From 417c878e9b6964f132caa96207241d0199a74384 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Thu, 1 Oct 2026 10:35:03 +0000 Subject: [PATCH 1/6] test: exercise live completion authz revocation Signed-off-by: Perfloop Agent --- .../completion_replay_auth_test.go | 579 ++++++++++++++++++ 1 file changed, 579 insertions(+) create mode 100644 tests/integration/completion_replay_auth_test.go diff --git a/tests/integration/completion_replay_auth_test.go b/tests/integration/completion_replay_auth_test.go new file mode 100644 index 00000000..356f7765 --- /dev/null +++ b/tests/integration/completion_replay_auth_test.go @@ -0,0 +1,579 @@ +package integration + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync/atomic" + "testing" + "time" + + "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 +} + +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) { + testColdCompletionReplayAfterSignatureWindow(t) + }) +} + +func testCompleteMultipartUploadReplayAuthorization(t *testing.T, transparent bool) { + t.Helper() + bucket := completionReplayBucket(t) + otherBucket := bucket + "-other" + upstream, completionCalls, slowCalls := newCompletionReplayUpstream(t, bucket, transparent) + env := newCompletionReplayEnvironment(t, transparent, upstream) + defer env.Close() + + if transparent { + 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) + + 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: 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, completionCalls, slowCalls, validator) + +} + +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) (http.HandlerFunc, *atomic.Int32, *atomic.Int32) { + t.Helper() + var responseKeys string + if transparent { + responseKeys = completionReplaySigningKeysHeader(t, completionReplayAccessA, completionReplaySecretA) + } + var completionCalls atomic.Int32 + var mainCalls atomic.Int32 + var aliasCalls atomic.Int32 + var slowCalls atomic.Int32 + writeSuccess := func(w http.ResponseWriter) { + if 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`) + } + } + + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Query().Get("uploadId") == "" { + http.NotFound(w, r) + return + } + 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 mainCalls.Add(1) == 1 { + writeSuccess(w) + return + } + writeError(w, http.StatusForbidden) + return + } + if requestBucket == bucket && key == completionReplayKey && uploadID == completionReplayAliasID && authErr == nil && info.AccessKey == completionReplayAccessA { + if aliasCalls.Add(1) == 1 { + writeSuccess(w) + 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 slowCalls.Add(1) == 1 { + time.Sleep(8 * time.Second) + writeSuccess(w) + return + } + writeError(w, http.StatusNotFound) + return + } + if requestBucket == bucket && key == completionReplayKey && uploadID == completionReplayMissUploadID { + writeError(w, http.StatusNotFound) + return + } + writeError(w, http.StatusForbidden) + }) + return handler, &completionCalls, &slowCalls +} + +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("COMPLETION_REPLAY_PROPERTY_VIOLATION: %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("COMPLETION_REPLAY_PROPERTY_VIOLATION: %s made %d upstream calls; want 1 authorization request", name, got-before) + } + if response.status != wantStatus { + t.Errorf("COMPLETION_REPLAY_PROPERTY_VIOLATION: 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: %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("COMPLETION_REPLAY_PROPERTY_VIOLATION: %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() + resp, err := client.Do(req) + if err != nil { + t.Fatalf("send completion request: %v", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read completion response: %v", err) + } + return completionReplayResponse{status: resp.StatusCode, contentType: resp.Header.Get("Content-Type"), etag: resp.Header.Get("ETag"), body: string(body)} +} + +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, completionCalls, slowCalls *atomic.Int32, validator *auth.RequestValidator) { + t.Helper() + signedAt := time.Now().UTC().Add(-14*time.Minute - 55*time.Second).Truncate(time.Second) + request := signCompletionReplayRequestAt(t, env.TAGServer.URL, bucket, completionReplayKey, completionReplaySlowUploadID, completionReplayAccessA, completionReplaySecretA, signedAt) + if accessKey, err := validator.ValidateRequest(request); err != nil || accessKey != completionReplayAccessA { + t.Fatalf("near-expiry signature invalid at dispatch: accessKey=%q err=%v", accessKey, err) + } + + callsBefore := completionCalls.Load() + firstResponse := executeCompletionReplayRequest(t, env.TAGServer.Client(), request) + if firstResponse.status != http.StatusOK || firstResponse.etag != `"completed-etag"` || firstResponse.body != string(completionReplaySuccessBody) { + t.Fatalf("near-expiry completion = %#v; want upstream success", firstResponse) + } + if got := completionCalls.Load(); got != callsBefore+1 { + t.Fatalf("near-expiry completion made %d upstream calls; want one", got-callsBefore) + } + if got := slowCalls.Load(); got != 1 { + t.Fatalf("near-expiry 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: same-principal retry after the original signature expired = %#v; want the captured response %#v", retry, firstResponse) + } + if got := completionCalls.Load(); got != callsBefore+1 { + t.Errorf("COMPLETION_REPLAY_PROPERTY_VIOLATION: same-principal retry after signature expiry made %d upstream calls; want saved response", got-callsBefore-1) + } + if got := slowCalls.Load(); got != 1 { + t.Errorf("COMPLETION_REPLAY_PROPERTY_VIOLATION: near-expiry upload was forwarded %d times; want one", got) + } +} + +func testColdCompletionReplayAfterSignatureWindow(t *testing.T) { + t.Helper() + bucket := completionReplayBucket(t) + upstream, completionCalls, slowCalls := newCompletionReplayUpstream(t, bucket, true) + env := NewTestEnvironmentWithTransparentAuth(t, upstream) + defer env.Close() + if env.DerivedKeyStore.HasKey(completionReplayAccessA) || env.AuthzCache.IsAuthorized(completionReplayAccessA, bucket) { + t.Fatal("test did not begin with cold transparent authentication state") + } + + signedAt := time.Now().UTC().Add(-14*time.Minute - 55*time.Second).Truncate(time.Second) + firstRequest := signCompletionReplayRequestAt(t, env.TAGServer.URL, bucket, completionReplayKey, completionReplaySlowUploadID, completionReplayAccessA, completionReplaySecretA, signedAt) + credentials := auth.NewCredentialStore() + credentials.AddCredential(completionReplayAccessA, completionReplaySecretA) + if accessKey, err := auth.NewRequestValidator(credentials).ValidateRequest(firstRequest); err != nil || accessKey != completionReplayAccessA { + t.Fatalf("near-expiry request is not valid at dispatch: accessKey=%q err=%v", accessKey, err) + } + + firstResponse := executeCompletionReplayRequest(t, env.TAGServer.Client(), firstRequest) + if firstResponse.status != http.StatusOK || firstResponse.body != string(completionReplaySuccessBody) { + t.Fatalf("cold-auth completion = %#v; want upstream success", firstResponse) + } + if got := completionCalls.Load(); got != 1 || slowCalls.Load() != 1 { + t.Fatalf("cold-auth completion made %d completion calls and %d delayed calls; want one each", got, slowCalls.Load()) + } + if !env.DerivedKeyStore.HasKey(completionReplayAccessA) || !env.AuthzCache.IsAuthorized(completionReplayAccessA, bucket) { + t.Fatal("successful upstream response did not teach local signing and bucket authorization state") + } + if _, err := auth.NewRequestValidator(env.DerivedKeyStore).ValidateRequest(firstRequest); err == nil { + t.Fatal("original completion signature did not expire during upstream work") + } + + before := completionCalls.Load() + 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: cold-auth same-principal retry = %#v; want captured response %#v", retry, firstResponse) + } + if got := completionCalls.Load(); got != before { + t.Errorf("COMPLETION_REPLAY_PROPERTY_VIOLATION: cold-auth retry made %d upstream calls; want saved response", got-before) + } +} From 1ee0e9eb5650542533916fe8163216e8255bf0a2 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Fri, 2 Oct 2026 05:18:48 +0000 Subject: [PATCH 2/6] test: bind integration cache to loopback Signed-off-by: Perfloop Agent --- tests/integration/testutil.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/integration/testutil.go b/tests/integration/testutil.go index 9d7fab8d..29797f52 100644 --- a/tests/integration/testutil.go +++ b/tests/integration/testutil.go @@ -851,12 +851,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 { From d072699361bffe781ddfce235f77af9e52bdb834 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Fri, 2 Oct 2026 06:30:48 +0000 Subject: [PATCH 3/6] test: keep replay assertions independent of result labels Signed-off-by: Perfloop Agent --- .../completion_replay_auth_test.go | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/tests/integration/completion_replay_auth_test.go b/tests/integration/completion_replay_auth_test.go index 356f7765..0a442deb 100644 --- a/tests/integration/completion_replay_auth_test.go +++ b/tests/integration/completion_replay_auth_test.go @@ -96,7 +96,7 @@ func testCompleteMultipartUploadReplayAuthorization(t *testing.T, transparent bo before = completionCalls.Load() samePrincipal := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, false) if samePrincipal != first { - t.Errorf("COMPLETION_REPLAY_PROPERTY_VIOLATION: validated same-principal replay = %#v; want saved response %#v", samePrincipal, first) + t.Errorf("validated same-principal replay = %#v; want saved response %#v", samePrincipal, first) } assertCompletionReplayCallCount(t, "same-principal replay", completionCalls.Load(), before) @@ -127,7 +127,7 @@ func testCompleteMultipartUploadReplayAuthorization(t *testing.T, transparent bo 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: valid same-principal presigned retry = %#v; want saved response %#v", presigned, first) + t.Errorf("valid same-principal presigned retry = %#v; want saved response %#v", presigned, first) } assertCompletionReplayCallCount(t, "presigned replay", completionCalls.Load(), before) @@ -149,7 +149,7 @@ func testCompleteMultipartUploadReplayAuthorization(t *testing.T, transparent bo 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: cache miss status = %d; want upstream 404", miss.status) + t.Errorf("cache miss status = %d; want upstream 404", miss.status) } assertCompletionReplayForwarded(t, "cache miss", miss, completionCalls.Load(), before, http.StatusNotFound) @@ -326,31 +326,31 @@ func seedTransparentPrincipal(t *testing.T, env *TestEnvironment, accessKey, sec 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: %s received status=%d ETag=%q body=%q from a cached completion", name, got.status, got.etag, got.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("COMPLETION_REPLAY_PROPERTY_VIOLATION: %s made %d upstream calls; want 1 authorization request", name, got-before) + t.Errorf("%s made %d upstream calls; want 1 authorization request", name, got-before) } if response.status != wantStatus { - t.Errorf("COMPLETION_REPLAY_PROPERTY_VIOLATION: forwarded %s response status = %d; want upstream status %d", name, 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: %s made %d upstream calls; want local rejection", name, 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: %s made %d upstream calls; want cached response", name, got-before) + t.Errorf("%s made %d upstream calls; want cached response", name, got-before) } } @@ -526,13 +526,13 @@ func testCompletionReplayAfterSignatureWindow(t *testing.T, env *TestEnvironment 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: same-principal retry after the original signature expired = %#v; want the captured response %#v", retry, firstResponse) + t.Errorf("same-principal retry after the original signature expired = %#v; want the captured response %#v", retry, firstResponse) } if got := completionCalls.Load(); got != callsBefore+1 { - t.Errorf("COMPLETION_REPLAY_PROPERTY_VIOLATION: same-principal retry after signature expiry made %d upstream calls; want saved response", got-callsBefore-1) + t.Errorf("same-principal retry after signature expiry made %d upstream calls; want saved response", got-callsBefore-1) } if got := slowCalls.Load(); got != 1 { - t.Errorf("COMPLETION_REPLAY_PROPERTY_VIOLATION: near-expiry upload was forwarded %d times; want one", got) + t.Errorf("near-expiry upload was forwarded %d times; want one", got) } } @@ -571,9 +571,9 @@ func testColdCompletionReplayAfterSignatureWindow(t *testing.T) { before := completionCalls.Load() 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("COMPLETION_REPLAY_PROPERTY_VIOLATION: cold-auth same-principal retry = %#v; want captured response %#v", retry, firstResponse) + t.Errorf("cold-auth same-principal retry = %#v; want captured response %#v", retry, firstResponse) } if got := completionCalls.Load(); got != before { - t.Errorf("COMPLETION_REPLAY_PROPERTY_VIOLATION: cold-auth retry made %d upstream calls; want saved response", got-before) + t.Errorf("cold-auth retry made %d upstream calls; want saved response", got-before) } } From 36d0c890b3efa8c19beeb1d56154a7a4f42d813a Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Fri, 2 Oct 2026 09:08:21 +0000 Subject: [PATCH 4/6] test: align replay comparison with repaired route schedule Signed-off-by: Perfloop Agent --- .../completion_replay_auth_test.go | 341 +++++++++++++----- tests/integration/testutil.go | 8 +- 2 files changed, 267 insertions(+), 82 deletions(-) diff --git a/tests/integration/completion_replay_auth_test.go b/tests/integration/completion_replay_auth_test.go index 0a442deb..b1e841ac 100644 --- a/tests/integration/completion_replay_auth_test.go +++ b/tests/integration/completion_replay_auth_test.go @@ -6,15 +6,18 @@ import ( "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" @@ -47,6 +50,52 @@ type completionReplayResponse struct { 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) @@ -55,7 +104,10 @@ func TestCompleteMultipartUploadReplayAuthorization(t *testing.T) { testCompleteMultipartUploadReplayAuthorization(t, false) }) t.Run("transparent-cold-auth", func(t *testing.T) { - testColdCompletionReplayAfterSignatureWindow(t) + testColdCompletionReplay(t) + }) + t.Run("transparent-revocation-during-cache-read", func(t *testing.T) { + testCompletionReplayGrantRevocationDuringRead(t) }) } @@ -63,8 +115,10 @@ func testCompleteMultipartUploadReplayAuthorization(t *testing.T, transparent bo t.Helper() bucket := completionReplayBucket(t) otherBucket := bucket + "-other" - upstream, completionCalls, slowCalls := newCompletionReplayUpstream(t, bucket, transparent) - env := newCompletionReplayEnvironment(t, transparent, upstream) + upstream := newCompletionReplayUpstream(t, bucket, transparent) + defer upstream.releaseSlow() + completionCalls := &upstream.completionCalls + env := newCompletionReplayEnvironment(t, transparent, upstream.handler) defer env.Close() if transparent { @@ -178,9 +232,122 @@ func testCompleteMultipartUploadReplayAuthorization(t *testing.T, transparent bo 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) || !env.AuthzCache.IsAuthorized(completionReplayAccessA, bucket) { + t.Fatal("successful upstream response did not teach local signing and bucket authorization state") + } + + before := upstream.completionCalls.Load() + retry := sendCompletionReplayRequest(t, client, env.TAGServer.URL, bucket, completionReplayKey, completionReplayUploadID, completionReplayAccessA, completionReplaySecretA, false, false) + if retry.status != first.status || retry.contentType != first.contentType || retry.etag != first.etag || retry.body != first.body { + t.Errorf("cold-auth same-principal retry = %#v; want captured response %#v", retry, first) + } + assertCompletionReplayCallCount(t, "cold-auth same-principal replay", upstream.completionCalls.Load(), before) +} + +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) } - testCompletionReplayAfterSignatureWindow(t, env, bucket, completionCalls, slowCalls, validator) + 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) { @@ -215,18 +382,18 @@ func newCompletionReplayEnvironment(t *testing.T, transparent bool, upstream htt return NewTestEnvironmentWithCacheHandler(upstream) } -func newCompletionReplayUpstream(t *testing.T, bucket string, transparent bool) (http.HandlerFunc, *atomic.Int32, *atomic.Int32) { +func newCompletionReplayUpstream(t *testing.T, bucket string, transparent bool) *completionReplayUpstreamFixture { t.Helper() var responseKeys string if transparent { responseKeys = completionReplaySigningKeysHeader(t, completionReplayAccessA, completionReplaySecretA) } - var completionCalls atomic.Int32 - var mainCalls atomic.Int32 - var aliasCalls atomic.Int32 - var slowCalls atomic.Int32 - writeSuccess := func(w http.ResponseWriter) { - if responseKeys != "" { + 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"`) @@ -243,26 +410,31 @@ func newCompletionReplayUpstream(t *testing.T, bucket string, transparent bool) } } - handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + 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 } - completionCalls.Add(1) + 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 mainCalls.Add(1) == 1 { - writeSuccess(w) + 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 aliasCalls.Add(1) == 1 { - writeSuccess(w) + if upstream.aliasCalls.Add(1) == 1 { + writeSuccess(w, true) return } writeError(w, http.StatusForbidden) @@ -273,9 +445,10 @@ func newCompletionReplayUpstream(t *testing.T, bucket string, transparent bool) return } if requestBucket == bucket && key == completionReplayKey && uploadID == completionReplaySlowUploadID && authErr == nil && info.AccessKey == completionReplayAccessA { - if slowCalls.Add(1) == 1 { - time.Sleep(8 * time.Second) - writeSuccess(w) + if upstream.slowCalls.Add(1) == 1 { + close(upstream.slowEntered) + <-upstream.slowRelease + writeSuccess(w, false) return } writeError(w, http.StatusNotFound) @@ -287,7 +460,7 @@ func newCompletionReplayUpstream(t *testing.T, bucket string, transparent bool) } writeError(w, http.StatusForbidden) }) - return handler, &completionCalls, &slowCalls + return upstream } func completionReplaySigningKeysHeader(t *testing.T, accessKey, secretKey string) string { @@ -403,16 +576,24 @@ func sendPresignedCompletionReplayRequest(t *testing.T, client *http.Client, end func executeCompletionReplayRequest(t *testing.T, client *http.Client, req *http.Request) completionReplayResponse { t.Helper() - resp, err := client.Do(req) + 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 { - t.Fatalf("read completion response: %v", err) + return completionReplayResponse{}, err } - return completionReplayResponse{status: resp.StatusCode, contentType: resp.Header.Get("Content-Type"), etag: resp.Header.Get("ETag"), body: string(body)} + 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 { @@ -504,76 +685,74 @@ func signCompletionReplayRequestAt(t *testing.T, endpoint, bucket, key, uploadID return request } -func testCompletionReplayAfterSignatureWindow(t *testing.T, env *TestEnvironment, bucket string, completionCalls, slowCalls *atomic.Int32, validator *auth.RequestValidator) { +func testCompletionReplayAfterSignatureWindow(t *testing.T, env *TestEnvironment, bucket string, upstream *completionReplayUpstreamFixture, validator *auth.RequestValidator) { t.Helper() - signedAt := time.Now().UTC().Add(-14*time.Minute - 55*time.Second).Truncate(time.Second) + 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) - if accessKey, err := validator.ValidateRequest(request); err != nil || accessKey != completionReplayAccessA { - t.Fatalf("near-expiry signature invalid at dispatch: accessKey=%q err=%v", accessKey, err) - } - - callsBefore := completionCalls.Load() - firstResponse := executeCompletionReplayRequest(t, env.TAGServer.Client(), request) + 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("near-expiry completion = %#v; want upstream success", firstResponse) + t.Fatalf("slow completion = %#v; want upstream success without response signing-key metadata", firstResponse) } - if got := completionCalls.Load(); got != callsBefore+1 { - t.Fatalf("near-expiry completion made %d upstream calls; want one", got-callsBefore) + if got := upstream.completionCalls.Load(); got != callsBefore+1 { + t.Fatalf("slow completion made %d upstream calls; want one", got-callsBefore) } - if got := slowCalls.Load(); got != 1 { - t.Fatalf("near-expiry upload reached upstream %d times; want one", got) + 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 := completionCalls.Load(); got != callsBefore+1 { + 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 := slowCalls.Load(); got != 1 { + if got := upstream.slowCalls.Load(); got != 1 { t.Errorf("near-expiry upload was forwarded %d times; want one", got) } } - -func testColdCompletionReplayAfterSignatureWindow(t *testing.T) { - t.Helper() - bucket := completionReplayBucket(t) - upstream, completionCalls, slowCalls := newCompletionReplayUpstream(t, bucket, true) - env := NewTestEnvironmentWithTransparentAuth(t, upstream) - defer env.Close() - if env.DerivedKeyStore.HasKey(completionReplayAccessA) || env.AuthzCache.IsAuthorized(completionReplayAccessA, bucket) { - t.Fatal("test did not begin with cold transparent authentication state") - } - - signedAt := time.Now().UTC().Add(-14*time.Minute - 55*time.Second).Truncate(time.Second) - firstRequest := signCompletionReplayRequestAt(t, env.TAGServer.URL, bucket, completionReplayKey, completionReplaySlowUploadID, completionReplayAccessA, completionReplaySecretA, signedAt) - credentials := auth.NewCredentialStore() - credentials.AddCredential(completionReplayAccessA, completionReplaySecretA) - if accessKey, err := auth.NewRequestValidator(credentials).ValidateRequest(firstRequest); err != nil || accessKey != completionReplayAccessA { - t.Fatalf("near-expiry request is not valid at dispatch: accessKey=%q err=%v", accessKey, err) - } - - firstResponse := executeCompletionReplayRequest(t, env.TAGServer.Client(), firstRequest) - if firstResponse.status != http.StatusOK || firstResponse.body != string(completionReplaySuccessBody) { - t.Fatalf("cold-auth completion = %#v; want upstream success", firstResponse) - } - if got := completionCalls.Load(); got != 1 || slowCalls.Load() != 1 { - t.Fatalf("cold-auth completion made %d completion calls and %d delayed calls; want one each", got, slowCalls.Load()) - } - if !env.DerivedKeyStore.HasKey(completionReplayAccessA) || !env.AuthzCache.IsAuthorized(completionReplayAccessA, bucket) { - t.Fatal("successful upstream response did not teach local signing and bucket authorization state") - } - if _, err := auth.NewRequestValidator(env.DerivedKeyStore).ValidateRequest(firstRequest); err == nil { - t.Fatal("original completion signature did not expire during upstream work") - } - - before := completionCalls.Load() - 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("cold-auth same-principal retry = %#v; want captured response %#v", retry, firstResponse) - } - if got := completionCalls.Load(); got != before { - t.Errorf("cold-auth retry made %d upstream calls; want saved response", got-before) - } -} diff --git a/tests/integration/testutil.go b/tests/integration/testutil.go index 29797f52..c0055ed7 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" @@ -996,6 +997,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") @@ -1050,7 +1056,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) From dc07b3c1ecb8cbaa66f90a4a5b4591e33bf95954 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Fri, 2 Oct 2026 09:08:21 +0000 Subject: [PATCH 5/6] fix: recheck multipart completion grants after cache reads Signed-off-by: Perfloop Agent --- README.md | 1 + cache/completion.go | 46 ++++++++++++++++---- cache/completion_test.go | 70 ++++++++++++++++++++++++++---- docs/s3-compatibility.md | 3 +- proxy/forwarder.go | 57 +++++++++++++++++++++--- proxy/forwarder_signing.go | 23 +++++++++- proxy/forwarder_transparent.go | 61 +++++++++++++++++++++++--- proxy/multipart_complete_test.go | 4 +- proxy/revalidation_test.go | 5 +++ proxy/service.go | 74 +++++++++++++++++++++++++++----- 10 files changed, 300 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index c8d6fdb7..2cdba0e2 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,7 @@ When configuring S3 clients, ensure path-style addressing is enabled. See [docs/ - Among cacheable objects (within `size_threshold`), those at or above `block_size` are cached as fixed-size blocks (block caching, on by default), so a range read fetches only the covering blocks; smaller objects are whole-cached - Set `block_caching_enabled: false` to cache whole objects instead — a range request then triggers a background fetch of the full object (if within threshold) - PUT/DELETE operations invalidate the cache entry +- 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 See [docs/cache-control.md](docs/cache-control.md) for detailed cache control and revalidation documentation. For tuning and sizing block caching in production, see [docs/deploy.md](docs/deploy.md#block-aligned-caching). diff --git a/cache/completion.go b/cache/completion.go index 6aa9021f..32cfa194 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" @@ -24,13 +26,31 @@ type CompletionEntry struct { Body []byte `json:"body"` } +// completionCacheRecord keeps caller identity in the cache's internal record; +// it is never returned to callers. +type completionCacheRecord struct { + AccessKey string `json:"access_key,omitempty"` + StatusCode int `json:"status_code"` + Headers map[string]string `json:"headers"` + Body []byte `json:"body"` +} + // 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) { +// GetCompletion retrieves a cached completion response only for a caller that +// matches the identity stored with the successful completion. The callback is +// evaluated only when a bound cache entry exists. +func (c *Cache) GetCompletion(ctx context.Context, bucket, key, uploadId string, callerAccessKey func() (string, bool)) (*CompletionEntry, bool, error) { if !c.IsEnabled() { return nil, false, nil } @@ -47,19 +67,28 @@ func (c *Cache) GetCompletion(ctx context.Context, bucket, key, uploadId string) return nil, false, nil } - var entry CompletionEntry + var entry completionCacheRecord if err := json.Unmarshal(data, &entry); err != nil { 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 == "" || callerAccessKey == nil { + return nil, false, nil + } + accessKey, authenticated := callerAccessKey() + if !authenticated || accessKey == "" || entry.AccessKey != accessKey { + return nil, false, nil + } log.Debug().Str("bucket", bucket).Str("uploadId", uploadId).Msg("Completion cache hit") - return &entry, true, nil + return &CompletionEntry{StatusCode: entry.StatusCode, Headers: entry.Headers, Body: entry.Body}, 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 replayable. +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 } @@ -71,7 +100,8 @@ func (c *Cache) PutCompletion(ctx context.Context, bucket, key, uploadId string, } } - entry := CompletionEntry{ + entry := completionCacheRecord{ + AccessKey: accessKey, StatusCode: statusCode, Headers: headerMap, Body: body, diff --git a/cache/completion_test.go b/cache/completion_test.go index c3dada0b..4da9c54f 100644 --- a/cache/completion_test.go +++ b/cache/completion_test.go @@ -18,9 +18,13 @@ func TestCompletionCache(t *testing.T) { key := "test-key" uploadId := "upload-123" - // Test 1: GetCompletion returns not found for non-existent entry + // Test 1: GetCompletion returns not found for non-existent entry without asking for caller identity. t.Run("GetCompletion_NotFound", func(t *testing.T) { - entry, found, err := cache.GetCompletion(ctx, bucket, key, uploadId) + identityChecked := false + entry, found, err := cache.GetCompletion(ctx, bucket, key, uploadId, func() (string, bool) { + identityChecked = true + return "client-a", true + }) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -30,6 +34,9 @@ func TestCompletionCache(t *testing.T) { if entry != nil { t.Error("expected nil entry, got non-nil") } + if identityChecked { + t.Error("caller identity was checked for a cache miss") + } }) // Test 2: PutCompletion stores and GetCompletion retrieves @@ -42,13 +49,15 @@ 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, func() (string, bool) { + return "client-a", true + }) if err != nil { t.Fatalf("GetCompletion failed: %v", err) } @@ -70,11 +79,23 @@ func TestCompletionCache(t *testing.T) { if string(entry.Body) != string(body) { t.Errorf("expected body %q, got %q", body, entry.Body) } + + wrongEntry, wrongFound, err := cache.GetCompletion(ctx, bucket, key, uploadId, func() (string, bool) { + return "client-b", true + }) + 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) + } }) // 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", func() (string, bool) { + return "client-a", true + }) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -86,7 +107,23 @@ 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", func() (string, bool) { + return "client-a", true + }) + 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 +131,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 +155,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", nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -115,7 +167,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 4ec6c94d..6cc069db 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=` | | @@ -83,6 +83,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** — PUT, DELETE, CopyObject, and DeleteObjects invalidate the cache _before_ forwarding to Tigris to prevent stale reads. - **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/forwarder.go b/proxy/forwarder.go index 076f79fa..b4917f7d 100644 --- a/proxy/forwarder.go +++ b/proxy/forwarder.go @@ -80,6 +80,44 @@ type RequestForwarder interface { DoConditionalHeadRequest(ctx context.Context, bucket, key, accessKey, secretKey, etag string, lastModified int64) (*http.Response, error) } +// completionAccessKeyValidator identifies a caller only after the forwarder's +// local authentication and, where available, bucket-authorization checks pass. +type completionAccessKeyValidator interface { + authenticatedCompletionAccessKey(r *http.Request) (accessKey string, ok bool) +} + +// completionAccessKeyAtRequestStart snapshots the caller identity before forwarding +// and can recheck its bucket authorization after a cache read. +type completionAccessKeyAtRequestStart interface { + authenticatedCompletionAccessKeyAtRequestStart(r *http.Request) (accessKey string, ok bool) + completionAccessKeyStillAuthorized(accessKey string, r *http.Request) bool +} + +// completionRequestForLocalValidation supplies the unsigned payload marker that +// presigned SigV4 validation expects without changing the request sent upstream. +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 +} + // AuthErrorCode represents the type of authentication error. type AuthErrorCode int @@ -162,9 +200,9 @@ var ( // ResponseInterceptor is called after receiving the upstream response but before // headers are sent to the client. Used by transparentForwarder to extract signing -// keys and strip internal headers from the response. -// The originalReq parameter is the original client request (needed for parsing auth info). -type ResponseInterceptor func(resp *http.Response, originalReq *http.Request) +// keys, strip internal headers, and report a client identity authenticated by the +// response. The originalReq is needed to parse the client's auth info. +type ResponseInterceptor func(resp *http.Response, originalReq *http.Request) (accessKey string, authenticated bool) // baseForwarder contains shared HTTP execution logic used by both // signingForwarder and transparentForwarder. @@ -228,7 +266,7 @@ func (b *baseForwarder) executeAndStreamWithMeta(w http.ResponseWriter, fwdReq * // Run response interceptor before sending headers to client if b.responseInterceptor != nil { - b.responseInterceptor(resp, originalReq) + _, _ = b.responseInterceptor(resp, originalReq) } var respHeaders http.Header @@ -269,8 +307,10 @@ func (b *baseForwarder) executeAndCapture(w http.ResponseWriter, fwdReq *http.Re defer resp.Body.Close() // Run response interceptor before sending headers to client + var authenticatedAccessKey string + var authenticated bool if b.responseInterceptor != nil { - b.responseInterceptor(resp, originalReq) + authenticatedAccessKey, authenticated = b.responseInterceptor(resp, originalReq) } // Capture response @@ -278,6 +318,9 @@ func (b *baseForwarder) executeAndCapture(w http.ResponseWriter, fwdReq *http.Re StatusCode: resp.StatusCode, Headers: resp.Header.Clone(), } + if authenticated { + capture.authenticatedAccessKey = authenticatedAccessKey + } // Copy headers to response writer copyHeaders(w.Header(), resp.Header) @@ -319,7 +362,7 @@ func (b *baseForwarder) executeRequest(fwdReq *http.Request, inContentLength int // Run response interceptor (e.g., signing key learning, header stripping) if b.responseInterceptor != nil { - b.responseInterceptor(resp, originalReq) + _, _ = b.responseInterceptor(resp, originalReq) } return resp, nil @@ -585,6 +628,8 @@ type ResponseCapture struct { Headers http.Header Body []byte Complete bool // True if body was fully captured without errors + + authenticatedAccessKey string // Set after local validation or trusted upstream key learning. } // ContentLength returns the content length from headers or body length. diff --git a/proxy/forwarder_signing.go b/proxy/forwarder_signing.go index 9f802d90..94194c99 100644 --- a/proxy/forwarder_signing.go +++ b/proxy/forwarder_signing.go @@ -133,7 +133,28 @@ func (f *signingForwarder) ForwardWithCapture(ctx context.Context, w http.Respon } prepareForwardedRequest(fwdReq, contentLength, chunked) - return f.executeAndCapture(w, fwdReq, contentLength, nil) + capture, err := f.executeAndCapture(w, fwdReq, contentLength, nil) + if capture != nil && err == nil { + capture.authenticatedAccessKey = accessKey + } + return capture, err +} + +// authenticatedCompletionAccessKey returns the client identity only when its +// signature and credentials validate locally. +func (f *signingForwarder) authenticatedCompletionAccessKey(r *http.Request) (string, bool) { + validationRequest, ok := completionRequestForLocalValidation(r) + if !ok { + return "", false + } + accessKey, err := f.validator.ValidateRequest(validationRequest) + if err != nil { + return "", false + } + if _, err := f.credStore.GetSecretKey(accessKey); err != nil { + return "", false + } + return accessKey, accessKey != "" } // ValidateAndGetCredentials validates the request signature and returns credentials. diff --git a/proxy/forwarder_transparent.go b/proxy/forwarder_transparent.go index 9a9ab6f4..bb046c0b 100644 --- a/proxy/forwarder_transparent.go +++ b/proxy/forwarder_transparent.go @@ -43,9 +43,10 @@ func (f *transparentForwarder) initInterceptor() { // interceptResponse is called by base forwarder methods after receiving the // upstream response but before headers are sent to the client. It extracts // signing keys from successful responses and revokes authZ on 403s. -func (f *transparentForwarder) interceptResponse(resp *http.Response, originalReq *http.Request) { - f.learnSigningKeys(resp, originalReq) +func (f *transparentForwarder) interceptResponse(resp *http.Response, originalReq *http.Request) (string, bool) { + accessKey, authenticated := f.learnSigningKeys(resp, originalReq) f.handleAuthzRevocation(resp, originalReq) + return accessKey, authenticated } // shallowHeaderCopy gives the forwarded request its own header map while @@ -202,6 +203,48 @@ func (f *transparentForwarder) ValidateAndGetCredentials(r *http.Request) (AuthR return result, f.proxySigner.AccessKey(), f.proxySigner.SecretKey(), nil } +// authenticatedCompletionAccessKey returns the caller identity after local +// signature and bucket-authorization checks pass. Without local auth configured, +// transparent requests cannot be bound to a validated principal for replay. +func (f *transparentForwarder) authenticatedCompletionAccessKey(r *http.Request) (string, bool) { + if f.validator == nil || f.derivedKeyStore == nil || f.authzCache == nil { + return "", false + } + validationRequest, ok := completionRequestForLocalValidation(r) + if !ok { + return "", false + } + result, err := f.validateLocally(validationRequest) + if err != nil || result != AuthValidated { + return "", false + } + authInfo, err := auth.ParseAuthInfo(r) + if err != nil || authInfo.AccessKey == "" { + return "", false + } + return authInfo.AccessKey, true +} + +// authenticatedCompletionAccessKeyAtRequestStart captures transparent-mode +// identity before an upstream round trip can age out the request signature. +func (f *transparentForwarder) authenticatedCompletionAccessKeyAtRequestStart(r *http.Request) (string, bool) { + return f.authenticatedCompletionAccessKey(r) +} + +// completionAccessKeyStillAuthorized rechecks the grant at the replay decision +// point after a potentially remote completion-cache read has finished. +func (f *transparentForwarder) completionAccessKeyStillAuthorized(accessKey string, r *http.Request) bool { + if accessKey == "" || f.authzCache == nil { + return false + } + authInfo, err := auth.ParseAuthInfo(r) + if err != nil || authInfo.AccessKey != accessKey { + return false + } + bucket, _ := ParseBucketKey(r) + return f.authzCache.IsAuthorized(accessKey, bucket) +} + // validateLocally performs local SigV4 validation of the client's request. func (f *transparentForwarder) validateLocally(r *http.Request) (AuthResult, error) { // If local auth is not configured, always treat as validated (legacy behavior) @@ -286,13 +329,14 @@ func (f *transparentForwarder) DoRequestWithCreds(ctx context.Context, r *http.R // learnSigningKeys extracts and caches derived signing keys from the Tigris response. // The signing keys header is always stripped before the response reaches the client. -func (f *transparentForwarder) learnSigningKeys(resp *http.Response, r *http.Request) { +// It returns the caller identity only when a successful response supplied a usable key. +func (f *transparentForwarder) learnSigningKeys(resp *http.Response, r *http.Request) (string, bool) { // Always strip the internal header, even when local auth is disabled. headerVal := resp.Header.Get(signingKeysHeader) resp.Header.Del(signingKeysHeader) if f.keyUnwrapper == nil { - return + return "", false } // Header may be absent on 2xx if feature is disabled on Tigris side, or non-proxy request @@ -300,22 +344,23 @@ func (f *transparentForwarder) learnSigningKeys(resp *http.Response, r *http.Req if headerVal == "" && resp.StatusCode >= 200 && resp.StatusCode < 300 { log.Debug().Int("status", resp.StatusCode).Msg("Signing keys header absent from successful response") } - return + return "", false } authInfo, err := auth.ParseAuthInfo(r) if err != nil { log.Debug().Err(err).Msg("Failed to parse auth info for signing key learning") - return + return "", false } entries, err := f.keyUnwrapper.Unwrap(headerVal, authInfo.AccessKey) if err != nil { log.Warn().Err(err).Str("access_key", authInfo.AccessKey).Msg("Failed to unwrap signing keys") - return + return "", false } newKeys := 0 + usableKey := false for _, entry := range entries { keyBytes, err := hex.DecodeString(entry.SigningKey) if err != nil { @@ -326,6 +371,7 @@ func (f *transparentForwarder) learnSigningKeys(resp *http.Response, r *http.Req newKeys++ } f.derivedKeyStore.Store(authInfo.AccessKey, entry.Date, entry.Region, keyBytes) + usableKey = true } bucket, _ := ParseBucketKey(r) @@ -342,6 +388,7 @@ func (f *transparentForwarder) learnSigningKeys(resp *http.Response, r *http.Req metrics.ProxySigningKeysReceived.Inc() metrics.DerivedKeyStoreSize.Set(float64(f.derivedKeyStore.Count())) metrics.AuthzCacheSize.Set(float64(f.authzCache.Count())) + return authInfo.AccessKey, usableKey && f.authzCache.IsAuthorized(authInfo.AccessKey, bucket) } // handleAuthzRevocation revokes authorization when Tigris returns 403. diff --git a/proxy/multipart_complete_test.go b/proxy/multipart_complete_test.go index b7f84ff7..a46cfa6d 100644 --- a/proxy/multipart_complete_test.go +++ b/proxy/multipart_complete_test.go @@ -89,7 +89,9 @@ 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, func() (string, bool) { + return "access", true + }); found { t.Error("200-with-error-body completion was cached as a successful completion — an idempotent replay would return the error as success") } } diff --git a/proxy/revalidation_test.go b/proxy/revalidation_test.go index be50dfeb..7b0e5526 100644 --- a/proxy/revalidation_test.go +++ b/proxy/revalidation_test.go @@ -59,6 +59,11 @@ func (m *mockForwarder) ValidateAndGetCredentials(r *http.Request) (AuthResult, return AuthValidated, "access", "secret", nil } +func (m *mockForwarder) authenticatedCompletionAccessKey(r *http.Request) (string, bool) { + result, accessKey, _, err := m.ValidateAndGetCredentials(r) + return accessKey, err == nil && result == AuthValidated && accessKey != "" +} + func (m *mockForwarder) DoRequestWithCreds(ctx context.Context, r *http.Request, accessKey, secretKey string) (*http.Response, error) { if m.doRequestFunc != nil { return m.doRequestFunc(ctx, r, accessKey, secretKey) diff --git a/proxy/service.go b/proxy/service.go index eb9e9958..f0a9b167 100644 --- a/proxy/service.go +++ b/proxy/service.go @@ -1179,8 +1179,8 @@ 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. +// Fully captured successful responses are cached for five seconds and replayed only +// for the same locally validated access key and bucket. func (s *Service) HandleCompleteMultipartUpload(w http.ResponseWriter, r *http.Request) error { bucket, key := ParseBucketKey(r) uploadId := r.URL.Query().Get("uploadId") @@ -1188,13 +1188,35 @@ func (s *Service) HandleCompleteMultipartUpload(w http.ResponseWriter, r *http.R log.Debug().Str("bucket", bucket).Str("key", key).Str("uploadId", uploadId).Msg("HandleCompleteMultipartUpload") + // Preserve transparent-mode identity before forwarding: a long completion can + // outlive the SigV4 age window before its response is captured. + var ( + requestAccessKey string + requestAuthValidated bool + requestAuthWasChecked bool + ) + if s.cache.IsEnabled() { + requestAccessKey, requestAuthValidated, requestAuthWasChecked = s.completionAccessKeyAtRequestStart(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 { + callerAccessKey := func() (string, bool) { + if requestAuthWasChecked { + // Keep the request-start signature result, but recheck the grant after + // GetCompletion's potentially remote read; that is the replay decision point. + if !requestAuthValidated || !s.completionAccessKeyStillAuthorized(requestAccessKey, r) { + return requestAccessKey, false + } + return requestAccessKey, true + } + return s.completionAccessKey(r) + } + entry, found, err := s.cache.GetCompletion(ctx, bucket, key, uploadId, callerAccessKey) + if err == nil && found && entry != nil { log.Debug().Str("uploadId", uploadId).Msg("CompleteMultipartUpload cache hit - returning cached response") for k, v := range entry.Headers { w.Header().Set(k, v) @@ -1271,19 +1293,49 @@ 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 { - log.Debug().Err(cacheErr).Msg("Failed to cache completion response") - // Don't fail the request if caching fails + // Cache successful completions only when the validated caller identity is + // available. The response is replayable only for that principal and bucket. + if completed && capture.Complete && s.cache.IsEnabled() { + accessKey := capture.authenticatedAccessKey + if accessKey == "" { + accessKey = requestAccessKey + } + if accessKey == "" { + accessKey, _ = s.completionAccessKey(r) + } + if accessKey != "" { + if cacheErr := s.cache.PutCompletion(ctx, bucket, key, uploadId, accessKey, 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 + } } } return nil } +func (s *Service) completionAccessKey(r *http.Request) (string, bool) { + validator, ok := s.forwarder.(completionAccessKeyValidator) + if !ok { + return "", false + } + return validator.authenticatedCompletionAccessKey(r) +} + +func (s *Service) completionAccessKeyAtRequestStart(r *http.Request) (string, bool, bool) { + validator, ok := s.forwarder.(completionAccessKeyAtRequestStart) + if !ok { + return "", false, false + } + accessKey, authenticated := validator.authenticatedCompletionAccessKeyAtRequestStart(r) + return accessKey, authenticated, true +} + +func (s *Service) completionAccessKeyStillAuthorized(accessKey string, r *http.Request) bool { + validator, ok := s.forwarder.(completionAccessKeyAtRequestStart) + return ok && validator.completionAccessKeyStillAuthorized(accessKey, r) +} + // ParseBucketKey extracts bucket and key from request path. func ParseBucketKey(r *http.Request) (bucket, key string) { path := strings.TrimPrefix(r.URL.Path, "/") From e2afe03620bfac5f2bef36a1f47b2018077d755c Mon Sep 17 00:00:00 2001 From: Ovais Tariq Date: Wed, 21 Oct 2026 14:34:29 -0700 Subject: [PATCH 6/6] test(proxy): remove duplicate lockedLogBuffer helper #238 and #256 each added an identical lockedLogBuffer to the proxy test package, so it no longer compiles on main. Keep the copy in lifecycle_test.go. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01NGTev5ePU5jHubGtdoUfK6 --- proxy/reinvalidation_test.go | 18 ------------------ 1 file changed, 18 deletions(-) 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