Roman
28f1bbde31
- Colonne users.content_preference (anime|serie|both, défaut both) avec migration automatique des bases existantes. - GET /auth/me expose la préférence ; PUT /auth/preferences la met à jour (422 si valeur invalide). - /api/search : filtre les résultats par media_type ET ignore les sources hors périmètre (une source anime-only n'est pas interrogée en mode séries). Les films restent associés aux animés. - /api/discover : nouveautés filtrées selon la même règle. - Sélecteur compact dans la topbar (⛩/📺/✨), persisté serveur. - 7 tests dédiés (roundtrip, rejet invalide, filtrage search/discover, skip de sources) — 114 au total, ruff clean. Prête pour la première source de séries : il ne reste qu'un module SourceScraper avec media_types=("serie",) à écrire.
129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""Routes d'authentification (cookies httponly, adaptées à l'UI htmx)."""
|
|
|
|
import logging
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Body, Depends, Form, HTTPException, Request, Response, status
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
from app import auth
|
|
from app.auth import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
ACCESS_COOKIE = "ohm_access"
|
|
REFRESH_COOKIE = "ohm_refresh"
|
|
|
|
|
|
def set_auth_cookies(response: Response, user: User, refresh_token: str) -> None:
|
|
response.set_cookie(
|
|
ACCESS_COOKIE, auth.create_access_token(user), httponly=True, samesite="lax"
|
|
)
|
|
response.set_cookie(REFRESH_COOKIE, refresh_token, httponly=True, samesite="lax", path="/")
|
|
|
|
|
|
def clear_auth_cookies(response: Response) -> None:
|
|
response.delete_cookie(ACCESS_COOKIE)
|
|
response.delete_cookie(REFRESH_COOKIE, path="/")
|
|
|
|
|
|
# ---------------------------------------------------------------- dépendances
|
|
|
|
|
|
async def current_user(request: Request, response: Response) -> User:
|
|
"""Utilisateur courant via cookie d'accès ; tente un refresh si expiré."""
|
|
token = request.cookies.get(ACCESS_COOKIE)
|
|
if token:
|
|
payload = auth.decode_access_token(token)
|
|
if payload:
|
|
user = await auth.get_user(int(payload["sub"]))
|
|
if user:
|
|
return user
|
|
|
|
refresh = request.cookies.get(REFRESH_COOKIE)
|
|
if refresh:
|
|
user = await auth.use_refresh_token(refresh)
|
|
if user:
|
|
new_refresh = await auth.create_refresh_token(user.id)
|
|
set_auth_cookies(response, user, new_refresh)
|
|
return user
|
|
|
|
raise HTTPException(status.HTTP_303_SEE_OTHER, headers={"Location": "/login"})
|
|
|
|
|
|
async def require_admin(user: Annotated[User, Depends(current_user)]) -> User:
|
|
if not user.is_admin:
|
|
raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Droits administrateur requis")
|
|
return user
|
|
|
|
|
|
CurrentUser = Annotated[User, Depends(current_user)]
|
|
AdminUser = Annotated[User, Depends(require_admin)]
|
|
|
|
|
|
# ---------------------------------------------------------------- routes
|
|
|
|
|
|
@router.post("/register")
|
|
async def register(
|
|
response: Response,
|
|
username: Annotated[str, Form(min_length=3, max_length=32)],
|
|
password: Annotated[str, Form(min_length=6)],
|
|
) -> RedirectResponse:
|
|
row = await auth.db.fetchone("SELECT id FROM users WHERE username = ?", (username.strip(),))
|
|
if row is not None:
|
|
raise HTTPException(status.HTTP_409_CONFLICT, detail="Nom d'utilisateur déjà pris")
|
|
user = await auth.create_user(username, password)
|
|
refresh = await auth.create_refresh_token(user.id)
|
|
redirect = RedirectResponse("/", status.HTTP_303_SEE_OTHER)
|
|
set_auth_cookies(redirect, user, refresh)
|
|
return redirect
|
|
|
|
|
|
@router.post("/login")
|
|
async def login(
|
|
response: Response,
|
|
username: Annotated[str, Form()],
|
|
password: Annotated[str, Form()],
|
|
) -> RedirectResponse:
|
|
user = await auth.authenticate(username, password)
|
|
if user is None:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Identifiants invalides")
|
|
refresh = await auth.create_refresh_token(user.id)
|
|
redirect = RedirectResponse("/", status.HTTP_303_SEE_OTHER)
|
|
set_auth_cookies(redirect, user, refresh)
|
|
return redirect
|
|
|
|
|
|
@router.post("/logout")
|
|
async def logout(user: CurrentUser) -> RedirectResponse:
|
|
await auth.revoke_all_refresh_tokens(user.id)
|
|
redirect = RedirectResponse("/login", status.HTTP_303_SEE_OTHER)
|
|
clear_auth_cookies(redirect)
|
|
return redirect
|
|
|
|
|
|
@router.get("/me")
|
|
async def me(user: CurrentUser) -> dict:
|
|
return {
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"is_admin": user.is_admin,
|
|
"content_preference": user.content_preference,
|
|
}
|
|
|
|
|
|
@router.put("/preferences")
|
|
async def update_preferences(
|
|
user: CurrentUser,
|
|
content_preference: Annotated[str, Body(embed=True)],
|
|
) -> dict:
|
|
"""Préférence de contenus : animés, séries, ou les deux."""
|
|
try:
|
|
await auth.set_content_preference(user.id, content_preference)
|
|
except ValueError as exc:
|
|
raise HTTPException(422, detail=str(exc)) from exc
|
|
return {"content_preference": content_preference}
|