s3tables: honor resource policy Deny on remaining operations (#11825)

* s3tables: honor a table policy Deny for maintenance operations

authorizeMaintenanceTable evaluated only the bucket policy, so a Deny in the table policy never vetoed PutTableMaintenanceConfiguration, GetTableMaintenanceConfiguration or GetTableMaintenanceJobStatus even though they act on the table ARN. The stored table policy is now checked for a matching Deny first via deniedByPolicy, a veto helper that keeps the admin exemption, and resource and bucket tags are loaded into the policy context so tag-conditioned Denies evaluate. readBucketPolicy is renamed to readPolicy and reused for the table path.

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

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

* s3tables: honor the stored policy's Deny on policy management operations

PutTablePolicy, GetTablePolicy and DeleteTablePolicy only evaluated the bucket policy, so a Deny in the table's own policy was ignored and a bucket-level Allow could strip or replace the policy guarding a table. The existing table policy is now loaded and a matching Deny vetoes the call. PutTableBucketPolicy likewise consults the existing bucket policy for a Deny before replacing it. Resource and bucket tags populate the policy context so tag-conditioned Denies evaluate, and policy writes assert the policy authorized against is still in force, returning a conflict when it changed mid-request.

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

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

* s3tables: apply a table policy Deny to tag operations

TagResource, ListTagsForResource and UntagResource evaluated only the bucket policy; a Deny in the table's own policy is now honored when the target ARN is a table. ListTagsForResource and UntagResource also returned early when the resource carried no tags attribute, skipping the permission check entirely; the check now always runs. Policy evaluation uses the canonical resource ARN rebuilt from stored metadata, so a request ARN with a forged account or region cannot dodge the Deny.

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

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

* s3tables: regression tests for resource policy denies

Cover the case where a bucket policy grants an identity s3tables:* while the table policy denies it: the policy, tag and maintenance operations must stay denied so the protection cannot be stripped. Also cover a bucket policy that denies PutTableBucketPolicy under the open fallback, and a tag-conditioned Deny, which vetoes only while the stored tag matches.

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

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

---------

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

weed/s3api/s3tables/handler_maintenance.go

@@ -339,7 +339,7 @@ func (h *S3TablesHandler) authorizeMaintenanceBucket(r *http.Request, filerClien
 		if err := json.Unmarshal(data, &bucketMetadata); err != nil {
 			return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
 		}
-		bucketPolicy, err = h.readBucketPolicy(r, client, bucketPath)
+		bucketPolicy, err = h.readPolicy(r, client, bucketPath)
 		return err
 	})
 	if err != nil {
@@ -368,7 +368,8 @@ func (h *S3TablesHandler) authorizeMaintenanceTable(r *http.Request, filerClient
 	bucketPath := GetTableBucketPath(bucketName)
 
 	var metadata tableMetadataInternal
-	var bucketPolicy string
+	var bucketPolicy, tablePolicy string
+	var bucketTags, tableTags map[string]string
 	err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
 		if err != nil {
@@ -377,7 +378,16 @@ func (h *S3TablesHandler) authorizeMaintenanceTable(r *http.Request, filerClient
 		if err := json.Unmarshal(data, &metadata); err != nil {
 			return fmt.Errorf("failed to unmarshal table metadata: %w", err)
 		}
-		bucketPolicy, err = h.readBucketPolicy(r, client, bucketPath)
+		if bucketPolicy, err = h.readPolicy(r, client, bucketPath); err != nil {
+			return err
+		}
+		if bucketTags, err = h.readTags(r.Context(), client, bucketPath); err != nil {
+			return err
+		}
+		if tablePolicy, err = h.readPolicy(r, client, tablePath); err != nil {
+			return err
+		}
+		tableTags, err = h.readTags(r.Context(), client, tablePath)
 		return err
 	})
 	if err != nil {
@@ -386,30 +396,34 @@ func (h *S3TablesHandler) authorizeMaintenanceTable(r *http.Request, filerClient
 
 	tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName)
 	principal := h.getAccountID(r)
-	if !CheckPermissionWithContext(operation, principal, metadata.OwnerAccountID, bucketPolicy, tableARN, &PolicyContext{
+	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
 		TableName:        tableName,
+		TableBucketTags:  bucketTags,
+		ResourceTags:     tableTags,
 		IdentityActions:  getIdentityActions(r),
 		DefaultAllow:     h.defaultAllowFor(r),
-	}) {
+	}
+	if deniedByPolicy(operation, principal, tablePolicy, tableARN, ctx) ||
+		!CheckPermissionWithContext(operation, principal, metadata.OwnerAccountID, bucketPolicy, tableARN, ctx) {
 		return "", NewAuthError(operation, principal, "not authorized to "+operation)
 	}
 
 	return tableARN, nil
 }
 
-// readBucketPolicy returns the bucket policy, or an empty string when the
-// bucket has none.
-func (h *S3TablesHandler) readBucketPolicy(r *http.Request, client filer_pb.SeaweedFilerClient, bucketPath string) (string, error) {
-	data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
+// readPolicy returns the policy stored on path, or an empty string when the
+// resource has none.
+func (h *S3TablesHandler) readPolicy(r *http.Request, client filer_pb.SeaweedFilerClient, path string) (string, error) {
+	data, err := h.getExtendedAttribute(r.Context(), client, path, ExtendedKeyPolicy)
 	if err != nil {
 		if errors.Is(err, ErrAttributeNotFound) {
 			return "", nil
 		}
-		return "", fmt.Errorf("failed to read bucket policy: %w", err)
+		return "", fmt.Errorf("failed to read policy: %w", err)
 	}
 	return string(data), nil
 }

weed/s3api/s3tables/handler_policy.go

@@ -1,6 +1,7 @@
 package s3tables
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -68,6 +69,8 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h
 	// Check if bucket exists and get metadata for ownership check
 	bucketPath := GetTableBucketPath(bucketName)
 	var bucketMetadata tableBucketMetadata
+	var bucketPolicy string
+	var bucketTags map[string]string
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
 		if err != nil {
@@ -76,7 +79,11 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h
 		if err := json.Unmarshal(data, &bucketMetadata); err != nil {
 			return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
 		}
-		return nil
+		if bucketPolicy, err = h.readPolicy(r, client, bucketPath); err != nil {
+			return err
+		}
+		bucketTags, err = h.readTags(r.Context(), client, bucketPath)
+		return err
 	})
 
 	if err != nil {
@@ -91,24 +98,39 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h
 	bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
 	principal := h.getAccountID(r)
 	identityActions := getIdentityActions(r)
-	if !CheckPermissionWithContext("PutTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, "", bucketARN, &PolicyContext{
+	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
+		TableBucketTags:  bucketTags,
+		ResourceTags:     bucketTags,
 		IdentityActions:  identityActions,
 		DefaultAllow:     h.defaultAllowFor(r),
-	}) {
+	}
+	if deniedByPolicy("PutTableBucketPolicy", principal, bucketPolicy, bucketARN, ctx) ||
+		!CheckPermissionWithContext("PutTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, "", bucketARN, ctx) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to put table bucket policy")
 		return NewAuthError("PutTableBucketPolicy", principal, "not authorized to put table bucket policy")
 	}
 
-	// Write policy
+	// Conditional on the policy this request was authorized against: a Deny
+	// landing mid-request sends the caller back through authorization.
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return h.setExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy, []byte(req.ResourcePolicy))
+		return h.mutateEntryExtended(r.Context(), client, bucketPath, func(extended map[string][]byte) error {
+			if !bytes.Equal(extended[ExtendedKeyPolicy], []byte(bucketPolicy)) {
+				return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy)
+			}
+			extended[ExtendedKeyPolicy] = []byte(req.ResourcePolicy)
+			return nil
+		})
 	})
 
 	if err != nil {
-		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to put table bucket policy")
+		if errors.Is(err, ErrConcurrentUpdate) {
+			h.writeError(w, http.StatusConflict, ErrCodeConflict, "table bucket policy changed concurrently, retry the request")
+		} else {
+			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to put table bucket policy")
+		}
 		return err
 	}
 
