"""JWT helpers, password-less auth utilities and FastAPI dependencies.""" from __future__ import annotations import base64 import hashlib import hmac import secrets from datetime import UTC, datetime, timedelta from typing import Any from fastapi import Depends, HTTPException, Request, status from jose import JWTError, jwt from app.core.config import Settings, get_settings ALGO = "HS256" COOKIE_NAME = "uqo_session" def hash_user_id(user_id: str) -> str: return hashlib.sha256(user_id.encode()).hexdigest() def create_token(sub: str, role: str, settings: Settings, ttl: timedelta | None = None, kind: str = "access") -> str: now = datetime.now(UTC) exp = now + (ttl or timedelta(hours=settings.JWT_TTL_HOURS)) payload: dict[str, Any] = {"sub": sub, "role": role, "kind": kind, "iat": now, "exp": exp} return jwt.encode(payload, settings.JWT_SECRET.get_secret_value(), algorithm=ALGO) def decode_token(token: str, settings: Settings) -> dict[str, Any] | None: try: return jwt.decode(token, settings.JWT_SECRET.get_secret_value(), algorithms=[ALGO]) except JWTError: return None def new_magic_token() -> str: return secrets.token_urlsafe(32) def role_for_email(email: str, settings: Settings) -> str: e = email.lower() if e in settings.admin_emails: return "admin" if e in settings.professor_emails: return "professor" return "student" def email_allowed(email: str, settings: Settings) -> bool: e = email.lower().strip() if "@" not in e: return False if e in settings.invited_emails or e in settings.professor_emails or e in settings.admin_emails: return True domain = e.split("@", 1)[1] return domain in settings.allowed_domains class AuthUser: def __init__(self, user_id: str, role: str) -> None: self.id = user_id self.role = role @property def is_professor(self) -> bool: return self.role in {"professor", "admin"} @property def is_admin(self) -> bool: return self.role == "admin" def _extract_token(request: Request) -> str | None: auth = request.headers.get("authorization", "") if auth.lower().startswith("bearer "): return auth[7:].strip() return request.cookies.get(COOKIE_NAME) async def get_current_user(request: Request, settings: Settings = Depends(get_settings)) -> AuthUser: token = _extract_token(request) if not token: raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Connexion requise.") payload = decode_token(token, settings) if not payload or payload.get("kind") != "access": raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Session expirée. Reconnecte-toi.") return AuthUser(str(payload["sub"]), str(payload.get("role", "student"))) async def get_optional_user(request: Request, settings: Settings = Depends(get_settings)) -> AuthUser | None: try: return await get_current_user(request, settings) except HTTPException: return None async def require_professor(user: AuthUser = Depends(get_current_user)) -> AuthUser: if not user.is_professor: raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Réservé au professeur.") return user async def require_admin(user: AuthUser = Depends(get_current_user)) -> AuthUser: if not user.is_admin: raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Réservé à l'administrateur.") return user # ------------------------------------------------------------------ passwords (PBKDF2-SHA256) def hash_password(password: str, iterations: int = 310_000) -> str: salt = secrets.token_bytes(16) dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, iterations) return f"pbkdf2${iterations}${base64.b64encode(salt).decode()}${base64.b64encode(dk).decode()}" def verify_password(password: str, stored: str | None) -> bool: if not stored or not stored.startswith("pbkdf2$"): return False try: _, it, salt_b64, dk_b64 = stored.split("$", 3) dk = hashlib.pbkdf2_hmac("sha256", password.encode(), base64.b64decode(salt_b64), int(it)) return hmac.compare_digest(dk, base64.b64decode(dk_b64)) except (ValueError, TypeError): return False