diff --git a/seaweed-worker/crates/lance/src/catalog/namespace.rs b/seaweed-worker/crates/lance/src/catalog/namespace.rs index c026ae2c69..61a0e099f3 100644 --- a/seaweed-worker/crates/lance/src/catalog/namespace.rs +++ b/seaweed-worker/crates/lance/src/catalog/namespace.rs @@ -34,6 +34,7 @@ struct ListTablesResponse { pub struct NamespaceClient { base_url: String, http: reqwest::Client, + api_key: Option, } /// A namespace call that has not answered by now is not going to. Without this @@ -41,12 +42,23 @@ pub struct NamespaceClient { /// slot open forever, and the sweep never finishes. const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +const MAX_REDIRECT_HOPS: usize = 10; impl NamespaceClient { pub fn new(base_url: impl Into) -> Self { let http = reqwest::Client::builder() .timeout(REQUEST_TIMEOUT) .connect_timeout(CONNECT_TIMEOUT) + // reqwest forwards custom headers on a follow, so only a same-origin + // hop is allowed: anywhere else would carry x-api-key to a listener + // the credential was never meant for. + .redirect(reqwest::redirect::Policy::custom(|attempt| { + if may_follow(attempt.previous(), attempt.url()) { + attempt.follow() + } else { + attempt.stop() + } + })) .build() // The builder only fails on a bad TLS backend, which would break // every call anyway; a client with no timeouts is worse than a panic @@ -58,6 +70,32 @@ impl NamespaceClient { Self { base_url: base_url.into().trim_end_matches('/').to_string(), http, + api_key: None, + } + } + + /// The catalog accepts an S3 credential as `x-api-key`; where the worker + /// already holds one for storage it authenticates the namespace calls too. + pub fn with_fallback(mut self, fallback: &crate::dataset::FallbackOptions) -> Self { + if let (Some(access), Some(secret)) = ( + fallback.get("aws_access_key_id"), + fallback.get("aws_secret_access_key"), + ) { + if !protected_transport(&self.base_url) { + tracing::warn!( + "catalog url {} exposes the x-api-key credential in cleartext", + self.base_url + ); + } + self.api_key = Some(format!("{access}:{secret}")); + } + self + } + + fn authed(&self, request: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + match &self.api_key { + Some(key) => request.header("x-api-key", key), + None => request, } } @@ -65,8 +103,7 @@ impl NamespaceClient { pub async fn list_all_tables(&self) -> Result> { let url = format!("{}/v1/table", self.base_url); let response: ListTablesResponse = self - .http - .get(&url) + .authed(self.http.get(&url)) .send() .await .context("list tables")? @@ -80,8 +117,7 @@ impl NamespaceClient { pub async fn list_namespaces(&self, id: &[String]) -> Result> { let url = format!("{}/v1/namespace/{}/list", self.base_url, encode_id(id)); let response: ListNamespacesResponse = self - .http - .get(&url) + .authed(self.http.get(&url)) .send() .await .context("list namespaces")? @@ -98,8 +134,7 @@ impl NamespaceClient { let url = format!("{}/v1/table/{}/describe", self.base_url, encode_id(id)); let body = serde_json::json!({ "id": id, "vend_credentials": true }); let description: TableDescription = self - .http - .post(&url) + .authed(self.http.post(&url)) .json(&body) .send() .await @@ -111,6 +146,41 @@ impl NamespaceClient { } } +/// Without a hop bound a catalog that redirects to itself keeps the request +/// alive until the call timeout, so the custom policy keeps the limit the +/// default one had. +fn may_follow(previous: &[reqwest::Url], next: &reqwest::Url) -> bool { + previous.len() < MAX_REDIRECT_HOPS + && previous.last().is_some_and(|prev| same_origin(prev, next)) +} + +/// A redirect stays within an origin when scheme, host, and port all match, so +/// a catalog cannot bounce the api key onto a downgraded or foreign listener. +fn same_origin(from: &reqwest::Url, to: &reqwest::Url) -> bool { + from.scheme() == to.scheme() + && from.host_str() == to.host_str() + && from.port_or_known_default() == to.port_or_known_default() +} + +/// A catalog url protects the api key when the hop is TLS or never leaves the +/// host; anything else sends it where the network can read it. +fn protected_transport(base_url: &str) -> bool { + let Ok(url) = reqwest::Url::parse(base_url) else { + return false; + }; + if url.scheme() == "https" { + return true; + } + match url.host_str() { + Some("localhost") => true, + Some(host) => host + .trim_matches(['[', ']']) + .parse::() + .is_ok_and(|ip| ip.is_loopback()), + None => false, + } +} + fn encode_id(id: &[String]) -> String { if id.is_empty() { DELIMITER.to_string() @@ -127,3 +197,56 @@ pub fn parse_id(encoded: &str) -> Vec { .map(|part| part.to_string()) .collect() } + +#[cfg(test)] +mod tests { + use super::{MAX_REDIRECT_HOPS, may_follow, protected_transport, same_origin}; + + #[test] + fn may_follow_bounds_hops_within_the_origin() { + let url = |u: &str| reqwest::Url::parse(u).unwrap(); + let first = [url("https://c.local/ns")]; + let next = url("https://c.local/other"); + assert!(may_follow(&first, &next)); + assert!(!may_follow(&[], &next)); + assert!(!may_follow(&first, &url("http://c.local/other"))); + let full: Vec<_> = (0..MAX_REDIRECT_HOPS) + .map(|i| url(&format!("https://c.local/{i}"))) + .collect(); + assert!(!may_follow(&full, &next)); + } + + #[test] + fn same_origin_rejects_scheme_and_port_hops() { + for (from, to, expected) in [ + ("https://c.local/ns", "https://c.local/other", true), + ("http://c.local", "http://c.local:80/next", true), + ("https://c.local", "http://c.local/next", false), + ("https://c.local:8443", "https://c.local/next", false), + ("https://c.local", "https://other.local/next", false), + ] { + let from = reqwest::Url::parse(from).unwrap(); + let to = reqwest::Url::parse(to).unwrap(); + assert_eq!(same_origin(&from, &to), expected, "{from} -> {to}"); + } + } + + #[test] + fn protected_transport_covers_tls_and_loopback() { + for url in [ + "https://catalog.internal:8443", + "http://127.0.0.1:8080", + "http://localhost", + "http://[::1]:8080", + ] { + assert!(protected_transport(url), "{url} should be protected"); + } + for url in [ + "http://catalog.internal", + "http://127.0.0.1.evil.test", + "not a url", + ] { + assert!(!protected_transport(url), "{url} should warn"); + } + } +} diff --git a/seaweed-worker/crates/lance/src/jobs/cleanup.rs b/seaweed-worker/crates/lance/src/jobs/cleanup.rs index 608187ad48..8b84c4a738 100644 --- a/seaweed-worker/crates/lance/src/jobs/cleanup.rs +++ b/seaweed-worker/crates/lance/src/jobs/cleanup.rs @@ -55,7 +55,7 @@ impl CleanupVersionsHandler { } fn client(&self) -> NamespaceClient { - NamespaceClient::new(self.namespace_url.clone()) + NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback) } } diff --git a/seaweed-worker/crates/lance/src/jobs/compact.rs b/seaweed-worker/crates/lance/src/jobs/compact.rs index 2cad969630..da3339add4 100644 --- a/seaweed-worker/crates/lance/src/jobs/compact.rs +++ b/seaweed-worker/crates/lance/src/jobs/compact.rs @@ -59,7 +59,7 @@ impl CompactHandler { } fn client(&self) -> NamespaceClient { - NamespaceClient::new(self.namespace_url.clone()) + NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback) } } diff --git a/seaweed-worker/crates/lance/src/jobs/indices.rs b/seaweed-worker/crates/lance/src/jobs/indices.rs index 927748b9b3..5ca17224b3 100644 --- a/seaweed-worker/crates/lance/src/jobs/indices.rs +++ b/seaweed-worker/crates/lance/src/jobs/indices.rs @@ -58,7 +58,7 @@ impl OptimizeIndicesHandler { } fn client(&self) -> NamespaceClient { - NamespaceClient::new(self.namespace_url.clone()) + NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback) } } diff --git a/seaweed-worker/crates/lance/src/jobs/sort.rs b/seaweed-worker/crates/lance/src/jobs/sort.rs index 0bbf9b0639..2e48eab0ae 100644 --- a/seaweed-worker/crates/lance/src/jobs/sort.rs +++ b/seaweed-worker/crates/lance/src/jobs/sort.rs @@ -78,7 +78,7 @@ impl SortHandler { } fn client(&self) -> NamespaceClient { - NamespaceClient::new(self.namespace_url.clone()) + NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback) } } diff --git a/seaweed-worker/crates/lance/src/preview.rs b/seaweed-worker/crates/lance/src/preview.rs index 092791b69f..596d8700a0 100644 --- a/seaweed-worker/crates/lance/src/preview.rs +++ b/seaweed-worker/crates/lance/src/preview.rs @@ -35,7 +35,7 @@ impl PreviewProvider for LancePreview { } async fn preview(&self, object_id: &[String], row_limit: usize) -> Result { - let client = NamespaceClient::new(self.namespace_url.clone()); + let client = NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback); let table = dataset::open(&client, object_id, &self.fallback).await?; let total_rows = table.dataset.count_rows(None).await.unwrap_or(0) as i64; diff --git a/test/s3tables/catalog_duckdb_lance/duckdb_lance_test.go b/test/s3tables/catalog_duckdb_lance/duckdb_lance_test.go index 5cefd12059..99a1a20372 100644 --- a/test/s3tables/catalog_duckdb_lance/duckdb_lance_test.go +++ b/test/s3tables/catalog_duckdb_lance/duckdb_lance_test.go @@ -244,7 +244,13 @@ func (env *environment) tableBucketExists(t *testing.T, bucket string) bool { t.Helper() url := fmt.Sprintf("http://%s:%d/v1/namespace/%s/exists", env.bindIP, env.lancePort, bucket) - resp, err := http.Post(url, "application/json", strings.NewReader("{}")) + req, err := http.NewRequest(http.MethodPost, url, strings.NewReader("{}")) + if err != nil { + t.Fatalf("build the namespace exists request: %v", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("x-api-key", env.accessKey+":"+env.secretKey) + resp, err := http.DefaultClient.Do(req) if err != nil { t.Fatalf("ask the namespace whether %s exists: %v", bucket, err) } diff --git a/test/s3tables/catalog_duckdb_lance/seed_table.py b/test/s3tables/catalog_duckdb_lance/seed_table.py index 051f2f519f..82929eea8f 100644 --- a/test/s3tables/catalog_duckdb_lance/seed_table.py +++ b/test/s3tables/catalog_duckdb_lance/seed_table.py @@ -53,7 +53,13 @@ def main(): "aws_region": "us-east-1", } - ns = ln.connect("rest", {"uri": args.namespace_url}) + ns = ln.connect( + "rest", + { + "uri": args.namespace_url, + "header.x-api-key": f"{args.access_key}:{args.secret_key}", + }, + ) ns.create_namespace(ln.CreateNamespaceRequest(id=[args.bucket], mode="EXIST_OK")) ns.create_namespace( ln.CreateNamespaceRequest(id=[args.bucket, args.namespace], mode="EXIST_OK") diff --git a/test/s3tables/catalog_lancedb/lancedb_ops.py b/test/s3tables/catalog_lancedb/lancedb_ops.py index 9f8fe1d447..ff7622d165 100644 --- a/test/s3tables/catalog_lancedb/lancedb_ops.py +++ b/test/s3tables/catalog_lancedb/lancedb_ops.py @@ -45,7 +45,15 @@ def seed_table(namespace_url, storage, bucket, namespace, table, rows): LanceDB reads through the catalog; the writing half is pylance, because the namespace records where a table lives and does not carry its data. """ - ns = ln.connect("rest", {"uri": namespace_url}) + ns = ln.connect( + "rest", + { + "uri": namespace_url, + "header.x-api-key": ( + f"{storage['aws_access_key_id']}:{storage['aws_secret_access_key']}" + ), + }, + ) ns.create_namespace( ln.CreateNamespaceRequest(id=[bucket], mode="EXIST_OK") ) @@ -98,7 +106,12 @@ def main(): print(f"lancedb {lancedb.__version__} connecting to {args.namespace_url}") db = lancedb.connect_namespace( - "rest", {"uri": args.namespace_url}, storage_options=storage + "rest", + { + "uri": args.namespace_url, + "header.x-api-key": f"{args.access_key}:{args.secret_key}", + }, + storage_options=storage, ) # 1. The catalog is browsable: the bucket is a namespace, and the table is @@ -190,7 +203,10 @@ def main(): # declare-and-write - rather than a hang, a 404, or a half-made table. pushdown = lancedb.connect_namespace( "rest", - {"uri": args.namespace_url}, + { + "uri": args.namespace_url, + "header.x-api-key": f"{args.access_key}:{args.secret_key}", + }, storage_options=storage, namespace_client_pushdown_operations=["CreateTable"], ) diff --git a/test/s3tables/catalog_spark_lance/README.md b/test/s3tables/catalog_spark_lance/README.md index 7ba408b809..a3162ba0b8 100644 --- a/test/s3tables/catalog_spark_lance/README.md +++ b/test/s3tables/catalog_spark_lance/README.md @@ -41,6 +41,7 @@ deployment uses whatever address the gateway answers on: spark.sql.catalog.lance org.lance.spark.LanceNamespaceSparkCatalog spark.sql.catalog.lance.impl rest spark.sql.catalog.lance.uri http://seaweed:9101 +spark.sql.catalog.lance.headers.x-api-key : spark.sql.catalog.lance.storage.aws_endpoint http://seaweed:8333 spark.sql.catalog.lance.storage.allow_http true spark.sql.catalog.lance.storage.aws_access_key_id … diff --git a/test/s3tables/catalog_spark_lance/spark_lance_ops.py b/test/s3tables/catalog_spark_lance/spark_lance_ops.py index bb824e2ad1..b59a10a957 100644 --- a/test/s3tables/catalog_spark_lance/spark_lance_ops.py +++ b/test/s3tables/catalog_spark_lance/spark_lance_ops.py @@ -46,6 +46,10 @@ def main(): .config("spark.sql.catalog.lance", "org.lance.spark.LanceNamespaceSparkCatalog") .config("spark.sql.catalog.lance.impl", "rest") .config("spark.sql.catalog.lance.uri", args.namespace_url) + .config( + "spark.sql.catalog.lance.headers.x-api-key", + f"{args.access_key}:{args.secret_key}", + ) .config("spark.sql.catalog.lance.storage.aws_endpoint", args.s3_endpoint) .config("spark.sql.catalog.lance.storage.allow_http", "true") .config("spark.sql.catalog.lance.storage.aws_access_key_id", args.access_key) diff --git a/test/s3tables/catalog_spark_lance/spark_lance_test.go b/test/s3tables/catalog_spark_lance/spark_lance_test.go index 5bf1a0359f..f7e4468f62 100644 --- a/test/s3tables/catalog_spark_lance/spark_lance_test.go +++ b/test/s3tables/catalog_spark_lance/spark_lance_test.go @@ -239,7 +239,13 @@ func (env *environment) tableBucketExists(t *testing.T, bucket string) bool { t.Helper() url := fmt.Sprintf("http://%s:%d/v1/namespace/%s/exists", env.bindIP, env.lancePort, bucket) - resp, err := http.Post(url, "application/json", strings.NewReader("{}")) + req, err := http.NewRequest(http.MethodPost, url, strings.NewReader("{}")) + if err != nil { + t.Fatalf("build the namespace exists request: %v", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("x-api-key", env.accessKey+":"+env.secretKey) + resp, err := http.DefaultClient.Do(req) if err != nil { t.Fatalf("ask the namespace whether %s exists: %v", bucket, err) } diff --git a/test/s3tables/lifecycle/iceberg_lifecycle.py b/test/s3tables/lifecycle/iceberg_lifecycle.py index d8965abd7b..1d85bc658e 100644 --- a/test/s3tables/lifecycle/iceberg_lifecycle.py +++ b/test/s3tables/lifecycle/iceberg_lifecycle.py @@ -99,7 +99,8 @@ def connect(args): "uri": args.catalog_url, "warehouse": f"s3://{args.bucket}/", "prefix": args.bucket, - "auth": {"type": "noop"}, + "oauth2-server-uri": f"{args.catalog_url}/v1/oauth/tokens", + "credential": f"{args.access_key}:{args.secret_key}", "s3.endpoint": args.s3_endpoint, "s3.access-key-id": args.access_key, "s3.secret-access-key": args.secret_key, diff --git a/test/s3tables/lifecycle/lance_lifecycle.py b/test/s3tables/lifecycle/lance_lifecycle.py index 60d62604c7..ccae5f2287 100644 --- a/test/s3tables/lifecycle/lance_lifecycle.py +++ b/test/s3tables/lifecycle/lance_lifecycle.py @@ -83,7 +83,13 @@ def tally(dataset): def resolve(args): """Ask the namespace where the table lives, the way the worker does.""" - namespace = ln.connect("rest", {"uri": args.namespace_url}) + namespace = ln.connect( + "rest", + { + "uri": args.namespace_url, + "header.x-api-key": f"{args.access_key}:{args.secret_key}", + }, + ) table_id = [args.bucket, args.namespace, args.table] described = namespace.describe_table(ln.DescribeTableRequest(id=table_id)) return namespace, table_id, described.location, storage_options(args, described) @@ -115,7 +121,13 @@ def main(): args = parser.parse_args() if args.phase == "write": - namespace = ln.connect("rest", {"uri": args.namespace_url}) + namespace = ln.connect( + "rest", + { + "uri": args.namespace_url, + "header.x-api-key": f"{args.access_key}:{args.secret_key}", + }, + ) for parent in ([args.bucket], [args.bucket, args.namespace]): namespace.create_namespace(ln.CreateNamespaceRequest(id=parent, mode="EXIST_OK")) table_id = [args.bucket, args.namespace, args.table] diff --git a/weed/s3api/s3tables/handler_identity_test.go b/weed/s3api/s3tables/handler_identity_test.go index b4ab8d9a04..905c62947c 100644 --- a/weed/s3api/s3tables/handler_identity_test.go +++ b/weed/s3api/s3tables/handler_identity_test.go @@ -140,7 +140,7 @@ func TestGetAccountIDAdminActionKeepsAdminAccount(t *testing.T) { assert.Equal(t, s3_constants.AccountAdminId, h.getAccountID(req), "an admin identity keeps the admin account as principal") } -func TestDefaultAllowForOnlyAppliesToUnauthenticatedOrAnonymous(t *testing.T) { +func TestDefaultAllowForOnlyAppliesWithoutResolvedIdentity(t *testing.T) { h := NewS3TablesHandler() h.SetDefaultAllow(true) @@ -149,8 +149,8 @@ func TestDefaultAllowForOnlyAppliesToUnauthenticatedOrAnonymous(t *testing.T) { anon := httptest.NewRequest(http.MethodGet, "/", nil) anon = anon.WithContext(s3_constants.SetIdentityInContext(anon.Context(), - &testIdentity{Name: s3_constants.AccountAnonymousId})) - assert.True(t, h.defaultAllowFor(anon), "anonymous requests keep the open default") + &testIdentity{Name: s3_constants.AccountAnonymousId, Actions: []string{"Read:pub", "List:pub"}})) + assert.False(t, h.defaultAllowFor(anon), "a resolved anonymous identity is scoped by its grant, not the open default") authed := httptest.NewRequest(http.MethodGet, "/", nil) authed = authed.WithContext(s3_constants.SetIdentityInContext(authed.Context(), diff --git a/weed/s3api/s3tables/iam.go b/weed/s3api/s3tables/iam.go index 944d87ca88..0b66f98eb9 100644 --- a/weed/s3api/s3tables/iam.go +++ b/weed/s3api/s3tables/iam.go @@ -49,7 +49,8 @@ func (h *S3TablesHandler) shouldUseIAM(r *http.Request, identityActions, identit } // defaultAllowFor reports whether the open-by-default fallback applies: only for -// trusted tooling or unauthenticated/anonymous access. An authenticated principal +// trusted tooling or when no principal was resolved at all. A resolved identity +// — the configured anonymous identity included — is scoped by its grant and // must pass an explicit check. func (h *S3TablesHandler) defaultAllowFor(r *http.Request) bool { if h.trusted { @@ -60,10 +61,7 @@ func (h *S3TablesHandler) defaultAllowFor(r *http.Request) bool { } // The Manager path forwards only the identity name, so a name alone (no // identity object) still counts as an authenticated principal. - if s3_constants.GetIdentityFromContext(r) == nil && s3_constants.GetIdentityNameFromContext(r) == "" { - return true - } - return isAnonymousIdentity(r) + return s3_constants.GetIdentityFromContext(r) == nil && s3_constants.GetIdentityNameFromContext(r) == "" } func isAnonymousIdentity(r *http.Request) bool {