Source logins: refuse addresses with a backslash and a login; decode escapes in one pass

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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/apk-repos.md

@@ -58,8 +58,61 @@ form shows the pinned one ("Trusted on first use: …") so you can compare it.
 Re-adding an existing URL preserves its pin; changing it requires deliberately
 removing and re-adding it.
 
+### Repositories behind a login
+
+Some self-hosted repositories sit behind HTTP basic auth. As in F-Droid's own
+client, the login can be part of the address
+(`https://user:pass@example.org/fdroid/repo`), or given separately: in the
+store's **Add a source** form under "This repository needs a login", in the
+server API (`username` and `password` beside `url`), or on the command line,
+where the password is read from standard input so it stays out of shell history
+and `ps`:
+
+```sh
+printf '%s\n' "$PASSWORD" | python3 ui/apk_sources/fdroid.py add 'https://example.org/fdroid/repo' --username me --password-stdin
+```
+
+How the login is kept and used:
+
+- **Kept out of the address.** The `user:pass@` part is removed before anything
+  is saved, so the saved repository URL, its id, the source list, job results
+  and the page never contain it. The store only says "Uses a login".
+- **Stored separately.** Logins live in
+  `frame_host.data_dir('source-credentials.json')`
+  (`~/Library/Application Support/Frame Control/source-credentials.json` on
+  macOS), written with mode 0600 and keyed by source id. They are stored in
+  plain text there, protected by file permissions only (on Windows,
+  the folder's normal ACLs). Removing the repository deletes its login, and
+  the file goes when the last login is removed. Each change holds a lock
+  file (`source-credentials.lock`), so the app and the command line can't undo
+  each other's saves or removals.
+- **HTTPS only.** An `http://` address with a login is refused before any request.
+- **Inside the repository only.** The `Authorization` header is decided
+  separately for every request, redirects included: index, `entry.jar`,
+  APKs and the store's artwork proxy. It is sent only when the request's
+  scheme, host and port match the repository's and its path is inside the
+  repository's directory by whole path segments. A redirect to another host
+  or port, or to another path on the same server, gets no login, and that
+  request's own `Authorization` header is removed. Paths with `.` or `..`
+  segments (plain or percent-encoded), encoded `/` or `\`, double encoding
+  or `;` get no login, since a server could normalise them to a path
+  outside the repository. Index file names like that are refused outright.
+- **Never echoed.** Logins are not written to logs, error messages, API
+  responses or the cache. The CLI prints only the saved source.
+- **Checked before saving.** As with any repository, the whole index must
+  download and verify before anything is saved. HTTP 401 (or 403 when a login
+  was sent) stops the add with "check the username and password" and saves
+  nothing: no repository, no login, no rollback state. A 403 with a login is
+  not treated as rate limiting.
+- **Same verification.** The login changes only who may download. Signature,
+  fingerprint pin or TOFU, index hash and rollback checks apply exactly as for
+  any other repository.
+
+Re-adding the same address with a new login replaces the saved one. Re-adding
+it without a login keeps the saved one.
+
 The API for the search/server integration is in `ui/apk_sources/fdroid.py`:
-`add_repo(url, fingerprint=None, name=None)`, `remove_repo(source_id)`,
+`add_repo(url, fingerprint=None, name=None, username=None, password=None)`, `remove_repo(source_id)`,
 `set_enabled(source_id, enabled)`, and `user_repos()`. The module also exposes
 `sources`, `search`, `details`, and `download` from the shared source contract.
 This change supplies the CLI and API; the integrated source-management UI is

docs/apk-sources.md

@@ -45,10 +45,23 @@ search. Results outside the curated list stay page-only, even if a repository
 claims an open-source license. This prevents an arbitrary tagged mirror from
 becoming a trusted downloader. Extend the curated JSON after provenance review.
 
-Set optional `FRAME_GITHUB_TOKEN` in the process environment for a higher API
-quota. Tokens are sent only to `api.github.com`, never written to the cache,
-never sent to asset hosts, and removed on redirects. The adapter does not
-read `gh` credentials automatically. Metadata is cached for one hour under
+For a higher API quota, add a GitHub access token in the store's source
+settings (the GitHub row's token field) or through the server API
+(`POST /api/sources` with `action: "set-token"` or `"clear-token"` and
+`source: "github"`). `FRAME_GITHUB_TOKEN` in the process environment still
+works and wins over a saved token. A saved token is kept with repository
+logins in `source-credentials.json` (mode 0600, see
+[APK repositories](apk-repos.md#repositories-behind-a-login)). The source list
+reports only whether a token is set, where it came from and whether GitHub
+rejected it, never its value. Tokens are sent as `Authorization: Bearer` only to
+`api.github.com`, never written to the cache, never sent to asset hosts (public
+release downloads don't need them), and removed on redirects. If GitHub answers
+401 to the token, the request is retried once without it, as Obtainium does.
+The rejection is logged without the token and shown in source settings. The
+token is not tried again until it is saved again or the app restarts. Rate-limit
+backoff is kept per login (anonymous, or each token), so saving a token after
+anonymous requests were limited lets the next request go out with it. The
+adapter does not read `gh` credentials automatically. Metadata is cached for one hour under
 `frame_host.cache_dir('apk-sources', 'publisher')`. A cold details request
 fetches at most ten releases. Rate-limit errors are surfaced without retry
 loops. Asset IDs and release tags are not Android version codes: metadata

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
 

docs/streaming.md

@@ -261,6 +261,16 @@ How (**verified 2026-09-29**, SteamOS 0.4.1, build 20260925.6191901):
   -window_id <window> -i :<display>` works on gamescope's redirected windows,
   while grabbing the root gives black. It streams as H.264 like the headset view
   (about 30 fps at 720p).
+- **When no panel is found**, `frame_touch.py panels` says why (`"why"`), and the
+  Desktop view shows that instead of a bare "no app panel": X tools missing,
+  gamescope listing windows that aren't X11 windows on any display, the desktop
+  running but not listed, or nothing open. It reads gamescope's properties from
+  `:0` first (one `xprop` call) and probes the other displays, 2 s each, only if `:0`
+  has none. A whole run stops after 12 s, inside the server's 20 s, and says the headset
+  was too slow. Added for
+  [#73](https://github.com/saphid/frame-control/issues/73) (SteamOS 0.4.5, desktop
+  open but not found). The likely cause there was the page itself: a second
+  `loadPanels` (the panel switcher's) replaced the Desktop view's, fixed in #84.
 - Tested from the iPhone app (Simulator): a tap on the Desktop view focused a
   text box in the panel and the text field typed into it; a trackpad move went
   exactly (+40, +25).

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/page/desk_panels.mjs

@@ -0,0 +1,82 @@
+// The live view's Desktop panel list (#73), run in node against the real functions from
+// ui/index.html with a fake server: the headset's reason when it lists no panels, a failed
+// listing shown (first time too) rather than "Finding…", and recovery after it.
+import { readFileSync } from "fs";
+const src = readFileSync(new URL("../../ui/index.html", import.meta.url), "utf8");
+const grab = name => {
+  const m = src.match(new RegExp(`(?:\\nconst ${name} = [^\\n]*\\n)|((?:async )?function ${name}\\([\\s\\S]*?\\n}\\n)`));
+  if (!m) throw new Error("not found: " + name);
+  return m[0];
+};
+// The page's scripts share one global scope, so a second loadPanels would silently replace
+// the Desktop view's (what broke #73 until #84): the one grabbed here must be the only one.
+for (const name of ["loadPanels", "noPanelWhy", "deskPanel", "ctrlTarget"]) {
+  const count = (src.match(new RegExp(`\\n(?:async )?function ${name}\\(`, "g")) || []).length;
+  if (count !== 1) { console.log(`FAIL ${name} is defined ${count} times`); process.exit(1); }
+}
+const code = ["desk", "panelKey", "noPanelWhy", "deskPanel", "loadPanels", "ctrlTarget"].map(grab).join("");
+const fail = msg => { console.log("FAIL " + msg); process.exit(1); };
+
+function page(answer) {
+  const calls = [], logged = [];
+  const env = {
+    view: "flat", live: false, ctrlMode: () => "abs", renderPanelPick: () => {}, toggleLive: () => {}, ctrlShow: () => {},
+    log: (m) => logged.push(m),
+    api: async (path, body) => { calls.push([path, body]); return answer(); },
+  };
+  const fns = new Function(...Object.keys(env), `${code}
+    return { desk, noPanelWhy, loadPanels, ctrlTarget };`)(...Object.values(env));
+  return { ...fns, calls, logged };
+}
+
+// The Desktop view asks for the headset's panels with a plain GET (no switcher action).
+{
+  const p = page(() => ({ panels: [], focus: null, why: "The headset's desktop is running, but SteamOS isn't listing it as a panel." }));
+  if (p.ctrlTarget().why !== "Finding the headset's panels…") fail("before loading: " + p.ctrlTarget().why);
+  await p.loadPanels();
+  if (!(p.calls.length === 1 && p.calls[0][0] === "/api/panels" && p.calls[0][1] === undefined))
+    fail("request " + JSON.stringify(p.calls));
+  if (!p.desk.loaded) fail("not marked loaded");
+  if (!p.noPanelWhy().startsWith("The headset's desktop is running")) fail("why for capture: " + p.noPanelWhy());
+  if (!p.ctrlTarget().why.startsWith("The headset's desktop is running")) fail("why for control: " + p.ctrlTarget().why);
+}
+// An older headset script without "why": the generic advice.
+{
+  const p = page(() => ({ panels: [], focus: null }));
+  await p.loadPanels();
+  if (!p.noPanelWhy().startsWith("No app panel is open in the headset. Open the desktop")) fail("generic: " + p.noPanelWhy());
+}
+// The first listing fails: Control and Capture say so (logged once), then recover.
+{
+  let broken = true;
+  const p = page(() => {
+    if (broken) throw new Error("Timed out talking to frame");
+    return { panels: [{ window: 7, display: ":0", name: "Desktop" }], focus: 7, focus_display: ":0" };
+  });
+  await p.loadPanels(); await p.loadPanels();
+  const why = p.ctrlTarget().why;
+  if (why !== "Couldn't list the headset's panels: Timed out talking to frame") fail("initial failure in Control: " + why);
+  if (p.noPanelWhy() !== why) fail("initial failure in Capture: " + p.noPanelWhy());
+  if (p.logged.length !== 1) fail("logged " + p.logged.length + " times");
+  broken = false;
+  await p.loadPanels();
+  if (p.desk.error) fail("error kept after recovery");
+  const t = p.ctrlTarget();
+  if (!t.why || t.why.includes("Couldn't list")) fail("after recovery: " + JSON.stringify(t));
+  if (p.desk.panels.length !== 1) fail("panels after recovery");
+}
+// Waiting for the Frame: shown in Control, not logged every poll. An answer about the
+// previous headset (stale) is ignored.
+{
+  let err;
+  const p = page(() => { throw err; });
+  err = Object.assign(new Error("Can't reach the Frame"), { offline: true });
+  await p.loadPanels(); await p.loadPanels();
+  if (p.logged.length) fail("offline logged: " + p.logged);
+  if (p.ctrlTarget().why !== "Couldn't list the headset's panels: Can't reach the Frame") fail("offline: " + p.ctrlTarget().why);
+  const q = page(() => { throw Object.assign(new Error("old headset"), { offline: true, stale: true }); });
+  await q.loadPanels();
+  if (q.logged.length || q.desk.error) fail("stale answer used: " + q.desk.error);
+  if (q.ctrlTarget().why !== "Finding the headset's panels…") fail("stale: " + q.ctrlTarget().why);
+}
+console.log("ok");

tests/page/source_login.mjs

@@ -0,0 +1,79 @@
+// The store's "Add a source" login handling and GitHub token field, run in node against
+// the real functions from ui/index.html. A login typed into the address must leave the
+// address field (into the login fields) before anything is sent, and stay out of it.
+import { readFileSync } from "fs";
+const src = readFileSync(new URL("../../ui/index.html", import.meta.url), "utf8");
+const lines = src.split("\n");
+const fail = msg => { console.log("FAIL " + msg); process.exit(1); };
+
+function grab(name) {
+  const start = lines.findIndex(l => new RegExp(`^(?:(?:async )?function ${name}\\(|const ${name} = )`).test(l));
+  if (start < 0) fail(`no top-level ${name} in index.html`);
+  let depth = 0;
+  for (let i = start; i < lines.length; i++) {
+    for (const c of lines[i]) depth += c === "{" ? 1 : c === "}" ? -1 : 0;
+    if (depth <= 0) return lines.slice(start, i + 1).join("\n") + "\n";
+  }
+  fail(`unbalanced ${name}`);
+}
+const { moveLogin, sourceToken } = new Function(
+  "const esc = s => String(s).replace(/[&<>\"']/g, c => `&#${c.charCodeAt(0)};`); const storeName = s => s;\n" +
+  grab("decodeLenient") + grab("moveLogin") + grab("sourceToken") + "return { moveLogin, sourceToken };")();
+
+function form(url, username = "", password = "") {
+  const details = { open: false };
+  const field = value => ({ value, closest: () => details });
+  return { details, elements: { url: field(url), username: field(username), password: field(password) } };
+}
+
+let f = form("https://alice:s3cr%40t@repo.example/fdroid/repo");
+moveLogin(f);
+if (f.elements.url.value !== "https://repo.example/fdroid/repo") fail("address still holds the login: " + f.elements.url.value);
+if (f.elements.username.value !== "alice" || f.elements.password.value !== "s3cr@t") fail("login not moved into its fields");
+if (!f.details.open) fail("login fields not shown");
+
+f = form("https://alice:fromurl@repo.example/repo", "typed", "typedpw");
+moveLogin(f);
+if (f.elements.url.value.includes("fromurl") || f.elements.username.value !== "typed" || f.elements.password.value !== "typedpw")
+  fail("typed login replaced, or address kept the login");
+
+// Undecodable logins and unusual spellings the URL parser accepts still lose their login; undecodable bytes stay as typed.
+for (const [url, address, user, pass] of [
+  ["https://alice:100%done@repo.example/private/repo", "https://repo.example/private/repo", "alice", "100%done"],
+  ["https:\t//alice:s3cret@repo.example/repo", "https://repo.example/repo", "alice", "s3cret"],
+  ["ht\ntps://alice:s3cret@repo.example/repo", "https://repo.example/repo", "alice", "s3cret"],
+  ["https:alice:s3cret@repo.example/repo", "https://repo.example/repo", "alice", "s3cret"],
+  ["\u0000https://alice:s3cret@repo.example/repo", "https://repo.example/repo", "alice", "s3cret"],
+  ["https://alice:%FF@repo.example/private/repo", "https://repo.example/private/repo", "alice", "%FF"],
+  ["https://al%20ice:p%40ss:word@repo.example/r", "https://repo.example/r", "al ice", "p@ss:word"],
+  ["https://alice:p%FF\u00e4@repo.example/repo", "https://repo.example/repo", "alice", "p%FF\u00e4"],
+  ["fdroidrepos://alice:pw@repo.example/repo?fingerprint=AB", "fdroidrepos://repo.example/repo?fingerprint=AB", "alice", "pw"],
+  ["https://alice@repo.example/repo", "https://repo.example/repo", "alice", ""],
+]) {
+  f = form(url);
+  moveLogin(f);
+  if (f.elements.url.value !== address) fail(`address kept the login: ${url} -> ${f.elements.url.value}`);
+  if (f.elements.username.value !== user || f.elements.password.value !== pass)
+    fail(`login for ${url}: ${f.elements.username.value} / ${f.elements.password.value}`);
+}
+
+// An address the parser rejects that may hold a login is refused and left as typed, never sent.
+for (const url of ["https://alice\\name:pw@repo.example/repo", "https:\\\\alice:s3cret@repo.example/repo", "fdroidrepos://alice:p\\w@repo.example/repo", "https://alice:s3cret@repo example/repo", "https:alice:s3cret@repo example/repo", "https://repo.example:bad\\a@b"]) {
+  f = form(url);
+  if (moveLogin(f) !== false) fail("accepted an invalid address with a login: " + url);
+  if (f.elements.username.value || f.elements.password.value) fail("guessed a login from an invalid address: " + url);
+}
+
+for (const url of ["https://repo.example/repo", "not a url", "https://repo.example/a@b", "fdroidrepos:///a@b", "fdroidrepos:/a@b", "https://repo.example/r?x=a@b#c@d"]) {
+  f = form(url);
+  moveLogin(f);
+  if (f.elements.url.value !== url || f.elements.username.value) fail("changed an address without a login: " + url);
+}
+
+const html = sourceToken({ id: "github", name: "GitHub releases", token: { set: true, from: "saved", rejected: true } });
+if (!html.includes("rejected") || !html.includes('data-clear-token="github"')) fail("token row: " + html);
+if (sourceToken({ id: "fdroid", name: "F-Droid" }) !== "") fail("token field on a source without one");
+const slow = "https://a:" + "%FF".repeat(4000) + "@repo.example/r", t0 = Date.now();
+f = form(slow); moveLogin(f);
+if (Date.now() - t0 > 200 || f.elements.password.value !== "%FF".repeat(4000)) fail("decoding malformed escapes is slow or wrong");
+console.log("ok");

tests/page/store_detail.mjs

@@ -0,0 +1,93 @@
+// The Android store's app page, run in node against the real functions from ui/index.html.
+// Issue #68: clicking an app opened an empty dialog (a bare horizontal line), because the
+// Devices page's renderDetail replaced the store's. The store's is renderAppDetail now.
+// (tests/test_server.py PageScripts checks no top-level name is declared twice.)
+import { readFileSync } from "fs";
+const src = readFileSync(new URL("../../ui/index.html", import.meta.url), "utf8");
+const lines = src.split("\n");
+const fail = msg => { console.log("FAIL " + msg); process.exit(1); };
+
+// A top-level declaration, from its first line to where its braces balance.
+function grab(name) {
+  const start = lines.findIndex(l => new RegExp(`^(?:(?:async )?function ${name}\\(|const ${name} = )`).test(l));
+  if (start < 0) fail(`no top-level ${name} in index.html`);
+  let depth = 0;
+  for (let i = start; i < lines.length; i++) {
+    for (const c of lines[i]) depth += c === "{" ? 1 : c === "}" ? -1 : 0;
+    if (depth <= 0) return lines.slice(start, i + 1).join("\n") + "\n";
+  }
+  fail(`unbalanced ${name}`);
+}
+const names = ["esc", "storeName", "storeHue", "storeInitial", "storeText", "storeOpenSource", "sourcePage", "storeImage",
+               "storeArt", "storeIcon", "sourceMark", "currentOffer", "detailBadges", "renderAppDetail", "renderAppDetailBody",
+               "updateInstall", "openStoreApp", "loadOfferDetails"];
+const code = names.map(grab).join("");
+
+// A fake page: elements by id. The install panel's elements exist only while
+// #detailBody holds the real app page, as in the browser.
+function page(api = async () => ({})) {
+  const els = { appDetail: { open: false, scrollTop: 0, showModal() { this.open = true; } } };
+  const inPage = id => ["detailInstall", "detailInstallNote", "detailProgress", "detailOffers"].includes(id);
+  const $ = id => inPage(id) && !(els.detailBody?.innerHTML || "").includes(`id="${id}"`) ? null
+    : (els[id] ||= { innerHTML: "", textContent: "", hidden: false, value: 0, removeAttribute() {}, querySelectorAll: () => [] });
+  const sourceState = { apps: [], detail: null, detailRequest: 0, jobs: new Map() };
+  const document = { activeElement: null, querySelector: () => null };
+  const console = { error() {} };
+  const calls = [];
+  const fns = new Function("$", "sourceState", "document", "console", "api", code + "; return { renderAppDetail, openStoreApp };")(
+    $, sourceState, document, console, (...a) => { calls.push(a); return api(...a); });
+  return { els, sourceState, calls, ...fns };
+}
+
+const offer = { source: "fdroid", id: "org.example.brush", source_name: "F-Droid", name: "Open Brush", summary: "Paint in VR",
+                description: "Paint in 3D.", version: "2.1", version_code: 21, package: "org.example.brush",
+                artwork: { icon: "/source-image/abc", screenshots: [] }, downloadable: true, verdict: { tone: "works", label: "Works" } };
+
+// An app from the catalogue: its page renders, with its name, install and sources.
+{
+  const p = page();
+  p.sourceState.detail = { app: { name: "Open Brush", offers: [{ ...offer }] }, choice: 0 };
+  if (p.renderAppDetail() !== true) fail("a good listing reported a render failure");
+  const html = p.els.detailBody.innerHTML;
+  if (!html.includes('id="detailTitle">Open Brush<')) fail("app page has no title: " + html.slice(0, 200));
+  if (!html.includes('id="detailInstall"') || !html.includes("Get it from")) fail("app page has no install panel");
+  if (html.includes('role="alert"')) fail("a good listing showed the error state");
+}
+
+// A listing that can't render (no offer to show) gets a clear message, not an empty dialog.
+{
+  const p = page();
+  p.sourceState.detail = { app: { name: "Broken <App>", offers: [] }, choice: 0 };
+  if (p.renderAppDetail() !== false) fail("a broken listing reported success");
+  const html = p.els.detailBody.innerHTML;
+  if (!html.includes('role="alert"') || !html.includes("We couldn't show Broken &lt;App&gt;")) fail("no error state: " + html);
+  if (!html.includes('data-close="appDetail"')) fail("error state has no way to close");
+}
+
+// Opening a listing that can't render: the error stays up, no details request, nothing thrown.
+{
+  const p = page(async () => { throw new Error("offline"); });
+  p.sourceState.apps = [{ name: "Broken", offers: [] }];
+  await p.openStoreApp(0);
+  if (!p.els.appDetail.open || !p.els.detailBody.innerHTML.includes('role="alert"')) fail("broken listing didn't open with the error");
+  if (p.calls.length) fail("asked for details of a listing that couldn't render");
+}
+
+// Opening a good listing: the dialog opens with its page, and fresher details fill it in.
+{
+  const p = page(async () => ({ description: "Fresh description." }));
+  p.sourceState.apps = [{ name: "Open Brush", offers: [{ ...offer }] }];
+  await p.openStoreApp(0);
+  if (!p.els.appDetail.open || !p.els.detailBody.innerHTML.includes("Fresh description.")) fail("details didn't fill in the page");
+  if (!String(p.calls[0]?.[0]).startsWith("/api/sources/details?")) fail("no details request: " + JSON.stringify(p.calls));
+}
+
+// A good listing whose details request fails keeps its page, with a note.
+{
+  const p = page(async () => { throw new Error("offline"); });
+  p.sourceState.apps = [{ name: "Open Brush", offers: [{ ...offer }] }];
+  await p.openStoreApp(0);
+  if (!p.els.detailBody.innerHTML.includes('id="detailTitle">Open Brush<')) fail("page lost after a failed details request");
+  if (!p.els.detailInstallNote.textContent.includes("couldn't be refreshed")) fail("no note after a failed details request");
+}
+console.log("ok");

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
@@ -324,7 +325,7 @@ class EndpointTests(SettingsTest):
             mod.add_repo.side_effect = SourceError('repository fingerprint mismatch')
             job = self.wait(self.request('POST', '/api/sources', {'action': 'add', 'url': link})[1]['job'])
             self.assertEqual(job['error'], 'repository fingerprint mismatch')  # no "SourceError:" prefix
-            for url in ('http://repo.example/repo', 'fdroidrepo://repo.example/repo', 'https://u@repo.example/'):
+            for url in ('http://repo.example/repo', 'fdroidrepo://repo.example/repo', 'https://repo.example:x/'):
                 self.assertEqual(self.request('POST', '/api/sources', {'action': 'add', 'url': url})[0], 400)
             self.assertEqual(self.request('POST', '/api/sources', {'action': 'remove', 'source': 'fdroid'})[0], 200)
             mod.remove_repo.assert_called_once_with(source_id='fdroid')

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
@@ -80,7 +81,7 @@ class Repositories(unittest.TestCase):
             state.clear()
             self.addCleanup(state.clear)
 
-    def fetch(self, url, path, maximum):
+    def fetch(self, url, path, maximum, auth=None):
         name = url.rsplit('/', 1)[-1]
         if self.v1 and name == 'entry.jar':
             raise urllib.error.HTTPError(url, 404, 'missing', None, None)
@@ -310,7 +311,7 @@ class Repositories(unittest.TestCase):
         source = self.add()
         other = dict(source, id='other-repo')
         started, release = threading.Event(), threading.Event()
-        def fetch(url, path, maximum):
+        def fetch(url, path, maximum, auth=None):
             if threading.current_thread().name == 'slow':
                 started.set()
                 release.wait(5)
@@ -354,7 +355,7 @@ class Repositories(unittest.TestCase):
         self.files['entry.jar'], _ = entry_jar(50)
         source = fdroid.add_repo(URL)
         jars = {'older': entry_jar(100)[0], 'newer': entry_jar(200)[0]}
-        def fetch(url, path, maximum):
+        def fetch(url, path, maximum, auth=None):
             name = threading.current_thread().name
             if name in jars and url.endswith('entry.jar'):
                 Path(path).write_bytes(jars[name])
@@ -400,7 +401,7 @@ class Repositories(unittest.TestCase):
         source = dict(id='overlap', name='Overlap', url=URL, fingerprint=None)
         self.v1 = True
         inner = []
-        def fetch(url, path, maximum):
+        def fetch(url, path, maximum, auth=None):
             if url.endswith('index-v1.jar') and not inner:
                 inner.append(1)  # the v1 load passed its fallback check; a v2 load finishes now
                 self.v1 = False
@@ -470,7 +471,7 @@ class Repositories(unittest.TestCase):
         source = self.add()
         self.expire(source)
         release = threading.Event()
-        def slow(url, path, maximum):
+        def slow(url, path, maximum, auth=None):
             release.wait(5)
             self.fetch(url, path, maximum)
         self.fetch_mock.side_effect = slow
@@ -527,7 +528,7 @@ class Repositories(unittest.TestCase):
         self.expire(source)
         before = self.fetch_mock.call_count
         release = __import__('threading').Event()
-        def slow(url, path, maximum):
+        def slow(url, path, maximum, auth=None):
             release.wait(5)
             self.fetch(url, path, maximum)
         self.fetch_mock.side_effect = slow

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

@@ -411,6 +411,59 @@ class ArtworkSettings(unittest.TestCase):
         self.assertLess(page.index("async function api("), page.index('<script src="/artwork-settings.js">'))
 
 
+@unittest.skipUnless(shutil.which("node"), "needs node")
+class PageScripts(unittest.TestCase):
+    """A page's scripts share one global scope, where a later function declaration
+    silently replaces an earlier one of the same name. Issue #68: the Devices page's
+    renderDetail replaced the Android store's, so every app opened as an empty
+    dialog (a bare horizontal line). Compiled together as one module, any top-level
+    name declared twice is a SyntaxError; node's parser handles nested scopes,
+    comments and strings."""
+
+    PAGES = ("index.html", "assistant.html", "mac-view.html")
+
+    @staticmethod
+    def page_scripts(html):
+        """A page's inline scripts, then its /*.js scripts, in one source."""
+        import re
+        parts = re.findall(r"<script>([\s\S]*?)</script>", html)
+        parts += [(ROOT / "ui" / src).read_text(encoding="utf-8") for src in re.findall(r'<script src="/([\w.-]+\.js)">', html)]
+        return "\n;\n".join(parts)
+
+    @staticmethod
+    def module_check(source):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "page.mjs"
+            path.write_text(source, encoding="utf-8")
+            r = subprocess.run(["node", "--check", str(path)], capture_output=True, text=True)
+        return r.returncode, r.stderr
+
+    def test_the_check_catches_top_level_duplicates_only(self):
+        dup = "<script>function a() {}\nfunction a() {}</script>"
+        across = "<script>function a() {}</script><p>x</p><script>  async function a() {}</script>"
+        same_line = "<script>function a() {} function a() {}</script>"
+        const_and_function = "<script>const a = 1, b = 2;\nfunction b() {}</script>"
+        for html in (dup, across, same_line, const_and_function):
+            rc, err = self.module_check(self.page_scripts(html))
+            self.assertNotEqual(rc, 0, html)
+            self.assertIn("has already been declared", err, html)
+        nested = ("<script>function a() {}\nfunction f() { function a() {} if (1) { const a = 2; } }\n"
+                  "// function a() {}\nconst s = `function a() {}`;</script>")
+        self.assertEqual(self.module_check(self.page_scripts(nested)), (0, ""))
+
+    def test_no_top_level_name_is_declared_twice(self):
+        for name in self.PAGES:
+            source = self.page_scripts((ROOT / "ui" / name).read_text(encoding="utf-8"))
+            self.assertIn("function", source, name)
+            rc, err = self.module_check(source)
+            self.assertEqual(rc, 0, f"{name}: {err}")
+
+    def test_store_app_page_renders(self):
+        """tests/page/store_detail.mjs runs the real store functions from index.html."""
+        r = subprocess.run(["node", str(ROOT / "tests" / "page" / "store_detail.mjs")], capture_output=True, text=True)
+        self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
+
+
 @unittest.skipIf(os.name == "nt", "runs on the Frame (Linux); local-bin/ssh is a POSIX shell script")
 class LocalMode(unittest.TestCase):
     """FRAME_LOCAL=1, as the iPhone app starts the server on the Frame: its own key
@@ -482,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",
         }
@@ -498,6 +555,47 @@ class UnreachableMessages(unittest.TestCase):
             self.assertEqual(self.server.error_body(raw), ({"error": raw}, None), raw)
 
 
+class ClipboardDesktopClosed(unittest.TestCase):
+    """The headset's desktop being closed is the person's to fix: a 409, not a 5xx fault for telemetry."""
+
+    @classmethod
+    def setUpClass(cls):
+        sys.path.insert(0, str(ROOT / "ui"))
+        import server
+        cls.server = server
+
+    def paste_failing_with(self, message):
+        s = self.server
+        saved = s.ssh
+
+        def fake_ssh(remote, **kw):
+            raise s.Failure(message)
+        s.ssh = fake_ssh
+        try:
+            s.clipboard({"text": "hello"})
+        finally:
+            s.ssh = saved
+
+    def test_desktop_closed_is_a_409(self):
+        with self.assertRaises(self.server.Failure) as cm:
+            self.paste_failing_with("The headset's desktop isn't open (plasmashell is not running). "
+                                    "Open the desktop in the headset, then try again.")
+        self.assertEqual(cm.exception.status, 409)
+        self.assertIn("Open the desktop", str(cm.exception))
+        # Older wording (scripts/paste-to-frame.sh) is recognised too.
+        with self.assertRaises(self.server.Failure) as cm:
+            self.paste_failing_with("plasmashell is not running: open the desktop in the headset first.")
+        self.assertEqual(cm.exception.status, 409)
+
+    def test_other_failures_stay_faults(self):
+        with self.assertRaises(self.server.Failure) as cm:
+            self.paste_failing_with("Klipper call failed (bus: none)")
+        self.assertEqual(cm.exception.status, 502)
+
+    def test_paste_script_says_it(self):
+        self.assertIn(self.server.DESKTOP_CLOSED, self.server.PASTE)
+
+
 class StatusProbe(unittest.TestCase):
     # frame_status.py only ever runs on the Frame (Linux); it needs os.statvfs.
     @unittest.skipIf(os.name == "nt", "Frame-side script; POSIX only")

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_source_logins.py

@@ -0,0 +1,566 @@
+"""Repositories and sources behind a login: F-Droid basic auth and GitHub tokens.
+
+A fake HTTPS layer stands in for the network (urllib's real redirect, error and
+request processing still run), so these tests see every request and its headers.
+"""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
+import base64
+import contextlib
+import http.client
+import io
+import json
+import os
+from pathlib import Path
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import Mock, patch
+import urllib.request
+import urllib.response
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
+from apk_sources import SourceAuthError, SourceError, SourceLimited, _credentials, _images, _web, fdroid, github, search
+import server
+
+FIXTURES = Path(__file__).parent / 'fixtures' / 'fdroid'
+PIN = (FIXTURES / 'fingerprint.txt').read_text().strip()
+REPO = 'https://repo.example/fdroid/repo/'
+USER, PASSWORD = 'alice', 's3cr@t-pässword'
+TOKEN = 'ghp_frameControlTestToken0123456789'
+SECRETS = (PASSWORD, TOKEN, base64.b64encode((USER + ':' + PASSWORD).encode()).decode())
+REAL_BUILD_OPENER = urllib.request.build_opener
+
+
+def basic(user, password):
+    return 'Basic ' + base64.b64encode((user + ':' + password).encode()).decode()
+
+
+class FakeHTTPS(urllib.request.HTTPSHandler):
+    """Answers https requests from route(url, authorization) -> (status, headers, body)."""
+
+    def __init__(self, route):
+        super().__init__()
+        self.route, self.seen = route, []
+
+    def https_open(self, req):
+        auth = req.get_header('Authorization')
+        self.seen.append((req.full_url, auth))
+        status, headers, body = self.route(req.full_url, auth)
+        message = http.client.HTTPMessage()
+        for key, value in headers.items():
+            message[key] = value
+        response = urllib.response.addinfourl(io.BytesIO(body), message, req.full_url, status)
+        response.msg = 'fake'
+        return response
+
+
+class Base(unittest.TestCase):
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.root = Path(tmp.name)
+        for name, value in [('data_dir', lambda *p: self.root.joinpath('data', *p)),
+                            ('cache_dir', lambda *p: self.root.joinpath('cache', *p))]:
+            mock = patch.object(fdroid.frame_host, name, value)
+            mock.start()
+            self.addCleanup(mock.stop)
+        self.web = FakeHTTPS(self.route)
+        mock = patch('urllib.request.build_opener', lambda *handlers: REAL_BUILD_OPENER(*handlers, self.web))
+        mock.start()
+        self.addCleanup(mock.stop)
+        mock = patch.dict(os.environ, {github.ENV: ''})
+        mock.start()
+        self.addCleanup(mock.stop)
+        for state in (_web._limited, fdroid._stale, fdroid._retry_at, fdroid._refreshing, github._rejected):
+            state.clear()
+            self.addCleanup(state.clear)
+        self.stderr = io.StringIO()
+        mock = patch('sys.stderr', self.stderr)
+        mock.start()
+        self.addCleanup(mock.stop)
+
+    def route(self, url, auth):
+        raise AssertionError('unexpected request ' + url)
+
+    def assertNoSecrets(self, *texts):
+        for text in texts:
+            for secret in SECRETS + (USER + ':',):
+                self.assertNotIn(secret, text)
+
+    def disk(self):
+        """Everything saved under the data and cache folders except the credentials file."""
+        out = []
+        for path in self.root.rglob('*'):
+            if path.is_file() and path.name != 'source-credentials.json':
+                out.append(path.read_bytes().decode('utf-8', 'replace'))
+        return '\n'.join(out)
+
+    def credentials_file(self):
+        return self.root / 'data' / 'source-credentials.json'
+
+
+class Repository(Base):
+    """A repository at REPO that wants USER and PASSWORD."""
+    def setUp(self):
+        super().setUp()
+        self.login = basic(USER, PASSWORD)
+        self.redirect = None  # file name -> absolute URL it redirects to
+        self.forbidden = 401
+
+    def serve(self, name):
+        return (FIXTURES / ('example.apk' if name.endswith('.apk') else name)).read_bytes()
+
+    def route(self, url, auth):
+        for mirror in ('https://mirror.example/fdroid/repo/', 'https://repo.example/elsewhere/'):
+            if url.startswith(mirror):  # serves anyone; a login must never get here unless same-origin
+                return 200, {}, self.serve(url[len(mirror):])
+        self.assertTrue(url.startswith(REPO), url)
+        if auth != self.login:
+            return self.forbidden, {'WWW-Authenticate': 'Basic realm="repo"'}, b'no'
+        name = url[len(REPO):]
+        if self.redirect and name in self.redirect:
+            return 302, {'Location': self.redirect[name]}, b''
+        return 200, {}, self.serve(name)
+
+    def add(self, **kw):
+        return fdroid.add_repo('https://alice:s3cr%40t-p%C3%A4ssword@repo.example/fdroid/repo', PIN, **kw)
+
+
+class RepositoryLogin(Repository):
+    def test_add_search_download_with_login_in_the_url(self):
+        source = self.add()
+        self.assertEqual(source['url'], REPO)
+        self.assertEqual(source['fingerprint'], PIN)
+        saved = fdroid.user_repos()[0]
+        self.assertEqual(saved['url'], REPO)
+        self.assertNotIn('login', saved)
+        self.assertEqual(len(fdroid.search(source, 'example offline')), 1)
+        downloaded = fdroid.download(source, 'org.example.app', 1)
+        self.assertTrue(downloaded['verified'])
+        self.assertTrue(self.web.seen)
+        self.assertTrue(all(auth == self.login for url, auth in self.web.seen))
+        self.assertTrue(all('@' not in url for url, auth in self.web.seen))  # never in the URL
+        self.assertTrue(next(s for s in fdroid.sources() if s['id'] == source['id'])['login'])
+        self.assertNoSecrets(json.dumps(source), json.dumps(fdroid.sources()), self.disk(), self.stderr.getvalue())
+        # The settings file holds no part of the login; the credentials file holds it, privately.
+        settings = (self.root / 'data' / 'apk-repos.json').read_text()
+        self.assertNotIn(USER, settings)
+        self.assertTrue(self.credentials_file().is_file())
+        if os.name != 'nt':
+            self.assertEqual(stat.S_IMODE(self.credentials_file().stat().st_mode), 0o600)
+
+    def test_login_given_apart_and_saved_login_reused_on_refresh(self):
+        source = fdroid.add_repo(REPO, PIN, username=USER, password=PASSWORD)
+        self.assertEqual(source['url'], REPO)
+        self.web.seen.clear()
+        fdroid._load(fdroid.user_repos()[0], force=True)  # as a background refresh would
+        self.assertTrue(self.web.seen)
+        self.assertTrue(all(auth == self.login for url, auth in self.web.seen))
+        fdroid.add_repo(REPO)  # re-adding without a login keeps the saved one
+        self.assertTrue(_credentials.has(source['id']))
+
+    def test_wrong_or_missing_login_saves_nothing(self):
+        for kw, message in [({'username': USER, 'password': 'wrong'}, 'check the username and password'),
+                            ({}, 'needs a username and password')]:
+            with self.subTest(kw=kw):
+                with self.assertRaisesRegex(SourceAuthError, message) as caught:
+                    fdroid.add_repo(REPO, PIN, **kw)
+                self.assertNoSecrets(str(caught.exception), repr(caught.exception))
+                self.assertNotIn('wrong', str(caught.exception))
+                self.assertEqual(fdroid.user_repos(), [])
+                self.assertFalse(self.credentials_file().exists())
+                self.assertFalse((self.root / 'data' / 'apk-repo-state.json').exists())
+                self.assertEqual(_web._limited, {})  # an auth failure isn't a rate limit
+
+    def test_403_with_a_login_is_a_login_error_not_a_rate_limit(self):
+        self.forbidden = 403
+        with self.assertRaisesRegex(SourceAuthError, 'check the username and password'):
+            fdroid.add_repo(REPO, PIN, username=USER, password='wrong')
+        self.assertEqual(_web._limited, {})
+        self.assertEqual(fdroid.user_repos(), [])
+
+    def test_signature_checks_still_apply(self):
+        with self.assertRaisesRegex(SourceError, 'fingerprint mismatch'):
+            fdroid.add_repo(REPO, '0' * 64, username=USER, password=PASSWORD)
+        self.assertEqual(fdroid.user_repos(), [])
+        self.assertFalse(self.credentials_file().exists())
+
+    def test_cross_origin_redirect_never_gets_the_login(self):
+        self.redirect = {'entry.jar': 'https://mirror.example/fdroid/repo/entry.jar',
+                         'index-v2.json': 'https://repo.example/elsewhere/index-v2.json'}
+        source = self.add()
+        self.assertEqual(source['fingerprint'], PIN)  # the mirror's copy still had to verify
+        mirror = [auth for url, auth in self.web.seen if url.startswith('https://mirror.example/')]
+        same_origin = [auth for url, auth in self.web.seen if url.startswith('https://repo.example/elsewhere/')]
+        self.assertEqual(mirror, [None])
+        self.assertEqual(same_origin, [None])  # same origin but outside the repository: no login either
+
+    def test_traversal_names_are_refused_and_never_get_the_login(self):
+        for name in ('%252e%252e/%252e%252e/evil/collect.apk', '..;/..;/evil/collect.apk', 'a%2f..%2f..%2fevil.apk',
+                     'a%5c..%5cevil.apk', '%2e%2e/evil.apk', 'x;y.apk', '100%25.apk'):
+            with self.subTest(name=name), self.assertRaisesRegex(SourceError, 'unsafe|outside'):
+                fdroid._child(REPO, name)
+        self.assertEqual(fdroid._child(REPO, '/app/en-US/phoneScreenshots/#0 a.png'),
+                         REPO + 'app/en-US/phoneScreenshots/%230%20a.png')
+        # Even if such a URL reached the downloader directly, the login stays home.
+        login = _credentials.basic(REPO, USER, PASSWORD)
+        for url in (REPO + '%252e%252e/%252e%252e/evil/collect.apk', REPO + '..;/..;/evil/collect.apk'):
+            with self.subTest(url=url), tempfile.TemporaryDirectory() as tmp:
+                self.web.seen.clear()
+                with contextlib.suppress(SourceError):  # the repository answers 401 without the login
+                    fdroid._fetch(url, Path(tmp) / 'out', 1 << 20, auth=login)
+                self.assertEqual(self.web.seen, [(url, None)])
+
+    def test_same_origin_redirect_outside_the_repository_gets_no_login(self):
+        self.redirect = {'example1.apk': 'https://repo.example/elsewhere/example1.apk'}
+        source = self.add()
+        self.web.seen.clear()
+        self.assertTrue(fdroid.download(source, 'org.example.app', 1)['verified'])
+        self.assertEqual(self.web.seen, [(REPO + 'example1.apk', self.login),
+                                         ('https://repo.example/elsewhere/example1.apk', None)])
+
+    def test_other_ports_and_http_never_get_the_login(self):
+        login = _credentials.basic(REPO, USER, PASSWORD)
+        for url in ('https://repo.example:8443/fdroid/repo/x', 'http://repo.example/fdroid/repo/x',
+                    'https://other.example/fdroid/repo/x', 'https://repo.example.evil/x'):
+            self.assertFalse(login.matches(url), url)
+        self.assertTrue(login.matches('https://REPO.example:443/anything'))
+        self.assertNoSecrets(repr(login))
+        handler = _credentials.Handler(login)
+        req = urllib.request.Request('https://other.example/x', headers={'Authorization': 'Basic stale'})
+        self.assertIsNone(handler.https_request(req).get_header('Authorization'))
+
+    def test_http_and_bad_logins_are_refused_before_any_request(self):
+        for url, kw in [('http://alice:pw@repo.example/fdroid/repo', {}),
+                        ('http://repo.example/fdroid/repo', {'username': USER, 'password': PASSWORD}),
+                        (REPO, {'username': 'a:b', 'password': PASSWORD}),
+                        (REPO, {'username': USER, 'password': 'x\r\nX-Evil: 1'}),
+                        (REPO, {'password': PASSWORD}),
+                        ('https://alice:pw@repo.example/fdroid/repo', {'username': 'bob'})]:
+            with self.subTest(url=url, kw=kw), self.assertRaises(SourceError) as caught:
+                fdroid.add_repo(url, PIN, **kw)
+            self.assertNotIn('pw', str(caught.exception).replace('password', ''))
+            self.assertNoSecrets(str(caught.exception))
+        self.assertEqual(self.web.seen, [])
+
+    def test_remove_deletes_the_login(self):
+        source = self.add()
+        other = _credentials.bearer('https://api.github.com/', TOKEN)
+        _credentials.put('github', other)
+        fdroid.remove_repo(source['id'])
+        self.assertFalse(_credentials.has(source['id']))
+        self.assertNotIn(PASSWORD, self.credentials_file().read_text())
+        _credentials.remove('github')
+        self.assertFalse(self.credentials_file().exists())  # nothing left: the file goes
+
+    def test_cli_reads_the_password_from_stdin(self):
+        out = io.StringIO()
+        argv = ['fdroid.py', 'add', REPO, '--fingerprint', PIN, '--username', USER, '--password-stdin']
+        with patch.object(sys, 'argv', argv), patch.object(sys, 'stdin', io.StringIO(PASSWORD + '\n')), \
+                contextlib.redirect_stdout(out):
+            fdroid.main()
+        self.assertEqual(json.loads(out.getvalue())['url'], REPO)
+        self.assertNoSecrets(out.getvalue())
+        self.assertTrue(_credentials.has(fdroid.user_repos()[0]['id']))
+
+    def test_artwork_from_the_repository_gets_the_login_and_only_there(self):
+        source = self.add()
+        self.assertEqual(_credentials.for_url(REPO + 'icons/a.png').header(), self.login)
+        self.assertTrue(_credentials.has(source['id']))
+        self.assertIsNone(_credentials.for_url('https://repo.example/other/a.png'))  # outside the repository
+        self.assertIsNone(_credentials.for_url('https://cdn.example/fdroid/repo/a.png'))
+
+
+class GitHubToken(Base):
+    def setUp(self):
+        super().setUp()
+        self.accept = {TOKEN}
+        self.releases = json.dumps([]).encode()
+        self.anonymous_limited = False
+
+    def route(self, url, auth):
+        if url.startswith('https://api.github.com/'):
+            if auth is not None and auth not in {'Bearer ' + t for t in self.accept}:
+                return 401, {}, b'{"message": "Bad credentials"}'
+            if auth is None and self.anonymous_limited:
+                return 403, {'X-RateLimit-Remaining': '0', 'X-RateLimit-Reset': str(int(time.time()) + 1800)}, b'{}'
+            if url.endswith('/redirected'):
+                return 302, {'Location': 'https://github.com/landed'}, b''
+            return 200, {}, self.releases
+        if url == 'https://github.com/landed':
+            return 200, {}, b'{"ok": true}'
+        raise AssertionError('unexpected request ' + url)
+
+    def test_saved_token_goes_only_to_the_api(self):
+        github.set_token(TOKEN)
+        self.assertEqual(github._api('/repos/a/b/releases'), [])
+        self.assertEqual(self.web.seen, [('https://api.github.com/repos/a/b/releases', 'Bearer ' + TOKEN)])
+        auth = github._credential(github._token()[0])
+        self.assertEqual(_web.read('https://api.github.com/redirected', ('api.github.com', 'github.com'), auth=auth),
+                         b'{"ok": true}')
+        self.assertEqual(self.web.seen[-2:], [('https://api.github.com/redirected', 'Bearer ' + TOKEN),
+                                              ('https://github.com/landed', None)])  # redirect: no token
+        status = github.sources()[0]['token']
+        self.assertEqual(status, {'set': True, 'from': 'saved', 'rejected': False})
+        self.assertNoSecrets(json.dumps(github.sources()), self.disk(), self.stderr.getvalue())
+        if os.name != 'nt':
+            self.assertEqual(stat.S_IMODE(self.credentials_file().stat().st_mode), 0o600)
+
+    def test_rejected_token_retries_once_without_it(self):
+        github.set_token(TOKEN)
+        self.accept = set()
+        self.assertEqual(github._api('/repos/a/b/releases'), [])
+        self.assertEqual([auth for _, auth in self.web.seen], ['Bearer ' + TOKEN, None])
+        self.assertTrue(github.token_status()['rejected'])
+        self.assertIn('rejected', self.stderr.getvalue())
+        self.assertNoSecrets(self.stderr.getvalue())
+        self.web.seen.clear()
+        github._api('/repos/a/b/other')
+        self.assertEqual([auth for _, auth in self.web.seen], [None])  # not tried again this run
+        github.set_token(TOKEN)  # saving it again gives it a fresh chance
+        self.assertFalse(github.token_status()['rejected'])
+
+    def test_environment_variable_wins(self):
+        github.set_token(TOKEN)
+        env = 'ghp_fromTheEnvironment'
+        self.accept = {env}
+        with patch.dict(os.environ, {github.ENV: env}):
+            github._api('/repos/a/b/releases')
+            self.assertEqual(self.web.seen[-1][1], 'Bearer ' + env)
+            self.assertEqual(github.token_status(), {'set': True, 'from': 'env', 'rejected': False})
+        self.assertEqual(github.token_status()['from'], 'saved')
+
+    def test_clear_removes_it_from_disk(self):
+        github.set_token(TOKEN)
+        github.clear_token()
+        self.assertFalse(self.credentials_file().exists())
+        self.assertEqual(github.token_status(), {'set': False, 'from': None, 'rejected': False})
+        github._api('/repos/a/b/releases')
+        self.assertEqual(self.web.seen[-1][1], None)
+
+    def test_saving_a_token_escapes_the_anonymous_rate_limit(self):
+        self.anonymous_limited = True
+        with self.assertRaisesRegex(SourceLimited, 'GitHub is limiting requests'):
+            github._api('/repos/a/b/releases')
+        self.assertEqual(self.web.seen[-1][1], None)
+        github.set_token(TOKEN)
+        self.assertEqual(github._api('/repos/a/b/other'), [])  # tried, with the token's own quota
+        self.assertEqual(self.web.seen[-1], ('https://api.github.com/repos/a/b/other', 'Bearer ' + TOKEN))
+        github.clear_token()
+        count = len(self.web.seen)
+        with self.assertRaisesRegex(SourceLimited, 'GitHub is limiting requests'):
+            github._api('/repos/a/b/third')
+        self.assertEqual(len(self.web.seen), count)  # anonymous is still backing off: no request
+
+    def test_bad_tokens_are_refused(self):
+        for token in ('', '   ', 'two words', 'x\nAuthorization: y'):
+            with self.subTest(token=token), self.assertRaises(SourceError):
+                github.set_token(token)
+        self.assertFalse(self.credentials_file().exists())
+
+
+PNG = b'\x89PNG\r\n\x1a\nfixture'
+PRIVATE = 'https://repo.example/private/repo/'
+TRAVERSALS = [PRIVATE + '%2e%2e/%2e%2e/evil/collect.png', PRIVATE + '%2E%2E/x.png', PRIVATE + '../../evil/x.png',
+              PRIVATE + './icons/a.png', PRIVATE + '..%2f..%2fevil/x.png', PRIVATE + 'a%5c..%5c..%5cevil.png',
+              PRIVATE + 'a;b/../../evil.png', PRIVATE + '%252e%252e/%252e%252e/evil.png',
+              'https://repo.example/private/repository/x.png', 'https://repo.example/private/repo%2fx.png']
+
+
+class ArtworkScope(Base):
+    """The artwork proxy: actual outgoing headers, through _images.get's own HTTP code."""
+
+    def setUp(self):
+        super().setUp()
+        self.sent, self.redirects = [], {}
+        _credentials.put('fdroid-user-private', _credentials.basic(PRIVATE, USER, PASSWORD))
+        test = self
+
+        class Response:
+            def __init__(self, status, location=''):
+                self.status, self.location, self.body = status, location, PNG if status == 200 else b''
+
+            def getheader(self, name, default=''):
+                return self.location if name == 'Location' else default
+
+            def read1(self, n):
+                chunk, self.body = self.body[:n], self.body[n:]
+                return chunk
+
+        class Connection:
+            def __init__(self, host, port, timeout=None):
+                self.host, self.port = host, port
+
+            def request(self, method, path, headers):
+                url = 'https://%s%s' % (self.host, path)
+                test.sent.append((url, headers.get('Authorization')))
+                self.url = url
+
+            def getresponse(self):
+                if self.url in test.redirects:
+                    return Response(302, test.redirects[self.url])
+                return Response(200)
+
+            def close(self):
+                pass
+
+        # A public address, no real socket or TLS, and Connection in place of http.client's.
+        for target, name, value in [(_images, '_resolve', lambda host, port, timeout: [(2, 1, 6, '', ('93.184.216.34', port))]),
+                                    (_images.socket, 'create_connection', lambda *a, **k: Mock()),
+                                    (_images.ssl, 'create_default_context', lambda: Mock()),
+                                    (_images.http.client, 'HTTPConnection', Connection)]:
+            mock = patch.object(target, name, value)
+            mock.start()
+            self.addCleanup(mock.stop)
+
+    def test_artwork_in_the_repository_gets_the_login(self):
+        self.assertEqual(_images.get(PRIVATE + 'icons/a%20b.png'), PNG)
+        self.assertEqual(self.sent, [(PRIVATE + 'icons/a%20b.png', basic(USER, PASSWORD))])
+
+    def test_paths_a_server_could_normalise_outside_the_repository_get_nothing(self):
+        for url in TRAVERSALS:
+            with self.subTest(url=url):
+                self.sent.clear()
+                self.assertEqual(_images.get(url), PNG)
+                self.assertEqual(self.sent, [(url, None)])
+                self.assertIsNone(_credentials.for_url(url))
+
+    def test_redirect_to_a_traversal_gets_nothing(self):
+        start = PRIVATE + 'icons/a.png'
+        self.redirects[start] = PRIVATE + '%2e%2e/%2e%2e/evil/collect.png'
+        self.assertEqual(_images.get(start), PNG)
+        self.assertEqual(self.sent, [(start, basic(USER, PASSWORD)), (self.redirects[start], None)])
+
+
+class ScopedHandler(Base):
+    """Handler(scoped=True), for callers (artwork-like) that must stay inside a scope, redirects included."""
+
+    def route(self, url, auth):
+        if url == PRIVATE + 'icons/a.png':
+            return 302, {'Location': PRIVATE + '%2e%2e/%2e%2e/evil/collect.png'}, b''
+        return 200, {}, PNG
+
+    def test_redirect_into_a_traversal_is_stripped(self):
+        login = _credentials.basic(PRIVATE, USER, PASSWORD)
+        opener = urllib.request.build_opener(_credentials.Handler(login, scoped=True))
+        with opener.open(PRIVATE + 'icons/a.png') as r:
+            self.assertEqual(r.read(), PNG)
+        self.assertEqual(self.web.seen, [(PRIVATE + 'icons/a.png', basic(USER, PASSWORD)),
+                                         (PRIVATE + '%2e%2e/%2e%2e/evil/collect.png', None)])
+        for url in TRAVERSALS:
+            self.assertFalse(_credentials.in_scope(login, url), url)
+        self.assertTrue(_credentials.in_scope(login, PRIVATE + 'a/b%20c.png?x=1'))
+        self.assertTrue(_credentials.in_scope(login, PRIVATE.rstrip('/')))
+
+
+CHURN = '''
+import sys
+sys.path.insert(0, sys.argv[1])
+from apk_sources import _credentials
+for i in range(int(sys.argv[3])):
+    key = sys.argv[2] + str(i)
+    _credentials.put(key, _credentials.basic('https://h.example/', 'u', 'p'))
+    if i % 2:
+        _credentials.remove(key)
+'''
+
+
+class TwoProcesses(Base):
+    def test_interleaved_saves_and_removals_are_not_lost(self):
+        ui = str(Path(__file__).resolve().parents[1] / 'ui')
+        env = dict(os.environ, FRAME_CONTROL_DATA_DIR=str(self.root / 'data'))
+        count = 60
+        procs = [subprocess.Popen([sys.executable, '-c', CHURN, ui, prefix, str(count)], env=env,
+                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE) for prefix in ('a', 'b')]
+        for p in procs:
+            out, err = p.communicate(timeout=120)
+            self.assertEqual(p.returncode, 0, err.decode())
+        saved = set(json.loads(self.credentials_file().read_text()))
+        self.assertEqual(saved, {p + str(i) for p in 'ab' for i in range(0, count, 2)})
+        if os.name != 'nt':
+            self.assertEqual(stat.S_IMODE(self.credentials_file().stat().st_mode), 0o600)
+
+
+@unittest.skipUnless(shutil.which('node'), 'needs node')
+class PageLogin(unittest.TestCase):
+    def test_login_leaves_the_address_field_before_sending(self):
+        """tests/page/source_login.mjs runs the real page functions from index.html."""
+        r = subprocess.run(['node', str(Path(__file__).parent / 'page' / 'source_login.mjs')],
+                           capture_output=True, text=True)
+        self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
+        self.assertIn('ok', r.stdout)
+
+
+class Endpoints(Repository):
+    """The server API: logins in, never out, in responses or the server log."""
+
+    def setUp(self):
+        super().setUp()
+        mock = patch.object(search, 'modules', return_value=([fdroid, github], []))
+        mock.start()
+        self.addCleanup(mock.stop)
+        self.httpd = server.ThreadingHTTPServer(('127.0.0.1', 0), server.Handler)
+        threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
+        self.addCleanup(self.httpd.server_close)
+        self.addCleanup(self.httpd.shutdown)
+
+    def request(self, method, path, body=None):
+        c = http.client.HTTPConnection('127.0.0.1', self.httpd.server_port)
+        c.request(method, path, json.dumps(body) if body is not None else None,
+                  {'X-Frame-UI': '1', 'Content-Type': 'application/json'})
+        r = c.getresponse()
+        raw = r.read().decode()
+        c.close()
+        self.replies.append(raw)
+        return r.status, json.loads(raw)
+
+    def job(self, reply):
+        for _ in range(500):
+            job = self.request('GET', '/api/job?id=' + reply['job'])[1]
+            if job['done']:
+                return job
+            time.sleep(.01)
+        self.fail('job did not finish')
+
+    def test_add_with_login_and_token_settings(self):
+        self.replies = []
+        bad = self.job(self.request('POST', '/api/sources', {'action': 'add', 'url': REPO, 'fingerprint': PIN,
+                                                             'username': USER, 'password': 'wrong'})[1])
+        self.assertIn('check the username and password', bad['error'])
+        self.assertEqual(fdroid.user_repos(), [])
+        code, reply = self.request('POST', '/api/sources', {'action': 'add', 'url': REPO, 'fingerprint': PIN,
+                                                            'username': USER, 'password': PASSWORD})
+        self.assertEqual(code, 200)
+        job = self.job(reply)
+        self.assertIsNone(job['error'])
+        listed = self.request('GET', '/api/sources')[1]['sources']
+        repo = next(s for s in listed if s['id'] == job['result']['source']['id'])
+        self.assertEqual((repo['url'], repo['login']), (REPO, True))
+        self.assertEqual(self.request('POST', '/api/sources', {'action': 'set-token', 'source': 'github',
+                                                               'token': TOKEN})[0], 200)
+        github_row = next(s for s in self.request('GET', '/api/sources')[1]['sources'] if s['id'] == 'github')
+        self.assertEqual(github_row['token'], {'set': True, 'from': 'saved', 'rejected': False})
+        self.assertEqual(self.request('POST', '/api/sources', {'action': 'set-token', 'source': 'fdroid',
+                                                               'token': TOKEN})[0], 400)
+        self.assertEqual(self.request('POST', '/api/sources', {'action': 'clear-token', 'source': 'github'})[0], 200)
+        self.assertFalse(_credentials.has('github'))
+        self.assertEqual(self.request('POST', '/api/sources', {'action': 'remove', 'source': repo['id']})[0], 200)
+        self.assertFalse(self.credentials_file().exists())
+        self.assertNoSecrets(*self.replies, self.stderr.getvalue(), self.disk())
+
+    def test_login_in_the_url_is_accepted(self):
+        self.replies = []
+        url = 'https://alice:s3cr%40t-p%C3%A4ssword@repo.example/fdroid/repo'
+        job = self.job(self.request('POST', '/api/sources', {'action': 'add', 'url': url, 'fingerprint': PIN})[1])
+        self.assertIsNone(job['error'])
+        self.assertEqual(fdroid.user_repos()[0]['url'], REPO)
+        self.assertNoSecrets(*self.replies, self.stderr.getvalue())
+
+
+if __name__ == '__main__':
+    unittest.main()

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,9 +2,15 @@
 
 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
 import shutil
 import subprocess
 import sys
+import tempfile
+import time
 import unittest
 from pathlib import Path
 
@@ -51,12 +57,9 @@ class Mapping(unittest.TestCase):
         self.assertEqual(t.locate(5, None)["display"], ":0")
 
     def test_focus_display_is_decoded(self):
-        t = self.t
-        saved = t.xprop_root
-        self.addCleanup(lambda: setattr(t, "xprop_root", saved))
         for values, want in (([12602, 0, 58], ":1"), ([12346], ":0"), ([], None), ([0x41], None)):
-            t.xprop_root = lambda d, n, v=values: v
-            self.assertEqual(t.focus_display(), want)
+            self.assertEqual(self.t.focus_display({"GAMESCOPE_FOCUS_DISPLAY": values}), want)
+        self.assertIsNone(self.t.focus_display({}))
 
     def test_ascii_table_covers_printable_characters(self):
         for code in range(0x20, 0x7F):
@@ -259,6 +262,236 @@ class ServerChecks(unittest.TestCase):
             self.s.remote_touch({"events": [{"dx": 1}] * (self.s.INPUT_BATCH_LIMIT + 1)})
 
 
+class FindingPanels(unittest.TestCase):
+    """panels() with faked headset probes: where gamescope's properties are, and why nothing was found (#73)."""
+
+    @classmethod
+    def setUpClass(cls):
+        import frame_touch
+        cls.t = frame_touch
+
+    def setUp(self):
+        t = self.t
+        names = ("read_props", "window_info", "window_pid", "displays", "desktop_running")
+        saved = {n: getattr(t, n) for n in names}
+        saved_which = t.shutil.which
+        self.real_displays = saved["displays"]
+        self.real_read_props = saved["read_props"]
+
+        def restore():
+            for n, v in saved.items():
+                setattr(t, n, v)
+            t.shutil.which = saved_which
+            t._ROOT.clear()
+        self.addCleanup(restore)
+        t._ROOT.clear()
+        self.props = {}  # (display, property) -> values
+        self.windows = {}  # (display, window) -> info
+        self.reads = []  # displays whose gamescope properties were read
+        t.read_props = lambda d, limit=5.0: self.reads.append(d) or {n: list(v) for (dd, n), v in self.props.items() if dd == d}
+        t.window_info = lambda d, w: self.windows.get((d, w))
+        t.window_pid = lambda d, w: None
+        t.displays = lambda: [":0", ":1", ":2"]
+        t.desktop_running = lambda: False
+        t.shutil.which = lambda tool: "/usr/bin/" + tool
+
+    def plasma_on(self, display):
+        # SteamOS 0.4.1: the nested Plasma desktop (kwin_wayland, app id 0) is a window on :0.
+        self.props[(display, "GAMESCOPE_FOCUSABLE_WINDOWS")] = [4194307, 0, 2222]
+        self.props[(display, "GAMESCOPE_FOCUSED_WINDOW")] = [4194307]
+        self.props[(display, "GAMESCOPE_FOCUS_DISPLAY")] = [0x303A]  # ":0"
+        self.windows[(":0", 4194307)] = {"name": "KDE Wayland Compositor", "x": 0, "y": 0, "width": 1280, "height": 800}
+        self.windows[(":0", "root")] = {"width": 1280, "height": 720}
+
+    def test_desktop_panel_on_display_0(self):
+        self.plasma_on(":0")
+        out = self.t.panels()
+        self.assertEqual([(p["window"], p["display"], p["focused"]) for p in out["panels"]], [(4194307, ":0", True)])
+        self.assertNotIn("why", out)
+
+    def test_display_0_costs_one_read(self):
+        # The layout of every build seen so far: one read of :0's properties for the whole run.
+        self.plasma_on(":0")
+        self.t.panels()
+        self.assertEqual(self.reads, [":0"])
+
+    def test_gamescope_properties_on_another_display_are_found(self):
+        # A build that numbers its displays differently: :0 has no gamescope properties.
+        self.plasma_on(":1")
+        self.assertEqual(self.t.gamescope()[0], ":1")
+        self.reads.clear()
+        self.t._ROOT.clear()
+        self.assertEqual(len(self.t.panels()["panels"]), 1)
+        self.assertEqual(self.reads, [":0", ":1"])  # read once, then reused by focus() and focusable()
+
+    def test_other_displays_are_probed_briefly(self):
+        limits = []
+        self.t.read_props = lambda d, limit=5.0: limits.append((d, limit)) or {}
+        self.assertEqual(self.t.gamescope(), (":0", {}))
+        self.assertEqual(limits, [(":0", 5.0), (":1", self.t.ROOT_PROBE_TIMEOUT), (":2", self.t.ROOT_PROBE_TIMEOUT)])
+
+    def test_input_helper_remembers_the_display(self):
+        # The input helper lives for a session: a found display is kept, a failed search retried later.
+        t = self.t
+        self.plasma_on(":1")
+        t.gamescope(); t.gamescope()
+        self.assertEqual(self.reads, [":0", ":1", ":1"])
+        self.props.clear()
+        self.reads.clear()
+        t._ROOT.clear()
+        t.gamescope(); t.gamescope()
+        self.assertEqual(self.reads, [":0", ":1", ":2", ":0"])  # given up: :0 for ROOT_RETRY
+        t._ROOT[1] -= t.ROOT_RETRY
+        self.reads.clear()
+        t.gamescope()
+        self.assertEqual(self.reads, [":0", ":1", ":2"])
+
+    def test_read_props_parses_xprop(self):
+        t = self.t
+        saved = t.subprocess.run
+        self.addCleanup(setattr, t.subprocess, "run", saved)
+        out = ("GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = 1, 0, 2, 3, 769, 4\n"
+               "GAMESCOPE_FOCUSED_WINDOW(CARDINAL) = 0\nGAMESCOPE_FOCUS_DISPLAY:  not found.\n")
+        t.subprocess.run = lambda *a, **k: subprocess.CompletedProcess(a, 0, out, "")
+        self.assertEqual(self.real_read_props(":0"), {"GAMESCOPE_FOCUSABLE_WINDOWS": [1, 0, 2, 3, 769, 4],
+                                                      "GAMESCOPE_FOCUSED_WINDOW": [0]})
+        t.subprocess.run = lambda *a, **k: subprocess.CompletedProcess(a, 0, "GAMESCOPE_FOCUSED_WINDOW:  not found.\n", "")
+        self.assertEqual(self.real_read_props(":0"), {})
+
+        def hang(*a, **k):
+            raise subprocess.TimeoutExpired(a[0], k.get("timeout"))
+        t.subprocess.run = hang
+        self.assertEqual(self.real_read_props(":0"), {})
+
+    def test_deadline_returns_what_was_found(self):
+        t, clock = self.t, [0.0]
+        self.plasma_on(":0")
+        self.props[(":0", "GAMESCOPE_FOCUSABLE_WINDOWS")] = [4194307, 0, 2222, 5, 0, 1, 6, 0, 1]
+        saved = t.time.monotonic
+        self.addCleanup(setattr, t.time, "monotonic", saved)
+        t.time.monotonic = lambda: clock[0]
+        info = self.windows.get
+
+        def slow(d, w):  # every xwininfo takes 1 s: the run is out of time after a few
+            t.budget()
+            clock[0] += 1
+            return info((d, w))
+        t.window_info = slow
+        out = t.panels(limit=7)  # focus: 2 reads; the first panel: 3 displays and its root
+        self.assertEqual([p["window"] for p in out["panels"]], [4194307])
+        self.assertIn("took too long", out["why"])
+        self.assertIn("Some may be missing", out["why"])
+        self.assertIsNone(t.DEADLINE[0])  # the next run (or the input helper) has no deadline left over
+
+    def test_listed_window_that_isnt_x11_explains_itself(self):
+        self.plasma_on(":0")
+        del self.windows[(":0", 4194307)]  # say, a desktop drawn as a Wayland window
+        out = self.t.panels()
+        self.assertEqual(out["panels"], [])
+        self.assertIn("can't find its window", out["why"])
+        self.assertIn("Headset view", out["why"])
+
+    def test_desktop_running_but_not_listed(self):
+        self.t.desktop_running = lambda: True
+        out = self.t.panels()
+        self.assertEqual(out["panels"], [])
+        self.assertIn("desktop is running", out["why"])
+
+    def test_nothing_open(self):
+        out = self.t.panels()
+        self.assertEqual(out, {"focus": None, "focus_display": None, "panels": [],
+                               "why": "No app panel is open in the headset. Open the desktop or an app in "
+                                      "the headset, then try again."})
+
+    def test_missing_x_tools_are_named(self):
+        self.plasma_on(":0")
+        self.t.shutil.which = lambda tool: None if tool == "xwininfo" else "/usr/bin/" + tool
+        self.windows.clear()  # xwininfo can't run, so nothing resolves
+        self.assertIn("no xwininfo", self.t.panels()["why"])
+
+    def test_displays_include_abstract_sockets(self):
+        # /tmp/.X11-unix not visible to the SSH session: the abstract sockets still name the displays.
+        import builtins
+        t = self.t
+        real_open, real_listdir = builtins.open, t.os.listdir
+
+        def fake_open(path, *a, **k):
+            if path == "/proc/net/unix":
+                return io.StringIO("0000: 00000002 0 ... @/tmp/.X11-unix/X1\n0000: ... @/tmp/.X11-unix/X0\n"
+                                   "0000: ... /run/user/1000/wayland-0\n")
+            return real_open(path, *a, **k)
+
+        def fake_listdir(path):
+            if path == "/tmp/.X11-unix":
+                raise FileNotFoundError(path)
+            return real_listdir(path)
+        builtins.open, t.os.listdir = fake_open, fake_listdir
+        try:
+            self.assertEqual(self.real_displays(), [":0", ":1"])
+        finally:
+            builtins.open, t.os.listdir = real_open, real_listdir
+
+
+FAKE_XPROP = """#!/bin/sh
+sleep "${FAKE_X_DELAY:-0}"
+if [ "$1" = "-root" ] && [ "$DISPLAY" = ":0" ]; then
+  echo "GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = 11, 0, 101, 12, 769, 102"
+  echo "GAMESCOPE_FOCUSED_WINDOW(CARDINAL) = 11"
+  echo "GAMESCOPE_FOCUS_DISPLAY(CARDINAL) = 12346"
+fi
+"""
+FAKE_XWININFO = """#!/bin/sh
+sleep "${FAKE_X_DELAY:-0}"
+[ "$DISPLAY" = ":0" ] || exit 1
+echo 'xwininfo: Window id: 0xb "Desktop"'
+echo '  Absolute upper-left X:  0'
+echo '  Absolute upper-left Y:  0'
+echo '  Width: 1280'
+echo '  Height: 800'
+echo '  Map State: IsViewable'
+"""
+
+
+@unittest.skipIf(sys.platform == "win32", "fake X tools are shell scripts; frame_touch.py runs on the Frame")
+class PanelsAsAFreshProcess(unittest.TestCase):
+    """`frame_touch.py panels` the way the server runs it: a new process each request, with
+    fake xprop and xwininfo first on PATH (two displays, :0 and :1)."""
+
+    def run_panels(self, delay):
+        with tempfile.TemporaryDirectory() as bin_dir:
+            for name, body in (("xprop", FAKE_XPROP), ("xwininfo", FAKE_XWININFO)):
+                path = Path(bin_dir) / name
+                path.write_text(body)
+                path.chmod(0o755)
+            env = dict(os.environ, PATH=bin_dir + os.pathsep + os.environ.get("PATH", ""), FAKE_X_DELAY=str(delay),
+                       PYTHONDONTWRITEBYTECODE="1")
+            code = ("import sys; sys.path.insert(0, sys.argv[1]); import frame_touch as t; "
+                    "t.displays = lambda: [':0', ':1']; t.desktop_running = lambda: False; "
+                    "sys.argv = ['frame_touch.py', 'panels']; sys.exit(t.main())")
+            start = time.monotonic()
+            r = subprocess.run([sys.executable, "-c", code, str(ROOT / "ui")], env=env, capture_output=True,
+                               text=True, timeout=60)
+            took = time.monotonic() - start
+        self.assertEqual(r.returncode, 0, r.stderr)
+        return json.loads(r.stdout), took
+
+    def test_lists_panels(self):
+        out, _ = self.run_panels(0)
+        self.assertEqual(out["focus"], 11)
+        self.assertEqual(out["focus_display"], ":0")
+        self.assertEqual([(p["window"], p["display"], p["focused"]) for p in out["panels"]],
+                         [(11, ":0", True), (12, ":0", False)])
+        self.assertNotIn("why", out)
+
+    def test_slow_x_tools_stay_inside_the_servers_timeout(self):
+        # Review of #83: X tools answering in 4.6 s each would take well over the server's
+        # 20 s; one deadline for the run (PANELS_DEADLINE, 12 s) returns in time, saying why.
+        out, took = self.run_panels(4.6)
+        self.assertLess(took, 15, out)
+        self.assertEqual(out["panels"], [])
+        self.assertIn("took too long", out["why"])
+
+
 @unittest.skipUnless(shutil.which("node"), "needs node")
 class PageGestures(unittest.TestCase):
     """Control's gestures and queue (tests/page/ctrl_gestures.mjs runs the real functions from index.html)."""
@@ -268,6 +501,12 @@ class PageGestures(unittest.TestCase):
                            timeout=60)
         self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
 
+    def test_desktop_panel_list(self):
+        # #73: the headset's reason for no panels, and a failed listing, reach Capture and Control.
+        r = subprocess.run(["node", str(ROOT / "tests" / "page" / "desk_panels.mjs")], capture_output=True, text=True,
+                           timeout=60)
+        self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
+
 
 class Script(unittest.TestCase):
     def test_helper_compiles_on_the_frames_python(self):

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/apk_sources/__init__.py

@@ -45,6 +45,10 @@ class SourceError(Exception):
     """User-readable failure from a source (network, format, verification)."""
 
 
+class SourceAuthError(SourceError):
+    """The server refused our login (HTTP 401, or 403 when a login was sent)."""
+
+
 class SourceLimited(SourceError):
     """The source's host asked us to slow down; retry_after is in seconds."""
 

ui/apk_sources/_credentials.py

@@ -0,0 +1,270 @@
+"""Logins for sources behind one: HTTP basic auth and access tokens.
+
+Saved apart from the source settings, in frame_host.data_dir('source-credentials.json')
+(mode 0600), keyed by a caller-chosen name: an F-Droid source id, 'github'. Each
+credential belongs to one exact origin (https, host, port) and is only ever sent
+to a request for that origin: Handler adds it to each request, redirects
+included, and removes any Authorization header from every other request.
+
+Scoped use (for_url, and Handler(..., scoped=True), which F-Droid repositories and
+artwork use) is stricter: the URL must also be under the credential's scope by whole
+path segments, with no dot segments, encoded slashes or ';' a server could normalise
+into a path outside it. Origin-only matching suits a host-wide token (GitHub's API).
+
+Secrets leave this module only as an Authorization header value. Credential's
+repr hides them, and nothing here puts them in an exception message.
+
+Each change is a read-modify-write under an inter-process lock (the CLI and the
+app may both save), so one process can't undo another's save or removal.
+"""
+import base64
+import contextlib
+import errno
+import json
+import os
+import tempfile
+import threading
+import urllib.parse
+import urllib.request
+
+import frame_host
+
+_lock = threading.RLock()
+MAX = 4096  # longest username, password or token accepted
+
+
+def _path():
+    return frame_host.data_dir('source-credentials.json')
+
+
+@contextlib.contextmanager
+def _changing():
+    """This thread, then this process, holds the credentials file for a read-modify-write."""
+    path = _path().with_name('source-credentials.lock')
+    with _lock:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with open(str(path), 'a+b') as f:
+            if os.name == 'nt':
+                import msvcrt
+                while True:
+                    try:
+                        f.seek(0)
+                        msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)
+                        break
+                    except OSError as e:  # LK_LOCK gives up after ~10 s of contention; keep waiting
+                        if e.errno not in (errno.EACCES, errno.EDEADLK):
+                            raise
+                try:
+                    yield
+                finally:
+                    f.seek(0)
+                    msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
+            else:
+                import fcntl
+                fcntl.flock(f.fileno(), fcntl.LOCK_EX)
+                try:
+                    yield
+                finally:
+                    fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+
+
+def path_segments(path):
+    """A URL path's decoded segments, or None if a server might read it as another path:
+    '.' or '..' (plain or percent-encoded), encoded '/' or '\\', '%' left after decoding
+    (double encoding), or ';' path parameters."""
+    if ';' in path or '\\' in path:
+        return None
+    parts = [urllib.parse.unquote(part) for part in path.split('/')]
+    if any(part in ('.', '..') or '/' in part or '\\' in part or '%' in part or ';' in part for part in parts):
+        return None
+    return parts
+
+
+def in_scope(credential, url):
+    """True if url is on credential's origin and under its scope by whole, plain path segments."""
+    if credential is None or not credential.matches(url):
+        return False
+    try:
+        path, scope = urllib.parse.urlsplit(url).path, urllib.parse.urlsplit(credential.scope).path
+    except ValueError:
+        return False
+    parts, base = path_segments(path), path_segments(scope)
+    if parts is None or base is None:
+        return False
+    if base and base[-1] == '':
+        base = base[:-1]  # 'https://h/repo/' covers '/repo/x', not '/repository'
+    return parts[:len(base)] == base
+
+
+def origin(url):
+    """'https://host:port' for an https URL; ValueError for anything else."""
+    p = urllib.parse.urlsplit(url)
+    port = p.port  # raises ValueError on a bad port
+    if p.scheme.lower() != 'https' or not p.hostname:
+        raise ValueError('credentials are only sent over HTTPS')
+    host = p.hostname.lower()
+    return 'https://%s:%d' % ('[%s]' % host if ':' in host else host, port or 443)
+
+
+def _check(value, what, allow_empty=False):
+    if not isinstance(value, str) or (not value and not allow_empty) or len(value) > MAX:
+        raise ValueError('invalid ' + what)
+    if any(ord(c) < 32 or ord(c) == 127 for c in value):
+        raise ValueError(what + ' must not contain control characters')
+    return value
+
+
+class Credential:
+    """A username and password (basic) or a token (bearer) for one origin.
+
+    scope is the URL prefix it belongs under: in_scope checks it (by path segments),
+    matches() checks only the origin."""
+    __slots__ = ('origin', 'scope', 'username', '_secret')
+
+    def __init__(self, scope, secret, username=None):
+        self.origin = origin(scope)
+        self.scope = scope
+        if username is not None:
+            _check(username, 'username')
+            if ':' in username:
+                raise ValueError('username must not contain ":"')
+            _check(secret, 'password', allow_empty=True)
+        else:
+            _check(secret, 'token')
+            if any(c.isspace() for c in secret):
+                raise ValueError('token must not contain spaces')
+        self.username, self._secret = username, secret
+
+    @property
+    def kind(self):
+        return 'bearer' if self.username is None else 'basic'
+
+    def header(self):
+        if self.username is None:
+            return 'Bearer ' + self._secret
+        return 'Basic ' + base64.b64encode((self.username + ':' + self._secret).encode('utf-8')).decode('ascii')
+
+    def matches(self, url):
+        """True only for a URL on exactly this credential's origin."""
+        try:
+            return origin(url) == self.origin
+        except ValueError:
+            return False
+
+    def same(self, other):
+        return isinstance(other, Credential) and (self.origin, self.username, self._secret) == (
+            other.origin, other.username, other._secret)
+
+    def __repr__(self):
+        return '<Credential %s for %s>' % (self.kind, self.origin)
+
+    def _json(self):
+        return {'scope': self.scope, 'username': self.username, 'secret': self._secret}
+
+
+def basic(scope, username, password):
+    """Basic auth for scope's origin, e.g. basic('https://example.org/fdroid/repo/', 'me', 'pw')."""
+    return Credential(scope, password, username)
+
+
+def bearer(scope, token):
+    """An access token for scope's origin, sent as 'Authorization: Bearer <token>'."""
+    return Credential(scope, token)
+
+
+def _read():
+    try:
+        with open(str(_path()), encoding='utf-8') as f:
+            data = json.load(f)
+        if not isinstance(data, dict):
+            raise ValueError
+        return data
+    except FileNotFoundError:
+        return {}
+    except (OSError, ValueError):
+        return {}  # unreadable: as if none were saved; the next save rewrites it
+
+
+def _save(data):
+    path = _path()
+    if not data:
+        try:
+            os.remove(str(path))
+        except FileNotFoundError:
+            pass
+        return
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix='.credentials-', suffix='.part')  # created 0600
+    try:
+        with os.fdopen(fd, 'w', encoding='utf-8') as f:
+            json.dump(data, f, separators=(',', ':'))
+        os.chmod(tmp, 0o600)
+        os.replace(tmp, str(path))
+    finally:
+        if os.path.exists(tmp):
+            os.unlink(tmp)
+
+
+def get(key):
+    """The saved Credential for key, or None."""
+    with _lock:
+        item = _read().get(key)
+    try:
+        return Credential(item['scope'], item['secret'], item.get('username'))
+    except (TypeError, KeyError, ValueError):
+        return None
+
+
+def has(key):
+    return get(key) is not None
+
+
+def put(key, credential):
+    if not isinstance(credential, Credential):
+        raise TypeError('put() needs a Credential')
+    with _changing():
+        data = _read()
+        data[key] = credential._json()
+        _save(data)
+
+
+def remove(key):
+    """Delete key's credential from disk; True if there was one."""
+    with _changing():
+        data = _read()
+        if key not in data:
+            return False
+        del data[key]
+        _save(data)
+        return True
+
+
+def for_url(url):
+    """A saved credential url is in the scope of (see in_scope), or None. Call it again for
+    each redirect hop: the answer depends only on the URL being requested."""
+    with _lock:
+        keys = list(_read())
+    for key in keys:
+        credential = get(key)
+        if in_scope(credential, url):
+            return credential
+    return None
+
+
+class Handler(urllib.request.BaseHandler):
+    """Adds the credential to requests for its origin (redirects included) and strips
+    Authorization from every other request, so a redirect elsewhere never gets it.
+    scoped=True also requires each request, redirects included, to be in_scope."""
+    handler_order = 900  # after the default processors
+
+    def __init__(self, credential, scoped=False):
+        self.credential, self.scoped = credential, scoped
+
+    def _request(self, req):
+        req.remove_header('Authorization')
+        allowed = in_scope if self.scoped else lambda c, url: c is not None and c.matches(url)
+        if allowed(self.credential, req.full_url):
+            req.add_unredirected_header('Authorization', self.credential.header())
+        return req
+
+    http_request = https_request = _request

