Démarrage rapide FastAPI et Django

pip install authlib httpx itsdangerous
# FastAPI:  pip install fastapi uvicorn
# Django:   pip install django

# honkio_auth.py: the refresh and the outage rule, shared by both frameworks.
import os
import time

import httpx

ISSUER = os.environ["HONKIO_AUTH_ISSUER"]  # https://login-dev.acme.ca
CLIENT_ID = os.environ["HONKIO_AUTH_CLIENT_ID"]
CLIENT_SECRET = os.environ["HONKIO_AUTH_CLIENT_SECRET"]
TOKEN_URL = ISSUER + "/token"
OUTAGE_GRACE_S = 24 * 60 * 60


def save_tokens(session: dict, token: dict) -> None:
    claims = token.get("userinfo")  # Authlib's parsed, verified ID token
    if claims:
        session["user"] = {"sub": claims["sub"], "email": claims.get("email")}
    if token.get("id_token"):
        session["id_token"] = token["id_token"]
    if token.get("refresh_token"):  # refresh tokens rotate: keep the newest
        session["refresh_token"] = token["refresh_token"]
    session["access_token"] = token["access_token"]
    session["access_expires_at"] = time.time() + int(token.get("expires_in", 600))
    session.pop("retry_refresh_at", None)


def ensure_fresh(session: dict) -> bool:
    """False means the session has ended: send the user to sign in."""
    if "user" not in session:
        return False
    now = time.time()
    expires_at = session["access_expires_at"]
    if now < expires_at - 30:
        return True
    within_grace = now < expires_at + OUTAGE_GRACE_S
    if within_grace and now < session.get("retry_refresh_at", 0):
        return True
    try:
        response = httpx.post(
            TOKEN_URL,
            data={"grant_type": "refresh_token", "refresh_token": session["refresh_token"]},
            auth=(CLIENT_ID, CLIENT_SECRET),
            timeout=5,
        )
    except httpx.TransportError:  # network error or timeout: HonkIO can't be reached
        response = None
    if response is None or response.status_code >= 500:
        if within_grace:
            session["retry_refresh_at"] = now + 60  # keep the session, retry in a minute
            return True
        session.clear()  # the outage outlasted 24 hours past expiry
        return False
    if response.status_code != 200:
        # invalid_grant: revoked, blocked, deleted or removed from the organization.
        session.clear()
        return False
    save_tokens(session, response.json())
    return True

# main.py
import os

from authlib.integrations.starlette_client import OAuth
from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse
from starlette.middleware.sessions import SessionMiddleware

from honkio_auth import CLIENT_ID, CLIENT_SECRET, ISSUER, ensure_fresh, save_tokens

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key=os.environ["SESSION_SECRET"], same_site="lax", https_only=True)

oauth = OAuth()
oauth.register(
    "honkio",
    server_metadata_url=ISSUER + "/.well-known/openid-configuration",
    client_id=CLIENT_ID,
    client_secret=CLIENT_SECRET,
    client_kwargs={"scope": "openid email", "code_challenge_method": "S256"},
)


@app.get("/auth/login")
async def login(request: Request):
    # Authlib adds state, nonce and the PKCE challenge.
    return await oauth.honkio.authorize_redirect(request, str(request.url_for("callback")))


@app.get("/auth/callback")
async def callback(request: Request):
    # Checks state, PKCE and the ID token: signature, iss, aud and nonce.
    token = await oauth.honkio.authorize_access_token(request)
    request.session.clear()
    save_tokens(request.session, token)
    return RedirectResponse("/")


@app.get("/")
async def home(request: Request):
    if not ensure_fresh(request.session):
        return RedirectResponse("/auth/login")
    user = request.session["user"]
    return {"signed_in_as": user.get("email") or user["sub"]}


@app.post("/auth/logout")
async def logout(request: Request):
    id_token = request.session.get("id_token")
    request.session.clear()
    url = ISSUER + "/logout?post_logout_redirect_uri=" + str(request.url_for("home"))
    if id_token:
        url += "&id_token_hint=" + id_token
    return RedirectResponse(url, status_code=303)

# views.py (needs django.contrib.sessions; wire login, callback and logout in urls.py)
from authlib.integrations.django_client import OAuth
from django.shortcuts import redirect
from django.urls import reverse
from django.views.decorators.http import require_POST

from honkio_auth import CLIENT_ID, CLIENT_SECRET, ISSUER, ensure_fresh, save_tokens

oauth = OAuth()
oauth.register(
    "honkio",
    server_metadata_url=ISSUER + "/.well-known/openid-configuration",
    client_id=CLIENT_ID,
    client_secret=CLIENT_SECRET,
    client_kwargs={"scope": "openid email", "code_challenge_method": "S256"},
)


def login(request):
    return oauth.honkio.authorize_redirect(request, request.build_absolute_uri(reverse("callback")))


def callback(request):
    token = oauth.honkio.authorize_access_token(request)
    request.session.cycle_key()
    save_tokens(request.session, token)
    return redirect("/")


def signed_in(view):
    def wrapper(request, *args, **kwargs):
        if not ensure_fresh(request.session):
            request.session.flush()
            return redirect("login")
        return view(request, *args, **kwargs)
    return wrapper


@require_POST
def logout(request):
    id_token = request.session.get("id_token")
    request.session.flush()
    url = ISSUER + "/logout?post_logout_redirect_uri=" + request.build_absolute_uri("/")
    if id_token:
        url += "&id_token_hint=" + id_token
    return redirect(url)