From 77d9615b07afb5ca2eae67080eb5b77f6275168f Mon Sep 17 00:00:00 2001 From: Tiago Kochenborger Date: Fri, 2 Oct 2026 17:15:44 -0300 Subject: [PATCH] fix(ias): verify JWT signature before stamping telemetry and audit identity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StarletteIASTelemetryMiddleware was calling parse_token() (unverified JWT decoder) and promoting sap_gtid/user_uuid onto OTel span attributes and the IAS auth context. An attacker could forge a JWT carrying a victim tenant/user ID and permanently pollute telemetry and AuditClient attribution. - Add IASVerifier: JWKS-backed verifier with PyJWKClient, RS256/ES256 alg pinning, issuer/audience/exp/nbf enforcement, key caching and rotation - Add IASVerifier.from_env(): auto-configures from VCAP_SERVICES (CF) or IAS_URL/IAS_CLIENT_ID env vars (K8s); raises IASConfigError if not found - Add VerifiedIASClaims frozen dataclass as provenance marker; add TokenVerifier type alias to sap_cloud_sdk.ias public API - StarletteIASTelemetryMiddleware auto-calls IASVerifier.from_env() at init; logs WARNING and disables identity attrs if no binding found (fail-closed) - _verify_and_extract calls verifier once per request; verified IASClaims flow to both set_auth_context (AuditClient) and OTel span attrs — forged tokens result in None auth context and empty identity attrs - parse_token() unchanged — available as unverified diagnostic decoder - 488 tests pass (22 new for IASVerifier, rewritten middleware tests with auto-config and auth-context coverage) --- .../telemetry/middleware/starlette_a2a.py | 126 +++++-- src/sap_cloud_sdk/ias/__init__.py | 8 +- src/sap_cloud_sdk/ias/_token.py | 20 +- src/sap_cloud_sdk/ias/_verifier.py | 153 ++++++++ .../middleware/test_starlette_a2a.py | 340 ++++++++++++------ tests/ias/unit/test_verifier.py | 203 +++++++++++ 6 files changed, 705 insertions(+), 145 deletions(-) create mode 100644 src/sap_cloud_sdk/ias/_verifier.py create mode 100644 tests/ias/unit/test_verifier.py diff --git a/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py b/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py index 3bbde413..b9f78305 100644 --- a/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py +++ b/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py @@ -2,7 +2,7 @@ import logging from contextvars import ContextVar -from typing import Any, Dict +from typing import Any, Dict, Optional, Tuple from sap_cloud_sdk.core.telemetry.constants import ( ATTR_SAP_TRIGGER_TYPE, @@ -10,7 +10,9 @@ ATTR_USER_ID, ) from sap_cloud_sdk.core.telemetry.middleware.base import TelemetryMiddleware -from sap_cloud_sdk.ias import parse_token, set_auth_context +from sap_cloud_sdk.ias import IASConfigError, IASVerifier, TokenVerifier, VerifiedIASClaims # noqa: F401 +from sap_cloud_sdk.ias import set_auth_context +from sap_cloud_sdk.ias._token import IASClaims try: from starlette.middleware.base import BaseHTTPMiddleware @@ -26,13 +28,19 @@ class _IASMiddleware(BaseHTTPMiddleware): - def __init__(self, app: Any, attrs_var: ContextVar[Dict[str, Any]]) -> None: + def __init__( + self, + app: Any, + attrs_var: ContextVar[Dict[str, Any]], + token_verifier: Optional[TokenVerifier], + ) -> None: super().__init__(app) self._attrs_var = attrs_var + self._token_verifier = token_verifier async def dispatch(self, request: Request, call_next: Any) -> Response: - claims, attrs = _parse_request(request) - set_auth_context(claims) + verified_claims, attrs = _verify_and_extract(request, self._token_verifier) + set_auth_context(verified_claims) token = self._attrs_var.set(attrs) try: return await call_next(request) @@ -41,18 +49,37 @@ async def dispatch(self, request: Request, call_next: Any) -> Response: class StarletteIASTelemetryMiddleware(TelemetryMiddleware): - """Starlette/FastAPI middleware that extracts IAS JWT claims as telemetry attributes. + """Starlette/FastAPI middleware that extracts verified IAS JWT claims as telemetry attributes. - Reads the ``Authorization: Bearer `` header on each request, - parses it as an IAS JWT, and exposes the following as span attributes: + Reads the ``Authorization: Bearer `` header on each request, verifies it using + a :class:`~sap_cloud_sdk.ias.IASVerifier`, and exposes the following as span attributes + on success: - ``sap.tenancy.tenant_id`` from the ``sap_gtid`` claim - ``user.id`` from the ``user_uuid`` claim - If the header is absent or the token cannot be parsed, no attributes are set - and the request continues normally. + The ``x-sap-origin`` header (trigger type, not JWT identity) is always stamped when + present, regardless of token verification outcome. - Each instance owns its own ContextVar to prevent cross-talk when multiple - middleware instances are registered on the same app. + Verified claims are also stored in the IAS auth context (see + :func:`~sap_cloud_sdk.ias.get_auth_context`) for downstream use by the AuditClient + auto-fill. When verification fails, the auth context is set to ``None`` so downstream + consumers see no identity rather than unverified claims. + + **Auto-configuration (recommended):** when no ``token_verifier`` is supplied, the + middleware automatically creates an :class:`~sap_cloud_sdk.ias.IASVerifier` from the + SAP BTP Identity service binding (``VCAP_SERVICES`` on CF, or ``IAS_URL`` env var on + Kubernetes). If the binding is not found, identity attributes are disabled and a + WARNING is logged — the app still starts normally. + + Each instance owns its own ContextVar to prevent cross-talk when multiple middleware + instances are registered on the same app. + + Args: + app: The Starlette/FastAPI application instance. + token_verifier: Optional. A callable that receives the raw ``Authorization`` header + value and returns :class:`~sap_cloud_sdk.ias.VerifiedIASClaims` on success, or + raises on any invalid token. When ``None`` (default), an + :class:`~sap_cloud_sdk.ias.IASVerifier` is auto-configured from the environment. Usage:: @@ -61,18 +88,26 @@ class StarletteIASTelemetryMiddleware(TelemetryMiddleware): from sap_cloud_sdk.core.telemetry.middleware import StarletteIASTelemetryMiddleware app = Starlette(...) + # Auto-configures from IAS service binding — no extra config needed auto_instrument(middlewares=[StarletteIASTelemetryMiddleware(app=app)]) """ - def __init__(self, app: Any) -> None: + def __init__(self, app: Any, token_verifier: Optional[TokenVerifier] = None) -> None: self.app = app + if token_verifier is None: + token_verifier = _auto_configure_verifier() + self._token_verifier = token_verifier self._attrs_var: ContextVar[Dict[str, Any]] = ContextVar( f"ias_attrs_{id(self)}", default={} ) def register(self) -> None: """Register the IAS JWT middleware with ``self.app``.""" - self.app.add_middleware(_IASMiddleware, attrs_var=self._attrs_var) + self.app.add_middleware( + _IASMiddleware, + attrs_var=self._attrs_var, + token_verifier=self._token_verifier, + ) logger.info("Registered IAS telemetry middleware on %r", self.app) def get_attributes(self) -> Dict[str, Any]: @@ -80,22 +115,57 @@ def get_attributes(self) -> Dict[str, Any]: return self._attrs_var.get() -def _parse_request(request: Request): - """Parse the Authorization header and return (IASClaims, telemetry_attrs).""" - auth = request.headers.get("authorization", "") - claims = None +def _verify_and_extract( + request: Request, token_verifier: Optional[TokenVerifier] +) -> Tuple[Optional[IASClaims], Dict[str, Any]]: + """Verify the request token once; return (verified_claims, telemetry_attrs). + + ``x-sap-origin`` is always stamped when present — it is a plain request + header, not JWT identity data, so it is independent of verification. + + Identity attrs and the returned claims are None/empty when: + - no Authorization header is present + - token_verifier is None (fail-closed default) + - the verifier raises for any reason (bad sig, wrong iss, expired, etc.) + """ attrs: Dict[str, Any] = {} - if auth: - try: - claims = parse_token(auth) - except Exception as e: - logger.debug("IAS token parsing failed, skipping telemetry attrs: %s", e) - if claims is not None: - if claims.sap_gtid: - attrs[ATTR_SAP_TENANT_ID] = claims.sap_gtid - if claims.user_uuid: - attrs[ATTR_USER_ID] = claims.user_uuid + origin = request.headers.get("x-sap-origin") if origin: attrs[ATTR_SAP_TRIGGER_TYPE] = origin + + auth = request.headers.get("authorization", "") + if not auth or token_verifier is None: + return None, attrs + + try: + verified = token_verifier(auth) + except Exception as exc: + logger.debug("IAS token verification failed, skipping identity attrs: %s", exc) + return None, attrs + + claims = verified.claims + if claims.sap_gtid: + attrs[ATTR_SAP_TENANT_ID] = claims.sap_gtid + if claims.user_uuid: + attrs[ATTR_USER_ID] = claims.user_uuid return claims, attrs + + +def _auto_configure_verifier() -> Optional[TokenVerifier]: + """Try to build an IASVerifier from the environment; warn and return None if not possible.""" + try: + verifier = IASVerifier.from_env() + logger.debug( + "StarletteIASTelemetryMiddleware: auto-configured IASVerifier from environment" + ) + return verifier + except IASConfigError as exc: + logger.warning( + "StarletteIASTelemetryMiddleware: IAS service binding not found — " + "sap.tenancy.tenant_id and user.id will NOT be stamped on spans. " + "Bind an SAP Identity service instance or set IAS_URL to enable " + "identity attributes. Details: %s", + exc, + ) + return None diff --git a/src/sap_cloud_sdk/ias/__init__.py b/src/sap_cloud_sdk/ias/__init__.py index c2459702..369f0df0 100644 --- a/src/sap_cloud_sdk/ias/__init__.py +++ b/src/sap_cloud_sdk/ias/__init__.py @@ -13,13 +13,19 @@ """ from sap_cloud_sdk.ias._context import get_auth_context, set_auth_context -from sap_cloud_sdk.ias._token import IASClaims, parse_token +from sap_cloud_sdk.ias._token import IASClaims, TokenVerifier, VerifiedIASClaims, parse_token +from sap_cloud_sdk.ias._verifier import IASConfigError, IASVerifier from sap_cloud_sdk.ias.exceptions import IASTokenError __all__ = [ "IASClaims", + "IASConfigError", "IASTokenError", + "IASVerifier", + "TokenVerifier", + "VerifiedIASClaims", "get_auth_context", "parse_token", "set_auth_context", ] + diff --git a/src/sap_cloud_sdk/ias/_token.py b/src/sap_cloud_sdk/ias/_token.py index c2ebc71c..2bef95e3 100644 --- a/src/sap_cloud_sdk/ias/_token.py +++ b/src/sap_cloud_sdk/ias/_token.py @@ -7,7 +7,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional, Union +from typing import Any, Callable, Dict, List, Optional, Union import jwt @@ -172,3 +172,21 @@ def parse_token(token: str) -> IASClaims: k: v for k, v in payload.items() if k not in _KNOWN_CLAIM_VALUES }, ) + + +@dataclass(frozen=True) +class VerifiedIASClaims: + """Claims proven to originate from a successfully verified IAS JWT. + + Construct ONLY after verifying the token's signature, issuer, audience, + algorithm, and time constraints. Its presence is the SDK's provenance + marker for security-sensitive consumers such as the telemetry middleware + and AuditClient auto-fill. + """ + + claims: IASClaims + + +# A verifier receives the raw Authorization header value (may include the +# "Bearer " prefix) and MUST raise (fail closed) if the token is not valid. +TokenVerifier = Callable[[str], VerifiedIASClaims] diff --git a/src/sap_cloud_sdk/ias/_verifier.py b/src/sap_cloud_sdk/ias/_verifier.py new file mode 100644 index 00000000..54a051ab --- /dev/null +++ b/src/sap_cloud_sdk/ias/_verifier.py @@ -0,0 +1,153 @@ +"""Built-in JWKS-backed IAS JWT verifier.""" + +import json +import logging +import os +from typing import Optional + +import jwt +from jwt import PyJWKClient + +from sap_cloud_sdk.ias._token import VerifiedIASClaims, parse_token +from sap_cloud_sdk.ias.exceptions import IASTokenError + +logger = logging.getLogger(__name__) + +_ENV_IAS_URL = "IAS_URL" +_ENV_IAS_CLIENT_ID = "IAS_CLIENT_ID" + + +class IASConfigError(Exception): + """Raised when IAS configuration cannot be resolved from the environment.""" + + +class IASVerifier: + """JWKS-backed IAS JWT verifier. + + Verifies the JWT signature using the IAS JWKS endpoint and validates + issuer, expiration, and not-before constraints. Optionally validates + the audience (``aud`` claim) against the application's client ID. + + Designed to be instantiated once at application startup and shared across + requests. ``PyJWKClient`` caches keys internally and handles key rotation + transparently. + + Args: + ias_url: IAS tenant base URL, e.g. ``https://.accounts.ondemand.com``. + The JWKS endpoint is derived as ``{ias_url}/oauth2/certs``. + client_id: Expected ``aud`` claim (the application's client ID in IAS). + When provided, tokens issued for other applications are rejected. + When ``None``, audience validation is skipped. + + Usage:: + + from sap_cloud_sdk.ias import IASVerifier + + # Auto-configure from the IAS service binding (recommended) + verifier = IASVerifier.from_env() + + # Or configure explicitly + verifier = IASVerifier( + ias_url="https://mytenant.accounts.ondemand.com", + client_id="my-app-client-id", + ) + + # Use as a TokenVerifier callable + verified = verifier("Bearer ") + print(verified.claims.sap_gtid) + """ + + def __init__(self, ias_url: str, client_id: Optional[str] = None) -> None: + self._ias_url = ias_url.rstrip("/") + self._client_id = client_id + jwks_url = f"{self._ias_url}/oauth2/certs" + self._jwk_client = PyJWKClient(jwks_url, cache_keys=True) + logger.debug( + "IASVerifier initialised (jwks=%s, client_id=%s)", + jwks_url, + client_id or "", + ) + + @classmethod + def from_env(cls) -> "IASVerifier": + """Auto-configure from the SAP BTP Identity service binding. + + Lookup order: + + 1. ``VCAP_SERVICES`` (Cloud Foundry) — + ``identity[0].credentials.{url, clientid}`` + 2. ``IAS_URL`` + ``IAS_CLIENT_ID`` environment variables (Kubernetes / manual) + + Returns: + A configured :class:`IASVerifier` instance. + + Raises: + IASConfigError: when no IAS configuration can be resolved. + """ + vcap_raw = os.getenv("VCAP_SERVICES") + if vcap_raw: + try: + vcap = json.loads(vcap_raw) + for svc_name in ("identity", "xsuaa"): + bindings = vcap.get(svc_name, []) + if bindings: + creds = bindings[0].get("credentials", {}) + url = creds.get("url") or creds.get("issuer") + client_id = creds.get("clientid") + if url: + logger.debug( + "IASVerifier.from_env: configured from VCAP_SERVICES[%s]", + svc_name, + ) + return cls(ias_url=url, client_id=client_id or None) + except (json.JSONDecodeError, KeyError, IndexError, TypeError) as exc: + logger.debug("IASVerifier.from_env: VCAP_SERVICES parse error: %s", exc) + + ias_url = os.getenv(_ENV_IAS_URL) + if ias_url: + client_id = os.getenv(_ENV_IAS_CLIENT_ID) or None + logger.debug("IASVerifier.from_env: configured from env vars") + return cls(ias_url=ias_url, client_id=client_id) + + raise IASConfigError( + f"Cannot auto-configure IASVerifier: no IAS service binding found. " + f"Bind an SAP Identity service instance (sets VCAP_SERVICES on CF or " + f"a Kubernetes secret) or set {_ENV_IAS_URL} (and optionally " + f"{_ENV_IAS_CLIENT_ID}) manually." + ) + + def __call__(self, authorization: str) -> VerifiedIASClaims: + """Verify the token and return its claims. + + Args: + authorization: Raw ``Authorization`` header value. + Accepts ``"Bearer "`` or a bare token string. + + Returns: + :class:`~sap_cloud_sdk.ias.VerifiedIASClaims` on success. + + Raises: + IASTokenError: if the token fails any validation check. + """ + raw = authorization.removeprefix("Bearer ").removeprefix("bearer ").strip() + try: + signing_key = self._jwk_client.get_signing_key_from_jwt(raw) + + options: dict = {"require": ["exp", "iss"]} + decode_kwargs: dict = { + "algorithms": ["RS256", "ES256"], + "issuer": self._ias_url, + "options": options, + } + if self._client_id: + decode_kwargs["audience"] = self._client_id + options["require"].append("aud") + else: + options["verify_aud"] = False + + jwt.decode(raw, signing_key.key, **decode_kwargs) + + except jwt.exceptions.PyJWTError as exc: + raise IASTokenError(f"IAS JWT verification failed: {exc}") from exc + + return VerifiedIASClaims(claims=parse_token(raw)) diff --git a/tests/core/unit/telemetry/middleware/test_starlette_a2a.py b/tests/core/unit/telemetry/middleware/test_starlette_a2a.py index 871798d9..bb5d8869 100644 --- a/tests/core/unit/telemetry/middleware/test_starlette_a2a.py +++ b/tests/core/unit/telemetry/middleware/test_starlette_a2a.py @@ -1,22 +1,24 @@ """Tests for StarletteIASTelemetryMiddleware.""" +import logging import pytest from unittest.mock import MagicMock, AsyncMock, patch from sap_cloud_sdk.core.telemetry.constants import ATTR_SAP_TRIGGER_TYPE, ATTR_SAP_TENANT_ID, ATTR_USER_ID from sap_cloud_sdk.core.telemetry.middleware.starlette_a2a import ( StarletteIASTelemetryMiddleware, - _parse_request, + _auto_configure_verifier, + _verify_and_extract, ) +from sap_cloud_sdk.ias import IASClaims, IASConfigError, IASTokenError, IASVerifier, VerifiedIASClaims -_PATCH_PARSE = "sap_cloud_sdk.core.telemetry.middleware.starlette_a2a.parse_token" +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- -def _make_claims(sap_gtid=None, user_uuid=None): - claims = MagicMock() - claims.sap_gtid = sap_gtid - claims.user_uuid = user_uuid - return claims +def _make_verified(sap_gtid=None, user_uuid=None) -> VerifiedIASClaims: + return VerifiedIASClaims(claims=IASClaims(sap_gtid=sap_gtid, user_uuid=user_uuid)) def _make_request(headers: dict): @@ -25,26 +27,74 @@ def _make_request(headers: dict): return request +def _passing_verifier(sap_gtid=None, user_uuid=None): + def verify(token: str) -> VerifiedIASClaims: + return _make_verified(sap_gtid=sap_gtid, user_uuid=user_uuid) + return verify + + +def _failing_verifier(exc=None): + def verify(token: str) -> VerifiedIASClaims: + raise (exc or IASTokenError("verification failed")) + return verify + + +# --------------------------------------------------------------------------- +# _auto_configure_verifier +# --------------------------------------------------------------------------- + +class TestAutoConfigureVerifier: + def test_returns_verifier_when_env_configured(self): + mock_verifier = MagicMock(spec=IASVerifier) + with patch("sap_cloud_sdk.core.telemetry.middleware.starlette_a2a.IASVerifier") as MockV: + MockV.from_env.return_value = mock_verifier + result = _auto_configure_verifier() + assert result is mock_verifier + + def test_returns_none_when_config_missing(self): + with patch("sap_cloud_sdk.core.telemetry.middleware.starlette_a2a.IASVerifier") as MockV: + MockV.from_env.side_effect = IASConfigError("no binding found") + result = _auto_configure_verifier() + assert result is None + + def test_logs_warning_when_config_missing(self, caplog): + with patch("sap_cloud_sdk.core.telemetry.middleware.starlette_a2a.IASVerifier") as MockV: + MockV.from_env.side_effect = IASConfigError("no IAS binding") + with caplog.at_level(logging.WARNING): + _auto_configure_verifier() + assert any("NOT be stamped" in r.message for r in caplog.records) + + def test_no_warning_when_configured(self, caplog): + with patch("sap_cloud_sdk.core.telemetry.middleware.starlette_a2a.IASVerifier") as MockV: + MockV.from_env.return_value = MagicMock(spec=IASVerifier) + with caplog.at_level(logging.WARNING): + _auto_configure_verifier() + assert not any("NOT be stamped" in r.message for r in caplog.records) + + +# --------------------------------------------------------------------------- +# StarletteIASTelemetryMiddleware construction +# --------------------------------------------------------------------------- + class TestStarletteIASTelemetryMiddleware: def test_register_calls_add_middleware_on_self_app(self): app = MagicMock() - mw = StarletteIASTelemetryMiddleware(app=app) + mw = StarletteIASTelemetryMiddleware(app=app, token_verifier=_passing_verifier()) mw.register() app.add_middleware.assert_called_once() def test_get_attributes_returns_empty_outside_request(self): - mw = StarletteIASTelemetryMiddleware(app=MagicMock()) + mw = StarletteIASTelemetryMiddleware(app=MagicMock(), token_verifier=_passing_verifier()) assert mw.get_attributes() == {} def test_each_instance_has_independent_context_var(self): - mw1 = StarletteIASTelemetryMiddleware(app=MagicMock()) - mw2 = StarletteIASTelemetryMiddleware(app=MagicMock()) + mw1 = StarletteIASTelemetryMiddleware(app=MagicMock(), token_verifier=_passing_verifier()) + mw2 = StarletteIASTelemetryMiddleware(app=MagicMock(), token_verifier=_passing_verifier()) assert mw1._attrs_var is not mw2._attrs_var def test_two_instances_do_not_interfere(self): - mw1 = StarletteIASTelemetryMiddleware(app=MagicMock()) - mw2 = StarletteIASTelemetryMiddleware(app=MagicMock()) - + mw1 = StarletteIASTelemetryMiddleware(app=MagicMock(), token_verifier=_passing_verifier()) + mw2 = StarletteIASTelemetryMiddleware(app=MagicMock(), token_verifier=_passing_verifier()) t1 = mw1._attrs_var.set({ATTR_SAP_TENANT_ID: "tenant-a"}) t2 = mw2._attrs_var.set({ATTR_USER_ID: "user-b"}) try: @@ -54,74 +104,122 @@ def test_two_instances_do_not_interfere(self): mw1._attrs_var.reset(t1) mw2._attrs_var.reset(t2) - -class TestParseRequest: - def test_extracts_tenant_and_user(self): - claims = _make_claims(sap_gtid="t1", user_uuid="u1") + def test_auto_configure_called_when_no_verifier_given(self): + mock_verifier = MagicMock(spec=IASVerifier) + with patch("sap_cloud_sdk.core.telemetry.middleware.starlette_a2a.IASVerifier") as MockV: + MockV.from_env.return_value = mock_verifier + mw = StarletteIASTelemetryMiddleware(app=MagicMock()) + assert mw._token_verifier is mock_verifier + + def test_auto_configure_failure_sets_none_verifier(self): + with patch("sap_cloud_sdk.core.telemetry.middleware.starlette_a2a.IASVerifier") as MockV: + MockV.from_env.side_effect = IASConfigError("no binding") + mw = StarletteIASTelemetryMiddleware(app=MagicMock()) + assert mw._token_verifier is None + + def test_no_binding_logs_warning(self, caplog): + with patch("sap_cloud_sdk.core.telemetry.middleware.starlette_a2a.IASVerifier") as MockV: + MockV.from_env.side_effect = IASConfigError("no binding") + with caplog.at_level(logging.WARNING): + StarletteIASTelemetryMiddleware(app=MagicMock()) + assert any("NOT be stamped" in r.message for r in caplog.records) + + def test_explicit_verifier_bypasses_auto_configure(self): + with patch("sap_cloud_sdk.core.telemetry.middleware.starlette_a2a.IASVerifier") as MockV: + explicit = _passing_verifier() + mw = StarletteIASTelemetryMiddleware(app=MagicMock(), token_verifier=explicit) + MockV.from_env.assert_not_called() + assert mw._token_verifier is explicit + + def test_explicit_verifier_no_warning(self, caplog): + with caplog.at_level(logging.WARNING): + StarletteIASTelemetryMiddleware(app=MagicMock(), token_verifier=_passing_verifier()) + assert not any("NOT be stamped" in r.message for r in caplog.records) + + +# --------------------------------------------------------------------------- +# _verify_and_extract: core security behaviour +# --------------------------------------------------------------------------- + +class TestVerifyAndExtract: + + def test_no_verifier_returns_none_claims_and_empty_attrs(self): request = _make_request({"authorization": "Bearer tok"}) - with patch(_PATCH_PARSE, return_value=claims): - _, result = _parse_request(request) - assert result == {ATTR_SAP_TENANT_ID: "t1", ATTR_USER_ID: "u1"} + claims, attrs = _verify_and_extract(request, None) + assert claims is None + assert ATTR_SAP_TENANT_ID not in attrs + assert ATTR_USER_ID not in attrs - def test_omits_missing_tenant(self): - claims = _make_claims(sap_gtid=None, user_uuid="u1") + def test_verified_token_returns_claims_and_stamps_identity(self): request = _make_request({"authorization": "Bearer tok"}) - with patch(_PATCH_PARSE, return_value=claims): - _, result = _parse_request(request) - assert result == {ATTR_USER_ID: "u1"} - assert ATTR_SAP_TENANT_ID not in result + claims, attrs = _verify_and_extract(request, _passing_verifier(sap_gtid="t1", user_uuid="u1")) + assert claims is not None + assert claims.sap_gtid == "t1" + assert attrs[ATTR_SAP_TENANT_ID] == "t1" + assert attrs[ATTR_USER_ID] == "u1" + + def test_forged_token_returns_none_claims_and_empty_identity(self): + request = _make_request({"authorization": "Bearer forged"}) + claims, attrs = _verify_and_extract(request, _failing_verifier(IASTokenError("bad sig"))) + assert claims is None + assert ATTR_SAP_TENANT_ID not in attrs + assert ATTR_USER_ID not in attrs + + def test_origin_stamped_when_no_verifier(self): + request = _make_request({"authorization": "Bearer tok", "x-sap-origin": "ui5"}) + _, attrs = _verify_and_extract(request, None) + assert attrs == {ATTR_SAP_TRIGGER_TYPE: "ui5"} - def test_omits_missing_user(self): - claims = _make_claims(sap_gtid="t1", user_uuid=None) - request = _make_request({"authorization": "Bearer tok"}) - with patch(_PATCH_PARSE, return_value=claims): - _, result = _parse_request(request) - assert result == {ATTR_SAP_TENANT_ID: "t1"} - assert ATTR_USER_ID not in result + def test_origin_stamped_when_verifier_raises(self): + request = _make_request({"authorization": "Bearer bad", "x-sap-origin": "job"}) + _, attrs = _verify_and_extract(request, _failing_verifier()) + assert attrs[ATTR_SAP_TRIGGER_TYPE] == "job" + assert ATTR_SAP_TENANT_ID not in attrs - def test_returns_empty_when_no_auth_header(self): + def test_origin_stamped_on_verified_path(self): + request = _make_request({"authorization": "Bearer tok", "x-sap-origin": "ui5"}) + _, attrs = _verify_and_extract(request, _passing_verifier(sap_gtid="t1", user_uuid="u1")) + assert attrs[ATTR_SAP_TRIGGER_TYPE] == "ui5" + assert attrs[ATTR_SAP_TENANT_ID] == "t1" + + def test_no_auth_header_returns_none_claims(self): request = _make_request({}) - with patch(_PATCH_PARSE) as mock_parse: - parsed_claims, result = _parse_request(request) - mock_parse.assert_not_called() - assert result == {} - assert parsed_claims is None + claims, attrs = _verify_and_extract(request, _passing_verifier(sap_gtid="t1")) + assert claims is None + assert attrs == {} - def test_returns_empty_on_parse_error(self): - request = _make_request({"authorization": "Bearer bad"}) - with patch(_PATCH_PARSE, side_effect=ValueError("bad token")): - parsed_claims, result = _parse_request(request) - assert result == {} - assert parsed_claims is None + def test_no_auth_header_origin_still_stamped(self): + request = _make_request({"x-sap-origin": "ui5"}) + _, attrs = _verify_and_extract(request, _passing_verifier(sap_gtid="t1")) + assert attrs == {ATTR_SAP_TRIGGER_TYPE: "ui5"} - def test_returns_empty_when_both_claims_absent(self): - claims = _make_claims(sap_gtid=None, user_uuid=None) + def test_omits_missing_tenant(self): request = _make_request({"authorization": "Bearer tok"}) - with patch(_PATCH_PARSE, return_value=claims): - _, result = _parse_request(request) - assert result == {} + _, attrs = _verify_and_extract(request, _passing_verifier(sap_gtid=None, user_uuid="u1")) + assert ATTR_SAP_TENANT_ID not in attrs + assert attrs[ATTR_USER_ID] == "u1" - def test_includes_origin_header_when_present(self): - claims = _make_claims(sap_gtid="t1", user_uuid="u1") - request = _make_request({"authorization": "Bearer tok", "x-sap-origin": "ui5"}) - with patch(_PATCH_PARSE, return_value=claims): - _, result = _parse_request(request) - assert result[ATTR_SAP_TRIGGER_TYPE] == "ui5" - - def test_omits_origin_attr_when_header_absent(self): - claims = _make_claims(sap_gtid="t1", user_uuid="u1") + def test_omits_missing_user(self): request = _make_request({"authorization": "Bearer tok"}) - with patch(_PATCH_PARSE, return_value=claims): - _, result = _parse_request(request) - assert ATTR_SAP_TRIGGER_TYPE not in result + _, attrs = _verify_and_extract(request, _passing_verifier(sap_gtid="t1", user_uuid=None)) + assert ATTR_USER_ID not in attrs + assert attrs[ATTR_SAP_TENANT_ID] == "t1" + + @pytest.mark.parametrize("exc_msg", [ + "alg=none rejected", "HS256 confusion", "unknown kid", + "wrong issuer", "wrong audience", "token expired", "not yet valid", + ]) + def test_invalid_token_variants_stamp_nothing(self, exc_msg): + request = _make_request({"authorization": "Bearer bad"}) + claims, attrs = _verify_and_extract(request, _failing_verifier(IASTokenError(exc_msg))) + assert claims is None + assert ATTR_SAP_TENANT_ID not in attrs + assert ATTR_USER_ID not in attrs - def test_returns_claims_object(self): - claims = _make_claims(sap_gtid="t1", user_uuid="u1") - request = _make_request({"authorization": "Bearer tok"}) - with patch(_PATCH_PARSE, return_value=claims): - parsed_claims, _ = _parse_request(request) - assert parsed_claims is claims +# --------------------------------------------------------------------------- +# Inner middleware dispatch — set_auth_context integration +# --------------------------------------------------------------------------- class TestInnerMiddlewareDispatch: def _get_inner_class_and_kwargs(self, mw: StarletteIASTelemetryMiddleware): @@ -132,78 +230,92 @@ def _get_inner_class_and_kwargs(self, mw: StarletteIASTelemetryMiddleware): return args[0], kwargs @pytest.mark.anyio - async def test_sets_attrs_in_context_var_during_request(self): - mw = StarletteIASTelemetryMiddleware(app=MagicMock()) - claims = _make_claims(sap_gtid="tenant-1", user_uuid="user-1") + async def test_sets_attrs_and_auth_context_on_verified_token(self): + mw = StarletteIASTelemetryMiddleware( + app=MagicMock(), + token_verifier=_passing_verifier(sap_gtid="t1", user_uuid="u1"), + ) inner_cls, kwargs = self._get_inner_class_and_kwargs(mw) - request = _make_request({"authorization": "Bearer tok"}) + captured_attrs = {} + captured_context = {} + + from sap_cloud_sdk.ias import get_auth_context + + async def call_next(req): + captured_attrs.update(mw._attrs_var.get()) + ctx = get_auth_context() + if ctx: + captured_context["sap_gtid"] = ctx.sap_gtid + return MagicMock() + + inner = inner_cls(app=MagicMock(), **kwargs) + await inner.dispatch(request, call_next) + + assert captured_attrs[ATTR_SAP_TENANT_ID] == "t1" + assert captured_context.get("sap_gtid") == "t1" + + @pytest.mark.anyio + async def test_forged_token_clears_auth_context(self): + from sap_cloud_sdk.ias import get_auth_context, set_auth_context + + set_auth_context(IASClaims(sap_gtid="stale")) + mw = StarletteIASTelemetryMiddleware( + app=MagicMock(), + token_verifier=_failing_verifier(IASTokenError("forged")), + ) + inner_cls, kwargs = self._get_inner_class_and_kwargs(mw) + request = _make_request({"authorization": "Bearer forged"}) captured = {} async def call_next(req): - captured.update(mw._attrs_var.get()) + captured["ctx"] = get_auth_context() return MagicMock() inner = inner_cls(app=MagicMock(), **kwargs) - with patch(_PATCH_PARSE, lambda t: claims): - await inner.dispatch(request, call_next) + await inner.dispatch(request, call_next) - assert captured == {ATTR_SAP_TENANT_ID: "tenant-1", ATTR_USER_ID: "user-1"} + assert captured["ctx"] is None @pytest.mark.anyio async def test_context_var_reset_after_request(self): - mw = StarletteIASTelemetryMiddleware(app=MagicMock()) - claims = _make_claims(sap_gtid="t1", user_uuid="u1") + mw = StarletteIASTelemetryMiddleware( + app=MagicMock(), + token_verifier=_passing_verifier(sap_gtid="t1", user_uuid="u1"), + ) inner_cls, kwargs = self._get_inner_class_and_kwargs(mw) - request = _make_request({"authorization": "Bearer tok"}) inner = inner_cls(app=MagicMock(), **kwargs) - with patch(_PATCH_PARSE, lambda t: claims): - await inner.dispatch(request, AsyncMock(return_value=MagicMock())) - + await inner.dispatch(request, AsyncMock(return_value=MagicMock())) assert mw._attrs_var.get() == {} @pytest.mark.anyio async def test_context_var_reset_on_exception(self): - mw = StarletteIASTelemetryMiddleware(app=MagicMock()) - claims = _make_claims(sap_gtid="t1", user_uuid="u1") + mw = StarletteIASTelemetryMiddleware( + app=MagicMock(), + token_verifier=_passing_verifier(sap_gtid="t1"), + ) inner_cls, kwargs = self._get_inner_class_and_kwargs(mw) - request = _make_request({"authorization": "Bearer tok"}) async def raises(req): raise RuntimeError("downstream") inner = inner_cls(app=MagicMock(), **kwargs) - with patch(_PATCH_PARSE, lambda t: claims): - with pytest.raises(RuntimeError): - await inner.dispatch(request, raises) - + with pytest.raises(RuntimeError): + await inner.dispatch(request, raises) assert mw._attrs_var.get() == {} - @pytest.mark.anyio - async def test_no_auth_header_sets_empty_attrs(self): - mw = StarletteIASTelemetryMiddleware(app=MagicMock()) - inner_cls, kwargs = self._get_inner_class_and_kwargs(mw) - - request = _make_request({}) - captured = {} - - async def call_next(req): - captured.update(mw._attrs_var.get()) - return MagicMock() - - inner = inner_cls(app=MagicMock(), **kwargs) - await inner.dispatch(request, call_next) - - assert captured == {} - @pytest.mark.anyio async def test_two_instances_independent_during_dispatch(self): - mw1 = StarletteIASTelemetryMiddleware(app=MagicMock()) - mw2 = StarletteIASTelemetryMiddleware(app=MagicMock()) - claims1 = _make_claims(sap_gtid="tenant-1", user_uuid=None) - claims2 = _make_claims(sap_gtid=None, user_uuid="user-2") + mw1 = StarletteIASTelemetryMiddleware( + app=MagicMock(), + token_verifier=_passing_verifier(sap_gtid="tenant-1", user_uuid=None), + ) + mw2 = StarletteIASTelemetryMiddleware( + app=MagicMock(), + token_verifier=_passing_verifier(sap_gtid=None, user_uuid="user-2"), + ) inner1_cls, kwargs1 = self._get_inner_class_and_kwargs(mw1) inner2_cls, kwargs2 = self._get_inner_class_and_kwargs(mw2) inner1 = inner1_cls(app=MagicMock(), **kwargs1) @@ -220,10 +332,8 @@ async def next2(r): captured2.update(mw2._attrs_var.get()) return MagicMock() - with patch(_PATCH_PARSE, lambda t: claims1): - await inner1.dispatch(req, next1) - with patch(_PATCH_PARSE, lambda t: claims2): - await inner2.dispatch(req, next2) + await inner1.dispatch(req, next1) + await inner2.dispatch(req, next2) assert captured1 == {ATTR_SAP_TENANT_ID: "tenant-1"} assert captured2 == {ATTR_USER_ID: "user-2"} diff --git a/tests/ias/unit/test_verifier.py b/tests/ias/unit/test_verifier.py new file mode 100644 index 00000000..e1181b12 --- /dev/null +++ b/tests/ias/unit/test_verifier.py @@ -0,0 +1,203 @@ +"""Unit tests for sap_cloud_sdk.ias IASVerifier.""" + +import json +import pytest +from unittest.mock import MagicMock, patch + +from sap_cloud_sdk.ias import IASClaims, IASConfigError, IASTokenError, IASVerifier, VerifiedIASClaims + + +class TestIASVerifierFromEnv: + def test_cf_vcap_services_identity_binding(self, monkeypatch): + vcap = {"identity": [{"credentials": {"url": "https://ias.example.com", "clientid": "my-app"}}]} + monkeypatch.setenv("VCAP_SERVICES", json.dumps(vcap)) + monkeypatch.delenv("IAS_URL", raising=False) + with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"): + v = IASVerifier.from_env() + assert v._ias_url == "https://ias.example.com" + assert v._client_id == "my-app" + + def test_cf_vcap_services_strips_trailing_slash(self, monkeypatch): + vcap = {"identity": [{"credentials": {"url": "https://ias.example.com/", "clientid": "cid"}}]} + monkeypatch.setenv("VCAP_SERVICES", json.dumps(vcap)) + monkeypatch.delenv("IAS_URL", raising=False) + with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"): + v = IASVerifier.from_env() + assert v._ias_url == "https://ias.example.com" + + def test_cf_vcap_services_no_client_id(self, monkeypatch): + vcap = {"identity": [{"credentials": {"url": "https://ias.example.com"}}]} + monkeypatch.setenv("VCAP_SERVICES", json.dumps(vcap)) + monkeypatch.delenv("IAS_URL", raising=False) + with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"): + v = IASVerifier.from_env() + assert v._client_id is None + + def test_cf_vcap_services_xsuaa_fallback(self, monkeypatch): + vcap = {"xsuaa": [{"credentials": {"url": "https://xsuaa.example.com", "clientid": "xc"}}]} + monkeypatch.setenv("VCAP_SERVICES", json.dumps(vcap)) + monkeypatch.delenv("IAS_URL", raising=False) + with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"): + v = IASVerifier.from_env() + assert v._ias_url == "https://xsuaa.example.com" + + def test_k8s_env_vars(self, monkeypatch): + monkeypatch.delenv("VCAP_SERVICES", raising=False) + monkeypatch.setenv("IAS_URL", "https://k8s-ias.example.com") + monkeypatch.setenv("IAS_CLIENT_ID", "k8s-client") + with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"): + v = IASVerifier.from_env() + assert v._ias_url == "https://k8s-ias.example.com" + assert v._client_id == "k8s-client" + + def test_k8s_env_vars_no_client_id(self, monkeypatch): + monkeypatch.delenv("VCAP_SERVICES", raising=False) + monkeypatch.setenv("IAS_URL", "https://k8s-ias.example.com") + monkeypatch.delenv("IAS_CLIENT_ID", raising=False) + with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"): + v = IASVerifier.from_env() + assert v._client_id is None + + def test_raises_when_nothing_configured(self, monkeypatch): + monkeypatch.delenv("VCAP_SERVICES", raising=False) + monkeypatch.delenv("IAS_URL", raising=False) + monkeypatch.delenv("IAS_CLIENT_ID", raising=False) + with pytest.raises(IASConfigError): + IASVerifier.from_env() + + def test_raises_with_malformed_vcap(self, monkeypatch): + monkeypatch.setenv("VCAP_SERVICES", "not-valid-json") + monkeypatch.delenv("IAS_URL", raising=False) + with pytest.raises(IASConfigError): + IASVerifier.from_env() + + def test_raises_with_empty_bindings(self, monkeypatch): + vcap = {"identity": [], "xsuaa": []} + monkeypatch.setenv("VCAP_SERVICES", json.dumps(vcap)) + monkeypatch.delenv("IAS_URL", raising=False) + with pytest.raises(IASConfigError): + IASVerifier.from_env() + + def test_vcap_takes_precedence_over_env_var(self, monkeypatch): + vcap = {"identity": [{"credentials": {"url": "https://vcap-ias.example.com", "clientid": "vc"}}]} + monkeypatch.setenv("VCAP_SERVICES", json.dumps(vcap)) + monkeypatch.setenv("IAS_URL", "https://env-ias.example.com") + with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"): + v = IASVerifier.from_env() + assert v._ias_url == "https://vcap-ias.example.com" + + +class TestIASVerifierCall: + def _make_verifier(self, ias_url="https://ias.example.com", client_id=None): + with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"): + v = IASVerifier(ias_url=ias_url, client_id=client_id) + return v + + def _mock_jwk_key(self, verifier): + mock_key = MagicMock() + mock_key.key = "mock-signing-key" + verifier._jwk_client.get_signing_key_from_jwt.return_value = mock_key + return mock_key + + def test_returns_verified_ias_claims_on_success(self): + import jwt as pyjwt + verifier = self._make_verifier() + raw_token = pyjwt.encode({"sap_gtid": "t1", "user_uuid": "u1"}, key="s", algorithm="HS256") + self._mock_jwk_key(verifier) + with patch("sap_cloud_sdk.ias._verifier.jwt.decode") as mock_decode: + mock_decode.return_value = {"sap_gtid": "t1", "user_uuid": "u1", "iss": "https://ias.example.com"} + result = verifier(f"Bearer {raw_token}") + assert isinstance(result, VerifiedIASClaims) + assert result.claims.sap_gtid == "t1" + + def test_strips_bearer_prefix(self): + import jwt as pyjwt + verifier = self._make_verifier() + raw_token = pyjwt.encode({"sub": "x"}, key="s", algorithm="HS256") + self._mock_jwk_key(verifier) + with patch("sap_cloud_sdk.ias._verifier.jwt.decode") as mock_decode: + mock_decode.return_value = {} + verifier(f"Bearer {raw_token}") + call_args = verifier._jwk_client.get_signing_key_from_jwt.call_args[0][0] + assert call_args == raw_token + + def test_raises_on_bad_signature(self): + import jwt.exceptions + verifier = self._make_verifier() + verifier._jwk_client.get_signing_key_from_jwt.side_effect = jwt.exceptions.InvalidSignatureError("bad") + with pytest.raises(IASTokenError, match="IAS JWT verification failed"): + verifier("Bearer bad.token") + + def test_raises_on_expired_token(self): + import jwt.exceptions + verifier = self._make_verifier() + verifier._jwk_client.get_signing_key_from_jwt.side_effect = jwt.exceptions.ExpiredSignatureError("exp") + with pytest.raises(IASTokenError): + verifier("Bearer expired") + + def test_raises_on_wrong_issuer(self): + import jwt.exceptions + verifier = self._make_verifier() + self._mock_jwk_key(verifier) + with patch("sap_cloud_sdk.ias._verifier.jwt.decode") as mock_decode: + mock_decode.side_effect = jwt.exceptions.InvalidIssuerError("iss") + with pytest.raises(IASTokenError): + verifier("Bearer tok") + + def test_raises_on_wrong_audience(self): + import jwt.exceptions + verifier = self._make_verifier(client_id="expected") + self._mock_jwk_key(verifier) + with patch("sap_cloud_sdk.ias._verifier.jwt.decode") as mock_decode: + mock_decode.side_effect = jwt.exceptions.InvalidAudienceError("aud") + with pytest.raises(IASTokenError): + verifier("Bearer tok") + + def test_pins_asymmetric_algorithms_only(self): + import jwt as pyjwt + verifier = self._make_verifier() + raw_token = pyjwt.encode({"sub": "x"}, key="s", algorithm="HS256") + self._mock_jwk_key(verifier) + with patch("sap_cloud_sdk.ias._verifier.jwt.decode") as mock_decode: + mock_decode.return_value = {} + verifier(f"Bearer {raw_token}") + _, kwargs = mock_decode.call_args_list[0] + assert kwargs["algorithms"] == ["RS256", "ES256"] + assert "none" not in kwargs["algorithms"] + assert "HS256" not in kwargs["algorithms"] + + def test_requires_exp_and_iss(self): + import jwt as pyjwt + verifier = self._make_verifier() + raw_token = pyjwt.encode({"sub": "x"}, key="s", algorithm="HS256") + self._mock_jwk_key(verifier) + with patch("sap_cloud_sdk.ias._verifier.jwt.decode") as mock_decode: + mock_decode.return_value = {} + verifier(f"Bearer {raw_token}") + _, kwargs = mock_decode.call_args_list[0] + assert "exp" in kwargs["options"]["require"] + assert "iss" in kwargs["options"]["require"] + + def test_includes_audience_when_client_id_set(self): + import jwt as pyjwt + verifier = self._make_verifier(client_id="my-client") + raw_token = pyjwt.encode({"sub": "x"}, key="s", algorithm="HS256") + self._mock_jwk_key(verifier) + with patch("sap_cloud_sdk.ias._verifier.jwt.decode") as mock_decode: + mock_decode.return_value = {} + verifier(f"Bearer {raw_token}") + _, kwargs = mock_decode.call_args_list[0] + assert kwargs.get("audience") == "my-client" + assert "aud" in kwargs["options"]["require"] + + def test_skips_audience_when_no_client_id(self): + import jwt as pyjwt + verifier = self._make_verifier(client_id=None) + raw_token = pyjwt.encode({"sub": "x"}, key="s", algorithm="HS256") + self._mock_jwk_key(verifier) + with patch("sap_cloud_sdk.ias._verifier.jwt.decode") as mock_decode: + mock_decode.return_value = {} + verifier(f"Bearer {raw_token}") + _, kwargs = mock_decode.call_args_list[0] + assert "audience" not in kwargs + assert kwargs["options"].get("verify_aud") is False