How to Secure File Uploads in a FastAPI App and Stop Disguised, Oversized, and Booby-Trapped Files

What you will build, and why uploads need layers

By the end of this tutorial you will have a small FastAPI service that accepts profile pictures and nothing else. It refuses anything that is not a genuine PNG, JPEG or WebP image, puts a ceiling on the size of every request so that an oversized upload is cut off after a few megabytes instead of being stored, rebuilds each image from its pixels so that nothing hidden in the file survives, stores the result under a name the server invents, and serves it back with headers that tell the browser not to run it.

You will also build a bench of eleven hostile uploads and run it against three versions of the service: the endpoint most tutorials show, the version with the checks most people add next, and the finished service. The scoreboard you will reproduce is 11 attacks that got through the first version, 7 that got through the second, and none that got through the third. The second number is the instructive one, because that version looks careful.

A file upload is the one feature where a stranger hands your server a name, a claimed type and a pile of bytes, and the server then writes the bytes to disk and often serves them to other people. MITRE files the mistake as CWE-434 and describes it this way: “The product allows the upload or transfer of dangerous file types that are automatically processed within its environment.” This is not a theoretical weakness. This site covered a real one in Elementor Pro Patches a Critical File Upload Bug That Enabled Remote Code Execution.

The OWASP File Upload Cheat Sheet sets the tone for everything below: “There is no silver bullet in validating user content.” Its answer is defense in depth, which here means several small checks, each covering a hole the others leave open. The CWE entry adds a useful warning about the name of the bug. It says the common label Unrestricted File Upload “could be interpreted as the lack of restrictions on the size or number of uploaded files, which is a resource consumption issue.” This tutorial covers both readings: what the file is, and how much of it the server will take.

Key terms in plain language

Multipart form data. The request format that browsers and curl -F use for file uploads. The body is a series of parts separated by a boundary string. Each part has its own headers, including Content-Disposition, which carries the field name and the file name, and an optional Content-Type, followed by the raw bytes. Whoever sends the request chooses every one of those values.

Content-Type header. What the sender claims the file is, for example image/png. OWASP puts it bluntly: “The Content-Type for uploaded files is provided by the user, and as such cannot be trusted, as it is trivial to spoof.”

File signature, or magic bytes. The first few bytes of a file identify its format. A PNG starts with the eight bytes 89 50 4E 47 0D 0A 1A 0A and a JPEG with FF D8 FF. Reading them tells you what the bytes claim to be, which is a different question from what the header or the extension claims.

Polyglot. A file that is valid in two formats at once, or valid in one format with extra data attached that another reader treats differently. In this lab it is a real PNG with a small web page glued on after its last chunk.

MIME sniffing. A browser guessing a file’s type from its content instead of trusting the Content-Type header. The response header X-Content-Type-Options: nosniff switches that guessing off.

Origin and stored XSS. A browser treats pages that share a scheme, host and port as one origin. A script in a page from your origin can do what your own pages can do, such as calling your API with the visitor’s session. If an attacker can store a web page on your origin and get a visitor to open it, that is stored cross-site scripting (XSS); How to Stop XSS Attacks in a Python Web App With a Content Security Policy covers the page side of the problem.

Decompression bomb, or pixel bomb. An image file that is tiny on disk but declares a huge canvas, so decoding it allocates far more memory than the file size suggests.

Allowlist and blocklist. An allowlist names what is permitted and refuses everything else. A blocklist names what is forbidden and permits everything else. OWASP says of blocking extensions that it “is a weak protection method on its own.”

Re-encoding. Decoding an image into raw pixels and writing those pixels into a brand-new file. Whatever is not a pixel, such as metadata or bytes after the end of the image, is left behind.

Prerequisites

• Python 3.11 or newer, because the pinned Pillow 13.0.0 and Starlette 1.9.0 require it. Everything here ran on Python 3.13.14 on Windows 11. Results that depend on Windows are labeled, and macOS and Linux behave differently in those places.
• A terminal, plus curl for Step 9. Windows 10 and 11 and macOS include it, and Linux distributions package it.
• A Chromium-based browser for the browser checks. The lab uses Microsoft Edge 155 in headless mode; Chrome works the same way once you change one path. If you skip the browser, the results are printed in this article.
• Basic Python and a feel for HTTP status codes. You do not need to know FastAPI or Pillow; each piece is explained when it first appears.
• About ninety minutes. Each listing carries its file name in the first line. Save every listing under that name in one folder.

Step 1: Create the project and the hostile files

Create the environment

Make a folder, create a virtual environment inside it, and install the packages. The versions below are the ones this tutorial ran on. An older FastAPI may not have the request body limit that Step 5 uses, so if anything behaves differently from this page, install the pinned versions first.

mkdir upload-lab
cd upload-lab
python -m venv .venv
.venv\Scripts\activate
python -m pip install fastapi==0.146.0 uvicorn==0.54.0 python-multipart==0.0.32 pillow==13.0.0 requests==2.34.2 httpx2==2.14.0 pytest==9.1.1 psutil==7.2.2

FastAPI 0.146.0 brings in Starlette 1.9.0, the toolkit underneath it. python-multipart is the parser that reads upload requests, Pillow is the image library, httpx2 is the client that Starlette’s test client uses in Step 10, and psutil only measures memory in two demos. In PowerShell, activate with .venv\Scripts\Activate.ps1. On macOS or Linux, use source .venv/bin/activate.

A helper that runs an app in the background

Save this as labserver.py. It starts a FastAPI app on localhost in a background thread, so a script can send requests to it and then shut it down.

# labserver.py
import threading
import time

import uvicorn


class LiveServer:
    """Run a FastAPI app on 127.0.0.1 in a background thread for the demos."""

    def __init__(self, app, port):
        config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="critical")
        self.server = uvicorn.Server(config)
        self.thread = threading.Thread(target=self.server.run, daemon=True)
        self.url = f"http://127.0.0.1:{port}"

    def __enter__(self):
        self.thread.start()
        while not self.server.started:
            if not self.thread.is_alive():
                raise RuntimeError("the server did not start (is the port already in use?)")
            time.sleep(0.02)
        return self

    def __exit__(self, *exc):
        self.server.should_exit = True
        self.thread.join(timeout=5)

Build the hostile files

Save this as attacks.py. Every hostile file in the tutorial is built here with Pillow, so you can read exactly what each one contains. The script that proves an attack worked is the same in all the web pages: it stamps the page with the origin it ran in. A real attacker would use that foothold to call your API as the visitor or read what the page can reach.

# attacks.py - the hostile files used throughout the lab
import io

from PIL import Image, PngImagePlugin

# Proof that a script ran in the page's own origin: it stamps the document element, and the headless
# browser dump shows the stamp. A real attacker would read cookies or call the API instead.
MARK = 'document.documentElement.setAttribute("data-xss", location.origin)'

HTML_PAGE = f"<!doctype html><title>profile</title><script>{MARK}</script>".encode()
XHTML_PAGE = (
    '<html xmlns="http://www.w3.org/1999/xhtml"><head><title>profile</title></head>'
    f"<body><script>{MARK}</script></body></html>"
).encode()
SVG_IMAGE = f'<svg xmlns="http://www.w3.org/2000/svg" width="40" height="40"><script>{MARK}</script></svg>'.encode()
TRAILER = f"<script>{MARK}</script>".encode()


def png_bytes(size=(8, 8), color="red", **save_args):
    buffer = io.BytesIO()
    Image.new("RGB", size, color).save(buffer, "PNG", **save_args)
    return buffer.getvalue()


def png_with_trailer():
    """A real, decodable PNG with a web page glued on after the last chunk."""
    return png_bytes() + TRAILER


def png_canvas(side):
    """A tiny file that decodes to side x side pixels: a solid grey canvas compresses about 1000 to 1."""
    buffer = io.BytesIO()
    Image.new("L", (side, side), 0).save(buffer, "PNG", compress_level=9)
    return buffer.getvalue()


def jpeg_with_gps(size=(40, 20), orientation=6):
    """A phone-style JPEG: camera make, a rotation flag and a GPS position in the Exif block."""
    exif = Image.Exif()
    exif[0x010F] = "LabCam"
    exif[0x0112] = orientation
    gps = exif.get_ifd(0x8825)
    gps[1], gps[2] = "N", (48.0, 51.0, 30.0)
    gps[3], gps[4] = "E", (2.0, 17.0, 40.0)
    buffer = io.BytesIO()
    Image.new("RGB", size, "blue").save(buffer, "JPEG", exif=exif)
    return buffer.getvalue()


def has_gps(data):
    return bool(Image.open(io.BytesIO(data)).getexif().get_ifd(0x8825))


def png_with_text_chunk():
    info = PngImagePlugin.PngInfo()
    info.add_text("Comment", "<script>" + MARK + "</script>")
    return png_bytes(pnginfo=info)

Three details are worth knowing before you move on. A solid grey canvas compresses about a thousand to one, so png_canvas(10000), a picture of 100 million pixels, is a file under 100 KiB. jpeg_with_gps builds what a phone produces: a camera make, a rotation flag in the Exif block, and a GPS position. And png_with_trailer is a valid, decodable PNG with a web page glued on after the last chunk, which image readers ignore.

Ask a browser whether the script ran

Save this as browser.py. It starts Edge in headless mode, which loads the page like a visitor would, runs its scripts, and prints the resulting DOM. If the attacker’s script ran, the DOM contains the stamp. On macOS or Linux, point EDGE at Chrome or Chromium instead.

# browser.py - open a URL in headless Edge and report whether the attacker's script ran
import shutil
import subprocess
import tempfile

EDGE = r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe"  # any Chromium-based browser works


def dom(url, budget_ms=4000):
    profile = tempfile.mkdtemp(prefix="edgeprofile")
    try:
        done = subprocess.run(
            [EDGE, "--headless=new", "--disable-gpu", "--no-first-run", f"--user-data-dir={profile}",
             f"--virtual-time-budget={budget_ms}", "--dump-dom", url],
            capture_output=True, timeout=120,
        )
        return done.stdout.decode("utf-8", "replace")
    finally:
        shutil.rmtree(profile, ignore_errors=True)


def script_ran(url):
    """True when the page stamped itself with data-xss="<origin>", which only happens if its script executed."""
    return 'data-xss="http://127.0.0.1' in dom(url)

Step 2: Build the naive endpoint and attack it

The endpoint most tutorials show

Save this as app_v1.py. It takes an UploadFile, writes it into a folder under whatever name the client sent, and serves that folder with StaticFiles. FastAPI’s documentation explains why UploadFile is the usual choice: it uses a spooled file, which the documentation describes as “A file stored in memory up to a maximum size limit, and after passing this limit it will be stored on disk.” Nothing else in this version is defensive.

# app_v1.py - the upload endpoint most tutorials show
from pathlib import Path

from fastapi import FastAPI, UploadFile
from fastapi.staticfiles import StaticFiles


def make_app(upload_dir):
    upload_dir = Path(upload_dir)
    upload_dir.mkdir(parents=True, exist_ok=True)
    app = FastAPI()

    @app.post("/upload")
    async def upload(file: UploadFile):
        target = upload_dir / file.filename
        target.write_bytes(await file.read())
        return {"url": f"/files/{file.filename}"}

    app.mount("/files", StaticFiles(directory=upload_dir), name="files")
    return app

