path: root/packages/meshbay-hub/tests/harness/chat_send_probe.py

#!/usr/bin/env python3
"""
Does sending a chat message come back?

Mounts **the real `ChatPanel` over the real `MeshBayTransport`** — both shipped
modules, neither a model of the other — and types a message into the composer
the way a person does. Only the DataChannel is replaced, by a stand-in that
answers what the node answers.

It exists because the defect it was written for is invisible to a structural
test and to `chat_scroll_probe.py` alike: nothing in `chat-app.js` is wrong,
and the transport routes every message it knows how to route. The node's reply
to a chat message is a bare `{"type": "ack"}` that names no request, so it fell
through to `_dispatch`'s arrival-order guess and was handed to whichever
request happened to be waiting — a `media_meta_req` from the Videos tab, say,
which the node never answered because it refused the file_id with a bare
`error` that named no request either. The chat send then waited out its own 30s
timeout with the composer disabled, so the tab looked frozen and the message
never appeared, while the node had stored it all along.

    chat_send_probe.py

It now drives two shapes of reply, because there were two ways for one to go
astray and only the first was ever fixed:

  `ack`    the node accepts the message and answers `{"type": "ack"}`, which
           names no request. Routed by request type since 2026-08-30.
  `error`  the node *refuses* it — every failure in `_dispatch_message` ends at
           one catch-all sending `{"type": "error", "detail": "Request failed"}`,
           and 238 of this module's 240 error sends name nothing either. That
           reply reached no caller at all: it went to whatever request happened
           to be waiting, and the send sat out its own 30s timeout with the
           composer disabled.

Both are run with an older request already pending — the condition that turns
"guess by arrival order" from usually-right into wrong — and both must come
back inside a second and a half.

Since MNP 2.0 a send also has to **seal and sign for real** before it goes
anywhere, so this drives `chatKeys()`, `openGroup`, `sealChat` and a genuine
Ed25519 signature rather than a model of any of them. The device key is
generated in the page — `signBytes` imports a pkcs8 key and WebCrypto will not
be fooled by a stand-in — and the `chat_keys_resp` the stub answers with is
sealed **by the shipped Python**, because a payload the page built itself would
prove only that the page agrees with the page.

That path found two defects the moment it first ran, neither visible in any
source file: `chat_keys_resp` was routed by arrival order and handed to the
older pending request (this defect, in a message type that did not exist when
the probe was written), and `_asText` had been deleted along with an unrelated
helper beside it — its only caller sits inside a promise the panel catches, so
every conversation rendered empty with nothing in the console.

Prints JSON: `scenarios`, the state of the panel at each stage of each, and
`log`, what the transport sent and how the deliberately-unanswered request
ended up.
"""
import http.server
import json
import socketserver
import subprocess
import sys
import tempfile
import threading
import time
from pathlib import Path

STATIC = Path(__file__).resolve().parents[2] / "src" / "meshbay_hub" / "static"
PORT = 8755
RECORDS = []
socketserver.TCPServer.allow_reuse_address = True

GROUP_ID = "g" * 32
GEK = bytes.fromhex("5a" * 32)
EPOCH_KEY = bytes.fromhex("7c" * 32)


def _page() -> str:
    """
    The page, with a real sealed `chat_keys_resp` baked in.

    Sealed here, by the shipped Python, rather than assembled in the browser:
    msgpack is private to transport.js and exported to nothing, and a payload
    the page built itself would prove only that the page agrees with the page.
    """
    import msgpack  # noqa: F401  (imported for the failure it gives if absent)

    from meshbay_common.groupbox import PURPOSE_CHAT_KEYS, seal

    sealed = seal(GEK, PURPOSE_CHAT_KEYS, "chat_keys_resp", GROUP_ID,
                  {"epochs": [{"epoch": 1, "key": EPOCH_KEY}], "current": 1})
    return (PAGE_TEMPLATE
            .replace("__GROUP_ID__", GROUP_ID)
            .replace("__GEK_HEX__", GEK.hex())
            .replace("__KEYS_NONCE_HEX__", sealed["nonce"].hex())
            .replace("__KEYS_CT_HEX__", sealed["ct"].hex()))

