Merge pull request #88 from saphid/fix/tests-real-ssh-dir

Tests: keep every test (and the servers they start) off the real ~/.ssh

app/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "frame-control",
-  "version": "0.4.2",
+  "version": "0.4.4",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "frame-control",
-      "version": "0.4.2",
+      "version": "0.4.4",
       "license": "MIT",
       "devDependencies": {
         "electron": "^44.4.5",

app/package.json

@@ -1,7 +1,7 @@
 {
   "name": "frame-control",
   "productName": "Frame Control",
-  "version": "0.4.2",
+  "version": "0.4.4",
   "description": "Desktop app for managing a Valve Steam Frame over SSH",
   "private": true,
   "main": "main.js",

docs/devices.md

@@ -104,7 +104,9 @@ connection is an SSH ControlMaster that every command shares; when it dies (the
 headset slept or left the network) the connector notices and starts again. On
 Windows, where OpenSSH can't share a connection, the same handshake runs once
 and each command then connects on its own; a command that can't reach the
-headset makes the connector start again.
+headset makes the connector start again. There ssh's stderr goes to a file that
+the connector reads as it grows, not a pipe: Windows' OpenSSH can block writing
+to a redirected pipe.
 
 Once connected, every `ssh`, `scp` and `rsync` the app runs gets
 `-o HostName=<address> -o HostKeyAlias=frame-control-<id>
@@ -121,9 +123,57 @@ lease that moved) would look like a new host there. The app keeps one known_host
 file per headset instead, so saving or forgetting one headset's key never touches
 another's: a different device answering at one of its
 addresses is refused, and the pill says so. A headset's first connection trusts
-the key it shows, as Set Up Connection does. After reinstalling SteamOS the
-headset has a new key; **Forget identity** on the Devices tab lets the next
-connection save the new one.
+the key it shows, as Set Up Connection does: while a headset has never had an
+identity here, every ssh command gets `StrictHostKeyChecking=accept-new` (the
+first contact saves it, whichever command makes it), and from then on
+`StrictHostKeyChecking=yes`. `accept-new` never replaces a saved key, so it
+can't hide a change. The saved key's fingerprint is also recorded in
+`devices.json` (`identity`), along with a flag that never clears
+(`identity_established`), as soon as ssh saves the key: even if the login is then
+refused. A headset counts as having had an identity if the flag is set, a
+fingerprint is recorded, its pin file exists at all, or a connection to it ever
+worked (older files get the flag from that when loaded, so editing or removing
+the address that worked doesn't undo it); then
+a pin file that's deleted, empty, unreadable or not parseable **fails closed**
+(*Checking the headset's identity* fails, "missing or can't be read") rather than
+trusting whatever answers, and only **Check its key…** saves a key again. It
+says when the key it reads matches the recorded fingerprint.
+
+**When the key changes** (SteamOS reinstalled or reset, a new headset at the
+same address, or a different device answering there), nothing is accepted by
+itself. The connection fails at *Checking the headset's identity*; the banner
+and the headset on the Devices tab say so, with **Check the new key…** (also
+**Check its key…** on the Devices tab at any time). That reads the key the
+headset shows now, and nothing more: ssh writes it to a throwaway known_hosts
+file during the key exchange, before any login, and is stopped as soon as it
+has. Every other kind of contact is ruled out on the command line, ahead of
+anything `~/.ssh/config` says: no authentication of any kind, no agent
+(`IdentityAgent=none`, `ForwardAgent=no`), no X11 or port forwarding, no local
+or remote command (`-N`), no terminal, and `-F none`. Through a ProxyJump host
+it builds that hop itself, as a ProxyCommand whose ssh forwards nothing and runs
+no local command (ProxyJump's own ssh would take `ForwardAgent` and
+`LocalCommand` from `~/.ssh/config`); it still logs in to the jump host as
+configured. Every word of that ProxyCommand is quoted for the shell that runs it
+(and `%` doubled for ssh's own tokens), the forward is `[%h]:%p` so an IPv6
+headset works, and an IPv6 jump host is passed bare with its port. Through a ProxyCommand from `~/.ssh/config`, or several jump hosts,
+it refuses, since those can't be made safe from here: add an address it can
+reach directly, or compare the key by hand. It shows the key's SHA256 fingerprint next to
+the saved one, with the command that prints it on the headset
+(`ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub`, in Konsole on its desktop).
+**This is my headset: trust its new key** then saves exactly that key, replacing
+the old one, and reconnects. The server keeps the key it showed for 10 minutes
+and saves it only if the page sends back the same fingerprint, and only while
+the headset's port, user and that address are unchanged; it never saves whatever
+answers next. The API's old `forget-identity`, which deleted the pin so that the
+next contact saved any key unseen, is refused. Set Up Connection can't fix a changed key: it checks your
+own `known_hosts`, not the headset's file.
+
+**When the headset refuses this computer's key** (`Permission denied
+(publickey,…)`), the connection stops at *Logging in*, and the banner and the
+Devices tab offer **Set Up Connection…**, which pairs this computer again.
+
+Neither is reported as a fault: the diagnostics report each kind at most once
+a session ([privacy.md](privacy.md)).
 
 ## One server at a time
 
@@ -144,7 +194,7 @@ All under the usual `/api/` guards (loopback `Host`, `X-Frame-UI` header).
 | `GET /api/devices` | Headsets, the current network, known networks, the next free alias |
 | `GET /api/devices/tailscale?id=` | Tailscale peers, likely headsets first |
 | `GET /api/devices/mdns?id=` | Headsets found on this network |
-| `POST /api/devices` | `{"action": ...}`: `use`, `update` (name, user, port), `remove`, `address-add`, `address-update`, `address-remove`, `address-move`, `test`, `forget-identity`, `name-network`, `setup` (alias, optional host), `retry` |
+| `POST /api/devices` | `{"action": ...}`: `use`, `update` (name, user, port), `remove`, `address-add`, `address-update`, `address-remove`, `address-move`, `test`, `identity-scan` (read the key the headset shows: `scan` with `type`, `fingerprint`, `host`, `saved`, `recorded`, `matches`, `recorded_match`), `identity-trust` (`fingerprint`: save the key `identity-scan` showed), `forget-identity` (refused: use the two before), `name-network`, `setup` (alias, optional host), `retry` |
 
 Every host, alias and user is checked against strict patterns before it's
 stored, because they end up in ssh arguments and `~/.ssh/config`; nothing goes
@@ -161,7 +211,7 @@ share the format later (it still connects to one host; see
 ```json
 {"version": 1, "active": "f67f8b7e",
  "devices": [{"id": "f67f8b7e", "name": "Steam Frame", "alias": "frame", "user": "steamos", "port": 22,
-              "identity_files": ["~/.ssh/id_ed25519_frame"],
+              "identity_files": ["~/.ssh/id_ed25519_frame"], "identity": ["SHA256:…"], "identity_established": true,
               "addresses": [{"host": "frame.local", "kind": "mdns", "label": "",
                              "networks": ["n-e0998baa61"], "last_ok": 1790593550.4, "last_rtt_ms": 0.9}]}],
  "networks": {"n-e0998baa61": {"name": "Home Wi-Fi", "ssid": null, "gateway": "192.168.1.1",

docs/privacy.md

@@ -101,8 +101,10 @@ scrubbed:
 - `token=`, `key=`, `password=` and similar values are replaced
 
 The same error is sent at most once every 10 minutes. When ssh reports that
-it couldn't reach the Frame (asleep, away, or not set up yet), that is sent
-at most once per kind each time Frame Control runs.
+it couldn't reach the Frame (asleep, away, or not set up yet), or that ssh and
+the Frame refused each other's keys (a changed host key, or this computer's
+key not accepted), that is sent at most once per kind each time Frame Control
+runs.
 
 ## Report a problem
 

tests/apk_fixtures.py

@@ -0,0 +1,87 @@
+"""Small APKs built in memory (binary manifest plus resource table), for test_frame_apk
+and the end-to-end tests. Deliberately doesn't import sandbox: tests/e2e runs in a
+throwaway container and keeps its real home, which sandbox would move."""
+import io
+import os
+import struct
+import sys
+import zipfile
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'ui'))
+import frame_apk  # noqa: E402
+
+
+def pool(strings, utf8=False):
+    """A ResStringPool chunk."""
+    data, offsets = b'', []
+    for s in strings:
+        offsets.append(len(data))
+        if utf8:
+            b = s.encode()
+            data += bytes([len(s), len(b)]) + b + b'\0'
+        else:
+            data += struct.pack('<H', len(s)) + s.encode('utf-16-le') + b'\0\0'
+    data += b'\0' * (-len(data) % 4)
+    start = 28 + 4 * len(strings)
+    body = struct.pack(f'<{len(strings)}I', *offsets) + data
+    return struct.pack('<HHIIIIII', 1, 28, 28 + len(body), len(strings), 0, 0x100 if utf8 else 0, start, 0) + body
+
+
+def manifest(package, label_ref, version_ref, min_sdk, package_raw=True, foreign_label=False):
+    """<manifest package versionName><uses-sdk minSdkVersion/><application label icon/></manifest>.
+
+    package_raw=False drops the package's raw string (as some repackers do);
+    foreign_label adds a non-android `label` attribute after android:label.
+    """
+    strings = ['label', 'icon', 'versionName', 'minSdkVersion', 'package', 'manifest', 'uses-sdk',
+               'application', package, 'junk', 'label', 'versionCode']  # the second 'label' has no android id
+    resmap = struct.pack('<4I', 0x01010001, 0x01010002, 0x0101021c, 0x0101020c)
+    resmap = struct.pack('<HHI', 0x0180, 8, 8 + len(resmap)) + resmap
+
+    def element(name, attrs):
+        body = struct.pack('<IIHHHHHH', 0xffffffff, name, 20, 20, len(attrs), 0, 0, 0)
+        for aname, raw, dtype, value in attrs:
+            body += struct.pack('<IIIHBBI', 0xffffffff, aname, raw, 8, 0, dtype, value)
+        return struct.pack('<HHIII', 0x0102, 16, 16 + len(body), 1, 0xffffffff) + body
+
+    none = 0xffffffff
+    chunks = (pool(strings) + resmap
+              + element(5, [(4, 8 if package_raw else none, frame_apk.T_STRING, 8),
+                            (2, none, frame_apk.T_REF, version_ref),
+                            (11, none, frame_apk.T_INT_DEC, 210)])
+              + element(6, [(3, none, frame_apk.T_INT_DEC, min_sdk)])
+              + element(7, [(0, none, frame_apk.T_REF, label_ref), (1, none, frame_apk.T_REF, 0x7f020000)]
+                        + ([(10, 9, frame_apk.T_STRING, 9)] if foreign_label else [])))
+    return struct.pack('<HHI', 3, 8, 8 + len(chunks)) + chunks
+
+
+def resources(values):
+    """resources.arsc with package 0x7f; values: {(type id, entry, language, density): global string index}."""
+    strings = ['French label', 'App label', '2.1', 'res/icon_lo.png', 'res/icon_hi.png', 'res/icon.xml']
+    pkg_body = b''
+    for (tid, lang, density), entries in values.items():
+        cfg = struct.pack('<I4x2s4xH', 64, lang.encode().ljust(2, b'\0'), density).ljust(64, b'\0')
+        count = max(entries) + 1
+        offsets, data = [], b''
+        for i in range(count):
+            if i in entries:
+                offsets.append(len(data))
+                data += struct.pack('<HHI', 8, 0, 0) + struct.pack('<HBBI', 8, 0, frame_apk.T_STRING, entries[i])
+            else:
+                offsets.append(0xffffffff)
+        header = 20 + 64
+        estart = header + 4 * count
+        body = struct.pack(f'<{count}I', *offsets) + data
+        pkg_body += struct.pack('<HHIBBHII', 0x0201, header, header + len(body), tid, 0, 0, count, estart) + cfg + body
+    pkg_header = struct.pack('<HHII', 0x0200, 288, 288 + len(pkg_body), 0x7f).ljust(288, b'\0')
+    pkg = pkg_header + pkg_body
+    table = pool(strings, utf8=True) + pkg
+    return struct.pack('<HHII', 2, 12, 12 + len(table), 1) + table
+
+
+def apk(files):
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, 'w') as z:
+        for name, data in files.items():
+            z.writestr(name, data)
+    return buf.getvalue()

tests/e2e/harness.py

@@ -10,6 +10,7 @@ import atexit
 import http.client
 import json
 import os
+import shutil
 import socket
 import subprocess
 import sys
@@ -22,7 +23,22 @@ from pathlib import Path
 ROOT = Path(__file__).resolve().parents[2]
 sys.path[:0] = [str(ROOT / 'ui'), str(ROOT / 'tests'), str(ROOT / 'tests' / 'smoke')]
 
-ENABLED = os.environ.get('FRAME_E2E') == '1'
+# Only inside the compose `host` container (FRAME_E2E_CONTAINER from compose.yaml, plus
+# the container runtime's marker file): these tests drive ui/server.py with the real
+# home, ~/.ssh and cache, which is only safe in a throwaway container.
+REQUESTED = os.environ.get('FRAME_E2E') == '1'
+IN_CONTAINER = os.environ.get('FRAME_E2E_CONTAINER') == '1' and (
+    os.path.exists('/.dockerenv') or os.path.exists('/run/.containerenv'))
+ENABLED = REQUESTED and IN_CONTAINER
+if ENABLED:
+    # What tests/sandbox.py does, minus moving HOME: the container's real ~/.ssh is what
+    # points `ssh frame` at the fake Frame, and OpenSSH reads ~/.ssh from the password
+    # database, not $HOME. (These tests used to get this through test_frame_apk.)
+    _data = tempfile.mkdtemp(prefix='frame-control-e2e-')
+    atexit.register(shutil.rmtree, _data, ignore_errors=True)
+    os.environ['FRAME_CONTROL_DATA_DIR'] = _data
+    os.environ['FRAME_CONTROL_TELEMETRY'] = '0'
+    os.environ['FRAME_COMPAT_DB_URL'] = 'http://127.0.0.1:9'
 CTL = os.environ.get('FAKEFRAME_CTL', 'http://fakeframe:9999').rstrip('/')
 FAKE_HOST = os.environ.get('FAKEFRAME_HOST', 'fakeframe')
 HOME = '/home/steamos'
@@ -30,9 +46,14 @@ SERVER_LOG = os.path.join(tempfile.gettempdir(), 'fakeframe-e2e-server.log')
 
 
 def require():
-    """Call at module level: skips the module unless the fake Frame is up."""
-    if not ENABLED:
-        raise unittest.SkipTest('needs the fake Frame: run scripts/e2e.sh (sets FRAME_E2E=1)')
+    """Skips (the module, at module level; the class, in setUpClass) unless this is
+    scripts/e2e.sh's container with the fake Frame up."""
+    if ENABLED:
+        return
+    if REQUESTED:
+        raise unittest.SkipTest('FRAME_E2E=1 outside the e2e container: these tests use the real home, ~/.ssh '
+                                'and app cache, so they only run in scripts/e2e.sh\'s throwaway container')
+    raise unittest.SkipTest('needs the fake Frame: run scripts/e2e.sh (sets FRAME_E2E=1)')
 
 
 # ---- the fake Frame's control port --------------------------------------------
@@ -97,6 +118,7 @@ class Server:
 
     @classmethod
     def start(cls):
+        require()  # never a server on this computer's real data
         if cls.proc and cls.proc.poll() is None:
             return
         with socket.socket() as s:
@@ -193,6 +215,7 @@ class FrameTestCase(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
+        require()
         Server.start()
 
     def setUp(self):

tests/e2e/test_agents.py

@@ -6,6 +6,8 @@ import sys
 import harness
 from harness import api, ok, finished, ssh
 
+harness.require()
+
 sys.path.insert(0, str(harness.ROOT / 'ui'))
 import frame_mcp
 

tests/e2e/test_android.py

@@ -5,7 +5,7 @@ import unittest
 
 import harness
 from harness import HOME, exists, ok, state, upload, wait_for
-from test_frame_apk import apk, manifest, resources
+from apk_fixtures import apk, manifest, resources
 
 harness.require()
 

tests/fakeframe/compose.yaml

@@ -43,6 +43,7 @@ services:
       FAKEFRAME_CTL: http://fakeframe:9999
       FAKEFRAME_HOST: fakeframe
       FRAME_E2E: "1"
+      FRAME_E2E_CONTAINER: "1"  # tests/e2e/harness.py refuses to run anywhere else
     healthcheck:
       test: ["CMD", "test", "-f", "/home/tester/.ready"]
       interval: 1s

tests/fakessh/ssh

@@ -3,10 +3,20 @@
 each step of a connection, and plays a ControlMaster. What each HostName does comes
 from $FAKESSH_HOSTS (JSON: host -> "ok", "wrong" (a different host key), "denied" or
 "slow" (hangs after connecting);
-every call is appended to $FAKESSH_LOG as a JSON line. POSIX only."""
+every call is appended to $FAKESSH_LOG as a JSON line. POSIX only.
+
+With $FAKESSH_PINS set, host keys are checked as OpenSSH does against UserKnownHostsFile under
+HostKeyAlias: "wrong" is then just a different key from the others' (a reinstalled headset),
+StrictHostKeyChecking=yes refuses a key not saved, accept-new saves a first one, and a saved key
+that differs is refused. "evil" is a device with its own key that lets anyone in without
+authentication ("none") and asks for the agent: it records in $FAKESSH_DIR what it got
+(agent-forwarded, command-ran). Like ssh, the first value of an option counts."""
+import base64
+import hashlib
 import json
 import os
 import signal
+import subprocess
 import sys
 import time
 
@@ -17,12 +27,12 @@ hosts = json.loads(os.environ.get("FAKESSH_HOSTS", "{}"))
 opts = {}
 i = 0
 while i < len(args) and args[i].startswith("-"):
-    if args[i] in ("-o", "-O", "-p", "-l"):
+    if args[i] in ("-o", "-O", "-p", "-l", "-F", "-W"):
         key = args[i]
         val = args[i + 1]
         if key == "-o":
             k, _, v = val.partition("=")
-            opts[k.lower()] = v
+            opts.setdefault(k.lower(), v)
         else:
             opts[key] = val
         i += 2
@@ -30,6 +40,12 @@ while i < len(args) and args[i].startswith("-"):
         opts[args[i]] = True
         i += 1
 alias = args[i] if i < len(args) else ""
+command = args[i + 1:]
+if opts.get("-F") != "none":
+    # $FAKESSH_CONFIG: what ~/.ssh/config says (JSON, lower-case option -> value), below the
+    # command line's options, as ssh has it; `-F none` reads no config.
+    for k, v in json.loads(os.environ.get("FAKESSH_CONFIG", "{}")).items():
+        opts.setdefault(k, v)
 host = opts.get("hostname", alias).replace("%%", "%")
 marker = os.path.join(os.environ["FAKESSH_DIR"], "master-" + host.replace("/", "_"))
 say = lambda s: (sys.stderr.write(s + "\n"), sys.stderr.flush())
@@ -44,6 +60,23 @@ if opts.get("-O") == "exit":
         os.unlink(marker)
     sys.exit(0)
 
+if "-W" in opts:  # a jump host's ssh, forwarding to the headset: the call is logged, that's all
+    sys.exit(0)
+if opts.get("proxycommand", "none") != "none":
+    # As OpenSSH: expand %-tokens, then `$SHELL -c "exec <line>"` (here /bin/sh).
+    line, out, n = opts["proxycommand"], "", 0
+    while n < len(line):
+        if line[n] == "%" and n + 1 < len(line):
+            out += {"%": "%", "h": host, "p": opts.get("port", "22"), "r": opts.get("user", "")}.get(
+                line[n + 1], line[n:n + 2])
+            n += 2
+        else:
+            out += line[n]
+            n += 1
+    if subprocess.run(["/bin/sh", "-c", "exec " + out], stdin=subprocess.DEVNULL,
+                      stdout=subprocess.DEVNULL).returncode:
+        say("kex_exchange_identification: Connection closed by remote host")
+        sys.exit(255)
 what = hosts.get(host)
 if what is None:
     say(f"ssh: Could not resolve hostname {host}: nodename nor servname provided, or not known")
@@ -53,6 +86,66 @@ say("debug1: Connection established.")
 if what == "slow":
     time.sleep(30)
 say(f"debug1: Authenticating to {host}:22 as '{opts.get('user', 'tester')}'")
+if os.environ.get("FAKESSH_PINS") and "hostkeyalias" in opts and "userknownhostsfile" in opts:
+    name, path = opts["hostkeyalias"], os.path.expanduser(opts["userknownhostsfile"])
+    blob = base64.b64encode(b"fake-" + {"wrong": b"other", "evil": b"evil"}.get(what, b"headset") + b"-key").decode()
+    fp = base64.b64encode(hashlib.sha256(base64.b64decode(blob)).digest()).decode().rstrip("=")
+    say(f"debug1: Server host key: ssh-ed25519 SHA256:{fp}")
+    try:
+        saved = [ln.split() for ln in open(path, errors="replace").read().splitlines() if ln.strip()]
+    except OSError:
+        saved = []
+    saved = [f[2] for f in saved if len(f) >= 3 and name in f[0].split(",")]
+    strict = opts.get("stricthostkeychecking", "ask")
+    if saved and blob not in saved:
+        say("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@")
+        say("@    WARNING: REMOTE HOST IDENTIFICATION HAS CHANGED!     @")
+        say("Host key verification failed.")
+        sys.exit(255)
+    if not saved and strict != "accept-new":
+        say(f"No ED25519 host key is known for {name} and you have requested strict checking.")
+        say("Host key verification failed.")
+        sys.exit(255)
+    if not saved:
+        try:
+            with open(path, "a") as f:
+                f.write(f"{name} ssh-ed25519 {blob}\n")
+            say(f"Warning: Permanently added '{name}' (ED25519) to the list of known hosts.")
+        except OSError:  # ssh doesn't create the folder: it goes on, saving nothing
+            say(f"Failed to add the host to the list of known hosts ({path}).")
+    else:
+        say(f"debug1: Host '{name}' is known and matches the ED25519 host key.")
+    say("debug1: Next authentication method: publickey")
+    if what == "evil":
+        say(f'Authenticated to {host} ([127.0.0.1]:22) using "none".')
+        got = os.path.join(os.environ["FAKESSH_DIR"], "evil-")
+        if opts.get("forwardagent", "no") != "no" and opts.get("identityagent", "") != "none":
+            open(got + "agent-forwarded", "w").close()
+        if "-N" not in opts and (command or opts.get("remotecommand", "none") != "none"):
+            open(got + "command-ran", "w").close()
+            sys.exit(0)
+        time.sleep(30)  # -N: holds the session open
+        sys.exit(0)
+    if opts.get("preferredauthentications") == "none":
+        say(f"tester@{host}: Permission denied (publickey,password).")
+        sys.exit(255)
+    if what == "denied":
+        say(f"tester@{host}: Permission denied (publickey).")
+        sys.exit(255)
+    what = "ok"
+    say(f'Authenticated to {host} ([127.0.0.1]:22) using "publickey".')
+    if opts.get("controlmaster") != "yes":
+        sys.exit(0)
+    open(marker, "w").close()
+    def bye(*_):
+        if os.path.exists(marker):
+            os.unlink(marker)
+        sys.exit(0)
+    signal.signal(signal.SIGTERM, bye)
+    while True:
+        time.sleep(0.2)
+        if not os.path.exists(marker):
+            sys.exit(0)
 say("debug1: Server host key: ssh-ed25519 SHA256:fakefakefakefakefakefakefakefakefakefakefak")
 if what == "wrong":
     say("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@")

tests/sandbox.py

@@ -1,12 +1,20 @@
 """Imported first by every test module: nothing a test does reaches this person's
-app data, their telemetry, or the shared compatibility database.
+app data, their ~/.ssh, their telemetry, or the shared compatibility database.
 
 Must run before any ui module is imported, since those read these at import time.
+The variables are set in os.environ, so servers and scripts the tests start
+inherit them too (they pass {**os.environ, ...}).
 """
 import atexit
 import os
 import shutil
 import tempfile
+from pathlib import Path
+
+# The environment and real ~/.ssh from before the changes below: test_sandbox
+# reruns tests from that environment and checks ~/.ssh is left alone.
+ORIGINAL_ENV = dict(os.environ)
+REAL_SSH_DIR = Path.home() / ".ssh"
 
 _dir = tempfile.mkdtemp(prefix="frame-control-tests-")
 atexit.register(shutil.rmtree, _dir, ignore_errors=True)
@@ -14,3 +22,25 @@ os.environ["FRAME_CONTROL_DATA_DIR"] = _dir
 os.environ["FRAME_CONTROL_TELEMETRY"] = "0"
 # A maintainer's machine holds the database key; send anything that slips through nowhere.
 os.environ["FRAME_COMPAT_DB_URL"] = "http://127.0.0.1:9"
+
+# A throwaway home, so Path.home() / "~" land here: ~/.ssh (config, host-key pins,
+# its lock file) and the other per-user folders some modules use. A server started
+# without FRAME_CONTROL_SSH_DIR used to import the real ~/.ssh/config and seed a pin
+# into the real ~/.ssh/frame-control-hosts on every run.
+#
+# tests/e2e must not import this (directly or through a test_* module): they run in
+# a throwaway container whose real ~/.ssh points `ssh frame` at the fake Frame, and
+# OpenSSH finds ~/.ssh from the password database, not $HOME. Shared helpers they
+# need live in sandbox-free modules such as apk_fixtures.py.
+HOME = os.path.join(_dir, "home")
+SSH_DIR = os.path.join(HOME, ".ssh")
+os.makedirs(SSH_DIR, mode=0o700)
+os.environ["HOME"] = HOME
+os.environ["USERPROFILE"] = HOME  # Path.home() on Windows
+os.environ["FRAME_CONTROL_SSH_DIR"] = SSH_DIR  # frame_devices.ssh_dir(), frame_connect.SSH_DIR
+# The per-user folders that don't follow HOME when these are set: frame_host.cache_dir()
+# (servers warm indexes there and sweep its staging folders) and data_dir()'s fallbacks.
+for _name, _sub in (("LOCALAPPDATA", "AppData/Local"), ("APPDATA", "AppData/Roaming"),
+                    ("XDG_CACHE_HOME", ".cache"), ("XDG_DATA_HOME", ".local/share"),
+                    ("XDG_STATE_HOME", ".local/state"), ("XDG_CONFIG_HOME", ".config")):
+    os.environ[_name] = os.path.join(HOME, *_sub.split("/"))

tests/test_agent.py

@@ -1,4 +1,5 @@
 """MCP protocol, exact-action approvals and explicit assistant data sharing."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import io
 import json
 import os

tests/test_apk_artwork.py

@@ -1,3 +1,4 @@
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import http.client
 import json
 from pathlib import Path

tests/test_apk_more_sources.py

@@ -1,3 +1,4 @@
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import hashlib, io, json, os, sys, tempfile, time, unittest, urllib.error, urllib.request, zipfile
 from pathlib import Path
 from unittest.mock import patch

tests/test_apk_search.py

@@ -1,3 +1,4 @@
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import http.client
 import json
 from pathlib import Path

tests/test_comfort.py

@@ -1,4 +1,5 @@
 """Fake-Frame session clock/actions plus real helper serialization and sensor probes."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import os
 import shutil
 import json

tests/test_comfort_ui.py

@@ -1,4 +1,5 @@
 """Run the actual shared page's comfort renderer against a minimal DOM/bridge."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import pathlib
 import shutil
 import subprocess

tests/test_devices.py

@@ -264,6 +264,30 @@ class Pins(Base):
         self.assertFalse(fd.pinned("d1"))
         self.assertFalse(fd.forget_pin("d1"))
 
+    def test_a_trusted_key_replaces_the_saved_one_and_fingerprints_match_ssh_keygen(self):
+        key = self.dir / "hostkey"
+        frame_host.run_ssh(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(key)], capture_output=True,
+                           stdin=subprocess.DEVNULL, check=True, timeout=20)
+        kind, blob = key.with_suffix(".pub").read_text().split()[:2]
+        want = frame_host.run_ssh(["ssh-keygen", "-l", "-E", "sha256", "-f", str(key.with_suffix(".pub"))],
+                                  capture_output=True, text=True, stdin=subprocess.DEVNULL, timeout=10).stdout.split()[1]
+        self.assertEqual(fd.fingerprint(blob), want)
+        self.assertIsNone(fd.fingerprint("not base64!"))
+        (self.ssh / "known_hosts").write_text(f"frame.local {KEY}\n")
+        fd.seed_pin("d1", ["frame.local"])
+        old = fd.pinned_keys("d1")
+        fd.pin_key("d1", kind, blob)
+        self.assertEqual(fd.known_hosts("d1").read_text(), f"frame-control-d1 {kind} {blob}\n")  # old key gone
+        self.assertEqual(fd.pinned_keys("d1"), [(kind, want)])
+        self.assertNotEqual(old, fd.pinned_keys("d1"))
+        self.assertTrue(fd.pinned("d1"))
+        for bad in (("ssh-ed25519\nHost *", blob), ("ssh-ed25519", "AAAA AAAA"), ("evil", blob)):
+            with self.assertRaises(fd.DeviceError):
+                fd.pin_key("d1", *bad)
+        # Only plain entries for this headset's alias count.
+        self.assertIsNone(fd.key_entry(f"frame-control-d2 {kind} {blob}", "frame-control-d1"))
+        self.assertIsNone(fd.key_entry(f"# frame-control-d1 {kind} {blob}", "frame-control-d1"))
+
     def test_each_headset_has_its_own_file(self):
         (self.ssh / "known_hosts").write_text(f"a.local {KEY}\nb.local {KEY}\n")
         fd.seed_pin("da", ["a.local"])

tests/test_fdroid_sources.py

@@ -1,4 +1,5 @@
 """Offline authenticated repository fixtures; no tests contact a server."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import io
 import json
 import os

tests/test_frame_android_data.py

@@ -1,3 +1,4 @@
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import io
 import json
 import os

tests/test_frame_android_library.py

@@ -1,4 +1,5 @@
 """Offline artwork, Steam API and launcher supervision regressions."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import importlib.util
 import json
 import os
@@ -263,11 +264,14 @@ class ArtworkTests(unittest.TestCase):
     def test_steamgriddb_uses_the_bounded_fetch_without_redirects(self):
         import frame_steamgriddb as sgdb
         from apk_sources import _images
+        deadline = time.monotonic() + 5
         with patch.object(_images, 'get', return_value=b'{"success": true, "data": [1]}') as get:
-            self.assertEqual(sgdb._get('/search/x', 'secret', time.monotonic() + 5), [1])
+            self.assertEqual(sgdb._get('/search/x', 'secret', deadline), [1])
         self.assertEqual(get.call_args.kwargs['redirects'], 0)
         self.assertEqual(get.call_args.args[1]['Authorization'], 'Bearer secret')
-        self.assertLessEqual(get.call_args.kwargs['deadline'] - time.monotonic(), 5)
+        # The caller's deadline is passed on, never extended. (Comparing against a fresh
+        # time.monotonic() flaked on Windows, whose clock ticks coarsely.)
+        self.assertLessEqual(get.call_args.kwargs['deadline'], deadline)
 
     def test_supplied_jpeg(self):
         data = (FIXTURES / 'icon.jpg').read_bytes()

tests/test_frame_apk.py

@@ -11,82 +11,7 @@ import zipfile
 
 sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'ui'))
 import frame_apk  # noqa: E402
-
-
-def pool(strings, utf8=False):
-    """A ResStringPool chunk."""
-    data, offsets = b'', []
-    for s in strings:
-        offsets.append(len(data))
-        if utf8:
-            b = s.encode()
-            data += bytes([len(s), len(b)]) + b + b'\0'
-        else:
-            data += struct.pack('<H', len(s)) + s.encode('utf-16-le') + b'\0\0'
-    data += b'\0' * (-len(data) % 4)
-    start = 28 + 4 * len(strings)
-    body = struct.pack(f'<{len(strings)}I', *offsets) + data
-    return struct.pack('<HHIIIIII', 1, 28, 28 + len(body), len(strings), 0, 0x100 if utf8 else 0, start, 0) + body
-
-
-def manifest(package, label_ref, version_ref, min_sdk, package_raw=True, foreign_label=False):
-    """<manifest package versionName><uses-sdk minSdkVersion/><application label icon/></manifest>.
-
-    package_raw=False drops the package's raw string (as some repackers do);
-    foreign_label adds a non-android `label` attribute after android:label.
-    """
-    strings = ['label', 'icon', 'versionName', 'minSdkVersion', 'package', 'manifest', 'uses-sdk',
-               'application', package, 'junk', 'label', 'versionCode']  # the second 'label' has no android id
-    resmap = struct.pack('<4I', 0x01010001, 0x01010002, 0x0101021c, 0x0101020c)
-    resmap = struct.pack('<HHI', 0x0180, 8, 8 + len(resmap)) + resmap
-
-    def element(name, attrs):
-        body = struct.pack('<IIHHHHHH', 0xffffffff, name, 20, 20, len(attrs), 0, 0, 0)
-        for aname, raw, dtype, value in attrs:
-            body += struct.pack('<IIIHBBI', 0xffffffff, aname, raw, 8, 0, dtype, value)
-        return struct.pack('<HHIII', 0x0102, 16, 16 + len(body), 1, 0xffffffff) + body
-
-    none = 0xffffffff
-    chunks = (pool(strings) + resmap
-              + element(5, [(4, 8 if package_raw else none, frame_apk.T_STRING, 8),
-                            (2, none, frame_apk.T_REF, version_ref),
-                            (11, none, frame_apk.T_INT_DEC, 210)])
-              + element(6, [(3, none, frame_apk.T_INT_DEC, min_sdk)])
-              + element(7, [(0, none, frame_apk.T_REF, label_ref), (1, none, frame_apk.T_REF, 0x7f020000)]
-                        + ([(10, 9, frame_apk.T_STRING, 9)] if foreign_label else [])))
-    return struct.pack('<HHI', 3, 8, 8 + len(chunks)) + chunks
-
-
-def resources(values):
-    """resources.arsc with package 0x7f; values: {(type id, entry, language, density): global string index}."""
-    strings = ['French label', 'App label', '2.1', 'res/icon_lo.png', 'res/icon_hi.png', 'res/icon.xml']
-    pkg_body = b''
-    for (tid, lang, density), entries in values.items():
-        cfg = struct.pack('<I4x2s4xH', 64, lang.encode().ljust(2, b'\0'), density).ljust(64, b'\0')
-        count = max(entries) + 1
-        offsets, data = [], b''
-        for i in range(count):
-            if i in entries:
-                offsets.append(len(data))
-                data += struct.pack('<HHI', 8, 0, 0) + struct.pack('<HBBI', 8, 0, frame_apk.T_STRING, entries[i])
-            else:
-                offsets.append(0xffffffff)
-        header = 20 + 64
-        estart = header + 4 * count
-        body = struct.pack(f'<{count}I', *offsets) + data
-        pkg_body += struct.pack('<HHIBBHII', 0x0201, header, header + len(body), tid, 0, 0, count, estart) + cfg + body
-    pkg_header = struct.pack('<HHII', 0x0200, 288, 288 + len(pkg_body), 0x7f).ljust(288, b'\0')
-    pkg = pkg_header + pkg_body
-    table = pool(strings, utf8=True) + pkg
-    return struct.pack('<HHII', 2, 12, 12 + len(table), 1) + table
-
-
-def apk(files):
-    buf = io.BytesIO()
-    with zipfile.ZipFile(buf, 'w') as z:
-        for name, data in files.items():
-            z.writestr(name, data)
-    return buf.getvalue()
+from apk_fixtures import apk, manifest, pool, resources  # noqa: E402,F401
 
 
 class ApkInfo(unittest.TestCase):

tests/test_frame_apk_vr.py

@@ -1,4 +1,5 @@
 """Local-only VR manifest, raw ZIP and independent signature checks."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import io
 import os
 from pathlib import Path
@@ -15,7 +16,7 @@ import frame_apk
 import frame_apk_vr as vr
 import frame_apk_sign as signing
 import frame_android
-from test_frame_apk import pool
+from apk_fixtures import pool
 
 
 DEFAULT = 'android.intent.category.DEFAULT'

tests/test_input.py

@@ -6,6 +6,7 @@ identity line, then starts TLS as the *client*.
 
 Run: python3 -m unittest discover -s tests
 """
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import json
 import os
 import shutil

tests/test_library_entrypoints.py

@@ -1,4 +1,5 @@
 """Every public sideload entry point reaches the mandatory five-slot writer."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import contextlib
 import io
 import json

tests/test_link.py

@@ -9,6 +9,7 @@ import http.client
 import io
 import json
 import os
+import shlex
 import shutil
 import socket
 import subprocess
@@ -199,7 +200,7 @@ class Connecting(unittest.TestCase):
         rows = {p["host"]: p for p in s["probes"]}
         self.assertEqual(rows["nothing.invalid"]["state"], "unresolved")
         self.assertEqual(rows["::1"]["state"], "sshfailed")
-        self.assertIn("different headset", rows["::1"]["detail"])
+        self.assertIn("different SSH identity", rows["::1"]["detail"])
         # Every ssh command was pointed at the winner, with the host key pinned per device.
         alias, opts = self.routes[-1]
         self.assertEqual(alias, "frame-t")
@@ -208,9 +209,10 @@ class Connecting(unittest.TestCase):
         self.assertIn(f"Port={self.port}", opts)
         master = [c for c in self.calls() if "ControlMaster=yes" in c][-1]
         self.assertIn("StrictHostKeyChecking=accept-new", master)  # first connection: nothing pinned yet
-        # ssh takes an option's first value: accept-new must come before the commands' own "yes".
-        self.assertLess(master.index("StrictHostKeyChecking=accept-new"), master.index("StrictHostKeyChecking=yes"))
-        self.assertIn("StrictHostKeyChecking=yes", opts)  # every other command checks the pinned key
+        self.assertNotIn("StrictHostKeyChecking=yes", master)
+        # This stand-in ssh saved no key, so commands may still save the first one (accept-new);
+        # test_a_first_connection_pins_and_then_checks_strictly has real pins.
+        self.assertIn("StrictHostKeyChecking=accept-new", opts)
         self.assertTrue(fd.known_hosts(d["id"]).parent.is_dir())  # where ssh saves the key it accepts
         # It learned: 127.0.0.1 works on this network.
         learned = {a["host"]: a for a in self.reg.get(d["id"])["addresses"]}
@@ -257,6 +259,7 @@ class Connecting(unittest.TestCase):
         s = self.link.snapshot()
         self.assertEqual((s["phase"], s["error"]["stage"]), ("failed", "login"))
         self.assertIn("SSH key", s["error"]["message"])
+        self.assertEqual(s["error"]["auth"], "denied")  # the page offers Set Up Connection
 
     def test_pinned_identity_is_checked_strictly(self):
         d = self.device("localhost")
@@ -266,6 +269,426 @@ class Connecting(unittest.TestCase):
         master = [c for c in self.calls() if "ControlMaster=yes" in c][-1]
         self.assertIn("StrictHostKeyChecking=yes", master)
 
+    def pins(self):
+        """The stand-in ssh checks host keys as OpenSSH does (tests/fakessh/ssh)."""
+        p = mock.patch.dict(os.environ, {"FAKESSH_PINS": "1"})
+        p.start()
+        self.addCleanup(p.stop)
+
+    def command(self, device, host="127.0.0.1"):
+        """One ssh command as the server runs it, with the route the connector applied."""
+        return subprocess.run(["ssh", *self.link.host_opts(device, host), device["alias"], "true"],
+                              capture_output=True, text=True, timeout=20)
+
+    def test_a_first_connection_pins_and_then_checks_strictly(self):
+        # 0.4.2 on Windows: "No ED25519 host key is known for frame-control-<id> and you have
+        # requested strict checking." Every command said "yes" before anything was saved, so one
+        # that ran before the connector's handshake saved the key (or without it) failed.
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "ok"})
+        r = self.command(d)
+        self.assertEqual(r.returncode, 0, r.stderr)  # the first contact saves the key...
+        self.assertIn("Permanently added", r.stderr)
+        self.assertTrue(fd.pinned(d["id"]))
+        self.assertIn("StrictHostKeyChecking=yes", self.link.host_opts(d, "127.0.0.1"))  # ...then it's checked
+        self.assertEqual(self.reg.get(d["id"])["identity"], [k[1] for k in fd.pinned_keys(d["id"])])  # and recorded
+        # The connector's first handshake, for another new headset.
+        self.reg.remove_device(d["id"])
+        d = self.device("localhost")
+        self.link.connect(["start"])
+        s = self.link.snapshot()
+        self.assertEqual(s["phase"], "connected", s["error"])
+        self.assertTrue(fd.pinned(d["id"]))
+        self.assertTrue(self.reg.get(d["id"])["identity"])
+        alias, opts = self.routes[-1]
+        self.assertIn("StrictHostKeyChecking=yes", opts)  # every command after it checks strictly
+        self.assertNotIn("StrictHostKeyChecking=accept-new", opts)
+
+    def lose_pin(self, device_id, how):
+        f = fd.known_hosts(device_id)
+        if how == "deleted":
+            f.unlink()
+        elif how == "empty":
+            f.write_text("")
+        elif how == "unreadable":
+            f.chmod(0)
+            self.addCleanup(lambda: f.exists() and f.chmod(0o600))
+        elif how == "another alias":
+            f.write_text(f.read_text().replace(f"frame-control-{device_id}", "frame-control-someone-else"))
+        elif how == "not utf-8":
+            f.write_bytes(b"\xff\xfe" + f.read_bytes())
+
+    def test_a_lost_or_unusable_pin_fails_closed(self):
+        # Review pr85-review-r1: accept-new can't tell a new headset from one whose saved key
+        # went missing; real OpenSSH then trusted a replacement key. Once a headset had an
+        # identity, nothing is trusted again until the person checks it.
+        self.pins()
+        cases = ["deleted", "empty", "another alias", "not utf-8"]
+        if os.name == "posix" and os.geteuid() != 0:
+            cases.append("unreadable")  # root reads anything
+        for n, how in enumerate(cases):
+            with self.subTest(how=how):
+                self.hosts({"localhost": "ok"})
+                d = self.reg.add_device(f"frame-l{n}", port=self.port, hosts=[])
+                self.reg.add_address(d["id"], "localhost", kind="lan")
+                self.link.override = None
+                self.reg.set_active(d["id"])
+                self.link.connect(["switch"])
+                self.assertEqual(self.link.snapshot()["phase"], "connected")
+                self.link.close_master()
+                d = self.reg.get(d["id"])
+                self.lose_pin(d["id"], how)
+                self.hosts({"localhost": "wrong"})  # another device, or a reinstalled headset
+                r = self.command(d)
+                self.assertEqual(r.returncode, 255, r.stderr)
+                self.assertIn("Host key verification failed", r.stderr)
+                self.assertNotIn("Permanently added", r.stderr)
+                self.link.connect(["dropped"])
+                s = self.link.snapshot()
+                self.assertEqual((s["phase"], s["error"]["stage"]), ("failed", "identity"))
+                self.assertEqual(s["error"]["hostkey"]["state"], "missing")
+                self.assertIn("missing or can't be read", s["error"]["message"])
+                view = next(x for x in fl.devices_view(self.link)["devices"] if x["id"] == d["id"])
+                self.assertTrue(view["identity_lost"])
+                scan = fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)["scan"]
+                self.assertFalse(scan["matches"] or scan["recorded_match"])  # not the key it had
+                self.hosts({"localhost": "ok"})
+                scan = fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)["scan"]
+                self.assertTrue(scan["recorded_match"])  # its own key, as recorded
+                self.assertFalse(scan["matches"])  # but not usable from the file: trusting restores it
+                fl.devices_action(self.link, {"action": "identity-trust", "id": d["id"],
+                                              "fingerprint": scan["fingerprint"]}, None)
+                self.link.connect(["switch"])
+                self.assertEqual(self.link.snapshot()["phase"], "connected", self.link.snapshot()["error"])
+                self.link.close_master()
+
+    def test_a_key_check_only_reads_the_key(self):
+        # Review pr85-review-r1: with ForwardAgent yes in ~/.ssh/config, a device that accepts
+        # "none" got the agent and ran a command before anyone trusted it.
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "evil"})
+        os.environ["FAKESSH_CONFIG"] = json.dumps({"forwardagent": "yes", "remotecommand": "touch /tmp/x",
+                                                   "permitlocalcommand": "yes", "requesttty": "force"})
+        self.addCleanup(os.environ.pop, "FAKESSH_CONFIG", None)
+        for proxied in (False, True):
+            route = ("jump", "me@jumpy:2222") if proxied else (None, None)
+            with self.subTest(proxied=proxied), mock.patch.object(fl, "proxy_route", return_value=route):
+                t0 = time.monotonic()
+                scan = fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)["scan"]
+                self.assertLess(time.monotonic() - t0, 10)  # stopped once it had the key, not left open
+                self.assertTrue(scan["fingerprint"].startswith("SHA256:"))
+                self.assertEqual([p.name for p in self.dir.iterdir() if p.name.startswith("evil-")], [])
+                call = [c for c in self.calls() if "PreferredAuthentications=none" in c][-1]
+                self.assertEqual(call[call.index("-F") + 1], "none")  # never ~/.ssh/config for the device
+                self.assertIn("-N", call)
+                hop = [c for c in call if c.startswith("ProxyCommand=")]
+                if proxied:
+                    # Review pr85-review-r2: ProxyJump's own ssh inherits the config (ForwardAgent,
+                    # LocalCommand); the scan builds the jump host's ssh itself, hardened.
+                    self.assertEqual(len(hop), 1)
+                    words = shlex.split(hop[0][len("ProxyCommand="):])
+                    self.assertEqual(words[-7:], ["-l", "me", "-p", "2222", "-W", "[%h]:%p", "jumpy"])
+                    for o in fl.Link.HOP_OPTS:
+                        self.assertIn(o, words)
+                    self.assertLess(call.index(hop[0]), call.index("HostKeyAlias=frame-control-" + d["id"]))
+                else:
+                    self.assertEqual(hop, [])
+                for o in ("ForwardAgent=no", "IdentityAgent=none", "ClearAllForwardings=yes", "RemoteCommand=none",
+                          "PermitLocalCommand=no", "RequestTTY=no", "PubkeyAuthentication=no"):
+                    self.assertIn(o, call)
+                self.assertLess(call.index("StrictHostKeyChecking=accept-new"), call.index("HostKeyAlias=frame-control-"
+                                                                                          + d["id"]))
+        self.assertFalse(fd.pinned(d["id"]))
+        self.assertEqual(sorted(p.name for p in fd.known_hosts(d["id"]).parent.iterdir()), [])  # no scan files
+
+    @unittest.skipIf(os.name != "posix", "the stand-in ssh runs ProxyCommand with /bin/sh")
+    def test_the_jump_hosts_command_line_is_quoted_whatever_the_names(self):
+        # Review pr85-review-r3: the hop's words went into the ProxyCommand line unquoted, so an
+        # ssh at /tmp/x/bin$(:>INJECTED)/ssh ran shell code; and an IPv6 jump host broke.
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "ok"})
+        for n, name in enumerate(["with space", "bin$(touch INJECTED-0)", "back`touch INJECTED-1`tick",
+                                  "quote'it\"s", "per%cent %h %%", "semi;touch INJECTED-2"]):
+            with self.subTest(name=name):
+                bindir = self.dir / f"b{n}" / name
+                bindir.mkdir(parents=True)
+                (bindir / "ssh").symlink_to(FAKESSH / "ssh")
+                for spec, at in (("me@jumpy:2222", "jumpy"), ("[::1]:2200", "::1")):
+                    before = len(self.calls())
+                    with mock.patch.dict(os.environ, {"PATH": f"{bindir}{os.pathsep}{os.environ['PATH']}"}), \
+                            mock.patch.object(fl, "proxy_route", return_value=("jump", spec)):
+                        scan = fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)["scan"]
+                    self.assertTrue(scan["fingerprint"].startswith("SHA256:"))
+                    hops = [c for c in self.calls()[before:] if "-W" in c]
+                    self.assertEqual(len(hops), 1, self.calls()[before:])
+                    self.assertEqual(hops[0][-1], at)  # the jump host, bare
+                    self.assertEqual(hops[0][hops[0].index("-W") + 1], "[localhost]:%d" % self.port)
+                    outer = [c for c in self.calls()[before:] if "PreferredAuthentications=none" in c][-1]
+                    line = next(c for c in outer if c.startswith("ProxyCommand="))
+                    self.assertTrue(line[len("ProxyCommand="):].startswith(shlex.quote(str(bindir / "ssh")
+                                                                                       .replace("%", "%%"))))
+                self.assertEqual([p.name for p in self.dir.rglob("INJECTED*")], [])
+                self.assertFalse(list(Path.cwd().glob("INJECTED*")))
+
+    def test_a_key_check_refuses_routes_it_cant_harden(self):
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "ok"})
+        before = len(self.calls())
+        for route in (("command", "nc %h %p"), ("jump", "a,b"), ("jump", "-oProxyCommand=x"), ("jump", "ssh://a"),
+                      ("jump", "a b"), ("jump", "a%h"), ("jump", "jumpy:99999")):
+            with self.subTest(route=route), mock.patch.object(fl, "proxy_route", return_value=route):
+                with self.assertRaises(fd.DeviceError) as caught:
+                    fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)
+                self.assertIn("can't make safe", str(caught.exception))
+        self.assertFalse(any("PreferredAuthentications=none" in c for c in self.calls()[before:]))
+
+    def test_a_key_read_before_the_address_or_port_changed_isnt_trusted(self):
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "ok"})
+        for change in ({"action": "update", "id": d["id"], "port": self.port + 1},
+                       {"action": "address-remove", "id": d["id"], "host": "localhost"}):
+            with self.subTest(change=change["action"]):
+                scan = fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)["scan"]
+                fl.devices_action(self.link, change, None)
+                with self.assertRaises(fd.DeviceError):
+                    fl.devices_action(self.link, {"action": "identity-trust", "id": d["id"],
+                                                  "fingerprint": scan["fingerprint"]}, None)
+                self.assertFalse(fd.pinned(d["id"]))
+                self.reg.update_device(d["id"], port=self.port)
+                if not self.reg.get(d["id"])["addresses"]:
+                    self.reg.add_address(d["id"], "localhost", kind="lan")
+
+    def test_terminals_judge_identity_by_the_registry_not_a_frozen_route(self):
+        # Review pr85-review-r2: named_route used the device as routed (before its first key was
+        # recorded), so a Terminal window fell back to accept-new once the pin was deleted.
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "ok"})
+        self.link.connect(["start"])
+        self.assertEqual(self.link.snapshot()["phase"], "connected")
+        self.assertFalse(self.link.routed_device.get("identity"))  # the frozen copy knows nothing of it
+        fd.known_hosts(d["id"]).unlink()
+        alias, opts = self.link.named_route()
+        self.assertIn("StrictHostKeyChecking=yes", opts)
+        self.assertNotIn("StrictHostKeyChecking=accept-new", opts)
+        self.hosts({"localhost": "wrong"})
+        r = subprocess.run(["ssh", *opts, alias, "true"], capture_output=True, text=True, timeout=20)
+        self.assertEqual(r.returncode, 255, r.stderr)
+        self.assertFalse(fd.pinned(d["id"]))
+
+    def test_a_key_saved_before_a_refused_login_is_kept_and_checked(self):
+        # Review pr85-review-r2: ssh saves the key during the key exchange; if the login is then
+        # refused, it was neither recorded nor protected, and the route stayed accept-new.
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "denied"})
+        self.link.connect(["start"])
+        s = self.link.snapshot()
+        self.assertEqual((s["phase"], s["error"]["stage"]), ("failed", "login"))
+        got = self.reg.get(d["id"])
+        self.assertEqual(got["identity"], [f for _, f in fd.pinned_keys(d["id"])])
+        self.assertTrue(got["identity_established"])
+        if os.name == "posix":
+            self.assertEqual(fd.known_hosts(d["id"]).stat().st_mode & 0o777, 0o600)
+        self.assertIn("StrictHostKeyChecking=yes", self.routes[-1][1])
+        fd.known_hosts(d["id"]).unlink()
+        self.hosts({"localhost": "wrong"})
+        r = self.command(self.reg.get(d["id"]))
+        self.assertEqual(r.returncode, 255)
+        self.assertNotIn("Permanently added", r.stderr)
+
+    def test_a_key_a_command_saved_tightens_the_route(self):
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "denied"})
+        with self.link.route_lock:
+            self.link.routed_device = d
+            self.link.apply(d["alias"], self.link.host_opts(d, "127.0.0.1"))
+        opts = self.routes[-1][1]
+        self.assertIn("StrictHostKeyChecking=accept-new", opts)
+        r = subprocess.run(["ssh", *opts, d["alias"], "true"], capture_output=True, text=True, timeout=20)
+        self.assertIn("Permanently added", r.stderr)
+        self.assertEqual(r.returncode, 255)  # saved the key, then refused the login
+        self.link.settle_pin()  # what the connector's loop does every tick while accept-new is applied
+        self.assertIn("StrictHostKeyChecking=yes", self.routes[-1][1])
+        self.assertNotIn("StrictHostKeyChecking=accept-new", self.routes[-1][1])
+        self.assertTrue(self.reg.get(d["id"])["identity"])
+
+    def test_editing_the_address_that_worked_keeps_the_identity_history(self):
+        # Review pr85-review-r2: last_ok lives on the address, so editing or removing it made a
+        # headset whose pin was gone (and had no fingerprint recorded) new again.
+        path = self.dir / "legacy.json"
+        path.write_text(json.dumps({"version": 1, "active": "abcd1234", "networks": {}, "devices": [
+            {"id": "abcd1234", "name": "Old", "alias": "frame-old", "user": "steamos", "port": 22,
+             "identity_files": [], "addresses": [{"host": "10.0.0.5", "kind": "lan", "label": "",
+                                                  "networks": [], "last_ok": 1790000000.0, "last_rtt_ms": 3}]}]}))
+        reg = fd.Registry(path)
+        self.assertTrue(reg.get("abcd1234")["identity_established"])  # migrated from last_ok
+        reg.update_address("abcd1234", "10.0.0.5", new_host="10.0.0.6")
+        reg.remove_address("abcd1234", "10.0.0.6")
+        reg.add_address("abcd1234", "10.0.0.7")
+        d = fd.Registry(path).get("abcd1234")  # saved, too
+        self.assertFalse(any(a.get("last_ok") for a in d["addresses"]))
+        self.assertTrue(fd.identity_established(d))
+        self.assertEqual(self.link.strictness(d), "yes")
+        # A connection that works sets it as well, for a headset added since.
+        n = self.reg.add_device("frame-new", hosts=["10.0.0.8"])
+        self.reg.record_success(n["id"], "10.0.0.8", None, 1)
+        self.reg.remove_address(n["id"], "10.0.0.8")
+        self.assertTrue(self.reg.get(n["id"])["identity_established"])
+
+    def test_ssh_last_words_arrive_even_when_the_reader_lags(self):
+        # Review pr85-review-r2: ssh's exit with nothing queued yet was taken as the end, and a
+        # late reader lost "Permission denied" ("ssh stopped").
+        self.pins()
+        link = fl.Link(self.reg, env_alias=None, mux_base=["ssh", "-o", "BatchMode=yes"], control=None,
+                       apply=lambda *a: None, explain=explain)
+        self.addCleanup(link.stop)
+        real = threading.Thread
+
+        class Late(real):
+            def __init__(self, *a, target=None, **kw):
+                if target is not None and getattr(target, "__name__", "") == "tail":
+                    inner = target
+
+                    def target():
+                        time.sleep(1)
+                        inner()
+                super().__init__(*a, target=target, **kw)
+        self.device("localhost")
+        self.hosts({"localhost": "denied"})
+        with mock.patch.object(fl.threading, "Thread", Late):
+            link.connect(["start"])
+        s = link.snapshot()
+        self.assertEqual((s["error"]["stage"], s["error"].get("auth")), ("login", "denied"), s["error"])
+        self.assertIn("SSH key", s["error"]["message"])
+
+    def test_forgetting_an_identity_is_refused(self):
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "ok"})
+        self.link.connect(["start"])
+        saved = fd.known_hosts(d["id"]).read_text()
+        with self.assertRaises(fd.DeviceError) as caught:
+            fl.devices_action(self.link, {"action": "forget-identity", "id": d["id"]}, None)
+        self.assertIn("identity-scan", str(caught.exception))
+        self.assertEqual(fd.known_hosts(d["id"]).read_text(), saved)
+
+    def test_without_a_shared_connection_ssh_writes_to_a_file_not_a_pipe(self):
+        # Windows: no ControlMaster, and its OpenSSH can block writing to a stderr pipe.
+        self.pins()
+        tmp = self.dir / "tmp"
+        tmp.mkdir()
+        link = fl.Link(self.reg, env_alias=None, mux_base=["ssh", "-o", "BatchMode=yes"], control=None,
+                       apply=lambda *a: None, explain=explain)
+        self.addCleanup(link.stop)
+        real = subprocess.Popen
+        seen = []
+
+        def popen(argv, **kw):
+            if "-v" in argv:  # the handshake (run_ssh calls go through Popen too)
+                seen.append(kw.get("stderr"))
+            return real(argv, **kw)
+        with mock.patch.object(tempfile, "tempdir", str(tmp)), mock.patch.object(fl.subprocess, "Popen", popen):
+            d = self.device("localhost")
+            self.hosts({"localhost": "ok"})
+            link.connect(["start"])
+            s = link.snapshot()
+            self.assertEqual(s["phase"], "connected", s["error"])
+            self.assertEqual([st["state"] for st in s["stages"]], ["done"] * 5)
+            self.assertIn("saved this headset's identity", next(st for st in s["stages"] if st["id"] == "identity")["detail"])
+            self.reg.update_device(d["id"], user="nobody")
+            self.hosts({"localhost": "denied"})
+            link.connect(["switch"])
+            s = link.snapshot()
+            self.assertEqual((s["phase"], s["error"]["stage"], s["error"].get("auth")), ("failed", "login", "denied"))
+        self.assertTrue(seen and all(isinstance(x, int) for x in seen))  # a file descriptor, never PIPE
+        self.assertNotIn(subprocess.PIPE, seen)
+        for _ in range(50):  # the reader removes its file once ssh has exited
+            if not list(tmp.iterdir()):
+                break
+            time.sleep(0.05)
+        self.assertEqual(list(tmp.iterdir()), [])
+
+    def test_a_changed_key_is_shown_never_accepted_and_trusted_on_request(self):
+        # 0.4.2 on Linux and Windows: "WARNING: REMOTE HOST IDENTIFICATION HAS CHANGED!", with
+        # "Run Set Up Connection again" as the advice, which can't replace the saved key.
+        self.pins()
+        d = self.device("localhost")
+        self.hosts({"localhost": "ok"})
+        self.link.connect(["start"])
+        self.assertEqual(self.link.snapshot()["phase"], "connected")
+        saved = fd.known_hosts(d["id"]).read_text()
+        old = fd.pinned_keys(d["id"])
+        self.link.close_master()
+        self.hosts({"localhost": "wrong"})  # SteamOS reinstalled: a new host key
+        self.link.connect(["dropped"])
+        s = self.link.snapshot()
+        self.assertEqual((s["phase"], s["error"]["stage"]), ("failed", "identity"))
+        shown = s["error"]["hostkey"]
+        self.assertEqual((shown["state"], shown["type"], shown["host"]), ("changed", "ssh-ed25519", "localhost"))
+        self.assertTrue(shown["fingerprint"].startswith("SHA256:"))
+        self.assertIn("Devices tab", s["error"]["message"])
+        self.assertEqual(fd.known_hosts(d["id"]).read_text(), saved)  # never accepted on its own
+        r = self.command(d)
+        self.assertEqual(r.returncode, 255)
+        self.assertIn("REMOTE HOST IDENTIFICATION HAS CHANGED", r.stderr)  # nor by any command
+        # Trusting needs the key read first, and exactly the fingerprint shown.
+        with self.assertRaises(fd.DeviceError):
+            fl.devices_action(self.link, {"action": "identity-trust", "id": d["id"],
+                                          "fingerprint": shown["fingerprint"]}, None)
+        out = fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)
+        scan = out["scan"]
+        self.assertEqual(scan["fingerprint"], shown["fingerprint"])
+        self.assertFalse(scan["matches"])
+        self.assertEqual(scan["saved"], [{"type": t, "fingerprint": f} for t, f in old])
+        self.assertEqual(fd.known_hosts(d["id"]).read_text(), saved)  # reading it saves nothing
+        self.assertEqual(sorted(p.name for p in fd.known_hosts(d["id"]).parent.iterdir()), [d["id"]])  # no scan files
+        call = [c for c in self.calls() if "PreferredAuthentications=none" in c][-1]  # no login offered
+        self.assertLess(call.index("StrictHostKeyChecking=accept-new"), call.index("StrictHostKeyChecking=yes"))
+        self.assertNotIn(f"UserKnownHostsFile={fd.known_hosts_opt(d['id'])}",
+                         call[:call.index("StrictHostKeyChecking=yes")])  # a throwaway file, first
+        with self.assertRaises(fd.DeviceError):
+            fl.devices_action(self.link, {"action": "identity-trust", "id": d["id"], "fingerprint": "SHA256:other"}, None)
+        self.link.scans[d["id"]]["at"] -= fl.SCAN_TTL + 1
+        with self.assertRaises(fd.DeviceError):  # shown too long ago
+            fl.devices_action(self.link, {"action": "identity-trust", "id": d["id"],
+                                          "fingerprint": shown["fingerprint"]}, None)
+        fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)
+        with mock.patch.object(self.link, "kick") as kick:
+            out = fl.devices_action(self.link, {"action": "identity-trust", "id": d["id"],
+                                                "fingerprint": shown["fingerprint"]}, None)
+        kick.assert_called_with("switch")
+        self.assertEqual(fd.pinned_keys(d["id"]), [("ssh-ed25519", shown["fingerprint"])])  # the old key is gone
+        self.assertEqual(next(x for x in out["devices"] if x["id"] == d["id"])["keys"],
+                         [{"type": "ssh-ed25519", "fingerprint": shown["fingerprint"]}])
+        self.link.connect(["switch"])
+        self.assertEqual(self.link.snapshot()["phase"], "connected", self.link.snapshot()["error"])
+        out = fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)
+        self.assertTrue(out["scan"]["matches"])
+
+    def test_test_now_never_saves_a_key(self):
+        self.pins()
+        d = self.device("127.0.0.1")
+        self.hosts({"127.0.0.1": "ok"})
+        self.link.test(d["id"])
+        rows = self.link.snapshot()["tests"][d["id"]]["rows"]
+        self.assertEqual(rows[0]["ssh"], "unpinned")
+        self.assertFalse(fd.pinned(d["id"]))
+
+    def test_reading_a_key_from_a_headset_that_doesnt_answer_says_why(self):
+        self.pins()
+        d = self.device("nothing.invalid")
+        with self.assertRaises(fd.DeviceError) as caught:
+            fl.devices_action(self.link, {"action": "identity-scan", "id": d["id"]}, None)
+        self.assertIn("Can't find the Frame", str(caught.exception))
+        self.assertEqual([p.name for p in fd.known_hosts(d["id"]).parent.iterdir()], [])
+
     def test_ssh_goes_to_the_ipv4_address_that_answered(self):
         self.device("localhost")
         self.hosts({"localhost": "ok"})