ui/apk_sources/_images.py

@@ -9,7 +9,7 @@ import threading
 import time
 from urllib.parse import urljoin, urlsplit
 
-from apk_sources import SourceError
+from apk_sources import SourceError, _credentials
 
 _lock = threading.Lock()
 _urls = OrderedDict()
@@ -141,7 +141,11 @@ def get(url, headers=None, redirects=3, deadline=None, limit=MAX_IMAGE):
         path = p.path or '/'
         if p.query:
             path += '?' + p.query
-        conn.request('GET', path, headers={'User-Agent': 'FrameControl/0.3.1', **(headers or {})})
+        sent = {'User-Agent': 'FrameControl/0.3.1', **(headers or {})}
+        login = _credentials.for_url(url)  # a repository behind a login: its own artwork only, per hop
+        if login:
+            sent['Authorization'] = login.header()
+        conn.request('GET', path, headers=sent)
         live[0].settimeout(min(10, left()))
         response = conn.getresponse()
         if response.status in (301, 302, 303, 307, 308) and redirects:

ui/apk_sources/_web.py

@@ -3,7 +3,7 @@ import hashlib, os, shutil, tempfile, threading, time, urllib.error, urllib.pars
 from email.utils import parsedate_to_datetime
 
 import frame_host
-from . import SourceError, SourceLimited
+from . import SourceAuthError, SourceError, SourceLimited, _credentials
 
 UA = 'FrameControl/0.1'
 APK_CAP = 2 * 1024 ** 3  # cached APKs across all sources, least recently used go first
