import urllib.request import urllib.error import json import time import uuid import base64 import asyncio import websockets import pymysql from cryptography.hazmat.primitives.asymmetric import ed25519, x25519 from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives import hashes BASE_URL = "http://127.0.0.1:8080" WS_URL = "ws://127.0.0.1:8080/api/v1/ws" import os def load_env_db_config(): pw = os.environ.get("DB_PASSWORD", "") if not pw and os.path.exists(".env"): with open(".env") as f: for line in f: if line.startswith("DB_PASSWORD="): pw = line.strip().split("=", 1)[1] return { "host": os.environ.get("DB_HOST", "94.136.190.60"), "user": os.environ.get("DB_USER", "chatpingdart_ravi"), "password": pw, "database": os.environ.get("DB_NAME", "chatpingdart_ucall"), } DB_CONFIG = load_env_db_config() def http_req(path, method="GET", payload=None, token=None): url = f"{BASE_URL}{path}" data = json.dumps(payload).encode() if payload is not None else None headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=10) as resp: return resp.status, json.loads(resp.read().decode()) except urllib.error.HTTPError as e: body = e.read().decode() try: return e.code, json.loads(body) except: return e.code, body print("=" * 70) print("UCALL E2EE & REAL-TIME RECONNECT SYNC VERIFICATION SUITE") print("=" * 70) # Helper: Register & create test users def register_user(label): phone = f"+1{int(time.time() * 1000) % 10000000000:010d}" st, res = http_req("/api/v1/auth/register", "POST", {"phone": phone, "display_name": label}) assert st == 200, f"Register failed: {res}" user_id = res["data"]["user_id"] token = res["data"]["access_token"] uid = res["data"]["universal_id"] return { "label": label, "phone": phone, "user_id": user_id, "universal_id": uid, "token": token, } # Register Alice and Bob alice = register_user("Alice E2EE") time.sleep(0.05) bob = register_user("Bob E2EE") print(f" ✓ Registered Alice (UID: {alice['universal_id']}) and Bob (UID: {bob['universal_id']})") # ------------------------------------------------------------------------------ # TEST 1: Cryptographic Device Identity & PreKey Generation # ------------------------------------------------------------------------------ print("\n[TEST 1] Device Identity & Signed PreKey Publication...") # Generate Alice device keys alice_ik_priv = ed25519.Ed25519PrivateKey.generate() alice_ik_pub_bytes = alice_ik_priv.public_key().public_bytes_raw() alice_ik_pub_b64 = base64.b64encode(alice_ik_pub_bytes).decode() alice_spk_priv = x25519.X25519PrivateKey.generate() alice_spk_pub_bytes = alice_spk_priv.public_key().public_bytes_raw() alice_spk_pub_b64 = base64.b64encode(alice_spk_pub_bytes).decode() alice_spk_sig = alice_ik_priv.sign(alice_spk_pub_bytes) alice_spk_sig_b64 = base64.b64encode(alice_spk_sig).decode() alice_device_id = str(uuid.uuid4()) # Generate Bob device keys bob_ik_priv = ed25519.Ed25519PrivateKey.generate() bob_ik_pub_bytes = bob_ik_priv.public_key().public_bytes_raw() bob_ik_pub_b64 = base64.b64encode(bob_ik_pub_bytes).decode() bob_spk_priv = x25519.X25519PrivateKey.generate() bob_spk_pub_bytes = bob_spk_priv.public_key().public_bytes_raw() bob_spk_pub_b64 = base64.b64encode(bob_spk_pub_bytes).decode() bob_spk_sig = bob_ik_priv.sign(bob_spk_pub_bytes) bob_spk_sig_b64 = base64.b64encode(bob_spk_sig).decode() bob_device_id = str(uuid.uuid4()) # Bob generates 5 one-time prekeys bob_opks = [] bob_opk_privs = {} for i in range(1, 6): priv = x25519.X25519PrivateKey.generate() pub_bytes = priv.public_key().public_bytes_raw() pub_b64 = base64.b64encode(pub_bytes).decode() bob_opks.append({"key_id": i, "public_key": pub_b64}) bob_opk_privs[i] = priv # 1a. Test forged signature rejection print(" Subtest 1a: Verifying forged signature rejection...") forged_sig_b64 = base64.b64encode(b"\x00" * 64).decode() st, res = http_req("/api/v1/e2ee/keys", "POST", { "device_id": bob_device_id, "identity_key": bob_ik_pub_b64, "signed_prekey_id": 1, "signed_prekey": bob_spk_pub_b64, "signed_prekey_sig": forged_sig_b64, "one_time_prekeys": bob_opks, }, token=bob["token"]) assert st in [400, 422], f"Expected 400 for forged signature, got {st}: {res}" print(" ✓ Server cryptographically rejected forged prekey signature with HTTP 400") # 1b. Test valid publication print(" Subtest 1b: Publishing valid Bob PreKeys...") st, res = http_req("/api/v1/e2ee/keys", "POST", { "device_id": bob_device_id, "identity_key": bob_ik_pub_b64, "signed_prekey_id": 1, "signed_prekey": bob_spk_pub_b64, "signed_prekey_sig": bob_spk_sig_b64, "one_time_prekeys": bob_opks, }, token=bob["token"]) assert st == 200, f"Publish keys failed: {res}" assert res["data"]["published_opk_count"] == 5 assert res["data"]["total_available_opks"] == 5 print(f" ✓ Bob published 5 OPKs successfully (available: {res['data']['total_available_opks']})") # Also publish Alice keys st, res = http_req("/api/v1/e2ee/keys", "POST", { "device_id": alice_device_id, "identity_key": alice_ik_pub_b64, "signed_prekey_id": 1, "signed_prekey": alice_spk_pub_b64, "signed_prekey_sig": alice_spk_sig_b64, "one_time_prekeys": [], }, token=alice["token"]) assert st == 200 # ------------------------------------------------------------------------------ # TEST 2: Atomic OPK Consumption & Multi-Device Fanout # ------------------------------------------------------------------------------ print("\n[TEST 2] Atomic Single-Use OPK Consumption...") # Alice fetches Bob's prekey bundle st, res = http_req(f"/api/v1/e2ee/keys/{bob['universal_id']}", "GET", token=alice["token"]) assert st == 200, f"Fetch bundle failed: {res}" bundles = res["data"] assert len(bundles) == 1 bundle1 = bundles[0] assert bundle1["device_id"] == bob_device_id assert bundle1["identity_key"] == bob_ik_pub_b64 assert bundle1["one_time_prekey"] is not None opk1_id = bundle1["one_time_prekey"]["key_id"] print(f" ✓ Fetched Bob bundle, claimed OPK ID {opk1_id}") # Alice fetches Bob's prekey bundle a second time (retry protection test) st, retry_res = http_req(f"/api/v1/e2ee/keys/{bob['universal_id']}", "GET", token=alice["token"]) assert retry_res["data"][0]["one_time_prekey"]["key_id"] == opk1_id print(f" ✓ Alice retry returned cached OPK ID {opk1_id} (protected against retry depletion)") # Carol (distinct caller) fetches Bob's prekey bundle carol = register_user("Carol E2EE") st, res2 = http_req(f"/api/v1/e2ee/keys/{bob['universal_id']}", "GET", token=carol["token"]) bundle2 = res2["data"][0] opk2_id = bundle2["one_time_prekey"]["key_id"] assert opk1_id != opk2_id, f"OPK was not consumed for new caller! Repeated key_id: {opk1_id}" print(f" ✓ Second distinct caller (Carol) consumed next OPK ID {opk2_id}") # Check remaining count for Bob st, count_res = http_req(f"/api/v1/e2ee/keys/count?device_id={bob_device_id}", "GET", token=bob["token"]) assert st == 200 assert count_res["data"]["available_opks"] == 3, f"Expected 3 remaining OPKs, got {count_res['data']['available_opks']}" print(f" ✓ Remaining available OPKs for Bob: {count_res['data']['available_opks']}") # Verify MariaDB row state conn = pymysql.connect(**DB_CONFIG) cur = conn.cursor() cur.execute("SELECT key_id, is_consumed, consumed_at FROM e2ee_one_time_prekeys WHERE device_id = %s ORDER BY key_id ASC", (bob_device_id,)) rows = cur.fetchall() consumed_keys = [r[0] for r in rows if r[1] == 1] assert set(consumed_keys) == {opk1_id, opk2_id} print(f" ✓ Remote MariaDB rows verified: keys {consumed_keys} marked is_consumed=1 with non-null consumed_at") conn.close() # ------------------------------------------------------------------------------ # TEST 3: X3DH Key Agreement & Double Ratchet End-to-End Encryption # ------------------------------------------------------------------------------ print("\n[TEST 3] X3DH Key Agreement & Zero Plaintext Transmission...") alice_ek_priv = x25519.X25519PrivateKey.generate() alice_ek_pub_bytes = alice_ek_priv.public_key().public_bytes_raw() bob_spk_pub = x25519.X25519PublicKey.from_public_bytes(base64.b64decode(bundle1["signed_prekey"])) bob_opk_pub = x25519.X25519PublicKey.from_public_bytes(base64.b64decode(bundle1["one_time_prekey"]["public_key"])) dh_shared = alice_ek_priv.exchange(bob_spk_pub) + alice_ek_priv.exchange(bob_opk_pub) hkdf = HKDF( algorithm=hashes.SHA256(), length=32, salt=b"\x00" * 32, info=b"UCall-E2EE-X3DH-v1", ) session_key = hkdf.derive(dh_shared) # Alice encrypts a confidential plaintext message aesgcm = AESGCM(session_key) nonce = b"\x01" * 12 plaintext = b"WhatsApp-style ultra secret message from Alice to Bob" ciphertext = aesgcm.encrypt(nonce, plaintext, None) envelope = { "sender_device_id": alice_device_id, "recipient_device_id": bob_device_id, "dh_pub": base64.b64encode(alice_ek_pub_bytes).decode(), "sequence_num": 1, "previous_chain_len": 0, "ciphertext": base64.b64encode(ciphertext).decode(), "nonce": base64.b64encode(nonce).decode(), "one_time_prekey_id": opk1_id, } # Alice sends encrypted message via POST /api/v1/messages st, send_res = http_req("/api/v1/messages", "POST", { "recipient_universal_id": bob["universal_id"], "message_type": "e2ee", "payload": json.dumps(envelope), "sender_device_id": alice_device_id, "recipient_device_id": bob_device_id, }, token=alice["token"]) assert st == 200, f"Send message failed: {send_res}" msg_data = send_res["data"] assert msg_data["status"] == "queued" assert msg_data["seq"] is not None msg_seq = msg_data["seq"] print(f" ✓ Encrypted envelope sent to Bob; server assigned monotonic sequence seq={msg_seq}") # 3b. Bob derives same key and decrypts bob_opk_priv = bob_opk_privs[opk1_id] alice_ek_pub = x25519.X25519PublicKey.from_public_bytes(alice_ek_pub_bytes) bob_dh_shared = bob_spk_priv.exchange(alice_ek_pub) + bob_opk_priv.exchange(alice_ek_pub) bob_hkdf = HKDF( algorithm=hashes.SHA256(), length=32, salt=b"\x00" * 32, info=b"UCall-E2EE-X3DH-v1", ) bob_session_key = bob_hkdf.derive(bob_dh_shared) assert bob_session_key == session_key, "Cryptographic key derivation mismatch!" bob_aesgcm = AESGCM(bob_session_key) decrypted = bob_aesgcm.decrypt(nonce, ciphertext, None) assert decrypted == plaintext print(f" ✓ Bob derived matching key and decrypted: '{decrypted.decode()}'") # 3c. Verify tampered ciphertext is rejected print(" Subtest 3c: Testing corrupt ciphertext rejection...") tampered_ct = bytearray(ciphertext) tampered_ct[5] ^= 0xFF # Flip bits in ciphertext try: bob_aesgcm.decrypt(nonce, bytes(tampered_ct), None) assert False, "Decryption succeeded on tampered ciphertext!" except Exception as e: print(" ✓ AES-GCM authentication tag rejected tampered ciphertext") # ------------------------------------------------------------------------------ # TEST 4: Cursor-Based Reconnect Synchronization (WhatsApp-Style) # ------------------------------------------------------------------------------ print("\n[TEST 4] Cursor-Based Reconnect Synchronization...") # Alice sends 2 more queued messages while Bob is offline for i in range(2, 4): pt = f"Queued offline message #{i}".encode() ct = aesgcm.encrypt(nonce, pt, None) env = { "sender_device_id": alice_device_id, "recipient_device_id": bob_device_id, "sequence_num": i, "ciphertext": base64.b64encode(ct).decode(), "nonce": base64.b64encode(nonce).decode(), } st, resp = http_req("/api/v1/messages", "POST", { "recipient_universal_id": bob["universal_id"], "message_type": "e2ee", "payload": json.dumps(env), }, token=alice["token"]) assert st == 200 # Bob reconnects and requests sync from seq=0 st, sync_resp = http_req(f"/api/v1/messages/sync?since_seq=0&limit=50", "GET", token=bob["token"]) assert st == 200, f"Sync failed: {sync_resp}" sync_data = sync_resp["data"] messages = sync_data["messages"] assert len(messages) >= 3, f"Expected at least 3 synced messages, got {len(messages)}" # Verify monotonic order seqs = [m["seq"] for m in messages] assert seqs == sorted(seqs), f"Messages not in monotonic sequence order: {seqs}" last_seq = sync_data["last_seq"] print(f" ✓ Bob synced {len(messages)} messages since seq=0; last_seq={last_seq}") # Bob requests sync from last_seq st, catchup_resp = http_req(f"/api/v1/messages/sync?since_seq={last_seq}&limit=50", "GET", token=bob["token"]) assert st == 200 assert len(catchup_resp["data"]["messages"]) == 0 print(f" ✓ Bob caught up to last_seq={last_seq}; 0 new messages returned") # ------------------------------------------------------------------------------ # TEST 5: Real-Time WebSocket Delivery, E2EE Envelope & Delivery Receipts # ------------------------------------------------------------------------------ print("\n[TEST 5] Real-Time WebSocket E2EE Delivery & Receipts...") async def run_ws_test(): # 1. Obtain single-use WS tickets for Alice and Bob st_a, ticket_a = http_req("/api/v1/auth/ws-ticket", "POST", token=alice["token"]) st_b, ticket_b = http_req("/api/v1/auth/ws-ticket", "POST", token=bob["token"]) assert st_a == 200 and st_b == 200 t_a = ticket_a["data"]["ticket"] t_b = ticket_b["data"]["ticket"] ws_url_a = f"{WS_URL}?ticket={t_a}" ws_url_b = f"{WS_URL}?ticket={t_b}" async with websockets.connect(ws_url_a) as ws_alice, websockets.connect(ws_url_b) as ws_bob: # Both send presence online await ws_alice.send(json.dumps({"method": "presence.set", "params": {"status": "online"}})) await ws_bob.send(json.dumps({"method": "presence.set", "params": {"status": "online"}})) await asyncio.sleep(0.1) # Alice sends E2EE message over WebSocket ws_plaintext = b"Live instant message over WebSocket E2EE" ws_ct = aesgcm.encrypt(nonce, ws_plaintext, None) ws_env = { "sender_device_id": alice_device_id, "recipient_device_id": bob_device_id, "ciphertext": base64.b64encode(ws_ct).decode(), "nonce": base64.b64encode(nonce).decode(), "sequence_num": 10, } client_msg_id = str(uuid.uuid4()) send_cmd = { "method": "msg.e2ee", "params": { "recipient_universal_id": bob["universal_id"], "sender_device_id": alice_device_id, "recipient_device_id": bob_device_id, "payload": json.dumps(ws_env), "client_message_id": client_msg_id, } } await ws_alice.send(json.dumps(send_cmd)) # 1. Alice should receive msg.ack ack_raw = await asyncio.wait_for(ws_alice.recv(), timeout=5.0) ack = json.loads(ack_raw) assert ack["method"] == "msg.ack" assert ack["params"]["client_message_id"] == client_msg_id assigned_msg_id = ack["params"]["message_id"] print(f" ✓ Alice received msg.ack for msg_id={assigned_msg_id}") # 2. Bob should immediately receive msg.receive recv_raw = await asyncio.wait_for(ws_bob.recv(), timeout=5.0) recv_msg = json.loads(recv_raw) assert recv_msg["method"] == "msg.receive" assert recv_msg["params"]["message_type"] == "e2ee" received_payload = json.loads(recv_msg["params"]["payload"]) rec_ct = base64.b64decode(received_payload["ciphertext"]) decrypted_live = bob_aesgcm.decrypt(nonce, rec_ct, None) assert decrypted_live == ws_plaintext print(f" ✓ Bob received real-time msg.receive and decrypted: '{decrypted_live.decode()}'") # 3. Bob marks message as delivered await ws_bob.send(json.dumps({ "method": "msg.delivered", "params": {"message_id": assigned_msg_id} })) # 4. Bob marks message as read await ws_bob.send(json.dumps({ "method": "msg.read", "params": {"message_id": assigned_msg_id} })) # 5. Alice receives read receipt receipt_raw = await asyncio.wait_for(ws_alice.recv(), timeout=5.0) receipt = json.loads(receipt_raw) assert receipt["method"] == "msg.receipt" assert receipt["params"]["message_id"] == assigned_msg_id assert receipt["params"]["status"] == "read" print(f" ✓ Alice received real-time read receipt for msg_id={assigned_msg_id}") # 6. Test WebSocket cursor sync (msg.sync) await ws_bob.send(json.dumps({ "method": "msg.sync", "params": {"since_seq": 0, "limit": 10} })) sync_batch_raw = await asyncio.wait_for(ws_bob.recv(), timeout=5.0) sync_batch = json.loads(sync_batch_raw) assert sync_batch["method"] == "msg.sync_batch" assert len(sync_batch["params"]["messages"]) > 0 print(f" ✓ Bob executed msg.sync over WebSocket; received {len(sync_batch['params']['messages'])} messages") asyncio.run(run_ws_test()) print("\n" + "=" * 70) print("ALL E2EE & REAL-TIME RECONNECT SYNC TESTS PASSED (100%)") print("=" * 70)