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/api/routers/files.py
+83 -2
View File
@@ -6,9 +6,15 @@ from pathlib import Path
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
from fastapi.responses import FileResponse
from PIL import Image
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.backend.config import settings
from src.backend.database import get_db
from src.backend.api.middleware.api_key import get_current_user, require_maker
from src.backend.models.orm.measurement import Measurement
from src.backend.models.orm.recipe import Recipe, RecipeVersion
from src.backend.models.orm.task import RecipeSubtask, RecipeTask
from src.backend.models.orm.user import User
logger = logging.getLogger(__name__)
@@ -20,6 +26,59 @@ ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
ALLOWED_PDF_TYPE = "application/pdf"
ALLOWED_TYPES = ALLOWED_IMAGE_TYPES | {ALLOWED_PDF_TYPE}
# Top-level folders nothing may DELETE through this router: statistics/reports are
# audit records, the logo is only ever replaced via POST /api/settings/logo.
_UNDELETABLE_PREFIXES = ("statistics/", "reports/", "logos/")
# GET under these needs a role that is meant to see production data, not just
# any authenticated user (a Maker or MeasurementTec has no business reading
# every production's statistics/report file by guessing its name).
_RESTRICTED_GET_PREFIXES = ("statistics/", "reports/")
def _can_read_restricted(user: User) -> bool:
return user.is_admin or user.has_role("Metrologist") or user.has_role("Supervisor")
async def _is_recipe_file_protected(db: AsyncSession, relative_path: str) -> bool:
"""True when `relative_path` (uploads-relative, forward slashes) must not be
deleted: it is a recipe's cover image, or a task/subtask file belonging to a
version that is either not the current one (kept for history) or already has
measurements against it (the audit trail needs it to make sense of them).
A path not referenced anywhere is safe to delete (orphaned upload).
"""
recipe_hit = (await db.execute(
select(Recipe.id).where(Recipe.image_path == relative_path)
)).scalar_one_or_none()
if recipe_hit is not None:
return True
version_id = (await db.execute(
select(RecipeTask.version_id).where(RecipeTask.file_path == relative_path)
)).scalar_one_or_none()
if version_id is None:
version_id = (await db.execute(
select(RecipeTask.version_id)
.join(RecipeSubtask, RecipeSubtask.task_id == RecipeTask.id)
.where(RecipeSubtask.image_path == relative_path)
)).scalar_one_or_none()
if version_id is None:
return False
version = (await db.execute(
select(RecipeVersion).where(RecipeVersion.id == version_id)
)).scalar_one_or_none()
if version is None or not version.is_current:
return True
has_measurements = (await db.execute(
select(Measurement.id).where(Measurement.version_id == version_id).limit(1)
)).scalar_one_or_none() is not None
return has_measurements
def validate_file_type(content_type: str) -> bool:
"""Validate if file type is allowed."""
@@ -195,8 +254,14 @@ async def get_file(
):
"""Serve a file from the uploads directory.
Requires authentication but no specific role.
Requires authentication; statistics/ and reports/ additionally require
Metrologist, Supervisor or admin, since those are production data.
"""
if file_path.startswith(_RESTRICTED_GET_PREFIXES) and not _can_read_restricted(user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Metrologist, Supervisor or admin role required",
)
full_path = resolve_upload_path(file_path)
return FileResponse(full_path)
@@ -205,13 +270,29 @@ async def get_file(
async def delete_file(
file_path: str,
user: User = Depends(require_maker),
db: AsyncSession = Depends(get_db),
):
"""Delete a file from the uploads directory.
Also deletes associated thumbnail if it exists.
Also deletes associated thumbnail if it exists. statistics/, reports/ and
logos/ can never be deleted here (403); a recipe file still referenced by
its current version or by a version with measurements is kept (409).
"""
if file_path.startswith(_UNDELETABLE_PREFIXES):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="This folder cannot be deleted through the files API.",
)
full_path = resolve_upload_path(file_path)
relative_path = str(full_path.relative_to(settings.upload_path)).replace("\\", "/")
if await _is_recipe_file_protected(db, relative_path):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="File is referenced by a current or measured recipe version.",
)
# Delete thumbnail if it exists
if full_path.parent.name != "thumbnails":
thumbnail_path = full_path.parent / "thumbnails" / full_path.name
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."""