gdbclientutils.py

importctypes
importerrno
importio
importthreading
importsocket
importtraceback
fromlldbsuite.supportimportseven


defchecksum(message):
"""
 Calculate the GDB server protocol checksum of the message.

 The GDB server protocol uses a simple modulo 256 sum.
 """
check=0
forcinmessage:
check+=ord(c)
returncheck%256


defframe_packet(message):
"""
 Create a framed packet that's ready to send over the GDB connection
 channel.

 Framing includes surrounding the message between $ and #, and appending
 a two character hex checksum.
 """
return"$%s#%02x"% (message, checksum(message))


defescape_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=""
forcinmessage:
d=ord(c)
ifdin (0x23, 0x24, 0x7D):
out+=chr(0x7D)
out+=chr(d^0x20)
else:
out+=c
returnout


defhex_encode_bytes(message):
"""
 Encode the binary message by converting each byte into a two-character
 hex string.
 """
out=""
forcinmessage:
out+="%02x"%ord(c)
returnout


defhex_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
whilei<hex_len-1:
out+=chr(int(hex_bytes[i : i+2], 16))
i+=2
returnout


classMockGDBServerResponder:
"""
 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=40
packetLog=None

classRESPONSE_DISCONNECT:
pass

def__init__(self):
self.packetLog= []

defrespond(self, packet):
"""
 Return the unframed packet data that the server should issue in response
 to the given packet received from the client.
 """
self.packetLog.append(packet)
ifpacketisMockGDBServer.PACKET_INTERRUPT:
returnself.interrupt()
ifpacket=="c":
returnself.cont()
ifpacket.startswith("vCont;c"):
returnself.vCont(packet)
ifpacket[0] =="A":
returnself.A(packet)
ifpacket[0] =="D":
returnself.D(packet)
ifpacket[0] =="g":
returnself.readRegisters()
ifpacket[0] =="G":
# Gxxxxxxxxxxx
# Gxxxxxxxxxxx;thread:1234;
returnself.writeRegisters(packet[1:].split(";")[0])
ifpacket[0] =="p":
regnum=packet[1:].split(";")[0]
returnself.readRegister(int(regnum, 16))
ifpacket[0] =="P":
register, value=packet[1:].split("=")
returnself.writeRegister(int(register, 16), value)
ifpacket[0] =="m":
addr, length= [int(x, 16) forxinpacket[1:].split(",")]
returnself.readMemory(addr, length)
ifpacket[0] =="x":
addr, length= [int(x, 16) forxinpacket[1:].split(",")]
returnself.x(addr, length)
ifpacket[0] =="M":
location, encoded_data=packet[1:].split(":")
addr, length= [int(x, 16) forxinlocation.split(",")]
returnself.writeMemory(addr, encoded_data)
ifpacket[0:7] =="qSymbol":
returnself.qSymbol(packet[8:])
ifpacket[0:10] =="qSupported":
returnself.qSupported(packet[11:].split(";"))
ifpacket=="qfThreadInfo":
returnself.qfThreadInfo()
ifpacket=="qsThreadInfo":
returnself.qsThreadInfo()
ifpacket=="qC":
returnself.qC()
ifpacket=="QEnableErrorStrings":
returnself.QEnableErrorStrings()
ifpacket=="?":
returnself.haltReason()
ifpacket=="s":
returnself.haltReason()
ifpacket[0] =="H":
tid=packet[2:]
if"."intid:
asserttid.startswith("p")
# TODO: do we want to do anything with PID?
tid=tid.split(".", 1)[1]
returnself.selectThread(packet[1], int(tid, 16))
ifpacket[0:6] =="qXfer:":
obj, read, annex, location=packet[6:].split(":")
offset, length= [int(x, 16) forxinlocation.split(",")]
data, has_more=self.qXferRead(obj, annex, offset, length)
ifdataisnotNone:
returnself._qXferResponse(data, has_more)
return""
ifpacket.startswith("vAttach;"):
pid=packet.partition(";")[2]
returnself.vAttach(int(pid, 16))
ifpacket[0] =="Z":
returnself.setBreakpoint(packet)
ifpacket.startswith("qThreadStopInfo"):
threadnum=int(packet[15:], 16)
returnself.threadStopInfo(threadnum)
ifpacket=="QThreadSuffixSupported":
returnself.QThreadSuffixSupported()
ifpacket=="QListThreadsInStopReply":
returnself.QListThreadsInStopReply()
ifpacket.startswith("qMemoryRegionInfo:"):
returnself.qMemoryRegionInfo(int(packet.split(":")[1], 16))
ifpacket=="qQueryGDBServer":
returnself.qQueryGDBServer()
ifpacket=="qHostInfo":
returnself.qHostInfo()
ifpacket=="qGetWorkingDir":
returnself.qGetWorkingDir()
ifpacket=="qOffsets":
returnself.qOffsets()
ifpacket=="qProcessInfo":
returnself.qProcessInfo()
ifpacket=="qsProcessInfo":
returnself.qsProcessInfo()
ifpacket.startswith("qfProcessInfo"):
returnself.qfProcessInfo(packet)
ifpacket.startswith("jGetLoadedDynamicLibrariesInfos"):
returnself.jGetLoadedDynamicLibrariesInfos(packet)
ifpacket.startswith("qPathComplete:"):
returnself.qPathComplete()
ifpacket.startswith("vFile:"):
returnself.vFile(packet)
ifpacket.startswith("vRun;"):
returnself.vRun(packet)
ifpacket.startswith("qLaunchGDBServer;"):
_, host=packet.partition(";")[2].split(":")
returnself.qLaunchGDBServer(host)
ifpacket.startswith("qLaunchSuccess"):
returnself.qLaunchSuccess()
ifpacket.startswith("QEnvironment:"):
returnself.QEnvironment(packet)
ifpacket.startswith("QEnvironmentHexEncoded:"):
returnself.QEnvironmentHexEncoded(packet)
ifpacket.startswith("qRegisterInfo"):
regnum=int(packet[len("qRegisterInfo") :], 16)
returnself.qRegisterInfo(regnum)
ifpacket=="k":
returnself.k()

returnself.other(packet)

defqsProcessInfo(self):
return"E04"

defqfProcessInfo(self, packet):
return"E04"

defjGetLoadedDynamicLibrariesInfos(self, packet):
return""

defqGetWorkingDir(self):
return"2f"

defqOffsets(self):
return""

defqProcessInfo(self):
return""

defqHostInfo(self):
return"ptrsize:8;endian:little;"

defqQueryGDBServer(self):
return"E04"

definterrupt(self):
raiseself.UnexpectedPacketException()

defcont(self):
raiseself.UnexpectedPacketException()

defvCont(self, packet):
raiseself.UnexpectedPacketException()

defA(self, packet):
return""

defD(self, packet):
return"OK"

defreadRegisters(self):
return"00000000"*self.registerCount

defreadRegister(self, register):
return"00000000"

defwriteRegisters(self, registers_hex):
return"OK"

defwriteRegister(self, register, value_hex):
return"OK"

defreadMemory(self, addr, length):
return"00"*length

defx(self, addr, length):
return""

defwriteMemory(self, addr, data_hex):
return"OK"

defqSymbol(self, symbol_args):
return"OK"

defqSupported(self, client_supported):
return"qXfer:features:read+;PacketSize=3fff;QStartNoAckMode+"

defqfThreadInfo(self):
return"l"

defqsThreadInfo(self):
return"l"

defqC(self):
return"QC0"

defQEnableErrorStrings(self):
return"OK"

defhaltReason(self):
# SIGINT is 2, return type is 2 digit hex string
return"S02"

defqXferRead(self, obj, annex, offset, length):
returnNone, False

def_qXferResponse(self, data, has_more):
return"%s%s"% ("m"ifhas_moreelse"l", escape_binary(data))

defvAttach(self, pid):
raiseself.UnexpectedPacketException()

defselectThread(self, op, thread_id):
return"OK"

defsetBreakpoint(self, packet):
raiseself.UnexpectedPacketException()

defthreadStopInfo(self, threadnum):
return""

defother(self, packet):
# empty string means unsupported
return""

defQThreadSuffixSupported(self):
return""

defQListThreadsInStopReply(self):
return""

defqMemoryRegionInfo(self, addr):
return""

defqPathComplete(self):
return""

defvFile(self, packet):
return""

defvRun(self, packet):
return""

defqLaunchGDBServer(self, host):
raiseself.UnexpectedPacketException()

defqLaunchSuccess(self):
return""

defQEnvironment(self, packet):
return"OK"

defQEnvironmentHexEncoded(self, packet):
return"OK"

defqRegisterInfo(self, num):
return""

defk(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.
 """