@@ -501,7 +924,7 @@ class Connecting(unittest.TestCase):
                      {"action": "update", "id": d["id"], "port": 2222},
                      {"action": "address-remove", "id": d["id"], "host": "localhost"},
                      {"action": "address-update", "id": d["id"], "host": "localhost", "newHost": "127.0.0.1"},
-                     {"action": "forget-identity", "id": d["id"]}):
+                     {"action": "identity-trust", "id": d["id"], "fingerprint": "SHA256:x"}):
             with self.assertRaises(fd.DeviceError, msg=body):
                 fl.devices_action(self.link, body, open_setup=None, busy=lambda: 1)
         # Renaming, or changing another headset, is fine.

tests/test_macview.py

@@ -7,6 +7,7 @@ input reaching the source, and closing.
 
 Run: python3 -m unittest discover -s tests
 """
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import base64
 import http.client
 import json

tests/test_media.py

@@ -1,4 +1,5 @@
 """Owned media planning, eye isolation, decoder choice and fake-Frame ownership."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import argparse
 import io
 import json

tests/test_panels.py

@@ -1,4 +1,5 @@
 """Fake-Frame panel responses and the headset page's real HTTP guards."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import http.client
 import json
 from pathlib import Path

tests/test_sandbox.py

@@ -0,0 +1,122 @@
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
+import os
+import re
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "ui"))
+import frame_connect  # noqa: E402
+import frame_devices  # noqa: E402
+import frame_host  # noqa: E402
+
+
+def inside(path, folder):
+    path, folder = os.path.realpath(path), os.path.realpath(folder)
+    return path == folder or path.startswith(folder + os.sep)
+
+
+def snapshot(folder):
+    """{relative path: (size, mtime)} for everything under FOLDER."""
+    out = {}
+    for base, dirs, files in os.walk(folder):
+        for name in dirs + files:
+            p = os.path.join(base, name)
+            try:
+                st = os.lstat(p)
+            except OSError:
+                continue
+            out[os.path.relpath(p, folder)] = (st.st_size, st.st_mtime_ns)
+    return out
+
+
+class Sandbox(unittest.TestCase):
+    """Tests once wrote a host-key pin into the real ~/.ssh/frame-control-hosts on every
+    run (a server started without FRAME_CONTROL_SSH_DIR imported the real ~/.ssh/config)."""
+
+    def test_home_and_ssh_dir_are_throwaway(self):
+        tmp = tempfile.gettempdir()
+        self.assertTrue(inside(Path.home(), tmp), Path.home())
+        self.assertTrue(inside(os.path.expanduser("~"), tmp))
+        for d in (frame_devices.ssh_dir(), frame_devices.pin_dir(), frame_connect.SSH_DIR,
+                  os.environ["FRAME_CONTROL_SSH_DIR"], frame_host.cache_dir(), frame_host.data_dir()):
+            self.assertTrue(inside(d, tmp), d)
+            self.assertFalse(inside(d, sandbox.REAL_SSH_DIR), d)
+
+    def test_cache_folders_on_every_platform_are_throwaway(self):
+        """cache_dir() reads LOCALAPPDATA on Windows and XDG_CACHE_HOME on Linux, which a
+        moved HOME alone doesn't cover: servers sweep staging folders there."""
+        tmp = tempfile.gettempdir()
+        for platform in ("MAC", "WINDOWS", None):
+            with mock.patch.object(frame_host, "MAC", platform == "MAC"), \
+                    mock.patch.object(frame_host, "WINDOWS", platform == "WINDOWS"):
+                self.assertTrue(inside(frame_host.cache_dir("apk-sources"), tmp), platform)
+        for name in ("LOCALAPPDATA", "APPDATA", "XDG_CACHE_HOME", "XDG_DATA_HOME", "XDG_STATE_HOME"):
+            self.assertTrue(inside(os.environ[name], tmp), name)
+
+    def test_end_to_end_tests_stay_out_of_it(self):
+        """tests/e2e keeps the container's real home (OpenSSH reads ~/.ssh from the password
+        database, not $HOME), so nothing there may import sandbox, even through a test_* module."""
+        here = Path(__file__).resolve().parent
+        pattern = re.compile(r"^\s*(import sandbox|from sandbox |import test_|from test_)", re.M)
+        for f in [*(here / "e2e").glob("*.py"), *(here / "smoke").glob("*.py"), here / "apk_fixtures.py"]:
+            self.assertIsNone(pattern.search(f.read_text(encoding="utf-8")), f)
+
+    def test_end_to_end_tests_start_no_server_outside_their_container(self):
+        """Plain discovery over tests/e2e, with or without FRAME_E2E=1, must skip everything
+        and never start ui/server.py: outside scripts/e2e.sh's container it would run on
+        this computer's real app data, ~/.ssh and cache."""
+        e2e = str(Path(__file__).resolve().parent / "e2e")
+        code = ("import io, sys, unittest; sys.path.insert(0, sys.argv[1]); import harness; calls = []\n"
+                "harness.Server.start = classmethod(lambda cls: calls.append(cls))\n"
+                "suite = unittest.defaultTestLoader.discover(sys.argv[1], top_level_dir=sys.argv[1])\n"
+                "r = unittest.TextTestRunner(stream=io.StringIO()).run(suite)\n"
+                "print(len(calls), r.testsRun, len(r.skipped), len(r.errors) + len(r.failures))")
+        for extra in ({}, {"FRAME_E2E": "1"}):
+            env = {k: v for k, v in os.environ.items() if not k.startswith("FRAME_E2E")}
+            env.update(extra, PYTHONDONTWRITEBYTECODE="1")
+            out = subprocess.run([sys.executable, "-c", code, e2e], env=env, capture_output=True, text=True,
+                                 timeout=120)
+            self.assertEqual(out.returncode, 0, out.stderr[-2000:])
+            started, ran, skipped, broken = map(int, out.stdout.split())
+            self.assertEqual((started, broken), (0, 0), (extra, out.stdout))
+            self.assertEqual(ran, skipped, (extra, out.stdout))
+            self.assertGreater(skipped, 0, extra)
+
+    def test_child_processes_inherit_it(self):
+        out = subprocess.run(
+            [sys.executable, "-c", "import sys; sys.path.insert(0, sys.argv[1]); import frame_devices, frame_connect;"
+             " from pathlib import Path; print(Path.home()); print(frame_devices.ssh_dir()); print(frame_connect.SSH_DIR)",
+             str(ROOT / "ui")], env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
+            capture_output=True, text=True, check=True).stdout.split("\n")
+        for d in filter(None, out):
+            self.assertFalse(inside(d, sandbox.REAL_SSH_DIR), d)
+            self.assertTrue(inside(d, tempfile.gettempdir()), d)
+
+    def test_real_ssh_folder_untouched_by_tests_that_start_servers(self):
+        """Reruns, from the environment the suite was started in, the tests that used to
+        leak (servers started without FRAME_CONTROL_SSH_DIR) plus the device and connect
+        tests, and checks nothing under the real ~/.ssh changed."""
+        real = sandbox.REAL_SSH_DIR
+        if inside(real, tempfile.gettempdir()):
+            self.skipTest("this run's HOME is already a temporary folder")
+        before = snapshot(real)
+        env = {k: v for k, v in sandbox.ORIGINAL_ENV.items() if k != "FRAME_CONTROL_SSH_DIR"}
+        env["PYTHONDONTWRITEBYTECODE"] = "1"
+        r = subprocess.run([sys.executable, "-m", "unittest", "-q", "test_server.OneServer", "test_server.LocalMode",
+                            "test_devices", "test_connect"],
+                           cwd=str(Path(__file__).resolve().parent), env=env, capture_output=True, text=True,
+                           timeout=300)
+        self.assertEqual(r.returncode, 0, r.stderr[-3000:])
+        after = snapshot(real)
+        changed = sorted(k for k in set(before) | set(after) if before.get(k) != after.get(k)
+                         and not k.startswith("agent"))  # ssh-agent sockets come and go on their own
+        self.assertEqual(changed, [], f"tests changed the real {real}")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_server.py

