app/auth.py

"""Identité : numéro de téléphone français, PIN à 6 chiffres, jeton de session."""

from __future__ import annotations

import hashlib
import hmac
import os
import re
import secrets
import time
from dataclasses import dataclass

import jwt
from fastapi import Header, HTTPException, status

from . import config, db

# --- Numéro de téléphone --------------------------------------------------

_NON_DIGITS = re.compile(r"[^0-9+]")
_FRENCH_MOBILE = re.compile(r"^0[67][0-9]{8}$")


def normalize_phone(raw: str) -> str:
    """Ramène toutes les écritures courantes à la forme canonique 0XXXXXXXXX.

    Accepte « 06 12 34 56 78 », « +33 6 12 34 56 78 », « 0033612345678 »,
    « 612345678 ». Lève ValueError si ce n'est pas un mobile français.
    """
    if not raw:
        raise ValueError("Numéro de téléphone manquant.")
    s = _NON_DIGITS.sub("", str(raw))
    if s.startswith("+33"):
        s = "0" + s[3:]
    elif s.startswith("0033"):
        s = "0" + s[4:]
    elif s.startswith("33") and len(s) == 11:
        s = "0" + s[2:]
    s = s.lstrip("+")
    # « 612345678 » saisi sans le zéro initial.
    if len(s) == 9 and s[0] in "67":
        s = "0" + s
    if not _FRENCH_MOBILE.match(s):
        raise ValueError("Numéro invalide : il doit commencer par 06 ou 07 et compter 10 chiffres.")
    return s


def format_phone(phone: str) -> str:
    """0612345678 -> « 06 12 34 56 78 » (affichage)."""
    if len(phone) != 10:
        return phone
    return " ".join(phone[i : i + 2] for i in range(0, 10, 2))


# --- PIN ------------------------------------------------------------------

_PIN_RE = re.compile(r"^[0-9]{6}$")


# Codes les plus fréquents dans les fuites publiques : à refuser d'office.
_COMMON_PINS = {
    "123456", "654321", "012345", "543210", "111111", "000000", "121212",
    "123123", "112233", "696969", "666666", "159753", "147258", "789456",
    "142536", "102030", "123321", "010203", "987654", "246810", "135790",
    "159357", "852456", "abcdef",
}


def validate_pin(pin: str) -> str:
    """Refuse les codes triviaux : la seule barrière avant le compte est ce PIN."""
    pin = (pin or "").strip()
    if not _PIN_RE.match(pin):
        raise ValueError("Le code PIN doit contenir exactement 6 chiffres.")

    generic = "Code PIN trop simple : choisissez une combinaison moins évidente."

    if pin in _COMMON_PINS:
        raise ValueError(generic)
    if len(set(pin)) == 1:
        raise ValueError("Code PIN trop simple : évitez six fois le même chiffre.")
    if len(set(pin)) == 2:
        raise ValueError("Code PIN trop simple : utilisez au moins trois chiffres différents.")

    digits = [int(c) for c in pin]
    steps = {b - a for a, b in zip(digits, digits[1:])}
    # Suites arithmétiques : 123456, 246802, 975310…
    if len(steps) == 1:
        raise ValueError(generic)
    # Motifs répétés : ABABAB ou ABCABC.
    if pin[:2] * 3 == pin or pin[:3] * 2 == pin:
        raise ValueError(generic)

    return pin


def hash_pin(pin: str, salt: bytes | None = None) -> tuple[str, str]:
    """scrypt (bibliothèque standard) — pas de dépendance native supplémentaire."""
    salt = salt or os.urandom(16)
    digest = hashlib.scrypt(pin.encode("utf-8"), salt=salt, n=2**14, r=8, p=1, dklen=32)
    return digest.hex(), salt.hex()


def verify_pin(pin: str, pin_hash: str, pin_salt: str) -> bool:
    if not pin_hash or not pin_salt:
        return False
    try:
        candidate, _ = hash_pin(pin, bytes.fromhex(pin_salt))
    except (ValueError, TypeError):
        return False
    return hmac.compare_digest(candidate, pin_hash)


