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()})