| from abc import ABC, abstractmethod | |
| import ctypes | |
| import errno | |
| import io | |
| import re | |
| import threading | |
| import socket | |
| import traceback | |
| from enum import Enum | |
| from lldbsuite.support import seven | |
| from typing import Optional, List, Tuple, Union, Sequence | |
| def checksum(message): | |
| """ | |
| Calculate the GDB server protocol checksum of the message. | |
| The GDB server protocol uses a simple modulo 256 sum. | |
| """ | |
| check = 0 | |
| for c in message: | |
| check += ord(c) | |
| return check % 256 | |
| def frame_packet(message, prefix): | |
| """ | |
| Create a framed packet that's ready to send over the GDB connection | |
| channel. | |
| Framing means: | |
| * Attaching the prefix. Which is usually '$' but for notifications will be | |
| '%'. | |
| * Appending a '#'. | |
| * Adding a two character hex checksum. | |
| """ | |
| return "%s%s#%02x" % (prefix, message, checksum(message)) | |
| def escape_binary(message): | |
| """ | |
| Escape the binary message using the process described in the GDB server | |
| protocol documentation. | |
| Most bytes are sent through as-is, but $, #, and { are escaped by writing | |
| a { followed by the original byte mod 0x20. | |
| """ | |
| out = "" | |
| for c in message: | |
| d = ord(c) | |
| if d in (0x23, 0x24, 0x7D): | |
| out += chr(0x7D) | |
| out += chr(d ^ 0x20) | |
| else: | |
| out += c | |
| return out | |
| def hex_encode_bytes(message): | |
| """ | |
| Encode the binary message by converting each byte into a two-character | |
| hex string. | |
| """ | |
| out = "" | |
| for c in message: | |
| out += "%02x" % ord(c) | |
| return out | |
| def hex_decode_bytes(hex_bytes): | |
| """ | |
| Decode the hex string into a binary message by converting each two-character | |
| hex string into a single output byte. | |
| """ | |
| out = "" | |
| hex_len = len(hex_bytes) | |
| i = 0 | |
| while i < hex_len - 1: | |
| out += chr(int(hex_bytes[i : i + 2], 16)) | |
| i += 2 | |
| return out | |
| def parse_memory_read_packet(packet): | |
| """ | |
| Parse a memory-read packet ("m<addr>,<len>" or "x<addr>,<len>") into its | |
| (addr, length) integers, or return None if it isn't a memory read. | |
| """ | |
| if not packet or packet[0] not in ("m", "x"): | |
| return None | |
| try: | |
| addr, length = [int(x, 16) for x in packet[1:].split(",")] | |
| except ValueError: | |
| return None | |
| return addr, length | |
| def parse_memory_read_ranges(packet: str) -> List[Tuple[int, int]]: | |
| """ | |
| Parse every (addr, length) a memory-read packet asks the stub for, and return | |
| an empty list if the packet isn't a memory read. Unlike | |
| parse_memory_read_packet this also covers "MultiMemRead", which carries | |
| several ranges in one packet. | |
| """ | |
| single = parse_memory_read_packet(packet) | |
| if single is not None: | |
| return [single] | |
| prefix = "MultiMemRead:ranges:" | |
| if not packet or not packet.startswith(prefix): | |
| return [] | |
| body = packet[len(prefix) :] | |
| end = body.find(";") | |
| if end < 0: | |
| return [] | |
| try: | |
| numbers = [int(n, 16) for n in body[:end].split(",")] | |
| except ValueError: | |
| return [] | |
| if len(numbers) % 2: | |
| return [] | |
| return list(zip(numbers[0::2], numbers[1::2])) | |
| class PacketDirection(Enum): | |
| RECV = "recv" | |
| SEND = "send" | |
| # A line in a "log enable gdb-remote packets" file, e.g. | |
| # < 22> send packet: $x100000,40#ad | |
| # < 6> read packet: $OK#9a | |
| # The leading "<...>" size column is optional; the body is the framed packet | |
| # between "$" and the "#" checksum. | |
| _PACKET_LOG_RE = re.compile( | |
| r"(?P<direction>send|read) packet: \$(?P<body>.*)#[0-9a-fA-F]{2}" | |
| ) | |
| def parse_packet_log(lines): | |
| """ | |
| Parse the lines of a "log enable gdb-remote packets" file into a list of | |
| (PacketDirection, body) tuples, where body is the packet contents. | |
| "send packet:" lines (client -> stub) map to PacketDirection.SEND and | |
| "read packet:" lines (stub -> client) map to PacketDirection.RECV, matching | |
| the perspective of the client whose log this is. | |
| """ | |
| out = [] | |
| for line in lines: | |
| m = _PACKET_LOG_RE.search(line) | |
| if not m: | |
| continue | |
| direction = ( | |
| PacketDirection.SEND | |
| if m.group("direction") == "send" | |
| else PacketDirection.RECV | |
| ) | |
| out.append((direction, m.group("body"))) | |
| return out | |
| class PacketLog: | |
| def __init__(self): | |
| self._packets: list[tuple[PacketDirection, str]] = [] | |
| def add_sent(self, packet: str): | |
| self._packets.append((PacketDirection.SEND, packet)) | |
| def add_received(self, packet: str): | |
| self._packets.append((PacketDirection.RECV, packet)) | |
| def get_sent(self): | |
| return [ | |
| pkt for direction, pkt in self._packets if direction == PacketDirection.SEND | |
| ] | |
| def get_received(self): | |
| return [ | |
| pkt for direction, pkt in self._packets if direction == PacketDirection.RECV | |
| ] | |
| def __iter__(self): | |
| return iter(self._packets) | |
| class MockGDBServerResponder: | |
| """ | |
| A base class for handling client packets and issuing server responses for | |
| GDB tests. | |
| This handles many typical situations, while still allowing subclasses to | |
| completely customize their responses. | |
| Most subclasses will be interested in overriding the other() method, which | |
| handles any packet not recognized in the common packet handling code. | |
| """ | |
| registerCount: int = 40 | |
| class SpecialResponse(Enum): | |
| RESPONSE_DISCONNECT = 0 | |
| RESPONSE_NONE = 1 | |
| RESPONSE_DISCONNECT = SpecialResponse.RESPONSE_DISCONNECT | |
| RESPONSE_NONE = SpecialResponse.RESPONSE_NONE | |
| Response = Union[str, SpecialResponse] | |
| def __init__(self): | |
| self.packetLog = PacketLog() | |
| def respond(self, packet: str) -> Sequence[Response]: | |
| """ | |
| Return the unframed packet data that the server should issue in response | |
| to the given packet received from the client. | |
| """ | |
| self.packetLog.add_received(packet) | |
| response = self._respond_impl(packet) | |
| if not isinstance(response, list): | |
| response = [response] | |
| for part in response: | |
| if isinstance(part, self.SpecialResponse): | |
| continue | |
| self.packetLog.add_sent(part) | |
| return response | |
| def _respond_impl(self, packet) -> Union[Response, List[Response]]: | |
| if packet is MockGDBServer.PACKET_INTERRUPT: | |
| return self.interrupt() | |
| if packet == "c": | |
| return self.cont() | |
| if packet.startswith("vCont;c"): | |
| return self.vCont(packet) | |
| if packet[0] == "A": | |
| return self.A(packet) | |
| if packet[0] == "D": | |
| return self.D(packet) | |
| if packet[0] == "g": | |
| return self.readRegisters() | |
| if packet[0] == "G": | |
| # Gxxxxxxxxxxx | |
| # Gxxxxxxxxxxx;thread:1234; | |
| return self.writeRegisters(packet[1:].split(";")[0]) | |
| if packet[0] == "p": | |
| regnum = packet[1:].split(";")[0] | |
| return self.readRegister(int(regnum, 16)) | |
| if packet[0] == "P": | |
| register, value = packet[1:].split("=") | |
| return self.writeRegister(int(register, 16), value) | |
| if packet[0] == "m": | |
| addr, length = parse_memory_read_packet(packet) | |
| return self.readMemory(addr, length) | |
| if packet[0] == "x": | |
| addr, length = parse_memory_read_packet(packet) | |
| return self.x(addr, length) | |
| if packet[0] == "M": | |
| location, encoded_data = packet[1:].split(":") | |
| addr, length = [int(x, 16) for x in location.split(",")] | |
| return self.writeMemory(addr, encoded_data) | |
| if packet[0:7] == "qSymbol": | |
| return self.qSymbol(packet[8:]) | |
| if packet[0:10] == "qSupported": | |
| return self.qSupported(packet[11:].split(";")) | |
| if packet == "qfThreadInfo": | |
| return self.qfThreadInfo() | |
| if packet == "qsThreadInfo": | |
| return self.qsThreadInfo() | |
| if packet == "qC": | |
| return self.qC() | |
| if packet == "QEnableErrorStrings": | |
| return self.QEnableErrorStrings() | |
| if packet == "?": | |
| return self.haltReason() | |
| if packet == "s": | |
| return self.haltReason() | |
| if packet[0] == "H": | |
| tid = packet[2:] | |
| if "." in tid: | |
| assert tid.startswith("p") | |
| # TODO: do we want to do anything with PID? | |
| tid = tid.split(".", 1)[1] | |
| return self.selectThread(packet[1], int(tid, 16)) | |
| if packet[0:6] == "qXfer:": | |
| obj, read, annex, location = packet[6:].split(":") | |
| offset, length = [int(x, 16) for x in location.split(",")] | |
| data, has_more = self.qXferRead(obj, annex, offset, length) | |
| if data is not None: | |
| return self._qXferResponse(data, has_more) | |
| return "" | |
| if packet.startswith("vAttach;"): | |
| pid = packet.partition(";")[2] | |
| return self.vAttach(int(pid, 16)) | |
| if packet[0] == "Z": | |
| return self.setBreakpoint(packet) | |
| if packet[0] == "z": | |
| return self.clearBreakpoint(packet) | |
| if packet.startswith("qThreadStopInfo"): | |
| threadnum = int(packet[15:], 16) | |
| return self.threadStopInfo(threadnum) | |
| if packet == "QThreadSuffixSupported": | |
| return self.QThreadSuffixSupported() | |
| if packet == "QListThreadsInStopReply": | |
| return self.QListThreadsInStopReply() | |
| if packet == "jThreadsInfo": | |
| return self.jThreadsInfo() | |
| if packet.startswith("qMemoryRegionInfo:"): | |
| return self.qMemoryRegionInfo(int(packet.split(":")[1], 16)) | |
| if packet == "qQueryGDBServer": | |
| return self.qQueryGDBServer() | |
| if packet == "qHostInfo": | |
| return self.qHostInfo() | |
| if packet.startswith("qEcho"): | |
| return self.qEcho(int(packet.split(":")[1])) | |
| if packet == "qGetWorkingDir": | |
| return self.qGetWorkingDir() | |
| if packet == "qOffsets": | |
| return self.qOffsets() | |
| if packet == "qProcessInfo": | |
| return self.qProcessInfo() | |
| if packet == "qsProcessInfo": | |
| return self.qsProcessInfo() | |
| if packet.startswith("qfProcessInfo"): | |
| return self.qfProcessInfo(packet) | |
| if packet.startswith("jGetLoadedDynamicLibrariesInfos"): | |
| return self.jGetLoadedDynamicLibrariesInfos(packet) | |
| if packet.startswith("qPathComplete:"): | |
| return self.qPathComplete() | |
| if packet.startswith("vFile:"): | |
| return self.vFile(packet) | |
| if packet.startswith("vRun;"): | |
| return self.vRun(packet) | |
| if packet.startswith("qLaunchGDBServer;"): | |
| _, host = packet.partition(";")[2].split(":") | |
| return self.qLaunchGDBServer(host) | |
| if packet.startswith("qLaunchSuccess"): | |
| return self.qLaunchSuccess() | |
| if packet.startswith("QEnvironment:"): | |
| return self.QEnvironment(packet) | |
| if packet.startswith("QEnvironmentHexEncoded:"): | |
| return self.QEnvironmentHexEncoded(packet) | |
| if packet.startswith("qRegisterInfo"): | |
| regnum = int(packet[len("qRegisterInfo") :], 16) | |
| return self.qRegisterInfo(regnum) | |
| if packet == "k": | |
| return self.k() | |
| return self.other(packet) | |
| def qsProcessInfo(self) -> str: | |
| return "E04" | |
| def qfProcessInfo(self, packet) -> str: | |
| return "E04" | |
| def jGetLoadedDynamicLibrariesInfos(self, packet) -> str: | |
| return "" | |
| def qGetWorkingDir(self) -> str: | |
| return "2f" | |
| def qOffsets(self) -> str: | |
| return "" | |
| def jThreadsInfo(self) -> str: | |
| return "" | |
| def qProcessInfo(self) -> str: | |
| return "" | |
| def qHostInfo(self) -> str: | |
| return "ptrsize:8;endian:little;" | |
| def qEcho(self, num: int) -> str: | |
| return "E04" | |
| def qQueryGDBServer(self) -> str: | |
| return "E04" | |
| def interrupt(self): | |
| raise self.UnexpectedPacketException() | |
| def cont(self): | |
| raise self.UnexpectedPacketException() | |
| def vCont(self, packet): | |
| raise self.UnexpectedPacketException() | |
| def A(self, packet) -> str: | |
| return "" | |
| def D(self, packet) -> str: | |
| return "OK" | |
| def readRegisters(self) -> str: | |
| return "00000000" * self.registerCount | |
| def readRegister(self, register: int) -> str: | |
| return "00000000" | |
| def writeRegisters(self, registers_hex) -> str: | |
| return "OK" | |
| def writeRegister(self, register, value_hex) -> str: | |
| return "OK" | |
| def readMemory(self, addr, length) -> str: | |
| return "00" * length | |
| def x(self, addr, length) -> str: | |
| return "" | |
| def writeMemory(self, addr, data_hex) -> str: | |
| return "OK" | |
| def qSymbol(self, symbol_args) -> str: | |
| return "OK" | |
| def qSupported(self, client_supported) -> str: | |
| return "qXfer:features:read+;PacketSize=3fff;QStartNoAckMode+" | |
| def qfThreadInfo(self) -> str: | |
| return "l" | |
| def qsThreadInfo(self) -> str: | |
| return "l" | |
| def qC(self) -> str: | |
| return "QC0" | |
| def QEnableErrorStrings(self) -> str: | |
| return "OK" | |
| def haltReason(self) -> str: | |
| # SIGINT is 2, return type is 2 digit hex string | |
| return "S02" | |
| def qXferRead( | |
| self, obj: str, annex: str, offset: int, length: int | |
| ) -> Tuple[Optional[str], bool]: | |
| return None, False | |
| def _qXferResponse(self, data, has_more): | |
| return "%s%s" % ("m" if has_more else "l", escape_binary(data)) | |
| def vAttach(self, pid): | |
| raise self.UnexpectedPacketException() | |
| def selectThread(self, op, thread_id) -> str: | |
| return "OK" | |
| def setBreakpoint(self, packet) -> str: | |
| raise self.UnexpectedPacketException() | |
| def clearBreakpoint(self, packet) -> str: | |
| raise self.UnexpectedPacketException() | |
| def threadStopInfo(self, threadnum) -> str: | |
| return "" | |
| def other(self, packet) -> str: | |
| # empty string means unsupported | |
| return "" | |
| def QThreadSuffixSupported(self) -> str: | |
| return "" | |
| def QListThreadsInStopReply(self) -> str: | |
| return "" | |
| def qMemoryRegionInfo(self, addr) -> str: | |
| return "" | |
| def qPathComplete(self) -> str: | |
| return "" | |
| def vFile(self, packet) -> str: | |
| return "" | |
| def vRun(self, packet) -> str: | |
| return "" | |
| def qLaunchGDBServer(self, host): | |
| raise self.UnexpectedPacketException() | |
| def qLaunchSuccess(self) -> str: | |
| return "" | |
| def QEnvironment(self, packet) -> str: | |
| return "OK" | |
| def QEnvironmentHexEncoded(self, packet) -> str: | |
| return "OK" | |
| def qRegisterInfo(self, num) -> str: | |
| return "" | |
| def k(self): | |
| return ["W01", self.RESPONSE_DISCONNECT] | |
| """ | |
| Raised when we receive a packet for which there is no default action. | |
| Override the responder class to implement behavior suitable for the test at | |
| hand. | |
| """ | |
| class UnexpectedPacketException(Exception): | |
| pass | |
| class MockGDBServerXMLResponder(MockGDBServerResponder): | |
| def __init__(self, docs, register_data): | |
| super().__init__() | |
| self.docs = docs | |
| self.register_data = register_data | |
| def qXferRead(self, obj, annex, offset, length): | |
| try: | |
| return self.docs[annex], False | |
| except KeyError: | |
| return None, False | |
| def readRegister(self, regnum): | |
| return "E01" | |
| def readRegisters(self): | |
| return self.register_data | |
| class ServerChannel(ABC): | |
| """ | |
| A wrapper class for TCP or pty-based server. | |
| """ | |
| @abstractmethod | |
| def get_connect_address(self) -> str: | |
| """Get address for the client to connect to.""" | |
| @abstractmethod | |
| def get_connect_url(self) -> str: | |
| """Get URL suitable for process connect command.""" | |
| def close_server(self): | |
| """Close all resources used by the server.""" | |
| def accept(self): | |
| """Accept a single client connection to the server.""" | |
| def close_connection(self): | |
| """Close all resources used by the accepted connection.""" | |
| @abstractmethod | |
| def recv(self) -> bytes: | |
| """Receive a data packet from the connected client.""" | |
| @abstractmethod | |
| def sendall(self, data: bytes) -> None: | |
| """Send the data to the connected client.""" | |
| class ServerSocket(ServerChannel): | |
| def __init__(self, family, type, proto, addr): | |
| self._server_socket = socket.socket(family, type, proto) | |
| self._connection = None | |
| self._server_socket.bind(addr) | |
| self._server_socket.listen(1) | |
| def close_server(self): | |
| self._server_socket.close() | |
| def accept(self): | |
| assert self._connection is None | |
| # accept() is stubborn and won't fail even when the socket is | |
| # shutdown, so we'll use a timeout | |
| self._server_socket.settimeout(30.0) | |
| client, client_addr = self._server_socket.accept() | |
| # The connected client inherits its timeout from self._socket, | |
| # but we'll use a blocking socket for the client | |
| client.settimeout(None) | |
| self._connection = client | |
| def close_connection(self): | |
| assert self._connection is not None | |
| self._connection.close() | |
| self._connection = None | |
| def recv(self) -> bytes: | |
| assert self._connection is not None | |
| return self._connection.recv(4096) | |
| def sendall(self, data: bytes) -> None: | |
| assert self._connection is not None | |
| return self._connection.sendall(data) | |
| class TCPServerSocket(ServerSocket): | |
| def __init__(self): | |
| family, type, proto, _, addr = socket.getaddrinfo( | |
| "localhost", 0, proto=socket.IPPROTO_TCP | |
| )[0] | |
| super().__init__(family, type, proto, addr) | |
| def get_connect_address(self) -> str: | |
| return "[{}]:{}".format(*self._server_socket.getsockname()) | |
| def get_connect_url(self) -> str: | |
| return "connect://" + self.get_connect_address() | |
| class UnixServerSocket(ServerSocket): | |
| def __init__(self, addr): | |
| super().__init__(socket.AF_UNIX, socket.SOCK_STREAM, 0, addr) | |
| def get_connect_address(self) -> str: | |
| return self._server_socket.getsockname() | |
| def get_connect_url(self) -> str: | |
| return "unix-connect://" + self.get_connect_address() | |
| class PtyServerSocket(ServerChannel): | |
| def __init__(self): | |
| import pty | |
| import tty | |
| primary, secondary = pty.openpty() | |
| tty.setraw(primary) | |
| self._primary = io.FileIO(primary, "r+b") | |
| self._secondary = io.FileIO(secondary, "r+b") | |
| def get_connect_address(self) -> str: | |
| libc = ctypes.CDLL(None) | |
| libc.ptsname.argtypes = (ctypes.c_int,) | |
| libc.ptsname.restype = ctypes.c_char_p | |
| return libc.ptsname(self._primary.fileno()).decode() | |
| def get_connect_url(self): | |
| return "serial://" + self.get_connect_address() | |
| def close_server(self): | |
| self._secondary.close() | |
| self._primary.close() | |
| def recv(self) -> bytes: | |
| try: | |
| return self._primary.read(4096) | |
| except OSError as e: | |
| # closing the pty results in EIO on Linux, convert it to EOF | |
| if e.errno == errno.EIO: | |
| return b"" | |
| raise | |
| def sendall(self, data: bytes) -> None: | |
| self._primary.write(data) | |
| class MockGDBServer: | |
| """ | |
| A simple TCP-based GDB server that can test client behavior by receiving | |
| commands and issuing custom-tailored responses. | |
| Responses are generated via the .responder property, which should be an | |
| instance of a class based on MockGDBServerResponder. | |
| """ | |
| responder = None | |
| _socket = None | |
| _thread = None | |
| _receivedData = None | |
| _receivedDataOffset = None | |
| _shouldSendAck = True | |
| def __init__(self, socket): | |
| self._socket = socket | |
| self.responder = MockGDBServerResponder() | |
| def start(self): | |
| # Start a thread that waits for a client connection. | |
| self._thread = threading.Thread(target=self.run) | |
| self._thread.start() | |
| def stop(self): | |
| if self._thread is not None: | |
| self._thread.join() | |
| self._thread = None | |
| def get_connect_address(self) -> str: | |
| assert self._socket is not None | |
| return self._socket.get_connect_address() | |
| def get_connect_url(self) -> str: | |
| assert self._socket is not None | |
| return self._socket.get_connect_url() | |
| def run(self): | |
| assert self._socket is not None | |
| # For testing purposes, we only need to worry about one client | |
| # connecting just one time. | |
| try: | |
| self._socket.accept() | |
| except Exception: | |
| traceback.print_exc() | |
| return | |
| self._shouldSendAck = True | |
| self._receivedData = "" | |
| self._receivedDataOffset = 0 | |
| data = None | |
| try: | |
| while True: | |
| data = seven.bitcast_to_string(self._socket.recv()) | |
| if data is None or len(data) == 0: | |
| break | |
| self._receive(data) | |
| except self.TerminateConnectionException: | |
| pass | |
| except Exception: | |
| print( | |
| "An exception happened when receiving the response from the gdb server. Closing the client..." | |
| ) | |
| traceback.print_exc() | |
| finally: | |
| self._socket.close_connection() | |
| self._socket.close_server() | |
| def _receive(self, data): | |
| """ | |
| Collects data, parses and responds to as many packets as exist. | |
| Any leftover data is kept for parsing the next time around. | |
| """ | |
| self._receivedData += data | |
| packet = self._parsePacket() | |
| while packet is not None: | |
| self._handlePacket(packet) | |
| packet = self._parsePacket() | |
| def _parsePacket(self): | |
| """ | |
| Reads bytes from self._receivedData, returning: | |
| - a packet's contents if a valid packet is found | |
| - the PACKET_ACK unique object if we got an ack | |
| - None if we only have a partial packet | |
| Raises an InvalidPacketException if unexpected data is received | |
| or if checksums fail. | |
| Once a complete packet is found at the front of self._receivedData, | |
| its data is removed form self._receivedData. | |
| """ | |
| assert self._receivedData is not None | |
| data = self._receivedData | |
| assert self._receivedDataOffset is not None | |
| i = self._receivedDataOffset | |
| data_len = len(data) | |
| if data_len == 0: | |
| return None | |
| if i == 0: | |
| # If we're looking at the start of the received data, that means | |
| # we're looking for the start of a new packet, denoted by a $. | |
| # It's also possible we'll see an ACK here, denoted by a + | |
| if data[0] == "+": | |
| self._receivedData = data[1:] | |
| return self.PACKET_ACK | |
| if ord(data[0]) == 3: | |
| self._receivedData = data[1:] | |
| return self.PACKET_INTERRUPT | |
| if data[0] == "$": | |
| i += 1 | |
| else: | |
| raise self.InvalidPacketException( | |
| "Unexpected leading byte: %s" % data[0] | |
| ) | |
| # If we're looking beyond the start of the received data, then we're | |
| # looking for the end of the packet content, denoted by a #. | |
| # Note that we pick up searching from where we left off last time | |
| while i < data_len and data[i] != "#": | |
| i += 1 | |
| # If there isn't enough data left for a checksum, just remember where | |
| # we left off so we can pick up there the next time around | |
| if i > data_len - 3: | |
| self._receivedDataOffset = i | |
| return None | |
| # If we have enough data remaining for the checksum, extract it and | |
| # compare to the packet contents | |
| packet = data[1:i] | |
| i += 1 | |
| try: | |
| check = int(data[i : i + 2], 16) | |
| except ValueError: | |
| raise self.InvalidPacketException("Checksum is not valid hex") | |
| i += 2 | |
| if check != checksum(packet): | |
| raise self.InvalidPacketException( | |
| "Checksum %02x does not match content %02x" % (check, checksum(packet)) | |
| ) | |
| # remove parsed bytes from _receivedData and reset offset so parsing | |
| # can start on the next packet the next time around | |
| self._receivedData = data[i:] | |
| self._receivedDataOffset = 0 | |
| return packet | |
| def _sendPacket(self, packet: str, prefix="$"): | |
| assert self._socket is not None | |
| framed_packet = seven.bitcast_to_bytes(frame_packet(packet, prefix)) | |
| self._socket.sendall(framed_packet) | |
| def _handlePacket(self, packet): | |
| assert self._socket is not None | |
| if packet is self.PACKET_ACK: | |
| # Ignore ACKs from the client. For the future, we can consider | |
| # adding validation code to make sure the client only sends ACKs | |
| # when it's supposed to. | |
| return | |
| response = [""] | |
| # We'll handle the ack stuff here since it's not something any of the | |
| # tests will be concerned about, and it'll get turned off quickly anyway. | |
| if self._shouldSendAck: | |
| self._socket.sendall(seven.bitcast_to_bytes("+")) | |
| if packet == "QStartNoAckMode": | |
| self._shouldSendAck = False | |
| response = ["OK"] | |
| elif self.responder is not None: | |
| # Delegate everything else to our responder | |
| response = self.responder.respond(packet) | |
| # MockGDBServerResponder no longer returns non-lists but others like | |
| # ReverseTestBase still do | |
| if not isinstance(response, list): | |
| response = [response] | |
| for part in response: | |
| if part is MockGDBServerResponder.RESPONSE_NONE: | |
| continue | |
| if part is MockGDBServerResponder.RESPONSE_DISCONNECT: | |
| raise self.TerminateConnectionException() | |
| # Should have handled the non-str's above | |
| assert isinstance(part, str) | |
| self._sendPacket(part) | |
| PACKET_ACK = object() | |
| PACKET_INTERRUPT = object() | |
| class TerminateConnectionException(Exception): | |
| pass | |
| class InvalidPacketException(Exception): | |
| pass |