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 97a71b429b..bc9af0db59 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -31,7 +31,19 @@ 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, + 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, @@ -253,11 +265,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" @@ -426,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) @@ -441,10 +454,70 @@ 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(): + 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 + + 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: + 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") + 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") + # 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") @@ -454,17 +527,30 @@ 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"), + # Sign only catalog requests; an external oauth2-server-uri stays unsigned. + signing_url_prefix=self.uri, + ) @staticmethod def _resolve_storage_credentials(storage_credentials: list[StorageCredential], location: str | None) -> Properties: @@ -489,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 @@ -767,64 +858,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: - 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 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"] - # 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 - - aws_request = AWSRequest( - method=request.method, url=url, params=params, data=request.body, headers=dict(request.headers) - ) - - SigV4Auth(credentials, service, region).add_auth(aws_request) - original_header = request.headers - signed_headers = 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( @@ -869,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 602074282c..b344459580 100644 --- a/pyiceberg/catalog/rest/auth.py +++ b/pyiceberg/catalog/rest/auth.py @@ -21,12 +21,13 @@ 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 -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 @@ -36,6 +37,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 +80,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 +351,149 @@ 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", + signing_url_prefix: str | None = None, + ): + """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. + 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 + + 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 + + # 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=dict(signing_headers)) + + _iceberg_sigv4_auth_class()(credentials, self._service, region).add_auth(aws_request) + + original_header = CaseInsensitiveDict(request.headers) + signed_headers = CaseInsensitiveDict(dict(aws_request.headers)) + relocated_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 + + 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,19 +515,86 @@ 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. """ + # 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. + 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: _registry: dict[str, type["AuthManager"]] = {} 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 43790f778a..405446cd7e 100644 --- a/tests/catalog/test_rest.py +++ b/tests/catalog/test_rest.py @@ -24,7 +24,6 @@ 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 @@ -33,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, @@ -589,77 +587,36 @@ 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 + # 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( + assert 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) - - assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256") - assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" - assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256 - - -def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None: - existing_token = "existing_token" - - catalog = RestCatalog( - "rest", - **{ - "uri": TEST_URI, - "token": existing_token, + "token": TEST_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) - - assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256") - assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}" - assert prepared.headers.get("x-amz-content-sha256") != EMPTY_BODY_SHA256 + ).list_tables(namespace) == [("examples", "fooshare")] + 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", @@ -678,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", @@ -697,29 +657,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( @@ -877,18 +814,30 @@ 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 + # 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 @@ -2751,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( @@ -2759,7 +2711,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 @@ -2783,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 @@ -2793,7 +2758,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..08c6fe71c5 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,1154 @@ 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 ") + # 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 + + 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_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 + + 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 + + +@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" + + 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 + + +@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" + + 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 + + +@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 + + 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 + + +@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 + + 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() + + +@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":{}}' + 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) + + +@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 + + # 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, + ) + + +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()) 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" },