diff --git a/.golangci.yml b/.golangci.yml index 5d450009..8011474c 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -26,6 +26,13 @@ linters: linters: - errorlint - bodyclose + # The cargo proxy tests route every request through a get() helper that + # reads the body and closes it in a defer. bodyclose tracks the + # *http.Response the helper returns, not what the helper does with it, + # so it flags all 30-odd call sites. Same reasoning as localtunnel above. + - path: "pkg/proxies/cargo/.*_test\\.go" + linters: + - bodyclose - path: "api/v1" linters: - errorlint diff --git a/README.md b/README.md index 117bd969..11882065 100644 --- a/README.md +++ b/README.md @@ -389,6 +389,16 @@ enabled = true # run a GOPROXY on the bridge gateway # upstream = "https://proxy.golang.org" # upstream to fetch from on cache miss # cleanup = true # wipe cache on shutdown +# Cargo/crates caching proxy — pull-through cache for the crates.io sparse +# index, .crate tarballs, and rustup toolchains. No workflow changes needed. +[cargo_proxy] +enabled = true # run the Cargo proxy on the bridge gateway +# port = 8083 # default listen port +# upstream = "https://index.crates.io" # sparse registry index +# rustup_upstream = "https://static.rust-lang.org" +# index_ttl = "10m" # index revalidation (tarballs are immutable) +# cleanup = false # keep the cache across restarts + [log] level = "info" # debug, info, warn, error format = "text" # text or json diff --git a/cmd/ephemerd/cache.go b/cmd/ephemerd/cache.go index 2bc67a1e..16e84ed5 100644 --- a/cmd/ephemerd/cache.go +++ b/cmd/ephemerd/cache.go @@ -65,6 +65,18 @@ func managedCaches() []cacheEntry { Description: "Go module proxy cache (GOPROXY) served to job containers", LiveSafe: true, }, + { + Name: "cargo", + Rel: filepath.Join("cache", "cargo"), + Description: "Cargo proxy cache (crates.io sparse index, .crate tarballs, rustup toolchains) " + + "served to job containers", + // Live-safe: every entry is a pull-through copy of public + // registry content, so a running job that misses simply + // refetches it. The generated container config that jobs + // bind-mount deliberately lives OUTSIDE this dir (/cargo), + // so clearing the cache cannot yank it from a running job. + LiveSafe: true, + }, { Name: "buildkit", Rel: "buildkit", diff --git a/cmd/ephemerd/main.go b/cmd/ephemerd/main.go index 17867408..6a84c0d3 100644 --- a/cmd/ephemerd/main.go +++ b/cmd/ephemerd/main.go @@ -27,6 +27,7 @@ import ( "github.com/ephpm/ephemerd/pkg/providers/gitea" githubProv "github.com/ephpm/ephemerd/pkg/providers/github" "github.com/ephpm/ephemerd/pkg/proxies" + cargoproxy "github.com/ephpm/ephemerd/pkg/proxies/cargo" goproxy "github.com/ephpm/ephemerd/pkg/proxies/go" "github.com/ephpm/ephemerd/pkg/runner" "github.com/ephpm/ephemerd/pkg/runtime" @@ -409,6 +410,13 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP if cfg.ModuleProxy.Enabled { gatewayPorts = append(gatewayPorts, modProxyPort) } + cargoProxyPort := cfg.CargoProxy.Port + if cargoProxyPort == 0 { + cargoProxyPort = 8083 + } + if cfg.CargoProxy.Enabled { + gatewayPorts = append(gatewayPorts, cargoProxyPort) + } // Initialize container networking net, err := networking.New(networking.Config{ @@ -436,16 +444,11 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP if upstream == "" { upstream = "https://proxy.golang.org" } - cleanup := cfg.ModuleProxy.Cleanup - if !cleanup { - cleanup = true - } - goProxy := goproxy.New(goproxy.Config{ CacheDir: joinPath(configDir, "cache", "gomod"), Upstream: upstream, ListenAddr: fmt.Sprintf("%s:%d", net.GatewayIP(), modProxyPort), - Cleanup: cleanup, + Cleanup: cfg.ModuleProxy.CleanupEnabled(), Log: log, }) if err := goProxy.Start(); err != nil { @@ -460,10 +463,43 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP } } - // Collect env vars from all cache proxies for injection into containers. + // Start Cargo/crates caching proxy if enabled + if cfg.CargoProxy.Enabled { + cargoProxy := cargoproxy.New(cargoproxy.Config{ + CacheDir: joinPath(configDir, "cache", "cargo"), + // Config dir sits OUTSIDE the cache so `ephemerd cache clear + // cargo` cannot delete the file mounted into running jobs. + ConfDir: joinPath(configDir, "cargo"), + IndexUpstream: cfg.CargoProxy.Upstream, + RustupUpstream: cfg.CargoProxy.RustupUpstream, + ListenAddr: fmt.Sprintf("%s:%d", net.GatewayIP(), cargoProxyPort), + IndexTTL: cfg.CargoProxy.IndexTTL, + Cleanup: cfg.CargoProxy.CleanupEnabled(), + Log: log, + }) + if err := cargoProxy.Start(); err != nil { + log.Warn("failed to start Cargo proxy, continuing without it", "error", err) + } else { + cacheProxies = append(cacheProxies, cargoProxy) + defer func() { + if err := cargoProxy.Stop(); err != nil { + log.Warn("error stopping Cargo proxy", "error", err) + } + }() + } + } + + // Collect env vars and mounts from all cache proxies for injection into + // containers. Only proxies that actually STARTED are in cacheProxies, so + // a failed proxy is never advertised to a job — that is the outer + // fail-open: jobs go straight to the upstream registry instead. var cacheProxyEnvVars []string + var cacheProxyMounts []proxies.Mount for _, cp := range cacheProxies { cacheProxyEnvVars = append(cacheProxyEnvVars, cp.EnvVars()...) + if mp, ok := cp.(proxies.MountProvider); ok { + cacheProxyMounts = append(cacheProxyMounts, mp.Mounts()...) + } } // Start the shared embedded BuildKit solver. One solver serves every @@ -516,6 +552,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP DindEnabled: cfg.Dind.Enabled, DindAllowPrivileged: cfg.Dind.ResolvedAllowPrivileged(), CacheProxyEnv: cacheProxyEnvVars, + CacheProxyMounts: cacheProxyMounts, Rlimits: cfg.Runtime.Rlimits.Resolved(), AllowNewPrivileges: cfg.Runtime.ResolvedAllowNewPrivileges(), Network: net, diff --git a/config.example.toml b/config.example.toml index 88cc680e..8661779b 100644 --- a/config.example.toml +++ b/config.example.toml @@ -203,6 +203,36 @@ shutdown_timeout = "5m" # [dispatch] # token = "..." +# Go module caching proxy. Runs a GOPROXY on the bridge gateway and injects +# GOPROXY into every job container, so `go mod download` hits the local cache +# instead of proxy.golang.org on each build. +# [module_proxy] +# enabled = false +# port = 8082 # listen port on the bridge gateway +# upstream = "https://proxy.golang.org" # fetched from on a cache miss +# cleanup = true # wipe the cache on shutdown + +# Cargo/crates caching proxy. A pull-through cache for the crates.io sparse +# index, .crate tarballs, and rustup toolchain artifacts. +# +# Jobs need no workflow changes: rustup is pointed at the proxy with +# RUSTUP_DIST_SERVER, and Cargo with a generated .cargo/config.toml that +# ephemerd bind-mounts read-only at the container's filesystem root. A repo +# that ships its own .cargo/config.toml still wins. +# +# Fails open: if the proxy does not start, nothing is injected and jobs go +# straight to crates.io. If an upstream is unreachable, cached data is served +# stale and uncached downloads are redirected to the origin. +# [cargo_proxy] +# enabled = false +# port = 8083 # listen port on the bridge gateway +# upstream = "https://index.crates.io" # sparse registry index +# rustup_upstream = "https://static.rust-lang.org" # toolchain distribution +# index_ttl = "10m" # index revalidation interval; +# # .crate tarballs are immutable +# # and cached permanently +# cleanup = false # keep the cache across restarts + [log] # Log level: debug, info, warn, error level = "info" diff --git a/docs/arch/sccache-evaluation.md b/docs/arch/sccache-evaluation.md new file mode 100644 index 00000000..8cb70e4e --- /dev/null +++ b/docs/arch/sccache-evaluation.md @@ -0,0 +1,197 @@ +# sccache for Rust compilation caching — evaluation + +**Status:** design note, no implementation. Written alongside the `[cargo_proxy]` +pull-through cache (`pkg/proxies/cargo`). + +**Bottom line:** worth building, but **after** the disk-pressure GC lands, and +only with a hard size cap wired into that GC from day one. A compile cache is +the single biggest available win on Rust CI wall-clock — and, if left +unbounded, the most likely next disk-fill incident. + +--- + +## Why consider it at all + +The `[cargo_proxy]` this note accompanies removes *download* cost: the sparse +index, `.crate` tarballs, and rustup toolchains stop crossing the network on +every build. That is a real saving, but it is not where Rust CI time goes. + +A cold Rust CI build spends a small minority of its wall-clock fetching crates +and the overwhelming majority in `rustc`. Dependencies are the bulk of it: a +project with a few hundred transitive crates compiles all of them before it +compiles a line of its own code, and it does so again on every fresh runner +because ephemerd's containers are ephemeral by design. `cargo build` alone +caches nothing across jobs — `target/` dies with the container. + +sccache addresses exactly that: it wraps `rustc`, hashes the compilation +inputs, and returns a cached object file on a hit. Unlike the registry proxy, +its benefit scales with dependency *compile* cost rather than dependency +*download* size, which is the dominant term. + +## Backend choice: local disk vs shared + +**Local disk (`SCCACHE_DIR`) on a host volume.** Simplest by far — a directory +on the runner host, bind-mounted into each job container. Hits come only from +jobs that previously ran *on this host*, which for a small fleet is most of +them. It has a built-in size cap (`SCCACHE_CACHE_SIZE`) with LRU eviction, +which is the single most important property given our history. No new network +service, no credentials, no cross-host trust boundary. + +**Shared/remote (S3, Redis, memcached, or `sccache --start-server` over +HTTP).** Hits are fleet-wide, so a cold host benefits from a warm one, and a +scale-out fleet converges on one cache instead of N. The costs are real +though: object storage or a Redis to operate, credentials to distribute into +job containers (a secret that untrusted CI code can read), and a write path +that lets any job poison the cache for every other host. It also reintroduces +network cost on the read path, partially undoing the thing we are optimising. + +**Recommendation: start local-disk-only.** It captures most of the benefit at a +fraction of the operational and security cost, and it is the configuration +whose eviction story we can actually enforce. Treat a shared backend as a +later, opt-in addition once there is evidence that cross-host misses matter. + +## Integration with ephemerd's job containers + +The mechanism is already built — `[cargo_proxy]` needed the same two levers, +and both generalise: + +- **Env vars** via `proxies.CacheProxy.EnvVars()`. sccache is configured + entirely through the environment: `RUSTC_WRAPPER=sccache`, `SCCACHE_DIR`, + `SCCACHE_CACHE_SIZE`. Unlike Cargo's source replacement, this needs no + config file, so no `MountProvider` is required for configuration. +- **A bind mount** via `proxies.MountProvider` for the cache directory itself + — but read-**write**, which is a materially different proposition from the + Cargo config mount (read-only). See the trust caveat below. + +The awkward part is the binary. `RUSTC_WRAPPER=sccache` requires an `sccache` +executable on `PATH` inside the container, and stock runner images do not have +one. Options, in increasing order of intrusiveness: + +1. Bind-mount a host-side `sccache` binary into the container and put its + directory on `PATH` (or set `RUSTC_WRAPPER` to the absolute mounted path, + avoiding the `PATH` edit entirely). Needs a statically-linked binary + matching the container's libc — musl builds exist and are the obvious + choice. This is the cheapest option and mirrors how the GitHub Actions + runner itself is already mounted in. +2. Ship it in ephemerd's embedded assets like the runner and CNI plugins, then + mount as above. Same runtime shape, adds a download to `mage download`. +3. Require the image to provide it. Rejected: it forces a workflow/image + change, which the `[cargo_proxy]` design explicitly avoided. + +Option 1 or 2. Note this makes the feature Linux-container-first; Windows and +the macOS native path would need separate handling and should be out of scope +for a first cut. + +## Correctness caveats + +sccache is conservative by design, but the failure mode of a compile cache is +*wrong output*, not a slow build, so this deserves care: + +- **Hash inputs.** sccache keys on the preprocessed source, the compiler + binary's hash, the full argument list, and the relevant env vars. That is + sound for ordinary `rustc` invocations. +- **Proc macros and build scripts are not cached.** `build.rs` output and + proc-macro expansion are outside sccache's model; crates that lean on them + see less benefit. This is a benefit ceiling, not a correctness risk. +- **Incremental compilation is incompatible.** sccache refuses to cache when + `CARGO_INCREMENTAL=1`. CI builds should set `CARGO_INCREMENTAL=0` anyway; + ephemerd should inject it alongside `RUSTC_WRAPPER` so the two settings can + never disagree. +- **Absolute paths leak into debug info.** Cached objects embed the paths they + were compiled from, so per-job workdir names (`/jobs//…`) + reduce the hit rate and can put a stale path in a backtrace. Mitigated with + `--remap-path-prefix`, which is worth injecting from the start rather than + retrofitting. +- **Toolchain churn invalidates everything.** A nightly bump changes the + compiler hash and orphans the entire cache. With a size cap and LRU this + self-heals; without one it silently doubles the footprint. +- **Trust boundary.** The cache directory is written by untrusted CI code and + read by the *next* job on the host. A malicious job can plant an object + under a key it predicts and have a later job link it. This is the same + category of risk the existing per-repo dind image cache manages by + namespacing per (provider, repo). The compile cache should do the same: + **partition the cache directory per repo**, not one shared pool. That + reduces the hit rate across repos and is the right trade. + +## Disk footprint — the part that must not be skipped + +We have just had two disk-exhaustion outages, both from caches that grew +without an eviction policy (most recently ~44 GB of BuildKit build cache in the +shared `buildkit` containerd namespace: 76 image records, 481 leases, 302 +snapshots, never cleaned). A compile cache is exactly the same shape of risk — +it is *designed* to accumulate — so it must not ship with the same gap. + +Non-negotiables for any implementation: + +- **A hard size cap, configured and enforced.** `SCCACHE_CACHE_SIZE` with + sccache's own LRU eviction, defaulting to something modest (10 GB is a + reasonable starting point) and surfaced as a config key, not a constant. + sccache enforces this itself, which makes it strictly better behaved than + BuildKit's cache has been. +- **Registration in `managedCaches()`** in `cmd/ephemerd/cache.go`, so + `ephemerd cache list` shows its size and `ephemerd cache clear sccache` + works. It is `LiveSafe: true` — a running job that loses a cache entry + simply recompiles. Follow the `cargo`/`gomod` entries as the template. +- **Participation in the disk-pressure GC** currently being built (which now + also covers the `buildkit` namespace). The GC must be able to evict from the + compile cache under pressure, and the compile cache must be low in the + eviction priority order — below anything that would force a re-download, and + well below live job state, but above nothing at all. A cache the GC cannot + see is a cache that will eventually fill the disk. +- **Per-repo partitioning interacts with the cap.** N repos × a per-repo cap is + the real ceiling. Either cap the aggregate and let the GC arbitrate, or size + the per-repo cap knowing the multiplier. Do not set a per-repo cap and quote + it as the total. + +## Interaction with BuildKit layer caching + +These two caches overlap and can double-count, which is worth designing around +rather than discovering later. + +BuildKit caches at *layer* granularity: a `docker build` whose Dockerfile runs +`cargo build` produces a layer keyed on the build context and the preceding +layers. sccache caches at *compilation unit* granularity, inside that build. +When both are active for the same work you can store the same compiled output +twice — once as object files in the sccache dir, once inside a BuildKit +snapshot — and the BuildKit copy is the one that has already caused an +incident. + +Two clean positions: + +- **Jobs that compile Rust directly on the runner** (`cargo build` as a + workflow step, which is the ephpm case): sccache applies, BuildKit is not + involved. No overlap. This is the case worth optimising for. +- **Jobs that compile Rust inside `docker build`**: BuildKit's layer cache + already covers a clean rebuild, and threading sccache into the builder means + plumbing the cache into the build (a cache mount) plus the env vars. The + incremental benefit over a working BuildKit cache is small. + +**Recommendation: scope sccache to the direct-`cargo build` path and do not +wire it into `docker build`.** That keeps the two caches disjoint by +construction, avoids two overlapping unbounded stores, and targets the case +that actually dominates. If BuildKit-side Rust caching is wanted later, the +right lever is a BuildKit cache mount, not a second copy of sccache. + +## Recommendation + +Build it, with these conditions: + +1. **Sequence it after the disk-pressure GC.** The GC is the prerequisite, not + a follow-up. Shipping an unbounded compile cache into a fleet that has just + had two disk outages would be repeating the mistake. +2. **Local disk backend, per-repo partitioned, hard size cap, registered in + `managedCaches()` and in the GC's eviction order.** +3. **Inject `RUSTC_WRAPPER`, `SCCACHE_DIR`, `SCCACHE_CACHE_SIZE`, + `CARGO_INCREMENTAL=0`, and `--remap-path-prefix` together**, via the + existing `CacheProxy`/`MountProvider` plumbing. Reuse it; do not invent a + parallel mechanism. +4. **Scope to the direct `cargo build` path**, Linux containers first. +5. **Fail open**, exactly as `[cargo_proxy]` does: if the binary or the cache + dir is unavailable, inject nothing and let the build compile normally. + sccache's own `SCCACHE_ERROR_LOG` should be surfaced, but a cache failure + must never fail a job. + +Expected payoff is a large reduction in Rust CI wall-clock on warm hosts — +substantially bigger than the registry proxy's, because it removes compile +time rather than download time. The risk is entirely in the disk dimension, +and it is fully mitigable with a cap the tool already implements. diff --git a/docs/cli/cache.md b/docs/cli/cache.md index 5e8d1328..a48759af 100644 --- a/docs/cli/cache.md +++ b/docs/cli/cache.md @@ -20,6 +20,7 @@ ephemerd cache clear --all [--yes] |------|-----------------------------|-------------|-------------------------| | `images` | `images/` | Staged OCI image tarballs imported into containerd on startup | yes | | `gomod` | `cache/gomod/` | Go module proxy cache (GOPROXY) served to job containers | yes | +| `cargo` | `cache/cargo/` | Cargo proxy cache (crates.io sparse index, `.crate` tarballs, rustup toolchains) served to job containers | yes | | `buildkit` | `buildkit/` | Embedded BuildKit solver cache + history (`docker build` layers) | no | | `worker` | `worker/` | BuildKit worker snapshot/content root | no | | `runners` | `runners/` | Extracted GitHub Actions runner binaries (per version/OS) | no | diff --git a/docs/getting-started/configuration.md b/docs/getting-started/configuration.md index 630c0da1..d4bcd4f2 100644 --- a/docs/getting-started/configuration.md +++ b/docs/getting-started/configuration.md @@ -117,6 +117,21 @@ max_concurrent = 4 # max simultaneous jobs # cache_prune_interval = "24h" # how often the per-repo image cache pruner runs # cache_max_age = "168h" # evict cached image records inactive longer than this (7 days) +# --- Package caching proxies -------------------------------------------------- +# [module_proxy] +# enabled = false # run a GOPROXY on the bridge gateway +# port = 8082 # listen port +# upstream = "https://proxy.golang.org" +# cleanup = true # wipe the cache on shutdown + +# [cargo_proxy] +# enabled = false # pull-through cache for crates.io + rustup +# port = 8083 # listen port +# upstream = "https://index.crates.io" +# rustup_upstream = "https://static.rust-lang.org" +# index_ttl = "10m" # sparse-index revalidation interval +# cleanup = false # keep the cache across restarts + # --- Metrics ------------------------------------------------------------------ [metrics] # enabled = false # expose Prometheus /metrics endpoint @@ -284,6 +299,57 @@ The cache namespace persists across jobs and across ephemerd restarts. Per-job s **Disabling caching.** Setting `cache_max_age = "0"` disables eviction (the cache grows unbounded — useful for debugging but not recommended in production). Setting `cache_prune_interval = "0"` disables the pruner goroutine entirely; equivalent to "keep everything forever, even empty namespaces." +### `[module_proxy]` + +Go module caching proxy. ephemerd runs a single GOPROXY on the bridge gateway and injects `GOPROXY=http://:|direct` into every job container, so repeated `go mod download` runs hit the local disk cache instead of `proxy.golang.org`. + +| Field | Type | Default | Description | +|---|---|---|---| +| `enabled` | boolean | `false` | Run the Go module proxy | +| `port` | integer | `8082` | Listen port on the bridge gateway | +| `upstream` | string | `"https://proxy.golang.org"` | Fetched from on a cache miss | +| `cleanup` | boolean | `true` | Wipe the cache directory on shutdown. Set `false` to keep it across restarts. | + +Immutable module files (`.info`, `.mod`, `.zip`) are cached; mutable endpoints (`@latest`, `@v/list`) and `sumdb` requests pass through. The `|direct` separator means the go command falls back to the origin on **any** proxy error, so a broken cache slows a build rather than failing it. + +### `[cargo_proxy]` + +Cargo/crates caching proxy — the Rust counterpart to `[module_proxy]`. One HTTP server on the bridge gateway serves three routes: + +| Route | Upstream | Caching | +|---|---|---| +| `/index/…` | `upstream` (sparse registry index) | **Mutable** — served from cache for `index_ttl`, then revalidated with a conditional GET | +| `/crates/{name}/{version}/download` | the registry's own `dl` template | **Immutable** — cached permanently, never refetched | +| `/rustup/…` | `rustup_upstream` | Dated artifacts (`dist/YYYY-MM-DD/…`) immutable; channel manifests revalidated on `index_ttl` | + +| Field | Type | Default | Description | +|---|---|---|---| +| `enabled` | boolean | `false` | Run the Cargo proxy | +| `port` | integer | `8083` | Listen port on the bridge gateway | +| `upstream` | string | `"https://index.crates.io"` | Sparse registry index | +| `rustup_upstream` | string | `"https://static.rust-lang.org"` | Toolchain distribution server | +| `index_ttl` | duration | `"10m"` | How long a cached index entry is served before revalidation. A negative value revalidates on every request. | +| `cleanup` | boolean | `false` | Wipe the cache on shutdown. Defaults to **false**, unlike `[module_proxy]` — a pull-through cache that empties itself on every restart saves nothing. | + +**How jobs pick it up — no workflow changes required.** + +- **rustup** reads its mirror from the environment, so ephemerd injects `RUSTUP_DIST_SERVER`. +- **Cargo does not.** Source replacement (`[source.crates-io] replace-with`) is the only mechanism Cargo offers for redirecting crates.io, and it is read **exclusively from config files** — `CARGO_SOURCE_*` environment variables are silently ignored. ephemerd therefore generates a `.cargo/config.toml` under `/cargo/` and bind-mounts it **read-only** at the container's filesystem root (`/.cargo`, or `C:\.cargo` on Windows). Cargo searches the current directory and *every ancestor* for `.cargo/config.toml`, so a file at the root applies to any workspace path a job checks out — no knowledge of the checkout location, the job user's home, or the image's `CARGO_HOME` is needed, and `CARGO_HOME` itself is left untouched so it stays writable. + +A repository that ships its own `.cargo/config.toml` still wins: Cargo prefers config closer to the workspace. + +**Fail-open behaviour.** A cache must never turn a registry hiccup into a red CI job, so failures degrade in three layers: + +1. If the proxy does not start, nothing is injected and no mount is added — jobs go straight to crates.io. +2. If an upstream is unreachable or returns 5xx and a cached copy exists, the **stale copy is served** with a warning. +3. If nothing is cached, crate and rustup downloads are answered with a `307` redirect to the real origin, so the job fetches it directly (slower, uncached) instead of failing. + +Genuine `404`s are passed through as `404` — a nonexistent crate version must stay distinguishable from an outage. + +**Scope.** The mount lands in the runner container. Jobs that run their steps inside a *further* container (a `container:` image spawned via Docker-in-Docker) do not inherit it. + +**Cache location.** Cached content lives at `/cache/cargo/` and is visible to `ephemerd cache list` / `ephemerd cache clear cargo`. The generated container config lives at `/cargo/` — deliberately outside the cache root, so clearing the cache cannot pull the mounted config out from under a running job. + ### `[metrics]` Prometheus metrics endpoint. diff --git a/pkg/config/config.go b/pkg/config/config.go index 6381722b..ea80bc9e 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -28,6 +28,7 @@ type Config struct { VM VMConfig `toml:"vm"` Dind DindConfig `toml:"dind"` ModuleProxy ModuleProxyConfig `toml:"module_proxy"` + CargoProxy CargoProxyConfig `toml:"cargo_proxy"` Runtime RuntimeConfig `toml:"runtime"` Runner RunnerConfig `toml:"runner"` Metrics MetricsConfig `toml:"metrics"` @@ -380,7 +381,59 @@ type ModuleProxyConfig struct { Enabled bool `toml:"enabled"` // enable Go module caching proxy Port int `toml:"port"` // listen port on bridge gateway (default 8082) Upstream string `toml:"upstream"` // upstream proxy URL (default "https://proxy.golang.org") - Cleanup bool `toml:"cleanup"` // wipe cache on shutdown (default true) + Cleanup *bool `toml:"cleanup"` // wipe cache on shutdown (default true) +} + +// CleanupEnabled reports whether the Go module cache is wiped on shutdown. +// Defaults to true, preserving the historical behavior. +// +// Pointer-typed so "unset" is distinguishable from an explicit false. The +// previous plain bool could not express that: the call site coerced any +// false value back to true, which silently ignored `cleanup = false`. +func (m *ModuleProxyConfig) CleanupEnabled() bool { + if m.Cleanup == nil { + return true + } + return *m.Cleanup +} + +// CargoProxyConfig configures the Cargo/crates caching proxy. +// +// When enabled, ephemerd runs a pull-through cache for the crates.io sparse +// index, .crate tarballs, and rustup toolchain artifacts on the bridge +// gateway. Job containers are pointed at it automatically: rustup via +// RUSTUP_DIST_SERVER, and Cargo via a generated .cargo/config.toml that is +// bind-mounted read-only at the container's filesystem root (Cargo ignores +// CARGO_SOURCE_* environment variables, so a file is the only mechanism). +type CargoProxyConfig struct { + // Enabled turns the proxy on. Default false. + Enabled bool `toml:"enabled"` + // Port is the listen port on the bridge gateway. Default 8083. + Port int `toml:"port"` + // Upstream is the sparse registry index base URL. + // Default "https://index.crates.io". + Upstream string `toml:"upstream"` + // RustupUpstream is the toolchain distribution server. + // Default "https://static.rust-lang.org". + RustupUpstream string `toml:"rustup_upstream"` + // IndexTTL is how long a cached sparse-index entry is served before a + // conditional revalidation. Default 10m. Crate tarballs ignore this — + // they are immutable and cached permanently. + IndexTTL time.Duration `toml:"index_ttl"` + // Cleanup wipes the cache on shutdown. Default FALSE, unlike the Go + // module proxy: the whole point of a pull-through cache is to survive + // restarts, and wiping it on every shutdown is what made the module + // proxy's cache worthless. + Cleanup *bool `toml:"cleanup"` +} + +// CleanupEnabled reports whether the Cargo cache is wiped on shutdown. +// Defaults to false — see the field comment. +func (c *CargoProxyConfig) CleanupEnabled() bool { + if c.Cleanup == nil { + return false + } + return *c.Cleanup } // VMConfig configures virtual machines for cross-OS job execution. diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index ccdbe1d9..d1b3a691 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -908,8 +908,78 @@ cleanup = false if cfg.ModuleProxy.Upstream != "https://goproxy.io" { t.Errorf("ModuleProxy.Upstream = %q, want %q", cfg.ModuleProxy.Upstream, "https://goproxy.io") } - if cfg.ModuleProxy.Cleanup { - t.Error("ModuleProxy.Cleanup = true, want false") + if cfg.ModuleProxy.CleanupEnabled() { + t.Error("ModuleProxy.CleanupEnabled() = true, want false (cleanup = false was set explicitly)") + } +} + +// TestModuleProxyCleanupDefault pins that an unset cleanup still defaults to +// true — the pointer field must not change existing behavior, only make an +// explicit false actually take effect. +func TestModuleProxyCleanupDefault(t *testing.T) { + var m ModuleProxyConfig + if !m.CleanupEnabled() { + t.Error("ModuleProxy.CleanupEnabled() = false when unset, want true") + } +} + +// --- CargoProxy config --- + +func TestLoad_CargoProxyConfig(t *testing.T) { + t.Setenv("GITHUB_TOKEN", "") + tmp := t.TempDir() + path := filepath.Join(tmp, "config.toml") + if err := os.WriteFile(path, []byte(` +[github] +token = "ghp_test" +owner = "org" + +[cargo_proxy] +enabled = true +port = 9100 +upstream = "https://index.example.test" +rustup_upstream = "https://static.example.test" +index_ttl = "5m" +cleanup = true +`), 0644); err != nil { + t.Fatal(err) + } + + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load() error: %v", err) + } + + if !cfg.CargoProxy.Enabled { + t.Error("CargoProxy.Enabled = false, want true") + } + if cfg.CargoProxy.Port != 9100 { + t.Errorf("CargoProxy.Port = %d, want 9100", cfg.CargoProxy.Port) + } + if cfg.CargoProxy.Upstream != "https://index.example.test" { + t.Errorf("CargoProxy.Upstream = %q", cfg.CargoProxy.Upstream) + } + if cfg.CargoProxy.RustupUpstream != "https://static.example.test" { + t.Errorf("CargoProxy.RustupUpstream = %q", cfg.CargoProxy.RustupUpstream) + } + if cfg.CargoProxy.IndexTTL != 5*time.Minute { + t.Errorf("CargoProxy.IndexTTL = %v, want 5m", cfg.CargoProxy.IndexTTL) + } + if !cfg.CargoProxy.CleanupEnabled() { + t.Error("CargoProxy.CleanupEnabled() = false, want true") + } +} + +// TestCargoProxyDefaults pins the defaults that matter: the proxy is opt-in, +// and — unlike the Go module proxy — its cache is NOT wiped on shutdown. A +// pull-through cache that empties itself on every restart saves nothing. +func TestCargoProxyDefaults(t *testing.T) { + var c CargoProxyConfig + if c.Enabled { + t.Error("CargoProxy.Enabled = true when unset, want false (opt-in)") + } + if c.CleanupEnabled() { + t.Error("CargoProxy.CleanupEnabled() = true when unset, want false") } } @@ -2397,12 +2467,12 @@ func TestResolvedReconcileInterval(t *testing.T) { in string want time.Duration }{ - {"", 30 * time.Minute}, // default (backstop only; event-driven self-heal is primary) - {"2m", 2 * time.Minute}, // custom - {"90s", 90 * time.Second}, // custom - {"garbage", 30 * time.Minute}, // unparseable -> default - {"0s", 0}, // explicit disable - {"-1m", 0}, // negative -> disabled + {"", 30 * time.Minute}, // default (backstop only; event-driven self-heal is primary) + {"2m", 2 * time.Minute}, // custom + {"90s", 90 * time.Second}, // custom + {"garbage", 30 * time.Minute}, // unparseable -> default + {"0s", 0}, // explicit disable + {"-1m", 0}, // negative -> disabled } for _, c := range cases { w := &WebhookConfig{ReconcileInterval: c.in} diff --git a/pkg/proxies/cargo/cache.go b/pkg/proxies/cargo/cache.go new file mode 100644 index 00000000..a532bc77 --- /dev/null +++ b/pkg/proxies/cargo/cache.go @@ -0,0 +1,355 @@ +package cargoproxy + +import ( + "encoding/json" + "fmt" + "path" + "path/filepath" + "regexp" + "strings" + "time" +) + +// freshness is the decision the cache makes for one request. Kept as a pure +// value so the TTL/revalidation policy can be table-tested without a server, +// a clock, or a filesystem. +type freshness int + +const ( + // fetchFresh: nothing usable on disk — fetch unconditionally. + fetchFresh freshness = iota + // serveCached: the cached copy is within its TTL (or immutable) — serve + // it and do not touch the network at all. + serveCached + // revalidate: a cached copy exists but has aged out — issue a + // conditional GET (If-None-Match / If-Modified-Since) so the common + // "nothing changed" case costs one 304 instead of a full body. + revalidate +) + +func (f freshness) String() string { + switch f { + case fetchFresh: + return "fetch" + case serveCached: + return "hit" + case revalidate: + return "revalidate" + default: + return "unknown" + } +} + +// entryMeta is the sidecar record stored next to every cached body. It holds +// the upstream validators so a stale entry can be revalidated cheaply. +type entryMeta struct { + ETag string `json:"etag,omitempty"` + LastModified string `json:"last_modified,omitempty"` + ContentType string `json:"content_type,omitempty"` + Fetched time.Time `json:"fetched"` +} + +// decideFreshness is the whole cache policy in one pure function. +// +// - immutable content (.crate tarballs, dated rustup artifacts) is +// content-addressed by its URL: once cached it is NEVER refetched and +// never revalidated, whatever its age. +// - mutable content (the sparse index, rustup channel manifests) is served +// from cache inside ttl and conditionally revalidated after that, so we +// never serve an indefinitely stale index. +// +// A zero or negative ttl means "always revalidate", which is the safe +// direction for mutable data. +func decideFreshness(cached bool, immutable bool, fetched, now time.Time, ttl time.Duration) freshness { + if !cached { + return fetchFresh + } + if immutable { + return serveCached + } + if ttl > 0 && now.Sub(fetched) < ttl { + return serveCached + } + return revalidate +} + +// crateNameRe matches the crate names crates.io permits: ASCII alphanumerics +// plus '-' and '_'. Anything else is rejected rather than sanitized — a name +// we do not recognise must never reach the filesystem. +var crateNameRe = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) + +// crateVersionRe is deliberately broader than strict semver (pre-release and +// build metadata are common) but still excludes separators and traversal. +var crateVersionRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._+-]{0,63}$`) + +// safeSegments splits a URL path into segments and rejects anything that +// could escape the cache root or confuse the host filesystem: traversal, +// empty segments, absolute/UNC forms, backslashes, drive letters, control +// characters, and Windows-hostile names. This is the single choke point every +// on-disk cache path goes through. +func safeSegments(urlPath string) ([]string, error) { + clean := strings.Trim(urlPath, "/") + if clean == "" { + return nil, fmt.Errorf("empty path") + } + segs := strings.Split(clean, "/") + out := make([]string, 0, len(segs)) + for _, s := range segs { + if s == "" { + return nil, fmt.Errorf("empty path segment in %q", urlPath) + } + if s == "." || s == ".." { + return nil, fmt.Errorf("path traversal segment %q in %q", s, urlPath) + } + if strings.ContainsAny(s, `\:*?"<>|`) { + return nil, fmt.Errorf("illegal character in path segment %q", s) + } + for _, r := range s { + if r < 0x20 || r == 0x7f { + return nil, fmt.Errorf("control character in path segment %q", s) + } + } + out = append(out, s) + } + return out, nil +} + +// cachePathUnder joins validated URL-path segments onto a cache root. +func cachePathUnder(root, urlPath string) (string, error) { + segs, err := safeSegments(urlPath) + if err != nil { + return "", err + } + return filepath.Join(append([]string{root}, segs...)...), nil +} + +// indexCachePath maps a sparse-index request path (e.g. "/se/rd/serde") to +// its on-disk location under /index. +func indexCachePath(cacheDir, indexPath string) (string, error) { + p, err := cachePathUnder(filepath.Join(cacheDir, "index"), indexPath) + if err != nil { + return "", fmt.Errorf("index cache path for %q: %w", indexPath, err) + } + return p, nil +} + +// crateCachePath maps a crate name+version to its on-disk .crate location. +// Both components are validated against the registry's own character rules, +// so a hostile "name" like "../../etc/passwd" is rejected outright rather +// than escaping the cache root. +func crateCachePath(cacheDir, name, version string) (string, error) { + if !crateNameRe.MatchString(name) { + return "", fmt.Errorf("invalid crate name %q", name) + } + if !crateVersionRe.MatchString(version) { + return "", fmt.Errorf("invalid crate version %q for %q", version, name) + } + lower := strings.ToLower(name) + return filepath.Join(cacheDir, "crates", lower, lower+"-"+version+".crate"), nil +} + +// rustupCachePath maps a rustup dist path to its on-disk location. +func rustupCachePath(cacheDir, distPath string) (string, error) { + p, err := cachePathUnder(filepath.Join(cacheDir, "rustup"), distPath) + if err != nil { + return "", fmt.Errorf("rustup cache path for %q: %w", distPath, err) + } + return p, nil +} + +// metaPath is the sidecar path for a cached body. Crate names and index +// entries never contain a dot, and rustup artifacts are immutable (no +// sidecar), so ".meta" cannot collide with a real cache entry. +func metaPath(bodyPath string) string { return bodyPath + ".meta" } + +// datedRustupRe matches the YYYY-MM-DD segment that rustup uses for pinned, +// immutable nightly/dated artifacts. +var datedRustupRe = regexp.MustCompile(`(^|/)\d{4}-\d{2}-\d{2}(/|$)`) + +// isImmutableRustupPath reports whether a rustup dist path is content-stable. +// +// Dated artifacts (dist/2026-08-01/rust-std-nightly-x86_64...tar.xz) are +// published once and never rewritten, so they cache forever. The channel +// manifests (dist/channel-rust-nightly.toml and its .sha256/.asc siblings) +// are rewritten daily in place, so they must revalidate — caching those +// forever would pin every job to the toolchain that happened to be current +// when the daemon first started. +func isImmutableRustupPath(distPath string) bool { + return datedRustupRe.MatchString(distPath) +} + +// isImmutableIndexPath reports whether a sparse-index path is immutable. +// Nothing in the index is: entries gain a line whenever a version is +// published or yanked, and config.json can be re-pointed. Kept as a named +// function so the call sites read symmetrically and the intent is explicit +// rather than an unexplained "false". +func isImmutableIndexPath(string) bool { return false } + +// indexPrefix returns the sparse-index directory prefix for a crate name, +// per the registry index layout: 1-char names live under "1", 2-char under +// "2", 3-char under "3/", and everything else under the first +// two characters then the next two. +func indexPrefix(name string) string { + switch len(name) { + case 0: + return "" + case 1: + return "1" + case 2: + return "2" + case 3: + return path.Join("3", name[0:1]) + default: + return path.Join(name[0:2], name[2:4]) + } +} + +// defaultDLTemplate is the download URL used when the upstream registry's +// config.json is not available (first crate request after a restart with an +// empty index cache). It is crates.io's published template. +const defaultDLTemplate = "https://static.crates.io/crates/{crate}/{crate}-{version}.crate" + +// registryConfig is the subset of a sparse registry's config.json we care +// about. Everything else is preserved verbatim on rewrite via rawConfig. +type registryConfig struct { + DL string `json:"dl"` + API string `json:"api,omitempty"` +} + +// parseDLTemplate extracts the "dl" template from a registry config.json. +// A config.json we cannot parse falls back to the crates.io default rather +// than failing the request — the worst case is that we proxy from the +// canonical CDN instead of a mirror. +func parseDLTemplate(raw []byte) string { + var cfg registryConfig + if err := json.Unmarshal(raw, &cfg); err != nil || strings.TrimSpace(cfg.DL) == "" { + return defaultDLTemplate + } + return cfg.DL +} + +// dlMarkers are the substitutions a registry may use in its "dl" template. +var dlMarkers = []string{"{crate}", "{version}", "{prefix}", "{lowerprefix}", "{sha256-checksum}"} + +// expandDL turns a registry "dl" template into a concrete upstream URL for +// one crate version, implementing the registry-index rules: +// +// - {crate}, {version}, {prefix}, {lowerprefix} are substituted; +// - a template containing NO markers gets "/{crate}/{version}/download" +// appended, which is the documented default; +// - {sha256-checksum} cannot be expanded here (the checksum lives in the +// index entry, which we do not parse), so such a template is reported +// unusable and the caller falls back to the crates.io default. +func expandDL(tmpl, name, version string) (string, bool) { + tmpl = strings.TrimSpace(tmpl) + if tmpl == "" { + return "", false + } + if strings.Contains(tmpl, "{sha256-checksum}") { + return "", false + } + + hasMarker := false + for _, m := range dlMarkers { + if strings.Contains(tmpl, m) { + hasMarker = true + break + } + } + if !hasMarker { + return strings.TrimRight(tmpl, "/") + "/" + name + "/" + version + "/download", true + } + + lower := strings.ToLower(name) + r := strings.NewReplacer( + "{crate}", name, + "{version}", version, + "{prefix}", indexPrefix(name), + "{lowerprefix}", indexPrefix(lower), + ) + return r.Replace(tmpl), true +} + +// rewriteConfigJSON rewrites a registry config.json so Cargo fetches crate +// tarballs from THIS proxy instead of the upstream CDN, while leaving every +// other field — notably "api", which `cargo publish`/`cargo search` use — +// pointing at the real registry. The rewritten "dl" is deliberately +// marker-free, so Cargo appends "/{crate}/{version}/download" and we get a +// single, unambiguous request shape to route. +// +// Unknown fields are preserved: the config is decoded into a generic map so +// a future registry key is passed through rather than silently dropped. +func rewriteConfigJSON(raw []byte, base string) ([]byte, error) { + var generic map[string]any + if err := json.Unmarshal(raw, &generic); err != nil { + return nil, fmt.Errorf("parsing registry config.json: %w", err) + } + generic["dl"] = strings.TrimRight(base, "/") + cratesRoute + out, err := json.Marshal(generic) + if err != nil { + return nil, fmt.Errorf("re-encoding registry config.json: %w", err) + } + return out, nil +} + +// parseCrateDownloadPath splits the crate-download route Cargo generates +// from our rewritten "dl" — "/crates///download" — into its +// components. Any other shape is rejected. +func parseCrateDownloadPath(urlPath string) (name, version string, ok bool) { + rest := strings.TrimPrefix(urlPath, cratesRoute+"/") + if rest == urlPath { + return "", "", false + } + parts := strings.Split(rest, "/") + if len(parts) != 3 || parts[2] != "download" { + return "", "", false + } + if !crateNameRe.MatchString(parts[0]) || !crateVersionRe.MatchString(parts[1]) { + return "", "", false + } + return parts[0], parts[1], true +} + +// containerConfigTOML renders the Cargo config that gets mounted into every +// job container. +// +// Source replacement is the ONLY mechanism Cargo offers for redirecting +// crates.io, and it is read exclusively from config files — CARGO_SOURCE_* +// environment variables are silently ignored (verified against cargo stable: +// with the env vars set, cargo still reported "Updating crates.io index"). +// Hence a generated file rather than an env var. +func containerConfigTOML(base string) string { + base = strings.TrimRight(base, "/") + return `# Generated by ephemerd. Do NOT edit — regenerated on every daemon start. +# +# Routes crates.io through ephemerd's on-host pull-through cache so repeated +# CI builds do not re-download the same .crate tarballs. Delete the +# [cargo_proxy] block from ephemerd's config.toml to turn this off. +# +# A repository's own .cargo/config.toml still wins: Cargo prefers config +# closer to the workspace, so a project that pins its own mirror is +# unaffected by this file. + +[source.crates-io] +replace-with = "ephemerd" + +[source.ephemerd] +registry = "sparse+` + base + indexRoute + `/" + +[net] +retry = 3 +` +} + +// containerConfigDest is the container path the generated .cargo directory is +// mounted at. Cargo walks the current directory and EVERY ancestor looking +// for .cargo/config.toml, so placing it at the filesystem root makes it apply +// to any workspace path a job might use — no knowledge of the checkout +// location, the job user's home, or the image's CARGO_HOME required, and +// CARGO_HOME itself is left alone so it stays writable. +func containerConfigDest(goos string) string { + if goos == "windows" { + return `C:\.cargo` + } + return "/.cargo" +} diff --git a/pkg/proxies/cargo/cache_test.go b/pkg/proxies/cargo/cache_test.go new file mode 100644 index 00000000..9c00de4e --- /dev/null +++ b/pkg/proxies/cargo/cache_test.go @@ -0,0 +1,542 @@ +package cargoproxy + +import ( + "encoding/json" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestDecideFreshness(t *testing.T) { + now := time.Date(2026, 8, 11, 12, 0, 0, 0, time.UTC) + ttl := 10 * time.Minute + + tests := []struct { + name string + cached bool + immutable bool + fetched time.Time + ttl time.Duration + want freshness + }{ + { + name: "nothing cached fetches", + want: fetchFresh, + }, + { + name: "nothing cached fetches even when immutable", + immutable: true, + want: fetchFresh, + }, + { + name: "immutable is served regardless of age", + cached: true, + immutable: true, + fetched: now.Add(-5000 * time.Hour), + ttl: ttl, + want: serveCached, + }, + { + name: "immutable with zero fetch time is still served", + cached: true, + immutable: true, + ttl: ttl, + want: serveCached, + }, + { + name: "mutable inside ttl is served", + cached: true, + fetched: now.Add(-1 * time.Minute), + ttl: ttl, + want: serveCached, + }, + { + name: "mutable exactly at ttl revalidates", + cached: true, + fetched: now.Add(-ttl), + ttl: ttl, + want: revalidate, + }, + { + name: "mutable past ttl revalidates", + cached: true, + fetched: now.Add(-1 * time.Hour), + ttl: ttl, + want: revalidate, + }, + { + name: "zero ttl always revalidates", + cached: true, + fetched: now, + ttl: 0, + want: revalidate, + }, + { + name: "negative ttl always revalidates", + cached: true, + fetched: now, + ttl: -time.Minute, + want: revalidate, + }, + { + name: "clock skew (future fetch time) still serves", + cached: true, + fetched: now.Add(time.Hour), + ttl: ttl, + want: serveCached, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := decideFreshness(tt.cached, tt.immutable, tt.fetched, now, tt.ttl) + if got != tt.want { + t.Errorf("decideFreshness(cached=%v, immutable=%v, age=%v, ttl=%v) = %v, want %v", + tt.cached, tt.immutable, now.Sub(tt.fetched), tt.ttl, got, tt.want) + } + }) + } +} + +func TestIndexPrefix(t *testing.T) { + tests := []struct{ name, want string }{ + {"a", "1"}, + {"go", "2"}, + {"bar", "3/b"}, + {"serde", "se/rd"}, + {"tokio", "to/ki"}, + {"cc", "2"}, + {"", ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := indexPrefix(tt.name); got != tt.want { + t.Errorf("indexPrefix(%q) = %q, want %q", tt.name, got, tt.want) + } + }) + } +} + +func TestSafeSegments_RejectsTraversalAndJunk(t *testing.T) { + bad := []string{ + "", + "/", + "//", + "..", + "/../etc/passwd", + "/se/../../etc/passwd", + "/se/rd/../../../root", + "/./serde", + `/se\rd\serde`, + "/C:/windows", + "/se//rd", + "/se/rd/ser\x00de", + "/se/rd/ser\nde", + "/se/rd/ser|de", + "/se/rd/ser*de", + } + for _, p := range bad { + t.Run(strings.ReplaceAll(p, "\x00", "NUL"), func(t *testing.T) { + if segs, err := safeSegments(p); err == nil { + t.Errorf("safeSegments(%q) = %v, want error", p, segs) + } + }) + } + + good := map[string][]string{ + "/se/rd/serde": {"se", "rd", "serde"}, + "se/rd/serde": {"se", "rd", "serde"}, + "/1/a": {"1", "a"}, + "/3/b/bar": {"3", "b", "bar"}, + "/config.json": {"config.json"}, + "/dist/x.tar.z": {"dist", "x.tar.z"}, + } + for p, want := range good { + t.Run("ok"+p, func(t *testing.T) { + got, err := safeSegments(p) + if err != nil { + t.Fatalf("safeSegments(%q): %v", p, err) + } + if strings.Join(got, "/") != strings.Join(want, "/") { + t.Errorf("safeSegments(%q) = %v, want %v", p, got, want) + } + }) + } +} + +// TestCachePaths_StayUnderRoot is the security property: no request path may +// produce a cache location outside the cache root. +func TestCachePaths_StayUnderRoot(t *testing.T) { + root := filepath.Join("cachedir") + + t.Run("index", func(t *testing.T) { + got, err := indexCachePath(root, "/se/rd/serde") + if err != nil { + t.Fatalf("indexCachePath: %v", err) + } + want := filepath.Join(root, "index", "se", "rd", "serde") + if got != want { + t.Errorf("indexCachePath = %q, want %q", got, want) + } + if _, err := indexCachePath(root, "/../../escape"); err == nil { + t.Error("indexCachePath accepted a traversal path") + } + }) + + t.Run("rustup", func(t *testing.T) { + got, err := rustupCachePath(root, "/dist/2026-08-01/rust-std.tar.xz") + if err != nil { + t.Fatalf("rustupCachePath: %v", err) + } + want := filepath.Join(root, "rustup", "dist", "2026-08-01", "rust-std.tar.xz") + if got != want { + t.Errorf("rustupCachePath = %q, want %q", got, want) + } + if _, err := rustupCachePath(root, "/dist/../../escape"); err == nil { + t.Error("rustupCachePath accepted a traversal path") + } + }) +} + +func TestCrateCachePath(t *testing.T) { + root := "cachedir" + tests := []struct { + name, crate, version, want string + wantErr bool + }{ + {name: "simple", crate: "serde", version: "1.0.203", want: filepath.Join(root, "crates", "serde", "serde-1.0.203.crate")}, + {name: "underscores and dashes", crate: "proc-macro2", version: "1.0.86", want: filepath.Join(root, "crates", "proc-macro2", "proc-macro2-1.0.86.crate")}, + {name: "case normalised", crate: "Inflector", version: "0.11.4", want: filepath.Join(root, "crates", "inflector", "inflector-0.11.4.crate")}, + {name: "prerelease version", crate: "tokio", version: "1.0.0-beta.1", want: filepath.Join(root, "crates", "tokio", "tokio-1.0.0-beta.1.crate")}, + {name: "build metadata", crate: "tokio", version: "1.0.0+build.5", want: filepath.Join(root, "crates", "tokio", "tokio-1.0.0+build.5.crate")}, + + {name: "traversal in name", crate: "../../etc/passwd", version: "1.0.0", wantErr: true}, + {name: "slash in name", crate: "a/b", version: "1.0.0", wantErr: true}, + {name: "dot in name", crate: "..", version: "1.0.0", wantErr: true}, + {name: "empty name", crate: "", version: "1.0.0", wantErr: true}, + {name: "traversal in version", crate: "serde", version: "../../x", wantErr: true}, + {name: "empty version", crate: "serde", version: "", wantErr: true}, + {name: "backslash in version", crate: "serde", version: `1.0\..\x`, wantErr: true}, + {name: "version starting with dot", crate: "serde", version: ".hidden", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := crateCachePath(root, tt.crate, tt.version) + if tt.wantErr { + if err == nil { + t.Fatalf("crateCachePath(%q, %q) = %q, want error", tt.crate, tt.version, got) + } + return + } + if err != nil { + t.Fatalf("crateCachePath(%q, %q): %v", tt.crate, tt.version, err) + } + if got != tt.want { + t.Errorf("crateCachePath = %q, want %q", got, tt.want) + } + }) + } +} + +func TestIsImmutableRustupPath(t *testing.T) { + tests := []struct { + path string + want bool + }{ + // Dated artifacts are published once and never rewritten. + {"/dist/2026-08-01/rust-std-nightly-x86_64-unknown-linux-gnu.tar.xz", true}, + {"/dist/2026-08-01/channel-rust-nightly.toml", true}, + {"/dist/2026-08-01/channel-rust-nightly.toml.sha256", true}, + {"dist/2024-12-31/rustc-nightly-src.tar.gz", true}, + {"/dist/2026-08-01", true}, + + // Channel manifests are rewritten in place — must revalidate. + {"/dist/channel-rust-nightly.toml", false}, + {"/dist/channel-rust-stable.toml", false}, + {"/dist/channel-rust-beta.toml.sha256", false}, + {"/rustup/dist/x86_64-unknown-linux-gnu/rustup-init", false}, + {"/rustup/release-stable.toml", false}, + {"", false}, + // Near-misses must not be mistaken for a date segment. + {"/dist/20260801/x.tar.xz", false}, + {"/dist/v2026-08-01x/x.tar.xz", false}, + } + for _, tt := range tests { + t.Run(tt.path, func(t *testing.T) { + if got := isImmutableRustupPath(tt.path); got != tt.want { + t.Errorf("isImmutableRustupPath(%q) = %v, want %v", tt.path, got, tt.want) + } + }) + } +} + +func TestIsImmutableIndexPath(t *testing.T) { + // Every index path is mutable — a publish or a yank rewrites the entry. + for _, p := range []string{"/se/rd/serde", "/config.json", "/1/a"} { + if isImmutableIndexPath(p) { + t.Errorf("isImmutableIndexPath(%q) = true, want false", p) + } + } +} + +func TestExpandDL(t *testing.T) { + tests := []struct { + name string + tmpl string + crate string + version string + want string + wantOK bool + }{ + { + name: "crates.io marker template", + tmpl: "https://static.crates.io/crates/{crate}/{crate}-{version}.crate", + crate: "serde", + version: "1.0.203", + want: "https://static.crates.io/crates/serde/serde-1.0.203.crate", + wantOK: true, + }, + { + name: "marker-free template appends the default suffix", + tmpl: "https://static.crates.io/crates", + crate: "serde", + version: "1.0.203", + want: "https://static.crates.io/crates/serde/1.0.203/download", + wantOK: true, + }, + { + name: "marker-free template with trailing slash", + tmpl: "https://example.test/dl/", + crate: "tokio", + version: "1.2.3", + want: "https://example.test/dl/tokio/1.2.3/download", + wantOK: true, + }, + { + name: "prefix markers", + tmpl: "https://example.test/{prefix}/{crate}/{version}.crate", + crate: "serde", + version: "1.0.0", + want: "https://example.test/se/rd/serde/1.0.0.crate", + wantOK: true, + }, + { + name: "lowerprefix marker lowercases", + tmpl: "https://example.test/{lowerprefix}/{crate}", + crate: "Inflector", + version: "0.1.0", + want: "https://example.test/in/fl/Inflector", + wantOK: true, + }, + { + name: "short name prefix", + tmpl: "https://example.test/{prefix}/{crate}-{version}.crate", + crate: "cc", + version: "1.0.0", + want: "https://example.test/2/cc-1.0.0.crate", + wantOK: true, + }, + { + name: "sha256 marker is unusable", + tmpl: "https://example.test/{sha256-checksum}/{crate}", + crate: "serde", + version: "1.0.0", + wantOK: false, + }, + { + name: "empty template", + tmpl: "", + crate: "serde", + wantOK: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := expandDL(tt.tmpl, tt.crate, tt.version) + if ok != tt.wantOK { + t.Fatalf("expandDL ok = %v, want %v (got %q)", ok, tt.wantOK, got) + } + if ok && got != tt.want { + t.Errorf("expandDL = %q, want %q", got, tt.want) + } + }) + } +} + +func TestParseDLTemplate(t *testing.T) { + tests := []struct { + name string + raw string + want string + }{ + { + name: "real crates.io config", + raw: `{"dl":"https://static.crates.io/crates/{crate}/{crate}-{version}.crate","api":"https://crates.io"}`, + want: "https://static.crates.io/crates/{crate}/{crate}-{version}.crate", + }, + { + name: "marker-free dl", + raw: `{"dl":"https://example.test/crates","api":"https://example.test"}`, + want: "https://example.test/crates", + }, + {name: "malformed json falls back", raw: `{not json`, want: defaultDLTemplate}, + {name: "missing dl falls back", raw: `{"api":"https://crates.io"}`, want: defaultDLTemplate}, + {name: "blank dl falls back", raw: `{"dl":" "}`, want: defaultDLTemplate}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := parseDLTemplate([]byte(tt.raw)); got != tt.want { + t.Errorf("parseDLTemplate(%s) = %q, want %q", tt.raw, got, tt.want) + } + }) + } +} + +func TestRewriteConfigJSON(t *testing.T) { + raw := []byte(`{"dl":"https://static.crates.io/crates/{crate}/{crate}-{version}.crate","api":"https://crates.io","auth-required":false}`) + + out, err := rewriteConfigJSON(raw, "http://10.88.0.1:8083") + if err != nil { + t.Fatalf("rewriteConfigJSON: %v", err) + } + + var got map[string]any + if err := json.Unmarshal(out, &got); err != nil { + t.Fatalf("rewritten config is not valid JSON: %v", err) + } + + if got["dl"] != "http://10.88.0.1:8083/crates" { + t.Errorf("dl = %v, want the proxy's /crates route", got["dl"]) + } + // "api" must survive untouched so `cargo publish` and `cargo search` + // still reach the real registry. + if got["api"] != "https://crates.io" { + t.Errorf("api = %v, want https://crates.io (must not be rewritten)", got["api"]) + } + // Unknown keys must be preserved rather than dropped. + if _, ok := got["auth-required"]; !ok { + t.Error("rewrite dropped the unknown auth-required key") + } + + if _, err := rewriteConfigJSON([]byte("not json"), "http://x"); err == nil { + t.Error("rewriteConfigJSON accepted invalid JSON") + } +} + +// TestRewrittenDLRoundTrips pins the contract between the two halves of the +// crate route: whatever we advertise as "dl" must expand to a URL our own +// parseCrateDownloadPath can decode. +func TestRewrittenDLRoundTrips(t *testing.T) { + base := "http://10.88.0.1:8083" + out, err := rewriteConfigJSON([]byte(`{"dl":"https://static.crates.io/crates"}`), base) + if err != nil { + t.Fatalf("rewriteConfigJSON: %v", err) + } + tmpl := parseDLTemplate(out) + + url, ok := expandDL(tmpl, "serde", "1.0.203") + if !ok { + t.Fatal("expandDL rejected our own advertised template") + } + reqPath := strings.TrimPrefix(url, base) + + name, version, ok := parseCrateDownloadPath(reqPath) + if !ok { + t.Fatalf("parseCrateDownloadPath(%q) failed for our own route", reqPath) + } + if name != "serde" || version != "1.0.203" { + t.Errorf("round trip = (%q, %q), want (serde, 1.0.203)", name, version) + } +} + +func TestParseCrateDownloadPath(t *testing.T) { + tests := []struct { + path string + name, vers string + ok bool + }{ + {path: "/crates/serde/1.0.203/download", name: "serde", vers: "1.0.203", ok: true}, + {path: "/crates/proc-macro2/1.0.86/download", name: "proc-macro2", vers: "1.0.86", ok: true}, + {path: "/crates/tokio/1.0.0-beta.1/download", name: "tokio", vers: "1.0.0-beta.1", ok: true}, + + {path: "/crates/serde/1.0.203", ok: false}, + {path: "/crates/serde/1.0.203/fetch", ok: false}, + {path: "/crates/serde/1.0.203/download/extra", ok: false}, + {path: "/crates/serde", ok: false}, + {path: "/crates/", ok: false}, + {path: "/index/se/rd/serde", ok: false}, + {path: "/crates/../../etc/1.0.0/download", ok: false}, + {path: "/crates/se rde/1.0.0/download", ok: false}, + } + for _, tt := range tests { + t.Run(tt.path, func(t *testing.T) { + name, vers, ok := parseCrateDownloadPath(tt.path) + if ok != tt.ok { + t.Fatalf("parseCrateDownloadPath(%q) ok = %v, want %v", tt.path, ok, tt.ok) + } + if ok && (name != tt.name || vers != tt.vers) { + t.Errorf("= (%q, %q), want (%q, %q)", name, vers, tt.name, tt.vers) + } + }) + } +} + +func TestContainerConfigTOML(t *testing.T) { + got := containerConfigTOML("http://10.88.0.1:8083/") + + // Source replacement is the whole point — without both halves Cargo + // silently keeps using crates.io. + for _, want := range []string{ + "[source.crates-io]", + `replace-with = "ephemerd"`, + "[source.ephemerd]", + `registry = "sparse+http://10.88.0.1:8083/index/"`, + } { + if !strings.Contains(got, want) { + t.Errorf("generated config missing %q:\n%s", want, got) + } + } + // A doubled slash would produce an index URL Cargo cannot use. + if strings.Contains(got, "8083//index") { + t.Errorf("trailing slash in base leaked into the index URL:\n%s", got) + } +} + +func TestContainerConfigDest(t *testing.T) { + if got := containerConfigDest("linux"); got != "/.cargo" { + t.Errorf("containerConfigDest(linux) = %q, want /.cargo", got) + } + if got := containerConfigDest("darwin"); got != "/.cargo" { + t.Errorf("containerConfigDest(darwin) = %q, want /.cargo", got) + } + if got := containerConfigDest("windows"); got != `C:\.cargo` { + t.Errorf(`containerConfigDest(windows) = %q, want C:\.cargo`, got) + } +} + +func TestMatchesETag(t *testing.T) { + tests := []struct { + ifNoneMatch, tag string + want bool + }{ + {`"abc"`, `"abc"`, true}, + {`W/"abc"`, `"abc"`, true}, + {`"abc"`, `W/"abc"`, true}, + {`"x", "abc"`, `"abc"`, true}, + {`*`, `"abc"`, true}, + {`"abc"`, `"def"`, false}, + {``, `"abc"`, false}, + {`"abc"`, ``, false}, + {``, ``, false}, + } + for _, tt := range tests { + t.Run(tt.ifNoneMatch+"|"+tt.tag, func(t *testing.T) { + if got := matchesETag(tt.ifNoneMatch, tt.tag); got != tt.want { + t.Errorf("matchesETag(%q, %q) = %v, want %v", tt.ifNoneMatch, tt.tag, got, tt.want) + } + }) + } +} diff --git a/pkg/proxies/cargo/cargoproxy.go b/pkg/proxies/cargo/cargoproxy.go new file mode 100644 index 00000000..959c2d48 --- /dev/null +++ b/pkg/proxies/cargo/cargoproxy.go @@ -0,0 +1,752 @@ +// Package cargoproxy implements proxies.CacheProxy for the Rust ecosystem. +// +// It is a pull-through cache for three upstreams that a Rust CI job hits on +// every run: +// +// /index/… the crates.io SPARSE registry index (mutable) +// /crates/{c}/{v}/… .crate tarballs (immutable, content-addressed) +// /rustup/… rustup toolchain artifacts (dated ones immutable) +// +// ephemerd runs one shared instance on the bridge gateway IP so every job +// container can reach it. Jobs only ever issue HTTP GETs; they have no write +// access to the cache. +// +// Why both a mount and env vars: rustup takes its mirror from the +// environment (RUSTUP_DIST_SERVER), but Cargo does NOT — source replacement +// is only honoured from a config file, so the proxy generates a +// .cargo/config.toml on the host and declares it as a read-only mount at the +// container's filesystem root, where Cargo's ancestor-directory config search +// always finds it. See containerConfigTOML/containerConfigDest. +package cargoproxy + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net" + "net/http" + "os" + "path/filepath" + goruntime "runtime" + "strings" + "sync" + "time" + + "github.com/ephpm/ephemerd/pkg/proxies" +) + +// hostGOOS is the OS this daemon runs on. Job containers share the host's +// OS (Linux jobs on Windows/macOS hosts run inside a Linux VM that has its +// own ephemerd), so it doubles as the default container OS. +const hostGOOS = goruntime.GOOS + +// Route prefixes. These are part of the contract with the generated +// container config (containerConfigTOML) and with the rewritten config.json. +const ( + indexRoute = "/index" + cratesRoute = "/crates" + rustupRoute = "/rustup" + healthRoute = "/healthz" +) + +// Defaults. Kept here rather than in the config package so the proxy is +// usable standalone (and testable) without a config file. +const ( + // DefaultIndexUpstream is the crates.io sparse index. + DefaultIndexUpstream = "https://index.crates.io" + // DefaultRustupUpstream is the rustup/toolchain distribution server. + DefaultRustupUpstream = "https://static.rust-lang.org" + // DefaultIndexTTL is how long a cached sparse-index entry is served + // without contacting upstream. Short by design: the index is mutable + // and a new dependency version must not take an hour to become + // visible. Revalidation after the TTL is a conditional GET, so the + // steady-state cost is a 304, not a re-download. + DefaultIndexTTL = 10 * time.Minute + // defaultTimeout bounds a single upstream request. + defaultTimeout = 120 * time.Second +) + +// Config for the Cargo caching proxy. +type Config struct { + // CacheDir is the on-disk cache root (e.g. /cache/cargo). + CacheDir string + // ConfDir is where the container-side Cargo config is generated. It is + // deliberately OUTSIDE CacheDir so `ephemerd cache clear cargo` cannot + // pull the mounted config out from under a running job. + ConfDir string + // IndexUpstream is the sparse registry index base URL. + IndexUpstream string + // RustupUpstream is the rustup distribution server base URL. + RustupUpstream string + // ListenAddr is the address to bind (e.g. "10.88.0.1:8083"). It is also + // the address advertised to containers — see advertiseBase. + ListenAddr string + // IndexTTL is the revalidation interval for mutable index entries. + // Zero means "unset" and takes DefaultIndexTTL; a NEGATIVE value means + // "revalidate on every request" (useful for tests and for operators who + // want the index never served without a conditional GET). + IndexTTL time.Duration + // Cleanup wipes the cache dir on Stop. Defaults to false: a + // pull-through cache that is emptied on every restart saves nothing. + Cleanup bool + // ContainerOS is the OS of the job containers this proxy serves, used + // only to pick the mount destination. Defaults to the host GOOS. + ContainerOS string + Log *slog.Logger +} + +// Compile-time interface checks. +var ( + _ proxies.CacheProxy = (*Proxy)(nil) + _ proxies.MountProvider = (*Proxy)(nil) +) + +// Proxy is a caching proxy for the Cargo registry and rustup distribution. +type Proxy struct { + cfg Config + server *http.Server + listener net.Listener + client *http.Client + inflight sync.Map // per-path mutex: collapses duplicate upstream fetches + + mu sync.RWMutex + // dlTemplate is the upstream registry's "dl" template, learned from + // config.json. Guarded because index requests (writers) and crate + // downloads (readers) run concurrently. + dlTemplate string +} + +// New creates a Cargo caching proxy. Call Start() to begin serving. +func New(cfg Config) *Proxy { + if cfg.IndexUpstream == "" { + cfg.IndexUpstream = DefaultIndexUpstream + } + if cfg.RustupUpstream == "" { + cfg.RustupUpstream = DefaultRustupUpstream + } + if cfg.IndexTTL == 0 { + cfg.IndexTTL = DefaultIndexTTL + } + if cfg.Log == nil { + cfg.Log = slog.Default() + } + cfg.IndexUpstream = strings.TrimRight(cfg.IndexUpstream, "/") + cfg.RustupUpstream = strings.TrimRight(cfg.RustupUpstream, "/") + + return &Proxy{ + cfg: cfg, + client: &http.Client{Timeout: defaultTimeout}, + dlTemplate: defaultDLTemplate, + } +} + +// Start begins serving the proxy. Returns after the listener is bound and +// the container-side Cargo config has been written. +func (p *Proxy) Start() error { + if err := os.MkdirAll(p.cfg.CacheDir, 0o755); err != nil { + return fmt.Errorf("creating cargo cache dir: %w", err) + } + if err := p.writeContainerConfig(); err != nil { + return fmt.Errorf("writing container cargo config: %w", err) + } + // Recover the upstream "dl" template from a previous run so the first + // crate download after a restart does not have to guess. + p.loadDLTemplateFromCache() + + // proxies.Listen, not net.Listen: the bridge gateway IP is not assigned + // to any interface until CNI creates the bridge for the first job. + ln, err := proxies.Listen(p.cfg.ListenAddr, p.cfg.Log) + if err != nil { + return err + } + p.listener = ln + + mux := http.NewServeMux() + mux.HandleFunc("/", p.handle) + p.server = &http.Server{Handler: mux, ReadHeaderTimeout: 30 * time.Second} + + go func() { + if err := p.server.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + p.cfg.Log.Error("cargo proxy server error", "error", err) + } + }() + + p.cfg.Log.Info("cargo proxy started", + "addr", ln.Addr().String(), + "advertised", p.advertiseBase(), + "cache", p.cfg.CacheDir, + "index_upstream", p.cfg.IndexUpstream, + "rustup_upstream", p.cfg.RustupUpstream, + "index_ttl", p.cfg.IndexTTL) + return nil +} + +// Stop shuts down the proxy and optionally wipes the cache. +func (p *Proxy) Stop() error { + var errs []error + + if p.server != nil { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := p.server.Shutdown(ctx); err != nil { + errs = append(errs, fmt.Errorf("shutting down cargo proxy: %w", err)) + } + } + + if p.cfg.Cleanup { + p.cfg.Log.Info("cleaning up cargo cache", "dir", p.cfg.CacheDir) + if err := os.RemoveAll(p.cfg.CacheDir); err != nil { + errs = append(errs, fmt.Errorf("cleaning cargo cache: %w", err)) + } + } + + return errors.Join(errs...) +} + +// Addr returns the address the proxy is listening on. +func (p *Proxy) Addr() string { + if p.listener != nil { + return p.listener.Addr().String() + } + return p.cfg.ListenAddr +} + +// advertiseBase is the base URL containers use to reach this proxy. It is +// built from the CONFIGURED address, not the bound one: after a wildcard +// fallback the bound address is "[::]:8083", which means nothing inside a +// container. +func (p *Proxy) advertiseBase() string { + addr := p.cfg.ListenAddr + if addr == "" { + addr = p.Addr() + } + return "http://" + addr +} + +// EnvVars returns environment variables to inject into job containers. +// +// Only rustup is configured here. Cargo's registry redirect cannot be done +// with environment variables (see Mounts). +func (p *Proxy) EnvVars() []string { + return []string{ + "RUSTUP_DIST_SERVER=" + p.advertiseBase() + rustupRoute, + } +} + +// Mounts returns the read-only bind mount that carries the generated Cargo +// config into job containers. +func (p *Proxy) Mounts() []proxies.Mount { + return []proxies.Mount{{ + Source: p.hostConfigDir(), + Destination: containerConfigDest(p.containerOS()), + ReadOnly: true, + }} +} + +// Name returns the proxy name for logging. +func (p *Proxy) Name() string { return "cargo" } + +func (p *Proxy) containerOS() string { + if p.cfg.ContainerOS != "" { + return p.cfg.ContainerOS + } + return hostGOOS +} + +// hostConfigDir is the host-side directory that is mounted as the +// container's ".cargo" directory. +func (p *Proxy) hostConfigDir() string { + return filepath.Join(p.cfg.ConfDir, ".cargo") +} + +// writeContainerConfig regenerates the container-side Cargo config. Written +// atomically so a job that starts mid-write never sees a truncated file. +func (p *Proxy) writeContainerConfig() error { + dir := p.hostConfigDir() + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("creating cargo conf dir %q: %w", dir, err) + } + target := filepath.Join(dir, "config.toml") + tmp, err := os.CreateTemp(dir, ".config-*.toml") + if err != nil { + return fmt.Errorf("creating temp cargo config: %w", err) + } + if _, err := tmp.WriteString(containerConfigTOML(p.advertiseBase())); err != nil { + _ = tmp.Close() + _ = os.Remove(tmp.Name()) + return fmt.Errorf("writing temp cargo config: %w", err) + } + if err := tmp.Close(); err != nil { + _ = os.Remove(tmp.Name()) + return fmt.Errorf("closing temp cargo config: %w", err) + } + if err := os.Chmod(tmp.Name(), 0o644); err != nil { + p.cfg.Log.Debug("chmod cargo config", "error", err) + } + if err := os.Rename(tmp.Name(), target); err != nil { + _ = os.Remove(tmp.Name()) + return fmt.Errorf("installing cargo config at %q: %w", target, err) + } + p.cfg.Log.Info("wrote container cargo config", "path", target, "mount", containerConfigDest(p.containerOS())) + return nil +} + +// --- routing --------------------------------------------------------------- + +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 + } + + path := r.URL.Path + switch { + case path == healthRoute: + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, "ok\n") + + case path == indexRoute+"/config.json": + p.handleRegistryConfig(w, r) + + case strings.HasPrefix(path, indexRoute+"/"): + p.handleIndex(w, r, strings.TrimPrefix(path, indexRoute)) + + case strings.HasPrefix(path, cratesRoute+"/"): + p.handleCrate(w, r) + + case strings.HasPrefix(path, rustupRoute+"/"): + p.handleRustup(w, r, strings.TrimPrefix(path, rustupRoute)) + + default: + http.NotFound(w, r) + } +} + +// handleRegistryConfig serves the registry's config.json with "dl" rewritten +// to point crate downloads at this proxy. The upstream copy is cached (and +// revalidated) like any other index entry, and its "dl" template is retained +// so crate requests can be mapped back to the real CDN. +func (p *Proxy) handleRegistryConfig(w http.ResponseWriter, r *http.Request) { + cachePath, err := indexCachePath(p.cfg.CacheDir, "/config.json") + if err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + body, _, err := p.fetchCached(r.Context(), cachePath, p.cfg.IndexUpstream+"/config.json", false, p.cfg.IndexTTL) + if err != nil { + p.cfg.Log.Warn("registry config.json unavailable", "error", err) + http.Error(w, "upstream error", http.StatusBadGateway) + return + } + + p.setDLTemplate(parseDLTemplate(body)) + + rewritten, err := rewriteConfigJSON(body, p.advertiseBase()) + if err != nil { + // Cannot rewrite: send the upstream config unchanged so cargo + // downloads straight from the CDN rather than failing. + p.cfg.Log.Warn("serving upstream registry config unrewritten", "error", err) + rewritten = body + } + + w.Header().Set("Content-Type", "application/json") + // No ETag: the body is ours, not upstream's, so upstream validators + // would be wrong. The document is tiny. + w.Header().Set("Cache-Control", "no-cache") + writeBody(w, r, rewritten) +} + +// handleIndex serves a sparse-index entry. Index data is MUTABLE, so a +// cached entry is served for IndexTTL and then revalidated with a +// conditional GET. +func (p *Proxy) handleIndex(w http.ResponseWriter, r *http.Request, indexPath string) { + cachePath, err := indexCachePath(p.cfg.CacheDir, indexPath) + if err != nil { + p.cfg.Log.Warn("rejecting index path", "path", indexPath, "error", err) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + body, meta, err := p.fetchCached(r.Context(), cachePath, p.cfg.IndexUpstream+indexPath, isImmutableIndexPath(indexPath), p.cfg.IndexTTL) + if err != nil { + if isNotFound(err) { + http.NotFound(w, r) + return + } + // Nothing cached and upstream is unreachable. Cargo treats this as + // fatal either way; report it honestly. + p.cfg.Log.Warn("index fetch failed", "path", indexPath, "error", err) + http.Error(w, "upstream error", http.StatusBadGateway) + return + } + + // Cheap client-side revalidation: Cargo keeps its own copy in CARGO_HOME + // and sends If-None-Match, so most warm-cache requests can be a 304. + if meta.ETag != "" { + w.Header().Set("ETag", meta.ETag) + if matchesETag(r.Header.Get("If-None-Match"), meta.ETag) { + w.WriteHeader(http.StatusNotModified) + return + } + } + if meta.LastModified != "" { + w.Header().Set("Last-Modified", meta.LastModified) + } + w.Header().Set("Content-Type", contentTypeOr(meta.ContentType, "text/plain; charset=utf-8")) + writeBody(w, r, body) +} + +// handleCrate serves a .crate tarball. Tarballs are IMMUTABLE — a published +// (crate, version) is never republished with different bytes — so a cached +// file is served forever and never revalidated. +func (p *Proxy) handleCrate(w http.ResponseWriter, r *http.Request) { + name, version, ok := parseCrateDownloadPath(r.URL.Path) + if !ok { + p.cfg.Log.Warn("rejecting crate download path", "path", r.URL.Path) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + cachePath, err := crateCachePath(p.cfg.CacheDir, name, version) + if err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + upstreamURL, ok := expandDL(p.dlTemplateOrDefault(), name, version) + if !ok { + upstreamURL, _ = expandDL(defaultDLTemplate, name, version) + } + + body, _, err := p.fetchCached(r.Context(), cachePath, upstreamURL, true, 0) + if err != nil { + if isNotFound(err) { + http.NotFound(w, r) + return + } + // FAIL OPEN: we could not fetch it, but the container can reach the + // internet directly. Redirect to the real CDN so the build proceeds + // (slower, uncached) instead of failing. + p.cfg.Log.Warn("crate fetch failed; redirecting job to upstream", + "crate", name, "version", version, "upstream", upstreamURL, "error", err) + http.Redirect(w, r, upstreamURL, http.StatusTemporaryRedirect) + return + } + + w.Header().Set("Content-Type", "application/x-tar") + // Immutable by construction — let Cargo and any intermediary cache hard. + w.Header().Set("Cache-Control", "public, max-age=31536000, immutable") + writeBody(w, r, body) +} + +// handleRustup serves rustup distribution artifacts. Dated artifacts are +// immutable; channel manifests are revalidated on the index TTL. +func (p *Proxy) handleRustup(w http.ResponseWriter, r *http.Request, distPath string) { + cachePath, err := rustupCachePath(p.cfg.CacheDir, distPath) + if err != nil { + p.cfg.Log.Warn("rejecting rustup path", "path", distPath, "error", err) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + upstreamURL := p.cfg.RustupUpstream + distPath + immutable := isImmutableRustupPath(distPath) + + body, meta, err := p.fetchCached(r.Context(), cachePath, upstreamURL, immutable, p.cfg.IndexTTL) + if err != nil { + if isNotFound(err) { + http.NotFound(w, r) + return + } + // FAIL OPEN, same reasoning as crate tarballs. + p.cfg.Log.Warn("rustup fetch failed; redirecting job to upstream", + "path", distPath, "error", err) + http.Redirect(w, r, upstreamURL, http.StatusTemporaryRedirect) + return + } + + if meta.ETag != "" { + w.Header().Set("ETag", meta.ETag) + } + w.Header().Set("Content-Type", contentTypeOr(meta.ContentType, "application/octet-stream")) + if immutable { + w.Header().Set("Cache-Control", "public, max-age=31536000, immutable") + } + writeBody(w, r, body) +} + +// --- cache core ------------------------------------------------------------ + +// notFoundError marks an upstream 404/410 so handlers can pass it through as +// a 404 rather than treating it as an outage. +type notFoundError struct{ url string } + +func (e *notFoundError) Error() string { return "upstream 404 for " + e.url } + +func isNotFound(err error) bool { + var nf *notFoundError + return errors.As(err, &nf) +} + +// fetchCached is the single read path for every route. +// +// It resolves the freshness decision (decideFreshness), then: +// +// - serveCached → return the on-disk bytes, no network at all; +// - revalidate → conditional GET; 304 refreshes the timestamp and the +// cached bytes are returned; 200 replaces them; +// - fetchFresh → unconditional GET, then store. +// +// FAIL-OPEN AT THIS LAYER: if the upstream request errors or returns 5xx and +// we HAVE a cached copy, the stale copy is served rather than propagating +// the failure. A registry outage then degrades to "slightly stale index" +// instead of a red CI job. Callers layer a second fail-open on top (a +// redirect to the origin) for the case where nothing is cached. +func (p *Proxy) fetchCached(ctx context.Context, cachePath, upstreamURL string, immutable bool, ttl time.Duration) ([]byte, entryMeta, error) { + // Fast path: a fresh (or immutable) hit needs no lock and no network. + if body, meta, ok := p.readCache(cachePath); ok { + if decideFreshness(true, immutable, meta.Fetched, time.Now(), ttl) == serveCached { + p.cfg.Log.Debug("cargo cache hit", "url", upstreamURL, "immutable", immutable) + return body, meta, nil + } + } + + // Collapse concurrent misses for the same object: without this, ten + // parallel jobs starting together each pull the same 40 MB toolchain. + mu := p.lockFor(cachePath) + mu.Lock() + defer mu.Unlock() + + body, meta, cached := p.readCache(cachePath) + decision := decideFreshness(cached, immutable, meta.Fetched, time.Now(), ttl) + if decision == serveCached { + p.cfg.Log.Debug("cargo cache hit (after lock)", "url", upstreamURL) + return body, meta, nil + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil) + if err != nil { + if cached { + return body, meta, nil + } + return nil, entryMeta{}, fmt.Errorf("building upstream request for %s: %w", upstreamURL, err) + } + if decision == revalidate { + if meta.ETag != "" { + req.Header.Set("If-None-Match", meta.ETag) + } + if meta.LastModified != "" { + req.Header.Set("If-Modified-Since", meta.LastModified) + } + } + + p.cfg.Log.Debug("cargo cache miss", "url", upstreamURL, "decision", decision.String()) + resp, err := p.client.Do(req) + if err != nil { + if cached { + p.cfg.Log.Warn("upstream unreachable; serving stale cache", "url", upstreamURL, "error", err) + return body, meta, nil + } + return nil, entryMeta{}, fmt.Errorf("fetching %s: %w", upstreamURL, err) + } + defer func() { + if cerr := resp.Body.Close(); cerr != nil { + p.cfg.Log.Debug("closing upstream body", "error", cerr) + } + }() + + switch { + case resp.StatusCode == http.StatusNotModified && cached: + meta.Fetched = time.Now() + p.writeMeta(cachePath, meta) + return body, meta, nil + + case resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone: + return nil, entryMeta{}, ¬FoundError{url: upstreamURL} + + case resp.StatusCode != http.StatusOK: + if cached { + p.cfg.Log.Warn("upstream error; serving stale cache", + "url", upstreamURL, "status", resp.StatusCode) + return body, meta, nil + } + return nil, entryMeta{}, fmt.Errorf("fetching %s: upstream status %d", upstreamURL, resp.StatusCode) + } + + fresh, err := io.ReadAll(resp.Body) + if err != nil { + if cached { + p.cfg.Log.Warn("upstream read failed; serving stale cache", "url", upstreamURL, "error", err) + return body, meta, nil + } + return nil, entryMeta{}, fmt.Errorf("reading %s: %w", upstreamURL, err) + } + + newMeta := entryMeta{ + ETag: resp.Header.Get("ETag"), + LastModified: resp.Header.Get("Last-Modified"), + ContentType: resp.Header.Get("Content-Type"), + Fetched: time.Now(), + } + // A write failure must not fail the request — the bytes are already in + // hand, the job just loses the caching benefit. + if err := p.writeCache(cachePath, fresh, newMeta, immutable); err != nil { + p.cfg.Log.Warn("caching response failed; serving anyway", "url", upstreamURL, "error", err) + } + return fresh, newMeta, nil +} + +// readCache loads a cached body and its sidecar metadata. A body with no +// sidecar is still usable (it reports a zero Fetched time, so mutable +// entries revalidate and immutable ones are served as-is). +func (p *Proxy) readCache(cachePath string) ([]byte, entryMeta, bool) { + body, err := os.ReadFile(cachePath) + if err != nil { + return nil, entryMeta{}, false + } + var meta entryMeta + if raw, err := os.ReadFile(metaPath(cachePath)); err == nil { + if err := json.Unmarshal(raw, &meta); err != nil { + p.cfg.Log.Debug("discarding unreadable cache metadata", "path", cachePath, "error", err) + meta = entryMeta{} + } + } + return body, meta, true +} + +// writeCache stores a body and its metadata atomically (write temp, rename) +// so a concurrent reader never sees a partial file. Immutable entries get no +// sidecar: there is nothing to revalidate, and skipping it halves the inode +// count for the tarball cache. +func (p *Proxy) writeCache(cachePath string, body []byte, meta entryMeta, immutable bool) error { + dir := filepath.Dir(cachePath) + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("creating cache dir %q: %w", dir, err) + } + if err := atomicWrite(cachePath, body); err != nil { + return err + } + if immutable { + return nil + } + p.writeMeta(cachePath, meta) + return nil +} + +// writeMeta persists the sidecar. Best-effort: a missing sidecar only costs +// a revalidation next time. +func (p *Proxy) writeMeta(cachePath string, meta entryMeta) { + raw, err := json.Marshal(meta) + if err != nil { + p.cfg.Log.Debug("encoding cache metadata", "path", cachePath, "error", err) + return + } + if err := atomicWrite(metaPath(cachePath), raw); err != nil { + p.cfg.Log.Debug("writing cache metadata", "path", cachePath, "error", err) + } +} + +func atomicWrite(target string, data []byte) error { + dir := filepath.Dir(target) + tmp, err := os.CreateTemp(dir, ".cargoproxy-*") + if err != nil { + return fmt.Errorf("creating temp file in %q: %w", dir, err) + } + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + _ = os.Remove(tmp.Name()) + return fmt.Errorf("writing %q: %w", target, err) + } + if err := tmp.Close(); err != nil { + _ = os.Remove(tmp.Name()) + return fmt.Errorf("closing temp file for %q: %w", target, err) + } + if err := os.Rename(tmp.Name(), target); err != nil { + _ = os.Remove(tmp.Name()) + return fmt.Errorf("renaming temp file to %q: %w", target, err) + } + return nil +} + +func (p *Proxy) lockFor(key string) *sync.Mutex { + v, _ := p.inflight.LoadOrStore(key, &sync.Mutex{}) + return v.(*sync.Mutex) +} + +func (p *Proxy) setDLTemplate(tmpl string) { + if tmpl == "" { + return + } + p.mu.Lock() + p.dlTemplate = tmpl + p.mu.Unlock() +} + +func (p *Proxy) dlTemplateOrDefault() string { + p.mu.RLock() + defer p.mu.RUnlock() + if p.dlTemplate == "" { + return defaultDLTemplate + } + return p.dlTemplate +} + +// loadDLTemplateFromCache recovers the upstream "dl" template across a +// daemon restart. Cargo caches config.json in CARGO_HOME too, so it may go +// straight to a crate download without asking us for config.json first. +func (p *Proxy) loadDLTemplateFromCache() { + cachePath, err := indexCachePath(p.cfg.CacheDir, "/config.json") + if err != nil { + return + } + raw, err := os.ReadFile(cachePath) + if err != nil { + return + } + p.setDLTemplate(parseDLTemplate(raw)) +} + +// --- small helpers --------------------------------------------------------- + +// writeBody writes the response body, honouring HEAD (headers only). +func writeBody(w http.ResponseWriter, r *http.Request, body []byte) { + w.Header().Set("Content-Length", fmt.Sprintf("%d", len(body))) + if r.Method == http.MethodHead { + w.WriteHeader(http.StatusOK) + return + } + w.WriteHeader(http.StatusOK) + if _, err := w.Write(body); err != nil { + // The client went away mid-response; nothing to do but note it. + _ = err + } +} + +// matchesETag reports whether an If-None-Match header matches the given tag. +// Handles the "*" wildcard, comma-separated lists, and the W/ weak prefix. +func matchesETag(ifNoneMatch, tag string) bool { + if ifNoneMatch == "" || tag == "" { + return false + } + if strings.TrimSpace(ifNoneMatch) == "*" { + return true + } + want := strings.TrimPrefix(tag, "W/") + for _, candidate := range strings.Split(ifNoneMatch, ",") { + if strings.TrimPrefix(strings.TrimSpace(candidate), "W/") == want { + return true + } + } + return false +} + +func contentTypeOr(got, fallback string) string { + if got == "" { + return fallback + } + return got +} diff --git a/pkg/proxies/cargo/cargoproxy_test.go b/pkg/proxies/cargo/cargoproxy_test.go new file mode 100644 index 00000000..10a18e17 --- /dev/null +++ b/pkg/proxies/cargo/cargoproxy_test.go @@ -0,0 +1,678 @@ +package cargoproxy + +import ( + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// NOTE: every test here runs against an httptest fake upstream. Nothing in +// this file touches crates.io, static.rust-lang.org, or any other network. + +// alwaysRevalidate is a negative TTL, which Config documents as "never serve +// a mutable entry without a conditional GET". Tests use it to reach the +// aged-out state deterministically instead of sleeping. A ZERO TTL will not +// do: New() reads zero as "unset" and substitutes DefaultIndexTTL. +const alwaysRevalidate = -1 * time.Second + +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// fakeUpstream is a stand-in for index.crates.io / static.crates.io / +// static.rust-lang.org, with per-path hit counting so tests can assert that +// the cache actually prevented a second fetch. +type fakeUpstream struct { + *httptest.Server + + mu sync.Mutex + hits map[string]int + bodies map[string]string + etags map[string]string + status map[string]int + failing atomic.Bool + condHits atomic.Int64 // conditional (If-None-Match) requests received +} + +func newFakeUpstream(t *testing.T) *fakeUpstream { + t.Helper() + f := &fakeUpstream{ + hits: map[string]int{}, + bodies: map[string]string{}, + etags: map[string]string{}, + status: map[string]int{}, + } + f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + f.hits[r.URL.Path]++ + body, known := f.bodies[r.URL.Path] + etag := f.etags[r.URL.Path] + status := f.status[r.URL.Path] + f.mu.Unlock() + + if f.failing.Load() { + http.Error(w, "upstream down", http.StatusInternalServerError) + return + } + if status != 0 { + w.WriteHeader(status) + return + } + if !known { + http.NotFound(w, r) + return + } + if inm := r.Header.Get("If-None-Match"); inm != "" { + f.condHits.Add(1) + if etag != "" && inm == etag { + w.Header().Set("ETag", etag) + w.WriteHeader(http.StatusNotModified) + return + } + } + if etag != "" { + w.Header().Set("ETag", etag) + } + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + _, _ = io.WriteString(w, body) + })) + t.Cleanup(f.Close) + return f +} + +func (f *fakeUpstream) set(path, body, etag string) { + f.mu.Lock() + defer f.mu.Unlock() + f.bodies[path] = body + if etag != "" { + f.etags[path] = etag + } +} + +func (f *fakeUpstream) setStatus(path string, code int) { + f.mu.Lock() + defer f.mu.Unlock() + f.status[path] = code +} + +func (f *fakeUpstream) hitCount(path string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.hits[path] +} + +// newTestProxy starts a proxy bound to loopback with both upstreams pointed +// at the fake, and returns it plus its base URL. +func newTestProxy(t *testing.T, up *fakeUpstream, mutate func(*Config)) (*Proxy, string) { + t.Helper() + dir := t.TempDir() + cfg := Config{ + CacheDir: filepath.Join(dir, "cache"), + ConfDir: filepath.Join(dir, "conf"), + IndexUpstream: up.URL, + RustupUpstream: up.URL, + ListenAddr: "127.0.0.1:0", + IndexTTL: time.Hour, + ContainerOS: "linux", + Log: discardLogger(), + } + if mutate != nil { + mutate(&cfg) + } + p := New(cfg) + 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, "http://" + p.Addr() +} + +func get(t *testing.T, url string, hdr map[string]string) (*http.Response, string) { + t.Helper() + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + t.Fatalf("building request for %s: %v", url, err) + } + for k, v := range hdr { + req.Header.Set(k, v) + } + // Do not follow redirects: the fail-open path is a 307 we want to assert. + client := &http.Client{ + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + Timeout: 10 * time.Second, + } + resp, err := client.Do(req) + if err != nil { + t.Fatalf("GET %s: %v", url, err) + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("reading body of %s: %v", url, err) + } + return resp, string(body) +} + +// --- index (mutable) ------------------------------------------------------- + +// TestIndex_CachedWithinTTL: the sparse index is mutable, but inside the TTL +// it must be served from disk with no upstream request at all. +func TestIndex_CachedWithinTTL(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", `{"name":"serde","vers":"1.0.203"}`, `"v1"`) + _, base := newTestProxy(t, up, nil) + + for i := range 3 { + resp, body := get(t, base+"/index/se/rd/serde", nil) + if resp.StatusCode != http.StatusOK { + t.Fatalf("request %d: status = %d, want 200", i, resp.StatusCode) + } + if !strings.Contains(body, `"serde"`) { + t.Fatalf("request %d: body = %q", i, body) + } + } + if got := up.hitCount("/se/rd/serde"); got != 1 { + t.Errorf("upstream hits = %d, want 1 (later requests must come from cache)", got) + } +} + +// TestIndex_RevalidatesAfterTTL: past the TTL the proxy must NOT serve stale +// data blindly — it must issue a conditional GET, and it must pick up new +// content when the entry actually changed. +func TestIndex_RevalidatesAfterTTL(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", "v1-body", `"v1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + if _, body := get(t, base+"/index/se/rd/serde", nil); body != "v1-body" { + t.Fatalf("first body = %q, want v1-body", body) + } + + // Unchanged upstream: the revalidation must be conditional and the + // cached bytes reused. + before := up.condHits.Load() + if _, body := get(t, base+"/index/se/rd/serde", nil); body != "v1-body" { + t.Fatalf("second body = %q, want v1-body", body) + } + if up.condHits.Load() <= before { + t.Error("revalidation was not conditional (no If-None-Match sent)") + } + + // Changed upstream: a new version was published, the proxy must see it. + up.set("/se/rd/serde", "v2-body", `"v2"`) + if _, body := get(t, base+"/index/se/rd/serde", nil); body != "v2-body" { + t.Errorf("after upstream change body = %q, want v2-body (stale index served)", body) + } +} + +// TestIndex_ServesClient304 keeps warm-cache traffic cheap: Cargo keeps its +// own copy and sends If-None-Match, which must produce a bodyless 304. +func TestIndex_ServesClient304(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", "body", `"v1"`) + _, base := newTestProxy(t, up, nil) + + resp, _ := get(t, base+"/index/se/rd/serde", nil) + etag := resp.Header.Get("ETag") + if etag == "" { + t.Fatal("proxy did not surface the upstream ETag") + } + + resp2, body2 := get(t, base+"/index/se/rd/serde", map[string]string{"If-None-Match": etag}) + if resp2.StatusCode != http.StatusNotModified { + t.Errorf("status = %d, want 304", resp2.StatusCode) + } + if body2 != "" { + t.Errorf("304 carried a body: %q", body2) + } +} + +// TestIndex_FailOpenToStaleCache: a registry outage must not turn into a red +// CI job when we already hold the data. +func TestIndex_FailOpenToStaleCache(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", "cached-body", `"v1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + if _, body := get(t, base+"/index/se/rd/serde", nil); body != "cached-body" { + t.Fatalf("priming failed: %q", body) + } + + up.failing.Store(true) + resp, body := get(t, base+"/index/se/rd/serde", nil) + if resp.StatusCode != http.StatusOK { + t.Errorf("status = %d, want 200 (stale cache should be served)", resp.StatusCode) + } + if body != "cached-body" { + t.Errorf("body = %q, want the stale cached copy", body) + } +} + +// TestIndex_UpstreamNotFoundIsPassedThrough: a crate that does not exist is +// a 404, not a 502 — Cargo needs to distinguish them. +func TestIndex_UpstreamNotFoundIsPassedThrough(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + + resp, _ := get(t, base+"/index/no/pe/nope", nil) + if resp.StatusCode != http.StatusNotFound { + t.Errorf("status = %d, want 404", resp.StatusCode) + } +} + +// TestIndex_RejectsTraversal keeps a hostile request from writing outside +// the cache root. +func TestIndex_RejectsTraversal(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + + // http.Client normalises "..", so build the raw request by hand. + resp, _ := get(t, base+"/index/se/rd/%2e%2e%2f%2e%2e%2fescape", nil) + if resp.StatusCode != http.StatusBadRequest && resp.StatusCode != http.StatusNotFound { + t.Errorf("status = %d, want 400 or 404 for a traversal attempt", resp.StatusCode) + } +} + +// --- config.json rewrite --------------------------------------------------- + +func TestRegistryConfig_RewritesDLToProxy(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+"UPSTREAM"+`/crates/{crate}/{crate}-{version}.crate","api":"https://crates.io"}`, "") + p, base := newTestProxy(t, up, nil) + + resp, body := get(t, base+"/index/config.json", nil) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + if !strings.Contains(body, `"dl":"`+p.advertiseBase()+`/crates"`) { + t.Errorf("dl was not rewritten to the proxy: %s", body) + } + if !strings.Contains(body, `"api":"https://crates.io"`) { + t.Errorf("api must be preserved so cargo publish still works: %s", body) + } +} + +// --- crate tarballs (immutable) -------------------------------------------- + +func TestCrate_CachedForeverAfterFirstFetch(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + up.set("/crates/serde/serde-1.0.203.crate", "TARBALL-BYTES", `"e1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + // Cargo asks for config.json first; that is what teaches us the dl template. + if resp, _ := get(t, base+"/index/config.json", nil); resp.StatusCode != http.StatusOK { + t.Fatalf("config.json status = %d", resp.StatusCode) + } + + for i := range 3 { + resp, body := get(t, base+"/crates/serde/1.0.203/download", nil) + if resp.StatusCode != http.StatusOK { + t.Fatalf("request %d: status = %d, want 200", i, resp.StatusCode) + } + if body != "TARBALL-BYTES" { + t.Fatalf("request %d: body = %q", i, body) + } + } + + // Immutable content: exactly one upstream fetch, and no revalidation + // even though the index TTL is zero. + if got := up.hitCount("/crates/serde/serde-1.0.203.crate"); got != 1 { + t.Errorf("upstream hits = %d, want 1 — .crate tarballs are immutable and must never be refetched", got) + } +} + +// TestCrate_ImmutableCacheSurvivesUpstreamOutage: once cached, a tarball is +// served with the network completely gone. +func TestCrate_ImmutableCacheSurvivesUpstreamOutage(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + up.set("/crates/serde/serde-1.0.203.crate", "TARBALL-BYTES", "") + _, base := newTestProxy(t, up, nil) + + get(t, base+"/index/config.json", nil) + get(t, base+"/crates/serde/1.0.203/download", nil) + + up.failing.Store(true) + resp, body := get(t, base+"/crates/serde/1.0.203/download", nil) + if resp.StatusCode != http.StatusOK || body != "TARBALL-BYTES" { + t.Errorf("status = %d body = %q, want a cache hit despite the outage", resp.StatusCode, body) + } +} + +// TestCrate_FailsOpenWithRedirect is the core fail-open guarantee: when the +// proxy cannot produce the tarball, the job must be sent to the real CDN +// rather than handed an error. +func TestCrate_FailsOpenWithRedirect(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + _, base := newTestProxy(t, up, nil) + get(t, base+"/index/config.json", nil) + + // 500 from upstream with nothing cached. + up.setStatus("/crates/serde/serde-9.9.9.crate", http.StatusInternalServerError) + + resp, _ := get(t, base+"/crates/serde/9.9.9/download", nil) + if resp.StatusCode != http.StatusTemporaryRedirect { + t.Fatalf("status = %d, want 307 (fail open to upstream)", resp.StatusCode) + } + loc := resp.Header.Get("Location") + if !strings.HasSuffix(loc, "/crates/serde/serde-9.9.9.crate") { + t.Errorf("Location = %q, want the upstream tarball URL", loc) + } +} + +// TestCrate_UnknownVersionIs404 keeps a genuine "no such version" distinct +// from an outage — redirecting on a 404 would just make Cargo fetch a 404. +func TestCrate_UnknownVersionIs404(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + _, base := newTestProxy(t, up, nil) + get(t, base+"/index/config.json", nil) + + resp, _ := get(t, base+"/crates/serde/0.0.0/download", nil) + if resp.StatusCode != http.StatusNotFound { + t.Errorf("status = %d, want 404", resp.StatusCode) + } +} + +func TestCrate_RejectsMalformedPaths(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + + for _, p := range []string{ + "/crates/serde/1.0.0", + "/crates/serde/1.0.0/fetch", + "/crates/serde%20x/1.0.0/download", + } { + resp, _ := get(t, base+p, nil) + if resp.StatusCode == http.StatusOK { + t.Errorf("GET %s = 200, want a rejection", p) + } + } +} + +// TestCrate_ConcurrentMissesFetchUpstreamOnce pins the single-flight +// behaviour that stops N parallel jobs pulling the same tarball N times. +func TestCrate_ConcurrentMissesFetchUpstreamOnce(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/crates/{crate}/{crate}-{version}.crate"}`, "") + up.set("/crates/tokio/tokio-1.2.3.crate", "BIG-TARBALL", "") + _, base := newTestProxy(t, up, nil) + get(t, base+"/index/config.json", nil) + + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + resp, err := http.Get(base + "/crates/tokio/1.2.3/download") + if err != nil { + return + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + }() + } + wg.Wait() + + if got := up.hitCount("/crates/tokio/tokio-1.2.3.crate"); got != 1 { + t.Errorf("upstream hits = %d, want 1 (concurrent misses must be collapsed)", got) + } +} + +// --- rustup ---------------------------------------------------------------- + +func TestRustup_DatedArtifactCachedForever(t *testing.T) { + up := newFakeUpstream(t) + up.set("/dist/2026-08-01/rust-std-nightly.tar.xz", "TOOLCHAIN", `"e1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + for range 3 { + resp, body := get(t, base+"/rustup/dist/2026-08-01/rust-std-nightly.tar.xz", nil) + if resp.StatusCode != http.StatusOK || body != "TOOLCHAIN" { + t.Fatalf("status = %d body = %q", resp.StatusCode, body) + } + } + if got := up.hitCount("/dist/2026-08-01/rust-std-nightly.tar.xz"); got != 1 { + t.Errorf("upstream hits = %d, want 1 (dated toolchain artifacts are immutable)", got) + } +} + +func TestRustup_ChannelManifestRevalidates(t *testing.T) { + up := newFakeUpstream(t) + up.set("/dist/channel-rust-nightly.toml", "manifest-v1", `"m1"`) + _, base := newTestProxy(t, up, func(c *Config) { c.IndexTTL = alwaysRevalidate }) + + if _, body := get(t, base+"/rustup/dist/channel-rust-nightly.toml", nil); body != "manifest-v1" { + t.Fatalf("first fetch = %q", body) + } + // The channel manifest is rewritten daily in place; the proxy must not + // pin jobs to yesterday's nightly. + up.set("/dist/channel-rust-nightly.toml", "manifest-v2", `"m2"`) + if _, body := get(t, base+"/rustup/dist/channel-rust-nightly.toml", nil); body != "manifest-v2" { + t.Errorf("second fetch = %q, want manifest-v2 (stale channel manifest served)", body) + } +} + +func TestRustup_FailsOpenWithRedirect(t *testing.T) { + up := newFakeUpstream(t) + up.setStatus("/dist/2026-08-01/x.tar.xz", http.StatusInternalServerError) + _, base := newTestProxy(t, up, nil) + + resp, _ := get(t, base+"/rustup/dist/2026-08-01/x.tar.xz", nil) + if resp.StatusCode != http.StatusTemporaryRedirect { + t.Fatalf("status = %d, want 307 (fail open)", resp.StatusCode) + } + if want := up.URL + "/dist/2026-08-01/x.tar.xz"; resp.Header.Get("Location") != want { + t.Errorf("Location = %q, want %q", resp.Header.Get("Location"), want) + } +} + +// --- lifecycle / wiring ---------------------------------------------------- + +// TestStart_WritesContainerConfigAndDeclaresMount is what makes the proxy +// actually get used: without the generated file (and the mount that carries +// it) Cargo silently keeps talking to crates.io. +func TestStart_WritesContainerConfigAndDeclaresMount(t *testing.T) { + up := newFakeUpstream(t) + p, _ := newTestProxy(t, up, nil) + + mounts := p.Mounts() + if len(mounts) != 1 { + t.Fatalf("Mounts() = %d entries, want 1", len(mounts)) + } + m := mounts[0] + if m.Destination != "/.cargo" { + t.Errorf("mount destination = %q, want /.cargo", m.Destination) + } + if !m.ReadOnly { + t.Error("the cargo config mount must be read-only — a job must not rewrite the next job's config") + } + + raw, err := os.ReadFile(filepath.Join(m.Source, "config.toml")) + if err != nil { + t.Fatalf("reading generated config: %v", err) + } + if !strings.Contains(string(raw), `replace-with = "ephemerd"`) { + t.Errorf("generated config is not a source replacement:\n%s", raw) + } + if !strings.Contains(string(raw), "sparse+"+p.advertiseBase()+"/index/") { + t.Errorf("generated config does not point at this proxy:\n%s", raw) + } +} + +// TestConfigDirIsOutsideCacheDir: `ephemerd cache clear cargo` wipes the +// cache root. If the mounted config lived there it would vanish from under a +// running job. +func TestConfigDirIsOutsideCacheDir(t *testing.T) { + up := newFakeUpstream(t) + p, _ := newTestProxy(t, up, nil) + + confAbs, err := filepath.Abs(p.Mounts()[0].Source) + if err != nil { + t.Fatalf("abs conf dir: %v", err) + } + cacheAbs, err := filepath.Abs(p.cfg.CacheDir) + if err != nil { + t.Fatalf("abs cache dir: %v", err) + } + rel, err := filepath.Rel(cacheAbs, confAbs) + if err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + t.Errorf("container config dir %q is inside the clearable cache dir %q", confAbs, cacheAbs) + } +} + +func TestEnvVars_PointRustupAtTheProxy(t *testing.T) { + up := newFakeUpstream(t) + p, _ := newTestProxy(t, up, nil) + + env := p.EnvVars() + if len(env) != 1 { + t.Fatalf("EnvVars() = %v, want exactly the rustup dist server", env) + } + want := "RUSTUP_DIST_SERVER=" + p.advertiseBase() + "/rustup" + if env[0] != want { + t.Errorf("EnvVars()[0] = %q, want %q", env[0], want) + } +} + +// TestAdvertiseBase_UsesConfiguredAddress: after a wildcard fallback the +// bound address is "[::]:port", which is useless inside a container. The +// advertised address must stay the configured gateway address. +func TestAdvertiseBase_UsesConfiguredAddress(t *testing.T) { + p := New(Config{ListenAddr: "10.88.0.1:8083", Log: discardLogger()}) + if got := p.advertiseBase(); got != "http://10.88.0.1:8083" { + t.Errorf("advertiseBase() = %q, want http://10.88.0.1:8083", got) + } +} + +func TestStop_CleanupHonoursConfig(t *testing.T) { + up := newFakeUpstream(t) + up.set("/se/rd/serde", "body", "") + + t.Run("cleanup disabled keeps the cache", func(t *testing.T) { + dir := t.TempDir() + p := New(Config{ + CacheDir: filepath.Join(dir, "cache"), ConfDir: filepath.Join(dir, "conf"), + IndexUpstream: up.URL, RustupUpstream: up.URL, + ListenAddr: "127.0.0.1:0", ContainerOS: "linux", Cleanup: false, Log: discardLogger(), + }) + if err := p.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + get(t, "http://"+p.Addr()+"/index/se/rd/serde", nil) + if err := p.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + if _, err := os.Stat(filepath.Join(dir, "cache", "index", "se", "rd", "serde")); err != nil { + t.Errorf("cache was wiped despite cleanup=false: %v", err) + } + }) + + t.Run("cleanup enabled wipes the cache", func(t *testing.T) { + dir := t.TempDir() + p := New(Config{ + CacheDir: filepath.Join(dir, "cache"), ConfDir: filepath.Join(dir, "conf"), + IndexUpstream: up.URL, RustupUpstream: up.URL, + ListenAddr: "127.0.0.1:0", ContainerOS: "linux", Cleanup: true, Log: discardLogger(), + }) + if err := p.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + get(t, "http://"+p.Addr()+"/index/se/rd/serde", nil) + if err := p.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + if _, err := os.Stat(filepath.Join(dir, "cache")); !os.IsNotExist(err) { + t.Errorf("cache survived cleanup=true (err=%v)", err) + } + }) +} + +// TestDLTemplateSurvivesRestart: Cargo caches config.json in CARGO_HOME, so +// after a daemon restart it may go straight to a crate download. The proxy +// must recover the upstream dl template from its own cache. +func TestDLTemplateSurvivesRestart(t *testing.T) { + up := newFakeUpstream(t) + up.set("/config.json", `{"dl":"`+up.URL+`/mirror/{crate}/{version}.crate"}`, "") + up.set("/mirror/serde/1.0.203.crate", "TARBALL", "") + + dir := t.TempDir() + mk := func() *Proxy { + return New(Config{ + CacheDir: filepath.Join(dir, "cache"), ConfDir: filepath.Join(dir, "conf"), + IndexUpstream: up.URL, RustupUpstream: up.URL, + ListenAddr: "127.0.0.1:0", ContainerOS: "linux", Log: discardLogger(), + }) + } + + p1 := mk() + if err := p1.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + get(t, "http://"+p1.Addr()+"/index/config.json", nil) + if err := p1.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + + p2 := mk() + if err := p2.Start(); err != nil { + t.Fatalf("restart: %v", err) + } + defer func() { _ = p2.Stop() }() + + if got := p2.dlTemplateOrDefault(); got != up.URL+"/mirror/{crate}/{version}.crate" { + t.Fatalf("dl template after restart = %q, want the cached upstream template", got) + } + resp, body := get(t, "http://"+p2.Addr()+"/crates/serde/1.0.203/download", nil) + if resp.StatusCode != http.StatusOK || body != "TARBALL" { + t.Errorf("status = %d body = %q, want the tarball via the recovered template", resp.StatusCode, body) + } +} + +func TestHealthz(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + if resp, _ := get(t, base+healthRoute, nil); resp.StatusCode != http.StatusOK { + t.Errorf("healthz status = %d, want 200", resp.StatusCode) + } +} + +func TestUnknownRouteIs404(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + if resp, _ := get(t, base+"/nope", nil); resp.StatusCode != http.StatusNotFound { + t.Errorf("status = %d, want 404", resp.StatusCode) + } +} + +func TestNonGetMethodsRejected(t *testing.T) { + up := newFakeUpstream(t) + _, base := newTestProxy(t, up, nil) + + req, err := http.NewRequest(http.MethodPost, base+"/index/se/rd/serde", nil) + if err != nil { + t.Fatalf("building request: %v", err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("POST: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusMethodNotAllowed { + t.Errorf("status = %d, want 405", resp.StatusCode) + } +} diff --git a/pkg/proxies/go/goproxy.go b/pkg/proxies/go/goproxy.go index 4ba2e405..6fa639ed 100644 --- a/pkg/proxies/go/goproxy.go +++ b/pkg/proxies/go/goproxy.go @@ -28,10 +28,10 @@ import ( // Config for the Go module caching proxy. type Config struct { - CacheDir string // on-disk cache directory - Upstream string // upstream proxy URL (default: https://proxy.golang.org) - ListenAddr string // address to listen on (e.g., "10.88.0.1:8082") - Cleanup bool // wipe cache dir on Stop + CacheDir string // on-disk cache directory + Upstream string // upstream proxy URL (default: https://proxy.golang.org) + ListenAddr string // address to listen on (e.g., "10.88.0.1:8082") + Cleanup bool // wipe cache dir on Stop Log *slog.Logger } @@ -68,9 +68,13 @@ func (p *Proxy) Start() error { return fmt.Errorf("creating cache dir: %w", err) } - ln, err := net.Listen("tcp", p.cfg.ListenAddr) + // proxies.Listen (not net.Listen): the bridge gateway IP does not exist + // yet at daemon boot — CNI creates the ephemerd0 bridge with the first + // job container — so a direct bind fails with EADDRNOTAVAIL and the + // proxy silently never starts. See proxies.Listen for the full story. + ln, err := proxies.Listen(p.cfg.ListenAddr, p.cfg.Log) if err != nil { - return fmt.Errorf("listening on %s: %w", p.cfg.ListenAddr, err) + return err } p.listener = ln @@ -123,10 +127,28 @@ func (p *Proxy) Addr() string { } // EnvVars returns the environment variables to inject into job containers. +// +// The advertised address is the CONFIGURED one (the bridge gateway), not +// p.Addr(): when the listener falls back to the wildcard, Addr() reports +// "[::]:8082", which is meaningless inside a container. +// +// The "|" separator (not ",") is what makes this fail open. With ",direct" +// the go command only falls through to the origin on 404/410 — a proxy that +// is down, wedged, or returning 5xx hard-fails the build. With "|direct" it +// falls through on ANY error, so a broken cache degrades to a slower build +// instead of a red job. func (p *Proxy) EnvVars() []string { return []string{ - "GOPROXY=http://" + p.Addr() + ",direct", + "GOPROXY=http://" + p.advertiseAddr() + "|direct", + } +} + +// advertiseAddr is the address containers should use to reach this proxy. +func (p *Proxy) advertiseAddr() string { + if p.cfg.ListenAddr != "" { + return p.cfg.ListenAddr } + return p.Addr() } // Name returns the proxy name for logging. diff --git a/pkg/proxies/listen.go b/pkg/proxies/listen.go new file mode 100644 index 00000000..f19d8431 --- /dev/null +++ b/pkg/proxies/listen.go @@ -0,0 +1,79 @@ +package proxies + +import ( + "fmt" + "log/slog" + "net" +) + +// Listen binds a cache-proxy listener on addr, falling back to the wildcard +// address on the same port when addr itself cannot be bound. +// +// WHY THIS EXISTS (this is the bug that made the Go module proxy cache stay +// empty forever — see docs and the review notes on this change): +// +// Cache proxies are asked to listen on the CNI bridge gateway IP (e.g. +// 10.88.0.1:8082) so job containers can reach them. But ephemerd deletes the +// ephemerd0 bridge on shutdown and again on startup, and the CNI bridge +// plugin only (re)creates the bridge — and only then assigns the gateway IP +// to it — when the FIRST job container is networked. Proxies start long +// before that, during daemon boot, so net.Listen("tcp", "10.88.0.1:8082") +// fails with EADDRNOTAVAIL every single time. The caller logs a warning and +// carries on without the proxy, which means its env vars are never injected +// into any container: the cache directory is created and then never written +// to again. +// +// Binding the wildcard address fixes this without racing the bridge: a +// wildcard (INADDR_ANY) socket accepts connections addressed to interface +// addresses that appear AFTER the bind, so the gateway IP becomes reachable +// the moment CNI brings the bridge up. +// +// The requested address is always tried first, so a host where the gateway +// already exists keeps the narrower binding. When both fail (e.g. the port is +// already in use, which fails on the wildcard too) the ORIGINAL error is +// returned — the fallback never masks a real problem. +// +// SECURITY NOTE: the wildcard binding also exposes the proxy on the host's +// other interfaces. These proxies only ever serve public package-registry +// content and hold no credentials, so the exposure is limited to an open +// caching mirror. Operators who need it closed should firewall the port at +// the host edge; see docs/getting-started/configuration.md. +func Listen(addr string, log *slog.Logger) (net.Listener, error) { + ln, err := net.Listen("tcp", addr) + if err == nil { + return ln, nil + } + + wildcard, werr := wildcardAddr(addr) + if werr != nil { + return nil, fmt.Errorf("listening on %s: %w", addr, err) + } + + ln2, err2 := net.Listen("tcp", wildcard) + if err2 != nil { + // The wildcard failed too (port in use, permissions). Report the + // original failure — it is the one the operator configured for. + return nil, fmt.Errorf("listening on %s: %w", addr, err) + } + + if log != nil { + log.Info("cache proxy bound to wildcard address; the requested address is not on any interface yet (the CNI bridge is created with the first job container)", + "requested", addr, "bound", ln2.Addr().String(), "reason", err) + } + return ln2, nil +} + +// wildcardAddr rewrites a host:port address to the wildcard form ":port", +// which binds every interface. Returns an error for input that is not a +// valid host:port pair, so callers can fall back to reporting the original +// bind failure rather than listening somewhere unintended. +func wildcardAddr(addr string) (string, error) { + _, port, err := net.SplitHostPort(addr) + if err != nil { + return "", fmt.Errorf("splitting address %q: %w", addr, err) + } + if port == "" { + return "", fmt.Errorf("address %q has no port", addr) + } + return net.JoinHostPort("", port), nil +} diff --git a/pkg/proxies/listen_test.go b/pkg/proxies/listen_test.go new file mode 100644 index 00000000..a4fd7e23 --- /dev/null +++ b/pkg/proxies/listen_test.go @@ -0,0 +1,122 @@ +package proxies + +import ( + "io" + "log/slog" + "net" + "testing" +) + +func testLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +func TestWildcardAddr(t *testing.T) { + tests := []struct { + name string + addr string + want string + wantErr bool + }{ + {name: "gateway ipv4", addr: "10.88.0.1:8082", want: ":8082"}, + {name: "loopback", addr: "127.0.0.1:1", want: ":1"}, + {name: "already wildcard", addr: ":8083", want: ":8083"}, + {name: "zero port keeps zero", addr: "10.88.0.1:0", want: ":0"}, + {name: "ipv6 host", addr: "[fd00::1]:8082", want: ":8082"}, + {name: "hostname", addr: "gateway.internal:9000", want: ":9000"}, + {name: "no port", addr: "10.88.0.1", wantErr: true}, + {name: "empty", addr: "", wantErr: true}, + {name: "trailing colon only", addr: "10.88.0.1:", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := wildcardAddr(tt.addr) + if tt.wantErr { + if err == nil { + t.Fatalf("wildcardAddr(%q) = %q, want error", tt.addr, got) + } + return + } + if err != nil { + t.Fatalf("wildcardAddr(%q): %v", tt.addr, err) + } + if got != tt.want { + t.Errorf("wildcardAddr(%q) = %q, want %q", tt.addr, got, tt.want) + } + }) + } +} + +// TestListen_BindableAddressIsUsedDirectly pins that the fallback does not +// kick in when the requested address is actually assignable: loopback is +// always present, so the listener must end up on 127.0.0.1 and not on the +// wildcard. +func TestListen_BindableAddressIsUsedDirectly(t *testing.T) { + ln, err := Listen("127.0.0.1:0", testLogger()) + if err != nil { + t.Fatalf("Listen: %v", err) + } + defer func() { _ = ln.Close() }() + + host, _, err := net.SplitHostPort(ln.Addr().String()) + if err != nil { + t.Fatalf("SplitHostPort(%q): %v", ln.Addr().String(), err) + } + if host != "127.0.0.1" { + t.Errorf("bound host = %q, want 127.0.0.1 (fallback should not trigger)", host) + } +} + +// TestListen_FallsBackToWildcard is the regression test for the empty-cache +// bug: binding an address that is not on any interface (the bridge gateway +// before CNI creates the bridge) must NOT fail the proxy, it must fall back +// to the wildcard so the address works once the interface appears. +// +// 203.0.113.0/24 is TEST-NET-3 (RFC 5737) and is never assigned to a real +// interface, so this reproduces EADDRNOTAVAIL without any network access. +func TestListen_FallsBackToWildcard(t *testing.T) { + ln, err := Listen("203.0.113.1:0", testLogger()) + if err != nil { + t.Fatalf("Listen fell over instead of falling back: %v", err) + } + defer func() { _ = ln.Close() }() + + host, port, err := net.SplitHostPort(ln.Addr().String()) + if err != nil { + t.Fatalf("SplitHostPort(%q): %v", ln.Addr().String(), err) + } + if host == "203.0.113.1" { + t.Fatalf("bound the TEST-NET address %q — the test assumption is broken", ln.Addr()) + } + if port == "0" || port == "" { + t.Errorf("bound port = %q, want a real ephemeral port", port) + } + // A wildcard listener must be reachable over loopback. + c, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", port)) + if err != nil { + t.Fatalf("wildcard listener not reachable on loopback: %v", err) + } + _ = c.Close() +} + +// TestListen_ReturnsOriginalErrorWhenWildcardAlsoFails pins that the +// fallback never masks a genuine failure: when the port is already taken the +// wildcard bind fails too and the caller gets an error, not a silent success. +func TestListen_ReturnsOriginalErrorWhenWildcardAlsoFails(t *testing.T) { + // Occupy a port on the wildcard address so no further bind can succeed. + blocker, err := net.Listen("tcp", ":0") + if err != nil { + t.Fatalf("setting up blocker listener: %v", err) + } + defer func() { _ = blocker.Close() }() + _, port, err := net.SplitHostPort(blocker.Addr().String()) + if err != nil { + t.Fatalf("SplitHostPort: %v", err) + } + + ln, err := Listen(net.JoinHostPort("203.0.113.1", port), testLogger()) + if err == nil { + _ = ln.Close() + t.Skip("host allows rebinding an in-use wildcard port; cannot exercise this path here") + } +} diff --git a/pkg/proxies/proxy.go b/pkg/proxies/proxy.go index dbd5f242..bf20354a 100644 --- a/pkg/proxies/proxy.go +++ b/pkg/proxies/proxy.go @@ -6,6 +6,7 @@ // Implementations live in sub-packages: // // pkg/proxies/go/ — Go module proxy (GOPROXY) +// pkg/proxies/cargo/ — Cargo sparse registry + crate/rustup proxy // pkg/proxies/npm/ — (future) npm registry proxy // pkg/proxies/pip/ — (future) pip index proxy package proxies @@ -30,3 +31,28 @@ type CacheProxy interface { // Name returns a human-readable name for logging (e.g., "go", "npm"). Name() string } + +// Mount is a host→container bind mount a cache proxy needs in every job +// container. Env vars are enough for toolchains that take their proxy from +// the environment (Go, rustup); Cargo is not one of them — its source +// replacement is only read from a config file, never from CARGO_* env vars +// (verified empirically; see pkg/proxies/cargo). Such proxies generate the +// file on the host and declare it here. +type Mount struct { + // Source is an absolute host path (a directory). + Source string + // Destination is the absolute path inside the container. + Destination string + // ReadOnly mounts the source read-only. Config material should always + // be read-only: a job must never be able to rewrite what the next job + // on this host will read. + ReadOnly bool +} + +// MountProvider is an OPTIONAL interface a CacheProxy may implement when +// env vars alone cannot point a toolchain at the proxy. Callers type-assert +// for it; a proxy that does not implement it needs no mounts. +type MountProvider interface { + // Mounts returns the bind mounts to add to every job container spec. + Mounts() []Mount +} diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index b6bf8add..69024114 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -25,6 +25,7 @@ import ( "github.com/ephpm/ephemerd/pkg/config" "github.com/ephpm/ephemerd/pkg/dind" "github.com/ephpm/ephemerd/pkg/networking" + "github.com/ephpm/ephemerd/pkg/proxies" craneTarball "github.com/google/go-containerregistry/pkg/v1/tarball" ocispec "github.com/opencontainers/runtime-spec/specs-go" ) @@ -82,6 +83,12 @@ type Config struct { // config.DindConfig.AllowPrivileged for the threat model. DindAllowPrivileged bool CacheProxyEnv []string // extra env vars from cache proxies (e.g., GOPROXY=...) + // CacheProxyMounts are read-only bind mounts requested by cache proxies + // for toolchains that cannot be redirected with an env var. The Cargo + // proxy uses this to place a generated .cargo/config.toml at the + // container's filesystem root, where Cargo's ancestor-directory config + // search finds it for any workspace path. + CacheProxyMounts []proxies.Mount // Rlimits sets POSIX resource limits on each runner container's OCI // process. Zero values fall back to the containerd default (1024). // Applies on Linux only; ignored on Windows (HCS uses a different model). @@ -96,7 +103,7 @@ type Config struct { // construction site that forgets this field breaks `sudo` rather // than silently loosening the sandbox. AllowNewPrivileges bool - Network *networking.Manager + Network *networking.Manager // WindowsMemoryBytes is the memory limit for Hyper-V isolated Windows // runner containers. Zero leaves the OCI spec field unset, which gives // the HCS default (~1 GB) — too small for MSVC builds. Caller should @@ -666,6 +673,12 @@ func (r *Runtime) Create(ctx context.Context, cfg CreateConfig) (*RunnerEnv, err // Docker-in-Docker is not supported (no CAP_SYS_ADMIN/CAP_NET_ADMIN). oci.WithCapabilities(containerCapabilities), } + // Cache-proxy config mounts (e.g. the Cargo source-replacement config). + // Read-only: a job must never be able to rewrite what the next job on + // this host will read. + if len(r.cfg.CacheProxyMounts) > 0 { + opts = append(opts, withCacheProxyMounts(r.cfg.CacheProxyMounts)) + } opts = append(opts, seccompOpts()...) // AppArmor is an additional, independent layer over what the default spec // above already does (read-only /proc/sys and /sys, masked /proc paths, @@ -1276,6 +1289,44 @@ func withRunnerMount(hostDir, containerDir string) oci.SpecOpts { } } +// withCacheProxyMounts adds the bind mounts a cache proxy needs in every job +// container. The sources are host directories ephemerd generates and owns; +// runc creates the destination if the image does not have it. +func withCacheProxyMounts(mounts []proxies.Mount) oci.SpecOpts { + return func(_ context.Context, _ oci.Client, _ *containers.Container, s *oci.Spec) error { + for _, m := range mounts { + if m.Source == "" || m.Destination == "" { + continue + } + if goruntime.GOOS == "windows" { + opts := []string{"rw"} + if m.ReadOnly { + opts = []string{"ro"} + } + s.Mounts = append(s.Mounts, ocispec.Mount{ + Destination: m.Destination, + Source: m.Source, + Options: opts, + }) + continue + } + opts := []string{"rbind", "rw"} + if m.ReadOnly { + // "ro" alone is not enough on a recursive bind: without + // rprivate a later host-side mount could propagate in. + opts = []string{"rbind", "ro", "rprivate"} + } + s.Mounts = append(s.Mounts, ocispec.Mount{ + Destination: m.Destination, + Type: "bind", + Source: m.Source, + Options: opts, + }) + } + return nil + } +} + // copyDirForJob creates a writable copy of src at dst for a single job. // On Linux, uses hardlinks (cp -al) for instant, space-efficient copies. // On Windows, uses a native Go walk+copy — xcopy returned exit 4 (init