url reader: follow html meta-refresh redirects (getnews shortlinks) with ssrf re-guard

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"))
+        html = body.decode("utf-8", "ignore")
-        clean = sanitize_external_text(text, int(config.get("url-max-chars", DEFAULT_MAX_CHARS)))
+        clean = sanitize_external_text(self._to_text(html), int(config.get("url-max-chars", DEFAULT_MAX_CHARS)))
-        images = await self._ingest_images(body.decode("utf-8", "ignore"), final_url, channel, user)
+        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}")
+            logging.debug(f"html parse failed: {err!r}")
-        return re.sub(r"\s+\n", "\n", " ".join(extractor.parts))
+        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()
+        extractor = self._extract(html)
-        try:
-            extractor.feed(html)
-        except Exception as err:
-            logging.debug(f"html parse (images) failed: {err!r}")
         candidates = ([extractor.og_image] if extractor.og_image else []) + extractor.images
         limit = int(self._config().get("url-max-images", DEFAULT_MAX_IMAGES))
         cached = 0

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)
 

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'<html><head><meta http-equiv="refresh" content="0;url=https://pushsquare.com/real"></head><body>Redirecting...</body></html>'
+        )
+        article = b"<html><body><h1>MARVEL Tokon</h1><p>Full article text here</p></body></html>"
+        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'<meta http-equiv="refresh" content="0; url=http://127.0.0.1/secret">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."""