@@ -535,7 +535,11 @@ class UnreachableMessages(unittest.TestCase):
             "ssh: connect to host 192.168.1.9 port 22: No route to host": "isn't answering",
             "ssh: connect to host 192.168.1.9 port 22: Connection refused": "refused",
             "steamos@192.168.1.9: Permission denied (publickey,password).": "SSH key",
-            "Host key verification failed.": "identity changed",
+            "Host key verification failed.": "identity has changed",
+            "@@@@\n@    WARNING: REMOTE HOST IDENTIFICATION HAS CHANGED!     @\nHost key verification failed.":
+                "Check the new key on the Devices tab",
+            "No ED25519 host key is known for frame-control-f7b4e9ac and you have requested strict checking.\n"
+            "Host key verification failed.": "isn't saved yet",
             "kex_exchange_identification: read: Connection reset by peer": "dropped",
             "Timed out talking to frame": "too long",
         }

tests/test_sidequest.py

@@ -1,3 +1,4 @@
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import json
 from pathlib import Path
 import sys

tests/test_telemetry.py

@@ -335,6 +335,26 @@ class LinkFailureProvenance(Base):
         tm.diagnostic("job steam", e)
         self.assertEqual([p["properties"]["error_category"] for p in self.queued()], ["frame_unreachable"])
 