@@ -314,7 +336,8 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re
 	bucketPath := GetTableBucketPath(bucketName)
 
 	var metadata tableMetadataInternal
-	var bucketPolicy string
+	var bucketPolicy, tablePolicy string
+	var bucketTags, tableTags map[string]string
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
 		if err != nil {
@@ -324,18 +347,17 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re
 			return fmt.Errorf("failed to unmarshal table metadata: %w", err)
 		}
 
-		// Fetch bucket policy if it exists
-		policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
-		if err != nil {
-			if !errors.Is(err, ErrAttributeNotFound) {
-				return fmt.Errorf("failed to read bucket policy: %w", err)
-			}
-			// Policy not found is not an error; bucketPolicy remains empty
-		} else {
-			bucketPolicy = string(policyData)
+		if bucketPolicy, err = h.readPolicy(r, client, bucketPath); err != nil {
+			return err
 		}
-
-		return nil
+		if bucketTags, err = h.readTags(r.Context(), client, bucketPath); err != nil {
+			return err
+		}
+		if tablePolicy, err = h.readPolicy(r, client, tablePath); err != nil {
+			return err
+		}
+		tableTags, err = h.readTags(r.Context(), client, tablePath)
+		return err
 	})
 
 	if err != nil {
@@ -350,26 +372,39 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re
 	tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName)
 	principal := h.getAccountID(r)
 	identityActions := getIdentityActions(r)
