Harden private connector address validation

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,9 +320,12 @@ 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):
+    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"

tests/test_http_fetch.py

@@ -65,11 +65,26 @@ class HttpFetchTests(unittest.TestCase):
         allowed_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, "", ("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"})
         with patch(
@@ -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,