Adriano/TieMeasureFlow
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix(csv): neutralizza injection di formule nei campi testuali export (#22)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
AdrianoDev
2026-09-21 22:33:32 +02:00
parent d1833ba203
commit 0dce2299d0
4 changed files with 416 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/api/routers/measurements.py
+3 -2
View File
@@ -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,
src/backend/services/production_export_service.py
+25 -5
View File
@@ -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),
src/backend/tests/test_measurements.py
+352
View File
@@ -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"
src/backend/tests/test_production_export_service.py
+36 -1
View File
@@ -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: