"""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 TestMetaRefresh(unittest.IsolatedAsyncioTestCase): async def test_follows_meta_refresh_to_real_article(self): """URL-04: a getnews-style meta-refresh stub is followed to the real article.""" reader = URLReader(lambda: {}, None) stub = ( b'Redirecting...' ) article = b"

MARVEL Tokon

Full article text here

" calls = [] async def fake_get(session, url, max_bytes): calls.append(url) return (url, stub if "stub" in url else article) reader._get = fake_get # type: ignore with patch("fjerkroa_bot.url_reader.guard_url", return_value=None): import fjerkroa_bot.url_reader as ur # patch the session context so fetch() runs against fake_get class FakeCM: async def __aenter__(self): return object() async def __aexit__(self, *a): return False with patch.object(ur.aiohttp, "ClientSession", return_value=FakeCM()): result = await reader.fetch("https://gggemein.de/url/stub.html", "chat", "alice") self.assertIn("Full article text", result["text"]) self.assertEqual(result["url"], "https://pushsquare.com/real") self.assertIn("https://pushsquare.com/real", calls) async def test_meta_refresh_to_internal_is_not_followed(self): """URL-04: a meta-refresh pointing at an internal IP is refused (SSRF).""" reader = URLReader(lambda: {}, None) stub = b'Redirecting' async def fake_get(session, url, max_bytes): return (url, stub) reader._get = fake_get # type: ignore import fjerkroa_bot.url_reader as ur class FakeCM: async def __aenter__(self): return object() async def __aexit__(self, *a): return False def guard(u): return "refused" if "127.0.0.1" in u else None with patch("fjerkroa_bot.url_reader.guard_url", side_effect=guard): with patch.object(ur.aiohttp, "ClientSession", return_value=FakeCM()): result = await reader.fetch("https://safe.com/x", "chat", "alice") self.assertEqual(result["url"], "https://safe.com/x") # did not follow to 127.0.0.1 class TestTextExtraction(unittest.TestCase): def test_html_reduced_to_text(self): """URL-05: scripts/styles dropped, tags stripped.""" reader = URLReader(lambda: {}, None) html = "

Titel

Inhalt hier

" 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"t", b"

middle

", b"

end

"] 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 = ("

@everyone " + "x" * 5000 + "

").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 + 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 = ( '' '' ) # 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)