# --- Secret de signature --------------------------------------------------


def _secret() -> str:
    """Secret HMAC persistant : généré une fois, conservé en base."""
    env = os.environ.get("SECRET_KEY", "").strip()
    if env:
        return env
    existing = db.get_setting("secret_key")
    if existing:
        return existing
    generated = secrets.token_urlsafe(48)
    db.set_setting("secret_key", generated)
    return generated


# --- Jetons ---------------------------------------------------------------


def issue_token(phone: str, token_epoch: int) -> str:
    now = int(time.time())
    payload = {
        "sub": phone,
        "ep": token_epoch,
        "iat": now,
        "exp": now + config.TOKEN_TTL_DAYS * 86400,
    }
    return jwt.encode(payload, _secret(), algorithm="HS256")


def decode_token(token: str) -> dict:
    return jwt.decode(token, _secret(), algorithms=["HS256"])


@dataclass
class CurrentUser:
    phone: str
    first_name: str
    last_name: str
    nickname: str
    is_superadmin: bool
    status: str

    @property
    def display_name(self) -> str:
        return self.nickname or f"{self.first_name} {self.last_name}".strip() or format_phone(self.phone)


def user_from_token(token: str) -> CurrentUser:
    """Valide un jeton et renvoie l'utilisateur, ou lève une HTTPException 401."""
    creds_error = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Session expirée ou invalide. Reconnectez-vous.",
    )
    try:
        payload = decode_token(token)
    except jwt.PyJWTError:
        raise creds_error
    phone = payload.get("sub")
    row = db.query_one(
        "SELECT phone, first_name, last_name, nickname, is_superadmin, status, token_epoch "
        "FROM users WHERE phone = ?",
        (phone,),
    )
    if row is None or row["status"] != "active":
        raise creds_error
    # Un changement de PIN ou un blocage incrémente token_epoch et invalide
    # d'un coup toutes les sessions ouvertes.
    if int(payload.get("ep", -1)) != int(row["token_epoch"]):
        raise creds_error
    return CurrentUser(
        phone=row["phone"],
        first_name=row["first_name"],
        last_name=row["last_name"],
        nickname=row["nickname"],
        is_superadmin=bool(row["is_superadmin"]),
        status=row["status"],
    )


def _bearer(authorization: str | None) -> str:
    if not authorization or not authorization.lower().startswith("bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentification requise.",
        )
    return authorization.split(" ", 1)[1].strip()


async def current_user(authorization: str | None = Header(default=None)) -> CurrentUser:
    """Dépendance FastAPI : utilisateur authentifié."""
    return user_from_token(_bearer(authorization))


async def current_superadmin(authorization: str | None = Header(default=None)) -> CurrentUser:
    """Dépendance FastAPI : réservé au superadmin."""
    user = user_from_token(_bearer(authorization))
    if not user.is_superadmin:
        raise HTTPException(status_code=403, detail="Action réservée au superadmin.")
    return user


# --- Rôles de groupe ------------------------------------------------------


def group_role(phone: str, group_id: int, is_superadmin: bool = False) -> str | None:
    """Renvoie 'admin', 'member' ou None. Le superadmin est admin partout."""
    row = db.query_one(
        "SELECT role FROM memberships WHERE group_id = ? AND phone = ?",
        (group_id, phone),
    )
    if row is not None:
        return "admin" if (is_superadmin or row["role"] == "admin") else "member"
    return "admin" if is_superadmin else None


def require_member(user: CurrentUser, group_id: int) -> str:
    role = group_role(user.phone, group_id, user.is_superadmin)
    if role is None:
        raise HTTPException(status_code=403, detail="Vous n'êtes pas membre de ce groupe.")
    return role


def require_group_admin(user: CurrentUser, group_id: int) -> str:
    role = group_role(user.phone, group_id, user.is_superadmin)
    if role != "admin":
        raise HTTPException(status_code=403, detail="Action réservée aux administrateurs du groupe.")
    return role