diff --git a/CHANGELOG.md b/CHANGELOG.md index ffd6a04..67602f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,11 @@ ## Unreleased +- Add a read-only wired connection doctor for a PC using the Frame's observed + USB-C network: show the local route, SSH port and pinned-key login result + without changing IP, host keys or headset settings. + [Usage and limits](docs/wired-connection.md). + - Start the Android app catalogue without waiting for the headset's installed-app list. Browsing remains available if that status request is slow or fails; installation still waits for a confirmed list to avoid duplicate installs. diff --git a/docs/wired-connection.md b/docs/wired-connection.md new file mode 100644 index 0000000..f6c6de8 --- /dev/null +++ b/docs/wired-connection.md @@ -0,0 +1,27 @@ +# Wired connection doctor + +If you use a USB-C cable because the Wireless Adapter is unreliable, open +**Devices & connection → Wired connection doctor** in the desktop app. Enter +the headset's configured SSH alias and click **Check cable connection**. From +a source checkout, run `python3 scripts/frame-wired-doctor.py --alias frame`. + +The on-demand check reads the computer's selected route to the Frame USB +address, checks whether port 22 answers, then runs remote `true` over SSH with +the existing pinned host key. It does not change network settings or the +headset, accept a new key, or attempt a password. A failure is shown as unknown +or warning; the doctor never claims a cable works from an IP address alone. + +The known address is `10.86.200.233` on the Frame's `10.86.200.232/29` USB +network. A Mac with the Frame attached used `10.86.200.234/29` and completed a +strict SSH login on 2026-09-28; see [the observed link](how-the-frame-works.md). +If the route check says unknown, first confirm a data-capable USB-C cable, a +visible PC network interface and a current Frame OS build. Other OS builds may +use another subnet. If the port answers but SSH identity fails, connect using +the normal pinned headset alias and resolve the pairing or key warning there; +do not disable host-key checking. + +This checks **Frame Control's SSH route**, not whether SteamVR's game stream +uses the cable. The Wireless Adapter and Steam streaming can follow a different +path. No cable is attached to the Linux test host, so this exact UI has only +synthetic route/SSH tests. Windows and Linux physical USB-C behavior, and a +fresh wired readback on the current Frame OS, remain unverified. diff --git a/scripts/frame-wired-doctor.py b/scripts/frame-wired-doctor.py new file mode 100644 index 0000000..8e212cb --- /dev/null +++ b/scripts/frame-wired-doctor.py @@ -0,0 +1,10 @@ +#!/usr/bin/env python3 +"""Source-checkout entry point for the PC's read-only USB-C route checks.""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'ui')) +from frame_wired_doctor import main # noqa: E402 + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/tests/page/wired_doctor.mjs b/tests/page/wired_doctor.mjs new file mode 100644 index 0000000..cd61324 --- /dev/null +++ b/tests/page/wired_doctor.mjs @@ -0,0 +1,41 @@ +// Run the actual wired-doctor page script without browser or device I/O. +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import vm from 'node:vm'; + +class Element { + constructor() { this.value = ''; this.disabled = false; this.textContent = ''; this.children = []; } + append(...nodes) { this.children.push(...nodes); } + prepend(...nodes) { this.children.unshift(...nodes); } + replaceChildren(...nodes) { this.children = nodes; } +} +const elements = Object.fromEntries(['back','alias','check','message','results'].map(id => [id, new Element()])); +elements.alias.value = 'frame'; +const requests = []; +let respond; +const context = vm.createContext({ + document: {getElementById: id => elements[id], createElement: () => new Element()}, + URLSearchParams, location: {search: '?key=test-key'}, + fetch: (url, options) => { requests.push({url, options}); return new Promise(resolve => {respond = resolve;}); }, +}); +const source = readFileSync(new URL('../../ui/wired-doctor.html', import.meta.url), 'utf8'); +vm.runInContext(source.match(/',status: 'unknown',detail: 'No known route'}]})}); +await promise; +assert.equal(elements.results.children.length, 1); +assert.equal(elements.results.children[0].children[0].textContent, '', 'device text must stay plain text'); +assert.equal(elements.check.disabled, false); +promise = elements.check.onclick(); +assert.equal(elements.results.children.length, 0, 'stale result cleared before retry'); +respond({ok: false, json: async () => ({error: 'PC route unavailable'})}); +await promise; +assert.match(elements.message.textContent, /route unavailable/); +assert.equal(elements.results.children.length, 0); +assert.equal(elements.check.disabled, false); +console.log('wired doctor UI: on-demand, escaped data, stale-result clearing and retry passed'); diff --git a/tests/test_wired_doctor.py b/tests/test_wired_doctor.py new file mode 100644 index 0000000..bb3ee88 --- /dev/null +++ b/tests/test_wired_doctor.py @@ -0,0 +1,138 @@ +"""The wired doctor never requires a cable or contacts the real Frame in tests.""" +import sandbox # noqa: F401 +from pathlib import Path +import http.client +import json +import subprocess +import sys +import threading +import unittest +from unittest.mock import patch +import shutil + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'ui')) +import frame_wired_doctor as wired + + +class FakeProbe: + def __init__(self, source='10.86.200.234', port=True, ssh_code=0, stderr=''): + self.source, self.port, self.ssh_code, self.stderr = source, port, ssh_code, stderr + self.calls = [] + + def route_source(self): + self.calls.append('route') + return self.source + + def port_open(self): + self.calls.append('port') + return self.port + + def host_key_name(self, alias): + self.calls.append('config') + return alias + '.example.ts.net' + + def ssh_check(self, alias, name): + self.calls.append(('ssh', alias, name)) + return subprocess.CompletedProcess([], self.ssh_code, '', self.stderr) + + +class WiredDoctor(unittest.TestCase): + @unittest.skipUnless(shutil.which('node'), 'Node is needed for the page script') + def test_page_is_on_demand_and_clears_stale_results(self): + result = subprocess.run(['node', str(Path(__file__).parent / 'page/wired_doctor.mjs')], + capture_output=True, text=True, timeout=15) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_known_usb_route_and_pinned_login(self): + probe = FakeProbe() + report = wired.diagnose('frame', probe=probe) + self.assertEqual(report['state'], 'ready') + self.assertEqual([c['status'] for c in report['checks']], ['ok', 'ok', 'ok']) + self.assertEqual(probe.calls, ['route', 'port', 'config', ('ssh', 'frame', 'frame.example.ts.net')]) + + def test_other_route_never_contacts_target(self): + for source in ('100.91.1.2', '192.168.1.10', '127.0.0.1'): + probe = FakeProbe(source=source) + report = wired.diagnose(probe=probe) + self.assertEqual(report['state'], 'unknown') + self.assertEqual(probe.calls, ['route']) + self.assertIn('no headset was contacted', report['checks'][-1]['detail']) + + def test_closed_port_skips_ssh_and_key_mismatch_is_not_bypassed(self): + closed = FakeProbe(port=False) + self.assertEqual(wired.diagnose(probe=closed)['state'], 'partial') + self.assertEqual(closed.calls, ['route', 'port']) + mismatch = FakeProbe(ssh_code=255, stderr='Host key verification failed.') + report = wired.diagnose(probe=mismatch) + self.assertEqual(report['checks'][-1]['status'], 'warning') + self.assertIn('do not bypass', report['checks'][-1]['detail']) + + def test_invalid_alias_and_headset_mode_do_not_probe(self): + probe = FakeProbe() + with self.assertRaises(ValueError): + wired.diagnose('-oProxyCommand=bad', probe=probe) + self.assertFalse(wired.diagnose(local=True, probe=probe)['supported']) + self.assertEqual(probe.calls, []) + + def test_real_command_uses_strict_identity_and_disables_proxy(self): + with patch.object(wired.frame_host, 'run_ssh', return_value=subprocess.CompletedProcess([], 0, '', '')) as run: + wired.Probe().ssh_check('frame', 'frame.example.ts.net') + argv = run.call_args.args[0] + self.assertIn('StrictHostKeyChecking=yes', argv) + self.assertIn('HostName=10.86.200.233', argv) + self.assertIn('HostKeyAlias=frame.example.ts.net', argv) + self.assertIn('ProxyCommand=none', argv) + self.assertIn('ControlMaster=no', argv) + self.assertEqual(argv[-2:], ['frame', 'true']) + self.assertEqual(run.call_args.kwargs['stdin'], subprocess.DEVNULL) + + +class HttpBoundary(unittest.TestCase): + @classmethod + def setUpClass(cls): + import server + cls.server_module = server + cls.httpd = server.LoopbackServer(('127.0.0.1', 0), server.Handler) + cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True) + cls.thread.start() + + @classmethod + def tearDownClass(cls): + cls.httpd.shutdown() + cls.httpd.server_close() + cls.thread.join() + + def request(self, path, authorized=True): + conn = http.client.HTTPConnection('127.0.0.1', self.httpd.server_address[1], timeout=5) + headers = {'X-Frame-UI': self.server_module.UI_KEY} if authorized else {} + conn.request('GET', path, headers=headers) + response = conn.getresponse() + result = response.status, response.read() + conn.close() + return result + + def test_auth_alias_and_page_without_headset_link(self): + with patch.object(self.server_module, 'ssh', side_effect=AssertionError('no generic SSH')), \ + patch.object(self.server_module, 'working', side_effect=AssertionError('no headset lock')), \ + patch.object(wired, 'diagnose', return_value={'supported': True, 'checks': []}) as probe: + self.assertEqual(self.request('/api/host/wired', authorized=False)[0], 403) + probe.assert_not_called() + code, data = self.request('/api/host/wired?alias=bedroom') + self.assertEqual(code, 200) + self.assertTrue(json.loads(data)['supported']) + probe.assert_called_once_with('bedroom', local=self.server_module.LOCAL) + code, page = self.request('/wired-doctor') + self.assertEqual(code, 200) + self.assertIn(b'No checks have run', page) + + def test_invalid_alias_and_headset_page_guard(self): + self.assertEqual(self.request('/api/host/wired?alias=-bad')[0], 400) + with patch.object(self.server_module, 'LOCAL', True): + self.assertEqual(self.request('/wired-doctor')[0], 400) + code, body = self.request('/api/host/wired') + self.assertEqual(code, 200) + self.assertFalse(json.loads(body)['supported']) + + +if __name__ == '__main__': + unittest.main() diff --git a/ui/frame_wired_doctor.py b/ui/frame_wired_doctor.py new file mode 100644 index 0000000..dfdcf0f --- /dev/null +++ b/ui/frame_wired_doctor.py @@ -0,0 +1,129 @@ +"""Read-only PC checks for the Frame's observed USB-C network path. + +The known Frame USB address was observed on a Mac in 2026-09. It is a clue, +not a promise that every Frame OS or host uses the same address. Never set an +interface address, accept a new SSH host key, or change a network route here. +""" +import argparse +import ipaddress +import json +import re +import socket +import subprocess + +import frame_host + +TARGET = '10.86.200.233' +SUBNET = ipaddress.ip_network('10.86.200.232/29') +ALIAS = re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$') + + +class Probe: + def route_source(self): + # UDP connect chooses a route but sends no packet. + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.connect((TARGET, 22)) + return sock.getsockname()[0] + + def port_open(self): + with socket.create_connection((TARGET, 22), timeout=2): + return True + + def host_key_name(self, alias): + result = frame_host.run_ssh(['ssh', '-G', alias], stdin=subprocess.DEVNULL, + capture_output=True, text=True, timeout=5) + if result.returncode: + raise OSError('SSH configuration unavailable') + settings = dict(line.split(None, 1) for line in result.stdout.splitlines() if ' ' in line) + name = settings.get('hostkeyalias') or settings.get('hostname') or alias + if not name or len(name) > 253 or name.startswith('-') or any(c.isspace() for c in name): + raise ValueError('Invalid SSH host-key name') + return name + + def ssh_check(self, alias, host_key_name): + return frame_host.run_ssh([ + 'ssh', '-T', '-o', 'BatchMode=yes', '-o', 'StrictHostKeyChecking=yes', + '-o', 'NumberOfPasswordPrompts=0', '-o', 'ConnectTimeout=5', + '-o', 'ConnectionAttempts=1', '-o', 'ProxyCommand=none', + '-o', 'ProxyJump=none', '-o', 'PermitLocalCommand=no', + '-o', 'RemoteCommand=none', '-o', 'ClearAllForwardings=yes', + '-o', 'ControlMaster=no', '-o', 'ControlPath=none', + '-o', f'HostName={TARGET}', + '-o', f'HostKeyAlias={host_key_name}', alias, 'true', + ], stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=9) + + +def check(identifier, title, status, detail): + return {'id': identifier, 'title': title, 'status': status, 'detail': detail} + + +def diagnose(alias='frame', *, probe=None, local=False): + """No headset request unless the PC chooses the known local USB subnet.""" + if local: + return {'supported': False, 'state': 'unsupported', 'checks': [], + 'message': 'Run this doctor on your computer, not the headset.'} + if not ALIAS.fullmatch(alias): + raise ValueError('Choose one configured headset alias.') + probe = probe or Probe() + checks = [] + try: + source = probe.route_source() + wired = ipaddress.ip_address(source) in SUBNET and source != TARGET + except (OSError, ValueError): + source, wired = None, False + if wired: + checks.append(check('route', 'PC route to Frame USB address', 'ok', + f'This PC chooses {source} on the previously observed Frame USB subnet. This alone does not prove the cable or headset identity.')) + else: + checks.append(check('route', 'PC route to Frame USB address', 'unknown', + 'No route using the previously observed 10.86.200.232/29 USB subnet. Check a data-capable USB-C cable, port and the PC network interface. This OS may use a different subnet.')) + checks += [check('port', 'USB path SSH port', 'unknown', 'Skipped: the known local USB route was not observed.'), + check('ssh', 'Pinned headset SSH identity', 'unknown', 'Skipped: no known USB route; no headset was contacted.')] + return {'supported': True, 'state': 'unknown', 'alias': alias, 'checks': checks} + try: + open_port = probe.port_open() + except OSError: + open_port = False + if not open_port: + checks += [check('port', 'USB path SSH port', 'warning', + 'Port 22 did not answer on the observed USB route. Check the cable and whether Developer Mode SSH is enabled on the Frame.'), + check('ssh', 'Pinned headset SSH identity', 'unknown', 'Skipped: port 22 did not answer.')] + return {'supported': True, 'state': 'partial', 'alias': alias, 'checks': checks} + checks.append(check('port', 'USB path SSH port', 'ok', 'Port 22 answered on the observed USB route. This does not authenticate the headset.')) + try: + host_key_name = probe.host_key_name(alias) + result = probe.ssh_check(alias, host_key_name) + except (OSError, ValueError, subprocess.TimeoutExpired): + result = None + if result and result.returncode == 0: + checks.append(check('ssh', 'Pinned headset SSH identity', 'ok', + 'A read-only SSH command succeeded over the USB address using strict host-key checking.')) + else: + error = (result.stderr or '') if result else '' + reason = ('The saved host key was not accepted. Verify and pair through the normal headset alias first; do not bypass host-key checking.' + if 'Host key verification failed' in error or 'No ' in error and 'host key is known' in error + else 'The USB SSH login did not complete. Check the saved headset key and Developer Mode SSH; no key or security setting was changed.') + checks.append(check('ssh', 'Pinned headset SSH identity', 'warning', reason)) + return {'supported': True, 'state': 'ready' if checks[-1]['status'] == 'ok' else 'partial', + 'alias': alias, 'checks': checks} + + +def main(argv=None): + parser = argparse.ArgumentParser(description='Read-only wired Frame connection doctor') + parser.add_argument('--alias', default='frame', help='configured SSH alias (default: frame)') + parser.add_argument('--json', action='store_true', help='print a JSON report') + args = parser.parse_args(argv) + try: + report = diagnose(args.alias) + except ValueError as error: + parser.error(str(error)) + if args.json: + print(json.dumps(report, indent=2)) + else: + for item in report['checks']: + print(f"{item['status'].upper():7} {item['title']}: {item['detail']}") + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/ui/index.html b/ui/index.html index fcdbd0b..7ee6232 100644 --- a/ui/index.html +++ b/ui/index.html @@ -1865,6 +1865,7 @@
Frame Control talks to one headset at a time. Each can be reached at several addresses; it tries them all at once and uses the best one that answers on the network you're on.
+

