fix(etl): borne la réponse de l'API Mock avant écriture en base
L'API Mock est le seul item OWASP API10 du projet, et ce script en est le premier consommateur. Des quatre garde-fous exigés par la traçabilité OWASP, seul le timeout était en place. - plafonne la taille des réponses : MAX_SITES sites, au plus --limit mesures ; - borne chaque grandeur physique par PHYSICAL_BOUNDS, une valeur hors plage, d'un type inattendu, NaN ou infinie devenant NULL avec sa raison dans null_reasons et data_quality à degraded ; - ne recopie vers la base que les champs attendus, via build_site_row() et build_reading_row(), au lieu de passer les dictionnaires de l'API en paramètres SQL ; - écarte une data_quality que ck_reading_quality refuserait, plutôt que de faire échouer le lot entier ; - nomme la cible du ON CONFLICT, qui avalait jusqu'ici toute violation d'unicité, y compris celle de la clé primaire. raw_data conserve la réponse d'origine intacte : rien n'est perdu, seule son exploitation est bornée.
This commit is contained in:
Johan LEROY
@@ -1,3 +1,11 @@
|
||||
# Contrainte : la réponse de l'API Mock est une entrée hostile, pas une source de confiance.
|
||||
# Voir OWASP API10 dans docs/architecture/owasp-traceabilite.md. Rien de ce qu'elle renvoie
|
||||
# n'atteint la base sans passer par build_site_row() ou build_reading_row() : seuls les champs
|
||||
# attendus sont recopiés, les grandeurs physiques sont bornées par PHYSICAL_BOUNDS et la taille
|
||||
# des tableaux est plafonnée par MAX_SITES et par --limit. Une valeur hors bornes devient NULL
|
||||
# et laisse sa trace dans null_reasons plutôt que de lever : le mock émet des anomalies par
|
||||
# construction, et raw_data conserve de toute façon la réponse d'origine intacte.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
@@ -14,6 +22,25 @@ from app.core.config import get_settings
|
||||
|
||||
SOURCE_HISTORY = "api_history"
|
||||
|
||||
MAX_SITES = 100
|
||||
|
||||
MAX_LIMIT = 1000
|
||||
|
||||
# Les quatre seules valeurs que la contrainte ck_reading_quality accepte.
|
||||
ACCEPTED_QUALITIES = frozenset({"good", "partial", "degraded", "critical"})
|
||||
|
||||
PHYSICAL_BOUNDS: dict[str, tuple[float, float]] = {
|
||||
"consumption_kw": (0.0, 100_000.0),
|
||||
"consumption_kwh": (0.0, 100_000.0),
|
||||
"voltage_v": (0.0, 1_000.0),
|
||||
"current_a": (0.0, 10_000.0),
|
||||
"power_factor": (0.0, 1.0),
|
||||
"temperature_celsius": (-90.0, 60.0),
|
||||
"humidity_percent": (0.0, 100.0),
|
||||
}
|
||||
|
||||
CAPACITY_BOUNDS = (0.0, 100_000.0)
|
||||
|
||||
|
||||
def create_mock_api_client() -> httpx.AsyncClient:
|
||||
settings = get_settings()
|
||||
@@ -31,6 +58,53 @@ def create_mock_api_client() -> httpx.AsyncClient:
|
||||
)
|
||||
|
||||
|
||||
def read_text(payload: dict[str, Any], key: str) -> str:
|
||||
value = payload.get(key)
|
||||
|
||||
if not isinstance(value, str) or not value:
|
||||
raise ValueError(f"Champ {key} absent ou invalide dans la réponse de l'API Mock.")
|
||||
|
||||
return value
|
||||
|
||||
|
||||
def optional_text(value: Any) -> str | None:
|
||||
return value if isinstance(value, str) else None
|
||||
|
||||
|
||||
def coerce_measure(
|
||||
value: Any,
|
||||
bounds: tuple[float, float],
|
||||
) -> float | None:
|
||||
if isinstance(value, bool) or not isinstance(value, int | float):
|
||||
return None
|
||||
|
||||
lower, upper = bounds
|
||||
|
||||
# Écarte aussi NaN et les infinis, qu'aucune comparaison de bornes ne retient.
|
||||
return float(value) if lower <= value <= upper else None
|
||||
|
||||
|
||||
def resolve_quality(
|
||||
value: Any,
|
||||
rejected: list[str],
|
||||
) -> str | None:
|
||||
quality = value if isinstance(value, str) and value in ACCEPTED_QUALITIES else None
|
||||
|
||||
if rejected:
|
||||
return "critical" if quality == "critical" else "degraded"
|
||||
|
||||
return quality
|
||||
|
||||
|
||||
def resolve_null_reasons(
|
||||
value: Any,
|
||||
rejected: list[str],
|
||||
) -> list[str]:
|
||||
reported = [str(reason) for reason in value] if isinstance(value, list) else []
|
||||
|
||||
return reported + rejected
|
||||
|
||||
|
||||
async def fetch_sites(
|
||||
client: httpx.AsyncClient,
|
||||
) -> list[dict[str, Any]]:
|
||||
@@ -43,14 +117,32 @@ async def fetch_sites(
|
||||
if not isinstance(payload, list):
|
||||
raise ValueError("La réponse /api/v1/sites doit être une liste.")
|
||||
|
||||
if len(payload) > MAX_SITES:
|
||||
raise ValueError(f"La réponse /api/v1/sites dépasse le plafond de {MAX_SITES} sites.")
|
||||
|
||||
return payload
|
||||
|
||||
|
||||
def build_site_row(
|
||||
site: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"site_id": read_text(site, "site_id"),
|
||||
"site_type": read_text(site, "site_type"),
|
||||
"site_name": read_text(site, "site_name"),
|
||||
"location": optional_text(site.get("location")),
|
||||
"capacity_kw": coerce_measure(site.get("capacity_kw"), CAPACITY_BOUNDS),
|
||||
"status": optional_text(site.get("status")),
|
||||
}
|
||||
|
||||
|
||||
async def upsert_sites(
|
||||
connection: AsyncConnection,
|
||||
sites: list[dict[str, Any]],
|
||||
) -> None:
|
||||
if not sites:
|
||||
rows = [build_site_row(site) for site in sites]
|
||||
|
||||
if not rows:
|
||||
return
|
||||
|
||||
await connection.execute(
|
||||
@@ -81,7 +173,7 @@ async def upsert_sites(
|
||||
status = EXCLUDED.status
|
||||
"""
|
||||
),
|
||||
sites,
|
||||
rows,
|
||||
)
|
||||
|
||||
|
||||
@@ -90,7 +182,7 @@ async def fetch_readings(
|
||||
site_id: str,
|
||||
start_time: datetime,
|
||||
end_time: datetime,
|
||||
limit: int = 1000,
|
||||
limit: int = MAX_LIMIT,
|
||||
) -> list[dict[str, Any]]:
|
||||
response = await client.get(
|
||||
"/api/v1/readings",
|
||||
@@ -109,30 +201,36 @@ async def fetch_readings(
|
||||
if not isinstance(payload, list):
|
||||
raise ValueError("La réponse /api/v1/readings doit être une liste.")
|
||||
|
||||
if len(payload) > limit:
|
||||
raise ValueError(f"La réponse /api/v1/readings dépasse la limite demandée de {limit}.")
|
||||
|
||||
return payload
|
||||
|
||||
|
||||
def build_reading_row(
|
||||
reading: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
timestamp = datetime.fromisoformat(reading["timestamp"].replace("Z", "+00:00"))
|
||||
measures: dict[str, float | None] = {}
|
||||
rejected: list[str] = []
|
||||
|
||||
for name, bounds in PHYSICAL_BOUNDS.items():
|
||||
received = reading.get(name)
|
||||
measures[name] = coerce_measure(received, bounds)
|
||||
|
||||
if received is not None and measures[name] is None:
|
||||
rejected.append(f"out_of_physical_bounds:{name}")
|
||||
|
||||
return {
|
||||
"site_id": reading["site_id"],
|
||||
"timestamp": timestamp,
|
||||
"site_id": read_text(reading, "site_id"),
|
||||
"timestamp": parse_datetime(read_text(reading, "timestamp")),
|
||||
"source": SOURCE_HISTORY,
|
||||
"dataset_id": None,
|
||||
"consumption_kw": reading.get("consumption_kw"),
|
||||
"consumption_kwh": reading.get("consumption_kwh"),
|
||||
**measures,
|
||||
"consumption_euros": None,
|
||||
"voltage_v": reading.get("voltage_v"),
|
||||
"current_a": reading.get("current_a"),
|
||||
"power_factor": reading.get("power_factor"),
|
||||
"temperature_celsius": reading.get("temperature_celsius"),
|
||||
"humidity_percent": reading.get("humidity_percent"),
|
||||
"solar_irradiance_wm2": None,
|
||||
"is_working_hours": None,
|
||||
"data_quality": reading.get("data_quality"),
|
||||
"null_reasons": reading.get("null_reasons"),
|
||||
"data_quality": resolve_quality(reading.get("data_quality"), rejected),
|
||||
"null_reasons": resolve_null_reasons(reading.get("null_reasons"), rejected),
|
||||
"imputed_values": None,
|
||||
"imputation_method": None,
|
||||
"raw_data": json.dumps(
|
||||
@@ -142,6 +240,8 @@ def build_reading_row(
|
||||
}
|
||||
|
||||
|
||||
# Le conflit vise l'index unique uq_reading_source plutôt que la table entière : sans cible
|
||||
# nommée, DO NOTHING avalerait aussi une violation de clé primaire.
|
||||
READING_INSERT = text(
|
||||
"""
|
||||
INSERT INTO reading (
|
||||
@@ -186,7 +286,8 @@ READING_INSERT = text(
|
||||
:imputation_method,
|
||||
CAST(:raw_data AS jsonb)
|
||||
)
|
||||
ON CONFLICT DO NOTHING
|
||||
ON CONFLICT (site_id, timestamp, source, (coalesce(dataset_id, 0)))
|
||||
DO NOTHING
|
||||
"""
|
||||
)
|
||||
|
||||
@@ -213,7 +314,7 @@ async def import_mock_api_history(
|
||||
all_readings: list[dict[str, Any]] = []
|
||||
|
||||
for site in sites:
|
||||
site_id = site["site_id"]
|
||||
site_id = read_text(site, "site_id")
|
||||
|
||||
readings = await fetch_readings(
|
||||
client=client,
|
||||
@@ -281,7 +382,7 @@ def parse_args() -> argparse.Namespace:
|
||||
parser.add_argument(
|
||||
"--limit",
|
||||
type=int,
|
||||
default=1000,
|
||||
default=MAX_LIMIT,
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
@@ -295,8 +396,8 @@ def parse_args() -> argparse.Namespace:
|
||||
def main() -> None:
|
||||
args = parse_args()
|
||||
|
||||
if args.limit < 1 or args.limit > 1000:
|
||||
raise ValueError("--limit doit être compris entre 1 et 1000.")
|
||||
if args.limit < 1 or args.limit > MAX_LIMIT:
|
||||
raise ValueError(f"--limit doit être compris entre 1 et {MAX_LIMIT}.")
|
||||
|
||||
if args.start_time >= args.end_time:
|
||||
raise ValueError("--start-time doit être antérieur à --end-time.")
|
||||
|
||||
@@ -13,10 +13,12 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
import app.etl.mock_api_import as mock_api_import
|
||||
from app.etl.mock_api_import import (
|
||||
MAX_SITES,
|
||||
READING_INSERT,
|
||||
SOURCE_HISTORY,
|
||||
build_reading_batch,
|
||||
build_reading_row,
|
||||
build_site_row,
|
||||
fetch_readings,
|
||||
fetch_sites,
|
||||
upsert_sites,
|
||||
@@ -73,6 +75,26 @@ async def test_fetch_sites_returns_sites() -> None:
|
||||
assert sites[0]["site_type"] == "office"
|
||||
|
||||
|
||||
async def test_fetch_sites_rejects_non_list_response() -> None:
|
||||
def handler(request: Request) -> Response:
|
||||
return Response(
|
||||
status_code=200,
|
||||
json={"unexpected": "payload"},
|
||||
)
|
||||
|
||||
transport = MockTransport(handler)
|
||||
|
||||
async with AsyncClient(
|
||||
transport=transport,
|
||||
base_url="https://mock.test",
|
||||
) as client:
|
||||
with pytest.raises(
|
||||
ValueError,
|
||||
match="La réponse /api/v1/sites doit être une liste",
|
||||
):
|
||||
await fetch_sites(client)
|
||||
|
||||
|
||||
async def test_fetch_readings_sends_expected_query_parameters() -> None:
|
||||
captured_params: dict[str, str] = {}
|
||||
|
||||
@@ -576,6 +598,148 @@ def test_main_runs_import(
|
||||
)
|
||||
|
||||
|
||||
async def test_fetch_sites_rejects_a_response_above_the_cap() -> None:
|
||||
def handler(request: Request) -> Response:
|
||||
return Response(
|
||||
status_code=200,
|
||||
json=[make_site() for _ in range(MAX_SITES + 1)],
|
||||
)
|
||||
|
||||
transport = MockTransport(handler)
|
||||
|
||||
async with AsyncClient(
|
||||
transport=transport,
|
||||
base_url="https://mock.test",
|
||||
) as client:
|
||||
with pytest.raises(
|
||||
ValueError,
|
||||
match=f"dépasse le plafond de {MAX_SITES} sites",
|
||||
):
|
||||
await fetch_sites(client)
|
||||
|
||||
|
||||
async def test_fetch_readings_rejects_a_response_above_the_requested_limit() -> None:
|
||||
def handler(request: Request) -> Response:
|
||||
return Response(
|
||||
status_code=200,
|
||||
json=[make_reading(), make_reading(), make_reading()],
|
||||
)
|
||||
|
||||
transport = MockTransport(handler)
|
||||
|
||||
async with AsyncClient(
|
||||
transport=transport,
|
||||
base_url="https://mock.test",
|
||||
) as client:
|
||||
with pytest.raises(
|
||||
ValueError,
|
||||
match="dépasse la limite demandée de 2",
|
||||
):
|
||||
await fetch_readings(
|
||||
client=client,
|
||||
site_id="SITE001",
|
||||
start_time=datetime.fromisoformat("2024-06-15T12:00:00"),
|
||||
end_time=datetime.fromisoformat("2024-06-15T13:00:00"),
|
||||
limit=2,
|
||||
)
|
||||
|
||||
|
||||
def test_build_reading_row_neutralises_values_outside_physical_bounds() -> None:
|
||||
reading = make_reading()
|
||||
|
||||
reading["power_factor"] = 42.0
|
||||
reading["temperature_celsius"] = 1e30
|
||||
reading["humidity_percent"] = -1.0
|
||||
|
||||
row = build_reading_row(reading)
|
||||
|
||||
assert row["power_factor"] is None
|
||||
assert row["temperature_celsius"] is None
|
||||
assert row["humidity_percent"] is None
|
||||
|
||||
assert row["null_reasons"] == [
|
||||
"out_of_physical_bounds:power_factor",
|
||||
"out_of_physical_bounds:temperature_celsius",
|
||||
"out_of_physical_bounds:humidity_percent",
|
||||
]
|
||||
|
||||
assert row["data_quality"] == "degraded"
|
||||
|
||||
assert json.loads(row["raw_data"])["power_factor"] == 42.0
|
||||
|
||||
|
||||
def test_build_reading_row_rejects_a_measure_that_is_not_a_number() -> None:
|
||||
reading = make_reading()
|
||||
|
||||
reading["consumption_kw"] = "87.34"
|
||||
|
||||
row = build_reading_row(reading)
|
||||
|
||||
assert row["consumption_kw"] is None
|
||||
assert "out_of_physical_bounds:consumption_kw" in row["null_reasons"]
|
||||
|
||||
|
||||
def test_build_reading_row_drops_a_quality_the_database_refuses() -> None:
|
||||
reading = make_reading()
|
||||
|
||||
reading["data_quality"] = "unknown"
|
||||
|
||||
row = build_reading_row(reading)
|
||||
|
||||
assert row["data_quality"] is None
|
||||
|
||||
|
||||
def test_build_reading_row_requires_an_identifier() -> None:
|
||||
reading = make_reading()
|
||||
|
||||
del reading["site_id"]
|
||||
|
||||
with pytest.raises(
|
||||
ValueError,
|
||||
match="Champ site_id absent ou invalide",
|
||||
):
|
||||
build_reading_row(reading)
|
||||
|
||||
|
||||
def test_build_site_row_keeps_only_the_expected_columns() -> None:
|
||||
site = make_site()
|
||||
|
||||
site["unexpected"] = "valeur hostile"
|
||||
site["capacity_kw"] = -5.0
|
||||
site["status"] = 12
|
||||
|
||||
row = build_site_row(site)
|
||||
|
||||
assert set(row) == {
|
||||
"site_id",
|
||||
"site_type",
|
||||
"site_name",
|
||||
"location",
|
||||
"capacity_kw",
|
||||
"status",
|
||||
}
|
||||
|
||||
assert row["capacity_kw"] is None
|
||||
assert row["status"] is None
|
||||
|
||||
|
||||
async def test_upsert_sites_sends_only_the_expected_columns() -> None:
|
||||
connection = AsyncMock()
|
||||
|
||||
site = make_site()
|
||||
site["unexpected"] = "valeur hostile"
|
||||
|
||||
await upsert_sites(
|
||||
connection,
|
||||
[site],
|
||||
)
|
||||
|
||||
rows = connection.execute.await_args.args[1]
|
||||
|
||||
assert "unexpected" not in rows[0]
|
||||
assert rows[0]["site_id"] == "SITE001"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_reading_insert_is_idempotent(
|
||||
session: AsyncSession,
|
||||
@@ -620,3 +784,51 @@ async def test_reading_insert_is_idempotent(
|
||||
assert result.scalar_one() == 1
|
||||
|
||||
await session.rollback()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_out_of_bounds_reading_is_stored_neutralised(
|
||||
session: AsyncSession,
|
||||
) -> None:
|
||||
reading = make_reading()
|
||||
reading["power_factor"] = 42.0
|
||||
|
||||
row = build_reading_row(reading)
|
||||
|
||||
connection = await session.connection()
|
||||
|
||||
await upsert_sites(
|
||||
connection,
|
||||
[make_site()],
|
||||
)
|
||||
|
||||
await session.execute(
|
||||
READING_INSERT,
|
||||
[row],
|
||||
)
|
||||
|
||||
result = await session.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT power_factor, data_quality, null_reasons, raw_data ->> 'power_factor'
|
||||
FROM reading
|
||||
WHERE site_id = :site_id
|
||||
AND timestamp = :timestamp
|
||||
AND source = :source
|
||||
"""
|
||||
),
|
||||
{
|
||||
"site_id": row["site_id"],
|
||||
"timestamp": row["timestamp"],
|
||||
"source": row["source"],
|
||||
},
|
||||
)
|
||||
|
||||
stored = result.one()
|
||||
|
||||
await session.rollback()
|
||||
|
||||
assert stored[0] is None
|
||||
assert stored[1] == "degraded"
|
||||
assert stored[2] == ["out_of_physical_bounds:power_factor"]
|
||||
assert stored[3] == "42.0"
|
||||
|
||||
@@ -41,6 +41,7 @@ lecture seule ; plusieurs lignes resteront à compléter une fois les endpoints
|
||||
| En-têtes `nosniff`, `DENY`, `no-referrer`, et `no-store` sur les routes d'authentification | `app/api/middleware.py` | A05 |
|
||||
| Refus de rétrograder ou désactiver le dernier administrateur actif | `app/services/user.py` | A04 Insecure Design |
|
||||
| Amorçage du premier administrateur hors dépôt, mot de passe jamais dans `argv` ni dans Git | `app/cli.py` | A02, A05 |
|
||||
| Réponse de l'API Mock bornée avant écriture : timeout, plafond de sites et de mesures, bornes physiques par grandeur, recopie des seuls champs attendus | `app/etl/mock_api_import.py` | API10 Unsafe Consumption of APIs |
|
||||
| CI bloquante : format, lint avec règles Bandit, typage strict, tests avec seuil de couverture | `.github/workflows/backend.yml` | A06 Vulnerable and Outdated Components |
|
||||
|
||||
Note sur A06 : le jeu de règles `S` de ruff, déjà actif dans `pyproject.toml`, est le portage des
|
||||
@@ -53,7 +54,7 @@ règles Bandit. Ajouter Bandit à la CI serait redondant, contrairement à ce qu
|
||||
| **API1 Broken Object Level Authorization** | **ouvert** | Les rôles sont globaux, il n'y a pas de portée par site : `GET /sites/{site_id}` et `GET /recommendations/{recommendation_id}` répondent à tout compte `lecteur` pour n'importe quel site ou recommandation, sans vérifier une affectation compte-site qui n'existe pas encore. Un opérateur du site A pourra agir sur le site B dès que les endpoints d'écriture métier existeront. Correctif prévu : table d'affectation compte-site, contrôle d'appartenance dans la même dépendance que le contrôle de rôle. |
|
||||
| **API4, lectures de séries temporelles** | **partiel** | `GET /readings` plafonne la fenêtre temporelle (90 jours) et la pagination (`limit` ≤ 2000), voir plus haut. Reste ouvert : pagination en `limit`/`offset` simple plutôt qu'en curseur (un `offset` élevé sur une fenêtre dense reste coûteux), et aucun `statement_timeout` au niveau de la connexion pour borner une requête individuelle si les plafonds au-dessus s'avéraient insuffisants. |
|
||||
| **API8 Security Misconfiguration, transport** | **ouvert** | Pas de TLS, donc ni HSTS, ni cookie `Secure` réellement posé en production. Ils appartiennent au terminateur TLS, qui n'existe pas. |
|
||||
| **API10 Unsafe Consumption of APIs** | **ouvert, et spécifique à ce projet** | L'API Mock de l'école n'a aucune authentification, tourne en HTTP clair sur le réseau de l'école, et expose un endpoint mutatif à quiconque. Sa réponse doit être traitée comme une entrée hostile : bornes physiques, taille de tableau plafonnée, timeout, et frontière d'anti-corruption. La conséquence la plus sérieuse n'est pas la fausse alerte, c'est l'empoisonnement du jeu d'entraînement du modèle de prédiction. |
|
||||
| **API10 Unsafe Consumption of APIs** | **partiel, et spécifique à ce projet** | L'API Mock de l'école n'a aucune authentification, tourne en HTTP clair sur le réseau de l'école, et expose un endpoint mutatif à quiconque. Sa réponse est traitée comme une entrée hostile par `app/etl/mock_api_import.py`, son seul consommateur à ce jour : les quatre garde-fous attendus sont en place, voir la ligne correspondante plus haut. Reste ouvert : le plafond de taille s'applique après désérialisation de la réponse, borner le corps HTTP lui-même demanderait une lecture en flux ; et `APP_MOCK_API_BASE_URL` n'impose pas `https`, donc les identifiants Basic partiraient en clair sur une URL en `http`. La conséquence la plus sérieuse n'est pas la fausse alerte, c'est l'empoisonnement du jeu d'entraînement du modèle de prédiction. |
|
||||
| **A08 Software and Data Integrity Failures** | **partiel** | La CI vérifie le code mais n'analyse ni les dépendances ni les images. `.terraform.lock.hcl` reste ignoré par git, ce qui contredit une chaîne d'approvisionnement maîtrisée. |
|
||||
| **A10 Server-Side Request Forgery** | **sans objet aujourd'hui** | Aucune URL sortante n'est pilotée par une donnée utilisateur. Le jour où l'adresse d'une source devient un champ de configuration, il faudra une liste blanche de schémas et d'hôtes, sans suivi de redirection. |
|
||||
| **Cantonnement des accès ETL et ML** | **dette assumée** | Le compte applicatif porte l'identité, le rôle PostgreSQL porterait le cantonnement. Voir ADR 0003. |
|
||||
|
||||
Reference in New Issue
Block a user