Fix SSRF DNS rebinding TOCTOU and yt-dlp URL bypass (#591, #592)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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)

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)
 

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" }

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):