open/llm-ingestion-pipeline-security
1
0
Fork 0
Code Issues Pull requests Releases Activity

feat(inbox): .zip container threats — zip-slip, zip-bomb, symlink (stage 2b)

Browse source 
The front-end reads zip entries in memory (never extracts to disk), so it owns
the container caps while a traversal entry maps onto the guard's path gate:

- zip-slip: a '../../evil.md' entry materializes onto a traversal concept path
  (preserved verbatim, not normalized) -> stage-2 T4 -> FAIL_SECURE -> REJECT.
- zip-bomb: per-entry + per-archive uncompressed-size caps (OWASP LLM10) refuse
  an oversize entry before its bytes are read; a bounded read defends a lying
  header. Detach-proof: a generous cap admits the same archive, so the cap is
  load-bearing.
- symlink entry: refused at the front-end (no legitimate concept meaning).

Caps are kwargs on extract_inbox/receive (small in tests, generous by default).
Tests 288 -> 293.
This commit is contained in:
Kjell Tore Guttormsen 2026-07-06 11:07:46 +02:00
commit 52aa40b17a
2 changed files with 156 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

96
tests/inbox_frontend.py
View file

@ -17,6 +17,8 @@ Slice 2a covers the text formats — ``.txt`` and ``.md`` (stdlib only). ``.zip`
"""
from __future__ import annotations
import stat
import zipfile
from dataclasses import dataclass
from pathlib import Path
@ -39,6 +41,12 @@ _MATERIALIZE_PREFIX = "uploads"
# The text formats this slice reads directly (no parser dependency).
_TEXT_SUFFIXES = {".txt", ".md"}
# Zip self-safety caps (OWASP LLM10). Bounded so a decompression bomb is refused
# before its uncompressed bytes are read into memory. Defaults are generous for a
# document inbox; tests pass small caps to exercise the gate.
MAX_ENTRY_BYTES = 5_000_000 # per uncompressed entry
MAX_TOTAL_BYTES = 25_000_000 # per archive, summed across entries
@dataclass(frozen=True)
class Provenance:
@ -63,41 +71,97 @@ class InboxExtract:
rejected: tuple
def _materialize_path(source_name: str) -> str:
"""Assign an OKF concept path to a dropped file: ``<prefix>/<stem>.md``."""
stem = Path(source_name).stem
return f"{_MATERIALIZE_PREFIX}/{stem}.md"
def _materialize_path(rel_name: str) -> str:
"""Assign an OKF concept path to a dropped file: ``<prefix>/<rel>.md``.
The relative name is preserved verbatim, including any ``..`` — a zip-slip
entry (``../../evil.md``) thus lands on a traversal concept path that the
stage-2 path gate (T4) rejects, rather than being silently normalized away.
"""
rel = Path(rel_name).with_suffix(".md").as_posix()
return f"{_MATERIALIZE_PREFIX}/{rel}"
def extract_inbox(paths) -> InboxExtract:
def _is_symlink_entry(info: zipfile.ZipInfo) -> bool:
"""True if a zip entry encodes a Unix symlink (mode bits in external_attr)."""
return stat.S_ISLNK(info.external_attr >> 16)
def _extract_zip(path, bundle, provenance, rejected, max_entry_bytes, max_total_bytes):
"""Read a ``.zip`` in memory, materializing its text entries; refuse bombs,
symlinks and oversize entries at the front-end (container threats)."""
total = 0
with zipfile.ZipFile(path) as zf:
for info in zf.infolist():
name = info.filename
if name.endswith("/"):
continue # directory entry — no content
if _is_symlink_entry(info):
rejected.append((name, "symlink entry refused (container threat)"))
continue
# Fast reject on the declared uncompressed size (a bomb, before reading).
if info.file_size > max_entry_bytes:
rejected.append((name, f"entry exceeds {max_entry_bytes}-byte cap (declared {info.file_size})"))
continue
if total + info.file_size > max_total_bytes:
rejected.append((name, f"archive exceeds {max_total_bytes}-byte total cap"))
continue
if Path(name).suffix.lower() not in _TEXT_SUFFIXES:
continue # only text concepts are materialized in this slice
# Bounded read defends against a header that lies about file_size.
with zf.open(info) as f:
data = f.read(max_entry_bytes + 1)
if len(data) > max_entry_bytes:
rejected.append((name, f"entry expands past {max_entry_bytes}-byte cap on read"))
continue
total += len(data)
concept_path = _materialize_path(name)
bundle[concept_path] = data.decode("utf-8", errors="replace")
provenance.append(Provenance(concept_path, name, "zip"))
def extract_inbox(
paths,
*,
max_entry_bytes: int = MAX_ENTRY_BYTES,
max_total_bytes: int = MAX_TOTAL_BYTES,
) -> InboxExtract:
"""Read dropped files and materialize them into an OKF bundle + provenance.
``paths`` is an iterable of file paths. Each ``.txt`` / ``.md`` becomes one
concept; a ``.md`` keeps its OKF frontmatter verbatim, a ``.txt`` becomes a
body-only concept. Non-text suffixes are not handled in this slice.
concept (a ``.md`` keeps its OKF frontmatter verbatim). A ``.zip`` is read in
memory and its text entries materialized, with bomb/symlink/oversize entries
refused into ``InboxExtract.rejected``. Other suffixes are not handled yet.
"""
bundle: dict = {}
provenance: list = []
rejected: list = []
for path in paths:
path = Path(path)
suffix = path.suffix.lower()
if suffix not in _TEXT_SUFFIXES:
if suffix == ".zip":
_extract_zip(path, bundle, provenance, rejected, max_entry_bytes, max_total_bytes)
elif suffix in _TEXT_SUFFIXES:
concept_path = _materialize_path(path.name)
bundle[concept_path] = path.read_text(encoding="utf-8")
provenance.append(Provenance(concept_path, path.name, suffix.lstrip(".")))
else:
raise ValueError(f"unsupported upload format in this slice: {path.name!r}")
concept_path = _materialize_path(path.name)
bundle[concept_path] = path.read_text(encoding="utf-8")
provenance.append(
Provenance(concept_path, path.name, suffix.lstrip("."))
)
return InboxExtract(bundle, tuple(provenance), ())
return InboxExtract(bundle, tuple(provenance), tuple(rejected))
def receive(paths) -> tuple[InboxExtract, BundleResult, str]:
def receive(
paths,
*,
max_entry_bytes: int = MAX_ENTRY_BYTES,
max_total_bytes: int = MAX_TOTAL_BYTES,
) -> tuple[InboxExtract, BundleResult, str]:
"""The full two-stage inbox: extract & materialize, then guard, then verdict.
Returns ``(extracted, guard_result, verdict)``. A front-end refusal forces a
REJECT regardless of the guard's aggregate — the guard never saw that drop.
"""
extracted = extract_inbox(paths)
extracted = extract_inbox(paths, max_entry_bytes=max_entry_bytes, max_total_bytes=max_total_bytes)
result = import_bundle(extracted.bundle, origin=Origin.EXTERNAL, channel=Channel.AUTOMATIC)
verdict = "REJECT" if extracted.rejected else _VERDICT[result.disposition]
return extracted, result, verdict

76
tests/test_okf_inbox_uploads.py
View file

@ -14,6 +14,9 @@ showcase/dev-scoped (PLAN §247), and the core package stays stdlib-only
"""
from __future__ import annotations
import stat
import zipfile
from inbox_frontend import receive, extract_inbox, InboxExtract
from llm_ingestion_guard.disposition import Disposition
@ -76,6 +79,77 @@ def test_detach_proof_extraction_carries_the_payload(tmp_path, monkeypatch):
p = _write(tmp_path, "notes.txt", "Some notes.\n" + _INJECTION + "\n")
import inbox_frontend as fe
monkeypatch.setattr(fe, "extract_inbox", lambda paths: InboxExtract({}, (), ()))
monkeypatch.setattr(fe, "extract_inbox", lambda paths, **kw: InboxExtract({}, (), ()))
_extracted, _result, verdict = fe.receive([p])
assert verdict == "ADMIT"
# --- slice 2b: .zip container threats ---------------------------------------
# The front-end reads zip entries in memory (never extracts to disk), so an
# on-disk zip-slip / symlink escape cannot happen. It owns the container caps
# (bomb / symlink); a traversal entry name becomes a concept path the guard's
# T4 gate rejects. Zips are built in the test so the crafted entries are explicit.
def _make_zip(tmp_path, entries, name="drop.zip"):
"""Build a zip. Each entry is (name, data) or (name, data, external_attr)."""
zp = tmp_path / name
with zipfile.ZipFile(zp, "w") as zf:
for entry in entries:
if len(entry) == 3:
ename, data, attr = entry
info = zipfile.ZipInfo(ename)
info.external_attr = attr
zf.writestr(info, data)
else:
ename, data = entry
zf.writestr(ename, data)
return zp
def test_zip_clean_entries_admit(tmp_path):
zp = _make_zip(tmp_path, [
("a.md", "---\ntype: t\n---\nA clean concept.\n"),
("docs/b.txt", "A clean note."),
])
extracted, _result, verdict = receive([zp])
assert set(extracted.bundle) == {"uploads/a.md", "uploads/docs/b.md"}
assert verdict == "ADMIT"
assert all(pr.source_type == "zip" for pr in extracted.provenance)
def test_zip_slip_entry_is_rejected_by_the_path_gate(tmp_path):
zp = _make_zip(tmp_path, [("../../evil.md", "---\ntype: t\n---\npayload\n")])
_extracted, result, verdict = receive([zp])
by_path = {c.path: c for c in result.concepts}
slip = "uploads/../../evil.md"
assert slip in by_path # traversal preserved
assert by_path[slip].error is not None # T4 rejected it
assert by_path[slip].disposition is Disposition.FAIL_SECURE
assert verdict == "REJECT"
def test_zip_bomb_is_refused_by_the_size_cap(tmp_path):
zp = _make_zip(tmp_path, [("big.txt", "A" * 5000)])
extracted, _result, verdict = receive([zp], max_entry_bytes=1024, max_total_bytes=1024)
assert extracted.bundle == {} # never read into the bundle
assert any("big.txt" in n for n, _reason in extracted.rejected)
assert verdict == "REJECT"
def test_zip_bomb_detach_proof(tmp_path):
# The same archive under a generous cap is NOT refused -> the cap is what
# rejected it above, not the archive shape.
zp = _make_zip(tmp_path, [("big.txt", "A" * 5000)])
extracted, _result, _verdict = receive([zp], max_entry_bytes=10_000, max_total_bytes=10_000)
assert extracted.rejected == ()
assert "uploads/big.md" in extracted.bundle
def test_zip_symlink_entry_is_refused(tmp_path):
attr = (stat.S_IFLNK | 0o777) << 16
zp = _make_zip(tmp_path, [("link.md", "/etc/passwd", attr)])
extracted, _result, verdict = receive([zp])
assert any("link.md" in n for n, _reason in extracted.rejected)
assert "uploads/link.md" not in extracted.bundle
assert verdict == "REJECT"