From 8d2b417c694d13a148e3a05a8780b837b612fcb0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 20 Oct 2026 07:16:27 -0700 Subject: [PATCH] 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 | 8 +- weed/s3api/iceberg/handlers_table.go | 57 +++++++++++++-- weed/s3api/iceberg/handlers_view.go | 8 +- weed/s3api/iceberg/handlers_view_update.go | 2 +- .../iceberg/iceberg_access_delegation_test.go | 4 +- .../iceberg_credential_vending_test.go | 36 +++++++-- weed/s3api/iceberg/iceberg_issue_9103_test.go | 4 +- weed/s3api/iceberg/server.go | 7 +- weed/s3api/lance/handlers_table.go | 8 +- weed/s3api/lance/server.go | 7 +- weed/s3api/lance/storage.go | 24 +++++- weed/s3api/s3api_iceberg_credentials.go | 25 ++++--- weed/s3api/s3api_iceberg_credentials_test.go | 48 +++++++++--- weed/s3api/s3tables/handler_table.go | 73 ++++++++++++++----- weed/s3api/s3tables/handler_view.go | 59 +++++++++++---- weed/s3api/s3tables/manager.go | 37 ++++++++++ 16 files changed, 324 insertions(+), 83 deletions(-) diff --git a/weed/command/s3.go b/weed/command/s3.go index 76308b24f4..036e3813eb 100644 --- a/weed/command/s3.go +++ b/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 } diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 855870ac47..6cdbf0e661 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/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) diff --git a/weed/s3api/iceberg/handlers_view.go b/weed/s3api/iceberg/handlers_view.go index 543655cc2f..0eb621b4dc 100644 --- a/weed/s3api/iceberg/handlers_view.go +++ b/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, diff --git a/weed/s3api/iceberg/handlers_view_update.go b/weed/s3api/iceberg/handlers_view_update.go index 60358829fe..e7edc7db2f 100644 --- a/weed/s3api/iceberg/handlers_view_update.go +++ b/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, diff --git a/weed/s3api/iceberg/iceberg_access_delegation_test.go b/weed/s3api/iceberg/iceberg_access_delegation_test.go index 3e024a9597..8e004a375a 100644 --- a/weed/s3api/iceberg/iceberg_access_delegation_test.go +++ b/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) } } diff --git a/weed/s3api/iceberg/iceberg_credential_vending_test.go b/weed/s3api/iceberg/iceberg_credential_vending_test.go index aee6a56a75..83f81a0124 100644 --- a/weed/s3api/iceberg/iceberg_credential_vending_test.go +++ b/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) diff --git a/weed/s3api/iceberg/iceberg_issue_9103_test.go b/weed/s3api/iceberg/iceberg_issue_9103_test.go index e321fa2af9..8fb72499b6 100644 --- a/weed/s3api/iceberg/iceberg_issue_9103_test.go +++ b/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") } diff --git a/weed/s3api/iceberg/server.go b/weed/s3api/iceberg/server.go index 8980ebc09d..5403654753 100644 --- a/weed/s3api/iceberg/server.go +++ b/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. diff --git a/weed/s3api/lance/handlers_table.go b/weed/s3api/lance/handlers_table.go index 40f23a2d88..7c63fd6ce8 100644 --- a/weed/s3api/lance/handlers_table.go +++ b/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 diff --git a/weed/s3api/lance/server.go b/weed/s3api/lance/server.go index 33ab48dc6e..936bd391e7 100644 --- a/weed/s3api/lance/server.go +++ b/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. diff --git a/weed/s3api/lance/storage.go b/weed/s3api/lance/storage.go index e04a1d83d1..c4ccd67385 100644 --- a/weed/s3api/lance/storage.go +++ b/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 { diff --git a/weed/s3api/s3api_iceberg_credentials.go b/weed/s3api/s3api_iceberg_credentials.go index 16b51a7180..830a2ca98f 100644 --- a/weed/s3api/s3api_iceberg_credentials.go +++ b/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)}, }, { diff --git a/weed/s3api/s3api_iceberg_credentials_test.go b/weed/s3api/s3api_iceberg_credentials_test.go index 33c5638f5c..77df42e8b6 100644 --- a/weed/s3api/s3api_iceberg_credentials_test.go +++ b/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") } } diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 7091dbda59..b9ce7d8834 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/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, }) diff --git a/weed/s3api/s3tables/handler_view.go b/weed/s3api/s3tables/handler_view.go index 3d2c98e6ab..d3327783f8 100644 --- a/weed/s3api/s3tables/handler_view.go +++ b/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, }) diff --git a/weed/s3api/s3tables/manager.go b/weed/s3api/s3tables/manager.go index 33dba70aee..301be1eb59 100644 --- a/weed/s3api/s3tables/manager.go +++ b/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 {