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

Block SDK-managed secondary connector peers

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 15:43:04 +02:00
parent 28a0a596a6
commit 0845e99cf6
3 changed files with 35 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
tests/test_api_smoke.py
View File

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

34
tests/test_http_fetch.py
View File

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