From 5c7e5d046f7c0ae774bb0d268a99f4499b14ce2c Mon Sep 17 00:00:00 2001 From: Thioye97 Date: Fri, 7 Aug 2026 16:15:57 +0000 Subject: [PATCH] =?UTF-8?q?Journal=20des=20connexions=20:=20tra=C3=A7abili?= =?UTF-8?q?t=C3=A9=20des=20acc=C3=A8s=20et=20alerte=20sur=20contexte=20inc?= =?UTF-8?q?onnu?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Le TDR exige des traces historisées systématiquement. Les tentatives d'accès n'étaient écrites que dans le journal applicatif (stdout de l'hébergeur) : ni requêtables, ni conservées. Impossible donc de répondre à « qui s'est connecté, depuis où, et quand » — ni de voir une série d'échecs. Table `login_events` (sans clé étrangère vers `users` : la trace doit survivre à la suppression du compte, et un échec sur une adresse inconnue n'a aucun utilisateur à référencer), alimentée aux quatre points de /auth : connexion réussie, échec avec son motif, déconnexion, prolongation de session. Plus `users.last_login_at` / `last_login_ip`. Une connexion est signalée « contexte inconnu » quand son adresse IP ou son appareil n'a jamais été observé pour ce compte ; la première connexion l'est donc toujours. Elle déclenche un courriel par le relais Brevo déjà en place, envoyé en tâche de fond pour ne pas peser sur le temps de connexion. Les identifiants SMTP n'existaient que sur les crons : ils sont ajoutés au service web, faute de quoi l'alerte serait restée inerte en production. Toute écriture passe par `record_safe` : un journal en panne dégrade la traçabilité, il ne doit pas fermer l'accès à la plateforme. Console de sécurité `/security/*` réservée au super-administrateur, en trois lectures : qui a tenté d'accéder, qui est connecté maintenant, et qui ne s'est jamais connecté. Ces endpoints passent par le rôle d'authentification et non par la session RLS : la politique `tenant_isolation_users` filtre par tenant sans échappatoire super-admin. Correction d'un défaut découvert au passage : `is_super_admin` n'était calculé nulle part, la revendication du jeton valait donc toujours faux. Toutes les portes qui en dépendent — bascule de tenant, contexte RLS élargi, écrans d'administration — étaient closes pour tout le monde. Elle est désormais lue depuis le rôle SUPER_ADMIN de `user_roles`, au login comme au renouvellement (sans quoi un refresh dégradait silencieusement le jeton). Aucun compte ne porte ce rôle par défaut : rien ne change pour les comptes existants. `app/scripts/grant_super_admin.py` permet de l'attribuer. 19 tests ajoutés (9 unitaires, 10 d'intégration). Migration s4 vérifiée à l'aller et au retour sur une base jetable. --- .env.example | 12 + alembic/versions/s4_login_events.py | 87 +++++ app/api/routes/auth.py | 97 +++++- app/api/routes/security.py | 125 +++++++ app/main.py | 2 + app/models/login_event.py | 75 +++++ app/models/users.py | 8 +- app/scripts/grant_super_admin.py | 117 +++++++ app/services/auth_service.py | 28 +- app/services/list_notifier.py | 28 +- app/services/login_audit_service.py | 468 ++++++++++++++++++++++++++ render.yaml | 18 + tests/integration/test_login_audit.py | 222 ++++++++++++ tests/unit/test_login_audit_alert.py | 84 +++++ 14 files changed, 1348 insertions(+), 23 deletions(-) create mode 100644 alembic/versions/s4_login_events.py create mode 100644 app/api/routes/security.py create mode 100644 app/models/login_event.py create mode 100644 app/scripts/grant_super_admin.py create mode 100644 app/services/login_audit_service.py create mode 100644 tests/integration/test_login_audit.py create mode 100644 tests/unit/test_login_audit_alert.py diff --git a/.env.example b/.env.example index 9b01161..34dcfd0 100644 --- a/.env.example +++ b/.env.example @@ -84,3 +84,15 @@ CELERY_TASK_SOFT_LIMIT_SECONDS=240 # --- CORS -------------------------------------------------------------------- # Comma-separated list. Defaults are coded in app/core/config.py. # CORS_ORIGINS=https://app.simandou-screening.com,http://localhost:5173 + +# --- Journal de connexions : alerte sur contexte inconnu --------------------- +# Une connexion depuis une IP ou un appareil jamais vus déclenche un courriel. +# Sans identifiants SMTP, le journal enregistre quand même tout : seul l'envoi +# est ignoré. Destinataire : LOGIN_ALERT_TO_EMAIL, à défaut BREVO_TO_EMAIL. +LOGIN_ALERT_ENABLED=true +LOGIN_ALERT_TO_EMAIL= +BREVO_SMTP_LOGIN= +BREVO_SMTP_KEY= +BREVO_FROM_EMAIL= +BREVO_FROM_NAME=Simandou Conformité +BREVO_TO_EMAIL= diff --git a/alembic/versions/s4_login_events.py b/alembic/versions/s4_login_events.py new file mode 100644 index 0000000..4255b1c --- /dev/null +++ b/alembic/versions/s4_login_events.py @@ -0,0 +1,87 @@ +"""S4: journal des connexions (login_events) + dernière connexion sur users + +Revision ID: s4_login_events +Revises: m13_offshore_relations +Create Date: 2026-08-07 +""" +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +# revision identifiers +revision = "s4_login_events" +down_revision = "m13_offshore_relations" +branch_labels = None +depends_on = None + +# Le rôle applicatif qui contourne RLS pour l'authentification (cf. +# settings.AUTH_BYPASS_ROLE). Les événements de connexion sont écrits SOUS ce +# rôle, puisque l'écriture a lieu pendant le login — avant qu'un contexte de +# tenant n'existe. Sans le GRANT ci-dessous, l'insertion échouerait en +# production alors qu'elle passe en test (où le rôle reçoit tous les droits). +AUTH_BYPASS_ROLE = "auth_bypass_rls" + + +def upgrade() -> None: + op.create_table( + "login_events", + sa.Column( + "id", + postgresql.UUID(as_uuid=True), + primary_key=True, + server_default=sa.text("gen_random_uuid()"), + ), + sa.Column("event", sa.String(16), nullable=False), + sa.Column("reason", sa.String(32), nullable=True), + # Pas de ForeignKey volontairement : la trace doit survivre à la + # suppression du compte, et un échec sur une adresse inconnue n'a + # aucun utilisateur à référencer. + sa.Column("user_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("email", sa.Text(), nullable=True), + sa.Column("tenant_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("ip", sa.String(64), nullable=True), + sa.Column("user_agent", sa.Text(), nullable=True), + sa.Column( + "is_new_context", + sa.Boolean(), + nullable=False, + server_default=sa.text("false"), + ), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("now()"), + ), + ) + op.create_index("ix_login_events_created_at", "login_events", ["created_at"]) + op.create_index("ix_login_events_user_created", "login_events", ["user_id", "created_at"]) + op.create_index("ix_login_events_email", "login_events", ["email"]) + op.create_index("ix_login_events_ip", "login_events", ["ip"]) + + op.add_column("users", sa.Column("last_login_at", sa.DateTime(timezone=True), nullable=True)) + op.add_column("users", sa.Column("last_login_ip", sa.String(64), nullable=True)) + + # Droits pour le rôle d'authentification, s'il existe sur cette instance. + op.execute( + f""" + DO $$ BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{AUTH_BYPASS_ROLE}') THEN + GRANT SELECT, INSERT ON public.login_events TO {AUTH_BYPASS_ROLE}; + GRANT SELECT, UPDATE ON public.users TO {AUTH_BYPASS_ROLE}; + END IF; + END $$; + """ + ) + + +def downgrade() -> None: + op.drop_column("users", "last_login_ip") + op.drop_column("users", "last_login_at") + op.drop_index("ix_login_events_ip", table_name="login_events") + op.drop_index("ix_login_events_email", table_name="login_events") + op.drop_index("ix_login_events_user_created", table_name="login_events") + op.drop_index("ix_login_events_created_at", table_name="login_events") + op.drop_table("login_events") diff --git a/app/api/routes/auth.py b/app/api/routes/auth.py index 0c0e5f0..3dc50f9 100644 --- a/app/api/routes/auth.py +++ b/app/api/routes/auth.py @@ -1,7 +1,7 @@ # app/api/routes/auth.py from typing import Annotated -from fastapi import APIRouter, Body, Depends, HTTPException, Request +from fastapi import APIRouter, BackgroundTasks, Body, Depends, HTTPException, Request from sqlalchemy import text from sqlalchemy.orm import Session @@ -16,6 +16,7 @@ RefreshRequest, TokenPairResponse, ) +from app.services import login_audit_service from app.services.auth_service import ( consume_refresh_token, create_access_token, @@ -61,6 +62,7 @@ def _access_token_for(user: dict) -> tuple[str, int]: def login( request: Request, payload: Annotated[LoginRequest, Body()], + background: BackgroundTasks, db: Session = Depends(get_db_public), ): _reset_session(db) @@ -71,11 +73,21 @@ def login( logger.exception("SET ROLE auth_bypass failed") raise HTTPException(status_code=500, detail="Auth DB role misconfigured") + ip, ua = _client_meta(request) + try: u = get_user_by_email(db, payload.email) if not u: logger.info("login_failed", extra={"email": payload.email, "reason": "unknown_user"}) + login_audit_service.record_safe( + db, + event=login_audit_service.EVENT_LOGIN_FAILED, + email=payload.email, + ip=ip, + user_agent=ua, + reason="unknown_user", + ) raise HTTPException(status_code=401, detail="Invalid credentials") if not verify_password(payload.password, u.get("password_hash") or ""): @@ -83,6 +95,16 @@ def login( "login_failed", extra={"email": payload.email, "user_id": u.get("id"), "reason": "bad_password"}, ) + login_audit_service.record_safe( + db, + event=login_audit_service.EVENT_LOGIN_FAILED, + email=payload.email, + user_id=str(u["id"]), + tenant_id=str(u["tenant_id"]) if u.get("tenant_id") else None, + ip=ip, + user_agent=ua, + reason="bad_password", + ) raise HTTPException(status_code=401, detail="Invalid credentials") if not u.get("is_active", True) or u.get("status") != "ACTIVE": @@ -90,12 +112,21 @@ def login( "login_failed", extra={"email": payload.email, "user_id": u.get("id"), "reason": "disabled"}, ) + login_audit_service.record_safe( + db, + event=login_audit_service.EVENT_LOGIN_FAILED, + email=payload.email, + user_id=str(u["id"]), + tenant_id=str(u["tenant_id"]) if u.get("tenant_id") else None, + ip=ip, + user_agent=ua, + reason="disabled", + ) raise HTTPException(status_code=403, detail="User disabled") # Émet l'access token + persiste un refresh token access_token, expires_in = _access_token_for(u) - ip, ua = _client_meta(request) refresh_token, refresh_expires_at = issue_refresh_token( db, user_id=str(u["id"]), @@ -105,6 +136,29 @@ def login( ) db.commit() + # Journal de connexion — APRÈS le commit de la session : une panne du + # journal ne doit pas annuler une connexion déjà acquise. + evenement = login_audit_service.record_safe( + db, + event=login_audit_service.EVENT_LOGIN_OK, + email=u.get("email") or payload.email, + user_id=str(u["id"]), + tenant_id=str(u["tenant_id"]), + ip=ip, + user_agent=ua, + detect_new_context=True, + ) + login_audit_service.touch_last_login(db, user_id=str(u["id"]), ip=ip) + + # Adresse ou appareil jamais vus → alerte à la Conformité, hors du + # chemin de réponse (une session SMTP prend plusieurs secondes). + if evenement and evenement.get("is_new_context"): + background.add_task( + login_audit_service.notify_new_context, + evenement, + full_name=u.get("full_name"), + ) + user_id_ctx.set(str(u["id"])) logger.info( "login_success", @@ -154,10 +208,14 @@ def refresh( # Recharge l'utilisateur (status peut avoir changé entre login et refresh) u = db.execute( text(""" - SELECT id::text AS id, email, full_name, is_active, status, - tenant_id::text AS tenant_id - FROM public.users - WHERE id = CAST(:uid AS uuid) + SELECT u.id::text AS id, u.email, u.full_name, u.is_active, u.status, + u.tenant_id::text AS tenant_id, + EXISTS ( + SELECT 1 FROM public.user_roles ur + WHERE ur.user_id = u.id AND ur.role = 'SUPER_ADMIN' + ) AS is_super_admin + FROM public.users u + WHERE u.id = CAST(:uid AS uuid) """), {"uid": found["user_id"]}, ).mappings().first() @@ -181,6 +239,16 @@ def refresh( user_agent=ua, ) db.commit() + login_audit_service.record_safe( + db, + event=login_audit_service.EVENT_REFRESH, + email=u["email"], + user_id=str(u["id"]), + tenant_id=str(u["tenant_id"]), + ip=ip, + user_agent=ua, + reason="rotated", + ) logger.info("refresh_rotated", extra={"user_id": str(u["id"])}) return TokenPairResponse( access_token=access_token, @@ -210,6 +278,7 @@ def refresh( @router.post("/logout", status_code=204) def logout( + request: Request, payload: LogoutRequest, user=Depends(get_current_user), db: Session = Depends(get_db_public), @@ -230,14 +299,30 @@ def logout( logger.exception("SET ROLE auth_bypass failed (logout)") raise HTTPException(status_code=500, detail="Auth DB role misconfigured") + ip, ua = _client_meta(request) + + def _trace(motif: str) -> None: + login_audit_service.record_safe( + db, + event=login_audit_service.EVENT_LOGOUT, + email=user.get("email"), + user_id=str(user["id"]), + tenant_id=str(user["tenant_id"]) if user.get("tenant_id") else None, + ip=ip, + user_agent=ua, + reason=motif, + ) + try: if payload.all_devices: n = revoke_all_user_tokens(db, user_id=str(user["id"]), reason="logout_all") db.commit() + _trace("logout_all") logger.info("logout_all", extra={"user_id": str(user["id"]), "revoked": n}) elif payload.refresh_token: revoke_refresh_token(db, token=payload.refresh_token, reason="logout") db.commit() + _trace("logout") logger.info("logout_single", extra={"user_id": str(user["id"])}) else: # Pas de refresh fourni, pas de all_devices → noop accepté diff --git a/app/api/routes/security.py b/app/api/routes/security.py new file mode 100644 index 0000000..85865a3 --- /dev/null +++ b/app/api/routes/security.py @@ -0,0 +1,125 @@ +# app/api/routes/security.py +""" +Console de sécurité — journal des connexions (Module 2). + +Réservée au super-administrateur : ces écrans exposent des adresses IP, des +appareils et les comptes de TOUS les tenants. C'est aussi la raison pour +laquelle ils s'appuient sur `get_db_public` + le rôle d'authentification et non +sur la session RLS : la politique `tenant_isolation_users` filtre par tenant +sans échappatoire super-admin, donc une session RLS ne verrait qu'un tenant. +Le contournement est ici volontaire, borné à ces quatre lectures, et refermé +par un `RESET ROLE` systématique. +""" +from __future__ import annotations + +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.api.deps.auth import get_current_user +from app.api.deps.db import get_db_public +from app.core.config import settings +from app.core.logging import get_logger +from app.services import login_audit_service + +router = APIRouter(prefix="/security", tags=["security"]) +logger = get_logger("simandou.security") + + +def _require_super_admin(user: dict) -> dict: + if not user.get("is_super_admin"): + raise HTTPException(status_code=403, detail="Réservé au super-administrateur") + return user + + +def _open(db: Session) -> None: + try: + db.execute(text(f"SET ROLE {settings.AUTH_BYPASS_ROLE}")) + except Exception: + logger.exception("security_set_role_failed") + raise HTTPException(status_code=500, detail="Auth DB role misconfigured") + + +def _close(db: Session) -> None: + try: + db.execute(text("RESET ROLE")) + except Exception: + logger.exception("security_reset_role_failed") + + +@router.get("/login-events") +def login_events( + limit: int = Query(50, ge=1, le=200), + offset: int = Query(0, ge=0), + event: Optional[str] = Query(None, description="LOGIN_OK | LOGIN_FAILED | LOGOUT | REFRESH"), + email: Optional[str] = None, + days: Optional[int] = Query(None, ge=1, le=365), + only_new_context: bool = False, + user=Depends(get_current_user), + db: Session = Depends(get_db_public), +): + """Journal des tentatives d'accès, du plus récent au plus ancien.""" + _require_super_admin(user) + + if event and event not in login_audit_service.EVENTS: + raise HTTPException(status_code=422, detail=f"Événement inconnu : {event}") + + _open(db) + try: + return login_audit_service.list_events( + db, + limit=limit, + offset=offset, + event=event, + email=email, + days=days, + only_new_context=only_new_context, + ) + finally: + _close(db) + + +@router.get("/login-summary") +def login_summary( + user=Depends(get_current_user), + db: Session = Depends(get_db_public), +): + """Chiffres de tête : connexions 24 h / 7 j, comptes distincts, échecs, contextes inconnus.""" + _require_super_admin(user) + _open(db) + try: + return login_audit_service.summary(db) + finally: + _close(db) + + +@router.get("/sessions") +def sessions( + limit: int = Query(100, ge=1, le=500), + user=Depends(get_current_user), + db: Session = Depends(get_db_public), +): + """Sessions ouvertes — qui est connecté en ce moment.""" + _require_super_admin(user) + _open(db) + try: + return {"items": login_audit_service.active_sessions(db, limit=limit)} + finally: + _close(db) + + +@router.get("/accounts") +def accounts( + limit: int = Query(200, ge=1, le=500), + user=Depends(get_current_user), + db: Session = Depends(get_db_public), +): + """Comptes et dernière connexion — met en évidence les comptes jamais utilisés.""" + _require_super_admin(user) + _open(db) + try: + return {"items": login_audit_service.accounts(db, limit=limit)} + finally: + _close(db) diff --git a/app/main.py b/app/main.py index ebaa437..0f67cf6 100644 --- a/app/main.py +++ b/app/main.py @@ -40,6 +40,7 @@ from app.api.routes.reporting import router as reporting_router # noqa: E402 from app.api.routes.scoring import router as scoring_router # noqa: E402 from app.api.routes.screening import router as screening_router # noqa: E402 +from app.api.routes.security import router as security_router # noqa: E402 from app.api.routes.settings_routes import router as settings_router # noqa: E402 @@ -121,3 +122,4 @@ async def _rate_limit_handler(request, exc: RateLimitExceeded): app.include_router(rbac_router) app.include_router(adverse_media_router) app.include_router(blacklist_router) +app.include_router(security_router) diff --git a/app/models/login_event.py b/app/models/login_event.py new file mode 100644 index 0000000..2a8ca13 --- /dev/null +++ b/app/models/login_event.py @@ -0,0 +1,75 @@ +# app/models/login_event.py +""" +Journal des connexions (Module 2 — Contrôle Accès et Sécurité). + +Le TDR exige des traces historisées systématiquement. Les tentatives d'accès +n'étaient jusqu'ici écrites que dans le journal applicatif (stdout de +l'hébergeur) : ni requêtables, ni conservées. On ne pouvait donc pas répondre +à « qui s'est connecté, depuis où, et quand » — qui est précisément la +question que se pose la Conformité après une remise d'offre ou un incident. + +Deux choix de conception à connaître : + +1. Les ÉCHECS sont journalisés au même titre que les succès. Une série + d'échecs sur une adresse valide est le premier signe d'une attaque par + force brute ; ne garder que les succès revient à ne voir que les attaques + qui ont réussi. + +2. Aucune clé étrangère vers `users`. Un événement doit survivre à la + suppression du compte — sinon la piste d'audit s'efface en même temps que + le compte suspect, ce qui la rend sans valeur probante. Et un échec sur une + adresse inconnue n'a, par définition, aucun utilisateur à référencer : + c'est `email` qui porte alors l'information. +""" +from __future__ import annotations + +import uuid + +from sqlalchemy import Boolean, Column, DateTime, Index, String, Text, text +from sqlalchemy.dialects.postgresql import UUID +from sqlalchemy.sql import func + +from app.models.base import Base + + +class LoginEvent(Base): + __tablename__ = "login_events" + + id = Column( + UUID(as_uuid=True), + primary_key=True, + default=uuid.uuid4, + server_default=text("gen_random_uuid()"), + ) + + # LOGIN_OK | LOGIN_FAILED | LOGOUT | REFRESH + event = Column(String(16), nullable=False) + # Motif d'échec ou précision : unknown_user | bad_password | disabled | rotated | logout_all + reason = Column(String(32), nullable=True) + + # NULL si l'adresse saisie ne correspond à aucun compte. + user_id = Column(UUID(as_uuid=True), nullable=True) + # Toujours renseigné : c'est la saisie de l'utilisateur, y compris quand + # elle ne correspond à rien. + email = Column(Text, nullable=True) + tenant_id = Column(UUID(as_uuid=True), nullable=True) + + ip = Column(String(64), nullable=True) + user_agent = Column(Text, nullable=True) + + # Vrai quand cette connexion vient d'une adresse IP ou d'un appareil + # jamais observés pour ce compte. C'est ce drapeau qui déclenche l'alerte + # par courriel — et qui permet de retrouver les connexions notables sans + # relire tout le journal. + is_new_context = Column(Boolean, nullable=False, server_default=text("false")) + + created_at = Column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + Index("ix_login_events_created_at", "created_at"), + Index("ix_login_events_user_created", "user_id", "created_at"), + Index("ix_login_events_email", "email"), + Index("ix_login_events_ip", "ip"), + ) diff --git a/app/models/users.py b/app/models/users.py index 6343ece..df364f4 100644 --- a/app/models/users.py +++ b/app/models/users.py @@ -2,7 +2,7 @@ from __future__ import annotations import uuid -from sqlalchemy import Column, Text, Boolean, DateTime, ForeignKey +from sqlalchemy import Column, Text, Boolean, DateTime, ForeignKey, String from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.sql import func from sqlalchemy.orm import relationship @@ -32,5 +32,11 @@ class User(Base): tenant_id = Column(UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="RESTRICT"), nullable=True, index=True) tenant = relationship("Tenant", lazy="joined") + # ✅ dernière connexion réussie — dénormalisé depuis `login_events` pour + # répondre en une requête à « ce compte sert-il encore ? » sans balayer + # le journal. Le détail (échecs, IP successives) reste dans login_events. + last_login_at = Column(DateTime(timezone=True), nullable=True) + last_login_ip = Column(String(64), nullable=True) + created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now()) updated_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()) diff --git a/app/scripts/grant_super_admin.py b/app/scripts/grant_super_admin.py new file mode 100644 index 0000000..fb0fdbe --- /dev/null +++ b/app/scripts/grant_super_admin.py @@ -0,0 +1,117 @@ +""" +Attribution (ou retrait) du rôle SUPER_ADMIN à un compte. + +Le super-administrateur est le seul profil qui voit au-delà d'un tenant : +bascule de tenant (`X-Tenant-Id`), contexte RLS élargi, et console de sécurité +(`/security/*`). Il ne s'attribue donc pas depuis l'interface — c'est une +opération d'exploitation, tracée par ce script. + + python -m app.scripts.grant_super_admin conformite@bcrg-guinee.org + python -m app.scripts.grant_super_admin conformite@bcrg-guinee.org --revoke + python -m app.scripts.grant_super_admin --list + +Le rôle est écrit dans `user_roles` (table des rôles structurels), et non dans +`rbac_user_roles` (habilitations paramétrables par tenant) : c'est cette +première table que lit l'émission du jeton. + +Prendre effet exige une NOUVELLE connexion : la revendication est portée par +l'access token, pas relue à chaque requête. +""" +from __future__ import annotations + +import sys + +from sqlalchemy import text + +from app.core.config import settings +from app.core.db import engine + +ROLE = "SUPER_ADMIN" + + +def _bypass(conn) -> None: + """RLS contourné : `users` est isolé par tenant, or on agit hors tenant.""" + conn.execute(text(f"SET ROLE {settings.AUTH_BYPASS_ROLE}")) + + +def lister() -> int: + with engine.begin() as conn: + _bypass(conn) + rows = conn.execute( + text(f""" + SELECT u.email, u.full_name, ur.created_at + FROM public.user_roles ur + JOIN public.users u ON u.id = ur.user_id + WHERE ur.role = '{ROLE}' + ORDER BY ur.created_at + """) + ).mappings().all() + + if not rows: + print("Aucun super-administrateur déclaré.") + return 0 + print(f"{len(rows)} super-administrateur(s) :") + for r in rows: + print(f" · {r['email']} ({r['full_name']}) depuis le {r['created_at']:%d/%m/%Y}") + return 0 + + +def appliquer(email: str, *, revoke: bool) -> int: + with engine.begin() as conn: + _bypass(conn) + + row = conn.execute( + text(""" + SELECT id::text AS id, tenant_id::text AS tenant_id, full_name + FROM public.users + WHERE lower(trim(email)) = lower(trim(:e)) + """), + {"e": email}, + ).mappings().first() + + if not row: + print(f"✗ Aucun compte pour {email}") + return 1 + if not row["tenant_id"]: + print(f"✗ {email} n'est rattaché à aucun tenant — rattachez-le d'abord.") + return 1 + + if revoke: + n = conn.execute( + text(f""" + DELETE FROM public.user_roles + WHERE user_id = CAST(:uid AS uuid) AND role = '{ROLE}' + """), + {"uid": row["id"]}, + ).rowcount + print(f"✓ SUPER_ADMIN retiré à {email}" if n else f"= {email} n'était pas SUPER_ADMIN") + else: + conn.execute( + text(f""" + INSERT INTO public.user_roles (id, tenant_id, user_id, role) + VALUES (gen_random_uuid(), CAST(:tid AS uuid), CAST(:uid AS uuid), '{ROLE}') + ON CONFLICT (tenant_id, user_id, role) DO NOTHING + """), + {"tid": row["tenant_id"], "uid": row["id"]}, + ) + print(f"✓ {email} ({row['full_name']}) est super-administrateur.") + + print("→ Effectif à la prochaine connexion (la revendication est dans le jeton).") + return 0 + + +def main() -> int: + args = [a for a in sys.argv[1:] if a] + if not args or "--list" in args: + return lister() + + revoke = "--revoke" in args + emails = [a for a in args if not a.startswith("--")] + if len(emails) != 1: + print(__doc__) + return 2 + return appliquer(emails[0], revoke=revoke) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/app/services/auth_service.py b/app/services/auth_service.py index fa53ace..66d177d 100644 --- a/app/services/auth_service.py +++ b/app/services/auth_service.py @@ -98,18 +98,28 @@ def decode_access_token(token: str) -> Dict[str, Any]: def get_user_by_email(db: Session, email: str): email_norm = (email or "").strip().lower() + # `is_super_admin` est calculé ici, à la source du jeton. Sans cela, la + # revendication émise dans l'access token valait toujours `false` et TOUTES + # les portes qui en dépendent (bascule de tenant, console de sécurité, + # contexte RLS super-admin) restaient fermées pour tout le monde. Le rôle + # SUPER_ADMIN n'étant attribué à personne par défaut, ce calcul ne change + # rien aux comptes existants. row = db.execute( text(""" SELECT - id::text AS id, - email, - full_name, - password_hash, - is_active, - status, - tenant_id::text AS tenant_id - FROM public.users - WHERE lower(trim(email)) = :email + u.id::text AS id, + u.email, + u.full_name, + u.password_hash, + u.is_active, + u.status, + u.tenant_id::text AS tenant_id, + EXISTS ( + SELECT 1 FROM public.user_roles ur + WHERE ur.user_id = u.id AND ur.role = 'SUPER_ADMIN' + ) AS is_super_admin + FROM public.users u + WHERE lower(trim(u.email)) = :email LIMIT 1 """), {"email": email_norm}, diff --git a/app/services/list_notifier.py b/app/services/list_notifier.py index cb409af..d4bb63e 100644 --- a/app/services/list_notifier.py +++ b/app/services/list_notifier.py @@ -126,19 +126,33 @@ def _render_html(report: dict) -> str: # ── Brevo SMTP ──────────────────────────────────────────────────────────────── -def _send_brevo_smtp(subject: str, html: str): - if not (BREVO_SMTP_LOGIN and BREVO_SMTP_KEY and BREVO_FROM_EMAIL and BREVO_TO_EMAIL): +def send_html_email(subject: str, html: str, recipients: Optional[str] = None) -> None: + """ + Envoi générique par le relais déjà configuré. + + Exposé pour les autres modules qui doivent notifier la Conformité (journal + de connexions, par exemple) sans redéclarer une configuration SMTP. + """ + _send_brevo_smtp(subject, html, recipients=recipients) + + +def _send_brevo_smtp(subject: str, html: str, recipients: Optional[str] = None): + # `recipients` (liste séparée par des virgules) permet à un module de viser + # une autre boîte que celle du compte rendu des listes. + brut = recipients or BREVO_TO_EMAIL + + if not (BREVO_SMTP_LOGIN and BREVO_SMTP_KEY and BREVO_FROM_EMAIL and brut): logger.info("Brevo SMTP not fully configured – skipping email") return - recipients = [r.strip() for r in BREVO_TO_EMAIL.split(",") if r.strip()] - if not recipients: + destinataires = [r.strip() for r in brut.split(",") if r.strip()] + if not destinataires: return msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = f"{BREVO_FROM_NAME} <{BREVO_FROM_EMAIL}>" - msg["To"] = ", ".join(recipients) + msg["To"] = ", ".join(destinataires) msg.attach(MIMEText(html, "html", "utf-8")) try: @@ -146,8 +160,8 @@ def _send_brevo_smtp(subject: str, html: str): srv.ehlo() srv.starttls() srv.login(BREVO_SMTP_LOGIN, BREVO_SMTP_KEY) - srv.sendmail(BREVO_FROM_EMAIL, recipients, msg.as_string()) - logger.info("Brevo SMTP email sent to %s", recipients) + srv.sendmail(BREVO_FROM_EMAIL, destinataires, msg.as_string()) + logger.info("Brevo SMTP email sent to %s", destinataires) except Exception as exc: logger.error("Brevo SMTP email failed: %s", exc) diff --git a/app/services/login_audit_service.py b/app/services/login_audit_service.py new file mode 100644 index 0000000..fc03afc --- /dev/null +++ b/app/services/login_audit_service.py @@ -0,0 +1,468 @@ +""" +Journal des connexions — écriture, détection de contexte inconnu, lecture. + +Ce service répond à une question opérationnelle simple : « qui s'est connecté, +depuis où, et quand ». Il alimente la table `login_events` (cf. le modèle pour +les choix de conception) et met à jour `users.last_login_at`. + +Principe directeur : **journaliser ne doit jamais empêcher de se connecter**. +Toutes les écritures passent par `record_safe`, qui avale ses erreurs après les +avoir tracées. Une panne du journal dégrade la traçabilité ; elle ne doit pas +fermer l'accès à la plateforme de conformité. + +Détection du contexte inconnu : une connexion est signalée quand son adresse IP +ou son appareil n'a jamais été observé pour ce compte. La toute première +connexion d'un compte est donc signalée — c'est voulu : c'est l'événement qu'on +attend après avoir remis des accès à un tiers. +""" +from __future__ import annotations + +import logging +import os +from typing import Any, Optional + +from sqlalchemy import text +from sqlalchemy.orm import Session + +logger = logging.getLogger("simandou.login_audit") + +EVENT_LOGIN_OK = "LOGIN_OK" +EVENT_LOGIN_FAILED = "LOGIN_FAILED" +EVENT_LOGOUT = "LOGOUT" +EVENT_REFRESH = "REFRESH" + +EVENTS = (EVENT_LOGIN_OK, EVENT_LOGIN_FAILED, EVENT_LOGOUT, EVENT_REFRESH) + +EVENT_LABELS = { + EVENT_LOGIN_OK: "Connexion réussie", + EVENT_LOGIN_FAILED: "Échec de connexion", + EVENT_LOGOUT: "Déconnexion", + EVENT_REFRESH: "Session prolongée", +} + +REASON_LABELS = { + "unknown_user": "adresse inconnue", + "bad_password": "mot de passe incorrect", + "disabled": "compte désactivé", + "rotated": "renouvellement du jeton", + "logout": "déconnexion", + "logout_all": "déconnexion de tous les appareils", +} + + +# ── Écriture ────────────────────────────────────────────────────────────────── + +def _known_context( + db: Session, + *, + user_id: Optional[str], + email: Optional[str], + ip: Optional[str], + user_agent: Optional[str], +) -> tuple[bool, bool, int]: + """ + (adresse déjà vue, appareil déjà vu, nombre de connexions réussies passées). + + On se restreint aux connexions RÉUSSIES : un échec depuis une adresse + inconnue ne doit pas « blanchir » cette adresse pour la connexion suivante. + """ + if user_id: + where = "user_id = CAST(:uid AS uuid)" + params: dict[str, Any] = {"uid": str(user_id)} + elif email: + where = "lower(email) = lower(:email)" + params = {"email": email} + else: + return (False, False, 0) + + params.update({"ip": ip, "ua": user_agent}) + + row = db.execute( + text(f""" + SELECT + COALESCE(bool_or(ip IS NOT DISTINCT FROM :ip), false) AS ip_seen, + COALESCE(bool_or(user_agent IS NOT DISTINCT FROM :ua), false) AS ua_seen, + COUNT(*)::int AS previous + FROM public.login_events + WHERE event = '{EVENT_LOGIN_OK}' AND {where} + """), + params, + ).mappings().first() + + if not row: + return (False, False, 0) + return (bool(row["ip_seen"]), bool(row["ua_seen"]), int(row["previous"])) + + +def record( + db: Session, + *, + event: str, + email: Optional[str] = None, + user_id: Optional[str] = None, + tenant_id: Optional[str] = None, + ip: Optional[str] = None, + user_agent: Optional[str] = None, + reason: Optional[str] = None, + detect_new_context: bool = False, +) -> dict: + """Écrit un événement et le renvoie. Peut lever — préférer `record_safe`.""" + known_ip = known_device = True + previous = 1 + + if detect_new_context: + known_ip, known_device, previous = _known_context( + db, user_id=user_id, email=email, ip=ip, user_agent=user_agent + ) + + is_new_context = detect_new_context and (not known_ip or not known_device) + + row = db.execute( + text(""" + INSERT INTO public.login_events + (event, reason, user_id, email, tenant_id, ip, user_agent, is_new_context) + VALUES + (:event, :reason, CAST(:uid AS uuid), :email, CAST(:tid AS uuid), + :ip, :ua, :new_ctx) + RETURNING id::text AS id, created_at + """), + { + "event": event, + "reason": reason, + "uid": str(user_id) if user_id else None, + "email": (email or "")[:320] or None, + "tid": str(tenant_id) if tenant_id else None, + "ip": (ip or "")[:64] or None, + "ua": (user_agent or "")[:1024] or None, + "new_ctx": is_new_context, + }, + ).mappings().first() + + return { + "id": row["id"], + "created_at": row["created_at"], + "event": event, + "reason": reason, + "user_id": str(user_id) if user_id else None, + "email": email, + "tenant_id": str(tenant_id) if tenant_id else None, + "ip": ip, + "user_agent": user_agent, + "is_new_context": is_new_context, + "is_first_login": detect_new_context and previous == 0, + "new_ip": detect_new_context and not known_ip, + "new_device": detect_new_context and not known_device, + } + + +def record_safe(db: Session, **kwargs) -> Optional[dict]: + """ + `record` + commit, sans jamais propager d'erreur. + + Appelée depuis les routes d'authentification : un journal indisponible ne + doit pas transformer une connexion valide en erreur 500. + """ + try: + out = record(db, **kwargs) + db.commit() + return out + except Exception: + logger.exception("login_event_write_failed", extra={"event": kwargs.get("event")}) + try: + db.rollback() + except Exception: + pass + return None + + +def touch_last_login(db: Session, *, user_id: str, ip: Optional[str]) -> None: + """Met à jour la dernière connexion réussie du compte. Ne lève jamais.""" + try: + db.execute( + text(""" + UPDATE public.users + SET last_login_at = now(), + last_login_ip = :ip + WHERE id = CAST(:uid AS uuid) + """), + {"uid": str(user_id), "ip": (ip or "")[:64] or None}, + ) + db.commit() + except Exception: + logger.exception("last_login_update_failed", extra={"user_id": str(user_id)}) + try: + db.rollback() + except Exception: + pass + + +# ── Alerte par courriel ─────────────────────────────────────────────────────── + +def _alert_enabled() -> bool: + return (os.getenv("LOGIN_ALERT_ENABLED", "true") or "").strip().lower() not in { + "0", "false", "no", "off", + } + + +def _alert_recipients() -> Optional[str]: + """Destinataires dédiés, sinon la boîte Conformité déjà configurée.""" + return os.getenv("LOGIN_ALERT_TO_EMAIL") or os.getenv("BREVO_TO_EMAIL") + + +def _motif(event: dict) -> str: + if event.get("is_first_login"): + return "première connexion de ce compte" + bouts = [] + if event.get("new_ip"): + bouts.append("adresse IP jamais vue") + if event.get("new_device"): + bouts.append("appareil ou navigateur jamais vu") + return " et ".join(bouts) or "contexte inhabituel" + + +def build_alert_email(event: dict, *, full_name: Optional[str] = None) -> tuple[str, str]: + """(sujet, corps HTML) de l'alerte de connexion inhabituelle.""" + qui = full_name or event.get("email") or "compte inconnu" + quand = event.get("created_at") + quand_txt = quand.strftime("%d/%m/%Y à %H:%M UTC") if hasattr(quand, "strftime") else str(quand) + + sujet = f"[LBC/FT] Nouvelle connexion — {qui}" + html = f""" +
+

Connexion depuis un contexte inconnu

+

Motif : {_motif(event)}.

+ + + + + + +
Compte{qui}
Adresse électronique{event.get('email') or '—'}
Date{quand_txt}
Adresse IP{event.get('ip') or '—'}
Appareil{(event.get('user_agent') or '—')[:200]}
+

+ Le détail est consultable dans Journal de connexions (espace administrateur). + Si cette connexion n'est pas attendue, révoquez les sessions du compte + et changez son mot de passe. +

+
+ """ + return sujet, html + + +def notify_new_context(event: dict, *, full_name: Optional[str] = None) -> None: + """ + Envoie l'alerte de connexion inhabituelle. Ne lève jamais. + + Appelée en tâche de fond : l'ouverture d'une session SMTP prend parfois + plusieurs secondes, ce qui ralentirait la connexion de l'utilisateur. + """ + if not event or not event.get("is_new_context"): + return + if not _alert_enabled(): + return + + destinataires = _alert_recipients() + if not destinataires: + logger.info("login_alert_skipped_no_recipient") + return + + try: + from app.services import list_notifier + + sujet, html = build_alert_email(event, full_name=full_name) + list_notifier.send_html_email(sujet, html, recipients=destinataires) + except Exception: + logger.exception("login_alert_failed") + + +# ── Lecture (console de sécurité) ───────────────────────────────────────────── + +def list_events( + db: Session, + *, + limit: int = 50, + offset: int = 0, + event: Optional[str] = None, + email: Optional[str] = None, + days: Optional[int] = None, + only_new_context: bool = False, +) -> dict: + """Journal paginé, du plus récent au plus ancien.""" + clauses: list[str] = [] + params: dict[str, Any] = {"limit": max(1, min(limit, 200)), "offset": max(0, offset)} + + if event: + clauses.append("e.event = :event") + params["event"] = event + if email: + clauses.append("e.email ILIKE :email") + params["email"] = f"%{email.strip()}%" + if days: + clauses.append("e.created_at >= now() - make_interval(days => :days)") + params["days"] = int(days) + if only_new_context: + clauses.append("e.is_new_context") + + where = f"WHERE {' AND '.join(clauses)}" if clauses else "" + + total = db.execute( + text(f"SELECT COUNT(*)::int FROM public.login_events e {where}"), params + ).scalar() or 0 + + rows = db.execute( + text(f""" + SELECT + e.id::text AS id, + e.event, + e.reason, + e.email, + e.user_id::text AS user_id, + e.tenant_id::text AS tenant_id, + e.ip, + e.user_agent, + e.is_new_context, + e.created_at, + u.full_name, + t.name AS tenant_name + FROM public.login_events e + LEFT JOIN public.users u ON u.id = e.user_id + LEFT JOIN public.tenants t ON t.id = e.tenant_id + {where} + ORDER BY e.created_at DESC + LIMIT :limit OFFSET :offset + """), + params, + ).mappings().all() + + return { + "items": [_serialize_event(r) for r in rows], + "total": total, + "limit": params["limit"], + "offset": params["offset"], + } + + +def _serialize_event(r) -> dict: + d = dict(r) + d["event_label"] = EVENT_LABELS.get(d.get("event"), d.get("event")) + d["reason_label"] = REASON_LABELS.get(d.get("reason")) if d.get("reason") else None + created = d.get("created_at") + d["created_at"] = created.isoformat() if hasattr(created, "isoformat") else created + return d + + +def summary(db: Session) -> dict: + """Chiffres de tête de la console de sécurité (fenêtres 24 h et 7 jours).""" + row = db.execute( + text(f""" + SELECT + COUNT(*) FILTER ( + WHERE event = '{EVENT_LOGIN_OK}' AND created_at >= now() - interval '24 hours' + )::int AS logins_24h, + COUNT(*) FILTER ( + WHERE event = '{EVENT_LOGIN_OK}' AND created_at >= now() - interval '7 days' + )::int AS logins_7d, + COUNT(DISTINCT user_id) FILTER ( + WHERE event = '{EVENT_LOGIN_OK}' AND created_at >= now() - interval '7 days' + )::int AS users_7d, + COUNT(*) FILTER ( + WHERE event = '{EVENT_LOGIN_FAILED}' AND created_at >= now() - interval '24 hours' + )::int AS failures_24h, + COUNT(*) FILTER ( + WHERE is_new_context AND created_at >= now() - interval '7 days' + )::int AS new_contexts_7d + FROM public.login_events + """) + ).mappings().first() + + derniere = db.execute( + text(f""" + SELECT created_at, email, ip + FROM public.login_events + WHERE event = '{EVENT_LOGIN_OK}' + ORDER BY created_at DESC + LIMIT 1 + """) + ).mappings().first() + + out = dict(row or {}) + if derniere: + out["last_login"] = { + "created_at": derniere["created_at"].isoformat(), + "email": derniere["email"], + "ip": derniere["ip"], + } + else: + out["last_login"] = None + return out + + +def active_sessions(db: Session, *, limit: int = 100) -> list[dict]: + """ + Sessions ouvertes = jetons de rafraîchissement ni révoqués ni expirés. + + C'est la réponse à « qui est connecté en ce moment », par opposition au + journal qui répond à « qui s'est connecté ». + """ + rows = db.execute( + text(""" + SELECT + r.id::text AS id, + r.user_id::text AS user_id, + u.email, + u.full_name, + t.name AS tenant_name, + r.issued_at, + r.expires_at, + r.client_ip AS ip, + r.user_agent + FROM public.refresh_tokens r + LEFT JOIN public.users u ON u.id = r.user_id + LEFT JOIN public.tenants t ON t.id = r.tenant_id + WHERE r.revoked_at IS NULL + AND r.expires_at > now() + ORDER BY r.issued_at DESC + LIMIT :limit + """), + {"limit": max(1, min(limit, 500))}, + ).mappings().all() + + out = [] + for r in rows: + d = dict(r) + for champ in ("issued_at", "expires_at"): + v = d.get(champ) + d[champ] = v.isoformat() if hasattr(v, "isoformat") else v + out.append(d) + return out + + +def accounts(db: Session, *, limit: int = 200) -> list[dict]: + """Inventaire des comptes avec leur dernière connexion — comptes dormants inclus.""" + rows = db.execute( + text(""" + SELECT + u.id::text AS id, + u.email, + u.full_name, + u.status, + u.is_active, + t.name AS tenant_name, + u.created_at, + u.last_login_at, + u.last_login_ip + FROM public.users u + LEFT JOIN public.tenants t ON t.id = u.tenant_id + ORDER BY u.last_login_at DESC NULLS LAST, u.created_at DESC + LIMIT :limit + """), + {"limit": max(1, min(limit, 500))}, + ).mappings().all() + + out = [] + for r in rows: + d = dict(r) + for champ in ("created_at", "last_login_at"): + v = d.get(champ) + d[champ] = v.isoformat() if hasattr(v, "isoformat") else v + out.append(d) + return out diff --git a/render.yaml b/render.yaml index 7fccb2d..6b1b9c3 100644 --- a/render.yaml +++ b/render.yaml @@ -83,6 +83,24 @@ services: - key: ANTHROPIC_API_KEY sync: false + # --- Alerte de connexion inhabituelle (journal des connexions) --------- + # L'alerte part du service WEB (c'est lui qui authentifie), d'où la + # reprise des identifiants SMTP déjà utilisés par les crons. Sans eux, + # le journal continue de tout enregistrer : seul le courriel est ignoré. + - key: BREVO_SMTP_LOGIN + sync: false + - key: BREVO_SMTP_KEY + sync: false + - key: BREVO_FROM_EMAIL + sync: false + - key: BREVO_FROM_NAME + value: Simandou Conformité + # Destinataire des alertes de connexion. À défaut, BREVO_TO_EMAIL. + - key: LOGIN_ALERT_TO_EMAIL + sync: false + - key: LOGIN_ALERT_ENABLED + value: "true" + # --- Seed des 2 comptes (facultatif : surcharge les mots de passe) ----- # - key: SEED_COMPLIANCE_PASSWORD # sync: false diff --git a/tests/integration/test_login_audit.py b/tests/integration/test_login_audit.py new file mode 100644 index 0000000..c6422ea --- /dev/null +++ b/tests/integration/test_login_audit.py @@ -0,0 +1,222 @@ +""" +Tests d'intégration du journal des connexions. + +Ce qui est vérifié ici tient en quatre points : une connexion laisse une trace, +un échec aussi, un contexte déjà vu n'est pas signalé deux fois, et la console +de sécurité reste fermée à qui n'est pas super-administrateur. +""" +from __future__ import annotations + +import uuid + +import pytest +from sqlalchemy import text + +from app.services.auth_service import hash_password + +pytestmark = pytest.mark.integration + + +def _bypass(db): + db.execute(text("RESET ROLE")) + db.execute(text("SET ROLE auth_bypass_rls")) + + +@pytest.fixture +def compte(db): + """Un tenant + un compte actif. Le journal est purgé en sortie.""" + tenant_id = uuid.uuid4() + user_id = uuid.uuid4() + email = f"{user_id.hex[:8]}@example.com" + password = "test-pass-12345" + + _bypass(db) + db.execute( + text("INSERT INTO tenants (id, name, slug, status) VALUES (:tid, :n, :s, 'ACTIVE')"), + {"tid": tenant_id, "n": f"tenant-{tenant_id.hex[:6]}", "s": f"t-{tenant_id.hex[:8]}"}, + ) + db.execute( + text(""" + INSERT INTO users (id, email, full_name, password_hash, is_active, status, tenant_id) + VALUES (:uid, :email, 'Test User', :pw, true, 'ACTIVE', :tid) + """), + {"uid": user_id, "email": email, "pw": hash_password(password), "tid": tenant_id}, + ) + db.commit() + + yield { + "user_id": str(user_id), + "tenant_id": str(tenant_id), + "email": email, + "password": password, + } + + _bypass(db) + db.execute(text("DELETE FROM login_events WHERE user_id = :uid OR email = :e"), + {"uid": user_id, "e": email}) + db.execute(text("DELETE FROM user_roles WHERE user_id = :uid"), {"uid": user_id}) + db.execute(text("DELETE FROM refresh_tokens WHERE user_id = :uid"), {"uid": user_id}) + db.execute(text("DELETE FROM users WHERE id = :uid"), {"uid": user_id}) + db.execute(text("DELETE FROM tenants WHERE id = :tid"), {"tid": tenant_id}) + db.execute(text("RESET ROLE")) + db.commit() + + +def _events(db, *, user_id=None, email=None) -> list[dict]: + _bypass(db) + clause = "user_id = CAST(:uid AS uuid)" if user_id else "email = :email" + rows = db.execute( + text(f""" + SELECT event, reason, ip, user_agent, is_new_context, tenant_id::text AS tenant_id + FROM login_events WHERE {clause} ORDER BY created_at + """), + {"uid": user_id, "email": email}, + ).mappings().all() + db.execute(text("RESET ROLE")) + return [dict(r) for r in rows] + + +class TestEcriture: + def test_connexion_reussie_est_journalisee(self, client, compte, db): + r = client.post("/auth/login", json={"email": compte["email"], "password": compte["password"]}) + assert r.status_code == 200, r.text + + evs = _events(db, user_id=compte["user_id"]) + assert [e["event"] for e in evs] == ["LOGIN_OK"] + assert evs[0]["tenant_id"] == compte["tenant_id"] + # Premier accès de ce compte → contexte forcément inconnu. + assert evs[0]["is_new_context"] is True + + def test_derniere_connexion_reportee_sur_le_compte(self, client, compte, db): + client.post("/auth/login", json={"email": compte["email"], "password": compte["password"]}) + + _bypass(db) + row = db.execute( + text("SELECT last_login_at, last_login_ip FROM users WHERE id = CAST(:uid AS uuid)"), + {"uid": compte["user_id"]}, + ).mappings().first() + db.execute(text("RESET ROLE")) + assert row["last_login_at"] is not None + + def test_second_acces_meme_contexte_non_signale(self, client, compte, db): + for _ in range(2): + client.post("/auth/login", json={"email": compte["email"], "password": compte["password"]}) + + evs = [e for e in _events(db, user_id=compte["user_id"]) if e["event"] == "LOGIN_OK"] + assert len(evs) == 2 + assert evs[0]["is_new_context"] is True + assert evs[1]["is_new_context"] is False, "même IP et même appareil : rien de nouveau" + + def test_mauvais_mot_de_passe_journalise(self, client, compte, db): + r = client.post("/auth/login", json={"email": compte["email"], "password": "faux"}) + assert r.status_code == 401 + + evs = _events(db, user_id=compte["user_id"]) + assert [(e["event"], e["reason"]) for e in evs] == [("LOGIN_FAILED", "bad_password")] + + def test_adresse_inconnue_journalisee_sans_compte(self, client, db): + inconnu = f"fantome-{uuid.uuid4().hex[:8]}@example.com" + r = client.post("/auth/login", json={"email": inconnu, "password": "peu importe"}) + assert r.status_code == 401 + + evs = _events(db, email=inconnu) + assert [(e["event"], e["reason"]) for e in evs] == [("LOGIN_FAILED", "unknown_user")] + + _bypass(db) + db.execute(text("DELETE FROM login_events WHERE email = :e"), {"e": inconnu}) + db.execute(text("RESET ROLE")) + db.commit() + + def test_deconnexion_journalisee(self, client, compte, db): + login = client.post( + "/auth/login", json={"email": compte["email"], "password": compte["password"]} + ).json() + r = client.post( + "/auth/logout", + json={"refresh_token": login["refresh_token"]}, + headers={"Authorization": f"Bearer {login['access_token']}"}, + ) + assert r.status_code == 204 + + evs = _events(db, user_id=compte["user_id"]) + assert [e["event"] for e in evs] == ["LOGIN_OK", "LOGOUT"] + + +class TestConsoleDeSecurite: + def _token(self, client, compte) -> str: + return client.post( + "/auth/login", json={"email": compte["email"], "password": compte["password"]} + ).json()["access_token"] + + def test_refus_sans_super_admin(self, client, compte): + tok = self._token(client, compte) + for url in ("/security/login-events", "/security/sessions", + "/security/login-summary", "/security/accounts"): + r = client.get(url, headers={"Authorization": f"Bearer {tok}"}) + assert r.status_code == 403, f"{url} devrait être fermé : {r.status_code}" + + def test_refus_sans_jeton(self, client): + assert client.get("/security/login-events").status_code == 401 + + def test_super_admin_lit_le_journal(self, client, compte, db): + # Le rôle structurel SUPER_ADMIN est ce que lit l'émission du jeton. + _bypass(db) + db.execute( + text(""" + INSERT INTO user_roles (id, tenant_id, user_id, role) + VALUES (gen_random_uuid(), CAST(:tid AS uuid), CAST(:uid AS uuid), 'SUPER_ADMIN') + """), + {"tid": compte["tenant_id"], "uid": compte["user_id"]}, + ) + db.execute(text("RESET ROLE")) + db.commit() + + tok = self._token(client, compte) + entetes = {"Authorization": f"Bearer {tok}"} + + r = client.get("/security/login-events", headers=entetes) + assert r.status_code == 200, r.text + corps = r.json() + assert corps["total"] >= 1 + assert any(e["email"] == compte["email"] for e in corps["items"]) + + s = client.get("/security/login-summary", headers=entetes) + assert s.status_code == 200 + assert s.json()["logins_24h"] >= 1 + + sess = client.get("/security/sessions", headers=entetes) + assert sess.status_code == 200 + assert any(x["email"] == compte["email"] for x in sess.json()["items"]) + + acc = client.get("/security/accounts", headers=entetes) + assert acc.status_code == 200 + moi = [x for x in acc.json()["items"] if x["email"] == compte["email"]] + assert moi and moi[0]["last_login_at"] is not None + + def test_filtre_par_evenement(self, client, compte, db): + _bypass(db) + db.execute( + text(""" + INSERT INTO user_roles (id, tenant_id, user_id, role) + VALUES (gen_random_uuid(), CAST(:tid AS uuid), CAST(:uid AS uuid), 'SUPER_ADMIN') + """), + {"tid": compte["tenant_id"], "uid": compte["user_id"]}, + ) + db.execute(text("RESET ROLE")) + db.commit() + + client.post("/auth/login", json={"email": compte["email"], "password": "faux"}) + tok = self._token(client, compte) + + r = client.get( + "/security/login-events?event=LOGIN_FAILED&limit=50", + headers={"Authorization": f"Bearer {tok}"}, + ) + assert r.status_code == 200 + assert all(e["event"] == "LOGIN_FAILED" for e in r.json()["items"]) + + mauvais = client.get( + "/security/login-events?event=NIMPORTE_QUOI", + headers={"Authorization": f"Bearer {tok}"}, + ) + assert mauvais.status_code == 422 diff --git a/tests/unit/test_login_audit_alert.py b/tests/unit/test_login_audit_alert.py new file mode 100644 index 0000000..14497bd --- /dev/null +++ b/tests/unit/test_login_audit_alert.py @@ -0,0 +1,84 @@ +""" +Tests unitaires du courriel d'alerte de connexion (aucune base, aucun réseau). +""" +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from app.services import login_audit_service as svc + +pytestmark = pytest.mark.unit + + +def _event(**kw) -> dict: + base = { + "email": "conformite@bcrg-guinee.org", + "ip": "41.82.10.5", + "user_agent": "Mozilla/5.0", + "created_at": datetime(2026, 8, 7, 9, 30, tzinfo=timezone.utc), + "is_new_context": True, + "is_first_login": False, + "new_ip": True, + "new_device": False, + } + base.update(kw) + return base + + +class TestMotif: + def test_premiere_connexion_prime_sur_le_reste(self): + assert svc._motif(_event(is_first_login=True, new_ip=True, new_device=True)) == \ + "première connexion de ce compte" + + def test_ip_seule(self): + assert svc._motif(_event()) == "adresse IP jamais vue" + + def test_ip_et_appareil(self): + m = svc._motif(_event(new_device=True)) + assert "adresse IP" in m and "appareil" in m + + +class TestCourriel: + def test_sujet_et_corps_portent_les_faits(self): + sujet, html = svc.build_alert_email(_event(), full_name="Cellule de Conformité") + assert "Cellule de Conformité" in sujet + assert "41.82.10.5" in html + assert "07/08/2026" in html + assert "conformite@bcrg-guinee.org" in html + + def test_repli_sur_l_adresse_si_nom_absent(self): + sujet, _ = svc.build_alert_email(_event()) + assert "conformite@bcrg-guinee.org" in sujet + + +class TestGardeFous: + def test_rien_a_notifier_si_contexte_connu(self, monkeypatch): + appels = [] + monkeypatch.setattr(svc, "_alert_recipients", lambda: appels.append("lu") or "x@y.z") + svc.notify_new_context(_event(is_new_context=False)) + assert appels == [], "un contexte connu ne doit même pas chercher de destinataire" + + def test_desactivable_par_variable_d_environnement(self, monkeypatch): + monkeypatch.setenv("LOGIN_ALERT_ENABLED", "false") + assert svc._alert_enabled() is False + monkeypatch.setenv("LOGIN_ALERT_ENABLED", "true") + assert svc._alert_enabled() is True + + def test_sans_destinataire_aucun_envoi(self, monkeypatch): + monkeypatch.delenv("LOGIN_ALERT_TO_EMAIL", raising=False) + monkeypatch.delenv("BREVO_TO_EMAIL", raising=False) + # Ne doit pas lever : l'absence de configuration n'est pas une erreur. + svc.notify_new_context(_event()) + + def test_une_panne_d_envoi_reste_silencieuse(self, monkeypatch): + monkeypatch.setenv("LOGIN_ALERT_TO_EMAIL", "audit@bcrg-guinee.org") + + from app.services import list_notifier + + def _boom(*a, **k): + raise RuntimeError("SMTP down") + + monkeypatch.setattr(list_notifier, "send_html_email", _boom) + svc.notify_new_context(_event()) # ne doit pas lever