fix(security): remediate telecom compliance findings

backend/app/api/upload.py

@@ -1,10 +1,48 @@
 import uuid
-from fastapi import APIRouter, Depends
+from datetime import date
+from pathlib import PurePosixPath
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.database import get_db
 from app.middleware.auth import get_current_user, require_any_role
 from app.services.minio_client import generate_presigned_upload_url, generate_presigned_download_url
+from app.models.visit import Visit
 
 router = APIRouter(prefix="/upload", tags=["Upload"])
 
+_IMAGE_TYPES = {
+    "image/jpeg": "jpg",
+    "image/png": "png",
+    "image/webp": "webp",
+}
+
+
+def validate_image_upload(filename: str, content_type: str) -> str:
+    """Validate client-declared image metadata before issuing a short-lived PUT URL."""
+    extension = _IMAGE_TYPES.get(content_type.lower())
+    if not extension:
+        raise HTTPException(status_code=400, detail="Only JPEG, PNG, and WebP images are supported")
+    if PurePosixPath(filename).suffix.lower() not in {".jpg", ".jpeg", ".png", ".webp"}:
+        raise HTTPException(status_code=400, detail="Invalid image filename extension")
+    return extension
+
+
+def build_object_key(user_id: str, extension: str) -> str:
+    return f"uploads/{user_id}/{date.today().isoformat()}/{uuid.uuid4().hex}.{extension}"
+
+
+def is_owned_upload_key(object_key: str, user_id: str) -> bool:
+    path = PurePosixPath(object_key)
+    parts = path.parts
+    return (
+        len(parts) == 4
+        and parts[0] == "uploads"
+        and parts[1] == user_id
+        and ".." not in parts
+        and path.suffix.lower() in {".jpg", ".png", ".webp"}
+    )
+
 
 @router.post("/presigned-url")
 async def get_presigned_upload_url(
@@ -13,10 +51,8 @@ async def get_presigned_upload_url(
     current_user: dict = Depends(require_any_role),
 ):
     """Get a presigned PUT URL for direct MinIO upload."""
-    import datetime
-    today = datetime.date.today().isoformat()
-    user_id = current_user["user_id"][:8]
-    object_key = f"{today}/{user_id}/{uuid.uuid4()}.jpg"
+    extension = validate_image_upload(filename, content_type)
+    object_key = build_object_key(current_user["user_id"], extension)
 
     url = generate_presigned_upload_url(object_key)
 
@@ -30,7 +66,14 @@ async def get_presigned_upload_url(
 async def get_presigned_download_url(
     object_key: str,
     current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
 ):
     """Get a presigned GET URL for viewing a photo (1 hour validity)."""
+    result = await db.execute(select(Visit).where(Visit.photos.any(object_key)).limit(1))
+    visit = result.scalar_one_or_none()
+    if not visit:
+        raise HTTPException(status_code=404, detail="Photo not found")
+    if current_user["role"] == "manager" and str(visit.manager_id) != current_user["user_id"]:
+        raise HTTPException(status_code=403, detail="Access denied")
     url = generate_presigned_download_url(object_key)
     return {"download_url": url}