Spaces:
Runtime error
Runtime error
Claude Opus 5 (1M context)
feat(auth): inscription publique, connexion Google, et conversations rattachees a un compte
d21ea1e | """ | |
| Authentication utilities for JuriX. | |
| Provides JWT token generation, password hashing, and user authentication. | |
| Author: JuriX Team | |
| Date: 2026-01-12 | |
| """ | |
| import base64 | |
| import hashlib | |
| from datetime import datetime, timedelta, timezone | |
| from typing import Optional | |
| import bcrypt | |
| from fastapi import Depends, HTTPException, status | |
| from fastapi.security import OAuth2PasswordBearer | |
| from jose import JWTError, jwt | |
| from sqlalchemy import select | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| from app.core.config import settings | |
| from app.core.database import get_db | |
| from app.models.user import User | |
| # ==================== HACHAGE DES MOTS DE PASSE ==================== | |
| # | |
| # bcrypt est utilise directement, sans passlib. passlib 1.7.4 (derniere version, | |
| # 2020) lit `bcrypt.__about__.__version__` pour detecter le backend — attribut | |
| # supprime depuis bcrypt 4.1. Avec bcrypt 5.x, le hachage echouait totalement | |
| # avec "password cannot be longer than 72 bytes" MEME pour un mot de passe de | |
| # 13 caracteres. L'authentification etait donc inutilisable telle que declaree. | |
| # | |
| # bcrypt ignore silencieusement tout ce qui depasse 72 octets : deux mots de | |
| # passe partageant les 72 premiers octets seraient equivalents. On pre-hache donc | |
| # en SHA-256 puis on encode en base64, ce qui donne une entree de longueur fixe | |
| # (44 octets) et supporte les phrases de passe de n'importe quelle longueur. | |
| def _prepare(password: str) -> bytes: | |
| """Normalise un mot de passe en une entree bcrypt de longueur fixe.""" | |
| digest = hashlib.sha256(password.encode("utf-8")).digest() | |
| return base64.b64encode(digest) | |
| # OAuth2 scheme | |
| oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") | |
| # Variante tolerante : `auto_error=False` rend None au lieu de lever 401 quand | |
| # l'en-tete Authorization est absent. Utilisee par `get_current_user_optional`. | |
| _oauth2_optionnel = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login", auto_error=False) | |
| # JWT settings — lus depuis la configuration. | |
| # Auparavant ALGORITHM et ACCESS_TOKEN_EXPIRE_MINUTES etaient codes en dur ici | |
| # (7 jours) alors que config.py declare 30 minutes : les deux valeurs se | |
| # contredisaient et celle du fichier de configuration n'avait aucun effet. | |
| # Le garde-fou hasattr etait vestigial : SECRET_KEY existe toujours. | |
| SECRET_KEY = settings.SECRET_KEY | |
| ALGORITHM = settings.ALGORITHM | |
| ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES | |
| def hash_password(password: str) -> str: | |
| """ | |
| Hash a password using bcrypt. | |
| Args: | |
| password: Plain text password | |
| Returns: | |
| Hashed password | |
| """ | |
| return bcrypt.hashpw(_prepare(password), bcrypt.gensalt()).decode("utf-8") | |
| def verify_password(plain_password: str, hashed_password: str) -> bool: | |
| """ | |
| Verify a password against its hash. | |
| Args: | |
| plain_password: Plain text password to verify | |
| hashed_password: Bcrypt hash to verify against | |
| Returns: | |
| True if password matches, False otherwise | |
| """ | |
| # Un compte cree par Google n'a PAS de mot de passe : `hashed_password` est | |
| # NULL. Sans ce controle, `.encode()` leve un AttributeError — que le | |
| # `except (ValueError, TypeError)` ci-dessous n'attrape pas — et la simple | |
| # tentative de connexion par mot de passe sur un compte Google repondait 500. | |
| if not hashed_password: | |
| return False | |
| try: | |
| return bcrypt.checkpw(_prepare(plain_password), hashed_password.encode("utf-8")) | |
| except (ValueError, TypeError): | |
| # Empreinte illisible ou tronquee : on refuse plutot que de propager. | |
| return False | |
| def duree_de_session(user: User) -> timedelta: | |
| """ | |
| Duree de vie du jeton, selon le privilege du compte. | |
| Les valeurs sont lues sur `settings` A L'APPEL et non a l'import : les | |
| constantes de module ci-dessus sont figees au chargement, et un test qui | |
| modifierait le reglage ne verrait rien changer. | |
| """ | |
| minutes = ( | |
| settings.ADMIN_TOKEN_EXPIRE_MINUTES | |
| if user.is_admin() | |
| else settings.ACCESS_TOKEN_EXPIRE_MINUTES | |
| ) | |
| return timedelta(minutes=minutes) | |
| def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: | |
| """ | |
| Create a JWT access token. | |
| Args: | |
| data: Data to encode in token (typically {"sub": user_email}) | |
| expires_delta: Optional expiration time delta | |
| Returns: | |
| Encoded JWT token | |
| """ | |
| to_encode = data.copy() | |
| if expires_delta: | |
| expire = datetime.now(timezone.utc) + expires_delta | |
| else: | |
| expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) | |
| 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), db: AsyncSession = Depends(get_db) | |
| ) -> User: | |
| """ | |
| Get current authenticated user from JWT token. | |
| Args: | |
| token: JWT token from Authorization header | |
| db: Database session | |
| Returns: | |
| Authenticated User object | |
| Raises: | |
| HTTPException: If token is invalid or user not found | |
| """ | |
| 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 | |
| # Get user from database | |
| query = select(User).where(User.email == email) | |
| result = await db.execute(query) | |
| user = result.scalar_one_or_none() | |
| if user is None: | |
| raise credentials_exception | |
| if not user.is_active: | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, detail="User account is inactive" | |
| ) | |
| return user | |
| async def get_current_user_optional( | |
| token: Optional[str] = Depends(_oauth2_optionnel), db: AsyncSession = Depends(get_db) | |
| ) -> Optional[User]: | |
| """ | |
| L'utilisateur courant, ou None quand la requete est anonyme. | |
| Existe pour les routes qui doivent servir les deux publics — le chat reste | |
| utilisable sans compte, mais doit rattacher la conversation quand il y en a | |
| un. | |
| LA DISTINCTION QUI COMPTE : pas d'en-tete `Authorization` rend `None` ; un | |
| en-tete PRESENT mais dont le jeton est invalide ou expire leve 401, jamais | |
| `None`. Degrader silencieusement en anonyme serait le pire comportement : | |
| l'utilisateur croirait sa conversation enregistree alors qu'elle partirait | |
| en `user_id` NULL, invisible dans sa liste. Le front sait deja traiter un | |
| 401 (`apiFetch` : deconnexion puis redirection). | |
| """ | |
| if token is None: | |
| return None | |
| return await get_current_user(token=token, db=db) | |
| async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User: | |
| """ | |
| Get current active user. | |
| Args: | |
| current_user: Current user from token | |
| Returns: | |
| Active user | |
| Raises: | |
| HTTPException: If user is inactive | |
| """ | |
| if not current_user.is_active: | |
| raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user") | |
| return current_user | |
| async def get_current_admin_user(current_user: User = Depends(get_current_user)) -> User: | |
| """ | |
| Get current user with admin role. | |
| Args: | |
| current_user: Current user from token | |
| Returns: | |
| Admin user | |
| Raises: | |
| HTTPException: If user doesn't have admin role | |
| """ | |
| if not current_user.is_admin(): | |
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") | |
| return current_user | |
| async def get_current_superadmin_user(current_user: User = Depends(get_current_user)) -> User: | |
| """ | |
| Get current user with superadmin role. | |
| Args: | |
| current_user: Current user from token | |
| Returns: | |
| Superadmin user | |
| Raises: | |
| HTTPException: If user doesn't have superadmin role | |
| """ | |
| if not current_user.is_superadmin(): | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, detail="Superadmin access required" | |
| ) | |
| return current_user | |