Etape 11 : authentification reelle branchee (demande n°2)

app/core/auth.py (188 lignes, JWT + bcrypt + roles) etait complet et importe
par PERSONNE. Les 9 endpoints d'administration passaient par deux stubs
renvoyant {"id": 1, "role": "admin"} : ils etaient ouverts a tous.

BUG BLOQUANT decouvert en chemin : passlib 1.7.4 (derniere version, 2020) est
incompatible avec bcrypt >= 4.1 — il lit bcrypt.__about__.__version__, supprime
depuis. Avec bcrypt 5.x installe, hash_password echouait avec "password cannot
be longer than 72 bytes" MEME pour 13 caracteres : l'authentification aurait ete
inutilisable telle que declaree.
-> bcrypt utilise directement, passlib retire des manifestes. Pre-hachage
SHA-256 + base64 avant bcrypt : entree de longueur fixe, donc support des
phrases de passe longues sans la troncature silencieuse a 72 octets de bcrypt
(deux mots de passe partageant leurs 72 premiers octets etaient equivalents).

Ajouts :
- app/api/routes/auth.py : POST /login (formulaire OAuth2, l'URL exacte attendue
par oauth2_scheme, donc le bouton Authorize de /docs fonctionne),
POST /login/json (pour le front), GET /me, POST /logout.
Meme message d'erreur pour email inconnu et mot de passe faux : les distinguer
permettrait d'enumerer les comptes. Pas d'inscription publique.
- scripts/create_admin.py, idempotent, valide le mot de passe via UserCreate pour
que le script et l'API appliquent la meme politique.

Corrections :
- core/auth.py codait ALGORITHM et ACCESS_TOKEN_EXPIRE_MINUTES en dur (7 jours)
alors que config.py declare 30 minutes : la configuration n'avait aucun effet.
- datetime.utcnow (deprecie) -> datetime.now(timezone.utc), stocke en UTC naif
car les colonnes DateTime du schema sont sans fuseau.
- admin.py : liste _mock_users en memoire remplacee par le modele User ; les
schemas locaux masquaient app/schemas/user.py et le UserCreate local n'avait
AUCUN champ mot de passe (un compte cree par l'API n'aurait jamais pu se
connecter). Garde-fous ajoutes : pas d'auto-elevation, pas d'auto-suppression,
pas de suppression du dernier superadmin actif.

Endpoints fermes (12/12 verifies a 401 en acces anonyme) : ocr/detect,
ocr/process, upload/{id} DELETE, upload/cleanup, search/reindex, search/stats,
categories POST/PUT/DELETE, batch-upload upload+status, laws/admin *, admin/users.

FAILLE CORRIGEE : /ocr/detect et /ocr/process acceptaient un chemin serveur
arbitraire et renvoyaient le texte de tout fichier lisible par Tesseract.
resolve_upload_path() contraint desormais le fichier au repertoire d'upload —
motif strict, resolution des liens symboliques, verification is_file. Refus
verifie meme AVEC un jeton administrateur : l'auth seule ne suffisait pas.

Fixtures d'authentification ajoutees au conftest. Surcharger get_current_user
suffit aux trois niveaux d'acces, qui en dependent tous.

Verifie de bout en bout : mauvais mot de passe 401, email inconnu 401 (message
identique), connexion 200, /auth/me 200, endpoint admin 200 avec jeton, jeton
invalide 401, path traversal 400, connexion formulaire 200.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

app/api/routes/admin.py

@@ -19,9 +19,17 @@ from typing import Any, Dict, List, TypedDict, cast
 
 from fastapi import APIRouter, Depends, HTTPException, status
 from pydantic import BaseModel, EmailStr, Field
+from sqlalchemy import or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.auth import (
+    get_current_admin_user,
+    get_current_superadmin_user,
+    hash_password,
+)
 from app.core.database import get_db, health_check_db
+from app.models.user import User
+from app.schemas.user import UserCreate, UserResponse, UserUpdate
 
 logger = logging.getLogger(__name__)
 
@@ -29,84 +37,15 @@ router = APIRouter(tags=["Admin"])
 
 
 # ==================== SCHEMAS ====================
-
-
-class UserCreate(BaseModel):
-    """Schema for creating a user."""
-    
-    email: EmailStr = Field(..., description="User email")
-    username: str = Field(..., min_length=3, max_length=50, description="Username")
-    role: str = Field(default="user", description="User role (user or admin)")
-    is_active: bool = Field(default=True, description="User active status")
-
-
-class UserUpdate(BaseModel):
-    """Schema for updating a user."""
-    
-    email: EmailStr | None = None
-    username: str | None = Field(None, min_length=3, max_length=50)
-    role: str | None = None
-    is_active: bool | None = None
-
-
-class UserResponse(BaseModel):
-    """Schema for user response."""
-    
-    id: int
-    email: str
-    username: str
-    role: str
-    is_active: bool
-    created_at: datetime
- 
- 
-class UserDict(TypedDict):
-    """Internal dictionary type for mock storage."""
-    id: int
-    email: str
-    username: str
-    role: str
-    is_active: bool
-    created_at: datetime
-
-
-# ==================== MOCK USER STORAGE ====================
-# TODO: Replace with actual database model
-
-
-_mock_users: List[UserDict] = [
-    {
-        "id": 1,
-        "email": "admin@jurix.cm",
-        "username": "admin",
-        "role": "admin",
-        "is_active": True,
-        "created_at": datetime.now(),
-    },
-    {
-        "id": 2,
-        "email": "user@jurix.cm",
-        "username": "user1",
-        "role": "user",
-        "is_active": True,
-        "created_at": datetime.now(),
-    },
-]
-
-_next_user_id = 3
-
-
-# ==================== AUTHENTICATION ====================
-
-
-async def get_current_admin():
-    """
-    Dependency for admin authentication.
-    
-    TODO: Implement actual JWT authentication.
-    For now, returns mock admin user.
-    """
-    return {"id": 1, "role": "admin", "username": "admin"}
+#
+# Les schemas UserCreate / UserUpdate / UserResponse etaient redefinis ici et
+# MASQUAIENT ceux de app/schemas/user.py. Le UserCreate local n'avait aucun champ
+# mot de passe : un utilisateur cree par cette API n'aurait jamais pu se
+# connecter. Ils sont desormais importes depuis la source unique.
+#
+# La liste _mock_users (en memoire, perdue a chaque redemarrage, non partagee
+# entre workers) a ete supprimee au profit du modele User et de la table users,
+# qui existaient tous deux depuis la migration 0b21fb6a3651 sans etre utilises.
 
 
 # ==================== USER MANAGEMENT ====================
@@ -116,191 +55,155 @@ async def get_current_admin():
 async def list_users(
     skip: int = 0,
     limit: int = 100,
-    current_admin: dict = Depends(get_current_admin),
-) -> List[UserResponse]:
-    """
-    List all users (admin only).
-    
-    **Parameters:**
-    - skip: Number of records to skip
-    - limit: Maximum records to return
-    
-    **Returns:**
-    - List of users
-    
-    **Requires:** Admin authentication
-    """
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_admin_user),
+) -> List[User]:
+    """Liste les comptes (administrateurs)."""
     logger.info(f"📋 GET /admin/users - skip={skip}, limit={limit}")
-    
-    users = _mock_users[skip : skip + limit]
-    return [
-        UserResponse(
-            id=u["id"],
-            email=u["email"],
-            username=u["username"],
-            role=u["role"],
-            is_active=u["is_active"],
-            created_at=u["created_at"]
-        ) 
-        for u in users
-    ]
+    result = await db.execute(
+        select(User).order_by(User.id).offset(skip).limit(min(limit, 500))
+    )
+    return list(result.scalars().all())
 
 
 @router.post("/users", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
 async def create_user(
-    user: UserCreate,
-    current_admin: dict = Depends(get_current_admin),
-) -> UserResponse:
+    payload: UserCreate,
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_admin_user),
+) -> User:
     """
-    Create a new user (admin only).
-    
-    **Parameters:**
-    - email: User email (unique)
-    - username: Username (unique)
-    - role: User role (user or admin)
-    - is_active: Active status
-    
-    **Returns:**
-    - Created user
-    
-    **Raises:**
-    - 409: Email or username already exists
-    
-    **Requires:** Admin authentication
+    Crée un compte (administrateurs).
+
+    Un administrateur ne peut pas créer de compte plus privilégié que le sien :
+    sans cette règle, tout admin pourrait se hisser au rang de superadmin en
+    créant un second compte.
     """
