feat(tools): promote the resolved-IP SSRF re-check into a reusable primitive

is_safe_url is only syntactic; the DNS-rebinding re-check (re-resolve each hop,
refuse a private/reserved address) was left to each consumer's fetcher — so every
consumer re-derived it, and each independently broke behind a fake-IP proxy.

Add rlm_kit.tools.resolved_host_is_safe(host, port, *, allow_nets=()) + parse_cidrs:
the resolved-address guard now ships once, with an allow_nets carve-out for a host
behind a fake-IP proxy / split-DNS VPN (Clash/Mihomo/Surge map every public host into
the reserved 198.18.0.0/16, which the strict re-check would refuse — starving the model
of all fetched source). Empty allow_nets = unchanged strictness: is_safe_url still
refuses localhost/metadata regardless of allow_nets.

_hostname_is_blocked_literal_ip and the new _ip_blocked share one _ip_in_blocked_range,
so the block-category list has a single home. Additive public surface (tools __all__);
no schema/contract change. Consumer-driven hardening — surfaced by a downstream direct
fetcher refusing every host behind such a proxy.

166 tests pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: qazbnm456

CHANGELOG.md

@@ -12,6 +12,15 @@ surfaced by dogfooding a real downstream consumer.
 
 ### Added
 
+- **Reusable resolved-IP SSRF guard for the `direct`-fetch pattern** (`rlm_kit.tools.resolved_host_is_safe`
+  + `parse_cidrs`). `is_safe_url` is only syntactic; the DNS-rebinding re-check (re-resolve each hop,
+  refuse a private/reserved address) was left to each consumer's fetcher — and every consumer re-derived
+  it. `resolved_host_is_safe(host, port, *, allow_nets=())` now ships that check ONCE, with an
+  `allow_nets` carve-out (`parse_cidrs(["198.18.0.0/16"])`) for a host behind a fake-IP proxy / split-DNS
+  VPN (Clash/Mihomo/Surge map every public host into the reserved `198.18.0.0/16`, which the strict
+  re-check would refuse — starving the model of all fetched source). Empty `allow_nets` = unchanged
+  strictness (`is_safe_url` still refuses localhost/metadata regardless). Consumer-driven: surfaced by a
+  downstream `direct` fetcher refusing every host behind such a proxy.
 - **`max_output_chars` is now configurable** (`RLMConfig.max_output_chars`, env
   `RLM_MAX_OUTPUT_CHARS`, default `10000` — dspy's own default, so behaviour is
   unchanged). dspy.RLM head+tail-truncates each REPL output to this many CHARACTERS

rlm_kit/tools/__init__.py

@@ -1,13 +1,20 @@
 """Reusable tools that RLM tasks can expose to the model inside the REPL."""
 
