Skip to content

    Chapter 20, Intermediate

    OAuth2 and protected routes

    With the hash and the token ready, the flow is missing: the login route, the dependency that requires the token, and the control of who can do what. FastAPI already brings the OAuth2 standard, and Swagger understands it.

    The password flow with bearer token

    1. The client sends username and password to /token, like form (the OAuth2 standard).
    2. The server checks the password against the hash and returns a access_token with token_type: "bearer".
    3. In the following requests, the client sends Authorization: Bearer <token>.
    4. A dependency validates the token and delivers the user to the route, or refuses with 401.

    The OAuth2PasswordRequestForm reads the form, and the OAuth2PasswordBearer reads the header and informs Swagger where the login is (this is what enables the button Authorize). Both depend on the python-multipart, which comes in the fastapi[standard].

    intermediario/cap20_oauth2_rotas.pylinhas 10 a 32
    from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.testclient import TestClient
from pwdlib import PasswordHash

CHAVE = "uma-chave-secreta-com-no-minimo-32-bytes-0123"
ALGORITMO = "HS256"
EXPIRA_EM_MINUTOS = 30

hasher = PasswordHash.recommended()
SENHA_FALSA = hasher.hash("senha-que-ninguem-usa")

usuarios = {
    "ana": {"nome": "ana", "senha_hash": hasher.hash("1234"), "papel": "admin"},
    "bia": {"nome": "bia", "senha_hash": hasher.hash("abcd"), "papel": "usuario"},
}

oauth2 = OAuth2PasswordBearer(tokenUrl="token")
app = FastAPI()

    The login

    Two decisions deserve attention. The first: the error message is the same for "user does not exist" and "wrong password", to avoid revealing which users exist. The second: when the user does not exist, I still calculate a hash (against a fake password). Argon2 is intentionally slow, and if only existing users spent that time, an attacker would measure the response to find out which names are valid.

    intermediario/cap20_oauth2_rotas.pylinhas 37 a 58
    def autenticar(nome: str, senha: str) -> dict | None:
    usuario = usuarios.get(nome)
    senha_hash = usuario["senha_hash"] if usuario else SENHA_FALSA
    senha_confere = hasher.verify(senha, senha_hash)
    return usuario if usuario and senha_confere else None


def criar_token(nome: str, minutos: int = EXPIRA_EM_MINUTOS) -> str:
    exp = datetime.now(timezone.utc) + timedelta(minutes=minutos)
    return jwt.encode({"sub": nome, "exp": exp}, CHAVE, algorithm=ALGORITMO)


@app.post("/token")
def login(formulario: Annotated[OAuth2PasswordRequestForm, Depends()]):
    usuario = autenticar(formulario.username, formulario.password)
    if usuario is None:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            "Usuário ou senha inválidos",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {"access_token": criar_token(usuario["nome"]), "token_type": "bearer"}

    Protect routes with a dependency

    The dependency usuario_atual reads the token, validates it, and returns the user. Any route that requests it is protected. Note that it captures InvalidTokenError (and not a except: loose, which would hide even your bugs):

    intermediario/cap20_oauth2_rotas.pylinhas 63 a 84
    def usuario_atual(token: Annotated[str, Depends(oauth2)]) -> dict:
    erro = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        "Token inválido ou expirado",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        carga = jwt.decode(token, CHAVE, algorithms=[ALGORITMO])
    except jwt.InvalidTokenError:
        raise erro
    usuario = usuarios.get(carga.get("sub", ""))
    if usuario is None:
        raise erro
    return usuario


UsuarioAtual = Annotated[dict, Depends(usuario_atual)]


@app.get("/eu")
def eu(usuario: UsuarioAtual):
    return {"nome": usuario["nome"], "papel": usuario["papel"]}

    Who can do what: roles

    Being authenticated is not enough for everything. A dependency factory creates the role verification, and the 403 (and not the 401) says "I know who you are, but you cannot":

    intermediario/cap20_oauth2_rotas.pylinhas 89 a 100
    def requer_papel(papel: str):
    def verificador(usuario: UsuarioAtual) -> dict:
        if usuario["papel"] != papel:
            raise HTTPException(status.HTTP_403_FORBIDDEN, "Sem permissão para esta ação")
        return usuario

    return verificador


@app.get("/admin/segredos")
def segredos(usuario: Annotated[dict, Depends(requer_papel("admin"))]):
    return {"segredo": "só para administradores", "visto_por": usuario["nome"]}

    The flow in practice

    The login sends form (data=), and not JSON. The token goes in the header Authorization:

    intermediario/cap20_oauth2_rotas.pylinhas 105 a 116
    cliente = TestClient(app)

certo = cliente.post("/token", data={"username": "ana", "password": "1234"})
print(certo.status_code, certo.json()["token_type"], certo.json()["access_token"].count("."))
errado = cliente.post("/token", data={"username": "ana", "password": "errada"})
inexistente = cliente.post("/token", data={"username": "zeca", "password": "1234"})
print(errado.status_code, errado.json() == inexistente.json())


def cabecalho(nome: str, senha: str) -> dict:
    token = cliente.post("/token", data={"username": nome, "password": senha}).json()["access_token"]
    return {"Authorization": f"Bearer {token}"}
    Output
    200 bearer 2
401 True
    intermediario/cap20_oauth2_rotas.pylinhas 118 a 127
    sem_token = cliente.get("/eu")
print(sem_token.status_code, sem_token.headers["www-authenticate"])
print(cliente.get("/eu", headers=cabecalho("ana", "1234")).json())
print(cliente.get("/eu", headers={"Authorization": "Bearer lixo"}).status_code)

vencido = criar_token("ana", minutos=-1)
print(cliente.get("/eu", headers={"Authorization": f"Bearer {vencido}"}).status_code)

print(cliente.get("/admin/segredos", headers=cabecalho("ana", "1234")).status_code)
print(cliente.get("/admin/segredos", headers=cabecalho("bia", "abcd")).status_code)
    Output
    401 Bearer
{'nome': 'ana', 'papel': 'admin'}
401
401
200
403

    What Swagger gains

    Since the route uses OAuth2PasswordBearer, the OpenAPI document declares the security scheme, and the page /docs gets the button Authorize: you log in once, and all protected routes start sending the token on their own.

    intermediario/cap20_oauth2_rotas.pylinhas 132 a 133
    esquemas = app.openapi()["components"]["securitySchemes"]
print(list(esquemas), esquemas["OAuth2PasswordBearer"]["flows"]["password"]["tokenUrl"])
    Output
    ['OAuth2PasswordBearer'] token

    What a JWT does not do

    A JWT is stateless: the server does not store anything, and therefore cannot cancel a token before it expires (a true "logout", or the revocation of a stolen token). That’s why the validity is short (minutes). For long sessions, the standard is a second token for renewal (refresh token), with a long lifespan, stored on the server and can be revoked. And, always, only in HTTPS: a token in plain HTTP is a stolen token.

    Exercise 1

    An expired token denies access

    Check that a token of bia with negative validity is rejected in /eu, and the valid token shows the role usuario.