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