OAuth login flow¶
The other pages check a credential. The OAuth helpers obtain one: they handle the tricky parts of a "Log in with..." button (discovery, CSRF protection, PKCE, code exchange, ID token validation), while you keep the routes.
sequenceDiagram
participant B as Browser
participant A as Your app
participant P as Provider
B->>A: GET /oauth/login
A-->>B: redirect to the provider
B->>P: log in
P-->>B: redirect to /oauth/callback
B->>A: GET /oauth/callback?code&state
A->>P: exchange the code
P-->>A: tokens
A-->>B: your session cookie 1. Redirect to the provider¶
@app.get("/oauth/login")
async def oauth_login(request: Request, next: str = "/"):
endpoints = await oauth_resolve_provider_urls(DISCOVERY_URL)
verifier, challenge = oauth_generate_pkce_pair()
flow = {
"state": oauth_generate_state_token(),
"nonce": oauth_generate_nonce(),
"verifier": verifier,
}
request.session["oauth"] = flow
return oauth_build_authorization_redirect(
endpoints.authorization_endpoint,
client_id=CLIENT_ID,
scopes="openid email profile",
redirect_uri=REDIRECT_URI,
destination=next,
state_token=flow["state"],
code_challenge=challenge,
nonce=flow["nonce"],
)
The state, nonce and PKCE verifier are kept in the server session (Starlette's SessionMiddleware) until the user comes back.
2. Handle the callback¶
@app.get("/oauth/callback")
async def oauth_callback(request: Request, code: str, state: str | None = None):
endpoints = await oauth_resolve_provider_urls(DISCOVERY_URL)
flow = request.session.pop("oauth", {})
try:
destination = oauth_decode_state(
state, expected_state_token=flow.get("state", ""), fallback="/"
)
token = await oauth_exchange_code(
token_url=endpoints.token_endpoint,
code=code,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
redirect_uri=REDIRECT_URI,
code_verifier=flow.get("verifier"),
)
if endpoints.issuer is None or endpoints.jwks_uri is None:
raise OAuthError("the provider does not support OpenID Connect")
claims = await oauth_validate_id_token(
token.get("id_token", ""),
issuer=endpoints.issuer,
jwks_uri=endpoints.jwks_uri,
client_id=CLIENT_ID,
nonce=flow.get("nonce", ""),
)
except OAuthError:
return RedirectResponse("/login?error=oauth", status_code=303)
key = (claims["iss"], claims["sub"])
user_id = USERS.setdefault(key, str(len(USERS) + 1))
response = RedirectResponse(destination, status_code=303)
session.set_cookie(response, user_id)
return response
oauth_decode_staterefuses a callback this browser did not start, and only allows relativenextURLs by default.oauth_validate_id_tokenchecks the ID token's signature, issuer, audience, expiry and nonce. Never trust an ID token before that.- Every helper raises a subclass of
OAuthError, so oneexceptcovers the whole callback.
The OpenID Connect recipe shows the full file.
The discovery URL is configuration
Never build it from request input.