+    def test_host_key_and_login_refusals_are_sent_once(self):
+        # 0.4.2: each of these came in as a $exception from the status poll, again every 10 minutes.
+        for _ in range(3):
+            for stderr in ("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@\r\n"
+                           "@    WARNING: REMOTE HOST IDENTIFICATION HAS CHANGED!     @\r\n"
+                           "IT IS POSSIBLE THAT SOMEONE IS DOING SOMETHING NASTY!\r\nHost key verification failed.\r\n",
+                           "No ED25519 host key is known for frame-control-f7b4e9ac and you have requested strict "
+                           "checking.\r\nHost key verification failed.\r\n",
+                           "jane doe@frame-control-f7b4e9ac: Permission denied (publickey,password).\r\n",
+                           "steamos@192.168.1.9: Permission denied (publickey,password).\n"):
+                e = self.ssh_fails(self.server, 255, stderr)
+                self.assertTrue(getattr(e, "frame_link_failed", False), stderr)
+                tm.diagnostic("POST /api/comfort status", e)
+        self.assertEqual([p["properties"]["error_category"] for p in self.queued()], ["frame_auth"])
+        # A command on the Frame that was refused something isn't ssh refusing the headset.
+        for code, stderr in ((1, "steamos@frame: Permission denied (publickey).\n"),
+                             (255, "cat: /root/x: Permission denied\n"),
+                             (255, "scp: /home/steamos/x: Permission denied\n")):
+            self.assertFalse(getattr(self.ssh_fails(self.server, code, stderr), "frame_link_failed", False), stderr)
+
     def test_a_command_that_fails_on_the_frame_is_not_marked(self):
         self.poll_fails()
         for module in (self.server, self.android):

