""" End-to-End Integration Test for FreeExile Gateway & Security Pipeline. Tests the complete client-to-server lifecycle: 1. Apple App Attest hardware authentication handshake. 2. Key exchange and AEAD packet encryption. 3. Anti-replay sliding window verification. 4. Touch biometrics human validation on iOS. 5. Authoritative server movement and spatial grid updates. 6. Detection and rejection of tampering, bots, and replay attacks. """ import unittest import asyncio import os import sys import json import struct sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../server"))) from gateway.authoritative_gateway_service import AuthoritativeGatewayService from gateway.network_gateway import PacketCodec from security.app_attest import AppleAppAttestValidator from security.packet_cipher import PacketCipherEngine class TestEndToEndGateway(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.host = "127.0.0.1" self.port = 19988 self.service = AuthoritativeGatewayService(host=self.host, port=self.port) await self.service.start() async def asyncTearDown(self): await self.service.stop() async def test_end_to_end_genuine_iphone_client(self): """Genuine iPhone connects, attests hardware, sends human touch movement, receives snapshot.""" reader, writer = await asyncio.open_connection(self.host, self.port) # 1. Client Handshake with Apple App Attest validator = AppleAppAttestValidator() token = validator.generate_mock_attestation( device_id="iphone_15_pro_001", bundle_id="com.freeexile.game.ios", counter=1 ) handshake_payload = json.dumps({ "type": "handshake", "account_id": "player_hero_01", "device_id": token.device_id, "bundle_id": token.bundle_id, "public_key_hash": token.public_key_hash, "signature": token.signature, "counter": token.counter, "timestamp_ms": token.timestamp_ms }).encode("utf-8") writer.write(PacketCodec.encode_frame(handshake_payload)) await writer.drain() # Read handshake response header = await reader.readexactly(4) (length,) = struct.unpack(">I", header) resp_payload = await reader.readexactly(length) resp = json.loads(resp_payload.decode("utf-8")) self.assertEqual(resp.get("status"), "ok") self.assertIn("session_id", resp) self.assertIn("shared_key", resp) entity_id = resp["entity_id"] shared_key = bytes.fromhex(resp["shared_key"]) # 2. Client prepares AEAD Cipher client_cipher = PacketCipherEngine(shared_key) # 3. Client sends movement packet with genuine human touch biometrics seq_num = 1 touch_points = [ {"x": 100.0, "y": 200.0, "timestamp_ms": 1000.0, "major_radius": 18.5, "force": 1.0}, {"x": 104.2, "y": 202.1, "timestamp_ms": 1016.0, "major_radius": 19.0, "force": 1.02}, {"x": 109.8, "y": 205.7, "timestamp_ms": 1033.0, "major_radius": 18.2, "force": 0.98}, {"x": 116.5, "y": 210.4, "timestamp_ms": 1049.0, "major_radius": 17.8, "force": 1.05}, {"x": 124.0, "y": 216.0, "timestamp_ms": 1066.0, "major_radius": 18.6, "force": 1.01}, ] move_msg = json.dumps({ "type": "move", "dir_x": 1.0, "dir_y": 0.0, "dt": 0.1, "touch_points": touch_points }).encode("utf-8") ciphertext, tag, nonce = client_cipher.encrypt(move_msg, seq_num) packet_body = struct.pack(">Q", seq_num) + nonce + tag + ciphertext writer.write(PacketCodec.encode_frame(packet_body)) await writer.drain() # 4. Server receives, validates, and sends back encrypted snapshot header = await reader.readexactly(4) (length,) = struct.unpack(">I", header) resp_packet_body = await reader.readexactly(length) resp_seq = struct.unpack(">Q", resp_packet_body[:8])[0] resp_nonce = resp_packet_body[8:20] resp_tag = resp_packet_body[20:36] resp_ciphertext = resp_packet_body[36:] decrypted_snapshot_bytes = client_cipher.decrypt(resp_ciphertext, resp_tag, resp_nonce, resp_seq) snapshot = json.loads(decrypted_snapshot_bytes.decode("utf-8")) self.assertEqual(snapshot.get("type"), "snapshot") self.assertEqual(snapshot.get("ack_seq"), 1) self.assertEqual(snapshot.get("entity_id"), entity_id) self.assertGreater(snapshot.get("x"), 0.0) writer.close() await writer.wait_closed() async def test_end_to_end_rejects_cracked_bundle(self): """Cracked IPA bundle ID is rejected during handshake.""" reader, writer = await asyncio.open_connection(self.host, self.port) handshake_payload = json.dumps({ "type": "handshake", "account_id": "player_hacker", "device_id": "fake_device", "bundle_id": "com.cracked.freeexile", "public_key_hash": "deadbeef", "signature": "bad_sig", "counter": 1, "timestamp_ms": 1000 }).encode("utf-8") writer.write(PacketCodec.encode_frame(handshake_payload)) await writer.drain() header = await reader.readexactly(4) (length,) = struct.unpack(">I", header) resp_payload = await reader.readexactly(length) resp = json.loads(resp_payload.decode("utf-8")) self.assertEqual(resp.get("status"), "error") self.assertIn("tampering", resp.get("message", "").lower()) writer.close() await writer.wait_closed() async def test_end_to_end_rejects_bot_straight_line_touch(self): """Scripted bot moving with synthetic contact or perfect straight line is rejected.""" reader, writer = await asyncio.open_connection(self.host, self.port) # Valid Handshake validator = AppleAppAttestValidator() token = validator.generate_mock_attestation( device_id="iphone_bot_target", bundle_id="com.freeexile.game.ios", counter=1 ) handshake_payload = json.dumps({ "type": "handshake", "account_id": "bot_acc", "device_id": token.device_id, "bundle_id": token.bundle_id, "public_key_hash": token.public_key_hash, "signature": token.signature, "counter": token.counter, "timestamp_ms": token.timestamp_ms }).encode("utf-8") writer.write(PacketCodec.encode_frame(handshake_payload)) await writer.drain() header = await reader.readexactly(4) (length,) = struct.unpack(">I", header) resp = json.loads((await reader.readexactly(length)).decode("utf-8")) shared_key = bytes.fromhex(resp["shared_key"]) client_cipher = PacketCipherEngine(shared_key) # Bot touch: Exact straight line / constant radius synthetic pattern bot_touch_points = [ {"x": 10.0, "y": 10.0, "timestamp_ms": 1000.0, "major_radius": 15.0, "force": 1.0}, {"x": 20.0, "y": 20.0, "timestamp_ms": 1010.0, "major_radius": 15.0, "force": 1.0}, {"x": 30.0, "y": 30.0, "timestamp_ms": 1020.0, "major_radius": 15.0, "force": 1.0}, {"x": 40.0, "y": 40.0, "timestamp_ms": 1030.0, "major_radius": 15.0, "force": 1.0}, {"x": 50.0, "y": 50.0, "timestamp_ms": 1040.0, "major_radius": 15.0, "force": 1.0}, ] move_msg = json.dumps({ "type": "move", "dir_x": 1.0, "dir_y": 0.0, "dt": 0.1, "touch_points": bot_touch_points }).encode("utf-8") seq_num = 1 ciphertext, tag, nonce = client_cipher.encrypt(move_msg, seq_num) packet_body = struct.pack(">Q", seq_num) + nonce + tag + ciphertext writer.write(PacketCodec.encode_frame(packet_body)) await writer.drain() # Server response should indicate anomaly rejection header = await reader.readexactly(4) (length,) = struct.unpack(">I", header) resp_packet_body = await reader.readexactly(length) resp_seq = struct.unpack(">Q", resp_packet_body[:8])[0] resp_nonce = resp_packet_body[8:20] resp_tag = resp_packet_body[20:36] resp_ciphertext = resp_packet_body[36:] decrypted_resp = json.loads(client_cipher.decrypt(resp_ciphertext, resp_tag, resp_nonce, resp_seq).decode("utf-8")) self.assertEqual(decrypted_resp.get("status"), "rejected") reason_lower = decrypted_resp.get("reason", "").lower() self.assertTrue( any(w in reason_lower for w in ["synthetic", "bot", "script", "emulator", "detected"]), f"Expected bot anomaly detection reason, got: {reason_lower}" ) writer.close() await writer.wait_closed() if __name__ == "__main__": unittest.main()