Merge remote-tracking branch 'origin/dev' into feat/service-de-scoring

apps/backend/tests/api/acces.py

@@ -0,0 +1,87 @@
+# Pourquoi : classification unique des routes du contrat, lue par test_route_protection.py,
+# test_openapi.py et test_matrice_acces.py. Trois listes séparées dérivaient auparavant chacune
+# de leur côté, et deux entrées de ROUTES_A_ROLE ne correspondaient plus à aucune route sans que
+# rien ne le signale.
+# Piège : les trois ensembles doivent rester disjoints et couvrir tout le schéma. C'est
+# `test_every_declared_route_is_classified` qui le vérifie, pas la relecture.
+
+from typing import Final
+
+from app.core.roles import Role
+
+Route = tuple[str, str]
+
+ROUTES_PUBLIQUES: Final[frozenset[Route]] = 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"),
+        ("POST", "/api/v1/auth/forgot-password"),
+        # Protégée par le jeton dans le corps de la requête, pas par un `Principal` : aucune
+        # authentification préalable ne s'applique, c'est la validité du jeton qui tranche.
+        ("POST", "/api/v1/auth/reset-password"),
+        # Même raison : lecture seule, protégée par le jeton passé en paramètre, pas par un
+        # `Principal`. Le jeton est un secret de 256 bits, non brute-forçable.
+        ("GET", "/api/v1/auth/reset-password/validate"),
+        ("GET", "/metrics"),
+    }
+)
+
+# Le cookie opaque porte seul l'autorisation : sans lui la route rend 401, mais aucun `Principal`
+# n'est construit et `require_role` n'entre jamais en jeu.
+ROUTE_COOKIE: Final[frozenset[Route]] = frozenset({("POST", "/api/v1/auth/refresh")})
+
+# Authentifiées par `CurrentPrincipalDep` nu, donc hors de `require_role` et, avec lui, hors du
+# refus `password_change_required`. Volontaire pour `/auth/password`, qui est la sortie de l'état
+# provisoire ; subi pour `/auth/logout-all`, cf. test_matrice_acces.py.
+ROUTES_SANS_ROLE: Final[frozenset[Route]] = frozenset(
+    {
+        ("GET", "/api/v1/auth/me"),
+        ("POST", "/api/v1/auth/password"),
+        ("POST", "/api/v1/auth/logout-all"),
+    }
+)
+
+ROLE_MINIMUM: Final[dict[Route, Role]] = {
+    ("GET", "/api/v1/sites"): Role.LECTEUR,
+    ("GET", "/api/v1/sites/{site_id}"): Role.LECTEUR,
+    ("GET", "/api/v1/sites/{site_id}/current"): Role.LECTEUR,
+    ("GET", "/api/v1/alerts"): Role.LECTEUR,
+    ("GET", "/api/v1/recommendations"): Role.LECTEUR,
+    ("GET", "/api/v1/recommendations/{recommendation_id}"): Role.LECTEUR,
+    ("GET", "/api/v1/stats/summary"): Role.LECTEUR,
+    ("GET", "/api/v1/readings"): Role.LECTEUR,
+    ("GET", "/api/v1/predictions"): Role.LECTEUR,
+    ("GET", "/api/v1/sensors/status"): Role.ADMIN,
+    ("GET", "/api/v1/users"): Role.ADMIN,
+    ("POST", "/api/v1/users"): Role.ADMIN,
+    ("PATCH", "/api/v1/users/{user_id}"): Role.ADMIN,
+    ("POST", "/api/v1/users/{user_id}/password-reset"): Role.ADMIN,
+}
+
+# Piège : `{recommendation_id}` est typé `int` et `{user_id}` est un UUID. Une substitution
+# uniforme par une chaîne quelconque rendrait 422 avant d'atteindre la garde de rôle, et le test
+# passerait en prouvant autre chose que ce qu'il annonce.
+SUBSTITUTIONS: Final[dict[str, str]] = {
+    "{user_id}": "00000000-0000-0000-0000-000000000000",
+    "{site_id}": "site-absent-du-jeu-de-donnees",
+    "{recommendation_id}": "999999999",
+}
+
+
+def chemin_concret(chemin: str) -> str:
+    for gabarit, valeur in SUBSTITUTIONS.items():
+        chemin = chemin.replace(gabarit, valeur)
+    return chemin
+
+
+def routes_du_schema(schema: dict[str, object]) -> list[Route]:
+    chemins: dict[str, dict[str, object]] = schema["paths"]  # type: ignore[assignment]
+    return [
+        (methode.upper(), chemin)
+        for chemin, operations in chemins.items()
+        for methode in operations
+        if methode.upper() in {"GET", "POST", "PATCH", "PUT", "DELETE"}
+    ]

