"""A registry of configured providers for multi-provider apps.
An app that offers more than one provider ("Log in with Google *or*
Microsoft", plus a couple of connected services) needs one
:class:`~fsh_lib.oauth.client.OAuthClient` -- and, for OIDC logins, one
:class:`~fsh_lib.oauth.oidc.IdTokenVerifier` -- per provider, all built
once at startup and looked up per request by a URL-safe *slug*.
:class:`ProviderRegistry` is that lookup table. A :class:`Provider`
bundles a provider's client with its (optional) verifier;
:meth:`ProviderRegistry.register` builds both from a
:class:`~fsh_lib.oauth.providers.ProviderConfig` +
:class:`~fsh_lib.oauth.providers.ClientCredentials`, and
:meth:`ProviderRegistry.get` resolves a slug to a bundle or raises
HTTP 404. The slug doubles as the path segment in the login /
callback routes (``/auth/{provider}/login``) and as the value stored
in :attr:`~fsh_lib.oauth.connection.OAuthConnectionMixin.provider`, so
the connected-service refresh path can pick the right client back out.
Typical wiring -- build once in the FastAPI lifespan, look up per
request::
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, Request, Response
from fastapi.responses import RedirectResponse
from fsh_lib.oauth import (
ClientCredentials, Provider, ProviderRegistry,
google, microsoft,
issue_flow_state, read_flow_state, verify_state,
)
@asynccontextmanager
async def lifespan(app: FastAPI):
registry = ProviderRegistry()
registry.register(
"google", google(),
ClientCredentials(
client_id="...", client_secret_env="GOOGLE_SECRET",
redirect_uri="https://app/auth/google/callback"),
)
registry.register(
"microsoft", microsoft("common"),
ClientCredentials(
client_id="...", client_secret_env="MS_SECRET",
redirect_uri="https://app/auth/microsoft/callback"),
)
app.state.oauth = registry
try:
yield
finally:
await registry.aclose_all()
app = FastAPI(lifespan=lifespan)
def current_provider(provider: str, request: Request) -> Provider:
return request.app.state.oauth.get(provider) # 404s if unknown
Dep = Annotated[Provider, Depends(current_provider)]
@app.get("/auth/{provider}/login")
async def login(provider: str, p: Dep) -> RedirectResponse:
url, flow = p.client.begin(scopes=["openid", "email"], use_nonce=True)
resp = RedirectResponse(url)
issue_flow_state(
resp, flow, secret_env="JWT_SECRET", algorithm="HS256",
cookie_name=f"oauth_flow_{provider}", # per-provider cookie
)
return resp
@app.get("/auth/{provider}/callback")
async def callback(
provider: str, code: str, state: str, request: Request, p: Dep,
):
flow = read_flow_state(
request.cookies.get(f"oauth_flow_{provider}"),
secret_env="JWT_SECRET", algorithm="HS256",
)
verify_state(received=state, stored=flow.state)
tokens = await p.client.exchange_code(
code, code_verifier=flow.code_verifier,
)
identity = await p.require_verifier().verify(
tokens.id_token, nonce=flow.nonce,
)
... # map identity to a user, then fsh_lib.auth.issue_session(...)
Requires the ``oauth`` extra (``pip install 'fsh-lib[oauth]'``).
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
from fastapi import HTTPException, status
from fsh_lib.oauth.client import OAuthClient, OAuthError
from fsh_lib.oauth.oidc import DEFAULT_ALGORITHMS, IdTokenVerifier
from fsh_lib.oauth.providers import OAuthConfigError
if TYPE_CHECKING:
from collections.abc import Iterator, Mapping, Sequence
from typing import Self
import httpx
from fsh_lib.oauth.providers import ClientCredentials, ProviderConfig
[docs]
@dataclass(frozen=True)
class Provider:
"""One provider's ready-to-use client + optional ID-token verifier.
Attributes:
client: The provider's :class:`~fsh_lib.oauth.client.OAuthClient`.
verifier: The OIDC ID-token
:class:`~fsh_lib.oauth.oidc.IdTokenVerifier`, or ``None``
for a connected service that issues no ID token (no
``jwks_uri``).
"""
client: OAuthClient
verifier: IdTokenVerifier | None = None
[docs]
def require_verifier(self) -> IdTokenVerifier:
"""Return :attr:`verifier`, or raise if this provider has none.
A convenience for social-login callbacks, which always need a
verifier -- turns a misconfiguration (an OIDC route pointed at
a non-OIDC provider) into a clear error instead of an
:class:`AttributeError` on ``None``.
Raises:
OAuthError: This provider was registered without a
verifier.
"""
if self.verifier is None:
msg = "provider has no ID-token verifier (not an OIDC provider?)"
raise OAuthError(msg)
return self.verifier
[docs]
class ProviderRegistry:
"""A slug -> :class:`Provider` lookup table, built once at startup.
Args:
providers: Optional initial ``{slug: Provider}`` mapping. Most
callers start empty and use :meth:`register`.
"""
def __init__(self, providers: Mapping[str, Provider] | None = None) -> None:
"""Store the initial provider mapping (copied, not aliased)."""
self._providers: dict[str, Provider] = dict(providers or {})
[docs]
def register(
self,
slug: str,
config: ProviderConfig,
credentials: ClientCredentials,
*,
client: httpx.AsyncClient | None = None,
verify_id_tokens: bool = True,
algorithms: Sequence[str] = DEFAULT_ALGORITHMS,
leeway: float = 0.0,
) -> Provider:
"""Build and store a :class:`Provider` under *slug*.
Constructs an :class:`~fsh_lib.oauth.client.OAuthClient` from
*config* + *credentials* and, unless *verify_id_tokens* is
``False`` or *config* has no ``jwks_uri``, an
:class:`~fsh_lib.oauth.oidc.IdTokenVerifier` keyed on the
credentials' ``client_id``.
Args:
slug: URL-safe identifier, e.g. ``"google"``. Reused as
the route path segment and the stored
:attr:`~fsh_lib.oauth.connection.OAuthConnectionMixin.provider`
value.
config: The provider's endpoints (see
:class:`~fsh_lib.oauth.providers.ProviderConfig`).
credentials: This app's client id / secret-env /
redirect URI for the provider.
client: A shared :class:`httpx.AsyncClient` for the new
:class:`~fsh_lib.oauth.client.OAuthClient` to borrow
(so several providers can pool one connection pool).
When ``None`` the client creates and owns its own --
and :meth:`aclose_all` will close it.
verify_id_tokens: Build a verifier when the provider
supports OIDC. Set ``False`` for a pure connected
service.
algorithms: Accepted ID-token signature algorithms for
the verifier. Defaults to
:data:`~fsh_lib.oauth.oidc.DEFAULT_ALGORITHMS`.
leeway: Clock-skew tolerance (seconds) for the verifier.
Returns:
The stored :class:`Provider` (also retrievable via
:meth:`get`).
Raises:
OAuthConfigError: *slug* is already registered.
"""
self._check_free(slug)
oauth = OAuthClient(config, credentials, client=client)
verifier = (
IdTokenVerifier.from_provider(
config,
client_id=credentials.client_id,
algorithms=algorithms,
leeway=leeway,
)
if verify_id_tokens and config.jwks_uri is not None
else None
)
provider = Provider(client=oauth, verifier=verifier)
self._providers[slug] = provider
return provider
[docs]
def add(self, slug: str, provider: Provider) -> None:
"""Store a pre-built :class:`Provider` under *slug*.
The explicit alternative to :meth:`register` when the caller
has already assembled the client / verifier.
Raises:
OAuthConfigError: *slug* is already registered.
"""
self._check_free(slug)
self._providers[slug] = provider
def _check_free(self, slug: str) -> None:
"""Raise unless *slug* is unregistered.
A silent overwrite would orphan the displaced provider's
owned ``httpx.AsyncClient`` (nothing would ever close it) and
almost certainly signals a config mistake -- e.g. a login
provider and a connection sharing a slug.
"""
if slug in self._providers:
msg = f"OAuth provider slug already registered: {slug!r}"
raise OAuthConfigError(msg)
[docs]
def get(self, slug: str) -> Provider:
"""Return the :class:`Provider` for *slug*, or raise HTTP 404.
Shaped to drop straight into a FastAPI dependency that takes
the ``{provider}`` path param.
Raises:
HTTPException: 404 when *slug* isn't registered.
"""
provider = self._providers.get(slug)
if provider is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Unknown OAuth provider: {slug}",
)
return provider
[docs]
async def aclose_all(self) -> None:
"""Close every registered client's owned HTTP client.
Call on shutdown (e.g. in the FastAPI lifespan). A provider
registered with an externally-owned ``client`` is left open --
the registry only closes what its clients own.
"""
for provider in self._providers.values():
await provider.client.aclose()
async def __aenter__(self) -> Self:
"""Enter the async context, returning ``self``."""
return self
async def __aexit__(self, *_exc: object) -> None:
"""Close all owned clients on context exit."""
await self.aclose_all()
[docs]
def slugs(self) -> list[str]:
"""Return the registered slugs (e.g. for a login-button list)."""
return list(self._providers)
def __getitem__(self, slug: str) -> Provider:
"""Return the provider for *slug* (raises ``KeyError``)."""
return self._providers[slug]
def __contains__(self, slug: object) -> bool:
"""Return whether *slug* is registered."""
return slug in self._providers
def __iter__(self) -> Iterator[str]:
"""Iterate the registered slugs."""
return iter(self._providers)
def __len__(self) -> int:
"""Return the number of registered providers."""
return len(self._providers)