File uploads and object storage

pip install fastapi uvicorn python-multipart

import re
import uuid
from pathlib import Path

from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.responses import FileResponse

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

MAX_BYTES = 2 * 1024 * 1024  # 2 MB
ALLOWED_TYPES = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"}

app = FastAPI(title="Brew & Bean photo uploads")


def looks_like_image(ext: str, head: bytes) -> bool:
    """Check the first bytes of the file, because the declared type can be faked."""
    if ext == ".jpg":
        return head.startswith(b"\xff\xd8\xff")
    if ext == ".png":
        return head.startswith(b"\x89PNG\r\n\x1a\n")
    if ext == ".webp":
        return head[:4] == b"RIFF" and head[8:12] == b"WEBP"
    return False


@app.post("/menu/{item_id}/photo", status_code=201)
async def upload_photo(item_id: str, file: UploadFile = File(...)):
    if not re.fullmatch(r"[a-z0-9-]{1,40}", item_id):
        raise HTTPException(status_code=400, detail="Bad item id")

    ext = ALLOWED_TYPES.get(file.content_type)
    if ext is None:
        raise HTTPException(status_code=415, detail="Only JPEG, PNG or WebP images are allowed")

    data = await file.read(MAX_BYTES + 1)  # never read more than the limit plus one byte
    if len(data) > MAX_BYTES:
        raise HTTPException(status_code=413, detail="File is larger than 2 MB")
    if not looks_like_image(ext, data[:12]):
        raise HTTPException(status_code=400, detail="File content does not match its type")

    # Our own name: never use the filename the user sent
    stored_name = f"{item_id}-{uuid.uuid4().hex}{ext}"
    (UPLOAD_DIR / stored_name).write_bytes(data)
    return {"filename": stored_name, "size": len(data), "url": f"/photos/{stored_name}"}


@app.get("/photos/{stored_name}")
def get_photo(stored_name: str):
    # Block ../ tricks: the name must be a plain file name
    if Path(stored_name).name != stored_name:
        raise HTTPException(status_code=400, detail="Bad file name")
    path = UPLOAD_DIR / stored_name
    if not path.is_file():
        raise HTTPException(status_code=404, detail="Photo not found")
    return FileResponse(path)

curl -X POST http://127.0.0.1:8000/menu/espresso/photo -F "file=@espresso.png;type=image/png"
curl -X POST http://127.0.0.1:8000/menu/espresso/photo -F "file=@notes.txt;type=text/plain" 

pip install fastapi uvicorn boto3

import os
import re
import uuid

import boto3
from botocore.config import Config
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

BUCKET = os.environ.get("S3_BUCKET", "brewbean-photos")

# For real AWS S3, leave S3_ENDPOINT unset and put credentials in the standard
# AWS environment variables or ~/.aws. For local MinIO set S3_ENDPOINT=http://localhost:9000.
s3 = boto3.client(
    "s3",
    endpoint_url=os.environ.get("S3_ENDPOINT"),
    region_name=os.environ.get("AWS_REGION", "ap-south-1"),
    config=Config(signature_version="s3v4"),
)

ALLOWED_TYPES = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"}

app = FastAPI(title="Brew & Bean presigned uploads")


class UploadRequest(BaseModel):
    content_type: str


@app.post("/menu/{item_id}/photo-upload-url")
def photo_upload_url(item_id: str, body: UploadRequest):
    if not re.fullmatch(r"[a-z0-9-]{1,40}", item_id):
        raise HTTPException(status_code=400, detail="Bad item id")
    ext = ALLOWED_TYPES.get(body.content_type)
    if ext is None:
        raise HTTPException(status_code=415, detail="Only JPEG, PNG or WebP images are allowed")

    key = f"menu/{item_id}/{uuid.uuid4().hex}{ext}"
    url = s3.generate_presigned_url(
        "put_object",
        Params={"Bucket": BUCKET, "Key": key, "ContentType": body.content_type},
        ExpiresIn=300,  # the link works for 5 minutes only
    )
    return {"upload_url": url, "key": key, "method": "PUT", "content_type": body.content_type}

// In the browser, after getting upload_url from your API
async function uploadPhoto(file, uploadUrl) {
  const response = await fetch(uploadUrl, {
    method: "PUT",
    headers: { "Content-Type": file.type },
    body: file,
  });
  if (!response.ok) throw new Error("Upload failed: " + response.status);
}

Review this file upload endpoint for security and reliability problems. Consider file size, type checks, filenames, path traversal, storage location, and behaviour with many servers. For each problem, explain the attack or failure in one sentence and show a small fix.

[paste your upload code]