From 32885f72398dc476ac9893017c9b0c241d94061c Mon Sep 17 00:00:00 2001 From: Tiago Kochenborger Date: Mon, 26 Oct 2026 09:54:56 -0300 Subject: [PATCH 1/4] fix(runtime_context): verify JWT signature in IASContextProvider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit IASContextProvider was calling parse_token() — a deliberately unverified JWT decoder — and writing app_tid, sap_gtid, and user_uuid directly into the request-scoped RuntimeContext. This allowed an attacker to forge a JWT carrying any tenant/user identity and poison the runtime context consumed by RuntimeContextSpanProcessor, IdentityLogFilter, AuditClient, and CBCClient. Apply the same TokenVerifier / fail-closed pattern already used in StarletteIASTelemetryMiddleware: auto-configure an IASVerifier from the environment at construction time; return empty context when no binding is found or when verification fails. Security: fixes incomplete CWE-347 remediation from v0.62.0. --- pyproject.toml | 2 +- .../core/runtime_context/providers/_ias.py | 66 +++++++++++---- .../core/runtime_context/user-guide.md | 2 +- .../runtime_context/test_runtime_context.py | 80 +++++++++++++++---- 4 files changed, 117 insertions(+), 33 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d4835fa4..05685d02 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "sap-cloud-sdk" -version = "0.62.0" +version = "0.62.1" description = "SAP Cloud SDK for Python" readme = "README.md" license = "Apache-2.0" 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..be74c1a2 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, 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,39 @@ 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 Exception as e: + logger.debug("IAS token verification failed, 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..6cee1d9d 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` | diff --git a/tests/core/unit/runtime_context/test_runtime_context.py b/tests/core/unit/runtime_context/test_runtime_context.py index bbb40bc5..f1d3c367 100644 --- a/tests/core/unit/runtime_context/test_runtime_context.py +++ b/tests/core/unit/runtime_context/test_runtime_context.py @@ -36,8 +36,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 +246,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 +266,86 @@ 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): + import logging + 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) # --------------------------------------------------------------------------- From b88d2cb85e98840f6e3ecf67e3d0e2eb710f9f6c Mon Sep 17 00:00:00 2001 From: Tiago Kochenborger Date: Mon, 26 Oct 2026 10:04:12 -0300 Subject: [PATCH 2/4] revert: remove manual version bump (managed by release pipeline) --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 05685d02..d4835fa4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "sap-cloud-sdk" -version = "0.62.1" +version = "0.62.0" description = "SAP Cloud SDK for Python" readme = "README.md" license = "Apache-2.0" From eba0b42bb6851e77f8efb029b99863e975bb85b1 Mon Sep 17 00:00:00 2001 From: Tiago Kochenborger Date: Mon, 26 Oct 2026 10:09:03 -0300 Subject: [PATCH 3/4] fix(runtime_context): split IASTokenError vs infra exception log levels in IASContextProvider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit IASTokenError (bad token) logged at DEBUG — expected, caller-side issue. Any other exception (DNS/TLS/JWKS unreachable) logged at WARNING — operator needs to know identity context is consistently broken. Addresses G6 review W3. --- .../core/runtime_context/providers/_ias.py | 11 +++++++++-- .../core/unit/runtime_context/test_runtime_context.py | 10 ++++++++++ 2 files changed, 19 insertions(+), 2 deletions(-) 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 be74c1a2..2bb8882f 100644 --- a/src/sap_cloud_sdk/core/runtime_context/providers/_ias.py +++ b/src/sap_cloud_sdk/core/runtime_context/providers/_ias.py @@ -7,7 +7,7 @@ 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 IASConfigError, IASVerifier, TokenVerifier +from sap_cloud_sdk.ias import IASConfigError, IASTokenError, IASVerifier, TokenVerifier logger = logging.getLogger(__name__) @@ -68,9 +68,16 @@ def extract(self, envelope: RequestEnvelope) -> RuntimeContext: return RuntimeContext() try: verified = self._token_verifier(auth) - except Exception as e: + 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.app_tid: diff --git a/tests/core/unit/runtime_context/test_runtime_context.py b/tests/core/unit/runtime_context/test_runtime_context.py index f1d3c367..eabe253e 100644 --- a/tests/core/unit/runtime_context/test_runtime_context.py +++ b/tests/core/unit/runtime_context/test_runtime_context.py @@ -347,6 +347,16 @@ def test_auto_configure_warns_when_no_binding(self, caplog): 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): + import logging + 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) + # --------------------------------------------------------------------------- # SAPTriggerContextProvider From b869340c99955b4d33abd24a7c5ce22de0074d49 Mon Sep 17 00:00:00 2001 From: Tiago Kochenborger Date: Mon, 26 Oct 2026 11:51:37 -0300 Subject: [PATCH 4/4] fix(tests): move lazy logging import to top level; add multi-tenancy section to user-guide --- src/sap_cloud_sdk/core/runtime_context/user-guide.md | 11 +++++++++++ .../core/unit/runtime_context/test_runtime_context.py | 3 +-- 2 files changed, 12 insertions(+), 2 deletions(-) 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 6cee1d9d..cc2396c5 100644 --- a/src/sap_cloud_sdk/core/runtime_context/user-guide.md +++ b/src/sap_cloud_sdk/core/runtime_context/user-guide.md @@ -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 eabe253e..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 @@ -340,7 +341,6 @@ def test_missing_auth_header_returns_empty_context(self): assert ctx.get(USER_ID) is None def test_auto_configure_warns_when_no_binding(self, caplog): - import logging with caplog.at_level(logging.WARNING, logger="sap_cloud_sdk"): with patch(_PATCH_FROM_ENV, side_effect=IASConfigError("no binding")): provider = IASContextProvider() @@ -348,7 +348,6 @@ def test_auto_configure_warns_when_no_binding(self, caplog): assert any("IASContextProvider" in r.message for r in caplog.records) def test_infra_error_logs_warning_not_debug(self, caplog): - import logging def _infra_fail(auth: str) -> VerifiedIASClaims: raise ConnectionError("DNS failure") envelope = _make_envelope({"authorization": "Bearer tok"})