158 lines
8.3 KiB
Python
158 lines
8.3 KiB
Python
|
"""Protocol fixtures exercise bounded guest probes, not a real GUI session."""
|
||
|
|
import base64
|
||
|
|
import hashlib
|
||
|
|
import http.server
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
import threading
|
||
|
|
from types import SimpleNamespace
|
||
|
|
import unittest
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
SOURCE = Path(__file__).resolve().parents[1] / 'crates/supervisor/src/runtime_probe.py'
|
||
|
|
module = {'__name__': 'runtime_probe_test'}
|
||
|
|
exec(compile(SOURCE.read_text(), str(SOURCE), 'exec'), module)
|
||
|
|
|
||
|
|
|
||
|
|
class ProbeTests(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
fixture = self
|
||
|
|
self.mode = 'valid'
|
||
|
|
self.requests = []
|
||
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
||
|
|
protocol_version = 'HTTP/1.1'
|
||
|
|
def log_message(self, *_):
|
||
|
|
pass
|
||
|
|
def do_GET(self):
|
||
|
|
self.close_connection = True
|
||
|
|
fixture.requests.append(self.path)
|
||
|
|
if self.path == '/websockify':
|
||
|
|
self.send_response(200 if fixture.mode == 'no-upgrade' else 101)
|
||
|
|
key = self.headers['Sec-WebSocket-Key']
|
||
|
|
accepted = base64.b64encode(hashlib.sha1((key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11').encode()).digest()).decode()
|
||
|
|
self.send_header('Upgrade', 'websocket')
|
||
|
|
self.send_header('Connection', 'Upgrade')
|
||
|
|
self.send_header('Sec-WebSocket-Protocol', 'binary')
|
||
|
|
self.send_header('Sec-WebSocket-Accept', 'wrong' if fixture.mode == 'wrong-accept' else accepted)
|
||
|
|
self.end_headers()
|
||
|
|
data = {
|
||
|
|
'bad-rfb': b'\x82\x0cNOT A VIEWER',
|
||
|
|
'oversized-frame': b'\x82\x7e',
|
||
|
|
'masked-frame': b'\x82\x8c',
|
||
|
|
'split-frame': b'\x82\x04RFB \x82\x08003.008\n',
|
||
|
|
}.get(fixture.mode, b'\x82\x0cRFB 003.008\n')
|
||
|
|
try:
|
||
|
|
self.wfile.write(data)
|
||
|
|
self.wfile.flush()
|
||
|
|
except (BrokenPipeError, ConnectionResetError):
|
||
|
|
pass
|
||
|
|
self.close_connection = True
|
||
|
|
return
|
||
|
|
if self.path == '/json/version':
|
||
|
|
body = json.dumps({'Browser': 'Chrome/fixture', 'Protocol-Version': '1.3',
|
||
|
|
'webSocketDebuggerUrl': f'ws://127.0.0.1:{fixture.port}/devtools/browser/test'}).encode()
|
||
|
|
if fixture.mode == 'foreign-browser':
|
||
|
|
body = body.replace(b'127.0.0.1', b'example.com')
|
||
|
|
if fixture.mode == 'bad-browser':
|
||
|
|
body = b'not JSON'
|
||
|
|
if fixture.mode == 'oversized-browser':
|
||
|
|
body = b' ' * 65537
|
||
|
|
elif self.path == '/vnc_lite.html':
|
||
|
|
body = b'<!doctype html><title>viewer fixture</title>'
|
||
|
|
else:
|
||
|
|
self.send_error(404)
|
||
|
|
return
|
||
|
|
self.send_response(302 if fixture.mode == 'redirect' else 200)
|
||
|
|
self.send_header('Content-Type', 'text/plain' if fixture.mode == 'not-html' else 'text/html')
|
||
|
|
self.send_header('Content-Length', str(len(body)))
|
||
|
|
self.end_headers()
|
||
|
|
try:
|
||
|
|
self.wfile.write(body)
|
||
|
|
except (BrokenPipeError, ConnectionResetError):
|
||
|
|
pass
|
||
|
|
self.server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler)
|
||
|
|
self.port = self.server.server_port
|
||
|
|
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
||
|
|
self.addCleanup(self.server.server_close)
|
||
|
|
self.addCleanup(self.server.shutdown)
|
||
|
|
|
||
|
|
def test_browser_requires_bounded_local_devtools_discovery(self):
|
||
|
|
for mode, expected in [('valid', True), ('foreign-browser', False), ('bad-browser', False),
|
||
|
|
('oversized-browser', False), ('redirect', False)]:
|
||
|
|
self.mode = mode
|
||
|
|
self.assertEqual(module['browser_probe'](self.port), expected, mode)
|
||
|
|
self.assertEqual(self.requests, ['/json/version'] * 5)
|
||
|
|
|
||
|
|
def test_viewer_requires_websocket_to_rfb_not_just_http(self):
|
||
|
|
for mode, expected in [('valid', True), ('split-frame', True), ('bad-rfb', False),
|
||
|
|
('wrong-accept', False), ('no-upgrade', False), ('oversized-frame', False),
|
||
|
|
('masked-frame', False), ('not-html', False), ('redirect', False)]:
|
||
|
|
self.mode = mode
|
||
|
|
self.assertEqual(module['viewer_probe'](self.port), expected, mode)
|
||
|
|
self.assertNotIn('/json/version', self.requests)
|
||
|
|
|
||
|
|
def test_desktop_needs_a_window_manager_as_well_as_x11(self):
|
||
|
|
run = module['subprocess'].run
|
||
|
|
with patch.object(module['subprocess'], 'run', side_effect=[
|
||
|
|
SimpleNamespace(returncode=0), SimpleNamespace(returncode=0, stdout=b'_NET_SUPPORTING_WM_CHECK(WINDOW): window id # 0x200001'),
|
||
|
|
SimpleNamespace(returncode=0, stdout=b'_NET_WM_NAME(UTF8_STRING) = "Xfwm4"')]) as call:
|
||
|
|
self.assertTrue(module['desktop_probe'](2))
|
||
|
|
self.assertEqual(call.call_args_list[0].args[0], ['xdpyinfo', '-display', ':3'])
|
||
|
|
self.assertIn('0x200001', call.call_args_list[2].args[0])
|
||
|
|
with patch.object(module['subprocess'], 'run', side_effect=[
|
||
|
|
SimpleNamespace(returncode=0), SimpleNamespace(returncode=0, stdout=b'not found')]):
|
||
|
|
self.assertFalse(module['desktop_probe'](2))
|
||
|
|
with patch.object(module['subprocess'], 'run', side_effect=FileNotFoundError):
|
||
|
|
self.assertFalse(module['desktop_probe'](2))
|
||
|
|
self.assertIs(module['subprocess'].run, run)
|
||
|
|
|
||
|
|
def test_no_assignment_never_probes_a_peer_and_rejects_invalid_slots(self):
|
||
|
|
original = {key: module[key] for key in ('desktop_probe', 'browser_probe', 'viewer_probe')}
|
||
|
|
try:
|
||
|
|
for key in original:
|
||
|
|
module[key] = lambda *_: self.fail('unassigned slot must not be probed')
|
||
|
|
result = module['probe'](None)
|
||
|
|
self.assertIsNone(result['desktop'])
|
||
|
|
self.assertIsNone(result['browser'])
|
||
|
|
self.assertIsNone(result['viewer'])
|
||
|
|
for slot in (True, False, -1, 8, '1', [], {}):
|
||
|
|
with self.assertRaisesRegex(ValueError, 'INVALID_SCREEN_SLOT'):
|
||
|
|
module['probe'](slot)
|
||
|
|
finally:
|
||
|
|
module.update(original)
|
||
|
|
|
||
|
|
def test_root_timeout_errors_do_not_echo_the_helper_or_create_a_job(self):
|
||
|
|
scope = {'__name__': 'root_health_test'}
|
||
|
|
for filename in ('runner_files.py', 'runner_jobs.py'):
|
||
|
|
source = SOURCE.parent / filename
|
||
|
|
exec(compile(source.read_text(), str(source), 'exec'), scope)
|
||
|
|
scope.update(connect=lambda: object(), RUNTIME_PROBE_SOURCE='PRIVATE_HELPER_CANARY')
|
||
|
|
request = {'action': 'runtime_health', 'identity': {'computer_id': 'computer', 'bot_id': 'bot', 'generation': 1},
|
||
|
|
'input': '{"slot":null}'}
|
||
|
|
errors = [scope['subprocess'].TimeoutExpired(['PRIVATE_HELPER_CANARY'], 8), OSError('PRIVATE_HELPER_CANARY')]
|
||
|
|
for error in errors:
|
||
|
|
with patch.object(scope['subprocess'], 'run', side_effect=error) as run:
|
||
|
|
with self.assertRaises(ValueError) as caught:
|
||
|
|
scope['handle'](request)
|
||
|
|
self.assertEqual(str(caught.exception), 'RUNTIME_HEALTH_UNAVAILABLE')
|
||
|
|
self.assertEqual(run.call_args.kwargs['timeout'], 8)
|
||
|
|
self.assertEqual(run.call_args.kwargs['user'], 1000)
|
||
|
|
|
||
|
|
def test_components_fail_independently_and_ports_follow_the_slot(self):
|
||
|
|
original = {key: module[key] for key in ('desktop_probe', 'browser_probe', 'viewer_probe')}
|
||
|
|
seen = []
|
||
|
|
try:
|
||
|
|
module['desktop_probe'] = lambda slot: seen.append(('desktop', slot)) or False
|
||
|
|
module['browser_probe'] = lambda port: seen.append(('browser', port)) or True
|
||
|
|
module['viewer_probe'] = lambda port: seen.append(('viewer', port)) or False
|
||
|
|
with patch.object(module['os'], 'getuid', return_value=1000):
|
||
|
|
result = module['probe'](3)
|
||
|
|
self.assertEqual(result, {'version': 1, 'slot': 3, 'runner': True, 'desktop': False, 'browser': True, 'viewer': False})
|
||
|
|
self.assertEqual(seen, [('desktop', 3), ('browser', 9225), ('viewer', 6083)])
|
||
|
|
finally:
|
||
|
|
module.update(original)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
unittest.main()
|