-    global _next_user_id
-    
-    assert user is not None, "UserCreate must not be None"
-    assert isinstance(user.email, str) and len(user.email) > 0, "Email must be a non-empty string"
-    assert isinstance(user.username, str) and len(user.username) >= 3, "Username must be at least 3 characters"
+    if payload.role in ("admin", "superadmin") and not current_admin.is_superadmin():
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Seul un superadmin peut créer un compte privilégié",
+        )
 
-    logger.info(f"➕ POST /admin/users - Creating user: {user.email}")
-    
-    # Check for duplicate email or username
-    for existing_user in _mock_users:
-        if existing_user["email"] == user.email:
-            raise HTTPException(
-                status_code=status.HTTP_409_CONFLICT,
-                detail=f"Email already exists: {user.email}",
-            )
-        if existing_user["username"] == user.username:
-            raise HTTPException(
-                status_code=status.HTTP_409_CONFLICT,
-                detail=f"Username already exists: {user.username}",
-            )
-    
-    # Create new user
-    new_user: UserDict = {
-        "id": _next_user_id,
-        "email": str(user.email),
-        "username": user.username,
-        "role": user.role,
-        "is_active": user.is_active,
-        "created_at": datetime.now(),
-    }
-    
-    _mock_users.append(new_user)
-    _next_user_id += 1
-    
-    logger.info(f"✅ User created: ID={new_user['id']}, email={new_user['email']}")
-    return UserResponse(
-        id=new_user["id"],
-        email=new_user["email"],
-        username=new_user["username"],
-        role=new_user["role"],
-        is_active=new_user["is_active"],
-        created_at=new_user["created_at"]
+    existing = await db.execute(
+        select(User).where(
+            or_(User.email == payload.email, User.username == payload.username)
+        )
     )
+    if existing.scalar_one_or_none() is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Un compte existe déjà avec cet email ou ce nom d'utilisateur",
+        )
+
+    user = User(
+        email=payload.email,
+        username=payload.username,
+        full_name=payload.full_name,
+        role=payload.role,
+        is_active=payload.is_active,
+        hashed_password=hash_password(payload.password),
+    )
+    db.add(user)
+    await db.commit()
+    await db.refresh(user)
+    logger.info(f"✅ Compte créé : {user.email} (rôle {user.role})")
+    return user
 
 
 @router.put("/users/{user_id}", response_model=UserResponse)
 async def update_user(
     user_id: int,
-    user_update: UserUpdate,
-    current_admin: dict = Depends(get_current_admin),
-) -> UserResponse:
+    payload: UserUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_admin_user),
+) -> User:
     """
-    Update a user (admin only).
-    
-    **Parameters:**
-    - user_id: User ID
-    - Partial update fields
-    
-    **Returns:**
-    - Updated user
-    
-    **Raises:**
-    - 404: User not found
-    
-    **Requires:** Admin authentication
+    Met à jour un compte (administrateurs).
+
+    Deux garde-fous : seul un superadmin peut modifier un rôle, et personne ne
+    peut se retirer ses propres droits — sinon la première erreur de manipulation
+    ferme définitivement l'accès à l'instance.
     """
-    assert isinstance(user_id, int) and user_id > 0, "user_id must be a positive integer"
-    assert user_update is not None, "UserUpdate must not be None"
+    user = (
+        await db.execute(select(User).where(User.id == user_id))
+    ).scalar_one_or_none()
+    if user is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail=f"Compte {user_id} introuvable"
+        )
 
-    logger.info(f"✏️  PUT /admin/users/{user_id}")
-    
-    # Find user
-    user = next((u for u in _mock_users if u["id"] == user_id), None)
-    
-    if not user:
+    changes = payload.model_dump(exclude_unset=True)
+
+    if "role" in changes and not current_admin.is_superadmin():
         raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"User not found: {user_id}",
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Seul un superadmin peut modifier un rôle",
+        )
+    if user.id == current_admin.id and (
+        changes.get("role") not in (None, current_admin.role)
+        or changes.get("is_active") is False
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Impossible de réduire ses propres droits",
         )
-    
-    # Update fields
-    update_data = user_update.model_dump(exclude_unset=True)
-    user_raw = cast(Dict[str, Any], user)
-    for field, value in update_data.items():
-        user_raw[field] = value
-    
-    logger.info(f"✅ User {user_id} updated")
-    return UserResponse(
-        id=user["id"],
-        email=user["email"],
-        username=user["username"],
-        role=user["role"],
-        is_active=user["is_active"],
-        created_at=user["created_at"]
-    )
+
+    password = changes.pop("password", None)
+    if password:
+        user.hashed_password = hash_password(password)
+    for field, value in changes.items():
+        setattr(user, field, value)
+
+    await db.commit()
+    await db.refresh(user)
+    logger.info(f"✏️ Compte mis à jour : {user.email}")
+    return user
 
 
 @router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_user(
     user_id: int,
-    current_admin: dict = Depends(get_current_admin),
-):
-    """
-    Delete a user (admin only).
-    
-    **Parameters:**
-    - user_id: User ID
-    
-    **Returns:**
-    - 204 No Content
-    
-    **Raises:**
-    - 404: User not found
-    
-    **Requires:** Admin authentication
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_superadmin_user),
+) -> None:
     """
-    assert isinstance(user_id, int) and user_id > 0, "user_id must be a positive integer"
+    Supprime un compte (superadministrateurs).
 
-    logger.info(f"🗑️  DELETE /admin/users/{user_id}")
-    
-    # Find user
-    user = next((u for u in _mock_users if u["id"] == user_id), None)
-    
-    if not user:
+    Refuse la suppression de soi-même et celle du dernier superadmin actif : les
+    deux mènent à une instance dont plus personne n'a les clés.
+    """
+    user = (
+        await db.execute(select(User).where(User.id == user_id))
+    ).scalar_one_or_none()
+    if user is None:
         raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"User not found: {user_id}",
+            status_code=status.HTTP_404_NOT_FOUND, detail=f"Compte {user_id} introuvable"
         )
-    
-    # Remove user
-    _mock_users.remove(user)
-    
-    logger.info(f"✅ User {user_id} deleted")
+    if user.id == current_admin.id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Impossible de supprimer son propre compte",
+        )
+    if user.role == "superadmin":
+        remaining = (
+            await db.execute(
+                select(User).where(
+                    User.role == "superadmin",
+                    User.is_active.is_(True),
+                    User.id != user.id,
+                )
+            )
+        ).scalars().first()
+        if remaining is None:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Impossible de supprimer le dernier superadmin actif",
+            )
+
+    await db.delete(user)
+    await db.commit()
+    logger.info(f"🗑️ Compte supprimé : {user.email}")
     return None
 
 
@@ -310,7 +213,7 @@ async def delete_user(
 @router.get("/system")
 async def get_system_info(
     db: AsyncSession = Depends(get_db),
-    current_admin: dict = Depends(get_current_admin),
+    current_admin: User = Depends(get_current_admin_user),
 ) -> Dict:
     """
     Get system information (admin only).

app/api/routes/auth.py

