# 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 ``` 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.