diff --git a/.agents/skills/a2a-workflow/SKILL.md b/.agents/skills/a2a-workflow/SKILL.md index 7e080b0..d33bf0f 100644 --- a/.agents/skills/a2a-workflow/SKILL.md +++ b/.agents/skills/a2a-workflow/SKILL.md @@ -249,6 +249,22 @@ Helpful environment variables for appliance environments that use an internal CA - `REQUESTS_CA_BUNDLE` for HTTP requests - `WEBSOCKET_CLIENT_CA_BUNDLE` for SignalR/WebSocket traffic +#### TLS 1.3 (SPP 9.0) + +If **async** A2A credential retrieval or `CertificateAuth` login fails on SPP 9.0 with `60094 Authorization is denied` while the **sync** path works, the cause is TLS 1.3 post-handshake authentication. The async `SSLContext` must set `post_handshake_auth = True` (done in `AsyncSafeguardClient._create_ssl_context`); without it the client never answers the server's post-handshake `CertificateRequest`. The sync (`requests`/urllib3) path enables this by default. + +To force a TLS version, pass the opt-in `min_tls_version` / `max_tls_version` (`ssl.TLSVersion | None`) to `A2AContext` / `AsyncA2AContext` (also on `SafeguardClient` / `AsyncSafeguardClient` and the `quick_*` classmethods). Examples: + +```python +import ssl +# Require TLS 1.3 +A2AContext(host, cert, key, min_tls_version=ssl.TLSVersion.TLSv1_3) +# Interim: cap at TLS 1.2 +A2AContext(host, cert, key, max_tls_version=ssl.TLSVersion.TLSv1_2) +``` + +Defaults are `None` (negotiate normally). Cert auth requires HTTP/1.1 — never enable HTTP/2, which disallows the post-handshake `CertificateRequest`. + ### Safe debugging rules - never print or log `password.value` or private-key plaintext in committed samples/tests diff --git a/.agents/skills/api-patterns/SKILL.md b/.agents/skills/api-patterns/SKILL.md index 74abc1e..380168e 100644 --- a/.agents/skills/api-patterns/SKILL.md +++ b/.agents/skills/api-patterns/SKILL.md @@ -349,6 +349,24 @@ client = SafeguardClient("host", auth=auth, verify=False) Set these when the appliance uses a certificate signed by an internal CA. +### TLS Version Pinning (opt-in) + +`SafeguardClient`, `AsyncSafeguardClient`, `A2AContext`, `AsyncA2AContext` +(and the A2A `quick_*` classmethods) accept optional `min_tls_version` / +`max_tls_version` (`ssl.TLSVersion | None`, default `None` = negotiate). + +```python +import ssl +# Require TLS 1.3 (SPP 9.0) +client = SafeguardClient("host", auth=auth, min_tls_version=ssl.TLSVersion.TLSv1_3) +# Interim: cap at TLS 1.2 +client = SafeguardClient("host", auth=auth, max_tls_version=ssl.TLSVersion.TLSv1_2) +``` + +Pins govern the client's request transport (all API, token, and A2A calls). +Async cert/A2A auth enables `post_handshake_auth` so it works over TLS 1.3; +the sync path enables it by default. Keep HTTP/1.1 (no HTTP/2). + ## Common Patterns ### Query parameters diff --git a/AGENTS.md b/AGENTS.md index dc9940c..187a3fe 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,6 +79,23 @@ and publishing workflow. - Prefer CA bundle verification over `verify=False`; use `REQUESTS_CA_BUNDLE` / `WEBSOCKET_CLIENT_CA_BUNDLE` when needed. +### TLS 1.3 + +SPP 9.0 enables TLS 1.3. Certificate/A2A auth over TLS 1.3 needs +post-handshake authentication (RFC 8446 §4.6.2): the sync (`requests`/urllib3) +path enables it by default, and the async (`aiohttp`) path enables +`post_handshake_auth` in `AsyncSafeguardClient._create_ssl_context`. Do not +remove it or async cert/A2A auth fails on 9.0 (error 60094). + +Both clients and both A2A contexts accept optional, opt-in +`min_tls_version` / `max_tls_version` (`ssl.TLSVersion | None`, default +`None` = negotiate). They govern the client's request transport (all API, +token, and A2A traffic): async applies them in `_create_ssl_context`; sync +mounts a `_TlsVersionAdapter` using urllib3's native +`ssl_minimum_version` / `ssl_maximum_version` only when a pin is set, so the +default path is unchanged. Keep HTTP/1.1 (never enable HTTP/2), since cert +auth's post-handshake CertificateRequest is disallowed under HTTP/2. + ## Versioning `pyproject.toml` holds the base semantic version. CI stamps tagged releases and diff --git a/README.md b/README.md index 980f959..21e4933 100644 --- a/README.md +++ b/README.md @@ -121,6 +121,82 @@ adapting a sample for production, remove `verify=False` and configure trust via `REQUESTS_CA_BUNDLE` (and `WEBSOCKET_CLIENT_CA_BUNDLE` if you use SignalR) or pass an explicit CA bundle path to `verify`. +> **TLS 1.3 / SPP 9.0:** if you are connecting to SPP 9.0 (which enables +> TLS 1.3), see [TLS 1.3 and SPP 9.0](#tls-13-and-spp-90) below — especially +> if you build your own `aiohttp`/`ssl` context for certificate or A2A auth. + +## TLS 1.3 and SPP 9.0 + +Starting with **SPP 9.0**, the appliance enables **TLS 1.3**. PySafeguard +`8.2.0` and later negotiate TLS 1.3 automatically — for most applications there +is **nothing to change**. Certificate-based login and A2A credential retrieval +continue to work over TLS 1.3 on both the sync and async clients. + +### What changed under the hood + +TLS 1.3 moves client-certificate authentication to a **post-handshake** +exchange (RFC 8446 §4.6.2). Instead of the client presenting its certificate +during the initial handshake, the server sends a `CertificateRequest` *after* +the handshake completes, and the client must answer it. Python only answers +that request when the underlying `ssl.SSLContext` has +`post_handshake_auth = True`. + +- **Sync client (`requests` / `urllib3`):** already enabled by default, so it + was never affected. +- **Async client (`aiohttp`):** PySafeguard now sets `post_handshake_auth` + explicitly on the SSL context it builds for certificate/A2A auth. Before + `8.2.0`, async certificate and A2A auth failed against SPP 9.0 with + `60094 Authorization is denied`. + +### Pinning the TLS version (opt-in) + +`SafeguardClient`, `AsyncSafeguardClient`, `A2AContext`, `AsyncA2AContext` +(and the A2A `quick_*` classmethods) accept optional `min_tls_version` and +`max_tls_version` arguments (`ssl.TLSVersion | None`, default `None` = +negotiate normally): + +```python +import ssl + +# Require TLS 1.3 (e.g. to enforce it against SPP 9.0) +client = SafeguardClient("host", auth=auth, min_tls_version=ssl.TLSVersion.TLSv1_3) + +# Interim: cap the connection at TLS 1.2 +client = SafeguardClient("host", auth=auth, max_tls_version=ssl.TLSVersion.TLSv1_2) + +# Also available on the A2A contexts +with A2AContext("host", "cert.pem", "key.pem", + min_tls_version=ssl.TLSVersion.TLSv1_3) as ctx: + password = ctx.retrieve_password(api_key) +``` + +The pins govern the client's request transport (all API, token, and A2A +traffic). Leaving them at `None` lets the platform negotiate the highest +mutually supported version, which is the recommended default. + +### Python-specific gotchas + +- **Post-handshake auth is required for cert/A2A auth on TLS 1.3.** If you + build your **own** `aiohttp`/`ssl` context instead of letting PySafeguard + create it (for example, a custom `AsyncSafeguardClient` subclass or a + hand-rolled A2A call), you **must** set `ssl_ctx.post_handshake_auth = True` + or cert/A2A auth will fail on SPP 9.0 with error `60094`. +- **Keep HTTP/1.1 — do not enable HTTP/2.** The post-handshake + `CertificateRequest` is disallowed under HTTP/2, so certificate auth breaks + over HTTP/2. Both `requests` and `aiohttp` default to HTTP/1.1; PySafeguard + relies on that and does not enable HTTP/2. +- **Your Python `ssl` must be built against OpenSSL 1.1.1 or newer** for + TLS 1.3 support. This is true of all supported CPython builds (3.10+), but + can bite on old or custom OpenSSL builds; check with + `ssl.HAS_TLSv1_3`. +- **Use the `ssl.TLSVersion` enum**, not integers or strings, for + `min_tls_version` / `max_tls_version` (e.g. `ssl.TLSVersion.TLSv1_3`). +- **A version pin forces a real SSL context even with `verify=False`.** On the + async client, `verify=False` normally skips building an SSL context entirely; + setting `min_tls_version`/`max_tls_version` makes PySafeguard build one anyway + so the floor/ceiling can be applied. Certificate verification stays disabled — + only the TLS version bounds are added. + ## Getting Started > **Note:** Recent versions of Safeguard have Resource Owner Grant (ROG) diff --git a/pyproject.toml b/pyproject.toml index fa3f6a0..16f2733 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api" [project] name = "pysafeguard" description = "One Identity Safeguard Python Package" -version = "8.1.0" +version = "8.2.0" readme = { file = "README.md", content-type = "text/markdown" } keywords = ["safeguard", "oneidentity"] license = "Apache" diff --git a/src/pysafeguard/a2a.py b/src/pysafeguard/a2a.py index 0165ed6..71ee2c2 100644 --- a/src/pysafeguard/a2a.py +++ b/src/pysafeguard/a2a.py @@ -33,6 +33,7 @@ from __future__ import annotations +import ssl import typing from typing import TYPE_CHECKING from types import TracebackType @@ -56,6 +57,10 @@ class A2AContext: :param key_file: Path to the certificate private key. :param verify: TLS verification — ``True``, ``False``, or a CA bundle path. :param api_version: API version (default ``"v4"``). + :param min_tls_version: Optional minimum TLS version to negotiate (e.g. + ``ssl.TLSVersion.TLSv1_3``). ``None`` (default) negotiates normally. + :param max_tls_version: Optional maximum TLS version to negotiate (e.g. + ``ssl.TLSVersion.TLSv1_2``). ``None`` (default) negotiates normally. """ def __init__( @@ -66,6 +71,8 @@ def __init__( *, verify: bool | str = True, api_version: LiteralString = "v4", + min_tls_version: ssl.TLSVersion | None = None, + max_tls_version: ssl.TLSVersion | None = None, ) -> None: if not cert_file or not key_file: raise ValueError("cert_file and key_file are required for A2A context") @@ -75,7 +82,13 @@ def __init__( self._verify = verify self._api_version = api_version - self._conn = SafeguardClient(host, verify=verify, api_version=api_version) + self._conn = SafeguardClient( + host, + verify=verify, + api_version=api_version, + min_tls_version=min_tls_version, + max_tls_version=max_tls_version, + ) self._user_authenticated = False # -- lifecycle ----------------------------------------------------------- @@ -281,6 +294,8 @@ def quick_retrieve_password( *, verify: bool | str = True, api_version: LiteralString = "v4", + min_tls_version: ssl.TLSVersion | None = None, + max_tls_version: ssl.TLSVersion | None = None, ) -> HiddenString: """One-shot password retrieval without creating a context. @@ -290,9 +305,19 @@ def quick_retrieve_password( :param key_file: Path to certificate key. :param verify: TLS verification setting. :param api_version: API version. + :param min_tls_version: Optional minimum TLS version to negotiate. + :param max_tls_version: Optional maximum TLS version to negotiate. :returns: The password wrapped in a :class:`~pysafeguard.HiddenString`. """ - with cls(host, cert_file, key_file, verify=verify, api_version=api_version) as ctx: + with cls( + host, + cert_file, + key_file, + verify=verify, + api_version=api_version, + min_tls_version=min_tls_version, + max_tls_version=max_tls_version, + ) as ctx: return ctx.retrieve_password(api_key) @classmethod @@ -306,6 +331,8 @@ def quick_retrieve_private_key( key_format: SshKeyFormat = SshKeyFormat.OPENSSH, verify: bool | str = True, api_version: LiteralString = "v4", + min_tls_version: ssl.TLSVersion | None = None, + max_tls_version: ssl.TLSVersion | None = None, ) -> HiddenString: """One-shot private key retrieval without creating a context. @@ -316,9 +343,19 @@ def quick_retrieve_private_key( :param key_format: Key format (default :attr:`SshKeyFormat.OPENSSH`). :param verify: TLS verification setting. :param api_version: API version. + :param min_tls_version: Optional minimum TLS version to negotiate. + :param max_tls_version: Optional maximum TLS version to negotiate. :returns: The private key wrapped in a :class:`~pysafeguard.HiddenString`. """ - with cls(host, cert_file, key_file, verify=verify, api_version=api_version) as ctx: + with cls( + host, + cert_file, + key_file, + verify=verify, + api_version=api_version, + min_tls_version=min_tls_version, + max_tls_version=max_tls_version, + ) as ctx: return ctx.retrieve_private_key(api_key, key_format=key_format) # -- Internal helpers ---------------------------------------------------- diff --git a/src/pysafeguard/async_a2a.py b/src/pysafeguard/async_a2a.py index 319283a..4c6fcab 100644 --- a/src/pysafeguard/async_a2a.py +++ b/src/pysafeguard/async_a2a.py @@ -15,6 +15,7 @@ from __future__ import annotations +import ssl import typing from typing import TYPE_CHECKING @@ -37,6 +38,10 @@ class AsyncA2AContext: :param key_file: Path to the certificate private key. :param verify: TLS verification — ``True``, ``False``, or a CA bundle path. :param api_version: API version (default ``"v4"``). + :param min_tls_version: Optional minimum TLS version to negotiate (e.g. + ``ssl.TLSVersion.TLSv1_3``). ``None`` (default) negotiates normally. + :param max_tls_version: Optional maximum TLS version to negotiate (e.g. + ``ssl.TLSVersion.TLSv1_2``). ``None`` (default) negotiates normally. """ def __init__( @@ -47,6 +52,8 @@ def __init__( *, verify: bool | str = True, api_version: LiteralString = "v4", + min_tls_version: ssl.TLSVersion | None = None, + max_tls_version: ssl.TLSVersion | None = None, ) -> None: if not cert_file or not key_file: raise ValueError("cert_file and key_file are required for A2A context") @@ -56,7 +63,13 @@ def __init__( self._verify = verify self._api_version = api_version - self._conn = AsyncSafeguardClient(host, verify=verify, api_version=api_version) + self._conn = AsyncSafeguardClient( + host, + verify=verify, + api_version=api_version, + min_tls_version=min_tls_version, + max_tls_version=max_tls_version, + ) self._user_authenticated = False # -- lifecycle ----------------------------------------------------------- @@ -255,6 +268,8 @@ async def quick_retrieve_password( *, verify: bool | str = True, api_version: LiteralString = "v4", + min_tls_version: ssl.TLSVersion | None = None, + max_tls_version: ssl.TLSVersion | None = None, ) -> HiddenString: """One-shot async password retrieval without creating a context. @@ -264,9 +279,19 @@ async def quick_retrieve_password( :param key_file: Path to certificate key. :param verify: TLS verification setting. :param api_version: API version. + :param min_tls_version: Optional minimum TLS version to negotiate. + :param max_tls_version: Optional maximum TLS version to negotiate. :returns: The password wrapped in a :class:`~pysafeguard.HiddenString`. """ - async with cls(host, cert_file, key_file, verify=verify, api_version=api_version) as ctx: + async with cls( + host, + cert_file, + key_file, + verify=verify, + api_version=api_version, + min_tls_version=min_tls_version, + max_tls_version=max_tls_version, + ) as ctx: return await ctx.retrieve_password(api_key) @classmethod @@ -280,6 +305,8 @@ async def quick_retrieve_private_key( key_format: SshKeyFormat = SshKeyFormat.OPENSSH, verify: bool | str = True, api_version: LiteralString = "v4", + min_tls_version: ssl.TLSVersion | None = None, + max_tls_version: ssl.TLSVersion | None = None, ) -> HiddenString: """One-shot async private key retrieval without creating a context. @@ -290,9 +317,19 @@ async def quick_retrieve_private_key( :param key_format: Key format (default :attr:`SshKeyFormat.OPENSSH`). :param verify: TLS verification setting. :param api_version: API version. + :param min_tls_version: Optional minimum TLS version to negotiate. + :param max_tls_version: Optional maximum TLS version to negotiate. :returns: The private key wrapped in a :class:`~pysafeguard.HiddenString`. """ - async with cls(host, cert_file, key_file, verify=verify, api_version=api_version) as ctx: + async with cls( + host, + cert_file, + key_file, + verify=verify, + api_version=api_version, + min_tls_version=min_tls_version, + max_tls_version=max_tls_version, + ) as ctx: return await ctx.retrieve_private_key(api_key, key_format=key_format) # -- Internal helpers ---------------------------------------------------- diff --git a/src/pysafeguard/async_client.py b/src/pysafeguard/async_client.py index 5cece7e..0c90e3c 100644 --- a/src/pysafeguard/async_client.py +++ b/src/pysafeguard/async_client.py @@ -49,6 +49,12 @@ class AsyncSafeguardClient: :param timeout: Request timeout in seconds (default 300). :param auto_refresh: If ``True``, automatically refresh the token before each request when the token has expired. + :param min_tls_version: Optional minimum TLS version to negotiate (e.g. + ``ssl.TLSVersion.TLSv1_3`` to require TLS 1.3). ``None`` (default) + negotiates normally. + :param max_tls_version: Optional maximum TLS version to negotiate (e.g. + ``ssl.TLSVersion.TLSv1_2`` to cap at TLS 1.2). ``None`` (default) + negotiates normally. """ def __init__( @@ -60,11 +66,15 @@ def __init__( api_version: LiteralString = "v4", timeout: int = DEFAULT_TIMEOUT, auto_refresh: bool = False, + min_tls_version: ssl.TLSVersion | None = None, + max_tls_version: ssl.TLSVersion | None = None, ) -> None: self.host = host self.verify = verify self.api_version = api_version self.auto_refresh = auto_refresh + self._min_tls_version = min_tls_version + self._max_tls_version = max_tls_version self._auth = auth self._user_token: str | None = None @@ -436,7 +446,7 @@ async def _check_and_refresh_token(self) -> None: def _create_ssl_context(self, cert: tuple[str, str] | None = None) -> ssl.SSLContext | bool: """Build an SSL context based on verification and client certificate settings.""" - if self.verify is False and cert is None: + if self.verify is False and cert is None and self._min_tls_version is None and self._max_tls_version is None: return False ctx = SSLContext(ssl.PROTOCOL_TLS_CLIENT) @@ -447,6 +457,16 @@ def _create_ssl_context(self, cert: tuple[str, str] | None = None) -> ssl.SSLCon ctx.verify_mode = ssl.CERT_NONE if cert is not None: ctx.load_cert_chain(cert[0], cert[1]) + # Enable post-handshake authentication (RFC 8446 s4.6.2) so the client + # answers the server's post-handshake CertificateRequest under TLS 1.3. + # Without this, aiohttp certificate/A2A auth fails on TLS 1.3 (SPP 9.0) + # with error 60094. The sync (requests/urllib3) path enables this by + # default, which is why only the async path needed the fix. + ctx.post_handshake_auth = True + if self._min_tls_version is not None: + ctx.minimum_version = self._min_tls_version + if self._max_tls_version is not None: + ctx.maximum_version = self._max_tls_version return ctx async def _get_session(self) -> ClientSession: diff --git a/src/pysafeguard/client.py b/src/pysafeguard/client.py index 85eb47c..808a714 100644 --- a/src/pysafeguard/client.py +++ b/src/pysafeguard/client.py @@ -18,6 +18,7 @@ from __future__ import annotations import json +import ssl import typing from collections.abc import Mapping from pathlib import Path @@ -25,6 +26,7 @@ from typing import IO, TYPE_CHECKING from requests import Response, Session +from requests.adapters import HTTPAdapter from requests.structures import CaseInsensitiveDict from .auth import Auth @@ -39,6 +41,41 @@ DEFAULT_STREAM_CHUNK_SIZE = 8192 +class _TlsVersionAdapter(HTTPAdapter): + """``requests`` adapter that pins the negotiated TLS version. + + Uses urllib3's native ``ssl_minimum_version`` / ``ssl_maximum_version`` + pool options so the existing certificate-verification behavior (CA + bundle handling, per-request client certs, and post-handshake auth, + which urllib3 enables by default) is preserved unchanged. + """ + + def __init__( + self, + min_tls_version: ssl.TLSVersion | None, + max_tls_version: ssl.TLSVersion | None, + ) -> None: + self._min_tls_version = min_tls_version + self._max_tls_version = max_tls_version + super().__init__() + + def _tls_kwargs(self) -> dict[str, ssl.TLSVersion]: + kwargs: dict[str, ssl.TLSVersion] = {} + if self._min_tls_version is not None: + kwargs["ssl_minimum_version"] = self._min_tls_version + if self._max_tls_version is not None: + kwargs["ssl_maximum_version"] = self._max_tls_version + return kwargs + + def init_poolmanager(self, *args: typing.Any, **kwargs: typing.Any) -> None: + kwargs.update(self._tls_kwargs()) + super().init_poolmanager(*args, **kwargs) + + def proxy_manager_for(self, *args: typing.Any, **kwargs: typing.Any) -> typing.Any: + kwargs.update(self._tls_kwargs()) + return super().proxy_manager_for(*args, **kwargs) + + class SafeguardClient: """Synchronous client for the One Identity Safeguard Web API. @@ -51,6 +88,12 @@ class SafeguardClient: :param timeout: Request timeout in seconds (default 300). :param auto_refresh: If ``True``, automatically refresh the token before each request when the token has expired. + :param min_tls_version: Optional minimum TLS version to negotiate (e.g. + ``ssl.TLSVersion.TLSv1_3`` to require TLS 1.3). ``None`` (default) + negotiates normally. + :param max_tls_version: Optional maximum TLS version to negotiate (e.g. + ``ssl.TLSVersion.TLSv1_2`` to cap at TLS 1.2). ``None`` (default) + negotiates normally. """ def __init__( @@ -62,17 +105,25 @@ def __init__( api_version: LiteralString = "v4", timeout: int = DEFAULT_TIMEOUT, auto_refresh: bool = False, + min_tls_version: ssl.TLSVersion | None = None, + max_tls_version: ssl.TLSVersion | None = None, ) -> None: self.host = host self.verify = verify self.api_version = api_version self.auto_refresh = auto_refresh + self._min_tls_version = min_tls_version + self._max_tls_version = max_tls_version self._auth = auth self._user_token: str | None = None self._timeout = timeout self._session = Session() self._session.verify = verify + # Only override transport TLS when a version pin is requested, so the + # default path keeps requests/urllib3's stock behavior untouched. + if min_tls_version is not None or max_tls_version is not None: + self._session.mount("https://", _TlsVersionAdapter(min_tls_version, max_tls_version)) self._headers = CaseInsensitiveDict({"accept": "application/json"}) # -- Properties ---------------------------------------------------------- diff --git a/tests/integration/test_a2a.py b/tests/integration/test_a2a.py index c60fb60..5875765 100644 --- a/tests/integration/test_a2a.py +++ b/tests/integration/test_a2a.py @@ -400,6 +400,43 @@ async def test_async_get_retrievable_accounts(self, a2a_env): assert len(accounts) >= 1 +# =========================================================================== +# A2A over TLS 1.3 (issues #41 / #43) +# =========================================================================== + + +class TestA2ATls13: + """A2A credential retrieval must succeed over an enforced TLS 1.3 handshake. + + Regression guard for the async ``post_handshake_auth`` fix: A2A uses + client-certificate auth, which under TLS 1.3 requires answering a + post-handshake CertificateRequest. + """ + + def test_sync_retrieve_password_over_tls13(self, a2a_env): + with A2AContext( + a2a_env.host, + a2a_env.cert_file, + a2a_env.key_file, + verify=a2a_env.verify, + min_tls_version=ssl.TLSVersion.TLSv1_3, + ) as a2a: + pw = a2a.retrieve_password(a2a_env.api_key) + assert pw.value == a2a_env.original_password + + @pytest.mark.asyncio + async def test_async_retrieve_password_over_tls13(self, a2a_env): + async with AsyncA2AContext( + a2a_env.host, + a2a_env.cert_file, + a2a_env.key_file, + verify=a2a_env.verify, + min_tls_version=ssl.TLSVersion.TLSv1_3, + ) as a2a: + pw = await a2a.retrieve_password(a2a_env.api_key) + assert pw.value == a2a_env.original_password + + # =========================================================================== # A2A event listener lifecycle tests # =========================================================================== diff --git a/tests/integration/test_certificate_auth.py b/tests/integration/test_certificate_auth.py index e0dae54..3ab0e40 100644 --- a/tests/integration/test_certificate_auth.py +++ b/tests/integration/test_certificate_auth.py @@ -241,3 +241,44 @@ async def test_async_context_manager(self, cert_env): assert client.is_authenticated resp = await client.get(Service.CORE, "Me") assert resp.status == 200 + + +# =========================================================================== +# TLS 1.3 enforcement (issues #41 / #43) +# =========================================================================== + + +class TestCertificateAuthTls13: + """Certificate auth must succeed over TLS 1.3 on SPP 9.0. + + Pinning the minimum version to TLS 1.3 forces a TLS 1.3 handshake, which + on cert auth requires the client to answer a post-handshake + CertificateRequest. This is the regression guard for the async + ``post_handshake_auth`` fix; the sync path is included for parity. + """ + + def test_sync_cert_auth_over_tls13(self, cert_env): + with SafeguardClient( + cert_env.host, + auth=CertificateAuth(cert_env.cert_file, cert_env.key_file), + verify=cert_env.verify, + min_tls_version=ssl.TLSVersion.TLSv1_3, + ) as client: + assert client.is_authenticated + resp = client.get(Service.CORE, "Me") + assert resp.status_code == 200 + assert resp.json()["Name"] == "PySg_CertAuthUser" + + @pytest.mark.asyncio + async def test_async_cert_auth_over_tls13(self, cert_env): + async with AsyncSafeguardClient( + cert_env.host, + auth=CertificateAuth(cert_env.cert_file, cert_env.key_file), + verify=cert_env.verify, + min_tls_version=ssl.TLSVersion.TLSv1_3, + ) as client: + assert client.is_authenticated + resp = await client.get(Service.CORE, "Me") + assert resp.status == 200 + me = await resp.json() + assert me["Name"] == "PySg_CertAuthUser" diff --git a/tests/test_tls.py b/tests/test_tls.py new file mode 100644 index 0000000..9b82b3d --- /dev/null +++ b/tests/test_tls.py @@ -0,0 +1,123 @@ +# Copyright (c) One Identity LLC. All rights reserved. +# Licensed under the Apache License, Version 2.0. + +"""Tests for TLS 1.3 support: async post-handshake auth and TLS version pinning. + +Covers the fix from issues #41/#43 (async cert auth on TLS 1.3) and the +opt-in ``min_tls_version`` / ``max_tls_version`` controls on both clients +and the A2A contexts. +""" + +from __future__ import annotations + +import ssl + +from requests.adapters import HTTPAdapter + +from pysafeguard.a2a import A2AContext +from pysafeguard.async_client import AsyncSafeguardClient +from pysafeguard.client import SafeguardClient, _TlsVersionAdapter + + +class TestAsyncSslContext: + def test_post_handshake_auth_enabled(self): + """The async cert-auth context must answer TLS 1.3 post-handshake auth.""" + client = AsyncSafeguardClient("host", verify=False) + ctx = client._create_ssl_context(cert=None) + # verify=False with no cert and no pins still returns False (no context). + assert ctx is False + + secure = AsyncSafeguardClient("host") + ctx = secure._create_ssl_context(cert=None) + assert isinstance(ctx, ssl.SSLContext) + assert ctx.post_handshake_auth is True + + def test_default_no_version_pin(self): + client = AsyncSafeguardClient("host") + ctx = client._create_ssl_context(cert=None) + assert isinstance(ctx, ssl.SSLContext) + # No explicit pin: minimum stays at the library default (TLS 1.2). + assert ctx.maximum_version == ssl.TLSVersion.MAXIMUM_SUPPORTED + + def test_min_tls_version_applied(self): + client = AsyncSafeguardClient("host", min_tls_version=ssl.TLSVersion.TLSv1_3) + ctx = client._create_ssl_context(cert=None) + assert isinstance(ctx, ssl.SSLContext) + assert ctx.minimum_version == ssl.TLSVersion.TLSv1_3 + assert ctx.post_handshake_auth is True + + def test_max_tls_version_applied(self): + client = AsyncSafeguardClient("host", max_tls_version=ssl.TLSVersion.TLSv1_2) + ctx = client._create_ssl_context(cert=None) + assert isinstance(ctx, ssl.SSLContext) + assert ctx.maximum_version == ssl.TLSVersion.TLSv1_2 + + def test_version_pin_forces_context_even_without_verify(self): + """A pin must build a real context even when verify=False and no cert.""" + client = AsyncSafeguardClient("host", verify=False, min_tls_version=ssl.TLSVersion.TLSv1_3) + ctx = client._create_ssl_context(cert=None) + assert isinstance(ctx, ssl.SSLContext) + assert ctx.minimum_version == ssl.TLSVersion.TLSv1_3 + assert ctx.verify_mode == ssl.CERT_NONE + + +class TestSyncTlsAdapter: + def test_default_uses_stock_adapter(self): + client = SafeguardClient("host") + adapter = client._session.get_adapter("https://host") + assert type(adapter) is HTTPAdapter + client.close() + + def test_min_pin_mounts_version_adapter(self): + client = SafeguardClient("host", min_tls_version=ssl.TLSVersion.TLSv1_3) + adapter = client._session.get_adapter("https://host") + assert isinstance(adapter, _TlsVersionAdapter) + assert adapter.poolmanager.connection_pool_kw["ssl_minimum_version"] == ssl.TLSVersion.TLSv1_3 + client.close() + + def test_max_pin_mounts_version_adapter(self): + client = SafeguardClient("host", max_tls_version=ssl.TLSVersion.TLSv1_2) + adapter = client._session.get_adapter("https://host") + assert isinstance(adapter, _TlsVersionAdapter) + assert adapter.poolmanager.connection_pool_kw["ssl_maximum_version"] == ssl.TLSVersion.TLSv1_2 + client.close() + + def test_both_pins_mounted(self): + client = SafeguardClient( + "host", + min_tls_version=ssl.TLSVersion.TLSv1_2, + max_tls_version=ssl.TLSVersion.TLSv1_3, + ) + adapter = client._session.get_adapter("https://host") + assert isinstance(adapter, _TlsVersionAdapter) + kw = adapter.poolmanager.connection_pool_kw + assert kw["ssl_minimum_version"] == ssl.TLSVersion.TLSv1_2 + assert kw["ssl_maximum_version"] == ssl.TLSVersion.TLSv1_3 + client.close() + + +class TestA2ATlsForwarding: + def test_sync_a2a_forwards_pins(self): + ctx = A2AContext( + "host", + "cert.pem", + "key.pem", + verify=False, + min_tls_version=ssl.TLSVersion.TLSv1_3, + ) + assert ctx._conn._min_tls_version == ssl.TLSVersion.TLSv1_3 + adapter = ctx._conn._session.get_adapter("https://host") + assert isinstance(adapter, _TlsVersionAdapter) + ctx.close() + + def test_async_a2a_forwards_pins(self): + from pysafeguard.async_a2a import AsyncA2AContext + + ctx = AsyncA2AContext( + "host", + "cert.pem", + "key.pem", + verify=False, + max_tls_version=ssl.TLSVersion.TLSv1_2, + ) + assert ctx._conn._max_tls_version == ssl.TLSVersion.TLSv1_2