@@ -0,0 +1,133 @@
+"""
+Routes d'authentification.
+
+`app/core/auth.py` implémentait déjà l'intégralité de la chaîne JWT — hachage
+bcrypt, création de jeton, dépendances par rôle — mais n'était importé nulle
+part, et aucune route de connexion n'existait. Les endpoints « admin » passaient
+par deux stubs renvoyant un dictionnaire en dur.
+
+Deux points d'entrée de connexion, volontairement :
+
+- `POST /login` accepte un formulaire OAuth2. C'est exactement l'URL déclarée
+  par `oauth2_scheme` (`tokenUrl="/api/v1/auth/login"`), donc le bouton
+  *Authorize* de /docs fonctionne — précieux pour administrer l'instance tant
+  que l'interface d'administration n'est pas terminée.
+- `POST /login/json` accepte du JSON, plus naturel pour le front SvelteKit.
+
+Les deux partagent le même `_authenticate`.
+
+Il n'y a **pas** de route d'inscription publique : la création de comptes passe
+par `POST /api/v1/admin/users` ou par `scripts/create_admin.py`. Une inscription
+ouverte donnerait à n'importe qui l'accès au RAG, donc à la facturation Gemini.
+
+Author: JuriX Team
+"""
+
+import logging
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.auth import (
+    create_access_token,
+    get_current_active_user,
+    verify_password,
+)
+from app.core.database import get_db
+from app.models.user import User
+from app.schemas.user import UserLogin, UserResponse, UserWithToken
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# Message unique pour « email inconnu » et « mot de passe faux » : les
+# distinguer permettrait d'énumérer les comptes existants.
+_INVALID = "Identifiants invalides"
+
+
+async def _authenticate(db: AsyncSession, email: str, password: str) -> User:
+    """
+    Vérifie un couple email / mot de passe.
+
+    Raises:
+        HTTPException 401: identifiants invalides
+        HTTPException 403: compte désactivé
+    """
+    result = await db.execute(select(User).where(User.email == email.lower().strip()))
+    user = result.scalar_one_or_none()
+
+    if user is None or not verify_password(password, user.hashed_password):
+        logger.warning(f"🔒 Échec de connexion pour {email!r}")
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=_INVALID,
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    if not user.is_active:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail="Compte désactivé"
+        )
+
+    # Les colonnes DateTime du modele sont SANS fuseau : y ecrire un datetime
+    # avec fuseau leve "can't subtract offset-naive and offset-aware datetimes"
+    # au moment du flush. On stocke donc de l'UTC naif, comme le reste du schema.
+    user.last_login_at = datetime.now(timezone.utc).replace(tzinfo=None)
+    await db.commit()
+    await db.refresh(user)
+
+    logger.info(f"✅ Connexion de {user.email} (rôle {user.role})")
+    return user
+
+
+@router.post("/login", response_model=UserWithToken, status_code=status.HTTP_200_OK)
+async def login(
+    form_data: OAuth2PasswordRequestForm = Depends(),
+    db: AsyncSession = Depends(get_db),
+) -> UserWithToken:
+    """
+    Connexion au format formulaire OAuth2 (utilisée par /docs).
+
+    Le champ `username` porte l'adresse email.
+    """
+    user = await _authenticate(db, form_data.username, form_data.password)
+    return UserWithToken(
+        **UserResponse.model_validate(user).model_dump(),
+        access_token=create_access_token({"sub": user.email}),
+    )
+
+
+@router.post("/login/json", response_model=UserWithToken, status_code=status.HTTP_200_OK)
+async def login_json(
+    credentials: UserLogin,
+    db: AsyncSession = Depends(get_db),
+) -> UserWithToken:
+    """Connexion au format JSON (utilisée par le front)."""
+    user = await _authenticate(db, credentials.email, credentials.password)
+    return UserWithToken(
+        **UserResponse.model_validate(user).model_dump(),
+        access_token=create_access_token({"sub": user.email}),
+    )
+
+
+@router.get("/me", response_model=UserResponse, status_code=status.HTTP_200_OK)
+async def read_me(current_user: User = Depends(get_current_active_user)) -> User:
+    """Renvoie l'utilisateur associé au jeton fourni."""
+    return current_user
+
+
+@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
+async def logout(current_user: User = Depends(get_current_active_user)) -> None:
+    """
+    Déconnexion.
+
+    Les JWT sont sans état : rien n'est révoqué côté serveur, le client jette son
+    jeton. L'endpoint existe pour donner un point d'appel explicite au front et
+    pour tracer la déconnexion.
+    """
+    logger.info(f"👋 Déconnexion de {current_user.email}")
+    return None

app/api/routes/batch_upload.py

@@ -21,6 +21,8 @@ from app.core.database import AsyncSessionLocal, get_db
 from app.models.law import Law
 from app.tasks.process_law import process_law_async
 from app.core.config import settings
+from app.core.auth import get_current_admin_user
+from app.models.user import User
 
 logger = logging.getLogger(__name__)
 
@@ -89,7 +91,8 @@ async def websocket_endpoint(websocket: WebSocket, session_id: str):
 async def batch_upload(
     files: List[UploadFile] = File(...),
     session_id: str | None = None,
-    db: AsyncSession = Depends(get_db)
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_admin_user),
 ):
     """
     Batch upload multiple PDF files.
@@ -241,7 +244,8 @@ async def process_batch(laws: List[dict], session_id: str) -> None:
 
                 law.status = "processing"
                 law.processing_progress = 0
-                law.processing_started_at = datetime.now(timezone.utc)
+                # UTC naif : la colonne est un DateTime sans fuseau.
+                law.processing_started_at = datetime.now(timezone.utc).replace(tzinfo=None)
                 await db.commit()
 
             await manager.send_progress(session_id, {
@@ -305,7 +309,8 @@ async def process_batch(laws: List[dict], session_id: str) -> None:
 @router.get("/status")
 async def get_batch_status(
     status: str | None = None,
-    db: AsyncSession = Depends(get_db)
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_admin_user),
 ):
     """
     Get all laws with optional status filter.

app/api/routes/categories.py

@@ -31,6 +31,8 @@ from app.services.category_service import (
     CategoryServiceError
 )
 from app.schemas.law import CategoryCreate, CategoryUpdate, CategoryResponse, CategoryStats
+from app.core.auth import get_current_admin_user
+from app.models.user import User
 
 # Configure logger
 logger = logging.getLogger(__name__)
@@ -59,7 +61,8 @@ router = APIRouter(
 )
 async def create_category(
     category_data: CategoryCreate,
-    db: AsyncSession = Depends(get_db)
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_admin_user),
 ) -> CategoryResponse:
     """
     Create a new legal category.
@@ -274,7 +277,8 @@ async def get_category(
 async def update_category(
     category_id: int,
     category_data: CategoryUpdate,
-    db: AsyncSession = Depends(get_db)
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_admin_user),
 ) -> CategoryResponse:
     """
     Update a category.
@@ -341,7 +345,8 @@ async def update_category(
 async def delete_category(
     category_id: int,
     force: bool = Query(False, description="Force delete even if laws are associated"),
-    db: AsyncSession = Depends(get_db)
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_admin_user),
 ):
     """
     Delete a category.

app/api/routes/laws.py

@@ -23,6 +23,8 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
+from app.core.auth import get_current_admin_user
+from app.models.user import User
 from app.core.database import get_db, AsyncSessionLocal
 from app.models.law import Category, Law
 from app.schemas.law import (
@@ -54,17 +56,10 @@ router = APIRouter(tags=["Laws"])
 # ==================== DEPENDENCIES ====================
 
 
-# TODO: Implement authentication
-# For now, this is a placeholder
-async def get_current_admin_user():
-    """
-    Dependency for admin authentication.
-
-    TODO: Implement actual authentication with JWT tokens.
-    For now, returns a mock admin user.
-    """
-    # In production, verify JWT token and check admin role
-    return {"id": 1, "role": "admin"}
+# L'authentification reelle vit dans app/core/auth.py (JWT + bcrypt + roles).
+# Un stub renvoyant {"id": 1, "role": "admin"} occupait cette place : les quatre
+# endpoints d'administration ci-dessous etaient donc ouverts a tous.
+# get_current_admin_user est importe en tete de fichier.
 
 
 # ==================== PUBLIC ENDPOINTS ====================
@@ -502,7 +497,7 @@ async def get_law_pdf_page_image(
 async def ingest_law(
     request: LawIngestRequest,
     db: AsyncSession = Depends(get_db),
-    current_user: dict = Depends(get_current_admin_user),
+    current_user: User = Depends(get_current_admin_user),
 ):
     """
     Ingest a new law from an uploaded file.
