"""The OAuth 2.0 authorization-code client and its value objects.
:class:`OAuthClient` drives the three server-side legs of the
authorization-code flow against a
:class:`~fsh_lib.oauth.providers.ProviderConfig`:
#. :meth:`OAuthClient.authorization_url` (or :meth:`OAuthClient.begin`)
builds the URL the user-agent is redirected to.
#. :meth:`OAuthClient.exchange_code` swaps the returned ``code`` for a
:class:`TokenResponse`.
#. :meth:`OAuthClient.refresh` trades a refresh token for a fresh one.
:meth:`OAuthClient.fetch_userinfo` is the optional OIDC UserInfo
call.
The leg *between* the redirect and the callback -- carrying the CSRF
``state``, the PKCE verifier, and the OIDC ``nonce`` -- is handled by
:mod:`fsh_lib.oauth.state`; ID-token verification by
:mod:`fsh_lib.oauth.oidc`.
Requires the ``oauth`` extra (``pip install 'fsh-lib[oauth]'``)
for :mod:`httpx`.
"""
from __future__ import annotations
import base64
import datetime
import hashlib
import os
import secrets
from typing import TYPE_CHECKING, Any, Self
from urllib.parse import urlencode, urlsplit, urlunsplit
import httpx
from pydantic import BaseModel, ConfigDict, Field, ValidationError
if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
from fsh_lib.oauth.providers import ClientCredentials, ProviderConfig
#: Bytes of CSPRNG entropy behind a PKCE verifier / a ``state`` /
#: a ``nonce``. 32 bytes -> a 43-character base64url string, the
#: minimum a 256-bit value needs and comfortably inside RFC 7636's
#: 43--128 verifier range.
_ENTROPY_BYTES = 32
#: Default per-request timeout, in seconds, for the token / userinfo
#: round-trips -- short, since they sit on an interactive login path.
_DEFAULT_TIMEOUT = 10.0
[docs]
class OAuthError(RuntimeError):
"""An OAuth round-trip failed.
Raised when the provider was unreachable, timed out, or answered
a token / userinfo request with a non-2xx status. ID-token
verification raises this too (see :mod:`fsh_lib.oauth.oidc`).
"""
[docs]
class TokenResponse(BaseModel):
"""The parsed body of a successful token-endpoint response.
Attributes:
access_token: The bearer token used to call the provider's
API.
token_type: The token type the provider returned, normally
``"Bearer"``.
expires_in: Lifetime of :attr:`access_token` in seconds, or
``None`` when the provider omits it.
refresh_token: A token to mint a new :attr:`access_token`
without user interaction, or ``None``. A refresh
response often omits it -- keep the previous one in that
case.
scope: Space-delimited scopes actually granted, or ``None``.
id_token: The OIDC ID token (a signed JWT), present only when
``openid`` was among the requested scopes.
raw: The full decoded response body, for provider-specific
fields (e.g. Intuit's ``x_refresh_token_expires_in``).
"""
model_config = ConfigDict(frozen=True, extra="ignore")
access_token: str
token_type: str = "Bearer" # noqa: S105 -- a token type, not a secret
expires_in: int | None = None
refresh_token: str | None = None
scope: str | None = None
id_token: str | None = None
raw: dict[str, Any] = Field(default_factory=dict)
[docs]
def expires_at(
self,
now: datetime.datetime,
) -> datetime.datetime | None:
"""Return the absolute expiry of :attr:`access_token`.
Args:
now: The reference time the relative :attr:`expires_in`
is measured from -- normally the moment the token was
issued. Pass an aware datetime.
Returns:
``now + expires_in`` seconds, or ``None`` when the
provider didn't send :attr:`expires_in`.
"""
if self.expires_in is None:
return None
return now + datetime.timedelta(seconds=self.expires_in)
[docs]
class AuthorizationState(BaseModel):
"""The per-attempt secrets that bridge the redirect and callback.
Generated by :meth:`OAuthClient.begin`, stashed by the caller
(see :mod:`fsh_lib.oauth.state` for the stateless signed-cookie
helper), and checked when the provider redirects back.
Attributes:
state: The CSRF token echoed back as the ``state`` query
param; compared against the stored value on callback.
code_verifier: The PKCE verifier whose challenge went out in
the authorize URL; replayed to
:meth:`OAuthClient.exchange_code`. ``None`` when PKCE
is disabled.
nonce: The OIDC nonce embedded in the authorize URL; matched
against the ID token's ``nonce`` claim. ``None`` when
not requested.
"""
model_config = ConfigDict(frozen=True)
state: str
code_verifier: str | None = None
nonce: str | None = None
def _b64url(raw: bytes) -> str:
"""Return *raw* as unpadded base64url text (the OAuth token form)."""
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
[docs]
def generate_state() -> str:
"""Return a fresh CSRF ``state`` value (256 bits, base64url)."""
return _b64url(secrets.token_bytes(_ENTROPY_BYTES))
[docs]
def generate_pkce() -> tuple[str, str]:
"""Return a ``(code_verifier, code_challenge)`` PKCE pair.
The verifier is a 256-bit CSPRNG value; the challenge is its
SHA-256 digest, base64url-encoded (the ``S256`` method from RFC
7636). Keep the verifier server-side and send only the
challenge in the authorize URL.
"""
verifier = _b64url(secrets.token_bytes(_ENTROPY_BYTES))
challenge = _b64url(hashlib.sha256(verifier.encode("ascii")).digest())
return verifier, challenge
[docs]
class OAuthClient:
"""Async client for one provider + one registered client.
Wraps an :class:`httpx.AsyncClient`. Construct one per
``(provider, credentials)`` pair (it is cheap to keep open) and
close it on shutdown via :meth:`aclose`, or use it as an async
context manager.
Args:
provider: The provider's endpoints and conventions.
credentials: The registered client's id / redirect URI and
the *name* of the env var holding its secret.
client: An existing :class:`httpx.AsyncClient` to reuse.
When ``None`` (the default) one is created and owned, and
:meth:`aclose` closes it.
timeout: Per-request timeout in seconds. Ignored when an
explicit *client* is supplied.
"""
def __init__(
self,
provider: ProviderConfig,
credentials: ClientCredentials,
*,
client: httpx.AsyncClient | None = None,
timeout: float = _DEFAULT_TIMEOUT,
) -> None:
"""Store the config and the (owned or borrowed) client."""
self._provider = provider
self._credentials = credentials
self._owns_client = client is None
self._client = client or httpx.AsyncClient(timeout=timeout)
async def __aenter__(self) -> Self:
"""Enter the async context, returning ``self``."""
return self
async def __aexit__(self, *_exc: object) -> None:
"""Close an owned client on context exit."""
await self.aclose()
[docs]
async def aclose(self) -> None:
"""Close the underlying client iff this instance owns it."""
if self._owns_client:
await self._client.aclose()
[docs]
def authorization_url(
self,
*,
state: str,
scopes: Sequence[str] | None = None,
code_challenge: str | None = None,
nonce: str | None = None,
extra_params: Mapping[str, str] | None = None,
) -> str:
"""Build the URL the user-agent is redirected to for consent.
Args:
state: The CSRF token to round-trip (see
:func:`generate_state`).
scopes: Scopes to request; falls back to the provider's
:attr:`~fsh_lib.oauth.providers.ProviderConfig.default_scopes`
when ``None``.
code_challenge: A PKCE ``S256`` challenge (see
:func:`generate_pkce`). Omitted from the URL when
``None``.
nonce: An OIDC nonce to bind to the ID token. Omitted
when ``None``.
extra_params: Provider-specific query params to add
verbatim, e.g. ``{"access_type": "offline"}`` /
``{"prompt": "consent"}`` for Google to force a
refresh token, or ``{"login_hint": email}``.
Returns:
The fully-formed authorization URL.
"""
chosen = (
tuple(scopes)
if scopes is not None
else (self._provider.default_scopes)
)
params: dict[str, str] = {
"response_type": "code",
"client_id": self._credentials.client_id,
"redirect_uri": self._credentials.redirect_uri,
"scope": " ".join(chosen),
"state": state,
}
if code_challenge is not None:
params["code_challenge"] = code_challenge
params["code_challenge_method"] = "S256"
if nonce is not None:
params["nonce"] = nonce
if extra_params:
params.update(extra_params)
return _with_query(self._provider.authorization_endpoint, params)
[docs]
def begin(
self,
*,
scopes: Sequence[str] | None = None,
use_pkce: bool = True,
use_nonce: bool = False,
extra_params: Mapping[str, str] | None = None,
) -> tuple[str, AuthorizationState]:
"""Start a flow: mint the secrets and build the authorize URL.
Convenience over :meth:`authorization_url` that generates the
``state`` (always), a PKCE pair (when *use_pkce*), and an
OIDC ``nonce`` (when *use_nonce*), returning both the URL to
redirect to and the :class:`AuthorizationState` to stash for
the callback.
Args:
scopes: Scopes to request (see
:meth:`authorization_url`).
use_pkce: Generate and attach a PKCE challenge. On by
default -- recommended for every flow.
use_nonce: Generate and attach an OIDC ``nonce``. Turn
on for social-login (OIDC) flows so the ID token can
be bound to this request.
extra_params: Passed through to
:meth:`authorization_url`.
Returns:
``(authorization_url, authorization_state)``.
"""
verifier, challenge = generate_pkce() if use_pkce else (None, None)
nonce = generate_state() if use_nonce else None
state = generate_state()
url = self.authorization_url(
state=state,
scopes=scopes,
code_challenge=challenge,
nonce=nonce,
extra_params=extra_params,
)
return url, AuthorizationState(
state=state,
code_verifier=verifier,
nonce=nonce,
)
[docs]
async def exchange_code(
self,
code: str,
*,
code_verifier: str | None = None,
redirect_uri: str | None = None,
) -> TokenResponse:
"""Exchange an authorization ``code`` for tokens.
Args:
code: The authorization code from the callback's ``code``
query param.
code_verifier: The PKCE verifier from the stored
:class:`AuthorizationState`; required iff the
authorize request carried a challenge.
redirect_uri: Overrides the credentials' redirect URI for
the (rare) provider that wants a different value here;
defaults to the registered one.
Returns:
The parsed :class:`TokenResponse`.
Raises:
OAuthError: The provider was unreachable, timed out, or
rejected the exchange.
"""
data = {
"grant_type": "authorization_code",
"code": code,
"redirect_uri": redirect_uri or self._credentials.redirect_uri,
}
if code_verifier is not None:
data["code_verifier"] = code_verifier
return await self._token_request(data)
[docs]
async def refresh(self, refresh_token: str) -> TokenResponse:
"""Trade a refresh token for a fresh access token.
Args:
refresh_token: The stored refresh token.
Returns:
The parsed :class:`TokenResponse`. Providers often omit
``refresh_token`` here -- keep the previous one when the
response's :attr:`TokenResponse.refresh_token` is
``None``.
Raises:
OAuthError: The provider was unreachable, timed out, or
rejected the refresh (e.g. a revoked grant).
"""
return await self._token_request(
{
"grant_type": "refresh_token",
"refresh_token": refresh_token,
},
)
[docs]
async def fetch_userinfo(self, access_token: str) -> dict[str, Any]:
"""Fetch the OIDC UserInfo claims for *access_token*.
Args:
access_token: A token granted the ``openid`` (and usually
``profile`` / ``email``) scopes.
Returns:
The UserInfo claims as a dict.
Raises:
OAuthError: No UserInfo endpoint is configured, or the
request failed.
"""
if self._provider.userinfo_endpoint is None:
msg = "provider has no userinfo_endpoint configured"
raise OAuthError(msg)
try:
response = await self._client.get(
self._provider.userinfo_endpoint,
headers={"Authorization": f"Bearer {access_token}"},
)
response.raise_for_status()
claims: dict[str, Any] = response.json()
except httpx.HTTPError as exc:
msg = f"userinfo request failed: {exc}"
raise OAuthError(msg) from exc
except ValueError as exc: # json.JSONDecodeError
msg = "userinfo response is not valid JSON"
raise OAuthError(msg) from exc
return claims
async def _token_request(
self,
data: dict[str, str],
) -> TokenResponse:
"""POST *data* to the token endpoint and parse the response.
Applies the provider's client-authentication method: Basic
``Authorization`` for ``client_secret_basic``, ``client_id``
/ ``client_secret`` form fields for ``client_secret_post``.
Raises:
OAuthError: The request failed or the client secret is
required but its env var is unset.
"""
body = dict(data)
secret = self._client_secret()
# ``auth`` is only added for the Basic case -- omitting the
# key entirely lets httpx apply its client default (no auth)
# rather than passing a ``None`` the post() signature rejects.
extra: dict[str, Any] = {}
auth_method = self._provider.token_endpoint_auth_method
if auth_method == "client_secret_basic":
if secret is None:
msg = "client_secret is required for client_secret_basic"
raise OAuthError(msg)
extra["auth"] = (self._credentials.client_id, secret)
else:
body["client_id"] = self._credentials.client_id
if secret is not None:
body["client_secret"] = secret
try:
response = await self._client.post(
self._provider.token_endpoint,
data=body,
headers={"Accept": "application/json"},
**extra,
)
response.raise_for_status()
payload = response.json()
except httpx.HTTPError as exc:
msg = f"token request failed: {exc}"
raise OAuthError(msg) from exc
except ValueError as exc: # json.JSONDecodeError
msg = "token response is not valid JSON"
raise OAuthError(msg) from exc
if not isinstance(payload, dict):
msg = "token response is not a JSON object"
raise OAuthError(msg)
try:
return TokenResponse.model_validate({**payload, "raw": payload})
except ValidationError as exc:
msg = f"malformed token response: {exc}"
raise OAuthError(msg) from exc
def _client_secret(self) -> str | None:
"""Return the client secret from its env var, or ``None``.
``None`` when no ``client_secret_env`` was configured (a
public / PKCE-only client) *or* when the named env var is
unset.
"""
env = self._credentials.client_secret_env
if env is None:
return None
return os.environ.get(env)
def _with_query(url: str, params: Mapping[str, str]) -> str:
"""Return *url* with *params* merged into its query string.
Preserves any query already present on *url* (some providers
publish authorize endpoints with fixed params) and appends the
new pairs after it.
"""
parts = urlsplit(url)
existing = parts.query
encoded = urlencode(params)
query = f"{existing}&{encoded}" if existing else encoded
return urlunsplit(
(parts.scheme, parts.netloc, parts.path, query, parts.fragment),
)