diff --git a/pyproject.toml b/pyproject.toml index c6b0bcde..beceb77f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "sap-cloud-sdk" -version = "0.45.1" +version = "0.46.0" description = "SAP Cloud SDK for Python" readme = "README.md" license = "Apache-2.0" @@ -65,7 +65,6 @@ dev = [ "pytest-bdd>=7.2.0", "python-dotenv>=1.0.0", "ty==0.0.64", - "cryptography>=46.0.3", "ruff==0.16.0", "starlette>=0.40.0", "anyio>=3.6.2", diff --git a/src/sap_cloud_sdk/destination/__init__.py b/src/sap_cloud_sdk/destination/__init__.py index 6073a98e..223d9170 100644 --- a/src/sap_cloud_sdk/destination/__init__.py +++ b/src/sap_cloud_sdk/destination/__init__.py @@ -68,6 +68,7 @@ HttpError, DestinationOperationError, DestinationNotFoundError, + DestinationCertificateError, ) @@ -253,4 +254,5 @@ def create_certificate_client( "HttpError", "DestinationOperationError", "DestinationNotFoundError", + "DestinationCertificateError", ] diff --git a/src/sap_cloud_sdk/destination/_cert_loader.py b/src/sap_cloud_sdk/destination/_cert_loader.py new file mode 100644 index 00000000..3096bce9 --- /dev/null +++ b/src/sap_cloud_sdk/destination/_cert_loader.py @@ -0,0 +1,230 @@ +"""Client-certificate loading for mTLS destinations. + +Parses PEM and PKCS12 keystores from the Destination Service v2 certificate +payload and builds a stdlib ssl.SSLContext for mTLS. + +Supported formats (selected by the file extension of Certificate.name): + pem — combined PEM bundle (cert + optional chain + private key; key may be + encrypted via KeyStorePassword) + p12 — PKCS12 binary keystore (requires KeyStorePassword in practice) + pfx — PKCS12 binary keystore (alternate extension) + +""" + +from __future__ import annotations + +import base64 +import binascii +import os +import ssl +import tempfile +from typing import Optional + +from cryptography.hazmat.primitives.serialization import ( + Encoding, + NoEncryption, + PrivateFormat, +) +from cryptography.hazmat.primitives.serialization import pkcs12 + +from sap_cloud_sdk.destination._models import Authentication, Certificate, Destination +from sap_cloud_sdk.destination.exceptions import DestinationCertificateError + +_SUPPORTED_EXTENSIONS = frozenset({"pem", "p12", "pfx"}) + + +def build_client_cert_context(destination: Destination) -> Optional[ssl.SSLContext]: + """Return an mTLS SSL context for the destination, or None if not applicable. + + Returns None when: + - The destination does not use ClientCertificateAuthentication. + - The certificate list contains no PEM/PKCS12 entry and no KeyStoreLocation is set. + + Raises DestinationCertificateError when client-cert auth is required but no + usable certificate can be loaded (wrong format, malformed content, key mismatch). + """ + if not _is_client_certificate_auth(destination): + return None + + cert = _select_certificate(destination) + if cert is None: + raise DestinationCertificateError( + f"Destination '{destination.name}' uses ClientCertificateAuthentication " + "but no usable certificate is available in the destination's certificate list." + ) + + try: + return _load_cert_into_context(cert, destination) + except DestinationCertificateError: + raise + except Exception as e: + raise DestinationCertificateError( + f"Failed to load client certificate '{cert.name}': {e}" + ) from e + + +def _is_client_certificate_auth(destination: Destination) -> bool: + auth = destination.authentication + auth_value = getattr(auth, "value", auth) + return str(auth_value) == Authentication.CLIENT_CERTIFICATE_AUTHENTICATION.value + + +def _select_certificate(destination: Destination) -> Optional[Certificate]: + certs = destination.certificates + if not certs: + return None + + props = destination.properties or {} + ks_location = props.get("KeyStoreLocation") + + if ks_location: + for cert in certs: + if cert.name == ks_location: + ext = cert.name.rsplit(".", 1)[-1].lower() if "." in cert.name else "" + if ext not in _SUPPORTED_EXTENSIONS: + raise DestinationCertificateError( + f"Certificate '{cert.name}' has unsupported format '.{ext}'. " + f"Supported formats: {sorted(_SUPPORTED_EXTENSIONS)}. " + "JKS is not supported (Java-specific format)." + ) + return cert + return None + + for cert in certs: + ext = cert.name.rsplit(".", 1)[-1].lower() if "." in cert.name else "" + if ext in _SUPPORTED_EXTENSIONS: + return cert + + return None + + +def _load_cert_into_context( + cert: Certificate, destination: Destination +) -> ssl.SSLContext: + ext = cert.name.rsplit(".", 1)[-1].lower() if "." in cert.name else "" + password = _get_key_password(destination) + + if ext == "pem": + return _load_pem(cert.content, password, cert.name) + + if ext in ("p12", "pfx"): + return _load_pkcs12(cert.content, password, cert.name) + + raise DestinationCertificateError( + f"Certificate '{cert.name}' has unsupported format '.{ext}'. " + f"Supported: {sorted(_SUPPORTED_EXTENSIONS)}." + ) + + +def _load_pem(content: str, password: Optional[bytes], name: str) -> ssl.SSLContext: + pem = _decode_pem_bytes(content, name) + return _build_context(pem, password) + + +def _load_pkcs12(content: str, password: Optional[bytes], name: str) -> ssl.SSLContext: + try: + der = base64.b64decode(content) + except (binascii.Error, ValueError) as e: + raise DestinationCertificateError( + f"Certificate '{name}' content is not valid base64: {e}" + ) from e + + try: + private_key, leaf, extra_certs = pkcs12.load_key_and_certificates(der, password) + except Exception as e: + raise DestinationCertificateError( + f"Failed to load PKCS12 certificate '{name}': {e}" + ) from e + + if leaf is None or private_key is None: + raise DestinationCertificateError( + f"PKCS12 certificate '{name}' is missing a certificate or private key." + ) + + # PKCS12 gives us parsed objects (no file), so serialize leaf + chain + an + # unencrypted key into a single PEM bundle. The key is already decrypted by + # load_key_and_certificates, so no password is passed to _build_context. + key_pem = private_key.private_bytes( + encoding=Encoding.PEM, + format=PrivateFormat.PKCS8, + encryption_algorithm=NoEncryption(), + ) + leaf_pem = leaf.public_bytes(Encoding.PEM) + chain_pem = b"".join(c.public_bytes(Encoding.PEM) for c in (extra_certs or [])) + return _build_context(leaf_pem + chain_pem + key_pem, password=None) + + +def _build_context( + bundle_pem: bytes, + password: Optional[bytes], +) -> ssl.SSLContext: + # Write the combined PEM bundle (cert chain + key) to a temp file, + # load it into an SSLContext, then immediately delete. + str_password: Optional[str] = password.decode("utf-8") if password else None + + # Guard against an encrypted key with no password + if str_password is None and any( + marker in bundle_pem + for marker in ( + b"-----BEGIN ENCRYPTED PRIVATE KEY-----", + b"Proc-Type: 4,ENCRYPTED", + ) + ): + raise DestinationCertificateError( + "The private key is encrypted but no KeyStorePassword was provided." + ) + + fd, path = tempfile.mkstemp(suffix=".pem") + try: + with os.fdopen(fd, "wb") as fh: + fh.write(bundle_pem) + ctx = ssl.create_default_context() + ctx.load_cert_chain(path, password=str_password) + except ssl.SSLError as e: + if getattr(e, "reason", None) == "KEY_VALUES_MISMATCH": + raise DestinationCertificateError( + "The certificate and private key do not match." + ) from e + raise DestinationCertificateError( + "Could not load the client certificate/private key (possible causes: " + f"wrong password, malformed PEM, or a missing certificate/key block): {e}" + ) from e + except OSError as e: + raise DestinationCertificateError( + f"Could not load the client certificate/private key: {e}" + ) from e + finally: + os.unlink(path) + + return ctx + + +def _decode_pem_bytes(content: str, name: str) -> bytes: + if not content or not content.strip(): + raise DestinationCertificateError(f"Certificate '{name}' content is empty.") + + pem = content.strip() + + if "-----BEGIN " not in pem: + try: + decoded = base64.b64decode("".join(pem.split())) + except (binascii.Error, ValueError) as e: + raise DestinationCertificateError( + f"Certificate '{name}' content is not valid base64-encoded PEM: {e}" + ) from e + try: + pem = decoded.decode("utf-8") + except UnicodeDecodeError as e: + raise DestinationCertificateError( + f"Certificate '{name}' content is not valid UTF-8 PEM text." + ) from e + + return pem.encode("utf-8") + + +def _get_key_password(destination: Destination) -> Optional[bytes]: + props = destination.properties or {} + password = props.get("KeyStorePassword") + if password and password.strip(): + return password.encode("utf-8") + return None diff --git a/src/sap_cloud_sdk/destination/_destination_http_client.py b/src/sap_cloud_sdk/destination/_destination_http_client.py index 9e948486..f97de1a4 100644 --- a/src/sap_cloud_sdk/destination/_destination_http_client.py +++ b/src/sap_cloud_sdk/destination/_destination_http_client.py @@ -2,25 +2,44 @@ from __future__ import annotations +import ssl from typing import Any, Dict, Optional import requests from requests import Response +from requests.adapters import HTTPAdapter +from sap_cloud_sdk.destination._cert_loader import build_client_cert_context from sap_cloud_sdk.destination._models import Destination, DestinationType +class _ClientCertAdapter(HTTPAdapter): + """requests HTTPAdapter that injects a stdlib SSLContext for mTLS.""" + + def __init__(self, ssl_ctx: ssl.SSLContext, **kwargs: Any) -> None: + self._ssl_ctx = ssl_ctx + super().__init__(**kwargs) + + def init_poolmanager(self, *args: Any, **kwargs: Any) -> None: + kwargs["ssl_context"] = self._ssl_ctx + super().init_poolmanager(*args, **kwargs) + + def proxy_manager_for(self, *args: Any, **kwargs: Any) -> Any: + kwargs["ssl_context"] = self._ssl_ctx + return super().proxy_manager_for(*args, **kwargs) + + class DestinationHttpClient: """Wraps requests.Session to call the target system described by a Destination. Pre-bakes headers derived from the destination — ERP headers (sap-client, - sap-language), URL.headers.* properties, and auth tokens. + sap-language), URL.headers.* properties, and auth tokens. Certificates from the + destination's certificate list are mounted into the session. - Usage: + Use as a context manager to ensure the underlying session is closed: - dest = client.get_destination("my-erp") - http = DestinationHttpClient(dest) - response = http.request("GET", "/api/resource") + with DestinationHttpClient(dest) as http: + response = http.request("GET", "/api/resource") """ def __init__(self, destination: Destination) -> None: @@ -33,6 +52,10 @@ def __init__(self, destination: Destination) -> None: self._session.headers.update(destination.get_headers()) self._base_url = destination.url.rstrip("/") if destination.url else "" + ssl_ctx = build_client_cert_context(destination) + if ssl_ctx is not None: + self._session.mount("https://", _ClientCertAdapter(ssl_ctx)) + def request( self, method: str, @@ -65,3 +88,10 @@ def request( headers=headers, **kwargs, ) + + def __enter__(self) -> "DestinationHttpClient": + return self + + def __exit__(self, *exc: Any) -> bool: + self._session.close() + return False diff --git a/src/sap_cloud_sdk/destination/exceptions.py b/src/sap_cloud_sdk/destination/exceptions.py index 8f3278b5..7d1d36fe 100644 --- a/src/sap_cloud_sdk/destination/exceptions.py +++ b/src/sap_cloud_sdk/destination/exceptions.py @@ -49,3 +49,9 @@ class DestinationNotFoundError(DestinationOperationError): """Raised when a requested Destination is not found (HTTP 404).""" pass + + +class DestinationCertificateError(DestinationError): + """Raised when a client certificate cannot be loaded or wired into the HTTP session.""" + + pass diff --git a/src/sap_cloud_sdk/destination/user-guide.md b/src/sap_cloud_sdk/destination/user-guide.md index 5039d07c..e315cc01 100644 --- a/src/sap_cloud_sdk/destination/user-guide.md +++ b/src/sap_cloud_sdk/destination/user-guide.md @@ -433,6 +433,24 @@ http = DestinationHttpClient(dest) response = http.request("GET", "/api/resource") ``` +### Client-Certificate (mTLS) Authentication + +When a destination's `Authentication` is `ClientCertificateAuthentication`, `DestinationHttpClient` automatically configures the underlying session for mutual TLS. + +```python +from sap_cloud_sdk.destination import create_client, DestinationHttpClient + +client = create_client(instance="default") +dest = client.get_destination("my-mtls-target") + +with DestinationHttpClient(dest) as http: # mTLS is wired automatically + response = http.request("GET", "/api/resource") +``` + +- **`KeyStoreLocation`** destination property: selects a specific certificate by name when multiple are present. +- **`KeyStorePassword`** destination property: used to decrypt an encrypted private key. +- **Supported formats**: PEM (`.pem`) and PKCS12 (`.p12` / `.pfx`). + ### What headers are pre-baked When `DestinationHttpClient` is constructed, it reads the destination and pre-bakes the following headers into every request: @@ -905,6 +923,7 @@ Entries with a `"tenant"` field are treated as subscriber-specific. Entries with - `DestinationNotFoundError`: mapped from HTTP 404 where applicable - `DestinationOperationError`: general operation failures - `HttpError`: HTTP-related or local store read/write errors with `status_code` and `response_text` when applicable +- `DestinationCertificateError`: raised when a client certificate cannot be loaded or wired into the HTTP session (unsupported format, wrong/missing KeyStorePassword, malformed content, cert/key mismatch) ## Configuration diff --git a/tests/destination/integration/destination.feature b/tests/destination/integration/destination.feature index 81a17374..151d3b01 100644 --- a/tests/destination/integration/destination.feature +++ b/tests/destination/integration/destination.feature @@ -253,6 +253,27 @@ Feature: Destination Service Integration And I clean up the instance destination "test-v2-full-options" And I clean up the instance fragment "test-v2-full-fragment" + Scenario: DestinationHttpClient mounts mTLS adapter for PEM certificate with encrypted key + Given I have a subaccount destination with a generated encrypted PEM certificate named "test-mtls-pem" + When I fetch the destination "test-mtls-pem" using the v2 API at subaccount level + Then the DestinationHttpClient mounts a client certificate adapter + And I clean up the subaccount destination "test-mtls-pem" + And I clean up the subaccount certificate "test-mtls-pem.pem" + + Scenario: DestinationHttpClient mounts mTLS adapter for PKCS12 P12 certificate + Given I have a subaccount destination with a generated P12 certificate named "test-mtls-p12" + When I fetch the destination "test-mtls-p12" using the v2 API at subaccount level + Then the DestinationHttpClient mounts a client certificate adapter + And I clean up the subaccount destination "test-mtls-p12" + And I clean up the subaccount certificate "test-mtls-p12.p12" + + Scenario: DestinationHttpClient mounts mTLS adapter for PKCS12 PFX certificate + Given I have a subaccount destination with a generated PFX certificate named "test-mtls-pfx" + When I fetch the destination "test-mtls-pfx" using the v2 API at subaccount level + Then the DestinationHttpClient mounts a client certificate adapter + And I clean up the subaccount destination "test-mtls-pfx" + And I clean up the subaccount certificate "test-mtls-pfx.pfx" + Scenario: DestinationHttpClient sends an authenticated request using token fetched from BTP Given I have a destination named "sdk-test-http-client" of type "HTTP" And the destination has URL "https://httpbin.org" diff --git a/tests/destination/integration/test_destination_bdd.py b/tests/destination/integration/test_destination_bdd.py index 3670c42f..58ceea2e 100644 --- a/tests/destination/integration/test_destination_bdd.py +++ b/tests/destination/integration/test_destination_bdd.py @@ -1589,6 +1589,148 @@ def certificate_should_have_label(context, key, value): ), f"Expected label key='{key}' value='{value}' in {context.retrieved_labels}" +# ==================== MTLS / CLIENT CERTIFICATE STEPS ==================== + +def _generate_encrypted_pem() -> tuple[str, str]: + """Return (base64-encoded combined PEM, password) for an encrypted-key PEM cert.""" + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + from datetime import datetime, timedelta, timezone + import base64 + + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "test-mtls")]) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.now(timezone.utc)) + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=365)) + .sign(key, hashes.SHA256()) + ) + password = "testpassword" + key_pem = key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.BestAvailableEncryption(password.encode()), + ) + cert_pem = cert.public_bytes(serialization.Encoding.PEM) + combined = base64.b64encode(key_pem + cert_pem).decode() + return combined, password + + +def _generate_pkcs12(extension: str) -> tuple[str, str]: + """Return (base64-encoded PKCS12 bytes, password) for a P12/PFX cert.""" + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.hazmat.primitives.serialization import pkcs12 + from cryptography.x509.oid import NameOID + from datetime import datetime, timedelta, timezone + import base64 + + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, f"test-mtls-{extension}")]) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.now(timezone.utc)) + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=365)) + .sign(key, hashes.SHA256()) + ) + password = "testpassword" + p12_bytes = pkcs12.serialize_key_and_certificates( + name=b"test-mtls", + key=key, + cert=cert, + cas=None, + encryption_algorithm=serialization.BestAvailableEncryption(password.encode()), + ) + return base64.b64encode(p12_bytes).decode(), password + + +def _create_mtls_destination_and_cert( + context, + destination_client, + certificate_client, + name: str, + cert_filename: str, + content: str, + password: str, +) -> None: + """Upload a certificate and create a matching ClientCertificateAuthentication destination.""" + cert = Certificate(name=cert_filename, content=content) + certificate_client.create_certificate(cert, level=Level.SUB_ACCOUNT) + context.cleanup_certificates.append((cert_filename, Level.SUB_ACCOUNT, None)) + + dest = Destination.from_dict({ + "Name": name, + "Type": "HTTP", + "URL": "https://httpbin.org", + "Authentication": "ClientCertificateAuthentication", + "KeyStore.Source": "DestinationService", + "KeyStoreLocation": cert_filename, + "KeyStorePassword": password, + }) + destination_client.create_destination(dest, level=Level.SUB_ACCOUNT) + context.cleanup_destinations.append((name, Level.SUB_ACCOUNT, None)) + context.destination = dest + + +@given(parsers.parse('I have a subaccount destination with a generated encrypted PEM certificate named "{name}"')) +def have_mtls_pem_destination(context, destination_client, certificate_client, name): + content, password = _generate_encrypted_pem() + _create_mtls_destination_and_cert( + context, destination_client, certificate_client, + name=name, cert_filename=f"{name}.pem", content=content, password=password, + ) + + +@given(parsers.parse('I have a subaccount destination with a generated P12 certificate named "{name}"')) +def have_mtls_p12_destination(context, destination_client, certificate_client, name): + content, password = _generate_pkcs12("p12") + _create_mtls_destination_and_cert( + context, destination_client, certificate_client, + name=name, cert_filename=f"{name}.p12", content=content, password=password, + ) + + +@given(parsers.parse('I have a subaccount destination with a generated PFX certificate named "{name}"')) +def have_mtls_pfx_destination(context, destination_client, certificate_client, name): + content, password = _generate_pkcs12("pfx") + _create_mtls_destination_and_cert( + context, destination_client, certificate_client, + name=name, cert_filename=f"{name}.pfx", content=content, password=password, + ) + + +@when(parsers.parse('I fetch the destination "{name}" using the v2 API at subaccount level')) +def fetch_destination_v2_subaccount(context, destination_client, name): + from sap_cloud_sdk.destination._models import ConsumptionLevel + context.retrieved_destination = destination_client.get_destination( + name, level=ConsumptionLevel.PROVIDER_SUBACCOUNT + ) + assert context.retrieved_destination is not None, f"Destination '{name}' not found via v2 API" + + +@then("the DestinationHttpClient mounts a client certificate adapter") +def assert_client_cert_adapter_mounted(context): + from sap_cloud_sdk.destination._destination_http_client import _ClientCertAdapter + with DestinationHttpClient(context.retrieved_destination) as http: + adapter = http._session.get_adapter("https://example.com") + assert isinstance(adapter, _ClientCertAdapter), ( + f"Expected _ClientCertAdapter but got {type(adapter).__name__}. " + "The SDK is not applying the client certificate to the HTTP session." + ) + + # ==================== DESTINATION HTTP CLIENT STEPS ==================== @given("the destination has OAuth2 credentials from environment") diff --git a/tests/destination/unit/test_cert_loader.py b/tests/destination/unit/test_cert_loader.py new file mode 100644 index 00000000..1cc3913c --- /dev/null +++ b/tests/destination/unit/test_cert_loader.py @@ -0,0 +1,343 @@ +"""Unit tests for build_client_cert_context (mTLS client-certificate loader).""" + +from __future__ import annotations + +import base64 +import ssl +from datetime import datetime, timedelta, timezone + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.serialization import ( + BestAvailableEncryption, + Encoding, + NoEncryption, + PrivateFormat, +) +from cryptography.hazmat.primitives.serialization import pkcs12 +from cryptography.x509.oid import NameOID + +from sap_cloud_sdk.destination._cert_loader import build_client_cert_context +from sap_cloud_sdk.destination._models import Destination +from sap_cloud_sdk.destination.exceptions import DestinationCertificateError + + +# --------------------------------------------------------------------------- +# Module-scoped key fixtures (RSA keygen is expensive — reuse across tests) +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def rsa_key_a(): + """Generate RSA key A once for the entire module.""" + return rsa.generate_private_key(public_exponent=65537, key_size=2048) + + +@pytest.fixture(scope="module") +def rsa_key_b(): + """Generate RSA key B once for the entire module.""" + return rsa.generate_private_key(public_exponent=65537, key_size=2048) + + +# --------------------------------------------------------------------------- +# Module-level helper functions +# --------------------------------------------------------------------------- + + +def _self_signed(key) -> x509.Certificate: + """Build a minimal self-signed certificate for the given key.""" + subject = issuer = x509.Name( + [x509.NameAttribute(NameOID.COMMON_NAME, "test.example.com")] + ) + return ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.now(timezone.utc)) + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=365)) + .sign(key, hashes.SHA256()) + ) + + +def _pem_bundle(cert, key, password: bytes | None = None) -> str: + """Return a PEM string: cert block + private key block (PKCS8). + + If password is given the key is encrypted with BestAvailableEncryption, + otherwise NoEncryption is used. + """ + enc_alg = BestAvailableEncryption(password) if password else NoEncryption() + key_pem = key.private_bytes( + encoding=Encoding.PEM, + format=PrivateFormat.PKCS8, + encryption_algorithm=enc_alg, + ) + cert_pem = cert.public_bytes(Encoding.PEM) + return (cert_pem + key_pem).decode("utf-8") + + +def _pkcs12_bytes(cert, key, password: bytes | None) -> str: + """Return base64-encoded PKCS12 bytes for the given cert/key pair.""" + enc_alg = BestAvailableEncryption(password) if password else NoEncryption() + der = pkcs12.serialize_key_and_certificates( + name=b"x", + key=key, + cert=cert, + cas=None, + encryption_algorithm=enc_alg, + ) + return base64.b64encode(der).decode("utf-8") + + +def _dest_with_cert( + name: str, + content: str, + *, + ks_location: str | None = None, + ks_password: str | None = None, +) -> Destination: + """Build a ClientCertificateAuthentication destination with one certificate. + + include_runtime_data=True is required or the certificates list is dropped. + """ + d: dict = { + "Name": "d", + "Type": "HTTP", + "URL": "https://example.com", + "Authentication": "ClientCertificateAuthentication", + "certificates": [{"Name": name, "Content": content}], + } + if ks_location is not None: + d["KeyStoreLocation"] = ks_location + if ks_password is not None: + d["KeyStorePassword"] = ks_password + return Destination.from_dict(d, include_runtime_data=True) + + +# --------------------------------------------------------------------------- +# TestSelection — certificate selection logic +# --------------------------------------------------------------------------- + + +class TestSelection: + """Tests for how build_client_cert_context selects (or skips) a certificate.""" + + def test_non_client_cert_auth_returns_none(self): + """Non-ClientCertificateAuthentication destinations return None.""" + dest = Destination.from_dict( + { + "Name": "d", + "Type": "HTTP", + "URL": "https://example.com", + "Authentication": "NoAuthentication", + } + ) + assert build_client_cert_context(dest) is None + + def test_client_cert_auth_empty_certificates_raises(self): + """ClientCertificateAuthentication with no certificates raises.""" + dest = Destination.from_dict( + { + "Name": "d", + "Type": "HTTP", + "URL": "https://example.com", + "Authentication": "ClientCertificateAuthentication", + "certificates": [], + }, + include_runtime_data=True, + ) + with pytest.raises(DestinationCertificateError, match="no usable certificate"): + build_client_cert_context(dest) + + def test_jks_only_cert_raises_unsupported_format(self): + """A JKS-only certificate list raises DestinationCertificateError.""" + dest = Destination.from_dict( + { + "Name": "d", + "Type": "HTTP", + "URL": "https://example.com", + "Authentication": "ClientCertificateAuthentication", + "certificates": [{"Name": "keystore.jks", "Content": "anycontent"}], + }, + include_runtime_data=True, + ) + with pytest.raises(DestinationCertificateError, match="no usable certificate"): + build_client_cert_context(dest) + + def test_ks_location_selects_specific_cert(self, rsa_key_a, rsa_key_b): + """KeyStoreLocation picks the named cert when multiple certs are present.""" + cert_a = _self_signed(rsa_key_a) + cert_b = _self_signed(rsa_key_b) + bundle_a = _pem_bundle(cert_a, rsa_key_a) + bundle_b = _pem_bundle(cert_b, rsa_key_b) + + dest = Destination.from_dict( + { + "Name": "d", + "Type": "HTTP", + "URL": "https://example.com", + "Authentication": "ClientCertificateAuthentication", + "KeyStoreLocation": "cert-b.pem", + "certificates": [ + {"Name": "cert-a.pem", "Content": bundle_a}, + {"Name": "cert-b.pem", "Content": bundle_b}, + ], + }, + include_runtime_data=True, + ) + ctx = build_client_cert_context(dest) + assert isinstance(ctx, ssl.SSLContext) + + +# --------------------------------------------------------------------------- +# TestPemHappy — successful PEM loading +# --------------------------------------------------------------------------- + + +class TestPemHappy: + """Tests for successful PEM keystore loading paths.""" + + def test_unencrypted_pem_returns_ssl_context(self, rsa_key_a): + """An unencrypted PEM bundle returns an SSLContext with secure defaults.""" + cert = _self_signed(rsa_key_a) + bundle = _pem_bundle(cert, rsa_key_a) + dest = _dest_with_cert("client.pem", bundle) + + ctx = build_client_cert_context(dest) + + assert isinstance(ctx, ssl.SSLContext) + assert ctx.verify_mode == ssl.CERT_REQUIRED + assert ctx.check_hostname is True + + def test_encrypted_pem_correct_password_returns_ssl_context(self, rsa_key_a): + """An encrypted PEM key with the correct KeyStorePassword returns an SSLContext.""" + cert = _self_signed(rsa_key_a) + bundle = _pem_bundle(cert, rsa_key_a, password=b"s3cr3t") + dest = _dest_with_cert("client.pem", bundle, ks_password="s3cr3t") + + ctx = build_client_cert_context(dest) + + assert isinstance(ctx, ssl.SSLContext) + + def test_base64_wrapped_pem_is_decoded(self, rsa_key_a): + """A base64-encoded PEM bundle (no BEGIN header visible) is decoded transparently.""" + cert = _self_signed(rsa_key_a) + bundle = _pem_bundle(cert, rsa_key_a) + # Wrap the whole PEM string in base64 — exercises _decode_pem_bytes + b64_content = base64.b64encode(bundle.encode()).decode("utf-8") + dest = _dest_with_cert("client.pem", b64_content) + + ctx = build_client_cert_context(dest) + + assert isinstance(ctx, ssl.SSLContext) + + def test_chain_cert_accepted(self, rsa_key_a, rsa_key_b): + """A PEM bundle with leaf + intermediate cert + key is accepted.""" + leaf_cert = _self_signed(rsa_key_a) + intermediate_cert = _self_signed(rsa_key_b) # acts as chain material + + leaf_pem = leaf_cert.public_bytes(Encoding.PEM) + intermediate_pem = intermediate_cert.public_bytes(Encoding.PEM) + key_pem = rsa_key_a.private_bytes( + encoding=Encoding.PEM, + format=PrivateFormat.PKCS8, + encryption_algorithm=NoEncryption(), + ) + # leaf + chain cert + key — matches the pattern ssl.load_cert_chain expects + bundle = (leaf_pem + intermediate_pem + key_pem).decode("utf-8") + dest = _dest_with_cert("client.pem", bundle) + + ctx = build_client_cert_context(dest) + + assert isinstance(ctx, ssl.SSLContext) + + +# --------------------------------------------------------------------------- +# TestPkcs12Happy — successful PKCS12 loading +# --------------------------------------------------------------------------- + + +class TestPkcs12Happy: + """Tests for successful PKCS12 keystore loading paths.""" + + def test_p12_with_password_returns_ssl_context(self, rsa_key_a): + """A PKCS12 (.p12) keystore with a password loads successfully.""" + cert = _self_signed(rsa_key_a) + p12_b64 = _pkcs12_bytes(cert, rsa_key_a, password=b"p12pass") + dest = _dest_with_cert("client.p12", p12_b64, ks_password="p12pass") + + ctx = build_client_cert_context(dest) + + assert isinstance(ctx, ssl.SSLContext) + + def test_pfx_extension_also_works(self, rsa_key_a): + """A PKCS12 keystore with .pfx extension is handled identically to .p12.""" + cert = _self_signed(rsa_key_a) + p12_b64 = _pkcs12_bytes(cert, rsa_key_a, password=b"pfxpass") + dest = _dest_with_cert("client.pfx", p12_b64, ks_password="pfxpass") + + ctx = build_client_cert_context(dest) + + assert isinstance(ctx, ssl.SSLContext) + + +# --------------------------------------------------------------------------- +# TestFailures — error paths +# --------------------------------------------------------------------------- + + +class TestFailures: + """Tests for DestinationCertificateError error paths.""" + + def test_malformed_pem_raises(self): + """Content that is neither valid PEM nor valid base64 raises DestinationCertificateError.""" + dest = _dest_with_cert("client.pem", "not-a-cert!!!") + with pytest.raises(DestinationCertificateError): + build_client_cert_context(dest) + + def test_encrypted_key_wrong_password_raises(self, rsa_key_a): + """An encrypted PEM key with the wrong password raises DestinationCertificateError.""" + cert = _self_signed(rsa_key_a) + bundle = _pem_bundle(cert, rsa_key_a, password=b"correct") + dest = _dest_with_cert("client.pem", bundle, ks_password="wrong") + + with pytest.raises(DestinationCertificateError): + build_client_cert_context(dest) + + def test_encrypted_key_missing_password_raises_and_does_not_hang(self, rsa_key_a): + """An encrypted PEM key with no KeyStorePassword raises immediately (no interactive prompt).""" + cert = _self_signed(rsa_key_a) + bundle = _pem_bundle(cert, rsa_key_a, password=b"somepass") + # No ks_password — the loader guards against the interactive-prompt footgun + dest = _dest_with_cert("client.pem", bundle) + + with pytest.raises(DestinationCertificateError, match="KeyStorePassword"): + build_client_cert_context(dest) + + def test_cert_key_mismatch_raises(self, rsa_key_a, rsa_key_b): + """A bundle where cert and private key belong to different keys raises DestinationCertificateError.""" + cert_a = _self_signed(rsa_key_a) + # cert signed by key_a, but private key is key_b — they don't match + cert_pem = cert_a.public_bytes(Encoding.PEM) + key_b_pem = rsa_key_b.private_bytes( + encoding=Encoding.PEM, + format=PrivateFormat.PKCS8, + encryption_algorithm=NoEncryption(), + ) + bundle = (cert_pem + key_b_pem).decode("utf-8") + dest = _dest_with_cert("client.pem", bundle) + + with pytest.raises(DestinationCertificateError, match="do not match"): + build_client_cert_context(dest) + + def test_cert_only_no_key_raises(self, rsa_key_a): + """A PEM bundle with only a cert block (no private key) raises DestinationCertificateError.""" + cert = _self_signed(rsa_key_a) + cert_only = cert.public_bytes(Encoding.PEM).decode("utf-8") + dest = _dest_with_cert("client.pem", cert_only) + + with pytest.raises(DestinationCertificateError): + build_client_cert_context(dest) diff --git a/tests/destination/unit/test_destination_http_client.py b/tests/destination/unit/test_destination_http_client.py index e0d8ea84..529e21e4 100644 --- a/tests/destination/unit/test_destination_http_client.py +++ b/tests/destination/unit/test_destination_http_client.py @@ -1,11 +1,16 @@ """Unit tests for DestinationHttpClient.""" +import ssl from unittest.mock import MagicMock, patch import pytest -from sap_cloud_sdk.destination._destination_http_client import DestinationHttpClient +from sap_cloud_sdk.destination._destination_http_client import ( + _ClientCertAdapter, + DestinationHttpClient, +) from sap_cloud_sdk.destination._models import AuthToken, Destination +from sap_cloud_sdk.destination.exceptions import DestinationCertificateError def _dest(**kwargs) -> Destination: @@ -15,7 +20,9 @@ def _dest(**kwargs) -> Destination: def _auth_token(key: str, value: str) -> AuthToken: - return AuthToken(type="Bearer", value="raw", http_header={"key": key, "value": value}) + return AuthToken( + type="Bearer", value="raw", http_header={"key": key, "value": value} + ) class TestDestinationHttpClientInit: @@ -77,30 +84,71 @@ def setup_method(self): self.mock_response = MagicMock() def test_constructs_full_url(self): - with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req: + with patch.object( + self.client._session, "request", return_value=self.mock_response + ) as mock_req: self.client.request("GET", "/api/v1/users") assert mock_req.call_args[1]["url"] == "https://example.com/api/v1/users" def test_uppercases_method(self): - with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req: + with patch.object( + self.client._session, "request", return_value=self.mock_response + ) as mock_req: self.client.request("get", "/resource") assert mock_req.call_args[1]["method"] == "GET" def test_passes_params(self): - with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req: + with patch.object( + self.client._session, "request", return_value=self.mock_response + ) as mock_req: self.client.request("GET", "/resource", params={"$top": "10"}) assert mock_req.call_args[1]["params"] == {"$top": "10"} def test_passes_json_body(self): - with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req: + with patch.object( + self.client._session, "request", return_value=self.mock_response + ) as mock_req: self.client.request("POST", "/resource", json={"key": "value"}) assert mock_req.call_args[1]["json"] == {"key": "value"} def test_passes_extra_headers(self): - with patch.object(self.client._session, "request", return_value=self.mock_response) as mock_req: + with patch.object( + self.client._session, "request", return_value=self.mock_response + ) as mock_req: self.client.request("GET", "/resource", headers={"X-Custom": "yes"}) assert mock_req.call_args[1]["headers"] == {"X-Custom": "yes"} def test_returns_response(self): - with patch.object(self.client._session, "request", return_value=self.mock_response): + with patch.object( + self.client._session, "request", return_value=self.mock_response + ): assert self.client.request("GET", "/resource") is self.mock_response + + +class TestDestinationHttpClientCert: + """Tests that DestinationHttpClient wires the mTLS cert adapter correctly.""" + + _PATCH_TARGET = ( + "sap_cloud_sdk.destination._destination_http_client.build_client_cert_context" + ) + + def test_ssl_context_mounts_client_cert_adapter(self): + """When build_client_cert_context returns a context, https:// uses _ClientCertAdapter.""" + ssl_ctx = ssl.create_default_context() + with patch(self._PATCH_TARGET, return_value=ssl_ctx): + client = DestinationHttpClient(_dest()) + assert isinstance(client._session.get_adapter("https://"), _ClientCertAdapter) + + def test_none_context_does_not_mount_client_cert_adapter(self): + """When build_client_cert_context returns None, https:// uses the default requests adapter.""" + with patch(self._PATCH_TARGET, return_value=None): + client = DestinationHttpClient(_dest()) + assert not isinstance( + client._session.get_adapter("https://"), _ClientCertAdapter + ) + + def test_cert_load_error_propagates(self): + """When build_client_cert_context raises DestinationCertificateError, the constructor propagates it.""" + with patch(self._PATCH_TARGET, side_effect=DestinationCertificateError("boom")): + with pytest.raises(DestinationCertificateError): + DestinationHttpClient(_dest()) diff --git a/uv.lock b/uv.lock index 2c98946e..cc69c3ff 100644 --- a/uv.lock +++ b/uv.lock @@ -161,9 +161,9 @@ name = "aiologic" version = "0.16.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "sniffio", marker = "python_full_version < '3.13'" }, - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, - { name = "wrapt", marker = "python_full_version < '3.13'" }, + { name = "sniffio" }, + { name = "typing-extensions" }, + { name = "wrapt" }, ] sdist = { url = "https://files.pythonhosted.org/packages/a8/13/50b91a3ea6b030d280d2654be97c48b6ed81753a50286ee43c646ba36d3c/aiologic-0.16.0.tar.gz", hash = "sha256:c267ccbd3ff417ec93e78d28d4d577ccca115d5797cdbd16785a551d9658858f", size = 225952, upload-time = "2025-11-27T23:48:41.195Z" } wheels = [ @@ -615,8 +615,8 @@ name = "culsans" version = "0.11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "aiologic", marker = "python_full_version < '3.13'" }, - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "aiologic" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/d9/e3/49afa1bc180e0d28008ec6bcdf82a4072d1c7a41032b5b759b60814ca4b0/culsans-0.11.0.tar.gz", hash = "sha256:0b43d0d05dce6106293d114c86e3fb4bfc63088cfe8ff08ed3fe36891447fe33", size = 107546, upload-time = "2025-12-31T23:15:38.196Z" } wheels = [ @@ -665,9 +665,9 @@ resolution-markers = [ "python_full_version < '3.12'", ] dependencies = [ - { name = "asgiref", marker = "python_full_version < '3.12'" }, - { name = "sqlparse", marker = "python_full_version < '3.12'" }, - { name = "tzdata", marker = "python_full_version < '3.12' and sys_platform == 'win32'" }, + { name = "asgiref" }, + { name = "sqlparse" }, + { name = "tzdata", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/a9/26/889449d521ae508b26de715954faecd8bcf3f740affb81b2d146a83b42a5/django-5.2.16.tar.gz", hash = "sha256:59ea02020c3136fce14bef0bbece21a10a4febef5eed1c51c22ae468efa22200", size = 10890894, upload-time = "2026-07-07T13:52:17.005Z" } wheels = [ @@ -685,9 +685,9 @@ resolution-markers = [ "python_full_version == '3.12.*'", ] dependencies = [ - { name = "asgiref", marker = "python_full_version >= '3.12'" }, - { name = "sqlparse", marker = "python_full_version >= '3.12'" }, - { name = "tzdata", marker = "python_full_version >= '3.12' and sys_platform == 'win32'" }, + { name = "asgiref" }, + { name = "sqlparse" }, + { name = "tzdata", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/89/55/664f24ff81c9ea19cb7dfc851afeae1f3c2390c7aee01d4ded68b5c1580d/django-6.0.7.tar.gz", hash = "sha256:2998503fc083124fb58037084bfa00de323c7c743f05f1b4284e77bff0ab8890", size = 10921299, upload-time = "2026-07-07T13:51:26.485Z" } wheels = [ @@ -3925,7 +3925,7 @@ wheels = [ [[package]] name = "sap-cloud-sdk" -version = "0.45.0" +version = "0.46.0" source = { editable = "." } dependencies = [ { name = "cryptography" }, @@ -3991,7 +3991,6 @@ dev = [ { name = "a2a-sdk" }, { name = "aiohttp" }, { name = "anyio" }, - { name = "cryptography" }, { name = "django", version = "5.2.16", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" }, { name = "django", version = "6.0.7", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, { name = "fastapi" }, @@ -4060,7 +4059,6 @@ dev = [ { name = "a2a-sdk", specifier = ">=0.2.0" }, { name = "aiohttp", specifier = ">=3.9.0" }, { name = "anyio", specifier = ">=3.6.2" }, - { name = "cryptography", specifier = ">=46.0.3" }, { name = "django", specifier = ">=4.0" }, { name = "fastapi", specifier = ">=0.100.0" }, { name = "flask", specifier = ">=3.0" },