AdamThomasandClaude Opus 5.5
d0fa1e109c
- Une ligne de log par requête (route modèle, statut, durée, code d'erreur, compte), identifiant X-Request-ID repris dans chaque log et affiché à l'utilisateur sous une erreur, exceptions non gérées tracées (500 JSON) - Journal d'accès d'uvicorn coupé : il exposait le jeton du formulaire public - Logs métier : paiement SumUp, connexion, import, plannings, solveur - Nom du bénévole dans la description du checkout SumUp - Code et identifiant de transaction SumUp enregistrés au paiement (migration 0023), affichés sur la fiche et dans l'export Excel Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
"""Dépendances d'authentification et d'autorisation."""
|
|
|
|
import uuid
|
|
from collections.abc import Callable
|
|
|
|
import jwt
|
|
from fastapi import Depends, HTTPException, Request, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlmodel import Session
|
|
|
|
from app.auth.tokens import decode_access_token
|
|
from app.db import get_session
|
|
from app.models import Account, Position, Role, Shift
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def _unauthorized() -> HTTPException:
|
|
return HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"code": "error.unauthorized"},
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
|
|
def _forbidden() -> HTTPException:
|
|
return HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={"code": "error.forbidden"},
|
|
)
|
|
|
|
|
|
def get_current_account(
|
|
request: Request,
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(_bearer),
|
|
session: Session = Depends(get_session),
|
|
) -> Account:
|
|
if credentials is None or credentials.scheme.lower() != "bearer":
|
|
raise _unauthorized()
|
|
try:
|
|
claims = decode_access_token(credentials.credentials)
|
|
account_id = uuid.UUID(claims["sub"])
|
|
except (jwt.PyJWTError, KeyError, ValueError) as exc:
|
|
raise _unauthorized() from exc
|
|
account = session.get(Account, account_id)
|
|
if account is None:
|
|
raise _unauthorized()
|
|
# Repris sur la ligne d'accès de la requête (`app/request_logging.py`) :
|
|
# qui a rencontré l'erreur, par identifiant seulement.
|
|
request.state.account_id = account.id
|
|
return account
|
|
|
|
|
|
def current_account_in_edition(
|
|
edition_id: uuid.UUID,
|
|
account: Account = Depends(get_current_account),
|
|
) -> Account:
|
|
"""Le compte doit appartenir à l'édition de l'URL."""
|
|
if account.edition_id != edition_id:
|
|
raise _forbidden()
|
|
return account
|
|
|
|
|
|
def require_roles(*roles: Role) -> Callable[..., Account]:
|
|
def dependency(
|
|
account: Account = Depends(current_account_in_edition),
|
|
) -> Account:
|
|
if account.role not in roles:
|
|
raise _forbidden()
|
|
return account
|
|
|
|
return dependency
|
|
|
|
|
|
# Singletons prêts à l'emploi (évite un appel de fonction dans un défaut d'arg).
|
|
require_coordinator = require_roles(Role.COORDINATOR)
|
|
require_staff = require_roles(Role.COORDINATOR, Role.LEAD)
|
|
|
|
|
|
def ensure_can_write_position(account: Account, position: Position) -> None:
|
|
"""COORDINATION partout ; RÉFÉRENT uniquement sur ses propres postes."""
|
|
if account.role is Role.COORDINATOR:
|
|
return
|
|
if account.role is Role.LEAD and position.lead_account_id == account.id:
|
|
return
|
|
raise _forbidden()
|
|
|
|
|
|
def ensure_can_write_shift(
|
|
account: Account, session: Session, shift: Shift
|
|
) -> None:
|
|
if account.role is Role.COORDINATOR:
|
|
return
|
|
position = session.get(Position, shift.position_id)
|
|
if position is None:
|
|
raise _forbidden()
|
|
ensure_can_write_position(account, position)
|