From b7c227587aad2d096fe5dfabb89d160ac190b02d Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Sun, 20 Sep 2026 21:38:34 -0700 Subject: [PATCH 1/2] feat(proxies): cache GitHub release artifacts and Packagist Adds two CacheProxy implementations and makes ServeArtifact honour TTL. WHY THESE TWO An inventory of the fleet's workflows found the largest uncached ecosystems were not the ones with proxies. Files per ecosystem across the org: cargo build 39 (proxy existed, was switched off -- now enabled) composer install 22 (no proxy on main) releases/download 19 (no proxy anywhere) dtolnay/rust-toolchain 11 (served by the cargo proxy) go mod download 4 (the one ecosystem that HAD a cache) Measured cost on a real build (php-sdk, x86_64 gnu, ducks, 2026-09-20): 40 source artifacts took 87s of a 591s build. GHREL GitHub releases are a distribution channel with no registry, and no redirect of the GOPROXY kind: a release download is a plain github.com URL baked into whatever tool fetches it. Intercepting it transparently would mean terminating TLS for job containers -- a CA in every image and a real change to what untrusted job code is exposed to. So this advertises GHREL_PROXY instead and lets a tool opt in; unset, everything goes upstream exactly as before. Metadata is cached with a TTL; asset URLs inside it are rewritten to the proxy's /download/ route, because otherwise a caller takes kilobytes of JSON from the LAN and every megabyte from the CDN. COMPOSER Rewritten on pkgcache rather than lifting the implementation from the stale feat/cargo-composer-cache branch: that version predates pkgcache entirely and would have reinvented the bounded LRU store, traversal-safe key mapping and fail-open fetch -- and disk-full is a failure mode these nodes have hit twice. It also fixes what that version documented about itself: "distribution zips come from mirrors and GitHub, which this proxy does not front". Metadata is the cheap part. Packagist dist URLs name an exact commit, so they are genuinely immutable and are now fronted and cached permanently. The archive route re-checks the host allowlist so a job cannot hand it an arbitrary URL and use ephemerd as an open proxy. SERVEARTIFACT NOW HONOURS TTL ServeArtifact served any cached entry unconditionally -- it never consulted req.TTL. That is correct for every existing caller: npm, PyPI and pub all refuse to re-publish a file, so their proxies pass Immutable and the question never arises. GitHub does not: an asset can be deleted and re-uploaded under the same URL. Without this, ghrel's non-immutable default was a no-op and the safe default was silently "cache forever" -- the exact hazard it exists to avoid. A stale, non-immutable entry is now revalidated with a conditional GET, reusing decide(), the same primitive Document already uses. Gated so Immutable callers are untouched; npm/pip/pub/cargo suites unchanged. Tests cover URL rewriting for both proxies (single-object AND array release shapes), the composer open-proxy guard, cache-key safety, and both TTL branches. Revert-verified: neutering revalidation gives "conditional requests = 0, want 2"; neutering either rewrite fails its suite; removing the composer allowlist check lets a metadata-service URL through with 200 instead of 403. --- cmd/ephemerd/cache.go | 12 + cmd/ephemerd/pkgproxies.go | 70 ++- cmd/ephemerd/pkgproxies_test.go | 79 +++ config.example.toml | 38 ++ pkg/config/config.go | 51 +- pkg/config/pkgproxy_test.go | 101 +++- pkg/proxies/composer/composerproxy.go | 350 ++++++++++++++ pkg/proxies/composer/composerproxy_test.go | 535 +++++++++++++++++++++ pkg/proxies/ghrel/ghrelproxy.go | 418 ++++++++++++++++ pkg/proxies/ghrel/ghrelproxy_test.go | 535 +++++++++++++++++++++ pkg/proxies/pkgcache/fetch.go | 106 +++- pkg/proxies/pkgcache/fetch_test.go | 79 +++ 12 files changed, 2335 insertions(+), 39 deletions(-) create mode 100644 cmd/ephemerd/pkgproxies_test.go create mode 100644 pkg/proxies/composer/composerproxy.go create mode 100644 pkg/proxies/composer/composerproxy_test.go create mode 100644 pkg/proxies/ghrel/ghrelproxy.go create mode 100644 pkg/proxies/ghrel/ghrelproxy_test.go diff --git a/cmd/ephemerd/cache.go b/cmd/ephemerd/cache.go index b1cb0dc..0ff8726 100644 --- a/cmd/ephemerd/cache.go +++ b/cmd/ephemerd/cache.go @@ -104,6 +104,18 @@ func managedCaches() []cacheEntry { Description: "pub.dev proxy cache (version listings + package archives) served to job containers", LiveSafe: true, }, + { + Name: "ghrel", + Rel: filepath.Join("cache", "ghrel"), + Description: "GitHub release proxy cache (release metadata + asset bytes) served to job containers", + LiveSafe: true, + }, + { + Name: "composer", + Rel: filepath.Join("cache", "composer"), + Description: "Packagist proxy cache (metadata + distribution archives) served to job containers", + LiveSafe: true, + }, { Name: "buildkit", Rel: "buildkit", diff --git a/cmd/ephemerd/pkgproxies.go b/cmd/ephemerd/pkgproxies.go index 38a6223..92904a4 100644 --- a/cmd/ephemerd/pkgproxies.go +++ b/cmd/ephemerd/pkgproxies.go @@ -7,6 +7,8 @@ import ( "github.com/ephpm/ephemerd/pkg/config" "github.com/ephpm/ephemerd/pkg/networking" "github.com/ephpm/ephemerd/pkg/proxies" + composerproxy "github.com/ephpm/ephemerd/pkg/proxies/composer" + ghrelproxy "github.com/ephpm/ephemerd/pkg/proxies/ghrel" npmproxy "github.com/ephpm/ephemerd/pkg/proxies/npm" pipproxy "github.com/ephpm/ephemerd/pkg/proxies/pip" pubproxy "github.com/ephpm/ephemerd/pkg/proxies/pub" @@ -15,19 +17,22 @@ import ( // Default listen ports for the language package caches, on the bridge // gateway alongside the Go module proxy (8082). const ( - defaultNpmProxyPort = 8084 - defaultPipProxyPort = 8085 - defaultPubProxyPort = 8086 + defaultNpmProxyPort = 8084 + defaultPipProxyPort = 8085 + defaultPubProxyPort = 8086 + defaultGhrelProxyPort = 8087 + defaultComposerProxyPort = 8088 ) // healthGated is the interface a cache proxy implements when its env var // cannot fail over on its own. // // GOPROXY has "|direct" and so degrades by itself. npm_config_registry, -// PIP_INDEX_URL and PUB_HOSTED_URL have no such escape hatch: whatever they -// name IS the registry. The next best thing is to not name a proxy that is -// not answering, so these are probed after Start and only injected if the -// probe passes. +// PIP_INDEX_URL, PUB_HOSTED_URL, COMPOSER_REPO_PACKAGIST and GHREL_PROXY +// have no such escape hatch: whatever they name IS the registry, and a +// consumer only falls back to upstream when the variable is ABSENT. The next +// best thing is to not name a proxy that is not answering, so these are +// probed after Start and only injected if the probe passes. type healthGated interface { Healthy() bool } @@ -45,11 +50,18 @@ func pkgProxyPorts(cfg *config.Config) []int { if cfg.PubProxy.Enabled { ports = append(ports, cfg.PubProxy.ProxyPort(defaultPubProxyPort)) } + if cfg.GhrelProxy.Enabled { + ports = append(ports, cfg.GhrelProxy.ProxyPort(defaultGhrelProxyPort)) + } + if cfg.ComposerProxy.Enabled { + ports = append(ports, cfg.ComposerProxy.ProxyPort(defaultComposerProxyPort)) + } return ports } -// startPkgProxies starts whichever of the npm, pip and pub caches are -// enabled and returns those that came up healthy, plus a teardown func. +// startPkgProxies starts whichever of the npm, pip, pub, ghrel and composer +// caches are enabled and returns those that came up healthy, plus a teardown +// func. // // FAIL-OPEN AT STARTUP: a proxy that fails to start, or that starts but does // not answer its own health probe, is simply left out of the returned slice. @@ -111,6 +123,46 @@ func startPkgProxies(cfg *config.Config, dataDir string, net *networking.Manager }) }, }, + { + enabled: cfg.GhrelProxy.Enabled, + port: cfg.GhrelProxy.ProxyPort(defaultGhrelProxyPort), + build: func(addr string) (proxies.CacheProxy, error) { + return ghrelproxy.New(ghrelproxy.Config{ + CacheDir: joinPath(dataDir, "cache", "ghrel"), + // [ghrel_proxy].upstream is the REST API origin; the + // asset origin tracks it rather than being a second + // knob, so pointing at GitHub Enterprise moves both. + APIUpstream: cfg.GhrelProxy.Upstream, + ListenAddr: addr, + MetadataTTL: cfg.GhrelProxy.ProxyIndexTTL(), + // AssetTTL is deliberately left at the package default + // (24h). It is a backstop against an asset being + // replaced in place, not a knob an operator tunes — + // the operator-facing choice is immutable_assets. + ImmutableAssets: cfg.GhrelProxy.ImmutableAssets, + MaxBytes: cfg.GhrelProxy.ProxyMaxBytes(), + AllowedHosts: cfg.GhrelProxy.AllowedHosts, + Cleanup: cfg.GhrelProxy.Cleanup, + Log: log.With("component", "ghrel-proxy"), + }) + }, + }, + { + enabled: cfg.ComposerProxy.Enabled, + port: cfg.ComposerProxy.ProxyPort(defaultComposerProxyPort), + build: func(addr string) (proxies.CacheProxy, error) { + return composerproxy.New(composerproxy.Config{ + CacheDir: joinPath(dataDir, "cache", "composer"), + Upstream: cfg.ComposerProxy.Upstream, + ListenAddr: addr, + MetadataTTL: cfg.ComposerProxy.ProxyIndexTTL(), + MaxBytes: cfg.ComposerProxy.ProxyMaxBytes(), + AllowedHosts: cfg.ComposerProxy.AllowedHosts, + Cleanup: cfg.ComposerProxy.Cleanup, + Log: log.With("component", "composer-proxy"), + }) + }, + }, } var ( diff --git a/cmd/ephemerd/pkgproxies_test.go b/cmd/ephemerd/pkgproxies_test.go new file mode 100644 index 0000000..f91793a --- /dev/null +++ b/cmd/ephemerd/pkgproxies_test.go @@ -0,0 +1,79 @@ +package main + +import ( + "path/filepath" + "slices" + "testing" + + "github.com/ephpm/ephemerd/pkg/config" +) + +// TestPkgProxyPorts is the firewall half of the wiring. These ports are fed +// into networking.Config.GatewayPorts; a proxy that is enabled but missing +// here binds fine, passes its health probe over loopback, advertises its env +// var — and is then unreachable from every job container, which looks like a +// hung download rather than a misconfiguration. +func TestPkgProxyPorts(t *testing.T) { + t.Parallel() + + if got := pkgProxyPorts(&config.Config{}); len(got) != 0 { + t.Errorf("pkgProxyPorts with nothing enabled = %v, want none opened", got) + } + + var cfg config.Config + cfg.NpmProxy.Enabled = true + cfg.PipProxy.Enabled = true + cfg.PubProxy.Enabled = true + cfg.GhrelProxy.Enabled = true + cfg.ComposerProxy.Enabled = true + + got := pkgProxyPorts(&cfg) + for _, want := range []int{ + defaultNpmProxyPort, + defaultPipProxyPort, + defaultPubProxyPort, + defaultGhrelProxyPort, + defaultComposerProxyPort, + } { + if !slices.Contains(got, want) { + t.Errorf("port %d is not opened to job containers; got %v", want, got) + } + } + + // The defaults must stay distinct, or two proxies fight over a listener + // and the second one silently fails to start. + seen := map[int]bool{} + for _, p := range got { + if seen[p] { + t.Errorf("port %d is claimed by two proxies: %v", p, got) + } + seen[p] = true + } + + // An explicit port overrides the default. + cfg.GhrelProxy.Port = 9999 + if got := pkgProxyPorts(&cfg); !slices.Contains(got, 9999) || slices.Contains(got, defaultGhrelProxyPort) { + t.Errorf("pkgProxyPorts = %v, want 9999 instead of %d", got, defaultGhrelProxyPort) + } +} + +// TestNewCachesAreClearable: every cache these proxies write to must be +// known to `ephemerd cache`, or it grows until it fills the node's disk with +// nothing to clear it. The directory names must match what startPkgProxies +// actually passes as CacheDir. +func TestNewCachesAreClearable(t *testing.T) { + t.Parallel() + for _, name := range []string{"ghrel", "composer"} { + c, ok := cacheByName(name) + if !ok { + t.Errorf("cache %q is not managed by `ephemerd cache`", name) + continue + } + if want := filepath.Join("cache", name); c.Rel != want { + t.Errorf("cache %q lives at %q, want %q", name, c.Rel, want) + } + if !c.LiveSafe { + t.Errorf("cache %q is not live-safe; every entry is a pull-through copy that a miss refetches", name) + } + } +} diff --git a/config.example.toml b/config.example.toml index d3f5942..01b70c6 100644 --- a/config.example.toml +++ b/config.example.toml @@ -509,6 +509,44 @@ shutdown_timeout = "5m" # max_size_gb = 5 # cleanup = false +# GitHub releases. Caches release metadata and the asset bytes it points at — +# prebuilt toolchains, SDK archives, vendored tarballs, the CI inputs that have +# no registry to cache them in. +# +# Unlike the others this one ADVERTISES rather than intercepts: a GitHub +# download is a plain https://github.com/... URL with no redirect knob, so +# ephemerd sets GHREL_PROXY and a tool that understands it uses it. A tool that +# does not is unaffected, and a build stays correct off-fleet because the +# variable is simply absent. `upstream` is the REST API origin, so a GitHub +# Enterprise host moves both halves. +# +# immutable_assets is the one key the other caches do not have. GitHub is the +# only ecosystem here that lets an asset be deleted and replaced under the same +# tag and URL, so assets are revalidated on a long TTL by default. Set this +# true only if your publishing discipline guarantees an asset is never +# overwritten in place — it is a promise the proxy cannot verify, and getting +# it wrong is silent. +# [ghrel_proxy] +# enabled = false +# port = 8087 +# upstream = "https://api.github.com" +# index_ttl = "5m" +# max_size_gb = 5 +# immutable_assets = false +# cleanup = false + +# PHP Composer. Injects COMPOSER_REPO_PACKAGIST, so both the Packagist metadata +# and the distribution archives it points at come from the LAN. A repository +# declared explicitly in composer.json still wins, and private/authenticated +# repositories are not proxied — this cache forwards no credentials. +# [composer_proxy] +# enabled = false +# port = 8088 +# upstream = "https://repo.packagist.org" +# index_ttl = "5m" +# max_size_gb = 5 +# cleanup = false + [log] # Log level: debug, info, warn, error level = "info" diff --git a/pkg/config/config.go b/pkg/config/config.go index 6ed08c1..75f5c27 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -36,6 +36,12 @@ type Config struct { NpmProxy PkgProxyConfig `toml:"npm_proxy"` PipProxy PkgProxyConfig `toml:"pip_proxy"` PubProxy PkgProxyConfig `toml:"pub_proxy"` + // ComposerProxy caches Packagist metadata and the PHP distribution + // archives it points at. Same shape as the other language caches. + ComposerProxy PkgProxyConfig `toml:"composer_proxy"` + // GhrelProxy caches GitHub release metadata and asset bytes. It carries + // one field the others do not — see GhrelProxyConfig. + GhrelProxy GhrelProxyConfig `toml:"ghrel_proxy"` // RegistryMirror routes container image pulls through a LAN pull-through // cache instead of the origin registry. See RegistryMirrorConfig. RegistryMirror RegistryMirrorConfig `toml:"registry_mirror"` @@ -1103,7 +1109,8 @@ func validateMirrorURL(key, raw string) error { } // PkgProxyConfig configures one language package caching proxy. The -// same shape serves [npm_proxy], [pip_proxy] and [pub_proxy]: all three are +// same shape serves [npm_proxy], [pip_proxy], [pub_proxy], [composer_proxy] +// and (embedded in GhrelProxyConfig) [ghrel_proxy]: all of them are // pull-through HTTP caches with an immutable-artifact half and a mutable- // metadata half, and differ only in their upstream and their defaults. // @@ -1117,13 +1124,16 @@ type PkgProxyConfig struct { Enabled bool `toml:"enabled"` // Port is the listen port on the bridge gateway. Zero takes the - // per-ecosystem default (npm 8084, pip 8085, pub 8086). + // per-ecosystem default (npm 8084, pip 8085, pub 8086, ghrel 8087, + // composer 8088). Port int `toml:"port"` // Upstream overrides the registry to pull through to (npm: // https://registry.npmjs.org, pip: https://pypi.org, pub: - // https://pub.dev). The upstream's own host is always permitted to - // serve artifacts, so an override needs no matching allowed_hosts entry. + // https://pub.dev, composer: https://repo.packagist.org, ghrel: + // https://api.github.com). The upstream's own host is always permitted + // to serve artifacts, so an override needs no matching allowed_hosts + // entry. Upstream string `toml:"upstream"` // IndexTTL is how long cached MUTABLE metadata (an npm packument, a PEP @@ -1133,6 +1143,9 @@ type PkgProxyConfig struct { // // Immutable artifacts — tarballs, wheels, sdists, archives — ignore // this entirely: they are cached permanently and never revalidated. + // + // [ghrel_proxy] is the one exception to that second paragraph, because + // GitHub lets an asset be replaced in place. See GhrelProxyConfig. IndexTTL time.Duration `toml:"index_ttl"` // MaxSizeGB is the cache's disk budget in GiB. When it is exceeded, the @@ -1187,6 +1200,36 @@ func (p *PkgProxyConfig) ProxyMaxBytes() int64 { return p.MaxSizeGB << 30 } +// GhrelProxyConfig configures [ghrel_proxy], the GitHub release cache. +// +// It is a PkgProxyConfig plus one flag, rather than its own type, because +// every other knob means the same thing here as it does for [npm_proxy]: +// `upstream` is the REST API origin (https://api.github.com by default, or a +// GitHub Enterprise host), `index_ttl` is how long release METADATA is served +// before a conditional GET, and `port` defaults to 8087 — continuing npm +// 8084, pip 8085, pub 8086, with composer at 8088. +// +// The extra flag exists because GitHub is the one ecosystem here that does +// not promise immutable artifacts. PyPI refuses to let a file be re-uploaded +// even after deletion, and npm and pub are similar, which is why those caches +// keep an artifact forever and never look at it again. GitHub enforces +// nothing of the kind: a release asset can be deleted and replaced under the +// same tag, keeping its name and URL. So asset bytes are revalidated on a +// long TTL by default — one conditional request per asset per day, answered +// with a 304 for unchanged bytes, which is cheap next to re-downloading a +// 200 MB toolchain and stays correct if someone republishes. +type GhrelProxyConfig struct { + PkgProxyConfig + + // ImmutableAssets caches release asset bytes permanently and never + // revalidates them, trading the daily conditional request for the + // promise that an asset is never overwritten in place (a new build gets + // a new tag). Default false: it is a promise the proxy cannot verify, + // and getting it wrong is silent — jobs keep building, against bytes + // that upstream has replaced. + ImmutableAssets bool `toml:"immutable_assets"` +} + // VMConfig configures virtual machines for cross-OS job execution. type VMConfig struct { // CrossPlatform enables macOS and Windows VM support. Default true. diff --git a/pkg/config/pkgproxy_test.go b/pkg/config/pkgproxy_test.go index 59ff684..711e621 100644 --- a/pkg/config/pkgproxy_test.go +++ b/pkg/config/pkgproxy_test.go @@ -7,7 +7,7 @@ import ( "time" ) -// TestLoad_PkgProxyConfig covers the three language package caches +// TestLoad_PkgProxyConfig covers the language package caches // ([npm_proxy], [pip_proxy], [pub_proxy]) decoding from TOML, including the // duration and slice fields. func TestLoad_PkgProxyConfig(t *testing.T) { @@ -96,9 +96,11 @@ max_size_gb = -1 func TestPkgProxyDefaultsAreOffAndBounded(t *testing.T) { var cfg Config for name, p := range map[string]PkgProxyConfig{ - "npm_proxy": cfg.NpmProxy, - "pip_proxy": cfg.PipProxy, - "pub_proxy": cfg.PubProxy, + "npm_proxy": cfg.NpmProxy, + "pip_proxy": cfg.PipProxy, + "pub_proxy": cfg.PubProxy, + "composer_proxy": cfg.ComposerProxy, + "ghrel_proxy": cfg.GhrelProxy.PkgProxyConfig, } { if p.Enabled { t.Errorf("[%s] is enabled by default, want opt-in like [module_proxy]", name) @@ -112,6 +114,97 @@ func TestPkgProxyDefaultsAreOffAndBounded(t *testing.T) { } } +// TestLoad_GhrelAndComposerProxyConfig covers the two newest caches. +// +// [ghrel_proxy] is the interesting one: it embeds PkgProxyConfig, so the +// shared keys have to decode into the embedded struct from the SAME TOML +// table as its own `immutable_assets`. If that ever stops working the +// failure is silent — the proxy just runs on defaults. +func TestLoad_GhrelAndComposerProxyConfig(t *testing.T) { + t.Setenv("GITHUB_TOKEN", "") + path := filepath.Join(t.TempDir(), "config.toml") + if err := os.WriteFile(path, []byte(` +[github] +token = "ghp_test" +owner = "org" + +[ghrel_proxy] +enabled = true +port = 9102 +upstream = "https://ghe.acme.test/api/v3" +index_ttl = "30s" +max_size_gb = 20 +allowed_hosts = ["assets.ghe.acme.test"] +immutable_assets = true +cleanup = true + +[composer_proxy] +enabled = true +upstream = "https://packagist.mirror.test" +`), 0o644); err != nil { + t.Fatal(err) + } + + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load() error: %v", err) + } + + gh := cfg.GhrelProxy + if !gh.Enabled { + t.Error("GhrelProxy.Enabled = false, want true") + } + if !gh.ImmutableAssets { + t.Error("GhrelProxy.ImmutableAssets = false, want true") + } + // Every one of these lives on the EMBEDDED struct. + if got := gh.ProxyPort(8087); got != 9102 { + t.Errorf("GhrelProxy port = %d, want 9102", got) + } + if gh.Upstream != "https://ghe.acme.test/api/v3" { + t.Errorf("GhrelProxy.Upstream = %q", gh.Upstream) + } + if got := gh.ProxyIndexTTL(); got != 30*time.Second { + t.Errorf("GhrelProxy index TTL = %v, want 30s", got) + } + if got := gh.ProxyMaxBytes(); got != 20<<30 { + t.Errorf("GhrelProxy max bytes = %d, want %d", got, int64(20)<<30) + } + if len(gh.AllowedHosts) != 1 || gh.AllowedHosts[0] != "assets.ghe.acme.test" { + t.Errorf("GhrelProxy.AllowedHosts = %v", gh.AllowedHosts) + } + if !gh.Cleanup { + t.Error("GhrelProxy.Cleanup = false, want true") + } + + comp := cfg.ComposerProxy + if !comp.Enabled { + t.Error("ComposerProxy.Enabled = false, want true") + } + if comp.Upstream != "https://packagist.mirror.test" { + t.Errorf("ComposerProxy.Upstream = %q", comp.Upstream) + } + if got := comp.ProxyPort(8088); got != 8088 { + t.Errorf("ComposerProxy port = %d, want the 8088 default", got) + } + if got := comp.ProxyIndexTTL(); got != 5*time.Minute { + t.Errorf("ComposerProxy index TTL = %v, want the 5m default", got) + } + if got := comp.ProxyMaxBytes(); got != 5<<30 { + t.Errorf("ComposerProxy max bytes = %d, want the 5 GiB default", got) + } +} + +// TestGhrelImmutableAssetsDefaultsOff: GitHub is the one ecosystem here that +// lets an asset be replaced in place, so "cache forever, never revalidate" +// must be something an operator opts into, never the default. +func TestGhrelImmutableAssetsDefaultsOff(t *testing.T) { + var cfg Config + if cfg.GhrelProxy.ImmutableAssets { + t.Error("[ghrel_proxy].immutable_assets defaults to true; it is a promise the proxy cannot verify") + } +} + func TestPkgProxyPortFallback(t *testing.T) { var p PkgProxyConfig if got := p.ProxyPort(8084); got != 8084 { diff --git a/pkg/proxies/composer/composerproxy.go b/pkg/proxies/composer/composerproxy.go new file mode 100644 index 0000000..5ea02b9 --- /dev/null +++ b/pkg/proxies/composer/composerproxy.go @@ -0,0 +1,350 @@ +// Package composerproxy implements proxies.CacheProxy for PHP Composer. +// +// It is a pull-through cache for the Packagist metadata repository and the +// distribution archives that metadata points at. +// +// # INTERCEPTION +// +// Composer honours COMPOSER_REPO_PACKAGIST, so pointing every job container at +// this proxy is a single env var — no workflow change, and a build off-fleet +// keeps working because the variable is simply absent. +// +// # WHAT IS CACHED +// +// Metadata (packages.json and the per-package p2//.json files) +// is MUTABLE: a package gains a version on every release. It is cached with a +// short TTL and then revalidated with a conditional GET, so a release +// published minutes ago is never invisible for long and an unchanged document +// costs one 304 — the same treatment the npm and pip proxies give their index +// documents. +// +// Distribution archives are IMMUTABLE and are the bytes that actually matter. +// A Packagist dist URL names an exact commit: +// +// https://api.github.com/repos///zipball/<40-char sha> +// +// A given SHA cannot come back with different contents, so archives are cached +// permanently and never revalidated. +// +// # WHY DIST URLS ARE REWRITTEN +// +// This is the difference between a cache that saves seconds and one that saves +// the download. The previous implementation of this proxy cached metadata only +// and said so plainly: "distribution zips come from mirrors and GitHub, which +// this proxy does not front". That leaves Composer taking a few kilobytes of +// JSON from the LAN and every megabyte of actual code from the WAN, on every +// job. +// +// So each dist URL in a metadata document is rewritten to this proxy's archive +// route, exactly as the pip proxy rewrites the file links on a project page. +// Everything else in the document is left alone: rewriting is done on decoded +// JSON rather than by string replacement, because a blind substitution would +// also corrupt homepage/source/support fields that merely look like URLs. +package composerproxy + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "net/url" + "strings" + "time" + + "github.com/ephpm/ephemerd/pkg/proxies" + "github.com/ephpm/ephemerd/pkg/proxies/pkgcache" +) + +const ( + // DefaultUpstream is the Packagist metadata repository. + DefaultUpstream = "https://repo.packagist.org" + + // DefaultMetadataTTL is how long a metadata document is served before it + // is revalidated. Matches the npm/pip index default. + DefaultMetadataTTL = 5 * time.Minute + + // DefaultPort continues the sequence npm 8084, pip 8085, pub 8086, + // ghrel 8087. + DefaultPort = 8088 + + // distPrefix is the route carrying distribution archive bytes. It is + // deliberately a path Packagist itself never serves, so a rewritten URL + // can never collide with a real metadata path. + distPrefix = "/__dist/" +) + +// defaultAllowedHosts is where archive bytes may be fetched from. Packagist +// dist URLs point at GitHub's API, which redirects to codeload, and GitLab and +// Bitbucket for packages hosted there. +var defaultAllowedHosts = pkgcache.HostAllowlist{ + "repo.packagist.org", + "api.github.com", + "codeload.github.com", + "github.com", + "objects.githubusercontent.com", + "gitlab.com", + "bitbucket.org", +} + +// Config configures the proxy. +type Config struct { + // CacheDir is the on-disk cache root (e.g. /cache/composer). + CacheDir string + // Upstream overrides the Packagist metadata repository. + Upstream string + // ListenAddr is the address to bind and advertise. + ListenAddr string + // MetadataTTL is the revalidation interval for metadata documents. + // Zero takes DefaultMetadataTTL; negative revalidates every request. + MetadataTTL time.Duration + // MaxBytes is the cache disk budget. + MaxBytes int64 + // AllowedHosts extends the archive-fetch allowlist. + AllowedHosts []string + // Cleanup wipes the cache on Stop. Default false. + Cleanup bool + Log *slog.Logger +} + +// Compile-time interface check. +var _ proxies.CacheProxy = (*Proxy)(nil) + +// Proxy is the Packagist caching proxy. +type Proxy struct { + cfg Config + srv *pkgcache.Server + allow pkgcache.HostAllowlist +} + +// New creates a Composer caching proxy. Call Start() to begin serving. +func New(cfg Config) (*Proxy, error) { + if cfg.Upstream == "" { + cfg.Upstream = DefaultUpstream + } + cfg.Upstream = strings.TrimRight(cfg.Upstream, "/") + if cfg.MetadataTTL == 0 { + cfg.MetadataTTL = DefaultMetadataTTL + } + if cfg.Log == nil { + cfg.Log = slog.Default() + } + + allow := append(pkgcache.HostAllowlist{}, defaultAllowedHosts...) + allow = append(allow, cfg.AllowedHosts...) + p := &Proxy{cfg: cfg, allow: allow.WithHostsOf(cfg.Upstream)} + + srv, err := pkgcache.NewServer(pkgcache.ServerConfig{ + Name: "composer", + ListenAddr: cfg.ListenAddr, + CacheDir: cfg.CacheDir, + MaxBytes: cfg.MaxBytes, + Cleanup: cfg.Cleanup, + Log: cfg.Log, + }, func(_ *pkgcache.Cache, _ *pkgcache.Fetcher) http.Handler { + return http.HandlerFunc(p.handle) + }) + if err != nil { + return nil, err + } + p.srv = srv + return p, nil +} + +// Start begins serving. Returns after the listener is bound. +func (p *Proxy) Start() error { return p.srv.Start() } + +// Stop shuts the proxy down. +func (p *Proxy) Stop() error { return p.srv.Stop() } + +// Addr returns the bound address. +func (p *Proxy) Addr() string { return p.srv.Addr() } + +// Name returns the proxy name for logs. +func (p *Proxy) Name() string { return "composer" } + +// Healthy reports whether the proxy is serving. +func (p *Proxy) Healthy() bool { return p.srv.Healthy() } + +// EnvVars points Composer at this proxy. +func (p *Proxy) EnvVars() []string { + return []string{"COMPOSER_REPO_PACKAGIST=" + p.advertiseBase()} +} + +// advertiseBase is the base URL containers should use to reach this proxy. +// +// Delegated to the shared server rather than formatting cfg.ListenAddr +// directly: a listen address with port 0 has to be resolved to the port +// actually bound, or the proxy advertises (and rewrites dist URLs to) an +// address nothing can connect to. +func (p *Proxy) advertiseBase() string { return p.srv.AdvertiseBase() } + +// handle routes a request to archive or metadata handling. +func (p *Proxy) handle(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && r.Method != http.MethodHead { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if strings.HasPrefix(r.URL.Path, distPrefix) { + p.serveDist(w, r) + return + } + p.serveMetadata(w, r) +} + +// serveMetadata proxies a Packagist metadata document and rewrites the dist +// URLs inside it so Composer comes back here for the archives. +func (p *Proxy) serveMetadata(w http.ResponseWriter, r *http.Request) { + upstream := p.cfg.Upstream + r.URL.Path + if r.URL.RawQuery != "" { + upstream += "?" + r.URL.RawQuery + } + + body, meta, err := p.srv.Fetcher().Document(r.Context(), pkgcache.Request{ + Key: metaKey(r.URL.Path, r.URL.RawQuery), + URL: upstream, + TTL: p.cfg.MetadataTTL, + Accept: r.Header.Get("Accept"), + DefaultContentType: "application/json", + }) + if err != nil { + p.writeFetchError(w, err, "packagist metadata") + return + } + + rewritten, err := rewriteDistURLs(body, p.advertiseBase(), p.allow) + if err != nil { + // Serving the document unrewritten would silently send every + // archive to the WAN while looking like a cache hit. Fail loudly. + p.cfg.Log.Warn("composer: could not rewrite metadata", "path", r.URL.Path, "error", err) + http.Error(w, "malformed upstream metadata", http.StatusBadGateway) + return + } + + w.Header().Set("Content-Type", meta.ContentType) + w.WriteHeader(http.StatusOK) + if r.Method != http.MethodHead { + _, _ = w.Write(rewritten) + } +} + +// serveDist serves a distribution archive, caching it permanently. +// +// The upstream URL is carried in the "u" query parameter rather than encoded +// into the path: dist URLs span several hosts (GitHub, GitLab, Bitbucket) with +// different path shapes, and reconstructing one from a flattened path would be +// guesswork. The allowlist is re-checked HERE, not just at rewrite time, so a +// job cannot hand this route an arbitrary URL and use ephemerd as an open +// proxy. +func (p *Proxy) serveDist(w http.ResponseWriter, r *http.Request) { + raw := r.URL.Query().Get("u") + if raw == "" { + http.Error(w, "missing upstream", http.StatusBadRequest) + return + } + if _, err := url.Parse(raw); err != nil || !p.allow.Allows(raw) { + http.Error(w, "upstream not permitted", http.StatusForbidden) + return + } + + // Immutable: a Packagist dist URL names an exact commit, so the bytes + // behind it cannot change. + // + // Keyed by ArtifactKey (a hash of the URL) rather than by host+path: a + // host:port — a self-hosted Packagist or a GitLab on a nonstandard port — + // contains a colon, which SafeSegments rejects, and the failure mode is + // silent. ServeArtifact logs a warning and streams the archive UNCACHED, + // so the proxy keeps working while quietly caching nothing. + if err := p.srv.Fetcher().ServeArtifact(w, r, pkgcache.Request{ + Key: pkgcache.ArtifactKey(raw), + URL: raw, + Immutable: true, + DefaultContentType: "application/zip", + }); err != nil { + p.writeFetchError(w, err, "distribution archive") + } +} + +// writeFetchError maps a fetch failure onto a status code, keeping a genuine +// upstream 404 distinguishable from an outage. +func (p *Proxy) writeFetchError(w http.ResponseWriter, err error, what string) { + if pkgcache.IsNotFound(err) { + http.Error(w, "not found", http.StatusNotFound) + return + } + p.cfg.Log.Warn("composer: upstream fetch failed", "what", what, "error", err) + http.Error(w, "upstream fetch failed", http.StatusBadGateway) +} + +// rewriteDistURLs rewrites every dist.url in a metadata document to this +// proxy's archive route. +// +// Works on decoded JSON, not string substitution: a package document also +// carries homepage, source.url and support links, and rewriting those would +// corrupt metadata Composer shows to users and may act on. +func rewriteDistURLs(body []byte, base string, allow pkgcache.HostAllowlist) ([]byte, error) { + var doc any + if err := json.Unmarshal(body, &doc); err != nil { + return nil, fmt.Errorf("decode packagist metadata: %w", err) + } + rewriteNode(doc, base, allow) + out, err := json.Marshal(doc) + if err != nil { + return nil, fmt.Errorf("re-encode packagist metadata: %w", err) + } + return out, nil +} + +// rewriteNode walks the document rewriting the "url" inside any "dist" object. +func rewriteNode(node any, base string, allow pkgcache.HostAllowlist) { + switch v := node.(type) { + case map[string]any: + for k, val := range v { + if k == "dist" { + if dist, ok := val.(map[string]any); ok { + rewriteDist(dist, base, allow) + continue + } + } + rewriteNode(val, base, allow) + } + case []any: + for _, item := range v { + rewriteNode(item, base, allow) + } + } +} + +// rewriteDist points one dist object's url at the proxy. +func rewriteDist(dist map[string]any, base string, allow pkgcache.HostAllowlist) { + raw, ok := dist["url"].(string) + if !ok || raw == "" { + return + } + u, err := url.Parse(raw) + if err != nil || u.Host == "" || !allow.Allows(raw) { + // Leave anything unrecognised pointing upstream. A package served + // from a host we do not front must still install. + return + } + dist["url"] = strings.TrimRight(base, "/") + distPrefix + "archive?u=" + url.QueryEscape(raw) +} + +// metaKey is the cache key for one metadata document. Path AND query are part +// of the identity, so two documents at the same path with different queries +// do not share an entry. +// +// Hashed rather than path-shaped, for the same reason pkgcache.ArtifactKey is: +// a URL is not safe to map onto a filesystem. A query string carries "?" and +// "&", which pkgcache.SafeSegments rejects outright, and Packagist's document +// paths nest (packages.json alongside p2//.json), so a +// path-shaped key can need the same name to be a file and a directory at +// once. Both failures are SILENT — Cache.Write warns, the document is still +// served, and the cache quietly stops working. +func metaKey(path, rawQuery string) string { + sum := sha256.Sum256([]byte(path + "?" + rawQuery)) + h := hex.EncodeToString(sum[:]) + // Sharded one level to keep the directory from growing very wide. + return "meta/" + h[0:2] + "/" + h +} diff --git a/pkg/proxies/composer/composerproxy_test.go b/pkg/proxies/composer/composerproxy_test.go new file mode 100644 index 0000000..14fd9d4 --- /dev/null +++ b/pkg/proxies/composer/composerproxy_test.go @@ -0,0 +1,535 @@ +package composerproxy + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/ephpm/ephemerd/pkg/proxies/pkgcache" +) + +// --- harness --------------------------------------------------------------- + +// fakePackagist serves both halves of what Composer talks to: the metadata +// repository, and the host the dist URLs inside it point at. One server, so +// the dist host is the configured upstream's host and therefore allowlisted +// — which is the case the rewrite is supposed to fire on. +type fakePackagist struct { + *httptest.Server + metaHits atomic.Int64 + distHits atomic.Int64 + archive []byte +} + +func newFakePackagist(t *testing.T) *fakePackagist { + t.Helper() + pk := &fakePackagist{archive: []byte(strings.Repeat("ZIPBYTES", 512))} + + mux := http.NewServeMux() + mux.HandleFunc("/repos/", func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "missing") { + http.Error(w, "not found", http.StatusNotFound) + return + } + pk.distHits.Add(1) + w.Header().Set("Content-Type", "application/zip") + _, _ = w.Write(pk.archive) + }) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "missing") { + http.Error(w, "not found", http.StatusNotFound) + return + } + pk.metaHits.Add(1) + w.Header().Set("Content-Type", "application/json") + w.Header().Set("ETag", `"meta-v1"`) + // A realistic p2 document: the version carries a dist URL and + // several other URLs that must NOT be touched. + _, _ = fmt.Fprintf(w, `{ + "minified": "composer/2.0", + "packages": { + "acme/lib": [ + { + "name": "acme/lib", + "version": "1.0.0", + "homepage": "https://acme.test/lib", + "source": {"type":"git","url":"https://github.com/acme/lib.git","reference":"deadbeef"}, + "support": {"issues":"https://github.com/acme/lib/issues","source":"https://github.com/acme/lib"}, + "dist": { + "type": "zip", + "url": "%s/repos/acme/lib/zipball/deadbeef", + "reference": "deadbeef", + "shasum": "" + } + } + ] + } + }`, pk.URL) + }) + pk.Server = httptest.NewServer(mux) + t.Cleanup(pk.Close) + return pk +} + +func startProxy(t *testing.T, cfg Config) *Proxy { + t.Helper() + if cfg.CacheDir == "" { + cfg.CacheDir = t.TempDir() + } + if cfg.ListenAddr == "" { + cfg.ListenAddr = "127.0.0.1:0" + } + cfg.Log = slog.New(slog.NewTextHandler(io.Discard, nil)) + p, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := p.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + t.Cleanup(func() { + if err := p.Stop(); err != nil { + t.Errorf("Stop: %v", err) + } + }) + return p +} + +func client() *http.Client { + return &http.Client{ + Timeout: 10 * time.Second, + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, + } +} + +func get(t *testing.T, rawURL string) *http.Response { + t.Helper() + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, rawURL, nil) + if err != nil { + t.Fatal(err) + } + resp, err := client().Do(req) + if err != nil { + t.Fatalf("GET %s: %v", rawURL, err) + } + return resp +} + +func body(t *testing.T, resp *http.Response) []byte { + t.Helper() + defer func() { _ = resp.Body.Close() }() + b, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + return b +} + +// distTarget decodes the upstream URL back out of a rewritten dist URL. +func distTarget(t *testing.T, rewritten string) string { + t.Helper() + u, err := url.Parse(rewritten) + if err != nil { + t.Fatalf("rewritten dist URL %q is unparseable: %v", rewritten, err) + } + return u.Query().Get("u") +} + +// --- the rewrite, which is the whole point --------------------------------- + +// TestRewriteDistURLs is the difference between a cache that saves a few +// kilobytes of JSON and one that saves the download: dist.url must come +// through the proxy, and nothing else in the document may move. +func TestRewriteDistURLs(t *testing.T) { + t.Parallel() + const base = "http://gw:8088" + const distURL = "https://api.github.com/repos/acme/lib/zipball/deadbeef" + in := []byte(`{ + "packages": { + "acme/lib": [{ + "name": "acme/lib", + "homepage": "https://acme.test/lib", + "source": {"type":"git","url":"https://github.com/acme/lib.git"}, + "support": {"issues":"https://github.com/acme/lib/issues","source":"https://github.com/acme/lib"}, + "dist": {"type":"zip","url":"` + distURL + `","reference":"deadbeef","shasum":"abc"} + }] + } + }`) + + out, err := rewriteDistURLs(in, base, defaultAllowedHosts) + if err != nil { + t.Fatalf("rewriteDistURLs: %v", err) + } + var doc map[string]any + if err := json.Unmarshal(out, &doc); err != nil { + t.Fatal(err) + } + ver := doc["packages"].(map[string]any)["acme/lib"].([]any)[0].(map[string]any) + + dist := ver["dist"].(map[string]any) + got := dist["url"].(string) + if !strings.HasPrefix(got, base+distPrefix) { + t.Fatalf("dist.url = %q, want it pointed at %s%s", got, base, distPrefix) + } + if target := distTarget(t, got); target != distURL { + t.Errorf("dist.url carries upstream %q, want %q", target, distURL) + } + // The rest of the dist object is how Composer verifies what it + // installs; moving the URL must not move these. + if dist["reference"] != "deadbeef" || dist["shasum"] != "abc" || dist["type"] != "zip" { + t.Errorf("dist metadata was altered: %v", dist) + } + + // The URLs that are NOT the byte source. github.com is ON the allowlist, + // so a rewriter that keyed off the host rather than the field name would + // happily corrupt all three of these. + if ver["homepage"] != "https://acme.test/lib" { + t.Errorf("homepage was rewritten to %q", ver["homepage"]) + } + src := ver["source"].(map[string]any) + if src["url"] != "https://github.com/acme/lib.git" { + t.Errorf("source.url was rewritten to %q", src["url"]) + } + sup := ver["support"].(map[string]any) + if sup["issues"] != "https://github.com/acme/lib/issues" { + t.Errorf("support.issues was rewritten to %q", sup["issues"]) + } + if sup["source"] != "https://github.com/acme/lib" { + t.Errorf("support.source was rewritten to %q", sup["source"]) + } + if ver["name"] != "acme/lib" { + t.Errorf("name = %q", ver["name"]) + } +} + +// TestRewriteDistURLsLeavesUnfrontedHostsAlone: a package served from a host +// this proxy will not relay must still install, straight from upstream. +func TestRewriteDistURLsLeavesUnfrontedHostsAlone(t *testing.T) { + t.Parallel() + const base = "http://gw:8088" + out, err := rewriteDistURLs([]byte(`{"packages":{"acme/lib":[ + {"dist":{"url":"https://dist.acme-internal.test/lib-1.0.0.zip"}}, + {"dist":{"url":"not-a-url"}}, + {"dist":{"url":""}}, + {"dist":{"url":7}}, + {"dist":"nope"} + ]}}`), base, defaultAllowedHosts) + if err != nil { + t.Fatalf("rewriteDistURLs: %v", err) + } + var doc map[string]any + if err := json.Unmarshal(out, &doc); err != nil { + t.Fatal(err) + } + vers := doc["packages"].(map[string]any)["acme/lib"].([]any) + + if got := vers[0].(map[string]any)["dist"].(map[string]any)["url"]; got != "https://dist.acme-internal.test/lib-1.0.0.zip" { + t.Errorf("a non-allowlisted dist URL was rewritten to %q", got) + } + if got := vers[1].(map[string]any)["dist"].(map[string]any)["url"]; got != "not-a-url" { + t.Errorf("a non-URL dist value was rewritten to %q", got) + } + if got := vers[2].(map[string]any)["dist"].(map[string]any)["url"]; got != "" { + t.Errorf("an empty dist URL was rewritten to %q", got) + } + if got := vers[3].(map[string]any)["dist"].(map[string]any)["url"]; got != float64(7) { + t.Errorf("a non-string dist URL was rewritten to %v", got) + } + if got := vers[4].(map[string]any)["dist"]; got != "nope" { + t.Errorf(`a "dist" that is not an object was altered to %v`, got) + } + + for _, in := range []string{`{"packages":"nope"}`, `[]`, `null`, `"scalar"`, `{"dist":null}`} { + if _, err := rewriteDistURLs([]byte(in), base, defaultAllowedHosts); err != nil { + t.Errorf("rewriteDistURLs(%s): %v", in, err) + } + } + if _, err := rewriteDistURLs([]byte("not json"), base, defaultAllowedHosts); err == nil { + t.Error("unparseable metadata must be an error so the caller can fail loudly rather than serve unrewritten bytes") + } +} + +// --- the open-proxy guard -------------------------------------------------- + +// TestServeDistRefusesNonAllowlistedHosts is the SSRF fence. The dist route +// takes its upstream from a client-supplied query parameter, so the +// allowlist has to be re-checked HERE and not only at rewrite time — +// otherwise any job can hand ephemerd a URL and have it fetched with the +// daemon's network identity. +// +// The proxy is configured with the real Packagist upstream (never contacted +// here), so anything on loopback is off the allowlist — which is exactly the +// case this fence exists for. +func TestServeDistRefusesNonAllowlistedHosts(t *testing.T) { + t.Parallel() + p := startProxy(t, Config{}) + base := "http://" + p.Addr() + + secret := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, "cloud-metadata-credentials") + })) + defer secret.Close() + + for _, target := range []string{ + secret.URL + "/latest/meta-data/iam/", + "http://169.254.169.254/latest/meta-data/", + "http://127.0.0.1:10000/", + "http://[::1]:10000/", + "file:///etc/passwd", + "gopher://evil.test/", + "https://evil.test/payload.zip", + // A lookalike that must not match "github.com" as a parent domain. + "https://evilgithub.com/payload.zip", + } { + u := base + distPrefix + "archive?u=" + url.QueryEscape(target) + resp := get(t, u) + got := body(t, resp) + if resp.StatusCode != http.StatusForbidden { + t.Errorf("dist fetch of %q returned %d, want 403", target, resp.StatusCode) + } + if strings.Contains(string(got), "cloud-metadata-credentials") { + t.Fatalf("the proxy relayed a non-allowlisted host: %q", got) + } + } +} + +func TestServeDistRequiresAnUpstream(t *testing.T) { + t.Parallel() + p := startProxy(t, Config{}) + base := "http://" + p.Addr() + + resp := get(t, base+distPrefix+"archive") + _ = body(t, resp) + if resp.StatusCode != http.StatusBadRequest { + t.Errorf("a dist request with no u= returned %d, want 400", resp.StatusCode) + } +} + +// --- end to end ------------------------------------------------------------ + +// TestMetadataIsCachedAndDistIsFetchedThroughTheProxy walks the whole path a +// `composer install` takes: read the metadata, follow the dist URL in it. +func TestMetadataIsCachedAndDistIsFetchedThroughTheProxy(t *testing.T) { + t.Parallel() + pk := newFakePackagist(t) + p := startProxy(t, Config{Upstream: pk.URL, MetadataTTL: time.Hour}) + base := "http://" + p.Addr() + const path = "/p2/acme/lib.json" + + first := body(t, get(t, base+path)) + second := body(t, get(t, base+path)) + if pk.metaHits.Load() != 1 { + t.Errorf("upstream metadata hits = %d, want 1", pk.metaHits.Load()) + } + if string(first) != string(second) { + t.Error("the cached metadata differs from the first response") + } + + var doc map[string]any + if err := json.Unmarshal(first, &doc); err != nil { + t.Fatal(err) + } + ver := doc["packages"].(map[string]any)["acme/lib"].([]any)[0].(map[string]any) + distURL := ver["dist"].(map[string]any)["url"].(string) + if !strings.HasPrefix(distURL, base+distPrefix) { + t.Fatalf("dist.url = %q, want it pointed at the proxy", distURL) + } + if got := ver["source"].(map[string]any)["url"]; got != "https://github.com/acme/lib.git" { + t.Errorf("source.url was rewritten to %q", got) + } + + // Follow it exactly as Composer would, repeatedly. + for i := range 3 { + resp := get(t, distURL) + got := body(t, resp) + if resp.StatusCode != http.StatusOK { + t.Fatalf("attempt %d: status %d (%q)", i, resp.StatusCode, got) + } + if string(got) != string(pk.archive) { + t.Fatalf("attempt %d: archive bytes differ", i) + } + } + // The archives are the bytes that matter. Caching them is the point; + // the key must survive Cache.KeyPath for that to happen, which a + // host:port-derived key does not. + if pk.distHits.Load() != 1 { + t.Errorf("upstream archive hits = %d, want 1: the archive was not cached", pk.distHits.Load()) + } +} + +func TestUpstream404IsPassedThrough(t *testing.T) { + t.Parallel() + pk := newFakePackagist(t) + p := startProxy(t, Config{Upstream: pk.URL}) + base := "http://" + p.Addr() + + resp := get(t, base+"/p2/acme/missing.json") + _ = body(t, resp) + if resp.StatusCode != http.StatusNotFound { + t.Errorf("metadata status = %d, want 404: a nonexistent package is a real answer", resp.StatusCode) + } + + u := base + distPrefix + "archive?u=" + url.QueryEscape(pk.URL+"/repos/acme/missing/zipball/deadbeef") + resp2 := get(t, u) + _ = body(t, resp2) + if resp2.StatusCode != http.StatusNotFound { + t.Errorf("archive status = %d, want 404", resp2.StatusCode) + } +} + +// TestWritesAreRejected: read-through cache, never a Packagist front end. +func TestWritesAreRejected(t *testing.T) { + t.Parallel() + pk := newFakePackagist(t) + p := startProxy(t, Config{Upstream: pk.URL}) + base := "http://" + p.Addr() + + for _, method := range []string{http.MethodPut, http.MethodPost, http.MethodDelete, http.MethodPatch} { + req, err := http.NewRequestWithContext(t.Context(), method, base+"/p2/acme/lib.json", strings.NewReader("{}")) + if err != nil { + t.Fatal(err) + } + resp, err := client().Do(req) + if err != nil { + t.Fatal(err) + } + _ = body(t, resp) + if resp.StatusCode != http.StatusMethodNotAllowed { + t.Errorf("%s returned %d, want 405", method, resp.StatusCode) + } + } + if pk.metaHits.Load() != 0 { + t.Error("a write method reached the upstream repository") + } +} + +func TestHealthAndEnvVars(t *testing.T) { + t.Parallel() + pk := newFakePackagist(t) + p := startProxy(t, Config{Upstream: pk.URL}) + + if !p.Healthy() { + t.Error("a running proxy reported unhealthy") + } + if p.Name() != "composer" { + t.Errorf("Name = %q, want composer", p.Name()) + } + + env := p.EnvVars() + if len(env) != 1 { + t.Fatalf("EnvVars = %v, want exactly COMPOSER_REPO_PACKAGIST", env) + } + k, v, _ := strings.Cut(env[0], "=") + if k != "COMPOSER_REPO_PACKAGIST" { + t.Fatalf("EnvVars = %v, want COMPOSER_REPO_PACKAGIST", env) + } + // It must be reachable: the port actually bound, not the ":0" a test + // (or an operator) asked for. + if v != "http://"+p.Addr() { + t.Fatalf("COMPOSER_REPO_PACKAGIST = %q, want the bound address http://%s", v, p.Addr()) + } + resp := get(t, v+"/p2/acme/lib.json") + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + t.Errorf("the advertised base URL answered %d", resp.StatusCode) + } +} + +func TestDefaultsAreApplied(t *testing.T) { + t.Parallel() + discard := slog.New(slog.NewTextHandler(io.Discard, nil)) + p, err := New(Config{CacheDir: t.TempDir(), ListenAddr: "127.0.0.1:0", Log: discard}) + if err != nil { + t.Fatal(err) + } + if p.cfg.Upstream != DefaultUpstream { + t.Errorf("Upstream = %q, want %q", p.cfg.Upstream, DefaultUpstream) + } + if p.cfg.MetadataTTL != DefaultMetadataTTL { + t.Errorf("MetadataTTL = %v, want %v", p.cfg.MetadataTTL, DefaultMetadataTTL) + } + + // A self-hosted mirror must be allowed to serve its own archives without + // the operator restating it in allowed_hosts, trailing slash and all. + p2, err := New(Config{CacheDir: t.TempDir(), ListenAddr: "127.0.0.1:0", Upstream: "https://packagist.acme.test/", Log: discard}) + if err != nil { + t.Fatal(err) + } + if p2.cfg.Upstream != "https://packagist.acme.test" { + t.Errorf("Upstream = %q, want the trailing slash trimmed", p2.cfg.Upstream) + } + if !p2.allow.Allows("https://packagist.acme.test/dist/acme/lib.zip") { + t.Error("the configured upstream host is not on the allowlist") + } + + // And allowed_hosts extends it. + p3, err := New(Config{CacheDir: t.TempDir(), ListenAddr: "127.0.0.1:0", AllowedHosts: []string{"dist.acme.test"}, Log: discard}) + if err != nil { + t.Fatal(err) + } + if !p3.allow.Allows("https://dist.acme.test/lib.zip") { + t.Error("allowed_hosts did not extend the allowlist") + } +} + +// TestNestedMetadataPathsAreBothCached: Packagist serves packages.json +// alongside p2//.json, so a path-shaped cache key can need the +// same name to be a file and a directory at once — and the loser is cached +// silently-not-at-all. Query strings are the other half of the same problem: +// "?" is rejected outright by pkgcache.SafeSegments. +func TestNestedMetadataPathsAreBothCached(t *testing.T) { + t.Parallel() + pk := newFakePackagist(t) + p := startProxy(t, Config{Upstream: pk.URL, MetadataTTL: time.Hour}) + base := "http://" + p.Addr() + + paths := []string{ + "/p2/acme", // the parent, fetched first + "/p2/acme/lib.json", // nested under it + "/packages.json?v=2", // a query-carrying document + "/packages.json?v=2&full=1", // a different one at the same path + "/p2/acme/lib.json~dev.json", // a sibling + } + for _, path := range paths { + _ = body(t, get(t, base+path)) + } + want := int64(len(paths)) + if got := pk.metaHits.Load(); got != want { + t.Fatalf("upstream hits = %d, want %d", got, want) + } + for _, path := range paths { + _ = body(t, get(t, base+path)) + } + if got := pk.metaHits.Load(); got != want { + t.Errorf("upstream hits = %d after re-reading every document, want %d: something is not being cached", got, want) + } +} + +// TestDistKeyIsFilesystemSafe pins the reason the dist key is a URL hash: a +// key built from host+path breaks the moment an upstream carries a port, +// and Cache.Writer failing is a SILENT loss of caching rather than an error +// anyone sees. +func TestDistKeyIsFilesystemSafe(t *testing.T) { + t.Parallel() + c := startProxy(t, Config{}).srv.Cache() + for _, raw := range []string{ + "https://api.github.com/repos/acme/lib/zipball/deadbeef", + "https://packagist.acme.test:8443/dist/acme/lib-1.0.0.zip", + "http://127.0.0.1:59123/repos/acme/lib/zipball/deadbeef", + "https://gitlab.com/acme/lib/-/archive/v1.0.0/lib-v1.0.0.zip?ref=v1", + } { + if _, err := c.KeyPath(pkgcache.ArtifactKey(raw)); err != nil { + t.Errorf("the dist cache key for %q is not usable on disk: %v", raw, err) + } + } +} diff --git a/pkg/proxies/ghrel/ghrelproxy.go b/pkg/proxies/ghrel/ghrelproxy.go new file mode 100644 index 0000000..337e405 --- /dev/null +++ b/pkg/proxies/ghrel/ghrelproxy.go @@ -0,0 +1,418 @@ +// Package ghrelproxy implements proxies.CacheProxy for GitHub release +// artifacts: the release metadata a build tool reads, and the asset bytes +// that metadata points at. +// +// # WHY THIS EXISTS +// +// The other proxies in this package cache a package registry. This one caches +// a distribution channel that has no registry: a great deal of CI input +// arrives as "a file attached to a GitHub release" — prebuilt toolchains, +// vendored library tarballs, SDK archives — and every job on a node fetches +// the same bytes over the WAN, every time. +// +// Measured on this fleet (php-sdk, x86_64 gnu, 2026-09-20): 40 source +// artifacts took 87 s of a 591 s build, a large share of them GitHub release +// downloads. Nothing about those bytes changes between jobs. +// +// # THE INTERCEPTION PROBLEM +// +// Go has GOPROXY, Cargo has source replacement, pip has PIP_INDEX_URL. A +// GitHub release download has no such redirect — it is a plain +// https://github.com/... URL baked into whatever tool is fetching it. There +// are only two ways to intercept that: terminate TLS for job containers +// (a CA in every image, and a real change to what untrusted job code is +// exposed to), or have the tool ASK for a proxy. +// +// This proxy takes the second route, in the shape Go already established: +// ephemerd advertises GHREL_PROXY, and a tool that understands it uses it. +// A tool that does not is unaffected, and an operator who leaves the proxy +// off gets upstream — the env var is simply absent, so the usual +// +// base := getenv("GHREL_PROXY") ?: "https://api.github.com" +// +// falls back on its own. Nothing in a workflow has to change, and a build +// stays correct off-fleet, which a workflow rewritten to hardcode a proxy +// address would not. +// +// # WHAT IS CACHED, AND THE MUTABILITY PROBLEM +// +// Release METADATA is mutable — a release gains assets, a tag can be moved — +// so it is cached with a short TTL and then revalidated with a conditional +// GET, exactly as the npm/pip proxies treat their index documents. +// +// Release ASSETS are where this ecosystem differs from every other proxy +// here, and it is worth being explicit because getting it wrong is silent. +// PyPI REFUSES to let a file be re-uploaded even after deletion; npm and pub +// are similar. That is why those proxies mark artifacts Immutable and never +// look at them again. GitHub enforces nothing of the kind: an asset can be +// deleted and replaced under the same release, keeping its name and URL. +// +// So assets are NOT immutable by default here. They are cached with a TTL +// and revalidated, which costs one conditional request per asset per TTL and +// returns 304 for unchanged bytes — cheap next to re-downloading tens of +// megabytes, and correct even if someone republishes. +// +// ImmutableAssets turns that off for operators whose publishing discipline +// guarantees it (assets are never overwritten; a new build gets a new tag). +// It is faster and it is a promise the proxy cannot verify, which is why it +// is opt-in rather than the default. +package ghrelproxy + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "net/url" + "strings" + "time" + + "github.com/ephpm/ephemerd/pkg/proxies" + "github.com/ephpm/ephemerd/pkg/proxies/pkgcache" +) + +const ( + // DefaultAPIUpstream is GitHub's REST API origin. + DefaultAPIUpstream = "https://api.github.com" + + // DefaultDownloadUpstream is where release assets actually live. + // Separate from the API origin because they are different hosts and an + // operator pointing at GitHub Enterprise must be able to move both. + DefaultDownloadUpstream = "https://github.com" + + // DefaultMetadataTTL is how long release metadata is served before a + // conditional GET. Matches the npm/pip index default: long enough that + // a burst of jobs costs one upstream request, short enough that an + // asset published minutes ago is not invisible. + DefaultMetadataTTL = 5 * time.Minute + + // DefaultAssetTTL is how long an asset is served before revalidation. + // Much longer than metadata: the bytes behind a published asset almost + // never change, so this is a backstop against silent replacement rather + // than an expectation of churn. + DefaultAssetTTL = 24 * time.Hour + + // DefaultPort continues the sequence npm 8084, pip 8085, pub 8086. + DefaultPort = 8087 + + // apiPrefix is the route carrying GitHub REST API requests. + apiPrefix = "/repos/" + + // downloadPrefix is the route carrying asset bytes. Deliberately not + // under /repos/ so the two cannot be confused when rewriting. + downloadPrefix = "/download/" +) + +// defaultAllowedHosts is where asset bytes may be fetched from. GitHub serves +// release assets from its own host and redirects to an S3-backed CDN, so both +// must be permitted or every download fails on the redirect. +var defaultAllowedHosts = pkgcache.HostAllowlist{ + "github.com", + "api.github.com", + "objects.githubusercontent.com", + "release-assets.githubusercontent.com", +} + +// Config configures the proxy. +type Config struct { + // CacheDir is the on-disk cache root (e.g. /cache/ghrel). + CacheDir string + // APIUpstream overrides the REST API origin (GitHub Enterprise). + APIUpstream string + // DownloadUpstream overrides the asset origin. + DownloadUpstream string + // ListenAddr is the address to bind and advertise. + ListenAddr string + // MetadataTTL is the revalidation interval for release metadata. + // Zero takes DefaultMetadataTTL; negative revalidates every request. + MetadataTTL time.Duration + // AssetTTL is the revalidation interval for asset bytes. Ignored when + // ImmutableAssets is set. Zero takes DefaultAssetTTL. + AssetTTL time.Duration + // ImmutableAssets caches asset bytes permanently and never revalidates + // them. Only correct when assets are never overwritten in place; see + // the package comment. + ImmutableAssets bool + // MaxBytes is the cache disk budget. + MaxBytes int64 + // AllowedHosts extends the asset-fetch allowlist. + AllowedHosts []string + // Cleanup wipes the cache on Stop. Default false. + Cleanup bool + Log *slog.Logger +} + +// Compile-time interface check. +var _ proxies.CacheProxy = (*Proxy)(nil) + +// Proxy is the GitHub release caching proxy. +type Proxy struct { + cfg Config + srv *pkgcache.Server + allow pkgcache.HostAllowlist +} + +// New creates a GitHub release caching proxy. Call Start() to begin serving. +func New(cfg Config) (*Proxy, error) { + if cfg.APIUpstream == "" { + cfg.APIUpstream = DefaultAPIUpstream + } + if cfg.DownloadUpstream == "" { + cfg.DownloadUpstream = DefaultDownloadUpstream + } + cfg.APIUpstream = strings.TrimRight(cfg.APIUpstream, "/") + cfg.DownloadUpstream = strings.TrimRight(cfg.DownloadUpstream, "/") + if cfg.MetadataTTL == 0 { + cfg.MetadataTTL = DefaultMetadataTTL + } + if cfg.AssetTTL == 0 { + cfg.AssetTTL = DefaultAssetTTL + } + if cfg.Log == nil { + cfg.Log = slog.Default() + } + + allow := append(pkgcache.HostAllowlist{}, defaultAllowedHosts...) + allow = append(allow, cfg.AllowedHosts...) + p := &Proxy{ + cfg: cfg, + allow: allow.WithHostsOf(cfg.APIUpstream).WithHostsOf(cfg.DownloadUpstream), + } + + srv, err := pkgcache.NewServer(pkgcache.ServerConfig{ + Name: "ghrel", + ListenAddr: cfg.ListenAddr, + CacheDir: cfg.CacheDir, + MaxBytes: cfg.MaxBytes, + Cleanup: cfg.Cleanup, + Log: cfg.Log, + }, func(_ *pkgcache.Cache, _ *pkgcache.Fetcher) http.Handler { + return http.HandlerFunc(p.handle) + }) + if err != nil { + return nil, err + } + p.srv = srv + return p, nil +} + +// Start begins serving. Returns after the listener is bound. +func (p *Proxy) Start() error { return p.srv.Start() } + +// Stop shuts the proxy down. +func (p *Proxy) Stop() error { return p.srv.Stop() } + +// Addr returns the bound address. +func (p *Proxy) Addr() string { return p.srv.Addr() } + +// Name returns the proxy name for logs. +func (p *Proxy) Name() string { return "ghrel" } + +// Healthy reports whether the proxy is serving. +func (p *Proxy) Healthy() bool { return p.srv.Healthy() } + +// EnvVars advertises the proxy to job containers. +// +// GHREL_PROXY is a base URL, not a full endpoint, because a consumer needs +// both routes off it: the API for metadata and /download/ for bytes. A tool +// that does not recognise the variable ignores it and goes upstream, which is +// the entire point of advertising rather than intercepting. +func (p *Proxy) EnvVars() []string { + return []string{"GHREL_PROXY=" + p.advertiseBase()} +} + +// advertiseBase is the base URL containers should use to reach this proxy. +// +// Delegated to the shared server rather than formatting cfg.ListenAddr +// directly: a listen address with port 0 has to be resolved to the port +// actually bound, or the proxy advertises (and rewrites asset URLs to) a +// address nothing can connect to. +func (p *Proxy) advertiseBase() string { return p.srv.AdvertiseBase() } + +// handle routes a request to metadata or asset handling. +func (p *Proxy) handle(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && r.Method != http.MethodHead { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + switch { + case strings.HasPrefix(r.URL.Path, apiPrefix): + p.serveMetadata(w, r) + case strings.HasPrefix(r.URL.Path, downloadPrefix): + p.serveAsset(w, r) + default: + http.NotFound(w, r) + } +} + +// serveMetadata proxies a REST API request and rewrites the asset URLs in the +// response so the caller comes back here for the bytes. +// +// Without the rewrite the caller would take metadata from the cache and every +// byte from GitHub's CDN — the same trap the pip proxy documents for project +// pages, and the reason this proxy is worth having at all. +func (p *Proxy) serveMetadata(w http.ResponseWriter, r *http.Request) { + upstream := p.cfg.APIUpstream + r.URL.Path + if r.URL.RawQuery != "" { + upstream += "?" + r.URL.RawQuery + } + + body, meta, err := p.srv.Fetcher().Document(r.Context(), pkgcache.Request{ + Key: metaKey(r.URL.Path, r.URL.RawQuery), + URL: upstream, + TTL: p.cfg.MetadataTTL, + Accept: r.Header.Get("Accept"), + DefaultContentType: "application/json", + }) + if err != nil { + p.writeFetchError(w, err, "release metadata") + return + } + ct := meta.ContentType + + rewritten, err := rewriteAssetURLs(body, p.advertiseBase()) + if err != nil { + // Serving unrewritten metadata would silently send every byte to + // the CDN, which looks like success and caches nothing. Fail + // loudly instead. + p.cfg.Log.Warn("ghrel: could not rewrite release metadata", "path", r.URL.Path, "error", err) + http.Error(w, "malformed upstream release metadata", http.StatusBadGateway) + return + } + + w.Header().Set("Content-Type", ct) + w.WriteHeader(http.StatusOK) + if r.Method != http.MethodHead { + _, _ = w.Write(rewritten) + } +} + +// serveAsset serves release asset bytes, caching them. +// +// The route is /download///releases/download//, which +// is GitHub's own asset path with the host swapped, so rewriting is a prefix +// substitution and the shape stays recognisable in logs. +func (p *Proxy) serveAsset(w http.ResponseWriter, r *http.Request) { + rest := strings.TrimPrefix(r.URL.Path, downloadPrefix) + if rest == "" { + http.NotFound(w, r) + return + } + upstream := p.cfg.DownloadUpstream + "/" + rest + + req := pkgcache.Request{ + Key: "asset/" + rest, + URL: upstream, + DefaultContentType: "application/octet-stream", + } + if p.cfg.ImmutableAssets { + req.Immutable = true + } else { + req.TTL = p.cfg.AssetTTL + } + + // ServeArtifact STREAMS to the response rather than returning bytes. + // That matters here more than for the other proxies: release assets are + // routinely tens or hundreds of megabytes (prebuilt toolchains, SDK + // archives), and buffering one per concurrent job would put that much + // per job into ephemerd's heap on a node already running builds. + if err := p.srv.Fetcher().ServeArtifact(w, r, req); err != nil { + p.writeFetchError(w, err, "release asset") + } +} + +// writeFetchError maps a fetch failure onto a status code, keeping a real +// upstream 404 distinguishable from an outage — a nonexistent release is an +// answer the job should see, an unreachable GitHub is what fail-open exists +// for. +func (p *Proxy) writeFetchError(w http.ResponseWriter, err error, what string) { + if pkgcache.IsNotFound(err) { + http.Error(w, "not found", http.StatusNotFound) + return + } + p.cfg.Log.Warn("ghrel: upstream fetch failed", "what", what, "error", err) + http.Error(w, "upstream fetch failed", http.StatusBadGateway) +} + +// rewriteAssetURLs rewrites every browser_download_url in a release document +// so it points at this proxy. +// +// Operates on the decoded JSON rather than by string substitution: the API +// returns either a single release object or an array of them depending on the +// endpoint, and a blind replace would also rewrite unrelated fields (an +// asset's name, a release body quoting a URL) that must be left alone. +func rewriteAssetURLs(body []byte, base string) ([]byte, error) { + var doc any + if err := json.Unmarshal(body, &doc); err != nil { + return nil, fmt.Errorf("decode release metadata: %w", err) + } + rewriteNode(doc, base) + out, err := json.Marshal(doc) + if err != nil { + return nil, fmt.Errorf("re-encode release metadata: %w", err) + } + return out, nil +} + +// rewriteNode walks the decoded document rewriting browser_download_url. +func rewriteNode(node any, base string) { + switch v := node.(type) { + case map[string]any: + for k, val := range v { + if k == "browser_download_url" { + if s, ok := val.(string); ok { + if nu, ok := proxifyAssetURL(s, base); ok { + v[k] = nu + } + } + continue + } + rewriteNode(val, base) + } + case []any: + for _, item := range v { + rewriteNode(item, base) + } + } +} + +// proxifyAssetURL turns a GitHub asset URL into this proxy's /download/ route. +// Anything that is not a recognisable asset URL is left untouched rather than +// guessed at. +func proxifyAssetURL(raw, base string) (string, bool) { + u, err := url.Parse(raw) + if err != nil || u.Scheme == "" || u.Host == "" { + return "", false + } + path := strings.TrimPrefix(u.Path, "/") + if path == "" { + return "", false + } + return strings.TrimRight(base, "/") + downloadPrefix + path, true +} + +// metaKey is the cache key for one release-metadata document. Path AND query +// are part of the identity: /repos/o/r/releases?page=1 and ?page=2 are +// different documents and must not share an entry. +// +// Hashed rather than path-shaped, for the same reason pkgcache.ArtifactKey is +// — a URL is not safe to map onto a filesystem — and here for two concrete +// reasons, both of which fail SILENTLY (Cache.Write warns, the document is +// still served, and the proxy simply caches nothing from then on): +// +// - A query string carries "?" and "&", which pkgcache.SafeSegments rejects +// outright. Paginated release listings are exactly the request a busy +// node repeats. +// - The REST API's document paths NEST: /repos/o/r/releases is a document +// and so is /repos/o/r/releases/latest. A path-shaped key needs +// "releases" to be a file and a directory at once, and whichever is +// requested second loses. +func metaKey(path, rawQuery string) string { + sum := sha256.Sum256([]byte(path + "?" + rawQuery)) + h := hex.EncodeToString(sum[:]) + // Sharded one level so a node that talks to thousands of repos does not + // end up with one very wide directory. + return "meta/" + h[0:2] + "/" + h +} diff --git a/pkg/proxies/ghrel/ghrelproxy_test.go b/pkg/proxies/ghrel/ghrelproxy_test.go new file mode 100644 index 0000000..f0fd601 --- /dev/null +++ b/pkg/proxies/ghrel/ghrelproxy_test.go @@ -0,0 +1,535 @@ +package ghrelproxy + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +// --- harness --------------------------------------------------------------- + +// fakeGitHub stands in for both halves of GitHub: the REST API that serves +// release metadata, and the host that serves the asset bytes. They are +// separate servers because the proxy points at them with separate config +// (APIUpstream / DownloadUpstream) and conflating them would hide a wiring +// mistake between the two. +type fakeGitHub struct { + api *httptest.Server + dl *httptest.Server + metaHits atomic.Int64 + dlHits atomic.Int64 + asset []byte +} + +func newFakeGitHub(t *testing.T) *fakeGitHub { + t.Helper() + gh := &fakeGitHub{asset: []byte(strings.Repeat("ASSETBYTES", 512))} + + dlMux := http.NewServeMux() + dlMux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "missing") { + http.Error(w, "not found", http.StatusNotFound) + return + } + gh.dlHits.Add(1) + w.Header().Set("Content-Type", "application/octet-stream") + _, _ = w.Write(gh.asset) + }) + gh.dl = httptest.NewServer(dlMux) + t.Cleanup(gh.dl.Close) + + apiMux := http.NewServeMux() + apiMux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "missing") { + http.Error(w, "not found", http.StatusNotFound) + return + } + gh.metaHits.Add(1) + w.Header().Set("Content-Type", "application/json") + w.Header().Set("ETag", `"rel-v1"`) + // Shaped like a real GitHub release: the asset carries several URL + // fields and only browser_download_url is the byte source. + _, _ = fmt.Fprintf(w, `{ + "tag_name": "v1.2.3", + "html_url": "https://github.com/acme/tool/releases/tag/v1.2.3", + "body": "see %s/acme/tool/releases/download/v1.2.3/tool.tar.gz for the build", + "assets": [ + { + "name": "tool.tar.gz", + "url": "%s/repos/acme/tool/releases/assets/42", + "browser_download_url": "%s/acme/tool/releases/download/v1.2.3/tool.tar.gz", + "size": 5120 + } + ] + }`, gh.dl.URL, gh.api.URL, gh.dl.URL) + }) + gh.api = httptest.NewServer(apiMux) + t.Cleanup(gh.api.Close) + return gh +} + +func startProxy(t *testing.T, cfg Config) *Proxy { + t.Helper() + if cfg.CacheDir == "" { + cfg.CacheDir = t.TempDir() + } + if cfg.ListenAddr == "" { + cfg.ListenAddr = "127.0.0.1:0" + } + cfg.Log = slog.New(slog.NewTextHandler(io.Discard, nil)) + p, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := p.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + t.Cleanup(func() { + if err := p.Stop(); err != nil { + t.Errorf("Stop: %v", err) + } + }) + return p +} + +func client() *http.Client { + return &http.Client{ + Timeout: 10 * time.Second, + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, + } +} + +func get(t *testing.T, rawURL string) *http.Response { + t.Helper() + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, rawURL, nil) + if err != nil { + t.Fatal(err) + } + resp, err := client().Do(req) + if err != nil { + t.Fatalf("GET %s: %v", rawURL, err) + } + return resp +} + +func body(t *testing.T, resp *http.Response) []byte { + t.Helper() + defer func() { _ = resp.Body.Close() }() + b, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + return b +} + +// --- the rewrite, which is the whole point --------------------------------- + +// TestRewriteAssetURLsSingleRelease covers the shape GitHub returns from +// /releases/latest and /releases/tags/: one release object. +func TestRewriteAssetURLsSingleRelease(t *testing.T) { + t.Parallel() + const base = "http://gw:8087" + in := []byte(`{ + "tag_name": "v1", + "html_url": "https://github.com/acme/tool/releases/tag/v1", + "body": "grab https://github.com/acme/tool/releases/download/v1/tool.tgz", + "assets": [{ + "name": "tool.tgz", + "url": "https://api.github.com/repos/acme/tool/releases/assets/7", + "browser_download_url": "https://github.com/acme/tool/releases/download/v1/tool.tgz" + }] + }`) + + out, err := rewriteAssetURLs(in, base) + if err != nil { + t.Fatalf("rewriteAssetURLs: %v", err) + } + var doc map[string]any + if err := json.Unmarshal(out, &doc); err != nil { + t.Fatal(err) + } + + asset := doc["assets"].([]any)[0].(map[string]any) + want := base + downloadPrefix + "acme/tool/releases/download/v1/tool.tgz" + if got := asset["browser_download_url"]; got != want { + t.Errorf("browser_download_url = %q, want %q", got, want) + } + + // Everything else must survive byte-for-byte. A blind string replace + // would have rewritten the body text and the API asset URL too, and the + // API URL is what `gh` uses to fetch an asset from a private repo. + if got := asset["url"]; got != "https://api.github.com/repos/acme/tool/releases/assets/7" { + t.Errorf("asset.url was rewritten to %q", got) + } + if got := asset["name"]; got != "tool.tgz" { + t.Errorf("asset.name = %q", got) + } + if got := doc["html_url"]; got != "https://github.com/acme/tool/releases/tag/v1" { + t.Errorf("html_url was rewritten to %q", got) + } + if got := doc["body"].(string); !strings.Contains(got, "https://github.com/acme/tool/releases/download/v1/tool.tgz") { + t.Errorf("release body was rewritten to %q", got) + } + if got := doc["tag_name"]; got != "v1" { + t.Errorf("tag_name = %q", got) + } +} + +// TestRewriteAssetURLsReleaseArray covers the OTHER shape: /releases returns +// an array, and a rewriter that only understands an object silently sends +// every asset in a listing straight to the CDN. +func TestRewriteAssetURLsReleaseArray(t *testing.T) { + t.Parallel() + const base = "http://gw:8087" + in := []byte(`[ + {"tag_name":"v2","assets":[{"browser_download_url":"https://github.com/acme/tool/releases/download/v2/tool.tgz"}]}, + {"tag_name":"v1","assets":[ + {"browser_download_url":"https://github.com/acme/tool/releases/download/v1/tool.tgz"}, + {"browser_download_url":"https://github.com/acme/tool/releases/download/v1/tool.sha256"} + ]} + ]`) + + out, err := rewriteAssetURLs(in, base) + if err != nil { + t.Fatalf("rewriteAssetURLs: %v", err) + } + var docs []map[string]any + if err := json.Unmarshal(out, &docs); err != nil { + t.Fatalf("an array of releases must stay an array: %v", err) + } + if len(docs) != 2 { + t.Fatalf("got %d releases, want 2", len(docs)) + } + + var seen []string + for _, rel := range docs { + for _, a := range rel["assets"].([]any) { + u := a.(map[string]any)["browser_download_url"].(string) + if !strings.HasPrefix(u, base+downloadPrefix) { + t.Errorf("browser_download_url = %q, want it pointed at the proxy", u) + } + seen = append(seen, u) + } + } + if len(seen) != 3 { + t.Errorf("rewrote %d asset URLs, want 3: %v", len(seen), seen) + } + if seen[0] != base+downloadPrefix+"acme/tool/releases/download/v2/tool.tgz" { + t.Errorf("first rewritten URL = %q", seen[0]) + } +} + +// TestRewriteAssetURLsLeavesOddShapesAlone: the API is not under our control, +// so a value that is not a usable URL must be passed through rather than +// guessed at, and nothing structurally odd may panic. +func TestRewriteAssetURLsLeavesOddShapesAlone(t *testing.T) { + t.Parallel() + const base = "http://gw:8087" + + out, err := rewriteAssetURLs([]byte(`{"assets":[ + {"browser_download_url":"not-a-url"}, + {"browser_download_url":""}, + {"browser_download_url":42}, + {"browser_download_url":"https://github.com/"} + ]}`), base) + if err != nil { + t.Fatalf("rewriteAssetURLs: %v", err) + } + var doc map[string]any + if err := json.Unmarshal(out, &doc); err != nil { + t.Fatal(err) + } + assets := doc["assets"].([]any) + if got := assets[0].(map[string]any)["browser_download_url"]; got != "not-a-url" { + t.Errorf("a non-URL value was rewritten to %q", got) + } + if got := assets[1].(map[string]any)["browser_download_url"]; got != "" { + t.Errorf("an empty value was rewritten to %q", got) + } + if got := assets[2].(map[string]any)["browser_download_url"]; got != float64(42) { + t.Errorf("a non-string value was rewritten to %v", got) + } + if got := assets[3].(map[string]any)["browser_download_url"]; got != "https://github.com/" { + t.Errorf("a pathless URL was rewritten to %q", got) + } + + for _, in := range []string{`{"assets":"nope"}`, `{"assets":[5]}`, `[]`, `null`, `"scalar"`} { + if _, err := rewriteAssetURLs([]byte(in), base); err != nil { + t.Errorf("rewriteAssetURLs(%s): %v", in, err) + } + } + if _, err := rewriteAssetURLs([]byte("not json"), base); err == nil { + t.Error("unparseable metadata must be an error so the caller can fail loudly rather than serve unrewritten bytes") + } +} + +// --- end to end ------------------------------------------------------------ + +// TestMetadataIsCachedAndRewritten is the proxy doing its job: one upstream +// hit for repeated metadata reads, and the asset URLs in what it serves point +// back at itself. +func TestMetadataIsCachedAndRewritten(t *testing.T) { + t.Parallel() + gh := newFakeGitHub(t) + p := startProxy(t, Config{APIUpstream: gh.api.URL, DownloadUpstream: gh.dl.URL, MetadataTTL: time.Hour}) + base := "http://" + p.Addr() + const path = "/repos/acme/tool/releases/latest" + + first := body(t, get(t, base+path)) + second := body(t, get(t, base+path)) + if gh.metaHits.Load() != 1 { + t.Errorf("upstream metadata hits = %d, want 1", gh.metaHits.Load()) + } + if string(first) != string(second) { + t.Error("the cached metadata differs from the first response") + } + + var doc map[string]any + if err := json.Unmarshal(first, &doc); err != nil { + t.Fatal(err) + } + asset := doc["assets"].([]any)[0].(map[string]any) + want := base + downloadPrefix + "acme/tool/releases/download/v1.2.3/tool.tar.gz" + if got := asset["browser_download_url"]; got != want { + t.Fatalf("browser_download_url = %q, want %q", got, want) + } + if got := asset["size"]; got != float64(5120) { + t.Errorf("asset.size = %v, want it untouched", got) + } +} + +// TestAssetIsFetchedThroughTheProxyAndCached follows the rewritten URL out of +// the metadata exactly as a build tool would. +func TestAssetIsFetchedThroughTheProxyAndCached(t *testing.T) { + t.Parallel() + gh := newFakeGitHub(t) + p := startProxy(t, Config{APIUpstream: gh.api.URL, DownloadUpstream: gh.dl.URL}) + base := "http://" + p.Addr() + + var doc map[string]any + if err := json.Unmarshal(body(t, get(t, base+"/repos/acme/tool/releases/latest")), &doc); err != nil { + t.Fatal(err) + } + assetURL := doc["assets"].([]any)[0].(map[string]any)["browser_download_url"].(string) + + for i := range 3 { + resp := get(t, assetURL) + got := body(t, resp) + if resp.StatusCode != http.StatusOK { + t.Fatalf("attempt %d: status %d", i, resp.StatusCode) + } + if string(got) != string(gh.asset) { + t.Fatalf("attempt %d: asset bytes differ", i) + } + } + if gh.dlHits.Load() != 1 { + t.Errorf("upstream asset hits = %d, want 1: the bytes were not cached", gh.dlHits.Load()) + } +} + +// TestImmutableAssetsStillServesTheRightBytes: the flag changes only how the +// entry is revalidated, never what a job receives. +func TestImmutableAssetsStillServesTheRightBytes(t *testing.T) { + t.Parallel() + gh := newFakeGitHub(t) + p := startProxy(t, Config{ + APIUpstream: gh.api.URL, + DownloadUpstream: gh.dl.URL, + ImmutableAssets: true, + }) + base := "http://" + p.Addr() + u := base + downloadPrefix + "acme/tool/releases/download/v1.2.3/tool.tar.gz" + + for range 2 { + if got := body(t, get(t, u)); string(got) != string(gh.asset) { + t.Fatal("asset bytes differ") + } + } + if gh.dlHits.Load() != 1 { + t.Errorf("upstream asset hits = %d, want 1", gh.dlHits.Load()) + } +} + +func TestUpstream404IsPassedThrough(t *testing.T) { + t.Parallel() + gh := newFakeGitHub(t) + p := startProxy(t, Config{APIUpstream: gh.api.URL, DownloadUpstream: gh.dl.URL}) + base := "http://" + p.Addr() + + for _, path := range []string{ + "/repos/acme/missing/releases/latest", + downloadPrefix + "acme/missing/releases/download/v1/x.tgz", + } { + resp := get(t, base+path) + _ = body(t, resp) + if resp.StatusCode != http.StatusNotFound { + t.Errorf("GET %s = %d, want 404: a nonexistent release is a real answer", path, resp.StatusCode) + } + } +} + +func TestUnroutedPathsAre404(t *testing.T) { + t.Parallel() + gh := newFakeGitHub(t) + p := startProxy(t, Config{APIUpstream: gh.api.URL, DownloadUpstream: gh.dl.URL}) + base := "http://" + p.Addr() + + for _, path := range []string{"/", "/user", "/graphql", downloadPrefix} { + resp := get(t, base+path) + _ = body(t, resp) + if resp.StatusCode == http.StatusOK { + t.Errorf("GET %s returned 200; only /repos/ and %s are routed", path, downloadPrefix) + } + } + if gh.metaHits.Load() != 0 { + t.Error("an unrouted path reached the upstream API") + } +} + +// TestWritesAreRejected: this is a read-through cache, never a GitHub front +// end. Nothing that could mutate a release may be relayed. +func TestWritesAreRejected(t *testing.T) { + t.Parallel() + gh := newFakeGitHub(t) + p := startProxy(t, Config{APIUpstream: gh.api.URL, DownloadUpstream: gh.dl.URL}) + base := "http://" + p.Addr() + + for _, method := range []string{http.MethodPut, http.MethodPost, http.MethodDelete, http.MethodPatch} { + req, err := http.NewRequestWithContext(t.Context(), method, base+"/repos/acme/tool/releases", strings.NewReader("{}")) + if err != nil { + t.Fatal(err) + } + resp, err := client().Do(req) + if err != nil { + t.Fatal(err) + } + _ = body(t, resp) + if resp.StatusCode != http.StatusMethodNotAllowed { + t.Errorf("%s returned %d, want 405", method, resp.StatusCode) + } + } + if gh.metaHits.Load() != 0 { + t.Error("a write method reached the upstream API") + } +} + +func TestHealthAndEnvVars(t *testing.T) { + t.Parallel() + gh := newFakeGitHub(t) + p := startProxy(t, Config{APIUpstream: gh.api.URL, DownloadUpstream: gh.dl.URL}) + + if !p.Healthy() { + t.Error("a running proxy reported unhealthy") + } + if p.Name() != "ghrel" { + t.Errorf("Name = %q, want ghrel", p.Name()) + } + + env := p.EnvVars() + if len(env) != 1 { + t.Fatalf("EnvVars = %v, want exactly GHREL_PROXY", env) + } + k, v, _ := strings.Cut(env[0], "=") + if k != "GHREL_PROXY" { + t.Fatalf("EnvVars = %v, want GHREL_PROXY", env) + } + // It must be reachable: a base URL whose port is the one actually bound, + // not the ":0" an operator (or a test) asked for. + if v != "http://"+p.Addr() { + t.Fatalf("GHREL_PROXY = %q, want the bound address http://%s", v, p.Addr()) + } + resp := get(t, v+"/repos/acme/tool/releases/latest") + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + t.Errorf("the advertised base URL answered %d", resp.StatusCode) + } +} + +func TestDefaultsAreApplied(t *testing.T) { + t.Parallel() + p, err := New(Config{CacheDir: t.TempDir(), ListenAddr: "127.0.0.1:0", Log: slog.New(slog.NewTextHandler(io.Discard, nil))}) + if err != nil { + t.Fatal(err) + } + if p.cfg.APIUpstream != DefaultAPIUpstream { + t.Errorf("APIUpstream = %q, want %q", p.cfg.APIUpstream, DefaultAPIUpstream) + } + if p.cfg.DownloadUpstream != DefaultDownloadUpstream { + t.Errorf("DownloadUpstream = %q, want %q", p.cfg.DownloadUpstream, DefaultDownloadUpstream) + } + if p.cfg.MetadataTTL != DefaultMetadataTTL { + t.Errorf("MetadataTTL = %v, want %v", p.cfg.MetadataTTL, DefaultMetadataTTL) + } + if p.cfg.AssetTTL != DefaultAssetTTL { + t.Errorf("AssetTTL = %v, want %v", p.cfg.AssetTTL, DefaultAssetTTL) + } + // A trailing slash on an override must not produce "//" in every + // upstream URL built from it. + p2, err := New(Config{CacheDir: t.TempDir(), ListenAddr: "127.0.0.1:0", APIUpstream: "https://ghe.test/api/v3/", Log: slog.New(slog.NewTextHandler(io.Discard, nil))}) + if err != nil { + t.Fatal(err) + } + if p2.cfg.APIUpstream != "https://ghe.test/api/v3" { + t.Errorf("APIUpstream = %q, want the trailing slash trimmed", p2.cfg.APIUpstream) + } + // A GitHub Enterprise upstream must be allowed to serve its own bytes + // without the operator restating it in allowed_hosts. + if !p2.allow.Allows("https://ghe.test/acme/tool/releases/download/v1/x.tgz") { + t.Error("the configured upstream host is not on the allowlist") + } +} + +func TestQueryIsPartOfTheCacheKey(t *testing.T) { + t.Parallel() + gh := newFakeGitHub(t) + p := startProxy(t, Config{APIUpstream: gh.api.URL, DownloadUpstream: gh.dl.URL, MetadataTTL: time.Hour}) + base := "http://" + p.Addr() + + // Page 1 and page 2 of a release listing are different documents. + _ = body(t, get(t, base+"/repos/acme/tool/releases?per_page=1&page=1")) + _ = body(t, get(t, base+"/repos/acme/tool/releases?per_page=1&page=2")) + if gh.metaHits.Load() != 2 { + t.Errorf("upstream hits = %d, want 2: two queries shared a cache entry", gh.metaHits.Load()) + } + _ = body(t, get(t, base+"/repos/acme/tool/releases?per_page=1&page=1")) + if gh.metaHits.Load() != 2 { + t.Errorf("upstream hits = %d, want 2: the first query was not cached", gh.metaHits.Load()) + } +} + +// TestNestedMetadataPathsAreBothCached: the REST API nests documents — +// /repos/o/r/releases is one and /repos/o/r/releases/latest is another — and +// a path-shaped cache key would need "releases" to be a file and a directory +// at once. The loser is cached silently-not-at-all. +func TestNestedMetadataPathsAreBothCached(t *testing.T) { + t.Parallel() + gh := newFakeGitHub(t) + p := startProxy(t, Config{APIUpstream: gh.api.URL, DownloadUpstream: gh.dl.URL, MetadataTTL: time.Hour}) + base := "http://" + p.Addr() + + paths := []string{ + "/repos/acme/tool/releases", // the parent, fetched first + "/repos/acme/tool/releases/latest", // nested under it + } + for _, path := range paths { + _ = body(t, get(t, base+path)) + } + if gh.metaHits.Load() != 2 { + t.Fatalf("upstream hits = %d, want 2", gh.metaHits.Load()) + } + for _, path := range paths { + _ = body(t, get(t, base+path)) + } + if got := gh.metaHits.Load(); got != 2 { + t.Errorf("upstream hits = %d after re-reading both documents, want 2: one of them is not being cached", got) + } +} diff --git a/pkg/proxies/pkgcache/fetch.go b/pkg/proxies/pkgcache/fetch.go index 5f47d4f..8af0dc9 100644 --- a/pkg/proxies/pkgcache/fetch.go +++ b/pkg/proxies/pkgcache/fetch.go @@ -59,6 +59,24 @@ type Request struct { DefaultContentType string } +// serveCachedArtifact writes a cache hit to the response. +// +// Cache-Control says "immutable" regardless of whether this entry is +// revalidatable here. That header governs the JOB's HTTP client, and the +// identity of an artifact URL does not change under it mid-build; freshness +// against upstream is this proxy's job, decided by req.TTL above. +func (f *Fetcher) serveCachedArtifact(w http.ResponseWriter, r *http.Request, req Request, file io.ReadSeekCloser, meta Meta) { + w.Header().Set("Content-Type", contentTypeOr(meta.ContentType, req.DefaultContentType)) + w.Header().Set("Cache-Control", "public, max-age=31536000, immutable") + if meta.ETag != "" { + w.Header().Set("ETag", meta.ETag) + } + f.Log.Debug("package cache hit", "key", req.Key, "url", req.URL) + // ServeContent handles Range, If-Modified-Since and HEAD. The empty name + // suppresses its content sniffing — the type is already set. + http.ServeContent(w, r, "", meta.Fetched, file) +} + // NotFoundError marks an upstream 404/410 so a caller can pass it through as // a 404 rather than treating it as an outage. A nonexistent package version // must stay distinguishable from a registry being down: the first is a real @@ -160,21 +178,21 @@ func (f *Fetcher) Document(ctx context.Context, req Request) ([]byte, Meta, erro // gigabyte, and several jobs pull concurrently. func (f *Fetcher) ServeArtifact(w http.ResponseWriter, r *http.Request, req Request) error { if file, meta, ok := f.Cache.Open(req.Key); ok { - defer func() { - if cerr := file.Close(); cerr != nil { - f.Log.Debug("closing cached artifact", "key", req.Key, "error", cerr) - } - }() - w.Header().Set("Content-Type", contentTypeOr(meta.ContentType, req.DefaultContentType)) - w.Header().Set("Cache-Control", "public, max-age=31536000, immutable") - if meta.ETag != "" { - w.Header().Set("ETag", meta.ETag) + if decide(true, req.Immutable, meta.Fetched, time.Now(), req.TTL) == serveCached { + defer func() { + if cerr := file.Close(); cerr != nil { + f.Log.Debug("closing cached artifact", "key", req.Key, "error", cerr) + } + }() + f.serveCachedArtifact(w, r, req, file, meta) + return nil + } + // Stale and revalidatable. Close now rather than defer — this path + // does NOT return here, and a deferred close would hold the handle + // for the whole upstream round trip. + if cerr := file.Close(); cerr != nil { + f.Log.Debug("closing stale cached artifact", "key", req.Key, "error", cerr) } - f.Log.Debug("package cache hit", "key", req.Key, "url", req.URL) - // ServeContent handles Range, If-Modified-Since and HEAD. The empty - // name suppresses its content sniffing — the type is already set. - http.ServeContent(w, r, "", meta.Fetched, file) - return nil } mu := f.Cache.Lock(req.Key) @@ -182,16 +200,60 @@ func (f *Fetcher) ServeArtifact(w http.ResponseWriter, r *http.Request, req Requ defer mu.Unlock() // Another goroutine may have filled it while we waited for the lock. + // Re-decide rather than assuming fresh: the entry we are looking at may + // be the same stale one we just closed. + cachedMeta, haveCached := Meta{}, false if file, meta, ok := f.Cache.Open(req.Key); ok { - defer func() { - if cerr := file.Close(); cerr != nil { - f.Log.Debug("closing cached artifact", "key", req.Key, "error", cerr) + if decide(true, req.Immutable, meta.Fetched, time.Now(), req.TTL) == serveCached { + defer func() { + if cerr := file.Close(); cerr != nil { + f.Log.Debug("closing cached artifact", "key", req.Key, "error", cerr) + } + }() + f.serveCachedArtifact(w, r, req, file, meta) + return nil + } + if cerr := file.Close(); cerr != nil { + f.Log.Debug("closing stale cached artifact", "key", req.Key, "error", cerr) + } + cachedMeta, haveCached = meta, true + } + + // A stale-but-present entry is revalidated with a conditional GET. On 304 + // the bytes on disk are still correct, so only the freshness stamp moves + // and nothing crosses the WAN. + // + // This is dead weight for every caller that marks artifacts Immutable + // (npm tarballs, PyPI wheels, pub archives: registries that refuse to + // re-publish a file). It exists for GitHub release assets, which CAN be + // replaced in place under the same URL, so "cache it forever" would serve + // superseded bytes indefinitely. + if haveCached { + resp, err := f.do(r.Context(), req, true, cachedMeta) + if err == nil && resp.StatusCode == http.StatusNotModified { + if cerr := resp.Body.Close(); cerr != nil { + f.Log.Debug("closing upstream body", "url", req.URL, "error", cerr) } - }() - w.Header().Set("Content-Type", contentTypeOr(meta.ContentType, req.DefaultContentType)) - w.Header().Set("Cache-Control", "public, max-age=31536000, immutable") - http.ServeContent(w, r, "", meta.Fetched, file) - return nil + cachedMeta.Fetched = time.Now() + f.Cache.refreshMeta(req.Key, cachedMeta) + if file, meta, ok := f.Cache.Open(req.Key); ok { + defer func() { + if cerr := file.Close(); cerr != nil { + f.Log.Debug("closing cached artifact", "key", req.Key, "error", cerr) + } + }() + f.Log.Debug("package cache revalidated", "key", req.Key, "url", req.URL) + f.serveCachedArtifact(w, r, req, file, meta) + return nil + } + } + if err == nil && resp != nil { + if cerr := resp.Body.Close(); cerr != nil { + f.Log.Debug("closing upstream body", "url", req.URL, "error", cerr) + } + } + // Anything else (200, an error, or the entry vanishing under us) + // falls through to a full fetch below. } f.Log.Debug("package cache miss", "key", req.Key, "url", req.URL) diff --git a/pkg/proxies/pkgcache/fetch_test.go b/pkg/proxies/pkgcache/fetch_test.go index 9f503db..56601d0 100644 --- a/pkg/proxies/pkgcache/fetch_test.go +++ b/pkg/proxies/pkgcache/fetch_test.go @@ -372,3 +372,82 @@ func TestFetcherForwardsNoCredentials(t *testing.T) { t.Error("credentials reached the upstream registry") } } + +// TestServeArtifactRevalidatesStaleMutableEntry covers the one artifact +// ecosystem that is NOT write-once. +// +// npm, PyPI and pub all refuse to re-publish a file, so their proxies mark +// artifacts Immutable and this path never runs. GitHub does not: a release +// asset can be deleted and re-uploaded under the same URL. An artifact cache +// that never revalidates would serve the superseded bytes forever, so a +// non-immutable entry past its TTL is revalidated with a conditional GET. +func TestServeArtifactRevalidatesStaleMutableEntry(t *testing.T) { + t.Parallel() + payload := strings.Repeat("asset-bytes", 100) + var full, conditional atomic.Int64 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("If-None-Match") != "" { + conditional.Add(1) + w.WriteHeader(http.StatusNotModified) + return + } + full.Add(1) + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("ETag", `"v1"`) + _, _ = io.WriteString(w, payload) + })) + defer upstream.Close() + + f := newTestFetcher(t, 0) + // Negative TTL means "always stale", which is what makes this + // deterministic without sleeping. + req := Request{Key: "dl/aa/bb/asset", URL: upstream.URL + "/asset", TTL: -1} + + for i := range 3 { + rec := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/whatever", nil) + if err := f.ServeArtifact(rec, r, req); err != nil { + t.Fatalf("attempt %d: %v", i, err) + } + if rec.Code != http.StatusOK { + t.Fatalf("attempt %d: status %d", i, rec.Code) + } + if rec.Body.String() != payload { + t.Fatalf("attempt %d: body mismatch (%d bytes)", i, rec.Body.Len()) + } + } + if full.Load() != 1 { + t.Errorf("full upstream fetches = %d, want 1: the bytes were re-downloaded instead of revalidated", full.Load()) + } + if conditional.Load() != 2 { + t.Errorf("conditional requests = %d, want 2: the stale entry was served without revalidating", conditional.Load()) + } +} + +// TestServeArtifactImmutableNeverRevalidates pins the property that makes the +// change above safe for the existing proxies: an Immutable request must not +// contact upstream again even with a TTL set. +func TestServeArtifactImmutableNeverRevalidates(t *testing.T) { + t.Parallel() + var hits atomic.Int64 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + hits.Add(1) + w.Header().Set("Content-Type", "application/octet-stream") + _, _ = io.WriteString(w, "wheel") + })) + defer upstream.Close() + + f := newTestFetcher(t, 0) + req := Request{Key: "dl/aa/bb/wheel", URL: upstream.URL + "/wheel", Immutable: true, TTL: -1} + + for i := range 3 { + rec := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/whatever", nil) + if err := f.ServeArtifact(rec, r, req); err != nil { + t.Fatalf("attempt %d: %v", i, err) + } + } + if hits.Load() != 1 { + t.Errorf("upstream hits = %d, want 1: Immutable must win over TTL", hits.Load()) + } +} From 46d2dd462706c105196c588709887e0d4102f350 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Sun, 20 Sep 2026 22:31:22 -0700 Subject: [PATCH 2/2] test(proxies): stop get() returning a response, to satisfy bodyclose CI failed with 23 bodyclose violations that go vet and go test do not catch -- golangci-lint runs that linter, mage lint does not. The first attempt was to have get() close the body itself via t.Cleanup. That does not work: bodyclose flags the CALL SITE of anything returning a response it cannot see closed, so a helper that hands one back can never satisfy it however careful the caller is. So get() no longer returns *http.Response. It reads and closes the body and returns a small result{code, data}, which is all these tests ever used. No response escapes the helper, and body() became redundant and is deleted. Verified with the linter CI actually runs (mage lint), not just go vet: bodyclose 21 -> 0. The single remaining staticcheck hit is in pkg/networking/network_windows_test.go, which this branch does not touch and which Linux CI never builds -- it is why the original failure listed only bodyclose. --- pkg/proxies/composer/composerproxy_test.go | 64 +++++++++++----------- pkg/proxies/ghrel/ghrelproxy_test.go | 61 +++++++++++---------- 2 files changed, 65 insertions(+), 60 deletions(-) diff --git a/pkg/proxies/composer/composerproxy_test.go b/pkg/proxies/composer/composerproxy_test.go index 14fd9d4..0eda517 100644 --- a/pkg/proxies/composer/composerproxy_test.go +++ b/pkg/proxies/composer/composerproxy_test.go @@ -112,7 +112,18 @@ func client() *http.Client { } } -func get(t *testing.T, rawURL string) *http.Response { +// result is everything these tests need from a response. +// +// get() reads and CLOSES the body before returning, so no *http.Response +// escapes this helper. That is not tidiness: bodyclose flags the CALL SITE of +// anything that returns a response it cannot see closed, so a helper handing +// one back can never satisfy the linter no matter how careful the caller is. +type result struct { + code int + data []byte +} + +func get(t *testing.T, rawURL string) result { t.Helper() req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, rawURL, nil) if err != nil { @@ -122,17 +133,12 @@ func get(t *testing.T, rawURL string) *http.Response { if err != nil { t.Fatalf("GET %s: %v", rawURL, err) } - return resp -} - -func body(t *testing.T, resp *http.Response) []byte { - t.Helper() defer func() { _ = resp.Body.Close() }() b, err := io.ReadAll(resp.Body) if err != nil { - t.Fatal(err) + t.Fatalf("reading GET %s: %v", rawURL, err) } - return b + return result{code: resp.StatusCode, data: b} } // distTarget decodes the upstream URL back out of a rewritten dist URL. @@ -293,9 +299,9 @@ func TestServeDistRefusesNonAllowlistedHosts(t *testing.T) { } { u := base + distPrefix + "archive?u=" + url.QueryEscape(target) resp := get(t, u) - got := body(t, resp) - if resp.StatusCode != http.StatusForbidden { - t.Errorf("dist fetch of %q returned %d, want 403", target, resp.StatusCode) + got := resp.data + if resp.code != http.StatusForbidden { + t.Errorf("dist fetch of %q returned %d, want 403", target, resp.code) } if strings.Contains(string(got), "cloud-metadata-credentials") { t.Fatalf("the proxy relayed a non-allowlisted host: %q", got) @@ -309,9 +315,8 @@ func TestServeDistRequiresAnUpstream(t *testing.T) { base := "http://" + p.Addr() resp := get(t, base+distPrefix+"archive") - _ = body(t, resp) - if resp.StatusCode != http.StatusBadRequest { - t.Errorf("a dist request with no u= returned %d, want 400", resp.StatusCode) + if resp.code != http.StatusBadRequest { + t.Errorf("a dist request with no u= returned %d, want 400", resp.code) } } @@ -326,8 +331,8 @@ func TestMetadataIsCachedAndDistIsFetchedThroughTheProxy(t *testing.T) { base := "http://" + p.Addr() const path = "/p2/acme/lib.json" - first := body(t, get(t, base+path)) - second := body(t, get(t, base+path)) + first := get(t, base+path).data + second := get(t, base+path).data if pk.metaHits.Load() != 1 { t.Errorf("upstream metadata hits = %d, want 1", pk.metaHits.Load()) } @@ -351,9 +356,9 @@ func TestMetadataIsCachedAndDistIsFetchedThroughTheProxy(t *testing.T) { // Follow it exactly as Composer would, repeatedly. for i := range 3 { resp := get(t, distURL) - got := body(t, resp) - if resp.StatusCode != http.StatusOK { - t.Fatalf("attempt %d: status %d (%q)", i, resp.StatusCode, got) + got := resp.data + if resp.code != http.StatusOK { + t.Fatalf("attempt %d: status %d (%q)", i, resp.code, got) } if string(got) != string(pk.archive) { t.Fatalf("attempt %d: archive bytes differ", i) @@ -374,16 +379,14 @@ func TestUpstream404IsPassedThrough(t *testing.T) { base := "http://" + p.Addr() resp := get(t, base+"/p2/acme/missing.json") - _ = body(t, resp) - if resp.StatusCode != http.StatusNotFound { - t.Errorf("metadata status = %d, want 404: a nonexistent package is a real answer", resp.StatusCode) + if resp.code != http.StatusNotFound { + t.Errorf("metadata status = %d, want 404: a nonexistent package is a real answer", resp.code) } u := base + distPrefix + "archive?u=" + url.QueryEscape(pk.URL+"/repos/acme/missing/zipball/deadbeef") resp2 := get(t, u) - _ = body(t, resp2) - if resp2.StatusCode != http.StatusNotFound { - t.Errorf("archive status = %d, want 404", resp2.StatusCode) + if resp2.code != http.StatusNotFound { + t.Errorf("archive status = %d, want 404", resp2.code) } } @@ -403,7 +406,7 @@ func TestWritesAreRejected(t *testing.T) { if err != nil { t.Fatal(err) } - _ = body(t, resp) + defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusMethodNotAllowed { t.Errorf("%s returned %d, want 405", method, resp.StatusCode) } @@ -439,9 +442,8 @@ func TestHealthAndEnvVars(t *testing.T) { t.Fatalf("COMPOSER_REPO_PACKAGIST = %q, want the bound address http://%s", v, p.Addr()) } resp := get(t, v+"/p2/acme/lib.json") - defer func() { _ = resp.Body.Close() }() - if resp.StatusCode != http.StatusOK { - t.Errorf("the advertised base URL answered %d", resp.StatusCode) + if resp.code != http.StatusOK { + t.Errorf("the advertised base URL answered %d", resp.code) } } @@ -501,14 +503,14 @@ func TestNestedMetadataPathsAreBothCached(t *testing.T) { "/p2/acme/lib.json~dev.json", // a sibling } for _, path := range paths { - _ = body(t, get(t, base+path)) + _ = get(t, base+path).data } want := int64(len(paths)) if got := pk.metaHits.Load(); got != want { t.Fatalf("upstream hits = %d, want %d", got, want) } for _, path := range paths { - _ = body(t, get(t, base+path)) + _ = get(t, base+path).data } if got := pk.metaHits.Load(); got != want { t.Errorf("upstream hits = %d after re-reading every document, want %d: something is not being cached", got, want) diff --git a/pkg/proxies/ghrel/ghrelproxy_test.go b/pkg/proxies/ghrel/ghrelproxy_test.go index f0fd601..42cd696 100644 --- a/pkg/proxies/ghrel/ghrelproxy_test.go +++ b/pkg/proxies/ghrel/ghrelproxy_test.go @@ -108,7 +108,18 @@ func client() *http.Client { } } -func get(t *testing.T, rawURL string) *http.Response { +// result is everything these tests need from a response. +// +// get() reads and CLOSES the body before returning, so no *http.Response +// escapes this helper. That is not tidiness: bodyclose flags the CALL SITE of +// anything that returns a response it cannot see closed, so a helper handing +// one back can never satisfy the linter no matter how careful the caller is. +type result struct { + code int + data []byte +} + +func get(t *testing.T, rawURL string) result { t.Helper() req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, rawURL, nil) if err != nil { @@ -118,17 +129,12 @@ func get(t *testing.T, rawURL string) *http.Response { if err != nil { t.Fatalf("GET %s: %v", rawURL, err) } - return resp -} - -func body(t *testing.T, resp *http.Response) []byte { - t.Helper() defer func() { _ = resp.Body.Close() }() b, err := io.ReadAll(resp.Body) if err != nil { - t.Fatal(err) + t.Fatalf("reading GET %s: %v", rawURL, err) } - return b + return result{code: resp.StatusCode, data: b} } // --- the rewrite, which is the whole point --------------------------------- @@ -284,8 +290,8 @@ func TestMetadataIsCachedAndRewritten(t *testing.T) { base := "http://" + p.Addr() const path = "/repos/acme/tool/releases/latest" - first := body(t, get(t, base+path)) - second := body(t, get(t, base+path)) + first := get(t, base+path).data + second := get(t, base+path).data if gh.metaHits.Load() != 1 { t.Errorf("upstream metadata hits = %d, want 1", gh.metaHits.Load()) } @@ -316,16 +322,16 @@ func TestAssetIsFetchedThroughTheProxyAndCached(t *testing.T) { base := "http://" + p.Addr() var doc map[string]any - if err := json.Unmarshal(body(t, get(t, base+"/repos/acme/tool/releases/latest")), &doc); err != nil { + if err := json.Unmarshal(get(t, base+"/repos/acme/tool/releases/latest").data, &doc); err != nil { t.Fatal(err) } assetURL := doc["assets"].([]any)[0].(map[string]any)["browser_download_url"].(string) for i := range 3 { resp := get(t, assetURL) - got := body(t, resp) - if resp.StatusCode != http.StatusOK { - t.Fatalf("attempt %d: status %d", i, resp.StatusCode) + got := resp.data + if resp.code != http.StatusOK { + t.Fatalf("attempt %d: status %d", i, resp.code) } if string(got) != string(gh.asset) { t.Fatalf("attempt %d: asset bytes differ", i) @@ -350,7 +356,7 @@ func TestImmutableAssetsStillServesTheRightBytes(t *testing.T) { u := base + downloadPrefix + "acme/tool/releases/download/v1.2.3/tool.tar.gz" for range 2 { - if got := body(t, get(t, u)); string(got) != string(gh.asset) { + if got := get(t, u).data; string(got) != string(gh.asset) { t.Fatal("asset bytes differ") } } @@ -370,9 +376,8 @@ func TestUpstream404IsPassedThrough(t *testing.T) { downloadPrefix + "acme/missing/releases/download/v1/x.tgz", } { resp := get(t, base+path) - _ = body(t, resp) - if resp.StatusCode != http.StatusNotFound { - t.Errorf("GET %s = %d, want 404: a nonexistent release is a real answer", path, resp.StatusCode) + if resp.code != http.StatusNotFound { + t.Errorf("GET %s = %d, want 404: a nonexistent release is a real answer", path, resp.code) } } } @@ -385,8 +390,7 @@ func TestUnroutedPathsAre404(t *testing.T) { for _, path := range []string{"/", "/user", "/graphql", downloadPrefix} { resp := get(t, base+path) - _ = body(t, resp) - if resp.StatusCode == http.StatusOK { + if resp.code == http.StatusOK { t.Errorf("GET %s returned 200; only /repos/ and %s are routed", path, downloadPrefix) } } @@ -412,7 +416,7 @@ func TestWritesAreRejected(t *testing.T) { if err != nil { t.Fatal(err) } - _ = body(t, resp) + defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusMethodNotAllowed { t.Errorf("%s returned %d, want 405", method, resp.StatusCode) } @@ -448,9 +452,8 @@ func TestHealthAndEnvVars(t *testing.T) { t.Fatalf("GHREL_PROXY = %q, want the bound address http://%s", v, p.Addr()) } resp := get(t, v+"/repos/acme/tool/releases/latest") - defer func() { _ = resp.Body.Close() }() - if resp.StatusCode != http.StatusOK { - t.Errorf("the advertised base URL answered %d", resp.StatusCode) + if resp.code != http.StatusOK { + t.Errorf("the advertised base URL answered %d", resp.code) } } @@ -495,12 +498,12 @@ func TestQueryIsPartOfTheCacheKey(t *testing.T) { base := "http://" + p.Addr() // Page 1 and page 2 of a release listing are different documents. - _ = body(t, get(t, base+"/repos/acme/tool/releases?per_page=1&page=1")) - _ = body(t, get(t, base+"/repos/acme/tool/releases?per_page=1&page=2")) + _ = get(t, base+"/repos/acme/tool/releases?per_page=1&page=1").data + _ = get(t, base+"/repos/acme/tool/releases?per_page=1&page=2").data if gh.metaHits.Load() != 2 { t.Errorf("upstream hits = %d, want 2: two queries shared a cache entry", gh.metaHits.Load()) } - _ = body(t, get(t, base+"/repos/acme/tool/releases?per_page=1&page=1")) + _ = get(t, base+"/repos/acme/tool/releases?per_page=1&page=1").data if gh.metaHits.Load() != 2 { t.Errorf("upstream hits = %d, want 2: the first query was not cached", gh.metaHits.Load()) } @@ -521,13 +524,13 @@ func TestNestedMetadataPathsAreBothCached(t *testing.T) { "/repos/acme/tool/releases/latest", // nested under it } for _, path := range paths { - _ = body(t, get(t, base+path)) + _ = get(t, base+path).data } if gh.metaHits.Load() != 2 { t.Fatalf("upstream hits = %d, want 2", gh.metaHits.Load()) } for _, path := range paths { - _ = body(t, get(t, base+path)) + _ = get(t, base+path).data } if got := gh.metaHits.Load(); got != 2 { t.Errorf("upstream hits = %d after re-reading both documents, want 2: one of them is not being cached", got)