diff --git a/src/govoplan_core/security/outbound_http.py b/src/govoplan_core/security/outbound_http.py index 868aac8..e04e95f 100644 --- a/src/govoplan_core/security/outbound_http.py +++ b/src/govoplan_core/security/outbound_http.py @@ -84,12 +84,10 @@ def validate_unpinned_sdk_http_url( """Fail closed when an SDK cannot connect to a prevalidated DNS answer.""" active_policy = policy or outbound_http_policy() - validated = validate_outbound_http_url(value, label=label, policy=active_policy) - if active_policy.allow_private_networks: - return validated + validate_outbound_http_url(value, label=label, policy=active_policy) raise OutboundHttpBlocked( f"{label} uses an SDK that cannot pin connection peers or revalidate every SDK-managed redirect; " - "it is disabled while application-enforced public-only egress is required" + "it is disabled until that transport supports connection-time DNS/IP pinning" ) @@ -163,10 +161,26 @@ def pinned_outbound_hostname( label: str = "Connector host", policy: OutboundHttpPolicy | None = None, ) -> str: - """Return the original allowed host or a public numeric connection target.""" + """Return a validated numeric target for a transport without socket injection. + + Public-only deployments may safely replace a DNS name with its validated + numeric answer. Private-network deployments fail closed for DNS names: a + protocol SDK may need the original name for authentication, so silently + replacing it would not be a generally safe pinning strategy. Explicit IP + endpoints remain usable after the normal address checks. + """ active_policy = policy or outbound_http_policy() host = str(hostname).strip().rstrip(".") + if active_policy.allow_private_networks: + try: + ipaddress.ip_address(host.split("%", 1)[0]) + except ValueError as exc: + raise OutboundHttpBlocked( + f"{label} uses a hostname with a transport that cannot pin DNS resolution while private-network " + "access is enabled; use an explicit IP endpoint or a connector transport with connection-time " + "DNS/IP pinning" + ) from exc records = _resolved_address_records(host, port=port, label=label, policy=active_policy) if active_policy.allow_private_networks: return host diff --git a/tests/test_http_fetch.py b/tests/test_http_fetch.py index b1dcbb5..890c073 100644 --- a/tests/test_http_fetch.py +++ b/tests/test_http_fetch.py @@ -14,6 +14,7 @@ from govoplan_core.security.outbound_http import ( bounded_response_bytes, create_outbound_connection, outbound_http_policy, + pinned_outbound_hostname, validate_outbound_http_url, validate_unpinned_sdk_http_url, ) @@ -135,16 +136,52 @@ class HttpFetchTests(unittest.TestCase): self.assertEqual(2, resolver.call_count) socket_factory.assert_not_called() - def test_unpinned_sdk_endpoints_fail_closed_for_dns_hosts(self) -> None: - policy = outbound_http_policy({"APP_ENV": "production"}) + def test_unpinned_sdk_endpoints_fail_closed_in_public_and_private_modes(self) -> None: + for allow_private, address in ((False, "93.184.216.34"), (True, "10.0.0.5")): + with self.subTest(allow_private=allow_private): + policy = outbound_http_policy( + { + "APP_ENV": "production", + "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": str(allow_private).lower(), + } + ) + with patch( + "govoplan_core.security.outbound_http.socket.getaddrinfo", + return_value=[(2, 1, 6, "", (address, 443))], + ), self.assertRaisesRegex(OutboundHttpBlocked, "until that transport supports.*DNS/IP pinning"): + validate_unpinned_sdk_http_url( + "https://objects.example.test", + label="S3 endpoint", + policy=policy, + ) + + def test_unpinned_hostname_transport_fails_closed_when_private_networks_are_enabled(self) -> None: + policy = outbound_http_policy( + {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"} + ) + with patch("govoplan_core.security.outbound_http.socket.getaddrinfo") as resolver, self.assertRaisesRegex( + OutboundHttpBlocked, + "cannot pin DNS resolution", + ): + pinned_outbound_hostname( + "files.internal.example", + port=445, + label="SMB endpoint", + policy=policy, + ) + resolver.assert_not_called() + + def test_explicit_private_ip_remains_a_pinned_transport_target(self) -> None: + policy = outbound_http_policy( + {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"} + ) with patch( "govoplan_core.security.outbound_http.socket.getaddrinfo", - return_value=[(2, 1, 6, "", ("93.184.216.34", 443))], - ), self.assertRaisesRegex(OutboundHttpBlocked, "cannot pin"): - validate_unpinned_sdk_http_url( - "https://objects.example.test", - label="S3 endpoint", - policy=policy, + return_value=[(2, 1, 6, "", ("10.0.0.5", 445))], + ): + self.assertEqual( + "10.0.0.5", + pinned_outbound_hostname("10.0.0.5", port=445, label="SMB endpoint", policy=policy), ) def test_core_redirects_strip_credentials_cross_origin_and_reject_https_downgrades(self) -> None: