"""Routes d'authentification : connexion, MFA, mot de passe, sessions.""" from __future__ import annotations from typing import Any import segno from fastapi import APIRouter, Depends, Request, Response from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.orm import Session from ..core import audit, security from ..core.config import get_config from ..core.db import get_db from ..models import STATUS_LABELS, User, UserSession from ..services import auth as auth_service from ..services import settings as settings_service from .deps import ApiError, client_ip, get_principal, get_principal_partial router = APIRouter(prefix="/api/auth", tags=["authentification"]) class LoginIn(BaseModel): username: str = Field(max_length=64) password: str = Field(max_length=256) class CodeIn(BaseModel): code: str = Field(max_length=32) class PasswordChangeIn(BaseModel): current_password: str = Field(max_length=256) new_password: str = Field(max_length=256) class PasswordIn(BaseModel): password: str = Field(max_length=256) def _set_cookie(response: Response, token: str) -> None: cfg = get_config() response.set_cookie( cfg.cookie_name, token, httponly=True, secure=cfg.cookie_secure, samesite="strict", path="/", ) def _clear_cookie(response: Response) -> None: cfg = get_config() response.delete_cookie(cfg.cookie_name, path="/", secure=cfg.cookie_secure, httponly=True, samesite="strict") def me_payload(db: Session, principal: auth_service.Principal) -> dict[str, Any]: u = principal.user cfg = get_config() return { "user": { "id": u.id, "username": u.username, "display_name": u.display_name, "role": u.role_code, "mfa_enabled": u.mfa_enabled, "can_export": "export.create" in principal.permissions, "last_login_at": u.last_login_at.isoformat() if u.last_login_at else None, }, "permissions": sorted(principal.permissions), "state": { "mfa_pending": not principal.session.mfa_verified, "mfa_setup_required": principal.needs_mfa_setup, "password_change_required": principal.needs_password_change, }, "csrf_token": principal.session.csrf_token, "app": { "env": cfg.env, "demo": cfg.env in {"demo", "development"}, "autosave_delay_ms": settings_service.get(db, "autosave_delay_ms"), "session_idle_minutes": settings_service.get(db, "session_idle_minutes"), "password_min_length": settings_service.get(db, "password_min_length"), "status_labels": STATUS_LABELS, }, } @router.post("/login") def login(body: LoginIn, request: Request, response: Response, db: Session = Depends(get_db)) -> dict[str, Any]: ip = client_ip(request) try: user = auth_service.authenticate(db, body.username, body.password, ip) except auth_service.AuthError as e: raise ApiError(e.status, e.message, e.code) from None # Rotation : une éventuelle session précédente sur ce navigateur est révoquée. old = auth_service.load_session(db, request.cookies.get(get_config().cookie_name)) if old is not None: old[0].revoked_at, old[0].revoked_reason = auth_service.utcnow(), "nouvelle connexion" needs_code = user.mfa_enabled token, sess = auth_service.create_session( db, user, mfa_verified=not needs_code, ip=ip, user_agent=request.headers.get("user-agent")) if needs_code: audit.record(db, "auth.password_ok", user_id=user.id, username=user.username, ip=ip, details={"etape": "code MFA attendu"}) else: user.last_login_at = auth_service.utcnow() audit.record(db, "auth.login", user_id=user.id, username=user.username, ip=ip) db.commit() _set_cookie(response, token) principal = auth_service.Principal(user, sess, auth_service.permissions_for(db, user), auth_service.mfa_required_for(db, user)) return me_payload(db, principal) @router.post("/mfa") def mfa_verify(body: CodeIn, request: Request, response: Response, db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal_partial)) -> dict[str, Any]: if principal.session.mfa_verified: return me_payload(db, principal) user = principal.user try: method = auth_service.verify_second_factor(db, user, body.code, client_ip(request)) except auth_service.AuthError as e: raise ApiError(e.status, e.message, e.code) from None token, sess = auth_service.rotate_session(db, principal.session, user, mfa_verified=True) user.last_login_at = auth_service.utcnow() audit.record(db, "auth.login", user_id=user.id, username=user.username, ip=client_ip(request), details={"mfa": method}) if method == "recovery": audit.record(db, "auth.recovery_code_used", user_id=user.id, username=user.username, ip=client_ip(request), details={"codes_restants": auth_service.remaining_recovery_codes(db, user.id)}) db.commit() _set_cookie(response, token) return me_payload(db, auth_service.Principal(user, sess, principal.permissions, principal.mfa_required)) @router.get("/me") def me(db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal_partial)) -> dict[str, Any]: return me_payload(db, principal) @router.post("/logout") def logout(request: Request, response: Response, db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal_partial)) -> dict[str, bool]: principal.session.revoked_at, principal.session.revoked_reason = auth_service.utcnow(), "déconnexion" audit.record(db, "auth.logout", user_id=principal.user.id, username=principal.user.username, ip=client_ip(request)) db.commit() _clear_cookie(response) return {"ok": True} @router.post("/password") def change_password(body: PasswordChangeIn, request: Request, response: Response, db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal_partial)) -> dict[str, Any]: if not principal.session.mfa_verified: raise ApiError(401, "Code de vérification (MFA) requis.", "mfa_required") user = principal.user if not security.verify_password(user.password_hash, body.current_password): auth_service.register_failure(db, user.username, client_ip(request)) db.commit() raise ApiError(400, "Le mot de passe actuel est incorrect.", "bad_password") problems = security.password_problems(body.new_password, user.username, int(settings_service.get(db, "password_min_length"))) if security.verify_password(user.password_hash, body.new_password): problems.append("Le nouveau mot de passe doit être différent de l'actuel.") if problems: raise ApiError(400, " ".join(problems), "weak_password") user.password_hash = security.hash_password(body.new_password) user.must_change_password = False user.password_changed_at = auth_service.utcnow() revoked = auth_service.revoke_user_sessions(db, user.id, "changement de mot de passe", principal.session.id) token, sess = auth_service.rotate_session(db, principal.session, user, mfa_verified=True) audit.record(db, "auth.password_changed", user_id=user.id, username=user.username, ip=client_ip(request), details={"sessions_revoquees": revoked}) db.commit() _set_cookie(response, token) return me_payload(db, auth_service.Principal(user, sess, principal.permissions, principal.mfa_required)) @router.post("/mfa/setup") def mfa_setup(request: Request, db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal_partial)) -> dict[str, Any]: if not principal.session.mfa_verified: raise ApiError(401, "Code de vérification (MFA) requis.", "mfa_required") user = principal.user if user.mfa_enabled: raise ApiError(409, "La double authentification est déjà active.", "mfa_already_enabled") secret = security.new_totp_secret() user.mfa_pending_secret_enc = security.encrypt_secret(secret) db.commit() uri = security.totp_uri(secret, user.username) qr = segno.make(uri, error="m") return {"secret": secret, "otpauth_uri": uri, "qr_svg": qr.svg_inline(scale=5, dark="#0b1f3a", light="#ffffff")} @router.post("/mfa/enable") def mfa_enable(body: CodeIn, request: Request, db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal_partial)) -> dict[str, Any]: if not principal.session.mfa_verified: raise ApiError(401, "Code de vérification (MFA) requis.", "mfa_required") user = principal.user if user.mfa_enabled or not user.mfa_pending_secret_enc: raise ApiError(409, "Aucune activation en cours.", "mfa_no_setup") secret = security.decrypt_secret(user.mfa_pending_secret_enc) step = security.verify_totp(secret, body.code, None) if step is None: raise ApiError(400, "Code incorrect. Vérifiez l'heure de votre téléphone et réessayez.", "mfa_failed") user.mfa_secret_enc, user.mfa_pending_secret_enc = user.mfa_pending_secret_enc, None user.mfa_enabled, user.mfa_last_step = True, step codes = auth_service.issue_recovery_codes(db, user) audit.record(db, "auth.mfa_enabled", user_id=user.id, username=user.username, ip=client_ip(request)) db.commit() return {"recovery_codes": codes, "me": me_payload(db, principal)} @router.post("/mfa/disable") def mfa_disable(body: PasswordIn, request: Request, db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal)) -> dict[str, Any]: user = principal.user if principal.mfa_required: raise ApiError(400, "La double authentification est obligatoire pour votre rôle.", "mfa_required_by_policy") if not security.verify_password(user.password_hash, body.password): raise ApiError(400, "Mot de passe incorrect.", "bad_password") user.mfa_enabled, user.mfa_secret_enc, user.mfa_last_step = False, None, None auth_service.issue_recovery_codes(db, user) # invalide les anciens codes audit.record(db, "auth.mfa_disabled", user_id=user.id, username=user.username, ip=client_ip(request)) db.commit() return me_payload(db, principal) @router.post("/recovery-codes") def regenerate_codes(body: PasswordIn, request: Request, db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal)) -> dict[str, Any]: user = principal.user if not user.mfa_enabled: raise ApiError(409, "Activez d'abord la double authentification.", "mfa_disabled") if not security.verify_password(user.password_hash, body.password): raise ApiError(400, "Mot de passe incorrect.", "bad_password") codes = auth_service.issue_recovery_codes(db, user) audit.record(db, "auth.recovery_codes_regenerated", user_id=user.id, username=user.username, ip=client_ip(request)) db.commit() return {"recovery_codes": codes} @router.get("/sessions") def my_sessions(db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal)) -> dict[str, Any]: rows = db.execute(select(UserSession).where( UserSession.user_id == principal.user.id, UserSession.revoked_at.is_(None), UserSession.expires_at > auth_service.utcnow(), ).order_by(UserSession.last_seen_at.desc())).scalars() return { "sessions": [{ "id": s.id[:12], "current": s.id == principal.session.id, "ip": s.ip, "user_agent": s.user_agent, "created_at": s.created_at.isoformat(), "last_seen_at": s.last_seen_at.isoformat(), } for s in rows], "recovery_codes_left": auth_service.remaining_recovery_codes(db, principal.user.id), } @router.post("/sessions/revoke-others") def revoke_others(request: Request, db: Session = Depends(get_db), principal: auth_service.Principal = Depends(get_principal)) -> dict[str, int]: n = auth_service.revoke_user_sessions(db, principal.user.id, "révocation par l'utilisateur", principal.session.id) audit.record(db, "auth.sessions_revoked", user_id=principal.user.id, username=principal.user.username, ip=client_ip(request), details={"nombre": n}) db.commit() return {"revoked": n} def get_user_by_username(db: Session, username: str) -> User | None: return db.execute(select(User).where(User.username == username.lower())).scalar_one_or_none()