Check cooldown on artifact downloads for cargo, gem, hex, pub, conda and helm

internal/database/database_test.go

@@ -1164,3 +1164,46 @@ func TestVersionPublishedAtPreserved(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)
+			}
+			// 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)
+			}
+		}
+
+		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")
+	})
+}

internal/database/queries.go

@@ -183,7 +183,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)
@@ -216,7 +216,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)
 	}

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,

internal/handler/conda.go

@@ -2,8 +2,11 @@ package handler
 
 import (
 	"bufio"
+	"bytes"
 	"compress/gzip"
 	"encoding/json"
+	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"slices"
@@ -127,6 +130,15 @@ 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, 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.
@@ -361,6 +373,101 @@ 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.
+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)
+	}
+	if err != nil {
+		return time.Time{}, false, err
+	}
+	published, ok := index[filename]
+	return published, ok, nil
+}
+
+// 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(reader)
+		if err != nil {
+			return nil, err
+		}
+		reader = decoded
+	}
+	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
+	}
+	req.Header.Set(headerAcceptEncoding, "gzip")
+	resp, err := h.proxy.HTTPClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	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
+}
+
 // proxyCached forwards a metadata request with caching.
 func (h *CondaHandler) proxyCached(w http.ResponseWriter, r *http.Request) {
 	cacheKey := h.metadataCacheKey(r.URL.Path)

internal/handler/download_cooldown.go

@@ -0,0 +1,144 @@
+package handler
+
+import (
+	"net/http"
+	"sync"
+	"time"
+
+	"golang.org/x/sync/singleflight"
+)
+
+// 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)
+}
+
+// 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
+}