# Patterns — Handling File Uploads

# Contents

Presigned direct-to-storage flow
Magic-byte validation table
Streaming multipart limit enforcement
Quarantine and scan promotion
Resumable uploads and lifecycle cleanup
Gotchas

# Presigned direct-to-storage flow

python

# 1) Client asks to upload
# POST /v1/uploads {"filename": "report.pdf", "content_type": "application/pdf", "size": 8123456}
import boto3, uuid

# 15 min: enough for a slow client on the declared size, short enough that a
# leaked URL is a bounded liability.
PRESIGN_TTL = 900

def create_upload(req):
    validate_declared(req.content_type, req.size)          # rule 2 pre-check
    file_id = str(uuid.uuid4())
    key = f"quarantine/{file_id}.pdf"                      # rule 4: server key
    url = s3.generate_presigned_post(
        Bucket="uploads", Key=key,
        Fields={"Content-Type": req.content_type},
        Conditions=[
            {"Content-Type": req.content_type},
            ["content-length-range", 1, 26_214_400],       # 25 MB doc cap
        ],
        ExpiresIn=PRESIGN_TTL)
    db.insert("uploads", id=file_id, key=key, status="pending",
              original_name=req.filename, declared_size=req.size)
    return {"file_id": file_id, "upload": url}

# 2) Client PUT/POSTs the file to `url`
# 3) POST /v1/uploads/{file_id}/complete → server HEADs the object, verifies
#    size matches, kicks the scan job, returns {"status": "scanning"}

# Magic-byte validation table

python

MAGIC = {
    "image/png":  [b"\x89PNG\r\n\x1a\n"],
    "image/jpeg": [b"\xff\xd8\xff"],
    "image/webp": [b"RIFF"],            # + b"WEBP" at offset 8
    "application/pdf": [b"%PDF"],
    "application/zip": [b"PK\x03\x04"], # also docx/xlsx/pptx containers
}

def sniff_ok(claimed: str, head: bytes) -> bool:
    sigs = MAGIC.get(claimed)
    return bool(sigs) and any(head.startswith(s) for s in sigs)
# Read the first 16 bytes from storage (ranged GET) — never rely on the
# client's Content-Type alone.

# Streaming multipart limit enforcement

python

# FastAPI/Starlette example — same idea in any framework: consume the stream
# in chunks and abort the moment the cap is crossed.
CHUNK = 64 * 1024
MAX_IMAGE = 10 * 1024 * 1024

async def save_stream(stream, dest, cap=MAX_IMAGE):
    written = 0
    with open(dest, "wb") as f:
        async for chunk in stream:
            written += len(chunk)
            if written > cap:
                f.close(); os.unlink(dest)
                raise Payload413(f"file exceeds {cap // 1_048_576} MB limit")
            f.write(chunk)
    return written

# Quarantine and scan promotion

text

quarantine/{uuid}.{ext}   status=pending   (not servable)
        │  scan job (ClamAV / provider malware scan)
        ├── clean    → copy to files/{uuid}.{ext}, status=clean, delete quarantine copy
        └── infected → delete object, status=rejected, notify uploader

Record sha256 at scan time for dedup/audit. The serving layer reads only files/ and only rows with status=clean.

# Resumable uploads and lifecycle cleanup

python

# S3 multipart: 8 MB parts — large enough to keep part count low (10k max),
# small enough that a retry wastes little.
PART_SIZE = 8 * 1024 * 1024

Bucket lifecycle rules (Terraform-style):

hcl

rule { id = "abort-incomplete"  abort_incomplete_multipart_upload_days = 1 }
rule { id = "purge-quarantine"  prefix = "quarantine/"  expiration_days = 1 }

For browser-based resumable uploads prefer tus (tusd server or provider equivalent) over hand-rolled chunk endpoints.

# Gotchas

Content-Length is a claim, not a fact — always cap the actual stream too (rule 2 belt-and-braces).
docx/xlsx/pptx sniff as zip (PK\x03\x04) — accept the zip signature, then check the internal [Content_Types].xml if you must distinguish.
Filename header injection: original filenames go into Content-Disposition later; store them escaped and serve with filename*=UTF-8''... encoding, or attackers smuggle header content.
Presigning with no Content-Type condition lets an attacker upload text/html to your bucket and phish from your domain.
EXIF in images may carry GPS/PII — strip metadata at scan/processing time if images are re-served publicly.
Multipart ETag ≠ MD5 on S3 multipart uploads — use explicit per-part checksums (ChecksumSHA256) for integrity, not ETag comparison.