-	if !CheckPermissionWithContext("PutTablePolicy", principal, metadata.OwnerAccountID, bucketPolicy, tableARN, &PolicyContext{
+	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
 		TableName:        tableName,
+		TableBucketTags:  bucketTags,
+		ResourceTags:     tableTags,
 		IdentityActions:  identityActions,
 		DefaultAllow:     h.defaultAllowFor(r),
-	}) {
+	}
+	if deniedByPolicy("PutTablePolicy", principal, tablePolicy, tableARN, ctx) ||
+		!CheckPermissionWithContext("PutTablePolicy", principal, metadata.OwnerAccountID, bucketPolicy, tableARN, ctx) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to put table policy")
 		return NewAuthError("PutTablePolicy", principal, "not authorized to put table policy")
 	}
 
-	// Write policy
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyPolicy, []byte(req.ResourcePolicy))
+		return h.mutateEntryExtended(r.Context(), client, tablePath, func(extended map[string][]byte) error {
+			if !bytes.Equal(extended[ExtendedKeyPolicy], []byte(tablePolicy)) {
+				return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy)
+			}
+			extended[ExtendedKeyPolicy] = []byte(req.ResourcePolicy)
+			return nil
+		})
 	})
 
 	if err != nil {
-		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to put table policy")
+		if errors.Is(err, ErrConcurrentUpdate) {
+			h.writeError(w, http.StatusConflict, ErrCodeConflict, "table policy changed concurrently, retry the request")
+		} else {
+			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to put table policy")
+		}
 		return err
 	}
 
@@ -413,6 +448,7 @@ func (h *S3TablesHandler) handleGetTablePolicy(w http.ResponseWriter, r *http.Re
 	var policy []byte
 	var metadata tableMetadataInternal
 	var bucketPolicy string
+	var bucketTags, tableTags map[string]string
 
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		// Get metadata for ownership check
@@ -430,18 +466,14 @@ func (h *S3TablesHandler) handleGetTablePolicy(w http.ResponseWriter, r *http.Re
 			return err
 		}
 
-		// Fetch bucket policy if it exists
-		policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
-		if err != nil {
-			if !errors.Is(err, ErrAttributeNotFound) {
-				return fmt.Errorf("failed to read bucket policy: %w", err)
-			}
-			// Policy not found is not an error; bucketPolicy remains empty
-		} else {
-			bucketPolicy = string(policyData)
+		if bucketPolicy, err = h.readPolicy(r, client, bucketPath); err != nil {
+			return err
 		}
-
-		return nil
+		if bucketTags, err = h.readTags(r.Context(), client, bucketPath); err != nil {
+			return err
+		}
+		tableTags, err = h.readTags(r.Context(), client, tablePath)
+		return err
 	})
 
 	if err != nil {
@@ -460,15 +492,19 @@ func (h *S3TablesHandler) handleGetTablePolicy(w http.ResponseWriter, r *http.Re
 	tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName)
 	principal := h.getAccountID(r)
 	identityActions := getIdentityActions(r)
-	if !CheckPermissionWithContext("GetTablePolicy", principal, metadata.OwnerAccountID, bucketPolicy, tableARN, &PolicyContext{
+	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
 		TableName:        tableName,
+		TableBucketTags:  bucketTags,
+		ResourceTags:     tableTags,
 		IdentityActions:  identityActions,
 		DefaultAllow:     h.defaultAllowFor(r),
-	}) {
+	}
+	if deniedByPolicy("GetTablePolicy", principal, string(policy), tableARN, ctx) ||
+		!CheckPermissionWithContext("GetTablePolicy", principal, metadata.OwnerAccountID, bucketPolicy, tableARN, ctx) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to get table policy")
 		return NewAuthError("GetTablePolicy", principal, "not authorized to get table policy")
 	}