tests/test_touch.py

@@ -2,6 +2,7 @@
 
 Run: python3 -m unittest discover -s tests
 """
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import io
 import json
 import os

tests/test_vr.py

@@ -1,4 +1,5 @@
 """Own VR telemetry, optional catalogue and entitlement guards; no headset needed."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import ctypes
 import json
 from pathlib import Path

ui/frame_connect.py

@@ -31,7 +31,8 @@ import frame_host  # noqa: E402
 FRAME_USER = os.environ.get("FRAME_USER", "steamos")
 USER_FROM_ENV = "FRAME_USER" in os.environ
 FRAME_ALIAS = os.environ.get("FRAME_ALIAS", "frame")
-SSH_DIR = Path.home() / ".ssh"
+# $FRAME_CONTROL_SSH_DIR in tests, as frame_devices.ssh_dir(), so they never touch the real one.
+SSH_DIR = Path(os.environ.get("FRAME_CONTROL_SSH_DIR") or Path.home() / ".ssh")
 KEY = SSH_DIR / "id_ed25519_frame"
 # The devkit service only accepts ssh-rsa keys (write_key in Valve's
 # steamos-devkit-service), so pairing uses a second key next to the ed25519 one.

ui/frame_devices.py

@@ -28,8 +28,10 @@ remembered IP is caught.
 
 Python stdlib only.
 """
+import base64
 import contextlib
 import copy
+import hashlib
 import json
 import os
 import re
@@ -418,19 +420,100 @@ def seed_pin(device_id, hosts, port=22, sources=None):
                 if len(f) >= 3 and not line.startswith("#") and not f[0].startswith("@"):
                     keys.append(f"{name} {f[1]} {f[2]}")
         if keys:
-            target = known_hosts(device_id)
-            target.parent.mkdir(**({} if frame_host.WINDOWS else {"mode": 0o700}), parents=True, exist_ok=True)
-            fd_, tmp = tempfile.mkstemp(prefix=".seed-", dir=str(target.parent))
-            with os.fdopen(fd_, "w", encoding="utf-8") as fh:
-                fh.write("\n".join(dict.fromkeys(keys)) + "\n")
-            os.replace(tmp, target)  # whole file at once: ssh never sees half of it
+            _write_pin(device_id, keys)
             return True
     return False
 
 
+def pin_dir():
+    """Where the per-headset files go, created private: ssh won't create it itself."""
+    d = ssh_dir() / PIN_DIR
+    d.mkdir(**({} if frame_host.WINDOWS else {"mode": 0o700}), parents=True, exist_ok=True)
+    return d
+
+
+def _write_pin(device_id, lines):
+    target = known_hosts(device_id)
+    fd_, tmp = tempfile.mkstemp(prefix=".seed-", dir=str(pin_dir()))
+    with os.fdopen(fd_, "w", encoding="utf-8") as fh:
+        fh.write("\n".join(dict.fromkeys(lines)) + "\n")
+    frame_host.make_private(tmp)  # Windows: this user only, not the folder's inherited ACL
+    os.replace(tmp, target)  # whole file at once: ssh never sees half of it
+
+
+# The host key types OpenSSH writes to known_hosts.
+KEY_TYPE_RE = re.compile(r"ssh-ed25519|ssh-rsa|ssh-dss|ecdsa-sha2-nistp(?:256|384|521)|"
+                         r"sk-ssh-ed25519@openssh\.com|sk-ecdsa-sha2-nistp256@openssh\.com")
+
+
+def fingerprint(blob):
+    """OpenSSH's SHA256 fingerprint of a base64 key blob (what `ssh-keygen -l` and ssh -v print),
+    or None if it isn't base64."""
+    try:
+        raw = base64.b64decode(blob, validate=True)
+    except (ValueError, TypeError):
+        return None
+    return "SHA256:" + base64.b64encode(hashlib.sha256(raw).digest()).decode().rstrip("=")
+
+
+def key_entry(line, name):
+    """(type, blob, fingerprint) from a known_hosts line for `name`, or None."""
+    f = (line or "").split()
+    if len(f) < 3 or line.startswith("#") or name not in f[0].split(",") or not KEY_TYPE_RE.fullmatch(f[1]):
+        return None
+    fp = fingerprint(f[2])
+    return (f[1], f[2], fp) if fp else None
+
+
+def pinned_keys(device_id):
+    """[(type, fingerprint)] of the keys saved for a headset (plain entries only)."""
+    try:
+        lines = known_hosts(device_id).read_text(encoding="utf-8").splitlines()
+    except (OSError, UnicodeDecodeError):
+        return []
+    name = host_key_alias(device_id)
+    return [(e[0], e[2]) for e in (key_entry(line, name) for line in lines) if e]
+
+
+def pin_key(device_id, key_type, blob):
+    """Save one key as the headset's whole identity, replacing whatever was saved: what the
+    person confirmed after seeing its fingerprint (see frame_link.Link.trust_identity)."""
+    if not KEY_TYPE_RE.fullmatch(key_type or "") or not fingerprint(blob):
+        raise DeviceError("That isn't an SSH host key")
+    _write_pin(device_id, [f"{host_key_alias(device_id)} {key_type} {blob}"])
+
+
+def scan_file(device_id):
+    """(path, how ssh is told) for a throwaway known_hosts file that a key check writes the
+    headset's key to, without touching the saved one. `~` as in known_hosts_opt."""
+    name = f".scan-{device_id}-{secrets.token_hex(4)}"
+    path = pin_dir() / name
+    path.touch(**({} if frame_host.WINDOWS else {"mode": 0o600}))
+    frame_host.make_private(path)
+    return path, (str(path) if os.environ.get("FRAME_CONTROL_SSH_DIR") else f"~/.ssh/{PIN_DIR}/{name}")
+
+
+def identity_established(device):
+    """Whether this headset ever had an SSH identity here: the device's own flag (set once it
+    had a key or a connection that worked, and never cleared, whatever happens to its
+    addresses), a fingerprint recorded in devices.json, a pin file at all (even empty,
+    unreadable or not parseable), or an address that worked. Then a missing or unusable pin
+    fails closed, needing Check its key; only a headset with none of these may save the
+    first key it meets (accept-new)."""
+    if device.get("identity_established") or device.get("identity"):
+        return True
+    try:
+        if os.path.lexists(known_hosts(device["id"])):
+            return True
+    except OSError:
+        return True
+    return any(a.get("last_ok") for a in device.get("addresses") or ())
+
+
 def forget_pin(device_id):
-    """Drop a headset's saved key, e.g. after SteamOS was reinstalled. The next connection
-    trusts whatever key the headset shows, as a first connection does."""
+    """Drop a headset's saved key (the headset was removed, or the API's forget-identity).
+    The next connection trusts whatever key the headset shows, as a first connection does;
+    the page's Trust a new key flow (frame_link) shows the key first instead."""
     try:
         known_hosts(device_id).unlink()
         return True
@@ -510,6 +593,10 @@ class Registry:
             for a in d["addresses"]:
                 a.setdefault("networks", [])
                 a.setdefault("label", "")
+            if d.get("identity") or any(a.get("last_ok") for a in d["addresses"]):
+                # Files from before the flag existed: it had an identity (or a connection that
+                # worked, which saved one). Kept from now on, whatever happens to its addresses.
+                d["identity_established"] = True
             return NAME_RE.fullmatch(d.get("id", "")) is not None
         except (KeyError, TypeError, DeviceError):
             return False
