catalog: refused writes answer 403 to callers that can read the entry (#11650)

* s3tables: answer a denied write with 403 when the caller can read the entry

Update/Delete/Rename answered every authorization refusal as not-found so
the denial leaked no existence signal. For a caller allowed to GetTable
(or GetView) the same entry, the veil hides nothing it could not load —
yet a refused write was still answered 404, so clients saw a table they
just loaded reported as missing.

When the write check fails, re-check read permission on the same entry:
readable entries get 403 AccessDenied; invisible ones keep the
not-found answer.

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

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

* iceberg: map UpdateTable errors through writeManagerError on commit paths

CommitTable and CommitTransaction answered any non-conflict UpdateTable
failure, including AccessDenied and NoSuchTable, with a bare 500. A 500
reads as outcome-unknown to clients (PyIceberg raises
CommitStateUnknownException) where a refused commit is a plain
ForbiddenException, matching what the drop and rename handlers already
emit through the same mapper.

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

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

* s3tables: keep the not-found veil over views and match the real read check on renames

UpdateTable and DeleteTable read shared metadata without checking the
entry kind, so a denied write on a view reported 403 to a GetTable-only
caller where a missing name reports 404. The rename visibility check
also fed resource tags into GetView evaluation that the real GetView
path never supplies.

* s3tables: gofmt handler_table.go

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/iceberg/handlers_commit.go

@@ -398,7 +398,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 			glog.V(1).Infof("Iceberg: failed to cleanup metadata file %s after update failure: %v", newMetadataLocation, cleanupErr)
 		}
 		glog.Errorf("Iceberg: CommitTable UpdateTable error: %v", err)
-		writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to commit table update: "+err.Error())
+		writeManagerError(w, err)
 		return
 	}
 }

weed/s3api/iceberg/handlers_transaction.go

@@ -110,7 +110,7 @@ func (s *Server) handleCommitTransaction(w http.ResponseWriter, r *http.Request)
 				return
 			}
 			glog.Errorf("Iceberg: CommitTransaction UpdateTable error: %v", err)
-			writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to commit table update: "+err.Error())
+			writeManagerError(w, err)
 			return
 		}
 	}

weed/s3api/iceberg/iceberg_invalid_name_test.go

@@ -58,6 +58,12 @@ func TestWriteManagerError(t *testing.T) {
 			// say how to name a real one.
 			wantMessage: []string{"table bucket warehouse not found", "warehouse=s3://<table-bucket>/", "/v1/<table-bucket>/"},
 		},