@@ -517,7 +553,8 @@ func (h *S3TablesHandler) handleDeleteTablePolicy(w http.ResponseWriter, r *http
 
 	// Check if table exists
 	var metadata tableMetadataInternal
-	var bucketPolicy string
+	var bucketPolicy, tablePolicy string
+	var bucketTags, tableTags map[string]string
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
 		if err != nil {
@@ -527,18 +564,17 @@ func (h *S3TablesHandler) handleDeleteTablePolicy(w http.ResponseWriter, r *http
 			return fmt.Errorf("failed to unmarshal table metadata: %w", err)
 		}
 
-		// Fetch bucket policy if it exists
-		policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
-		if err != nil {
-			if !errors.Is(err, ErrAttributeNotFound) {
-				return fmt.Errorf("failed to read bucket policy: %w", err)
-			}
-			// Policy not found is not an error; bucketPolicy remains empty
-		} else {
-			bucketPolicy = string(policyData)
+		if bucketPolicy, err = h.readPolicy(r, client, bucketPath); err != nil {
+			return err
 		}
-
-		return nil
+		if bucketTags, err = h.readTags(r.Context(), client, bucketPath); err != nil {
+			return err
+		}
+		if tablePolicy, err = h.readPolicy(r, client, tablePath); err != nil {
+			return err
+		}
+		tableTags, err = h.readTags(r.Context(), client, tablePath)
+		return err
 	})
 	if err != nil {
 		if errors.Is(err, filer_pb.ErrNotFound) {
@@ -552,25 +588,35 @@ func (h *S3TablesHandler) handleDeleteTablePolicy(w http.ResponseWriter, r *http
 	tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName)
 	principal := h.getAccountID(r)
 	identityActions := getIdentityActions(r)
-	if !CheckPermissionWithContext("DeleteTablePolicy", principal, metadata.OwnerAccountID, bucketPolicy, tableARN, &PolicyContext{
+	ctx := &PolicyContext{
 		TrustedIssuers:   h.trustedIssuers(r),
 		IsLocalAccountID: h.localAccountIDs,
 		TableBucketName:  bucketName,
 		Namespace:        namespaceName,
 		TableName:        tableName,
+		TableBucketTags:  bucketTags,
+		ResourceTags:     tableTags,
 		IdentityActions:  identityActions,
 		DefaultAllow:     h.defaultAllowFor(r),
-	}) {
+	}
+	if deniedByPolicy("DeleteTablePolicy", principal, tablePolicy, tableARN, ctx) ||
+		!CheckPermissionWithContext("DeleteTablePolicy", principal, metadata.OwnerAccountID, bucketPolicy, tableARN, ctx) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to delete table policy")
 		return NewAuthError("DeleteTablePolicy", principal, "not authorized to delete table policy")
 	}
 
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return h.deleteExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyPolicy)
+		return h.removeExtendedAttributesIf(r.Context(), client, tablePath,
+			map[string][]byte{ExtendedKeyPolicy: []byte(tablePolicy)},
+			ExtendedKeyPolicy)
 	})
 
 	if err != nil && !errors.Is(err, ErrAttributeNotFound) {
-		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table policy")
+		if errors.Is(err, ErrConcurrentUpdate) {
+			h.writeError(w, http.StatusConflict, ErrCodeConflict, "table policy changed concurrently, retry the request")
+		} else {
+			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table policy")
+		}
 		return err
 	}
 
@@ -605,8 +651,9 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
 
 	// Read existing tags and merge, AND check permissions based on metadata ownership
 	existingTags := make(map[string]string)
-	var bucketPolicy string
+	var bucketPolicy, tablePolicy string
 	var bucketTags map[string]string
+	var tableNamespace, tableName string
 	requestTagKeys := mapKeys(req.Tags)
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		// Read metadata for ownership check
@@ -623,14 +670,8 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
 		// Fetch bucket policy if we have a bucket name
 		if bucketName != "" {
 			bucketPath := GetTableBucketPath(bucketName)
-			policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
-			if err != nil {
-				if !errors.Is(err, ErrAttributeNotFound) {
-					return fmt.Errorf("failed to read bucket policy: %w", err)
-				}
-				// Policy not found is not an error; bucketPolicy remains empty
-			} else {
-				bucketPolicy = string(policyData)
+			if bucketPolicy, err = h.readPolicy(r, client, bucketPath); err != nil {
+				return err
 			}
 			bucketTags, err = h.readTags(r.Context(), client, bucketPath)
 			if err != nil {
@@ -638,6 +679,15 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
 			}
 		}
 
+		if rType == ResourceTypeTable {
+			if _, tableNamespace, tableName, err = parseTableFromARN(req.ResourceARN); err != nil {
+				return err
+			}
+			if tablePolicy, err = h.readPolicy(r, client, resourcePath); err != nil {
+				return err
+			}
+		}
+
 		// Read existing tags
 		data, err = h.getExtendedAttribute(r.Context(), client, resourcePath, extendedKey)
 		if err != nil {
@@ -648,20 +698,29 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
 			return err
 		}
 
-		resourceARN := req.ResourceARN
+		var resourceARN string
+		if rType == ResourceTypeTable {
+			resourceARN = h.generateTableARN(ownerAccountID, bucketName, tableNamespace+"/"+tableName)
+		} else {
+			resourceARN = h.generateTableBucketARN(ownerAccountID, bucketName)
+		}
 		principal := h.getAccountID(r)
 		identityActions := getIdentityActions(r)
-		if !CheckPermissionWithContext("TagResource", principal, ownerAccountID, bucketPolicy, resourceARN, &PolicyContext{
+		ctx := &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
+			Namespace:        tableNamespace,
+			TableName:        tableName,
 			TableBucketTags:  bucketTags,
 			RequestTags:      req.Tags,
 			TagKeys:          requestTagKeys,
 			ResourceTags:     existingTags,
 			IdentityActions:  identityActions,
 			DefaultAllow:     h.defaultAllowFor(r),
-		}) {
+		}
+		if deniedByPolicy("TagResource", principal, tablePolicy, resourceARN, ctx) ||
+			!CheckPermissionWithContext("TagResource", principal, ownerAccountID, bucketPolicy, resourceARN, ctx) {
 			return NewAuthError("TagResource", principal, "not authorized to tag resource")
 		}
 		return nil
@@ -727,8 +786,9 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht
 	}
 
 	tags := make(map[string]string)
