Bound npm publish body and make interrupted publishes retryable

2026-10-17 12:34:18 +11:00
parent d8e2446262
commit 4747693fa8
3 changed files with 248 additions and 20 deletions
@@ -30,8 +30,11 @@ import (
 //   <pkg>/-/dist-tags/<tag>        version string (mutable)

 var (
-	npmNameRe = regexp.MustCompile(`^(@[a-z0-9][a-z0-9._~-]*/)?[a-z0-9][a-z0-9._~-]*$`)
-	npmRefRe  = regexp.MustCompile(`^[0-9A-Za-z][0-9A-Za-z.+_-]*$`)
+	npmNameRe   = regexp.MustCompile(`^(@[a-z0-9][a-z0-9._~-]*/)?[a-z0-9][a-z0-9._~-]*$`)
+	npmRefRe    = regexp.MustCompile(`^[0-9A-Za-z][0-9A-Za-z.+_-]*$`)
+	npmSemverRe = regexp.MustCompile(`^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$`)
+
+	npmMaxPublishBytes int64 = 256 << 20
 )

 func (h *ProxyHandler) localNPMRemote(r *http.Request, name string) (*models.Remote, bool) {
@@ -86,7 +89,7 @@ func parseNPMPublish(body io.Reader, pkg string) (*npmRelease, error) {
 	for v, m := range doc.Versions {
 		rel.version, rel.manifest = v, m
 	}
-	if !npmRefRe.MatchString(rel.version) {
+	if !npmSemverRe.MatchString(rel.version) {
 		return nil, fmt.Errorf("invalid version %q", rel.version)
 	}
 	var meta struct {
@@ -145,7 +148,7 @@ func verifyNPMDist(tgz []byte, integrity, shasum string) error {
 func npmError(w http.ResponseWriter, status int, msg string) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(status)
-	json.NewEncoder(w).Encode(map[string]string{"error": msg})
+	_ = json.NewEncoder(w).Encode(map[string]string{"error": msg})
 }

 func (h *ProxyHandler) handleLocalPut(w http.ResponseWriter, r *http.Request) {
@@ -159,52 +162,74 @@ func (h *ProxyHandler) handleLocalPut(w http.ResponseWriter, r *http.Request) {
 		npmError(w, http.StatusBadRequest, "invalid package name")
 		return
 	}
-	rel, err := parseNPMPublish(r.Body, pkg)
+	rel, err := parseNPMPublish(http.MaxBytesReader(w, r.Body, npmMaxPublishBytes), pkg)
 	if err != nil {
+		var tooLarge *http.MaxBytesError
+		if errors.As(err, &tooLarge) {
+			npmError(w, http.StatusRequestEntityTooLarge, fmt.Sprintf("publish body exceeds %d bytes", npmMaxPublishBytes))
+			return
+		}
 		npmError(w, http.StatusBadRequest, err.Error())
 		return
 	}

 	base := pkg[strings.LastIndex(pkg, "/")+1:]
 	stem := pkg + "/-/" + base + "-" + rel.version
-	// The tarball's immutable row is the version's uniqueness gate.
-	if err := h.npmStore(r.Context(), remote.Name, stem+".tgz", rel.tarball, "application/octet-stream", false); err != nil {
-		if errors.Is(err, database.ErrAlreadyExists) {
-			npmError(w, http.StatusConflict, fmt.Sprintf("%s@%s already exists; overwrites are not allowed", pkg, rel.version))
+	conflict := func() {
+		npmError(w, http.StatusConflict, fmt.Sprintf("%s@%s already exists; overwrites are not allowed", pkg, rel.version))
+	}
+	// The manifest gates visibility, so it is written last; a tarball left by a
+	// failed publish is accepted again only with identical bytes.
+	hash, err := h.npmStore(r.Context(), remote.Name, stem+".tgz", rel.tarball, "application/octet-stream", false)
+	if errors.Is(err, database.ErrAlreadyExists) {
+		existing, getErr := h.db.GetLocalFile(r.Context(), remote.Name, stem+".tgz")
+		if getErr != nil || existing == nil {
+			npmError(w, http.StatusInternalServerError, fmt.Sprintf("read existing tarball: %v", getErr))
 			return
 		}
-		npmError(w, http.StatusInternalServerError, err.Error())
-		return
+		if existing.ContentHash != hash {
+			conflict()
+			return
+		}
+		err = nil
 	}
-	if err := h.npmStore(r.Context(), remote.Name, stem+".json", rel.manifest, "application/json", false); err != nil {
+	if err != nil {
 		npmError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
 	for _, tag := range rel.tags {
-		if err := h.npmStore(r.Context(), remote.Name, pkg+"/-/dist-tags/"+tag, []byte(rel.version), "text/plain", true); err != nil {
+		if _, err := h.npmStore(r.Context(), remote.Name, pkg+"/-/dist-tags/"+tag, []byte(rel.version), "text/plain", true); err != nil {
 			npmError(w, http.StatusInternalServerError, err.Error())
 			return
 		}
 	}
+	if _, err := h.npmStore(r.Context(), remote.Name, stem+".json", rel.manifest, "application/json", false); err != nil {
+		if errors.Is(err, database.ErrAlreadyExists) {
+			conflict()
+			return
+		}
+		npmError(w, http.StatusInternalServerError, err.Error())
+		return
+	}

 	slog.Info("local npm package published", "repo", remote.Name, "package", pkg, "version", rel.version)
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusCreated)
-	json.NewEncoder(w).Encode(map[string]any{"ok": true, "id": pkg})
+	_ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "id": pkg})
 }

-func (h *ProxyHandler) npmStore(ctx context.Context, repo, path string, data []byte, contentType string, mutable bool) error {
+func (h *ProxyHandler) npmStore(ctx context.Context, repo, path string, data []byte, contentType string, mutable bool) (string, error) {
 	result, err := h.cas.Store(ctx, bytes.NewReader(data), contentType)
 	if err != nil {
-		return err
+		return "", err
 	}
 	if err := h.db.UpsertBlob(ctx, result.ContentHash, result.S3Key, result.SizeBytes, contentType); err != nil {
-		return err
+		return "", err
 	}
 	if mutable {
-		return h.db.UpsertLocalFile(ctx, repo, path, result.ContentHash)
+		return result.ContentHash, h.db.UpsertLocalFile(ctx, repo, path, result.ContentHash)
 	}
-	return h.db.CreateLocalFile(ctx, repo, path, result.ContentHash)
+	return result.ContentHash, h.db.CreateLocalFile(ctx, repo, path, result.ContentHash)
 }

 func (h *ProxyHandler) readLocalBlob(ctx context.Context, contentHash string) ([]byte, error) {
@@ -236,7 +261,7 @@ func (h *ProxyHandler) serveNPMPackument(w http.ResponseWriter, r *http.Request,
 	w.Header().Set("Content-Type", "application/json")
 	w.Header().Set("X-Artifact-Source", "local")
 	w.WriteHeader(http.StatusOK)
-	json.NewEncoder(w).Encode(doc)
+	_ = json.NewEncoder(w).Encode(doc)
 }

 // buildNPMPackument returns nil when pkg has no published versions.
@@ -0,0 +1,201 @@
+package v1
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+
+	"git.unkin.net/unkin/artifactapi/internal/database"
+	"git.unkin.net/unkin/artifactapi/internal/storage"
+	"git.unkin.net/unkin/artifactapi/internal/testsupport"
+	"git.unkin.net/unkin/artifactapi/pkg/models"
+)
+
+var (
+	testDSN   string
+	testMinio testsupport.MinioConn
+)
+
+func TestMain(m *testing.M) {
+	ctx := context.Background()
+	dsn, termPG, err := testsupport.StartPostgres(ctx)
+	if err != nil {
+		os.Exit(m.Run())
+	}
+	conn, termMinio, err := testsupport.StartMinio(ctx)
+	if err != nil {
+		termPG()
+		os.Exit(m.Run())
+	}
+	testDSN, testMinio = dsn, conn
+	code := m.Run()
+	termMinio()
+	termPG()
+	os.Exit(code)
+}
+
+func npmTestServer(t *testing.T) (*ProxyHandler, *httptest.Server) {
+	t.Helper()
+	if testDSN == "" {
+		t.Skip("Docker unavailable")
+	}
+	db, err := database.New(testDSN)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(db.Close)
+	var store *storage.S3
+	for i := 0; i < 20; i++ {
+		if store, err = storage.NewS3(testMinio.Endpoint, testMinio.AccessKey, testMinio.SecretKey, "npm", false, ""); err == nil {
+			break
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	h := NewProxyHandler(nil, nil, db, store, nil)
+	r := chi.NewRouter()
+	r.Mount("/api/v1", h.Routes())
+	srv := httptest.NewServer(r)
+	t.Cleanup(srv.Close)
+	return h, srv
+}
+
+func npmRepo(t *testing.T, h *ProxyHandler, pt models.PackageType) string {
+	t.Helper()
+	name := fmt.Sprintf("npm-%s-%d", pt, time.Now().UnixNano())
+	if err := h.db.CreateRemote(context.Background(), &models.Remote{Name: name, PackageType: pt, RepoType: models.RepoTypeLocal}); err != nil {
+		t.Fatal(err)
+	}
+	return name
+}
+
+func npmDo(t *testing.T, method, url, body string) (int, string) {
+	t.Helper()
+	req, _ := http.NewRequest(method, url, strings.NewReader(body))
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	b, _ := io.ReadAll(resp.Body)
+	return resp.StatusCode, string(b)
+}
+
+func packumentVersions(t *testing.T, url string) map[string]any {
+	t.Helper()
+	code, body := npmDo(t, http.MethodGet, url, "")
+	if code != http.StatusOK {
+		t.Fatalf("GET %s = %d: %s", url, code, body)
+	}
+	var doc struct {
+		Versions map[string]any `json:"versions"`
+	}
+	if err := json.Unmarshal([]byte(body), &doc); err != nil {
+		t.Fatal(err)
+	}
+	return doc.Versions
+}
+
+func TestNPMPublishHandler(t *testing.T) {
+	h, srv := npmTestServer(t)
+	repo := npmRepo(t, h, models.PackageNPM)
+	base := srv.URL + "/api/v1/local/" + repo + "/"
+
+	t.Run("scoped name via %2f", func(t *testing.T) {
+		body := publishBody(t, "@acme/widget", []byte("scoped tgz"), publishOpts{})
+		if code, b := npmDo(t, http.MethodPut, base+"@acme%2fwidget", body); code != http.StatusCreated {
+			t.Fatalf("publish = %d: %s", code, b)
+		}
+		v := packumentVersions(t, base+"@acme%2fwidget")
+		dist := v["1.0.0"].(map[string]any)["dist"].(map[string]any)
+		tarball := dist["tarball"].(string)
+		if !strings.HasPrefix(tarball, base+"@acme/widget/-/widget-1.0.0.tgz") {
+			t.Fatalf("tarball = %s", tarball)
+		}
+		if code, b := npmDo(t, http.MethodGet, tarball, ""); code != http.StatusOK || b != "scoped tgz" {
+			t.Fatalf("tarball fetch = %d %q", code, b)
+		}
+	})
+
+	t.Run("republish", func(t *testing.T) {
+		if code, b := npmDo(t, http.MethodPut, base+"repub", publishBody(t, "repub", []byte("v1"), publishOpts{})); code != http.StatusCreated {
+			t.Fatalf("first publish = %d: %s", code, b)
+		}
+		if code, _ := npmDo(t, http.MethodPut, base+"repub", publishBody(t, "repub", []byte("different"), publishOpts{})); code != http.StatusConflict {
+			t.Errorf("different bytes = %d, want 409", code)
+		}
+		if code, _ := npmDo(t, http.MethodPut, base+"repub", publishBody(t, "repub", []byte("v1"), publishOpts{})); code != http.StatusConflict {
+			t.Errorf("identical complete republish = %d, want 409", code)
+		}
+	})
+
+	t.Run("retry after tarball-only write", func(t *testing.T) {
+		tgz := []byte("half published")
+		if _, err := h.npmStore(context.Background(), repo, "retry/-/retry-1.0.0.tgz", tgz, "application/octet-stream", false); err != nil {
+			t.Fatal(err)
+		}
+		if code, _ := npmDo(t, http.MethodGet, base+"retry", ""); code != http.StatusNotFound {
+			t.Fatalf("tarball-only version visible: %d", code)
+		}
+		if code, _ := npmDo(t, http.MethodPut, base+"retry", publishBody(t, "retry", []byte("other bytes"), publishOpts{})); code != http.StatusConflict {
+			t.Errorf("retry with different bytes = %d, want 409", code)
+		}
+		if code, b := npmDo(t, http.MethodPut, base+"retry", publishBody(t, "retry", tgz, publishOpts{})); code != http.StatusCreated {
+			t.Fatalf("retry with identical bytes = %d: %s", code, b)
+		}
+		if _, ok := packumentVersions(t, base+"retry")["1.0.0"]; !ok {
+			t.Error("retried version missing from packument")
+		}
+	})
+
+	t.Run("concurrent versions", func(t *testing.T) {
+		var wg sync.WaitGroup
+		codes := make([]int, 2)
+		for i, v := range []string{"1.0.0", "2.0.0"} {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				codes[i], _ = npmDo(t, http.MethodPut, base+"conc", publishBody(t, "conc", []byte("tgz "+v), publishOpts{version: v}))
+			}()
+		}
+		wg.Wait()
+		if codes[0] != http.StatusCreated || codes[1] != http.StatusCreated {
+			t.Fatalf("codes = %v", codes)
+		}
+		v := packumentVersions(t, base+"conc")
+		if _, ok := v["1.0.0"]; !ok {
+			t.Error("1.0.0 missing")
+		}
+		if _, ok := v["2.0.0"]; !ok {
+			t.Error("2.0.0 missing")
+		}
+	})
+
+	t.Run("body too large", func(t *testing.T) {
+		old := npmMaxPublishBytes
+		npmMaxPublishBytes = 64
+		defer func() { npmMaxPublishBytes = old }()
+		if code, _ := npmDo(t, http.MethodPut, base+"big", publishBody(t, "big", []byte("x"), publishOpts{})); code != http.StatusRequestEntityTooLarge {
+			t.Errorf("oversized body = %d, want 413", code)
+		}
+	})
+
+	t.Run("non-npm local", func(t *testing.T) {
+		generic := npmRepo(t, h, models.PackageGeneric)
+		if code, _ := npmDo(t, http.MethodPut, srv.URL+"/api/v1/local/"+generic+"/pkg", publishBody(t, "pkg", []byte("x"), publishOpts{})); code != http.StatusMethodNotAllowed {
+			t.Errorf("publish to generic local = %d, want 405", code)
+		}
+	})
+}
@@ -101,6 +101,8 @@ func TestParseNPMPublish(t *testing.T) {
 		"manifest name mismatch":    {manifestName: "other"},
 		"integrity mismatch":        {integrity: "sha512-AAAA"},
 		"path-like version":         {version: "../1.0.0"},
+		"non-semver version":        {version: "1..2"},
+		"leading-zero version":      {version: "01.0.0"},
 		"tag at another version":    {tags: map[string]string{"latest": "9.9.9"}},
 		"path-like tag":             {tags: map[string]string{"../x": "1.0.0"}},
 		"multiple versions at once": {extraVersion: true},