@@ -38,7 +38,7 @@ def release(path):
         else:
             _in_use.pop(path, None)
 BACKOFF = 600  # seconds to leave a host alone after 403/429 without Retry-After
-_limited = {}  # host -> time.time() before which we don't contact it
+_limited = {}  # host (+ login identity) -> time.time() before which we don't ask it with that login
 _limited_lock = threading.Lock()
 
 
@@ -46,8 +46,16 @@ def _host(url):
     return (urllib.parse.urlsplit(url).hostname or '').lower()
 
 
-def throttle(url, headers=None):
-    """Remember that url's host asked us to back off; return the delay in seconds."""
+def _who(url, auth=None):
+    """Backoff key: the host, per login. A token has its own quota, so anonymous
+    rate limiting mustn't hold back requests made with one (or the reverse)."""
+    if auth is None:
+        return _host(url)
+    return _host(url) + ' ' + hashlib.sha256(auth.header().encode()).hexdigest()[:16]
+
+
+def throttle(url, headers=None, auth=None):
+    """Remember that url's host asked us (with this login) to back off; return the delay in seconds."""
     headers = headers or {}
     now, delay = time.time(), None
     value = (headers.get('Retry-After') or '').strip()
@@ -62,14 +70,15 @@ def throttle(url, headers=None):
     if delay is None and headers.get('X-RateLimit-Remaining') == '0' and reset.isdigit():
         delay = int(reset) - now  # GitHub
     delay = min(max(delay if delay is not None else BACKOFF, 1), 6 * 3600)
