Pourquoi Authentification dans FastAPI ?
L'authentification est un élément essentiel des applications web et API modernes. Elle permet de contrôler l'accès aux ressources en garantissant que seuls les utilisateurs autorisés ont la possibilité d'y accéder. Dans le contexte réel, cela peut signifier que seul un utilisateur authentifié peut effectuer certaines actions, comme créer des tâches, publier des articles ou scraper des données.
Un cas concret est celui d'une application de gestion de projet. Imaginez que vous ayez une API qui permet aux utilisateurs de gérer leurs projets et leurs tâches. Vous voudriez qu'un utilisateur puisse créer, modifier et supprimer des tâches uniquement s'il est connecté et autorisé à le faire.
Prerequis
- Connaissances en Python 3.x
- Familiarité avec les bases de FastAPI
- Installation d'Unicorn (serveur ASGI recommandé pour FastAPI)
Pour installer Unicorn, exécutez la commande suivante dans votre terminal :
pip install uvicorn
Concepts fondamentaux
1. Middleware
Le middleware est un mécanisme qui permet d'exécuter du code avant ou après chaque requête HTTP traitée par FastAPI. Il peut être utilisé pour effectuer des tâches comme l'authentification, la validation des données et le logging.
from fastapi import Request, Response
from fastapi.middleware.base import BaseHTTPMiddleware
class AuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# Vérifiez ici si l'utilisateur est authentifié
if not is_user_authenticated(request):
return Response("Unauthorized", status_code=401)
response = await call_next(request)
return response
def is_user_authenticated(request):
# Logique d'authentification
pass
2. Dépendances
Les dépendances permettent de définir des fonctions qui seront exécutées avant chaque opération API. Elles sont utiles pour l'authentification, où vous pouvez vérifier les informations d'identification et renvoyer une erreur si elles ne sont pas valides.
from fastapi import Depends, HTTPException, status
def get_current_user(token: str = Depends(oauth2_scheme)):
# Vérifiez ici la validité du token et renvoyez l'utilisateur
if not is_valid_token(token):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
return get_user_from_token(token)
def is_valid_token(token: str):
# Logique de validation du token
pass
def get_user_from_token(token: str):
# Récupérez l'utilisateur à partir du token
pass
3. Sécurité des jetons d'accès
Pour une authentification sécurisée, les applications utilisent souvent des tokens d'accès, comme des JWT (JSON Web Tokens). FastAPI peut être utilisé avec des bibliothèques comme python-jose pour travailler avec ces tokens.
from fastapi import Depends, HTTPException, status
from jose import JWTError, jwt
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(email=email)
if user is None:
raise credentials_exception
return user
def get_user(email: str):
# Récupérez l'utilisateur à partir de l'email
pass
Mise en pratique : projet fil rouge
Nous allons créer un mini-projet complet d'un gestionnaire de tâches. Le but est que seul les utilisateurs authentifiés puissent créer, modifier et supprimer des tâches.
1. Initialisation du projet
Créez un nouveau répertoire pour le projet et initialisez un environnement virtuel :
mkdir fastapi-todo
cd fastapi-todo
python -m venv venv
source venv/bin/activate # Sous Windows : `venv\Scripts\activate`
2. Création de l'application FastAPI
Créez un fichier main.py avec le contenu suivant :
from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel
from jose import JWTError, jwt
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
app = FastAPI()
class Item(BaseModel):
id: int
title: str
description: str = None
class User(BaseModel):
username: str
email: str = None
full_name: str = None
disabled: bool = None
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # fake hashed password
}
}
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return User(**user_dict)
def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user or not verify_password(password, user.hashed_password):
return False
return user
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
3. Définition des routes
Ajoutez les routes pour gérer les tâches :
from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
app = FastAPI()
class Item(BaseModel):
id: int
title: str
description: str = None
class User(BaseModel):
username: str
email: str = None
full_name: str = None
disabled: bool = None
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # fake hashed password
}
}
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return User(**user_dict)
def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user or not verify_password(password, user.hashed_password):
return False
return user
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
fake_items_db = [{"id": 1, "title": "Foo", "description": "There goes my error"}]
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
return current_user
@app.post("/items/")
async def create_item(item: Item, current_user: User = Depends(get_current_active_user)):
item_dict = item.dict()
fake_db[item.id] = item_dict
return item_dict
@app.get("/items/{item_id}")
async def read_item(item_id: int, current_user: User = Depends(get_current_active_user)):
if item_id not in fake_items_db:
raise HTTPException(status_code=404, detail="Item not found")
item_dict = fake_items_db[item_id]
return item_dict
4. Exécution de l'application
Exécutez l'application avec Unicorn :
uvicorn main:app --reload
Allez à http://127.0.0.1:8000/docs pour accéder à la documentation interactive de FastAPI et tester les routes.
Erreurs frequentes et debugging
1. Token invalide
Code incorrect :
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=username)
if user is None:
raise credentials_exception
return user
Code correct :
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=username)
if not user:
raise credentials_exception
return user
2. Route non définie
Code incorrect :
@app.post("/items/")
async def create_item(item: Item):
item_dict = item.dict()
fake_db[item.id] = item_dict
return item_dict
Code correct :
@app.post("/items/")
async def create_item(item: Item, current_user: User = Depends(get_current_active_user)):
item_dict = item.dict()
fake_db[item.id] = item_dict
return item_dict
3. Erreur de dépendance
Code incorrect :
@app.post("/items/")
async def create_item(item: Item):
item_dict = item.dict()
fake_db[item.id] = item_dict
return item_dict
Code correct :
@app.post("/items/")
async def create_item(item: Item, current_user: User = Depends(get_current_active_user)):
item_dict = item.dict()
fake_db[item.id] = item_dict
return item_dict
Pour aller plus loin
Authentification OAuth 2.0 : Explorez comment implémenter OAuth 2.0 pour une authentification sécurisée avec des fournisseurs tiers comme Google, Facebook, etc.
Token refresh : Ajoutez la possibilité de rafraîchir les tokens d'accès pour une meilleure expérience utilisateur.
Gestion des permissions : Implémentez un système de gestion des permissions pour contrôler les actions que chaque utilisateur peut effectuer.
Défi pratique : Créez un script CLI pour gérer les utilisateurs en utilisant FastAPI et le middleware d'authentification.