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>

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