Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 38 additions & 11 deletions src/sap_cloud_sdk/ias/_verifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,13 @@
_SECRET_MODULE = "identity-service"
_SECRET_DEFAULT_INSTANCE = "default"

_NO_CLIENT_ID_HINT = (
"the audience (aud) claim cannot be validated. from_env() refuses to build a "
"verifier that silently accepts tokens issued for any application. Add a "
"clientid to the binding, or — only if you intentionally accept any audience "
"— construct the verifier explicitly: IASVerifier(ias_url=..., client_id=None)."
)


@dataclass
class _IASBindingData:
Expand Down Expand Up @@ -52,7 +59,9 @@ class IASVerifier:
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.
When ``None``, audience validation is skipped — an explicit
opt-out. :meth:`from_env` never produces this state silently;
it raises :class:`IASConfigError` if a binding lacks a clientid.

Usage::

Expand Down Expand Up @@ -99,7 +108,10 @@ def from_env(cls) -> "IASVerifier":
A configured :class:`IASVerifier` instance.

Raises:
IASConfigError: when no IAS configuration can be resolved.
IASConfigError: when no IAS configuration can be resolved, or when a
binding is resolved but carries no ``clientid`` (audience would be
unverifiable). To intentionally accept any audience, construct the
verifier explicitly with ``IASVerifier(ias_url=..., client_id=None)``.
"""
vcap_raw = os.getenv("VCAP_SERVICES")
if vcap_raw:
Expand All @@ -112,35 +124,50 @@ def from_env(cls) -> "IASVerifier":
url = creds.get("url") or creds.get("issuer")
client_id = creds.get("clientid")
if url:
if not client_id:
raise IASConfigError(
f"IAS binding VCAP_SERVICES[{svc_name}] has no "
f"'clientid'; {_NO_CLIENT_ID_HINT}"
) from None
logger.debug(
"IASVerifier.from_env: configured from VCAP_SERVICES[%s]",
svc_name,
)
return cls(ias_url=url, client_id=client_id or None)
return cls(ias_url=url, client_id=client_id)
except (json.JSONDecodeError, KeyError, IndexError, TypeError) as exc:
logger.debug("IASVerifier.from_env: VCAP_SERVICES parse error: %s", exc)

# Step 2 — Kubernetes volume mount (identity-service secret)
binding = _IASBindingData()
try:
binding = _IASBindingData()
_read_secret(
base_volume_mount=_SECRET_MOUNT_BASE,
base_var_name=_ENV_VAR_BASE,
module=_SECRET_MODULE,
instance=_SECRET_DEFAULT_INSTANCE,
target=binding,
)
if binding.url:
logger.debug(
"IASVerifier.from_env: configured from Kubernetes secret mount"
)
return cls(ias_url=binding.url, client_id=binding.clientid or None)
except Exception as exc:
logger.debug("IASVerifier.from_env: secret mount lookup failed: %s", exc)
if binding.url:
if not binding.clientid:
raise IASConfigError(
f"IAS identity-service secret mount has no 'clientid'; "
f"{_NO_CLIENT_ID_HINT}"
) from None
logger.debug(
"IASVerifier.from_env: configured from Kubernetes secret mount"
)
return cls(ias_url=binding.url, client_id=binding.clientid)

ias_url = os.getenv(_ENV_IAS_URL)
if ias_url:
client_id = os.getenv(_ENV_IAS_CLIENT_ID) or None
client_id = os.getenv(_ENV_IAS_CLIENT_ID)
if not client_id:
raise IASConfigError(
f"{_ENV_IAS_URL} is set but {_ENV_IAS_CLIENT_ID} is missing; "
f"{_NO_CLIENT_ID_HINT}"
) from None
logger.debug("IASVerifier.from_env: configured from env vars")
return cls(ias_url=ias_url, client_id=client_id)

Expand All @@ -149,7 +176,7 @@ def from_env(cls) -> "IASVerifier":
f"Set VCAP_SERVICES (CF), mount the identity-service secret at "
f"{_SECRET_MOUNT_BASE}/{_SECRET_MODULE}/{_SECRET_DEFAULT_INSTANCE}/ (Kubernetes), "
f"or set {_ENV_IAS_URL} (and optionally {_ENV_IAS_CLIENT_ID}) manually."
)
) from None

def __call__(self, authorization: str) -> VerifiedIASClaims:
"""Verify the token and return its claims.
Expand Down
21 changes: 19 additions & 2 deletions src/sap_cloud_sdk/ias/user-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,13 @@ add_span_attribute("enduser.id", claims.scim_id or claims.sub or "")

- **Cloud Foundry**: reads `VCAP_SERVICES` → `identity[0]` or `xsuaa[0]` credentials.
- **Kubernetes (managed runtime)**: reads the `identity-service` secret mounted at `/etc/secrets/appfnd/identity-service/default/` — mounted automatically by the agent deployment template, no `app.yaml` changes needed.
- **Manual / local**: reads `IAS_URL` (and optionally `IAS_CLIENT_ID`) environment variables.
- **Manual / local**: reads `IAS_URL` and `IAS_CLIENT_ID` environment variables.

`from_env()` is secure by default: it fails closed. If a binding is resolved but
carries no `clientid`, it raises `IASConfigError` rather than building a verifier
that silently skips audience (`aud`) validation and accepts tokens issued for any
application. To intentionally accept any audience, opt out explicitly by
constructing `IASVerifier(ias_url=..., client_id=None)`.

```python
from sap_cloud_sdk.ias import IASVerifier, IASConfigError
Expand All @@ -104,7 +110,7 @@ from sap_cloud_sdk.ias import IASVerifier, IASConfigError
try:
verifier = IASVerifier.from_env()
except IASConfigError:
# No IAS binding found — handle gracefully
# No IAS binding (or a binding without clientid) — handle gracefully
...

# Or configure explicitly
Expand Down Expand Up @@ -143,3 +149,14 @@ auto_instrument(middlewares=[StarletteIASTelemetryMiddleware(app=app)])
```

