()
| 4 | import locale |
| 5 | |
| 6 | def rpc_server(): |
| 7 | host = '0.0.0.0' |
| 8 | port = 5657 |
| 9 | |
| 10 | server_socket = socket.socket() |
| 11 | server_socket.bind((host, port)) |
| 12 | |
| 13 | server_socket.listen(1) |
| 14 | conn, address = server_socket.accept() # accept new connection |
| 15 | conn.settimeout(600) # 10 minute timeout |
| 16 | |
| 17 | print("[RPC] Connection from: " + str(address)) |
| 18 | |
| 19 | # First, receive the kernel data base address |
| 20 | try: |
| 21 | data = conn.recv(0x100) |
| 22 | if not data: |
| 23 | return |
| 24 | print("[RPC] Received kernel .data base: 0x" + data.decode('utf-8')) |
| 25 | except socket.timeout: |
| 26 | print("Timeout reached for receiving data (1 min)") |
| 27 | return |
| 28 | |
| 29 | # Now, process cmds |
| 30 | cmd = '' |
| 31 | while True: |
| 32 | print("> ", end = '') |
| 33 | cmd = input() |
| 34 | |
| 35 | cmd_parts = cmd.split(' ') |
| 36 | |
| 37 | if cmd_parts[0] == "r" or cmd_parts[0] == "read": |
| 38 | if len(cmd_parts) < 2: |
| 39 | print("Usage: r [addr]") |
| 40 | continue |
| 41 | |
| 42 | read_addr = cmd_parts[1] |
| 43 | read_addr = read_addr.replace("0x", "") |
| 44 | |
| 45 | if len(read_addr) < 16: |
| 46 | print("Usage: r [addr]") |
| 47 | continue |
| 48 | |
| 49 | read_addr_hi = int(read_addr[:8], 16) |
| 50 | read_addr_low = int(read_addr[8:], 16) |
| 51 | read_addr_bin = struct.pack("<II", read_addr_low, read_addr_hi) |
| 52 | |
| 53 | conn.send(struct.pack("<I", 1)); |
| 54 | conn.send(read_addr_bin) |
| 55 | data = conn.recv(0x14) |
| 56 | if not data: |
| 57 | break |
| 58 | |
| 59 | print(" ".join("{:02x}".format(x) for x in data)) |
| 60 | |
| 61 | if cmd_parts[0] == "w" or cmd_parts[0] == "write": |
| 62 | if len(cmd_parts) < 3: |
| 63 | print("Usage: w [addr] [qword]") |
no test coverage detected