+		{
+			name:     "denied catalog write is forbidden, not a server fault",
+			err:      &s3tables.S3TablesError{Type: s3tables.ErrCodeAccessDenied, Message: "not authorized to update table"},
+			wantCode: http.StatusForbidden,
+			wantType: "ForbiddenException",
+		},
 		{name: "everything else is a server fault", err: fmt.Errorf("all filers failed, last error: connection refused"), wantCode: http.StatusInternalServerError, wantType: "InternalServerError"},
 	}
 	for _, c := range cases {

weed/s3api/s3tables/handler_existence_hiding_test.go

@@ -126,3 +126,88 @@ func TestDeleteTableAuthorizedVersionMismatchStillConflicts(t *testing.T) {
 	assert.Equal(t, ErrCodeConflict, s3Err.Type)
 	assert.NotNil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t"))
 }
+
+func runRequestAs(t *testing.T, m *Manager, fs *s3tablestest.MemFiler, operation string, input interface{}, actions []string) s3TablesHTTPError {
+	t.Helper()
+
+	body, err := json.Marshal(input)
+	require.NoError(t, err)
+	identity := &testIdentity{Name: "reader", Account: &testIdentityAccount{Id: "reader"}, Actions: actions}
+	ctx := s3_constants.SetIdentityInContext(context.Background(), identity)
+	req, err := newManagerRequest(ctx, operation, body, "reader")
+	require.NoError(t, err)
+
+	recorder := httptest.NewRecorder()
+	m.handler.HandleRequest(recorder, req, NewManagerClient(fs.Client))
+	result := recorder.Result()
+	defer result.Body.Close()
+
+	var response S3TablesError
+	if result.StatusCode >= 400 {
+		require.NoError(t, json.NewDecoder(result.Body).Decode(&response))
+	}
+	return s3TablesHTTPError{status: result.StatusCode, body: response}
+}
+
+// A principal that may read the table is told the write was refused; the
+// not-found veil only hides the entry from callers that cannot see it.
+func TestDeniedWritesReportForbiddenToReaders(t *testing.T) {
+	fs, manager := startRenameManager(t)
+	manager.SetTrusted(false)
+	manager.SetDefaultAllow(false)
+	readActions := []string{"s3tables:GetTable"}
+
+	got := runRequestAs(t, manager, fs, "GetTable", &GetTableRequest{
+		TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "t"}, readActions)
+	assert.Equal(t, 200, got.status)
+
+	for _, tc := range []struct {
+		operation string
+		request   interface{}
+	}{
+		{"UpdateTable", &UpdateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "t", VersionToken: "any"}},
+		{"DeleteTable", &DeleteTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "t"}},
+		{"RenameTable", &RenameTableRequest{TableBucketARN: mustBucketARN(t), SourceNamespace: []string{"ns"}, SourceName: "t", DestNamespace: []string{"ns"}, DestName: "t2"}},
+	} {
+		t.Run(tc.operation, func(t *testing.T) {
+			got := runRequestAs(t, manager, fs, tc.operation, tc.request, readActions)
+			assert.Equal(t, 403, got.status)
+			assert.Equal(t, ErrCodeAccessDenied, got.body.Type)
+		})
+	}
+}
+
+// A view stored at a table name is hidden from table operations: a principal
+// with only s3tables:GetTable must get the same not-found it would see for a
+// genuinely missing name, never a 403 that reveals the entry exists.
+func TestDeniedTableWritesDoNotRevealViews(t *testing.T) {
+	fs, manager := startRenameManager(t)
+	manager.SetTrusted(false)
+	manager.SetDefaultAllow(false)
+
+	viewMeta, _ := json.Marshal(tableMetadataInternal{
+		Name:           "v",
+		Namespace:      "ns",
+		OwnerAccountID: DefaultAccountID,
+	})
+	fs.Put(GetNamespacePath(renameTestBucket, "ns"), "v", map[string][]byte{
+		ExtendedKeyMetadata:  viewMeta,
+		ExtendedKeyEntryType: []byte(EntryTypeView),
+	})
+
+	readActions := []string{"s3tables:GetTable"}
+	for _, tc := range []struct {
+		operation string
+		request   interface{}
+	}{
+		{"GetTable", &GetTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "v"}},
+		{"UpdateTable", &UpdateTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "v", VersionToken: "any"}},
+		{"DeleteTable", &DeleteTableRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "v"}},
+	} {
+		t.Run(tc.operation, func(t *testing.T) {
+			got := runRequestAs(t, manager, fs, tc.operation, tc.request, readActions)
+			assert.Equal(t, 404, got.status)
+			assert.Equal(t, ErrCodeNoSuchTable, got.body.Type)
+		})
+	}
+}

weed/s3api/s3tables/handler_table.go

@@ -2,6 +2,7 @@ package s3tables
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -340,17 +341,10 @@ func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request,
 
 	var metadata tableMetadataInternal
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		entry, err := h.lookupEntry(r.Context(), client, tablePath)
+		data, err := h.readTableMetadata(r.Context(), client, tablePath)
 		if err != nil {
 			return err
 		}
-		if EntryType(entry.Extended) == EntryTypeView {
-			return filer_pb.ErrNotFound
-		}
-		data, ok := entry.Extended[ExtendedKeyMetadata]
-		if !ok {
-			return fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata)
-		}
 		if err := json.Unmarshal(data, &metadata); err != nil {
 			return fmt.Errorf("failed to unmarshal table metadata: %w", err)
 		}
@@ -889,7 +883,7 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
 	var tableTags map[string]string
 	var bucketMetadata tableBucketMetadata
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
+		data, err := h.readTableMetadata(r.Context(), client, tablePath)
 		if err != nil {
 			return err
 		}
