fix(ias): verify JWT signature before stamping telemetry and audit identity

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

@@ -2,15 +2,17 @@
 
 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,
     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, 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 <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.
+    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,41 +88,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 _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

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",
 ]
+

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]

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://<tenant>.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 <token>")
+        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 "<not configured>",
+        )
+
+    @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 <token>"`` 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))