PAGE_TEMPLATE = r"""<!doctype html><html><head><meta charset=utf-8>
<link rel="stylesheet" href="/style.css"></head>
<body>
<div class="layout"><div class="main">
  <div class="group-header"><div><h2>a group</h2></div></div>
  <div class="group-tabs"><button class="group-tab active">Chat</button></div>
  <div id="root"></div>
</div></div>
<!-- The two the real page loads and the transport reaches for by global:
     `sealChat`/`openGroup` live in crypto.js, `signBytes` in keyderive.js.
     Without them a send fails with "cannot read properties of undefined". -->
<script src="/crypto.js"></script>
<script src="/keyderive.js"></script>
<script src="/transport.js"></script>
<script type="module">
import { html, render, useRef } from '/vendor/htm-preact.js';
import { ChatPanel } from '/chat-app.js';

const log = [];
window.addEventListener('error', e => log.push('error: ' + e.message));
window.addEventListener('unhandledrejection',
  e => log.push('rejected: ' + (e.reason && e.reason.message || e.reason)));

const hex = (s) => Uint8Array.from(s.match(/../g) || [], b => parseInt(b, 16));
const b64 = (buf) => btoa(String.fromCharCode(...new Uint8Array(buf)));

// One device key for both scenarios. Real, not stubbed: `signBytes` imports a
// pkcs8 key, so nothing else gets a signature past `verifyChatSignature`.
const kp = await crypto.subtle.generateKey({ name: 'Ed25519' }, true,
                                           ['sign', 'verify']);
const SK_ED_B64 = b64(await crypto.subtle.exportKey('pkcs8', kp.privateKey));
const DEVICE_PK_B64 = b64(await crypto.subtle.exportKey('raw', kp.publicKey));

const now = Date.now() / 1000;
const history = [];
for (let i = 0; i < 5; i++) {
  history.push({ id: 'm' + i, sender_id: 'someone', sender_name: 'someone',
                 payload: 'message ' + i, timestamp: now - (5 - i) * 60 });
}

const wait = ms => new Promise(r => setTimeout(r, ms));

// Stands in for the node, answering what webrtc_server.py answers — including
// stamping the reply with the id of the request it is answering, which is what
// `_send` does there. `chatReply` is the only difference between the two runs.
function makeTransport(name, chatReply) {
  // The real transport, with only the channel replaced: _send takes the plain
  // object _sendAndWait built, so the framing and msgpack are the only things
  // skipped — every pending entry, key and dispatch path below is the shipped
  // one.
  const tp = new window.MeshBayTransport('', 'token');
  tp._connected = true;
  tp._channel = { readyState: 'open', send() {} };
  // What a completed MNP 2.0 handshake leaves behind: the group and its key
  // from `connect`, the current epoch from the sealed ack, and the device this
  // connection identified itself as with `device_hello`.
  tp._groupId = '__GROUP_ID__';
  tp._gekRaw = hex('__GEK_HEX__');
  tp.chatEpoch = 1;
  tp._sessionKeys = { skEdB64: SK_ED_B64 };
  tp.devicePk = DEVICE_PK_B64;
  tp._send = (obj) => {
    log.push('sent ' + obj.type);
    const answer = (reply) => setTimeout(
      () => tp._dispatch({ ...reply, req_id: obj.req_id }), 10);
    if (obj.type === 'chat_hist') {
      answer({ type: 'chat_hist_resp', v: '0.2', messages: history, has_more: false });
    } else if (obj.type === 'chat_keys_req') {
      // Sealed under the group key, as `_do_chat_keys_req` sends it.
      answer({ type: 'chat_keys_resp', v: '2.0', group_id: tp._groupId,
               nonce: hex('__KEYS_NONCE_HEX__'), ct: hex('__KEYS_CT_HEX__') });
    } else if (obj.type === 'chat_msg') {
      // Recorded so the test can assert what actually left the browser, rather
      // than trusting that a composer which accepted the text sealed it.
      log.push(name + ': chat_msg format=' + obj.format + ' epoch=' + obj.epoch
               + ' ct=' + (obj.ct ? obj.ct.length : 0)
               + ' sig=' + (obj.sig ? obj.sig.length : 0)
               + ' plaintextLeak=' + JSON.stringify(obj).includes('hello'));
      answer(chatReply);
    }
    // music_meta_req is answered by nothing at all, on purpose: the node holds
    // one open for as long as the third-party lookup behind it takes, which
    // was measured at over 100 seconds with that service failing. It is the
    // older pending request every scenario here needs.
  };
  // The panel swallows a send failure into `setInput(text)`, which is right for
  // a person and useless for a probe: the symptom is a message not appearing,
  // with no reason anywhere. Surfaced here so a failure names itself — the
  // `error` scenario is *expected* to reach this.
  const _sendChat = tp.sendChat.bind(tp);
  tp.sendChat = (...a) => _sendChat(...a).catch((e) => {
    log.push(name + ': sendChat failed: ' + (e && e.message || e));
    throw e;
  });
  return tp;
}

function Host({ tp }) {
  const transportRef = useRef(tp);
  const gekRef = useRef(null);
  return html`<${ChatPanel} transportRef=${transportRef} gekRef=${gekRef}
    username="me" userId="user-me" entries=${[]} status="connected" />`;
}

function typeInto(root, text) {
  const c = root.querySelector('.chat-input');
  c.focus();
  // Preact reads e.target.value on input, so the native setter has to run.
  Object.getOwnPropertyDescriptor(window.HTMLTextAreaElement.prototype, 'value')
    .set.call(c, text);
  c.dispatchEvent(new Event('input', { bubbles: true }));
}

async function runScenario(name, chatReply) {
  const root = document.createElement('div');
  document.getElementById('root').appendChild(root);
  const tp = makeTransport(name, chatReply);
  render(html`<${Host} tp=${tp} />`, root);

  const steps = [];
  const snap = (label) => {
    const c = root.querySelector('.chat-input');
    steps.push({
      label,
      bubbles: root.querySelectorAll('.chat-bubble').length,
      lastText: [...root.querySelectorAll('.chat-text')].pop()?.textContent ?? null,
      // What a frozen tab actually is: the composer is disabled for as long as
      // a send is in flight.
      composerDisabled: c ? c.disabled : null,
      composerValue: c ? c.value : null,
      pending: tp._pending.size,
    });
  };

  await wait(500);
  snap('arrived');

  // The Music tab asked about a track and is still waiting on the node, which
  // is waiting on something else. Any older unanswered request will do; this
  // is the one that was live when the defect was found.
  tp.fetchMusicMeta('a-track-the-node-is-slow-about')
    .then(m => log.push(name + ': music_meta resolved with ' + m.type),
          e => log.push(name + ': music_meta rejected: ' + e.message));
  await wait(100);
  snap('older request pending');

  typeInto(root, 'hello');
  await wait(100);
  root.querySelector('.chat-input').dispatchEvent(new KeyboardEvent('keydown',
    { key: 'Enter', bubbles: true, cancelable: true }));

  // Far short of _sendAndWait's 30s timeout: a send that has not come back by
  // now is the freeze, not a slow node.
  await wait(1500);
  snap('after send');

  return { name, steps };
}

(async () => {
  const out = { scenarios: [], log };
  out.scenarios.push(await runScenario('ack', { type: 'ack', v: '0.14' }));
  out.scenarios.push(await runScenario(
    'error', { type: 'error', detail: 'Request failed' }));
  fetch('/log', { method: 'POST', body: JSON.stringify(out) });
})();
</script></body></html>"""


