"""File upload/download/delete router for images and PDFs.""" import logging import re from pathlib import Path from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status from fastapi.responses import FileResponse from PIL import Image 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 ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"} ALLOWED_PDF_TYPE = "application/pdf" ALLOWED_TYPES = ALLOWED_IMAGE_TYPES | {ALLOWED_PDF_TYPE} def validate_file_type(content_type: str) -> bool: """Validate if file type is allowed.""" return content_type in ALLOWED_TYPES def validate_file_size(file_size: int) -> bool: """Validate if file size is within limits.""" max_bytes = settings.max_upload_size_mb * 1024 * 1024 return file_size <= max_bytes def sanitize_filename(filename: str) -> str: """Sanitize an uploaded filename to prevent path traversal and injection. - Strips directory path separators (/ and \\). - Only allows alphanumeric characters, dots, hyphens, and underscores. - Rejects filenames starting with '.' (hidden files). - Rejects empty filenames. Returns: The sanitized filename. Raises: HTTPException: If filename is empty or starts with '.'. """ # Strip any path separators to prevent directory traversal filename = filename.replace("/", "").replace("\\", "") # Remove any characters not in the allowed set filename = re.sub(r"[^a-zA-Z0-9._-]", "", filename) if not filename: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Filename is empty or contains only invalid characters.", ) if filename.startswith("."): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Filenames starting with '.' are not allowed.", ) 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: with Image.open(image_path) as img: img.thumbnail(max_size, Image.Resampling.LANCZOS) img.save(thumbnail_path, quality=85) except Exception as e: # If thumbnail generation fails, we continue without it logger.warning("Thumbnail generation failed for %s: %s", image_path, e) @router.post("/upload") async def upload_file( file: UploadFile = File(...), recipe_id: int | None = None, version_id: int | None = None, user: User = Depends(require_maker), ): """Upload an image or PDF file. Files are stored in uploads/{recipe_id}/{version_id}/ Thumbnails are generated for images. """ # Validate file type if not validate_file_type(file.content_type): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"File type {file.content_type} not allowed. Must be image or PDF.", ) # Read file content content = await file.read() file_size = len(content) # Validate file size if not validate_file_size(file_size): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"File size {file_size} bytes exceeds maximum {settings.max_upload_size_mb}MB", ) # Determine storage path if recipe_id is not None and version_id is not None: storage_dir = settings.upload_path / str(recipe_id) / str(version_id) else: storage_dir = settings.upload_path / "general" storage_dir.mkdir(parents=True, exist_ok=True) # Sanitize the uploaded filename safe_filename = sanitize_filename(file.filename or "") # Generate unique filename if file already exists file_path = storage_dir / safe_filename counter = 1 while file_path.exists(): name_parts = safe_filename.rsplit(".", 1) if len(name_parts) == 2: file_path = storage_dir / f"{name_parts[0]}_{counter}.{name_parts[1]}" else: file_path = storage_dir / f"{safe_filename}_{counter}" counter += 1 # Write file file_path.write_bytes(content) # Generate thumbnail for images thumbnail_path = None if file.content_type in ALLOWED_IMAGE_TYPES: thumbnail_dir = storage_dir / "thumbnails" thumbnail_dir.mkdir(exist_ok=True) thumbnail_path = thumbnail_dir / file_path.name generate_thumbnail(file_path, thumbnail_path) # Return relative path from uploads directory relative_path = file_path.relative_to(settings.upload_path) thumbnail_relative = ( thumbnail_path.relative_to(settings.upload_path) if thumbnail_path else None ) return { "file_path": str(relative_path).replace("\\", "/"), "thumbnail_path": str(thumbnail_relative).replace("\\", "/") if thumbnail_relative else None, "file_type": file.content_type, "file_size": file_size, } @router.get("/{file_path:path}") async def get_file( file_path: str, user: User = Depends(get_current_user), ): """Serve a file from the uploads directory. Requires authentication but no specific role. """ full_path = resolve_upload_path(file_path) return FileResponse(full_path) @router.delete("/{file_path:path}", status_code=status.HTTP_204_NO_CONTENT) async def delete_file( file_path: str, user: User = Depends(require_maker), ): """Delete a file from the uploads directory. Also deletes associated thumbnail if it exists. """ full_path = resolve_upload_path(file_path) # Delete thumbnail if it exists if full_path.parent.name != "thumbnails": thumbnail_path = full_path.parent / "thumbnails" / full_path.name if thumbnail_path.exists(): thumbnail_path.unlink() # Delete file full_path.unlink()