Merge remote-tracking branch 'origin/main' into theatre-media

# Conflicts:
#	ui/server.py

tests/test_frame_apk_vr.py

@@ -0,0 +1,297 @@
+"""Local-only VR manifest, raw ZIP and independent signature checks."""
+import io
+import os
+from pathlib import Path
+import struct
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+import zipfile
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
+import frame_apk
+import frame_apk_vr as vr
+import frame_apk_sign as signing
+import frame_android
+from test_frame_apk import pool
+
+
+DEFAULT = 'android.intent.category.DEFAULT'
+
+
+def manifest(utf8=False, launcher=False, category=None, samsung=False, split=False, alias=False, splash=False):
+    strings = ['manifest', 'package', 'org.test.vr', 'application', 'activity', 'intent-filter',
+               'action', 'category', 'name', vr.MAIN, category or next(iter(sorted(vr.VR))),
+               vr.LAUNCHER, 'http://schemas.android.com/apk/res/android', 'meta-data', 'value',
+               'com.samsung.android.vr.application.mode', 'vr_only', 'activity-alias', DEFAULT, next(iter(sorted(vr.VR))), 'targetActivity', '.Splash', '.Game']
+    def start(tag, attrs=()):
+        body = struct.pack('<IIHHHHHH', 0xffffffff, strings.index(tag), 20, 20, len(attrs), 0, 0, 0)
+        for name, value in attrs:
+            ns = 0xffffffff if name == 'package' else 12
+            body += struct.pack('<IIIHBBI', ns, strings.index(name), strings.index(value), 8, 0, 3, strings.index(value))
+        return struct.pack('<HHIII', 0x102, 16, 16 + len(body), 1, 0xffffffff) + body
+    def end(tag):
+        return struct.pack('<HHIIIII', 0x103, 16, 24, 1, 0xffffffff, 0xffffffff, strings.index(tag))
+    def leaf(tag, attrs):
+        return start(tag, attrs) + end(tag)
+    b = pool(strings, utf8) + start('manifest', [('package', 'org.test.vr')]) + start('application')
+    if samsung:
+        b += leaf('meta-data', [('name', strings[15]), ('value', 'vr_only')])
+    if splash:  # a helper activity with its own MAIN filter, ahead of the game's
+        b += start('activity', [('name', '.Splash')]) + start('intent-filter') + leaf('action', [('name', vr.MAIN)])
+        b += leaf('category', [('name', DEFAULT)]) + end('intent-filter') + end('activity')
+    b += start('activity', [('name', '.Game')] if splash else []) + start('intent-filter') + leaf('action', [('name', vr.MAIN)])
+    b += leaf('category', [('name', strings[10])])
+    if split:
+        b += end('intent-filter') + start('intent-filter')
+    if launcher:
+        b += leaf('category', [('name', vr.LAUNCHER)])
+    b += end('intent-filter') + end('activity')
+    if alias:  # Godot 4: LAUNCHER only on an alias of the activity ('vr' or 'flat')
+        b += start('activity-alias', [('targetActivity', '.Game')] if splash else []) + start('intent-filter') + leaf('action', [('name', vr.MAIN)])
+        if alias == 'vr':
+            b += leaf('category', [('name', next(iter(sorted(vr.VR))))])
+        b += leaf('category', [('name', vr.LAUNCHER)])
+        b += end('intent-filter') + end('activity-alias')
+    b += end('application') + end('manifest')
+    return struct.pack('<HHI', 3, 8, len(b) + 8) + b
+
+
+class VRTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.tmp = tempfile.TemporaryDirectory()
+        cls.keypath = Path(cls.tmp.name) / 'key.json'
+        cls.key = signing.signing_key(cls.keypath)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.tmp.cleanup()
+
+    def test_manifest_patch(self):
+        for utf8 in (False, True):
+            for category in vr.VR:
+                original = manifest(utf8, category=category)
+                result = vr.add_launcher_category(original)
+                info, filters = vr.inspect(result)
+                self.assertTrue(info['launchable'])
+                self.assertTrue(info['vr'])
+                self.assertIn(vr.LAUNCHER, filters[0]['categories'])
+                self.assertEqual(struct.unpack_from('<I', result, 4)[0], len(result))
+                self.assertEqual(vr.add_launcher_category(result), result)
+                self.assertEqual(vr.add_launcher_category(manifest(utf8, True)), manifest(utf8, True))
+
+    def test_filter_boundaries(self):
+        self.assertFalse(vr.inspect(manifest(launcher=True, split=True))[0]['launchable'])
+        self.assertTrue(vr.inspect(vr.add_launcher_category(manifest(launcher=True, split=True)))[0]['launchable'])
+
+    def test_alias_launcher_is_not_enough(self):
+        # (manifest, still VR after patching)
+        cases = [(manifest(alias='vr'), True),                    # Godot 4 VR export
+                 (manifest(alias='vr', category=DEFAULT), True),  # VR category only on the alias
+                 (manifest(alias='flat', category=DEFAULT), False)]  # Godot 4 flat export
+        for original, is_vr in cases:
+            info, filters = vr.inspect(original)
+            self.assertFalse(info['launchable'])
+            self.assertTrue(info['repairable'])
+            self.assertEqual(len(filters), 1)
+            self.assertFalse(filters[0]['alias'])
+            result = vr.add_launcher_category(original)
+            info, _ = vr.inspect(result)
+            self.assertTrue(info['launchable'])
+            self.assertFalse(info['repairable'])
+            self.assertEqual(info['vr'], is_vr)
+
+    def test_alias_target_activity_is_patched(self):
+        original = manifest(alias='flat', category=DEFAULT, splash=True)
+        info, filters = vr.inspect(original)
+        self.assertTrue(info['repairable'])
+        self.assertEqual(filters[0]['activity'], 'org.test.vr.Game')
+        owner, launchers = None, []
+        for tag, attrs in frame_apk.manifest_elements(vr.add_launcher_category(original)):
+            if tag in ('activity', 'activity-alias'):
+                owner = (tag, attrs.get('name', (0, 0, None))[2])
+            if tag == 'category' and attrs['name'][2] == vr.LAUNCHER:
+                launchers.append(owner)
+        self.assertEqual(launchers, [('activity', '.Game'), ('activity-alias', None)])
+
+    def test_alias_with_launchable_activity_is_left_alone(self):
+        original = manifest(launcher=True, alias='vr')
+        info, _ = vr.inspect(original)
+        self.assertTrue(info['launchable'])
+        self.assertFalse(info['repairable'])
+        self.assertEqual(vr.add_launcher_category(original), original)
+
+    def test_patch_alias_apk(self):
+        with tempfile.TemporaryDirectory() as d:
+            src, dst = Path(d) / 'in.apk', Path(d) / 'out.apk'
+            with zipfile.ZipFile(src, 'w') as z:
+                z.writestr('AndroidManifest.xml', manifest(alias='flat', category=DEFAULT))
+            with patch.object(signing, 'signing_key', return_value=self.key):
+                result = frame_android.patch(src, dst)
+            self.assertEqual(result['patched'], ['launcher'])
+            self.assertTrue(frame_apk.apk_info(dst)['launchable'])
+            self.assertTrue(signing.verify(dst))
+
+    def test_styled_pool(self):
+        for utf8 in (False, True):
+            p = bytearray(pool(['styled'], utf8))
+            old_start = struct.unpack_from('<I', p, 20)[0]
+            p[old_start:old_start] = struct.pack('<I', 0)
+            style_start = len(p)
+            p += struct.pack('<III', 0, 0, 2) + b'\xff' * 12
+            struct.pack_into('<I', p, 4, len(p))
+            struct.pack_into('<I', p, 16, (0x100 if utf8 else 0) | 1)
+            struct.pack_into('<I', p, 12, 1)
+            struct.pack_into('<II', p, 20, old_start + 4, style_start)
+            result, idx = vr._append_string(bytes(p), vr.LAUNCHER)
+            self.assertEqual(frame_apk._string_pool(result, 0), ['styled', vr.LAUNCHER])
+            new_style = struct.unpack_from('<I', result, 24)[0]
+            self.assertEqual(result[new_style:], p[style_start:])
+            self.assertEqual(len(result) % 4, 0)
+
+    def test_detection(self):
+        names = {'lib/arm64-v8a/' + n for n in ('libvrapi.so', 'libopenxr_loader.so', 'libovrplatformloader.so')}
+        info = vr.detection(manifest(category='android.intent.category.LAUNCHER'), names)
+        self.assertTrue(info['vr'])
+        self.assertTrue(info['launchable'])
+        self.assertEqual(len(info['vr_issues']), 3)
+        self.assertFalse(vr.detection(manifest(category=vr.LAUNCHER), set())['vr'])
+        self.assertTrue(vr.detection(manifest(category=vr.LAUNCHER, samsung=True), set())['vr'])
+
+    def test_key_cache(self):
+        with patch.object(signing, '_prime', side_effect=AssertionError('regenerated')):
+            self.assertEqual(signing.signing_key(self.keypath), self.key)
+        if os.name != 'nt':
+            self.assertEqual(self.keypath.stat().st_mode & 0o777, 0o600)
+
+    def test_repack_sign_tamper(self):
+        with tempfile.TemporaryDirectory() as d:
+            src, dst = Path(d) / 'in.apk', Path(d) / 'out.apk'
+            with zipfile.ZipFile(src, 'w') as z:
+                z.writestr('AndroidManifest.xml', manifest(), compress_type=8)
+                z.writestr('classes.dex', b'compress me' * 10000, compress_type=8)
+                z.writestr('resources.arsc', b'1234')
+                z.writestr('lib/arm64-v8a/libx.so', b'ELF' * 500000)
+                z.writestr('META-INF/OLD.RSA', b'old')
+                z.writestr('META-INF/MANIFEST.MF', b'old')
+            with patch.object(signing, 'signing_key', return_value=self.key):
+                result = frame_android.patch(src, dst, {'assets/layer.json': b'{}', 'lib/arm64-v8a/liblayer.so': b'layer'})
+            self.assertEqual(result['patched'], ['launcher'])
+            self.assertTrue(frame_apk.apk_info(dst)['launchable'])
+            self.assertTrue(signing.verify(dst))
+            def compressed(path, info):
+                data = path.read_bytes()
+                nl, el = struct.unpack_from('<HH', data, info.header_offset + 26)
+                start = info.header_offset + 30 + nl + el
+                return data[start:start + info.compress_size], start
+            with zipfile.ZipFile(src) as a, zipfile.ZipFile(dst) as b:
+                self.assertNotIn('META-INF/OLD.RSA', b.namelist())
+                self.assertNotIn('META-INF/MANIFEST.MF', b.namelist())
+                for i in b.infolist():
+                    raw, start = compressed(dst, i)
+                    if i.compress_type == 0:
+                        self.assertEqual(start % (16384 if i.filename.endswith('.so') else 4), 0)
+                    if i.filename in ('classes.dex', 'resources.arsc', 'lib/arm64-v8a/libx.so'):
+                        self.assertEqual(raw, compressed(src, a.getinfo(i.filename))[0])
+                _, off = compressed(dst, b.getinfo('resources.arsc'))
+            original = dst.read_bytes()
+            data = bytearray(original)
+            eo, cd = signing._eocd(data)
+            size = struct.unpack_from('<Q', data, cd - 24)[0]
+            block_start = cd - size - 8
+            value = data[block_start + 20:cd - 24]
+            signer = signing._parts(signing._parts(value)[0])[0]
+            signed, signatures, pub = signing._parts(signer)
+            signature = signing._parts(signing._parts(signatures)[0][4:])[0]
+            sig_offset = data.index(signature, block_start)
+            data[sig_offset] ^= 1
+            dst.write_bytes(data)
+            with self.assertRaisesRegex(ValueError, 'RSA signature'):
+                signing.verify(dst)
+            data = bytearray(original)
+            data[off] ^= 1
+            dst.write_bytes(data)
+            with self.assertRaisesRegex(ValueError, 'digest'):
+                signing.verify(dst)
+
+    @unittest.skipUnless(Path('/usr/bin/openssl').exists(), 'openssl absent')
+    def test_openssl(self):
+        with tempfile.TemporaryDirectory() as d:
+            d = Path(d)
+            (d / 'cert.der').write_bytes(signing.certificate(self.key))
+            (d / 'message').write_bytes(b'independent signature check')
+            (d / 'signature').write_bytes(signing.rsa_sign(b'independent signature check', self.key))
+            def run(*args):
+                return subprocess.run(['/usr/bin/openssl', *args], check=True, capture_output=True).stdout
+            run('x509', '-inform', 'DER', '-in', str(d / 'cert.der'), '-out', str(d / 'cert.pem'))
+            run('verify', '-check_ss_sig', '-CAfile', str(d / 'cert.pem'), str(d / 'cert.pem'))
+            (d / 'pub.pem').write_bytes(run('x509', '-in', str(d / 'cert.pem'), '-pubkey', '-noout'))
+            output = run('dgst', '-sha256', '-verify', str(d / 'pub.pem'), '-signature', str(d / 'signature'), str(d / 'message'))
+            self.assertIn(b'Verified OK', output)
+
+    def test_install_auto_and_override(self):
+        base = {'package': 'org.test.vr', 'label': 'VR', 'abis': [], 'min_sdk': None,
+                'vr': True, 'vr_activity': True, 'launchable': False, 'repairable': True}
+        with patch.object(frame_android, 'apk_info', return_value=base), \
+                patch.object(frame_android, 'xr_compat_files', return_value={}), \
+                patch.object(frame_android, 'patch', return_value={'patched': ['launcher']}) as repair, \
+                patch.object(frame_android, '_install', return_value={}) as install:
+            frame_android.install('original.apk')
+            repair.assert_called_once()
+            self.assertFalse(install.call_args.args[3])
+            self.assertEqual(install.call_args.args[1]['patched'], ['launcher'])
+            frame_android.install('original.apk', flatscreen=True)
+            self.assertTrue(install.call_args.args[3])
+
+    def test_xr_compat_layer(self):
+        with tempfile.TemporaryDirectory() as d:
+            apk = Path(d) / 'a.apk'
+            with zipfile.ZipFile(apk, 'w') as z:
+                z.writestr('lib/arm64-v8a/libopenxr_loader.so', b'')
+            add = frame_android.xr_compat_files(str(apk))
+            self.assertEqual(set(add), set(frame_android.XR_COMPAT_FILES))
+            self.assertIn(b'XR_APILAYER_FRAME_compat', add['assets/openxr/1/api_layers/implicit.d/XrApiLayer_FRAME_compat.json'])
+            self.assertTrue(add['lib/arm64-v8a/libXrApiLayer_FRAME_compat.so'].startswith(b'\x7fELF'))
+            with zipfile.ZipFile(apk, 'a') as z:  # already injected: nothing more to add
+                z.writestr('lib/arm64-v8a/libXrApiLayer_FRAME_compat.so', b'')
+            self.assertEqual(frame_android.xr_compat_files(str(apk)), {})
+            flat = Path(d) / 'flat.apk'
+            with zipfile.ZipFile(flat, 'w') as z:
+                z.writestr('classes.dex', b'')
+            self.assertEqual(frame_android.xr_compat_files(str(flat)), {})
+
+    def test_xr_compat_layer_missing(self):
+        with tempfile.TemporaryDirectory() as d:
+            apk = Path(d) / 'a.apk'
+            with zipfile.ZipFile(apk, 'w') as z:
+                z.writestr('lib/arm64-v8a/libopenxr_loader.so', b'')
+            with patch.object(frame_android, 'XR_COMPAT', d):
+                with self.assertRaisesRegex(frame_android.FrameError, 'build.sh'):
+                    frame_android.xr_compat_files(str(apk))
+
+    def test_patch_rejects_corrupt_manifest_cleanly(self):
+        with patch.object(frame_android, 'apk_info', side_effect=struct.error('bad')):
+            with self.assertRaises(frame_android.FrameError):
+                frame_android.patch('x.apk', 'y.apk')
+
+    def test_install_adds_layer_to_vr_apps(self):
+        base = {'package': 'org.test.vr', 'label': 'VR', 'abis': [], 'min_sdk': None,
+                'vr': True, 'vr_activity': True, 'launchable': True, 'repairable': False}
+        layer = {'x': b''}
+        with patch.object(frame_android, 'apk_info', return_value=dict(base)), \
+                patch.object(frame_android, 'xr_compat_files', return_value=layer), \
+                patch.object(frame_android, 'patch', return_value={'patched': ['openxr-compat']}) as repair, \
+                patch.object(frame_android, '_install', return_value={}) as install:
+            frame_android.install('game.apk')
+            self.assertIs(repair.call_args.args[2], layer)
+            self.assertEqual(install.call_args.args[1]['patched'], ['openxr-compat'])
+            repair.reset_mock()
+            frame_android.install('game.apk', xr_compat=False)  # launchable, no layer: install as is
+            repair.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_macview.py

