test(table-catalog): verify vended credential data-plane scope (#3429)

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:
 

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:

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}