+    key = _who(url, auth)
     with _limited_lock:
-        _limited[_host(url)] = max(_limited.get(_host(url), 0), now + delay)
+        _limited[key] = max(_limited.get(key, 0), now + delay)
     return delay
 
 
-def wait_time(url):
+def wait_time(url, auth=None):
     with _limited_lock:
-        return max(0, _limited.get(_host(url), 0) - time.time())
+        return max(0, _limited.get(_who(url, auth), 0) - time.time())
 
 
 def limited_error(name, seconds, hint=''):
@@ -136,6 +145,8 @@ def checked_url(url, hosts):
 
 
 class Redirect(urllib.request.HTTPRedirectHandler):
+    """Only to the allowed hosts, and never carrying an Authorization header along:
+    a credential passed to open_url() is added again, by its Handler, only on its origin."""
     def __init__(self, hosts):
         self.hosts = hosts
 
@@ -147,13 +158,15 @@ class Redirect(urllib.request.HTTPRedirectHandler):
         return result
 
 
-def open_url(url, hosts, headers=None):
+def open_url(url, hosts, headers=None, auth=None):
+    """auth: an _credentials.Credential, sent only to requests on its own origin."""
     checked_url(url, hosts)
-    return urllib.request.build_opener(Redirect(hosts)).open(
+    handlers = [Redirect(hosts)] + ([_credentials.Handler(auth)] if auth else [])
+    return urllib.request.build_opener(*handlers).open(
         urllib.request.Request(url, headers={'User-Agent': UA, **(headers or {})}), timeout=60)
 
 
-def read(url, hosts, headers=None, ttl=3600, name=None, hint=''):
+def read(url, hosts, headers=None, ttl=3600, name=None, hint='', auth=None):
     path = os.path.join(cache(), hashlib.sha256(url.encode()).hexdigest() + '.data')
     name = name or _host(url) or 'The source'
 
@@ -165,13 +178,13 @@ def read(url, hosts, headers=None, ttl=3600, name=None, hint=''):
         if os.path.isfile(path) and time.time() - os.path.getmtime(path) < ttl:
             with open(path, 'rb') as f:
                 return f.read()
-        wait = wait_time(url)
+        wait = wait_time(url, auth)
         if wait:
             data = cached()
             if data is None:
                 raise limited_error(name, wait, hint)
             return data
-        with open_url(url, hosts, headers) as r:
+        with open_url(url, hosts, headers, auth) as r:
             data = r.read(8 * 1024 * 1024 + 1)
         if len(data) > 8 * 1024 * 1024:
             raise SourceError('Source index is too large')
@@ -185,8 +198,10 @@ def read(url, hosts, headers=None, ttl=3600, name=None, hint=''):
                 os.remove(tmp)
         return data
     except urllib.error.HTTPError as e:
+        if e.code == 401:  # not 403: GitHub also uses that for rate limits, token or not
+            raise SourceAuthError('%s refused the login (HTTP 401)' % name) from e
         if e.code in (403, 429):
-            delay = throttle(url, e.headers)
+            delay = throttle(url, e.headers, auth)
             data = cached()
             if data is None:
                 raise limited_error(name, delay, hint) from e

ui/apk_sources/fdroid.py

@@ -20,7 +20,7 @@ import zipfile
 
 if __package__ in (None, ''):
     sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-from apk_sources import SourceError, SourceLimited, _web
+from apk_sources import SourceAuthError, SourceError, SourceLimited, _credentials, _web
 import frame_host
 from frame_apk_sign import _der_parts, _cert_key, der
 from frame_catalog import _IndexReader, _reduce_index, _sha256
@@ -71,6 +71,39 @@ def _url(url, fingerprint=None):
     return urllib.parse.urlunsplit(('https', p.netloc.lower(), p.path.rstrip('/') + '/', '', '')), pin
 
 
+def _split_login(url, username=None, password=None):
+    """(url without user:pass, credential or None). The address may carry the login, as
+    F-Droid's client accepts (https://user:pass@host/fdroid/repo), or it may be given apart."""
+    try:
+        p = urllib.parse.urlsplit(url)
+        p.port
+    except ValueError as e:
+        raise SourceError('invalid repository URL') from e
+    if '@' in p.netloc:
+        if p.username is None:
+            raise SourceError('invalid repository URL')
+        linked = urllib.parse.unquote(p.username), None if p.password is None else urllib.parse.unquote(p.password)
+        if username not in (None, '', linked[0]) or (linked[1] is not None and password not in (None, '', linked[1])):
+            raise SourceError('the address and the username/password fields disagree')
+        username = linked[0]
+        password = linked[1] if linked[1] is not None else password
+        url = urllib.parse.urlunsplit((p.scheme, p.netloc.rpartition('@')[2], p.path, p.query, p.fragment))
+    if username in (None, ''):
+        if password:
+            raise SourceError('a password needs a username')
+        return url, None
+    clean, _ = _url(url)
+    try:
+        return url, _credentials.basic(clean, username, password or '')
+    except ValueError as e:
+        raise SourceError('cannot use that login: ' + str(e)) from None
+
+
+def _login(source):
+    """The saved login for a user repository, or None."""
+    return None if source.get('builtin') else _credentials.get(source['id'])
+
+
 def _child(base, name):
     name = str(name).lstrip('/')
     decoded = urllib.parse.unquote(name)
@@ -81,6 +114,9 @@ def _child(base, name):
     url = urllib.parse.urljoin(base, urllib.parse.quote(decoded, safe='/'))  # names may contain '#' or spaces
     if not url.startswith(base) or urllib.parse.urlsplit(url).query or urllib.parse.urlsplit(url).fragment:
         raise SourceError('repository file is outside its repository')
+    # Nothing a server could normalise elsewhere: double-encoded dots, ';' parameters, encoded slashes.
+    if _credentials.path_segments(urllib.parse.urlsplit(url).path) is None:
+        raise SourceError('unsafe repository file name')
     return url
 
 
@@ -91,14 +127,17 @@ class _HTTPSRedirect(urllib.request.HTTPRedirectHandler):
         return super().redirect_request(req, fp, code, msg, headers, newurl)
 
 
-def _fetch(url, path, maximum):
+def _fetch(url, path, maximum, auth=None):
+    """auth: the repository's login, sent only inside the repository (its origin and directory,
+    by whole plain path segments), on every request including redirects."""
     host = urllib.parse.urlsplit(url).hostname
-    wait = _web.wait_time(url)
+    wait = _web.wait_time(url, auth)
     if wait:
         raise _web.limited_error(host, wait)
     request = urllib.request.Request(url, headers={'User-Agent': 'FrameControl/1.0'})
+    handlers = [_HTTPSRedirect()] + ([_credentials.Handler(auth, scoped=True)] if auth else [])
     try:
-        with urllib.request.build_opener(_HTTPSRedirect()).open(request, timeout=60) as r, open(path, 'wb') as f:
+        with urllib.request.build_opener(*handlers).open(request, timeout=60) as r, open(path, 'wb') as f:
             total = 0
             while True:
                 chunk = r.read(1 << 20)
@@ -109,8 +148,11 @@ def _fetch(url, path, maximum):
                     raise SourceError('repository file exceeds size limit')
                 f.write(chunk)
     except urllib.error.HTTPError as e:
+        if e.code == 401 or (e.code == 403 and auth):
+            raise SourceAuthError('The repository refused the login (HTTP %d); check the username and password'
+                                  % e.code if auth else 'This repository needs a username and password') from None
         if e.code in (403, 429):
-            raise _web.limited_error(host, _web.throttle(url, e.headers)) from e
+            raise _web.limited_error(host, _web.throttle(url, e.headers, auth)) from e
         raise
 
 
@@ -342,7 +384,7 @@ def sources():
     return [dict(id=i, kind=KIND, name=n, url=u, fingerprint=p, builtin=True,
                  # The archive only holds superseded versions; it clutters search unless asked for.
                  enabled=settings['enabled'].get(i, i != 'fdroid-archive'), trust='community')
-            for i, n, u, p in builtins] + settings['repos']
+            for i, n, u, p in builtins] + [dict(s, login=_credentials.has(s['id'])) for s in settings['repos']]
 
 
 def _text(value):
