s3tables: scope the anonymous identity's grant on the table catalogs (#12056)

* s3tables: scope the resolved anonymous identity instead of defaulting to allow

Catalog authorization fell back to the open default whenever the request
resolved to the configured anonymous identity, ignoring its grant. The
fallback now applies only when no principal was resolved (zero-config),
so anonymous requests are checked against the identity's configured
actions like any other principal. The S3 Tables API and the Iceberg and
Lance catalogs share this handler, so all three are covered.

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test: authenticate the catalog clients in the s3tables suites

Anonymous calls into the Lance and Iceberg catalogs are now scoped by
the anonymous identity's grant, so the suites that used to rely on the
anonymous pass authenticate like their real deployments do: x-api-key
for the Lance probes and REST drivers, client-credentials OAuth for
pyiceberg.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* worker(lance): authenticate namespace calls with the storage credential

Anonymous catalog calls are now scoped by the anonymous grant, so a
worker that holds an S3 credential sends it as x-api-key on the
namespace requests instead of relying on the anonymous pass.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test: authenticate the pushdown catalog connection in catalog_lancedb

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* lance: follow catalog redirects only within the same host

reqwest forwards custom headers on a follow, so a redirect to another
host would carry the x-api-key credential to a server it was never
meant for. Keep same-host hops for proxies that normalize the path or
upgrade the scheme.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* lance: warn when the catalog api key travels in cleartext

The x-api-key pair is a real storage credential: over plain http to a
non-loopback host the network can read it. Warn at client build so the
operator sees it; https and loopback stay quiet, and the key still
sends since an http catalog is a valid deployment.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* lance: require the same origin on catalog redirects

A same-host hop can still downgrade https to http and put the api key on
the wire; matching scheme, host, and port keeps the credential on the
listener it was issued for.

* lance: bound catalog redirect hops so loops fail fast

The custom redirect policy drops reqwest's default 10-hop limit, so a
catalog that redirects back to itself would hold each call open until
the request timeout. Keep the bound alongside the origin check.

* lance: gofmt the redirect helpers

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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

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

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

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      <access-key>:<secret-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      …

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)

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

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,

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]