The attack bench

Now the bench. Save the next listing as matrix.py; it arrives in four pieces, and the pieces go into one file. The first piece builds multipart requests by hand, so the lab controls the exact bytes of every file name, and it defines the helper that most web-page attacks use: upload the file, fetch the URL the app returned, look at the Content-Type it came back with, and, if that is a type a browser runs scripts in, open it in the browser and see whether the script ran.

# matrix.py - send the same hostile uploads to one version of the app: python matrix.py v1|v2|v3|all
import importlib
import sys
import tempfile
from pathlib import Path

import requests

import attacks
from browser import script_ran
from labserver import LiveServer

ACTIVE = {"text/html", "application/xhtml+xml", "image/svg+xml"}  # types a browser runs scripts in
PORTS = {"v1": 8751, "v2": 8752, "v3": 8753}


def multipart(filename, content_type, body):
    boundary = "----labboundary"
    head = (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="').encode()
    head += filename.encode("utf-8") + f'"\r\nContent-Type: {content_type}\r\n\r\n'.encode()
    return head + body + f"\r\n--{boundary}--\r\n".encode(), {"Content-Type": f"multipart/form-data; boundary={boundary}"}


class Target:
    def __init__(self, base, uploads):
        self.base = base
        self.uploads = uploads

    def upload(self, filename, content_type, body):
        data, headers = multipart(filename, content_type, body)
        return requests.post(self.base + "/upload", data=data, headers=headers, timeout=60)

    def get(self, path):
        return requests.get(self.base + path, timeout=60)


def served_page(t, filename, content_type, body, fetch_as=None):
    reply = t.upload(filename, content_type, body)
    if reply.status_code >= 300:
        return False, f"refused with HTTP {reply.status_code}"
    url = fetch_as or reply.json()["url"]
    media = t.get(url).headers.get("content-type", "none").split(";")[0]
    ran = script_ran(t.base + url) if media in ACTIVE else False
    return ran, f"accepted, served as {media}, script ran in the app's origin: {'yes' if ran else 'no'}"

The first five attacks are all web pages in disguise. They differ in the file name and the claimed type, which is exactly what real filters look at.

# matrix.py (continued)
def html_page(t):
    return served_page(t, "evil.html", "text/html", attacks.HTML_PAGE)


def html_spoofed_type(t):
    return served_page(t, "evil.html", "image/png", attacks.HTML_PAGE)


def html_trailing_dot(t):
    return served_page(t, "evil.html.", "image/png", attacks.HTML_PAGE, fetch_as="/files/evil.html")


def xhtml_page(t):
    return served_page(t, "evil.xhtml", "image/png", attacks.XHTML_PAGE)


def svg_image(t):
    return served_page(t, "logo.svg", "image/svg+xml", attacks.SVG_IMAGE)

Six more attacks follow. Each returns whether it got through and a sentence saying how it was judged. Reading them tells you what getting through means in every row.

# matrix.py (continued)
def path_traversal(t):
    target = (t.uploads / ".." / ".." / "escaped.txt").resolve()
    target.unlink(missing_ok=True)
    reply = t.upload("../../escaped.txt", "image/png", attacks.png_bytes())
    escaped = target.exists()
    target.unlink(missing_ok=True)
    return escaped, f"HTTP {reply.status_code}, file written outside the upload folder: {'yes' if escaped else 'no'}"


def overwrite(t):
    url = t.upload("avatar.png", "image/png", attacks.png_bytes(color="red")).json()["url"]
    before = t.get(url).content
    t.upload("avatar.png", "image/png", attacks.png_bytes(color="blue"))  # another user, same file name
    changed = before != t.get(url).content
    return changed, f"the first user's file changed after a second upload with the same name: {'yes' if changed else 'no'}"


def glued_on_page(t):
    reply = t.upload("pixel.png", "image/png", attacks.png_with_trailer())
    if reply.status_code >= 300:
        return False, f"refused with HTTP {reply.status_code}"
    url = reply.json()["url"]
    kept = attacks.TRAILER in t.get(url).content
    ran = script_ran(t.base + url)  # a browser shows it as an image, so the glued-on page never runs
    return kept, f"accepted; the page glued onto the PNG is still in the stored file: {'yes' if kept else 'no'} (opened in the browser, script ran: {'yes' if ran else 'no'})"


def page_named_png(t):
    reply = t.upload("photo.png", "image/png", b"<html><body>this is not a picture</body></html>")
    accepted = reply.status_code < 300
    return accepted, "accepted and stored" if accepted else f"refused with HTTP {reply.status_code}"


def gps_photo(t):
    reply = t.upload("holiday.jpg", "image/jpeg", attacks.jpeg_with_gps())
    if reply.status_code >= 300:
        return False, f"refused with HTTP {reply.status_code}"
    kept = attacks.has_gps(t.get(reply.json()["url"]).content)
    return kept, f"accepted; GPS position still inside the stored file: {'yes' if kept else 'no'}"


def pixel_bomb(t):
    body = attacks.png_canvas(10000)
    reply = t.upload("canvas.png", "image/png", body)
    accepted = reply.status_code < 300
    return accepted, f"a {round(len(body) / 1024)} KiB file that decodes to 100 megapixels: " + ("accepted and stored" if accepted else f"refused with HTTP {reply.status_code}")

The last piece lists the attacks, runs them against one version of the app, and prints a grid when you pass all.

# matrix.py (continued)
ATTACKS = [html_page, html_spoofed_type, html_trailing_dot, xhtml_page, svg_image, path_traversal, overwrite,
           glued_on_page, page_named_png, gps_photo, pixel_bomb]


def run(version):
    module = importlib.import_module(f"app_{version}")
    results = []
    with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmp:
        uploads = Path(tmp) / "site" / "uploads"
        with LiveServer(module.make_app(uploads), PORTS[version]) as server:
            target = Target(server.url, uploads)
            print(f"== {version} on {server.url}")
            for attack in ATTACKS:
                hit, text = attack(target)
                results.append((attack.__name__, hit))
                print(f"{attack.__name__:18} {'GOT THROUGH' if hit else 'stopped    '}  {text}")
    print(f"{sum(hit for _, hit in results)} of {len(results)} attacks got through")
    return results


if __name__ == "__main__":
    wanted = sys.argv[1] if len(sys.argv) > 1 else "all"
    versions = ["v1", "v2", "v3"] if wanted == "all" else [wanted]
    table = {v: dict(run(v)) for v in versions}
    if wanted == "all":
        print("\nattack              " + "  ".join(f"{v:12}" for v in versions))
        for attack in ATTACKS:
            print(f"{attack.__name__:19} " + "  ".join(f"{'got through' if table[v][attack.__name__] else 'stopped':12}" for v in versions))

Here is what each attack tries, so you can read the results in the next section.

Run it against version 1

python matrix.py v1

Expect this output. The server runs in the same process, on port 8751, and the browser checks make the run take a while.

== v1 on http://127.0.0.1:8751
html_page          GOT THROUGH  accepted, served as text/html, script ran in the app's origin: yes
html_spoofed_type  GOT THROUGH  accepted, served as text/html, script ran in the app's origin: yes
html_trailing_dot  GOT THROUGH  accepted, served as text/html, script ran in the app's origin: yes
xhtml_page         GOT THROUGH  accepted, served as application/xhtml+xml, script ran in the app's origin: yes
svg_image          GOT THROUGH  accepted, served as image/svg+xml, script ran in the app's origin: yes
path_traversal     GOT THROUGH  HTTP 200, file written outside the upload folder: yes
overwrite          GOT THROUGH  the first user's file changed after a second upload with the same name: yes
glued_on_page      GOT THROUGH  accepted; the page glued onto the PNG is still in the stored file: yes (opened in the browser, script ran: no)
page_named_png     GOT THROUGH  accepted and stored
gps_photo          GOT THROUGH  accepted; GPS position still inside the stored file: yes
pixel_bomb         GOT THROUGH  a 95 KiB file that decodes to 100 megapixels: accepted and stored
11 of 11 attacks got through

Every attack worked. The first five lines are the serious ones: an uploaded page was served back by the application itself, with a type chosen from its file extension, and the script ran in the application’s own origin. That is stored XSS. Notice that the browser executed it from text/html, application/xhtml+xml and image/svg+xml alike, so calling a file an image is no defense when its type is SVG. The sixth line shows a name with ../../ in it writing a file outside the upload folder, and the seventh shows two people with the same file name sharing one file. The glued-on page, the GPS position and the pixel bomb are not attacks on a browser at all. They are bytes you now store and may hand to the next program that reads them.

The glued_on_page line has one more detail worth reading. Opened in the browser, the glued-on script did not run. The stored file is a PNG, served as image/png, and a browser shows it as a picture. The danger of a polyglot is any other reader, such as an image tool, a document converter or a server that treats the file as code, and the browser is merely the one reader you can test today.

Step 3: Apply the usual first fix, and see what it misses

Four familiar checks

Here is the version that many code reviews would wave through. Save it as app_v2.py. It allows three Content-Type values, blocks six risky extensions, keeps only the last part of the file name, and rejects files over 5 MiB.

# app_v2.py - the usual first fix: allow some Content-Types, block risky extensions, trim the path, cap the size
from pathlib import Path

from fastapi import FastAPI, HTTPException, UploadFile
from fastapi.staticfiles import StaticFiles

ALLOWED_TYPES = {"image/png", "image/jpeg", "image/gif"}
BLOCKED_EXTENSIONS = {".html", ".htm", ".svg", ".js", ".php", ".exe"}
MAX_BYTES = 5 * 1024 * 1024


def make_app(upload_dir):
    upload_dir = Path(upload_dir)
    upload_dir.mkdir(parents=True, exist_ok=True)
    app = FastAPI()

    @app.post("/upload")
    async def upload(file: UploadFile):
        if file.content_type not in ALLOWED_TYPES:
            raise HTTPException(415, "only png, jpeg and gif images")
        name = Path(file.filename or "").name  # drop any directory part
        if Path(name).suffix.lower() in BLOCKED_EXTENSIONS:
            raise HTTPException(415, "that file extension is not allowed")
        data = await file.read()
        if len(data) > MAX_BYTES:
            raise HTTPException(413, "file too large")
        (upload_dir / name).write_bytes(data)
        return {"url": f"/files/{name}"}

    app.mount("/files", StaticFiles(directory=upload_dir), name="files")
    return app

python matrix.py v2

== v2 on http://127.0.0.1:8752
html_page          stopped      refused with HTTP 415
html_spoofed_type  stopped      refused with HTTP 415
html_trailing_dot  GOT THROUGH  accepted, served as text/html, script ran in the app's origin: yes
xhtml_page         GOT THROUGH  accepted, served as application/xhtml+xml, script ran in the app's origin: yes
svg_image          stopped      refused with HTTP 415
path_traversal     stopped      HTTP 200, file written outside the upload folder: no
overwrite          GOT THROUGH  the first user's file changed after a second upload with the same name: yes
glued_on_page      GOT THROUGH  accepted; the page glued onto the PNG is still in the stored file: yes (opened in the browser, script ran: no)
page_named_png     GOT THROUGH  accepted and stored
gps_photo          GOT THROUGH  accepted; GPS position still inside the stored file: yes
pixel_bomb         GOT THROUGH  a 95 KiB file that decodes to 100 megapixels: accepted and stored
7 of 11 attacks got through

Four attacks are now stopped, and seven still get through. The checks are not useless; they stopped the obvious cases. They fail in specific, instructive ways.

The blocklist has holes

The stopped html_page and html_spoofed_type lines show the allowlist of types and the blocklist of extensions each doing their job. Then xhtml_page walks past the extension check, because .xhtml is not on the list. On the machine used here, Python’s mimetypes module maps that extension to application/xhtml+xml, which the browser treats as a page, and the script runs. OWASP warns about exactly this: “Please be aware that blocking specific extensions is a weak protection method on its own.”

The html_trailing_dot line is the Windows-only trick. The check looks at the suffix of evil.html., which is not .html, so the name passes. Windows then strips the dot when it creates the file. Microsoft documents the rule in its page on file path formats: “If the path doesn’t end in a separator, all trailing periods and spaces (U+0020) are removed.” The file on disk is evil.html, and an attacker who knows that links to it by that name. On Linux the same upload would be stored with the dot, so this particular bypass is a Windows trap, but the lesson applies anywhere: a filter that reasons about the name and a file system that reasons about it differently will disagree. It also makes the scoreboard platform dependent. On Linux or macOS the html_trailing_dot row should read stopped for both of the first two versions, which would make their totals 10 and 6. I measured on Windows only.

A file name is not an identity

The overwrite line is the quiet one. Two people upload avatar.png, and the second replaces the first, because the name is the storage key. Trimming the path with Path(...).name fixed traversal and left this behind.

The header is a claim and the extension is a label

The page_named_png line is accepted because the allowlist checks what the sender says the file is. HTML text called photo.png, with a Content-Type of image/png, satisfies both checks without being an image. OWASP’s advice is to check the type of the file itself and not to rely on the header, which the sections that follow do.

Nobody opened the file

The last three lines, glued_on_page, gps_photo and pixel_bomb, all share one cause: the checks never looked inside the file. The glued-on page, the GPS position and the 100-megapixel canvas are all stored as sent. Fixing that is the center of this tutorial, in Steps 6 and 7.

One problem the matrix cannot show

Look again at the order in app_v2.py: await file.read() comes before the size check. By the time the code decides a file is too large, the server has already received it, parsed it, and read it into memory. Step 5 measures what that costs.

Step 4: Keep the client’s file name away from the file system

The trailing dot in Step 3 showed that a file name does damage in two ways. A filter and a file system can read the same name differently, and the name decides which file gets written. The cleanest fix is to take the name out of the equation. First, see what actually arrives and what happens when you use it.

See what arrives, and what Windows does with it

Save this as step_names.py. It sends twelve file names to an echo endpoint as raw multipart requests, shows the filename that Starlette hands to your code, and then tries to write a file under that name in an uploads folder, reporting what the operating system really created.

# step_names.py - what filename does Starlette hand your code, and what does Windows create for it?
import shutil
from pathlib import Path, PurePosixPath

import requests
from fastapi import FastAPI, UploadFile

from labserver import LiveServer

LAB = Path(__file__).resolve().parent
ROOT = LAB / "scratch" / "names_demo"
UPLOADS = ROOT / "site" / "uploads"  # two folders up from here is ROOT, so a traversal name lands inside the demo

app = FastAPI()


@app.post("/echo")
async def echo(file: UploadFile):
    return {"filename": file.filename}


def raw_multipart(filename):
    boundary = b"----labboundary"
    body = (b"--" + boundary + b'\r\nContent-Disposition: form-data; name="file"; filename="' + filename
            + b'"\r\nContent-Type: image/png\r\n\r\nx\r\n--' + boundary + b"--\r\n")
    return body, {"Content-Type": "multipart/form-data; boundary=" + boundary.decode()}


RLO = chr(0x202E)  # right-to-left override: some file listings display the name below as "reportexe.jpg"
NAMES = [
    b"photo.png",
    b"../../evil.txt",
    b"..\\..\\evil.txt",
    b"C:\\Windows\\win.ini",
    b"/etc/passwd",
    b"evil.html.",
    b"evil.html ",
    b"evil.html::$DATA",
    b"x.php:.jpg",
    ("report" + RLO + "gpj.exe").encode("utf-8"),
    b"a" * 300 + b".png",
    b"na\x00me.png",
]


def snapshot():
    return {p for p in ROOT.rglob("*") if p.is_file()}


def short(text, limit=30):
    text = ascii(text)
    return text if len(text) <= limit else text[: limit - 3] + "..."


def where(path):
    try:
        return "uploads/" + ascii(path.relative_to(UPLOADS).as_posix())[1:-1]
    except ValueError:
        return "OUTSIDE uploads: " + path.relative_to(ROOT).as_posix()


def main():
    shutil.rmtree(ROOT, ignore_errors=True)
    UPLOADS.mkdir(parents=True)
    print(f"{'client sent':32}{'handler received':32}what uploads / name does on Windows (NTFS)")
    with LiveServer(app, 8731) as server:
        for raw in NAMES:
            body, headers = raw_multipart(raw)
            name = requests.post(server.url + "/echo", data=body, headers=headers, timeout=10).json()["filename"]
            before = snapshot()
            try:
                with open(UPLOADS / name, "wb") as handle:
                    handle.write(b"data")
                made = ", ".join(sorted(where(p) for p in snapshot() - before)) or "(nothing new)"
            except Exception as exc:  # noqa: BLE001 - print whatever the OS says
                made = type(exc).__name__
            for path in snapshot() - before:
                path.unlink()
            print(f"{short(raw.decode('utf-8', 'replace')):32}{short(name):32}{made}")
    shutil.rmtree(ROOT, ignore_errors=True)
    print("On Linux and macOS, pathlib throws away the base folder when the second part is absolute:")
    print("  PurePosixPath('uploads') / '/etc/passwd' ->", PurePosixPath("uploads") / "/etc/passwd")


if __name__ == "__main__":
    main()

python step_names.py

client sent                     handler received                what uploads / name does on Windows (NTFS)
'photo.png'                     'photo.png'                     uploads/photo.png
'../../evil.txt'                '../../evil.txt'                OUTSIDE uploads: evil.txt
'..\\..\\evil.txt'              '..\\..\\evil.txt'              OUTSIDE uploads: evil.txt
'C:\\Windows\\win.ini'          'win.ini'                       uploads/win.ini
'/etc/passwd'                   '/etc/passwd'                   FileNotFoundError
'evil.html.'                    'evil.html.'                    uploads/evil.html
'evil.html '                    'evil.html '                    uploads/evil.html
'evil.html::$DATA'              'evil.html::$DATA'              uploads/evil.html
'x.php:.jpg'                    'x.php:.jpg'                    uploads/x.php
'report\u202egpj.exe'           'report\u202egpj.exe'           uploads/report\u202egpj.exe
'aaaaaaaaaaaaaaaaaaaaaaaaaa...  'aaaaaaaaaaaaaaaaaaaaaaaaaa...  FileNotFoundError
'na\x00me.png'                  'na\x00me.png'                  ValueError
On Linux and macOS, pathlib throws away the base folder when the second part is absolute:
  PurePosixPath('uploads') / '/etc/passwd' -> /etc/passwd

The Windows column is the one measured here, on NTFS. Read it row by row:

• Directory traversal arrives untouched. ../../evil.txt and ..\..\evil.txt reach your code as sent and land two folders above uploads. The one cleanup I found in the stack is a comment in python-multipart 0.0.32: “Work around an IE6 bug where the full file path is sent instead of just the filename.” The code under it keeps only the last segment when a name starts with a drive letter and a colon, or with two backslashes, which is why C:\Windows\win.ini became win.ini. Relative paths are not covered. You can read the lines in python_multipart/multipart.py.
• An absolute path replaces your folder on Linux and macOS. On Windows, /etc/passwd failed with FileNotFoundError because there is no such folder. The last two lines of the output show the Linux and macOS behavior: when the second part of a pathlib join is absolute, the first part is thrown away, so upload_dir / file.filename would point at /etc/passwd itself. A process that is allowed to write there would overwrite it.
• Windows rewrites names. A trailing dot, a trailing space and ::$DATA all produced evil.html, and x.php:.jpg produced x.php. OWASP describes the last one: with an alternate data stream, “the actual file created on disk carries the extension before the colon”. Its advice is to reject any file name that contains a colon.
• Invisible characters survive. The name report, then U+202E, then gpj.exe was stored as sent. U+202E is the right-to-left override, and listings that honor bidirectional text show that name as reportexe.jpg, hiding the real extension. How to Detect and Strip Invisible Unicode in Python covers this family of tricks.
• Some names simply crash the write. A 300-character name raised FileNotFoundError, and a name with a NUL byte raised ValueError. In an endpoint with no error handling, each becomes an HTTP 500.

The fix: names the server invents

OWASP’s File Upload Cheat Sheet says what to do instead: “creating a random string as a filename, such as generating a UUID/GUID, is essential.” The finished service does that with secrets.token_urlsafe(16), which turns 16 random bytes into 22 URL-safe characters. The client’s name is kept only as a label to show back to people, and it never reaches a path.

Start the module that holds the building blocks. Save this as uploadsafe.py. It will grow over the next three steps, and each later piece starts with a comment that says it continues the file. The constants at the top are used in Steps 6 and 7.

# uploadsafe.py - decide what an upload really is, rebuild it, and keep it under a name the server chose
import io
import re
import secrets
import unicodedata
from pathlib import Path

from PIL import Image, ImageOps

MAX_PIXELS = 25_000_000  # 25 megapixels: a 6000 by 4000 camera photo fits, a 10000 by 10000 canvas does not
FORMATS = {  # the only formats we accept: Pillow format name -> (media type, extension we choose)
    "PNG": ("image/png", ".png"),
    "JPEG": ("image/jpeg", ".jpg"),
    "WEBP": ("image/webp", ".webp"),
}
SAVE_OPTIONS = {"JPEG": {"quality": 85}, "WEBP": {"quality": 85}}
ID_PATTERN = re.compile(r"[A-Za-z0-9_-]{22}")  # exactly what secrets.token_urlsafe(16) produces


class Rejected(Exception):
    def __init__(self, status, reason):
        super().__init__(reason)
        self.status = status
        self.reason = reason


def display_name(raw, limit=80):
    """A name that is safe to show back to people. It is never used to build a path."""
    name = (raw or "").replace("\\", "/").rsplit("/", 1)[-1]
    name = "".join(ch for ch in name if unicodedata.category(ch)[0] != "C")  # control and format characters
    return name.strip(" .")[:limit] or "upload"


class Store:
    """Files live in one private folder, under names the server invents."""

    def __init__(self, root):
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)

    def save(self, data, kind):
        file_id = secrets.token_urlsafe(16)
        with (self.root / (file_id + FORMATS[kind][1])).open("xb") as handle:  # "x" never overwrites
            handle.write(data)
        return file_id

    def find(self, file_id):
        if not ID_PATTERN.fullmatch(file_id):  # check before the string gets anywhere near a path
            return None
        for media_type, extension in FORMATS.values():
            path = self.root / (file_id + extension)
            if path.is_file():
                return path, media_type
        return None

