iceberg: vend table-scoped credentials to clients that ask for delegation (#10777)

* iceberg: vend table-scoped credentials to clients that ask for delegation

The catalog recognised X-Iceberg-Access-Delegation: vended-credentials
and then deliberately said nothing, because it had nothing to vend: it
withheld even the S3 endpoint so the client would keep the credentials it
was configured with. That left every engine expecting the catalog to hand
out access - Snowflake, Databricks, Trino with vending, any multi-tenant
setup - needing static S3 keys distributed out of band.

Mint an STS session per request instead, scoped by a session policy to
the table's own prefix plus the bucket listing needed to resolve it, and
return it in the load response config and storage-credentials. The role
to assume is named by -s3.iceberg.credentialRole; its trust policy is
what decides whether a caller may assume it, and vending stays off until
it is set. A failed mint falls back to the old silence rather than
handing back an endpoint the client cannot sign for.

* iceberg: keep vended credentials inside the table prefix

Review follow-ups on credential vending:

Listing was granted on the bucket ARN with no condition, so a credential
vended for one table could enumerate every other table's object names.
Constrain s3:prefix to the table's own prefix, which the S3 gateway
already populates for list requests.

A table location carrying * or ? would have gone into the policy's
resource pattern unescaped and widened the session to sibling prefixes.
Refuse to vend for such a location rather than escaping it; nothing the
catalog generates contains those characters.

DurationSeconds skipped the 900..43200 bounds the other assume-role paths
enforce, so -s3.iceberg.credentialDurationSeconds could ask for a session
outside them. The check is now shared by all three entry points.

* iceberg: return the vended credentials from buildFileIOConfig itself

buildStorageConfig was a second name for what buildFileIOConfig already
did; it now returns the storage credentials alongside the properties, and
callers that only want the properties drop them.

* iceberg: split the vended bucket grants, and refuse a whole-bucket scope

The prefix condition sat on a statement that also granted
GetBucketLocation and ListBucketMultipartUploads, neither of which carries
an s3:prefix to satisfy it, so both were denied for every vended
credential. GetBucketLocation moves to its own unconditioned statement.
ListBucketMultipartUploads is dropped: Iceberg writers complete and abort
by upload id, and granting it either leaks in-flight keys bucket-wide or
breaks on the same missing prefix.

A table whose location has no prefix - one registered at the bucket root -
would have been vended read and write over every other table in the
bucket. Refuse, the way a location with wildcards is refused.

weed/command/mini.go

@@ -498,6 +498,8 @@ func initMiniS3Flags() {
 	miniS3Options.portHttps = cmdMini.Flag.Int("s3.port.https", 0, "s3 server https listen port")
 	miniS3Options.portGrpc = cmdMini.Flag.Int("s3.port.grpc", 0, "s3 server grpc listen port")
 	miniS3Options.portIceberg = cmdMini.Flag.Int("s3.port.iceberg", 8181, "Iceberg REST Catalog server listen port (0 to disable)")
+	miniS3Options.icebergCredentialRole = cmdMini.Flag.String("s3.iceberg.credentialRole", "", "IAM role ARN the Iceberg catalog assumes to vend table-scoped credentials (empty disables vending)")
+	miniS3Options.icebergCredentialDuration = cmdMini.Flag.Int("s3.iceberg.credentialDurationSeconds", 3600, "lifetime of credentials vended by the Iceberg catalog")
 	miniS3Options.domainName = cmdMini.Flag.String("s3.domainName", "", "suffix of the host name in comma separated list, {bucket}.{domainName}")
 	miniS3Options.allowedOrigins = cmdMini.Flag.String("s3.allowedOrigins", "*", "comma separated list of allowed origins")
 	miniS3Options.tlsPrivateKey = cmdMini.Flag.String("s3.key.file", "", "path to the TLS private key file")

weed/command/s3.go

@@ -48,6 +48,8 @@ type S3Options struct {
 	portHttps                 *int
 	portGrpc                  *int
 	portIceberg               *int
+	icebergCredentialRole     *string
+	icebergCredentialDuration *int
 	config                    *string
 	iamConfig                 *string
 	domainName                *string
@@ -91,6 +93,8 @@ func init() {
 	s3StandaloneOptions.portHttps = cmdS3.Flag.Int("port.https", 0, "s3 server https listen port")
 	s3StandaloneOptions.portGrpc = cmdS3.Flag.Int("port.grpc", 0, "s3 server grpc listen port")
 	s3StandaloneOptions.portIceberg = cmdS3.Flag.Int("port.iceberg", 8181, "Iceberg REST Catalog server listen port (0 to disable)")
+	s3StandaloneOptions.icebergCredentialRole = cmdS3.Flag.String("iceberg.credentialRole", "", "IAM role ARN the Iceberg catalog assumes to vend table-scoped credentials (empty disables vending)")
+	s3StandaloneOptions.icebergCredentialDuration = cmdS3.Flag.Int("iceberg.credentialDurationSeconds", 3600, "lifetime of credentials vended by the Iceberg catalog")
 	s3StandaloneOptions.domainName = cmdS3.Flag.String("domainName", "", "suffix of the host name in comma separated list, {bucket}.{domainName}")
 	s3StandaloneOptions.allowedOrigins = cmdS3.Flag.String("allowedOrigins", "*", "comma separated list of allowed origins")
 	s3StandaloneOptions.dataCenter = cmdS3.Flag.String("dataCenter", "", "prefer to read and write to volumes in this data center")
@@ -544,6 +548,14 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) {
 	// Create Iceberg server using the S3ApiServer as filer client
 	icebergServer := iceberg.NewServer(s3ApiServer, s3ApiServer)
 	icebergServer.SetCredentialValidator(s3ApiServer)
+	if s3opt.icebergCredentialRole != nil && *s3opt.icebergCredentialRole != "" {
+		duration := int64(0)
+		if s3opt.icebergCredentialDuration != nil {
+			duration = int64(*s3opt.icebergCredentialDuration)
+		}
+		s3ApiServer.SetIcebergCredentialRole(*s3opt.icebergCredentialRole, duration)
+		icebergServer.SetCredentialVendor(icebergCredentialVendor{s3ApiServer})
+	}
 	icebergServer.SetS3Endpoint(s3opt.deriveS3AdvertisedEndpoint())
 	icebergServer.RegisterRoutes(icebergRouter)
 
@@ -615,3 +627,22 @@ func (s3opt *S3Options) deriveS3AdvertisedEndpoint() string {
 	}
 	return fmt.Sprintf("%s://%s", scheme, util.JoinHostPort(host, port))
 }
+
+// icebergCredentialVendor adapts the S3 gateway's STS-backed vending to the
+// catalog's interface, keeping the two packages independent of each other.
+type icebergCredentialVendor struct {
+	server *s3api.S3ApiServer
+}
+
+func (v icebergCredentialVendor) VendTableCredentials(ctx context.Context, principal, bucket, prefix string) (*iceberg.VendedCredentials, error) {
+	credentials, err := v.server.VendTableCredentials(ctx, principal, bucket, prefix)
+	if err != nil || credentials == nil {
+		return nil, err
+	}
+	return &iceberg.VendedCredentials{
+		AccessKeyID:     credentials.AccessKeyID,
+		SecretAccessKey: credentials.SecretAccessKey,
+		SessionToken:    credentials.SessionToken,
+		Expiration:      credentials.Expiration,
+	}, nil
+}

weed/command/server.go

@@ -164,6 +164,8 @@ func init() {
 	s3Options.portHttps = cmdServer.Flag.Int("s3.port.https", 0, "s3 server https listen port")
 	s3Options.portGrpc = cmdServer.Flag.Int("s3.port.grpc", 0, "s3 server grpc listen port")
 	s3Options.portIceberg = cmdServer.Flag.Int("s3.port.iceberg", 8181, "Iceberg REST Catalog server listen port (0 to disable)")
+	s3Options.icebergCredentialRole = cmdServer.Flag.String("s3.iceberg.credentialRole", "", "IAM role ARN the Iceberg catalog assumes to vend table-scoped credentials (empty disables vending)")
+	s3Options.icebergCredentialDuration = cmdServer.Flag.Int("s3.iceberg.credentialDurationSeconds", 3600, "lifetime of credentials vended by the Iceberg catalog")
 	s3Options.domainName = cmdServer.Flag.String("s3.domainName", "", "suffix of the host name in comma separated list, {bucket}.{domainName}")
 	s3Options.allowedOrigins = cmdServer.Flag.String("s3.allowedOrigins", "*", "comma separated list of allowed origins")
 	s3Options.tlsPrivateKey = cmdServer.Flag.String("s3.key.file", "", "path to the TLS private key file")

weed/iam/sts/credential_prefix_test.go

@@ -1,6 +1,7 @@
 package sts
 
 import (
+	"context"
 	"encoding/base64"
 	"encoding/hex"
 	"strings"
@@ -66,3 +67,34 @@ func TestTemporaryCredentialFormat(t *testing.T) {
 	// Verify session token is not empty
 	assert.NotEmpty(t, cred.SessionToken)
 }
+
+// A duration that came from configuration is bounded the same as one from a
+// request: the catalog passes -s3.iceberg.credentialDurationSeconds straight
+// through.
+func TestAssumeRoleForPrincipalValidatesDuration(t *testing.T) {
+	svc := NewSTSService()
+	err := svc.Initialize(&STSConfig{
+		TokenDuration:    FlexibleDuration{time.Hour},
+		MaxSessionLength: FlexibleDuration{12 * time.Hour},
+		Issuer:           "test-issuer",
+		SigningKey:       []byte("test-signing-key-at-least-32-bytes-long"),
+	})
+	if err != nil {
+		t.Fatalf("Initialize() error = %v", err)
+	}
+
+	for _, seconds := range []int64{60, 100000} {
+		duration := seconds
+		_, err := svc.AssumeRoleForPrincipal(context.Background(), &AssumeRoleForPrincipalRequest{
+			RoleArn:         "arn:aws:iam::role/IcebergTableAccess",
+			Principal:       "admin",
+			DurationSeconds: &duration,
+		})
+		if err == nil {
+			t.Fatalf("AssumeRoleForPrincipal(%ds) error = nil, want a validation error", seconds)
+		}
+		if !strings.Contains(err.Error(), "DurationSeconds") {
+			t.Errorf("AssumeRoleForPrincipal(%ds) error = %v, want it to name DurationSeconds", seconds, err)
+		}
+	}
+}

weed/iam/sts/sts_service.go

@@ -754,12 +754,20 @@ func (s *STSService) AssumeRoleWithCredentials(ctx context.Context, request *Ass
 		return nil, fmt.Errorf("role assumption denied: %w", err)
 	}
 
-	// 4. Calculate session duration
-	// For credential-based auth, there's no source token with expiration to cap against
-	sessionDuration := s.calculateSessionDuration(request.DurationSeconds, nil)
+	// 4-7. Mint the session. For credential-based auth there is no source token
+	// with an expiration to cap the duration against.
+	return s.issueSession(request.RoleArn, request.RoleSessionName, sessionPolicy,
+		request.DurationSeconds, nil, provider.Name(), externalIdentity.UserID)
+}
+
+// issueSession mints temporary credentials and the self-contained JWT that
+// carries the whole session, shared by every assume-role entry point.
+func (s *STSService) issueSession(roleArn, roleSessionName, sessionPolicy string,
+	durationSeconds *int64, tokenExpiration *time.Time, providerName, subject string) (*AssumeRoleResponse, error) {
+
+	sessionDuration := s.calculateSessionDuration(durationSeconds, tokenExpiration)
 	expiresAt := time.Now().Add(sessionDuration)
 
-	// 5. Generate session ID and temporary credentials
 	sessionId, err := GenerateSessionId()
 	if err != nil {
 		return nil, fmt.Errorf("failed to generate session ID: %w", err)
@@ -771,38 +779,112 @@ func (s *STSService) AssumeRoleWithCredentials(ctx context.Context, request *Ass
 		return nil, fmt.Errorf("failed to generate credentials: %w", err)
 	}
 
-	// 6. Create comprehensive JWT session token with all session information embedded
 	assumedRoleUser := &AssumedRoleUser{
-		AssumedRoleId: request.RoleArn,
-		Arn:           GenerateAssumedRoleArn(request.RoleArn, request.RoleSessionName),
-		Subject:       externalIdentity.UserID,
+		AssumedRoleId: roleArn,
+		Arn:           GenerateAssumedRoleArn(roleArn, roleSessionName),
+		Subject:       subject,
 	}
 
-	// Create rich JWT claims with all session information
 	sessionClaims := NewSTSSessionClaims(sessionId, s.Config.Issuer, expiresAt).
-		WithSessionName(request.RoleSessionName).
-		WithRoleInfo(request.RoleArn, assumedRoleUser.Arn, assumedRoleUser.Arn).
-		WithIdentityProvider(provider.Name(), externalIdentity.UserID, "").
+		WithSessionName(roleSessionName).
+		WithRoleInfo(roleArn, assumedRoleUser.Arn, assumedRoleUser.Arn).
+		WithIdentityProvider(providerName, subject, "").
 		WithMaxDuration(sessionDuration)
 	if sessionPolicy != "" {
 		sessionClaims.WithSessionPolicy(sessionPolicy)
 	}
 
-	// Generate self-contained JWT token with all session information
 	jwtToken, err := s.tokenGenerator.GenerateJWTWithClaims(sessionClaims)
 	if err != nil {
 		return nil, fmt.Errorf("failed to generate JWT session token: %w", err)
 	}
 	tempCredentials.SessionToken = jwtToken
 
-	// 7. Build and return response (no session storage needed!)
-
 	return &AssumeRoleResponse{
 		Credentials:     tempCredentials,
 		AssumedRoleUser: assumedRoleUser,
 	}, nil
 }
 
+// validateSessionDurationSeconds bounds a requested session lifetime the way
+// AWS STS does. Every assume-role entry point runs it, so a duration that came
+// from configuration is checked the same as one from a request.
+func validateSessionDurationSeconds(durationSeconds *int64) error {
+	if durationSeconds == nil {
+		return nil
+	}
+	if *durationSeconds < 900 || *durationSeconds > 43200 { // 15min to 12 hours
+		return fmt.Errorf("DurationSeconds must be between 900 and 43200 seconds")
+	}
+	return nil
+}
+
+// AssumeRoleForPrincipalRequest asks for a session on behalf of a principal the
+// calling service has already authenticated.
+type AssumeRoleForPrincipalRequest struct {
+	// RoleArn is the ARN of the role to assume.
+	RoleArn string
+
+	// Principal identifies the already-authenticated caller.
+	Principal string
+
+	// RoleSessionName names the session.
+	RoleSessionName string
+
+	// ProviderName records how the caller was authenticated.
+	ProviderName string
+
+	// Policy optionally narrows the session below the role's own permissions.
+	Policy *string
+
+	// DurationSeconds requests a session lifetime.
+	DurationSeconds *int64
+}
+
+// AssumeRoleForPrincipal issues session credentials for a caller that a
+// SeaweedFS service authenticated itself, such as the Iceberg catalog vending
+// scoped credentials for a table it has already authorized. There is no
+// external token left to verify at this point, so the role's trust policy is
+// the control point and is still enforced.
+func (s *STSService) AssumeRoleForPrincipal(ctx context.Context, request *AssumeRoleForPrincipalRequest) (*AssumeRoleResponse, error) {
+	if !s.initialized {
+		return nil, fmt.Errorf(ErrSTSServiceNotInitialized)
+	}
+	if request == nil {
+		return nil, fmt.Errorf("request cannot be nil")
+	}
+	if request.RoleArn == "" {
+		return nil, fmt.Errorf("role ARN cannot be empty")
+	}
+	if request.Principal == "" {
+		return nil, fmt.Errorf("principal cannot be empty")
+	}
+	if err := validateSessionDurationSeconds(request.DurationSeconds); err != nil {
+		return nil, fmt.Errorf("invalid request: %w", err)
+	}
+
+	sessionPolicy := ""
+	if request.Policy != nil {
+		normalized, err := NormalizeSessionPolicy(*request.Policy)
+		if err != nil {
+			return nil, fmt.Errorf("invalid session policy: %w", err)
+		}
+		sessionPolicy = normalized
+	}
+
+	identity := &providers.ExternalIdentity{
+		UserID:      request.Principal,
+		DisplayName: request.Principal,
+		Provider:    request.ProviderName,
+	}
+	if err := s.validateRoleAssumptionForCredentials(ctx, request.RoleArn, identity); err != nil {
+		return nil, fmt.Errorf("role assumption denied: %w", err)
+	}
+
+	return s.issueSession(request.RoleArn, request.RoleSessionName, sessionPolicy,
+		request.DurationSeconds, nil, request.ProviderName, request.Principal)
+}
+
 // ValidateSessionToken validates a session token and returns session information
 // This method is now completely stateless - all session information is extracted from the JWT token
 func (s *STSService) ValidateSessionToken(ctx context.Context, sessionToken string) (*SessionInfo, error) {
@@ -855,14 +937,7 @@ func (s *STSService) validateAssumeRoleWithWebIdentityRequest(request *AssumeRol
 		return fmt.Errorf("RoleSessionName is required")
 	}
 
-	// Validate session duration if provided
-	if request.DurationSeconds != nil {
-		if *request.DurationSeconds < 900 || *request.DurationSeconds > 43200 { // 15min to 12 hours
-			return fmt.Errorf("DurationSeconds must be between 900 and 43200 seconds")
-		}
-	}
-
-	return nil
+	return validateSessionDurationSeconds(request.DurationSeconds)
 }
 
 // validateWebIdentityToken validates the web identity token with strict issuer-to-provider mapping
@@ -1095,14 +1170,7 @@ func (s *STSService) validateAssumeRoleWithCredentialsRequest(request *AssumeRol
 		return fmt.Errorf("ProviderName is required")
 	}
 
-	// Validate session duration if provided
-	if request.DurationSeconds != nil {
-		if *request.DurationSeconds < 900 || *request.DurationSeconds > 43200 { // 15min to 12 hours
-			return fmt.Errorf("DurationSeconds must be between 900 and 43200 seconds")
-		}
-	}
-
-	return nil
+	return validateSessionDurationSeconds(request.DurationSeconds)
 }
 
 // ExpireSessionForTesting manually expires a session for testing purposes

weed/s3api/iceberg/handlers_table.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"os"
 	"path"
+	"strconv"
 	"strings"
 
 	"github.com/apache/iceberg-go"
@@ -273,10 +274,12 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 		if markerErr := s.writeStageCreateMarker(r.Context(), bucketName, namespace, tableName, tableUUID, location, stagedMetadataLocation); markerErr != nil {
 			glog.V(1).Infof("Iceberg: failed to persist stage-create marker for %s.%s: %v", flattenNamespacePath(namespace), tableName, markerErr)
 		}
+		config, storageCredentials := s.buildFileIOConfig(r, location)
 		result := LoadTableResult{
-			MetadataLocation: metadataLocation,
-			Metadata:         metadata,
-			Config:           s.buildFileIOConfig(r),
+			MetadataLocation:   metadataLocation,
+			Metadata:           metadata,
+			Config:             config,
+			StorageCredentials: storageCredentials,
 		}
 		writeLoadResult(w, http.StatusOK, result)
 		return
@@ -372,10 +375,12 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 		glog.V(1).Infof("Iceberg: failed to cleanup stage-create markers for %s.%s after create: %v", flattenNamespacePath(namespace), tableName, markerErr)
 	}
 
+	config, storageCredentials := s.buildFileIOConfig(r, location)
 	result := LoadTableResult{
-		MetadataLocation: finalLocation,
-		Metadata:         metadata,
-		Config:           s.buildFileIOConfig(r),
+		MetadataLocation:   finalLocation,
+		Metadata:           metadata,
+		Config:             config,
+		StorageCredentials: storageCredentials,
 	}
 	writeLoadResult(w, http.StatusOK, result)
 }
@@ -565,10 +570,12 @@ func (s *Server) buildLoadTableResult(r *http.Request, getResp s3tables.GetTable
 		return LoadTableResult{}, fmt.Errorf("build metadata for %s: %w", tableName, err)
 	}
 
+	config, storageCredentials := s.buildFileIOConfig(r, location)
 	return LoadTableResult{
-		MetadataLocation: getResp.MetadataLocation,
-		Metadata:         metadata,
-		Config:           s.buildFileIOConfig(r),
+		MetadataLocation:   getResp.MetadataLocation,
+		Metadata:           metadata,
+		Config:             config,
+		StorageCredentials: storageCredentials,
 	}, nil
 }
 
@@ -617,22 +624,79 @@ func applySnapshotsParam(r *http.Request, metadata table.Metadata) (table.Metada
 // separately discovering the endpoint. The region defaults to the same
 // value baked into table bucket ARNs so clients like DuckDB that require
 // a region on attach don't need to be told it out-of-band. See issue #9103.
-func (s *Server) buildFileIOConfig(r *http.Request) iceberg.Properties {
+// It also returns, for a client that asked for credential vending, the
+// credentials scoped to this table.
+func (s *Server) buildFileIOConfig(r *http.Request, location string) (iceberg.Properties, []StorageCredential) {
 	config := make(iceberg.Properties)
 	if s.s3Endpoint == "" {
-		return config
+		return config, nil
 	}
-	// A client asking for vended credentials builds its storage credential out
-	// of whatever comes back here and stops using the one it was configured
-	// with. We vend none, so an endpoint on its own leaves it sending unsigned
-	// requests; say nothing instead and let it keep its own credentials.
+
+	vended := iceberg.Properties(nil)
 	if wantsVendedCredentials(r) {
-		return config
+		vended = s.vendCredentials(r, location)
+		// A client asking for vended credentials builds its storage credential
+		// out of whatever comes back here and stops using the one it was
+		// configured with. With nothing to vend, an endpoint on its own leaves
+		// it sending unsigned requests; say nothing instead and let it keep its
+		// own credentials.
+		if vended == nil {
+			return config, nil
+		}
 	}
+
 	config["s3.endpoint"] = s.s3Endpoint
 	config["s3.path-style-access"] = "true"
 	config["s3.region"] = s3tables.DefaultRegion
-	return config
+	if vended == nil {
+		return config, nil
+	}
+
+	credentialConfig := make(iceberg.Properties, len(config)+len(vended))
+	for key, value := range config {
+		credentialConfig[key] = value
+	}
+	for key, value := range vended {
+		config[key] = value
+		credentialConfig[key] = value
+	}
+	return config, []StorageCredential{{Prefix: location, Config: credentialConfig}}
+}
+
+// vendCredentials mints credentials limited to this table's prefix. It returns
+// nil when the deployment has no vending configured or the mint failed, which
+// leaves the caller withholding the endpoint so the client keeps its own
+// credentials rather than falling back to unsigned requests.
+func (s *Server) vendCredentials(r *http.Request, location string) iceberg.Properties {
+	if s.credentialVendor == nil || location == "" {
+		return nil
+	}
+
+	bucket, prefix, err := parseS3Location(location)
+	if err != nil {
+		glog.V(1).Infof("Iceberg: cannot vend credentials for %s: %v", location, err)
+		return nil
+	}
+
+	principal := s3_constants.GetIdentityNameFromContext(r)
+	credentials, err := s.credentialVendor.VendTableCredentials(r.Context(), principal, bucket, prefix)
+	if err != nil {
+		glog.Warningf("Iceberg: failed to vend credentials for %s: %v", location, err)
+		return nil
+	}
+	if credentials == nil {
+		return nil
+	}
+
+	vended := iceberg.Properties{
+		"s3.access-key-id":     credentials.AccessKeyID,
+		"s3.secret-access-key": credentials.SecretAccessKey,
+		"s3.session-token":     credentials.SessionToken,
+	}
+	if !credentials.Expiration.IsZero() {
+		vended["s3.session-token-expires-at-ms"] = strconv.FormatInt(credentials.Expiration.UnixMilli(), 10)
+	}
+	return vended
 }
 
 // handleTableExists checks if a table exists.

weed/s3api/iceberg/handlers_view.go

@@ -218,10 +218,11 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) {
 	if finalLocation == "" {
 		finalLocation = metadataLocation
 	}
+	config, _ := s.buildFileIOConfig(r, location)
 	writeLoadResult(w, http.StatusOK, ViewResponse{
 		MetadataLocation: finalLocation,
 		Metadata:         metadata,
-		Config:           s.buildFileIOConfig(r),
+		Config:           config,
 	})
 }
 
@@ -348,10 +349,11 @@ func (s *Server) buildViewResponse(r *http.Request, getResp s3tables.GetViewResp
 	if err != nil {
 		return ViewResponse{}, fmt.Errorf("failed to parse view metadata: %w", err)
 	}
+	config, _ := s.buildFileIOConfig(r, tableLocationFromMetadataLocation(getResp.MetadataLocation))
 	return ViewResponse{
 		MetadataLocation: getResp.MetadataLocation,
 		Metadata:         metadata,
-		Config:           s.buildFileIOConfig(r),
+		Config:           config,
 	}, nil
 }
 

weed/s3api/iceberg/handlers_view_update.go

@@ -129,10 +129,11 @@ func (s *Server) handleUpdateView(w http.ResponseWriter, r *http.Request) {
 			return s.tablesManager.Execute(r.Context(), mgrClient, "UpdateView", updateReq, nil, identityName)
 		})
 		if err == nil {
+			config, _ := s.buildFileIOConfig(r, location)
 			writeLoadResult(w, http.StatusOK, ViewResponse{
 				MetadataLocation: newMetadataLocation,
 				Metadata:         newMetadata,
-				Config:           s.buildFileIOConfig(r),
+				Config:           config,
 			})
 			return
 		}

weed/s3api/iceberg/iceberg_access_delegation_test.go

@@ -35,20 +35,20 @@ func TestWantsVendedCredentials(t *testing.T) {
 }
 
 // A client that asked for vended credentials replaces its own storage
-// credentials with whatever the catalog returns. Since the catalog vends none,
-// an endpoint by itself would leave the client signing nothing and every data
-// file read and write would come back 403.
+// credentials with whatever the catalog returns. With no vending configured
+// there are none, so an endpoint by itself would leave the client signing
+// nothing and every data file read and write would come back 403.
 func TestBuildFileIOConfigWithholdsEndpointFromCredentialVendingClients(t *testing.T) {
 	s := &Server{s3Endpoint: "http://seaweed.example:8333"}
 
 	r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil)
 	r.Header.Set(accessDelegationHeader, "vended-credentials")
-	if got := s.buildFileIOConfig(r); len(got) != 0 {
+	if got, _ := s.buildFileIOConfig(r, "s3://warehouse/ns/t"); len(got) != 0 {
 		t.Fatalf("buildFileIOConfig() = %v, want empty for a vended-credentials request", got)
 	}
 
 	plain := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil)
-	if got := s.buildFileIOConfig(plain); got["s3.endpoint"] != s.s3Endpoint {
+	if got, _ := s.buildFileIOConfig(plain, "s3://warehouse/ns/t"); got["s3.endpoint"] != s.s3Endpoint {
 		t.Fatalf("s3.endpoint = %q, want %q", got["s3.endpoint"], s.s3Endpoint)
 	}
 }

weed/s3api/iceberg/iceberg_credential_vending_test.go

@@ -0,0 +1,111 @@
+package iceberg
+
+import (
+	"context"
+	"errors"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+type stubVendor struct {
+	credentials *VendedCredentials
+	err         error
+	gotBucket   string
+	gotPrefix   string
+	gotcaller   string
+}
+
+func (v *stubVendor) VendTableCredentials(_ context.Context, principal, bucket, prefix string) (*VendedCredentials, error) {
+	v.gotcaller, v.gotBucket, v.gotPrefix = principal, bucket, prefix
+	return v.credentials, v.err
+}
+
+func vendingServer(vendor CredentialVendor) *Server {
+	s := &Server{s3Endpoint: "http://s3.example:8333"}
+	s.SetCredentialVendor(vendor)
+	return s
+}
+
+func TestBuildFileIOConfigVendingVendsScopedCredentials(t *testing.T) {
+	expiry := time.Now().Add(time.Hour).Truncate(time.Millisecond)
+	vendor := &stubVendor{credentials: &VendedCredentials{
+		AccessKeyID:     "ASIAEXAMPLE",
+		SecretAccessKey: "secret",
+		SessionToken:    "token",
+		Expiration:      expiry,
+	}}
+	s := vendingServer(vendor)
+
+	r := httptest.NewRequest("GET", "/v1/namespaces/ns/tables/t", nil)
+	r.Header.Set(accessDelegationHeader, "vended-credentials")
+
+	config, storageCredentials := s.buildFileIOConfig(r, "s3://warehouse/ns/t")
+
+	if vendor.gotBucket != "warehouse" || vendor.gotPrefix != "ns/t" {
+		t.Errorf("vendor called with bucket=%q prefix=%q, want warehouse and ns/t", vendor.gotBucket, vendor.gotPrefix)
+	}
+	if config["s3.access-key-id"] != "ASIAEXAMPLE" || config["s3.secret-access-key"] != "secret" || config["s3.session-token"] != "token" {
+		t.Errorf("config missing vended credentials: %v", config)
+	}
+	if config["s3.endpoint"] != "http://s3.example:8333" {
+		t.Errorf("config missing the endpoint the credentials are for: %v", config)
+	}
+	if len(storageCredentials) != 1 {
+		t.Fatalf("storage-credentials = %d entries, want 1", len(storageCredentials))
+	}
+	if storageCredentials[0].Prefix != "s3://warehouse/ns/t" {
+		t.Errorf("storage credential prefix = %q", storageCredentials[0].Prefix)
+	}
+	if storageCredentials[0].Config["s3.session-token"] != "token" {
+		t.Errorf("storage credential is missing the session token: %v", storageCredentials[0].Config)
+	}
+}
+
+// Without a vendor the response must stay silent about the endpoint too: a
+// client that asked for vended credentials would otherwise drop its own and
+// start sending unsigned requests.
+func TestBuildFileIOConfigVendingWithoutVendorSaysNothing(t *testing.T) {
+	s := &Server{s3Endpoint: "http://s3.example:8333"}
+	r := httptest.NewRequest("GET", "/v1/namespaces/ns/tables/t", nil)
+	r.Header.Set(accessDelegationHeader, "vended-credentials")
+
+	config, storageCredentials := s.buildFileIOConfig(r, "s3://warehouse/ns/t")
+	if len(config) != 0 || storageCredentials != nil {
+		t.Errorf("config = %v, storage-credentials = %v, want both empty", config, storageCredentials)
+	}
+}
+
+func TestBuildFileIOConfigVendingFallsBackWhenVendingFails(t *testing.T) {
+	s := vendingServer(&stubVendor{err: errors.New("no role configured")})
+	r := httptest.NewRequest("GET", "/v1/namespaces/ns/tables/t", nil)
+	r.Header.Set(accessDelegationHeader, "vended-credentials")
+
+	config, storageCredentials := s.buildFileIOConfig(r, "s3://warehouse/ns/t")
+	if len(config) != 0 || storageCredentials != nil {
+		t.Errorf("config = %v, storage-credentials = %v, want both empty", config, storageCredentials)
+	}
+}
+
+// A client that did not ask for delegation keeps the plain endpoint config and
+// must never be handed credentials.
+func TestBuildFileIOConfigVendingWithoutDelegationHeader(t *testing.T) {
+	vendor := &stubVendor{credentials: &VendedCredentials{AccessKeyID: "ASIAEXAMPLE"}}
+	s := vendingServer(vendor)
+
+	r := httptest.NewRequest("GET", "/v1/namespaces/ns/tables/t", nil)
+	config, storageCredentials := s.buildFileIOConfig(r, "s3://warehouse/ns/t")
+
+	if _, vended := config["s3.access-key-id"]; vended {
+		t.Errorf("credentials vended without the delegation header: %v", config)
+	}
+	if config["s3.endpoint"] != "http://s3.example:8333" {
+		t.Errorf("config = %v, want the endpoint", config)
+	}
+	if storageCredentials != nil {
+		t.Errorf("storage-credentials = %v, want none", storageCredentials)
+	}
+	if vendor.gotBucket != "" {
+		t.Errorf("vendor was called for a request that did not ask for delegation")
+	}
+}

weed/s3api/iceberg/iceberg_issue_9103_test.go

@@ -61,7 +61,7 @@ func TestBuildFileIOConfig(t *testing.T) {
 
 	t.Run("no endpoint configured yields empty config", func(t *testing.T) {
 		s := &Server{}
-		got := s.buildFileIOConfig(loadTable())
+		got, _ := s.buildFileIOConfig(loadTable(), "s3://warehouse/ns/t")
 		if len(got) != 0 {
 			t.Fatalf("buildFileIOConfig() = %v, want empty", got)
 		}
@@ -69,7 +69,7 @@ func TestBuildFileIOConfig(t *testing.T) {
 
 	t.Run("endpoint is advertised with path-style-access and region", func(t *testing.T) {
 		s := &Server{s3Endpoint: "http://seaweed.example:8333"}
-		got := s.buildFileIOConfig(loadTable())
+		got, _ := s.buildFileIOConfig(loadTable(), "s3://warehouse/ns/t")
 		if got["s3.endpoint"] != "http://seaweed.example:8333" {
 			t.Fatalf("s3.endpoint = %q, want %q", got["s3.endpoint"], "http://seaweed.example:8333")
 		}

weed/s3api/iceberg/server.go

@@ -1,7 +1,9 @@
 package iceberg
 
 import (
+	"context"
 	"net/http"
+	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -33,6 +35,21 @@ type CredentialValidator interface {
 	GetCredentialByAccessKey(accessKey string) (identityName string, identity interface{}, secretKey string, err error)
 }
 
+// VendedCredentials are short-lived S3 credentials scoped to one table.
+type VendedCredentials struct {
+	AccessKeyID     string
+	SecretAccessKey string
+	SessionToken    string
+	Expiration      time.Time
+}
+
+// CredentialVendor mints credentials limited to a single table's prefix for a
+// caller the catalog has already authenticated and authorized. A nil result
+// with no error means the deployment has vending switched off.
+type CredentialVendor interface {
+	VendTableCredentials(ctx context.Context, principal, bucket, prefix string) (*VendedCredentials, error)
+}
+
 // Server implements the Iceberg REST Catalog API.
 type Server struct {
 	filerClient         FilerClient
@@ -40,6 +57,7 @@ type Server struct {
 	prefix              string // optional prefix for routes
 	authenticator       S3Authenticator
 	credentialValidator CredentialValidator
+	credentialVendor    CredentialVendor
 	s3Endpoint          string // http(s):// URL advertised in LoadTable FileIO config
 }
 
@@ -61,6 +79,12 @@ func NewServer(filerClient FilerClient, authenticator S3Authenticator) *Server {
 	}
 }
 
+// SetCredentialVendor enables credential vending for clients that ask for it
+// with X-Iceberg-Access-Delegation: vended-credentials.
+func (s *Server) SetCredentialVendor(vendor CredentialVendor) {
+	s.credentialVendor = vendor
+}
+
 // SetCredentialValidator sets the credential validator for OAuth token support.
 func (s *Server) SetCredentialValidator(cv CredentialValidator) {
 	s.credentialValidator = cv

weed/s3api/iceberg/types.go

@@ -101,16 +101,25 @@ type RegisterTableRequest struct {
 }
 
 type LoadTableResult struct {
-	MetadataLocation string             `json:"metadata-location,omitempty"`
-	Metadata         table.Metadata     `json:"metadata"`
-	Config           iceberg.Properties `json:"config"`
+	MetadataLocation   string              `json:"metadata-location,omitempty"`
+	Metadata           table.Metadata      `json:"metadata"`
+	Config             iceberg.Properties  `json:"config"`
+	StorageCredentials []StorageCredential `json:"storage-credentials,omitempty"`
+}
+
+// StorageCredential is one set of credentials scoped to a location prefix, as
+// returned to clients that asked for credential vending.
+type StorageCredential struct {
+	Prefix string             `json:"prefix"`
+	Config iceberg.Properties `json:"config"`
 }
 
 // loadTableResultAlias is used for custom JSON unmarshaling.
 type loadTableResultAlias struct {
-	MetadataLocation string             `json:"metadata-location,omitempty"`
-	RawMetadata      json.RawMessage    `json:"metadata"`
-	Config           iceberg.Properties `json:"config,omitempty"`
+	MetadataLocation   string              `json:"metadata-location,omitempty"`
+	RawMetadata        json.RawMessage     `json:"metadata"`
+	Config             iceberg.Properties  `json:"config,omitempty"`
+	StorageCredentials []StorageCredential `json:"storage-credentials,omitempty"`
 }
 
 // MarshalJSON serializes LoadTableResult while backfilling spec-required
@@ -126,13 +135,15 @@ func (r LoadTableResult) MarshalJSON() ([]byte, error) {
 	metaBytes = ensureMetadataSpecCompliance(metaBytes)
 
 	return json.Marshal(struct {
-		MetadataLocation string             `json:"metadata-location,omitempty"`
-		Metadata         json.RawMessage    `json:"metadata"`
-		Config           iceberg.Properties `json:"config"`
+		MetadataLocation   string              `json:"metadata-location,omitempty"`
+		Metadata           json.RawMessage     `json:"metadata"`
+		Config             iceberg.Properties  `json:"config"`
+		StorageCredentials []StorageCredential `json:"storage-credentials,omitempty"`
 	}{
-		MetadataLocation: r.MetadataLocation,
-		Metadata:         metaBytes,
-		Config:           r.Config,
+		MetadataLocation:   r.MetadataLocation,
+		Metadata:           metaBytes,
+		Config:             r.Config,
+		StorageCredentials: r.StorageCredentials,
 	})
 }
 
@@ -146,6 +157,7 @@ func (r *LoadTableResult) UnmarshalJSON(data []byte) error {
 
 	r.MetadataLocation = alias.MetadataLocation
 	r.Config = alias.Config
+	r.StorageCredentials = alias.StorageCredentials
 
 	if len(alias.RawMetadata) > 0 {
 		metadata, err := table.ParseMetadataBytes(alias.RawMetadata)

weed/s3api/s3api_iceberg_credentials.go

@@ -0,0 +1,177 @@
+package s3api
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/iam/sts"
+)
+
+// icebergVendedCredentialsProvider names the origin recorded on sessions the
+// catalog mints, so audit trails can tell them apart from ordinary logins.
+const icebergVendedCredentialsProvider = "iceberg-catalog"
+
+// SetIcebergCredentialRole configures the role the Iceberg catalog assumes when
+// a client asks for vended credentials. Vending stays off until it is set.
+func (s3a *S3ApiServer) SetIcebergCredentialRole(roleArn string, durationSeconds int64) {
+	s3a.icebergCredentialRole = strings.TrimSpace(roleArn)
+	s3a.icebergCredentialDuration = durationSeconds
+}
+
+// VendedCredentials are short-lived S3 credentials scoped to one table.
+type VendedCredentials struct {
+	AccessKeyID     string
+	SecretAccessKey string
+	SessionToken    string
+	Expiration      time.Time
+}
+
+// VendTableCredentials mints credentials that can read and write exactly one
+// table's prefix, for a caller the Iceberg catalog has already authenticated
+// and authorized. Returns nil when no role is configured, which leaves the
+// catalog telling the client to keep using its own credentials.
+func (s3a *S3ApiServer) VendTableCredentials(ctx context.Context, principal, bucket, prefix string) (*VendedCredentials, error) {
+	if s3a.icebergCredentialRole == "" {
+		return nil, nil
+	}
+	if principal == "" {
+		return nil, fmt.Errorf("no principal to vend credentials for")
+	}
+	if bucket == "" {
+		return nil, fmt.Errorf("no bucket to scope credentials to")
+	}
+
+	stsService := s3a.stsService()
+	if stsService == nil {
+		return nil, fmt.Errorf("STS is not configured")
+	}
+
+	policy, err := tablePrefixSessionPolicy(bucket, prefix)
+	if err != nil {
+		return nil, err
+	}
+
+	request := &sts.AssumeRoleForPrincipalRequest{
+		RoleArn:         s3a.icebergCredentialRole,
+		Principal:       principal,
+		RoleSessionName: icebergSessionName(principal, bucket, prefix),
+		ProviderName:    icebergVendedCredentialsProvider,
+		Policy:          &policy,
+	}
+	if s3a.icebergCredentialDuration > 0 {
+		duration := s3a.icebergCredentialDuration
+		request.DurationSeconds = &duration
+	}
+
+	resp, err := stsService.AssumeRoleForPrincipal(ctx, request)
+	if err != nil {
+		return nil, err
+	}
+	if resp == nil || resp.Credentials == nil {
+		return nil, fmt.Errorf("STS returned no credentials")
+	}
+
+	return &VendedCredentials{
+		AccessKeyID:     resp.Credentials.AccessKeyId,
+		SecretAccessKey: resp.Credentials.SecretAccessKey,
+		SessionToken:    resp.Credentials.SessionToken,
+		Expiration:      resp.Credentials.Expiration,
+	}, nil
+}
+
+func (s3a *S3ApiServer) stsService() *sts.STSService {
+	if s3a.iam == nil || s3a.iam.iamIntegration == nil {
+		return nil
+	}
+	provider, ok := s3a.iam.iamIntegration.(IAMManagerProvider)
+	if !ok {
+		return nil
+	}
+	manager := provider.GetIAMManager()
+	if manager == nil {
+		return nil
+	}
+	return manager.GetSTSService()
+}
+
+// tablePrefixSessionPolicy narrows a session to one table's files plus the
+// bucket listing that clients need to resolve them.
+func tablePrefixSessionPolicy(bucket, prefix string) (string, error) {
+	prefix = strings.Trim(prefix, "/")
+	// A resource pattern is matched with wildcards, so a location carrying one
+	// would widen the session to sibling tables. Refuse rather than escape it:
+	// nothing SeaweedFS generates contains these.
+	if strings.ContainsAny(bucket, "*?") || strings.ContainsAny(prefix, "*?") {
+		return "", fmt.Errorf("refusing to scope credentials to a location with wildcards: s3://%s/%s", bucket, prefix)
+	}
+	// Without a prefix there is nothing to scope to, and the session would carry
+	// read and write over every table in the bucket. A table registered at the
+	// bucket root is the only way to get here.
+	if prefix == "" {
+		return "", fmt.Errorf("refusing to scope credentials to a whole bucket: s3://%s", bucket)
+	}
+
+	policy := map[string]interface{}{
+		"Version": "2012-10-17",
+		"Statement": []map[string]interface{}{
+			{
+				"Effect":   "Allow",
+				"Action":   []string{"s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:AbortMultipartUpload", "s3:ListMultipartUploadParts"},
+				"Resource": []string{fmt.Sprintf("arn:aws:s3:::%s/%s/*", bucket, prefix)},
+			},
+			{
+				// Listing is granted on the bucket, so without the condition the
+				// session could enumerate every other table's object names.
+				"Effect":   "Allow",
+				"Action":   []string{"s3:ListBucket"},
+				"Resource": []string{fmt.Sprintf("arn:aws:s3:::%s", bucket)},
+				"Condition": map[string]interface{}{
+					"StringLike": map[string]interface{}{
+						"s3:prefix": []string{prefix, prefix + "/*"},
+					},
+				},
+			},
+			{
+				// GetBucketLocation carries no prefix to condition on, so it sits
+				// in its own statement rather than being denied by one.
+				// ListBucketMultipartUploads is deliberately absent: Iceberg
+				// writers complete and abort by upload id, and granting it either
+				// leaks in-flight keys bucket-wide or breaks on the same missing
+				// prefix.
+				"Effect":   "Allow",
+				"Action":   []string{"s3:GetBucketLocation"},
+				"Resource": []string{fmt.Sprintf("arn:aws:s3:::%s", bucket)},
+			},
+		},
+	}
+
+	encoded, err := json.Marshal(policy)
+	if err != nil {
+		return "", fmt.Errorf("build session policy: %w", err)
+	}
+	return string(encoded), nil
+}
+
+// icebergSessionName builds a session name that identifies the caller and the
+// table in audit logs, within the 64-character AWS limit.
+func icebergSessionName(principal, bucket, prefix string) string {
+	name := fmt.Sprintf("iceberg-%s-%s", principal, strings.ReplaceAll(strings.Trim(prefix, "/"), "/", "-"))
+	if strings.Trim(prefix, "/") == "" {
+		name = fmt.Sprintf("iceberg-%s-%s", principal, bucket)
+	}
+	name = strings.Map(func(r rune) rune {
+		switch {
+		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_':
+			return r
+		default:
+			return '-'
+		}
+	}, name)
+	if len(name) > 64 {
+		name = name[:64]
+	}
+	return name
+}

weed/s3api/s3api_iceberg_credentials_test.go

@@ -0,0 +1,178 @@
+package s3api
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func TestTablePrefixSessionPolicyScopesToTheTable(t *testing.T) {
+	raw, err := tablePrefixSessionPolicy("warehouse", "analytics/events")
+	if err != nil {
+		t.Fatalf("tablePrefixSessionPolicy() error = %v", err)
+	}
+
+	var policy struct {
+		Statement []struct {
+			Effect   string   `json:"Effect"`
+			Action   []string `json:"Action"`
+			Resource []string `json:"Resource"`
+		} `json:"Statement"`
+	}
+	if err := json.Unmarshal([]byte(raw), &policy); err != nil {
+		t.Fatalf("policy is not valid JSON: %v", err)
+	}
+	// objects, conditioned bucket listing, and unconditioned location lookup
+	if len(policy.Statement) != 3 {
+		t.Fatalf("policy has %d statements, want 3", len(policy.Statement))
+	}
+
+	objects := policy.Statement[0]
+	if objects.Resource[0] != "arn:aws:s3:::warehouse/analytics/events/*" {
+		t.Errorf("object resource = %q, want the table prefix only", objects.Resource[0])
+	}
+	if policy.Statement[1].Resource[0] != "arn:aws:s3:::warehouse" {
+		t.Errorf("bucket resource = %q", policy.Statement[1].Resource[0])
+	}
+	for _, statement := range policy.Statement {
+		if statement.Effect != "Allow" {
+			t.Errorf("statement effect = %q, want Allow", statement.Effect)
+		}
+	}
+}
+
+// A table at the bucket root has no prefix to scope to, and vending would hand
+// out read and write over every other table in the bucket.
+func TestTablePrefixSessionPolicyRefusesAWholeBucket(t *testing.T) {
+	if _, err := tablePrefixSessionPolicy("warehouse", ""); err == nil {
+		t.Error("tablePrefixSessionPolicy() error = nil, want a refusal for an empty prefix")
+	}
+	if _, err := tablePrefixSessionPolicy("warehouse", "/"); err == nil {
+		t.Error("a prefix of only separators was accepted")
+	}
+}
+
+// GetBucketLocation carries no prefix, so conditioning it would deny location
+// discovery to every vended credential.
+func TestTablePrefixSessionPolicyLeavesBucketLocationUnconditioned(t *testing.T) {
+	raw, err := tablePrefixSessionPolicy("warehouse", "analytics/events")
+	if err != nil {
+		t.Fatalf("tablePrefixSessionPolicy() error = %v", err)
+	}
+
+	var policy struct {
+		Statement []struct {
+			Action    []string       `json:"Action"`
+			Condition map[string]any `json:"Condition"`
+		} `json:"Statement"`
+	}
+	if err := json.Unmarshal([]byte(raw), &policy); err != nil {
+		t.Fatalf("policy is not valid JSON: %v", err)
+	}
+
+	for _, statement := range policy.Statement {
+		for _, action := range statement.Action {
+			if action == "s3:GetBucketLocation" && statement.Condition != nil {
+				t.Errorf("s3:GetBucketLocation carries a condition it can never satisfy: %v", statement.Condition)
+			}
+			if action == "s3:ListBucketMultipartUploads" {
+				t.Error("s3:ListBucketMultipartUploads is granted; it cannot be scoped by prefix")
+			}
+		}
+	}
+}
+
+func TestIcebergSessionNameIsBounded(t *testing.T) {
+	name := icebergSessionName(strings.Repeat("principal", 10), "warehouse", "ns/table")
+	if len(name) > 64 {
+		t.Errorf("session name is %d chars, want at most 64", len(name))
+	}
+	for _, r := range name {
+		switch {
+		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_':
+		default:
+			t.Errorf("session name %q contains %q, which AWS rejects", name, r)
+		}
+	}
+}
+
+// Vending stays off until an operator names a role, so the catalog keeps
+// telling clients to use their own credentials.
+func TestVendTableCredentialsDisabledWithoutRole(t *testing.T) {
+	s3a := &S3ApiServer{}
+	credentials, err := s3a.VendTableCredentials(context.Background(), "admin", "warehouse", "ns/t")
+	if err != nil {
+		t.Fatalf("VendTableCredentials() error = %v, want nil", err)
+	}
+	if credentials != nil {
+		t.Errorf("VendTableCredentials() = %v, want nil while no role is configured", credentials)
+	}
+}
+
+func TestVendTableCredentialsNeedsSTS(t *testing.T) {
+	s3a := &S3ApiServer{}
+	s3a.SetIcebergCredentialRole("arn:aws:iam::role/IcebergTableAccess", 3600)
+
+	if _, err := s3a.VendTableCredentials(context.Background(), "admin", "warehouse", "ns/t"); err == nil {
+		t.Error("VendTableCredentials() error = nil, want an error when STS is not configured")
+	}
+}
+
+// Listing is granted on the bucket, so the prefix condition is what keeps a
+// vended credential from enumerating other tables.
+func TestTablePrefixSessionPolicyRestrictsListing(t *testing.T) {
+	raw, err := tablePrefixSessionPolicy("warehouse", "analytics/events")
+	if err != nil {
+		t.Fatalf("tablePrefixSessionPolicy() error = %v", err)
+	}
+
+	var policy struct {
+		Statement []struct {
+			Action    []string                       `json:"Action"`
+			Condition map[string]map[string][]string `json:"Condition"`
+		} `json:"Statement"`
+	}
+	if err := json.Unmarshal([]byte(raw), &policy); err != nil {
+		t.Fatalf("policy is not valid JSON: %v", err)
+	}
+
+	var bucketStatement *struct {
+		Action    []string                       `json:"Action"`
+		Condition map[string]map[string][]string `json:"Condition"`
+	}
+	for i := range policy.Statement {
+		for _, action := range policy.Statement[i].Action {
+			if action == "s3:ListBucket" {
+				bucketStatement = &policy.Statement[i]
+			}
+		}
+	}
+	if bucketStatement == nil {
+		t.Fatal("no statement grants s3:ListBucket")
+	}
+
+	prefixes := bucketStatement.Condition["StringLike"]["s3:prefix"]
+	want := map[string]bool{"analytics/events": true, "analytics/events/*": true}
+	if len(prefixes) != len(want) {
+		t.Fatalf("s3:prefix = %v, want %v", prefixes, want)
+	}
+	for _, prefix := range prefixes {
+		if !want[prefix] {
+			t.Errorf("s3:prefix contains %q, which is outside the table", prefix)
+		}
+	}
+}
+
+// A location carrying a wildcard would widen the resource pattern to sibling
+// tables, so it is refused instead of vended.
+func TestTablePrefixSessionPolicyRefusesWildcards(t *testing.T) {
+	for _, prefix := range []string{"analytics/*", "analytics/ev?nts", "*"} {
+		if _, err := tablePrefixSessionPolicy("warehouse", prefix); err == nil {
+			t.Errorf("tablePrefixSessionPolicy(%q) error = nil, want a refusal", prefix)
+		}
+	}
+	if _, err := tablePrefixSessionPolicy("ware*house", "analytics/events"); err == nil {
+		t.Error("a wildcard in the bucket name was accepted")
+	}
+}

weed/s3api/s3api_server.go

@@ -98,8 +98,12 @@ type S3ApiServer struct {
 	inFlightDataLimitCond *sync.Cond
 	embeddedIam           *EmbeddedIamApi // Embedded IAM API server (when enabled)
 	stsHandlers           *STSHandlers    // STS HTTP handlers for AssumeRoleWithWebIdentity
-	cipher                bool            // encrypt data on volume servers
-	newObjectWriteLock    func(bucket, object string) objectWriteLock
+	// icebergCredentialRole is the role the Iceberg catalog assumes to vend
+	// table-scoped credentials; empty leaves vending off.
+	icebergCredentialRole     string
+	icebergCredentialDuration int64
+	cipher                    bool // encrypt data on volume servers
+	newObjectWriteLock        func(bucket, object string) objectWriteLock
 	// objectWriteLockClient resolves a key's owner filer for route-by-key.
 	objectWriteLockClient *cluster.LockClient
 	// unreachableOwners holds owners (pb.ServerAddress -> expiry time.Time) whose