Source code for fsh_lib.oauth.providers

"""Provider + client configuration for the OAuth 2.0 / OIDC flow.

Two value objects describe *who* the flow talks to and *as whom*:

* :class:`ProviderConfig` -- the provider's endpoints (authorize,
  token, userinfo, JWKS) and how its token endpoint authenticates
  the client.  Build one by hand, fetch it from an OpenID Connect
  discovery document with :func:`discover`, or start from a
  :func:`google` / :func:`microsoft` preset.

* :class:`ClientCredentials` -- the registered client's
  ``client_id`` and ``redirect_uri``, plus the *name* of the env
  var holding the ``client_secret``.  The secret itself never
  lives in the config object (or in generated source) -- only the
  env-var name does, matching :mod:`fsh_lib.auth`'s ``secret_env``
  convention.

The presets here are conveniences pinned to each provider's
published endpoints; verify them against the provider's current
documentation before relying on them.  The generic
:class:`ProviderConfig` / :func:`discover` pair is the real API.

Requires the ``oauth`` extra (``pip install 'fsh-lib[oauth]'``)
for :mod:`httpx`.
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Literal

import httpx
from pydantic import BaseModel, ConfigDict

if TYPE_CHECKING:
    from collections.abc import Sequence

#: Token-endpoint client-authentication method.  ``client_secret_post``
#: puts ``client_id`` / ``client_secret`` in the form body (Google,
#: Microsoft); ``client_secret_basic`` sends them as an HTTP Basic
#: ``Authorization`` header (Intuit / QuickBooks).
TokenAuthMethod = Literal["client_secret_post", "client_secret_basic"]

#: The default token-endpoint auth method -- an auth-method *name*,
#: not a secret (named so S105 doesn't false-positive on the literal
#: next to ``token_...`` the way it does at the field definition).
_DEFAULT_AUTH_METHOD: TokenAuthMethod = "client_secret_post"

#: Default per-request timeout, in seconds, for discovery and the
#: token / userinfo round-trips.  These sit on an interactive login
#: path, so a slow provider should fail fast.
_DEFAULT_TIMEOUT = 10.0

#: Google's OIDC issuer.  Its discovery document lives at
#: ``{GOOGLE_ISSUER}/.well-known/openid-configuration`` -- pass it to
#: :func:`discover` to build a :class:`ProviderConfig`.
GOOGLE_ISSUER = "https://accounts.google.com"

#: Microsoft tenant values that are *multi-tenant*: the ID token's
#: ``iss`` claim carries the caller's real tenant GUID, not the
#: literal, so a static issuer check can't be used (see
#: :func:`microsoft`).
_MS_WILDCARD_TENANTS = frozenset({"common", "organizations", "consumers"})


[docs] class OAuthConfigError(ValueError): """A provider configuration could not be built or discovered."""
[docs] class ClientCredentials(BaseModel): """The registered OAuth client's identity. Attributes: client_id: The provider-issued client identifier. client_secret_env: *Name* of the environment variable holding the client secret -- the secret is read from ``os.environ`` at token-exchange time, never stored here. A public (PKCE-only) client may leave this ``None``. redirect_uri: The callback URL registered with the provider; must match exactly on both the authorize request and the token exchange. """ model_config = ConfigDict(frozen=True) client_id: str client_secret_env: str | None = None redirect_uri: str
[docs] class ProviderConfig(BaseModel): """The endpoints and conventions of one OAuth 2.0 / OIDC provider. Attributes: authorization_endpoint: Where the user-agent is redirected to grant consent. token_endpoint: Where an authorization ``code`` (or a ``refresh_token``) is exchanged for tokens. userinfo_endpoint: OIDC UserInfo endpoint, or ``None`` when the provider has none. jwks_uri: JSON Web Key Set URL used to verify ID-token signatures, or ``None`` for non-OIDC providers. issuer: The ``iss`` value ID tokens must carry, or ``None`` to skip issuer validation (e.g. a multi-tenant endpoint -- see :func:`microsoft`). token_endpoint_auth_method: How the client authenticates to :attr:`token_endpoint` (see :data:`TokenAuthMethod`). default_scopes: Scopes requested when the caller doesn't pass an explicit list. """ model_config = ConfigDict(frozen=True, extra="ignore") authorization_endpoint: str token_endpoint: str userinfo_endpoint: str | None = None jwks_uri: str | None = None issuer: str | None = None token_endpoint_auth_method: TokenAuthMethod = _DEFAULT_AUTH_METHOD default_scopes: tuple[str, ...] = ()
[docs] async def discover( issuer: str, *, client: httpx.AsyncClient | None = None, default_scopes: Sequence[str] = ("openid", "email", "profile"), timeout: float = _DEFAULT_TIMEOUT, # noqa: ASYNC109 -- forwarded to httpx ) -> ProviderConfig: """Build a :class:`ProviderConfig` from an OIDC discovery document. Fetches ``{issuer}/.well-known/openid-configuration`` -- the standard OpenID Connect discovery endpoint -- and reads the authorize / token / userinfo / JWKS URLs and the issuer from it. Works for any spec-compliant OIDC provider (Google, Microsoft single-tenant, ...). This makes a network round-trip, so call it **once at app startup** (e.g. in the FastAPI lifespan) and reuse the returned config -- together with the :class:`~fsh_lib.oauth.client.OAuthClient` and :class:`~fsh_lib.oauth.oidc.IdTokenVerifier` built from it -- across requests; don't rebuild them per request. The :func:`google` / :func:`microsoft` presets need no discovery round-trip at all. Args: issuer: The provider's issuer URL, e.g. :data:`GOOGLE_ISSUER`. A trailing slash is tolerated. client: An existing :class:`httpx.AsyncClient` to reuse. When ``None`` a short-lived one is created and closed. default_scopes: Scopes baked into the returned config's :attr:`ProviderConfig.default_scopes`. timeout: Per-request timeout in seconds; ignored when *client* is supplied. Returns: The discovered :class:`ProviderConfig` (``token_endpoint_auth_method`` defaults to ``"client_secret_post"`` regardless of what the document advertises). Raises: OAuthConfigError: The document could not be fetched, is not valid JSON, is missing a required endpoint, or asserts an ``issuer`` different from the requested one (OIDC Discovery 4.3). """ url = f"{issuer.rstrip('/')}/.well-known/openid-configuration" owns_client = client is None client = client or httpx.AsyncClient(timeout=timeout) try: response = await client.get(url) response.raise_for_status() document = response.json() except httpx.HTTPError as exc: msg = f"OIDC discovery failed for {issuer!r}: {exc}" raise OAuthConfigError(msg) from exc except ValueError as exc: # json.JSONDecodeError msg = f"OIDC discovery document for {issuer!r} is not valid JSON" raise OAuthConfigError(msg) from exc finally: if owns_client: await client.aclose() missing = [ key for key in ("authorization_endpoint", "token_endpoint") if not document.get(key) ] if missing: msg = ( f"OIDC discovery document for {issuer!r} is missing " f"required field(s): {missing}" ) raise OAuthConfigError(msg) # OIDC Discovery 4.3: the document's issuer must match the one it # was retrieved for. A missing/null issuer falls back to the # requested one (never to None -- that would silently disable # ID-token issuer validation downstream). document_issuer = document.get("issuer") if document_issuer is not None and document_issuer.rstrip( "/" ) != issuer.rstrip("/"): msg = ( f"OIDC discovery document issuer {document_issuer!r} does " f"not match the requested issuer {issuer!r}" ) raise OAuthConfigError(msg) return ProviderConfig( authorization_endpoint=document["authorization_endpoint"], token_endpoint=document["token_endpoint"], userinfo_endpoint=document.get("userinfo_endpoint"), jwks_uri=document.get("jwks_uri"), issuer=document_issuer or issuer, default_scopes=tuple(default_scopes), )
[docs] def google( *, default_scopes: Sequence[str] = ("openid", "email", "profile"), ) -> ProviderConfig: """Return a :class:`ProviderConfig` for Google (OIDC) sign-in. Hard-codes Google's well-known OAuth 2.0 / OIDC endpoints, so -- unlike ``await discover(GOOGLE_ISSUER)`` -- it needs no network round-trip. Use :func:`discover` instead if you'd rather pick the endpoints up dynamically. To receive a refresh token from Google, the authorize request must also carry ``access_type=offline`` (and usually ``prompt=consent``); pass those via the ``extra_params`` of :meth:`~fsh_lib.oauth.client.OAuthClient.authorization_url` / :meth:`~fsh_lib.oauth.client.OAuthClient.begin`. Args: default_scopes: Scopes for :attr:`ProviderConfig.default_scopes`. Returns: Google's :class:`ProviderConfig`. """ return ProviderConfig( authorization_endpoint="https://accounts.google.com/o/oauth2/v2/auth", # S106: a URL, not a hard-coded password. token_endpoint="https://oauth2.googleapis.com/token", # noqa: S106 userinfo_endpoint="https://openidconnect.googleapis.com/v1/userinfo", jwks_uri="https://www.googleapis.com/oauth2/v3/certs", issuer=GOOGLE_ISSUER, default_scopes=tuple(default_scopes), )
[docs] def microsoft( tenant: str = "common", *, default_scopes: Sequence[str] = ("openid", "email", "profile"), ) -> ProviderConfig: """Return a :class:`ProviderConfig` for Microsoft Entra ID (v2.0). Builds the v2.0 endpoints for *tenant* directly, so no discovery round-trip is needed. Args: tenant: A directory (tenant) GUID, a verified domain, or one of the multi-tenant values ``"common"`` / ``"organizations"`` / ``"consumers"``. For the multi-tenant values the ID token's ``iss`` carries the *caller's* tenant GUID rather than the literal, so the returned config leaves :attr:`ProviderConfig.issuer` ``None`` (issuer validation skipped); pin a concrete *tenant* to get exact issuer validation. default_scopes: Scopes for :attr:`ProviderConfig.default_scopes`. Returns: The Microsoft v2.0 :class:`ProviderConfig`. """ base = f"https://login.microsoftonline.com/{tenant}" issuer = None if tenant in _MS_WILDCARD_TENANTS else f"{base}/v2.0" return ProviderConfig( authorization_endpoint=f"{base}/oauth2/v2.0/authorize", token_endpoint=f"{base}/oauth2/v2.0/token", userinfo_endpoint="https://graph.microsoft.com/oidc/userinfo", jwks_uri=f"{base}/discovery/v2.0/keys", issuer=issuer, default_scopes=tuple(default_scopes), )