fix: code review — security fixes, dedup, cleanup
- setup: validate roles against users.VALID_ROLES (Supervisor was missing) - files: fix path traversal prefix-match edge case (is_relative_to), dedupe path validation into resolve_upload_path(), use logging not print - measurements: extract shared _build_measurement_filters() helper - client app: prevent open redirect via Referer on /set-language - maker: guard resp.json() in parse-technical-sheet proxy - measure/maker: extract shared file proxy into services/file_proxy.py - measure: localize supervisor validation error messages - annotation-editor: remove global keydown listener in destroy() Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -1,5 +1,5 @@
|
||||
"""File upload/download/delete router for images and PDFs."""
|
||||
import os
|
||||
import logging
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
@@ -11,6 +11,8 @@ from src.backend.config import settings
|
||||
from src.backend.api.middleware.api_key import get_current_user, require_maker
|
||||
from src.backend.models.orm.user import User
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/api/files", tags=["files"])
|
||||
|
||||
# Allowed MIME types
|
||||
@@ -65,6 +67,37 @@ def sanitize_filename(filename: str) -> str:
|
||||
return filename
|
||||
|
||||
|
||||
def resolve_upload_path(file_path: str) -> Path:
|
||||
"""Resolve a user-supplied relative path inside the uploads directory.
|
||||
|
||||
Raises:
|
||||
HTTPException: 404 if the path escapes the uploads directory,
|
||||
cannot be resolved, or does not point to an existing file.
|
||||
"""
|
||||
try:
|
||||
full_path = (settings.upload_path / file_path).resolve()
|
||||
if not full_path.is_relative_to(settings.upload_path.resolve()):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Access denied",
|
||||
)
|
||||
except HTTPException:
|
||||
raise
|
||||
except (OSError, RuntimeError, ValueError):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="File not found",
|
||||
)
|
||||
|
||||
if not full_path.exists() or not full_path.is_file():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="File not found",
|
||||
)
|
||||
|
||||
return full_path
|
||||
|
||||
|
||||
def generate_thumbnail(image_path: Path, thumbnail_path: Path, max_size: tuple[int, int] = (200, 200)):
|
||||
"""Generate a thumbnail for an image."""
|
||||
try:
|
||||
@@ -73,7 +106,7 @@ def generate_thumbnail(image_path: Path, thumbnail_path: Path, max_size: tuple[i
|
||||
img.save(thumbnail_path, quality=85)
|
||||
except Exception as e:
|
||||
# If thumbnail generation fails, we continue without it
|
||||
print(f"Thumbnail generation failed: {e}")
|
||||
logger.warning("Thumbnail generation failed for %s: %s", image_path, e)
|
||||
|
||||
|
||||
@router.post("/upload")
|
||||
@@ -164,30 +197,7 @@ async def get_file(
|
||||
|
||||
Requires authentication but no specific role.
|
||||
"""
|
||||
# Construct full path
|
||||
full_path = settings.upload_path / file_path
|
||||
|
||||
# Security: ensure path is within uploads directory
|
||||
try:
|
||||
full_path = full_path.resolve()
|
||||
if not str(full_path).startswith(str(settings.upload_path.resolve())):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Access denied",
|
||||
)
|
||||
except Exception:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="File not found",
|
||||
)
|
||||
|
||||
# Check if file exists
|
||||
if not full_path.exists() or not full_path.is_file():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="File not found",
|
||||
)
|
||||
|
||||
full_path = resolve_upload_path(file_path)
|
||||
return FileResponse(full_path)
|
||||
|
||||
|
||||
@@ -200,29 +210,7 @@ async def delete_file(
|
||||
|
||||
Also deletes associated thumbnail if it exists.
|
||||
"""
|
||||
# Construct full path
|
||||
full_path = settings.upload_path / file_path
|
||||
|
||||
# Security: ensure path is within uploads directory
|
||||
try:
|
||||
full_path = full_path.resolve()
|
||||
if not str(full_path).startswith(str(settings.upload_path.resolve())):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Access denied",
|
||||
)
|
||||
except Exception:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="File not found",
|
||||
)
|
||||
|
||||
# Check if file exists
|
||||
if not full_path.exists() or not full_path.is_file():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="File not found",
|
||||
)
|
||||
full_path = resolve_upload_path(file_path)
|
||||
|
||||
# Delete thumbnail if it exists
|
||||
if full_path.parent.name != "thumbnails":
|
||||
|
||||
@@ -84,6 +84,44 @@ async def create_measurement_batch(
|
||||
)
|
||||
|
||||
|
||||
def _build_measurement_filters(
|
||||
recipe_id: int | None,
|
||||
version_id: int | None,
|
||||
subtask_id: int | None,
|
||||
measured_by: int | None,
|
||||
lot_number: str | None,
|
||||
serial_number: str | None,
|
||||
date_from: datetime | None,
|
||||
date_to: datetime | None,
|
||||
pass_fail: str | None,
|
||||
) -> list:
|
||||
"""Build SQLAlchemy filter conditions shared by list and CSV export."""
|
||||
filters = []
|
||||
if recipe_id is not None:
|
||||
# Filter by recipe via subquery on RecipeVersion
|
||||
version_ids = select(RecipeVersion.id).where(
|
||||
RecipeVersion.recipe_id == recipe_id
|
||||
)
|
||||
filters.append(Measurement.version_id.in_(version_ids))
|
||||
if version_id is not None:
|
||||
filters.append(Measurement.version_id == version_id)
|
||||
if subtask_id is not None:
|
||||
filters.append(Measurement.subtask_id == subtask_id)
|
||||
if measured_by is not None:
|
||||
filters.append(Measurement.measured_by == measured_by)
|
||||
if lot_number is not None:
|
||||
filters.append(Measurement.lot_number == lot_number)
|
||||
if serial_number is not None:
|
||||
filters.append(Measurement.serial_number == serial_number)
|
||||
if date_from is not None:
|
||||
filters.append(Measurement.measured_at >= date_from)
|
||||
if date_to is not None:
|
||||
filters.append(Measurement.measured_at <= date_to)
|
||||
if pass_fail is not None:
|
||||
filters.append(Measurement.pass_fail == pass_fail)
|
||||
return filters
|
||||
|
||||
|
||||
@router.get("/", response_model=MeasurementListResponse)
|
||||
async def get_measurements(
|
||||
recipe_id: int | None = Query(None),
|
||||
@@ -108,30 +146,10 @@ async def get_measurements(
|
||||
filters. Measurements carry no PII beyond numeric values + a recipe
|
||||
reference, so role-gating beyond authentication isn't justified.
|
||||
"""
|
||||
# Build filter conditions
|
||||
filters = []
|
||||
if recipe_id is not None:
|
||||
# Filter by recipe via subquery on RecipeVersion
|
||||
version_ids = select(RecipeVersion.id).where(
|
||||
RecipeVersion.recipe_id == recipe_id
|
||||
)
|
||||
filters.append(Measurement.version_id.in_(version_ids))
|
||||
if version_id is not None:
|
||||
filters.append(Measurement.version_id == version_id)
|
||||
if subtask_id is not None:
|
||||
filters.append(Measurement.subtask_id == subtask_id)
|
||||
if measured_by is not None:
|
||||
filters.append(Measurement.measured_by == measured_by)
|
||||
if lot_number is not None:
|
||||
filters.append(Measurement.lot_number == lot_number)
|
||||
if serial_number is not None:
|
||||
filters.append(Measurement.serial_number == serial_number)
|
||||
if date_from is not None:
|
||||
filters.append(Measurement.measured_at >= date_from)
|
||||
if date_to is not None:
|
||||
filters.append(Measurement.measured_at <= date_to)
|
||||
if pass_fail is not None:
|
||||
filters.append(Measurement.pass_fail == pass_fail)
|
||||
filters = _build_measurement_filters(
|
||||
recipe_id, version_id, subtask_id, measured_by, lot_number,
|
||||
serial_number, date_from, date_to, pass_fail,
|
||||
)
|
||||
|
||||
# Build query
|
||||
query = select(Measurement).where(and_(*filters) if filters else True)
|
||||
@@ -196,29 +214,10 @@ async def export_measurements_csv(
|
||||
decimal_setting = decimal_result.scalar_one_or_none()
|
||||
decimal_separator = decimal_setting.setting_value if decimal_setting else "."
|
||||
|
||||
# Build filter conditions (same as get_measurements)
|
||||
filters = []
|
||||
if recipe_id is not None:
|
||||
version_ids = select(RecipeVersion.id).where(
|
||||
RecipeVersion.recipe_id == recipe_id
|
||||
)
|
||||
filters.append(Measurement.version_id.in_(version_ids))
|
||||
if version_id is not None:
|
||||
filters.append(Measurement.version_id == version_id)
|
||||
if subtask_id is not None:
|
||||
filters.append(Measurement.subtask_id == subtask_id)
|
||||
if measured_by is not None:
|
||||
filters.append(Measurement.measured_by == measured_by)
|
||||
if lot_number is not None:
|
||||
filters.append(Measurement.lot_number == lot_number)
|
||||
if serial_number is not None:
|
||||
filters.append(Measurement.serial_number == serial_number)
|
||||
if date_from is not None:
|
||||
filters.append(Measurement.measured_at >= date_from)
|
||||
if date_to is not None:
|
||||
filters.append(Measurement.measured_at <= date_to)
|
||||
if pass_fail is not None:
|
||||
filters.append(Measurement.pass_fail == pass_fail)
|
||||
filters = _build_measurement_filters(
|
||||
recipe_id, version_id, subtask_id, measured_by, lot_number,
|
||||
serial_number, date_from, date_to, pass_fail,
|
||||
)
|
||||
|
||||
# Query all matching measurements
|
||||
query = select(Measurement).where(and_(*filters) if filters else True)
|
||||
|
||||
@@ -15,6 +15,7 @@ from pydantic import BaseModel
|
||||
from sqlalchemy import inspect as sa_inspect, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.backend.api.routers.users import VALID_ROLES
|
||||
from src.backend.config import settings
|
||||
from src.backend.database import Base, engine, async_session_factory
|
||||
from src.backend.models.orm import (
|
||||
@@ -544,7 +545,6 @@ async def manage_user(body: ManageUserBody):
|
||||
"""Create or update a user. user_id=null creates new, user_id=int updates."""
|
||||
_check_password(body.password)
|
||||
|
||||
VALID_ROLES = {"Maker", "MeasurementTec", "Metrologist"}
|
||||
invalid = set(body.roles) - VALID_ROLES
|
||||
if invalid:
|
||||
raise HTTPException(
|
||||
|
||||
Reference in New Issue
Block a user