Skip to content

Step-up authentication

Some routes can take over an account: changing the password, adding a second factor, creating an API token. A valid session is not enough there; you want proof the user logged in recently, or strongly enough.

Recent login: fresh()

import hashlib
import hmac
import os
import secrets
from typing import Annotated, TypedDict

from fastapi import FastAPI, Response, Security
from pydantic import BaseModel

from fastapi_multiauth import APIKeyCookieAuth, UnauthorizedError

SECRET_KEY = os.environ["SECRET_KEY"]


def hash_password(password: str, salt: bytes | None = None) -> bytes:
    salt = salt or secrets.token_bytes(16)
    return salt + hashlib.scrypt(password.encode(), salt=salt, n=2**14, r=8, p=1)


def check_password(password: str, stored: bytes) -> bool:
    return hmac.compare_digest(hash_password(password, stored[:16]), stored)


class User(TypedDict):
    password: bytes
    generation: int


USERS: dict[str, User] = {
    "alice": {"password": hash_password("wonderland"), "generation": 0}
}


class Password(BaseModel):
    password: str


async def validate_session(value: str) -> str:
    username, _, generation = value.rpartition(":")
    user = USERS.get(username)
    if user is None or generation != str(user["generation"]):
        raise UnauthorizedError()
    return username


session = APIKeyCookieAuth(
    name="__Host-session", validator=validate_session, secret_key=SECRET_KEY
)
FreshUser = Annotated[str, Security(session.fresh(max_age=300))]

app = FastAPI()


def sign_in(response: Response, username: str) -> None:
    session.set_cookie(response, f"{username}:{USERS[username]['generation']}")


@app.post("/me/re-auth")
async def re_auth(
    body: Password, response: Response, username: str = Security(session)
):
    if not check_password(body.password, USERS[username]["password"]):
        raise UnauthorizedError()
    sign_in(response, username)  # a new login time
    return {"ok": True}


@app.post("/me/password")
async def change_password(body: Password, response: Response, username: FreshUser):
    USERS[username]["password"] = hash_password(body.password)
    USERS[username]["generation"] += 1
    sign_in(response, username)
    return {"ok": True}

session.fresh(max_age=300) accepts only a login from the last 5 minutes. An older one gets a 401 (logging in again fixes it), and the re-auth route issues a new cookie, which counts as a new login. The cookie also carries the user's generation: a password change increments it, which ends every other session, and signs the current one back in.

A signed cookie knows when it was set. Other sources need to be told where the login time is, with authenticated_at=:

bearer.fresh(max_age=300, authenticated_at=lambda claims: claims.get("auth_time"))

Warning

Only issue a new credential on a real login. Refreshing it silently on a timer makes the user look fresh forever.

Strong login: stepped_up()

stepped_up() asks how the user logged in, from the token's acr claim:

from typing import Annotated

from fastapi import FastAPI, Security

from fastapi_multiauth import HTTPBearerAuth, JWTValidator

ISSUER = "https://idp.example.com/realms/main"

bearer = HTTPBearerAuth(
    validator=JWTValidator(
        jwks_url=f"{ISSUER}/protocol/openid-connect/certs",
        audience="my-api",
        issuer=ISSUER,
    )
)
StrongClaims = Annotated[
    dict,
    Security(
        bearer.stepped_up(acr_values=["urn:mfa"], acr=lambda claims: claims.get("acr"))
    ),
]

app = FastAPI()


@app.post("/payouts")
async def payout(claims: StrongClaims):
    return {"approved_by": claims["sub"]}

A weaker login gets a 401 whose WWW-Authenticate header names the required level (acr_values="urn:mfa"), so the client can send the user back to the provider. To check the methods used (the amr claim), compare against AuthenticationMethodReference.

A signed cookie carries its own level: pass acr= and amr= to set_cookie, and stepped_up() reads them back without an acr= reader:

import hmac
import os
from typing import Annotated

from fastapi import FastAPI, Request, Response, Security
from pydantic import BaseModel

from fastapi_multiauth import APIKeyCookieAuth, UnauthorizedError
from fastapi_multiauth import AuthenticationMethodReference as AMR

SECRET_KEY = os.environ["SECRET_KEY"]
OTP_CODES = {"alice": "123456"}  # stands in for your TOTP check


class Code(BaseModel):
    code: str


async def validate_session(username: str) -> str:
    return username


session = APIKeyCookieAuth(
    name="__Host-session", validator=validate_session, secret_key=SECRET_KEY
)
StrongUser = Annotated[str, Security(session.stepped_up(["aal2"]))]

app = FastAPI()


def sign_in(response: Response, username: str) -> None:
    session.set_cookie(response, username, acr="aal1", amr=[AMR.PASSWORD])


@app.post("/login/otp")
async def verify_otp(
    body: Code, request: Request, response: Response, username: str = Security(session)
):
    expected = OTP_CODES.get(username)
    if expected is None or not hmac.compare_digest(body.code, expected):
        raise UnauthorizedError()
    methods = [*(session.amr_of(request) or ()), AMR.ONE_TIME_PASSWORD]
    session.set_cookie(response, username, acr="aal2", amr=methods)
    return {"ok": True}


@app.post("/payouts")
async def payout(username: StrongUser):
    return {"approved_by": username}

Call sign_in from your login route once the password matches: the session starts at aal1. The OTP route re-issues the cookie at aal2, adding otp to the methods it already had, and /payouts refuses anything less. The level names are yours to pick; these follow NIST SP 800-63B. For the full flow with real TOTP codes, a pending-login cookie and recovery codes, see the two-factor recipe.

session.acr_of(request) and session.amr_of(request) read the current cookie's level and methods. Both are signed with the value, so the client cannot raise its own level, and a cookie set without them never passes stepped_up().

Step up every source of a MultiAuth

A source left out of the guard is never checked: MultiAuth(bearer, session.stepped_up(["mfa"])) lets any bearer token through. When sources read the level differently, guard each one: MultiAuth(bearer.stepped_up(["mfa"], acr=read_acr), session.stepped_up(["mfa"])).

Good to know

  • fresh() and stepped_up() can be chained, and work on MultiAuth too, passing the same reader to every source, a signed cookie included.
  • With session_id=True, the step-up set_cookie returns a new session id: move your stored session to it.
  • Neither works with optional().
  • Refusals raise StaleCredentialError or InsufficientAuthenticationError.