iceberg/s3tables: confine stored metadataLocation to the authorized table bucket (#11292)

* iceberg: confine commit/transaction/view-update write paths to authorized bucket

The create, register, and createView handlers already confine the client-
supplied metadata location to the caller table bucket and reject ".."
segments. The commit, create-on-commit, transaction, and view-update paths
read the stored metadataLocation back from the catalog and skipped the same
guard, so a location poisoned via the raw S3Tables UpdateTable API (which
persists metadataLocation verbatim) could escape the caller bucket through
a ".." segment that path.Join collapses in saveMetadataBlob.

Add confineMetadataLocation and apply it after parseS3Location on every
commit/update/transaction/view write path, mirroring the create/register/
createView check. Reject with 400 so a poisoned stored location fails the
commit instead of writing into another tenant bucket tree.

* s3tables: validate metadataLocation at the store layer

The raw S3Tables API (CreateTable, RegisterTable, UpdateTable, CreateView,
UpdateView) persisted the client-supplied metadataLocation verbatim with no
bucket-confinement or traversal check, so a caller could store a location
pointing outside its own bucket. The Iceberg REST gateway commit paths then
read that stored value back and wrote through it.

Add ValidateMetadataLocation and call it in every s3tables store handler
that accepts a metadataLocation, rejecting locations whose bucket differs
from the caller table bucket or whose path contains traversal segments. This
prevents a poisoned location from ever being persisted, complementing the
per-write-path guard added to the Iceberg commit handlers.

* iceberg/s3tables: validate location before repair and after idempotency check

Address review feedback:
- Move the commit-path confinement check ahead of repairManifests so a
  poisoned stored location cannot reach manifest repair I/O before the
  commit is rejected.
- Move ValidateMetadataLocation in CreateTable/CreateView to after the
  existing-resource check so idempotent retries that do not consume the
  requested location are not rejected for an unused bad location.
- Assert HTTP 400 in the cross-tenant reproduction tests so an unrelated
  failure cannot satisfy them.

* iceberg: confine staged metadata location before load in create-on-commit

The create-on-commit path parsed the staged metadata location from the
stage-create marker and called loadMetadataFile before validating that the
staged bucket/path stay within the authorized bucket. Add the same
confineMetadataLocation guard before the read so a tampered marker cannot
direct a cross-tenant metadata read.

* iceberg/s3tables: reject bucket-only metadata locations

ValidateMetadataLocation and confineMetadataLocation accepted s3://bucket
with an empty table path. metadataDirPath then maps every such table to the
shared <TablesPath>/<bucket>/metadata directory, so tables could overwrite
or read each other's metadata files. Require a non-empty table path in both
validators; the empty-location case (where the catalog derives one) is
unaffected.

* iceberg/s3tables: reject slash-only table paths in location validation

s3://bkt/// parses to tablePath="/" which passed the empty-string check
but path.Join cleans it away, mapping to the bucket-level metadata
directory shared across tables. Update isValidTablePath to require at
least one non-empty segment and mirror the same check in
ValidateMetadataLocation, closing the gap in all callers.

weed/s3api/iceberg/commit_helpers.go

@@ -187,6 +187,13 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi
 			message: "Invalid table location: " + err.Error(),
 		}
 	}
