From d448e9db7be1081ab18d2e4dc9c3e0033ad3b8f8 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 4 Aug 2026 18:14:34 -0700 Subject: [PATCH] iceberg: withhold the S3 endpoint from credential-vending clients (#10570) * iceberg: withhold the S3 endpoint from credential-vending clients A client that sends X-Iceberg-Access-Delegation: vended-credentials builds its storage credential out of the LoadTable config and drops the one it was configured with. We vend no credentials, so the endpoint we advertised left DuckDB signing nothing: every metadata and data file came back 403, and its attempt to refresh the empty credential 404ed on stage-created tables. Answer those clients with no config at all so they keep their own credentials. Clients that do not ask for delegation still get the endpoint. * iceberg: mark load responses as varying on the delegation header The FileIO config in a table or view load response now depends on whether the client asked for vended credentials, so a cache between us and the client must key on that header rather than on the URL alone. * test: cover the DuckDB vended-credentials access pattern Runs weed mini with -s3.externalUrl, which is what makes the catalog advertise an endpoint at all, and checks both halves: a plain LoadTable still gets the endpoint, while one asking for vended credentials never gets an endpoint without the credentials to sign with. The DuckDB round trip creates a table from a query and reads it back, which is the flow that failed with 403 on every data file. --- test/s3tables/catalog/duckdb_oauth_test.go | 12 +- .../catalog/duckdb_vended_credentials_test.go | 215 ++++++++++++++++++ weed/s3api/iceberg/handlers_table.go | 50 ++-- weed/s3api/iceberg/handlers_view.go | 16 +- weed/s3api/iceberg/handlers_view_update.go | 4 +- .../iceberg/iceberg_access_delegation_test.go | 79 +++++++ .../iceberg/iceberg_create_table_test.go | 5 +- weed/s3api/iceberg/iceberg_issue_9103_test.go | 9 +- weed/s3api/iceberg/utils.go | 28 +++ 9 files changed, 382 insertions(+), 36 deletions(-) create mode 100644 test/s3tables/catalog/duckdb_vended_credentials_test.go create mode 100644 weed/s3api/iceberg/iceberg_access_delegation_test.go diff --git a/test/s3tables/catalog/duckdb_oauth_test.go b/test/s3tables/catalog/duckdb_oauth_test.go index afbb4886f..4d9eb1f1b 100644 --- a/test/s3tables/catalog/duckdb_oauth_test.go +++ b/test/s3tables/catalog/duckdb_oauth_test.go @@ -38,6 +38,9 @@ type oauthTestEnv struct { weedCancel context.CancelFunc accessKey string secretKey string + // s3ExternalURL, when set before start, is the S3 endpoint the catalog + // advertises to clients in LoadTable FileIO config. + s3ExternalURL string } func newOAuthTestEnv(t *testing.T) *oauthTestEnv { @@ -108,7 +111,7 @@ func (env *oauthTestEnv) start(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) env.weedCancel = cancel - cmd := exec.CommandContext(ctx, env.weedBinary, "mini", + args := []string{"mini", "-master.port", fmt.Sprintf("%d", env.masterPort), "-master.port.grpc", fmt.Sprintf("%d", env.masterGrpcPort), "-volume.port", fmt.Sprintf("%d", env.volumePort), @@ -123,7 +126,12 @@ func (env *oauthTestEnv) start(t *testing.T) { "-ip", env.bindIP, "-ip.bind", "0.0.0.0", "-dir", env.dataDir, - ) + } + if env.s3ExternalURL != "" { + args = append(args, "-s3.externalUrl", env.s3ExternalURL) + } + + cmd := exec.CommandContext(ctx, env.weedBinary, args...) cmd.Dir = env.dataDir cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr diff --git a/test/s3tables/catalog/duckdb_vended_credentials_test.go b/test/s3tables/catalog/duckdb_vended_credentials_test.go new file mode 100644 index 000000000..518bff33f --- /dev/null +++ b/test/s3tables/catalog/duckdb_vended_credentials_test.go @@ -0,0 +1,215 @@ +// Tests for the credential-vending access pattern. +// +// DuckDB attaches a catalog with "X-Iceberg-Access-Delegation: vended-credentials" +// and rebuilds its S3 credential out of whatever LoadTable returns in `config`, +// dropping the S3 secret it was configured with. A catalog that advertises an +// endpoint there but vends no credentials therefore leaves DuckDB sending +// unsigned requests, and every metadata and data file comes back 403. +// +// These run against a weed mini started with -s3.externalUrl, which is what +// makes the catalog advertise an endpoint at all; with a wildcard bind and no +// external URL the config is empty and the bug cannot appear. +package catalog + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/test/testutil" +) + +// duckDBImage tracks whatever DuckDB users are actually running, matching the +// other DuckDB tests here: a client that changes how it handles vended +// credentials is exactly what this suite exists to catch. +const duckDBImage = "duckdb/duckdb:latest" + +// requireDuckDBIceberg skips when the image cannot install the iceberg +// extension -- no egress, or a build without it. Probing separately keeps the +// round trip below free to fail on any error instead of having to guess which +// DuckDB messages mean "no extension" and which mean "the catalog is broken". +func requireDuckDBIceberg(t *testing.T) { + t.Helper() + + const ready = "iceberg extension ready" + cmd := exec.Command("docker", "run", "--rm", + "--entrypoint", "duckdb", + duckDBImage, + "-c", fmt.Sprintf("INSTALL iceberg; LOAD iceberg; SELECT '%s' AS marker;", ready), + ) + output, err := cmd.CombinedOutput() + if err != nil || !strings.Contains(string(output), ready) { + t.Skipf("DuckDB image cannot load the iceberg extension: %v\n%s", err, output) + } +} + +func TestIcebergVendedCredentials(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + env := newOAuthTestEnv(t) + // The DuckDB container reaches S3 through host.docker.internal, so advertise + // the endpoint under the name the client can actually resolve. + env.s3ExternalURL = fmt.Sprintf("http://host.docker.internal:%d", env.s3Port) + defer env.cleanup(t) + env.start(t) + + bucketName := "vendcreds-" + randomSuffix() + namespace := "vendcreds_ns_" + randomSuffix() + tableName := "vendcreds_tbl_" + randomSuffix() + + createTableBucketViaShell(t, env, bucketName) + token := requestOAuthToken(t, env, env.accessKey, env.secretKey) + createNamespaceWithToken(t, env, token, bucketName, namespace) + createTableWithToken(t, env, token, bucketName, namespace, tableName) + + t.Run("plain load table advertises the endpoint", func(t *testing.T) { + config := loadTableConfig(t, env, token, bucketName, namespace, tableName, "") + if config["s3.endpoint"] != env.s3ExternalURL { + t.Fatalf("s3.endpoint = %q, want %q (clients that bring their own credentials still need the endpoint)", + config["s3.endpoint"], env.s3ExternalURL) + } + }) + + t.Run("vended credentials request gets no half-configured storage", func(t *testing.T) { + config := loadTableConfig(t, env, token, bucketName, namespace, tableName, "vended-credentials") + if config["s3.endpoint"] == "" { + return + } + // If the catalog ever does vend credentials, the endpoint may come back + // -- but only alongside the credentials that make it usable. + if config["s3.access-key-id"] == "" || config["s3.secret-access-key"] == "" { + t.Fatalf("LoadTable config = %v: a client that asked for vended credentials drops its own and signs with what it gets, so an endpoint without credentials makes every data file read 403", config) + } + }) + + t.Run("duckdb writes and reads back through the catalog", func(t *testing.T) { + if !testutil.HasDocker() { + t.Skip("Docker not available, skipping DuckDB round trip") + } + requireDuckDBIceberg(t) + duckDBRoundTrip(t, env, bucketName, namespace) + }) +} + +// duckDBRoundTrip runs the reporter's flow: attach the catalog, create a table +// from a query, then read it back. Both halves need signed S3 access to the +// table's metadata and data files. +func duckDBRoundTrip(t *testing.T, env *oauthTestEnv, bucketName, namespace string) { + t.Helper() + + const marker = "vended-credentials round trip ok" + sql := fmt.Sprintf(` +INSTALL iceberg; +LOAD iceberg; + +CREATE SECRET iceberg_secret ( + TYPE ICEBERG, + ENDPOINT 'http://host.docker.internal:%d', + CLIENT_ID '%s', + CLIENT_SECRET '%s' +); + +CREATE SECRET s3_secret ( + TYPE S3, + KEY_ID '%s', + SECRET '%s', + ENDPOINT 'host.docker.internal:%d', + URL_STYLE 'path', + USE_SSL false +); + +ATTACH 's3://%s' AS vend_cat ( + TYPE ICEBERG, + SECRET iceberg_secret, + ENDPOINT 'http://host.docker.internal:%d', + READ_ONLY false +); + +CREATE TABLE vend_cat.%s.round_trip AS SELECT 42 AS answer; + +SELECT CASE WHEN sum(answer) = 42 THEN '%s' ELSE 'wrong answer' END AS marker +FROM vend_cat.%s.round_trip; +`, + env.icebergPort, env.accessKey, env.secretKey, + env.accessKey, env.secretKey, env.s3Port, + bucketName, env.icebergPort, + namespace, marker, namespace, + ) + + sqlFile := filepath.Join(env.dataDir, "duckdb_vended_credentials.sql") + if err := os.WriteFile(sqlFile, []byte(sql), 0644); err != nil { + t.Fatalf("write SQL file: %v", err) + } + + cmd := exec.Command("docker", "run", "--rm", + "-v", fmt.Sprintf("%s:/test", env.dataDir), + "--add-host", "host.docker.internal:host-gateway", + "-e", "AWS_REGION=us-east-1", + "--entrypoint", "duckdb", + duckDBImage, + "-init", "/test/duckdb_vended_credentials.sql", + "-c", "SELECT 1", + ) + output, err := cmd.CombinedOutput() + outputStr := string(output) + t.Logf("DuckDB output:\n%s", outputStr) + + // Nothing below skips: requireDuckDBIceberg already established that the + // extension loads, so any failure from here is the catalog's. + if strings.Contains(outputStr, "trying to refresh secret") { + t.Fatalf("DuckDB fell back to refreshing the credential the catalog vended, which no stage-created table can satisfy:\n%s", outputStr) + } + // Match the phrasings rather than a bare "403", which a random port could + // carry. + for _, denial := range []string{"AccessDenied", "403 Forbidden", "code 403"} { + if strings.Contains(outputStr, denial) { + t.Fatalf("DuckDB signed its S3 requests with the catalog's credential-less config:\n%s", outputStr) + } + } + if !strings.Contains(outputStr, marker) { + t.Fatalf("DuckDB did not write and read the table back (err=%v):\n%s", err, outputStr) + } +} + +// loadTableConfig loads a table and returns the FileIO config the catalog +// advertises, optionally asking for an access delegation mechanism. +func loadTableConfig(t *testing.T, env *oauthTestEnv, token, bucketName, namespace, tableName, delegation string) map[string]string { + t.Helper() + + url := fmt.Sprintf("%s/v1/%s/namespaces/%s/tables/%s", env.icebergURL(), bucketName, namespace, tableName) + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + t.Fatalf("create request: %v", err) + } + req.Header.Set("Authorization", "Bearer "+token) + if delegation != "" { + req.Header.Set("X-Iceberg-Access-Delegation", delegation) + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("load table: %v", err) + } + defer resp.Body.Close() + + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + t.Fatalf("load table failed: status=%d body=%s", resp.StatusCode, body) + } + + var result struct { + Config map[string]string `json:"config"` + } + if err := json.Unmarshal(body, &result); err != nil { + t.Fatalf("decode LoadTableResult: %v", err) + } + return result.Config +} diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 309547c4a..616a6ac25 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -246,13 +246,13 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { if existsErr == nil { // Table already registered. Return the existing definition so CTAS/IF NOT // EXISTS flows see a stable response instead of a 409. - result, buildErr := s.buildLoadTableResult(existsResp, bucketName, namespace, tableName) + result, buildErr := s.buildLoadTableResult(r, existsResp, bucketName, namespace, tableName) if buildErr != nil { glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr) writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") return } - writeJSON(w, http.StatusOK, result) + writeLoadResult(w, http.StatusOK, result) return } if !isNoSuchTableError(existsErr) { @@ -275,9 +275,9 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { result := LoadTableResult{ MetadataLocation: metadataLocation, Metadata: metadata, - Config: s.buildFileIOConfig(), + Config: s.buildFileIOConfig(r), } - writeJSON(w, http.StatusOK, result) + writeLoadResult(w, http.StatusOK, result) return } if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil { @@ -324,13 +324,13 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error()) return } - result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, tableName) + result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, tableName) if buildErr != nil { glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr) writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") return } - writeJSON(w, http.StatusOK, result) + writeLoadResult(w, http.StatusOK, result) return } if strings.Contains(err.Error(), "already exists") { @@ -348,13 +348,13 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error()) return } - result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, tableName) + result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, tableName) if buildErr != nil { glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr) writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") return } - writeJSON(w, http.StatusOK, result) + writeLoadResult(w, http.StatusOK, result) return } glog.V(1).Infof("Iceberg: CreateTable error: %v", err) @@ -374,9 +374,9 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { result := LoadTableResult{ MetadataLocation: finalLocation, Metadata: metadata, - Config: s.buildFileIOConfig(), + Config: s.buildFileIOConfig(r), } - writeJSON(w, http.StatusOK, result) + writeLoadResult(w, http.StatusOK, result) } // handleRegisterTable registers an existing metadata.json under a new catalog @@ -465,13 +465,13 @@ func (s *Server) handleRegisterTable(w http.ResponseWriter, r *http.Request) { MetadataLocation: req.MetadataLocation, Metadata: &s3tables.TableMetadata{FullMetadata: json.RawMessage(metadataBytes)}, } - result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, req.Name) + result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, req.Name) if buildErr != nil { glog.Errorf("Iceberg: RegisterTable %s: %v", req.Name, buildErr) writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") return } - writeJSON(w, http.StatusOK, result) + writeLoadResult(w, http.StatusOK, result) } // handleLoadTable loads table metadata. @@ -514,16 +514,16 @@ func (s *Server) handleLoadTable(w http.ResponseWriter, r *http.Request) { return } - result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, tableName) + result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, tableName) if buildErr != nil { glog.Errorf("Iceberg: LoadTable %s: %v", tableName, buildErr) writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") return } - writeJSON(w, http.StatusOK, result) + writeLoadResult(w, http.StatusOK, result) } -func (s *Server) buildLoadTableResult(getResp s3tables.GetTableResponse, bucketName string, namespace []string, tableName string) (LoadTableResult, error) { +func (s *Server) buildLoadTableResult(r *http.Request, getResp s3tables.GetTableResponse, bucketName string, namespace []string, tableName string) (LoadTableResult, error) { location := tableLocationFromMetadataLocation(getResp.MetadataLocation) if location == "" { location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName)) @@ -562,7 +562,7 @@ func (s *Server) buildLoadTableResult(getResp s3tables.GetTableResponse, bucketN return LoadTableResult{ MetadataLocation: getResp.MetadataLocation, Metadata: metadata, - Config: s.buildFileIOConfig(), + Config: s.buildFileIOConfig(r), }, nil } @@ -571,13 +571,21 @@ func (s *Server) buildLoadTableResult(getResp s3tables.GetTableResponse, bucketN // separately discovering the endpoint. The region defaults to the same // value baked into table bucket ARNs so clients like DuckDB that require // a region on attach don't need to be told it out-of-band. See issue #9103. -func (s *Server) buildFileIOConfig() iceberg.Properties { +func (s *Server) buildFileIOConfig(r *http.Request) iceberg.Properties { config := make(iceberg.Properties) - if s.s3Endpoint != "" { - config["s3.endpoint"] = s.s3Endpoint - config["s3.path-style-access"] = "true" - config["s3.region"] = s3tables.DefaultRegion + if s.s3Endpoint == "" { + return config } + // 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. We vend none, so an endpoint on its own leaves it sending unsigned + // requests; say nothing instead and let it keep its own credentials. + if wantsVendedCredentials(r) { + return config + } + config["s3.endpoint"] = s.s3Endpoint + config["s3.path-style-access"] = "true" + config["s3.region"] = s3tables.DefaultRegion return config } diff --git a/weed/s3api/iceberg/handlers_view.go b/weed/s3api/iceberg/handlers_view.go index daa4151e0..ba3d29a45 100644 --- a/weed/s3api/iceberg/handlers_view.go +++ b/weed/s3api/iceberg/handlers_view.go @@ -158,12 +158,12 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) { // short-circuits with its stored definition (idempotent CreateView) so we // never overwrite the persisted metadata of an existing view. if existsResp, existsErr := s.getView(r, namespace, req.Name); existsErr == nil { - result, buildErr := s.buildViewResponse(existsResp, bucketName, namespace, req.Name) + result, buildErr := s.buildViewResponse(r, existsResp, bucketName, namespace, req.Name) if buildErr != nil { writeError(w, http.StatusInternalServerError, "InternalServerError", buildErr.Error()) return } - writeJSON(w, http.StatusOK, result) + writeLoadResult(w, http.StatusOK, result) return } else if !isViewNotFound(existsErr) { glog.V(1).Infof("Iceberg: CreateView existence check failed for %s.%s: %v", flattenNamespacePath(namespace), req.Name, existsErr) @@ -218,10 +218,10 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) { if finalLocation == "" { finalLocation = metadataLocation } - writeJSON(w, http.StatusOK, ViewResponse{ + writeLoadResult(w, http.StatusOK, ViewResponse{ MetadataLocation: finalLocation, Metadata: metadata, - Config: s.buildFileIOConfig(), + Config: s.buildFileIOConfig(r), }) } @@ -246,12 +246,12 @@ func (s *Server) handleLoadView(w http.ResponseWriter, r *http.Request) { return } - result, err := s.buildViewResponse(getResp, getBucketFromPrefix(r), namespace, viewName) + result, err := s.buildViewResponse(r, getResp, getBucketFromPrefix(r), namespace, viewName) if err != nil { writeError(w, http.StatusInternalServerError, "InternalServerError", err.Error()) return } - writeJSON(w, http.StatusOK, result) + writeLoadResult(w, http.StatusOK, result) } // handleViewExists checks if a view exists. @@ -340,7 +340,7 @@ func (s *Server) getView(r *http.Request, namespace []string, viewName string) ( } // buildViewResponse parses the stored view metadata into a ViewResponse. -func (s *Server) buildViewResponse(getResp s3tables.GetViewResponse, bucketName string, namespace []string, viewName string) (ViewResponse, error) { +func (s *Server) buildViewResponse(r *http.Request, getResp s3tables.GetViewResponse, bucketName string, namespace []string, viewName string) (ViewResponse, error) { if getResp.Metadata == nil || len(getResp.Metadata.FullMetadata) == 0 { return ViewResponse{}, fmt.Errorf("view %s has no metadata", viewName) } @@ -351,7 +351,7 @@ func (s *Server) buildViewResponse(getResp s3tables.GetViewResponse, bucketName return ViewResponse{ MetadataLocation: getResp.MetadataLocation, Metadata: metadata, - Config: s.buildFileIOConfig(), + Config: s.buildFileIOConfig(r), }, nil } diff --git a/weed/s3api/iceberg/handlers_view_update.go b/weed/s3api/iceberg/handlers_view_update.go index 001a641f3..73ebcadd2 100644 --- a/weed/s3api/iceberg/handlers_view_update.go +++ b/weed/s3api/iceberg/handlers_view_update.go @@ -130,10 +130,10 @@ func (s *Server) handleUpdateView(w http.ResponseWriter, r *http.Request) { return s.tablesManager.Execute(r.Context(), mgrClient, "UpdateView", updateReq, nil, identityName) }) if err == nil { - writeJSON(w, http.StatusOK, ViewResponse{ + writeLoadResult(w, http.StatusOK, ViewResponse{ MetadataLocation: newMetadataLocation, Metadata: newMetadata, - Config: s.buildFileIOConfig(), + Config: s.buildFileIOConfig(r), }) return } diff --git a/weed/s3api/iceberg/iceberg_access_delegation_test.go b/weed/s3api/iceberg/iceberg_access_delegation_test.go new file mode 100644 index 000000000..51a1931a4 --- /dev/null +++ b/weed/s3api/iceberg/iceberg_access_delegation_test.go @@ -0,0 +1,79 @@ +package iceberg + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +func TestWantsVendedCredentials(t *testing.T) { + tests := []struct { + name string + values []string + want bool + }{ + {name: "no header", want: false}, + {name: "vended credentials", values: []string{"vended-credentials"}, want: true}, + {name: "mechanism list", values: []string{"remote-signing,vended-credentials"}, want: true}, + {name: "spaced and mixed case", values: []string{"Remote-Signing, Vended-Credentials"}, want: true}, + {name: "repeated header", values: []string{"remote-signing", "vended-credentials"}, want: true}, + {name: "remote signing only", values: []string{"remote-signing"}, want: false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil) + for _, v := range tc.values { + r.Header.Add(accessDelegationHeader, v) + } + if got := wantsVendedCredentials(r); got != tc.want { + t.Fatalf("wantsVendedCredentials(%v) = %v, want %v", tc.values, got, tc.want) + } + }) + } +} + +// A client that asked for vended credentials replaces its own storage +// credentials with whatever the catalog returns. Since the catalog vends none, +// an endpoint by itself would leave the client signing nothing and every data +// file read and write would come back 403. +func TestBuildFileIOConfigWithholdsEndpointFromCredentialVendingClients(t *testing.T) { + s := &Server{s3Endpoint: "http://seaweed.example:8333"} + + r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil) + r.Header.Set(accessDelegationHeader, "vended-credentials") + if got := s.buildFileIOConfig(r); 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); got["s3.endpoint"] != s.s3Endpoint { + t.Fatalf("s3.endpoint = %q, want %q", got["s3.endpoint"], s.s3Endpoint) + } +} + +// The body of a load response depends on the delegation header, so a cache in +// front of the catalog must not key on the URL alone. +func TestWriteLoadResultVariesOnTheDelegationHeader(t *testing.T) { + rec := httptest.NewRecorder() + writeLoadResult(rec, http.StatusOK, LoadTableResult{}) + if got := rec.Header().Get("Vary"); got != accessDelegationHeader { + t.Fatalf("Vary = %q, want %q", got, accessDelegationHeader) + } +} + +func TestLoadTableResultOmitsConfigForCredentialVendingClients(t *testing.T) { + s := &Server{s3Endpoint: "http://seaweed.example:8333"} + r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil) + r.Header.Set(accessDelegationHeader, "vended-credentials") + + getResp := s3tables.GetTableResponse{MetadataLocation: "s3://bkt/ns/t/metadata/v1.metadata.json"} + result, err := s.buildLoadTableResult(r, getResp, "bkt", []string{"ns"}, "t") + if err != nil { + t.Fatalf("buildLoadTableResult() error = %v", err) + } + if len(result.Config) != 0 { + t.Fatalf("LoadTableResult.Config = %v, want empty", result.Config) + } +} diff --git a/weed/s3api/iceberg/iceberg_create_table_test.go b/weed/s3api/iceberg/iceberg_create_table_test.go index 721cb463e..ea09e4252 100644 --- a/weed/s3api/iceberg/iceberg_create_table_test.go +++ b/weed/s3api/iceberg/iceberg_create_table_test.go @@ -3,6 +3,8 @@ package iceberg import ( "encoding/json" "errors" + "net/http" + "net/http/httptest" "strings" "testing" @@ -96,7 +98,8 @@ func TestBuildLoadTableResultNeverReturnsNilMetadata(t *testing.T) { } for name, getResp := range cases { t.Run(name, func(t *testing.T) { - result, err := (&Server{}).buildLoadTableResult(getResp, "bkt", []string{"ns"}, "t") + r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil) + result, err := (&Server{}).buildLoadTableResult(r, getResp, "bkt", []string{"ns"}, "t") if err != nil { t.Fatalf("buildLoadTableResult() error = %v, want nil", err) } diff --git a/weed/s3api/iceberg/iceberg_issue_9103_test.go b/weed/s3api/iceberg/iceberg_issue_9103_test.go index 691779148..555d09c9e 100644 --- a/weed/s3api/iceberg/iceberg_issue_9103_test.go +++ b/weed/s3api/iceberg/iceberg_issue_9103_test.go @@ -1,6 +1,7 @@ package iceberg import ( + "net/http" "net/http/httptest" "testing" ) @@ -54,9 +55,13 @@ func TestGetBucketFromPrefix_WarehouseQueryFallback(t *testing.T) { } func TestBuildFileIOConfig(t *testing.T) { + loadTable := func() *http.Request { + return httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil) + } + t.Run("no endpoint configured yields empty config", func(t *testing.T) { s := &Server{} - got := s.buildFileIOConfig() + got := s.buildFileIOConfig(loadTable()) if len(got) != 0 { t.Fatalf("buildFileIOConfig() = %v, want empty", got) } @@ -64,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() + got := s.buildFileIOConfig(loadTable()) 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/utils.go b/weed/s3api/iceberg/utils.go index 52f4aa6c7..84f02c47a 100644 --- a/weed/s3api/iceberg/utils.go +++ b/weed/s3api/iceberg/utils.go @@ -75,6 +75,34 @@ func tableLocationFromMetadataLocation(metadataLocation string) string { return trimmed } +// accessDelegationHeader is how a client asks the catalog to hand back storage +// credentials along with the table metadata. +const accessDelegationHeader = "X-Iceberg-Access-Delegation" + +// wantsVendedCredentials reports whether the client asked for vended +// credentials. The header carries a comma-separated list of mechanisms. +func wantsVendedCredentials(r *http.Request) bool { + if r == nil { + return false + } + for _, value := range r.Header.Values(accessDelegationHeader) { + for _, mechanism := range strings.Split(value, ",") { + if strings.EqualFold(strings.TrimSpace(mechanism), "vended-credentials") { + return true + } + } + } + return false +} + +// writeLoadResult writes a table or view load response. The FileIO config it +// carries depends on the delegation the client asked for, so a cache between +// us and the client must key on that header and not on the URL alone. +func writeLoadResult(w http.ResponseWriter, status int, v interface{}) { + w.Header().Set("Vary", accessDelegationHeader) + writeJSON(w, status, v) +} + // writeJSON writes a JSON response. func writeJSON(w http.ResponseWriter, status int, v interface{}) { w.Header().Set("Content-Type", "application/json")