"""Legacy broker peer policy checks without sockets, desktops or live profiles.""" from __future__ import annotations import contextlib import io import tempfile import types import unittest from unittest import mock from framebridge_expanded import broker CONTROL_PATHS = ( '/fbpair/v1/control/node-status', '/fbpair/v1/control/request', '/fbpair/v1/control/admin-exec', '/fbpair/v1/control/admin-cua', '/fbpair/v1/control/operation', '/fbpair/v1/control/result', '/dev/authenticate-and-approve', '/dev/approve-authenticated', ) class BodyReadReached(Exception): pass class NoStoreCalls: def __getattr__(self, name): raise AssertionError('The peer gate must run before Store access') class LegacyBrokerPeerTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.original = broker.original_modules() def request_fixture(self, handler, path, peer): request = object.__new__(handler) request.path = path request.client_address = (peer, 12345) # Forwarding headers must never replace the actual transport peer. request.headers = {'X-Forwarded-For': '127.0.0.1', 'X-Real-IP': '127.0.0.1'} request.read_json = mock.Mock(side_effect=BodyReadReached) request.send_json = mock.Mock() return request def test_default_policy_preserves_only_the_original_loopback_peer(self): allowed = self.original.parse_control_peers() self.assertEqual(len(allowed), 1) self.assertTrue(self.original.control_peer_allowed('127.0.0.1', allowed)) for peer in ('192.0.2.15', '198.51.100.18', 'localhost', '', None): with self.subTest(kind=type(peer).__name__): self.assertFalse(self.original.control_peer_allowed(peer, allowed)) def test_explicit_literal_peers_are_canonical_and_narrow(self): allowed = self.original.parse_control_peers(['192.0.2.15', '2001:0db8:0:0::15']) self.assertEqual(len(allowed), 3) self.assertTrue(self.original.control_peer_allowed('192.0.2.15', allowed)) self.assertTrue(self.original.control_peer_allowed('2001:db8::15', allowed)) self.assertTrue(self.original.control_peer_allowed('127.0.0.1', allowed)) self.assertFalse(self.original.control_peer_allowed('192.0.2.16', allowed)) def test_malformed_or_implicit_peer_configuration_is_rejected(self): invalid = ('192.0.2.15', ['', '192.0.2.15'], ['*'], ['0.0.0.0/0'], ['0.0.0.0'], ['::'], ['example.com'], ['[::1]'], ['fe80::1%eth0'], [' 192.0.2.15'], [None], [123]) for value in invalid: with self.subTest(configuration_type=type(value).__name__): with self.assertRaisesRegex(ValueError, '^CONTROL_PEER_CONFIG_INVALID$'): self.original.parse_control_peers(value) def test_cli_rejects_invalid_configuration_before_store_or_listener(self): with mock.patch.object(self.original, 'Store') as store, \ mock.patch.object(self.original, 'ThreadingHTTPServer') as server, \ mock.patch('sys.argv', ['legacy-broker', '--control-peer', 'example.com']), \ contextlib.redirect_stderr(io.StringIO()) as errors: with self.assertRaises(SystemExit) as caught: self.original.main() self.assertEqual(caught.exception.code, 2) self.assertIn('CONTROL_PEER_CONFIG_INVALID', errors.getvalue()) self.assertNotIn('example.com', errors.getvalue()) store.assert_not_called() server.assert_not_called() def test_repeated_cli_peers_reach_the_handler_only_when_explicit(self): with mock.patch.object(self.original, 'Store'), \ mock.patch.object(self.original, 'ThreadingHTTPServer') as server, \ mock.patch.object(self.original, 'make_handler') as make_handler, \ mock.patch('sys.argv', ['legacy-broker', '--control-peer', '192.0.2.15', '--control-peer', '2001:db8::15']), contextlib.redirect_stdout(io.StringIO()): self.original.main() allowed = make_handler.call_args.kwargs['control_peers'] self.assertEqual(len(allowed), 3) self.assertTrue(self.original.control_peer_allowed('192.0.2.15', allowed)) self.assertTrue(self.original.control_peer_allowed('2001:db8::15', allowed)) server.return_value.serve_forever.assert_called_once() def test_all_eight_routes_deny_an_unconfigured_peer_before_body_or_store(self): handler = self.original.make_handler(NoStoreCalls(), 'http://localhost:18765') for path in CONTROL_PATHS: with self.subTest(path=path): request = self.request_fixture(handler, path, '192.0.2.15') handler.do_POST(request) request.read_json.assert_not_called() request.send_json.assert_called_once_with(403, {'ok': False, 'error': 'local_only'}) def test_loopback_and_explicit_peers_pass_the_gate_for_all_eight_routes(self): handler = self.original.make_handler(NoStoreCalls(), 'http://localhost:18765', control_peers=['192.0.2.15']) for path in CONTROL_PATHS: for peer in ('127.0.0.1', '192.0.2.15'): with self.subTest(path=path, source_kind='loopback' if peer.startswith('127.') else 'explicit'): request = self.request_fixture(handler, path, peer) with self.assertRaises(BodyReadReached): handler.do_POST(request) request.read_json.assert_called_once() def test_expanded_control_keeps_authentication_and_blocks_development_routes(self): def no_socket_server(address, handler): return types.SimpleNamespace(RequestHandlerClass=handler) with tempfile.TemporaryDirectory(prefix='framebridge-broker-policy-') as state, \ mock.patch.object(broker, '_config', return_value={}), \ mock.patch.object(broker, '_control_secret', return_value='fixture-control'), \ mock.patch.object(broker, 'ThreadingHTTPServer', side_effect=no_socket_server), \ mock.patch.object(self.original, 'control_peer_allowed', side_effect=AssertionError('legacy gate reached')): server = broker.make_server(host='localhost', state_dir=state) handler = server.RequestHandlerClass for path, status, error in ( ('/fbpair/v1/control/nodes', 403, 'CONTROL_AUTH_REQUIRED'), ('/dev/authenticate-and-approve', 404, 'ENDPOINT_NOT_AVAILABLE'), ('/dev/approve-authenticated', 404, 'ENDPOINT_NOT_AVAILABLE'), ): with self.subTest(path=path): request = self.request_fixture(handler, path, '127.0.0.1') handler.do_POST(request) request.read_json.assert_not_called() request.send_json.assert_called_once_with(status, {'ok': False, 'error': error}) if __name__ == '__main__': unittest.main()