Oleksandr Kozachuk
2caa18a17f
search_games used `where name ~ "<q>"*`: prefix-only, diacritic- and word-order-sensitive -- "MARVEL Tōkon" found nothing though IGDB has it. Switch to IGDB's native `search` clause (relevance-ranked, diacritic-insensitive). fetch_url and image downloads read bodies with content.read(n), which returns only the first buffered chunk (~7 KB): pages collapsed to their <title>. httpread.read_capped collects chunks up to the byte cap; image downloads read limit+1 so over-limit files are still rejected instead of cached truncated (IMG-10). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
184 lines
7.9 KiB
Python
184 lines
7.9 KiB
Python
"""Unit coverage for SPEC-011 URL reading (URL-01..07)."""
|
|
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from fjerkroa_bot.openai_responder import OpenAIResponder
|
|
from fjerkroa_bot.url_reader import FETCH_URL_TOOL, URLReader, guard_url
|
|
|
|
CONFIG = {"openai-token": "t", "model": "m", "system": "s", "history-limit": 5}
|
|
|
|
|
|
class TestToolOffered(unittest.TestCase):
|
|
def test_tool_present_only_when_enabled(self):
|
|
"""URL-01: fetch_url appears in the tool list only with enable-url-reading."""
|
|
off = OpenAIResponder(CONFIG, "chat")
|
|
self.assertNotIn("fetch_url", [f["name"] for f in off._available_tools()])
|
|
on = OpenAIResponder(dict(CONFIG, **{"enable-url-reading": True}), "chat")
|
|
self.assertIn("fetch_url", [f["name"] for f in on._available_tools()])
|
|
self.assertEqual(FETCH_URL_TOOL["name"], "fetch_url")
|
|
|
|
|
|
class TestSchemeGuard(unittest.TestCase):
|
|
def test_non_http_schemes_refused(self):
|
|
"""URL-02: only http/https pass the guard."""
|
|
self.assertIsNone(guard_url("https://example.com/article"))
|
|
for bad in ("file:///etc/passwd", "ftp://host/x", "data:text/html,x", "gopher://h", "no-scheme.com/x"):
|
|
self.assertIsNotNone(guard_url(bad))
|
|
|
|
|
|
class TestSSRFGuard(unittest.TestCase):
|
|
def test_private_and_loopback_refused(self):
|
|
"""URL-03: private/loopback/link-local literals are refused without DNS."""
|
|
for bad in (
|
|
"http://127.0.0.1/admin",
|
|
"http://localhost/x", # resolves to loopback
|
|
"http://10.0.0.5/x",
|
|
"http://192.168.1.1/x",
|
|
"http://169.254.169.254/latest/meta-data", # cloud metadata
|
|
"http://[::1]/x",
|
|
):
|
|
self.assertIsNotNone(guard_url(bad), f"{bad} should be refused")
|
|
|
|
def test_public_ip_allowed(self):
|
|
"""URL-03: a public IP literal passes."""
|
|
self.assertIsNone(guard_url("http://93.184.216.34/"))
|
|
|
|
@patch("fjerkroa_bot.url_reader.socket.getaddrinfo")
|
|
def test_dns_to_private_refused(self, getaddrinfo):
|
|
"""URL-03: a hostname resolving to a private IP is refused."""
|
|
getaddrinfo.return_value = [(2, 1, 6, "", ("10.1.2.3", 0))]
|
|
self.assertIsNotNone(guard_url("http://evil.example.com/x"))
|
|
|
|
|
|
class TestRedirectRevalidation(unittest.IsolatedAsyncioTestCase):
|
|
async def test_redirect_to_internal_refused(self):
|
|
"""URL-04: a public URL redirecting to localhost is refused at the hop."""
|
|
reader = URLReader(lambda: {}, None)
|
|
|
|
class FakeResp:
|
|
status = 302
|
|
headers = {"Location": "http://127.0.0.1/secret"}
|
|
url = "http://safe.example.com"
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
class FakeSession:
|
|
def get(self, url, allow_redirects=False):
|
|
return FakeResp()
|
|
|
|
with patch("fjerkroa_bot.url_reader.guard_url", side_effect=[None, "refused internal"]):
|
|
with self.assertRaises(ValueError):
|
|
await reader._get(FakeSession(), "http://safe.example.com", 1000)
|
|
|
|
|
|
class TestTextExtraction(unittest.TestCase):
|
|
def test_html_reduced_to_text(self):
|
|
"""URL-05: scripts/styles dropped, tags stripped."""
|
|
reader = URLReader(lambda: {}, None)
|
|
html = "<html><head><style>x{}</style></head><body><h1>Titel</h1><script>evil()</script><p>Inhalt hier</p></body></html>"
|
|
text = reader._to_text(html)
|
|
self.assertIn("Titel", text)
|
|
self.assertIn("Inhalt hier", text)
|
|
self.assertNotIn("evil", text)
|
|
self.assertNotIn("x{}", text)
|
|
|
|
|
|
class TestBodyReadCollectsAllChunks(unittest.IsolatedAsyncioTestCase):
|
|
async def test_get_reads_past_first_chunk(self):
|
|
"""URL-05 regression: body arrives in many chunks; all are collected up to the cap."""
|
|
reader = URLReader(lambda: {}, None)
|
|
chunks = [b"<title>t</title>", b"<p>middle</p>", b"<p>end</p>"]
|
|
|
|
class FakeContent:
|
|
@staticmethod
|
|
async def iter_chunked(size):
|
|
for chunk in chunks:
|
|
yield chunk
|
|
|
|
class FakeResp:
|
|
status = 200
|
|
headers = {}
|
|
url = "http://safe.example.com"
|
|
content = FakeContent()
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
class FakeSession:
|
|
def get(self, url, allow_redirects=False):
|
|
return FakeResp()
|
|
|
|
with patch("fjerkroa_bot.url_reader.guard_url", return_value=None):
|
|
_, body = await reader._get(FakeSession(), "http://safe.example.com", 1000)
|
|
self.assertEqual(body, b"".join(chunks))
|
|
|
|
with patch("fjerkroa_bot.url_reader.guard_url", return_value=None):
|
|
_, body = await reader._get(FakeSession(), "http://safe.example.com", 20)
|
|
self.assertEqual(body, b"".join(chunks)[:20])
|
|
|
|
|
|
class TestFetchSanitizes(unittest.IsolatedAsyncioTestCase):
|
|
async def test_fetch_result_is_sanitized_and_capped(self):
|
|
"""URL-05: fetch output is length-capped and @everyone-neutralized."""
|
|
reader = URLReader(lambda: {"url-max-chars": 50}, None)
|
|
payload = ("<p>@everyone " + "x" * 5000 + "</p>").encode()
|
|
with patch.object(reader, "_get", new=AsyncMock(return_value=("http://x.com", payload))):
|
|
result = await reader.fetch("http://x.com", "chat", "alice")
|
|
self.assertLessEqual(len(result["text"]), 50)
|
|
self.assertNotIn("@everyone", result["text"])
|
|
|
|
async def test_fetch_error_is_reported_not_raised(self):
|
|
"""URL-05: a fetch failure returns an error dict the model can relay."""
|
|
reader = URLReader(lambda: {}, None)
|
|
with patch.object(reader, "_get", new=AsyncMock(side_effect=ValueError("refused non-public address"))):
|
|
result = await reader.fetch("http://10.0.0.1", "chat", "alice")
|
|
self.assertIn("error", result)
|
|
|
|
|
|
class TestImageIngest(unittest.IsolatedAsyncioTestCase):
|
|
async def test_page_images_go_to_cache_ssrf_guarded(self):
|
|
"""URL-06: og:image + <img> ingested (cap honored), internal srcs skipped."""
|
|
cache = type("C", (), {})()
|
|
cache.ingest_url = AsyncMock(side_effect=["sha1", "sha2", "sha3"])
|
|
reader = URLReader(lambda: {"url-max-images": 2}, cache)
|
|
html = (
|
|
'<meta property="og:image" content="https://cdn.example.com/hero.jpg">'
|
|
'<img src="https://cdn.example.com/a.png"><img src="http://127.0.0.1/internal.png">'
|
|
)
|
|
|
|
# guard by scheme/loopback only, no real DNS in the test
|
|
def fake_guard(url):
|
|
return "refused" if "127.0.0.1" in url else None
|
|
|
|
with patch("fjerkroa_bot.url_reader.guard_url", side_effect=fake_guard):
|
|
count = await reader._ingest_images(html, "https://example.com", "chat", "alice")
|
|
self.assertEqual(count, 2) # og:image + first public img, cap 2
|
|
ingested = [call.args[0] for call in cache.ingest_url.await_args_list]
|
|
self.assertNotIn("http://127.0.0.1/internal.png", ingested)
|
|
|
|
|
|
class TestPerUserCap(unittest.IsolatedAsyncioTestCase):
|
|
async def test_dispatch_caps_fetches(self):
|
|
"""URL-07: over url-daily-per-user, fetch_url returns an error without fetching."""
|
|
responder = OpenAIResponder(dict(CONFIG, **{"enable-url-reading": True, "url-daily-per-user": 2}), "chat")
|
|
responder.url_reader.fetch = AsyncMock(return_value={"url": "x", "text": "ok"})
|
|
for _ in range(2):
|
|
await responder._dispatch_tool("fetch_url", {"url": "http://x.com"}, "alice")
|
|
blocked = await responder._dispatch_tool("fetch_url", {"url": "http://x.com"}, "alice")
|
|
self.assertIn("error", blocked)
|
|
self.assertEqual(responder.url_reader.fetch.await_count, 2)
|