open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 4 Activity

test(assets): the cost of a link, and a property over all 1 110 chains

Browse source 
Round 3 of the 0.10.1 review. Rounds 1-3 each bound a NUMBER and the bomb
moved to the next one: the declared size, then the first `FlateDecode`, then
every `FlateDecode`. It now lives in a link this package had documented as
safe. `ASCII85Decode` was called bounded "by its own input because it
shrinks"; `z` is ASCII85's shorthand for four zero bytes, so one input byte
becomes four, and `base64.a85decode` appends one 4-byte object per group to a
list, so the DECODER costs about a hundred bytes of memory per byte of input
(measured on CPython 3.14: 101.4x at 1 MiB, 96.1x at 4 MiB, 94.5x at 16 MiB).

Measured on the pinned `0c3c490` tree, its own interpreter, idle machine: a
33 475-byte PDF decoding an image through `/Filter [/FlateDecode
/ASCII85Decode]` cost 3 827 003 392 bytes of peak RSS -- 114 000x the file --
and the picture was CARRIED, with no rejection at all.

Seven tests red, three green:

* the two bombs above, in their own interpreters, at the shipped bound;
* the cost ratio as a re-measurable known-negative, so the constant cannot
  rot the first time CPython changes `a85decode`;
* the input cap against the corpora (9 668 image objects over 77 PDFs, 16
  behind an `ASCII85Decode` link, largest input 450 739 bytes);
* THE PROPERTY, both payload fills: every chain of length 1-3 over the ten
  filters pdfminer decodes -- 1 110 of them -- is either delivered under the
  bound or refused with a code in the published vocabulary, and never paid
  for on the way, which `tracemalloc` measures because that is where
  `a85decode`'s cost lives.

Green but previously uncovered, which is the MAJOR of the same checkpoint:
`check_payload` at the END of `_check_stream_cost` could be deleted with the
whole suite still passing, because the second one after `get_data()` gives
the same code one step later. The two differ in whether the payment was made,
so the test asserts `get_data` was never called. The known-positive beside
the property -- every bounded chain still carries a 64-byte image -- is green
too, and a rule that refuses everything would pass the property alone.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-18 17:20:51 +02:00
commit 33d3269380
Signed by: ktg
SSH key fingerprint: SHA256:JakMjO6FTBBzN0Bhfj9saOoEjaFxlSdYuZQQpM/lF9Q
1 changed files with 352 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

352
tests/test_asset_limits.py
View file

