summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-07 18:03:52 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-07 18:03:52 +0200
commite1383e1d545b994f4ad61694f868339defb0bdef (patch)
tree67a3b1933f2a9c5caf107d01d9ff91c44a975df6
parent36cebf25d0e0f24cf63be4380ccb5d03da726a74 (diff)
parent8980a8e42d94ab7c0bc9739283d39f938f8402b0 (diff)
downloadmeshbay-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
-rw-r--r--CLAUDE.md27
-rw-r--r--docs/USERGUIDE.md2
-rw-r--r--docs/devel-phases-next.md2
-rw-r--r--packages/meshbay-common/src/meshbay_common/__init__.py53
-rw-r--r--packages/meshbay-common/src/meshbay_common/groupbox.py17
-rw-r--r--packages/meshbay-common/src/meshbay_common/protocol.py151
-rw-r--r--packages/meshbay-common/tests/test_js_python_parity.py13
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/static/crypto.js11
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/static/transport.js253
-rw-r--r--packages/meshbay-hub/tests/harness/chat_send_probe.py242
-rw-r--r--packages/meshbay-hub/tests/harness/upload_seal_probe.mjs102
-rw-r--r--packages/meshbay-hub/tests/test_chat_send.py158
-rw-r--r--packages/meshbay-hub/tests/test_index_seal_client.py30
-rw-r--r--packages/meshbay-hub/tests/test_transport_contracts.py50
-rw-r--r--packages/meshbay-hub/tests/test_upload_seal_client.py169
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/quic_client.py19
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/quic_server.py76
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py309
-rw-r--r--packages/meshbay-node/tests/conftest.py31
-rw-r--r--packages/meshbay-node/tests/test_reply_correlation.py165
-rw-r--r--packages/meshbay-node/tests/test_root_writable_policy.py12
-rw-r--r--packages/meshbay-node/tests/test_security_regressions.py137
-rw-r--r--packages/meshbay-node/tests/test_task_lifetime.py9
-rw-r--r--packages/meshbay-node/tests/test_upload_sealed.py285
-rw-r--r--packages/meshbay-node/tests/test_webrtc_transport.py30
25 files changed, 1747 insertions, 606 deletions
diff --git a/CLAUDE.md b/CLAUDE.md
index 808ab47..ec5b1b5 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -471,11 +471,28 @@ anything that assumes one key per person.
next visit to the tab, since the node had stored it and answered. Every line
of `chat-app.js` is correct and every routed message in `transport.js` is
routed correctly; the defect is in the seam, which is why
- `tests/harness/chat_send_probe.py` drives the two together. `ack` is now
- matched by request type (`chat_msg`, or the keypair-bundle store/delete that
- name themselves in `detail`). Anything left to the "oldest pending" guess is
- a latent version of this bug: a request type deserves a key, and a reply
- deserves something to key it by
+ `tests/harness/chat_send_probe.py` drives the two together. `ack` was matched
+ by request type (`chat_msg`, or the keypair-bundle store/delete that name
+ themselves in `detail`), and that closed the instance — **but it left the
+ class open, and it came back on 2026-09-06 through the other door.** A
+ refusal has no type of its own to key on: `_dispatch_message`'s catch-all
+ answers every unforeseen failure with `{"type": "error", "detail": "Request
+ failed"}`, and 238 of `webrtc_server.py`'s 240 error sends name nothing
+ either. So a chat send the node refused was routed by luck all over again —
+ same frozen composer, same 30 s, and rare enough (it needs an older request
+ still waiting, which a `music_meta_req` behind a failing third-party lookup
+ supplies for over a hundred seconds) to look like once every couple of days.
+ The keys were never the fix, only a workaround for a protocol that carried no
+ correlation id at all: `_seqId` existed, indexed `_pending`, and was never put
+ on the wire. It is now (`req_id`, see `protocol.py`) — the node stamps it on
+ the reply from `_send`, via a ContextVar so a handler's spawned work still
+ answers under the right id, and never on a broadcast, which answers nothing.
+ With that, the arrival-order fallback is gone for any node that stamps.
+ The lesson is not "key the replies": it is that **matching by arrival order
+ is a guess that fails silently and asymmetrically** — the victim is never
+ the request that was answered wrongly, it is the unrelated one that now
+ waits for a reply already delivered elsewhere. A reply needs an identifier
+ the protocol guarantees, not a field it happens to have
- **A refusal that never rejects.** Denying Chromium's `fullscreen` permission
does not make `requestFullscreen()` throw — the promise never settles. The
diff --git a/docs/USERGUIDE.md b/docs/USERGUIDE.md
index de5ce3f..32f31dd 100644
--- a/docs/USERGUIDE.md
+++ b/docs/USERGUIDE.md
@@ -1127,7 +1127,7 @@ handshake that establishes what they are allowed to see:
| `join_request` | `join_result` | Pairing code, or recognition of a pinned identity; carries the wrapped GEK on success |
| `index_sync` | `index_sync` / `index_delta` | The group index |
| `file_req` | `file_chunk` | One encrypted 1 MB chunk |
-| `file_upload` | `file_upload_ack` | Upload into `uploads/`; the ack names the file as stored |
+| `file_upload` | `file_upload_ack` | One chunk into the folder you are browsing; both halves sealed under the group key, so the filename and the bytes never appear on the wire in clear. The ack names the file as stored |
| `dir_create` | `dir_create_ack` | Create a directory |
| `file_delete` | `file_delete_ack` | Delete a file you uploaded |
| `stream_req` | `stream_init`, `stream_data`, `stream_end` | MSE video |
diff --git a/docs/devel-phases-next.md b/docs/devel-phases-next.md
index cbee283..a155652 100644
--- a/docs/devel-phases-next.md
+++ b/docs/devel-phases-next.md
@@ -503,7 +503,7 @@ Browser Node
- No seeking beyond buffered range (user must wait for data to arrive)
- No adaptive bitrate (single quality stream)
-- Requires ffmpeg/ffprobe on the node (already a dependency for the old STREAM_SEGMENT handler)
+- Requires ffmpeg/ffprobe on the node (already a dependency for the live streaming path)
---
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)