apps/backend/tests/api/test_matrice_acces.py

@@ -0,0 +1,279 @@
+# Pourquoi : la matrice rôle x route sur les routes réelles. `test_authorization.py` la joue déjà,
+# mais contre une route jetable montée par une fixture, ce qui ne dit rien du niveau effectivement
+# posé sur `/sites` ou `/users`. `ROLE_MINIMUM` (tests/api/acces.py) est la référence, et ce
+# fichier est ce qui la confronte au comportement observé.
+# Piège : l'assertion porte sur le refus de la garde, pas sur un 200. Un rôle suffisant peut
+# légitimement recevoir 404 ou 422 selon les données ; ce qui compte est qu'il ne reçoive pas le
+# 403 `Droits insuffisants`. Sans cette nuance, le test dépendrait du contenu de la base.
+# Les tests `integration` en fin de fichier rejouent la même matrice avec de vrais jetons, donc en
+# traversant le décodage du JWT et la relecture du compte, ce que l'override court-circuite.
+
+import uuid
+from collections.abc import AsyncIterator, Callable, Iterator
+
+import pytest
+from fastapi import FastAPI
+from httpx import AsyncClient, Response
+from sqlalchemy import text
+
+from app.api.deps import get_current_principal
+from app.core.hashing import build_hasher
+from app.core.principal import Principal
+from app.core.roles import AccountKind, Role, has_at_least
+from app.db.session import get_session, get_session_factory
+from app.repositories.user import UserRepository
+from tests.api.acces import ROLE_MINIMUM, chemin_concret
+
+ROLES = [Role.LECTEUR, Role.OPERATEUR, Role.ADMIN]
+IDS_DE_ROLE = ["lecteur", "operateur", "admin"]
+REFUS_DE_DROITS = "Droits insuffisants"
+REFUS_DE_MOT_DE_PASSE = "password_change_required"
+MOT_DE_PASSE = "un-mot-de-passe-de-recette"
+
+
+# `FakeSession` de tests/factories.py rend un unique objet pour les trois formes d'appel, ce qui
+# suffit à un test d'endpoint ciblé mais pas à balayer 13 routes qui interrogent chacune la base
+# à sa façon. Ce double rend un résultat vide quelle que soit la forme demandée, pour que la
+# réponse observée vienne de la garde de rôle et jamais d'un double mal ajusté.
+class ResultatVide:
+    def scalars(self) -> ResultatVide:
+        return self
+
+    def all(self) -> list[object]:
+        return []
+
+    def first(self) -> None:
+        return None
+
+    def one_or_none(self) -> None:
+        return None
+
+    def scalar_one_or_none(self) -> None:
+        return None
+
+    def mappings(self) -> ResultatVide:
+        return self
+
+    def __iter__(self) -> Iterator[object]:
+        return iter(())
+
+
+class SessionMuette:
+    async def scalar(self, *_: object, **__: object) -> None:
+        return None
+
+    async def execute(self, *_: object, **__: object) -> ResultatVide:
+        return ResultatVide()
+
+    async def scalars(self, *_: object, **__: object) -> ResultatVide:
+        return ResultatVide()
+
+    async def get(self, *_: object, **__: object) -> None:
+        return None
+
+    async def flush(self) -> None:
+        return None
+
+    async def commit(self) -> None:
+        return None
+
+    async def rollback(self) -> None:
+        return None
+
+    def add(self, *_: object, **__: object) -> None:
+        return None
+
+
+@pytest.fixture
+def base_muette(app: FastAPI) -> None:
+    async def override() -> AsyncIterator[SessionMuette]:
+        yield SessionMuette()
+
+    app.dependency_overrides[get_session] = override
+
+
+def principal(role: Role, *, must_change_password: bool = False) -> Principal:
+    return Principal(
+        id=uuid.uuid4(),
+        email=f"matrice-{role.value}@enervision.fr",
+        role=role,
+        kind=AccountKind.HUMAIN,
+        must_change_password=must_change_password,
+    )
+
+
+@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 appelle(client: AsyncClient, methode: str, chemin: str, **kwargs: object) -> Response:
+    return await client.request(methode, chemin_concret(chemin), json={}, **kwargs)  # type: ignore[arg-type]
+
+
+def motif_du_refus(response: Response) -> str | None:
+    if response.status_code != 403:
+        return None
+    detail = response.json().get("detail")
+    return detail if isinstance(detail, str) else None
+
+
+@pytest.mark.parametrize("role", ROLES, ids=IDS_DE_ROLE)
+async def test_a_role_below_the_minimum_is_refused_on_every_guarded_route(
+    connecte: Callable[[Principal], None],
+    client: AsyncClient,
+    base_muette: None,
+    role: Role,
+) -> None:
+    connecte(principal(role))
+    laissees_passer: list[tuple[str, str, int]] = []
+
+    for (methode, chemin), minimum in ROLE_MINIMUM.items():
+        if has_at_least(role, minimum):
+            continue
+        response = await appelle(client, methode, chemin)
+        if motif_du_refus(response) != REFUS_DE_DROITS:
+            laissees_passer.append((methode, chemin, response.status_code))
+
+    assert laissees_passer == []
+
+
+# Le pendant du test précédent : sans lui, une garde posée trop haut, par exemple `AdminDep` sur
+# `/sites`, ne ferait échouer aucun test du dépôt.
+@pytest.mark.parametrize("role", ROLES, ids=IDS_DE_ROLE)
+async def test_a_role_at_or_above_the_minimum_is_never_refused_by_the_guard(
+    connecte: Callable[[Principal], None],
+    client: AsyncClient,
+    base_muette: None,
+    role: Role,
+) -> None:
+    connecte(principal(role))
+    refusees: list[tuple[str, str]] = []
+
+    for (methode, chemin), minimum in ROLE_MINIMUM.items():
+        if not has_at_least(role, minimum):
+            continue
+        response = await appelle(client, methode, chemin)
+        if motif_du_refus(response) == REFUS_DE_DROITS:
+            refusees.append((methode, chemin))
+
+    assert refusees == []
+
+
+async def test_a_pending_password_change_is_refused_on_every_guarded_route(
+    connecte: Callable[[Principal], None],
+    client: AsyncClient,
+    base_muette: None,
+) -> None:
+    connecte(principal(Role.ADMIN, must_change_password=True))
+    laissees_passer: list[tuple[str, str, int]] = []
+
+    for methode, chemin in ROLE_MINIMUM:
+        response = await appelle(client, methode, chemin)
+        if motif_du_refus(response) != REFUS_DE_MOT_DE_PASSE:
+            laissees_passer.append((methode, chemin, response.status_code))
+
+    assert laissees_passer == []
+
+
+@pytest.fixture
+async def comptes_par_role() -> AsyncIterator[dict[Role, str]]:
+    marque = uuid.uuid4().hex[:12]
+    hacheur = build_hasher(time_cost=1, memory_cost_kib=8192, parallelism=1, max_concurrency=2)
+    empreinte = await hacheur.hash(MOT_DE_PASSE)
+    adresses = {role: f"matrice-{marque}-{role.value}@enervision.fr" for role in ROLES}
+
+    async with get_session_factory()() as session:
+        depot = UserRepository(session)
+        for role, email in adresses.items():
+            await depot.create(email=email, password_hash=empreinte, role=role)
+        await session.commit()
+
+    yield adresses
+
+    async with get_session_factory()() as session:
+        await session.execute(
+            text("delete from app_user where email like :motif"), {"motif": f"matrice-{marque}-%"}
+        )
+        await session.commit()
+
+
+async def authentifie(client: AsyncClient, email: str) -> dict[str, str]:
+    reponse = await client.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']}"}
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("role", ROLES, ids=IDS_DE_ROLE)
+async def test_a_real_token_reaches_exactly_the_routes_of_its_rank(
+    comptes_par_role: dict[Role, str], client: AsyncClient, role: Role
+) -> None:
+    entetes = await authentifie(client, comptes_par_role[role])
+    ecarts: list[tuple[str, str, int, str]] = []
+
+    for (methode, chemin), minimum in ROLE_MINIMUM.items():
+        response = await appelle(client, methode, chemin, headers=entetes)
+        refuse = motif_du_refus(response) == REFUS_DE_DROITS
+        if refuse is has_at_least(role, minimum):
+            ecarts.append((methode, chemin, response.status_code, response.text[:120]))
+
+    assert ecarts == []
+
+
+# Contrainte : `operateur` n'ouvre aujourd'hui aucune route de plus que `lecteur`, faute d'écriture
+# métier dans l'API. Figer l'égalité rend la régression visible le jour où une route d'opérateur
+# arrive sans que `ROLE_MINIMUM` soit mis à jour.
+@pytest.mark.integration
+async def test_the_operator_rank_opens_nothing_more_than_the_reader_rank(
+    comptes_par_role: dict[Role, str], client: AsyncClient
+) -> None:
+    lecteur = await authentifie(client, comptes_par_role[Role.LECTEUR])
+    operateur = await authentifie(client, comptes_par_role[Role.OPERATEUR])
+    divergences: list[tuple[str, str]] = []
+
+    for methode, chemin in ROLE_MINIMUM:
+        cote_lecteur = await appelle(client, methode, chemin, headers=lecteur)
+        cote_operateur = await appelle(client, methode, chemin, headers=operateur)
+        if cote_lecteur.status_code != cote_operateur.status_code:
+            divergences.append((methode, chemin))
+
+    assert divergences == []
+
+
+# Piège : `/auth/logout-all` prend un `CurrentPrincipalDep` nu, donc elle échappe au gate
+# `must_change_password` que seul `require_role` applique. Comportement figé ici, pas corrigé.
+@pytest.mark.integration
+async def test_a_temporary_password_blocks_the_business_routes_but_not_logout_all(
+    client: AsyncClient,
+) -> None:
+    marque = uuid.uuid4().hex[:12]
+    email = f"matrice-{marque}-provisoire@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.ADMIN, must_change_password=True
+        )
+        await session.commit()
+
+    try:
+        entetes = await authentifie(client, email)
+        sites = await client.get("/api/v1/sites", headers=entetes)
+        identite = await client.get("/api/v1/auth/me", headers=entetes)
+        fermeture = await client.post("/api/v1/auth/logout-all", headers=entetes)
+
+        assert motif_du_refus(sites) == REFUS_DE_MOT_DE_PASSE
+        assert identite.status_code == 200
+        assert fermeture.status_code == 204
+    finally:
+        async with get_session_factory()() as session:
+            await session.execute(text("delete from app_user where email = :e"), {"e": email})
+            await session.commit()