@ -730,3 +730,355 @@ def test_a_decoded_stream_under_the_bound_is_still_read() -> None:
stream.get_data()
image = extract_module._pdf_image(stream, "decoded")
assert (image.width, image.height) == (8, 8)
# --- BLOCKER of the 18.09 round-3 checkpoint: the COST of a link, not the ---
# --- size of its output -----------------------------------------------------
#
# Round 1 bound what the dictionary DECLARES; the bomb moved into the stream.
# Round 2 bound the first `FlateDecode`; it moved into the second one. Round 3
# bound every `FlateDecode` link; it moved into a link this package had called
# safe. `ASCII85Decode` was documented as bounded "by its own input because it
# shrinks", and that is false: `z` is ASCII85's shorthand for four zero bytes,
# so one input byte becomes four. Worse, the DECODER's working set is far
# larger than either number -- `base64.a85decode` appends one 4-byte object per
# group to a list -- so the cost is about a hundred bytes of memory per byte of
# INPUT. Measured on CPython 3.14: 101.4x at 1 MiB of `z`, 96.1x at 4 MiB,
# 94.5x at 16 MiB. Nothing in the output size says so.
#
# Measured 2026-09-18 on the pinned `0c3c490` tree, its own interpreter, idle
# machine: a 33 475-byte PDF whose image decodes through `/Filter
# [/FlateDecode /ASCII85Decode]` cost 3 827 003 392 bytes of peak RSS -- about
# 114 000x the file -- and the picture was CARRIED, with no rejection at all.
#
# So the rule under test is not "bound the output of a link" but "bound what
# decoding a link COSTS", and it has to hold for every chain rather than for
# the ones a corpus happens to hold.
#: The filters pdfminer will decode, by their canonical names, plus `/Crypt`.
#: The property test below runs EVERY chain of length 1-3 over these.
DECODED_FILTERS = (
"FlateDecode",
"ASCII85Decode",
"ASCIIHexDecode",
"DCTDecode",
"JPXDecode",
"JBIG2Decode",
"LZWDecode",
"RunLengthDecode",
"CCITTFaxDecode",
"Crypt",
)
#: What a rejection may be CALLED. A refusal outside this vocabulary is a
#: defect even when it stops the bomb: a consumer counts these.
ASSET_REJECTION_CODES = frozenset(
{
"asset_too_large",
"asset_size_invalid",
"asset_pdf_unbounded",
"asset_pdf_unsupported",
"asset_samples_invalid",
"asset_type_unknown",
}
)
def _encode_for(literal: str, data: bytes) -> bytes:
"""Encode `data` so that DECODING it with `literal` gives `data` back.
The filters with no encoder here are the ones this package refuses unread,
so a chain containing one never reaches a decode at all.
"""
import base64
import binascii
if literal in ("FlateDecode", "Fl"):
return zlib.compress(data, 9)
if literal in ("ASCII85Decode", "A85"):
return base64.a85encode(data, adobe=True)
if literal in ("ASCIIHexDecode", "AHx"):
return binascii.hexlify(data) + b">"
return data
def _chain_stream(chain: tuple[str, ...], payload: bytes) -> bytes:
"""The stream bytes a document must hold for `chain` to decode to
`payload`. The LAST filter is applied to the payload first.
A payload of zeros is the amplifying case for BOTH ends of the chain: it
deflates to almost nothing, and `a85encode` writes a run of `z`, which is
the shorthand this bound exists for.
"""
for literal in reversed(chain):
payload = _encode_for(literal, payload)
return payload
def _filters_entry(chain: tuple[str, ...]) -> str:
return "[" + " ".join("/" + name for name in chain) + "]"
_CHAIN_CHILD = """
import resource, sys
sys.path.insert(0, {tests!r})
from test_asset_limits import _bomb, _chain_stream, _filters_entry
from llm_ingestion_okf.extract import extract_document
chain = {chain!r}
document = _bomb(
1, payload=_chain_stream(chain, b"\\x00" * {payload}), filters=_filters_entry(chain)
)
extracted = extract_document("bomb.pdf", document, assets=True)
peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
print(
len(document),
len(extracted.images),
",".join(r.code for r in extracted.rejected) or "-",
peak if sys.platform == "darwin" else peak * 1024,
)
"""
def _run_chain(chain: tuple[str, ...], *, payload: int) -> tuple[int, int, str, int]:
"""One chain's bomb in its own interpreter, so peak RSS is ITS peak and not
the high-water mark of every test that ran before it."""
completed = subprocess.run(
[
sys.executable,
"-c",
_CHAIN_CHILD.format(tests=str(Path(__file__).parent), chain=chain, payload=payload),
],
capture_output=True,
text=True,
check=True,
)
size, carried, codes, peak = completed.stdout.split()
return int(size), int(carried), codes, int(peak)
def test_an_ascii85_link_behind_a_flate_link_is_bounded_too() -> None:
"""The round-3 BLOCKER at the shipped bound, in its own interpreter."""
pytest.importorskip("pdfplumber")
size, carried, codes, peak = _run_chain(
("FlateDecode", "ASCII85Decode"), payload=128 * 1024 * 1024
)
assert size < 2 * 1024 * 1024, "the fixture must stay a small file, or it proves nothing"
assert carried == 0, "a run of `z` was carried as a 1x1 picture"
assert codes in ASSET_REJECTION_CODES, codes
assert peak < PEAK_RSS_BOUND, f"peak RSS {peak} bytes for a {size}-byte file"
def test_an_ascii85_link_on_its_own_is_bounded() -> None:
"""The same amplification with no filter in front of it: the stream IS the
run of `z`, so the cost must not be a multiple of the file."""
pytest.importorskip("pdfplumber")
size, carried, codes, peak = _run_chain(("ASCII85Decode",), payload=32 * 1024 * 1024)
assert carried == 0
assert codes in ASSET_REJECTION_CODES, codes
assert peak < PEAK_RSS_BOUND, f"peak RSS {peak} bytes for a {size}-byte file"
_COST_CHILD = """
import base64, resource, sys
base = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
data = b"z" * {count}
out = base64.a85decode(data)
peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
scale = 1 if sys.platform == "darwin" else 1024
print((peak - base) * scale / len(data))
"""
@pytest.mark.parametrize("count", [1 << 20, 4 << 20])
def test_the_ascii85_cost_ratio_is_not_below_the_one_this_package_measured(count: int) -> None:
"""The bound rests on a number measured from the DECODER, so the number has
to stay re-measurable or it rots the first time CPython changes
`a85decode`. This is the known-negative beside the bound: if the ratio here
ever rises above the constant, every cap derived from it is too generous
and this says so before a corpus does.
"""
completed = subprocess.run(
[sys.executable, "-c", _COST_CHILD.format(count=count)],
capture_output=True,
text=True,
check=True,
)
measured = float(completed.stdout.strip())
assert measured > 1.0, "the measurement itself must see the cost, or it proves nothing"
assert assets.PDF_FILTER_COST_RATIO["ASCII85Decode"] >= measured, (
f"a85decode costs {measured:.1f} bytes per input byte at {count} bytes of input, "
f"above the {assets.PDF_FILTER_COST_RATIO['ASCII85Decode']} this package budgets"
)
def test_the_ascii85_input_cap_stands_above_every_such_stream_in_the_corpora() -> None:
"""The cap is READ OFF the corpora, the posture `MAX_IMAGE_PIXELS` has.
Measured 2026-09-18 over the 9 668 image objects of the 77 PDFs on this
machine: 16 decode through an `ASCII85Decode` link, and the largest input
to one of them is 450 739 bytes.
"""
largest_measured = 450_739
cap = assets.MAX_FILTER_DECODE_BYTES / assets.PDF_FILTER_COST_RATIO["ASCII85Decode"]
assert cap > 10 * largest_measured, (
f"the cap is {cap:.0f} bytes, not an order of magnitude above the "
f"{largest_measured}-byte stream the corpora hold"
)
def _all_chains() -> list[tuple[str, ...]]:
import itertools
return [
tuple(chain)
for length in (1, 2, 3)
for chain in itertools.product(DECODED_FILTERS, repeat=length)
]
ALL_CHAINS = _all_chains()
@pytest.mark.parametrize("fill", [b"\x00", b"\xab"], ids=["zeros", "repeat"])
def test_no_chain_of_up_to_three_filters_is_carried_over_the_bound(fill: bytes) -> None:
"""THE PROPERTY, over every chain of length 1-3 the filters pdfminer
decodes can form -- 1 110 of them -- rather than over the shapes a corpus
happens to hold.
Each chain is given an amplifying payload and a small budget, and the
requirement is one sentence: the picture is either delivered with its bytes
under the bound, or refused with a code in the published vocabulary; never
carried over the bound, and never paid for on the way. `tracemalloc`
measures the paying, because it counts Python's own allocations, which is
exactly where `a85decode`'s cost lives.
"""
pytest.importorskip("pdfplumber")
import tracemalloc
from pdfminer.pdftypes import PDFStream
from pdfminer.psparser import LIT
from llm_ingestion_okf import extract as extract_module
budget = 8 * 1024
payload = fill * (128 * 1024)
ceiling = 4 * 1024 * 1024
carried_over_bound: list[tuple[str, ...]] = []
wrong_code: list[tuple[tuple[str, ...], str]] = []
over_ceiling: list[tuple[tuple[str, ...], int]] = []
monkey = pytest.MonkeyPatch()
try:
monkey.setattr(assets, "MAX_IMAGE_BYTES", budget)
monkey.setattr(assets, "MAX_FILTER_DECODE_BYTES", 2 * budget)
for chain in ALL_CHAINS:
stream = PDFStream(
{
"Width": 1,
"Height": 1,
"BitsPerComponent": 8,
"Filter": [LIT(name) for name in chain],
},
_chain_stream(chain, payload),
)
stream.set_objid(1, 0)
tracemalloc.start()
try:
extract_module._check_stream_cost(stream, "bomb")
data = stream.get_data()
except ExtractionError as exc:
if exc.code not in ASSET_REJECTION_CODES:
wrong_code.append((chain, exc.code))
except Exception:
pass # the reader behind the bound reports it in its own words
else:
if len(data) > budget:
carried_over_bound.append(chain)
finally:
_, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
if peak > ceiling:
over_ceiling.append((chain, peak))
finally:
monkey.undo()
assert carried_over_bound == [], (
f"{len(carried_over_bound)} of {len(ALL_CHAINS)} chains carried over the bound: "
f"{carried_over_bound[:4]}"
)
assert wrong_code == [], f"{len(wrong_code)} of {len(ALL_CHAINS)}: {wrong_code[:4]}"
assert over_ceiling == [], (
f"{len(over_ceiling)} of {len(ALL_CHAINS)} chains paid over {ceiling} bytes: "
f"{over_ceiling[:4]}"
)
def test_every_bounded_chain_still_carries_a_small_image() -> None:
"""The known-positive beside the property: over the same chain space,
restricted to the filters this package bounds, an image UNDER the bound
must still be delivered. A rule that refuses everything passes the property
above and is worthless."""
pytest.importorskip("pdfplumber")
import itertools
from pdfminer.pdftypes import PDFStream
from pdfminer.psparser import LIT
from llm_ingestion_okf import extract as extract_module
bounded = tuple(sorted(extract_module.bounded_pdf_filters()))
chains = [
tuple(chain) for length in (1, 2, 3) for chain in itertools.product(bounded, repeat=length)
]
refused: list[tuple[tuple[str, ...], str]] = []
for chain in chains:
stream = PDFStream(
{
"Width": 8,
"Height": 8,
"BitsPerComponent": 8,
"Filter": [LIT(name) for name in chain],
},
_chain_stream(chain, b"\x00" * 64),
)
stream.set_objid(1, 0)
try:
extract_module._check_stream_cost(stream, "small")
except ExtractionError as exc:
refused.append((chain, exc.code))
assert refused == [], f"{len(refused)} of {len(chains)} bounded chains refused a 64-byte image"
def test_the_stream_bound_refuses_before_get_data_is_ever_called() -> None:
"""MAJOR of the 18.09 checkpoint: `check_payload` at the END of
`_check_stream_cost` is what refuses a stream no filter in the chain
expands -- an unfiltered one, or one behind `DCTDecode` -- and deleting
that line passed the whole suite, because the SECOND `check_payload`, after
`get_data()`, produces the same code and the same words one step later.
A test reading the code cannot tell the two apart. What separates them is
whether the payment was made, so this one asserts `get_data` was never
called.
"""
pytest.importorskip("pdfplumber")
from pdfminer.pdftypes import PDFStream
from llm_ingestion_okf import extract as extract_module
stream = PDFStream({"Width": 1, "Height": 1, "BitsPerComponent": 8}, b"\x00" * 4096)
stream.set_objid(1, 0)
calls: list[int] = []
original = stream.get_data
def spy() -> bytes:
calls.append(1)
return original()
stream.get_data = spy # type: ignore[method-assign]
monkey = pytest.MonkeyPatch()
try:
monkey.setattr(assets, "MAX_IMAGE_BYTES", 1024)
with pytest.raises(ExtractionError) as excinfo:
extract_module._check_stream_cost(stream, "unfiltered")
finally:
monkey.undo()
assert excinfo.value.code == "asset_too_large"
assert calls == [], "refused only after the stream was decoded, which is the backstop"