Add local ansible Galaxy repositories

internal/api/v1/ansible.go

@@ -0,0 +1,289 @@
+package v1
+
+import (
+	"bytes"
+	"cmp"
+	"context"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/url"
+	"slices"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+
+	"git.unkin.net/unkin/artifactapi/internal/database"
+	"git.unkin.net/unkin/artifactapi/internal/provider/ansible"
+	"git.unkin.net/unkin/artifactapi/internal/virtual"
+	"git.unkin.net/unkin/artifactapi/pkg/models"
+)
+
+var ansibleMaxPublishBytes int64 = 256 << 20
+
+// ansibleError writes a Galaxy v3 error document, which ansible-galaxy prints.
+func ansibleError(w http.ResponseWriter, status int, msg string) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(map[string]any{"errors": []map[string]string{{
+		"status": strconv.Itoa(status), "code": strings.ToLower(strings.ReplaceAll(http.StatusText(status), " ", "_")),
+		"title": http.StatusText(status), "detail": msg,
+	}}})
+}
+
+func writeAnsibleJSON(w http.ResponseWriter, status int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("Cache-Control", "no-store")
+	w.Header().Set("Vary", "Host, X-Forwarded-Proto")
+	w.WriteHeader(status)
+	enc := json.NewEncoder(w)
+	enc.SetEscapeHTML(false)
+	_ = enc.Encode(v)
+}
+
+func isAnsiblePublish(path string) bool {
+	return strings.TrimSuffix(path, "/")+"/" == ansible.PublishPath
+}
+
+func (h *ProxyHandler) handleAnsiblePublish(w http.ResponseWriter, r *http.Request) {
+	remote, err := h.db.GetRemote(r.Context(), chi.URLParam(r, "localName"))
+	if err != nil || remote.RepoType != models.RepoTypeLocal || remote.PackageType != models.PackageAnsible {
+		ansibleError(w, http.StatusMethodNotAllowed, "publish is only supported on local ansible repositories")
+		return
+	}
+	base, ok := requestBaseURL(w, r)
+	if !ok {
+		return
+	}
+	r.Body = http.MaxBytesReader(w, r.Body, ansibleMaxPublishBytes)
+	file, sum, err := readAnsibleUpload(r)
+	if err != nil {
+		var tooLarge *http.MaxBytesError
+		if errors.As(err, &tooLarge) {
+			ansibleError(w, http.StatusRequestEntityTooLarge, fmt.Sprintf("upload exceeds %d bytes", ansibleMaxPublishBytes))
+			return
+		}
+		ansibleError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+	digest := sha256.Sum256(file)
+	if got := hex.EncodeToString(digest[:]); sum != "" && !strings.EqualFold(sum, got) {
+		ansibleError(w, http.StatusBadRequest, "sha256 does not match the uploaded file")
+		return
+	}
+	e, err := ansible.ParseCollection(bytes.NewReader(file))
+	if err != nil {
+		ansibleError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+	e.SHA256, e.Size, e.CreatedAt = hex.EncodeToString(digest[:]), int64(len(file)), time.Now().UTC()
+
+	id := e.Namespace + "." + e.Name + ":" + e.Version
+	unlock, err := h.engine.Lock(r.Context(), remote.Name, "_ansiblelock/"+e.TaskID(), gemsLockWait, 5*time.Minute)
+	if err != nil {
+		ansibleError(w, http.StatusServiceUnavailable, id+" is being published, retry")
+		return
+	}
+	defer unlock()
+	conflict := func() {
+		ansibleError(w, http.StatusConflict, id+" already exists; overwrites are not allowed")
+	}
+	if f, err := h.db.GetLocalFile(r.Context(), remote.Name, e.IndexPath()); err != nil {
+		ansibleError(w, http.StatusInternalServerError, err.Error())
+		return
+	} else if f != nil {
+		conflict()
+		return
+	}
+	// The version is not live, so a tarball left by a failed publish is safe to drop.
+	artifact := ansible.ArtifactPrefix + e.Filename()
+	fail := func(status int, msg string) {
+		if err := h.db.DeleteLocalFile(context.WithoutCancel(r.Context()), remote.Name, artifact); err != nil {
+			slog.Error("ansible publish cleanup failed", "repo", remote.Name, "path", artifact, "error", err)
+		}
+		ansibleError(w, status, msg)
+	}
+	if ok, err := h.npmStoreImmutable(r.Context(), remote.Name, artifact, file, "application/gzip"); err != nil {
+		fail(http.StatusInternalServerError, err.Error())
+		return
+	} else if !ok {
+		fail(http.StatusConflict, id+" already exists; overwrites are not allowed")
+		return
+	}
+	entry, err := json.Marshal(e)
+	if err != nil {
+		fail(http.StatusInternalServerError, err.Error())
+		return
+	}
+	// The index entry gates visibility, so it is written last.
+	if _, err := h.npmStore(r.Context(), remote.Name, e.IndexPath(), entry, "application/json", false); err != nil {
+		if errors.Is(err, database.ErrAlreadyExists) {
+			conflict()
+			return
+		}
+		fail(http.StatusInternalServerError, err.Error())
+		return
+	}
+	slog.Info("local ansible collection published", "repo", remote.Name, "collection", id)
+	repoBase := base + "/api/v1/local/" + url.PathEscape(remote.Name)
+	writeAnsibleJSON(w, http.StatusAccepted, map[string]string{"task": repoBase + "/api/v3/imports/collections/" + e.TaskID() + "/"})
+}
+
+// readAnsibleUpload returns the multipart "file" part, which ansible-galaxy
+// base64-encodes, and the optional "sha256" field sent with it.
+func readAnsibleUpload(r *http.Request) ([]byte, string, error) {
+	mr, err := r.MultipartReader()
+	if err != nil {
+		return nil, "", errors.New("expected a multipart/form-data upload")
+	}
+	var file []byte
+	var sum string
+	for {
+		part, err := mr.NextPart()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return nil, "", err
+		}
+		switch part.FormName() {
+		case "file":
+			if file != nil {
+				return nil, "", errors.New("multiple file parts")
+			}
+			var src io.Reader = part
+			switch enc := strings.ToLower(part.Header.Get("Content-Transfer-Encoding")); enc {
+			case "base64":
+				src = base64.NewDecoder(base64.StdEncoding, part)
+			case "", "binary", "7bit", "8bit":
+			default:
+				return nil, "", fmt.Errorf("unsupported Content-Transfer-Encoding %q", enc)
+			}
+			if file, err = io.ReadAll(src); err != nil {
+				return nil, "", err
+			}
+		case "sha256":
+			b, err := io.ReadAll(io.LimitReader(part, 128))
+			if err != nil {
+				return nil, "", err
+			}
+			sum = strings.TrimSpace(string(b))
+		}
+	}
+	if file == nil {
+		return nil, "", errors.New("missing file part")
+	}
+	return file, sum, nil
+}
+
+// serveLocalAnsible answers the Galaxy v3 API of a local ansible repo.
+func (h *ProxyHandler) serveLocalAnsible(w http.ResponseWriter, r *http.Request, remote *models.Remote, ap ansible.APIPath) {
+	base, ok := requestBaseURL(w, r)
+	if !ok {
+		return
+	}
+	body, err := h.localAnsibleDoc(r.Context(), remote.Name, ap, r.URL.Query(), base+"/api/v1/local/"+url.PathEscape(remote.Name))
+	if errors.Is(err, virtual.ErrNotFound) {
+		ansibleError(w, http.StatusNotFound, "not found")
+		return
+	}
+	if err != nil {
+		ansibleError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	w.Header().Set("X-Artifact-Source", "local")
+	writeAnsibleJSON(w, http.StatusOK, body)
+}
+
+// LocalAnsible serves a local ansible repo's collection, full versions list
+// or version detail to ansible virtuals, with root-relative URLs.
+func (h *ProxyHandler) LocalAnsible(ctx context.Context, repo, path string) ([]byte, error) {
+	doc, err := h.localAnsibleDoc(ctx, repo, ansible.ParseAPIPath(path), nil, "/api/v1/local/"+url.PathEscape(repo))
+	if err != nil {
+		return nil, err
+	}
+	return json.Marshal(doc)
+}
+
+// localAnsibleDoc builds the document for ap; a nil q lists every version.
+func (h *ProxyHandler) localAnsibleDoc(ctx context.Context, repo string, ap ansible.APIPath, q url.Values, base string) (any, error) {
+	switch ap.Kind {
+	case ansible.APIRoot:
+		return ansible.Root(), nil
+	case ansible.APIImport:
+		ns, rest, _ := strings.Cut(ap.TaskID, "-")
+		name, version, _ := strings.Cut(rest, "-")
+		entries, err := h.localAnsibleEntries(ctx, repo, ns, name)
+		if err != nil {
+			return nil, err
+		}
+		for _, e := range entries {
+			if e.Version == version {
+				return ansible.ImportTask(e), nil
+			}
+		}
+	case ansible.APICollection, ansible.APIVersions, ansible.APIVersion:
+		entries, err := h.localAnsibleEntries(ctx, repo, ap.Namespace, ap.Name)
+		if err != nil || len(entries) == 0 {
+			return nil, cmp.Or(err, virtual.ErrNotFound)
+		}
+		refs := make([]ansible.VersionRef, len(entries))
+		for i, e := range entries {
+			refs[i] = e.Ref()
+		}
+		switch ap.Kind {
+		case ansible.APICollection:
+			return ansible.CollectionDoc(base, ap.Namespace, ap.Name, refs), nil
+		case ansible.APIVersions:
+			limit, offset := len(refs), 0
+			if q != nil {
+				limit, offset = ansible.Page(q)
+			}
+			return ansible.VersionsPage(base, ap.Namespace, ap.Name, refs, limit, offset), nil
+		}
+		for _, e := range entries {
+			if e.Version == ap.Version {
+				return ansible.VersionDoc(base, e), nil
+			}
+		}
+	}
+	return nil, virtual.ErrNotFound
+}
+
+// localAnsibleEntries returns a collection's versions, newest publish first.
+func (h *ProxyHandler) localAnsibleEntries(ctx context.Context, repo, ns, name string) ([]ansible.Entry, error) {
+	if !ansible.NameRe.MatchString(ns) || !ansible.NameRe.MatchString(name) {
+		return nil, nil
+	}
+	prefix := ansible.IndexPrefix + ns + "/" + name + "/"
+	files, err := h.db.ListLocalFilesByPrefix(ctx, repo, prefix)
+	if err != nil {
+		return nil, err
+	}
+	slices.SortFunc(files, func(a, b database.LocalFile) int { return cmp.Compare(b.ID, a.ID) })
+	var out []ansible.Entry
+	for _, f := range files {
+		// LIKE treats '_' as a wildcard, so the prefix match can over-select.
+		if !strings.HasPrefix(f.FilePath, prefix) {
+			continue
+		}
+		raw, err := h.readLocalBlob(ctx, f.ContentHash)
+		if err != nil {
+			return nil, err
+		}
+		var e ansible.Entry
+		if err := json.Unmarshal(raw, &e); err != nil {
+			return nil, fmt.Errorf("decode %s: %w", f.FilePath, err)
+		}
+		out = append(out, e)
+	}
+	return out, nil
+}

internal/api/v1/ansible_handler_test.go

@@ -0,0 +1,212 @@
+package v1
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"encoding/base64"
+	"encoding/json"
+	"mime/multipart"
+	"net/http"
+	"net/http/httptest"
+	"net/textproto"
+	"strings"
+	"testing"
+
+	"git.unkin.net/unkin/artifactapi/pkg/models"
+)
+
+func testCollection(t *testing.T, ns, name, version string) []byte {
+	t.Helper()
+	manifest := `{"collection_info":{"namespace":"` + ns + `","name":"` + name + `","version":"` + version + `","dependencies":{"ansible.posix":">=1.0.0"}}}`
+	var buf bytes.Buffer
+	zw := gzip.NewWriter(&buf)
+	tw := tar.NewWriter(zw)
+	_ = tw.WriteHeader(&tar.Header{Name: "MANIFEST.json", Mode: 0o644, Size: int64(len(manifest))})
+	_, _ = tw.Write([]byte(manifest))
+	_ = tw.Close()
+	_ = zw.Close()
+	return buf.Bytes()
+}
+
+// ansiblePublish uploads like ansible-galaxy: the file part base64-encoded
+// whenever a sha256 field is sent, raw otherwise.
+func ansiblePublish(t *testing.T, url string, tarball []byte, sum string) (int, string) {
+	t.Helper()
+	var body bytes.Buffer
+	mw := multipart.NewWriter(&body)
+	hdr := textproto.MIMEHeader{"Content-Disposition": {`form-data; name="file"; filename="c.tar.gz"`}}
+	data := tarball
+	if sum != "" {
+		_ = mw.WriteField("sha256", sum)
+		hdr.Set("Content-Transfer-Encoding", "base64")
+		data = []byte(base64.StdEncoding.EncodeToString(tarball))
+	}
+	fw, _ := mw.CreatePart(hdr)
+	_, _ = fw.Write(data)
+	_ = mw.Close()
+	req, _ := http.NewRequest(http.MethodPost, url, &body)
+	req.Header.Set("Content-Type", mw.FormDataContentType())
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	var b bytes.Buffer
+	_, _ = b.ReadFrom(resp.Body)
+	return resp.StatusCode, b.String()
+}
+
+func getJSON(t *testing.T, url string, v any) {
+	t.Helper()
+	code, body := npmDo(t, http.MethodGet, url, "")
+	if code != http.StatusOK {
+		t.Fatalf("GET %s = %d: %s", url, code, body)
+	}
+	if err := json.Unmarshal([]byte(body), v); err != nil {
+		t.Fatalf("GET %s: %v: %s", url, err, body)
+	}
+}
+
+func TestAnsiblePublishAndServe(t *testing.T) {
+	h, srv := npmTestServer(t)
+	repo := npmRepo(t, h, models.PackageAnsible)
+	base := srv.URL + "/api/v1/local/" + repo + "/"
+	publish := base + "api/v3/artifacts/collections/"
+
+	v1 := testCollection(t, "unkin", "e2e_hello", "1.0.0")
+	code, body := ansiblePublish(t, publish, v1, sha256Hex(v1))
+	if code != http.StatusAccepted {
+		t.Fatalf("publish = %d: %s", code, body)
+	}
+	var task struct{ Task string }
+	_ = json.Unmarshal([]byte(body), &task)
+	if task.Task != base+"api/v3/imports/collections/unkin-e2e_hello-1.0.0/" {
+		t.Fatalf("task = %q", task.Task)
+	}
+	var st map[string]any
+	getJSON(t, task.Task, &st)
+	if st["state"] != "completed" || st["finished_at"] == "" {
+		t.Fatalf("task = %v", st)
+	}
+	if code, _ := npmDo(t, http.MethodGet, base+"api/v3/imports/collections/unkin-e2e_hello-9.9.9/", ""); code != http.StatusNotFound {
+		t.Errorf("unknown task = %d, want 404", code)
+	}
+
+	for _, v := range []string{"1.1.0", "2.0.0"} {
+		tb := testCollection(t, "unkin", "e2e_hello", v)
+		if code, body := ansiblePublish(t, publish, tb, ""); code != http.StatusAccepted {
+			t.Fatalf("publish %s = %d: %s", v, code, body)
+		}
+	}
+	if code, body := ansiblePublish(t, publish, v1, ""); code != http.StatusConflict || !strings.Contains(body, "already exists") {
+		t.Errorf("identical republish = %d: %s, want 409", code, body)
+	}
+	if code, _ := ansiblePublish(t, publish, testCollection(t, "unkin", "e2e_hello", "1.0.0")[:20], ""); code != http.StatusBadRequest {
+		t.Errorf("truncated tarball = %d, want 400", code)
+	}
+
+	var root map[string]map[string]string
+	getJSON(t, base+"api/", &root)
+	if root["available_versions"]["v3"] != "v3/" {
+		t.Fatalf("root = %v", root)
+	}
+
+	var page struct {
+		Meta  struct{ Count int }
+		Links struct{ Next, Previous *string }
+		Data  []struct{ Version, Href string }
+	}
+	getJSON(t, base+"api/v3/collections/unkin/e2e_hello/versions/?limit=2", &page)
+	if page.Meta.Count != 3 || len(page.Data) != 2 || page.Data[0].Version != "2.0.0" || page.Links.Next == nil || page.Links.Previous != nil {
+		t.Fatalf("page 1 = %+v", page)
+	}
+	next := *page.Links.Next
+	page.Links.Next = nil
+	getJSON(t, next, &page)
+	if len(page.Data) != 1 || page.Data[0].Version != "1.0.0" || page.Links.Next != nil {
+		t.Fatalf("page 2 (%s) = %+v", next, page)
+	}
+
+	var detail struct {
+		DownloadURL string `json:"download_url"`
+		Artifact    struct{ Sha256 string }
+		Metadata    struct{ Dependencies map[string]string }
+		Namespace   struct{ Name string }
+		Collection  struct{ Name string }
+	}
+	getJSON(t, page.Data[0].Href, &detail)
+	if detail.DownloadURL != base+"artifacts/unkin-e2e_hello-1.0.0.tar.gz" || detail.Artifact.Sha256 != sha256Hex(v1) ||
+		detail.Metadata.Dependencies["ansible.posix"] != ">=1.0.0" || detail.Namespace.Name != "unkin" || detail.Collection.Name != "e2e_hello" {
+		t.Fatalf("detail = %+v", detail)
+	}
+	if code, got := npmDo(t, http.MethodGet, detail.DownloadURL, ""); code != http.StatusOK || got != string(v1) {
+		t.Fatalf("download = %d (%d bytes)", code, len(got))
+	}
+
+	var coll map[string]any
+	getJSON(t, base+"api/v3/collections/unkin/e2e_hello/", &coll)
+	if coll["versions_url"] != base+"api/v3/collections/unkin/e2e_hello/versions/" || coll["updated_at"] == "" {
+		t.Fatalf("collection = %v", coll)
+	}
+	for _, p := range []string{"api/v3/collections/unkin/nope/", "api/v3/collections/unkin/e2e_hello/versions/9.0.0/", "_index/unkin/e2e_hello/1.0.0", "other"} {
+		if code, _ := npmDo(t, http.MethodGet, base+p, ""); code != http.StatusNotFound {
+			t.Errorf("GET %s = %d, want 404", p, code)
+		}
+	}
+}
+
+func TestAnsiblePublishRejects(t *testing.T) {
+	h, srv := npmTestServer(t)
+	repo := npmRepo(t, h, models.PackageAnsible)
+	publish := srv.URL + "/api/v1/local/" + repo + "/api/v3/artifacts/collections/"
+	tb := testCollection(t, "unkin", "rej", "1.0.0")
+
+	if code, _ := ansiblePublish(t, publish, tb, strings.Repeat("0", 64)); code != http.StatusBadRequest {
+		t.Errorf("sha256 mismatch = %d, want 400", code)
+	}
+	if code, _ := npmDo(t, http.MethodPost, publish, "raw"); code != http.StatusBadRequest {
+		t.Errorf("non-multipart = %d, want 400", code)
+	}
+	old := ansibleMaxPublishBytes
+	ansibleMaxPublishBytes = 64
+	if code, _ := ansiblePublish(t, publish, tb, ""); code != http.StatusRequestEntityTooLarge {
+		t.Errorf("oversize = %d, want 413", code)
+	}
+	ansibleMaxPublishBytes = old
+	generic := npmRepo(t, h, models.PackageGeneric)
+	if code, _ := ansiblePublish(t, srv.URL+"/api/v1/local/"+generic+"/api/v3/artifacts/collections/", tb, ""); code != http.StatusMethodNotAllowed {
+		t.Errorf("publish to generic local = %d, want 405", code)
+	}
+	if code, _ := ansiblePublish(t, publish, tb, ""); code != http.StatusAccepted {
+		t.Errorf("publish after rejects = %d, want 202", code)
+	}
+}
+
+func TestAnsibleLocalPerHost(t *testing.T) {
+	h, srv := npmTestServer(t)
+	repo := npmRepo(t, h, models.PackageAnsible)
+	if code, b := ansiblePublish(t, srv.URL+"/api/v1/local/"+repo+"/api/v3/artifacts/collections/", testCollection(t, "unkin", "host", "1.0.0"), ""); code != http.StatusAccepted {
+		t.Fatalf("publish = %d: %s", code, b)
+	}
+	routes := h.Routes()
+	path := "/local/" + repo + "/api/v3/collections/unkin/host/versions/1.0.0/"
+	r := httptest.NewRequest(http.MethodGet, path, nil)
+	r.Host = "x@evil.com"
+	w := httptest.NewRecorder()
+	routes.ServeHTTP(w, r)
+	if w.Code != http.StatusBadRequest {
+		t.Errorf("bad Host = %d, want 400", w.Code)
+	}
+	r = httptest.NewRequest(http.MethodGet, path, nil)
+	r.Host = "aa.example"
+	r.Header.Set("X-Forwarded-Proto", "https")
+	w = httptest.NewRecorder()
+	routes.ServeHTTP(w, r)
+	if w.Code != http.StatusOK || !strings.Contains(w.Body.String(), `"download_url":"https://aa.example/api/v1/local/`+repo+`/artifacts/unkin-host-1.0.0.tar.gz"`) {
+		t.Fatalf("detail = %d: %s", w.Code, w.Body)
+	}
+	if cc, vary := w.Header().Get("Cache-Control"), w.Header().Get("Vary"); cc != "no-store" || vary != "Host, X-Forwarded-Proto" {
+		t.Errorf("Cache-Control %q Vary %q", cc, vary)
+	}
+}

internal/api/v1/gems.go

@@ -64,6 +64,10 @@ func (h *ProxyHandler) lockGemVersion(w http.ResponseWriter, r *http.Request, re
 }
 
 func (h *ProxyHandler) handleLocalPost(w http.ResponseWriter, r *http.Request) {
+	if isAnsiblePublish(chi.URLParam(r, "*")) {
+		h.handleAnsiblePublish(w, r)
+		return
+	}
 	if chi.URLParam(r, "*") != "api/v1/gems" {
 		gemsText(w, http.StatusNotFound, "not found")
 		return

internal/api/v1/proxy.go

@@ -17,6 +17,7 @@ import (
 	v2 "git.unkin.net/unkin/artifactapi/internal/api/v2"
 	"git.unkin.net/unkin/artifactapi/internal/database"
 	"git.unkin.net/unkin/artifactapi/internal/provider"
+	"git.unkin.net/unkin/artifactapi/internal/provider/ansible"
 	"git.unkin.net/unkin/artifactapi/internal/proxy"
 	"git.unkin.net/unkin/artifactapi/internal/storage"
 	"git.unkin.net/unkin/artifactapi/internal/virtual"
@@ -275,6 +276,17 @@ func (h *ProxyHandler) handleLocal(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
+	if remote.PackageType == models.PackageAnsible {
+		if ap := ansible.ParseAPIPath(path); ap.Kind != ansible.APIUnknown {
+			h.serveLocalAnsible(w, r, remote, ap)
+			return
+		}
+		if !strings.HasPrefix(path, ansible.ArtifactPrefix) {
+			http.Error(w, "not found", http.StatusNotFound)
+			return
+		}
+	}
+
 	prov, _ := provider.Get(remote.PackageType)
 	if indexer, ok := prov.(provider.LocalIndexer); ok {
 		if indexer.ServeLocalIndex(w, r, h.db, remote.Name, path) {