Four things in that listing are worth a closer look:

• display_name keeps only the last path segment, removes control and format characters (the category that includes the right-to-left override), trims dots and spaces, and cuts the result to 80 characters. Use it for display, never for storage.
• Store.save opens the file with mode "xb", where the x means create the file and fail if it already exists. A collision between two 128-bit random names is not a realistic risk, but a write that cannot silently overwrite is cheap insurance. The extension comes from the format the server detected, never from the client.
• Store.find checks the identifier against a pattern before it builds a path. A download URL is as hostile as an upload name: ..\..\secret is a legal path segment on Windows. Step 8 proves the check works.
• Nothing serves this folder directly. Files come back only through a route you write, which is the subject of Step 8.

Common mistakes with names

Sanitizing the name and then using it anyway is the most common one. Even a careful sanitizer inherits every file system quirk in the table above. If you must keep the original name, store it in a database column with a parameterized query (How to Prevent SQL Injection in Python With Parameterized Queries) and escape it when you show it. If your service also unpacks archives, How to Prevent Path Traversal in Python File Downloads and Archive Extraction covers the member names inside them.

Step 5: Put a ceiling on the size of the request

Step 3 ended with a warning that app_v2.py checks the size after reading the file. To see why that matters, measure what a 200 MiB upload costs a server under four designs. Starlette’s form parser stores each uploaded file in a SpooledTemporaryFile that stays in memory up to 1 MiB and moves to disk beyond that. Starlette’s documentation is explicit that the size limit it provides for form parts does not cover files. It says the parameter “limits the size of each non-file field; uploaded files are spooled to temporary storage and are not limited by max_part_size”. FastAPI’s documentation describes the same spooling, and warns about the alternative of declaring the parameter as bytes: “Keep in mind that this means that the whole contents will be stored in memory.”

