From 3456b394beb793db9a5926e26d565c8ccbc7eb52 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Sun, 18 Oct 2026 21:30:21 +0200 Subject: [PATCH 1/6] Cargo, gem, hex, pub, conda and helm now check cooldown on artifact downloads too, so a version inside the window is withheld even when it was cached before. Publish times are stored in UTC. Fixes #521. --- internal/database/database_test.go | 41 +++ internal/database/queries.go | 4 +- internal/handler/cargo.go | 28 ++ internal/handler/conda.go | 44 +++ internal/handler/download_cooldown.go | 82 +++++ internal/handler/download_cooldown_test.go | 347 +++++++++++++++++++++ internal/handler/filename_download.go | 11 + internal/handler/gem.go | 30 ++ internal/handler/helm.go | 54 ++++ internal/handler/hex.go | 25 ++ internal/handler/pub.go | 34 ++ 11 files changed, 698 insertions(+), 2 deletions(-) create mode 100644 internal/handler/download_cooldown.go create mode 100644 internal/handler/download_cooldown_test.go diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 9ac4140d..91734e11 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -1204,3 +1204,44 @@ func TestGetFetchedVersionPURLs(t *testing.T) { } }) } + +// Publish times are written in UTC whatever zone the caller's time is in, so +// a TIMESTAMP column without a zone reads back the same instant. +func TestVersionPublishedAtIsUTC(t *testing.T) { + tokyo, err := time.LoadLocation("Asia/Tokyo") + if err != nil { + t.Skipf("time zone data unavailable: %v", err) + } + prev := time.Local + time.Local = tokyo + t.Cleanup(func() { time.Local = prev }) + + runWithBothDatabases(t, func(t *testing.T, db *DB) { + written := time.Date(2026, 10, 10, 21, 0, 1, 0, tokyo) + want := written.UTC() + check := func(purl string) { + t.Helper() + got, err := db.GetVersionByPURL(purl) + if err != nil || got == nil || !got.PublishedAt.Valid { + t.Fatalf("GetVersionByPURL(%s) = %+v, %v", purl, got, err) + } + if !got.PublishedAt.Time.Equal(want) || got.PublishedAt.Time.Location() != time.UTC { + t.Errorf("%s PublishedAt read back as %v, want %v", purl, got.PublishedAt.Time, want) + } + } + + if err := db.SetVersionPublishedAt("pkg:cargo/demo@1.0.0", "pkg:cargo/demo", written); err != nil { + t.Fatal(err) + } + check("pkg:cargo/demo@1.0.0") + + if err := db.UpsertVersion(&Version{ + PURL: "pkg:cargo/demo@2.0.0", + PackagePURL: "pkg:cargo/demo", + PublishedAt: sql.NullTime{Time: written, Valid: true}, + }); err != nil { + t.Fatal(err) + } + check("pkg:cargo/demo@2.0.0") + }) +} diff --git a/internal/database/queries.go b/internal/database/queries.go index 23d55df8..86f31ef9 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -185,7 +185,7 @@ func (db *DB) upsertVersion(exec execer, v *Version) error { _, err := exec.Exec(query, v.PURL, v.PackagePURL, v.License, v.Integrity, - v.PublishedAt, v.Yanked, v.EnrichedAt, now, now, + inUTC(v.PublishedAt), v.Yanked, v.EnrichedAt, now, now, ) if err != nil { return fmt.Errorf("upserting version: %w", err) @@ -218,7 +218,7 @@ func (db *DB) SetVersionPublishedAt(versionPURL, packagePURL string, publishedAt ` } - _, err := db.Exec(query, versionPURL, packagePURL, publishedAt, now, now) + _, err := db.Exec(query, versionPURL, packagePURL, publishedAt.UTC(), now, now) if err != nil { return fmt.Errorf("setting version publish time: %w", err) } diff --git a/internal/handler/cargo.go b/internal/handler/cargo.go index 0c265519..43bcbc4d 100644 --- a/internal/handler/cargo.go +++ b/internal/handler/cargo.go @@ -2,6 +2,7 @@ package handler import ( "bufio" + "bytes" "encoding/json" "errors" "fmt" @@ -186,6 +187,26 @@ func (h *CargoHandler) applyCooldownFiltering(downstreamResponse http.ResponseWr } } +// publishTime reads a crate version's pubtime from the sparse index, sharing +// the index cache with handleIndex. Entries published before crates.io +// recorded pubtime have none. +func (h *CargoHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, error) { + upstreamURL := fmt.Sprintf("%s/%s", h.indexURL, h.buildIndexPath(name)) + body, _, err := h.proxy.FetchOrCacheMetadata(r.Context(), "cargo", name, upstreamURL, "text/plain") + if err != nil { + return time.Time{}, false, err + } + for line := range bytes.SplitSeq(body, []byte("\n")) { + if vers, _, _ := lookupJSONString(line, "vers"); vers != version { + continue + } + pubtime, _, _ := lookupJSONString(line, "pubtime") + published, err := time.Parse(time.RFC3339, pubtime) + return published, err == nil, nil + } + return time.Time{}, false, nil +} + // buildIndexPath builds the sparse index path for a crate name. func (h *CargoHandler) buildIndexPath(name string) string { name = strings.ToLower(name) @@ -217,6 +238,13 @@ func (h *CargoHandler) handleDownload(w http.ResponseWriter, r *http.Request) { h.proxy.Logger.Debug("cargo download request", "crate", name, "version", version, "filename", filename) + if h.proxy.withholdInCooldown(w, downloadVersion{ + ecosystem: "cargo", policyName: name, storedName: name, version: version, + lookup: func() (time.Time, bool, error) { return h.publishTime(r, name, version) }, + }) { + return + } + downloadURL := fmt.Sprintf( "%s/%s/%s", h.downloadURL, diff --git a/internal/handler/conda.go b/internal/handler/conda.go index 8581c888..22ec603a 100644 --- a/internal/handler/conda.go +++ b/internal/handler/conda.go @@ -139,6 +139,13 @@ func (h *CondaHandler) handleDownload(w http.ResponseWriter, r *http.Request) { h.proxy.Logger.Debug("conda download request", "channel", channel, "arch", arch, "name", name, "version", version) + if h.proxy.withholdInCooldown(w, downloadVersion{ + ecosystem: "conda", policyName: name, storedName: packageName, version: version, + lookup: func() (time.Time, bool, error) { return h.publishTime(r, channel, arch, filename) }, + }) { + return + } + upstreamURL := h.upstreamURL + r.URL.Path // The same filename can hold different bytes in each subdir, so the // subdir is part of the cache key. @@ -531,6 +538,43 @@ func (h *CondaHandler) metadataCacheKey(urlPath string) string { return hashedCacheKey(h.upstreamURL, urlPath) } +// publishTime reads a package file's timestamp from the subdir's +// repodata.json. It shares proxyCached's cache entry and encoding, so the +// repodata a client just fetched is not downloaded again, and reads the one +// entry in place rather than decoding the whole index. +func (h *CondaHandler) publishTime(r *http.Request, channel, arch, filename string) (time.Time, bool, error) { + repodataPath := "/" + channel + "/" + arch + "/repodata.json" + cacheKey := h.metadataCacheKey(repodataPath) + body, _, encoding, err := h.proxy.fetchOrCacheMetadata(r.Context(), "conda", cacheKey, h.upstreamURL+repodataPath, "gzip", nil, "*/*") + if err != nil { + return time.Time{}, false, err + } + if strings.EqualFold(strings.TrimSpace(encoding), "gzip") { + decoded, err := gzip.NewReader(bytes.NewReader(body)) + if err != nil { + return time.Time{}, false, err + } + body, err = h.proxy.ReadMetadata(decoded) + if err != nil { + return time.Time{}, false, err + } + } + + key := "packages" + if strings.HasSuffix(filename, ".conda") { + key = "packages.conda" + } + raw, err := lookupJSON(body, key, filename, "timestamp") + if err != nil || len(raw) == 0 { + return time.Time{}, false, err + } + var ts float64 + if err := json.Unmarshal(raw, &ts); err != nil || ts == 0 { + return time.Time{}, false, nil + } + return time.Unix(int64(ts)/condaTimestampDivisor, 0), true, nil +} + // proxyCached forwards a metadata request with caching. func (h *CondaHandler) proxyCached(w http.ResponseWriter, r *http.Request) { cacheKey := h.metadataCacheKey(r.URL.Path) diff --git a/internal/handler/download_cooldown.go b/internal/handler/download_cooldown.go new file mode 100644 index 00000000..94554d3b --- /dev/null +++ b/internal/handler/download_cooldown.go @@ -0,0 +1,82 @@ +package handler + +import ( + "net/http" + "time" +) + +// publishTimeLookup reads a version's publish time from upstream metadata. ok +// is false when the metadata has no usable time for the version. +type publishTimeLookup func() (published time.Time, ok bool, err error) + +// downloadVersion names the version an artifact download is for, so its +// cooldown can be checked before the artifact is served. +type downloadVersion struct { + ecosystem string + // policyName is the package name the cooldown policy is keyed by, + // canonicalPackagePURL(ecosystem, policyName). + policyName string + // storedName and version name the versions row the publish time is kept + // in, the same names the artifact is cached under. An empty storedName + // keeps no time, and every check asks lookup. + storedName string + version string + lookup publishTimeLookup +} + +// withholdInCooldown reports whether a download is inside the cooldown +// window, and if so answers it with a 404. Filtering metadata is not enough on +// its own: lockfiles record artifact URLs, so clients reach the download path +// without reading the metadata first, and a cached artifact would be served +// even after the policy was raised. +// +// A publish time never changes, so the stored versions row is read first and +// lookup is only asked for a version the proxy has no time for yet; the time +// it finds is stored. A version with no usable publish time is allowed +// through, matching how the metadata filters treat it. +func (p *Proxy) withholdInCooldown(w http.ResponseWriter, d downloadVersion) bool { + if !p.versionInCooldown(d) { + return false + } + p.Logger.Info("cooldown: withholding "+d.ecosystem+" download", + "name", d.policyName, "version", d.version) + http.Error(w, ErrVersionNotFound.Error(), http.StatusNotFound) + return true +} + +func (p *Proxy) versionInCooldown(d downloadVersion) bool { + if p.Cooldown == nil || !p.Cooldown.Enabled() { + return false + } + policyKey := canonicalPackagePURL(d.ecosystem, d.policyName) + if p.Cooldown.For(d.ecosystem, policyKey) <= 0 { + return false + } + + var versionPURL string + if d.storedName != "" { + versionPURL = canonicalVersionPURL(d.ecosystem, d.storedName, d.version) + if ver, err := p.DB.GetVersionByPURL(versionPURL); err == nil && ver != nil && ver.PublishedAt.Valid { + return !p.Cooldown.IsAllowed(d.ecosystem, policyKey, ver.PublishedAt.Time) + } + } + + published, ok, err := d.lookup() + if err != nil { + p.Logger.Warn("cooldown: could not read publish time for download check", + "ecosystem", d.ecosystem, "name", d.policyName, "version", d.version, "error", err) + return false + } + if !ok { + return false + } + + if versionPURL != "" { + if err := p.DB.SetVersionPublishedAt(versionPURL, canonicalPackagePURL(d.ecosystem, d.storedName), published); err != nil { + p.Logger.Warn("cooldown: could not store publish time", + "ecosystem", d.ecosystem, "name", d.policyName, "version", d.version, "error", err) + } + } + + return !p.Cooldown.IsAllowed(d.ecosystem, policyKey, published) +} diff --git a/internal/handler/download_cooldown_test.go b/internal/handler/download_cooldown_test.go new file mode 100644 index 00000000..9b791bed --- /dev/null +++ b/internal/handler/download_cooldown_test.go @@ -0,0 +1,347 @@ +package handler + +import ( + "compress/gzip" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/git-pkgs/cooldown" +) + +// Versions every download cooldown fixture lists: one published before the +// window, one inside it, and one without a publish time. +const ( + cooldownOldVersion = "1.0.0" + cooldownRecentVersion = "2.0.0" + cooldownUntimed = "3.0.0" +) + +// downloadCooldownCase drives one ecosystem's artifact endpoint. The package +// is always named demo, so the policy key is pkg:{ecosystem}/demo. +type downloadCooldownCase struct { + ecosystem string + // storedName is the name the artifact is cached under; empty when the + // ecosystem keeps no publish time in the versions table. + storedName string + // artifactVersion maps a release to the version the artifact is cached + // under, and filename and path to its file and download path. + artifactVersion func(version string) string + filename func(version string) string + path func(version string) string + // metadata serves the upstream document holding the publish times; a + // zero time is listed without one. + metadata func(w http.ResponseWriter, r *http.Request, times map[string]time.Time) + handler func(proxy *Proxy, upstreamURL string) http.Handler +} + +func sameVersion(version string) string { return version } + +func chartContent(version string) string { return "chart " + version } + +func downloadCooldownCases() []downloadCooldownCase { + return []downloadCooldownCase{ + { + ecosystem: "cargo", + storedName: "demo", + artifactVersion: sameVersion, + filename: func(v string) string { return "demo-" + v + ".crate" }, + path: func(v string) string { return "/crates/demo/" + v + "/download" }, + metadata: cargoCooldownMetadata, + handler: func(p *Proxy, u string) http.Handler { return NewCargoHandler(p, "http://proxy", u, u).Routes() }, + }, + { + ecosystem: "gem", + storedName: "demo", + // Platform gems carry the platform in the filename and the + // cached version, but not in the versions API number. + artifactVersion: func(v string) string { return v + "-x86_64-linux" }, + filename: func(v string) string { return "demo-" + v + "-x86_64-linux.gem" }, + path: func(v string) string { return "/gems/demo-" + v + "-x86_64-linux.gem" }, + metadata: gemCooldownMetadata, + handler: func(p *Proxy, u string) http.Handler { return NewGemHandlerWithUpstream(p, "http://proxy", u).Routes() }, + }, + { + ecosystem: "hex", + storedName: "demo", + artifactVersion: sameVersion, + filename: func(v string) string { return "demo-" + v + ".tar" }, + path: func(v string) string { return "/tarballs/demo-" + v + ".tar" }, + metadata: hexCooldownMetadata, + handler: func(p *Proxy, u string) http.Handler { + return NewHexHandlerWithUpstreams(p, "http://proxy", u, u).Routes() + }, + }, + { + ecosystem: "pub", + storedName: "demo", + artifactVersion: sameVersion, + filename: func(v string) string { return "demo-" + v + ".tar.gz" }, + path: func(v string) string { return "/packages/demo/versions/" + v + ".tar.gz" }, + metadata: pubCooldownMetadata, + handler: func(p *Proxy, u string) http.Handler { return NewPubHandlerWithUpstream(p, "http://proxy", u).Routes() }, + }, + { + // Conda caches under {channel}/{name}, but the policy is keyed by + // the name alone, as the repodata filter keys it. + ecosystem: "conda", + storedName: "conda-forge/demo", + artifactVersion: sameVersion, + filename: func(v string) string { return "noarch/demo-" + v + "-py_0.conda" }, + path: func(v string) string { return "/conda-forge/noarch/demo-" + v + "-py_0.conda" }, + metadata: condaCooldownMetadata, + handler: func(p *Proxy, u string) http.Handler { + return NewCondaHandlerWithUpstream(p, "http://proxy", u).Routes() + }, + }, + { + // Helm caches charts under the repository and digest, so the + // chart name and time come from the index on every check. + ecosystem: "helm", + artifactVersion: func(v string) string { return sha256Hex(chartContent(v)) }, + filename: func(v string) string { return "demo-" + v + ".tgz" }, + path: func(v string) string { return "/stable/charts/" + sha256Hex(chartContent(v)) + "/demo-" + v + ".tgz" }, + metadata: helmCooldownMetadata, + handler: func(p *Proxy, u string) http.Handler { + return NewHelmHandler(p, "http://proxy", map[string]string{"stable": u + "/charts"}).Routes() + }, + }, + } +} + +func cargoCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[string]time.Time) { + if r.URL.Path != "/de/mo/demo" { + http.NotFound(w, r) + return + } + for version, published := range times { + line := map[string]string{"name": "demo", "vers": version} + if !published.IsZero() { + line["pubtime"] = published.Format(time.RFC3339) + } + encoded, _ := json.Marshal(line) + _, _ = fmt.Fprintf(w, "%s\n", encoded) + } +} + +func gemCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[string]time.Time) { + if r.URL.Path != "/api/v1/versions/demo.json" { + http.NotFound(w, r) + return + } + var versions []gemVersion + for version, published := range times { + v := gemVersion{Number: version, Platform: "x86_64-linux"} + if !published.IsZero() { + v.CreatedAt = published.Format(time.RFC3339) + } + versions = append(versions, v) + } + _ = json.NewEncoder(w).Encode(versions) +} + +func hexCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[string]time.Time) { + if r.URL.Path != "/api/packages/demo" { + http.NotFound(w, r) + return + } + var pkg hexPackageAPI + for version, published := range times { + release := hexRelease{Version: version} + if !published.IsZero() { + release.InsertedAt = published.Format(time.RFC3339Nano) + } + pkg.Releases = append(pkg.Releases, release) + } + _ = json.NewEncoder(w).Encode(pkg) +} + +func pubCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[string]time.Time) { + if r.URL.Path != "/api/packages/demo" { + http.NotFound(w, r) + return + } + var versions []map[string]string + for version, published := range times { + v := map[string]string{"version": version} + if !published.IsZero() { + v["published"] = published.Format(time.RFC3339) + } + versions = append(versions, v) + } + _ = json.NewEncoder(w).Encode(map[string]any{"name": "demo", "versions": versions}) +} + +func condaCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[string]time.Time) { + if r.URL.Path != "/conda-forge/noarch/repodata.json" { + http.NotFound(w, r) + return + } + packages := map[string]map[string]any{} + for version, published := range times { + entry := map[string]any{"name": "demo", "version": version} + if !published.IsZero() { + entry["timestamp"] = published.UnixMilli() + } + packages["demo-"+version+"-py_0.conda"] = entry + } + // The download check asks for gzip, as proxyCached does. + w.Header().Set(headerContentEncoding, "gzip") + gz := gzip.NewWriter(w) + _ = json.NewEncoder(gz).Encode(map[string]any{"packages.conda": packages}) + _ = gz.Close() +} + +func helmCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[string]time.Time) { + if r.URL.Path != "/charts/index.yaml" { + http.NotFound(w, r) + return + } + _, _ = fmt.Fprintln(w, "apiVersion: v1\nentries:\n demo:") + for version, published := range times { + _, _ = fmt.Fprintf(w, " - digest: %s\n urls: [demo-%s.tgz]\n version: %s\n", + sha256Hex(chartContent(version)), version, version) + if !published.IsZero() { + _, _ = fmt.Fprintf(w, " created: %s\n", published.Format(time.RFC3339)) + } + } +} + +// seedCachedArtifact caches the release's artifact the way an earlier, +// cooldown-free download would have. +func (c downloadCooldownCase) seedCachedArtifact(t *testing.T, proxy *Proxy, version string) { + t.Helper() + db, store := proxy.DB, proxy.Storage.(*mockStorage) + name := c.storedName + content := "artifact " + version + if c.ecosystem == "helm" { + name, content = "stable", chartContent(version) + } + seedPackage(t, db, store, c.ecosystem, name, c.artifactVersion(version), c.filename(version), content) +} + +// TestDownloadCooldownWithholdsCachedArtifacts requests each ecosystem's +// artifact endpoint for a version that was cached before the cooldown was +// raised. A version inside the window must be withheld, from the cache too; +// one past the window, or without a known publish time, is served. +func TestDownloadCooldownWithholdsCachedArtifacts(t *testing.T) { + now := time.Now().Truncate(time.Second) + times := map[string]time.Time{ + cooldownOldVersion: now.Add(-30 * 24 * time.Hour), + cooldownRecentVersion: now.Add(-time.Hour), + cooldownUntimed: {}, + } + statuses := map[string]int{ + cooldownOldVersion: http.StatusOK, + cooldownRecentVersion: http.StatusNotFound, + cooldownUntimed: http.StatusOK, + } + + for _, c := range downloadCooldownCases() { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c.metadata(w, r, times) + })) + t.Cleanup(upstream.Close) + + for _, version := range []string{cooldownOldVersion, cooldownRecentVersion, cooldownUntimed} { + t.Run(c.ecosystem+"/"+version, func(t *testing.T) { + proxy, db, _, fetcher := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + fetcher.fetchErr = errors.New("artifact must be served from the cache") + c.seedCachedArtifact(t, proxy, version) + proxy.Cooldown = &cooldown.Config{Packages: map[string]string{ + canonicalPackagePURL(c.ecosystem, "demo"): "7d", + }} + + w := httptest.NewRecorder() + c.handler(proxy, upstream.URL).ServeHTTP(w, httptest.NewRequest(http.MethodGet, c.path(version), nil)) + + if w.Code != statuses[version] { + t.Fatalf("status = %d, want %d (body %q)", w.Code, statuses[version], w.Body.String()) + } + if w.Code == http.StatusOK && w.Body.String() == "" { + t.Error("served an empty body") + } + if c.storedName == "" || times[version].IsZero() { + return + } + ver, err := db.GetVersionByPURL(canonicalVersionPURL(c.ecosystem, c.storedName, c.artifactVersion(version))) + if err != nil || ver == nil || !ver.PublishedAt.Valid { + t.Fatalf("publish time not stored: %+v, %v", ver, err) + } + if got, want := ver.PublishedAt.Time, times[version]; !got.Equal(want) { + t.Errorf("stored publish time = %v, want %v", got, want) + } + }) + } + } +} + +// TestDownloadCooldownUsesStoredPublishTime checks that a stored publish +// time decides the download without asking the upstream, which here fails. +func TestDownloadCooldownUsesStoredPublishTime(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "unavailable", http.StatusServiceUnavailable) + })) + t.Cleanup(upstream.Close) + + for _, c := range downloadCooldownCases() { + if c.storedName == "" { + continue + } + t.Run(c.ecosystem, func(t *testing.T) { + proxy, db, _, fetcher := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + fetcher.fetchErr = errors.New("artifact must be served from the cache") + c.seedCachedArtifact(t, proxy, cooldownRecentVersion) + versionPURL := canonicalVersionPURL(c.ecosystem, c.storedName, c.artifactVersion(cooldownRecentVersion)) + packagePURL := canonicalPackagePURL(c.ecosystem, c.storedName) + if err := db.SetVersionPublishedAt(versionPURL, packagePURL, time.Now().Add(-time.Hour)); err != nil { + t.Fatalf("seeding publish time: %v", err) + } + proxy.Cooldown = &cooldown.Config{Default: "7d"} + + w := httptest.NewRecorder() + c.handler(proxy, upstream.URL).ServeHTTP(w, httptest.NewRequest(http.MethodGet, c.path(cooldownRecentVersion), nil)) + if w.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404 (body %q)", w.Code, w.Body.String()) + } + }) + } +} + +// TestDownloadCooldownOtherPackagesSkipLookup checks that a download outside +// every cooldown rule is served without reading upstream metadata. Helm +// needs the index to learn the chart name, so it is left out. +func TestDownloadCooldownOtherPackagesSkipLookup(t *testing.T) { + for _, c := range downloadCooldownCases() { + if c.storedName == "" { + continue + } + t.Run(c.ecosystem, func(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Errorf("unexpected upstream request %s", r.URL.Path) + http.Error(w, "unexpected", http.StatusInternalServerError) + })) + t.Cleanup(upstream.Close) + proxy, _, _, fetcher := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + fetcher.fetchErr = errors.New("artifact must be served from the cache") + c.seedCachedArtifact(t, proxy, cooldownRecentVersion) + proxy.Cooldown = &cooldown.Config{Packages: map[string]string{ + canonicalPackagePURL(c.ecosystem, "other"): "7d", + }} + + w := httptest.NewRecorder() + c.handler(proxy, upstream.URL).ServeHTTP(w, httptest.NewRequest(http.MethodGet, c.path(cooldownRecentVersion), nil)) + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200 (body %q)", w.Code, strings.TrimSpace(w.Body.String())) + } + }) + } +} diff --git a/internal/handler/filename_download.go b/internal/handler/filename_download.go index 153c0e6e..dbc0e641 100644 --- a/internal/handler/filename_download.go +++ b/internal/handler/filename_download.go @@ -3,6 +3,7 @@ package handler import ( "net/http" "strings" + "time" ) type filenameDownload struct { @@ -12,6 +13,9 @@ type filenameDownload struct { parseErr string fetchErr string parse func(string) (name, version string) + // publishTime reads a version's publish time from upstream metadata for + // the download-time cooldown check. + publishTime func(r *http.Request, name, version string) (time.Time, bool, error) } func (p *Proxy) handleFilenameDownload(w http.ResponseWriter, r *http.Request, d filenameDownload) { @@ -30,6 +34,13 @@ func (p *Proxy) handleFilenameDownload(w http.ResponseWriter, r *http.Request, d p.Logger.Debug(d.ecosystem+" download request", "name", name, "version", version, "filename", filename) + if p.withholdInCooldown(w, downloadVersion{ + ecosystem: d.ecosystem, policyName: name, storedName: name, version: version, + lookup: func() (time.Time, bool, error) { return d.publishTime(r, name, version) }, + }) { + return + } + downloadURL := d.upstreamURL + r.URL.Path result, err := p.GetOrFetchArtifactFromURL( r.Context(), d.ecosystem, name, version, filename, downloadURL, diff --git a/internal/handler/gem.go b/internal/handler/gem.go index 6e575598..009e0f51 100644 --- a/internal/handler/gem.go +++ b/internal/handler/gem.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "net/http" + "net/url" "regexp" "strings" "time" @@ -82,6 +83,7 @@ func (h *GemHandler) handleDownload(w http.ResponseWriter, r *http.Request) { parseErr: "could not parse gem filename", fetchErr: "failed to fetch gem", parse: h.parseGemFilename, + publishTime: h.publishTime, }) } @@ -304,6 +306,34 @@ func parseGemVersions(body []byte) (publishTimes, error) { return times, nil } +// publishTime reads a gem version's created_at from the versions API. The +// download check caches the response, unlike fetchFilteredVersions, so each +// download does not cost an upstream request. version carries the platform +// the way gem filenames do, as in 1.0.0-x86_64-linux. +func (h *GemHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, error) { + versionsURL := fmt.Sprintf("%s/api/v1/versions/%s.json", h.upstreamURL, url.PathEscape(name)) + body, _, err := h.proxy.FetchOrCacheMetadata(r.Context(), "gem", "api/v1/versions/"+name, versionsURL) + if err != nil { + return time.Time{}, false, err + } + var versions []gemVersion + if err := json.Unmarshal(body, &versions); err != nil { + return time.Time{}, false, err + } + for _, v := range versions { + number := v.Number + if v.Platform != "" && v.Platform != "ruby" { + number += "-" + v.Platform + } + if number != version { + continue + } + createdAt, err := time.Parse(time.RFC3339, v.CreatedAt) + return createdAt, err == nil, nil + } + return time.Time{}, false, nil +} + // proxyCached forwards a metadata request with caching. func (h *GemHandler) proxyCached(w http.ResponseWriter, r *http.Request) { upstreamURL := h.upstreamURL + r.URL.Path diff --git a/internal/handler/helm.go b/internal/handler/helm.go index fdc000ed..9fb947f9 100644 --- a/internal/handler/helm.go +++ b/internal/handler/helm.go @@ -127,6 +127,12 @@ func (h *HelmHandler) handleChart(w http.ResponseWriter, r *http.Request) { return } if cached != nil { + if h.withholdCachedChart(w, r, repository, upstreamURL, digest) { + if cached.Reader != nil { + _ = cached.Reader.Close() + } + return + } h.serveChart(w, r, repository, digest, filename, cached) return } @@ -376,6 +382,54 @@ func (h *HelmHandler) buildChartLookup(upstreamURL string, body []byte) (*helmCh return lookup, nil } +// withholdCachedChart checks a cached chart against the cooldown before it is +// served. An uncached chart is checked by findChartDownload, but a cached one +// may have been fetched under an earlier policy. Charts are cached under the +// repository and digest rather than the chart name the policy is keyed by, so +// the name and publish time come from the index on every check. +func (h *HelmHandler) withholdCachedChart(w http.ResponseWriter, r *http.Request, repository, upstreamURL, digest string) bool { + if h.proxy.Cooldown == nil || !h.proxy.Cooldown.Enabled() { + return false + } + index := h.fetchIndex(r, repository, upstreamURL) + if index.err != nil { + h.proxy.Logger.Warn("cooldown: could not fetch Helm index for download check", + "repository", repository, "digest", digest, "error", index.err) + return false + } + chartName, created, ok := h.findChartRelease(upstreamURL, index.body, digest) + if !ok { + return false + } + return h.proxy.withholdInCooldown(w, downloadVersion{ + ecosystem: helmMetadataEcosystem, policyName: chartName, version: digest, + lookup: func() (time.Time, bool, error) { return created, !created.IsZero(), nil }, + }) +} + +// findChartRelease returns the name and created time of the index release +// with the given digest. +func (h *HelmHandler) findChartRelease(upstreamURL string, body []byte, digest string) (string, time.Time, bool) { + _, entries, err := parseHelmIndex(body) + if err != nil { + return "", time.Time{}, false + } + for i := 0; i < len(entries.Content); i += 2 { + chartName := entries.Content[i].Value + releases := entries.Content[i+1] + if releases.Kind != yaml.SequenceNode { + continue + } + for _, release := range releases.Content { + chart, err := h.parseChartRelease(chartName, upstreamURL, release) + if err == nil && chart.digest == digest { + return chartName, chart.created, true + } + } + } + return "", time.Time{}, false +} + func (h *HelmHandler) chartOnCooldown(chartName string, created time.Time) bool { return !created.IsZero() && h.proxy.Cooldown != nil && h.proxy.Cooldown.Enabled() && !h.proxy.cooldown().IsAllowed(helmMetadataEcosystem, canonicalPackagePURL(helmMetadataEcosystem, chartName), created) diff --git a/internal/handler/hex.go b/internal/handler/hex.go index f2a93adf..6836dadf 100644 --- a/internal/handler/hex.go +++ b/internal/handler/hex.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "net/http" + "net/url" "strings" "time" @@ -78,6 +79,7 @@ func (h *HexHandler) handleDownload(w http.ResponseWriter, r *http.Request) { parseErr: "could not parse tarball filename", fetchErr: "failed to fetch package", parse: h.parseTarballFilename, + publishTime: h.publishTime, }) } @@ -241,6 +243,29 @@ func parseHexReleases(body []byte) (publishTimes, error) { return times, nil } +// publishTime reads a release's inserted_at from the Hex HTTP API. The +// download check caches the response, unlike fetchFilteredVersions, so each +// download does not cost an upstream request. +func (h *HexHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, error) { + apiURL := fmt.Sprintf("%s/api/packages/%s", h.apiURL, url.PathEscape(name)) + body, _, err := h.proxy.FetchOrCacheMetadata(r.Context(), "hex", "api/packages/"+name, apiURL) + if err != nil { + return time.Time{}, false, err + } + var pkg hexPackageAPI + if err := json.Unmarshal(body, &pkg); err != nil { + return time.Time{}, false, err + } + for _, release := range pkg.Releases { + if release.Version != version { + continue + } + insertedAt, err := time.Parse(time.RFC3339Nano, release.InsertedAt) + return insertedAt, err == nil, nil + } + return time.Time{}, false, nil +} + // signedPayload decompresses a gzipped Signed message, with the metadata // limit applied to the decompressed size, and returns its payload. func (h *HexHandler) signedPayload(gzippedData []byte) ([]byte, error) { diff --git a/internal/handler/pub.go b/internal/handler/pub.go index 24fdd02c..22540f14 100644 --- a/internal/handler/pub.go +++ b/internal/handler/pub.go @@ -94,6 +94,13 @@ func (h *PubHandler) handleDownload(w http.ResponseWriter, r *http.Request) { h.proxy.Logger.Debug("pub download request", "name", name, "version", version) + if h.proxy.withholdInCooldown(w, downloadVersion{ + ecosystem: "pub", policyName: name, storedName: name, version: version, + lookup: func() (time.Time, bool, error) { return h.publishTime(r, name, version) }, + }) { + return + } + downloadURL := fmt.Sprintf("%s/packages/%s/versions/%s%s", h.upstreamURL, url.PathEscape(name), url.PathEscape(version), pubArchiveExt) h.discardVersionPage(name, version, filename) @@ -134,6 +141,33 @@ func (h *PubHandler) discardVersionPage(name, version, filename string) { } } +// publishTime reads a version's published time from the package metadata, +// sharing the metadata cache with handlePackageMetadata. +func (h *PubHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, error) { + upstreamURL := fmt.Sprintf("%s/api/packages/%s", h.upstreamURL, name) + body, _, err := h.proxy.FetchOrCacheMetadata(r.Context(), "pub", name, upstreamURL) + if err != nil { + return time.Time{}, false, err + } + var metadata struct { + Versions []struct { + Version string `json:"version"` + Published string `json:"published"` + } `json:"versions"` + } + if err := json.Unmarshal(body, &metadata); err != nil { + return time.Time{}, false, err + } + for _, v := range metadata.Versions { + if v.Version != version { + continue + } + published, err := time.Parse(time.RFC3339, v.Published) + return published, err == nil, nil + } + return time.Time{}, false, nil +} + // handlePackageMetadata proxies package metadata and rewrites archive URLs. func (h *PubHandler) handlePackageMetadata(w http.ResponseWriter, r *http.Request) { name := r.PathValue("name") From 89c60abd1a51a551aceef46d030f55f0f0b94cb6 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Sun, 18 Oct 2026 21:43:57 +0200 Subject: [PATCH 2/6] Conda now checks each package file against its own timestamp, because builds of one version differ. Conda repodata and helm indexes are parsed once per cached document instead of on every download. --- internal/handler/conda.go | 55 +++++++---- internal/handler/download_cooldown.go | 62 ++++++++++++ internal/handler/download_cooldown_test.go | 108 +++++++++++++++++---- internal/handler/handler.go | 4 + internal/handler/helm.go | 62 +++++++----- 5 files changed, 232 insertions(+), 59 deletions(-) diff --git a/internal/handler/conda.go b/internal/handler/conda.go index 22ec603a..d7cf196d 100644 --- a/internal/handler/conda.go +++ b/internal/handler/conda.go @@ -139,8 +139,10 @@ func (h *CondaHandler) handleDownload(w http.ResponseWriter, r *http.Request) { h.proxy.Logger.Debug("conda download request", "channel", channel, "arch", arch, "name", name, "version", version) + // Builds and subdirs of one version each have their own timestamp, so + // the check is per file and keeps no time in the shared versions row. if h.proxy.withholdInCooldown(w, downloadVersion{ - ecosystem: "conda", policyName: name, storedName: packageName, version: version, + ecosystem: "conda", policyName: name, version: version, lookup: func() (time.Time, bool, error) { return h.publishTime(r, channel, arch, filename) }, }) { return @@ -539,40 +541,57 @@ func (h *CondaHandler) metadataCacheKey(urlPath string) string { } // publishTime reads a package file's timestamp from the subdir's -// repodata.json. It shares proxyCached's cache entry and encoding, so the -// repodata a client just fetched is not downloaded again, and reads the one -// entry in place rather than decoding the whole index. +// repodata.json, fetched with the cache key and encoding proxyCached uses. +// The repodata is indexed once per document (see publishIndex), since a +// subdir's index runs to tens of megabytes and an install downloads many +// files from it. func (h *CondaHandler) publishTime(r *http.Request, channel, arch, filename string) (time.Time, bool, error) { repodataPath := "/" + channel + "/" + arch + "/repodata.json" cacheKey := h.metadataCacheKey(repodataPath) - body, _, encoding, err := h.proxy.fetchOrCacheMetadata(r.Context(), "conda", cacheKey, h.upstreamURL+repodataPath, "gzip", nil, "*/*") + res := h.proxy.lookupMetadata(r.Context(), "conda", cacheKey, h.upstreamURL+repodataPath, "*/*", "gzip", nil) + if res.err != nil { + return time.Time{}, false, res.err + } + index, err := publishIndex(h.proxy, "conda\x00"+cacheKey, res, h.indexRepodata) if err != nil { return time.Time{}, false, err } + published, ok := index[filename] + return published, ok, nil +} + +// indexRepodata maps each package file in a repodata.json to its timestamp, +// leaving out files without one. +func (h *CondaHandler) indexRepodata(body []byte, encoding string) (map[string]time.Time, error) { if strings.EqualFold(strings.TrimSpace(encoding), "gzip") { decoded, err := gzip.NewReader(bytes.NewReader(body)) if err != nil { - return time.Time{}, false, err + return nil, err } body, err = h.proxy.ReadMetadata(decoded) if err != nil { - return time.Time{}, false, err + return nil, err } } - - key := "packages" - if strings.HasSuffix(filename, ".conda") { - key = "packages.conda" + type entry struct { + Timestamp float64 `json:"timestamp"` } - raw, err := lookupJSON(body, key, filename, "timestamp") - if err != nil || len(raw) == 0 { - return time.Time{}, false, err + var repodata struct { + Packages map[string]entry `json:"packages"` + PackagesConda map[string]entry `json:"packages.conda"` + } + if err := json.Unmarshal(body, &repodata); err != nil { + return nil, err } - var ts float64 - if err := json.Unmarshal(raw, &ts); err != nil || ts == 0 { - return time.Time{}, false, nil + index := make(map[string]time.Time, len(repodata.Packages)+len(repodata.PackagesConda)) + for _, files := range []map[string]entry{repodata.Packages, repodata.PackagesConda} { + for filename, e := range files { + if e.Timestamp != 0 { + index[filename] = time.Unix(int64(e.Timestamp)/condaTimestampDivisor, 0) + } + } } - return time.Unix(int64(ts)/condaTimestampDivisor, 0), true, nil + return index, nil } // proxyCached forwards a metadata request with caching. diff --git a/internal/handler/download_cooldown.go b/internal/handler/download_cooldown.go index 94554d3b..5aec9599 100644 --- a/internal/handler/download_cooldown.go +++ b/internal/handler/download_cooldown.go @@ -2,7 +2,10 @@ package handler import ( "net/http" + "sync" "time" + + "golang.org/x/sync/singleflight" ) // publishTimeLookup reads a version's publish time from upstream metadata. ok @@ -80,3 +83,62 @@ func (p *Proxy) versionInCooldown(d downloadVersion) bool { return !p.Cooldown.IsAllowed(d.ecosystem, policyKey, published) } + +// maxPublishIndexes bounds how many documents publishIndexCache keeps an +// index for. +const maxPublishIndexes = 64 + +// publishIndexCache keeps, for each large metadata document, the publish +// times a download check reads from it, so a document is decompressed and +// parsed once per version from upstream rather than on every download. +// Concurrent checks for a document not indexed yet share one parse. +type publishIndexCache struct { + mu sync.Mutex + entries map[string]publishIndexEntry + group singleflight.Group +} + +type publishIndexEntry struct { + digest string + index any +} + +// publishIndex returns build's index of res, reusing the one built for the +// same document under key. res.digest identifies the document; without one, +// as when the metadata cache is off, the index is built for this call alone. +func publishIndex[V any](p *Proxy, key string, res metadataResult, build func([]byte, string) (map[string]V, error)) (map[string]V, error) { + c := &p.publishIndexes + if res.digest == "" { + return build(res.body, res.contentEncoding) + } + c.mu.Lock() + if e, ok := c.entries[key]; ok && e.digest == res.digest { + c.mu.Unlock() + return e.index.(map[string]V), nil + } + c.mu.Unlock() + + v, err, _ := c.group.Do(key+"\x00"+res.digest, func() (any, error) { + index, err := build(res.body, res.contentEncoding) + if err != nil { + return nil, err + } + c.mu.Lock() + defer c.mu.Unlock() + if c.entries == nil { + c.entries = make(map[string]publishIndexEntry) + } + if _, ok := c.entries[key]; !ok && len(c.entries) >= maxPublishIndexes { + for evict := range c.entries { + delete(c.entries, evict) + break + } + } + c.entries[key] = publishIndexEntry{digest: res.digest, index: index} + return index, nil + }) + if err != nil { + return nil, err + } + return v.(map[string]V), nil +} diff --git a/internal/handler/download_cooldown_test.go b/internal/handler/download_cooldown_test.go index 9b791bed..c504edb2 100644 --- a/internal/handler/download_cooldown_test.go +++ b/internal/handler/download_cooldown_test.go @@ -26,9 +26,10 @@ const ( // is always named demo, so the policy key is pkg:{ecosystem}/demo. type downloadCooldownCase struct { ecosystem string - // storedName is the name the artifact is cached under; empty when the - // ecosystem keeps no publish time in the versions table. - storedName string + // cachedName is the name the artifact is cached under, and keepsTime + // whether the publish time is kept in its versions row. + cachedName string + keepsTime bool // artifactVersion maps a release to the version the artifact is cached // under, and filename and path to its file and download path. artifactVersion func(version string) string @@ -48,7 +49,8 @@ func downloadCooldownCases() []downloadCooldownCase { return []downloadCooldownCase{ { ecosystem: "cargo", - storedName: "demo", + cachedName: "demo", + keepsTime: true, artifactVersion: sameVersion, filename: func(v string) string { return "demo-" + v + ".crate" }, path: func(v string) string { return "/crates/demo/" + v + "/download" }, @@ -57,7 +59,8 @@ func downloadCooldownCases() []downloadCooldownCase { }, { ecosystem: "gem", - storedName: "demo", + cachedName: "demo", + keepsTime: true, // Platform gems carry the platform in the filename and the // cached version, but not in the versions API number. artifactVersion: func(v string) string { return v + "-x86_64-linux" }, @@ -68,7 +71,8 @@ func downloadCooldownCases() []downloadCooldownCase { }, { ecosystem: "hex", - storedName: "demo", + cachedName: "demo", + keepsTime: true, artifactVersion: sameVersion, filename: func(v string) string { return "demo-" + v + ".tar" }, path: func(v string) string { return "/tarballs/demo-" + v + ".tar" }, @@ -79,7 +83,8 @@ func downloadCooldownCases() []downloadCooldownCase { }, { ecosystem: "pub", - storedName: "demo", + cachedName: "demo", + keepsTime: true, artifactVersion: sameVersion, filename: func(v string) string { return "demo-" + v + ".tar.gz" }, path: func(v string) string { return "/packages/demo/versions/" + v + ".tar.gz" }, @@ -88,9 +93,10 @@ func downloadCooldownCases() []downloadCooldownCase { }, { // Conda caches under {channel}/{name}, but the policy is keyed by - // the name alone, as the repodata filter keys it. + // the name alone, as the repodata filter keys it. Each build has + // its own timestamp, so none is kept in the version's row. ecosystem: "conda", - storedName: "conda-forge/demo", + cachedName: "conda-forge/demo", artifactVersion: sameVersion, filename: func(v string) string { return "noarch/demo-" + v + "-py_0.conda" }, path: func(v string) string { return "/conda-forge/noarch/demo-" + v + "-py_0.conda" }, @@ -102,6 +108,7 @@ func downloadCooldownCases() []downloadCooldownCase { { // Helm caches charts under the repository and digest, so the // chart name and time come from the index on every check. + cachedName: "stable", ecosystem: "helm", artifactVersion: func(v string) string { return sha256Hex(chartContent(v)) }, filename: func(v string) string { return "demo-" + v + ".tgz" }, @@ -217,12 +224,11 @@ func helmCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[stri func (c downloadCooldownCase) seedCachedArtifact(t *testing.T, proxy *Proxy, version string) { t.Helper() db, store := proxy.DB, proxy.Storage.(*mockStorage) - name := c.storedName content := "artifact " + version if c.ecosystem == "helm" { - name, content = "stable", chartContent(version) + content = chartContent(version) } - seedPackage(t, db, store, c.ecosystem, name, c.artifactVersion(version), c.filename(version), content) + seedPackage(t, db, store, c.ecosystem, c.cachedName, c.artifactVersion(version), c.filename(version), content) } // TestDownloadCooldownWithholdsCachedArtifacts requests each ecosystem's @@ -267,10 +273,10 @@ func TestDownloadCooldownWithholdsCachedArtifacts(t *testing.T) { if w.Code == http.StatusOK && w.Body.String() == "" { t.Error("served an empty body") } - if c.storedName == "" || times[version].IsZero() { + if !c.keepsTime || times[version].IsZero() { return } - ver, err := db.GetVersionByPURL(canonicalVersionPURL(c.ecosystem, c.storedName, c.artifactVersion(version))) + ver, err := db.GetVersionByPURL(canonicalVersionPURL(c.ecosystem, c.cachedName, c.artifactVersion(version))) if err != nil || ver == nil || !ver.PublishedAt.Valid { t.Fatalf("publish time not stored: %+v, %v", ver, err) } @@ -291,7 +297,7 @@ func TestDownloadCooldownUsesStoredPublishTime(t *testing.T) { t.Cleanup(upstream.Close) for _, c := range downloadCooldownCases() { - if c.storedName == "" { + if !c.keepsTime { continue } t.Run(c.ecosystem, func(t *testing.T) { @@ -299,8 +305,8 @@ func TestDownloadCooldownUsesStoredPublishTime(t *testing.T) { proxy.HTTPClient = upstream.Client() fetcher.fetchErr = errors.New("artifact must be served from the cache") c.seedCachedArtifact(t, proxy, cooldownRecentVersion) - versionPURL := canonicalVersionPURL(c.ecosystem, c.storedName, c.artifactVersion(cooldownRecentVersion)) - packagePURL := canonicalPackagePURL(c.ecosystem, c.storedName) + versionPURL := canonicalVersionPURL(c.ecosystem, c.cachedName, c.artifactVersion(cooldownRecentVersion)) + packagePURL := canonicalPackagePURL(c.ecosystem, c.cachedName) if err := db.SetVersionPublishedAt(versionPURL, packagePURL, time.Now().Add(-time.Hour)); err != nil { t.Fatalf("seeding publish time: %v", err) } @@ -320,7 +326,7 @@ func TestDownloadCooldownUsesStoredPublishTime(t *testing.T) { // needs the index to learn the chart name, so it is left out. func TestDownloadCooldownOtherPackagesSkipLookup(t *testing.T) { for _, c := range downloadCooldownCases() { - if c.storedName == "" { + if c.ecosystem == "helm" { continue } t.Run(c.ecosystem, func(t *testing.T) { @@ -345,3 +351,69 @@ func TestDownloadCooldownOtherPackagesSkipLookup(t *testing.T) { }) } } + +// TestCondaDownloadCooldownPerBuild checks that builds of one version are +// judged by their own timestamps, whichever is requested first: a rebuild +// inside the window is withheld and an old build stays available. +func TestCondaDownloadCooldownPerBuild(t *testing.T) { + now := time.Now() + repodata := map[string]any{"packages.conda": map[string]any{ + "demo-1.0.0-py_0.conda": map[string]any{"name": "demo", "version": "1.0.0", "timestamp": now.Add(-400 * 24 * time.Hour).UnixMilli()}, + "demo-1.0.0-py_1.conda": map[string]any{"name": "demo", "version": "1.0.0", "timestamp": now.Add(-time.Hour).UnixMilli()}, + }} + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode(repodata) + })) + t.Cleanup(upstream.Close) + + builds := map[string]int{"py_0": http.StatusOK, "py_1": http.StatusNotFound} + for _, order := range [][]string{{"py_0", "py_1", "py_0"}, {"py_1", "py_0", "py_1"}} { + t.Run(strings.Join(order, ","), func(t *testing.T) { + proxy, db, store, fetcher := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + fetcher.fetchErr = errors.New("artifact must be served from the cache") + for build := range builds { + filename := "noarch/demo-1.0.0-" + build + ".conda" + seedPackage(t, db, store, "conda", "conda-forge/demo", "1.0.0", filename, "artifact "+build) + } + proxy.Cooldown = &cooldown.Config{Default: "7d"} + h := NewCondaHandlerWithUpstream(proxy, "http://proxy", upstream.URL).Routes() + + for _, build := range order { + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/conda-forge/noarch/demo-1.0.0-"+build+".conda", nil)) + if w.Code != builds[build] { + t.Fatalf("%s status = %d, want %d", build, w.Code, builds[build]) + } + } + }) + } +} + +// TestPublishIndexBuildsOncePerDocument checks that an index is reused for +// the same document, rebuilt for a new one, and not kept without a digest. +func TestPublishIndexBuildsOncePerDocument(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + builds := 0 + build := func(body []byte, _ string) (map[string]bool, error) { + builds++ + return map[string]bool{string(body): true}, nil + } + steps := []struct { + res metadataResult + want string + wantBuilds int + }{ + {metadataResult{body: []byte("a"), digest: "d1"}, "a", 1}, + {metadataResult{body: []byte("ignored"), digest: "d1"}, "a", 1}, + {metadataResult{body: []byte("b"), digest: "d2"}, "b", 2}, + {metadataResult{body: []byte("c")}, "c", 3}, + {metadataResult{body: []byte("c")}, "c", 4}, + } + for i, step := range steps { + got, err := publishIndex(proxy, "key", step.res, build) + if err != nil || !got[step.want] || builds != step.wantBuilds { + t.Fatalf("step %d: publishIndex = %v, %v after %d builds, want %q after %d", i, got, err, builds, step.want, step.wantBuilds) + } + } +} diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 196f6487..125b1174 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -292,6 +292,10 @@ type Proxy struct { // Nil leaves every request to rewrite its own copy. rewrites *rewriteCache + // publishIndexes keeps publish times read out of large metadata + // documents for the download cooldown check; see publishIndex. + publishIndexes publishIndexCache + // now is the clock rewritten metadata is dated by, and started the time // the process started; see metadataDate. now func() time.Time diff --git a/internal/handler/helm.go b/internal/handler/helm.go index 9fb947f9..cbbf8b0b 100644 --- a/internal/handler/helm.go +++ b/internal/handler/helm.go @@ -120,6 +120,10 @@ func (h *HelmHandler) handleChart(w http.ResponseWriter, r *http.Request) { return } + if h.withholdChart(w, r, repository, upstreamURL, digest) { + return + } + cached, err := h.proxy.GetCachedArtifact(r.Context(), helmMetadataEcosystem, repository, digest, filename) if err != nil { h.proxy.Logger.Error("failed to check Helm chart cache", "error", err) @@ -127,12 +131,6 @@ func (h *HelmHandler) handleChart(w http.ResponseWriter, r *http.Request) { return } if cached != nil { - if h.withholdCachedChart(w, r, repository, upstreamURL, digest) { - if cached.Reader != nil { - _ = cached.Reader.Close() - } - return - } h.serveChart(w, r, repository, digest, filename, cached) return } @@ -382,38 +380,55 @@ func (h *HelmHandler) buildChartLookup(upstreamURL string, body []byte) (*helmCh return lookup, nil } -// withholdCachedChart checks a cached chart against the cooldown before it is -// served. An uncached chart is checked by findChartDownload, but a cached one +// withholdChart checks a chart against the cooldown before it is served. +// findChartDownload already leaves out charts in cooldown, but a cached chart // may have been fetched under an earlier policy. Charts are cached under the // repository and digest rather than the chart name the policy is keyed by, so -// the name and publish time come from the index on every check. -func (h *HelmHandler) withholdCachedChart(w http.ResponseWriter, r *http.Request, repository, upstreamURL, digest string) bool { +// the name and publish time come from the index, indexed once per document +// (see publishIndex). +func (h *HelmHandler) withholdChart(w http.ResponseWriter, r *http.Request, repository, upstreamURL, digest string) bool { if h.proxy.Cooldown == nil || !h.proxy.Cooldown.Enabled() { return false } - index := h.fetchIndex(r, repository, upstreamURL) - if index.err != nil { + res := h.fetchIndex(r, repository, upstreamURL) + if res.err != nil { h.proxy.Logger.Warn("cooldown: could not fetch Helm index for download check", - "repository", repository, "digest", digest, "error", index.err) + "repository", repository, "digest", digest, "error", res.err) return false } - chartName, created, ok := h.findChartRelease(upstreamURL, index.body, digest) + index, err := publishIndex(h.proxy, "helm\x00"+h.indexCacheKey(repository, upstreamURL), res, + func(body []byte, _ string) (map[string]helmChartTime, error) { + return h.indexChartTimes(upstreamURL, body) + }) + if err != nil { + h.proxy.Logger.Warn("cooldown: could not read Helm index for download check", + "repository", repository, "digest", digest, "error", err) + return false + } + chart, ok := index[digest] if !ok { return false } return h.proxy.withholdInCooldown(w, downloadVersion{ - ecosystem: helmMetadataEcosystem, policyName: chartName, version: digest, - lookup: func() (time.Time, bool, error) { return created, !created.IsZero(), nil }, + ecosystem: helmMetadataEcosystem, policyName: chart.name, version: digest, + lookup: func() (time.Time, bool, error) { return chart.created, !chart.created.IsZero(), nil }, }) } -// findChartRelease returns the name and created time of the index release -// with the given digest. -func (h *HelmHandler) findChartRelease(upstreamURL string, body []byte, digest string) (string, time.Time, bool) { +// helmChartTime is the chart a release digest belongs to and when it was +// created. +type helmChartTime struct { + name string + created time.Time +} + +// indexChartTimes maps each release digest in an index to its chart. +func (h *HelmHandler) indexChartTimes(upstreamURL string, body []byte) (map[string]helmChartTime, error) { _, entries, err := parseHelmIndex(body) if err != nil { - return "", time.Time{}, false + return nil, err } + index := make(map[string]helmChartTime) for i := 0; i < len(entries.Content); i += 2 { chartName := entries.Content[i].Value releases := entries.Content[i+1] @@ -422,12 +437,13 @@ func (h *HelmHandler) findChartRelease(upstreamURL string, body []byte, digest s } for _, release := range releases.Content { chart, err := h.parseChartRelease(chartName, upstreamURL, release) - if err == nil && chart.digest == digest { - return chartName, chart.created, true + if err != nil { + continue } + index[chart.digest] = helmChartTime{name: chartName, created: chart.created} } } - return "", time.Time{}, false + return index, nil } func (h *HelmHandler) chartOnCooldown(chartName string, created time.Time) bool { From 5db16cbb991c25714673aebd9fd860ca451da9dc Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 26 Oct 2026 07:28:44 +0100 Subject: [PATCH 3/6] Conda reads repodata as a stream for the download check, under the same bound as the repodata filter, so a repodata.json over metadata_max_size no longer lets a package in cooldown through. If even the compressed copy is too large to cache, the check streams it from upstream. --- internal/handler/conda.go | 91 ++++++++++++++++------ internal/handler/download_cooldown_test.go | 48 ++++++++++++ 2 files changed, 114 insertions(+), 25 deletions(-) diff --git a/internal/handler/conda.go b/internal/handler/conda.go index d7cf196d..5d79b3f9 100644 --- a/internal/handler/conda.go +++ b/internal/handler/conda.go @@ -543,16 +543,23 @@ func (h *CondaHandler) metadataCacheKey(urlPath string) string { // publishTime reads a package file's timestamp from the subdir's // repodata.json, fetched with the cache key and encoding proxyCached uses. // The repodata is indexed once per document (see publishIndex), since a -// subdir's index runs to tens of megabytes and an install downloads many -// files from it. +// subdir's index runs to hundreds of megabytes and an install downloads many +// files from it. A compressed copy over metadata_max_size can't be cached, +// so it is streamed from upstream for each check, as handleRepodata does. func (h *CondaHandler) publishTime(r *http.Request, channel, arch, filename string) (time.Time, bool, error) { repodataPath := "/" + channel + "/" + arch + "/repodata.json" cacheKey := h.metadataCacheKey(repodataPath) res := h.proxy.lookupMetadata(r.Context(), "conda", cacheKey, h.upstreamURL+repodataPath, "*/*", "gzip", nil) - if res.err != nil { - return time.Time{}, false, res.err + var index map[string]time.Time + var err error + switch { + case errors.Is(res.err, ErrMetadataTooLarge): + index, err = h.streamRepodataIndex(r, h.upstreamURL+repodataPath) + case res.err != nil: + err = res.err + default: + index, err = publishIndex(h.proxy, "conda\x00"+cacheKey, res, h.indexRepodata) } - index, err := publishIndex(h.proxy, "conda\x00"+cacheKey, res, h.indexRepodata) if err != nil { return time.Time{}, false, err } @@ -560,36 +567,70 @@ func (h *CondaHandler) publishTime(r *http.Request, channel, arch, filename stri return published, ok, nil } -// indexRepodata maps each package file in a repodata.json to its timestamp, -// leaving out files without one. +// indexRepodata indexes a cached repodata.json body. func (h *CondaHandler) indexRepodata(body []byte, encoding string) (map[string]time.Time, error) { + var reader io.Reader = bytes.NewReader(body) if strings.EqualFold(strings.TrimSpace(encoding), "gzip") { - decoded, err := gzip.NewReader(bytes.NewReader(body)) - if err != nil { - return nil, err - } - body, err = h.proxy.ReadMetadata(decoded) + decoded, err := gzip.NewReader(reader) if err != nil { return nil, err } + reader = decoded } - type entry struct { - Timestamp float64 `json:"timestamp"` - } - var repodata struct { - Packages map[string]entry `json:"packages"` - PackagesConda map[string]entry `json:"packages.conda"` + return readRepodataIndex(reader) +} + +// streamRepodataIndex indexes repodata.json as it streams from upstream. +func (h *CondaHandler) streamRepodataIndex(r *http.Request, upstreamURL string) (map[string]time.Time, error) { + req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, upstreamURL, nil) + if err != nil { + return nil, err } - if err := json.Unmarshal(body, &repodata); err != nil { + req.Header.Set(headerAcceptEncoding, "gzip") + resp, err := h.proxy.HTTPClient.Do(req) + if err != nil { return nil, err } - index := make(map[string]time.Time, len(repodata.Packages)+len(repodata.PackagesConda)) - for _, files := range []map[string]entry{repodata.Packages, repodata.PackagesConda} { - for filename, e := range files { - if e.Timestamp != 0 { - index[filename] = time.Unix(int64(e.Timestamp)/condaTimestampDivisor, 0) - } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("repodata returned %d", resp.StatusCode) + } + var reader io.Reader = resp.Body + if strings.EqualFold(strings.TrimSpace(resp.Header.Get(headerContentEncoding)), "gzip") { + decoded, err := gzip.NewReader(resp.Body) + if err != nil { + return nil, err + } + defer func() { _ = decoded.Close() }() + reader = decoded + } + return readRepodataIndex(reader) +} + +// readRepodataIndex maps each package file in a repodata.json to its +// timestamp, leaving out files without one. It reads the document as it +// streams, under the same bound as filterRepodata, so the decompressed size +// is not limited by metadata_max_size. +func readRepodataIndex(r io.Reader) (map[string]time.Time, error) { + index := make(map[string]time.Time) + err := jsonstream.CopyObject(json.NewDecoder(io.LimitReader(r, maxRepodataSize)), io.Discard, func(m *jsonstream.Member) error { + if m.Key != "packages" && m.Key != "packages.conda" { + return m.Copy() } + return m.Object(func(entry *jsonstream.Member) error { + raw, err := entry.Raw() + if err != nil { + return err + } + var e repodataEntry + if json.Unmarshal(raw, &e) == nil && e.Timestamp != 0 { + index[entry.Key] = time.Unix(int64(e.Timestamp)/condaTimestampDivisor, 0) + } + return nil + }) + }) + if err != nil { + return nil, err } return index, nil } diff --git a/internal/handler/download_cooldown_test.go b/internal/handler/download_cooldown_test.go index c504edb2..1dd44e0a 100644 --- a/internal/handler/download_cooldown_test.go +++ b/internal/handler/download_cooldown_test.go @@ -417,3 +417,51 @@ func TestPublishIndexBuildsOncePerDocument(t *testing.T) { } } } + +// TestCondaDownloadCooldownPastMetadataLimit checks that a cached package is +// withheld when its repodata is over metadata_max_size, as the repodata +// endpoint hides it: once only decompressed, and once already compressed. +func TestCondaDownloadCooldownPastMetadataLimit(t *testing.T) { + plain := largeRepodata(500) + compressed := gzipPayload(t, plain) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set(headerContentEncoding, "gzip") + _, _ = w.Write(compressed) + })) + t.Cleanup(upstream.Close) + + for _, tt := range []struct { + name string + limit int64 + }{ + {"decompressed over the limit", int64(len(compressed)) + 1}, + {"compressed over the limit", int64(len(compressed)) - 1}, + } { + t.Run(tt.name, func(t *testing.T) { + proxy, db, store, fetcher := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + proxy.MetadataMaxSize = tt.limit + fetcher.fetchErr = errors.New("artifact must be served from the cache") + for _, file := range []string{"pkg-0-old.conda", "pkg-1-new.conda"} { + seedPackage(t, db, store, "conda", "conda-forge/pkg", strings.Split(file, "-")[1], "linux-64/"+file, "artifact "+file) + } + proxy.Cooldown = &cooldown.Config{Default: "3d"} + h := NewCondaHandlerWithUpstream(proxy, "http://proxy", upstream.URL).Routes() + + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/conda-forge/linux-64/repodata.json", nil)) + if w.Code != http.StatusOK || strings.Contains(w.Body.String(), "pkg-1-new.conda") { + t.Fatalf("repodata status = %d, want 200 without pkg-1-new.conda", w.Code) + } + for file, want := range map[string]int{"pkg-0-old.conda": http.StatusOK, "pkg-1-new.conda": http.StatusNotFound} { + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/conda-forge/linux-64/"+file, nil)) + if w.Code != want { + t.Errorf("%s status = %d, want %d (body %q)", file, w.Code, want, w.Body.String()) + } + } + }) + } +} From 3ab7dcc1dc0fdcd821f08cc9eb23d020274a8625 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 26 Oct 2026 08:55:25 +0100 Subject: [PATCH 4/6] The UTC test compares the zone offset, since Postgres reads TIMESTAMP back as a fixed +0000 zone rather than time.UTC. --- internal/database/database_test.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 91734e11..500ad0d8 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -1225,7 +1225,9 @@ func TestVersionPublishedAtIsUTC(t *testing.T) { if err != nil || got == nil || !got.PublishedAt.Valid { t.Fatalf("GetVersionByPURL(%s) = %+v, %v", purl, got, err) } - if !got.PublishedAt.Time.Equal(want) || got.PublishedAt.Time.Location() != time.UTC { + // Postgres reads TIMESTAMP back as a fixed +0000 zone rather + // than time.UTC, so compare the offset. + if _, offset := got.PublishedAt.Time.Zone(); !got.PublishedAt.Time.Equal(want) || offset != 0 { t.Errorf("%s PublishedAt read back as %v, want %v", purl, got.PublishedAt.Time, want) } } From b9dcbd27fc065e6e82e8b930b36a243170d9befb Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Thu, 5 Nov 2026 18:41:03 +0000 Subject: [PATCH 5/6] Handle cooldown lookup failures and reuse download indexes --- docs/configuration.md | 4 + internal/handler/cargo.go | 16 ++- internal/handler/conda.go | 39 +++--- internal/handler/cooldown_times.go | 15 +++ internal/handler/download_cooldown.go | 98 ++++++++------- .../handler/download_cooldown_cache_test.go | 26 ++++ .../handler/download_cooldown_errors_test.go | 101 +++++++++++++++ internal/handler/download_cooldown_test.go | 11 +- internal/handler/filename_download.go | 4 +- internal/handler/gem.go | 27 +--- internal/handler/helm.go | 117 ++++++++---------- internal/handler/hex.go | 22 +--- internal/handler/pub.go | 6 +- 13 files changed, 294 insertions(+), 192 deletions(-) create mode 100644 internal/handler/download_cooldown_cache_test.go create mode 100644 internal/handler/download_cooldown_errors_test.go diff --git a/docs/configuration.md b/docs/configuration.md index bf34b23b..38765e08 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1201,6 +1201,10 @@ Maven caveats: Helm uses each release's `created` time from the repository's `index.yaml`. Releases in cooldown are left out of the index the proxy serves, and their archives aren't served; a release without `created` is let through. Charts pulled over OCI have no publish time and aren't held back. +Cargo, RubyGems, Hex, pub.dev, Conda and Helm check artifact downloads against cooldown, including files cached before the policy changed. When a publish-time lookup fails, an already-cached file is still served; an uncached download fails with 502, or 503 with `Retry-After` when the rate-limit gate holds the request. Conda checks timestamps per file because builds of one version can have different publish times. + +Conda repodata and Helm indexes are parsed once per cached document for download checks. With `cache_metadata` off, the parsed index is kept for one minute, including the fetch, so an install can reuse it across files. Conda streams repodata when its compressed copy exceeds `metadata_max_size`, under the same decompressed size bound as the metadata filter. + RubyGems, Hex and PyPI keep their publish times outside the metadata clients read: the RubyGems versions API, the Hex HTTP API (`upstream.hex_api`) and the PyPI JSON API. The proxy fetches them alongside each compact index, package or simple page it filters, and for PyPI downloads. Cooldown fails closed: when it applies to a package and those times can't be fetched or parsed, or the metadata can't be rewritten, the request fails with 502, or with 503 and `Retry-After` while the upstream is rate limiting the proxy, instead of serving the metadata unfiltered. The same goes for npm, Composer and pub.dev metadata that can't be rewritten, and for npm tarball downloads whose publish time can't be read. A PyPI file or npm tarball that is already cached is still served when its publish time can't be read; only a download that would have to be fetched fails. A PyPI release's time is the upload time of its newest file. The proxy stores it when a download looks it up, and serves files of that release that are already cached against the stored time without reading the JSON API again. A file that isn't cached yet is always looked up, since it may have been added to the release since, and the newer time is stored. Packages exempted with a duration of `0` are served as before and don't need the times. A PyPI package the JSON API returns 404 for has no times to go by and is served unfiltered. For RubyGems and Hex, both documents go through the metadata cache. With [`cache_metadata`](#metadata-caching) they are served from it within `metadata_ttl` without asking the upstream, revalidated with their ETags after it, and filtered from the last cached copies, marked stale, while the upstream is down; without it they are still fetched once for concurrent requests. Each document has its own TTL, so the compact index or Hex package can be newer than the cached times and list a version they don't have. The times are then fetched again before filtering, and if that fails, such versions are left out until the times can be read. The filtered index or package carries an ETag of its own, so Bundler and Mix revalidate it with `304 Not Modified`. Publish times are compared with the current time on each request, and the filtered result is kept until the first version it held back leaves cooldown (see [rewritten metadata cache](#rewritten-metadata-cache)); a newly published version shows up within `metadata_ttl`, as with other cached metadata. Upstream errors are answered as for other cached metadata (404, or 502) rather than relayed. diff --git a/internal/handler/cargo.go b/internal/handler/cargo.go index 43bcbc4d..fa93a747 100644 --- a/internal/handler/cargo.go +++ b/internal/handler/cargo.go @@ -197,11 +197,17 @@ func (h *CargoHandler) publishTime(r *http.Request, name, version string) (time. return time.Time{}, false, err } for line := range bytes.SplitSeq(body, []byte("\n")) { - if vers, _, _ := lookupJSONString(line, "vers"); vers != version { + if len(bytes.TrimSpace(line)) == 0 { continue } - pubtime, _, _ := lookupJSONString(line, "pubtime") - published, err := time.Parse(time.RFC3339, pubtime) + var entry crateIndexEntry + if err := json.Unmarshal(line, &entry); err != nil { + return time.Time{}, false, err + } + if entry.Version != version { + continue + } + published, err := time.Parse(time.RFC3339, entry.PublishTime) return published, err == nil, nil } return time.Time{}, false, nil @@ -238,8 +244,8 @@ func (h *CargoHandler) handleDownload(w http.ResponseWriter, r *http.Request) { h.proxy.Logger.Debug("cargo download request", "crate", name, "version", version, "filename", filename) - if h.proxy.withholdInCooldown(w, downloadVersion{ - ecosystem: "cargo", policyName: name, storedName: name, version: version, + if h.proxy.withholdInCooldown(w, r, downloadVersion{ + ecosystem: "cargo", policyName: name, storedName: name, version: version, cacheName: name, filename: filename, lookup: func() (time.Time, bool, error) { return h.publishTime(r, name, version) }, }) { return diff --git a/internal/handler/conda.go b/internal/handler/conda.go index 5d79b3f9..de6ae8a8 100644 --- a/internal/handler/conda.go +++ b/internal/handler/conda.go @@ -141,18 +141,16 @@ func (h *CondaHandler) handleDownload(w http.ResponseWriter, r *http.Request) { // Builds and subdirs of one version each have their own timestamp, so // the check is per file and keeps no time in the shared versions row. - if h.proxy.withholdInCooldown(w, downloadVersion{ - ecosystem: "conda", policyName: name, version: version, + // Subdirs distinguish cached files with the same filename. + cacheFilename := arch + "/" + filename + if h.proxy.withholdInCooldown(w, r, downloadVersion{ + ecosystem: "conda", policyName: name, version: version, cacheName: packageName, filename: cacheFilename, lookup: func() (time.Time, bool, error) { return h.publishTime(r, channel, arch, filename) }, }) { return } upstreamURL := h.upstreamURL + r.URL.Path - // The same filename can hold different bytes in each subdir, so the - // subdir is part of the cache key. - cacheFilename := arch + "/" + filename - result, err := h.proxy.GetOrFetchArtifactFromURL(r.Context(), "conda", packageName, version, cacheFilename, upstreamURL) if err != nil { h.proxy.serveArtifactError(w, err, "failed to fetch package") @@ -540,26 +538,21 @@ func (h *CondaHandler) metadataCacheKey(urlPath string) string { return hashedCacheKey(h.upstreamURL, urlPath) } -// publishTime reads a package file's timestamp from the subdir's -// repodata.json, fetched with the cache key and encoding proxyCached uses. -// The repodata is indexed once per document (see publishIndex), since a -// subdir's index runs to hundreds of megabytes and an install downloads many -// files from it. A compressed copy over metadata_max_size can't be cached, -// so it is streamed from upstream for each check, as handleRepodata does. +// publishTime streams compressed repodata when it exceeds metadata_max_size. func (h *CondaHandler) publishTime(r *http.Request, channel, arch, filename string) (time.Time, bool, error) { repodataPath := "/" + channel + "/" + arch + "/repodata.json" cacheKey := h.metadataCacheKey(repodataPath) - res := h.proxy.lookupMetadata(r.Context(), "conda", cacheKey, h.upstreamURL+repodataPath, "*/*", "gzip", nil) - var index map[string]time.Time - var err error - switch { - case errors.Is(res.err, ErrMetadataTooLarge): - index, err = h.streamRepodataIndex(r, h.upstreamURL+repodataPath) - case res.err != nil: - err = res.err - default: - index, err = publishIndex(h.proxy, "conda\x00"+cacheKey, res, h.indexRepodata) - } + index, err := publishIndex(h.proxy, "conda\x00"+cacheKey, func() metadataResult { + return h.proxy.lookupMetadata(r.Context(), "conda", cacheKey, h.upstreamURL+repodataPath, "*/*", "gzip", nil) + }, func(res metadataResult) (map[string]time.Time, error) { + if errors.Is(res.err, ErrMetadataTooLarge) { + return h.streamRepodataIndex(r, h.upstreamURL+repodataPath) + } + if res.err != nil { + return nil, res.err + } + return h.indexRepodata(res.body, res.contentEncoding) + }) if err != nil { return time.Time{}, false, err } diff --git a/internal/handler/cooldown_times.go b/internal/handler/cooldown_times.go index 5d95d16a..b7a05162 100644 --- a/internal/handler/cooldown_times.go +++ b/internal/handler/cooldown_times.go @@ -16,6 +16,21 @@ import ( // through. type publishTimes map[string]time.Time +func (p *Proxy) lookupPublishTime(ctx context.Context, t timedMetadata, version string) (time.Time, bool, error) { + var times publishTimes + validate := func(body []byte) error { + var err error + times, err = t.parseTimes(body) + return err + } + res := p.lookupMetadata(ctx, t.ecosystem, t.timesKey, t.timesURL, t.timesAccept, "", validate) + if res.err != nil { + return time.Time{}, false, res.err + } + published, ok := times[version] + return published, ok && !published.IsZero(), nil +} + // timedMetadata describes metadata whose publish times live in a second // document, as with the RubyGems compact index and the versions API, or a // Hex package and the Hex HTTP API. Both are read through the metadata cache. diff --git a/internal/handler/download_cooldown.go b/internal/handler/download_cooldown.go index 5aec9599..47d8b704 100644 --- a/internal/handler/download_cooldown.go +++ b/internal/handler/download_cooldown.go @@ -24,21 +24,22 @@ type downloadVersion struct { // keeps no time, and every check asks lookup. storedName string version string + cacheName string + filename string lookup publishTimeLookup } // withholdInCooldown reports whether a download is inside the cooldown -// window, and if so answers it with a 404. Filtering metadata is not enough on -// its own: lockfiles record artifact URLs, so clients reach the download path -// without reading the metadata first, and a cached artifact would be served -// even after the policy was raised. -// -// A publish time never changes, so the stored versions row is read first and -// lookup is only asked for a version the proxy has no time for yet; the time -// it finds is stored. A version with no usable publish time is allowed -// through, matching how the metadata filters treat it. -func (p *Proxy) withholdInCooldown(w http.ResponseWriter, d downloadVersion) bool { - if !p.versionInCooldown(d) { +// window, or a lookup failure prevents an uncached download. +func (p *Proxy) withholdInCooldown(w http.ResponseWriter, r *http.Request, d downloadVersion) bool { + withheld, err := p.versionInCooldown(d) + if err != nil { + if !p.serveCachedDespiteCooldownError(w, r, d.ecosystem, d.cacheName, d.version, d.filename, err) { + p.writeCooldownLookupError(w, d.ecosystem, d.policyName, err) + } + return true + } + if !withheld { return false } p.Logger.Info("cooldown: withholding "+d.ecosystem+" download", @@ -47,31 +48,26 @@ func (p *Proxy) withholdInCooldown(w http.ResponseWriter, d downloadVersion) boo return true } -func (p *Proxy) versionInCooldown(d downloadVersion) bool { - if p.Cooldown == nil || !p.Cooldown.Enabled() { - return false - } +func (p *Proxy) versionInCooldown(d downloadVersion) (bool, error) { policyKey := canonicalPackagePURL(d.ecosystem, d.policyName) - if p.Cooldown.For(d.ecosystem, policyKey) <= 0 { - return false + if !p.cooldownApplies(d.ecosystem, d.policyName) { + return false, nil } var versionPURL string if d.storedName != "" { versionPURL = canonicalVersionPURL(d.ecosystem, d.storedName, d.version) if ver, err := p.DB.GetVersionByPURL(versionPURL); err == nil && ver != nil && ver.PublishedAt.Valid { - return !p.Cooldown.IsAllowed(d.ecosystem, policyKey, ver.PublishedAt.Time) + return !p.cooldown().IsAllowed(d.ecosystem, policyKey, ver.PublishedAt.Time), nil } } published, ok, err := d.lookup() if err != nil { - p.Logger.Warn("cooldown: could not read publish time for download check", - "ecosystem", d.ecosystem, "name", d.policyName, "version", d.version, "error", err) - return false + return false, err } if !ok { - return false + return false, nil } if versionPURL != "" { @@ -81,7 +77,7 @@ func (p *Proxy) versionInCooldown(d downloadVersion) bool { } } - return !p.Cooldown.IsAllowed(d.ecosystem, policyKey, published) + return !p.cooldown().IsAllowed(d.ecosystem, policyKey, published), nil } // maxPublishIndexes bounds how many documents publishIndexCache keeps an @@ -99,27 +95,40 @@ type publishIndexCache struct { } type publishIndexEntry struct { - digest string - index any + digest string + index any + expires time.Time } -// publishIndex returns build's index of res, reusing the one built for the -// same document under key. res.digest identifies the document; without one, -// as when the metadata cache is off, the index is built for this call alone. -func publishIndex[V any](p *Proxy, key string, res metadataResult, build func([]byte, string) (map[string]V, error)) (map[string]V, error) { +// publishIndex reuses parsed documents by digest, or for a minute without one. +func publishIndex[V any](p *Proxy, key string, load func() metadataResult, build func(metadataResult) (V, error)) (V, error) { //nolint:ireturn // Supports timestamp maps and Helm lookups. c := &p.publishIndexes - if res.digest == "" { - return build(res.body, res.contentEncoding) + fresh := func() (V, bool) { + c.mu.Lock() + defer c.mu.Unlock() + if e, ok := c.entries[key]; ok && !e.expires.IsZero() && p.clock().Before(e.expires) { + return e.index.(V), true + } + var zero V + return zero, false } - c.mu.Lock() - if e, ok := c.entries[key]; ok && e.digest == res.digest { - c.mu.Unlock() - return e.index.(map[string]V), nil + if index, ok := fresh(); ok { + return index, nil } - c.mu.Unlock() - - v, err, _ := c.group.Do(key+"\x00"+res.digest, func() (any, error) { - index, err := build(res.body, res.contentEncoding) + v, err, _ := c.group.Do(key, func() (any, error) { + if index, ok := fresh(); ok { + return index, nil + } + res := load() + if res.err == nil && res.digest != "" { + c.mu.Lock() + entry, ok := c.entries[key] + c.mu.Unlock() + if ok && entry.digest == res.digest { + return entry.index, nil + } + } + index, err := build(res) if err != nil { return nil, err } @@ -134,11 +143,16 @@ func publishIndex[V any](p *Proxy, key string, res metadataResult, build func([] break } } - c.entries[key] = publishIndexEntry{digest: res.digest, index: index} + entry := publishIndexEntry{digest: res.digest, index: index} + if res.digest == "" || !p.CacheMetadata { + entry.expires = p.clock().Add(time.Minute) + } + c.entries[key] = entry return index, nil }) if err != nil { - return nil, err + var zero V + return zero, err } - return v.(map[string]V), nil + return v.(V), nil } diff --git a/internal/handler/download_cooldown_cache_test.go b/internal/handler/download_cooldown_cache_test.go new file mode 100644 index 00000000..be488ded --- /dev/null +++ b/internal/handler/download_cooldown_cache_test.go @@ -0,0 +1,26 @@ +package handler + +import ( + "net/http" + "testing" +) + +func TestGemDownloadSharesPublishTimeMetadata(t *testing.T) { + u, proxy, h := gemCacheSetup(t) + wantStatus(t, serveGet(t, h, gemInfoPath), http.StatusOK, "") + seedPackage(t, proxy.DB, proxy.Storage.(*mockStorage), "gem", "testgem", "1.0.0", "testgem-1.0.0.gem", "gem bytes") + wantStatus(t, serveGet(t, h, "/gems/testgem-1.0.0.gem"), http.StatusOK, "") + if n, _ := u.counts(gemVersionsPath); n != 1 { + t.Fatalf("versions API fetched %d times, want 1", n) + } +} + +func TestHexDownloadSharesPublishTimeMetadata(t *testing.T) { + u, proxy, h, _ := hexCacheSetup(t, 0) + wantStatus(t, serveGet(t, h, hexPackagePath), http.StatusOK, "") + seedPackage(t, proxy.DB, proxy.Storage.(*mockStorage), "hex", "testpkg", "1.0.0", "testpkg-1.0.0.tar", "hex bytes") + wantStatus(t, serveGet(t, h, "/tarballs/testpkg-1.0.0.tar"), http.StatusOK, "") + if n, _ := u.counts(hexAPIPath); n != 1 { + t.Fatalf("Hex API fetched %d times, want 1", n) + } +} diff --git a/internal/handler/download_cooldown_errors_test.go b/internal/handler/download_cooldown_errors_test.go new file mode 100644 index 00000000..b214db26 --- /dev/null +++ b/internal/handler/download_cooldown_errors_test.go @@ -0,0 +1,101 @@ +package handler + +import ( + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/cooldown" + registryclient "github.com/git-pkgs/registries/client" +) + +func TestDownloadCooldownLookupFailures(t *testing.T) { + for _, c := range downloadCooldownCases() { + for _, failure := range []string{"upstream error", "invalid metadata", "rate limited"} { + for _, cached := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/%s/cached=%t", c.ecosystem, failure, cached), func(t *testing.T) { + assertDownloadLookupFailure(t, c, failure, cached) + }) + } + } + } +} + +func assertDownloadLookupFailure(t *testing.T, c downloadCooldownCase, failure string, cached bool) { + t.Helper() + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if failure == "upstream error" { + w.WriteHeader(http.StatusInternalServerError) + } + _, _ = w.Write([]byte("invalid metadata")) + })) + t.Cleanup(upstream.Close) + proxy, _, _, fetcher := setupTestProxy(t) + proxy.Cooldown = &cooldown.Config{Default: "7d"} + proxy.HTTPClient = upstream.Client() + if failure == "rate limited" { + proxy.HTTPClient = &http.Client{Transport: roundTripperFunc(func(_ *http.Request) (*http.Response, error) { + return nil, ®istryclient.HeldError{Host: "upstream.example", Until: time.Now().Add(time.Minute), RetryAfter: time.Minute} + })} + } + if cached { + c.seedCachedArtifact(t, proxy, cooldownRecentVersion) + } + w := serveGet(t, c.handler(proxy, upstream.URL), c.path(cooldownRecentVersion)) + want := http.StatusBadGateway + if failure == "rate limited" { + want = http.StatusServiceUnavailable + } + if cached { + want = http.StatusOK + if w.Body.Len() == 0 { + t.Fatal("cached artifact body is empty") + } + } + wantStatus(t, w, want, "") + if fetcher.fetchCalled { + t.Fatal("lookup failure fetched an unchecked artifact") + } +} + +func TestDownloadPublishIndexesHaveShortTTLWithoutMetadataCache(t *testing.T) { + for _, c := range downloadCooldownCases() { + if c.ecosystem != "conda" && c.ecosystem != "helm" { + continue + } + t.Run(c.ecosystem, func(t *testing.T) { + var calls atomic.Int64 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + c.metadata(w, r, map[string]time.Time{ + cooldownOldVersion: time.Now().Add(-30 * 24 * time.Hour), + cooldownRecentVersion: time.Now().Add(-time.Hour), + }) + })) + t.Cleanup(upstream.Close) + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + proxy.Cooldown = &cooldown.Config{Default: "7d"} + clock := &testClock{t: time.Now()} + proxy.now = clock.now + c.seedCachedArtifact(t, proxy, cooldownOldVersion) + c.seedCachedArtifact(t, proxy, cooldownRecentVersion) + h := c.handler(proxy, upstream.URL) + for range 3 { + wantStatus(t, serveGet(t, h, c.path(cooldownOldVersion)), http.StatusOK, "") + wantStatus(t, serveGet(t, h, c.path(cooldownRecentVersion)), http.StatusNotFound, "") + } + if n := calls.Load(); n != 1 { + t.Fatalf("index downloads within TTL = %d, want 1", n) + } + clock.advance(time.Minute + time.Second) + wantStatus(t, serveGet(t, h, c.path(cooldownRecentVersion)), http.StatusNotFound, "") + if n := calls.Load(); n != 2 { + t.Errorf("index downloads after expiry = %d, want 2", n) + } + }) + } +} diff --git a/internal/handler/download_cooldown_test.go b/internal/handler/download_cooldown_test.go index 1dd44e0a..a3c89ae9 100644 --- a/internal/handler/download_cooldown_test.go +++ b/internal/handler/download_cooldown_test.go @@ -391,13 +391,14 @@ func TestCondaDownloadCooldownPerBuild(t *testing.T) { } // TestPublishIndexBuildsOncePerDocument checks that an index is reused for -// the same document, rebuilt for a new one, and not kept without a digest. +// the same document, rebuilt for a new one, and kept briefly without a digest. func TestPublishIndexBuildsOncePerDocument(t *testing.T) { proxy, _, _, _ := setupTestProxy(t) + proxy.CacheMetadata = true builds := 0 - build := func(body []byte, _ string) (map[string]bool, error) { + build := func(res metadataResult) (map[string]bool, error) { builds++ - return map[string]bool{string(body): true}, nil + return map[string]bool{string(res.body): true}, nil } steps := []struct { res metadataResult @@ -408,10 +409,10 @@ func TestPublishIndexBuildsOncePerDocument(t *testing.T) { {metadataResult{body: []byte("ignored"), digest: "d1"}, "a", 1}, {metadataResult{body: []byte("b"), digest: "d2"}, "b", 2}, {metadataResult{body: []byte("c")}, "c", 3}, - {metadataResult{body: []byte("c")}, "c", 4}, + {metadataResult{body: []byte("c")}, "c", 3}, } for i, step := range steps { - got, err := publishIndex(proxy, "key", step.res, build) + got, err := publishIndex(proxy, "key", func() metadataResult { return step.res }, build) if err != nil || !got[step.want] || builds != step.wantBuilds { t.Fatalf("step %d: publishIndex = %v, %v after %d builds, want %q after %d", i, got, err, builds, step.want, step.wantBuilds) } diff --git a/internal/handler/filename_download.go b/internal/handler/filename_download.go index dbc0e641..373c8f45 100644 --- a/internal/handler/filename_download.go +++ b/internal/handler/filename_download.go @@ -34,8 +34,8 @@ func (p *Proxy) handleFilenameDownload(w http.ResponseWriter, r *http.Request, d p.Logger.Debug(d.ecosystem+" download request", "name", name, "version", version, "filename", filename) - if p.withholdInCooldown(w, downloadVersion{ - ecosystem: d.ecosystem, policyName: name, storedName: name, version: version, + if p.withholdInCooldown(w, r, downloadVersion{ + ecosystem: d.ecosystem, policyName: name, storedName: name, version: version, cacheName: name, filename: filename, lookup: func() (time.Time, bool, error) { return d.publishTime(r, name, version) }, }) { return diff --git a/internal/handler/gem.go b/internal/handler/gem.go index 009e0f51..e9ee55af 100644 --- a/internal/handler/gem.go +++ b/internal/handler/gem.go @@ -5,7 +5,6 @@ import ( "encoding/json" "fmt" "net/http" - "net/url" "regexp" "strings" "time" @@ -306,32 +305,8 @@ func parseGemVersions(body []byte) (publishTimes, error) { return times, nil } -// publishTime reads a gem version's created_at from the versions API. The -// download check caches the response, unlike fetchFilteredVersions, so each -// download does not cost an upstream request. version carries the platform -// the way gem filenames do, as in 1.0.0-x86_64-linux. func (h *GemHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, error) { - versionsURL := fmt.Sprintf("%s/api/v1/versions/%s.json", h.upstreamURL, url.PathEscape(name)) - body, _, err := h.proxy.FetchOrCacheMetadata(r.Context(), "gem", "api/v1/versions/"+name, versionsURL) - if err != nil { - return time.Time{}, false, err - } - var versions []gemVersion - if err := json.Unmarshal(body, &versions); err != nil { - return time.Time{}, false, err - } - for _, v := range versions { - number := v.Number - if v.Platform != "" && v.Platform != "ruby" { - number += "-" + v.Platform - } - if number != version { - continue - } - createdAt, err := time.Parse(time.RFC3339, v.CreatedAt) - return createdAt, err == nil, nil - } - return time.Time{}, false, nil + return h.proxy.lookupPublishTime(r.Context(), h.timedIndex(name), version) } // proxyCached forwards a metadata request with caching. diff --git a/internal/handler/helm.go b/internal/handler/helm.go index cbbf8b0b..661ca98f 100644 --- a/internal/handler/helm.go +++ b/internal/handler/helm.go @@ -120,8 +120,22 @@ func (h *HelmHandler) handleChart(w http.ResponseWriter, r *http.Request) { return } - if h.withholdChart(w, r, repository, upstreamURL, digest) { - return + var charts *helmChartLookup + if h.proxy.cooldownEnabled() { + var err error + charts, err = h.downloadChartLookup(r, repository, upstreamURL) + if err != nil { + cached, cacheErr := h.proxy.GetCachedArtifact(r.Context(), helmMetadataEcosystem, repository, digest, filename) + if cacheErr == nil && cached != nil { + h.serveChart(w, r, repository, digest, filename, cached) + } else { + h.proxy.writeCooldownLookupError(w, helmMetadataEcosystem, repository, err) + } + return + } + if h.withholdChart(w, charts, digest, filename) { + return + } } cached, err := h.proxy.GetCachedArtifact(r.Context(), helmMetadataEcosystem, repository, digest, filename) @@ -135,13 +149,17 @@ func (h *HelmHandler) handleChart(w http.ResponseWriter, r *http.Request) { return } - index := h.fetchIndex(r.Context(), repository, upstreamURL) - if index.err != nil { - h.serveIndexError(w, index.err) - return + var downloadURL string + if charts == nil { + index := h.fetchIndex(r.Context(), repository, upstreamURL) + if index.err != nil { + h.serveIndexError(w, index.err) + return + } + downloadURL, err = h.findChartDownload(repository, upstreamURL, index, digest, filename) + } else { + downloadURL, err = h.findChartURL(charts, digest, filename) } - - downloadURL, err := h.findChartDownload(repository, upstreamURL, index, digest, filename) if err != nil { if errors.Is(err, errHelmChartNotFound) { http.NotFound(w, r) @@ -269,6 +287,10 @@ func (h *HelmHandler) findChartDownload(repository, upstreamURL string, index me if err != nil { return "", err } + return h.findChartURL(charts, digest, filename) +} + +func (h *HelmHandler) findChartURL(charts *helmChartLookup, digest, filename string) (string, error) { for _, chart := range charts.byDigest[digest] { if h.chartOnCooldown(chart.name, chart.created) { continue @@ -380,70 +402,35 @@ func (h *HelmHandler) buildChartLookup(upstreamURL string, body []byte) (*helmCh return lookup, nil } -// withholdChart checks a chart against the cooldown before it is served. -// findChartDownload already leaves out charts in cooldown, but a cached chart -// may have been fetched under an earlier policy. Charts are cached under the -// repository and digest rather than the chart name the policy is keyed by, so -// the name and publish time come from the index, indexed once per document -// (see publishIndex). -func (h *HelmHandler) withholdChart(w http.ResponseWriter, r *http.Request, repository, upstreamURL, digest string) bool { - if h.proxy.Cooldown == nil || !h.proxy.Cooldown.Enabled() { - return false - } - res := h.fetchIndex(r, repository, upstreamURL) - if res.err != nil { - h.proxy.Logger.Warn("cooldown: could not fetch Helm index for download check", - "repository", repository, "digest", digest, "error", res.err) - return false - } - index, err := publishIndex(h.proxy, "helm\x00"+h.indexCacheKey(repository, upstreamURL), res, - func(body []byte, _ string) (map[string]helmChartTime, error) { - return h.indexChartTimes(upstreamURL, body) - }) - if err != nil { - h.proxy.Logger.Warn("cooldown: could not read Helm index for download check", - "repository", repository, "digest", digest, "error", err) - return false - } - chart, ok := index[digest] - if !ok { - return false - } - return h.proxy.withholdInCooldown(w, downloadVersion{ - ecosystem: helmMetadataEcosystem, policyName: chart.name, version: digest, - lookup: func() (time.Time, bool, error) { return chart.created, !chart.created.IsZero(), nil }, +func (h *HelmHandler) downloadChartLookup(r *http.Request, repository, upstreamURL string) (*helmChartLookup, error) { + key := "helm\x00" + h.indexCacheKey(repository, upstreamURL) + return publishIndex(h.proxy, key, func() metadataResult { + return h.fetchIndex(r.Context(), repository, upstreamURL) + }, func(res metadataResult) (*helmChartLookup, error) { + if res.err != nil { + return nil, res.err + } + return h.chartLookup(repository, upstreamURL, res) }) } -// helmChartTime is the chart a release digest belongs to and when it was -// created. -type helmChartTime struct { - name string - created time.Time -} - -// indexChartTimes maps each release digest in an index to its chart. -func (h *HelmHandler) indexChartTimes(upstreamURL string, body []byte) (map[string]helmChartTime, error) { - _, entries, err := parseHelmIndex(body) - if err != nil { - return nil, err - } - index := make(map[string]helmChartTime) - for i := 0; i < len(entries.Content); i += 2 { - chartName := entries.Content[i].Value - releases := entries.Content[i+1] - if releases.Kind != yaml.SequenceNode { - continue - } - for _, release := range releases.Content { - chart, err := h.parseChartRelease(chartName, upstreamURL, release) - if err != nil { +func (h *HelmHandler) withholdChart(w http.ResponseWriter, charts *helmChartLookup, digest, filename string) bool { + withheld := false + for _, chart := range charts.byDigest[digest] { + for _, download := range chart.downloads { + if download.filename != filename { continue } - index[chart.digest] = helmChartTime{name: chartName, created: chart.created} + if !h.chartOnCooldown(chart.name, chart.created) { + return false + } + withheld = true } } - return index, nil + if withheld { + http.Error(w, ErrVersionNotFound.Error(), http.StatusNotFound) + } + return withheld } func (h *HelmHandler) chartOnCooldown(chartName string, created time.Time) bool { diff --git a/internal/handler/hex.go b/internal/handler/hex.go index 6836dadf..ceaf5879 100644 --- a/internal/handler/hex.go +++ b/internal/handler/hex.go @@ -6,7 +6,6 @@ import ( "encoding/json" "fmt" "net/http" - "net/url" "strings" "time" @@ -243,27 +242,8 @@ func parseHexReleases(body []byte) (publishTimes, error) { return times, nil } -// publishTime reads a release's inserted_at from the Hex HTTP API. The -// download check caches the response, unlike fetchFilteredVersions, so each -// download does not cost an upstream request. func (h *HexHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, error) { - apiURL := fmt.Sprintf("%s/api/packages/%s", h.apiURL, url.PathEscape(name)) - body, _, err := h.proxy.FetchOrCacheMetadata(r.Context(), "hex", "api/packages/"+name, apiURL) - if err != nil { - return time.Time{}, false, err - } - var pkg hexPackageAPI - if err := json.Unmarshal(body, &pkg); err != nil { - return time.Time{}, false, err - } - for _, release := range pkg.Releases { - if release.Version != version { - continue - } - insertedAt, err := time.Parse(time.RFC3339Nano, release.InsertedAt) - return insertedAt, err == nil, nil - } - return time.Time{}, false, nil + return h.proxy.lookupPublishTime(r.Context(), h.timedPackage(name), version) } // signedPayload decompresses a gzipped Signed message, with the metadata diff --git a/internal/handler/pub.go b/internal/handler/pub.go index 22540f14..001b0c66 100644 --- a/internal/handler/pub.go +++ b/internal/handler/pub.go @@ -94,8 +94,9 @@ func (h *PubHandler) handleDownload(w http.ResponseWriter, r *http.Request) { h.proxy.Logger.Debug("pub download request", "name", name, "version", version) - if h.proxy.withholdInCooldown(w, downloadVersion{ - ecosystem: "pub", policyName: name, storedName: name, version: version, + h.discardVersionPage(name, version, filename) + if h.proxy.withholdInCooldown(w, r, downloadVersion{ + ecosystem: "pub", policyName: name, storedName: name, version: version, cacheName: name, filename: filename, lookup: func() (time.Time, bool, error) { return h.publishTime(r, name, version) }, }) { return @@ -103,7 +104,6 @@ func (h *PubHandler) handleDownload(w http.ResponseWriter, r *http.Request) { downloadURL := fmt.Sprintf("%s/packages/%s/versions/%s%s", h.upstreamURL, url.PathEscape(name), url.PathEscape(version), pubArchiveExt) - h.discardVersionPage(name, version, filename) result, err := h.proxy.GetOrFetchArtifactFromURL( r.Context(), "pub", name, version, filename, downloadURL, ) From 61ac273862069436dac9412c366aa6dff9cec895 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Thu, 5 Nov 2026 20:00:19 +0000 Subject: [PATCH 6/6] Check primary metadata before allowing untimed downloads --- internal/handler/cargo.go | 5 +- internal/handler/conda.go | 5 +- internal/handler/cooldown_times.go | 28 +++--- internal/handler/download_cooldown.go | 9 +- .../handler/download_cooldown_cache_test.go | 89 +++++++++++++++++++ internal/handler/download_cooldown_test.go | 15 ++++ internal/handler/filename_download.go | 4 +- internal/handler/gem.go | 4 +- internal/handler/hex.go | 4 +- internal/handler/pub.go | 5 +- 10 files changed, 144 insertions(+), 24 deletions(-) diff --git a/internal/handler/cargo.go b/internal/handler/cargo.go index fa93a747..aca7416e 100644 --- a/internal/handler/cargo.go +++ b/internal/handler/cargo.go @@ -246,7 +246,10 @@ func (h *CargoHandler) handleDownload(w http.ResponseWriter, r *http.Request) { if h.proxy.withholdInCooldown(w, r, downloadVersion{ ecosystem: "cargo", policyName: name, storedName: name, version: version, cacheName: name, filename: filename, - lookup: func() (time.Time, bool, error) { return h.publishTime(r, name, version) }, + lookup: func() (time.Time, bool, bool, error) { + published, ok, err := h.publishTime(r, name, version) + return published, ok, false, err + }, }) { return } diff --git a/internal/handler/conda.go b/internal/handler/conda.go index de6ae8a8..1c14980a 100644 --- a/internal/handler/conda.go +++ b/internal/handler/conda.go @@ -145,7 +145,10 @@ func (h *CondaHandler) handleDownload(w http.ResponseWriter, r *http.Request) { cacheFilename := arch + "/" + filename if h.proxy.withholdInCooldown(w, r, downloadVersion{ ecosystem: "conda", policyName: name, version: version, cacheName: packageName, filename: cacheFilename, - lookup: func() (time.Time, bool, error) { return h.publishTime(r, channel, arch, filename) }, + lookup: func() (time.Time, bool, bool, error) { + published, ok, err := h.publishTime(r, channel, arch, filename) + return published, ok, false, err + }, }) { return } diff --git a/internal/handler/cooldown_times.go b/internal/handler/cooldown_times.go index b7a05162..b765f9a5 100644 --- a/internal/handler/cooldown_times.go +++ b/internal/handler/cooldown_times.go @@ -7,6 +7,7 @@ import ( "crypto/sha256" "encoding/hex" "fmt" + "slices" "strings" "time" ) @@ -16,19 +17,24 @@ import ( // through. type publishTimes map[string]time.Time -func (p *Proxy) lookupPublishTime(ctx context.Context, t timedMetadata, version string) (time.Time, bool, error) { - var times publishTimes - validate := func(body []byte) error { +func (p *Proxy) lookupPublishTime(ctx context.Context, t timedMetadata, version string, listed func(metadataResult) ([]string, error)) (time.Time, bool, bool, error) { + var versions []string + docs, err := p.fetchTimedMetadata(ctx, t, func(doc metadataResult) ([]string, error) { var err error - times, err = t.parseTimes(body) - return err - } - res := p.lookupMetadata(ctx, t.ecosystem, t.timesKey, t.timesURL, t.timesAccept, "", validate) - if res.err != nil { - return time.Time{}, false, res.err + versions, err = listed(doc) + return versions, err + }) + switch { + case docs.doc.err != nil: + return time.Time{}, false, false, docs.doc.err + case docs.times.err != nil: + return time.Time{}, false, false, docs.times.err + case err != nil: + return time.Time{}, false, false, err } - published, ok := times[version] - return published, ok && !published.IsZero(), nil + published, ok := docs.published[version] + withholdUnknown := !ok && docs.withholdUnknown && slices.Contains(versions, version) + return published, ok && !published.IsZero(), withholdUnknown, nil } // timedMetadata describes metadata whose publish times live in a second diff --git a/internal/handler/download_cooldown.go b/internal/handler/download_cooldown.go index 47d8b704..c491ef3d 100644 --- a/internal/handler/download_cooldown.go +++ b/internal/handler/download_cooldown.go @@ -9,8 +9,9 @@ import ( ) // publishTimeLookup reads a version's publish time from upstream metadata. ok -// is false when the metadata has no usable time for the version. -type publishTimeLookup func() (published time.Time, ok bool, err error) +// is false when the metadata has no usable time for the version. withholdUnknown +// marks a listed version missing from stale publish-time metadata. +type publishTimeLookup func() (published time.Time, ok, withholdUnknown bool, err error) // downloadVersion names the version an artifact download is for, so its // cooldown can be checked before the artifact is served. @@ -62,12 +63,12 @@ func (p *Proxy) versionInCooldown(d downloadVersion) (bool, error) { } } - published, ok, err := d.lookup() + published, ok, withholdUnknown, err := d.lookup() if err != nil { return false, err } if !ok { - return false, nil + return withholdUnknown, nil } if versionPURL != "" { diff --git a/internal/handler/download_cooldown_cache_test.go b/internal/handler/download_cooldown_cache_test.go index be488ded..d51b862d 100644 --- a/internal/handler/download_cooldown_cache_test.go +++ b/internal/handler/download_cooldown_cache_test.go @@ -1,8 +1,12 @@ package handler import ( + "errors" + "fmt" "net/http" + "net/http/httptest" "testing" + "time" ) func TestGemDownloadSharesPublishTimeMetadata(t *testing.T) { @@ -15,6 +19,91 @@ func TestGemDownloadSharesPublishTimeMetadata(t *testing.T) { } } +type downloadPublishTimeCase struct { + name string + down bool + absent bool + untimed bool + refresh bool + wantStatus int +} + +func TestDownloadCooldownRefreshesMissingPublishTimes(t *testing.T) { + for _, c := range downloadCooldownCases() { + if c.ecosystem != "gem" && c.ecosystem != "hex" { + continue + } + for _, scenario := range []downloadPublishTimeCase{ + {name: "refreshed times", refresh: true, wantStatus: http.StatusNotFound}, + {name: "stale times and listed version", down: true, wantStatus: http.StatusNotFound}, + {name: "stale times and absent version", down: true, absent: true, wantStatus: http.StatusOK}, + {name: "fresh times without version", wantStatus: http.StatusOK}, + {name: "stale times with untimed version", down: true, untimed: true, wantStatus: http.StatusOK}, + } { + for _, cached := range []bool{false, true} { + if !cached && scenario.wantStatus == http.StatusOK { + continue + } + t.Run(fmt.Sprintf("%s/%s/cached=%t", c.ecosystem, scenario.name, cached), func(t *testing.T) { + assertDownloadPublishTime(t, c, scenario, cached) + }) + } + } + } +} + +func assertDownloadPublishTime(t *testing.T, c downloadCooldownCase, scenario downloadPublishTimeCase, cached bool) { + t.Helper() + u := newCooldownCountingUpstream(t) + proxy := cooldownCacheProxy(t, u) + fetcher := proxy.Fetcher.(*mockFetcher) + fetcher.fetchErr = errors.New("unchecked artifact download") + h := c.handler(proxy, u.URL) + docPath, timesPath := "/info/demo", "/api/v1/versions/demo.json" + if c.ecosystem == "hex" { + docPath, timesPath = "/packages/demo", "/api/packages/demo" + } + oldTimes := map[string]time.Time{cooldownOldVersion: time.Now().Add(-30 * 24 * time.Hour)} + if scenario.untimed { + oldTimes[cooldownRecentVersion] = time.Time{} + } + setDownloadMetadata(t, u, c, docPath, oldTimes) + setDownloadMetadata(t, u, c, timesPath, oldTimes) + wantStatus(t, serveGet(t, h, docPath), http.StatusOK, "") + + newTimes := map[string]time.Time{ + cooldownOldVersion: oldTimes[cooldownOldVersion], + cooldownUntimed: time.Now().Add(-time.Minute), + } + if !scenario.absent { + newTimes[cooldownRecentVersion] = time.Now().Add(-time.Hour) + } + setDownloadMetadata(t, u, c, docPath, newTimes) + expireCachedMetadata(t, proxy, c.ecosystem, docPath[1:]) + if scenario.refresh { + setDownloadMetadata(t, u, c, timesPath, newTimes) + } + u.setDown(timesPath, scenario.down) + if cached { + c.seedCachedArtifact(t, proxy, cooldownRecentVersion) + } + wantStatus(t, serveGet(t, h, c.path(cooldownRecentVersion)), scenario.wantStatus, "") + if fetcher.fetchCalled { + t.Error("fetched an unchecked artifact") + } + if n, _ := u.counts(timesPath); n != 2 { + t.Errorf("publish times fetched %d times, want 2", n) + } +} + +func setDownloadMetadata(t *testing.T, u *cooldownCountingUpstream, c downloadCooldownCase, path string, times map[string]time.Time) { + t.Helper() + w := httptest.NewRecorder() + c.metadata(w, httptest.NewRequest(http.MethodGet, path, nil), times) + wantStatus(t, w, http.StatusOK, "") + u.set(path, w.Body.Bytes()) +} + func TestHexDownloadSharesPublishTimeMetadata(t *testing.T) { u, proxy, h, _ := hexCacheSetup(t, 0) wantStatus(t, serveGet(t, h, hexPackagePath), http.StatusOK, "") diff --git a/internal/handler/download_cooldown_test.go b/internal/handler/download_cooldown_test.go index a3c89ae9..1fd915f8 100644 --- a/internal/handler/download_cooldown_test.go +++ b/internal/handler/download_cooldown_test.go @@ -137,6 +137,13 @@ func cargoCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[str } func gemCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[string]time.Time) { + if r.URL.Path == "/info/demo" { + _, _ = fmt.Fprintln(w, "---") + for version := range times { + _, _ = fmt.Fprintf(w, "%s-x86_64-linux |checksum:a\n", version) + } + return + } if r.URL.Path != "/api/v1/versions/demo.json" { http.NotFound(w, r) return @@ -153,6 +160,14 @@ func gemCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[strin } func hexCooldownMetadata(w http.ResponseWriter, r *http.Request, times map[string]time.Time) { + if r.URL.Path == "/packages/demo" { + var versions []string + for version := range times { + versions = append(versions, version) + } + _, _ = w.Write(buildHexSigned(buildHexPackage("demo", versions))) + return + } if r.URL.Path != "/api/packages/demo" { http.NotFound(w, r) return diff --git a/internal/handler/filename_download.go b/internal/handler/filename_download.go index 373c8f45..c37d07d8 100644 --- a/internal/handler/filename_download.go +++ b/internal/handler/filename_download.go @@ -15,7 +15,7 @@ type filenameDownload struct { parse func(string) (name, version string) // publishTime reads a version's publish time from upstream metadata for // the download-time cooldown check. - publishTime func(r *http.Request, name, version string) (time.Time, bool, error) + publishTime func(r *http.Request, name, version string) (time.Time, bool, bool, error) } func (p *Proxy) handleFilenameDownload(w http.ResponseWriter, r *http.Request, d filenameDownload) { @@ -36,7 +36,7 @@ func (p *Proxy) handleFilenameDownload(w http.ResponseWriter, r *http.Request, d if p.withholdInCooldown(w, r, downloadVersion{ ecosystem: d.ecosystem, policyName: name, storedName: name, version: version, cacheName: name, filename: filename, - lookup: func() (time.Time, bool, error) { return d.publishTime(r, name, version) }, + lookup: func() (time.Time, bool, bool, error) { return d.publishTime(r, name, version) }, }) { return } diff --git a/internal/handler/gem.go b/internal/handler/gem.go index e9ee55af..fa7c24ba 100644 --- a/internal/handler/gem.go +++ b/internal/handler/gem.go @@ -305,8 +305,8 @@ func parseGemVersions(body []byte) (publishTimes, error) { return times, nil } -func (h *GemHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, error) { - return h.proxy.lookupPublishTime(r.Context(), h.timedIndex(name), version) +func (h *GemHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, bool, error) { + return h.proxy.lookupPublishTime(r.Context(), h.timedIndex(name), version, h.indexVersions) } // proxyCached forwards a metadata request with caching. diff --git a/internal/handler/hex.go b/internal/handler/hex.go index ceaf5879..d394c4df 100644 --- a/internal/handler/hex.go +++ b/internal/handler/hex.go @@ -242,8 +242,8 @@ func parseHexReleases(body []byte) (publishTimes, error) { return times, nil } -func (h *HexHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, error) { - return h.proxy.lookupPublishTime(r.Context(), h.timedPackage(name), version) +func (h *HexHandler) publishTime(r *http.Request, name, version string) (time.Time, bool, bool, error) { + return h.proxy.lookupPublishTime(r.Context(), h.timedPackage(name), version, h.packageVersions) } // signedPayload decompresses a gzipped Signed message, with the metadata diff --git a/internal/handler/pub.go b/internal/handler/pub.go index 001b0c66..31e00d8b 100644 --- a/internal/handler/pub.go +++ b/internal/handler/pub.go @@ -97,7 +97,10 @@ func (h *PubHandler) handleDownload(w http.ResponseWriter, r *http.Request) { h.discardVersionPage(name, version, filename) if h.proxy.withholdInCooldown(w, r, downloadVersion{ ecosystem: "pub", policyName: name, storedName: name, version: version, cacheName: name, filename: filename, - lookup: func() (time.Time, bool, error) { return h.publishTime(r, name, version) }, + lookup: func() (time.Time, bool, bool, error) { + published, ok, err := h.publishTime(r, name, version) + return published, ok, false, err + }, }) { return }