Add ansible virtual repositories (#158)

Clients need one Galaxy server to install both internal collections and galaxy.ansible.com ones; ansible has no virtual type.

- merge member collection version lists in member order, the earlier member winning a duplicate version, cached with the shared virtual merge cache
- paginate limit/offset over the merged list, following remote member pages
- serve version detail from the owning member with href and download_url via the virtual; downloads 302 to the member through the member-redirect validation
- fail closed with 502 on a member error or a member versions list shorter than its count
- add docs/ansible/virtual.md and an e2e-docker case driving real ansible-galaxy

requires unkin/artifactapi#157

Reviewed-on: #158
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>

internal/virtual/ansible_merger.go

@@ -0,0 +1,182 @@
+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(m.Refs) >= doc.Meta.Count {
+			return m, nil
+		}
+		if len(doc.Data) == 0 {
+			return nil, fmt.Errorf("%s/%s: empty page after %d of %d versions", member, path, len(m.Refs), doc.Meta.Count)
+		}
+	}
+	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)
+}

internal/virtual/ansible_merger_test.go

@@ -0,0 +1,206 @@
+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 TestAnsibleVersionsShortRemote(t *testing.T) {
+	e := ansibleEngine(func(_ models.Remote, path string) ([]byte, error) {
+		if strings.Contains(path, "offset=") {
+			return json.Marshal(map[string]any{"meta": map[string]int{"count": 3}, "data": []any{}})
+		}
+		return galaxyPages([]string{"3.0.0", "2.0.0", "1.0.0"}, path)
+	})
+	if _, err := e.AnsibleVersions(context.Background(), ansibleVirt("galaxy"), "a", "b"); err == nil || errors.Is(err, ErrNotFound) {
+		t.Fatalf("empty page before count: 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)
+	}
+}

internal/virtual/engine.go

@@ -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
 	}