API REST in Python: autenticazione con JWT e autorizzazione

API REST in Python: autenticazione con JWT e autorizzazione

La nostra API è funzionale e robusta, ma completamente aperta: chiunque può creare, modificare ed eliminare libri. In questa puntata la mettiamo in sicurezza. Introduciamo l'autenticazione con token JWT, sfruttando gli strumenti di sicurezza integrati in FastAPI, e l'autorizzazione basata sulla proprietà delle risorse, che stabilisce non solo chi è l'utente ma anche che cosa gli è permesso fare. Al termine avremo endpoint pubblici in lettura e protetti in scrittura, con token che il client presenta a ogni richiesta.

Autenticazione e autorizzazione

Conviene distinguere subito due concetti che spesso si confondono. L'autenticazione risponde alla domanda: chi sei? Verifica l'identità di chi effettua la richiesta. L'autorizzazione risponde a una domanda diversa: che cosa ti è permesso fare? Presuppone l'identità già nota e decide se quell'utente può compiere una determinata azione su una determinata risorsa. Un utente autenticato non è necessariamente autorizzato a tutto: potrà magari leggere qualsiasi libro ma modificare soltanto quelli che ha inserito.

I token JWT in breve

Un JSON Web Token è una stringa firmata che racchiude alcune informazioni, dette claim, tra cui l'identità dell'utente e la scadenza. Il server lo emette al momento del login e il client lo conserva, allegandolo a ogni richiesta successiva nell'intestazione Authorization. Poiché il token è firmato con una chiave segreta nota solo al server, quest'ultimo può verificarne l'autenticità senza consultare il database a ogni richiesta: basta ricalcolare la firma. È il modello ideale per API stateless consumate da app mobili, servizi esterni o frontend a pagina singola.

Installiamo le due librerie che ci servono: pyjwt per creare e verificare i token e passlib con bcrypt per l'hashing sicuro delle password.

pip install pyjwt "passlib[bcrypt]"

Il modello User

Aggiungiamo un modello utente in app/models.py. Memorizzeremo la password solo sotto forma di hash, mai in chiaro.

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True, index=True)
    # Contiene l'hash della password, non la password in chiaro
    hashed_password: str

Per legare ogni libro al suo autore, aggiungiamo al modello Book una colonna user_id che fa riferimento all'utente creatore. In sviluppo la aggiungiamo al modello e ricreiamo il database; in produzione genereremmo una migrazione Alembic, come visto nella terza puntata.