classUnexpectedPacketException(Exception):
pass


classServerChannel:
"""
 A wrapper class for TCP or pty-based server.
 """

defget_connect_address(self):
"""Get address for the client to connect to."""

defget_connect_url(self):
"""Get URL suitable for process connect command."""

defclose_server(self):
"""Close all resources used by the server."""

defaccept(self):
"""Accept a single client connection to the server."""

defclose_connection(self):
"""Close all resources used by the accepted connection."""

defrecv(self):
"""Receive a data packet from the connected client."""

defsendall(self, data):
"""Send the data to the connected client."""


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

defclose_server(self):
self._server_socket.close()

defaccept(self):
assertself._connectionisNone
# 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

defclose_connection(self):
assertself._connectionisnotNone
self._connection.close()
self._connection=None

defrecv(self):
assertself._connectionisnotNone
returnself._connection.recv(4096)

defsendall(self, data):
assertself._connectionisnotNone
returnself._connection.sendall(data)


classTCPServerSocket(ServerSocket):
def__init__(self):
family, type, proto, _, addr=socket.getaddrinfo(
"localhost", 0, proto=socket.IPPROTO_TCP
 )[0]
super().__init__(family, type, proto, addr)

defget_connect_address(self):
return"[{}]:{}".format(*self._server_socket.getsockname())

