fix(ias): require verified provenance for IAS telemetry identity attributes

pyproject.toml

@@ -21,6 +21,7 @@ dependencies = [
     "opentelemetry-instrumentation-langchain>=0.61.0",
     "httpx>=0.27.0",
     "PyJWT>=2.13.0",
+    "cryptography>=44.0.0",
     "protobuf>=4.25.0",
     "protovalidate>=0.13.0",
     "grpcio>=1.60.0",

src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py

@@ -2,15 +2,15 @@
 
 import logging
 from contextvars import ContextVar
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 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.base import TelemetryMiddleware
-from sap_cloud_sdk.ias import parse_token
+from sap_cloud_sdk.ias import IASConfigError, IASVerifier, TokenVerifier, VerifiedIASClaims  # noqa: F401
 
 try:
     from starlette.middleware.base import BaseHTTPMiddleware
@@ -26,31 +26,55 @@
 
 
 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:
-        token = self._attrs_var.set(_extract_ias_attrs(request))
+        token = self._attrs_var.set(_extract_ias_attrs(request, self._token_verifier))
         try:
             return await call_next(request)
         finally:
             self._attrs_var.reset(token)
 
 
 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 <token>`` header on each request,
-    parses it as an IAS JWT, and exposes the following as span attributes:
+    Reads the ``Authorization: Bearer <token>`` 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.
+    **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.
+
+    If the verifier raises for any reason (bad signature, wrong issuer, expired token,
+    unknown algorithm, etc.), the identity attributes are silently omitted and the request
+    continues 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::
 
@@ -59,42 +83,84 @@ 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]:
         """Return IAS JWT attributes extracted from the current request."""
         return self._attrs_var.get()
 
 
-def _extract_ias_attrs(request: Request) -> Dict[str, Any]:
-    """Parse the Authorization header and return telemetry attributes."""
+def _extract_ias_attrs(
+    request: Request, token_verifier: Optional[TokenVerifier]
+) -> Dict[str, Any]:
+    """Extract telemetry attributes from the request.
+
+    ``x-sap-origin`` (trigger type) is always included when present — it is a plain
+    request header, not JWT identity data.
+
+    Identity attributes (``sap.tenancy.tenant_id``, ``user.id``) are included only when
+    ``token_verifier`` is provided and succeeds for the ``Authorization`` header.
+    """
+    attrs: Dict[str, Any] = {}
+
+    # x-sap-origin is not JWT identity — stamp it regardless of verification outcome.
+    origin = request.headers.get("x-sap-origin")
+    if origin:
+        attrs[ATTR_SAP_TRIGGER_TYPE] = origin
+
     auth = request.headers.get("authorization", "")
     if not auth:
-        return {}
+        return attrs
+
+    if token_verifier is None:
+        return attrs  # fail-closed for identity; warned once at construction
+
     try:
-        claims = parse_token(auth)
+        verified = token_verifier(auth)
     except Exception as e:
-        logger.debug("IAS token parsing failed, skipping telemetry attrs: %s", e)
-        return {}
+        logger.debug("IAS token verification failed, skipping identity attrs: %s", e)
+        return attrs
 
-    attrs: Dict[str, Any] = {}
+    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
-    origin = request.headers.get("x-sap-origin")
-    if origin:
-        attrs[ATTR_SAP_TRIGGER_TYPE] = origin
     return 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

src/sap_cloud_sdk/core/telemetry/user-guide.md

@@ -267,21 +267,41 @@ auto_instrument(middlewares=[MyMiddleware(app=app)])
 
 ### Built-in: `StarletteIASTelemetryMiddleware`
 
-For Starlette/FastAPI apps with IAS authentication, the SDK ships a ready-to-use middleware that reads the `Authorization: Bearer <token>` header on each request, parses it as an IAS JWT, and injects:
+For Starlette/FastAPI apps with IAS authentication, the SDK ships a ready-to-use middleware that reads the `Authorization: Bearer <token>` header on each request and, after successful token verification, injects:
 - `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) is always stamped when present, regardless of token verification outcome.
+
+#### `token_verifier` parameter
+
+A `token_verifier` callable is **required to enable identity attributes**. Without it, the middleware logs a one-time warning and stamps **no** `sap.tenancy.tenant_id` / `user.id`. This is a safe default — the app runs normally, but identity attributes are absent from spans until you supply a verifier.
 
 ```python
 from starlette.applications import Starlette
 from sap_cloud_sdk.core.telemetry import auto_instrument
 from sap_cloud_sdk.core.telemetry.middleware import StarletteIASTelemetryMiddleware
+from sap_cloud_sdk.ias import VerifiedIASClaims, parse_token
+
+# Option A: platform pre-verified (e.g. Kyma Istio / UCL already verified the JWT)
+def platform_pre_verified(authorization: str) -> VerifiedIASClaims:
+    return VerifiedIASClaims(claims=parse_token(authorization))
+
+# Option B: real JWKS verification (see the IAS user guide for the full implementation)
+# verifier = make_ias_verifier(jwks_url="https://<tenant>.accounts.ondemand.com/oauth2/certs",
+#                              issuer="https://<tenant>.accounts.ondemand.com",
+#                              audience="<your-client-id>")
 
 app = Starlette(...)
-auto_instrument(middlewares=[StarletteIASTelemetryMiddleware(app=app)])
+auto_instrument(middlewares=[
+    StarletteIASTelemetryMiddleware(app=app, token_verifier=platform_pre_verified)
+])
 ```
 
+If the verifier raises for any reason (bad signature, wrong issuer, expired token, etc.), identity attributes are silently omitted and the request continues normally.
+
+See the [IAS user guide](../../ias/user-guide.md#verified-claims-security-sensitive-consumers) for a full JWKS-based verifier implementation.
+
 ---
 
 ## Configuration

src/sap_cloud_sdk/ias/__init__.py

@@ -1,6 +1,6 @@
 """SAP Cloud SDK for Python - IAS module
 
-Utilities for parsing SAP Identity Authentication Service (IAS) JWT tokens.
+Utilities for parsing and verifying SAP Identity Authentication Service (IAS) JWT tokens.
 
 Usage:
     from sap_cloud_sdk.ias import parse_token, IASClaims
@@ -12,11 +12,16 @@
     print(claims.email)             # user email (when email scope requested)
 """
 
-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",
-    "parse_token",
+    "IASConfigError",
     "IASTokenError",
+    "IASVerifier",
+    "TokenVerifier",
+    "VerifiedIASClaims",
+    "parse_token",
 ]

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
 
@@ -102,6 +102,30 @@ class IASClaims:
     custom_attributes: Dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass(frozen=True)
+class VerifiedIASClaims:
+    """Claims proven to originate from a successfully verified IAS JWT.
+
+    Construct this 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 telemetry
+    identity stamping.
+    """
+
+    claims: IASClaims
+
+
+TokenVerifier = Callable[[str], VerifiedIASClaims]
+"""Callable contract for IAS JWT verifiers.
+
+Receives the raw ``Authorization`` header value (may include the ``"Bearer "``
+prefix) and must raise (fail closed) on any invalid token — bad signature,
+wrong issuer/audience, expired, not-yet-valid, or unknown algorithm.
+
+Returns a :class:`VerifiedIASClaims` instance on success.
+"""
+
+
 def parse_token(token: str) -> IASClaims:
     """Parse an SAP IAS JWT token and return its claims.