@@ -632,7 +627,7 @@ async def ingest_law(
 async def create_law(
     law: LawCreate,
     db: AsyncSession = Depends(get_db),
-    current_user: dict = Depends(get_current_admin_user),
+    current_user: User = Depends(get_current_admin_user),
 ):
     """
     Create a new law (admin only).
@@ -709,7 +704,7 @@ async def update_law(
     law_id: int,
     law_update: LawUpdate,
     db: AsyncSession = Depends(get_db),
-    current_user: dict = Depends(get_current_admin_user),
+    current_user: User = Depends(get_current_admin_user),
 ):
     """
     Update an existing law (admin only).
@@ -771,7 +766,7 @@ async def update_law(
 async def delete_law(
     law_id: int,
     db: AsyncSession = Depends(get_db),
-    current_user: dict = Depends(get_current_admin_user),
+    current_user: User = Depends(get_current_admin_user),
 ):
     """
     Delete a law (admin only).

app/api/routes/ocr.py

@@ -13,7 +13,7 @@ Date: 2026-01-11
 import logging
 from pathlib import Path
 
-from fastapi import APIRouter, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.responses import JSONResponse
 
 from app.schemas.ocr import (
@@ -23,6 +23,9 @@ from app.schemas.ocr import (
     PDFTypeDetectionRequest,
     PDFTypeDetectionResult,
 )
+from app.core.auth import get_current_admin_user
+from app.models.user import User
+from app.utils.file_utils import resolve_upload_path
 from app.services.ocr_service import OCRError, OCRService
 
 logger = logging.getLogger(__name__)
@@ -65,7 +68,10 @@ def get_ocr_service() -> OCRService:
     - OCR recommendation
     """,
 )
-async def detect_pdf_type(request: PDFTypeDetectionRequest) -> PDFTypeDetectionResult:
+async def detect_pdf_type(
+    request: PDFTypeDetectionRequest,
+    current_admin: User = Depends(get_current_admin_user),
+) -> PDFTypeDetectionResult:
     """
     Detect PDF type without processing.
     
@@ -81,19 +87,24 @@ async def detect_pdf_type(request: PDFTypeDetectionRequest) -> PDFTypeDetectionR
         500: Detection error
     """
     assert request is not None, "PDFTypeDetectionRequest must not be None"
-    assert isinstance(request.pdf_path, str) and len(request.pdf_path) > 0, "pdf_path must be a non-empty string"
 
-    logger.info(f"🔍 PDF type detection request: {request.pdf_path}")
+    logger.info(f"🔍 PDF type detection request: {request.file_id}")
     
-    pdf_path = Path(request.pdf_path)
-    
-    # Validate path
-    if not pdf_path.exists():
+    # resolve_upload_path contraint le fichier au repertoire d'upload.
+    # Auparavant : Path(request.pdf_path) sur un chemin serveur arbitraire, sans
+    # authentification — enumeration et lecture de n'importe quel fichier lisible.
+    try:
+        pdf_path = resolve_upload_path(request.file_id)
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    except FileNotFoundError:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"PDF file not found: {request.pdf_path}",
+            detail=f"Fichier introuvable : {request.file_id}",
         )
     
+    # Validate path
+    
     try:
         service = get_ocr_service()
         result = await service.detect_pdf_type(pdf_path)
@@ -126,7 +137,10 @@ async def detect_pdf_type(request: PDFTypeDetectionRequest) -> PDFTypeDetectionR
     **Languages:** French (fra), English (eng), or both (fra+eng)
     """,
 )
-async def process_pdf(request: OCRRequest) -> OCRResult:
+async def process_pdf(
+    request: OCRRequest,
+    current_admin: User = Depends(get_current_admin_user),
+) -> OCRResult:
     """
     Process PDF with OCR.
     
@@ -142,22 +156,27 @@ async def process_pdf(request: OCRRequest) -> OCRResult:
         500: Processing error
     """
     assert request is not None, "OCRRequest must not be None"
-    assert isinstance(request.pdf_path, str) and len(request.pdf_path) > 0, "pdf_path must be a non-empty string"
 
     logger.info(
-        f"📄 OCR processing request: {request.pdf_path} "
+        f"📄 OCR processing request: {request.file_id} "
         f"(lang: {request.language}, dpi: {request.dpi})"
     )
     
-    pdf_path = Path(request.pdf_path)
-    
-    # Validate path
-    if not pdf_path.exists():
+    # resolve_upload_path contraint le fichier au repertoire d'upload.
+    # Auparavant : Path(request.pdf_path) sur un chemin serveur arbitraire, sans
+    # authentification — enumeration et lecture de n'importe quel fichier lisible.
+    try:
+        pdf_path = resolve_upload_path(request.file_id)
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    except FileNotFoundError:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"PDF file not found: {request.pdf_path}",
+            detail=f"Fichier introuvable : {request.file_id}",
         )
     
+    # Validate path
+    
     try:
         service = get_ocr_service()
         result = await service.process_pdf(

app/api/routes/search.py

@@ -26,6 +26,8 @@ from app.schemas.search import (
     SearchStats,
 )
 from app.services.search_service import SearchService, SearchServiceError
+from app.core.auth import get_current_admin_user
+from app.models.user import User
 
 logger = logging.getLogger(__name__)
 
@@ -345,7 +347,8 @@ async def find_article(
     # dependencies=[Depends(require_admin)]
 )
 async def reindex_all(
-    background_tasks: BackgroundTasks, search_service: SearchService = Depends(get_search_service)
+    background_tasks: BackgroundTasks, search_service: SearchService = Depends(get_search_service),
+    current_admin: User = Depends(get_current_admin_user),
 ) -> ReindexResponse:
     """
     Reconstruit les tsvector de toutes les lois publiées (admin uniquement).
@@ -434,7 +437,10 @@ async def reindex_all(
     response_model=SearchStats,
     status_code=status.HTTP_200_OK,
 )
-async def search_stats(db: AsyncSession = Depends(get_db)) -> SearchStats:
+async def search_stats(
+    db: AsyncSession = Depends(get_db),
+    current_admin: User = Depends(get_current_admin_user),
+) -> SearchStats:
     """
     Statistiques d'indexation (administrateurs).
 

app/api/routes/upload.py

@@ -15,11 +15,13 @@ import logging
 from pathlib import Path
 from typing import Dict
 
-from fastapi import APIRouter, File, HTTPException, UploadFile, status
+from fastapi import Depends, APIRouter, File, HTTPException, UploadFile, status
 from fastapi.responses import JSONResponse
 
 from app.schemas.file_upload import FileUploadResult, UploadServiceHealth
 from app.services.file_upload_service import FileUploadError, FileUploadService
+from app.core.auth import get_current_admin_user
+from app.models.user import User
 
 logger = logging.getLogger(__name__)
 
@@ -189,7 +191,10 @@ async def get_upload_status(file_id: str) -> Dict:
     summary="Delete uploaded file",
     description="Delete an uploaded file from temporary storage.",
 )
-async def delete_upload(file_id: str) -> Dict:
+async def delete_upload(
+    file_id: str,
+    current_admin: User = Depends(get_current_admin_user),
+) -> Dict:
     """
     Delete an uploaded file.
 
@@ -233,7 +238,10 @@ async def delete_upload(file_id: str) -> Dict:
     summary="Cleanup old files",
     description="Remove files older than specified hours (default: 24h).",
 )
-async def cleanup_old_files(max_age_hours: int = 24) -> Dict:
+async def cleanup_old_files(
+    max_age_hours: int = 24,
+    current_admin: User = Depends(get_current_admin_user),
+) -> Dict:
     """
     Cleanup old uploaded files.
 

app/core/auth.py

@@ -7,13 +7,16 @@ Author: JuriX Team
 Date: 2026-01-12
 """
 
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Optional
 
 from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
