# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and contributors # For license information, please see license.txt """Pure unit tests for the SSRF guard (``http.validate_url`` / ``http._is_blocked_ip``). These are fully offline: ``socket.getaddrinfo`` is monkeypatched so DNS resolution is deterministic and no network call is ever made. The guard must reject loopback, link-local, private and reserved addresses (incl. the cloud-metadata endpoint 169.254.169.254), reject non-http(s) schemes, honor allow/block domain lists, and allow a normal public host. The private-network rejection is unconditional -- there is no setting to disable it. """ from __future__ import annotations from contextlib import contextmanager from unittest import mock from frappe.tests import UnitTestCase from crm.domain_enrichment import http from crm.domain_enrichment.http import SSRFError, validate_url from crm.domain_enrichment.tests.fixtures import make_config @contextmanager def _resolves_to(ip_map): """Patch socket.getaddrinfo so a host resolves to the given IP(s). ``ip_map`` maps hostname -> ip string (or list of ip strings). An unknown host raises gaierror, mimicking a real resolver failure. """ import socket def fake_getaddrinfo(host, *args, **kwargs): if host not in ip_map: raise socket.gaierror(f"unknown host {host}") ips = ip_map[host] if isinstance(ips, str): ips = [ips] # getaddrinfo returns 5-tuples; element 4 is the sockaddr (ip, port, ...). return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 0)) for ip in ips] with mock.patch.object(http.socket, "getaddrinfo", side_effect=fake_getaddrinfo): yield class SSRFGuardTest(UnitTestCase): def setUp(self): self.cfg = make_config() # default: no allow/block lists # -- rejection of internal / dangerous addresses ----------------------- # def test_rejects_loopback_127(self): with _resolves_to({"evil.example": "127.0.0.1"}): with self.assertRaises(SSRFError): validate_url("http://evil.example/", self.cfg) def test_rejects_cloud_metadata_link_local(self): # 169.254.169.254 is the AWS/GCP metadata endpoint — a classic SSRF target. with _resolves_to({"metadata.example": "169.254.169.254"}): with self.assertRaises(SSRFError): validate_url("http://metadata.example/latest/meta-data/", self.cfg) def test_rejects_private_10_range(self): with _resolves_to({"intranet.example": "10.0.0.5"}): with self.assertRaises(SSRFError): validate_url("http://intranet.example/", self.cfg) def test_rejects_localhost(self): with _resolves_to({"localhost": "127.0.0.1"}): with self.assertRaises(SSRFError): validate_url("http://localhost:8000/", self.cfg) def test_rejects_ipv6_loopback(self): with _resolves_to({"v6.example": "::1"}): with self.assertRaises(SSRFError): validate_url("http://v6.example/", self.cfg) # -- scheme validation ------------------------------------------------- # def test_rejects_non_http_scheme(self): with self.assertRaises(SSRFError): validate_url("ftp://example.com/x", self.cfg) with self.assertRaises(SSRFError): validate_url("file:///etc/passwd", self.cfg) with self.assertRaises(SSRFError): validate_url("gopher://example.com/", self.cfg) def test_rejects_url_without_host(self): with self.assertRaises(SSRFError): validate_url("http:///nohost", self.cfg) def test_rejects_unresolvable_host(self): with _resolves_to({}): # nothing resolves with self.assertRaises(SSRFError): validate_url("http://does-not-resolve.example/", self.cfg) # -- the happy path ---------------------------------------------------- # def test_allows_public_host(self): with _resolves_to({"frappe.io": "104.21.0.1"}): self.assertEqual(validate_url("https://frappe.io/", self.cfg), "https://frappe.io/") def test_rejects_when_any_resolved_ip_is_private(self): # DNS-rebinding style: one public + one private IP -> must reject. with _resolves_to({"mixed.example": ["8.8.8.8", "10.1.2.3"]}): with self.assertRaises(SSRFError): validate_url("http://mixed.example/", self.cfg) # -- the private-network guard is unconditional ------------------------ # def test_private_network_guard_cannot_be_disabled(self): # There is no setting to allow private networks -- a would-be bypass flag # is inert, the guard still rejects a private address. cfg = make_config(settings={"allow_private_networks": 1}) with _resolves_to({"intranet.example": "10.0.0.5"}): with self.assertRaises(SSRFError): validate_url("http://intranet.example/", cfg) # -- allow / block lists ----------------------------------------------- # def test_blocked_domain_is_always_rejected(self): cfg = make_config(blocked_domains=["evil.com"]) with _resolves_to({"evil.com": "8.8.8.8"}): with self.assertRaises(SSRFError): validate_url("http://evil.com/", cfg) # Subdomains of a blocked domain are blocked too. with _resolves_to({"sub.evil.com": "8.8.8.8"}): with self.assertRaises(SSRFError): validate_url("http://sub.evil.com/", cfg) def test_trailing_dot_fqdn_does_not_bypass_block_list(self): # evil.com. resolves the same as evil.com in DNS, so the trailing dot must not # slip past the block rule (the block check fires before resolution). cfg = make_config(blocked_domains=["evil.com"]) with self.assertRaises(SSRFError): validate_url("http://evil.com./", cfg) with self.assertRaises(SSRFError): validate_url("http://sub.evil.com./", cfg) # And a block entry that itself carries a trailing dot still matches. self.assertTrue(http._domain_in_list("evil.com", ["evil.com."])) def test_trailing_dot_host_is_normalized_for_allow_list_and_resolution(self): # The stripped host feeds both the allow-list check and DNS resolution: the # resolver only knows "good.com" (no dot), so passing "good.com." proves the # trailing dot was stripped before both steps. cfg = make_config(allowed_domains=["good.com"]) with _resolves_to({"good.com": "8.8.8.8"}): self.assertEqual(validate_url("http://good.com./", cfg), "http://good.com./") def test_allow_list_restricts_to_listed_hosts(self): cfg = make_config(allowed_domains=["good.com"]) with _resolves_to({"good.com": "8.8.8.8", "other.com": "8.8.8.8"}): self.assertEqual(validate_url("http://good.com/", cfg), "http://good.com/") with self.assertRaises(SSRFError): validate_url("http://other.com/", cfg) # -- direct IP-classification unit ------------------------------------- # def test_is_blocked_ip_classification(self): for ip in ("127.0.0.1", "10.0.0.5", "172.16.0.1", "192.168.1.1", "169.254.169.254", "::1", "0.0.0.0"): self.assertTrue(http._is_blocked_ip(ip), ip) for ip in ("8.8.8.8", "104.21.0.1", "1.1.1.1"): self.assertFalse(http._is_blocked_ip(ip), ip) # Unparseable -> rejected. self.assertTrue(http._is_blocked_ip("not-an-ip")) class PinnedConnectionTest(UnitTestCase): """DNS-rebinding defense: the socket must go to the IP the guard validated. ``_pinned_get`` rewrites the URL netloc to the validated IP and keeps the original hostname in the Host header / TLS SNI, so a resolver answer that changes after validation can no longer steer the connection. """ def _capture_send(self, captured): import requests def fake_send(adapter, request, **kwargs): captured["url"] = request.url captured["host_header"] = request.headers.get("Host") raise requests.exceptions.ConnectionError("offline test: no socket") return mock.patch.object(http._PinnedIPAdapter, "send", autospec=True, side_effect=fake_send) def test_pinned_get_connects_to_ip_with_hostname_host_header(self): import requests captured = {} with self._capture_send(captured), self.assertRaises(requests.exceptions.ConnectionError): http._pinned_get(requests.Session(), "https://example.com/about", "93.184.216.34", 5, 2) self.assertEqual(captured["url"], "https://93.184.216.34/about") self.assertEqual(captured["host_header"], "example.com") def test_pinned_get_brackets_ipv6_and_keeps_explicit_port(self): import requests captured = {} with self._capture_send(captured), self.assertRaises(requests.exceptions.ConnectionError): http._pinned_get(requests.Session(), "https://example.com:8443/x", "2606:2800:220:1::", 5, 2) self.assertEqual(captured["url"], "https://[2606:2800:220:1::]:8443/x") self.assertEqual(captured["host_header"], "example.com:8443") def test_pinned_adapter_keeps_tls_verification_on_hostname(self): adapter = http._PinnedIPAdapter("example.com") # urllib3 uses server_hostname for SNI and falls back to it for certificate # matching, so verification stays against the hostname, not the IP literal. self.assertEqual(adapter.poolmanager.connection_pool_kw.get("server_hostname"), "example.com") def test_fetch_connects_to_the_validated_ip(self): import requests captured = {} with _resolves_to({"rebind.example": "8.8.8.8"}), self._capture_send(captured): status, _html, error, _final = http.fetch("http://rebind.example/", make_config()) self.assertEqual(captured["url"], "http://8.8.8.8/") self.assertEqual(status, 0) self.assertIn("connection error", error)