"""OpenID Connect ID-token verification for social login.
The social-login (OIDC) case ends with the provider handing back an
*ID token* -- a JWT, signed with the provider's rotating RSA keys,
asserting who the user is. :class:`IdTokenVerifier` checks that
token end to end:
* signature, against the provider's published JWKS (keys fetched and
cached by :class:`jwt.PyJWKClient`);
* ``aud``, against the registered ``client_id``;
* ``iss``, against the provider's issuer (when one is configured);
* ``exp`` / ``iat``, with a small clock-skew leeway;
* ``nonce``, against the value bound into the authorize request.
A verified token yields an :class:`OidcIdentity`; the consumer maps
that to its own user record and mints a first-party session with
:func:`fsh_lib.auth.issue_session` -- the provider's tokens are not
kept.
Typical callback wiring (after :func:`fsh_lib.oauth.state.verify_state`)::
tokens = await client.exchange_code(code, code_verifier=flow.code_verifier)
identity = await verifier.verify(tokens.id_token, nonce=flow.nonce)
user = await users.upsert_from_oidc(identity) # consumer's mapping
token_id = await tokens_store.mint(user, source=TokenSource.OAUTH)
return issue_session(response, token_id, ...)
Requires the ``oauth`` extra (``pip install 'fsh-lib[oauth]'``)
-- ``cryptography`` (pulled in by the extra) backs pyjwt's RS256
verification.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Self
import anyio
import jwt
from jwt import PyJWKClient
from pydantic import BaseModel, ConfigDict, Field
from fsh_lib.oauth.client import OAuthError
if TYPE_CHECKING:
from collections.abc import Sequence
from fsh_lib.oauth.providers import ProviderConfig
#: ID tokens from the mainstream IdPs (Google, Microsoft, Intuit)
#: are RS256-signed. Restricting the accepted set here -- rather
#: than trusting the token's own ``alg`` header -- closes the
#: algorithm-confusion class of attacks.
DEFAULT_ALGORITHMS: tuple[str, ...] = ("RS256",)
[docs]
class OidcIdentity(BaseModel):
"""The identity asserted by a verified ID token.
Attributes:
sub: The provider's stable, unique subject identifier -- the
key to map onto a local user (never the email, which can
change hands).
email: The user's email, when the ``email`` scope was
granted.
email_verified: Whether the provider vouches the email is
verified. Treat a missing value as "unknown", not
"verified".
name: The user's display name, when ``profile`` was granted.
raw: All decoded claims, for anything beyond the common four.
"""
model_config = ConfigDict(frozen=True, extra="ignore")
sub: str
email: str | None = None
email_verified: bool | None = None
name: str | None = None
raw: dict[str, Any] = Field(default_factory=dict)
[docs]
class IdTokenVerifier:
"""Verifies OIDC ID tokens against one provider's keys + audience.
Construct one per ``(provider, client_id)`` and reuse it: the
underlying :class:`jwt.PyJWKClient` caches the provider's signing
keys, so steady-state verification needs no network call.
Args:
jwks_uri: The provider's JWKS URL (from
:attr:`~fsh_lib.oauth.providers.ProviderConfig.jwks_uri`).
audience: The expected ``aud`` -- the registered
``client_id``.
issuer: The expected ``iss``, or ``None`` to skip issuer
validation (e.g. Microsoft's multi-tenant ``common``
endpoint, whose ``iss`` is per-tenant).
algorithms: Accepted signature algorithms. Defaults to
``("RS256",)``; widen only if a provider demands it.
leeway: Clock-skew tolerance in seconds applied to ``exp`` /
``iat`` checks.
"""
def __init__(
self,
*,
jwks_uri: str,
audience: str,
issuer: str | None = None,
algorithms: Sequence[str] = DEFAULT_ALGORITHMS,
leeway: float = 0.0,
) -> None:
"""Store the verification parameters and the JWKS client."""
self._audience = audience
self._issuer = issuer
self._algorithms = list(algorithms)
self._leeway = leeway
self._jwk_client = PyJWKClient(jwks_uri)
[docs]
@classmethod
def from_provider(
cls,
provider: ProviderConfig,
*,
client_id: str,
algorithms: Sequence[str] = DEFAULT_ALGORITHMS,
leeway: float = 0.0,
) -> Self:
"""Build a verifier from a provider config and client id.
Args:
provider: A :class:`~fsh_lib.oauth.providers.ProviderConfig`
carrying a non-``None`` ``jwks_uri`` (an OIDC
provider).
client_id: The registered client id, used as the expected
``aud``.
algorithms: Accepted signature algorithms.
leeway: Clock-skew tolerance in seconds.
Returns:
A configured :class:`IdTokenVerifier`.
Raises:
OAuthError: The provider has no ``jwks_uri`` -- it can't
be used for OIDC ID-token verification.
"""
if provider.jwks_uri is None:
msg = "provider has no jwks_uri; cannot verify ID tokens"
raise OAuthError(msg)
return cls(
jwks_uri=provider.jwks_uri,
audience=client_id,
issuer=provider.issuer,
algorithms=algorithms,
leeway=leeway,
)
[docs]
async def verify(
self,
id_token: str | None,
*,
nonce: str | None = None,
) -> OidcIdentity:
"""Verify *id_token* and return the asserted identity.
Args:
id_token: The raw ID-token JWT from
:attr:`~fsh_lib.oauth.client.TokenResponse.id_token`.
nonce: The nonce from the stored
:class:`~fsh_lib.oauth.client.AuthorizationState`.
When given, the token's ``nonce`` claim must equal it;
pass it whenever the authorize request set one.
Returns:
The verified :class:`OidcIdentity`.
Raises:
OAuthError: The token is missing, its signature / ``aud``
/ ``iss`` / ``exp`` are invalid, the signing key
can't be fetched, or the ``nonce`` doesn't match.
"""
if not id_token:
msg = "no id_token to verify (was 'openid' among the scopes?)"
raise OAuthError(msg)
try:
# PyJWKClient fetches/caches keys over the network -- run
# the blocking call off the event loop.
signing_key = await anyio.to_thread.run_sync(
self._jwk_client.get_signing_key_from_jwt,
id_token,
)
# ``audience`` already forces the ``aud`` claim to be
# present and to match; ``verify_exp`` (on by default)
# checks ``exp`` when present.
claims: dict[str, Any] = jwt.decode(
id_token,
signing_key.key,
algorithms=self._algorithms,
audience=self._audience,
issuer=self._issuer,
leeway=self._leeway,
)
except jwt.PyJWTError as exc:
msg = f"ID token verification failed: {exc}"
raise OAuthError(msg) from exc
# Require an expiry to be present -- an ID token with no
# ``exp`` would otherwise never be treated as expired.
if "exp" not in claims:
msg = "ID token has no exp claim"
raise OAuthError(msg)
# The nonce is application state, not a registered claim, so
# jwt.decode can't check it -- compare it ourselves.
if nonce is not None and claims.get("nonce") != nonce:
msg = "ID token nonce mismatch"
raise OAuthError(msg)
return OidcIdentity.model_validate({**claims, "raw": claims})