s3tables: Fix ListTables authorization and policy parsing

Make ListTables authorization consistent with GetTable/CreateTable:

1. ListTables authorization now evaluates policies instead of owner-only checks:
   - For namespace listing: checks namespace policy AND bucket policy
   - For bucket-wide listing: checks bucket policy
   - Uses CanListTables permission framework

2. Remove owner-only filter in listTablesWithClient that prevented policy-based
   sharing of tables. Authorization is now enforced at the handler level, so all
   tables in the namespace/bucket are returned to authorized callers (who have
   access either via ownership or policy).

3. Add flexible PolicyDocument.UnmarshalJSON to support both single-object and
   array forms of Statement field:
   - Handles: {"Statement": {...}}
   - Handles: {"Statement": [{...}, {...}]}
   - Improves AWS IAM compatibility

This ensures cross-account table listing works when delegated via bucket/namespace
policies, consistent with the authorization model for other operations.

weed/s3api/s3tables/handler_table.go

@@ -403,10 +403,17 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		var err error
+		accountID := h.getAccountID(r)
+		
 		if len(req.Namespace) > 0 {
 			// Namespace has already been validated above
 			namespacePath := getNamespacePath(bucketName, namespaceName)
+			bucketPath := getTableBucketPath(bucketName)
 			var nsMeta namespaceMetadata
+			var bucketMeta tableBucketMetadata
+			var namespacePolicy, bucketPolicy string
+
+			// Fetch namespace metadata and policy
 			data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
 			if err != nil {
 				return err // Not Found handled by caller
@@ -414,15 +421,47 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			if err := json.Unmarshal(data, &nsMeta); err != nil {
 				return err
 			}
-			if accountID := h.getAccountID(r); accountID != nsMeta.OwnerAccountID {
+
+			// Fetch namespace policy if it exists
+			policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy)
+			if err == nil {
+				namespacePolicy = string(policyData)
+			} else if !errors.Is(err, ErrAttributeNotFound) {
+				return fmt.Errorf("failed to fetch namespace policy: %v", err)
+			}
+
+			// Fetch bucket metadata and policy
+			data, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
+			if err == nil {
+				if err := json.Unmarshal(data, &bucketMeta); err != nil {
+					return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
+				}
+			} else if !errors.Is(err, ErrAttributeNotFound) {
+				return fmt.Errorf("failed to fetch bucket metadata: %v", err)
+			}
+
+			policyData, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
+			if err == nil {
+				bucketPolicy = string(policyData)
+			} else if !errors.Is(err, ErrAttributeNotFound) {
+				return fmt.Errorf("failed to fetch bucket policy: %v", err)
+			}
+
+			// Authorize listing: namespace policy OR bucket policy OR ownership
+			nsAllowed := CanListTables(accountID, nsMeta.OwnerAccountID, namespacePolicy)
+			bucketAllowed := CanListTables(accountID, bucketMeta.OwnerAccountID, bucketPolicy)
+			if !nsAllowed && !bucketAllowed {
 				return ErrAccessDenied
 			}
 
 			tables, paginationToken, err = h.listTablesInNamespaceWithClient(r, client, bucketName, namespaceName, req.Prefix, req.ContinuationToken, maxTables)
 		} else {
-			// Check permission (check bucket ownership)
+			// List tables across all namespaces in bucket
 			bucketPath := getTableBucketPath(bucketName)
 			var bucketMeta tableBucketMetadata
+			var bucketPolicy string
+
+			// Fetch bucket metadata and policy
 			data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
 			if err != nil {
 				return err
@@ -430,7 +469,17 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 			if err := json.Unmarshal(data, &bucketMeta); err != nil {
 				return err
 			}
-			if accountID := h.getAccountID(r); accountID != bucketMeta.OwnerAccountID {
+
+			// Fetch bucket policy if it exists
+			policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
+			if err == nil {
+				bucketPolicy = string(policyData)
+			} else if !errors.Is(err, ErrAttributeNotFound) {
+				return fmt.Errorf("failed to fetch bucket policy: %v", err)
+			}
+
+			// Authorize listing: bucket policy OR ownership
+			if !CanListTables(accountID, bucketMeta.OwnerAccountID, bucketPolicy) {
 				return ErrAccessDenied
 			}
 
@@ -530,9 +579,9 @@ func (h *S3TablesHandler) listTablesWithClient(r *http.Request, client filer_pb.
 				continue
 			}
 
-			if metadata.OwnerAccountID != h.getAccountID(r) {
-				continue
-			}
+			// Note: Authorization (ownership or policy-based access) is checked at the handler level
+			// before calling this function. This filter is removed to allow policy-based sharing.
+			// The caller has already been verified to have ListTables permission for this namespace/bucket.
 
 			tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+entry.Entry.Name)
 

weed/s3api/s3tables/permissions.go

@@ -2,6 +2,7 @@ package s3tables
 
 import (
 	"encoding/json"
+	"fmt"
 	"strings"
 
 	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
@@ -16,6 +17,53 @@ type PolicyDocument struct {
 	Statement []Statement `json:"Statement"`
 }
 
+// UnmarshalJSON handles both single statement object and array of statements
+// AWS allows {"Statement": {...}} or {"Statement": [{...}]}
+func (pd *PolicyDocument) UnmarshalJSON(data []byte) error {
+	type Alias PolicyDocument
+	aux := &struct {
+		Statement interface{} `json:"Statement"`
+		*Alias
+	}{
+		Alias: (*Alias)(pd),
+	}
+
+	if err := json.Unmarshal(data, &aux); err != nil {
+		return err
+	}
+
+	// Handle Statement as either a single object or array
+	switch s := aux.Statement.(type) {
+	case map[string]interface{}:
+		// Single statement object - unmarshal to one Statement
+		stmtData, err := json.Marshal(s)
+		if err != nil {
+			return fmt.Errorf("failed to marshal single statement: %w", err)
+		}
+		var stmt Statement
+		if err := json.Unmarshal(stmtData, &stmt); err != nil {
+			return fmt.Errorf("failed to unmarshal single statement: %w", err)
+		}
+		pd.Statement = []Statement{stmt}
+	case []interface{}:
+		// Array of statements - normal handling
+		stmtData, err := json.Marshal(s)
+		if err != nil {
+			return fmt.Errorf("failed to marshal statement array: %w", err)
+		}
+		if err := json.Unmarshal(stmtData, &pd.Statement); err != nil {
+			return fmt.Errorf("failed to unmarshal statement array: %w", err)
+		}
+	case nil:
+		// No statements
+		pd.Statement = []Statement{}
+	default:
+		return fmt.Errorf("Statement must be an object or array, got %T", aux.Statement)
+	}
+
+	return nil
+}
+
 type Statement struct {
 	Effect    string      `json:"Effect"`    // "Allow" or "Deny"
 	Principal interface{} `json:"Principal"` // Can be string, []string, or map