"""Tests for remote access to the MCP bridge (a domain name, e.g. through frp, with a password): python3 -m unittest scripts/test_mcp_remote.py """ import http.client import json import pathlib import sys import tempfile import threading import unittest HERE = pathlib.Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import mcp_bridge # noqa: E402 import mcp_remote # noqa: E402 import serve # noqa: E402 PY = sys.executable PASSWORD = 'correct horse battery staple' HOST = 'mcp.example.com' class Clock: def __init__(self): self.t = 1000.0 def __call__(self): return self.t def remote(cfg=None, password=PASSWORD, clock=None): return mcp_remote.RemoteAccess.from_config({'hosts': [HOST], **(cfg or {})}, {'WEBTT_MCP_PASSWORD': password} if password else {}, clock=clock or Clock()) class ConfigTest(unittest.TestCase): def test_no_remote_section_means_no_remote_access(self): self.assertIsNone(mcp_remote.RemoteAccess.from_config(None, {})) def test_a_long_enough_password_is_required(self): with self.assertRaisesRegex(ValueError, 'WEBTT_MCP_PASSWORD'): remote(password=None) with self.assertRaisesRegex(ValueError, '16'): remote(password='short') def test_bad_sections_are_rejected(self): for bad in ({'hosts': []}, {'hosts': 'x.com'}, {'hosts': ['bad host']}, {'hosts': [HOST], 'extra': 1}, {'hosts': [HOST], 'session_hours': 0}, {'hosts': [HOST], 'insecure_http': 'yes'}, {'hosts': [HOST], 'trust_localhost': 1}, {'hosts': ['localhost:8289']}, {'hosts': ['127.0.0.1']}, {'hosts': ['LOCALHOST']}): with self.assertRaises(ValueError, msg=bad): mcp_remote.RemoteAccess.from_config(bad, {'WEBTT_MCP_PASSWORD': PASSWORD}) def test_hosts_match_case_insensitively_and_with_their_port(self): r = remote({'hosts': ['MCP.example.com', 'b.example.com:8443']}) self.assertTrue(r.is_remote_host('mcp.EXAMPLE.com')) self.assertTrue(r.is_remote_host('b.example.com:8443')) self.assertFalse(r.is_remote_host('b.example.com')) self.assertFalse(r.is_remote_host('evil.com')) class SessionTest(unittest.TestCase): def setUp(self): self.clock = Clock() self.r = remote({'session_hours': 2}, clock=self.clock) def test_login_and_session(self): self.assertIsNone(self.r.login('wrong password here').token) token = self.r.login(PASSWORD).token self.assertTrue(self.r.valid(token)) self.assertFalse(self.r.valid('forged')) self.assertFalse(self.r.valid('')) def test_sessions_expire_and_logout_ends_them(self): token = self.r.login(PASSWORD).token self.clock.t += 2 * 3600 - 1 self.assertTrue(self.r.valid(token)) self.clock.t += 2 self.assertFalse(self.r.valid(token)) other = self.r.login(PASSWORD).token self.r.logout(other) self.assertFalse(self.r.valid(other)) def test_failures_lock_login_with_growing_delays(self): for _ in range(mcp_remote.FREE_ATTEMPTS): self.assertIsNone(self.r.login('nope').token) locked = self.r.login(PASSWORD) self.assertIsNone(locked.token, 'even the right password waits out the lock') self.assertEqual(locked.retry_after, 60) self.clock.t += 61 self.assertIsNone(self.r.login('nope').token) # one more failure locks again, twice as long self.assertEqual(self.r.login(PASSWORD).retry_after, 120) self.clock.t += 121 self.assertIsNotNone(self.r.login(PASSWORD).token) # success resets self.assertIsNone(self.r.login('nope').retry_after) def test_lock_is_capped(self): for _ in range(40): self.r.login('nope') self.clock.t += 10_000 self.r.login('nope') self.assertLessEqual(self.r.login(PASSWORD).retry_after, mcp_remote.MAX_LOCK) def test_cookie_attributes(self): cookie = self.r.cookie_header('tok') for part in ('mcp_session=tok', 'HttpOnly', 'Secure', 'SameSite=Strict', 'Path=/api;', 'Max-Age=7200'): self.assertIn(part, cookie) self.assertNotIn('Secure', remote({'insecure_http': True}).cookie_header('tok')) self.assertIn('Max-Age=0', self.r.clear_cookie_header()) def test_origins(self): self.assertTrue(self.r.origin_ok(HOST, f'https://{HOST}')) self.assertFalse(self.r.origin_ok(HOST, f'http://{HOST}')) self.assertFalse(self.r.origin_ok(HOST, 'https://evil.com')) self.assertTrue(self.r.origin_ok(HOST.upper(), f'https://{HOST}')) self.assertTrue(self.r.origin_ok('localhost:8289', 'http://localhost:8289'), 'localhost is a secure context over http') self.assertTrue(remote({'insecure_http': True}).origin_ok(HOST, f'http://{HOST}')) class BridgeHarness: """A real bridge; requests arrive from 127.0.0.1 like frp's, with the domain as Host. Subclasses set REMOTE, the config's "remote" section.""" REMOTE = None @classmethod def setUpClass(cls): cls.tmp = tempfile.TemporaryDirectory() config = { 'remote': cls.REMOTE, 'servers': [ {'id': 'demo', 'command': [PY, str(HERE / 'mcp_demo_server.py')]}, {'id': 'sh', 'shell': {'allow': ['echo'], 'confirm': False}}, {'id': 'sh-remote', 'remote': True, 'shell': {'allow': ['echo'], 'confirm': False}}, {'id': 'hidden', 'remote': False, 'cli': [{'name': 'wc', 'command': ['wc']}]}, ]} path = pathlib.Path(cls.tmp.name) / 'mcp.config.json' path.write_text(json.dumps(config), encoding='utf-8') cls.bridge = mcp_bridge.Bridge(path, 0, env={'WEBTT_MCP_PASSWORD': PASSWORD}) cls.httpd = serve.make_server(0, cls.bridge, host='127.0.0.1', quiet=True) cls.port = cls.bridge.port = cls.httpd.server_address[1] cls.local = f'localhost:{cls.port}' threading.Thread(target=cls.httpd.serve_forever, daemon=True).start() @classmethod def tearDownClass(cls): cls.httpd.shutdown() cls.bridge.shutdown() cls.tmp.cleanup() def setUp(self): self.bridge.remote.reset_limits() def request(self, method, path, body=None, host=HOST, origin=None, cookie=None, headers=None): conn = http.client.HTTPConnection('127.0.0.1', self.port, timeout=10) h = {'Host': host, **(headers or {})} if origin: h['Origin'] = origin if cookie: h['Cookie'] = cookie data = None if body is not None: data = json.dumps(body).encode() h['Content-Type'] = 'application/json' conn.request(method, path, body=data, headers=h) resp = conn.getresponse() text = resp.read().decode() conn.close() return resp.status, dict(resp.getheaders()), (json.loads(text) if text.startswith(('{', '[')) else text) def login(self, password=PASSWORD, host=HOST, origin=None): status, headers, body = self.request('POST', '/api/mcp/login', {'password': password}, host=host, origin=origin or f'https://{host}') cookie = headers.get('Set-Cookie', '').split(';')[0] return status, cookie, body def page_cookie(self, host, headers=None): """The cookie a page load sets, if any.""" _, h, _ = self.request('GET', '/www/mcp/', host=host, headers=headers) return h.get('Set-Cookie', '').split(';')[0] def ids(self, cookie, host=HOST): status, _, servers = self.request('GET', '/api/mcp/servers', host=host, cookie=cookie) return status, [s['id'] for s in servers] if status == 200 else servers class RemoteBridgeTest(BridgeHarness, unittest.TestCase): REMOTE = {'hosts': [HOST]} # ── with remote access configured, "localhost" is not trusted on its own ── def test_localhost_page_gets_no_token_and_must_log_in(self): # A proxy that rewrites Host to localhost would otherwise hand the internet the local token self.assertEqual(self.page_cookie(self.local), '') status, body = self.ids('', host=self.local) self.assertEqual((status, body.get('login')), (401, True)) status, cookie, _ = self.login(host=self.local, origin=f'http://{self.local}') self.assertEqual(status, 200, 'localhost logs in over plain http') self.assertEqual(self.ids(cookie, host=self.local), (200, ['demo', 'sh-remote']), 'same reach as remote') self.assertEqual(self.request('GET', '/api/mcp/whoami', host=self.local, cookie=cookie)[2]['mode'], 'remote') # ── remote access needs a login ── def test_remote_requests_need_a_session(self): self.assertEqual(self.page_cookie(HOST), '', 'no token for remote page loads') status, body = self.ids('') self.assertEqual((status, body.get('login')), (401, True)) self.assertEqual(self.request('GET', '/api/mcp/servers', host='evil.example')[0], 403) def test_login_logout_and_what_a_session_can_reach(self): self.assertEqual(self.login('wrong password here')[0], 401) status, cookie, _ = self.login() self.assertEqual(status, 200) self.assertTrue(cookie.startswith('mcp_session=')) self.assertEqual(self.ids(cookie), (200, ['demo', 'sh-remote']), 'shells need "remote": true') init = {'jsonrpc': '2.0', 'id': 1, 'method': 'initialize', 'params': {'protocolVersion': '2025-06-18', 'capabilities': {}}} status, _, reply = self.request('POST', '/api/mcp/demo/rpc', init, cookie=cookie, origin=f'https://{HOST}') self.assertEqual((status, reply['result']['serverInfo']['name']), (200, 'webtt-demo')) for hidden in ('sh', 'hidden'): self.assertEqual(self.request('POST', f'/api/mcp/{hidden}/rpc', init, cookie=cookie)[0], 404, hidden) self.assertEqual(self.request('POST', f'/api/mcp/{hidden}/restart', {}, cookie=cookie)[0], 404, hidden) whoami = self.request('GET', '/api/mcp/whoami', cookie=cookie)[2] self.assertEqual(whoami['mode'], 'remote') self.assertGreater(whoami['expires_in'], 0) status, headers, _ = self.request('POST', '/api/mcp/logout', {}, cookie=cookie, origin=f'https://{HOST}') self.assertEqual(status, 200) self.assertIn('Max-Age=0', headers['Set-Cookie']) self.assertEqual(self.ids(cookie)[0], 401) def test_host_case_does_not_matter(self): status, cookie, _ = self.login(host=HOST.upper(), origin=f'https://{HOST}') self.assertEqual(status, 200) def test_login_needs_this_sites_origin(self): self.assertEqual(self.request('POST', '/api/mcp/login', {'password': PASSWORD})[0], 403) self.assertEqual(self.request('POST', '/api/mcp/login', {'password': PASSWORD}, origin='https://evil.example')[0], 403) self.assertEqual(self.request('POST', '/api/mcp/login', {'password': PASSWORD}, origin=f'http://{HOST}')[0], 403, 'https only unless insecure_http') def test_cross_site_and_foreign_origin_requests_are_refused_even_with_a_session(self): _, cookie, _ = self.login() self.assertEqual(self.request('GET', '/api/mcp/servers', cookie=cookie, origin='https://evil.example')[0], 403) self.assertEqual(self.request('GET', '/api/mcp/servers', cookie=cookie, headers={'Sec-Fetch-Site': 'cross-site'})[0], 403) def test_repeated_failures_answer_429_with_retry_after(self): for _ in range(mcp_remote.FREE_ATTEMPTS): self.login('wrong password here') status, _, body = self.request('POST', '/api/mcp/login', {'password': PASSWORD}, origin=f'https://{HOST}') self.assertEqual(status, 429) self.assertGreater(body['retry_after'], 0) def test_bad_login_bodies(self): for body in ({}, {'password': 123}, []): self.assertIn(self.request('POST', '/api/mcp/login', body, origin=f'https://{HOST}')[0], (400, 401), body) def test_pages_and_api_cannot_be_framed(self): # Clickjacking: another site must not show the console in a frame for path in ('/www/mcp/', '/www/', '/api/mcp/whoami'): _, headers, _ = self.request('GET', path) self.assertEqual(headers.get('X-Frame-Options'), 'DENY', path) self.assertIn("frame-ancestors 'none'", headers.get('Content-Security-Policy', ''), path) def test_idle_connections_time_out(self): handler = self.httpd.RequestHandlerClass.func # make_server wraps it in functools.partial self.assertGreater(handler.timeout or 0, 0) class TrustLocalhostTest(BridgeHarness, unittest.TestCase): """"trust_localhost": true keeps the no-login local flow (only safe when no proxy in front can rewrite Host to localhost).""" REMOTE = {'hosts': [HOST], 'trust_localhost': True} def test_local_page_gets_the_token_and_sees_every_server(self): cookie = self.page_cookie(self.local) self.assertTrue(cookie.startswith('mcp_token=')) self.assertEqual(self.ids(cookie, host=self.local), (200, ['demo', 'sh', 'sh-remote', 'hidden'])) self.assertEqual(self.request('GET', '/api/mcp/whoami', host=self.local, cookie=cookie)[2], {'mode': 'local'}) def test_localhost_with_forwarding_headers_is_not_local(self): for header in ('X-Forwarded-For', 'X-Real-IP', 'Forwarded', 'X-Forwarded-Host'): self.assertEqual(self.page_cookie(self.local, headers={header: '203.0.113.9'}), '', header) status, _, _ = self.request('GET', '/api/mcp/servers', host=self.local, cookie=self.page_cookie(self.local), headers={header: '203.0.113.9'}) self.assertEqual(status, 403, header) def test_the_local_token_is_no_good_from_outside(self): self.assertEqual(self.ids(self.page_cookie(self.local))[0], 401) class BridgeConfigTest(unittest.TestCase): def test_remote_hosts_without_a_password_refuse_to_start(self): with tempfile.NamedTemporaryFile('w', suffix='.json', delete=False) as f: json.dump({'remote': {'hosts': [HOST]}, 'servers': []}, f) self.addCleanup(pathlib.Path(f.name).unlink) with self.assertRaisesRegex(ValueError, 'WEBTT_MCP_PASSWORD'): mcp_bridge.Bridge(f.name, 0, env={}) def test_server_remote_flag_must_be_boolean(self): with tempfile.NamedTemporaryFile('w', suffix='.json', delete=False) as f: json.dump({'servers': [{'id': 'a', 'remote': 'yes', 'command': ['x']}]}, f) self.addCleanup(pathlib.Path(f.name).unlink) with self.assertRaises(ValueError): mcp_bridge.load_config(f.name) def test_startup_lines_say_what_is_reachable_remotely(self): with tempfile.NamedTemporaryFile('w', suffix='.json', delete=False) as f: json.dump({'remote': {'hosts': [HOST], 'insecure_http': True}, 'servers': [{'id': 'a', 'command': ['x']}, {'id': 's', 'shell': {'allow': ['ls'], 'confirm': False}}]}, f) self.addCleanup(pathlib.Path(f.name).unlink) text = '\n'.join(mcp_bridge.Bridge(f.name, 0, env={'WEBTT_MCP_PASSWORD': PASSWORD}).startup_lines()) self.assertIn(HOST, text) self.assertIn('a', text) self.assertIn('明文', text, 'insecure_http is called out') if __name__ == '__main__': unittest.main()