benevoles/backend/app/api/auth.py

"""Authentification.

Deux systèmes, selon le compte :
- **mot de passe** (`POST .../auth/login`) pour les comptes qui en ont un —
  la coordination ;
- **lien magique** (`.../auth/magic-link` puis `.../auth/verify`) pour les
  comptes sans mot de passe — référents et bénévoles.
"""

import datetime as dt
import logging
import uuid

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import func
from sqlmodel import Session, select

from app.api.schemas import (
    AccountRead,
    ChangePasswordRequest,
    LoginRequest,
    MagicLinkRequest,
    TokenResponse,
    VerifyRequest,
)
from app.auth.deps import get_current_account
from app.auth.passwords import MIN_LENGTH, hash_password, verify_password
from app.auth.tokens import create_access_token, hash_token, new_raw_token
from app.config import get_settings
from app.db import get_session
from app.models import Account, MagicLinkToken, local_now

router = APIRouter(tags=["auth"])
logger = logging.getLogger("app.auth")


def _account_by_email(
    session: Session, edition_id: uuid.UUID, email: str
) -> Account | None:
    return session.exec(
        select(Account)
        .where(Account.edition_id == edition_id)
        .where(func.lower(Account.email) == email.strip().lower())
    ).first()


def _issue(account: Account) -> TokenResponse:
    return TokenResponse(
        access_token=create_access_token(account),
        token_type="bearer",
        account=AccountRead.model_validate(account),
    )


def _invalid_credentials() -> HTTPException:
    return HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail={"code": "auth.invalid_credentials"},
    )


def _invalid_token() -> HTTPException:
    return HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail={"code": "auth.invalid_token"},
    )


# --------------------------------------------------------------------------- #
# Mot de passe (admins)                                                        #
# --------------------------------------------------------------------------- #
@router.post("/editions/{edition_id}/auth/login", response_model=TokenResponse)
def login(
    edition_id: uuid.UUID,
    payload: LoginRequest,
    session: Session = Depends(get_session),
) -> TokenResponse:
    account = _account_by_email(session, edition_id, payload.email)
    if (
        account is None
        or account.password_hash is None
        or not verify_password(payload.password, account.password_hash)
    ):
        raise _invalid_credentials()
    return _issue(account)


@router.post(
    "/editions/{edition_id}/auth/change-password",
    status_code=status.HTTP_204_NO_CONTENT,
)
def change_password(
    edition_id: uuid.UUID,
    payload: ChangePasswordRequest,
    account: Account = Depends(get_current_account),
    session: Session = Depends(get_session),
) -> None:
    if account.edition_id != edition_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail={"code": "error.forbidden"},
        )
    if account.password_hash is None or not verify_password(
        payload.current_password, account.password_hash
    ):
        raise _invalid_credentials()
    if len(payload.new_password) < MIN_LENGTH:
        raise HTTPException(
            status_code=422,
            detail={"code": "auth.password_too_short", "min": MIN_LENGTH},
        )
    account.password_hash = hash_password(payload.new_password)
    session.add(account)
    session.commit()


# --------------------------------------------------------------------------- #
# Lien magique (référents, bénévoles)                                          #
# --------------------------------------------------------------------------- #
@router.post(
    "/editions/{edition_id}/auth/magic-link",
    status_code=status.HTTP_202_ACCEPTED,
)
def request_magic_link(
    edition_id: uuid.UUID,
    payload: MagicLinkRequest,
    session: Session = Depends(get_session),
) -> dict[str, str]:
    """Émet un lien magique si un compte **sans mot de passe** correspond.
    Réponse identique dans tous les cas (pas de divulgation)."""
    account = _account_by_email(session, edition_id, payload.email)
    if account is not None and account.password_hash is None:
        raw = new_raw_token()
        token = MagicLinkToken(
            account_id=account.id,
            token_hash=hash_token(raw),
            expires_at=local_now()
            + dt.timedelta(minutes=get_settings().magic_link_ttl_minutes),
        )
        session.add(token)
        session.commit()
        # STUB : en production, ce lien part par email. Ici on le journalise
        # (identifiants uniquement, aucune donnée nominative).
        logger.info(
            "STUB lien magique — compte=%s édition=%s expire=%s jeton=%s",
            account.id,
            edition_id,
            token.expires_at.isoformat(),
            raw,
        )
    return {"status": "accepted"}


@router.post("/editions/{edition_id}/auth/verify", response_model=TokenResponse)
def verify_magic_link(
    edition_id: uuid.UUID,
    payload: VerifyRequest,
    session: Session = Depends(get_session),
) -> TokenResponse:
    token = session.exec(
        select(MagicLinkToken).where(
            MagicLinkToken.token_hash == hash_token(payload.token.strip())
        )
    ).first()
    if (
        token is None
        or token.used_at is not None
        or token.expires_at < local_now()
    ):
        raise _invalid_token()
    account = session.get(Account, token.account_id)
    if account is None or account.edition_id != edition_id:
        raise _invalid_token()
    token.used_at = local_now()
    session.commit()
    session.refresh(account)
    return _issue(account)


@router.get("/auth/me", response_model=AccountRead)
def read_me(account: Account = Depends(get_current_account)) -> Account:
    return account