fix(web): re-check Firecrawl final URLs for SSRF

This commit is contained in:
zapabob 2026-05-31 20:58:09 +09:00 committed by Teknium
parent 7136b5382a
commit 2e12401ed4
2 changed files with 67 additions and 0 deletions

View file

@ -51,6 +51,7 @@ import os
from typing import Any, Dict, List, Optional, TYPE_CHECKING
from agent.web_search_provider import WebSearchProvider
from tools.url_safety import is_safe_url
from tools.website_policy import check_website_access
logger = logging.getLogger(__name__)
@ -523,6 +524,26 @@ class FirecrawlWebSearchProvider(WebSearchProvider):
title = metadata.get("title", "")
final_url = metadata.get("sourceURL", url)
# Re-check SSRF safety after any redirect reported by Firecrawl.
if not is_safe_url(final_url):
logger.info(
"Blocked redirected web_extract for unsafe final URL: %s",
final_url,
)
results.append(
{
"url": final_url,
"title": title,
"content": "",
"raw_content": "",
"error": (
"Blocked: URL targets a private or internal "
"network address"
),
}
)
continue
# Re-check website-access policy after any redirect
final_blocked = check_website_access(final_url)
if final_blocked:

View file

@ -413,6 +413,7 @@ class TestWebToolPolicy:
return True
monkeypatch.setattr(web_tools, "async_is_safe_url", _allow_ssrf)
monkeypatch.setattr(firecrawl_provider, "is_safe_url", lambda url: True)
def fake_check(url):
if url == "https://allowed.test":
@ -449,6 +450,51 @@ class TestWebToolPolicy:
assert result["results"][0]["content"] == ""
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
@pytest.mark.asyncio
async def test_web_extract_blocks_firecrawl_unsafe_final_url(self, monkeypatch):
from tools import web_tools
from plugins.web.firecrawl import provider as firecrawl_provider
async def _allow_ssrf(_url: str) -> bool:
return True
monkeypatch.setattr(web_tools, "async_is_safe_url", _allow_ssrf)
monkeypatch.setattr(
firecrawl_provider,
"is_safe_url",
lambda url: url != "http://169.254.169.254/latest/meta-data/",
)
checked_urls = []
def fake_check(url):
checked_urls.append(url)
if url == "https://allowed.test":
return None
pytest.fail(f"unexpected website policy check for unsafe URL: {url}")
class FakeFirecrawlClient:
def scrape(self, url, formats):
return {
"markdown": "metadata credentials",
"metadata": {
"title": "Metadata",
"sourceURL": "http://169.254.169.254/latest/meta-data/",
},
}
monkeypatch.setattr(firecrawl_provider, "check_website_access", fake_check)
monkeypatch.setattr(firecrawl_provider, "_get_firecrawl_client", lambda: FakeFirecrawlClient())
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
result = json.loads(await web_tools.web_extract_tool(["https://allowed.test"], use_llm_processing=False))
assert checked_urls == ["https://allowed.test"]
assert result["results"][0]["url"] == "http://169.254.169.254/latest/meta-data/"
assert result["results"][0]["content"] == ""
assert "private or internal network" in result["results"][0]["error"]
def test_check_website_access_fails_open_on_malformed_config(tmp_path, monkeypatch):
"""Malformed config with default path should fail open (return None), not crash."""