101 lines
4.1 KiB
Python
101 lines
4.1 KiB
Python
import uuid
|
|
import io
|
|
import warnings
|
|
from datetime import date
|
|
from pathlib import PurePosixPath
|
|
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
|
|
from PIL import Image, UnidentifiedImageError
|
|
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_download_url, put_validated_image
|
|
from app.models.visit import Visit
|
|
|
|
router = APIRouter(prefix="/upload", tags=["Upload"])
|
|
|
|
_IMAGE_TYPES = {
|
|
"image/jpeg": "jpg",
|
|
"image/png": "png",
|
|
"image/webp": "webp",
|
|
}
|
|
MAX_IMAGE_BYTES = 10 * 1024 * 1024
|
|
MAX_IMAGE_PIXELS = 25_000_000
|
|
|
|
|
|
def validate_image_upload(filename: str, content_type: str) -> str:
|
|
"""Validate image metadata before accepting bytes for server-side verification."""
|
|
if content_type.lower() not in _IMAGE_TYPES:
|
|
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 _IMAGE_TYPES[content_type.lower()]
|
|
|
|
|
|
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"}
|
|
)
|
|
|
|
|
|
def normalize_image(image_bytes: bytes) -> bytes:
|
|
"""Verify image bytes, enforce a pixel limit, and strip active/metadata content by re-encoding."""
|
|
if not image_bytes or len(image_bytes) > MAX_IMAGE_BYTES:
|
|
raise HTTPException(status_code=413, detail="Image must be between 1 byte and 10 MB")
|
|
try:
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("error", Image.DecompressionBombWarning)
|
|
with Image.open(io.BytesIO(image_bytes)) as image:
|
|
image.verify()
|
|
with Image.open(io.BytesIO(image_bytes)) as image:
|
|
if image.width * image.height > MAX_IMAGE_PIXELS:
|
|
raise HTTPException(status_code=413, detail="Image dimensions are too large")
|
|
normalized = image.convert("RGB")
|
|
output = io.BytesIO()
|
|
normalized.save(output, format="JPEG", quality=85, optimize=True)
|
|
return output.getvalue()
|
|
except HTTPException:
|
|
raise
|
|
except (Image.DecompressionBombError, Image.DecompressionBombWarning, OSError, UnidentifiedImageError, ValueError) as exc:
|
|
raise HTTPException(status_code=400, detail="Invalid image content") from exc
|
|
|
|
|
|
@router.post("/image")
|
|
async def upload_image(
|
|
file: UploadFile = File(...),
|
|
current_user: dict = Depends(require_any_role),
|
|
):
|
|
"""Validate and normalize an image before persisting it to private object storage."""
|
|
validate_image_upload(file.filename or "upload.jpg", file.content_type or "")
|
|
normalized = normalize_image(await file.read(MAX_IMAGE_BYTES + 1))
|
|
object_key = build_object_key(current_user["user_id"], "jpg")
|
|
put_validated_image(object_key, normalized)
|
|
return {"object_key": object_key}
|
|
|
|
|
|
@router.get("/download-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}
|