From 0845e99cf6789ffdd8970f342dbbec243dadee18 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 15:43:04 +0200 Subject: [PATCH] Block SDK-managed secondary connector peers --- src/govoplan_core/security/outbound_http.py | 55 +++++--------- tests/test_api_smoke.py | 80 +++------------------ tests/test_http_fetch.py | 34 +++------ 3 files changed, 35 insertions(+), 134 deletions(-) diff --git a/src/govoplan_core/security/outbound_http.py b/src/govoplan_core/security/outbound_http.py index e04e95f..976cf98 100644 --- a/src/govoplan_core/security/outbound_http.py +++ b/src/govoplan_core/security/outbound_http.py @@ -85,8 +85,26 @@ def validate_unpinned_sdk_http_url( active_policy = policy or outbound_http_policy() validate_outbound_http_url(value, label=label, policy=active_policy) + _raise_unpinned_transport(label) + + +def validate_unpinned_sdk_host( + hostname: str, + *, + port: int, + label: str, + policy: OutboundHttpPolicy | None = None, +) -> None: + """Validate a host and then reject an SDK that may select another peer.""" + + active_policy = policy or outbound_http_policy() + validate_outbound_host(hostname, port=port, label=label, policy=active_policy) + _raise_unpinned_transport(label) + + +def _raise_unpinned_transport(label: str) -> None: raise OutboundHttpBlocked( - f"{label} uses an SDK that cannot pin connection peers or revalidate every SDK-managed redirect; " + f"{label} uses an SDK that cannot pin every connection peer or revalidate SDK-managed redirects/referrals; " "it is disabled until that transport supports connection-time DNS/IP pinning" ) @@ -154,41 +172,6 @@ def create_outbound_connection( raise OutboundHttpBlocked(f"{label} hostname did not resolve to a usable address") -def pinned_outbound_hostname( - hostname: str, - *, - port: int, - label: str = "Connector host", - policy: OutboundHttpPolicy | None = None, -) -> str: - """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 - 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] - - def build_outbound_http_opener(*handlers: urllib.request.BaseHandler) -> urllib.request.OpenerDirector: """Build a proxy-free urllib opener whose sockets use approved addresses.""" diff --git a/tests/test_api_smoke.py b/tests/test_api_smoke.py index 0412ef0..514d57e 100644 --- a/tests/test_api_smoke.py +++ b/tests/test_api_smoke.py @@ -2341,74 +2341,13 @@ class ApiSmokeTests(unittest.TestCase): self.assertEqual(unchanged["file"]["id"], nextcloud_file["id"]) self.assertEqual(unchanged["current_version_id"], synced["current_version_id"]) - case = self - - class _FakeSmbStat: - st_size = 18 - st_mtime = 1783425600 - st_mtime_ns = 1783425600000000000 - - class _FakeSmbEntry: - def __init__(self, name: str, is_dir: bool, size: int = 18) -> None: - self.name = name - self._is_dir = is_dir - self._stat = _FakeSmbStat() - self._stat.st_size = size - - def is_dir(self) -> bool: - return self._is_dir - - def stat(self): - return self._stat - - class _FakeSmbScandir: - def __enter__(self): - return iter([ - _FakeSmbEntry("Archive", True, 0), - _FakeSmbEntry("smb-notice.txt", False, 18), - ]) - - def __exit__(self, exc_type, exc, traceback): - return False - - class _FakeSmbFile: - def __enter__(self): - return self - - def __exit__(self, exc_type, exc, traceback): - return False - - def read(self, size: int) -> bytes: - case.assertGreater(size, 18) - return b"smb connector data" - - class _FakeSmbClient: - def scandir(self, path: str, **kwargs): - case.assertEqual(path, r"\\127.0.0.1\files\root\shared") - case.assertEqual(kwargs["port"], 1445) - case.assertEqual(kwargs["username"], "govoplan") - case.assertEqual(kwargs["password"], "super-secret-smb") - case.assertTrue(kwargs["require_signing"]) - return _FakeSmbScandir() - - def stat(self, path: str, **kwargs): - case.assertEqual(path, r"\\127.0.0.1\files\root\shared\smb-notice.txt") - case.assertEqual(kwargs["port"], 1445) - return _FakeSmbStat() - - def open_file(self, path: str, mode: str, **kwargs): - case.assertEqual(path, r"\\127.0.0.1\files\root\shared\smb-notice.txt") - case.assertEqual(mode, "rb") - case.assertEqual(kwargs["password"], "super-secret-smb") - return _FakeSmbFile() - - fake_smb = _FakeSmbClient() - with patch("govoplan_files.backend.storage.connector_browse._smbclient_module", return_value=fake_smb): + with patch("govoplan_files.backend.storage.connector_browse._smbclient_module") as smb_sdk: smb_browse = self.client.get("/api/v1/files/connectors/profiles/tenant-smb/browse?path=shared", headers=headers) - self.assertEqual(smb_browse.status_code, 200, smb_browse.text) - self.assertEqual([(item["kind"], item["path"]) for item in smb_browse.json()["items"]], [("folder", "shared/Archive"), ("file", "shared/smb-notice.txt")]) + self.assertEqual(smb_browse.status_code, 400, smb_browse.text) + self.assertIn("redirects/referrals", smb_browse.json()["detail"]) + smb_sdk.assert_not_called() - with patch("govoplan_files.backend.storage.connector_imports._smbclient_module", return_value=fake_smb): + with patch("govoplan_files.backend.storage.connector_imports._smbclient_module") as smb_import_sdk: smb_import = self.client.post( "/api/v1/files/connectors/profiles/tenant-smb/import", headers=headers, @@ -2420,12 +2359,9 @@ class ApiSmokeTests(unittest.TestCase): "target_folder": "imports", }, ) - self.assertEqual(smb_import.status_code, 200, smb_import.text) - smb_file = smb_import.json()["files"][0] - self.assertEqual(smb_file["display_path"], "imports/smb-notice.txt") - self.assertEqual(smb_file["source_provenance"]["provider"], "smb") - self.assertEqual(smb_file["source_provenance"]["external_id"], "files:shared/smb-notice.txt") - self.assertEqual(smb_file["source_provenance"]["metadata"]["share"], "files") + self.assertEqual(smb_import.status_code, 400, smb_import.text) + self.assertIn("redirects/referrals", smb_import.json()["detail"]) + smb_import_sdk.assert_not_called() filtered = self.client.get("/api/v1/files/connectors/profiles?provider=smb", headers=headers) self.assertEqual(filtered.status_code, 200, filtered.text) diff --git a/tests/test_http_fetch.py b/tests/test_http_fetch.py index 890c073..2d46a75 100644 --- a/tests/test_http_fetch.py +++ b/tests/test_http_fetch.py @@ -14,8 +14,8 @@ 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_host, validate_unpinned_sdk_http_url, ) @@ -155,34 +155,16 @@ class HttpFetchTests(unittest.TestCase): policy=policy, ) - def test_unpinned_hostname_transport_fails_closed_when_private_networks_are_enabled(self) -> None: + def test_unpinned_host_sdk_fails_closed_for_dns_and_explicit_ip_targets(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, "", ("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), - ) + for host in ("files.internal.example", "10.0.0.5"): + with self.subTest(host=host), patch( + "govoplan_core.security.outbound_http.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("10.0.0.5", 445))], + ), self.assertRaisesRegex(OutboundHttpBlocked, "redirects/referrals.*DNS/IP pinning"): + validate_unpinned_sdk_host(host, port=445, label="SMB endpoint", policy=policy) def test_core_redirects_strip_credentials_cross_origin_and_reject_https_downgrades(self) -> None: import urllib.request