apps/backend/tests/api/test_openapi.py

@@ -8,6 +8,7 @@ from typing import Any
 import pytest
 
 from app import cli
+from tests.api.acces import ROLE_MINIMUM
 
 METHODES = {"get", "post", "patch", "put", "delete"}
 
@@ -24,22 +25,11 @@ ORIGINE_VERIFIEE = {
 
 # Toute route derrière `require_role` (LecteurDep, OperateurDep, AdminDep) peut rendre 403 pour
 # `password_change_required`, pas seulement les routes `admin`.
-ROUTES_A_ROLE = {
-    ("GET", "/api/v1/users"),
-    ("POST", "/api/v1/users"),
-    ("PATCH", "/api/v1/users/{id}"),
-    ("POST", "/api/v1/users/{id}/password-reset"),
-    ("GET", "/api/v1/sites"),
-    ("GET", "/api/v1/sites/{site_id}"),
-    ("GET", "/api/v1/sites/{site_id}/current"),
-    ("GET", "/api/v1/alerts"),
-    ("GET", "/api/v1/recommendations"),
-    ("GET", "/api/v1/recommendations/{recommendation_id}"),
-    ("GET", "/api/v1/stats/summary"),
-    ("GET", "/api/v1/readings"),
-    ("GET", "/api/v1/sensors/status"),
-    ("GET", "/api/v1/predictions"),
-}
+# Piège : cette liste était recopiée ici, et deux de ses entrées portaient `{id}` là où le contrat
+# expose `{user_id}`. Elles ne correspondaient donc à aucune opération, et le test ci-dessous
+# passait au vert sans rien vérifier sur ces deux routes. Elle est maintenant dérivée, et
+# `test_every_declared_route_is_classified` interdit l'entrée morte.
+ROUTES_A_ROLE = frozenset(ROLE_MINIMUM)
 
 
 @pytest.fixture(scope="module")

apps/backend/tests/api/test_route_protection.py

@@ -1,6 +1,6 @@
 # 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.
+# `ROUTES_PUBLIQUES` dans `tests/api/acces.py`, 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.
@@ -11,58 +11,64 @@ 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"),
-        ("POST", "/api/v1/auth/forgot-password"),
-        # Protégée par le jeton dans le corps de la requête, pas par un `Principal` : aucune
-        # authentification préalable ne s'applique, c'est la validité du jeton qui tranche.
-        ("POST", "/api/v1/auth/reset-password"),
-        # Même raison : lecture seule, protégée par le jeton passé en paramètre, pas par un
-        # `Principal`. Le jeton est un secret de 256 bits, non brute-forçable.
-        ("GET", "/api/v1/auth/reset-password/validate"),
-        ("GET", "/metrics"),
-    }
+from tests.api.acces import (
+    ROLE_MINIMUM,
+    ROUTE_COOKIE,
+    ROUTES_PUBLIQUES,
+    ROUTES_SANS_ROLE,
+    Route,
+    chemin_concret,
+    routes_du_schema,
 )
 
-VALEURS_DE_SUBSTITUTION = "00000000-0000-0000-0000-000000000000"
 STATUTS_DE_REFUS = {401, 403}
+HORS_SCHEMA = {("GET", "/metrics")}
 
 
-def routes_declarees(app: FastAPI) -> list[tuple[str, str]]:
+def routes_declarees(app: FastAPI) -> list[Route]:
     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"}
-    ]
+    return routes_du_schema(schema)
 
 
-def routes_protegees(app: FastAPI) -> list[tuple[str, str]]:
+def routes_protegees(app: FastAPI) -> list[Route]:
     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")}
+    declarees = set(routes_declarees(app)) | HORS_SCHEMA
 
     inconnues = ROUTES_PUBLIQUES - declarees
 
     assert inconnues == set()
 
 
+# Sans lui, une route ajoutée sans être classée n'est vue par aucun test de rôle : elle hérite
+# du seul contrôle anonyme, et une garde posée au mauvais niveau passe inaperçue.
+def test_every_declared_route_is_classified(app: FastAPI) -> None:
+    classees = ROUTES_PUBLIQUES | ROUTE_COOKIE | ROUTES_SANS_ROLE | set(ROLE_MINIMUM)
+
+    non_classees = set(routes_declarees(app)) - classees
+    fantomes = classees - set(routes_declarees(app)) - HORS_SCHEMA
+
+    assert non_classees == set(), "classer la route dans tests/api/acces.py"
+    assert fantomes == set(), "entrée morte : la route n'existe plus sous ce chemin"
+
+
+def test_the_four_classes_of_routes_stay_disjoint() -> None:
+    classes = [ROUTES_PUBLIQUES, ROUTE_COOKIE, ROUTES_SANS_ROLE, frozenset(ROLE_MINIMUM)]
+
+    for rang, classe in enumerate(classes):
+        for autre in classes[rang + 1 :]:
+            assert classe & autre == frozenset()
+
+
 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={})
+        response = await client.request(methode, chemin_concret(chemin), json={})
         if response.status_code not in STATUTS_DE_REFUS:
             ouvertes.append((methode, chemin, response.status_code))
 
@@ -95,4 +101,5 @@ async def test_the_documentation_routes_are_public_by_design(
     app: FastAPI, client: AsyncClient, chemin: str
 ) -> None:
     response = await client.get(chemin)
+
     assert response.status_code == 200