diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 97 |
1 files changed, 11 insertions, 86 deletions
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 1f069e7..370a8b4 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -93,8 +93,8 @@ from meshbay_common.join import ( ROLE_OPERATOR, join_transcript, ) -from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes -from meshbay_common.protocol import MNP, index_entry_wire +from meshbay_common.protocol import MNP, chunk_ciphertext, file_chunk_wire +from meshbay_node.transport.wire import index_sync_message from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import linkpreview, ops @@ -2608,50 +2608,7 @@ class WebRTCPeerSession: def _do_index_sync(self) -> None: ctx = self._group_ctx() - idx = ctx["index"] - entries = [index_entry_wire(e) for e in idx.entries] - self._send({ - "type": MNP.INDEX_SYNC, - "v": MNP_VERSION, - "group_id": idx.group_id, - "version": idx.version, - "entries": entries, - # Directories are not index entries, so the client used to infer them - # from file paths — which means a folder someone just created, or one - # they emptied, simply did not exist as far as the UI was concerned. - "dirs": self._list_dirs(ctx.get("roots")), - # Which top-level folders are roots, and whether each is readable. - # A frozen root's files stay listed, so without this a member cannot - # tell "the drive is unplugged" from "it is all still there". - "roots": ctx["roots"].describe() if ctx.get("roots") else [], - }) - - @staticmethod - def _list_dirs(roots: RootSet | None) -> list[str]: - """ - Every directory in the group, as members address them, sorted. - - Each root appears as a directory in its own right, so a root holding no - files yet is still somewhere a member can navigate to and upload into. - An unavailable root is listed too — its content is frozen, not gone, and - hiding it would look exactly like deletion. - """ - if not roots: - return [] - out: list[str] = [] - for root in roots: - out.append(root.name) - if not root.available: - continue - try: - for path in sorted(root.path.rglob("*")): - if path.is_dir() and not path.name.startswith("."): - rel = path.relative_to(root.path) - if not any(part.startswith(".") for part in rel.parts): - out.append(f"{root.name}/{rel.as_posix()}") - except OSError: - continue - return sorted(out)[:2000] + self._send(index_sync_message(ctx["index"], ctx.get("roots"))) async def _try_serve_thumbnail( self, thumb_hash: str, chunk_index: int, gek: bytes | None, @@ -2677,10 +2634,8 @@ class WebRTCPeerSession: if start > len(blob) or (start == len(blob) and chunk_index != 0): return None piece = blob[start:start + CHUNK_SIZE] - return _encrypt_chunk_bytes( - self._ctx["sk_node"], gek, piece, chunk_index, - bytes.fromhex(thumb_hash), thumb_hash, - ) + return file_chunk_wire( + gek, piece, chunk_index, bytes.fromhex(thumb_hash), thumb_hash) async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() @@ -2707,13 +2662,7 @@ class WebRTCPeerSession: getattr(self._channel, "bufferedAmount", "?")) file_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( - self._ctx["sk_node"], - ctx["gek"], - file_path, - chunk_index, - file_hash, - entry.id, - ) + ctx["gek"], file_path, chunk_index, file_hash, entry.id) # Backpressure. Without it the node hands the whole window to the # channel at once and the reader sees the first chunk, then nothing for # as long as the link takes to drain the rest. @@ -4435,8 +4384,9 @@ class WebRTCPeerSession: data = await proc.stdout.read(STREAM_SEGMENT_SIZE) if not data: break - ckey = chunk_key_aes(gek, file_hash, index) - nonce, ct = encrypt_chunk_aes(ckey, data) + # Same derivation as a file chunk, indexed by segment: one + # implementation, in `meshbay_common.protocol`. + nonce, ct = chunk_ciphertext(gek, data, index, file_hash) self._send({ "type": MNP.STREAM_DATA, "v": MNP_VERSION, @@ -4546,43 +4496,18 @@ class WebRTCPeerSession: await self._pc.close() -def _encrypt_chunk_bytes( - sk_node: Ed25519PrivateKey, - gek: bytes, - plaintext: bytes, - chunk_index: int, - file_hash: bytes, - file_id: str = "", -) -> dict: - ckey = chunk_key_aes(gek, file_hash, chunk_index) - nonce, ct = encrypt_chunk_aes(ckey, plaintext) - - return { - "type": MNP.FILE_CHUNK, - "v": MNP_VERSION, - # Named so a client running several downloads at once can tell whose - # reply this is. It used to carry only the index, which made matching a - # reply to its request a question of arrival order. - "file_id": file_id, - "chunk_index": chunk_index, - "plaintext_size": len(plaintext), - "nonce": nonce, - "ct": ct, - } - - def _read_and_encrypt( - sk_node: Ed25519PrivateKey, gek: bytes, file_path: Path, chunk_index: int, file_hash: bytes, file_id: str = "", ) -> dict: + """Read one chunk off disk and encrypt it. Blocking; the caller keeps it short.""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) - return _encrypt_chunk_bytes(sk_node, gek, plaintext, chunk_index, file_hash, file_id) + return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) async def _transcode_audio_to_aac(file_path: Path) -> bytes: |