Wired connection doctor ↗
Check the USB-C network route and pinned SSH login from this computer.

This computer's network

Checking…
diff --git a/ui/server.py b/ui/server.py index 8b17530..ec5671f 100755 --- a/ui/server.py +++ b/ui/server.py @@ -76,6 +76,7 @@ import frame_wireless # noqa: E402 import frame_vr # noqa: E402 import frame_utilities # noqa: E402 import frame_wireless_doctor # noqa: E402 +import frame_wired_doctor # noqa: E402 import frame_compat_db # noqa: E402 frame_host.trust_bundled_cas() @@ -3497,11 +3498,22 @@ class Handler(BaseHTTPRequestHandler): self.send_json(frame_agent.approvals.inspect(token)) elif path == "/wireless-doctor": self.send_bytes((HERE / "wireless-doctor.html").read_bytes(), "text/html; charset=utf-8") + elif path == "/wired-doctor": + if LOCAL: + raise Failure("Wired connection doctor runs on a computer, not the headset", 400) + self.send_bytes((HERE / "wired-doctor.html").read_bytes(), "text/html; charset=utf-8") elif path == "/api/wireless-doctor": # PC-only and independent of LINK/working(): never start or wait for SSH. args = parse_qs(url.query) self.send_json(frame_wireless_doctor.diagnose( (args.get("device") or [None])[0], local=LOCAL)) + elif path == "/api/host/wired": + args = parse_qs(url.query) + try: + self.send_json(frame_wired_doctor.diagnose( + (args.get("alias") or ["frame"])[0], local=LOCAL)) + except ValueError as error: + raise Failure(str(error), 400) from error elif path == "/api/steam-pipewire": self.send_json(steam_pipewire_state()) elif path == "/api/host": diff --git a/ui/wired-doctor.html b/ui/wired-doctor.html new file mode 100644 index 0000000..f277e0a --- /dev/null +++ b/ui/wired-doctor.html @@ -0,0 +1,40 @@ + + +Wired connection doctor · Frame Control +
+← Devices & connection +

Wired connection doctor

This computer · Read-only · On demand +

Check whether this computer has the Frame's previously observed USB-C network route, whether SSH answers on it, and whether a strictly checked SSH identity works. This does not change network or headset settings.

+
+

No checks have run.

+
+

The observed USB address is 10.86.200.233 on Frame OS tested in 2026. A different OS build may use another subnet. An unknown result does not prove the cable is broken. The doctor never changes IP settings, accepts a new host key, or bypasses SSH warnings.

+