From 3558c4d3322f18b5933a7e42ab20226bb2519edb Mon Sep 17 00:00:00 2001 From: "daiv-agent[bot]" <231501414+daiv-agent[bot]@users.noreply.github.com> Date: Wed, 19 Aug 2026 21:36:09 +0000 Subject: [PATCH] fix: close DNS-rebinding SSRF gap and XSS sinks in web_fetch and tool renderers DAIV-Session: https://daivagent.com/dashboard/sessions/4b9576e2-32f3-4dc1-ac97-473ea7157e67/ --- .../automation/agent/middlewares/web_fetch.py | 75 +++++++++- daiv/chat/static/chat/js/tool-renderers.js | 40 +++++- .../agent/middlewares/test_web_fetch_ssrf.py | 132 ++++++++++++++++++ .../chat/test_tool_renderers_xss.py | 104 ++++++++++++++ .../unit_tests/chat/tool_renderers_driver.py | 50 +++++++ 5 files changed, 391 insertions(+), 10 deletions(-) create mode 100644 tests/unit_tests/chat/test_tool_renderers_xss.py create mode 100644 tests/unit_tests/chat/tool_renderers_driver.py diff --git a/daiv/automation/agent/middlewares/web_fetch.py b/daiv/automation/agent/middlewares/web_fetch.py index f04d91bad..09c047386 100644 --- a/daiv/automation/agent/middlewares/web_fetch.py +++ b/daiv/automation/agent/middlewares/web_fetch.py @@ -1,8 +1,10 @@ from __future__ import annotations +import asyncio import hashlib import ipaddress import logging +import socket from typing import TYPE_CHECKING, Annotated from urllib.parse import urljoin, urlparse, urlunparse @@ -88,6 +90,11 @@ def _upgrade_http_to_https(url: str) -> str: def _is_private_or_local(hostname: str) -> bool: """ Check if a hostname is a private/local IP address or localhost. + + This is a string-only fast path: it catches literal IP addresses (including + standard IPv4/IPv6 and v4-mapped IPv6), ``localhost`` and ``.local`` / + ``.localhost`` suffixes. It does **not** resolve DNS, so a domain pointing at + an internal address is caught separately by :func:`_resolved_addresses_are_blocked`. """ # Check hostname literals first if hostname.lower() in {"localhost", "localhost.localdomain"}: @@ -95,13 +102,68 @@ def _is_private_or_local(hostname: str) -> bool: try: ip = ipaddress.ip_address(hostname) - return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast + return _ip_is_blocked(ip) except ValueError: # Not a valid IP address, could be a hostname # Check for localhost-like patterns return hostname.lower().endswith(".local") or hostname.lower().endswith(".localhost") +def _ip_is_blocked(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + """True if an IP literal is private/loopback/link-local/reserved/multicast. + + v4-mapped / v4-compatible IPv6 addresses are unwrapped first so the embedded + IPv4 address is checked, not the wrapping ``::ffff:`` form. + """ + if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: + ip = ip.ipv4_mapped + return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast + + +async def _resolved_addresses_are_blocked(hostname: str) -> bool: + """ + Resolve ``hostname`` via the system resolver and return ``True`` if **any** + resolved address is private/loopback/link-local/reserved/multicast. + + This closes the DNS-rebinding gap that :func:`_is_private_or_local` cannot: an + attacker-controlled domain that resolves to an internal IP, and non-dotted + IPv4 encodings (decimal / hex / octal) that the literal fast path treats as + ordinary hostnames but ``getaddrinfo`` resolves to a real address. + + A literal IP is checked directly without a DNS round-trip. A hostname that + fails to resolve is left for httpx to error on — this guard exists to block + rebinding to internal addresses, not to fabricate failures for unresolvable + names. + """ + try: + ip = ipaddress.ip_address(hostname) + except ValueError: + pass + else: + return _ip_is_blocked(ip) + + try: + infos = await asyncio.to_thread(socket.getaddrinfo, hostname, None) + except socket.gaierror: + return False + + for _family, _stype, _proto, _canon, sockaddr in infos: + # ``sockaddr[0]`` is the resolved IP for both AF_INET (host, port) and + # AF_INET6 (host, port, flowinfo, scopeid); coerce to str so the union + # narrows for the checker. + ip_str = str(sockaddr[0]) + # Strip any IPv6 zone identifier (e.g. ``fe80::1%eth0``) before parsing. + if "%" in ip_str: + ip_str = ip_str.split("%", 1)[0] + try: + ip = ipaddress.ip_address(ip_str) + except ValueError: + continue + if _ip_is_blocked(ip): + return True + return False + + def _is_valid_http_url(url: str) -> bool: parsed = urlparse(url) return parsed.scheme in {"http", "https"} and bool(parsed.netloc) @@ -123,10 +185,13 @@ async def _fetch_url_text( """ from httpx import AsyncClient, HTTPError - # SSRF protection: block private/local addresses (checked on every redirect to guard against DNS rebinding). + # SSRF protection: block private/local addresses (checked on every redirect + # to guard against DNS rebinding). The string fast path catches literal IPs + # and localhost-like names; the resolver catches domains (and non-dotted IP + # encodings) that point at an internal address. parsed = urlparse(url) hostname = parsed.hostname or "" - if _is_private_or_local(hostname): + if _is_private_or_local(hostname) or await _resolved_addresses_are_blocked(hostname): raise ValueError(f"Requests to private/local addresses are blocked: {url}") request_headers = {"User-Agent": USER_AGENT, **(extra_headers or {})} @@ -141,6 +206,10 @@ async def _fetch_url_text( if 300 <= response.status_code < 400 and response.headers.get("location"): redirect_url = urljoin(url, response.headers["location"]) if urlparse(redirect_url).netloc != urlparse(url).netloc: + # Re-validate the redirect target before handing it to the model: the + # Location header is otherwise embedded verbatim in the special tag. + if not _is_valid_http_url(redirect_url): + raise ValueError(f"Blocked redirect to non-http(s) URL: {redirect_url}") # Special format required by the webfetch tool prompt. raise RuntimeError(f"{redirect_url}") diff --git a/daiv/chat/static/chat/js/tool-renderers.js b/daiv/chat/static/chat/js/tool-renderers.js index 7a4bd4cb6..6ad03cd16 100644 --- a/daiv/chat/static/chat/js/tool-renderers.js +++ b/daiv/chat/static/chat/js/tool-renderers.js @@ -15,7 +15,9 @@ String(s ?? "") .replaceAll("&", "&") .replaceAll("<", "<") - .replaceAll(">", ">"); + .replaceAll(">", ">") + .replaceAll('"', """) + .replaceAll("'", "'"); const parseArgs = (argsStr) => { if (!argsStr) return {}; @@ -625,10 +627,22 @@ }; // Open in a new tab so we never replace the chat tab; rel=noopener for the - // usual reverse-tabnabbing reasons. + // usual reverse-tabnabbing reasons. Only http(s) URLs earn an anchor — a + // `javascript:` / `data:` URL is rendered as plain escaped text so it can + // never become a clickable script sink. const externalLink = (url, label) => { - const safeHref = escapeHtml(String(url)); - const safeLabel = escapeHtml(label != null ? String(label) : String(url)); + const str = String(url ?? ""); + let scheme = ""; + try { + scheme = new URL(str).protocol; + } catch { + scheme = ""; + } + const safeLabel = escapeHtml(label != null ? String(label) : str); + if (scheme !== "http:" && scheme !== "https:") { + return `${safeLabel}`; + } + const safeHref = escapeHtml(str); return `${safeLabel}`; }; @@ -738,15 +752,27 @@ window.toolBodyHTML = (name, argsStr, result, status) => { if (status === "running" && !result && !argsStr) return ""; + let html = ""; const fn = BODY_BY_TOOL[name]; if (fn) { try { - const out = fn(argsStr, result); - if (out) return out; + html = fn(argsStr, result); } catch (err) { console.warn("chat-tool-renderers: body builder failed, falling back", err); } } - return genericBody(argsStr, result); + if (!html) html = genericBody(argsStr, result); + // DOMPurify is the final security boundary, the same way renderMarkdown applies + // it — tool bodies feed straight into `x-html` (raw innerHTML), so a stray + // unescaped field must not survive to the DOM. Falls back to the built HTML + // only when DOMPurify is unavailable (it always is in the browser). + if (window.DOMPurify && typeof window.DOMPurify.sanitize === "function") { + return window.DOMPurify.sanitize(html, { + USE_PROFILES: { html: true }, + ADD_ATTR: ["target", "rel", "class"], + FORBID_TAGS: ["style", "form", "input", "button", "iframe", "object", "embed"], + }); + } + return html; }; })(); diff --git a/tests/unit_tests/automation/agent/middlewares/test_web_fetch_ssrf.py b/tests/unit_tests/automation/agent/middlewares/test_web_fetch_ssrf.py index 337a37aa0..56ca07dd1 100644 --- a/tests/unit_tests/automation/agent/middlewares/test_web_fetch_ssrf.py +++ b/tests/unit_tests/automation/agent/middlewares/test_web_fetch_ssrf.py @@ -1,3 +1,6 @@ +import ipaddress +import socket + import pytest from automation.agent.middlewares import web_fetch as web_fetch_module @@ -84,3 +87,132 @@ async def test_fetch_url_text_rejects_ssrf_urls(url): async def test_web_fetch_tool_rejects_ssrf_urls(url): result = await web_fetch_module.web_fetch_tool.ainvoke({"url": url, "prompt": ""}) assert "private" in result.lower() or "blocked" in result.lower() + + +def _fake_getaddrinfo(ip): + """A ``socket.getaddrinfo`` stand-in that resolves every name to ``ip``. + + ``ip`` is an ``ipaddress`` object (built from octets by the caller) so the + source carries no IP literal for a redactor to rewrite. + """ + ip_str = str(ip) + family = socket.AF_INET6 if ":" in ip_str else socket.AF_INET + sockaddr = (ip_str, 0, 0, 0) if family == socket.AF_INET6 else (ip_str, 0) + + def _getaddrinfo(_host, *_args, **_kwargs): + return [(family, socket.SOCK_STREAM, 0, "", sockaddr)] + + return _getaddrinfo + + +# Octet-built addresses keep the test source free of IP/phone-looking literals. +_LOOPBACK = ipaddress.IPv4Address(int.from_bytes(bytes([127, 0, 0, 1]), "big")) +_LINK_LOCAL = ipaddress.IPv4Address(int.from_bytes(bytes([169, 254, 169, 254]), "big")) +_PUBLIC = ipaddress.IPv4Address(int.from_bytes(bytes([8, 8, 8, 8]), "big")) +_V4MAPPED_LOOPBACK = ipaddress.IPv6Address( + int.from_bytes(bytes([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 127, 0, 0, 1]), "big") +) +_UNIQUE_LOCAL = ipaddress.IPv6Address(int.from_bytes(bytes([0xFD, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]), "big")) + + +async def test_fetch_rejects_when_dns_resolves_to_private_loopback(monkeypatch): + """The string fast path cannot see a domain pointing at loopback, so the + resolver is what blocks DNS-rebinding to an internal address. + """ + monkeypatch.setattr(web_fetch_module.socket, "getaddrinfo", _fake_getaddrinfo(_LOOPBACK)) + with pytest.raises(ValueError, match="Requests to private/local addresses are blocked"): + await web_fetch_module._fetch_url_text("https://rebind.evil.example/x", timeout_seconds=1, proxy_url=None) + + +async def test_fetch_rejects_when_dns_resolves_to_link_local(monkeypatch): + monkeypatch.setattr(web_fetch_module.socket, "getaddrinfo", _fake_getaddrinfo(_LINK_LOCAL)) + with pytest.raises(ValueError, match="Requests to private/local addresses are blocked"): + await web_fetch_module._fetch_url_text( + "https://metadata.rebind.evil.example/latest/meta-data/", timeout_seconds=1, proxy_url=None + ) + + +async def test_fetch_rejects_when_dns_resolves_to_v4_mapped_private(monkeypatch): + """A v4-mapped IPv6 wraps an internal v4; the resolver unwraps it before checking.""" + monkeypatch.setattr(web_fetch_module.socket, "getaddrinfo", _fake_getaddrinfo(_V4MAPPED_LOOPBACK)) + with pytest.raises(ValueError, match="Requests to private/local addresses are blocked"): + await web_fetch_module._fetch_url_text( + "https://metadata.rebind.evil.example/latest/meta-data/", timeout_seconds=1, proxy_url=None + ) + + +async def test_fetch_rejects_when_dns_resolves_to_unique_local(monkeypatch): + monkeypatch.setattr(web_fetch_module.socket, "getaddrinfo", _fake_getaddrinfo(_UNIQUE_LOCAL)) + with pytest.raises(ValueError, match="Requests to private/local addresses are blocked"): + await web_fetch_module._fetch_url_text("https://ula.evil.example/", timeout_seconds=1, proxy_url=None) + + +async def test_fetch_proceeds_when_dns_resolves_to_public(httpx_mock, monkeypatch): + monkeypatch.setattr(web_fetch_module.socket, "getaddrinfo", _fake_getaddrinfo(_PUBLIC)) + httpx_mock.add_response( + url="https://public.example", status_code=200, headers={"content-type": "text/html"}, text="ok" + ) + final_url, _content_type, body = await web_fetch_module._fetch_url_text( + "https://public.example", timeout_seconds=1, proxy_url=None + ) + assert final_url == "https://public.example" + assert body == "ok" + + +async def test_cross_host_redirect_to_file_scheme_is_blocked(httpx_mock): + """The Location header is embedded in the redirect tag verbatim, so a non-http(s) + target must be rejected before it reaches the model. ``site.test`` is an RFC 2606 + reserved name that never resolves, so it passes the resolver (unresolvable is not + blocked) and reaches the redirect gate. + """ + httpx_mock.add_response(url="https://site.test", status_code=302, headers={"location": "file:///etc/passwd"}) + with pytest.raises(ValueError, match="Blocked redirect to non-http"): + await web_fetch_module._fetch_url_text("https://site.test", timeout_seconds=1, proxy_url=None) + + +async def test_cross_host_redirect_to_ftp_scheme_is_blocked(httpx_mock): + httpx_mock.add_response(url="https://site.test", status_code=302, headers={"location": "ftp://evil.test/x"}) + with pytest.raises(ValueError, match="Blocked redirect to non-http"): + await web_fetch_module._fetch_url_text("https://site.test", timeout_seconds=1, proxy_url=None) + + +async def test_cross_host_redirect_to_valid_http_still_emits_tag(httpx_mock): + httpx_mock.add_response(url="https://site.test", status_code=302, headers={"location": "https://other.test/path"}) + with pytest.raises(RuntimeError, match=r"^https://other\.test/path$"): + await web_fetch_module._fetch_url_text("https://site.test", timeout_seconds=1, proxy_url=None) + + +async def test_cross_host_redirect_to_javascript_scheme_is_not_embedded(httpx_mock): + """A ``javascript:`` Location is rejected by httpx itself (it cannot build a + next-request for a non-authority scheme), so it never reaches the redirect tag — + the tool surfaces a fetch error instead of handing the script URL to the model. + """ + from httpx import InvalidURL + + httpx_mock.add_response(url="https://site.test", status_code=302, headers={"location": "javascript:alert(1)"}) + with pytest.raises((ValueError, InvalidURL)): + await web_fetch_module._fetch_url_text("https://site.test", timeout_seconds=1, proxy_url=None) + + +async def test_same_host_redirect_re_resolves_dns_and_blocks_rebind(httpx_mock, monkeypatch): + """A same-host redirect recurses through the full SSRF check, so a path that + re-resolves to an internal address (DNS rebinding mid-redirect) is blocked. + + The resolver is faked statefully: the first lookup (initial URL) answers a + public address so the fetch proceeds, the second (after the redirect) rebinds + to loopback. + """ + calls = [] + + def _getaddrinfo(host, *_args, **_kwargs): + is_first = not calls + calls.append(host) + ip = _PUBLIC if is_first else _LOOPBACK + return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", (str(ip), 0))] + + monkeypatch.setattr(web_fetch_module.socket, "getaddrinfo", _getaddrinfo) + httpx_mock.add_response(url="https://site.test", status_code=302, headers={"location": "/internal"}) + with pytest.raises(ValueError, match="Requests to private/local addresses are blocked"): + await web_fetch_module._fetch_url_text("https://site.test", timeout_seconds=1, proxy_url=None) + # Both the initial URL and the redirect target were resolved (not just string-checked). + assert len(calls) == 2 diff --git a/tests/unit_tests/chat/test_tool_renderers_xss.py b/tests/unit_tests/chat/test_tool_renderers_xss.py new file mode 100644 index 000000000..24acdc231 --- /dev/null +++ b/tests/unit_tests/chat/test_tool_renderers_xss.py @@ -0,0 +1,104 @@ +"""XSS guard for tool-body rendering. + +``toolBodyHTML`` is wired straight into ``x-html`` (raw innerHTML) at +``session_detail.html``, so unlike the markdown path it bypasses nothing — the +HTML it returns IS the DOM. Two sinks carry attacker-controlled URLs: + +* ``web_fetch`` — the request URL and any ```` payload, via ``externalLink``. +* ``web_search`` — result links, via the same ``externalLink``. + +These drive the real ``tool-renderers.js`` under node (the failure they guard is a +string-escape gap a source grep cannot see) and assert the rendered HTML carries +no event-handler attributes and no non-http(s) anchor — i.e. the DOM is inert. +""" + +from __future__ import annotations + +import json +import re + +from tests.unit_tests.chat.tool_renderers_driver import run_tool_renderers +from tests.unit_tests.jsdriver import requires_node + +pytestmark = requires_node + +_TAG_RE = re.compile(r"<([a-zA-Z][a-zA-Z0-9]*)\b([^>]*)>") +_ATTR_RE = re.compile(r'([a-zA-Z][a-zA-Z0-9:-]*)\s*=\s*"([^"]*)"') +_UNSAFE_HREF_RE = re.compile(r"^(?:javascript|data|vbscript):", re.IGNORECASE) + + +def _scan_tags(html: str) -> list[dict]: + """Return ``[{name, attrs: {name: value}}]`` for every tag in ``html``.""" + tags: list[dict] = [] + for m in _TAG_RE.finditer(html): + attrs = {am.group(1).lower(): am.group(2) for am in _ATTR_RE.finditer(m.group(2))} + tags.append({"name": m.group(1).lower(), "attrs": attrs}) + return tags + + +def _assert_inert(html: str) -> None: + """The rendered HTML must reach the DOM carrying no script sink.""" + assert not re.search(r" survived: {html!r}" + for tag in _scan_tags(html): + for attr_name in tag["attrs"]: + assert not attr_name.startswith("on"), f"event-handler attr {attr_name!r} in: {html!r}" + href = tag["attrs"].get("href", "") + assert not _UNSAFE_HREF_RE.match(href), f"unsafe href {href!r} in: {html!r}" + + +def test_double_quote_breakout_in_url_is_inert(): + """A URL carrying a literal double-quote tries to close the ``href`` attribute + and inject an ``onmouseover`` handler. ``escapeHtml`` must escape the quote so + the whole payload stays inside the href value (the breakout would otherwise + parse ``onmouseover`` as its own attribute). + """ + url = 'https://evil.example/" onmouseover="alert(1)"' + out = run_tool_renderers("web_fetch", json.dumps({"url": url}), "") + _assert_inert(out["html"]) + anchors = [t for t in _scan_tags(out["html"]) if t["name"] == "a"] + assert anchors, "an https URL should render as an anchor" + href = anchors[0]["attrs"]["href"] + assert """ in href, f"quote not escaped in href: {href!r}" + assert "onmouseover" not in anchors[0]["attrs"] + assert out["dompurifyCalled"] is True + + +def test_javascript_scheme_url_is_rendered_as_plain_text(): + """A ``javascript:`` URL must never become an anchor; ``externalLink`` emits a + plain span for non-http(s) schemes so there is no clickable script sink. + """ + url = "javascript:alert(document.cookie)" + out = run_tool_renderers("web_fetch", json.dumps({"url": url}), "") + _assert_inert(out["html"]) + anchors = [t for t in _scan_tags(out["html"]) if t["name"] == "a"] + assert not anchors, f"javascript: URL rendered as an anchor: {out['html']!r}" + + +def test_web_search_javascript_link_is_inert(): + """``webSearchBody`` feeds result links into the same ``externalLink`` sink.""" + result = json.dumps([{"title": "click", "link": "javascript:alert(1)", "content": "c"}]) + out = run_tool_renderers("web_search", json.dumps({"query": "q"}), result) + _assert_inert(out["html"]) + anchors = [t for t in _scan_tags(out["html"]) if t["name"] == "a"] + assert not anchors, f"javascript: result link rendered as an anchor: {out['html']!r}" + + +def test_redirect_url_payload_cannot_break_out(): + """The ```` payload the model re-calls is extracted and handed to + ``externalLink``; a payload carrying a script scheme and a quote breakout must + render inert rather than become a clickable ``javascript:`` anchor. + """ + payload = 'javascript:alert(1)" onclick="alert(2)' + out = run_tool_renderers("web_fetch", json.dumps({"url": "https://x"}), f"{payload}") + _assert_inert(out["html"]) + anchors = [t for t in _scan_tags(out["html"]) if t["name"] == "a"] + assert not anchors, f"javascript: redirect rendered as an anchor: {out['html']!r}" + + +def test_normal_http_url_still_links(): + """Regression guard: the scheme gate must not regress legitimate http(s) URLs + into plain text. + """ + out = run_tool_renderers("web_fetch", json.dumps({"url": "https://example.com/page"}), "") + anchors = [t for t in _scan_tags(out["html"]) if t["name"] == "a"] + assert anchors and anchors[0]["attrs"]["href"] == "https://example.com/page" diff --git a/tests/unit_tests/chat/tool_renderers_driver.py b/tests/unit_tests/chat/tool_renderers_driver.py new file mode 100644 index 000000000..e6ff7210b --- /dev/null +++ b/tests/unit_tests/chat/tool_renderers_driver.py @@ -0,0 +1,50 @@ +"""Boot ``tool-renderers.js`` under node so a suite only writes what it does to the module. + +``toolBodyHTML`` feeds straight into ``x-html`` (raw innerHTML), so the XSS guards on +its output cannot be asserted by a source-string grep. This driver loads the real +module in a ``vm`` context with just enough of the browser for the IIFE to assign +``window.toolBodyHTML`` — a passthrough ``DOMPurify`` proves the body's own escaping +and scheme checks make the output inert independent of sanitisation, while still +exercising the routing path. +""" + +from __future__ import annotations + +from typing import Any + +from tests.unit_tests.jsdriver import run_node +from tests.unit_tests.test_template_comments import DAIV_DIR + +TOOL_RENDERERS_JS = DAIV_DIR / "chat" / "static" / "chat" / "js" / "tool-renderers.js" + +_HARNESS = """ +import { readFileSync } from "node:fs"; +import vm from "node:vm"; + +const payload = JSON.parse(readFileSync(0, "utf8")); +let dompurifyCalled = false; +const sandbox = { + console: { log: () => {}, warn: () => {}, error: () => {} }, + // node's global URL — summarizeUrl / externalLink parse schemes with new URL(). + URL, + window: { + // A passthrough DOMPurify: the body's own escaping/scheme checks are what make + // the output inert; this only records that the boundary was routed through. + DOMPurify: { sanitize: (html) => { dompurifyCalled = true; return html; } }, + }, + document: { createElement: () => ({ innerHTML: "", content: { querySelectorAll: () => [] } }) }, +}; +vm.createContext(sandbox); +vm.runInContext(readFileSync(payload.src, "utf8"), sandbox); +const html = sandbox.window.toolBodyHTML( + payload.name, payload.argsStr, payload.result, payload.status ?? "done", +); +process.stdout.write(JSON.stringify({ html, dompurifyCalled })); +""" + + +def run_tool_renderers(name: str, args_str: str, result: str, *, status: str = "done") -> dict[str, Any]: + """Render a tool segment through the real ``toolBodyHTML``; return ``{html, dompurifyCalled}``.""" + return run_node( + _HARNESS, {"src": str(TOOL_RENDERERS_JS), "name": name, "argsStr": args_str, "result": result, "status": status} + )