diff --git a/.claude/rules/transparent-proxy.md b/.claude/rules/transparent-proxy.md
index d26731cc..0b2c4be6 100644
--- a/.claude/rules/transparent-proxy.md
+++ b/.claude/rules/transparent-proxy.md
@@ -42,7 +42,8 @@ TAG adds four proxy headers using HMAC-SHA256 over `forwarded_host\ntimestamp\nm
 ## Per-Bucket Authorization
 
 - `AuthzCache` keyed on `(accessKey, bucket)`. Auth for one bucket does not grant access to others.
-- Granted when Tigris returns 2xx with signing keys
+- `AuthzCache` authorization is granted only after a successful upstream client `GetObject` or `HeadObject`; writes and other passthrough operations do not grant cached-read access.
+- Signing keys may be learned from any successful Tigris response carrying the internal header; they prove identity, not read permission.
 - Revoked when Tigris returns 403
 - TTL: 10 minutes (configurable via `TAG_AUTHZ_CACHE_TTL`)
 
diff --git a/README.md b/README.md
index c8d6fdb7..1b66c869 100644
--- a/README.md
+++ b/README.md
@@ -176,7 +176,7 @@ See [docs/metrics.md](docs/metrics.md) for complete metrics reference.
 
 ## Security
 
-TAG supports transparent proxy mode (default) with local SigV4 validation and per-bucket authorization caching, as well as signing mode with local credential stores.
+TAG supports transparent proxy mode (default) with local SigV4 validation and per-bucket authorization caching, as well as signing mode with local credential stores. In transparent mode, TAG grants cached-read authorization for a bucket only after a successful upstream `GetObject` or `HeadObject`. Successful writes and other passthrough responses do not grant cached-read access.
 
 See [docs/security.md](docs/security.md) for authentication, access control, and security architecture.
 
diff --git a/auth/authz_cache.go b/auth/authz_cache.go
index 57aee55a..90fdf761 100644
--- a/auth/authz_cache.go
+++ b/auth/authz_cache.go
@@ -15,7 +15,7 @@ const DefaultAuthzCacheTTL = 10 * time.Minute
 const maxAuthzCacheSize = 1_000_000
 
 // AuthzCache caches authorization decisions at the (access_key, bucket) level.
-// Entries are granted when Tigris returns a successful response with signing keys,
+// Entries are granted after a client GetObject or HeadObject succeeds upstream,
 // revoked when Tigris returns 403, and expire after a configurable TTL.
 //
 // Uses hashicorp/golang-lru/v2/expirable for TTL-based expiration and bounded size.
diff --git a/docs/security.md b/docs/security.md
index 53e533ba..0ea6a227 100644
--- a/docs/security.md
+++ b/docs/security.md
@@ -131,13 +131,13 @@ Request arrives at TAG
 
 ## Authorization Lifecycle
 
-Authorization decisions are cached per `(accessKey, bucket)` pair:
+Authorization decisions are cached per `(accessKey, bucket)` pair. TAG may learn signing keys from any successful response with the internal header. Those keys prove identity, not read permission. TAG grants bucket authorization only after a successful upstream `GetObject` or `HeadObject` response. Successful writes and other passthrough operations do not grant cached-read access.
 
-| Event                                | Action                                    |
-| ------------------------------------ | ----------------------------------------- |
-| Tigris returns 2xx with signing keys | `AuthzCache.Grant(accessKey, bucket)`     |
-| Tigris returns 403                   | `AuthzCache.Revoke(accessKey, bucket)`    |
-| TTL expires (10 min default)         | Entry removed, next request re-authorizes |
+| Event                                              | Action                                    |
+| -------------------------------------------------- | ----------------------------------------- |
+| Upstream 2xx to client `GetObject` or `HeadObject` | `AuthzCache.Grant(accessKey, bucket)`     |
+| Tigris returns 403                                 | `AuthzCache.Revoke(accessKey, bucket)`    |
+| TTL expires (10 min default)                       | Entry removed, next request re-authorizes |
 
 Authorization is strictly per-bucket. A client may have access to some buckets but not others, and TAG enforces this at the cache level.
 
diff --git a/proxy/forwarder_transparent.go b/proxy/forwarder_transparent.go
index 9a9ab6f4..44709627 100644
--- a/proxy/forwarder_transparent.go
+++ b/proxy/forwarder_transparent.go
@@ -41,13 +41,51 @@ 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.
+// upstream response but before headers are sent to the client. It learns
+// signing keys, grants authorization only for successful object reads, and
+// revokes authZ on 403s.
 func (f *transparentForwarder) interceptResponse(resp *http.Response, originalReq *http.Request) {
 	f.learnSigningKeys(resp, originalReq)
+	f.grantReadAuthorization(resp, originalReq)
 	f.handleAuthzRevocation(resp, originalReq)
 }
 
+// isReadAuthorizationQuery accepts only selectors and authentication metadata
+// known not to select a different S3 operation. The basic object route also
+// receives unrecognized GET/HEAD subresources, so an unknown or malformed
+// query cannot prove that the origin authorized GetObject or HeadObject.
+func isReadAuthorizationQuery(rawQuery string) bool {
+	if rawQuery == "" {
+		return true
+	}
+
+	query, err := url.ParseQuery(rawQuery)
+	if err != nil {
+		return false
+	}
+
+	for key := range query {
+		switch key {
+		case "versionId", "partNumber", "response-cache-control", "response-content-disposition",
+			"response-content-encoding", "response-content-language", "response-content-type", "response-expires",
+			"X-Amz-Checksum-Mode", "x-amz-expected-bucket-owner", "x-amz-request-payer":
+			continue
+		}
+
+		allowedAuthOrSDKParam := false
+		for _, param := range originlessIgnoredParams {
+			if key == param {
+				allowedAuthOrSDKParam = true
+				break
+			}
+		}
+		if !allowedAuthOrSDKParam {
+			return false
+		}
+	}
+	return true
+}
+
 // shallowHeaderCopy gives the forwarded request its own header map while
 // reusing the immutable value slices owned by the client request. Header.Set
 // and Header.Del replace or remove entries, so the values for headers changed
@@ -286,6 +324,7 @@ 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.
+// Learning identity keys does not grant authorization to read cached objects.
 func (f *transparentForwarder) learnSigningKeys(resp *http.Response, r *http.Request) {
 	// Always strip the internal header, even when local auth is disabled.
 	headerVal := resp.Header.Get(signingKeysHeader)
@@ -329,7 +368,6 @@ func (f *transparentForwarder) learnSigningKeys(resp *http.Response, r *http.Req
 	}
 
 	bucket, _ := ParseBucketKey(r)
-	f.authzCache.Grant(authInfo.AccessKey, bucket)
 
 	if newKeys > 0 {
 		log.Debug().
@@ -344,6 +382,29 @@ func (f *transparentForwarder) learnSigningKeys(resp *http.Response, r *http.Req
 	metrics.AuthzCacheSize.Set(float64(f.authzCache.Count()))
 }
 
+// grantReadAuthorization records bucket authorization only after an upstream
+// success from a client GetObject or HeadObject. The method, object path, and
+// query classifier reject other operations that can reach the basic object route.
+func (f *transparentForwarder) grantReadAuthorization(resp *http.Response, r *http.Request) {
+	if f.validator == nil || f.keyUnwrapper == nil || f.authzCache == nil ||
+		resp.StatusCode < 200 || resp.StatusCode >= 300 || r == nil ||
+		(r.Method != http.MethodGet && r.Method != http.MethodHead) || !isReadAuthorizationQuery(r.URL.RawQuery) {
+		return
+	}
+
+	authInfo, err := auth.ParseAuthInfo(r)
+	if err != nil {
+		return
+	}
+	bucket, key := ParseBucketKey(r)
+	if bucket == "" || key == "" {
+		return
+	}
+
+	f.authzCache.Grant(authInfo.AccessKey, bucket)
+	metrics.AuthzCacheSize.Set(float64(f.authzCache.Count()))
+}
+
 // handleAuthzRevocation revokes authorization when Tigris returns 403.
 func (f *transparentForwarder) handleAuthzRevocation(resp *http.Response, r *http.Request) {
 	if f.authzCache == nil || resp.StatusCode != http.StatusForbidden {
diff --git a/proxy/forwarder_transparent_test.go b/proxy/forwarder_transparent_test.go
index b6c6a69d..0ded84cc 100644
--- a/proxy/forwarder_transparent_test.go
+++ b/proxy/forwarder_transparent_test.go
@@ -1,6 +1,13 @@
 package proxy
 
 import (
+	"context"
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
 	"errors"
 	"io"
 	"net/http"
@@ -8,6 +15,7 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/tigrisdata/tag/auth"
 )
@@ -522,3 +530,165 @@ func TestTransparentForwarderSynchronousPathsKeepHeadersPrivate(t *testing.T) {
 		})
 	}
 }
+
+const transparentResponseTestSecret = "response-interceptor-test-secret"
+
+func transparentResponseTestEnvelope(t *testing.T, accessKey string, plaintext []byte) string {
+	t.Helper()
+	hash := sha256.Sum256([]byte(transparentResponseTestSecret))
+	block, err := aes.NewCipher(hash[:])
+	if err != nil {
+		t.Fatal(err)
+	}
+	gcm, err := cipher.NewGCM(block)
+	if err != nil {
+		t.Fatal(err)
+	}
+	nonce := make([]byte, gcm.NonceSize())
+	if _, err := rand.Read(nonce); err != nil {
+		t.Fatal(err)
+	}
+	raw := gcm.Seal(nonce, nonce, plaintext, []byte(accessKey))
+	return base64.StdEncoding.EncodeToString(raw)
+}
+
+func transparentResponseTestRequest(t *testing.T, method, path, accessKey string) *http.Request {
+	t.Helper()
+	request, err := auth.NewRequestSigner("https://upstream.example.com", "us-east-1").SignRequest(
+		context.Background(), method, path, nil, "", accessKey, "client-secret", nil,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return request
+}
+
+func TestResponseInterceptorStripsSigningKeysOnFailureAndDisabledAuth(t *testing.T) {
+	const accessKey = "INTERCEPTORKEY"
+	entries, err := json.Marshal([]auth.SigningKeyEntry{{
+		Date:       "20250211",
+		Region:     "auto",
+		SigningKey: strings.Repeat("a", 64),
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	validEnvelope := transparentResponseTestEnvelope(t, accessKey, entries)
+	wrongAADEnvelope := transparentResponseTestEnvelope(t, "OTHERACCESSKEY", entries)
+	corruptRaw, err := base64.StdEncoding.DecodeString(validEnvelope)
+	if err != nil {
+		t.Fatal(err)
+	}
+	corruptRaw[len(corruptRaw)-1] ^= 0x80
+	corruptEnvelope := base64.StdEncoding.EncodeToString(corruptRaw)
+	malformedEnvelope := transparentResponseTestEnvelope(t, accessKey, []byte("{"))
+	emptyEnvelope := transparentResponseTestEnvelope(t, accessKey, []byte("[]"))
+
+	tests := []struct {
+		name     string
+		status   int
+		envelope string
+		authOn   bool
+	}{
+		{name: "wrong AAD", status: http.StatusOK, envelope: wrongAADEnvelope, authOn: true},
+		{name: "corrupt ciphertext", status: http.StatusOK, envelope: corruptEnvelope, authOn: true},
+		{name: "malformed list", status: http.StatusOK, envelope: malformedEnvelope, authOn: true},
+		{name: "empty list", status: http.StatusOK, envelope: emptyEnvelope, authOn: true},
+		{name: "non-2xx response", status: http.StatusForbidden, envelope: validEnvelope, authOn: true},
+		{name: "local authentication disabled", status: http.StatusOK, envelope: validEnvelope, authOn: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			keys := auth.NewDerivedKeyStore(auth.DefaultDerivedKeyTTL)
+			authz := auth.NewAuthzCache(time.Minute)
+			fwd := &transparentForwarder{authzCache: authz}
+			if tt.authOn {
+				unwrapper, err := auth.NewKeyUnwrapper(transparentResponseTestSecret)
+				if err != nil {
+					t.Fatal(err)
+				}
+				fwd.derivedKeyStore = keys
+				fwd.validator = auth.NewRequestValidator(keys)
+				fwd.keyUnwrapper = unwrapper
+			}
+			response := &http.Response{
+				StatusCode: tt.status,
+				Header:     http.Header{signingKeysHeader: []string{tt.envelope}},
+			}
+			fwd.interceptResponse(response, transparentResponseTestRequest(t, http.MethodPost, "/bucket/key?uploads", accessKey))
+
+			if got := response.Header.Get(signingKeysHeader); got != "" {
+				t.Fatalf("internal signing-keys header was not stripped: %q", got)
+			}
+			if keys.HasKey(accessKey) {
+				t.Fatal("failed, empty, non-2xx, or disabled response unexpectedly learned a signing key")
+			}
+			if authz.IsAuthorized(accessKey, "bucket") {
+				t.Fatal("passthrough response unexpectedly granted read authorization")
+			}
+		})
+	}
+}
+
+func TestSuccessfulObjectReadGrantsAuthorizationWithoutSigningKeyHeader(t *testing.T) {
+	const accessKey = "READWITHOUTKEYHEADER"
+	keys := auth.NewDerivedKeyStore(auth.DefaultDerivedKeyTTL)
+	authz := auth.NewAuthzCache(time.Minute)
+	unwrapper, err := auth.NewKeyUnwrapper(transparentResponseTestSecret)
+	if err != nil {
+		t.Fatal(err)
+	}
+	fwd := &transparentForwarder{
+		derivedKeyStore: keys,
+		validator:       auth.NewRequestValidator(keys),
+		keyUnwrapper:    unwrapper,
+		authzCache:      authz,
+	}
+	request := transparentResponseTestRequest(t, http.MethodGet, "/bucket/key", accessKey)
+	response := &http.Response{StatusCode: http.StatusOK, Header: make(http.Header)}
+
+	fwd.interceptResponse(response, request)
+
+	if !authz.IsAuthorized(accessKey, "bucket") {
+		t.Fatal("successful client GetObject did not grant bucket authorization without a signing-keys header")
+	}
+	if keys.HasKey(accessKey) {
+		t.Fatal("read authorization unexpectedly created a signing key")
+	}
+}
+
+func TestReadAuthorizationQueryClassifierAcceptsOnlyObjectReadQueries(t *testing.T) {
+	const presignedReadParams = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKID%2Fdate%2Fregion%2Fs3%2Faws4_request&X-Amz-Date=20261005T180039Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Signature=signature&X-Amz-Security-Token=token&X-Amz-Checksum-Mode=ENABLED&x-amz-expected-bucket-owner=123456789012&x-amz-request-payer=requester"
+	tests := []struct {
+		name  string
+		query string
+		want  bool
+	}{
+		{name: "plain object read", want: true},
+		{name: "versioned read", query: "versionId=v1", want: true},
+		{name: "part read", query: "partNumber=2", want: true},
+		{name: "response overrides", query: "response-cache-control=no-cache&response-content-disposition=attachment&response-content-encoding=gzip&response-content-language=en&response-content-type=text%2Fplain&response-expires=Wed%2C21%20Oct%202015%2007%3A28%3A00%20GMT", want: true},
+		{name: "presigned GetObject", query: presignedReadParams + "&x-id=GetObject", want: true},
+		{name: "presigned HeadObject", query: presignedReadParams, want: true},
+		{name: "GetObjectAttributes", query: "attributes", want: false},
+		{name: "attribute selector with read selectors", query: "attributes&versionId=v1&x-id=GetObjectAttributes", want: false},
+		{name: "object tagging", query: "tagging", want: false},
+		{name: "object ACL", query: "acl", want: false},
+		{name: "list parts", query: "uploadId=upload", want: false},
+		{name: "object lock retention", query: "retention", want: false},
+		{name: "object lock legal hold", query: "legal-hold", want: false},
+		{name: "torrent", query: "torrent", want: false},
+		{name: "unknown operation", query: "unknown=1", want: false},
+		{name: "malformed query", query: "attributes%ZZ", want: false},
+		{name: "malformed separator", query: "attributes;versionId=v1", want: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isReadAuthorizationQuery(tt.query); got != tt.want {
+				t.Fatalf("read authorization query = %v, want %v for %q", got, tt.want, tt.query)
+			}
+		})
+	}
+}
diff --git a/tests/integration/transparent_authz_scope_test.go b/tests/integration/transparent_authz_scope_test.go
new file mode 100644
index 00000000..99b0bb67
--- /dev/null
+++ b/tests/integration/transparent_authz_scope_test.go
@@ -0,0 +1,430 @@
+package integration
+
+import (
+	"context"
+	"encoding/hex"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/tigrisdata/tag/auth"
+)
+
+const (
+	authzScopeBucket            = "scope-write-bucket"
+	authzScopeObject            = "private-object"
+	authzScopeBody              = "private reader-only object body"
+	authzScopeWriter            = "WRITEONLYKEY"
+	authzScopeWriterSecret      = "writer-secret"
+	authzScopeReader            = "READERKEY"
+	authzScopeReaderSecret      = "reader-secret"
+	authzScopeHeadReader        = "HEADONLYKEY"
+	authzScopeHeadSecret        = "head-reader-secret"
+	authzScopeOtherReader       = "OTHERREADERKEY"
+	authzScopeOtherReaderSecret = "other-reader-secret"
+)
+
+// This origin enforces permissions independently of TAG's AuthzCache.
+type authzScopeOrigin struct {
+	bucket      string
+	otherBucket string
+	validator   *auth.RequestValidator
+
+	mu        sync.RWMutex
+	envelopes map[string]string
+
+	writerReads            atomic.Int64
+	writerAttributes       atomic.Int64
+	readerReads            atomic.Int64
+	readerOtherBucketReads atomic.Int64
+	headReads              atomic.Int64
+	writerTags             atomic.Int64
+	otherReaderReads       atomic.Int64
+	readerDenied           atomic.Bool
+}
+
+func newAuthzScopeTestEnvironment(t *testing.T, bucket string) (*TestEnvironment, *authzScopeOrigin, *auth.CredentialStore) {
+	t.Helper()
+	credentials := auth.NewCredentialStore()
+	credentials.AddCredential(authzScopeWriter, authzScopeWriterSecret)
+	credentials.AddCredential(authzScopeReader, authzScopeReaderSecret)
+	credentials.AddCredential(authzScopeHeadReader, authzScopeHeadSecret)
+	credentials.AddCredential(authzScopeOtherReader, authzScopeOtherReaderSecret)
+	origin := &authzScopeOrigin{
+		bucket:      bucket,
+		otherBucket: bucket + "-other",
+		validator:   auth.NewRequestValidator(credentials),
+		envelopes:   make(map[string]string),
+	}
+	env := NewTestEnvironmentWithTransparentAuth(t, origin.serveHTTP)
+	return env, origin, credentials
+}
+
+func (o *authzScopeOrigin) setSigningKeys(accessKey, envelope string) {
+	o.mu.Lock()
+	o.envelopes[accessKey] = envelope
+	o.mu.Unlock()
+}
+
+func (o *authzScopeOrigin) signingKeys(accessKey string) string {
+	o.mu.RLock()
+	defer o.mu.RUnlock()
+	return o.envelopes[accessKey]
+}
+
+func (o *authzScopeOrigin) serveHTTP(w http.ResponseWriter, r *http.Request) {
+	original := r.Clone(r.Context())
+	if forwardedHost := r.Header.Get("X-Tigris-Forwarded-Host"); forwardedHost != "" {
+		original.Host = forwardedHost
+	}
+	accessKey, err := o.validator.ValidateRequest(original)
+	if err != nil {
+		authzScopeDeny(w)
+		return
+	}
+
+	objectPath := "/" + o.bucket + "/" + authzScopeObject
+	otherObjectPath := "/" + o.otherBucket + "/" + authzScopeObject
+	switch {
+	case accessKey == authzScopeWriter && r.Method == http.MethodPost && r.URL.Path == objectPath && r.URL.Query().Has("uploads"):
+		envelope := o.signingKeys(authzScopeWriter)
+		if envelope == "" {
+			authzScopeDeny(w)
+			return
+		}
+		w.Header().Set("X-Tigris-Proxy-Signing-Keys", envelope)
+		w.Header().Set("Content-Type", "application/xml")
+		w.WriteHeader(http.StatusOK)
+		_, _ = io.WriteString(w, "<InitiateMultipartUploadResult><UploadId>scope-upload</UploadId></InitiateMultipartUploadResult>")
+	case accessKey == authzScopeWriter && r.Method == http.MethodGet && r.URL.Path == objectPath && r.URL.Query().Has("attributes"):
+		o.writerAttributes.Add(1)
+		envelope := o.signingKeys(authzScopeWriter)
+		if envelope == "" {
+			authzScopeDeny(w)
+			return
+		}
+		w.Header().Set("X-Tigris-Proxy-Signing-Keys", envelope)
+		w.Header().Set("Cache-Control", "no-store")
+		w.Header().Set("Content-Type", "application/xml")
+		w.WriteHeader(http.StatusOK)
+		_, _ = io.WriteString(w, "<GetObjectAttributesOutput/>")
+	case accessKey == authzScopeWriter && r.Method == http.MethodGet && r.URL.Path == objectPath && r.URL.Query().Has("tagging"):
+		o.writerTags.Add(1)
+		envelope := o.signingKeys(authzScopeWriter)
+		if envelope == "" {
+			authzScopeDeny(w)
+			return
+		}
+		w.Header().Set("X-Tigris-Proxy-Signing-Keys", envelope)
+		w.Header().Set("Content-Type", "application/xml")
+		w.WriteHeader(http.StatusOK)
+		_, _ = io.WriteString(w, "<Tagging/>")
+	case r.URL.Path == objectPath && r.URL.RawQuery == "" && r.Method == http.MethodGet && accessKey == authzScopeWriter:
+		o.writerReads.Add(1)
+		authzScopeDeny(w)
+	case r.URL.Path == objectPath && r.URL.RawQuery == "" && r.Method == http.MethodGet && accessKey == authzScopeReader:
+		o.readerReads.Add(1)
+		if o.readerDenied.Load() {
+			authzScopeDeny(w)
+			return
+		}
+		envelope := o.signingKeys(authzScopeReader)
+		if envelope == "" {
+			authzScopeDeny(w)
+			return
+		}
+		authzScopeWriteObject(w, true, envelope)
+	case r.URL.Path == objectPath && r.URL.RawQuery == "" && r.Method == http.MethodHead && accessKey == authzScopeHeadReader:
+		o.headReads.Add(1)
+		envelope := o.signingKeys(authzScopeHeadReader)
+		if envelope == "" {
+			authzScopeDeny(w)
+			return
+		}
+		authzScopeWriteObject(w, false, envelope)
+	case r.URL.Path == otherObjectPath && r.URL.RawQuery == "" && r.Method == http.MethodGet && accessKey == authzScopeOtherReader:
+		o.otherReaderReads.Add(1)
+		envelope := o.signingKeys(authzScopeOtherReader)
+		if envelope == "" {
+			authzScopeDeny(w)
+			return
+		}
+		authzScopeWriteObject(w, true, envelope)
+	case r.URL.Path == otherObjectPath && r.URL.RawQuery == "" && r.Method == http.MethodGet && accessKey == authzScopeReader:
+		o.readerOtherBucketReads.Add(1)
+		authzScopeDeny(w)
+	default:
+		authzScopeDeny(w)
+	}
+}
+
+func authzScopeDeny(w http.ResponseWriter) {
+	w.Header().Set("Content-Type", "application/xml")
+	w.WriteHeader(http.StatusForbidden)
+	_, _ = io.WriteString(w, "<Error><Code>AccessDenied</Code></Error>")
+}
+
+func authzScopeWriteObject(w http.ResponseWriter, withBody bool, envelope string) {
+	w.Header().Set("ETag", `"scope-etag"`)
+	w.Header().Set("Content-Length", "31")
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.Header().Set("X-Amz-Acl", "private")
+	w.Header().Set("X-Tigris-Proxy-Signing-Keys", envelope)
+	w.WriteHeader(http.StatusOK)
+	if withBody {
+		_, _ = io.WriteString(w, authzScopeBody)
+	}
+}
+
+func authzScopeRequest(t *testing.T, endpoint, method, path, accessKey, secret string) *http.Request {
+	t.Helper()
+	request, err := auth.NewRequestSigner(endpoint, TestRegion).SignRequest(context.Background(), method, path, nil, "", accessKey, secret, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return request
+}
+
+func authzScopeKeysForRequest(t *testing.T, credentials *auth.CredentialStore, accessKey string, request *http.Request) (string, []string) {
+	t.Helper()
+	info, err := auth.ParseAuthInfo(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	date, err := time.Parse("20060102", info.Date)
+	if err != nil {
+		t.Fatalf("parse signing date %q: %v", info.Date, err)
+	}
+	dates := []string{info.Date, date.Add(-24 * time.Hour).Format("20060102")}
+	entries := make([]auth.SigningKeyEntry, 0, len(dates))
+	for _, signingDate := range dates {
+		key, err := credentials.GetSigningKey(accessKey, signingDate, info.Region)
+		if err != nil {
+			t.Fatal(err)
+		}
+		entries = append(entries, auth.SigningKeyEntry{Date: signingDate, Region: info.Region, SigningKey: hex.EncodeToString(key)})
+	}
+	return encryptSigningKeysHeader(t, TestProxySecretKey, accessKey, entries), dates
+}
+
+func authzScopeRead(t *testing.T, client *http.Client, request *http.Request) (int, http.Header, string) {
+	t.Helper()
+	response, err := client.Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	body, err := io.ReadAll(response.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return response.StatusCode, response.Header, string(body)
+}
+
+func waitForAuthzScopeCache(env *TestEnvironment, bucket, key string, timeout time.Duration) bool {
+	ticker := time.NewTicker(time.Millisecond)
+	defer ticker.Stop()
+	deadline := time.After(timeout)
+	for {
+		if env.IsCached(bucket, key) {
+			return true
+		}
+		select {
+		case <-ticker.C:
+		case <-deadline:
+			return false
+		}
+	}
+}
+
+func TestTransparentAuth_WriteOnlyMultipartDoesNotGrantCachedRead(t *testing.T) {
+	env, origin, credentials := newAuthzScopeTestEnvironment(t, authzScopeBucket)
+	defer env.Close()
+	client := &http.Client{Timeout: 5 * time.Second}
+	path := "/" + authzScopeBucket + "/" + authzScopeObject
+
+	// The origin denies a valid writer GetObject independently of TAG.
+	status, _, body := authzScopeRead(t, client, authzScopeRequest(t, env.UpstreamServer.URL, http.MethodGet, path, authzScopeWriter, authzScopeWriterSecret))
+	if status != http.StatusForbidden || strings.Contains(body, authzScopeBody) {
+		t.Fatalf("origin must deny writer GetObject: status=%d private_body=%v", status, strings.Contains(body, authzScopeBody))
+	}
+
+	// An allowed reader populates the private object through TAG's real router.
+	readerRequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodGet, path, authzScopeReader, authzScopeReaderSecret)
+	readerEnvelope, _ := authzScopeKeysForRequest(t, credentials, authzScopeReader, readerRequest)
+	origin.setSigningKeys(authzScopeReader, readerEnvelope)
+	status, headers, responseBody := authzScopeRead(t, client, readerRequest)
+	if status != http.StatusOK || responseBody != authzScopeBody || headers.Get("X-Tigris-Proxy-Signing-Keys") != "" {
+		t.Fatalf("authorized reader GetObject: status=%d body=%q leaked_header=%q", status, responseBody, headers.Get("X-Tigris-Proxy-Signing-Keys"))
+	}
+	if !waitForAuthzScopeCache(env, authzScopeBucket, authzScopeObject, 2*time.Second) {
+		t.Fatal("authorized reader did not populate the private object cache")
+	}
+	meta, found, err := env.Cache.GetMeta(context.Background(), authzScopeBucket, authzScopeObject)
+	if err != nil || !found || meta == nil || meta.IsPublicRead() {
+		t.Fatalf("cached object must remain private: found=%v meta=%v error=%v", found, meta, err)
+	}
+
+	readerOriginGets := origin.readerReads.Load()
+	status, headers, responseBody = authzScopeRead(t, client, readerRequest.Clone(context.Background()))
+	if status != http.StatusOK || headers.Get("X-Cache") != "HIT" || responseBody != authzScopeBody || origin.readerReads.Load() != readerOriginGets {
+		t.Fatalf("authorized reader cache hit: status=%d cache=%q body=%q origin GET delta=%d", status, headers.Get("X-Cache"), responseBody, origin.readerReads.Load()-readerOriginGets)
+	}
+
+	// A separate HeadObject identity also learns keys from an upstream HEAD and
+	// then serves the object's metadata from the shared integration cache.
+	headRequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodHead, path, authzScopeHeadReader, authzScopeHeadSecret)
+	headEnvelope, _ := authzScopeKeysForRequest(t, credentials, authzScopeHeadReader, headRequest)
+	origin.setSigningKeys(authzScopeHeadReader, headEnvelope)
+	status, headers, responseBody = authzScopeRead(t, client, headRequest)
+	if status != http.StatusOK || responseBody != "" || headers.Get("X-Tigris-Proxy-Signing-Keys") != "" || headers.Get("Content-Length") != "31" || !env.AuthzCache.IsAuthorized(authzScopeHeadReader, authzScopeBucket) {
+		t.Fatalf("authorized HeadObject: status=%d body=%q content_length=%q leaked_header=%q grant=%v", status, responseBody, headers.Get("Content-Length"), headers.Get("X-Tigris-Proxy-Signing-Keys"), env.AuthzCache.IsAuthorized(authzScopeHeadReader, authzScopeBucket))
+	}
+	headOriginGets := origin.headReads.Load()
+	status, headers, responseBody = authzScopeRead(t, client, headRequest.Clone(context.Background()))
+	if status != http.StatusOK || responseBody != "" || headers.Get("Content-Length") != "31" || headers.Get("X-Cache") != "HIT" || origin.headReads.Load() != headOriginGets {
+		t.Fatalf("authorized HeadObject cache hit: status=%d body=%q cache=%q origin HEAD delta=%d", status, responseBody, headers.Get("X-Cache"), origin.headReads.Load()-headOriginGets)
+	}
+
+	// The signing-key response still completes multipart initiation and teaches
+	// the writer the keys for its exact later GetObject signature.
+	writerReadRequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodGet, path, authzScopeWriter, authzScopeWriterSecret)
+	writeRequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodPost, path+"?uploads", authzScopeWriter, authzScopeWriterSecret)
+	writerEnvelope, writerDates := authzScopeKeysForRequest(t, credentials, authzScopeWriter, writeRequest)
+	origin.setSigningKeys(authzScopeWriter, writerEnvelope)
+	status, headers, responseBody = authzScopeRead(t, client, writeRequest)
+	if status != http.StatusOK || responseBody != "<InitiateMultipartUploadResult><UploadId>scope-upload</UploadId></InitiateMultipartUploadResult>" || headers.Get("X-Tigris-Proxy-Signing-Keys") != "" {
+		t.Fatalf("multipart initiation: status=%d body=%q leaked_header=%q", status, responseBody, headers.Get("X-Tigris-Proxy-Signing-Keys"))
+	}
+	if env.AuthzCache.IsAuthorized(authzScopeWriter, authzScopeBucket) {
+		t.Fatal("successful multipart initiation granted the writer cached-read access")
+	}
+	for _, date := range writerDates {
+		key, err := env.DerivedKeyStore.GetSigningKey(authzScopeWriter, date, TestRegion)
+		if err != nil || len(key) != 32 {
+			t.Fatalf("multipart response did not teach the writer's 32-byte key for %s: length=%d error=%v", date, len(key), err)
+		}
+	}
+
+	// A successful tagging subresource is still passthrough, not GetObject.
+	tagRequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodGet, path+"?tagging", authzScopeWriter, authzScopeWriterSecret)
+	status, headers, _ = authzScopeRead(t, client, tagRequest)
+	if status != http.StatusOK || headers.Get("X-Tigris-Proxy-Signing-Keys") != "" || origin.writerTags.Load() != 1 {
+		t.Fatalf("object-tagging passthrough: status=%d leaked_header=%q origin calls=%d", status, headers.Get("X-Tigris-Proxy-Signing-Keys"), origin.writerTags.Load())
+	}
+	if env.AuthzCache.IsAuthorized(authzScopeWriter, authzScopeBucket) {
+		t.Fatal("successful object-tagging passthrough granted the writer cached-read access")
+	}
+
+	// GetObjectAttributes also uses GET, but its success does not prove that the
+	// client can read the object's body. The basic object route must not turn it
+	// into a bucket cache-read grant.
+	attributesRequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodGet, path+"?attributes", authzScopeWriter, authzScopeWriterSecret)
+	attributesEnvelope, _ := authzScopeKeysForRequest(t, credentials, authzScopeWriter, attributesRequest)
+	origin.setSigningKeys(authzScopeWriter, attributesEnvelope)
+	attributesRequest.Header.Set("Cache-Control", "no-store")
+	status, headers, responseBody = authzScopeRead(t, client, attributesRequest)
+	if status != http.StatusOK || responseBody != "<GetObjectAttributesOutput/>" || headers.Get("X-Tigris-Proxy-Signing-Keys") != "" || origin.writerAttributes.Load() != 1 {
+		t.Fatalf("GetObjectAttributes: status=%d body=%q leaked_header=%q origin calls=%d", status, responseBody, headers.Get("X-Tigris-Proxy-Signing-Keys"), origin.writerAttributes.Load())
+	}
+	if env.AuthzCache.IsAuthorized(authzScopeWriter, authzScopeBucket) {
+		t.Fatal("successful GetObjectAttributes granted the writer cached-read access")
+	}
+
+	// Check that this request is locally verifiable from the keys learned above.
+	if _, err := auth.NewRequestValidator(env.DerivedKeyStore).ValidateRequest(writerReadRequest); err != nil {
+		t.Fatalf("multipart response keys do not validate the writer's GetObject signature: %v", err)
+	}
+	writerOriginGets := origin.writerReads.Load()
+	status, headers, responseBody = authzScopeRead(t, client, writerReadRequest)
+	if status != http.StatusForbidden || strings.Contains(responseBody, authzScopeBody) || origin.writerReads.Load() != writerOriginGets+1 {
+		t.Fatalf("writer GetObject must reach origin and be denied: status=%d cache=%q private_body=%v origin GET delta=%d", status, headers.Get("X-Cache"), strings.Contains(responseBody, authzScopeBody), origin.writerReads.Load()-writerOriginGets)
+	}
+}
+
+func TestTransparentAuth_ReadGrantRevokedAfter403(t *testing.T) {
+	const bucket = "scope-revoke-bucket"
+	env, origin, credentials := newAuthzScopeTestEnvironment(t, bucket)
+	defer env.Close()
+	client := &http.Client{Timeout: 5 * time.Second}
+	path := "/" + bucket + "/" + authzScopeObject
+	readerRequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodGet, path, authzScopeReader, authzScopeReaderSecret)
+	readerEnvelope, _ := authzScopeKeysForRequest(t, credentials, authzScopeReader, readerRequest)
+	origin.setSigningKeys(authzScopeReader, readerEnvelope)
+	status, _, responseBody := authzScopeRead(t, client, readerRequest)
+	if status != http.StatusOK || responseBody != authzScopeBody || !env.AuthzCache.IsAuthorized(authzScopeReader, bucket) {
+		t.Fatalf("initial authorized GetObject: status=%d grant=%v", status, env.AuthzCache.IsAuthorized(authzScopeReader, bucket))
+	}
+
+	origin.readerDenied.Store(true)
+	request := authzScopeRequest(t, env.TAGServer.URL, http.MethodGet, path, authzScopeReader, authzScopeReaderSecret)
+	request.Header.Set("Cache-Control", "no-store")
+	status, _, responseBody = authzScopeRead(t, client, request)
+	if status != http.StatusForbidden || strings.Contains(responseBody, authzScopeBody) {
+		t.Fatalf("origin-denied GetObject: status=%d private_body=%v", status, strings.Contains(responseBody, authzScopeBody))
+	}
+	if env.AuthzCache.IsAuthorized(authzScopeReader, bucket) {
+		t.Fatal("upstream 403 did not revoke the read grant")
+	}
+}
+
+func TestTransparentAuth_AuthzGrantRemainsBucketScoped(t *testing.T) {
+	const bucketA = "scope-authz-bucket-a"
+	bucketB := bucketA + "-other"
+	env, origin, credentials := newAuthzScopeTestEnvironment(t, bucketA)
+	defer env.Close()
+	client := &http.Client{Timeout: 5 * time.Second}
+	pathA := "/" + bucketA + "/" + authzScopeObject
+	pathB := "/" + bucketB + "/" + authzScopeObject
+
+	// The same access key is allowed to read bucket A but not bucket B.
+	readerARequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodGet, pathA, authzScopeReader, authzScopeReaderSecret)
+	readerBRequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodGet, pathB, authzScopeReader, authzScopeReaderSecret)
+	readerEnvelope, _ := authzScopeKeysForRequest(t, credentials, authzScopeReader, readerBRequest)
+	origin.setSigningKeys(authzScopeReader, readerEnvelope)
+	status, headers, body := authzScopeRead(t, client, readerARequest)
+	if status != http.StatusOK || body != authzScopeBody || headers.Get("X-Tigris-Proxy-Signing-Keys") != "" {
+		t.Fatalf("authorized bucket-A GetObject: status=%d body=%q leaked_header=%q", status, body, headers.Get("X-Tigris-Proxy-Signing-Keys"))
+	}
+	if !waitForAuthzScopeCache(env, bucketA, authzScopeObject, 2*time.Second) {
+		t.Fatal("reader A did not populate bucket A")
+	}
+	if !env.AuthzCache.IsAuthorized(authzScopeReader, bucketA) || env.AuthzCache.IsAuthorized(authzScopeReader, bucketB) {
+		t.Fatalf("bucket A grant leaked to bucket B: A=%v B=%v", env.AuthzCache.IsAuthorized(authzScopeReader, bucketA), env.AuthzCache.IsAuthorized(authzScopeReader, bucketB))
+	}
+
+	// Another identity populates a private cache entry in bucket B. Reader A's
+	// bucket-A grant must not make this private object readable from cache.
+	otherReaderRequest := authzScopeRequest(t, env.TAGServer.URL, http.MethodGet, pathB, authzScopeOtherReader, authzScopeOtherReaderSecret)
+	otherReaderEnvelope, _ := authzScopeKeysForRequest(t, credentials, authzScopeOtherReader, otherReaderRequest)
+	origin.setSigningKeys(authzScopeOtherReader, otherReaderEnvelope)
+	status, headers, body = authzScopeRead(t, client, otherReaderRequest)
+	if status != http.StatusOK || body != authzScopeBody || headers.Get("X-Tigris-Proxy-Signing-Keys") != "" {
+		t.Fatalf("authorized bucket-B cache fill: status=%d body=%q leaked_header=%q", status, body, headers.Get("X-Tigris-Proxy-Signing-Keys"))
+	}
+	if !waitForAuthzScopeCache(env, bucketB, authzScopeObject, 2*time.Second) || !env.AuthzCache.IsAuthorized(authzScopeOtherReader, bucketB) {
+		t.Fatal("other reader did not authorize and populate bucket B")
+	}
+	meta, found, err := env.Cache.GetMeta(context.Background(), bucketB, authzScopeObject)
+	if err != nil || !found || meta == nil || meta.IsPublicRead() {
+		t.Fatalf("bucket-B cache entry must remain private: found=%v meta=%v error=%v", found, meta, err)
+	}
+	otherReaderOriginGets := origin.otherReaderReads.Load()
+	status, headers, body = authzScopeRead(t, client, otherReaderRequest.Clone(context.Background()))
+	if status != http.StatusOK || headers.Get("X-Cache") != "HIT" || body != authzScopeBody || origin.otherReaderReads.Load() != otherReaderOriginGets {
+		t.Fatalf("authorized other-reader bucket-B cache hit: status=%d cache=%q body=%q origin GET delta=%d", status, headers.Get("X-Cache"), body, origin.otherReaderReads.Load()-otherReaderOriginGets)
+	}
+
+	readerOriginGets := origin.readerOtherBucketReads.Load()
+	status, headers, body = authzScopeRead(t, client, readerBRequest)
+	if status != http.StatusForbidden || strings.Contains(body, authzScopeBody) || origin.readerOtherBucketReads.Load() != readerOriginGets+1 {
+		t.Fatalf("reader A must reach origin and be denied in bucket B: status=%d cache=%q private_body=%v origin GET delta=%d", status, headers.Get("X-Cache"), strings.Contains(body, authzScopeBody), origin.readerOtherBucketReads.Load()-readerOriginGets)
+	}
+	if !env.AuthzCache.IsAuthorized(authzScopeReader, bucketA) || env.AuthzCache.IsAuthorized(authzScopeReader, bucketB) {
+		t.Fatalf("bucket-B denial changed bucket grants: A=%v B=%v", env.AuthzCache.IsAuthorized(authzScopeReader, bucketA), env.AuthzCache.IsAuthorized(authzScopeReader, bucketB))
+	}
+}