Pin S3 and SMB connector peers

tests/test_connector_providers.py

@@ -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 ConnectorImportError, read_connector_file
+from govoplan_files.backend.storage.connector_imports import 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
 
@@ -23,6 +23,10 @@ class FakeS3Client:
         self.list_objects_request: dict[str, object] | None = None
         self.head_request: dict[str, object] | None = None
         self.get_request: dict[str, object] | None = None
+        self.closed = False
+
+    def close(self) -> None:
+        self.closed = True
 
     def list_objects_v2(self, **kwargs: object) -> dict[str, object]:
         self.list_objects_request = dict(kwargs)
@@ -75,42 +79,39 @@ def s3_profile(**overrides: object) -> ConnectorProfile:
 
 
 class ConnectorProviderTests(unittest.TestCase):
-    def test_smb_sdk_transport_fails_closed_before_client_creation_in_all_modes(self) -> None:
+    def test_smb_browse_uses_the_files_owned_pinned_connection_cache(self) -> None:
         profile = ConnectorProfile(
             id="smb",
             label="SMB",
             provider="smb",
             endpoint_url="smb://files.example.test/share",
         )
-        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()
+        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))],
+        ), patch("govoplan_files.backend.storage.connector_browse._smbclient_module") as sdk:
+            sdk.return_value.scandir.return_value.__enter__.return_value = iter(())
+            self.assertEqual([], browse_connector_profile(profile, path=""))
 
-    def test_smb_explicit_ip_still_fails_closed_because_the_sdk_may_follow_referrals(self) -> None:
+        kwargs = sdk.return_value.scandir.call_args.kwargs
+        self.assertIsInstance(kwargs["connection_cache"], dict)
+        self.assertTrue(kwargs["require_signing"])
+
+    def test_smb_endpoint_preflight_applies_private_network_policy(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"},
+            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
         ), patch(
             "govoplan_core.security.outbound_http.socket.getaddrinfo",
             return_value=[(2, 1, 6, "", ("10.0.0.5", 445))],
-        ), self.assertRaisesRegex(ConnectorBrowseError, "redirects/referrals.*DNS/IP pinning"):
+        ), self.assertRaisesRegex(ConnectorBrowseError, "non-public network"):
             _smb_location(profile)
 
-    def test_smb_import_surfaces_fail_closed_policy_as_an_import_error(self) -> None:
+    def test_smb_import_uses_the_same_pinned_connection_cache(self) -> None:
         profile = ConnectorProfile(id="smb", label="SMB", provider="smb", endpoint_url="smb://10.0.0.5/share")
         with patch.dict(
             "os.environ",
@@ -118,12 +119,13 @@ class ConnectorProviderTests(unittest.TestCase):
         ), 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()
+        ), patch("govoplan_files.backend.storage.connector_imports._smbclient_module") as sdk:
+            sdk.return_value.stat.return_value.st_size = 4
+            sdk.return_value.open_file.return_value.__enter__.return_value.read.return_value = b"test"
+            downloaded = read_connector_file(profile, library_id="", path="notice.txt", max_bytes=1024)
+
+        self.assertEqual(b"test", downloaded.data)
+        self.assertIsInstance(sdk.return_value.stat.call_args.kwargs["connection_cache"], dict)
 
     def test_provider_descriptors_include_s3_and_reserved_microsoft_providers(self) -> None:
         descriptors = {descriptor.provider: descriptor for descriptor in connector_provider_descriptors()}
@@ -172,6 +174,7 @@ class ConnectorProviderTests(unittest.TestCase):
         self.assertEqual("govoplan:root/report.xlsx", items[1].external_id)
         self.assertEqual("root/report.xlsx", items[1].metadata["key"])
         self.assertEqual("next-page", items[1].metadata["next_continuation_token"])
+        self.assertTrue(client.closed)
 
     def test_s3_browse_lists_buckets_when_profile_has_no_bucket(self) -> None:
         client = FakeS3Client()
@@ -181,27 +184,33 @@ class ConnectorProviderTests(unittest.TestCase):
 
         self.assertEqual(["archive"], [item.path for item in items])
 