+import base64
+import hashlib
+
+import bcrypt
 from jose import JWTError, jwt
-from passlib.context import CryptContext
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -21,20 +24,36 @@ from app.core.config import settings
 from app.core.database import get_db
 from app.models.user import User
 
-# Password hashing
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+# ==================== 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")
 
-# JWT settings
-SECRET_KEY = (
-    settings.SECRET_KEY
-    if hasattr(settings, "SECRET_KEY")
-    else "dev-secret-key-change-in-production"
-)
-ALGORITHM = "HS256"
-ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7  # 7 days
+# 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:
@@ -47,7 +66,7 @@ def hash_password(password: str) -> str:
     Returns:
         Hashed password
     """
-    return pwd_context.hash(password)
+    return bcrypt.hashpw(_prepare(password), bcrypt.gensalt()).decode("utf-8")
 
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
@@ -61,7 +80,11 @@ def verify_password(plain_password: str, hashed_password: str) -> bool:
     Returns:
         True if password matches, False otherwise
     """
-    return pwd_context.verify(plain_password, hashed_password)
+    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 create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
@@ -77,9 +100,9 @@ def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -
     """
     to_encode = data.copy()
     if expires_delta:
-        expire = datetime.utcnow() + expires_delta
+        expire = datetime.now(timezone.utc) + expires_delta
     else:
-        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+        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)

app/main.py

@@ -7,7 +7,7 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from app.core.config import settings
 from app.core.database import close_db
-from app.api.routes import language, classifier, articles, search, rag, categories, personas, upload, ocr, laws, analytics, admin, batch_upload
+from app.api.routes import auth, language, classifier, articles, search, rag, categories, personas, upload, ocr, laws, analytics, admin, batch_upload
 
 logger = logging.getLogger(__name__)
 
@@ -75,6 +75,7 @@ async def health():
 
 
 # Include routers with global /api/v1 prefix
+app.include_router(auth.router, prefix="/api/v1/auth", tags=["auth"])
 app.include_router(language.router, prefix="/api/v1/language", tags=["language"])
 app.include_router(classifier.router, prefix="/api/v1/classifier", tags=["classifier"])
 app.include_router(search.router, prefix="/api/v1/search", tags=["search"])

app/schemas/ocr.py

@@ -112,7 +112,8 @@ class OCRRequest(BaseModel):
     Request schema for OCR processing.
     """
 
