From 763b673350aaa7f4c32d3c976c7365e7f003b759 Mon Sep 17 00:00:00 2001 From: Safi Date: Tue, 2 Jun 2026 22:50:33 +0100 Subject: [PATCH] tighten F2 and F3 from review: bounded decompression, ollama DNS + clean error F2: replace the header-trust ratio check with an authoritative bounded streaming-decompression pass. The zip central-directory sizes are attacker-controlled, so a member that under-declares its size could dodge the declared-size checks; now every member is stream-decompressed with a hard byte ceiling, so actual expansion past the cap is caught regardless of the headers. The cheap declared-size pre-filter stays as a fast reject for honest bombs. F3: _validate_ollama_base_url now resolves the host, so an alias that points at a link-local/metadata IP is blocked too, not just literal IPs. The extract command gains an early gate that turns the metadata block into a clean 'error: ...' exit(2) instead of a deep traceback; a warn toggle keeps the single user-facing LAN warning in the in-flow call. Co-Authored-By: Claude Opus 4.8 --- graphify/__main__.py | 11 +++++++ graphify/detect.py | 33 +++++++++++++------ graphify/llm.py | 63 +++++++++++++++++++++++++++---------- tests/test_office_limits.py | 27 ++++++++++++++++ tests/test_ollama.py | 22 +++++++++++++ 5 files changed, 130 insertions(+), 26 deletions(-) diff --git a/graphify/__main__.py b/graphify/__main__.py index 543779d..12e5b9b 100644 --- a/graphify/__main__.py +++ b/graphify/__main__.py @@ -3769,6 +3769,17 @@ def main() -> None: file=sys.stderr, ) sys.exit(1) + if backend == "ollama": + # Fail closed with a clean message (not a deep traceback) if + # OLLAMA_BASE_URL points at a link-local/metadata address. warn=False: + # the later in-flow call owns the user-facing warning for LAN hosts. + from graphify.llm import _validate_ollama_base_url + _oll_url = os.environ.get("OLLAMA_BASE_URL", _BACKENDS["ollama"].get("base_url", "")) + try: + _validate_ollama_base_url(_oll_url, warn=False) + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + sys.exit(2) if not _get_backend_api_key(backend): # Ollama on a loopback URL ignores auth entirely; don't block # the run just because OLLAMA_API_KEY is unset (issue #792). diff --git a/graphify/detect.py b/graphify/detect.py index 96bde4c..f770a3a 100644 --- a/graphify/detect.py +++ b/graphify/detect.py @@ -55,10 +55,15 @@ def _file_within_size_cap(path: Path, cap: int = _OFFICE_MAX_RAW_BYTES) -> bool: def _zip_within_caps(path: Path) -> bool: - """Reject a zip-based office file that looks like a zip/XML bomb. + """Reject a zip-based office file that is a likely zip/XML bomb. - Checks on-disk size, the summed uncompressed size of every member, and the - overall compression ratio before openpyxl/python-docx decompress and parse. + Two layers, because the zip central-directory sizes are attacker-controlled: + 1. A cheap pre-filter on the declared sizes (on-disk cap, summed-uncompressed + cap, compression ratio) that rejects an honest bomb without decompressing. + 2. An authoritative pass that stream-decompresses every member with a hard + byte ceiling, so a member that under-declares its size in the central + directory cannot expand past the cap undetected. Decompression is chunked + and bounded, so checking a bomb never materializes more than the ceiling. """ import zipfile if not _file_within_size_cap(path): @@ -67,12 +72,22 @@ def _zip_within_caps(path: Path) -> bool: with zipfile.ZipFile(path) as zf: infos = zf.infolist() compressed = sum(i.compress_size for i in infos) or 1 - uncompressed = sum(i.file_size for i in infos) - except (zipfile.BadZipFile, OSError): - return False - if uncompressed > _OFFICE_MAX_DECOMPRESSED_BYTES: - return False - if uncompressed / compressed > _OFFICE_MAX_COMPRESSION_RATIO: + declared = sum(i.file_size for i in infos) + if declared > _OFFICE_MAX_DECOMPRESSED_BYTES: + return False + if declared / compressed > _OFFICE_MAX_COMPRESSION_RATIO: + return False + total = 0 + for info in infos: + with zf.open(info) as member: + while True: + chunk = member.read(1024 * 1024) + if not chunk: + break + total += len(chunk) + if total > _OFFICE_MAX_DECOMPRESSED_BYTES: + return False + except (zipfile.BadZipFile, OSError, EOFError): return False return True diff --git a/graphify/llm.py b/graphify/llm.py index f9a0cd4..20f3ff1 100644 --- a/graphify/llm.py +++ b/graphify/llm.py @@ -1250,42 +1250,71 @@ def estimate_cost(backend: str, input_tokens: int, output_tokens: int) -> float: return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000 -def _validate_ollama_base_url(url: str) -> None: +def _ollama_host_is_link_local_or_metadata(host: str) -> bool: + """True if *host* is, or resolves to, a link-local / cloud-metadata address. + + Resolves the name so an alias pointing at 169.254.169.254 is caught too, not + just a literal IP. General private/LAN addresses are deliberately NOT treated + as metadata: people do run Ollama on trusted LAN boxes, so those only warn. + """ + import ipaddress + import socket + if host in ("metadata.google.internal", "metadata.google.com", "0.0.0.0", "::", "[::]"): # nosec B104 - blocklist, not a bind + return True + if host.startswith("169.254."): # link-local literal, includes the metadata IP + return True + try: + infos = socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM) + except (socket.gaierror, UnicodeError, OSError): + return False + for info in infos: + try: + ip = ipaddress.ip_address(info[4][0]) + except ValueError: + continue + if ip.is_link_local: # 169.254.0.0/16 and fe80::/10 (includes the metadata IP) + return True + return False + + +def _validate_ollama_base_url(url: str, *, warn: bool = True) -> None: """Warn if OLLAMA_BASE_URL looks unsafe; hard-block link-local/metadata (F3). Sending an entire corpus to a non-loopback http:// endpoint silently leaks proprietary code, but some users genuinely run Ollama on a LAN host they trust, so a general non-loopback target only warns. A link-local or cloud - metadata address (169.254.x, metadata.google.*) is never a legitimate Ollama - host and is a classic SSRF target, so we fail closed with a ValueError there. + metadata address (169.254.x, metadata.google.*, or any host that resolves to + one) is never a legitimate Ollama host and is a classic SSRF target, so we + fail closed with a ValueError there regardless of *warn*. Pass warn=False for + an early gate that should hard-block but leave the user-facing warning to the + later in-flow call. """ try: from urllib.parse import urlparse parsed = urlparse(url) except Exception: - print( - f"[graphify] WARNING: OLLAMA_BASE_URL={url!r} is not a parseable URL.", - file=sys.stderr, - ) + if warn: + print( + f"[graphify] WARNING: OLLAMA_BASE_URL={url!r} is not a parseable URL.", + file=sys.stderr, + ) return if parsed.scheme not in ("http", "https"): - print( - f"[graphify] WARNING: OLLAMA_BASE_URL has unexpected scheme {parsed.scheme!r}; " - "expected http or https.", - file=sys.stderr, - ) + if warn: + print( + f"[graphify] WARNING: OLLAMA_BASE_URL has unexpected scheme {parsed.scheme!r}; " + "expected http or https.", + file=sys.stderr, + ) return host = (parsed.hostname or "").lower() - if ( - host.startswith("169.254.") # link-local, includes the 169.254.169.254 metadata IP - or host in ("metadata.google.internal", "metadata.google.com", "0.0.0.0", "::", "[::]") # nosec B104 - blocklist, not a bind - ): + if _ollama_host_is_link_local_or_metadata(host): raise ValueError( f"OLLAMA_BASE_URL points at a link-local/metadata address ({host!r}); refusing to " "send the corpus there. Set it to a real Ollama host." ) is_loopback = host in ("localhost", "127.0.0.1", "::1") or host.startswith("127.") - if not is_loopback: + if warn and not is_loopback: scheme_note = " (UNENCRYPTED)" if parsed.scheme == "http" else "" print( f"[graphify] WARNING: OLLAMA_BASE_URL points to non-loopback host {host!r}{scheme_note}. " diff --git a/tests/test_office_limits.py b/tests/test_office_limits.py index a1e2b42..d4f70c9 100644 --- a/tests/test_office_limits.py +++ b/tests/test_office_limits.py @@ -51,6 +51,33 @@ def test_converters_return_empty_for_bomb(tmp_path): assert detect.xlsx_to_markdown(bomb) == "" +def test_legit_multi_member_passes_streaming(tmp_path): + """A normal multi-member office zip passes the streaming-ceiling pass.""" + ok = tmp_path / "ok.xlsx" + with zipfile.ZipFile(ok, "w", zipfile.ZIP_DEFLATED) as zf: + zf.writestr("[Content_Types].xml", b"") + zf.writestr("xl/workbook.xml", b"" * 100) + zf.writestr("xl/worksheets/sheet1.xml", b"rows" * 500) + assert detect._zip_within_caps(ok) is True + + +def test_streaming_ceiling_rejects_oversized_actual(tmp_path, monkeypatch): + """With a low decompressed cap, content whose actual bytes exceed it is rejected. + + This exercises the authoritative bounded-decompression pass: the function + reads real decompressed bytes (not the attacker-declared central-directory + sizes) and stops once the ceiling is crossed. + """ + monkeypatch.setattr(detect, "_OFFICE_MAX_DECOMPRESSED_BYTES", 64 * 1024) # 64 KiB + f = tmp_path / "big.xlsx" + # ~512 KiB of incompressible data: low ratio (passes the ratio pre-filter), + # but real decompressed size far exceeds the 64 KiB ceiling. + import os as _os + with zipfile.ZipFile(f, "w", zipfile.ZIP_DEFLATED) as zf: + zf.writestr("xl/x.xml", _os.urandom(512 * 1024)) + assert detect._zip_within_caps(f) is False + + def test_pdf_over_cap_returns_empty(tmp_path, monkeypatch): """A PDF larger than the raw cap is skipped before pypdf opens it.""" big = tmp_path / "big.pdf" diff --git a/tests/test_ollama.py b/tests/test_ollama.py index 4991557..c90d610 100644 --- a/tests/test_ollama.py +++ b/tests/test_ollama.py @@ -26,6 +26,28 @@ def test_ollama_loopback_and_lan_do_not_raise(capsys): assert "non-loopback" in capsys.readouterr().err +def test_ollama_alias_resolving_to_link_local_blocked(monkeypatch): + """A hostname that RESOLVES to a link-local IP is blocked, not just literals (F3).""" + from graphify import llm + + def fake_getaddrinfo(host, *a, **k): + return [(2, 1, 6, "", ("169.254.169.254", 0))] # alias -> metadata IP + + monkeypatch.setattr("socket.getaddrinfo", fake_getaddrinfo) + with pytest.raises(ValueError): + llm._validate_ollama_base_url("http://innocent-looking-host/v1") + + +def test_ollama_warn_false_still_hard_blocks_but_stays_quiet(capsys): + """warn=False suppresses the LAN warning but never the metadata hard-block (F3).""" + # LAN host with warn=False: allowed, and no warning emitted (early-gate use). + _validate_ollama_base_url("http://192.168.1.50:11434/v1", warn=False) + assert capsys.readouterr().err == "" + # metadata host with warn=False: still raises. + with pytest.raises(ValueError): + _validate_ollama_base_url("http://169.254.169.254/v1", warn=False) + + def test_ollama_in_backends(): assert "ollama" in BACKENDS assert BACKENDS["ollama"]["pricing"]["input"] == 0.0