import urllib.request import urllib.error import json import time import uuid import base64 import threading import concurrent.futures import os import pymysql BASE_URL = "http://127.0.0.1:8080" 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=15) 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 CONCURRENCY, ATOMIC OPK & SYNC RACE REGRESSION SUITE") print("=" * 70) 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}" return { "label": label, "user_id": res["data"]["user_id"], "token": res["data"]["access_token"], "universal_id": res["data"]["universal_id"], } # 1. Register test users alice = register_user("Alice Stress") time.sleep(0.05) bob = register_user("Bob Recipient") time.sleep(0.05) carol = register_user("Carol Stress") print(f" ✓ Registered test users: Alice, Bob, and Carol") # ------------------------------------------------------------------------------ # TEST 1: Atomic One-Time Prekey (OPK) Concurrent Race Verification # ------------------------------------------------------------------------------ print("\n[TEST 1] Verifying Atomic Single-Use OPK Consumption Under Concurrency...") # Bob publishes 20 OPKs from cryptography.hazmat.primitives.asymmetric import ed25519, x25519 bob_ik = ed25519.Ed25519PrivateKey.generate() bob_ik_pub_b64 = base64.b64encode(bob_ik.public_key().public_bytes_raw()).decode() bob_spk = x25519.X25519PrivateKey.generate() bob_spk_pub_bytes = bob_spk.public_key().public_bytes_raw() bob_spk_pub_b64 = base64.b64encode(bob_spk_pub_bytes).decode() bob_spk_sig_b64 = base64.b64encode(bob_ik.sign(bob_spk_pub_bytes)).decode() bob_dev_id = str(uuid.uuid4()) bob_opks = [] for i in range(1, 21): priv = x25519.X25519PrivateKey.generate() pub_b64 = base64.b64encode(priv.public_key().public_bytes_raw()).decode() bob_opks.append({"key_id": i, "public_key": pub_b64}) st, pub_res = http_req("/api/v1/e2ee/keys", "POST", { "device_id": bob_dev_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 and pub_res["data"]["total_available_opks"] == 20 print(f" ✓ Bob published 20 OPKs") # 1a. Security Check: Unauthenticated request MUST be rejected print(" Subtest 1a: Verifying unauthenticated prekey fetch is blocked...") st, res = http_req(f"/api/v1/e2ee/keys/{bob['universal_id']}", "GET", token=None) assert st in [401, 403], f"Expected 401 Unauthorized, got {st}" print(" ✓ Unauthenticated prekey request blocked with HTTP 401") # 1b. Concurrency Race: Multiple distinct users fetch Bob's bundle simultaneously print(" Subtest 1b: Running 6 concurrent prekey claims from distinct callers...") other_tokens = [alice["token"], carol["token"]] for i in range(4): u = register_user(f"Caller {i}") other_tokens.append(u["token"]) claimed_opk_ids = [] lock = threading.Lock() def claim_bundle(token): s, r = http_req(f"/api/v1/e2ee/keys/{bob['universal_id']}", "GET", token=token) if s == 200: b = r["data"][0] # Verify server NEVER returns private keys assert "private_key" not in b assert "secret" not in b opk = b.get("one_time_prekey") if opk: with lock: claimed_opk_ids.append(opk["key_id"]) with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor: futures = [executor.submit(claim_bundle, t) for t in other_tokens] concurrent.futures.wait(futures) # Every claimed OPK id MUST be distinct (zero duplicates) print(f" Claimed OPK IDs: {claimed_opk_ids}") assert len(claimed_opk_ids) == len(set(claimed_opk_ids)), f"Duplicate OPKs issued under race! {claimed_opk_ids}" print(f" ✓ 100% unique OPK issuance under concurrency: 0 duplicates across {len(claimed_opk_ids)} claims") # 1c. Retry Protection: Alice calls 3 times rapidly; should receive same cached bundle, NOT burn 3 keys print(" Subtest 1c: Testing retry protection against OPK depletion...") st, r1 = http_req(f"/api/v1/e2ee/keys/{bob['universal_id']}", "GET", token=alice["token"]) st, r2 = http_req(f"/api/v1/e2ee/keys/{bob['universal_id']}", "GET", token=alice["token"]) st, r3 = http_req(f"/api/v1/e2ee/keys/{bob['universal_id']}", "GET", token=alice["token"]) assert r1["data"][0]["one_time_prekey"]["key_id"] == r2["data"][0]["one_time_prekey"]["key_id"] == r3["data"][0]["one_time_prekey"]["key_id"] print(" ✓ Rapid retries reuse active claim window: 0 extra OPKs burned on retries") # ------------------------------------------------------------------------------ # TEST 2: Fix Verification: Concurrent-Transaction Cursor Race Prevention # ------------------------------------------------------------------------------ print("\n[TEST 2] Verifying Concurrent-Transaction Cursor Race Prevention (Critical Issue C)...") # Multiple senders (Alice and Carol) simultaneously send 10 messages to Bob num_messages = 10 sent_results = [] send_lock = threading.Lock() def send_concurrent_msg(sender_token, index): envelope = { "text": f"Concurrent message #{index}", "nonce": str(uuid.uuid4()), } s, r = http_req("/api/v1/messages", "POST", { "recipient_universal_id": bob["universal_id"], "message_type": "e2ee", "payload": json.dumps(envelope), }, token=sender_token) if s == 200: with send_lock: sent_results.append(r["data"]) tokens_cycle = [alice["token"], carol["token"]] * (num_messages // 2) with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: futures = [executor.submit(send_concurrent_msg, tokens_cycle[i], i) for i in range(num_messages)] concurrent.futures.wait(futures) assert len(sent_results) == num_messages, f"Expected {num_messages} sends, got {len(sent_results)}" # Verify that recipient_seq assigned to Bob's mailbox is strictly monotonic and contiguous recipient_seqs = sorted([m["recipient_seq"] for m in sent_results]) print(f" Assigned recipient_seq sequence: {recipient_seqs}") expected_contiguous = list(range(recipient_seqs[0], recipient_seqs[0] + num_messages)) assert recipient_seqs == expected_contiguous, f"Gaps detected in recipient_seq! Expected {expected_contiguous}, got {recipient_seqs}" print(f" ✓ Row-locked user_mailboxes guaranteed strictly contiguous recipient sequence: {recipient_seqs}") # Verify Bob's cursor sync: Bob syncs incrementally st, sync1 = http_req(f"/api/v1/messages/sync?since_seq={recipient_seqs[0] - 1}&limit=5", "GET", token=bob["token"]) assert st == 200 batch1 = sync1["data"]["messages"] assert len(batch1) == 5 assert sync1["data"]["has_more"] is True cursor1 = sync1["data"]["last_seq"] print(f" ✓ Batch 1: synced 5 messages, cursor safely advanced to {cursor1}") # Bob syncs remainder st, sync2 = http_req(f"/api/v1/messages/sync?since_seq={cursor1}&limit=10", "GET", token=bob["token"]) assert st == 200 batch2 = sync2["data"]["messages"] assert len(batch2) == 5 cursor2 = sync2["data"]["last_seq"] print(f" ✓ Batch 2: synced remaining 5 messages, cursor advanced to {cursor2}") # Verify ALL messages were retrieved with ZERO loss all_synced_ids = {m["id"] for m in batch1 + batch2} all_sent_ids = {m["id"] for m in sent_results} assert all_synced_ids == all_sent_ids, f"Lost messages during cursor progression! Synced: {len(all_synced_ids)}, Sent: {len(all_sent_ids)}" print(f" ✓ 100% of concurrently sent messages retrieved across cursor checkpoints: ZERO loss!") print("\n" + "=" * 70) print("ALL CONCURRENCY & RACE-FREE CURSOR TESTS PASSED (100%)") print("=" * 70)