diff --git a/src/sap_cloud_sdk/core/runtime_context/providers/_ias.py b/src/sap_cloud_sdk/core/runtime_context/providers/_ias.py index f23b4418..2bb8882f 100644 --- a/src/sap_cloud_sdk/core/runtime_context/providers/_ias.py +++ b/src/sap_cloud_sdk/core/runtime_context/providers/_ias.py @@ -1,12 +1,13 @@ """IAS context provider and its typed keys.""" import logging +from typing import Optional from sap_cloud_sdk.core.runtime_context._context import RuntimeContext from sap_cloud_sdk.core.runtime_context._envelope import RequestEnvelope from sap_cloud_sdk.core.runtime_context._keys import ContextKey from sap_cloud_sdk.core.runtime_context._protocol import ContextProvider -from sap_cloud_sdk.ias import parse_token +from sap_cloud_sdk.ias import IASConfigError, IASTokenError, IASVerifier, TokenVerifier logger = logging.getLogger(__name__) @@ -15,6 +16,23 @@ USER_ID = ContextKey[str]("ias.user_uuid") +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("IASContextProvider: auto-configured IASVerifier from environment") + return verifier + except IASConfigError as exc: + logger.warning( + "IASContextProvider: IAS service binding not found — " + "APP_TENANT_ID, GLOBAL_TENANT_ID, and USER_ID will not be populated. " + "Bind an SAP Identity service instance or set IAS_URL to enable " + "identity context. Details: %s", + exc, + ) + return None + + class IASContextProvider(ContextProvider): """Extracts tenant/user context from an IAS JWT ``Authorization`` header. @@ -26,25 +44,46 @@ class IASContextProvider(ContextProvider): - :data:`APP_TENANT_ID` from ``app_tid`` claim - :data:`GLOBAL_TENANT_ID` from ``sap_gtid`` claim - :data:`USER_ID` from ``user_uuid`` claim + + The JWT signature is verified using an :class:`~sap_cloud_sdk.ias.IASVerifier` + before any claims are trusted. When no ``token_verifier`` is supplied, one is + auto-configured from the environment (VCAP_SERVICES → K8s secret mount → + ``IAS_URL`` env var). If no IAS binding is found, a WARNING is logged and the + provider returns an empty context for every request (fail-closed). + + Args: + token_verifier: Optional callable that verifies an IAS JWT and returns + :class:`~sap_cloud_sdk.ias.VerifiedIASClaims`. When ``None``, + auto-configuration from the environment is attempted. """ + def __init__(self, token_verifier: Optional[TokenVerifier] = None) -> None: + if token_verifier is None: + token_verifier = _auto_configure_verifier() + self._token_verifier = token_verifier + def extract(self, envelope: RequestEnvelope) -> RuntimeContext: auth = envelope.headers.get("authorization", "") - - claims = None - if auth: - try: - claims = parse_token(auth) - except Exception as e: - logger.debug("IAS token parsing failed, context will be empty: %s", e) - + if not auth or self._token_verifier is None: + return RuntimeContext() + try: + verified = self._token_verifier(auth) + except IASTokenError as e: + logger.debug("IAS token verification failed, context will be empty: %s", e) + return RuntimeContext() + except Exception as e: + logger.warning( + "IASContextProvider: unexpected error during token verification " + "(IAS unreachable?), context will be empty: %s", + e, + ) + return RuntimeContext() + claims = verified.claims values = {} - if claims: - if claims.app_tid: - values[APP_TENANT_ID] = claims.app_tid - if claims.sap_gtid: - values[GLOBAL_TENANT_ID] = claims.sap_gtid - if claims.user_uuid: - values[USER_ID] = claims.user_uuid - + if claims.app_tid: + values[APP_TENANT_ID] = claims.app_tid + if claims.sap_gtid: + values[GLOBAL_TENANT_ID] = claims.sap_gtid + if claims.user_uuid: + values[USER_ID] = claims.user_uuid return RuntimeContext(values) diff --git a/src/sap_cloud_sdk/core/runtime_context/user-guide.md b/src/sap_cloud_sdk/core/runtime_context/user-guide.md index 152f4a92..cc2396c5 100644 --- a/src/sap_cloud_sdk/core/runtime_context/user-guide.md +++ b/src/sap_cloud_sdk/core/runtime_context/user-guide.md @@ -93,7 +93,7 @@ metadata or a message queue envelope — as long as the adapter populates | Provider | Reads | Sets | |---|---|---| -| `IASContextProvider` | `Authorization: Bearer ` | `TENANT_ID`, `USER_ID`, `GLOBAL_TENANT_ID` | +| `IASContextProvider` | `Authorization: Bearer ` (signature verified via `IASVerifier`) | `TENANT_ID`, `USER_ID`, `GLOBAL_TENANT_ID` | | `SAPTriggerContextProvider` | `x-sap-origin` | `TRIGGER_TYPE` | | `DWCContextProvider` | `dwc-subdomain`, `dwc-tenant`, `dwc-stage-configuration` | `DWC_SUBDOMAIN`, `DWC_TENANT`, `FEATURE_TOGGLES` | @@ -261,3 +261,14 @@ Returns an empty list if `bootstrap()` has not been called yet. Each entry corre ```bash uv run pytest tests/core/unit/runtime_context/ ``` + +--- + +## Multi-tenancy + +| Field | Value | +|---|---| +| **Supported** | Yes | +| **Authentication** | IAS | +| **How to use** | `IASContextProvider` extracts `app_tid` (→ `APP_TENANT_ID`), `sap_gtid` (→ `GLOBAL_TENANT_ID`), and `user_uuid` (→ `USER_ID`) from the IAS JWT on every request. All claims are verified via `IASVerifier` (JWKS-backed signature check). Downstream SDK components (`RuntimeContextSpanProcessor`, `IdentityLogFilter`, `AuditClient`) read these keys to attribute spans, logs, and audit events to the correct tenant. No extra configuration is required — `bootstrap(app)` registers `IASContextProvider` as the first provider by default. | +| **Further reading** | [IAS user guide](../../ias/user-guide.md) · [StarletteIASTelemetryMiddleware](../telemetry/user-guide.md) | diff --git a/tests/core/unit/runtime_context/test_runtime_context.py b/tests/core/unit/runtime_context/test_runtime_context.py index bbb40bc5..4329a7bd 100644 --- a/tests/core/unit/runtime_context/test_runtime_context.py +++ b/tests/core/unit/runtime_context/test_runtime_context.py @@ -2,6 +2,7 @@ import base64 import json +import logging import pytest from unittest.mock import MagicMock, patch @@ -36,8 +37,9 @@ USER_ID, ) from sap_cloud_sdk.core.runtime_context.starlette import _merge +from sap_cloud_sdk.ias import IASConfigError, IASTokenError, VerifiedIASClaims -_PATCH_PARSE = "sap_cloud_sdk.core.runtime_context.providers._ias.parse_token" +_PATCH_FROM_ENV = "sap_cloud_sdk.core.runtime_context.providers._ias.IASVerifier.from_env" # --------------------------------------------------------------------------- @@ -245,12 +247,18 @@ def _make_envelope(headers: dict) -> RequestEnvelope: return RequestEnvelope(headers=headers) +def _make_verifier(claims_obj): + """Return a TokenVerifier that always succeeds with the given claims.""" + def _v(auth: str) -> VerifiedIASClaims: + return VerifiedIASClaims(claims=claims_obj) + return _v + + class TestIASContextProvider: def test_extracts_tenant_and_user(self): claims = _make_claims(app_tid="t-1", user_uuid="u-1") envelope = _make_envelope({"authorization": "Bearer tok"}) - with patch(_PATCH_PARSE, return_value=claims): - ctx = IASContextProvider().extract(envelope) + ctx = IASContextProvider(token_verifier=_make_verifier(claims)).extract(envelope) assert ctx.get(APP_TENANT_ID) == "t-1" assert ctx.get(USER_ID) == "u-1" @@ -259,41 +267,94 @@ def test_does_not_set_trigger_type(self): envelope = _make_envelope( {"authorization": "Bearer tok", "x-sap-origin": "ui5"} ) - with patch(_PATCH_PARSE, return_value=claims): - ctx = IASContextProvider().extract(envelope) + ctx = IASContextProvider(token_verifier=_make_verifier(claims)).extract(envelope) assert ctx.get(TRIGGER_TYPE) is None def test_extracts_global_tenant_id(self): claims = _make_claims(app_tid="t-1", user_uuid="u-1", sap_gtid="g-1") envelope = _make_envelope({"authorization": "Bearer tok"}) - with patch(_PATCH_PARSE, return_value=claims): - ctx = IASContextProvider().extract(envelope) + ctx = IASContextProvider(token_verifier=_make_verifier(claims)).extract(envelope) assert ctx.get(GLOBAL_TENANT_ID) == "g-1" def test_returns_empty_context_when_no_auth_header(self): + claims = _make_claims(app_tid="t-1", user_uuid="u-1") envelope = _make_envelope({}) - ctx = IASContextProvider().extract(envelope) + ctx = IASContextProvider(token_verifier=_make_verifier(claims)).extract(envelope) assert ctx.get(APP_TENANT_ID) is None assert ctx.get(USER_ID) is None assert ctx.get(GLOBAL_TENANT_ID) is None - def test_returns_empty_context_on_parse_error(self): + def test_returns_empty_context_on_verification_error(self): + def _boom(auth: str) -> VerifiedIASClaims: + raise IASTokenError("signature verification failed") envelope = _make_envelope({"authorization": "Bearer bad"}) - with patch(_PATCH_PARSE, side_effect=ValueError("bad")): - ctx = IASContextProvider().extract(envelope) + ctx = IASContextProvider(token_verifier=_boom).extract(envelope) assert ctx.get(APP_TENANT_ID) is None assert ctx.get(USER_ID) is None def test_tenant_id_none_when_claim_absent(self): claims = _make_claims(app_tid=None, user_uuid="u-1") envelope = _make_envelope({"authorization": "Bearer tok"}) - with patch(_PATCH_PARSE, return_value=claims): - ctx = IASContextProvider().extract(envelope) + ctx = IASContextProvider(token_verifier=_make_verifier(claims)).extract(envelope) assert ctx.get(APP_TENANT_ID) is None assert ctx.get(USER_ID) == "u-1" def test_satisfies_context_provider_protocol(self): - assert isinstance(IASContextProvider(), ContextProvider) + with patch(_PATCH_FROM_ENV, side_effect=IASConfigError("no binding")): + assert isinstance(IASContextProvider(), ContextProvider) + + def test_no_verifier_returns_empty_context(self): + """Fail-closed: no IAS binding → empty context, no crash.""" + with patch(_PATCH_FROM_ENV, side_effect=IASConfigError("no binding")): + provider = IASContextProvider() + assert provider._token_verifier is None + envelope = _make_envelope({"authorization": "Bearer forged.token"}) + ctx = provider.extract(envelope) + assert ctx.get(APP_TENANT_ID) is None + assert ctx.get(GLOBAL_TENANT_ID) is None + assert ctx.get(USER_ID) is None + + def test_forged_token_returns_empty_context(self): + """Verifier raises on forged token → no claims leak into context.""" + def _reject(auth: str) -> VerifiedIASClaims: + raise IASTokenError("invalid signature") + envelope = _make_envelope({"authorization": "Bearer forged"}) + ctx = IASContextProvider(token_verifier=_reject).extract(envelope) + assert ctx.get(APP_TENANT_ID) is None + assert ctx.get(GLOBAL_TENANT_ID) is None + assert ctx.get(USER_ID) is None + + def test_verified_token_populates_all_keys(self): + claims = _make_claims(app_tid="app-t", user_uuid="u-x", sap_gtid="g-x") + envelope = _make_envelope({"authorization": "Bearer real"}) + ctx = IASContextProvider(token_verifier=_make_verifier(claims)).extract(envelope) + assert ctx.get(APP_TENANT_ID) == "app-t" + assert ctx.get(GLOBAL_TENANT_ID) == "g-x" + assert ctx.get(USER_ID) == "u-x" + + def test_missing_auth_header_returns_empty_context(self): + claims = _make_claims(app_tid="app-t", user_uuid="u-x", sap_gtid="g-x") + envelope = _make_envelope({}) + ctx = IASContextProvider(token_verifier=_make_verifier(claims)).extract(envelope) + assert ctx.get(APP_TENANT_ID) is None + assert ctx.get(GLOBAL_TENANT_ID) is None + assert ctx.get(USER_ID) is None + + def test_auto_configure_warns_when_no_binding(self, caplog): + with caplog.at_level(logging.WARNING, logger="sap_cloud_sdk"): + with patch(_PATCH_FROM_ENV, side_effect=IASConfigError("no binding")): + provider = IASContextProvider() + assert provider._token_verifier is None + assert any("IASContextProvider" in r.message for r in caplog.records) + + def test_infra_error_logs_warning_not_debug(self, caplog): + def _infra_fail(auth: str) -> VerifiedIASClaims: + raise ConnectionError("DNS failure") + envelope = _make_envelope({"authorization": "Bearer tok"}) + with caplog.at_level(logging.WARNING, logger="sap_cloud_sdk"): + ctx = IASContextProvider(token_verifier=_infra_fail).extract(envelope) + assert ctx.get(APP_TENANT_ID) is None + assert any("unexpected error" in r.message for r in caplog.records) # ---------------------------------------------------------------------------