@@ -514,10 +556,12 @@ def _source_lock(source_id):
         return _load_locks.setdefault(source_id, threading.Lock())
 
 
-def _load(source, force=False):
+def _load(source, force=False, auth=None):
+    """auth: a login to use instead of the saved one (while adding, before anything is saved)."""
     if not re.fullmatch(r'[a-z0-9-]+', source['id']):
         raise SourceError('invalid source id')
     _url(source['url'], source.get('fingerprint'))
+    auth = auth or _login(source)
     cache = frame_host.cache_dir('apk-sources', source['id'] + '.json')
     found = None if force else _cached(source, cache)
     if found:
@@ -535,7 +579,7 @@ def _load(source, force=False):
                 jar, raw = Path(tmp) / 'index.jar', Path(tmp) / 'index.json'
                 v2 = True
                 try:
-                    _fetch(source['url'] + 'entry.jar', jar, 8 * 1024 * 1024)
+                    _fetch(source['url'] + 'entry.jar', jar, 8 * 1024 * 1024, auth=auth)
                 except urllib.error.HTTPError as e:
                     if e.code not in (404, 410):
                         raise
@@ -543,7 +587,7 @@ def _load(source, force=False):
                         raise SourceError('repository no longer serves its signed v2 index; '
                                           'refusing to fall back to the older v1 index') from e
                     v2 = False
