ohm_streaming/app/auth.py

"""Authentification : JWT court + refresh token longue durée, rôles admin/user."""

import hashlib
import logging
import secrets
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta

import jwt
from pwdlib import PasswordHash

from app.config import get_settings
from app.db import db

logger = logging.getLogger(__name__)

password_hash = PasswordHash.recommended()

ALGORITHM = "HS256"


@dataclass
class User:
    id: int
    username: str
    is_admin: bool
    is_active: bool
    content_preference: str = "both"  # anime | serie | both


VALID_CONTENT_PREFERENCES = ("anime", "serie", "both")


# ---------------------------------------------------------------- mots de passe


def hash_password(password: str) -> str:
    return password_hash.hash(password)


def verify_password(password: str, hashed: str) -> bool:
    return password_hash.verify(password, hashed)


# ---------------------------------------------------------------- utilisateurs


async def create_user(username: str, password: str) -> User:
    """Crée un compte ; le tout premier utilisateur devient admin."""
    row = await db.fetchone("SELECT COUNT(*) AS n FROM users")
    is_first = row["n"] == 0
    cursor = await db.execute(
        "INSERT INTO users (username, password_hash, is_admin) VALUES (?, ?, ?)",
        (username.strip(), hash_password(password), int(is_first)),
    )
    user = User(id=cursor.lastrowid, username=username.strip(), is_admin=is_first, is_active=True)
    logger.info("Utilisateur créé : %s (admin=%s)", user.username, user.is_admin)
    return user


async def authenticate(username: str, password: str) -> User | None:
    row = await db.fetchone(
        "SELECT id, username, password_hash, is_admin, is_active, content_preference "
        "FROM users WHERE username = ?",
        (username.strip(),),
    )
    if row is None or not verify_password(password, row["password_hash"]):
        logger.warning("Échec d'authentification pour %r", username)
        return None
    if not row["is_active"]:
        logger.warning("Compte désactivé : %r", username)
        return None
    return User(
        id=row["id"],
        username=row["username"],
        is_admin=bool(row["is_admin"]),
        is_active=True,
        content_preference=row["content_preference"],
    )


async def get_user(user_id: int) -> User | None:
    row = await db.fetchone(
        "SELECT id, username, is_admin, is_active, content_preference FROM users WHERE id = ?",
        (user_id,),
    )
    if row is None or not row["is_active"]:
        return None
    return User(
        id=row["id"],
        username=row["username"],
        is_admin=bool(row["is_admin"]),
        is_active=True,
        content_preference=row["content_preference"],
    )


async def set_content_preference(user_id: int, preference: str) -> None:
    if preference not in VALID_CONTENT_PREFERENCES:
        raise ValueError(f"Préférence invalide : {preference!r}")
    await db.execute(
        "UPDATE users SET content_preference = ? WHERE id = ?", (preference, user_id)
    )


# ---------------------------------------------------------------- tokens


def create_access_token(user: User) -> str:
    settings = get_settings()
    payload = {
        "sub": str(user.id),
        "username": user.username,
        "admin": user.is_admin,
        "exp": datetime.now(UTC) + timedelta(minutes=settings.access_token_ttl_minutes),
    }
    return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM)


def decode_access_token(token: str) -> dict | None:
    try:
        return jwt.decode(token, get_settings().secret_key, algorithms=[ALGORITHM])
    except jwt.PyJWTError as exc:
        logger.debug("Token d'accès invalide : %s", exc)
        return None


def _hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


async def create_refresh_token(user_id: int) -> str:
    """Refresh token opaque ; seul son hash est stocké en DB (révocable)."""
    token = secrets.token_urlsafe(48)
    expires = datetime.now(UTC) + timedelta(days=get_settings().refresh_token_ttl_days)
    await db.execute(
        "INSERT INTO refresh_tokens (user_id, token_hash, expires_at) VALUES (?, ?, ?)",
        (user_id, _hash_token(token), expires.strftime("%Y-%m-%d %H:%M:%S")),
    )
    return token


async def use_refresh_token(token: str) -> User | None:
    """Valide un refresh token, le révoque (rotation) et retourne l'utilisateur."""
    row = await db.fetchone(
        "SELECT id, user_id FROM refresh_tokens "
        "WHERE token_hash = ? AND revoked = 0 AND expires_at > datetime('now')",
        (_hash_token(token),),
    )
    if row is None:
        logger.warning("Refresh token invalide ou expiré")
        return None
    await db.execute("UPDATE refresh_tokens SET revoked = 1 WHERE id = ?", (row["id"],))
    user = await get_user(row["user_id"])
    if user is None:
        logger.warning("Refresh token d'un compte supprimé/désactivé (user_id=%s)", row["user_id"])
    return user


async def revoke_all_refresh_tokens(user_id: int) -> None:
    await db.execute("UPDATE refresh_tokens SET revoked = 1 WHERE user_id = ?", (user_id,))