upstream-frappe-crm/crm/domain_enrichment/tests/test_ssrf.py

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