From 0dce2299d052dfd073ad0aff8b76e915cae796a7 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Mon, 21 Sep 2026 22:33:32 +0200 Subject: [PATCH] fix(csv): neutralizza injection di formule nei campi testuali export (#22) lot_number, serial_number (export manuale e di produzione), station.code, recipe.code, subtask.description finivano nel CSV cosi' come arrivavano: un lotto '=HYPERLINK(...)' apriva un link vivo in Excel/LibreOffice. Nuova neutralise_csv_cell() in production_export_service.py, importata da measurements.py: se il campo (spazi iniziali esclusi) comincia con =, +, -, @, TAB o CR antepone un apice, cosi' resta testo letterale. Applicata solo ai campi testuali, mai ai numeri. Co-Authored-By: Claude Sonnet 5 --- src/backend/api/routers/measurements.py | 5 +- .../services/production_export_service.py | 30 +- src/backend/tests/test_measurements.py | 352 ++++++++++++++++++ .../tests/test_production_export_service.py | 37 +- 4 files changed, 416 insertions(+), 8 deletions(-) diff --git a/src/backend/api/routers/measurements.py b/src/backend/api/routers/measurements.py index 15f9fdf..39d26b3 100644 --- a/src/backend/api/routers/measurements.py +++ b/src/backend/api/routers/measurements.py @@ -28,6 +28,7 @@ from src.backend.models.api.measurement import ( ) from src.backend.services import auth_service, measurement_service from src.backend.services.measurement_service import save_measurement +from src.backend.services.production_export_service import neutralise_csv_cell router = APIRouter(prefix="/api/measurements", tags=["measurements"]) @@ -342,8 +343,8 @@ async def export_measurements_csv( value_str, m.pass_fail, deviation_str, - m.lot_number or "", - m.serial_number or "", + neutralise_csv_cell(m.lot_number or ""), + neutralise_csv_cell(m.serial_number or ""), m.input_method, m.measured_at.isoformat(), m.synced_to_csv, diff --git a/src/backend/services/production_export_service.py b/src/backend/services/production_export_service.py index 1c92160..ed1d02b 100644 --- a/src/backend/services/production_export_service.py +++ b/src/backend/services/production_export_service.py @@ -38,6 +38,26 @@ EXPORT_SUBDIR = "statistics" _VALID_CSV_DELIMITERS = {",", ";", "\t", "|"} _FALLBACK_CSV_DELIMITER = ";" +# A cell starting with one of these (after leading spaces) is read as a formula +# by Excel/LibreOffice when the CSV is opened — CSV/formula injection (OWASP). +_DANGEROUS_CSV_PREFIXES = ("=", "+", "-", "@", "\t", "\r") + + +def neutralise_csv_cell(value: str) -> str: + """Defuse a formula-injection payload in a text CSV cell. + + Prepends a single quote when the value, stripped of leading spaces, starts + with =, +, -, @, TAB or CR: this keeps spreadsheet software from evaluating + it as a formula while the value stays valid, literal CSV text. Only for + free-text fields (lot, serial, descriptions, names, codes) — never call + this on numeric cells, the quote would make them text. + """ + stripped = value.lstrip(" ") + if stripped and stripped[0] in _DANGEROUS_CSV_PREFIXES: + return "'" + value + return value + + HEADER = [ "production_run_id", "station_code", @@ -127,15 +147,15 @@ async def build_statistics_csv(db: AsyncSession, run: ProductionRun) -> str: for measurement, subtask in rows: writer.writerow([ run.id, - station.code if station else "", - recipe.code if recipe else "", + neutralise_csv_cell(station.code) if station else "", + neutralise_csv_cell(recipe.code) if recipe else "", measurement.version_id, - measurement.lot_number or "", - measurement.serial_number or "", + neutralise_csv_cell(measurement.lot_number or ""), + neutralise_csv_cell(measurement.serial_number or ""), measurement.id, measurement.subtask_id, subtask.marker_number if subtask else "", - subtask.description if subtask else "", + neutralise_csv_cell(subtask.description) if subtask else "", number(subtask.nominal) if subtask else "", number(measurement.value), number(measurement.deviation), diff --git a/src/backend/tests/test_measurements.py b/src/backend/tests/test_measurements.py index b387890..4c11711 100644 --- a/src/backend/tests/test_measurements.py +++ b/src/backend/tests/test_measurements.py @@ -1,11 +1,15 @@ """Tests for measurements router (/api/measurements).""" +import csv + import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession +from src.backend.models.orm.station import Station from src.backend.models.orm.user import User from src.backend.models.orm.recipe import RecipeVersion from src.backend.models.orm.task import RecipeTask, RecipeSubtask +from src.backend.services import auth_service from src.backend.tests.conftest import auth_headers, create_test_recipe @@ -223,6 +227,281 @@ class TestMeasurementPassFail: assert resp.json()["pass_fail"] == "fail" +class TestMeasurementValueBounds: + """value rejects NaN/Infinity and out-of-DECIMAL(12,6)-scale numbers (#20).""" + + @pytest.mark.parametrize("bad_value", [1e300, 1_000_000]) + async def test_out_of_scale_value_is_422( + self, + client: AsyncClient, + measurement_tec_user: User, + db_session: AsyncSession, + bad_value: float, + ): + recipe = await create_test_recipe(db_session, measurement_tec_user.id) + subtask_id, version_id = await _get_subtask_and_version( + client, measurement_tec_user, recipe.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={ + "subtask_id": subtask_id, + "version_id": version_id, + "value": bad_value, + }, + ) + assert resp.status_code == 422 + + @pytest.mark.parametrize("raw_token", ["NaN", "Infinity", "-Infinity"]) + async def test_nan_and_infinity_are_422( + self, + client: AsyncClient, + measurement_tec_user: User, + db_session: AsyncSession, + raw_token: str, + ): + """NaN/Infinity are not valid JSON: httpx's own encoder refuses to send + them via ``json=``, so the raw body is built by hand here - exactly + what a client that is not httpx could still put on the wire.""" + recipe = await create_test_recipe(db_session, measurement_tec_user.id) + subtask_id, version_id = await _get_subtask_and_version( + client, measurement_tec_user, recipe.id + ) + + body = ( + '{"subtask_id": %d, "version_id": %d, "value": %s}' + % (subtask_id, version_id, raw_token) + ) + resp = await client.post( + "/api/measurements/", + headers={**auth_headers(measurement_tec_user), "Content-Type": "application/json"}, + content=body, + ) + assert resp.status_code == 422 + + async def test_value_is_rounded_before_the_verdict( + self, + client: AsyncClient, + measurement_tec_user: User, + db_session: AsyncSession, + ): + """10.5000004 rounds to 10.5 (=utl) first, so it is not a fail.""" + recipe = await create_test_recipe(db_session, measurement_tec_user.id) + subtask_id, version_id = await _get_subtask_and_version( + client, measurement_tec_user, recipe.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={ + "subtask_id": subtask_id, + "version_id": version_id, + "value": 10.5000004, + }, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["value"] == pytest.approx(10.5) + # 10.5 is > uwl(10.3) but not > utl(10.5): warning, not fail. + assert data["pass_fail"] == "warning" + + +class TestSaveMeasurementCoherence: + """save_measurement checks quote/version/run coherence before insert (#5).""" + + async def _station(self, db_session, user_id, code): + station = Station( + code=code, name=f"Stazione {code}", active=True, created_by=user_id, + ) + db_session.add(station) + await db_session.commit() + await db_session.refresh(station) + return station + + async def _supervisor(self, db_session, username): + user = await auth_service.create_user( + db_session, username=username, password="TurnoPwd1", + display_name="Capo Turno", roles=["Supervisor"], + ) + await db_session.commit() + return user, "TurnoPwd1" + + async def test_quota_of_another_recipe_version_is_422( + self, client: AsyncClient, measurement_tec_user: User, db_session: AsyncSession, + ): + recipe_a = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-A") + recipe_b = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-B") + subtask_a, _version_a = await _get_subtask_and_version( + client, measurement_tec_user, recipe_a.id + ) + _subtask_b, version_b = await _get_subtask_and_version( + client, measurement_tec_user, recipe_b.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={"subtask_id": subtask_a, "version_id": version_b, "value": 10.0}, + ) + assert resp.status_code == 422 + + async def test_measurement_of_other_recipe_in_run_is_422( + self, client: AsyncClient, measurement_tec_user: User, admin_user: User, + db_session: AsyncSession, + ): + recipe_a = await create_test_recipe( + db_session, measurement_tec_user.id, code="COH-RUNA" + ) + recipe_b = await create_test_recipe( + db_session, measurement_tec_user.id, code="COH-RUNB" + ) + station = await self._station(db_session, admin_user.id, "ST-COH1") + + run = (await client.post( + "/api/production-runs", + headers=auth_headers(measurement_tec_user), + json={"station_code": station.code, "recipe_id": recipe_a.id}, + )).json() + + subtask_b, version_b = await _get_subtask_and_version( + client, measurement_tec_user, recipe_b.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={ + "subtask_id": subtask_b, + "version_id": version_b, + "value": 10.0, + "production_run_id": run["id"], + }, + ) + assert resp.status_code == 422 + + async def test_measurement_on_closed_run_is_409( + self, client: AsyncClient, measurement_tec_user: User, admin_user: User, + db_session: AsyncSession, + ): + recipe = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-CLOSED") + station = await self._station(db_session, admin_user.id, "ST-COH2") + run = (await client.post( + "/api/production-runs", + headers=auth_headers(measurement_tec_user), + json={"station_code": station.code, "recipe_id": recipe.id}, + )).json() + supervisor, password = await self._supervisor(db_session, "capo-coh-close") + await client.post( + f"/api/production-runs/{run['id']}/close", + headers=auth_headers(measurement_tec_user), + json={ + "supervisor_username": supervisor.username, + "supervisor_password": password, + }, + ) + subtask_id, version_id = await _get_subtask_and_version( + client, measurement_tec_user, recipe.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={ + "subtask_id": subtask_id, + "version_id": version_id, + "value": 10.0, + "production_run_id": run["id"], + }, + ) + assert resp.status_code == 409 + + async def test_measurement_on_paused_run_is_409( + self, client: AsyncClient, measurement_tec_user: User, admin_user: User, + db_session: AsyncSession, + ): + recipe = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-PAUSED") + station = await self._station(db_session, admin_user.id, "ST-COH3") + run = (await client.post( + "/api/production-runs", + headers=auth_headers(measurement_tec_user), + json={"station_code": station.code, "recipe_id": recipe.id}, + )).json() + supervisor, password = await self._supervisor(db_session, "capo-coh-pause") + await client.post( + f"/api/production-runs/{run['id']}/pause", + headers=auth_headers(measurement_tec_user), + json={ + "supervisor_username": supervisor.username, + "supervisor_password": password, + }, + ) + subtask_id, version_id = await _get_subtask_and_version( + client, measurement_tec_user, recipe.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={ + "subtask_id": subtask_id, + "version_id": version_id, + "value": 10.0, + "production_run_id": run["id"], + }, + ) + assert resp.status_code == 409 + + async def test_valid_measurement_in_matching_run_is_200( + self, client: AsyncClient, measurement_tec_user: User, admin_user: User, + db_session: AsyncSession, + ): + recipe = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-OK") + station = await self._station(db_session, admin_user.id, "ST-COH4") + run = (await client.post( + "/api/production-runs", + headers=auth_headers(measurement_tec_user), + json={"station_code": station.code, "recipe_id": recipe.id}, + )).json() + subtask_id, version_id = await _get_subtask_and_version( + client, measurement_tec_user, recipe.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={ + "subtask_id": subtask_id, + "version_id": version_id, + "value": 10.0, + "production_run_id": run["id"], + }, + ) + assert resp.status_code == 200 + + async def test_measurement_on_unknown_run_is_404( + self, client: AsyncClient, measurement_tec_user: User, db_session: AsyncSession, + ): + recipe = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-404") + subtask_id, version_id = await _get_subtask_and_version( + client, measurement_tec_user, recipe.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={ + "subtask_id": subtask_id, + "version_id": version_id, + "value": 10.0, + "production_run_id": 999999, + }, + ) + assert resp.status_code == 404 + + class TestListMeasurements: """GET /api/measurements/ tests.""" @@ -366,3 +645,76 @@ class TestMeasurementStatistics: # deviation = 10.2 - nominal(10.0) = 0.2 assert data["deviation"] is not None assert abs(data["deviation"] - 0.2) < 0.001 + + +class TestExportCsv: + """GET /api/measurements/export/csv tests (gitea#22).""" + + async def test_lot_number_formula_is_neutralised( + self, + client: AsyncClient, + measurement_tec_user: User, + db_session: AsyncSession, + ): + """A lot_number starting with '=' must not reach the CSV as a live formula (#22).""" + recipe = await create_test_recipe(db_session, measurement_tec_user.id) + subtask_id, version_id = await _get_subtask_and_version( + client, measurement_tec_user, recipe.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={ + "subtask_id": subtask_id, + "version_id": version_id, + "value": 10.0, + "lot_number": '=HYPERLINK("http://x","y")', + }, + ) + assert resp.status_code == 200 + + export_resp = await client.get( + "/api/measurements/export/csv", + headers=auth_headers(measurement_tec_user), + ) + assert export_resp.status_code == 200 + rows = list(csv.reader(export_resp.text.splitlines())) + header, row = rows[0], rows[1] + lot_cell = row[header.index("lot_number")] + # Quoted with a leading ' so a spreadsheet reads it as literal text, + # not the raw formula that would otherwise be live in Excel/LibreOffice. + assert lot_cell == '\'=HYPERLINK("http://x","y")' + + async def test_negative_value_stays_numeric( + self, + client: AsyncClient, + measurement_tec_user: User, + db_session: AsyncSession, + ): + """Neutralisation must never touch numeric cells like the measured value.""" + recipe = await create_test_recipe(db_session, measurement_tec_user.id) + subtask_id, version_id = await _get_subtask_and_version( + client, measurement_tec_user, recipe.id + ) + + resp = await client.post( + "/api/measurements/", + headers=auth_headers(measurement_tec_user), + json={ + "subtask_id": subtask_id, + "version_id": version_id, + "value": -1.25, + }, + ) + assert resp.status_code == 200 + + export_resp = await client.get( + "/api/measurements/export/csv", + headers=auth_headers(measurement_tec_user), + ) + assert export_resp.status_code == 200 + rows = list(csv.reader(export_resp.text.splitlines())) + header, row = rows[0], rows[1] + value_cell = row[header.index("value")] + assert value_cell == "-1.250000" diff --git a/src/backend/tests/test_production_export_service.py b/src/backend/tests/test_production_export_service.py index 5e2ddfe..8e30bce 100644 --- a/src/backend/tests/test_production_export_service.py +++ b/src/backend/tests/test_production_export_service.py @@ -2,11 +2,46 @@ - CSV format defenses (gitea#23): a dirty csv_delimiter read from the DB must never make close_run fail with a csv.writer TypeError. +- Formula-injection neutralisation (gitea#22): text cells starting with + =, +, -, @, TAB or CR must be defused before they reach a CSV cell. """ from sqlalchemy.ext.asyncio import AsyncSession from src.backend.models.orm.setting import SystemSetting -from src.backend.services.production_export_service import _csv_format +from src.backend.services.production_export_service import ( + _csv_format, + neutralise_csv_cell, +) + + +class TestNeutraliseCsvCell: + """gitea#22: formula-injection prefixes must be quoted, plain text untouched.""" + + def test_equals_prefix_is_quoted(self): + assert neutralise_csv_cell('=HYPERLINK("http://x","y")') == \ + '\'=HYPERLINK("http://x","y")' + + def test_plus_minus_at_prefixes_are_quoted(self): + assert neutralise_csv_cell("+1;2").startswith("'") + assert neutralise_csv_cell("-cmd|calc").startswith("'") + assert neutralise_csv_cell("@SUM(A1)").startswith("'") + + def test_tab_and_cr_prefixes_are_quoted(self): + assert neutralise_csv_cell("\t=1+1").startswith("'") + assert neutralise_csv_cell("\r=1+1").startswith("'") + + def test_leading_spaces_before_dangerous_prefix_are_quoted(self): + assert neutralise_csv_cell(" =cmd").startswith("'") + + def test_plain_text_is_untouched(self): + assert neutralise_csv_cell("LOT-2026-001") == "LOT-2026-001" + assert neutralise_csv_cell("") == "" + + def test_negative_number_as_text_is_still_quoted(self): + """The function only sees text cells; callers must not pass numeric + cells to it (that responsibility is the caller's, see -1.25 test in + test_measurements.py). Given a string, a leading '-' is neutralised.""" + assert neutralise_csv_cell("-1.25").startswith("'") class TestCsvFormatDefense: