path: root/packages/meshbay-common/tests/test_js_python_parity.py

"""
Cross-language parity: the browser's transcripts must be byte-identical to Python's.

The handshake proof and the admin signature are computed independently on both sides
and compared by producing the same bytes. Nothing on the wire carries the transcript,
which is the point — but it also means a one-byte disagreement between `crypto.js` and
`meshbay_common` is invisible to every other test and produces a total outage: no
browser can complete a handshake with any node.

The rest of the suite runs in Python only, so nothing else crosses this boundary. These
tests drive the real `crypto.js` under node and compare against the real Python.

Skipped when node is unavailable; that is a coverage gap, not a pass.
"""

import base64
import json
import shutil
import subprocess
import tempfile
from pathlib import Path

import pytest

from meshbay_common.adminop import admin_transcript
from meshbay_common.handshake import handshake_transcript, webrtc_binding
from meshbay_common.join import join_transcript

CRYPTO_JS = (Path(__file__).resolve().parents[2]
             / "meshbay-hub" / "src" / "meshbay_hub" / "static" / "crypto.js")

pytestmark = pytest.mark.skipif(
    shutil.which("node") is None or not CRYPTO_JS.exists(),
    reason="node or crypto.js unavailable — parity cannot be checked",
)

# (role, group_id, nonce_c hex, nonce_s hex, offer_fp hex, answer_fp hex)
HANDSHAKE_VECTORS = [
    ("client", "g" * 32, "01" * 32, "02" * 32, "aa" * 32, "bb" * 32),
    ("node",   "g" * 32, "01" * 32, "02" * 32, "aa" * 32, "bb" * 32),
    # Short and empty group ids — length prefixing must keep these distinct.
    ("client", "g",  "03" * 32, "04" * 32, "cc" * 32, "dd" * 32),
    ("client", "",   "03" * 32, "04" * 32, "cc" * 32, "dd" * 32),
    # Non-ASCII: JS TextEncoder and Python .encode() must agree on UTF-8.
    ("client", "groupe-café-日本", "05" * 32, "06" * 32, "ee" * 32, "ff" * 32),
    # Fields that could run together under naive concatenation.
    ("client", "gg", "07" * 32, "08" * 32, "11" * 32, "22" * 32),
]

# (op, node_pk_b64, group_id, subject, nonce hex, ts)
ADMIN_VECTORS = [
    ("file_delete", "Tk9ERVBL", "g" * 32, "file-1", "01" * 32, 1_700_000_000),
    ("gek_bundle_store", "Tk9ERVBL", "g" * 32, "user-2", "02" * 32, 1_700_000_001),
    ("file_delete", "Tk9ERVBL", "", "", "03" * 32, 0),
    ("file_delete", "Tk9ERVBL", "café", "fichier é.mp4", "04" * 32, 1_700_000_002),
]

# (node_pk_b64, group_id, user_id, pk_ed b64, pk_x b64, nonce hex, ts)
JOIN_VECTORS = [
    # Operator pairing: group_id is empty and must stay distinguishable from a
    # request that names a group.
    ("Tk9ERVBL", "", "grenet", "QUFB", "QkJC", "01" * 32, 1_700_000_000),
    ("Tk9ERVBL", "g" * 32, "grenet", "QUFB", "QkJC", "01" * 32, 1_700_000_000),
    # Identical to the previous vector except that the two keys are swapped —
    # they are adjacent fields, so this isolates the ordering.
    ("Tk9ERVBL", "g" * 32, "grenet", "QkJC", "QUFB", "01" * 32, 1_700_000_000),
    ("Tk9ERVBL", "café", "utilisateur-é", "QUFB", "QkJC", "03" * 32, 0),
]