-    def test_s3_sdk_transport_fails_closed_before_client_creation_in_private_mode(self) -> None:
+    def test_s3_client_is_constructed_through_the_pinned_transport(self) -> None:
         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, "", ("127.0.0.1", 9000))],
-        ), patch("govoplan_files.backend.storage.connector_browse.import_module") as importer, self.assertRaisesRegex(
-            ConnectorBrowseError,
-            "until that transport supports.*DNS/IP pinning",
-        ):
+        ), patch(
+            "govoplan_files.backend.storage.connector_browse.create_pinned_s3_client",
+            return_value=FakeS3Client(),
+        ) as factory:
             browse_connector_profile(s3_profile(), path="")
-        importer.assert_not_called()
 
-    def test_s3_sdk_endpoint_discovery_fails_closed(self) -> None:
-        with patch("govoplan_files.backend.storage.connector_browse.import_module") as importer, self.assertRaisesRegex(
-            ConnectorBrowseError,
-            "endpoint discovery.*cannot guarantee.*DNS/IP pinning",
-        ):
+        kwargs = factory.call_args.kwargs
+        self.assertEqual("http://127.0.0.1:9000", kwargs["endpoint_url"])
+        self.assertEqual("access-key", kwargs["aws_access_key_id"])
+        self.assertEqual("secret-key", kwargs["aws_secret_access_key"])
+        self.assertEqual({}, kwargs["config"].proxies)
+
+    def test_s3_endpoint_discovery_uses_the_same_pinned_transport(self) -> None:
+        with patch(
+            "govoplan_files.backend.storage.connector_browse.create_pinned_s3_client",
+            return_value=FakeS3Client(),
+        ) as factory:
             browse_connector_profile(s3_profile(endpoint_url=None), path="")
-        importer.assert_not_called()
+
+        self.assertNotIn("endpoint_url", factory.call_args.kwargs)
 
     def test_s3_import_downloads_object_and_preserves_remote_identity(self) -> None:
         client = FakeS3Client()
@@ -217,6 +226,7 @@ class ConnectorProviderTests(unittest.TestCase):
         self.assertEqual("govoplan:root/report.txt", downloaded.external_id)
         self.assertEqual("s3://govoplan/root/report.txt", downloaded.external_url)
         self.assertEqual("checksum", downloaded.metadata["checksum_sha256"])
+        self.assertTrue(client.closed)
 
 
 if __name__ == "__main__":

tests/test_connector_visibility.py

@@ -10,6 +10,9 @@ from govoplan_files.backend.storage.connector_visibility import (
     connector_profile_usable_for_import,
     visible_connector_profiles_for_actor,
 )
+from govoplan_files.backend.storage.connector_providers import (
+    connector_provider_descriptors,
+)
 
 
 def _profile(
@@ -168,12 +171,16 @@ class ConnectorVisibilityTests(unittest.TestCase):
         self,
     ) -> None:
         self.assertTrue(connector_profile_usable_for_import(_profile("webdav")))
