lldbgdbproxy.py

importlogging
importos
importos.path
importrandom

importlldb
fromlldbsuite.test.lldbtestimport*
fromlldbsuite.test.gdbclientutilsimport*
importlldbgdbserverutils
fromlldbsuite.supportimportseven


classGDBProxyTestBase(TestBase):
"""
 Base class for gdbserver proxy tests.

 This class will setup and start a mock GDB server for the test to use.
 It pases through requests to a regular lldb-server/debugserver and
 forwards replies back to the LLDB under test.
 """

"""The gdbserver that we implement."""
server=None
"""The inner lldb-server/debugserver process that we proxy requests into."""
monitor_server=None
monitor_sock=None

server_socket_class=TCPServerSocket

DEFAULT_TIMEOUT=20* (10if ("ASAN_OPTIONS"inos.environ) else1)

_verbose_log_handler=None
_log_formatter=logging.Formatter(fmt="%(asctime)-15s %(levelname)-8s %(message)s")

defsetUpBaseLogging(self):
self.logger=logging.getLogger(__name__)

self.logger.propagate=False
self.logger.setLevel(logging.DEBUG)

# log all warnings to stderr
self._stderr_log_handler=logging.StreamHandler()
self._stderr_log_handler.setLevel(
logging.DEBUGifself.TraceOn() elselogging.WARNING
 )
self._stderr_log_handler.setFormatter(self._log_formatter)
self.logger.addHandler(self._stderr_log_handler)

defsetUp(self):
TestBase.setUp(self)

self.setUpBaseLogging()

ifself.isVerboseLoggingRequested():
# If requested, full logs go to a log file
log_file_name=self.getLogBasenameForCurrentTest() +"-proxy.log"
self._verbose_log_handler=logging.FileHandler(log_file_name)
self._verbose_log_handler.setFormatter(self._log_formatter)
self._verbose_log_handler.setLevel(logging.DEBUG)
self.logger.addHandler(self._verbose_log_handler)

iflldbplatformutil.getPlatform() =="macosx":
self.debug_monitor_exe=lldbgdbserverutils.get_debugserver_exe()
self.debug_monitor_extra_args= []
else:
self.debug_monitor_exe=lldbgdbserverutils.get_lldb_server_exe()
self.debug_monitor_extra_args= ["gdbserver"]
self.assertIsNotNone(self.debug_monitor_exe)

self.server=MockGDBServer(self.server_socket_class())
self.server.responder=self

deftearDown(self):
# TestBase.tearDown will kill the process, but we need to kill it early
# so its client connection closes and we can stop the server before
# finally calling the base tearDown.
ifself.process() isnotNone:
self.process().Kill()
self.server.stop()

self.logger.removeHandler(self._verbose_log_handler)
self._verbose_log_handler=None
self.logger.removeHandler(self._stderr_log_handler)
self._stderr_log_handler=None

TestBase.tearDown(self)

defisVerboseLoggingRequested(self):
# We will report our detailed logs if the user requested that the "gdb-remote" channel is
# logged.
returnany(("gdb-remote"inchannel) forchannelinlldbtest_config.channels)

defconnect(self, target):
"""
 Create a process by connecting to the mock GDB server.
 """
self.prep_debug_monitor_and_inferior()
self.server.start()

listener=self.dbg.GetListener()
error=lldb.SBError()
process=target.ConnectRemote(
listener, self.server.get_connect_url(), "gdb-remote", error
 )
self.assertTrue(error.Success(), error.description)
self.assertTrue(process, PROCESS_IS_VALID)
returnprocess

defprep_debug_monitor_and_inferior(self):
inferior_exe_path=self.getBuildArtifact("a.out")
self.connect_to_debug_monitor([inferior_exe_path])
self.assertIsNotNone(self.monitor_server)
self.initial_handshake()

definitial_handshake(self):
self.monitor_server.send_packet(seven.bitcast_to_bytes("+"))
reply=seven.bitcast_to_string(self.monitor_server.get_normal_packet())
self.assertEqual(reply, "+")
self.monitor_server.send_packet(seven.bitcast_to_bytes("QStartNoAckMode"))
reply=seven.bitcast_to_string(self.monitor_server.get_normal_packet())
self.assertEqual(reply, "+")
reply=seven.bitcast_to_string(self.monitor_server.get_normal_packet())
self.assertEqual(reply, "OK")
self.monitor_server.set_validate_checksums(False)
self.monitor_server.send_packet(seven.bitcast_to_bytes("+"))
reply=seven.bitcast_to_string(self.monitor_server.get_normal_packet())
self.assertEqual(reply, "+")

defget_debug_monitor_command_line_args(self, connect_address, launch_args):
return (
self.debug_monitor_extra_args
+ ["--reverse-connect", connect_address]
+launch_args
 )

deflaunch_debug_monitor(self, launch_args):
family, type, proto, _, addr=socket.getaddrinfo(
"localhost", 0, proto=socket.IPPROTO_TCP
 )[0]
sock=socket.socket(family, type, proto)
sock.settimeout(self.DEFAULT_TIMEOUT)
sock.bind(addr)
sock.listen(1)
addr=sock.getsockname()
connect_address="[{}]:{}".format(*addr)

commandline_args=self.get_debug_monitor_command_line_args(
connect_address, launch_args
 )

# Start the server.
self.logger.info(f"Spawning monitor {commandline_args}")
monitor_process=self.spawnSubprocess(
self.debug_monitor_exe, commandline_args, install_remote=False
 )
self.assertIsNotNone(monitor_process)

self.monitor_sock=sock.accept()[0]
self.monitor_sock.settimeout(self.DEFAULT_TIMEOUT)
returnmonitor_process

defconnect_to_debug_monitor(self, launch_args):
monitor_process=self.launch_debug_monitor(launch_args)
# Turn off checksum validation because debugserver does not produce
# correct checksums.
self.monitor_server=lldbgdbserverutils.Server(
self.monitor_sock, monitor_process
 )

defrespond(self, packet):
"""Subclasses can override this to change how packets are handled."""
returnself.pass_through(packet)

defpass_through(self, packet):
self.logger.info(f"Sending packet {packet}")
self.monitor_server.send_packet(seven.bitcast_to_bytes(packet))
reply=seven.bitcast_to_string(self.monitor_server.get_normal_packet())
self.logger.info(f"Received reply {reply}")
returnreply