-                    _fetch(source['url'] + 'index-v1.jar', jar, 256 * 1024 * 1024)
+                    _fetch(source['url'] + 'index-v1.jar', jar, 256 * 1024 * 1024, auth=auth)
                     content, pin = _jar(jar, 'index-v1.json', source.get('fingerprint'))
                     timestamp = _v1(content, raw)
                     _check_timestamp(source, timestamp)
@@ -552,7 +596,7 @@ def _load(source, force=False):
                     signed = json.loads(content)
                     timestamp, entry = signed.get('timestamp'), signed['index']
                     _check_timestamp(source, timestamp)
-                    _fetch(_child(source['url'], entry['name']), raw, 512 * 1024 * 1024)
+                    _fetch(_child(source['url'], entry['name']), raw, 512 * 1024 * 1024, auth=auth)
                     if _sha256(raw) != entry['sha256'] or (entry.get('size') is not None and raw.stat().st_size != entry['size']):
                         raise SourceError('index SHA-256 or size mismatch')
                 apps = _reduce(raw, source)
@@ -572,7 +616,10 @@ def _load(source, force=False):
             raise SourceError('cannot load repository: ' + str(e)) from e
 
 
-def add_repo(url, fingerprint=None, name=None):
+def add_repo(url, fingerprint=None, name=None, username=None, password=None):
+    """Check and save a repository. A login (in the URL or apart) is kept out of the
+    saved URL, in _credentials, and saved only once the repository's index verified."""
+    url, login = _split_login(url, username, password)
     url, pin = _url(url, fingerprint)
     with _LOCK:
         existing = next((s for s in _read()['repos'] if s['url'] == url), None)
@@ -583,15 +630,26 @@ def add_repo(url, fingerprint=None, name=None):
     source = dict(id='fdroid-user-' + hashlib.sha256(url.encode()).hexdigest()[:20], kind=KIND,
                   name=name or (existing or {}).get('name') or urllib.parse.urlsplit(url).hostname,
                   url=url, builtin=False, enabled=True, trust='user', fingerprint=pin)
-    _, source['fingerprint'] = _load(source, force=True)  # network work outside the settings lock
+    # Network work outside the settings lock; without a new login, a saved one (re-adding) is used.
+    _, source['fingerprint'] = _load(source, force=True, auth=login)
     source['trust_on_first_use'] = existing.get('trust_on_first_use', False) if existing else pin is None
     with _LOCK:
         settings = _read()
         current = next((s for s in settings['repos'] if s['id'] == source['id']), None)
         if current and current['fingerprint'] != source['fingerprint']:
             raise SourceError('repository already has a different pinned fingerprint; remove it first')
-        settings['repos'] = [s for s in settings['repos'] if s['id'] != source['id']] + [source]
-        _write(_storage(), settings)
+        previous = _credentials.get(source['id'])
+        if login:
+            _credentials.put(source['id'], login)
+        try:
+            settings['repos'] = [s for s in settings['repos'] if s['id'] != source['id']] + [source]
+            _write(_storage(), settings)
+        except BaseException:
+            if login and previous:  # not saved after all: leave the login as it was
+                _credentials.put(source['id'], previous)
+            elif login:
+                _credentials.remove(source['id'])
+            raise
     return source
 
 
@@ -602,6 +660,7 @@ def remove_repo(source_id):
             raise SourceError('unknown user repository')
         settings['repos'] = [s for s in settings['repos'] if s['id'] != source_id]
         _write(_storage(), settings)
+        _credentials.remove(source_id)
     with _state_file_lock(), _LOCK:  # lock order: state file, then _LOCK
         states = _states()
         if states.pop(source_id, None) is not None:  # re-adding is a deliberate new trust decision
@@ -663,7 +722,7 @@ def download(source, entry_id, version_code=None):
             fd, tmp = tempfile.mkstemp(dir=str(path.parent), suffix='.part')
             os.close(fd)
             try:
-                _fetch(_child(source['url'], version['name']), tmp, 4 * 1024 ** 3)
+                _fetch(_child(source['url'], version['name']), tmp, 4 * 1024 ** 3, auth=_login(source))
                 if _sha256(tmp) != sha:
                     raise SourceError('APK SHA-256 mismatch; download discarded')
                 os.replace(tmp, path)
@@ -684,6 +743,8 @@ def main():
     add.add_argument('url')
     add.add_argument('--fingerprint')
     add.add_argument('--name')
+    add.add_argument('--username', help='for a repository behind a login (HTTP basic auth)')
+    add.add_argument('--password-stdin', action='store_true', help='read the password from the first line of stdin')
     sub.add_parser('list')
     remove = sub.add_parser('remove')
     remove.add_argument('source')
@@ -702,7 +763,12 @@ def main():
 def _cli(parser, args):
     try:
         if args.command == 'add':
-            result = add_repo(args.url, args.fingerprint, args.name)
+            password = None
+            if args.password_stdin:
+                if not args.username:
+                    raise SourceError('--password-stdin needs --username')
+                password = sys.stdin.readline().rstrip('\r\n')
+            result = add_repo(args.url, args.fingerprint, args.name, args.username, password)
         elif args.command == 'list':
             result = sources()
         elif args.command == 'remove':

ui/apk_sources/github.py

@@ -1,17 +1,77 @@
 """Curated publisher releases; optional topic discovery is page-only."""
-import fnmatch, json, os, re, urllib.parse
+import fnmatch, hashlib, json, os, re, sys, threading, urllib.parse
 
-from . import SourceError, _web
+from . import SourceAuthError, SourceError, _credentials, _web
 
 KIND = 'github'
 API = 'https://api.github.com'
+ENV = 'FRAME_GITHUB_TOKEN'  # wins over a token saved in source settings
+CREDENTIAL = 'github'  # _credentials key; only ever sent to api.github.com
+_rejected = set()  # SHA-256 of tokens GitHub answered 401 to (this run only)
+_rejected_lock = threading.Lock()
 TOPICS = ('oculus-quest', 'openxr', 'quest')
 HOSTS = ('github.com', 'release-assets.githubusercontent.com', 'objects.githubusercontent.com')
 
 
 def sources():
     return [{'id': KIND, 'kind': KIND, 'name': 'GitHub releases', 'url': 'https://github.com',
-             'builtin': True, 'enabled': True, 'trust': 'community'}]
+             'builtin': True, 'enabled': True, 'trust': 'community', 'token': token_status()}]
+
+
+def _token():
+    """(token, where it came from) or (None, None). The environment variable wins."""
+    env = os.environ.get(ENV)
+    if env:
+        return env, 'env'
+    saved = _credentials.get(CREDENTIAL)
+    if saved and saved.kind == 'bearer':
+        return saved, 'saved'
+    return None, None
+
+
+def _credential(token):
+    if isinstance(token, _credentials.Credential):
+        return token
+    try:
+        return _credentials.bearer(API + '/', token)
+    except ValueError as e:
+        raise SourceError('%s is not a usable GitHub token: %s' % (ENV, e)) from None
+
+
+def _digest(credential):
+    return hashlib.sha256(credential.header().encode()).hexdigest()
+
+
+def token_status():
+    """Whether a token is set, where from, and whether GitHub rejected it; never the token."""
+    token, where = _token()
+    rejected = False
+    if token is not None:
+        try:
+            with _rejected_lock:
+                rejected = _digest(_credential(token)) in _rejected
+        except SourceError:
+            rejected = True
+    return {'set': token is not None, 'from': where, 'rejected': rejected}
+
+
+def set_token(token):
+    """Save a personal access token for api.github.com (FRAME_GITHUB_TOKEN still wins)."""
+    if not isinstance(token, str) or not token.strip():
+        raise SourceError('Paste a GitHub token')
+    try:
+        credential = _credentials.bearer(API + '/', token.strip())
+    except ValueError as e:
+        raise SourceError('That is not a usable GitHub token: ' + str(e)) from None
+    _credentials.put(CREDENTIAL, credential)
+    with _rejected_lock:
+        _rejected.discard(_digest(credential))  # a fresh chance
+    return {'message': 'GitHub token saved' + (' (%s is set and is used instead)' % ENV if os.environ.get(ENV) else '')}
+
+
+def clear_token():
+    _credentials.remove(CREDENTIAL)
+    return {'message': 'GitHub token removed'}
 
 
 def _curated():
@@ -20,13 +80,33 @@ def _curated():
 
 
 def _api(path):
+    """GET from the GitHub API, with the user's token if there is one. If GitHub rejects
+    the token (401), retry once without it, like Obtainium, and remember the rejection."""
     headers = {'Accept': 'application/vnd.github+json', 'X-GitHub-Api-Version': '2022-11-28'}
-    token = os.environ.get('FRAME_GITHUB_TOKEN')
-    if token:
-        headers['Authorization'] = 'Bearer ' + token
+    token, _ = _token()
+    auth = _credential(token) if token is not None else None
+    with _rejected_lock:
+        if auth and _digest(auth) in _rejected:
+            auth = None  # already refused this run: don't spend a request on it again
+    hint = ' (add a GitHub token in the store\'s source settings, or set %s, to raise the limit)' % ENV
+    if token is not None and not auth:
+        hint = ' (GitHub rejected your token; replace it in the store\'s source settings)'
+
+    def get(auth):
+        return _web.read(API + path, ('api.github.com',), headers, name='GitHub',
+                         hint='' if auth else hint, auth=auth)
     try:
-        return json.loads(_web.read(API + path, ('api.github.com',), headers, name='GitHub',
-                                    hint='' if token else ' (set FRAME_GITHUB_TOKEN to raise the limit)'))
+        try:
+            data = get(auth)
+        except SourceAuthError:
+            if not auth:
+                raise
+            with _rejected_lock:
+                _rejected.add(_digest(auth))
+            print('GitHub rejected the access token (HTTP 401); continuing without it', file=sys.stderr)
+            hint = ' (GitHub rejected your token; replace it in the store\'s source settings)'
+            data = get(None)
+        return json.loads(data)
     except (ValueError, TypeError) as e:
         raise SourceError('Invalid GitHub response') from e
 

ui/apk_sources/search.py

@@ -113,6 +113,14 @@ def manage_repo(action, **kwargs):
     return getattr(module, action)(**kwargs)
 
 
+def set_token(source_id, token):
+    """Save (token) or remove (None) a source's access token; status is in sources()."""
+    module, _ = resolve(source_id)
+    if not hasattr(module, 'set_token'):
+        raise SourceError('This source does not use an access token')
+    return module.set_token(token) if token is not None else module.clear_token()
+
+
 def normalise(name):
     return ' '.join(re.findall(r'\w+', unicodedata.normalize('NFKC', name or '').casefold()))
 

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/frame_touch.py

@@ -25,7 +25,9 @@ Standard library only (ctypes for libei), like the rest of what runs on the Fram
 import ctypes
 import json
 import os
+import re
 import select
+import shutil
 import subprocess
 import sys
 import time
@@ -53,14 +55,41 @@ def say(state, **more):
 
 # ---- which panel has focus -----------------------------------------------------
 
-def xprop_root(display, name):
+PROPS = ("GAMESCOPE_FOCUSABLE_WINDOWS", "GAMESCOPE_FOCUSED_WINDOW", "GAMESCOPE_FOCUS_DISPLAY")
+TOOL_TIMEOUT = 5.0  # seconds for any one xprop or xwininfo
+PANELS_DEADLINE = 12.0  # seconds for a whole `panels` run: well inside the server's 20 s for /api/panels
+DEADLINE = [None]  # when the current `panels` run must stop asking X tools (monotonic), if it has one
+
+
+class OutOfTime(Exception):
+    """The current run's deadline passed: stop and report what was found."""
+
+
+def budget(limit=TOOL_TIMEOUT):
+    """Seconds an X tool may take now: its own limit, cut to what's left of the run's deadline."""
+    if DEADLINE[0] is None:
+        return limit
+    left = DEADLINE[0] - time.monotonic()
+    if left <= 0:
+        raise OutOfTime()
+    return min(limit, left)
+
+
+def read_props(display, limit=TOOL_TIMEOUT):
+    """gamescope's root properties on one display, in one xprop call: {name: [values]} for
+    those the display has, {} if none (or xprop didn't answer)."""
     try:
-        out = subprocess.run(["xprop", "-root", name], env=dict(os.environ, DISPLAY=display),
-                             capture_output=True, text=True, timeout=5).stdout
+        out = subprocess.run(["xprop", "-root", *PROPS], env=dict(os.environ, DISPLAY=display),
+                             capture_output=True, text=True, timeout=budget(limit)).stdout
     except (OSError, subprocess.SubprocessError):
-        return []
-    values = out.split("=", 1)[1] if "=" in out else ""
-    return [int(v) for v in values.replace(",", " ").split() if v.isdigit()]
+        return {}
+    props = {}
+    for line in out.splitlines():
+        name, sep, values = line.partition("=")
+        name = name.split("(", 1)[0].strip()
+        if sep and name in PROPS:
+            props[name] = [int(v) for v in values.replace(",", " ").split() if v.isdigit()]
+    return props
 
 
 def window_info(display, window):
@@ -68,7 +97,7 @@ def window_info(display, window):
     try:
         which = ["-root"] if window == "root" else ["-id", str(window)]
         out = subprocess.run(["xwininfo", *which], env=dict(os.environ, DISPLAY=display),
-                             capture_output=True, text=True, timeout=5).stdout
+                             capture_output=True, text=True, timeout=budget()).stdout
     except (OSError, subprocess.SubprocessError):
         return None
     if "IsViewable" not in out:
@@ -86,13 +115,58 @@ def window_info(display, window):
 
 
 def displays():
-    return sorted(f":{n[1:]}" for n in os.listdir("/tmp/.X11-unix") if n[1:].isdigit())
+    """Every X display on the Frame: the sockets in /tmp/.X11-unix, and the abstract ones
+    X servers on Linux usually listen on as well (in /proc/net/unix), in case a SteamOS
+    build stops sharing /tmp/.X11-unix with SSH sessions."""
+    found = set()
+    try:
+        found.update(int(n[1:]) for n in os.listdir("/tmp/.X11-unix") if n[:1] == "X" and n[1:].isdigit())
+    except OSError:
+        pass
+    try:
+        with open("/proc/net/unix") as f:
+            found.update(int(n) for n in re.findall(r"@/tmp/\.X11-unix/X(\d+)", f.read()))
+    except OSError:
+        pass
+    return [f":{n}" for n in sorted(found)]
+
+
+# The input helper (`frame_touch.py` reading events) lives for a whole Control session and
+# reads focus often, so it remembers which display has gamescope's properties: found, for
+# good; not found, for ROOT_RETRY. A `panels` run is a fresh process each time, so this
+# saves it nothing; it costs nothing either while :0 has the properties.
+_ROOT = []  # [display, when it was given up on, or None if found]
+ROOT_PROBE_TIMEOUT = 2.0  # seconds for each display other than :0, so a stalled one can't use the run
+ROOT_RETRY = 30.0
+
+
+def gamescope():
+    """(display, properties): the X display whose root window carries gamescope's panel
+    properties (GAMESCOPE_FOCUSABLE_WINDOWS and friends), and what they say now.
+
+    That's :0 on every build seen so far (SteamOS 0.3.0 to 0.4.1), so :0 is read first and,
+    when it has them, used with no other probe. Only if it hasn't are the other displays
+    asked, briefly each; with none, :0 and no properties.
+    """
+    if _ROOT and (_ROOT[1] is None or time.monotonic() - _ROOT[1] < ROOT_RETRY):
+        return _ROOT[0], read_props(_ROOT[0])
+    props, found = read_props(":0"), ":0"
+    if not props:
+        found = None
+        for display in displays():
+            if display != ":0":
+                props = read_props(display, ROOT_PROBE_TIMEOUT)
+                if props:
+                    found = display
+                    break
+    _ROOT[:] = [found, None] if found else [":0", time.monotonic()]
+    return _ROOT[0], props
 
 
 def window_pid(display, window):
     try:
         out = subprocess.run(["xprop", "-id", str(window), "_NET_WM_PID"], env=dict(os.environ, DISPLAY=display),
-                             capture_output=True, text=True, timeout=5).stdout
+                             capture_output=True, text=True, timeout=budget()).stdout
     except (OSError, subprocess.SubprocessError):
         return None
     value = out.rsplit("=", 1)[-1].strip() if "=" in out else ""
