iceberg, lance: vend read-only credentials to callers that may only read (#11859)

* s3tables: share the UpdateTable authorization gate

Extract the validation, load, and permission check from handleUpdateTable
into authorizeUpdateTable so the table an update was authorized against is
the same snapshot the mutation writes to. Manager gains
AuthorizeUpdateTable, mirroring AuthorizeCreateTable, so callers can ask the
question without updating anything.

* s3tables: share the UpdateView authorization gate

Extract the lookup, policy load, and permission check from handleUpdateView
into authorizeUpdateView, mirroring authorizeUpdateTable, and expose it as
Manager.AuthorizeUpdateView for callers that need the answer without the
update.

* s3api: scope vended table credentials to read-only on request

VendTableCredentials takes a readOnly flag and tablePrefixSessionPolicy
emits only s3:GetObject for the table prefix when it is set, keeping the
conditioned ListBucket and unconditioned GetBucketLocation statements that
reads still need.

* iceberg: vend read-only credentials when the caller may not write

LoadTable and friends vend storage credentials only for the authority the
caller holds: the vend now checks UpdateTable/UpdateView permission and a
caller that may only read the catalog entry gets a read-only session. Paths
that just passed a write gate (stage-create, view commit) keep the full
session.

* lance: vend read-only credentials when the caller may not write

storage_options vends the same read-only session as the Iceberg catalog
when the caller lacks UpdateTable on the table.

* lance: keep write credentials for callers authorized to declare

A Lance commit is a direct S3 write that never revisits the catalog, so a
caller that just passed CreateTable still gets a writable session; the
UpdateTable gate applies to describe, the read path.

* iceberg, lance: log a failed write check instead of only reading it as denied

weed/command/s3.go

@@ -762,8 +762,8 @@ 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)
+func (v icebergCredentialVendor) VendTableCredentials(ctx context.Context, principal, bucket, prefix string, readOnly bool) (*iceberg.VendedCredentials, error) {
+	credentials, err := v.server.VendTableCredentials(ctx, principal, bucket, prefix, readOnly)
 	if err != nil || credentials == nil {
 		return nil, err
 	}
@@ -781,8 +781,8 @@ type lanceCredentialVendor struct {
 	server *s3api.S3ApiServer
 }
 
-func (v lanceCredentialVendor) VendTableCredentials(ctx context.Context, principal, bucket, prefix string) (*lance.VendedCredentials, error) {
-	credentials, err := v.server.VendTableCredentials(ctx, principal, bucket, prefix)
+func (v lanceCredentialVendor) VendTableCredentials(ctx context.Context, principal, bucket, prefix string, readOnly bool) (*lance.VendedCredentials, error) {
+	credentials, err := v.server.VendTableCredentials(ctx, principal, bucket, prefix, readOnly)
 	if err != nil || credentials == nil {
 		return nil, err
 	}

weed/s3api/iceberg/handlers_table.go

@@ -294,7 +294,10 @@ 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)
+		// The staged table is not registered yet, so there is nothing to
+		// authorize an update against; the create gate above is the write
+		// authority for the prefix the staged table will occupy.
+		config, storageCredentials := s.buildFileIOConfig(r, location, func() bool { return true })
 		result := LoadTableResult{
 			MetadataLocation:   metadataLocation,
 			Metadata:           metadata,
@@ -395,7 +398,9 @@ 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)
+	config, storageCredentials := s.buildFileIOConfig(r, location, func() bool {
+		return s.mayUpdateTable(r, bucketName, namespace, req.Name)
+	})
 	result := LoadTableResult{
 		MetadataLocation:   finalLocation,
 		Metadata:           metadata,
@@ -590,7 +595,9 @@ 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)
+	config, storageCredentials := s.buildFileIOConfig(r, location, func() bool {
+		return s.mayUpdateTable(r, bucketName, namespace, tableName)
+	})
 	return LoadTableResult{
 		MetadataLocation:   getResp.MetadataLocation,
 		Metadata:           metadata,
@@ -646,7 +653,7 @@ func applySnapshotsParam(r *http.Request, metadata table.Metadata) (table.Metada
 // a region on attach don't need to be told it out-of-band. See issue #9103.
 // 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) {
+func (s *Server) buildFileIOConfig(r *http.Request, location string, canWrite func() bool) (iceberg.Properties, []StorageCredential) {
 	config := make(iceberg.Properties)
 	if s.s3Endpoint == "" {
 		return config, nil
@@ -654,7 +661,7 @@ func (s *Server) buildFileIOConfig(r *http.Request, location string) (iceberg.Pr
 
 	vended := iceberg.Properties(nil)
 	if wantsVendedCredentials(r) {
-		vended = s.vendCredentials(r, location)
+		vended = s.vendCredentials(r, location, canWrite)
 		// 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
@@ -687,7 +694,7 @@ func (s *Server) buildFileIOConfig(r *http.Request, location string) (iceberg.Pr
 // 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 {
+func (s *Server) vendCredentials(r *http.Request, location string, canWrite func() bool) iceberg.Properties {
 	if s.credentialVendor == nil || location == "" {
 		return nil
 	}
@@ -699,7 +706,7 @@ func (s *Server) vendCredentials(r *http.Request, location string) iceberg.Prope
 	}
 
 	principal := s3_constants.GetIdentityNameFromContext(r)
-	credentials, err := s.credentialVendor.VendTableCredentials(r.Context(), principal, bucket, prefix)
+	credentials, err := s.credentialVendor.VendTableCredentials(r.Context(), principal, bucket, prefix, canWrite == nil || !canWrite())
 	if err != nil {
 		glog.Warningf("Iceberg: failed to vend credentials for %s: %v", location, err)
 		return nil
@@ -719,6 +726,42 @@ func (s *Server) vendCredentials(r *http.Request, location string) iceberg.Prope
 	return vended
 }
 
+// mayUpdateTable reports whether the caller may commit to the table the
+// credentials would cover. The session stays read-only when it may not, so
+// the vend carries no more authority than the caller holds at the catalog.
+func (s *Server) mayUpdateTable(r *http.Request, bucketName string, namespace []string, tableName string) bool {
+	updateReq := &s3tables.UpdateTableRequest{
+		TableBucketARN: buildTableBucketARN(bucketName),
+		Namespace:      namespace,
+		Name:           tableName,
+	}
+	identityName := s3_constants.GetIdentityNameFromContext(r)
+	err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return s.tablesManager.AuthorizeUpdateTable(r.Context(), s3tables.NewManagerClient(client), updateReq, identityName)
+	})
+	if err != nil {
+		glog.V(1).Infof("Iceberg: UpdateTable check for %s/%s failed, vending read-only: %v", bucketName, tableName, err)
+	}
+	return err == nil
+}
+
+// mayUpdateView is mayUpdateTable for a view, whose commit path is UpdateView.
+func (s *Server) mayUpdateView(r *http.Request, bucketName string, namespace []string, viewName string) bool {
+	updateReq := &s3tables.UpdateViewRequest{
+		TableBucketARN: buildTableBucketARN(bucketName),
+		Namespace:      namespace,
+		Name:           viewName,
+	}
+	identityName := s3_constants.GetIdentityNameFromContext(r)
+	err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return s.tablesManager.AuthorizeUpdateView(r.Context(), s3tables.NewManagerClient(client), updateReq, identityName)
+	})
+	if err != nil {
+		glog.V(1).Infof("Iceberg: UpdateView check for %s/%s failed, vending read-only: %v", bucketName, viewName, err)
+	}
+	return err == nil
+}
+
 // handleTableExists checks if a table exists.
 func (s *Server) handleTableExists(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)

weed/s3api/iceberg/handlers_view.go

@@ -218,7 +218,9 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) {
 	if finalLocation == "" {
 		finalLocation = metadataLocation
 	}
-	config, _ := s.buildFileIOConfig(r, location)
+	config, _ := s.buildFileIOConfig(r, location, func() bool {
+		return s.mayUpdateView(r, bucketName, namespace, req.Name)
+	})
 	writeLoadResult(w, http.StatusOK, ViewResponse{
 		MetadataLocation: finalLocation,
 		Metadata:         metadata,
@@ -349,7 +351,9 @@ 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))
+	config, _ := s.buildFileIOConfig(r, tableLocationFromMetadataLocation(getResp.MetadataLocation), func() bool {
+		return s.mayUpdateView(r, bucketName, namespace, viewName)
+	})
 	return ViewResponse{
 		MetadataLocation: getResp.MetadataLocation,
 		Metadata:         metadata,

weed/s3api/iceberg/handlers_view_update.go

@@ -133,7 +133,7 @@ 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)
+			config, _ := s.buildFileIOConfig(r, location, func() bool { return true })
 			writeLoadResult(w, http.StatusOK, ViewResponse{
 				MetadataLocation: newMetadataLocation,
 				Metadata:         newMetadata,

weed/s3api/iceberg/iceberg_access_delegation_test.go

@@ -43,12 +43,12 @@ func TestBuildFileIOConfigWithholdsEndpointFromCredentialVendingClients(t *testi
 
 	r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil)
 	r.Header.Set(accessDelegationHeader, "vended-credentials")
-	if got, _ := s.buildFileIOConfig(r, "s3://warehouse/ns/t"); len(got) != 0 {
+	if got, _ := s.buildFileIOConfig(r, "s3://warehouse/ns/t", nil); 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, "s3://warehouse/ns/t"); got["s3.endpoint"] != s.s3Endpoint {
+	if got, _ := s.buildFileIOConfig(plain, "s3://warehouse/ns/t", nil); got["s3.endpoint"] != s.s3Endpoint {
 		t.Fatalf("s3.endpoint = %q, want %q", got["s3.endpoint"], s.s3Endpoint)
 	}
 }

weed/s3api/iceberg/iceberg_credential_vending_test.go

@@ -14,10 +14,11 @@ type stubVendor struct {
 	gotBucket   string
 	gotPrefix   string
 	gotcaller   string
+	gotReadOnly bool
 }
 
-func (v *stubVendor) VendTableCredentials(_ context.Context, principal, bucket, prefix string) (*VendedCredentials, error) {
-	v.gotcaller, v.gotBucket, v.gotPrefix = principal, bucket, prefix
+func (v *stubVendor) VendTableCredentials(_ context.Context, principal, bucket, prefix string, readOnly bool) (*VendedCredentials, error) {
+	v.gotcaller, v.gotBucket, v.gotPrefix, v.gotReadOnly = principal, bucket, prefix, readOnly
 	return v.credentials, v.err
 }
 
@@ -40,7 +41,7 @@ func TestBuildFileIOConfigVendingVendsScopedCredentials(t *testing.T) {
 	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")
+	config, storageCredentials := s.buildFileIOConfig(r, "s3://warehouse/ns/t", nil)
 
 	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)
@@ -70,7 +71,7 @@ func TestBuildFileIOConfigVendingWithoutVendorSaysNothing(t *testing.T) {
 	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")
+	config, storageCredentials := s.buildFileIOConfig(r, "s3://warehouse/ns/t", nil)
 	if len(config) != 0 || storageCredentials != nil {
 		t.Errorf("config = %v, storage-credentials = %v, want both empty", config, storageCredentials)
 	}
@@ -81,12 +82,35 @@ func TestBuildFileIOConfigVendingFallsBackWhenVendingFails(t *testing.T) {
 	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")
+	config, storageCredentials := s.buildFileIOConfig(r, "s3://warehouse/ns/t", nil)
 	if len(config) != 0 || storageCredentials != nil {
 		t.Errorf("config = %v, storage-credentials = %v, want both empty", config, storageCredentials)
 	}
 }
 
+// The vend stays read-only unless the caller proved write authority: a nil
+// or false canWrite must reach the vendor as readOnly.
+func TestBuildFileIOConfigVendingReadOnlyUnlessCanWrite(t *testing.T) {
+	vendor := &stubVendor{credentials: &VendedCredentials{AccessKeyID: "ASIAEXAMPLE"}}
+	s := vendingServer(vendor)
+
+	r := httptest.NewRequest("GET", "/v1/namespaces/ns/tables/t", nil)
+	r.Header.Set(accessDelegationHeader, "vended-credentials")
+
+	s.buildFileIOConfig(r, "s3://warehouse/ns/t", nil)
+	if !vendor.gotReadOnly {
+		t.Error("vendCredentials() passed readOnly=false for a nil canWrite, want readOnly=true")
+	}
+	s.buildFileIOConfig(r, "s3://warehouse/ns/t", func() bool { return false })
+	if !vendor.gotReadOnly {
+		t.Error("vendCredentials() passed readOnly=false for canWrite=false, want readOnly=true")
+	}
+	s.buildFileIOConfig(r, "s3://warehouse/ns/t", func() bool { return true })
+	if vendor.gotReadOnly {
+		t.Error("vendCredentials() passed readOnly=true for canWrite=true, want readOnly=false")
+	}
+}
+
 // A client that did not ask for delegation keeps the plain endpoint config and
 // must never be handed credentials.
 func TestBuildFileIOConfigVendingWithoutDelegationHeader(t *testing.T) {
@@ -94,7 +118,7 @@ func TestBuildFileIOConfigVendingWithoutDelegationHeader(t *testing.T) {
 	s := vendingServer(vendor)
 
 	r := httptest.NewRequest("GET", "/v1/namespaces/ns/tables/t", nil)
-	config, storageCredentials := s.buildFileIOConfig(r, "s3://warehouse/ns/t")
+	config, storageCredentials := s.buildFileIOConfig(r, "s3://warehouse/ns/t", nil)
 
 	if _, vended := config["s3.access-key-id"]; vended {
 		t.Errorf("credentials vended without the delegation header: %v", config)

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(), "s3://warehouse/ns/t")
+		got, _ := s.buildFileIOConfig(loadTable(), "s3://warehouse/ns/t", nil)
 		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(), "s3://warehouse/ns/t")
+		got, _ := s.buildFileIOConfig(loadTable(), "s3://warehouse/ns/t", nil)
 		if got["s3.endpoint"] != "http://seaweed.example:8333" {
 			t.Fatalf("s3.endpoint = %q, want %q", got["s3.endpoint"], "http://seaweed.example:8333")
 		}

weed/s3api/iceberg/server.go

@@ -45,10 +45,11 @@ type VendedCredentials struct {
 }
 
 // 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.
+// caller the catalog has already authenticated and authorized, read-only when
+// readOnly is set. 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)
+	VendTableCredentials(ctx context.Context, principal, bucket, prefix string, readOnly bool) (*VendedCredentials, error)
 }
 
 // Server implements the Iceberg REST Catalog API.

weed/s3api/lance/handlers_table.go

@@ -150,7 +150,9 @@ func (s *Server) handleDeclareTable(w http.ResponseWriter, r *http.Request) {
 		glog.V(2).Infof("lance: could not clear %s for %s: %v", deregisteredMarker, location, err)
 	}
 
-	options, err := s.storageOptions(r, bucket, location, wants(r, "vend_credentials", req.VendCredentials))
+	// A Lance commit is a direct S3 write that never revisits the catalog, so
+	// the create authorization above is the write evidence for the dataset.
+	options, err := s.storageOptions(r, bucket, location, wants(r, "vend_credentials", req.VendCredentials), func() bool { return true })
 	if err != nil {
 		writeError(w, r, http.StatusInternalServerError, codeInternal, err.Error())
 		return
@@ -210,7 +212,9 @@ func (s *Server) handleDescribeTable(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	options, err := s.storageOptions(r, bucket, location, wants(r, "vend_credentials", req.VendCredentials))
+	options, err := s.storageOptions(r, bucket, location, wants(r, "vend_credentials", req.VendCredentials), func() bool {
+		return s.mayUpdateTable(r, bucket, ns, name)
+	})
 	if err != nil {
 		writeError(w, r, http.StatusInternalServerError, codeInternal, err.Error())
 		return

weed/s3api/lance/server.go

@@ -45,10 +45,11 @@ type VendedCredentials struct {
 }
 
 // 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.
+// caller the catalog has already authenticated and authorized, read-only when
+// readOnly is set. 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)
+	VendTableCredentials(ctx context.Context, principal, bucket, prefix string, readOnly bool) (*VendedCredentials, error)
 }
 
 // Server implements the Lance Namespace REST spec.

weed/s3api/lance/storage.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 
 	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
@@ -135,7 +136,7 @@ func confineLocation(bucket, location string) error {
 // storageOptions builds the object_store settings a Lance client needs to reach
 // the dataset. The key names are the aws_-prefixed forms Lance clients pass
 // through to object_store.
-func (s *Server) storageOptions(r *http.Request, bucket, location string, vend bool) (map[string]string, error) {
+func (s *Server) storageOptions(r *http.Request, bucket, location string, vend bool, canWrite func() bool) (map[string]string, error) {
 	options := map[string]string{}
 	if s.s3Endpoint != "" {
 		options["aws_endpoint"] = s.s3Endpoint
@@ -154,7 +155,7 @@ func (s *Server) storageOptions(r *http.Request, bucket, location string, vend b
 	}
 
 	principal := s3_constants.GetIdentityNameFromContext(r)
-	credentials, err := s.credentialVendor.VendTableCredentials(r.Context(), principal, bucket, locationPrefix(location))
+	credentials, err := s.credentialVendor.VendTableCredentials(r.Context(), principal, bucket, locationPrefix(location), canWrite == nil || !canWrite())
 	if err != nil {
 		return nil, err
 	}
@@ -172,6 +173,25 @@ func (s *Server) storageOptions(r *http.Request, bucket, location string, vend b
 	return options, nil
 }
 
+// mayUpdateTable reports whether the caller may commit to the table the
+// credentials would cover, the same gate the Iceberg catalog applies so a
+// read-only caller never gets write-capable storage_options.
+func (s *Server) mayUpdateTable(r *http.Request, bucket string, ns []string, name string) bool {
+	req := &s3tables.UpdateTableRequest{
+		TableBucketARN: bucketARN(bucket),
+		Namespace:      ns,
+		Name:           name,
+	}
+	identityName := s3_constants.GetIdentityNameFromContext(r)
+	err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return s.tablesManager.AuthorizeUpdateTable(r.Context(), s3tables.NewManagerClient(client), req, identityName)
+	})
+	if err != nil {
+		glog.V(1).Infof("lance: UpdateTable check for %s/%s failed, vending read-only: %v", bucket, name, err)
+	}
+	return err == nil
+}
+
 // locationPrefix strips the s3://bucket/ part of a location, leaving the key
 // prefix a credential is scoped to.
 func locationPrefix(location string) string {

weed/s3api/s3api_iceberg_credentials.go

@@ -29,11 +29,12 @@ type VendedCredentials struct {
 	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) {
+// VendTableCredentials mints credentials that can read exactly one table's
+// prefix, and write it too unless readOnly, 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, readOnly bool) (*VendedCredentials, error) {
 	if s3a.icebergCredentialRole == "" {
 		return nil, nil
 	}
@@ -49,7 +50,7 @@ func (s3a *S3ApiServer) VendTableCredentials(ctx context.Context, principal, buc
 		return nil, fmt.Errorf("STS is not configured")
 	}
 
-	policy, err := tablePrefixSessionPolicy(bucket, prefix)
+	policy, err := tablePrefixSessionPolicy(bucket, prefix, readOnly)
 	if err != nil {
 		return nil, err
 	}
@@ -98,8 +99,9 @@ func (s3a *S3ApiServer) stsService() *sts.STSService {
 }
 
 // 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) {
+// bucket listing that clients need to resolve them. A read-only session drops
+// the write verbs: the catalog vends no more authority than the caller holds.
+func tablePrefixSessionPolicy(bucket, prefix string, readOnly bool) (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:
@@ -114,12 +116,17 @@ func tablePrefixSessionPolicy(bucket, prefix string) (string, error) {
 		return "", fmt.Errorf("refusing to scope credentials to a whole bucket: s3://%s", bucket)
 	}
 
+	objectActions := []string{"s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:AbortMultipartUpload", "s3:ListMultipartUploadParts"}
+	if readOnly {
+		objectActions = []string{"s3:GetObject"}
+	}
+
 	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"},
+				"Action":   objectActions,
 				"Resource": []string{fmt.Sprintf("arn:aws:s3:::%s/%s/*", bucket, prefix)},
 			},
 			{

weed/s3api/s3api_iceberg_credentials_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestTablePrefixSessionPolicyScopesToTheTable(t *testing.T) {
-	raw, err := tablePrefixSessionPolicy("warehouse", "analytics/events")
+	raw, err := tablePrefixSessionPolicy("warehouse", "analytics/events", false)
 	if err != nil {
 		t.Fatalf("tablePrefixSessionPolicy() error = %v", err)
 	}
@@ -42,13 +42,43 @@ func TestTablePrefixSessionPolicyScopesToTheTable(t *testing.T) {
 	}
 }
 
+// A caller the catalog would not let commit gets a session that can read the
+// table's files and nothing more.
+func TestTablePrefixSessionPolicyReadOnly(t *testing.T) {
+	raw, err := tablePrefixSessionPolicy("warehouse", "analytics/events", true)
+	if err != nil {
+		t.Fatalf("tablePrefixSessionPolicy() error = %v", err)
+	}
+
+	var policy struct {
+		Statement []struct {
+			Action []string `json:"Action"`
+		} `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 {
+			switch action {
+			case "s3:PutObject", "s3:DeleteObject", "s3:AbortMultipartUpload", "s3:ListMultipartUploadParts":
+				t.Errorf("read-only session policy grants %s", action)
+			}
+		}
+	}
+	if got := policy.Statement[0].Action; len(got) != 1 || got[0] != "s3:GetObject" {
+		t.Errorf("object actions = %v, want [s3:GetObject]", got)
+	}
+}
+
 // 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 {
+	if _, err := tablePrefixSessionPolicy("warehouse", "", false); err == nil {
 		t.Error("tablePrefixSessionPolicy() error = nil, want a refusal for an empty prefix")
 	}
-	if _, err := tablePrefixSessionPolicy("warehouse", "/"); err == nil {
+	if _, err := tablePrefixSessionPolicy("warehouse", "/", false); err == nil {
 		t.Error("a prefix of only separators was accepted")
 	}
 }
@@ -56,7 +86,7 @@ func TestTablePrefixSessionPolicyRefusesAWholeBucket(t *testing.T) {
 // 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")
+	raw, err := tablePrefixSessionPolicy("warehouse", "analytics/events", false)
 	if err != nil {
 		t.Fatalf("tablePrefixSessionPolicy() error = %v", err)
 	}
@@ -101,7 +131,7 @@ func TestIcebergSessionNameIsBounded(t *testing.T) {
 // telling clients to use their own credentials.
 func TestVendTableCredentialsDisabledWithoutRole(t *testing.T) {
 	s3a := &S3ApiServer{}
-	credentials, err := s3a.VendTableCredentials(context.Background(), "admin", "warehouse", "ns/t")
+	credentials, err := s3a.VendTableCredentials(context.Background(), "admin", "warehouse", "ns/t", false)
 	if err != nil {
 		t.Fatalf("VendTableCredentials() error = %v, want nil", err)
 	}
@@ -114,7 +144,7 @@ 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 {
+	if _, err := s3a.VendTableCredentials(context.Background(), "admin", "warehouse", "ns/t", false); err == nil {
 		t.Error("VendTableCredentials() error = nil, want an error when STS is not configured")
 	}
 }
@@ -122,7 +152,7 @@ func TestVendTableCredentialsNeedsSTS(t *testing.T) {
 // 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")
+	raw, err := tablePrefixSessionPolicy("warehouse", "analytics/events", false)
 	if err != nil {
 		t.Fatalf("tablePrefixSessionPolicy() error = %v", err)
 	}
@@ -168,11 +198,11 @@ func TestTablePrefixSessionPolicyRestrictsListing(t *testing.T) {
 // 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 {
+		if _, err := tablePrefixSessionPolicy("warehouse", prefix, false); err == nil {
 			t.Errorf("tablePrefixSessionPolicy(%q) error = nil, want a refusal", prefix)
 		}
 	}
-	if _, err := tablePrefixSessionPolicy("ware*house", "analytics/events"); err == nil {
+	if _, err := tablePrefixSessionPolicy("ware*house", "analytics/events", false); err == nil {
 		t.Error("a wildcard in the bucket name was accepted")
 	}
 }

weed/s3api/s3tables/handler_table.go

@@ -1465,40 +1465,51 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ
 	return nil
 }
 
-// handleUpdateTable updates table metadata
-func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error {
-	var req UpdateTableRequest
-	if err := h.readRequestBody(r, &req); err != nil {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
-	}
+// updateTableTarget is the table state an update was authorized against: the
+// stored values the optimistic-concurrency check compares and the fields the
+// response echoes back.
+type updateTableTarget struct {
+	bucketName     string
+	namespaceName  string
+	tableName      string
+	tablePath      string
+	tableARN       string
+	metadata       tableMetadataInternal
+	storedMetadata []byte
+	storedPolicy   []byte
+}
 
+// authorizeUpdateTable validates the names an update names, loads the table
+// and the policies governing it, and checks the caller may update it, writing
+// the error response itself. Credential vending asks the same question without
+// updating anything, so it runs this gate through Manager.AuthorizeUpdateTable.
+func (h *S3TablesHandler) authorizeUpdateTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient, req *UpdateTableRequest) (*updateTableTarget, error) {
 	if req.TableBucketARN == "" || len(req.Namespace) == 0 || req.Name == "" {
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "tableBucketARN, namespace, and name are required")
-		return fmt.Errorf("missing required parameters")
+		return nil, fmt.Errorf("missing required parameters")
 	}
 
 	namespaceName, err := validateNamespace(req.Namespace)
 	if err != nil {
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
+		return nil, err
 	}
 
 	bucketName, err := parseBucketNameFromARN(req.TableBucketARN)
 	if err != nil {
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
+		return nil, err
 	}
 
 	tableName, err := validateTableName(req.Name)
 	if err != nil {
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
+		return nil, err
 	}
 
 	if err := ValidateMetadataLocation(req.MetadataLocation, bucketName, namespaceName); err != nil {
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
+		return nil, err
 	}
 
 	tablePath := GetTablePath(bucketName, namespaceName, tableName)
@@ -1567,7 +1578,7 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		} else {
 			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, err.Error())
 		}
-		return err
+		return nil, err
 	}
 
 	// Authorization Check
@@ -1596,9 +1607,35 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		} else {
 			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, "table not found")
 		}
-		return NewAuthError("UpdateTable", principal, "not authorized to update table")
+		return nil, NewAuthError("UpdateTable", principal, "not authorized to update table")
 	}
 
+	return &updateTableTarget{
+		bucketName:     bucketName,
+		namespaceName:  namespaceName,
+		tableName:      tableName,
+		tablePath:      tablePath,
+		tableARN:       tableARN,
+		metadata:       metadata,
+		storedMetadata: storedMetadata,
+		storedPolicy:   storedPolicy,
+	}, nil
+}
+
+// handleUpdateTable updates table metadata
+func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error {
+	var req UpdateTableRequest
+	if err := h.readRequestBody(r, &req); err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return err
+	}
+
+	target, err := h.authorizeUpdateTable(w, r, filerClient, &req)
+	if err != nil {
+		return err
+	}
+	metadata := target.metadata
+
 	// Check version token if provided
 	if req.VersionToken != "" && req.VersionToken != metadata.VersionToken {
 		h.writeError(w, http.StatusConflict, ErrCodeConflict, "Version token mismatch")
@@ -1645,15 +1682,15 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 	// snapshot. mutateEntryExtended retries on a changed entry, so the check
 	// lives in the mutation, where it sees the value that is current now.
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return h.mutateEntryExtended(r.Context(), client, tablePath, func(extended map[string][]byte) error {
-			if !bytes.Equal(extended[ExtendedKeyMetadata], storedMetadata) {
+		return h.mutateEntryExtended(r.Context(), client, target.tablePath, func(extended map[string][]byte) error {
+			if !bytes.Equal(extended[ExtendedKeyMetadata], target.storedMetadata) {
 				return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata)
 			}
 			// The policy this request was authorized against must still be the
 			// one in force: an administrator restricting it mid-commit should
 			// send the caller back through authorization, not have its decision
 			// applied afterwards.
-			if !bytes.Equal(extended[ExtendedKeyPolicy], storedPolicy) {
+			if !bytes.Equal(extended[ExtendedKeyPolicy], target.storedPolicy) {
 				return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy)
 			}
 			extended[ExtendedKeyMetadata] = metadataBytes
@@ -1671,7 +1708,7 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 	}
 
 	h.writeJSON(w, http.StatusOK, &UpdateTableResponse{
-		TableARN:         tableARN,
+		TableARN:         target.tableARN,
 		MetadataLocation: metadata.MetadataLocation,
 		VersionToken:     metadata.VersionToken,
 	})

weed/s3api/s3tables/handler_view.go

@@ -396,23 +396,31 @@ func (h *S3TablesHandler) listViewsInNamespace(r *http.Request, client filer_pb.
 	return views, lastFileName, nil
 }
 
-// handleUpdateView replaces the metadata pointer for a view, mirroring UpdateTable.
-func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error {
-	var req UpdateViewRequest
-	if err := h.readRequestBody(r, &req); err != nil {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
-	}
+// updateViewTarget is the view state an update was authorized against: the
+// metadata the version-token check reads and the fields the response echoes.
+type updateViewTarget struct {
+	bucketName    string
+	namespaceName string
+	viewName      string
+	viewPath      string
+	viewARN       string
+	metadata      tableMetadataInternal
+}
 
+// authorizeUpdateView validates the names an update names, loads the view and
+// the policies governing it, and checks the caller may update it, writing the
+// error response itself. Credential vending asks the same question without
+// updating anything, so it runs this gate through Manager.AuthorizeUpdateView.
+func (h *S3TablesHandler) authorizeUpdateView(w http.ResponseWriter, r *http.Request, filerClient FilerClient, req *UpdateViewRequest) (*updateViewTarget, error) {
 	bucketName, namespaceName, viewName, err := h.parseViewRef(req.TableBucketARN, req.Namespace, req.Name)
 	if err != nil {
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
+		return nil, err
 	}
 
 	if err := ValidateMetadataLocation(req.MetadataLocation, bucketName, namespaceName); err != nil {
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
+		return nil, err
 	}
 
 	viewPath := GetTablePath(bucketName, namespaceName, viewName)
@@ -437,21 +445,46 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques
 		} else {
 			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, err.Error())
 		}
-		return err
+		return nil, err
 	}
 
 	accountID := h.getAccountID(r)
 	viewPolicy, bucketPolicy, bucketTags, bucketMetadata, err := h.loadAuthContext(r, filerClient, viewPath, GetTableBucketPath(bucketName))
 	if err != nil {
 		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, err.Error())
-		return err
+		return nil, err
 	}
 	viewARN := h.generateViewARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+viewName)
 	if !h.authorizeViewOp(r, "UpdateView", accountID, metadata.OwnerAccountID, bucketMetadata.OwnerAccountID, viewPolicy, bucketPolicy, viewARN, bucketName, namespaceName, viewName, bucketTags) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to update view")
-		return NewAuthError("UpdateView", accountID, "not authorized to update view")
+		return nil, NewAuthError("UpdateView", accountID, "not authorized to update view")
 	}
 
+	return &updateViewTarget{
+		bucketName:    bucketName,
+		namespaceName: namespaceName,
+		viewName:      viewName,
+		viewPath:      viewPath,
+		viewARN:       viewARN,
+		metadata:      metadata,
+	}, nil
+}
+
+// handleUpdateView replaces the metadata pointer for a view, mirroring UpdateTable.
+func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error {
+	var req UpdateViewRequest
+	if err := h.readRequestBody(r, &req); err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return err
+	}
+
+	target, err := h.authorizeUpdateView(w, r, filerClient, &req)
+	if err != nil {
+		return err
+	}
+	metadata := target.metadata
+	viewPath := target.viewPath
+
 	if req.VersionToken != "" && req.VersionToken != metadata.VersionToken {
 		h.writeError(w, http.StatusConflict, ErrCodeConflict, "Version token mismatch")
 		return ErrVersionTokenMismatch
@@ -499,7 +532,7 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques
 	}
 
 	h.writeJSON(w, http.StatusOK, &UpdateViewResponse{
-		ViewARN:          viewARN,
+		ViewARN:          target.viewARN,
 		MetadataLocation: metadata.MetadataLocation,
 		VersionToken:     metadata.VersionToken,
 	})

weed/s3api/s3tables/manager.go

@@ -81,6 +81,43 @@ func (m *Manager) AuthorizeCreateTable(ctx context.Context, filerClient FilerCli
 	return authErr
 }
 
+// AuthorizeUpdateTable checks that identity may update the table the request
+// describes, without updating anything. Credential vending gates on the same
+// permission a commit needs.
+func (m *Manager) AuthorizeUpdateTable(ctx context.Context, filerClient FilerClient, req *UpdateTableRequest, identity string) error {
+	httpReq, err := newManagerRequest(ctx, "UpdateTable", nil, identity)
+	if err != nil {
+		return err
+	}
+	recorder := httptest.NewRecorder()
+	_, authErr := m.handler.authorizeUpdateTable(recorder, httpReq, filerClient, req)
+	if authErr == nil {
+		return nil
+	}
+	if decoded := decodeS3TablesHTTPResponse(recorder, nil); decoded != nil {
+		return decoded
+	}
+	return authErr
+}
+
+// AuthorizeUpdateView checks that identity may update the view the request
+// describes, without updating anything.
+func (m *Manager) AuthorizeUpdateView(ctx context.Context, filerClient FilerClient, req *UpdateViewRequest, identity string) error {
+	httpReq, err := newManagerRequest(ctx, "UpdateView", nil, identity)
+	if err != nil {
+		return err
+	}
+	recorder := httptest.NewRecorder()
+	_, authErr := m.handler.authorizeUpdateView(recorder, httpReq, filerClient, req)
+	if authErr == nil {
+		return nil
+	}
+	if decoded := decodeS3TablesHTTPResponse(recorder, nil); decoded != nil {
+		return decoded
+	}
+	return authErr
+}
+
 func newManagerRequest(ctx context.Context, operation string, body []byte, identity string) (*http.Request, error) {
 	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, "/", bytes.NewReader(body))
 	if err != nil {