fix(runtime_context): verify JWT signature in IASContextProvider

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)

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 <JWT>` | `TENANT_ID`, `USER_ID`, `GLOBAL_TENANT_ID` |
+| `IASContextProvider` | `Authorization: Bearer <JWT>` (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) |

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)
 
 
 # ---------------------------------------------------------------------------