@@ -0,0 +1,421 @@
+"""Mac in the headset (ui/frame_macview.py and the frame-mac-view agent).
+
+The Python checks run anywhere. On macOS the agent is built and driven over
+HTTP and WebSocket with its test pattern, which needs no Screen Recording
+permission: status, the token, the viewer page, H.264 keyframes, pointer
+input reaching the source, and closing.
+
+Run: python3 -m unittest discover -s tests
+"""
+import base64
+import http.client
+import json
+import os
+import shutil
+import socket
+import struct
+import subprocess
+import sys
+import tempfile
+import time
+import threading
+import unittest
+from unittest import mock
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "ui"))
+
+import frame_macview  # noqa: E402
+
+
+class Helpers(unittest.TestCase):
+    def test_panel_id_is_stable_and_in_range(self):
+        a = frame_macview.panel_id("window:123")
+        self.assertEqual(a, frame_macview.panel_id("window:123"))
+        self.assertNotEqual(a, frame_macview.panel_id("window:124"))
+        self.assertTrue(2_001_000_000 <= a < 2_002_000_000)
+
+    def test_fit_keeps_aspect_inside_the_panel(self):
+        self.assertEqual(frame_macview.fit(2560, 1440), (1920, 1080))
+        w, h = frame_macview.fit(800, 1600)
+        self.assertEqual(h, 1080)
+        self.assertAlmostEqual(w / h, 0.5, places=2)
+
+    @unittest.skipUnless(shutil.which("bash"), "needs bash")
+    @unittest.skipIf(os.name == "nt", "Windows' bash.exe is WSL's launcher, and runners have no distribution")
+    def test_launch_script_parses(self):
+        r = subprocess.run(["bash", "-n"], input=frame_macview.LAUNCH, text=True, capture_output=True)
+        self.assertEqual(r.returncode, 0, r.stderr)
+
+    def test_show_checks_the_source_before_anything_else(self):
+        mv = frame_macview.MacView(["ssh"], lambda *a, **k: self.fail("no ssh"), "frame")
+        with self.assertRaises(frame_macview.MacViewError):
+            mv.show("rm -rf /")
+
+    def test_missing_browser_is_explained(self):
+        calls = []
+
+        def run(remote, stdin=None, timeout=30):
+            calls.append(remote)
+            e = RuntimeError("exit 3")
+            e.stdout = "NO_BROWSER\n"
+            raise e
+
+        mv = frame_macview.MacView(["ssh"], run, "frame")
+        mv.call = lambda path, **kw: {"screen": True, "ticket": "tk"}
+        mv.ensure_tunnel = lambda **kw: None
+        mv.remote_port = 47900
+        with self.assertRaises(frame_macview.MacViewError) as cm:
+            mv.show("window:5")
+        self.assertIn("Chromium", str(cm.exception))
+        self.assertTrue(calls[0].startswith("bash -s -- "))
+
+    def test_separate_windows_need_accessibility(self):
+        mv = frame_macview.MacView(["ssh"], lambda *a, **k: self.fail("no ssh"), "frame")
+        mv.call = lambda path, **kw: {"screen": True, "accessibility": False}
+        with self.assertRaises(frame_macview.MacViewError) as cm:
+            mv.show("separate:42")
+        self.assertIn("Accessibility", str(cm.exception))
+
+    def test_screen_permission_is_checked_before_the_headset(self):
+        mv = frame_macview.MacView(["ssh"], lambda *a, **k: self.fail("no ssh"), "frame")
+        mv.call = lambda path, **kw: {"screen": False}
+        with self.assertRaises(frame_macview.MacViewError) as cm:
+            mv.show("display:1")
+        self.assertIn("Screen Recording", str(cm.exception))
+
+    def test_stop_all_ends_the_viewer_browser_unless_shown_again(self):
+        calls = []
+        mv = frame_macview.MacView(["ssh"], lambda remote, **kw: calls.append(remote) or "", "frame")
+        mv.agent = mock.Mock(poll=lambda: None)
+        mv.call = lambda path, **kw: {"closed": 1}
+        mv.shown = {"window:5"}
+        with mock.patch.object(frame_macview.time, "sleep"):
+            gen = mv.shows
+            mv.shown.clear()
+            mv._end_viewer_browser(gen)
+            self.assertEqual(len(calls), 1)
+            self.assertIn("pkill -f '[f]rame-control/mac-view", calls[0])
+            mv.shows += 1  # Show pressed during the wait: leave the new viewer alone
+            mv._end_viewer_browser(gen)
+            self.assertEqual(len(calls), 1)
+            mv.launching = 1  # a Show replacing its own stream is still launching
+            mv._end_viewer_browser(mv.shows)
+            self.assertEqual(len(calls), 1)
+        # A Show waits while the cleanup checks and runs pkill.
+        mv.launching = 0
+        in_pkill, release, entered = threading.Event(), threading.Event(), threading.Event()
+
+        def blocking_pkill(remote, **kw):
+            in_pkill.set()
+            release.wait(5)
+        mv.run = blocking_pkill
+        mv._show = lambda *a: entered.set() or "shown"
+        with mock.patch.object(frame_macview.time, "sleep"):
+            cleanup = threading.Thread(target=mv._end_viewer_browser, args=(mv.shows,))
+            cleanup.start()
+            self.assertTrue(in_pkill.wait(2))
+            shower = threading.Thread(target=mv.show, args=("window:5",))
+            shower.start()
+            self.assertFalse(entered.wait(0.3), "Show started while the cleanup held the lock")
+            release.set()
+            self.assertTrue(entered.wait(2))
+            cleanup.join()
+            shower.join()
+
+    def test_tunnel_prefers_the_usb_c_network_when_plugged_in(self):
+        mv = frame_macview.MacView(["ssh"], lambda remote, **kw: "13: usb0 inet 10.86.200.233/29 scope global", "frame")
+        ssh_g = mock.Mock(stdout="user steamos\nhostname frame.example.ts.net\n")
+        with mock.patch.object(frame_macview.socket, "create_connection") as conn, \
+                mock.patch.object(frame_macview.subprocess, "run", return_value=ssh_g):
+            self.assertEqual(mv._usb_route(), ["-o", "HostName=10.86.200.233", "-o", "HostKeyAlias=frame.example.ts.net"])
+            conn.assert_called_once_with(("10.86.200.233", 22), timeout=1)
+            ssh_g.stdout = "hostname frame.example.ts.net\nhostkeyalias paired-frame\n"  # a configured alias wins
+            conn.side_effect = None
+            self.assertIn("HostKeyAlias=paired-frame", mv._usb_route())
+            conn.side_effect = OSError("unplugged")
+            self.assertEqual(mv._usb_route(), [])
+        mv.run = lambda remote, **kw: ""  # no usb0
+        self.assertEqual(mv._usb_route(), [])
+        mv.prefer_usb = False
+        mv.run = lambda remote, **kw: self.fail("no ssh when USB is off")
+        self.assertEqual(mv._usb_route(), [])
+
+    def test_a_failed_usb_tunnel_falls_back_to_the_network(self):
+        mv = frame_macview.MacView(["ssh"], lambda *a, **k: "", "frame")
+        mv._usb_route = lambda: ["-o", "HostName=10.86.200.233"]
+        tried = []
+
+        def attempt(via, ports):
+            tried.append(list(via))
+            mv._last_tunnel_error = "Connection refused"
+            return not via  # USB fails, the normal path works
+        mv._open_tunnel = attempt
+        mv.tunnel_up = lambda: False
+        mv.ensure_tunnel()
+        self.assertEqual(tried, [["-o", "HostName=10.86.200.233"], []])
+        self.assertEqual(mv.route, "network")
+
+
+class WS:
+    """A minimal WebSocket client (masked frames out, plain frames in)."""
+
+    def __init__(self, port, path):
+        self.s = socket.create_connection(("127.0.0.1", port), timeout=10)
+        key = base64.b64encode(os.urandom(16)).decode()
+        self.s.sendall(f"GET {path} HTTP/1.1\r\nHost: x\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
+                       f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n".encode())
+        head = b""
+        while b"\r\n\r\n" not in head:
+            head += self.s.recv(1)
+        self.status = int(head.split()[1])
+        self.buf = b""
+
+    def _read(self, n):
+        while len(self.buf) < n:
+            chunk = self.s.recv(65536)
+            if not chunk:
+                raise EOFError
+            self.buf += chunk
+        out, self.buf = self.buf[:n], self.buf[n:]
+        return out
+
+    def recv(self):
+        b0, b1 = self._read(2)
+        n = b1 & 0x7F
+        if n == 126:
+            n = struct.unpack(">H", self._read(2))[0]
+        elif n == 127:
+            n = struct.unpack(">Q", self._read(8))[0]
+        return b0 & 0x0F, self._read(n)
+
+    def send_text(self, text):
+        data, mask = text.encode(), os.urandom(4)
+        head = bytes([0x81, 0x80 | len(data)]) if len(data) < 126 else bytes([0x81, 0xFE]) + struct.pack(">H", len(data))
+        self.s.sendall(head + mask + bytes(c ^ mask[i % 4] for i, c in enumerate(data)))
+
+    def close(self):
+        self.s.close()
+
+
+@unittest.skipUnless(sys.platform == "darwin" and shutil.which("xcrun"), "the agent is macOS-only")
+class Agent(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.tmp = tempfile.mkdtemp()
+        cls.bin = Path(cls.tmp) / "frame-mac-view"
+        r = subprocess.run(["/bin/sh", str(ROOT / "mac" / "frame-mac-view" / "build.sh"), str(cls.bin)],
+                           capture_output=True, text=True, timeout=600)
+        if r.returncode:  # a real failure on a Mac with Xcode: don't hide it as a skip
+            raise AssertionError("agent didn't build:\n" + (r.stderr or r.stdout)[-2000:])
+        cls.token = "t0ken-" + os.urandom(6).hex()
+        cls.proc = subprocess.Popen([str(cls.bin), "serve", "--port", "0", "--page", str(ROOT / "ui" / "mac-view.html"),
+                                     "--exit-on-eof"], env={**os.environ, "FRAME_MAC_VIEW_TOKEN": cls.token},
+                                    stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
+        line = cls.proc.stdout.readline()
+        cls.port = int(line.rsplit(":", 1)[1])
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.proc.stdin.close()  # --exit-on-eof
+        cls.proc.stdout.close()
+        try:
+            cls.proc.wait(5)
+        except subprocess.TimeoutExpired:
+            cls.proc.kill()
+        shutil.rmtree(cls.tmp, ignore_errors=True)
+
+    def get(self, path, method="GET"):
+        c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        c.request(method, path)
+        r = c.getresponse()
+        return r.status, r.read()
+
+    def test_token_is_required(self):
+        self.assertEqual(self.get("/status")[0], 403)
+        self.assertEqual(self.get("/status?k=wrong")[0], 403)
+        status, body = self.get(f"/status?k={self.token}")
+        self.assertEqual(status, 200)
+        self.assertIn("screen", json.loads(body))
+
+    def test_serves_the_viewer_page(self):
+        status, body = self.get(f"/view?k={self.token}&src=test")
+        self.assertEqual(status, 200)
+        self.assertIn(b"VideoDecoder", body)
+
+    def test_snapshot_needs_the_key(self):
+        self.assertEqual(self.get("/snapshot?display=1")[0], 403)
+
+    def test_separate_without_accessibility_explains(self):
+        if json.loads(self.get(f"/status?k={self.token}")[1])["accessibility"]:
+            self.skipTest("this Mac allows Accessibility here")
+        ws = WS(self.port, f"/stream?k={self.token}&src=separate:1")
+        self.assertEqual(ws.status, 101)
+        for _ in range(5):
+            op, data = ws.recv()
+            msg = json.loads(data) if op == 1 else {}
+            if msg.get("t") in ("error", "closed"):
+                break
+        self.assertIn("Accessibility", msg.get("message", msg.get("reason", "")))
+        ws.close()
+
+    def test_lists_displays(self):
+        status, body = self.get(f"/displays?k={self.token}")
+        self.assertEqual(status, 200)
+        self.assertIsInstance(json.loads(body)["displays"], list)
+
+    def ticket(self, src):
+        status, body = self.get(f"/ticket?k={self.token}&src={src}", method="POST")
+        self.assertEqual(status, 200)
+        return json.loads(body)["ticket"]
+
+    def test_ping_and_page_are_open_but_streams_are_not(self):
+        self.assertEqual(self.get("/ping"), (200, b"frame-mac-view"))
+        self.assertEqual(WS(self.port, "/stream?src=test").status, 403)
+        self.assertEqual(self.get("/ticket?src=test", method="POST")[0], 403)
+
+    def test_tickets_are_single_use_and_tied_to_one_source(self):
+        t = self.ticket("test")
+        self.assertEqual(WS(self.port, f"/stream?src=display:1&t={t}").status, 403)  # wrong source
+        ws = WS(self.port, f"/stream?src=test&t={t}")
+        self.assertEqual(ws.status, 101)
+        hello = json.loads(ws.recv()[1])
+        self.assertEqual(hello["t"], "hello")
+        # Until the viewer acknowledges, a retry (the hello got lost) gets the
+        # same key, and replaces the first viewer rather than adding one.
+        retry = WS(self.port, f"/stream?src=test&t={t}")
+        self.assertEqual(retry.status, 101)
+        self.assertEqual(json.loads(retry.recv()[1])["r"], hello["r"])
+        time.sleep(0.3)
+        _, body = self.get(f"/status?k={self.token}")
+        self.assertEqual(len(json.loads(body)["streams"]), 1)
+        ws.close()
+        ws = retry
+        ws.send_text(json.dumps({"t": "ack"}))
+        time.sleep(0.3)
+        self.assertEqual(WS(self.port, f"/stream?src=test&t={t}").status, 403)  # spent
+        again = WS(self.port, f"/stream?src=test&r={hello['r']}")  # the viewer reconnecting
+        self.assertEqual(again.status, 101)
+        again.close()
+        ws.close()
+        # Stop revokes the reconnect key.
+        self.get(f"/close?k={self.token}&src=test", method="POST")
+        self.assertEqual(WS(self.port, f"/stream?src=test&r={hello['r']}").status, 403)
+
+    def test_stop_revokes_tickets_not_yet_used(self):
+        t = self.ticket("test")
+        self.get(f"/close?k={self.token}&src=test", method="POST")
+        self.assertEqual(WS(self.port, f"/stream?src=test&t={t}").status, 403)
+
+    def test_bad_frame_lengths_close_the_socket_not_the_agent(self):
+        ws = WS(self.port, f"/stream?src=test&t={self.ticket('test')}")
+        self.assertEqual(ws.status, 101)
+        # A masked frame claiming 2^63 bytes.
+        ws.s.sendall(bytes([0x81, 0xFF]) + struct.pack(">Q", 1 << 63) + os.urandom(4))
+        with self.assertRaises((EOFError, OSError)):
+            for _ in range(1000):
+                ws.recv()
+        ws.close()
+        self.assertEqual(self.get(f"/status?k={self.token}")[0], 200)
+
+    def test_stream_input_and_close(self):
+        ws = WS(self.port, f"/stream?k={self.token}&src=test&codec=h264&fps=30&max=640")
+        self.assertEqual(ws.status, 101)
+        info = None
+        key = None
+        for _ in range(200):
+            op, data = ws.recv()
+            if op == 1:
+                msg = json.loads(data)
+                if msg["t"] == "hello":
+                    continue
+                if msg["t"] == "error":
+                    self.skipTest("no H.264 encoder here: " + msg["message"])
+                if msg["t"] == "info":
+                    info = msg
+            elif op == 2 and data[0] == 1:
+                key = data
+            if info and key:
+                break
+        self.assertEqual(info["src"], "test")
+        self.assertTrue(info["input"])
+        # A keyframe: flags, 8-byte timestamp, sequence number, input echoed,
+        # then Annex B with the SPS first.
+        self.assertEqual(key[17:21], b"\x00\x00\x00\x01")
+        self.assertEqual(key[21] & 0x1F, 7)  # NAL type 7, SPS
+        ws.send_text(json.dumps({"t": "m", "e": "down", "b": 0, "x": 0.5, "y": 0.5}))
+        ws.send_text(json.dumps({"t": "key-frame"}))
+        got_key = False
+        for _ in range(200):
+            op, data = ws.recv()
+            if op == 2 and data[0] == 1:
+                got_key = True
+                break
+        self.assertTrue(got_key, "no keyframe after asking for one")
+        _, body = self.get(f"/status?k={self.token}")
+        self.assertEqual([s["src"] for s in json.loads(body)["streams"]], ["test"])
+        status, body = self.get(f"/close?k={self.token}", method="POST")
+        self.assertEqual(json.loads(body)["closed"], 1)
+        for _ in range(400):
+            op, data = ws.recv()
+            if op == 1 and json.loads(data)["t"] == "close":
+                break
+        self.assertEqual(json.loads(data)["t"], "close")
+        # Without the viewer doing anything, the agent ends the stream itself.
+        time.sleep(1)
+        _, body = self.get(f"/status?k={self.token}")
+        self.assertEqual(json.loads(body)["streams"], [])
+        ws.close()
+
+    def test_timing_reports_and_input_echo(self):
+        # What a viewer does: sync clocks, report each frame, stamp input.
+        ws = WS(self.port, f"/stream?k={self.token}&src=test&codec=h264&fps=30&max=640")
+        self.assertEqual(ws.status, 101)
+        ws.send_text(json.dumps({"t": "w"}))  # keep-warm filler: ignored
+        ws.send_text(json.dumps({"t": "ping", "c": 1.5}))
+        pong, echoed, seqs, info = None, None, [], None
+        clicked = False
+        deadline = time.time() + 10
+        while time.time() < deadline and not (pong and echoed):
+            op, data = ws.recv()
+            if op == 1:
+                msg = json.loads(data)
+                if msg["t"] == "error":
+                    self.skipTest("no H.264 encoder here: " + msg["message"])
+                if msg["t"] == "pong":
+                    pong = msg
+                if msg["t"] == "info":
+                    info = msg
+            elif op == 2:
+                seq, echo = struct.unpack(">II", data[9:17])
+                seqs.append(seq)
+                now = pong["a"] if pong else 0
+                ws.send_text(json.dumps({"t": "rx", "s": seq, "r": now}))
+                ws.send_text(json.dumps({"t": "fd", "f": [[seq, now + 1, now + 2, now + 3]], "drop": 0}))
+                if echo == 7:
+                    echoed = seq
+                if len(seqs) == 3 and not clicked:
+                    ws.send_text(json.dumps({"t": "m", "e": "down", "b": 0, "x": 0.5, "y": 0.5, "i": 7, "tv": now}))
+                    clicked = True
+        self.assertEqual(pong["c"], 1.5)
+        self.assertGreater(pong["a"], 0)
+        self.assertEqual(seqs[:3], sorted(seqs[:3]))
+        self.assertIsNotNone(echoed, "no frame was tagged as the first reply to the click")
+        time.sleep(1.7)  # frames are reported once the viewer has had time
+        stream = json.loads(self.get(f"/stats?k={self.token}")[1])["streams"][0]
+        first = stream["frames"][0]
+        self.assertGreater(first["e1"], first["e0"])
+        self.assertGreaterEqual(first["e0"], first["cap"])
+        self.assertEqual(first["vs"] - first["rx"], 3)
+        self.assertEqual([i["frame"] for i in stream["inputs"] if i["id"] == 7], [echoed])
+        self.assertIn("total", stream["summary"])
+        self.assertEqual(info["warm"], 0)  # off unless FRAME_MAC_VIEW_WARM is set
+        live = json.loads(self.get(f"/status?k={self.token}")[1])["streams"][0]
+        self.assertEqual(live["controller"]["tier"], 0)
+        self.assertIn("fps", live["stats"])
+        ws.close()
+
+
+if __name__ == "__main__":
+    unittest.main()