_HARNESS = r"""
const fs = require('fs');

// crypto.js ends with `window.MeshBayCrypto = {...}` and references SubtleCrypto in
// functions we do not call. A stub is enough to evaluate the module body.
globalThis.window = {};
globalThis.crypto = globalThis.crypto || {};

const src = fs.readFileSync(process.argv[2], 'utf8');
const load = new Function(
  src + '\nreturn { handshakeTranscript, adminTranscript, joinTranscript, '
      + 'webrtcBinding, b64encode };');
const M = load();

const hex = (s) => {
  const out = new Uint8Array(s.length / 2);
  for (let i = 0; i < s.length; i += 2) out[i / 2] = parseInt(s.substr(i, 2), 16);
  return out;
};
const toHex = (u8) =>
  Array.from(u8).map(b => b.toString(16).padStart(2, '0')).join('');

const input = JSON.parse(fs.readFileSync(process.argv[3], 'utf8'));
const out = { handshake: [], admin: [], join: [] };

for (const v of input.handshake) {
  const binding = M.webrtcBinding(hex(v.offer_fp), hex(v.answer_fp));
  out.handshake.push(toHex(M.handshakeTranscript(
    v.role, v.group_id, hex(v.nonce_c), hex(v.nonce_s), binding)));
}

for (const v of input.admin) {
  out.admin.push(toHex(M.adminTranscript(
    v.op, v.node_pk, v.group_id, v.subject, M.b64encode(hex(v.nonce)), v.ts)));
}

for (const v of input.join) {
  out.join.push(toHex(M.joinTranscript(
    v.node_pk, v.group_id, v.user_id, v.pk_ed, v.pk_x, hex(v.nonce), v.ts)));
}

process.stdout.write(JSON.stringify(out));
"""


@pytest.fixture(scope="module")
def js_output(tmp_path_factory):
    """Run the real crypto.js under node and return its transcripts as hex."""
    d = tmp_path_factory.mktemp("parity")
    harness = d / "harness.js"
    harness.write_text(_HARNESS)

    payload = d / "vectors.json"
    payload.write_text(json.dumps({
        "handshake": [
            {"role": r, "group_id": g, "nonce_c": nc,
             "nonce_s": ns, "offer_fp": ofp, "answer_fp": afp}
            for r, g, nc, ns, ofp, afp in HANDSHAKE_VECTORS
        ],
        "admin": [
            {"op": op, "node_pk": pk, "group_id": g,
             "subject": s, "nonce": n, "ts": ts}
            for op, pk, g, s, n, ts in ADMIN_VECTORS
        ],
        "join": [
            {"node_pk": pk, "group_id": g, "user_id": u,
             "pk_ed": pe, "pk_x": px, "nonce": n, "ts": ts}
            for pk, g, u, pe, px, n, ts in JOIN_VECTORS
        ],
    }))

    proc = subprocess.run(
        ["node", str(harness), str(CRYPTO_JS), str(payload)],
        capture_output=True, text=True, timeout=60,
    )
    if proc.returncode != 0:
        pytest.fail(f"node harness failed:\n{proc.stderr}")
    return json.loads(proc.stdout)


@pytest.mark.parametrize("idx,vector", list(enumerate(HANDSHAKE_VECTORS)))
def test_handshake_transcript_parity(idx, vector, js_output):
    """
    A mismatch here means no browser can complete a handshake with any node —
    the GEK proof would never verify, and no other test would notice.
    """
    role, group_id, nonce_c, nonce_s, offer_fp, answer_fp = vector

    expected = handshake_transcript(
        role=role,
        group_id=group_id,
        nonce_client=bytes.fromhex(nonce_c),
        nonce_node=bytes.fromhex(nonce_s),
        binding=webrtc_binding(bytes.fromhex(offer_fp), bytes.fromhex(answer_fp)),
    )
    assert js_output["handshake"][idx] == expected.hex(), (
        f"crypto.js and meshbay_common.handshake disagree for role={role!r} "
        f"group={group_id!r}"
    )


@pytest.mark.parametrize("idx,vector", list(enumerate(ADMIN_VECTORS)))
def test_admin_transcript_parity(idx, vector, js_output):
    """
    A mismatch here means the browser signs bytes the node did not ask for, so every
    file deletion and GEK bundle store is rejected.
    """
    op, node_pk, group_id, subject, nonce, ts = vector

    expected = admin_transcript(
        op=op,
        node_pk_b64=node_pk,
        group_id=group_id,
        subject=subject,
        nonce=bytes.fromhex(nonce),
        ts=ts,
    )
    assert js_output["admin"][idx] == expected.hex(), (
        f"crypto.js and meshbay_common.adminop disagree for op={op!r} "
        f"subject={subject!r}"
    )


@pytest.mark.parametrize("idx,vector", list(enumerate(JOIN_VECTORS)))
def test_join_transcript_parity(idx, vector, js_output):
    """
    A mismatch here means no browser can pair with a node and no member can be
    recognised — the node would reject every signature as invalid, and, as with
    the other two, nothing else in the suite crosses this boundary.
    """
    node_pk, group_id, user_id, pk_ed, pk_x, nonce, ts = vector

    expected = join_transcript(
        node_pk_b64=node_pk,
        group_id=group_id,
        user_id=user_id,
        pk_ed25519_b64=pk_ed,
        pk_x25519_b64=pk_x,
        nonce_node=bytes.fromhex(nonce),
        ts=ts,
    )
    assert js_output["join"][idx] == expected.hex(), (
        f"crypto.js and meshbay_common.join disagree for user={user_id!r} "
        f"group={group_id!r}"
    )


def test_join_transcript_binds_the_two_keys_in_order(js_output):
    """
    The X25519 key is trusted only because the Ed25519 identity signed it, so the
    two must not be interchangeable: swapping them has to produce different bytes.
    """
    assert js_output["join"][1] != js_output["join"][2]


def test_operator_pairing_is_distinguishable_from_a_group_join(js_output):
    """An empty group_id (node-wide operator authority) must not collide."""
    assert js_output["join"][0] != js_output["join"][1]


def test_length_prefixing_actually_disambiguates(js_output):
    """
    The reason both sides length-prefix: two different field splits must not collide.
    Verified across the language boundary, since a JS implementation that concatenated
    naively would still agree with itself.
    """
    a = js_output["handshake"][2]   # group_id "g"
    b = js_output["handshake"][3]   # group_id ""
    assert a != b, "JS transcripts collide across different group ids"


# ── groupbox: the sealed payload, both directions ────────────────────────────
#
# Unlike the transcripts above, this one has a wire format to disagree about as
# well as a derivation: the HKDF salt (Python's `salt=None` against WebCrypto's
# `salt: new Uint8Array(0)`) and the AAD's UTF-8 encoding are both invisible to
# every other test, and a disagreement in either means no browser can open an
# index or a handshake ack from any node — with the AEAD reporting only "it did
# not open", which is the same thing a wrong key reports.

# (purpose, msg_type, group_id)
GROUPBOX_VECTORS = [
    ("index", "index_sync", "g" * 32),
    ("index", "index_delta", "g" * 32),
    ("ack", "handshake_ack", "g" * 32),
    # Empty group id — the operator-pairing shape, and the one a naive
    # concatenation would let collide with a short id.
    ("ack", "handshake_ack", ""),
    # Non-ASCII: TextEncoder and Python's .encode() must agree on the AAD.
    ("index", "index_sync", "groupe-café-日本"),
    # A '|' inside the group id, which is the AAD's own separator.
    ("index", "index_sync", "a|b"),
    # MNP 2.0 — the upload, and the only purpose the *browser* seals in
    # production. A disagreement here means no file can be uploaded from any
    # browser to any node, and the node reports only "it did not open".
    ("upload", "file_upload", "g" * 32),
    ("upload", "file_upload_ack", "g" * 32),
]

GROUPBOX_GEK = bytes.fromhex("5a" * 32)

_GROUPBOX_HARNESS = r"""
const fs = require('fs');

globalThis.window = {};
const src = fs.readFileSync(process.argv[2], 'utf8');
const M = new Function(src + '\nreturn { sealGroup, openGroup };')();

const hex = (s) => {
  const out = new Uint8Array(s.length / 2);
  for (let i = 0; i < s.length; i += 2) out[i / 2] = parseInt(s.substr(i, 2), 16);
  return out;
};
const toHex = (u8) =>
  Array.from(u8).map(b => b.toString(16).padStart(2, '0')).join('');

(async () => {
  const input = JSON.parse(fs.readFileSync(process.argv[3], 'utf8'));
  const gek = hex(input.gek);
  const out = { opened: [], sealed: [] };

  for (const v of input.vectors) {
    // Python sealed it; open it here.
    out.opened.push(toHex(await M.openGroup(
      gek, v.purpose, v.msg_type, v.group_id,
      { nonce: hex(v.nonce), ct: hex(v.ct) })));
    // Seal the same plaintext here, for Python to open.
    const sealed = await M.sealGroup(
      gek, v.purpose, v.msg_type, v.group_id, hex(v.plaintext));
    out.sealed.push({ nonce: toHex(sealed.nonce), ct: toHex(sealed.ct) });
  }

  process.stdout.write(JSON.stringify(out));
})().catch((e) => { console.error(e); process.exit(1); });
"""


def _groupbox_payload(idx: int) -> dict:
    """A distinct payload per vector, so a crossed result cannot pass."""
    return {"n": idx, "name": f"entry-{idx}.bin", "flags": [True, None, idx * 7]}


@pytest.fixture(scope="module")
def groupbox_js(tmp_path_factory):
    import msgpack

    from meshbay_common.groupbox import seal

    d = tmp_path_factory.mktemp("groupbox-parity")
    harness = d / "harness.js"
    harness.write_text(_GROUPBOX_HARNESS)

    vectors = []
    for i, (purpose, msg_type, group_id) in enumerate(GROUPBOX_VECTORS):
        payload = _groupbox_payload(i)
        sealed = seal(GROUPBOX_GEK, purpose, msg_type, group_id, payload)
        vectors.append({
            "purpose": purpose, "msg_type": msg_type, "group_id": group_id,
            "nonce": sealed["nonce"].hex(), "ct": sealed["ct"].hex(),
            "plaintext": msgpack.packb(payload, use_bin_type=True).hex(),
        })

    payload_file = d / "vectors.json"
    payload_file.write_text(json.dumps({"gek": GROUPBOX_GEK.hex(),
                                        "vectors": vectors}))

    proc = subprocess.run(
        ["node", str(harness), str(CRYPTO_JS), str(payload_file)],
        capture_output=True, text=True, timeout=60,
    )
    if proc.returncode != 0:
        pytest.fail(f"node groupbox harness failed:\n{proc.stderr}")
    return json.loads(proc.stdout)


@pytest.mark.parametrize("idx,vector", list(enumerate(GROUPBOX_VECTORS)))
def test_browser_opens_what_python_sealed(idx, vector, groupbox_js):
    """A mismatch means no browser can read an index or a handshake ack."""
    import msgpack

    opened = bytes.fromhex(groupbox_js["opened"][idx])
    assert msgpack.unpackb(opened, raw=False) == _groupbox_payload(idx), (
        f"crypto.js and groupbox.py disagree for {vector!r}")


@pytest.mark.parametrize("idx,vector", list(enumerate(GROUPBOX_VECTORS)))
def test_python_opens_what_the_browser_sealed(idx, vector, groupbox_js):
    """
    The other direction, and since MNP 2.0 it is a shipping path rather than a
    precaution: `uploadFile` seals every chunk under `upload`, and the node
    opens it with `unseal`. The index vectors above still only ever run one way
    in production, and are kept because a codec whose encoder is untested is a
    codec with half a test.
    """
    from meshbay_common.groupbox import unseal

    purpose, msg_type, group_id = vector
    sealed = groupbox_js["sealed"][idx]
    msg = {"nonce": bytes.fromhex(sealed["nonce"]),
           "ct": bytes.fromhex(sealed["ct"])}
    assert unseal(GROUPBOX_GEK, purpose, msg_type, group_id, msg) == \
        _groupbox_payload(idx)


def test_the_browser_refuses_a_payload_sealed_for_another_message(groupbox_js):
    """
    The AAD, checked across the boundary rather than only within Python: a JS
    `openGroup` that dropped `additionalData` would still round-trip against
    itself and against Python, and would pass every other test here.
    """
    from meshbay_common.groupbox import seal

    sealed = seal(GROUPBOX_GEK, "index", "index_sync", "g1", {"x": 1})
    script = (
        "globalThis.window = {};\n"
        "const fs = require('fs');\n"
        "const M = new Function(fs.readFileSync(process.argv[2], 'utf8')\n"
        "  + '\\nreturn { openGroup };')();\n"
        "const hex = (s) => Uint8Array.from(s.match(/../g).map(b => parseInt(b, 16)));\n"
        "M.openGroup(hex(process.argv[3]), 'index', 'index_delta', 'g1',\n"
        "  { nonce: hex(process.argv[4]), ct: hex(process.argv[5]) })\n"
        "  .then(() => { console.log('OPENED'); })\n"
        "  .catch(() => { console.log('REFUSED'); });\n"
    )
    with tempfile.TemporaryDirectory() as tmp:
        h = Path(tmp) / "aad.js"
        h.write_text(script)
        proc = subprocess.run(
            ["node", str(h), str(CRYPTO_JS), GROUPBOX_GEK.hex(),
             sealed["nonce"].hex(), sealed["ct"].hex()],
            capture_output=True, text=True, timeout=60)
    assert proc.stdout.strip() == "REFUSED", proc.stdout + proc.stderr


# ── chatbox: a chat message, sealed and signed, both directions ──────────────
#
# Same class of invisible disagreement as groupbox above, with one more moving
# part: the per-device subkey is derived from a *string* that carries the group
# id and the device's base64 key, so a mismatch in how either is encoded means
# messages that encrypt fine and never decrypt — and AES-GCM reports that
# exactly the way it reports a wrong key.
#
# The signature is checked in both directions too. It is the half that
# establishes who spoke, and unlike the ciphertext it is verified by clients
# that may never hold the epoch key at all.

# (group_id, epoch)
CHAT_VECTORS = [
    ("g" * 32, 1),
    # Epoch is inside the AAD and the transcript as a decimal string; 0 and a
    # large value must not collide with each other or with the empty field.
    ("g" * 32, 0),
    ("g" * 32, 4294967296),
    # Empty group id, the operator-pairing shape.
    ("", 1),
    # Non-ASCII: TextEncoder and Python's .encode() must agree.
    ("groupe-café-日本", 2),
    # A '|' inside the group id — the separator used by both the AAD and the
    # HKDF info string.
    ("a|b", 3),
]

CHAT_EPOCH_KEY = bytes.fromhex("7c" * 32)

_CHATBOX_HARNESS = r"""
const fs = require('fs');

globalThis.window = {};
const src = fs.readFileSync(process.argv[2], 'utf8');
const M = new Function(src +
  '\nreturn { sealChat, openChat, chatSigningTranscript, verifyChatSignature };')();

const hex = (s) => {
  const out = new Uint8Array(s.length / 2);
  for (let i = 0; i < s.length; i += 2) out[i / 2] = parseInt(s.substr(i, 2), 16);
  return out;
};
const toHex = (u8) =>
  Array.from(u8).map(b => b.toString(16).padStart(2, '0')).join('');

(async () => {
  const input = JSON.parse(fs.readFileSync(process.argv[3], 'utf8'));
  const epochKey = hex(input.epoch_key);
  const out = { opened: [], sealed: [], transcripts: [], verified: [] };

  for (const v of input.vectors) {
    // Python sealed it; open it here.
    out.opened.push(toHex(await M.openChat(
      epochKey, v.group_id, v.epoch, v.device_b64, hex(v.nonce), hex(v.ct))));
    // Seal the same plaintext here, for Python to open.
    const sealed = await M.sealChat(
      epochKey, v.group_id, v.epoch, v.device_b64, hex(v.plaintext));
    out.sealed.push({ nonce: toHex(sealed.nonce), ct: toHex(sealed.ct) });
    // The signed bytes, and whether Python's signature verifies here.
    out.transcripts.push(toHex(M.chatSigningTranscript(
      v.group_id, v.epoch, hex(v.device_raw), hex(v.nonce), hex(v.ct))));
    out.verified.push(await M.verifyChatSignature(
      hex(v.device_raw), v.group_id, v.epoch, hex(v.nonce), hex(v.ct),
      hex(v.sig)));
  }

  process.stdout.write(JSON.stringify(out));
})().catch((e) => { console.error(e); process.exit(1); });
"""


def _chat_payload(idx: int) -> dict:
    """A distinct message per vector, so a crossed result cannot pass."""
    return {"text": f"message {idx} — café", "thread_id": None,
            "sender_name": f"member-{idx}", "sent_at": 1_700_000_000 + idx}


@pytest.fixture(scope="module")
def chatbox_js(tmp_path_factory):
    import msgpack
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.asymmetric.ed25519 import (
        Ed25519PrivateKey,
    )

    from meshbay_common.chatbox import seal

    d = tmp_path_factory.mktemp("chatbox-parity")
    harness = d / "harness.js"
    harness.write_text(_CHATBOX_HARNESS)

    vectors = []
    for i, (group_id, epoch) in enumerate(CHAT_VECTORS):
        # A distinct device per vector: the subkey is derived from the device's
        # own base64 key, so reusing one would hide a derivation that ignored it.
        sk = Ed25519PrivateKey.from_private_bytes(bytes([i + 1]) * 32)
        device_raw = sk.public_key().public_bytes(
            serialization.Encoding.Raw, serialization.PublicFormat.Raw)
        device_b64 = base64.b64encode(device_raw).decode()
        payload = _chat_payload(i)
        sealed = seal(CHAT_EPOCH_KEY, group_id, epoch, device_b64, device_raw,
                      sk, payload)
        vectors.append({
            "group_id": group_id, "epoch": epoch,
            "device_b64": device_b64, "device_raw": device_raw.hex(),
            "nonce": sealed["nonce"].hex(), "ct": sealed["ct"].hex(),
            "sig": sealed["sig"].hex(),
            "plaintext": msgpack.packb(payload, use_bin_type=True).hex(),
        })

    payload_file = d / "vectors.json"
    payload_file.write_text(json.dumps({"epoch_key": CHAT_EPOCH_KEY.hex(),
                                        "vectors": vectors}))

    proc = subprocess.run(
        ["node", str(harness), str(CRYPTO_JS), str(payload_file)],
        capture_output=True, text=True, timeout=60,
    )
    if proc.returncode != 0:
        pytest.fail(f"node chatbox harness failed:\n{proc.stderr}")
    return json.loads(proc.stdout), vectors


@pytest.mark.parametrize("idx,vector", list(enumerate(CHAT_VECTORS)))
def test_browser_opens_a_chat_message_python_sealed(idx, vector, chatbox_js):
    import msgpack

    js, _ = chatbox_js
    assert msgpack.unpackb(bytes.fromhex(js["opened"][idx]), raw=False) == \
        _chat_payload(idx)


@pytest.mark.parametrize("idx,vector", list(enumerate(CHAT_VECTORS)))
def test_python_opens_a_chat_message_the_browser_sealed(idx, vector, chatbox_js):
    import msgpack

    from meshbay_common.chatbox import open_message

    js, vectors = chatbox_js
    group_id, epoch = vector
    sealed = js["sealed"][idx]
    opened = open_message(
        CHAT_EPOCH_KEY, group_id, epoch, vectors[idx]["device_b64"],
        bytes.fromhex(sealed["nonce"]), bytes.fromhex(sealed["ct"]))
    assert opened == _chat_payload(idx)
    # And the msgpack the browser produced is what Python produces, so the two
    # are not merely each self-consistent.
    assert msgpack.packb(opened, use_bin_type=True) == \
        bytes.fromhex(vectors[idx]["plaintext"])


@pytest.mark.parametrize("idx,vector", list(enumerate(CHAT_VECTORS)))
def test_the_signing_transcript_is_byte_identical(idx, vector, chatbox_js):
    from meshbay_common.chatbox import signing_transcript

    js, vectors = chatbox_js
    group_id, epoch = vector
    v = vectors[idx]
    expected = signing_transcript(
        group_id, epoch, bytes.fromhex(v["device_raw"]),
        bytes.fromhex(v["nonce"]), bytes.fromhex(v["ct"]))
    assert js["transcripts"][idx] == expected.hex()


@pytest.mark.parametrize("idx,vector", list(enumerate(CHAT_VECTORS)))
def test_the_browser_verifies_a_signature_python_made(idx, vector, chatbox_js):
    js, _ = chatbox_js
    assert js["verified"][idx] is True


def test_a_message_does_not_open_under_another_epoch(chatbox_js):
    """
    The epoch is in the AAD, so a message cannot be re-presented as belonging
    to a later key. Without it, a member who kept an old epoch key could make
    an old message look current — and an AEAD that ignored `additionalData`
    would round-trip against itself and pass every other test here.
    """
    import pytest as _pytest

    from meshbay_common.chatbox import open_message

    _js, vectors = chatbox_js
    v = vectors[0]
    group_id, epoch = CHAT_VECTORS[0]
    with _pytest.raises(Exception):
        open_message(CHAT_EPOCH_KEY, group_id, epoch + 1, v["device_b64"],
                     bytes.fromhex(v["nonce"]), bytes.fromhex(v["ct"]))


def test_a_message_does_not_open_under_another_devices_key(chatbox_js):
    """
    Each device has its own subkey, derived from its own public key. That is
    what makes nonce reuse across two devices of one person impossible without
    any coordination — the property per-device ratchet chains were wanted for.
    """
    import pytest as _pytest

    from meshbay_common.chatbox import open_message

    _js, vectors = chatbox_js
    group_id, epoch = CHAT_VECTORS[0]
    with _pytest.raises(Exception):
        open_message(CHAT_EPOCH_KEY, group_id, epoch, vectors[1]["device_b64"],
                     bytes.fromhex(vectors[0]["nonce"]),
                     bytes.fromhex(vectors[0]["ct"]))