feat(guard): runtime DNS SSRF bypass detection — WRD-RES-EXFIL-DNS-SSRF (#11)

.github/workflows/integrity-gate.yml

@@ -168,7 +168,7 @@ jobs:
             --lock e2e.warden.lock \
             --verify \
             --certificate-identity \
-              "https://github.com/ernestprovo23/mcp-warden/.github/workflows/integrity-gate.yml@${GITHUB_REF}" \
+              "https://github.com/DataScience-EngineeringExperts/mcp-warden/.github/workflows/integrity-gate.yml@${GITHUB_REF}" \
             --certificate-oidc-issuer "https://token.actions.githubusercontent.com"
 
       # NEGATIVE PROOF (real crypto): tamper the signed lock's overall_digest and
@@ -185,7 +185,7 @@ jobs:
                --lock e2e.warden.lock \
                --verify \
                --certificate-identity \
-                 "https://github.com/ernestprovo23/mcp-warden/.github/workflows/integrity-gate.yml@${GITHUB_REF}" \
+                 "https://github.com/DataScience-EngineeringExperts/mcp-warden/.github/workflows/integrity-gate.yml@${GITHUB_REF}" \
                --certificate-oidc-issuer "https://token.actions.githubusercontent.com" ; then
             echo "ERROR: real sigstore verified a TAMPERED overall_digest — fail closed is broken"
             exit 1

CHANGELOG.md

@@ -22,15 +22,25 @@ CI. The v0.3 `guard` proxy adds deterministic runtime *result* inspection
 **Explicitly out of scope in v1 (documented post-1.0 roadmap):**
 
 - **HTTP/SSE transport** — v1 is stdio-only; HTTP/SSE is the headline v1.x item (#9).
-- **DNS-name resolution** of exfil-domain matches (raw-IP-literal handling is the D6
-  work item) and **prompt-injection default-block** (stays opt-in / MONITOR until
-  field false-positive data justifies blocking by default).
+- **Prompt-injection default-block** — stays opt-in / MONITOR until field
+  false-positive data justifies blocking by default.
 - Behavioral-attack defense (`T-BEHAVE`), full agent-firewall mediation, and any
   compliance/regulatory claim. See `docs/THREAT_MODEL.md` for the limits.
 
 ## [Unreleased]
 
-_No unreleased changes yet._
+### Added
+
+- **Runtime DNS resolution SSRF bypass detection (`WRD-RES-EXFIL-DNS-SSRF`)** (#11):
+  the `guard` proxy now resolves URL hostnames from `tools/call` results at runtime
+  and blocks (error-replace) when any resolved IP falls in a deny range
+  (`SSRF_NETWORKS` — link-local, loopback, RFC1918, IPv6 ULA/loopback/link-local).
+  This closes the bypass where `WRD-RES-EXFIL-IP-LITERAL` could not fire because the
+  result contained a DNS hostname (e.g. `169.254.169.254.nip.io`) rather than a raw
+  IP literal. Resolution is bounded by 1 s across all hostnames per result frame,
+  fail-open (any DNS error = no hit), and opt-out via `--no-block-exfil-dns-ssrf`
+  (or `--no-block-deterministic`). Raw IP literals and the offline `inspect` command
+  are unchanged. New module `res_dns.py`; 23 new tests.
 
 ## [1.0.1] — 2026-06-13
 

src/mcp_warden/cli_guard.py

@@ -85,6 +85,7 @@ def guard(
         no_block_exfil_domain: bool = typer.Option(False, "--no-block-exfil-domain", help="Demote WRD-RES-EXFIL-DOMAIN to shadow"),
         allow_exfil_domain: bool = typer.Option(False, "--allow-exfil-domain", help="Alias of --no-block-exfil-domain"),
         no_block_exfil_ip_literal: bool = typer.Option(False, "--no-block-exfil-ip-literal", help="Demote WRD-RES-EXFIL-IP-LITERAL to shadow"),
+        no_block_exfil_dns_ssrf: bool = typer.Option(False, "--no-block-exfil-dns-ssrf", help="Demote WRD-RES-EXFIL-DNS-SSRF to shadow (disables runtime DNS resolution)"),
         no_block_list_changed: bool = typer.Option(False, "--no-block-list-changed", help="Demote tools/list_changed gate to shadow"),
         no_block_policy: bool = typer.Option(False, "--no-block-policy", help="Demote argument-policy deny to shadow"),
         no_block_deterministic: bool = typer.Option(False, "--no-block-deterministic", help="Demote the WHOLE deterministic tier + both gates"),
@@ -177,6 +178,7 @@ def guard(
             no_block_secret_echo=no_block_secret_echo or no_block_deterministic,
             no_block_exfil_domain=no_block_exfil_domain or allow_exfil_domain or no_block_deterministic,
             no_block_exfil_ip_literal=no_block_exfil_ip_literal or no_block_deterministic,
+            no_block_exfil_dns_ssrf=no_block_exfil_dns_ssrf or no_block_deterministic,
             no_block_list_changed=no_block_list_changed or no_block_deterministic,
             no_block_policy=no_block_policy or no_block_deterministic,
             block_inject_phrase=block_inject_phrase,

src/mcp_warden/guard_loop.py

@@ -86,6 +86,7 @@ class GuardConfig:
     no_block_secret_echo: bool = False
     no_block_exfil_domain: bool = False
     no_block_exfil_ip_literal: bool = False
+    no_block_exfil_dns_ssrf: bool = False
     no_block_list_changed: bool = False
     no_block_policy: bool = False
     block_inject_phrase: bool = False
@@ -116,6 +117,7 @@ class GuardConfig:
         "WRD-RES-SECRET-ECHO": "no_block_secret_echo",
         "WRD-RES-EXFIL-DOMAIN": "no_block_exfil_domain",
         "WRD-RES-EXFIL-IP-LITERAL": "no_block_exfil_ip_literal",
+        "WRD-RES-EXFIL-DNS-SSRF": "no_block_exfil_dns_ssrf",
     }
 
     def category_enabled(self, rule_id: str) -> bool:

src/mcp_warden/guard_result.py

@@ -98,6 +98,17 @@ def handle_s2c(state, frame: Frame, mode: str) -> bytes:
             ) from None
         state.emit(_frame_error_note("s2c", rpc_id, f"inspect error: {exc}"))
         return frame.raw
+
+    # WRD-RES-EXFIL-DNS-SSRF: resolve URL hostnames to catch SSRF bypasses
+    # (fail-open — any DNS error produces no hits, never aborts).
+    if state.config.category_enabled("WRD-RES-EXFIL-DNS-SSRF"):
+        try:
+            dns_findings = _dns_ssrf_findings(result, tool)
+            if dns_findings:
+                findings = list(findings) + dns_findings
+        except Exception as exc:  # pragma: no cover
+            state.emit(_frame_error_note("s2c", rpc_id, f"dns-ssrf error: {exc}"))
+
     return _apply_result_findings(state, frame, mode, rpc_id, tool, result, findings, pol)
 
 
@@ -179,6 +190,47 @@ def _tool_name_from_result(obj: dict[str, Any]) -> str:
     return ""
 
 
+def _dns_ssrf_findings(result: dict, tool: str) -> list[ResultFinding]:
+    """Resolve URL hostnames in ``result`` blocks; return WRD-RES-EXFIL-DNS-SSRF findings.
+
+    Fail-open: DNS errors within :func:`~mcp_warden.res_dns.resolve_ssrf_hits`
+    are already swallowed there (return ``[]``). This function raises only if
+    the catalog/extract plumbing itself fails — caught by the caller.
+
+    Args:
+        result: The ``tools/call`` result dict.
+        tool: The tool name (for finding messages).
+
+    Returns:
+        Per-block ``WRD-RES-EXFIL-DNS-SSRF`` findings (empty if no SSRF hits).
+    """
+    from . import res_catalog, res_dns
+
+    blocks, _ = res_catalog.extract_blocks(result)
+    if not blocks:
+        return []
+
+    # Collect unique candidates across all blocks (resolve once, match per block).
+    all_candidates: set[str] = set()
+    for _idx, text in blocks:
+        all_candidates.update(res_dns.extract_dns_candidates(text))
+    if not all_candidates:
+        return []
+
+    hits = res_dns.resolve_ssrf_hits(sorted(all_candidates))
+    if not hits:
+        return []
+
+    hit_map: dict[str, tuple[str, str]] = {h: (ip, lbl) for h, ip, lbl in hits}
+    findings: list[ResultFinding] = []
+    for idx, text in blocks:
+        block_candidates = res_dns.extract_dns_candidates(text)
+        block_hits = [(h, hit_map[h][0], hit_map[h][1]) for h in block_candidates if h in hit_map]
+        if block_hits:
+            findings.extend(res_catalog.inspect_exfil_dns_ssrf(block_hits, tool, idx))
+    return findings
+
+
 def _apply_result_findings(
     state,
     frame: Frame,
@@ -197,7 +249,7 @@ def _apply_result_findings(
     if state.config.block_inject_phrase and not state.config.audit_only:
         block_findings += [f for f in findings if f.tier == TIER_MONITOR]
 
-    error_rules = {"WRD-RES-EXFIL-DOMAIN", "WRD-RES-INJECT-PHRASE"}
+    error_rules = {"WRD-RES-EXFIL-DOMAIN", "WRD-RES-INJECT-PHRASE", "WRD-RES-EXFIL-DNS-SSRF"}
     if not state.config.redact_secret_echo:
         error_rules.add("WRD-RES-SECRET-ECHO")
     redact_rules = {"WRD-RES-ANSI"}

src/mcp_warden/res_catalog.py

@@ -187,6 +187,42 @@ def inspect_inject(text: str, tool: str, idx: int, phrases: tuple[str, ...] | li
     ]
 
 
+def inspect_exfil_dns_ssrf(
+    dns_hits: list[tuple[str, str, str]],
+    tool: str,
+    idx: int,
+) -> list["ResultFinding"]:
+    """WRD-RES-EXFIL-DNS-SSRF: a URL hostname resolved to a private/SSRF IP at runtime.
+
+    Closes the bypass where ``WRD-RES-EXFIL-IP-LITERAL`` would miss a destination
+    whose hostname (not a raw IP literal) resolves to a deny-range IP.
+
+    Args:
+        dns_hits: Pre-resolved ``(host, resolved_ip, range_label)`` tuples from
+            :func:`mcp_warden.res_dns.resolve_ssrf_hits`.
+        tool: The tool name (for the finding message).
+        idx: The content-block index.
+
+    Returns:
+        A single block-tier finding when ``dns_hits`` is non-empty, else ``[]``.
+    """
+    if not dns_hits:
+        return []
+    detail = ", ".join(f"{h} -> {ip} ({label})" for h, ip, label in dns_hits)
+    return [
+        _RF(
+            rule_id="WRD-RES-EXFIL-DNS-SSRF",
+            severity="high",
+            tier=TIER_BLOCK,
+            message=(
+                f"tools/{tool}: result URL hostname(s) resolve to "
+                f"private/SSRF IP(s): {detail}"
+            ),
+            block_index=idx,
+        )
+    ]
+
+
 def uninspectable_note(tool: str, idx: int) -> "ResultFinding":
     """WRD-RES-UNINSPECTABLE (§5.2): a content block could not be inspected."""
     return _RF(

src/mcp_warden/res_dns.py

@@ -0,0 +1,108 @@
+"""DNS-resolution SSRF bypass detection for runtime result inspection.
+
+Resolves extracted hostnames at runtime to catch the class of SSRF bypass
+where a non-literal host resolves to a private/loopback/metadata IP — a gap
+``WRD-RES-EXFIL-IP-LITERAL`` cannot close (that rule only matches raw IP
+literals already present in result text).
+
+Design constraints (POLICY_MODEL.md §5):
+- All DNS IO is isolated here; callers that are pure (``inspect_result``)
+  receive pre-resolved hits, not raw hostnames to resolve themselves.
+- Fail-open: any ``OSError``, timeout, or unexpected error returns no hits for
+  the affected host — the guard continues normally.
+- Raw IP literals are skipped (already handled by ``WRD-RES-EXFIL-IP-LITERAL``).
+- Resolution is bounded by ``timeout`` seconds across ALL hosts combined.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures
+import socket
+from typing import Sequence
+
+from .net_rules import SSRF_NETWORKS, parse_ip
+from .res_net import extract_urls
+
+
+def _resolve_ips(host: str) -> list[str]:
+    """Return string-form IPs from ``getaddrinfo`` for ``host``, or ``[]``.
+
+    Any ``OSError`` (NXDOMAIN, refused, unreachable) silently returns empty.
+
+    Args:
+        host: A DNS hostname (never a raw IP literal).
+
+    Returns:
+        IP address strings for all addresses ``getaddrinfo`` returns.
+    """
+    try:
+        return [info[4][0] for info in socket.getaddrinfo(host, None)]
+    except OSError:
+        return []
+
+
+def extract_dns_candidates(text: str) -> list[str]:
+    """Return unique non-IP-literal hostnames from ``scheme://`` URLs in ``text``.
+
+    Only scheme-qualified URLs (``https://``, ``http://``, etc.) are
+    considered — bare hostname tokens are not resolved (too noisy). Raw IP
+    literals are filtered out (already handled by ``WRD-RES-EXFIL-IP-LITERAL``).
+
+    Args:
+        text: Raw result text.
+
+    Returns:
+        Sorted, de-duplicated list of DNS name candidates.
+    """
+    seen: set[str] = set()
+    for host, _path, _full in extract_urls(text):
+        if parse_ip(host) is None and host not in seen:
+            seen.add(host)
+    return sorted(seen)
+
+
+def resolve_ssrf_hits(
+    hosts: Sequence[str],
+    *,
+    timeout: float = 1.0,
+) -> list[tuple[str, str, str]]:
+    """Resolve ``hosts`` and return those whose IPs fall in ``SSRF_NETWORKS``.
+
+    DNS lookups run in a :class:`concurrent.futures.ThreadPoolExecutor` so the
+    total wall-clock is bounded by ``timeout`` seconds (``wait`` semantics: any
+    host not resolved within the budget is silently skipped — fail-open).
+
+    Args:
+        hosts: DNS name candidates (no raw IP literals; use
+            :func:`extract_dns_candidates` to derive these from result text).
+        timeout: Max seconds to wait for ALL resolutions combined (default 1.0).
+
+    Returns:
+        Sorted ``(host, resolved_ip_str, range_label)`` for every host that
+        resolved to at least one IP inside a deny range. Empty on timeout or
+        error.
+    """
+    if not hosts:
+        return []
+    workers = min(len(hosts), 8)
+    hits: list[tuple[str, str, str]] = []
+    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as exe:
+        futs: dict[concurrent.futures.Future[list[str]], str] = {
+            exe.submit(_resolve_ips, h): h for h in hosts
+        }
+        done, _ = concurrent.futures.wait(futs, timeout=timeout)
+        for fut in done:
+            host = futs[fut]
+            try:
+                ips = fut.result()
+            except Exception:  # pragma: no cover — _resolve_ips already swallows
+                continue
+            for ip_str in ips:
+                ip = parse_ip(ip_str)
+                if ip is None:
+                    continue
+                for net, label in SSRF_NETWORKS:
+                    if ip in net:
+                        hits.append((host, ip_str, label))
+                        break  # first matching range wins per resolved IP
+    return sorted(hits)

src/mcp_warden/result_inspection.py

@@ -32,7 +32,13 @@
 TIER_NOTE = "note"
 
 BLOCK_RULES = frozenset(
-    {"WRD-RES-ANSI", "WRD-RES-SECRET-ECHO", "WRD-RES-EXFIL-DOMAIN", "WRD-RES-EXFIL-IP-LITERAL"}
+    {
+        "WRD-RES-ANSI",
+        "WRD-RES-SECRET-ECHO",
+        "WRD-RES-EXFIL-DOMAIN",
+        "WRD-RES-EXFIL-IP-LITERAL",
+        "WRD-RES-EXFIL-DNS-SSRF",
+    }
 )
 
 #: Severity -> SARIF level (CHECKS.md §2), mirrored for WRD-RES-*.

tests/test_guard_posture.py

@@ -144,6 +144,25 @@ def test_exfil_ip_literal_no_block_deterministic_fold_demotes():
     assert GuardConfig(no_block_exfil_ip_literal=True).category_enabled("WRD-RES-EXFIL-IP-LITERAL") is False
 
 
+# --- WRD-RES-EXFIL-DNS-SSRF category posture (DSE-58) -------------------------
+
+
+def test_exfil_dns_ssrf_blocks_by_default():
+    assert GuardConfig().category_enabled("WRD-RES-EXFIL-DNS-SSRF") is True
+
+
+def test_exfil_dns_ssrf_opt_out_demotes():
+    assert GuardConfig(no_block_exfil_dns_ssrf=True).category_enabled("WRD-RES-EXFIL-DNS-SSRF") is False
+
+
+def test_exfil_dns_ssrf_audit_only_demotes():
+    assert GuardConfig(audit_only=True).category_enabled("WRD-RES-EXFIL-DNS-SSRF") is False
+
+
+def test_exfil_dns_ssrf_no_block_deterministic_demotes():
+    assert GuardConfig(no_block_exfil_dns_ssrf=True).category_enabled("WRD-RES-EXFIL-DNS-SSRF") is False
+
+
 # --- framing: both modes round-trip ------------------------------------------