Measure four endpoints

Save this as limit_server.py. It starts one of four endpoints in its own process: naive only reports the size, checked reads the whole file and then rejects it, bytes declares the parameter as bytes, and limited adds Starlette’s RequestBodyLimitMiddleware with a 6 MiB ceiling.

# limit_server.py - four upload endpoints, one per process: python limit_server.py naive|checked|bytes|limited PORT
import sys

import uvicorn
from fastapi import FastAPI, File, HTTPException, UploadFile
from starlette.middleware.body_limit import RequestBodyLimitMiddleware

MAX_REQUEST_BYTES = 6 * 1024 * 1024


def make_app(kind):
    app = FastAPI()

    if kind == "bytes":

        @app.post("/upload")
        async def upload_bytes(file: bytes = File()):
            return {"received": len(file)}

    elif kind == "checked":

        @app.post("/upload")
        async def upload_checked(file: UploadFile):
            data = await file.read()  # the size check comes after the whole upload has been read
            if len(data) > MAX_REQUEST_BYTES:
                raise HTTPException(413, "file too large")
            return {"received": len(data)}

    else:

        @app.post("/upload")
        async def upload(file: UploadFile):
            return {"received": file.size}

    if kind == "limited":
        app.add_middleware(RequestBodyLimitMiddleware, max_body_size=MAX_REQUEST_BYTES)
    return app


if __name__ == "__main__":
    uvicorn.run(make_app(sys.argv[1]), host="127.0.0.1", port=int(sys.argv[2]), log_level="critical")

Now the measuring script. Save it as step_limit.py. It sends a 200 MiB upload twice, once with a Content-Length header and once as a chunked transfer, which has no declared length, and records the response, the time, how many megabytes the server process wrote to disk, and its peak memory.

# step_limit.py - what does a big upload cost the server with and without a request body limit?
import subprocess
import sys
import time
from pathlib import Path

import psutil
import requests

LAB = Path(__file__).resolve().parent
MIB = 1024 * 1024
BOUNDARY = "----labboundary"
HEAD = (f'--{BOUNDARY}\r\nContent-Disposition: form-data; name="file"; filename="big.bin"\r\n'
        "Content-Type: application/octet-stream\r\n\r\n").encode()
TAIL = f"\r\n--{BOUNDARY}--\r\n".encode()


