diff --git a/docs/configuration.md b/docs/configuration.md index d887509a..9c170573 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1042,6 +1042,18 @@ A repository's files are cached under its name and a hash of its upstream URL, so pointing a repository at a different URL stops serving what the old one sent. +In every Maven repository, `maven-metadata.xml` follows `metadata_ttl`, and +its `.md5`, `.sha1`, `.sha256` and `.sha512` files are computed from the +cached metadata rather than fetched, so they always match it. Other files are +cached for good, except SNAPSHOT files named with `SNAPSHOT` rather than a +build timestamp (`lib-1.0-SNAPSHOT.jar`, as JitPack and non-unique snapshot +repositories publish them), which each build replaces: those are fetched +again for every `GET`, through the denylist, cooldown, scanning and the +Plugin Portal fallback like any other file. The new copy replaces the cached +one once stored; while upstream can't be reached the cached copy is served, +and `HEAD` is answered from cache. Timestamped SNAPSHOT files +(`lib-1.0-20240101.120000-1.jar`) never change and are cached. + This field has no environment variable, as with the other named upstream maps. `PROXY_UPSTREAM_MAVEN` still sets `upstream.maven`. @@ -1242,7 +1254,7 @@ Currently supported for npm, PyPI, pub.dev, Composer, Cargo, NuGet, Conda, RubyG Go has no publish time. The module proxy's `.info` gives the commit time of the revision a version points at, so a version tagged on an old commit counts as old. The proxy reads each listed version's `.info` once, stores the time, and leaves versions in cooldown out of `@v/list` and `@latest`. When upstream's `@latest` is in cooldown, it serves the newest release out of cooldown instead. Requests for a version in cooldown, including its `.info`, `.mod` and `.zip`, and a branch or commit query that resolves to one, get 403 rather than 404, because the go command moves on to the next `GOPROXY` entry, such as `direct`, after a 404. If a version's `.info` can't be fetched, the list or download fails with 502 instead of letting the version through. -Maven metadata has no publish time per version. The proxy uses the `Last-Modified` header of each version's `.pom`, which on Maven Central is the time the version was deployed. It reads the header once per version and stores the time, for each repository under `upstream.maven_repositories` separately: from the response when the `.pom` or Gradle `.module` itself is downloaded, which clients do before a version's other files, and otherwise with a `HEAD` request. A `.pom` or `.module` refused this way is still cached, so it counts toward `max_size` before the version leaves cooldown. To keep requests to the upstream down, it only checks versions that can still be in cooldown: none when the metadata's `` is older than the cooldown, and otherwise the versions from the end of the list back to the first one that is old enough. Versions in cooldown are left out of `maven-metadata.xml`, `` and `` are set to the last versions left, and the metadata's `.md5`, `.sha1`, `.sha256` and `.sha512` files are computed from what is served. Requests for any file of a version in cooldown get 403, as for a version on the denylist. If a `.pom` can't be checked, the metadata or download fails with 502 instead of letting the version through, or with 503 and `Retry-After` when the upstream rate-limits the proxy; a version without a `.pom` is let through. +Maven metadata has no publish time per version. The proxy uses the `Last-Modified` header of each version's `.pom`, which on Maven Central is the time the version was deployed. It reads the header once per version and stores the time, for each repository under `upstream.maven_repositories` separately: from the response when the `.pom` or Gradle `.module` itself is downloaded, which clients do before a version's other files, and otherwise with a `HEAD` request. A `.pom` or `.module` refused this way is still cached, so it counts toward `max_size` before the version leaves cooldown. To keep requests to the upstream down, it only checks versions that can still be in cooldown: none when the metadata's `` is older than the cooldown, and otherwise the versions from the end of the list back to the first one that is old enough. Versions in cooldown are left out of `maven-metadata.xml`, `` and `` are set to the last versions left, and the metadata's `.md5`, `.sha1`, `.sha256` and `.sha512` files are computed from what is served, as they are without cooldown. Requests for any file of a version in cooldown get 403, as for a version on the denylist. If a `.pom` can't be checked, the metadata or download fails with 502 instead of letting the version through, or with 503 and `Retry-After` when the upstream rate-limits the proxy; a version without a `.pom` is let through. Maven caveats: @@ -1509,7 +1521,7 @@ cache_metadata: true Or via environment variable: `PROXY_CACHE_METADATA=true`. -Some metadata never changes once published, and is cached like an artifact, with or without `cache_metadata`, and served without revalidation whatever `metadata_ttl` says: a Go module's `.mod` and `.info` for a canonical version (a release, pre-release, pseudo-version or `+incompatible` version; the checksum database fixes their content), and RPM repodata files named after their own checksum (`repodata/-primary.xml.gz`). A stored copy is checked against the digest recorded for it and fetched again if it doesn't match. Go branch and commit queries, `@v/list` and `@latest`, and RPM `repomd.xml` and plainly named repodata files follow `metadata_ttl` as before. A checksum-named repodata file over `metadata_max_size` is streamed from upstream uncached. +Some metadata never changes once published, and is cached like an artifact, with or without `cache_metadata`, and served without revalidation whatever `metadata_ttl` says: a Go module's `.mod` and `.info` for a canonical version (a release, pre-release, pseudo-version or `+incompatible` version; the checksum database fixes their content), and RPM repodata files named after their own checksum (`repodata/-primary.xml.gz`). A stored copy is checked against the digest recorded for it and fetched again if it doesn't match. Go branch and commit queries, `@v/list` and `@latest`, and RPM `repomd.xml` follow `metadata_ttl` as before; plainly named repodata files follow the cached `repomd.xml` (see [Metadata TTL](#metadata-ttl)). A checksum-named repodata file over `metadata_max_size` is streamed from upstream uncached. Composer archive downloads look up their upstream URL in the same cached metadata, so an archive is resolved without asking Packagist again and stays installable while it is down. Only a version or file the cached metadata does not list yet is looked up upstream. With or without `cache_metadata`, a cached archive of a tagged release from the default repository is served without any metadata lookup; with cooldown enabled it is checked against the publish time recorded when the archive was first resolved. Branch (dev) versions, routed packages and archives picked by a query string such as `?sha=` are always resolved from the metadata. @@ -1529,7 +1541,9 @@ Or via environment variable: `PROXY_METADATA_TTL=10m`. Set to `"0"` to always revalidate with upstream (ETag-based conditional requests still avoid re-downloading unchanged content). -When upstream is unreachable and the cached entry is past its TTL, the proxy serves the stale cached copy with a `Warning: 110 - "Response is Stale"` header so clients can tell the data may be outdated. +When upstream is unreachable and the cached entry is past its TTL (with a `metadata_ttl` of `"0"`, on every such request), the proxy serves the stale cached copy with a `Warning: 110 - "Response is Stale"` header so clients can tell the data may be outdated. The `ETag` and `Last-Modified` sent with a document are always those of the bytes served, so a document fetched from upstream but not stored, for example because storage failed, carries the upstream's validators rather than the older cached copy's. + +APT and RPM clients fetch a release file first (`InRelease` or `Release`, `repomd.xml`) and reject an index file whose checksum doesn't match it. So an index file the cached release file lists (`Packages`, `Sources`, a plainly named `primary.xml.gz`, …) does not follow its own `metadata_ttl`: its cached copy is served while it matches the cached release file's sha256, however old, and otherwise it is fetched from upstream. If upstream's copy doesn't match either, the repository has changed since the release file was cached; that copy is served and the release file is expired, so the next `apt update` or `dnf makecache` gets a matching pair. Where a suite has both `InRelease` and `Release` cached, the one fetched last decides. The release file itself follows `metadata_ttl`. Index files it doesn't list, such as `by-hash` paths, follow `metadata_ttl` too. ### Change feeds diff --git a/internal/handler/arch.go b/internal/handler/arch.go index b53f25c0..da1a6f1c 100644 --- a/internal/handler/arch.go +++ b/internal/handler/arch.go @@ -199,12 +199,12 @@ func (h *ArchHandler) handleDatabase(w http.ResponseWriter, r *http.Request, rep d := h.database(repository, upstreamURL, path) if _, hit := p.cachedMetadataState(r.Context(), archEcosystem, d.dbKey, nil); hit != nil { metrics.RecordMetadataCacheHit(archEcosystem, len(hit.body)) - h.writeMetadata(w, r, d.dbKey, *hit) + h.writeMetadata(w, r, *hit) return } p.recordMetadataCacheMiss(archEcosystem) res := h.sharedRefresh(r.Context(), d, true) - h.writeMetadata(w, r, d.dbKey, res) + h.writeMetadata(w, r, res) } // handleDatabaseSignature serves the signature stored for the cached @@ -218,7 +218,7 @@ func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Req d := h.database(repository, upstreamURL, strings.TrimSuffix(path, archSignatureSuffix)) if res, ok := h.cachedSignature(r.Context(), d); ok { metrics.RecordMetadataCacheHit(archEcosystem, len(res.body)) - h.writeMetadata(w, r, d.sigKey, res) + h.writeMetadata(w, r, res) return } p.recordMetadataCacheMiss(archEcosystem) @@ -242,7 +242,7 @@ func (h *ArchHandler) handleDatabaseSignature(w http.ResponseWriter, r *http.Req } return metadataResult{err: errArchSignatureUnpaired} }) - h.writeMetadata(w, r, d.sigKey, res) + h.writeMetadata(w, r, res) } var errArchSignatureUnpaired = errors.New("database signature could not be paired with the database") @@ -324,7 +324,7 @@ func (h *ArchHandler) fetchDatabase(ctx context.Context, d archDatabase, entry * return metadataResult{err: fmt.Errorf("upstream failed and no cached metadata: %w", err)}, nil } p.Logger.Warn("upstream metadata fetch failed, checking cache", "ecosystem", archEcosystem, "key", d.dbKey, "error", err) - return h.readDatabase(ctx, p.currentMetadataEntry(archEcosystem, d.dbKey, entry), err) + return h.readDatabase(ctx, p.currentMetadataEntry(archEcosystem, d.dbKey, entry), err, true) } // storeDatabase caches meta under its digest and returns the database the @@ -345,15 +345,15 @@ func (h *ArchHandler) storeDatabase(ctx context.Context, d archDatabase, meta *u row, _ := p.DB.GetMetadataCache(archEcosystem, d.dbKey) if row == nil { // Nothing cached at all: serve it like an uncached proxy would. - return metadataResult{body: meta.body, contentType: meta.contentType, contentEncoding: meta.contentEncoding}, nil + return fetchedResult(meta, ""), nil } if row.StoragePath != path { - return h.readDatabase(ctx, row, errArchDatabaseNotCached) + return h.readDatabase(ctx, row, errArchDatabaseNotCached, false) } if previous != nil && previous.StoragePath != path { _ = p.Storage.Delete(ctx, previous.StoragePath) } - return metadataResult{body: meta.body, contentType: meta.contentType, contentEncoding: meta.contentEncoding}, row + return fetchedResult(meta, row.ContentDigest.String), row } // blobExists reports whether path is stored. Size eviction can remove a blob @@ -377,13 +377,18 @@ func (h *ArchHandler) touchDatabase(row *database.MetadataCacheEntry, meta *upst var errArchDatabaseNotCached = errors.New("database could not be cached") // readDatabase returns the cached database row's bytes, or cause if they are -// unreadable. -func (h *ArchHandler) readDatabase(ctx context.Context, row *database.MetadataCacheEntry, cause error) (metadataResult, *database.MetadataCacheEntry) { +// unreadable. stale marks them as served because the upstream failed with +// cause. +func (h *ArchHandler) readDatabase(ctx context.Context, row *database.MetadataCacheEntry, cause error, stale bool) (metadataResult, *database.MetadataCacheEntry) { body, contentType, err := h.proxy.readCachedMetadata(ctx, row, nil) if err != nil { return metadataResult{err: fmt.Errorf("cached database unusable (%v): %w", err, cause)}, nil } - return metadataResult{body: body, contentType: contentType, contentEncoding: row.ContentEncoding.String}, row + res := metadataResult{body: body, contentType: contentType, contentEncoding: row.ContentEncoding.String} + if stale { + res.stale, res.upstreamErr = true, cause + } + return res.describedBy(row), row } // lock serializes cache writes for one database and its signature, so a row @@ -433,7 +438,7 @@ func (h *ArchHandler) storedSignature(ctx context.Context, d archDatabase, dbDig if err != nil { return metadataResult{}, false } - return metadataResult{body: body, contentType: contentType, contentEncoding: entry.ContentEncoding.String}, true + return metadataResult{body: body, contentType: contentType, contentEncoding: entry.ContentEncoding.String}.describedBy(entry), true } return metadataResult{}, false } @@ -478,7 +483,7 @@ func archDigest(body []byte) string { return "sha256:" + hex.EncodeToString(digest[:]) } -func (h *ArchHandler) writeMetadata(w http.ResponseWriter, r *http.Request, cacheKey string, res metadataResult) { +func (h *ArchHandler) writeMetadata(w http.ResponseWriter, r *http.Request, res metadataResult) { if res.err != nil { if errors.Is(res.err, ErrUpstreamNotFound) { http.Error(w, "not found", http.StatusNotFound) @@ -491,7 +496,7 @@ func (h *ArchHandler) writeMetadata(w http.ResponseWriter, r *http.Request, cach http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) return } - h.proxy.writeMetadataCachedResponseWithEncoding(w, r, archEcosystem, cacheKey, res.body, res.contentType, res.contentEncoding) + writeMetadataResult(w, r, res) } // metadataCacheKey hashes repository, upstream URL and path, as APKHandler does. diff --git a/internal/handler/arch_test.go b/internal/handler/arch_test.go index 3affad5c..02c87fa3 100644 --- a/internal/handler/arch_test.go +++ b/internal/handler/arch_test.go @@ -1,6 +1,7 @@ package handler import ( + "database/sql" "fmt" "net/http" "net/http/httptest" @@ -648,3 +649,55 @@ func serveArchRequest(h *ArchHandler, method, target string, header http.Header) h.Routes().ServeHTTP(w, req) return w } + +// TestArchHandler_DatabaseValidatorsDescribeBody checks a fetched database +// goes out with its own Last-Modified, not that of the cache row, which +// another instance may have rewritten while the signature was fetched. +func TestArchHandler_DatabaseValidatorsDescribeBody(t *testing.T) { + const dbPath, sigPath = "/core/os/x86_64/core.db", "/core/os/x86_64/core.db.sig" + fetched := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) + rewritten := fetched.Add(time.Hour) + upstream := newArchUpstream(t, map[string][]byte{dbPath: []byte("core-db-v1"), sigPath: []byte("sig")}) + upstream.modified[dbPath] = fetched + proxy := newArchTestProxy(t, upstream) + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + key := h.metadataCacheKey("archlinux", upstream.URL, "core/os/x86_64/core.db") + upstream.served = func(path string) { + if path != sigPath { + return + } + entry, err := proxy.DB.GetMetadataCache(archEcosystem, key) + if err != nil || entry == nil { + t.Errorf("database row while fetching the signature: %v", err) + return + } + entry.LastModified = sql.NullTime{Time: rewritten, Valid: true} + if err := proxy.DB.UpsertMetadataCache(entry); err != nil { + t.Error(err) + } + } + + w := serveArchRequest(h, http.MethodGet, "/archlinux"+dbPath, nil) + assertArchBody(t, w, "database", []byte("core-db-v1")) + if got := w.Header().Get("Last-Modified"); got != fetched.Format(http.TimeFormat) { + t.Errorf("Last-Modified = %q, want the fetched database's %q", got, fetched.Format(http.TimeFormat)) + } +} + +// TestArchHandler_StaleDatabaseMarkedWithoutTTL checks the stale fallback +// carries the Warning header with metadata_ttl 0. +func TestArchHandler_StaleDatabaseMarkedWithoutTTL(t *testing.T) { + db := []byte("core-db-bytes") + upstream := newArchUpstream(t, map[string][]byte{"/core/os/x86_64/core.db": db}) + proxy := newArchTestProxy(t, upstream) + proxy.MetadataTTL = 0 + h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"archlinux": upstream.URL}) + + assertArchBody(t, serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db", nil), "seed", db) + upstream.available.Store(false) + w := serveArchRequest(h, http.MethodGet, "/archlinux/core/os/x86_64/core.db", nil) + assertArchBody(t, w, "stale", db) + if got := w.Header().Get("Warning"); got != staleWarning { + t.Errorf("Warning = %q, want %q", got, staleWarning) + } +} diff --git a/internal/handler/conda_shards.go b/internal/handler/conda_shards.go index 8269f1db..dbfaee2b 100644 --- a/internal/handler/conda_shards.go +++ b/internal/handler/conda_shards.go @@ -58,11 +58,12 @@ func (h *CondaHandler) handleShardIndex(w http.ResponseWriter, r *http.Request) upstreamURL := h.upstreamURL + r.URL.Path cacheKey := h.metadataCacheKey(r.URL.Path) - body, contentType, contentEncoding, err := h.proxy.fetchOrCacheMetadata(r.Context(), "conda", cacheKey, upstreamURL, "identity", nil, "*/*") - if err != nil { - h.proxy.writeMetadataFetchError(w, err) + res := h.proxy.lookupMetadata(r.Context(), "conda", cacheKey, upstreamURL, "*/*", "identity", nil) + if res.err != nil { + h.proxy.writeMetadataFetchError(w, res.err) return } + body := res.body if !h.cooldownEnabled() { index, err := decodeCondaShardIndex(body) @@ -74,7 +75,7 @@ func (h *CondaHandler) handleShardIndex(w http.ResponseWriter, r *http.Request) http.NotFound(w, r) return } - h.proxy.writeMetadataCachedResponseWithEncoding(w, r, "conda", cacheKey, body, contentType, contentEncoding) + writeMetadataResult(w, r, res) return } @@ -121,7 +122,7 @@ func (h *CondaHandler) handleCooldownShard(w http.ResponseWriter, r *http.Reques } upstreamPath := "/" + channel + "/" + arch + "/" + dir + hex.EncodeToString(masked) + condaShardSuffix - body, _, _, err := h.proxy.fetchOrCacheMetadata(r.Context(), "conda", h.metadataCacheKey(upstreamPath), h.upstreamURL+upstreamPath, "identity", nil, "*/*") + body, _, err := h.proxy.fetchOrCacheMetadata(r.Context(), "conda", h.metadataCacheKey(upstreamPath), h.upstreamURL+upstreamPath, "identity", nil, "*/*") if err != nil { h.proxy.writeMetadataFetchError(w, err) return diff --git a/internal/handler/debian.go b/internal/handler/debian.go index 565f181a..bcb04396 100644 --- a/internal/handler/debian.go +++ b/internal/handler/debian.go @@ -5,6 +5,7 @@ import ( "encoding/hex" "fmt" "net/http" + pathpkg "path" "regexp" "sort" "strings" @@ -187,13 +188,49 @@ func (h *DebianHandler) handlePackageDownload( // handleMetadata serves repository metadata files through the metadata cache, // so they keep resolving while the upstream archive is unreachable. +// Index files the cached InRelease or Release lists follow that listing (see +// proxyListedIndex). func (h *DebianHandler) handleMetadata( w http.ResponseWriter, r *http.Request, repository, upstreamURL, path string, ) { cacheKey := h.metadataCacheKeyFor(repository, upstreamURL, path) - h.proxy.ProxyCached(w, r, fmt.Sprintf("%s/%s", upstreamURL, path), "debian", cacheKey, "*/*") + fileURL := fmt.Sprintf("%s/%s", upstreamURL, path) + switch pathpkg.Base(path) { + case "InRelease", "Release", "Release.gpg": + h.proxy.ProxyCached(w, r, fileURL, "debian", cacheKey, "*/*") + return + } + release, want := h.releaseListing(r, repository, upstreamURL, path) + h.proxy.proxyListedIndex(w, r, fileURL, "debian", cacheKey, release, want) +} + +// releaseListing finds the cached release file of the suite path belongs +// to and the sha256 it lists for path, nearest directory first, skipping +// release files that list nothing for path (such as the legacy +// per-component Release). Where both InRelease and Release list it, the one +// fetched last wins: apt reads InRelease and debootstrap Release, and a +// client asking for the index has most likely just fetched one of them. +func (h *DebianHandler) releaseListing(r *http.Request, repository, upstreamURL, path string) (*cachedRelease, string) { + for dir := pathpkg.Dir(path); strings.HasPrefix(dir, "dists/"); dir = pathpkg.Dir(dir) { + var found *cachedRelease + var want string + for _, name := range []string{"InRelease", "Release"} { + release := h.proxy.readCachedRelease(r.Context(), "debian", h.metadataCacheKeyFor(repository, upstreamURL, dir+"/"+name)) + if release == nil { + continue + } + sum := debianReleaseSHA256(release.body, strings.TrimPrefix(path, dir+"/")) + if sum != "" && (found == nil || release.fetchedAt.After(found.fetchedAt)) { + found, want = release, sum + } + } + if found != nil { + return found, want + } + } + return nil, "" } // artifactCacheFilename scopes a package by repository, since the same diff --git a/internal/handler/download_test.go b/internal/handler/download_test.go index 216a2798..c53585e2 100644 --- a/internal/handler/download_test.go +++ b/internal/handler/download_test.go @@ -1,7 +1,9 @@ package handler import ( + "crypto/sha1" "database/sql" + "encoding/hex" "fmt" "io" "net/http" @@ -1074,8 +1076,7 @@ func TestMavenHandler_GradlePluginMarkerFallbackAndCache(t *testing.T) { func TestMavenHandler_GradlePluginMarkerMetadataFallback(t *testing.T) { paths := map[string]string{ - "/com/diffplug/spotless/com.diffplug.spotless.gradle.plugin/maven-metadata.xml": "", - "/com/diffplug/spotless/com.diffplug.spotless.gradle.plugin/maven-metadata.xml.sha1": "sha1", + "/com/diffplug/spotless/com.diffplug.spotless.gradle.plugin/maven-metadata.xml": "", } primaryHits := map[string]int{} @@ -1131,6 +1132,18 @@ func TestMavenHandler_GradlePluginMarkerMetadataFallback(t *testing.T) { t.Fatalf("GET %s did not hit plugin portal fallback", reqPath) } } + + // The checksum is computed from the metadata served from the fallback. + resp, err := http.Get(srv.URL + "/com/diffplug/spotless/com.diffplug.spotless.gradle.plugin/maven-metadata.xml.sha1") + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + sum := sha1.Sum([]byte("")) //nolint:gosec // Maven publishes SHA-1 checksums. + if resp.StatusCode != http.StatusOK || string(body) != hex.EncodeToString(sum[:]) { + t.Fatalf("GET maven-metadata.xml.sha1: %d %q, want %x", resp.StatusCode, body, sum) + } } func TestMavenHandler_GradlePluginChecksumFallbackToPluginPortal(t *testing.T) { diff --git a/internal/handler/go_sumdb.go b/internal/handler/go_sumdb.go index 8b382066..e96e5abd 100644 --- a/internal/handler/go_sumdb.go +++ b/internal/handler/go_sumdb.go @@ -383,7 +383,9 @@ func (h *GoHandler) proxySumDB(w http.ResponseWriter, r *http.Request, name, ups // the upstream timeout before the fetch falls back to it. if h.sumDBOffline && h.sumDBProbes.recentlyFailed(name) { if entry, body, contentType, ok := h.readCachedSumDB(r.Context(), cacheKey, check); ok { - p.writeMetadataCachedResponseWithEncoding(w, r, goEcosystem, cacheKey, body, contentType, entry.ContentEncoding.String) + cm := p.cachedMetaFor(entry) + cm.stale = true // upstream was not asked + writeMetadataBody(w, r, cm, body, contentType, entry.ContentEncoding.String) return } } @@ -416,7 +418,7 @@ func (h *GoHandler) proxySumDB(w http.ResponseWriter, r *http.Request, name, ups return } - p.writeMetadataCachedResponseWithEncoding(w, r, goEcosystem, cacheKey, res.body, res.contentType, res.contentEncoding) + writeMetadataResult(w, r, res) } // fetchSumDB fetches a checksum database response through the metadata diff --git a/internal/handler/go_sumdb_test.go b/internal/handler/go_sumdb_test.go index 6a0c47b4..d47e432c 100644 --- a/internal/handler/go_sumdb_test.go +++ b/internal/handler/go_sumdb_test.go @@ -985,3 +985,26 @@ func TestGoSumDB_ClientRoundTrip(t *testing.T) { t.Fatalf("lookup from the proxy cache with the database down: %v", err) } } + +// TestGoSumDB_OfflineCopyMarkedStaleWithoutTTL: with metadata_ttl 0 the +// cached copy offline mode serves while the database is unreachable still +// carries the Warning header. +func TestGoSumDB_OfflineCopyMarkedStaleWithoutTTL(t *testing.T) { + const lookup = "/lookup/golang.org/x/text@v0.14.0" + db := newFakeSumDB(t, map[string]string{lookup: testSumDBRecord}) + h := newSumDBTestHandler(t, db.server.URL, true, 0) + h.sumDBOffline = true + + if w := serveGoRequest(h, "/sumdb/sum.example"+lookup); w.Code != http.StatusOK { + t.Fatalf("online: status = %d, want 200", w.Code) + } + db.down.Store(true) + serveGoRequest(h, "/sumdb/sum.example"+lookup) + if !h.sumDBProbes.recentlyFailed(testSumDBName) { + t.Fatal("the failure was not remembered") + } + w := serveGoRequest(h, "/sumdb/sum.example"+lookup) + if w.Code != http.StatusOK || w.Header().Get("Warning") != staleWarning { + t.Errorf("offline copy = %d Warning %q, want 200 %q", w.Code, w.Header().Get("Warning"), staleWarning) + } +} diff --git a/internal/handler/go_vulndb.go b/internal/handler/go_vulndb.go index d3c1598f..4cd2970f 100644 --- a/internal/handler/go_vulndb.go +++ b/internal/handler/go_vulndb.go @@ -57,7 +57,7 @@ func (h *GoHandler) handleVulnDB(w http.ResponseWriter, r *http.Request, path st p.writeMetadataFetchError(w, res.err) return } - p.writeMetadataCachedResponseWithEncoding(w, r, goEcosystem, cacheKey, res.body, res.contentType, res.contentEncoding) + writeMetadataResult(w, r, res) } // validGoVulnDBPath reports whether path is a file the vulnerability diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 9612e510..55aa8c96 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -4,7 +4,9 @@ package handler import ( "bytes" "context" + "crypto/sha256" "database/sql" + "encoding/hex" "errors" "fmt" "io" @@ -1626,8 +1628,7 @@ var errStale304 = fmt.Errorf("upstream returned 304 but cached file is missing") // cacheKey is typically the package name but can include subpath components. // Optional acceptHeaders specify the Accept header(s) to send; defaults to application/json. func (p *Proxy) FetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL string, acceptHeaders ...string) ([]byte, string, error) { - body, contentType, _, err := p.fetchOrCacheMetadata(ctx, ecosystem, cacheKey, upstreamURL, "", nil, acceptHeaders...) - return body, contentType, err + return p.fetchOrCacheMetadata(ctx, ecosystem, cacheKey, upstreamURL, "", nil, acceptHeaders...) } // fetchOrCacheMetadata implements FetchOrCacheMetadata. acceptEncoding controls @@ -1643,13 +1644,13 @@ func (p *Proxy) FetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, u // it can also decode the document into request-local state; a caller that // joined a fetch gets its own validation error directly. It must not modify the // body, which joined callers share. -func (p *Proxy) fetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, acceptEncoding string, validate func([]byte) error, acceptHeaders ...string) ([]byte, string, string, error) { +func (p *Proxy) fetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, acceptEncoding string, validate func([]byte) error, acceptHeaders ...string) ([]byte, string, error) { accept := contentTypeJSON if len(acceptHeaders) > 0 && acceptHeaders[0] != "" { accept = acceptHeaders[0] } res := p.lookupMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate) - return res.body, res.contentType, res.contentEncoding, res.err + return res.body, res.contentType, res.err } // fetchMetadataDocument is FetchOrCacheMetadata for callers that rewrite the @@ -1789,23 +1790,29 @@ func (p *Proxy) refreshMetadata(ctx context.Context, ecosystem, cacheKey, upstre // and falls back to the cached copy if upstream fails. entry is the row to // revalidate against, or nil. func (p *Proxy) fetchMetadataFromUpstream(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding string, validate func([]byte) error, entry *database.MetadataCacheEntry) metadataResult { - meta, err := p.fetchUpstreamMetadata(ctx, upstreamURL, entry, accept, acceptEncoding) - if errors.Is(err, errStale304) { - // 304 but cached file is gone; retry without ETag - meta, err = p.fetchUpstreamMetadata(ctx, upstreamURL, nil, accept, acceptEncoding) + fetch := func(entry *database.MetadataCacheEntry) (*upstreamMetadata, error) { + meta, err := p.fetchUpstreamMetadata(ctx, upstreamURL, entry, accept, acceptEncoding) + if err == nil && validate != nil { + err = validate(meta.body) + } + return meta, err } - if err == nil && validate != nil { - err = validate(meta.body) + meta, err := fetch(entry) + var digest string + // Validated first, so a 304 for a rejected copy doesn't mark it fresh. + if err == nil && p.CacheMetadata { + digest, err = p.refreshMetadataBlob(ecosystem, cacheKey, meta) + } + if errors.Is(err, errStale304) { + // 304 but the cached copy is gone or was replaced; retry without + // validators. + meta, err = fetch(nil) } if err == nil { - var digest string - if p.CacheMetadata { - digest = p.refreshMetadataBlob(ecosystem, cacheKey, meta) - if digest == "" { - digest = p.cacheMetadataBlob(ctx, ecosystem, cacheKey, metadataStoragePath(ecosystem, cacheKey), meta) - } + if p.CacheMetadata && digest == "" { + digest = p.cacheMetadataBlob(ctx, ecosystem, cacheKey, metadataStoragePath(ecosystem, cacheKey), meta) } - return metadataResult{body: meta.body, contentType: meta.contentType, contentEncoding: meta.contentEncoding, digest: digest} + return fetchedResult(meta, digest) } return p.staleMetadata(ctx, ecosystem, cacheKey, validate, entry, err) @@ -1827,17 +1834,25 @@ func (p *Proxy) staleMetadata(ctx context.Context, ecosystem, cacheKey string, v // Re-read the row so the encoding describes the blob as it is now: a // concurrent refetch may have replaced both since entry was read above - // (an identity blob swapped for a gzip one during rollout). - entry = p.currentMetadataEntry(ecosystem, cacheKey, entry) - - data, ct, readErr := p.readCachedMetadata(ctx, entry, validate) - if readErr != nil { - return metadataResult{err: fmt.Errorf("upstream failed and cached metadata unusable (%v): %w", readErr, err)} + // (an identity blob swapped for a gzip one during rollout). Bytes that + // don't match the row were replaced after it was read, so the row is + // read once more; if they still don't match, no row describes them. + for attempt := 0; ; attempt++ { + entry = p.currentMetadataEntry(ecosystem, cacheKey, entry) + data, ct, readErr := p.readCachedMetadata(ctx, entry, validate) + if readErr == nil && !matchesEntryDigest(entry, data) { + if attempt == 0 { + continue + } + readErr = errors.New("replaced while being read") + } + if readErr != nil { + return metadataResult{err: fmt.Errorf("upstream failed and cached metadata unusable (%v): %w", readErr, err)} + } + p.Logger.Info("serving metadata from cache", + "ecosystem", ecosystem, "key", cacheKey) + return metadataResult{body: data, contentType: ct, contentEncoding: entry.ContentEncoding.String, stale: true, upstreamErr: err}.describedBy(entry) } - - p.Logger.Info("serving metadata from cache", - "ecosystem", ecosystem, "key", cacheKey) - return metadataResult{body: data, contentType: ct, contentEncoding: entry.ContentEncoding.String, stale: true, upstreamErr: err}.describedBy(entry) } // metadataResult is what a metadata lookup hands back. A shared fetch gives @@ -1850,6 +1865,10 @@ type metadataResult struct { // empty otherwise. digest string err error + // etag and lastModified are the validators of body: upstream's for a + // fetched document, the cache row's for a stored one. + etag string + lastModified time.Time // stale is set when the body is the cached copy served because upstream // failed, and upstreamErr is that failure. stale bool @@ -1863,15 +1882,37 @@ type metadataResult struct { entry *database.MetadataCacheEntry } -// describedBy returns res with the digest that entry, the cache row its body -// was read for, records. +// describedBy returns res with the digest and validators that entry, the +// cache row its body was read for, records. func (res metadataResult) describedBy(entry *database.MetadataCacheEntry) metadataResult { if entry.ContentDigest.Valid { res.digest = entry.ContentDigest.String } + res.etag = entry.ETag.String + if entry.LastModified.Valid { + res.lastModified = entry.LastModified.Time + } return res } +// fetchedResult returns the result for meta, a document from upstream (or +// the stored copy it answered 304 for), cached under digest. +func fetchedResult(meta *upstreamMetadata, digest string) metadataResult { + return metadataResult{ + body: meta.body, + contentType: meta.contentType, + contentEncoding: meta.contentEncoding, + digest: digest, + etag: meta.etag, + lastModified: meta.lastModified, + } +} + +// cachedMeta returns the validators and staleness of res's body. +func (res metadataResult) cachedMeta() cachedMeta { + return cachedMeta{etag: res.etag, lastModified: res.lastModified, stale: res.stale} +} + // inflightMetadata is one metadata fetch that concurrent callers share. res is // written before done closes and read only after, so the close is the handoff. type inflightMetadata struct { @@ -2015,7 +2056,8 @@ func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, e } defer func() { _ = cached.Close() }() data, readErr := p.ReadMetadataSized(cached, metadataEntrySize(entry)) - if readErr != nil { + if readErr != nil || !matchesEntryDigest(entry, data) { + // Gone, or replaced by another instance since entry was read. return nil, errStale304 } meta := &upstreamMetadata{body: data, contentType: contentTypeJSON, etag: entry.ETag.String, notModified: entry} @@ -2082,6 +2124,17 @@ func (p *Proxy) requestUpstreamMetadata(ctx context.Context, upstreamURL string, return resp, nil } +// matchesEntryDigest reports whether data is the document entry records, +// when entry has a sha256 digest to check it by. +func matchesEntryDigest(entry *database.MetadataCacheEntry, data []byte) bool { + want, ok := strings.CutPrefix(entry.ContentDigest.String, "sha256:") + if !entry.ContentDigest.Valid || !ok { + return true + } + sum := sha256.Sum256(data) + return hex.EncodeToString(sum[:]) == want +} + // upstreamMetadataStatus returns the error for a metadata response other // than 200. Callers handle a 304 they asked for before. func upstreamMetadataStatus(resp *http.Response) error { @@ -2099,23 +2152,28 @@ func upstreamMetadataStatus(resp *http.Response) error { // document meta was read from, by moving the row's fetched_at on. The bytes in // storage are unchanged, so they are neither written nor hashed again. It // returns the row's digest, or "" when the caller has to store meta as for a -// new document: when upstream sent one, the row records no digest, or a -// concurrent fetch has replaced the document since it was read. -func (p *Proxy) refreshMetadataBlob(ecosystem, cacheKey string, meta *upstreamMetadata) string { +// new document: when upstream sent one or the row records no digest. When +// a concurrent fetch has replaced the document since it was read, it returns +// errStale304: storing meta would put the old document back under the new +// one. A failed database write only leaves fetched_at behind, so the +// confirmed bytes are still served. +func (p *Proxy) refreshMetadataBlob(ecosystem, cacheKey string, meta *upstreamMetadata) (string, error) { entry := meta.notModified if entry == nil || p.DB == nil || !entry.ContentDigest.Valid { - return "" + return "", nil } refreshed, err := p.DB.RefreshMetadataCache(ecosystem, cacheKey, entry.ContentDigest.String, time.Now()) if err != nil { + // The bytes were checked against the row and upstream confirmed + // them; only fetched_at stays behind, so they are still served. p.Logger.Warn("failed to refresh cached metadata", "ecosystem", ecosystem, "key", cacheKey, "error", err) - return "" + return entry.ContentDigest.String, nil } if !refreshed { - return "" + return "", errStale304 } p.recordMetadataRead(entry) - return entry.ContentDigest.String + return entry.ContentDigest.String, nil } // cacheMetadataBlob stores metadata bytes in storage and updates the @@ -2190,18 +2248,6 @@ type cachedMeta struct { stale bool } -// lookupCachedMeta retrieves cache validators for a metadata entry. -func (p *Proxy) lookupCachedMeta(ecosystem, cacheKey string) cachedMeta { - if p.DB == nil { - return cachedMeta{} - } - entry, err := p.DB.GetMetadataCache(ecosystem, cacheKey) - if err != nil || entry == nil { - return cachedMeta{} - } - return p.cachedMetaFor(entry) -} - // cachedMetaFor returns the cache validators and freshness state of entry. func (p *Proxy) cachedMetaFor(entry *database.MetadataCacheEntry) cachedMeta { var cm cachedMeta @@ -2259,16 +2305,20 @@ func (p *Proxy) proxyCachedWithEncoding(w http.ResponseWriter, r *http.Request, return } - body, contentType, contentEncoding, err := p.fetchOrCacheMetadata(r.Context(), ecosystem, cacheKey, upstreamURL, acceptEncoding, nil, acceptHeaders...) - if err != nil { - p.writeMetadataFetchError(w, err) + accept := contentTypeJSON + if len(acceptHeaders) > 0 && acceptHeaders[0] != "" { + accept = acceptHeaders[0] + } + res := p.lookupMetadata(r.Context(), ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, nil) + if res.err != nil { + p.writeMetadataFetchError(w, res.err) return } if hooks != nil && hooks.fetched != nil { - hooks.fetched(body, contentType, contentEncoding) + hooks.fetched(res.body, res.contentType, res.contentEncoding) } - p.writeMetadataCachedResponseWithEncoding(w, r, ecosystem, cacheKey, body, contentType, contentEncoding) + writeMetadataResult(w, r, res) } // writeMetadataFetchError answers a failed metadata fetch. @@ -2284,19 +2334,11 @@ func (p *Proxy) writeMetadataFetchError(w http.ResponseWriter, err error) { http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) } -// writeMetadataCachedResponse writes a cached metadata response and handles -// conditional request headers using metadata cache validators. -func (p *Proxy) writeMetadataCachedResponse(w http.ResponseWriter, r *http.Request, ecosystem, cacheKey string, body []byte, contentType string) { - p.writeMetadataCachedResponseWithEncoding(w, r, ecosystem, cacheKey, body, contentType, "") -} - -// writeMetadataCachedResponseWithEncoding is writeMetadataCachedResponse with -// an explicit Content-Encoding. contentEncoding must describe the body being -// written; it is passed in rather than re-read from the cache row, which is -// missing or stale when the metadata cache write failed and would otherwise -// mislabel the bytes. -func (p *Proxy) writeMetadataCachedResponseWithEncoding(w http.ResponseWriter, r *http.Request, ecosystem, cacheKey string, body []byte, contentType, contentEncoding string) { - writeMetadataBody(w, r, p.lookupCachedMeta(ecosystem, cacheKey), body, contentType, contentEncoding) +// writeMetadataResult writes the document res holds with its own validators +// and staleness, rather than the cache row's, which may describe other bytes +// when the cache write failed or another instance replaced it. +func writeMetadataResult(w http.ResponseWriter, r *http.Request, res metadataResult) { + writeMetadataBody(w, r, res.cachedMeta(), res.body, res.contentType, res.contentEncoding) } // writeMetadataBody writes a metadata document with the validators and @@ -2574,6 +2616,39 @@ func (p *Proxy) getCachedArtifactWithUpstreamHash(ctx context.Context, pkgPURL, return nil, nil } +// refetchArtifactFromURL fetches a file upstream replaces in place, which +// a cached copy can't stand in for while upstream answers. The new copy +// replaces the cached one only once it is stored, and the cached one is +// served when the fetch fails for any reason but the file being missing or +// blocked. Concurrent callers share one fetch. +func (p *Proxy) refetchArtifactFromURL(ctx context.Context, ecosystem, name, version, filename, downloadURL string) (*CacheResult, error) { + pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, name, version) + if err != nil { + return nil, err + } + // Streaming never serves a cached copy, and a denied version is refused + // there. + if p.StreamArtifacts || p.versionDenied(ecosystem, name, version) || p.Denylist.Denied(versionPURL) { + return p.getOrFetchArtifactFromURL(ctx, ecosystem, name, version, filename, downloadURL, nil, "", "") + } + metrics.RecordArtifactCacheMiss(ecosystem) + key := artifactCoalesceKey(versionPURL, filename, downloadURL, "", "") + noRecheck := func() (artifacts.Artifact, string, bool) { return artifacts.Artifact{}, "", false } + result, err := p.coalesceFetch(ctx, key, noRecheck, func(fetchCtx context.Context) (artifacts.Artifact, string, error) { + return p.fetchAndCacheFromURL(fetchCtx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, nil, "", "") + }) + if err == nil || errors.Is(err, ErrUpstreamNotFound) || errors.Is(err, ErrArtifactBlocked) || ctx.Err() != nil { + return result, err + } + cached, cacheErr := p.checkCache(ctx, pkgPURL, versionPURL, filename) + if cacheErr != nil || cached == nil { + return nil, err + } + p.Logger.Warn("upstream fetch failed, serving cached copy", + "purl", versionPURL, "filename", filename, "error", err) + return cached, nil +} + // discardStaleArtifact clears the cached entry when its digest disagrees with // upstreamHash, queueing its object for deletion. It runs under the coalescing // key, after the recheck, so an entry a previous fetch refreshed is kept. diff --git a/internal/handler/maven.go b/internal/handler/maven.go index e841d083..50d30467 100644 --- a/internal/handler/maven.go +++ b/internal/handler/maven.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "errors" "fmt" + "io" "net/http" "path" "strings" @@ -175,32 +176,51 @@ func (h *MavenHandler) handleMetadata(w http.ResponseWriter, r *http.Request, so return } - body, contentType, cacheKey, err := h.fetchMetadata(r.Context(), source, urlPath) - if err != nil { - h.writeMetadataError(w, err) + // The checksums of maven-metadata.xml are computed from the cached + // metadata, as with cooldown, so they never describe another deploy. + if suffix, ok := strings.CutPrefix(path.Base(urlPath), mavenMetadataFile); ok && mavenChecksums[suffix] != nil { + res, _ := h.fetchMetadata(r.Context(), source, path.Join(path.Dir(urlPath), mavenMetadataFile)) + var body []byte + if res.err == nil { + body, res.err = h.proxy.decodedMetadataBody(res) + } + if res.err != nil { + h.writeMetadataError(w, res.err) + return + } + sum := mavenChecksums[suffix]() + _, _ = sum.Write(body) + w.Header().Set(headerContentType, "text/plain") + markStale(w, res.stale) + _, _ = io.WriteString(w, hex.EncodeToString(sum.Sum(nil))) return } - h.proxy.writeMetadataCachedResponse(w, r, "maven", cacheKey, body, contentType) + res, _ := h.fetchMetadata(r.Context(), source, urlPath) + if res.err != nil { + h.writeMetadataError(w, res.err) + return + } + writeMetadataResult(w, r, res) } // fetchMetadata fetches a metadata file from the source's upstream, or from // its fallback when the upstream doesn't have it, and returns it with the // key it is cached under. -func (h *MavenHandler) fetchMetadata(ctx context.Context, source mavenSource, urlPath string) (body []byte, contentType, cacheKey string, err error) { - cacheKey = source.metadataCacheKey(urlPath) +func (h *MavenHandler) fetchMetadata(ctx context.Context, source mavenSource, urlPath string) (metadataResult, string) { + cacheKey := source.metadataCacheKey(urlPath) upstreamURL := fmt.Sprintf("%s/%s", source.upstreamURL, urlPath) - body, contentType, err = h.proxy.FetchOrCacheMetadata(ctx, "maven", cacheKey, upstreamURL, "*/*") - if err != nil && source.fallbackURL != "" { - if errors.Is(err, ErrUpstreamNotFound) { + res := h.proxy.lookupMetadata(ctx, "maven", cacheKey, upstreamURL, "*/*", "", nil) + if res.err != nil && source.fallbackURL != "" { + if errors.Is(res.err, ErrUpstreamNotFound) { pluginPortalURL := fmt.Sprintf("%s/%s", source.fallbackURL, urlPath) h.proxy.Logger.Info("maven metadata unavailable in primary upstream, trying Gradle Plugin Portal", "path", urlPath) - body, contentType, err = h.proxy.FetchOrCacheMetadata(ctx, "maven", cacheKey, pluginPortalURL, "*/*") + res = h.proxy.lookupMetadata(ctx, "maven", cacheKey, pluginPortalURL, "*/*", "", nil) } } - return body, contentType, cacheKey, err + return res, cacheKey } func (h *MavenHandler) writeMetadataError(w http.ResponseWriter, err error) { @@ -249,13 +269,21 @@ func (h *MavenHandler) handleDownload(w http.ResponseWriter, r *http.Request, so upstreamURL := fmt.Sprintf("%s/%s", source.upstreamURL, urlPath) cacheFilename := source.cacheFilename(filename) - result, err := h.proxy.GetOrFetchArtifactFromURL(ctx, "maven", name, version, cacheFilename, upstreamURL) + // Each build replaces a SNAPSHOT file named without a timestamp, so a + // GET fetches it again rather than serving the cached copy. HEAD is + // answered from cache. + get := h.proxy.GetOrFetchArtifactFromURL + if h.isMutableSnapshotFile(urlPath) && r.Method != http.MethodHead { + get = h.proxy.refetchArtifactFromURL + } + + result, err := get(ctx, "maven", name, version, cacheFilename, upstreamURL) if err != nil && source.fallbackURL != "" { if errors.Is(err, ErrUpstreamNotFound) { pluginPortalURL := fmt.Sprintf("%s/%s", source.fallbackURL, urlPath) h.proxy.Logger.Info("maven artifact not found in primary upstream, trying Gradle Plugin Portal", "group", group, "artifact", artifact, "version", version, "filename", filename) - result, err = h.proxy.GetOrFetchArtifactFromURL(ctx, "maven", name, version, cacheFilename, pluginPortalURL) + result, err = get(ctx, "maven", name, version, cacheFilename, pluginPortalURL) } } if err != nil { @@ -336,6 +364,15 @@ func (h *MavenHandler) isCoordinateFile(urlPath string) bool { return strings.HasPrefix(filename, artifact+"-"+strings.TrimSuffix(version, "SNAPSHOT")) } +// isMutableSnapshotFile reports whether urlPath is a SNAPSHOT file named +// with "SNAPSHOT" rather than a build timestamp (lib-1.0-SNAPSHOT.jar), which +// each new build replaces. Timestamped SNAPSHOT files never change. +func (h *MavenHandler) isMutableSnapshotFile(urlPath string) bool { + _, artifact, version, filename := h.parsePath(urlPath) + return artifact != "" && strings.HasSuffix(version, mavenSnapshot) && + strings.HasPrefix(filename, artifact+"-"+version) +} + // isVersionListingFile matches maven-metadata.xml and its checksums. The // listing changes with every release, so it is cached for the metadata TTL // rather than kept like an artifact. diff --git a/internal/handler/maven_consistency_test.go b/internal/handler/maven_consistency_test.go new file mode 100644 index 00000000..e5484474 --- /dev/null +++ b/internal/handler/maven_consistency_test.go @@ -0,0 +1,157 @@ +package handler + +import ( + "crypto/sha1" //nolint:gosec // Maven publishes SHA-1 checksums. + "encoding/hex" + "io" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/registries/fetch" +) + +// mavenGenerations serves every path with the current generation's bytes, +// and a matching .sha1 for maven-metadata.xml, counting requests per path. +type mavenGenerations struct { + gen atomic.Int32 + requests map[string]*atomic.Int32 + // down makes every request fail with 502. + down atomic.Bool +} + +func (m *mavenGenerations) body(path string) string { + return path + " generation " + string(rune('0'+m.gen.Load())) +} + +func newMavenGenerations(t *testing.T, paths ...string) (*mavenGenerations, http.Handler) { + t.Helper() + m := &mavenGenerations{requests: map[string]*atomic.Int32{}} + for _, p := range paths { + m.requests[p] = &atomic.Int32{} + } + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if n := m.requests[r.URL.Path]; n != nil { + n.Add(1) + } + if m.down.Load() { + http.Error(w, "unavailable", http.StatusBadGateway) + return + } + const metadata = "/com/example/lib/maven-metadata.xml" + if r.URL.Path == metadata+".sha1" { + sum := sha1.Sum([]byte(m.body(metadata))) //nolint:gosec // as above + _, _ = io.WriteString(w, hex.EncodeToString(sum[:])) + return + } + _, _ = io.WriteString(w, m.body(r.URL.Path)) + })) + t.Cleanup(srv.Close) + + proxy, _, store, _ := setupTestProxy(t) + store.seekable = true + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + proxy.HTTPClient = srv.Client() + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(srv.Client()), fetch.WithMaxRetries(0)) + proxy.Fetcher = fetcher + t.Cleanup(func() { _ = fetcher.Close() }) + return m, NewMavenHandler(proxy, "http://localhost", srv.URL, srv.URL).Routes() +} + +// TestMavenMetadataChecksumMatchesServedMetadata checks that the checksum +// files of maven-metadata.xml describe the metadata the proxy serves, even +// when upstream has deployed since it was cached. +func TestMavenMetadataChecksumMatchesServedMetadata(t *testing.T) { + m, h := newMavenGenerations(t) + const metadata = "/com/example/lib/maven-metadata.xml" + + served := getMaven(h, metadata).Body.String() + m.gen.Store(1) + w := getMaven(h, metadata+".sha1") + sum := sha1.Sum([]byte(served)) //nolint:gosec // as above + if w.Code != http.StatusOK || w.Body.String() != hex.EncodeToString(sum[:]) { + t.Errorf(".sha1 = %d %q, want the sha1 of the served metadata %x", w.Code, w.Body, sum) + } +} + +// TestMavenNonUniqueSnapshotNotKept checks that a SNAPSHOT file without a +// timestamp, which each build replaces, is not served from cache for good, +// while a timestamped one is. +func TestMavenNonUniqueSnapshotNotKept(t *testing.T) { + const ( + snapshot = "/com/example/lib/1.0-SNAPSHOT/lib-1.0-SNAPSHOT.jar" + timestamped = "/com/example/lib/1.0-SNAPSHOT/lib-1.0-20240101.120000-1.jar" + ) + m, h := newMavenGenerations(t, timestamped) + + getMaven(h, snapshot) + getMaven(h, snapshot+".sha1") + getMaven(h, timestamped) + m.gen.Store(1) + if w := getMaven(h, snapshot); w.Body.String() != m.body(snapshot) { + t.Errorf("SNAPSHOT jar = %q, want the new build %q", w.Body, m.body(snapshot)) + } + if w := getMaven(h, snapshot+".sha1"); w.Body.String() != m.body(snapshot+".sha1") { + t.Errorf("SNAPSHOT jar checksum = %q, want the new build's", w.Body) + } + if w := getMaven(h, timestamped); w.Code != http.StatusOK || m.requests[timestamped].Load() != 1 { + t.Errorf("timestamped SNAPSHOT jar fetched %d times, want once", m.requests[timestamped].Load()) + } +} + +// TestMavenMutableSnapshotKeepsPolicyAndFallback checks that a SNAPSHOT file +// fetched again for each request still goes through the denylist, and falls +// back to the Gradle Plugin Portal when the primary upstream lacks it. +func TestMavenMutableSnapshotKeepsPolicyAndFallback(t *testing.T) { + const snapshot = "/com/example/lib/1.0-SNAPSHOT/lib-1.0-SNAPSHOT.jar" + primary := httptest.NewServer(http.NotFoundHandler()) + t.Cleanup(primary.Close) + portal := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, "portal build") + })) + t.Cleanup(portal.Close) + + proxy, _, store, _ := setupTestProxy(t) + store.seekable = true + proxy.HTTPClient = primary.Client() + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(primary.Client()), fetch.WithMaxRetries(0)) + proxy.Fetcher = fetcher + t.Cleanup(func() { _ = fetcher.Close() }) + h := NewMavenHandler(proxy, "http://localhost", primary.URL, portal.URL).Routes() + + if w := getMaven(h, snapshot); w.Code != http.StatusOK || w.Body.String() != "portal build" { + t.Errorf("SNAPSHOT from the fallback = %d %q, want 200 portal build", w.Code, w.Body) + } + setTestDenylist(t, proxy, "pkg:maven/com.example/lib@1.0-SNAPSHOT") + if w := getMaven(h, snapshot); w.Code != http.StatusForbidden { + t.Errorf("denied SNAPSHOT = %d, want 403", w.Code) + } +} + +// TestMavenMutableSnapshotServesCacheWhenUpstreamFails checks that a +// SNAPSHOT file fetched again for each request is still served from cache +// while upstream is down, and that HEAD does not download it again. +func TestMavenMutableSnapshotServesCacheWhenUpstreamFails(t *testing.T) { + const snapshot = "/com/example/lib/1.0-SNAPSHOT/lib-1.0-SNAPSHOT.jar" + m, h := newMavenGenerations(t, snapshot) + + built := getMaven(h, snapshot).Body.String() + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodHead, snapshot, nil)) + if w.Code != http.StatusOK || m.requests[snapshot].Load() != 1 { + t.Errorf("HEAD = %d after %d upstream requests, want 200 from cache", w.Code, m.requests[snapshot].Load()) + } + + m.down.Store(true) + if w := getMaven(h, snapshot); w.Code != http.StatusOK || w.Body.String() != built { + t.Errorf("SNAPSHOT while upstream is down = %d %q, want the cached build %q", w.Code, w.Body, built) + } + m.down.Store(false) + m.gen.Store(1) + if w := getMaven(h, snapshot); w.Body.String() != m.body(snapshot) { + t.Errorf("SNAPSHOT once upstream is back = %q, want the new build", w.Body) + } +} diff --git a/internal/handler/maven_cooldown.go b/internal/handler/maven_cooldown.go index 4da839ee..6155e0d7 100644 --- a/internal/handler/maven_cooldown.go +++ b/internal/handler/maven_cooldown.go @@ -242,11 +242,12 @@ func (h *MavenHandler) serveCooldownMetadata(w http.ResponseWriter, r *http.Requ return false } metadataPath := path.Join(path.Dir(urlPath), mavenMetadataFile) - body, contentType, cacheKey, err := h.fetchMetadata(r.Context(), source, metadataPath) - if err != nil { - h.writeMetadataError(w, err) + res, cacheKey := h.fetchMetadata(r.Context(), source, metadataPath) + if res.err != nil { + h.writeMetadataError(w, res.err) return true } + body, contentType := res.body, res.contentType m, err := h.proxy.cachedRewrite(r.Context(), "maven", h.proxyURL, cacheKey, body, "", func(in []byte) ([]byte, []string, error) { // The rewrite is shared by requests waiting on it and cached, so it @@ -273,7 +274,7 @@ func (h *MavenHandler) serveCooldownMetadata(w http.ResponseWriter, r *http.Requ if contentType == "" || contentType == contentTypeJSON { contentType = "text/xml" } - markStale(w, h.proxy.lookupCachedMeta("maven", cacheKey).stale) + markStale(w, res.stale) serveRewrittenMetadataAs(w, r, m, contentType) case mavenChecksums[suffix] != nil: sum := mavenChecksums[suffix]() diff --git a/internal/handler/metadata_consistency_test.go b/internal/handler/metadata_consistency_test.go new file mode 100644 index 00000000..3e0220af --- /dev/null +++ b/internal/handler/metadata_consistency_test.go @@ -0,0 +1,324 @@ +package handler + +import ( + "bytes" + "compress/gzip" + "context" + "database/sql" + "errors" + "io" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/database" +) + +// proxyCachedRequest runs one ProxyCached request for key "pkg". +func proxyCachedRequest(proxy *Proxy, upstreamURL string, header http.Header) *httptest.ResponseRecorder { + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/pkg", nil) + for k, v := range header { + r.Header[k] = v + } + proxy.ProxyCached(w, r, upstreamURL+"/pkg", "test", "pkg") + return w +} + +// TestFetchedMetadataValidatorsDescribeBody checks that a document fetched +// fresh but not stored is served with upstream's validators, not those of the +// older row still in the cache. +func TestFetchedMetadataValidatorsDescribeBody(t *testing.T) { + var requests atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set(headerContentType, contentTypeJSON) + if requests.Add(1) == 1 { + w.Header().Set(headerETag, `"v1"`) + _, _ = w.Write([]byte(`{"v":1}`)) + return + } + w.Header().Set(headerETag, `"v2"`) + _, _ = w.Write([]byte(`{"v":2}`)) + })) + t.Cleanup(upstream.Close) + + proxy, _, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Millisecond + proxy.HTTPClient = upstream.Client() + + if w := proxyCachedRequest(proxy, upstream.URL, nil); w.Code != http.StatusOK { + t.Fatalf("first request: %d", w.Code) + } + time.Sleep(5 * time.Millisecond) + store.mu.Lock() + store.storeErr = errors.New("disk full") + store.mu.Unlock() + + w := proxyCachedRequest(proxy, upstream.URL, nil) + if w.Code != http.StatusOK || w.Body.String() != `{"v":2}` { + t.Fatalf("second request = %d %q, want v2", w.Code, w.Body) + } + if got := w.Header().Get(headerETag); got != `"v2"` { + t.Errorf("ETag = %q, want %q", got, `"v2"`) + } + if got := w.Header().Get("Warning"); got != "" { + t.Errorf("Warning = %q on a fresh document", got) + } + + // A client holding v1 must get v2, not a 304. + w = proxyCachedRequest(proxy, upstream.URL, http.Header{"If-None-Match": {`"v1"`}}) + if w.Code != http.StatusOK || w.Body.String() != `{"v":2}` { + t.Errorf("conditional request = %d %q, want 200 v2", w.Code, w.Body) + } +} + +// TestStaleMetadataMarkedWithoutTTL checks that the stale fallback carries +// the Warning header with metadata_ttl 0, where every request revalidates. +func TestStaleMetadataMarkedWithoutTTL(t *testing.T) { + upstream := upstreamThen(t, `{"v":1}`, http.StatusBadGateway) + proxy, _, _, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = 0 + proxy.HTTPClient = upstream.Client() + + if w := proxyCachedRequest(proxy, upstream.URL, nil); w.Code != http.StatusOK { + t.Fatalf("first request: %d", w.Code) + } + w := proxyCachedRequest(proxy, upstream.URL, nil) + if w.Code != http.StatusOK || w.Body.String() != `{"v":1}` { + t.Fatalf("fallback = %d %q", w.Code, w.Body) + } + if got := w.Header().Get("Warning"); got != staleWarning { + t.Errorf("Warning = %q, want %q", got, staleWarning) + } +} + +// seedMetadataRow stores {"v":1} for key "pkg" with ETag "v1" and a row past +// its TTL. +func seedMetadataRow(t *testing.T, db *database.DB, store *mockStorage) string { + t.Helper() + const body, etag = `{"v":1}`, `"v1"` + path := metadataStoragePath("test", "pkg") + store.mu.Lock() + store.files[path] = []byte(body) + store.mu.Unlock() + if err := db.UpsertMetadataCache(&database.MetadataCacheEntry{ + Ecosystem: "test", + Name: "pkg", + StoragePath: path, + ETag: sql.NullString{String: etag, Valid: true}, + ContentType: sql.NullString{String: contentTypeJSON, Valid: true}, + ContentDigest: sql.NullString{String: "sha256:" + sha256Hex(body), Valid: true}, + Size: sql.NullInt64{Int64: int64(len(body)), Valid: true}, + FetchedAt: sql.NullTime{Time: time.Now().Add(-time.Hour), Valid: true}, + }); err != nil { + t.Fatal(err) + } + return path +} + +// v2Upstream answers If-None-Match "v1" with 304 after running onNotModified, +// and any other request with v2. +func v2Upstream(t *testing.T, onNotModified func()) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("If-None-Match") == `"v1"` { + onNotModified() + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set(headerContentType, contentTypeJSON) + w.Header().Set(headerETag, `"v2"`) + _, _ = w.Write([]byte(`{"v":2}`)) + })) + t.Cleanup(srv.Close) + return srv +} + +// TestNotModifiedChecksStoredBytes checks that a 304 for a row whose blob has +// meanwhile been replaced (by another instance sharing the store) does not +// serve the new bytes under the old ETag. +func TestNotModifiedChecksStoredBytes(t *testing.T) { + proxy, db, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Minute + path := seedMetadataRow(t, db, store) + upstream := v2Upstream(t, func() { + // The other instance has written v2 but not yet its row. + store.mu.Lock() + store.files[path] = []byte(`{"v":2}`) + store.mu.Unlock() + }) + proxy.HTTPClient = upstream.Client() + + w := proxyCachedRequest(proxy, upstream.URL, nil) + if w.Code != http.StatusOK || w.Body.String() != `{"v":2}` { + t.Fatalf("response = %d %q", w.Code, w.Body) + } + if got := w.Header().Get(headerETag); got != `"v2"` { + t.Errorf("ETag = %q, want %q", got, `"v2"`) + } + row, err := db.GetMetadataCache("test", "pkg") + if err != nil || row == nil { + t.Fatal(row, err) + } + if row.ETag.String != `"v2"` || row.ContentDigest.String != "sha256:"+sha256Hex(`{"v":2}`) { + t.Errorf("row = etag %q digest %q, want v2's", row.ETag.String, row.ContentDigest.String) + } +} + +// TestNotModifiedAfterRowReplaced checks that a 304 for a row another +// instance has replaced since it was read does not store the old document +// back over the new one. +func TestNotModifiedAfterRowReplaced(t *testing.T) { + proxy, db, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Minute + path := seedMetadataRow(t, db, store) + upstream := v2Upstream(t, func() { + // The other instance has recorded v2's row; its blob is not written + // yet. + _ = db.UpsertMetadataCache(&database.MetadataCacheEntry{ + Ecosystem: "test", + Name: "pkg", + StoragePath: path, + ETag: sql.NullString{String: `"v2"`, Valid: true}, + ContentType: sql.NullString{String: contentTypeJSON, Valid: true}, + ContentDigest: sql.NullString{String: "sha256:" + sha256Hex(`{"v":2}`), Valid: true}, + Size: sql.NullInt64{Int64: 7, Valid: true}, + FetchedAt: sql.NullTime{Time: time.Now(), Valid: true}, + }) + }) + proxy.HTTPClient = upstream.Client() + + res := proxy.fetchMetadataDocument(context.Background(), "test", "pkg", upstream.URL+"/pkg", contentTypeJSON, nil) + if res.err != nil || string(res.body) != `{"v":2}` { + t.Fatalf("fetch = %q, %v; want v2", res.body, res.err) + } + row, err := db.GetMetadataCache("test", "pkg") + if err != nil || row == nil { + t.Fatal(row, err) + } + if row.ETag.String != `"v2"` { + t.Errorf("row etag = %q, want %q", row.ETag.String, `"v2"`) + } +} + +// TestNotModifiedRejectedBodyNotRefreshed checks that a 304 for a stored +// document validate rejects does not mark that document fresh. +func TestNotModifiedRejectedBodyNotRefreshed(t *testing.T) { + proxy, db, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Minute + seedMetadataRow(t, db, store) + before, err := db.GetMetadataCache("test", "pkg") + if err != nil || before == nil { + t.Fatal(before, err) + } + upstream := v2Upstream(t, func() {}) + proxy.HTTPClient = upstream.Client() + + reject := func(body []byte) error { + if string(body) == `{"v":1}` { + return errors.New("damaged") + } + return nil + } + proxy.fetchMetadataDocument(context.Background(), "test", "pkg", upstream.URL+"/pkg", contentTypeJSON, reject) + row, err := db.GetMetadataCache("test", "pkg") + if err != nil || row == nil { + t.Fatal(row, err) + } + if !row.FetchedAt.Time.Equal(before.FetchedAt.Time) { + t.Errorf("rejected v1 row fetched_at moved from %v to %v", before.FetchedAt.Time, row.FetchedAt.Time) + } +} + +// openHookStorage runs onOpen before the first Open. +type openHookStorage struct { + *mockStorage + once sync.Once + onOpen func() +} + +func (s *openHookStorage) Open(ctx context.Context, path string) (io.ReadCloser, error) { + s.once.Do(s.onOpen) + return s.mockStorage.Open(ctx, path) +} + +// TestStaleMetadataChecksStoredBytes checks that the stale fallback does not +// serve a blob another instance replaced after the row was read under that +// row's ETag. +func TestStaleMetadataChecksStoredBytes(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadGateway) + })) + t.Cleanup(upstream.Close) + proxy, db, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = 0 + proxy.HTTPClient = upstream.Client() + path := seedMetadataRow(t, db, store) + proxy.Storage = &openHookStorage{mockStorage: store, onOpen: func() { + // The other instance has written v2 but not yet its row. + store.mu.Lock() + store.files[path] = []byte(`{"v":2}`) + store.mu.Unlock() + }} + + // No row describes v2 yet, so its encoding is unknown: nothing is served. + w := proxyCachedRequest(proxy, upstream.URL, http.Header{"If-None-Match": {`"v1"`}}) + if w.Code < http.StatusInternalServerError || w.Header().Get(headerETag) == `"v1"` { + t.Errorf("response = %d ETag %q for v2's bytes, want an error without v1 validators", w.Code, w.Header().Get(headerETag)) + } +} + +// TestStaleMetadataRereadsReplacedRow checks that when the stored bytes no +// longer match the row, the stale fallback reads the row again and serves +// the blob with the encoding and validators of the row that describes it. +func TestStaleMetadataRereadsReplacedRow(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadGateway) + })) + t.Cleanup(upstream.Close) + proxy, db, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = 0 + proxy.HTTPClient = upstream.Client() + path := seedMetadataRow(t, db, store) + var gz bytes.Buffer + zw := gzip.NewWriter(&gz) + _, _ = zw.Write([]byte(`{"v":2}`)) + _ = zw.Close() + v2 := gz.Bytes() + proxy.Storage = &openHookStorage{mockStorage: store, onOpen: func() { + // The other instance stores v2 gzipped, then its row. + store.mu.Lock() + store.files[path] = v2 + store.mu.Unlock() + if err := db.UpsertMetadataCache(&database.MetadataCacheEntry{ + Ecosystem: "test", + Name: "pkg", + StoragePath: path, + ETag: sql.NullString{String: `"v2"`, Valid: true}, + ContentType: sql.NullString{String: contentTypeJSON, Valid: true}, + ContentEncoding: sql.NullString{String: "gzip", Valid: true}, + ContentDigest: sql.NullString{String: "sha256:" + sha256Hex(string(v2)), Valid: true}, + Size: sql.NullInt64{Int64: int64(len(v2)), Valid: true}, + FetchedAt: sql.NullTime{Time: time.Now().Add(-time.Hour), Valid: true}, + }); err != nil { + t.Error(err) + } + }} + + w := proxyCachedRequest(proxy, upstream.URL, http.Header{"Accept-Encoding": {"gzip"}}) + if w.Code != http.StatusOK || w.Header().Get("Content-Encoding") != "gzip" || + w.Header().Get(headerETag) != `"v2"` || !bytes.Equal(w.Body.Bytes(), v2) { + t.Errorf("response = %d, Content-Encoding %q, ETag %q; want v2 gzipped under its own row", + w.Code, w.Header().Get("Content-Encoding"), w.Header().Get(headerETag)) + } +} diff --git a/internal/handler/metadata_immutable.go b/internal/handler/metadata_immutable.go index 691fbc1d..214077c4 100644 --- a/internal/handler/metadata_immutable.go +++ b/internal/handler/metadata_immutable.go @@ -40,9 +40,8 @@ func (p *Proxy) proxyImmutable(w http.ResponseWriter, r *http.Request, upstreamU p.writeMetadataFetchError(w, res.err) return } - cm := p.lookupCachedMeta(ecosystem, cacheKey) - cm.stale = false - writeMetadataBody(w, r, cm, res.body, res.contentType, res.contentEncoding) + res.stale = false + writeMetadataResult(w, r, res) } // fetchImmutableMetadata returns an immutable metadata document from the @@ -74,7 +73,7 @@ func (p *Proxy) fetchImmutableMetadata(ctx context.Context, ecosystem, cacheKey, return metadataResult{err: err} } digest := p.cacheMetadataBlob(fetchCtx, ecosystem, cacheKey, metadataStoragePath(ecosystem, cacheKey), meta) - return metadataResult{body: meta.body, contentType: meta.contentType, contentEncoding: meta.contentEncoding, digest: digest} + return fetchedResult(meta, digest) }) return res } diff --git a/internal/handler/metadata_stream.go b/internal/handler/metadata_stream.go index 56606e24..c165b819 100644 --- a/internal/handler/metadata_stream.go +++ b/internal/handler/metadata_stream.go @@ -35,6 +35,12 @@ func (p *Proxy) serveStoredMetadata(w http.ResponseWriter, r *http.Request, ecos if entry == nil { return false } + return p.serveMetadataEntry(w, r, ecosystem, cacheKey, entry, immutable) +} + +// serveMetadataEntry is serveStoredMetadata for entry, a row +// storedMetadataEntry or streamableMetadata accepted. +func (p *Proxy) serveMetadataEntry(w http.ResponseWriter, r *http.Request, ecosystem, cacheKey string, entry *database.MetadataCacheEntry, immutable bool) bool { body, err := p.openMetadataEntry(r.Context(), ecosystem, cacheKey, entry) if err != nil { return false @@ -88,14 +94,20 @@ func (p *Proxy) storedMetadataEntry(ecosystem, cacheKey string, immutable bool) if !immutable && time.Since(entry.FetchedAt.Time) >= p.MetadataTTL { return nil } + if !p.streamableMetadata(entry) { + return nil + } + return entry +} + +// streamableMetadata reports whether entry records a sha256 digest and a +// size within the metadata size limit, so its document can be streamed. +func (p *Proxy) streamableMetadata(entry *database.MetadataCacheEntry) bool { limit := p.MetadataMaxSize if limit <= 0 { limit = defaultMetadataMaxSize } - if !entry.ContentDigest.Valid || !strings.HasPrefix(entry.ContentDigest.String, "sha256:") || !entry.Size.Valid || entry.Size.Int64 > limit { - return nil - } - return entry + return entry.ContentDigest.Valid && strings.HasPrefix(entry.ContentDigest.String, "sha256:") && entry.Size.Valid && entry.Size.Int64 <= limit } // openMetadataEntry opens the stored document of a row freshMetadataEntry diff --git a/internal/handler/npm.go b/internal/handler/npm.go index 7eef4240..32a5eaf0 100644 --- a/internal/handler/npm.go +++ b/internal/handler/npm.go @@ -607,7 +607,7 @@ func (h *NPMHandler) handleFilteredNPMVersionMetadata(w http.ResponseWriter, r * return } - packument, _, _, err := h.proxy.fetchOrCacheMetadata(r.Context(), "npm", cacheKey, packumentURL, "", h.packumentValidator(), accept) + packument, _, err := h.proxy.fetchOrCacheMetadata(r.Context(), "npm", cacheKey, packumentURL, "", h.packumentValidator(), accept) if err != nil { if errors.Is(err, ErrUpstreamNotFound) { JSONError(w, http.StatusNotFound, ErrPackageNotFound.Error()) diff --git a/internal/handler/nuget_cooldown.go b/internal/handler/nuget_cooldown.go index b9345c5f..7a0a17ab 100644 --- a/internal/handler/nuget_cooldown.go +++ b/internal/handler/nuget_cooldown.go @@ -44,7 +44,7 @@ func (h *NuGetHandler) nugetMetadata(ctx context.Context, path string) (map[stri document, err = h.decodeNuGetMetadata(body) return err } - _, _, _, err := h.proxy.fetchOrCacheMetadata(ctx, "nuget", key, target, "", validate) + _, _, err := h.proxy.fetchOrCacheMetadata(ctx, "nuget", key, target, "", validate) if err != nil { return nil, err } diff --git a/internal/handler/release_index.go b/internal/handler/release_index.go new file mode 100644 index 00000000..626792a2 --- /dev/null +++ b/internal/handler/release_index.go @@ -0,0 +1,133 @@ +package handler + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/xml" + "net/http" + "strings" + "time" +) + +// APT and RPM repositories have a release file (InRelease, Release, +// repomd.xml) that lists the sha256 of each index file. Clients fetch the +// release file first and reject an index that doesn't match it. Each file +// has its own metadata cache entry and TTL, so without care a client could +// get a cached release file and a freshly fetched index, or the reverse. +// +// An index the cached release file lists is therefore served by that +// listing rather than by its own TTL: the cached copy when it matches, +// whatever its age, and otherwise a copy fetched from upstream. If upstream +// only has an index the cached release file doesn't list, upstream has moved +// on; that index is served and the release file is expired, so the next +// update gets a matching pair. + +// cachedRelease is a release file in the metadata cache. +type cachedRelease struct { + ecosystem string + cacheKey string + body []byte + fetchedAt time.Time +} + +// readCachedRelease returns the cached release file stored under cacheKey, +// or nil. +func (p *Proxy) readCachedRelease(ctx context.Context, ecosystem, cacheKey string) *cachedRelease { + if !p.CacheMetadata || p.DB == nil || p.Storage == nil { + return nil + } + entry, err := p.DB.GetMetadataCache(ecosystem, cacheKey) + if err != nil || entry == nil || entry.ContentEncoding.String != "" { + return nil + } + body, _, err := p.readCachedMetadata(ctx, entry, nil) + // Bytes another instance replaced since the row was read don't describe + // the release the row records. + if err != nil || !matchesEntryDigest(entry, body) { + return nil + } + return &cachedRelease{ecosystem: ecosystem, cacheKey: cacheKey, body: body, fetchedAt: entry.FetchedAt.Time} +} + +// proxyListedIndex serves an index file that release lists with the sha256 +// hex want. Without a listing it is served as other metadata. +func (p *Proxy) proxyListedIndex(w http.ResponseWriter, r *http.Request, upstreamURL, ecosystem, cacheKey string, release *cachedRelease, want string) { + if release == nil || want == "" { + p.ProxyCached(w, r, upstreamURL, ecosystem, cacheKey, "*/*") + return + } + if entry, _ := p.DB.GetMetadataCache(ecosystem, cacheKey); entry != nil && p.streamableMetadata(entry) && + entry.ContentDigest.String == "sha256:"+want && entry.ContentEncoding.String == "" { + if p.serveMetadataEntry(w, r, ecosystem, cacheKey, entry, true) { + return + } + } + p.recordMetadataCacheMiss(ecosystem) + res := p.refreshMetadata(r.Context(), ecosystem, cacheKey, upstreamURL, "*/*", "identity", nil) + if res.err != nil { + p.writeMetadataFetchError(w, res.err) + return + } + if sum := sha256.Sum256(res.body); !res.stale && hex.EncodeToString(sum[:]) != want { + p.Logger.Info("index does not match the cached release file, expiring it", + "ecosystem", ecosystem, "url", upstreamURL) + if _, err := p.DB.ExpireMetadataCache(ecosystem, []string{release.cacheKey}, time.Now()); err != nil { + p.Logger.Warn("failed to expire cached release file", "ecosystem", ecosystem, "error", err) + } + } + writeMetadataResult(w, r, res) +} + +// releaseLineMax bounds a line of a Release file. +const releaseLineMax = 1 << 20 + +// debianReleaseSHA256 returns the sha256 an APT Release or InRelease file +// lists for rel, a path relative to the release file's directory, or "". +func debianReleaseSHA256(release []byte, rel string) string { + inSection := false + scanner := bufio.NewScanner(bytes.NewReader(release)) + scanner.Buffer(nil, releaseLineMax) + for scanner.Scan() { + line := strings.TrimRight(scanner.Text(), "\r") + if !strings.HasPrefix(line, " ") { + inSection = line == "SHA256:" + continue + } + if !inSection { + continue + } + if f := strings.Fields(line); len(f) == 3 && f[2] == rel { + return strings.ToLower(f[0]) + } + } + return "" +} + +// repomdSHA256 returns the sha256 an RPM repomd.xml lists for href, a path +// relative to the repository root, or "" when it lists none or another +// checksum type. +func repomdSHA256(repomd []byte, href string) string { + var doc struct { + Data []struct { + Checksum struct { + Type string `xml:"type,attr"` + Value string `xml:",chardata"` + } `xml:"checksum"` + Location struct { + Href string `xml:"href,attr"` + } `xml:"location"` + } `xml:"data"` + } + if xml.Unmarshal(repomd, &doc) != nil { + return "" + } + for _, d := range doc.Data { + if d.Location.Href == href && d.Checksum.Type == "sha256" { + return strings.ToLower(strings.TrimSpace(d.Checksum.Value)) + } + } + return "" +} diff --git a/internal/handler/release_index_test.go b/internal/handler/release_index_test.go new file mode 100644 index 00000000..b69f4edb --- /dev/null +++ b/internal/handler/release_index_test.go @@ -0,0 +1,265 @@ +package handler + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +// releaseArchive is an upstream repository whose release file and index +// change together, one generation at a time. +type releaseArchive struct { + gen atomic.Int32 + // releasePath and indexPath are the request paths, release renders the + // release file listing index's sha256 for a generation. + releasePath, indexPath string + release func(indexSHA256 string, size int) string +} + +func indexBody(gen int32) string { return fmt.Sprintf("index generation %d\n", gen) } + +func (a *releaseArchive) serve(w http.ResponseWriter, r *http.Request) { + gen := a.gen.Load() + index := indexBody(gen) + switch r.URL.Path { + case "/" + a.releasePath: + _, _ = w.Write([]byte(a.release(sha256Hex(index), len(index)))) + case "/" + a.indexPath: + _, _ = w.Write([]byte(index)) + default: + http.NotFound(w, r) + } +} + +type releaseIndexCase struct { + name string + ecosystem string + archive *releaseArchive + handler func(p *Proxy, upstreamURL string) (http.Handler, func(path string) string) +} + +func releaseIndexCases() []releaseIndexCase { + return []releaseIndexCase{ + { + name: "debian InRelease", + ecosystem: "debian", + archive: &releaseArchive{ + releasePath: "dists/stable/InRelease", + indexPath: "dists/stable/main/binary-amd64/Packages", + release: func(sum string, size int) string { + return "-----BEGIN PGP SIGNED MESSAGE-----\nHash: SHA512\n\nSuite: stable\nMD5Sum:\n 00000000000000000000000000000000 1 main/binary-amd64/Packages\nSHA256:\n" + + fmt.Sprintf(" %s %d main/binary-amd64/Packages\n", sum, size) + + "-----BEGIN PGP SIGNATURE-----\n\nxx\n-----END PGP SIGNATURE-----\n" + }, + }, + handler: func(p *Proxy, upstreamURL string) (http.Handler, func(string) string) { + h := NewDebianHandler(p, "http://proxy.test", upstreamURL, nil) + return h.Routes(), func(path string) string { return h.metadataCacheKeyFor("", h.upstreamURL, path) } + }, + }, + { + name: "rpm repomd.xml", + ecosystem: "rpm", + archive: &releaseArchive{ + releasePath: "f/39/os/repodata/repomd.xml", + indexPath: "f/39/os/repodata/primary.xml.gz", + release: func(sum string, _ int) string { + return ` +1 +` + sum + ` +` + }, + }, + handler: func(p *Proxy, upstreamURL string) (http.Handler, func(string) string) { + h := NewRPMHandlerWithUpstream(p, "http://proxy.test", upstreamURL) + return h.Routes(), h.metadataCacheKey + }, + }, + } +} + +func getBody(t *testing.T, h http.Handler, path string) string { + t.Helper() + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/"+path, nil)) + if w.Code != http.StatusOK { + t.Fatalf("GET %s = %d %s", path, w.Code, w.Body) + } + return w.Body.String() +} + +// TestIndexFollowsCachedRelease checks that the index files a release file +// lists are served as that release file lists them, whatever the index's own +// cache entry says. +func TestIndexFollowsCachedRelease(t *testing.T) { + for _, tc := range releaseIndexCases() { + t.Run(tc.name, func(t *testing.T) { + a := tc.archive + upstream := httptest.NewServer(http.HandlerFunc(a.serve)) + t.Cleanup(upstream.Close) + proxy, db, _, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + proxy.HTTPClient = upstream.Client() + h, key := tc.handler(proxy, upstream.URL) + expire := func(path string) { + t.Helper() + if _, err := db.ExpireMetadataCache(tc.ecosystem, []string{key(path)}, time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + } + + getBody(t, h, a.releasePath) + if got := getBody(t, h, a.indexPath); got != indexBody(0) { + t.Fatalf("index = %q", got) + } + + // The index entry expires before the release file's while + // upstream moves on: the cached release still lists generation 0. + a.gen.Store(1) + expire(a.indexPath) + if got := getBody(t, h, a.indexPath); got != indexBody(0) { + t.Errorf("index after its entry expired = %q, want the one the cached release lists", got) + } + + // The release file is refetched while the index is still cached: + // the index must follow it. + expire(a.releasePath) + getBody(t, h, a.releasePath) + if got := getBody(t, h, a.indexPath); got != indexBody(1) { + t.Errorf("index after release refetch = %q, want %q", got, indexBody(1)) + } + }) + } +} + +// TestIndexMismatchExpiresRelease checks that when upstream only has an +// index newer than the cached release file, the release file is refetched +// on its next request, so the next update gets a matching pair. +func TestIndexMismatchExpiresRelease(t *testing.T) { + for _, tc := range releaseIndexCases() { + t.Run(tc.name, func(t *testing.T) { + a := tc.archive + upstream := httptest.NewServer(http.HandlerFunc(a.serve)) + t.Cleanup(upstream.Close) + proxy, _, _, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + proxy.HTTPClient = upstream.Client() + h, _ := tc.handler(proxy, upstream.URL) + + first := getBody(t, h, a.releasePath) + a.gen.Store(1) + if got := getBody(t, h, a.indexPath); got != indexBody(1) { + t.Fatalf("index = %q", got) + } + if got := getBody(t, h, a.releasePath); got == first { + t.Errorf("release file still the cached one, which does not list the index served") + } + }) + } +} + +// debianSuite serves an APT suite with both InRelease and Release, and the +// legacy per-component Release that lists no checksums. +func debianSuite(t *testing.T) (*atomic.Int32, http.Handler, func(path string)) { + t.Helper() + const index = "main/binary-amd64/Packages" + var gen atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body := indexBody(gen.Load()) + release := fmt.Sprintf("Suite: stable\nSHA256:\n %s %d %s\n", sha256Hex(body), len(body), index) + switch r.URL.Path { + case "/dists/stable/InRelease": + _, _ = w.Write([]byte("-----BEGIN PGP SIGNED MESSAGE-----\n\n" + release + "-----BEGIN PGP SIGNATURE-----\n")) + case "/dists/stable/Release": + _, _ = w.Write([]byte(release)) + case "/dists/stable/main/binary-amd64/Release": + _, _ = w.Write([]byte("Archive: stable\nComponent: main\nArchitecture: amd64\n")) + case "/dists/stable/" + index: + _, _ = w.Write([]byte(body)) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(upstream.Close) + proxy, db, _, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + proxy.HTTPClient = upstream.Client() + h := NewDebianHandler(proxy, "http://proxy.test", upstream.URL, nil) + expire := func(path string) { + t.Helper() + key := h.metadataCacheKeyFor("", h.upstreamURL, path) + if _, err := db.ExpireMetadataCache("debian", []string{key}, time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + } + return &gen, h.Routes(), expire +} + +// TestDebianIndexFollowsLatestReleaseFile checks that with both InRelease +// and Release cached, the index follows the one fetched last, which is what +// the client asking for the index just read. +func TestDebianIndexFollowsLatestReleaseFile(t *testing.T) { + const index = "dists/stable/main/binary-amd64/Packages" + gen, h, expire := debianSuite(t) + + getBody(t, h, "dists/stable/InRelease") + getBody(t, h, index) + gen.Store(1) + time.Sleep(time.Millisecond) + getBody(t, h, "dists/stable/Release") + if got := getBody(t, h, index); got != indexBody(1) { + t.Errorf("index after a fresh Release = %q, want %q", got, indexBody(1)) + } + + // And back: a fresh InRelease wins over the older Release. + gen.Store(2) + expire("dists/stable/InRelease") + time.Sleep(time.Millisecond) + getBody(t, h, "dists/stable/InRelease") + if got := getBody(t, h, index); got != indexBody(2) { + t.Errorf("index after a fresh InRelease = %q, want %q", got, indexBody(2)) + } +} + +// TestDebianIndexSkipsReleaseWithoutListing checks that a cached release +// file that lists nothing for the index, such as the legacy per-component +// Release, does not stop the lookup at the suite's release file. +func TestDebianIndexSkipsReleaseWithoutListing(t *testing.T) { + const index = "dists/stable/main/binary-amd64/Packages" + gen, h, expire := debianSuite(t) + + getBody(t, h, "dists/stable/InRelease") + getBody(t, h, "dists/stable/main/binary-amd64/Release") + getBody(t, h, index) + gen.Store(1) + expire("dists/stable/InRelease") + getBody(t, h, "dists/stable/InRelease") + if got := getBody(t, h, index); got != indexBody(1) { + t.Errorf("index after InRelease refetch = %q, want %q", got, indexBody(1)) + } +} + +// TestCachedReleaseChecksStoredBytes checks that a release file whose stored +// bytes no longer match its row, because another instance replaced them, is +// not used to decide which index to serve. +func TestCachedReleaseChecksStoredBytes(t *testing.T) { + proxy, db, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + path := seedMetadataRow(t, db, store) + if proxy.readCachedRelease(context.Background(), "test", "pkg") == nil { + t.Fatal("matching release file not read") + } + store.mu.Lock() + store.files[path] = []byte(`{"v":2}`) + store.mu.Unlock() + if release := proxy.readCachedRelease(context.Background(), "test", "pkg"); release != nil { + t.Errorf("read a release file whose bytes don't match its row: %q", release.body) + } +} diff --git a/internal/handler/rpm.go b/internal/handler/rpm.go index 62b2798c..7363cca1 100644 --- a/internal/handler/rpm.go +++ b/internal/handler/rpm.go @@ -132,14 +132,27 @@ func (h *RPMHandler) handlePackageDownload(w http.ResponseWriter, r *http.Reques // handleMetadata proxies repository metadata files (repomd.xml, // primary.xml.gz, etc.). repomd.xml changes with every repository update and // follows the metadata TTL. The files it points at are usually named after -// their own checksum and so never change; those are kept for good. +// their own checksum and so never change; those are kept for good. Plainly +// named ones follow the cached repomd.xml (see proxyListedIndex). func (h *RPMHandler) handleMetadata(w http.ResponseWriter, r *http.Request, path string) { upstreamURL := fmt.Sprintf("%s/%s", h.upstreamURL, path) if rpmContentAddressed(path) { h.proxy.proxyImmutable(w, r, upstreamURL, "rpm", h.metadataCacheKey(path), "*/*") return } - h.proxy.ProxyCached(w, r, upstreamURL, "rpm", h.metadataCacheKey(path), "*/*") + i := strings.LastIndex(path, "/repodata/") + if strings.HasPrefix(path[i:], "/repodata/repomd.xml") { + h.proxy.ProxyCached(w, r, upstreamURL, "rpm", h.metadataCacheKey(path), "*/*") + return + } + // Plainly named files (createrepo --simple-md-filenames) follow the + // cached repomd.xml that lists them. + repomd := h.proxy.readCachedRelease(r.Context(), "rpm", h.metadataCacheKey(path[:i]+"/repodata/repomd.xml")) + var want string + if repomd != nil { + want = repomdSHA256(repomd.body, path[i+1:]) + } + h.proxy.proxyListedIndex(w, r, upstreamURL, "rpm", h.metadataCacheKey(path), repomd, want) } // rpmChecksumPrefix matches a repodata filename that createrepo prefixed