class H(http.server.BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        RECORDS.append(json.loads(
            self.rfile.read(int(self.headers["Content-Length"])).decode()))
        self.send_response(204)
        self.end_headers()

    def do_GET(self):
        if self.path == "/":
            body, ctype = _page().encode(), "text/html; charset=utf-8"
        else:
            path = (STATIC / self.path.lstrip("/")).resolve()
            if not str(path).startswith(str(STATIC)) or not path.is_file():
                self.send_response(404)
                self.end_headers()
                return
            body = path.read_bytes()
            ctype = ("text/css" if path.suffix == ".css"
                     else "text/javascript" if path.suffix == ".js"
                     else "application/octet-stream")
        self.send_response(200)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


def main() -> int:
    with socketserver.TCPServer(("127.0.0.1", PORT), H) as srv:
        threading.Thread(target=srv.serve_forever, daemon=True).start()
        # ignore_cleanup_errors: Chrome's children (zygote, renderer, gpu)
        # outlive terminate() on the parent by a moment and go on writing into
        # the profile. rmtree then walks a directory that gains a file between
        # its readdir and its rmdir and raises "Directory not empty" -- which
        # failed the probe, which failed every test in the file, intermittently
        # and for a reason nowhere near the chat code they were testing. A few
        # bytes left in a throwaway profile are harmless; failing the run is not.
        with tempfile.TemporaryDirectory(
                ignore_cleanup_errors=True) as profile:
            proc = subprocess.Popen(
                ["google-chrome", "--headless=new", "--disable-gpu", "--no-sandbox",
                 f"--user-data-dir={profile}", "--window-size=1100,800",
                 f"http://127.0.0.1:{PORT}/"],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            for _ in range(300):
                if RECORDS:
                    break
                time.sleep(0.1)
            proc.terminate()
            try:
                proc.wait(timeout=10)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait()
    if not RECORDS:
        print(json.dumps({"error": "no measurement"}), file=sys.stderr)
        return 1
    print(json.dumps(RECORDS[0], indent=1))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())