295 lines
12 KiB
Python
295 lines
12 KiB
Python
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()
|