Merge pull request #108 from ineszang/test/matrice-acces-roles

test(backend): matrice d'accès par rôle et CI d'intégration
This commit is contained in:
Johan LEROY
2026-09-18 14:25:40 +02:00
committed by GitHub
7 changed files with 509 additions and 56 deletions
+60
View File
@@ -57,6 +57,66 @@ jobs:
- name: Tests et couverture - name: Tests et couverture
run: uv run pytest --cov-fail-under=85 run: uv run pytest --cov-fail-under=85
# Piège : l'image est celle de docker-compose.yml, pas une image `postgres` nue. La première
# migration (`5353c0e4f094`) échoue volontairement si l'extension TimescaleDB manque, et un
# écart d'image entre la CI et le poste rendrait ce job vert sur une base qui n'est pas la nôtre.
integration:
name: Tests exigeant une base
runs-on: ubuntu-latest
defaults:
run:
working-directory: apps/backend
services:
db:
image: timescale/timescaledb-ha:pg17
env:
POSTGRES_USER: enervision
POSTGRES_PASSWORD: change_me
POSTGRES_DB: enervision_test
ports:
- "5433:5432"
options: >-
--health-cmd "pg_isready -U enervision -d enervision_test"
--health-interval 10s
--health-timeout 5s
--health-retries 12
--health-start-period 40s
env:
DATABASE_URL: postgresql+asyncpg://enervision:change_me@localhost:5433/enervision_test
APP_SECRET_KEY: secret-de-test-assez-long-pour-le-validateur
PGPASSWORD: change_me
steps:
- name: Récupère le dépôt
uses: actions/checkout@v4
- name: Installe uv
uses: astral-sh/setup-uv@v5
with:
enable-cache: true
cache-dependency-glob: apps/backend/uv.lock
- name: Installe l'interpréteur déclaré par .python-version
run: uv python install
- name: Synchronise les dépendances sans dévier du verrou
run: uv sync --all-groups --frozen
# Sur le poste, c'est db/init/110-test-database.sql qui pose l'extension. Ce fichier n'est
# pas monté ici, et sans lui `alembic upgrade head` s'arrête sur la garde de la révision 1.
- name: Active TimescaleDB sur la base de test
run: psql -h localhost -p 5433 -U enervision -d enervision_test -c "CREATE EXTENSION IF NOT EXISTS timescaledb"
- name: Applique les migrations
run: uv run alembic upgrade head
# `-m` en ligne de commande écrase celui d'`addopts`. La couverture est désactivée : ce job
# ne joue qu'une partie de la suite, son taux n'aurait aucun sens face au seuil de 85 %.
- name: Tests d'intégration
run: uv run pytest -m integration --no-cov
security-audit: security-audit:
name: Audit des dépendances name: Audit des dépendances
runs-on: ubuntu-latest runs-on: ubuntu-latest
+21 -3
View File
@@ -123,6 +123,11 @@ async def test_repository_reads_back_what_it_wrote(session: AsyncSession) -> Non
defaut, ce qui garde `make check` jouable sans Docker. Tout autre marqueur doit etre defaut, ce qui garde `make check` jouable sans Docker. Tout autre marqueur doit etre
declare dans `pyproject.toml` : `--strict-markers` refuse les marqueurs inconnus. declare dans `pyproject.toml` : `--strict-markers` refuse les marqueurs inconnus.
Ces tests ne sont pas pour autant facultatifs : le job `integration` de
`.github/workflows/backend.yml` monte un service TimescaleDB, applique les migrations et
les joue a chaque poussee. Un test `integration` casse donc la CI comme un autre. En local,
`make db-up` puis `make test-integration`.
## Couverture ## Couverture
Les branches sont mesurees, pas seulement les lignes. Le seuil de 85 % ne s'applique Les branches sont mesurees, pas seulement les lignes. Le seuil de 85 % ne s'applique
@@ -142,14 +147,27 @@ uv run pytest tests/api/test_health.py # un seul fichier
uv run pytest -k readiness # par motif de nom uv run pytest -k readiness # par motif de nom
``` ```
## Trois fichiers à connaître avant de toucher à l'authentification ## Quatre fichiers à connaître avant de toucher à l'authentification
`tests/api/acces.py` porte la classification des routes du contrat, en quatre ensembles :
`ROUTES_PUBLIQUES`, `ROUTE_COOKIE`, `ROUTES_SANS_ROLE` et la table `ROLE_MINIMUM`. Ce n'est pas
un fichier de test, c'est la référence que les trois autres confrontent au comportement observé.
**Toute route ajoutée doit y être classée** : `test_every_declared_route_is_classified` échoue
sinon, et échoue aussi sur une entrée qui ne correspond plus à aucune route.
`tests/api/test_route_protection.py` interroge réellement chaque route sans identifiant et `tests/api/test_route_protection.py` interroge réellement chaque route sans identifiant et
échoue si l'une d'elles répond autre chose qu'un 401 ou un 403. Il n'inspecte pas l'arbre de échoue si l'une d'elles répond autre chose qu'un 401 ou un 403. Il n'inspecte pas l'arbre de
dépendances : celui-ci n'est accessible que par l'API privée de FastAPI, et surtout une route dépendances : celui-ci 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. **Rendre une route publique impose peut porter la bonne dépendance tout en répondant quand même. **Rendre une route publique impose
donc de modifier la liste `ROUTES_PUBLIQUES` de ce fichier**, ce qui apparaît en clair dans la donc de modifier `ROUTES_PUBLIQUES` dans `acces.py`**, ce qui apparaît en clair dans la diff
diff d'une pull request. d'une pull request.
`tests/api/test_matrice_acces.py` croise chaque route gardée avec chacun des trois rôles, dans
les deux sens : un rôle insuffisant reçoit un 403 `Droits insuffisants`, un rôle suffisant ne le
reçoit jamais. Le second sens est ce qui rend visible une garde posée trop haut, par exemple
`AdminDep` sur une route de lecture. La même matrice est rejouée sous `integration` avec de vrais
jetons, donc en traversant le décodage du JWT et la relecture du compte en base, que
`dependency_overrides` court-circuite.
`tests/services/test_auth.py` donne au faux hacheur un **compteur d'appels**. C'est ce qui rend `tests/services/test_auth.py` donne au faux hacheur un **compteur d'appels**. C'est ce qui rend
possibles les deux assertions qui prouvent la conception, et qu'aucune autre forme de test possibles les deux assertions qui prouvent la conception, et qu'aucune autre forme de test
+86
View File
@@ -0,0 +1,86 @@
# 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/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"}
]
@@ -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()
+6 -15
View File
@@ -8,6 +8,7 @@ from typing import Any
import pytest import pytest
from app import cli from app import cli
from tests.api.acces import ROLE_MINIMUM
METHODES = {"get", "post", "patch", "put", "delete"} METHODES = {"get", "post", "patch", "put", "delete"}
@@ -24,21 +25,11 @@ ORIGINE_VERIFIEE = {
# Toute route derrière `require_role` (LecteurDep, OperateurDep, AdminDep) peut rendre 403 pour # Toute route derrière `require_role` (LecteurDep, OperateurDep, AdminDep) peut rendre 403 pour
# `password_change_required`, pas seulement les routes `admin`. # `password_change_required`, pas seulement les routes `admin`.
ROUTES_A_ROLE = { # Piège : cette liste était recopiée ici, et deux de ses entrées portaient `{id}` là où le contrat
("GET", "/api/v1/users"), # expose `{user_id}`. Elles ne correspondaient donc à aucune opération, et le test ci-dessous
("POST", "/api/v1/users"), # passait au vert sans rien vérifier sur ces deux routes. Elle est maintenant dérivée, et
("PATCH", "/api/v1/users/{id}"), # `test_every_declared_route_is_classified` interdit l'entrée morte.
("POST", "/api/v1/users/{id}/password-reset"), ROUTES_A_ROLE = frozenset(ROLE_MINIMUM)
("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"),
}
@pytest.fixture(scope="module") @pytest.fixture(scope="module")
+37 -30
View File
@@ -1,6 +1,6 @@
# Ce test est le garde-fou de l'autorisation : rendre une route publique oblige à modifier # 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 # `ROUTES_PUBLIQUES` dans `tests/api/acces.py`, ce qui apparaît en clair dans la diff d'une pull
# demande une justification au relecteur. # request et demande une justification au relecteur.
# Pourquoi : il interroge réellement chaque route sans jeton au lieu d'inspecter l'arbre de # 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 # 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. # peut porter la bonne dépendance tout en répondant quand même.
@@ -11,58 +11,64 @@ import pytest
from fastapi import FastAPI from fastapi import FastAPI
from httpx import AsyncClient from httpx import AsyncClient
ROUTES_PUBLIQUES = frozenset( from tests.api.acces import (
{ ROLE_MINIMUM,
("GET", "/api/v1/health/live"), ROUTE_COOKIE,
("GET", "/api/v1/health/ready"), ROUTES_PUBLIQUES,
("POST", "/api/v1/auth/login"), ROUTES_SANS_ROLE,
# Sans cookie, la déconnexion ne fait rien et répond 204 : elle est idempotente. Route,
("POST", "/api/v1/auth/logout"), chemin_concret,
("POST", "/api/v1/auth/forgot-password"), routes_du_schema,
# 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"),
}
) )
VALEURS_DE_SUBSTITUTION = "00000000-0000-0000-0000-000000000000"
STATUTS_DE_REFUS = {401, 403} 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() schema: dict[str, Any] = app.openapi()
return [ return routes_du_schema(schema)
(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]]: def routes_protegees(app: FastAPI) -> list[Route]:
return [route for route in routes_declarees(app) if route not in ROUTES_PUBLIQUES] 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: 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 inconnues = ROUTES_PUBLIQUES - declarees
assert inconnues == set() 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( async def test_every_route_rejects_an_anonymous_caller_unless_explicitly_public(
app: FastAPI, client: AsyncClient app: FastAPI, client: AsyncClient
) -> None: ) -> None:
ouvertes: list[tuple[str, str, int]] = [] ouvertes: list[tuple[str, str, int]] = []
for methode, chemin in routes_protegees(app): for methode, chemin in routes_protegees(app):
concret = chemin.replace("{user_id}", VALEURS_DE_SUBSTITUTION) response = await client.request(methode, chemin_concret(chemin), json={})
response = await client.request(methode, concret, json={})
if response.status_code not in STATUTS_DE_REFUS: if response.status_code not in STATUTS_DE_REFUS:
ouvertes.append((methode, chemin, response.status_code)) 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 app: FastAPI, client: AsyncClient, chemin: str
) -> None: ) -> None:
response = await client.get(chemin) response = await client.get(chemin)
assert response.status_code == 200 assert response.status_code == 200
+20 -8
View File
@@ -155,10 +155,12 @@ Deux fichiers d'environnement, deux usages : `.env` à la racine alimente `docke
Les codes de la dernière colonne sont ceux que le schéma **déclare**, et le fichier Les codes de la dernière colonne sont ceux que le schéma **déclare**, et le fichier
`openapi.json` versionné interdit qu'ils divergent de ce que les routes rendent. `openapi.json` versionné interdit qu'ils divergent de ce que les routes rendent.
**Quatre routes seulement sont publiques** : les deux sondes, `/auth/login` et `/auth/logout`. **Sept routes du contrat sont publiques** : les deux sondes, `/auth/login`, `/auth/logout`,
`/auth/forgot-password` et les deux routes de réinitialisation, qui portent leur autorisation dans
le jeton à usage unique plutôt que dans un `Principal`.
`tests/api/test_route_protection.py` interroge réellement chaque autre route sans identifiant et `tests/api/test_route_protection.py` interroge réellement chaque autre route sans identifiant et
échoue si l'une d'elles répond autre chose qu'un 401 ou un 403. Rendre une route publique impose échoue si l'une d'elles répond autre chose qu'un 401 ou un 403. Rendre une route publique impose
donc de modifier la liste dans ce fichier de test. donc de modifier `ROUTES_PUBLIQUES` dans `tests/api/acces.py`.
`GET /sites` et `GET /sites/{site_id}` sont la première route métier, et le gabarit repris pour `GET /sites` et `GET /sites/{site_id}` sont la première route métier, et le gabarit repris pour
`GET /alerts` puis pour les suivantes (`dataset`, `prediction`) : les quatre couches `GET /alerts` puis pour les suivantes (`dataset`, `prediction`) : les quatre couches
@@ -273,11 +275,16 @@ Checklist pour toute nouvelle route sur le gabarit `sites`/`alerts`/`recommendat
au niveau de l'`include_router()` du routeur, `REPONSE_VALIDATION` et les codes locaux au niveau de l'`include_router()` du routeur, `REPONSE_VALIDATION` et les codes locaux
(404, 409, ...) directement sur l'endpoint qui les rend. (404, 409, ...) directement sur l'endpoint qui les rend.
2. Décrire son tag dans `TAGS`. 2. Décrire son tag dans `TAGS`.
3. Si elle passe par `require_role` (`LecteurDep`/`OperateurDep`/`AdminDep`), l'ajouter à 3. **La classer dans `tests/api/acces.py`** : `ROLE_MINIMUM` avec son rôle minimum si elle passe
`ROUTES_A_ROLE` dans `tests/api/test_openapi.py`. Si elle passe par `require_trusted_origin`, par `require_role` (`LecteurDep`/`OperateurDep`/`AdminDep`), `ROUTES_SANS_ROLE` si elle se
l'ajouter à `ORIGINE_VERIFIEE`. **Ces deux listes sont maintenues à la main, pas dérivées** : contente de `CurrentPrincipalDep`, `ROUTES_PUBLIQUES` si elle est ouverte. L'oubli n'est plus
une route oubliée n'y est pas détectée automatiquement. silencieux : `test_every_declared_route_is_classified` échoue sur une route non classée comme
4. `make openapi`, puis `uv run pytest tests/api/test_openapi.py`. sur une entrée qui ne correspond plus à aucune route. `ROUTES_A_ROLE` de `test_openapi.py` en
est dérivée, et `test_matrice_acces.py` vérifie le niveau réellement monté.
4. Si elle passe par `require_trusted_origin`, l'ajouter à `ORIGINE_VERIFIEE` dans
`tests/api/test_openapi.py`. **Cette liste-là reste maintenue à la main.**
5. `make openapi`, puis `uv run pytest tests/api/test_openapi.py tests/api/test_route_protection.py
tests/api/test_matrice_acces.py`.
## Sécurité ## Sécurité
@@ -328,9 +335,14 @@ Le reste, par ordre de surface :
Conventions, gabarits et arborescence : [`apps/backend/TESTING.md`](../../apps/backend/TESTING.md). Conventions, gabarits et arborescence : [`apps/backend/TESTING.md`](../../apps/backend/TESTING.md).
Trois fichiers méritent d'être connus avant de toucher à l'authentification : Quatre fichiers méritent d'être connus avant de toucher à l'authentification :
- `tests/api/acces.py` : la classification des routes, `ROUTES_PUBLIQUES` et `ROLE_MINIMUM` en
tête. Ce n'est pas un test, c'est la référence que les deux suivants confrontent au
comportement observé.
- `tests/api/test_route_protection.py` : le garde-fou de l'autorisation, décrit plus haut. - `tests/api/test_route_protection.py` : le garde-fou de l'autorisation, décrit plus haut.
- `tests/api/test_matrice_acces.py` : chaque route gardée croisée avec chacun des trois rôles,
dans les deux sens, puis rejouée sous `integration` avec de vrais jetons.
- `tests/services/test_auth.py` : le faux hacheur y porte un compteur d'appels, ce qui permet les - `tests/services/test_auth.py` : le faux hacheur y porte un compteur d'appels, ce qui permet les
deux assertions qui prouvent le design, à savoir un appel quand l'adresse est inconnue et zéro deux assertions qui prouvent le design, à savoir un appel quand l'adresse est inconnue et zéro
appel quand la limite est atteinte. appel quand la limite est atteinte.