How to Stop Decompression Bombs in Python With Hard Output Limits for gzip, Brotli, zstd, ZIP, and tar.gz
A hands-on tutorial that measures how far six compression formats can stretch a tiny payload, builds and tests a bounded decompressor, and applies it to a FastAPI gzip endpoint, an HTTP client and...
A decompression bomb is a small compressed file that unpacks into something enormous. The trick is old and it keeps working, because compression exists to make repetitive data tiny and an attacker can make data as repetitive as they like. If your Python service accepts a gzip request body, downloads a page from a server it does not control, or extracts a ZIP that a user uploaded, then someone else decides how much memory and disk space your code will spend on that input, unless you decide first.
Table Of Content
- What you will build, and the words you need first
- Prerequisites
- Step 1: Create the folder and the codec kit
- Step 2: Measure how far each format can stretch a payload
- Step 3: Watch ordinary code fall over
- Step 4: Build a decompressor with a hard cap
- The idea
- Run it on the bombs
- Step 5: Seven traps that look like limits
- Step 6: Door one, request bodies in FastAPI
- Step 7: Door two, responses from servers you do not control
- A server that answers with test payloads
- Two ways to download, and the naive one
- Measure all three
- The same test on an older urllib3
- Step 8: Door three, ZIP and tar.gz uploads
- Builders for test archives
- The limits
- Naive against limited
- Declared sizes can be checked before you decompress anything
- Declared sizes can also lie
- Honest sizes can still add up
- Step 9: Prove it with tests
- Check that the tests can fail
- Step 10: Choose limits for your own service
- Confirm everything works end to end
- Where to go next
- Sources and further reading
In this tutorial you will measure how far six compression formats can stretch a tiny payload, watch ordinary decompression code spend hundreds of megabytes on a payload of a few hundred kilobytes at most, and then build a decompressor that refuses to produce more output than you allow. You will put it in the three places where Python services meet untrusted compressed data: a FastAPI endpoint that accepts Content-Encoding: gzip, an HTTP client that fetches from servers you do not control, and an upload endpoint that extracts ZIP and tar.gz files. Every fix comes with a memory measurement, and the last step turns the work into tests, including a check that the tests catch nine bugs planted on purpose.
This is not a theoretical weakness. MITRE lists it as CWE-409, “Improper Handling of Highly Compressed Data (Data Amplification)”, and describes it as a product that “does not handle or incorrectly handles a compressed input with a very high compression ratio that produces a large output.” urllib3, the library under requests and a great deal of other Python HTTP code, has published four advisories about this bug class since December 2025: CVE-2025-66418 (an unbounded chain of codings), CVE-2025-66471 (the streaming API decoding too much at once), CVE-2026-21441 (redirect responses) and CVE-2026-44432 (parts of the streaming API, including a second read on a Brotli response). They were fixed in urllib3 2.6.0, 2.6.0, 2.6.3 and 2.7.0. If the HTTP library most Python programs depend on needed four rounds of fixes, your own decompression code deserves a deliberate look. In Step 7 you will see the effect of one of those bugs on an older urllib3, and why counting bytes yourself is the safer habit.
One safety note. The payloads in this lab are runs of zero bytes, the most compressible input there is. Nothing here targets anyone, and the largest expansion the lab performs is 256 MiB on your own machine. Run it only on a computer you own, with about 2 GiB of free memory and 1 GB of free disk space.
What you will build, and the words you need first
By the end you will have these pieces:
- a table of how far each of six formats can expand a small payload (Step 2);
- a measuring tool that reports a task’s peak memory in a fresh process (Step 3);
decompress_limited, a function of about 30 lines that decodes gzip, deflate, bzip2, xz, Zstandard and Brotli with a hard cap on output size and on time, anddecode_http, which applies it to aContent-Encodingheader (Step 4);- a FastAPI route that applies the cap to request bodies (Step 6), a download function for
requests(Step 7), and ZIP and tar.gz extractors with member, size and total limits (Step 8); - 59 tests and a script that breaks the decompressor nine ways to prove the tests notice (Step 9).
Some terms, defined before they are used. The compression ratio is the decompressed size divided by the compressed size, so a ratio of 1,000 to 1 means that 1 KB of input produces 1 MB of output. The wire size is the number of compressed bytes that arrive from the network or the file. The output cap is the largest number of decompressed bytes you are willing to produce for one input. Peak memory is the highest amount of memory a process has used so far. It only ever goes up, which is why each measurement below runs in a brand new process.
Prerequisites
- Python 3.13 or newer. Every command ran on Python 3.13.14 on Windows 11. The tar extraction filter used in Step 8 needs Python 3.12 or newer, and the zstd module is built in from Python 3.14 (earlier versions use the
backports.zstdpackage, installed below). - A bash shell (Git Bash on Windows). Linux and macOS should behave the same way, but I ran the lab only on Windows 11, so the Linux and macOS branch of the memory reading in
probe.py(Step 3) was not run. - Packages: FastAPI 0.162.0, uvicorn 0.54.0, requests 2.34.2 (which brings urllib3 2.8.0), brotli 1.2.1, backports.zstd 1.8.0 (Python before 3.14 only), psutil 7.2.2 and pytest 9.1.1. psutil reads memory on Windows and stops the servers the lab starts.
- Resources: about 2 GiB of free memory and 1 GB of free disk space.
- Background: you can read a Python function and a class. No compression theory is needed; each idea is explained where it appears.
Step 1: Create the folder and the codec kit
Make a folder with a virtual environment (a private set of installed packages for this project):
mkdir bomb-lab
cd bomb-lab
python -m venv .venv
source .venv/bin/activate # Git Bash on Windows: source .venv/Scripts/activate
pip install fastapi==0.162.0 uvicorn==0.54.0 requests==2.34.2 brotli==1.2.1 psutil==7.2.2 pytest==9.1.1
pip install "backports.zstd==1.8.0; python_version < '3.14'"The second pip line installs the Zstandard backport only on Python 3.10 to 3.13. All files below go in this one folder, and the first line of every code block is a comment with the file name. A block whose first line says “(continued)” belongs at the end of the file started in the block above it.
Start with codecs_kit.py. It puts the six formats behind one interface so that every later step can loop over them:
# codecs_kit.py
"""Compress and decompress with the six codecs used on the web and in uploads."""
import bz2
import gzip
import json
import lzma
import random
import zlib
import brotli
try: # Python 3.14 and newer
from compression import zstd
except ImportError: # Python 3.10 to 3.13: pip install backports.zstd
from backports import zstd
MiB = 1024 * 1024
CODECS = ("gzip", "deflate", "bz2", "xz", "zstd", "br")
_COMPRESS = {
"gzip": lambda data: gzip.compress(data, 9),
"deflate": lambda data: zlib.compress(data, 9), # HTTP "deflate" is the zlib format
"bz2": bz2.compress,
"xz": lzma.compress,
"zstd": zstd.compress,
"br": brotli.compress,
}
_NAIVE = {
"gzip": gzip.decompress,
"deflate": zlib.decompress,
"bz2": bz2.decompress,
"xz": lzma.decompress,
"zstd": zstd.decompress,
"br": brotli.decompress,
}
def compress(codec, data):
return _COMPRESS[codec](data)
def decompress_naive(codec, data):
"""The one-liner most code uses. It has no limit on the output."""
return _NAIVE[codec](data)
def make_test_payload(codec, size):
"""Compress `size` zero bytes: harmless, and as compressible as data gets."""
return compress(codec, bytes(size))
def sample_document(records=20000):
"""A realistic JSON document of about 1.5 MiB, the same every run."""
rng = random.Random(7)
notes = ["ok", "backorder", "damaged box", "gift", ""]
rows = [{"id": i, "sku": f"SKU-{rng.randrange(10**6):06d}", "qty": rng.randrange(1, 500),
"note": rng.choice(notes)} for i in range(records)]
return json.dumps(rows).encode()Four things are worth noticing. compress and decompress_naive are the one-liners most code uses: they hand the whole input to the library and return the whole output, and neither has any limit on how large that output can be. HTTP’s deflate coding is the zlib format (it differs from gzip only in its header and trailer), so it gets its own entry. make_test_payload compresses a run of zero bytes, which is as compressible as data gets, so it shows the worst case for each format. And sample_document builds an ordinary JSON document of about 1.3 MB that later steps use to prove that the limits do not break legitimate input.
Step 2: Measure how far each format can stretch a payload
Before defending against a stretch you should know how big it can be. step01_ratios.py compresses 256 MiB of zeros with each format, saves the results in a payloads folder for later steps, and prints the ratio and what a 1 MiB upload would become at that ratio:
# step01_ratios.py
"""How far can each codec stretch a tiny upload? Compress 256 MiB of zeros and see."""
import os
from codecs_kit import CODECS, MiB, make_test_payload
SIZE = 256 * MiB
os.makedirs("payloads", exist_ok=True)
print(f"{'codec':<8}{'compressed':>14}{'ratio':>16} a 1 MiB upload could expand to")
for codec in CODECS:
data = make_test_payload(codec, SIZE)
with open(f"payloads/{codec}.bin", "wb") as f:
f.write(data)
ratio = SIZE / len(data)
print(f"{codec:<8}{len(data):>12,} B{ratio:>14,.0f}:1 {ratio / 1024:>8,.1f} GiB")Run python step01_ratios.py:
codec compressed ratio a 1 MiB upload could expand to
gzip 260,934 B 1,029:1 1.0 GiB
deflate 260,922 B 1,029:1 1.0 GiB
bz2 208 B 1,290,555:1 1,260.3 GiB
xz 39,184 B 6,851:1 6.7 GiB
zstd 8,211 B 32,692:1 31.9 GiB
br 421 B 637,614:1 622.7 GiBRead the table from the bottom up. Brotli turned 256 MiB into 421 bytes, and bzip2 into 208 bytes, a ratio of about 1.3 million to 1. A 1 MiB upload in bzip2 format, which many upload limits would accept without a second look, could expand to about 1,260 GiB if the ratio held. gzip and deflate are the tamest at 1,029 to 1, but that still means a quarter of a megabyte of request body is a quarter of a gigabyte of data.
Two lessons follow already. First, a limit on the compressed size is not a limit on the work: at these ratios the same 1 MiB cap lets through anything from 1 GiB to over a terabyte of output, depending on the format. Second, the OWASP File Upload Cheat Sheet says the check has to sit on the far side of the decompression: “If the system is going to extract the files or process them, the file size limit should be considered after file decompression is conducted and by using secure methods to calculate zip files size.” The rest of this tutorial is about doing that safely.
Step 3: Watch ordinary code fall over
Now decompress those payloads the way most code does, in a fresh process each time, and record the peak memory. Create probe.py. It has two jobs: a function that reads the process’s peak memory on Windows, Linux and macOS, and a runner that executes one decompression in a child process and reports back as a line of JSON:
# probe.py
"""Run one decompression in a fresh process and report that process's peak memory."""
import json
import subprocess
import sys
import time
MiB = 1024 * 1024
def peak_mib():
"""Highest memory this process has used so far, in MiB (Windows, Linux and macOS)."""
if sys.platform == "win32":
import psutil
return psutil.Process().memory_info().peak_wset / MiB
import resource
peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss # KiB on Linux, bytes on macOS
return peak / (MiB if sys.platform == "darwin" else 1024)
def run_probe(mode, codec, payload_path, limit=0):
"""Start `python probe.py ...` and parse the one JSON line it prints."""
cmd = [sys.executable, __file__, mode, codec, payload_path, str(limit)]
done = subprocess.run(cmd, capture_output=True, text=True, timeout=300, check=True)
return json.loads(done.stdout.strip().splitlines()[-1])
if __name__ == "__main__":
mode, codec, path, limit = sys.argv[1:5]
from codecs_kit import decompress_naive
from safe_decompress import Refused, decompress_limited
with open(path, "rb") as f:
data = f.read()
base, started = peak_mib(), time.perf_counter()
try:
if mode == "naive":
out = decompress_naive(codec, data)
else:
out = decompress_limited(codec, data, int(limit))
result = f"ok, {len(out) // MiB} MiB"
except Refused as exc:
result = f"refused: {exc.reason}"
print(json.dumps({"result": result, "peak_mib": round(peak_mib() - base),
"seconds": round(time.perf_counter() - started, 2)}))On Windows the number comes from psutil’s peak_wset, the peak working set, which is the most physical memory the process has held at once. On Linux and macOS it comes from resource.getrusage, the same idea under another name (the code comment records that Linux reports kilobytes and macOS bytes). The child records a baseline right before it starts and reports the growth above it, so interpreter start-up and library imports do not count. Create step02_naive.py:
# step02_naive.py
"""Decompress each payload the way most code does and watch the memory."""
from codecs_kit import CODECS
from probe import run_probe
print(f"{'codec':<8}{'result':<14}{'peak memory':>14}{'seconds':>10}")
for codec in CODECS:
r = run_probe("naive", codec, f"payloads/{codec}.bin")
print(f"{codec:<8}{r['result']:<14}{r['peak_mib']:>10,} MiB{r['seconds']:>10}")Run python step02_naive.py:
codec result peak memory seconds
gzip ok, 256 MiB 512 MiB 0.28
deflate ok, 256 MiB 512 MiB 0.27
bz2 ok, 256 MiB 516 MiB 0.49
xz ok, 256 MiB 520 MiB 0.6
zstd ok, 256 MiB 514 MiB 0.13
br ok, 256 MiB 516 MiB 0.25Every format produced exactly the 256 MiB it was asked to and cost about 512 to 520 MiB of memory to do it, roughly twice the output size. I did not dig into why each library needs that much; the point is that memory grows with the output, and the output is chosen by whoever made the file. None of these six calls would fit in a container limited to 512 MiB. They also took under a second each, so asking costs an attacker almost nothing.
Step 4: Build a decompressor with a hard cap
The idea
Every one of these libraries can decompress a stream incrementally: you feed in a little compressed data and ask for at most N bytes of output. If you ask for N bytes at a time and add up what you receive, you can stop the moment the total passes your cap, with no more than one small step of excess in memory. The three families spell this differently:
- zlib (gzip and deflate) uses
zlib.decompressobj()anddecompress(data, max_length). Input the object did not get to is kept inunconsumed_tailand has to be fed back in. The documentation says: “If the optional parameter max_length is non-zero then the return value will be no longer than max_length.” Note the words non-zero; Step 5 returns to them. - bzip2, xz and Zstandard share one interface:
decompress(data, max_length)plus the flagsneeds_input(does the decoder want more compressed bytes?),eof(has the stream ended?) andunused_data(bytes found after the end). - Brotli uses
Decompressor.process(data, output_buffer_limit=N)pluscan_accept_more_data()andis_finished().
The code below hides those three dialects behind small adapter classes with the same four members, feed, eof, wants_input and unused, so that one loop can drive all of them. Create safe_decompress.py, starting with the imports, two constants, the table of HTTP coding names and the exception:
# safe_decompress.py
"""Decompress untrusted bytes with hard limits on output size and time."""
import bz2
import lzma
import time
import zlib
import brotli
from codecs_kit import MiB, zstd
CHUNK = 64 * 1024 # compressed bytes fed to the decoder per step
STEP = 1 * MiB # most output we ask for per step
HTTP_CODINGS = {"gzip": "gzip", "x-gzip": "gzip", "deflate": "deflate", "br": "br", "zstd": "zstd"}
class Refused(Exception):
"""The payload broke a limit. `reason` is a short code to log or map to a status."""
def __init__(self, reason, detail=""):
super().__init__(f"{reason}: {detail}" if detail else reason)
self.reason = reasonRefused carries a short reason code, such as too_large or truncated, so callers can log it or map it to an HTTP status without parsing text. CHUNK is how many compressed bytes the loop feeds per step, and STEP is the most output it asks for per step. Next, the three adapters:
# safe_decompress.py (continued)
class _Zlib: # gzip and deflate
def __init__(self, codec):
wbits = zlib.MAX_WBITS | 16 if codec == "gzip" else zlib.MAX_WBITS
self.d, self.tail = zlib.decompressobj(wbits), b""
eof = property(lambda self: self.d.eof)
wants_input = property(lambda self: not self.tail)
unused = property(lambda self: self.d.unused_data)
def feed(self, data, room): # room is at least 1: max_length=0 would mean "no limit"
out = self.d.decompress(self.tail + data, room)
self.tail = self.d.unconsumed_tail
return out
class _MaxLength: # bz2, xz and zstd share one interface
def __init__(self, decoder):
self.d = decoder
eof = property(lambda self: self.d.eof)
wants_input = property(lambda self: self.d.needs_input)
unused = property(lambda self: self.d.unused_data)
def feed(self, data, room):
return self.d.decompress(data, max_length=room)
class _Brotli:
def __init__(self):
self.d = brotli.Decompressor()
eof = property(lambda self: self.d.is_finished())
wants_input = property(lambda self: self.d.can_accept_more_data())
unused = b"" # Brotli raises an error on trailing bytes instead
def feed(self, data, room):
return self.d.process(data, output_buffer_limit=room)The zlib adapter has to cope with unconsumed_tail: when a call stops because of max_length, the rest of the input waits there and the next call must start with it, so its wants_input means “nothing is left over”. The bzip2, xz and Zstandard adapter is thin because their interfaces already match. The Brotli adapter leaves unused empty because, when I appended bytes to a valid Brotli stream, the decoder raised brotli.error instead of reporting them. For that reason the trailing-bytes test in Step 9 accepts either trailing_data or bad_data. Now the function that picks an adapter, with the two memory settings that Step 5 explains:
# safe_decompress.py (continued)
_ERRORS = (OSError, EOFError, ValueError, zlib.error, lzma.LZMAError, brotli.error, zstd.ZstdError)
def _new_decoder(codec, memlimit, window_log_max):
if codec in ("gzip", "deflate"):
return _Zlib(codec)
if codec == "bz2":
return _MaxLength(bz2.BZ2Decompressor())
if codec == "xz":
return _MaxLength(lzma.LZMADecompressor(memlimit=memlimit))
if codec == "zstd":
options = {zstd.DecompressionParameter.window_log_max: window_log_max}
return _MaxLength(zstd.ZstdDecompressor(options=options))
if codec == "br":
return _Brotli()
raise Refused("unsupported_encoding", codec)And the loop itself, the heart of this tutorial:
# safe_decompress.py (continued)
def decompress_limited(codec, data, max_output, *, max_seconds=None,
memlimit=128 * MiB, window_log_max=27):
"""Return the decompressed bytes or raise Refused. Memory stays near max_output."""
dec = _new_decoder(codec, memlimit, window_log_max)
deadline = None if max_seconds is None else time.monotonic() + max_seconds
out, pos = bytearray(), 0
while not dec.eof:
if dec.wants_input and pos < len(data):
chunk, pos = data[pos:pos + CHUNK], pos + CHUNK
else:
chunk = b""
room = min(max_output - len(out) + 1, STEP) # one byte more than allowed: overflow shows
try:
piece = dec.feed(chunk, room)
except _ERRORS as exc:
raise Refused("bad_data", type(exc).__name__) from None
out += piece
if len(out) > max_output:
raise Refused("too_large", f"more than {max_output} bytes")
if deadline is not None and time.monotonic() > deadline:
raise Refused("too_slow", f"more than {max_seconds} s")
if not piece and not chunk and not dec.eof:
raise Refused("truncated", "the stream ends before its end marker")
if dec.unused or pos < len(data):
raise Refused("trailing_data", "bytes follow the end of the stream")
return bytes(out)Read the loop slowly, because every line is there for a reason:
while not dec.eofruns until the decoder reports the end of the stream.- Each pass feeds the next
CHUNKof compressed bytes if the decoder wants input and any remain. Otherwise it feeds an empty bytes object, which means “give me more of the output you are already holding”. roomis the cap minus what we hold, plus one byte, limited toSTEP(1 MiB). The extra byte is what makes overflow detectable: if the stream holds exactlymax_outputbytes we get them all and finish, and if it holds one more, that byte arrives, the total passes the cap, and we refuse. Without the extra byte, an output of exactly the cap and an output of the cap plus a million would look the same until the next call.- Any library error becomes
Refused("bad_data"), with only the exception’s class name as detail and never the payload. - After adding the piece to the output we check the cap, then the clock. The deadline matters because a payload can be slow without being large.
- If a step fed no new input and produced no output, and the stream has not ended, nothing more can ever happen: the data was cut off. We refuse it as
truncatedinstead of looping forever. - After the loop, any leftover bytes mean the input held more than one stream, which we refuse as
trailing_data.
The last function applies the same cap to HTTP’s Content-Encoding header, which can list several codings in the order they were applied:
# safe_decompress.py (continued)
def decode_http(body, content_encoding, max_output, *, max_layers=1, max_seconds=None):
"""Undo a Content-Encoding header. Codings are listed in the order they were applied."""
codings = [c.strip().lower() for c in content_encoding.split(",") if c.strip()]
codings = [c for c in codings if c != "identity"]
if len(codings) > max_layers:
raise Refused("too_many_layers", f"{len(codings)} codings, limit {max_layers}")
for coding in reversed(codings):
if coding not in HTTP_CODINGS:
raise Refused("unsupported_encoding", coding)
body = decompress_limited(HTTP_CODINGS[coding], body, max_output,
max_seconds=max_seconds, window_log_max=23)
if len(body) > max_output:
raise Refused("too_large", f"more than {max_output} bytes")
return bodyThree details to notice. Codings are undone in reverse order (a header of gzip, br means gzip was applied first, so Brotli comes off first). The number of codings is capped at one by default, because a legitimate client almost never stacks them and Step 5 shows why stacking is dangerous. And a body with no coding at all still goes through the size check on the last lines, so plain bodies and compressed bodies obey the same cap. The Zstandard window is limited to 2 to the power of 23 bytes (8 MiB) for HTTP bodies; Step 5 explains where that number comes from.
Run it on the bombs
Now use it. step03_bounded.py runs the same 256 MiB payloads through decompress_limited with an 8 MiB cap, in fresh processes again, and then checks that a legitimate document survives:
# step03_bounded.py
"""The same payloads through decompress_limited with an 8 MiB cap, then a legitimate document."""
from codecs_kit import CODECS, MiB, compress, sample_document
from probe import run_probe
from safe_decompress import decompress_limited
print(f"{'codec':<8}{'result':<20}{'peak memory':>14}{'seconds':>10}")
for codec in CODECS:
r = run_probe("bounded", codec, f"payloads/{codec}.bin", 8 * MiB)
print(f"{codec:<8}{r['result']:<20}{r['peak_mib']:>10,} MiB{r['seconds']:>10}")
doc = sample_document()
print(f"\nlegitimate JSON document: {len(doc):,} bytes")
for codec in CODECS:
packed = compress(codec, doc)
same = decompress_limited(codec, packed, 8 * MiB) == doc
print(f"{codec:<8}{len(packed):>10,} B round trip identical: {same}")Run python step03_bounded.py:
codec result peak memory seconds
gzip refused: too_large 16 MiB 0.02
deflate refused: too_large 16 MiB 0.02
bz2 refused: too_large 20 MiB 0.03
xz refused: too_large 24 MiB 0.03
zstd refused: too_large 18 MiB 0.02
br refused: too_large 20 MiB 0.03
legitimate JSON document: 1,288,469 bytes
gzip 199,422 B round trip identical: True
deflate 199,410 B round trip identical: True
bz2 145,064 B round trip identical: True
xz 145,208 B round trip identical: True
zstd 214,273 B round trip identical: True
br 150,246 B round trip identical: TrueAll six payloads are refused after 0.02 to 0.03 seconds, with a peak of 16 to 24 MiB instead of about 512 MiB. The memory that remains is the output buffer, which holds up to the 8 MiB cap and grows by copying, plus whatever state each decoder keeps. What matters is that the peak is set by your cap and not by the file, and that the legitimate 1.3 MB document came back byte for byte identical through all six formats.
Step 5: Seven traps that look like limits
The function above avoids several mistakes that are easy to make and hard to notice. step04_gotchas.py demonstrates each one. The script is long, so it appears in three parts, each followed by its output. Run python step04_gotchas.py to see all of it at once. The first part covers three traps in the libraries’ own parameters:
# step04_gotchas.py
"""Traps that look like limits but are not, each one measured."""
import gzip
import lzma
import zlib
import brotli
from codecs_kit import MiB, sample_document, zstd
from safe_decompress import Refused, decode_http, decompress_limited
bomb = zlib.compress(bytes(8 * MiB))
print("1. bufsize is a starting buffer size, not a cap")
print(f" zlib.decompress(bomb, bufsize=1024) returned {len(zlib.decompress(bomb, bufsize=1024)):,} bytes")
print("2. max_length=0 means no limit in zlib")
print(f" decompress(bomb, 0) returned {len(zlib.decompressobj().decompress(bomb, 0)):,} bytes,"
f" decompress(bomb, 1) returned {len(zlib.decompressobj().decompress(bomb, 1))} byte")
print("3. Brotli's output_buffer_limit is a soft limit")
packed = brotli.compress(bytes(8 * MiB))
for limit in (1000, 65536, MiB):
first = brotli.Decompressor().process(packed, output_buffer_limit=limit)
print(f" output_buffer_limit={limit:>9,} gave a first piece of {len(first):>9,} bytes")1. bufsize is a starting buffer size, not a cap
zlib.decompress(bomb, bufsize=1024) returned 8,388,608 bytes
2. max_length=0 means no limit in zlib
decompress(bomb, 0) returned 8,388,608 bytes, decompress(bomb, 1) returned 1 byte
3. Brotli's output_buffer_limit is a soft limit
output_buffer_limit= 1,000 gave a first piece of 32,752 bytes
output_buffer_limit= 65,536 gave a first piece of 98,272 bytes
output_buffer_limit=1,048,576 gave a first piece of 2,064,288 bytesTrap 1: bufsize is not a limit. zlib.decompress(data, bufsize=...) looks like it should cap the output, but the documentation says: “bufsize is the initial size of the buffer used to hold decompressed data. If more space is required, the buffer size will be increased as needed, so you don’t have to get this value exactly right; tuning it will only save a few calls to malloc().” A bufsize of 1,024 still returned all 8 MiB.
Trap 2: in zlib, a max_length of zero means no limit. That is what “non-zero” in the documentation sentence quoted in Step 4 was telling you. Passing 0 returned all 8 MiB, and passing 1 returned one byte. This is why room is always at least 1, and why the tests include a zero-byte cap for every format. A cap computed as “allowed minus already produced” reaches zero exactly when you are at the limit, which is the worst moment to lose it.
Trap 3: Brotli’s limit is soft. The package’s own documentation for output_buffer_limit says: “If set, the output buffer will not grow once its size equal or exceeding that value.” That describes a buffer that stops growing, not a promise about the size of what you receive. With a limit of 1,000 bytes the first piece was 32,752 bytes, and with 1 MiB it was 2,064,288 bytes, nearly double. The loop handles this in two ways: it clamps every request to STEP so that the overshoot stays near 2 MiB whatever the cap, and it always measures the real length of what it received instead of trusting the limit it asked for.
The second part covers two traps in how streams are put together:
# step04_gotchas.py (continued)
print("4. gzip.decompress accepts two members glued together")
two = gzip.compress(b"first part;") + gzip.compress(b"second part")
print(" gzip.decompress ->", gzip.decompress(two))
try:
decompress_limited("gzip", two, MiB)
except Refused as exc:
print(" decompress_limited ->", exc)
print("5. Content-Encoding can stack codings")
layer = bytes(64 * MiB)
for n in range(1, 4):
layer = gzip.compress(layer, 9, mtime=0)
print(f" after {n} gzip layer(s): {len(layer):,} bytes")
if n == 2:
two_layers = layer
print(f" 64 MiB of zeros in {len(two_layers):,} bytes is {64 * MiB / len(two_layers):,.0f}:1")
try:
decode_http(two_layers, "gzip, gzip", 64 * MiB)
except Refused as exc:
print(" decode_http, default max_layers=1 ->", exc)
out = decode_http(two_layers, "gzip, gzip", 64 * MiB, max_layers=2)
print(f" decode_http with max_layers=2 -> {len(out):,} bytes")4. gzip.decompress accepts two members glued together
gzip.decompress -> b'first part;second part'
decompress_limited -> trailing_data: bytes follow the end of the stream
5. Content-Encoding can stack codings
after 1 gzip layer(s): 65,250 bytes
after 2 gzip layer(s): 269 bytes
after 3 gzip layer(s): 247 bytes
64 MiB of zeros in 269 bytes is 249,475:1
decode_http, default max_layers=1 -> too_many_layers: 2 codings, limit 1
decode_http with max_layers=2 -> 67,108,864 bytesTrap 4: gzip.decompress accepts several members glued together. A gzip file may hold more than one stream back to back, and gzip.decompress joins their outputs, as the first line of output shows. Our function refuses the second member as trailing_data on purpose, because an endpoint that expects one document should not quietly accept two. If you need concatenated members (some tools produce them), add that as a deliberate feature with its own count limit.
Trap 5: codings can be stacked. The script gzips 64 MiB of zeros once and gets 65,250 bytes, then gzips that result again and gets only 269 bytes. That is 249,475 to 1, far past the 1,029 to 1 ceiling of a single gzip layer, because the output of the first layer is itself very repetitive. HTTP allows Content-Encoding: gzip, gzip, and the urllib3 advisory for CVE-2025-66418 describes a malicious server inserting “a virtually unlimited number of compression steps”; urllib3 2.6.0 limits the chain to 5 links. decode_http allows one by default and caps each layer’s output separately, so the total work is bounded by the number of layers times the cap.
The third part covers the last two traps, one about ratios and one about memory:
# step04_gotchas.py (continued)
print("6. Legitimate data has no typical ratio")
csv = "".join(f"{i},,,,,,,,,,,,,,,,,\n" for i in range(100000)).encode()
for name, data in (("JSON records", sample_document()), ("CSV with empty columns", csv),
("sparse disk image", b"header" + bytes(32 * MiB))):
print(f" {name:<24}{len(data):>12,} B gzip ratio {len(data) / len(gzip.compress(data)):>7,.0f}:1")
print("7. Decoder memory can be asked for in the header, not only the output")
c = zstd.ZstdCompressor(options={zstd.CompressionParameter.window_log: 27})
frame = c.compress(b"hello world" * 10) + c.flush(zstd.ZstdCompressor.FLUSH_FRAME)
for window_log in (27, 23):
options = {zstd.DecompressionParameter.window_log_max: window_log}
try:
out = zstd.ZstdDecompressor(options=options).decompress(frame)
print(f" zstd, 128 MiB window frame, window_log_max={window_log}: decoded {len(out)} bytes")
except zstd.ZstdError as exc:
print(f" zstd, 128 MiB window frame, window_log_max={window_log}: ZstdError: {exc}")
xz = lzma.compress(b"hello", format=lzma.FORMAT_XZ,
filters=[{"id": lzma.FILTER_LZMA2, "dict_size": 64 * MiB}])
for memlimit in (128 * MiB, 64 * MiB, 16 * MiB):
try:
out = lzma.LZMADecompressor(memlimit=memlimit).decompress(xz)
print(f" xz, 64 MiB dictionary, memlimit={memlimit // MiB} MiB: decoded {out!r}")
except lzma.LZMAError as exc:
print(f" xz, 64 MiB dictionary, memlimit={memlimit // MiB} MiB: LZMAError: {exc}")6. Legitimate data has no typical ratio
JSON records 1,288,469 B gzip ratio 6:1
CSV with empty columns 2,288,890 B gzip ratio 9:1
sparse disk image 33,554,438 B gzip ratio 1,028:1
7. Decoder memory can be asked for in the header, not only the output
zstd, 128 MiB window frame, window_log_max=27: decoded 110 bytes
zstd, 128 MiB window frame, window_log_max=23: ZstdError: Unable to decompress Zstandard data: Frame requires too much memory for decoding
xz, 64 MiB dictionary, memlimit=128 MiB: decoded b'hello'
xz, 64 MiB dictionary, memlimit=64 MiB: LZMAError: Memory usage limit exceeded
xz, 64 MiB dictionary, memlimit=16 MiB: LZMAError: Memory usage limit exceededTrap 6: there is no typical ratio. It is tempting to refuse anything above, say, 100 to 1. The script measured three legitimate files. JSON records compress 6 to 1 and a CSV with many empty columns 9 to 1, but a sparse disk image, which is mostly zeros, compresses 1,028 to 1, which an attack cannot be told apart from by ratio alone. A ratio threshold either rejects honest sparse files or lets real bombs through. An absolute cap, chosen from what your endpoint legitimately needs, has neither problem.
Trap 7: a stream can ask for memory in its header, not only in its output. Two formats let the sender choose how much memory the decoder must use. A Zstandard frame declares a window size, which is how far back the decoder must remember. The Python documentation for window_log_max says it “can be useful to limit the amount of memory used when decompressing data”. RFC 9659 “updates the window size limit in RFC 8878 from a recommendation to a requirement in HTTP contexts”: “decoders MUST support a Window_Size of up to and including 8 MB, and encoders MUST NOT generate frames requiring a Window_Size larger than 8 MB”. It also warns that decoders “still need to take into account that they can receive oversized frames”. That is why decode_http passes window_log_max=23, which is 8 MiB. In the script, a tiny frame compressed with a 128 MiB window setting decodes with the limit at 27 and fails with “Frame requires too much memory for decoding” at 23. For xz the equivalent is memlimit: “decompression will fail with an LZMAError if it is not possible to decompress the input within the given memory limit.” A stream with a 64 MiB dictionary needed more than a 64 MiB limit and decoded under 128 MiB, so the function’s default is 128 MiB. Lower it if you know your files come from smaller presets.
Step 6: Door one, request bodies in FastAPI
Now apply the function to the first place where untrusted compressed data enters: HTTP request bodies. Clients sometimes gzip their uploads, and FastAPI’s page on a custom Request and APIRoute class lists “Decompressing gzip-compressed request bodies” among its use cases. Its first example does exactly that, and the page calls it “a toy example to demonstrate how it works”. Examples like this get copied into services, so it is a realistic starting point for a test. The file below keeps the page’s GzipRequest and GzipRoute classes as they are and swaps the page’s number-summing endpoint for a notes endpoint that takes a text field. Create app_v1.py:
# app_v1.py
"""Notes API that decompresses gzip request bodies the way the FastAPI docs recipe does."""
import gzip
from collections.abc import Callable
from fastapi import FastAPI, Request, Response
from fastapi.routing import APIRoute
from pydantic import BaseModel
from probe import peak_mib
class GzipRequest(Request):
async def body(self) -> bytes:
if not hasattr(self, "_body"):
body = await super().body()
if "gzip" in self.headers.getlist("Content-Encoding"):
body = gzip.decompress(body)
self._body = body
return self._body
class GzipRoute(APIRoute):
def get_route_handler(self) -> Callable:
original = super().get_route_handler()
async def handler(request: Request) -> Response:
return await original(GzipRequest(request.scope, request.receive))
return handler
class Note(BaseModel):
title: str
text: str
app = FastAPI()
app.router.route_class = GzipRoute
@app.post("/notes")
async def add_note(note: Note):
return {"title": note.title, "chars": len(note.text)}
@app.get("/_peak") # lab only: lets the lab script read this process's peak memory
async def peak():
return {"peak_mib": round(peak_mib())}The line that matters is gzip.decompress(body). The class reads the whole body and then decompresses all of it, with no limit on either step. The /_peak route at the bottom exists only for the lab: it lets a script ask the server process how much memory it has used at its peak. Next, the hardened version. Create app.py:
# app.py
"""Notes API with bounded request decompression. Compare app_v1.py, the docs recipe."""
from collections.abc import Callable
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.routing import APIRoute
from pydantic import BaseModel
from codecs_kit import MiB
from probe import peak_mib
from safe_decompress import Refused, decode_http
MAX_WIRE = 256 * 1024 # compressed bytes we are willing to read from the socket
MAX_BODY = 2 * MiB # bytes allowed after decompression
STATUS = {"wire_too_large": 413, "too_large": 413, "unsupported_encoding": 415} # others: 400
class BoundedRequest(Request):
async def body(self) -> bytes:
if not hasattr(self, "_body"):
raw = bytearray()
try:
async for chunk in self.stream():
raw += chunk
if len(raw) > MAX_WIRE:
raise Refused("wire_too_large", f"more than {MAX_WIRE} bytes")
coding = self.headers.get("content-encoding", "")
self._body = decode_http(bytes(raw), coding, MAX_BODY, max_seconds=1)
except Refused as exc:
raise HTTPException(STATUS.get(exc.reason, 400), detail=exc.reason) from None
return self._body
class BoundedRoute(APIRoute):
def get_route_handler(self) -> Callable:
original = super().get_route_handler()
async def handler(request: Request) -> Response:
return await original(BoundedRequest(request.scope, request.receive))
return handler
class Note(BaseModel):
title: str
text: str
app = FastAPI()
app.router.route_class = BoundedRoute
@app.post("/notes")
async def add_note(note: Note):
return {"title": note.title, "chars": len(note.text)}
@app.get("/_peak") # lab only: lets the lab script read this process's peak memory
async def peak():
return {"peak_mib": round(peak_mib())}It applies two caps at two moments. While reading from the socket, BoundedRequest.body counts compressed bytes and stops at MAX_WIRE (256 KiB), so a client cannot make you buffer a huge compressed body first. After reading, decode_http enforces MAX_BODY (2 MiB) on the decoded bytes, and the same call also covers bodies sent without any Content-Encoding. Each failure becomes an HTTPException with a status from the STATUS table: 413 for too much data, 415 for a coding the service does not accept, and 400 for the rest. It has to be an HTTPException. In FastAPI 0.162.0, routing.py re-raises an HTTPException raised while the body is read, but turns every other exception into a generic 400 with the detail “There was an error parsing the body”, so a Refused would lose its meaning on the way out. For a standard error format, see the RFC 9457 problem details tutorial, and for limiting the compressed body with Starlette’s own middleware, the secure file uploads tutorial.
Now compare the two apps. step05_fastapi.py starts each one in its own process, sends a normal gzip note and then a bomb, and reads each server’s peak memory before and after. The bomb is a JSON note whose text field holds 128 MiB of the letter A, gzip-compressed to 130,512 bytes. The helper class in labkit.py starts a server, waits until it answers, and always stops it:
# labkit.py
"""Start a lab server in its own process, wait until it answers, and always stop it."""
import subprocess
import time
import psutil
import requests
class Server:
def __init__(self, cmd, port, ready_path="/_peak"):
self.cmd, self.url, self.ready_path = cmd, f"http://127.0.0.1:{port}", ready_path
def __enter__(self):
self.proc = subprocess.Popen(self.cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
for _ in range(150):
try:
requests.get(self.url + self.ready_path, timeout=1)
return self
except requests.RequestException:
time.sleep(0.2)
self.__exit__()
raise RuntimeError("the server did not start")
def peak(self):
"""The server's own peak memory in MiB (every lab app has a /_peak route)."""
return requests.get(self.url + "/_peak", timeout=10).json()["peak_mib"]
def __exit__(self, *exc):
try:
parent = psutil.Process(self.proc.pid)
for proc in parent.children(recursive=True) + [parent]:
proc.kill()
except psutil.NoSuchProcess:
pass# step05_fastapi.py
"""Send the same tiny gzip body to the docs-recipe app and to the bounded app."""
import gzip
import json
import os
import sys
import requests
from codecs_kit import MiB
from labkit import Server
def packed(text_size, title="note"):
return gzip.compress(json.dumps({"title": title, "text": "A" * text_size}).encode(), 9)
LEGIT = packed(1000, "hello")
BOMB = packed(128 * MiB, "bomb")
print(f"bomb body: {len(BOMB):,} bytes on the wire, {128 * MiB:,} characters of text inside\n")
def post(url, data, encoding="gzip"):
headers = {"Content-Type": "application/json"}
if encoding:
headers["Content-Encoding"] = encoding
return requests.post(url + "/notes", data=data, headers=headers, timeout=120)
for module, port in (("app_v1", 8201), ("app", 8202)):
cmd = [sys.executable, "-m", "uvicorn", f"{module}:app", "--port", str(port), "--log-level", "warning"]
with Server(cmd, port) as srv:
before = srv.peak()
good = post(srv.url, LEGIT)
bad = post(srv.url, BOMB)
grew = srv.peak() - before
print(f"{module:<7} legit -> {good.status_code} {good.json()}")
print(f"{'':<7} bomb -> {bad.status_code} {bad.text[:50]} server peak memory grew by {grew} MiB")
if module == "app":
cases = {
"3 MiB JSON, no coding": (json.dumps({"title": "t", "text": "y" * 3 * MiB}).encode(), ""),
"unsupported coding 'compress'": (LEGIT, "compress"),
"gzip cut in half": (LEGIT[:len(LEGIT) // 2], "gzip"),
"two gzip layers": (gzip.compress(LEGIT), "gzip, gzip"),
"300 KiB on the wire": (os.urandom(300 * 1024), "gzip"),
}
for name, (data, encoding) in cases.items():
r = post(srv.url, data, encoding)
print(f"{'':<7} {name:<31}-> {r.status_code} {r.json()['detail']}")Run python step05_fastapi.py (it needs ports 8201 and 8202):
bomb body: 130,512 bytes on the wire, 134,217,728 characters of text inside
app_v1 legit -> 200 {'title': 'hello', 'chars': 1000}
bomb -> 200 {"title":"bomb","chars":134217728} server peak memory grew by 384 MiB
app legit -> 200 {'title': 'hello', 'chars': 1000}
bomb -> 413 {"detail":"too_large"} server peak memory grew by 5 MiB
3 MiB JSON, no coding -> 413 wire_too_large
unsupported coding 'compress' -> 415 unsupported_encoding
gzip cut in half -> 400 truncated
two gzip layers -> 400 too_many_layers
300 KiB on the wire -> 413 wire_too_largeThe docs-style app accepted the bomb with a 200 and a correct character count, and its peak memory grew by 384 MiB for a request body of 130,512 bytes. That is three times the size of the text inside, which suggests that several copies of the text were alive at once; I did not trace them. The bounded app answered 413 and grew by 5 MiB. The extra cases at the bottom show the other refusals: a 3 MiB body with no coding trips the wire cap, an unknown coding gets 415, a truncated gzip gets 400, and two stacked gzip layers get 400.
Step 7: Door two, responses from servers you do not control
Your service probably also calls other servers: webhooks, link previews, feed readers, package indexes. There the other side picks the Content-Encoding of the response, and your HTTP library decodes it for you by default.
A server that answers with test payloads
The lab server answers three paths: /ok, a normal gzip JSON document; /zeros-gzip, 128 MiB of zeros gzipped to about 130 KB; and /zeros-br, the same zeros in Brotli. /accept-encoding echoes the header your client sent. Create bomb_server.py:
# bomb_server.py
"""A tiny web server that answers with harmless compressed test payloads (repeated zeros)."""
import gzip
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import brotli
from codecs_kit import MiB, sample_document
SIZE = 128 * MiB
PAGES = {
"/ok": ("gzip", gzip.compress(sample_document())),
"/zeros-gzip": ("gzip", gzip.compress(bytes(SIZE), 9)),
"/zeros-br": ("br", brotli.compress(bytes(SIZE))),
}
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path in ("/ready", "/accept-encoding"):
body = self.headers.get("Accept-Encoding", "(none)").encode()
coding = None
else:
coding, body = PAGES[self.path]
self.send_response(200)
self.send_header("Content-Type", "application/json")
if coding:
self.send_header("Content-Encoding", coding)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
ThreadingHTTPServer(("127.0.0.1", int(sys.argv[1])), Handler).serve_forever()Two ways to download, and the naive one
Create safe_fetch.py with two strategies to put next to the naive requests.get(url).content:
# safe_fetch.py
"""Download a small resource over HTTP with limits on the wire size and the decoded size."""
import requests
from safe_decompress import Refused, decode_http
ACCEPT = "gzip, deflate, br, zstd" # only codings that decode_http understands
def fetch_streamed(url, max_output, chunk_size=64 * 1024):
"""The common fix: stream, count decoded bytes, stop early. It trusts urllib3 to decode in small steps."""
with requests.get(url, stream=True, timeout=(3, 10), allow_redirects=False) as resp:
body = bytearray()
for chunk in resp.iter_content(chunk_size):
body += chunk
if len(body) > max_output:
raise Refused("too_large", f"more than {max_output} bytes")
return bytes(body)
def fetch_limited(url, max_wire, max_output):
"""Read the compressed bytes ourselves and decode them with our own limits."""
headers = {"Accept-Encoding": ACCEPT}
with requests.get(url, stream=True, headers=headers, timeout=(3, 10), allow_redirects=False) as resp:
wire = bytearray()
while chunk := resp.raw.read(64 * 1024, decode_content=False):
wire += chunk
if len(wire) > max_wire:
raise Refused("wire_too_large", f"more than {max_wire} bytes on the wire")
coding = resp.headers.get("Content-Encoding", "")
return decode_http(bytes(wire), coding, max_output, max_seconds=2)By default requests downloads the whole body at once. The requests documentation explains that you can “defer downloading the response body until you access the Response.content attribute with the stream parameter”, and both functions use that. They differ in who does the decoding:
fetch_streamedis the common fix. It asks requests for the body in 64 KiB pieces that are already decoded, counts them, and stops past the cap. It relies on the HTTP library to decode in small steps.fetch_limitedreads the compressed bytes withdecode_content=False, enforces a wire cap, and hands them todecode_http. The HTTP library never decodes anything, so a bug in its decoder cannot hurt you. It also sends anAccept-Encodingthat lists only the codingsdecode_httpunderstands.
Both set allow_redirects=False. When requests follows a redirect, it reads the redirect response’s body to free the connection (the line resp.content # Consume socket so it can be released in its resolve_redirects code), and that read decodes the content with no cap of yours in the way. urllib3’s CVE-2026-21441 describes the library-side version of the same path. Follow redirects yourself by calling the same function on each Location.
Measure all three
Create step06_http.py. It starts the server, then runs each strategy against each page in a fresh child process (the same file serves as the child when it receives arguments), and prints the urllib3 version next to each result:
# step06_http.py
"""Three ways to download a hostile response, on the current urllib3 and on an old one."""
import json
import os
import subprocess
import sys
import time
import requests
import urllib3
from labkit import Server
from probe import MiB, peak_mib
from safe_decompress import Refused
from safe_fetch import fetch_limited, fetch_streamed
def child(mode, url):
"""Fetch one URL one way in this fresh process and print its peak memory as JSON."""
base, started = peak_mib(), time.perf_counter()
try:
if mode == "naive":
size = len(requests.get(url, timeout=30).content)
elif mode == "streamed":
size = len(fetch_streamed(url, 8 * MiB))
else:
size = len(fetch_limited(url, 1 * MiB, 8 * MiB))
result = f"ok, {size:,} bytes"
except Refused as exc:
result = f"refused: {exc.reason}"
print(json.dumps({"urllib3": urllib3.__version__, "result": result,
"peak_mib": round(peak_mib() - base), "seconds": round(time.perf_counter() - started, 2)}))
def parent():
old = os.environ.get("OLD_PYTHON") # a venv python with urllib3 2.5.0 (optional)
port = 8301
runs = [("naive", "/zeros-gzip"), ("streamed", "/zeros-gzip"), ("limited", "/zeros-gzip"),
("limited", "/zeros-br"), ("limited", "/ok")]
with Server([sys.executable, "bomb_server.py", str(port)], port, ready_path="/ready") as srv:
print("Accept-Encoding sent by requests here:", requests.get(srv.url + "/accept-encoding").text)
print(f"\n{'urllib3':<9}{'strategy':<10}{'page':<13}{'result':<26}{'peak memory':>12}{'seconds':>9}")
for python in [sys.executable] + ([old] if old else []):
for mode, path in runs:
done = subprocess.run([python, __file__, mode, srv.url + path],
capture_output=True, text=True, timeout=300, check=True)
r = json.loads(done.stdout.strip().splitlines()[-1])
print(f"{r['urllib3']:<9}{mode:<10}{path:<13}{r['result']:<26}{r['peak_mib']:>8,} MiB{r['seconds']:>9}")
if __name__ == "__main__":
child(*sys.argv[1:3]) if len(sys.argv) > 1 else parent()Run python step06_http.py (it uses port 8301):
Accept-Encoding sent by requests here: gzip, deflate, br, zstd
urllib3 strategy page result peak memory seconds
2.8.0 naive /zeros-gzip ok, 134,217,728 bytes 267 MiB 0.18
2.8.0 streamed /zeros-gzip refused: too_large 16 MiB 0.03
2.8.0 limited /zeros-gzip refused: too_large 17 MiB 0.03
2.8.0 limited /zeros-br refused: too_large 20 MiB 0.03
2.8.0 limited /ok ok, 1,288,469 bytes 3 MiB 0.01On the current urllib3 (2.8.0) the naive call took 267 MiB for a 128 MiB body. The streamed loop refused after the cap with a 16 MiB peak, and the limited fetch refused both bombs, gzip and Brotli, within 17 and 20 MiB. The last row is the check that matters for legitimate use: the normal 1.3 MB page came through intact.
The line at the top is a quiet lesson of its own. This copy of requests sent Accept-Encoding: gzip, deflate, br, zstd. urllib3 builds that header from what it can import: in its util/request.py it appends br when the brotli or brotlicffi package imports, and zstd when compression.zstd (Python 3.14 and newer) or backports.zstd does. Installing a package for some other reason therefore changes which decoders every server you contact can aim at, which matters in a week when a Brotli decoder flaw forced patches in Node.js, .NET and Python (our report).
The same test on an older urllib3
A streaming API is supposed to keep memory flat, so does fetch_streamed deserve trust? The four advisories say it did not always. Create a second environment with urllib3 2.5.0, a release that predates the fixes, and run the same script against both:
python -m venv .venv-old
.venv-old/bin/pip install requests==2.34.2 urllib3==2.5.0 brotli==1.2.1 psutil==7.2.2
.venv-old/bin/pip install "backports.zstd==1.8.0; python_version < '3.14'"
export OLD_PYTHON="$PWD/.venv-old/bin/python" # Git Bash on Windows: $PWD/.venv-old/Scripts/python.exe
python step06_http.pyAccept-Encoding sent by requests here: gzip, deflate, br, zstd
urllib3 strategy page result peak memory seconds
2.8.0 naive /zeros-gzip ok, 134,217,728 bytes 267 MiB 0.18
2.8.0 streamed /zeros-gzip refused: too_large 16 MiB 0.03
2.8.0 limited /zeros-gzip refused: too_large 17 MiB 0.03
2.8.0 limited /zeros-br refused: too_large 20 MiB 0.03
2.8.0 limited /ok ok, 1,288,469 bytes 3 MiB 0.01
2.5.0 naive /zeros-gzip ok, 134,217,728 bytes 274 MiB 15.4
2.5.0 streamed /zeros-gzip refused: too_large 129 MiB 2.53
2.5.0 limited /zeros-gzip refused: too_large 17 MiB 0.03
2.5.0 limited /zeros-br refused: too_large 20 MiB 0.03
2.5.0 limited /ok ok, 1,288,469 bytes 3 MiB 0.01Compare the two streamed rows. On urllib3 2.8.0 the loop stayed at 16 MiB. On 2.5.0 the same loop, with the same cap and the same 64 KiB reads, peaked at 129 MiB, essentially the whole 128 MiB body: the library decoded far more than one chunk inside a single read, before your counting code saw a byte. The advisory for CVE-2025-66471 describes this kind of behavior: “The decompression logic could cause urllib3 to fully decode a small amount of highly compressed data in a single operation.” A cap in your own code cannot help when the blow-up happens inside the call that is supposed to return one chunk. The limited rows are flat on both versions (17 and 20 MiB) because the old library never decoded anything. Two lessons: keep urllib3 current, and for input you do not trust, prefer decoding it yourself over relying on any library’s idea of a streaming limit. (The naive row on 2.5.0 also took 15.4 seconds against 0.18 on 2.8.0; I did not investigate why.)
Step 8: Door three, ZIP and tar.gz uploads
Archives add two dangers to the ones above: disk space instead of memory, and the sheer number of files. Python’s own documentation says it plainly. The zipfile page lists “Resources limitations”: “The lack of memory or disk volume would lead to decompression failed. For example, decompression bombs (aka ZIP bomb) apply to zipfile library that can cause disk volume exhaustion.” For tar, the extraction filters page says: “Among other issues, the pre-defined filters do not prevent denial-of-service attacks. Users should do additional checks.” So extractall, with or without filter="data", does not protect you from size.
Builders for test archives
First, a small module that builds the test archives. ZIP lets each member choose its own compression method, including bzip2, which is how the first test below fits 256 MiB into a few hundred bytes. zip_with_lying_size builds an archive whose headers claim 1,000 bytes for a member that really holds 4 MiB, by overwriting the uncompressed-size field of the local file header (at byte 22) and of the central directory entry (at byte 24). Create archive_kit.py:
# archive_kit.py
"""Build small test archives. The 'bombs' are zeros and live only in memory or a temp folder."""
import io
import struct
import tarfile
import zipfile
from codecs_kit import MiB
def make_zip(members, compression=zipfile.ZIP_DEFLATED):
"""members: list of (name, bytes). Returns the archive as bytes."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", compression) as zf:
for name, data in members:
zf.writestr(name, data)
return buf.getvalue()
def zip_with_lying_size(real_size=4 * MiB, claimed=1000):
"""One member whose headers claim `claimed` bytes while its stream holds `real_size`."""
raw = bytearray(make_zip([("big.bin", bytes(real_size))]))
local, central = raw.find(b"PK\x03\x04"), raw.find(b"PK\x01\x02")
struct.pack_into("<I", raw, local + 22, claimed) # uncompressed size in the local header
struct.pack_into("<I", raw, central + 24, claimed) # and in the central directory
return bytes(raw)
def make_targz(members):
"""members: list of (name, bytes). Returns a .tar.gz as bytes."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
for name, data in members:
info = tarfile.TarInfo(name)
info.size = len(data)
tar.addfile(info, io.BytesIO(data))
return buf.getvalue()The limits
Now the extractors. Limits holds three numbers: the most members, the largest member, and the largest total. LimitedReader is a file-like wrapper that raises Refused once more than limit bytes have been read through it, so anything that reads from it, such as shutil.copyfileobj or tarfile, gets the cap for free. Create safe_archive.py:
# safe_archive.py
"""Extract ZIP and tar.gz uploads with limits on member count, member size and total size."""
import gzip
import io
import shutil
import tarfile
import zipfile
from dataclasses import dataclass
from pathlib import Path
from codecs_kit import MiB
from safe_decompress import Refused
@dataclass
class Limits:
max_members: int = 1_000
max_member_bytes: int = 32 * MiB
max_total_bytes: int = 64 * MiB
class LimitedReader(io.RawIOBase):
"""Wrap a readable stream and refuse to hand out more than `limit` bytes in total."""
def __init__(self, raw, limit):
self.raw, self.limit, self.count = raw, limit, 0
def readable(self):
return True
def readinto(self, buffer):
data = self.raw.read(len(buffer))
self.count += len(data)
if self.count > self.limit:
raise Refused("too_large", f"more than {self.limit} bytes")
buffer[:len(data)] = data
return len(data)check_zip and extract_zip come next. check_zip looks only at the central directory, the list of member names and declared sizes at the end of the file, so it can refuse a bomb before decompressing a byte. extract_zip then streams each member through a LimitedReader whose limit is the smaller of the member cap and the budget left, so the real bytes are counted whatever the headers claim. The is_relative_to check keeps names inside the destination folder; the path traversal tutorial covers that topic in depth.
# safe_archive.py (continued)
def check_zip(zf, limits):
"""Cheap checks on the central directory, before a single byte is decompressed."""
infos = zf.infolist()
if len(infos) > limits.max_members:
raise Refused("too_many_members", f"{len(infos)} members, limit {limits.max_members}")
if any(info.file_size > limits.max_member_bytes for info in infos):
raise Refused("member_too_large", f"a member is declared above {limits.max_member_bytes} bytes")
total = sum(info.file_size for info in infos)
if total > limits.max_total_bytes:
raise Refused("archive_too_large", f"{total} bytes declared, limit {limits.max_total_bytes}")
def extract_zip(zf, dest, limits):
check_zip(zf, limits)
dest = Path(dest).resolve()
budget = limits.max_total_bytes
for info in zf.infolist():
target = (dest / info.filename).resolve()
if not target.is_relative_to(dest):
raise Refused("path_escape", "a member name points outside the destination")
if info.is_dir():
target.mkdir(parents=True, exist_ok=True)
continue
target.parent.mkdir(parents=True, exist_ok=True)
with zf.open(info) as src, open(target, "wb") as out:
reader = LimitedReader(src, min(limits.max_member_bytes, budget))
try:
shutil.copyfileobj(reader, out)
except zipfile.BadZipFile as exc:
raise Refused("bad_data", str(exc)) from None
budget -= reader.countFor tar.gz, the whole decompressed stream goes through one LimitedReader set to the total cap. The archive is read in streaming mode (r|), one member at a time, so nothing is extracted before its header has been checked, and extraction uses the data filter for the path and link rules:
# safe_archive.py (continued)
def extract_targz(data, dest, limits):
stream = LimitedReader(gzip.GzipFile(fileobj=io.BytesIO(data)), limits.max_total_bytes)
with tarfile.open(fileobj=stream, mode="r|") as tar:
for number, member in enumerate(tar, 1):
if number > limits.max_members:
raise Refused("too_many_members", f"more than {limits.max_members} members")
if member.size > limits.max_member_bytes:
raise Refused("member_too_large", f"a member is declared above {limits.max_member_bytes} bytes")
tar.extract(member, dest, filter="data")Naive against limited
step07_archives.py builds five archives and extracts each twice into temporary folders, once with the library’s own extractall and once with the limited extractors, measuring how many bytes ended up on disk:
# step07_archives.py
"""Archive uploads: naive extraction next to the limited one, measured in bytes written to disk."""
import io
import tarfile
import tempfile
import time
import zipfile
from pathlib import Path
from archive_kit import make_targz, make_zip, zip_with_lying_size
from codecs_kit import MiB
from safe_archive import Limits, extract_targz, extract_zip
from safe_decompress import Refused
LIMITS = Limits()
def on_disk(folder):
return sum(f.stat().st_size for f in Path(folder).rglob("*") if f.is_file())
def human(size):
return f"{size:,} bytes" if size < MiB else f"{size / MiB:,.0f} MiB"
def attempt(label, action, folder):
started = time.perf_counter()
try:
action()
verdict = "extracted"
except Refused as exc:
verdict = f"refused: {exc.reason}"
except (zipfile.BadZipFile, tarfile.TarError) as exc:
verdict = f"library error: {type(exc).__name__}"
print(f" {label:<9}{verdict:<30}{human(on_disk(folder)) + ' on disk':>22}{time.perf_counter() - started:>8.2f} s")
def zip_case(name, data):
print(f"{name}: archive is {len(data):,} bytes")
with tempfile.TemporaryDirectory() as plain, tempfile.TemporaryDirectory() as safe:
attempt("naive", lambda: zipfile.ZipFile(io.BytesIO(data)).extractall(plain), plain)
attempt("limited", lambda: extract_zip(zipfile.ZipFile(io.BytesIO(data)), safe, LIMITS), safe)
def tar_case(name, data):
print(f"{name}: archive is {len(data):,} bytes")
with tempfile.TemporaryDirectory() as plain, tempfile.TemporaryDirectory() as safe:
attempt("naive", lambda: tarfile.open(fileobj=io.BytesIO(data)).extractall(plain, filter="data"), plain)
attempt("limited", lambda: extract_targz(data, safe, LIMITS), safe)
zip_case("ZIP with one bzip2 member holding 256 MiB of zeros",
make_zip([("zeros.bin", bytes(256 * MiB))], zipfile.ZIP_BZIP2))
zip_case("ZIP with 20,000 empty members", make_zip([(f"d/{i}.txt", b"") for i in range(20000)]))
zip_case("ZIP whose headers claim 1,000 bytes for a 4 MiB member", zip_with_lying_size())
tar_case("tar.gz with one member of 256 MiB", make_targz([("zeros.bin", bytes(256 * MiB))]))
tar_case("tar.gz with 20 members of 30 MiB each", make_targz([(f"part{i}.bin", bytes(30 * MiB)) for i in range(20)]))Run python step07_archives.py. It writes up to 600 MiB to a temporary folder at a time and deletes it afterwards:
ZIP with one bzip2 member holding 256 MiB of zeros: archive is 324 bytes
naive extracted 256 MiB on disk 0.69 s
limited refused: member_too_large 0 bytes on disk 0.00 s
ZIP with 20,000 empty members: archive is 1,977,802 bytes
naive extracted 0 bytes on disk 4.66 s
limited refused: too_many_members 0 bytes on disk 0.03 s
ZIP whose headers claim 1,000 bytes for a 4 MiB member: archive is 4,192 bytes
naive library error: BadZipFile 0 bytes on disk 0.00 s
limited refused: bad_data 0 bytes on disk 0.00 s
tar.gz with one member of 256 MiB: archive is 260,994 bytes
naive extracted 256 MiB on disk 0.48 s
limited refused: member_too_large 0 bytes on disk 0.00 s
tar.gz with 20 members of 30 MiB each: archive is 612,689 bytes
naive extracted 600 MiB on disk 1.11 s
limited refused: too_large 64 MiB on disk 0.14 sDeclared sizes can be checked before you decompress anything
The first archive is 324 bytes long. Naive extraction wrote 256 MiB to disk in under a second; the limited extractor refused it as member_too_large from the central directory alone and wrote nothing. The tar.gz with one 256 MiB member (260,994 bytes) went the same way: tar headers record the true size of each member, so the header check caught it. The ZIP with 20,000 empty members is a different bomb, one made of file count: naive extraction needed 4.66 seconds to create 20,000 files, while the limited extractor refused on the member count in 0.03 seconds. Note that Python has parsed the central directory by the time that check runs, so keep the wire cap in front of it.
Declared sizes can also lie
The fourth case is the ZIP whose headers claim 1,000 bytes for a 4 MiB member. In CPython the declared size works as a cap: ZipExtFile sets self._left = zipinfo.file_size and truncates each read with data = data[:self._left], and the CRC check then fails, so the naive extractall raised BadZipFile and wrote nothing. The limited extractor also counts real bytes and reported bad_data. I tested only CPython’s zipfile, and a lying header may behave differently in another ZIP library, which is a good reason to count the bytes you actually receive instead of trusting a number in the file.
Honest sizes can still add up
The last archive has 20 members of 30 MiB, each under the 32 MiB member cap, so no header check objects. Naive extraction wrote 600 MiB. The limited extractor stopped at 64 MiB on disk after 0.14 seconds, because the total counter sits on the decompressed stream and does not care how the archive is divided into members. This is the case that per-file checks alone cannot catch.
Step 9: Prove it with tests
A fix you cannot re-check is a fix that will quietly break. Create test_safe.py. The first part covers the decompressor on all six formats. Most tests are parametrized, which means pytest runs them once per format:
# test_safe.py
"""Tests for the bounded decompressor, the HTTP coding helper and the archive extractors."""
import gzip
import io
import tracemalloc
import types
import zipfile
import pytest
import safe_decompress
from archive_kit import make_targz, make_zip, zip_with_lying_size
from codecs_kit import CODECS, MiB, compress, make_test_payload, sample_document, zstd
from safe_archive import Limits, extract_targz, extract_zip
from safe_decompress import Refused, decode_http, decompress_limited
DOC = sample_document(3000)
SMALL = Limits(max_members=10, max_member_bytes=1 * MiB, max_total_bytes=2 * MiB)
def disk(folder):
return sum(p.stat().st_size for p in folder.rglob("*") if p.is_file())
@pytest.mark.parametrize("codec", CODECS)
def test_round_trip(codec):
assert decompress_limited(codec, compress(codec, DOC), 4 * MiB) == DOC
@pytest.mark.parametrize("codec", CODECS)
def test_bomb_is_refused_and_memory_stays_small(codec):
bomb = make_test_payload(codec, 32 * MiB)
tracemalloc.start()
with pytest.raises(Refused) as caught:
decompress_limited(codec, bomb, 1 * MiB)
_, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
assert caught.value.reason == "too_large"
assert peak < 16 * MiB # xz alone holds its 8 MiB dictionary on top of our 1 MiB cap
@pytest.mark.parametrize("codec", CODECS)
def test_limit_is_inclusive(codec):
data = compress(codec, bytes(5000))
assert decompress_limited(codec, data, 5000) == bytes(5000)
with pytest.raises(Refused) as caught:
decompress_limited(codec, data, 4999)
assert caught.value.reason == "too_large"
@pytest.mark.parametrize("codec", CODECS)
def test_zero_limit_never_means_unlimited(codec):
assert decompress_limited(codec, compress(codec, b""), 0) == b""
with pytest.raises(Refused) as caught:
decompress_limited(codec, compress(codec, b"x"), 0)
assert caught.value.reason == "too_large"
@pytest.mark.parametrize("codec", CODECS)
def test_truncated_stream(codec):
data = compress(codec, DOC)
with pytest.raises(Refused) as caught:
decompress_limited(codec, data[: len(data) // 2], 4 * MiB)
assert caught.value.reason == "truncated"
@pytest.mark.parametrize("codec", CODECS)
def test_bytes_after_the_end(codec):
with pytest.raises(Refused) as caught:
decompress_limited(codec, compress(codec, DOC) + b"junk", 4 * MiB)
assert caught.value.reason in {"trailing_data", "bad_data"}
@pytest.mark.parametrize("codec", CODECS)
def test_garbage_input(codec):
with pytest.raises(Refused) as caught:
decompress_limited(codec, bytes(range(256)) * 4, 4 * MiB)
assert caught.value.reason in {"bad_data", "truncated"}A few of these deserve a comment. test_bomb_is_refused_and_memory_stays_small uses tracemalloc to check that the peak stays under 16 MiB while refusing a 32 MiB expansion. My first threshold was 8 MiB, and the xz case failed at 10.5 MB: the xz decoder holds its own 8 MiB dictionary on top of our 1 MiB cap, so the threshold became 16 MiB with a comment saying why. test_limit_is_inclusive checks the off-by-one: exactly the cap passes and one byte more is refused. test_zero_limit_never_means_unlimited guards Trap 2. The second part covers the multi-member policy, the clock, and the HTTP helper:
# test_safe.py (continued)
def test_two_gzip_members_are_refused_on_purpose():
two = gzip.compress(b"one;") + gzip.compress(b"two")
assert gzip.decompress(two) == b"one;two"
with pytest.raises(Refused) as caught:
decompress_limited("gzip", two, 1 * MiB)
assert caught.value.reason == "trailing_data"
def test_deadline(monkeypatch):
clock = iter(range(0, 1000, 10))
monkeypatch.setattr(safe_decompress, "time", types.SimpleNamespace(monotonic=lambda: next(clock)))
with pytest.raises(Refused) as caught:
decompress_limited("gzip", make_test_payload("gzip", 8 * MiB), 64 * MiB, max_seconds=15)
assert caught.value.reason == "too_slow"
def test_codings_are_undone_in_reverse_order():
body = compress("br", compress("gzip", DOC)) # gzip was applied first, then br
assert decode_http(body, "gzip, br", 4 * MiB, max_layers=2) == DOC
def test_layer_limit_and_unknown_coding():
layered = gzip.compress(gzip.compress(b"x"))
with pytest.raises(Refused) as caught:
decode_http(layered, "gzip, gzip", 1 * MiB)
assert caught.value.reason == "too_many_layers"
with pytest.raises(Refused) as caught:
decode_http(b"x", "compress", 1 * MiB)
assert caught.value.reason == "unsupported_encoding"
def test_zstd_window_is_capped_at_8_mib_for_http_bodies():
packer = zstd.ZstdCompressor(options={zstd.CompressionParameter.window_log: 27})
frame = packer.compress(b"hello") + packer.flush(zstd.ZstdCompressor.FLUSH_FRAME)
assert decompress_limited("zstd", frame, 1 * MiB) == b"hello" # fine as an uploaded file
with pytest.raises(Refused) as caught:
decode_http(frame, "zstd", 1 * MiB) # refused as an HTTP body
assert caught.value.reason == "bad_data"
def test_an_uncoded_body_is_capped_too():
assert decode_http(b"abc", "identity", 1 * MiB) == b"abc"
with pytest.raises(Refused) as caught:
decode_http(bytes(2 * MiB), "", 1 * MiB)
assert caught.value.reason == "too_large"test_deadline replaces the module’s time with a fake clock that jumps 10 seconds per call, so it needs no real waiting. test_codings_are_undone_in_reverse_order uses two different formats on purpose; with two gzip layers a wrong order would go unnoticed. The third part covers the archive extractors, each with its own folder from pytest’s tmp_path:
# test_safe.py (continued)
def test_zip_bomb_is_refused_from_the_headers_alone(tmp_path):
bomb = make_zip([("zeros.bin", bytes(8 * MiB))], zipfile.ZIP_BZIP2)
with pytest.raises(Refused) as caught:
extract_zip(zipfile.ZipFile(io.BytesIO(bomb)), tmp_path, SMALL)
assert caught.value.reason == "member_too_large"
assert disk(tmp_path) == 0
def test_zip_member_count_and_total_size(tmp_path):
many = make_zip([(f"{i}.txt", b"") for i in range(50)])
with pytest.raises(Refused) as caught:
extract_zip(zipfile.ZipFile(io.BytesIO(many)), tmp_path, SMALL)
assert caught.value.reason == "too_many_members"
big = make_zip([(f"{i}.bin", bytes(900_000)) for i in range(3)])
with pytest.raises(Refused) as caught:
extract_zip(zipfile.ZipFile(io.BytesIO(big)), tmp_path, SMALL)
assert caught.value.reason == "archive_too_large"
def test_zip_with_a_lying_size_writes_at_most_the_claimed_bytes(tmp_path):
zf = zipfile.ZipFile(io.BytesIO(zip_with_lying_size(real_size=4 * MiB, claimed=1000)))
with pytest.raises(Refused) as caught:
extract_zip(zf, tmp_path, SMALL)
assert caught.value.reason == "bad_data"
assert disk(tmp_path) <= 1000
def test_zip_name_cannot_leave_the_folder(tmp_path):
evil = make_zip([("../evil.txt", b"x")])
with pytest.raises(Refused) as caught:
extract_zip(zipfile.ZipFile(io.BytesIO(evil)), tmp_path / "inside", SMALL)
assert caught.value.reason == "path_escape"
def test_honest_zip_and_targz_extract(tmp_path):
members = [("a.txt", b"alpha"), ("sub/b.txt", b"beta" * 100)]
extract_zip(zipfile.ZipFile(io.BytesIO(make_zip(members))), tmp_path / "z", SMALL)
extract_targz(make_targz(members), tmp_path, SMALL)
for name, data in members:
assert (tmp_path / "z" / name).read_bytes() == data
assert (tmp_path / name).read_bytes() == data
def test_targz_total_is_enforced_on_the_decompressed_stream(tmp_path):
members = [(f"part{i}.bin", bytes(900_000)) for i in range(5)] # each under the member cap
with pytest.raises(Refused) as caught:
extract_targz(make_targz(members), tmp_path, SMALL)
assert caught.value.reason == "too_large"
assert disk(tmp_path) <= SMALL.max_total_bytesThe API tests run against a real uvicorn process, so they exercise the same path as a client would. Create test_app.py:
# test_app.py
"""Tests for the bounded notes API, run against a real uvicorn process."""
import gzip
import json
import sys
import pytest
import requests
from codecs_kit import MiB
from labkit import Server
@pytest.fixture(scope="module")
def url():
cmd = [sys.executable, "-m", "uvicorn", "app:app", "--port", "8203", "--log-level", "warning"]
with Server(cmd, 8203) as srv:
yield srv.url
def post(url, data, encoding="gzip"):
headers = {"Content-Type": "application/json"}
if encoding:
headers["Content-Encoding"] = encoding
return requests.post(url + "/notes", data=data, headers=headers, timeout=60)
def note(size):
return json.dumps({"title": "t", "text": "A" * size}).encode()
def test_a_normal_gzip_note_is_accepted(url):
r = post(url, gzip.compress(note(1000)))
assert r.status_code == 200 and r.json()["chars"] == 1000
def test_a_small_plain_note_is_accepted(url):
assert post(url, note(1000), encoding="").status_code == 200
def test_a_decompression_bomb_gets_413(url):
r = post(url, gzip.compress(note(64 * MiB), 9))
assert r.status_code == 413 and r.json()["detail"] == "too_large"
def test_an_unknown_coding_gets_415(url):
assert post(url, gzip.compress(note(10)), encoding="compress").status_code == 415
def test_a_truncated_body_gets_400(url):
packed = gzip.compress(note(5000))
assert post(url, packed[: len(packed) // 2]).status_code == 400Run python -m pytest test_safe.py test_app.py -q:
........................................................... [100%]
59 passed in 3.14sCheck that the tests can fail
A passing suite proves little until you have seen it fail. mutations.py copies the lab to a temporary folder, breaks one line of safe_decompress.py at a time in nine different ways, and runs the tests against each broken copy. A mutant that survives points at a missing test:
# mutations.py
"""Break safe_decompress.py on purpose, nine ways, and check that the tests notice each time."""
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
LAB = Path(__file__).resolve().parent
MUTANTS = [
("room without the extra byte", "room = min(max_output - len(out) + 1, STEP)", "room = min(max_output - len(out), STEP)"),
("limit is exclusive", "if len(out) > max_output:\n raise Refused(\"too_large\"", "if len(out) >= max_output:\n raise Refused(\"too_large\""),
("truncation ignored", "raise Refused(\"truncated\", \"the stream ends before its end marker\")", "break"),
("trailing bytes ignored", "if dec.unused or pos < len(data):", "if False:"),
("deadline ignored", "if deadline is not None and time.monotonic() > deadline:", "if False:"),
("layer limit ignored", "if len(codings) > max_layers:", "if False:"),
("codings undone in the wrong order", "for coding in reversed(codings):", "for coding in codings:"),
("uncoded body not capped", " if len(body) > max_output:\n raise Refused(\"too_large\", f\"more than {max_output} bytes\")\n return body", " return body"),
("zstd window cap dropped", "options = {zstd.DecompressionParameter.window_log_max: window_log_max}", "options = {}"),
]
env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1", COLUMNS="170")
caught = 0
for name, old, new in MUTANTS:
with tempfile.TemporaryDirectory() as tmp:
for path in LAB.glob("*.py"):
shutil.copy(path, tmp)
target = Path(tmp) / "safe_decompress.py"
text = target.read_text(encoding="utf-8")
assert text.count(old) == 1, f"pattern for {name!r} must occur once"
target.write_text(text.replace(old, new), encoding="utf-8", newline="\n")
run = subprocess.run([sys.executable, "-m", "pytest", "test_safe.py", "-x", "-q", "-p", "no:cacheprovider"],
cwd=tmp, env=env, capture_output=True, text=True, timeout=600)
failed = [line for line in run.stdout.splitlines() if line.startswith("FAILED")]
verdict = f"caught by {failed[0].split()[1].split('::')[1]}" if failed else "SURVIVED"
caught += bool(failed)
print(f"{name:<36}{verdict}")
print(f"\n{caught} of {len(MUTANTS)} mutants caught")Run python mutations.py:
room without the extra byte caught by test_bomb_is_refused_and_memory_stays_small[gzip]
limit is exclusive caught by test_limit_is_inclusive[gzip]
truncation ignored caught by test_truncated_stream[gzip]
trailing bytes ignored caught by test_bytes_after_the_end[gzip]
deadline ignored caught by test_deadline
layer limit ignored caught by test_layer_limit_and_unknown_coding
codings undone in the wrong order caught by test_codings_are_undone_in_reverse_order
uncoded body not capped caught by test_an_uncoded_body_is_capped_too
zstd window cap dropped caught by test_zstd_window_is_capped_at_8_mib_for_http_bodies
9 of 9 mutants caughtAll nine mutants were caught, and the output names the first test that noticed each one. It was not 9 of 9 the first time. The mutant that removes the Zstandard window cap survived, because no test used a frame with a large window; test_zstd_window_is_capped_at_8_mib_for_http_bodies was written to kill it. Look also at what the first mutant, which drops the “plus one byte” from room, does to each format. For gzip and deflate the verdict stays correct but the peak memory rises to about 82 MB instead of staying under 16 MiB, because a max_length of zero means no limit (Trap 2), so only the memory test fails. For bzip2, xz and Zstandard a room of zero produces nothing and the loop wrongly reports truncated, so the boundary and zero-limit tests fail as well. A suite without a memory assertion would have missed the gzip case.
Step 10: Choose limits for your own service
The numbers in this lab suit a small notes API and are not recommendations. Choose yours from what your endpoint legitimately handles, starting from the largest real input you expect and adding headroom:
| Limit | What it bounds | Lab value | How to choose |
|---|---|---|---|
| Wire size | Compressed bytes read before decoding | 256 KiB (API), 1 MiB (client) | Largest legitimate compressed payload plus headroom. Count while reading, not after. |
| Output size | Decoded bytes, applied to every layer and to uncoded bodies | 2 MiB (API), 8 MiB (client), 64 MiB (archive total) | Largest legitimate decoded payload plus headroom. |
| Time | Wall-clock time for one decode | 1 s (API), 2 s (client) | Several times the longest legitimate decode you have measured. Decoding blocks the event loop in an async service, so move large caps to a worker thread or process. |
| Layers | Codings in one Content-Encoding header |
1 | One, unless a real client needs more. |
| Archive members | Files in one archive | 1,000 | Largest real archive plus headroom. |
| Member size | Declared and counted bytes of one file | 32 MiB | Largest legitimate single file. |
| Decoder memory | Zstandard window, xz dictionary | 8 MiB window for HTTP, 128 MiB xz limit | Keep 8 MiB for HTTP bodies (RFC 9659). Raise it only for files you know came from long-window modes. |
When you refuse an input, return a clear status (413, 415 or 400 as in Step 6), and log the reason code, the format, the wire size, the cap and the client address. Do not log payload bytes, and treat archive member names as data that can contain surprises. Beyond the code, the usual operating measures still apply: a memory limit on the container, a timeout on the proxy in front, a separate low-privilege worker process for archive extraction, and rate limits per client. None of those was tested in this lab, but they are the backstop for the day a limit is set too high.
Confirm everything works end to end
Create run_all.py, which runs every step in order and saves each output in an out folder:
# run_all.py
"""Run every step in order and save each output under out/."""
import os
import subprocess
import sys
STEPS = ["step01_ratios", "step02_naive", "step03_bounded", "step04_gotchas",
"step05_fastapi", "step06_http", "step07_archives"]
os.makedirs("out", exist_ok=True)
env = dict(os.environ, COLUMNS="170")
for step in STEPS:
done = subprocess.run([sys.executable, f"{step}.py"], capture_output=True, text=True, env=env, timeout=900)
with open(f"out/{step}.txt", "w", encoding="utf-8", newline="\n") as f:
f.write(done.stdout)
print(f"{step}: exit {done.returncode}, {len(done.stdout.splitlines())} lines")
for name, cmd in (("pytest", [sys.executable, "-m", "pytest", "test_safe.py", "test_app.py", "-q", "-p", "no:cacheprovider"]),
("mutations", [sys.executable, "mutations.py"])):
done = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=900)
with open(f"out/{name}.txt", "w", encoding="utf-8", newline="\n") as f:
f.write(done.stdout)
print(f"{name}: exit {done.returncode}, last line: {done.stdout.strip().splitlines()[-1]}")Run python run_all.py, which takes a couple of minutes. Every step should exit with 0, pytest should end with 59 passed, and the mutation script with 9 of 9 mutants caught. Your numbers will differ a little (memory by a few MiB, times by fractions of a second) but not in kind: the naive rows should sit near the size of the output, and every bounded row near your cap.
Where to go next
- Make it a rule. Route every decompression of untrusted input through
decompress_limited, and add a code review check or a smallastscript that flags direct calls togzip.decompress,zlib.decompressand friends. The open redirect tutorial shows one way to write such an audit. - Other formats have their own bombs. Images have a pixel count instead of a byte count (the secure file uploads tutorial measures Pillow’s limit), and XML has entity expansion (the untrusted XML tutorial).
- Put your own data through it. The Zstandard dictionary tutorial applies a decompression guard to records you compress yourself.
- Stay current on the library under you. Pin and update urllib3 deliberately, and subscribe to its security advisories, since the four above came in five months.
Sources and further reading
- CWE-409: Improper Handling of Highly Compressed Data (Data Amplification), MITRE.
- urllib3 security advisories GHSA-gm62-xv2j-4w53, GHSA-2xpw-w6gg-jr37, GHSA-38jv-5279-wg99 and GHSA-mf9v-mfxr-j63j.
- Python documentation: zlib, bz2, lzma, compression.zstd, zipfile decompression pitfalls and tarfile extraction filters.
- RFC 9659, Window Sizing for Zstandard Content Encoding.
- OWASP File Upload Cheat Sheet.
- FastAPI: Custom Request and APIRoute class and Requests: Advanced Usage.
- Brotli on PyPI, whose
Decompressor.processdocstring documentsoutput_buffer_limit.
No Comment! Be the first one.