def chunked_body(size):
    yield HEAD  # no Content-Length is sent for a generator: requests uses Transfer-Encoding: chunked
    block = b"\0" * MIB
    for _ in range(size // MIB):
        yield block
    yield TAIL


def sized_body(size):
    return HEAD + b"\0" * size + TAIL  # requests adds a Content-Length header


def start(kind, port):
    proc = subprocess.Popen([sys.executable, str(LAB / "limit_server.py"), kind, str(port)])
    for _ in range(100):
        try:
            requests.get(f"http://127.0.0.1:{port}/openapi.json", timeout=1)
            return proc
        except requests.RequestException:
            time.sleep(0.1)
    raise RuntimeError("server did not start")


def tree(proc):
    root = psutil.Process(proc.pid)
    return [root, *root.children(recursive=True)]


def trial(kind, port, size_mib, mode):
    proc = start(kind, port)
    wrote_before = sum(p.io_counters().write_bytes for p in tree(proc))
    headers = {"Content-Type": f"multipart/form-data; boundary={BOUNDARY}"}
    started = time.perf_counter()
    try:
        data = chunked_body(size_mib * MIB) if mode == "chunked" else sized_body(size_mib * MIB)
        reply = requests.post(f"http://127.0.0.1:{port}/upload", data=data, headers=headers, timeout=120)
        outcome = f"HTTP {reply.status_code} {reply.text[:38]}"
    except requests.RequestException as exc:
        outcome = type(exc).__name__
    seconds = time.perf_counter() - started
    members = tree(proc)
    wrote = (sum(p.io_counters().write_bytes for p in members) - wrote_before) / MIB
    peak = max(p.memory_info().peak_wset for p in members) / MIB
    for p in members:
        p.kill()
    print(f"{kind:8} {mode:8} {size_mib:4} MiB  {outcome:50} {seconds:4.1f} s  server wrote {wrote:6.1f} MiB  peak memory {peak:4.0f} MiB")


def main():
    port = 8741
    for kind, size_mib, mode in [
        ("naive", 1, "length"),
        ("naive", 200, "length"),
        ("naive", 200, "chunked"),
        ("checked", 200, "length"),
        ("bytes", 200, "length"),
        ("limited", 1, "length"),
        ("limited", 200, "length"),
        ("limited", 200, "chunked"),
    ]:
        trial(kind, port, size_mib, mode)
        port += 1


if __name__ == "__main__":
    main()

python step_limit.py

naive    length      1 MiB  HTTP 200 {"received":1048576}                       0.0 s  server wrote    0.0 MiB  peak memory   53 MiB
naive    length    200 MiB  HTTP 200 {"received":209715200}                     0.5 s  server wrote  200.0 MiB  peak memory   53 MiB
naive    chunked   200 MiB  HTTP 200 {"received":209715200}                     0.4 s  server wrote  200.0 MiB  peak memory   54 MiB
checked  length    200 MiB  HTTP 413 {"detail":"file too large"}                0.6 s  server wrote  200.0 MiB  peak memory  252 MiB
bytes    length    200 MiB  HTTP 200 {"received":209715200}                     0.6 s  server wrote  200.0 MiB  peak memory  252 MiB
limited  length      1 MiB  HTTP 200 {"received":1048576}                       0.0 s  server wrote    0.0 MiB  peak memory   53 MiB
limited  length    200 MiB  HTTP 413 Content Too Large                          0.2 s  server wrote    0.0 MiB  peak memory   51 MiB
limited  chunked   200 MiB  HTTP 413 {"detail":"Content Too Large"}             0.4 s  server wrote    5.9 MiB  peak memory   54 MiB

Read it as four stories:

• naive. A 1 MiB upload wrote nothing to disk, because it stayed in the spool’s memory. A 200 MiB upload, sent either way, made the server write 200 MiB to a temporary file, and it answered 200. Memory stayed near 53 MiB, so spooling protects RAM and does nothing for disk.
• checked. This is the shape of app_v2.py. The answer was 413, but only after the server had written 200 MiB to disk and then read the file into memory, which pushed its peak to 252 MiB. The client received an error for work the server had already done.
• bytes. The same 252 MiB peak, with an answer of 200. Declaring the parameter as bytes is fine for small files and a memory problem for large ones.
• limited. With the middleware, a 1 MiB upload still worked. The 200 MiB upload with a declared length was refused with 413 and plain text before the server wrote a byte. The chunked upload, which declares no length, was counted as it arrived and refused with a JSON 413 after the server had written 5.9 MiB, just under the 6 MiB ceiling.

The fix: a body limit

Starlette’s documentation describes the tool: “To limit the complete request body, including uploaded files and multipart encoding overhead, configure max_body_size on the application, router, mount, or route, or use RequestBodyLimitMiddleware.” With FastAPI 0.146.0 there is a catch. Save this as step_kwarg.py and run it:

# step_kwarg.py - Starlette documents max_body_size on the application, but does FastAPI accept it?
from fastapi import FastAPI

try:
    FastAPI(max_body_size=1000)
except TypeError as exc:
    print(type(exc).__name__ + ":", exc)

python step_kwarg.py

TypeError: FastAPI.__init__() got an unexpected keyword argument 'max_body_size'

FastAPI’s constructor does not accept the keyword that Starlette’s does, and it says so loudly instead of ignoring it. Add the middleware class directly, as limit_server.py did with app.add_middleware(RequestBodyLimitMiddleware, max_body_size=...). The finished app in Step 8 does the same.

Choosing the number

The ceiling applies to the whole request, so it has to be a little larger than the largest file you will accept, to leave room for the multipart framing. The finished app allows 5 MiB per file and sets the request ceiling at 5 MiB plus 64 KiB. Both numbers are choices for this lab; tune them to your product. It is also worth setting a similar limit in your reverse proxy or load balancer, so that oversized bodies never reach Python at all.

A platform note on the measurements

step_limit.py and step_bomb.py use two psutil features that depend on the operating system: peak_wset, the peak working set, exists on Windows only, and per-process I/O counters are not available on macOS. On Linux you can read the process’s peak resident memory from the resource module, or watch the process in top.

Step 6: Decide what the file is from its bytes

Everything the client says about a file, its name and its Content-Type, is a claim. The bytes are the only evidence you have. Add the first function that looks at them. It reads the first sixteen bytes and names the format they announce, and it knows exactly three.

# uploadsafe.py (continued)
def sniff(head):
    """Name the format the first bytes claim to be, or None. The client's Content-Type and file name play no part."""
    if head.startswith(b"\x89PNG\r\n\x1a\n"):
        return "PNG"
    if head.startswith(b"\xff\xd8\xff"):
        return "JPEG"
    if head[:4] == b"RIFF" and head[8:12] == b"WEBP":
        return "WEBP"
    return None

A PNG starts with an eight-byte signature, a JPEG with FF D8 FF, and a WebP is a RIFF container whose bytes 8 to 11 spell WEBP. HTML, SVG, XHTML and GIF all return None, so they are refused by default: the function is an allowlist, not a blocklist. That is a deliberate policy decision, and it is the most effective single change in this tutorial: HTML, XHTML and SVG, the three page formats the matrix used, are refused without anyone having to name them on a list.

Why the signature is only the first gate

OWASP is blunt about signature checks: “This should not be used on its own, as bypassing it is pretty common and easy.” The glued_on_page attack shows why. It starts with a perfect PNG signature and is a perfect PNG all the way to its last chunk, and then it carries a web page. A signature check says yes. The next step therefore does not stop at the header; it decodes the whole file and rebuilds it.

One more safeguard comes from Pillow itself. Image.open accepts a list of formats to try, and the documentation explains it: “A list or tuple of formats to attempt to load the file in. This can be used to restrict the set of formats checked.” The next listing passes only the format it sniffed, so a file that announced itself as PNG is offered to the PNG decoder and nothing else.

Step 7: Rebuild the image inside hard limits

Why a 95 KiB file can cost 500 MiB

Pillow has a built-in guard against decompression bombs, and its behavior is easy to misread. The documentation says: “Pillow will issue a DecompressionBombWarning if the number of pixels in an image is over a certain limit, MAX_IMAGE_PIXELS.” Then comes the part that matters: “If the number of pixels is greater than twice MAX_IMAGE_PIXELS, then a DecompressionBombError will be raised instead.” The default is 89,478,485 pixels, which the documentation calls “approximately 0.25GB for a 24-bit (3 bpp) image.” Between one and two times the limit, you get a warning and the image decodes anyway.

Save this as step_bomb.py. It makes solid-grey PNG canvases of several sizes, then opens each one in a fresh process, converts it to RGB as a thumbnailer would, and reports the warnings, errors and peak memory.

# step_bomb.py - what does Pillow warn about, refuse and allocate for tiny PNGs with huge canvases?
import subprocess
import sys
from pathlib import Path

from PIL import Image

SCRATCH = Path(__file__).resolve().parent / "scratch"

CHILD = """
import sys, time, warnings
import psutil
from PIL import Image
path, cap, strict = sys.argv[1], int(sys.argv[2]), sys.argv[3] == "strict"
if cap:
    Image.MAX_IMAGE_PIXELS = cap
started = time.perf_counter()
with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    if strict:
        warnings.simplefilter("error", Image.DecompressionBombWarning)
    try:
        image = Image.open(path)
        print("  opened", image.format, image.size, image.mode)
        rgb = image.convert("RGB")
        print("  decoded and converted to RGB")
    except Exception as exc:
        print("  raised", type(exc).__name__)
for note in caught:
    print("  warning", note.category.__name__)
peak = psutil.Process().memory_info().peak_wset / 2**20
print(f"  peak working set {peak:.0f} MiB, {time.perf_counter() - started:.2f} s")
"""


def canvas(side):
    path = SCRATCH / f"canvas_{side}.png"
    if not path.exists():
        Image.new("L", (side, side), 0).save(path, compress_level=9)  # one solid grey: compresses about 1000 to 1
    return path


def run(side, cap, mode, label):
    path = canvas(side)
    print(f"== {side} by {side} = {side * side / 1e6:.0f} megapixels, {path.stat().st_size / 1024:.0f} KiB on disk, {label}")
    done = subprocess.run([sys.executable, "-c", CHILD, str(path), str(cap), mode], capture_output=True, text=True)
    print(done.stdout.rstrip())


def main():
    SCRATCH.mkdir(exist_ok=True)
    print("Image.MAX_IMAGE_PIXELS:", Image.MAX_IMAGE_PIXELS, "| twice that:", 2 * Image.MAX_IMAGE_PIXELS)
    for side in (1000, 10000, 13000, 14200):
        run(side, 0, "default", "default settings")
    run(6000, 25_000_000, "strict", "limit 25,000,000, warnings as errors")
    run(6000, 25_000_000, "default", "limit 25,000,000, warnings left alone")
    run(10000, 25_000_000, "default", "limit 25,000,000, warnings left alone")


if __name__ == "__main__":
    main()

python step_bomb.py

Image.MAX_IMAGE_PIXELS: 89478485 | twice that: 178956970
== 1000 by 1000 = 1 megapixels, 1 KiB on disk, default settings
  opened PNG (1000, 1000) L
  decoded and converted to RGB
  peak working set 29 MiB, 0.01 s
== 10000 by 10000 = 100 megapixels, 95 KiB on disk, default settings
  opened PNG (10000, 10000) L
  decoded and converted to RGB
  warning DecompressionBombWarning
  peak working set 501 MiB, 0.18 s
== 13000 by 13000 = 169 megapixels, 160 KiB on disk, default settings
  opened PNG (13000, 13000) L
  decoded and converted to RGB
  warning DecompressionBombWarning
  peak working set 830 MiB, 0.30 s
== 14200 by 14200 = 202 megapixels, 191 KiB on disk, default settings
  raised DecompressionBombError
  peak working set 24 MiB, 0.01 s
== 6000 by 6000 = 36 megapixels, 34 KiB on disk, limit 25,000,000, warnings as errors
  raised DecompressionBombWarning
  peak working set 24 MiB, 0.01 s
== 6000 by 6000 = 36 megapixels, 34 KiB on disk, limit 25,000,000, warnings left alone
  opened PNG (6000, 6000) L
  decoded and converted to RGB
  warning DecompressionBombWarning
  peak working set 196 MiB, 0.07 s
== 10000 by 10000 = 100 megapixels, 95 KiB on disk, limit 25,000,000, warnings left alone
  raised DecompressionBombError
  peak working set 24 MiB, 0.01 s

With default settings, a 95 KiB file that declared 100 megapixels triggered the warning, decoded, and pushed the process to 501 MiB. A 160 KiB file at 169 megapixels reached 830 MiB. Only the 202-megapixel file, beyond twice the limit, was refused, and it was refused before anything was allocated. Lowering the limit to 25 million pixels moves the zones: a 36-megapixel file is a warning and 196 MiB if you leave warnings alone, an exception if you turn warnings into errors, and a 100-megapixel file is refused outright. The lesson is that you should not rely on a warning to stop anything. The decoder in the next listing reads the image header, which costs almost nothing, and compares the pixel count with its own limit before it decodes a single pixel.

The function that rebuilds an upload

Here is the heart of the service. It takes the bytes of an upload and either returns brand-new bytes plus the format name, or raises Rejected with an HTTP status and a reason.

# uploadsafe.py (continued)
def clean_image(data):
    """Decode the upload inside hard limits and return (brand-new bytes, format name)."""
    kind = sniff(data[:16])
    if kind is None:
        raise Rejected(415, "only PNG, JPEG and WebP images are accepted")
    try:
        with Image.open(io.BytesIO(data), formats=[kind]) as image:  # only the decoder for the format we sniffed
            if image.width * image.height > MAX_PIXELS:  # open() read the header only, nothing is allocated yet
                raise Rejected(422, "the image has too many pixels")
            if getattr(image, "n_frames", 1) > 1:
                raise Rejected(422, "animated images are not accepted")
            image.load()  # decode everything now, so a corrupt or truncated file fails here
            upright = ImageOps.exif_transpose(image)  # apply the rotation flag before we throw the metadata away
            mode = "RGBA" if kind != "JPEG" and upright.has_transparency_data else "RGB"
            pixels = upright.convert(mode)
    except Rejected:
        raise
    except Image.DecompressionBombError as exc:  # Pillow's own guard, which only raises beyond twice its limit
        raise Rejected(422, "the image has too many pixels") from exc
    except Exception as exc:  # any decoder failure means the file is not one we want
        raise Rejected(422, "the image could not be decoded") from exc
    fresh = Image.frombytes(mode, pixels.size, pixels.tobytes())  # a new image: no Exif, ICC profile, text chunks or trailer
    out = io.BytesIO()
    fresh.save(out, format=kind, **SAVE_OPTIONS.get(kind, {}))
    return out.getvalue(), kind

Walk through it in order:

1. Sniff. No recognized signature means a 415 right away.
2. Open with one decoder. Image.open reads only the header here, so the size is known before any pixel is allocated.
3. Check the pixels. More than 25 million pixels is a 422. This comes first, before the expensive part.
4. Refuse animation. An animated PNG or WebP would silently lose every frame but the first, so the policy is to refuse it. A real avatar service might accept animation and handle it properly.
5. Decode everything. image.load() forces the full decode now. A corrupt or truncated file fails here, inside the try, with a clear 422 instead of failing later in some other code.
6. Apply the rotation. Phones store pictures sideways and record the rotation in the Exif block. Pillow’s documentation for exif_transpose says: “If an image has an EXIF Orientation tag, other than 1, transpose the image accordingly, and remove the orientation data.” If you strip the metadata without applying the rotation first, portraits come out sideways.
7. Pick a mode and convert. JPEG has no transparency, so it becomes RGB. PNG and WebP keep an alpha channel only if the image really has transparency.
8. Rebuild. Image.frombytes creates a new image object from the raw pixels. A fresh object has no info dictionary, so there is nowhere for Exif, a colour profile, text chunks or trailing bytes to ride along. Saving it with the sniffed format writes a file that contains the pixels and little else.

The error handling has three arms. Rejected is re-raised as it is. Image.DecompressionBombError, Pillow’s own guard beyond twice its limit, is mapped to the same too-many-pixels answer. Anything else the decoder throws becomes a generic 422, because for an upload service every decoder failure means the same thing: not a file we want.

What the rebuild removes

Save this as step_rebuild.py. For four inputs, it prints what the file contained before and after clean_image.

# step_rebuild.py - what does clean_image() keep, and what does it throw away?
import io

from PIL import Image, ImageCms

import attacks
from uploadsafe import clean_image


def facts(data):
    image = Image.open(io.BytesIO(data))
    exif = image.getexif()
    return {
        "bytes": len(data),
        "size": image.size,
        "glued-on page": attacks.TRAILER in data,
        "Exif tags": len(exif),
        "GPS block": bool(exif.get_ifd(0x8825)),
        "ICC profile": bool(image.info.get("icc_profile")),
        "PNG text chunk": "Comment" in image.info,
    }


def show(label, data):
    out, kind = clean_image(data)
    before, after = facts(data), facts(out)
    print(f"== {label} ({kind})")
    for key in before:
        marker = "" if before[key] == after[key] else "   <- changed"
        print(f"   {key:15} {str(before[key]):>14} -> {str(after[key]):<14}{marker}")


profile = ImageCms.ImageCmsProfile(ImageCms.createProfile("sRGB")).tobytes()
show("PNG with a web page glued on after the last chunk", attacks.png_with_trailer())
show("phone JPEG: rotation flag, camera make and GPS position", attacks.jpeg_with_gps(size=(40, 20), orientation=6))
show("PNG with an embedded sRGB colour profile", attacks.png_bytes(icc_profile=profile))
show("PNG with a text chunk that holds a script", attacks.png_with_text_chunk())

python step_rebuild.py

== PNG with a web page glued on after the last chunk (PNG)
   bytes                      158 -> 75               <- changed
   size                    (8, 8) -> (8, 8)        
   glued-on page             True -> False            <- changed
   Exif tags                    0 -> 0             
   GPS block                False -> False         
   ICC profile              False -> False         
   PNG text chunk           False -> False         
== phone JPEG: rotation flag, camera make and GPS position (JPEG)
   bytes                      823 -> 654              <- changed
   size                  (40, 20) -> (20, 40)         <- changed
   glued-on page            False -> False         
   Exif tags                    3 -> 0                <- changed
   GPS block                 True -> False            <- changed
   ICC profile              False -> False         
   PNG text chunk           False -> False         
== PNG with an embedded sRGB colour profile (PNG)
   bytes                      461 -> 75               <- changed
   size                    (8, 8) -> (8, 8)        
   glued-on page            False -> False         
   Exif tags                    0 -> 0             
   GPS block                False -> False         
   ICC profile               True -> False            <- changed
   PNG text chunk           False -> False         
== PNG with a text chunk that holds a script (PNG)
   bytes                      178 -> 75               <- changed
   size                    (8, 8) -> (8, 8)        
   glued-on page             True -> False            <- changed
   Exif tags                    0 -> 0             
   GPS block                False -> False         
   ICC profile              False -> False         
   PNG text chunk            True -> False            <- changed

The glued-on page and the script-bearing text chunk are gone. The phone JPEG lost its camera make and its GPS block, and the rotation flag was applied first: a picture stored as 40 wide and 20 high came back 20 wide and 40 high, which is how a viewer would have shown it. The embedded colour profile was dropped too. The byte counts for that sample can differ by a few bytes when you run it, because the profile Pillow generates carries its creation time in its header. Losing the profile is a real trade-off: it can change how wide-gamut photos look, and the same goes for recompressing a JPEG at quality 85. These are choices you can revisit, but each one gives something back to an attacker or to a privacy leak if you reverse it carelessly.

Be modest about what this proves. OWASP’s advice on images is to decode and re-encode, then add a caution: “Rewriting does not guarantee that all malicious content is removed, and the image processor itself handles untrusted input.” It continues: “Use an up-to-date library with restricted formats, resource limits, and sandboxing.” In this lab that means pinning Pillow, accepting three formats, limiting pixels and request size, and, in production, thinking about running the decoder in a separate low-privilege process.

What the rebuild costs

A legitimate large photo is not free to process. Save this as step_cost.py. It builds a 6000 by 4000 pixel JPEG, 24 megapixels, which is just under the ceiling, and measures clean_image in a separate process. Then it serves the function two ways, directly on the event loop and through asyncio.to_thread, and pings the server while a 24-megapixel upload is being cleaned.

# step_cost.py - what does cleaning a legitimate 24-megapixel photo cost, and does a worker thread keep the server responsive?
import asyncio
import io
import statistics
import subprocess
import sys
import threading
import time
from pathlib import Path

import requests
from fastapi import FastAPI, UploadFile
from PIL import Image

from labserver import LiveServer
from uploadsafe import clean_image

LAB = Path(__file__).resolve().parent
SCRATCH = LAB / "scratch"


def photo(width=6000, height=4000):
    base = Image.radial_gradient("L").resize((width, height))  # smooth, so the JPEG stays small
    flipped = base.transpose(Image.Transpose.FLIP_LEFT_RIGHT)
    buffer = io.BytesIO()
    Image.merge("RGB", (base, flipped, base)).save(buffer, "JPEG", quality=85)
    return buffer.getvalue()


CHILD = """
import sys, time
import psutil
from uploadsafe import clean_image
data = open(sys.argv[1], "rb").read()
started = time.perf_counter()
out, kind = clean_image(data)
seconds = time.perf_counter() - started
peak = psutil.Process().memory_info().peak_wset / 2**20
print(f"clean_image: {len(data) / 1024:.0f} KiB in, {len(out) / 1024:.0f} KiB out, {seconds:.2f} s, peak working set {peak:.0f} MiB")
"""


def make_app(offload):
    app = FastAPI()

    @app.get("/ping")
    async def ping():
        return {"ok": True}

    @app.post("/upload")
    async def upload(file: UploadFile):
        data = await file.read()
        if offload:
            await asyncio.to_thread(clean_image, data)
        else:
            clean_image(data)  # runs on the event loop
        return {"ok": True}

    return app


def worst_ping(offload, port, data):
    with LiveServer(make_app(offload), port) as server:
        requests.get(server.url + "/ping")
        busy = threading.Thread(target=lambda: requests.post(server.url + "/upload", files={"file": ("p.jpg", data, "image/jpeg")}))
        times = []
        busy.start()
        while busy.is_alive():
            started = time.perf_counter()
            requests.get(server.url + "/ping", timeout=30)
            times.append((time.perf_counter() - started) * 1000)
        busy.join()
    return max(times), statistics.median(times), len(times)


def main():
    SCRATCH.mkdir(exist_ok=True)
    data = photo()
    path = SCRATCH / "photo_24mp.jpg"
    path.write_bytes(data)
    done = subprocess.run([sys.executable, "-c", CHILD, str(path)], capture_output=True, text=True, cwd=LAB)
    print(done.stdout.strip() or done.stderr.strip()[-300:])
    for offload, port in ((False, 8771), (True, 8772)):
        worst, median, count = worst_ping(offload, port, data)
        label = "asyncio.to_thread(clean_image, data)" if offload else "clean_image(data) on the event loop   "
        print(f"{label}: {count:3d} pings while one 24 MP upload is cleaned, median {median:5.1f} ms, worst {worst:6.1f} ms")


if __name__ == "__main__":
    main()

python step_cost.py

clean_image: 550 KiB in, 564 KiB out, 0.21 s, peak working set 462 MiB
clean_image(data) on the event loop   :   2 pings while one 24 MP upload is cleaned, median 104.7 ms, worst  205.7 ms
asyncio.to_thread(clean_image, data):  70 pings while one 24 MP upload is cleaned, median   1.9 ms, worst   45.3 ms

A 550 KiB photo took roughly a fifth of a second and peaked at about 460 MiB, because several copies of the pixels exist at the same time during the rebuild. Called directly inside an async route, the function froze the event loop: only two pings got through during the upload, with a median of about 100 milliseconds. Through asyncio.to_thread the server answered 70 pings with a median of about 2 milliseconds. So run the rebuild in a worker thread, and remember the memory: ten simultaneous large uploads can hold several gigabytes. Choose the pixel ceiling for your product (an avatar rarely needs more than a few megapixels), and cap how many rebuilds run at once. How to Add Backpressure and Load Shedding to a Python Service shows how to do that.

Step 8: Serve files safely, and wire everything into the finished app

The finished service

All the pieces now exist. Save this as app_v3.py. It adds the body limit from Step 5, caps the file at 5 MiB, rebuilds the image off the event loop, stores it under an invented name, and serves it through a route you control.

# app_v3.py - the upload endpoint with every layer in place
import asyncio

from fastapi import FastAPI, HTTPException, Response, UploadFile
from starlette.middleware.body_limit import RequestBodyLimitMiddleware

from uploadsafe import FORMATS, Rejected, Store, clean_image, display_name

MAX_FILE_BYTES = 5 * 1024 * 1024
MAX_REQUEST_BYTES = MAX_FILE_BYTES + 64 * 1024  # the file plus room for the multipart framing
SAFE_HEADERS = {
    "X-Content-Type-Options": "nosniff",
    "Content-Security-Policy": "default-src 'none'; sandbox",
    "Cache-Control": "private, max-age=3600",
}


def make_app(store_root):
    store = Store(store_root)
    app = FastAPI()
    app.add_middleware(RequestBodyLimitMiddleware, max_body_size=MAX_REQUEST_BYTES)

    @app.post("/upload", status_code=201)
    async def upload(file: UploadFile):
        data = await file.read(MAX_FILE_BYTES + 1)
        if len(data) > MAX_FILE_BYTES:
            raise HTTPException(413, "file too large")
        try:
            clean, kind = await asyncio.to_thread(clean_image, data)  # decoding is CPU work: keep it off the event loop
        except Rejected as problem:
            raise HTTPException(problem.status, problem.reason) from None
        file_id = store.save(clean, kind)
        return {
            "id": file_id,
            "url": f"/files/{file_id}",
            "type": FORMATS[kind][0],
            "bytes": len(clean),
            "name": display_name(file.filename),
        }

    @app.get("/files/{file_id}")
    async def download(file_id: str):
        found = store.find(file_id)
        if found is None:
            raise HTTPException(404, "no such file")
        path, media_type = found
        return Response(path.read_bytes(), media_type=media_type, headers=SAFE_HEADERS)

    return app

Read the upload route from top to bottom. The middleware has already refused any request over 5 MiB plus 64 KiB. file.read(MAX_FILE_BYTES + 1) then reads at most one byte more than the allowed size, so a file just over the cap is detected without reading an unbounded amount. That check duplicates the middleware on purpose: if someone later raises the request ceiling for another route, the file cap still holds. clean_image runs in a worker thread. A Rejected problem becomes an HTTPException with the status that clean_image chose. The reply gives back an identifier, a URL, the detected type, the stored size, and the sanitized display name.

The download route is deliberately small. It looks the identifier up in the store, answers 404 for anything that is not an identifier the server made, and returns the stored bytes with a media type taken from the store, not from the request and not from a file extension.

Headers that tell the browser not to run anything

Three headers go on every download. Each one covers a failure that the others do not.

• X-Content-Type-Options: nosniff. MDN explains that the header “allows you to avoid MIME type sniffing by specifying that the MIME types are deliberately configured.” With it, the browser uses the declared type as it is, instead of examining the content and deciding for itself.
• Content-Security-Policy: default-src 'none'; sandbox. The default-src 'none' part forbids the document from loading anything else. The sandbox directive, in MDN’s words, “applies restrictions to a page’s actions including preventing popups, preventing the execution of plugins and scripts, and enforcing a same-origin policy.” It is a second lock for the case where a file ever gets served as something executable.
• Cache-Control: private, max-age=3600. This keeps shared caches, such as a CDN, from storing the file, which is what you want until you decide that a file is public.

If you later accept formats that you should not render inline, such as PDF or Office documents, serve them with Content-Disposition: attachment. MDN’s example makes the effect plain: an HTML file sent that way “will be downloaded rather than displayed in the browser.” Those formats can carry active content of their own, so they also call for the antivirus scan and content disarm steps that OWASP lists.

Check the headers, the identifier rule and the SVG sandbox

Save this as step_headers.py. It uploads a PNG called ../../x.png while claiming to be text/html, prints the reply and the download headers, asks for four identifiers the server never issued, and then runs a small experiment: the same SVG with a script, served once as image/svg+xml and once with Content-Security-Policy: sandbox.

# step_headers.py - what the safe download route sends, and what Content-Security-Policy: sandbox does to an SVG
import tempfile
from pathlib import Path

import requests
from fastapi import FastAPI, Response

import app_v3
import attacks
from browser import script_ran
from labserver import LiveServer


def download_headers():
    with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmp:
        with LiveServer(app_v3.make_app(Path(tmp) / "files"), 8761) as server:
            files = {"file": ("../../x.png", attacks.png_bytes(), "text/html")}
            created = requests.post(server.url + "/upload", files=files).json()
            print("upload reply:", created)
            reply = requests.get(server.url + created["url"])
            print("download status:", reply.status_code)
            for name in ("content-type", "x-content-type-options", "content-security-policy", "cache-control"):
                print(f"  {name}: {reply.headers[name]}")
            for odd in ("..%5C..%5Csecret", "..", "abcdefghijklmnopqrstuv", "x" * 23):
                print(f"GET /files/{odd[:24]:24} -> {requests.get(server.url + '/files/' + odd).status_code}")


svg_app = FastAPI()


@svg_app.get("/plain.svg")
def plain():
    return Response(attacks.SVG_IMAGE, media_type="image/svg+xml")


@svg_app.get("/sandboxed.svg")
def sandboxed():
    return Response(attacks.SVG_IMAGE, media_type="image/svg+xml", headers={"Content-Security-Policy": "sandbox"})


def svg_experiment():
    with LiveServer(svg_app, 8762) as server:
        for path, label in (("/plain.svg", "SVG with a script, served as image/svg+xml"),
                            ("/sandboxed.svg", "same SVG plus Content-Security-Policy: sandbox")):
            print(f"{label}: script ran in the app's origin: {'yes' if script_ran(server.url + path) else 'no'}")


if __name__ == "__main__":
    download_headers()
    svg_experiment()

python step_headers.py

upload reply: {'id': 'vZX54f7yLr5bS9YKYjcMVg', 'url': '/files/vZX54f7yLr5bS9YKYjcMVg', 'type': 'image/png', 'bytes': 75, 'name': 'x.png'}
download status: 200
  content-type: image/png
  x-content-type-options: nosniff
  content-security-policy: default-src 'none'; sandbox
  cache-control: private, max-age=3600
GET /files/..%5C..%5Csecret         -> 404
GET /files/..                       -> 404
GET /files/abcdefghijklmnopqrstuv   -> 404
GET /files/xxxxxxxxxxxxxxxxxxxxxxx  -> 404
SVG with a script, served as image/svg+xml: script ran in the app's origin: yes
same SVG plus Content-Security-Policy: sandbox: script ran in the app's origin: no

The upload was accepted because its bytes were a PNG; both the traversal-style name and the text/html claim were ignored, and the reply shows the display name reduced to x.png. The download carries the three headers. All four bad identifiers got 404, including the percent-encoded backslash traversal. Three do not match the 22-character pattern the store issues, and the fourth matches the pattern but names no file. The last two lines are the SVG experiment, measured in Edge 155: with no policy the script ran in the application’s origin, and with sandbox it did not.

If you must accept SVG

This service does not accept SVG, and that is the strongest policy. If your product needs it, treat the sandbox header as one layer of several: sanitize the file with a library built for SVG, serve it from an origin that does not share cookies with your application, and decide deliberately whether it may be shown inline at all. The experiment above shows only that the header stopped this one script in this one browser.

Step 9: Run the finished service and try it with curl

Save these two lines as main.py. They create the app with a folder called data/files as the store.

# main.py - run the finished service with: uvicorn main:app
from app_v3 import make_app

app = make_app("data/files")

Start the server in one terminal. It listens on port 8000 by default.

uvicorn main:app

In a second terminal, create two sample files with the helpers you already have, then upload them. In Windows PowerShell, type curl.exe instead of curl, because curl is an alias for a different command there, and on macOS or Linux replace NUL with /dev/null. The lab ran the same commands against port 8765, so your identifier and port will differ.

python -c "import attacks; open('holiday.jpg','wb').write(attacks.jpeg_with_gps()); open('evil.html','wb').write(attacks.HTML_PAGE)"

$ curl -s -F "[email protected];type=image/jpeg" http://127.0.0.1:8765/upload
{"id":"W2Q2wY-h4R9DF8USBuBlVQ","url":"/files/W2Q2wY-h4R9DF8USBuBlVQ","type":"image/jpeg","bytes":654,"name":"holiday.jpg"}
$ curl -s -w "\nHTTP %{http_code}\n" -F "[email protected];type=image/png" http://127.0.0.1:8765/upload
{"detail":"only PNG, JPEG and WebP images are accepted"}
HTTP 415
$ curl -s -D - -o NUL http://127.0.0.1:8765/files/W2Q2wY-h4R9DF8USBuBlVQ
HTTP/1.1 200 OK
server: uvicorn
x-content-type-options: nosniff
content-security-policy: default-src 'none'; sandbox
cache-control: private, max-age=3600
content-length: 654
content-type: image/jpeg

The first upload is a phone-style JPEG with a GPS position. It came back as a stored JPEG of 654 bytes, down from 823, with an invented identifier and a display name. The second is a web page that claims to be a PNG. It got a 415 and a plain explanation. The third command fetches the stored image and prints only the headers, where the safe set is visible. Open the same URL in a browser to see the picture itself.

Step 10: Prove it with the matrix and the tests

The whole scoreboard

Run all three versions and print the grid. This repeats the browser checks, so it takes a while.

python matrix.py all

The end of the output is the grid:

attack              v1            v2            v3          
html_page           got through   stopped       stopped     
html_spoofed_type   got through   stopped       stopped     
html_trailing_dot   got through   got through   stopped     
xhtml_page          got through   got through   stopped     
svg_image           got through   stopped       stopped     
path_traversal      got through   stopped       stopped     
overwrite           got through   got through   stopped     
glued_on_page       got through   got through   stopped     
page_named_png      got through   got through   stopped     
gps_photo           got through   got through   stopped     
pixel_bomb          got through   got through   stopped     

Version 1 loses all eleven, version 2 loses seven, and version 3 loses none. You can also run python matrix.py v3 alone to see the evidence behind the last column.

== v3 on http://127.0.0.1:8753
html_page          stopped      refused with HTTP 415
html_spoofed_type  stopped      refused with HTTP 415
html_trailing_dot  stopped      refused with HTTP 415
xhtml_page         stopped      refused with HTTP 415
svg_image          stopped      refused with HTTP 415
path_traversal     stopped      HTTP 201, file written outside the upload folder: no
overwrite          stopped      the first user's file changed after a second upload with the same name: no
glued_on_page      stopped      accepted; the page glued onto the PNG is still in the stored file: no (opened in the browser, script ran: no)
page_named_png     stopped      refused with HTTP 415
gps_photo          stopped      accepted; GPS position still inside the stored file: no
pixel_bomb         stopped      a 95 KiB file that decodes to 100 megapixels: refused with HTTP 422
0 of 11 attacks got through

When the pixel-bomb file arrives, Pillow also prints a DecompressionBombWarning to the terminal. That is the default guard from Step 7 speaking up, and the check in clean_image is what refuses the file.

Look at the third-version lines that report an acceptance. path_traversal was accepted, because it is a valid PNG, and no file appeared outside the store, because the name was never used. glued_on_page and gps_photo were accepted too, and the stored files no longer contain the glued-on page or the GPS position. Accepting a harmless file and neutralizing what rode along is better behavior than refusing everything odd.

The tests

A matrix is a good way to explore, and tests are what stop a regression next month. Save the next three files. conftest.py lets you pick the app under test with --app, so the same policy tests run against all three versions. It turns a crash inside the app into a 500 response, so a hostile name shows up as a failed assertion instead of an error.

# conftest.py - run the same policy tests against app_v1, app_v2 or app_v3
import importlib

import pytest
from fastapi.testclient import TestClient


def pytest_addoption(parser):
    parser.addoption("--app", default="app_v3", help="app module under test: app_v1, app_v2 or app_v3")


@pytest.fixture
def site(tmp_path):
    return tmp_path / "site"


@pytest.fixture
def client(request, site):
    module = importlib.import_module(request.config.getoption("--app"))
    # a crash inside the app becomes a 500 response, so a bad name shows up as a failed assertion, not an error
    return TestClient(module.make_app(site / "uploads"), raise_server_exceptions=False)

Save the policy tests as test_upload_policy.py. They state what a safe endpoint must do, whichever version is behind it.

# test_upload_policy.py - what a safe upload endpoint must do, whichever app is under test
import pytest

import attacks

REFUSED = (400, 413, 415, 422)


def upload(client, name, content_type, body):
    return client.post("/upload", files={"file": (name, body, content_type)})


def stored(client, reply):
    return client.get(reply.json()["url"]).content


@pytest.mark.parametrize(
    "name, content_type, body",
    [
        ("evil.html", "text/html", attacks.HTML_PAGE),
        ("evil.html", "image/png", attacks.HTML_PAGE),
        ("evil.html.", "image/png", attacks.HTML_PAGE),
        ("evil.xhtml", "image/png", attacks.XHTML_PAGE),
        ("logo.svg", "image/svg+xml", attacks.SVG_IMAGE),
        ("photo.png", "image/png", b"<html><body>not a picture</body></html>"),
    ],
    ids=["html", "html-spoofed-type", "html-trailing-dot", "xhtml", "svg", "page-named-png"],
)
def test_a_web_page_is_never_stored(client, name, content_type, body):
    assert upload(client, name, content_type, body).status_code in REFUSED


def test_a_good_png_is_accepted_and_comes_back_as_an_image(client):
    reply = upload(client, "avatar.png", "image/png", attacks.png_bytes())
    assert reply.status_code in (200, 201)
    assert client.get(reply.json()["url"]).headers["content-type"].startswith("image/png")


def test_a_traversal_name_cannot_write_outside_the_upload_folder(client, site):
    upload(client, "../../escaped.txt", "image/png", attacks.png_bytes())
    assert not (site.parent / "escaped.txt").exists()


def test_the_same_file_name_does_not_replace_another_users_file(client):
    first = upload(client, "avatar.png", "image/png", attacks.png_bytes(color="red"))
    before = stored(client, first)
    upload(client, "avatar.png", "image/png", attacks.png_bytes(color="blue"))
    assert stored(client, first) == before


def test_a_page_glued_onto_a_png_does_not_survive(client):
    reply = upload(client, "pixel.png", "image/png", attacks.png_with_trailer())
    assert reply.status_code in (200, 201)
    assert attacks.TRAILER not in stored(client, reply)


def test_the_gps_position_does_not_survive(client):
    reply = upload(client, "holiday.jpg", "image/jpeg", attacks.jpeg_with_gps())
    assert reply.status_code in (200, 201)
    assert not attacks.has_gps(stored(client, reply))


def test_a_pixel_bomb_is_refused(client):
    assert upload(client, "canvas.png", "image/png", attacks.png_canvas(10000)).status_code in REFUSED


def test_a_body_over_the_limit_is_refused(client):
    assert upload(client, "big.png", "image/png", b"\0" * (6 * 1024 * 1024)).status_code == 413


def test_downloads_say_what_they_are_and_forbid_sniffing(client):
    reply = upload(client, "avatar.png", "image/png", attacks.png_bytes())
    download = client.get(reply.json()["url"])
    assert download.headers["x-content-type-options"] == "nosniff"
    assert "sandbox" in download.headers["content-security-policy"]

And save the unit tests for the building blocks as test_uploadsafe.py.

# test_uploadsafe.py - unit tests for the building blocks in uploadsafe.py
import io

import pytest
from PIL import Image, ImageCms

import attacks
from uploadsafe import Rejected, Store, clean_image, display_name, sniff


def test_sniff_reads_bytes_and_ignores_names():
    assert sniff(attacks.png_bytes()[:16]) == "PNG"
    assert sniff(attacks.jpeg_with_gps()[:16]) == "JPEG"
    assert sniff(attacks.HTML_PAGE[:16]) is None
    assert sniff(attacks.SVG_IMAGE[:16]) is None
    assert sniff(b"GIF89a" + bytes(10)) is None


@pytest.mark.parametrize("body", [attacks.HTML_PAGE, attacks.SVG_IMAGE, b"GIF89a" + bytes(20), b""], ids=["html", "svg", "gif", "empty"])
def test_other_formats_are_refused_with_415(body):
    with pytest.raises(Rejected) as problem:
        clean_image(body)
    assert problem.value.status == 415


def test_a_truncated_png_is_refused_with_422():
    data = attacks.png_bytes((256, 256))
    with pytest.raises(Rejected) as problem:
        clean_image(data[: len(data) // 2])
    assert problem.value.status == 422


def test_an_animated_webp_is_refused_with_422():
    frames = [Image.new("RGB", (16, 16), color) for color in ("red", "blue")]
    buffer = io.BytesIO()
    frames[0].save(buffer, "WEBP", save_all=True, append_images=frames[1:], duration=100)
    with pytest.raises(Rejected) as problem:
        clean_image(buffer.getvalue())
    assert problem.value.status == 422


def test_rotation_is_applied_and_every_exif_tag_is_gone():
    out, kind = clean_image(attacks.jpeg_with_gps(size=(40, 20), orientation=6))
    image = Image.open(io.BytesIO(out))
    assert kind == "JPEG" and image.size == (20, 40)
    assert dict(image.getexif()) == {}


def test_icc_profiles_and_text_chunks_do_not_survive():
    profile = ImageCms.ImageCmsProfile(ImageCms.createProfile("sRGB")).tobytes()
    with_profile = attacks.png_bytes(icc_profile=profile)
    assert Image.open(io.BytesIO(with_profile)).info.get("icc_profile")
    out, _ = clean_image(with_profile)
    assert "icc_profile" not in Image.open(io.BytesIO(out)).info
    out, _ = clean_image(attacks.png_with_text_chunk())
    assert not Image.open(io.BytesIO(out)).info.get("Comment")


def test_png_pixels_survive_exactly():
    source = Image.effect_noise((32, 32), 40).convert("RGB")
    buffer = io.BytesIO()
    source.save(buffer, "PNG")
    out, _ = clean_image(buffer.getvalue())
    assert Image.open(io.BytesIO(out)).tobytes() == source.tobytes()


def test_display_names_lose_paths_and_invisible_characters():
    assert display_name("../../photo.png") == "photo.png"
    assert display_name("C:\\Users\\me\\photo.png") == "photo.png"
    assert display_name("report" + chr(0x202E) + "gpj.exe") == "reportgpj.exe"
    assert display_name(None) == "upload"
    assert len(display_name("a" * 500 + ".png")) == 80


def test_the_store_only_opens_ids_it_made(tmp_path):
    store = Store(tmp_path / "files")
    file_id = store.save(b"data", "PNG")
    assert store.find(file_id)[1] == "image/png"
    for bad in ("..", "..\\..\\secret", "a" * 22, file_id + "x", file_id[:-1]):
        assert store.find(bad) is None

Run them against all three versions

python -m pytest -q --tb=no -rf --app=app_v1 test_upload_policy.py
python -m pytest -q --tb=no -rf --app=app_v2 test_upload_policy.py
python -m pytest -q --tb=no -rf --app=app_v3 test_upload_policy.py
python -m pytest -q --tb=no -rf test_uploadsafe.py

Version 1 fails 13 of the 14 policy tests. The list is long, so only the summary is shown:

FFFFFF.FFFFFFF [100%]
13 failed, 1 passed in 0.43s

Version 2 fails 8, and its list is the same list of gaps the matrix found:

..FF.F..FFFF.F [100%]
=== short test summary info ===
FAILED test_upload_policy.py::test_a_web_page_is_never_stored[html-trailing-dot] - assert 200 in (400, 413, 415, 422)
FAILED test_upload_policy.py::test_a_web_page_is_never_stored[xhtml] - assert 200 in (400, 413, 415, 422)
FAILED test_upload_policy.py::test_a_web_page_is_never_stored[page-named-png] - AssertionError: assert 200 in (400, 413, 415, 422)
FAILED test_upload_policy.py::test_the_same_file_name_does_not_replace_another_users_file - AssertionError: assert b'\x89PNG\r\n...END\xaeB`\x82' == b'\x89PNG\r\n...EN...
FAILED test_upload_policy.py::test_a_page_glued_onto_a_png_does_not_survive - assert b'<script>document.documentElement.setAttribute("data-xss", location.origin)</scri...
FAILED test_upload_policy.py::test_the_gps_position_does_not_survive - AssertionError: assert not True
FAILED test_upload_policy.py::test_a_pixel_bomb_is_refused - assert 200 in (400, 413, 415, 422)
FAILED test_upload_policy.py::test_downloads_say_what_they_are_and_forbid_sniffing - KeyError: 'x-content-type-options'
8 failed, 6 passed in 0.41s

Version 3 passes all 14. The warning is Pillow’s own DecompressionBombWarning from the pixel-bomb test, which is the default guard from Step 7 speaking up. Our own check is what refuses the file.

.............. [100%]
=== warnings summary ===
test_upload_policy.py::test_a_pixel_bomb_is_refused
  .venv\Lib\site-packages\PIL\Image.py:3593: DecompressionBombWarning: Image size (100000000 pixels) exceeds limit of 89478485 pixels, could be decompression bomb DOS attack.
    warnings.warn(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
14 passed, 1 warning in 0.36s

And the unit tests for the building blocks:

............ [100%]
12 passed in 0.08s

Make sure the tests can fail

A green run means something only if the tests can go red. The lab made four one-line breakages to the finished service, one at a time, and ran the policy and unit tests after each:

The last row is the lesson. With the middleware gone, the in-route check still answers 413 for a body over 5 MiB, so a test that looks at status codes cannot tell the difference. The difference is the 200 MiB that Step 5 measured being received, spooled to disk and read, and no status-code assertion can see that. Resource behavior needs a measurement, like step_limit.py, and not only a test.

Common mistakes and gotchas

Trusting the label

The file name and the Content-Type are both written by the sender. Decide from the bytes, and decide against an allowlist of the few formats your product needs.

Serving the upload folder as static files

The first version’s StaticFiles mount gave every uploaded file a type chosen from its extension. A folder that web clients can read directly should hold only files you built.

Checking the size after reading

The check in app_v2.py is correct and too late. Put the ceiling in front of the parser, as Step 5 does.

Decoding on the event loop

Image work is CPU work. Step 7 measured two pings getting through while a 24-megapixel image was cleaned on the loop, and 70 when it ran in a thread. Also cap how many cleanings can run at once, because each one can hold hundreds of megabytes.

Stripping metadata without applying the rotation

If exif_transpose is missing, your privacy fix turns every portrait photo sideways, and someone will file a bug that blames the upload code.

Testing on one operating system

The trailing-dot bypass and the alternate data streams in Step 4 are Windows behavior, and the absolute-path join is Linux and macOS behavior. Run the matrix and the tests on the platform you deploy to, and if you develop on another one, run them there as well.

Leaving the warning as a warning

Pillow’s decompression bomb guard warns first and fails only beyond twice its limit. Do your own size comparison from the header, as clean_image does.

Treating the result as proof

Re-encoding is a strong measure and not a guarantee. OWASP says so itself, as quoted in Step 7. Keep Pillow current, accept few formats, and consider running the decoder in a low-privilege process.

Check the whole thing end to end, then keep going

You are done when these commands print the results shown in this article: python matrix.py v3 ends with 0 of 11 attacks got through; python -m pytest -q --tb=no -rf --app=app_v3 test_upload_policy.py reports 14 passed, 1 warning; and python -m pytest -q --tb=no -rf test_uploadsafe.py reports 12 passed. Then run the matrix against v1 once more to confirm that the tool can fail, which is how you know a pass means something.

To keep learning, try these in order:

1. Decide who may upload and who may read a file. This service has no accounts, so anyone can upload and anyone with an identifier can read. How to Prevent Broken Object Level Authorization (IDOR) in a FastAPI App shows how to enforce that per object, and How to Stop CSRF Attacks in a Python Web App With Synchronizer Tokens covers OWASP’s reminder to protect the upload itself from CSRF.
2. Move storage off the application server. OWASP ranks the options: a different host first, then outside the web root. An object store with signed upload URLs removes the large bodies from your process altogether, and a separate domain for user content keeps uploaded files out of your application’s origin.
3. Add quotas and limits on concurrent rebuilds, and answer overload honestly with the techniques in the backpressure tutorial linked in Step 7.
4. Add a way to report abuse. OWASP notes that “The File Upload service should allow users to report illegal content, and copyright owners to report abuse.”
5. Extend the allowlist carefully. Archives need the member-name checks from the path traversal tutorial linked in Step 4, and XML-based formats such as SVG or Office documents bring the parser risks in How to Parse Untrusted XML Safely in Python.
6. If you also want to catch repeated or near-identical pictures, How to Detect Duplicate and Near-Duplicate Images With Perceptual Hashing in Python builds that on top of Pillow.
7. For documents, add an antivirus scan and content disarm and reconstruction, which OWASP lists next to the controls built here. They are outside this lab.

Sources and method

• OWASP File Upload Cheat Sheet
• CWE-434: Unrestricted Upload of File with Dangerous Type
• Starlette documentation, Requests (request files, form limits, max_body_size) and starlette/formparsers.py at 1.9.0
• python-multipart 0.0.32, multipart.py
• FastAPI documentation, Request Files
• Pillow documentation, Image module (open, MAX_IMAGE_PIXELS, decompression bomb behavior) and ImageOps (exif_transpose)
• MDN, X-Content-Type-Options, Content-Disposition and Content-Security-Policy: sandbox
• Microsoft Learn, File path formats on Windows systems

Method. Everything was run on Windows 11 with Python 3.13.14, FastAPI 0.146.0, Starlette 1.9.0, uvicorn 0.54.0, python-multipart 0.0.32, Pillow 13.0.0, pytest 9.1.1 and Edge 155 in headless mode. The listings in this article were rebuilt from the page into a clean virtual environment and run again. The browser results come from one browser. Linux and macOS behavior is described from documentation and from the pathlib line in Step 4, not measured. The lab does not test a reverse proxy, an antivirus scanner, object storage or authentication, and the memory and timing figures are from one machine, so expect different absolute numbers and the same pattern.