1 files changed, 100 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_challenge_signature_client.py b/packages/meshbay-hub/tests/test_challenge_signature_client.py
new file mode 100644
index 0000000..90aa5f2
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_challenge_signature_client.py
@@ -0,0 +1,100 @@
+"""
+The browser checks the node's challenge signature with the code it ships (MNP 3.4).
+
+The node signs `handshake_challenge` so a client can hold it to the key it
+announces *before* a join, which goes out ahead of the ack. The Python side is
+tested against a real connection in the node suite; this runs the other half —
+the real `_challengeProvesNodeKey` out of `transport.js` over the real
+`crypto.js` — against a challenge Python signed, because the two agreeing on
+paper (the parity test) is not the browser accepting what the node sends.
+"""
+
+import base64
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.crypto import pk_to_b64
+from meshbay_common.handshake import challenge_transcript, webrtc_binding
+
+STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
+
+pytestmark = pytest.mark.skipif(shutil.which("node") is None,
+                                reason="node is not available")
+
+GROUP = "g" * 32
+
+_HARNESS = r"""
+const fs = require('fs');
+globalThis.window = globalThis;
+globalThis.addEventListener = () => {};
+globalThis.removeEventListener = () => {};
+globalThis.location = { hash: '' };
+globalThis.document = {
+  addEventListener() {}, removeEventListener() {}, visibilityState: 'visible',
+};
+const STATIC = process.argv[2];
+new Function(fs.readFileSync(`${STATIC}/crypto.js`, 'utf8'))();
+const src = fs.readFileSync(`${STATIC}/transport.js`, 'utf8');
+const { _challengeProvesNodeKey } =
+  new Function(src + '\nreturn { _challengeProvesNodeKey };')();
+
+const cases = JSON.parse(fs.readFileSync(process.argv[3], 'utf8'));
+(async () => {
+  const out = [];
+  for (const c of cases) {
+    const nonceC = Uint8Array.from(Buffer.from(c.nonce_c, 'base64'));
+    try {
+      out.push(String(await _challengeProvesNodeKey(
+        c.reply, c.group_id, nonceC, c.offer_sdp, c.answer_sdp)));
+    } catch (e) {
+      out.push('refused');
+    }
+  }
+  process.stdout.write(JSON.stringify(out));
+})();
+"""
+
+
+def _sdp(fp: bytes) -> str:
+    return "v=0\r\na=fingerprint:sha-256 " + ":".join(f"{b:02X}" for b in fp) + "\r\n"
+
+
+def test_the_browser_holds_the_node_to_its_challenge(tmp_path):
+    sk_node, sk_other = Ed25519PrivateKey.generate(), Ed25519PrivateKey.generate()
+    nonce_c, nonce_s = os.urandom(32), os.urandom(32)
+    offer_fp, answer_fp = os.urandom(32), os.urandom(32)
+    sig = sk_node.sign(challenge_transcript(
+        GROUP, nonce_c, nonce_s, webrtc_binding(offer_fp, answer_fp)))
+
+    def case(*, sig=sig, pk=sk_node, group=GROUP, answer=answer_fp, nonce=nonce_c):
+        reply = {"type": "handshake_challenge", "nonce": base64.b64encode(nonce_s).decode(),
+                 "node_pk": pk_to_b64(pk.public_key())}
+        if sig is not None:
+            reply["sig"] = base64.b64encode(sig).decode()
+        return {"reply": reply, "group_id": group,
+                "nonce_c": base64.b64encode(nonce).decode(),
+                "offer_sdp": _sdp(offer_fp), "answer_sdp": _sdp(answer)}
+
+    cases = {
+        "signed over this connection": (case(), "true"),
+        "an older node, no signature": (case(sig=None), "false"),
+        "another key announced": (case(pk=sk_other), "refused"),
+        "a relay's fingerprint": (case(answer=os.urandom(32)), "refused"),
+        "a replay under another nonce": (case(nonce=os.urandom(32)), "refused"),
+        "another group": (case(group="other"), "refused"),
+        "garbage for a signature": (case(sig=b"\x00" * 64), "refused"),
+    }
+    harness = tmp_path / "harness.js"
+    harness.write_text(_HARNESS)
+    payload = tmp_path / "cases.json"
+    payload.write_text(json.dumps([c for c, _ in cases.values()]))
+    proc = subprocess.run(["node", str(harness), str(STATIC), str(payload)],
+                          capture_output=True, text=True, timeout=60)
+    assert proc.returncode == 0, proc.stderr
+    got = dict(zip(cases, json.loads(proc.stdout)))
+    assert got == {name: want for name, (_, want) in cases.items()}