@@ -974,7 +968,11 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
 		DefaultAllow:    h.defaultAllowFor(r),
 	})
 	if !tableAllowed && !bucketAllowed {
-		h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName))
+		if h.canReadEntry("GetTable", principal, metadata.OwnerAccountID, bucketMetadata.OwnerAccountID, tablePolicy, bucketPolicy, tableARN, bucketARN, bucketName, namespaceName, tableName, bucketTags, tableTags, identityActions, r) {
+			h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to delete table")
+		} else {
+			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName))
+		}
 		return NewAuthError("DeleteTable", principal, "not authorized to delete table")
 	}
 	if req.VersionToken != "" && metadata.VersionToken != req.VersionToken {
@@ -1043,10 +1041,15 @@ var renamedTableAttributes = []string{
 // catalogEntryKind describes the entry a rename operates on, so tables and
 // views share one implementation of the catalog-only move.
 type catalogEntryKind struct {
-	entryType    string
-	noun         string
-	renameOp     string
-	createOp     string
+	entryType string
+	noun      string
+	renameOp  string
+	createOp  string
+	readOp    string
+	// tagsForRead mirrors whether the read handler supplies resource tags to
+	// the policy check, so the denied-write visibility test cannot accept a
+	// tag-conditioned allow the real read would not evaluate.
+	tagsForRead  bool
 	notFoundCode string
 	existsCode   string
 	// resourceARN builds the ARN a policy scoped to this entry would name, so a
@@ -1060,6 +1063,8 @@ var (
 		noun:         "table",
 		renameOp:     "RenameTable",
 		createOp:     "CreateTable",
+		readOp:       "GetTable",
+		tagsForRead:  true,
 		notFoundCode: ErrCodeNoSuchTable,
 		existsCode:   ErrCodeTableAlreadyExists,
 		resourceARN: func(h *S3TablesHandler, ownerAccountID, bucketName, id string) string {
@@ -1071,6 +1076,8 @@ var (
 		noun:         "view",
 		renameOp:     "RenameView",
 		createOp:     "CreateView",
+		readOp:       "GetView",
+		tagsForRead:  false,
 		notFoundCode: ErrCodeNoSuchView,
 		existsCode:   ErrCodeViewAlreadyExists,
 		resourceARN: func(h *S3TablesHandler, ownerAccountID, bucketName, id string) string {
@@ -1079,6 +1086,42 @@ var (
 	}
 )
 
+// readTableMetadata returns the entry's table-metadata xattr. A view stored at
+// the same name is reported absent, the way GetTable and the renamer's
+// entry-type check already treat kind mismatches.
+func (h *S3TablesHandler) readTableMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient, tablePath string) ([]byte, error) {
+	entry, err := h.lookupEntry(ctx, client, tablePath)
+	if err != nil {
+		return nil, err
+	}
+	if EntryType(entry.Extended) == EntryTypeView {
+		return nil, filer_pb.ErrNotFound
+	}
+	data, ok := entry.Extended[ExtendedKeyMetadata]
+	if !ok {
+		return nil, fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata)
+	}
+	return data, nil
+}
+
+// canReadEntry reports whether the principal may read the catalog entry a
+// denied write would touch. A denied write is then answered with 403 — the
+// refusal leaks nothing a reader could not load anyway — while a caller that
+// cannot see the entry keeps the not-found veil.
+func (h *S3TablesHandler) canReadEntry(readOp, principal, tableOwner, bucketOwner, tablePolicy, bucketPolicy, tableARN, bucketARN, bucketName, namespace, name string, bucketTags, tableTags map[string]string, identityActions []string, r *http.Request) bool {
+	ctx := &PolicyContext{
+		TableBucketName: bucketName,
+		Namespace:       namespace,
+		TableName:       name,
+		TableBucketTags: bucketTags,
+		ResourceTags:    tableTags,
+		IdentityActions: identityActions,
+		DefaultAllow:    h.defaultAllowFor(r),
+	}
+	return CheckPermissionWithContext(readOp, principal, tableOwner, tablePolicy, tableARN, ctx) ||
+		CheckPermissionWithContext(readOp, principal, bucketOwner, bucketPolicy, bucketARN, ctx)
+}
+
 // handleRenameTable moves a table's catalog entry to a new namespace/name within
 // the same bucket. It is catalog-only: the metadata.json and data files stay put,
 // the destination keeps the source's MetadataLocation, and the source name is
@@ -1266,7 +1309,15 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ
 		DefaultAllow:    h.defaultAllowFor(r),
 	})
 	if !tableAllowed && !bucketAllowed {
-		h.writeError(w, http.StatusNotFound, kind.notFoundCode, fmt.Sprintf("%s %s not found", kind.noun, srcName))
+		readTags := tableTags
+		if !kind.tagsForRead {
+			readTags = nil
+		}
+		if h.canReadEntry(kind.readOp, principal, metadata.OwnerAccountID, bucketMetadata.OwnerAccountID, tablePolicy, bucketPolicy, tableARN, bucketARN, bucketName, srcNamespace, srcName, bucketTags, readTags, identityActions, r) {
+			h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to rename "+kind.noun)
+		} else {
+			h.writeError(w, http.StatusNotFound, kind.notFoundCode, fmt.Sprintf("%s %s not found", kind.noun, srcName))
+		}
 		return NewAuthError(kind.renameOp, principal, "not authorized to rename "+kind.noun)
 	}
 
@@ -1462,7 +1513,7 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		// 1. Get Table Metadata
-		data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
+		data, err := h.readTableMetadata(r.Context(), client, tablePath)
 		if err != nil {
 			return err
 		}
@@ -1543,7 +1594,11 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 	})
 
 	if !tableAllowed && !bucketAllowed {
-		h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, "table not found")
+		if h.canReadEntry("GetTable", principal, metadata.OwnerAccountID, bucketMetadata.OwnerAccountID, tablePolicy, bucketPolicy, tableARN, bucketARN, bucketName, namespaceName, tableName, bucketTags, tableTags, identityActions, r) {
+			h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to update table")
+		} else {
+			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, "table not found")
+		}
 		return NewAuthError("UpdateTable", principal, "not authorized to update table")
 	}