Add ansible virtual repositories
ci/woodpecker/pr/pre-commit Pipeline was successful
ci/woodpecker/pr/build Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful

This commit is contained in:
2026-10-18 03:45:30 +11:00
parent ba8d3a3c39
commit 4b202b1852
11 changed files with 604 additions and 6 deletions
+56
View File
@@ -4,6 +4,7 @@ import (
"archive/tar"
"bytes"
"compress/gzip"
"context"
"encoding/base64"
"encoding/json"
"mime/multipart"
@@ -13,6 +14,7 @@ import (
"strings"
"testing"
"git.unkin.net/unkin/artifactapi/internal/virtual"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
@@ -210,3 +212,57 @@ func TestAnsibleLocalPerHost(t *testing.T) {
t.Errorf("Cache-Control %q Vary %q", cc, vary)
}
}
func TestAnsibleVirtualOverLocals(t *testing.T) {
h, srv := npmTestServer(t)
h.virtualEngine = virtual.NewEngine(h.db, nil)
h.virtualEngine.AnsibleLocal = h.LocalAnsible
a, b := npmRepo(t, h, models.PackageAnsible), npmRepo(t, h, models.PackageAnsible)
for _, p := range []struct{ repo, version string }{{a, "1.0.0"}, {b, "1.0.0"}, {b, "2.0.0"}} {
if code, body := ansiblePublish(t, srv.URL+"/api/v1/local/"+p.repo+"/api/v3/artifacts/collections/", testCollection(t, "unkin", "virt", p.version), ""); code != http.StatusAccepted {
t.Fatalf("publish %s to %s = %d: %s", p.version, p.repo, code, body)
}
}
virt := "v" + a
if err := h.db.CreateVirtual(context.Background(), &models.Virtual{Name: virt, PackageType: models.PackageAnsible, Members: []string{a, b}}); err != nil {
t.Fatal(err)
}
base := srv.URL + "/api/v1/virtual/" + virt + "/"
var page struct {
Meta struct{ Count int }
Links struct{ Next *string }
Data []struct{ Version, Href string }
}
getJSON(t, base+"api/v3/collections/unkin/virt/versions/?limit=1", &page)
if page.Meta.Count != 2 || len(page.Data) != 1 || page.Data[0].Version != "1.0.0" || page.Links.Next == nil {
t.Fatalf("page 1 = %+v", page)
}
href := page.Data[0].Href
getJSON(t, *page.Links.Next, &page)
if len(page.Data) != 1 || page.Data[0].Version != "2.0.0" || page.Links.Next != nil {
t.Fatalf("page 2 = %+v", page)
}
var detail struct {
DownloadURL string `json:"download_url"`
}
getJSON(t, href, &detail)
if detail.DownloadURL != base+a+"/artifacts/unkin-virt-1.0.0.tar.gz" {
t.Fatalf("download_url = %q, want the first member via the virtual", detail.DownloadURL)
}
noFollow := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
resp, err := noFollow.Get(detail.DownloadURL)
if err != nil {
t.Fatal(err)
}
_ = resp.Body.Close()
if resp.StatusCode != http.StatusFound || resp.Header.Get("Location") != srv.URL+"/api/v1/local/"+a+"/artifacts/unkin-virt-1.0.0.tar.gz" {
t.Fatalf("download = %d -> %q", resp.StatusCode, resp.Header.Get("Location"))
}
for _, p := range []string{"api/v3/collections/unkin/nope/", "notamember/x.tar.gz"} {
if code, _ := npmDo(t, http.MethodGet, base+p, ""); code != http.StatusNotFound {
t.Errorf("GET %s = %d, want 404", p, code)
}
}
}
+64
View File
@@ -0,0 +1,64 @@
package v1
import (
"encoding/json"
"errors"
"log/slog"
"net/http"
"net/url"
"git.unkin.net/unkin/artifactapi/internal/provider/ansible"
"git.unkin.net/unkin/artifactapi/internal/virtual"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
// serveVirtualAnsible serves the merged Galaxy v3 API and 302s downloads to
// the member owning that version.
func (h *ProxyHandler) serveVirtualAnsible(w http.ResponseWriter, r *http.Request, virt models.Virtual, path, proxyBaseURL string) {
ap := ansible.ParseAPIPath(path)
base := proxyBaseURL + "/api/v1/virtual/" + url.PathEscape(virt.Name)
var doc any
var err error
switch ap.Kind {
case ansible.APIRoot:
doc = ansible.Root()
case ansible.APIVersion:
var body []byte
if body, err = h.virtualEngine.AnsibleVersion(r.Context(), virt, ap.Namespace, ap.Name, ap.Version, proxyBaseURL); err == nil {
doc = json.RawMessage(body)
}
case ansible.APICollection, ansible.APIVersions:
var d *virtual.AnsibleDoc
if d, err = h.virtualEngine.AnsibleVersions(r.Context(), virt, ap.Namespace, ap.Name); err == nil {
if ap.Kind == ansible.APICollection {
doc = ansible.CollectionDoc(base, ap.Namespace, ap.Name, d.Refs)
} else {
limit, offset := ansible.Page(r.URL.Query())
doc = ansible.VersionsPage(base, ap.Namespace, ap.Name, d.Refs, limit, offset)
}
}
default:
loc, ok, err := h.virtualEngine.MemberRedirect(r.Context(), virt, path, r.URL.RawQuery, proxyBaseURL)
switch {
case err != nil:
ansibleError(w, http.StatusBadRequest, err.Error())
case !ok:
ansibleError(w, http.StatusNotFound, "not found")
default:
w.Header().Set("Cache-Control", "no-store")
w.Header().Set("Vary", "Host, X-Forwarded-Proto")
http.Redirect(w, r, loc, http.StatusFound)
}
return
}
switch {
case errors.Is(err, virtual.ErrNotFound):
ansibleError(w, http.StatusNotFound, "not found")
case err != nil:
slog.Error("virtual ansible fetch failed", "virtual", virt.Name, "path", path, "error", err)
ansibleError(w, http.StatusBadGateway, "bad gateway")
default:
w.Header().Set("X-Artifact-Source", "virtual")
writeAnsibleJSON(w, http.StatusOK, doc)
}
}
+4
View File
@@ -211,6 +211,10 @@ func (h *ProxyHandler) handleVirtual(w http.ResponseWriter, r *http.Request) {
h.serveVirtualGems(w, r, *virt, path, proxyBaseURL)
return
}
if virt.PackageType == models.PackageAnsible {
h.serveVirtualAnsible(w, r, *virt, path, proxyBaseURL)
return
}
loc, ok, err := h.virtualEngine.MemberRedirect(r.Context(), *virt, path, r.URL.RawQuery, proxyBaseURL)
if err != nil {
+1
View File
@@ -168,6 +168,7 @@ func (s *Server) routes() chi.Router {
proxyHandler := v1.NewProxyHandler(s.engine, s.virtEngine, s.db, s.store, s.localHandler)
s.virtEngine.NPMLocal = proxyHandler.LocalNPMPackument
s.virtEngine.GemsLocal = proxyHandler.LocalGemsIndex
s.virtEngine.AnsibleLocal = proxyHandler.LocalAnsible
r.Mount("/api/v1", proxyHandler.Routes())
r.Mount("/v2", proxyHandler.DockerV2Routes())
+179
View File
@@ -0,0 +1,179 @@
package virtual
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"net/url"
"strconv"
"strings"
"git.unkin.net/unkin/artifactapi/internal/provider/ansible"
"git.unkin.net/unkin/artifactapi/internal/proxy"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
const (
maxAnsibleCollections = 1024
ansibleMemberPage = 100
maxAnsibleMemberPages = 100
)
// AnsibleMember is one member's versions of a collection; Found is false when
// the member does not have it.
type AnsibleMember struct {
RemoteName string
Found bool
Refs []ansible.VersionRef
}
// AnsibleDoc is a collection's merged versions; Owners maps each version to
// the member serving it.
type AnsibleDoc struct {
Refs []ansible.VersionRef
Owners map[string]string
body []byte
}
func (d *AnsibleDoc) index() []byte { return d.body }
func (d *AnsibleDoc) files() map[string][]byte { return nil }
// MergeAnsibleVersions concatenates member versions in member order; the
// first member wins a duplicate version. ErrNotFound means no member has the
// collection.
func MergeAnsibleVersions(members []AnsibleMember) (*AnsibleDoc, error) {
d := &AnsibleDoc{Refs: []ansible.VersionRef{}, Owners: map[string]string{}}
found := false
for _, m := range members {
found = found || m.Found
for _, r := range m.Refs {
if _, dup := d.Owners[r.Version]; dup {
continue
}
d.Owners[r.Version] = m.RemoteName
d.Refs = append(d.Refs, r)
}
}
if !found {
return nil, ErrNotFound
}
var err error
d.body, err = json.Marshal(d.Refs)
return d, err
}
// AnsibleVersions returns the merged versions of ns.name across members.
func (e *Engine) AnsibleVersions(ctx context.Context, virt models.Virtual, ns, name string) (*AnsibleDoc, error) {
prefix := "ansible\x00" + virt.Name + "\x00"
inserted := false
repo, err := e.merged(ctx, prefix+ns+"/"+name, func(ctx context.Context) (mergedRepo, error) {
inserted = true
members, err := fetchMembers(virt, func(m string) (*AnsibleMember, error) { return e.ansibleMember(ctx, m, ns, name) })
if err != nil {
return nil, err
}
return MergeAnsibleVersions(members)
})
if err != nil {
return nil, err
}
if inserted {
e.evictOldest(prefix, maxAnsibleCollections)
}
return repo.(*AnsibleDoc), nil
}
// AnsibleVersion serves the owning member's version detail with href and
// download_url pointing at the virtual; the download 302s to the member.
// Member documents carry root-relative URLs, so anything else is rejected.
func (e *Engine) AnsibleVersion(ctx context.Context, virt models.Virtual, ns, name, version, proxyBaseURL string) ([]byte, error) {
d, err := e.AnsibleVersions(ctx, virt, ns, name)
if err != nil {
return nil, err
}
owner, ok := d.Owners[version]
if !ok {
return nil, ErrNotFound
}
remote, err := e.getRemote(ctx, owner)
if err != nil {
return nil, fmt.Errorf("remote %q: %w", owner, err)
}
body, err := e.ansibleDoc(ctx, *remote, ansible.VersionPath(ns, name, version)[1:])
if err != nil {
return nil, fmt.Errorf("member %q: %v", owner, err)
}
var detail map[string]any
if err := json.Unmarshal(body, &detail); err != nil {
return nil, fmt.Errorf("member %q version detail: %v", owner, err)
}
dl, _ := detail["download_url"].(string)
u, err := url.Parse(dl)
if err != nil {
return nil, fmt.Errorf("member %q download_url %q: %v", owner, dl, err)
}
rest, ok := strings.CutPrefix(u.EscapedPath(), "/api/v1/"+string(remote.RepoType)+"/"+url.PathEscape(owner)+"/")
if !ok || u.Scheme != "" || u.Host != "" {
return nil, fmt.Errorf("member %q download_url %q is not served by the member", owner, dl)
}
base := strings.TrimRight(proxyBaseURL, "/") + "/api/v1/virtual/" + url.PathEscape(virt.Name)
detail["download_url"] = base + "/" + url.PathEscape(owner) + "/" + rest
detail["href"] = base + ansible.VersionPath(ns, name, version)
return json.Marshal(detail)
}
// ansibleMember reads every version a member lists for ns.name, following
// the member's pagination for remotes.
func (e *Engine) ansibleMember(ctx context.Context, member, ns, name string) (*AnsibleMember, error) {
remote, err := e.getRemote(ctx, member)
if err != nil {
return nil, fmt.Errorf("remote %q: %w", member, err)
}
m := &AnsibleMember{RemoteName: member}
list := ansible.CollectionPath(ns, name)[1:] + "versions/"
for page := 0; page < maxAnsibleMemberPages; page++ {
// Locals list every version at once; remote pages match the keys the
// remote caches client requests under.
path := list
if remote.RepoType != models.RepoTypeLocal {
path += "?limit=" + strconv.Itoa(ansibleMemberPage)
if page > 0 {
path += "&offset=" + strconv.Itoa(len(m.Refs))
}
}
body, err := e.ansibleDoc(ctx, *remote, path)
var pe *proxy.ProxyError
if errors.Is(err, ErrNotFound) || errors.As(err, &pe) && pe.Status == http.StatusNotFound {
return m, nil
}
if err != nil {
return nil, err
}
var doc struct {
Meta struct{ Count int }
Data []ansible.VersionRef
}
if err := json.Unmarshal(body, &doc); err != nil {
return nil, fmt.Errorf("%s/%s: %v", member, path, err)
}
m.Found = true
m.Refs = append(m.Refs, doc.Data...)
if remote.RepoType == models.RepoTypeLocal || len(doc.Data) == 0 || len(m.Refs) >= doc.Meta.Count {
return m, nil
}
}
return nil, fmt.Errorf("%s: more than %d pages of %s.%s versions", member, maxAnsibleMemberPages, ns, name)
}
// ansibleMemberDoc reads a Galaxy API document from a member.
func (e *Engine) ansibleMemberDoc(ctx context.Context, remote models.Remote, path string) ([]byte, error) {
if remote.RepoType != models.RepoTypeLocal {
return e.fetchMemberPath(ctx, remote, path)
}
if e.AnsibleLocal == nil {
return nil, errors.New("local ansible indexes are not wired")
}
return e.AnsibleLocal(ctx, remote.Name, path)
}
+194
View File
@@ -0,0 +1,194 @@
package virtual
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"strings"
"sync"
"testing"
"time"
"git.unkin.net/unkin/artifactapi/internal/provider/ansible"
"git.unkin.net/unkin/artifactapi/internal/proxy"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
func refs(vs ...string) []ansible.VersionRef {
out := make([]ansible.VersionRef, len(vs))
for i, v := range vs {
out[i] = ansible.VersionRef{Version: v}
}
return out
}
func TestMergeAnsibleVersionsMemberOrderWins(t *testing.T) {
d, err := MergeAnsibleVersions([]AnsibleMember{
{RemoteName: "internal", Found: true, Refs: refs("2.0.0", "1.0.0")},
{RemoteName: "missing"},
{RemoteName: "galaxy", Found: true, Refs: refs("3.0.0", "2.0.0", "0.9.0")},
})
if err != nil {
t.Fatal(err)
}
var got []string
for _, r := range d.Refs {
got = append(got, r.Version)
}
if strings.Join(got, ",") != "2.0.0,1.0.0,3.0.0,0.9.0" {
t.Fatalf("merged = %v", got)
}
for v, owner := range map[string]string{"2.0.0": "internal", "1.0.0": "internal", "3.0.0": "galaxy", "0.9.0": "galaxy"} {
if d.Owners[v] != owner {
t.Errorf("owner of %s = %q, want %q", v, d.Owners[v], owner)
}
}
if _, err := MergeAnsibleVersions([]AnsibleMember{{RemoteName: "a"}, {RemoteName: "b"}}); !errors.Is(err, ErrNotFound) {
t.Fatalf("no member has it: want ErrNotFound, got %v", err)
}
if d, err := MergeAnsibleVersions([]AnsibleMember{{RemoteName: "a", Found: true}}); err != nil || len(d.Refs) != 0 {
t.Fatalf("found with no versions = %v %v", d, err)
}
}
// galaxyPages answers a remote's versions list like Galaxy, two per page.
func galaxyPages(versions []string, path string) ([]byte, error) {
_, q, _ := strings.Cut(path, "?")
offset := 0
if _, off, ok := strings.Cut(q, "offset="); ok {
_, _ = fmt.Sscan(off, &offset)
}
end := min(offset+2, len(versions))
return json.Marshal(map[string]any{"meta": map[string]int{"count": len(versions)}, "data": refs(versions[min(offset, end):end]...)})
}
func ansibleEngine(doc func(remote models.Remote, path string) ([]byte, error)) *Engine {
return &Engine{
mergeTTL: time.Minute,
getRemote: func(_ context.Context, name string) (*models.Remote, error) {
rt := models.RepoTypeRemote
if name == "internal" {
rt = models.RepoTypeLocal
}
return &models.Remote{Name: name, RepoType: rt}, nil
},
ansibleDoc: func(_ context.Context, remote models.Remote, path string) ([]byte, error) { return doc(remote, path) },
}
}
func ansibleVirt(members ...string) models.Virtual {
return models.Virtual{Name: "ansible", PackageType: models.PackageAnsible, Members: members}
}
func TestAnsibleVersionsFollowsRemotePages(t *testing.T) {
var mu sync.Mutex
var asked []string
e := ansibleEngine(func(remote models.Remote, path string) ([]byte, error) {
mu.Lock()
asked = append(asked, remote.Name+" "+path)
mu.Unlock()
if remote.Name == "internal" {
return galaxyPages([]string{"1.0.0", "5.0.0"}, "")
}
return galaxyPages([]string{"5.0.0", "4.0.0", "3.0.0", "2.0.0", "1.5.0"}, path)
})
virt := ansibleVirt("internal", "galaxy")
d, err := e.AnsibleVersions(context.Background(), virt, "community", "general")
if err != nil {
t.Fatal(err)
}
if len(d.Refs) != 6 || d.Owners["5.0.0"] != "internal" || d.Owners["1.5.0"] != "galaxy" {
t.Fatalf("merged = %+v", d)
}
want := map[string]bool{
"internal api/v3/collections/community/general/versions/": true,
"galaxy api/v3/collections/community/general/versions/?limit=100": true,
"galaxy api/v3/collections/community/general/versions/?limit=100&offset=2": true,
"galaxy api/v3/collections/community/general/versions/?limit=100&offset=4": true,
}
if len(asked) != len(want) {
t.Fatalf("asked %v", asked)
}
for _, a := range asked {
if !want[a] {
t.Errorf("unexpected fetch %q", a)
}
}
if _, err := e.AnsibleVersions(context.Background(), virt, "community", "general"); err != nil || len(asked) != len(want) {
t.Fatalf("merge not cached: %v, %d fetches", err, len(asked))
}
}
func TestAnsibleVersionsFailsClosed(t *testing.T) {
e := ansibleEngine(func(remote models.Remote, path string) ([]byte, error) {
switch remote.Name {
case "internal":
return nil, ErrNotFound
case "gone":
return nil, &proxy.ProxyError{Status: http.StatusNotFound}
}
return nil, errors.New("upstream down")
})
if _, err := e.AnsibleVersions(context.Background(), ansibleVirt("internal", "gone"), "a", "b"); !errors.Is(err, ErrNotFound) {
t.Fatalf("all members lack it: want ErrNotFound, got %v", err)
}
if _, err := e.AnsibleVersions(context.Background(), ansibleVirt("internal", "galaxy"), "a", "b"); err == nil || errors.Is(err, ErrNotFound) {
t.Fatalf("member error: want a non-404 error, got %v", err)
}
}
func TestAnsibleVersionRoutesViaVirtual(t *testing.T) {
detail := func(dl string) []byte {
b, _ := json.Marshal(map[string]any{"version": "1.0.0", "download_url": dl, "artifact": map[string]string{"sha256": "ab"}})
return b
}
dls := map[string]string{
"internal": "/api/v1/local/internal/artifacts/a-b-1.0.0.tar.gz",
"galaxy": "/api/v1/remote/galaxy/api/v3/plugin/ansible/content/published/collections/artifacts/a-b-2.0.0.tar.gz",
"evil": "https://evil.example/api/v1/remote/evil/x.tar.gz",
"other": "/api/v1/remote/someoneelse/x.tar.gz",
}
e := ansibleEngine(func(remote models.Remote, path string) ([]byte, error) {
if strings.Contains(path, "/versions/?") || strings.HasSuffix(path, "/versions/") {
v := map[string]string{"internal": "1.0.0", "galaxy": "2.0.0", "evil": "3.0.0", "other": "4.0.0"}[remote.Name]
return galaxyPages([]string{v}, path)
}
return detail(dls[remote.Name]), nil
})
virt := ansibleVirt("internal", "galaxy", "evil", "other")
for v, want := range map[string]string{
"1.0.0": "https://h/api/v1/virtual/ansible/internal/artifacts/a-b-1.0.0.tar.gz",
"2.0.0": "https://h/api/v1/virtual/ansible/galaxy/api/v3/plugin/ansible/content/published/collections/artifacts/a-b-2.0.0.tar.gz",
} {
body, err := e.AnsibleVersion(context.Background(), virt, "a", "b", v, "https://h")
if err != nil {
t.Fatal(err)
}
var got struct {
Href string
DownloadURL string `json:"download_url"`
Artifact struct{ Sha256 string }
}
_ = json.Unmarshal(body, &got)
if got.DownloadURL != want || got.Href != "https://h/api/v1/virtual/ansible/api/v3/collections/a/b/versions/"+v+"/" || got.Artifact.Sha256 != "ab" {
t.Errorf("%s detail = %+v", v, got)
}
loc, ok, err := e.MemberRedirect(context.Background(), virt, strings.TrimPrefix(want, "https://h/api/v1/virtual/ansible/"), "", "https://h")
if err != nil || !ok || loc != "https://h"+dls[map[string]string{"1.0.0": "internal", "2.0.0": "galaxy"}[v]] {
t.Errorf("%s redirect = %q %v %v", v, loc, ok, err)
}
}
for _, v := range []string{"3.0.0", "4.0.0"} {
if _, err := e.AnsibleVersion(context.Background(), virt, "a", "b", v, "https://h"); err == nil || errors.Is(err, ErrNotFound) {
t.Errorf("%s: foreign download_url accepted: %v", v, err)
}
}
if _, err := e.AnsibleVersion(context.Background(), virt, "a", "b", "9.9.9", "https://h"); !errors.Is(err, ErrNotFound) {
t.Errorf("unknown version: want ErrNotFound, got %v", err)
}
if _, _, err := e.MemberRedirect(context.Background(), virt, "galaxy/../internal/x", "", "https://h"); !errors.Is(err, ErrBadPath) {
t.Errorf("escaping redirect: want ErrBadPath, got %v", err)
}
}
+11 -4
View File
@@ -38,6 +38,11 @@ type Engine struct {
// ErrNotFound.
GemsLocal func(ctx context.Context, repo, path string) ([]byte, error)
// AnsibleLocal returns a local ansible repo's collection, full versions list
// or version detail with root-relative URLs, or ErrNotFound.
AnsibleLocal func(ctx context.Context, repo, path string) ([]byte, error)
ansibleDoc func(ctx context.Context, remote models.Remote, path string) ([]byte, error)
mergeTTL time.Duration
sf singleflight.Group
@@ -68,6 +73,7 @@ func NewEngine(db *database.DB, proxyEngine *proxy.Engine) *Engine {
e.rpmMember = e.fetchRPMMember
e.npmMember = e.fetchNPMMember
e.gemsMember = e.fetchGemsMember
e.ansibleDoc = e.ansibleMemberDoc
return e
}
@@ -192,16 +198,17 @@ var (
ErrBadPath = errors.New("invalid path")
)
// MemberRedirect maps an rpm or npm virtual package path (<member>/<href>, as
// written by MergeRPM/MergeNPM) to the owning member's route. ok is false for
// any other path; ErrBadPath means the path is absolute or escapes the member.
// MemberRedirect maps an rpm, npm or ansible virtual package path
// (<member>/<href>, as written by MergeRPM/MergeNPM/AnsibleVersion) to the
// owning member's route. ok is false for any other path; ErrBadPath means the
// path is absolute or escapes the member.
func (e *Engine) MemberRedirect(ctx context.Context, virt models.Virtual, path, rawQuery, proxyBaseURL string) (string, bool, error) {
switch virt.PackageType {
case models.PackageRPM:
if strings.HasPrefix(path, "repodata/") {
return "", false, nil
}
case models.PackageNPM:
case models.PackageNPM, models.PackageAnsible:
default:
return "", false, nil
}