From 89699f2b676ebb0fcac774ad0f61f76ed17b2e2d Mon Sep 17 00:00:00 2001 From: ShivSankalp <121006829+ShivSankalp@users.noreply.github.com> Date: Wed, 2 Sep 2026 00:18:18 +0200 Subject: [PATCH] Add standards-based Customer Identity OIDC client --- README.md | 65 ++++++----- pyproject.toml | 5 +- src/namoid/__init__.py | 20 +++- src/namoid/_client.py | 233 ++++++++++++++++++++++++++++++++------ src/namoid/oidc.py | 174 ++++++++++++++++++++++++++++ tests/test_hosted_auth.py | 162 +++++++++++++++++++++----- 6 files changed, 567 insertions(+), 92 deletions(-) create mode 100644 src/namoid/oidc.py diff --git a/README.md b/README.md index 5db9e0f..08ca749 100644 --- a/README.md +++ b/README.md @@ -51,9 +51,9 @@ catches both surfaces without importing the one you do not use. ## Hosted Auth -Hosted Auth redirects the user to a branded NamoID sign-in page and returns a -one-time code. The Client ID resolves the application, its environment, and its -Hosted Auth domain, so there is no issuer or application UUID to configure. +Hosted Auth uses standard OpenID Connect Authorization Code flow with S256 PKCE. +The Client ID resolves the application and issuer; discovery supplies the +authorization, token, UserInfo, revocation, JWKS, and logout endpoints. ```python from namoid import NamoIDClient @@ -63,32 +63,32 @@ namoid = NamoIDClient( client_secret=os.environ["NAMOID_CLIENT_SECRET"], # server-side only ) -# 1. Start a state-bound transaction and keep the verifier in the user's session. -transaction = namoid.create_transaction() +# 1. Start a state-, nonce-, and PKCE-bound transaction. Keep it server-side. +transaction = namoid.create_oidc_transaction("https://app.example/auth/callback") session["namoid_state"] = transaction.state +session["namoid_nonce"] = transaction.nonce session["namoid_verifier"] = transaction.code_verifier -# 2. Send the browser to the application's own hosted sign-in page. -url = namoid.hosted_auth_url( - return_to="https://app.example/auth/callback", - state=transaction.state, - completion_mode="confidential", - code_challenge=transaction.code_challenge, -) +# 2. Send the browser to the discovered authorization endpoint. +url = namoid.authorization_url(transaction) -# 3. On the callback, compare state, then exchange the code on the server. +# 3. On the callback, compare state, then exchange using the same redirect URI. tokens = namoid.exchange_code( code=request.args["code"], code_verifier=session.pop("namoid_verifier"), + redirect_uri="https://app.example/auth/callback", ) -# 4. Confirm the token and create your own application session. -result = namoid.validate_access_token(tokens.access_token) -if not result.valid: - raise Unauthorized() +# 4. Verify the ID token signature and callback-bound nonce, then fetch UserInfo. +claims = namoid.validate_id_token(tokens.raw["id_token"], + nonce=session.pop("namoid_nonce")) +user = namoid.user_info(tokens.access_token) +assert claims["sub"] == user["sub"] -# 5. On sign-out, revoke the NamoID session too. -namoid.revoke_session(access_token=tokens.access_token, refresh_token=tokens.refresh_token) +# 5. On sign-out, revoke the refresh token and redirect through provider logout. +namoid.revoke_token(tokens.refresh_token, token_type_hint="refresh_token") +url = namoid.logout_url(id_token_hint=tokens.raw["id_token"], + post_logout_redirect_uri="https://app.example/signed-out") ``` `AsyncNamoIDClient` has exactly the same methods with `await`, for FastAPI, @@ -98,25 +98,36 @@ Starlette, or any async framework: from namoid import AsyncNamoIDClient async with AsyncNamoIDClient(client_id=..., client_secret=...) as namoid: - tokens = await namoid.exchange_code(code=code, code_verifier=verifier) + tokens = await namoid.exchange_code( + code=code, code_verifier=verifier, + redirect_uri="https://app.example/auth/callback", + ) ``` Both accept an `http_client` if you want to supply your own configured `httpx.Client` / `httpx.AsyncClient`, and cache the auth config after the first fetch. -For a browser-only public client, redirect with `completion_mode="public"` and -exchange with `confidential=False` — PKCE protects the flow and no secret is -involved. Never put a Client Secret anywhere a browser can reach. +For a public client, omit `client_secret`; PKCE protects the code exchange. For +a confidential web application, the SDK sends the secret using HTTP Basic +authentication at the discovered token endpoint. Never put a Client Secret +anywhere a browser can reach. | Method | Endpoint | |---|---| | `get_auth_config()` | `GET /v1/auth/config` | -| `hosted_auth_url(...)` | builds the URL, no request | -| `exchange_code(...)` | `POST /v1/auth/hosted/exchange` | -| `refresh(...)` | `POST /v1/auth/refresh` | +| `get_oidc_discovery()` | issuer `/.well-known/openid-configuration` | +| `authorization_url(...)` | discovered authorization endpoint | +| `exchange_code(...)` | discovered token endpoint | +| `refresh(...)` | discovered token endpoint | +| `user_info(...)` | discovered UserInfo endpoint | +| `validate_id_token(...)` | discovered JWKS endpoint; local verification | +| `revoke_token(...)` | discovered revocation endpoint | +| `logout_url(...)` | discovered end-session endpoint | | `validate_access_token(...)` | `POST /v1/auth/tokens/validate` | -| `revoke_session(...)` | `POST /v1/auth/logout` | + +The older `hosted_auth_url(...)` and `revoke_session(...)` helpers remain for +applications using NamoID's legacy Hosted Auth contract. Every failure raises `NamoIDError`, carrying `status`, `code` (the API's own error code when present), and the parsed `detail`. diff --git a/pyproject.toml b/pyproject.toml index fff4208..634c69d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "namoid" -version = "0.1.0" +version = "0.2.0" description = "Python SDK for NamoID, enterprise identity for India (OAuth 2.1 / OIDC)." readme = "README.md" requires-python = ">=3.10" @@ -44,18 +44,17 @@ classifiers = [ dependencies = [ "httpx>=0.28", + "joserfc>=1.0", ] [project.optional-dependencies] # Protect an MCP server. Framework-agnostic core: discovery, audience-bound # token verification, and RFC 9728 metadata. mcp = [ - "joserfc>=1.0", ] # The same core wired into FastMCP. `fastmcp` itself requires a newer Python # than this package's floor, so pip enforces that when the extra is installed. fastmcp = [ - "joserfc>=1.0", "fastmcp>=3.4.5,<4", ] diff --git a/src/namoid/__init__.py b/src/namoid/__init__.py index f3fb7d4..e67cd70 100644 --- a/src/namoid/__init__.py +++ b/src/namoid/__init__.py @@ -23,7 +23,7 @@ from importlib import import_module from typing import TYPE_CHECKING, Any -__version__ = "0.1.0" +__version__ = "0.2.0" __homepage__ = "https://namoid.in" # Public name -> the module that defines it. Resolved on first attribute access @@ -37,6 +37,12 @@ "HostedAuthTransaction": "namoid.hosted_auth", "TokenResponse": "namoid.hosted_auth", "TokenValidation": "namoid.hosted_auth", + "OIDCDiscovery": "namoid.oidc", + "OIDCTransaction": "namoid.oidc", + "build_authorization_url": "namoid.oidc", + "build_logout_url": "namoid.oidc", + "create_oidc_transaction": "namoid.oidc", + "validate_id_token": "namoid.oidc", "build_configured_hosted_auth_url": "namoid.hosted_auth", "build_hosted_auth_url": "namoid.hosted_auth", "create_hosted_auth_transaction": "namoid.hosted_auth", @@ -48,12 +54,18 @@ "HostedAuthTransaction", "NamoIDClient", "NamoIDError", + "OIDCDiscovery", + "OIDCTransaction", "TokenResponse", "TokenValidation", "__homepage__", "__version__", "build_configured_hosted_auth_url", "build_hosted_auth_url", + "build_authorization_url", + "build_logout_url", + "create_oidc_transaction", + "validate_id_token", "create_hosted_auth_transaction", ] @@ -84,6 +96,12 @@ def __dir__() -> list[str]: from namoid.hosted_auth import ( build_configured_hosted_auth_url as build_configured_hosted_auth_url, ) + from namoid.oidc import OIDCDiscovery as OIDCDiscovery + from namoid.oidc import OIDCTransaction as OIDCTransaction + from namoid.oidc import build_authorization_url as build_authorization_url + from namoid.oidc import build_logout_url as build_logout_url + from namoid.oidc import create_oidc_transaction as create_oidc_transaction + from namoid.oidc import validate_id_token as validate_id_token from namoid.hosted_auth import build_hosted_auth_url as build_hosted_auth_url from namoid.hosted_auth import ( create_hosted_auth_transaction as create_hosted_auth_transaction, diff --git a/src/namoid/_client.py b/src/namoid/_client.py index 21916e3..d6a3d91 100644 --- a/src/namoid/_client.py +++ b/src/namoid/_client.py @@ -7,7 +7,8 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, Mapping +import base64 +from typing import Any, Mapping, Sequence import httpx @@ -21,6 +22,14 @@ build_configured_hosted_auth_url, create_hosted_auth_transaction, ) +from namoid.oidc import ( + OIDCDiscovery, + OIDCTransaction, + build_authorization_url, + build_logout_url, + create_oidc_transaction, + validate_id_token, +) __all__ = ["AsyncNamoIDClient", "NamoIDClient"] @@ -35,6 +44,7 @@ class _Call: path: str params: Mapping[str, Any] | None = None json: Mapping[str, Any] | None = None + data: Mapping[str, Any] | None = None headers: Mapping[str, str] | None = None expect_body: bool = True failure_code: str = "namoid_request_failed" @@ -51,36 +61,37 @@ def _config_call(client_id: str) -> _Call: ) -def _exchange_call( +def _token_call( *, + endpoint: str, code: str, - code_verifier: str | None, - client_id: str | None, + redirect_uri: str, + code_verifier: str, + client_id: str, client_secret: str | None, - device_id: str | None, ) -> _Call: + headers = _oauth_client_headers(client_id, client_secret) return _Call( "POST", - "/v1/auth/hosted/exchange", - json=_compact( - { - "code": code, - "code_verifier": code_verifier, - "device_id": device_id, - "client_id": client_id, - "client_secret": client_secret, - } - ), + endpoint, + data={"grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri, + "code_verifier": code_verifier, "client_id": client_id}, + headers=headers, failure_code="hosted_auth_exchange_failed", - failure_label="Hosted Auth exchange", + failure_label="OIDC code exchange", ) -def _refresh_call(refresh_token: str) -> _Call: +def _refresh_call(*, endpoint: str, refresh_token: str, client_id: str, + client_secret: str | None, scopes: Sequence[str] | None = None) -> _Call: + data = {"grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": client_id} + if scopes: + data["scope"] = " ".join(dict.fromkeys(scopes)) return _Call( "POST", - "/v1/auth/refresh", - json={"refresh_token": refresh_token}, + endpoint, + data=data, + headers=_oauth_client_headers(client_id, client_secret), failure_code="token_refresh_failed", failure_label="Token refresh", ) @@ -98,6 +109,21 @@ def _validate_call(*, token: str, client_id: str | None, client_secret: str | No ) +def _userinfo_call(*, endpoint: str, access_token: str) -> _Call: + return _Call("GET", endpoint, headers={"authorization": f"Bearer {access_token}"}, + failure_code="userinfo_failed", failure_label="OIDC UserInfo") + + +def _revoke_call(*, endpoint: str, token: str, token_type_hint: str | None, + client_id: str, client_secret: str | None) -> _Call: + data = {"token": token, "client_id": client_id} + if token_type_hint: + data["token_type_hint"] = token_type_hint + return _Call("POST", endpoint, data=data, + headers=_oauth_client_headers(client_id, client_secret), expect_body=False, + failure_code="token_revocation_failed", failure_label="OIDC token revocation") + + def _logout_call(*, access_token: str, refresh_token: str | None) -> _Call: return _Call( "POST", @@ -110,6 +136,14 @@ def _logout_call(*, access_token: str, refresh_token: str | None) -> _Call: ) +def _oauth_client_headers(client_id: str, client_secret: str | None) -> Mapping[str, str]: + headers = {"content-type": "application/x-www-form-urlencoded"} + if client_secret: + credentials = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() + headers["authorization"] = f"Basic {credentials}" + return headers + + class _ClientBase: def __init__( self, @@ -126,6 +160,7 @@ def __init__( self._api_base_url = api_base_url.rstrip("/") self._timeout = timeout self._config: AuthConfig | None = None + self._discovery: OIDCDiscovery | None = None @property def client_id(self) -> str: @@ -136,9 +171,20 @@ def create_transaction() -> HostedAuthTransaction: """Fresh ``state`` and PKCE material for one sign-in attempt.""" return create_hosted_auth_transaction() + @staticmethod + def create_oidc_transaction(redirect_uri: str) -> OIDCTransaction: + return create_oidc_transaction(redirect_uri) + def _url(self, path: str) -> str: + if path.startswith(("https://", "http://")): + return path return f"{self._api_base_url}{path}" + def _authorization_url(self, discovery: OIDCDiscovery, transaction: OIDCTransaction, + scopes: Sequence[str], extra_params: Mapping[str, str] | None) -> str: + return build_authorization_url(discovery, self._client_id, transaction, + scopes=scopes, extra_params=extra_params) + def _require_secret(self, provided: str | None) -> str: secret = provided or self._client_secret if not secret: @@ -183,6 +229,23 @@ def get_auth_config(self, *, refresh: bool = False) -> AuthConfig: self._config = AuthConfig.from_payload(self._send(_config_call(self._client_id))) return self._config + def get_oidc_discovery(self, *, refresh: bool = False) -> OIDCDiscovery: + """Fetch and cache the issuer's validated OpenID Connect metadata.""" + if self._discovery is None or refresh: + issuer = self.get_auth_config(refresh=refresh).issuer.rstrip("/") + payload = self._send( + _Call("GET", f"{issuer}/.well-known/openid-configuration", + failure_code="oidc_discovery_failed", failure_label="OIDC discovery") + ) + self._discovery = OIDCDiscovery.from_payload(payload, expected_issuer=issuer) + return self._discovery + + def authorization_url(self, transaction: OIDCTransaction, *, + scopes: Sequence[str] = ("openid", "profile", "email"), + extra_params: Mapping[str, str] | None = None) -> str: + """Build a standard OIDC Authorization Code + PKCE URL.""" + return self._authorization_url(self.get_oidc_discovery(), transaction, scopes, extra_params) + def hosted_auth_url(self, **kwargs: Any) -> str: """Build the Hosted Auth URL for this application. @@ -195,10 +258,10 @@ def exchange_code( self, *, code: str, - code_verifier: str | None = None, + code_verifier: str, + redirect_uri: str, client_secret: str | None = None, - device_id: str | None = None, - confidential: bool = True, + confidential: bool | None = None, ) -> TokenResponse: """Exchange a one-time Hosted Auth code for a NamoID session. @@ -207,21 +270,60 @@ def exchange_code( matching ``completion_mode="confidential"`` on the redirect. Pass false for the browser-only PKCE flow. """ - secret = self._require_secret(client_secret) if confidential else None + secret = client_secret or self._client_secret + if confidential is True: + secret = self._require_secret(client_secret) + discovery = self.get_oidc_discovery() payload = self._send( - _exchange_call( + _token_call( + endpoint=discovery.token_endpoint, code=code, + redirect_uri=redirect_uri, code_verifier=code_verifier, client_id=self._client_id, client_secret=secret, - device_id=device_id, ) ) return TokenResponse.from_payload(payload) - def refresh(self, refresh_token: str) -> TokenResponse: + def refresh(self, refresh_token: str, *, scopes: Sequence[str] | None = None, + client_secret: str | None = None) -> TokenResponse: """Rotate a refresh token for a new session.""" - return TokenResponse.from_payload(self._send(_refresh_call(refresh_token))) + discovery = self.get_oidc_discovery() + return TokenResponse.from_payload(self._send(_refresh_call( + endpoint=discovery.token_endpoint, refresh_token=refresh_token, + client_id=self._client_id, client_secret=client_secret or self._client_secret, + scopes=scopes, + ))) + + def user_info(self, access_token: str) -> Mapping[str, Any]: + return self._send(_userinfo_call( + endpoint=self.get_oidc_discovery().userinfo_endpoint, access_token=access_token + )) + + def validate_id_token(self, id_token: str, *, nonce: str) -> Mapping[str, Any]: + discovery = self.get_oidc_discovery() + jwks = self._send(_Call("GET", discovery.jwks_uri, failure_code="jwks_unavailable", + failure_label="OIDC signing keys")) + return validate_id_token(id_token, jwks=jwks, issuer=discovery.issuer, + client_id=self._client_id, nonce=nonce) + + def revoke_token(self, token: str, *, token_type_hint: str | None = None, + client_secret: str | None = None) -> None: + discovery = self.get_oidc_discovery() + if not discovery.revocation_endpoint: + raise NamoIDError("The issuer does not advertise token revocation", + code="revocation_unavailable") + self._send(_revoke_call( + endpoint=discovery.revocation_endpoint, token=token, + token_type_hint=token_type_hint, client_id=self._client_id, + client_secret=client_secret or self._client_secret, + )) + + def logout_url(self, *, id_token_hint: str, post_logout_redirect_uri: str | None = None, + state: str | None = None) -> str: + return build_logout_url(self.get_oidc_discovery(), id_token_hint=id_token_hint, + post_logout_redirect_uri=post_logout_redirect_uri, state=state) def validate_access_token( self, token: str, *, client_secret: str | None = None @@ -254,6 +356,7 @@ def _send(self, call: _Call) -> Any: self._url(call.path), params=dict(call.params or {}) or None, json=dict(call.json) if call.json is not None else None, + data=dict(call.data) if call.data is not None else None, headers={"accept": "application/json", **(call.headers or {})}, ) except httpx.HTTPError as exc: @@ -290,6 +393,23 @@ async def get_auth_config(self, *, refresh: bool = False) -> AuthConfig: ) return self._config + async def get_oidc_discovery(self, *, refresh: bool = False) -> OIDCDiscovery: + if self._discovery is None or refresh: + issuer = (await self.get_auth_config(refresh=refresh)).issuer.rstrip("/") + payload = await self._send( + _Call("GET", f"{issuer}/.well-known/openid-configuration", + failure_code="oidc_discovery_failed", failure_label="OIDC discovery") + ) + self._discovery = OIDCDiscovery.from_payload(payload, expected_issuer=issuer) + return self._discovery + + async def authorization_url(self, transaction: OIDCTransaction, *, + scopes: Sequence[str] = ("openid", "profile", "email"), + extra_params: Mapping[str, str] | None = None) -> str: + return self._authorization_url( + await self.get_oidc_discovery(), transaction, scopes, extra_params + ) + async def hosted_auth_url(self, **kwargs: Any) -> str: return self._hosted_url(await self.get_auth_config(), kwargs) @@ -297,25 +417,67 @@ async def exchange_code( self, *, code: str, - code_verifier: str | None = None, + code_verifier: str, + redirect_uri: str, client_secret: str | None = None, - device_id: str | None = None, - confidential: bool = True, + confidential: bool | None = None, ) -> TokenResponse: - secret = self._require_secret(client_secret) if confidential else None + secret = client_secret or self._client_secret + if confidential is True: + secret = self._require_secret(client_secret) + discovery = await self.get_oidc_discovery() payload = await self._send( - _exchange_call( + _token_call( + endpoint=discovery.token_endpoint, code=code, + redirect_uri=redirect_uri, code_verifier=code_verifier, client_id=self._client_id, client_secret=secret, - device_id=device_id, ) ) return TokenResponse.from_payload(payload) - async def refresh(self, refresh_token: str) -> TokenResponse: - return TokenResponse.from_payload(await self._send(_refresh_call(refresh_token))) + async def refresh(self, refresh_token: str, *, scopes: Sequence[str] | None = None, + client_secret: str | None = None) -> TokenResponse: + discovery = await self.get_oidc_discovery() + return TokenResponse.from_payload(await self._send(_refresh_call( + endpoint=discovery.token_endpoint, refresh_token=refresh_token, + client_id=self._client_id, client_secret=client_secret or self._client_secret, + scopes=scopes, + ))) + + async def user_info(self, access_token: str) -> Mapping[str, Any]: + return await self._send(_userinfo_call( + endpoint=(await self.get_oidc_discovery()).userinfo_endpoint, + access_token=access_token, + )) + + async def validate_id_token(self, id_token: str, *, nonce: str) -> Mapping[str, Any]: + discovery = await self.get_oidc_discovery() + jwks = await self._send(_Call("GET", discovery.jwks_uri, + failure_code="jwks_unavailable", + failure_label="OIDC signing keys")) + return validate_id_token(id_token, jwks=jwks, issuer=discovery.issuer, + client_id=self._client_id, nonce=nonce) + + async def revoke_token(self, token: str, *, token_type_hint: str | None = None, + client_secret: str | None = None) -> None: + discovery = await self.get_oidc_discovery() + if not discovery.revocation_endpoint: + raise NamoIDError("The issuer does not advertise token revocation", + code="revocation_unavailable") + await self._send(_revoke_call( + endpoint=discovery.revocation_endpoint, token=token, + token_type_hint=token_type_hint, client_id=self._client_id, + client_secret=client_secret or self._client_secret, + )) + + async def logout_url(self, *, id_token_hint: str, + post_logout_redirect_uri: str | None = None, + state: str | None = None) -> str: + return build_logout_url(await self.get_oidc_discovery(), id_token_hint=id_token_hint, + post_logout_redirect_uri=post_logout_redirect_uri, state=state) async def validate_access_token( self, token: str, *, client_secret: str | None = None @@ -341,6 +503,7 @@ async def _send(self, call: _Call) -> Any: self._url(call.path), params=dict(call.params or {}) or None, json=dict(call.json) if call.json is not None else None, + data=dict(call.data) if call.data is not None else None, headers={"accept": "application/json", **(call.headers or {})}, ) except httpx.HTTPError as exc: diff --git a/src/namoid/oidc.py b/src/namoid/oidc.py new file mode 100644 index 0000000..bd6b52b --- /dev/null +++ b/src/namoid/oidc.py @@ -0,0 +1,174 @@ +"""Standards-based OpenID Connect primitives for NamoID Customer Identity.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Mapping, Sequence +from urllib.parse import urlencode, urlsplit, urlunsplit + +from namoid._errors import NamoIDError +from namoid.hosted_auth import pkce_challenge, random_base64url + +__all__ = [ + "OIDCDiscovery", "OIDCTransaction", "build_authorization_url", "build_logout_url", + "create_oidc_transaction", "validate_id_token", +] + + +@dataclass(frozen=True) +class OIDCDiscovery: + issuer: str + authorization_endpoint: str + token_endpoint: str + userinfo_endpoint: str + jwks_uri: str + revocation_endpoint: str | None = None + end_session_endpoint: str | None = None + code_challenge_methods_supported: Sequence[str] = () + raw: Mapping[str, Any] = field(default_factory=dict) + + @classmethod + def from_payload(cls, payload: Mapping[str, Any], *, expected_issuer: str) -> "OIDCDiscovery": + issuer = _normalize_issuer(expected_issuer) + declared = _normalize_issuer(str(payload.get("issuer") or "")) + if declared != issuer: + raise NamoIDError("OIDC discovery issuer mismatch", code="issuer_mismatch") + required = ("authorization_endpoint", "token_endpoint", "userinfo_endpoint", "jwks_uri") + missing = [name for name in required if not payload.get(name)] + if missing: + raise NamoIDError( + f"OIDC discovery document is missing {missing[0]!r}", + code="invalid_discovery_document", + ) + endpoints = [payload[name] for name in required] + endpoints += [payload.get("revocation_endpoint"), payload.get("end_session_endpoint")] + for endpoint in filter(None, endpoints): + _require_trusted_endpoint(issuer, str(endpoint)) + methods = list(payload.get("code_challenge_methods_supported") or []) + if "S256" not in methods: + raise NamoIDError("The issuer does not advertise PKCE S256", code="pkce_s256_unavailable") + return cls( + issuer=declared, + authorization_endpoint=str(payload["authorization_endpoint"]), + token_endpoint=str(payload["token_endpoint"]), + userinfo_endpoint=str(payload["userinfo_endpoint"]), + jwks_uri=str(payload["jwks_uri"]), + revocation_endpoint=payload.get("revocation_endpoint"), + end_session_endpoint=payload.get("end_session_endpoint"), + code_challenge_methods_supported=methods, + raw=dict(payload), + ) + + +@dataclass(frozen=True) +class OIDCTransaction: + state: str + nonce: str + code_verifier: str + code_challenge: str + redirect_uri: str + code_challenge_method: str = "S256" + + +def create_oidc_transaction(redirect_uri: str) -> OIDCTransaction: + if not redirect_uri: + raise NamoIDError("redirect_uri is required", code="missing_redirect_uri") + verifier = random_base64url(48) + return OIDCTransaction( + state=random_base64url(32), + nonce=random_base64url(32), + code_verifier=verifier, + code_challenge=pkce_challenge(verifier), + redirect_uri=redirect_uri, + ) + + +def build_authorization_url( + discovery: OIDCDiscovery, + client_id: str, + transaction: OIDCTransaction, + *, + scopes: Sequence[str] = ("openid", "profile", "email"), + extra_params: Mapping[str, str] | None = None, +) -> str: + normalized_scopes = list(dict.fromkeys(("openid", *filter(None, scopes)))) + params = { + "response_type": "code", + "client_id": client_id, + "redirect_uri": transaction.redirect_uri, + "scope": " ".join(normalized_scopes), + "state": transaction.state, + "nonce": transaction.nonce, + "code_challenge": transaction.code_challenge, + "code_challenge_method": "S256", + } + reserved = set(params) + for key, value in (extra_params or {}).items(): + if key not in reserved: + params[key] = value + parts = urlsplit(discovery.authorization_endpoint) + return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(params), "")) + + +def build_logout_url( + discovery: OIDCDiscovery, + *, + id_token_hint: str, + post_logout_redirect_uri: str | None = None, + state: str | None = None, +) -> str: + if not discovery.end_session_endpoint: + raise NamoIDError( + "The issuer does not advertise RP-initiated logout", code="logout_unavailable" + ) + params = {"id_token_hint": id_token_hint} + if post_logout_redirect_uri: + params["post_logout_redirect_uri"] = post_logout_redirect_uri + if state: + params["state"] = state + parts = urlsplit(discovery.end_session_endpoint) + return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(params), "")) + + +def validate_id_token( + id_token: str, + *, + jwks: Mapping[str, Any], + issuer: str, + client_id: str, + nonce: str, + clock_tolerance_seconds: int = 30, +) -> Mapping[str, Any]: + """Verify an RS256 ID token and its callback-bound nonce.""" + # Keep ordinary URL construction and public-client use lightweight. JOSE is + # imported only when a server actually validates an ID token. + from joserfc import jwk, jwt + from joserfc.errors import JoseError + + try: + keys = jwk.KeySet.import_key_set(dict(jwks)) + decoded = jwt.decode(id_token, keys, algorithms=["RS256"]) + registry = jwt.JWTClaimsRegistry( + leeway=clock_tolerance_seconds, + iss={"essential": True, "value": issuer.rstrip("/")}, + aud={"essential": True, "value": client_id}, + sub={"essential": True}, + iat={"essential": True}, + exp={"essential": True}, + nonce={"essential": True, "value": nonce}, + ) + registry.validate(decoded.claims) + except (JoseError, ValueError, TypeError, KeyError) as exc: + raise NamoIDError("ID token validation failed", code="invalid_id_token") from exc + return dict(decoded.claims) + + +def _normalize_issuer(value: str) -> str: + return value.rstrip("/") + + +def _require_trusted_endpoint(issuer: str, endpoint: str) -> None: + issuer_parts = urlsplit(issuer) + endpoint_parts = urlsplit(endpoint) + if endpoint_parts.scheme != issuer_parts.scheme or endpoint_parts.netloc != issuer_parts.netloc: + raise NamoIDError("OIDC endpoint is outside the configured issuer", code="untrusted_oidc_endpoint") diff --git a/tests/test_hosted_auth.py b/tests/test_hosted_auth.py index 0683f6c..33de099 100644 --- a/tests/test_hosted_auth.py +++ b/tests/test_hosted_auth.py @@ -4,10 +4,12 @@ import base64 import hashlib +import time from urllib.parse import parse_qs, urlsplit import httpx import pytest +from joserfc import jwk, jwt from namoid import ( AsyncNamoIDClient, @@ -57,6 +59,28 @@ "user_id": "11111111-1111-1111-1111-111111111111", } +DISCOVERY_PAYLOAD = { + "issuer": CONFIG_PAYLOAD["issuer"], + "authorization_endpoint": f'{CONFIG_PAYLOAD["issuer"]}/oauth/authorize', + "token_endpoint": f'{CONFIG_PAYLOAD["issuer"]}/v1/oauth/token', + "userinfo_endpoint": f'{CONFIG_PAYLOAD["issuer"]}/v1/oauth/userinfo', + "jwks_uri": f'{CONFIG_PAYLOAD["issuer"]}/v1/oauth/jwks.json', + "revocation_endpoint": f'{CONFIG_PAYLOAD["issuer"]}/v1/oauth/revoke', + "end_session_endpoint": f'{CONFIG_PAYLOAD["issuer"]}/oauth/logout', + "code_challenge_methods_supported": ["S256"], +} + + +def oidc_handler(final): + def handler(request): + if request.url.path == "/v1/auth/config": + return httpx.Response(200, json=CONFIG_PAYLOAD) + if request.url.path == "/.well-known/openid-configuration": + return httpx.Response(200, json=DISCOVERY_PAYLOAD) + return final(request) + + return handler + def recorder(handler): """Collect the requests a client makes, alongside a response handler.""" @@ -251,39 +275,45 @@ def test_builds_the_hosted_url_from_the_fetched_config(): def test_exchanges_a_code_with_a_client_secret(): - import json - seen, client = sync_client( - lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD), client_secret=CLIENT_SECRET + oidc_handler(lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD)), + client_secret=CLIENT_SECRET, ) - tokens = client.exchange_code(code="c" * 40, code_verifier="v" * 50) + tokens = client.exchange_code( + code="c" * 40, + code_verifier="v" * 50, + redirect_uri="https://app.example/callback", + ) assert tokens.access_token == "access-token-value" assert tokens.refresh_token == "refresh-token-value" assert tokens.expires_in == 900 assert tokens.user_id == "11111111-1111-1111-1111-111111111111" - body = json.loads(seen[0].content) - assert seen[0].url.path == "/v1/auth/hosted/exchange" - assert body["client_secret"] == CLIENT_SECRET - assert body["code_verifier"] == "v" * 50 - # Unset optional fields are omitted rather than sent as null. - assert "device_id" not in body + request = seen[-1] + body = parse_qs(request.content.decode()) + assert request.url.path == "/v1/oauth/token" + assert request.headers["authorization"].startswith("Basic ") + assert body["grant_type"] == ["authorization_code"] + assert body["redirect_uri"] == ["https://app.example/callback"] + assert body["code_verifier"] == ["v" * 50] def test_public_exchange_sends_no_secret(): - import json - - seen, client = sync_client(lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD)) - client.exchange_code(code="c" * 40, code_verifier="v" * 50, confidential=False) - assert "client_secret" not in json.loads(seen[0].content) + seen, client = sync_client( + oidc_handler(lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD)) + ) + client.exchange_code(code="c" * 40, code_verifier="v" * 50, + redirect_uri="https://app.example/callback", confidential=False) + assert "authorization" not in seen[-1].headers def test_confidential_calls_refuse_to_run_without_a_secret(): _seen, client = sync_client(lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD)) with pytest.raises(NamoIDError, match="client_secret is required"): - client.exchange_code(code="c" * 40) + client.exchange_code(code="c" * 40, code_verifier="v" * 50, + redirect_uri="https://app.example/callback", confidential=True) with pytest.raises(NamoIDError, match="client_secret is required"): client.validate_access_token("token") @@ -313,10 +343,81 @@ def test_validates_an_access_token(): def test_refreshes_a_session(): - seen, client = sync_client(lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD)) + seen, client = sync_client( + oidc_handler(lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD)) + ) tokens = client.refresh("refresh-token-value") assert tokens.access_token == "access-token-value" - assert seen[0].url.path == "/v1/auth/refresh" + assert seen[-1].url.path == "/v1/oauth/token" + assert parse_qs(seen[-1].content.decode())["grant_type"] == ["refresh_token"] + + +def test_builds_standard_authorization_url_with_nonce_and_pkce(): + seen, client = sync_client(oidc_handler(lambda _r: httpx.Response(404))) + transaction = client.create_oidc_transaction("https://app.example/callback") + url = client.authorization_url(transaction) + query = parse_qs(urlsplit(url).query) + + assert urlsplit(url).path == "/oauth/authorize" + assert query["client_id"] == [CLIENT_ID] + assert query["redirect_uri"] == [transaction.redirect_uri] + assert query["state"] == [transaction.state] + assert query["nonce"] == [transaction.nonce] + assert query["code_challenge_method"] == ["S256"] + assert transaction.code_verifier not in url + assert [request.url.path for request in seen] == [ + "/v1/auth/config", "/.well-known/openid-configuration" + ] + + +def test_userinfo_revocation_and_logout_use_discovered_endpoints(): + def final(request): + if request.url.path == "/v1/oauth/userinfo": + return httpx.Response(200, json={"sub": "user-1", "email": "user@example.com"}) + if request.url.path == "/v1/oauth/revoke": + return httpx.Response(200) + return httpx.Response(404) + + seen, client = sync_client(oidc_handler(final), client_secret=CLIENT_SECRET) + assert client.user_info("access-token-value")["sub"] == "user-1" + client.revoke_token("refresh-token-value", token_type_hint="refresh_token") + logout = client.logout_url( + id_token_hint="id-token-value", + post_logout_redirect_uri="https://app.example/signed-out", + state="logout-state", + ) + + assert seen[-2].headers["authorization"] == "Bearer access-token-value" + revocation = seen[-1] + assert revocation.url.path == "/v1/oauth/revoke" + assert revocation.headers["authorization"].startswith("Basic ") + assert parse_qs(revocation.content.decode())["token_type_hint"] == ["refresh_token"] + logout_query = parse_qs(urlsplit(logout).query) + assert urlsplit(logout).path == "/oauth/logout" + assert logout_query["post_logout_redirect_uri"] == ["https://app.example/signed-out"] + assert logout_query["state"] == ["logout-state"] + + +def test_validates_id_token_signature_claims_and_nonce(): + key = jwk.RSAKey.generate_key(2048, parameters={"kid": "id-key", "alg": "RS256"}) + now = int(time.time()) + token = jwt.encode( + {"alg": "RS256", "kid": "id-key"}, + {"iss": CONFIG_PAYLOAD["issuer"], "aud": CLIENT_ID, "sub": "user-1", + "iat": now, "exp": now + 300, "nonce": "expected-nonce"}, + key, + ) + + def final(request): + if request.url.path == "/v1/oauth/jwks.json": + return httpx.Response(200, json={"keys": [key.as_dict(private=False)]}) + return httpx.Response(404) + + _seen, client = sync_client(oidc_handler(final)) + assert client.validate_id_token(token, nonce="expected-nonce")["sub"] == "user-1" + with pytest.raises(NamoIDError) as excinfo: + client.validate_id_token(token, nonce="attacker-nonce") + assert excinfo.value.code == "invalid_id_token" def test_revokes_a_session_and_tolerates_an_empty_204(): @@ -329,14 +430,15 @@ def test_revokes_a_session_and_tolerates_an_empty_204(): def test_surfaces_the_api_error_message_and_code(): _seen, client = sync_client( - lambda _r: httpx.Response( + oidc_handler(lambda _r: httpx.Response( 400, json={"error": "invalid_grant", "message": "authorization code expired"} - ), + )), client_secret=CLIENT_SECRET, ) with pytest.raises(NamoIDError) as excinfo: - client.exchange_code(code="c" * 40) + client.exchange_code(code="c" * 40, code_verifier="v" * 50, + redirect_uri="https://app.example/callback") error = excinfo.value assert "authorization code expired" in str(error) @@ -346,7 +448,7 @@ def test_surfaces_the_api_error_message_and_code(): def test_falls_back_to_a_generic_message_for_an_opaque_failure(): - _seen, client = sync_client(lambda _r: httpx.Response(502, text="upstream boom")) + _seen, client = sync_client(oidc_handler(lambda _r: httpx.Response(502, text="upstream boom"))) with pytest.raises(NamoIDError) as excinfo: client.refresh("refresh-token-value") assert excinfo.value.status == 502 @@ -365,7 +467,9 @@ def explode(_request): def test_rejects_a_token_response_without_an_access_token(): - _seen, client = sync_client(lambda _r: httpx.Response(200, json={"expires_in": 900})) + _seen, client = sync_client( + oidc_handler(lambda _r: httpx.Response(200, json={"expires_in": 900})) + ) with pytest.raises(NamoIDError, match="did not include an access_token"): client.refresh("refresh-token-value") @@ -383,7 +487,9 @@ async def test_async_client_mirrors_the_sync_one(): def handler(request: httpx.Response) -> httpx.Response: if request.url.path == "/v1/auth/config": return httpx.Response(200, json=CONFIG_PAYLOAD) - if request.url.path == "/v1/auth/hosted/exchange": + if request.url.path == "/.well-known/openid-configuration": + return httpx.Response(200, json=DISCOVERY_PAYLOAD) + if request.url.path == "/v1/oauth/token": return httpx.Response(200, json=TOKEN_PAYLOAD) if request.url.path == "/v1/auth/logout": return httpx.Response(204) @@ -401,14 +507,18 @@ def handler(request: httpx.Response) -> httpx.Response: ) assert parse_qs(urlsplit(url).query)["client_id"] == [CLIENT_ID] - tokens = await client.exchange_code(code="c" * 40, code_verifier="v" * 50) + tokens = await client.exchange_code( + code="c" * 40, code_verifier="v" * 50, + redirect_uri="https://app.example/callback" + ) assert tokens.access_token == "access-token-value" await client.revoke_session(access_token=tokens.access_token) assert [r.url.path for r in seen] == [ "/v1/auth/config", - "/v1/auth/hosted/exchange", + "/.well-known/openid-configuration", + "/v1/oauth/token", "/v1/auth/logout", ]