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
|
import struct
import random
import enum
import socket
CURRENT_REQUEST_VERSION = 1
MAX_REQUEST_DATA_SIZE = 32
MAX_PACKET_SIZE = 48
class RequestType(enum.IntEnum):
ReadMemory = 1,
WriteMemory = 2
CITRA_PORT = 45987
class Citra:
def __init__(self, address="127.0.0.1", port=CITRA_PORT):
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.address = address
def is_connected(self):
return self.socket is not None
def _generate_header(self, request_type, data_size):
request_id = random.getrandbits(32)
return (struct.pack("IIII", CURRENT_REQUEST_VERSION, request_id, request_type, data_size), request_id)
def _read_and_validate_header(self, raw_reply, expected_id, expected_type):
reply_version, reply_id, reply_type, reply_data_size = struct.unpack("IIII", raw_reply[:4*4])
if (CURRENT_REQUEST_VERSION == reply_version and
expected_id == reply_id and
expected_type == reply_type and
reply_data_size == len(raw_reply[4*4:])):
return raw_reply[4*4:]
return None
def read_memory(self, read_address, read_size):
"""
>>> c.read_memory(0x100000, 4)
b'\\x07\\x00\\x00\\xeb'
"""
result = bytes()
while read_size > 0:
temp_read_size = min(read_size, MAX_REQUEST_DATA_SIZE)
request_data = struct.pack("II", read_address, temp_read_size)
request, request_id = self._generate_header(RequestType.ReadMemory, len(request_data))
request += request_data
self.socket.sendto(request, (self.address, CITRA_PORT))
raw_reply = self.socket.recv(MAX_PACKET_SIZE)
reply_data = self._read_and_validate_header(raw_reply, request_id, RequestType.ReadMemory)
if reply_data:
result += reply_data
read_size -= len(reply_data)
read_address += len(reply_data)
else:
return None
return result
def write_memory(self, write_address, write_contents):
"""
>>> c.write_memory(0x100000, b"\\xff\\xff\\xff\\xff")
True
>>> c.read_memory(0x100000, 4)
b'\\xff\\xff\\xff\\xff'
>>> c.write_memory(0x100000, b"\\x07\\x00\\x00\\xeb")
True
>>> c.read_memory(0x100000, 4)
b'\\x07\\x00\\x00\\xeb'
"""
write_size = len(write_contents)
while write_size > 0:
temp_write_size = min(write_size, MAX_REQUEST_DATA_SIZE - 8)
request_data = struct.pack("II", write_address, temp_write_size)
request_data += write_contents[:temp_write_size]
request, request_id = self._generate_header(RequestType.WriteMemory, len(request_data))
request += request_data
self.socket.sendto(request, (self.address, CITRA_PORT))
raw_reply = self.socket.recv(MAX_PACKET_SIZE)
reply_data = self._read_and_validate_header(raw_reply, request_id, RequestType.WriteMemory)
if None != reply_data:
write_address += temp_write_size
write_size -= temp_write_size
write_contents = write_contents[temp_write_size:]
else:
return False
return True
if "__main__" == __name__:
import doctest
doctest.testmod(extraglobs={'c': Citra()})
|