@@ -712,8 +799,25 @@ class Registry:
                 a["networks"] = (a["networks"] + [network_id])[-16:]
             a["last_ok"] = time.time()
             a["last_rtt_ms"] = rtt_ms
+            self._find(device_id)["identity_established"] = True  # survives editing the address away
             self.save()
 
+    def record_identity(self, device_id, fingerprints):
+        """Remember the fingerprints of the key saved for a headset, so that losing the pin
+        file later can't turn into trusting whatever answers (identity_established)."""
+        fingerprints = [f for f in fingerprints if isinstance(f, str) and f.startswith("SHA256:")][:8]
+        if not fingerprints:
+            return
+        with self._changing():
+            try:
+                d = self._find(device_id)
+            except DeviceError:
+                return
+            if d.get("identity") != fingerprints or not d.get("identity_established"):
+                d["identity"] = fingerprints
+                d["identity_established"] = True
+                self.save()
+
     def undismiss(self, alias):
         """Set Up Connection is about to run for this alias: import its block again."""
         with self._changing():
@@ -797,15 +901,17 @@ class Registry:
                         d["addresses"].append(dict(new_address(host), label="From Set Up Connection"))
                     d["config_host"] = host
                     changed = True
-                    if seed and host:
-                        seed_pin(d["id"], [host], b["port"])
+                    if seed and host and seed_pin(d["id"], [host], b["port"]):
+                        d["identity"] = [f for _, f in pinned_keys(d["id"])][:8] or d.get("identity")
+                        d["identity_established"] = True
                 elif host and host != d.get("config_host"):
                     # Set Up Connection ran again and found the headset somewhere new.
                     d["config_host"] = host
                     if not any(a["host"] == host for a in d["addresses"]):
                         d["addresses"].insert(0, dict(new_address(host), label="From Set Up Connection"))
-                    if seed:
-                        seed_pin(d["id"], [host], b["port"])
+                    if seed and seed_pin(d["id"], [host], b["port"]) and not d.get("identity"):
+                        d["identity"] = [f for _, f in pinned_keys(d["id"])][:8] or None
+                        d["identity_established"] = True
                     changed = True
                 if d.get("config_login") != [user, b["port"]]:
                     # Set Up Connection (or an edit) changed who to log in as, or the port.

ui/frame_host.py

@@ -38,9 +38,13 @@ class HostError(RuntimeError):
 
 
 # The start of a line in which ssh itself says the link to the headset failed (not the command it ran).
+# That includes ssh refusing the headset's host key, and the headset refusing this computer's
+# key ("user@host: Permission denied (publickey,...)."): either way no command ran.
 SSH_LINK_FAILED = re.compile(r"^(?:ssh: connect to host |ssh: Could not resolve hostname |banner exchange: |"
                              r"kex_exchange_identification: |mux_client_\w+: |client_loop: |"
-                             r"Connection (?:closed|reset) by \S+ port \d+|Connection timed out during banner exchange)",
+                             r"Connection (?:closed|reset) by \S+ port \d+|Connection timed out during banner exchange|"
+                             r"Host key verification failed\.|No \S+ host key is known for |"
+                             r"[^\n@:]{1,64}@\S+: Permission denied \()",  # a Windows user name may have spaces
                              re.M)
 
 

ui/frame_link.py

@@ -28,11 +28,14 @@ Python stdlib only. Runs on this computer, never on the Frame.
 import copy
 import hashlib
 import ipaddress
+import os
 import queue
 import re
+import shlex
 import socket
 import subprocess
 import sys
+import tempfile
 import threading
 import time
 
@@ -50,6 +53,7 @@ NETWORK_EVERY = 5      # how often the network fingerprint is read
 TAILSCALE_EVERY = 30
 RETRY = (5, 10, 20, 30)  # seconds before automatic retries after a failure
 REQUEST_GAP = 5        # a request may start a new attempt this long after the last one
