Add read-only wired connection doctor (#178)

Co-authored-by: SAPHID <saphid@users.noreply.github.com>

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.

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.

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())

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(/<script>([\s\S]*)<\/script>/)[1], context);
+assert.equal(requests.length, 0, 'page load must not probe the PC or Frame');
+assert.equal(elements.back.href, '/?key=test-key#devices');
+let promise = elements.check.onclick();
+assert.equal(elements.check.disabled, true);
+assert.equal(requests[0].url, '/api/host/wired?alias=frame');
+assert.equal(requests[0].options.headers['X-Frame-UI'], 'test-key');
+respond({ok: true, json: async () => ({state: 'unknown', checks: [{title: '<script>bad</script>',status: 'unknown',detail: 'No known route'}]})});
+await promise;
+assert.equal(elements.results.children.length, 1);
+assert.equal(elements.results.children[0].children[0].textContent, '<script>bad</script>', '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');

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()

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())

ui/index.html

@@ -1865,6 +1865,7 @@
       <div class="hint">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.</div>
       <p id="wirelessDoctorLink" hidden><a href="/wireless-doctor">Linux Wireless Adapter doctor ↗</a><br><span class="sub">Read-only PC checks; works without a connected Frame.</span></p>
+      <p><a href="/wired-doctor">Wired connection doctor ↗</a><br><span class="sub">Check the USB-C network route and pinned SSH login from this computer.</span></p>
       <h3 style="margin-top:22px">This computer's network</h3>
       <div id="netNow" class="sub">Checking…</div>
       <div class="row" style="margin-top:8px; flex-wrap:nowrap">

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":

ui/wired-doctor.html

@@ -0,0 +1,40 @@
+<!doctype html>
+<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
+<title>Wired connection doctor · Frame Control</title>
+<style>
+:root{color-scheme:dark;--bg:#101820;--card:#1b2838;--text:#e4edf5;--sub:#acbccb;--line:#3a5065;--blue:#66c0f4;--warn:#f0c67a}
+*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--text);font:16px/1.55 system-ui,sans-serif}main{max-width:850px;margin:auto;padding:32px 24px 64px}a{color:var(--blue)}h1{font-size:30px;line-height:1.2;margin:20px 0 12px}h2{font-size:18px;margin:0 0 8px}p{margin:8px 0 16px}.sub{color:var(--sub)}.badge{font-size:12px;letter-spacing:.05em;text-transform:uppercase;color:var(--blue)}section{background:var(--card);border:1px solid var(--line);border-radius:10px;padding:20px;margin:18px 0}.row{display:flex;gap:12px;align-items:end;flex-wrap:wrap}label{display:block}input,button{font:inherit;min-height:44px;border:1px solid var(--line);border-radius:5px;padding:9px 14px;background:#263f54;color:var(--text)}input{display:block;min-width:220px;margin-top:6px}button{cursor:pointer;background:#205b81;border-color:var(--blue)}button:disabled{opacity:.5;cursor:default}.status{font-size:12px;font-weight:700;text-transform:uppercase;margin-right:10px;color:var(--blue)}.warning .status{color:var(--warn)}.unknown .status{color:var(--sub)}#message{min-height:24px}:focus-visible{outline:2px solid var(--blue);outline-offset:4px}@media(max-width:540px){main{padding:20px 16px}h1{font-size:25px}section{padding:16px}}
+</style></head><body><main>
+<a id="back" href="/#devices">← Devices &amp; connection</a>
+<header><h1>Wired connection doctor</h1><span class="badge">This computer · Read-only · On demand</span>
+<p>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.</p></header>
+<section><div class="row"><label for="alias">Your configured headset SSH alias<input id="alias" value="frame" maxlength="64" autocomplete="off"></label>
+<button id="check">Check cable connection</button></div><p id="message" role="status" aria-live="polite">No checks have run.</p></section>
+<div id="results" aria-live="polite"></div>
+<p class="sub">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.</p>
+</main><script>
+'use strict';
+const $ = id => document.getElementById(id);
+const key = new URLSearchParams(location.search).get('key') || '1';
+$('back').href = '/?key=' + encodeURIComponent(key) + '#devices';
+const make = (tag, value) => { const node = document.createElement(tag); node.textContent = value; return node; };
+$('check').onclick = async () => {
+  $('check').disabled = true;
+  $('results').replaceChildren();
+  $('message').textContent = 'Checking this computer…';
+  try {
+    const response = await fetch('/api/host/wired?alias=' + encodeURIComponent($('alias').value.trim()),
+      {headers:{'X-Frame-UI':key}});
+    const report = await response.json();
+    if (!response.ok) throw new Error(report.error || 'Wired check failed');
+    for (const item of report.checks) {
+      const section = make('section', ''); section.className = item.status;
+      const title = make('h2', item.title); const status = make('span', item.status + ' · '); status.className = 'status';
+      title.prepend(status); section.append(title, make('p', item.detail)); $('results').append(section);
+    }
+    $('message').textContent = report.state === 'ready' ? 'The known USB path and pinned SSH identity worked.'
+      : 'Review each check and its limits below.';
+  } catch (error) { $('message').textContent = error.message; }
+  finally { $('check').disabled = false; }
+};
+</script></body></html>