fix(client): link a device from the application; approve one match only

"Get a linking code" called the connection the page had just closed on the
node's unknown_device refusal, so it never filed anything. It now retries
with the request filed on that connection, before the proof (MNP 6.3), and
the page tries the group again after each refusal until a linked device
approves the code. Offered only in the desktop application: a browser's
minted identity is gone by its next connection.

Approving refuses a code that matches more than one pending request:
filing needs only the account's token, so somebody holding the passphrase
who saw a code could file their own keys under it.

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

1 files changed, 106 insertions, 0 deletions

diff --git a/packages/meshbay-hub/tests/test_device_approval.py b/packages/meshbay-hub/tests/test_device_approval.py
new file mode 100644
index 0000000..529f7c6
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_device_approval.py
@@ -0,0 +1,106 @@
+"""
+Approving a device by its code, in the shipped `approveDevice` (transport-devices.js).
+
+A device request is filed before the proof, with the account's token alone
+(docs/MESHBAY_DESIGN.md §3.3, MNP 6.3). So somebody holding the passphrase who
+saw a code on screen can file a second request with that code and keys of their
+own, and both hashes are right. The approving device must then approve neither:
+taking the first match would hand the choice to whoever filed first, or pushed
+the other out.
+
+Run under node against requests hashed by the shipped Python
+(`meshbay_common.device.device_code_hash`), with the node's answer and the
+countersignature stood in for.
+"""
+
+import base64
+import json
+import secrets
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+from meshbay_common.device import device_code_hash
+from spa_source import transport_argv
+
+STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
+CRYPTO = STATIC / "crypto.js"
+
+pytestmark = pytest.mark.skipif(shutil.which("node") is None, reason="node is not available")
+
+CODE = "K7P2-9WQX"
+
+_HARNESS = r"""
+const fs = require('fs');
+globalThis.window = globalThis;
+globalThis.addEventListener = () => {};
+globalThis.removeEventListener = () => {};
+globalThis.location = { hash: '' };
+globalThis.document = {
+  addEventListener() {}, removeEventListener() {}, visibilityState: 'visible',
+};
+globalThis.localStorage = { getItem() { return null; }, setItem() {} };
+new Function(fs.readFileSync(process.argv[2], 'utf8'))();
+const T = new Function(
+  process.argv[3].split(require('path').delimiter)
+    .map((p) => fs.readFileSync(p, 'utf8')).join('\n')
+  + '\nreturn { MeshBayTransport };')();
+
+(async () => {
+  const input = JSON.parse(fs.readFileSync(process.argv[4], 'utf8'));
+  const t = Object.create(T.MeshBayTransport.prototype);
+  t._identity = { pkEdB64: 'me', pkXB64: 'me' };
+  t._nonceNode = new Uint8Array(32);
+  t.nodePk = 'node';
+  t._sendAndWait = async () => ({ type: 'device_lookup_result', requests: input.requests });
+  t._countersign = async (userId, codeHash, pkEd) => ({ approved: pkEd });
+  try {
+    process.stdout.write(JSON.stringify(await t.approveDevice('u-1', input.code)));
+  } catch (e) {
+    process.stdout.write(JSON.stringify({ refused: e.message }));
+  }
+})().catch((e) => { console.error(e); process.exit(1); });
+"""
+
+
+def _keys() -> tuple[str, str]:
+    return (base64.b64encode(secrets.token_bytes(32)).decode(),
+            base64.b64encode(secrets.token_bytes(32)).decode())
+
+
+def _request(code: str) -> dict:
+    ed, x = _keys()
+    return {"pk_ed25519": ed, "pk_x25519": x, "created_at": "2026-10-14T20:00:00+00:00",
+            "code_hash": device_code_hash(code.replace("-", ""), ed, x)}
+
+
+def _approve(tmp_path, requests: list[dict], code: str = CODE) -> dict:
+    h = tmp_path / "h.js"
+    h.write_text(_HARNESS, encoding="utf-8")
+    payload = tmp_path / "in.json"
+    payload.write_text(json.dumps({"requests": requests, "code": code}), encoding="utf-8")
+    run = subprocess.run(["node", str(h), str(CRYPTO), transport_argv(), str(payload)],
+                         capture_output=True, timeout=60)
+    assert run.returncode == 0, run.stderr.decode()[-2000:]
+    return json.loads(run.stdout.decode())
+
+
+def test_the_one_request_with_the_code_is_approved(tmp_path):
+    mine = _request(CODE)
+    out = _approve(tmp_path, [_request("ZZZZ-ZZZZ"), mine, _request("AAAA-1111")])
+    assert out == {"approved": mine["pk_ed25519"]}
+
+
+def test_two_requests_with_the_same_code_are_both_refused(tmp_path):
+    """Whichever came first: the person's own, or one filed by somebody who saw it."""
+    first, second = _request(CODE), _request(CODE)
+    for requests in ([first, second], [second, first]):
+        out = _approve(tmp_path, requests)
+        assert "approved" not in out, out
+        assert "More than one device" in out["refused"]
+
+
+def test_no_request_with_the_code_approves_nothing(tmp_path):
+    out = _approve(tmp_path, [_request("ZZZZ-ZZZZ")])
+    assert out == {"refused": "No device is waiting with that code"}