+SCAN_TTL = 600         # a key shown for "Trust a new key" may be trusted this long after it was read
 
 STAGES = [("network", "Checking this computer's network"), ("find", "Finding the headset"),
           ("ssh", "Opening SSH"), ("identity", "Checking the headset's identity"),
@@ -90,6 +94,32 @@ def ssh_g(alias):
     return got.get("hostname") or alias, port, got.get("user"), proxied
 
 
+def proxy_route(alias):
+    """How ~/.ssh/config sends ALIAS: (None, None) directly, ("jump", spec) through
+    ProxyJump, or ("command", text) through a ProxyCommand."""
+    try:
+        out = frame_host.run_ssh(["ssh", "-G", alias], capture_output=True, stdin=subprocess.DEVNULL, text=True,
+                                 timeout=10).stdout
+    except (OSError, subprocess.TimeoutExpired):
+        out = ""
+    got = {}
+    for line in out.splitlines():
+        k, _, v = line.partition(" ")
+        if k in ("proxyjump", "proxycommand") and k not in got:
+            got[k] = v.strip()
+    if got.get("proxyjump") not in (None, "", "none"):
+        return "jump", got["proxyjump"]
+    if got.get("proxycommand") not in (None, "", "none"):
+        return "command", got["proxycommand"]
+    return None, None
+
+
+# One jump host, as ProxyJump names it: [user@]host[:port], nothing that could start an
+# option, split a shell word or carry an ssh % token.
+JUMP_RE = re.compile(r"(?:([A-Za-z0-9][A-Za-z0-9._-]{0,63})@)?"
+                     r"(\[[0-9A-Fa-f:.]{2,45}\]|[A-Za-z0-9][A-Za-z0-9.-]{0,252})(?::(\d{1,5}))?")
+
+
 def probe(host, port, timeout=PROBE_TIMEOUT, update=None):
     """Try a TCP connection to host:port. -> {"state", "detail", "ip", "rtt_ms"}.
 
@@ -257,6 +287,43 @@ def ssh_target(host, ip):
         return host
 
 
+def _unlink(path):
+    try:
+        os.unlink(path)
+    except OSError:
+        pass
+
+
+def _scanned_key(path, name):
+    """The first whole, valid known_hosts line for `name` that a key check wrote, or None.
+    Only lines ssh has finished (ending in a newline): a half-written key mustn't count."""
+    try:
+        with open(path, encoding="utf-8") as fh:
+            text = fh.read()
+    except (OSError, UnicodeDecodeError):
+        return None
+    for line in text.splitlines(True):
+        if line.endswith("\n"):
+            entry = frame_devices.key_entry(line.strip(), name)
+            if entry:
+                return entry
+    return None
+
+
+def proxy_command(literal):
+    """A ProxyCommand that runs `literal` (argv, ending with the jump host) with `-W` to the
+    headset, each word exactly as given. ssh first expands its %-tokens in the line, so every
+    % of ours is doubled; then it hands the line to a shell (`$SHELL -c` on macOS and Linux)
+    or, on Windows, starts it as a command line, so every word is quoted for that: a path
+    with spaces, $(...), backticks or quotes stays one plain word. The forward is
+    [%h]:%p, so an IPv6 address (with a zone) reads as one host."""
+    words = [w.replace("%", "%%") for w in literal]
+    target = "[%h]:%p"
+    if frame_host.WINDOWS:
+        return subprocess.list2cmdline([*words[:-1], "-W", target, words[-1]])
+    return " ".join(shlex.quote(w) for w in [*words[:-1], "-W", target, words[-1]])
+
+
 def probe_raw(host, port, result):
     """ssh's own wording for a failed probe, so the server's UNREACHABLE table explains it."""
     return {"unresolved": f"ssh: Could not resolve hostname {host}: not found",
@@ -275,7 +342,8 @@ class Link:
         self.session_alias = env_alias  # ...and stays selectable after switching away
         self.mux_base = list(mux_base)  # ["ssh", "-o", "BatchMode=yes", ControlPath...]
         self.control = control          # ControlPath, or None where ssh can't share connections
-        self.apply = apply              # apply(alias, host_opts): point every ssh command at the headset
+        self._apply = apply             # apply(alias, host_opts): point every ssh command at the headset
+        self.applied = (None, [])       # what it was last given
         self.explain = explain          # ssh error text -> plain reason, or None
         self.cond = threading.Condition()
         self.version = 0
@@ -301,6 +369,7 @@ class Link:
         self.routed_device = None
         self.last_failure = None        # {"stage", "category", "at"}: for report diagnostics, no free text
         self.attempt_device = None      # the headset the current attempt is for
+        self.scans = {}                 # device id -> the host key it showed for "Trust a new key"
 
     # ---- publishing ----
     def publish(self, **fields):
@@ -505,13 +574,76 @@ class Link:
             return [*mux, *(device.get("frozen") or [])]  # what ~/.ssh/config said when it was routed
         if not host:  # a headset with no addresses: reach nothing, not whatever ~/.ssh/config says
             return ["-o", "HostName=no-address.invalid"]
-        # StrictHostKeyChecking=yes: whatever ~/.ssh/config says for Host *, every command
-        # checks the headset's pinned key (only the connector's first handshake may save one).
-        return [*mux, "-o", "StrictHostKeyChecking=yes", "-o", f"HostName={frame_devices.ssh_host(host)}",
+        # Whatever ~/.ssh/config says for Host *, every command checks the headset's pinned key:
+        # StrictHostKeyChecking=yes. Only a headset that never had an identity here (just
+        # added, never connected) gets accept-new: the first contact saves its key, as Set Up
+        # Connection and the connector's first handshake do, rather than every command failing
+        # with "No host key is known" until the connector gets there. Once it had one, a pin
+        # file that's gone, empty or unreadable fails closed (identity_established): accept-new
+        # can't tell that from a new headset and would trust whatever answers.
+        strict = self.strictness(device)
+        if strict == "accept-new":
+            try:
+                frame_devices.pin_dir()  # ssh won't create the folder its known_hosts file goes in
+            except OSError:
+                pass  # ssh then says it couldn't save the key, and goes on
+        return [*mux, "-o", f"StrictHostKeyChecking={strict}", "-o", f"HostName={frame_devices.ssh_host(host)}",
                 "-o", f"HostKeyAlias={frame_devices.host_key_alias(device['id'])}",
                 "-o", f"UserKnownHostsFile={frame_devices.known_hosts_opt(device['id'])}", "-o", "HashKnownHosts=no",
                 "-o", f"User={device['user']}", "-o", f"Port={device['port']}"]
 
+    def apply(self, alias, opts):
+        self.applied = (alias, list(opts))
+        self._apply(alias, opts)
+
+    def strictness(self, device):
+        """yes, or accept-new for a headset that never had an identity. Judged on what the
+        registry says now, not only the copy handed in (a route frozen at connect, say, from
+        before its first key was recorded). A key saved since is recorded on the way."""
+        try:
+            current = self.reg.get(device["id"])
+        except frame_devices.DeviceError:
+            current = {}
+        if not (frame_devices.identity_established(device) or
+                (current and frame_devices.identity_established(current))):
+            return "accept-new"
+        self.settle_pin(dict(device, **current) if current else device, tighten=False)
+        return "yes"
+
+    def settle_pin(self, device=None, tighten=True):
+        """A key ssh saved on a first contact, whatever happened next (logged in or refused,
+        by the handshake or any command): record it, make the file private, and from then on
+        route every command strictly. Idempotent and cheap when there's nothing to do."""
+        device = device or self.routed_device
+        if not device or device.get("transient") or device.get("none") or not device.get("id"):
+            return
+        did = device["id"]
+        if frame_devices.pinned(did):
+            fps = [f for _, f in frame_devices.pinned_keys(did)]
+            if fps and (device.get("identity") != fps or not device.get("identity_established")):
+                try:
+                    current = self.reg.get(did)
+                except frame_devices.DeviceError:
+                    return
+                if current.get("identity") != fps or not current.get("identity_established"):
+                    frame_host.make_private(frame_devices.known_hosts(did))  # ssh wrote it
+                    self.reg.record_identity(did, fps)
+        if not tighten:
+            return
+        with self.route_lock:
+            alias, opts = self.applied
+            if alias == device.get("alias") and "StrictHostKeyChecking=accept-new" in opts and \
+                    frame_devices.identity_established(dict(device, **self._current(did))):
+                self.opts = [("StrictHostKeyChecking=yes" if o == "StrictHostKeyChecking=accept-new" else o)
+                             for o in opts]
+                self.apply(alias, self.opts)
+
+    def _current(self, device_id):
+        try:
+            return self.reg.get(device_id)
+        except frame_devices.DeviceError:
+            return {}
+
     def public_device(self, d):
         return {k: d.get(k) for k in ("id", "name", "alias", "user", "port", "transient")}
 
@@ -541,6 +673,8 @@ class Link:
                 if t - last_ts >= TAILSCALE_EVERY and self.state["network"] and not reasons:
                     last_ts = t
                     self.refresh_network()
+                if "StrictHostKeyChecking=accept-new" in self.applied[1]:
+                    self.settle_pin()  # a command's first contact may have saved the key
                 phase = self.state["phase"]
                 if phase == "connected" and not self.alive():
                     reasons.append("dropped")
@@ -675,10 +809,11 @@ class Link:
             return "Starting up"
         return "Connecting"
 
-    def fail(self, sid, message, raw=""):
+    def fail(self, sid, message, raw="", **more):
+        """A stage failed. `more`: what the page can offer to fix it (hostkey, auth)."""
         self.stage(sid, "failed", message)
         with self.cond:
-            self.state["error"] = {"stage": sid, "message": message, "raw": raw}
+            self.state["error"] = dict({"stage": sid, "message": message, "raw": raw}, **more)
             probes = copy.deepcopy(self.state["probes"])
         self.note_failure(sid, message, raw, probes)
 
@@ -867,6 +1002,7 @@ class Link:
         if device.get("transient"):
             return
         self.reg.record_success(device["id"], host, net.get("id"), rtt)
+        self.reg.record_identity(device["id"], [f for _, f in frame_devices.pinned_keys(device["id"])])
         # Set Up Connection may have changed the block while this attempt ran: take that
         # in (and reconnect) rather than writing this attempt's older settings over it.
         self.watch_config()
@@ -921,7 +1057,17 @@ class Link:
 
     def handshake(self, device, a, found, user):
         """SSH to one address, following ssh -v through stages 3-5.
-        -> "ok", "next" (try another address) or "stop"."""
+        -> "ok", "next" (try another address) or "stop". A key ssh saved is recorded and
+        protected however it ends: a login refused after the key was saved, a timeout, a stop."""
+        try:
+            return self._handshake(device, a, found, user)
+        finally:
+            if not device.get("transient"):
+                with self.route_lock:
+                    current = self.attempt_gen == self.gen
+                self.settle_pin(device, tighten=current)
+
+    def _handshake(self, device, a, found, user):
         # The IP that answered (with a link-local IPv6 address's zone), so ssh doesn't
         # look the name up again and stall on an address that didn't answer.
         opts = self.host_opts(device, ssh_target(a["host"], found.get("ip")))
@@ -939,15 +1085,9 @@ class Link:
             return "ok"
         extra = []
         if not device.get("transient"):
-            if frame_devices.pinned(device["id"]):
-                extra = ["-o", "StrictHostKeyChecking=yes"]
-            else:
-                # First connection since this headset was added: trust what it shows
-                # (as Set Up Connection does), and pin it from now on. ssh won't create
-                # the folder its known_hosts file goes in.
-                extra = ["-o", "StrictHostKeyChecking=accept-new"]
-                pins = frame_devices.known_hosts(device["id"]).parent
-                pins.mkdir(**({} if frame_host.WINDOWS else {"mode": 0o700}), parents=True, exist_ok=True)
+            # A headset's first connection trusts what it shows (as Set Up Connection does) and
+            # pins it; any later one checks the pin strictly (see host_opts).
+            extra = ["-o", f"StrictHostKeyChecking={self.strictness(device)}"]
         if self.control:
             # No ConnectTimeout: with it, OpenSSH's master takes ~5s to open its socket.
             # `extra` first: ssh takes the first value of an option, and it may say accept-new.
@@ -955,27 +1095,70 @@ class Link:
                     "-o", "ServerAliveCountMax=2", "-N", alias]
         else:
             argv = [*self.mux_base, *extra, *opts, "-v", "-o", "ConnectTimeout=10", alias, "true"]
+        lines = queue.Queue()
+        collecting = [True]
+        reader_done = threading.Event()
+        errfile = None
         try:
-            proc = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
-                                    stderr=subprocess.PIPE, **frame_host.DETACHED)
+            if self.control:
+                proc = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
+                                        stderr=subprocess.PIPE, **frame_host.DETACHED)
+            else:
+                # No master (Windows): ssh runs `true` and exits. Its stderr goes to a file read
+                # as it grows, not a pipe: Windows' OpenSSH can block writing to a redirected
+                # pipe (see frame_host.run_ssh), and the handshake would never finish.
+                fd_, errfile = tempfile.mkstemp(prefix="frame-ssh-")
+                try:
+                    proc = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
+                                            stderr=fd_, **frame_host.DETACHED)
+                finally:
+                    os.close(fd_)
         except OSError as e:
+            if errfile:
+                _unlink(errfile)
             self.fail("ssh", f"Couldn't run ssh: {e}", str(e))
             return "stop"
         self.pending = proc  # so stop() can end it mid-handshake
-        lines = queue.Queue()
-        collecting = [True]
 
         def read():
             for raw in iter(proc.stderr.readline, b""):
                 if collecting[0]:
                     lines.put(raw.decode("utf-8", "replace").rstrip())
+            reader_done.set()
             lines.put(None)
             proc.stderr.close()
-        threading.Thread(target=read, daemon=True).start()
+
+        def tail():
+            buf = b""
+            try:
+                with open(errfile, "rb") as fh:
+                    while True:
+                        chunk = fh.read()
+                        if not chunk:
+                            if proc.poll() is None:
+                                time.sleep(0.05)
+                                continue
+                            chunk = fh.read()  # whatever it wrote just before exiting
+                            if not chunk:
+                                break
+                        buf += chunk
+                        *done, buf = buf.split(b"\n")
+                        for raw in done:
+                            lines.put(raw.decode("utf-8", "replace").rstrip())
+            except OSError:
+                pass
+            if buf:
+                lines.put(buf.decode("utf-8", "replace").rstrip())
+            reader_done.set()
+            lines.put(None)
+            _unlink(errfile)
+        threading.Thread(target=read if errfile is None else tail, daemon=True).start()
 
         step, said, authed = "ssh", [], False
         deadline = time.monotonic() + HANDSHAKE_TIMEOUT
         mismatch = False
+        shown = None  # (type, fingerprint) of the key the headset showed
+        exited = None  # when ssh was first seen to have exited
         while True:
             left = deadline - time.monotonic()
             if self.stopped:
@@ -993,7 +1176,14 @@ class Link:
                 if authed and self.control and self.check(opts, alias):
                     break
                 if proc.poll() is not None and lines.empty():
-                    line = None
+                    # ssh exited: its last words (Permission denied…) may still be on their way.
+                    # The reader's end marker says when they're all in; a pipe that something
+                    # else holds open (a ProxyCommand's child) gets a short grace instead.
+                    exited = exited or time.monotonic()
+                    if reader_done.is_set() and lines.empty() or time.monotonic() - exited > 2:
+                        line = None
+                    else:
+                        continue
                 else:
                     continue
             if line is None:  # ssh exited
@@ -1002,7 +1192,7 @@ class Link:
                     break
                 if authed and self.control and self.check(opts, alias):
                     break
-                return self.failed(step, said, mismatch, alias)
+                return self.failed(step, said, mismatch, alias, shown, a["host"], device)
             if not line.startswith("debug"):
                 said.append(line)
             m = CONNECTING.search(line)
@@ -1015,12 +1205,15 @@ class Link:
                 self.stage("identity", "active", "Waiting for the headset's host key")
             elif HOSTKEY.search(line):
                 m = HOSTKEY.search(line)
+                shown = (m.group(1), m.group(2))
                 self.stage("identity", "active", f"It shows {m.group(1)} key {m.group(2)[:20]}…")
             elif KNOWN.search(line):
                 self.stage("identity", "done", "Matches the identity saved for this headset")
                 step = "login"
                 self.stage("login", "active", f"Logging in as {user}")
             elif ADDED.search(line):
+                if not device.get("transient"):
+                    self.settle_pin(device, tighten=False)  # now, before the login can fail
                 self.stage("identity", "done", "First connection: saved this headset's identity")
                 step = "login"
                 self.stage("login", "active", f"Logging in as {user}")
@@ -1059,18 +1252,34 @@ class Link:
         # OpenSSH lower-cases host names (FRAME.LOCAL logs as frame.local).
         return not m or m.group(1).lower() in (host.lower(), alias.lower()) or "Authentication succeeded" in line
 
-    def failed(self, step, said, mismatch, alias):
+    def failed(self, step, said, mismatch, alias, shown=None, host=None, device=None):
         text = "\n".join(said).strip()
         if mismatch:
-            self.fail("identity", "This address answered as a different headset (its SSH identity doesn't match). "
-                      "If SteamOS was reinstalled, use Forget Identity on the Devices tab.", text)
+            # A key that differs from the saved one, or (strict, pinned) none of the saved key's
+            # type any more: either way not the identity saved. Never accepted here: the page
+            # shows the new key's fingerprint and the person decides (Trust a new key).
+            name = (device or {}).get("name") or alias
+            lost = bool(device) and not frame_devices.pinned(device["id"]) and \
+                frame_devices.identity_established(device)
+            if lost:
+                message = (f"The SSH identity saved for {name} is missing or can't be read, so Frame Control won't "
+                           "trust whatever answers. Check its key and trust it on the Devices tab.")
+            else:
+                message = (f"This address answered with a different SSH identity from the one saved for {name}. "
+                           "If SteamOS was reinstalled or reset, or this is a new headset, check its new key and trust "
+                           "it on the Devices tab. If not, a different device may be answering here.")
+            self.fail("identity", message, text,
+                      hostkey={"state": "missing" if lost else "changed", "type": shown[0] if shown else None,
+                               "fingerprint": shown[1] if shown else None, "host": host})
             return "next"
         # A refused key is the same at every address: stop. (Judged by ssh's own words, not
         # the step: a jump host's progress lines look like the headset's.) A forward that
         # a jump host couldn't open is about this address only: try the next.
         forward = re.search(r"open failed|forwarding failed|Connection refused|Connection closed|timed out", text)
         if re.search(r"Permission denied", text) and not forward:
-            self.fail("login", self.explain(text) or "The headset didn't accept this computer's key.", text)
+            # Not something a retry fixes: the headset needs this computer's key (Set Up Connection).
+            self.fail("login", self.explain(text) or "The headset didn't accept this computer's key. Run Set Up "
+                      "Connection: it pairs this computer with the headset again.", text, auth="denied")
             return "stop"
         if step == "connected":
             self.fail("login", "Logged in, but the shared SSH connection didn't start.", text)
@@ -1110,9 +1319,11 @@ class Link:
             lead = f"Answered in {res['rtt_ms']:g} ms" if res.get("rtt_ms") is not None else "Through the jump host"
             rows[i]["ssh"] = "checking"
             put()
+            # "yes" before host_opts (ssh takes an option's first value): a test only looks, so
+            # it reports a key not saved yet rather than saving it as host_opts' accept-new would.
             argv = [*self.mux_base[:3], "-o", "ControlPath=none", "-o", "ConnectTimeout=8",
-                    *self.host_opts(device, ssh_target(a["host"], res.get("ip"))),
-                    "-o", "StrictHostKeyChecking=yes", device["alias"], "true"]
+                    "-o", "StrictHostKeyChecking=yes",
+                    *self.host_opts(device, ssh_target(a["host"], res.get("ip"))), device["alias"], "true"]
             try:
                 r = frame_host.run_ssh(argv, capture_output=True, stdin=subprocess.DEVNULL, text=True,
                                        errors="replace", timeout=20)
@@ -1140,6 +1351,156 @@ class Link:
         put(done=True, finished=now())
         self.devices_changed()
 
+    # ---- Trust a new key ----
+    # Every way a key check could do more than read the key: no login of any kind, no agent
+    # (nor forwarding it), no X11, port or tunnel forwarding, no local or remote command, no
+    # terminal, no shared connection, no learning keys. Given first, so ssh takes these over
+    # anything ~/.ssh/config (or host_opts) says. -N: no command, even if a device accepts "none".
+    SCAN_OPTS = ["BatchMode=yes", "ControlMaster=no", "ControlPath=none", "ConnectTimeout=8",
+                 "StrictHostKeyChecking=accept-new", "UpdateHostKeys=no", "HashKnownHosts=no",
+                 "PreferredAuthentications=none", "PubkeyAuthentication=no", "PasswordAuthentication=no",
+                 "KbdInteractiveAuthentication=no", "HostbasedAuthentication=no", "IdentityAgent=none",
+                 "ForwardAgent=no", "ForwardX11=no", "ForwardX11Trusted=no", "ClearAllForwardings=yes",
+                 "Tunnel=no", "PermitLocalCommand=no", "RemoteCommand=none", "RequestTTY=no"]
+    SCAN_TIMEOUT = 20
+
+    def scan_identity(self, device_id):
+        """Read the headset's host key without trusting it or logging in. ssh writes it to a
+        throwaway known_hosts file during the key exchange, before any authentication, and is
+        stopped as soon as it has (or when it gives up); SCAN_OPTS and -N rule out anything else.
+        `-F none` too, unless ~/.ssh/config sends the alias through a jump host (then it's
+        needed for that, and SCAN_OPTS still win). The address that showed the changed key goes
+        first, then the rest in order. -> {"type", "fingerprint", "host", "saved", "recorded",
+        "matches", "recorded_match"}; the key is kept (SCAN_TTL) for trust_identity, which pins
+        exactly it."""
+        device = self.reg.get(device_id)
+        if not device["addresses"]:
+            raise frame_devices.DeviceError("This headset has no addresses to ask")
+        hosts = [a["host"] for a in device["addresses"]]
+        with self.cond:
+            err = self.state.get("error") or {}
+            mine = (self.state.get("device") or {}).get("id") == device_id
+        first = (err.get("hostkey") or {}).get("host") if mine else None
+        if first in hosts:
+            hosts.remove(first)
+            hosts.insert(0, first)
+        route = self.scan_route(device)
+        reasons = []
+        for host in hosts:
+            key, err_text = self._scan_one(device, host, route)
+            if key:
+                saved = frame_devices.pinned_keys(device_id)
+                recorded = [f for f in device.get("identity") or [] if isinstance(f, str)]
+                scan = {"type": key[0], "blob": key[1], "fingerprint": key[2], "host": host,
+                        "port": device["port"], "user": device["user"], "at": now()}
+                with self.cond:
+                    self.scans[device_id] = scan
+                return {"type": key[0], "fingerprint": key[2], "host": host,
+                        "saved": [{"type": t, "fingerprint": f} for t, f in saved], "recorded": recorded,
+                        "matches": any(f == key[2] for _, f in saved), "recorded_match": key[2] in recorded}
+            text = err_text.strip()
+            reasons.append(f"{host}: " + (self.explain(text) or (text.splitlines() or ["no answer"])[-1]))
+        raise frame_devices.DeviceError("Couldn't read the headset's key. " + "; ".join(reasons))
+
+    # The jump host's own ssh (a ProxyCommand the scan builds itself): it may log in to the jump
+    # host as configured (keys, the agent), but it never forwards the agent, X11 or ports, runs
+    # no local or remote command, has no terminal, shares no connection and goes no further.
+    HOP_OPTS = ["BatchMode=yes", "ForwardAgent=no", "ForwardX11=no", "ForwardX11Trusted=no",
+                "ClearAllForwardings=yes", "Tunnel=no", "PermitLocalCommand=no", "RemoteCommand=none",
+                "RequestTTY=no", "ControlMaster=no", "ControlPath=none", "ProxyCommand=none", "ProxyJump=none"]
+
+    def scan_route(self, device):
+        """The options that take a key check to the headset: `-F none` and nothing else when
+        it's reached directly. Through a jump host (ProxyJump), `-F none` and a ProxyCommand
+        built here, whose ssh gets HOP_OPTS ahead of ~/.ssh/config: config options can't be
+        overridden for ProxyJump's own ssh, which inherits ForwardAgent, LocalCommand and the
+        rest. Through a ProxyCommand from ~/.ssh/config, or several jump hosts, it refuses:
+        those can't be made safe from here."""
+        kind, spec = proxy_route(device["alias"])
+        if kind is None:
+            return ["-F", "none"]
+        m = JUMP_RE.fullmatch(spec or "") if kind == "jump" else None
+        if not m or (m.group(3) and not 1 <= int(m.group(3)) <= 65535):
+            how = "a ProxyCommand" if kind == "command" else "this jump host setting"
+            raise frame_devices.DeviceError(
+                f"{device['name']} is reached through {how} in ~/.ssh/config, which Frame Control can't make safe "
+                "for talking to a device it doesn't trust yet, so it won't read the key that way. Add an address "
+                "it can reach directly (or through a single ProxyJump host) and check its key again; or compare "
+                "the key by hand (on the headset, in Konsole: ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub)")
+        user, host, port = m.groups()
+        # The ssh this server runs, by its full path: the shell that runs a ProxyCommand may set
+        # up another PATH.
+        ssh = frame_host.which(self.mux_base[0])
+        if not ssh:
+            raise frame_devices.DeviceError(f"Couldn't find {self.mux_base[0]} to reach the jump host with")
+        hop = [ssh, *(a for o in self.HOP_OPTS for a in ("-o", o)), *(["-l", user] if user else []),
+               *(["-p", port] if port else []), host.strip("[]")]  # ssh takes a bare IPv6 address here
+        return ["-F", "none", "-o", "ProxyCommand=" + proxy_command(hop)]
+
+    def _scan_one(self, device, host, route):
+        """-> ((type, blob, fingerprint) or None, what ssh said)."""
+        name = frame_devices.host_key_alias(device["id"])
+        path, opt = frame_devices.scan_file(device["id"])
+        opts = [a for o in [*self.SCAN_OPTS, f"UserKnownHostsFile={opt}"] for a in ("-o", o)]
+        argv = [self.mux_base[0], *route, *opts, *self.host_opts(device, host), "-N", device["alias"]]
+        fd_, errfile = tempfile.mkstemp(prefix="frame-scan-")  # a file, not a pipe (see frame_host.run_ssh)
+        key, proc = None, None
+        try:
+            try:
+                proc = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=fd_,
+                                        **frame_host.DETACHED)
+            finally:
+                os.close(fd_)
+            deadline = time.monotonic() + self.SCAN_TIMEOUT
+            while key is None:
+                ended = proc.poll() is not None
+                key = _scanned_key(path, name)
+                if key or ended or self.stopped or time.monotonic() > deadline:
+                    break
+                time.sleep(0.05)
+        except OSError as e:
+            return None, str(e)
+        finally:
+            if proc and proc.poll() is None:
+                proc.kill()  # it has the key (or gave up): nothing more to do with that device
+            if proc:
+                try:
+                    proc.wait(5)
+                except subprocess.TimeoutExpired:
+                    pass
+            try:
+                with open(errfile, "rb") as fh:
+                    said = fh.read().decode("utf-8", "replace")
+            except OSError:
+                said = ""
+            _unlink(errfile)
+            _unlink(path)
+        if key is None and not said.strip() and proc and proc.returncode is not None and proc.returncode < 0:
+            said = f"Timed out talking to {device['alias']}"
+        return key, said
+
+    def trust_identity(self, device_id, fingerprint):
+        """Pin the key scan_identity read, and only if `fingerprint` is the one it showed (what
+        the person saw and confirmed), replacing whatever was saved; and only while the headset
+        is still reached as it was then (same port and user, the address still one of its own).
+        -> the scan."""
+        with self.cond:
+            scan = self.scans.get(device_id)
+            if not scan or now() - scan["at"] > SCAN_TTL:
+                self.scans.pop(device_id, None)
+                raise frame_devices.DeviceError("Check the headset's key again first: the one shown has expired")
+            if not isinstance(fingerprint, str) or fingerprint != scan["fingerprint"]:
+                raise frame_devices.DeviceError("That isn't the key the headset showed. Check its key again")
+            self.scans.pop(device_id)
+        d = self.reg.get(device_id)  # still a headset here
+        if (d["port"], d["user"]) != (scan["port"], scan["user"]) or \
+                scan["host"] not in [a["host"] for a in d["addresses"]]:
+            raise frame_devices.DeviceError("The headset's address, port or user changed since its key was read. "
+                                            "Check its key again")
+        frame_devices.pin_key(device_id, scan["type"], scan["blob"])
+        self.reg.record_identity(device_id, [scan["fingerprint"]])
+        return scan
+
 
 # ---- the page's API: /api/devices -------------------------------------------------
 
@@ -1158,6 +1519,9 @@ def devices_view(link):
         view = {k: v for k, v in d.items() if k not in ("config_host", "addresses")}
         view["active"] = d["id"] == active["id"]
         view["pinned"] = frame_devices.pinned(d["id"])
+        view["keys"] = [{"type": t, "fingerprint": f} for t, f in frame_devices.pinned_keys(d["id"])]
+        # It had an identity, but the saved key is gone or unusable: nothing is trusted until checked.
+        view["identity_lost"] = not view["pinned"] and frame_devices.identity_established(d)
         view["addresses"] = [dict(a, network_names=[names.get(n, "an unnamed network") for n in a["networks"]])
                              for a in d["addresses"]]
         devices.append(view)
@@ -1190,7 +1554,7 @@ def devices_action(link, body, open_setup, busy=lambda: 0):
     is_active = did == active["id"]
     moves = action == "use" or (action == "retry" and link.alive()) or (is_active and (
         # (a retry while connected would cut the install's connection)
-        action in ("remove", "address-remove", "forget-identity")
+        action in ("remove", "address-remove", "identity-trust")
         or (action == "update" and login_change(reg, did, body))
         or (action == "address-update" and body.get("newHost") not in (None, body.get("host")))))
     if moves and busy():
@@ -1270,11 +1634,25 @@ def devices_action(link, body, open_setup, busy=lambda: 0):
         threading.Thread(target=link.test, args=(did,), daemon=True).start()
         msg = f"Testing {len(d['addresses'])} address{'es' * (len(d['addresses']) != 1)}"
     elif action == "forget-identity":
+        # Dropping the pin used to let the next contact save whatever key answered, unseen.
+        # A new key is now always shown first: identity-scan, then identity-trust.
+        reg.get(did)
+        raise frame_devices.DeviceError("Forgetting a headset's identity is no longer supported: use Check its key "
+                                        "on the Devices tab (identity-scan, then identity-trust), which shows the "
+                                        "new key before saving it")
+    elif action == "identity-scan":
         d = reg.get(did)
-        frame_devices.forget_pin(did)
+        scan = link.scan_identity(did)
+        msg = (f"{d['name']} shows the key already saved for it" if scan["matches"]
+               else f"{d['name']} shows a {scan['type']} key, {scan['fingerprint']}")
+        link.devices_changed()
+        return dict(devices_view(link), message=msg, scan=scan)
+    elif action == "identity-trust":
+        d = reg.get(did)
+        link.trust_identity(did, body.get("fingerprint"))
         if is_active:
-            link.kick("switch")
-        msg = f"Forgot {d['name']}'s SSH identity; the next connection saves the one it shows"
+            link.kick("switch")  # reconnect with the key now saved
+        msg = f"Saved {d['name']}'s new SSH identity; reconnecting"
     elif action == "name-network":
         reg.name_network(body.get("network"), body.get("name"))
         if link.state["network"]:

ui/frame_telemetry.py

@@ -53,12 +53,14 @@ SENT_KEEP = 200
 OUTBOX_MAX = 2000  # events kept while offline; the oldest go first
 FLUSH_EVERY = 60
 REPEAT_WINDOW = 600  # the same diagnostic error is sent at most once in this many seconds
-# Not faults in Frame Control: the headset asleep, away or not set up yet. The status poll
+# Not faults in Frame Control: the headset asleep, away or not set up yet, or ssh and the headset
+# not trusting each other (a host key that changed, a key not authorised): the page says what to do
+# about each (Devices tab, Set Up Connection). The status poll
 # meets these every few seconds, so each is sent at most once per session, whatever the wording,
 # but only for an error marked where ssh ran as ssh failing to reach the headset
 # (frame_host.link_failure). The message alone isn't evidence: a download's "Connection reset by
 # peer", or a file name with ssh's words in it, keeps the usual window.
-EXPECTED_CATEGORIES = ('frame_unreachable', 'frame_not_set_up')
+EXPECTED_CATEGORIES = ('frame_unreachable', 'frame_not_set_up', 'frame_auth')
 DEFAULT_HOST = 'https://us.i.posthog.com'
 
 LEVELS = ('usage', 'compat', 'diagnostics')
@@ -268,7 +270,8 @@ CATEGORIES = [
     ('cant_run_on_frame', re.compile(r"can't run on the Frame")),
     ('steam_shortcut', re.compile(r'(?i)steam did not return a shortcut|shortcut list|no Steam shortcut')),
     ('frame_not_set_up', re.compile(r'(?i)Could not resolve hostname|no "?frame"? (?:SSH )?alias')),
-    ('frame_auth', re.compile(r'(?i)Permission denied|Host key verification failed')),
+    ('frame_auth', re.compile(r'(?i)Permission denied|Host key verification failed|'
+                              r'REMOTE HOST IDENTIFICATION HAS CHANGED|host key is known for')),
     # Anything ssh says about its own connection: "ssh: connect to host … port 22: <reason>" (Windows
     # says "Unknown error"), a dropped shared connection (mux_client_…, client_loop), and Windows'
     # "banner exchange: Connection to UNKNOWN port -1" (its ssh can't name a peer whose connect

ui/index.html

@@ -700,6 +700,7 @@
     <div class="grow"><div class="t" id="offMsg">Can't reach the Frame</div>
       <div class="s" id="offDetail">Retrying every few seconds. Everything fills in when it answers.</div></div>
     <button class="small desk-only" id="offDetails" hidden>Details</button>
+    <button class="small desk-only" id="offTrust" hidden>Check the new key…</button>
     <button class="small" id="offSetup" hidden>Set Up Connection…</button>
     <button class="action small" id="offRetry">Retry now</button>
   </div>
@@ -1358,6 +1359,18 @@
       <button type="submit" class="small danger" id="rmGo">Remove</button></div>
   </form>
 </dialog>
+<dialog id="trustDlg" class="dlg" aria-labelledby="trustTitle">
+  <form method="dialog" id="trustForm">
+    <h2 id="trustTitle">Check the headset's new key</h2>
+    <div class="sub" id="trustText"></div>
+    <dl class="facts" id="trustKeys"></dl>
+    <div class="sub" id="trustHow" hidden>To check it, on the headset open the desktop from the SteamVR dashboard, start
+      Konsole and run <code id="trustCmd"></code>. The <code>SHA256:</code> value it prints must match the key above.</div>
+    <div class="row rep-actions"><span class="sub" id="trustMsg" role="status"></span><span class="spacer"></span>
+      <button type="button" class="small" id="trustCancel">Cancel</button>
+      <button type="submit" class="small action" id="trustGo" disabled>This is my headset: trust its new key</button></div>
+  </form>
+</dialog>
 <div class="toast" id="toast"></div>
 
 <script>
@@ -4542,6 +4555,10 @@ function renderBanner() {
   const s = link.s;
   if (!link.live || !s) return;
   $("offDetails").hidden = false;
+  // A changed host key: never accepted by itself. The person checks the new key and decides.
+  const hk = s.phase === "failed" && s.error && s.error.hostkey && s.device ? s.device.id : null;
+  $("offTrust").hidden = !hk;
+  if (hk) $("offTrust").onclick = () => openTrust(hk);
   if (s.phase === "failed" && s.error) {
     $("offline").hidden = false;
     $("offMsg").textContent = s.error.message;
@@ -4642,6 +4659,9 @@ function onConnection(s) {
     loadDevices();
   }
   link.phase = s.phase; link.device = devId;
+  // What the Devices tab offers for the failure (check a new key, pair again) follows it.
+  const problem = s.phase === "failed" && s.error ? (s.error.hostkey ? "hostkey" : s.error.auth || "") : "";
+  if (problem !== link.problem) { link.problem = problem; renderDevices(); }
   if (page === "devices") renderTests();
 }
 
@@ -4801,8 +4821,15 @@ function renderDetail() {
       <button class="small" data-mv="-1" title="Try earlier"${i ? "" : " disabled"}>↑</button>
       <button class="small" data-mv="1" title="Try later"${i < d.addresses.length - 1 ? "" : " disabled"}>↓</button>
       <button class="small" data-ed>Edit</button><button class="small danger" data-rm>Remove</button></div>`).join("");
+  const err = d.active && link.s && link.s.phase === "failed" ? link.s.error : null;
+  const problem = err && err.hostkey ? `<div class="item" role="alert"><div class="grow"><div class="t">${err.hostkey.state === "missing"
+        ? "Its saved SSH identity is missing" : "Its SSH identity has changed"}</div>
+        <div class="s">${esc(err.message)}</div></div><button class="small action" id="dvTrustNow">Check the new key…</button></div>`
+    : err && err.auth === "denied" ? `<div class="item" role="alert"><div class="grow"><div class="t">The headset didn't accept this computer's key</div>
+        <div class="s">Set Up Connection pairs this computer with it again. On the headset, Developer Mode must be on.</div></div>
+        <button class="small action" id="dvPair">Set Up Connection…</button></div>` : "";
   el.innerHTML = `<div class="shelf-head"><h2>${esc(d.name)}</h2>${d.active ? '<span class="count">In use</span>' : ""}<span class="spacer"></span>
-      ${d.active ? "" : `<button class="small action" id="dvUse">Use this headset</button>`}</div>
+      ${d.active ? "" : `<button class="small action" id="dvUse">Use this headset</button>`}</div>${problem}
     <form class="dev-form" id="dvForm">
       <label>Name<input type="text" id="dvName" value="${esc(d.name)}" maxlength="60"></label>
       <label>SSH alias<input type="text" value="${esc(d.alias)}" readonly title="Terminal: ssh ${esc(d.alias)}"></label>
@@ -4823,10 +4850,12 @@ function renderDetail() {
       <button class="small" id="dvTs">Find on Tailscale</button>
       <button class="small" id="dvMdns">Find on this network</button>
       <span class="spacer"></span>
-      <button class="small" id="dvForget" title="After reinstalling SteamOS the headset shows a new identity">Forget identity</button>
+      <button class="small" id="dvTrust" title="After SteamOS is reinstalled or reset the headset shows a new identity">Check its key…</button>
       <button class="small danger" id="dvRemove">Remove…</button></div>
     <div class="found" id="dvFound"></div>
-    <div class="hint">Identity: ${d.pinned ? "saved. A different device answering at one of these addresses is refused."
+    <div class="hint">Identity: ${d.pinned ? `saved${(d.keys || []).length ? ` (${esc(d.keys.map(k => k.fingerprint).join(", "))})` : ""}.
+      A different device answering at one of these addresses is refused.`
+      : d.identity_lost ? "the saved key is missing or can't be read, so nothing is trusted until you check its key."
       : "not saved yet; the next connection saves it."} Kind "auto" is picked from the address:
       <code>.local</code> names are mDNS, <code>100.x</code> and <code>.ts.net</code> are Tailscale.</div>`;
   $("adKind").insertAdjacentHTML("afterbegin", `<option value="" selected>Kind: auto</option>`);
@@ -4859,7 +4888,9 @@ function renderDetail() {
   $("dvTest").onclick = e => devAction(`Test ${d.name}'s addresses`, { action: "test", id: d.id }, e.currentTarget);
   $("dvTs").onclick = e => findOn("tailscale", d, e.currentTarget);
   $("dvMdns").onclick = e => findOn("mdns", d, e.currentTarget);
-  $("dvForget").onclick = e => devAction(`Forget ${d.name}'s identity`, { action: "forget-identity", id: d.id }, e.currentTarget);
+  $("dvTrust").onclick = () => openTrust(d.id);
+  if ($("dvTrustNow")) $("dvTrustNow").onclick = () => openTrust(d.id);
+  if ($("dvPair")) $("dvPair").onclick = () => setUpHeadset(d.alias);
   $("dvRemove").onclick = () => askRemove(d);
   renderFound(d);
   renderTests();
@@ -4912,6 +4943,48 @@ function renderFound(d) {
   });
 }
 
+// Trust a new key: read the key the headset shows now (nothing is saved), show it next to the
+// saved one, and save it only when the person says it's their headset. The server pins exactly
+// the key it showed here (frame_link.Link.trust_identity), never whatever answers next.
+const HOST_KEY_FILE = { "ssh-ed25519": "ed25519", "ssh-rsa": "rsa", "ssh-dss": "dsa" };
+async function openTrust(id) {
+  const d = dv.list.find(x => x.id === id);
+  const name = d ? d.name : "the headset";
+  $("trustTitle").textContent = `Check ${name}'s key`;
+  $("trustText").textContent = "Asking the headset for its key…";
+  $("trustKeys").innerHTML = ""; $("trustMsg").textContent = "";
+  $("trustHow").hidden = true; $("trustGo").disabled = true;
+  $("trustCancel").onclick = () => $("trustDlg").close();
+  $("trustForm").onsubmit = e => e.preventDefault();
+  if (!$("trustDlg").open) $("trustDlg").showModal();
+  let res;
+  try { res = await api("/api/devices", { action: "identity-scan", id }); }
+  catch (e) { $("trustText").textContent = e.message; return; }
+  const scan = res.scan;
+  const saved = (scan.saved || []).map(k => `${k.type} ${k.fingerprint}`).join("\n")
+    || ((scan.recorded || []).length ? `${scan.recorded.join("\n")}\n(recorded; the saved file is missing or unreadable)` : "None");
+  $("trustKeys").innerHTML = [["Saved key", saved], ["It shows now", `${scan.type} ${scan.fingerprint}`], ["At", scan.host]]
+    .map(([k, v]) => `<dt>${esc(k)}</dt><dd style="white-space:pre-wrap; word-break:break-all"><code>${esc(v)}</code></dd>`).join("");
+  if (scan.matches) {
+    $("trustText").textContent = `${name} shows the key already saved for it, so there's nothing to change. If it still won't connect, try Test now.`;
+    return;
+  }
+  $("trustText").textContent = scan.recorded_match
+    ? `${name} shows the key recorded for it before, but the saved copy is missing or can't be read. Trust it to save it again.`
+    : (scan.saved || []).length || (scan.recorded || []).length
+    ? `${name} now shows a different SSH key from the one saved. That's expected after SteamOS was reinstalled or reset, or if this is a new headset at the same address. If none of that happened, don't trust it: another device may be answering in its place.`
+    : `No key is saved for ${name} yet. Trust this one if it's your headset.`;
+  $("trustCmd").textContent = `ssh-keygen -lf /etc/ssh/ssh_host_${HOST_KEY_FILE[scan.type] || "ecdsa"}_key.pub`;
+  $("trustHow").hidden = false;
+  $("trustGo").disabled = false;
+  $("trustForm").onsubmit = async e => {
+    e.preventDefault();
+    const ok = await devAction(`Trust ${name}'s new key`, { action: "identity-trust", id, fingerprint: scan.fingerprint }, $("trustGo"));
+    if (ok) $("trustDlg").close();
+    else $("trustMsg").textContent = "Not saved. Close this and check the key again.";
+  };
+}
+
 function askRemove(d) {
   $("rmDevTitle").textContent = `Remove ${d.name}?`;
   $("rmDevText").textContent = `Frame Control forgets this headset, its ${d.addresses.length} address${d.addresses.length === 1 ? "" : "es"} and its saved identity. Its SSH keys stay.`;

ui/server.py

@@ -211,9 +211,14 @@ UNREACHABLE = [
     (re.compile(r"port \d+: Connection refused"),
      "The Frame refused the connection. Check Developer Mode is still on."),
     (re.compile(r"Permission denied \(publickey"),
-     "The Frame didn't accept this computer's SSH key. Run Set Up Connection again."),
-    (re.compile(r"Host key verification failed"),
-     "The Frame's SSH identity changed (after a reinstall, or a different device). Run Set Up Connection again."),
+     "The Frame didn't accept this computer's SSH key. Set Up Connection pairs this computer with it "
+     "again (Developer Mode must be on)."),
+    # Before "Host key verification failed", which ssh prints after each of these.
+    (re.compile(r"No \S+ host key is known for "),
+     "The Frame's SSH identity isn't saved yet. Reconnecting saves it; or check and trust it on the Devices tab."),
+    (re.compile(r"REMOTE HOST IDENTIFICATION HAS CHANGED|Host key verification failed"),
+     "The Frame's SSH identity has changed: after SteamOS was reinstalled or reset, or a different device "
+     "answering at its address. Check the new key on the Devices tab before trusting it."),
     (re.compile(r"kex_exchange_identification|Connection closed by .* port \d+|Connection reset by .* port \d+"),
      "The connection to the Frame dropped. Try again."),
 ]