1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
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"}
|