+	if err := confineMetadataLocation(metadataBucket, metadataPath, input.markerBucket); err != nil {
+		return nil, &icebergRequestError{
+			status:  http.StatusBadRequest,
+			errType: "BadRequestException",
+			message: err.Error(),
+		}
+	}
 	if err := s.saveMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes, false); err != nil {
 		return nil, &icebergRequestError{
 			status:  http.StatusInternalServerError,

weed/s3api/iceberg/handlers_commit.go

@@ -146,6 +146,10 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 						writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid staged metadata location: "+parseLocationErr.Error())
 						return
 					}
+					if err := confineMetadataLocation(stagedBucket, stagedPath, bucketName); err != nil {
+						writeError(w, http.StatusBadRequest, "BadRequestException", err.Error())
+						return
+					}
 					stagedMetadataBytes, loadErr := s.loadMetadataFile(r.Context(), stagedBucket, stagedPath, stagedFileName)
 					if loadErr != nil {
 						if !errors.Is(loadErr, filer_pb.ErrNotFound) {
@@ -201,6 +205,16 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 					}
 				}
 
+				createBucket, createPath, createLocErr := parseS3Location(location)
+				if createLocErr != nil {
+					writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid table location: "+createLocErr.Error())
+					return
+				}
+				if err := confineMetadataLocation(createBucket, createPath, bucketName); err != nil {
+					writeError(w, http.StatusBadRequest, "BadRequestException", err.Error())
+					return
+				}
+
 				repairManifests(location)
 
 				result, reqErr := s.finalizeCreateOnCommit(r.Context(), createOnCommitInput{
@@ -233,6 +247,15 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 		if location == "" {
 			location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
 		}
+		locBucket, locPath, locErr := parseS3Location(location)
+		if locErr != nil {
+			writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid table location: "+locErr.Error())
+			return
+		}
+		if err := confineMetadataLocation(locBucket, locPath, bucketName); err != nil {
+			writeError(w, http.StatusBadRequest, "BadRequestException", err.Error())
+			return
+		}
 		tableUUID := uuid.Nil
 		if getResp.Metadata != nil && getResp.Metadata.Iceberg != nil && getResp.Metadata.Iceberg.TableUUID != "" {
 			if parsed, parseErr := uuid.Parse(getResp.Metadata.Iceberg.TableUUID); parseErr == nil {
@@ -320,6 +343,10 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid table location: "+err.Error())
 			return
 		}
+		if err := confineMetadataLocation(metadataBucket, metadataPath, bucketName); err != nil {
+			writeError(w, http.StatusBadRequest, "BadRequestException", err.Error())
+			return
+		}
 		metadataFileName, newMetadataLocation, err = s.stageCommitMetadata(r.Context(), metadataBucket, metadataPath, location, metadataFileName, metadataBytes)
 		if err != nil {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save metadata file: "+err.Error())

weed/s3api/iceberg/handlers_transaction.go

@@ -207,6 +207,9 @@ func (s *Server) prepareTableCommit(ctx context.Context, bucketName, bucketARN,
 	if err != nil {
 		return nil, &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Invalid table location: " + err.Error()}
 	}
+	if err := confineMetadataLocation(metadataBucket, metadataPath, bucketName); err != nil {
+		return nil, &icebergRequestError{http.StatusBadRequest, "BadRequestException", err.Error()}
+	}
 
 	return &preparedTableCommit{
 		namespace:           namespace,

weed/s3api/iceberg/handlers_view_update.go

@@ -102,6 +102,10 @@ func (s *Server) handleUpdateView(w http.ResponseWriter, r *http.Request) {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid view location: "+err.Error())
 			return
 		}
+		if err := confineMetadataLocation(metadataBucket, metadataPath, bucketName); err != nil {
+			writeError(w, http.StatusBadRequest, "BadRequestException", err.Error())
+			return
+		}
 		metadataFileName, newMetadataLocation, err = s.stageCommitMetadata(r.Context(), metadataBucket, metadataPath, location, metadataFileName, metadataBytes)
 		if err != nil {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save view metadata file: "+err.Error())

weed/s3api/iceberg/iceberg_commit_location_test.go

@@ -0,0 +1,97 @@
+package iceberg
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+// seedPoisonedTable registers a table entry whose stored metadataLocation
+// points outside its own bucket via a ".." segment, simulating a value
+// persisted verbatim by the raw S3Tables UpdateTable API.
+func seedPoisonedTable(t *testing.T, fc *memFiler, bucket, namespace, tableName, metadataLocation string) {
+	t.Helper()
+	meta := s3tables.TableMetadata{
+		Iceberg: &s3tables.IcebergMetadata{TableUUID: "00000000-0000-0000-0000-000000000001"},
+	}
+	internal := map[string]any{
+		"name":             tableName,
+		"namespace":        namespace,
+		"format":           "ICEBERG",
+		"ownerAccountId":   s3_constants.AccountAdminId,
+		"versionToken":     "v1",
+		"metadataVersion":  1,
+		"metadataLocation": metadataLocation,
+		"metadata":         meta,
+	}
+	metaBytes, _ := json.Marshal(internal)
+	fc.seed(s3tables.GetTablePath(bucket, namespace, tableName), &filer_pb.Entry{
+		Name:        tableName,
+		IsDirectory: true,
+		Extended:    map[string][]byte{s3tables.ExtendedKeyMetadata: metaBytes},
+	})
+}
+
+func TestCommitTableRejectsStoredTraversalLocation(t *testing.T) {
+	const attacker = "attacker"
+	const victim = "victim"
+	fc := newMemFiler()
+	seedNamespace(fc, attacker, "ns", s3_constants.AccountAdminId)
+	seedPoisonedTable(t, fc, attacker, "ns", "t",
+		"s3://attacker/../victim/planted/metadata/v1.metadata.json")
+
+	s := NewServer(fc, nil)
+
+	r := httptest.NewRequest(http.MethodPost, "/v1/"+attacker+"/namespaces/ns/tables/t",
+		strings.NewReader(`{"requirements":[],"updates":[]}`))
+	r = mux.SetURLVars(r, map[string]string{"prefix": attacker, "namespace": "ns", "table": "t"})
+	r = r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), s3_constants.AccountAdminId))
+
+	w := httptest.NewRecorder()
+	s.handleUpdateTable(w, r)
+
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want %d (body=%s)", w.Code, http.StatusBadRequest, w.Body.String())
+	}
+	for p := range fc.entries {
+		if strings.Contains(p, "/"+victim+"/") && !strings.Contains(p, "/"+attacker+"/") {
+			t.Fatalf("cross-tenant write escaped into victim tree at %s (status=%d body=%s)", p, w.Code, w.Body.String())
+		}
+	}
+}
+
+func TestCommitTransactionRejectsStoredTraversalLocation(t *testing.T) {
+	const attacker = "attacker"
+	const victim = "victim"
+	fc := newMemFiler()
+	seedNamespace(fc, attacker, "ns", s3_constants.AccountAdminId)
+	seedPoisonedTable(t, fc, attacker, "ns", "t",
+		"s3://attacker/../victim/planted/metadata/v1.metadata.json")
+
+	s := NewServer(fc, nil)
+
+	body := `{"table-changes":[{"identifier":{"namespace":["ns"],"name":"t"},"requirements":[],"updates":[]}]}`
+	r := httptest.NewRequest(http.MethodPost, "/v1/"+attacker+"/transactions/commit",
+		strings.NewReader(body))
+	r = mux.SetURLVars(r, map[string]string{"prefix": attacker})
+	r = r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), s3_constants.AccountAdminId))
+
+	w := httptest.NewRecorder()
+	s.handleCommitTransaction(w, r)
+
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want %d (body=%s)", w.Code, http.StatusBadRequest, w.Body.String())
+	}
+	for p := range fc.entries {
+		if strings.Contains(p, "/"+victim+"/") && !strings.Contains(p, "/"+attacker+"/") {
+			t.Fatalf("cross-tenant transaction write escaped into victim tree at %s (status=%d body=%s)", p, w.Code, w.Body.String())
+		}
+	}
+}

