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 {