""" Tests for the proxy (redsocks) exit type — configure_proxy validation, redsocks.conf generation (golden strings, no injection), apply_routes REDIRECT rules, _exit_status bridging, egress_manager mirroring, and the /api/connectivity/exits/proxy route (never echoes secrets). """ import os import stat import sys import shutil import tempfile import unittest from pathlib import Path from unittest.mock import MagicMock, patch sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'api')) import connectivity_manager as cm_module from connectivity_manager import ConnectivityManager import egress_manager as em_module _SENTINEL = object() def _make_manager(tmp_dir=None, peer_registry=_SENTINEL, config_manager=None): if tmp_dir is None: tmp_dir = tempfile.mkdtemp() if config_manager is None: config_manager = MagicMock() config_manager.get_identity.return_value = { 'cell_name': 'test', 'ip_range': '172.20.0.0/16', } config_manager.get_connectivity_config.return_value = { 'exits': {}, 'peer_exit_map': {}, } config_manager.get_installed_services.return_value = {} if peer_registry is _SENTINEL: peer_registry = MagicMock() peer_registry.list_peers.return_value = [] with patch.object(ConnectivityManager, '_subscribe_to_events', lambda self: None): mgr = ConnectivityManager( config_manager=config_manager, peer_registry=peer_registry, data_dir=tmp_dir, config_dir=tmp_dir, ) return mgr def _valid_cfg(**overrides): cfg = { 'scheme': 'socks5', 'host': 'proxy.example.com', 'port': 1080, } cfg.update(overrides) return cfg def _mock_subprocess_ok(): return MagicMock(returncode=0, stdout='', stderr='') # --------------------------------------------------------------------------- # configure_proxy — validation # --------------------------------------------------------------------------- class TestConfigureProxyValidation(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.mgr = _make_manager(tmp_dir=self.tmp) def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) def test_valid_socks5_config_returns_ok(self): result = self.mgr.configure_proxy(_valid_cfg()) self.assertTrue(result['ok'], result) def test_valid_http_config_returns_ok(self): result = self.mgr.configure_proxy(_valid_cfg(scheme='http', port=3128)) self.assertTrue(result['ok']) def test_non_dict_config_rejected(self): result = self.mgr.configure_proxy([1, 2, 3]) self.assertFalse(result['ok']) def test_missing_scheme_rejected(self): cfg = _valid_cfg() del cfg['scheme'] result = self.mgr.configure_proxy(cfg) self.assertFalse(result['ok']) self.assertIn('scheme', result['error']) def test_invalid_scheme_rejected(self): result = self.mgr.configure_proxy(_valid_cfg(scheme='socks4')) self.assertFalse(result['ok']) def test_https_scheme_rejected(self): result = self.mgr.configure_proxy(_valid_cfg(scheme='https')) self.assertFalse(result['ok']) def test_missing_host_rejected(self): cfg = _valid_cfg() del cfg['host'] result = self.mgr.configure_proxy(cfg) self.assertFalse(result['ok']) self.assertIn('host', result['error']) def test_host_with_semicolon_rejected(self): result = self.mgr.configure_proxy(_valid_cfg(host='evil;injected')) self.assertFalse(result['ok']) def test_host_with_quote_rejected(self): result = self.mgr.configure_proxy(_valid_cfg(host='a"b')) self.assertFalse(result['ok']) def test_host_with_newline_rejected(self): result = self.mgr.configure_proxy(_valid_cfg(host='a\nb')) self.assertFalse(result['ok']) def test_ip_host_accepted(self): result = self.mgr.configure_proxy(_valid_cfg(host='203.0.113.99')) self.assertTrue(result['ok']) def test_missing_port_rejected(self): cfg = _valid_cfg() del cfg['port'] result = self.mgr.configure_proxy(cfg) self.assertFalse(result['ok']) self.assertIn('port', result['error']) def test_port_zero_rejected(self): result = self.mgr.configure_proxy(_valid_cfg(port=0)) self.assertFalse(result['ok']) def test_port_above_65535_rejected(self): result = self.mgr.configure_proxy(_valid_cfg(port=65536)) self.assertFalse(result['ok']) def test_port_non_numeric_rejected(self): result = self.mgr.configure_proxy(_valid_cfg(port='oops')) self.assertFalse(result['ok']) def test_user_with_injection_chars_rejected(self): result = self.mgr.configure_proxy( _valid_cfg(user='user";\nip = evil')) self.assertFalse(result['ok']) def test_password_with_double_quote_rejected(self): result = self.mgr.configure_proxy( _valid_cfg(user='bob', password='pa"ss')) self.assertFalse(result['ok']) def test_password_with_backslash_rejected(self): result = self.mgr.configure_proxy( _valid_cfg(user='bob', password='pa\\ss')) self.assertFalse(result['ok']) def test_password_with_newline_rejected(self): result = self.mgr.configure_proxy( _valid_cfg(user='bob', password='pa\nss')) self.assertFalse(result['ok']) def test_password_without_user_rejected(self): result = self.mgr.configure_proxy(_valid_cfg(password='secret')) self.assertFalse(result['ok']) def test_result_never_contains_password(self): result = self.mgr.configure_proxy( _valid_cfg(user='bob', password='topsecret99')) self.assertNotIn('topsecret99', str(result)) # --------------------------------------------------------------------------- # configure_proxy — redsocks.conf generation # --------------------------------------------------------------------------- class TestRedsocksConfGeneration(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.mgr = _make_manager(tmp_dir=self.tmp) def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) def _conf_path(self): return Path(self.mgr.proxy_dir, 'redsocks.conf') def test_socks5_conf_golden(self): self.mgr.configure_proxy(_valid_cfg()) expected = ( 'base {\n' ' log_debug = off;\n' ' log_info = on;\n' ' log = stderr;\n' ' daemon = off;\n' ' redirector = iptables;\n' '}\n' '\n' 'redsocks {\n' ' local_ip = 0.0.0.0;\n' ' local_port = 12345;\n' ' ip = proxy.example.com;\n' ' port = 1080;\n' ' type = socks5;\n' '}\n' ) self.assertEqual(self._conf_path().read_text(), expected) def test_http_conf_uses_http_connect_type(self): self.mgr.configure_proxy(_valid_cfg(scheme='http', port=3128)) conf = self._conf_path().read_text() self.assertIn('type = http-connect;', conf) self.assertIn('port = 3128;', conf) def test_auth_conf_golden_with_login_and_password(self): self.mgr.configure_proxy(_valid_cfg(user='bob', password='s3cret!')) conf = self._conf_path().read_text() self.assertIn(' login = "bob";\n', conf) self.assertIn(' password = "s3cret!";\n', conf) def test_conf_without_auth_has_no_login_lines(self): self.mgr.configure_proxy(_valid_cfg()) conf = self._conf_path().read_text() self.assertNotIn('login', conf) self.assertNotIn('password', conf) def test_conf_file_mode_0600(self): self.mgr.configure_proxy(_valid_cfg(user='bob', password='s3cret!')) mode = stat.S_IMODE(os.stat(self._conf_path()).st_mode) self.assertEqual(mode, 0o600) def test_password_not_persisted_in_config_manager(self): self.mgr.configure_proxy(_valid_cfg(user='bob', password='s3cret!')) self.mgr.config_manager.set_connectivity_field.assert_called_once() field, exits = self.mgr.config_manager.set_connectivity_field.call_args[0] self.assertEqual(field, 'exits') self.assertEqual(exits['proxy']['scheme'], 'socks5') self.assertEqual(exits['proxy']['user'], 'bob') self.assertNotIn('password', exits['proxy']) def test_write_failure_returns_ok_false(self): with patch.object(self.mgr, '_write_secure', side_effect=OSError('disk full')): result = self.mgr.configure_proxy(_valid_cfg()) self.assertFalse(result['ok']) # --------------------------------------------------------------------------- # apply_routes — proxy REDIRECT # --------------------------------------------------------------------------- class TestApplyRoutesProxy(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) def test_proxy_peer_gets_redirect_to_12345(self): pr = MagicMock() pr.list_peers.return_value = [ {'peer': 'bob', 'exit_via': 'proxy'}, ] pr.get_peer.return_value = {'peer': 'bob', 'ip': '172.20.0.60/32'} mgr = _make_manager(tmp_dir=self.tmp, peer_registry=pr) with patch.object(cm_module, 'subprocess') as mock_sp: mock_sp.run.return_value = _mock_subprocess_ok() mgr.apply_routes() redirect_calls = [ c for c in mock_sp.run.call_args_list if 'REDIRECT' in c.args[0] ] self.assertEqual(len(redirect_calls), 1) args = redirect_calls[0].args[0] self.assertEqual(args[args.index('--to-ports') + 1], '12345') self.assertIn('172.20.0.60', args) def test_proxy_peer_gets_mark_0x50(self): pr = MagicMock() pr.list_peers.return_value = [ {'peer': 'bob', 'exit_via': 'proxy'}, ] pr.get_peer.return_value = {'peer': 'bob', 'ip': '172.20.0.60/32'} mgr = _make_manager(tmp_dir=self.tmp, peer_registry=pr) with patch.object(cm_module, 'subprocess') as mock_sp: mock_sp.run.return_value = _mock_subprocess_ok() mgr.apply_routes() mark_calls = [ c for c in mock_sp.run.call_args_list if 'MARK' in c.args[0] and '172.20.0.60' in c.args[0] ] self.assertEqual(len(mark_calls), 1) args = mark_calls[0].args[0] self.assertEqual(args[args.index('--set-mark') + 1], hex(0x50)) def test_ip_rule_added_for_proxy_table_150(self): mgr = _make_manager(tmp_dir=self.tmp) with patch.object(cm_module, 'subprocess') as mock_sp: mock_sp.run.return_value = MagicMock(returncode=1, stdout='', stderr='') mgr.apply_routes() rule_adds = [ c for c in mock_sp.run.call_args_list if 'rule' in c.args[0] and 'add' in c.args[0] and hex(0x50) in c.args[0] ] self.assertEqual(len(rule_adds), 1) self.assertIn('150', rule_adds[0].args[0]) def test_tor_redirect_still_uses_9040(self): """Regression: tor redirect must be unaffected by the new exits.""" pr = MagicMock() pr.list_peers.return_value = [ {'peer': 'carol', 'exit_via': 'tor'}, ] pr.get_peer.return_value = {'peer': 'carol', 'ip': '172.20.0.70/32'} mgr = _make_manager(tmp_dir=self.tmp, peer_registry=pr) with patch.object(cm_module, 'subprocess') as mock_sp: mock_sp.run.return_value = _mock_subprocess_ok() mgr.apply_routes() redirect_calls = [ c for c in mock_sp.run.call_args_list if 'REDIRECT' in c.args[0] ] self.assertEqual(len(redirect_calls), 1) args = redirect_calls[0].args[0] self.assertEqual(args[args.index('--to-ports') + 1], '9040') # --------------------------------------------------------------------------- # egress_manager mirror — marks/tables/redirect ports # --------------------------------------------------------------------------- class TestEgressManagerMirror(unittest.TestCase): def test_exit_types_include_sshuttle_and_proxy(self): self.assertIn('sshuttle', em_module.EXIT_TYPES) self.assertIn('proxy', em_module.EXIT_TYPES) def test_marks_do_not_collide_with_connectivity(self): self.assertEqual(em_module.MARKS['sshuttle'], 0x140) self.assertEqual(em_module.MARKS['proxy'], 0x150) self.assertNotIn(em_module.MARKS['sshuttle'], ConnectivityManager.MARKS.values()) self.assertNotIn(em_module.MARKS['proxy'], ConnectivityManager.MARKS.values()) def test_tables(self): self.assertEqual(em_module.TABLES['sshuttle'], 240) self.assertEqual(em_module.TABLES['proxy'], 250) def _make_egress(self, exit_via): config_manager = MagicMock() manifest = { 'id': 'svc', 'container_name': 'cell-svc', 'has_egress': True, 'egress': {'default': exit_via, 'allowed': list(em_module.EXIT_TYPES)}, } config_manager.get_installed_services.return_value = { 'svc': {'manifest': manifest}, } config_manager.configs = {'egress_overrides': {}} return em_module.EgressManager(config_manager=config_manager) def test_apply_service_sshuttle_redirects_to_12300(self): em = self._make_egress('sshuttle') with patch.object(em_module, 'subprocess') as mock_sp: mock_sp.run.return_value = MagicMock( returncode=0, stdout='172.21.0.5', stderr='') result = em.apply_service('svc') self.assertTrue(result['ok'], result) redirect_calls = [ c for c in mock_sp.run.call_args_list if 'REDIRECT' in c.args[0] ] self.assertEqual(len(redirect_calls), 1) args = redirect_calls[0].args[0] self.assertEqual(args[args.index('--to-ports') + 1], '12300') def test_apply_service_proxy_redirects_to_12345(self): em = self._make_egress('proxy') with patch.object(em_module, 'subprocess') as mock_sp: mock_sp.run.return_value = MagicMock( returncode=0, stdout='172.21.0.5', stderr='') result = em.apply_service('svc') self.assertTrue(result['ok'], result) redirect_calls = [ c for c in mock_sp.run.call_args_list if 'REDIRECT' in c.args[0] ] self.assertEqual(len(redirect_calls), 1) args = redirect_calls[0].args[0] self.assertEqual(args[args.index('--to-ports') + 1], '12345') # --------------------------------------------------------------------------- # _exit_status — proxy bridge # --------------------------------------------------------------------------- class TestProxyExitStatus(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) def _mgr(self, installed=None): config_manager = MagicMock() config_manager.get_identity.return_value = {'ip_range': '172.20.0.0/16'} config_manager.get_installed_services.return_value = installed or {} return _make_manager(tmp_dir=self.tmp, config_manager=config_manager) def test_not_configured_initially(self): mgr = self._mgr() with patch.object(cm_module, 'subprocess') as mock_sp: mock_sp.run.return_value = MagicMock(returncode=1, stdout='', stderr='') info = mgr._exit_status('proxy') self.assertFalse(info['configured']) self.assertEqual(info['status'], 'not_configured') def test_configured_after_configure_proxy(self): mgr = self._mgr() mgr.configure_proxy(_valid_cfg()) with patch.object(cm_module, 'subprocess') as mock_sp: mock_sp.run.return_value = MagicMock(returncode=1, stdout='', stderr='') info = mgr._exit_status('proxy') self.assertTrue(info['configured']) self.assertEqual(info['status'], 'configured') def test_configured_when_store_service_installed(self): mgr = self._mgr(installed={'proxy': {'manifest': {'id': 'proxy'}}}) with patch.object(cm_module, 'subprocess') as mock_sp: mock_sp.run.return_value = MagicMock(returncode=1, stdout='', stderr='') info = mgr._exit_status('proxy') self.assertTrue(info['configured']) def test_configured_when_redsocks_container_running(self): mgr = self._mgr() def fake_run(cmd, **kwargs): if 'inspect' in cmd and 'cell-redsocks' in cmd: return MagicMock(returncode=0, stdout='true\n', stderr='') return MagicMock(returncode=1, stdout='', stderr='') with patch.object(cm_module, 'subprocess') as mock_sp: mock_sp.run.side_effect = fake_run info = mgr._exit_status('proxy') self.assertTrue(info['configured']) def test_proxy_in_list_exits(self): mgr = self._mgr() with patch.object(cm_module, 'subprocess') as mock_sp: mock_sp.run.return_value = _mock_subprocess_ok() exits = mgr.list_exits() types = {e['type'] for e in exits} self.assertIn('proxy', types) # --------------------------------------------------------------------------- # POST /api/connectivity/exits/proxy — route behaviour # --------------------------------------------------------------------------- class TestProxyRoute(unittest.TestCase): def setUp(self): import app as app_module self.app_module = app_module app_module.app.config['TESTING'] = True self.client = app_module.app.test_client() def test_valid_config_returns_200_ok_only(self): mock_cm = MagicMock() mock_cm.configure_proxy.return_value = {'ok': True} with patch.object(self.app_module, 'connectivity_manager', mock_cm): resp = self.client.post('/api/connectivity/exits/proxy', json=_valid_cfg(user='bob', password='pw123')) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.get_json(), {'ok': True}) def test_invalid_config_returns_400(self): mock_cm = MagicMock() mock_cm.configure_proxy.return_value = {'ok': False, 'error': 'invalid scheme'} with patch.object(self.app_module, 'connectivity_manager', mock_cm): resp = self.client.post('/api/connectivity/exits/proxy', json={'scheme': 'gopher'}) self.assertEqual(resp.status_code, 400) self.assertFalse(resp.get_json()['ok']) def test_response_never_echoes_password(self): mock_cm = MagicMock() mock_cm.configure_proxy.return_value = {'ok': True} with patch.object(self.app_module, 'connectivity_manager', mock_cm): resp = self.client.post( '/api/connectivity/exits/proxy', json=_valid_cfg(user='bob', password='ultra-secret-pw')) self.assertNotIn('ultra-secret-pw', resp.get_data(as_text=True)) def test_exception_returns_500_without_details(self): mock_cm = MagicMock() mock_cm.configure_proxy.side_effect = Exception('boom secret-detail') with patch.object(self.app_module, 'connectivity_manager', mock_cm): resp = self.client.post('/api/connectivity/exits/proxy', json=_valid_cfg()) self.assertEqual(resp.status_code, 500) self.assertNotIn('secret-detail', resp.get_data(as_text=True)) if __name__ == '__main__': unittest.main()