beyhan/graphify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tighten F2 and F3 from review: bounded decompression, ollama DNS + clean error

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Safi
2026-06-02 22:50:33 +01:00
co-authored by Claude Opus 4.8
parent 46a1d4cabb
commit 763b673350
5 changed files with 130 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
graphify/__main__.py
+11
View File
@@ -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).
graphify/detect.py
+24 -9
View File
@@ -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
graphify/llm.py
+46 -17
View File
@@ -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}. "
tests/test_office_limits.py
+27
View File
@@ -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"<types/>")
zf.writestr("xl/workbook.xml", b"<workbook/>" * 100)
zf.writestr("xl/worksheets/sheet1.xml", b"<sheetData>rows</sheetData>" * 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"
tests/test_ollama.py
+22
View File
@@ -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