diff --git a/graphify/security.py b/graphify/security.py index 0d906013..42c08fd1 100644 --- a/graphify/security.py +++ b/graphify/security.py @@ -1,6 +1,7 @@ # Security helpers - URL validation, safe fetch, path guards, label sanitisation from __future__ import annotations +import contextlib import html import re import urllib.error @@ -58,12 +59,45 @@ def validate_url(url: str) -> str: f"Blocked private/internal IP {addr} (resolved from '{hostname}'). " f"Got: {url!r}" ) - except socket.gaierror: - pass # DNS failure will surface later during fetch + except socket.gaierror as exc: + raise ValueError( + f"DNS resolution failed for '{hostname}': {exc}. Got: {url!r}" + ) from exc return url +@contextlib.contextmanager +def _ssrf_guarded_socket(): + """Patch socket.getaddrinfo for the duration of a fetch to catch DNS rebinding. + + Validates every IP that urllib resolves so a DNS server cannot return a public IP + for validate_url and swap to a private IP for the actual connection (TOCTOU fix). + Not thread-safe, but graphify is a single-threaded CLI tool. + """ + original = socket.getaddrinfo + + def _guarded(host, port, *args, **kwargs): + results = original(host, port, *args, **kwargs) + for info in results: + addr = info[4][0] + try: + ip = ipaddress.ip_address(addr) + except ValueError: + continue + if ip.is_private or ip.is_reserved or ip.is_loopback or ip.is_link_local: + raise OSError( + f"SSRF blocked: IP {addr} resolved from '{host}' is private/reserved" + ) + return results + + socket.getaddrinfo = _guarded + try: + yield + finally: + socket.getaddrinfo = original + + class _NoFileRedirectHandler(urllib.request.HTTPRedirectHandler): """Redirect handler that re-validates every redirect target. @@ -104,7 +138,7 @@ def safe_fetch(url: str, max_bytes: int = _MAX_FETCH_BYTES, timeout: int = 30) - opener = _build_opener() req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 graphify/1.0"}) - with opener.open(req, timeout=timeout) as resp: + with _ssrf_guarded_socket(), opener.open(req, timeout=timeout) as resp: # urllib raises HTTPError for non-2xx when using urlopen directly; # with a custom opener we check manually to be safe. status = getattr(resp, "status", None) or getattr(resp, "code", None) diff --git a/graphify/transcribe.py b/graphify/transcribe.py index 70000757..701fdb4c 100644 --- a/graphify/transcribe.py +++ b/graphify/transcribe.py @@ -51,6 +51,8 @@ def download_audio(url: str, output_dir: Path) -> Path: Returns the path to the downloaded audio file (.m4a or .opus). Uses cached file if already downloaded. """ + from graphify.security import validate_url + validate_url(url) # blocks private IPs, bad schemes before yt-dlp runs yt_dlp = _get_yt_dlp() output_dir.mkdir(parents=True, exist_ok=True) diff --git a/pyproject.toml b/pyproject.toml index d270cfdb..a63559b9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "graphifyy" -version = "0.5.3" +version = "0.5.4" description = "AI coding assistant skill (Claude Code, Codex, OpenCode, Cursor, Gemini CLI, Aider, OpenClaw, Factory Droid, Trae, Hermes, Kiro, Google Antigravity) - turn any folder of code, docs, papers, images, or videos into a queryable knowledge graph" readme = "README.md" license = { file = "LICENSE" } diff --git a/tests/test_cache.py b/tests/test_cache.py index fd57cad1..c3f19dd6 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -67,11 +67,13 @@ def test_cached_files(tmp_path, cache_root): def test_clear_cache(tmp_file, cache_root): - """clear_cache removes all .json files from graphify-out/cache/.""" + """clear_cache removes all .json files from graphify-out/cache/ (all subdirs).""" save_cached(tmp_file, {"nodes": [], "edges": []}, root=cache_root) - assert len(list((cache_root / "graphify-out" / "cache").glob("*.json"))) > 0 + # Since v0.5.3 entries go into cache/ast/, not the flat cache/ dir + cache_base = cache_root / "graphify-out" / "cache" + assert len(list(cache_base.rglob("*.json"))) > 0 clear_cache(cache_root) - assert len(list((cache_root / "graphify-out" / "cache").glob("*.json"))) == 0 + assert len(list(cache_base.rglob("*.json"))) == 0 def test_md_frontmatter_only_change_same_hash(tmp_path):