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()