spb/uqo-chat

1"""JWT helpers, password-less auth utilities and FastAPI dependencies."""23from __future__ import annotations45import base646import hashlib7import hmac8import secrets9from datetime import UTC, datetime, timedelta10from typing import Any1112from fastapi import Depends, HTTPException, Request, status13from jose import JWTError, jwt1415from app.core.config import Settings, get_settings1617ALGO = "HS256"18COOKIE_NAME = "uqo_session"192021def hash_user_id(user_id: str) -> str:22    return hashlib.sha256(user_id.encode()).hexdigest()232425def create_token(sub: str, role: str, settings: Settings, ttl: timedelta | None = None,26                 kind: str = "access") -> str:27    now = datetime.now(UTC)28    exp = now + (ttl or timedelta(hours=settings.JWT_TTL_HOURS))29    payload: dict[str, Any] = {"sub": sub, "role": role, "kind": kind, "iat": now, "exp": exp}30    return jwt.encode(payload, settings.JWT_SECRET.get_secret_value(), algorithm=ALGO)313233def decode_token(token: str, settings: Settings) -> dict[str, Any] | None:34    try:35        return jwt.decode(token, settings.JWT_SECRET.get_secret_value(), algorithms=[ALGO])36    except JWTError:37        return None383940def new_magic_token() -> str:41    return secrets.token_urlsafe(32)424344def role_for_email(email: str, settings: Settings) -> str:45    e = email.lower()46    if e in settings.admin_emails:47        return "admin"48    if e in settings.professor_emails:49        return "professor"50    return "student"515253def email_allowed(email: str, settings: Settings) -> bool:54    e = email.lower().strip()55    if "@" not in e:56        return False57    if e in settings.invited_emails or e in settings.professor_emails or e in settings.admin_emails:58        return True59    domain = e.split("@", 1)[1]60    return domain in settings.allowed_domains616263class AuthUser:64    def __init__(self, user_id: str, role: str) -> None:65        self.id = user_id66        self.role = role6768    @property69    def is_professor(self) -> bool:70        return self.role in {"professor", "admin"}7172    @property73    def is_admin(self) -> bool:74        return self.role == "admin"757677def _extract_token(request: Request) -> str | None:78    auth = request.headers.get("authorization", "")79    if auth.lower().startswith("bearer "):80        return auth[7:].strip()81    return request.cookies.get(COOKIE_NAME)828384async def get_current_user(request: Request, settings: Settings = Depends(get_settings)) -> AuthUser:85    token = _extract_token(request)86    if not token:87        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Connexion requise.")88    payload = decode_token(token, settings)89    if not payload or payload.get("kind") != "access":90        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Session expirée. Reconnecte-toi.")91    return AuthUser(str(payload["sub"]), str(payload.get("role", "student")))929394async def get_optional_user(request: Request,95                            settings: Settings = Depends(get_settings)) -> AuthUser | None:96    try:97        return await get_current_user(request, settings)98    except HTTPException:99        return None100101102async def require_professor(user: AuthUser = Depends(get_current_user)) -> AuthUser:103    if not user.is_professor:104        raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Réservé au professeur.")105    return user106107108async def require_admin(user: AuthUser = Depends(get_current_user)) -> AuthUser:109    if not user.is_admin:110        raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Réservé à l'administrateur.")111    return user112113114# ------------------------------------------------------------------ passwords (PBKDF2-SHA256)115def hash_password(password: str, iterations: int = 310_000) -> str:116    salt = secrets.token_bytes(16)117    dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, iterations)118    return f"pbkdf2${iterations}${base64.b64encode(salt).decode()}${base64.b64encode(dk).decode()}"119120121def verify_password(password: str, stored: str | None) -> bool:122    if not stored or not stored.startswith("pbkdf2$"):123        return False124    try:125        _, it, salt_b64, dk_b64 = stored.split("$", 3)126        dk = hashlib.pbkdf2_hmac("sha256", password.encode(), base64.b64decode(salt_b64), int(it))127        return hmac.compare_digest(dk, base64.b64decode(dk_b64))128    except (ValueError, TypeError):129        return False130