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")