GovOPlaN/govoplan-core
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

Fail closed for unpinned connector transports

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 15:36:32 +02:00
parent ae74189588
commit 28a0a596a6
2 changed files with 64 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
tests/test_http_fetch.py
View File

@@ -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: