Merge remote-tracking branch 'origin/dev' into feat/data-schema

# Conflicts:
#	apps/backend/app/models/__init__.py

apps/backend/tests/api/test_auth.py

@@ -0,0 +1,208 @@
+from collections.abc import Iterator
+from uuid import uuid4
+
+import pytest
+from fastapi import FastAPI
+from httpx import AsyncClient
+
+from app.api.deps import get_auth_service
+from app.core.principal import Principal
+from app.core.roles import AccountKind, Role
+from app.services.auth import (
+    AuthenticatedSession,
+    InvalidCredentialsError,
+    RateLimitedError,
+    SessionRejectedError,
+)
+
+IDENTIFIANTS = {"email": "operateur@enervision.fr", "password": "un-mot-de-passe-valide"}
+
+PRINCIPAL = Principal(
+    id=uuid4(),
+    email="operateur@enervision.fr",
+    role=Role.OPERATEUR,
+    kind=AccountKind.HUMAIN,
+    must_change_password=False,
+)
+
+
+class FauxService:
+    def __init__(self, erreur: Exception | None = None) -> None:
+        self._erreur = erreur
+
+    async def refresh(self, **_: object) -> AuthenticatedSession:
+        return await self.authenticate()
+
+    async def logout(self, **_: object) -> None:
+        return None
+
+    async def authenticate(self, **_: object) -> AuthenticatedSession:
+        if self._erreur is not None:
+            raise self._erreur
+        return AuthenticatedSession(
+            principal=PRINCIPAL,
+            access_token="un.jeton.factice",
+            expires_in=900,
+            refresh_secret="un-secret-opaque",
+        )
+
+
+@pytest.fixture
+def fake_auth_service(app: FastAPI) -> Iterator[list[Exception | None]]:
+    programme: list[Exception | None] = [None]
+    app.dependency_overrides[get_auth_service] = lambda: FauxService(programme[0])
+    yield programme
+    app.dependency_overrides.pop(get_auth_service, None)
+
+
+async def test_login_returns_the_token_and_the_principal_when_credentials_match(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
+
+    assert response.status_code == 200
+    corps = response.json()
+    assert corps["access_token"] == "un.jeton.factice"
+    assert corps["token_type"] == "bearer"
+    assert corps["principal"]["role"] == "operateur"
+
+
+async def test_login_forbids_intermediaries_from_caching_the_response(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
+
+    assert response.headers["cache-control"] == "no-store"
+
+
+async def test_login_never_reveals_which_half_of_the_credentials_was_wrong(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    fake_auth_service[0] = InvalidCredentialsError("Identifiants invalides")
+
+    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
+
+    assert response.status_code == 401
+    assert response.json() == {"detail": "Identifiants invalides"}
+
+
+async def test_login_returns_429_with_a_retry_after_when_the_rate_limit_is_reached(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    fake_auth_service[0] = RateLimitedError(900)
+
+    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
+
+    assert response.status_code == 429
+    assert response.headers["retry-after"] == "900"
+
+
+@pytest.mark.parametrize(
+    "corps",
+    [
+        {"email": "pas-une-adresse", "password": "un-mot-de-passe-valide"},
+        {"email": "operateur@enervision.fr"},
+        {"email": "operateur@enervision.fr", "password": "x" * 129},
+    ],
+    ids=["adresse_invalide", "mot_de_passe_absent", "mot_de_passe_trop_long"],
+)
+async def test_login_rejects_a_malformed_body_without_echoing_the_password(
+    fake_auth_service: list[Exception | None], client: AsyncClient, corps: dict[str, str]
+) -> None:
+    response = await client.post("/api/v1/auth/login", json=corps)
+
+    assert response.status_code == 422
+    assert "un-mot-de-passe-valide" not in response.text
+    assert "x" * 129 not in response.text
+
+
+async def test_login_posts_an_http_only_refresh_cookie_scoped_to_the_auth_routes(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
+
+    depose = response.headers["set-cookie"]
+    assert depose.startswith("ev_refresh=un-secret-opaque")
+    assert "HttpOnly" in depose
+    assert "SameSite=strict" in depose
+    assert "Path=/api/v1/auth" in depose
+
+
+async def test_login_keeps_the_refresh_secret_out_of_the_response_body(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    response = await client.post("/api/v1/auth/login", json=IDENTIFIANTS)
+
+    assert "un-secret-opaque" not in response.text
+
+
+async def test_refresh_returns_401_when_no_cookie_is_presented(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    response = await client.post("/api/v1/auth/refresh")
+
+    assert response.status_code == 401
+
+
+async def test_refresh_rotates_the_cookie_when_the_session_is_still_valid(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    client.cookies.set("ev_refresh", "un-secret-opaque")
+
+    response = await client.post("/api/v1/auth/refresh")
+
+    assert response.status_code == 200
+    assert "ev_refresh=" in response.headers["set-cookie"]
+
+
+async def test_refresh_clears_the_cookie_when_the_session_is_rejected(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    fake_auth_service[0] = SessionRejectedError("Session révoquée")
+    client.cookies.set("ev_refresh", "un-secret-rejoue")
+
+    response = await client.post("/api/v1/auth/refresh")
+
+    assert response.status_code == 401
+    assert 'ev_refresh=""' in response.headers["set-cookie"]
+    assert "Path=/api/v1/auth" in response.headers["set-cookie"]
+
+
+async def test_logout_answers_204_and_clears_the_cookie(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    client.cookies.set("ev_refresh", "un-secret-opaque")
+
+    response = await client.post("/api/v1/auth/logout")
+
+    assert response.status_code == 204
+    assert 'ev_refresh=""' in response.headers["set-cookie"]
+
+
+async def test_logout_stays_idempotent_without_a_cookie(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    response = await client.post("/api/v1/auth/logout")
+
+    assert response.status_code == 204
+
+
+@pytest.mark.parametrize(
+    "chemin",
+    ["/api/v1/auth/refresh", "/api/v1/auth/logout"],
+    ids=["rotation", "deconnexion"],
+)
+async def test_a_cookie_bearing_route_refuses_a_foreign_origin(
+    fake_auth_service: list[Exception | None], client: AsyncClient, chemin: str
+) -> None:
+    response = await client.post(chemin, headers={"Origin": "https://malveillant.example"})
+
+    assert response.status_code == 403
+
+
+async def test_a_cookie_bearing_route_accepts_a_request_without_origin(
+    fake_auth_service: list[Exception | None], client: AsyncClient
+) -> None:
+    response = await client.post("/api/v1/auth/logout")
+
+    assert response.status_code != 403

apps/backend/tests/api/test_authorization.py

@@ -0,0 +1,113 @@
+from collections.abc import Callable, Iterator
+from uuid import uuid4
+
+import pytest
+from fastapi import FastAPI
+from httpx import AsyncClient
+
+from app.api.deps import AdminDep, get_current_principal, require_role
+from app.core.principal import Principal
+from app.core.roles import AccountKind, Role
+
+CHEMIN_ADMIN = "/api/v1/essai-admin"
+
+
+def principal(role: Role = Role.LECTEUR, *, must_change_password: bool = False) -> Principal:
+    return Principal(
+        id=uuid4(),
+        email=f"{role.value}@enervision.fr",
+        role=role,
+        kind=AccountKind.HUMAIN,
+        must_change_password=must_change_password,
+    )
+
+
+@pytest.fixture
+def route_admin(app: FastAPI) -> None:
+    @app.get(CHEMIN_ADMIN)
+    async def _reserve_aux_admins(acteur: AdminDep) -> dict[str, str]:
+        return {"email": acteur.email}
+
+
+@pytest.fixture
+def connecte(app: FastAPI) -> Iterator[Callable[[Principal], None]]:
+    def installe(acteur: Principal) -> None:
+        app.dependency_overrides[get_current_principal] = lambda: acteur
+
+    yield installe
+    app.dependency_overrides.pop(get_current_principal, None)
+
+
+async def test_me_returns_401_when_no_credentials_are_sent(client: AsyncClient) -> None:
+    response = await client.get("/api/v1/auth/me")
+
+    assert response.status_code == 401
+    assert "Bearer" in response.headers["www-authenticate"]
+
+
+async def test_me_returns_401_when_the_token_is_not_readable(client: AsyncClient) -> None:
+    response = await client.get(
+        "/api/v1/auth/me", headers={"Authorization": "Bearer nimporte.quoi.ici"}
+    )
+
+    assert response.status_code == 401
+    assert 'error="invalid_token"' in response.headers["www-authenticate"]
+
+
+async def test_me_describes_the_connected_account(
+    connecte: Callable[[Principal], None], client: AsyncClient
+) -> None:
+    acteur = principal(Role.OPERATEUR)
+    connecte(acteur)
+
+    response = await client.get("/api/v1/auth/me")
+
+    assert response.status_code == 200
+    assert response.json()["email"] == acteur.email
+
+
+@pytest.mark.parametrize(
+    ("role", "attendu"),
+    [(Role.LECTEUR, 403), (Role.OPERATEUR, 403), (Role.ADMIN, 200)],
+    ids=["lecteur_refuse", "operateur_refuse", "admin_accepte"],
+)
+async def test_an_admin_route_only_answers_to_an_admin(
+    route_admin: None,
+    connecte: Callable[[Principal], None],
+    client: AsyncClient,
+    role: Role,
+    attendu: int,
+) -> None:
+    connecte(principal(role))
+
+    response = await client.get(CHEMIN_ADMIN)
+
+    assert response.status_code == attendu
+
+
+async def test_a_pending_password_change_blocks_every_business_route(
+    route_admin: None, connecte: Callable[[Principal], None], client: AsyncClient
+) -> None:
+    connecte(principal(Role.ADMIN, must_change_password=True))
+
+    response = await client.get(CHEMIN_ADMIN)
+
+    assert response.status_code == 403
+    assert response.json()["detail"] == "password_change_required"
+
+
+async def test_a_pending_password_change_still_allows_reading_ones_own_account(
+    connecte: Callable[[Principal], None], client: AsyncClient
+) -> None:
+    connecte(principal(Role.LECTEUR, must_change_password=True))
+
+    response = await client.get("/api/v1/auth/me")
+
+    assert response.status_code == 200
+    assert response.json()["must_change_password"] is True
+
+
+def test_require_role_builds_one_guard_per_minimum_level() -> None:
+    garde = require_role(Role.OPERATEUR)
+
+    assert callable(garde)

apps/backend/tests/api/test_hardening.py

@@ -0,0 +1,111 @@
+import pytest
+from httpx import ASGITransport, AsyncClient
+from httpx import Response as HttpResponse
+
+from app.main import create_app
+from tests.factories import make_settings
+
+ORIGINE = "https://enervision.fr"
+
+
+async def interroge(
+    settings_overrides: dict[str, object], chemin: str, **kwargs: object
+) -> HttpResponse:
+    application = create_app(make_settings(**settings_overrides))
+    transport = ASGITransport(app=application)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        return await client.get(chemin, **kwargs)  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize(
+    ("entete", "valeur"),
+    [
+        ("x-content-type-options", "nosniff"),
+        ("x-frame-options", "DENY"),
+        ("referrer-policy", "no-referrer"),
+    ],
+    ids=["nosniff", "anti_iframe", "referrer"],
+)
+async def test_every_response_carries_the_security_headers(
+    client: AsyncClient, entete: str, valeur: str
+) -> None:
+    response = await client.get("/api/v1/health/live")
+
+    assert response.headers[entete] == valeur
+
+
+async def test_the_application_never_sets_hsts_itself(client: AsyncClient) -> None:
+    response = await client.get("/api/v1/health/live")
+
+    assert "strict-transport-security" not in response.headers
+
+
+@pytest.mark.parametrize(
+    "env",
+    ["staging", "prod"],
+    ids=["preproduction", "production"],
+)
+async def test_the_documentation_disappears_outside_development(env: str) -> None:
+    surcharges = {"env": env, "cors_origins": ORIGINE}
+
+    for chemin in ("/docs", "/openapi.json"):
+        assert (await interroge(surcharges, chemin)).status_code == 404
+
+
+@pytest.mark.parametrize("env", ["local", "dev"], ids=["local", "developpement"])
+async def test_the_documentation_stays_available_while_developing(env: str) -> None:
+    surcharges = {"env": env, "cors_origins": ORIGINE}
+
+    assert (await interroge(surcharges, "/openapi.json")).status_code == 200
+
+
+async def test_an_explicit_override_can_reopen_the_documentation() -> None:
+    surcharges = {"env": "prod", "cors_origins": ORIGINE, "expose_api_docs": True}
+
+    assert (await interroge(surcharges, "/openapi.json")).status_code == 200
+
+
+async def test_metrics_stay_open_when_no_token_is_configured(client: AsyncClient) -> None:
+    response = await client.get("/metrics")
+
+    assert response.status_code == 200
+
+
+async def test_metrics_demand_the_token_once_one_is_configured() -> None:
+    surcharges = {"metrics_token": "un-jeton-de-supervision-assez-long"}
+
+    assert (await interroge(surcharges, "/metrics")).status_code == 401
+
+
+async def test_metrics_answer_to_the_right_token() -> None:
+    surcharges = {"metrics_token": "un-jeton-de-supervision-assez-long"}
+    entetes = {"Authorization": "Bearer un-jeton-de-supervision-assez-long"}
+
+    response = await interroge(surcharges, "/metrics", headers=entetes)
+
+    assert response.status_code == 200
+
+
+async def test_metrics_refuse_a_token_that_is_almost_right() -> None:
+    surcharges = {"metrics_token": "un-jeton-de-supervision-assez-long"}
+    entetes = {"Authorization": "Bearer un-jeton-de-supervision-assez-lon"}
+
+    response = await interroge(surcharges, "/metrics", headers=entetes)
+
+    assert response.status_code == 401
+
+
+async def test_an_unhandled_error_returns_a_correlation_id_and_no_traceback() -> None:
+    application = create_app(make_settings())
+
+    @application.get("/api/v1/essai-panne")
+    async def _casse() -> None:
+        raise RuntimeError("secret interne de la pile")
+
+    transport = ASGITransport(app=application, raise_app_exceptions=False)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get("/api/v1/essai-panne")
+
+    assert response.status_code == 500
+    assert "secret interne de la pile" not in response.text
+    assert response.json()["correlation"]

apps/backend/tests/api/test_health.py

@@ -17,7 +17,7 @@ async def test_liveness_exposes_service_metadata(client: AsyncClient) -> None:
     }
 
 
-async def test_readiness_reports_the_timescaledb_version(
+async def test_readiness_confirms_the_extension_without_leaking_its_version(
     fake_session: Callable[..., None], client: AsyncClient
 ) -> None:
     fake_session(result="2.22.1")
@@ -28,8 +28,9 @@ async def test_readiness_reports_the_timescaledb_version(
     assert response.json() == {
         "status": "ready",
         "database": "reachable",
-        "timescaledb": "2.22.1",
+        "timescaledb": "loaded",
     }
+    assert "2.22.1" not in response.text
 
 
 async def test_readiness_returns_503_when_the_extension_is_missing(
@@ -59,7 +60,7 @@ async def test_readiness_returns_503_when_database_is_unreachable(
     response = await client.get("/api/v1/health/ready")
 
     assert response.status_code == 503
-    assert response.json()["detail"] == "Base de donnees injoignable"
+    assert response.json()["detail"] == "Base de données injoignable"
 
 
 @pytest.mark.parametrize("path", ["/openapi.json", "/metrics"])
@@ -75,4 +76,4 @@ async def test_readiness_reaches_the_real_database(client: AsyncClient) -> None:
     body = response.json()
     assert body["status"] == "ready"
     assert body["database"] == "reachable"
-    assert body["timescaledb"]
+    assert body["timescaledb"] == "loaded"

apps/backend/tests/api/test_parcours_authentification.py

@@ -0,0 +1,162 @@
+# Parcours complet contre la vraie base, sans serveur ni port ouvert. C'est ce fichier qui
+# prouve que le câblage tient : la connexion, la rotation, la détection de réutilisation et la
+# révocation immédiate passent par les vrais dépôts, les vraies transactions et les vrais
+# déclencheurs PostgreSQL.
+
+import uuid
+from collections.abc import AsyncIterator
+
+import pytest
+from fastapi import FastAPI
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import text
+
+from app.core.hashing import build_hasher
+from app.core.roles import Role
+from app.db.session import get_session_factory
+from app.repositories.user import UserRepository
+
+pytestmark = pytest.mark.integration
+
+MOT_DE_PASSE = "un-mot-de-passe-de-recette"
+
+
+@pytest.fixture
+async def compte_operateur() -> AsyncIterator[str]:
+    email = f"parcours-{uuid.uuid4().hex[:12]}@enervision.fr"
+    hacheur = build_hasher(time_cost=1, memory_cost_kib=8192, parallelism=1, max_concurrency=2)
+    empreinte = await hacheur.hash(MOT_DE_PASSE)
+
+    async with get_session_factory()() as session:
+        await UserRepository(session).create(
+            email=email, password_hash=empreinte, role=Role.OPERATEUR
+        )
+        await session.commit()
+
+    yield email
+
+    async with get_session_factory()() as session:
+        await session.execute(text("delete from app_user where email = :e"), {"e": email})
+        await session.commit()
+
+
+@pytest.fixture
+async def navigateur(app: FastAPI) -> AsyncIterator[AsyncClient]:
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        yield client
+
+
+async def connecte(navigateur: AsyncClient, email: str) -> dict[str, str]:
+    reponse = await navigateur.post(
+        "/api/v1/auth/login", json={"email": email, "password": MOT_DE_PASSE}
+    )
+    assert reponse.status_code == 200, reponse.text
+    return {"Authorization": f"Bearer {reponse.json()['access_token']}"}
+
+
+async def test_a_full_session_runs_from_login_to_logout(
+    compte_operateur: str, navigateur: AsyncClient
+) -> None:
+    entetes = await connecte(navigateur, compte_operateur)
+
+    identite = await navigateur.get("/api/v1/auth/me", headers=entetes)
+    rotation = await navigateur.post("/api/v1/auth/refresh")
+    deconnexion = await navigateur.post("/api/v1/auth/logout")
+
+    assert identite.status_code == 200
+    assert identite.json()["role"] == "operateur"
+    assert rotation.status_code == 200
+    assert deconnexion.status_code == 204
+
+
+async def test_replaying_a_rotated_cookie_kills_the_whole_family(
+    compte_operateur: str, navigateur: AsyncClient
+) -> None:
+    await connecte(navigateur, compte_operateur)
+    vole = navigateur.cookies["ev_refresh"]
+    premiere_rotation = await navigateur.post("/api/v1/auth/refresh")
+    vivant = navigateur.cookies["ev_refresh"]
+
+    navigateur.cookies.set("ev_refresh", vole)
+    rejeu = await navigateur.post("/api/v1/auth/refresh")
+
+    navigateur.cookies.set("ev_refresh", vivant)
+    apres = await navigateur.post("/api/v1/auth/refresh")
+
+    assert premiere_rotation.status_code == 200
+    assert rejeu.status_code == 401
+    assert apres.status_code == 401, "la session vivante doit tomber avec sa famille"
+
+
+async def test_the_reuse_leaves_a_trace_in_the_append_only_audit_log(
+    compte_operateur: str, navigateur: AsyncClient
+) -> None:
+    await connecte(navigateur, compte_operateur)
+    vole = navigateur.cookies["ev_refresh"]
+    await navigateur.post("/api/v1/auth/refresh")
+
+    navigateur.cookies.set("ev_refresh", vole)
+    await navigateur.post("/api/v1/auth/refresh")
+
+    async with get_session_factory()() as session:
+        traces = await session.scalar(
+            text("select count(*) from audit_log where action = 'auth.refresh_reuse_detected'")
+        )
+    assert traces is not None
+    assert traces >= 1
+
+
+async def test_disabling_an_account_invalidates_its_access_token_at_once(
+    compte_operateur: str, navigateur: AsyncClient
+) -> None:
+    entetes = await connecte(navigateur, compte_operateur)
+    avant = await navigateur.get("/api/v1/auth/me", headers=entetes)
+
+    async with get_session_factory()() as session:
+        depot = UserRepository(session)
+        compte = await depot.get_by_email(compte_operateur)
+        assert compte is not None
+        await depot.set_active(compte.id, is_active=False)
+        await session.commit()
+
+    apres = await navigateur.get("/api/v1/auth/me", headers=entetes)
+
+    assert avant.status_code == 200
+    assert apres.status_code == 401, "la révocation doit être immédiate, pas dans 15 minutes"
+
+
+async def test_changing_a_role_invalidates_the_token_that_still_carries_the_old_one(
+    compte_operateur: str, navigateur: AsyncClient
+) -> None:
+    entetes = await connecte(navigateur, compte_operateur)
+
+    async with get_session_factory()() as session:
+        depot = UserRepository(session)
+        compte = await depot.get_by_email(compte_operateur)
+        assert compte is not None
+        await depot.set_role(compte.id, Role.LECTEUR)
+        await session.commit()
+
+    apres = await navigateur.get("/api/v1/auth/me", headers=entetes)
+
+    assert apres.status_code == 401
+    assert "token_stale" in apres.headers["www-authenticate"]
+
+
+async def test_a_failed_login_is_recorded_even_for_an_unknown_address(
+    navigateur: AsyncClient,
+) -> None:
+    inconnu = f"inconnu-{uuid.uuid4().hex[:12]}@enervision.fr"
+
+    reponse = await navigateur.post(
+        "/api/v1/auth/login", json={"email": inconnu, "password": "peu-importe-ici"}
+    )
+
+    async with get_session_factory()() as session:
+        tentatives = await session.scalar(
+            text("select count(*) from login_attempt where email_tried = :e"), {"e": inconnu}
+        )
+    assert reponse.status_code == 401
+    assert reponse.json() == {"detail": "Identifiants invalides"}
+    assert tentatives == 1, "sans cette ligne, le 429 deviendrait un oracle d'existence"

apps/backend/tests/api/test_route_protection.py

@@ -0,0 +1,76 @@
+# Ce test est le garde-fou de l'autorisation : rendre une route publique oblige à modifier
+# `ROUTES_PUBLIQUES` ci-dessous, ce qui apparaît en clair dans la diff d'une pull request et
+# demande une justification au relecteur.
+# Pourquoi : il interroge réellement chaque route sans jeton au lieu d'inspecter l'arbre de
+# dépendances. L'arbre n'est accessible que par l'API privée de FastAPI, et surtout une route
+# peut porter la bonne dépendance tout en répondant quand même.
+
+from typing import Any
+
+import pytest
+from fastapi import FastAPI
+from httpx import AsyncClient
+
+ROUTES_PUBLIQUES = frozenset(
+    {
+        ("GET", "/api/v1/health/live"),
+        ("GET", "/api/v1/health/ready"),
+        ("POST", "/api/v1/auth/login"),
+        # Sans cookie, la déconnexion ne fait rien et répond 204 : elle est idempotente.
+        ("POST", "/api/v1/auth/logout"),
+        ("GET", "/metrics"),
+    }
+)
+
+VALEURS_DE_SUBSTITUTION = "00000000-0000-0000-0000-000000000000"
+STATUTS_DE_REFUS = {401, 403}
+
+
+def routes_declarees(app: FastAPI) -> list[tuple[str, str]]:
+    schema: dict[str, Any] = app.openapi()
+    return [
+        (methode.upper(), chemin)
+        for chemin, operations in schema["paths"].items()
+        for methode in operations
+        if methode.upper() in {"GET", "POST", "PATCH", "PUT", "DELETE"}
+    ]
+
+
+def routes_protegees(app: FastAPI) -> list[tuple[str, str]]:
+    return [route for route in routes_declarees(app) if route not in ROUTES_PUBLIQUES]
+
+
+def test_the_public_allow_list_has_no_stale_entry(app: FastAPI) -> None:
+    declarees = set(routes_declarees(app)) | {("GET", "/metrics")}
+
+    inconnues = ROUTES_PUBLIQUES - declarees
+
+    assert inconnues == set()
+
+
+async def test_every_route_rejects_an_anonymous_caller_unless_explicitly_public(
+    app: FastAPI, client: AsyncClient
+) -> None:
+    ouvertes: list[tuple[str, str, int]] = []
+
+    for methode, chemin in routes_protegees(app):
+        concret = chemin.replace("{user_id}", VALEURS_DE_SUBSTITUTION)
+        response = await client.request(methode, concret, json={})
+        if response.status_code not in STATUTS_DE_REFUS:
+            ouvertes.append((methode, chemin, response.status_code))
+
+    assert ouvertes == []
+
+
+async def test_the_declared_routes_are_actually_reachable(app: FastAPI) -> None:
+    assert ("POST", "/api/v1/auth/login") in routes_declarees(app)
+    assert ("GET", "/api/v1/auth/me") in routes_declarees(app)
+
+
+@pytest.mark.parametrize(
+    "chemin",
+    ["/api/v1/health/live", "/api/v1/health/ready"],
+    ids=["sonde_de_vie", "sonde_de_disponibilite"],
+)
+def test_the_health_probes_stay_public(app: FastAPI, chemin: str) -> None:
+    assert ("GET", chemin) in ROUTES_PUBLIQUES

apps/backend/tests/api/test_users.py

@@ -0,0 +1,208 @@
+from collections.abc import Callable, Iterator
+from datetime import UTC, datetime
+from uuid import UUID, uuid4
+
+import pytest
+from fastapi import FastAPI
+from httpx import AsyncClient
+
+from app.api.deps import get_current_principal, get_user_service
+from app.core.principal import Principal
+from app.core.roles import AccountKind, Role
+from app.services.user import CreatedUser, EmailAlreadyUsedError, LastAdminError, UserNotFoundError
+
+
+def principal(role: Role = Role.ADMIN) -> Principal:
+    return Principal(
+        id=uuid4(),
+        email=f"{role.value}@enervision.fr",
+        role=role,
+        kind=AccountKind.HUMAIN,
+        must_change_password=False,
+    )
+
+
+class FauxCompte:
+    def __init__(self, role: Role = Role.LECTEUR) -> None:
+        self.id = uuid4()
+        self.email = "cible@enervision.fr"
+        self.role = role.value
+        self.kind = "human"
+        self.is_active = True
+        self.must_change_password = True
+        self.full_name = None
+        self.last_login_at: datetime | None = None
+        self.created_at = datetime.now(UTC)
+
+
+class FauxService:
+    def __init__(self, erreur: Exception | None = None) -> None:
+        self._erreur = erreur
+        self.compte = FauxCompte()
+
+    def _leve(self) -> None:
+        if self._erreur is not None:
+            raise self._erreur
+
+    async def list_all(self) -> list[FauxCompte]:
+        return [self.compte]
+
+    async def create(self, **_: object) -> CreatedUser:
+        self._leve()
+        return CreatedUser(user=self.compte, temporary_password="mot-de-passe-provisoire")  # type: ignore[arg-type]
+
+    async def change_role(self, **_: object) -> FauxCompte:
+        self._leve()
+        return self.compte
+
+    async def set_active(self, **_: object) -> FauxCompte:
+        self._leve()
+        return self.compte
+
+    async def reset_password(self, **_: object) -> CreatedUser:
+        self._leve()
+        return CreatedUser(user=self.compte, temporary_password="mot-de-passe-provisoire")  # type: ignore[arg-type]
+
+
+@pytest.fixture
+def administre(app: FastAPI) -> Iterator[Callable[[Exception | None], FauxService]]:
+    services: list[FauxService] = []
+
+    def installe(erreur: Exception | None = None) -> FauxService:
+        service = FauxService(erreur)
+        services.append(service)
+        app.dependency_overrides[get_user_service] = lambda: service
+        app.dependency_overrides[get_current_principal] = lambda: principal()
+        return service
+
+    yield installe
+    app.dependency_overrides.pop(get_user_service, None)
+    app.dependency_overrides.pop(get_current_principal, None)
+
+
+@pytest.fixture
+def lecteur_connecte(app: FastAPI) -> Iterator[None]:
+    app.dependency_overrides[get_current_principal] = lambda: principal(Role.LECTEUR)
+    yield
+    app.dependency_overrides.pop(get_current_principal, None)
+
+
+async def test_list_users_returns_the_accounts_without_their_digest(
+    administre: Callable[..., FauxService], client: AsyncClient
+) -> None:
+    administre()
+
+    response = await client.get("/api/v1/users")
+
+    assert response.status_code == 200
+    corps = response.json()
+    assert "password_hash" not in corps[0]
+    assert corps[0]["email"] == "cible@enervision.fr"
+
+
+async def test_create_user_returns_the_temporary_password_once(
+    administre: Callable[..., FauxService], client: AsyncClient
+) -> None:
+    administre()
+
+    response = await client.post(
+        "/api/v1/users", json={"email": "nouveau@enervision.fr", "role": "operateur"}
+    )
+
+    assert response.status_code == 201
+    assert response.json()["temporary_password"] == "mot-de-passe-provisoire"
+    assert response.headers["cache-control"] == "no-store"
+
+
+async def test_create_user_refuses_an_address_already_taken(
+    administre: Callable[..., FauxService], client: AsyncClient
+) -> None:
+    administre(EmailAlreadyUsedError("cible@enervision.fr"))
+
+    response = await client.post(
+        "/api/v1/users", json={"email": "cible@enervision.fr", "role": "lecteur"}
+    )
+
+    assert response.status_code == 409
+
+
+async def test_create_user_never_accepts_a_caller_chosen_digest(
+    administre: Callable[..., FauxService], client: AsyncClient
+) -> None:
+    administre()
+
+    response = await client.post(
+        "/api/v1/users",
+        json={
+            "email": "nouveau@enervision.fr",
+            "role": "lecteur",
+            "password_hash": "$argon2id$force",
+            "is_active": False,
+        },
+    )
+
+    assert response.status_code == 201
+
+
+async def test_update_user_refuses_to_strand_the_last_administrator(
+    administre: Callable[..., FauxService], client: AsyncClient
+) -> None:
+    administre(LastAdminError("x"))
+
+    response = await client.patch(f"/api/v1/users/{uuid4()}", json={"is_active": False})
+
+    assert response.status_code == 409
+
+
+async def test_update_user_returns_404_for_an_unknown_account(
+    administre: Callable[..., FauxService], client: AsyncClient
+) -> None:
+    administre(UserNotFoundError("x"))
+
+    response = await client.patch(f"/api/v1/users/{uuid4()}", json={"role": "admin"})
+
+    assert response.status_code == 404
+
+
+async def test_update_user_refuses_an_empty_body(
+    administre: Callable[..., FauxService], client: AsyncClient
+) -> None:
+    administre()
+
+    response = await client.patch(f"/api/v1/users/{uuid4()}", json={})
+
+    assert response.status_code == 400
+
+
+async def test_reset_password_returns_a_new_temporary_password(
+    administre: Callable[..., FauxService], client: AsyncClient
+) -> None:
+    administre()
+
+    response = await client.post(f"/api/v1/users/{uuid4()}/password-reset")
+
+    assert response.status_code == 200
+    assert response.json()["temporary_password"] == "mot-de-passe-provisoire"
+    assert response.headers["cache-control"] == "no-store"
+
+
+@pytest.mark.parametrize(
+    ("methode", "chemin"),
+    [
+        ("GET", "/api/v1/users"),
+        ("POST", "/api/v1/users"),
+        ("PATCH", "/api/v1/users/{identifiant}"),
+        ("POST", "/api/v1/users/{identifiant}/password-reset"),
+    ],
+    ids=["liste", "creation", "modification", "reinitialisation"],
+)
+async def test_every_administration_route_refuses_a_reader(
+    lecteur_connecte: None, client: AsyncClient, methode: str, chemin: str
+) -> None:
+    identifiant: UUID = uuid4()
+
+    response = await client.request(
+        methode, chemin.format(identifiant=identifiant), json={"role": "admin"}
+    )
+
+    assert response.status_code == 403

apps/backend/tests/conftest.py

@@ -12,8 +12,8 @@ from app.main import create_app
 from tests.factories import FakeSession
 
 
-# Piege : les variables d'environnement priment sur apps/backend/.env. Celles qu'on ne
-# pose pas ici, c'est le .env du poste qui les decide, et les assertions avec.
+# Piège : les variables d'environnement priment sur apps/backend/.env. Celles qu'on ne
+# pose pas ici, c'est le .env du poste qui les décide, et les assertions avec.
 @pytest.fixture(autouse=True, scope="session")
 def environment() -> Iterator[None]:
     os.environ.update(
@@ -22,7 +22,7 @@ def environment() -> Iterator[None]:
             "APP_DEBUG": "false",
             "APP_LOG_LEVEL": "WARNING",
             "APP_CORS_ORIGINS": "",
-            "APP_SECRET_KEY": "secret-de-test",
+            "APP_SECRET_KEY": "secret-de-test-assez-long-pour-le-validateur",
         }
     )
     os.environ.setdefault(
@@ -33,8 +33,8 @@ def environment() -> Iterator[None]:
     get_settings.cache_clear()
 
 
-# Piege : get_engine est lru_cache et pytest-asyncio ouvre une boucle par test. Sans ce
-# recyclage, le 2e test touchant vraiment la base heriterait d une boucle morte.
+# Piège : get_engine est lru_cache et pytest-asyncio ouvre une boucle par test. Sans ce
+# recyclage, le 2e test touchant vraiment la base hériterait d'une boucle morte.
 @pytest.fixture(autouse=True)
 async def engine_per_test() -> AsyncIterator[None]:
     yield
@@ -67,7 +67,7 @@ def fake_session(app: FastAPI) -> Callable[..., None]:
     return install
 
 
-# Contrainte : ouvre une vraie connexion, donc reservee aux tests `integration`.
+# Contrainte : ouvre une vraie connexion, donc réservée aux tests `integration`.
 @pytest.fixture
 async def session() -> AsyncIterator[AsyncSession]:
     async with get_session_factory()() as async_session:

apps/backend/tests/core/test_config.py

@@ -0,0 +1,78 @@
+import pytest
+from pydantic import ValidationError
+
+from tests.factories import make_settings
+
+SECRET_VALIDE = "un-secret-de-test-de-plus-de-trente-deux-caracteres"
+
+
+@pytest.mark.parametrize(
+    "surcharges",
+    [
+        {"secret_key": "trop-court"},
+        {"secret_key": "change_me"},
+        {"env": "prod", "debug": True, "cors_origins": "https://enervision.fr"},
+        {"cors_origins": "*"},
+        {"env": "prod", "cors_origins": ""},
+        {"cookie_samesite": "none", "cookie_secure": False},
+    ],
+    ids=[
+        "secret_trop_court",
+        "secret_sentinelle",
+        "debug_en_production",
+        "joker_dans_les_origines",
+        "origines_vides_hors_local",
+        "samesite_none_sans_secure",
+    ],
+)
+def test_settings_refuses_to_build_when_the_configuration_is_unsafe(
+    surcharges: dict[str, object],
+) -> None:
+    with pytest.raises(ValidationError):
+        make_settings(**surcharges)
+
+
+def test_settings_accepts_debug_in_local_environment() -> None:
+    settings = make_settings(env="local", debug=True)
+
+    assert settings.debug is True
+
+
+@pytest.mark.parametrize(
+    ("env", "attendu"),
+    [("local", False), ("dev", True), ("staging", True), ("prod", True)],
+    ids=["local", "dev", "staging", "production"],
+)
+def test_cookies_are_secure_follows_the_environment(env: str, attendu: bool) -> None:
+    settings = make_settings(env=env, cors_origins="https://enervision.fr")
+
+    assert settings.cookies_are_secure is attendu
+
+
+def test_cookies_are_secure_honours_an_explicit_override() -> None:
+    settings = make_settings(env="prod", cors_origins="https://enervision.fr", cookie_secure=False)
+
+    assert settings.cookies_are_secure is False
+
+
+@pytest.mark.parametrize(
+    ("env", "attendu"),
+    [("local", True), ("dev", True), ("staging", False), ("prod", False)],
+    ids=["local", "dev", "staging", "production"],
+)
+def test_api_docs_are_exposed_closes_staging_and_production(env: str, attendu: bool) -> None:
+    settings = make_settings(env=env, cors_origins="https://enervision.fr")
+
+    assert settings.api_docs_are_exposed is attendu
+
+
+def test_api_docs_are_exposed_honours_an_explicit_override() -> None:
+    settings = make_settings(env="prod", cors_origins="https://enervision.fr", expose_api_docs=True)
+
+    assert settings.api_docs_are_exposed is True
+
+
+def test_allowed_origins_splits_and_trims_the_list() -> None:
+    settings = make_settings(cors_origins=" http://localhost:4200 , https://enervision.fr ")
+
+    assert settings.allowed_origins == ["http://localhost:4200", "https://enervision.fr"]

apps/backend/tests/core/test_cookies.py

@@ -0,0 +1,58 @@
+from app.core.cookies import RefreshCookie, cookie_name
+from tests.factories import make_settings
+
+
+def test_build_marks_the_cookie_http_only_and_scopes_it_to_the_auth_routes() -> None:
+    settings = make_settings(env="local")
+
+    cookie = RefreshCookie.build(settings, "un-secret-opaque")
+
+    assert cookie.httponly is True
+    assert cookie.samesite == "strict"
+    assert cookie.path == "/api/v1/auth"
+    assert cookie.max_age == settings.refresh_token_ttl_seconds
+
+
+def test_build_prefixes_and_secures_the_cookie_outside_local() -> None:
+    settings = make_settings(env="prod", cors_origins="https://enervision.fr")
+
+    cookie = RefreshCookie.build(settings, "un-secret-opaque")
+
+    assert cookie.secure is True
+    assert cookie.key.startswith("__Secure-")
+
+
+def test_build_leaves_the_cookie_unprefixed_in_local() -> None:
+    settings = make_settings(env="local")
+
+    cookie = RefreshCookie.build(settings, "un-secret-opaque")
+
+    assert cookie.key == "ev_refresh"
+
+
+def test_expired_reuses_the_exact_name_and_path_of_the_posted_cookie() -> None:
+    settings = make_settings(env="prod", cors_origins="https://enervision.fr")
+
+    pose = RefreshCookie.build(settings, "un-secret-opaque")
+    suppression = RefreshCookie.expired(settings)
+
+    assert suppression.key == pose.key
+    assert suppression.path == pose.path
+    assert suppression.secure == pose.secure
+    assert suppression.samesite == pose.samesite
+    assert suppression.max_age == 0
+    assert suppression.value == ""
+
+
+def test_as_kwargs_matches_the_starlette_set_cookie_signature() -> None:
+    settings = make_settings(env="local")
+
+    arguments = RefreshCookie.build(settings, "un-secret-opaque").as_kwargs()
+
+    assert set(arguments) == {"key", "value", "max_age", "path", "secure", "httponly", "samesite"}
+
+
+def test_cookie_name_follows_the_configured_name() -> None:
+    settings = make_settings(env="local", refresh_cookie_name="autre_nom")
+
+    assert cookie_name(settings) == "autre_nom"

apps/backend/tests/core/test_hashing.py

@@ -0,0 +1,61 @@
+from app.core.hashing import Argon2Hasher, build_hasher
+
+MOT_DE_PASSE = "un-mot-de-passe-de-test-assez-long"
+
+
+def fabrique(time_cost: int = 1, max_concurrency: int = 2) -> Argon2Hasher:
+    return build_hasher(
+        time_cost=time_cost,
+        memory_cost_kib=8192,
+        parallelism=1,
+        max_concurrency=max_concurrency,
+    )
+
+
+async def test_hash_produces_a_distinct_digest_for_the_same_password() -> None:
+    hacheur = fabrique()
+
+    premier = await hacheur.hash(MOT_DE_PASSE)
+    second = await hacheur.hash(MOT_DE_PASSE)
+
+    assert premier != second
+    assert premier.startswith("$argon2id$")
+
+
+async def test_verify_accepts_the_right_password_and_rejects_the_others() -> None:
+    hacheur = fabrique()
+
+    empreinte = await hacheur.hash(MOT_DE_PASSE)
+
+    assert await hacheur.verify(empreinte, MOT_DE_PASSE) is True
+    assert await hacheur.verify(empreinte, "un-autre-mot-de-passe") is False
+
+
+async def test_verify_returns_false_when_the_stored_digest_is_malformed() -> None:
+    hacheur = fabrique()
+
+    accorde = await hacheur.verify("pas-une-empreinte-argon2", MOT_DE_PASSE)
+
+    assert accorde is False
+
+
+async def test_needs_rehash_is_true_when_the_parameters_changed() -> None:
+    ancien = fabrique(time_cost=1)
+    recent = fabrique(time_cost=3)
+
+    empreinte = await ancien.hash(MOT_DE_PASSE)
+
+    assert ancien.needs_rehash(empreinte) is False
+    assert recent.needs_rehash(empreinte) is True
+
+
+def test_needs_rehash_is_true_when_the_stored_digest_is_malformed() -> None:
+    hacheur = fabrique()
+
+    assert hacheur.needs_rehash("pas-une-empreinte-argon2") is True
+
+
+async def test_verify_dummy_completes_without_revealing_anything() -> None:
+    hacheur = fabrique()
+
+    await hacheur.verify_dummy()

apps/backend/tests/core/test_logging.py

@@ -0,0 +1,72 @@
+import logging
+
+import pytest
+
+from app.core.logging import CAVIARDAGE, RedactingFilter, redact
+
+
+@pytest.mark.parametrize(
+    "message",
+    [
+        "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.charge-utile-assez-longue.signature",
+        "jeton brut eyJhbGciOiJIUzI1NiJ9abcdefghijklmnopqrstuvwxyz",
+        "INSERT ... ('$argon2id$v=19$m=19456,t=2,p=1$sel-en-clair$empreinte-en-clair')",
+        '{"password": "le-mot-de-passe-du-client"}',
+        "current_password=le-mot-de-passe",
+        "Cookie: ev_refresh=abcdefghijklmnopqrstuvwxyz0123456789",
+    ],
+    ids=[
+        "en_tete_bearer",
+        "jeton_jwt_nu",
+        "empreinte_argon2",
+        "mot_de_passe_json",
+        "mot_de_passe_en_paire",
+        "cookie_de_rafraichissement",
+    ],
+)
+def test_redact_removes_every_known_secret_shape(message: str) -> None:
+    expurge = redact(message)
+
+    assert CAVIARDAGE in expurge
+    for suspect in ("le-mot-de-passe", "empreinte-en-clair", "abcdefghijklmnopqrstuvwxyz"):
+        assert suspect not in expurge
+
+
+def test_redact_leaves_an_innocent_message_untouched() -> None:
+    message = "auth.login.success user_id=3f2a ip=203.0.113.10"
+
+    assert redact(message) == message
+
+
+def test_the_filter_rewrites_the_record_before_it_reaches_the_handler() -> None:
+    enregistrement = logging.LogRecord(
+        name="app",
+        level=logging.INFO,
+        pathname=__file__,
+        lineno=1,
+        msg='requete {"password": "%s"}',
+        args=("secret-du-client",),
+        exc_info=None,
+    )
+
+    conserve = RedactingFilter().filter(enregistrement)
+
+    assert conserve is True
+    assert "secret-du-client" not in enregistrement.getMessage()
+
+
+def test_the_filter_keeps_a_record_that_holds_no_secret() -> None:
+    enregistrement = logging.LogRecord(
+        name="app",
+        level=logging.INFO,
+        pathname=__file__,
+        lineno=1,
+        msg="requete %s",
+        args=("/api/v1/health/live",),
+        exc_info=None,
+    )
+
+    conserve = RedactingFilter().filter(enregistrement)
+
+    assert conserve is True
+    assert enregistrement.getMessage() == "requete /api/v1/health/live"

apps/backend/tests/core/test_roles.py

@@ -0,0 +1,40 @@
+import pytest
+
+from app.core.roles import Role, has_at_least
+
+
+@pytest.mark.parametrize(
+    ("actual", "required", "expected"),
+    [
+        (Role.LECTEUR, Role.LECTEUR, True),
+        (Role.LECTEUR, Role.OPERATEUR, False),
+        (Role.LECTEUR, Role.ADMIN, False),
+        (Role.OPERATEUR, Role.LECTEUR, True),
+        (Role.OPERATEUR, Role.OPERATEUR, True),
+        (Role.OPERATEUR, Role.ADMIN, False),
+        (Role.ADMIN, Role.LECTEUR, True),
+        (Role.ADMIN, Role.OPERATEUR, True),
+        (Role.ADMIN, Role.ADMIN, True),
+    ],
+    ids=[
+        "lecteur_sur_lecteur",
+        "lecteur_sur_operateur",
+        "lecteur_sur_admin",
+        "operateur_sur_lecteur",
+        "operateur_sur_operateur",
+        "operateur_sur_admin",
+        "admin_sur_lecteur",
+        "admin_sur_operateur",
+        "admin_sur_admin",
+    ],
+)
+def test_has_at_least_orders_the_three_roles(actual: Role, required: Role, expected: bool) -> None:
+    accorde = has_at_least(actual, required)
+
+    assert accorde is expected
+
+
+def test_role_values_stay_ascii_for_the_wire_format() -> None:
+    valeurs = [role.value for role in Role]
+
+    assert all(valeur.isascii() for valeur in valeurs)

apps/backend/tests/core/test_security.py

@@ -0,0 +1,194 @@
+import base64
+import json
+from datetime import UTC, datetime, timedelta
+from uuid import uuid4
+
+import jwt
+import pytest
+
+from app.core.security import (
+    AccessClaims,
+    TokenExpiredError,
+    TokenInvalidError,
+    TokenPolicy,
+    decode_access_token,
+    encode_access_token,
+    fingerprint_refresh,
+    generate_refresh_secret,
+)
+
+POLITIQUE = TokenPolicy(
+    secret="un-secret-de-test-de-plus-de-trente-deux-caracteres",
+    issuer="enervision-api",
+    audience="enervision-web",
+    access_ttl=timedelta(minutes=15),
+)
+
+
+def emets(**surcharges: object) -> str:
+    charge = {
+        "iss": POLITIQUE.issuer,
+        "aud": POLITIQUE.audience,
+        "sub": str(uuid4()),
+        "iat": datetime.now(UTC),
+        "exp": datetime.now(UTC) + timedelta(minutes=15),
+        "jti": str(uuid4()),
+        "typ": "access",
+        "role": "lecteur",
+        "kind": "human",
+    }
+    charge.update(surcharges)
+    return jwt.encode(charge, POLITIQUE.secret, algorithm="HS256")
+
+
+def test_decode_access_token_returns_the_claims_when_the_token_is_valid() -> None:
+    sujet = uuid4()
+
+    jeton = encode_access_token(POLITIQUE, subject=sujet, role="operateur", kind="human")
+    claims = decode_access_token(POLITIQUE, jeton)
+
+    assert isinstance(claims, AccessClaims)
+    assert claims.subject == sujet
+    assert claims.role == "operateur"
+    assert claims.kind == "human"
+
+
+def test_decode_access_token_raises_expired_when_the_lifetime_has_passed() -> None:
+    passe = datetime.now(UTC) - timedelta(hours=2)
+
+    jeton = encode_access_token(POLITIQUE, subject=uuid4(), role="lecteur", kind="human", now=passe)
+
+    with pytest.raises(TokenExpiredError):
+        decode_access_token(POLITIQUE, jeton)
+
+
+def test_decode_access_token_raises_invalid_when_the_signature_was_forged() -> None:
+    autre = TokenPolicy(
+        secret="un-autre-secret-tout-aussi-long-que-le-premier",
+        issuer=POLITIQUE.issuer,
+        audience=POLITIQUE.audience,
+        access_ttl=POLITIQUE.access_ttl,
+    )
+
+    jeton = encode_access_token(autre, subject=uuid4(), role="lecteur", kind="human")
+
+    with pytest.raises(TokenInvalidError):
+        decode_access_token(POLITIQUE, jeton)
+
+
+@pytest.mark.parametrize(
+    "surcharges",
+    [
+        {"aud": "un-autre-public"},
+        {"iss": "un-autre-emetteur"},
+        {"typ": "refresh"},
+    ],
+    ids=["audience_invalide", "emetteur_invalide", "jeton_de_rafraichissement"],
+)
+def test_decode_access_token_raises_invalid_when_a_claim_is_wrong(
+    surcharges: dict[str, object],
+) -> None:
+    jeton = emets(**surcharges)
+
+    with pytest.raises(TokenInvalidError):
+        decode_access_token(POLITIQUE, jeton)
+
+
+@pytest.mark.parametrize(
+    "claim",
+    ["jti", "typ", "role", "kind"],
+    ids=["identifiant", "type", "role", "nature_du_compte"],
+)
+def test_decode_access_token_raises_invalid_when_a_required_claim_is_missing(claim: str) -> None:
+    charge = {
+        "iss": POLITIQUE.issuer,
+        "aud": POLITIQUE.audience,
+        "sub": str(uuid4()),
+        "iat": datetime.now(UTC),
+        "exp": datetime.now(UTC) + timedelta(minutes=15),
+        "jti": str(uuid4()),
+        "typ": "access",
+        "role": "lecteur",
+        "kind": "human",
+    }
+    del charge[claim]
+
+    jeton = jwt.encode(charge, POLITIQUE.secret, algorithm="HS256")
+
+    with pytest.raises(TokenInvalidError):
+        decode_access_token(POLITIQUE, jeton)
+
+
+def test_decode_access_token_rejects_a_token_forged_with_the_none_algorithm() -> None:
+    def encode(donnees: dict[str, object]) -> str:
+        brut = json.dumps(donnees, separators=(",", ":")).encode()
+        return base64.urlsafe_b64encode(brut).rstrip(b"=").decode()
+
+    entete = encode({"alg": "none", "typ": "JWT"})
+    charge = encode(
+        {
+            "iss": POLITIQUE.issuer,
+            "aud": POLITIQUE.audience,
+            "sub": str(uuid4()),
+            "iat": int(datetime.now(UTC).timestamp()),
+            "exp": int((datetime.now(UTC) + timedelta(minutes=15)).timestamp()),
+            "jti": str(uuid4()),
+            "typ": "access",
+            "role": "admin",
+            "kind": "human",
+        }
+    )
+
+    with pytest.raises(TokenInvalidError):
+        decode_access_token(POLITIQUE, f"{entete}.{charge}.")
+
+
+def test_decode_access_token_rejects_a_token_signed_with_another_algorithm() -> None:
+    charge = {
+        "iss": POLITIQUE.issuer,
+        "aud": POLITIQUE.audience,
+        "sub": str(uuid4()),
+        "iat": datetime.now(UTC),
+        "exp": datetime.now(UTC) + timedelta(minutes=15),
+        "jti": str(uuid4()),
+        "typ": "access",
+        "role": "admin",
+        "kind": "human",
+    }
+
+    jeton = jwt.encode(charge, POLITIQUE.secret * 2, algorithm="HS512")
+
+    with pytest.raises(TokenInvalidError):
+        decode_access_token(POLITIQUE, jeton)
+
+
+@pytest.mark.parametrize(
+    "surcharges",
+    [{"sub": "pas-un-uuid"}, {"jti": "pas-un-uuid"}],
+    ids=["sujet_illisible", "identifiant_illisible"],
+)
+def test_decode_access_token_raises_invalid_when_an_identifier_is_not_a_uuid(
+    surcharges: dict[str, object],
+) -> None:
+    jeton = emets(**surcharges)
+
+    with pytest.raises(TokenInvalidError):
+        decode_access_token(POLITIQUE, jeton)
+
+
+def test_generate_refresh_secret_returns_distinct_url_safe_values() -> None:
+    secrets_generes = {generate_refresh_secret() for _ in range(100)}
+
+    assert len(secrets_generes) == 100
+    assert all(len(valeur) >= 43 for valeur in secrets_generes)
+
+
+def test_fingerprint_refresh_is_stable_and_distinguishes_two_secrets() -> None:
+    premier = generate_refresh_secret()
+    second = generate_refresh_secret()
+
+    empreinte = fingerprint_refresh(premier)
+
+    assert len(empreinte) == 32
+    assert empreinte == fingerprint_refresh(premier)
+    assert empreinte != fingerprint_refresh(second)

apps/backend/tests/factories.py

@@ -7,7 +7,7 @@ SETTINGS_DE_TEST: dict[str, Any] = {
     "debug": False,
     "log_level": "WARNING",
     "cors_origins": "",
-    "secret_key": "secret-de-test",
+    "secret_key": "secret-de-test-assez-long-pour-le-validateur",
     "database_url": "postgresql+asyncpg://enervision:change_me@localhost:5433/enervision_test",
 }
 
@@ -31,7 +31,7 @@ class FakeSession:
         return self._result
 
 
-# Piege : les arguments nommes priment sur l'environnement et sur .env, contrairement
-# aux variables posees par la fixture `environment`, qui restent surchargeables.
+# Piège : les arguments nommés priment sur l'environnement et sur .env, contrairement
+# aux variables posées par la fixture `environment`, qui restent surchargeables.
 def make_settings(**overrides: Any) -> Settings:
     return Settings(**{**SETTINGS_DE_TEST, **overrides})

apps/backend/tests/repositories/test_audit_log.py

@@ -0,0 +1,142 @@
+# Les trois refus ci-dessous sont la preuve que l'ajout seul est une propriété de la base et
+# non une convention de code Python. Ce sont eux qu'il faut montrer, pas la classe du dépôt.
+
+import uuid
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.exc import DBAPIError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.principal import Principal
+from app.core.roles import AccountKind, Role
+from app.models.audit_log import AuditAction, AuditOutcome
+from app.repositories.audit_log import (
+    CLES_DE_DETAIL_AUTORISEES,
+    AuditLogRepository,
+    assemble_detail,
+)
+
+pytestmark = pytest.mark.integration
+
+ACTEUR = Principal(
+    id=uuid.uuid4(),
+    email="admin@enervision.fr",
+    role=Role.ADMIN,
+    kind=AccountKind.HUMAIN,
+    must_change_password=False,
+)
+
+
+async def une_ligne(session: AsyncSession) -> None:
+    await AuditLogRepository(session).record(
+        action=AuditAction.COMPTE_CREE, actor=ACTEUR, target_type="app_user", target_id="x"
+    )
+    await session.flush()
+
+
+@pytest.mark.parametrize(
+    "instruction",
+    [
+        "update audit_log set action = 'falsifie'",
+        "delete from audit_log",
+        "truncate audit_log",
+    ],
+    ids=["modification", "suppression", "vidage"],
+)
+async def test_the_database_refuses_to_mutate_the_audit_log(
+    session: AsyncSession, instruction: str
+) -> None:
+    await une_ligne(session)
+
+    with pytest.raises(DBAPIError, match="ajout seul"):
+        await session.execute(text(instruction))
+    await session.rollback()
+
+
+async def test_record_keeps_a_snapshot_of_the_actor(session: AsyncSession) -> None:
+    depot = AuditLogRepository(session)
+    cible = uuid.uuid4().hex
+
+    await depot.record(action=AuditAction.COMPTE_DESACTIVE, actor=ACTEUR, target_id=cible)
+    await session.flush()
+    ligne = (
+        await session.execute(
+            text(
+                "select actor_id, actor_email, actor_role, outcome from audit_log "
+                "where target_id = :c"
+            ),
+            {"c": cible},
+        )
+    ).one()
+    await session.rollback()
+
+    assert ligne.actor_id == ACTEUR.id
+    assert ligne.actor_email == ACTEUR.email
+    assert ligne.actor_role == Role.ADMIN.value
+    assert ligne.outcome == AuditOutcome.SUCCES.value
+
+
+async def test_record_accepts_a_label_when_there_is_no_authenticated_actor(
+    session: AsyncSession,
+) -> None:
+    depot = AuditLogRepository(session)
+
+    cible = uuid.uuid4().hex
+    await depot.record(action=AuditAction.ADMIN_AMORCE, actor_label="cli", target_id=cible)
+    await session.flush()
+    ligne = (
+        await session.execute(
+            text("select actor_id, actor_email from audit_log where target_id = :c"),
+            {"c": cible},
+        )
+    ).one()
+    await session.rollback()
+
+    assert ligne.actor_id is None
+    assert ligne.actor_email == "cli"
+
+
+async def test_record_drops_the_detail_keys_outside_the_allow_list(
+    session: AsyncSession,
+) -> None:
+    depot = AuditLogRepository(session)
+
+    cible = uuid.uuid4().hex
+    await depot.record(
+        action=AuditAction.COMPTE_ROLE_CHANGE,
+        actor=ACTEUR,
+        target_id=cible,
+        detail={"role_avant": "lecteur", "mot_de_passe": "ne-doit-pas-passer"},
+    )
+    await session.flush()
+    detail = (
+        await session.execute(
+            text("select detail from audit_log where target_id = :c"), {"c": cible}
+        )
+    ).scalar_one()
+    await session.rollback()
+
+    assert detail == {"role_avant": "lecteur"}
+
+
+@pytest.mark.parametrize(
+    ("brut", "attendu"),
+    [
+        (None, {}),
+        ({}, {}),
+        ({"motif": "reutilisation"}, {"motif": "reutilisation"}),
+        ({"password": "x"}, {}),
+    ],
+    ids=["absent", "vide", "cle_autorisee", "cle_refusee"],
+)
+def test_assemble_detail_only_keeps_the_allowed_keys(
+    brut: dict[str, str] | None, attendu: dict[str, str]
+) -> None:
+    assert assemble_detail(brut) == attendu
+
+
+def test_the_allow_list_never_mentions_a_secret() -> None:
+    suspects = {"password", "mot_de_passe", "token", "jeton", "secret", "hash"}
+
+    assert CLES_DE_DETAIL_AUTORISEES & suspects == set()

apps/backend/tests/repositories/test_login_attempt.py

@@ -0,0 +1,118 @@
+import uuid
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.login_attempt import LoginOutcome
+from app.repositories.login_attempt import LoginAttemptRepository
+
+pytestmark = pytest.mark.integration
+
+IP = "203.0.113.10"
+AUTRE_IP = "198.51.100.7"
+
+
+def adresse() -> str:
+    return f"tentative-{uuid.uuid4().hex[:12]}@enervision.fr"
+
+
+async def echoue(
+    depot: LoginAttemptRepository, email: str, ip: str | None, combien: int = 1
+) -> None:
+    for _ in range(combien):
+        await depot.record(email=email, client_ip=ip, outcome=LoginOutcome.IDENTIFIANTS_INVALIDES)
+
+
+async def test_count_recent_failures_separates_the_three_counters(
+    session: AsyncSession,
+) -> None:
+    depot = LoginAttemptRepository(session)
+    cible, voisin = adresse(), adresse()
+    await echoue(depot, cible, IP, combien=3)
+    await echoue(depot, cible, AUTRE_IP, combien=2)
+    await echoue(depot, voisin, IP, combien=4)
+    await session.flush()
+
+    compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
+    await session.rollback()
+
+    assert compteurs.per_identifier_and_ip == 3
+    assert compteurs.per_identifier == 5
+    assert compteurs.per_ip == 7
+
+
+async def test_count_recent_failures_ignores_successful_attempts(
+    session: AsyncSession,
+) -> None:
+    depot = LoginAttemptRepository(session)
+    cible = adresse()
+    await echoue(depot, cible, IP, combien=2)
+    await depot.record(email=cible, client_ip=IP, outcome=LoginOutcome.SUCCES)
+    await session.flush()
+
+    compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
+    await session.rollback()
+
+    assert compteurs.per_identifier_and_ip == 2
+
+
+async def test_count_recent_failures_forgets_what_falls_outside_the_window(
+    session: AsyncSession,
+) -> None:
+    depot = LoginAttemptRepository(session)
+    cible = adresse()
+    await echoue(depot, cible, IP, combien=2)
+    await session.flush()
+    await session.execute(
+        text(
+            "update login_attempt set occurred_at = now() - interval '2 hours' "
+            "where email_tried = :e"
+        ),
+        {"e": cible},
+    )
+
+    compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
+    await session.rollback()
+
+    assert compteurs.per_identifier_and_ip == 0
+
+
+async def test_count_recent_failures_still_counts_when_the_address_is_unknown(
+    session: AsyncSession,
+) -> None:
+    depot = LoginAttemptRepository(session)
+    inconnu = adresse()
+    await echoue(depot, inconnu, IP, combien=5)
+    await session.flush()
+
+    compteurs = await depot.count_recent_failures(email=inconnu, client_ip=IP, window_seconds=900)
+    await session.rollback()
+
+    assert compteurs.per_identifier_and_ip == 5
+
+
+async def test_record_normalises_the_address_before_counting(session: AsyncSession) -> None:
+    depot = LoginAttemptRepository(session)
+    cible = adresse()
+    await echoue(depot, cible.upper(), IP, combien=2)
+    await session.flush()
+
+    compteurs = await depot.count_recent_failures(email=cible, client_ip=IP, window_seconds=900)
+    await session.rollback()
+
+    assert compteurs.per_identifier_and_ip == 2
+
+
+async def test_count_recent_failures_tolerates_a_missing_client_address(
+    session: AsyncSession,
+) -> None:
+    depot = LoginAttemptRepository(session)
+    cible = adresse()
+    await echoue(depot, cible, None, combien=2)
+    await session.flush()
+
+    compteurs = await depot.count_recent_failures(email=cible, client_ip=None, window_seconds=900)
+    await session.rollback()
+
+    assert compteurs.per_identifier == 2

apps/backend/tests/repositories/test_refresh_token.py

@@ -0,0 +1,190 @@
+# Le premier test de ce fichier est le seul endroit où l'atomicité de la rotation se démontre :
+# sur un double, deux appels concurrents réussiraient tous les deux.
+
+import uuid
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.roles import Role
+from app.core.security import fingerprint_refresh, generate_refresh_secret
+from app.models.refresh_token import RevocationReason
+from app.repositories.refresh_token import RefreshTokenRepository
+from app.repositories.user import UserRepository
+
+pytestmark = pytest.mark.integration
+
+DUREE = timedelta(days=7)
+
+
+async def un_compte(session: AsyncSession) -> uuid.UUID:
+    compte = await UserRepository(session).create(
+        email=f"jeton-{uuid.uuid4().hex[:12]}@enervision.fr",
+        password_hash="$argon2id$x",
+        role=Role.LECTEUR,
+    )
+    return compte.id
+
+
+async def un_jeton(
+    depot: RefreshTokenRepository,
+    user_id: uuid.UUID,
+    *,
+    family_id: uuid.UUID | None = None,
+    duree: timedelta = DUREE,
+) -> tuple[str, uuid.UUID]:
+    secret = generate_refresh_secret()
+    jeton = await depot.create(
+        user_id=user_id,
+        family_id=family_id or uuid.uuid4(),
+        token_hash=fingerprint_refresh(secret),
+        expires_at=datetime.now(UTC) + duree,
+        client_ip="203.0.113.10",
+        user_agent="pytest",
+    )
+    return secret, jeton.family_id
+
+
+async def test_claim_for_rotation_only_succeeds_once(session: AsyncSession) -> None:
+    depot = RefreshTokenRepository(session)
+    secret, _ = await un_jeton(depot, await un_compte(session))
+
+    premier = await depot.claim_for_rotation(fingerprint_refresh(secret))
+    second = await depot.claim_for_rotation(fingerprint_refresh(secret))
+    await session.rollback()
+
+    assert premier is not None
+    assert second is None
+
+
+async def test_claim_for_rotation_refuses_an_expired_token(session: AsyncSession) -> None:
+    depot = RefreshTokenRepository(session)
+    secret, _ = await un_jeton(depot, await un_compte(session), duree=-timedelta(minutes=1))
+
+    revendique = await depot.claim_for_rotation(fingerprint_refresh(secret))
+    await session.rollback()
+
+    assert revendique is None
+
+
+async def test_claim_for_rotation_returns_nothing_for_an_unknown_fingerprint(
+    session: AsyncSession,
+) -> None:
+    revendique = await RefreshTokenRepository(session).claim_for_rotation(
+        fingerprint_refresh(generate_refresh_secret())
+    )
+
+    assert revendique is None
+
+
+async def test_inspect_finds_a_token_that_rotation_already_refused(
+    session: AsyncSession,
+) -> None:
+    depot = RefreshTokenRepository(session)
+    secret, _ = await un_jeton(depot, await un_compte(session))
+    await depot.claim_for_rotation(fingerprint_refresh(secret))
+
+    ligne = await depot.inspect(fingerprint_refresh(secret))
+    rotation, motif = (ligne.rotated_at, ligne.revoked_reason) if ligne else (None, None)
+    await session.rollback()
+
+    assert rotation is not None
+    assert motif == RevocationReason.ROTATION.value
+
+
+async def test_revoke_family_touches_every_living_token_of_that_family_only(
+    session: AsyncSession,
+) -> None:
+    depot = RefreshTokenRepository(session)
+    compte = await un_compte(session)
+    famille = uuid.uuid4()
+    await un_jeton(depot, compte, family_id=famille)
+    await un_jeton(depot, compte, family_id=famille)
+    autre_secret, _ = await un_jeton(depot, compte)
+
+    revoquees = await depot.revoke_family(famille, RevocationReason.REUTILISATION)
+    intacte = await depot.claim_for_rotation(fingerprint_refresh(autre_secret))
+    await session.rollback()
+
+    assert revoquees == 2
+    assert intacte is not None
+
+
+async def test_revoke_family_is_idempotent(session: AsyncSession) -> None:
+    depot = RefreshTokenRepository(session)
+    compte = await un_compte(session)
+    famille = uuid.uuid4()
+    await un_jeton(depot, compte, family_id=famille)
+
+    premier = await depot.revoke_family(famille, RevocationReason.DECONNEXION)
+    second = await depot.revoke_family(famille, RevocationReason.DECONNEXION)
+    await session.rollback()
+
+    assert premier == 1
+    assert second == 0
+
+
+async def test_revoke_all_for_user_closes_every_family_at_once(session: AsyncSession) -> None:
+    depot = RefreshTokenRepository(session)
+    compte = await un_compte(session)
+    await un_jeton(depot, compte)
+    await un_jeton(depot, compte)
+    await un_jeton(depot, compte)
+
+    revoquees = await depot.revoke_all_for_user(compte, RevocationReason.CHANGEMENT_MOT_DE_PASSE)
+    await session.rollback()
+
+    assert revoquees == 3
+
+
+async def test_link_replacement_records_the_successor(session: AsyncSession) -> None:
+    depot = RefreshTokenRepository(session)
+    compte = await un_compte(session)
+    ancien_secret, famille = await un_jeton(depot, compte)
+    revendique = await depot.claim_for_rotation(fingerprint_refresh(ancien_secret))
+    assert revendique is not None
+    nouveau_secret = generate_refresh_secret()
+    nouveau = await depot.create(
+        user_id=compte,
+        family_id=famille,
+        token_hash=fingerprint_refresh(nouveau_secret),
+        expires_at=revendique.expires_at,
+        client_ip=None,
+        user_agent=None,
+    )
+
+    await depot.link_replacement(revendique.id, nouveau.id)
+    ligne = await depot.inspect(fingerprint_refresh(ancien_secret))
+    successeur = ligne.replaced_by if ligne else None
+    await session.rollback()
+
+    assert successeur == nouveau.id
+
+
+async def test_the_database_refuses_two_tokens_sharing_a_fingerprint(
+    session: AsyncSession,
+) -> None:
+    depot = RefreshTokenRepository(session)
+    compte = await un_compte(session)
+    secret = generate_refresh_secret()
+    await depot.create(
+        user_id=compte,
+        family_id=uuid.uuid4(),
+        token_hash=fingerprint_refresh(secret),
+        expires_at=datetime.now(UTC) + DUREE,
+        client_ip=None,
+        user_agent=None,
+    )
+
+    with pytest.raises(IntegrityError):
+        await depot.create(
+            user_id=compte,
+            family_id=uuid.uuid4(),
+            token_hash=fingerprint_refresh(secret),
+            expires_at=datetime.now(UTC) + DUREE,
+            client_ip=None,
+            user_agent=None,
+        )
+    await session.rollback()

apps/backend/tests/repositories/test_user.py

@@ -0,0 +1,196 @@
+import uuid
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.roles import AccountKind, Role
+from app.repositories.user import UserRepository
+
+pytestmark = pytest.mark.integration
+
+
+def adresse() -> str:
+    return f"compte-{uuid.uuid4().hex[:12]}@enervision.fr"
+
+
+async def test_create_normalises_the_email_to_lower_case(session: AsyncSession) -> None:
+    depot = UserRepository(session)
+    saisie = adresse().upper()
+
+    compte = await depot.create(email=saisie, password_hash="$argon2id$x", role=Role.LECTEUR)
+    enregistre = compte.email
+    await session.rollback()
+
+    assert enregistre == saisie.lower()
+
+
+async def test_the_database_refuses_an_email_written_in_upper_case(
+    session: AsyncSession,
+) -> None:
+    saisie = adresse().upper()
+
+    with pytest.raises(IntegrityError):
+        await session.execute(
+            text(
+                "insert into app_user (email, password_hash, role) "
+                "values (:e, '$argon2id$x', 'lecteur')"
+            ),
+            {"e": saisie},
+        )
+    await session.rollback()
+
+
+async def test_the_database_refuses_two_accounts_sharing_an_email(
+    session: AsyncSession,
+) -> None:
+    depot = UserRepository(session)
+    saisie = adresse()
+
+    await depot.create(email=saisie, password_hash="$argon2id$x", role=Role.LECTEUR)
+
+    with pytest.raises(IntegrityError):
+        await depot.create(email=saisie, password_hash="$argon2id$y", role=Role.ADMIN)
+    await session.rollback()
+
+
+async def test_get_by_email_is_case_insensitive(session: AsyncSession) -> None:
+    depot = UserRepository(session)
+    saisie = adresse()
+    await depot.create(email=saisie, password_hash="$argon2id$x", role=Role.OPERATEUR)
+
+    trouve = await depot.get_by_email(saisie.upper())
+    role = trouve.role if trouve else None
+    await session.rollback()
+
+    assert role == Role.OPERATEUR.value
+
+
+async def test_get_by_email_returns_nothing_for_an_unknown_address(
+    session: AsyncSession,
+) -> None:
+    trouve = await UserRepository(session).get_by_email(adresse())
+
+    assert trouve is None
+
+
+async def test_set_role_moves_the_credentials_marker_forward(session: AsyncSession) -> None:
+    depot = UserRepository(session)
+    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
+    avant = compte.credentials_changed_at
+
+    await depot.set_role(compte.id, Role.ADMIN)
+    await session.refresh(compte)
+    apres, role = compte.credentials_changed_at, compte.role
+    await session.rollback()
+
+    assert role == Role.ADMIN.value
+    assert apres > avant
+
+
+async def test_set_active_moves_the_credentials_marker_forward(session: AsyncSession) -> None:
+    depot = UserRepository(session)
+    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
+    avant = compte.credentials_changed_at
+
+    await depot.set_active(compte.id, is_active=False)
+    await session.refresh(compte)
+    apres, actif = compte.credentials_changed_at, compte.is_active
+    await session.rollback()
+
+    assert actif is False
+    assert apres > avant
+
+
+async def test_rehash_password_leaves_the_credentials_marker_untouched(
+    session: AsyncSession,
+) -> None:
+    depot = UserRepository(session)
+    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
+    avant = compte.credentials_changed_at
+
+    await depot.rehash_password(compte.id, "$argon2id$plus-recent")
+    await session.refresh(compte)
+    apres, empreinte = compte.credentials_changed_at, compte.password_hash
+    await session.rollback()
+
+    assert empreinte == "$argon2id$plus-recent"
+    assert apres == avant
+
+
+async def test_update_password_moves_the_credentials_marker_forward(
+    session: AsyncSession,
+) -> None:
+    depot = UserRepository(session)
+    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
+    avant = compte.credentials_changed_at
+
+    await depot.update_password(compte.id, "$argon2id$neuf", must_change_password=False)
+    await session.refresh(compte)
+    apres = compte.credentials_changed_at
+    await session.rollback()
+
+    assert apres > avant
+
+
+async def test_touch_last_login_records_the_connection_date(session: AsyncSession) -> None:
+    depot = UserRepository(session)
+    compte = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.LECTEUR)
+
+    await depot.touch_last_login(compte.id)
+    await session.refresh(compte)
+    date = compte.last_login_at
+    await session.rollback()
+
+    assert date is not None
+
+
+async def test_count_active_admins_only_counts_enabled_administrators(
+    session: AsyncSession,
+) -> None:
+    depot = UserRepository(session)
+    depart = await depot.count_active_admins()
+
+    await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.ADMIN)
+    desactive = await depot.create(email=adresse(), password_hash="$argon2id$x", role=Role.ADMIN)
+    await depot.set_active(desactive.id, is_active=False)
+    total = await depot.count_active_admins()
+    await session.rollback()
+
+    assert total == depart + 1
+
+
+async def test_create_accepts_a_service_account(session: AsyncSession) -> None:
+    depot = UserRepository(session)
+
+    compte = await depot.create(
+        email=adresse(),
+        password_hash="$argon2id$x",
+        role=Role.OPERATEUR,
+        kind=AccountKind.SERVICE,
+    )
+    nature = compte.kind
+    await session.rollback()
+
+    assert nature == AccountKind.SERVICE.value
+
+
+async def test_list_all_returns_the_accounts_sorted_by_email(session: AsyncSession) -> None:
+    depot = UserRepository(session)
+    await depot.create(email=f"zz-{adresse()}", password_hash="$argon2id$x", role=Role.LECTEUR)
+    await depot.create(email=f"aa-{adresse()}", password_hash="$argon2id$x", role=Role.LECTEUR)
+
+    comptes = await depot.list_all()
+    emails = [compte.email for compte in comptes]
+    await session.rollback()
+
+    assert emails == sorted(emails)
+
+
+async def test_get_by_id_returns_nothing_for_an_unknown_identifier(
+    session: AsyncSession,
+) -> None:
+    trouve = await UserRepository(session).get_by_id(uuid.uuid4())
+
+    assert trouve is None

apps/backend/tests/services/test_auth.py

@@ -0,0 +1,495 @@
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from uuid import UUID, uuid4
+
+import pytest
+
+from app.core.principal import Principal
+from app.core.roles import AccountKind, Role
+from app.core.security import (
+    TokenPolicy,
+    decode_access_token,
+    fingerprint_refresh,
+)
+from app.models.login_attempt import LoginOutcome
+from app.models.refresh_token import RevocationReason
+from app.repositories.login_attempt import FailureCounts
+from app.repositories.refresh_token import ClaimedToken
+from app.services.auth import (
+    AuthService,
+    InvalidCredentialsError,
+    LoginPolicy,
+    RateLimitedError,
+    SessionRejectedError,
+)
+
+POLITIQUE_JETON = TokenPolicy(
+    secret="un-secret-de-test-de-plus-de-trente-deux-caracteres",
+    issuer="enervision-api",
+    audience="enervision-web",
+    access_ttl=timedelta(minutes=15),
+)
+POLITIQUE_CONNEXION = LoginPolicy(
+    window_seconds=900,
+    max_failures_per_identifier_and_ip=5,
+    max_failures_per_ip=20,
+    max_failures_per_identifier=50,
+)
+
+
+@dataclass
+class FauxCompte:
+    id: UUID = field(default_factory=uuid4)
+    email: str = "operateur@enervision.fr"
+    password_hash: str = "$argon2id$factice"
+    role: str = "operateur"
+    kind: str = "human"
+    is_active: bool = True
+    must_change_password: bool = False
+    credentials_changed_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+
+
+class FauxDepotComptes:
+    def __init__(self, compte: FauxCompte | None) -> None:
+        self.compte = compte
+        self.rehachages = 0
+        self.connexions_datees = 0
+        self.mots_de_passe_changes = 0
+
+    async def get_by_email(self, email: str) -> FauxCompte | None:
+        return self.compte
+
+    async def get_by_id(self, user_id: UUID) -> FauxCompte | None:
+        return self.compte
+
+    async def rehash_password(self, user_id: UUID, password_hash: str) -> None:
+        self.rehachages += 1
+
+    async def update_password(self, user_id: UUID, password_hash: str, **_: object) -> None:
+        self.mots_de_passe_changes += 1
+
+    async def touch_last_login(self, user_id: UUID) -> None:
+        self.connexions_datees += 1
+
+
+class FauxDepotTentatives:
+    def __init__(self, compteurs: FailureCounts | None = None) -> None:
+        self.compteurs = compteurs or FailureCounts(0, 0, 0)
+        self.enregistrees: list[str] = []
+
+    async def count_recent_failures(self, **_: object) -> FailureCounts:
+        return self.compteurs
+
+    async def record(self, *, outcome: object, **_: object) -> None:
+        self.enregistrees.append(str(outcome))
+
+
+class FauxDepotAudit:
+    def __init__(self) -> None:
+        self.lignes: list[tuple[str, Mapping[str, Any] | None]] = []
+
+    async def record(self, *, action: object, detail: Any = None, **_: object) -> None:
+        self.lignes.append((str(action), detail))
+
+
+@dataclass
+class FauxJeton:
+    id: UUID = field(default_factory=uuid4)
+    family_id: UUID = field(default_factory=uuid4)
+    user_id: UUID = field(default_factory=uuid4)
+    expires_at: datetime = field(default_factory=lambda: datetime.now(UTC) + timedelta(days=7))
+    rotated_at: datetime | None = None
+    revoked_at: datetime | None = None
+
+
+class FauxDepotJetons:
+    def __init__(
+        self, revendique: ClaimedToken | None = None, connu: FauxJeton | None = None
+    ) -> None:
+        self.revendique = revendique
+        self.connu = connu
+        self.crees: list[UUID] = []
+        self.familles_revoquees: list[tuple[UUID, str]] = []
+        self.revocations_par_compte: list[tuple[UUID, str]] = []
+        self.liaisons: list[tuple[UUID, UUID]] = []
+
+    async def create(self, *, user_id: UUID, family_id: UUID, **_: object) -> FauxJeton:
+        jeton = FauxJeton(user_id=user_id, family_id=family_id)
+        self.crees.append(jeton.id)
+        return jeton
+
+    async def claim_for_rotation(self, token_hash: bytes) -> ClaimedToken | None:
+        return self.revendique
+
+    async def inspect(self, token_hash: bytes) -> FauxJeton | None:
+        return self.connu
+
+    async def link_replacement(self, ancien_id: UUID, nouveau_id: UUID) -> None:
+        self.liaisons.append((ancien_id, nouveau_id))
+
+    async def revoke_family(self, family_id: UUID, reason: RevocationReason) -> int:
+        self.familles_revoquees.append((family_id, reason.value))
+        return 2
+
+    async def revoke_all_for_user(self, user_id: UUID, reason: RevocationReason) -> int:
+        self.revocations_par_compte.append((user_id, reason.value))
+        return 3
+
+
+class FauxHacheur:
+    def __init__(self, *, accepte: bool = True, rehachage_requis: bool = False) -> None:
+        self.verifications = 0
+        self.hachages = 0
+        self._accepte = accepte
+        self._rehachage_requis = rehachage_requis
+
+    async def hash(self, password: str) -> str:
+        self.hachages += 1
+        return "$argon2id$nouvelle"
+
+    async def verify(self, stored: str, password: str) -> bool:
+        self.verifications += 1
+        return self._accepte
+
+    async def verify_dummy(self) -> None:
+        self.verifications += 1
+
+    def needs_rehash(self, stored: str) -> bool:
+        return self._rehachage_requis
+
+
+class FausseTransaction:
+    def __init__(self) -> None:
+        self.validations = 0
+
+    async def commit(self) -> None:
+        self.validations += 1
+
+
+@dataclass
+class Attirail:
+    service: AuthService
+    comptes: FauxDepotComptes
+    tentatives: FauxDepotTentatives
+    jetons: FauxDepotJetons
+    audit: FauxDepotAudit
+    hacheur: FauxHacheur
+
+
+def fabrique_service(
+    *,
+    compte: FauxCompte | None = None,
+    compteurs: FailureCounts | None = None,
+    hacheur: FauxHacheur | None = None,
+    jetons: FauxDepotJetons | None = None,
+) -> Attirail:
+    comptes = FauxDepotComptes(compte)
+    tentatives = FauxDepotTentatives(compteurs)
+    depot_jetons = jetons or FauxDepotJetons()
+    audit = FauxDepotAudit()
+    hacheur = hacheur or FauxHacheur()
+    service = AuthService(
+        users=comptes,  # type: ignore[arg-type]
+        attempts=tentatives,  # type: ignore[arg-type]
+        refresh_tokens=depot_jetons,  # type: ignore[arg-type]
+        audit=audit,  # type: ignore[arg-type]
+        hasher=hacheur,  # type: ignore[arg-type]
+        transaction=FausseTransaction(),
+        token_policy=POLITIQUE_JETON,
+        login_policy=POLITIQUE_CONNEXION,
+        refresh_ttl=timedelta(days=7),
+    )
+    return Attirail(service, comptes, tentatives, depot_jetons, audit, hacheur)
+
+
+async def connecte(service: AuthService, mot_de_passe: str = "un-mot-de-passe-valide") -> object:
+    return await service.authenticate(
+        email="operateur@enervision.fr",
+        password=mot_de_passe,
+        client_ip="203.0.113.10",
+        user_agent="pytest",
+    )
+
+
+async def rafraichit(service: AuthService, secret: str = "un-secret-opaque") -> object:
+    return await service.refresh(secret=secret, client_ip="203.0.113.10", user_agent="pytest")
+
+
+async def test_authenticate_returns_a_readable_access_token_when_credentials_match() -> None:
+    compte = FauxCompte()
+    attirail = fabrique_service(compte=compte)
+
+    session = await connecte(attirail.service)
+
+    claims = decode_access_token(POLITIQUE_JETON, session.access_token)  # type: ignore[attr-defined]
+    assert claims.subject == compte.id
+    assert claims.role == "operateur"
+    assert attirail.tentatives.enregistrees == [LoginOutcome.SUCCES.value]
+    assert attirail.comptes.connexions_datees == 1
+
+
+async def test_authenticate_opens_one_refresh_family_per_login() -> None:
+    attirail = fabrique_service(compte=FauxCompte())
+
+    session = await connecte(attirail.service)
+
+    assert len(attirail.jetons.crees) == 1
+    assert session.refresh_secret  # type: ignore[attr-defined]
+
+
+async def test_authenticate_verifies_a_decoy_digest_when_the_email_is_unknown() -> None:
+    attirail = fabrique_service(compte=None)
+
+    with pytest.raises(InvalidCredentialsError):
+        await connecte(attirail.service)
+
+    assert attirail.hacheur.verifications == 1
+    assert attirail.tentatives.enregistrees == [LoginOutcome.IDENTIFIANTS_INVALIDES.value]
+
+
+async def test_authenticate_skips_hashing_entirely_when_the_rate_limit_is_reached() -> None:
+    compteurs = FailureCounts(per_identifier_and_ip=5, per_ip=5, per_identifier=5)
+    attirail = fabrique_service(compte=FauxCompte(), compteurs=compteurs)
+
+    with pytest.raises(RateLimitedError):
+        await connecte(attirail.service)
+
+    assert attirail.hacheur.verifications == 0
+    assert attirail.hacheur.hachages == 0
+    assert attirail.tentatives.enregistrees == [LoginOutcome.LIMITE.value]
+    assert attirail.audit.lignes == []
+
+
+async def test_authenticate_audits_when_the_identifier_threshold_alone_is_reached() -> None:
+    compteurs = FailureCounts(per_identifier_and_ip=0, per_ip=0, per_identifier=50)
+    attirail = fabrique_service(compte=FauxCompte(), compteurs=compteurs)
+
+    with pytest.raises(RateLimitedError):
+        await connecte(attirail.service)
+
+    assert len(attirail.audit.lignes) == 1
+    assert "identifier_throttled" in attirail.audit.lignes[0][0]
+
+
+async def test_authenticate_rejects_a_wrong_password_with_the_generic_error() -> None:
+    attirail = fabrique_service(compte=FauxCompte(), hacheur=FauxHacheur(accepte=False))
+
+    with pytest.raises(InvalidCredentialsError):
+        await connecte(attirail.service)
+
+    assert attirail.tentatives.enregistrees == [LoginOutcome.IDENTIFIANTS_INVALIDES.value]
+
+
+@pytest.mark.parametrize(
+    "compte",
+    [FauxCompte(is_active=False), FauxCompte(kind="service")],
+    ids=["compte_desactive", "compte_de_service"],
+)
+async def test_authenticate_rejects_unavailable_accounts_after_checking_the_password(
+    compte: FauxCompte,
+) -> None:
+    attirail = fabrique_service(compte=compte)
+
+    with pytest.raises(InvalidCredentialsError):
+        await connecte(attirail.service)
+
+    assert attirail.hacheur.verifications == 1
+    assert attirail.tentatives.enregistrees == [LoginOutcome.COMPTE_INDISPONIBLE.value]
+
+
+async def test_authenticate_rehashes_the_password_when_the_parameters_changed() -> None:
+    attirail = fabrique_service(compte=FauxCompte(), hacheur=FauxHacheur(rehachage_requis=True))
+
+    await connecte(attirail.service)
+
+    assert attirail.comptes.rehachages == 1
+
+
+async def test_authenticate_leaves_the_digest_alone_when_the_parameters_match() -> None:
+    attirail = fabrique_service(compte=FauxCompte())
+
+    await connecte(attirail.service)
+
+    assert attirail.comptes.rehachages == 0
+
+
+async def test_refresh_rotates_the_token_and_keeps_the_family() -> None:
+    compte = FauxCompte()
+    revendique = ClaimedToken(
+        id=uuid4(),
+        family_id=uuid4(),
+        user_id=compte.id,
+        expires_at=datetime.now(UTC) + timedelta(days=5),
+    )
+    attirail = fabrique_service(compte=compte, jetons=FauxDepotJetons(revendique=revendique))
+
+    session = await rafraichit(attirail.service)
+
+    assert session.refresh_secret  # type: ignore[attr-defined]
+    assert len(attirail.jetons.crees) == 1
+    assert attirail.jetons.liaisons == [(revendique.id, attirail.jetons.crees[0])]
+    assert attirail.jetons.familles_revoquees == []
+
+
+async def test_refresh_inherits_the_absolute_expiry_of_its_predecessor() -> None:
+    compte = FauxCompte()
+    echeance = datetime.now(UTC) + timedelta(days=2)
+    revendique = ClaimedToken(id=uuid4(), family_id=uuid4(), user_id=compte.id, expires_at=echeance)
+    attirail = fabrique_service(compte=compte, jetons=FauxDepotJetons(revendique=revendique))
+
+    await rafraichit(attirail.service)
+
+    assert revendique.expires_at == echeance
+
+
+async def test_refresh_rejects_an_unknown_secret_without_touching_any_family() -> None:
+    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons())
+
+    with pytest.raises(SessionRejectedError):
+        await rafraichit(attirail.service)
+
+    assert attirail.jetons.familles_revoquees == []
+    assert attirail.audit.lignes == []
+
+
+async def test_refresh_rejects_an_expired_token_without_revoking_its_family() -> None:
+    perime = FauxJeton(expires_at=datetime.now(UTC) - timedelta(minutes=1))
+    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons(connu=perime))
+
+    with pytest.raises(SessionRejectedError):
+        await rafraichit(attirail.service)
+
+    assert attirail.jetons.familles_revoquees == []
+    assert attirail.audit.lignes == []
+
+
+async def test_refresh_revokes_the_whole_family_when_a_rotated_token_comes_back() -> None:
+    rejoue = FauxJeton(rotated_at=datetime.now(UTC), revoked_at=datetime.now(UTC))
+    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons(connu=rejoue))
+
+    with pytest.raises(SessionRejectedError):
+        await rafraichit(attirail.service)
+
+    assert attirail.jetons.familles_revoquees == [
+        (rejoue.family_id, RevocationReason.REUTILISATION.value)
+    ]
+    assert "refresh_reuse_detected" in attirail.audit.lignes[0][0]
+
+
+async def test_refresh_revokes_the_family_when_the_account_was_disabled_meanwhile() -> None:
+    compte = FauxCompte(is_active=False)
+    revendique = ClaimedToken(
+        id=uuid4(),
+        family_id=uuid4(),
+        user_id=compte.id,
+        expires_at=datetime.now(UTC) + timedelta(days=5),
+    )
+    attirail = fabrique_service(compte=compte, jetons=FauxDepotJetons(revendique=revendique))
+
+    with pytest.raises(SessionRejectedError):
+        await rafraichit(attirail.service)
+
+    assert attirail.jetons.familles_revoquees == [
+        (revendique.family_id, RevocationReason.ADMINISTRATION.value)
+    ]
+
+
+async def test_logout_revokes_only_the_presented_family() -> None:
+    connu = FauxJeton()
+    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons(connu=connu))
+
+    await attirail.service.logout(secret="un-secret-opaque")
+
+    assert attirail.jetons.familles_revoquees == [
+        (connu.family_id, RevocationReason.DECONNEXION.value)
+    ]
+    assert attirail.jetons.revocations_par_compte == []
+
+
+async def test_logout_stays_silent_when_the_cookie_points_at_nothing() -> None:
+    attirail = fabrique_service(compte=FauxCompte(), jetons=FauxDepotJetons())
+
+    await attirail.service.logout(secret="un-secret-inconnu")
+
+    assert attirail.jetons.familles_revoquees == []
+
+
+async def test_logout_all_revokes_every_session_and_leaves_an_audit_trail() -> None:
+    compte = FauxCompte()
+    attirail = fabrique_service(compte=compte)
+    acteur = Principal(
+        id=compte.id,
+        email=compte.email,
+        role=Role.OPERATEUR,
+        kind=AccountKind.HUMAIN,
+        must_change_password=False,
+    )
+
+    revoquees = await attirail.service.logout_all(acteur)
+
+    assert revoquees == 3
+    assert attirail.jetons.revocations_par_compte == [
+        (compte.id, RevocationReason.DECONNEXION.value)
+    ]
+    assert "all_sessions_revoked" in attirail.audit.lignes[0][0]
+
+
+def test_fingerprint_is_what_the_service_stores_not_the_secret_itself() -> None:
+    secret = "un-secret-opaque"
+
+    empreinte = fingerprint_refresh(secret)
+
+    assert secret.encode() not in empreinte
+
+
+async def test_change_password_revokes_every_session_then_reopens_the_current_one() -> None:
+    compte = FauxCompte()
+    attirail = fabrique_service(compte=compte)
+    acteur = Principal(
+        id=compte.id,
+        email=compte.email,
+        role=Role.OPERATEUR,
+        kind=AccountKind.HUMAIN,
+        must_change_password=True,
+    )
+
+    session = await attirail.service.change_password(
+        principal=acteur,
+        current_password="l-ancien-mot-de-passe",
+        new_password="le-nouveau-mot-de-passe",
+        client_ip="203.0.113.10",
+        user_agent="pytest",
+    )
+
+    assert attirail.jetons.revocations_par_compte == [
+        (compte.id, RevocationReason.CHANGEMENT_MOT_DE_PASSE.value)
+    ]
+    assert len(attirail.jetons.crees) == 1, "l'appareil courant doit repartir avec une session"
+    assert session.refresh_secret
+    assert "password_changed" in attirail.audit.lignes[0][0]
+
+
+async def test_change_password_refuses_a_wrong_current_password() -> None:
+    compte = FauxCompte()
+    attirail = fabrique_service(compte=compte, hacheur=FauxHacheur(accepte=False))
+    acteur = Principal(
+        id=compte.id,
+        email=compte.email,
+        role=Role.OPERATEUR,
+        kind=AccountKind.HUMAIN,
+        must_change_password=False,
+    )
+
+    with pytest.raises(InvalidCredentialsError):
+        await attirail.service.change_password(
+            principal=acteur,
+            current_password="mauvais",
+            new_password="le-nouveau-mot-de-passe",
+            client_ip=None,
+            user_agent=None,
+        )
+
+    assert attirail.jetons.revocations_par_compte == []
+    assert attirail.jetons.crees == []

apps/backend/tests/services/test_user.py

@@ -0,0 +1,239 @@
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from typing import Any
+from uuid import UUID, uuid4
+
+import pytest
+
+from app.core.principal import Principal
+from app.core.roles import AccountKind, Role
+from app.models.refresh_token import RevocationReason
+from app.services.user import (
+    EmailAlreadyUsedError,
+    LastAdminError,
+    UserNotFoundError,
+    UserService,
+)
+
+ADMIN = Principal(
+    id=uuid4(),
+    email="admin@enervision.fr",
+    role=Role.ADMIN,
+    kind=AccountKind.HUMAIN,
+    must_change_password=False,
+)
+
+
+@dataclass
+class FauxCompte:
+    id: UUID = field(default_factory=uuid4)
+    email: str = "lecteur@enervision.fr"
+    password_hash: str = "$argon2id$factice"
+    role: str = "lecteur"
+    kind: str = "human"
+    is_active: bool = True
+    must_change_password: bool = False
+    full_name: str | None = None
+    last_login_at: datetime | None = None
+    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+
+
+class FauxDepotComptes:
+    def __init__(
+        self, compte: FauxCompte | None = None, *, admins_actifs: int = 2, existe: bool = False
+    ) -> None:
+        self.compte = compte
+        self.admins_actifs = admins_actifs
+        self.existe = existe
+        self.crees: list[str] = []
+        self.roles_poses: list[tuple[UUID, str]] = []
+        self.activations: list[tuple[UUID, bool]] = []
+        self.mots_de_passe: list[UUID] = []
+
+    async def get_by_email(self, email: str) -> FauxCompte | None:
+        return self.compte if self.existe else None
+
+    async def get_by_id(self, user_id: UUID) -> FauxCompte | None:
+        return self.compte
+
+    async def count_active_admins(self) -> int:
+        return self.admins_actifs
+
+    async def create(self, *, email: str, **_: object) -> FauxCompte:
+        self.crees.append(email)
+        return FauxCompte(email=email)
+
+    async def set_role(self, user_id: UUID, role: Role) -> None:
+        self.roles_poses.append((user_id, role.value))
+
+    async def set_active(self, user_id: UUID, *, is_active: bool) -> None:
+        self.activations.append((user_id, is_active))
+
+    async def update_password(self, user_id: UUID, password_hash: str, **_: object) -> None:
+        self.mots_de_passe.append(user_id)
+
+
+class FauxDepotJetons:
+    def __init__(self) -> None:
+        self.revocations: list[tuple[UUID, str]] = []
+
+    async def revoke_all_for_user(self, user_id: UUID, reason: RevocationReason) -> int:
+        self.revocations.append((user_id, reason.value))
+        return 2
+
+
+class FauxDepotAudit:
+    def __init__(self) -> None:
+        self.lignes: list[tuple[str, Any]] = []
+
+    async def record(self, *, action: object, detail: Any = None, **_: object) -> None:
+        self.lignes.append((str(action), detail))
+
+
+class FauxHacheur:
+    async def hash(self, password: str) -> str:
+        return "$argon2id$nouvelle"
+
+
+class FausseTransaction:
+    async def commit(self) -> None:
+        return None
+
+
+@dataclass
+class Attirail:
+    service: UserService
+    comptes: FauxDepotComptes
+    jetons: FauxDepotJetons
+    audit: FauxDepotAudit
+
+
+def fabrique(
+    compte: FauxCompte | None = None, *, admins_actifs: int = 2, existe: bool = False
+) -> Attirail:
+    comptes = FauxDepotComptes(compte, admins_actifs=admins_actifs, existe=existe)
+    jetons = FauxDepotJetons()
+    audit = FauxDepotAudit()
+    service = UserService(
+        users=comptes,  # type: ignore[arg-type]
+        refresh_tokens=jetons,  # type: ignore[arg-type]
+        audit=audit,  # type: ignore[arg-type]
+        hasher=FauxHacheur(),  # type: ignore[arg-type]
+        transaction=FausseTransaction(),
+    )
+    return Attirail(service, comptes, jetons, audit)
+
+
+async def test_create_returns_a_temporary_password_shown_once() -> None:
+    attirail = fabrique()
+
+    cree = await attirail.service.create(
+        actor=ADMIN, email="nouveau@enervision.fr", role=Role.LECTEUR, full_name=None
+    )
+
+    assert len(cree.temporary_password) >= 18
+    assert attirail.comptes.crees == ["nouveau@enervision.fr"]
+    assert "user.created" in attirail.audit.lignes[0][0]
+
+
+async def test_create_refuses_an_address_already_taken() -> None:
+    attirail = fabrique(FauxCompte(), existe=True)
+
+    with pytest.raises(EmailAlreadyUsedError):
+        await attirail.service.create(
+            actor=ADMIN, email="lecteur@enervision.fr", role=Role.LECTEUR, full_name=None
+        )
+
+
+async def test_change_role_revokes_every_session_of_the_target() -> None:
+    cible = FauxCompte()
+    attirail = fabrique(cible)
+
+    await attirail.service.change_role(actor=ADMIN, user_id=cible.id, role=Role.OPERATEUR)
+
+    assert attirail.comptes.roles_poses == [(cible.id, "operateur")]
+    assert attirail.jetons.revocations == [(cible.id, RevocationReason.ADMINISTRATION.value)]
+
+
+async def test_change_role_does_nothing_when_the_role_is_already_the_right_one() -> None:
+    cible = FauxCompte(role="operateur")
+    attirail = fabrique(cible)
+
+    await attirail.service.change_role(actor=ADMIN, user_id=cible.id, role=Role.OPERATEUR)
+
+    assert attirail.comptes.roles_poses == []
+    assert attirail.jetons.revocations == []
+
+
+async def test_change_role_refuses_to_demote_the_last_active_administrator() -> None:
+    dernier = FauxCompte(role="admin")
+    attirail = fabrique(dernier, admins_actifs=1)
+
+    with pytest.raises(LastAdminError):
+        await attirail.service.change_role(actor=ADMIN, user_id=dernier.id, role=Role.LECTEUR)
+
+
+async def test_change_role_accepts_a_demotion_when_another_administrator_remains() -> None:
+    admin = FauxCompte(role="admin")
+    attirail = fabrique(admin, admins_actifs=2)
+
+    await attirail.service.change_role(actor=ADMIN, user_id=admin.id, role=Role.LECTEUR)
+
+    assert attirail.comptes.roles_poses == [(admin.id, "lecteur")]
+
+
+async def test_set_active_refuses_to_disable_the_last_active_administrator() -> None:
+    dernier = FauxCompte(role="admin")
+    attirail = fabrique(dernier, admins_actifs=1)
+
+    with pytest.raises(LastAdminError):
+        await attirail.service.set_active(actor=ADMIN, user_id=dernier.id, is_active=False)
+
+
+async def test_set_active_revokes_the_sessions_when_disabling() -> None:
+    cible = FauxCompte()
+    attirail = fabrique(cible)
+
+    await attirail.service.set_active(actor=ADMIN, user_id=cible.id, is_active=False)
+
+    assert attirail.comptes.activations == [(cible.id, False)]
+    assert attirail.jetons.revocations == [(cible.id, RevocationReason.ADMINISTRATION.value)]
+
+
+async def test_set_active_leaves_the_sessions_alone_when_enabling() -> None:
+    cible = FauxCompte(is_active=False)
+    attirail = fabrique(cible)
+
+    await attirail.service.set_active(actor=ADMIN, user_id=cible.id, is_active=True)
+
+    assert attirail.jetons.revocations == []
+
+
+async def test_reset_password_closes_every_session_and_forces_a_change() -> None:
+    cible = FauxCompte()
+    attirail = fabrique(cible)
+
+    reinitialise = await attirail.service.reset_password(actor=ADMIN, user_id=cible.id)
+
+    assert len(reinitialise.temporary_password) >= 18
+    assert attirail.comptes.mots_de_passe == [cible.id]
+    assert attirail.jetons.revocations == [
+        (cible.id, RevocationReason.CHANGEMENT_MOT_DE_PASSE.value)
+    ]
+
+
+@pytest.mark.parametrize(
+    "action",
+    ["change_role", "set_active", "reset_password"],
+    ids=["changement_de_role", "activation", "reinitialisation"],
+)
+async def test_every_operation_refuses_an_unknown_account(action: str) -> None:
+    attirail = fabrique(None)
+    arguments: dict[str, Any] = {"actor": ADMIN, "user_id": uuid4()}
+    if action == "change_role":
+        arguments["role"] = Role.ADMIN
+    if action == "set_active":
+        arguments["is_active"] = False
+
+    with pytest.raises(UserNotFoundError):
+        await getattr(attirail.service, action)(**arguments)

apps/backend/tests/test_cli.py

@@ -0,0 +1,57 @@
+import pytest
+
+from app import cli
+
+
+def test_build_parser_reads_the_create_admin_arguments() -> None:
+    arguments = cli.build_parser().parse_args(
+        ["create-admin", "--email", "admin@enervision.fr", "--generate", "--force"]
+    )
+
+    assert arguments.commande == "create-admin"
+    assert arguments.email == "admin@enervision.fr"
+    assert arguments.generate is True
+    assert arguments.force is True
+
+
+def test_build_parser_requires_a_subcommand() -> None:
+    with pytest.raises(SystemExit):
+        cli.build_parser().parse_args([])
+
+
+def test_build_parser_requires_an_email() -> None:
+    with pytest.raises(SystemExit):
+        cli.build_parser().parse_args(["create-admin"])
+
+
+def test_read_password_generates_a_long_secret_when_asked(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    mot_de_passe = cli.read_password(generate=True)
+
+    assert len(mot_de_passe) >= cli.LONGUEUR_MOT_DE_PASSE_GENERE
+    assert mot_de_passe in capsys.readouterr().out
+
+
+def test_read_password_accepts_two_matching_entries(monkeypatch: pytest.MonkeyPatch) -> None:
+    saisies = iter(["un-mot-de-passe-valide", "un-mot-de-passe-valide"])
+    monkeypatch.setattr(cli, "getpass", lambda _: next(saisies))
+
+    assert cli.read_password(generate=False) == "un-mot-de-passe-valide"
+
+
+def test_read_password_refuses_a_password_below_the_minimum_length(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(cli, "getpass", lambda _: "court")
+
+    with pytest.raises(SystemExit):
+        cli.read_password(generate=False)
+
+
+def test_read_password_refuses_two_different_entries(monkeypatch: pytest.MonkeyPatch) -> None:
+    saisies = iter(["un-mot-de-passe-valide", "un-autre-mot-de-passe"])
+    monkeypatch.setattr(cli, "getpass", lambda _: next(saisies))
+
+    with pytest.raises(SystemExit):
+        cli.read_password(generate=False)