Files
ENI-projet-piscine/apps/backend/tests/api/test_hardening.py
T
Johan LEROY dc952d13aa feat(monitoring): supervise l'API, la base et l'hôte avec Prometheus et Grafana
L'API exposait /metrics, mais aucun collecteur ne le lisait : monitoring/ ne contenait que des
.gitkeep.

Sous le profil Compose `monitoring` : prometheus, alertmanager, grafana, postgres-exporter,
node-exporter et cadvisor. Tous ont un mem_limit, pour environ 700 Mo au total sur la VM de 8 Go,
et leurs interfaces n'écoutent que sur 127.0.0.1. Le profil est actif en prod via
COMPOSE_PROFILES, donc à chaque déploiement, et se lance à la demande ailleurs
(make monitoring-up).

- Neuf règles d'alerte (API, base, hôte, cibles). Chacune a un cas dans les tests joués par
  `promtool test rules`, en CI comme par make monitoring-check.
- Alertmanager route les alertes par courriel vers Mailpit ; un critical masque le warning de la
  même cible.
- Grafana est provisionné : sources Prometheus et TimescaleDB, et trois tableaux de bord (API,
  données et dérive du modèle, infrastructure).
- Le rôle PostgreSQL `supervision` est en lecture seule sur les seules tables métier
  (db/roles/supervision.sql), posé par make db-ensure-supervision et par stack-up quand le
  profil est actif.
- Le jeton de /metrics passe à Prometheus en secret Compose (APP_METRICS_TOKEN) ;
  provision-host.sh génère ce secret et les deux autres.

Backend :
- un APP_METRICS_TOKEN vide vaut absent ;
- les sondes de santé ne comptent plus dans les métriques ;
- seaux de latence fins autour de 500 ms ;
- un registre Prometheus par application, sans quoi toute application créée après la première
  (dans les tests) ne mesurait rien.

Réf : #26
2026-09-23 09:41:12 +02:00

126 lines
4.3 KiB
Python

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_an_empty_metrics_token_means_no_token() -> None:
assert (await interroge({"metrics_token": ""}, "/metrics")).status_code == 200
async def test_metrics_ignore_health_probes_but_count_business_routes(client: AsyncClient) -> None:
await client.get("/api/v1/health/live")
await client.get("/api/v1/sites")
exposition = (await client.get("/metrics")).text
assert 'handler="/api/v1/health/live"' not in exposition
assert 'handler="/api/v1/sites"' in exposition
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"]