- app/email/ : sender (smtplib, STARTTLS ou SSL, connexion réutilisée pour un
lot) + templates (rendu, textes dans locales/fr.json). EMAIL_ENABLED=false
(défaut) ⇒ journalisé seulement, sans destinataire ni corps (RGPD).
- Lien magique : POST .../auth/magic-link envoie désormais l'email
(PUBLIC_BASE_URL + magic_link_path + ?token=). Helper app/auth/magic_link.py
partagé. Sans SMTP, le jeton non nominatif reste journalisé pour le dev.
- Récapitulatif de planning : POST /editions/{id}/notifications/planning
(COORDINATION). subject/body sont des modèles ({prenom}, {nom}, {nb_creneaux},
{nb_heures}, {creneaux}, {lien_connexion} → lien magique personnel émis).
Cible tous les bénévoles avec email ou une liste. Envoi en BackgroundTasks.
- Config : PUBLIC_BASE_URL, EMAIL_ENABLED, SMTP_* ; .env.example,
docker-compose (dev + prod), deploy/README.
- 9 tests (sender activé/désactivé, rendu, notifications : contenu, gabarit par
défaut, lien magique embarqué, sous-ensemble, réservé COORDINATION).
137 tests au total ; ruff + alembic check verts.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017uUsRg7XzopkCPWimwg9ZM
181 lines
5.8 KiB
Python
181 lines
5.8 KiB
Python
"""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 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.magic_link import build_magic_link_url, issue_magic_link_token
|
|
from app.auth.passwords import MIN_LENGTH, hash_password, verify_password
|
|
from app.auth.tokens import create_access_token, hash_token
|
|
from app.config import get_settings
|
|
from app.db import get_session
|
|
from app.email import render_magic_link, send
|
|
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:
|
|
settings = get_settings()
|
|
raw = issue_magic_link_token(session, account)
|
|
session.commit()
|
|
subject, body = render_magic_link(
|
|
link=build_magic_link_url(raw),
|
|
minutes=settings.magic_link_ttl_minutes,
|
|
)
|
|
send(account.email, subject, body)
|
|
if not settings.email_enabled:
|
|
# Aide au dev sans SMTP : le jeton (non nominatif) est journalisé.
|
|
logger.info(
|
|
"Lien magique (email désactivé) — compte=%s jeton=%s",
|
|
account.id,
|
|
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
|