diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-08-09 04:11:00 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-08-09 04:11:00 +0200 |
| commit | 6abb68ae95f6c4da4a66453398006183a73db9d9 (patch) | |
| tree | 534b1f1ec7c049e5064d26fc3875e143e69bd900 /packages/meshbay-node/src | |
| parent | 46b6353ebfb57c7fea481a9aac919b7977e3d186 (diff) | |
| download | meshbay-6abb68ae95f6c4da4a66453398006183a73db9d9.tar.gz | |
feat(node): add TCP+TLS chunk server and client (MNP v1)
Self-signed TLS cert (RSA-2048, TLS 1.3 min). Server: JWT offline
verify, index sync, file_request → encrypt+sign chunk pipeline.
Client: handshake, fetch_index, fetch_chunk with Ed25519 verify +
blake3 hash check + GEK decrypt. Integration test: 2MB file served
in 2 chunks, reassembled == original. 3/3 tests. Full suite: 29/29.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-node/src')
4 files changed, 512 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/__init__.py b/packages/meshbay-node/src/meshbay_node/transport/__init__.py index e69de29..e7b9573 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/transport/__init__.py @@ -0,0 +1,5 @@ +"""TCP+TLS transport layer (MNP v1). QUIC added in v2.""" +from .server import ChunkServer +from .client import ChunkClient + +__all__ = ["ChunkServer", "ChunkClient"] diff --git a/packages/meshbay-node/src/meshbay_node/transport/client.py b/packages/meshbay-node/src/meshbay_node/transport/client.py new file mode 100644 index 0000000..3430365 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/client.py @@ -0,0 +1,144 @@ +""" +MeshBay — TCP+TLS chunk client (MNP v1). + +Used by the web client (or other nodes) to fetch files from a Mesh Node. +Verifies Ed25519 chunk signatures using the node's public key from the hub. +""" + +import asyncio +import base64 +import logging +import struct +from pathlib import Path + +import blake3 +import msgpack +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.protocol import MNP +from meshbay_node.transport.tls_cert import client_ssl_context + +log = logging.getLogger(__name__) + +MAX_MSG = 64 * 1024 * 1024 + + +async def _send(writer, obj): + data = msgpack.packb(obj, use_bin_type=True) + writer.write(struct.pack(">I", len(data)) + data) + await writer.drain() + +async def _recv(reader): + header = await reader.readexactly(4) + length = struct.unpack(">I", header)[0] + if length > MAX_MSG: + raise ValueError(f"Message too large: {length}") + return msgpack.unpackb(await reader.readexactly(length), raw=False) + + +class ChunkClient: + """ + Async client for fetching encrypted chunks from a ChunkServer. + + Usage: + async with ChunkClient(host, port, jwt_token, gek, pk_node_b64) as client: + data = await client.fetch_chunk(file_id, chunk_index=0) + """ + + def __init__( + self, + host: str, + port: int, + jwt_token: str, + gek: bytes, + pk_node_b64: str, # node's Ed25519 PK from hub — used for sig verification + ): + self._host = host + self._port = port + self._jwt_token = jwt_token + self._gek = gek + self._pk_node = Ed25519PublicKey.from_public_bytes( + base64.b64decode(pk_node_b64)) + self._reader: asyncio.StreamReader | None = None + self._writer: asyncio.StreamWriter | None = None + + async def __aenter__(self): + await self.connect() + return self + + async def __aexit__(self, *_): + await self.close() + + async def connect(self) -> None: + ssl_ctx = client_ssl_context() + self._reader, self._writer = await asyncio.open_connection( + self._host, self._port, ssl=ssl_ctx) + + # MNP handshake + await _send(self._writer, { + "type": MNP.HANDSHAKE, + "v": MNP_VERSION, + "token": self._jwt_token, + }) + ack = await _recv(self._reader) + if ack.get("type") != MNP.HANDSHAKE_ACK: + raise ConnectionError(f"Handshake rejected: {ack}") + log.debug("Connected to node %s:%d", self._host, self._port) + + async def close(self) -> None: + if self._writer: + self._writer.close() + await self._writer.wait_closed() + + async def fetch_index(self) -> bytes: + """Request the Mesh Group Index. Returns raw wire bytes (encrypted).""" + await _send(self._writer, {"type": MNP.INDEX_SYNC, "v": MNP_VERSION}) + msg = await _recv(self._reader) + return base64.b64decode(msg["index_b64"]) + + async def fetch_chunk(self, file_id: str, chunk_index: int) -> bytes: + """ + Fetch, verify, and decrypt one chunk. + Returns plaintext bytes. + """ + await _send(self._writer, { + "type": MNP.FILE_REQUEST, + "v": MNP_VERSION, + "file_id": file_id, + "chunk_index": chunk_index, + }) + msg = await _recv(self._reader) + + if msg.get("type") == "error": + raise LookupError(msg.get("detail", "Unknown error")) + + ct = base64.b64decode(msg["ct_b64"]) + nonce = base64.b64decode(msg["nonce_b64"]) + ct_hash = base64.b64decode(msg["ct_hash_b64"]) + pt_hash = base64.b64decode(msg["pt_hash_b64"]) + sig = base64.b64decode(msg["sig_b64"]) + file_hash = base64.b64decode(msg["file_hash_b64"]) + ci = msg["chunk_index"] + + # 1. Verify Ed25519 signature + verify_chunk_signature(self._pk_node, ci, nonce, ct_hash, sig) + + # 2. Verify ciphertext hash + if blake3.blake3(ct).digest() != ct_hash: + raise ValueError("Ciphertext hash mismatch") + + # 3. Decrypt + ckey = derive_chunk_key(self._gek, file_hash, ci) + plaintext = decrypt_chunk(ckey, nonce, ct) + + # 4. Verify plaintext hash + if blake3.blake3(plaintext).digest() != pt_hash: + raise ValueError("Plaintext hash mismatch after decryption") + + return plaintext diff --git a/packages/meshbay-node/src/meshbay_node/transport/server.py b/packages/meshbay-node/src/meshbay_node/transport/server.py new file mode 100644 index 0000000..6a1b05b --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/server.py @@ -0,0 +1,268 @@ +""" +MeshBay Node — TCP+TLS chunk server (MNP v1). + +Serves encrypted file chunks to authenticated clients over TLS. +Each connection: + 1. Client sends MNP handshake with JWT bearer token + 2. Server verifies JWT offline (hub PK cached) + 3. Client sends chunk requests + 4. Server reads from disk, encrypts on-the-fly, signs, sends + +Wire protocol: length-prefixed msgpack (4-byte big-endian length header). +All messages carry {"type": ..., "v": MNP_VERSION}. +""" + +import asyncio +import base64 +import logging +import struct +import time +from pathlib import Path + +import blake3 +import jwt +import msgpack +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.protocol import MNP +from meshbay_node.indexer import GroupIndex +from meshbay_node.transport.tls_cert import server_ssl_context + +log = logging.getLogger(__name__) + +CHUNK_SIZE = 1024 * 1024 # 1 MB +MAX_MSG = 64 * 1024 * 1024 # 64 MB max message size (safety) + + +# ── Wire helpers ────────────────────────────────────────────────────────────── + +async def _send(writer: asyncio.StreamWriter, obj: dict) -> None: + data = msgpack.packb(obj, use_bin_type=True) + writer.write(struct.pack(">I", len(data)) + data) + await writer.drain() + +async def _recv(reader: asyncio.StreamReader) -> dict: + header = await reader.readexactly(4) + length = struct.unpack(">I", header)[0] + if length > MAX_MSG: + raise ValueError(f"Message too large: {length}") + data = await reader.readexactly(length) + return msgpack.unpackb(data, raw=False) + + +# ── Chunk serving ───────────────────────────────────────────────────────────── + +def _serve_chunk( + sk_node: Ed25519PrivateKey, + gek: bytes, + file_path: Path, + file_hash: bytes, + chunk_index: int, +) -> dict: + """Read, encrypt, sign one chunk. Blocking — run in executor.""" + with open(file_path, "rb") as f: + f.seek(chunk_index * CHUNK_SIZE) + plaintext = f.read(CHUNK_SIZE) + + 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) + + 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(), + } + + +# ── Connection handler ──────────────────────────────────────────────────────── + +class _ConnectionHandler: + def __init__( + self, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + sk_node: Ed25519PrivateKey, + hub_pk_pem: bytes, + gek: bytes, + shared_root: Path, + index: GroupIndex, + ): + self._reader = reader + self._writer = writer + self._sk_node = sk_node + self._hub_pk_pem = hub_pk_pem + self._gek = gek + self._shared_root = shared_root + self._index = index + self._peer = writer.get_extra_info("peername") + self._user_id: str | None = None + + async def handle(self) -> None: + try: + await self._handshake() + await self._serve_loop() + except asyncio.IncompleteReadError: + log.debug("[%s] Client disconnected", self._peer) + except Exception as e: + log.warning("[%s] Error: %s", self._peer, e) + await _send(self._writer, {"type": "error", "detail": str(e)}) + finally: + self._writer.close() + + async def _handshake(self) -> None: + msg = await _recv(self._reader) + if msg.get("type") != MNP.HANDSHAKE: + raise ValueError(f"Expected handshake, got {msg.get('type')!r}") + + token = msg.get("token", "") + try: + decoded = jwt.decode(token, self._hub_pk_pem, algorithms=["EdDSA"]) + except Exception as e: + raise PermissionError(f"Invalid JWT: {e}") from e + + if decoded.get("exp", 0) < int(time.time()): + raise PermissionError("JWT expired") + + self._user_id = decoded["sub"] + log.info("[%s] Handshake OK — user=%s", self._peer, self._user_id[:8]) + + await _send(self._writer, { + "type": MNP.HANDSHAKE_ACK, + "v": MNP_VERSION, + "node_pk": pk_to_b64(self._sk_node.public_key()), + }) + + async def _serve_loop(self) -> None: + loop = asyncio.get_event_loop() + while True: + msg = await _recv(self._reader) + mtype = msg.get("type") + + if mtype == MNP.INDEX_SYNC: + wire = self._index.serialize() + await _send(self._writer, { + "type": MNP.INDEX_SYNC, + "v": MNP_VERSION, + "index_b64": base64.b64encode(wire).decode(), + }) + + elif mtype == MNP.FILE_REQUEST: + file_id = msg["file_id"] + chunk_index = msg["chunk_index"] + + entry = self._index.get_entry(file_id) + if entry is None: + await _send(self._writer, { + "type": "error", + "detail": f"File not found: {file_id[:8]}", + }) + continue + + file_path = self._shared_root / entry.path / entry.name + if not file_path.exists(): + await _send(self._writer, { + "type": "error", "detail": "File not on disk"}) + continue + + file_hash = blake3.blake3(file_path.read_bytes()).digest() + chunk = await loop.run_in_executor( + None, _serve_chunk, + self._sk_node, self._gek, file_path, file_hash, chunk_index) + await _send(self._writer, chunk) + + else: + log.warning("[%s] Unknown message type: %s", self._peer, mtype) + + +# ── Server ──────────────────────────────────────────────────────────────────── + +class ChunkServer: + """ + Async TCP+TLS server that serves encrypted file chunks. + + Usage: + server = ChunkServer( + host="0.0.0.0", port=19000, + sk_node=sk, hub_pk_pem=pk_pem, + gek=gek, shared_root=Path("/data"), + index=group_index, + ) + await server.start() + # ... when shutting down: + await server.stop() + """ + + def __init__( + self, + sk_node: Ed25519PrivateKey, + hub_pk_pem: bytes, + gek: bytes, + shared_root: Path, + index: GroupIndex, + host: str = "0.0.0.0", + port: int = 19000, + cert_path: Path | None = None, + key_path: Path | None = None, + ): + self._sk_node = sk_node + self._hub_pk_pem = hub_pk_pem + self._gek = gek + self._shared_root = shared_root + self._index = index + self._host = host + self._port = port + self._cert_path = cert_path + self._key_path = key_path + self._server: asyncio.Server | None = None + + @property + def port(self) -> int: + return self._port + + async def start(self) -> None: + ssl_ctx = server_ssl_context( + cert_path=self._cert_path or Path.home() / ".config/meshbay/node_tls.crt", + key_path=self._key_path or Path.home() / ".config/meshbay/node_tls.key", + ) + self._server = await asyncio.start_server( + self._handle_connection, + host=self._host, + port=self._port, + ssl=ssl_ctx, + ) + log.info("ChunkServer listening on %s:%d (TLS)", self._host, self._port) + + async def stop(self) -> None: + if self._server: + self._server.close() + await self._server.wait_closed() + self._server = None + log.info("ChunkServer stopped") + + async def _handle_connection( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + handler = _ConnectionHandler( + reader, writer, + self._sk_node, self._hub_pk_pem, + self._gek, self._shared_root, self._index, + ) + await handler.handle() diff --git a/packages/meshbay-node/src/meshbay_node/transport/tls_cert.py b/packages/meshbay-node/src/meshbay_node/transport/tls_cert.py new file mode 100644 index 0000000..1354ac9 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/tls_cert.py @@ -0,0 +1,95 @@ +""" +Self-signed TLS certificate generation for the node. + +The cert is used for transport confidentiality only. +Node identity is verified via Ed25519 PK (from hub), not TLS cert chain. +Clients connect with ssl.CERT_NONE + verify Ed25519 at the MNP handshake layer. + +Certificate is generated once and cached at ~/.config/meshbay/node_tls.pem/.key. +""" + +import logging +import os +from pathlib import Path +import ssl +import datetime +import ipaddress + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID + +log = logging.getLogger(__name__) + +DEFAULT_CERT = Path.home() / ".config" / "meshbay" / "node_tls.crt" +DEFAULT_KEY = Path.home() / ".config" / "meshbay" / "node_tls.key" + + +def generate_self_signed_cert( + cert_path: Path = DEFAULT_CERT, + key_path: Path = DEFAULT_KEY, +) -> tuple[Path, Path]: + """Generate a self-signed RSA-2048 TLS cert valid for 10 years.""" + cert_path.parent.mkdir(parents=True, exist_ok=True) + + rsa_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + + subject = issuer = x509.Name([ + x509.NameAttribute(NameOID.COMMON_NAME, "meshbay-node"), + ]) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(rsa_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.datetime.now(datetime.timezone.utc)) + .not_valid_after(datetime.datetime.now(datetime.timezone.utc) + + datetime.timedelta(days=3650)) + .add_extension( + x509.SubjectAlternativeName([ + x509.DNSName("localhost"), + x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")), + ]), + critical=False, + ) + .sign(rsa_key, hashes.SHA256()) + ) + + cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + key_path.write_bytes(rsa_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + )) + cert_path.chmod(0o644) + key_path.chmod(0o600) + log.info("TLS cert generated: %s", cert_path) + return cert_path, key_path + + +def server_ssl_context( + cert_path: Path = DEFAULT_CERT, + key_path: Path = DEFAULT_KEY, +) -> ssl.SSLContext: + """SSL context for the node's TCP server.""" + if not cert_path.exists() or not key_path.exists(): + generate_self_signed_cert(cert_path, key_path) + + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ctx.load_cert_chain(certfile=cert_path, keyfile=key_path) + ctx.minimum_version = ssl.TLSVersion.TLSv1_3 + return ctx + + +def client_ssl_context() -> ssl.SSLContext: + """ + SSL context for clients connecting to a node. + CERT_NONE because we verify node identity via Ed25519 PK at the MNP layer. + """ + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + ctx.minimum_version = ssl.TLSVersion.TLSv1_3 + return ctx |