From 1651541d388db21f32414a93ea4241e56afe166e Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Mon, 7 Sep 2026 05:28:06 +0800 Subject: [PATCH] fix(s3-tables): preserve encoded paths in client signatures (#7309) --- scripts/table-catalog/duckdb_smoke.py | 3 +- scripts/table-catalog/pyiceberg_smoke.py | 59 +++++--- .../table-catalog/test_pyiceberg_signing.py | 142 ++++++++++++++++++ scripts/table-catalog/test_pyiceberg_smoke.py | 2 +- 4 files changed, 184 insertions(+), 22 deletions(-) create mode 100644 scripts/table-catalog/test_pyiceberg_signing.py diff --git a/scripts/table-catalog/duckdb_smoke.py b/scripts/table-catalog/duckdb_smoke.py index b4089fd0f..236cc26fe 100755 --- a/scripts/table-catalog/duckdb_smoke.py +++ b/scripts/table-catalog/duckdb_smoke.py @@ -438,8 +438,7 @@ def run_smoke(args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps) -> Du stage_table = table_name(args.table, "stage") v3_table = table_name(args.table, "v3") smoke_tables = [seed_table, write_table, purge_table, drop_table, stage_table, v3_table] - catalog = deps.load_catalog(iceberg_args.catalog_name, **pyiceberg_smoke.catalog_properties(iceberg_args)) - pyiceberg_smoke.install_rustfs_rest_sigv4_adapter(catalog, iceberg_args, deps) + catalog = pyiceberg_smoke.load_rest_catalog(iceberg_args, deps) namespace_preexisting = bool(catalog.namespace_exists(args.namespace)) prepare_smoke_tables(catalog, args.namespace, smoke_tables, args.replace) pyiceberg_smoke.ensure_namespace(catalog, args.namespace) diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index 8c055c9ff..02e766c82 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -41,6 +41,8 @@ TABLE_MAINTENANCE_CONFIG_VERSION = 1 IDENTIFIER_SEGMENT_MAX_LEN = 64 MAX_PAGINATION_PROBE_PAGES = 16 +RUSTFS_PROFILES = {"rustfs", "rustfs-compat", CATALOG_VENDED_PROFILE} + PROFILE_DEFAULTS: dict[str, dict[str, Any]] = { "rustfs": { "catalog_uri": "{endpoint}/iceberg", @@ -264,6 +266,7 @@ class RuntimeDeps: botocore_config: Any botocore_credentials: Any botocore_auth: Any + botocore_s3_auth: Any botocore_awsrequest: Any pyarrow: Any load_catalog: Any @@ -469,13 +472,13 @@ def load_runtime_deps() -> RuntimeDeps: pyarrow = None missing.append("pyarrow") try: - from botocore.auth import SigV4Auth + from botocore.auth import S3SigV4Auth, SigV4Auth from botocore.awsrequest import AWSRequest from botocore.exceptions import ClientError from botocore.config import Config from botocore.credentials import Credentials except ModuleNotFoundError: - ClientError = Config = Credentials = SigV4Auth = AWSRequest = None + ClientError = Config = Credentials = S3SigV4Auth = SigV4Auth = AWSRequest = None missing.append("botocore") try: from pyiceberg.catalog import load_catalog @@ -496,6 +499,7 @@ def load_runtime_deps() -> RuntimeDeps: botocore_config=Config, botocore_credentials=Credentials, botocore_auth=SigV4Auth, + botocore_s3_auth=S3SigV4Auth, botocore_awsrequest=AWSRequest, pyarrow=pyarrow, load_catalog=load_catalog, @@ -529,6 +533,12 @@ def unsigned_ssl_context(insecure: bool) -> ssl.SSLContext | None: return ssl._create_unverified_context() +def sign_rest_request(args: argparse.Namespace, deps: RuntimeDeps, request: Any) -> None: + credentials = deps.botocore_credentials(args.access_key, args.secret_key) + signer = deps.botocore_s3_auth if args.profile in RUSTFS_PROFILES else deps.botocore_auth + signer(credentials, args.rest_signing_name, args.region).add_auth(request) + + def signed_rest_request(args: argparse.Namespace, deps: RuntimeDeps, method: str, path: str, body: dict[str, Any] | None = None) -> dict[str, Any]: endpoint = normalized_endpoint(args.endpoint) url = f"{endpoint}{path}" @@ -541,8 +551,7 @@ def signed_rest_request(args: argparse.Namespace, deps: RuntimeDeps, method: str headers["content-type"] = "application/json" aws_request = deps.botocore_awsrequest(method=method, url=url, data=payload, headers=headers) - credentials = deps.botocore_credentials(args.access_key, args.secret_key) - deps.botocore_auth(credentials, args.rest_signing_name, args.region).add_auth(aws_request) + sign_rest_request(args, deps, aws_request) prepared = aws_request.prepare() request = urllib.request.Request( @@ -1053,9 +1062,7 @@ def write_live_evidence(args: argparse.Namespace, result: SmokeResult) -> None: output_path.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8") -def install_rustfs_rest_sigv4_adapter(catalog: Any, args: argparse.Namespace, deps: RuntimeDeps) -> None: - from urllib import parse - +def install_rustfs_rest_sigv4_adapter(catalog: Any, args: argparse.Namespace, deps: RuntimeDeps, session: Any = None) -> None: from requests.adapters import HTTPAdapter class RustfsSigV4Adapter(HTTPAdapter): @@ -1068,21 +1075,37 @@ def install_rustfs_rest_sigv4_adapter(catalog: Any, args: argparse.Namespace, de if "connection" in request.headers: del request.headers["connection"] - url = str(request.url).split("?")[0] - query = str(parse.urlsplit(request.url).query) - params = dict(parse.parse_qsl(query)) - credentials = deps.botocore_credentials(args.access_key, args.secret_key) aws_request = deps.botocore_awsrequest( method=request.method, - url=url, - params=params, + url=request.url, data=body, headers=dict(request.headers), ) - deps.botocore_auth(credentials, args.rest_signing_name, args.region).add_auth(aws_request) + sign_rest_request(args, deps, aws_request) request.headers.update(aws_request.headers) - catalog._session.mount(catalog.uri, RustfsSigV4Adapter()) + (session if session is not None else catalog._session).mount(catalog.uri, RustfsSigV4Adapter()) + + +def load_rest_catalog(args: argparse.Namespace, deps: RuntimeDeps, storage_credential: StorageCredential | None = None) -> Any: + properties = catalog_properties(args, storage_credential) + if args.profile not in RUSTFS_PROFILES: + catalog = deps.load_catalog(args.catalog_name, **properties) + install_rustfs_rest_sigv4_adapter(catalog, args, deps) + return catalog + + from pyiceberg.catalog import _ENV_CONFIG + from pyiceberg.catalog.rest import RestCatalog + from pyiceberg.utils.config import merge_config + + properties = merge_config(_ENV_CONFIG.get_catalog_config(args.catalog_name) or {}, properties) + + class RustfsRestCatalog(RestCatalog): + def _init_sigv4(self, session: Any) -> None: + # RestCatalog fetches /config before constructing its persistent session. + install_rustfs_rest_sigv4_adapter(self, args, deps, session=session) + + return RustfsRestCatalog(args.catalog_name, **properties) def table_identifier(args: argparse.Namespace) -> tuple[str, str]: @@ -1399,8 +1422,7 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> SmokeResult: enable_table_bucket(args, deps) print(f"[3/10] 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) + catalog = load_rest_catalog(args, deps) identifier = table_identifier(args) if args.replace: @@ -1430,8 +1452,7 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> SmokeResult: 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) + catalog = load_rest_catalog(args, deps, storage_credential=storage_credential) else: print(f"[6/10] using configured S3 credentials for data-plane operations") print(f"[7/10] skipping vended credential data-plane scope probe") diff --git a/scripts/table-catalog/test_pyiceberg_signing.py b/scripts/table-catalog/test_pyiceberg_signing.py new file mode 100644 index 000000000..2ccb4f56b --- /dev/null +++ b/scripts/table-catalog/test_pyiceberg_signing.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Signing regressions requiring the PyIceberg smoke runtime dependencies.""" + +from __future__ import annotations + +import copy +import hashlib +import json +import os +import sys +import unittest +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest import mock + +from botocore.auth import S3SigV4Auth, SigV4Auth +from botocore.awsrequest import AWSRequest +from botocore.credentials import Credentials +from requests import Request, Response, Session +from requests.adapters import HTTPAdapter + +import pyiceberg_smoke + + +RUSTFS_PROFILES = {"rustfs", "rustfs-compat", "rustfs-vended-credentials"} + + +class PyIcebergSigningTest(unittest.TestCase): + def setUp(self) -> None: + self.deps = SimpleNamespace( + botocore_auth=SigV4Auth, + botocore_s3_auth=S3SigV4Auth, + botocore_credentials=Credentials, + botocore_awsrequest=AWSRequest, + ) + clock = mock.patch("botocore.auth.get_current_datetime", return_value=datetime(2026, 1, 1, tzinfo=timezone.utc)) + clock.start() + self.addCleanup(clock.stop) + + def args(self, profile: str = "rustfs") -> object: + with mock.patch.dict(os.environ, {}, clear=True), mock.patch.object( + sys, "argv", ["pyiceberg_smoke.py", "--profile", profile, "--endpoint", "http://127.0.0.1:29500", "--bucket", "warehouse"] + ): + return pyiceberg_smoke.parse_args() + + def expected_signature(self, request: object, args: object, signer: object = S3SigV4Auth) -> str: + headers = {key: value for key, value in request.headers.items() if key.lower() != "authorization"} + expected = AWSRequest(method=request.method, url=request.url, data=request.body, headers=headers) + signer(Credentials(args.access_key, args.secret_key), args.rest_signing_name, args.region).add_auth(expected) + return expected.headers["Authorization"] + + def test_adapter_signs_the_encoded_wire_path_and_full_query(self) -> None: + for profile in sorted(RUSTFS_PROFILES): + args = self.args(profile) + session = Session() + self.addCleanup(session.close) + catalog = SimpleNamespace(uri=f"{args.endpoint}{args.rest_path}", _session=session) + pyiceberg_smoke.install_rustfs_rest_sigv4_adapter(catalog, args, self.deps) + for namespace in ["sales", "sales%1Ftax", "literal%251F", "with%20space", "with%2Bplus", "a%2Fb"]: + with self.subTest(profile=profile, namespace=namespace): + url = f"{catalog.uri}/v1/warehouse/namespaces/{namespace}?key=b&key=a&empty=&plus=%2B" + request = Request("POST", url, data=b'{"properties":{"owner":"analytics"}}').prepare() + session.get_adapter(url).add_headers(request) + self.assertEqual(request.url, url) + self.assertEqual(request.headers["x-amz-content-sha256"], hashlib.sha256(request.body).hexdigest()) + self.assertEqual(request.headers["Authorization"], self.expected_signature(request, args)) + + def test_signature_changes_when_method_path_or_body_changes(self) -> None: + args = self.args() + session = Session() + self.addCleanup(session.close) + catalog = SimpleNamespace(uri=f"{args.endpoint}{args.rest_path}", _session=session) + pyiceberg_smoke.install_rustfs_rest_sigv4_adapter(catalog, args, self.deps) + request = Request("POST", f"{catalog.uri}/v1/warehouse/namespaces/sales%1Ftax", data=b"{}").prepare() + session.get_adapter(request.url).add_headers(request) + for attribute, value in [("method", "DELETE"), ("url", request.url.replace("%1F", "%251F")), ("body", b'{"changed":true}')]: + with self.subTest(attribute=attribute): + changed = copy.copy(request) + changed.headers = request.headers.copy() + setattr(changed, attribute, value) + self.assertNotEqual(request.headers["Authorization"], self.expected_signature(changed, args)) + + def test_direct_rest_requests_use_the_same_path_contract(self) -> None: + args = self.args() + response = mock.MagicMock() + response.__enter__.return_value.read.return_value = b"{}" + path = f"{args.rest_path}/v1/warehouse/namespaces/sales%1Ftax?key=b&key=a&empty=" + with mock.patch.object(pyiceberg_smoke.urllib.request, "urlopen", return_value=response) as send: + pyiceberg_smoke.signed_rest_request(args, self.deps, "GET", path) + wire = send.call_args.args[0] + request = Request(wire.method, wire.full_url, headers=dict(wire.header_items())).prepare() + self.assertEqual(request.headers["Authorization"], self.expected_signature(request, args)) + + def test_vendor_profiles_keep_generic_sigv4_normalization(self) -> None: + for profile in sorted(set(pyiceberg_smoke.PROFILE_DEFAULTS) - RUSTFS_PROFILES): + with self.subTest(profile=profile): + args = SimpleNamespace(profile=profile, access_key="test-access", secret_key="test-secret", rest_signing_name="s3tables", region="us-east-1") + request = AWSRequest(method="GET", url="https://catalog.example/namespaces/sales%1Ftax") + pyiceberg_smoke.sign_rest_request(args, self.deps, request) + expected = AWSRequest(method="GET", url=request.url) + SigV4Auth(Credentials(args.access_key, args.secret_key), args.rest_signing_name, args.region).add_auth(expected) + self.assertEqual(request.headers["Authorization"], expected.headers["Authorization"]) + + def test_initial_config_and_recreated_sessions_are_signed(self) -> None: + for profile in sorted(RUSTFS_PROFILES): + args = self.args(profile) + seen = [] + + def send(adapter: HTTPAdapter, request: object, **kwargs: object) -> Response: + adapter.add_headers(request, **kwargs) + self.assertEqual(request.headers["x-amz-content-sha256"], hashlib.sha256(b"").hexdigest()) + self.assertEqual(request.headers["Authorization"], self.expected_signature(request, args)) + seen.append(request.url) + response = Response() + response.status_code = 200 + response._content = json.dumps( + {"defaults": {}, "overrides": {}} if "/v1/config" in request.url else {"namespace": ["sales", "tax"], "properties": {}} + ).encode() + return response + + credential = pyiceberg_smoke.StorageCredential( + prefix="s3://warehouse/tables/test/", + config={"s3.access-key-id": "temporary-access", "s3.secret-access-key": "temporary-secret", "s3.session-token": "temporary-token"}, + ) + named_config = {"uri": "https://configured.example", "ssl": {"cabundle": "catalog-ca.pem", "client": {"cert": "client.pem", "key": "client-key.pem"}}} + with self.subTest(profile=profile), mock.patch.object(HTTPAdapter, "send", autospec=True, side_effect=send), mock.patch( + "pyiceberg.catalog._ENV_CONFIG.get_catalog_config", return_value=named_config + ): + for storage_credential in [None, credential]: + catalog = pyiceberg_smoke.load_rest_catalog(args, self.deps, storage_credential) + self.addCleanup(catalog._session.close) + self.assertEqual(catalog.uri, f"{args.endpoint}{args.rest_path}") + self.assertEqual(catalog._session.verify, "catalog-ca.pem") + self.assertEqual(catalog._session.cert, ("client.pem", "client-key.pem")) + self.assertEqual(catalog.load_namespace_properties(("sales", "tax")), {}) + self.assertEqual(len(seen), 4) + self.assertIn("/v1/config", seen[0]) + self.assertIn("/namespaces/sales%1Ftax", seen[1]) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index 6fd1e5b58..9c321f180 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -669,7 +669,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): with mock.patch.object(pyiceberg_smoke, "ensure_aws_env"): with mock.patch.object(pyiceberg_smoke, "ensure_bucket"): with mock.patch.object(pyiceberg_smoke, "enable_table_bucket"): - with mock.patch.object(pyiceberg_smoke, "install_rustfs_rest_sigv4_adapter"): + with mock.patch.object(pyiceberg_smoke, "load_rest_catalog", side_effect=lambda *_args, **_kwargs: FakeCatalog()): with mock.patch.object(pyiceberg_smoke, "ensure_namespace"): with mock.patch.object( pyiceberg_smoke,