defget_connect_url(self):
return"connect://"+self.get_connect_address()


classUnixServerSocket(ServerSocket):
def__init__(self, addr):
super().__init__(socket.AF_UNIX, socket.SOCK_STREAM, 0, addr)

defget_connect_address(self):
returnself._server_socket.getsockname()

defget_connect_url(self):
return"unix-connect://"+self.get_connect_address()


classPtyServerSocket(ServerChannel):
def__init__(self):
importpty
importtty

primary, secondary=pty.openpty()
tty.setraw(primary)
self._primary=io.FileIO(primary, "r+b")
self._secondary=io.FileIO(secondary, "r+b")

defget_connect_address(self):
libc=ctypes.CDLL(None)
libc.ptsname.argtypes= (ctypes.c_int,)
libc.ptsname.restype=ctypes.c_char_p
returnlibc.ptsname(self._primary.fileno()).decode()

defget_connect_url(self):
return"serial://"+self.get_connect_address()

defclose_server(self):
self._secondary.close()
self._primary.close()

defrecv(self):
try:
returnself._primary.read(4096)
exceptOSErrorase:
# closing the pty results in EIO on Linux, convert it to EOF
ife.errno==errno.EIO:
returnb""
raise

defsendall(self, data):
returnself._primary.write(data)


classMockGDBServer:
"""
 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()

defstart(self):
# Start a thread that waits for a client connection.
self._thread=threading.Thread(target=self.run)
self._thread.start()

defstop(self):
ifself._threadisnotNone:
self._thread.join()
self._thread=None

defget_connect_address(self):
returnself._socket.get_connect_address()

defget_connect_url(self):
returnself._socket.get_connect_url()

defrun(self):
# For testing purposes, we only need to worry about one client
# connecting just one time.
try:
self._socket.accept()
except:
traceback.print_exc()
return
self._shouldSendAck=True
self._receivedData=""
self._receivedDataOffset=0
data=None
try:
whileTrue:
data=seven.bitcast_to_string(self._socket.recv())
ifdataisNoneorlen(data) ==0:
break
self._receive(data)
exceptself.TerminateConnectionException:
pass
exceptExceptionase:
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()
whilepacketisnotNone:
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.
 """
data=self._receivedData
i=self._receivedDataOffset
data_len=len(data)
ifdata_len==0:
returnNone
ifi==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 +
ifdata[0] =="+":
self._receivedData=data[1:]
returnself.PACKET_ACK
iford(data[0]) ==3:
self._receivedData=data[1:]
returnself.PACKET_INTERRUPT
ifdata[0] =="$":
i+=1
else:
raiseself.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
whilei<data_lenanddata[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
ifi>data_len-3:
self._receivedDataOffset=i
returnNone

# 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)
exceptValueError:
raiseself.InvalidPacketException("Checksum is not valid hex")
i+=2
ifcheck!=checksum(packet):
raiseself.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
returnpacket

def_sendPacket(self, packet):
self._socket.sendall(seven.bitcast_to_bytes(frame_packet(packet)))

def_handlePacket(self, packet):
ifpacketisself.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.
ifself._shouldSendAck:
self._socket.sendall(seven.bitcast_to_bytes("+"))
ifpacket=="QStartNoAckMode":
self._shouldSendAck=False
response="OK"
elifself.responderisnotNone:
# Delegate everything else to our responder
response=self.responder.respond(packet)
ifnotisinstance(response, list):
response= [response]
forpartinresponse:
ifpartisMockGDBServerResponder.RESPONSE_DISCONNECT:
raiseself.TerminateConnectionException()
self._sendPacket(part)

PACKET_ACK=object()
PACKET_INTERRUPT=object()

classTerminateConnectionException(Exception):
pass

classInvalidPacketException(Exception):
pass