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

fix(files): un Maker non cancella statistics/reports/logo, GET ristretto (#21)

Browse Source 
DELETE /api/files/{path} non aveva nessun legame con ricetta/proprietario:
un Maker poteva cancellare qualunque file sotto uploads, compresi i CSV di
statistica di produzioni altrui. Ora: statistics/, reports/, logos/ sono
403 sempre (il logo si cambia da PUT /api/settings/logo); un file di
ricetta (task.file_path, subtask.image_path, recipe.image_path) e' 409 se
referenziato da una versione non corrente o che ha gia' misurazioni - un
file non referenziato resta cancellabile. GET di statistics/* e reports/*
richiede Metrologist, Supervisor o admin (prima bastava essere autenticati).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
AdrianoDev
2026-09-21 22:36:49 +02:00
parent 20957e517d
commit e8a8aeed66
2 changed files with 397 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/tests/test_files.py
+314 -1
View File
@@ -7,9 +7,13 @@ from unittest.mock import patch
import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.backend.models.orm.recipe import RecipeVersion
from src.backend.models.orm.task import RecipeTask
from src.backend.models.orm.user import User
from src.backend.tests.conftest import auth_headers
from src.backend.tests.conftest import auth_headers, create_test_recipe
class TestUploadFile:
@@ -134,6 +138,79 @@ class TestGetFile:
assert resp.status_code == 404
class TestGetRestrictedFile:
"""gitea#21: GET statistics/* and reports/* need Metrologist/Supervisor/admin."""
async def test_measurement_tec_forbidden_on_statistics(
self, client: AsyncClient, measurement_tec_user: User, tmp_path: Path
):
stats_dir = tmp_path / "statistics"
stats_dir.mkdir()
(stats_dir / "production_1_x.csv").write_text("a,b\n1,2\n")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.get(
"/api/files/statistics/production_1_x.csv",
headers=auth_headers(measurement_tec_user),
)
assert resp.status_code == 403
async def test_metrologist_allowed_on_statistics(
self, client: AsyncClient, metrologist_user: User, tmp_path: Path
):
stats_dir = tmp_path / "statistics"
stats_dir.mkdir()
(stats_dir / "production_1_x.csv").write_text("a,b\n1,2\n")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.get(
"/api/files/statistics/production_1_x.csv",
headers=auth_headers(metrologist_user),
)
assert resp.status_code == 200
async def test_admin_allowed_on_reports(
self, client: AsyncClient, admin_user: User, tmp_path: Path
):
reports_dir = tmp_path / "reports"
reports_dir.mkdir()
(reports_dir / "spc_1.pdf").write_bytes(b"%PDF-1.4")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.get(
"/api/files/reports/spc_1.pdf",
headers=auth_headers(admin_user),
)
assert resp.status_code == 200
async def test_maker_unrestricted_path_still_allowed(
self, client: AsyncClient, maker_user: User, tmp_path: Path
):
"""The restriction is scoped to statistics/reports, not every file."""
test_file = tmp_path / "1" / "1" / "drawing.png"
test_file.parent.mkdir(parents=True)
test_file.write_bytes(b"\x89PNG")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.get(
"/api/files/1/1/drawing.png",
headers=auth_headers(maker_user),
)
assert resp.status_code == 200
class TestDeleteFile:
"""DELETE /api/files/{file_path} tests."""
@@ -155,6 +232,242 @@ class TestDeleteFile:
assert resp.status_code == 204
assert not test_file.exists()
async def test_delete_unreferenced_recipe_image_allowed(
self, client: AsyncClient, maker_user: User, tmp_path: Path
):
"""A file under a recipe/version path but referenced nowhere is safe."""
test_file = tmp_path / "5" / "12" / "images" / "orphan.png"
test_file.parent.mkdir(parents=True)
test_file.write_bytes(b"\x89PNG")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.delete(
"/api/files/5/12/images/orphan.png",
headers=auth_headers(maker_user),
)
assert resp.status_code == 204
assert not test_file.exists()
class TestDeleteProtectedFolders:
"""gitea#21: statistics/, reports/, logos/ can never be deleted here."""
async def test_maker_delete_statistics_forbidden(
self, client: AsyncClient, maker_user: User, tmp_path: Path
):
stats_dir = tmp_path / "statistics"
stats_dir.mkdir()
test_file = stats_dir / "production_1_x.csv"
test_file.write_text("a,b\n1,2\n")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.delete(
"/api/files/statistics/production_1_x.csv",
headers=auth_headers(maker_user),
)
assert resp.status_code == 403
assert test_file.exists()
async def test_admin_delete_reports_forbidden(
self, client: AsyncClient, admin_user: User, tmp_path: Path
):
"""Even an admin cannot delete through this route: not just a role gate."""
reports_dir = tmp_path / "reports"
reports_dir.mkdir()
test_file = reports_dir / "spc_1.pdf"
test_file.write_bytes(b"%PDF-1.4")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.delete(
"/api/files/reports/spc_1.pdf",
headers=auth_headers(admin_user),
)
assert resp.status_code == 403
assert test_file.exists()
async def test_delete_logo_forbidden(
self, client: AsyncClient, maker_user: User, tmp_path: Path
):
"""The logo is only ever replaced via POST /api/settings/logo."""
logos_dir = tmp_path / "logos"
logos_dir.mkdir()
test_file = logos_dir / "company_logo.png"
test_file.write_bytes(b"\x89PNG")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.delete(
"/api/files/logos/company_logo.png",
headers=auth_headers(maker_user),
)
assert resp.status_code == 403
assert test_file.exists()
class TestDeleteReferencedRecipeFile:
"""gitea#21: a recipe file backing the current version, a historical
version, or measured data must survive a DELETE (409)."""
async def test_current_version_no_measurements_is_allowed(
self,
client: AsyncClient,
maker_user: User,
db_session: AsyncSession,
tmp_path: Path,
):
"""Still a draft (current version, nothing measured against it yet):
swapping the drawing out is exactly what editing a recipe does."""
recipe = await create_test_recipe(db_session, maker_user.id, code="FILE-409-A")
version = recipe.versions[0]
task = (await db_session.execute(
select(RecipeTask).where(RecipeTask.version_id == version.id)
)).scalar_one()
task.file_path = "recipe-a/1/drawing.png"
await db_session.commit()
test_file = tmp_path / "recipe-a" / "1" / "drawing.png"
test_file.parent.mkdir(parents=True)
test_file.write_bytes(b"\x89PNG")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.delete(
"/api/files/recipe-a/1/drawing.png",
headers=auth_headers(maker_user),
)
assert resp.status_code == 204
assert not test_file.exists()
async def test_current_version_with_measurements_is_409(
self,
client: AsyncClient,
maker_user: User,
db_session: AsyncSession,
tmp_path: Path,
):
"""Once something has been measured against this version, its drawing
must stay: the audit trail needs it to make sense of the numbers."""
recipe = await create_test_recipe(db_session, maker_user.id, code="FILE-409-D")
version = recipe.versions[0]
task = (await db_session.execute(
select(RecipeTask).where(RecipeTask.version_id == version.id)
)).scalar_one()
task.file_path = "recipe-d/1/drawing.png"
from src.backend.models.orm.measurement import Measurement
from src.backend.models.orm.task import RecipeSubtask
subtask = (await db_session.execute(
select(RecipeSubtask).where(RecipeSubtask.task_id == task.id)
)).scalar_one()
db_session.add(Measurement(
subtask_id=subtask.id,
version_id=version.id,
measured_by=maker_user.id,
value=10.0,
pass_fail="pass",
input_method="manual",
))
await db_session.commit()
test_file = tmp_path / "recipe-d" / "1" / "drawing.png"
test_file.parent.mkdir(parents=True)
test_file.write_bytes(b"\x89PNG")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.delete(
"/api/files/recipe-d/1/drawing.png",
headers=auth_headers(maker_user),
)
assert resp.status_code == 409
assert test_file.exists()
async def test_noncurrent_version_task_file_is_409(
self,
client: AsyncClient,
maker_user: User,
db_session: AsyncSession,
tmp_path: Path,
):
"""A file kept only for a superseded version is history, not garbage."""
recipe = await create_test_recipe(db_session, maker_user.id, code="FILE-409-B")
old_version = RecipeVersion(
recipe_id=recipe.id,
version_number=2,
is_current=False,
created_by=maker_user.id,
change_notes="Superseded",
)
db_session.add(old_version)
await db_session.flush()
old_task = RecipeTask(
version_id=old_version.id,
order_index=0,
title="Old drawing",
task_type="drawing",
file_path="recipe-b/2/old-drawing.png",
)
db_session.add(old_task)
await db_session.commit()
test_file = tmp_path / "recipe-b" / "2" / "old-drawing.png"
test_file.parent.mkdir(parents=True)
test_file.write_bytes(b"\x89PNG")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.delete(
"/api/files/recipe-b/2/old-drawing.png",
headers=auth_headers(maker_user),
)
assert resp.status_code == 409
assert test_file.exists()
async def test_recipe_cover_image_is_409(
self,
client: AsyncClient,
maker_user: User,
db_session: AsyncSession,
tmp_path: Path,
):
recipe = await create_test_recipe(db_session, maker_user.id, code="FILE-409-C")
recipe.image_path = "recipe-c/cover.png"
await db_session.commit()
test_file = tmp_path / "recipe-c" / "cover.png"
test_file.parent.mkdir(parents=True)
test_file.write_bytes(b"\x89PNG")
with patch("src.backend.api.routers.files.settings") as mock_settings:
mock_settings.upload_path = tmp_path
resp = await client.delete(
"/api/files/recipe-c/cover.png",
headers=auth_headers(maker_user),
)
assert resp.status_code == 409
assert test_file.exists()
class TestFilenameSanitization:
"""Path traversal protection tests."""