Autorisation JWT FastAPI : code et schémas

Mise en œuvre de l'authentification JWT dans FastAPI : Guide complet avec code

Le JWT (JSON Web Token) permet une authentification sans état dans les API. Un jeton se compose d'un en-tête, d'un chargement utile et d'une signature — encodés en base64 et séparés par des points.

En-tête précise l'algorithme de signature :

{
  "alg": "HS256",
  "typ": "JWT"
}

Chargement utile contient des revendications comme user_id (sub), l'heure d'émission (iat) et l'expiration (exp).

{
  "sub": "user_id_123",
  "exp": 1735689600,
  "iat": 1735603200
}

Signature garantit l'intégrité en utilisant HMAC-SHA256 avec une clé secrète.

Pour une utilisation en production, utilisez un système à deux jetons :

• Jeton d'accès (15–30 minutes) — pour les requêtes API
• Jeton de rafraîchissement (7–30 jours) — pour renouveler les jetons d'accès

Architecture du projet

Structure en couches simplifie le test et l'évolutivité :

project/
├── app/
│   ├── core/          # config.py, security.py, dependencies.py
│   ├── models/        # user.py (SQLAlchemy)
│   ├── schemas/       # user.py, token.py (Pydantic)
│   ├── routers/       # auth.py, users.py
│   ├── services/      # user.py
│   └── main.py
├── requirements.txt
└── .env

Les routeurs gèrent la logique HTTP, les schémas valident les entrées, les modèles définissent la structure de la base de données, et les services contiennent les règles métier.

Configuration et outils de sécurité

Dans core/config.py :

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    SECRET_KEY: str = "your-secret-key-change-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
    
    class Config:
        env_file = ".env"

settings = Settings()

Fonctions clés dans core/security.py :

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from .config import settings

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

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

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

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def create_refresh_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        return payload
    except JWTError:
        return None

Schémas Pydantic pour la validation

schemas/user.py :

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str
    full_name: str

class UserResponse(BaseModel):
    id: int
    email: EmailStr
    full_name: str
    
    class Config:
        from_attributes = True

schemas/token.py :

from pydantic import BaseModel

class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    user_id: int

Dépendance pour l'utilisateur actuel

core/dependencies.py implémente OAuth2PasswordBearer :

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    payload = decode_token(token)
    if payload is None:
        raise credentials_exception
    
    user_id = payload.get("sub")
    if user_id is None:
        raise credentials_exception
    
    token_data = TokenData(user_id=int(user_id))
    user = db.query(User).filter(User.id == token_data.user_id).first()
    if user is None:
        raise credentials_exception
    
    return user

Points d'entrée d'authentification

Dans routers/auth.py :

• POST /api/auth/register — crée un utilisateur avec hachage bcrypt
• POST /api/auth/login — retourne les jetons d'accès et de rafraîchissement
• POST /api/auth/refresh — renouvelle le jeton d'accès

Implémentation complète de la connexion :

@router.post("/login", response_model=Token)
def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    user = db.query(User).filter(User.email == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token = create_access_token(data={"sub": str(user.id)})
    refresh_token = create_refresh_token(data={"sub": str(user.id)})
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

Ressources protégées

routers/users.py :

@router.get("/me", response_model=UserResponse)
def get_current_user_info(current_user: User = Depends(get_current_user)):
    return current_user

Cet endpoint nécessite un jeton Bearer dans l'en-tête Authorization.

Application principale

main.py :

from fastapi import FastAPI
from .routers import auth, users

app = FastAPI(title="FastAPI JWT Auth", version="1.0.0")

app.include_router(auth.router)
app.include_router(users.router)

@app.get("/")
def root():
    return {"message": "Welcome to FastAPI JWT Auth API"}

Points clés à retenir

• Utilisez HS256 avec une clé secrète forte (au moins 32 caractères)
• Stockez les jetons de rafraîchissement dans des cookies HttpOnly pour éviter les attaques XSS
• Mettez en œuvre la rotation des jetons de rafraîchissement à chaque rafraîchissement
• Enregistrez les tentatives d'authentification échouées
• Testez les jetons expirés (vérifiez la revendication exp)

— Editorial Team