From 321665db5e88d54d94a173c49f18ac7319fdf0d3 Mon Sep 17 00:00:00 2001 From: fuziontech Date: Thu, 3 Sep 2026 23:33:17 +0000 Subject: [PATCH 1/2] fix(cache-proxy): scope cache keys by SigV4 credential and lock down relay paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Security fix for two findings in the unauthenticated forward proxy: (A) Cache keys mixed no credential material, so a warm entry (local or from a peer) was served with zero authorization. In the managed-warehouse topology tenants share a bucket with per-org path prefixes, so one tenant could read another tenant's objects from a warm cache. The tenant scope — the SigV4 access key ID from the Authorization header, unique per issued STS credential set — is now part of every CacheKey and BlockKey hash input (scope + "\x00" + url + ...). A cross-tenant request misses and goes to the origin, where S3 authorization applies. Unsigned requests share the empty-scope namespace, which is correct for public objects. Peer traffic carries only opaque keys, so the peer protocol is unchanged. (B) handleConnect dialed any host:port and the plain-HTTP forward path forwarded to any absolute URL — open relay and SSRF primitives reachable cluster-wide via the hostPort. CONNECT now allows only port 443, refuses loopback/link-local/unspecified IP literals, and honors the new CONNECT_ALLOWED_SUFFIXES hostname list when configured. The plain-HTTP forward path forwards only to hosts matching CACHE_HOST_SUFFIXES when that list is configured (signed S3 traffic matches by definition); legacy no-suffix mode stays unrestricted for backward compatibility. Co-authored-by: Shelley --- cmd/cache-proxy/README.md | 39 ++++- cmd/cache-proxy/block_serve.go | 29 ++-- cmd/cache-proxy/block_serve_test.go | 58 +++---- cmd/cache-proxy/blocks.go | 9 +- cmd/cache-proxy/blocks_test.go | 10 +- cmd/cache-proxy/cache.go | 16 +- cmd/cache-proxy/cache_test.go | 8 +- cmd/cache-proxy/main.go | 14 ++ cmd/cache-proxy/peers_test.go | 2 +- cmd/cache-proxy/proxy.go | 110 +++++++++++-- cmd/cache-proxy/proxy_test.go | 186 ++++++++++++++++++++- cmd/cache-proxy/scope.go | 35 ++++ cmd/cache-proxy/scope_test.go | 231 +++++++++++++++++++++++++++ cmd/cache-proxy/summary_pull_test.go | 6 +- 14 files changed, 671 insertions(+), 82 deletions(-) create mode 100644 cmd/cache-proxy/scope.go create mode 100644 cmd/cache-proxy/scope_test.go diff --git a/cmd/cache-proxy/README.md b/cmd/cache-proxy/README.md index 1b844128..c78d7b71 100644 --- a/cmd/cache-proxy/README.md +++ b/cmd/cache-proxy/README.md @@ -19,7 +19,8 @@ available, and forwards cache misses to origin object storage. | `CACHE_MAX_CONCURRENT_PEER_PROBES` | `64` | Per-pod non-blocking cap on active summary-mode `/cache/has` HTTP requests and sockets. It is not a process goroutine limit. When exhausted, confirmations are skipped and the request fetches origin. `CACHE_MAX_PEER_PROBES_IN_FLIGHT` is a deprecated alias. | | `CACHE_PROXY_ID` | pod name, node name, then hostname | Stable opaque receiver identity used for deterministic peer-summary selection; it must not be a customer or object identifier. | | `HEALTH_ADDR` | `:8082` | Health and Prometheus metrics listener. | -| `CACHE_HOST_SUFFIXES` | empty | Empty means all `GET` hosts are cacheable. Otherwise, cache only hosts containing one of the comma-separated suffixes. | +| `CACHE_HOST_SUFFIXES` | empty | Empty means all `GET` hosts are cacheable. Otherwise, cache only hosts containing one of the comma-separated suffixes. When non-empty, the plain-HTTP forward path also refuses targets outside the list with `403`; when empty (legacy mode), forwarding stays unrestricted. | +| `CONNECT_ALLOWED_SUFFIXES` | empty | Optional hostname substrings a `CONNECT` target must contain. Empty means any hostname on port 443 is allowed. The port-443 and local-IP-refusal rules always apply. | | `CACHE_BLOCK_MODE` | `off` | `on` enables block-aligned caching; any other value (including unset) keeps the legacy exact-range path. See [Block-aligned mode](#block-aligned-mode). | | `CACHE_BLOCK_SIZE_BYTES` | `8388608` (8 MiB) | Fixed block size for block-aligned mode. Ignored when block mode is off. | | `CACHE_BLOCK_MAX_SPAN_BLOCKS` | `8` | Max blocks coalesced into one origin range fetch. Ignored when block mode is off. | @@ -91,9 +92,40 @@ All lookup state comes from the in-memory index under the cache mutex, not from filesystem stats, so `/cache/has` and eviction/size accounting can never disagree about whether an entry exists. +## Security boundaries + +The proxy binds a hostPort and serves unauthenticated requests from any pod in +the cluster. Two boundaries keep that safe in the managed-warehouse topology, +where tenants share a bucket with per-org path prefixes: + +- **Cache keys are tenant-scoped.** The scope is the SigV4 access key ID from + the request's `Authorization` header + (`AWS4-HMAC-SHA256 Credential=/...`). STS access key IDs are + unique per issued credential set, so the scope separates tenants. The legacy + key is `sha256(scope + "\x00" + url + "|" + range)` and the block key is + `sha256(scope + "\x00" + url + "|blk|" + idx + "|" + blockSize)`. A warm + entry for one tenant — local or from a peer — is never served to a request + signed by another tenant; it simply misses and goes to the origin, where + S3 authorization applies. Requests without a parseable SigV4 header share + the empty-scope namespace, which is correct for public objects. The access + key ID is an identifier, not a secret; the secret key and the signature + never enter the key. Peer traffic carries only these opaque keys, so the + peer protocol is unchanged. Changing the key format invalidates all + existing on-disk entries: they become unreachable and age out by LRU. +- **The relay surface is locked down.** `CONNECT` only dials port 443, refuses + loopback, link-local, and unspecified IP literals, and optionally requires + the hostname to contain a `CONNECT_ALLOWED_SUFFIXES` entry. Port 443 for + arbitrary hostnames stays allowed because DuckDB reads external HTTPS + sources through the proxy while `http_proxy` is set globally. The + plain-HTTP forward path (non-`GET`, passthrough, non-cache hosts) forwards + only to hosts matching `CACHE_HOST_SUFFIXES` when that list is configured; + signed S3 traffic matches by definition. With no suffixes configured the + forward path stays unrestricted for backward compatibility. + ## Block-aligned mode -The legacy cache key is `sha256(url|range)` — an exact match on the client's +The legacy cache key is `sha256(scope + "\x00" + url + "|" + range)` — an +exact match on the client's `Range` header. DuckDB's Parquet reader rarely issues the same byte range twice, even across repeat runs of the same query: footer probes, row-group reads, and column-chunk reads all drift by a few bytes depending on prior @@ -104,7 +136,8 @@ miss rate on a workload that should have been fully warm. Block-aligned mode fixes this by keying the cache on fixed-size blocks of the underlying object instead of the client's exact range. The key is -`sha256(url|blk|idx|blockSize)`, where `idx` is the block index (`start / +`sha256(scope + "\x00" + url + "|blk|" + idx + "|" + blockSize)`, where `idx` +is the block index (`start / blockSize`) and `blockSize` is part of the key so a config change can't serve a wrong-sized entry — old-size entries just become unreachable and age out normally. A request is served by locating the blocks its range overlaps diff --git a/cmd/cache-proxy/block_serve.go b/cmd/cache-proxy/block_serve.go index 834e03d2..fa8b2407 100644 --- a/cmd/cache-proxy/block_serve.go +++ b/cmd/cache-proxy/block_serve.go @@ -85,13 +85,14 @@ func writeRangeNotSatisfiable(w http.ResponseWriter, objectSize int64) { } // fetchOriginSpan fetches blocks [firstIdx, lastIdx] of r.URL in ONE origin -// range GET and commits each block to the store under its BlockKey. Rewriting +// range GET and commits each block to the store under its BlockKey. scope is +// the tenant scope from TenantScope and is part of every block key. Rewriting // the Range header is legal: DuckDB httpfs signs only // host;x-amz-content-sha256;x-amz-date (see forwardUncached), so Range is not // covered by the SigV4 signature. Content-Range is validated before any block // is committed, and each selected block must contain exactly the advertised // number of bytes. -func (p *CacheProxy) fetchOriginSpan(r *http.Request, blockSize, firstIdx, lastIdx int64) error { +func (p *CacheProxy) fetchOriginSpan(r *http.Request, blockSize, firstIdx, lastIdx int64, scope string) error { timeout := p.originTimeout if timeout <= 0 { timeout = defaultOriginTimeout @@ -160,7 +161,7 @@ func (p *CacheProxy) fetchOriginSpan(r *http.Request, blockSize, firstIdx, lastI remaining := expectedBodySize for idx := firstIdx; idx <= lastIdx && remaining > 0; idx++ { blockBytes := min(blockSize, remaining) - size, err := p.store.PutStream(BlockKey(r.URL.String(), idx, blockSize), &exactLengthReader{ + size, err := p.store.PutStream(BlockKey(scope, r.URL.String(), idx, blockSize), &exactLengthReader{ r: resp.Body, remaining: blockBytes, }) @@ -190,9 +191,11 @@ func (p *CacheProxy) blockPresent(key string) bool { // serveBlockAligned serves a cacheable GET whose Range is an absolute // bytes=start-end pair from block-aligned cache entries: local disk, then // peers, then coalesced origin fetches for contiguous missing runs (chunked -// at maxSpanBlocks per origin request). Returns false when the request shape -// is not block-servable; the caller then runs the legacy exact-range path. -func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, rangeHeader string) bool { +// at maxSpanBlocks per origin request). scope is the tenant scope from +// TenantScope and is part of every block key. Returns false when the request +// shape is not block-servable; the caller then runs the legacy exact-range +// path. +func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, rangeHeader, scope string) bool { requestStart := time.Now() var peerDur, s3Dur, writeDur time.Duration @@ -251,7 +254,7 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r // if only lo keyed the call, the loser would adopt the winner's // (shorter) fetch result while believing its own longer span was // covered — silently leaving trailing blocks unfetched. - flightKey := fmt.Sprintf("%s|%d", BlockKey(urlStr, lo, p.blockSize), hi) + flightKey := fmt.Sprintf("%s|%d", BlockKey(scope, urlStr, lo, p.blockSize), hi) _, err := p.flights.Do(flightKey, func() (fetchResult, error) { fetchStart := time.Now() // Retry transient origin failures inside the flight so every @@ -259,7 +262,7 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r // absorbed here instead of reaching DuckDB as a 502. _, fetchSpan := proxyTracer.Start(r.Context(), "cache.origin_span_fetch") fetchErr := p.retryOriginFetch(r, fetchSpan, func() error { - return p.fetchOriginSpan(r, p.blockSize, lo, hi) + return p.fetchOriginSpan(r, p.blockSize, lo, hi, scope) }) fetchSpan.End() s3Dur += time.Since(fetchStart) @@ -300,7 +303,7 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r return true } for idx := firstIdx; idx <= lastIdx; idx++ { - key := BlockKey(urlStr, idx, p.blockSize) + key := BlockKey(scope, urlStr, idx, p.blockSize) if p.blockPresent(key) { if !flushRun(idx - 1) { return true // error already written @@ -376,7 +379,7 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r fetchStart := time.Now() _, fetchSpan := proxyTracer.Start(r.Context(), "cache.origin_span_refetch") err := p.retryOriginFetch(r, fetchSpan, func() error { - return p.fetchOriginSpan(r, p.blockSize, lo, runEnd) + return p.fetchOriginSpan(r, p.blockSize, lo, runEnd, scope) }) fetchSpan.End() s3Dur += time.Since(fetchStart) @@ -388,7 +391,7 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r nOrigin += runEnd - lo + 1 } for idx := firstIdx; idx <= lastIdx; idx++ { - if p.blockPresent(BlockKey(urlStr, idx, p.blockSize)) { + if p.blockPresent(BlockKey(scope, urlStr, idx, p.blockSize)) { reverify(idx - 1) continue } @@ -398,7 +401,7 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r } reverify(lastIdx) for idx := firstIdx; idx <= lastIdx; idx++ { - if !p.blockPresent(BlockKey(urlStr, idx, p.blockSize)) { + if !p.blockPresent(BlockKey(scope, urlStr, idx, p.blockSize)) { slog.Error("Block still missing after presence re-fetch; failing closed.", "url", urlStr, "block", idx) http.Error(w, "block cache entry missing after re-fetch", http.StatusBadGateway) @@ -425,7 +428,7 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r } } for idx := firstIdx; idx <= lastIdx; idx++ { - reader, size, ok := p.store.openFile(BlockKey(urlStr, idx, p.blockSize)) + reader, size, ok := p.store.openFile(BlockKey(scope, urlStr, idx, p.blockSize)) if !ok { closeOpened() blockFallbackTotal.WithLabelValues("entry_vanished").Inc() diff --git a/cmd/cache-proxy/block_serve_test.go b/cmd/cache-proxy/block_serve_test.go index 39aebad5..cd9f475d 100644 --- a/cmd/cache-proxy/block_serve_test.go +++ b/cmd/cache-proxy/block_serve_test.go @@ -71,13 +71,13 @@ func TestFetchOriginSpan(t *testing.T) { req := &http.Request{Method: http.MethodGet, URL: u, Host: u.Host, Header: http.Header{}} // Fetch blocks 1..3 in one span (block 3 is the short tail). - if err := p.fetchOriginSpan(req, blockSize, 1, 3); err != nil { + if err := p.fetchOriginSpan(req, blockSize, 1, 3, ""); err != nil { t.Fatalf("fetchOriginSpan: %v", err) } // Every block in the span must now be a complete, correct cache entry. for idx := int64(1); idx <= 3; idx++ { - key := BlockKey(u.String(), idx, blockSize) + key := BlockKey("", u.String(), idx, blockSize) reader, size, ok := store.Open(key) if !ok { t.Fatalf("block %d not committed to store", idx) @@ -99,7 +99,7 @@ func TestFetchOriginSpan(t *testing.T) { } // Block 0 was outside the span and must not exist. - if store.Has(BlockKey(u.String(), 0, blockSize)) { + if store.Has(BlockKey("", u.String(), 0, blockSize)) { t.Fatal("block 0 should not have been fetched") } } @@ -135,7 +135,7 @@ func TestFetchOriginSpanRejects200(t *testing.T) { u, _ := url.Parse(origin.URL + "/bucket/f.parquet") req := &http.Request{Method: http.MethodGet, URL: u, Host: u.Host, Header: http.Header{}} - err = p.fetchOriginSpan(req, blockSize, 1, 2) + err = p.fetchOriginSpan(req, blockSize, 1, 2, "") if err == nil { t.Fatal("expected fetchOriginSpan to fail closed on a 200 response to a ranged request") } @@ -143,7 +143,7 @@ func TestFetchOriginSpanRejects200(t *testing.T) { t.Fatalf("error %q should mention the unexpected status code", err.Error()) } for idx := int64(0); idx <= 3; idx++ { - if store.Has(BlockKey(u.String(), idx, blockSize)) { + if store.Has(BlockKey("", u.String(), idx, blockSize)) { t.Fatalf("block %d must not be committed when the origin ignored Range", idx) } } @@ -170,11 +170,11 @@ func TestFetchOriginSpanRejectsMismatchedContentRange(t *testing.T) { u, _ := url.Parse(origin.URL + "/bucket/f.parquet") req := &http.Request{Method: http.MethodGet, URL: u, Host: u.Host, Header: http.Header{}} - if err := p.fetchOriginSpan(req, blockSize, 1, 2); err == nil { + if err := p.fetchOriginSpan(req, blockSize, 1, 2, ""); err == nil { t.Fatal("expected mismatched Content-Range to be rejected") } for idx := int64(1); idx <= 2; idx++ { - if store.Has(BlockKey(u.String(), idx, blockSize)) { + if store.Has(BlockKey("", u.String(), idx, blockSize)) { t.Fatalf("block %d committed from a mismatched Content-Range", idx) } } @@ -199,10 +199,10 @@ func TestFetchOriginSpanRejectsCleanShortBody(t *testing.T) { u, _ := url.Parse(origin.URL + "/bucket/f.parquet") req := &http.Request{Method: http.MethodGet, URL: u, Host: u.Host, Header: http.Header{}} - if err := p.fetchOriginSpan(req, blockSize, 1, 2); err == nil { + if err := p.fetchOriginSpan(req, blockSize, 1, 2, ""); err == nil { t.Fatal("expected clean short 206 body to be rejected") } - if store.Has(BlockKey(u.String(), 2, blockSize)) { + if store.Has(BlockKey("", u.String(), 2, blockSize)) { t.Fatal("truncated block committed from a short 206 body") } } @@ -230,7 +230,7 @@ func TestServeBlockAlignedFailsClosedOn200Origin(t *testing.T) { req = req.WithContext(context.Background()) w := httptest.NewRecorder() - if !p.serveBlockAligned(w, req, "bytes=1500-2500") { + if !p.serveBlockAligned(w, req, "bytes=1500-2500", "") { t.Fatal("expected serveBlockAligned to handle the request (not fall back)") } if w.Code != http.StatusBadGateway { @@ -240,7 +240,7 @@ func TestServeBlockAlignedFailsClosedOn200Origin(t *testing.T) { t.Fatalf("Content-Range = %q, want unset: headers must not be committed before the 502", got) } for idx := int64(0); idx <= 3; idx++ { - if store.Has(BlockKey(u.String(), idx, blockSize)) { + if store.Has(BlockKey("", u.String(), idx, blockSize)) { t.Fatalf("block %d must not be committed when the origin ignored Range", idx) } } @@ -332,7 +332,7 @@ func TestServeBlockAlignedRetriesTruncatedOriginBody(t *testing.T) { // The truncated first attempt must not have left partial blocks behind: // blocks 1 and 2 must be complete. for idx := int64(1); idx <= 2; idx++ { - _, size, ok := store.Open(BlockKey(u.String(), idx, blockSize)) + _, size, ok := store.Open(BlockKey("", u.String(), idx, blockSize)) if !ok || size != blockSize { t.Fatalf("block %d: present=%v size=%d, want present size %d", idx, ok, size, blockSize) } @@ -405,7 +405,7 @@ func TestFetchOriginSpanSendsBlockAlignedRange(t *testing.T) { req := &http.Request{Method: http.MethodGet, URL: u, Host: u.Host, Header: http.Header{ "Range": []string{"bytes=1500-2500"}, // client's original, must be ignored }} - if err := p.fetchOriginSpan(req, blockSize, 1, 2); err != nil { + if err := p.fetchOriginSpan(req, blockSize, 1, 2, ""); err != nil { t.Fatal(err) } want := "bytes=" + strconv.Itoa(1*blockSize) + "-" + strconv.Itoa(3*blockSize-1) @@ -437,7 +437,7 @@ func doBlockRequest(t *testing.T, p *CacheProxy, rawURL, rangeHeader string) *ht Header: http.Header{"Range": []string{rangeHeader}}} req = req.WithContext(context.Background()) w := httptest.NewRecorder() - if !p.serveBlockAligned(w, req, rangeHeader) { + if !p.serveBlockAligned(w, req, rangeHeader, "") { t.Fatalf("serveBlockAligned returned false for %q", rangeHeader) } return w @@ -553,7 +553,7 @@ func TestServeBlockAlignedFallsBackOnRangeShape(t *testing.T) { req = req.WithContext(context.Background()) before := counterValue(t, blockFallbackTotal.WithLabelValues(tt.reason)) - if p.serveBlockAligned(httptest.NewRecorder(), req, tt.rangeHeader) { + if p.serveBlockAligned(httptest.NewRecorder(), req, tt.rangeHeader, "") { t.Fatalf("range %q must return false (legacy fallback)", tt.rangeHeader) } if got := counterValue(t, blockFallbackTotal.WithLabelValues(tt.reason)); got != before+1 { @@ -605,7 +605,7 @@ func TestServeBlockAlignedPeerFillCountsAsLocalMiss(t *testing.T) { for i := range blockData { blockData[i] = byte(i % 251) } - key := BlockKey(target, 0, blockSize) + key := BlockKey("", target, 0, blockSize) var hasCalls, getCalls int32 peerAddr := newPeerServer(t, key, blockData, http.StatusOK, &hasCalls, &getCalls) @@ -691,7 +691,7 @@ func TestServeBlockAlignedDoesNotReverifyPastObjectEOF(t *testing.T) { req = req.WithContext(context.Background()) w := httptest.NewRecorder() - if !p.serveBlockAligned(w, req, "bytes=0-4095") { + if !p.serveBlockAligned(w, req, "bytes=0-4095", "") { t.Fatal("expected serveBlockAligned to handle the request (not fall back)") } if w.Code != http.StatusPartialContent { @@ -730,7 +730,7 @@ func TestServeBlockAlignedReturns416ForColdPastEOFStart(t *testing.T) { req = req.WithContext(context.Background()) w := httptest.NewRecorder() - if !p.serveBlockAligned(w, req, "bytes=2200-3000") { + if !p.serveBlockAligned(w, req, "bytes=2200-3000", "") { t.Fatal("expected serveBlockAligned to handle the request (not fall back)") } if w.Code != http.StatusRequestedRangeNotSatisfiable { @@ -742,7 +742,7 @@ func TestServeBlockAlignedReturns416ForColdPastEOFStart(t *testing.T) { if got := w.Body.Len(); got != 0 { t.Fatalf("body length = %d, want 0", got) } - if !store.Has(BlockKey(target, 2, blockSize)) { + if !store.Has(BlockKey("", target, 2, blockSize)) { t.Fatal("validated tail block was not cached while learning object size") } } @@ -767,10 +767,10 @@ func TestHandleProxyRoutesToBlockMode(t *testing.T) { t.Fatalf("block-mode HandleProxy: status %d len %d", w.Code, w.Body.Len()) } // Blocks 0-2 stored under block keys; the legacy exact-range key must NOT exist. - if store.Has(CacheKey(u.String(), "bytes=100-2100")) { + if store.Has(CacheKey("", u.String(), "bytes=100-2100")) { t.Fatal("legacy key written in block mode") } - if !store.Has(BlockKey(u.String(), 0, blockSize)) { + if !store.Has(BlockKey("", u.String(), 0, blockSize)) { t.Fatal("block 0 missing after block-mode request") } } @@ -790,10 +790,10 @@ func TestHandleProxyBlockModeOffUsesLegacyPath(t *testing.T) { req.Header.Set("Range", "bytes=100-2100") p.HandleProxy(httptest.NewRecorder(), req) - if !store.Has(CacheKey(u.String(), "bytes=100-2100")) { + if !store.Has(CacheKey("", u.String(), "bytes=100-2100")) { t.Fatal("legacy key missing with block mode off") } - if store.Has(BlockKey(u.String(), 0, blockSize)) { + if store.Has(BlockKey("", u.String(), 0, blockSize)) { t.Fatal("block key written with block mode off") } } @@ -838,7 +838,7 @@ func TestServeBlockAlignedConcurrentDriftedRanges(t *testing.T) { w := httptest.NewRecorder() // t.Errorf (not Fatalf) below: FailNow must only be called from // the goroutine running the test function, not spawned ones. - if !p.serveBlockAligned(w, req, rangeHeader) { + if !p.serveBlockAligned(w, req, rangeHeader, "") { t.Errorf("goroutine %d: serveBlockAligned returned false (legacy fallback) for %q", i, rangeHeader) return } @@ -896,7 +896,7 @@ func TestServeBlockAlignedRejectsDegenerateConfig(t *testing.T) { req := &http.Request{Method: http.MethodGet, URL: u, Host: u.Host, Header: http.Header{"Range": []string{"bytes=0-100"}}} req = req.WithContext(context.Background()) - if p.serveBlockAligned(httptest.NewRecorder(), req, "bytes=0-100") { + if p.serveBlockAligned(httptest.NewRecorder(), req, "bytes=0-100", "") { t.Fatal("expected false (legacy fallback) for degenerate config") } }) @@ -931,7 +931,7 @@ func TestServeBlockAlignedPinsBlocksBeforeHeaders(t *testing.T) { for i := range data { data[i] = byte((idx*blockSize + int64(i)) % 251) } - if _, err := store.PutStream(BlockKey(target, idx, blockSize), strings.NewReader(string(data))); err != nil { + if _, err := store.PutStream(BlockKey("", target, idx, blockSize), strings.NewReader(string(data))); err != nil { t.Fatalf("seed block %d: %v", idx, err) } } @@ -951,7 +951,7 @@ func TestServeBlockAlignedPinsBlocksBeforeHeaders(t *testing.T) { } } - if !p.serveBlockAligned(w, req, "bytes=0-2047") { + if !p.serveBlockAligned(w, req, "bytes=0-2047", "") { t.Fatal("serveBlockAligned unexpectedly fell back") } if w.Code != http.StatusPartialContent || w.Body.Len() != 2*int(blockSize) { @@ -1021,11 +1021,11 @@ func TestHandleProxyOversizedBlockSpanFallsBackToLegacy(t *testing.T) { if w.Code != http.StatusPartialContent || w.Body.Len() != 3*int(blockSize) { t.Fatalf("response status=%d len=%d, want legacy 206 and %d bytes", w.Code, w.Body.Len(), 3*blockSize) } - if !store.Has(CacheKey(target, "bytes=0-3071")) { + if !store.Has(CacheKey("", target, "bytes=0-3071")) { t.Fatal("legacy exact-range entry was not written") } for idx := int64(0); idx < 3; idx++ { - if store.Has(BlockKey(target, idx, blockSize)) { + if store.Has(BlockKey("", target, idx, blockSize)) { t.Fatalf("block %d was written before oversized-span fallback", idx) } } diff --git a/cmd/cache-proxy/blocks.go b/cmd/cache-proxy/blocks.go index 0f894f3b..055ca439 100644 --- a/cmd/cache-proxy/blocks.go +++ b/cmd/cache-proxy/blocks.go @@ -85,8 +85,13 @@ func blockSpan(start, end, blockSize int64) (firstIdx, lastIdx int64) { // BlockKey computes the cache key for one block of an object. blockSize is // part of the key so a block-size config change can never serve a // wrong-sized entry — old-size entries simply become unreachable and age out. -func BlockKey(url string, blockIdx, blockSize int64) string { +// scope is the SigV4 access key ID (see TenantScope), so two tenants reading +// the same object URL never share a block. +// +// Hash input format: scope + "\x00" + url + "|blk|" + idx + "|" + blockSize. +// The NUL separator makes the scope boundary unambiguous, same as CacheKey. +func BlockKey(scope, url string, blockIdx, blockSize int64) string { h := sha256.New() - _, _ = fmt.Fprintf(h, "%s|blk|%d|%d", url, blockIdx, blockSize) + _, _ = fmt.Fprintf(h, "%s\x00%s|blk|%d|%d", scope, url, blockIdx, blockSize) return fmt.Sprintf("%x", h.Sum(nil)) } diff --git a/cmd/cache-proxy/blocks_test.go b/cmd/cache-proxy/blocks_test.go index ce5a85ec..93a9f18e 100644 --- a/cmd/cache-proxy/blocks_test.go +++ b/cmd/cache-proxy/blocks_test.go @@ -59,9 +59,9 @@ func TestBlockSpan(t *testing.T) { } func TestBlockKey(t *testing.T) { - k1 := BlockKey("http://s3/bucket/f.parquet", 0, 8<<20) - k2 := BlockKey("http://s3/bucket/f.parquet", 1, 8<<20) - k3 := BlockKey("http://s3/bucket/f.parquet", 0, 16<<20) + k1 := BlockKey("", "http://s3/bucket/f.parquet", 0, 8<<20) + k2 := BlockKey("", "http://s3/bucket/f.parquet", 1, 8<<20) + k3 := BlockKey("", "http://s3/bucket/f.parquet", 0, 16<<20) if !IsValidCacheKey(k1) { t.Fatalf("BlockKey output %q is not a valid cache key", k1) } @@ -71,10 +71,10 @@ func TestBlockKey(t *testing.T) { if k1 == k3 { t.Fatal("different block sizes must produce different keys") } - if k1 != BlockKey("http://s3/bucket/f.parquet", 0, 8<<20) { + if k1 != BlockKey("", "http://s3/bucket/f.parquet", 0, 8<<20) { t.Fatal("BlockKey must be deterministic") } - if k1 == CacheKey("http://s3/bucket/f.parquet", "bytes=0-100") { + if k1 == CacheKey("", "http://s3/bucket/f.parquet", "bytes=0-100") { t.Fatal("block keys must not collide with legacy keys") } } diff --git a/cmd/cache-proxy/cache.go b/cmd/cache-proxy/cache.go index 60066acc..997e786d 100644 --- a/cmd/cache-proxy/cache.go +++ b/cmd/cache-proxy/cache.go @@ -62,12 +62,18 @@ var ( }) ) -// CacheKey computes a deterministic cache key from a full URL and byte range. -// The URL includes scheme, host, path, and query — so different buckets, regions, -// or query-signed URLs naturally produce different keys. -func CacheKey(url, rangeHeader string) string { +// CacheKey computes a deterministic cache key from a tenant scope, a full +// URL, and a byte range. The URL includes scheme, host, path, and query — so +// different buckets, regions, or query-signed URLs naturally produce +// different keys. The scope is the SigV4 access key ID (see TenantScope), so +// two tenants reading the same object URL never share an entry. +// +// Hash input format: scope + "\x00" + url + "|" + range. The NUL separator +// makes the scope boundary unambiguous: no URL or range byte sequence can +// shift bytes into or out of the scope field. +func CacheKey(scope, url, rangeHeader string) string { h := sha256.New() - _, _ = fmt.Fprintf(h, "%s|%s", url, rangeHeader) + _, _ = fmt.Fprintf(h, "%s\x00%s|%s", scope, url, rangeHeader) return fmt.Sprintf("%x", h.Sum(nil)) } diff --git a/cmd/cache-proxy/cache_test.go b/cmd/cache-proxy/cache_test.go index a6971280..3862e7c8 100644 --- a/cmd/cache-proxy/cache_test.go +++ b/cmd/cache-proxy/cache_test.go @@ -92,15 +92,15 @@ func TestIsValidCacheKey(t *testing.T) { } func TestCacheKeyDeterministic(t *testing.T) { - a := CacheKey("http://s3/bucket/file.parquet", "bytes=0-1023") - b := CacheKey("http://s3/bucket/file.parquet", "bytes=0-1023") + a := CacheKey("", "http://s3/bucket/file.parquet", "bytes=0-1023") + b := CacheKey("", "http://s3/bucket/file.parquet", "bytes=0-1023") if a != b { t.Fatalf("CacheKey not deterministic: %s != %s", a, b) } if !IsValidCacheKey(a) { t.Errorf("CacheKey output %q is not a valid key", a) } - c := CacheKey("http://s3/bucket/file.parquet", "bytes=0-2047") + c := CacheKey("", "http://s3/bucket/file.parquet", "bytes=0-2047") if a == c { t.Fatal("different ranges produced identical keys") } @@ -670,7 +670,7 @@ func BenchmarkTouchLargeCache(b *testing.B) { c := &DiskCache{order: list.New(), index: make(map[string]*list.Element)} keys := make([]string, 285000) for i := range keys { - keys[i] = CacheKey(fmt.Sprintf("http://bucket/f%d.parquet", i), "bytes=0-1") + keys[i] = CacheKey("", fmt.Sprintf("http://bucket/f%d.parquet", i), "bytes=0-1") c.addLocked(keys[i], 1) } b.ResetTimer() diff --git a/cmd/cache-proxy/main.go b/cmd/cache-proxy/main.go index f6a85125..6d1bb049 100644 --- a/cmd/cache-proxy/main.go +++ b/cmd/cache-proxy/main.go @@ -151,6 +151,18 @@ func main() { } } + // Comma-separated hostname substrings a CONNECT target must contain. + // Empty means any hostname on port 443 is allowed; the port and local-IP + // rules apply either way. See connectRefusalReason. + var connectAllowedSuffixes []string + if raw := os.Getenv("CONNECT_ALLOWED_SUFFIXES"); raw != "" { + for _, s := range strings.Split(raw, ",") { + if s = strings.TrimSpace(s); s != "" { + connectAllowedSuffixes = append(connectAllowedSuffixes, s) + } + } + } + slog.Info("Starting cache-proxy.", "cache_dir", cacheDir, "max_percent", maxPercent, @@ -164,6 +176,7 @@ func main() { "peer_service", peerService, "peer_lookup_mode", lookupMode, "cache_host_suffixes", cacheHostSuffixes, + "connect_allowed_suffixes", connectAllowedSuffixes, ) // Block-aligned cache mode: fixed-size, content-addressed blocks instead of @@ -213,6 +226,7 @@ func main() { proxy.blockMode = blockMode proxy.blockSize = blockSize proxy.maxSpanBlocks = maxSpanBlocks + proxy.connectAllowedSuffixes = connectAllowedSuffixes // Forward HTTP proxy (DuckDB httpfs traffic). ServeMux can't match absolute // URLs in forward-proxy requests, so use the handler directly. diff --git a/cmd/cache-proxy/peers_test.go b/cmd/cache-proxy/peers_test.go index 2a22234d..848385bb 100644 --- a/cmd/cache-proxy/peers_test.go +++ b/cmd/cache-proxy/peers_test.go @@ -372,7 +372,7 @@ func TestPeerServesBlockKeys(t *testing.T) { } // Create a block key for a parquet block. - key := BlockKey("http://s3/bucket/f.parquet", 3, 8<<20) + key := BlockKey("", "http://s3/bucket/f.parquet", 3, 8<<20) blockContent := "block-content" // Store the block. diff --git a/cmd/cache-proxy/proxy.go b/cmd/cache-proxy/proxy.go index 9ae63aca..5e6204ad 100644 --- a/cmd/cache-proxy/proxy.go +++ b/cmd/cache-proxy/proxy.go @@ -39,15 +39,20 @@ func requestSpanAttrs(r *http.Request) []attribute.KeyValue { // CacheProxy is a forward HTTP proxy that caches responses on local NVMe. // DuckDB httpfs sends each S3 request as a signed plain-HTTP request to the -// proxy; the proxy caches by URL+Range and forwards misses verbatim. The -// SigV4 signature stays valid because Host/URL are unchanged, so the proxy -// needs no AWS credentials of its own. +// proxy; the proxy caches by tenant scope (SigV4 access key ID) + URL + Range +// and forwards misses verbatim. The SigV4 signature stays valid because +// Host/URL are unchanged, so the proxy needs no AWS credentials of its own. type CacheProxy struct { store *DiskCache peers *PeerManager client *http.Client flights singleFlight + // connectDial dials CONNECT tunnel targets. It is net.DialTimeout in + // production and a seam in tests, where an allowed target (port 443, + // non-local hostname) cannot resolve to a real listener. + connectDial func(network, addr string, timeout time.Duration) (net.Conn, error) + originTimeout time.Duration originRetryMaxAttempts int originRetryInitialBackoff time.Duration @@ -55,9 +60,18 @@ type CacheProxy struct { // cacheHostSuffixes are the Host substrings that identify DuckLake bucket // traffic worth caching. Requests whose Host doesn't contain any of these - // are passed through without caching. + // are passed through without caching. When the list is non-empty it also + // bounds the plain-HTTP forward path: a host matching no suffix is + // refused with 403 (see forward). cacheHostSuffixes []string + // connectAllowedSuffixes optionally bounds CONNECT tunnel targets. When + // non-empty, the target hostname must contain one of these substrings + // (same strings.Contains semantics as cacheHostSuffixes). Empty means any + // hostname on port 443 is allowed; the port and local-IP rules always + // apply. See connectRefusalReason. + connectAllowedSuffixes []string + // blockMode, blockSize, and maxSpanBlocks configure the block-aligned // serve path (serveBlockAligned): whether it's active, the fixed block // size in bytes, and the max number of blocks coalesced into one origin @@ -147,6 +161,7 @@ func NewCacheProxy(store *DiskCache, peers *PeerManager, cacheHostSuffixes []str store: store, peers: peers, client: &http.Client{Timeout: defaultOriginTimeout}, + connectDial: net.DialTimeout, originTimeout: defaultOriginTimeout, originRetryMaxAttempts: defaultOriginRetryMaxAttempts, originRetryInitialBackoff: defaultOriginRetryInitialBackoff, @@ -196,6 +211,12 @@ func (p *CacheProxy) shouldCache(r *http.Request) bool { // and spot dial failures. For full request/response visibility on writes, // DuckDB has to actually use plain HTTP via forwardUncached (s3_use_ssl = // false), which httpfs has been observed ignoring for some PUT paths. +// +// The proxy binds a hostPort and is reachable cluster-wide without +// authentication, so the tunnel target is restricted (connectRefusalReason): +// port 443 only, no local IP literals, optionally a hostname suffix list. +// Without this the CONNECT path is an open relay into every TCP service +// reachable from the node's network context. func (p *CacheProxy) handleConnect(w http.ResponseWriter, r *http.Request) { connectStart := time.Now() target := r.Host @@ -203,7 +224,14 @@ func (p *CacheProxy) handleConnect(w http.ResponseWriter, r *http.Request) { attribute.String("server.address", target), attribute.String("client.address", r.RemoteAddr), )) - upstream, err := net.DialTimeout("tcp", target, 10*time.Second) + if reason := p.connectRefusalReason(target); reason != "" { + span.SetStatus(codes.Error, reason) + span.End() + slog.Warn("Forward-proxy CONNECT target refused.", "target", target, "reason", reason, "client", r.RemoteAddr) + http.Error(w, "CONNECT target not allowed: "+reason, http.StatusForbidden) + return + } + upstream, err := p.connectDial("tcp", target, 10*time.Second) if err != nil { span.SetStatus(codes.Error, err.Error()) span.End() @@ -278,6 +306,44 @@ func (p *CacheProxy) handleConnect(w http.ResponseWriter, r *http.Request) { }() } +// connectRefusalReason returns "" when a CONNECT target may be dialed, else +// a short reason for the Warn log. Only port 443 is allowed: DuckDB tunnels +// external HTTPS reads through the proxy while http_proxy is set globally, +// and 443 is the only port those reads need. Refusing every other port keeps +// the tunnel from relaying into node-internal services (peer APIs, the +// control plane, Kubelet). IP literals that name loopback, link-local, or +// unspecified addresses are refused outright, so the tunnel cannot reach the +// node itself or the cloud metadata endpoint. When connectAllowedSuffixes is +// configured the hostname must also contain one of its suffixes. +// +// Known limit: a hostname is checked as a string, not resolved. A public +// hostname whose DNS record points at a local address still passes. Closing +// that gap requires a resolve-and-pin dialer; the port-443 restriction keeps +// the residual exposure small. +func (p *CacheProxy) connectRefusalReason(target string) string { + host, port, err := net.SplitHostPort(target) + if err != nil { + return "target is not host:port" + } + if port != "443" { + return "port is not 443" + } + if ip := net.ParseIP(host); ip != nil { + if ip.IsLoopback() || ip.IsUnspecified() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() { + return "local or non-routable IP literal" + } + } + if len(p.connectAllowedSuffixes) > 0 { + for _, s := range p.connectAllowedSuffixes { + if strings.Contains(host, s) { + return "" + } + } + return "hostname not in CONNECT_ALLOWED_SUFFIXES" + } + return "" +} + // Hop-by-hop headers per RFC 7230 §6.1 — must not be forwarded. var hopByHop = map[string]bool{ "connection": true, @@ -322,28 +388,32 @@ func (p *CacheProxy) HandleProxy(w http.ResponseWriter, r *http.Request) { // Non-GET (HEAD, etc.) is never cached — forward and return. if r.Method != http.MethodGet { - p.forwardUncached(w, r) + p.forward(w, r) return } if r.Header.Get(cachePassthroughHeader) == "true" { - p.forwardUncached(w, r) + p.forward(w, r) return } // Only cache URLs that look like DuckLake bucket traffic. Anything else // (non-bucket HTTP) is a passthrough. if !p.shouldCache(r) { - p.forwardUncached(w, r) + p.forward(w, r) return } rangeHeader := r.Header.Get("Range") + // The tenant scope (SigV4 access key ID) is part of every cache key, so + // an entry written for one tenant's credentials is never served to a + // request signed by another tenant. + scope := TenantScope(r) - if p.blockMode && p.serveBlockAligned(w, r, rangeHeader) { + if p.blockMode && p.serveBlockAligned(w, r, rangeHeader, scope) { return } // Legacy exact-range path (also the fallback for non-absolute ranges). - cacheKey := CacheKey(r.URL.String(), rangeHeader) + cacheKey := CacheKey(scope, r.URL.String(), rangeHeader) // Requests without a propagated parent still start a standalone trace. // Thread the cache request span context into origin/peer work below. @@ -783,6 +853,26 @@ func (p *CacheProxy) serveStream(w http.ResponseWriter, r io.Reader, size int64, _, _ = io.Copy(w, r) } +// forward gates the uncached plain-HTTP forward path on the configured host +// list, then forwards. The proxy binds a hostPort and is reachable +// cluster-wide without authentication, so an unbounded forward path is an +// open relay to any URL reachable from the node. When cacheHostSuffixes is +// configured, a target host matching no suffix is refused with 403. Signed +// S3 traffic is unaffected: S3 endpoints match the suffixes by definition. +// When no suffixes are configured (legacy mode) every host is allowed — +// deployments predating CACHE_HOST_SUFFIXES rely on unrestricted +// passthrough, and with no configured host list there is no boundary to +// enforce. +func (p *CacheProxy) forward(w http.ResponseWriter, r *http.Request) { + if len(p.cacheHostSuffixes) > 0 && !p.shouldCache(r) { + slog.Warn("Forward-proxy target host not in CACHE_HOST_SUFFIXES; refusing.", + "method", r.Method, "url", r.URL.String(), "client", r.RemoteAddr) + http.Error(w, "forward target host not allowed", http.StatusForbidden) + return + } + p.forwardUncached(w, r) +} + // forwardUncached forwards a request to the origin without caching. Used for // HEAD and other non-GET methods that shouldn't consume cache space. // diff --git a/cmd/cache-proxy/proxy_test.go b/cmd/cache-proxy/proxy_test.go index a03c7c94..b8072b21 100644 --- a/cmd/cache-proxy/proxy_test.go +++ b/cmd/cache-proxy/proxy_test.go @@ -221,7 +221,7 @@ func TestHandleProxyPassthroughGETSkipsCacheAndStripsMarker(t *testing.T) { if traceHeadersReachedOrigin.Load() { t.Fatal("trace propagation headers reached origin") } - if _, _, ok := proxy.store.Open(CacheKey(originURL+"/bucket/file.parquet", "")); ok { + if _, _, ok := proxy.store.Open(CacheKey("", originURL+"/bucket/file.parquet", "")); ok { t.Fatal("passthrough request populated the cache") } } @@ -1138,8 +1138,14 @@ func TestHandleConnectLogsOpenAndClose(t *testing.T) { // Proxy: wrap HandleProxy in an httptest server. CONNECT requests go // through Go's standard server hijack path, which is what the real - // cache-proxy does in production. + // cache-proxy does in production. The CONNECT lockdown only allows port + // 443 and refuses local IP literals, so the test target is + // "example.com:443" and the dial seam redirects it to the local echo + // origin. proxy := newTestProxy(t) + proxy.connectDial = func(_, _ string, _ time.Duration) (net.Conn, error) { + return net.Dial("tcp", origin.Addr().String()) + } proxySrv := httptest.NewServer(http.HandlerFunc(proxy.HandleProxy)) defer proxySrv.Close() proxyURL, _ := url.Parse(proxySrv.URL) @@ -1153,7 +1159,7 @@ func TestHandleConnectLogsOpenAndClose(t *testing.T) { } defer func() { _ = pconn.Close() }() - connectReq := fmt.Sprintf("CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\n", origin.Addr().String(), origin.Addr().String()) + connectReq := "CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n" if _, err := pconn.Write([]byte(connectReq)); err != nil { t.Fatalf("write CONNECT: %v", err) } @@ -1202,8 +1208,8 @@ func TestHandleConnectLogsOpenAndClose(t *testing.T) { if !strings.Contains(out, `msg="Forward-proxy CONNECT closed."`) { t.Errorf("expected close log, got:\n%s", out) } - if !strings.Contains(out, fmt.Sprintf(`target=%s`, origin.Addr().String())) { - t.Errorf("expected target= attr matching origin addr, got:\n%s", out) + if !strings.Contains(out, `target=example.com:443`) { + t.Errorf("expected target= attr matching the CONNECT target, got:\n%s", out) } } @@ -1215,6 +1221,11 @@ func TestHandleConnectLogsDialFailure(t *testing.T) { defer restore() proxy := newTestProxy(t) + // The CONNECT lockdown only allows port 443 targets, so the dial seam + // stands in for an unreachable upstream. + proxy.connectDial = func(_, _ string, _ time.Duration) (net.Conn, error) { + return nil, fmt.Errorf("connection refused") + } proxySrv := httptest.NewServer(http.HandlerFunc(proxy.HandleProxy)) defer proxySrv.Close() proxyURL, _ := url.Parse(proxySrv.URL) @@ -1225,8 +1236,7 @@ func TestHandleConnectLogsDialFailure(t *testing.T) { } defer func() { _ = pconn.Close() }() - // 127.0.0.1:1 is reserved/unbound — kernel rejects the dial fast. - connectReq := "CONNECT 127.0.0.1:1 HTTP/1.1\r\nHost: 127.0.0.1:1\r\n\r\n" + connectReq := "CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n" if _, err := pconn.Write([]byte(connectReq)); err != nil { t.Fatalf("write CONNECT: %v", err) } @@ -1507,3 +1517,165 @@ func TestForwardUncachedPreservesMethod(t *testing.T) { }) } } + +// TestConnectRefusalReason covers the CONNECT relay lockdown: only port 443, +// no local or non-routable IP literals, and the optional hostname suffix list. +func TestConnectRefusalReason(t *testing.T) { + cases := []struct { + target string + suffixes []string + wantAllow bool + }{ + // Internal TCP services must not be reachable through the tunnel. + {target: "worker:8816", wantAllow: false}, + {target: "peer.cache-proxy:8081", wantAllow: false}, + {target: "example.com:80", wantAllow: false}, + {target: "example.com:4430", wantAllow: false}, + {target: "example.com", wantAllow: false}, + // Local and non-routable IP literals are refused even on 443. + {target: "127.0.0.1:443", wantAllow: false}, + {target: "127.1.2.3:443", wantAllow: false}, + {target: "[::1]:443", wantAllow: false}, + {target: "169.254.169.254:443", wantAllow: false}, + {target: "169.254.169.254:80", wantAllow: false}, + {target: "[fe80::1]:443", wantAllow: false}, + {target: "0.0.0.0:443", wantAllow: false}, + // Public HTTPS targets are the documented product need. + {target: "example.com:443", wantAllow: true}, + {target: "datasets.clickhouse.com:443", wantAllow: true}, + {target: "8.8.8.8:443", wantAllow: true}, + // The suffix list narrows allowed hostnames when configured. + {target: "example.com:443", suffixes: []string{"s3.amazonaws.com"}, wantAllow: false}, + {target: "bucket.s3.amazonaws.com:443", suffixes: []string{"s3.amazonaws.com"}, wantAllow: true}, + } + for _, c := range cases { + p := &CacheProxy{connectAllowedSuffixes: c.suffixes} + reason := p.connectRefusalReason(c.target) + if gotAllow := reason == ""; gotAllow != c.wantAllow { + t.Errorf("connectRefusalReason(%q, suffixes=%v) allow = %v (reason %q), want allow = %v", + c.target, c.suffixes, gotAllow, reason, c.wantAllow) + } + } +} + +// TestHandleConnectRefused: a CONNECT to an internal service port is refused +// with 403 and a Warn log before any dial is attempted. +func TestHandleConnectRefused(t *testing.T) { + buf, restore := captureSlog(t) + defer restore() + + proxy := newTestProxy(t) + dialed := false + proxy.connectDial = func(_, _ string, _ time.Duration) (net.Conn, error) { + dialed = true + return nil, fmt.Errorf("must not be called") + } + + for _, target := range []string{"worker:8816", "169.254.169.254:80", "127.0.0.1:443"} { + req := httptest.NewRequest(http.MethodConnect, "//"+target, nil) + req.Host = target + rec := httptest.NewRecorder() + proxy.HandleProxy(rec, req) + if rec.Code != http.StatusForbidden { + t.Errorf("CONNECT %s: status = %d, want 403", target, rec.Code) + } + } + if dialed { + t.Fatal("refused CONNECT reached the dialer") + } + if out := buf.String(); !strings.Contains(out, `Forward-proxy CONNECT target refused.`) { + t.Errorf("expected refusal Warn log, got:\n%s", out) + } +} + +// TestHandleConnectAllowedTargetIsNotForbidden: an allowed target +// (port 443, public hostname) passes the gate; the dial then fails in the +// test environment and must surface as 502, not 403. +func TestHandleConnectAllowedTargetIsNotForbidden(t *testing.T) { + proxy := newTestProxy(t) + dialed := false + proxy.connectDial = func(_, _ string, _ time.Duration) (net.Conn, error) { + dialed = true + return nil, fmt.Errorf("connection refused") + } + + req := httptest.NewRequest(http.MethodConnect, "//example.com:443", nil) + req.Host = "example.com:443" + rec := httptest.NewRecorder() + proxy.HandleProxy(rec, req) + if rec.Code == http.StatusForbidden { + t.Fatalf("allowed CONNECT target got 403: %s", rec.Body.String()) + } + if rec.Code != http.StatusBadGateway { + t.Fatalf("allowed CONNECT target with failed dial: status = %d, want 502", rec.Code) + } + if !dialed { + t.Fatal("allowed CONNECT never reached the dialer") + } +} + +// TestForwardRefusesNonSuffixHost: with CACHE_HOST_SUFFIXES configured, the +// plain-HTTP forward path must refuse targets outside the suffix list — +// otherwise it is an open relay to any URL reachable from the node. Requests +// to matching hosts (signed S3 traffic by definition) keep working. +func TestForwardRefusesNonSuffixHost(t *testing.T) { + var originCalls atomic.Int32 + _, originURL := newTestServer(t, func(w http.ResponseWriter, r *http.Request) { + originCalls.Add(1) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }) + // The httptest origin is a 127.0.0.1 URL; use it as the allowed suffix so + // matching-host traffic reaches it, and refuse everything else. + proxy := NewCacheProxy(newTestCache(t), nil, []string{"127.0.0.1"}) + + // GET, PUT, and passthrough GET to a non-suffix host: all refused. + for _, c := range []struct { + method string + headers http.Header + }{ + {method: http.MethodGet}, + {method: http.MethodPut}, + {method: http.MethodGet, headers: http.Header{cachePassthroughHeader: []string{"true"}}}, + } { + rec := doForwardProxyRequest(proxy, c.method, "http://evil.internal:8816/x", c.headers) + if rec.Code != http.StatusForbidden { + t.Errorf("%s to non-suffix host: status = %d, want 403", c.method, rec.Code) + } + } + if got := originCalls.Load(); got != 0 { + t.Fatalf("origin calls = %d, want 0 for refused targets", got) + } + + // A matching host still forwards (non-GET) and caches (GET). + if rec := doForwardProxyRequest(proxy, http.MethodPut, originURL+"/bucket/k", nil); rec.Code != http.StatusOK { + t.Errorf("PUT to suffix host: status = %d, want 200", rec.Code) + } + if rec := doForwardProxyRequest(proxy, http.MethodGet, originURL+"/bucket/k", http.Header{ + cachePassthroughHeader: []string{"true"}, + }); rec.Code != http.StatusOK { + t.Errorf("passthrough GET to suffix host: status = %d, want 200", rec.Code) + } + if got := originCalls.Load(); got != 2 { + t.Fatalf("origin calls = %d, want 2 for allowed suffix-host traffic", got) + } +} + +// TestForwardLegacyModeAllowsAllHosts: with no CACHE_HOST_SUFFIXES +// configured the forward path stays unrestricted (legacy mode) — deployments +// predating the setting rely on unrestricted passthrough. +func TestForwardLegacyModeAllowsAllHosts(t *testing.T) { + proxy := newTestProxy(t) // no suffixes + var originCalls atomic.Int32 + _, originURL := newTestServer(t, func(w http.ResponseWriter, r *http.Request) { + originCalls.Add(1) + w.WriteHeader(http.StatusOK) + }) + + if rec := doForwardProxyRequest(proxy, http.MethodHead, originURL+"/bucket/k", nil); rec.Code != http.StatusOK { + t.Errorf("legacy-mode HEAD: status = %d, want 200", rec.Code) + } + if got := originCalls.Load(); got != 1 { + t.Fatalf("origin calls = %d, want 1", got) + } +} diff --git a/cmd/cache-proxy/scope.go b/cmd/cache-proxy/scope.go new file mode 100644 index 00000000..0a6234d8 --- /dev/null +++ b/cmd/cache-proxy/scope.go @@ -0,0 +1,35 @@ +package main + +import ( + "net/http" + "strings" +) + +// TenantScope extracts the stable, non-secret tenant dimension of a request: +// the access key ID from the SigV4 Authorization header +// ("AWS4-HMAC-SHA256 Credential=///s3/aws4_request, ..."). +// STS access key IDs are unique per issued credential set, so the ID +// separates tenants that share a bucket with per-org path prefixes. Cache +// keys mix in this scope, so a warm entry for one tenant is never served to +// a request signed by another tenant's credentials. +// +// A request without a parseable SigV4 header gets the empty scope. Unsigned +// requests therefore share one cache namespace, which is correct for public +// objects. The scope is an identifier, not a secret: the secret access key +// and the signature never enter the cache key. +func TenantScope(r *http.Request) string { + auth := r.Header.Get("Authorization") + if !strings.HasPrefix(auth, "AWS4-HMAC-SHA256 ") { + return "" + } + const marker = "Credential=" + i := strings.Index(auth, marker) + if i < 0 { + return "" + } + id, _, found := strings.Cut(auth[i+len(marker):], "/") + if !found || id == "" { + return "" + } + return id +} diff --git a/cmd/cache-proxy/scope_test.go b/cmd/cache-proxy/scope_test.go new file mode 100644 index 00000000..a7561342 --- /dev/null +++ b/cmd/cache-proxy/scope_test.go @@ -0,0 +1,231 @@ +package main + +import ( + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" +) + +func TestTenantScope(t *testing.T) { + cases := []struct { + name string + auth string + want string + }{ + { + name: "sigv4 header", + auth: "AWS4-HMAC-SHA256 Credential=ASIAEXAMPLE123/20260101/us-east-1/s3/aws4_request, SignedHeaders=host;x-amz-date, Signature=abc", + want: "ASIAEXAMPLE123", + }, + { + name: "static credentials", + auth: "AWS4-HMAC-SHA256 Credential=AKIAEXAMPLE/20260101/eu-west-1/s3/aws4_request, SignedHeaders=host, Signature=def", + want: "AKIAEXAMPLE", + }, + {name: "no header", auth: "", want: ""}, + {name: "not sigv4", auth: "Bearer token", want: ""}, + {name: "algorithm without credential", auth: "AWS4-HMAC-SHA256 SignedHeaders=host, Signature=abc", want: ""}, + {name: "credential without scope path", auth: "AWS4-HMAC-SHA256 Credential=ASIAEXAMPLE123", want: ""}, + {name: "empty access key id", auth: "AWS4-HMAC-SHA256 Credential=/20260101/us-east-1/s3/aws4_request", want: ""}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + r := httptest.NewRequest(http.MethodGet, "http://s3/bucket/f", nil) + if c.auth != "" { + r.Header.Set("Authorization", c.auth) + } + if got := TenantScope(r); got != c.want { + t.Errorf("TenantScope(%q) = %q, want %q", c.auth, got, c.want) + } + }) + } +} + +// TestCacheKeyTenantScope: the tenant scope is part of the cache key. Two +// access key IDs reading the same URL+range must produce different keys, and +// unsigned requests (empty scope) must share one namespace. +func TestCacheKeyTenantScope(t *testing.T) { + const url = "http://s3/bucket/org-a/data.parquet" + const rng = "bytes=0-1023" + a := CacheKey("ASIAAAAA", url, rng) + b := CacheKey("ASIABBBB", url, rng) + if a == b { + t.Fatal("different access key IDs must produce different cache keys") + } + if a != CacheKey("ASIAAAAA", url, rng) { + t.Fatal("CacheKey must be deterministic for a fixed scope") + } + if CacheKey("", url, rng) == a || CacheKey("", url, rng) == b { + t.Fatal("unsigned requests must not collide with a signed tenant namespace") + } + if CacheKey("", url, rng) != CacheKey("", url, rng) { + t.Fatal("unsigned requests must share one cache namespace") + } + + if BlockKey("ASIAAAAA", url, 0, 8<<20) == BlockKey("ASIABBBB", url, 0, 8<<20) { + t.Fatal("different access key IDs must produce different block keys") + } + if BlockKey("", url, 0, 8<<20) != BlockKey("", url, 0, 8<<20) { + t.Fatal("unsigned requests must share one block namespace") + } +} + +func sigv4Header(accessKeyID string) string { + return "AWS4-HMAC-SHA256 Credential=" + accessKeyID + "/20260101/us-east-1/s3/aws4_request, SignedHeaders=host;x-amz-date, Signature=abc" +} + +// TestHandleProxyTenantIsolation: a warm cache entry for tenant A must never +// be served to tenant B. Same URL, same Range, different SigV4 access key +// IDs: B's request must MISS and hit the origin, not receive A's bytes from +// cache. +func TestHandleProxyTenantIsolation(t *testing.T) { + proxy := newTestProxy(t) + + var originCalls atomic.Int32 + _, originURL := newTestServer(t, func(w http.ResponseWriter, r *http.Request) { + originCalls.Add(1) + w.Header().Set("Content-Type", "application/octet-stream") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("tenant-shared-object-bytes")) + }) + + target := originURL + "/bucket/org-a/data.parquet" + rangeHeader := "bytes=0-24" + + // Tenant A warms the cache. + rec := doForwardProxyRequest(proxy, http.MethodGet, target, http.Header{ + "Range": []string{rangeHeader}, + "Authorization": []string{sigv4Header("ASIATENANTAAAAA")}, + }) + if rec.Code != http.StatusPartialContent { + t.Fatalf("tenant A: status = %d, want 206", rec.Code) + } + if got := originCalls.Load(); got != 1 { + t.Fatalf("origin calls after tenant A = %d, want 1", got) + } + + // Tenant A again: cache hit, no origin call. + rec = doForwardProxyRequest(proxy, http.MethodGet, target, http.Header{ + "Range": []string{rangeHeader}, + "Authorization": []string{sigv4Header("ASIATENANTAAAAA")}, + }) + if rec.Code != http.StatusPartialContent { + t.Fatalf("tenant A repeat: status = %d, want 206", rec.Code) + } + if got := originCalls.Load(); got != 1 { + t.Fatalf("origin calls after tenant A repeat = %d, want 1 (cache hit)", got) + } + + // Tenant B requests the same URL+range. It must MISS and go to origin — + // under no circumstances may it be served tenant A's cached entry. + rec = doForwardProxyRequest(proxy, http.MethodGet, target, http.Header{ + "Range": []string{rangeHeader}, + "Authorization": []string{sigv4Header("ASIATENANTBBBBB")}, + }) + if rec.Code != http.StatusPartialContent { + t.Fatalf("tenant B: status = %d, want 206", rec.Code) + } + if got := originCalls.Load(); got != 2 { + t.Fatalf("origin calls after tenant B = %d, want 2 (B must miss A's entry)", got) + } + + // Both tenants now hold independent entries: a repeat from each hits its + // own namespace and the origin sees no further calls. + for _, akid := range []string{"ASIATENANTAAAAA", "ASIATENANTBBBBB"} { + rec = doForwardProxyRequest(proxy, http.MethodGet, target, http.Header{ + "Range": []string{rangeHeader}, + "Authorization": []string{sigv4Header(akid)}, + }) + if rec.Code != http.StatusPartialContent { + t.Fatalf("%s repeat: status = %d, want 206", akid, rec.Code) + } + } + if got := originCalls.Load(); got != 2 { + t.Fatalf("origin calls after warm repeats = %d, want 2", got) + } +} + +// TestHandleProxyUnsignedRequestsShareCache: requests without a SigV4 +// Authorization header share one cache namespace (correct for public +// objects). +func TestHandleProxyUnsignedRequestsShareCache(t *testing.T) { + proxy := newTestProxy(t) + + var originCalls atomic.Int32 + _, originURL := newTestServer(t, func(w http.ResponseWriter, r *http.Request) { + originCalls.Add(1) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("public-object-bytes")) + }) + + target := originURL + "/bucket/public/data.parquet" + headers := http.Header{"Range": []string{"bytes=0-18"}} + + if rec := doForwardProxyRequest(proxy, http.MethodGet, target, headers); rec.Code != http.StatusPartialContent { + t.Fatalf("first unsigned: status = %d, want 206", rec.Code) + } + if rec := doForwardProxyRequest(proxy, http.MethodGet, target, headers); rec.Code != http.StatusPartialContent { + t.Fatalf("second unsigned: status = %d, want 206", rec.Code) + } + if got := originCalls.Load(); got != 1 { + t.Fatalf("origin calls = %d, want 1 (unsigned requests share one entry)", got) + } +} + +// TestBlockModeTenantIsolation: the block-aligned path keys blocks by tenant +// scope too. Tenant B reading the same object blocks as a warm tenant A must +// miss and fetch its own span from the origin. +func TestBlockModeTenantIsolation(t *testing.T) { + const blockSize = 1024 + body := make([]byte, 4*blockSize) + for i := range body { + body[i] = byte(i % 251) + } + var originCalls atomic.Int32 + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + originCalls.Add(1) + serveSyntheticRanged(w, r, body) + })) + defer origin.Close() + + p, _ := newBlockProxy(t, origin, blockSize) + p.blockMode = true + target := origin.URL + "/bucket/org-a/f.parquet" + + get := func(akid string) *httptest.ResponseRecorder { + headers := http.Header{"Range": []string{"bytes=0-2047"}} + if akid != "" { + headers.Set("Authorization", sigv4Header(akid)) + } + return doForwardProxyRequest(p, http.MethodGet, target, headers) + } + + if rec := get("ASIATENANTAAAAA"); rec.Code != http.StatusPartialContent { + t.Fatalf("tenant A: status = %d, want 206 (body: %s)", rec.Code, rec.Body.String()) + } + if got := originCalls.Load(); got != 1 { + t.Fatalf("origin calls after tenant A = %d, want 1", got) + } + // Tenant A repeat: warm hit, no origin call. + if rec := get("ASIATENANTAAAAA"); rec.Code != http.StatusPartialContent { + t.Fatalf("tenant A repeat: status = %d, want 206", rec.Code) + } + if got := originCalls.Load(); got != 1 { + t.Fatalf("origin calls after tenant A repeat = %d, want 1 (cache hit)", got) + } + // Tenant B: same URL, same range, same blocks — but a different scope, so + // it must fetch from the origin rather than adopt A's cached blocks. + if rec := get("ASIATENANTBBBBB"); rec.Code != http.StatusPartialContent { + t.Fatalf("tenant B: status = %d, want 206", rec.Code) + } + if got := originCalls.Load(); got != 2 { + t.Fatalf("origin calls after tenant B = %d, want 2 (B must miss A's blocks)", got) + } + if rec := get("ASIATENANTBBBBB"); rec.Code != http.StatusPartialContent { + t.Fatalf("tenant B repeat: status = %d, want 206", rec.Code) + } + if got := originCalls.Load(); got != 2 { + t.Fatalf("origin calls after tenant B repeat = %d, want 2", got) + } +} diff --git a/cmd/cache-proxy/summary_pull_test.go b/cmd/cache-proxy/summary_pull_test.go index 9672a4b3..80ca22d0 100644 --- a/cmd/cache-proxy/summary_pull_test.go +++ b/cmd/cache-proxy/summary_pull_test.go @@ -42,7 +42,7 @@ func TestBlockPresentRejectsUntrackedDiskFile(t *testing.T) { if err != nil { t.Fatal(err) } - key := BlockKey("https://example.invalid/object", 0, 1024) + key := BlockKey("", "https://example.invalid/object", 0, 1024) if err := os.WriteFile(filepath.Join(store.dir, key), []byte("stray"), 0o640); err != nil { t.Fatal(err) } @@ -707,7 +707,7 @@ func TestBlockRequestSharesOneConfirmationBudgetAcrossAllBlocks(t *testing.T) { target := origin.URL + "/bucket/large.parquet" keys := make([]string, 8) for i := range keys { - keys[i] = BlockKey(target, int64(i), blockSize) + keys[i] = BlockKey("", target, int64(i), blockSize) } var hasCalls, getCalls int32 @@ -752,7 +752,7 @@ func TestBlockRequestStopsSummaryLookupsAfterGETBudgetExhaustion(t *testing.T) { target := origin.URL + "/bucket/get-budget.parquet" keys := make([]string, 4) for i := range keys { - keys[i] = BlockKey(target, int64(i), blockSize) + keys[i] = BlockKey("", target, int64(i), blockSize) } var hasCalls, getCalls int32 From da5747abee38cc353599dc6372b0ccd5a48eaa8c Mon Sep 17 00:00:00 2001 From: fuziontech Date: Thu, 3 Sep 2026 23:46:42 +0000 Subject: [PATCH 2/2] test(cache-proxy): appease staticcheck SA4000 in tenant-scope key test Co-authored-by: Shelley --- cmd/cache-proxy/scope_test.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/cmd/cache-proxy/scope_test.go b/cmd/cache-proxy/scope_test.go index a7561342..7bfcf352 100644 --- a/cmd/cache-proxy/scope_test.go +++ b/cmd/cache-proxy/scope_test.go @@ -56,17 +56,19 @@ func TestCacheKeyTenantScope(t *testing.T) { if a != CacheKey("ASIAAAAA", url, rng) { t.Fatal("CacheKey must be deterministic for a fixed scope") } - if CacheKey("", url, rng) == a || CacheKey("", url, rng) == b { + unsigned := CacheKey("", url, rng) + if unsigned == a || unsigned == b { t.Fatal("unsigned requests must not collide with a signed tenant namespace") } - if CacheKey("", url, rng) != CacheKey("", url, rng) { + if CacheKey("", url, rng) != unsigned { t.Fatal("unsigned requests must share one cache namespace") } if BlockKey("ASIAAAAA", url, 0, 8<<20) == BlockKey("ASIABBBB", url, 0, 8<<20) { t.Fatal("different access key IDs must produce different block keys") } - if BlockKey("", url, 0, 8<<20) != BlockKey("", url, 0, 8<<20) { + unsignedBlock := BlockKey("", url, 0, 8<<20) + if BlockKey("", url, 0, 8<<20) != unsignedBlock { t.Fatal("unsigned requests must share one block namespace") } }