-	var bucketPolicy string
+	var bucketPolicy, tablePolicy string
 	var bucketTags map[string]string
+	var tableNamespace, tableName string
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		// Read metadata for ownership check
 		data, err := h.getExtendedAttribute(r.Context(), client, resourcePath, ExtendedKeyMetadata)
@@ -744,14 +804,8 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht
 		// Fetch bucket policy if we have a bucket name
 		if bucketName != "" {
 			bucketPath := GetTableBucketPath(bucketName)
-			policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
-			if err != nil {
-				if !errors.Is(err, ErrAttributeNotFound) {
-					return fmt.Errorf("failed to read bucket policy: %w", err)
-				}
-				// Policy not found is not an error; bucketPolicy remains empty
-			} else {
-				bucketPolicy = string(policyData)
+			if bucketPolicy, err = h.readPolicy(r, client, bucketPath); err != nil {
+				return err
 			}
 			bucketTags, err = h.readTags(r.Context(), client, bucketPath)
 			if err != nil {
@@ -759,29 +813,45 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht
 			}
 		}
 
+		if rType == ResourceTypeTable {
+			if _, tableNamespace, tableName, err = parseTableFromARN(req.ResourceARN); err != nil {
+				return err
+			}
+			if tablePolicy, err = h.readPolicy(r, client, resourcePath); err != nil {
+				return err
+			}
+		}
+
 		data, err = h.getExtendedAttribute(r.Context(), client, resourcePath, extendedKey)
 		if err != nil {
-			if errors.Is(err, ErrAttributeNotFound) {
-				return nil // No tags is not an error.
+			if !errors.Is(err, ErrAttributeNotFound) {
+				return err
 			}
-			return err // Propagate other errors.
-		}
-		if err := json.Unmarshal(data, &tags); err != nil {
+		} else if err := json.Unmarshal(data, &tags); err != nil {
 			return err
 		}
 
-		resourceARN := req.ResourceARN
+		var resourceARN string
+		if rType == ResourceTypeTable {
+			resourceARN = h.generateTableARN(ownerAccountID, bucketName, tableNamespace+"/"+tableName)
+		} else {
+			resourceARN = h.generateTableBucketARN(ownerAccountID, bucketName)
+		}
 		principal := h.getAccountID(r)
 		identityActions := getIdentityActions(r)
-		if !CheckPermissionWithContext("ListTagsForResource", principal, ownerAccountID, bucketPolicy, resourceARN, &PolicyContext{
+		ctx := &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
+			Namespace:        tableNamespace,
+			TableName:        tableName,
 			TableBucketTags:  bucketTags,
 			ResourceTags:     tags,
 			IdentityActions:  identityActions,
 			DefaultAllow:     h.defaultAllowFor(r),
-		}) {
+		}
+		if deniedByPolicy("ListTagsForResource", principal, tablePolicy, resourceARN, ctx) ||
+			!CheckPermissionWithContext("ListTagsForResource", principal, ownerAccountID, bucketPolicy, resourceARN, ctx) {
 			return NewAuthError("ListTagsForResource", principal, "not authorized to list tags for resource")
 		}
 		return nil
@@ -836,8 +906,9 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
 
 	// Read existing tags, check permission
 	tags := make(map[string]string)
-	var bucketPolicy string
+	var bucketPolicy, tablePolicy string
 	var bucketTags map[string]string
+	var tableNamespace, tableName string
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		// Read metadata for ownership check
 		data, err := h.getExtendedAttribute(r.Context(), client, resourcePath, ExtendedKeyMetadata)
@@ -853,14 +924,8 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
 		// Fetch bucket policy if we have a bucket name
 		if bucketName != "" {
 			bucketPath := GetTableBucketPath(bucketName)
-			policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
-			if err != nil {
-				if !errors.Is(err, ErrAttributeNotFound) {
-					return fmt.Errorf("failed to read bucket policy: %w", err)
-				}
-				// Policy not found is not an error; bucketPolicy remains empty
-			} else {
-				bucketPolicy = string(policyData)
+			if bucketPolicy, err = h.readPolicy(r, client, bucketPath); err != nil {
+				return err
 			}
 			bucketTags, err = h.readTags(r.Context(), client, bucketPath)
 			if err != nil {
@@ -868,30 +933,46 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
 			}
 		}
 
+		if rType == ResourceTypeTable {
+			if _, tableNamespace, tableName, err = parseTableFromARN(req.ResourceARN); err != nil {
+				return err
+			}
+			if tablePolicy, err = h.readPolicy(r, client, resourcePath); err != nil {
+				return err
+			}
+		}
+
 		data, err = h.getExtendedAttribute(r.Context(), client, resourcePath, extendedKey)
 		if err != nil {
-			if errors.Is(err, ErrAttributeNotFound) {
-				return nil
+			if !errors.Is(err, ErrAttributeNotFound) {
+				return err
 			}
-			return err
-		}
-		if err := json.Unmarshal(data, &tags); err != nil {
+		} else if err := json.Unmarshal(data, &tags); err != nil {
 			return err
 		}
 
-		resourceARN := req.ResourceARN
+		var resourceARN string
+		if rType == ResourceTypeTable {
+			resourceARN = h.generateTableARN(ownerAccountID, bucketName, tableNamespace+"/"+tableName)
+		} else {
+			resourceARN = h.generateTableBucketARN(ownerAccountID, bucketName)
+		}
 		principal := h.getAccountID(r)
 		identityActions := getIdentityActions(r)
-		if !CheckPermissionWithContext("UntagResource", principal, ownerAccountID, bucketPolicy, resourceARN, &PolicyContext{
+		ctx := &PolicyContext{
 			TrustedIssuers:   h.trustedIssuers(r),
 			IsLocalAccountID: h.localAccountIDs,
 			TableBucketName:  bucketName,
+			Namespace:        tableNamespace,
+			TableName:        tableName,
 			TableBucketTags:  bucketTags,
 			TagKeys:          req.TagKeys,
 			ResourceTags:     tags,
 			IdentityActions:  identityActions,
 			DefaultAllow:     h.defaultAllowFor(r),
-		}) {
+		}
+		if deniedByPolicy("UntagResource", principal, tablePolicy, resourceARN, ctx) ||
+			!CheckPermissionWithContext("UntagResource", principal, ownerAccountID, bucketPolicy, resourceARN, ctx) {
 			return NewAuthError("UntagResource", principal, "not authorized to untag resource")
 		}
 		return nil

weed/s3api/s3tables/handler_policy_deny_test.go

@@ -0,0 +1,218 @@
+package s3tables
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"path"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const policyDenyBucket = "denybkt"
+const policyDenyWriter = "writer"
+
+func policyDenyBucketARN(t *testing.T) string {
+	t.Helper()
+	arn, err := BuildBucketARN(DefaultRegion, DefaultAccountID, policyDenyBucket)
+	require.NoError(t, err)
+	return arn
+}
+
+func policyDenyTableARN(t *testing.T) string {
+	t.Helper()
+	arn, err := BuildTableARN(DefaultRegion, DefaultAccountID, policyDenyBucket, "ns", "t")
+	require.NoError(t, err)
+	return arn
+}
+
+// startPolicyDenyManager seeds a bucket/namespace/table where the bucket policy
+// grants writer s3tables:* while the table policy denies writer everything.
+func startPolicyDenyManager(t *testing.T) (*s3tablestest.MemFiler, *Manager) {
+	t.Helper()
+	fs := s3tablestest.Start(t)
+
+	bucketMeta, _ := json.Marshal(tableBucketMetadata{Name: policyDenyBucket, OwnerAccountID: DefaultAccountID})
+	bucketAllow := fmt.Sprintf(`{"Statement":[{"Effect":"Allow","Principal":"%s","Action":"s3tables:*","Resource":"*"}]}`, policyDenyWriter)
+	fs.Put(TablesPath, policyDenyBucket, map[string][]byte{
+		ExtendedKeyTableBucket: []byte("{}"),
+		ExtendedKeyMetadata:    bucketMeta,
+		ExtendedKeyPolicy:      []byte(bucketAllow),
+	})
+
+	nsMeta, _ := json.Marshal(namespaceMetadata{Namespace: []string{"ns"}, OwnerAccountID: DefaultAccountID})
+	fs.Put(GetTableBucketPath(policyDenyBucket), "ns", map[string][]byte{ExtendedKeyMetadata: nsMeta})
+
+	tableMeta, _ := json.Marshal(tableMetadataInternal{
+		Name:             "t",
+		Namespace:        "ns",
+		Format:           "ICEBERG",
+		OwnerAccountID:   DefaultAccountID,
+		MetadataVersion:  3,
+		MetadataLocation: "s3://" + policyDenyBucket + "/ns/t/metadata/v3.metadata.json",
+	})
+	tableDeny := fmt.Sprintf(`{"Statement":[{"Effect":"Deny","Principal":"%s","Action":"s3tables:*","Resource":%q}]}`, policyDenyWriter, policyDenyTableARN(t))
+	fs.Put(GetNamespacePath(policyDenyBucket, "ns"), "t", map[string][]byte{
+		ExtendedKeyMetadata:        tableMeta,
+		ExtendedKeyMetadataVersion: []byte("3"),
+		ExtendedKeyPolicy:          []byte(tableDeny),
+	})
+
+	tablePath := GetTablePath(policyDenyBucket, "ns", "t")
+	fs.Put(tablePath, "metadata", nil)
+	fs.Put(tablePath, "data", nil)
+	fs.Put(path.Join(tablePath, "metadata"), "v3.metadata.json", nil)
+
+	return fs, NewManager()
+}
+
+func runPolicyOp(t *testing.T, m *Manager, fs *s3tablestest.MemFiler, op string, req interface{}, identity string) error {
+	t.Helper()
+	return m.Execute(context.Background(), NewManagerClient(fs.Client), op, req, nil, identity)
+}
+
+func deniedTableReq(t *testing.T) (string, string, string) {
+	return policyDenyBucketARN(t), "ns", "t"
+}
+
+// A table policy Deny must hold even when the bucket policy allows the same
+// action: the writer may not strip or replace the policy guarding the table.
+func TestTablePolicyDenyProtectsPolicyOps(t *testing.T) {
+	fs, m := startPolicyDenyManager(t)
+	bucketARN, ns, name := deniedTableReq(t)
+
+	require.Error(t, runPolicyOp(t, m, fs, "GetTablePolicy", &GetTablePolicyRequest{
+		TableBucketARN: bucketARN, Namespace: []string{ns}, Name: name,
+	}, policyDenyWriter), "writer must not read the table policy")
+
+	require.Error(t, runPolicyOp(t, m, fs, "PutTablePolicy", &PutTablePolicyRequest{
+		TableBucketARN: bucketARN, Namespace: []string{ns}, Name: name,
+		ResourcePolicy: `{"Statement":[{"Effect":"Allow","Principal":"writer","Action":"s3tables:*","Resource":"*"}]}`,
+	}, policyDenyWriter), "writer must not replace the table policy")
+
+	require.Error(t, runPolicyOp(t, m, fs, "DeleteTablePolicy", &DeleteTablePolicyRequest{
+		TableBucketARN: bucketARN, Namespace: []string{ns}, Name: name,
+	}, policyDenyWriter), "writer must not delete the table policy")
+
+	// The deny still holds, so the table cannot be read or deleted.
+	require.Error(t, runPolicyOp(t, m, fs, "GetTable", &GetTableRequest{
+		TableBucketARN: bucketARN, Namespace: []string{ns}, Name: name,
+	}, policyDenyWriter))
+	require.Error(t, runPolicyOp(t, m, fs, "DeleteTable", &DeleteTableRequest{
+		TableBucketARN: bucketARN, Namespace: []string{ns}, Name: name,
+	}, policyDenyWriter))
+
+	entry := fs.Get(GetNamespacePath(policyDenyBucket, "ns"), "t")
+	require.NotNil(t, entry)
+	_, ok := entry.Extended[ExtendedKeyPolicy]
+	assert.True(t, ok, "table policy must still be in place")
+	_, ok = entry.Extended[ExtendedKeyMetadata]
+	assert.True(t, ok, "table must still exist")
+}
+
+// Tag and maintenance operations on a table carry the table ARN as their
+// resource, so the table's own policy Deny applies to them as well.
+func TestTablePolicyDenyCoversTagAndMaintenanceOps(t *testing.T) {
+	fs, m := startPolicyDenyManager(t)
+	bucketARN, ns, name := deniedTableReq(t)
+	tableARN := policyDenyTableARN(t)
+
+	require.Error(t, runPolicyOp(t, m, fs, "TagResource", &TagResourceRequest{
+		ResourceARN: tableARN, Tags: map[string]string{"k": "v"},
+	}, policyDenyWriter), "writer must not tag the table")
+
+	require.Error(t, runPolicyOp(t, m, fs, "UntagResource", &UntagResourceRequest{
+		ResourceARN: tableARN, TagKeys: []string{"k"},
+	}, policyDenyWriter), "writer must not untag the table")
+
+	require.Error(t, runPolicyOp(t, m, fs, "ListTagsForResource", &ListTagsForResourceRequest{
+		ResourceARN: tableARN,
+	}, policyDenyWriter), "writer must not list the table's tags")
+
+	forged, err := BuildTableARN(DefaultRegion, "111111111111", policyDenyBucket, ns, name)
+	require.NoError(t, err)
+	require.Error(t, runPolicyOp(t, m, fs, "TagResource", &TagResourceRequest{
+		ResourceARN: forged, Tags: map[string]string{"k": "v"},
+	}, policyDenyWriter), "a forged account in the ARN must not bypass the table Deny")
+
+	require.Error(t, runPolicyOp(t, m, fs, "PutTableMaintenanceConfiguration", &PutTableMaintenanceConfigurationRequest{
+		TableBucketARN: bucketARN, Namespace: []string{ns}, Name: name,
+		Type: MaintenanceTypeIcebergCompaction, Value: &MaintenanceConfigurationValue{Status: MaintenanceStatusEnabled},
+	}, policyDenyWriter), "writer must not change the table's maintenance configuration")
+
+	require.Error(t, runPolicyOp(t, m, fs, "GetTableMaintenanceJobStatus", &GetTableMaintenanceJobStatusRequest{
+		TableBucketARN: bucketARN, Namespace: []string{ns}, Name: name,
+	}, policyDenyWriter), "writer must not read the table's maintenance status")
+}
+
+// A Deny conditioned on a resource tag applies only while the stored tag
+// matches, so the veto must evaluate the tags held on the table.
+func TestTablePolicyDenyHonorsTagCondition(t *testing.T) {
+	bucketARN, ns, name := deniedTableReq(t)
+	conditionalDeny := fmt.Sprintf(`{"Statement":[{"Effect":"Deny","Principal":%q,"Action":"s3tables:*","Resource":%q,"Condition":{"StringEquals":{"aws:ResourceTag/protected":"true"}}}]}`,
+		policyDenyWriter, policyDenyTableARN(t))
+
+	seed := func(t *testing.T, tags map[string]string) (*s3tablestest.MemFiler, *Manager) {
+		t.Helper()
+		fs, m := startPolicyDenyManager(t)
+		entry := fs.Get(GetNamespacePath(policyDenyBucket, ns), name)
+		require.NotNil(t, entry)
+		entry.Extended[ExtendedKeyPolicy] = []byte(conditionalDeny)
+		if tagData, err := json.Marshal(tags); err == nil {
+			entry.Extended[ExtendedKeyTags] = tagData
+		}
+		return fs, m
+	}
+
+	putPolicy := &PutTablePolicyRequest{
+		TableBucketARN: bucketARN, Namespace: []string{ns}, Name: name,
+		ResourcePolicy: `{"Statement":[{"Effect":"Allow","Principal":"writer","Action":"s3tables:*","Resource":"*"}]}`,
+	}
+	putMaintenance := &PutTableMaintenanceConfigurationRequest{
+		TableBucketARN: bucketARN, Namespace: []string{ns}, Name: name,
+		Type: MaintenanceTypeIcebergCompaction, Value: &MaintenanceConfigurationValue{Status: MaintenanceStatusEnabled},
+	}
+
+	t.Run("matching tag", func(t *testing.T) {
+		fs, m := seed(t, map[string]string{"protected": "true"})
+		require.Error(t, runPolicyOp(t, m, fs, "PutTablePolicy", putPolicy, policyDenyWriter))
+		require.Error(t, runPolicyOp(t, m, fs, "PutTableMaintenanceConfiguration", putMaintenance, policyDenyWriter))
+	})
+
+	t.Run("other tag value", func(t *testing.T) {
+		fs, m := seed(t, map[string]string{"protected": "false"})
+		require.NoError(t, runPolicyOp(t, m, fs, "PutTablePolicy", putPolicy, policyDenyWriter))
+		require.NoError(t, runPolicyOp(t, m, fs, "PutTableMaintenanceConfiguration", putMaintenance, policyDenyWriter))
+	})
+}
+
+// The policy being replaced is still the resource's policy: a Deny on
+// PutTableBucketPolicy inside the existing bucket policy must veto the write
+// even for the bucket owner under the open fallback.
+func TestBucketPolicyDenyVetoesPolicyReplacement(t *testing.T) {
+	fs := s3tablestest.Start(t)
+
+	bucketMeta, _ := json.Marshal(tableBucketMetadata{Name: policyDenyBucket, OwnerAccountID: DefaultAccountID})
+	denyReplace := `{"Statement":[{"Effect":"Deny","Principal":"*","Action":"s3tables:PutTableBucketPolicy","Resource":"*"}]}`
+	fs.Put(TablesPath, policyDenyBucket, map[string][]byte{
+		ExtendedKeyTableBucket: []byte("{}"),
+		ExtendedKeyMetadata:    bucketMeta,
+		ExtendedKeyPolicy:      []byte(denyReplace),
+	})
+
+	m := NewManager()
+	m.SetDefaultAllow(true)
+
+	err := runPolicyOp(t, m, fs, "PutTableBucketPolicy", &PutTableBucketPolicyRequest{
+		TableBucketARN: policyDenyBucketARN(t),
+		ResourcePolicy: `{"Statement":[{"Effect":"Allow","Principal":"*","Action":"s3tables:*","Resource":"*"}]}`,
+	}, "")
+	require.Error(t, err, "existing bucket policy Deny must veto replacing it")
+
+	entry := fs.Get(TablesPath, policyDenyBucket)
+	require.NotNil(t, entry)
+	assert.Equal(t, denyReplace, string(entry.Extended[ExtendedKeyPolicy]), "bucket policy must be unchanged")
+}

weed/s3api/s3tables/permissions.go

@@ -141,6 +141,13 @@ func CheckTableOrBucketPermission(operation, principal, resourceOwner, resourceP
 	return resourceOK || bucketOK
 }
 
+// deniedByPolicy reports whether another applicable resource policy vetoes the
+// request. The admin account is exempt, matching CheckPermissionWithContext.
+func deniedByPolicy(operation, principal, resourcePolicy, resourceARN string, ctx *PolicyContext) bool {
+	return principal != s3_constants.AccountAdminId &&
+		matchesDenyStatement(operation, principal, resourcePolicy, resourceARN, ctx)
+}
+
 // matchesDenyStatement reports whether the JSON policy carries a statement
 // matching this request with Effect Deny. Malformed policies return false;
 // the caller's normal evaluation still denies them.