package handler import ( "context" "encoding/json" "fmt" "io" "net/http" "net/http/httptest" "net/url" "strings" "sync/atomic" "testing" "time" "github.com/git-pkgs/cooldown" registryclient "github.com/git-pkgs/registries/client" "github.com/git-pkgs/registries/fetch" ) func TestNPMContentAddressedTarball(t *testing.T) { for _, tc := range []struct { cacheMetadata, stream bool metadata, tarballs int64 }{ {false, false, 2, 1}, {false, true, 3, 2}, {true, false, 1, 1}, {true, true, 1, 2}, } { t.Run(fmt.Sprintf("metadata=%t/stream=%t", tc.cacheMetadata, tc.stream), func(t *testing.T) { var metadataCalls, tarballCalls atomic.Int64 upstream := npmContentAddressedRegistry(t, &metadataCalls, &tarballCalls) p, _, _, _ := setupTestProxy(t) p.CacheMetadata = tc.cacheMetadata p.MetadataTTL = time.Hour p.StreamArtifacts = tc.stream p.Cooldown = &cooldown.Config{Default: "7d"} p.HTTPClient = upstream.Client() fetcher := fetch.NewFetcher(fetch.WithHTTPClient(upstream.Client()), fetch.WithMaxRetries(0)) p.Fetcher = fetcher t.Cleanup(func() { _ = fetcher.Close() }) h := NewNPMHandler(p, "http://proxy.test", upstream.URL+"/registry") routes := http.StripPrefix("/npm", h.Routes()) metadata := httptest.NewRecorder() routes.ServeHTTP(metadata, httptest.NewRequest(http.MethodGet, "/npm/@example/widget", nil)) if metadata.Code != http.StatusOK { t.Fatalf("metadata status = %d: %s", metadata.Code, metadata.Body.String()) } tarball := npmVersionTarball(metadata.Body.Bytes(), testVersion100) if want := "http://proxy.test/npm/@example%2Fwidget/-/widget-1.0.0.tgz"; tarball != want { t.Fatalf("rewritten tarball = %q, want %q", tarball, want) } for i := range 2 { response := httptest.NewRecorder() routes.ServeHTTP(response, httptest.NewRequest(http.MethodGet, tarball, nil)) if response.Code != http.StatusOK || response.Body.String() != "tarball bytes" { t.Fatalf("download %d = %d %q", i, response.Code, response.Body.String()) } if !tc.stream { upstream.Close() } } if metadataCalls.Load() != tc.metadata || tarballCalls.Load() != tc.tarballs { t.Errorf("upstream requests: metadata=%d tarballs=%d, want %d %d", metadataCalls.Load(), tarballCalls.Load(), tc.metadata, tc.tarballs) } }) } } func npmContentAddressedRegistry(t *testing.T, metadataCalls, tarballCalls *atomic.Int64) *httptest.Server { t.Helper() var upstream *httptest.Server upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/registry/@example/widget": metadataCalls.Add(1) w.Header().Set("Content-Type", contentTypeJSON) _, _ = fmt.Fprintf(w, `{"name":"@example/widget","time":{"1.0.0":%q},"versions":{"1.0.0":{"dist":{"tarball":%q}}}}`, time.Now().Add(-30*24*time.Hour).Format(time.RFC3339), upstream.URL+"/registry/download/@example/widget/1.0.0/abc123?token=example") case "/registry/download/@example/widget/1.0.0/abc123": if r.URL.RawQuery != "token=example" { t.Errorf("tarball query = %q", r.URL.RawQuery) } tarballCalls.Add(1) _, _ = io.WriteString(w, "tarball bytes") default: t.Errorf("unexpected upstream request: %s", r.URL) http.NotFound(w, r) } })) t.Cleanup(upstream.Close) return upstream } func TestNPMTarballMetadataFallback(t *testing.T) { for _, body := range []string{`{"versions":{}}`, `{"versions":{"1.0.0":{"dist":{}}}}`} { t.Run(body, func(t *testing.T) { upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = io.WriteString(w, body) })) t.Cleanup(upstream.Close) p, _, _, fetcher := setupTestProxy(t) p.CacheMetadata = true p.MetadataTTL = time.Hour fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("package"))} h := NewNPMHandler(p, "http://proxy.test", upstream.URL) if _, _, err := p.FetchOrCacheMetadata(t.Context(), "npm", "@example/widget", upstream.URL); err != nil { t.Fatal(err) } upstream.Close() response := httptest.NewRecorder() h.Routes().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/@example/widget/-/widget-1.0.0.tgz", nil)) if response.Code != http.StatusOK { t.Fatalf("status = %d: %s", response.Code, response.Body.String()) } if want := upstream.URL + "/@example/widget/-/widget-1.0.0.tgz"; fetcher.fetchedURL != want { t.Errorf("fetched URL = %q, want %q", fetcher.fetchedURL, want) } }) } } // TestNPMTarballWithoutMetadataIsNotFetched checks that a tarball whose // packument can't be read, so its integrity is unknown, is neither fetched // nor cached. func TestNPMTarballWithoutMetadataIsNotFetched(t *testing.T) { for _, tc := range []struct { name string status int body string limit int64 }{ {name: "unavailable", status: http.StatusServiceUnavailable, body: "unavailable"}, {name: "over metadata_max_size", status: http.StatusOK, body: `{"versions":{"1.0.0":{"dist":{"integrity":"` + sha512SRI("package") + `"}}}}`, limit: 16}, } { t.Run(tc.name, func(t *testing.T) { upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(tc.status) _, _ = io.WriteString(w, tc.body) })) t.Cleanup(upstream.Close) p, _, _, fetcher := setupTestProxy(t) if tc.limit > 0 { p.MetadataMaxSize = tc.limit } fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("tampered"))} h := NewNPMHandler(p, "http://proxy.test", upstream.URL) response := httptest.NewRecorder() h.Routes().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/@example/widget/-/widget-1.0.0.tgz", nil)) if response.Code != http.StatusBadGateway { t.Fatalf("status = %d, want 502: %s", response.Code, response.Body.String()) } if fetcher.fetchedURL != "" { t.Errorf("fetched %q without the metadata to verify it", fetcher.fetchedURL) } cached, err := p.GetCachedArtifact(t.Context(), "npm", "@example/widget", "1.0.0", "widget-1.0.0.tgz") if err != nil { t.Fatal(err) } if cached != nil { _ = cached.Reader.Close() t.Error("tarball was cached without verification") } }) } } // serveEmptyPackuments answers every metadata request with a packument that // lists no versions, for tests about the tarball download itself. func serveEmptyPackuments(p *Proxy) { p.HTTPClient = &http.Client{Transport: pypiRoundTripFunc(func(r *http.Request) (*http.Response, error) { return pypiHTTPResponse(r, "application/json", `{"versions":{}}`), nil })} } func TestNPMTarballRejectsUnsafeMetadataURL(t *testing.T) { for _, target := range []string{ "https://outside.invalid/package.tgz", "//outside.invalid/package.tgz", "/registry/package.tgz", "UPSTREAM/outside/package.tgz", "UPSTREAM/registry-other/package.tgz", "UPSTREAM/registry/../outside/package.tgz", "UPSTREAM/registry/%2e%2e/outside/package.tgz", "UPSTREAM/registry/%252e%252e/outside/package.tgz", "UPSTREAM/registry/..%2foutside/package.tgz", "UPSTREAM/registry/..%5coutside/package.tgz", "UPSTREAM/registry/package.tgz#fragment", "USERINFO/registry/package.tgz", "SCHEME/registry/package.tgz", } { t.Run(target, func(t *testing.T) { var upstream *httptest.Server upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { address, _ := url.Parse(upstream.URL) address.User = url.UserPassword("user", "secret") tarball := strings.NewReplacer("UPSTREAM", upstream.URL, "USERINFO", address.String(), "SCHEME", strings.Replace(upstream.URL, "http:", "https:", 1)).Replace(target) _ = json.NewEncoder(w).Encode(map[string]any{"versions": map[string]any{testVersion100: map[string]any{"dist": map[string]string{"tarball": tarball}}}}) })) t.Cleanup(upstream.Close) p, _, _, fetcher := setupTestProxy(t) h := NewNPMHandler(p, "http://proxy.test", upstream.URL+"/registry") response := httptest.NewRecorder() h.Routes().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/widget/-/widget-1.0.0.tgz", nil)) if response.Code != http.StatusBadGateway || fetcher.fetchCalled { t.Errorf("status = %d, fetched = %t; body: %s", response.Code, fetcher.fetchCalled, response.Body.String()) } }) } } // TestNPMTarballCooldownRefetchesStalePackument checks that a tarball for a // version the cached packument does not list yet is decided on the packument // upstream serves now, not let through for want of a publish time. func TestNPMTarballCooldownRefetchesStalePackument(t *testing.T) { for _, tc := range []struct { name string published time.Time down bool want int }{ {"recent", time.Now().Add(-time.Hour), false, http.StatusNotFound}, // Without upstream the version can't be checked, so it isn't served. {"upstream down", time.Now().Add(-time.Hour), true, http.StatusBadGateway}, // The refetched entry's integrity is checked, so a tarball that // doesn't match it is refused rather than served unverified. {"old", time.Now().Add(-30 * 24 * time.Hour), false, http.StatusBadGateway}, } { t.Run(tc.name, func(t *testing.T) { var released atomic.Bool upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { old := time.Now().Add(-60 * 24 * time.Hour).Format(time.RFC3339) if released.Load() && tc.down { w.WriteHeader(http.StatusServiceUnavailable) return } if !released.Load() { _, _ = fmt.Fprintf(w, `{"name":"widget","time":{"1.0.0":%q},"versions":{"1.0.0":{"dist":{}}}}`, old) return } _, _ = fmt.Fprintf(w, `{"name":"widget","time":{"1.0.0":%q,"2.0.0":%q},"versions":{"1.0.0":{"dist":{}},"2.0.0":{"dist":{"integrity":%q}}}}`, old, tc.published.Format(time.RFC3339), sha512SRI("package")) })) t.Cleanup(upstream.Close) p, _, _, fetcher := setupTestProxy(t) p.CacheMetadata = true p.MetadataTTL = time.Hour p.Cooldown = &cooldown.Config{Default: "7d"} p.HTTPClient = upstream.Client() fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("tampered"))} h := NewNPMHandler(p, "http://proxy.test", upstream.URL) metadata := httptest.NewRecorder() h.Routes().ServeHTTP(metadata, httptest.NewRequest(http.MethodGet, "/widget", nil)) if metadata.Code != http.StatusOK { t.Fatalf("metadata status = %d: %s", metadata.Code, metadata.Body.String()) } released.Store(true) response := httptest.NewRecorder() h.Routes().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/widget/-/widget-2.0.0.tgz", nil)) if response.Code != tc.want { t.Fatalf("status = %d, want %d: %s", response.Code, tc.want, response.Body.String()) } }) } } // widgetPackument is a packument listing only widget 1.0.0, published long // ago. func widgetPackument() string { old := time.Now().Add(-60 * 24 * time.Hour).Format(time.RFC3339) return fmt.Sprintf(`{"name":"widget","time":{"1.0.0":%q},"versions":{"1.0.0":{"dist":{}}}}`, old) } // cooldownNPMHandler returns an npm handler for upstream with cooldown on // and a tarball fetcher that would serve anything asked of it. func cooldownNPMHandler(t *testing.T, upstream *httptest.Server, ttl time.Duration) *NPMHandler { t.Helper() p, _, _, fetcher := setupTestProxy(t) p.CacheMetadata = true p.MetadataTTL = ttl p.Cooldown = &cooldown.Config{Default: "7d"} p.HTTPClient = upstream.Client() fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("package"))} return NewNPMHandler(p, "http://proxy.test", upstream.URL) } // TestNPMTarballCooldownStalePackument checks that a stale packument served // because upstream failed doesn't let through a version it doesn't list, // and that the upstream failure decides the response. func TestNPMTarballCooldownStalePackument(t *testing.T) { for _, tc := range []struct { name string status int limited bool want int }{ {"upstream down", http.StatusServiceUnavailable, false, http.StatusBadGateway}, {"not found", http.StatusNotFound, false, http.StatusNotFound}, {"rate limited", http.StatusTooManyRequests, true, http.StatusServiceUnavailable}, } { t.Run(tc.name, func(t *testing.T) { var down atomic.Bool upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { if down.Load() { w.Header().Set("Retry-After", "60") w.WriteHeader(tc.status) return } _, _ = io.WriteString(w, widgetPackument()) })) t.Cleanup(upstream.Close) // With no TTL the packument is read in memory and every read // asks upstream. h := cooldownNPMHandler(t, upstream, 0) if tc.limited { gate := registryclient.NewRateLimitGate(http.DefaultTransport) h.proxy.HTTPClient = &http.Client{Transport: gate} h.proxy.RateLimits = gate } metadata := httptest.NewRecorder() h.Routes().ServeHTTP(metadata, httptest.NewRequest(http.MethodGet, "/widget", nil)) if metadata.Code != http.StatusOK { t.Fatalf("metadata status = %d: %s", metadata.Code, metadata.Body.String()) } down.Store(true) response := httptest.NewRecorder() h.Routes().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/widget/-/widget-2.0.0.tgz", nil)) if response.Code != tc.want { t.Fatalf("status = %d, want %d: %s", response.Code, tc.want, response.Body.String()) } if tc.limited && response.Header().Get("Retry-After") == "" { t.Error("no Retry-After on a rate-limited response") } }) } } // TestNPMTarballCooldownFetchesMissingVersionOnce checks that a packument // just fetched from upstream isn't fetched again because it doesn't list the // requested version. func TestNPMTarballCooldownFetchesMissingVersionOnce(t *testing.T) { var requests atomic.Int32 upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { requests.Add(1) _, _ = io.WriteString(w, widgetPackument()) })) t.Cleanup(upstream.Close) h := cooldownNPMHandler(t, upstream, time.Hour) response := httptest.NewRecorder() h.Routes().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/widget/-/widget-99.0.0.tgz", nil)) if n := requests.Load(); n != 1 { t.Errorf("upstream packument requests = %d, want 1", n) } } // TestNPMTarballCooldownRefreshIgnoresSharedStale checks that a refetch of a // stored packument lacking the version, joining another caller's fetch that // falls back to the stored copy, doesn't take that copy as upstream's answer. func TestNPMTarballCooldownRefreshIgnoresSharedStale(t *testing.T) { var blocked atomic.Bool arrived := make(chan struct{}, 1) release := make(chan struct{}) upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if !blocked.Load() { _, _ = io.WriteString(w, widgetPackument()) return } arrived <- struct{}{} select { case <-release: case <-r.Context().Done(): } w.WriteHeader(http.StatusServiceUnavailable) })) t.Cleanup(upstream.Close) h := cooldownNPMHandler(t, upstream, time.Hour) p := h.proxy metadata := httptest.NewRecorder() h.Routes().ServeHTTP(metadata, httptest.NewRequest(http.MethodGet, "/widget", nil)) if metadata.Code != http.StatusOK { t.Fatalf("metadata status = %d: %s", metadata.Code, metadata.Body.String()) } registryURL, cacheKey := h.sourceFor("widget") packumentURL := registryURL + "/widget" accept := h.packumentAccept() row, err := p.DB.GetMetadataCache("npm", cacheKey) if err != nil || row == nil { t.Fatalf("no stored packument row: %v", err) } setFetchedAt := func(at time.Time) { t.Helper() if ok, err := p.DB.RefreshMetadataCache("npm", cacheKey, row.ContentDigest.String, at); err != nil || !ok { t.Fatalf("setting fetched_at: %t, %v", ok, err) } } // Another caller revalidates the expired copy, and upstream holds it. setFetchedAt(time.Now().Add(-2 * time.Hour)) blocked.Store(true) other := make(chan error, 1) go func() { _, err := p.storedMetadataDocument(context.Background(), "npm", cacheKey, packumentURL, accept, h.packumentStreamValidator()) other <- err }() <-arrived // The tarball request reads the copy as within its TTL and refetches it, // joining that fetch. setFetchedAt(time.Now()) response := httptest.NewRecorder() done := make(chan struct{}) go func() { defer close(done) h.Routes().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/widget/-/widget-2.0.0.tgz", nil)) }() waitForMetadataWaiters(t, p, metadataCoalesceKey("npm", cacheKey, packumentURL, accept, "", false)+"\x00stored", 1) close(release) <-done if err := <-other; err != nil { t.Fatalf("other caller: %v", err) } if response.Code != http.StatusBadGateway { t.Fatalf("status = %d, want %d: %s", response.Code, http.StatusBadGateway, response.Body.String()) } }