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