From 95e67ba6c0ae66ca81ccc002d902e4b05d2e08d7 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Wed, 4 Mar 2026 18:33:42 +0800 Subject: [PATCH 01/10] Fix SigV4 auth to use base64-encoded content SHA256 and custom canonical request --- pyiceberg/catalog/rest/__init__.py | 32 ++++++++++--- tests/catalog/test_rest.py | 72 ++++++++++++++++++++++++++++-- 2 files changed, 94 insertions(+), 10 deletions(-) diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py index 97a71b429b..1a07663003 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -768,6 +768,8 @@ def _split_identifier_for_json(self, identifier: str | Identifier) -> dict[str, return {"namespace": identifier_tuple[:-1], "name": identifier_tuple[-1]} def _init_sigv4(self, session: Session) -> None: + import base64 + import hashlib from urllib import parse import boto3 @@ -776,6 +778,12 @@ def _init_sigv4(self, session: Session) -> None: from requests import PreparedRequest from requests.adapters import HTTPAdapter + class _IcebergSigV4Auth(SigV4Auth): + def canonical_request(self, request: Any) -> str: + cr = super().canonical_request(request) + # Replace the last line (body_checksum) with hex-encoded payload hash. + return cr.rsplit("\n", 1)[0] + "\n" + self.payload(request) + class SigV4Adapter(HTTPAdapter): def __init__(self, **properties: str): self._properties = properties @@ -802,17 +810,27 @@ def add_headers(self, request: PreparedRequest, **kwargs: Any) -> None: # pylin # remove the connection header as it will be updated after signing if "connection" in request.headers: del request.headers["connection"] - # For empty bodies, explicitly set the content hash header to the SHA256 of an empty string - if not request.body: - request.headers["x-amz-content-sha256"] = EMPTY_BODY_SHA256 + + # Compute the x-amz-content-sha256 header to match Iceberg Java SDK: + # - empty body → hex (EMPTY_BODY_SHA256) + # - non-empty body → base64 + if request.body: + body_bytes = request.body.encode("utf-8") if isinstance(request.body, str) else request.body + content_sha256_header = base64.b64encode(hashlib.sha256(body_bytes).digest()).decode() + else: + content_sha256_header = EMPTY_BODY_SHA256 + + signing_headers = dict(request.headers) + signing_headers["x-amz-content-sha256"] = content_sha256_header aws_request = AWSRequest( - method=request.method, url=url, params=params, data=request.body, headers=dict(request.headers) + method=request.method, url=url, params=params, data=request.body, headers=signing_headers ) - SigV4Auth(credentials, service, region).add_auth(aws_request) - original_header = request.headers - signed_headers = aws_request.headers + _IcebergSigV4Auth(credentials, service, region).add_auth(aws_request) + + original_header = dict(request.headers) + signed_headers = dict(aws_request.headers) relocated_headers = {} # relocate headers if there is a conflict with signed headers diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py index 43790f778a..a89a641260 100644 --- a/tests/catalog/test_rest.py +++ b/tests/catalog/test_rest.py @@ -624,9 +624,10 @@ def test_sigv4_sign_request_without_body(rest_mock: Mocker) -> None: assert isinstance(adapter, HTTPAdapter) adapter.add_headers(prepared) - assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256") + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256 + assert "SignedHeaders=" in prepared.headers["Authorization"] def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: @@ -655,9 +656,74 @@ def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: assert isinstance(adapter, HTTPAdapter) adapter.add_headers(prepared) - assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256") + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert "SignedHeaders=" in prepared.headers["Authorization"] + # Conflicting Authorization header is relocated assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" - assert prepared.headers.get("x-amz-content-sha256") != EMPTY_BODY_SHA256 + assert prepared.headers["x-amz-content-sha256"] == "nhKdVGKGU3IMGjYlod9xKUVc7/H5K6zTWj60yJOM80k=" + + +def test_sigv4_content_sha256_with_bytes_body(rest_mock: Mocker) -> None: + existing_token = "existing_token" + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": existing_token, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + body_content = b'{"namespace": "test_namespace"}' + prepared = catalog._session.prepare_request( + Request( + "POST", + f"{TEST_URI}v1/namespaces", + data=body_content, + ) + ) + adapter = catalog._session.adapters[catalog.uri] + assert isinstance(adapter, HTTPAdapter) + adapter.add_headers(prepared) + + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert "SignedHeaders=" in prepared.headers["Authorization"] + assert prepared.headers["x-amz-content-sha256"] == "sD20bEQP+WnwKPT7jxn7PIACGciAeWjQPlzFCK5Fifo=" + + +def test_sigv4_conflicting_sigv4_headers(rest_mock: Mocker) -> None: + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + prepared = catalog._session.prepare_request(Request("GET", f"{TEST_URI}v1/config")) + adapter = catalog._session.adapters[catalog.uri] + assert isinstance(adapter, HTTPAdapter) + + # Inject conflicting SigV4 headers before signing + prepared.headers["x-amz-content-sha256"] = "fake" + prepared.headers["X-Amz-Date"] = "fake" + + adapter.add_headers(prepared) + + # Matching Java SDK: conflicting headers are relocated with "Original-" prefix + assert prepared.headers.get("Original-x-amz-content-sha256") == "fake" + assert prepared.headers.get("Original-X-Amz-Date") == "fake" + # SigV4 headers are set correctly after signing + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256 + assert "X-Amz-Date" in prepared.headers def test_sigv4_adapter_default_retry_config(rest_mock: Mocker) -> None: From 6dd594a8930551ff9acb1b4f3843bdf93d541277 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 21 Mar 2026 16:43:35 +0800 Subject: [PATCH 02/10] Refactor _IcebergSigV4Auth to reuse canonical_request logic instead of rsplit --- pyiceberg/catalog/rest/__init__.py | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py index 1a07663003..be5a999671 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -780,9 +780,20 @@ def _init_sigv4(self, session: Session) -> None: class _IcebergSigV4Auth(SigV4Auth): def canonical_request(self, request: Any) -> str: - cr = super().canonical_request(request) - # Replace the last line (body_checksum) with hex-encoded payload hash. - return cr.rsplit("\n", 1)[0] + "\n" + self.payload(request) + # Reuses the logic from botocore's SigV4Auth.canonical_request + # (https://github.com/boto/botocore/blob/develop/botocore/auth.py) + # but always uses self.payload(request) for the body checksum. + cr = [request.method.upper()] + path = self._normalize_url_path(parse.urlsplit(request.url).path) + cr.append(path) + cr.append(self.canonical_query_string(request)) + headers_to_sign = self.headers_to_sign(request) + cr.append(self.canonical_headers(headers_to_sign) + "\n") + cr.append(self.signed_headers(headers_to_sign)) + # Always use hex-encoded payload hash per SigV4 spec, + # regardless of the x-amz-content-sha256 header value (which may be base64). + cr.append(self.payload(request)) + return "\n".join(cr) class SigV4Adapter(HTTPAdapter): def __init__(self, **properties: str): From 0bd644bfd84efd0fb3bcfcb6d405012cca6d2c72 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 21 Mar 2026 17:19:09 +0800 Subject: [PATCH 03/10] update test --- tests/catalog/test_rest.py | 32 ++++++++++++++++++++++++++------ 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py index a89a641260..31b7413bad 100644 --- a/tests/catalog/test_rest.py +++ b/tests/catalog/test_rest.py @@ -18,6 +18,7 @@ from __future__ import annotations import base64 +import hashlib import os from collections.abc import Callable from typing import Any, cast @@ -624,10 +625,16 @@ def test_sigv4_sign_request_without_body(rest_mock: Mocker) -> None: assert isinstance(adapter, HTTPAdapter) adapter.add_headers(prepared) - assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + auth_header = prepared.headers["Authorization"] + assert auth_header.startswith("AWS4-HMAC-SHA256 Credential=") assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256 - assert "SignedHeaders=" in prepared.headers["Authorization"] + # Verify the signature format: Credential, SignedHeaders, Signature + assert "Credential=" in auth_header + assert "SignedHeaders=" in auth_header + assert "Signature=" in auth_header + # x-amz-content-sha256 should be in signed headers + assert "x-amz-content-sha256" in auth_header def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: @@ -656,11 +663,19 @@ def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: assert isinstance(adapter, HTTPAdapter) adapter.add_headers(prepared) - assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") - assert "SignedHeaders=" in prepared.headers["Authorization"] + auth_header = prepared.headers["Authorization"] + assert auth_header.startswith("AWS4-HMAC-SHA256 Credential=") + assert "SignedHeaders=" in auth_header # Conflicting Authorization header is relocated assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" - assert prepared.headers["x-amz-content-sha256"] == "nhKdVGKGU3IMGjYlod9xKUVc7/H5K6zTWj60yJOM80k=" + # Non-empty body should have base64-encoded SHA256 + content_sha256 = prepared.headers["x-amz-content-sha256"] + assert content_sha256 == "nhKdVGKGU3IMGjYlod9xKUVc7/H5K6zTWj60yJOM80k=" + # Verify it's valid base64 and matches the body + decoded = base64.b64decode(content_sha256) + assert len(decoded) == 32 # SHA256 produces 32 bytes + # x-amz-content-sha256 should be in signed headers + assert "x-amz-content-sha256" in auth_header def test_sigv4_content_sha256_with_bytes_body(rest_mock: Mocker) -> None: @@ -692,7 +707,12 @@ def test_sigv4_content_sha256_with_bytes_body(rest_mock: Mocker) -> None: assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") assert "SignedHeaders=" in prepared.headers["Authorization"] - assert prepared.headers["x-amz-content-sha256"] == "sD20bEQP+WnwKPT7jxn7PIACGciAeWjQPlzFCK5Fifo=" + content_sha256 = prepared.headers["x-amz-content-sha256"] + assert content_sha256 == "sD20bEQP+WnwKPT7jxn7PIACGciAeWjQPlzFCK5Fifo=" + # Verify it's valid base64 and matches the body + decoded = base64.b64decode(content_sha256) + assert len(decoded) == 32 # SHA256 produces 32 bytes + assert decoded == hashlib.sha256(body_content).digest() def test_sigv4_conflicting_sigv4_headers(rest_mock: Mocker) -> None: From 77c3e5f4efcc903d39b66f19e6ca603910567d42 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Wed, 8 Apr 2026 10:54:23 +0800 Subject: [PATCH 04/10] Improve SigV4 tests and add botocore version reference --- pyiceberg/catalog/rest/__init__.py | 2 + tests/catalog/test_rest.py | 65 +++++++++++++++++++++++++----- 2 files changed, 58 insertions(+), 9 deletions(-) diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py index be5a999671..7170d82e8d 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -783,6 +783,8 @@ def canonical_request(self, request: Any) -> str: # Reuses the logic from botocore's SigV4Auth.canonical_request # (https://github.com/boto/botocore/blob/develop/botocore/auth.py) # but always uses self.payload(request) for the body checksum. + # Validated against botocore <= 1.42.x + # (https://github.com/boto/botocore/blob/1.42.85/botocore/auth.py#L622-L637) cr = [request.method.upper()] path = self._normalize_url_path(parse.urlsplit(request.url).path) cr.append(path) diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py index 31b7413bad..cd1038e3fe 100644 --- a/tests/catalog/test_rest.py +++ b/tests/catalog/test_rest.py @@ -670,10 +670,9 @@ def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" # Non-empty body should have base64-encoded SHA256 content_sha256 = prepared.headers["x-amz-content-sha256"] - assert content_sha256 == "nhKdVGKGU3IMGjYlod9xKUVc7/H5K6zTWj60yJOM80k=" - # Verify it's valid base64 and matches the body - decoded = base64.b64decode(content_sha256) - assert len(decoded) == 32 # SHA256 produces 32 bytes + body_bytes = prepared.body.encode("utf-8") if isinstance(prepared.body, str) else prepared.body + expected_sha256 = base64.b64encode(hashlib.sha256(body_bytes).digest()).decode() + assert content_sha256 == expected_sha256 # x-amz-content-sha256 should be in signed headers assert "x-amz-content-sha256" in auth_header @@ -708,11 +707,8 @@ def test_sigv4_content_sha256_with_bytes_body(rest_mock: Mocker) -> None: assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") assert "SignedHeaders=" in prepared.headers["Authorization"] content_sha256 = prepared.headers["x-amz-content-sha256"] - assert content_sha256 == "sD20bEQP+WnwKPT7jxn7PIACGciAeWjQPlzFCK5Fifo=" - # Verify it's valid base64 and matches the body - decoded = base64.b64decode(content_sha256) - assert len(decoded) == 32 # SHA256 produces 32 bytes - assert decoded == hashlib.sha256(body_content).digest() + expected_sha256 = base64.b64encode(hashlib.sha256(body_content).digest()).decode() + assert content_sha256 == expected_sha256 def test_sigv4_conflicting_sigv4_headers(rest_mock: Mocker) -> None: @@ -746,6 +742,57 @@ def test_sigv4_conflicting_sigv4_headers(rest_mock: Mocker) -> None: assert "X-Amz-Date" in prepared.headers +def test_sigv4_canonical_request_uses_hex_payload(rest_mock: Mocker) -> None: + """Verify that the canonical request uses hex-encoded payload hash, not the base64 header value.""" + from unittest.mock import patch + + from botocore.auth import SigV4Auth + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": "token", + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + body_content = b'{"namespace": "test"}' + prepared = catalog._session.prepare_request( + Request( + "POST", + f"{TEST_URI}v1/namespaces", + data=body_content, + ) + ) + adapter = catalog._session.adapters[catalog.uri] + assert isinstance(adapter, HTTPAdapter) + + # Capture the canonical request string during signing + captured_canonical = [] + original_add_auth = SigV4Auth.add_auth + + def capturing_add_auth(self: Any, request: Any) -> None: + captured_canonical.append(self.canonical_request(request)) + original_add_auth(self, request) + + with patch.object(SigV4Auth, "add_auth", capturing_add_auth): + adapter.add_headers(prepared) + + assert len(captured_canonical) == 1 + canonical_lines = captured_canonical[0].split("\n") + # Last line of canonical request is the payload hash + payload_hash = canonical_lines[-1] + # Must be hex-encoded (64 hex chars), not base64 + assert len(payload_hash) == 64 + assert payload_hash == hashlib.sha256(body_content).hexdigest() + # Meanwhile the header is base64-encoded + assert prepared.headers["x-amz-content-sha256"] == base64.b64encode(hashlib.sha256(body_content).digest()).decode() + + def test_sigv4_adapter_default_retry_config(rest_mock: Mocker) -> None: catalog = RestCatalog( "rest", From fd6081ec62bd28b6a34b5c597d3df619a5e21f9a Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Wed, 8 Apr 2026 15:35:35 +0800 Subject: [PATCH 05/10] Fix mypy error: assert prepared.body is not None before hashing --- tests/catalog/test_rest.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py index cd1038e3fe..54d7525a69 100644 --- a/tests/catalog/test_rest.py +++ b/tests/catalog/test_rest.py @@ -670,6 +670,7 @@ def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" # Non-empty body should have base64-encoded SHA256 content_sha256 = prepared.headers["x-amz-content-sha256"] + assert prepared.body is not None body_bytes = prepared.body.encode("utf-8") if isinstance(prepared.body, str) else prepared.body expected_sha256 = base64.b64encode(hashlib.sha256(body_bytes).digest()).decode() assert content_sha256 == expected_sha256 From a4afff00ae8d0f73c59b5171a85999c9d8e4786d Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sun, 17 May 2026 15:50:27 +0800 Subject: [PATCH 06/10] add Java cross-language reference vector, harden body type handling --- pyiceberg/catalog/rest/__init__.py | 30 ++++++++++------- pyproject.toml | 2 +- tests/catalog/test_rest.py | 53 ++++++++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 13 deletions(-) diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py index 7170d82e8d..330b4bfdd7 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -779,12 +779,11 @@ def _init_sigv4(self, session: Session) -> None: from requests.adapters import HTTPAdapter class _IcebergSigV4Auth(SigV4Auth): - def canonical_request(self, request: Any) -> str: - # Reuses the logic from botocore's SigV4Auth.canonical_request - # (https://github.com/boto/botocore/blob/develop/botocore/auth.py) - # but always uses self.payload(request) for the body checksum. - # Validated against botocore <= 1.42.x - # (https://github.com/boto/botocore/blob/1.42.85/botocore/auth.py#L622-L637) + def canonical_request(self, request: AWSRequest) -> str: + # Override forces hex payload hash in the canonical request even when + # x-amz-content-sha256 header is base64 (see body-hash block below). + # Mirrors botocore <=1.42.x SigV4Auth.canonical_request layout: + # https://github.com/boto/botocore/blob/1.42.85/botocore/auth.py#L622-L637 cr = [request.method.upper()] path = self._normalize_url_path(parse.urlsplit(request.url).path) cr.append(path) @@ -792,8 +791,6 @@ def canonical_request(self, request: Any) -> str: headers_to_sign = self.headers_to_sign(request) cr.append(self.canonical_headers(headers_to_sign) + "\n") cr.append(self.signed_headers(headers_to_sign)) - # Always use hex-encoded payload hash per SigV4 spec, - # regardless of the x-amz-content-sha256 header value (which may be base64). cr.append(self.payload(request)) return "\n".join(cr) @@ -824,11 +821,20 @@ def add_headers(self, request: PreparedRequest, **kwargs: Any) -> None: # pylin if "connection" in request.headers: del request.headers["connection"] - # Compute the x-amz-content-sha256 header to match Iceberg Java SDK: - # - empty body → hex (EMPTY_BODY_SHA256) - # - non-empty body → base64 + # Match Iceberg Java's AWS SDK v2 flexible-checksum signing: + # x-amz-content-sha256 header is base64 for non-empty bodies, hex for empty. + # The SigV4 canonical request still uses hex (enforced in _IcebergSigV4Auth above). + # Ref: https://github.com/apache/iceberg/blob/main/aws/src/main/java/org/apache/iceberg/aws/RESTSigV4AuthSession.java if request.body: - body_bytes = request.body.encode("utf-8") if isinstance(request.body, str) else request.body + if isinstance(request.body, str): + body_bytes = request.body.encode("utf-8") + elif isinstance(request.body, (bytes, bytearray)): + body_bytes = request.body + else: + raise TypeError( + f"Unsupported request body type for SigV4 signing: " + f"{type(request.body).__name__}; expected str or bytes." + ) content_sha256_header = base64.b64encode(hashlib.sha256(body_bytes).digest()).decode() else: content_sha256_header = EMPTY_BODY_SHA256 diff --git a/pyproject.toml b/pyproject.toml index 894a946667..4f87e3da80 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -91,7 +91,7 @@ sql-postgres = [ ] sql-sqlite = ["sqlalchemy>=2.0.18,<3"] gcsfs = ["gcsfs>=2023.1.0"] -rest-sigv4 = ["boto3>=1.24.59"] +rest-sigv4 = ["boto3>=1.24.59", "botocore<2"] hf = ["huggingface-hub>=0.24.0"] pyiceberg-core = ["pyiceberg-core>=0.5.1,<0.10.0"] datafusion = ["datafusion>=52,<53"] diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py index 54d7525a69..01466b10b1 100644 --- a/tests/catalog/test_rest.py +++ b/tests/catalog/test_rest.py @@ -794,6 +794,59 @@ def capturing_add_auth(self: Any, request: Any) -> None: assert prepared.headers["x-amz-content-sha256"] == base64.b64encode(hashlib.sha256(body_content).digest()).decode() +def test_sigv4_content_sha256_matches_iceberg_java_reference(rest_mock: Mocker) -> None: + """Pin byte-for-byte equivalence with Iceberg Java TestRESTSigV4AuthSession (L121, L177).""" + java_reference_body = b'{"namespace":["ns"],"properties":{}}' + java_reference_base64 = "yc5oAKPWjHY4sW8XQq0l/3aNrrXJKBycVFNnDEGMfww=" + java_reference_empty_hex = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + adapter = catalog._session.adapters[catalog.uri] + assert isinstance(adapter, HTTPAdapter) + + # Non-empty body: must match Java's base64 reference value exactly + prepared_with_body = catalog._session.prepare_request(Request("POST", f"{TEST_URI}v1/namespaces", data=java_reference_body)) + adapter.add_headers(prepared_with_body) + assert prepared_with_body.headers["x-amz-content-sha256"] == java_reference_base64 + + # Empty body: must match Java's hex reference value exactly + prepared_empty = catalog._session.prepare_request(Request("GET", f"{TEST_URI}v1/config")) + adapter.add_headers(prepared_empty) + assert prepared_empty.headers["x-amz-content-sha256"] == java_reference_empty_hex + + +def test_sigv4_unsupported_body_type_raises(rest_mock: Mocker) -> None: + """Unsupported body types (e.g. file-like) raise a clear error rather than crashing in hashlib.""" + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + adapter = catalog._session.adapters[catalog.uri] + assert isinstance(adapter, HTTPAdapter) + + prepared = catalog._session.prepare_request(Request("POST", f"{TEST_URI}v1/namespaces")) + # Inject an unsupported body type (a list — not str/bytes) + prepared.body = ["not", "a", "valid", "body"] # type: ignore[assignment] + + with pytest.raises(TypeError, match="Unsupported request body type for SigV4 signing"): + adapter.add_headers(prepared) + + def test_sigv4_adapter_default_retry_config(rest_mock: Mocker) -> None: catalog = RestCatalog( "rest", From fe8c3dfc8dd11b34f14d5fd2bf6223baf99cbf55 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Thu, 21 May 2026 14:40:35 +0800 Subject: [PATCH 07/10] Move SigV4 request signing into the AuthManager abstraction --- pyiceberg/catalog/rest/__init__.py | 175 ++++++------- pyiceberg/catalog/rest/auth.py | 134 +++++++++- tests/catalog/test_rest.py | 312 ++++-------------------- tests/catalog/test_rest_auth.py | 378 +++++++++++++++++++++++++++++ 4 files changed, 618 insertions(+), 381 deletions(-) diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py index 330b4bfdd7..0181080962 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -31,7 +31,15 @@ from pyiceberg import __version__ from pyiceberg.catalog import BOTOCORE_SESSION, TOKEN, URI, WAREHOUSE_LOCATION, Catalog, PropertiesUpdateSummary -from pyiceberg.catalog.rest.auth import AUTH_MANAGER, AuthManager, AuthManagerAdapter, AuthManagerFactory, LegacyOAuth2AuthManager +from pyiceberg.catalog.rest.auth import ( + AUTH_MANAGER, + AuthManager, + AuthManagerAdapter, + AuthManagerFactory, + LegacyOAuth2AuthManager, + NoopAuthManager, + SigV4AuthManager, +) from pyiceberg.catalog.rest.response import _handle_non_200_response from pyiceberg.catalog.rest.scan_planning import ( FetchScanTasksRequest, @@ -253,11 +261,11 @@ class ScanPlanningMode(Enum): CA_BUNDLE = "cabundle" SSL = "ssl" SIGV4 = "rest.sigv4-enabled" +SIGV4_AUTH_TYPE = "sigv4" SIGV4_REGION = "rest.signing-region" SIGV4_SERVICE = "rest.signing-name" SIGV4_MAX_RETRIES = "rest.sigv4.max-retries" SIGV4_MAX_RETRIES_DEFAULT = 10 -EMPTY_BODY_SHA256: str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" OAUTH2_SERVER_URI = "oauth2-server-uri" SNAPSHOT_LOADING_MODE = "snapshot-loading-mode" AUTH = "auth" @@ -441,10 +449,49 @@ def _create_session(self) -> Session: elif ssl_client_cert := ssl_client.get(CERT): session.cert = ssl_client_cert + self._auth_manager = self._build_auth_manager(session) + session.auth = AuthManagerAdapter(self._auth_manager) + + # SigV4 retry is decoupled from signing: mount a plain retry adapter. + if self._is_sigv4_enabled(): + from requests.adapters import HTTPAdapter + + max_retries = property_as_int(self.properties, SIGV4_MAX_RETRIES, SIGV4_MAX_RETRIES_DEFAULT) + session.mount(self.uri, HTTPAdapter(max_retries=max_retries)) + + return session + + def _is_sigv4_enabled(self) -> bool: + """Return True if SigV4 signing is requested via either config path.""" + if property_as_bool(self.properties, SIGV4, False): + return True + auth_config = self.properties.get(AUTH) + return auth_config is not None and auth_config.get("type") == SIGV4_AUTH_TYPE + + def _build_auth_manager(self, session: Session) -> AuthManager: + """Build the AuthManager, wrapping the delegate in SigV4 when enabled.""" + delegate = self._build_delegate_auth_manager(session) + if self._is_sigv4_enabled(): + return self._build_sigv4_auth_manager(delegate) + return delegate + + def _build_delegate_auth_manager(self, session: Session) -> AuthManager: + """Build the header-based AuthManager (the SigV4 delegate, or the manager used directly).""" if auth_config := self.properties.get(AUTH): auth_type = auth_config.get("type") if auth_type is None: raise ValueError("auth.type must be defined") + + if auth_type == SIGV4_AUTH_TYPE: + # The delegate is configured under auth.sigv4.delegate.* + sigv4_config = auth_config.get(SIGV4_AUTH_TYPE, {}) + delegate_config = sigv4_config.get("delegate") + if not delegate_config or "type" not in delegate_config: + # No delegate configured: SigV4-only auth, with no header-based delegate. + return NoopAuthManager() + delegate_type = delegate_config["type"] + return AuthManagerFactory.create(delegate_type, delegate_config.get(delegate_type, {})) + auth_type_config = auth_config.get(auth_type, {}) auth_impl = auth_config.get("impl") @@ -454,17 +501,28 @@ def _create_session(self) -> Session: if auth_type != CUSTOM and auth_impl: raise ValueError("auth.impl can only be specified when using custom auth.type") - self._auth_manager = AuthManagerFactory.create(auth_impl or auth_type, auth_type_config) - session.auth = AuthManagerAdapter(self._auth_manager) - else: - self._auth_manager = self._create_legacy_oauth2_auth_manager(session) - session.auth = AuthManagerAdapter(self._auth_manager) + return AuthManagerFactory.create(auth_impl or auth_type, auth_type_config) - # Configure SigV4 Request Signing - if property_as_bool(self.properties, SIGV4, False): - self._init_sigv4(session) + return self._create_legacy_oauth2_auth_manager(session) - return session + def _build_sigv4_auth_manager(self, delegate: AuthManager) -> AuthManager: + """Wrap the delegate AuthManager in a SigV4AuthManager.""" + import boto3 + + boto_session = boto3.Session( + profile_name=get_first_property_value(self.properties, AWS_PROFILE_NAME), + region_name=get_first_property_value(self.properties, AWS_REGION), + botocore_session=self.properties.get(BOTOCORE_SESSION), + aws_access_key_id=get_first_property_value(self.properties, AWS_ACCESS_KEY_ID), + aws_secret_access_key=get_first_property_value(self.properties, AWS_SECRET_ACCESS_KEY), + aws_session_token=get_first_property_value(self.properties, AWS_SESSION_TOKEN), + ) + return SigV4AuthManager( + delegate=delegate, + boto_session=boto_session, + region=self.properties.get(SIGV4_REGION), + service=self.properties.get(SIGV4_SERVICE, "execute-api"), + ) @staticmethod def _resolve_storage_credentials(storage_credentials: list[StorageCredential], location: str | None) -> Properties: @@ -767,101 +825,6 @@ def _split_identifier_for_json(self, identifier: str | Identifier) -> dict[str, identifier_tuple = self._identifier_to_validated_tuple(identifier) return {"namespace": identifier_tuple[:-1], "name": identifier_tuple[-1]} - def _init_sigv4(self, session: Session) -> None: - import base64 - import hashlib - from urllib import parse - - import boto3 - from botocore.auth import SigV4Auth - from botocore.awsrequest import AWSRequest - from requests import PreparedRequest - from requests.adapters import HTTPAdapter - - class _IcebergSigV4Auth(SigV4Auth): - def canonical_request(self, request: AWSRequest) -> str: - # Override forces hex payload hash in the canonical request even when - # x-amz-content-sha256 header is base64 (see body-hash block below). - # Mirrors botocore <=1.42.x SigV4Auth.canonical_request layout: - # https://github.com/boto/botocore/blob/1.42.85/botocore/auth.py#L622-L637 - cr = [request.method.upper()] - path = self._normalize_url_path(parse.urlsplit(request.url).path) - cr.append(path) - cr.append(self.canonical_query_string(request)) - headers_to_sign = self.headers_to_sign(request) - cr.append(self.canonical_headers(headers_to_sign) + "\n") - cr.append(self.signed_headers(headers_to_sign)) - cr.append(self.payload(request)) - return "\n".join(cr) - - class SigV4Adapter(HTTPAdapter): - def __init__(self, **properties: str): - self._properties = properties - max_retries = property_as_int(self._properties, SIGV4_MAX_RETRIES, SIGV4_MAX_RETRIES_DEFAULT) - super().__init__(max_retries=max_retries) - self._boto_session = boto3.Session( - profile_name=get_first_property_value(self._properties, AWS_PROFILE_NAME), - region_name=get_first_property_value(self._properties, AWS_REGION), - botocore_session=self._properties.get(BOTOCORE_SESSION), - aws_access_key_id=get_first_property_value(self._properties, AWS_ACCESS_KEY_ID), - aws_secret_access_key=get_first_property_value(self._properties, AWS_SECRET_ACCESS_KEY), - aws_session_token=get_first_property_value(self._properties, AWS_SESSION_TOKEN), - ) - - def add_headers(self, request: PreparedRequest, **kwargs: Any) -> None: # pylint: disable=W0613 - credentials = self._boto_session.get_credentials().get_frozen_credentials() - region = self._properties.get(SIGV4_REGION, self._boto_session.region_name) - service = self._properties.get(SIGV4_SERVICE, "execute-api") - - url = str(request.url).split("?")[0] - query = str(parse.urlsplit(request.url).query) - params = dict(parse.parse_qsl(query)) - - # remove the connection header as it will be updated after signing - if "connection" in request.headers: - del request.headers["connection"] - - # Match Iceberg Java's AWS SDK v2 flexible-checksum signing: - # x-amz-content-sha256 header is base64 for non-empty bodies, hex for empty. - # The SigV4 canonical request still uses hex (enforced in _IcebergSigV4Auth above). - # Ref: https://github.com/apache/iceberg/blob/main/aws/src/main/java/org/apache/iceberg/aws/RESTSigV4AuthSession.java - if request.body: - if isinstance(request.body, str): - body_bytes = request.body.encode("utf-8") - elif isinstance(request.body, (bytes, bytearray)): - body_bytes = request.body - else: - raise TypeError( - f"Unsupported request body type for SigV4 signing: " - f"{type(request.body).__name__}; expected str or bytes." - ) - content_sha256_header = base64.b64encode(hashlib.sha256(body_bytes).digest()).decode() - else: - content_sha256_header = EMPTY_BODY_SHA256 - - signing_headers = dict(request.headers) - signing_headers["x-amz-content-sha256"] = content_sha256_header - - aws_request = AWSRequest( - method=request.method, url=url, params=params, data=request.body, headers=signing_headers - ) - - _IcebergSigV4Auth(credentials, service, region).add_auth(aws_request) - - original_header = dict(request.headers) - signed_headers = dict(aws_request.headers) - relocated_headers = {} - - # relocate headers if there is a conflict with signed headers - for header, value in original_header.items(): - if header in signed_headers and signed_headers[header] != value: - relocated_headers[f"Original-{header}"] = value - - request.headers.update(relocated_headers) - request.headers.update(signed_headers) - - session.mount(self.uri, SigV4Adapter(**self.properties)) - def _response_to_table(self, identifier_tuple: tuple[str, ...], table_response: TableResponse) -> Table: # Per Iceberg spec: storage-credentials take precedence over config credential_config = self._resolve_storage_credentials( diff --git a/pyiceberg/catalog/rest/auth.py b/pyiceberg/catalog/rest/auth.py index 602074282c..3f42708dd7 100644 --- a/pyiceberg/catalog/rest/auth.py +++ b/pyiceberg/catalog/rest/auth.py @@ -21,7 +21,7 @@ import threading import time from abc import ABC, abstractmethod -from functools import cached_property +from functools import cache, cached_property from typing import Any import requests @@ -36,6 +36,37 @@ COLON = ":" logger = logging.getLogger(__name__) +# SHA-256 of an empty payload. Used as the x-amz-content-sha256 header value for +# empty-body requests, matching Iceberg Java's RESTSigV4AuthSession workaround. +EMPTY_BODY_SHA256 = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + + +@cache +def _iceberg_sigv4_auth_class() -> type: + """Lazily build the botocore SigV4Auth subclass (botocore is an optional dependency).""" + from urllib import parse + + from botocore.auth import SigV4Auth + from botocore.awsrequest import AWSRequest + + class _IcebergSigV4Auth(SigV4Auth): + def canonical_request(self, request: AWSRequest) -> str: + # Override forces the hex payload hash in the canonical request even when + # the x-amz-content-sha256 header is base64 (see SigV4AuthManager.sign_request). + # Mirrors botocore <=1.42.x SigV4Auth.canonical_request layout: + # https://github.com/boto/botocore/blob/1.42.85/botocore/auth.py#L622-L637 + cr = [request.method.upper()] + path = self._normalize_url_path(parse.urlsplit(request.url).path) + cr.append(path) + cr.append(self.canonical_query_string(request)) + headers_to_sign = self.headers_to_sign(request) + cr.append(self.canonical_headers(headers_to_sign) + "\n") + cr.append(self.signed_headers(headers_to_sign)) + cr.append(self.payload(request)) + return "\n".join(cr) + + return _IcebergSigV4Auth + class AuthManager(ABC): """ @@ -48,6 +79,14 @@ class AuthManager(ABC): def auth_header(self) -> str | None: """Return the Authorization header value, or None if not applicable.""" + def sign_request(self, request: PreparedRequest) -> PreparedRequest: + """Optionally sign or otherwise modify the prepared request. + + The default implementation is a no-op. Override for request-signing + schemes such as SigV4 that must inspect the full request. + """ + return request + class NoopAuthManager(AuthManager): """Auth Manager implementation with no auth.""" @@ -311,6 +350,91 @@ def auth_header(self) -> str: return f"Bearer {self._get_token()}" +class SigV4AuthManager(AuthManager): + """AuthManager that signs requests with AWS SigV4, wrapping a delegate AuthManager. + + Mirrors Iceberg Java's RESTSigV4AuthManager: the delegate AuthManager handles + header-based auth (e.g. OAuth2), then SigV4 signs the resulting request. + """ + + def __init__( + self, + delegate: AuthManager, + boto_session: Any, + region: str | None, + service: str = "execute-api", + ): + """Initialize SigV4AuthManager. + + Args: + delegate: AuthManager that supplies header-based auth before signing. + boto_session: A boto3.Session used to resolve AWS credentials. + region: SigV4 signing region; falls back to the boto session's region. + service: SigV4 signing service name. + """ + self._delegate = delegate + self._boto_session = boto_session + self._region = region + self._service = service + + def auth_header(self) -> str | None: + return self._delegate.auth_header() + + def sign_request(self, request: PreparedRequest) -> PreparedRequest: + import hashlib + from urllib import parse + + from botocore.awsrequest import AWSRequest + + credentials = self._boto_session.get_credentials().get_frozen_credentials() + region = self._region or self._boto_session.region_name + + url = str(request.url).split("?")[0] + query = str(parse.urlsplit(request.url).query) + params = dict(parse.parse_qsl(query)) + + # remove the connection header as it will be updated after signing + if "connection" in request.headers: + del request.headers["connection"] + + # Match Iceberg Java's AWS SDK v2 flexible-checksum signing: + # x-amz-content-sha256 header is base64 for non-empty bodies, hex for empty. + # The SigV4 canonical request still uses hex (enforced in _iceberg_sigv4_auth_class). + # Ref: https://github.com/apache/iceberg/blob/main/aws/src/main/java/org/apache/iceberg/aws/RESTSigV4AuthSession.java + if request.body: + if isinstance(request.body, str): + body_bytes = request.body.encode("utf-8") + elif isinstance(request.body, (bytes, bytearray)): + body_bytes = bytes(request.body) + else: + raise TypeError( + f"Unsupported request body type for SigV4 signing: {type(request.body).__name__}; expected str or bytes." + ) + content_sha256_header = base64.b64encode(hashlib.sha256(body_bytes).digest()).decode() + else: + content_sha256_header = EMPTY_BODY_SHA256 + + signing_headers = dict(request.headers) + signing_headers["x-amz-content-sha256"] = content_sha256_header + + aws_request = AWSRequest(method=request.method, url=url, params=params, data=request.body, headers=signing_headers) + + _iceberg_sigv4_auth_class()(credentials, self._service, region).add_auth(aws_request) + + original_header = dict(request.headers) + signed_headers = dict(aws_request.headers) + relocated_headers = {} + + # relocate headers if there is a conflict with signed headers + for header, value in original_header.items(): + if header in signed_headers and signed_headers[header] != value: + relocated_headers[f"Original-{header}"] = value + + request.headers.update(relocated_headers) + request.headers.update(signed_headers) + return request + + class AuthManagerAdapter(AuthBase): """A `requests.auth.AuthBase` adapter for integrating an `AuthManager` into a `requests.Session`. @@ -332,17 +456,19 @@ def __init__(self, auth_manager: AuthManager): def __call__(self, request: PreparedRequest) -> PreparedRequest: """ - Modify the outgoing request to include the Authorization header. + Modify the outgoing request to include the Authorization header and any signature. Args: request (requests.PreparedRequest): The HTTP request being prepared. Returns: - requests.PreparedRequest: The modified request with Authorization header. + requests.PreparedRequest: The modified request. """ if auth_header := self.auth_manager.auth_header(): request.headers["Authorization"] = auth_header - return request + # Header first, then sign: a request-signing AuthManager (e.g. SigV4) must + # see the Authorization header so it can relocate it before signing. + return self.auth_manager.sign_request(request) class AuthManagerFactory: diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py index 01466b10b1..33bf29a547 100644 --- a/tests/catalog/test_rest.py +++ b/tests/catalog/test_rest.py @@ -18,14 +18,12 @@ from __future__ import annotations import base64 -import hashlib import os from collections.abc import Callable from typing import Any, cast from unittest import mock import pytest -from requests import Request from requests.adapters import HTTPAdapter from requests.exceptions import HTTPError from requests_mock import Mocker @@ -34,7 +32,6 @@ from pyiceberg.catalog import PropertiesUpdateSummary, load_catalog from pyiceberg.catalog.rest import ( DEFAULT_ENDPOINTS, - EMPTY_BODY_SHA256, OAUTH2_SERVER_URI, PAGE_SIZE, SIGV4_MAX_RETRIES, @@ -592,259 +589,26 @@ def test_list_tables_page_size(rest_mock: Mocker) -> None: def test_list_tables_200_sigv4(rest_mock: Mocker) -> None: namespace = "examples" + # SigV4 signing replaces the bearer Authorization header with an AWS4-HMAC-SHA256 + # signature, so the request headers are not matched against TEST_HEADERS here. rest_mock.get( f"{TEST_URI}v1/namespaces/{namespace}/tables", json={"identifiers": [{"namespace": ["examples"], "name": "fooshare"}]}, status_code=200, - request_headers=TEST_HEADERS, - ) - - assert RestCatalog("rest", **{"uri": TEST_URI, "token": TEST_TOKEN, "rest.sigv4-enabled": "true"}).list_tables(namespace) == [ - ("examples", "fooshare") - ] - assert rest_mock.called - - -def test_sigv4_sign_request_without_body(rest_mock: Mocker) -> None: - existing_token = "existing_token" - - catalog = RestCatalog( - "rest", - **{ - "uri": TEST_URI, - "token": existing_token, - "rest.sigv4-enabled": "true", - "rest.signing-region": "us-west-2", - "client.access-key-id": "id", - "client.secret-access-key": "secret", - }, - ) - - prepared = catalog._session.prepare_request(Request("GET", f"{TEST_URI}v1/config")) - adapter = catalog._session.adapters[catalog.uri] - assert isinstance(adapter, HTTPAdapter) - adapter.add_headers(prepared) - - auth_header = prepared.headers["Authorization"] - assert auth_header.startswith("AWS4-HMAC-SHA256 Credential=") - assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" - assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256 - # Verify the signature format: Credential, SignedHeaders, Signature - assert "Credential=" in auth_header - assert "SignedHeaders=" in auth_header - assert "Signature=" in auth_header - # x-amz-content-sha256 should be in signed headers - assert "x-amz-content-sha256" in auth_header - - -def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: - existing_token = "existing_token" - - catalog = RestCatalog( - "rest", - **{ - "uri": TEST_URI, - "token": existing_token, - "rest.sigv4-enabled": "true", - "rest.signing-region": "us-west-2", - "client.access-key-id": "id", - "client.secret-access-key": "secret", - }, - ) - - prepared = catalog._session.prepare_request( - Request( - "POST", - f"{TEST_URI}v1/namespaces", - data={"namespace": "asdfasd"}, - ) - ) - adapter = catalog._session.adapters[catalog.uri] - assert isinstance(adapter, HTTPAdapter) - adapter.add_headers(prepared) - - auth_header = prepared.headers["Authorization"] - assert auth_header.startswith("AWS4-HMAC-SHA256 Credential=") - assert "SignedHeaders=" in auth_header - # Conflicting Authorization header is relocated - assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" - # Non-empty body should have base64-encoded SHA256 - content_sha256 = prepared.headers["x-amz-content-sha256"] - assert prepared.body is not None - body_bytes = prepared.body.encode("utf-8") if isinstance(prepared.body, str) else prepared.body - expected_sha256 = base64.b64encode(hashlib.sha256(body_bytes).digest()).decode() - assert content_sha256 == expected_sha256 - # x-amz-content-sha256 should be in signed headers - assert "x-amz-content-sha256" in auth_header - - -def test_sigv4_content_sha256_with_bytes_body(rest_mock: Mocker) -> None: - existing_token = "existing_token" - - catalog = RestCatalog( - "rest", - **{ - "uri": TEST_URI, - "token": existing_token, - "rest.sigv4-enabled": "true", - "rest.signing-region": "us-west-2", - "client.access-key-id": "id", - "client.secret-access-key": "secret", - }, ) - body_content = b'{"namespace": "test_namespace"}' - prepared = catalog._session.prepare_request( - Request( - "POST", - f"{TEST_URI}v1/namespaces", - data=body_content, - ) - ) - adapter = catalog._session.adapters[catalog.uri] - assert isinstance(adapter, HTTPAdapter) - adapter.add_headers(prepared) - - assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") - assert "SignedHeaders=" in prepared.headers["Authorization"] - content_sha256 = prepared.headers["x-amz-content-sha256"] - expected_sha256 = base64.b64encode(hashlib.sha256(body_content).digest()).decode() - assert content_sha256 == expected_sha256 - - -def test_sigv4_conflicting_sigv4_headers(rest_mock: Mocker) -> None: - catalog = RestCatalog( - "rest", - **{ - "uri": TEST_URI, - "rest.sigv4-enabled": "true", - "rest.signing-region": "us-west-2", - "client.access-key-id": "id", - "client.secret-access-key": "secret", - }, - ) - - prepared = catalog._session.prepare_request(Request("GET", f"{TEST_URI}v1/config")) - adapter = catalog._session.adapters[catalog.uri] - assert isinstance(adapter, HTTPAdapter) - - # Inject conflicting SigV4 headers before signing - prepared.headers["x-amz-content-sha256"] = "fake" - prepared.headers["X-Amz-Date"] = "fake" - - adapter.add_headers(prepared) - - # Matching Java SDK: conflicting headers are relocated with "Original-" prefix - assert prepared.headers.get("Original-x-amz-content-sha256") == "fake" - assert prepared.headers.get("Original-X-Amz-Date") == "fake" - # SigV4 headers are set correctly after signing - assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") - assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256 - assert "X-Amz-Date" in prepared.headers - - -def test_sigv4_canonical_request_uses_hex_payload(rest_mock: Mocker) -> None: - """Verify that the canonical request uses hex-encoded payload hash, not the base64 header value.""" - from unittest.mock import patch - - from botocore.auth import SigV4Auth - - catalog = RestCatalog( + assert RestCatalog( "rest", **{ "uri": TEST_URI, - "token": "token", + "token": TEST_TOKEN, "rest.sigv4-enabled": "true", "rest.signing-region": "us-west-2", "client.access-key-id": "id", "client.secret-access-key": "secret", }, - ) - - body_content = b'{"namespace": "test"}' - prepared = catalog._session.prepare_request( - Request( - "POST", - f"{TEST_URI}v1/namespaces", - data=body_content, - ) - ) - adapter = catalog._session.adapters[catalog.uri] - assert isinstance(adapter, HTTPAdapter) - - # Capture the canonical request string during signing - captured_canonical = [] - original_add_auth = SigV4Auth.add_auth - - def capturing_add_auth(self: Any, request: Any) -> None: - captured_canonical.append(self.canonical_request(request)) - original_add_auth(self, request) - - with patch.object(SigV4Auth, "add_auth", capturing_add_auth): - adapter.add_headers(prepared) - - assert len(captured_canonical) == 1 - canonical_lines = captured_canonical[0].split("\n") - # Last line of canonical request is the payload hash - payload_hash = canonical_lines[-1] - # Must be hex-encoded (64 hex chars), not base64 - assert len(payload_hash) == 64 - assert payload_hash == hashlib.sha256(body_content).hexdigest() - # Meanwhile the header is base64-encoded - assert prepared.headers["x-amz-content-sha256"] == base64.b64encode(hashlib.sha256(body_content).digest()).decode() - - -def test_sigv4_content_sha256_matches_iceberg_java_reference(rest_mock: Mocker) -> None: - """Pin byte-for-byte equivalence with Iceberg Java TestRESTSigV4AuthSession (L121, L177).""" - java_reference_body = b'{"namespace":["ns"],"properties":{}}' - java_reference_base64 = "yc5oAKPWjHY4sW8XQq0l/3aNrrXJKBycVFNnDEGMfww=" - java_reference_empty_hex = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" - - catalog = RestCatalog( - "rest", - **{ - "uri": TEST_URI, - "rest.sigv4-enabled": "true", - "rest.signing-region": "us-east-1", - "client.access-key-id": "id", - "client.secret-access-key": "secret", - }, - ) - adapter = catalog._session.adapters[catalog.uri] - assert isinstance(adapter, HTTPAdapter) - - # Non-empty body: must match Java's base64 reference value exactly - prepared_with_body = catalog._session.prepare_request(Request("POST", f"{TEST_URI}v1/namespaces", data=java_reference_body)) - adapter.add_headers(prepared_with_body) - assert prepared_with_body.headers["x-amz-content-sha256"] == java_reference_base64 - - # Empty body: must match Java's hex reference value exactly - prepared_empty = catalog._session.prepare_request(Request("GET", f"{TEST_URI}v1/config")) - adapter.add_headers(prepared_empty) - assert prepared_empty.headers["x-amz-content-sha256"] == java_reference_empty_hex - - -def test_sigv4_unsupported_body_type_raises(rest_mock: Mocker) -> None: - """Unsupported body types (e.g. file-like) raise a clear error rather than crashing in hashlib.""" - catalog = RestCatalog( - "rest", - **{ - "uri": TEST_URI, - "rest.sigv4-enabled": "true", - "rest.signing-region": "us-east-1", - "client.access-key-id": "id", - "client.secret-access-key": "secret", - }, - ) - adapter = catalog._session.adapters[catalog.uri] - assert isinstance(adapter, HTTPAdapter) - - prepared = catalog._session.prepare_request(Request("POST", f"{TEST_URI}v1/namespaces")) - # Inject an unsupported body type (a list — not str/bytes) - prepared.body = ["not", "a", "valid", "body"] # type: ignore[assignment] - - with pytest.raises(TypeError, match="Unsupported request body type for SigV4 signing"): - adapter.add_headers(prepared) + ).list_tables(namespace) == [("examples", "fooshare")] + assert rest_mock.called def test_sigv4_adapter_default_retry_config(rest_mock: Mocker) -> None: @@ -884,29 +648,6 @@ def test_sigv4_adapter_override_retry_config(rest_mock: Mocker) -> None: assert adapter.max_retries.total == 3 -def test_sigv4_uses_client_profile_name(rest_mock: Mocker) -> None: - with mock.patch("boto3.Session") as mock_session: - RestCatalog( - "rest", - **{ - "uri": TEST_URI, - "token": TEST_TOKEN, - "rest.sigv4-enabled": "true", - "rest.signing-region": "us-west-2", - "client.profile-name": "rest-profile", - }, - ) - - mock_session.assert_called_with( - profile_name="rest-profile", - region_name=None, - botocore_session=None, - aws_access_key_id=None, - aws_secret_access_key=None, - aws_session_token=None, - ) - - def test_list_tables_404(rest_mock: Mocker) -> None: namespace = "examples" rest_mock.get( @@ -1066,16 +807,25 @@ def test_list_views_invalid_page_size(rest_mock: Mocker) -> None: def test_list_views_200_sigv4(rest_mock: Mocker) -> None: namespace = "examples" + # SigV4 signing replaces the bearer Authorization header with an AWS4-HMAC-SHA256 + # signature, so the request headers are not matched against TEST_HEADERS here. rest_mock.get( f"{TEST_URI}v1/namespaces/{namespace}/views", json={"identifiers": [{"namespace": ["examples"], "name": "fooshare"}]}, status_code=200, - request_headers=TEST_HEADERS, ) - assert RestCatalog("rest", **{"uri": TEST_URI, "token": TEST_TOKEN, "rest.sigv4-enabled": "true"}).list_views(namespace) == [ - ("examples", "fooshare") - ] + assert RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": TEST_TOKEN, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ).list_views(namespace) == [("examples", "fooshare")] assert rest_mock.called @@ -2946,7 +2696,17 @@ def test_rest_catalog_close_sigv4(self, rest_mock: Mocker) -> None: status_code=200, ) - catalog = RestCatalog("rest", **{"uri": TEST_URI, "token": TEST_TOKEN, "rest.sigv4-enabled": "true"}) + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": TEST_TOKEN, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) catalog.close() assert hasattr(catalog, "_session") assert len(catalog._session.adapters) == self.EXPECTED_ADAPTERS_SIGV4 @@ -2980,7 +2740,17 @@ def test_rest_catalog_context_manager_with_exception_sigv4(self, rest_mock: Mock ) try: - with RestCatalog("rest", **{"uri": TEST_URI, "token": TEST_TOKEN, "rest.sigv4-enabled": "true"}) as cat: + with RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": TEST_TOKEN, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) as cat: catalog = cat raise ValueError("Test exception") except ValueError: diff --git a/tests/catalog/test_rest_auth.py b/tests/catalog/test_rest_auth.py index ae5d40f5aa..29f6612ff3 100644 --- a/tests/catalog/test_rest_auth.py +++ b/tests/catalog/test_rest_auth.py @@ -16,12 +16,14 @@ # under the License. import base64 +import hashlib from unittest.mock import MagicMock, patch import pytest import requests from requests_mock import Mocker +from pyiceberg.catalog.rest import RestCatalog from pyiceberg.catalog.rest.auth import AuthManagerAdapter, BasicAuthManager, EntraAuthManager, GoogleAuthManager, NoopAuthManager TEST_URI = "https://iceberg-test-catalog/" @@ -35,6 +37,11 @@ def rest_mock(requests_mock: Mocker) -> Mocker: json={}, status_code=200, ) + requests_mock.get( + f"{TEST_URI}v1/config", + json={"defaults": {}, "overrides": {}}, + status_code=200, + ) return requests_mock @@ -249,3 +256,374 @@ def test_entra_auth_manager_token_failure(mock_default_cred: MagicMock, rest_moc # Verify no requests were made with a blank/missing auth header history = rest_mock.request_history assert len(history) == 0 + + +def test_sign_request_default_is_noop() -> None: + """AuthManager.sign_request default implementation must not modify the request.""" + manager = NoopAuthManager() + prepared = requests.Request("GET", TEST_URI).prepare() + original_headers = dict(prepared.headers) + + result = manager.sign_request(prepared) + + assert result is prepared + assert dict(result.headers) == original_headers + + +def test_sigv4_auth_manager_signs_with_java_reference_values() -> None: + """SigV4AuthManager.sign_request must match Iceberg Java reference header values.""" + import boto3 + + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + boto_session = boto3.Session( + aws_access_key_id="id", + aws_secret_access_key="secret", + region_name="us-east-1", + ) + manager = SigV4AuthManager( + delegate=NoopAuthManager(), + boto_session=boto_session, + region="us-east-1", + service="execute-api", + ) + + # Non-empty body: base64 SHA-256 (Iceberg Java TestRESTSigV4AuthSession.java L177) + body = b'{"namespace":["ns"],"properties":{}}' + prepared = requests.Request("POST", "https://example.com/v1/namespaces", data=body).prepare() + manager.sign_request(prepared) + assert prepared.headers["x-amz-content-sha256"] == base64.b64encode(hashlib.sha256(body).digest()).decode() + assert prepared.headers["x-amz-content-sha256"] == "yc5oAKPWjHY4sW8XQq0l/3aNrrXJKBycVFNnDEGMfww=" + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + + # Empty body: hex EMPTY_BODY_SHA256 (Iceberg Java TestRESTSigV4AuthSession.java L121) + prepared_empty = requests.Request("GET", "https://example.com/v1/config").prepare() + manager.sign_request(prepared_empty) + assert prepared_empty.headers["x-amz-content-sha256"] == "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + + +def test_sigv4_auth_manager_relocates_delegate_authorization() -> None: + """When the delegate sets Authorization, SigV4 relocates it to Original-Authorization.""" + import boto3 + + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager( + delegate=BasicAuthManager(username="user", password="pass"), + boto_session=boto_session, + region="us-east-1", + service="execute-api", + ) + adapter = AuthManagerAdapter(manager) + + prepared = requests.Request("GET", "https://example.com/v1/config").prepare() + adapter(prepared) + + # SigV4 owns Authorization; the delegate's Basic header is relocated. + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert prepared.headers["Original-Authorization"].startswith("Basic ") + + +def test_sigv4_legacy_config_builds_sigv4_auth_manager(rest_mock: Mocker) -> None: + """Legacy rest.sigv4-enabled config produces a SigV4AuthManager.""" + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + assert isinstance(catalog._auth_manager, SigV4AuthManager) + + +def test_sigv4_auth_type_config_builds_sigv4_auth_manager(rest_mock: Mocker) -> None: + """New auth.type=sigv4 config produces a SigV4AuthManager wrapping the delegate.""" + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + catalog = RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "auth": {"type": "sigv4", "sigv4": {"delegate": {"type": "noop"}}}, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + assert isinstance(catalog._auth_manager, SigV4AuthManager) + + +def test_sigv4_sign_request_without_body(rest_mock: Mocker) -> None: + from pyiceberg.catalog.rest.auth import EMPTY_BODY_SHA256 + + existing_token = "existing_token" + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": existing_token, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + # prepare_request applies session.auth, which signs via SigV4AuthManager. + prepared = catalog._session.prepare_request(requests.Request("GET", f"{TEST_URI}v1/config")) + + auth_header = prepared.headers["Authorization"] + assert auth_header.startswith("AWS4-HMAC-SHA256 Credential=") + assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" + assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256 + # Verify the signature format: Credential, SignedHeaders, Signature + assert "Credential=" in auth_header + assert "SignedHeaders=" in auth_header + assert "Signature=" in auth_header + # x-amz-content-sha256 should be in signed headers + assert "x-amz-content-sha256" in auth_header + + +def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: + existing_token = "existing_token" + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": existing_token, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + prepared = catalog._session.prepare_request( + requests.Request( + "POST", + f"{TEST_URI}v1/namespaces", + data={"namespace": "asdfasd"}, + ) + ) + + auth_header = prepared.headers["Authorization"] + assert auth_header.startswith("AWS4-HMAC-SHA256 Credential=") + assert "SignedHeaders=" in auth_header + # Conflicting Authorization header is relocated + assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" + # Non-empty body should have base64-encoded SHA256 + content_sha256 = prepared.headers["x-amz-content-sha256"] + assert prepared.body is not None + body_bytes = prepared.body.encode("utf-8") if isinstance(prepared.body, str) else prepared.body + expected_sha256 = base64.b64encode(hashlib.sha256(body_bytes).digest()).decode() + assert content_sha256 == expected_sha256 + # x-amz-content-sha256 should be in signed headers + assert "x-amz-content-sha256" in auth_header + + +def test_sigv4_content_sha256_with_bytes_body(rest_mock: Mocker) -> None: + existing_token = "existing_token" + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": existing_token, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + body_content = b'{"namespace": "test_namespace"}' + prepared = catalog._session.prepare_request( + requests.Request( + "POST", + f"{TEST_URI}v1/namespaces", + data=body_content, + ) + ) + + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert "SignedHeaders=" in prepared.headers["Authorization"] + content_sha256 = prepared.headers["x-amz-content-sha256"] + expected_sha256 = base64.b64encode(hashlib.sha256(body_content).digest()).decode() + assert content_sha256 == expected_sha256 + + +def test_sigv4_conflicting_sigv4_headers(rest_mock: Mocker) -> None: + from pyiceberg.catalog.rest.auth import EMPTY_BODY_SHA256 + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + # Build an unsigned prepared request, then inject conflicting SigV4 headers. + prepared = requests.Request("GET", f"{TEST_URI}v1/config").prepare() + prepared.headers["x-amz-content-sha256"] = "fake" + prepared.headers["X-Amz-Date"] = "fake" + + # session.auth is the AuthManagerAdapter; calling it signs the request. + auth = catalog._session.auth + assert isinstance(auth, AuthManagerAdapter) + auth(prepared) + + # Matching Java SDK: conflicting headers are relocated with "Original-" prefix + assert prepared.headers.get("Original-x-amz-content-sha256") == "fake" + assert prepared.headers.get("Original-X-Amz-Date") == "fake" + # SigV4 headers are set correctly after signing + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256 + assert "X-Amz-Date" in prepared.headers + + +def test_sigv4_canonical_request_uses_hex_payload(rest_mock: Mocker) -> None: + """Verify that the canonical request uses hex-encoded payload hash, not the base64 header value.""" + from typing import Any + + from botocore.auth import SigV4Auth + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": "token", + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + body_content = b'{"namespace": "test"}' + + # Capture the canonical request string during signing + captured_canonical = [] + original_add_auth = SigV4Auth.add_auth + + def capturing_add_auth(self: Any, request: Any) -> None: + captured_canonical.append(self.canonical_request(request)) + original_add_auth(self, request) + + # Signing now happens inside prepare_request (via session.auth). + with patch.object(SigV4Auth, "add_auth", capturing_add_auth): + prepared = catalog._session.prepare_request( + requests.Request( + "POST", + f"{TEST_URI}v1/namespaces", + data=body_content, + ) + ) + + assert len(captured_canonical) == 1 + canonical_lines = captured_canonical[0].split("\n") + # Last line of canonical request is the payload hash + payload_hash = canonical_lines[-1] + # Must be hex-encoded (64 hex chars), not base64 + assert len(payload_hash) == 64 + assert payload_hash == hashlib.sha256(body_content).hexdigest() + # Meanwhile the header is base64-encoded + assert prepared.headers["x-amz-content-sha256"] == base64.b64encode(hashlib.sha256(body_content).digest()).decode() + + +def test_sigv4_content_sha256_matches_iceberg_java_reference(rest_mock: Mocker) -> None: + """Pin byte-for-byte equivalence with Iceberg Java TestRESTSigV4AuthSession (L121, L177).""" + java_reference_body = b'{"namespace":["ns"],"properties":{}}' + java_reference_base64 = "yc5oAKPWjHY4sW8XQq0l/3aNrrXJKBycVFNnDEGMfww=" + java_reference_empty_hex = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + + catalog = RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + # Non-empty body: must match Java's base64 reference value exactly + prepared_with_body = catalog._session.prepare_request( + requests.Request("POST", f"{TEST_URI}v1/namespaces", data=java_reference_body) + ) + assert prepared_with_body.headers["x-amz-content-sha256"] == java_reference_base64 + + # Empty body: must match Java's hex reference value exactly + prepared_empty = catalog._session.prepare_request(requests.Request("GET", f"{TEST_URI}v1/config")) + assert prepared_empty.headers["x-amz-content-sha256"] == java_reference_empty_hex + + +def test_sigv4_unsupported_body_type_raises() -> None: + """Unsupported body types (e.g. file-like) raise a clear error rather than crashing in hashlib.""" + import boto3 + + from pyiceberg.catalog.rest.auth import NoopAuthManager, SigV4AuthManager + + boto_session = boto3.Session( + aws_access_key_id="id", + aws_secret_access_key="secret", + region_name="us-east-1", + ) + manager = SigV4AuthManager( + delegate=NoopAuthManager(), + boto_session=boto_session, + region="us-east-1", + service="execute-api", + ) + + prepared = requests.Request("POST", f"{TEST_URI}v1/namespaces").prepare() + # Inject an unsupported body type (a list — not str/bytes) + prepared.body = ["not", "a", "valid", "body"] # type: ignore[assignment] + + with pytest.raises(TypeError, match="Unsupported request body type for SigV4 signing"): + manager.sign_request(prepared) + + +def test_sigv4_uses_client_profile_name(rest_mock: Mocker) -> None: + import boto3 + + # Use a real boto3.Session for credential resolution (signing runs during + # config fetch), but spy on the constructor to assert the profile is honored. + real_session = boto3.Session( + aws_access_key_id="id", + aws_secret_access_key="secret", + region_name="us-west-2", + ) + + with patch("boto3.Session", return_value=real_session) as mock_session: + RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "token": "token", + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-west-2", + "client.profile-name": "rest-profile", + }, + ) + + mock_session.assert_called_with( + profile_name="rest-profile", + region_name=None, + botocore_session=None, + aws_access_key_id=None, + aws_secret_access_key=None, + aws_session_token=None, + ) From ccbba516742566294a194c9aa16e2ef3cdc810fb Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Fri, 12 Jun 2026 14:55:32 +0800 Subject: [PATCH 08/10] Align SigV4 auth with Java: sign relocated Authorization, validate config, add docs --- mkdocs/docs/configuration.md | 22 +++++++++ pyiceberg/catalog/rest/__init__.py | 10 +++++ pyiceberg/catalog/rest/auth.py | 3 ++ tests/catalog/test_rest.py | 18 ++++++++ tests/catalog/test_rest_auth.py | 71 ++++++++++++++++++++++++++++++ 5 files changed, 124 insertions(+) diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md index 44b5e395a9..2811e999d9 100644 --- a/mkdocs/docs/configuration.md +++ b/mkdocs/docs/configuration.md @@ -423,6 +423,8 @@ Legacy OAuth2 Properties will be removed in PyIceberg 1.0 in place of pluggable | rest.signing-region | us-east-1 | The region to use when SigV4 signing a request | | rest.signing-name | execute-api | The service signing name to use when SigV4 signing a request | +SigV4 can also be enabled as `auth.type: sigv4`, which additionally lets you choose the wrapped header-based auth (see the AuthManager section below). + ##### Pluggable Authentication via AuthManager The RESTCatalog supports pluggable authentication via the `auth` configuration block. This allows you to specify which how the access token will be fetched and managed for use with the HTTP requests to the RESTCatalog server. The authentication method is selected by setting the `auth.type` property, and additional configuration can be provided as needed for each method. @@ -435,6 +437,7 @@ The RESTCatalog supports pluggable authentication via the `auth` configuration b - `custom`: Custom authentication manager (requires `auth.impl`). - `google`: Google Authentication support - `entra`: Microsoft Entra ID (Azure AD) authentication support +- `sigv4`: AWS SigV4 request signing, optionally wrapping a delegate auth type. ###### Configuration Properties @@ -463,6 +466,7 @@ catalog: | `auth.custom` | If type is `custom` | Block containing configuration for the custom AuthManager. | | `auth.google` | If type is `google` | Block containing `credentials_path` to a service account file (if using). Will default to using Application Default Credentials. | | `auth.entra` | If type is `entra` | Block containing Entra ID configuration. Will default to using DefaultAzureCredential. | +| `auth.sigv4` | If type is `sigv4` | Block containing an optional `delegate` auth block whose `Authorization` header is preserved as `Original-Authorization` after signing. Signing region/name come from `rest.signing-region`/`rest.signing-name`; AWS credentials from `client.*` or the standard boto3 chain. | ###### Examples @@ -508,6 +512,24 @@ auth: property2: value2 ``` +SigV4 Signing (wrapping OAuth2): + +```yaml +auth: + type: sigv4 + sigv4: + delegate: + type: oauth2 + oauth2: + client_id: my-client-id + client_secret: my-client-secret + token_url: https://auth.example.com/oauth/token +rest.signing-region: us-east-1 +rest.signing-name: execute-api +client.access-key-id: my-access-key +client.secret-access-key: my-secret-key +``` + ###### Notes - If `auth.type` is `custom`, you **must** specify `auth.impl` with the full class path to your custom AuthManager. diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py index 0181080962..3d8782db1a 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -472,6 +472,12 @@ def _build_auth_manager(self, session: Session) -> AuthManager: """Build the AuthManager, wrapping the delegate in SigV4 when enabled.""" delegate = self._build_delegate_auth_manager(session) if self._is_sigv4_enabled(): + if property_as_bool(self.properties, SIGV4, False): + deprecation_message( + deprecated_in="0.11.0", + removed_in="1.0.0", + help_message=f"The property {SIGV4} is deprecated. Please use auth.type={SIGV4_AUTH_TYPE} instead", + ) return self._build_sigv4_auth_manager(delegate) return delegate @@ -483,6 +489,8 @@ def _build_delegate_auth_manager(self, session: Session) -> AuthManager: raise ValueError("auth.type must be defined") if auth_type == SIGV4_AUTH_TYPE: + if auth_config.get("impl"): + raise ValueError("auth.impl can only be specified when using custom auth.type") # The delegate is configured under auth.sigv4.delegate.* sigv4_config = auth_config.get(SIGV4_AUTH_TYPE, {}) delegate_config = sigv4_config.get("delegate") @@ -490,6 +498,8 @@ def _build_delegate_auth_manager(self, session: Session) -> AuthManager: # No delegate configured: SigV4-only auth, with no header-based delegate. return NoopAuthManager() delegate_type = delegate_config["type"] + if delegate_type == SIGV4_AUTH_TYPE: + raise ValueError("Cannot delegate a SigV4 auth manager to another SigV4 auth manager") return AuthManagerFactory.create(delegate_type, delegate_config.get(delegate_type, {})) auth_type_config = auth_config.get(auth_type, {}) diff --git a/pyiceberg/catalog/rest/auth.py b/pyiceberg/catalog/rest/auth.py index 3f42708dd7..6773692902 100644 --- a/pyiceberg/catalog/rest/auth.py +++ b/pyiceberg/catalog/rest/auth.py @@ -415,6 +415,9 @@ def sign_request(self, request: PreparedRequest) -> PreparedRequest: content_sha256_header = EMPTY_BODY_SHA256 signing_headers = dict(request.headers) + # Relocate Authorization before signing so it lands in SignedHeaders, like Java. + if "Authorization" in signing_headers: + signing_headers["Original-Authorization"] = signing_headers.pop("Authorization") signing_headers["x-amz-content-sha256"] = content_sha256_header aws_request = AWSRequest(method=request.method, url=url, params=params, data=request.body, headers=signing_headers) diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py index 33bf29a547..405446cd7e 100644 --- a/tests/catalog/test_rest.py +++ b/tests/catalog/test_rest.py @@ -587,6 +587,9 @@ def test_list_tables_page_size(rest_mock: Mocker) -> None: ] +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_list_tables_200_sigv4(rest_mock: Mocker) -> None: namespace = "examples" # SigV4 signing replaces the bearer Authorization header with an AWS4-HMAC-SHA256 @@ -611,6 +614,9 @@ def test_list_tables_200_sigv4(rest_mock: Mocker) -> None: assert rest_mock.called +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_adapter_default_retry_config(rest_mock: Mocker) -> None: catalog = RestCatalog( "rest", @@ -629,6 +635,9 @@ def test_sigv4_adapter_default_retry_config(rest_mock: Mocker) -> None: assert adapter.max_retries.total == SIGV4_MAX_RETRIES_DEFAULT +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_adapter_override_retry_config(rest_mock: Mocker) -> None: catalog = RestCatalog( "rest", @@ -805,6 +814,9 @@ def test_list_views_invalid_page_size(rest_mock: Mocker) -> None: assert str(e.value) == "rest-page-size must be a positive integer" +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_list_views_200_sigv4(rest_mock: Mocker) -> None: namespace = "examples" # SigV4 signing replaces the bearer Authorization header with an AWS4-HMAC-SHA256 @@ -2688,6 +2700,9 @@ def test_catalog_close(self, rest_mock: Mocker) -> None: # Second close should not raise any exception catalog.close() + @pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" + ) def test_rest_catalog_close_sigv4(self, rest_mock: Mocker) -> None: catalog = None rest_mock.get( @@ -2730,6 +2745,9 @@ def test_rest_catalog_context_manager_with_exception(self, rest_mock: Mocker) -> assert catalog is not None and hasattr(catalog, "_session") assert len(catalog._session.adapters) == self.EXPECTED_ADAPTERS + @pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" + ) def test_rest_catalog_context_manager_with_exception_sigv4(self, rest_mock: Mocker) -> None: """Test RestCatalog context manager properly closes with exceptions.""" catalog = None diff --git a/tests/catalog/test_rest_auth.py b/tests/catalog/test_rest_auth.py index 29f6612ff3..b6fb2a9882 100644 --- a/tests/catalog/test_rest_auth.py +++ b/tests/catalog/test_rest_auth.py @@ -323,8 +323,13 @@ def test_sigv4_auth_manager_relocates_delegate_authorization() -> None: # SigV4 owns Authorization; the delegate's Basic header is relocated. assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") assert prepared.headers["Original-Authorization"].startswith("Basic ") + # Relocated header is signed (in SignedHeaders), matching Iceberg Java. + assert "original-authorization" in prepared.headers["Authorization"] +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_legacy_config_builds_sigv4_auth_manager(rest_mock: Mocker) -> None: """Legacy rest.sigv4-enabled config produces a SigV4AuthManager.""" from pyiceberg.catalog.rest.auth import SigV4AuthManager @@ -359,6 +364,54 @@ def test_sigv4_auth_type_config_builds_sigv4_auth_manager(rest_mock: Mocker) -> assert isinstance(catalog._auth_manager, SigV4AuthManager) +def test_sigv4_auth_type_rejects_auth_impl(rest_mock: Mocker) -> None: + """auth.impl is only valid with auth.type=custom, not sigv4.""" + with pytest.raises(ValueError, match="auth.impl can only be specified when using custom auth.type"): + RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "auth": {"type": "sigv4", "impl": "my.custom.AuthManager"}, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + +def test_sigv4_rejects_sigv4_delegate(rest_mock: Mocker) -> None: + """A SigV4 delegate cannot itself be sigv4, matching Iceberg Java's AuthManagers check.""" + with pytest.raises(ValueError, match="Cannot delegate a SigV4 auth manager to another SigV4 auth manager"): + RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "auth": {"type": "sigv4", "sigv4": {"delegate": {"type": "sigv4"}}}, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + +def test_sigv4_legacy_flag_emits_deprecation_warning(rest_mock: Mocker) -> None: + """The legacy rest.sigv4-enabled flag warns and points at auth.type=sigv4, matching Iceberg Java.""" + with pytest.warns(DeprecationWarning, match="rest.sigv4-enabled is deprecated"): + RestCatalog( + "rest", + **{ + "uri": TEST_URI, + "rest.sigv4-enabled": "true", + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_sign_request_without_body(rest_mock: Mocker) -> None: from pyiceberg.catalog.rest.auth import EMPTY_BODY_SHA256 @@ -391,6 +444,9 @@ def test_sigv4_sign_request_without_body(rest_mock: Mocker) -> None: assert "x-amz-content-sha256" in auth_header +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: existing_token = "existing_token" @@ -429,6 +485,9 @@ def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: assert "x-amz-content-sha256" in auth_header +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_content_sha256_with_bytes_body(rest_mock: Mocker) -> None: existing_token = "existing_token" @@ -460,6 +519,9 @@ def test_sigv4_content_sha256_with_bytes_body(rest_mock: Mocker) -> None: assert content_sha256 == expected_sha256 +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_conflicting_sigv4_headers(rest_mock: Mocker) -> None: from pyiceberg.catalog.rest.auth import EMPTY_BODY_SHA256 @@ -493,6 +555,9 @@ def test_sigv4_conflicting_sigv4_headers(rest_mock: Mocker) -> None: assert "X-Amz-Date" in prepared.headers +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_canonical_request_uses_hex_payload(rest_mock: Mocker) -> None: """Verify that the canonical request uses hex-encoded payload hash, not the base64 header value.""" from typing import Any @@ -542,6 +607,9 @@ def capturing_add_auth(self: Any, request: Any) -> None: assert prepared.headers["x-amz-content-sha256"] == base64.b64encode(hashlib.sha256(body_content).digest()).decode() +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_content_sha256_matches_iceberg_java_reference(rest_mock: Mocker) -> None: """Pin byte-for-byte equivalence with Iceberg Java TestRESTSigV4AuthSession (L121, L177).""" java_reference_body = b'{"namespace":["ns"],"properties":{}}' @@ -596,6 +664,9 @@ def test_sigv4_unsupported_body_type_raises() -> None: manager.sign_request(prepared) +@pytest.mark.filterwarnings( + "ignore:Deprecated in 0.11.0, will be removed in 1.0.0. The property rest.sigv4-enabled is deprecated:DeprecationWarning" +) def test_sigv4_uses_client_profile_name(rest_mock: Mocker) -> None: import boto3 From f9b87f5563ed0e9e396f877b40b9622f38ea8d92 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Fri, 31 Jul 2026 03:24:39 -0400 Subject: [PATCH 09/10] chore: sync uv.lock after rebase onto main --- uv.lock | 2 ++ 1 file changed, 2 insertions(+) diff --git a/uv.lock b/uv.lock index 3351821808..b3a981c0a6 100644 --- a/uv.lock +++ b/uv.lock @@ -4965,6 +4965,7 @@ ray = [ ] rest-sigv4 = [ { name = "boto3" }, + { name = "botocore" }, ] s3fs = [ { name = "s3fs" }, @@ -5031,6 +5032,7 @@ requires-dist = [ { name = "boto3", marker = "extra == 'dynamodb'", specifier = ">=1.24.59" }, { name = "boto3", marker = "extra == 'glue'", specifier = ">=1.24.59" }, { name = "boto3", marker = "extra == 'rest-sigv4'", specifier = ">=1.24.59" }, + { name = "botocore", marker = "extra == 'rest-sigv4'", specifier = "<2" }, { name = "cachetools", specifier = ">=5.5" }, { name = "click", specifier = ">=7.1.1" }, { name = "daft", marker = "extra == 'daft'", specifier = ">=0.7.10" }, From 13982624258bbd40164c09d55f8778fa06d8ae92 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Fri, 31 Jul 2026 05:21:48 -0400 Subject: [PATCH 10/10] Harden SigV4 auth: redirect re-signing, credential hygiene, delegate fixes --- pyiceberg/catalog/rest/__init__.py | 36 +- pyiceberg/catalog/rest/auth.py | 135 +++++- tests/catalog/test_rest_auth.py | 709 +++++++++++++++++++++++++++++ 3 files changed, 868 insertions(+), 12 deletions(-) diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py index 3d8782db1a..bc9af0db59 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -38,8 +38,12 @@ AuthManagerFactory, LegacyOAuth2AuthManager, NoopAuthManager, + ReauthenticatingSession, SigV4AuthManager, ) +from pyiceberg.catalog.rest.auth import ( + EMPTY_BODY_SHA256 as EMPTY_BODY_SHA256, # public re-export (0.11.x compat) +) from pyiceberg.catalog.rest.response import _handle_non_200_response from pyiceberg.catalog.rest.scan_planning import ( FetchScanTasksRequest, @@ -434,7 +438,8 @@ def __init__(self, name: str, **properties: str): def _create_session(self) -> Session: """Create a request session with provided catalog configuration.""" - session = Session() + # Re-applies auth on redirects, but only to the catalog's own origin. + session = ReauthenticatingSession(trusted_auth_origin=self.uri) # Set HTTP headers self._config_headers(session) @@ -494,13 +499,24 @@ def _build_delegate_auth_manager(self, session: Session) -> AuthManager: # The delegate is configured under auth.sigv4.delegate.* sigv4_config = auth_config.get(SIGV4_AUTH_TYPE, {}) delegate_config = sigv4_config.get("delegate") - if not delegate_config or "type" not in delegate_config: - # No delegate configured: SigV4-only auth, with no header-based delegate. + if not delegate_config: + # Parity with the deprecated rest.sigv4-enabled flag: legacy + # token/credential properties keep their OAuth header auth. + if self.properties.get(TOKEN) or self.properties.get(CREDENTIAL): + return self._create_legacy_oauth2_auth_manager(session) return NoopAuthManager() + if "type" not in delegate_config: + raise ValueError("auth.sigv4.delegate.type must be defined") delegate_type = delegate_config["type"] if delegate_type == SIGV4_AUTH_TYPE: raise ValueError("Cannot delegate a SigV4 auth manager to another SigV4 auth manager") - return AuthManagerFactory.create(delegate_type, delegate_config.get(delegate_type, {})) + # Custom delegates resolve through their impl, like top-level auth. + delegate_impl = delegate_config.get("impl") + if delegate_type == CUSTOM and not delegate_impl: + raise ValueError("auth.sigv4.delegate.impl must be specified when using custom delegate type") + if delegate_type != CUSTOM and delegate_impl: + raise ValueError("auth.sigv4.delegate.impl can only be specified when using custom delegate type") + return AuthManagerFactory.create(delegate_impl or delegate_type, delegate_config.get(delegate_type, {})) auth_type_config = auth_config.get(auth_type, {}) auth_impl = auth_config.get("impl") @@ -532,6 +548,8 @@ def _build_sigv4_auth_manager(self, delegate: AuthManager) -> AuthManager: boto_session=boto_session, region=self.properties.get(SIGV4_REGION), service=self.properties.get(SIGV4_SERVICE, "execute-api"), + # Sign only catalog requests; an external oauth2-server-uri stays unsigned. + signing_url_prefix=self.uri, ) @staticmethod @@ -557,7 +575,12 @@ def _resolve_storage_credentials(storage_credentials: list[StorageCredential], l def _load_file_io(self, properties: Properties = EMPTY_DICT, location: str | None = None) -> FileIO: merged_properties = {**self.properties, **properties} if self._auth_manager: - merged_properties[AUTH_MANAGER] = self._auth_manager + # Propagate the delegate, not the SigV4 wrapper: storage requests are not + # catalog-signed, and the wrapper's boto3.Session is not pickleable. + auth_manager = self._auth_manager + if isinstance(auth_manager, SigV4AuthManager): + auth_manager = auth_manager.delegate + merged_properties[AUTH_MANAGER] = auth_manager return load_file_io(merged_properties, location) @override @@ -879,6 +902,9 @@ def _refresh_token(self) -> None: # instead of retrying on Auth Exceptions. Keeping refresh behavior for the LegacyOAuth2AuthManager # for backward compatibility auth_manager = self._session.auth.auth_manager # type: ignore[union-attr] + # Unwrap SigV4 so legacy OAuth refresh reaches the delegate. + if isinstance(auth_manager, SigV4AuthManager): + auth_manager = auth_manager.delegate if isinstance(auth_manager, LegacyOAuth2AuthManager): auth_manager._refresh_token() diff --git a/pyiceberg/catalog/rest/auth.py b/pyiceberg/catalog/rest/auth.py index 6773692902..b344459580 100644 --- a/pyiceberg/catalog/rest/auth.py +++ b/pyiceberg/catalog/rest/auth.py @@ -25,8 +25,9 @@ from typing import Any import requests -from requests import HTTPError, PreparedRequest, Session +from requests import HTTPError, PreparedRequest, Response, Session from requests.auth import AuthBase +from requests.structures import CaseInsensitiveDict from pyiceberg.catalog.rest.response import TokenResponse, _handle_non_200_response from pyiceberg.exceptions import OAuthError @@ -363,6 +364,7 @@ def __init__( boto_session: Any, region: str | None, service: str = "execute-api", + signing_url_prefix: str | None = None, ): """Initialize SigV4AuthManager. @@ -371,21 +373,66 @@ def __init__( boto_session: A boto3.Session used to resolve AWS credentials. region: SigV4 signing region; falls back to the boto session's region. service: SigV4 signing service name. + signing_url_prefix: When set, only URLs under this prefix are signed; + the delegate header still applies everywhere. """ self._delegate = delegate self._boto_session = boto_session self._region = region self._service = service + self._signing_url_prefix = signing_url_prefix + + @property + def delegate(self) -> AuthManager: + """The wrapped header-based AuthManager.""" + return self._delegate def auth_header(self) -> str | None: return self._delegate.auth_header() + def __getstate__(self) -> dict[str, Any]: + """Serialize without the boto3 session; restored instances refuse to sign.""" + state = self.__dict__.copy() + state["_boto_session"] = None + return state + + def _in_signing_scope(self, url: str) -> bool: + """Prefix match with normalized scheme/host/default port.""" + from urllib import parse + + default_ports = {"http": 80, "https": 443} + + def parts(value: str) -> tuple[str, str, int | None, str]: + split = parse.urlsplit(value) + scheme = split.scheme.lower() + return scheme, (split.hostname or "").lower(), split.port or default_ports.get(scheme), split.path.lower() + + p_scheme, p_host, p_port, p_path = parts(self._signing_url_prefix or "") + u_scheme, u_host, u_port, u_path = parts(url) + return (p_scheme, p_host, p_port) == (u_scheme, u_host, u_port) and u_path.startswith(p_path) + def sign_request(self, request: PreparedRequest) -> PreparedRequest: import hashlib from urllib import parse from botocore.awsrequest import AWSRequest + # Apply the delegate's own hook first so its mutations are signed too. + request = self._delegate.sign_request(request) + + # Only sign catalog requests (case-insensitive, like adapter mounts); a + # redirect leaving the signing scope must not keep stale artifacts. + if self._signing_url_prefix and not self._in_signing_scope(str(request.url)): + if str(request.headers.get("Authorization", "")).startswith("AWS4-HMAC-SHA256"): + del request.headers["Authorization"] + for header in ("X-Amz-Date", "x-amz-content-sha256", "X-Amz-Security-Token"): + request.headers.pop(header, None) + for header in [h for h in request.headers if h.lower().startswith("original-")]: + del request.headers[header] + return request + + if self._boto_session is None: + raise ValueError("SigV4AuthManager cannot sign after deserialization: the boto3 session is not serialized") credentials = self._boto_session.get_credentials().get_frozen_credentials() region = self._region or self._boto_session.region_name @@ -414,22 +461,31 @@ def sign_request(self, request: PreparedRequest) -> PreparedRequest: else: content_sha256_header = EMPTY_BODY_SHA256 - signing_headers = dict(request.headers) + # Case-insensitive: a plain dict would keep duplicate header spellings and + # let the signed canonical headers diverge from what is sent. + signing_headers = CaseInsensitiveDict(request.headers) + # A previous hop's own SigV4 signature is never a delegate header: discard, + # don't relocate. + if str(signing_headers.get("Authorization", "")).startswith("AWS4-HMAC-SHA256"): + signing_headers.pop("Authorization") # Relocate Authorization before signing so it lands in SignedHeaders, like Java. if "Authorization" in signing_headers: signing_headers["Original-Authorization"] = signing_headers.pop("Authorization") signing_headers["x-amz-content-sha256"] = content_sha256_header - aws_request = AWSRequest(method=request.method, url=url, params=params, data=request.body, headers=signing_headers) + aws_request = AWSRequest(method=request.method, url=url, params=params, data=request.body, headers=dict(signing_headers)) _iceberg_sigv4_auth_class()(credentials, self._service, region).add_auth(aws_request) - original_header = dict(request.headers) - signed_headers = dict(aws_request.headers) + original_header = CaseInsensitiveDict(request.headers) + signed_headers = CaseInsensitiveDict(dict(aws_request.headers)) relocated_headers = {} - # relocate headers if there is a conflict with signed headers + # Relocate conflicting headers; Authorization/Original-Authorization are + # managed above and must never re-relocate into Original-Original-* chains. for header, value in original_header.items(): + if header.lower() in ("authorization", "original-authorization"): + continue if header in signed_headers and signed_headers[header] != value: relocated_headers[f"Original-{header}"] = value @@ -467,11 +523,76 @@ def __call__(self, request: PreparedRequest) -> PreparedRequest: Returns: requests.PreparedRequest: The modified request. """ + # Capture tracking before signing: sign_request may return a fresh request. + tracked = {str(header) for header in getattr(request, "_auth_applied_headers", ())} + before = {key.lower(): value for key, value in request.headers.items()} if auth_header := self.auth_manager.auth_header(): request.headers["Authorization"] = auth_header # Header first, then sign: a request-signing AuthManager (e.g. SigV4) must # see the Authorization header so it can relocate it before signing. - return self.auth_manager.sign_request(request) + request = self.auth_manager.sign_request(request) + # Union with earlier hops: a stable credential re-applied with an unchanged + # value yields an empty diff and would otherwise be forgotten. + tracked.update(key for key, value in request.headers.items() if before.get(key.lower()) != value) + request._auth_applied_headers = sorted(tracked) # type: ignore[attr-defined] + return request + + +class ReauthenticatingSession(Session): + """A Session that re-applies `self.auth` to redirected requests. + + `requests` invokes session auth only on the original request, so a + request-signing AuthManager (e.g. SigV4) would send a signature computed for the + pre-redirect URL. Auth is re-applied only to `trusted_auth_origin` (the catalog + URI): a redirect chain that leaves it stays stripped of all credential-bearing + headers, and is re-authenticated only if it returns. + """ + + # Class-level default: Session pickles only __attrs__; old pickles fall back here. + _trusted_auth_origin: str | None = None + __attrs__ = [*Session.__attrs__, "_trusted_auth_origin"] + + def __init__(self, trusted_auth_origin: str | None = None) -> None: + super().__init__() + self._trusted_auth_origin = trusted_auth_origin + + # Stripped on cross-origin redirects (requests strips only Authorization); + # relocated credentials are stripped separately by their Original-* prefix. + _SIGNING_ARTIFACT_HEADERS = ("X-Amz-Security-Token", "X-Amz-Date", "x-amz-content-sha256") + + def rebuild_auth(self, prepared_request: PreparedRequest, response: Response) -> None: + original_url = str(response.request.url) + super().rebuild_auth(prepared_request, response) + cross_origin_hop = self.should_strip_auth(original_url, str(prepared_request.url)) + if cross_origin_hop: + # Strip whatever auth introduced, plus relocated headers and artifacts. + for header in getattr(response.request, "_auth_applied_headers", ()): + prepared_request.headers.pop(header, None) + for header in [h for h in prepared_request.headers if h.lower().startswith("original-")]: + del prepared_request.headers[header] + for header in self._SIGNING_ARTIFACT_HEADERS: + prepared_request.headers.pop(header, None) + if callable(self.auth) and self._may_reapply_auth(response, prepared_request, cross_origin_hop): + # copy() drops attributes: carry tracking forward for the adapter's union. + if prev_tracked := getattr(response.request, "_auth_applied_headers", ()): + prepared_request._auth_applied_headers = list(prev_tracked) # type: ignore[attr-defined] + # prepare_auth honors auth callables that return a replacement request; + # a caller-supplied static Authorization survives header-less managers. + prepared_request.prepare_auth(self.auth) + + def _may_reapply_auth(self, response: Response, prepared_request: PreparedRequest, cross_origin_hop: bool) -> bool: + """Whether re-applying auth on this hop is safe. + + A hop that is same-origin with its predecessor may still sit on a foreign + host reached through an earlier cross-origin redirect (A -> B -> B), so the + decision is made against the trusted origin, not the previous hop. + """ + if self._trusted_auth_origin is not None: + return not self.should_strip_auth(self._trusted_auth_origin, str(prepared_request.url)) + if cross_origin_hop: + return False + # No trusted origin: re-apply only if the previous hop still carried auth. + return "Authorization" in response.request.headers class AuthManagerFactory: diff --git a/tests/catalog/test_rest_auth.py b/tests/catalog/test_rest_auth.py index b6fb2a9882..08c6fe71c5 100644 --- a/tests/catalog/test_rest_auth.py +++ b/tests/catalog/test_rest_auth.py @@ -698,3 +698,712 @@ def test_sigv4_uses_client_profile_name(rest_mock: Mocker) -> None: aws_secret_access_key=None, aws_session_token=None, ) + + +def test_empty_body_sha256_importable_from_rest_package() -> None: + """EMPTY_BODY_SHA256 was public in released 0.11.x; the re-export must survive.""" + from pyiceberg.catalog.rest import EMPTY_BODY_SHA256 + + assert EMPTY_BODY_SHA256 == "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + + +def test_sigv4_relocates_lowercase_authorization() -> None: + """Header handling is case-insensitive: a lowercase authorization header is still relocated.""" + import boto3 + + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager(delegate=NoopAuthManager(), boto_session=boto_session, region="us-east-1") + + prepared = requests.Request("GET", "https://example.com/v1/config", headers={"authorization": "Bearer tok"}).prepare() + manager.sign_request(prepared) + + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert prepared.headers["Original-Authorization"] == "Bearer tok" + + +def test_sigv4_overwrites_conflicting_content_sha256_casing() -> None: + """A caller-supplied X-Amz-Content-SHA256 in different casing must not duplicate or survive signing.""" + import boto3 + + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager(delegate=NoopAuthManager(), boto_session=boto_session, region="us-east-1") + + body = b'{"k":"v"}' + prepared = requests.Request( + "POST", "https://example.com/v1/namespaces", data=body, headers={"X-Amz-Content-SHA256": "bogus"} + ).prepare() + manager.sign_request(prepared) + + expected = base64.b64encode(hashlib.sha256(body).digest()).decode() + values = [value for key, value in prepared.headers.items() if key.lower() == "x-amz-content-sha256"] + assert values == [expected] + assert prepared.headers["Original-X-Amz-Content-SHA256"] == "bogus" + + +def test_refresh_token_unwraps_sigv4_auth_manager() -> None: + """Legacy OAuth refresh must reach the delegate through a SigV4 wrapper.""" + from requests import Session + + from pyiceberg.catalog.rest.auth import LegacyOAuth2AuthManager, SigV4AuthManager + + legacy = MagicMock(spec=LegacyOAuth2AuthManager) + manager = SigV4AuthManager(delegate=legacy, boto_session=MagicMock(), region="us-east-1") + catalog = object.__new__(RestCatalog) + session = Session() + session.auth = AuthManagerAdapter(manager) + catalog._session = session + + catalog._refresh_token() + + legacy._refresh_token.assert_called_once() + + +def test_sigv4_custom_delegate_resolves_impl(rest_mock: Mocker) -> None: + """A custom delegate resolves through delegate.impl, like top-level custom auth.""" + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + catalog = RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "auth": { + "type": "sigv4", + "sigv4": { + "delegate": { + "type": "custom", + "impl": "pyiceberg.catalog.rest.auth.BasicAuthManager", + "custom": {"username": "u", "password": "p"}, + } + }, + }, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + assert isinstance(catalog._auth_manager, SigV4AuthManager) + assert isinstance(catalog._auth_manager.delegate, BasicAuthManager) + + +def test_sigv4_custom_delegate_requires_impl(rest_mock: Mocker) -> None: + """A custom delegate without impl fails fast instead of importing the literal string.""" + with pytest.raises(ValueError, match="auth.sigv4.delegate.impl must be specified"): + RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "auth": {"type": "sigv4", "sigv4": {"delegate": {"type": "custom"}}}, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + +def test_sigv4_non_custom_delegate_rejects_impl(rest_mock: Mocker) -> None: + """delegate.impl is only valid for a custom delegate, mirroring top-level auth.impl.""" + with pytest.raises(ValueError, match="auth.sigv4.delegate.impl can only be specified"): + RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "auth": { + "type": "sigv4", + "sigv4": {"delegate": {"type": "noop", "impl": "pyiceberg.catalog.rest.auth.BasicAuthManager"}}, + }, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + +def test_reauthenticating_session_resigns_same_origin_redirect(requests_mock: Mocker) -> None: + """Same-origin redirects are re-signed for the redirected URL, mirroring Java per-send signing.""" + import boto3 + + from pyiceberg.catalog.rest.auth import ReauthenticatingSession, SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager( + delegate=BasicAuthManager(username="u", password="p"), boto_session=boto_session, region="us-east-1" + ) + + session = ReauthenticatingSession() + session.auth = AuthManagerAdapter(manager) + + requests_mock.get("https://example.com/v1/old", status_code=302, headers={"Location": "https://example.com/v1/new"}) + requests_mock.get("https://example.com/v1/new", json={}) + + response = session.get("https://example.com/v1/old") + + assert response.status_code == 200 + first, second = requests_mock.request_history + assert second.url == "https://example.com/v1/new" + assert first.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert second.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert second.headers["Authorization"] != first.headers["Authorization"] + assert second.headers["Original-Authorization"].startswith("Basic ") + + +def test_reauthenticating_session_strips_credentials_cross_origin(requests_mock: Mocker) -> None: + """Cross-origin redirects must not forward any credential-bearing header.""" + import boto3 + + from pyiceberg.catalog.rest.auth import ReauthenticatingSession, SigV4AuthManager + + boto_session = boto3.Session( + aws_access_key_id="id", aws_secret_access_key="secret", aws_session_token="sts-token", region_name="us-east-1" + ) + manager = SigV4AuthManager( + delegate=BasicAuthManager(username="u", password="p"), boto_session=boto_session, region="us-east-1" + ) + + session = ReauthenticatingSession() + session.auth = AuthManagerAdapter(manager) + + requests_mock.get("https://example.com/v1/old", status_code=302, headers={"Location": "https://other.example.net/v1/new"}) + requests_mock.get("https://other.example.net/v1/new", json={}) + + # Stripping must be by Original-* prefix, not a fixed list. + response = session.get("https://example.com/v1/old", headers={"Original-Original-Authorization": "Bearer stale"}) + + assert response.status_code == 200 + first, second = requests_mock.request_history + assert first.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert first.headers["Original-Authorization"].startswith("Basic ") + assert first.headers["X-Amz-Security-Token"] == "sts-token" + for header in ( + "Authorization", + "Original-Authorization", + "Original-Original-Authorization", + "X-Amz-Security-Token", + "X-Amz-Date", + "x-amz-content-sha256", + ): + assert header not in second.headers + + +def test_sigv4_signing_scoped_to_url_prefix(requests_mock: Mocker) -> None: + """URLs outside signing_url_prefix are not signed; the delegate header still applies.""" + import boto3 + + from pyiceberg.catalog.rest.auth import ReauthenticatingSession, SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager( + delegate=BasicAuthManager(username="u", password="p"), + boto_session=boto_session, + region="us-east-1", + signing_url_prefix="https://catalog.example.com/", + ) + + session = ReauthenticatingSession() + session.auth = AuthManagerAdapter(manager) + + requests_mock.post("https://auth.other.example.net/token", json={"access_token": "t", "token_type": "bearer"}) + requests_mock.get("https://catalog.example.com/v1/config", json={}) + + session.post("https://auth.other.example.net/token", data={"grant_type": "client_credentials"}) + session.get("https://catalog.example.com/v1/config") + + token_request, catalog_request = requests_mock.request_history + assert token_request.headers["Authorization"].startswith("Basic ") + assert "X-Amz-Date" not in token_request.headers + assert catalog_request.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + + +def test_sigv4_catalog_wires_signing_prefix_to_catalog_uri(rest_mock: Mocker) -> None: + """The catalog scopes SigV4 signing to its own URI.""" + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + catalog = RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "auth": {"type": "sigv4", "sigv4": {"delegate": {"type": "noop"}}}, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + assert isinstance(catalog._auth_manager, SigV4AuthManager) + assert catalog._auth_manager._signing_url_prefix == TEST_URI + + +def test_plain_manager_keeps_static_authorization_on_same_origin_redirect(requests_mock: Mocker) -> None: + """A header-less manager must not lose a caller-supplied Authorization on same-origin redirects.""" + from pyiceberg.catalog.rest.auth import ReauthenticatingSession + + session = ReauthenticatingSession() + session.auth = AuthManagerAdapter(NoopAuthManager()) + + requests_mock.get("https://example.com/v1/old", status_code=307, headers={"Location": "https://example.com/v1/new"}) + requests_mock.get("https://example.com/v1/new", json={}) + + session.get("https://example.com/v1/old", headers={"Authorization": "Bearer table-token"}) + + _, second = requests_mock.request_history + assert second.headers["Authorization"] == "Bearer table-token" + + +def test_sigv4_discards_stale_self_signature_and_keeps_static_bearer(requests_mock: Mocker) -> None: + """A stale SigV4 signature is discarded on redirect; a static bearer relocates once.""" + import boto3 + + from pyiceberg.catalog.rest.auth import ReauthenticatingSession, SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager(delegate=NoopAuthManager(), boto_session=boto_session, region="us-east-1") + + session = ReauthenticatingSession() + session.auth = AuthManagerAdapter(manager) + + requests_mock.get("https://example.com/v1/old", status_code=307, headers={"Location": "https://example.com/v1/new"}) + requests_mock.get("https://example.com/v1/new", json={}) + + session.get("https://example.com/v1/old", headers={"Authorization": "Bearer table-token"}) + + first, second = requests_mock.request_history + for hop in (first, second): + assert hop.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert hop.headers["Original-Authorization"] == "Bearer table-token" + assert "Original-Original-Authorization" not in second.headers + + +def test_sigv4_rotating_delegate_does_not_spawn_recursive_relocation(requests_mock: Mocker) -> None: + """A delegate that rotates its token across hops must not resurrect the old token.""" + import boto3 + + from pyiceberg.catalog.rest.auth import AuthManager, ReauthenticatingSession, SigV4AuthManager + + class RotatingAuthManager(AuthManager): + def __init__(self) -> None: + self.calls = 0 + + def auth_header(self) -> str: + self.calls += 1 + return f"Bearer token-{self.calls}" + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager(delegate=RotatingAuthManager(), boto_session=boto_session, region="us-east-1") + + session = ReauthenticatingSession() + session.auth = AuthManagerAdapter(manager) + + requests_mock.get("https://example.com/v1/old", status_code=307, headers={"Location": "https://example.com/v1/new"}) + requests_mock.get("https://example.com/v1/new", json={}) + + session.get("https://example.com/v1/old") + + first, second = requests_mock.request_history + assert first.headers["Original-Authorization"] == "Bearer token-1" + assert second.headers["Original-Authorization"] == "Bearer token-2" + assert "Original-Original-Authorization" not in second.headers + assert "token-1" not in str(second.headers) + + +def test_sigv4_signing_prefix_match_is_case_insensitive() -> None: + """Prefix matching mirrors requests' adapter-mount matching (case-insensitive).""" + import boto3 + + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager( + delegate=NoopAuthManager(), + boto_session=boto_session, + region="us-east-1", + signing_url_prefix="https://Catalog.Example.COM/", + ) + + prepared = requests.Request("GET", "https://catalog.example.com/v1/config").prepare() + manager.sign_request(prepared) + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + + +def test_sigv4_file_io_receives_pickleable_delegate(rest_mock: Mocker) -> None: + """FileIO gets the pickleable delegate, not the SigV4 wrapper.""" + import pickle + + from pyiceberg.catalog.rest.auth import AUTH_MANAGER, SigV4AuthManager + + catalog = RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "auth": {"type": "sigv4", "sigv4": {"delegate": {"type": "noop"}}}, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + file_io = catalog._load_file_io() + assert isinstance(file_io.properties[AUTH_MANAGER], NoopAuthManager) + assert not isinstance(file_io.properties[AUTH_MANAGER], SigV4AuthManager) + pickle.dumps(file_io) + + +def test_sigv4_delegate_without_type_raises(rest_mock: Mocker) -> None: + """A non-empty delegate block without a type is a config error, not 'no delegate'.""" + with pytest.raises(ValueError, match="auth.sigv4.delegate.type must be defined"): + RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "auth": {"type": "sigv4", "sigv4": {"delegate": {"oauth2": {"client_id": "x"}}}}, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + + +def test_stripped_auth_stays_stripped_across_foreign_same_origin_hops(requests_mock: Mocker) -> None: + """A -> B (cross-origin, stripped) -> B (same-origin) must NOT re-apply catalog auth on B.""" + import boto3 + + from pyiceberg.catalog.rest.auth import ReauthenticatingSession, SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager( + delegate=BasicAuthManager(username="u", password="p"), + boto_session=boto_session, + region="us-east-1", + signing_url_prefix="https://catalog.example.com/", + ) + + session = ReauthenticatingSession(trusted_auth_origin="https://catalog.example.com/") + session.auth = AuthManagerAdapter(manager) + + requests_mock.get("https://catalog.example.com/v1/a", status_code=302, headers={"Location": "https://other.example.net/b"}) + requests_mock.get("https://other.example.net/b", status_code=302, headers={"Location": "https://other.example.net/c"}) + requests_mock.get("https://other.example.net/c", json={}) + + session.get("https://catalog.example.com/v1/a") + + first, second, third = requests_mock.request_history + assert first.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + for hop in (second, third): + assert "Authorization" not in hop.headers + assert "Original-Authorization" not in hop.headers + + +def test_auth_reapplied_when_chain_returns_to_trusted_origin(requests_mock: Mocker) -> None: + """A -> B (stripped) -> back to A: the trusted origin gets fresh auth again.""" + import boto3 + + from pyiceberg.catalog.rest.auth import ReauthenticatingSession, SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager( + delegate=BasicAuthManager(username="u", password="p"), + boto_session=boto_session, + region="us-east-1", + signing_url_prefix="https://catalog.example.com/", + ) + + session = ReauthenticatingSession(trusted_auth_origin="https://catalog.example.com/") + session.auth = AuthManagerAdapter(manager) + + requests_mock.get("https://catalog.example.com/v1/a", status_code=302, headers={"Location": "https://other.example.net/b"}) + requests_mock.get("https://other.example.net/b", status_code=302, headers={"Location": "https://catalog.example.com/v1/c"}) + requests_mock.get("https://catalog.example.com/v1/c", json={}) + + session.get("https://catalog.example.com/v1/a") + + _, second, third = requests_mock.request_history + assert "Authorization" not in second.headers + assert third.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert third.headers["Original-Authorization"].startswith("Basic ") + + +def test_sigv4_without_delegate_preserves_legacy_token(rest_mock: Mocker) -> None: + """Migrating rest.sigv4-enabled + token to auth.type=sigv4 keeps the legacy OAuth delegate.""" + from pyiceberg.catalog.rest.auth import LegacyOAuth2AuthManager, SigV4AuthManager + + catalog = RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "token": "legacy-token", + "auth": {"type": "sigv4"}, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + assert isinstance(catalog._auth_manager, SigV4AuthManager) + assert isinstance(catalog._auth_manager.delegate, LegacyOAuth2AuthManager) + assert catalog._auth_manager.delegate.auth_header() == "Bearer legacy-token" + + +def test_redirect_leaving_signing_prefix_clears_stale_artifacts(requests_mock: Mocker) -> None: + """A same-origin redirect that leaves the signing scope must not carry the old hop's signature.""" + import boto3 + + from pyiceberg.catalog.rest.auth import ReauthenticatingSession, SigV4AuthManager + + boto_session = boto3.Session( + aws_access_key_id="id", aws_secret_access_key="secret", aws_session_token="sts-token", region_name="us-east-1" + ) + manager = SigV4AuthManager( + delegate=BasicAuthManager(username="u", password="p"), + boto_session=boto_session, + region="us-east-1", + signing_url_prefix="https://host.example.com/catalog/", + ) + + session = ReauthenticatingSession(trusted_auth_origin="https://host.example.com/catalog/") + session.auth = AuthManagerAdapter(manager) + + requests_mock.get( + "https://host.example.com/catalog/v1/a", status_code=302, headers={"Location": "https://host.example.com/other"} + ) + requests_mock.get("https://host.example.com/other", json={}) + + session.get("https://host.example.com/catalog/v1/a") + + first, second = requests_mock.request_history + assert first.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert second.headers["Authorization"].startswith("Basic ") + for header in ("X-Amz-Date", "X-Amz-Security-Token", "x-amz-content-sha256", "Original-Authorization"): + assert header not in second.headers + + +def test_redirect_leaving_signing_prefix_with_headerless_delegate(requests_mock: Mocker) -> None: + """With a header-less delegate, the stale AWS4 Authorization itself must be dropped.""" + import boto3 + + from pyiceberg.catalog.rest.auth import ReauthenticatingSession, SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager( + delegate=NoopAuthManager(), + boto_session=boto_session, + region="us-east-1", + signing_url_prefix="https://host.example.com/catalog/", + ) + + session = ReauthenticatingSession(trusted_auth_origin="https://host.example.com/catalog/") + session.auth = AuthManagerAdapter(manager) + + requests_mock.get( + "https://host.example.com/catalog/v1/a", status_code=302, headers={"Location": "https://host.example.com/other"} + ) + requests_mock.get("https://host.example.com/other", json={}) + + session.get("https://host.example.com/catalog/v1/a") + + _, second = requests_mock.request_history + for header in ("Authorization", "X-Amz-Date", "x-amz-content-sha256", "Original-Authorization"): + assert header not in second.headers + + +def test_sigv4_applies_delegate_sign_request_hook() -> None: + """A delegate's sign_request mutations must be applied (and signed) when nested.""" + import boto3 + + from pyiceberg.catalog.rest.auth import AuthManager, SigV4AuthManager + + class HookedAuthManager(AuthManager): + def auth_header(self) -> None: + return None + + def sign_request(self, request: requests.PreparedRequest) -> requests.PreparedRequest: + request.headers["X-Delegate-Hook"] = "applied" + return request + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager(delegate=HookedAuthManager(), boto_session=boto_session, region="us-east-1") + + prepared = requests.Request("GET", "https://example.com/v1/config").prepare() + manager.sign_request(prepared) + + assert prepared.headers["X-Delegate-Hook"] == "applied" + assert "x-delegate-hook" in prepared.headers["Authorization"] + + +def test_reauthenticating_session_survives_pickle() -> None: + """The trusted origin must survive a Session pickle round trip.""" + import pickle + + from pyiceberg.catalog.rest.auth import ReauthenticatingSession + + session = ReauthenticatingSession(trusted_auth_origin="https://cat.example.com/") + restored = pickle.loads(pickle.dumps(session)) + assert restored._trusted_auth_origin == "https://cat.example.com/" + + +def test_cross_origin_strips_custom_auth_headers(requests_mock: Mocker) -> None: + """Headers introduced by a custom sign_request hook must not cross origins.""" + from pyiceberg.catalog.rest.auth import AuthManager, ReauthenticatingSession + + class ApiKeyAuthManager(AuthManager): + def auth_header(self) -> None: + return None + + def sign_request(self, request: requests.PreparedRequest) -> requests.PreparedRequest: + request.headers["X-Api-Key"] = "secret-key" + return request + + session = ReauthenticatingSession(trusted_auth_origin="https://example.com/") + session.auth = AuthManagerAdapter(ApiKeyAuthManager()) + + requests_mock.get("https://example.com/v1/a", status_code=302, headers={"Location": "https://other.example.net/b"}) + requests_mock.get("https://other.example.net/b", json={}) + + session.get("https://example.com/v1/a") + + first, second = requests_mock.request_history + assert first.headers["X-Api-Key"] == "secret-key" + assert "X-Api-Key" not in second.headers + + +def test_redirect_honors_copy_returning_sign_request(requests_mock: Mocker) -> None: + """An auth hook that returns a replacement request must take effect on redirects.""" + import boto3 + + from pyiceberg.catalog.rest.auth import AuthManager, ReauthenticatingSession, SigV4AuthManager + + class CopyingAuthManager(AuthManager): + def auth_header(self) -> None: + return None + + def sign_request(self, request: requests.PreparedRequest) -> requests.PreparedRequest: + replacement = request.copy() + replacement.headers["X-Copy"] = "yes" + return replacement + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager(delegate=CopyingAuthManager(), boto_session=boto_session, region="us-east-1") + + session = ReauthenticatingSession(trusted_auth_origin="https://example.com/") + session.auth = AuthManagerAdapter(manager) + + requests_mock.get("https://example.com/v1/a", status_code=302, headers={"Location": "https://example.com/v1/b"}) + requests_mock.get("https://example.com/v1/b", json={}) + + session.get("https://example.com/v1/a") + + first, second = requests_mock.request_history + assert first.headers["X-Copy"] == "yes" + assert second.headers["X-Copy"] == "yes" + # The redirected hop carries a fresh signature, not the discarded stale one. + assert second.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + assert second.headers["Authorization"] != first.headers["Authorization"] + + +def test_stable_custom_auth_header_stripped_after_intermediate_hop(requests_mock: Mocker) -> None: + """A -> A (same-origin) -> B: tracking must survive the middle hop so B gets no credential.""" + from pyiceberg.catalog.rest.auth import AuthManager, ReauthenticatingSession + + class ApiKeyAuthManager(AuthManager): + def auth_header(self) -> None: + return None + + def sign_request(self, request: requests.PreparedRequest) -> requests.PreparedRequest: + request.headers["X-Api-Key"] = "secret-key" + return request + + session = ReauthenticatingSession(trusted_auth_origin="https://example.com/") + session.auth = AuthManagerAdapter(ApiKeyAuthManager()) + + requests_mock.get("https://example.com/v1/a", status_code=302, headers={"Location": "https://example.com/v1/b"}) + requests_mock.get("https://example.com/v1/b", status_code=302, headers={"Location": "https://other.example.net/c"}) + requests_mock.get("https://other.example.net/c", json={}) + + session.get("https://example.com/v1/a") + + first, second, third = requests_mock.request_history + assert first.headers["X-Api-Key"] == "secret-key" + assert second.headers["X-Api-Key"] == "secret-key" + assert "X-Api-Key" not in third.headers + + +def test_copy_returning_manager_tracking_survives_chain(requests_mock: Mocker) -> None: + """A -> A -> B with a copy-returning manager: stable credentials still stripped at B.""" + from pyiceberg.catalog.rest.auth import AuthManager, ReauthenticatingSession + + class CopyingApiKeyManager(AuthManager): + def auth_header(self) -> None: + return None + + def sign_request(self, request: requests.PreparedRequest) -> requests.PreparedRequest: + replacement = request.copy() + replacement.headers["X-Api-Key"] = "secret-key" + return replacement + + session = ReauthenticatingSession(trusted_auth_origin="https://example.com/") + session.auth = AuthManagerAdapter(CopyingApiKeyManager()) + + requests_mock.get("https://example.com/v1/a", status_code=302, headers={"Location": "https://example.com/v1/b"}) + requests_mock.get("https://example.com/v1/b", status_code=302, headers={"Location": "https://other.example.net/c"}) + requests_mock.get("https://other.example.net/c", json={}) + + session.get("https://example.com/v1/a") + + first, second, third = requests_mock.request_history + assert first.headers["X-Api-Key"] == "secret-key" + assert second.headers["X-Api-Key"] == "secret-key" + assert "X-Api-Key" not in third.headers + + +def test_signing_scope_normalizes_default_port() -> None: + """An explicit default port in the configured prefix must still match normalized URLs.""" + import boto3 + + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager( + delegate=NoopAuthManager(), + boto_session=boto_session, + region="us-east-1", + signing_url_prefix="https://catalog.example.com:443/", + ) + + prepared = requests.Request("GET", "https://catalog.example.com/v1/config").prepare() + manager.sign_request(prepared) + assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256 Credential=") + + +def test_sigv4_file_io_pickles_with_legacy_delegate(rest_mock: Mocker) -> None: + """FileIO from a sigv4 catalog with the legacy token fallback must pickle.""" + import pickle + + catalog = RestCatalog( + "rest", + **{ # type: ignore + "uri": TEST_URI, + "token": "legacy-token", + "auth": {"type": "sigv4"}, + "rest.signing-region": "us-east-1", + "client.access-key-id": "id", + "client.secret-access-key": "secret", + }, + ) + pickle.dumps(catalog._load_file_io()) + + +def test_deserialized_sigv4_manager_refuses_to_sign() -> None: + """A pickled SigV4AuthManager restores auth_header/delegate but refuses to sign.""" + import pickle + + import boto3 + + from pyiceberg.catalog.rest.auth import SigV4AuthManager + + boto_session = boto3.Session(aws_access_key_id="id", aws_secret_access_key="secret", region_name="us-east-1") + manager = SigV4AuthManager( + delegate=BasicAuthManager(username="u", password="p"), boto_session=boto_session, region="us-east-1" + ) + + restored = pickle.loads(pickle.dumps(manager)) + assert restored.auth_header().startswith("Basic ") + with pytest.raises(ValueError, match="cannot sign after deserialization"): + restored.sign_request(requests.Request("GET", "https://example.com/v1/config").prepare())