diff --git a/src/govoplan_core/security/outbound_http.py b/src/govoplan_core/security/outbound_http.py index 0c57fec..a5e9666 100644 --- a/src/govoplan_core/security/outbound_http.py +++ b/src/govoplan_core/security/outbound_http.py @@ -100,26 +100,8 @@ def validate_outbound_host( policy: OutboundHttpPolicy | None = None, ) -> tuple[str, ...]: active_policy = policy or outbound_http_policy() - host = str(hostname).strip().rstrip(".") - if not host: - raise OutboundHttpBlocked(f"{label} must include a hostname") - if not 1 <= int(port) <= 65535: - raise OutboundHttpBlocked(f"{label} has an invalid port") - if active_policy.allow_private_networks: - return () - try: - results = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) - except socket.gaierror as exc: - raise OutboundHttpBlocked(f"{label} hostname could not be resolved") from exc - addresses = tuple(dict.fromkeys(str(item[4][0]).split("%", 1)[0] for item in results if item[4])) - if not addresses: - raise OutboundHttpBlocked(f"{label} hostname did not resolve to an address") - forbidden = [address for address in addresses if not _is_public_address(address)] - if forbidden: - raise OutboundHttpBlocked( - f"{label} resolves to a non-public network; " - "set GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS=true for this deployment to permit it" - ) + records = _resolved_address_records(hostname, port=port, label=label, policy=active_policy) + addresses = tuple(dict.fromkeys(str(item[4][0]).split("%", 1)[0] for item in records if item[4])) return addresses @@ -184,9 +166,9 @@ def pinned_outbound_hostname( active_policy = policy or outbound_http_policy() host = str(hostname).strip().rstrip(".") + records = _resolved_address_records(host, port=port, label=label, policy=active_policy) if active_policy.allow_private_networks: return host - records = _resolved_address_records(host, port=port, label=label, policy=active_policy) addresses = tuple(dict.fromkeys(str(item[4][0]).split("%", 1)[0] for item in records if item[4])) ipv4 = next((address for address in addresses if ":" not in address), None) return ipv4 or addresses[0] @@ -301,6 +283,24 @@ def _is_public_address(value: str) -> bool: return False +def _is_forbidden_special_address(value: str) -> bool: + """Keep connector access away from host-local and non-unicast address space. + + Enabling private-network connectors deliberately permits internal and + loopback destinations, but it must not expose link-local metadata services + or addresses that cannot identify a single remote peer. + """ + + try: + address = ipaddress.ip_address(value) + except ValueError: + return True + mapped = getattr(address, "ipv4_mapped", None) + if mapped is not None: + address = mapped + return address.is_link_local or address.is_multicast or address.is_unspecified + + def _resolved_address_records( hostname: str, *, @@ -320,13 +320,16 @@ def _resolved_address_records( records = tuple(results) if not records: raise OutboundHttpBlocked(f"{label} hostname did not resolve to an address") - if not policy.allow_private_networks: - addresses = tuple(str(item[4][0]).split("%", 1)[0] for item in records if item[4]) - if not addresses or any(not _is_public_address(address) for address in addresses): - raise OutboundHttpBlocked( - f"{label} resolves to a non-public network; " - "set GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS=true for this deployment to permit it" - ) + addresses = tuple(str(item[4][0]).split("%", 1)[0] for item in records if item[4]) + if not addresses: + raise OutboundHttpBlocked(f"{label} hostname did not resolve to an address") + if any(_is_forbidden_special_address(address) for address in addresses): + raise OutboundHttpBlocked(f"{label} resolves to a forbidden special-purpose network") + if not policy.allow_private_networks and any(not _is_public_address(address) for address in addresses): + raise OutboundHttpBlocked( + f"{label} resolves to a non-public network; " + "set GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS=true for this deployment to permit it" + ) return records diff --git a/tests/test_http_fetch.py b/tests/test_http_fetch.py index 112d11c..029e58a 100644 --- a/tests/test_http_fetch.py +++ b/tests/test_http_fetch.py @@ -65,10 +65,25 @@ class HttpFetchTests(unittest.TestCase): allowed_policy = outbound_http_policy( {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"} ) - self.assertEqual( - "https://127.0.0.1/path", - validate_outbound_http_url("https://127.0.0.1/path", policy=allowed_policy), + with patch( + "govoplan_core.security.outbound_http.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("127.0.0.1", 443))], + ): + self.assertEqual( + "https://127.0.0.1/path", + validate_outbound_http_url("https://127.0.0.1/path", policy=allowed_policy), + ) + + def test_private_network_policy_still_blocks_non_peer_and_metadata_addresses(self) -> None: + policy = outbound_http_policy( + {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"} ) + for address in ("0.0.0.0", "169.254.169.254", "224.0.0.1", "::", "fe80::1", "ff02::1"): + with self.subTest(address=address), patch( + "govoplan_core.security.outbound_http.socket.getaddrinfo", + return_value=[(2, 1, 6, "", (address, 443))], + ), self.assertRaisesRegex(OutboundHttpBlocked, "forbidden special-purpose network"): + validate_outbound_http_url("https://connector.example.test/path", policy=policy) def test_outbound_policy_rejects_mixed_public_and_private_dns_answers(self) -> None: policy = outbound_http_policy({"APP_ENV": "production"}) @@ -131,7 +146,10 @@ class HttpFetchTests(unittest.TestCase): headers={"Authorization": "Bearer secret", "X-Request-ID": "request-1"}, ) handler = _PolicyRedirectHandler(label="Catalog URL") - with patch.dict("os.environ", {"APP_ENV": "test"}): + with patch.dict("os.environ", {"APP_ENV": "test"}), patch( + "govoplan_core.security.outbound_http.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("127.0.0.1", 443))], + ): redirected = handler.redirect_request( request, None,