"""Stateless carrier for the redirect->callback flow secrets.
The CSRF ``state``, PKCE ``code_verifier``, and OIDC ``nonce`` an
:class:`~fsh_lib.oauth.client.AuthorizationState` holds must survive
the round-trip to the provider and back, with no server-side
session. These helpers pack that state into a short-lived, signed,
``httpOnly`` cookie -- reusing :mod:`fsh_lib.auth`'s
:func:`~fsh_lib.auth.encode_jwt` / :func:`~fsh_lib.auth.decode_jwt`
so the same ``JWT_SECRET`` and algorithm protect it.
The cookie is signed (tamper-evident) and ``httpOnly`` (out of
reach of page JS), and its short TTL bounds how long a captured
authorize URL stays replayable. The PKCE verifier inside it is the
defence when the *authorize URL* (state + challenge) leaks: the
verifier never leaves the cookie, so the stolen ``state`` can't
complete a token exchange. A leaked cookie, by contrast, hands an
attacker the whole flow -- the httpOnly/Secure flags and the short
TTL are the defences on that side.
The token is signed with the same secret/algorithm as the session
JWTs but stamped with the reserved
:data:`fsh_lib.auth.FLOW_STATE_PURPOSE` claim, so the two token
kinds can't substitute for each other: :func:`read_flow_state`
requires the claim, and :func:`fsh_lib.auth.session_auth` rejects
any token carrying it.
Typical wiring::
# login route
url, flow = client.begin(scopes=[...], use_pkce=True, use_nonce=True)
response = RedirectResponse(url)
issue_flow_state(
response, flow,
secret_env="JWT_SECRET", algorithm="HS256",
cookie_name="oauth_flow",
)
return response
# callback route
flow = read_flow_state(
request.cookies.get("oauth_flow"),
secret_env="JWT_SECRET", algorithm="HS256",
)
verify_state(received=request.query_params["state"], stored=flow.state)
tokens = await client.exchange_code(
code, code_verifier=flow.code_verifier,
)
"""
from __future__ import annotations
import datetime
import secrets
from typing import TYPE_CHECKING
from fastapi import HTTPException, status
from pydantic import ValidationError
from fsh_lib.auth import FLOW_STATE_PURPOSE, decode_jwt, encode_jwt
from fsh_lib.oauth.client import AuthorizationState
if TYPE_CHECKING:
from fastapi import Response
from fsh_lib.auth import SameSite
#: Default lifetime of the flow-state cookie. Long enough for a
#: user to sign in at the provider, short enough to bound replay of
#: a captured authorize URL.
DEFAULT_FLOW_TTL = datetime.timedelta(minutes=10)
def _bad_request(detail: str) -> HTTPException:
"""400 for a missing, malformed, or mismatched flow state.
A small factory so call sites stay one line; the message is a
parameter (not a literal at the ``raise``), keeping the exception
text out of the constructor call the linters flag.
"""
return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail)
[docs]
def issue_flow_state(
response: Response,
state: AuthorizationState,
*,
secret_env: str,
algorithm: str,
cookie_name: str,
ttl: datetime.timedelta = DEFAULT_FLOW_TTL,
cookie_secure: bool = True,
cookie_samesite: SameSite = "lax",
) -> None:
"""Sign *state* into a short-lived ``httpOnly`` cookie on *response*.
Args:
response: The redirect response the cookie is attached to.
state: The per-attempt secrets from
:meth:`~fsh_lib.oauth.client.OAuthClient.begin`.
secret_env: Name of the env var holding the signing secret
(the same one :mod:`fsh_lib.auth` uses).
algorithm: JWT signing algorithm, e.g. ``"HS256"``.
cookie_name: Name of the cookie to set; pass the same name to
:func:`read_flow_state` / :func:`clear_flow_state`.
ttl: Cookie lifetime; also the token's ``exp``. Defaults to
:data:`DEFAULT_FLOW_TTL`.
cookie_secure: Set the ``Secure`` attribute. Leave ``True``
outside local development.
cookie_samesite: ``SameSite`` policy. ``"lax"`` is correct
for the top-level redirect the provider performs back to
the callback.
"""
token = encode_jwt(
{**state.model_dump(), "purpose": FLOW_STATE_PURPOSE},
secret_env=secret_env,
algorithm=algorithm,
ttl=ttl,
)
response.set_cookie(
key=cookie_name,
value=token,
max_age=int(ttl.total_seconds()),
httponly=True,
secure=cookie_secure,
samesite=cookie_samesite,
)
[docs]
def read_flow_state(
cookie_value: str | None,
*,
secret_env: str,
algorithm: str,
) -> AuthorizationState:
"""Decode the flow-state cookie back into an AuthorizationState.
Args:
cookie_value: The raw cookie value from the callback request
(``request.cookies.get(cookie_name)``), or ``None`` when
absent.
secret_env: Name of the signing-secret env var (must match
:func:`issue_flow_state`).
algorithm: JWT algorithm (must match).
Returns:
The validated :class:`~fsh_lib.oauth.client.AuthorizationState`.
Raises:
HTTPException: 400 when the cookie is missing, expired,
tampered with, or otherwise un-decodable -- a callback
without a valid flow cookie is a bad request, never a
silent pass.
"""
if not cookie_value:
detail = "Missing OAuth flow state"
raise _bad_request(detail)
try:
claims = decode_jwt(
cookie_value,
secret_env=secret_env,
algorithm=algorithm,
)
except HTTPException as exc:
# decode_jwt raises 401 for an invalid/expired token; for a
# callback the right signal is "bad request", not
# "unauthenticated".
detail = "Invalid OAuth flow state"
raise _bad_request(detail) from exc
# Only accept tokens minted by issue_flow_state -- a session JWT
# (same secret, no purpose claim) must not pass as flow state.
if claims.get("purpose") != FLOW_STATE_PURPOSE:
detail = "Invalid OAuth flow state"
raise _bad_request(detail)
try:
return AuthorizationState.model_validate(claims)
except ValidationError as exc:
# A validly-signed token whose claims don't fit the model is
# still a bad callback, not a server error.
detail = "Invalid OAuth flow state"
raise _bad_request(detail) from exc
[docs]
def verify_state(*, received: str | None, stored: str) -> None:
"""Constant-time check that the callback ``state`` matches.
Args:
received: The ``state`` query param the provider echoed back
(``None`` when absent).
stored: The ``state`` from the decoded flow cookie.
Raises:
HTTPException: 400 on any mismatch -- the CSRF guard for the
whole flow.
"""
# Compare as bytes: compare_digest raises TypeError on non-ASCII
# str input, and ``received`` is attacker-controlled.
if not received or not secrets.compare_digest(
received.encode("utf-8"),
stored.encode("utf-8"),
):
detail = "OAuth state mismatch"
raise _bad_request(detail)
[docs]
def clear_flow_state(
response: Response,
*,
cookie_name: str,
cookie_secure: bool = True,
cookie_samesite: SameSite = "lax",
) -> None:
"""Delete the flow-state cookie once the callback is handled.
``cookie_secure`` / ``cookie_samesite`` must match the values
:func:`issue_flow_state` used -- browsers refuse to overwrite a
cookie when either attribute differs.
"""
response.delete_cookie(
key=cookie_name,
httponly=True,
secure=cookie_secure,
samesite=cookie_samesite,
)