import requests import json import uuid import time import sys BASE_URL = "http://127.0.0.1:8080" def run_tests(): print("==================================================================") print(" UCALL GAP ANALYSIS & IMPLEMENTATION VERIFICATION SUITE ") print("==================================================================") suffix = str(int(time.time()))[-6:] phone1 = f"+91987{suffix}11" phone2 = f"+91987{suffix}22" # --- 1. Register Users --- print("\n[1] Testing User Registration...") reg1 = requests.post(f"{BASE_URL}/api/v1/auth/register", json={ "phone": phone1, "display_name": "Alice Tester" }).json()["data"] u1_id = reg1["user_id"] u1_uid = reg1["universal_id"] token1 = reg1["access_token"] ref_token1 = reg1["refresh_token"] reg2 = requests.post(f"{BASE_URL}/api/v1/auth/register", json={ "phone": phone2, "display_name": "Bob Communications" }).json()["data"] u2_uid = reg2["universal_id"] token2 = reg2["access_token"] print(f" ✅ Alice registered: UID={u1_uid}, phone={phone1}") print(f" ✅ Bob registered: UID={u2_uid}, phone={phone2}") # --- 2. Login by Phone and Universal ID --- print("\n[2] Testing Auth Login (by Phone & Universal ID)...") login_phone = requests.post(f"{BASE_URL}/api/v1/auth/login", json={ "identifier": phone1 }) assert login_phone.status_code == 200, login_phone.text l1 = login_phone.json()["data"] assert l1["universal_id"] == u1_uid print(f" ✅ Logged in by phone ({phone1}) -> UID: {l1['universal_id']}") login_uid = requests.post(f"{BASE_URL}/api/v1/auth/login", json={ "identifier": u1_uid }) assert login_uid.status_code == 200, login_uid.text l2 = login_uid.json()["data"] assert l2["user_id"] == u1_id active_refresh = l2["refresh_token"] print(f" ✅ Logged in by 12-digit UID ({u1_uid}) -> user_id: {l2['user_id']}") # --- 3. Refresh Token Rotation & Reuse Detection --- print("\n[3] Testing Refresh Token Rotation & Reuse Detection...") rot_res = requests.post(f"{BASE_URL}/api/v1/auth/refresh", json={ "refresh_token": active_refresh }) assert rot_res.status_code == 200, rot_res.text rot_data = rot_res.json()["data"] new_refresh = rot_data["refresh_token"] token1 = rot_data["access_token"] print(" ✅ Refresh token rotated successfully to a fresh token") # Attempt to REUSE the old refresh token (Attacker replay simulation) reuse_res = requests.post(f"{BASE_URL}/api/v1/auth/refresh", json={ "refresh_token": active_refresh }) assert reuse_res.status_code == 401, f"Expected 401 for reuse, got {reuse_res.status_code}" print(f" ✅ Token reuse blocked: HTTP 401 Unauthorized (Error: {reuse_res.json().get('error', {}).get('message')})") # Verify that the entire compromised family was revoked compromised_res = requests.post(f"{BASE_URL}/api/v1/auth/refresh", json={ "refresh_token": new_refresh }) assert compromised_res.status_code == 401 print(" ✅ Token family cascade-invalidated after reuse detection") # Fresh login to continue fresh_login = requests.post(f"{BASE_URL}/api/v1/auth/login", json={"identifier": phone1}).json()["data"] token1 = fresh_login["access_token"] ref_to_logout = fresh_login["refresh_token"] # --- 4. User Profile (/users/me) --- print("\n[4] Testing GET & PATCH /api/v1/users/me...") me_res = requests.get(f"{BASE_URL}/api/v1/users/me", headers={"Authorization": f"Bearer {token1}"}) assert me_res.status_code == 200, me_res.text prof = me_res.json()["data"] assert prof["universal_id"] == u1_uid assert prof["display_name"] == "Alice Tester" print(f" ✅ GET /users/me: name='{prof['display_name']}', UID={prof['universal_id']}, phone={prof['phone_number']}") patch_res = requests.patch(f"{BASE_URL}/api/v1/users/me", json={ "display_name": "Alice Senior Architect" }, headers={"Authorization": f"Bearer {token1}"}) assert patch_res.status_code == 200 assert patch_res.json()["data"]["display_name"] == "Alice Senior Architect" print(" ✅ PATCH /users/me: name updated to 'Alice Senior Architect'") # --- 5. Device & Endpoint Management + Deletion --- print("\n[5] Testing Device and Endpoint Registration & Revocation...") hw_id = f"hw-uuid-{suffix}" dev_res = requests.post(f"{BASE_URL}/api/v1/endpoints/device", json={ "universal_id": u1_uid, "hardware_id": hw_id, "platform": "android", "model": "Pixel 9 Pro" }) assert dev_res.status_code == 200, dev_res.text device_id = dev_res.json()["data"]["device"]["id"] print(f" ✅ Device registered: id={device_id}") # Delete device del_dev = requests.delete(f"{BASE_URL}/api/v1/devices/{device_id}", headers={"Authorization": f"Bearer {token1}"}) assert del_dev.status_code == 200, del_dev.text print(f" ✅ DELETE /api/v1/devices/{device_id} succeeded") # --- 6. Contact Management (Sync, List, Delete, Clear) --- print("\n[6] Testing Contact Management Endpoints...") sync_res = requests.post(f"{BASE_URL}/api/v1/contacts/sync", json={ "contacts": [ {"client_contact_id": "c1", "phone_number": phone2, "contact_name": "Bob Contact"}, {"client_contact_id": "c2", "phone_number": "+919000011111", "contact_name": "Charlie Plumber"} ] }, headers={"Authorization": f"Bearer {token1}"}) assert sync_res.status_code == 200 list_res = requests.get(f"{BASE_URL}/api/v1/contacts", headers={"Authorization": f"Bearer {token1}"}) assert list_res.status_code == 200 saved_contacts = list_res.json()["data"] print(f" ✅ GET /api/v1/contacts returned {len(saved_contacts)} contacts:") for sc in saved_contacts: print(f" - {sc['contact_name']} ({sc['contact_phone']}) is_ucall={sc['is_ucall_user']} UID={sc.get('universal_id')}") assert len(saved_contacts) >= 2 # Delete single contact first_cid = saved_contacts[0]["id"] del_c = requests.delete(f"{BASE_URL}/api/v1/contacts/{first_cid}", headers={"Authorization": f"Bearer {token1}"}) assert del_c.status_code == 200 print(f" ✅ DELETE /api/v1/contacts/{first_cid} succeeded") # Clear all contacts (Consent withdrawal) clear_c = requests.delete(f"{BASE_URL}/api/v1/contacts", headers={"Authorization": f"Bearer {token1}"}) assert clear_c.status_code == 200 empty_list = requests.get(f"{BASE_URL}/api/v1/contacts", headers={"Authorization": f"Bearer {token1}"}).json()["data"] assert len(empty_list) == 0 print(" ✅ DELETE /api/v1/contacts cleared all contacts on consent withdrawal") # --- 7. Call State Machine & Transition Hardening --- print("\n[7] Testing Call State Machine Hardening & Event Ledger...") call_res = requests.post(f"{BASE_URL}/api/v1/calls", json={ "recipient_uid": u2_uid, "call_type": "audio" }, headers={"Authorization": f"Bearer {token1}"}) assert call_res.status_code == 200 call_id = call_res.json()["data"]["call_id"] print(f" ✅ Call created: id={call_id}") # Reject call rej_res = requests.post(f"{BASE_URL}/api/v1/calls/{call_id}/reject", json={"reason": "busy"}) assert rej_res.status_code == 200 assert rej_res.json()["data"]["status"] == "rejected" print(f" ✅ Call {call_id} rejected") # Attempt to ACCEPT the rejected call (Invalid transition) fake_ep = str(uuid.uuid4()) bad_accept = requests.post(f"{BASE_URL}/api/v1/calls/{call_id}/accept", json={"endpoint_id": fake_ep}) assert bad_accept.status_code == 409, f"Expected 409 Conflict, got {bad_accept.status_code}" print(" ✅ Invalid transition blocked: HTTP 409 Conflict (cannot accept a rejected/ended call)") # --- 8. Conversations & Messaging --- print("\n[8] Testing Conversations & Messaging Delivery State...") msg_res = requests.post(f"{BASE_URL}/api/v1/messages", json={ "recipient_uid": u2_uid, "payload": "Hi Bob, are you available for a UCall consultation?", "client_message_id": f"msg-req-{suffix}" }, headers={"Authorization": f"Bearer {token1}"}) assert msg_res.status_code == 200, msg_res.text msg_id = msg_res.json()["data"]["id"] print(f" ✅ Message sent from Alice to Bob: msg_id={msg_id}") # Mark delivered deliv_res = requests.post(f"{BASE_URL}/api/v1/messages/{msg_id}/delivered", headers={"Authorization": f"Bearer {token2}"}) assert deliv_res.status_code == 200, deliv_res.text print(f" ✅ POST /api/v1/messages/{msg_id}/delivered marked message delivered") # Get conversations list for Alice conv_res = requests.get(f"{BASE_URL}/api/v1/conversations", headers={"Authorization": f"Bearer {token1}"}) assert conv_res.status_code == 200, conv_res.text convs = conv_res.json()["data"] assert len(convs) >= 1 c_bob = next(c for c in convs if c["peer_universal_id"] == u2_uid) print(f" ✅ GET /api/v1/conversations for Alice: peer_uid={c_bob['peer_universal_id']}, latest='{c_bob['last_message_payload']}'") # Bob marks read read_res = requests.post(f"{BASE_URL}/api/v1/messages/{msg_id}/read", headers={"Authorization": f"Bearer {token2}"}) assert read_res.status_code == 200 print(" ✅ Bob marked message as read") # --- 9. Logout --- print("\n[9] Testing Session Logout...") logout_res = requests.post(f"{BASE_URL}/api/v1/auth/logout", json={ "refresh_token": ref_to_logout }, headers={"Authorization": f"Bearer {token1}"}) assert logout_res.status_code == 200 print(" ✅ Session successfully revoked via POST /api/v1/auth/logout") print("\n==================================================================") print(" ALL GAP ANALYSIS & IMPLEMENTATION TESTS PASSED (100%) ") print("==================================================================") if __name__ == "__main__": run_tests()