diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
5 files changed, 111 insertions, 35 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/http_server.py b/packages/meshbay-node/src/meshbay_node/transport/http_server.py index 7554086..151c2e8 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/http_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/http_server.py @@ -37,7 +37,8 @@ from fastapi.responses import FileResponse, JSONResponse, StreamingResponse from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION -from meshbay_common.crypto import chunk_key as derive_chunk_key, encrypt_chunk, sign_chunk, pk_to_b64 +from meshbay_common.crypto import sign_chunk, pk_to_b64 +from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, encrypt_chunk_aes as encrypt_chunk from meshbay_node import __version__ from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.group_index import GroupIndex @@ -189,7 +190,7 @@ def create_http_app( if not plaintext: raise HTTPException(status_code=416, detail="Chunk out of range") - file_hash = blake3.blake3(file_path.read_bytes()).digest() + file_hash = bytes.fromhex(entry.id) pt_hash = blake3.blake3(plaintext).digest() if gek: 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 83b729e..288465f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py @@ -23,11 +23,8 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from meshbay_common import MNP_VERSION -from meshbay_common.crypto import ( - chunk_key as derive_chunk_key, - decrypt_chunk, - verify_chunk_signature, -) +from meshbay_common.crypto import verify_chunk_signature +from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk from meshbay_common.protocol import MNP log = logging.getLogger(__name__) 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 43c1026..f439e62 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -35,11 +35,10 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION from meshbay_common.crypto import ( - chunk_key as derive_chunk_key, - encrypt_chunk, sign_chunk, pk_to_b64, ) +from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, encrypt_chunk_aes as encrypt_chunk from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex from meshbay_node.transport.tls_cert import server_ssl_context @@ -217,11 +216,13 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return + file_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( self._ctx["sk_node"], ctx["gek"], file_path, chunk_index, + file_hash, ) self._send(stream_id, chunk_data) @@ -304,13 +305,13 @@ def _read_and_encrypt( gek: bytes, file_path: Path, chunk_index: int, + file_hash: bytes, ) -> dict: """Read and encrypt one chunk (blocking — runs in executor).""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) - file_hash = blake3.blake3(file_path.read_bytes()).digest() pt_hash = blake3.blake3(plaintext).digest() ckey = derive_chunk_key(gek, file_hash, chunk_index) nonce, ct = encrypt_chunk(ckey, plaintext) diff --git a/packages/meshbay-node/src/meshbay_node/transport/server.py b/packages/meshbay-node/src/meshbay_node/transport/server.py index 76ac13a..b77f1f2 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/server.py @@ -197,7 +197,7 @@ class _ConnectionHandler: "type": "error", "detail": "File not on disk"}) continue - file_hash = blake3.blake3(file_path.read_bytes()).digest() + file_hash = bytes.fromhex(entry.id) chunk = await loop.run_in_executor( None, _serve_chunk, self._sk_node, self._gek, file_path, file_hash, chunk_index) 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 89391b9..7da6623 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -29,19 +29,14 @@ import struct from pathlib import Path from typing import Any -import blake3 import jwt import msgpack from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION -from meshbay_common.crypto import ( - chunk_key as derive_chunk_key, - encrypt_chunk, - sign_chunk, - pk_to_b64, -) +from meshbay_common.crypto import pk_to_b64 +from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex @@ -101,6 +96,7 @@ class WebRTCPeerSession: def _handle_message(self, msg: dict) -> None: mtype = msg.get("type") + log.debug("WebRTC recv: %s", mtype) try: if mtype == MNP.HANDSHAKE: self._do_handshake(msg) @@ -112,8 +108,12 @@ class WebRTCPeerSession: self._do_file_request(msg) elif mtype == MNP.STREAM_SEGMENT: self._do_stream_segment(msg) + elif mtype == MNP.GEK_REQUEST: + self._do_gek_request() elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message(msg) + elif mtype == MNP.CHAT_HISTORY: + self._do_chat_history(msg) else: log.warning("Unknown MNP message type on DataChannel: %s", mtype) except Exception as e: @@ -145,6 +145,10 @@ class WebRTCPeerSession: self._user_id = decoded["sub"] self._group_id = group_id + peers = self._ctx.get("_peers") + if peers is not None: + peers[self._user_id] = self + log.info("WebRTC handshake OK — user=%s group=%s", self._user_id[:8], group_id[:8] if group_id else "none") self._send({ @@ -160,11 +164,32 @@ class WebRTCPeerSession: def _do_index_sync(self) -> None: ctx = self._group_ctx() - wire = ctx["index"].serialize() + idx = ctx["index"] + entries = [ + { + "id": e.id, "name": e.name, "path": e.path, + "size": e.size, "type": e.type, "added_at": e.added_at, + } + for e in idx.entries + ] self._send({ "type": MNP.INDEX_SYNC, "v": MNP_VERSION, - "index_b64": base64.b64encode(wire).decode(), + "group_id": idx.group_id, + "version": idx.version, + "entries": entries, + }) + + def _do_gek_request(self) -> None: + ctx = self._group_ctx() + gek = ctx.get("gek") + if not gek: + self._send({"type": "error", "detail": "No GEK available"}) + return + self._send({ + "type": MNP.GEK_RESPONSE, + "v": MNP_VERSION, + "gek_b64": base64.b64encode(gek).decode(), }) def _do_file_request(self, msg: dict) -> None: @@ -173,6 +198,7 @@ class WebRTCPeerSession: chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) if not entry: + log.warning("File not found: %s", file_id[:16]) self._send({"type": "error", "detail": "File not found"}) return @@ -181,11 +207,13 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not on disk"}) return + file_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( self._ctx["sk_node"], ctx["gek"], file_path, chunk_index, + file_hash, ) self._send(chunk_data) @@ -235,21 +263,77 @@ class WebRTCPeerSession: def _do_chat_message(self, msg: dict) -> None: chat_store = self._ctx.get("chat_store") + payload = msg.get("payload", "") if chat_store: + raw = payload.encode() if isinstance(payload, str) else payload asyncio.ensure_future(chat_store.save_message( sender_id=msg.get("sender_id", self._user_id), iteration=msg.get("iteration", 0), - payload=msg.get("payload", b"").encode() - if isinstance(msg.get("payload"), str) else msg.get("payload", b""), + payload=raw, thread_id=msg.get("thread_id"), )) + + peers = self._ctx.get("_peers", {}) + broadcast = { + "type": MNP.CHAT_MESSAGE, + "v": MNP_VERSION, + "sender_id": msg.get("sender_id", self._user_id), + "payload": payload, + "thread_id": msg.get("thread_id"), + "timestamp": __import__("time").time(), + } + for uid, session in peers.items(): + if uid != self._user_id and session is not self: + try: + session._send(broadcast) + except Exception: + pass + self._send({"type": "ack", "v": MNP_VERSION}) + def _do_chat_history(self, msg: dict) -> None: + chat_store = self._ctx.get("chat_store") + if not chat_store: + self._send({ + "type": MNP.CHAT_HISTORY_RESPONSE, + "v": MNP_VERSION, + "messages": [], + }) + return + + since = msg.get("since", 0) + limit = msg.get("limit", 100) + asyncio.ensure_future(self._send_chat_history(chat_store, since, limit)) + + async def _send_chat_history(self, chat_store, since: float, limit: int) -> None: + msgs = await chat_store.get_messages(since=since, limit=limit) + self._send({ + "type": MNP.CHAT_HISTORY_RESPONSE, + "v": MNP_VERSION, + "messages": [ + { + "id": m.id, + "sender_id": m.sender_id, + "payload": m.payload.decode("utf-8", errors="replace") + if isinstance(m.payload, bytes) else m.payload, + "timestamp": m.timestamp, + "thread_id": m.thread_id, + } + for m in msgs + ], + }) + def _send(self, obj: dict) -> None: if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) + else: + log.warning("WebRTC send skipped: channel=%s", + self._channel.readyState if self._channel else "none") async def close(self) -> None: + peers = self._ctx.get("_peers") + if peers and self._user_id: + peers.pop(self._user_id, None) await self._pc.close() @@ -258,30 +342,22 @@ def _read_and_encrypt( gek: bytes, file_path: Path, chunk_index: int, + file_hash: bytes, ) -> dict: with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) - file_hash = blake3.blake3(file_path.read_bytes()).digest() - pt_hash = blake3.blake3(plaintext).digest() - ckey = derive_chunk_key(gek, file_hash, chunk_index) - nonce, ct = encrypt_chunk(ckey, plaintext) - ct_hash = blake3.blake3(ct).digest() - sig = sign_chunk(sk_node, chunk_index, nonce, ct_hash) + ckey = chunk_key_aes(gek, file_hash, chunk_index) + nonce, ct = encrypt_chunk_aes(ckey, plaintext) return { "type": MNP.FILE_CHUNK, "v": MNP_VERSION, "chunk_index": chunk_index, "plaintext_size": len(plaintext), - "nonce_b64": base64.b64encode(nonce).decode(), - "ct_b64": base64.b64encode(ct).decode(), - "ct_hash_b64": base64.b64encode(ct_hash).decode(), - "pt_hash_b64": base64.b64encode(pt_hash).decode(), - "sig_b64": base64.b64encode(sig).decode(), - "pk_node_b64": pk_to_b64(sk_node.public_key()), - "file_hash_b64": base64.b64encode(file_hash).decode(), + "nonce": nonce, + "ct": ct, } @@ -312,6 +388,7 @@ class WebRTCTransport: "gek": gek, "shared_root": shared_root, "index": index, + "_peers": {}, } if groups: self._ctx["groups"] = groups |