fix: authenticate cached multipart completion replays

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.
 

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,21 +19,32 @@ 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"`
 }
 
 // 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,

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(`<CompleteMultipartUploadResult><ETag>"abc123"</ETag></CompleteMultipartUploadResult>`)
 		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,14 +106,43 @@ 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"
 		if key != expected {
 			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)
 	}

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
 

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 <Error> 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)
+	}
 }

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