weed/s3api/iceberg/path_validation.go

@@ -1,6 +1,7 @@
 package iceberg
 
 import (
+	"fmt"
 	"net/http"
 	"strings"
 
@@ -58,8 +59,11 @@ func validateRequestPath(next http.Handler) http.Handler {
 
 // isValidTablePath reports whether a "/"-separated table path (the part below
 // the bucket) is free of segments that path.Join would collapse to escape the
-// bucket directory. Empty segments (from leading/duplicate slashes) are ignored.
+// bucket directory. Empty segments (from leading/duplicate slashes) are
+// ignored, but the path must resolve to at least one real segment so it does
+// not collapse to the bucket-level metadata directory.
 func isValidTablePath(tablePath string) bool {
+	hasSegment := false
 	for _, segment := range strings.Split(tablePath, "/") {
 		if segment == "" {
 			continue
@@ -67,8 +71,9 @@ func isValidTablePath(tablePath string) bool {
 		if !isValidNameSegment(segment) {
 			return false
 		}
+		hasSegment = true
 	}
-	return true
+	return hasSegment
 }
 
 // isValidNameSegment rejects a single path-segment value (bucket prefix slot,
@@ -83,3 +88,17 @@ func isValidNameSegment(s string) bool {
 	}
 	return !strings.ContainsAny(s, "/\\\x00")
 }
+
+// confineMetadataLocation checks that a parsed s3 location stays within the
+// authorized table bucket and rejects traversal segments that path.Join in
+// saveMetadataBlob would collapse to escape it. The table path must resolve
+// to at least one real segment so its metadata directory is table-specific.
+func confineMetadataLocation(metadataBucket, metadataPath, bucketName string) error {
+	if metadataBucket != bucketName {
+		return fmt.Errorf("table location must be within bucket %s", bucketName)
+	}
+	if !isValidTablePath(metadataPath) {
+		return fmt.Errorf("invalid table location path")
+	}
+	return nil
+}

weed/s3api/iceberg/path_validation_test.go

@@ -16,7 +16,9 @@ func TestIsValidTablePath(t *testing.T) {
 		{"sales.orders", true},
 		{"ns/table", true},
 		{"ns//table", true},
-		{"", true},
+		{"", false},
+		{"/", false},
+		{"//", false},
 		{"../other", false},
 		{"ns/../../etc", false},
 		{"ns/./x", false},

weed/s3api/s3tables/handler_table.go

@@ -99,6 +99,11 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 		return err
 	}
 
+	if err := ValidateMetadataLocation(req.MetadataLocation, bucketName); err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return err
+	}
+
 	// Create the table
 	now := time.Now()
 	versionToken := generateVersionToken()
@@ -227,6 +232,11 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req
 	}
 	bucketName, namespaceName, tableName := target.bucketName, target.namespaceName, target.tableName
 
+	if err := ValidateMetadataLocation(req.MetadataLocation, bucketName); err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return err
+	}
+
 	tablePath := GetTablePath(bucketName, namespaceName, tableName)
 
 	// Table must be absent.
