From 7522e17b6d76b436d704a580661b9a884f3c8b6a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 16 Aug 2026 12:57:12 -0700 Subject: [PATCH] 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 | 2 + weed/command/s3.go | 31 +++ weed/command/server.go | 2 + weed/iam/sts/credential_prefix_test.go | 32 ++++ weed/iam/sts/sts_service.go | 130 ++++++++++--- weed/s3api/iceberg/handlers_table.go | 98 ++++++++-- weed/s3api/iceberg/handlers_view.go | 6 +- weed/s3api/iceberg/handlers_view_update.go | 3 +- .../iceberg/iceberg_access_delegation_test.go | 10 +- .../iceberg_credential_vending_test.go | 111 +++++++++++ weed/s3api/iceberg/iceberg_issue_9103_test.go | 4 +- weed/s3api/iceberg/server.go | 24 +++ weed/s3api/iceberg/types.go | 36 ++-- weed/s3api/s3api_iceberg_credentials.go | 177 +++++++++++++++++ weed/s3api/s3api_iceberg_credentials_test.go | 178 ++++++++++++++++++ weed/s3api/s3api_server.go | 8 +- 16 files changed, 780 insertions(+), 72 deletions(-) create mode 100644 weed/s3api/iceberg/iceberg_credential_vending_test.go create mode 100644 weed/s3api/s3api_iceberg_credentials.go create mode 100644 weed/s3api/s3api_iceberg_credentials_test.go diff --git a/weed/command/mini.go b/weed/command/mini.go index ec1132ae6..c02a6c274 100644 --- a/weed/command/mini.go +++ b/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") diff --git a/weed/command/s3.go b/weed/command/s3.go index 6d1037071..7d7c4ad7c 100644 --- a/weed/command/s3.go +++ b/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 +} diff --git a/weed/command/server.go b/weed/command/server.go index 71cc38134..0eaca136d 100644 --- a/weed/command/server.go +++ b/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") diff --git a/weed/iam/sts/credential_prefix_test.go b/weed/iam/sts/credential_prefix_test.go index b3bd0d331..bc44801e5 100644 --- a/weed/iam/sts/credential_prefix_test.go +++ b/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) + } + } +} diff --git a/weed/iam/sts/sts_service.go b/weed/iam/sts/sts_service.go index e7ca17262..99aacba5d 100644 --- a/weed/iam/sts/sts_service.go +++ b/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 diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 8929693dc..c31d73151 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/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. diff --git a/weed/s3api/iceberg/handlers_view.go b/weed/s3api/iceberg/handlers_view.go index 108ca5226..543655cc2 100644 --- a/weed/s3api/iceberg/handlers_view.go +++ b/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 } diff --git a/weed/s3api/iceberg/handlers_view_update.go b/weed/s3api/iceberg/handlers_view_update.go index 9ee23065c..45c14d249 100644 --- a/weed/s3api/iceberg/handlers_view_update.go +++ b/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 } diff --git a/weed/s3api/iceberg/iceberg_access_delegation_test.go b/weed/s3api/iceberg/iceberg_access_delegation_test.go index 51a1931a4..3e024a959 100644 --- a/weed/s3api/iceberg/iceberg_access_delegation_test.go +++ b/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) } } diff --git a/weed/s3api/iceberg/iceberg_credential_vending_test.go b/weed/s3api/iceberg/iceberg_credential_vending_test.go new file mode 100644 index 000000000..aee6a56a7 --- /dev/null +++ b/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") + } +} diff --git a/weed/s3api/iceberg/iceberg_issue_9103_test.go b/weed/s3api/iceberg/iceberg_issue_9103_test.go index 760dd4871..e321fa2af 100644 --- a/weed/s3api/iceberg/iceberg_issue_9103_test.go +++ b/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") } diff --git a/weed/s3api/iceberg/server.go b/weed/s3api/iceberg/server.go index 9a3709f59..3a08320a3 100644 --- a/weed/s3api/iceberg/server.go +++ b/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 diff --git a/weed/s3api/iceberg/types.go b/weed/s3api/iceberg/types.go index 43f37913f..e66c209e4 100644 --- a/weed/s3api/iceberg/types.go +++ b/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) diff --git a/weed/s3api/s3api_iceberg_credentials.go b/weed/s3api/s3api_iceberg_credentials.go new file mode 100644 index 000000000..16b51a718 --- /dev/null +++ b/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 +} diff --git a/weed/s3api/s3api_iceberg_credentials_test.go b/weed/s3api/s3api_iceberg_credentials_test.go new file mode 100644 index 000000000..33c5638f5 --- /dev/null +++ b/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") + } +} diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 7b09e0855..22bef931b 100644 --- a/weed/s3api/s3api_server.go +++ b/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