If no binding is found, a WARNING is logged at startup and identity attributes are not stamped until a binding is added. See [Telemetry user guide](../core/telemetry/user-guide.md#built-in-starletteiastelemetrymiddleware) for details.

---

## Multi-tenancy

| Field | Value |
|-------|-------|
| **Supported** | Yes |
| **Authentication** | IAS |
| **How to use** | The `app_tid` claim in `IASClaims` carries the SAP tenant ID for the authenticated user. `IASVerifier` enforces the `aud` claim against the application's own `clientid`, preventing tokens issued for sibling applications (cross-tenant or same-tenant) from being accepted. For subscriber-tenant routing (e.g. deriving a per-tenant IAS URL), use `IASVerifier(ias_url=..., client_id=...)` constructed explicitly per tenant rather than `from_env()`. |
| **Further reading** | [SAP IAS multi-tenancy documentation](https://help.sap.com/docs/identity-authentication/identity-authentication/tenant-model-and-licensing) |
33 changes: 27 additions & 6 deletions tests/ias/unit/test_verifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,13 +25,15 @@ def test_cf_vcap_services_strips_trailing_slash(self, monkeypatch):
v = IASVerifier.from_env()
assert v._ias_url == "https://ias.example.com"

def test_cf_vcap_services_no_client_id(self, monkeypatch):
def test_cf_vcap_services_no_client_id_raises(self, monkeypatch):
# Security (FINDING-444): a binding without clientid must not silently
# disable audience validation — from_env fails closed.
vcap = {"identity": [{"credentials": {"url": "https://ias.example.com"}}]}
monkeypatch.setenv("VCAP_SERVICES", json.dumps(vcap))
monkeypatch.delenv("IAS_URL", raising=False)
with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"):
v = IASVerifier.from_env()
assert v._client_id is None
with pytest.raises(IASConfigError, match="clientid"):
IASVerifier.from_env()

def test_cf_vcap_services_xsuaa_fallback(self, monkeypatch):
vcap = {"xsuaa": [{"credentials": {"url": "https://xsuaa.example.com", "clientid": "xc"}}]}
Expand All @@ -50,13 +52,14 @@ def test_k8s_env_vars(self, monkeypatch):
assert v._ias_url == "https://k8s-ias.example.com"
assert v._client_id == "k8s-client"

def test_k8s_env_vars_no_client_id(self, monkeypatch):
def test_k8s_env_vars_no_client_id_raises(self, monkeypatch):
# Security (FINDING-444): IAS_URL without IAS_CLIENT_ID must fail closed.
monkeypatch.delenv("VCAP_SERVICES", raising=False)
monkeypatch.setenv("IAS_URL", "https://k8s-ias.example.com")
monkeypatch.delenv("IAS_CLIENT_ID", raising=False)
with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"):
v = IASVerifier.from_env()
assert v._client_id is None
with pytest.raises(IASConfigError, match="IAS_CLIENT_ID"):
IASVerifier.from_env()

def test_raises_when_nothing_configured(self, monkeypatch):
monkeypatch.delenv("VCAP_SERVICES", raising=False)
Expand Down Expand Up @@ -103,6 +106,7 @@ def _fake_read_secret(base_volume_mount, base_var_name, module, instance, target
def test_k8s_mount_no_url_falls_through_to_env_var(self, monkeypatch):
monkeypatch.delenv("VCAP_SERVICES", raising=False)
monkeypatch.setenv("IAS_URL", "https://env-ias.example.com")
monkeypatch.setenv("IAS_CLIENT_ID", "env-client")

def _fake_read_secret(base_volume_mount, base_var_name, module, instance, target):
target.clientid = "mount-client"
Expand All @@ -113,9 +117,25 @@ def _fake_read_secret(base_volume_mount, base_var_name, module, instance, target
v = IASVerifier.from_env()
assert v._ias_url == "https://env-ias.example.com"

def test_k8s_mount_with_url_no_client_id_raises(self, monkeypatch):
# Security (FINDING-444): a mounted binding with url but no clientid
# must fail closed rather than fall through or skip audience.
monkeypatch.delenv("VCAP_SERVICES", raising=False)
monkeypatch.delenv("IAS_URL", raising=False)

def _fake_read_secret(base_volume_mount, base_var_name, module, instance, target):
target.url = "https://mount-ias.example.com"
# clientid stays ""

with patch("sap_cloud_sdk.ias._verifier._read_secret", side_effect=_fake_read_secret):
with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"):
with pytest.raises(IASConfigError, match="clientid"):
IASVerifier.from_env()

def test_k8s_mount_raises_falls_through_to_env_var(self, monkeypatch):
monkeypatch.delenv("VCAP_SERVICES", raising=False)
monkeypatch.setenv("IAS_URL", "https://env-ias.example.com")
monkeypatch.setenv("IAS_CLIENT_ID", "env-client")

with patch("sap_cloud_sdk.ias._verifier._read_secret", side_effect=RuntimeError("mount not found")):
with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"):
Expand All @@ -141,6 +161,7 @@ def test_mount_takes_precedence_over_env_var(self, monkeypatch):

def _fake_read_secret(base_volume_mount, base_var_name, module, instance, target):
target.url = "https://mount-ias.example.com"
target.clientid = "mount-client"

with patch("sap_cloud_sdk.ias._verifier._read_secret", side_effect=_fake_read_secret):
with patch("sap_cloud_sdk.ias._verifier.PyJWKClient"):
Expand Down
Loading