From a514ff652cf40ab72116b21d22e88dad4f9f37f8 Mon Sep 17 00:00:00 2001 From: Oleksandr Kozachuk Date: Mon, 13 Jul 2026 20:09:46 +0200 Subject: [PATCH] url reader: follow html meta-refresh redirects (getnews shortlinks) with ssrf re-guard --- fjerkroa_bot/url_reader.py | 41 +++++++++++++++++------- specs/SPEC-011-url-reading.md | 5 ++- tests/test_spec_url.py | 59 +++++++++++++++++++++++++++++++++++ 3 files changed, 93 insertions(+), 12 deletions(-) diff --git a/fjerkroa_bot/url_reader.py b/fjerkroa_bot/url_reader.py index f119972..a7dddae 100644 --- a/fjerkroa_bot/url_reader.py +++ b/fjerkroa_bot/url_reader.py @@ -38,6 +38,9 @@ FETCH_URL_TOOL = { } +_META_REFRESH_URL = re.compile(r"url\s*=\s*['\"]?([^'\";\s]+)", re.I) + + class _Extractor(HTMLParser): def __init__(self) -> None: super().__init__() @@ -45,6 +48,7 @@ class _Extractor(HTMLParser): self.parts: List[str] = [] self.images: List[str] = [] self.og_image: Optional[str] = None + self.refresh_url: Optional[str] = None def handle_starttag(self, tag: str, attrs) -> None: if tag in ("script", "style", "noscript", "svg"): @@ -55,6 +59,12 @@ class _Extractor(HTMLParser): self.images.append(src) if tag == "meta" and attr.get("property") == "og:image" and attr.get("content"): self.og_image = attr["content"] + # meta-refresh redirect (link shorteners, getnews stubs) — URL-04 + content = attr.get("content") + if tag == "meta" and (attr.get("http-equiv") or "").lower() == "refresh" and content: + match = _META_REFRESH_URL.search(content) + if match and self.refresh_url is None: + self.refresh_url = match.group(1) def handle_endtag(self, tag: str) -> None: if tag in ("script", "style", "noscript", "svg") and self._skip > 0: @@ -126,29 +136,38 @@ class URLReader: try: async with aiohttp.ClientSession(timeout=timeout, headers={"User-Agent": "FjerkroaBot/1.0"}) as session: final_url, body = await self._get(session, url, max_bytes) + # follow a meta-refresh redirect (link shorteners / getnews stubs), re-guarded — URL-04 + for _ in range(2): + extractor = self._extract(body.decode("utf-8", "ignore")) + if not extractor.refresh_url: + break + target = urljoin(final_url, extractor.refresh_url) + if guard_url(target) is not None or target == final_url: + break + logging.info(f"url reader: following meta-refresh -> {target}") + final_url, body = await self._get(session, target, max_bytes) except Exception as err: return {"error": str(err)} - text = self._to_text(body.decode("utf-8", "ignore")) - clean = sanitize_external_text(text, int(config.get("url-max-chars", DEFAULT_MAX_CHARS))) - images = await self._ingest_images(body.decode("utf-8", "ignore"), final_url, channel, user) + html = body.decode("utf-8", "ignore") + clean = sanitize_external_text(self._to_text(html), int(config.get("url-max-chars", DEFAULT_MAX_CHARS))) + images = await self._ingest_images(html, final_url, channel, user) return {"url": final_url, "text": clean, "images_cached": images} - def _to_text(self, html: str) -> str: + def _extract(self, html: str) -> "_Extractor": extractor = _Extractor() try: extractor.feed(html) except Exception as err: - logging.debug(f"html parse (text) failed: {err!r}") - return re.sub(r"\s+\n", "\n", " ".join(extractor.parts)) + logging.debug(f"html parse failed: {err!r}") + return extractor + + def _to_text(self, html: str) -> str: + return re.sub(r"\s+\n", "\n", " ".join(self._extract(html).parts)) async def _ingest_images(self, html: str, base_url: str, channel: str, user: str) -> int: if self.image_cache is None: return 0 - extractor = _Extractor() - try: - extractor.feed(html) - except Exception as err: - logging.debug(f"html parse (images) failed: {err!r}") + extractor = self._extract(html) candidates = ([extractor.og_image] if extractor.og_image else []) + extractor.images limit = int(self._config().get("url-max-images", DEFAULT_MAX_IMAGES)) cached = 0 diff --git a/specs/SPEC-011-url-reading.md b/specs/SPEC-011-url-reading.md index 011438e..a2a7055 100644 --- a/specs/SPEC-011-url-reading.md +++ b/specs/SPEC-011-url-reading.md @@ -31,7 +31,10 @@ refused without DNS. Redirects are followed manually; each hop's target passes URL-02 and URL-03 again. A public URL that 302-redirects to `localhost` or an -internal IP is refused at the redirect, not fetched. +internal IP is refused at the redirect, not fetched. **HTML +meta-refresh** redirects (link shorteners, the old getnews stubs) are +also followed — the target is SSRF-re-guarded and fetched, so the +reader returns the real article, not the "Redirecting…" stub. ### URL-05 — Fetched text is bounded and sanitized (coverage: test) diff --git a/tests/test_spec_url.py b/tests/test_spec_url.py index be317b4..d4ccbb4 100644 --- a/tests/test_spec_url.py +++ b/tests/test_spec_url.py @@ -79,6 +79,65 @@ class TestRedirectRevalidation(unittest.IsolatedAsyncioTestCase): 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."""