-from .fetch import is_safe_url, make_fetch_tool
+from .fetch import (
+    is_safe_url,
+    make_fetch_tool,
+    parse_cidrs,
+    resolved_host_is_safe,
+)
 from .model import ModelToolResult, make_model_tool
 from .search import make_web_search_tool, normalise_search_results
 from .validation import make_schema_validator
 
 __all__ = [
     "make_schema_validator",
     "is_safe_url",
+    "resolved_host_is_safe",
+    "parse_cidrs",
     "make_fetch_tool",
     "make_web_search_tool",
     "normalise_search_results",

rlm_kit/tools/fetch.py

@@ -11,18 +11,26 @@
 ``RLMTask(tools=…)`` — dspy.RLM invokes tools synchronously, so the tool must be sync.
 
 NOTE: a syntactic guard does not stop DNS rebinding (a public hostname that
-resolves to a private address). Full protection requires re-checking the
-*resolved* address at connection time inside your fetcher.
+resolves to a private address). ``resolved_host_is_safe`` is that missing
+resolved-address re-check — call it INSIDE your fetcher at connection time (and
+on every redirect hop). Its ``allow_nets`` carve-out (build with ``parse_cidrs``)
+accommodates a fake-IP proxy / split-DNS VPN that maps public hosts into a
+reserved range; it is a re-usable primitive so each consumer's ``direct`` fetcher
+shares ONE correct implementation instead of re-deriving it.
 """
 
 from __future__ import annotations
 
 import ipaddress
+import logging
+import socket
 from typing import Callable
 from urllib.parse import urlparse
 
 from ..trace import record_tool_call
 
+logger = logging.getLogger(__name__)
+
 ALLOWED_SCHEMES = frozenset({"http", "https"})
 
 # Hostnames that must never be fetched regardless of resolution.
@@ -34,13 +42,9 @@
 )
 
 
-def _hostname_is_blocked_literal_ip(hostname: str) -> bool:
-    try:
-        ip = ipaddress.ip_address(hostname)
-    except ValueError:
-        return False
-    # Block loopback, private, link-local (incl. 169.254.169.254 cloud metadata),
-    # reserved, and unspecified ranges.
+def _ip_in_blocked_range(ip) -> bool:
+    """True if a parsed ``ip_address`` falls in a loopback / private / link-local (incl.
+    169.254.169.254 cloud metadata) / reserved / unspecified / multicast range."""
     return (
         ip.is_private
         or ip.is_loopback
@@ -51,6 +55,57 @@ def _hostname_is_blocked_literal_ip(hostname: str) -> bool:
     )
 
 
+def _hostname_is_blocked_literal_ip(hostname: str) -> bool:
+    try:
+        ip = ipaddress.ip_address(hostname)
+    except ValueError:
+        return False
+    return _ip_in_blocked_range(ip)
+
+
+def parse_cidrs(cidrs) -> tuple:
+    """Parse an iterable of CIDR strings into networks for ``resolved_host_is_safe(allow_nets=…)``.
+    An unparseable entry is warned-and-skipped (never fatal) so a typo can't sink a run."""
+    nets = []
+    for c in cidrs or ():
+        try:
+            nets.append(ipaddress.ip_network(c, strict=False))
+        except ValueError:
+            logger.warning("ignoring invalid allow-CIDR %r", c)
+    return tuple(nets)
+
+
+def _ip_blocked(ip_str: str, allow_nets=()) -> bool:
+    """True if ``ip_str`` must be refused. Unparseable → blocked (fail closed). An operator-listed
+    ``allow_nets`` range is treated as external (the proxy, not the resolved address, is the real
+    endpoint) — everything else falls through to the standard blocked-range check."""
+    try:
+        ip = ipaddress.ip_address(ip_str)
+    except ValueError:
+        return True
+    if any(ip in n for n in allow_nets):
+        return False
+    return _ip_in_blocked_range(ip)
+
+
+def resolved_host_is_safe(host: str, port: int, *, allow_nets=()) -> bool:
+    """The DNS-rebinding defence: resolve ``host`` and return True only if EVERY resolved address is
+    external. Call this INSIDE your fetcher, re-checking each redirect hop — ``is_safe_url`` is
+    syntactic and cannot see what a hostname resolves to.
+
+    ``allow_nets`` (from ``parse_cidrs``) carves out operator-trusted ranges: a fake-IP proxy /
+    split-DNS VPN (e.g. Clash/Mihomo/Surge default ``198.18.0.0/16``) maps every public host into a
+    RESERVED range that would otherwise be refused, starving the model of source. Empty by default =
+    full strictness (``is_safe_url`` still refuses localhost/metadata regardless of ``allow_nets``)."""
+    if not host:
+        return False
+    try:
+        infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
+    except OSError:
+        return False
+    return bool(infos) and all(not _ip_blocked(info[4][0], allow_nets) for info in infos)
+
+
 def is_safe_url(url: str) -> bool:
     """Return True if ``url`` is an http(s) URL not pointing at an obvious
     internal/loopback/metadata target.

tests/test_tools.py

@@ -4,7 +4,14 @@
 from pydantic import BaseModel
 
 from rlm_kit.optimize import exact_field_metric, schema_valid_metric
-from rlm_kit.tools.fetch import is_safe_url, make_fetch_tool
+import rlm_kit.tools.fetch as fetch_mod
+from rlm_kit.tools.fetch import (
+    _ip_blocked,
+    is_safe_url,
+    make_fetch_tool,
+    parse_cidrs,
+    resolved_host_is_safe,
+)
 from rlm_kit.tools.model import ModelToolResult, make_model_tool
 from rlm_kit.tools.search import make_web_search_tool, normalise_search_results
 from rlm_kit.tools.validation import make_schema_validator
@@ -171,6 +178,46 @@ def test_unsafe_urls_blocked(url):
     assert is_safe_url(url) is False
 
 
+# ---- resolved-IP re-check (DNS-rebinding defence) + allow-list -----------
+
+def test_parse_cidrs_skips_bad_entries():
+    nets = parse_cidrs(("198.18.0.0/16", "garbage", "10.0.0.0/8"))
+    assert len(nets) == 2                                    # the bad entry is dropped, not fatal
+    assert parse_cidrs(()) == () and parse_cidrs(None) == ()
+
+
+def test_ip_blocked_default_allow_list_and_fail_closed():
+    assert _ip_blocked("8.8.8.8") is False                  # public → allowed
+    assert _ip_blocked("198.18.2.128") is True              # reserved (benchmarking) → blocked
+    assert _ip_blocked("not-an-ip") is True                 # unparseable → fail closed
+    nets = parse_cidrs(("198.18.0.0/16",))
+    assert _ip_blocked("198.18.2.128", nets) is False       # allow-listed proxy range → external
+    assert _ip_blocked("127.0.0.1", nets) is True           # loopback still blocked
+    assert _ip_blocked("169.254.169.254", nets) is True     # cloud metadata still blocked
+
+
+def test_resolved_host_is_safe_honors_allow_list(monkeypatch):
+    # A fake-IP proxy resolves every host into 198.18.x.x; refused by default, allowed when listed.
+    monkeypatch.setattr(fetch_mod.socket, "getaddrinfo",
+                        lambda host, port, *a, **k: [(2, 1, 6, "", ("198.18.2.128", port))])
+    assert resolved_host_is_safe("plugins.svn.example", 443) is False
+    assert resolved_host_is_safe("plugins.svn.example", 443,
+                                 allow_nets=parse_cidrs(("198.18.0.0/16",))) is True
+    # a host that resolves to a REAL public address is safe without any allow-list
+    monkeypatch.setattr(fetch_mod.socket, "getaddrinfo",
+                        lambda host, port, *a, **k: [(2, 1, 6, "", ("93.184.216.34", port))])
+    assert resolved_host_is_safe("example.com", 443) is True
+
+
+def test_resolved_host_is_safe_fails_closed_on_resolution_error(monkeypatch):
+    def boom(*a, **k):
+        raise OSError("no DNS")
+
+    monkeypatch.setattr(fetch_mod.socket, "getaddrinfo", boom)
+    assert resolved_host_is_safe("x", 443, allow_nets=parse_cidrs(("198.18.0.0/16",))) is False
+    assert resolved_host_is_safe("", 443) is False          # empty host → refused
+
+
 def test_fetch_tool_blocks_before_calling_fetcher():
     called = {"n": 0}