-
Notifications
You must be signed in to change notification settings - Fork 92
/
rpcserver.py
102 lines (77 loc) · 2.91 KB
/
rpcserver.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
import socket
import time
import struct
import locale
def rpc_server():
host = '0.0.0.0'
port = 5657
server_socket = socket.socket()
server_socket.bind((host, port))
server_socket.listen(1)
conn, address = server_socket.accept() # accept new connection
conn.settimeout(600) # 10 minute timeout
print("[RPC] Connection from: " + str(address))
# First, receive the kernel data base address
try:
data = conn.recv(0x100)
if not data:
return
print("[RPC] Received kernel .data base: 0x" + data.decode('utf-8'))
except socket.timeout:
print("Timeout reached for receiving data (1 min)")
return
# Now, process cmds
cmd = ''
while True:
print("> ", end = '')
cmd = input()
cmd_parts = cmd.split(' ')
if cmd_parts[0] == "r" or cmd_parts[0] == "read":
if len(cmd_parts) < 2:
print("Usage: r [addr]")
continue
read_addr = cmd_parts[1]
read_addr = read_addr.replace("0x", "")
if len(read_addr) < 16:
print("Usage: r [addr]")
continue
read_addr_hi = int(read_addr[:8], 16)
read_addr_low = int(read_addr[8:], 16)
read_addr_bin = struct.pack("<II", read_addr_low, read_addr_hi)
conn.send(struct.pack("<I", 1));
conn.send(read_addr_bin)
data = conn.recv(0x14)
if not data:
break
print(" ".join("{:02x}".format(x) for x in data))
if cmd_parts[0] == "w" or cmd_parts[0] == "write":
if len(cmd_parts) < 3:
print("Usage: w [addr] [qword]")
continue
write_addr = cmd_parts[1]
write_addr = write_addr.replace("0x", "")
if len(write_addr) < 16:
print("Usage: w [addr] [qword]")
continue
write_addr_hi = int(read_addr[:8], 16)
write_addr_low = int(read_addr[8:], 16)
write_addr_bin = struct.pack("<II", write_addr_low, write_addr_hi)
write_val = cmd_parts[2]
write_val = write_val.replace("0x", "")
write_val_hi = int(0)
write_val_low = int(0)
if len(write_val) > 8:
write_val_hi = int(write_val[:8], 16)
write_val_low = int(write_val[8:], 16)
else:
write_val_low = int(write_val, 16)
write_val_bin = struct.pack("<II", write_val_low, write_val_hi)
conn.send(struct.pack("<I", 2));
conn.send(write_addr_bin)
conn.send(write_val_bin)
print("Wrote qword.")
if cmd_parts[0] == "exit":
break
conn.close()
if __name__ == '__main__':
rpc_server()