#!/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("10.0.0.7") PEER_IP = socket.inet_aton("10.0.0.1") LOGIN_CLIENT_IP = socket.inet_aton("10.0.0.2") 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!" 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 == PEER_IP return ( sender_mac + PEER_MAC + b"\x08\x06" + b"\x00\x01\x08\x00\x06\x04\x00\x02" + PEER_MAC + PEER_IP + sender_mac + sender_ip ) def icmp_request(sequence: int, magic: bytes) -> bytes: payload = b"\0" * 8 + magic + magic icmp = struct.pack("!BBHHH", 8, 0, 0, 0x4748, sequence) + payload icmp = icmp[:2] + struct.pack("!H", checksum(icmp)) + icmp[4:] total_length = 20 + len(icmp) ip = struct.pack( "!BBHHHBBH4s4s", 0x45, 0, total_length, 0x484C, 0, 64, 1, 0, LOGIN_CLIENT_IP, GUEST_IP, ) ip = ip[:10] + struct.pack("!H", checksum(ip)) + ip[12:] return GUEST_MAC + PEER_MAC + b"\x08\x00" + ip + icmp def validate_reply(frame: bytes, 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] == LOGIN_CLIENT_IP assert frame[34] == 0 assert frame[50:66] == magic assert frame[66:82] == NATIVE_ACK_MAGIC assert checksum(frame[34:]) == 0 def main() -> None: global GUEST_IP, PEER_IP, LOGIN_CLIENT_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("--login-only", action="store_true") parser.add_argument("--guest-ip", default="10.0.0.7") parser.add_argument("--peer-ip", default="10.0.0.1") parser.add_argument("--login-client-ip", default="10.0.0.2") args = parser.parse_args() GUEST_IP = socket.inet_aton(args.guest_ip) PEER_IP = socket.inet_aton(args.peer_ip) LOGIN_CLIENT_IP = socket.inet_aton(args.login_client_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_sent = False reply_count = 0 login_reply_count = 0 resident_login_reply_count = 0 recovery_reply_verified = False command_phase = "login" def write_receipt( *, commit_verified: bool, branch_verified: bool, ) -> None: with open(args.receipt, "w", encoding="utf-8") as output: output.write( "arp_gateway_reply: VERIFIED\n" f"icmp_login_request_sent: {str(login_sent).lower()}\n" "icmp_login_reply_verified: true\n" f"icmp_login_reply_count: {login_reply_count}\n" "code_commit_reply_verified: " f"{str(commit_verified).lower()}\n" "branch_move_reply_verified: " f"{str(branch_verified).lower()}\n" f"resident_login_reply_count: {resident_login_reply_count}\n" "recovery_reply_verified: " f"{str(recovery_reply_verified).lower()}\n" "login_magic: HLDP-GHOS-LOGIN!\n" ) def phase_magic() -> bytes: return { "login": LOGIN_MAGIC, "commit": COMMIT_MAGIC, "branch": BRANCH_MAGIC, "resident_login": LOGIN_MAGIC, "recovery": RECOVERY_MAGIC, }[command_phase] while time.monotonic() < deadline: try: frame = peer.recv(4096) except TimeoutError: if arp_verified: peer.sendto( icmp_request(reply_count + 1, phase_magic()), qemu, ) login_sent = True 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": magic = phase_magic() validate_reply(frame, magic) reply_count += 1 if command_phase == "login": login_reply_count += 1 if command_phase == "login" and reply_count < 3: peer.sendto(icmp_request(reply_count + 1, LOGIN_MAGIC), qemu) continue if command_phase == "login": if args.login_only: write_receipt( commit_verified=False, branch_verified=False, ) return command_phase = "commit" peer.sendto(icmp_request(4, COMMIT_MAGIC), qemu) continue if command_phase == "commit": command_phase = "branch" peer.sendto(icmp_request(5, BRANCH_MAGIC), qemu) continue if command_phase == "branch" and args.resident: command_phase = "resident_login" peer.sendto(icmp_request(6, LOGIN_MAGIC), qemu) continue if command_phase == "resident_login": resident_login_reply_count += 1 if resident_login_reply_count < 10: peer.sendto( icmp_request(6 + resident_login_reply_count, LOGIN_MAGIC), qemu, ) continue command_phase = "recovery" peer.sendto(icmp_request(16, RECOVERY_MAGIC), qemu) continue if command_phase == "recovery": recovery_reply_verified = True write_receipt(commit_verified=True, branch_verified=True) return raise SystemExit("timed out waiting for native ICMP login reply") if __name__ == "__main__": main()