s3: end a catalog-vended session with the access key it was vended for (#11945)

* s3: end a catalog-vended session with the access key it was vended for

A table credential the Iceberg catalog vends is a self-contained STS session:
it stays valid until it expires (an hour by default) even after the access key
that loaded the table is deleted or deactivated, or its identity is removed
or disabled. An operator revoking a leaked key cannot cut off the table
credentials that key already obtained.

Sessions are stateless JWTs, so there is no revocation list to consult. This
change follows the same pattern as role sessions, which are already refused
once their role's ID no longer matches (checkSessionRoleBinding): a vended
session carries the access key it was vended for (src_ak, recorded by the
previous change), and validateSTSSessionToken checks it against the identity
table held in memory. A session of provider iceberg-catalog that carries a
source key is refused (InvalidAccessKeyId, as for an unknown key) once its
identity is gone or disabled, or the key is no longer one of that identity's
active, unexpired credentials.

Scope:
- Only catalog-vended sessions are checked; web-identity, AssumeRole and
  every other STS session are validated as before.
- The vendor binds a session to the caller's key only when the caller is a
  stored identity holding that key. A caller that signed with another STS
  session's key, an OIDC user, or a request that recorded no key gets an
  unbound session, as today; binding it would refuse every request made with
  it. A session minted before this change carries no key and is unbound.
- The check is one map lookup under the read lock and a scan of one
  identity's keys (about 11 ns, 0 allocations, with 1,000 identities).

Tests (s3api_iceberg_vended_session_test.go) mint real sessions through the
catalog vendor and validate them the way a data-path request is validated:
deactivating, expiring or deleting key A ends A's session while the session
vended through key B on the same identity keeps working; disabling or
deleting the identity ends it; callers that are not a stored identity holding
the key get unbound sessions that keep working; other STS sessions are not
checked. The first two tests and the third fail without this change.

* s3: keep the vended-session key binding when the key dies before vend

vendedSessionKeyLive gated the binding, so a key revoked between the
request's authentication and the vend skipped it and minted an unbound
session that stayed usable until it expired. Every catalog auth path
pairs the identity name in context with the key that authenticated, so
a stored-identity principal always means the request signed with that
identity's own key. Bind whenever the principal is a stored identity: a
session bound to a now-dead key is refused on its first request, which
is what the binding is for.

* s3: gate the vended-session key binding on the authenticated caller type

lookupByIdentityName(principal) named only that some identity exists, not
that the request signed with a stored key: an STS session or OIDC principal
that collides with an identity name got a session bound to its temporary
key, which vendedSessionKeyLive then refused on every request. Verifiers
now record the verified key as a stored credential, and the vendor binds
only when that marker is present — keeping the binding even when the key
was revoked between authentication and the vend.

* s3: enforce the vended key binding under a dedicated claim

Sessions minted before the binding existed carry the caller's key in the
audit-only SourceAccessKeyId — including temporary STS keys that can never
resolve in the identity table. Enforcing on that field would end live table
credentials at upgrade; the binding now travels under BoundAccessKeyId,
leaving legacy sessions unbound as before.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: bind vended sessions to a credential fingerprint

The key id alone cannot identify a credential: an operator can delete a
key and re-create it under the same id with a different secret, reviving
sessions minted for the original. Verifiers now record a fingerprint of
the secret at authentication; the session dies unless it still matches.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: keep recording the caller's key on every catalog session

SourceAccessKeyId was unconditional audit metadata before the binding
existed; gating it on the stored-credential marker dropped the record
for callers that authenticated another way. With the binding under its
own claim there is no ambiguity left, so the field records the caller's
key again regardless of how it authenticated.

* s3: key the vended-session credential fingerprint with the signing key

The bound-credential fingerprint travels inside the vended session
token. As a plain SHA-256 of the secret it gave any token holder an
offline guessing oracle for weak secrets, and CodeQL flags it twice.

Store the verified secret in the request marker instead and HMAC it
with the STS signing key at vend and at validation. The claim can no
longer be mined, and the per-request auth path stops hashing secrets.

---------

Co-authored-by: Javier de la Torre <jatorre@lakehousebox.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/iam/sts/session_claims.go

@@ -105,6 +105,13 @@ type STSSessionClaims struct {
 	// a SeaweedFS service minted the session for it (AssumeRoleForPrincipal);
 	// empty for every other session type.
 	SourceAccessKeyId string `json:"src_ak,omitempty"`
+
+	// BoundAccessKeyId is the stored credential the session is bound to: the
+	// session ends when that key is gone. It is a claim of its own because
+	// tokens minted before the binding existed carry SourceAccessKeyId as an
+	// audit field, never a binding — enforcing those would break them.
+	BoundAccessKeyId          string `json:"bind_ak,omitempty"`
+	BoundAccessKeyFingerprint string `json:"bind_fp,omitempty"`
 }
 
 // NewSTSSessionClaims creates new STS session claims with all required information
@@ -163,6 +170,9 @@ func (c *STSSessionClaims) ToSessionInfo(credGen *CredentialGenerator) *SessionI
 		ParentUser:       c.ParentUser,
 		// The access key the session was minted for, if recorded
 		SourceAccessKeyId: c.SourceAccessKeyId,
+		// The stored credential the session is bound to, if any
+		BoundAccessKeyId:          c.BoundAccessKeyId,
+		BoundAccessKeyFingerprint: c.BoundAccessKeyFingerprint,
 		// Provide the Subject (sub) from registered claims
 		Subject:     c.Subject,
 		Credentials: credentials,
@@ -263,6 +273,14 @@ func (c *STSSessionClaims) WithSourceAccessKeyId(accessKeyId string) *STSSession
 	return c
 }
 