-        self.assertFalse(
-            connector_profile_usable_for_import(_profile("s3", provider="s3"))
-        )
-        self.assertFalse(
-            connector_profile_usable_for_import(_profile("smb", provider="smb"))
-        )
+        descriptors = {
+            item.provider: item for item in connector_provider_descriptors()
+        }
+        for provider in ("s3", "smb"):
+            self.assertEqual(
+                descriptors[provider].installed,
+                connector_profile_usable_for_import(
+                    _profile(provider, provider=provider)
+                ),
+            )
         self.assertFalse(
             connector_profile_usable_for_import(
                 _profile("sharepoint", provider="sharepoint")

tests/test_documentation.py

@@ -142,6 +142,7 @@ class FilesRuntimeDocumentationTests(unittest.TestCase):
             base_path="/classified",
             credential_mode="basic",
             password_value="credential-secret",
+            secret_ref="runtime/connector-secret",
         )
         with patch(
             "govoplan_files.backend.documentation.visible_connector_profiles_for_actor",

tests/test_manifest_documentation.py

@@ -4,6 +4,7 @@ import unittest
 
 
 STATIC_TOPIC_IDS = {
+    "files.search.managed-content",
     "files.workflow.organize-managed-files",
     "files.workflow.find-and-download-files",
     "files.workflow.share-managed-files",
@@ -152,7 +153,7 @@ class FilesManifestDocumentationTests(unittest.TestCase):
             "/api/v1/files/connectors/credentials", {link.href for link in topic.links}
         )
 
-    def test_operator_topic_covers_recovery_and_fail_closed_s3_smb(self) -> None:
+    def test_operator_topic_covers_recovery_and_pinned_s3_smb(self) -> None:
         topic = self.topic(
             "files.reference.integrity-recovery-and-fail-closed-transports"
         )
@@ -163,6 +164,7 @@ class FilesManifestDocumentationTests(unittest.TestCase):
         self.assertIn("SMB", topic.body)
         self.assertIn("fail closed", topic.body)
         self.assertIn("DFS referrals", topic.body)
+        self.assertIn("ambient credential discovery", topic.body)
         self.assertIn("does not remove backend blob objects", topic.body)
         self.assertIn("bounded resumable integrity scan", topic.body)
         self.assertIn("quarantined", topic.body)

tests/test_sdk_peer_pinning.py

@@ -0,0 +1,294 @@
+from __future__ import annotations
+
+import socket
+import sys
+import threading
+import types
+import unittest
+from importlib.util import find_spec
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from unittest.mock import patch
+
+from govoplan_core.security.outbound_http import (
+    OutboundHttpBlocked,
+    create_outbound_connection as core_create_outbound_connection,
+)
+from govoplan_files.backend.storage.sdk_peer_pinning import (
+    SdkPeerPinningError,
+    _botocore_transport_types,
+    create_pinned_s3_client,
+    install_pinned_smb_transport,
+    pinned_smb_connection_cache,
+)
+
+
+class _Socket:
+    def __init__(self) -> None:
+        self.connected_to: object | None = None
+
+    def settimeout(self, _value: object) -> None:
+        return None
+
+    def setsockopt(self, *_args: object) -> None:
+        return None
+
+    def connect(self, value: object) -> None:
+        self.connected_to = value
+
+    def close(self) -> None:
+        return None
+
+
+class _S3Handler(BaseHTTPRequestHandler):
+    def do_GET(self) -> None:  # noqa: N802 - stdlib handler contract
+        body = (
+            b'<?xml version="1.0" encoding="UTF-8"?>'
+            b'<ListAllMyBucketsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
+            b'<Owner><ID>govoplan</ID><DisplayName>GovOPlaN</DisplayName></Owner>'
+            b'<Buckets><Bucket><Name>evidence</Name><CreationDate>2026-08-04T00:00:00Z</CreationDate>'
+            b'</Bucket></Buckets></ListAllMyBucketsResult>'
+        )
+        self.send_response(200)
+        self.send_header("Content-Type", "application/xml")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, _format: str, *_args: object) -> None:
+        return None
+
+
+@unittest.skipUnless(find_spec("botocore") and find_spec("boto3"), "boto3 extra is not installed")
+class S3PeerPinningTests(unittest.TestCase):
+    def test_client_is_born_with_the_pinned_http_session(self) -> None:
+        client = create_pinned_s3_client(
+            endpoint_url="http://127.0.0.1:9000",
+            region_name="eu-central-1",
+            aws_access_key_id="access",
+            aws_secret_access_key="secret",
+        )
+        try:
+            session_type, _, _ = _botocore_transport_types()
+            self.assertIsInstance(client._endpoint.http_session, session_type)
+            self.assertEqual({}, client._endpoint.http_session._proxy_config._proxies)
+        finally:
+            client.close()
+
+    def test_each_connection_attempt_resolves_and_pins_again(self) -> None:
+        _, http_connection, _ = _botocore_transport_types()
+        connection = http_connection(host="objects.example.test", port=80)
+        sockets = [_Socket(), _Socket()]
+        with patch(
+            "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+            side_effect=sockets,
+        ) as connector:
+            self.assertIs(sockets[0], connection._new_conn())
+            self.assertIs(sockets[1], connection._new_conn())
+
+        self.assertEqual(2, connector.call_count)
+        self.assertEqual("objects.example.test", connector.call_args.args[0])
+
+    def test_real_sdk_request_uses_the_pinned_socket_path(self) -> None:
+        server = ThreadingHTTPServer(("127.0.0.1", 0), _S3Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        endpoint = f"http://127.0.0.1:{server.server_port}"
+        try:
+            with patch.dict(
+                "os.environ",
+                {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"},
+            ), patch(
+                "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+                wraps=core_create_outbound_connection,
+            ) as connector:
+                client = create_pinned_s3_client(
+                    endpoint_url=endpoint,
+                    region_name="eu-central-1",
+                    aws_access_key_id="access",
+                    aws_secret_access_key="secret",
+                )
+                try:
+                    response = client.list_buckets()
+                finally:
+                    client.close()
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join(timeout=2)
+
+        self.assertEqual("evidence", response["Buckets"][0]["Name"])
+        self.assertEqual("127.0.0.1", connector.call_args.args[0])
+
+    def test_every_sdk_selected_origin_uses_a_pinned_pool(self) -> None:
+        session_type, http_connection, https_connection = _botocore_transport_types()
+        session = session_type(proxies={})
+        try:
+            first = session._manager.connection_from_url("http://one.example.test/root")
+            redirected = session._manager.connection_from_url("https://two.example.test/root")
+            self.assertIs(first.ConnectionCls, http_connection)
+            self.assertIs(redirected.ConnectionCls, https_connection)
+        finally:
+            session.close()
+
+    def test_tls_connection_retains_the_configured_authority_for_sni(self) -> None:
+        _, _, https_connection = _botocore_transport_types()
+        connection = https_connection(host="objects.example.test", port=443)
+        with patch(
+            "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+            return_value=_Socket(),
+        ) as connector:
+            connection._new_conn()
+
+        self.assertEqual("objects.example.test", connection.host)
+        self.assertEqual("objects.example.test", connection._dns_host)
+        self.assertEqual("objects.example.test", connector.call_args.args[0])
+
+    def test_mixed_answer_and_peer_change_fail_closed(self) -> None:
+        _, http_connection, _ = _botocore_transport_types()
+        connection = http_connection(host="objects.example.test", port=80)
+        records = [
+            [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80))],
+            [
+                (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80)),
+                (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 80)),
+            ],
+        ]
+        with patch.dict(
+            "os.environ",
+            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
+        ), patch(
+            "govoplan_core.security.outbound_http.socket.getaddrinfo",
+            side_effect=records,
+        ), patch(
+            "govoplan_core.security.outbound_http.socket.socket",
+            return_value=_Socket(),
+        ):
+            connection._new_conn()
+            with self.assertRaisesRegex(Exception, "non-public network"):
+                connection._new_conn()
+
+
+class _FakeTcp:
+    def __init__(self, server: str, port: int, timeout: float | None = None) -> None:
+        self.server = server
+        self.port = port
+        self.timeout = timeout
+        self.connected = False
+        self._sock = None
+        self._sock_lock = threading.Lock()
+
+
+class _FakeConnection:
+    def __init__(self, _guid: object, server: str, port: int, *, require_signing: bool) -> None:
+        self.server = server
+        self.port = port
+        self.require_signing = require_signing
+        self.transport = _FakeTcp(server, port)
+        self.session_table: dict[str, object] = {}
+
+    def connect(self, *, timeout: float) -> None:
+        self.transport.timeout = timeout
+        self.transport.connect()
+
+
+class _FakeSession:
+    def __init__(
+        self,
+        connection: _FakeConnection,
+        *,
+        username: str | None,
+        password: str | None,
+        require_encryption: bool,
+        auth_protocol: str,
+    ) -> None:
+        self.connection = connection
+        self.username = username
+        self.password = password
+        self.encrypt_data = require_encryption
+        self.auth_protocol = auth_protocol
+        self.encrypt = require_encryption
+
+    def connect(self) -> None:
+        self.connection.session_table[self.username or "anonymous"] = self
+
+
+class SmbPeerPinningTests(unittest.TestCase):
+    def setUp(self) -> None:
+        pinned_smb_connection_cache.cache_clear()
+
+    def _modules(self) -> tuple[types.ModuleType, dict[str, types.ModuleType]]:
+        smbclient = types.ModuleType("smbclient")
+        pool = types.ModuleType("smbclient._pool")
+
+        def original_register_session(
+            server: str,
+            username: str | None = None,
+            password: str | None = None,
+            port: int = 445,
+            encrypt: bool | None = None,
+            connection_timeout: float = 60,
+            connection_cache: dict[str, object] | None = None,
+            auth_protocol: str = "negotiate",
+            require_signing: bool = True,
+        ) -> None:
+            del server, username, password, port, encrypt, connection_timeout
+            del connection_cache, auth_protocol, require_signing
+
+        pool.register_session = original_register_session
+        pool.ClientConfig = lambda: types.SimpleNamespace(client_guid="guid")
+        connection = types.ModuleType("smbprotocol.connection")
+        connection.Connection = _FakeConnection
+        connection.Tcp = _FakeTcp
+        session = types.ModuleType("smbprotocol.session")
+        session.Session = _FakeSession
+        transport = types.ModuleType("smbprotocol.transport")
+        transport.Tcp = _FakeTcp
+        return smbclient, {
+            "smbclient._pool": pool,
+            "smbprotocol.connection": connection,
+            "smbprotocol.session": session,
+            "smbprotocol.transport": transport,
+        }
+
+    def test_initial_reconnect_and_referral_hosts_are_each_pinned(self) -> None:
+        smbclient, modules = self._modules()
+        sockets = [_Socket(), _Socket(), _Socket()]
+        with patch.dict(sys.modules, modules), patch(
+            "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+            side_effect=sockets,
+        ) as connector:
+            install_pinned_smb_transport(smbclient)
+            register = modules["smbclient._pool"].register_session
+            cache: dict[str, object] = {}
+            first = register("files.example.test", connection_cache=cache)
+            first.connection.transport.connected = False
+            register("files.example.test", connection_cache=cache)
+            register("dfs-target.example.test", connection_cache=cache)
+
+        self.assertEqual(
+            ["files.example.test", "files.example.test", "dfs-target.example.test"],
+            [call.args[0] for call in connector.call_args_list],
+        )
+
+    def test_referral_to_disallowed_peer_fails_before_session_creation(self) -> None:
+        smbclient, modules = self._modules()
+        with patch.dict(sys.modules, modules), patch(
+            "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+            side_effect=OutboundHttpBlocked("non-public network"),
+        ):
+            install_pinned_smb_transport(smbclient)
+            register = modules["smbclient._pool"].register_session
+            with self.assertRaisesRegex(ValueError, "non-public network"):
+                register("private-referral.example.test", connection_cache={})
+
+    def test_transport_tampering_after_installation_fails_closed(self) -> None:
+        smbclient, modules = self._modules()
+        with patch.dict(sys.modules, modules):
+            install_pinned_smb_transport(smbclient)
+            modules["smbprotocol.connection"].Tcp = _FakeTcp
+            with self.assertRaisesRegex(SdkPeerPinningError, "changed after peer pinning"):
+                install_pinned_smb_transport(smbclient)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_search_source.py

@@ -9,6 +9,7 @@ from sqlalchemy.orm import Session
 from govoplan_access.backend.db.models import Account, Group, User
 from govoplan_core.auth import ApiPrincipal
 from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.change_sequence import ChangeSequenceEntry
 from govoplan_core.core.events import EventObjectRef, EventTenantRef, PlatformEvent
 from govoplan_core.core.search import (
     SearchAuthorizationRequest,
@@ -35,6 +36,7 @@ class FilesSearchSourceTests(unittest.TestCase):
                 FileAsset.__table__,
                 FileFolder.__table__,
                 FileShare.__table__,
+                ChangeSequenceEntry.__table__,
             ),
         )
         self.session = Session(self.engine)