#!/usr/bin/env python3 import argparse import socket import struct import time GUEST_MAC = bytes.fromhex("525400267198") PEER_MAC = bytes.fromhex("525400123401") GUEST_IP = socket.inet_aton("172.16.0.6") GATEWAY_IP = socket.inet_aton("172.16.0.1") RELAY_IP = socket.inet_aton("43.153.193.169") LOGIN_MAGIC = b"HLDP-GHOS-LOGIN!" COMMIT_MAGIC = b"HLDP-CODE-COMMIT" BRANCH_MAGIC = b"HLDP-BRANCH-MOVE" RECOVERY_MAGIC = b"HLDP-RECOVER-OS!" NATIVE_ACK_MAGIC = b"HLDP-NATIVE-ACK!" ANCHOR_PORT = 3922 ANCHOR_CLIENT_PORT = 40392 ANCHOR_CLIENT_SEQUENCE = 0x10203040 ANCHOR_BODY = ( b'{"schema":"guanghu.native-public-anchor/v1",' b'"anchor_id":"GLW-PUBLIC-NAV-ANCHOR-001",' b'"entry_path":"LL-CMPN-0001","world_node_id":"SYS-GLW-0001",' b'"repository_id":"REPO-012","branch":"main",' b'"code_entry":"https://guanghulab.com/code/bingshuo/guanghu-ice-heart",' b'"runtime":"GUANGHU_OS_NATIVE"}' ) def checksum(payload: bytes) -> int: if len(payload) % 2: payload += b"\0" words = struct.unpack(f"!{len(payload) // 2}H", payload) total = sum(words) while total >> 16: total = (total & 0xFFFF) + (total >> 16) return (~total) & 0xFFFF def arp_reply(request: bytes) -> bytes: assert request[12:14] == b"\x08\x06" assert request[20:22] == b"\x00\x01" sender_mac = request[22:28] sender_ip = request[28:32] target_ip = request[38:42] assert sender_mac == GUEST_MAC assert sender_ip == GUEST_IP assert target_ip == GATEWAY_IP return ( sender_mac + PEER_MAC + b"\x08\x06" + b"\x00\x01\x08\x00\x06\x04\x00\x02" + PEER_MAC + GATEWAY_IP + sender_mac + sender_ip ) def verified_request(frame: bytes, sequence: int, magic: bytes) -> None: assert frame[0:6] == PEER_MAC assert frame[6:12] == GUEST_MAC assert frame[12:14] == b"\x08\x00" assert frame[26:30] == GUEST_IP assert frame[30:34] == RELAY_IP assert checksum(frame[14:34]) == 0 assert frame[34] == 8 assert frame[35] == 0 assert frame[38:40] == b"\x47\x48" assert int.from_bytes(frame[40:42], "big") == sequence assert frame[50:66] == magic assert frame[66:82] == magic assert checksum(frame[34:82]) == 0 def ordinary_reply(frame: bytes) -> bytes: reply = bytearray(frame) reply[0:6] = GUEST_MAC reply[6:12] = PEER_MAC reply[26:30] = RELAY_IP reply[30:34] = GUEST_IP reply[24:26] = b"\0\0" reply[24:26] = struct.pack("!H", checksum(bytes(reply[14:34]))) reply[34] = 0 reply[36:38] = b"\0\0" reply[36:38] = struct.pack("!H", checksum(bytes(reply[34:82]))) return bytes(reply) def authenticated_reply(frame: bytes, response_capability: bytes = NATIVE_ACK_MAGIC) -> bytes: if len(response_capability) != 16: raise ValueError("response capability must be exactly 16 bytes") reply = bytearray(ordinary_reply(frame)) reply[36:38] = b"\0\0" reply[66:82] = response_capability reply[36:38] = struct.pack("!H", checksum(bytes(reply[34:82]))) return bytes(reply) def tcp_frame( flags: int, sequence: int, acknowledgement: int, payload: bytes = b"", options: bytes = b"", ) -> bytes: if len(options) % 4 or len(options) > 40: raise ValueError("TCP options must be 32-bit aligned and no more than 40 bytes") tcp = bytearray( struct.pack( "!HHIIBBHHH", ANCHOR_CLIENT_PORT, ANCHOR_PORT, sequence, acknowledgement, (5 + len(options) // 4) << 4, flags, 16384, 0, 0, ) + options + payload ) pseudo = GATEWAY_IP + GUEST_IP + b"\0\x06" + struct.pack("!H", len(tcp)) tcp[16:18] = struct.pack("!H", checksum(pseudo + tcp)) ip = bytearray( b"\x45\x00" + struct.pack("!H", 20 + len(tcp)) + b"\x47\x41\x00\x00\x40\x06\x00\x00" + GATEWAY_IP + GUEST_IP ) ip[10:12] = struct.pack("!H", checksum(ip)) return GUEST_MAC + PEER_MAC + b"\x08\x00" + bytes(ip) + bytes(tcp) def verify_anchor_tcp(frame: bytes, expected_flags: int) -> tuple[int, int, bytes]: assert frame[0:6] == PEER_MAC assert frame[6:12] == GUEST_MAC assert frame[12:14] == b"\x08\x00" assert frame[23] == 6 assert frame[26:30] == GUEST_IP assert frame[30:34] == GATEWAY_IP assert checksum(frame[14:34]) == 0 total_length = int.from_bytes(frame[16:18], "big") tcp = frame[34 : 14 + total_length] assert int.from_bytes(tcp[0:2], "big") == ANCHOR_PORT assert int.from_bytes(tcp[2:4], "big") == ANCHOR_CLIENT_PORT assert tcp[13] == expected_flags pseudo = GUEST_IP + GATEWAY_IP + b"\0\x06" + struct.pack("!H", len(tcp)) assert checksum(pseudo + tcp) == 0 return ( int.from_bytes(tcp[4:8], "big"), int.from_bytes(tcp[8:12], "big"), tcp[20:], ) def main() -> None: global GUEST_IP, GATEWAY_IP, RELAY_IP parser = argparse.ArgumentParser() parser.add_argument("--listen-port", type=int, required=True) parser.add_argument("--qemu-port", type=int, required=True) parser.add_argument("--receipt", required=True) parser.add_argument("--resident", action="store_true") parser.add_argument("--final-resident", action="store_true") parser.add_argument("--recovery-token-file") parser.add_argument("--login-only", action="store_true") parser.add_argument("--anchor-http", action="store_true") parser.add_argument("--guest-ip", default="172.16.0.6") parser.add_argument("--peer-ip", default="172.16.0.1") parser.add_argument("--relay-ip", default="43.153.193.169") args = parser.parse_args() if sum((args.resident, args.final_resident, args.login_only)) > 1: raise SystemExit("--resident, --final-resident and --login-only are mutually exclusive") if args.final_resident and not args.recovery_token_file: raise SystemExit("--final-resident requires --recovery-token-file") recovery_capability = None if args.recovery_token_file: token_hex = open(args.recovery_token_file, encoding="ascii").read().strip() recovery_capability = bytes.fromhex(token_hex) if len(recovery_capability) != 16: raise SystemExit("recovery token must be exactly 16 bytes") GUEST_IP = socket.inet_aton(args.guest_ip) GATEWAY_IP = socket.inet_aton(args.peer_ip) RELAY_IP = socket.inet_aton(args.relay_ip) peer = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) peer.bind(("127.0.0.1", args.listen_port)) peer.settimeout(0.2) qemu = ("127.0.0.1", args.qemu_port) deadline = time.monotonic() + 15 arp_verified = False login_count = 0 resident_login_count = 0 commit_verified = False branch_verified = False recovery_verified = False anchor_syn_sent = False anchor_syn_ack_verified = False anchor_server_sequence = None anchor_get_sent = False anchor_http_verified = False while time.monotonic() < deadline: try: frame = peer.recv(4096) except TimeoutError: continue if frame[12:14] == b"\x08\x06": peer.sendto(arp_reply(frame), qemu) arp_verified = True continue if frame[12:14] != b"\x08\x00": continue if frame[23] == 6: if not anchor_syn_ack_verified: server_sequence, acknowledgement, payload = verify_anchor_tcp(frame, 0x12) assert acknowledgement == ANCHOR_CLIENT_SEQUENCE + 1 assert payload == b"" anchor_server_sequence = server_sequence anchor_syn_ack_verified = True request = ( b"GET /v1/anchor HTTP/1.1\r\n" b"Host: native.guanghulab.com\r\n" b"Connection: close\r\n\r\n" ) peer.sendto( tcp_frame( 0x18, ANCHOR_CLIENT_SEQUENCE + 1, server_sequence + 1, request, options=b"\x01\x01\x08\x0a\xd8\x8d\xec\x56\x00\x00\x00\x01", ), qemu, ) anchor_get_sent = True continue server_sequence, acknowledgement, payload = verify_anchor_tcp(frame, 0x19) assert anchor_server_sequence is not None assert server_sequence == anchor_server_sequence + 1 assert acknowledgement > ANCHOR_CLIENT_SEQUENCE + 1 header, body = payload.split(b"\r\n\r\n", 1) assert b"HTTP/1.1 200 OK" in header assert b"Content-Length: 286" in header assert body == ANCHOR_BODY anchor_http_verified = True continue sequence = int.from_bytes(frame[40:42], "big") if sequence <= 3: magic = LOGIN_MAGIC login_count += 1 elif sequence == 4: magic = COMMIT_MAGIC commit_verified = True elif sequence == 5: magic = BRANCH_MAGIC branch_verified = True elif args.final_resident and sequence == 6: magic = LOGIN_MAGIC resident_login_count += 1 elif 6 <= sequence <= 15: magic = LOGIN_MAGIC resident_login_count += 1 elif sequence == 16: magic = RECOVERY_MAGIC recovery_verified = True else: raise AssertionError(f"unexpected native sequence {sequence}") verified_request(frame, sequence, magic) peer.sendto(ordinary_reply(frame), qemu) time.sleep(0.05) response_capability = NATIVE_ACK_MAGIC if ( args.final_resident and resident_login_count >= 3 and (not args.anchor_http or anchor_http_verified) ): response_capability = recovery_capability recovery_verified = True reply = authenticated_reply(frame, response_capability) for repetition in range(4): peer.sendto(reply, qemu) if repetition < 3: time.sleep(0.01) if ( args.anchor_http and args.final_resident and sequence == 6 and not anchor_syn_sent ): peer.sendto( tcp_frame( 0x02, ANCHOR_CLIENT_SEQUENCE, 0, options=( b"\x02\x04\x05\x90\x04\x02\x08\x0a" b"\xd8\x8d\xec\x56\x00\x00\x00\x00\x01\x03\x03\x07" ), ), qemu, ) anchor_syn_sent = True terminal = ( sequence == 3 if args.login_only else recovery_verified and (not args.anchor_http or anchor_http_verified) if args.final_resident else sequence == 16 if args.resident else sequence == 5 ) if not terminal: continue with open(args.receipt, "w", encoding="utf-8") as output: output.write( "arp_gateway_reply: VERIFIED\n" "handshake_direction: NATIVE_INITIATED_OUTBOUND_ICMP\n" "authenticated_relay_reply: VERIFIED\n" "authenticated_relay_reply_burst: 4\n" "ordinary_echo_reply_ignored: true\n" "icmp_login_request_sent: true\n" "icmp_login_reply_verified: true\n" f"icmp_login_reply_count: {login_count}\n" f"code_commit_reply_verified: {str(commit_verified).lower()}\n" f"branch_move_reply_verified: {str(branch_verified).lower()}\n" f"resident_login_reply_count: {resident_login_count}\n" f"recovery_reply_verified: {str(recovery_verified).lower()}\n" f"final_resident_control: {str(args.final_resident).lower()}\n" f"anchor_syn_sent: {str(anchor_syn_sent).lower()}\n" f"anchor_syn_ack_verified: {str(anchor_syn_ack_verified).lower()}\n" f"anchor_get_sent: {str(anchor_get_sent).lower()}\n" f"anchor_http_verified: {str(anchor_http_verified).lower()}\n" "recovery_selected_by_native: false\n" "login_magic: HLDP-GHOS-LOGIN!\n" ) return raise SystemExit("timed out waiting for native outbound ICMP exchange") if __name__ == "__main__": main()