""" MeshBay Spike 5 — Encrypted Transfer NODE (runs on Fedora, behind NAT) Acts as the "file server": - Creates a 5 MB test file if not present - Generates or loads GEK from node_state.json - Connects TCP OUT to meshbay.org:19003 (outgoing = always works through NAT) - Receives chunk request - Reads chunk from disk, derives per-chunk key via HKDF, encrypts, signs - Sends encrypted chunk with metadata - Reports timing Note: node initiates the TCP connection (reversed for POC — in production, the QUIC client connects to the node using the hole-punching from Spike 4). """ import asyncio, json, base64, struct, os, time from pathlib import Path from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305 from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives import hashes, serialization import blake3 MESHBAY_HOST = "meshbay.org" MESHBAY_PORT = 19003 CHUNK_SIZE = 1024 * 1024 # 1 MB TEST_FILE = Path("testfile.bin") STATE_FILE = Path("node_state.json") PASS = "✓"; FAIL = "✗" # ── Wire helpers ─────────────────────────────────────────────────────────────── async def send_msg(writer, obj: dict): data = json.dumps(obj).encode() writer.write(struct.pack('>I', len(data)) + data) await writer.drain() async def recv_msg(reader) -> dict: raw_len = await reader.readexactly(4) length = struct.unpack('>I', raw_len)[0] data = await reader.readexactly(length) return json.loads(data) # ── Chunk encryption ─────────────────────────────────────────────────────────── def make_encrypted_chunk( sk_node: Ed25519PrivateKey, gek_raw: bytes, file_path: Path, file_hash: bytes, chunk_index: int ) -> dict: """Read chunk from disk, compress (skipped), encrypt, sign. Return wire dict.""" # Read plaintext chunk with open(file_path, 'rb') as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) pt_hash = blake3.blake3(plaintext).digest() # Derive per-chunk key (deterministic from GEK + file identity + chunk index) chunk_key = HKDF( algorithm=hashes.SHA256(), length=32, salt=None, info=b"file:" + file_hash + b":chunk:" + chunk_index.to_bytes(4, 'big') ).derive(gek_raw) # Encrypt with ChaCha20-Poly1305 nonce = os.urandom(12) ciphertext = ChaCha20Poly1305(chunk_key).encrypt(nonce, plaintext, None) ct_hash = blake3.blake3(ciphertext).digest() # Sign: chunk_index || nonce || ct_hash sig_payload = chunk_index.to_bytes(4, 'big') + nonce + ct_hash signature = sk_node.sign(sig_payload) pk_bytes = sk_node.public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw) return { "chunk_index": chunk_index, "plaintext_size": len(plaintext), "nonce_b64": base64.b64encode(nonce).decode(), "ct_b64": base64.b64encode(ciphertext).decode(), "ct_hash_b64": base64.b64encode(ct_hash).decode(), "pt_hash_b64": base64.b64encode(pt_hash).decode(), "sig_b64": base64.b64encode(signature).decode(), "pk_node_b64": base64.b64encode(pk_bytes).decode(), "file_hash_b64": base64.b64encode(file_hash).decode(), "gek_b64": base64.b64encode(gek_raw).decode(), # POC ONLY } # ── Main ────────────────────────────────────────────────────────────────────── async def main(): print("\n=== MeshBay Spike 5 — Encrypted Transfer NODE ===\n") # Load state state = json.loads(STATE_FILE.read_text()) if STATE_FILE.exists() else {} # Load node Ed25519 key sk_node = Ed25519PrivateKey.from_private_bytes( base64.b64decode(state["sk_ed25519_b64"])) pk_bytes = sk_node.public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw) print(f" Node PK: {base64.b64encode(pk_bytes).decode()[:24]}...") # GEK — load or generate if "gek_b64" not in state: gek_raw = ChaCha20Poly1305.generate_key() state["gek_b64"] = base64.b64encode(gek_raw).decode() STATE_FILE.write_text(json.dumps(state, indent=2)) print(f" {PASS} GEK generated and saved") else: gek_raw = base64.b64decode(state["gek_b64"]) print(f" {PASS} GEK loaded from state") # Create test file if needed (5 MB random) if not TEST_FILE.exists(): print(f" Creating {TEST_FILE} (5 MB)...") with open(TEST_FILE, 'wb') as f: f.write(os.urandom(5 * 1024 * 1024)) file_size = TEST_FILE.stat().st_size print(f" {PASS} Test file: {TEST_FILE} ({file_size/1024/1024:.1f} MB)") # Pre-compute file hash (used in HKDF info string — identifies the file) print(f" Computing file hash...") t0 = time.perf_counter() file_hash = blake3.blake3(TEST_FILE.read_bytes()).digest() print(f" {PASS} File hash: {file_hash.hex()[:16]}... ({(time.perf_counter()-t0)*1000:.0f}ms)") # Connect to meshbay.org print(f"\n Connecting to {MESHBAY_HOST}:{MESHBAY_PORT} ...") reader, writer = await asyncio.open_connection(MESHBAY_HOST, MESHBAY_PORT) print(f" {PASS} TCP connection established") # Receive chunk request req = await recv_msg(reader) assert req["type"] == "chunk_req", f"Unexpected message type: {req['type']}" chunk_index = req["chunk_index"] file_id = req["file_id"] print(f" {PASS} Request received: file={file_id} chunk={chunk_index}") # Encrypt chunk on-the-fly print(f"\n Encrypting chunk {chunk_index} ({CHUNK_SIZE/1024:.0f} KB)...") t_enc = time.perf_counter() chunk_data = make_encrypted_chunk(sk_node, gek_raw, TEST_FILE, file_hash, chunk_index) enc_ms = (time.perf_counter() - t_enc) * 1000 ct_size = len(base64.b64decode(chunk_data["ct_b64"])) print(f" {PASS} Encrypted: {CHUNK_SIZE/1024:.0f} KB → {ct_size/1024:.1f} KB ciphertext") print(f" {PASS} Encrypt+sign time: {enc_ms:.1f} ms") # Send encrypted chunk print(f"\n Sending to {MESHBAY_HOST}:{MESHBAY_PORT}...") t_send = time.perf_counter() await send_msg(writer, chunk_data) writer.close() await writer.wait_closed() send_ms = (time.perf_counter() - t_send) * 1000 total_ms = enc_ms + send_ms payload_kb = (4 + len(json.dumps(chunk_data).encode())) / 1024 throughput = (payload_kb / 1024) / (send_ms / 1000) print(f" {PASS} Sent {payload_kb:.1f} KB in {send_ms:.0f}ms ({throughput:.1f} MB/s)") print(f"\n {'='*50}") print(f" Timings (node side):") print(f" Encrypt + sign : {enc_ms:.1f} ms") print(f" TCP send : {send_ms:.0f} ms") print(f" Total : {total_ms:.0f} ms") print(f"\n {PASS} Spike 5 NODE — chunk served successfully.") asyncio.run(main())