fix(image-search): share public result URL screening (#6740)

* fix(image-search): screen DuckDuckGo result URLs

* fix(image-search): unify provider result URL screening

* docs(image-search): scope URL guidance to avoid inherited size limits

---------

Co-authored-by: huyengiang101086-spec <270684046+huyengiang101086-spec@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -1526,7 +1526,7 @@ Public-skill CI waivers are exact, expiring exceptions in `.github/skill-review-
 
 Tools follow the same philosophy. DeerFlow comes with a core toolset — web search, web fetch, rendered web capture, file operations, bash execution — and supports custom tools via MCP servers and Python functions. The bundled DDG, Brave, Tavily, SearXNG, and Serper search providers accept an optional `time_range` of `day`, `week`, `month`, or `year`; omitting it preserves existing search behavior. For DDG recency searches, DeerFlow excludes DDGS backends that ignore time limits. Swap anything. Add anything.
 
-For DDG web search and DDG image search, `max_results` in `config.yaml` can be a positive integer or an environment-variable reference such as `max_results: $DDG_MAX_RESULTS` with `DDG_MAX_RESULTS=5`. The configured value takes precedence over the tool call's `max_results` argument. Invalid values (for example, `abc`, an empty string, or `3.5`), zero, and negative counts produce a warning and fall back to the default of 5 results.
+For DDG web search and DDG image search, `max_results` in `config.yaml` can be a positive integer or an environment-variable reference such as `max_results: $DDG_MAX_RESULTS` with `DDG_MAX_RESULTS=5`. The configured value takes precedence over the tool call's `max_results` argument. Invalid values (for example, `abc`, an empty string, or `3.5`), zero, and negative counts produce a warning and fall back to the default of 5 results. DDG, Brave, and Serper image search share one result-URL screen that omits non-HTTP(S), localhost, cloud-metadata hostnames, and non-public address literals, including multicast and IPv6 spellings embedding private IPv4 addresses. The screen does not resolve DNS; downloaders must validate the resolved address when connecting. A result with neither a safe full image URL nor a safe thumbnail is omitted.
 
 Stdio MCP servers can set `cwd` in `extensions_config.json` when their entrypoint
 or data files depend on a specific working directory. The setting applies to

backend/AGENTS.md

@@ -64,6 +64,8 @@ Every code change must keep docs accurate and current: update `README.md` for
 user-facing behavior and the relevant `AGENTS.md` for development changes.
 `CLAUDE.md` imports `AGENTS.md`; do not edit the shim.
 
+Image result URL policy: [image search guide](packages/harness/deerflow/community/image_search/AGENTS.md).
+
 ### Backend Benchmarks
 
 `scripts/benchmark/context_snapshot/`: explicit `run-live` needs provider env

backend/packages/harness/deerflow/community/brave/AGENTS.md

@@ -16,3 +16,8 @@ through an offline HTTPX transport with synthetic JSON bodies. Do not mock the
 normalizer or substitute its final output. Patch the provider's `httpx` reference
 with a local client factory and `HTTPStatusError`; never replace the shared
 `httpx.Client`. Cover both the unchanged global client and HTTP error handling.
+
+Image result URLs use `community.url_safety.safe_public_result_url` (imported
+as `_safe_public_url`); do not add a provider-local copy. It checks strings
+without DNS and shares the non-public address predicate with fetch validation.
+Tests: `tests/test_image_result_url_safety.py` plus the existing provider suite.

backend/packages/harness/deerflow/community/brave/tools.py

@@ -13,13 +13,12 @@ giving structured results, authenticated quota, and a documented SLA.
 import json
 import logging
 import os
-from ipaddress import IPv4Address, IPv6Address, ip_address, ip_network
-from urllib.parse import urlparse
 
 import httpx
 from langchain.tools import tool
 
 from deerflow.community.search_time_range import BRAVE_FRESHNESS_BY_TIME_RANGE, SearchTimeRange
+from deerflow.community.url_safety import safe_public_result_url as _safe_public_url
 from deerflow.config import get_app_config
 
 logger = logging.getLogger(__name__)
@@ -31,8 +30,6 @@ _DEFAULT_MAX_RESULTS = 5
 _BRAVE_WEB_MAX_COUNT = 20
 # Brave Image Search supports larger batches than web search.
 _BRAVE_IMAGE_MAX_COUNT = 200
-# NAT64 well-known prefix (RFC 6052): IPv6 literals embedding an IPv4 address.
-_NAT64_PREFIX = ip_network("64:ff9b::/96")
 _api_key_warned: set[str] = set()
 
 
@@ -94,110 +91,10 @@ def _unexpected_format_error(query: str, *, service_name: str = "Brave Search")
     )
 
 
-def _decode_ipv4(host: str) -> IPv4Address | None:
-    """Decode obfuscated IPv4 literals that ``ip_address`` rejects.
-
-    Mirrors the permissive ``inet_aton`` parsing many HTTP clients use, so that
-    integer (``2130706433``), hex (``0x7f000001``) and octal (``0177.0.0.1``)
-    encodings of an address are recognized.
-    """
-    parts = host.split(".")
-    if not 1 <= len(parts) <= 4:
-        return None
-
-    values: list[int] = []
-    for part in parts:
-        if not part:
-            return None
-        try:
-            if part.startswith(("0x", "0X")):
-                values.append(int(part, 16))
-            elif part.startswith("0") and len(part) > 1:
-                values.append(int(part, 8))
-            else:
-                values.append(int(part, 10))
-        except ValueError:
-            return None
-
-    *leading, last = values
-    for value in leading:
-        if not 0 <= value <= 0xFF:
-            return None
-    max_last = (1 << (8 * (4 - len(leading)))) - 1
-    if not 0 <= last <= max_last:
-        return None
-
-    result = 0
-    for value in leading:
-        result = (result << 8) | value
-    result = (result << (8 * (4 - len(leading)))) | last
-    return IPv4Address(result)
-
-
 def _is_url_present(value: object) -> bool:
     return isinstance(value, str) and bool(value.strip())
 
 
-def _embedded_ipv4(ip: IPv6Address) -> IPv4Address | None:
-    """Extract an IPv4 address embedded in an IPv6 literal, if any.
-
-    Covers IPv4-mapped (``::ffff:a.b.c.d``), 6to4 (``2002::/16``), NAT64
-    (``64:ff9b::/96``), and IPv4-compatible (``::a.b.c.d``) forms. These all
-    smuggle a v4 destination through the IPv6 path, where ``is_global`` on the
-    v6 literal alone would otherwise report a loopback/private target as safe.
-    """
-    if ip.ipv4_mapped is not None:
-        return ip.ipv4_mapped
-    if ip.sixtofour is not None:
-        return ip.sixtofour
-    if ip in _NAT64_PREFIX:
-        return IPv4Address(int(ip) & 0xFFFFFFFF)
-    # IPv4-compatible ``::a.b.c.d`` (high 96 bits zero, excluding ::/:: 1).
-    packed = int(ip)
-    if packed >> 32 == 0 and packed > 1:
-        return IPv4Address(packed & 0xFFFFFFFF)
-    return None
-
-
-def _safe_public_url(value: object) -> str:
-    """Return ``value`` only if it is a safe, public http(s) URL, else "".
-
-    This is a best-effort SSRF guard that rejects non-http(s) schemes,
-    ``localhost``, and private/non-global IP literals (including obfuscated
-    decimal/hex/octal encodings and IPv6 literals embedding a non-global IPv4).
-    It only inspects the URL string and cannot catch public hostnames that
-    resolve to internal IPs; any consumer that actually downloads these URLs
-    must re-validate the resolved IP at fetch time.
-    """
-    if not isinstance(value, str):
-        return ""
-    url = value.strip()
-    try:
-        parsed = urlparse(url)
-    except ValueError:
-        return ""
-    if parsed.scheme not in {"http", "https"} or not parsed.netloc or not parsed.hostname:
-        return ""
-
-    host = parsed.hostname.lower().rstrip(".")
-    if not host:
-        return ""
-    if host == "localhost" or host.endswith(".localhost"):
-        return ""
-
-    try:
-        ip = ip_address(host)
-    except ValueError:
-        ip = _decode_ipv4(host)
-        if ip is None:
-            return url
-    if isinstance(ip, IPv6Address):
-        embedded = _embedded_ipv4(ip)
-        if embedded is not None and not embedded.is_global:
-            return ""
-    return url if ip.is_global else ""
-
-
 def _brave_get(
     endpoint: str,
     api_key: str,

backend/packages/harness/deerflow/community/image_search/AGENTS.md

@@ -0,0 +1,9 @@
+# Image Search
+
+DDG, Brave, and Serper `image_search` screen returned image and thumbnail URLs with
+`community.url_safety.safe_public_result_url`. This pure string check does not
+resolve hostnames; any later downloader must enforce its own connection-time
+address policy. The screen reuses `is_blocked_address` for literal and embedded
+IPv4 addresses, including multicast/reserved forms. Shared regression coverage
+is in `tests/test_image_result_url_safety.py`; DDG tool coverage is in
+`tests/test_ddg_image_search_urls.py`.

backend/packages/harness/deerflow/community/image_search/tools.py

@@ -9,6 +9,7 @@ import logging
 from langchain.tools import tool
 
 from deerflow.community.search_max_results import DEFAULT_MAX_RESULTS, coerce_max_results
+from deerflow.community.url_safety import safe_public_result_url
 from deerflow.config import get_app_config
 
 logger = logging.getLogger(__name__)
@@ -132,14 +133,25 @@ async def image_search_tool(
     if not results:
         return json.dumps({"error": "No images found", "query": query}, ensure_ascii=False)
 
-    normalized_results = [
-        {
-            "title": r.get("title", ""),
-            "image_url": r.get("image", ""),
-            "thumbnail_url": r.get("thumbnail", ""),
-        }
-        for r in results
-    ]
+    normalized_results = []
+    for row in results:
+        if not isinstance(row, dict):
+            continue
+        image_url = safe_public_result_url(row.get("image"))
+        thumbnail_url = safe_public_result_url(row.get("thumbnail"))
+        if not image_url and not thumbnail_url:
+            continue
+        title = row.get("title")
+        normalized_results.append(
+            {
+                "title": title if isinstance(title, str) else "",
+                "image_url": image_url,
+                "thumbnail_url": thumbnail_url,
+            }
+        )
+
+    if not normalized_results:
+        return json.dumps({"error": "No safe image URLs found", "query": query}, ensure_ascii=False)
 
     output = {
         "query": query,

backend/packages/harness/deerflow/community/serper/AGENTS.md

@@ -45,3 +45,8 @@ Budget aborts before and after backoff log a warning identifying the stage,
 without query/key values, and preserve the last structured error. Date tests
 pin asctime UTC anchoring (under a non-UTC host timezone where supported) and
 the RFC 850 50-year cutoff down to the second.
+
+Image result URLs use `community.url_safety.safe_public_result_url` (imported
+as `_safe_public_url`); do not add a provider-local copy. It checks strings
+without DNS and shares the non-public address predicate with fetch validation.
+Tests: `tests/test_image_result_url_safety.py` plus the existing provider suite.

backend/packages/harness/deerflow/community/serper/tools.py

@@ -22,6 +22,7 @@ import httpx
 from langchain.tools import tool
 
 from deerflow.community.search_time_range import SearchTimeRange
+from deerflow.community.url_safety import safe_public_result_url as _safe_public_url
 from deerflow.config import get_app_config
 
 logger = logging.getLogger(__name__)
@@ -216,44 +217,6 @@ def _is_url_present(value: object) -> bool:
     return isinstance(value, str) and bool(value.strip())
 
 
-def _safe_public_url(value: object) -> str:
-    """Return ``value`` only if it is a safe, public http(s) URL, else "".
-
-    This is a best-effort SSRF guard that rejects non-http(s) schemes,
-    ``localhost``, and private/non-global IP literals (including obfuscated
-    decimal/hex/octal encodings). It only inspects the URL string and cannot
-    catch public hostnames that resolve to internal IPs (e.g. DNS rebinding);
-    any consumer that actually downloads these URLs must re-validate the
-    resolved IP at fetch time.
-    """
-    if not isinstance(value, str):
-        return ""
-    url = value.strip()
-    try:
-        parsed = urlparse(url)
-    except ValueError:
-        return ""
-    if parsed.scheme not in {"http", "https"} or not parsed.netloc or not parsed.hostname:
-        return ""
-
-    # Strip a single trailing dot (FQDN root label). ``localhost.`` and
-    # ``127.0.0.1.`` resolve to loopback on common resolvers but would
-    # otherwise slip past the localhost/IP checks below.
-    host = parsed.hostname.lower().rstrip(".")
-    if not host:
-        return ""
-    if host == "localhost" or host.endswith(".localhost"):
-        return ""
-
-    try:
-        ip = ip_address(host)
-    except ValueError:
-        ip = _decode_ipv4(host)
-        if ip is None:
-            return url
-    return url if ip.is_global else ""
-
-
 def _retry_options(config) -> dict:
     extra = config.model_extra if config is not None else {}
     return {name: extra[name] for name in ("max_retries", "retry_budget_seconds") if name in extra}

backend/packages/harness/deerflow/community/url_safety.py

@@ -10,6 +10,67 @@ from urllib.parse import urlparse
 _BLOCKED_HOSTNAMES = {"localhost", "metadata.google.internal"}
 
 
+def safe_public_result_url(value: object) -> str:
+    """Screen an untrusted result URL without doing DNS or network I/O.
+
+    This only filters URLs returned to callers. Downloaders must still check
+    the resolved address at connection time.
+    """
+    if not isinstance(value, str):
+        return ""
+    url = value.strip()
+    try:
+        parsed = urlparse(url)
+        host = parsed.hostname
+    except ValueError:
+        return ""
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc or not host:
+        return ""
+    host = host.lower().rstrip(".")
+    if not host or host in _BLOCKED_HOSTNAMES or host.endswith(".localhost"):
+        return ""
+
+    try:
+        address = ipaddress.ip_address(host)
+    except ValueError:
+        address = _decode_result_ipv4(host)
+        if address is None:
+            return url
+    if isinstance(address, ipaddress.IPv6Address):
+        embedded = address.ipv4_mapped or address.sixtofour
+        if embedded is None and address in ipaddress.ip_network("64:ff9b::/96"):
+            embedded = ipaddress.IPv4Address(int(address) & 0xFFFFFFFF)
+        if embedded is None and 1 < int(address) < 1 << 32:
+            embedded = ipaddress.IPv4Address(int(address) & 0xFFFFFFFF)
+        if embedded is not None and is_blocked_address(embedded):
+            return ""
+    return "" if is_blocked_address(address) else url
+
+
+def _decode_result_ipv4(host: str) -> ipaddress.IPv4Address | None:
+    """Recognize decimal, octal and hex IPv4 spellings accepted by clients."""
+    parts = host.split(".")
+    if not 1 <= len(parts) <= 4:
+        return None
+    values: list[int] = []
+    for part in parts:
+        try:
+            base = 16 if part.startswith(("0x", "0X")) else 8 if part.startswith("0") and len(part) > 1 else 10
+            values.append(int(part, base))
+        except ValueError:
+            return None
+    *leading, last = values
+    if any(not 0 <= value <= 255 for value in leading):
+        return None
+    remaining_bits = 8 * (4 - len(leading))
+    if not 0 <= last < 1 << remaining_bits:
+        return None
+    result = 0
+    for value in leading:
+        result = result << 8 | value
+    return ipaddress.IPv4Address(result << remaining_bits | last)
+
+
 def resolve_host_addresses(hostname: str) -> list[ipaddress._BaseAddress]:
     """Resolve a hostname to all IP addresses for SSRF screening.
 

backend/tests/test_agent_guidance_check.py

@@ -22,6 +22,7 @@ EXPECTED_GUIDANCE_PATHS = {
     "backend/packages/harness/deerflow/agents/task_continuity/AGENTS.md",
     "backend/packages/harness/deerflow/community/brave/AGENTS.md",
     "backend/packages/harness/deerflow/community/groundroute/AGENTS.md",
+    "backend/packages/harness/deerflow/community/image_search/AGENTS.md",
     "backend/packages/harness/deerflow/community/jina_ai/AGENTS.md",
     "backend/packages/harness/deerflow/community/ragflow/AGENTS.md",
     "backend/packages/harness/deerflow/community/serper/AGENTS.md",

backend/tests/test_ddg_image_search_urls.py

@@ -0,0 +1,52 @@
+import json
+from unittest.mock import patch
+
+import pytest
+
+from deerflow.community.image_search import tools
+from deerflow.community.url_safety import safe_public_result_url
+
+
+@pytest.mark.asyncio
+async def test_image_search_drops_unsafe_result_urls() -> None:
+    rows = [
+        {"title": "safe", "image": "https://images.example/full.jpg", "thumbnail": "https://images.example/thumb.jpg"},
+        {"title": "unsafe original", "image": "http://127.0.0.1/private", "thumbnail": "https://images.example/thumb2.jpg"},
+        {"title": "unsafe thumbnail", "image": "https://images.example/full3.jpg", "thumbnail": "javascript:alert(1)"},
+        {"title": "both unsafe", "image": "http://2130706433/private", "thumbnail": "http://localhost/thumb"},
+    ]
+    with patch.object(tools, "_search_images", return_value=rows), patch.object(tools, "get_app_config") as config:
+        config.return_value.get_tool_config.return_value = None
+        result = json.loads(await tools.image_search_tool.ainvoke({"query": "reference"}))
+
+    assert result["total_results"] == 3
+    assert result["results"] == [
+        {"title": "safe", "image_url": "https://images.example/full.jpg", "thumbnail_url": "https://images.example/thumb.jpg"},
+        {"title": "unsafe original", "image_url": "", "thumbnail_url": "https://images.example/thumb2.jpg"},
+        {"title": "unsafe thumbnail", "image_url": "https://images.example/full3.jpg", "thumbnail_url": ""},
+    ]
+
+
+@pytest.mark.asyncio
+async def test_image_search_reports_when_all_urls_are_unsafe() -> None:
+    with patch.object(tools, "_search_images", return_value=[{"title": "local", "image": "http://[::1]/image"}]), patch.object(tools, "get_app_config") as config:
+        config.return_value.get_tool_config.return_value = None
+        result = json.loads(await tools.image_search_tool.ainvoke({"query": "reference"}))
+
+    assert result == {"error": "No safe image URLs found", "query": "reference"}
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "file:///etc/passwd",
+        "http://metadata.google.internal/computeMetadata/v1/",
+        "http://127.0.0.1/image",
+        "http://0177.0.0.1/image",
+        "http://0x7f000001/image",
+        "http://[::ffff:127.0.0.1]/image",
+        "http://[64:ff9b::7f00:1]/image",
+    ],
+)
+def test_image_search_rejects_local_result_url_spellings(url: str) -> None:
+    assert safe_public_result_url(url) == ""

backend/tests/test_image_result_url_safety.py

@@ -0,0 +1,45 @@
+"""The image providers enforce the same string-only result URL policy."""
+
+import importlib
+from unittest.mock import patch
+
+import pytest
+
+
+@pytest.fixture(params=["url_safety", "brave.tools", "serper.tools"])
+def screen(request):
+    module = importlib.import_module(f"deerflow.community.{request.param}")
+    return module.safe_public_result_url if request.param == "url_safety" else module._safe_public_url
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "http://metadata.google.internal/image",
+        "http://[64:ff9b::7f00:1]/image",
+        "http://[::ffff:127.0.0.1]/image",
+        "http://[2002:7f00:1::]/image",
+        "http://[::10.0.0.1]/image",
+        "http://224.0.0.1/image",
+        "http://3758096385/image",
+        "http://[ff02::1]/image",
+        "http://[2002:e000:1::]/image",
+        "http://240.0.0.1/image",
+        "http://127.0.0.1/image",
+        "http://0x7f000001/image",
+        "http://0177.0.0.1/image",
+        "http://localhost./image",
+        "file:///etc/passwd",
+    ],
+)
+def test_image_providers_reject_non_public_results(screen, url):
+    assert screen(url) == ""
+
+
+@pytest.mark.parametrize(
+    "url",
+    ["https://images.example/image", "http://8.8.8.8/image", "http://134744072/image", "http://[2606:4700:4700::1111]/image"],
+)
+def test_image_providers_keep_public_results_without_dns(screen, url):
+    with patch("socket.getaddrinfo", side_effect=AssertionError("Result screening must not perform DNS")):
+        assert screen(url) == url