diff --git a/README.md b/README.md index 72be0794..8cf29259 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,7 @@ When configuring S3 clients, ensure path-style addressing is enabled. See [docs/ - A dropped whole-object cache listener or upstream body-read error does not publish the partial body as a new cache entry - In tiered mode, a PUT routed to the local tier checks a concrete SigV4 `X-Amz-Content-Sha256` digest after admission and any local precondition pass and the full body passes its declared-length and any Content-MD5 checks; conditional writes against upstream-tier markers, `UNSIGNED-PAYLOAD`, and aws-chunked uploads keep their existing paths - PUT/DELETE operations invalidate affected cache entries. For cache-enabled bulk DeleteObjects requests with a nonempty parsed key list, TAG completes an upstream 2xx response only after the post-success invalidation attempts return, bounded by a 10-second fence timeout; a fence that does not converge in time is logged as a warning and the 2xx is still returned. +- Fully captured successful multipart completions are cached for five seconds when TAG can identify the caller; replay requires local validation for the same access key and exact bucket, object key, and upload ID - Legacy ordered metadata populates are refused when the tombstone cannot be read or is malformed, rather than assuming no invalidation. The detached metadata warm remains best-effort, so a successful S3 write response is unchanged. - If a whole-object cache write returns an error during a successful GET, the upstream response is unchanged and the cache-populate slot becomes available after relay cleanup, so a later eligible GET can retry. A store call that never returns has no completion guarantee. diff --git a/cache/completion.go b/cache/completion.go index 6aa9021f..4c036861 100644 --- a/cache/completion.go +++ b/cache/completion.go @@ -3,6 +3,8 @@ package cache import ( "context" "net/http" + "net/url" + "strings" json "github.com/goccy/go-json" "github.com/rs/zerolog/log" @@ -17,8 +19,11 @@ const ( completionKeyPrefix = "complete:" ) -// CompletionEntry stores a cached CompleteMultipartUpload response. +// CompletionEntry stores a cached CompleteMultipartUpload response, bound to the +// access key that completed the upload. An entry without an access key (written +// before identity binding existed) is never replayed. type CompletionEntry struct { + AccessKey string `json:"access_key,omitempty"` StatusCode int `json:"status_code"` Headers map[string]string `json:"headers"` Body []byte `json:"body"` @@ -26,12 +31,20 @@ type CompletionEntry struct { // MakeCompletionKey creates a cache key for a completion response. func MakeCompletionKey(bucket, key, uploadId string) string { + if strings.ContainsAny(bucket, "|") || strings.ContainsAny(key, "|") || strings.ContainsAny(uploadId, "|") { + // Keep the legacy form for ordinary tuples. For components containing its + // delimiter, escape every component and use a separator that QueryEscape + // also escapes. The encoded form contains no '|', so it cannot alias a + // legacy key, which always contains two delimiters. + return completionKeyPrefix + url.QueryEscape(bucket) + ":" + url.QueryEscape(key) + ":" + url.QueryEscape(uploadId) + } return completionKeyPrefix + bucket + "|" + key + "|" + uploadId } -// GetCompletion retrieves a cached completion response. -func (c *Cache) GetCompletion(ctx context.Context, bucket, key, uploadId string) (*CompletionEntry, bool, error) { - if !c.IsEnabled() { +// GetCompletion retrieves the cached completion response bound to accessKey. +// Entries bound to another principal, or to none, are reported as misses. +func (c *Cache) GetCompletion(ctx context.Context, bucket, key, uploadId, accessKey string) (*CompletionEntry, bool, error) { + if !c.IsEnabled() || accessKey == "" { return nil, false, nil } @@ -52,14 +65,20 @@ func (c *Cache) GetCompletion(ctx context.Context, bucket, key, uploadId string) log.Debug().Err(err).Str("key", cacheKey).Msg("Completion cache decode error") return nil, false, nil // Treat decode errors as cache miss } + if entry.AccessKey == "" || entry.AccessKey != accessKey { + log.Debug().Str("bucket", bucket).Str("uploadId", uploadId).Msg("Completion cache entry bound to another principal") + return nil, false, nil + } log.Debug().Str("bucket", bucket).Str("uploadId", uploadId).Msg("Completion cache hit") return &entry, true, nil } -// PutCompletion stores a completion response in cache. -func (c *Cache) PutCompletion(ctx context.Context, bucket, key, uploadId string, statusCode int, headers http.Header, body []byte) error { - if !c.IsEnabled() { +// PutCompletion stores a completion response in cache, bound to the access key +// that successfully completed the upload. Responses without an identified +// principal are not stored, so they can never be replayed. +func (c *Cache) PutCompletion(ctx context.Context, bucket, key, uploadId, accessKey string, statusCode int, headers http.Header, body []byte) error { + if !c.IsEnabled() || accessKey == "" { return nil } @@ -72,6 +91,7 @@ func (c *Cache) PutCompletion(ctx context.Context, bucket, key, uploadId string, } entry := CompletionEntry{ + AccessKey: accessKey, StatusCode: statusCode, Headers: headerMap, Body: body, diff --git a/cache/completion_test.go b/cache/completion_test.go index c3dada0b..fdbd8b8d 100644 --- a/cache/completion_test.go +++ b/cache/completion_test.go @@ -20,7 +20,7 @@ func TestCompletionCache(t *testing.T) { // Test 1: GetCompletion returns not found for non-existent entry t.Run("GetCompletion_NotFound", func(t *testing.T) { - entry, found, err := cache.GetCompletion(ctx, bucket, key, uploadId) + entry, found, err := cache.GetCompletion(ctx, bucket, key, uploadId, "client-a") if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -42,13 +42,13 @@ func TestCompletionCache(t *testing.T) { body := []byte(`"abc123"`) statusCode := 200 - err := cache.PutCompletion(ctx, bucket, key, uploadId, statusCode, headers, body) + err := cache.PutCompletion(ctx, bucket, key, uploadId, "client-a", statusCode, headers, body) if err != nil { t.Fatalf("PutCompletion failed: %v", err) } - // Retrieve and verify - entry, found, err := cache.GetCompletion(ctx, bucket, key, uploadId) + // Retrieve and verify for the caller bound to this completion. + entry, found, err := cache.GetCompletion(ctx, bucket, key, uploadId, "client-a") if err != nil { t.Fatalf("GetCompletion failed: %v", err) } @@ -70,11 +70,31 @@ func TestCompletionCache(t *testing.T) { if string(entry.Body) != string(body) { t.Errorf("expected body %q, got %q", body, entry.Body) } + + if entry.AccessKey != "client-a" { + t.Errorf("expected entry bound to %q, got %q", "client-a", entry.AccessKey) + } + + wrongEntry, wrongFound, err := cache.GetCompletion(ctx, bucket, key, uploadId, "client-b") + if err != nil { + t.Fatalf("GetCompletion for a different principal failed: %v", err) + } + if wrongFound || wrongEntry != nil { + t.Fatalf("completion for a different principal was returned: entry=%+v found=%t", wrongEntry, wrongFound) + } + + noEntry, noFound, err := cache.GetCompletion(ctx, bucket, key, uploadId, "") + if err != nil { + t.Fatalf("GetCompletion without a principal failed: %v", err) + } + if noFound || noEntry != nil { + t.Fatalf("completion was returned to an unidentified caller: entry=%+v found=%t", noEntry, noFound) + } }) // Test 3: Different uploadId returns not found t.Run("Different_UploadId_NotFound", func(t *testing.T) { - entry, found, err := cache.GetCompletion(ctx, bucket, key, "different-upload-id") + entry, found, err := cache.GetCompletion(ctx, bucket, key, "different-upload-id", "client-a") if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -86,7 +106,21 @@ func TestCompletionCache(t *testing.T) { } }) - // Test 4: MakeCompletionKey format + // Test 4: Responses without an authenticated principal cannot be replayed. + t.Run("PutCompletion_WithoutPrincipal", func(t *testing.T) { + if err := cache.PutCompletion(ctx, bucket, key, "anonymous-upload", "", http.StatusOK, nil, nil); err != nil { + t.Fatalf("PutCompletion failed: %v", err) + } + entry, found, err := cache.GetCompletion(ctx, bucket, key, "anonymous-upload", "client-a") + if err != nil { + t.Fatalf("GetCompletion failed: %v", err) + } + if found || entry != nil { + t.Fatalf("completion without a principal was cached: entry=%+v found=%t", entry, found) + } + }) + + // Test 5: MakeCompletionKey format t.Run("MakeCompletionKey_Format", func(t *testing.T) { key := MakeCompletionKey("bucket", "path/to/object", "upload-abc") expected := "complete:bucket|path/to/object|upload-abc" @@ -94,6 +128,21 @@ func TestCompletionCache(t *testing.T) { t.Errorf("expected key %q, got %q", expected, key) } }) + + // Test 6: Components containing the legacy separator cannot alias. + t.Run("MakeCompletionKey_DelimiterAliases", func(t *testing.T) { + first := MakeCompletionKey("bucket", "object", "id|suffix") + second := MakeCompletionKey("bucket", "object|id", "suffix") + if first == second { + t.Fatalf("distinct completion tuples share key %q", first) + } + if first != "complete:bucket:object:id%7Csuffix" { + t.Errorf("unexpected encoded key for first tuple: %q", first) + } + if second != "complete:bucket:object%7Cid:suffix" { + t.Errorf("unexpected encoded key for second tuple: %q", second) + } + }) } func TestCompletionCache_DisabledCache(t *testing.T) { @@ -103,7 +152,7 @@ func TestCompletionCache_DisabledCache(t *testing.T) { ctx := context.Background() // GetCompletion should return not found without error - entry, found, err := cache.GetCompletion(ctx, "bucket", "key", "uploadId") + entry, found, err := cache.GetCompletion(ctx, "bucket", "key", "uploadId", "client") if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -115,7 +164,7 @@ func TestCompletionCache_DisabledCache(t *testing.T) { } // PutCompletion should succeed without error (no-op) - err = cache.PutCompletion(ctx, "bucket", "key", "uploadId", 200, nil, nil) + err = cache.PutCompletion(ctx, "bucket", "key", "uploadId", "client", 200, nil, nil) if err != nil { t.Fatalf("unexpected error for disabled cache: %v", err) } diff --git a/docs/s3-compatibility.md b/docs/s3-compatibility.md index 5d66f1bc..e4b5f7b7 100644 --- a/docs/s3-compatibility.md +++ b/docs/s3-compatibility.md @@ -53,7 +53,7 @@ All bucket operations support both `/{bucket}` and `/{bucket}/` path forms for c | InitiateMultipartUpload | POST | `/{bucket}/{key}?uploads` | | | UploadPart | PUT | `/{bucket}/{key}?uploadId=&partNumber=` | | | UploadPartCopy | PUT | `/{bucket}/{key}?uploadId=&partNumber=` | Detected via `X-Amz-Copy-Source` header | -| CompleteMultipartUpload | POST | `/{bucket}/{key}?uploadId=` | Idempotent — successful completions cached | +| CompleteMultipartUpload | POST | `/{bucket}/{key}?uploadId=` | | | AbortMultipartUpload | DELETE | `/{bucket}/{key}?uploadId=` | | | ListParts | GET | `/{bucket}/{key}?uploadId=` | | @@ -92,6 +92,7 @@ Key caching behaviors: - **Conditional requests** — `If-None-Match` and `If-Modified-Since` are evaluated against cached metadata and can return 304 without hitting Tigris. - **Write-through invalidation** — In proxy mode, PUT, DELETE, CopyObject, and DeleteObjects invalidate affected cache entries before forwarding. For cache-enabled DeleteObjects requests with a nonempty parsed key list, TAG completes an upstream 2xx only after the post-success invalidation attempts return, bounded by a 10-second fence timeout; a fence that does not converge in time is logged as a warning and the 2xx is still returned. - **Tombstone protection** — A short-lived tombstone is written on DELETE to prevent in-flight background cache writes from resurrecting deleted objects. +- **Multipart completion replay** — A fully captured successful response is cached for five seconds when TAG can identify a caller. A replay must match the exact bucket, object key, and upload ID and pass local validation for the saved access key; transparent mode also checks the caller's current bucket authorization. Entries without a saved caller identity are treated as misses; other requests follow normal forwarding and authorization. ## Addressing Style diff --git a/proxy/multipart_complete_test.go b/proxy/multipart_complete_test.go index b7f84ff7..69dbe2f0 100644 --- a/proxy/multipart_complete_test.go +++ b/proxy/multipart_complete_test.go @@ -16,8 +16,19 @@ const ( mpUpload = "upload-123" ) +// completeMultipartRequest builds a completion signed as the mock-validated +// principal "access": the Authorization header identifies the caller and the +// content hash lets the request be validated locally (the mock accepts any). func completeMultipartRequest() *http.Request { - return httptest.NewRequest(http.MethodPost, "/"+mpBucket+"/"+mpKey+"?uploadId="+mpUpload, nil) + return completeMultipartRequestAs("access") +} + +func completeMultipartRequestAs(accessKey string) *http.Request { + r := httptest.NewRequest(http.MethodPost, "/"+mpBucket+"/"+mpKey+"?uploadId="+mpUpload, nil) + r.Header.Set("Authorization", "AWS4-HMAC-SHA256 Credential="+accessKey+"/20260101/us-east-1/s3/aws4_request, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=deadbeef") + r.Header.Set("X-Amz-Content-Sha256", "UNSIGNED-PAYLOAD") + r.Header.Set("X-Amz-Date", "20260101T000000Z") + return r } // completion result body (a genuine success — not an document). @@ -89,7 +100,7 @@ func TestHandleCompleteMultipartUpload_200ErrorBodyKeepsRacingRefill(t *testing. } // A failed completion must not be cached as a successful idempotent replay. - if _, found, _ := c.GetCompletion(context.Background(), mpBucket, mpKey, mpUpload); found { + if _, found, _ := c.GetCompletion(context.Background(), mpBucket, mpKey, mpUpload, "access"); found { t.Error("200-with-error-body completion was cached as a successful completion — an idempotent replay would return the error as success") } } @@ -119,7 +130,9 @@ func TestHandleCompleteMultipartUpload_InvalidatesBeforeForward(t *testing.T) { } // A successful completion is cached for idempotent replay: a second call with the -// same uploadId must return the stored response without forwarding upstream again. +// same uploadId by the same validated principal must return the stored response +// without forwarding upstream again. The entry is bound to that principal, so a +// different caller for the same uploadId is forwarded upstream instead. func TestHandleCompleteMultipartUpload_SuccessCachedForIdempotentReplay(t *testing.T) { var forwards int succeed := func(ctx context.Context, w http.ResponseWriter, r *http.Request) (*ResponseCapture, error) { @@ -143,4 +156,12 @@ func TestHandleCompleteMultipartUpload_SuccessCachedForIdempotentReplay(t *testi if forwards != 1 { t.Errorf("upstream forwarded %d times; the second identical completion must be served from the idempotency cache", forwards) } + + w := httptest.NewRecorder() + if err := svc.HandleCompleteMultipartUpload(w, completeMultipartRequestAs("other")); err != nil { + t.Fatalf("HandleCompleteMultipartUpload as another principal: %v", err) + } + if forwards != 2 { + t.Errorf("upstream forwarded %d times; a completion by another principal must not be served from a caller-bound cache entry", forwards) + } } diff --git a/proxy/reinvalidation_test.go b/proxy/reinvalidation_test.go index 37483045..aa444494 100644 --- a/proxy/reinvalidation_test.go +++ b/proxy/reinvalidation_test.go @@ -1,7 +1,6 @@ package proxy import ( - "bytes" "context" "errors" "net/http" @@ -542,23 +541,6 @@ func (c *stalledMetaDeleteClient) Delete(ctx context.Context, key string) error return c.CacheClient.Delete(ctx, key) } -type lockedLogBuffer struct { - mu sync.Mutex - buf bytes.Buffer -} - -func (b *lockedLogBuffer) Write(p []byte) (int, error) { - b.mu.Lock() - defer b.mu.Unlock() - return b.buf.Write(p) -} - -func (b *lockedLogBuffer) String() string { - b.mu.Lock() - defer b.mu.Unlock() - return b.buf.String() -} - // A cache peer that never answers the post-success fence must not hold the // staged DeleteObjects response past bulkDeleteFenceTimeout; the client still // gets the upstream 2xx byte-exact, and the expiry is logged with the bucket diff --git a/proxy/service.go b/proxy/service.go index 8f1186e2..4208b1ac 100644 --- a/proxy/service.go +++ b/proxy/service.go @@ -12,6 +12,7 @@ import ( "github.com/hashicorp/golang-lru/v2/expirable" "github.com/rs/zerolog/log" + "github.com/tigrisdata/tag/auth" "github.com/tigrisdata/tag/cache" "github.com/tigrisdata/tag/config" "github.com/tigrisdata/tag/metrics" @@ -1209,8 +1210,10 @@ func (s *Service) HandlePassthrough(w http.ResponseWriter, r *http.Request) erro } // HandleCompleteMultipartUpload handles CompleteMultipartUpload with idempotency caching. -// This caches successful completion responses in ocache to support idempotent calls, -// matching tigris-os behavior where a second CompleteMultipartUpload call returns success. +// A fully captured successful response is cached for five seconds, bound to the +// locally validated caller, and replayed only to that same principal for the exact +// bucket, key, and upload ID. Any other caller is forwarded upstream, which gives +// the authoritative answer and never leaks the completed object's ETag or existence. func (s *Service) HandleCompleteMultipartUpload(w http.ResponseWriter, r *http.Request) error { bucket, key := ParseBucketKey(r) uploadId := r.URL.Query().Get("uploadId") @@ -1218,13 +1221,24 @@ func (s *Service) HandleCompleteMultipartUpload(w http.ResponseWriter, r *http.R log.Debug().Str("bucket", bucket).Str("key", key).Str("uploadId", uploadId).Msg("HandleCompleteMultipartUpload") + // Identify the caller before forwarding. Only a locally validated principal can + // replay a cached completion or have one saved for it, and the identity must be + // taken now: a long completion can outlive the SigV4 age window before its + // response is captured, so it could not be validated afterwards. + var principal string + if s.cache.IsEnabled() { + principal = s.completionPrincipal(r) + } + // Check ocache first for idempotent completion (works across TAG pods). // A replay returns the already-completed response without touching upstream and // without changing the object, so it needs no read-cache invalidation — the first // completion (below) already invalidated it. - if s.cache.IsEnabled() { - entry, found, err := s.cache.GetCompletion(ctx, bucket, key, uploadId) - if err == nil && found { + if principal != "" { + entry, found, err := s.cache.GetCompletion(ctx, bucket, key, uploadId, principal) + // The cache read may have waited on a remote peer, during which an upstream + // 403 can revoke the caller's grant; re-validate at the replay decision point. + if err == nil && found && s.completionPrincipal(r) == principal { log.Debug().Str("uploadId", uploadId).Msg("CompleteMultipartUpload cache hit - returning cached response") for k, v := range entry.Headers { w.Header().Set(k, v) @@ -1306,11 +1320,12 @@ func (s *Service) HandleCompleteMultipartUpload(w http.ResponseWriter, r *http.R } } - // Cache successful completions in ocache for idempotent replays. Only cache a - // genuine success (not a 200-with- body) that was fully captured, so we - // never replay a corrupted or error response as a successful completion. - if completed && capture.Complete { - if cacheErr := s.cache.PutCompletion(ctx, bucket, key, uploadId, capture.StatusCode, capture.Headers, capture.Body); cacheErr != nil { + // Cache a genuine, fully captured success (not a 200-with- body) for + // idempotent replays, bound to the caller validated at request start. A + // completion whose caller could not be identified locally is not cached, so + // it can never be replayed to anyone. + if completed && capture.Complete && principal != "" { + if cacheErr := s.cache.PutCompletion(ctx, bucket, key, uploadId, principal, capture.StatusCode, capture.Headers, capture.Body); cacheErr != nil { log.Debug().Err(cacheErr).Msg("Failed to cache completion response") // Don't fail the request if caching fails } @@ -1319,6 +1334,58 @@ func (s *Service) HandleCompleteMultipartUpload(w http.ResponseWriter, r *http.R return nil } +// completionPrincipal returns the access key of a caller the forwarder validates +// locally, or "" when the caller cannot be identified. In signing mode that is a +// signature check against the credential store; in transparent mode it is the +// signature check against learned keys plus the caller's current bucket grant — +// the same validation that gates serving cached objects. The access key comes +// from the client's own auth info in both modes (transparent mode's +// ValidateAndGetCredentials returns the proxy's upstream credentials). +func (s *Service) completionPrincipal(r *http.Request) string { + validationRequest, ok := completionRequestForLocalValidation(r) + if !ok { + return "" + } + result, _, _, err := s.forwarder.ValidateAndGetCredentials(validationRequest) + if err != nil || result != AuthValidated { + return "" + } + authInfo, err := auth.ParseAuthInfo(r) + if err != nil { + return "" + } + return authInfo.AccessKey +} + +// completionRequestForLocalValidation returns the request to validate locally. +// The validator requires X-Amz-Content-Sha256 on non-GET requests, but a presigned +// POST completion legitimately omits it and is signed over UNSIGNED-PAYLOAD; supply +// that marker on a clone so validation can run without changing the request that +// is forwarded upstream. A request that neither carries the hash nor is presigned +// cannot be validated locally. +func completionRequestForLocalValidation(r *http.Request) (*http.Request, bool) { + if r.Header.Get("X-Amz-Content-Sha256") != "" { + return r, true + } + if !auth.IsPresignedRequest(r) { + return r, false + } + authInfo, err := auth.ParseAuthInfo(r) + if err != nil || !authInfo.IsPresigned { + return r, false + } + for _, header := range authInfo.SignedHeaders { + if strings.EqualFold(header, "x-amz-content-sha256") { + return r, false + } + } + + validationRequest := r.Clone(r.Context()) + validationRequest.Header = r.Header.Clone() + validationRequest.Header.Set("X-Amz-Content-Sha256", "UNSIGNED-PAYLOAD") + return validationRequest, true +} + // ParseBucketKey extracts bucket and key from request path. func ParseBucketKey(r *http.Request) (bucket, key string) { path := strings.TrimPrefix(r.URL.Path, "/") diff --git a/tests/integration/completion_replay_auth_test.go b/tests/integration/completion_replay_auth_test.go new file mode 100644 index 00000000..bae7cca6 --- /dev/null +++ b/tests/integration/completion_replay_auth_test.go @@ -0,0 +1,772 @@ +package integration + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + cacheclient "github.com/tigrisdata/ocache/client" + "github.com/tigrisdata/tag/auth" + "github.com/tigrisdata/tag/cache" + "github.com/tigrisdata/tag/proxy" +) + +const ( + completionReplayKey = "object" + completionReplayUploadID = "upload-123" + completionReplayMissUploadID = "missing-upload" + completionReplaySlowUploadID = "near-expiry-upload" + completionReplayUnboundID = "unbound-upload" + completionReplayAliasID = "id|suffix" + completionReplayAliasKey = "object|id" + completionReplayAliasUpload = "suffix" + completionReplayAccessA = TestAccessKey + completionReplaySecretA = TestSecretKey + completionReplayAccessB = "CLIENT-B" + completionReplaySecretB = "secret-b" +) + +var ( + completionReplayRequestBody = []byte(`1"part"`) + completionReplaySuccessBody = []byte(`"completed"`) +) + +type completionReplayResponse struct { + status int + contentType string + etag string + body string +} + +type completionReplayAttempt struct { + response completionReplayResponse + err error +} + +type completionReplayUpstreamFixture struct { + handler http.HandlerFunc + completionCalls atomic.Int32 + slowCalls atomic.Int32 + revocationCalls atomic.Int32 + mainCalls atomic.Int32 + aliasCalls atomic.Int32 + slowEntered chan struct{} + slowRelease chan struct{} + releaseOnce sync.Once +} + +func (u *completionReplayUpstreamFixture) releaseSlow() { + u.releaseOnce.Do(func() { close(u.slowRelease) }) +} + +type completionReplayReadGate struct { + cacheclient.CacheClient + key string + armed atomic.Bool + entered chan struct{} + release chan struct{} + releaseOnce sync.Once +} + +func (g *completionReplayReadGate) releaseRead() { + g.releaseOnce.Do(func() { close(g.release) }) +} + +func (g *completionReplayReadGate) Get(ctx context.Context, key string) ([]byte, error) { + if key == g.key && g.armed.CompareAndSwap(true, false) { + close(g.entered) + select { + case <-g.release: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return g.CacheClient.Get(ctx, key) +} + +func TestCompleteMultipartUploadReplayAuthorization(t *testing.T) { + t.Run("transparent", func(t *testing.T) { + testCompleteMultipartUploadReplayAuthorization(t, true) + }) + t.Run("signing", func(t *testing.T) { + testCompleteMultipartUploadReplayAuthorization(t, false) + }) + t.Run("transparent-cold-auth", func(t *testing.T) { + testColdCompletionReplay(t) + }) + t.Run("transparent-revocation-during-cache-read", func(t *testing.T) { + testCompletionReplayGrantRevocationDuringRead(t) + }) +} + +func testCompleteMultipartUploadReplayAuthorization(t *testing.T, transparent bool) { + t.Helper() + bucket := completionReplayBucket(t) + otherBucket := bucket + "-other" + upstream := newCompletionReplayUpstream(t, bucket, transparent) + defer upstream.releaseSlow() + completionCalls := &upstream.completionCalls + env := newCompletionReplayEnvironment(t, transparent, upstream.handler) + defer env.Close() + + if transparent { + // Replay binds to a principal TAG can validate locally: learned signing keys + // plus a bucket grant, which only a successful object read establishes. Seed + // both principals as if they had already read from the bucket. + seedTransparentPrincipal(t, env, completionReplayAccessA, completionReplaySecretA, bucket) + seedTransparentPrincipal(t, env, completionReplayAccessB, completionReplaySecretB, bucket) + // Exercise the cache's bucket dimension even when the same principal is + // authorized for both buckets. + env.AuthzCache.Grant(completionReplayAccessA, otherBucket) + } else { + env.CredStore.AddCredential(completionReplayAccessB, completionReplaySecretB) + } + + client := env.TAGServer.Client() + unboundEntry := completionReplayResponse{status: http.StatusOK, contentType: "application/xml", etag: `"legacy-etag"`, body: ``} + seedUnboundCompletionEntry(t, env, bucket, completionReplayKey, completionReplayUnboundID, unboundEntry) + before := completionCalls.Load() + unbound := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUnboundID, completionReplayAccessA, completionReplaySecretA, false, false) + assertCompletionReplayDenied(t, "completion without a saved caller identity", unbound, unboundEntry) + assertCompletionReplayForwarded(t, "completion without a saved caller identity", unbound, completionCalls.Load(), before, http.StatusForbidden) + if transparent { + // The forwarded denial revoked the caller's bucket grant; restore it so the + // completion below is made by a principal TAG can still validate locally. + env.AuthzCache.Grant(completionReplayAccessA, bucket) + } + + before = completionCalls.Load() + first := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + if first.status != http.StatusOK || first.contentType != "application/xml" || first.etag != `"completed-etag"` || first.body != string(completionReplaySuccessBody) { + t.Fatalf("initial completion = %#v; want successful upstream response", first) + } + if got := completionCalls.Load(); got != before+1 { + t.Fatalf("initial completion made %d upstream completion calls; want 1", got-before) + } + + before = completionCalls.Load() + samePrincipal := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + if samePrincipal != first { + t.Errorf("validated same-principal replay = %#v; want saved response %#v", samePrincipal, first) + } + assertCompletionReplayCallCount(t, "same-principal replay", completionCalls.Load(), before) + + if transparent { + data, err := env.EmbeddedCache.Get(context.Background(), cache.MakeCompletionKey(bucket, completionReplayKey, completionReplayUploadID)) + if err != nil || data == nil { + t.Fatalf("completion cache entry was not live before authorization revocation: data=%d bytes err=%v", len(data), err) + } + var entry struct { + StatusCode int `json:"status_code"` + Body []byte `json:"body"` + } + if err := json.Unmarshal(data, &entry); err != nil || entry.StatusCode != first.status || string(entry.Body) != first.body { + t.Fatalf("completion cache entry = status %d body %q err=%v; want live saved response", entry.StatusCode, entry.Body, err) + } + env.AuthzCache.Revoke(completionReplayAccessA, bucket) + before = completionCalls.Load() + unauthorizedBucket := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + assertCompletionReplayDenied(t, "same principal without bucket authorization", unauthorizedBucket, first) + assertCompletionReplayForwarded(t, "same principal without bucket authorization", unauthorizedBucket, completionCalls.Load(), before, http.StatusForbidden) + env.AuthzCache.Grant(completionReplayAccessA, bucket) + } + + validator := auth.NewRequestValidator(env.CredStore) + if transparent { + validator = auth.NewRequestValidator(env.DerivedKeyStore) + } + before = completionCalls.Load() + presigned := sendPresignedCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, time.Now().UTC().Truncate(time.Second), validator) + if presigned != first { + t.Errorf("valid same-principal presigned retry = %#v; want saved response %#v", presigned, first) + } + assertCompletionReplayCallCount(t, "presigned replay", completionCalls.Load(), before) + + aliasSeed := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayAliasID, completionReplayAccessA, completionReplaySecretA, false, false) + if aliasSeed.status != http.StatusOK || aliasSeed.etag != `"completed-etag"` || aliasSeed.body != string(completionReplaySuccessBody) { + t.Fatalf("delimiter-alias seed completion = %#v; want successful upstream response", aliasSeed) + } + aliasCallsBefore := completionCalls.Load() + alias := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayAliasKey, completionReplayAliasUpload, completionReplayAccessA, completionReplaySecretA, false, false) + assertCompletionReplayDenied(t, "same-principal delimiter-alias tuple", alias, aliasSeed) + assertCompletionReplayForwarded(t, "same-principal delimiter-alias tuple", alias, completionCalls.Load(), aliasCallsBefore, http.StatusNotFound) + + before = completionCalls.Load() + otherBucketResponse := sendCompletionReplayRequest(t, client, env.TAGServer.URL, otherBucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + assertCompletionReplayDenied(t, "same principal in another bucket", otherBucketResponse, first) + assertCompletionReplayForwarded(t, "same principal in another bucket", otherBucketResponse, completionCalls.Load(), before, http.StatusForbidden) + + before = completionCalls.Load() + miss := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayMissUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + assertCompletionReplayDenied(t, "cache miss", miss, first) + if miss.status != http.StatusNotFound { + t.Errorf("cache miss status = %d; want upstream 404", miss.status) + } + assertCompletionReplayForwarded(t, "cache miss", miss, completionCalls.Load(), before, http.StatusNotFound) + + before = completionCalls.Load() + differentPrincipal := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessB, completionReplaySecretB, false, false) + assertCompletionReplayDenied(t, "different validated principal", differentPrincipal, first) + assertCompletionReplayForwarded(t, "different validated principal", differentPrincipal, completionCalls.Load(), before, http.StatusForbidden) + + before = completionCalls.Load() + anonymous := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, "", "", true, false) + assertCompletionReplayDenied(t, "anonymous caller", anonymous, first) + if transparent { + assertCompletionReplayForwarded(t, "anonymous caller", anonymous, completionCalls.Load(), before, http.StatusForbidden) + } else { + assertCompletionReplayNotForwarded(t, "signing-mode anonymous caller", completionCalls.Load(), before) + } + + before = completionCalls.Load() + badSignature := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, true) + assertCompletionReplayDenied(t, "bad-signature caller", badSignature, first) + if transparent { + assertCompletionReplayForwarded(t, "bad-signature caller", badSignature, completionCalls.Load(), before, http.StatusForbidden) + } else { + assertCompletionReplayNotForwarded(t, "signing-mode bad-signature caller", completionCalls.Load(), before) + } + + if transparent { + env.AuthzCache.Grant(completionReplayAccessA, bucket) + testCompletionReplayAfterSignatureWindow(t, env, bucket, upstream, validator) + } + +} + +func testColdCompletionReplay(t *testing.T) { + t.Helper() + bucket := completionReplayBucket(t) + upstream := newCompletionReplayUpstream(t, bucket, true) + defer upstream.releaseSlow() + env := NewTestEnvironmentWithTransparentAuth(t, upstream.handler) + defer env.Close() + if env.DerivedKeyStore.HasKey(completionReplayAccessA) || env.AuthzCache.IsAuthorized(completionReplayAccessA, bucket) { + t.Fatal("test did not begin with cold transparent authentication state") + } + + client := env.TAGServer.Client() + first := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + if first.status != http.StatusOK || first.etag != `"completed-etag"` || first.body != string(completionReplaySuccessBody) { + t.Fatalf("cold-auth completion = %#v; want upstream success", first) + } + if got := upstream.completionCalls.Load(); got != 1 { + t.Fatalf("cold-auth completion made %d upstream calls; want one", got) + } + if !env.DerivedKeyStore.HasKey(completionReplayAccessA) { + t.Fatal("successful upstream response did not teach local signing keys") + } + // A successful write teaches signing keys but grants no bucket authorization; + // only a successful object read does. The caller could not be validated locally + // at request start, so the completion was not bound to a principal and is not + // replayable: the same caller's retry goes upstream for the authoritative answer. + if env.AuthzCache.IsAuthorized(completionReplayAccessA, bucket) { + t.Fatal("successful completion granted bucket authorization; only object reads may") + } + + before := upstream.completionCalls.Load() + retry := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + assertCompletionReplayDenied(t, "cold-auth same-principal retry", retry, first) + assertCompletionReplayForwarded(t, "cold-auth same-principal retry", retry, upstream.completionCalls.Load(), before, http.StatusForbidden) +} + +func testCompletionReplayGrantRevocationDuringRead(t *testing.T) { + t.Helper() + bucket := completionReplayBucket(t) + upstream := newCompletionReplayUpstream(t, bucket, true) + defer upstream.releaseSlow() + + gate := &completionReplayReadGate{ + CacheClient: sharedEmbeddedCache, + key: cache.MakeCompletionKey(bucket, completionReplayKey, completionReplayUploadID), + entered: make(chan struct{}), + release: make(chan struct{}), + } + env := newTestEnvironmentWithTransparentAuthCacheClient(t, upstream.handler, gate) + defer env.Close() + defer gate.releaseRead() + seedTransparentPrincipal(t, env, completionReplayAccessA, completionReplaySecretA, bucket) + + client := env.TAGServer.Client() + first := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + if first.status != http.StatusOK || first.etag != `"completed-etag"` || first.body != string(completionReplaySuccessBody) { + t.Fatalf("initial completion = %#v; want a cached successful response", first) + } + if got := upstream.completionCalls.Load(); got != 1 { + t.Fatalf("initial completion made %d upstream calls; want 1", got) + } + data, err := env.EmbeddedCache.Get(context.Background(), gate.key) + if err != nil || data == nil { + t.Fatalf("completion cache entry is not live before the gated read: bytes=%d err=%v", len(data), err) + } + var stored struct { + AccessKey string `json:"access_key"` + StatusCode int `json:"status_code"` + } + if err := json.Unmarshal(data, &stored); err != nil || stored.AccessKey != completionReplayAccessA || stored.StatusCode != first.status { + t.Fatalf("completion cache binding = access_key %q, status %d, err %v; want the successful caller and response", stored.AccessKey, stored.StatusCode, err) + } + + gate.armed.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + replayRequest := signCompletionReplayRequestAt(t, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, time.Now().UTC()) + replayRequest = replayRequest.WithContext(ctx) + result := make(chan completionReplayAttempt, 1) + callsBefore := upstream.completionCalls.Load() + go func() { + response, err := doCompletionReplayRequest(client, replayRequest) + result <- completionReplayAttempt{response: response, err: err} + }() + + select { + case <-gate.entered: + case <-time.After(5 * time.Second): + t.Fatal("replay did not reach the completion-cache read gate") + } + if !env.AuthzCache.IsAuthorized(completionReplayAccessA, bucket) { + t.Fatal("bucket grant was absent before the concurrent upstream denial") + } + + revocationRequest, err := env.Signer.SignRequest(context.Background(), http.MethodGet, "/"+bucket+"/revoke", nil, "", completionReplayAccessA, completionReplaySecretA, http.Header{}) + if err != nil { + t.Fatalf("sign concurrent revocation request: %v", err) + } + revocation, err := doCompletionReplayRequest(client, revocationRequest) + if err != nil { + t.Fatalf("send concurrent revocation request: %v", err) + } + if revocation.status != http.StatusForbidden || upstream.revocationCalls.Load() != 1 { + t.Fatalf("concurrent upstream denial = status %d, calls %d; want one 403", revocation.status, upstream.revocationCalls.Load()) + } + if env.AuthzCache.IsAuthorized(completionReplayAccessA, bucket) { + t.Fatal("upstream 403 did not revoke the caller's bucket grant") + } + + gate.releaseRead() + attempt := <-result + if attempt.err != nil { + t.Fatalf("replay after revocation: %v", attempt.err) + } + assertCompletionReplayDenied(t, "replay after a grant was revoked during cache read", attempt.response, first) + assertCompletionReplayForwarded(t, "replay after a grant was revoked during cache read", attempt.response, upstream.completionCalls.Load(), callsBefore, http.StatusForbidden) +} + +func seedUnboundCompletionEntry(t *testing.T, env *TestEnvironment, bucket, key, uploadID string, response completionReplayResponse) { + t.Helper() + entry := cache.CompletionEntry{ + StatusCode: response.status, + Headers: map[string]string{ + "Content-Type": response.contentType, + "ETag": response.etag, + }, + Body: []byte(response.body), + } + data, err := json.Marshal(entry) + if err != nil { + t.Fatalf("marshal unbound completion entry: %v", err) + } + if err := env.EmbeddedCache.Put(context.Background(), cache.MakeCompletionKey(bucket, key, uploadID), data, int64(cache.CompletionCacheTTL)); err != nil { + t.Fatalf("seed unbound completion entry: %v", err) + } +} + +func completionReplayBucket(t *testing.T) string { + t.Helper() + return fmt.Sprintf("replay-%d", time.Now().UnixNano()) +} + +func newCompletionReplayEnvironment(t *testing.T, transparent bool, upstream http.HandlerFunc) *TestEnvironment { + t.Helper() + if transparent { + return NewTestEnvironmentWithTransparentAuth(t, upstream) + } + return NewTestEnvironmentWithCacheHandler(upstream) +} + +func newCompletionReplayUpstream(t *testing.T, bucket string, transparent bool) *completionReplayUpstreamFixture { + t.Helper() + var responseKeys string + if transparent { + responseKeys = completionReplaySigningKeysHeader(t, completionReplayAccessA, completionReplaySecretA) + } + upstream := &completionReplayUpstreamFixture{ + slowEntered: make(chan struct{}), + slowRelease: make(chan struct{}), + } + writeSuccess := func(w http.ResponseWriter, includeSigningKeys bool) { + if includeSigningKeys && responseKeys != "" { + w.Header().Set("X-Tigris-Proxy-Signing-Keys", responseKeys) + } + w.Header().Set("ETag", `"completed-etag"`) + w.Header().Set("Content-Type", "application/xml") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(completionReplaySuccessBody) + } + writeError := func(w http.ResponseWriter, status int) { + w.WriteHeader(status) + if status == http.StatusNotFound { + _, _ = io.WriteString(w, `NoSuchUpload`) + } else { + _, _ = io.WriteString(w, `AccessDenied`) + } + } + + upstream.handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet && r.URL.Path == "/"+bucket+"/revoke" { + upstream.revocationCalls.Add(1) + writeError(w, http.StatusForbidden) + return + } + if r.Method != http.MethodPost || r.URL.Query().Get("uploadId") == "" { + http.NotFound(w, r) + return + } + upstream.completionCalls.Add(1) + requestBucket, key := proxy.ParseBucketKey(r) + uploadID := r.URL.Query().Get("uploadId") + info, authErr := auth.ParseAuthInfo(r) + if requestBucket == bucket && key == completionReplayKey && uploadID == completionReplayUploadID && authErr == nil && info.AccessKey == completionReplayAccessA { + if upstream.mainCalls.Add(1) == 1 { + writeSuccess(w, true) + return + } + writeError(w, http.StatusForbidden) + return + } + if requestBucket == bucket && key == completionReplayKey && uploadID == completionReplayAliasID && authErr == nil && info.AccessKey == completionReplayAccessA { + if upstream.aliasCalls.Add(1) == 1 { + writeSuccess(w, true) + return + } + writeError(w, http.StatusForbidden) + return + } + if requestBucket == bucket && key == completionReplayAliasKey && uploadID == completionReplayAliasUpload { + writeError(w, http.StatusNotFound) + return + } + if requestBucket == bucket && key == completionReplayKey && uploadID == completionReplaySlowUploadID && authErr == nil && info.AccessKey == completionReplayAccessA { + if upstream.slowCalls.Add(1) == 1 { + close(upstream.slowEntered) + <-upstream.slowRelease + writeSuccess(w, false) + return + } + writeError(w, http.StatusNotFound) + return + } + if requestBucket == bucket && key == completionReplayKey && uploadID == completionReplayMissUploadID { + writeError(w, http.StatusNotFound) + return + } + writeError(w, http.StatusForbidden) + }) + return upstream +} + +func completionReplaySigningKeysHeader(t *testing.T, accessKey, secretKey string) string { + t.Helper() + credentials := auth.NewCredentialStore() + credentials.AddCredential(accessKey, secretKey) + now := time.Now().UTC() + entries := make([]auth.SigningKeyEntry, 0, 3) + for offset := -1; offset <= 1; offset++ { + date := now.AddDate(0, 0, offset).Format("20060102") + key, err := credentials.GetSigningKey(accessKey, date, TestRegion) + if err != nil { + t.Fatalf("derive response signing key for %s: %v", accessKey, err) + } + entries = append(entries, auth.SigningKeyEntry{Date: date, Region: TestRegion, SigningKey: hex.EncodeToString(key)}) + } + return encryptSigningKeysHeader(t, TestProxySecretKey, accessKey, entries) +} + +func seedTransparentPrincipal(t *testing.T, env *TestEnvironment, accessKey, secretKey, bucket string) { + t.Helper() + credentials := auth.NewCredentialStore() + credentials.AddCredential(accessKey, secretKey) + now := time.Now().UTC() + for offset := -1; offset <= 1; offset++ { + date := now.AddDate(0, 0, offset).Format("20060102") + key, err := credentials.GetSigningKey(accessKey, date, TestRegion) + if err != nil { + t.Fatalf("derive local signing key for %s: %v", accessKey, err) + } + env.DerivedKeyStore.Store(accessKey, date, TestRegion, key) + } + env.AuthzCache.Grant(accessKey, bucket) +} + +func assertCompletionReplayDenied(t *testing.T, name string, got, cached completionReplayResponse) { + t.Helper() + if got.status >= http.StatusOK && got.status < http.StatusMultipleChoices || got.etag == cached.etag || got.body == cached.body { + t.Errorf("%s received status=%d ETag=%q body=%q from a cached completion", name, got.status, got.etag, got.body) + } +} + +func assertCompletionReplayForwarded(t *testing.T, name string, response completionReplayResponse, got, before int32, wantStatus int) { + t.Helper() + if got != before+1 { + t.Errorf("%s made %d upstream calls; want 1 authorization request", name, got-before) + } + if response.status != wantStatus { + t.Errorf("forwarded %s response status = %d; want upstream status %d", name, response.status, wantStatus) + } +} + +func assertCompletionReplayNotForwarded(t *testing.T, name string, got, before int32) { + t.Helper() + if got != before { + t.Errorf("%s made %d upstream calls; want local rejection", name, got-before) + } +} + +func assertCompletionReplayCallCount(t *testing.T, name string, got, before int32) { + t.Helper() + if got != before { + t.Errorf("%s made %d upstream calls; want cached response", name, got-before) + } +} + +func sendCompletionReplayRequest(t *testing.T, client *http.Client, endpoint, bucket, key, uploadID, accessKey, secretKey string, anonymous, badSignature bool) completionReplayResponse { + t.Helper() + path := "/" + bucket + "/" + key + "?" + url.Values{"uploadId": {uploadID}}.Encode() + var req *http.Request + var err error + if anonymous { + req, err = http.NewRequest(http.MethodPost, endpoint+path, bytes.NewReader(completionReplayRequestBody)) + if err == nil { + req.Header.Set("Content-Type", "application/xml") + } + } else { + sum := sha256.Sum256(completionReplayRequestBody) + req, err = auth.NewRequestSigner(endpoint, TestRegion).SignRequest(context.Background(), http.MethodPost, path, bytes.NewReader(completionReplayRequestBody), hex.EncodeToString(sum[:]), accessKey, secretKey, http.Header{"Content-Type": {"application/xml"}}) + } + if err != nil { + t.Fatalf("create completion request: %v", err) + } + if badSignature { + authorization := req.Header.Get("Authorization") + marker := "Signature=" + i := strings.LastIndex(authorization, marker) + if i < 0 { + t.Fatalf("signed request has no %q field: %q", marker, authorization) + } + i += len(marker) + if i >= len(authorization) { + t.Fatalf("signed request has empty signature: %q", authorization) + } + if authorization[i] == '0' { + authorization = authorization[:i] + "1" + authorization[i+1:] + } else { + authorization = authorization[:i] + "0" + authorization[i+1:] + } + req.Header.Set("Authorization", authorization) + } + return executeCompletionReplayRequest(t, client, req) +} + +func sendPresignedCompletionReplayRequest(t *testing.T, client *http.Client, endpoint, bucket, key, uploadID, accessKey, secretKey string, signedAt time.Time, validator *auth.RequestValidator) completionReplayResponse { + t.Helper() + request := presignCompletionReplayRequest(t, endpoint, bucket, key, uploadID, accessKey, secretKey, signedAt, validator) + if request.Header.Get("X-Amz-Content-Sha256") != "" { + t.Fatal("presigned request unexpectedly includes X-Amz-Content-Sha256") + } + return executeCompletionReplayRequest(t, client, request) +} + +func executeCompletionReplayRequest(t *testing.T, client *http.Client, req *http.Request) completionReplayResponse { + t.Helper() + response, err := doCompletionReplayRequest(client, req) + if err != nil { + t.Fatalf("send completion request: %v", err) + } + return response +} + +func doCompletionReplayRequest(client *http.Client, req *http.Request) (completionReplayResponse, error) { + resp, err := client.Do(req) + if err != nil { + return completionReplayResponse{}, err + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return completionReplayResponse{}, err + } + return completionReplayResponse{status: resp.StatusCode, contentType: resp.Header.Get("Content-Type"), etag: resp.Header.Get("ETag"), body: string(body)}, nil +} + +func presignCompletionReplayRequest(t *testing.T, endpoint, bucket, key, uploadID, accessKey, secretKey string, signedAt time.Time, validator *auth.RequestValidator) *http.Request { + t.Helper() + path := "/" + bucket + "/" + key + query := url.Values{"uploadId": {uploadID}} + dateTime := signedAt.UTC().Format(auth.TimeFormat) + shortDate := signedAt.UTC().Format("20060102") + scope := shortDate + "/" + TestRegion + "/s3/aws4_request" + query.Set("X-Amz-Algorithm", "AWS4-HMAC-SHA256") + query.Set("X-Amz-Credential", accessKey+"/"+scope) + query.Set("X-Amz-Date", dateTime) + query.Set("X-Amz-Expires", "300") + query.Set("X-Amz-SignedHeaders", "host") + request, err := http.NewRequest(http.MethodPost, endpoint+path+"?"+query.Encode(), bytes.NewReader(completionReplayRequestBody)) + if err != nil { + t.Fatalf("create presigned completion request: %v", err) + } + request.Host = request.URL.Host + request.Header.Set("Content-Type", "application/xml") + canonicalRequest := strings.Join([]string{ + request.Method, + request.URL.EscapedPath(), + query.Encode(), + "host:" + request.Host + "\n", + "host", + "UNSIGNED-PAYLOAD", + }, "\n") + canonicalHash := sha256.Sum256([]byte(canonicalRequest)) + stringToSign := "AWS4-HMAC-SHA256\n" + dateTime + "\n" + scope + "\n" + hex.EncodeToString(canonicalHash[:]) + credentials := auth.NewCredentialStore() + credentials.AddCredential(accessKey, secretKey) + signingKey, err := credentials.GetSigningKey(accessKey, shortDate, TestRegion) + if err != nil { + t.Fatalf("derive presigned signing key: %v", err) + } + query.Set("X-Amz-Signature", hex.EncodeToString(hmacSHA256(signingKey, []byte(stringToSign)))) + request.URL.RawQuery = query.Encode() + validationRequest := request.Clone(request.Context()) + validationRequest.Header = request.Header.Clone() + validationRequest.Header.Set("X-Amz-Content-Sha256", "UNSIGNED-PAYLOAD") + if got, err := validator.ValidateRequest(validationRequest); err != nil || got != accessKey { + t.Fatalf("presigned signature invalid with its standard unsigned-payload hash: accessKey=%q err=%v", got, err) + } + return request +} + +func signCompletionReplayRequestAt(t *testing.T, endpoint, bucket, key, uploadID, accessKey, secretKey string, signedAt time.Time) *http.Request { + t.Helper() + path := "/" + bucket + "/" + key + "?" + url.Values{"uploadId": {uploadID}}.Encode() + request, err := http.NewRequest(http.MethodPost, endpoint+path, bytes.NewReader(completionReplayRequestBody)) + if err != nil { + t.Fatalf("create completion request: %v", err) + } + request.Host = request.URL.Host + request.Header.Set("Content-Type", "application/xml") + sum := sha256.Sum256(completionReplayRequestBody) + bodyHash := hex.EncodeToString(sum[:]) + request.Header.Set("X-Amz-Content-Sha256", bodyHash) + dateTime := signedAt.UTC().Format(auth.TimeFormat) + shortDate := signedAt.UTC().Format("20060102") + signedHeaders := "content-type;host;x-amz-content-sha256;x-amz-date" + request.Header.Set("X-Amz-Date", dateTime) + canonicalHeaders := strings.Join([]string{ + "content-type:application/xml", + "host:" + request.Host, + "x-amz-content-sha256:" + bodyHash, + "x-amz-date:" + dateTime, + }, "\n") + "\n" + canonicalRequest := strings.Join([]string{ + request.Method, + request.URL.EscapedPath(), + "uploadId=" + uploadID, + canonicalHeaders, + signedHeaders, + bodyHash, + }, "\n") + canonicalHash := sha256.Sum256([]byte(canonicalRequest)) + credentialScope := shortDate + "/" + TestRegion + "/s3/aws4_request" + stringToSign := "AWS4-HMAC-SHA256\n" + dateTime + "\n" + credentialScope + "\n" + hex.EncodeToString(canonicalHash[:]) + credentials := auth.NewCredentialStore() + credentials.AddCredential(accessKey, secretKey) + signingKey, err := credentials.GetSigningKey(accessKey, shortDate, TestRegion) + if err != nil { + t.Fatalf("derive signing key: %v", err) + } + signature := hex.EncodeToString(hmacSHA256(signingKey, []byte(stringToSign))) + request.Header.Set("Authorization", fmt.Sprintf("AWS4-HMAC-SHA256 Credential=%s/%s, SignedHeaders=%s, Signature=%s", accessKey, credentialScope, signedHeaders, signature)) + return request +} + +func testCompletionReplayAfterSignatureWindow(t *testing.T, env *TestEnvironment, bucket string, upstream *completionReplayUpstreamFixture, validator *auth.RequestValidator) { + t.Helper() + signedAt := time.Now().UTC().Add(-14*time.Minute - 50*time.Second).Truncate(time.Second) + request := signCompletionReplayRequestAt(t, env.TAGServer.URL, bucket, completionReplayKey, completionReplaySlowUploadID, completionReplayAccessA, completionReplaySecretA, signedAt) + validationRequest := request.Clone(context.Background()) + validationRequest.Header = request.Header.Clone() + validationRequest.Body = nil + if accessKey, err := validator.ValidateRequest(validationRequest); err != nil || accessKey != completionReplayAccessA { + t.Fatalf("near-expiry signature invalid before dispatch: accessKey=%q err=%v", accessKey, err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + request = request.WithContext(ctx) + result := make(chan completionReplayAttempt, 1) + callsBefore := upstream.completionCalls.Load() + go func() { + response, err := doCompletionReplayRequest(env.TAGServer.Client(), request) + result <- completionReplayAttempt{response: response, err: err} + }() + + select { + case <-upstream.slowEntered: + case <-time.After(5 * time.Second): + t.Fatal("near-expiry completion did not reach upstream") + } + if accessKey, err := validator.ValidateRequest(validationRequest); err != nil || accessKey != completionReplayAccessA { + upstream.releaseSlow() + t.Fatalf("near-expiry signature was no longer valid when upstream received it: accessKey=%q err=%v", accessKey, err) + } + + // Do not release the controlled upstream response until the request-start + // signature crosses the validator's 15-minute age limit. The 10-second + // dispatch margin avoids the near-cutoff timing race while keeping the test short. + untilExpired := time.Until(signedAt.Add(15*time.Minute + time.Millisecond)) + if untilExpired > 0 { + timer := time.NewTimer(untilExpired) + <-timer.C + } + if _, err := validator.ValidateRequest(validationRequest); !errors.Is(err, auth.ErrExpiredRequest) { + upstream.releaseSlow() + t.Fatalf("original completion signature error after the wait = %v; want %v", err, auth.ErrExpiredRequest) + } + upstream.releaseSlow() + + attempt := <-result + if attempt.err != nil { + t.Fatalf("complete slow multipart upload: %v", attempt.err) + } + firstResponse := attempt.response + if firstResponse.status != http.StatusOK || firstResponse.etag != `"completed-etag"` || firstResponse.body != string(completionReplaySuccessBody) { + t.Fatalf("slow completion = %#v; want upstream success without response signing-key metadata", firstResponse) + } + if got := upstream.completionCalls.Load(); got != callsBefore+1 { + t.Fatalf("slow completion made %d upstream calls; want one", got-callsBefore) + } + if got := upstream.slowCalls.Load(); got != 1 { + t.Fatalf("slow upload reached upstream %d times; want one", got) + } + + retry := sendCompletionReplayRequest(t, env.TAGServer.Client(), env.TAGServer.URL, bucket, completionReplayKey, completionReplaySlowUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + if retry.status != firstResponse.status || retry.contentType != firstResponse.contentType || retry.etag != firstResponse.etag || retry.body != firstResponse.body { + t.Errorf("same-principal retry after the original signature expired = %#v; want the captured response %#v", retry, firstResponse) + } + if got := upstream.completionCalls.Load(); got != callsBefore+1 { + t.Errorf("same-principal retry after signature expiry made %d upstream calls; want saved response", got-callsBefore-1) + } + if got := upstream.slowCalls.Load(); got != 1 { + t.Errorf("near-expiry upload was forwarded %d times; want one", got) + } +} diff --git a/tests/integration/testutil.go b/tests/integration/testutil.go index 5d52a104..db184b32 100644 --- a/tests/integration/testutil.go +++ b/tests/integration/testutil.go @@ -28,6 +28,7 @@ import ( "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/johannesboyne/gofakes3" "github.com/johannesboyne/gofakes3/backend/s3mem" + cacheclient "github.com/tigrisdata/ocache/client" "github.com/tigrisdata/ocache/embedded" "github.com/tigrisdata/tag/auth" "github.com/tigrisdata/tag/cache" @@ -854,12 +855,13 @@ func setupSharedCache() error { return fmt.Errorf("failed to get free port for gRPC: %w", err) } - // Initialize embedded cache + // Initialize the single-node test cache with a loopback memberlist address; + // tests do not need remote peers or host-private IP discovery. embeddedCache, err := embedded.New(&embedded.Config{ DiskPath: tempDir, TTL: config.DefaultCacheTTL, NodeID: "test-node", - ClusterAddr: fmt.Sprintf(":%d", clusterPort), + ClusterAddr: fmt.Sprintf("127.0.0.1:%d", clusterPort), GRPCAddr: fmt.Sprintf(":%d", grpcPort), }) if err != nil { @@ -998,6 +1000,11 @@ func newSigningKeysUpstreamHandler(t *testing.T, backend *s3mem.Backend) http.Ha // and local auth enabled. Uses shared embedded cache for cache-hit testing. func NewTestEnvironmentWithTransparentAuth(t *testing.T, upstreamHandler http.HandlerFunc) *TestEnvironment { t.Helper() + return newTestEnvironmentWithTransparentAuthCacheClient(t, upstreamHandler, sharedEmbeddedCache) +} + +func newTestEnvironmentWithTransparentAuthCacheClient(t *testing.T, upstreamHandler http.HandlerFunc, cacheClient cacheclient.CacheClient) *TestEnvironment { + t.Helper() if sharedEmbeddedCache == nil { panic("Shared embedded cache not initialized - TestMain must run first") @@ -1052,7 +1059,7 @@ func NewTestEnvironmentWithTransparentAuth(t *testing.T, upstreamHandler http.Ha }, } - testCache := cache.NewCacheWithClient(sharedEmbeddedCache, &cfg.Cache) + testCache := cache.NewCacheWithClient(cacheClient, &cfg.Cache) forwarder := proxy.NewForwarder(credStore, cfg.Upstream.Endpoint, cfg.Upstream.Region, cfg.Upstream.MaxIdleConnsPerHost, proxySigner, localAuth) service := proxy.NewService(forwarder, testCache, cfg)