+// WithBoundAccessKey binds the session to a stored credential: the key id it
+// was vended for and a fingerprint of its secret at mint time.
+func (c *STSSessionClaims) WithBoundAccessKey(accessKeyId, fingerprint string) *STSSessionClaims {
+	c.BoundAccessKeyId = accessKeyId
+	c.BoundAccessKeyFingerprint = fingerprint
+	return c
+}
+
 // WithParentUser sets the stable per-identity hash for the session. See
 // ComputeParentUser for the derivation rule.
 func (c *STSSessionClaims) WithParentUser(parentUser string) *STSSessionClaims {

weed/iam/sts/sts_service.go

@@ -320,6 +320,16 @@ type SessionInfo struct {
 	// SourceAccessKeyId is the access key the session was minted for, when the
 	// minting service recorded it (AssumeRoleForPrincipal).
 	SourceAccessKeyId string `json:"sourceAccessKeyId,omitempty"`
+
+	// BoundAccessKeyId is the stored credential the session is bound to, when
+	// the minting service recorded one (AssumeRoleForPrincipal). A legacy
+	// SourceAccessKeyId is never treated as a binding.
+	BoundAccessKeyId string `json:"boundAccessKeyId,omitempty"`
+
+	// BoundAccessKeyFingerprint identifies the credential instance the
+	// session is bound to, so a key re-created under the same id does not
+	// match.
+	BoundAccessKeyFingerprint string `json:"boundAccessKeyFingerprint,omitempty"`
 }
 
 // NewSTSService creates a new STS service
@@ -901,6 +911,14 @@ type AssumeRoleForPrincipalRequest struct {
 	// SourceAccessKeyId optionally records the access key the caller
 	// authenticated with, so the session can be tied to that key.
 	SourceAccessKeyId string
+
+	// BoundAccessKeyId binds the session to a stored credential: the session
+	// ends when that key is gone. Distinct from SourceAccessKeyId, which is
+	// audit-only on sessions minted before the binding existed.
+	BoundAccessKeyId string
+
+	// BoundAccessKeyFingerprint identifies the bound credential instance.
+	BoundAccessKeyFingerprint string
 }
 
 // AssumeRoleForPrincipal issues session credentials for a caller that a
@@ -947,7 +965,10 @@ func (s *STSService) AssumeRoleForPrincipal(ctx context.Context, request *Assume
 	// carry no role ID and stay bound by name.
 	return s.issueSession(request.RoleArn, "", request.RoleSessionName, sessionPolicy,
 		request.DurationSeconds, request.ProviderName, request.Principal,
-		func(c *STSSessionClaims) { c.WithSourceAccessKeyId(request.SourceAccessKeyId) })
+		func(c *STSSessionClaims) {
+			c.WithSourceAccessKeyId(request.SourceAccessKeyId)
+			c.WithBoundAccessKey(request.BoundAccessKeyId, request.BoundAccessKeyFingerprint)
+		})
 }
 
 // ValidateSessionToken validates a session token and returns session information

weed/s3api/auth_signature_v2.go

@@ -110,6 +110,7 @@ func (iam *IdentityAccessManagement) doesPolicySignatureV2Match(ctx context.Cont
 		return nil, s3err.ErrSignatureDoesNotMatch
 	}
 	s3_constants.RecordAccessKeyId(ctx, cred.AccessKey)
+	s3_constants.RecordStoredAccessKeyId(ctx, cred.AccessKey, cred.SecretKey)
 	return identity, s3err.ErrNone
 }
 
@@ -168,6 +169,7 @@ func (iam *IdentityAccessManagement) doesSignV2Match(r *http.Request) (*Identity
 		return nil, s3err.ErrSignatureDoesNotMatch
 	}
 	s3_constants.RecordAccessKeyId(r.Context(), cred.AccessKey)
+	s3_constants.RecordStoredAccessKeyId(r.Context(), cred.AccessKey, cred.SecretKey)
 	return identity, s3err.ErrNone
 }
 
@@ -227,6 +229,7 @@ func (iam *IdentityAccessManagement) doesPresignV2SignatureMatch(r *http.Request
 		return nil, s3err.ErrSignatureDoesNotMatch
 	}
 	s3_constants.RecordAccessKeyId(r.Context(), cred.AccessKey)
+	s3_constants.RecordStoredAccessKeyId(r.Context(), cred.AccessKey, cred.SecretKey)
 	return identity, s3err.ErrNone
 }
 

weed/s3api/auth_signature_v4.go