@@ -120,20 +194,20 @@ def locate(window, pid):
             "root": [root.get("width", info["width"]), root.get("height", info["height"])]}
 
 
-def focusable():
-    """gamescope's focusable windows as (window, app id, pid)."""
-    t = xprop_root(":0", "GAMESCOPE_FOCUSABLE_WINDOWS")
+def focusable(props=None):
+    """gamescope's focusable windows as (window, app id, pid), from props if already read."""
+    t = (gamescope()[1] if props is None else props).get("GAMESCOPE_FOCUSABLE_WINDOWS", [])
     return [tuple(t[i:i + 3]) for i in range(0, len(t) - 2, 3)]
 
 
-def focus_display():
-    """The display of the focused window, from GAMESCOPE_FOCUS_DISPLAY on :0's root.
+def focus_display(props=None):
+    """The display of the focused window, from GAMESCOPE_FOCUS_DISPLAY on gamescope's root.
 
     gamescope writes the name (":1") as 32-bit items, so its first four bytes land,
     little-endian, in the first value: 12602 is 0x313A, ":1" (steamcompmgr.cpp;
     seen 2026-09-29).
     """
-    values = xprop_root(":0", "GAMESCOPE_FOCUS_DISPLAY")
+    values = (gamescope()[1] if props is None else props).get("GAMESCOPE_FOCUS_DISPLAY")
     if not values:
         return None
     name = (values[0] & 0xFFFFFFFF).to_bytes(4, "little").split(b"\0", 1)[0].decode("ascii", "replace")
@@ -141,19 +215,23 @@ def focus_display():
 
 
 def focus_now():
-    """Just which window and display have focus: two property reads, for checking a press."""
-    window = (xprop_root(":0", "GAMESCOPE_FOCUSED_WINDOW") or [0])[0]
-    return (window or None, focus_display() if window else None)
+    """Just which window and display have focus: one property read, for checking a press."""
+    props = gamescope()[1]
+    window = (props.get("GAMESCOPE_FOCUSED_WINDOW") or [0])[0]
+    return (window or None, focus_display(props) if window else None)
 
 
-def focus():
+def focus(props=None):
     """The panel that has focus in the headset: window, display, name and sizes (gamescope
-    publishes the window and its display on :0's root)."""
-    window = (xprop_root(":0", "GAMESCOPE_FOCUSED_WINDOW") or [0])[0]
+    publishes the window and its display on its root display, :0 so far). props: gamescope's
+    properties if already read."""
+    if props is None:
+        props = gamescope()[1]
+    window = (props.get("GAMESCOPE_FOCUSED_WINDOW") or [0])[0]
     if not window:
         return {"window": None}
-    app, pid = next(((a, p) for w, a, p in focusable() if w == window), (None, None))
-    display = focus_display()
+    app, pid = next(((a, p) for w, a, p in focusable(props) if w == window), (None, None))
+    display = focus_display(props)
     info = window_info(display, window) if display else None
     if info:
         root = window_info(display, "root") or {}
@@ -164,18 +242,77 @@ def focus():
     return {**panel, "app": app} if panel else {"window": None}
 
 
-def panels():
-    """Every app panel (gamescope's focusable windows), for watching one that hasn't focus."""
-    now = focus()
-    found = []
-    for window, app, pid in focusable():
-        panel = locate(window, pid)
-        if panel and panel["width"] > 1 and panel["height"] > 1 and \
-                not any(f["window"] == window and f["display"] == panel["display"] for f in found):
-            panel.pop("root", None)
-            found.append({**panel, "app": app, "focused": (window, panel["display"]) ==
-                          (now.get("window"), now.get("display"))})
-    return {"focus": now.get("window"), "focus_display": now.get("display"), "panels": found}
+def panels(limit=None):
+    """Every app panel (gamescope's focusable windows), for watching one that hasn't focus.
+
+    One deadline (PANELS_DEADLINE) covers the whole run, inside the server's timeout: past
+    it, the panels found so far come back with a "why" saying the headset was too slow.
+    """
+    DEADLINE[0] = time.monotonic() + (PANELS_DEADLINE if limit is None else limit)
+    now, found, unfound, late = {}, [], 0, False
+    try:
+        props = gamescope()[1]
+        now = focus(props)
+        for window, app, pid in focusable(props):
+            panel = locate(window, pid)
+            unfound += panel is None
+            if panel and panel["width"] > 1 and panel["height"] > 1 and \
+                    not any(f["window"] == window and f["display"] == panel["display"] for f in found):
+                panel.pop("root", None)
+                found.append({**panel, "app": app, "focused": (window, panel["display"]) ==
+                              (now.get("window"), now.get("display"))})
+        late = time.monotonic() >= DEADLINE[0]  # the last read may have been cut short
+    except OutOfTime:
+        late = True
+    finally:
+        DEADLINE[0] = None
+    out = {"focus": now.get("window"), "focus_display": now.get("display"), "panels": found}
+    if late:
+        out["why"] = ("The headset took too long to list its panels (its X tools didn't answer in time). "
+                      + ("Some may be missing. " if found else "") + "Try again, or use the Headset view.")
+    elif not found:
+        out["why"] = why_none(unfound)
+    return out
+
+
+# The headset's Linux desktop, by process name (docs/how-the-frame-works.md).
+DESKTOP = ("plasmashell", "kwin_wayland", "kwin_wayland_wrapper", "steamos-nested-desktop", "startplasma-wayland")
+
+
+def desktop_running():
+    """Whether the headset's Linux desktop (the nested Plasma session) is running."""
+    try:
+        pids = [p for p in os.listdir("/proc") if p.isdigit()]
+    except OSError:
+        return False
+    for pid in pids:
+        try:
+            with open(f"/proc/{pid}/comm") as f:
+                if f.read().strip() in DESKTOP:
+                    return True
+        except OSError:
+            continue
+    return False
+
+
+def why_none(listed):
+    """Why panels() found nothing, in words the person at the computer can act on.
+    listed: how many of the windows gamescope lists aren't X11 windows on any display."""
+    missing = [tool for tool in ("xprop", "xwininfo") if not shutil.which(tool)]
+    if missing:
+        return (f"This SteamOS build has no {' or '.join(missing)}, which Frame Control uses to find the "
+                "headset's panels. Use the Headset view instead, and please report your SteamOS version.")
+    if listed:
+        # gamescope lists windows, but none is an X11 window on any display: a Wayland
+        # window (a desktop drawn straight to gamescope, say) can't be found or captured this way.
+        return (f"The headset has {listed} panel{'s' if listed > 1 else ''} open, but Frame Control can't "
+                f"find {'its' if listed == 1 else 'their'} window. This SteamOS build may draw "
+                f"{'it' if listed == 1 else 'them'} in a way the Desktop view can't capture yet. "
+                "Use the Headset view instead, and please report your SteamOS version.")
+    if desktop_running():
+        return ("The headset's desktop is running, but SteamOS isn't listing it as a panel. Look at it in "
+                "the headset (or close and reopen it), then try again, or use the Headset view.")
+    return "No app panel is open in the headset. Open the desktop or an app in the headset, then try again."
 
 
 def to_root(panel, fx, fy):

ui/index.html

@@ -611,6 +611,8 @@
   .source-repo-form input { margin-top:7px; display:block; border-radius:6px; }
   .source-repo-form details { font-size:11px; color:var(--muted); }
   .source-repo-form button { justify-self:start; border-radius:6px; }
+  .source-token { display:flex; flex-wrap:wrap; align-items:center; gap:7px; margin:8px 0 4px; }
+  .source-token input { flex:1; min-width:150px; border-radius:6px; font-size:12px; }
   #appStore :focus-visible, .store-dialog :focus-visible, .store-tabs :focus-visible { outline:2px solid var(--link); outline-offset:4px; }
   .store-search input:focus-visible { outline:0!important; }
   @media(min-width:1400px) { .store-grid { grid-template-columns:repeat(5,minmax(0,1fr)); } }
@@ -700,6 +702,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>
@@ -1206,7 +1209,7 @@
   <button class="store-close" data-close="sourcesDialog" aria-label="Close source settings">×</button>
   <div class="sources-intro"><div class="store-eyebrow">Your app shelves</div><h1 id="sourcesTitle">A world of apps.</h1><p>Choose where you discover games and apps. We bring them together for you.</p></div>
   <div id="sourcesList"></div>
-  <form id="addSource" class="source-repo-form"><h2>Add a source</h2><p>Have a favorite F-Droid repository? Add its address to browse its apps here.</p><label>Repository address<input name="url" type="url" required placeholder="https://example.org/fdroid/repo"></label><details><summary>Verify this source (optional)</summary><label>Repository fingerprint<input name="fingerprint" type="text" autocomplete="off" placeholder="Paste the fingerprint from the repository owner"></label></details><button class="action">Add source</button></form>
+  <form id="addSource" class="source-repo-form"><h2>Add a source</h2><p>Have a favorite F-Droid repository? Add its address to browse its apps here.</p><label>Repository address<input name="url" type="url" required placeholder="https://example.org/fdroid/repo"></label><details><summary>Verify this source (optional)</summary><label>Repository fingerprint<input name="fingerprint" type="text" autocomplete="off" placeholder="Paste the fingerprint from the repository owner"></label></details><details><summary>This repository needs a login (optional)</summary><label>Username<input name="username" type="text" autocomplete="off" autocapitalize="none" spellcheck="false"></label><label>Password<input name="password" type="password" autocomplete="new-password"></label><p>Kept on this computer, apart from the address, and only sent to this repository over HTTPS.</p></details><button class="action">Add source</button></form>
   <p id="sourceError" class="store-notice" role="status" aria-live="polite"></p>
 </dialog>
 
@@ -1358,6 +1361,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>
@@ -1889,7 +1904,7 @@ async function capture() {
     var shotPanel = null;
     if (view === "flat") {
       shotPanel = deskPanel() || (await loadPanels(), deskPanel());
-      if (!shotPanel) throw new Error("no app panel is open in the headset");
+      if (!shotPanel) throw new Error(noPanelWhy());
       url_ = `/api/screenshot?view=panel&window=${shotPanel.window}&display=${encodeURIComponent(shotPanel.display)}`;
     }
     const r = await fetch(url_, { headers: {"X-Frame-UI": UI_KEY} });
@@ -2101,8 +2116,14 @@ document.querySelectorAll("[data-eye]").forEach(b => b.onclick = () => setEye(b.
 // Control reaches, so the view follows it unless you pick another to watch.
 // Panels are "display/window": window ids are per X display, and the Frame has two.
 // desk.shown is the panel whose picture is on screen now; Control only acts on that one.
-const desk = { panels: [], focus: null, pick: "", timer: null, loaded: false, shown: null };
+// desk.why is the headset's own reason when it lists no panels; desk.error, why listing them failed.
+const desk = { panels: [], focus: null, pick: "", timer: null, loaded: false, shown: null, why: "", error: "" };
 const panelKey = p => p ? `${p.display}/${p.window}` : null;
+// Why the Desktop view has no panel to show: the headset's own reason when it gave one.
+function noPanelWhy() {
+  if (desk.error) return "Couldn't list the headset's panels: " + desk.error;
+  return desk.why || "No app panel is open in the headset. Open the desktop or an app in the headset, then try again.";
+}
 function deskPanel() {
   const want = desk.pick || desk.focus;
   // Nothing in use and nothing picked: show the first panel rather than nothing.
@@ -2112,7 +2133,8 @@ async function loadPanels() {
   try {
     const before = panelKey(deskPanel());
     const r = await api("/api/panels");
-    desk.panels = r.panels || []; desk.loaded = true;
+    desk.panels = r.panels || []; desk.loaded = true; desk.error = "";
+    desk.why = r.why || "";
     desk.focus = r.focus ? `${r.focus_display}/${r.focus}` : null;
     if (desk.pick && !desk.panels.some(p => panelKey(p) === desk.pick)) desk.pick = "";
     renderPanelPick();
@@ -2120,7 +2142,11 @@ async function loadPanels() {
     if (view === "flat" && live && now !== before) { toggleLive(false); toggleLive(true); }
     ctrlShow();
   } catch (e) {
-    log("Couldn't list the headset's panels: " + e.message, "e");
+    if (e.stale) return;  // an answer about the previous headset says nothing about this one
+    // Polled every few seconds: waiting for the Frame, or the same failure again, isn't news.
+    if (!e.offline && e.message !== desk.error) log("Couldn't list the headset's panels: " + e.message, "e");
+    desk.error = e.message;
+    ctrlShow();
   }
 }
 function watchPanels() {
@@ -2165,8 +2191,7 @@ function ctrlMode() { return view === "flat" ? "abs" : "rel"; }
 function ctrlTarget() {
   if (ctrlMode() === "rel") return { ok: true };
   const p = deskPanel();
-  if (!desk.loaded) return { why: "Finding the headset's panels…" };
-  if (!p) return { why: "No app panel is open in the headset." };
+  if (!p) return { why: desk.loaded || desk.error ? noPanelWhy() : "Finding the headset's panels…" };
   if (!desk.focus) return { why: "Nobody's using a panel in the headset yet. Taps work once one is in use." };
   if (panelKey(p) !== desk.focus) return { why: "Only the panel in use takes taps. Pick “Panel in use” to control it." };
   if (desk.shown !== desk.focus) return { why: live ? "Waiting for the picture of the panel in use…"
@@ -3116,12 +3141,18 @@ function storeIcon(o) { return `<span class="store-icon" style="--hue:${storeHue
 document.addEventListener("error", e => { if (e.target.matches?.(".store-art img, .store-icon img")) e.target.remove(); }, true);
 function sourceMark(s) { const name=storeName(s); return `<span class="offer-mark" style="--hue:${storeHue(name)}" aria-hidden="true">${esc(name === "SideQuest" ? "SQ" : name === "F-Droid" ? "Fd" : name.slice(0,2))}</span>`; }
 function sourceDescription(s) { return s.description || ({fdroid:"Independent, open-source apps",sidequest:"Discover something new in VR",itch:"Games from independent creators",github:"Direct from the people who make it"})[s.kind] || (s.name.includes("GitHub") ? "Direct from the people who make it" : s.name.includes("Droid") ? "Independent, open-source apps" : s.name.includes("SideQuest") ? "Discover something new in VR" : s.name.includes("itch") ? "Games from independent creators" : "Apps shared by this community"); }
+// GitHub's optional access token: only whether one is set is ever shown, never its value.
+function sourceToken(s) {
+  const t=s.token;if(!t)return "";
+  const note=t.rejected ? "GitHub rejected this token, so searches continue without it. Replace or clear it." : t.from==="env" ? "Using the token from FRAME_GITHUB_TOKEN." : t.set ? "A token is saved." : "Optional: a token raises GitHub's limit of 60 requests an hour.";
+  return `<form class="source-token" data-token="${esc(s.id)}"><input name="token" type="password" autocomplete="off" spellcheck="false" aria-label="${esc(storeName(s.name))} access token" placeholder="${t.set ? "Replace access token" : "Access token"}"><button class="small">Save</button>${t.from==="saved" ? `<button type="button" class="small" data-clear-token="${esc(s.id)}">Clear</button>` : ""}</form><small>${esc(note)}</small>`;
+}
 async function loadSources() {
   try {
     const {sources} = await api("/api/sources"); sourceState.sources = sources;
     $("sourceFilter").innerHTML = `<option value="">All sources</option>` + sources.filter(s=>s.enabled).map(s=>`<option value="${esc(s.id)}">${esc(storeName(s.name))}</option>`).join("");
     $("sourceFilter").value = sourceState.source;
-    $("sourcesList").innerHTML = sources.map(s=>`<div class="source-setting">${sourceMark(s.name)}<div class="grow"><strong>${esc(storeName(s.name))}</strong><p>${esc(sourceDescription(s))}</p><small>${esc(({official:"Official store",community:"Community",user:"Added by you"})[s.trust] || "Community")}</small>${s.builtin === false ? `<button class="small" data-remove="${esc(s.id)}" aria-label="Remove ${esc(s.name)}">Remove</button>` : ""}</div><input class="source-toggle" role="switch" type="checkbox" aria-label="Enable ${esc(storeName(s.name))}" data-enable="${esc(s.id)}" ${s.enabled ? "checked" : ""}></div>`).join("") || `<p class="store-notice">Your sources will appear here when they are available.</p>`;
+    $("sourcesList").innerHTML = sources.map(s=>`<div class="source-setting">${sourceMark(s.name)}<div class="grow"><strong>${esc(storeName(s.name))}</strong><p>${esc(sourceDescription(s))}</p><small>${esc(({official:"Official store",community:"Community",user:"Added by you"})[s.trust] || "Community")}${s.login ? " · Uses a login" : ""}</small>${sourceToken(s)}${s.builtin === false ? `<button class="small" data-remove="${esc(s.id)}" aria-label="Remove ${esc(s.name)}">Remove</button>` : ""}</div><input class="source-toggle" role="switch" type="checkbox" aria-label="Enable ${esc(storeName(s.name))}" data-enable="${esc(s.id)}" ${s.enabled ? "checked" : ""}></div>`).join("") || `<p class="store-notice">Your sources will appear here when they are available.</p>`;
   } catch (e) { $("sourceError").textContent = e.message; }
 }
 function storeCard(app) {
@@ -3172,7 +3203,19 @@ async function searchSources(quiet) {
 }
 function currentOffer() { return sourceState.detail?.app.offers[sourceState.detail.choice]; }
 function detailBadges(o) { return `<span class="store-badge">${o.vr ? "VR experience" : o.vr===false ? "Flat-screen app" : "Android app"}</span>${o.verdict?.tone==='works' ? '<span class="store-badge works">✓ Works on Frame</span>' : ''}<span class="store-badge">Free</span>${storeOpenSource(o) ? '<span class="store-badge">Open source</span>' : ''}`; }
-function renderDetail() {
+// The store's app page. (Not renderDetail: the Devices page declares that further down,
+// and a later function declaration silently replaces an earlier one; issue #68.)
+// True when the page rendered; false when it shows the error instead.
+function renderAppDetail() {
+  try { renderAppDetailBody(); return true; }
+  catch(e) {
+    console.error("App details failed to render", e);
+    const name=sourceState.detail?.app?.name;
+    $("detailBody").innerHTML=`<div class="store-empty detail-error" role="alert"><h2 id="detailTitle">${name ? `We couldn't show ${esc(name)}` : "We couldn't show this app"}</h2><p>Something went wrong opening its page. Close this and try again, or pick another app.</p><button class="action" data-close="appDetail">Close</button></div>`;
+    return false;
+  }
+}
+function renderAppDetailBody() {
   const restoreOfferFocus=document.activeElement?.name==='storeOffer';
   const d=sourceState.detail;if(!d)return;const o=currentOffer(), images=o.artwork?.screenshots||[], page=sourcePage(o.page);
   $("detailBody").innerHTML=storeArt(o,"detail-banner",true)+`<div class="detail-content"><div class="detail-heading">${storeIcon(o)}<div><h1 id="detailTitle">${esc(o.name)}</h1><p>${esc(o.developer || o.author || 'From '+storeName(o.source_name))}</p></div></div><div class="store-badges">${detailBadges(o)}</div><div class="detail-layout"><div><div class="detail-gallery" id="detailGallery" aria-label="App screenshots" tabindex="0">${images.map((p,i)=>storeArt(o,"",false,p,`${o.name} screenshot ${i+1}`)).join("")}</div>${images.length ? `<div class="gallery-controls"><span>${images.length} screenshots · scroll to explore</span><div><button data-gallery="-1" aria-label="Previous screenshot">←</button> <button data-gallery="1" aria-label="Next screenshot">→</button></div></div>` : ''}<h2>About this app</h2><div class="detail-description">${esc(storeText(o.description || o.summary || "Discover this app on its developer's page."))}</div>${page ? `<p><a href="${esc(page)}" target="_blank" rel="noopener noreferrer">Meet the creators ↗</a></p>` : ''}<details class="detail-technical"><summary>Technical details</summary><dl><dt>Package</dt><dd>${esc(o.package||"Unknown")}</dd><dt>Version</dt><dd>${esc(o.version||"Unknown")} (${esc(o.version_code??"unknown")})</dd><dt>Android API</dt><dd>${esc(o.min_sdk??"Unknown")}</dd><dt>Architecture</dt><dd>${esc(o.abis?.join(', ')||"Unknown / no native code")}</dd><dt>Updated</dt><dd>${esc(o.updated||"Unknown")}</dd><dt>Compatibility</dt><dd>${(o.fit?.reasons||[]).map(esc).join(' · ')||'No known installation blockers'}</dd></dl></details></div><aside class="detail-buy"><div class="detail-price">Free</div><p class="detail-small">No purchase needed.</p><p class="detail-verdict ${esc(o.verdict?.tone||'unknown')}">${esc(o.verdict?.label||'Not yet checked on the Frame')}</p><button id="detailInstall" class="action store-primary">Install</button><progress id="detailProgress" max="100" hidden aria-label="Installation progress"></progress><p id="detailInstallNote" class="detail-small" role="status" aria-live="polite"></p><h2>Get it from</h2><div id="detailOffers">${d.app.offers.map((v,j)=>`<label class="offer-choice"><input type="radio" name="storeOffer" value="${j}" ${j===d.choice?'checked':''}><span>${sourceMark(v.source_name)}</span><span><strong>${esc(storeName(v.source_name))}${j===0?' · Recommended':''}</strong><small>${v.verified===true?'✓ Verified download':'Verification not provided'}</small><small>${esc(v.version ? 'v'+v.version : 'Latest release')}${v.size!=null?' · '+(v.size/1048576).toFixed(0)+' MB':''}</small></span></label>`).join('')}</div></aside></div></div>`;
@@ -3192,8 +3235,8 @@ function updateInstall() {
 async function openStoreApp(index) {
   const app=sourceState.apps[index];if(!app)return;
   sourceState.detail={app:{...app,offers:app.offers.map(o=>({...o}))},choice:0};
-  renderDetail();if(!$("appDetail").open)$("appDetail").showModal();$("appDetail").scrollTop=0;
-  await loadOfferDetails();
+  const shown=renderAppDetail();if(!$("appDetail").open)$("appDetail").showModal();$("appDetail").scrollTop=0;
+  if(shown)await loadOfferDetails();
 }
 async function loadOfferDetails() {
   const d=sourceState.detail,o=currentOffer(), request=++sourceState.detailRequest;
@@ -3201,8 +3244,8 @@ async function loadOfferDetails() {
     const detail=await api('/api/sources/details?'+new URLSearchParams({source:o.source,id:o.id}));
     if(request!==sourceState.detailRequest || d!==sourceState.detail)return;
     const artwork={icon:detail.artwork?.icon||o.artwork?.icon,banner:detail.artwork?.banner||o.artwork?.banner,screenshots:detail.artwork?.screenshots?.length?detail.artwork.screenshots:o.artwork?.screenshots||[]};
-    d.app.offers[d.choice]={...o,...detail,artwork};renderDetail();
-  } catch(e) { if(d===sourceState.detail)$("detailInstallNote").textContent="Some app details couldn't be refreshed. You can still view this listing."; }
+    d.app.offers[d.choice]={...o,...detail,artwork};renderAppDetail();
+  } catch(e) { const note=$("detailInstallNote");if(d===sourceState.detail && note)note.textContent="Some app details couldn't be refreshed. You can still view this listing."; }
 }
 async function installStoreApp() {
   const o=currentOffer(), key=`${o.source}:${o.id}`, existing=sourceState.jobs.get(key);
@@ -3232,11 +3275,12 @@ $("sourceSearch").onsubmit=e=>{e.preventDefault();clearTimeout(storeSearchTimer)
 $("sourceFilter").onchange=e=>{sourceState.source=e.target.value;searchSources();};
 $("searchFilters").onclick=e=>{const b=e.target.closest('[data-filter]');if(!b)return;const f=b.dataset.filter;if(f==='all'){sourceState.vr=null;sourceState.installable=false;}else if(f==='installable')sourceState.installable=!sourceState.installable;else sourceState.vr=sourceState.vr===(f==='vr')?null:f==='vr';storeChips();searchSources();};
 $("searchGrid").onclick=e=>{const b=e.target.closest('[data-app]');if(b)openStoreApp(+b.dataset.app);if(e.target.closest('#storeReset')){$('sourceQ').value='';sourceState.vr=null;sourceState.installable=false;sourceState.source='';storeChips();searchSources();}};
-$("detailBody").onclick=e=>{if(e.target.closest('#detailInstall'))installStoreApp();const b=e.target.closest('[data-gallery]');if(b)$('detailGallery').scrollBy({left:+b.dataset.gallery*$('detailGallery').clientWidth*.86,behavior:matchMedia('(prefers-reduced-motion: reduce)').matches?'instant':'smooth'});};
-$("detailBody").onchange=e=>{if(e.target.name==='storeOffer'){sourceState.detail.choice=+e.target.value;renderDetail();loadOfferDetails();}};
+$("detailBody").onclick=e=>{if(e.target.closest('[data-close]'))return $("appDetail").close();if(e.target.closest('#detailInstall'))installStoreApp();const b=e.target.closest('[data-gallery]');if(b)$('detailGallery').scrollBy({left:+b.dataset.gallery*$('detailGallery').clientWidth*.86,behavior:matchMedia('(prefers-reduced-motion: reduce)').matches?'instant':'smooth'});};
+$("detailBody").onchange=e=>{if(e.target.name==='storeOffer'){sourceState.detail.choice=+e.target.value;renderAppDetail();loadOfferDetails();}};
 $("sourcesOpen").onclick=()=>{loadSources();$('sourcesDialog').showModal();};
 $("sourcesList").onchange=e=>{if(e.target.matches('[data-enable]'))changeSource({action:'enable',source:e.target.dataset.enable,enabled:e.target.checked});};
-$("sourcesList").onclick=e=>{const b=e.target.closest('[data-remove]');if(b)changeSource({action:'remove',source:b.dataset.remove});};
+$("sourcesList").onclick=e=>{const b=e.target.closest('[data-remove]');if(b)changeSource({action:'remove',source:b.dataset.remove});const c=e.target.closest('[data-clear-token]');if(c)changeSource({action:'clear-token',source:c.dataset.clearToken});};
+$("sourcesList").onsubmit=e=>{const f=e.target.closest('[data-token]');if(!f)return;e.preventDefault();const token=f.elements.token.value.trim();f.reset();if(token)changeSource({action:'set-token',source:f.dataset.token,token});};
 async function addSource(body) {
   $("sourceError").textContent="Checking the repository. Large ones can take a minute…";
   let error="";
@@ -3244,7 +3288,49 @@ async function addSource(body) {
   $("sourceError").textContent=result ? result.message : error || "We couldn't add that repository.";
   if(result){$("addSource").reset();sourceState.source='';await loadSources();await searchSources();}
 }
-$("addSource").onsubmit=e=>{e.preventDefault();const f=new FormData(e.target);addSource({action:'add',url:f.get('url'),fingerprint:f.get('fingerprint')});};
+// A login typed into the address moves into the login fields, on the page too, before anything
+// is sent: the address is shown and logged, the login never. It stays there if adding fails.
+// Plain text, not the URL parser: an address it rejects (or a login it can't decode) must
+// still lose its login first. Decoding falls back to the text as typed.
+// Percent-decode what decodes; keep any byte sequence that isn't UTF-8 as typed.
+// One forward pass: each lead byte gives its sequence length (at most 4 bytes).
+function decodeLenient(s) {
+  return s.replace(/(?:%[0-9a-f]{2})+/gi, run => {
+    const b=run.match(/%[0-9a-f]{2}/gi); let out='', i=0;
+    while(i<b.length){
+      const v=parseInt(b[i].slice(1),16);
+      const n=v<0x80 ? 1 : v>=0xC2&&v<0xE0 ? 2 : v>=0xE0&&v<0xF0 ? 3 : v>=0xF0&&v<0xF5 ? 4 : 0;
+      if(n){ try { out+=decodeURIComponent(b.slice(i,i+n).join('')); i+=n; continue; } catch {} }
+      out+=b[i]; i++;
+    }
+    return out;
+  });
+}
+// Moves a login typed into the address into the login fields. Returns false for an
+// address the URL parser rejects that may hold a login: it is refused, not sent.
+function moveLogin(form) {
+  // Normalise the way URL parsers do: trim C0 controls and spaces, drop tabs and newlines.
+  const el=form.elements;
+  el.url.value=el.url.value.replace(/^[\u0000- ]+|[\u0000- ]+$/g,'').replace(/[\t\n\r]/g,'');
+  let u;
+  // A backslash and an '@' together are read differently by browsers and by the
+  // server, so the address can't be split safely: refuse it rather than guess.
+  if(el.url.value.includes('@') && el.url.value.includes('\\'))return false;
+  try { u=new URL(el.url.value); } catch { return !el.url.value.includes('@'); }
+  if(!u.username && !u.password)return true;
+  const user=decodeLenient(u.username), pass=decodeLenient(u.password);
+  u.username=''; u.password=''; el.url.value=u.href;
+  if(!el.username.value)el.username.value=user;
+  if(!el.password.value)el.password.value=pass;
+  el.username.closest('details').open=true;
+  return true;
+}
+$("addSource").onsubmit=e=>{
+  e.preventDefault();
+  if(!moveLogin(e.target)){ $("sourceError").textContent="That address isn't valid. Put the username and password in the login fields below, not in the address."; return; }
+  const f=new FormData(e.target), url=f.get('url'), username=f.get('username'), password=f.get('password');
+  addSource({action:'add',url,fingerprint:f.get('fingerprint'),...(username ? {username,password} : {})});
+};
 document.querySelectorAll('[data-close]').forEach(b=>b.onclick=()=>$(b.dataset.close).close());
 $('appDetail').addEventListener('close',()=>{sourceState.detailRequest++;});
 document.querySelectorAll('[data-store-tab]').forEach(b=>b.onclick=()=>{const browse=b.dataset.storeTab==='browse';$('appStore').hidden=!browse;$('androidLibrary').hidden=browse;$('repNew').hidden=browse;$('sourcesOpen').hidden=!browse;document.querySelectorAll('[data-store-tab]').forEach(t=>{t.classList.toggle('on',t===b);t.setAttribute('aria-pressed',t===b);});if(!browse)loadAndroid();});
@@ -3785,7 +3871,7 @@ document.addEventListener("visibilitychange", () => { if (!document.hidden && on
 
 // ---- Panel switcher: our UI over SteamVR's panel API ----
 let panelSeq = 0;
-async function loadPanels() {
+async function loadPanelSwitcher() {
   const seq = ++panelSeq;
   try {
     const s = await api("/api/panels", {action: "list"});
@@ -3796,7 +3882,7 @@ async function loadPanels() {
       : '<div class="sub">No open panels.</div>';
   } catch (e) { if (seq === panelSeq) failed($("panelList"), e); }
 }
-$("panelsRefresh").onclick = loadPanels;
+$("panelsRefresh").onclick = loadPanelSwitcher;
 $("panelsOpen").onclick = e => act("Open panel switcher in headset",
   () => api("/api/panels", {action: "open"}), e.currentTarget);
 $("panelList").onclick = e => {
@@ -3894,7 +3980,7 @@ function showPage() {
   document.title = page === "home" ? "Frame Control" : `${page[0].toUpperCase() + page.slice(1)} · Frame Control`;
   const section = !PAGES.includes(id) && id && $(id);
   if (section) section.scrollIntoView(); else window.scrollTo(0, 0);
-  if (page === "tools") { loadMacView(); loadPanels(); }
+  if (page === "tools") { loadMacView(); loadPanelSwitcher(); }
 }
 window.addEventListener("hashchange", showPage);
 // While a text field has focus, phones hide the bottom tab bar (see body.typing in the CSS).
@@ -4520,6 +4606,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;
@@ -4620,6 +4710,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();
 }
 
@@ -4779,8 +4872,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>
@@ -4801,10 +4901,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>`);
@@ -4837,7 +4939,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();
@@ -4890,6 +4994,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."),
 ]
@@ -697,7 +702,7 @@ PASTE = r"""set -u
 text=$(cat; printf x); text=${text%x}
 pid=$(pgrep -u "$(id -u)" -x plasmashell | head -n 1)
 if [ -z "$pid" ]; then
-  echo "plasmashell is not running: open the desktop in the headset first." >&2
+  echo "The headset's desktop isn't open (plasmashell is not running). Open the desktop in the headset, then try again." >&2
   exit 2
 fi
 bus=$(tr '\0' '\n' < "/proc/$pid/environ" | sed -n 's/^DBUS_SESSION_BUS_ADDRESS=//p')
@@ -709,6 +714,7 @@ else
   exit 2
 fi
 """
+DESKTOP_CLOSED = "plasmashell is not running"  # what PASTE says when the headset's desktop isn't open
 # base64 keeps the script intact through every local shell's quoting rules.
 PASTE_CMD = 'bash -c "$(echo %s | base64 -d)"' % base64.b64encode(PASTE.encode()).decode()
 
@@ -725,7 +731,13 @@ def clipboard(body):
         text = body.get("text")
         if not isinstance(text, str) or not text:
             raise Failure("nothing to send", 400)
-    return {"message": ssh(PASTE_CMD, stdin=text, timeout=30).strip()}
+    try:
+        return {"message": ssh(PASTE_CMD, stdin=text, timeout=30).strip()}
+    except Failure as e:
+        if DESKTOP_CLOSED in str(e):
+            # The headset's state, for the person to change, not a fault: a 409 isn't reported as one.
+            raise Failure(str(e), 409) from None
+        raise
 
 
 # ---- keyboard and pointer (KDE Connect on the Frame, see frame_input_agent.py) ----
@@ -2141,6 +2153,16 @@ def source_text(body, key, optional=False):
     return value.strip()
 
 
+def login_text(body, key):
+    """A username, password or token: kept exactly (no stripping) and never echoed back."""
+    value = body.get(key)
+    if value in (None, ''):
+        return None
+    if not isinstance(value, str) or len(value) > 4096:
+        raise Failure('Provide a valid ' + key, 400)
+    return value
+
+
 def source_search(query):
     args = parse_qs(query)
     q = args.get('q', [''])[0]
@@ -2179,19 +2201,31 @@ def source_manage(body):
         if action == 'add':
             url = source_text(body, 'url')
             fingerprint, name = source_text(body, 'fingerprint', True), source_text(body, 'name', True)
-            parts = urlparse(url)
-            if parts.scheme not in ('https', 'fdroidrepos') or not parts.hostname or parts.username:
-                raise Failure('Use an HTTPS or fdroidrepos:// repository URL without credentials', 400)
+            username, password = login_text(body, 'username'), login_text(body, 'password')
+            try:
+                parts = urlparse(url)
+                parts.port
+            except ValueError:
+                raise Failure('Use an HTTPS or fdroidrepos:// repository URL', 400) from None
+            if parts.scheme not in ('https', 'fdroidrepos') or not parts.hostname:
+                raise Failure('Use an HTTPS or fdroidrepos:// repository URL', 400)
             apk_search.repo_module()  # fail now if this build can't manage repositories
 
             def add():  # downloads and verifies the whole index: a job, not a request
-                source = apk_search.manage_repo('add_repo', url=url, fingerprint=fingerprint, name=name)
+                # A login (here or in the URL) is stored apart and never comes back in a response.
+                login = {'username': username, 'password': password} if username or password else {}
+                source = apk_search.manage_repo('add_repo', url=url, fingerprint=fingerprint, name=name, **login)
                 message = 'Added ' + source['name']
                 if source.get('trust_on_first_use'):
                     message += '. Trusted on first use: ' + source['fingerprint'].upper()
                 return {'message': message, 'source': {k: source.get(k) for k in
                                                        ('id', 'name', 'fingerprint', 'trust_on_first_use')}}
             return start_job('Add repository', add)
+        if action in ('set-token', 'clear-token'):
+            token = login_text(body, 'token') if action == 'set-token' else None
+            if action == 'set-token' and not token:
+                raise Failure('Paste a token', 400)
+            return apk_search.set_token(source_text(body, 'source'), token)
         if action == 'game-data':
             package = source_text(body, 'package')
             return start_job('Add game data', lambda: apk_search.add_game_data(package))