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

Fail closed for SMB referral transports

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 15:43:56 +02:00
parent d5d0df792b
commit 5248e7de4a
7 changed files with 58 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
tests/test_connector_providers.py
View File

@@ -5,7 +5,7 @@ from datetime import UTC, datetime
from unittest.mock import patch
from govoplan_files.backend.storage.connector_browse import ConnectorBrowseError, _smb_location, browse_connector_profile
from govoplan_files.backend.storage.connector_imports import read_connector_file
from govoplan_files.backend.storage.connector_imports import ConnectorImportError, read_connector_file
from govoplan_files.backend.storage.connector_profiles import ConnectorProfile, connector_profiles_from_payload
from govoplan_files.backend.storage.connector_providers import connector_provider_descriptors
@@ -75,58 +75,55 @@ def s3_profile(**overrides: object) -> ConnectorProfile:
class ConnectorProviderTests(unittest.TestCase):
def test_smb_uses_validated_numeric_target_when_private_networks_are_disabled(self) -> None:
def test_smb_sdk_transport_fails_closed_before_client_creation_in_all_modes(self) -> None:
profile = ConnectorProfile(
id="public-smb",
label="Public SMB",
id="smb",
label="SMB",
provider="smb",
endpoint_url="smb://files.example.test/share",
)
with patch.dict(
"os.environ",
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
), patch(
"govoplan_core.security.outbound_http.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 445))],
):
location = _smb_location(profile)
for allow_private, address in ((False, "93.184.216.34"), (True, "10.0.0.5")):
with self.subTest(allow_private=allow_private), patch.dict(
"os.environ",
{
"APP_ENV": "production",
"GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": str(allow_private).lower(),
},
), patch(
"govoplan_core.security.outbound_http.socket.getaddrinfo",
return_value=[(2, 1, 6, "", (address, 445))],
), patch("govoplan_files.backend.storage.connector_browse._smbclient_module") as sdk, self.assertRaisesRegex(
ConnectorBrowseError,
"redirects/referrals.*DNS/IP pinning",
):
browse_connector_profile(profile, path="")
sdk.assert_not_called()
self.assertEqual("93.184.216.34", location.server)
def test_smb_dns_endpoint_fails_closed_when_private_networks_are_enabled(self) -> None:
profile = ConnectorProfile(
id="private-smb",
label="Private SMB",
provider="smb",
endpoint_url="smb://files.internal.example/share",
)
with patch.dict(
"os.environ",
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"},
), patch("govoplan_core.security.outbound_http.socket.getaddrinfo") as resolver, self.assertRaisesRegex(
ConnectorBrowseError,
"cannot pin DNS resolution",
):
_smb_location(profile)
resolver.assert_not_called()
def test_smb_explicit_private_ip_is_a_pinned_target(self) -> None:
profile = ConnectorProfile(
id="private-smb",
label="Private SMB",
provider="smb",
endpoint_url="smb://10.0.0.5/share",
)
def test_smb_explicit_ip_still_fails_closed_because_the_sdk_may_follow_referrals(self) -> None:
profile = ConnectorProfile(id="smb", label="SMB", provider="smb", endpoint_url="smb://10.0.0.5/share")
with patch.dict(
"os.environ",
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"},
), patch(
"govoplan_core.security.outbound_http.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("10.0.0.5", 445))],
):
location = _smb_location(profile)
), self.assertRaisesRegex(ConnectorBrowseError, "redirects/referrals.*DNS/IP pinning"):
_smb_location(profile)
self.assertEqual("10.0.0.5", location.server)
def test_smb_import_surfaces_fail_closed_policy_as_an_import_error(self) -> None:
profile = ConnectorProfile(id="smb", label="SMB", provider="smb", endpoint_url="smb://10.0.0.5/share")
with patch.dict(
"os.environ",
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"},
), patch(
"govoplan_core.security.outbound_http.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("10.0.0.5", 445))],
), patch("govoplan_files.backend.storage.connector_imports._smbclient_module") as sdk, self.assertRaisesRegex(
ConnectorImportError,
"redirects/referrals.*DNS/IP pinning",
):
read_connector_file(profile, library_id="", path="notice.txt", max_bytes=1024)
sdk.assert_not_called()
def test_provider_descriptors_include_s3_and_reserved_microsoft_providers(self) -> None:
descriptors = {descriptor.provider: descriptor for descriptor in connector_provider_descriptors()}