diff --git a/.secrets.baseline b/.secrets.baseline index 49f352545..f78c6ac53 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -290,13 +290,6 @@ "is_verified": false, "line_number": 141 }, - { - "type": "Hex High Entropy String", - "filename": ".semgrep/tenant-scope-allowlist.yml", - "hashed_secret": "fc54ea750dde43f8070720f70d68f7b7f25dde2c", - "is_verified": false, - "line_number": 147 - }, { "type": "Hex High Entropy String", "filename": ".semgrep/tenant-scope-allowlist.yml", @@ -3754,5 +3747,5 @@ } ] }, - "generated_at": "2026-10-25T02:01:01Z" + "generated_at": "2026-10-25T07:05:37Z" } diff --git a/.semgrep/tenant-scope-allowlist.yml b/.semgrep/tenant-scope-allowlist.yml index ac2210804..13c7fd237 100644 --- a/.semgrep/tenant-scope-allowlist.yml +++ b/.semgrep/tenant-scope-allowlist.yml @@ -144,7 +144,7 @@ baseline: scope: _query_tenant_events rule: sql-without-tenant-predicate target: fsma.cte_events,fsma.cte_kdes - qhash: "87bb48c02e1970fb" + qhash: "c9b46a2dd14590f8" # pragma: allowlist secret reason: "WHERE list seeded with e.tenant_id = :tenant_id; verified scoped at runtime (cte_kdes joined on the scoped event)." - file: services/ingestion/app/incident_router.py scope: list_incidents diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py index 9414afd31..278def6f3 100644 --- a/services/ingestion/app/epcis_export.py +++ b/services/ingestion/app/epcis_export.py @@ -13,7 +13,7 @@ from datetime import datetime, timezone from typing import Optional -from fastapi import APIRouter, Depends, Query +from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, Field from shared.database import get_db_safe @@ -32,10 +32,26 @@ ) +# Hard cap on events buffered into one synchronous export. Exceeding it +# returns HTTP 413 rather than silently dropping the oldest events (SEL-506), +# matching the compliance FDA spreadsheet export (services/compliance +# routes.py ``_MAX_EXPORT_EVENTS``). +_MAX_EXPORT_EVENTS = 50_000 + + +class ExportStorageUnavailable(RuntimeError): + """The event store could not be read; never report this as "no data".""" + + def _query_tenant_events( tenant_id: str, lot_code: str | None, date_from: str | None, date_to: str | None ) -> list[dict]: - """Query real CTE events from fsma.cte_events for export.""" + """Query real CTE events from fsma.cte_events for export. + + Returns at most ``_MAX_EXPORT_EVENTS + 1`` rows so callers can detect + an over-cap export. Raises :class:`ExportStorageUnavailable` on any + database error instead of returning an empty list (SEL-508). + """ db = get_db_safe() try: where = ["e.tenant_id = :tenant_id"] @@ -80,18 +96,40 @@ def _query_tenant_events( LEFT JOIN fsma.cte_kdes k_temp ON k_temp.cte_event_id = e.id AND k_temp.kde_key = 'temperature_celsius' WHERE {" AND ".join(where)} ORDER BY e.event_timestamp DESC - LIMIT 5000 + LIMIT :row_limit """), - params, + {**params, "row_limit": _MAX_EXPORT_EVENTS + 1}, ).fetchall() return [dict(row._mapping) for row in rows] except Exception as exc: - logger.warning("db_query_failed_for_export", extra={"error": str(exc)}) - return [] + logger.error("db_query_failed_for_export", extra={"error": str(exc)}) + raise ExportStorageUnavailable(str(exc)) from exc finally: db.close() +def _load_export_rows(tenant_id: str, request: "ExportRequest") -> list[dict]: + """Load rows for an export, failing loudly instead of returning partial data. + + 503 when storage is unavailable (never "No data yet"), 413 when the + result would exceed ``_MAX_EXPORT_EVENTS`` (never a silent truncation). + """ + try: + rows = _query_tenant_events(tenant_id, request.lot_code, request.date_from, request.date_to) + except ExportStorageUnavailable as exc: + raise HTTPException(status_code=503, detail="Event storage unavailable; export not generated") from exc + if len(rows) > _MAX_EXPORT_EVENTS: + logger.warning( + "export_exceeded_hard_cap", + extra={"tenant_id": tenant_id, "cap": _MAX_EXPORT_EVENTS}, + ) + raise HTTPException( + status_code=413, + detail=(f"Export exceeds the {_MAX_EXPORT_EVENTS:,}-event hard cap; narrow your date range or lot code"), + ) + return rows + + _CTE_TO_BIZSTEP = { "receiving": "urn:epcglobal:cbv:bizstep:receiving", "shipping": "urn:epcglobal:cbv:bizstep:shipping", @@ -186,7 +224,7 @@ async def export_epcis( now = datetime.now(timezone.utc) tenant_id = ctx.tenant_id - rows = _query_tenant_events(tenant_id, request.lot_code, request.date_from, request.date_to) + rows = _load_export_rows(tenant_id, request) data_source = "tenant" if rows else "empty" event_list = [] @@ -284,7 +322,7 @@ async def export_fda( now = datetime.now(timezone.utc) tenant_id = ctx.tenant_id - rows = _query_tenant_events(tenant_id, request.lot_code, request.date_from, request.date_to) + rows = _load_export_rows(tenant_id, request) data_source = "tenant" if rows else "empty" if rows: diff --git a/services/ingestion/tests/test_epcis_export_1342.py b/services/ingestion/tests/test_epcis_export_1342.py index b6f588157..f60db1b8b 100644 --- a/services/ingestion/tests/test_epcis_export_1342.py +++ b/services/ingestion/tests/test_epcis_export_1342.py @@ -103,11 +103,22 @@ def test_no_rows_returns_empty_list(self, monkeypatch: pytest.MonkeyPatch) -> No assert _query_tenant_events("t", None, None, None) == [] - def test_swallow_exception_returns_empty_list(self, monkeypatch: pytest.MonkeyPatch) -> None: - session = _FakeSession(execute_queue=[RuntimeError()]) + def test_db_error_raises_instead_of_empty_list(self, monkeypatch: pytest.MonkeyPatch) -> None: + # SEL-508: a storage failure must never look like an empty tenant. + session = _FakeSession(execute_queue=[RuntimeError("db down")]) monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session) - result = _query_tenant_events("t", None, None, None) - assert result == [] + with pytest.raises(epcis_export.ExportStorageUnavailable): + _query_tenant_events("t", None, None, None) + assert session.closed + + def test_query_is_bounded_by_cap_plus_one_not_5000(self, monkeypatch: pytest.MonkeyPatch) -> None: + # SEL-506: the old LIMIT 5000 silently dropped the oldest events. + session = _FakeSession(execute_queue=[[]]) + monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session) + _query_tenant_events("t", None, None, None) + sql, params = session.calls[0][1] + assert "LIMIT 5000" not in sql + assert params["row_limit"] == epcis_export._MAX_EXPORT_EVENTS + 1 # --------------------------------------------------------------------------- @@ -260,3 +271,42 @@ def test_lists_formats(self, client: TestClient) -> None: assert "name" in fmt assert "content_type" in fmt assert "retailers" in fmt + + +def _raise_storage(*_a: Any, **_k: Any) -> list[dict]: + raise epcis_export.ExportStorageUnavailable("db down") + + +class TestExportFailsLoudly: + """SEL-506 / SEL-508: partial or failed reads are never reported as complete.""" + + @pytest.mark.parametrize("path", ["/api/v1/export/epcis", "/api/v1/export/fda"]) + def test_storage_error_returns_503_not_no_data_yet( + self, client: TestClient, monkeypatch: pytest.MonkeyPatch, path: str + ) -> None: + session = _FakeSession(execute_queue=[RuntimeError("db down")]) + monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session) + response = client.post(path, json={"tenant_id": "t"}) + assert response.status_code == 503 + assert "No data yet" not in response.text + assert response.headers.get("X-RegEngine-Data-Source") != "empty" + + def test_5001_events_export_5001(self, client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None: + rows = [_event_row() for _ in range(5001)] + session = _FakeSession(execute_queue=[rows]) + monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session) + response = client.post("/api/v1/export/epcis", json={"tenant_id": "t"}) + assert response.status_code == 200 + assert len(response.json()["epcisBody"]["eventList"]) == 5001 + assert response.headers["X-RegEngine-Events-Count"] == "5001" + + @pytest.mark.parametrize("path", ["/api/v1/export/epcis", "/api/v1/export/fda"]) + def test_over_cap_returns_413_not_truncated( + self, client: TestClient, monkeypatch: pytest.MonkeyPatch, path: str + ) -> None: + monkeypatch.setattr(epcis_export, "_MAX_EXPORT_EVENTS", 3) + session = _FakeSession(execute_queue=[[_event_row() for _ in range(4)]]) + monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session) + response = client.post(path, json={"tenant_id": "t"}) + assert response.status_code == 413 + assert session.calls[0][1][1]["row_limit"] == 4