diff --git a/misp_modules/modules/expansion/html_to_markdown.py b/misp_modules/modules/expansion/html_to_markdown.py
index 55d5802b..c5637e2e 100755
--- a/misp_modules/modules/expansion/html_to_markdown.py
+++ b/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):
diff --git a/tests/test_html_to_markdown.py b/tests/test_html_to_markdown.py
index dc065317..e9747746 100644
--- a/tests/test_html_to_markdown.py
+++ b/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()