@@ -412,6 +412,9 @@ func (iam *IdentityAccessManagement) verifyV4Signature(r *http.Request, shouldCh
 		// catalog's vended sessions name it (never a key re-read from the request).
 		// Recorded before the retarget check so that refusal names it too.
 		s3_constants.RecordAccessKeyId(r.Context(), cred.AccessKey)
+		if found {
+			s3_constants.RecordStoredAccessKeyId(r.Context(), cred.AccessKey, cred.SecretKey)
+		}
 		if signedBucket, ok := bucketFromVirtualHost(matchedHost, iam.domain); ok {
 			if routedBucket, _ := s3_constants.GetBucketAndObject(r); routedBucket != "" && routedBucket != signedBucket {
 				glog.V(2).Infof("reject %s %s: signed host %q implies bucket %q but routed to %q",
@@ -496,6 +499,19 @@ func (iam *IdentityAccessManagement) validateSTSSessionToken(r *http.Request, se
 		return nil, nil, s3err.ErrExpiredToken
 	}
 
+	// A session the table catalog vended for a static access key is bound to
+	// that key, as a role session is bound to its role's ID: deleting or
+	// deactivating the key, or deleting or disabling its identity, ends the
+	// session at its next request instead of at its expiry. Sessions minted
+	// before the binding existed carry only the audit field and stay unbound.
+	if sessionInfo.IdentityProvider == icebergVendedCredentialsProvider && sessionInfo.BoundAccessKeyId != "" {
+		if errCode := iam.vendedSessionKeyLive(sessionInfo.ExternalUserId, sessionInfo.BoundAccessKeyId, sessionInfo.BoundAccessKeyFingerprint); errCode != s3err.ErrNone {
+			glog.V(2).Infof("vended STS session %s: identity %q or the access key it was vended for is gone, disabled or inactive",
+				sessionInfo.SessionId, sessionInfo.ExternalUserId)
+			return nil, nil, errCode
+		}
+	}
+
 	// Validate required credential fields
 	if sessionInfo.Credentials.SecretAccessKey == "" {
 		glog.Warningf("STS service returned empty SecretAccessKey in credentials")
@@ -554,6 +570,38 @@ func (iam *IdentityAccessManagement) validateSTSSessionToken(r *http.Request, se
 	return identity, cred, s3err.ErrNone
 }
 
+// vendedSessionKeyLive answers ErrNone while owner is a known, enabled
+// identity holding accessKey as an active, unexpired credential, and
+// ErrInvalidAccessKeyID (as for an unknown key) otherwise. A fingerprint set
+// at mint time must still match, so a key re-created under the same id does
+// not revive sessions the deleted original had carried. One map lookup under
+// the read lock.
+func (iam *IdentityAccessManagement) vendedSessionKeyLive(owner, accessKey, fingerprint string) s3err.ErrorCode {
+	if owner == "" || accessKey == "" {
+		return s3err.ErrInvalidAccessKeyID
+	}
+	fingerprintKey := iam.stsSigningKey()
+	iam.m.RLock()
+	defer iam.m.RUnlock()
+	identity, ok := iam.nameToIdentity[owner]
+	if !ok || identity == nil || identity.Disabled {
+		return s3err.ErrInvalidAccessKeyID
+	}
+	for _, credential := range identity.Credentials {
+		if credential == nil || credential.AccessKey != accessKey {
+			continue
+		}
+		if credential.Status == iamAccessKeyStatusInactive || credential.isCredentialExpired() {
+			return s3err.ErrInvalidAccessKeyID
+		}
+		if fingerprint != "" && s3_constants.StoredCredentialFingerprint(fingerprintKey, accessKey, credential.SecretKey) != fingerprint {
+			return s3err.ErrInvalidAccessKeyID
+		}
+		return s3err.ErrNone
+	}
+	return s3err.ErrInvalidAccessKeyID
+}
+
 // calculateAndVerifySignature contains the core logic for creating the canonical request,
 // string-to-sign, and comparing the final signature.
 func calculateAndVerifySignature(secretKey, method, urlPath, queryStr string, extractedSignedHeaders http.Header, authInfo *v4AuthInfo) (string, s3err.ErrorCode) {
@@ -874,6 +922,7 @@ func (iam *IdentityAccessManagement) doesPolicySignatureV4Match(ctx context.Cont
 		return nil, s3err.ErrSignatureDoesNotMatch
 	}
 	s3_constants.RecordAccessKeyId(ctx, cred.AccessKey)
+	s3_constants.RecordStoredAccessKeyId(ctx, cred.AccessKey, cred.SecretKey)
 	return identity, s3err.ErrNone
 }
 

weed/s3api/iceberg/handlers_oauth.go

@@ -324,28 +324,28 @@ func mintIcebergToken(identityName, accessKey, secret string, ttlSeconds int) (s
 // authenticateBearer validates a Bearer token from the Authorization header.
 // Returns the identity name, identity object, and whether auth succeeded.
 func (s *Server) authenticateBearer(r *http.Request) (string, interface{}, bool) {
-	identityName, identity, _, ok := s.authenticateBearerKey(r)
+	identityName, identity, _, _, ok := s.authenticateBearerKey(r)
 	return identityName, identity, ok
 }
 
 // authenticateBearerKey is authenticateBearer that also returns the access key
 // the token was minted from (its verified access_key claim), so the catalog
 // can tell apart the holders of several keys on one identity.
-func (s *Server) authenticateBearerKey(r *http.Request) (string, interface{}, string, bool) {
+func (s *Server) authenticateBearerKey(r *http.Request) (string, interface{}, string, string, bool) {
 	auth := r.Header.Get("Authorization")
 	if auth == "" {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 	if !strings.HasPrefix(strings.ToLower(auth), "bearer ") {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 	tokenString := strings.TrimSpace(auth[7:])
 	if tokenString == "" {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	if s.credentialValidator == nil {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	// Parse the token without verification first to get the access key,
@@ -355,18 +355,18 @@ func (s *Server) authenticateBearerKey(r *http.Request) (string, interface{}, st
 	_, _, err := parser.ParseUnverified(tokenString, unverified)
 	if err != nil {
 		glog.V(2).Infof("Iceberg OAuth: failed to parse token: %v", err)
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	if unverified.AccessKey == "" {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	// Look up the credential by access key to get the signing key for verification
 	identityName, identity, secretKey, err := s.credentialValidator.GetCredentialByAccessKey(unverified.AccessKey)
 	if err != nil {
 		glog.V(2).Infof("Iceberg OAuth: failed to get credential for access key: %v", err)
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	signingKey := deriveSigningKey(unverified.AccessKey, secretKey)
@@ -379,12 +379,12 @@ func (s *Server) authenticateBearerKey(r *http.Request) (string, interface{}, st
 	})
 	if err != nil || !verified.Valid {
 		glog.V(2).Infof("Iceberg OAuth: token verification failed: %v", err)
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	// The token's signing key is derived from this access key, so a verified
 	// token was minted from it.
-	return identityName, identity, unverified.AccessKey, true
+	return identityName, identity, unverified.AccessKey, secretKey, true
 }
 
 // deriveSigningKey derives a signing key from the access key and secret using HMAC-SHA256.

weed/s3api/iceberg/server.go

@@ -269,10 +269,11 @@ func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc {
 		// The auth scheme is case-insensitive (RFC 7235), matching
 		// authenticateBearer below.
 		if strings.HasPrefix(strings.ToLower(r.Header.Get("Authorization")), "bearer ") {
-			if identityName, identity, accessKey, ok := s.authenticateBearerKey(r); ok {
+			if identityName, identity, accessKey, secretKey, ok := s.authenticateBearerKey(r); ok {
 				ctx := r.Context()
 				ctx = s3_constants.SetIdentityNameInContext(ctx, identityName)
 				ctx = s3_constants.SetAccessKeyIdInContext(ctx, accessKey)
+				ctx = s3_constants.SetStoredAccessKeyIdInContext(ctx, accessKey, secretKey)
 				if identity != nil {
 					ctx = s3_constants.SetIdentityInContext(ctx, identity)
 				}

weed/s3api/lance/handlers_oauth.go

@@ -147,27 +147,27 @@ func mintToken(identityName, accessKey, secret string, ttlSeconds int) (string,
 // authenticateBearer validates a Bearer token from the Authorization header.
 // Returns the identity name, identity object, and whether auth succeeded.
 func (s *Server) authenticateBearer(r *http.Request) (string, interface{}, bool) {
-	identityName, identity, _, ok := s.authenticateBearerKey(r)
+	identityName, identity, _, _, ok := s.authenticateBearerKey(r)
 	return identityName, identity, ok
 }
 
 // authenticateBearerKey is authenticateBearer that also returns the access key
 // the token was minted from (its verified access_key claim).
-func (s *Server) authenticateBearerKey(r *http.Request) (string, interface{}, string, bool) {
+func (s *Server) authenticateBearerKey(r *http.Request) (string, interface{}, string, string, bool) {
 	auth := r.Header.Get("Authorization")
 	if auth == "" {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 	if !strings.HasPrefix(strings.ToLower(auth), "bearer ") {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 	tokenString := strings.TrimSpace(auth[7:])
 	if tokenString == "" {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	if s.credentialValidator == nil {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	// Parse the token without verification first to get the access key,
@@ -177,17 +177,17 @@ func (s *Server) authenticateBearerKey(r *http.Request) (string, interface{}, st
 	_, _, err := parser.ParseUnverified(tokenString, unverified)
 	if err != nil {
 		glog.V(2).Infof("Lance OAuth: failed to parse token: %v", err)
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	if unverified.AccessKey == "" {
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	identityName, identity, secretKey, err := s.credentialValidator.GetCredentialByAccessKey(unverified.AccessKey)
 	if err != nil {
 		glog.V(2).Infof("Lance OAuth: failed to get credential for access key: %v", err)
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	signingKey := deriveSigningKey(unverified.AccessKey, secretKey)
@@ -200,12 +200,12 @@ func (s *Server) authenticateBearerKey(r *http.Request) (string, interface{}, st
 	})
 	if err != nil || !verified.Valid {
 		glog.V(2).Infof("Lance OAuth: token verification failed: %v", err)
-		return "", nil, "", false
+		return "", nil, "", "", false
 	}
 
 	// The token's signing key is derived from this access key, so a verified
 	// token was minted from it.
-	return identityName, identity, unverified.AccessKey, true
+	return identityName, identity, unverified.AccessKey, secretKey, true
 }
 
 // authenticateApiKey validates an x-api-key header carrying an S3 credential

weed/s3api/lance/server.go

@@ -177,10 +177,11 @@ func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc {
 		// different error that clients cannot act on.
 		// The auth scheme is case-insensitive (RFC 7235).
 		if strings.HasPrefix(strings.ToLower(r.Header.Get("Authorization")), "bearer ") {
-			if identityName, identity, accessKey, ok := s.authenticateBearerKey(r); ok {
+			if identityName, identity, accessKey, secretKey, ok := s.authenticateBearerKey(r); ok {
 				ctx := r.Context()
 				ctx = s3_constants.SetIdentityNameInContext(ctx, identityName)
 				ctx = s3_constants.SetAccessKeyIdInContext(ctx, accessKey)
+				ctx = s3_constants.SetStoredAccessKeyIdInContext(ctx, accessKey, secretKey)
 				if identity != nil {
 					ctx = s3_constants.SetIdentityInContext(ctx, identity)
 				}
@@ -195,10 +196,11 @@ func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc {
 
 		if apiKey := r.Header.Get("x-api-key"); apiKey != "" {
 			if identityName, identity, ok := s.authenticateApiKey(apiKey); ok {
-				accessKey, _, _ := strings.Cut(apiKey, ":")
+				accessKey, secretKey, _ := strings.Cut(apiKey, ":")
 				ctx := r.Context()
 				ctx = s3_constants.SetIdentityNameInContext(ctx, identityName)
 				ctx = s3_constants.SetAccessKeyIdInContext(ctx, accessKey)
+				ctx = s3_constants.SetStoredAccessKeyIdInContext(ctx, accessKey, secretKey)
 				if identity != nil {
 					ctx = s3_constants.SetIdentityInContext(ctx, identity)
 				}

weed/s3api/s3_constants/header.go

@@ -18,6 +18,9 @@ package s3_constants
 
 import (
 	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
 	"net/http"
 	"strings"
 	"sync/atomic"
@@ -355,6 +358,7 @@ const (
 	contextKeyIgnoredSessionToken contextKey = "s3-ignored-session-token"
 	contextKeyOriginalRequestPath contextKey = "s3-original-request-path"
 	contextKeyAccessKeyId         contextKey = "s3-access-key-id"
+	contextKeyStoredAccessKeyId   contextKey = "s3-stored-access-key-id"
 )
 
 // identityHolder is a mutable container for the authenticated identity name,
@@ -365,10 +369,37 @@ const (
 // holder installed before authentication is shared across all copies, so the
 // name written by the inner handler is readable by the outer middleware.
 type identityHolder struct {
-	name          atomic.Pointer[string]
-	principalArn  atomic.Pointer[string]
-	identityClaim atomic.Pointer[string]
-	accessKeyId   atomic.Pointer[string]
+	name             atomic.Pointer[string]
+	principalArn     atomic.Pointer[string]
+	identityClaim    atomic.Pointer[string]
+	accessKeyId      atomic.Pointer[string]
+	storedCredential atomic.Pointer[storedCredential]
+}
+
+// storedCredential is the stored-identity credential a request authenticated
+// with: the access key id and its secret, so a credential vendor can bind
+// what it mints to this exact credential instance.
+type storedCredential struct {
+	accessKeyId string
+	secretKey   string
+}
+
+// StoredCredentialFingerprint identifies one credential instance. The key id
+// alone cannot: an operator can delete a key and re-create it under the same
+// id with a different secret. The fingerprint travels inside the vended
+// session token, so it is keyed with a server-held key — an unkeyed hash of
+// the secret would hand the token holder an offline guessing oracle.
+func StoredCredentialFingerprint(fingerprintKey []byte, accessKeyId, secretKey string) string {
+	if len(fingerprintKey) == 0 {
+		return ""
+	}
+	h := hmac.New(sha256.New, fingerprintKey)
+	h.Write([]byte("seaweedfs-stored-credential"))
+	h.Write([]byte{0})
+	h.Write([]byte(accessKeyId))
+	h.Write([]byte{0})
+	h.Write([]byte(secretKey))
+	return hex.EncodeToString(h.Sum(nil)[:16])
 }
 
 // EnsureIdentityHolder attaches a mutable identity holder to the request context
@@ -554,6 +585,70 @@ func AccessKeyIdFromContext(ctx context.Context) string {
 	return ""
 }
 
+// SetStoredAccessKeyIdInContext records the stored-identity credential the
+// request authenticated with — set by an authenticator that verified the
+// credential against the identity store, so a credential vendor can tell the
+// caller apart from an STS session or OIDC user holding a key the store never
+// issued.
+func SetStoredAccessKeyIdInContext(ctx context.Context, accessKeyId, secretKey string) context.Context {
+	if accessKeyId == "" {
+		return ctx
+	}
+	sc := storedCredential{accessKeyId, secretKey}
+	if h, ok := ctx.Value(contextKeyIdentityHolder).(*identityHolder); ok && h != nil {
+		h.storedCredential.Store(&sc)
+	}
+	return context.WithValue(ctx, contextKeyStoredAccessKeyId, sc)
+}
+
+// RecordStoredAccessKeyId stores the stored-credential marker into the
+// request's identity holder: a verifier calls it when the credential that
+// matched resolves to a key held by a configured identity. A no-op without a
+// holder (EnsureIdentityHolder) or without a key.
+func RecordStoredAccessKeyId(ctx context.Context, accessKeyId, secretKey string) {
+	if ctx == nil || accessKeyId == "" {
+		return
+	}
+	sc := storedCredential{accessKeyId, secretKey}
+	if h, ok := ctx.Value(contextKeyIdentityHolder).(*identityHolder); ok && h != nil {
+		h.storedCredential.Store(&sc)
+	}
+}
+
+// StoredAccessKeyIdFromContext retrieves the stored-identity access key id the
+// request authenticated with, or "" when it authenticated another way (STS
+// session, OIDC bearer, anonymous).
+func StoredAccessKeyIdFromContext(ctx context.Context) string {
+	sc, _ := storedCredentialFromContext(ctx)
+	return sc.accessKeyId
+}
+
+// StoredCredentialFromContext retrieves the stored-identity access key id and
+// secret the request authenticated with, or "", "" when it authenticated
+// another way (STS session, OIDC bearer, anonymous).
+func StoredCredentialFromContext(ctx context.Context) (string, string) {
+	sc, ok := storedCredentialFromContext(ctx)
+	if !ok {
+		return "", ""
+	}
+	return sc.accessKeyId, sc.secretKey
+}
+
+func storedCredentialFromContext(ctx context.Context) (storedCredential, bool) {
+	if ctx == nil {
+		return storedCredential{}, false
+	}
+	if sc, ok := ctx.Value(contextKeyStoredAccessKeyId).(storedCredential); ok && sc.accessKeyId != "" {
+		return sc, true
+	}
+	if h, ok := ctx.Value(contextKeyIdentityHolder).(*identityHolder); ok && h != nil {
+		if sc := h.storedCredential.Load(); sc != nil && sc.accessKeyId != "" {
+			return *sc, true
+		}
+	}
+	return storedCredential{}, false
+}
+
 // IgnoreSessionTokenInContext marks the request's session token as tolerated
 // rather than authenticated: a statically configured access key signed the
 // request, so the token (which credential vendors like Unity Catalog emit on

weed/s3api/s3_constants/header_test.go

@@ -313,3 +313,20 @@ func TestGetIdentityNameFromContextWithoutHolder(t *testing.T) {
 		t.Errorf("requester = %q, want admin", got)
 	}
 }
+
+func TestStoredCredentialFingerprintIsKeyed(t *testing.T) {
+	key := []byte("test-signing-key-32-characters!!")
+	fp := StoredCredentialFingerprint(key, "AKIAEXAMPLE", "secret")
+	if fp == "" {
+		t.Fatal("fingerprint is empty with a key")
+	}
+	if again := StoredCredentialFingerprint(key, "AKIAEXAMPLE", "secret"); again != fp {
+		t.Error("fingerprint is not deterministic")
+	}
+	if other := StoredCredentialFingerprint([]byte("another-signing-key-padded-out"), "AKIAEXAMPLE", "secret"); other == fp {
+		t.Error("fingerprint ignores the server key, leaving the secret guessable")
+	}
+	if noKey := StoredCredentialFingerprint(nil, "AKIAEXAMPLE", "secret"); noKey != "" {
+		t.Error("an empty key must yield an empty fingerprint")
+	}
+}

weed/s3api/s3api_iceberg_credentials.go

@@ -40,8 +40,11 @@ type VendedCredentials struct {
 //
 // The access key the caller authenticated with (from ctx) is recorded in the
 // session: a short hash of it ends the session name, so the audit log's
-// requester_arn tells apart the holders of several keys on one identity, and
-// the key id itself is carried in the session token.
+// requester_arn tells apart the holders of several keys on one identity. When
+// the caller is a stored identity holding that key, the key id is also carried
+// in the session token, binding the session to it: the session ends when the
+// key is deleted or deactivated, or its identity removed or disabled
+// (validateSTSSessionToken).
 func (s3a *S3ApiServer) VendTableCredentials(ctx context.Context, principal, bucket, prefix string, readOnly bool) (*VendedCredentials, error) {
 	if s3a.icebergCredentialRole == "" {
 		return nil, nil
@@ -72,6 +75,19 @@ func (s3a *S3ApiServer) VendTableCredentials(ctx context.Context, principal, buc
 		Policy:            &policy,
 		SourceAccessKeyId: accessKey,
 	}
+	// Bind the session to the caller's key when the caller authenticated with a
+	// stored identity's key — a marker its verifier records, so a caller that
+	// signed with an STS session key or OIDC token stays unbound even when its
+	// principal names a stored identity, whose session a binding would refuse
+	// on every request. The marker survives the key being revoked between
+	// authentication and this vend: the session then dies on its first request
+	// instead of living unbound until it expires. The binding travels under
+	// its own claim — SourceAccessKeyId is audit-only for compatibility with
+	// tokens minted before the binding existed.
+	if storedKey, storedSecret := s3_constants.StoredCredentialFromContext(ctx); storedKey != "" {
+		request.BoundAccessKeyId = storedKey
+		request.BoundAccessKeyFingerprint = s3_constants.StoredCredentialFingerprint(s3a.iam.stsSigningKey(), storedKey, storedSecret)
+	}
 	if s3a.icebergCredentialDuration > 0 {
 		duration := s3a.icebergCredentialDuration
 		request.DurationSeconds = &duration
@@ -94,10 +110,17 @@ func (s3a *S3ApiServer) VendTableCredentials(ctx context.Context, principal, buc
 }
 
 func (s3a *S3ApiServer) stsService() *sts.STSService {
-	if s3a.iam == nil || s3a.iam.iamIntegration == nil {
+	if s3a.iam == nil {
 		return nil
 	}
-	provider, ok := s3a.iam.iamIntegration.(IAMManagerProvider)
+	return s3a.iam.stsService()
+}
+
+func (iam *IdentityAccessManagement) stsService() *sts.STSService {
+	if iam.iamIntegration == nil {
+		return nil
+	}
+	provider, ok := iam.iamIntegration.(IAMManagerProvider)
 	if !ok {
 		return nil
 	}
@@ -108,6 +131,13 @@ func (s3a *S3ApiServer) stsService() *sts.STSService {
 	return manager.GetSTSService()
 }
 
+func (iam *IdentityAccessManagement) stsSigningKey() []byte {
+	if s := iam.stsService(); s != nil && s.Config != nil {
+		return s.Config.SigningKey
+	}
+	return nil
+}
+
 // tablePrefixSessionPolicy narrows a session to one table's files plus the
 // bucket listing that clients need to resolve them. A read-only session drops
 // the write verbs: the catalog vends no more authority than the caller holds.

weed/s3api/s3api_iceberg_session_name_test.go

@@ -76,12 +76,14 @@ func TestVendTableCredentialsRecordsTheCallersKey(t *testing.T) {
 	s3a := &S3ApiServer{iam: iam}
 	s3a.SetIcebergCredentialRole("arn:aws:iam::role/IcebergTableAccess", 3600)
 
-	for _, key := range []string{sharedKeyA, sharedKeyB, ""} {
+	for _, k := range []struct{ id, secret string }{{sharedKeyA, sharedSecretA}, {sharedKeyB, sharedSecretB}, {"", ""}} {
+		key := k.id
 		ctx := context.Background()
 		if key != "" {
 			// As the catalog's auth does: the holder is installed, the verifier records into it.
 			r := s3_constants.EnsureIdentityHolder(httptest.NewRequest(http.MethodGet, "/", nil))
 			s3_constants.RecordAccessKeyId(r.Context(), key)
+			s3_constants.RecordStoredAccessKeyId(r.Context(), k.id, k.secret)
 			ctx = r.Context()
 		}
 		creds, err := s3a.VendTableCredentials(ctx, "shared-user", "warehouse", "ns/t", false)

weed/s3api/s3api_iceberg_vended_session_test.go

@@ -0,0 +1,234 @@
+package s3api
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/seaweedfs/seaweedfs/weed/iam/integration"
+	"github.com/seaweedfs/seaweedfs/weed/iam/policy"
+	"github.com/seaweedfs/seaweedfs/weed/iam/sts"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+)
+
+// vendingServer is an S3ApiServer whose catalog role trusts the vendor, with
+// the identity of newSharedIdentityIAM (one identity, two keys).
+func vendingServer(t *testing.T) *S3ApiServer {
+	t.Helper()
+	mgr := newTestIAMManager(t)
+	require.NoError(t, mgr.CreateRole(context.Background(), "", "IcebergTableAccess", &integration.RoleDefinition{
+		RoleName: "IcebergTableAccess",
+		TrustPolicy: &policy.PolicyDocument{
+			Version: "2012-10-17",
+			Statement: []policy.Statement{{
+				Effect:    "Allow",
+				Principal: map[string]interface{}{"Federated": icebergVendedCredentialsProvider},
+				Action:    []string{"sts:AssumeRoleWithCredentials"},
+			}},
+		},
+	}))
+	iam := newSharedIdentityIAM()
+	iam.SetIAMIntegration(NewS3IAMIntegration(mgr, ""))
+	s3a := &S3ApiServer{iam: iam}
+	s3a.SetIcebergCredentialRole("arn:aws:iam::role/IcebergTableAccess", 3600)
+	return s3a
+}
+
+// vend mints table credentials for principal as if its request had been
+// signed with accessKey ("" for a request that recorded none); a non-empty
+// secret marks the credential as a stored identity's, as a verifier would.
+func vend(t *testing.T, s3a *S3ApiServer, principal, accessKey, secret string) *VendedCredentials {
+	t.Helper()
+	ctx := context.Background()
+	if accessKey != "" {
+		ctx = s3_constants.SetAccessKeyIdInContext(ctx, accessKey)
+	}
+	if secret != "" {
+		ctx = s3_constants.SetStoredAccessKeyIdInContext(ctx, accessKey, secret)
+	}
+	creds, err := s3a.VendTableCredentials(ctx, principal, "warehouse", "ns/t", false)
+	require.NoError(t, err)
+	require.NotNil(t, creds)
+	return creds
+}
+
+// use validates a vended session the way a data-path request carrying it is
+// validated.
+func use(t *testing.T, s3a *S3ApiServer, creds *VendedCredentials) s3err.ErrorCode {
+	t.Helper()
+	r, err := newTestRequest("GET", "https://example.com/warehouse/ns/t/data/f.parquet", 0, nil)
+	require.NoError(t, err)
+	_, _, errCode := s3a.iam.validateSTSSessionToken(r, creds.SessionToken, creds.AccessKeyID)
+	return errCode
+}
+
+func sessionInfo(t *testing.T, s3a *S3ApiServer, creds *VendedCredentials) *sts.SessionInfo {
+	t.Helper()
+	info, err := s3a.stsService().ValidateSessionToken(context.Background(), creds.SessionToken)
+	require.NoError(t, err)
+	return info
+}
+
+func TestVendedSessionEndsWithItsAccessKey(t *testing.T) {
+	s3a := vendingServer(t)
+	sessionA := vend(t, s3a, "shared-user", sharedKeyA, sharedSecretA)
+	sessionB := vend(t, s3a, "shared-user", sharedKeyB, sharedSecretB)
+	require.Equal(t, s3err.ErrNone, use(t, s3a, sessionA), "control: the session works while its key does")
+	require.Equal(t, s3err.ErrNone, use(t, s3a, sessionB))
+
+	identity := s3a.iam.nameToIdentity["shared-user"]
+	identity.Credentials[0].Status = iamAccessKeyStatusInactive
+	assert.Equal(t, s3err.ErrInvalidAccessKeyID, use(t, s3a, sessionA), "key A deactivated")
+	assert.Equal(t, s3err.ErrNone, use(t, s3a, sessionB), "the session vended through key B keeps working")
+
+	identity.Credentials[0].Status = iamAccessKeyStatusActive
+	assert.Equal(t, s3err.ErrNone, use(t, s3a, sessionA), "key A reactivated")
+
+	identity.Credentials[0].Expiration = time.Now().Add(-time.Minute).Unix()
+	assert.Equal(t, s3err.ErrInvalidAccessKeyID, use(t, s3a, sessionA), "key A expired")
+	identity.Credentials[0].Expiration = 0
+
+	identity.Credentials = identity.Credentials[1:]
+	assert.Equal(t, s3err.ErrInvalidAccessKeyID, use(t, s3a, sessionA), "key A deleted")
+	assert.Equal(t, s3err.ErrNone, use(t, s3a, sessionB), "the session vended through key B keeps working")
+}
+
+func TestVendedSessionEndsWithItsIdentity(t *testing.T) {
+	s3a := vendingServer(t)
+	session := vend(t, s3a, "shared-user", sharedKeyA, sharedSecretA)
+	require.Equal(t, s3err.ErrNone, use(t, s3a, session))
+
+	s3a.iam.nameToIdentity["shared-user"].Disabled = true
+	assert.Equal(t, s3err.ErrInvalidAccessKeyID, use(t, s3a, session), "identity disabled")
+	s3a.iam.nameToIdentity["shared-user"].Disabled = false
+	assert.Equal(t, s3err.ErrNone, use(t, s3a, session), "identity enabled again")
+
+	delete(s3a.iam.nameToIdentity, "shared-user")
+	assert.Equal(t, s3err.ErrInvalidAccessKeyID, use(t, s3a, session), "identity deleted")
+}
+
+// A caller that is not a stored identity (a caller signing with another STS
+// session's key, an OIDC user, a request that recorded no key) still gets
+// credentials, unbound as before: binding them would refuse every request
+// made with them.
+func TestVendedSessionIsBoundOnlyForAStoredIdentitysKey(t *testing.T) {
+	s3a := vendingServer(t)
+	control := sessionInfo(t, s3a, vend(t, s3a, "shared-user", sharedKeyA, sharedSecretA))
+	assert.Equal(t, sharedKeyA, control.SourceAccessKeyId, "control")
+	assert.Equal(t, sharedKeyA, control.BoundAccessKeyId, "control")
+
+	for _, tc := range []struct{ principal, key, why string }{
+		{"47ad4828c45b3f337bc3146081ba8f0f", "ASIATEMPORARYKEY0001", "an STS session's subject and temporary key"},
+		{"shared-user", "ASIATEMPORARYKEY0001", "an STS session whose subject names a stored identity"},
+		{"alice@example.com", "", "an OIDC user"},
+		{"shared-user", "", "no recorded key"},
+	} {
+		session := vend(t, s3a, tc.principal, tc.key, "")
+		info := sessionInfo(t, s3a, session)
+		assert.Equal(t, tc.key, info.SourceAccessKeyId, tc.why)
+		assert.Empty(t, info.BoundAccessKeyId, tc.why)
+		assert.Equal(t, s3err.ErrNone, use(t, s3a, session), tc.why)
+	}
+}
+
+// Tokens minted before the binding existed record the caller's key under the
+// audit claim only. Validating them as bound would end live table credentials
+// at upgrade — a chained STS caller's key can never resolve to a stored
+// credential. They stay valid until expiry, as they always did.
+func TestLegacyVendedSessionStaysUnbound(t *testing.T) {
+	iam := newSharedIdentityIAM()
+	iam.iamIntegration = &MockIAMIntegration{
+		validateSessionFunc: func(ctx context.Context, token string) (*sts.SessionInfo, error) {
+			return &sts.SessionInfo{
+				SessionId:         "sid",
+				AssumedRoleUser:   "arn:aws:sts::assumed-role/Reader/session",
+				Principal:         "arn:aws:sts::assumed-role/Reader/session",
+				IdentityProvider:  icebergVendedCredentialsProvider,
+				ExternalUserId:    "someone-not-in-the-identity-table",
+				SourceAccessKeyId: "ASIATEMPORARYKEY0001",
+				Credentials:       &sts.Credentials{AccessKeyId: "ASIALEGACY", SecretAccessKey: "secret"},
+				ExpiresAt:         time.Now().Add(time.Hour),
+			}, nil
+		},
+	}
+	r, err := newTestRequest("GET", "https://example.com/bucket/f.parquet", 0, nil)
+	require.NoError(t, err)
+	_, _, errCode := iam.validateSTSSessionToken(r, "token", "ASIALEGACY")
+	assert.Equal(t, s3err.ErrNone, errCode)
+}
+
+// A key re-created under the same id with a different secret does not revive
+// sessions bound to the deleted original: the fingerprint recorded at mint
+// time no longer matches.
+func TestRecreatedKeyDoesNotReviveVendedSession(t *testing.T) {
+	s3a := vendingServer(t)
+	session := vend(t, s3a, "shared-user", sharedKeyA, sharedSecretA)
+	require.Equal(t, s3err.ErrNone, use(t, s3a, session))
+
+	identity := s3a.iam.nameToIdentity["shared-user"]
+	identity.Credentials[0] = &Credential{AccessKey: sharedKeyA, SecretKey: "recreated-secret"}
+	assert.Equal(t, s3err.ErrInvalidAccessKeyID, use(t, s3a, session), "same id, different secret")
+
+	identity.Credentials[0] = &Credential{AccessKey: sharedKeyA, SecretKey: sharedSecretA}
+	assert.Equal(t, s3err.ErrNone, use(t, s3a, session), "the identical credential is indistinguishable")
+}
+
+// A session for a stored identity stays bound even when the key no longer
+// resolves at vend time — revoked between the request's authentication and
+// the vend, or held by a different identity now. The session dies on its
+// first request instead of living unbound until it expires.
+func TestVendedSessionStaysBoundWhenTheKeyIsGoneAtVend(t *testing.T) {
+	s3a := vendingServer(t)
+
+	identity := s3a.iam.nameToIdentity["shared-user"]
+	identity.Credentials = identity.Credentials[1:] // sharedKeyA revoked mid-request
+	session := vend(t, s3a, "shared-user", sharedKeyA, sharedSecretA)
+	assert.Equal(t, sharedKeyA, sessionInfo(t, s3a, session).SourceAccessKeyId, "still bound")
+	assert.Equal(t, s3err.ErrInvalidAccessKeyID, use(t, s3a, session), "the dead key's session is refused")
+}
+
+// Only sessions the catalog vended are bound; any other STS session (web
+// identity, AssumeRole) is validated as before.
+func TestOtherSessionsAreNotBoundToAKey(t *testing.T) {
+	iam := newSharedIdentityIAM()
+	iam.iamIntegration = &MockIAMIntegration{
+		validateSessionFunc: func(ctx context.Context, token string) (*sts.SessionInfo, error) {
+			return &sts.SessionInfo{
+				SessionId:         "sid",
+				AssumedRoleUser:   "arn:aws:sts::assumed-role/Reader/session",
+				Principal:         "arn:aws:sts::assumed-role/Reader/session",
+				IdentityProvider:  "oidc-provider",
+				ExternalUserId:    "someone-not-in-the-identity-table",
+				SourceAccessKeyId: "AKIAGONE",
+				Credentials:       &sts.Credentials{AccessKeyId: "ASIAWEBID", SecretAccessKey: "secret"},
+				ExpiresAt:         time.Now().Add(time.Hour),
+			}, nil
+		},
+	}
+	r, err := newTestRequest("GET", "https://example.com/bucket/f.parquet", 0, nil)
+	require.NoError(t, err)
+	_, _, errCode := iam.validateSTSSessionToken(r, "token", "ASIAWEBID")
+	assert.Equal(t, s3err.ErrNone, errCode)
+}
+
+// The check on the request path is one read-locked map lookup and a scan of
+// one identity's keys.
+func BenchmarkVendedSessionKeyLive(b *testing.B) {
+	iam := newSharedIdentityIAM()
+	for i := 0; i < 1000; i++ {
+		name := fmt.Sprintf("user-%04d", i)
+		iam.nameToIdentity[name] = &Identity{Name: name, Credentials: []*Credential{{AccessKey: name}}}
+	}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if iam.vendedSessionKeyLive("shared-user", sharedKeyB, "") != s3err.ErrNone {
+			b.Fatal("refused")
+		}
+	}
+}