-    pdf_path: str = Field(..., min_length=1, description="Path to PDF file")
+    file_id: str = Field(
+        ..., min_length=1, description="Identifiant du fichier uploadé (jamais un chemin serveur)")
     language: str = Field(
         default="fra+eng",
         description="Tesseract language codes (e.g., 'fra', 'eng', 'fra+eng')",
@@ -138,7 +139,7 @@ class OCRRequest(BaseModel):
     class Config:
         json_schema_extra = {
             "example": {
-                "pdf_path": "/path/to/document.pdf",
+                "file_id": "/path/to/document.pdf",
                 "language": "fra+eng",
                 "dpi": 300,
                 "timeout_seconds": 60,
@@ -149,10 +150,11 @@ class OCRRequest(BaseModel):
 class PDFTypeDetectionRequest(BaseModel):
     """Request schema for PDF type detection only."""
 
-    pdf_path: str = Field(..., min_length=1, description="Path to PDF file")
+    file_id: str = Field(
+        ..., min_length=1, description="Identifiant du fichier uploadé (jamais un chemin serveur)")
 
     class Config:
-        json_schema_extra = {"example": {"pdf_path": "/path/to/document.pdf"}}
+        json_schema_extra = {"example": {"file_id": "/path/to/document.pdf"}}
 
 
 # ============================================================================

app/utils/file_utils.py

@@ -14,6 +14,7 @@ Date: 2026-01-11
 
 import hashlib
 import logging
+import re
 import uuid
 from pathlib import Path
 from typing import Optional, Tuple
@@ -353,3 +354,51 @@ def get_mime_type(file_type: str) -> str:
         raise ValueError(f"Unsupported file type: {file_type}")
     
     return MIME_TYPES[file_type]
+
+
+# ==================== RESOLUTION SECURISEE DES FICHIERS ====================
+
+_FILE_ID_RE = re.compile(r"^[A-Za-z0-9_-]{8,64}$")
+
+
+def resolve_upload_path(file_id: str) -> Path:
+    """
+    Résout un identifiant de fichier en chemin, à l'intérieur du répertoire d'upload.
+
+    Motivation : `/api/v1/ocr/detect` et `/api/v1/ocr/process` acceptaient un
+    `pdf_path` arbitraire — un chemin du système de fichiers du serveur — et se
+    contentaient d'un `.exists()`. N'importe qui pouvait donc énumérer les
+    fichiers de la machine (404 contre 500) et se faire renvoyer le texte de tout
+    document lisible par Tesseract. L'authentification seule ne suffit pas : un
+    jeton d'administration volé rouvrirait la faille.
+
+    Trois barrières indépendantes :
+      1. le motif rejette `..`, `/`, `%2f`, les chemins absolus et les octets nuls
+         AVANT tout accès au disque ;
+      2. `resolve()` + `is_relative_to` bloquent les évasions par lien symbolique ;
+      3. `is_file()` écarte les répertoires et les fichiers spéciaux.
+
+    Args:
+        file_id: Identifiant du fichier, sans extension
+
+    Returns:
+        Chemin absolu du fichier
+
+    Raises:
+        ValueError: identifiant invalide ou chemin sortant du répertoire
+        FileNotFoundError: aucun fichier correspondant
+    """
+    if not _FILE_ID_RE.match(file_id or ""):
+        raise ValueError(f"Identifiant de fichier invalide : {file_id!r}")
+
+    from app.services.file_upload_service import get_upload_service
+
+    root = get_upload_service().storage_path.resolve()
+    for extension in (".pdf", ".docx"):
+        candidate = (root / f"{file_id}{extension}").resolve()
+        if not candidate.is_relative_to(root):
+            raise ValueError("Le chemin sort du répertoire d'upload")
+        if candidate.is_file():
+            return candidate
+
+    raise FileNotFoundError(file_id)

pyproject.toml

@@ -25,7 +25,8 @@ pgvector = "^0.2.3"
 
 # Security
 python-jose = {extras = ["cryptography"], version = "^3.3.0"}
-passlib = {extras = ["bcrypt"], version = "^1.7.4"}
+# bcrypt direct : passlib est incompatible avec bcrypt >= 4.1
+bcrypt = ">=4.0,<6.0"
 python-multipart = "^0.0.6"
 
 # (Meilisearch replaced by PostgreSQL tsvector + pg_trgm)

requirements.txt

@@ -19,7 +19,10 @@ pgvector>=0.2.3,<1.0.0
 
 # ── Security ──────────────────────────────────────────────────────────────────
 python-jose[cryptography]>=3.3.0,<4.0.0
-passlib[bcrypt]>=1.7.4,<2.0.0
+# bcrypt utilise directement, sans passlib : passlib 1.7.4 (derniere version,
+# 2020) est incompatible avec bcrypt >= 4.1 (il lit bcrypt.__about__, supprime),
+# et le hachage echouait totalement avec bcrypt 5.x.
+bcrypt>=4.0,<6.0
 python-multipart>=0.0.6,<1.0.0
 
 # ── Document Processing ───────────────────────────────────────────────────────

scripts/create_admin.py

@@ -0,0 +1,134 @@
+"""
+Crée ou met à jour le compte administrateur initial.
+
+Aucun utilisateur n'était créé nulle part dans le projet : le modèle `User` et la
+table `users` existaient depuis la migration 0b21fb6a3651 sans qu'aucun code ne
+les alimente. Une fois l'authentification branchée, sans ce script il devient
+impossible de se connecter à sa propre instance.
+
+Idempotent : relancé, il ne duplique rien et ne modifie que ce qui doit l'être.
+
+Usage:
+    python scripts/create_admin.py
+    python scripts/create_admin.py --email admin@jurix.cm --username admin
+    python scripts/create_admin.py --reset-password
+    python scripts/create_admin.py --role superadmin
+
+Les valeurs par défaut viennent de .env : ADMIN_EMAIL, ADMIN_USERNAME,
+ADMIN_PASSWORD. Si aucun mot de passe n'est fourni ni par option ni par
+l'environnement, il est demandé de façon interactive — jamais de valeur par
+défaut en dur.
+"""
+
+import argparse
+import asyncio
+import getpass
+import os
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from dotenv import load_dotenv  # noqa: E402
+from sqlalchemy import select  # noqa: E402
+
+load_dotenv()
+
+from app.core.auth import hash_password  # noqa: E402
+from app.core.database import AsyncSessionLocal  # noqa: E402
+from app.models.user import User  # noqa: E402
+from app.schemas.user import UserCreate  # noqa: E402
+
+
+def parse_args() -> argparse.Namespace:
+    p = argparse.ArgumentParser(description="Crée ou met à jour un administrateur JuriX")
+    p.add_argument("--email", default=os.getenv("ADMIN_EMAIL", "admin@jurix.cm"))
+    p.add_argument("--username", default=os.getenv("ADMIN_USERNAME", "admin"))
+    p.add_argument("--password", default=os.getenv("ADMIN_PASSWORD"))
+    p.add_argument(
+        "--role",
+        default=os.getenv("ADMIN_ROLE", "superadmin"),
+        choices=["admin", "superadmin"],
+    )
+    p.add_argument(
+        "--reset-password",
+        action="store_true",
+        help="Réinitialise le mot de passe d'un compte existant",
+    )
+    return p.parse_args()
+
+
+async def main() -> int:
+    args = parse_args()
+
+    password = args.password
+    if not password:
+        password = getpass.getpass(f"Mot de passe pour {args.email} : ")
+    if not password:
+        print("❌ Aucun mot de passe fourni.", file=sys.stderr)
+        return 1
+
+    # On valide via UserCreate pour que le script et l'API appliquent
+    # exactement la même politique de mot de passe.
+    try:
+        UserCreate(
+            email=args.email,
+            username=args.username,
+            password=password,
+            role=args.role,
+        )
+    except Exception as e:
+        print(f"❌ Données invalides : {e}", file=sys.stderr)
+        return 1
+
+    try:
+        async with AsyncSessionLocal() as db:
+            existing = (
+                await db.execute(select(User).where(User.email == args.email.lower()))
+            ).scalar_one_or_none()
+
+            if existing is None:
+                db.add(
+                    User(
+                        email=args.email.lower(),
+                        username=args.username.lower(),
+                        hashed_password=hash_password(password),
+                        role=args.role,
+                        is_active=True,
+                        is_verified=True,
+                    )
+                )
+                await db.commit()
+                print(f"✅ Compte créé : {args.email} (rôle {args.role})")
+                return 0
+
+            changed = []
+            if existing.role != args.role:
+                existing.role = args.role
+                changed.append("rôle")
+            if not existing.is_active:
+                existing.is_active = True
+                changed.append("réactivation")
+            if args.reset_password:
+                existing.hashed_password = hash_password(password)
+                changed.append("mot de passe")
+
+            if changed:
+                await db.commit()
+                print(f"✅ Compte mis à jour : {args.email} ({', '.join(changed)})")
+            else:
+                print(
+                    f"ℹ️  Compte déjà conforme : {args.email} (rôle {existing.role}). "
+                    "Utilisez --reset-password pour changer le mot de passe."
+                )
+            return 0
+
+    except Exception as e:
+        print(f"❌ Base de données inaccessible : {e}", file=sys.stderr)
+        print("   Vérifiez DATABASE_URL dans .env et que les migrations sont appliquées.",
+              file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

tests/conftest.py

@@ -89,6 +89,7 @@ _DB_FIXTURES = {
     "test_superadmin_user",
     "auth_headers",
     "admin_headers",
+    "as_admin",
 }
 
 _pg_available: bool | None = None
@@ -332,3 +333,89 @@ def override_get_db(async_db_session: AsyncSession):
         yield async_db_session
 
     return _override_get_db
+
+
+# ==================== AUTHENTIFICATION ====================
+#
+# Surcharger `get_current_user` suffit à débloquer les trois niveaux d'accès :
+# `get_current_active_user`, `get_current_admin_user` et
+# `get_current_superadmin_user` en dépendent tous. Les surcharger un par un
+# obligerait à traiter chaque endpoint séparément.
+#
+# `client` reste volontairement anonyme, pour que les tests qui vérifient les
+# 401 continuent de fonctionner.
+
+
+async def _make_user(db: AsyncSession, email: str, username: str, role: str):
+    from app.core.auth import hash_password
+    from app.models.user import User
+
+    user = User(
+        email=email,
+        username=username,
+        hashed_password=hash_password("MotDePasseTest123"),
+        role=role,
+        is_active=True,
+        is_verified=True,
+    )
+    db.add(user)
+    await db.commit()
+    await db.refresh(user)
+    return user
+
+
+@pytest_asyncio.fixture
+async def test_user(db_session: AsyncSession):
+    """Compte sans privilège."""
+    return await _make_user(db_session, "user@test.cm", "usertest", "user")
+
+
+@pytest_asyncio.fixture
+async def test_admin_user(db_session: AsyncSession):
+    """Compte administrateur."""
+    return await _make_user(db_session, "admin@test.cm", "admintest", "admin")
+
+
+@pytest_asyncio.fixture
+async def test_superadmin_user(db_session: AsyncSession):
+    """Compte superadministrateur."""
+    return await _make_user(db_session, "super@test.cm", "supertest", "superadmin")
+
+
+@pytest.fixture
+def auth_headers(test_user):
+    """En-tête Authorization pour un compte sans privilège (jeton réel)."""
+    from app.core.auth import create_access_token
+
+    return {"Authorization": f"Bearer {create_access_token({'sub': test_user.email})}"}
+
+
+@pytest.fixture
+def admin_headers(test_admin_user):
+    """En-tête Authorization pour un administrateur (jeton réel)."""
+    from app.core.auth import create_access_token
+
+    return {"Authorization": f"Bearer {create_access_token({'sub': test_admin_user.email})}"}
+
+
+@pytest.fixture
+def as_admin(test_admin_user):
+    """
+    Court-circuite l'authentification en se faisant passer pour un administrateur.
+
+    Utile aux tests qui portent sur la logique métier et non sur l'authentification :
+    ils n'ont pas à fabriquer un jeton ni à gérer son expiration.
+    """
+    from app.core.auth import get_current_user
+
+    app.dependency_overrides[get_current_user] = lambda: test_admin_user
+    try:
+        yield test_admin_user
+    finally:
+        app.dependency_overrides.pop(get_current_user, None)
+
+
+@pytest_asyncio.fixture
+async def admin_client(client: AsyncClient, as_admin) -> AsyncClient:
+    """Client HTTP authentifié comme administrateur."""
+    return client

tests/test_api/test_category_routes.py

@@ -11,9 +11,9 @@ class TestCreateCategoryAPI:
     """Tests for POST /api/v1/categories endpoint."""
 
     @pytest.mark.asyncio
-    async def test_create_category_success(self, client: AsyncClient):
+    async def test_create_category_success(self, admin_client: AsyncClient):
         """Test successful category creation via API."""
-        response = await client.post(
+        response = await admin_client.post(
             "/api/v1/categories",
             json={"name": "API Test Category", "description": "Created via API"}
         )
@@ -26,16 +26,16 @@ class TestCreateCategoryAPI:
         assert "created_at" in data
 
     @pytest.mark.asyncio
-    async def test_create_category_duplicate_name(self, client: AsyncClient):
+    async def test_create_category_duplicate_name(self, admin_client: AsyncClient):
         """Test creating duplicate category returns 409."""
         # First create
-        await client.post(
+        await admin_client.post(
             "/api/v1/categories",
             json={"name": "Duplicate Test", "description": "First"}
         )
 
         # Try to create again
-        response = await client.post(
+        response = await admin_client.post(
             "/api/v1/categories",
             json={"name": "Duplicate Test", "description": "Second"}
         )
@@ -44,9 +44,9 @@ class TestCreateCategoryAPI:
         assert "already exists" in response.json()["detail"].lower()
 
     @pytest.mark.asyncio
-    async def test_create_category_invalid_data(self, client: AsyncClient):
+    async def test_create_category_invalid_data(self, admin_client: AsyncClient):
         """Test creating category with invalid data returns 422."""
-        response = await client.post(
+        response = await admin_client.post(
             "/api/v1/categories",
             json={}  # Missing required 'name' field
         )
@@ -99,17 +99,17 @@ class TestGetCategoryAPI:
     """Tests for GET /api/v1/categories/{id} endpoint."""
 
     @pytest.mark.asyncio
-    async def test_get_category_success(self, client: AsyncClient):
+    async def test_get_category_success(self, admin_client: AsyncClient):
         """Test getting single category by ID."""
         # First create a category
-        create_response = await client.post(
+        create_response = await admin_client.post(
             "/api/v1/categories",
             json={"name": "Get Test Category"}
         )
         created = create_response.json()
 
         # Now get it
-        response = await client.get(f"/api/v1/categories/{created['id']}")
+        response = await admin_client.get(f"/api/v1/categories/{created['id']}")
 
         assert response.status_code == 200
         data = response.json()
@@ -129,17 +129,17 @@ class TestUpdateCategoryAPI:
     """Tests for PUT /api/v1/categories/{id} endpoint."""
 
     @pytest.mark.asyncio
-    async def test_update_category_success(self, client: AsyncClient):
+    async def test_update_category_success(self, admin_client: AsyncClient):
         """Test successful category update."""
         # Create a category
-        create_response = await client.post(
+        create_response = await admin_client.post(
             "/api/v1/categories",
             json={"name": "Update Test", "description": "Original"}
         )
         created = create_response.json()
 
         # Update it
-        response = await client.put(
+        response = await admin_client.put(
             f"/api/v1/categories/{created['id']}",
             json={"name": "Updated Name", "description": "Updated Desc"}
         )
@@ -151,9 +151,9 @@ class TestUpdateCategoryAPI:
         assert data["description"] == "Updated Desc"
 
     @pytest.mark.asyncio
-    async def test_update_category_not_found(self, client: AsyncClient):
+    async def test_update_category_not_found(self, admin_client: AsyncClient):
         """Test updating non-existent category returns 404."""
-        response = await client.put(
+        response = await admin_client.put(
             "/api/v1/categories/999999",
             json={"name": "New Name"}
         )
@@ -161,21 +161,21 @@ class TestUpdateCategoryAPI:
         assert response.status_code == 404
 
     @pytest.mark.asyncio
-    async def test_update_category_duplicate_name(self, client: AsyncClient):
+    async def test_update_category_duplicate_name(self, admin_client: AsyncClient):
         """Test updating to duplicate name returns 409."""
         # Create two categories
-        cat1 = (await client.post(
+        cat1 = (await admin_client.post(
             "/api/v1/categories",
             json={"name": "Category 1"}
         )).json()
 
-        cat2 = (await client.post(
+        cat2 = (await admin_client.post(
             "/api/v1/categories",
             json={"name": "Category 2"}
         )).json()
 
         # Try to update cat2 to have same name as cat1
-        response = await client.put(
+        response = await admin_client.put(
             f"/api/v1/categories/{cat2['id']}",
             json={"name": "Category 1"}
         )
@@ -187,43 +187,43 @@ class TestDeleteCategoryAPI:
     """Tests for DELETE /api/v1/categories/{id} endpoint."""
 
     @pytest.mark.asyncio
-    async def test_delete_category_success(self, client: AsyncClient):
+    async def test_delete_category_success(self, admin_client: AsyncClient):
         """Test successful category deletion."""
         # Create a category
-        create_response = await client.post(
+        create_response = await admin_client.post(
             "/api/v1/categories",
             json={"name": "Delete Test"}
         )
         created = create_response.json()
 
         # Delete it
-        response = await client.delete(f"/api/v1/categories/{created['id']}")
+        response = await admin_client.delete(f"/api/v1/categories/{created['id']}")
 
         assert response.status_code == 204
 
         # Verify it's gone
-        get_response = await client.get(f"/api/v1/categories/{created['id']}")
+        get_response = await admin_client.get(f"/api/v1/categories/{created['id']}")
         assert get_response.status_code == 404
 
     @pytest.mark.asyncio
-    async def test_delete_category_not_found(self, client: AsyncClient):
+    async def test_delete_category_not_found(self, admin_client: AsyncClient):
         """Test deleting non-existent category returns 404."""
-        response = await client.delete("/api/v1/categories/999999")
+        response = await admin_client.delete("/api/v1/categories/999999")
 
         assert response.status_code == 404
 
     @pytest.mark.asyncio
-    async def test_delete_category_with_force(self, client: AsyncClient):
+    async def test_delete_category_with_force(self, admin_client: AsyncClient):
         """Test force delete parameter."""
         # Create a category
-        create_response = await client.post(
+        create_response = await admin_client.post(
             "/api/v1/categories",
             json={"name": "Force Delete Test"}
         )
         created = create_response.json()
 
         # Force delete it
-        response = await client.delete(
+        response = await admin_client.delete(
             f"/api/v1/categories/{created['id']}?force=true"
         )
 
@@ -383,9 +383,9 @@ class TestErrorHandling:
         assert response.status_code in [200, 422]
 
     @pytest.mark.asyncio
-    async def test_create_category_with_extra_fields(self, client: AsyncClient):
+    async def test_create_category_with_extra_fields(self, admin_client: AsyncClient):
         """Test creating category with extra fields ignores them."""
-        response = await client.post(
+        response = await admin_client.post(
             "/api/v1/categories",
             json={
                 "name": "Extra Fields Test",
@@ -404,10 +404,10 @@ class TestEndToEndWorkflow:
     """End-to-end workflow tests."""
 
     @pytest.mark.asyncio
-    async def test_full_crud_workflow(self, client: AsyncClient):
+    async def test_full_crud_workflow(self, admin_client: AsyncClient):
         """Test complete CRUD workflow: create → read → update → delete."""
         # 1. Create
-        create_response = await client.post(
+        create_response = await admin_client.post(
             "/api/v1/categories",
             json={"name": "E2E Test Category", "description": "End to end test"}
         )
@@ -416,18 +416,18 @@ class TestEndToEndWorkflow:
         category_id = created["id"]
 
         # 2. Read (single)
-        get_response = await client.get(f"/api/v1/categories/{category_id}")
+        get_response = await admin_client.get(f"/api/v1/categories/{category_id}")
         assert get_response.status_code == 200
         assert get_response.json()["name"] == "E2E Test Category"
 
         # 3. Read (list)
-        list_response = await client.get("/api/v1/categories")
+        list_response = await admin_client.get("/api/v1/categories")
         assert list_response.status_code == 200
         category_ids = [cat["id"] for cat in list_response.json()]
         assert category_id in category_ids
 
         # 4. Update
-        update_response = await client.put(
+        update_response = await admin_client.put(
             f"/api/v1/categories/{category_id}",
             json={"name": "E2E Updated", "description": "Updated description"}
         )
@@ -435,15 +435,15 @@ class TestEndToEndWorkflow:
         assert update_response.json()["name"] == "E2E Updated"
 
         # 5. Verify update
-        get_updated = await client.get(f"/api/v1/categories/{category_id}")
+        get_updated = await admin_client.get(f"/api/v1/categories/{category_id}")
         assert get_updated.json()["name"] == "E2E Updated"
 
         # 6. Delete
-        delete_response = await client.delete(f"/api/v1/categories/{category_id}")
+        delete_response = await admin_client.delete(f"/api/v1/categories/{category_id}")
         assert delete_response.status_code == 204
 
         # 7. Verify deletion
-        get_deleted = await client.get(f"/api/v1/categories/{category_id}")
+        get_deleted = await admin_client.get(f"/api/v1/categories/{category_id}")
         assert get_deleted.status_code == 404
 
     @pytest.mark.asyncio
@@ -464,28 +464,28 @@ class TestEndToEndWorkflow:
             assert single_stats.json()["category_id"] == first_cat["category_id"]
 
     @pytest.mark.asyncio
-    async def test_mapping_workflow(self, client: AsyncClient):
+    async def test_mapping_workflow(self, admin_client: AsyncClient):
         """Test mapping utilities workflow."""
         # 1. Create a new category
-        create_response = await client.post(
+        create_response = await admin_client.post(
             "/api/v1/categories",
             json={"name": "Mapping Test Category"}
         )
         created = create_response.json()
 
         # 2. Get ID-to-name mapping
-        id_to_name = await client.get("/api/v1/categories/mapping/id-to-name")
+        id_to_name = await admin_client.get("/api/v1/categories/mapping/id-to-name")
         assert id_to_name.status_code == 200
         assert "Mapping Test Category" in id_to_name.json().values()
 
         # 3. Get name-to-ID mapping
-        name_to_id = await client.get("/api/v1/categories/mapping/name-to-id")
+        name_to_id = await admin_client.get("/api/v1/categories/mapping/name-to-id")
         assert name_to_id.status_code == 200
         assert "Mapping Test Category" in name_to_id.json().keys()
         assert name_to_id.json()["Mapping Test Category"] == created["id"]
 
         # Cleanup
-        await client.delete(f"/api/v1/categories/{created['id']}")
+        await admin_client.delete(f"/api/v1/categories/{created['id']}")
 
 
 # Summary:

tests/test_api/test_laws_api.py

@@ -133,7 +133,7 @@ class TestCreateLaw:
     """Tests pour POST /api/v1/admin/laws."""
 
     @pytest.mark.asyncio
-    async def test_create_law_success(self, client: AsyncClient):
+    async def test_create_law_success(self, admin_client: AsyncClient):
         """Test création loi."""
         new_law = {
             "reference": "LOI-2024-002",
@@ -144,7 +144,7 @@ class TestCreateLaw:
             "status": "active"
         }
         
-        response = await client.post("/api/v1/admin/laws", json=new_law)
+        response = await admin_client.post("/api/v1/admin/laws", json=new_law)
         
         assert response.status_code == 201
         law = response.json()
@@ -153,7 +153,7 @@ class TestCreateLaw:
         assert "id" in law
 
     @pytest.mark.asyncio
-    async def test_create_law_duplicate_reference(self, client: AsyncClient, sample_law):
+    async def test_create_law_duplicate_reference(self, admin_client: AsyncClient, sample_law):
         """Test création avec référence existante."""
         duplicate_law = {
             "reference": sample_law.reference,
@@ -163,20 +163,20 @@ class TestCreateLaw:
             "language": "fr"
         }
         
-        response = await client.post("/api/v1/admin/laws", json=duplicate_law)
+        response = await admin_client.post("/api/v1/admin/laws", json=duplicate_law)
         
         assert response.status_code == 409
         assert "already exists" in response.json()["detail"].lower()
 
     @pytest.mark.asyncio
-    async def test_create_law_validation_error(self, client: AsyncClient):
+    async def test_create_law_validation_error(self, admin_client: AsyncClient):
         """Test création avec données invalides."""
         invalid_law = {
             "reference": "",  # Invalid: empty
             "title": "Test"
         }
         
-        response = await client.post("/api/v1/admin/laws", json=invalid_law)
+        response = await admin_client.post("/api/v1/admin/laws", json=invalid_law)
         
         assert response.status_code == 422
 
@@ -188,14 +188,14 @@ class TestUpdateLaw:
     """Tests pour PUT /api/v1/admin/laws/{id}."""
 
     @pytest.mark.asyncio
-    async def test_update_law_success(self, client: AsyncClient, sample_law):
+    async def test_update_law_success(self, admin_client: AsyncClient, sample_law):
         """Test mise à jour loi."""
         update_data = {
             "title": "Titre mis à jour",
             "status": "archived"
         }
         
-        response = await client.put(
+        response = await admin_client.put(
             f"/api/v1/admin/laws/{sample_law.id}",
             json=update_data
         )
@@ -206,9 +206,9 @@ class TestUpdateLaw:
         assert law["status"] == update_data["status"]
 
     @pytest.mark.asyncio
-    async def test_update_law_not_found(self, client: AsyncClient):
+    async def test_update_law_not_found(self, admin_client: AsyncClient):
         """Test mise à jour loi non trouvée."""
-        response = await client.put(
+        response = await admin_client.put(
             "/api/v1/admin/laws/99999",
             json={"title": "Test"}
         )
@@ -223,20 +223,20 @@ class TestDeleteLaw:
     """Tests pour DELETE /api/v1/admin/laws/{id}."""
 
     @pytest.mark.asyncio
-    async def test_delete_law_success(self, client: AsyncClient, sample_law):
+    async def test_delete_law_success(self, admin_client: AsyncClient, sample_law):
         """Test suppression loi."""
-        response = await client.delete(f"/api/v1/admin/laws/{sample_law.id}")
+        response = await admin_client.delete(f"/api/v1/admin/laws/{sample_law.id}")
         
         assert response.status_code == 204
         
         # Vérifier que la loi est supprimée
-        get_response = await client.get(f"/api/v1/laws/{sample_law.id}")
+        get_response = await admin_client.get(f"/api/v1/laws/{sample_law.id}")
         assert get_response.status_code == 404
 
     @pytest.mark.asyncio
-    async def test_delete_law_not_found(self, client: AsyncClient):
+    async def test_delete_law_not_found(self, admin_client: AsyncClient):
         """Test suppression loi non trouvée."""
-        response = await client.delete("/api/v1/admin/laws/99999")
+        response = await admin_client.delete("/api/v1/admin/laws/99999")
         
         assert response.status_code == 404
 
@@ -248,7 +248,7 @@ class TestLawsIntegration:
     """Tests d'intégration."""
 
     @pytest.mark.asyncio
-    async def test_full_crud_workflow(self, client: AsyncClient):
+    async def test_full_crud_workflow(self, admin_client: AsyncClient):
         """Test workflow CRUD complet."""
         # 1. Create
         new_law = {
@@ -259,17 +259,17 @@ class TestLawsIntegration:
             "language": "fr",
             "status": "active"
         }
-        create_response = await client.post("/api/v1/admin/laws", json=new_law)
+        create_response = await admin_client.post("/api/v1/admin/laws", json=new_law)
         assert create_response.status_code == 201
         law_id = create_response.json()["id"]
         
         # 2. Read
-        read_response = await client.get(f"/api/v1/laws/{law_id}")
+        read_response = await admin_client.get(f"/api/v1/laws/{law_id}")
         assert read_response.status_code == 200
         assert read_response.json()["title"] == new_law["title"]
         
         # 3. Update
-        update_response = await client.put(
+        update_response = await admin_client.put(
             f"/api/v1/admin/laws/{law_id}",
             json={"title": "Titre modifié"}
         )
@@ -277,9 +277,9 @@ class TestLawsIntegration:
         assert update_response.json()["title"] == "Titre modifié"
         
         # 4. Delete
-        delete_response = await client.delete(f"/api/v1/admin/laws/{law_id}")
+        delete_response = await admin_client.delete(f"/api/v1/admin/laws/{law_id}")
         assert delete_response.status_code == 204
         
         # 5. Verify deletion
-        verify_response = await client.get(f"/api/v1/laws/{law_id}")
+        verify_response = await admin_client.get(f"/api/v1/laws/{law_id}")
         assert verify_response.status_code == 404