From 96cc1ee87f6a76d2420c26d80a998e0790c2369f Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 23 Oct 2026 12:07:01 +0200 Subject: [PATCH 1/2] Serve conda sharded repodata through cooldown and the cache The shard index (repodata_shards.msgpack.zst) and the shards of CEP 16 fell through to the uncached passthrough, so clients that use them, such as pixi, got no cooldown filtering and no offline copy. Without cooldown, the index and shards are now served through the metadata cache. With cooldown, shards have to be filtered, but clients cache them by the hash in the index and never fetch a hash twice. The proxy rewrites the index instead: each hash is XORed with a mask derived from channel, subdir and the current hour, and shards_base_url points at cooldown-shards/{hour}/. That route undoes the mask, fetches the upstream shard and drops records still in cooldown. The unmasked shard paths answer 404 while cooldown is on. When the index puts packages or shards somewhere the proxy doesn't route, or has an unknown version, the proxy answers 404 and clients fall back to repodata.json, which is filtered as before. Fixes #529 --- README.md | 2 + docs/configuration.md | 2 + go.mod | 4 +- go.sum | 4 + internal/handler/conda.go | 31 +- internal/handler/conda_shards.go | 469 ++++++++++++++++++++++++++ internal/handler/conda_shards_test.go | 275 +++++++++++++++ internal/handler/handler.go | 23 +- 8 files changed, 794 insertions(+), 16 deletions(-) create mode 100644 internal/handler/conda_shards.go create mode 100644 internal/handler/conda_shards_test.go diff --git a/README.md b/README.md index 0553e4a1..6e0e9d83 100644 --- a/README.md +++ b/README.md @@ -411,6 +411,8 @@ Or set via command: conda config --add channels http://localhost:8080/conda/main ``` +pixi and other rattler-based clients use sharded repodata ([CEP 16](https://github.com/conda/ceps/blob/main/cep-0016.md)) where a channel offers it, as conda-forge does. Point them at the same channel URLs, for example `pixi init -c http://localhost:8080/conda/conda-forge`. The shard index and shards are cached as metadata when `cache_metadata` is on. + ### CRAN (R) Set the repository in R: diff --git a/docs/configuration.md b/docs/configuration.md index 4da6f634..e151ea8d 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -982,6 +982,8 @@ Resolution order: exact package override, then package pattern, then ecosystem o Currently supported for npm, PyPI, pub.dev, Composer, Cargo, NuGet, Conda, RubyGems, and Hex. These ecosystems include publish timestamps in their metadata. +Conda cooldown covers `repodata.json` and sharded repodata (CEP 16). Clients cache shards by the hash in the shard index, so with cooldown on the proxy serves a rewritten index whose shard hashes change every hour and filters each shard as it is fetched. Versions leaving cooldown show up within the hour, and clients fetch the shards they need again once an hour. If a channel's shard index points packages or shards somewhere the proxy doesn't route, the proxy answers 404 for it and clients fall back to `repodata.json`. + Note: Hex cooldown requires disabling registry signature verification since the proxy re-encodes the protobuf payload without the original signature. Set `HEX_NO_VERIFY_REPO_ORIGIN=1` or configure your repo with `no_verify: true`. ## Version denylist diff --git a/go.mod b/go.mod index e2f860e0..7d17bfde 100644 --- a/go.mod +++ b/go.mod @@ -24,11 +24,13 @@ require ( github.com/git-pkgs/vulns v0.2.4 github.com/go-chi/chi/v5 v5.5.0 github.com/jmoiron/sqlx v1.4.0 + github.com/klauspost/compress v1.20.1 github.com/lib/pq v1.12.3 github.com/opencontainers/go-digest v1.0.0 github.com/prometheus/client_golang v1.25.0 github.com/prometheus/client_model v0.6.3 github.com/swaggo/swag v1.16.6 + github.com/vmihailenco/msgpack/v5 v5.4.1 gocloud.dev v0.47.0 golang.org/x/mod v0.42.0 golang.org/x/net v0.61.0 @@ -90,7 +92,6 @@ require ( github.com/googleapis/enterprise-certificate-proxy v0.3.21 // indirect github.com/googleapis/gax-go/v2 v2.24.1 // indirect github.com/josharian/intern v1.0.0 // indirect - github.com/klauspost/compress v1.20.1 // indirect github.com/kylelemons/godebug v1.1.0 // indirect github.com/mailru/easyjson v0.7.7 // indirect github.com/mattn/go-isatty v0.0.24 // indirect @@ -108,6 +109,7 @@ require ( github.com/rs/dnscache v0.0.0-20230804202142-fc85eb664529 // indirect github.com/rubyist/circuitbreaker v2.2.1+incompatible // indirect github.com/ulikunitz/xz v0.5.17 // indirect + github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel v1.46.0 // indirect go.opentelemetry.io/otel/metric v1.46.0 // indirect diff --git a/go.sum b/go.sum index f298c284..ee376526 100644 --- a/go.sum +++ b/go.sum @@ -275,6 +275,10 @@ github.com/swaggo/swag v1.16.6 h1:qBNcx53ZaX+M5dxVyTrgQ0PJ/ACK+NzhwcbieTt+9yI= github.com/swaggo/swag v1.16.6/go.mod h1:ngP2etMK5a0P3QBizic5MEwpRmluJZPHjXcMoj4Xesg= github.com/ulikunitz/xz v0.5.17 h1:flR0y/x1hgM8EGV1AW3Xll6T413G0glV8UfBwR617V4= github.com/ulikunitz/xz v0.5.17/go.mod h1:H9Rt/W6/Qj27PGauhQc6nfCDy7vHpzsOThBSaYDoEhw= +github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8= +github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok= +github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g= +github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/detectors/gcp v1.46.0 h1:PI8dGkqDaQkwJ8kOopqMhDTbrnK3UIeG/RCHH4HErbo= diff --git a/internal/handler/conda.go b/internal/handler/conda.go index 7f235e17..0f229fcd 100644 --- a/internal/handler/conda.go +++ b/internal/handler/conda.go @@ -82,6 +82,11 @@ func (h *CondaHandler) Routes() http.Handler { mux.HandleFunc("GET /{channel}/{arch}/repodata.json.bz2", h.proxyCached) mux.HandleFunc("GET /{channel}/{arch}/current_repodata.json", h.handleRepodata) + // Sharded repodata (CEP 16) + mux.HandleFunc("GET /{channel}/{arch}/"+condaShardIndexFile, h.handleShardIndex) + mux.HandleFunc("GET /{channel}/{arch}/shards/{filename}", h.handleShard) + mux.HandleFunc("GET /{channel}/{arch}/"+condaCooldownShardDir+"/{bucket}/{rest...}", h.handleCooldownShard) + // Package downloads (cache these) mux.HandleFunc("GET /{channel}/{arch}/{filename}", h.handleDownload) @@ -94,6 +99,11 @@ func (h *CondaHandler) handleDownload(w http.ResponseWriter, r *http.Request) { arch := r.PathValue("arch") filename := r.PathValue("filename") + if strings.HasSuffix(filename, condaShardSuffix) { + h.handleShard(w, r) + return + } + // Only cache actual package files if !h.isPackageFile(filename) { h.proxyUpstream(w, r) @@ -288,22 +298,31 @@ type repodataEntry struct { // Entries without a name or timestamp are kept. func (h *CondaHandler) inCooldown(filename string, raw []byte) bool { var entry repodataEntry - if err := json.Unmarshal(raw, &entry); err != nil || entry.Name == "" || entry.Timestamp == 0 { + if err := json.Unmarshal(raw, &entry); err != nil { + return false + } + return h.recordInCooldown(filename, entry.Name, entry.Version, entry.Timestamp) +} + +// recordInCooldown reports whether a package record with a millisecond +// timestamp was published too recently. Records without a name or +// timestamp are kept. +func (h *CondaHandler) recordInCooldown(filename, name, version string, timestamp float64) bool { + if name == "" || timestamp == 0 { return false } - publishedAt := time.Unix(int64(entry.Timestamp)/condaTimestampDivisor, 0) - if h.proxy.Cooldown.IsAllowed("conda", canonicalPackagePURL("conda", entry.Name), publishedAt) { + publishedAt := time.Unix(int64(timestamp)/condaTimestampDivisor, 0) + if h.proxy.Cooldown.IsAllowed("conda", canonicalPackagePURL("conda", name), publishedAt) { return false } h.proxy.Logger.Info("cooldown: filtering conda package", - "name", entry.Name, "version", entry.Version, "filename", filename) + "name", name, "version", version, "filename", filename) return true } // proxyCached forwards a metadata request with caching. func (h *CondaHandler) proxyCached(w http.ResponseWriter, r *http.Request) { - cacheKey := strings.TrimPrefix(r.URL.Path, "/") - cacheKey = strings.ReplaceAll(cacheKey, "/", "_") + cacheKey := condaCacheKey(r.URL.Path) // Large JSON indexes can exceed the metadata limit uncompressed. Conda // clients decode Content-Encoding, so cache and relay gzip verbatim. The // already-compressed repodata.json.bz2 route must stay on identity. diff --git a/internal/handler/conda_shards.go b/internal/handler/conda_shards.go new file mode 100644 index 00000000..f28f3484 --- /dev/null +++ b/internal/handler/conda_shards.go @@ -0,0 +1,469 @@ +package handler + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "net/http" + "net/url" + "path" + "strconv" + "strings" + "sync" + "time" + + "github.com/klauspost/compress/zstd" + "github.com/vmihailenco/msgpack/v5" +) + +// Sharded repodata (CEP 16) splits repodata.json into an index, +// /repodata_shards.msgpack.zst, that maps each package name to the +// sha256 of its shard, and one zstd-compressed msgpack shard per package +// stored under that hash. +// +// Without cooldown the index and the shards are passed through and cached as +// metadata. With cooldown, shards have to be filtered, and that changes their +// bytes. Clients cache shards by the hash in the index and never fetch the same +// hash again, so a filtered shard can't be served under its upstream hash. The +// proxy rewrites the index instead: every hash is XORed with a mask derived +// from the channel, the subdir and the current hour, and shards_base_url points +// at cooldown-shards/{hour}/. A shard request there undoes the mask, fetches +// the upstream shard and filters it. Each hour the keys change, so clients +// fetch shards again and see versions that have left cooldown since. +// +// When the index can't be served this way, the proxy answers 404 and clients +// fall back to repodata.json, which is filtered too. +const ( + condaShardIndexFile = "repodata_shards.msgpack.zst" + condaShardSuffix = ".msgpack.zst" + condaCooldownShardDir = "cooldown-shards" + condaShardsSubdir = "shards/" + condaShardBucket = time.Hour + condaShardIndexFormat = 1 + condaShardHashLen = sha256.Size + // maxCondaShardSize bounds a decompressed index or shard. conda-forge's + // largest index is a few MB. + maxCondaShardSize = 256 << 20 +) + +var ( + errCondaShardLayout = errors.New("unsupported sharded repodata layout") + + condaZstd = sync.OnceValues(func() (*zstd.Decoder, error) { + return zstd.NewReader(nil, zstd.WithDecoderMaxMemory(maxCondaShardSize), zstd.WithDecoderConcurrency(0)) + }) + condaZstdEncoder = sync.OnceValues(func() (*zstd.Encoder, error) { + return zstd.NewWriter(nil) + }) +) + +func (h *CondaHandler) cooldownEnabled() bool { + return h.proxy.Cooldown != nil && h.proxy.Cooldown.Enabled() +} + +// handleShardIndex serves repodata_shards.msgpack.zst. +func (h *CondaHandler) handleShardIndex(w http.ResponseWriter, r *http.Request) { + channel, arch := r.PathValue("channel"), r.PathValue("arch") + upstreamURL := h.upstreamURL + r.URL.Path + cacheKey := condaCacheKey(r.URL.Path) + + body, contentType, contentEncoding, err := h.proxy.fetchOrCacheMetadata(r.Context(), "conda", cacheKey, upstreamURL, "identity", nil, "*/*") + if err != nil { + h.proxy.writeMetadataFetchError(w, err) + return + } + + if !h.cooldownEnabled() { + index, err := decodeCondaShardIndex(body) + if err == nil { + _, err = index.shardsLayout(upstreamURL) + } + if err != nil { + h.proxy.Logger.Info("serving no sharded repodata", "url", upstreamURL, "error", err) + http.NotFound(w, r) + return + } + h.proxy.writeMetadataCachedResponseWithEncoding(w, r, "conda", cacheKey, body, contentType, contentEncoding) + return + } + + now := time.Now() + bucket := strconv.FormatInt(now.Unix()/int64(condaShardBucket/time.Second), 10) + rewritten, err := rewriteCondaShardIndex(body, upstreamURL, condaShardMask(channel, arch, bucket), bucket) + if err != nil { + h.proxy.Logger.Info("serving no sharded repodata", "url", upstreamURL, "error", err) + http.NotFound(w, r) + return + } + + next := now.Truncate(condaShardBucket).Add(condaShardBucket) + maxAge := max(int(next.Sub(now)/time.Second), 1) + w.Header().Set(headerContentType, "application/octet-stream") + w.Header().Set("Cache-Control", fmt.Sprintf("public, max-age=%d", maxAge)) + _, _ = w.Write(rewritten) +} + +// handleShard serves a shard under its upstream hash. While cooldown is on, +// the index never points here, and the unfiltered shard is refused. +func (h *CondaHandler) handleShard(w http.ResponseWriter, r *http.Request) { + if h.cooldownEnabled() || !isCondaShardFile(path.Base(r.URL.Path)) { + http.NotFound(w, r) + return + } + h.proxyCached(w, r) +} + +// handleCooldownShard serves a shard named in a rewritten index, filtered by +// cooldown. +func (h *CondaHandler) handleCooldownShard(w http.ResponseWriter, r *http.Request) { + channel, arch, bucket := r.PathValue("channel"), r.PathValue("arch"), r.PathValue("bucket") + dir, file := path.Split(r.PathValue("rest")) + if _, err := strconv.ParseUint(bucket, 10, 64); err != nil || (dir != "" && dir != condaShardsSubdir) || !isCondaShardFile(file) { + http.NotFound(w, r) + return + } + + masked, _ := hex.DecodeString(strings.TrimSuffix(file, condaShardSuffix)) + mask := condaShardMask(channel, arch, bucket) + for i := range masked { + masked[i] ^= mask[i] + } + upstreamPath := "/" + channel + "/" + arch + "/" + dir + hex.EncodeToString(masked) + condaShardSuffix + + body, _, _, err := h.proxy.fetchOrCacheMetadata(r.Context(), "conda", condaCacheKey(upstreamPath), h.upstreamURL+upstreamPath, "identity", nil, "*/*") + if err != nil { + h.proxy.writeMetadataFetchError(w, err) + return + } + + if h.cooldownEnabled() { + body, err = h.filterCondaShard(body) + if err != nil { + h.proxy.Logger.Error("failed to filter conda shard", "path", upstreamPath, "error", err) + http.Error(w, "failed to filter shard", http.StatusBadGateway) + return + } + } + w.Header().Set(headerContentType, "application/octet-stream") + _, _ = w.Write(body) +} + +func isCondaShardFile(name string) bool { + hash, ok := strings.CutSuffix(name, condaShardSuffix) + if !ok || len(hash) != 2*condaShardHashLen { + return false + } + _, err := hex.DecodeString(hash) + return err == nil && strings.ToLower(hash) == hash +} + +// condaShardMask is XORed with shard hashes in a rewritten index. It keeps +// filtered shards apart from upstream ones in client caches, which are keyed +// by hash alone. +func condaShardMask(channel, arch, bucket string) [condaShardHashLen]byte { + return sha256.Sum256([]byte("git-pkgs proxy conda cooldown shard\x00" + channel + "\x00" + arch + "\x00" + bucket)) +} + +// msgpackEntry is one member of a msgpack map, kept in its upstream order +// with its value undecoded. +type msgpackEntry struct { + key string + value msgpack.RawMessage +} + +func decodeMsgpackMap(b []byte) ([]msgpackEntry, error) { + dec := msgpack.NewDecoder(bytes.NewReader(b)) + n, err := dec.DecodeMapLen() + if err != nil { + return nil, err + } + if n < 0 { + return nil, errors.New("expected a msgpack map") + } + entries := make([]msgpackEntry, 0, n) + for range n { + key, err := dec.DecodeString() + if err != nil { + return nil, err + } + value, err := dec.DecodeRaw() + if err != nil { + return nil, err + } + entries = append(entries, msgpackEntry{key: key, value: value}) + } + return entries, nil +} + +func encodeMsgpackMap(entries []msgpackEntry) ([]byte, error) { + var buf bytes.Buffer + enc := msgpack.NewEncoder(&buf) + if err := enc.EncodeMapLen(len(entries)); err != nil { + return nil, err + } + for _, e := range entries { + if err := enc.EncodeString(e.key); err != nil { + return nil, err + } + if err := enc.Encode(e.value); err != nil { + return nil, err + } + } + return buf.Bytes(), nil +} + +func lookupMsgpackEntry(entries []msgpackEntry, key string) (int, bool) { + for i, e := range entries { + if e.key == key { + return i, true + } + } + return -1, false +} + +func decompressCondaShard(body []byte) ([]byte, error) { + dec, err := condaZstd() + if err != nil { + return nil, err + } + return dec.DecodeAll(body, nil) +} + +func compressCondaShard(body []byte) ([]byte, error) { + enc, err := condaZstdEncoder() + if err != nil { + return nil, err + } + return enc.EncodeAll(body, nil), nil +} + +// condaShardIndex is a decoded repodata_shards.msgpack.zst. +type condaShardIndex struct { + top []msgpackEntry + info []msgpackEntry +} + +func decodeCondaShardIndex(body []byte) (*condaShardIndex, error) { + raw, err := decompressCondaShard(body) + if err != nil { + return nil, err + } + top, err := decodeMsgpackMap(raw) + if err != nil { + return nil, err + } + i, ok := lookupMsgpackEntry(top, "version") + if !ok { + return nil, errors.New("sharded index has no version") + } + var version int + if err := msgpack.Unmarshal(top[i].value, &version); err != nil || version != condaShardIndexFormat { + return nil, fmt.Errorf("unsupported sharded index version %v", top[i].value) + } + i, ok = lookupMsgpackEntry(top, "info") + if !ok { + return nil, errors.New("sharded index has no info") + } + info, err := decodeMsgpackMap(top[i].value) + if err != nil { + return nil, err + } + return &condaShardIndex{top: top, info: info}, nil +} + +func (idx *condaShardIndex) infoString(key string) (string, error) { + i, ok := lookupMsgpackEntry(idx.info, key) + if !ok { + return "", nil + } + var s string + err := msgpack.Unmarshal(idx.info[i].value, &s) + return s, err +} + +// shardsLayout checks that packages and shards sit where the proxy routes +// them, below the subdir of indexURL, and returns the shards directory +// relative to the subdir: "" or "shards/". +func (idx *condaShardIndex) shardsLayout(indexURL string) (string, error) { + base, err := url.Parse(indexURL) + if err != nil { + return "", err + } + subdir := base.ResolveReference(&url.URL{Path: "./"}).String() + resolve := func(ref string) (string, error) { + if ref == "" { + return subdir, nil + } + u, err := url.Parse(ref) + if err != nil { + return "", err + } + return base.ResolveReference(u).String(), nil + } + + baseURL, err := idx.infoString("base_url") + if err != nil { + return "", err + } + if resolved, err := resolve(baseURL); err != nil || resolved != subdir { + return "", fmt.Errorf("%w: base_url %q", errCondaShardLayout, baseURL) + } + + shardsBaseURL, err := idx.infoString("shards_base_url") + if err != nil { + return "", err + } + resolved, err := resolve(shardsBaseURL) + if err != nil { + return "", err + } + rel, ok := strings.CutPrefix(resolved, subdir) + if !ok || (rel != "" && rel != condaShardsSubdir) { + return "", fmt.Errorf("%w: shards_base_url %q", errCondaShardLayout, shardsBaseURL) + } + return rel, nil +} + +// rewriteCondaShardIndex masks every shard hash and points shards_base_url +// at the cooldown shard route for bucket. +func rewriteCondaShardIndex(body []byte, indexURL string, mask [condaShardHashLen]byte, bucket string) ([]byte, error) { + idx, err := decodeCondaShardIndex(body) + if err != nil { + return nil, err + } + layout, err := idx.shardsLayout(indexURL) + if err != nil { + return nil, err + } + + shardsBase, err := msgpack.Marshal("./" + condaCooldownShardDir + "/" + bucket + "/" + layout) + if err != nil { + return nil, err + } + if i, ok := lookupMsgpackEntry(idx.info, "shards_base_url"); ok { + idx.info[i].value = shardsBase + } else { + idx.info = append(idx.info, msgpackEntry{key: "shards_base_url", value: shardsBase}) + } + info, err := encodeMsgpackMap(idx.info) + if err != nil { + return nil, err + } + i, _ := lookupMsgpackEntry(idx.top, "info") + idx.top[i].value = info + + i, ok := lookupMsgpackEntry(idx.top, "shards") + if !ok { + return nil, errors.New("sharded index has no shards") + } + shards, err := decodeMsgpackMap(idx.top[i].value) + if err != nil { + return nil, err + } + for j := range shards { + var hash []byte + if err := msgpack.Unmarshal(shards[j].value, &hash); err != nil || len(hash) != condaShardHashLen { + return nil, fmt.Errorf("shard hash for %q is not a sha256", shards[j].key) + } + for k := range hash { + hash[k] ^= mask[k] + } + if shards[j].value, err = msgpack.Marshal(hash); err != nil { + return nil, err + } + } + if idx.top[i].value, err = encodeMsgpackMap(shards); err != nil { + return nil, err + } + + raw, err := encodeMsgpackMap(idx.top) + if err != nil { + return nil, err + } + return compressCondaShard(raw) +} + +// condaShardRecord holds the fields cooldown needs from a shard record. +// Timestamps are milliseconds, encoded as an integer or a float. +type condaShardRecord struct { + Name string `msgpack:"name"` + Version string `msgpack:"version"` + Timestamp any `msgpack:"timestamp"` +} + +// filterCondaShard leaves out records in packages and packages.conda that +// are still inside their cooldown. Kept records keep their upstream bytes. +func (h *CondaHandler) filterCondaShard(body []byte) ([]byte, error) { + raw, err := decompressCondaShard(body) + if err != nil { + return nil, err + } + top, err := decodeMsgpackMap(raw) + if err != nil { + return nil, err + } + for i := range top { + if top[i].key != "packages" && top[i].key != "packages.conda" { + continue + } + records, err := decodeMsgpackMap(top[i].value) + if err != nil { + return nil, err + } + kept := records[:0] + for _, rec := range records { + var fields condaShardRecord + if err := msgpack.Unmarshal(rec.value, &fields); err != nil { + return nil, fmt.Errorf("record %q: %w", rec.key, err) + } + if h.recordInCooldown(rec.key, fields.Name, fields.Version, msgpackNumber(fields.Timestamp)) { + continue + } + kept = append(kept, rec) + } + if top[i].value, err = encodeMsgpackMap(kept); err != nil { + return nil, err + } + } + out, err := encodeMsgpackMap(top) + if err != nil { + return nil, err + } + return compressCondaShard(out) +} + +// msgpackNumber returns a decoded msgpack integer or float as float64, and 0 +// for anything else. +func msgpackNumber(v any) float64 { + switch n := v.(type) { + case int8: + return float64(n) + case int16: + return float64(n) + case int32: + return float64(n) + case int64: + return float64(n) + case uint8: + return float64(n) + case uint16: + return float64(n) + case uint32: + return float64(n) + case uint64: + return float64(n) + case float32: + return float64(n) + case float64: + return n + default: + return 0 + } +} + +// condaCacheKey is the metadata cache key for a request path. +func condaCacheKey(urlPath string) string { + return strings.ReplaceAll(strings.TrimPrefix(urlPath, "/"), "/", "_") +} diff --git a/internal/handler/conda_shards_test.go b/internal/handler/conda_shards_test.go new file mode 100644 index 00000000..43b507fb --- /dev/null +++ b/internal/handler/conda_shards_test.go @@ -0,0 +1,275 @@ +package handler + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "net/http" + "net/http/httptest" + "sort" + "strings" + "testing" + "time" + + "github.com/git-pkgs/cooldown" + "github.com/vmihailenco/msgpack/v5" +) + +func zstdMsgpack(t *testing.T, v any) []byte { + t.Helper() + raw, err := msgpack.Marshal(v) + if err != nil { + t.Fatal(err) + } + out, err := compressCondaShard(raw) + if err != nil { + t.Fatal(err) + } + return out +} + +func decodeZstdMsgpack(t *testing.T, body []byte, v any) { + t.Helper() + raw, err := decompressCondaShard(body) + if err != nil { + t.Fatal(err) + } + if err := msgpack.Unmarshal(raw, v); err != nil { + t.Fatal(err) + } +} + +type shardFixture struct { + upstream *httptest.Server + shardHash string + requests []string +} + +// newShardUpstream serves a sharded index for conda-forge/noarch with one +// shard for testpkg, holding an old and a recent build. +func newShardUpstream(t *testing.T, info map[string]any) *shardFixture { + t.Helper() + now := time.Now() + shard := zstdMsgpack(t, map[string]any{ + "packages": map[string]any{ + "testpkg-1.0.0-0.tar.bz2": map[string]any{ + "name": "testpkg", "version": "1.0.0", "timestamp": now.Add(-7 * 24 * time.Hour).UnixMilli(), + "sha256": bytes.Repeat([]byte{1}, 32), + }, + }, + "packages.conda": map[string]any{ + "testpkg-1.0.0-0.conda": map[string]any{ + "name": "testpkg", "version": "1.0.0", "timestamp": now.Add(-7 * 24 * time.Hour).UnixMilli(), + }, + "testpkg-2.0.0-0.conda": map[string]any{ + "name": "testpkg", "version": "2.0.0", "timestamp": now.Add(-time.Hour).UnixMilli(), + }, + }, + "removed": []string{"testpkg-0.1.0-0.tar.bz2"}, + }) + sum := sha256.Sum256(shard) + f := &shardFixture{shardHash: hex.EncodeToString(sum[:])} + + layout := "" + if s, _ := info["shards_base_url"].(string); strings.Contains(s, "shards/") { + layout = "shards/" + } + index := zstdMsgpack(t, map[string]any{ + "version": 1, + "info": info, + "shards": map[string]any{"testpkg": sum[:]}, + }) + + f.upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + f.requests = append(f.requests, r.URL.Path) + switch r.URL.Path { + case "/conda-forge/noarch/" + condaShardIndexFile: + _, _ = w.Write(index) + case "/conda-forge/noarch/" + layout + f.shardHash + condaShardSuffix: + _, _ = w.Write(shard) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(f.upstream.Close) + return f +} + +func shardTestHandler(f *shardFixture, cooldownDefault string) http.Handler { + proxy := testProxy() + if cooldownDefault != "" { + proxy.Cooldown = &cooldown.Config{Default: cooldownDefault} + } + return NewCondaHandlerWithUpstream(proxy, "http://proxy.local", f.upstream.URL).Routes() +} + +func serveShardRequest(h http.Handler, path string) *httptest.ResponseRecorder { + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil)) + return w +} + +type shardRecords struct { + Packages map[string]map[string]any `msgpack:"packages"` + PackagesConda map[string]map[string]any `msgpack:"packages.conda"` + Removed []string `msgpack:"removed"` +} + +func (s shardRecords) filenames() []string { + var names []string + for k := range s.Packages { + names = append(names, k) + } + for k := range s.PackagesConda { + names = append(names, k) + } + sort.Strings(names) + return names +} + +type shardIndex struct { + Version int `msgpack:"version"` + Info map[string]any `msgpack:"info"` + Shards map[string][]byte `msgpack:"shards"` +} + +func TestCondaShardsWithoutCooldown(t *testing.T) { + for _, shardsBaseURL := range []string{"", "./shards/"} { + t.Run(shardsBaseURL, func(t *testing.T) { + f := newShardUpstream(t, map[string]any{"base_url": "", "shards_base_url": shardsBaseURL, "subdir": "noarch"}) + h := shardTestHandler(f, "") + + w := serveShardRequest(h, "/conda-forge/noarch/"+condaShardIndexFile) + if w.Code != http.StatusOK { + t.Fatalf("index status = %d", w.Code) + } + var index shardIndex + decodeZstdMsgpack(t, w.Body.Bytes(), &index) + if got := hex.EncodeToString(index.Shards["testpkg"]); got != f.shardHash { + t.Fatalf("index hash = %s, want upstream %s", got, f.shardHash) + } + if index.Info["shards_base_url"] != shardsBaseURL { + t.Fatalf("shards_base_url = %v, want %q", index.Info["shards_base_url"], shardsBaseURL) + } + + dir := strings.TrimPrefix(shardsBaseURL, "./") + w = serveShardRequest(h, "/conda-forge/noarch/"+dir+f.shardHash+condaShardSuffix) + if w.Code != http.StatusOK { + t.Fatalf("shard status = %d", w.Code) + } + var shard shardRecords + decodeZstdMsgpack(t, w.Body.Bytes(), &shard) + if len(shard.filenames()) != 3 { + t.Fatalf("records = %v, want all 3", shard.filenames()) + } + }) + } +} + +func TestCondaShardsWithCooldown(t *testing.T) { + f := newShardUpstream(t, map[string]any{"base_url": "", "shards_base_url": "./shards/", "subdir": "noarch", "created_at": "2026-01-01T00:00:00Z"}) + h := shardTestHandler(f, "3d") + + w := serveShardRequest(h, "/conda-forge/noarch/"+condaShardIndexFile) + if w.Code != http.StatusOK { + t.Fatalf("index status = %d", w.Code) + } + if !strings.HasPrefix(w.Header().Get("Cache-Control"), "public, max-age=") { + t.Errorf("Cache-Control = %q", w.Header().Get("Cache-Control")) + } + var index shardIndex + decodeZstdMsgpack(t, w.Body.Bytes(), &index) + masked := hex.EncodeToString(index.Shards["testpkg"]) + if masked == f.shardHash || len(masked) != 64 { + t.Fatalf("index hash = %s, want a masked hash", masked) + } + if index.Info["created_at"] != "2026-01-01T00:00:00Z" || index.Info["subdir"] != "noarch" { + t.Errorf("info lost fields: %v", index.Info) + } + shardsBaseURL, _ := index.Info["shards_base_url"].(string) + if !strings.HasPrefix(shardsBaseURL, "./"+condaCooldownShardDir+"/") || !strings.HasSuffix(shardsBaseURL, "/shards/") { + t.Fatalf("shards_base_url = %q", shardsBaseURL) + } + + w = serveShardRequest(h, "/conda-forge/noarch/"+strings.TrimPrefix(shardsBaseURL, "./")+masked+condaShardSuffix) + if w.Code != http.StatusOK { + t.Fatalf("shard status = %d: %s", w.Code, w.Body) + } + var shard shardRecords + decodeZstdMsgpack(t, w.Body.Bytes(), &shard) + want := []string{"testpkg-1.0.0-0.conda", "testpkg-1.0.0-0.tar.bz2"} + if got := shard.filenames(); strings.Join(got, ",") != strings.Join(want, ",") { + t.Fatalf("records = %v, want %v", got, want) + } + if len(shard.Removed) != 1 { + t.Errorf("removed = %v, want it kept", shard.Removed) + } + if _, ok := shard.Packages["testpkg-1.0.0-0.tar.bz2"]["sha256"].([]byte); !ok { + t.Errorf("sha256 is no longer binary: %T", shard.Packages["testpkg-1.0.0-0.tar.bz2"]["sha256"]) + } + + // The unfiltered shard stays out of reach while cooldown is on. + w = serveShardRequest(h, "/conda-forge/noarch/shards/"+f.shardHash+condaShardSuffix) + if w.Code != http.StatusNotFound { + t.Fatalf("plain shard status = %d, want 404", w.Code) + } +} + +func TestCondaCooldownShardRejectsBadPaths(t *testing.T) { + f := newShardUpstream(t, map[string]any{"base_url": "", "shards_base_url": ""}) + h := shardTestHandler(f, "3d") + hash := strings.Repeat("a", 64) + condaShardSuffix + for _, p := range []string{ + "/conda-forge/noarch/cooldown-shards/abc/" + hash, + "/conda-forge/noarch/cooldown-shards/1/other/" + hash, + "/conda-forge/noarch/cooldown-shards/1/" + strings.Repeat("A", 64) + condaShardSuffix, + "/conda-forge/noarch/cooldown-shards/1/x.msgpack.zst", + } { + if w := serveShardRequest(h, p); w.Code != http.StatusNotFound { + t.Errorf("%s: status = %d, want 404", p, w.Code) + } + } + if len(f.requests) != 0 { + t.Errorf("upstream was asked for %v", f.requests) + } +} + +func TestCondaShardIndexUnsupportedLayout(t *testing.T) { + for name, info := range map[string]map[string]any{ + "packages elsewhere": {"base_url": "https://cdn.example.com/conda-forge/noarch/", "shards_base_url": ""}, + "shards elsewhere": {"base_url": "", "shards_base_url": "https://cdn.example.com/shards/"}, + "nested shards": {"base_url": "", "shards_base_url": "./a/b/"}, + } { + t.Run(name, func(t *testing.T) { + f := newShardUpstream(t, info) + for _, cd := range []string{"", "3d"} { + w := serveShardRequest(shardTestHandler(f, cd), "/conda-forge/noarch/"+condaShardIndexFile) + if w.Code != http.StatusNotFound { + t.Errorf("cooldown %q: status = %d, want 404 so clients use repodata.json", cd, w.Code) + } + } + }) + } +} + +func TestCondaShardIndexUnknownVersion(t *testing.T) { + index := zstdMsgpack(t, map[string]any{"version": 2, "info": map[string]any{}, "shards": map[string]any{}}) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write(index) + })) + defer upstream.Close() + + proxy := testProxy() + proxy.Cooldown = &cooldown.Config{Default: "3d"} + h := NewCondaHandlerWithUpstream(proxy, "http://proxy.local", upstream.URL).Routes() + if w := serveShardRequest(h, "/conda-forge/noarch/"+condaShardIndexFile); w.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404", w.Code) + } +} + +func TestCondaShardMaskDependsOnBucket(t *testing.T) { + a := condaShardMask("conda-forge", "noarch", "1") + if a == condaShardMask("conda-forge", "noarch", "2") || a == condaShardMask("conda-forge", "linux-64", "1") { + t.Fatal("mask must change with bucket and subdir") + } +} diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 022f3d8b..2d83d198 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -1800,21 +1800,26 @@ func (p *Proxy) proxyCachedWithEncoding(w http.ResponseWriter, r *http.Request, body, contentType, contentEncoding, err := p.fetchOrCacheMetadata(r.Context(), ecosystem, cacheKey, upstreamURL, acceptEncoding, nil, acceptHeaders...) if err != nil { - if errors.Is(err, ErrUpstreamNotFound) { - http.Error(w, "not found", http.StatusNotFound) - return - } - if writeUpstreamHeld(w, err) { - return - } - p.Logger.Error("metadata fetch failed", "error", err) - http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) + p.writeMetadataFetchError(w, err) return } p.writeMetadataCachedResponseWithEncoding(w, r, ecosystem, cacheKey, body, contentType, contentEncoding) } +// writeMetadataFetchError answers a failed metadata fetch. +func (p *Proxy) writeMetadataFetchError(w http.ResponseWriter, err error) { + if errors.Is(err, ErrUpstreamNotFound) { + http.Error(w, "not found", http.StatusNotFound) + return + } + if writeUpstreamHeld(w, err) { + return + } + p.Logger.Error("metadata fetch failed", "error", err) + http.Error(w, "failed to fetch from upstream", http.StatusBadGateway) +} + // writeMetadataCachedResponse writes a cached metadata response and handles // conditional request headers using metadata cache validators. func (p *Proxy) writeMetadataCachedResponse(w http.ResponseWriter, r *http.Request, ecosystem, cacheKey string, body []byte, contentType string) { From eb5ff1a9da7cde2ee7e2606afe94d1696c05b1c5 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Fri, 23 Oct 2026 19:06:07 +0200 Subject: [PATCH 2/2] Accept only relative paths in a conda shard index's base URLs Clients resolve base_url and shards_base_url against the proxy's URL. A scheme or host sends them around the proxy, and a leading slash leaves its mount, so neither layout is served. --- internal/handler/conda_shards.go | 36 +++++++++++---------------- internal/handler/conda_shards_test.go | 32 +++++++++++++++++------- 2 files changed, 37 insertions(+), 31 deletions(-) diff --git a/internal/handler/conda_shards.go b/internal/handler/conda_shards.go index f28f3484..f2de4758 100644 --- a/internal/handler/conda_shards.go +++ b/internal/handler/conda_shards.go @@ -18,23 +18,12 @@ import ( "github.com/vmihailenco/msgpack/v5" ) -// Sharded repodata (CEP 16) splits repodata.json into an index, -// /repodata_shards.msgpack.zst, that maps each package name to the -// sha256 of its shard, and one zstd-compressed msgpack shard per package -// stored under that hash. -// -// Without cooldown the index and the shards are passed through and cached as -// metadata. With cooldown, shards have to be filtered, and that changes their -// bytes. Clients cache shards by the hash in the index and never fetch the same -// hash again, so a filtered shard can't be served under its upstream hash. The -// proxy rewrites the index instead: every hash is XORed with a mask derived -// from the channel, the subdir and the current hour, and shards_base_url points -// at cooldown-shards/{hour}/. A shard request there undoes the mask, fetches -// the upstream shard and filters it. Each hour the keys change, so clients -// fetch shards again and see versions that have left cooldown since. -// -// When the index can't be served this way, the proxy answers 404 and clients -// fall back to repodata.json, which is filtered too. +// Sharded repodata (CEP 16) maps each package name to the sha256 of its +// shard. With cooldown the proxy filters shards, which changes their bytes, +// but clients cache shards by hash and never fetch the same hash again. So +// the rewritten index XORs every hash with a mask that changes each hour: +// filtered shards get hashes of their own, and clients fetch them again once +// versions have left cooldown. const ( condaShardIndexFile = "repodata_shards.msgpack.zst" condaShardSuffix = ".msgpack.zst" @@ -285,7 +274,10 @@ func (idx *condaShardIndex) infoString(key string) (string, error) { // shardsLayout checks that packages and shards sit where the proxy routes // them, below the subdir of indexURL, and returns the shards directory -// relative to the subdir: "" or "shards/". +// relative to the subdir: "" or "shards/". Clients resolve base_url and +// shards_base_url against the proxy's URL, so only relative paths stay on +// the proxy: a scheme or host leads clients around it, a leading / outside +// its mount. func (idx *condaShardIndex) shardsLayout(indexURL string) (string, error) { base, err := url.Parse(indexURL) if err != nil { @@ -300,6 +292,9 @@ func (idx *condaShardIndex) shardsLayout(indexURL string) (string, error) { if err != nil { return "", err } + if u.Scheme != "" || u.Host != "" || strings.HasPrefix(u.Path, "/") { + return "", errCondaShardLayout + } return base.ResolveReference(u).String(), nil } @@ -316,11 +311,8 @@ func (idx *condaShardIndex) shardsLayout(indexURL string) (string, error) { return "", err } resolved, err := resolve(shardsBaseURL) - if err != nil { - return "", err - } rel, ok := strings.CutPrefix(resolved, subdir) - if !ok || (rel != "" && rel != condaShardsSubdir) { + if err != nil || !ok || (rel != "" && rel != condaShardsSubdir) { return "", fmt.Errorf("%w: shards_base_url %q", errCondaShardLayout, shardsBaseURL) } return rel, nil diff --git a/internal/handler/conda_shards_test.go b/internal/handler/conda_shards_test.go index 43b507fb..27b0ad7f 100644 --- a/internal/handler/conda_shards_test.go +++ b/internal/handler/conda_shards_test.go @@ -74,12 +74,7 @@ func newShardUpstream(t *testing.T, info map[string]any) *shardFixture { if s, _ := info["shards_base_url"].(string); strings.Contains(s, "shards/") { layout = "shards/" } - index := zstdMsgpack(t, map[string]any{ - "version": 1, - "info": info, - "shards": map[string]any{"testpkg": sum[:]}, - }) - + var index []byte f.upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { f.requests = append(f.requests, r.URL.Path) switch r.URL.Path { @@ -92,6 +87,20 @@ func newShardUpstream(t *testing.T, info map[string]any) *shardFixture { } })) t.Cleanup(f.upstream.Close) + + // {upstream} in an info value stands for the upstream's URL, {host} for + // its host. + placeholders := strings.NewReplacer("{upstream}", f.upstream.URL, "{host}", strings.TrimPrefix(f.upstream.URL, "http://")) + for k, v := range info { + if s, ok := v.(string); ok { + info[k] = placeholders.Replace(s) + } + } + index = zstdMsgpack(t, map[string]any{ + "version": 1, + "info": info, + "shards": map[string]any{"testpkg": sum[:]}, + }) return f } @@ -236,9 +245,14 @@ func TestCondaCooldownShardRejectsBadPaths(t *testing.T) { func TestCondaShardIndexUnsupportedLayout(t *testing.T) { for name, info := range map[string]map[string]any{ - "packages elsewhere": {"base_url": "https://cdn.example.com/conda-forge/noarch/", "shards_base_url": ""}, - "shards elsewhere": {"base_url": "", "shards_base_url": "https://cdn.example.com/shards/"}, - "nested shards": {"base_url": "", "shards_base_url": "./a/b/"}, + "packages elsewhere": {"base_url": "https://cdn.example.com/conda-forge/noarch/", "shards_base_url": ""}, + "shards elsewhere": {"base_url": "", "shards_base_url": "https://cdn.example.com/shards/"}, + "nested shards": {"base_url": "", "shards_base_url": "./a/b/"}, + "packages upstream": {"base_url": "{upstream}/conda-forge/noarch/", "shards_base_url": ""}, + "shards upstream": {"base_url": "", "shards_base_url": "{upstream}/conda-forge/noarch/"}, + "no scheme upstream": {"base_url": "//{host}/conda-forge/noarch/", "shards_base_url": ""}, + "root-relative": {"base_url": "/conda-forge/noarch/", "shards_base_url": ""}, + "root-relative shards": {"base_url": "", "shards_base_url": "/conda-forge/noarch/shards/"}, } { t.Run(name, func(t *testing.T) { f := newShardUpstream(t, info)