@@ -1428,6 +1438,11 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		return err
 	}
 
+	if err := ValidateMetadataLocation(req.MetadataLocation, bucketName); err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return err
+	}
+
 	tablePath := GetTablePath(bucketName, namespaceName, tableName)
 
 	// Load existing metadata and policies for authorization

weed/s3api/s3tables/handler_view.go

@@ -121,6 +121,11 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques
 		return err
 	}
 
+	if err := ValidateMetadataLocation(req.MetadataLocation, bucketName); err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return err
+	}
+
 	now := time.Now()
 	versionToken := generateVersionToken()
 	metadata := &tableMetadataInternal{
@@ -405,6 +410,11 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques
 		return err
 	}
 
+	if err := ValidateMetadataLocation(req.MetadataLocation, bucketName); err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return err
+	}
+
 	viewPath := GetTablePath(bucketName, namespaceName, viewName)
 	var metadata tableMetadataInternal
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {

weed/s3api/s3tables/utils.go

@@ -126,6 +126,58 @@ func TableDataDirFromMetadataLocation(metadataLocation string) string {
 	return path.Join(TablesPath, loc)
 }
 
+// ValidateMetadataLocation checks that an s3:// metadata location stays within
+// the authorized table bucket and rejects traversal segments that path.Join
+// would collapse to escape the bucket directory. Empty locations are allowed
+// (the catalog derives one). A non-empty location must include a table path so
+// its metadata directory is table-specific, not shared at the bucket level.
+func ValidateMetadataLocation(metadataLocation, bucketName string) error {
+	if metadataLocation == "" {
+		return nil
+	}
+	bucket, tablePath, err := parseS3Location(metadataLocation)
+	if err != nil {
+		return err
+	}
+	if bucket != bucketName {
+		return fmt.Errorf("metadata location must be within bucket %s", bucketName)
+	}
+	hasSegment := false
+	for _, segment := range strings.Split(tablePath, "/") {
+		if segment == "" {
+			continue
+		}
+		if segment == "." || segment == ".." || strings.ContainsAny(segment, "\\\x00") {
+			return fmt.Errorf("invalid metadata location path")
+		}
+		hasSegment = true
+	}
+	if !hasSegment {
+		return fmt.Errorf("metadata location must include a table path")
+	}
+	return nil
+}
+
+func parseS3Location(location string) (bucket, tablePath string, err error) {
+	if !strings.HasPrefix(location, "s3://") {
+		return "", "", fmt.Errorf("unsupported location: %s", location)
+	}
+	trimmed := strings.TrimPrefix(location, "s3://")
+	trimmed = strings.TrimSuffix(trimmed, "/")
+	if trimmed == "" {
+		return "", "", fmt.Errorf("invalid location: %s", location)
+	}
+	parts := strings.SplitN(trimmed, "/", 2)
+	bucket = parts[0]
+	if bucket == "" {
+		return "", "", fmt.Errorf("invalid location bucket: %s", location)
+	}
+	if len(parts) == 2 {
+		tablePath = parts[1]
+	}
+	return bucket, tablePath, nil
+}
+
 // GetTableObjectRootDir returns the root path for table bucket object storage
 func GetTableObjectRootDir() string {
 	return path.Join(TablesPath, tableObjectRootDirName)

weed/s3api/s3tables/utils_location_test.go

@@ -0,0 +1,33 @@
+package s3tables
+
+import "testing"
+
+func TestValidateMetadataLocation(t *testing.T) {
+	tests := []struct {
+		name    string
+		loc     string
+		bucket  string
+		wantErr bool
+	}{
+		{"empty allowed", "", "bkt", false},
+		{"same bucket", "s3://bkt/ns/t/metadata/v1.metadata.json", "bkt", false},
+		{"cross bucket rejected", "s3://other/ns/t/metadata/v1.metadata.json", "bkt", true},
+		{"bucket only rejected", "s3://bkt", "bkt", true},
+		{"bucket with trailing slash rejected", "s3://bkt/", "bkt", true},
+		{"slash-only path rejected", "s3://bkt///", "bkt", true},
+		{"dotdot in path rejected", "s3://bkt/../victim/metadata/v1.metadata.json", "bkt", true},
+		{"dotdot mid path rejected", "s3://bkt/ns/../../victim/metadata/v1.metadata.json", "bkt", true},
+		{"dot segment rejected", "s3://bkt/ns/./t/metadata/v1.metadata.json", "bkt", true},
+		{"backslash segment rejected", "s3://bkt/ns/\\t/metadata/v1.metadata.json", "bkt", true},
+		{"non-s3 scheme rejected", "file:///bkt/ns/t", "bkt", true},
+		{"empty location string", "s3://", "bkt", true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := ValidateMetadataLocation(tt.loc, tt.bucket)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("ValidateMetadataLocation(%q, %q) err = %v, wantErr = %v", tt.loc, tt.bucket, err, tt.wantErr)
+			}
+		})
+	}
+}