fix: prevent HTML-to-Markdown URL parser SSRF bypass

misp_modules/modules/expansion/html_to_markdown.py

@@ -3,7 +3,7 @@
 
 import requests
 import ipaddress
-from urllib.parse import urlparse
+from urllib.parse import urljoin, urlparse
 from bs4 import BeautifulSoup
 from markdownify import markdownify
 
@@ -35,6 +35,7 @@
     ipaddress.ip_network("169.254.0.0/16"),
     ipaddress.ip_network("::1/128"),
 ]
+MAX_REDIRECTS = 10
 
 
 def _normalize_ip_address(ip_str: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
@@ -55,21 +56,53 @@ def _hostname_resolves_to_blocked_ip(hostname: str) -> bool:
         return True
 
 
-def is_safe_url(url: str) -> bool:
-    parsed = urlparse(url)
-    if parsed.scheme not in ("http", "https") or not parsed.hostname:
+def _canonicalize_url(url: str) -> str | None:
+    """Return the URL exactly as requests will send it, rejecting ambiguous input."""
+    if not isinstance(url, str) or "\\" in url:
+        return None
+    try:
+        return requests.Request("GET", url).prepare().url
+    except requests.RequestException:
+        return None
+
+
+def _is_safe_canonical_url(url: str) -> bool:
+    try:
+        parsed = urlparse(url)
+        hostname = parsed.hostname
+    except ValueError:
+        return False
+    if parsed.scheme not in ("http", "https") or not hostname:
         return False
     try:
-        return not _is_ip_blocked(parsed.hostname)
+        return not _is_ip_blocked(hostname)
     except ValueError:
-        return not _hostname_resolves_to_blocked_ip(parsed.hostname)
+        return not _hostname_resolves_to_blocked_ip(hostname)
+
+
+def is_safe_url(url: str) -> bool:
+    canonical_url = _canonicalize_url(url)
+    return canonical_url is not None and _is_safe_canonical_url(canonical_url)
 
 
 def fetchHTML(url):
-    if not is_safe_url(url):
+    canonical_url = _canonicalize_url(url)
+    if canonical_url is None or not _is_safe_canonical_url(canonical_url):
         raise ValueError(f"Blocked URL: {url}")
-    r = requests.get(url, timeout=10)
-    return r.text
+
+    with requests.Session() as session:
+        for _ in range(MAX_REDIRECTS + 1):
+            response = session.get(canonical_url, timeout=10, allow_redirects=False)
+            if not response.is_redirect:
+                return response.text
+
+            redirected_url = _canonicalize_url(urljoin(canonical_url, response.headers["location"]))
+            response.close()
+            if redirected_url is None or not _is_safe_canonical_url(redirected_url):
+                raise ValueError(f"Blocked redirect URL: {redirected_url}")
+            canonical_url = redirected_url
+
+    raise ValueError(f"Too many redirects for URL: {url}")
 
 
 def stripUselessTags(html):

tests/test_html_to_markdown.py

@@ -2,9 +2,9 @@
 # -*- coding: utf-8 -*-
 
 import unittest
-from unittest.mock import patch
+from unittest.mock import Mock, call, patch
 
-from misp_modules.modules.expansion.html_to_markdown import is_safe_url
+from misp_modules.modules.expansion.html_to_markdown import fetchHTML, is_safe_url
 
 
 class TestHtmlToMarkdownUrlSafety(unittest.TestCase):
@@ -45,6 +45,57 @@ def test_blocks_hostnames_resolving_to_ipv4_mapped_blocked_addresses(self):
     def test_rejects_url_without_hostname(self):
         self.assertFalse(is_safe_url("http:///missing-host"))
 
+    def test_rejects_backslash_authority_parser_differential(self):
+        self.assertFalse(is_safe_url(r"http://127.0.0.1:8877\@1.1.1.1/"))
+
+    @patch("misp_modules.modules.expansion.html_to_markdown.requests.Session")
+    def test_parser_differential_is_blocked_before_request(self, session_factory):
+        with self.assertRaisesRegex(ValueError, "Blocked URL"):
+            fetchHTML(r"http://127.0.0.1:8877\@1.1.1.1/")
+        session_factory.assert_not_called()
+
+    @patch("misp_modules.modules.expansion.html_to_markdown.requests.Session")
+    def test_blocks_redirect_to_loopback(self, session_factory):
+        session = session_factory.return_value.__enter__.return_value
+        response = Mock()
+        response.is_redirect = True
+        response.headers = {"location": "http://127.0.0.1/internal"}
+        session.get.return_value = response
+
+        with self.assertRaisesRegex(ValueError, "Blocked redirect URL"):
+            fetchHTML("http://93.184.216.34/")
+
+        session.get.assert_called_once_with(
+            "http://93.184.216.34/", timeout=10, allow_redirects=False
+        )
+        response.close.assert_called_once_with()
+
+    @patch("misp_modules.modules.expansion.html_to_markdown.requests.Session")
+    def test_reuses_session_when_following_safe_redirect(self, session_factory):
+        session = session_factory.return_value.__enter__.return_value
+        redirect_response = Mock(
+            is_redirect=True,
+            headers={"location": "http://93.184.216.35/destination"},
+        )
+        final_response = Mock(is_redirect=False, text="redirected content")
+        session.get.side_effect = [redirect_response, final_response]
+
+        self.assertEqual(fetchHTML("http://93.184.216.34/"), "redirected content")
+
+        session_factory.assert_called_once_with()
+        self.assertEqual(
+            session.get.call_args_list,
+            [
+                call("http://93.184.216.34/", timeout=10, allow_redirects=False),
+                call(
+                    "http://93.184.216.35/destination",
+                    timeout=10,
+                    allow_redirects=False,
+                ),
+            ],
+        )
+        redirect_response.close.assert_called_once_with()
+
 
 if __name__ == "__main__":
     unittest.main()