diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-09-07 18:03:52 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-09-07 18:03:52 +0200 |
| commit | e1383e1d545b994f4ad61694f868339defb0bdef (patch) | |
| tree | 67a3b1933f2a9c5caf107d01d9ff91c44a975df6 /packages | |
| parent | 36cebf25d0e0f24cf63be4380ccb5d03da726a74 (diff) | |
| parent | 8980a8e42d94ab7c0bc9739283d39f938f8402b0 (diff) | |
| download | meshbay-e1383e1d545b994f4ad61694f868339defb0bdef.tar.gz | |
Merge origin/main into the chat encryption work
Both sides landed a breaking MNP change and both called it 2.0, which is right:
the sealed upload, the removal of `stream_seg` and mandatory chat encryption
share one flag day. They are recorded as one version in `__init__.py` rather
than as a race between two.
The resolutions that were decisions rather than mechanics:
* **`MNP_MIN_SUPPORTED` moves to "2.0".** The sealed upload alone was a
*confined* break — a 1.x peer could still connect, browse, download, stream
and chat, with only its uploads refused by `upload_not_sealed` — so the floor
deliberately stayed at "1.0". Mandatory chat encryption ends that
confinement: a 1.x peer can neither produce a sealed chat message nor read
one, so it would connect, look fine, and be unable to say anything. Refusing
it at the handshake is the honest form. The per-message `upload_not_sealed`
path is untouched and still right if the floor is ever lowered.
* **`sendChat` throws on an `error` reply**, from origin, applied to the sealed
send. It matters more after this change, not less: the node now refuses a
stale epoch, a malformed envelope and a device claim that is not the
connection's own, so there are three new ways for a message to be rejected
and none of them may look like a message that was sent.
* **`req_id` supersedes the per-type routing** this branch added for
`chat_keys_resp` and `device_hello_ack`. Both blocks are kept beside the
existing `chat_hist_resp` one, for the same stated reason — a node too old to
stamp — and their comments no longer claim to be the mechanism that closes
the class. `req_id` is.
* **`chat_send_probe.py` is rebuilt on origin's structure**, not beside it: two
scenarios, a stub that stamps `req_id`, `music_meta_req` as the older pending
request. The encrypted path is layered on — a real Ed25519 device key
generated in the page, and a `chat_keys_resp` sealed by the shipped Python,
because a payload the page built itself would prove only that the page agrees
with the page.
* **`test_reply_correlation.py` now sends a sealed message.** Its subject is
which of the two messages leaving that handler carries the id; plaintext chat
was only the fixture, and the node refuses one now.
* `groupbox` keeps both new purposes (`upload`, `chat_keys`); `protocol.py`
keeps origin's removal of `STREAM_SEGMENT` and this branch's correction of
the "Double Ratchet message" comment on `CHAT_MESSAGE`, which was wrong when
it was written and is wrong differently now.
Full suite on the merged tree: 1993 passed, 11 failed — the same 11 that fail
on a pristine checkout (2 Windows service tests, 1 apps-enabled policy, 7
transcode tests that pass in isolation, and the WebRTC invite test that hangs
on its own).
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TZZxYjz8YeWRz13xDi8LJr
Diffstat (limited to 'packages')
22 files changed, 1723 insertions, 599 deletions
diff --git a/packages/meshbay-common/src/meshbay_common/__init__.py b/packages/meshbay-common/src/meshbay_common/__init__.py index ca4c6eb..cb6ad15 100644 --- a/packages/meshbay-common/src/meshbay_common/__init__.py +++ b/packages/meshbay-common/src/meshbay_common/__init__.py @@ -90,19 +90,48 @@ __version__ = "0.11.0" # makes the *next* breaking change cost a refusal message instead of a second # flag day. `MNP_MIN_SUPPORTED` in `handshake.py` is the other half. # -# **2.0 (2026-09-07): chat is encrypted, and there is no way to turn it off.** -# A MAJOR bump because it is a real break: a 1.x peer cannot produce a sealed -# chat message and cannot read one, so it is refused at the handshake with -# `version_too_old` rather than connecting and then failing to speak. Expressing -# the break in the version is what makes it a stated refusal instead of a -# conversation that silently does not work — `MNP_MIN_SUPPORTED` moves with it. +# **2.0 (2026-09-07): the write path is sealed, chat is encrypted, and the last +# unencrypted content message is gone.** Three changes that landed together and +# share one version, because they share one flag day. # -# There is deliberately no per-group switch. Every node in existence is a test -# node, so an opt-in flag would buy nothing and cost a compatibility path to -# maintain; existing node data is migrated by `QE/migrate-chat-encryption.py`. +# - `file_upload` and `file_upload_ack` travel sealed under a GEK-derived +# subkey (`groupbox.PURPOSE_UPLOAD`). The filename, the destination folder +# and the bytes all ride inside the seal; `upload_id` — a fresh +# client-chosen correlation id — and `chunk_index` stay in clear because +# the node routes and orders on them. `filename` used to be the +# correlation key and cannot be any more, which is what forced `upload_id`. +# - `chat_msg` is sealed under a per-device subkey of the group's chat epoch +# key (`meshbay_common.chatbox`) and signed over the ciphertext with the +# device key the node pinned. There is deliberately **no per-group switch**: +# every node in existence is a test node, so an opt-in flag would buy +# nothing and leave a plaintext branch reachable. Existing node data is +# migrated by `QE/migration/migrate_chat_encryption.py`. +# - `stream_seg` is **removed**. It answered with an MPEG-TS segment as +# base64 with no encryption at all, on both transports, to any +# authenticated member — the one content-plane message that never went +# through a GEK-derived key. `stream_data` has done the job properly since +# Phase 12, and `fetchStreamSegment`, its only browser caller, was defined +# and never once invoked. # -# The index at rest, `index_progress` (counters only, never a path — see -# `groupbox.py` and daemon.py `_push_index_progress`), and file content on the -# operator's disk are all deliberately unchanged. +# **Breaking, on the wire every deployed client speaks**, and MAJOR by the same +# rule 1.0 was. +# +# **`MNP_MIN_SUPPORTED` moves to "2.0" with it, and that is a change of plan +# worth reading.** The sealed upload alone was a *confined* break: a 1.x peer +# could still connect, browse, download, stream and chat, and only its uploads +# were refused — so the floor stayed at "1.0" and the refusal was per message +# (`upload_not_sealed`). Mandatory chat encryption ends that confinement. A 1.x +# peer can neither produce a sealed chat message nor read one, so it would +# connect, appear to work, and then be unable to say anything or read anything +# anyone else said. Refusing it at the handshake with `version_too_old` and a +# sentence saying so is the honest form: a stated refusal is a bug report, a +# chat that quietly does not work is a support case. The per-message +# `upload_not_sealed` path stays, unchanged — it is still the right answer if +# the floor is ever lowered again. +# +# Still deliberately in clear, and none of it is content: the handshake itself, +# `index_progress` (counters only — see daemon.py `_push_index_progress`), the +# admin and configuration acks, and the media-metadata replies. The index at +# rest and file content on the operator's disk are unchanged. MNP_VERSION = "2.0" MHP_VERSION = "0.1" diff --git a/packages/meshbay-common/src/meshbay_common/groupbox.py b/packages/meshbay-common/src/meshbay_common/groupbox.py index ff60bba..e020ed8 100644 --- a/packages/meshbay-common/src/meshbay_common/groupbox.py +++ b/packages/meshbay-common/src/meshbay_common/groupbox.py @@ -24,6 +24,14 @@ and group configuration. It buys nothing against a network observer (DTLS/TLS already covers that), nothing against the hub (it never sees channel traffic), and nothing against a member — who holds the GEK. That is the whole claim. +**Both, for the upload (MNP 2.0).** `file_upload` carried the filename and the raw +bytes in clear, and `file_upload_ack` carried the name it was stored under. The +download path had been sealed end to end since the beginning — so the same file was +ciphertext coming out of a node and plaintext going in, which is not a threat model, +it is an oversight. The node holds the GEK for its own group, so unlike the index +this direction seals *towards* the node: it opens the payload before it writes +anything to disk, and refuses a chunk that does not open rather than guessing. + Purpose separation is deliberate. `GroupIndex.serialize()` reuses `chunk_key_aes(gek, file_hash, chunk_index)` with a pseudo-file ("the index as chunk 0 of a virtual index file"), which borrows a file's key space for something that is @@ -41,6 +49,7 @@ from cryptography.hazmat.primitives.kdf.hkdf import HKDF PURPOSE_INDEX = "index" PURPOSE_ACK = "ack" +PURPOSE_UPLOAD = "upload" # The chat epoch keys themselves, on their way to a member. The keys are what # the chat archive is encrypted under; this is only how they travel, which is # why rotating the group key costs a re-wrap and not a re-encryption. @@ -52,9 +61,17 @@ PURPOSE_CHAT_KEYS = "chat_keys" _INFO = { PURPOSE_INDEX: b"meshbay:index:v1", PURPOSE_ACK: b"meshbay:ack:v1", + PURPOSE_UPLOAD: b"meshbay:upload:v1", PURPOSE_CHAT_KEYS: b"meshbay:chat_keys:v1", } +# One subkey per purpose, and `seal` draws a fresh 96-bit nonce per message, so +# the bound that matters is birthday collision under `PURPOSE_UPLOAD` — the only +# purpose with real volume, one message per 48 KiB chunk. 2**32 chunks is 200 TB +# uploaded under a single GEK before the collision probability reaches 2**-32, +# and `gek_rotate` exists. Deriving the nonce from the payload instead would be +# worse, not better: two chunks of identical bytes are ordinary in a file. + NONCE_LEN = 12 # 96-bit, the WebCrypto AES-GCM standard diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index dfb5d56..cc5324a 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -6,8 +6,31 @@ MHP (Mesh Bay Hub Protocol) — v0.1 All wire messages are length-prefixed msgpack (4-byte big-endian length header). Every message carries a "v" field for protocol version. + +**`req_id` — the correlation id (added 2026-09-07).** A request may carry one; +the reply to it carries the same value back, and nothing else on the wire does. +It is the caller's own key for its pending request, opaque to the node, and +unique only within one connection. + +There was none for a long time, and its absence was not neutral. A reply named +its own type and nothing else, so a caller with more than one request in flight +had to work out which one a message answered from the message itself — and the +replies that name nothing (a bare `ack`, and `{"type": "error"}`, which +webrtc_server.py sends from 240 places while two of them say what they are +about) could only be matched by arrival order. That is a guess, wrong whenever +two replies reorder, and it does not fail quietly: one request is resolved with +another's answer while the request that answer belonged to waits out its own +timeout. Live symptom (2026-09-06): a chat send whose reply went astray left +the composer disabled for thirty seconds, and the Chat tab read as frozen. + +Both halves are optional and degrade to what came before: a request without one +is answered without one, and a client that gets no id back falls back to +matching by type. Neither side may treat it as authentication or as a sequence +number — it is a label chosen by the peer, and the only thing it decides is +which local promise a reply belongs to. """ +import os from dataclasses import dataclass, field from typing import Any @@ -15,6 +38,7 @@ from typing import Any # second copy here said "0.1" while every message on the wire carried "0.2". # Nothing imported it, which is the only reason it was harmless. from meshbay_common import MNP_VERSION, MHP_VERSION # noqa: F401 (re-export) +from meshbay_common.groupbox import PURPOSE_UPLOAD, seal, unseal from meshbay_common.webcrypto import ( chunk_key_aes, decrypt_chunk_aes, @@ -37,13 +61,21 @@ class MNP: INDEX_PROGRESS = "index_progress" FILE_REQUEST = "file_req" # request chunk(s) FILE_CHUNK = "file_chunk" # encrypted chunk response - STREAM_SEGMENT = "stream_seg" # HLS/DASH segment + # STREAM_SEGMENT ("stream_seg") was removed in MNP 2.0. It served an + # MPEG-TS segment as base64 **with no encryption at all** — the one message + # on the content plane that never was under a GEK-derived key. It predates + # STREAM_DATA, which does the same job properly (`chunk_ciphertext`, keyed + # per segment), and its browser caller `fetchStreamSegment` was defined and + # never once invoked. A live handler on both transports, plaintext media, + # and no client: removed rather than repaired. + # # Not a Double Ratchet message, and never was — `first-review.md` C1 - # rejected exactly that for groups. Plaintext until a group turns - # encryption on, then AES-256-GCM under a per-device subkey of the group's - # chat epoch key, signed with the sending device's pinned Ed25519 key - # (`chatbox.py`, docs/chat-sender-keys.md). - CHAT_MESSAGE = "chat_msg" # one chat message, plain or sealed + # rejected exactly that for groups. Since MNP 2.0 it is AES-256-GCM under a + # per-device subkey of the group's chat epoch key, signed over the + # ciphertext with the sending device's pinned Ed25519 key. There is no + # plaintext form on the wire (`chatbox.py`, docs/chat-sender-keys.md); + # `format` distinguishes a *stored* pre-2.0 row, which is still served. + CHAT_MESSAGE = "chat_msg" # one chat message, sealed and signed CHAT_ATTACHMENT = "chat_attach" # attachment metadata CHAT_HISTORY = "chat_hist" # request message history (newest, or before a cursor) CHAT_HISTORY_RESPONSE = "chat_hist_resp" # history response with messages @@ -369,3 +401,110 @@ def file_chunk_plaintext( file_hash = bytes.fromhex(msg["file_id"]) ckey = chunk_key_aes(gek, file_hash, msg["chunk_index"]) return decrypt_chunk_aes(ckey, msg["nonce"], msg["ct"]) + + +# ── Uploads (MNP 2.0) ───────────────────────────────────────────────────────── +# +# The write path, sealed under the group key the way the read path always was. +# One encoder for both directions, here rather than in the client, for the reason +# `file_chunk` has one: two copies of a wire shape with a single consumer each is +# how `index_sync` and `file_chunk` forked (finding C6), and nothing noticed +# until someone went looking. +# +# What stays in clear, and why each has to: +# `type`, `v` — routed and version-checked before anything can be decrypted +# `upload_id` — the correlation key. It replaces `filename`, which used to +# play that role and cannot any more: naming the file in clear +# to match an ack against a request would give back exactly +# what the seal is for. Client-chosen, opaque to the node, +# unique within one connection; never an authorization input. +# `chunk_index` — ordering, which the node enforces before it opens anything +# `total_chunks` — how many to expect +# +# `group_id` is *not* on the message: the session already decided which group it +# is on, and the node uses that as the AAD. A client naming its own group here +# would be choosing which key its bytes are checked against. + +UPLOAD_ID_LEN = 16 # 128 bits of client-chosen correlation, hex on the wire + + +def new_upload_id() -> str: + """A fresh correlation id for one upload.""" + return os.urandom(UPLOAD_ID_LEN).hex() + + +def file_upload_wire( + gek: bytes, + group_id: str, + *, + upload_id: str, + chunk_index: int, + total_chunks: int, + filename: str, + data: bytes, + dir: str = "", + root: str = "", +) -> dict: + """ + One sealed `file_upload` chunk. + + `filename`, `dir` and `root` ride inside the seal with the bytes: sealing the + content and announcing the name beside it would be theatre. They are repeated + on every chunk rather than sent once — a hundred bytes against a 48 KiB chunk + — because a header that arrives once is state the node has to carry, and + upload state that can disagree with the chunk in hand is what `_free_name` and + the chunk-ordering rule exist to prevent. + """ + payload = {"filename": filename, "data": data, "dir": dir, "root": root} + return { + "type": MNP.FILE_UPLOAD, + "v": MNP_VERSION, + "upload_id": upload_id, + "chunk_index": chunk_index, + "total_chunks": total_chunks, + **seal(gek, PURPOSE_UPLOAD, MNP.FILE_UPLOAD, group_id, payload), + } + + +def file_upload_payload(gek: bytes, group_id: str, msg: dict) -> dict: + """ + Open a `file_upload`. Raises on anything that does not open. + + Never a partial result and never a default: a chunk that does not open is not + an empty file with an empty name, it is a peer we cannot talk to. `unseal` + says why at length. + """ + return unseal(gek, PURPOSE_UPLOAD, MNP.FILE_UPLOAD, group_id, msg) + + +def file_upload_ack_wire( + gek: bytes, + group_id: str, + *, + upload_id: str, + chunk_index: int, + filename: str, + stored_as: str, + dir: str = "", +) -> dict: + """ + The node's answer to one chunk, sealed the same way. + + `stored_as` is the name the node settled on — it finds a free one rather than + replacing anything — and `dir` is where it landed. Both name the operator's + content, so both belong inside the seal; only `upload_id` and `chunk_index` + stay out, because the client matches on them. + """ + payload = {"filename": filename, "stored_as": stored_as, "dir": dir} + return { + "type": MNP.FILE_UPLOAD_ACK, + "v": MNP_VERSION, + "upload_id": upload_id, + "chunk_index": chunk_index, + **seal(gek, PURPOSE_UPLOAD, MNP.FILE_UPLOAD_ACK, group_id, payload), + } + + +def file_upload_ack_payload(gek: bytes, group_id: str, msg: dict) -> dict: + """Open a `file_upload_ack`. Raises on anything that does not open.""" + return unseal(gek, PURPOSE_UPLOAD, MNP.FILE_UPLOAD_ACK, group_id, msg) diff --git a/packages/meshbay-common/tests/test_js_python_parity.py b/packages/meshbay-common/tests/test_js_python_parity.py index dffee36..9893a9f 100644 --- a/packages/meshbay-common/tests/test_js_python_parity.py +++ b/packages/meshbay-common/tests/test_js_python_parity.py @@ -259,6 +259,11 @@ GROUPBOX_VECTORS = [ ("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) @@ -350,9 +355,11 @@ def test_browser_opens_what_python_sealed(idx, vector, groupbox_js): @pytest.mark.parametrize("idx,vector", list(enumerate(GROUPBOX_VECTORS))) def test_python_opens_what_the_browser_sealed(idx, vector, groupbox_js): """ - The other direction. Nothing in the SPA seals today — `sealGroup` exists for - the chat plan, which needs the same primitive — but a codec that only ever - runs one way is a codec whose encoder is untested. + 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 diff --git a/packages/meshbay-hub/src/meshbay_hub/static/crypto.js b/packages/meshbay-hub/src/meshbay_hub/static/crypto.js index f9cff08..24d1399 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/crypto.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/crypto.js @@ -111,8 +111,9 @@ async function decryptChunkBin(gek, fileHashHex, chunkIndex, nonce, ct) { // ── Sealing a payload under the group key ──────────────────────────────────── // -// Mirrors meshbay_common/groupbox.py. `index_sync`, `index_delta` and the -// `handshake_ack` config payload travel sealed under a GEK-derived subkey; the +// Mirrors meshbay_common/groupbox.py. `index_sync`, `index_delta`, the +// `handshake_ack` config payload and both halves of an upload travel sealed +// under a GEK-derived subkey; the // routing fields (type, v, group_id) and the ack's own authentication (node_pk, // proof, sig) stay in clear, because a receiver must route, version-check and // *authenticate* before it would trust a decryption. @@ -126,6 +127,12 @@ async function decryptChunkBin(gek, fileHashHex, chunkIndex, nonce, ct) { const GROUPBOX_INFO = { index: new TextEncoder().encode('meshbay:index:v1'), ack: new TextEncoder().encode('meshbay:ack:v1'), + // MNP 2.0: `file_upload` and `file_upload_ack`. This is the one purpose that + // seals *towards* the node — it holds the GEK for its own group — and the one + // with real message volume, one per 48 KB chunk. groupbox.py carries the + // nonce-collision arithmetic that makes a random 96-bit nonce fine at that rate. + upload: new TextEncoder().encode('meshbay:upload:v1'), + // The group's chat epoch keys, on their way to a member. chat_keys: new TextEncoder().encode('meshbay:chat_keys:v1'), }; diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index 927956d..1f6dd8f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -251,7 +251,7 @@ window.addEventListener('hashchange', () => { // The `v: '0.1'` on every other message in this file is the historical value // and is read by nothing; it is left alone deliberately. The range is // negotiated once, at the start, not restated per message. -const MNP_V = '1.1'; +const MNP_V = '2.0'; const MNP_V_MIN = '1.0'; // Codes a NODE sends us, in its own vocabulary (meshbay_common/handshake.py's @@ -290,6 +290,11 @@ class MeshBayTransport { this._pc = null; this._channel = null; this._pending = new Map(); + // Set the first time this connection sees a reply that names the request + // it answers (see _dispatch). A node either stamps every reply or none, + // so one is proof for the connection — and once there is proof, the + // arrival-order fallback at the bottom of _dispatch is never right again. + this._correlates = false; this._seqId = 0; this._recvBuf = new Uint8Array(0); this._connected = false; @@ -299,10 +304,17 @@ class MeshBayTransport { this._onStreamEnd = null; this._onStreamError = null; this._onIndexSync = null; - // filename → the uploader waiting on it. Keyed rather than FIFO because - // several uploads may be in flight at once and their acks interleave; the - // node names the file in every one. + // upload_id → the uploader waiting on it. Keyed rather than FIFO because + // several uploads may be in flight at once and their acks interleave. + // + // It was keyed by filename until MNP 2.0, which is no longer possible: the + // name is sealed under the group key, and echoing it in clear so the two + // sides could match on it would give back precisely what the seal is for. + // `upload_id` is drawn per upload here and is opaque to the node. this._uploaders = new Map(); + // Names, not ids: the "already being uploaded" guard is about the file the + // caller passed, and two `uploadFile` calls for one file draw two ids. + this._inFlightUploads = new Set(); // Set once close() runs — stops the automatic reconnect from firing on a // connection the caller tore down on purpose (leaving the group, page // unload), which would otherwise race back in right as everything else @@ -379,6 +391,19 @@ class MeshBayTransport { if (!m) return false; return (Number(m[1]) > 1) || (Number(m[1]) === 1 && Number(m[2]) >= 1); } + /** + * Whether the node opens a sealed upload (MNP 2.0). + * + * A 1.x node reads `filename` and `data` off the message itself, finds + * neither — they are inside the seal — and answers "Missing filename or + * data", an error about the wrong thing that names no upload_id and so fails + * every upload in flight. Asked before sending rather than discovered after, + * for the same reason `supportsAppOps` is. + */ + get supportsSealedUpload() { + const m = /^(\d+)\.(\d+)$/.exec(this._nodeVersion || ''); + return !!m && Number(m[1]) >= 2; + } set onAppsEnabled(fn) { this._onAppsEnabled = fn; } set onAppDirectories(fn) { this._onAppDirectories = fn; } set onChatDirectory(fn) { this._onChatDirectory = fn; } @@ -1456,18 +1481,6 @@ class MeshBayTransport { return msg; } - async fetchStreamSegment(fileId, segmentIndex, segmentDuration) { - const msg = await this._sendAndWait({ - type: 'stream_seg', - v: '0.1', - file_id: fileId, - segment_index: segmentIndex, - segment_duration: segmentDuration || 4, - }); - if (msg.type === 'error') throw new Error(msg.detail); - return _b64decode(msg.data_b64); - } - /** * A page of chat history, newest first by default. * @@ -1583,7 +1596,7 @@ class MeshBayTransport { this._sessionKeys.skEdB64, C.chatSigningTranscript(gid, epoch, device, nonce, ct))); - return this._sendAndWait({ + const msg = await this._sendAndWait({ type: 'chat_msg', v: '2.0', format: 1, @@ -1596,6 +1609,15 @@ class MeshBayTransport { // Deliberately absent: the display name is inside the envelope now. sender_name: null, }); + // Every other request in this file refuses an `error` reply; this one + // returned it as though the node had accepted the message. It never + // mattered while a refusal reached the wrong caller anyway — now that a + // reply finds the request that made it, a message the node rejected would + // otherwise appear in the conversation as sent. It rejects more of them + // than it used to: a stale epoch, an envelope the node dislikes, or a + // device claim that is not this connection's all come back as `error`. + if (msg.type === 'error') throw new Error(msg.detail || 'chat send refused'); + return msg; } /** @@ -2080,10 +2102,21 @@ class MeshBayTransport { */ async uploadFile(file, { chunkSize, onProgress, signal, root, dir } = {}) { // The same file twice at once would confuse the node, which keys its own - // upload state by name — and would race for the same destination. - if (this._uploaders.has(file.name)) { + // upload state by name — and would race for the same destination. The guard + // is by name for that reason, even though the map below is keyed by id. + if (this._inFlightUploads.has(file.name)) { throw new Error(`${file.name} is already being uploaded`); } + if (!this._gekRaw) throw new Error('This group has no key on this device'); + if (!this.supportsSealedUpload) { + throw new Error( + 'This node is running an older MeshBay and cannot accept an upload ' + + 'from this page. Its operator has to update it.'); + } + const C = window.MeshBayCrypto; + const groupId = (this._connectArgs && this._connectArgs.groupId) || ''; + this._inFlightUploads.add(file.name); + const uploadId = _hex(crypto.getRandomValues(new Uint8Array(16))); const size = chunkSize || UPLOAD_CHUNK_SIZE; const total = Math.max(1, Math.ceil(file.size / size)); let acked = 0; @@ -2091,16 +2124,32 @@ class MeshBayTransport { let failure = null; const acks = []; - this._uploaders.set(file.name, (msg) => { - if (msg.type === 'error') { - failure = new Error(msg.detail || 'Upload refused'); - } else if (msg.stored_as) { - stored = msg; - } + const wake = () => { acked += 1; if (onProgress) onProgress(Math.min(file.size, acked * size), file.size); const waiter = acks.shift(); if (waiter) waiter(); + }; + this._uploaders.set(uploadId, (msg) => { + if (msg.type === 'error') { + failure = new Error(msg.detail || 'Upload refused'); + wake(); + return; + } + // The ack is sealed too — `stored_as` and the folder it landed in name + // the operator's content. Opening it is what makes the result usable, so + // a failure here fails the upload rather than being swallowed: a chat + // attachment that cannot learn its stored name would point at nothing. + C.openGroup(this._gekRaw, 'upload', 'file_upload_ack', groupId, msg) + .then((plain) => { + const payload = msgpack_decode(plain); + if (payload.stored_as) stored = payload; + }) + .catch((e) => { + failure = new Error( + `The node's upload reply did not open under the group key (${e.message})`); + }) + .finally(wake); }); const nextAck = () => new Promise(r => acks.push(r)); @@ -2122,15 +2171,20 @@ class MeshBayTransport { const buf = new Uint8Array( await file.slice(i * size, (i + 1) * size).arrayBuffer()); + // The name, the destination and the bytes go inside the seal together. + // Mirrors `file_upload_wire` in meshbay_common/protocol.py; only the + // fields the node routes on stay outside it. + const sealed = await C.sealGroup( + this._gekRaw, 'upload', 'file_upload', groupId, + msgpack_encode({ filename: file.name, data: buf, + dir: dir || '', root: root || '' })); this._send({ type: 'file_upload', v: '0.1', - filename: file.name, + upload_id: uploadId, chunk_index: i, total_chunks: total, - data: buf, - ...(root ? { root } : {}), - ...(dir ? { dir } : {}), + ...sealed, }); } while (acked < total) { @@ -2138,7 +2192,8 @@ class MeshBayTransport { if (failure) throw failure; } } finally { - this._uploaders.delete(file.name); + this._uploaders.delete(uploadId); + this._inFlightUploads.delete(file.name); } return stored || {}; } @@ -2458,12 +2513,15 @@ class MeshBayTransport { if (msg.type === 'index_sync') { if (this._onIndexSync) this._onIndexSync(opened); - for (const [, handler] of this._pending) { - if (handler._reqType === 'index_sync') { - handler.resolve(opened); - break; - } - } + // The node's first push to a newly connected peer is an index_sync + // nobody asked for, so there is not always a request to resolve. When + // there is, `req_id` says which one — the type match below is what a + // node too old to stamp one leaves us, and it is why two fetches in + // flight at once used to resolve the wrong one. + const handler = opened.req_id !== undefined && opened.req_id !== null + ? this._pending.get(opened.req_id) + : [...this._pending.values()].find(h => h._reqType === 'index_sync'); + if (handler) handler.resolve(opened); return; } if (this._onIndexDelta) this._onIndexDelta(opened); @@ -2550,7 +2608,13 @@ class MeshBayTransport { resolve: (msg) => { clearTimeout(timeout); this._pending.delete(id); resolve(msg); }, reject: (err) => { clearTimeout(timeout); this._pending.delete(id); reject(err); }, }); - this._send(obj); + // The id goes on the wire (MNP 1.1+): a node that understands it stamps + // the reply with it, and _dispatch matches on that alone. It used to be + // local to this map, which is why every reply had to be recognised by + // some field of its own — and why the ones that carry no such field + // reached their caller by luck. An older node ignores the extra key and + // is routed by the per-type fallbacks below, exactly as before. + this._send({ ...obj, req_id: id }); }); } @@ -2590,6 +2654,45 @@ class MeshBayTransport { } _dispatch(msg) { + // A reply that names the request it answers. Nothing below this needs to + // recognise it, and nothing below this may see it: every remaining branch + // exists to identify a reply by some field of its own, which is the job + // this makes unnecessary. + // + // What is left underneath is genuinely unsolicited — a broadcast to every + // connected client, a push, a challenge — or a reply from a node too old + // to stamp one, which is what the per-type keys are for now. + if (msg.req_id !== undefined && msg.req_id !== null) { + this._correlates = true; + // The one exception, and the only one: an index message is sealed under + // the GEK and cannot be handed to its caller until it is opened, which + // is not something this synchronous function can do. Resolving it here + // would give `fetchIndex` the envelope — nonce and ciphertext, no + // entries — and skip `_onIndexSync` entirely. `_queueIndexMessage` + // opens it and then resolves, by this same id. + const sealed = msg.type === 'index_sync' || msg.type === 'index_delta'; + if (!sealed) { + const handler = this._pending.get(msg.req_id); + if (handler) { + handler.resolve(msg); + // The acks whose *broadcast* half their own requester also needs: + // every other client learns the change from the broadcast, and the + // one that asked for it is the only one that would not, because its + // own request swallowed its copy. Same call the keyed `_ack` branch + // below makes, for the same reason. + if (BROADCAST_ACK_TYPES.has(msg.type)) _replayBroadcast(this, msg); + return; + } + // Answers a request that is no longer waiting: it gave up at its own + // timeout, or a reconnect rejected everything in flight. It belongs to + // nobody, and the whole point of this change is that it is not offered + // to somebody else instead. + console.warn('[MeshBay] late reply to req', msg.req_id, '(', msg.type, + ') — nothing waiting'); + return; + } + } + // Two-step admin-op flow (_authorizeAdminOp, ADMIN_OP_TYPES) — resolve // by (op) key before anything below gets a chance to steal it via the // generic "oldest pending" fallback further down. Returns as soon as a @@ -2647,21 +2750,21 @@ class MeshBayTransport { // While an upload is in flight the acks are its own, and there are many of // them: they must not be handed to whatever request happens to be oldest in // the pending map. - if (msg.type === 'file_upload_ack' && this._uploaders.has(msg.filename)) { - this._uploaders.get(msg.filename)(msg); + if (msg.type === 'file_upload_ack' && this._uploaders.has(msg.upload_id)) { + this._uploaders.get(msg.upload_id)(msg); return; } - // An upload refusal names the file it is about, so only that upload fails. - // It did not use to, and there was no way to tell whose error it was, so - // every upload in flight was failed together — send a second file whose - // name the node dislikes and both died. The broadcast is kept for a node - // that does not name it, where guessing wrong is worse than stopping. + // An upload refusal names the upload it is about, so only that upload + // fails. It did not use to, and there was no way to tell whose error it + // was, so every upload in flight was failed together — send a second file + // whose name the node dislikes and both died. The broadcast is kept for a + // refusal that names none, where guessing wrong is worse than stopping. if (msg.type === 'error' && this._uploaders.size) { - if (msg.filename && this._uploaders.has(msg.filename)) { - this._uploaders.get(msg.filename)(msg); + if (msg.upload_id && this._uploaders.has(msg.upload_id)) { + this._uploaders.get(msg.upload_id)(msg); return; } - if (!msg.filename) { + if (!msg.upload_id) { for (const handler of [...this._uploaders.values()]) handler(msg); return; } @@ -2955,10 +3058,10 @@ class MeshBayTransport { return; } - // device_hello_ack ends in `_ack` but is not an admin op, so the branch - // above looks it up under `admin:device_hello`, finds nothing, and drops it - // through to the arrival-order guess. Routed by request type instead: a - // request type deserves a key, and a reply deserves something to key it by. + // device_hello_ack ends in `_ack` but is not an admin op, so the admin + // branch looks it up under `admin:device_hello` and finds nothing. A 2.0 + // node stamps `req_id` and this is never reached; it is the per-type key + // for a node that does not, alongside chat_hist_resp above. if (msg.type === 'device_hello_ack') { for (const [, handler] of this._pending) { if (handler._reqType === 'device_hello') { handler.resolve(msg); return; } @@ -2968,14 +3071,13 @@ class MeshBayTransport { } // Same shape as chat_hist_resp above, and found the same way — by driving - // the panel rather than by reading this file. `chat_keys_resp` answers a - // `chat_keys_req` under a different type string, so without this it fell - // to the arrival-order guess at the end and was handed to whatever was - // oldest in `_pending`. `chat_send_probe.py` caught it on its first run: - // the Videos tab's unanswered `media_meta_req` swallowed the chat keys, - // and the send then waited out its own 30s timeout with the composer - // disabled — which is a frozen Chat tab, the exact defect that harness - // exists for, reappearing one feature later. + // the panel rather than by reading this file. Before `req_id` existed, + // `chat_keys_resp` fell to the arrival-order guess and was handed to + // whatever was oldest in `_pending`; `chat_send_probe.py` caught it on its + // first run, with the Videos tab's unanswered `media_meta_req` swallowing + // the chat keys and the send then waiting out its own 30s timeout with the + // composer disabled. `req_id` is what closes that class now, and this is + // the per-type key for a node that does not stamp one. if (msg.type === 'chat_keys_resp') { for (const [, handler] of this._pending) { if (handler._reqType === 'chat_keys_req') { handler.resolve(msg); return; } @@ -3020,10 +3122,28 @@ class MeshBayTransport { return; } - // Everything above is routed by something in the message. What is left is - // matched by arrival order, which is only ever a guess — and a wrong guess - // here hands one request's answer to another, which then waits for a reply - // that already came. Logged so that guess is visible. + // Everything above is routed by something in the message. What is left + // used to be matched by arrival order — a guess, and a wrong guess hands + // one request's answer to another, which then waits out its own 30s + // timeout for a reply that already came and went. That is how the Chat + // composer, disabled while a send is in flight, could stay disabled for + // thirty seconds on a message the node had already stored. + // + // A node that stamps its replies (`req_id`, handled at the top) has taken + // every one of its answers out of this path, so anything arriving here is + // unsolicited and the guess can only ever be wrong. Dropping it loses + // nothing and stops the theft. + if (this._correlates) { + console.warn('[MeshBay] unsolicited', msg.type, '— dropped (pending:', + this._pending.size, ')'); + return; + } + + // Only a node too old to stamp anything reaches here, where arrival order + // is still the only thing there is. Kept deliberately, and no wider than + // it was: the alternative for such a node is that half the protocol + // (device_list_result, join_result, the handshake's own replies) reaches + // nobody at all. const oldest = this._pending.entries().next(); if (!oldest.done) { const [, handler] = oldest.value; @@ -3276,11 +3396,8 @@ function _decodeMap(buf, view, offset, count) { return [obj, offset]; } -function _b64decode(b64) { - const binary = atob(b64); - const bytes = new Uint8Array(binary.length); - for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i); - return bytes; +function _hex(bytes) { + return [...bytes].map(b => b.toString(16).padStart(2, '0')).join(''); } function _extractDtlsFingerprint(sdp) { diff --git a/packages/meshbay-hub/tests/harness/chat_send_probe.py b/packages/meshbay-hub/tests/harness/chat_send_probe.py index 1f0557b..5f99beb 100644 --- a/packages/meshbay-hub/tests/harness/chat_send_probe.py +++ b/packages/meshbay-hub/tests/harness/chat_send_probe.py @@ -20,8 +20,40 @@ never appeared, while the node had stored it all along. chat_send_probe.py -Prints JSON: `steps`, the state of the panel at each stage, and `log`, what the -transport sent and how the deliberately-unanswered request ended up. +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 @@ -49,9 +81,9 @@ def _page() -> str: 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 test built itself would prove only that the page agrees with the page. + the page built itself would prove only that the page agrees with the page. """ - import msgpack + import msgpack # noqa: F401 (imported for the failure it gives if absent) from meshbay_common.groupbox import PURPOSE_CHAT_KEYS, seal @@ -73,9 +105,7 @@ PAGE_TEMPLATE = r"""<!doctype html><html><head><meta charset=utf-8> </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", - which is what this probe reported the first time it exercised the - encrypted path. --> + 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> @@ -87,37 +117,16 @@ 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 _warn = console.warn, _err = console.error; -console.warn = (...a) => { log.push('warn: ' + a.join(' ')); _warn(...a); }; -console.error = (...a) => { log.push('console error: ' + a.join(' ')); _err(...a); }; const hex = (s) => Uint8Array.from(s.match(/../g) || [], b => parseInt(b, 16)); +const b64 = (buf) => btoa(String.fromCharCode(...new Uint8Array(buf))); -// 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() {} }; - -// Chat is encrypted (MNP 2.0), so a send that is going to come back has to -// seal and sign for real. The device key is generated here rather than stubbed -// — `signBytes` imports a pkcs8 key and WebCrypto will not be fooled — and the -// group key and epoch keys come from Python, which sealed the `chat_keys_resp` -// below exactly as the node does. So this exercises `chatKeys()`, `openGroup`, -// `sealChat` and the real signature, not a model of any of them. -tp._groupId = '__GROUP_ID__'; -tp._gekRaw = hex('__GEK_HEX__'); -tp.chatEpoch = 1; - +// 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 b64 = (buf) => btoa(String.fromCharCode(...new Uint8Array(buf))); -tp._sessionKeys = { - skEdB64: b64(await crypto.subtle.exportKey('pkcs8', kp.privateKey)), -}; -// What `device_hello` sets on a live connection. -tp.devicePk = b64(await crypto.subtle.exportKey('raw', kp.publicKey)); +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 = []; @@ -126,75 +135,72 @@ for (let i = 0; i < 5; i++) { payload: 'message ' + i, timestamp: now - (5 - i) * 60 }); } -// Stands in for the node, answering exactly what webrtc_server.py answers. -// media_meta_req is answered with nothing at all, which is what a refusal -// amounts to for the request that asked: `_do_media_meta_request` sends a bare -// `error` for a file_id the index does not have, and a bare error names no -// request, so it reaches none. -tp._send = (obj) => { - log.push('sent ' + obj.type); - if (obj.type === 'chat_hist') { - setTimeout(() => { - log.push('answering chat_hist'); - tp._dispatch({ type: 'chat_hist_resp', v: '0.2', messages: history, - has_more: false }); - }, 10); - } else if (obj.type === 'chat_keys_req') { - // Sealed under the group key, as `_do_chat_keys_req` sends it. - setTimeout(() => tp._dispatch({ - type: 'chat_keys_resp', v: '2.0', group_id: tp._groupId, - nonce: hex('__KEYS_NONCE_HEX__'), ct: hex('__KEYS_CT_HEX__'), - }), 10); - } else if (obj.type === 'chat_msg') { - // Recorded so the test can assert the message really was sealed and - // signed rather than sent in clear past a composer that let it through. - log.push('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')); - setTimeout(() => tp._dispatch({ type: 'ack', v: '2.0' }), 10); - } -}; +const wait = ms => new Promise(r => setTimeout(r, ms)); -// The panel swallows a send failure into `setInput(text)`, which is right for -// a person and useless for a probe: the symptom is the message not appearing, -// with no reason anywhere. Surfaced here so a failure names itself. -const _sendChat = tp.sendChat.bind(tp); -tp.sendChat = (...a) => _sendChat(...a).catch((e) => { - log.push('sendChat failed: ' + (e && e.message || e)); - throw e; -}); +// 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() { +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" />`; } -render(html`<${Host} />`, document.getElementById('root')); -const out = { steps: [], log }; -const composer = () => document.querySelector('.chat-input'); - -function snap(label) { - const c = composer(); - out.steps.push({ - label, - bubbles: document.querySelectorAll('.chat-bubble').length, - msgs: document.querySelectorAll('.chat-msg').length, - lastText: [...document.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, - }); -} - -const wait = ms => new Promise(r => setTimeout(r, ms)); - -function typeInto(text) { - const c = composer(); +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') @@ -202,22 +208,42 @@ function typeInto(text) { c.dispatchEvent(new Event('input', { bubbles: true })); } -(async () => { +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 Videos tab asked about a file a moment ago and is still waiting. Any - // unanswered request will do; this is the one that was live when the defect - // was found. - tp.fetchMediaMeta('a-file-the-node-refused') - .then(m => log.push('media_meta resolved with ' + m.type), - e => log.push('media_meta rejected: ' + e.message)); + // 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('stale request pending'); + snap('older request pending'); - typeInto('hello'); + typeInto(root, 'hello'); await wait(100); - composer().dispatchEvent(new KeyboardEvent('keydown', + 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 @@ -225,6 +251,14 @@ function typeInto(text) { 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>""" diff --git a/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs b/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs new file mode 100644 index 0000000..0b77e42 --- /dev/null +++ b/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs @@ -0,0 +1,102 @@ +/** + * Does the browser's real `uploadFile` produce frames a real node can open — + * and does it read back what that node actually answered? + * + * Drives **the real `MeshBayTransport` over the real `crypto.js`**. Only the DOM + * and the DataChannel are stand-ins; the msgpack encode, the HKDF, the AES-GCM + * and the whole upload loop are the shipped code. + * + * It exists because a source-reading test cannot show either half of MNP 2.0's + * upload. `test_transport_contracts` can see that `sealGroup` is called; only + * this can show that what comes out opens under `meshbay_common.groupbox` — and + * that the caller of `uploadFile` is told the name the *node* chose, which now + * arrives sealed and would otherwise be `undefined` with nothing to notice it. + * + * node upload_seal_probe.mjs <static-dir> <input.json> + * + * Two modes, because the node half runs in Python between them: + * "send" — run the upload, print every frame it emits, answer nothing + * "receive" — run it again, answering with acks Python built. Their + * `upload_id` is retargeted to this run's; it is outside the + * seal, so the ciphertext stays exactly the one Python produced. + */ +import fs from 'fs'; + +const STATIC = process.argv[2]; +const input = JSON.parse(fs.readFileSync(process.argv[3], 'utf8')); + +for (const level of ['log', 'warn', 'error', 'info', 'debug']) { + console[level] = (...args) => process.stderr.write(args.join(' ') + '\n'); +} + +globalThis.window = globalThis; +globalThis.addEventListener = () => {}; +globalThis.removeEventListener = () => {}; +globalThis.location = { hash: '' }; +globalThis.document = { + addEventListener() {}, removeEventListener() {}, visibilityState: 'visible', +}; + +new Function(fs.readFileSync(`${STATIC}/crypto.js`, 'utf8'))(); +const transportSrc = fs.readFileSync(`${STATIC}/transport.js`, 'utf8'); +new Function(transportSrc)(); +// The msgpack codec is private to transport.js — pulled out the same way the +// groupbox parity harness pulls out sealGroup, so this probe encodes and +// decodes with the codec that is actually shipped rather than a second one. +const { msgpack_encode, msgpack_decode } = + new Function(transportSrc + '\nreturn { msgpack_encode, msgpack_decode };')(); + +const hex = (s) => Uint8Array.from(s.match(/../g).map((b) => parseInt(b, 16))); +const toHex = (u8) => + Array.from(u8).map((b) => b.toString(16).padStart(2, '0')).join(''); + +// Just enough of a File: a name, a size, and slices that yield ArrayBuffers. +const bytes = hex(input.file.data); +const file = { + name: input.file.name, + size: bytes.length, + slice(a, b) { + const part = bytes.slice(a, b); + return { arrayBuffer: async () => part.buffer.slice( + part.byteOffset, part.byteOffset + part.byteLength) }; + }, +}; + +const tp = new window.MeshBayTransport('', 'token'); +tp._connected = true; +tp._channel = { readyState: 'open', bufferedAmount: 0, send() {}, close() {} }; +tp._pc = { close() {} }; +tp._gekRaw = hex(input.gek); +tp._connectArgs = { groupId: input.group_id }; +tp._nodeVersion = input.node_version; + +const frames = []; +let uploadId = null; +tp._send = (msg) => { + frames.push(toHex(msgpack_encode(msg))); + if (msg.upload_id) uploadId = msg.upload_id; + if (input.mode !== 'receive') return; + // Answer as the node did, on the next turn of the loop so the send path + // finishes first — which is also how a real ack arrives. + const ack = msgpack_decode(hex(input.acks[msg.chunk_index])); + ack.upload_id = uploadId; + // Through the real `_dispatch`, so the routing under test — matching an + // ack to its uploader by `upload_id` — is the shipped one. + setImmediate(() => tp._dispatch(ack)); +}; + +const out = { frames, mode: input.mode }; +const done = tp.uploadFile(file, { chunkSize: input.chunk_size, + dir: input.dir, root: input.root }); + +if (input.mode === 'receive') { + done.then((stored) => { out.state = 'resolved'; out.stored = stored; }) + .catch((e) => { out.state = 'rejected'; out.message = e.message; }) + .finally(() => { out.upload_id = uploadId; + process.stdout.write(JSON.stringify(out)); }); +} else { + // Nothing will answer, so let the send loop run itself out and report. + done.catch((e) => { out.state = 'rejected'; out.message = e.message; }); + setTimeout(() => { out.upload_id = uploadId; + process.stdout.write(JSON.stringify(out)); }, 250); +} diff --git a/packages/meshbay-hub/tests/test_chat_send.py b/packages/meshbay-hub/tests/test_chat_send.py index 250c9a3..5db8f26 100644 --- a/packages/meshbay-hub/tests/test_chat_send.py +++ b/packages/meshbay-hub/tests/test_chat_send.py @@ -1,38 +1,29 @@ """ -Sending a chat message must come back — and must go out encrypted. +Sending a chat message must come back — accepted or refused. -The node answers a chat message with a bare `{"type": "ack"}` — no request id, -no type of its own — so `_dispatch` had nothing to match it on and left it to -the arrival-order guess at the end of the function. That guess is wrong as soon -as anything else this browser asked for is still waiting: the ack was handed to -*that* request, and the send waited out `_sendAndWait`'s 30s timeout. Since the -composer is disabled while a send is in flight, the Chat tab stopped taking -clicks and keys, the message never appeared — and it was there on the next -visit, because the node had stored it and answered. +The node's replies to a chat message name no request. The acceptance is a bare +`{"type": "ack"}`; the refusal is a bare `{"type": "error"}`, and it is not a +special case — `_dispatch_message`'s catch-all answers *every* failure that +way, and 238 of webrtc_server.py's 240 error sends name nothing either. So +`_dispatch` had nothing to match either reply on and left both to the +arrival-order guess at the end of the function. -An outstanding request is the ordinary case, not a rare one: the node refuses -an unknown file_id with a bare `error`, which names no request either, so a -Videos tab that asked about a file the index no longer has leaves a -`media_meta_req` in `_pending` for a full 30s. +That guess is wrong as soon as anything else this browser asked for is still +waiting, which is the ordinary case rather than a rare one: a `music_meta_req` +sits in `_pending` for as long as the third-party lookup behind it takes, and +that was measured live at over 100 seconds with the service failing. The reply +went to *that* request, and the send waited out `_sendAndWait`'s 30s timeout. +Since the composer is disabled while a send is in flight, the Chat tab stopped +taking clicks and keys, and the message never appeared. -None of that is visible in `chat-app.js`, where every line is correct, so this +The ack half was fixed by matching on request type. The refusal half could not +be: an `error` has no type of its own to match on. `req_id` is what closed it — +the caller's id, stamped on the reply by the node — so this now drives both +shapes of answer. + +None of it is visible in `chat-app.js`, where every line is correct, so this drives the real panel over the real transport in a browser rather than reading either source. - -Extended for MNP 2.0, where a send seals and signs before it goes anywhere. -That turned out to matter twice on its first run: - - * `chat_keys_resp` answers a `chat_keys_req` under a different type string, - so it fell through to the arrival-order guess and was handed to the very - `media_meta_req` this probe leaves outstanding — the original defect, one - feature later, in a message type that did not exist when it was written. - * `_asText` had been deleted along with an unrelated helper beside it. Its - only caller is inside `_openChatMessage`, whose rejection the panel - swallows, so the whole conversation rendered empty with nothing in the - console and the node answering perfectly. - -Neither is visible in any source file, and neither would have been caught by a -test that reads one. """ import json import shutil @@ -52,7 +43,7 @@ pytestmark = pytest.mark.skipif( @pytest.fixture(scope="module") def probe(): - # `sys.executable`, not a bare "python3": the harness now imports + # `sys.executable`, not a bare "python3": the harness imports # `meshbay_common` to seal the chat keys the way the node does, and the # system interpreter has neither that nor msgpack. The other probes get # away with "python3" because they import nothing from this project. @@ -60,51 +51,81 @@ def probe(): capture_output=True, timeout=180) assert run.returncode == 0, run.stderr.decode()[-2000:] data = json.loads(run.stdout.decode()) - return data, {s["label"]: s for s in data["steps"]} + steps = {sc["name"]: {s["label"]: s for s in sc["steps"]} + for sc in data["scenarios"]} + return data, steps + +@pytest.mark.parametrize("reply", ["ack", "error"]) +def test_the_composer_comes_back(probe, reply): + """The one thing a person sees: the tab is usable again. -def test_the_composer_comes_back(probe): - """The one thing a person sees: the tab is usable again.""" + Both answers have to release it. A refusal that reaches nobody leaves the + composer disabled exactly as long as an acceptance that reaches nobody — + the composer is not waiting for good news, it is waiting for an answer. + """ _, steps = probe - assert steps["stale request pending"]["composerDisabled"] is False, ( + assert steps[reply]["older request pending"]["composerDisabled"] is False, ( "the composer was already unusable before the send") - assert steps["after send"]["composerDisabled"] is False, ( + assert steps[reply]["after send"]["composerDisabled"] is False, ( "the composer is still disabled well inside the 30s request timeout -- " "the send never came back, which is what reads as a frozen Chat tab") -def test_the_message_is_displayed(probe): +def test_an_accepted_message_is_displayed(probe): """A sent message appears at once, not on the next visit to the tab.""" _, steps = probe - before = steps["stale request pending"]["bubbles"] - assert steps["after send"]["bubbles"] == before + 1, ( + before = steps["ack"]["older request pending"]["bubbles"] + assert steps["ack"]["after send"]["bubbles"] == before + 1, ( "the message was not added to the conversation") - assert steps["after send"]["lastText"] == "hello" - assert steps["after send"]["composerValue"] == "", ( + assert steps["ack"]["after send"]["lastText"] == "hello" + assert steps["ack"]["after send"]["composerValue"] == "", ( "the text came back into the composer, so the send was treated as failed") -def test_the_ack_is_not_handed_to_another_request(probe): - """The other half of the same defect: whatever was waiting got the ack and - carried on with a reply to a question it never asked.""" +def test_a_refused_message_is_not_displayed_as_sent(probe): + """The other direction, and the one routing this correctly makes possible. + + While a refusal reached the wrong caller it did not matter what `sendChat` + would have done with it. Now that it arrives, a message the node rejected + must not appear in the conversation as though it had been stored — it must + come back into the composer, where a person can see it did not go. + """ + _, steps = probe + before = steps["error"]["older request pending"]["bubbles"] + assert steps["error"]["after send"]["bubbles"] == before, ( + "a refused message was added to the conversation anyway") + assert steps["error"]["after send"]["composerValue"] == "hello", ( + "the refused text was dropped instead of being handed back") + + +@pytest.mark.parametrize("reply", ["ack", "error"]) +def test_the_reply_is_not_handed_to_another_request(probe, reply): + """The other half of the same defect: whatever was waiting got the reply + and carried on with an answer to a question it never asked.""" data, _ = probe - assert "media_meta resolved with ack" not in data["log"], ( - "the chat ack was routed to the pending media_meta_req -- that request " - "now believes it has an answer, and the chat send is waiting for a " - "reply that already arrived") + stolen = [line for line in data["log"] if line.startswith(f"{reply}: music_meta")] + assert not stolen, ( + f"the chat {reply} was routed to the pending music_meta_req ({stolen}) -- " + "that request now believes it has an answer, and the chat send is " + "waiting for a reply that already arrived") -def test_the_message_goes_out_sealed_and_signed(probe): +@pytest.mark.parametrize("reply", ["ack", "error"]) +def test_the_message_goes_out_sealed_and_signed(probe, reply): """ - What actually left the browser. A composer that let a plaintext message - through would be refused by the node, but the refusal arrives after the - fact and reads as "the message did not send" — so assert the shape here, - where the reason is visible. + What actually left the browser (MNP 2.0). + + Asserted for both scenarios because the composer is what decides to send: + a client that fell back to plaintext when something went wrong would be + refused by the node, but the refusal arrives after the fact and reads as + "the message did not send". There is no plaintext form on the wire. """ data, _ = probe - sent = [line for line in data["log"] if line.startswith("chat_msg ")] - assert sent, ("no chat_msg reached the stand-in node — the send did not " - f"complete. log: {data['log']}") + sent = [line for line in data["log"] + if line.startswith(f"{reply}: chat_msg ")] + assert sent, (f"no chat_msg reached the stand-in node in the {reply} " + f"scenario — the send did not complete. log: {data['log']}") assert "format=1" in sent[0], "the message was not sealed" assert "sig=64" in sent[0], "the message was not signed" assert "ct=" in sent[0] and "ct=0" not in sent[0], "there was no ciphertext" @@ -115,12 +136,25 @@ def test_the_message_goes_out_sealed_and_signed(probe): def test_the_chat_keys_answer_is_not_handed_to_another_request(probe): """ - The original defect's shape, in the message type that carries the group's - chat keys. An unanswered request is the ordinary case, not a rare one, and - the one this probe leaves outstanding swallowed the keys on the first run. + The same defect this file exists for, in the message type that carries the + group's chat keys — which did not exist when it was written, and which a + send now depends on. It went astray on the probe's first encrypted run. """ data, _ = probe - assert not any("media_meta resolved with chat_keys_resp" in line - for line in data["log"]), ( - "chat_keys_resp was routed by arrival order and handed to the stale " - "media_meta_req — the send then waits out its own 30s timeout") + stolen = [line for line in data["log"] if "music_meta resolved with" in line] + assert not stolen, ( + f"a reply was routed to the pending music_meta_req ({stolen}) — the " + "send then waits out its own 30s timeout with the composer disabled") + + +def test_history_still_renders(probe): + """ + Not about sending at all, and here because it broke without a sound: + `_asText` was deleted with an unrelated helper beside it, its only caller + sits inside a promise the panel catches, and every conversation rendered + empty with the node answering perfectly. + """ + _, steps = probe + assert steps["ack"]["older request pending"]["bubbles"] == 5, ( + "the five history messages did not render — the panel swallows a " + "failure in the transport's message reader, so this is silent") diff --git a/packages/meshbay-hub/tests/test_index_seal_client.py b/packages/meshbay-hub/tests/test_index_seal_client.py index ca2c7a2..e1135da 100644 --- a/packages/meshbay-hub/tests/test_index_seal_client.py +++ b/packages/meshbay-hub/tests/test_index_seal_client.py @@ -45,10 +45,15 @@ def _frame(msg: dict) -> str: return (struct.pack(">I", len(body)) + body).hex() -def _sync_frame(gek: bytes, *names: str, version: int = 3) -> str: +def _sync_frame(gek: bytes, *names: str, version: int = 3, + req_id: int | None = None) -> str: payload = {"version": version, "entries": [_entry(n) for n in names], "dirs": ["library"], "roots": [{"name": "library"}]} + # `req_id` is what a current node stamps on a *reply*; the push it sends a + # newly connected peer answers no request and carries none. Both shapes + # arrive here, and only one of them may resolve a waiting fetchIndex. return _frame({"type": "index_sync", "v": "1.0", "group_id": GROUP, + **({"req_id": req_id} if req_id is not None else {}), **seal(gek, PURPOSE_INDEX, "index_sync", GROUP, payload)}) @@ -141,3 +146,26 @@ def test_deltas_are_applied_in_arrival_order(): assert [e["additions"][0] for e in out["events"][1:]] == [ "added-0.mkv", "added-1.mkv", "added-2.mkv"] assert [e["base_version"] for e in out["events"][1:]] == [3, 4, 5] + + +def test_a_sealed_reply_is_opened_before_it_reaches_its_caller(): + """A stamped index_sync must not be short-circuited by its `req_id`. + + Every other reply a node stamps is resolved straight out of the pending + map, which is the whole point of the id. An index message cannot be: it is + sealed, opening it is asynchronous, and `_dispatch` is not. Handing it over + on the strength of the id alone gives `fetchIndex` the envelope — nonce and + ciphertext, no entries — and never calls `onIndexSync` at all. + + `req_id` is 0 here because it is the transport's first request, and a + falsy id is exactly the one a presence check gets wrong. + """ + out = _run([_sync_frame(GEK, "a-film.mkv", req_id=0)]) + + assert [e["event"] for e in out["events"]] == ["index_sync"], ( + "the consumer was never told about an index that arrived as a reply") + assert out["events"][0]["entries"] == ["a-film.mkv"] + assert out["events"][0]["hasCiphertext"] is False + assert out["fetchIndex"]["state"] == "resolved" + assert out["fetchIndex"]["entries"] == ["a-film.mkv"], ( + "the caller was handed the sealed envelope instead of the index") diff --git a/packages/meshbay-hub/tests/test_transport_contracts.py b/packages/meshbay-hub/tests/test_transport_contracts.py index b462242..879062b 100644 --- a/packages/meshbay-hub/tests/test_transport_contracts.py +++ b/packages/meshbay-hub/tests/test_transport_contracts.py @@ -307,26 +307,60 @@ def test_no_setter_survives_the_state_it_belonged_to(): # ── Parallel uploads ────────────────────────────────────────────────────────── -def test_an_upload_refusal_names_the_file_it_is_about(transport): +def test_an_upload_refusal_names_the_upload_it_is_about(transport): """Reported 2026-08-16: a second upload started in parallel killed both. - An error used to carry no filename, so the client could not tell whose it - was and failed every upload in flight — one name the node disliked took the - other file with it. The node names the file now, and only that upload stops. + An error used to carry nothing identifying, so the client could not tell + whose it was and failed every upload in flight — one name the node disliked + took the other file with it. + + The node named the *file* until MNP 2.0 and names the `upload_id` now: the + filename moved inside the seal, and echoing it in clear so the two sides + could match on it would give back precisely what sealing the upload is for. + The property is unchanged — one refusal, one failed upload. """ body = transport[transport.index("if (msg.type === 'error' && this._uploaders.size)"):] body = body[:body.index("\n if (msg.type === 'chat_msg'")] - assert "this._uploaders.has(msg.filename)" in body, ( + assert "this._uploaders.has(msg.upload_id)" in body, ( "a named refusal must reach one uploader, not all of them") - assert "if (!msg.filename)" in body, ( + assert "if (!msg.upload_id)" in body, ( "an unnamed error from an older node must still stop everything — " "guessing which upload it belongs to would be worse") -def test_uploads_are_tracked_per_file(transport): +def test_uploads_are_tracked_per_upload(transport): """Acks interleave when two files are in flight.""" assert "this._uploaders = new Map()" in transport - assert "this._uploaders.set(file.name" in transport + assert "this._uploaders.set(uploadId" in transport + # And the "already being uploaded" guard still speaks in filenames, because + # that is what the caller passed and what it would recognise in the error. + assert "this._inFlightUploads.has(file.name)" in transport + + +def test_the_upload_itself_is_sealed(transport): + """ + MNP 2.0. The filename, the destination and the bytes go inside the seal + together — sealing the content and announcing the name beside it would be + theatre — and only what the node routes on stays outside. + """ + start = transport.index(" async uploadFile(file,") + body = transport[start:transport.index("\n /** Create a directory", start)] + assert "sealGroup(" in body and "'file_upload'" in body, ( + "the upload must be sealed under the group key") + assert "openGroup(" in body and "'file_upload_ack'" in body, ( + "the ack carries the stored name and must be opened, not read") + # The message the node actually receives: everything between `this._send({` + # and its close. Read on its own, because the same field names appear a few + # lines above inside `msgpack_encode({...})`, which is the sealed half. + sent = body[body.index("this._send({"):] + sent = sent[:sent.index("});")] + assert "filename" not in sent, "the filename is on the message in clear" + assert "data" not in sent, "the bytes are on the message in clear" + assert "dir" not in sent and "root" not in sent, ( + "the destination is on the message in clear") + assert "...sealed," in sent, "the message must carry the sealed pair" + assert "supportsSealedUpload" in body, ( + "an older node must be refused before a chunk is sent, not after") # ── MNP 1.0: the sealed handshake ack ──────────────────────────────────────── diff --git a/packages/meshbay-hub/tests/test_upload_seal_client.py b/packages/meshbay-hub/tests/test_upload_seal_client.py new file mode 100644 index 0000000..d6f9156 --- /dev/null +++ b/packages/meshbay-hub/tests/test_upload_seal_client.py @@ -0,0 +1,169 @@ +""" +The browser half of MNP 2.0's sealed upload, measured rather than read. + +`test_upload_sealed.py` (node side) proves the node opens what the shared +encoder produces and refuses everything else. This proves the *shipped browser +code* produces it — and, the part that matters more, that a caller of +`uploadFile` is still told the name the node stored the file under, which now +arrives sealed and would otherwise be `undefined` with nothing on screen to say +so: a chat attachment would point at a file that is not there. + +Driven through `harness/upload_seal_probe.mjs`, which runs the shipped +`transport.js` over the shipped `crypto.js`. The node half in between is the +real `_do_file_upload`, writing to a real directory. + +A source-reading test can see that `sealGroup` is called. Only this can see +whether what comes out of it opens. +""" + +import json +import shutil +import subprocess +import tempfile +from pathlib import Path + +import msgpack +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_common.protocol import MNP +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roots import RootSet +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" +PROBE = Path(__file__).resolve().parent / "harness" / "upload_seal_probe.mjs" + +pytestmark = pytest.mark.skipif( + shutil.which("node") is None or not PROBE.exists(), + reason="node unavailable — the client half cannot be measured", +) + +GROUP = "g-upload-probe" +CHUNK = 32 +BODY = bytes(range(256)) * 3 # 768 bytes → 24 chunks of 32 + + +def _run_probe(payload: dict) -> dict: + with tempfile.TemporaryDirectory() as d: + f = Path(d) / "input.json" + f.write_text(json.dumps(payload)) + proc = subprocess.run( + ["node", str(PROBE), str(STATIC), str(f)], + capture_output=True, text=True, timeout=60, + ) + if proc.returncode != 0 or not proc.stdout: + pytest.fail(f"upload probe failed:\n{proc.stderr}") + return json.loads(proc.stdout) + + +def _node_session(tmp_path: Path, gek: bytes) -> WebRTCPeerSession: + root = tmp_path / "library" + root.mkdir() + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = { + "roots": RootSet.build([{"path": str(root), "name": "library", + "writable": True}]), + "index": index, "sk_node": index.sk_node, "gek": gek, + } + session._group_id = GROUP + session._user_id = "prober" + session._pk_user = "" + session._uploads = {} + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _probe_input(gek: bytes, mode: str, **extra) -> dict: + return { + "mode": mode, "gek": gek.hex(), "group_id": GROUP, + "node_version": "2.0", "chunk_size": CHUNK, "dir": "library", + "root": "library", + "file": {"name": "holiday.jpg", "data": BODY.hex()}, + **extra, + } + + +@pytest.fixture(scope="module") +def _gek(): + return generate_gek() + + +@pytest.fixture(scope="module") +def _sent(_gek): + """Every frame the shipped `uploadFile` puts on the wire, unanswered.""" + return _run_probe(_probe_input(_gek, "send")) + + +def test_the_browser_puts_no_filename_and_no_content_on_the_wire(_sent): + """The whole point, measured on the bytes rather than read off the source.""" + assert _sent["frames"], "the client sent nothing" + for hexframe in _sent["frames"]: + raw = bytes.fromhex(hexframe) + assert b"holiday.jpg" not in raw, "the filename is on the wire in clear" + msg = msgpack.unpackb(raw, raw=False) + assert set(msg) == {"type", "v", "upload_id", "chunk_index", + "total_chunks", "nonce", "ct"} + assert msg["type"] == MNP.FILE_UPLOAD + + +def test_a_real_node_opens_what_the_real_browser_sealed(tmp_path, _gek, _sent): + """ + End to end: the shipped browser encoder into the shipped node handler, with + the file that lands on disk as the assertion. A mismatch in the HKDF salt, + the AAD encoding or the payload shape shows up here as "did not open" — and + nowhere else until somebody tries to upload something. + """ + session = _node_session(tmp_path, _gek) + for hexframe in _sent["frames"]: + session._do_file_upload(msgpack.unpackb(bytes.fromhex(hexframe), raw=False)) + + errors = [m for m in session.sent if m.get("type") == "error"] + assert not errors, f"the node refused a frame the browser built: {errors[:1]}" + + root = session._ctx["roots"].roots[0].path + assert (root / "holiday.jpg").read_bytes() == BODY + assert not list(root.glob("*.part")), "a temp file was left behind" + + +def test_the_caller_is_told_the_name_the_node_chose(tmp_path, _gek, _sent): + """ + `stored_as` is sealed now, so reading it takes a decrypt that can fail + silently. It must not: the node finds a free name rather than replacing + anything, and a chat attachment that never learns which name points at + nothing. + + The acks below are the ones the node really produced — only their + `upload_id`, which is outside the seal, is retargeted to the second probe + run's own upload. + """ + session = _node_session(tmp_path, _gek) + # A file of that name is already there, so the node has to choose another. + (session._ctx["roots"].roots[0].path / "holiday.jpg").write_bytes(b"someone else's") + + for hexframe in _sent["frames"]: + session._do_file_upload(msgpack.unpackb(bytes.fromhex(hexframe), raw=False)) + acks = [msgpack.packb(m, use_bin_type=True).hex() + for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK] + assert len(acks) == len(_sent["frames"]) + + result = _run_probe(_probe_input(_gek, "receive", acks=acks)) + assert result["state"] == "resolved", result.get("message") + assert result["stored"]["stored_as"] == "holiday (2).jpg" + assert result["stored"]["dir"] == "library" + + +def test_the_client_refuses_an_older_node_before_sending_a_chunk(_gek): + """ + A 1.x node would answer "Missing filename or data" — an error about the + wrong thing, naming no upload, which fails every upload in flight. Asked + first instead, and nothing goes on the wire. + """ + result = _run_probe(_probe_input(_gek, "receive", acks=[], + node_version="1.1")) + assert result["state"] == "rejected" + assert "older MeshBay" in result["message"] + assert result["frames"] == [], "a chunk was sent to a node that cannot open it" diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py index b22b8df..af87b70 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py @@ -317,22 +317,3 @@ class QuicChunkClient: # substitutes it would otherwise choose which key we decrypt with. return file_chunk_plaintext( self._gek, msg, file_hash=bytes.fromhex(file_id)) - - async def fetch_stream_segment( - self, file_id: str, segment_index: int, segment_duration: int = 4, - ) -> bytes: - """Fetch one HLS segment (MPEG-TS bytes) over QUIC.""" - sid = self._new_stream() - self._proto._send(sid, { - "type": MNP.STREAM_SEGMENT, - "v": MNP_VERSION, - "file_id": file_id, - "segment_index": segment_index, - "segment_duration": segment_duration, - }) - msg = await self._proto._recv(sid, timeout=30.0) - - if msg.get("type") == "error": - raise LookupError(msg.get("detail", "Unknown error")) - - return base64.b64decode(msg["data_b64"]) diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 284b488..e34153c 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -22,7 +22,6 @@ import base64 import logging import os import struct -import subprocess import uuid from pathlib import Path from typing import Any, Callable @@ -54,7 +53,6 @@ from meshbay_common.groupbox import PURPOSE_ACK, seal from meshbay_common.protocol import MNP, file_chunk_wire from meshbay_node.indexer import GroupIndex from meshbay_node.transport.wire import index_sync_message -from meshbay_node import platform log = logging.getLogger(__name__) @@ -62,16 +60,6 @@ CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 ALPN = ["meshbay-mnp"] -# ffmpeg is spawned per STREAM_SEGMENT request, and `_extract_segment` runs -# `subprocess.run` synchronously — so without a bound, an authenticated peer can -# both fork-bomb the node and block its event loop for up to 30 s per request -# (finding M2c). Extraction now runs in a thread and passes through this -# semaphore. Small on purpose: the QUIC path has no shipping client yet, this is -# parity work with the WebRTC transcode cap. -_MAX_CONCURRENT_SEGMENTS = 4 -_segment_sem = asyncio.Semaphore(_MAX_CONCURRENT_SEGMENTS) - - class Denylist: """ Denylist for revoked users, groups and invalidated JWTs. @@ -247,8 +235,6 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._do_index_sync_sync(stream_id) elif mtype == MNP.FILE_REQUEST: self._do_file_request_sync(stream_id, msg) - elif mtype == MNP.STREAM_SEGMENT: - self._spawn(self._do_stream_segment(stream_id, msg)) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message_sync(stream_id, msg) elif mtype == MNP.PING: @@ -435,49 +421,6 @@ class _MNPServerProtocol(QuicConnectionProtocol): ctx["gek"], file_path, chunk_index, file_hash, entry.id) self._send(stream_id, chunk_data) - async def _do_stream_segment(self, stream_id: int, msg: dict) -> None: - """ - Extract and serve one segment via ffmpeg — off the event loop and behind - a concurrency bound, so one request can neither stall the whole node nor - fork-bomb it (finding M2c). The WebRTC path has had both since Phase 11.5. - """ - try: - ctx = self._group_ctx() - file_id = msg["file_id"] - segment_index = msg["segment_index"] - segment_duration = msg.get("segment_duration", 4) - - entry = ctx["index"].get_entry(file_id) - if not entry: - self._send(stream_id, {"type": "error", "detail": "File not found"}) - return - - file_path = entry_abs_path(ctx["roots"], entry) - if not file_path.exists(): - self._send(stream_id, {"type": "error", "detail": "File not on disk"}) - return - - start_time = segment_index * segment_duration - loop = asyncio.get_event_loop() - async with _segment_sem: - segment_data = await loop.run_in_executor( - None, _extract_segment, file_path, start_time, segment_duration) - if segment_data is None: - self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) - return - - self._send(stream_id, { - "type": MNP.STREAM_SEGMENT, - "v": MNP_VERSION, - "file_id": file_id, - "segment_index": segment_index, - "data_b64": base64.b64encode(segment_data).decode(), - "size": len(segment_data), - }) - except Exception as e: - log.error("stream_segment: %s", e) - self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) - def _do_chat_message_sync(self, stream_id: int, msg: dict) -> None: """ Store a chat message and broadcast it to the rest of THIS group. @@ -548,25 +491,6 @@ def _read_and_encrypt( return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) -def _extract_segment(file_path: Path, start_time: float, duration: float) -> bytes | None: - """Extract one HLS segment via ffmpeg. Returns MPEG-TS bytes or None on failure.""" - try: - result = subprocess.run( - [platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", - "-ss", str(start_time), - "-i", str(file_path), - "-t", str(duration), - "-c:v", "copy", "-c:a", "copy", - "-f", "mpegts", "pipe:1"], - capture_output=True, timeout=30, - ) - if result.returncode == 0 and result.stdout: - return result.stdout - return None - except Exception: - return None - - # ── QuicChunkServer ──────────────────────────────────────────────────────────── class QuicChunkServer: diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 9774831..4e4a23f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -24,6 +24,7 @@ Signaling flow (handled externally by the hub): import asyncio import base64 +import contextvars import hashlib import hmac import logging @@ -105,9 +106,18 @@ from meshbay_common.join import ( ROLE_OPERATOR, join_transcript, ) -from meshbay_common.protocol import MNP, chunk_ciphertext, file_chunk_wire -from meshbay_common.chatbox import NONCE_LEN as CHAT_NONCE_LEN, SIG_LEN as CHAT_SIG_LEN -from meshbay_node.chat import FORMAT_PLAIN, FORMAT_SEALED_V1, ReplayedMessage +from meshbay_common.chatbox import ( + NONCE_LEN as CHAT_NONCE_LEN, + SIG_LEN as CHAT_SIG_LEN, +) +from meshbay_common.protocol import ( + MNP, + chunk_ciphertext, + file_chunk_wire, + file_upload_ack_wire, + file_upload_payload, +) +from meshbay_node.chat import FORMAT_SEALED_V1, ReplayedMessage from meshbay_node.transport.wire import index_sync_message from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer @@ -263,6 +273,32 @@ def _pack(obj: dict) -> bytes: _WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1" _WEBRTC_TRACE_INTERVAL_S = 30.0 +# The request this session is currently answering, as (session, req_id). +# +# MNP has never carried a correlation id: a reply named its own type and +# nothing else, so a client with more than one request outstanding had to guess +# which one a message answered — by arrival order, for every reply the client +# could not key off a field of its own. The guess is wrong whenever two replies +# reorder, and catastrophically wrong for the replies that name *nothing*: this +# module sends `{"type": "error"}` from 240 places and two of them name what +# they are about. A refusal therefore reached no caller at all, and the request +# it belonged to waited out the client's 30s timeout while some unrelated +# request was resolved with the refusal instead. Live symptom, found 2026-09-06: +# the Chat composer is disabled while a send is in flight, so a chat message +# whose reply went astray froze the tab for 30 seconds. +# +# `req_id` closes it: whatever the caller put on the request is stamped on the +# reply. A ContextVar rather than a parameter because the alternative is +# threading an argument through all 240 send sites — and asyncio copies the +# current context into a task, so a handler that `_spawn`s its real work still +# answers under the id of the request that started it. +# +# The session is held alongside the id because a handler may send to *other* +# sessions as well as its own (a chat broadcast, an index push): those are not +# replies to anything and must not be stamped. _send checks the owner. +_REPLY_TO: contextvars.ContextVar[tuple] = contextvars.ContextVar( + "meshbay_reply_to", default=(None, None)) + class _DataChannelBuffer: """ @@ -416,6 +452,22 @@ class WebRTCPeerSession: ) def _handle_message(self, msg: dict) -> None: + """Answer one MNP message, under the correlation id it carries. + + The id is published for the whole handler — see _REPLY_TO — so that + every reply _send puts on the wire, including the ones a spawned task + sends much later and the generic refusal below, names the request it + answers. Resetting on the way out only clears it for *this* call: a + task spawned in between captured its own copy of the context when it + was created and keeps answering under the right id. + """ + token = _REPLY_TO.set((self, msg.get("req_id"))) + try: + self._dispatch_message(msg) + finally: + _REPLY_TO.reset(token) + + def _dispatch_message(self, msg: dict) -> None: mtype = msg.get("type") log.debug("WebRTC recv: %s", mtype) try: @@ -459,8 +511,6 @@ class WebRTCPeerSession: # Chunks are matched by file and index on the client, so # answering out of order is safe. self._spawn(self._do_file_request(msg)) - elif mtype == MNP.STREAM_SEGMENT: - self._do_stream_segment(msg) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message(msg) elif mtype == MNP.CHAT_HISTORY: @@ -3172,7 +3222,14 @@ class WebRTCPeerSession: def _group_ctx(self) -> dict: if "groups" in self._ctx and self._group_id: - return self._ctx["groups"][self._group_id] + # `.get`, not a bare subscript. A config reload removes a group + # from this map (daemon.py's reload does `groups_ctx.pop`) while + # sessions connected to it are still open, and the next request + # any of them made raised KeyError into _dispatch_message's + # catch-all. An absent group now reads the way an unconfigured + # one already does — the handlers all test for what they need — + # instead of failing every request the session has left. + return self._ctx["groups"].get(self._group_id) or {} return self._ctx def _indexing_status(self) -> dict: @@ -4005,71 +4062,6 @@ class WebRTCPeerSession: "director": director, } - def _do_stream_segment(self, msg: dict) -> None: - self._spawn(self._do_stream_segment_async(msg)) - - async def _do_stream_segment_async(self, msg: dict) -> None: - """ - Legacy HLS segment extraction (superseded by stream_req/MSE). - - Finding H6: this ran subprocess.run(..., timeout=30) directly inside the - event loop, so a single request stalled the whole daemon — every peer, - every group — for up to thirty seconds. Now async and under the same - transcode semaphore as _stream_video. - """ - ctx = self._group_ctx() - file_id = msg["file_id"] - segment_index = msg["segment_index"] - segment_duration = msg.get("segment_duration", 4) - - entry = ctx["index"].get_entry(file_id) - if not entry: - self._send({"type": "error", "detail": "File not found"}) - return - - file_path = entry_abs_path(ctx["roots"], entry) - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) - return - - sem = self._transcode_semaphore() - - try: - async with sem: - proc = await asyncio.create_subprocess_exec( - platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", - "-ss", str(segment_index * segment_duration), - "-i", str(file_path), - "-t", str(segment_duration), - "-c:v", "copy", "-c:a", "copy", - "-f", "mpegts", "pipe:1", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.DEVNULL, - ) - try: - stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30) - except asyncio.TimeoutError: - proc.kill() - await proc.wait() - self._send({"type": "error", "detail": "Segment extraction timed out"}) - return - if proc.returncode != 0 or not stdout: - self._send({"type": "error", "detail": "Segment extraction failed"}) - return - segment_data = stdout - except Exception: - self._send({"type": "error", "detail": "Segment extraction failed"}) - return - - self._send({ - "type": MNP.STREAM_SEGMENT, - "v": MNP_VERSION, - "file_id": file_id, - "segment_index": segment_index, - "data_b64": base64.b64encode(segment_data).decode(), - "size": len(segment_data), - }) - def _do_chat_message(self, msg: dict) -> None: """ Store one message and hand it to everyone else in this group. @@ -4417,27 +4409,88 @@ class WebRTCPeerSession: self._send(resp) def _do_file_upload(self, msg: dict) -> None: + """ + One chunk of an upload, sealed under the group key (MNP 2.0). + + Sealing this direction is not symmetry for its own sake. Downloads have + been under a GEK-derived key since the beginning; uploads carried the + filename and the raw bytes in plain msgpack, so the same file was + ciphertext leaving a node and plaintext arriving at one. The node holds + the GEK for its own group, so it opens the payload here — before it + decides a destination, before it touches the disk — and refuses a chunk + that does not open. + + `upload_id` is the correlation key and stays in clear; `filename`, `dir` + and `root` moved inside the seal, which is why every refusal below names + the upload rather than the file. A `code` says which refusal it is, and + the client already knows what it sent. + """ ctx = self._group_ctx() - filename = msg.get("filename", "") - chunk_index = msg.get("chunk_index", 0) - total_chunks = msg.get("total_chunks", 1) - data = msg.get("data") + upload_id = str(msg.get("upload_id") or "")[:64] + + gek = ctx.get("gek") + if not gek: + self._send({"type": "error", "detail": "Group encryption not initialized", + "code": "no_group_key", "upload_id": upload_id}) + return - if not filename or data is None: - self._send({"type": "error", "detail": "Missing filename or data", - "filename": filename}) + try: + payload = file_upload_payload(gek, self._group_id or "", msg) + except Exception: + # Deliberately one answer for "not sealed at all" and "sealed wrong": + # distinguishing them tells a peer which of the two it got right. + # An MNP 1.x client lands here, which is the whole of the upgrade + # story — everything else it does still works. + self._audit("upload_refused", "unsealed") + self._send({ + "type": "error", + "detail": "This upload did not open under the group key — the " + "client may be running an older version", + "code": "upload_not_sealed", + "upload_id": upload_id, + }) return + filename = payload.get("filename") or "" + data = payload.get("data") + # From the clear part of the message, so peer-controlled and unchecked + # by the AEAD. Everything below compares and adds to them. + try: + chunk_index = int(msg.get("chunk_index", 0)) + total_chunks = int(msg.get("total_chunks", 1)) + except (TypeError, ValueError): + self._send({"type": "error", "detail": "Invalid chunk index", + "code": "bad_chunk_index", "upload_id": upload_id}) + return + + def _refuse(detail: str, code: str = "") -> None: + """A refusal names the upload, never the file: the name is sealed.""" + out = {"type": "error", "detail": detail, "upload_id": upload_id} + if code: + out["code"] = code + self._send(out) + + # Types first, and before any state is created. What comes out of a + # sealed payload is authenticated, not validated: it is msgpack a + # member wrote, and `SAFE_UPLOAD_NAME.match(123)` raises where a + # refusal was meant. + if not isinstance(filename, str) or not filename: + _refuse("Missing filename or data", "upload_incomplete") + return + # Bytes, always: base64 was the shape of the old plaintext `data` field + # and there is no sealed message that can carry a string here. + if not isinstance(data, (bytes, bytearray)): + _refuse("Invalid chunk encoding", "bad_chunk_encoding") + return + chunk_bytes = bytes(data) + if not SAFE_UPLOAD_NAME.match(filename): - self._send({"type": "error", "detail": "Invalid filename", - "filename": filename}) + _refuse("Invalid filename", "invalid_filename") return roots: RootSet | None = ctx.get("roots") if not roots: - self._send({"type": "error", - "detail": "No directories configured for this group", - "filename": filename}) + _refuse("No directories configured for this group", "no_roots") return # The client names the root it is uploading into — it is browsing one, @@ -4448,47 +4501,32 @@ class WebRTCPeerSession: # An unknown name is refused rather than falling back to a writable # root, because "the file went somewhere else" is discovered weeks # later — the same reason the old single upload root was never guessed. - # A client that names nothing is an MNP 1.0 one, and there was exactly - # one destination in its world: the first writable root. # `dir` is the folder being browsed, as a virtual path # (`Media/Films/1999`); `root` is the older, coarser form and is what - # its first segment means on its own. - target_rel = str(msg.get("dir") or "").strip().strip("/") + # its first segment means on its own. Both are sealed now, so a refusal + # below can no longer quote them back. + target_rel = str(payload.get("dir") or "").strip().strip("/") target_root_name = (target_rel.split("/")[0] if target_rel - else str(msg.get("root") or "").strip()) + else str(payload.get("root") or "").strip()) upload_root = None if target_root_name: upload_root = roots.by_name(target_root_name) if upload_root is None: - self._send({"type": "error", - "detail": f"No directory named " - f"{target_root_name!r} in this group", - "code": "no_such_root", - "filename": filename}) + _refuse("No such directory in this group", "no_such_root") return else: writable = roots.writable_roots upload_root = writable[0] if writable else None if upload_root is None: - self._send({"type": "error", - "detail": "No writable directory in this group", - "code": "no_writable_root", - "filename": filename}) + _refuse("No writable directory in this group", "no_writable_root") return if not upload_root.writable: - self._send({"type": "error", - "detail": f"Directory '{upload_root.name}' is read-only", - "code": "root_read_only", - "filename": filename}) + _refuse("That directory is read-only", "root_read_only") self._audit("upload_refused", filename[:64]) return if not upload_root.available: - self._send({"type": "error", - "detail": f"Directory '{upload_root.name}' is " - f"currently unavailable", - "code": "root_unavailable", - "filename": filename}) + _refuse("That directory is currently unavailable", "root_unavailable") return # The folder the sender is looking at, and no subdirectory of the node's @@ -4512,23 +4550,17 @@ class WebRTCPeerSession: if target_rel: target_dir = roots.resolve(target_rel) if target_dir is None or not target_dir.is_dir(): - self._send({"type": "error", - "detail": "Not a directory in this group", - "code": "no_such_directory", - "filename": filename}) + _refuse("Not a directory in this group", "no_such_directory") return rel_dir = target_rel else: - # An MNP 1.0 client names nothing; the root itself is where its one - # destination now is. + # A client that names nothing: the first writable root is where its + # one destination is. target_dir = upload_root.path rel_dir = upload_root.name if not target_dir.is_dir(): - self._send({"type": "error", - "detail": f"Directory '{upload_root.name}' is " - f"currently unavailable", - "code": "root_unavailable", - "filename": filename}) + _refuse("That directory is currently unavailable", + "root_unavailable") return upload_key = f"{rel_dir}/{filename}" @@ -4544,33 +4576,24 @@ class WebRTCPeerSession: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. if final_path.exists(): - self._send({"type": "error", "detail": "File already exists", - "filename": filename}) + _refuse("File already exists", "already_exists") return state = {"next_index": 0, "bytes": 0, "stored_name": stored_name} self._uploads[upload_key] = state elif state is None: - self._send({"type": "error", "detail": "Upload not started", - "filename": filename}) + _refuse("Upload not started", "not_started") return # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends # blindly to whatever .part file is already on disk. if chunk_index != state["next_index"]: - self._send({"type": "error", "detail": "Unexpected chunk index", - "filename": filename}) + _refuse("Unexpected chunk index", "bad_chunk_index") return - if isinstance(data, str): - chunk_bytes = base64.b64decode(data) - else: - chunk_bytes = bytes(data) - if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES: self._uploads.pop(upload_key, None) tmp_path.unlink(missing_ok=True) - self._send({"type": "error", "detail": "Upload exceeds size limit", - "filename": filename}) + _refuse("Upload exceeds size limit", "too_large") return with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f: @@ -4578,16 +4601,16 @@ class WebRTCPeerSession: state["next_index"] = chunk_index + 1 state["bytes"] += len(chunk_bytes) - self._send({ - "type": MNP.FILE_UPLOAD_ACK, - "v": MNP_VERSION, - "chunk_index": chunk_index, - "filename": filename, + self._send(file_upload_ack_wire( + gek, self._group_id or "", + upload_id=upload_id, + chunk_index=chunk_index, + filename=filename, # What it is actually called on disk, which a chat attachment has to # reference and the uploader deserves to be told. - "stored_as": stored_name, - "dir": rel_dir, - }) + stored_as=stored_name, + dir=rel_dir, + )) if chunk_index + 1 >= total_chunks: self._uploads.pop(upload_key, None) @@ -5384,6 +5407,16 @@ class WebRTCPeerSession: self._audit("stream_video", entry.name) def _send(self, obj: dict) -> None: + # Stamp the reply with the id of the request being answered, so the + # caller never has to guess. Only for this session's own replies: a + # handler that also pushes to other peers (a chat broadcast, an index + # delta) reaches them through *their* _send, where the owner no longer + # matches and nothing is stamped — those messages answer no request. + # An explicit req_id already on the object wins, and an unsolicited + # push (no request in scope) carries none, exactly as before. + owner, req_id = _REPLY_TO.get() + if req_id is not None and owner is self and "req_id" not in obj: + obj = {**obj, "req_id": req_id} if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) else: diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py index 3dc9cd9..ba86c13 100644 --- a/packages/meshbay-node/tests/conftest.py +++ b/packages/meshbay-node/tests/conftest.py @@ -39,3 +39,34 @@ def one_root(path: Path, *, name: str = "", kind: str = "generic", """ return RootSet.build([{"path": str(path), "name": name, "kind": kind, "writable": writable}]) + + +def sealed_upload(session, *, filename: str, data: bytes, + chunk_index: int = 0, total_chunks: int = 1, + dir: str = "", root: str = "", + upload_id: str = "up-test") -> dict: + """ + A `file_upload` message as the shipping client builds one (MNP 2.0). + + Built through `file_upload_wire`, not by hand: a test that assembles the + wire shape itself is a second encoder, and a second encoder is how + `file_chunk` and `index_sync` forked between the transports (finding C6) + with nobody noticing. The key and the AAD are taken off the session, so + these agree with the handler by construction rather than by copying. + """ + from meshbay_common.protocol import file_upload_wire + + ctx = session._group_ctx() + return file_upload_wire( + ctx["gek"], session._group_id or "", + upload_id=upload_id, chunk_index=chunk_index, total_chunks=total_chunks, + filename=filename, data=data, dir=dir, root=root, + ) + + +def opened_ack(session, msg: dict) -> dict: + """The payload of a `file_upload_ack` the node sent, opened as a client would.""" + from meshbay_common.protocol import file_upload_ack_payload + + ctx = session._group_ctx() + return file_upload_ack_payload(ctx["gek"], session._group_id or "", msg) diff --git a/packages/meshbay-node/tests/test_reply_correlation.py b/packages/meshbay-node/tests/test_reply_correlation.py new file mode 100644 index 0000000..bfb336e --- /dev/null +++ b/packages/meshbay-node/tests/test_reply_correlation.py @@ -0,0 +1,165 @@ +""" +A reply names the request it answers. + +MNP carried no correlation id until 2026-09-07. A reply named its own type and +nothing else, so a client with more than one request outstanding had to work out +which one a message answered from the message itself — and for the replies that +name nothing, it could not. This module sends `{"type": "error"}` from 240 +places and two of them say what they are about; `_dispatch_message`'s catch-all +is one of the 238. Such a refusal reached no caller at all: the browser handed +it to whichever request happened to be waiting, and the request it belonged to +sat until its own 30s timeout. Live symptom (2026-09-06): the Chat composer is +disabled while a send is in flight, so a chat message whose refusal went astray +froze the tab for thirty seconds. + +`req_id` is the client's own pending-map key, put on the wire and stamped back +onto the reply by `_send`. What matters here, and what the browser cannot check +for itself: + + * a reply carries it, including the refusals that name nothing else; + * a *broadcast* does not — it answers no request, and stamping it would hand + another peer's client a reply to a request it never made; + * work handed to a background task still answers under the right id, which is + why this is a ContextVar and not an attribute on the session. +""" +import asyncio +import base64 + +import msgpack +import pytest +from meshbay_common.protocol import MNP +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +pytestmark = pytest.mark.asyncio + +# A device key's raw bytes. Only its length and its identity with the +# connection's own pin matter here; nothing verifies a signature over it. +_DEVICE = b"\x07" * 32 + + +class _Channel: + readyState = "open" + + def __init__(self): + self.sent = [] + + def send(self, framed): + # Skip the 4-byte length prefix _pack writes. + self.sent.append(msgpack.unpackb(framed[4:], raw=False)) + + +def _session(peer_id="p"): + s = WebRTCPeerSession.__new__(WebRTCPeerSession) + s._ctx = {} + s._peer_id = peer_id + s._user_id = None + s._group_id = "" + s._channel = _Channel() + s._tasks = set() + return s + + +async def test_a_refusal_that_names_nothing_else_names_the_request(): + """The reply at the root of the defect: no type of its own to match on.""" + s = _session() + # No handshake yet, so any other message is refused — a bare `error`, the + # same shape the catch-all sends and the same shape a browser could not + # route. + s._handle_message({"type": MNP.CHAT_MESSAGE, "req_id": 41}) + + (reply,) = s._channel.sent + assert reply["type"] == "error" + assert reply["req_id"] == 41, ( + "a refusal that names neither the request nor a type of its own is a " + "reply no caller can claim") + + +async def test_the_catch_all_refusal_names_the_request_too(): + """Every failure in the dispatch loop funnels into one generic reply.""" + s = _session() + s._user_id = "u" + + def _boom(msg): + raise RuntimeError("filesystem path that must not reach the peer") + s._do_chat_message = _boom + + s._handle_message({"type": MNP.CHAT_MESSAGE, "req_id": 7}) + + (reply,) = s._channel.sent + assert reply == {"type": "error", "detail": "Request failed", "req_id": 7}, ( + "the catch-all is where an unforeseen failure ends up, so it is exactly " + "the reply that must still be routable") + + +async def test_a_request_without_an_id_is_answered_without_one(): + """An older client sends none; nothing may be invented for it.""" + s = _session() + s._handle_message({"type": MNP.CHAT_MESSAGE}) + + (reply,) = s._channel.sent + assert "req_id" not in reply + + +async def test_a_broadcast_to_another_peer_is_not_stamped(): + """The reply goes to the asker; the broadcast goes to everyone else. + + They travel out of the same handler, and only the first answers anything. + Stamping the second would hand another browser a reply keyed to a pending + request of its own that it never sent — the very confusion this fixes. + """ + asker, other = _session("asker"), _session("other") + asker._user_id, other._user_id = "a", "b" + registry = {"ka": asker, "kb": other} + asker._peer_registry = lambda: registry + asker._user_names = lambda: {} + asker._audit = lambda *a, **k: None + asker._group_ctx = lambda: {} + asker._spawn = lambda coro: coro.close() + + asker._registry_key, other._registry_key = "ka", "kb" + asker._pinned_pk = base64.b64encode(_DEVICE).decode() + asker._device_confirmed = True + + # A sealed message, because MNP 2.0 has no plaintext chat and the node + # refuses one — the bytes need not decrypt, since nothing here opens them. + # What this test is about is unchanged: which of the two messages leaving + # this handler carries the id. + asker._handle_message({ + "type": MNP.CHAT_MESSAGE, "req_id": 3, + "format": 1, "epoch": 1, "device": _DEVICE, "ct": b"ciphertext", + "nonce": b"\x02" * 12, "sig": b"\x03" * 64, + }) + + (ack,) = asker._channel.sent + assert ack["type"] == "ack" and ack["req_id"] == 3 + (broadcast,) = other._channel.sent + assert broadcast["type"] == MNP.CHAT_MESSAGE + assert "req_id" not in broadcast, ( + "a broadcast answers no request and must not look like a reply") + + +async def test_work_handed_to_a_task_still_answers_under_the_right_id(): + """Most handlers `_spawn` their real work, and the reply leaves long after + the dispatch call that started it has returned. + + This is the reason the id lives in a ContextVar: asyncio copies the current + context into a task, so the answer keeps the id even though nothing passed + it along. An attribute on the session would have been overwritten by the + next message to arrive in the meantime. + """ + s = _session() + s._user_id = "u" + + async def _late(reply): + await asyncio.sleep(0.01) + s._send({"type": "roster_read_resp", "detail": reply}) + s._do_chat_message = lambda msg: s._spawn(_late(msg["payload"])) + + s._handle_message({"type": MNP.CHAT_MESSAGE, "payload": "first", "req_id": 11}) + # A second request arrives while the first one's task is still asleep. + s._handle_message({"type": MNP.CHAT_MESSAGE, "payload": "second", "req_id": 12}) + await asyncio.gather(*list(s._tasks)) + + by_id = {m["req_id"]: m["detail"] for m in s._channel.sent} + assert by_id == {11: "first", 12: "second"}, ( + "a late reply answered under whichever request arrived most recently") diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py index 7eb75fd..8345880 100644 --- a/packages/meshbay-node/tests/test_root_writable_policy.py +++ b/packages/meshbay-node/tests/test_root_writable_policy.py @@ -21,13 +21,14 @@ anything. A deprecated instruction that still works is not deprecated, and this one would reopen uploads group-wide. """ -import base64 from pathlib import Path import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from conftest import sealed_upload from meshbay_common.adminop import OP_ROOT_UPDATE, OP_ROOT_EJECT, OP_ROOT_PLUG +from meshbay_common.crypto import generate_gek from meshbay_common.protocol import MNP from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roots import RootSet @@ -47,6 +48,8 @@ def _session(tmp_path: Path, user_id: str, *, "index": index, "sk_node": index.sk_node, "node_user_id": operator, + # Uploads are sealed under the group key since MNP 2.0. + "gek": generate_gek(), } session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = ctx @@ -61,11 +64,8 @@ def _session(tmp_path: Path, user_id: str, *, def _upload(session, filename="clip.mp4", body=b"bytes"): - session._do_file_upload({ - "filename": filename, "dir": "shared", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(body).decode(), - }) + session._do_file_upload(sealed_upload( + session, filename=filename, data=body, dir="shared")) def _uploads_dir(session) -> Path: diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index 1a318f7..988aa46 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -16,10 +16,11 @@ from pathlib import Path import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roots import RootSet -from conftest import one_root +from conftest import one_root, opened_ack, sealed_upload from meshbay_node.transport.webrtc_server import WebRTCPeerSession @@ -156,7 +157,10 @@ def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession: shared_root = tmp_path / "shared" shared_root.mkdir(exist_ok=True) index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) - ctx = {"roots": one_root(shared_root), "index": index, "sk_node": index.sk_node} + # A group key, because uploads are sealed under it since MNP 2.0 — the + # handler opens the payload before it has a filename to refuse. + ctx = {"roots": one_root(shared_root), "index": index, + "sk_node": index.sk_node, "gek": generate_gek()} session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = ctx @@ -188,12 +192,8 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): original.write_bytes(b"operator's original content") attacker = _session(tmp_path, "attacker-user") - attacker._do_file_upload({ - "filename": "important.mp4", - "chunk_index": 0, - "total_chunks": 1, - "data": base64.b64encode(b"attacker content").decode(), - }) + attacker._do_file_upload(sealed_upload( + attacker, filename="important.mp4", data=b"attacker content")) assert original.read_bytes() == b"operator's original content", ( "an upload replaced an existing file (C5a)") @@ -203,12 +203,15 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path): """C5a: even the original uploader does not get to overwrite.""" session = _session(tmp_path, "user-1") - payload = {"filename": "movie.mp4", "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"first").decode()} - session._do_file_upload(dict(payload)) - session.sent.clear() + def _send_it(): + # Sealed afresh each time: a nonce is drawn per message, so re-sending + # the same dict would be a replay rather than a second upload. + session._do_file_upload(sealed_upload( + session, filename="movie.mp4", data=b"first")) - session._do_file_upload(dict(payload)) + _send_it() + session.sent.clear() + _send_it() uploads = _uploads_dir(session) assert (uploads / "movie.mp4").read_bytes() == b"first", ( "the first upload was replaced") @@ -252,11 +255,8 @@ def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path): for bad in ("../../etc", "/etc", "shared/../..", "shared/../../etc", "nope", "shared/missing"): session.sent.clear() - session._do_file_upload({ - "filename": "note.txt", "dir": bad, - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir=bad)) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal, f"{bad!r} was accepted" assert refusal[0].get("code") in ("no_such_root", "no_such_directory"), bad @@ -274,11 +274,8 @@ def test_an_upload_lands_in_the_folder_it_names(tmp_path): root = session._ctx["roots"].roots[0] (root.path / "Albums").mkdir() - session._do_file_upload({ - "filename": "note.txt", "dir": f"{root.name}/Albums", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir=f"{root.name}/Albums")) assert (root.path / "Albums" / "note.txt").read_bytes() == b"x" assert not (root.path / "Albums" / "uploads").exists(), ( @@ -304,11 +301,8 @@ def test_an_upload_goes_to_the_root_it_names(tmp_path): {"path": str(incoming), "writable": True}, ]) - session._do_file_upload({ - "filename": "note.txt", "dir": "Incoming", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir="Incoming")) assert (incoming / "note.txt").read_bytes() == b"x" assert not (media / "note.txt").exists(), "it went to the first root instead" @@ -327,11 +321,8 @@ def test_a_read_only_root_refuses_an_upload(tmp_path): session._ctx["roots"] = RootSet.build([{"path": str(published)}]) session._is_node_admin = lambda: True - session._do_file_upload({ - "filename": "note.txt", "dir": "Published", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir="Published")) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "root_read_only" @@ -349,11 +340,8 @@ def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path): session = _session(tmp_path, "user-1") session._ctx["roots"] = RootSet.build([{"path": str(published)}]) - session._do_file_upload({ - "filename": "note.txt", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x")) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "no_writable_root" @@ -375,11 +363,8 @@ def test_an_ejected_root_refuses_an_upload(tmp_path): roots.roots[0].available = False session._ctx["roots"] = roots - session._do_file_upload({ - "filename": "note.txt", "dir": "USB", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir="USB")) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "root_unavailable" @@ -392,22 +377,22 @@ def test_two_members_can_send_the_same_filename(tmp_path): IMG_1234.jpg. The second gets a free name; neither replaces the other. """ first = _session(tmp_path, "user-1") - first._do_file_upload({ - "filename": "IMG_1234.jpg", "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"first").decode(), - }) + first._do_file_upload(sealed_upload( + first, filename="IMG_1234.jpg", data=b"first")) second = _session(tmp_path, "user-2") - second._do_file_upload({ - "filename": "IMG_1234.jpg", "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"second").decode(), - }) + # Same group, so the same key: `_session` builds one per call, and two + # members of one group do not have two. + second._ctx["gek"] = first._ctx["gek"] + second._do_file_upload(sealed_upload( + second, filename="IMG_1234.jpg", data=b"second")) uploads = _uploads_dir(first) assert (uploads / "IMG_1234.jpg").read_bytes() == b"first" assert (uploads / "IMG_1234 (2).jpg").read_bytes() == b"second" ack = [m for m in second.sent if m.get("type") == "file_upload_ack"][-1] - assert ack["stored_as"] == "IMG_1234 (2).jpg", ( + assert "stored_as" not in ack, "the name the node chose must be sealed" + assert opened_ack(second, ack)["stored_as"] == "IMG_1234 (2).jpg", ( "the sender must be told the name that was used, or a chat attachment " "points at someone else's file") @@ -743,20 +728,48 @@ def test_pre_handshake_message_budget_is_small(): list(buf.messages()) -def test_stream_segment_is_not_synchronous(): +def test_no_transport_ships_media_outside_the_aead(): """ - H6: _do_stream_segment ran subprocess.run(timeout=30) inside the event loop, - stalling every peer on the node for up to thirty seconds per request. + `stream_seg` served an MPEG-TS segment as base64 with no encryption at all + — the one content-plane message that never went through a GEK-derived key, + on both transports, answering any authenticated member. Its browser caller + was defined and never invoked. Removed in MNP 2.0 rather than repaired: + `stream_data` already does the job under `chunk_ciphertext`. - Asserts the property (the worker is a coroutine, ffmpeg is spawned through - asyncio) rather than grepping for "subprocess.run" — which also matches the - comment that documents the old behaviour. + Asserted as the property, not as "the function is gone": what matters is + that no transport has a field carrying media bytes past the AEAD. The old + H6 test lived here — it pinned `_do_stream_segment_async` to a coroutine so + ffmpeg could not block the event loop — and the handler outliving that + concern is exactly what this replaces. """ - import ast - import inspect - from meshbay_node.transport.webrtc_server import WebRTCPeerSession + import re + + from meshbay_common.protocol import MNP + + assert not hasattr(MNP, "STREAM_SEGMENT"), ( + "the constant outliving the handlers is how a deleted endpoint keeps " + "looking like part of the wire contract") + + root = Path(__file__).parent.parent / "src" / "meshbay_node" / "transport" + for name in ("webrtc_server.py", "quic_server.py", "quic_client.py"): + source = (root / name).read_text(encoding="utf-8") + # Word boundaries: `_stream_segments` and `STREAM_SEGMENT_SIZE` belong + # to the live `stream_data` path, which is encrypted and stays. + assert not re.search(r"\bstream_seg\b", source), ( + f"{name} still speaks stream_seg") + assert not re.search(r"\bSTREAM_SEGMENT\b", source), ( + f"{name} still names the removed type") + assert "data_b64" not in source, ( + f"{name} carries a base64 media field — media leaves this node " + "encrypted or not at all") - assert inspect.iscoroutinefunction(WebRTCPeerSession._do_stream_segment_async) + +def test_ffmpeg_never_blocks_the_event_loop(): + """ + H6, the half that survives `stream_seg`: the live streaming path still + spawns ffmpeg, and a synchronous spawn stalls every peer on the node. + """ + import ast source = (Path(__file__).parent.parent / "src" / "meshbay_node" / "transport" / "webrtc_server.py").read_text(encoding="utf-8") diff --git a/packages/meshbay-node/tests/test_task_lifetime.py b/packages/meshbay-node/tests/test_task_lifetime.py index 3a5b8a5..9dffedb 100644 --- a/packages/meshbay-node/tests/test_task_lifetime.py +++ b/packages/meshbay-node/tests/test_task_lifetime.py @@ -245,8 +245,13 @@ def test_chunks_wait_for_room_on_the_channel(session): work through the rest — which is what "stuck at 1 MB" looks like, one chunk being exactly one megabyte. """ - fn = session[session.index("async def _do_file_request"):] - fn = fn[:fn.index("\n def _do_stream_segment")] + start = session.index("async def _do_file_request") + # Up to whatever the next member is. This used to end at + # "\n def _do_stream_segment" — a neighbour removed in MNP 2.0 — and an + # `index()` on a name that no longer exists fails the test for a reason + # that has nothing to do with what it is about. + nxt = re.search(r"\n (?:@|(?:async )?def )", session[start:]) + fn = session[start:start + nxt.start()] if nxt else session[start:] assert "DOWNLOAD_BUFFER_HIGH" in fn, "the send buffer has to be watched" assert "await asyncio.sleep" in fn, "waiting for room is the point" assert 'readyState != "open"' in fn, ( diff --git a/packages/meshbay-node/tests/test_upload_sealed.py b/packages/meshbay-node/tests/test_upload_sealed.py new file mode 100644 index 0000000..7c1be96 --- /dev/null +++ b/packages/meshbay-node/tests/test_upload_sealed.py @@ -0,0 +1,285 @@ +""" +The write path, sealed under the group key (MNP 2.0). + +Downloads have been encrypted under a GEK-derived key since the beginning: +`file_chunk` and `stream_data` both go through `chunk_ciphertext`. Uploads did +not. `file_upload` carried the filename and the raw bytes in plain msgpack and +`file_upload_ack` carried the name the node stored them under, so the same file +was ciphertext leaving a node and plaintext arriving at one — an asymmetry with +no threat model behind it. + +What sealing buys is what `groupbox.py` says and no more: nothing against a +network observer (DTLS covers that), nothing against the hub (never on this +channel), nothing against a member (they hold the GEK). It buys defence in +depth against our own next handshake bug, of a class already shipped twice — +C1, the unauthenticated node HTTP API, and C6, the transport that took a bare +JWT. Both were "a peer that had not finished the handshake was served data". +Sealed, the equivalent bug on this path leaks ciphertext instead of the +operator's filenames. +""" + +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_common.protocol import MNP, file_upload_wire +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root, opened_ack, sealed_upload + +GROUP = "g" * 32 + + +def _session(tmp_path: Path, *, gek: bytes | None = None) -> WebRTCPeerSession: + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = {"roots": one_root(shared_root), "index": index, + "sk_node": index.sk_node, "gek": gek or generate_gek()} + session._group_id = GROUP + session._user_id = "user-1" + session._pk_user = "" + session._uploads = {} + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _root(session): + return session._ctx["roots"].roots[0] + + +def _errors(session): + return [m for m in session.sent if m.get("type") == "error"] + + +def _wrote_anything(tmp_path) -> bool: + return any(p.is_file() for p in tmp_path.rglob("*")) + + +# ── The message itself ─────────────────────────────────────────────────────── + +def test_the_wire_message_carries_no_filename_and_no_plaintext(tmp_path): + """ + The point of the exercise. `upload_id` and `chunk_index` are outside the + seal because the node routes and orders on them before it can decrypt; + everything that names or is the operator's content is inside it. + """ + session = _session(tmp_path) + msg = sealed_upload(session, filename="holiday.jpg", data=b"JPEGDATA", + dir=f"{_root(session).name}") + + assert set(msg) == {"type", "v", "upload_id", "chunk_index", + "total_chunks", "nonce", "ct"} + blob = repr(msg).encode() + msg["ct"] + assert b"holiday.jpg" not in blob, "the filename is on the wire in clear" + assert b"JPEGDATA" not in blob, "the file content is on the wire in clear" + + +def test_the_ack_carries_no_stored_name(tmp_path): + """ + `stored_as` is the name the node settled on — it finds a free one rather + than replacing anything — and naming it in clear would hand back exactly + what the request took the trouble to hide. + """ + session = _session(tmp_path) + session._do_file_upload(sealed_upload( + session, filename="holiday.jpg", data=b"x", dir=_root(session).name)) + + ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1] + assert set(ack) == {"type", "v", "upload_id", "chunk_index", "nonce", "ct"} + assert b"holiday.jpg" not in repr(ack).encode() + ack["ct"] + assert opened_ack(session, ack) == { + "filename": "holiday.jpg", "stored_as": "holiday.jpg", + "dir": _root(session).name} + + +def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path): + """ + `filename` used to be the correlation key on both sides. It cannot be one + any more, and `upload_id` replaces it — a client-chosen label, opaque to + the node, never an authorization input. Without it a client running several + uploads could only match replies by arrival order, which is how a refusal + for one file used to fail every upload in flight. + """ + session = _session(tmp_path) + session._do_file_upload(sealed_upload( + session, filename="a.txt", data=b"x", dir=_root(session).name, + upload_id="upload-A")) + session._do_file_upload(sealed_upload( + session, filename="../evil", data=b"x", dir=_root(session).name, + upload_id="upload-B")) + + ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1] + assert ack["upload_id"] == "upload-A" + assert _errors(session)[0]["upload_id"] == "upload-B" + + +# ── What is refused ────────────────────────────────────────────────────────── + +def test_a_plaintext_upload_is_refused(tmp_path): + """ + The MNP 1.x shape, which is what an un-updated client sends. Refused with a + code and a message saying which side is old — never accepted "just this + once", because a path that still takes plaintext is not a sealed path. + """ + session = _session(tmp_path) + session._do_file_upload({ + "filename": "note.txt", "dir": _root(session).name, + "chunk_index": 0, "total_chunks": 1, "data": b"x", + }) + + assert _errors(session)[0]["code"] == "upload_not_sealed" + assert not _wrote_anything(tmp_path) + + +def test_a_tampered_chunk_is_refused(tmp_path): + """ + AES-GCM's tag, asserted where it matters: a flipped bit in the ciphertext + must stop the upload, not produce a corrupt file with a plausible name. + """ + session = _session(tmp_path) + msg = sealed_upload(session, filename="note.txt", data=b"x" * 64, + dir=_root(session).name) + msg["ct"] = bytes([msg["ct"][0] ^ 0x01]) + msg["ct"][1:] + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "upload_not_sealed" + assert not _wrote_anything(tmp_path) + + +def test_an_upload_sealed_for_another_group_is_refused(tmp_path): + """ + The group is the AAD, so a node hosting two groups cannot have a chunk + moved between them — and a member of one cannot write into the other by + reaching a session that is on it (finding H1's shape, on the write path). + """ + session = _session(tmp_path) + msg = file_upload_wire( + session._ctx["gek"], "some-other-group", + upload_id="u1", chunk_index=0, total_chunks=1, + filename="note.txt", data=b"x", dir=_root(session).name) + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "upload_not_sealed" + assert not _wrote_anything(tmp_path) + + +def test_an_upload_under_another_key_is_refused(tmp_path): + """A peer past the handshake with the wrong GEK still writes nothing.""" + session = _session(tmp_path) + msg = file_upload_wire( + generate_gek(), GROUP, + upload_id="u1", chunk_index=0, total_chunks=1, + filename="note.txt", data=b"x", dir=_root(session).name) + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "upload_not_sealed" + assert not _wrote_anything(tmp_path) + + +def test_an_ack_replayed_as_a_request_does_not_open(tmp_path): + """ + The message type is in the AAD, so the two halves of an upload cannot be + confused for each other. Cheap, and it closes a class that is tedious to + reason about after the fact. + """ + from cryptography.exceptions import InvalidTag + from meshbay_common.protocol import file_upload_ack_wire, file_upload_payload + + session = _session(tmp_path) + ack = file_upload_ack_wire( + session._ctx["gek"], GROUP, upload_id="u1", chunk_index=0, + filename="note.txt", stored_as="note.txt", dir="shared") + with pytest.raises(InvalidTag): + file_upload_payload(session._ctx["gek"], GROUP, ack) + + +def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path): + """ + A node whose group has no GEK yet cannot open anything. It must say so, not + read the message as though it were the old plaintext shape. + """ + session = _session(tmp_path) + msg = sealed_upload(session, filename="note.txt", data=b"x", + dir=_root(session).name) + session._ctx["gek"] = b"" + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "no_group_key" + assert not _wrote_anything(tmp_path) + + +# ── What must still work ───────────────────────────────────────────────────── + +def test_a_multi_chunk_upload_reassembles(tmp_path): + """ + Every chunk is sealed under its own nonce, and the node appends in order. + Nothing about the seal may change what lands on disk. + """ + session = _session(tmp_path) + body = bytes(range(256)) * 40 + parts = [body[i:i + 1024] for i in range(0, len(body), 1024)] + for i, part in enumerate(parts): + session._do_file_upload(sealed_upload( + session, filename="blob.bin", data=part, + chunk_index=i, total_chunks=len(parts), dir=_root(session).name)) + + assert (_root(session).path / "blob.bin").read_bytes() == body + assert not list(_root(session).path.glob("*.part")), "a temp file was left" + acks = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK] + assert [m["chunk_index"] for m in acks] == list(range(len(parts))) + + +def test_two_identical_chunks_do_not_reuse_a_nonce(tmp_path): + """ + A file of repeated bytes is ordinary, so the nonce must come from the RNG + and never from the payload. Cheap to assert and expensive to discover. + """ + session = _session(tmp_path) + a = sealed_upload(session, filename="f", data=b"same", chunk_index=0) + b = sealed_upload(session, filename="f", data=b"same", chunk_index=0) + assert a["nonce"] != b["nonce"] + assert a["ct"] != b["ct"] + + +def test_a_sealed_payload_is_authenticated_not_validated(tmp_path): + """ + Opening a payload proves a member wrote it, not that they wrote something + sensible. A member can seal anything, so the fields still need their types + checked — `SAFE_UPLOAD_NAME.match(123)` raises where a refusal was meant, + and the dispatcher's catch-all would turn that into "Request failed". + """ + from meshbay_common.groupbox import PURPOSE_UPLOAD, seal + + session = _session(tmp_path) + for payload in ({"filename": 123, "data": b"x"}, + {"filename": "note.txt", "data": "not bytes"}, + {"filename": "note.txt"}): + session.sent.clear() + session._do_file_upload({ + "type": MNP.FILE_UPLOAD, "v": "2.0", "upload_id": "u1", + "chunk_index": 0, "total_chunks": 1, + **seal(session._ctx["gek"], PURPOSE_UPLOAD, MNP.FILE_UPLOAD, + GROUP, payload), + }) + errs = _errors(session) + assert errs, f"{payload!r} was accepted" + assert errs[0]["code"] in ("upload_incomplete", "bad_chunk_encoding") + assert not _wrote_anything(tmp_path) + + +def test_a_peer_controlled_chunk_index_cannot_crash_the_handler(tmp_path): + """`chunk_index` is outside the seal by necessity, so it is unchecked input.""" + session = _session(tmp_path) + msg = sealed_upload(session, filename="note.txt", data=b"x", + dir=_root(session).name) + msg["chunk_index"] = "zero" + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "bad_chunk_index" + assert not _wrote_anything(tmp_path) diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index ea13d96..c1a5287 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -783,36 +783,6 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir): @pytest.mark.asyncio -async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir): - """WebRTC DataChannel: stream_segment for non-existent file returns error.""" - hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) - await indexer.initial_scan() - - transport = WebRTCTransport( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - roots=one_root(shared_dir), index=indexer.index, - stun_servers=[], - ) - - browser_pc, channel, received = await _setup_peer( - transport, sk_hub, gek, "peer-stream") - - channel.send(_pack({ - "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION, - "file_id": "nonexistent-file-id", - "segment_index": 0, "segment_duration": 4, - })) - - msg = await asyncio.wait_for(received.get(), timeout=5.0) - assert msg["type"] == "error" - assert "not found" in msg["detail"].lower() - - await browser_pc.close() - await transport.close_all() - - -@pytest.mark.asyncio async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: wrong GEK proof is rejected — hub admin can't fake membership.""" hub_pk_pem = _hub_pk_pem(sk_hub) |