class Book(BookBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    isbn: str | None = Field(default=None, unique=True, index=True)
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    # Riferimento all'utente che ha creato il libro
    user_id: int | None = Field(default=None, foreign_key="user.id")

Le funzioni di sicurezza

Raccogliamo in un modulo app/security.py le funzioni per l'hashing delle password e per la creazione e verifica dei token.

from datetime import datetime, timedelta, timezone

import jwt
from passlib.context import CryptContext

# In produzione la chiave va letta da una variabile d'ambiente, mai scritta nel codice
SECRET_KEY = "cambia-questa-chiave-in-produzione"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def verify_password(plain: str, hashed: str) -> bool:
    # Confronta la password in chiaro con l'hash memorizzato
    return pwd_context.verify(plain, hashed)


def create_access_token(subject: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    # "sub" identifica il soggetto del token; "exp" ne fissa la scadenza
    payload = {"sub": subject, "exp": expire}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

La chiave segreta è qui scritta nel codice solo a scopo illustrativo: in un'applicazione reale va conservata in una variabile d'ambiente e mai messa sotto controllo di versione. Il claim sub conterrà l'identificatore dell'utente, mentre exp fa scadere automaticamente il token dopo l'intervallo stabilito.

Gli endpoint di registrazione e login

Creiamo un router dedicato all'autenticazione in app/routers/auth.py. La registrazione crea l'utente con la password sottoposta ad hashing; il login verifica le credenziali ed emette un token.

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel, EmailStr
from sqlmodel import Session, select

from app.database import get_session
from app.models import User
from app.security import create_access_token, hash_password, verify_password

router = APIRouter(prefix="/auth", tags=["auth"])


class RegisterInput(BaseModel):
    name: str
    email: EmailStr
    password: str


class TokenOut(BaseModel):
    access_token: str
    token_type: str = "bearer"


@router.post("/register", response_model=TokenOut, status_code=status.HTTP_201_CREATED)
def register(data: RegisterInput, session: Session = Depends(get_session)):
    existing = session.exec(select(User).where(User.email == data.email)).first()
    if existing is not None:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email già registrata.")

    user = User(name=data.name, email=data.email, hashed_password=hash_password(data.password))
    session.add(user)
    session.commit()
    session.refresh(user)

    token = create_access_token(subject=str(user.id))
    return TokenOut(access_token=token)


@router.post("/login", response_model=TokenOut)
def login(
    form: OAuth2PasswordRequestForm = Depends(),
    session: Session = Depends(get_session),
):
    # OAuth2PasswordRequestForm legge i campi username e password dal form
    user = session.exec(select(User).where(User.email == form.username)).first()
    if user is None or not verify_password(form.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Credenziali non valide.",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_access_token(subject=str(user.id))
    return TokenOut(access_token=token)

Per il login usiamo OAuth2PasswordRequestForm, una dipendenza di FastAPI che legge le credenziali secondo lo standard OAuth2, dai campi username e password di un form. Adottare questo formato ci regala l'integrazione con la documentazione interattiva, che mostrerà un pulsante per autenticarsi direttamente dalla pagina. Il tipo EmailStr in registrazione valida che l'email sia formalmente corretta; richiede il pacchetto email-validator, che si installa con pip install email-validator.

La dipendenza dell'utente corrente

Il cuore dell'autenticazione è una dipendenza che, data una richiesta, estrae il token, lo verifica e restituisce l'utente corrispondente. La aggiungiamo in app/security.py o in un modulo dedicato alle dipendenze.

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from sqlmodel import Session

from app.database import get_session
from app.models import User

# Indica a FastAPI dove ottenere il token; alimenta anche i /docs
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/login")


def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: Session = Depends(get_session),
) -> User:
    credentials_error = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Token non valido o scaduto.",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        if user_id is None:
            raise credentials_error
    except InvalidTokenError:
        raise credentials_error

    user = session.get(User, int(user_id))
    if user is None:
        raise credentials_error
    return user

La dipendenza oauth2_scheme si occupa di estrarre il token dall'intestazione Authorization: Bearer .... Se il token manca, è malformato o è scaduto, la funzione solleva un 401. Ogni rotta che dichiara user: User = Depends(get_current_user) sarà quindi accessibile solo a un utente autenticato, e riceverà l'oggetto utente già pronto.

Proteggere le rotte dei libri

Vogliamo che la lettura del catalogo resti pubblica, ma che la scrittura sia riservata agli utenti autenticati. Aggiungiamo la dipendenza dell'utente corrente alle rotte di creazione, aggiornamento ed eliminazione, lasciando intatte quelle di lettura. Nella creazione ne approfittiamo per registrare chi ha inserito il libro.

from app.security import get_current_user


@router.post("", response_model=BookRead, status_code=status.HTTP_201_CREATED)
def store_book(
    data: BookCreate,
    session: Session = Depends(get_session),
    user: User = Depends(get_current_user),
):
    book = Book.model_validate(data)
    book.user_id = user.id  # associa il libro all'utente autenticato
    session.add(book)
    session.commit()
    session.refresh(book)
    return book

Ora un tentativo di creare un libro senza token riceve una risposta 401 Unauthorized. Per superarla, il client deve allegare il token ottenuto al login.

curl -i -X POST http://127.0.0.1:8000/api/books \
  -H "Authorization: Bearer IL_TUO_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"title":"Il barone rampante","author":"Italo Calvino","year":1957}'

Autorizzazione basata sulla proprietà

L'autenticazione ci dice chi è l'utente; l'autorizzazione decide cosa può fare. Vogliamo consentire la modifica e l'eliminazione di un libro soltanto al suo proprietario. Ora che ogni libro porta un user_id, possiamo scrivere una funzione che confronta il proprietario con l'utente corrente e solleva un 403 Forbidden in caso di mancata corrispondenza. La collochiamo accanto alle altre dipendenze.

def get_owned_book(
    book: Book = Depends(get_book_or_404),
    user: User = Depends(get_current_user),
) -> Book:
    # Compone due dipendenze: recupero del libro e verifica del proprietario
    if book.user_id != user.id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Non sei il proprietario di questo libro.",
        )
    return book

Questa dipendenza è essa stessa composta da altre due: get_book_or_404, che recupera il libro e gestisce il 404, e get_current_user, che autentica. FastAPI risolve l'intera catena. Colleghiamola alle rotte di aggiornamento ed eliminazione, che così ricevono un libro già recuperato e di cui l'utente è certamente proprietario.

@router.put("/{book_id}", response_model=BookRead)
def update_book(
    data: BookUpdate,
    book: Book = Depends(get_owned_book),
    session: Session = Depends(get_session),
):
    for field, value in data.model_dump(exclude_unset=True).items():
        setattr(book, field, value)
    session.add(book)
    session.commit()
    session.refresh(book)
    return book


@router.delete("/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def destroy(
    book: Book = Depends(get_owned_book),
    session: Session = Depends(get_session),
):
    session.delete(book)
    session.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

La bellezza di questo approccio è che la logica di autorizzazione vive in un unico punto, componibile e riutilizzabile. Le rotte tornano a occuparsi solo della loro azione, sicure che a monte l'accesso è già stato verificato.

Permessi granulari con gli scope

Lo standard OAuth2, su cui poggiano gli strumenti di FastAPI, prevede il concetto di scope: permessi granulari che un token può possedere o meno. Potremmo emettere un token con il solo scope di lettura e un altro con anche quello di scrittura, includendo gli scope tra i claim del token e verificandoli con la dipendenza Security, che estende Depends proprio per questo. È il modo in cui si costruisce un controllo degli accessi più fine, per esempio distinguendo i client di sola consultazione da quelli abilitati a modificare il catalogo. La meccanica ricalca quella dell'utente corrente, aggiungendo il confronto fra gli scope richiesti dalla rotta e quelli presenti nel token.

Conclusione

La nostra API è ora sotto controllo. Con i token JWT abbiamo aggiunto l'autenticazione, creato gli endpoint di registrazione e login con hashing sicuro delle password e costruito una dipendenza che identifica l'utente corrente a ogni richiesta. Abbiamo protetto le operazioni di scrittura lasciando pubblica la lettura, e introdotto l'autorizzazione basata sulla proprietà tramite una dipendenza componibile che restituisce un 403 a chi tenta di toccare un libro altrui. Abbiamo infine accennato agli scope per un controllo degli accessi granulare.

Manca un ultimo tassello per portare l'API a un livello davvero professionale: la gestione uniforme degli errori, la strategia di versioning, la protezione contro l'abuso tramite rate limiting e, soprattutto, i test automatici. Sarà l'argomento della settima e ultima puntata, che chiuderà la serie.