s3tables: keep legacy-subject owners and policy principals for federated STS sessions (#12615)

* s3tables: re-anchor legacy owners for federated STS sessions

A web-identity role session was keyed on its bare OIDC subject until
owners became issuer-scoped. Its sessions now carry the scoped aws:userid
but no asserted issuers, so the legacy subject check never applied and the
session lost the table buckets and namespaces it had created. Read the
configured issuer set for such sessions.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: keep token claims from asserting trusted issuers on STS sessions

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* iceberg: give the catalog's table handler the local accounts and issuer set

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3tables: test a federated session on a bucket its legacy subject owns

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* sts: drop a token-supplied aws:userid when a session has no parent user

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: take a session's aws:userid from its signed parent user

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: test that session claims keep only the server-asserted aws:userid

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3tables: match legacy subject patterns in table bucket policy principals

A policy written before issuer scoping can name role sessions by a pattern
over their bare OIDC subjects. Match such a pattern against the caller's
subject claim once that subject resolves to the session's scoped principal
under the single configured issuer.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3tables: test a federated session against a legacy subject pattern principal

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/iam/sts/sts_service.go

@@ -699,6 +699,8 @@ func (s *STSService) AssumeRoleWithWebIdentity(ctx context.Context, request *Ass
 		// Surface as aws:userid so policies can reference it directly without
 		// caring about token-rotation churn.
 		requestContext["aws:userid"] = parentUser
+	} else {
+		delete(requestContext, "aws:userid")
 	}
 
 	// Surface principal session tags as aws:PrincipalTag/<key>. The OIDC

weed/s3api/auth_signature_v4.go

@@ -530,6 +530,8 @@ func (iam *IdentityAccessManagement) validateSTSSessionToken(r *http.Request, se
 	for k, v := range sessionInfo.RequestContext {
 		claims[k] = v
 	}
+	// Token attributes reach the request context; only the bearer path asserts trusted issuers.
+	delete(claims, "seaweed:oidc-issuers")
 	// The session's user is the one it was issued to, never a claim the
 	// identity provider forwarded.
 	if sessionInfo.ExternalUserId != "" {
@@ -537,6 +539,11 @@ func (iam *IdentityAccessManagement) validateSTSSessionToken(r *http.Request, se
 	} else {
 		delete(claims, "aws:username")
 	}
+	if sessionInfo.ParentUser != "" {
+		claims["aws:userid"] = sessionInfo.ParentUser
+	} else {
+		delete(claims, "aws:userid")
+	}
 
 	// Key the principal on the OIDC subject so a session resolves to the same
 	// identity on the SigV4 and JWT paths (see s3_iam_middleware.go); fall back to

weed/s3api/auth_sts_identity_test.go

@@ -394,3 +394,41 @@ func newTestIdentity(policyNames []string) *Identity {
 		PolicyNames:  policyNames,
 	}
 }
+
+func TestSTSSessionClaimsKeepOnlyServerAssertedUserID(t *testing.T) {
+	stsService, config := setupTestSTSService(t)
+	iam := NewIdentityAccessManagementWithStore(&S3ApiServerOption{}, nil, "memory")
+	iam.SetIAMIntegration(&S3IAMIntegration{stsService: stsService})
+	victim := sts.ComputeParentUser("alice", "https://idp.example")
+	attacker := sts.ComputeParentUser("mallory", "https://idp.example")
+
+	for name, tc := range map[string]struct{ parentUser, want string }{
+		"no parent user": {"", ""},
+		"parent user":    {attacker, attacker},
+	} {
+		t.Run(name, func(t *testing.T) {
+			claims := sts.NewSTSSessionClaims("userid-"+strings.ReplaceAll(name, " ", "-"), config.Issuer, time.Now().Add(time.Hour)).
+				WithSessionName("userid").
+				WithRoleInfo("arn:aws:iam::role/testRole", "arn:aws:sts::assumed-role/testRole/userid", "arn:aws:sts::assumed-role/testRole/userid").
+				WithRequestContext(map[string]interface{}{"aws:userid": victim, "seaweed:oidc-issuers": "https://idp.example"})
+			if tc.parentUser != "" {
+				claims.WithParentUser(tc.parentUser)
+			}
+			token, err := sts.NewTokenGenerator(config.SigningKey, config.Issuer).GenerateJWTWithClaims(claims)
+			require.NoError(t, err)
+			sessionInfo, err := stsService.ValidateSessionToken(context.Background(), token)
+			require.NoError(t, err)
+			req, err := http.NewRequest(http.MethodGet, "/lake/key", nil)
+			require.NoError(t, err)
+
+			identity, _, errCode := iam.validateSTSSessionToken(req, token, sessionInfo.Credentials.AccessKeyId)
+			require.Equal(t, s3err.ErrNone, errCode)
+			_, hasUserID := identity.Claims["aws:userid"]
+			assert.Equal(t, tc.want != "", hasUserID)
+			if tc.want != "" {
+				assert.Equal(t, tc.want, identity.Claims["aws:userid"])
+			}
+			assert.NotContains(t, identity.Claims, "seaweed:oidc-issuers")
+		})
+	}
+}

weed/s3api/iceberg/server.go

@@ -73,6 +73,14 @@ func NewServer(filerClient FilerClient, authenticator S3Authenticator) *Server {
 	if authenticator != nil {
 		manager.SetDefaultAllow(authenticator.DefaultAllow())
 	}
+	if checker, ok := authenticator.(interface{ IsConfiguredAccountID(string) bool }); ok {
+		manager.SetLocalAccountChecker(checker.IsConfiguredAccountID)
+	}
+	if source, ok := authenticator.(interface {
+		TrustedOIDCIssuers(context.Context) []string
+	}); ok {
+		manager.SetTrustedIssuerSource(source.TrustedOIDCIssuers)
+	}
 	return &Server{
 		filerClient:   filerClient,
 		tablesManager: manager,

weed/s3api/s3api_server.go

@@ -1558,6 +1558,22 @@ func (s3a *S3ApiServer) DefaultAllow() bool {
 	return s3a.iam.iamIntegration.DefaultAllow()
 }
 
+// IsConfiguredAccountID reports whether id is the account of a configured identity.
+func (s3a *S3ApiServer) IsConfiguredAccountID(id string) bool {
+	return s3a.iam != nil && s3a.iam.isConfiguredAccountID(id)
+}
+
+// TrustedOIDCIssuers returns the configured OIDC issuer set.
+func (s3a *S3ApiServer) TrustedOIDCIssuers(ctx context.Context) []string {
+	if s3a.iam == nil {
+		return nil
+	}
+	if s3iam, ok := s3a.iam.iamIntegration.(*S3IAMIntegration); ok {
+		return s3iam.trustedOIDCIssuers(ctx)
+	}
+	return nil
+}
+
 // ValidateS3Credential validates an S3 access key / secret key pair.
 // Returns the identity name and identity object on success.
 func (s3a *S3ApiServer) ValidateS3Credential(accessKey, secretKey string) (string, interface{}, error) {

weed/s3api/s3api_tables.go

@@ -2,6 +2,7 @@ package s3api
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -63,6 +64,12 @@ func (st *S3TablesApiServer) SetEncryptionChecker(fn func() bool) {
 	st.handler.SetEncryptionChecker(fn)
 }
 
+// SetTrustedIssuerSource injects the configured OIDC issuer set for S3 Tables
+// legacy-owner checks of federated STS sessions.
+func (st *S3TablesApiServer) SetTrustedIssuerSource(fn func(context.Context) []string) {
+	st.handler.SetTrustedIssuerSource(fn)
+}
+
 // S3TablesHandler handles S3 Tables API requests
 func (st *S3TablesApiServer) S3TablesHandler(w http.ResponseWriter, r *http.Request) {
 	st.handler.HandleRequest(w, r, st)
@@ -85,6 +92,7 @@ func (s3a *S3ApiServer) registerS3TablesRoutes(router *mux.Router) {
 		s3TablesApi.SetDefaultAllow(s3a.iam.iamIntegration.DefaultAllow())
 		if s3Integration, ok := s3a.iam.iamIntegration.(*S3IAMIntegration); ok && s3Integration.iamManager != nil {
 			s3TablesApi.SetIAMAuthorizer(s3Integration.iamManager)
+			s3TablesApi.SetTrustedIssuerSource(s3Integration.trustedOIDCIssuers)
 		}
 	} else {
 		// If IAM is not configured, allow all access by default

weed/s3api/s3tables/handler.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/iam/sts"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 )
@@ -79,6 +80,7 @@ type S3TablesHandler struct {
 	trusted         bool // Trusted local tooling (shell/admin) bypasses authorization
 	iamAuthorizer   IAMAuthorizer
 	localAccountIDs func(string) bool
+	oidcIssuers     func(context.Context) []string
 	canEncrypt      func() bool
 }
 
@@ -123,6 +125,12 @@ func (h *S3TablesHandler) SetLocalAccountChecker(fn func(string) bool) {
 	h.localAccountIDs = fn
 }
 
+// SetTrustedIssuerSource injects the configured OIDC issuer set. A federated
+// STS session carries no asserted issuers, so its legacy owner checks read it.
+func (h *S3TablesHandler) SetTrustedIssuerSource(fn func(context.Context) []string) {
+	h.oidcIssuers = fn
+}
+
 // SetEncryptionChecker injects the predicate that tells whether the gateway
 // holds an SSE-S3 key, so a bucket whose declared AES256 the write path
 // applies is not created where every write would then fail.
@@ -390,9 +398,26 @@ func claimString(claims map[string]interface{}, key string) string {
 	return str
 }
 
-// trustedIssuers returns the OIDC issuers the auth layer asserted into the
-// identity's claims; they re-anchor legacy owner checks in checkPermission.
+// trustedIssuers returns the OIDC issuers that re-anchor legacy owner checks in
+// checkPermission: those the auth layer asserted into a bearer token's claims,
+// else, for a federated STS session, the configured set.
 func (h *S3TablesHandler) trustedIssuers(r *http.Request) []string {
+	claims := requestClaims(r)
+	if _, asserted := claims["seaweed:oidc-issuers"]; asserted {
+		return strings.Fields(claimString(claims, "seaweed:oidc-issuers"))
+	}
+	if h.oidcIssuers != nil && sts.IsParentUserID(claimString(claims, "aws:userid")) {
+		return h.oidcIssuers(r.Context())
+	}
+	return nil
+}
+
+// callerSubject returns the OIDC subject in the request identity's claims.
+func callerSubject(r *http.Request) string {
+	return claimString(requestClaims(r), "sub")
+}
+
+func requestClaims(r *http.Request) map[string]interface{} {
 	identityRaw := s3_constants.GetIdentityFromContext(r)
 	if identityRaw == nil {
 		return nil
@@ -404,7 +429,7 @@ func (h *S3TablesHandler) trustedIssuers(r *http.Request) []string {
 	if val.Kind() != reflect.Struct {
 		return nil
 	}
-	return strings.Fields(claimString(identityClaimsMap(val), "seaweed:oidc-issuers"))
+	return identityClaimsMap(val)
 }
 
 // normalizePrincipalID collapses ARN and identity strings to a key that is stable within a single account.
@@ -580,6 +605,7 @@ func (h *S3TablesHandler) authorizeAbsentTarget(r *http.Request, filerClient Fil
 	}
 	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespace,

weed/s3api/s3tables/handler_bucket_create.go

@@ -71,6 +71,7 @@ func (h *S3TablesHandler) handleCreateTableBucket(w http.ResponseWriter, r *http
 		}
 		if !CheckPermissionWithContext("CreateTableBucket", principal, owner, "", "", &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			IdentityActions:  identityActions,
 			DefaultAllow:     h.defaultAllowFor(r),

weed/s3api/s3tables/handler_bucket_get_list_delete.go

@@ -72,6 +72,7 @@ func (h *S3TablesHandler) handleGetTableBucket(w http.ResponseWriter, r *http.Re
 	identityActions := getIdentityActions(r)
 	if !CheckPermissionWithContext("GetTableBucket", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		IdentityActions:  identityActions,
@@ -197,6 +198,7 @@ func (h *S3TablesHandler) handleListTableBuckets(w http.ResponseWriter, r *http.
 				identityActions := getIdentityActions(r)
 				if !CheckPermissionWithContext("GetTableBucket", accountID, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 					TrustedIssuers:   h.trustedIssuers(r),
+					Subject:          callerSubject(r),
 					IsLocalAccountID: h.localAccountIDs,
 					TableBucketName:  entry.Entry.Name,
 					IdentityActions:  identityActions,
@@ -332,6 +334,7 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http
 		identityActions := getIdentityActions(r)
 		if !CheckPermissionWithContext("DeleteTableBucket", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
 			IdentityActions:  identityActions,

weed/s3api/s3tables/handler_maintenance.go

@@ -361,6 +361,7 @@ func (h *S3TablesHandler) authorizeMaintenanceBucket(r *http.Request, filerClien
 	principal := h.getAccountID(r)
 	if !CheckPermissionWithContext(operation, principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		IdentityActions:  getIdentityActions(r),
@@ -410,6 +411,7 @@ func (h *S3TablesHandler) authorizeMaintenanceTable(r *http.Request, filerClient
 	principal := h.getAccountID(r)
 	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,

weed/s3api/s3tables/handler_namespace.go

@@ -162,6 +162,7 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
 	glog.Infof("S3Tables: CreateNamespace permission check - principal=%s, owner=%s, actions=%v", principal, bucketMetadata.OwnerAccountID, identityActions)
 	if !CheckPermissionWithContext("CreateNamespace", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
@@ -388,6 +389,7 @@ func (h *S3TablesHandler) handleGetNamespace(w http.ResponseWriter, r *http.Requ
 	identityActions := getIdentityActions(r)
 	if !CheckPermissionWithContext("GetNamespace", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
@@ -480,6 +482,7 @@ func (h *S3TablesHandler) handleUpdateNamespace(w http.ResponseWriter, r *http.R
 	identityActions := getIdentityActions(r)
 	if !CheckPermissionWithContext("UpdateNamespace", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
@@ -600,6 +603,7 @@ func (h *S3TablesHandler) handleListNamespaces(w http.ResponseWriter, r *http.Re
 	identityActions := getIdentityActions(r)
 	if !CheckPermissionWithContext("ListNamespaces", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		TableBucketTags:  bucketTags,
@@ -798,6 +802,7 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R
 	identityActions := getIdentityActions(r)
 	if !CheckPermissionWithContext("DeleteNamespace", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,

weed/s3api/s3tables/handler_policy.go

@@ -167,6 +167,7 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h
 	identityActions := getIdentityActions(r)
 	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		TableBucketTags:  bucketTags,
@@ -267,6 +268,7 @@ func (h *S3TablesHandler) handleGetTableBucketPolicy(w http.ResponseWriter, r *h
 	identityActions := getIdentityActions(r)
 	if !CheckPermissionWithContext("GetTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, string(policy), bucketARN, &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		IdentityActions:  identityActions,
@@ -345,6 +347,7 @@ func (h *S3TablesHandler) handleDeleteTableBucketPolicy(w http.ResponseWriter, r
 	identityActions := getIdentityActions(r)
 	if !CheckPermissionWithContext("DeleteTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		IdentityActions:  identityActions,
@@ -466,6 +469,7 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re
 	identityActions := getIdentityActions(r)
 	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
@@ -591,6 +595,7 @@ func (h *S3TablesHandler) handleGetTablePolicy(w http.ResponseWriter, r *http.Re
 	identityActions := getIdentityActions(r)
 	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
@@ -687,6 +692,7 @@ func (h *S3TablesHandler) handleDeleteTablePolicy(w http.ResponseWriter, r *http
 	identityActions := getIdentityActions(r)
 	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
@@ -825,6 +831,7 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
 		identityActions := getIdentityActions(r)
 		ctx := &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
 			Namespace:        tableNamespace,
@@ -988,6 +995,7 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht
 		identityActions := getIdentityActions(r)
 		ctx := &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
 			Namespace:        tableNamespace,
@@ -1117,6 +1125,7 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
 		identityActions := getIdentityActions(r)
 		ctx := &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
 			Namespace:        tableNamespace,

weed/s3api/s3tables/handler_sts_legacy_owner_test.go

@@ -0,0 +1,141 @@
+package s3tables
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/iam/sts"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// A table bucket a web-identity role session created before owners were
+// issuer-scoped is owned by the bare OIDC subject. The same subject's later
+// sessions, signed with SigV4 session credentials, carry the scoped aws:userid
+// and no asserted issuers, and must still own the bucket.
+func TestSTSSessionKeepsLegacySubjectOwnedBucket(t *testing.T) {
+	const (
+		bucket = "lake-norm"
+		sub    = "project_path:dev/infra/datalake/normalisation:ref_type:branch:ref:main"
+		issuer = "https://gitlab.example.com"
+	)
+	arn, err := BuildBucketARN(DefaultRegion, DefaultAccountID, bucket)
+	require.NoError(t, err)
+	body, err := json.Marshal(&CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{"some_ns"}})
+	require.NoError(t, err)
+
+	createNamespace := func(t *testing.T, sessionIssuer string, configured ...string) int {
+		fs := s3tablestest.Start(t)
+		metadata, err := json.Marshal(&tableBucketMetadata{Name: bucket, CreatedAt: time.Now(), OwnerAccountID: sub})
+		require.NoError(t, err)
+		fs.Put(TablesPath, bucket, map[string][]byte{ExtendedKeyTableBucket: []byte("true"), ExtendedKeyMetadata: metadata})
+		m := NewManager()
+		m.SetDefaultAllow(false)
+		if configured != nil {
+			m.SetTrustedIssuerSource(func(context.Context) []string { return configured })
+		}
+		identity := &testIdentity{
+			Name:    sub,
+			Account: &testIdentityAccount{Id: sub},
+			Claims: map[string]interface{}{
+				"sub":        sub,
+				"aws:userid": sts.ComputeParentUser(sub, sessionIssuer),
+			},
+		}
+		req, err := newManagerRequest(s3_constants.SetIdentityInContext(context.Background(), identity), "CreateNamespace", body, sub)
+		require.NoError(t, err)
+		recorder := httptest.NewRecorder()
+		m.handler.HandleRequest(recorder, req, NewManagerClient(fs.Client))
+		return recorder.Code
+	}
+
+	assert.Equal(t, http.StatusOK, createNamespace(t, issuer, issuer), "the subject's own session lost its bucket")
+	assert.Equal(t, http.StatusForbidden, createNamespace(t, issuer), "no configured issuer set must not re-anchor")
+	assert.Equal(t, http.StatusForbidden, createNamespace(t, issuer, issuer, "https://other.example.com"),
+		"a legacy subject is ambiguous once two issuers are configured")
+	assert.Equal(t, http.StatusForbidden, createNamespace(t, "https://other.example.com", issuer),
+		"another issuer's user with the same subject must not take the bucket")
+}
+
+// Issuers asserted by the bearer path win, even when empty; an identity that
+// is not a federated session never reads the configured set.
+func TestTrustedIssuersSourceOnlyForFederatedSessions(t *testing.T) {
+	h := NewS3TablesHandler()
+	h.SetTrustedIssuerSource(func(context.Context) []string { return []string{"configured"} })
+	issuersFor := func(claims map[string]interface{}) []string {
+		req := httptest.NewRequest(http.MethodGet, "/", nil)
+		req = req.WithContext(s3_constants.SetIdentityInContext(req.Context(), &testIdentity{Claims: claims}))
+		return h.trustedIssuers(req)
+	}
+	scoped := sts.ComputeParentUser("alice", "configured")
+	assert.Equal(t, []string{"configured"}, issuersFor(map[string]interface{}{"aws:userid": scoped}))
+	assert.Equal(t, []string{"asserted"}, issuersFor(map[string]interface{}{"aws:userid": scoped, "seaweed:oidc-issuers": "asserted"}))
+	assert.Empty(t, issuersFor(map[string]interface{}{"aws:userid": scoped, "seaweed:oidc-issuers": ""}))
+	assert.Empty(t, issuersFor(map[string]interface{}{"aws:userid": "alice"}))
+	assert.Empty(t, issuersFor(map[string]interface{}{"sub": "alice"}))
+}
+
+// A table bucket policy written before issuer scoping names role sessions by
+// a pattern over their bare subjects. A session whose subject matches it keeps
+// the grant; a subject that does not hash to the session's aws:userid does not.
+func TestSTSSessionMatchesLegacySubjectPatternPrincipal(t *testing.T) {
+	const (
+		bucket = "lake-norm"
+		sub    = "project_path:dev/infra/datalake/normalisation:ref_type:branch:ref:main"
+		issuer = "https://gitlab.example.com"
+	)
+	arn, err := BuildBucketARN(DefaultRegion, DefaultAccountID, bucket)
+	require.NoError(t, err)
+	body, err := json.Marshal(&CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{"some_ns"}})
+	require.NoError(t, err)
+	policy := func(principal string) []byte {
+		return []byte(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":["` + principal +
+			`"],"Action":["s3tables:CreateNamespace"],"Resource":"*"}]}`)
+	}
+
+	createNamespace := func(t *testing.T, principalSpec, claimedSub, scopedSub, sessionIssuer string, configured ...string) int {
+		fs := s3tablestest.Start(t)
+		metadata, err := json.Marshal(&tableBucketMetadata{Name: bucket, CreatedAt: time.Now(), OwnerAccountID: s3_constants.AccountAdminId})
+		require.NoError(t, err)
+		fs.Put(TablesPath, bucket, map[string][]byte{
+			ExtendedKeyTableBucket: []byte("true"),
+			ExtendedKeyMetadata:    metadata,
+			ExtendedKeyPolicy:      policy(principalSpec),
+		})
+		m := NewManager()
+		m.SetDefaultAllow(false)
+		if configured != nil {
+			m.SetTrustedIssuerSource(func(context.Context) []string { return configured })
+		}
+		scoped := sts.ComputeParentUser(scopedSub, sessionIssuer)
+		identity := &testIdentity{
+			Name:    scoped,
+			Account: &testIdentityAccount{Id: scoped},
+			Claims:  map[string]interface{}{"sub": claimedSub, "aws:userid": scoped},
+		}
+		req, err := newManagerRequest(s3_constants.SetIdentityInContext(context.Background(), identity), "CreateNamespace", body, scoped)
+		require.NoError(t, err)
+		recorder := httptest.NewRecorder()
+		m.handler.HandleRequest(recorder, req, NewManagerClient(fs.Client))
+		return recorder.Code
+	}
+
+	const pattern = "project_path:dev/infra/datalake/normalisation:*"
+	assert.Equal(t, http.StatusOK, createNamespace(t, pattern, sub, sub, issuer, issuer), "the subject pattern no longer grants")
+	assert.Equal(t, http.StatusOK, createNamespace(t, sub, sub, sub, issuer, issuer), "the exact subject no longer grants")
+	assert.Equal(t, http.StatusForbidden, createNamespace(t, pattern, sub, sub, issuer), "no configured issuer set must not re-anchor")
+	assert.Equal(t, http.StatusForbidden, createNamespace(t, pattern, sub, sub, issuer, issuer, "https://other.example.com"),
+		"a subject pattern is ambiguous once two issuers are configured")
+	assert.Equal(t, http.StatusForbidden, createNamespace(t, pattern, sub, sub, "https://other.example.com", issuer),
+		"another issuer's user with the same subject must not match")
+	assert.Equal(t, http.StatusForbidden, createNamespace(t, pattern, sub, "project_path:other/repo:ref_type:branch:ref:main", issuer, issuer),
+		"a subject claim the session's aws:userid was not derived from must not match")
+	assert.Equal(t, http.StatusForbidden, createNamespace(t, "arn:aws:iam::*", sub, sub, issuer, issuer),
+		"an ARN pattern is not a subject pattern")
+}

weed/s3api/s3tables/handler_table.go

@@ -501,6 +501,7 @@ func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request,
 		metadata.OwnerAccountID, tablePolicy, tableARN,
 		bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
 			Namespace:        namespace,
@@ -652,6 +653,7 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 				nsMeta.OwnerAccountID, namespacePolicy, bucketARN,
 				bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 					TrustedIssuers:   h.trustedIssuers(r),
+					Subject:          callerSubject(r),
 					IsLocalAccountID: h.localAccountIDs,
 					TableBucketName:  bucketName,
 					Namespace:        namespaceName,
@@ -705,6 +707,7 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			identityActions := getIdentityActions(r)
 			if !CheckPermissionWithContext("ListTables", accountID, bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 				TrustedIssuers:   h.trustedIssuers(r),
+				Subject:          callerSubject(r),
 				IsLocalAccountID: h.localAccountIDs,
 				TableBucketName:  bucketName,
 				TableBucketTags:  bucketTags,
@@ -1059,6 +1062,7 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
 		metadata.OwnerAccountID, tablePolicy, tableARN,
 		bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
 			Namespace:        namespaceName,
@@ -1409,6 +1413,7 @@ func tablePurgeDir(entry *filer_pb.Entry, tablePath, bucketName string) (string,
 func (h *S3TablesHandler) canReadEntry(readOp, principal, tableOwner, bucketOwner, tablePolicy, bucketPolicy, tableARN, bucketARN, bucketName, namespace, name string, bucketTags, tableTags map[string]string, identityActions []string, r *http.Request) bool {
 	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespace,
@@ -1617,6 +1622,7 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ
 		metadata.OwnerAccountID, tablePolicy, tableARN,
 		bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
 			Namespace:        srcNamespace,
@@ -1686,6 +1692,7 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ
 		destNamespaceMetadata.OwnerAccountID, destNamespacePolicy, bucketARN,
 		bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
 			Namespace:        destNamespace,
@@ -2090,6 +2097,7 @@ func (h *S3TablesHandler) authorizeUpdateTable(w http.ResponseWriter, r *http.Re
 		metadata.OwnerAccountID, tablePolicy, tableARN,
 		bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
+			Subject:          callerSubject(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
 			Namespace:        namespaceName,
@@ -2367,6 +2375,7 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re
 	identityActions := getIdentityActions(r)
 	policyContext := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,

weed/s3api/s3tables/handler_view.go

@@ -807,6 +807,7 @@ func (h *S3TablesHandler) authorizeViewOp(r *http.Request, operation, principal,
 	identityActions := getIdentityActions(r)
 	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
+		Subject:          callerSubject(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespace,

weed/s3api/s3tables/manager.go

@@ -47,6 +47,16 @@ func (m *Manager) SetTrusted(trusted bool) {
 	m.handler.SetTrusted(trusted)
 }
 
+// SetLocalAccountChecker sets the handler's configured local account predicate.
+func (m *Manager) SetLocalAccountChecker(fn func(string) bool) {
+	m.handler.SetLocalAccountChecker(fn)
+}
+
+// SetTrustedIssuerSource sets the handler's configured OIDC issuer source.
+func (m *Manager) SetTrustedIssuerSource(fn func(context.Context) []string) {
+	m.handler.SetTrustedIssuerSource(fn)
+}
+
 // SetEncryptionChecker sets the handler's SSE-S3 availability predicate.
 func (m *Manager) SetEncryptionChecker(fn func() bool) {
 	m.handler.SetEncryptionChecker(fn)
@@ -70,7 +80,7 @@ func (m *Manager) PrincipalFor(r *http.Request) string {
 // action scoped to that bucket apply, as the handlers do.
 func (m *Manager) PolicyContext(r *http.Request, bucket string) *PolicyContext {
 	ctx := &PolicyContext{IsLocalAccountID: m.handler.localAccountIDs, IdentityActions: getIdentityActions(r),
-		TrustedIssuers: m.handler.trustedIssuers(r), TableBucketName: bucket}
+		TrustedIssuers: m.handler.trustedIssuers(r), Subject: callerSubject(r), TableBucketName: bucket}
 	if m.handler.trusted || (m.handler.defaultAllow && s3_constants.GetIdentityNameFromContext(r) == "") {
 		ctx.DefaultAllow = true
 	}

weed/s3api/s3tables/permissions.go

@@ -159,6 +159,9 @@ type PolicyContext struct {
 	// be re-anchored; a nil predicate exempts nothing beyond the built-in
 	// reserved, ARN, and scoped-id checks.
 	IsLocalAccountID func(string) bool
+	// Subject is the caller's OIDC subject, matched against subject patterns
+	// written before issuer scoping once it is shown to hash to the principal.
+	Subject string
 }
 
 // CheckPermissionWithContext checks permission with optional resource and condition context.
@@ -565,10 +568,10 @@ func matchCaller(principalSpec interface{}, principal string, ctx *PolicyContext
 	}
 	switch p := principalSpec.(type) {
 	case string:
-		return matchesLegacySubject(p, principal, ctx)
+		return matchesLegacyPrincipal(p, principal, ctx)
 	case []interface{}:
 		for _, item := range p {
-			if str, ok := item.(string); ok && matchesLegacySubject(str, principal, ctx) {
+			if str, ok := item.(string); ok && matchesLegacyPrincipal(str, principal, ctx) {
 				return true
 			}
 		}
@@ -604,6 +607,17 @@ func matchesLegacySubject(legacy, principal string, ctx *PolicyContext) bool {
 	return sts.ComputeParentUser(legacy, ctx.TrustedIssuers[0]) == principal
 }
 
+// matchesLegacyPrincipal is matchesLegacySubject for a policy principal, which
+// may also be a subject pattern: it matches the caller's subject only when that
+// subject itself resolves to the principal.
+func matchesLegacyPrincipal(spec, principal string, ctx *PolicyContext) bool {
+	if matchesLegacySubject(spec, principal, ctx) {
+		return true
+	}
+	return ctx != nil && strings.ContainsAny(spec, "*?") && !strings.HasPrefix(spec, "arn:") &&
+		wildcard.MatchesWildcard(spec, ctx.Subject) && matchesLegacySubject(ctx.Subject, principal, ctx)
+}
+
 // matchesPrincipal checks if the principal matches the statement's principal
 func matchesPrincipal(principalSpec interface{}, principal string) bool {
 	return matchPrincipal(principalSpec, principal, false)