From fc17e75fb2e0a3b836e29c34e357d818cb1d3588 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Sun, 14 Jun 2026 16:35:54 +0800 Subject: [PATCH] test(table-catalog): verify vended credential data-plane scope (#3429) --- scripts/table-catalog/README.md | 47 +++- scripts/table-catalog/pyiceberg_smoke.py | 243 ++++++++++++++++-- scripts/table-catalog/test_pyiceberg_smoke.py | 207 +++++++++++++++ 3 files changed, 471 insertions(+), 26 deletions(-) diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 590d19397..2fe42ae69 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -69,6 +69,38 @@ python3 scripts/table-catalog/pyiceberg_smoke.py \ --rest-signing-name s3 ``` +To verify catalog-vended table credentials, enable server-side credential +vending and use the vended credential profile: + +```text +RUSTFS_TABLE_CATALOG_CREDENTIAL_VENDING=enabled +``` + +```bash +python3 scripts/table-catalog/pyiceberg_smoke.py \ + --profile rustfs-vended-credentials \ + --endpoint http://127.0.0.1:9000 \ + --access-key rustfsadmin \ + --secret-key rustfsadmin \ + --bucket rustfs-s3table-smoke \ + --replace \ + --cleanup +``` + +This profile uses the configured principal to create the bucket, enable the +table bucket, and create the table. After the table exists, it calls the REST +credentials endpoint and reloads the PyIceberg catalog with the returned +table-scoped S3 access key, secret key, and session token before append, reload, +and scan operations. + +Before the PyIceberg append, the profile also checks that the returned +credential prefix exactly matches the created table warehouse location, then +runs a direct S3 data-plane scope probe with the returned temporary credentials: + +- `PutObject`, `HeadObject`, and `DeleteObject` must work inside the returned + table warehouse prefix. +- `PutObject` to the same bucket outside that prefix must be rejected. + ## Machine-Readable Inventories The script can print the current conformance inventories without importing @@ -89,7 +121,7 @@ added. | Client | Current status | Claim | |---|---|---| -| PyIceberg | Automated smoke target | create namespace, create table, append, reload, scan | +| PyIceberg | Automated smoke target | create namespace, create table, append, reload, scan, optional catalog-vended table credentials with exact-prefix data-plane scope probe | | Spark Iceberg REST catalog | Manual-ready | create/load/append/reload should be verified against a running RustFS endpoint | | Trino Iceberg REST catalog | Documented, not automated | no write compatibility claim yet | | DuckDB Iceberg | Documented, not automated | read-path reference only | @@ -102,6 +134,7 @@ added. |---|---|---|---|---| | `rustfs` | `{endpoint}/iceberg` | `s3` | static S3 credentials | automated smoke target | | `rustfs-compat` | `{endpoint}/_iceberg` | `s3tables` by default | static S3 credentials | compatibility smoke target | +| `rustfs-vended-credentials` | `{endpoint}/iceberg` | `s3` | catalog-vended table credentials after table creation | automated credential smoke target when server vending is enabled | | `aws-s3tables` | `https://s3tables.{region}.amazonaws.com/iceberg` | `s3tables` | AWS IAM/session credentials | reference only | | `minio-aistor` | `{endpoint}/_iceberg` | `s3tables` | policy-scoped S3 credentials | reference only | | `cloudflare-r2-data-catalog` | catalog URI returned by R2 | `s3` | catalog-vended credentials | reference only | @@ -112,7 +145,7 @@ added. Unsupported behavior is documented instead of hidden behind internal errors. The current unsupported inventory is: -- credential vending: table scope preview and credentials endpoint exist; temporary credentials are available only when explicitly enabled and are not yet covered by automated client profiles +- credential vending: automated after table bootstrap with exact-prefix validation and a data-plane scope probe; full no-long-term-data-credential bootstrap is not claimed - background maintenance worker: unsupported - manifest/data reachability cleanup: unsupported - snapshot expiration and compaction: unsupported @@ -133,8 +166,14 @@ The endpoint returns an empty `storage-credentials` list unless table catalog credential vending is explicitly enabled. When enabled, RustFS issues temporary table-scoped S3 credentials through the credentials endpoint. Those credentials are constrained to the table warehouse prefix and include a session token and -expiration. The automated smoke profiles still use configured S3 credentials for -object data access until a catalog-vended credential profile is added. +expiration. + +The `rustfs-vended-credentials` profile verifies the client handoff from the +catalog principal to the table-scoped temporary credentials. It still uses the +configured principal for setup and REST request signing; the vended credentials +are first checked against the created table warehouse location, then checked +with a direct S3 scope probe, and finally applied to PyIceberg S3 data-plane +access after the table has been created. Enablement is server-side and fail-closed: diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index 0aae26a6f..e382802ba 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -17,6 +17,13 @@ from dataclasses import dataclass from typing import Any DEFAULT_PROFILE = "rustfs" +CATALOG_VENDED_PROFILE = "rustfs-vended-credentials" +VENDED_CREDENTIAL_PROFILES = {CATALOG_VENDED_PROFILE} +REQUIRED_STORAGE_CREDENTIAL_KEYS = ( + "s3.access-key-id", + "s3.secret-access-key", + "s3.session-token", +) PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "rustfs": { @@ -35,6 +42,14 @@ PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "credential_mode": "static-s3-credentials", "status": "compatibility-smoke-target", }, + CATALOG_VENDED_PROFILE: { + "catalog_uri": "{endpoint}/iceberg", + "rest_path": "/iceberg", + "rest_signing_name": "s3", + "warehouse": "{bucket}", + "credential_mode": "catalog-vended-temporary-credentials", + "status": "automated-credential-smoke-target", + }, "aws-s3tables": { "catalog_uri": "https://s3tables.{region}.amazonaws.com/iceberg", "rest_path": "/iceberg", @@ -73,7 +88,7 @@ CLIENT_MATRIX: list[dict[str, str]] = [ { "client": "PyIceberg", "status": "automated", - "coverage": "create namespace, create table, append, reload, scan", + "coverage": "create namespace, create table, append, reload, scan, optional catalog-vended table credentials with exact-prefix data-plane scope probe", "entrypoint": "scripts/table-catalog/pyiceberg_smoke.py", }, { @@ -111,10 +126,10 @@ CLIENT_MATRIX: list[dict[str, str]] = [ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [ { "capability": "credential-vending", - "status": "manual-enable-temporary-credentials-no-client-profile", + "status": "automated-after-table-bootstrap", "roadmap_area": "credential-vending", "catalog_endpoint": "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/credentials", - "expected_behavior": "load table advertises the table credential scope; the credentials endpoint returns an empty storage-credentials list unless server-side temporary credential vending is explicitly enabled", + "expected_behavior": "the rustfs-vended-credentials profile requires server-side temporary credential vending, verifies the returned prefix exactly matches the table warehouse location, verifies scoped data-plane access, and switches append, reload, and scan operations to the returned table-scoped credentials after table creation", }, { "capability": "background-maintenance-worker", @@ -161,6 +176,12 @@ class RuntimeDeps: load_catalog: Any +@dataclass(frozen=True) +class StorageCredential: + prefix: str + config: dict[str, str] + + def env_or_none(key: str) -> str | None: value = os.getenv(key) if value is None or value == "": @@ -196,6 +217,8 @@ def apply_profile_defaults(args: argparse.Namespace) -> argparse.Namespace: args.rest_path = normalized_rest_path(args.rest_path) if args.rest_signing_name is None: args.rest_signing_name = defaults["rest_signing_name"] + if args.require_vended_credentials is None: + args.require_vended_credentials = args.profile in VENDED_CREDENTIAL_PROFILES return args @@ -213,6 +236,12 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--catalog-name", default=os.getenv("RUSTFS_TABLE_CATALOG_NAME", "rustfs")) parser.add_argument("--rest-path", default=env_or_none("RUSTFS_TABLE_REST_PATH")) parser.add_argument("--rest-signing-name", default=env_or_none("RUSTFS_TABLE_REST_SIGNING_NAME")) + parser.add_argument( + "--require-vended-credentials", + action="store_true", + default=None, + help="Require table-scoped credentials from the REST credentials endpoint before data-plane operations.", + ) parser.add_argument("--timeout", type=float, default=float(os.getenv("RUSTFS_TABLE_SMOKE_TIMEOUT", "20"))) parser.add_argument("--cleanup", action="store_true", help="Drop the smoke table and namespace before exiting.") parser.add_argument("--replace", action="store_true", help="Drop an existing table with the same identifier first.") @@ -330,15 +359,7 @@ def signed_rest_request(args: argparse.Namespace, deps: RuntimeDeps, method: str def ensure_bucket(args: argparse.Namespace, deps: RuntimeDeps) -> None: - client = deps.boto3.client( - "s3", - endpoint_url=normalized_endpoint(args.endpoint), - aws_access_key_id=args.access_key, - aws_secret_access_key=args.secret_key, - region_name=args.region, - verify=not args.insecure, - config=deps.botocore_config(signature_version="s3v4", s3={"addressing_style": "path"}), - ) + client = configured_s3_client(args, deps) try: client.head_bucket(Bucket=args.bucket) return @@ -354,12 +375,170 @@ def ensure_bucket(args: argparse.Namespace, deps: RuntimeDeps) -> None: client.create_bucket(Bucket=args.bucket, CreateBucketConfiguration={"LocationConstraint": args.region}) +def configured_s3_client(args: argparse.Namespace, deps: RuntimeDeps) -> Any: + return deps.boto3.client( + "s3", + endpoint_url=normalized_endpoint(args.endpoint), + aws_access_key_id=args.access_key, + aws_secret_access_key=args.secret_key, + region_name=args.region, + verify=not args.insecure, + config=deps.botocore_config(signature_version="s3v4", s3={"addressing_style": "path"}), + ) + + def enable_table_bucket(args: argparse.Namespace, deps: RuntimeDeps) -> None: encoded_bucket = urllib.parse.quote(args.bucket, safe="") signed_rest_request(args, deps, "PUT", f"{args.rest_path}/v1/buckets/{encoded_bucket}") -def catalog_properties(args: argparse.Namespace) -> dict[str, str]: +def credentials_endpoint_path(args: argparse.Namespace) -> str: + encoded_bucket = urllib.parse.quote(args.bucket, safe="") + encoded_namespace = urllib.parse.quote(args.namespace, safe="") + encoded_table = urllib.parse.quote(args.table, safe="") + return f"{args.rest_path}/v1/{encoded_bucket}/namespaces/{encoded_namespace}/tables/{encoded_table}/credentials" + + +def storage_credential_from_response(response: dict[str, Any]) -> StorageCredential: + credentials = response.get("storage-credentials") + if not isinstance(credentials, list) or not credentials: + raise RuntimeError("no storage credentials were returned by the table credentials endpoint") + + for credential in credentials: + if not isinstance(credential, dict): + continue + config = credential.get("config") + if not isinstance(config, dict): + continue + missing_keys = [key for key in REQUIRED_STORAGE_CREDENTIAL_KEYS if not config.get(key)] + if missing_keys: + continue + prefix = credential.get("prefix") + if not isinstance(prefix, str): + prefix = "" + return StorageCredential(prefix=prefix, config={str(key): str(value) for key, value in config.items()}) + + required = ", ".join(REQUIRED_STORAGE_CREDENTIAL_KEYS) + raise RuntimeError(f"no storage credentials contained the required keys: {required}") + + +def load_table_storage_credential(args: argparse.Namespace, deps: RuntimeDeps) -> StorageCredential: + response = signed_rest_request(args, deps, "GET", credentials_endpoint_path(args)) + return storage_credential_from_response(response) + + +def s3_scope_from_uri(s3_uri: str, description: str) -> tuple[str, str]: + parsed = urllib.parse.urlparse(s3_uri) + if parsed.scheme != "s3" or not parsed.netloc: + raise RuntimeError(f"{description} must be an s3 URI") + object_prefix = parsed.path.lstrip("/") + if not object_prefix: + raise RuntimeError(f"{description} must include an object prefix") + if not object_prefix.endswith("/"): + object_prefix = f"{object_prefix}/" + return parsed.netloc, object_prefix + + +def s3_scope_from_credential(storage_credential: StorageCredential) -> tuple[str, str]: + return s3_scope_from_uri(storage_credential.prefix, "storage credential prefix") + + +def string_value(value: Any) -> str | None: + if callable(value): + value = value() + if isinstance(value, str) and value: + return value + return None + + +def table_warehouse_location(table: Any) -> str: + for source in (getattr(table, "metadata", None), table): + if callable(source): + source = source() + if source is None: + continue + if isinstance(source, dict): + location = string_value(source.get("location")) + else: + location = string_value(getattr(source, "location", None)) + if location is not None: + return location + raise RuntimeError("created table did not expose a warehouse location") + + +def safe_probe_segment(namespace: str, table: str) -> str: + source = f"{namespace}-{table}" + return "".join(char if char.isalnum() or char in {"-", "_", "."} else "-" for char in source) + + +def scope_probe_keys(object_prefix: str, namespace: str, table: str) -> tuple[str, str]: + segment = safe_probe_segment(namespace, table) + suffix = f"{segment}-{int(time.time())}.txt" + return ( + f"{object_prefix}.rustfs-vended-scope-probe/{suffix}", + f"__rustfs-vended-scope-denied/{suffix}", + ) + + +def vended_s3_client(args: argparse.Namespace, deps: RuntimeDeps, storage_credential: StorageCredential) -> Any: + return deps.boto3.client( + "s3", + endpoint_url=normalized_endpoint(args.endpoint), + aws_access_key_id=storage_credential.config["s3.access-key-id"], + aws_secret_access_key=storage_credential.config["s3.secret-access-key"], + aws_session_token=storage_credential.config["s3.session-token"], + region_name=args.region, + verify=not args.insecure, + config=deps.botocore_config(signature_version="s3v4", s3={"addressing_style": "path"}), + ) + + +def is_access_denied_error(error: BaseException) -> bool: + response = getattr(error, "response", {}) + status_code = response.get("ResponseMetadata", {}).get("HTTPStatusCode") + error_code = response.get("Error", {}).get("Code") + return status_code == 403 or error_code in {"AccessDenied", "InvalidAccessKeyId", "SignatureDoesNotMatch"} + + +def verify_vended_credential_data_plane_scope( + args: argparse.Namespace, + deps: RuntimeDeps, + storage_credential: StorageCredential, + table_location: str, +) -> None: + bucket, object_prefix = s3_scope_from_credential(storage_credential) + table_bucket, table_object_prefix = s3_scope_from_uri(table_location, "table warehouse location") + if bucket != table_bucket or object_prefix != table_object_prefix: + raise RuntimeError("storage credential prefix does not match table warehouse location") + if table_bucket != args.bucket: + raise RuntimeError(f"table warehouse bucket {table_bucket} does not match expected table bucket {args.bucket}") + + inside_key, denied_key = scope_probe_keys(object_prefix, args.namespace, args.table) + client = vended_s3_client(args, deps, storage_credential) + put_inside = False + try: + client.put_object(Bucket=bucket, Key=inside_key, Body=b"rustfs table credential scope probe\n") + put_inside = True + client.head_object(Bucket=bucket, Key=inside_key) + finally: + if put_inside: + client.delete_object(Bucket=bucket, Key=inside_key) + + try: + client.put_object(Bucket=bucket, Key=denied_key, Body=b"rustfs denied table credential scope probe\n") + except deps.botocore_client_error as error: + if is_access_denied_error(error): + return + raise RuntimeError(f"scope-denied data-plane probe failed with unexpected S3 error: {error}") from error + + try: + configured_s3_client(args, deps).delete_object(Bucket=bucket, Key=denied_key) + except Exception as cleanup_error: + print(f"warning: failed to clean unexpected denied-scope probe object {denied_key}: {cleanup_error}", file=sys.stderr) + raise RuntimeError("vended table credentials unexpectedly wrote outside the table warehouse prefix") + + +def catalog_properties(args: argparse.Namespace, storage_credential: StorageCredential | None = None) -> dict[str, str]: endpoint = normalized_endpoint(args.endpoint) properties = { "type": "rest", @@ -376,6 +555,8 @@ def catalog_properties(args: argparse.Namespace) -> dict[str, str]: "rest.signing-region": args.region, "rest.signing-name": args.rest_signing_name, } + if storage_credential is not None: + properties.update(storage_credential.config) if args.insecure: properties["s3.verify-ssl"] = "false" return properties @@ -484,24 +665,24 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: ensure_local_proxy_bypass(endpoint) ensure_aws_env(args.access_key, args.secret_key, args.region) - print(f"[1/7] ensuring S3 bucket {args.bucket}") + print(f"[1/9] ensuring S3 bucket {args.bucket}") ensure_bucket(args, deps) - print(f"[2/7] enabling RustFS table bucket {args.bucket} through {args.rest_path}/v1") + print(f"[2/9] enabling RustFS table bucket {args.bucket} through {args.rest_path}/v1") enable_table_bucket(args, deps) - print(f"[3/7] loading PyIceberg REST catalog at {endpoint}{args.rest_path}") + print(f"[3/9] loading PyIceberg REST catalog at {endpoint}{args.rest_path}") catalog = deps.load_catalog(args.catalog_name, **catalog_properties(args)) install_rustfs_rest_sigv4_adapter(catalog, args, deps) identifier = table_identifier(args) if args.replace: - print(f"[4/7] replacing existing table {'.'.join(identifier)} if present") + print(f"[4/9] replacing existing table {'.'.join(identifier)} if present") drop_table_if_present(catalog, identifier) else: - print(f"[4/7] table {'.'.join(identifier)} is available") + print(f"[4/9] table {'.'.join(identifier)} is available") - print(f"[5/7] creating namespace and table {'.'.join(identifier)}") + print(f"[5/9] creating namespace and table {'.'.join(identifier)}") ensure_namespace(catalog, args.namespace) arrow_schema = deps.pyarrow.schema( [ @@ -509,9 +690,27 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: deps.pyarrow.field("payload", deps.pyarrow.string(), nullable=False), ] ) - table = catalog.create_table(identifier, schema=arrow_schema) + created_table = catalog.create_table(identifier, schema=arrow_schema) - print(f"[6/7] appending rows through PyIceberg") + storage_credential = None + if args.require_vended_credentials: + print(f"[6/9] loading table-scoped storage credentials") + storage_credential = load_table_storage_credential(args, deps) + print(f"credential scope: {storage_credential.prefix or 'not reported'}") + print(f"[7/9] verifying table-scoped data-plane access") + try: + expected_table_location = table_warehouse_location(created_table) + except RuntimeError: + expected_table_location = table_warehouse_location(catalog.load_table(identifier)) + verify_vended_credential_data_plane_scope(args, deps, storage_credential, expected_table_location) + catalog = deps.load_catalog(args.catalog_name, **catalog_properties(args, storage_credential=storage_credential)) + install_rustfs_rest_sigv4_adapter(catalog, args, deps) + else: + print(f"[6/9] using configured S3 credentials for data-plane operations") + print(f"[7/9] skipping vended credential data-plane scope probe") + + print(f"[8/9] appending rows through PyIceberg") + table = catalog.load_table(identifier) rows = deps.pyarrow.Table.from_pylist( [ {"id": 1, "payload": "alpha"}, @@ -521,7 +720,7 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: ) table.append(rows) - print(f"[7/7] reloading and scanning table") + print(f"[9/9] reloading and scanning table") loaded = catalog.load_table(identifier) scanned = loaded.scan().to_arrow() if scanned.num_rows != 2: diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index f5344b484..2aa11aa26 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -58,13 +58,220 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertIn("rustfs", profiles) self.assertIn("rustfs-compat", profiles) + self.assertIn("rustfs-vended-credentials", profiles) self.assertIn("aws-s3tables", profiles) self.assertIn("minio-aistor", profiles) self.assertIn("cloudflare-r2-data-catalog", profiles) self.assertIn("oss-tables", profiles) + self.assertEqual( + profiles["rustfs-vended-credentials"]["credential_mode"], + "catalog-vended-temporary-credentials", + ) self.assertEqual(profiles["minio-aistor"]["rest_signing_name"], "s3tables") self.assertEqual(profiles["cloudflare-r2-data-catalog"]["credential_mode"], "catalog-vended") + def test_vended_profile_requires_catalog_credentials_and_keeps_canonical_path(self) -> None: + args = self.parse_with_args([ + "--profile", + "rustfs-vended-credentials", + "--endpoint", + "http://rustfs.local:9000", + "--bucket", + "lake", + ]) + + self.assertTrue(args.require_vended_credentials) + self.assertEqual(args.rest_path, "/iceberg") + self.assertEqual(args.rest_signing_name, "s3") + self.assertEqual(pyiceberg_smoke.catalog_properties(args)["uri"], "http://rustfs.local:9000/iceberg") + + def test_credentials_endpoint_path_uses_encoded_table_identifier(self) -> None: + args = self.parse_with_args([ + "--bucket", + "lake bucket", + "--namespace", + "sales.analytics", + "--table", + "orders table", + ]) + + self.assertEqual( + pyiceberg_smoke.credentials_endpoint_path(args), + "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/tables/orders%20table/credentials", + ) + + def test_catalog_properties_can_use_vended_storage_credentials(self) -> None: + args = self.parse_with_args([ + "--access-key", + "root-access", + "--secret-key", + "root-secret", + ]) + credential = pyiceberg_smoke.storage_credential_from_response( + { + "storage-credentials": [ + { + "prefix": "s3://lake/tables/table-id/", + "config": { + "s3.access-key-id": "temp-access", + "s3.secret-access-key": "temp-secret", + "s3.session-token": "temp-token", + "rustfs.credential-mode": "catalog-vended-temporary-credentials", + }, + } + ] + } + ) + + properties = pyiceberg_smoke.catalog_properties(args, storage_credential=credential) + + self.assertEqual(properties["s3.access-key-id"], "temp-access") + self.assertEqual(properties["s3.secret-access-key"], "temp-secret") + self.assertEqual(properties["s3.session-token"], "temp-token") + self.assertEqual(properties["rustfs.credential-mode"], "catalog-vended-temporary-credentials") + + def test_empty_vended_credentials_response_is_rejected(self) -> None: + with self.assertRaisesRegex(RuntimeError, "no storage credentials"): + pyiceberg_smoke.storage_credential_from_response({"storage-credentials": []}) + + def test_storage_credential_prefix_parses_bucket_and_key_prefix(self) -> None: + bucket, key_prefix = pyiceberg_smoke.s3_scope_from_credential( + pyiceberg_smoke.StorageCredential( + prefix="s3://lake/tables/table-id/", + config={ + "s3.access-key-id": "temp-access", + "s3.secret-access-key": "temp-secret", + "s3.session-token": "temp-token", + }, + ) + ) + + self.assertEqual(bucket, "lake") + self.assertEqual(key_prefix, "tables/table-id/") + + def test_storage_credential_prefix_rejects_bucket_scope(self) -> None: + credential = pyiceberg_smoke.StorageCredential( + prefix="s3://lake", + config={ + "s3.access-key-id": "temp-access", + "s3.secret-access-key": "temp-secret", + "s3.session-token": "temp-token", + }, + ) + + with self.assertRaisesRegex(RuntimeError, "object prefix"): + pyiceberg_smoke.s3_scope_from_credential(credential) + + def test_table_warehouse_location_is_read_from_table_metadata(self) -> None: + class FakeMetadata: + location = "s3://lake/tables/table-id" + + class FakeTable: + metadata = FakeMetadata() + + self.assertEqual(pyiceberg_smoke.table_warehouse_location(FakeTable()), "s3://lake/tables/table-id") + + def test_scope_probe_keys_are_inside_and_outside_the_vended_prefix(self) -> None: + inside_key, denied_key = pyiceberg_smoke.scope_probe_keys("tables/table-id/", "namespace", "table") + + self.assertTrue(inside_key.startswith("tables/table-id/")) + self.assertFalse(denied_key.startswith("tables/table-id/")) + self.assertIn("namespace-table", inside_key) + self.assertIn("namespace-table", denied_key) + + def test_vended_s3_client_uses_session_token(self) -> None: + args = self.parse_with_args(["--endpoint", "http://rustfs.local:9000"]) + credential = pyiceberg_smoke.StorageCredential( + prefix="s3://lake/tables/table-id/", + config={ + "s3.access-key-id": "temp-access", + "s3.secret-access-key": "temp-secret", + "s3.session-token": "temp-token", + }, + ) + boto3 = mock.Mock() + deps = mock.Mock(boto3=boto3, botocore_config=mock.Mock()) + + pyiceberg_smoke.vended_s3_client(args, deps, credential) + + boto3.client.assert_called_once() + call_kwargs = boto3.client.call_args.kwargs + self.assertEqual(call_kwargs["aws_access_key_id"], "temp-access") + self.assertEqual(call_kwargs["aws_secret_access_key"], "temp-secret") + self.assertEqual(call_kwargs["aws_session_token"], "temp-token") + + def test_data_plane_scope_probe_requires_inside_access_and_outside_denial(self) -> None: + class FakeClientError(Exception): + def __init__(self, status_code: int, code: str) -> None: + self.response = { + "ResponseMetadata": {"HTTPStatusCode": status_code}, + "Error": {"Code": code}, + } + + class FakeS3Client: + def __init__(self) -> None: + self.calls: list[tuple[str, str]] = [] + + def put_object(self, *, Bucket: str, Key: str, Body: bytes) -> None: + self.calls.append(("put", Key)) + if Key == "outside/probe": + raise FakeClientError(403, "AccessDenied") + + def head_object(self, *, Bucket: str, Key: str) -> None: + self.calls.append(("head", Key)) + + def delete_object(self, *, Bucket: str, Key: str) -> None: + self.calls.append(("delete", Key)) + + args = self.parse_with_args(["--bucket", "lake"]) + credential = pyiceberg_smoke.StorageCredential( + prefix="s3://lake/tables/table-id/", + config={ + "s3.access-key-id": "temp-access", + "s3.secret-access-key": "temp-secret", + "s3.session-token": "temp-token", + }, + ) + fake_client = FakeS3Client() + deps = mock.Mock(botocore_client_error=FakeClientError) + + with mock.patch.object(pyiceberg_smoke, "vended_s3_client", return_value=fake_client): + with mock.patch.object(pyiceberg_smoke, "scope_probe_keys", return_value=("tables/table-id/probe", "outside/probe")): + pyiceberg_smoke.verify_vended_credential_data_plane_scope(args, deps, credential, "s3://lake/tables/table-id") + + self.assertEqual( + fake_client.calls, + [ + ("put", "tables/table-id/probe"), + ("head", "tables/table-id/probe"), + ("delete", "tables/table-id/probe"), + ("put", "outside/probe"), + ], + ) + + def test_data_plane_scope_probe_rejects_parent_prefix_before_s3_calls(self) -> None: + args = self.parse_with_args(["--bucket", "lake"]) + credential = pyiceberg_smoke.StorageCredential( + prefix="s3://lake/tables/", + config={ + "s3.access-key-id": "temp-access", + "s3.secret-access-key": "temp-secret", + "s3.session-token": "temp-token", + }, + ) + deps = mock.Mock() + + with mock.patch.object(pyiceberg_smoke, "vended_s3_client") as client_factory: + with self.assertRaisesRegex(RuntimeError, "does not match table warehouse location"): + pyiceberg_smoke.verify_vended_credential_data_plane_scope( + args, + deps, + credential, + "s3://lake/tables/table-id/", + ) + + client_factory.assert_not_called() + def test_unsupported_inventory_names_stable_boundaries(self) -> None: inventory = pyiceberg_smoke.unsupported_inventory() capabilities = {entry["capability"] for entry in inventory}