1""" 2Base class for gdb-remote test cases. 3""" 4 5from __future__ import print_function 6 7 8 9import errno 10import os 11import os.path 12import platform 13import random 14import re 15import select 16import signal 17import socket 18import subprocess 19import sys 20import tempfile 21import time 22from lldbsuite.test import configuration 23from lldbsuite.test.lldbtest import * 24from lldbgdbserverutils import * 25import logging 26 27class _ConnectionRefused(IOError): 28 pass 29 30class GdbRemoteTestCaseBase(TestBase): 31 32 _TIMEOUT_SECONDS = 7 33 34 _GDBREMOTE_KILL_PACKET = "$k#6b" 35 36 # Start the inferior separately, attach to the inferior on the stub command line. 37 _STARTUP_ATTACH = "attach" 38 # Start the inferior separately, start the stub without attaching, allow the test to attach to the inferior however it wants (e.g. $vAttach;pid). 39 _STARTUP_ATTACH_MANUALLY = "attach_manually" 40 # Start the stub, and launch the inferior with an $A packet via the initial packet stream. 41 _STARTUP_LAUNCH = "launch" 42 43 # GDB Signal numbers that are not target-specific used for common exceptions 44 TARGET_EXC_BAD_ACCESS = 0x91 45 TARGET_EXC_BAD_INSTRUCTION = 0x92 46 TARGET_EXC_ARITHMETIC = 0x93 47 TARGET_EXC_EMULATION = 0x94 48 TARGET_EXC_SOFTWARE = 0x95 49 TARGET_EXC_BREAKPOINT = 0x96 50 51 _verbose_log_handler = None 52 _log_formatter = logging.Formatter(fmt='%(asctime)-15s %(levelname)-8s %(message)s') 53 54 def setUpBaseLogging(self): 55 self.logger = logging.getLogger(__name__) 56 57 if len(self.logger.handlers) > 0: 58 return # We have set up this handler already 59 60 self.logger.propagate = False 61 self.logger.setLevel(logging.DEBUG) 62 63 # log all warnings to stderr 64 handler = logging.StreamHandler() 65 handler.setLevel(logging.WARNING) 66 handler.setFormatter(self._log_formatter) 67 self.logger.addHandler(handler) 68 69 70 def isVerboseLoggingRequested(self): 71 # We will report our detailed logs if the user requested that the "gdb-remote" channel is 72 # logged. 73 return any(("gdb-remote" in channel) for channel in lldbtest_config.channels) 74 75 def setUp(self): 76 TestBase.setUp(self) 77 78 self.setUpBaseLogging() 79 self._remote_server_log_file = None 80 self.debug_monitor_extra_args = [] 81 self._pump_queues = socket_packet_pump.PumpQueues() 82 83 if self.isVerboseLoggingRequested(): 84 # If requested, full logs go to a log file 85 self._verbose_log_handler = logging.FileHandler(self.log_basename + "-host.log") 86 self._verbose_log_handler.setFormatter(self._log_formatter) 87 self._verbose_log_handler.setLevel(logging.DEBUG) 88 self.logger.addHandler(self._verbose_log_handler) 89 90 self.test_sequence = GdbRemoteTestSequence(self.logger) 91 self.set_inferior_startup_launch() 92 self.port = self.get_next_port() 93 self.named_pipe_path = None 94 self.named_pipe = None 95 self.named_pipe_fd = None 96 self.stub_sends_two_stop_notifications_on_kill = False 97 if configuration.lldb_platform_url: 98 if configuration.lldb_platform_url.startswith('unix-'): 99 url_pattern = '(.+)://\[?(.+?)\]?/.*' 100 else: 101 url_pattern = '(.+)://(.+):\d+' 102 scheme, host = re.match(url_pattern, configuration.lldb_platform_url).groups() 103 if configuration.lldb_platform_name == 'remote-android' and host != 'localhost': 104 self.stub_device = host 105 self.stub_hostname = 'localhost' 106 else: 107 self.stub_device = None 108 self.stub_hostname = host 109 else: 110 self.stub_hostname = "localhost" 111 112 def tearDown(self): 113 self._pump_queues.verify_queues_empty() 114 115 if self._remote_server_log_file is not None: 116 lldb.remote_platform.Get(lldb.SBFileSpec(self._remote_server_log_file), 117 lldb.SBFileSpec(self.getLocalServerLogFile())) 118 lldb.remote_platform.Run(lldb.SBPlatformShellCommand("rm " + self._remote_server_log_file)) 119 self._remote_server_log_file = None 120 121 self.logger.removeHandler(self._verbose_log_handler) 122 self._verbose_log_handler = None 123 TestBase.tearDown(self) 124 125 def getLocalServerLogFile(self): 126 return self.log_basename + "-server.log" 127 128 def setUpServerLogging(self, is_llgs): 129 if len(lldbtest_config.channels) == 0: 130 return # No logging requested 131 132 if lldb.remote_platform: 133 log_file = lldbutil.join_remote_paths(lldb.remote_platform.GetWorkingDirectory(), "server.log") 134 self._remote_server_log_file = log_file 135 else: 136 log_file = self.getLocalServerLogFile() 137 138 if is_llgs: 139 self.debug_monitor_extra_args.append("--log-file=" + log_file) 140 self.debug_monitor_extra_args.append("--log-channels={}".format(":".join(lldbtest_config.channels))) 141 else: 142 self.debug_monitor_extra_args = ["--log-file=" + self.log_file, "--log-flags=0x800000"] 143 144 def get_next_port(self): 145 return 12000 + random.randint(0,3999) 146 147 def reset_test_sequence(self): 148 self.test_sequence = GdbRemoteTestSequence(self.logger) 149 150 def create_named_pipe(self): 151 # Create a temp dir and name for a pipe. 152 temp_dir = tempfile.mkdtemp() 153 named_pipe_path = os.path.join(temp_dir, "stub_port_number") 154 155 # Create the named pipe. 156 os.mkfifo(named_pipe_path) 157 158 # Open the read side of the pipe in non-blocking mode. This will return right away, ready or not. 159 named_pipe_fd = os.open(named_pipe_path, os.O_RDONLY | os.O_NONBLOCK) 160 161 # Create the file for the named pipe. Note this will follow semantics of 162 # a non-blocking read side of a named pipe, which has different semantics 163 # than a named pipe opened for read in non-blocking mode. 164 named_pipe = os.fdopen(named_pipe_fd, "r") 165 self.assertIsNotNone(named_pipe) 166 167 def shutdown_named_pipe(): 168 # Close the pipe. 169 try: 170 named_pipe.close() 171 except: 172 print("failed to close named pipe") 173 None 174 175 # Delete the pipe. 176 try: 177 os.remove(named_pipe_path) 178 except: 179 print("failed to delete named pipe: {}".format(named_pipe_path)) 180 None 181 182 # Delete the temp directory. 183 try: 184 os.rmdir(temp_dir) 185 except: 186 print("failed to delete temp dir: {}, directory contents: '{}'".format(temp_dir, os.listdir(temp_dir))) 187 None 188 189 # Add the shutdown hook to clean up the named pipe. 190 self.addTearDownHook(shutdown_named_pipe) 191 192 # Clear the port so the stub selects a port number. 193 self.port = 0 194 195 return (named_pipe_path, named_pipe, named_pipe_fd) 196 197 def get_stub_port_from_named_socket(self, read_timeout_seconds=5): 198 # Wait for something to read with a max timeout. 199 (ready_readers, _, _) = select.select([self.named_pipe_fd], [], [], read_timeout_seconds) 200 self.assertIsNotNone(ready_readers, "write side of pipe has not written anything - stub isn't writing to pipe.") 201 self.assertNotEqual(len(ready_readers), 0, "write side of pipe has not written anything - stub isn't writing to pipe.") 202 203 # Read the port from the named pipe. 204 stub_port_raw = self.named_pipe.read() 205 self.assertIsNotNone(stub_port_raw) 206 self.assertNotEqual(len(stub_port_raw), 0, "no content to read on pipe") 207 208 # Trim null byte, convert to int. 209 stub_port_raw = stub_port_raw[:-1] 210 stub_port = int(stub_port_raw) 211 self.assertTrue(stub_port > 0) 212 213 return stub_port 214 215 def run_shell_cmd(self, cmd): 216 platform = self.dbg.GetSelectedPlatform() 217 shell_cmd = lldb.SBPlatformShellCommand(cmd) 218 err = platform.Run(shell_cmd) 219 if err.Fail() or shell_cmd.GetStatus(): 220 m = "remote_platform.RunShellCommand('%s') failed:\n" % cmd 221 m += ">>> return code: %d\n" % shell_cmd.GetStatus() 222 if err.Fail(): 223 m += ">>> %s\n" % str(err).strip() 224 m += ">>> %s\n" % (shell_cmd.GetOutput() or 225 "Command generated no output.") 226 raise Exception(m) 227 return shell_cmd.GetOutput().strip() 228 229 def init_llgs_test(self, use_named_pipe=True): 230 if lldb.remote_platform: 231 # Remote platforms don't support named pipe based port negotiation 232 use_named_pipe = False 233 234 # Grab the ppid from /proc/[shell pid]/stat 235 shell_stat = self.run_shell_cmd("cat /proc/$$/stat") 236 # [pid] ([executable]) [state] [*ppid*] 237 pid = re.match(r"^\d+ \(.+\) . (\d+)", shell_stat).group(1) 238 ls_output = self.run_shell_cmd("ls -l /proc/%s/exe" % pid) 239 exe = ls_output.split()[-1] 240 241 # If the binary has been deleted, the link name has " (deleted)" appended. 242 # Remove if it's there. 243 self.debug_monitor_exe = re.sub(r' \(deleted\)$', '', exe) 244 else: 245 self.debug_monitor_exe = get_lldb_server_exe() 246 if not self.debug_monitor_exe: 247 self.skipTest("lldb-server exe not found") 248 249 self.debug_monitor_extra_args = ["gdbserver"] 250 self.setUpServerLogging(is_llgs=True) 251 252 if use_named_pipe: 253 (self.named_pipe_path, self.named_pipe, self.named_pipe_fd) = self.create_named_pipe() 254 255 def init_debugserver_test(self, use_named_pipe=True): 256 self.debug_monitor_exe = get_debugserver_exe() 257 if not self.debug_monitor_exe: 258 self.skipTest("debugserver exe not found") 259 self.setUpServerLogging(is_llgs=False) 260 if use_named_pipe: 261 (self.named_pipe_path, self.named_pipe, self.named_pipe_fd) = self.create_named_pipe() 262 # The debugserver stub has a race on handling the 'k' command, so it sends an X09 right away, then sends the real X notification 263 # when the process truly dies. 264 self.stub_sends_two_stop_notifications_on_kill = True 265 266 def forward_adb_port(self, source, target, direction, device): 267 adb = [ 'adb' ] + ([ '-s', device ] if device else []) + [ direction ] 268 def remove_port_forward(): 269 subprocess.call(adb + [ "--remove", "tcp:%d" % source]) 270 271 subprocess.call(adb + [ "tcp:%d" % source, "tcp:%d" % target]) 272 self.addTearDownHook(remove_port_forward) 273 274 def _verify_socket(self, sock): 275 # Normally, when the remote stub is not ready, we will get ECONNREFUSED during the 276 # connect() attempt. However, due to the way how ADB forwarding works, on android targets 277 # the connect() will always be successful, but the connection will be immediately dropped 278 # if ADB could not connect on the remote side. This function tries to detect this 279 # situation, and report it as "connection refused" so that the upper layers attempt the 280 # connection again. 281 triple = self.dbg.GetSelectedPlatform().GetTriple() 282 if not re.match(".*-.*-.*-android", triple): 283 return # Not android. 284 can_read, _, _ = select.select([sock], [], [], 0.1) 285 if sock not in can_read: 286 return # Data is not available, but the connection is alive. 287 if len(sock.recv(1, socket.MSG_PEEK)) == 0: 288 raise _ConnectionRefused() # Got EOF, connection dropped. 289 290 def create_socket(self): 291 sock = socket.socket() 292 logger = self.logger 293 294 triple = self.dbg.GetSelectedPlatform().GetTriple() 295 if re.match(".*-.*-.*-android", triple): 296 self.forward_adb_port(self.port, self.port, "forward", self.stub_device) 297 298 logger.info("Connecting to debug monitor on %s:%d", self.stub_hostname, self.port) 299 connect_info = (self.stub_hostname, self.port) 300 try: 301 sock.connect(connect_info) 302 except socket.error as serr: 303 if serr.errno == errno.ECONNREFUSED: 304 raise _ConnectionRefused() 305 raise serr 306 307 def shutdown_socket(): 308 if sock: 309 try: 310 # send the kill packet so lldb-server shuts down gracefully 311 sock.sendall(GdbRemoteTestCaseBase._GDBREMOTE_KILL_PACKET) 312 except: 313 logger.warning("failed to send kill packet to debug monitor: {}; ignoring".format(sys.exc_info()[0])) 314 315 try: 316 sock.close() 317 except: 318 logger.warning("failed to close socket to debug monitor: {}; ignoring".format(sys.exc_info()[0])) 319 320 self.addTearDownHook(shutdown_socket) 321 322 self._verify_socket(sock) 323 324 return sock 325 326 def set_inferior_startup_launch(self): 327 self._inferior_startup = self._STARTUP_LAUNCH 328 329 def set_inferior_startup_attach(self): 330 self._inferior_startup = self._STARTUP_ATTACH 331 332 def set_inferior_startup_attach_manually(self): 333 self._inferior_startup = self._STARTUP_ATTACH_MANUALLY 334 335 def get_debug_monitor_command_line_args(self, attach_pid=None): 336 if lldb.remote_platform: 337 commandline_args = self.debug_monitor_extra_args + ["*:{}".format(self.port)] 338 else: 339 commandline_args = self.debug_monitor_extra_args + ["localhost:{}".format(self.port)] 340 341 if attach_pid: 342 commandline_args += ["--attach=%d" % attach_pid] 343 if self.named_pipe_path: 344 commandline_args += ["--named-pipe", self.named_pipe_path] 345 return commandline_args 346 347 def run_platform_command(self, cmd): 348 platform = self.dbg.GetSelectedPlatform() 349 shell_command = lldb.SBPlatformShellCommand(cmd) 350 err = platform.Run(shell_command) 351 return (err, shell_command.GetOutput()) 352 353 def launch_debug_monitor(self, attach_pid=None, logfile=None): 354 # Create the command line. 355 commandline_args = self.get_debug_monitor_command_line_args(attach_pid=attach_pid) 356 357 # Start the server. 358 server = self.spawnSubprocess(self.debug_monitor_exe, commandline_args, install_remote=False) 359 self.addTearDownHook(self.cleanupSubprocesses) 360 self.assertIsNotNone(server) 361 362 # If we're receiving the stub's listening port from the named pipe, do that here. 363 if self.named_pipe: 364 self.port = self.get_stub_port_from_named_socket() 365 366 return server 367 368 def connect_to_debug_monitor(self, attach_pid=None): 369 if self.named_pipe: 370 # Create the stub. 371 server = self.launch_debug_monitor(attach_pid=attach_pid) 372 self.assertIsNotNone(server) 373 374 def shutdown_debug_monitor(): 375 try: 376 server.terminate() 377 except: 378 logger.warning("failed to terminate server for debug monitor: {}; ignoring".format(sys.exc_info()[0])) 379 self.addTearDownHook(shutdown_debug_monitor) 380 381 # Schedule debug monitor to be shut down during teardown. 382 logger = self.logger 383 384 # Attach to the stub and return a socket opened to it. 385 self.sock = self.create_socket() 386 return server 387 388 # We're using a random port algorithm to try not to collide with other ports, 389 # and retry a max # times. 390 attempts = 0 391 MAX_ATTEMPTS = 20 392 393 while attempts < MAX_ATTEMPTS: 394 server = self.launch_debug_monitor(attach_pid=attach_pid) 395 396 # Schedule debug monitor to be shut down during teardown. 397 logger = self.logger 398 def shutdown_debug_monitor(): 399 try: 400 server.terminate() 401 except: 402 logger.warning("failed to terminate server for debug monitor: {}; ignoring".format(sys.exc_info()[0])) 403 self.addTearDownHook(shutdown_debug_monitor) 404 405 connect_attemps = 0 406 MAX_CONNECT_ATTEMPTS = 10 407 408 while connect_attemps < MAX_CONNECT_ATTEMPTS: 409 # Create a socket to talk to the server 410 try: 411 logger.info("Connect attempt %d", connect_attemps+1) 412 self.sock = self.create_socket() 413 return server 414 except _ConnectionRefused as serr: 415 # Ignore, and try again. 416 pass 417 time.sleep(0.5) 418 connect_attemps += 1 419 420 # We should close the server here to be safe. 421 server.terminate() 422 423 # Increment attempts. 424 print("connect to debug monitor on port %d failed, attempt #%d of %d" % (self.port, attempts + 1, MAX_ATTEMPTS)) 425 attempts += 1 426 427 # And wait a random length of time before next attempt, to avoid collisions. 428 time.sleep(random.randint(1,5)) 429 430 # Now grab a new port number. 431 self.port = self.get_next_port() 432 433 raise Exception("failed to create a socket to the launched debug monitor after %d tries" % attempts) 434 435 def launch_process_for_attach(self, inferior_args=None, sleep_seconds=3, exe_path=None): 436 # We're going to start a child process that the debug monitor stub can later attach to. 437 # This process needs to be started so that it just hangs around for a while. We'll 438 # have it sleep. 439 if not exe_path: 440 exe_path = os.path.abspath("a.out") 441 442 args = [] 443 if inferior_args: 444 args.extend(inferior_args) 445 if sleep_seconds: 446 args.append("sleep:%d" % sleep_seconds) 447 448 inferior = self.spawnSubprocess(exe_path, args) 449 def shutdown_process_for_attach(): 450 try: 451 inferior.terminate() 452 except: 453 logger.warning("failed to terminate inferior process for attach: {}; ignoring".format(sys.exc_info()[0])) 454 self.addTearDownHook(shutdown_process_for_attach) 455 return inferior 456 457 def prep_debug_monitor_and_inferior(self, inferior_args=None, inferior_sleep_seconds=3, inferior_exe_path=None): 458 """Prep the debug monitor, the inferior, and the expected packet stream. 459 460 Handle the separate cases of using the debug monitor in attach-to-inferior mode 461 and in launch-inferior mode. 462 463 For attach-to-inferior mode, the inferior process is first started, then 464 the debug monitor is started in attach to pid mode (using --attach on the 465 stub command line), and the no-ack-mode setup is appended to the packet 466 stream. The packet stream is not yet executed, ready to have more expected 467 packet entries added to it. 468 469 For launch-inferior mode, the stub is first started, then no ack mode is 470 setup on the expected packet stream, then the verified launch packets are added 471 to the expected socket stream. The packet stream is not yet executed, ready 472 to have more expected packet entries added to it. 473 474 The return value is: 475 {inferior:<inferior>, server:<server>} 476 """ 477 inferior = None 478 attach_pid = None 479 480 if self._inferior_startup == self._STARTUP_ATTACH or self._inferior_startup == self._STARTUP_ATTACH_MANUALLY: 481 # Launch the process that we'll use as the inferior. 482 inferior = self.launch_process_for_attach(inferior_args=inferior_args, sleep_seconds=inferior_sleep_seconds, exe_path=inferior_exe_path) 483 self.assertIsNotNone(inferior) 484 self.assertTrue(inferior.pid > 0) 485 if self._inferior_startup == self._STARTUP_ATTACH: 486 # In this case, we want the stub to attach via the command line, so set the command line attach pid here. 487 attach_pid = inferior.pid 488 489 if self._inferior_startup == self._STARTUP_LAUNCH: 490 # Build launch args 491 if not inferior_exe_path: 492 inferior_exe_path = os.path.abspath("a.out") 493 494 if lldb.remote_platform: 495 remote_path = lldbutil.append_to_process_working_directory(os.path.basename(inferior_exe_path)) 496 remote_file_spec = lldb.SBFileSpec(remote_path, False) 497 err = lldb.remote_platform.Install(lldb.SBFileSpec(inferior_exe_path, True), remote_file_spec) 498 if err.Fail(): 499 raise Exception("remote_platform.Install('%s', '%s') failed: %s" % (inferior_exe_path, remote_path, err)) 500 inferior_exe_path = remote_path 501 502 launch_args = [inferior_exe_path] 503 if inferior_args: 504 launch_args.extend(inferior_args) 505 506 # Launch the debug monitor stub, attaching to the inferior. 507 server = self.connect_to_debug_monitor(attach_pid=attach_pid) 508 self.assertIsNotNone(server) 509 510 # Build the expected protocol stream 511 self.add_no_ack_remote_stream() 512 if self._inferior_startup == self._STARTUP_LAUNCH: 513 self.add_verified_launch_packets(launch_args) 514 515 return {"inferior":inferior, "server":server} 516 517 def expect_socket_recv(self, sock, expected_content_regex, timeout_seconds): 518 response = "" 519 timeout_time = time.time() + timeout_seconds 520 521 while not expected_content_regex.match(response) and time.time() < timeout_time: 522 can_read, _, _ = select.select([sock], [], [], timeout_seconds) 523 if can_read and sock in can_read: 524 recv_bytes = sock.recv(4096) 525 if recv_bytes: 526 response += recv_bytes 527 528 self.assertTrue(expected_content_regex.match(response)) 529 530 def expect_socket_send(self, sock, content, timeout_seconds): 531 request_bytes_remaining = content 532 timeout_time = time.time() + timeout_seconds 533 534 while len(request_bytes_remaining) > 0 and time.time() < timeout_time: 535 _, can_write, _ = select.select([], [sock], [], timeout_seconds) 536 if can_write and sock in can_write: 537 written_byte_count = sock.send(request_bytes_remaining) 538 request_bytes_remaining = request_bytes_remaining[written_byte_count:] 539 self.assertEqual(len(request_bytes_remaining), 0) 540 541 def do_handshake(self, stub_socket, timeout_seconds=5): 542 # Write the ack. 543 self.expect_socket_send(stub_socket, "+", timeout_seconds) 544 545 # Send the start no ack mode packet. 546 NO_ACK_MODE_REQUEST = "$QStartNoAckMode#b0" 547 bytes_sent = stub_socket.send(NO_ACK_MODE_REQUEST) 548 self.assertEqual(bytes_sent, len(NO_ACK_MODE_REQUEST)) 549 550 # Receive the ack and "OK" 551 self.expect_socket_recv(stub_socket, re.compile(r"^\+\$OK#[0-9a-fA-F]{2}$"), timeout_seconds) 552 553 # Send the final ack. 554 self.expect_socket_send(stub_socket, "+", timeout_seconds) 555 556 def add_no_ack_remote_stream(self): 557 self.test_sequence.add_log_lines( 558 ["read packet: +", 559 "read packet: $QStartNoAckMode#b0", 560 "send packet: +", 561 "send packet: $OK#9a", 562 "read packet: +"], 563 True) 564 565 def add_verified_launch_packets(self, launch_args): 566 self.test_sequence.add_log_lines( 567 ["read packet: %s" % build_gdbremote_A_packet(launch_args), 568 "send packet: $OK#00", 569 "read packet: $qLaunchSuccess#a5", 570 "send packet: $OK#00"], 571 True) 572 573 def add_thread_suffix_request_packets(self): 574 self.test_sequence.add_log_lines( 575 ["read packet: $QThreadSuffixSupported#e4", 576 "send packet: $OK#00", 577 ], True) 578 579 def add_process_info_collection_packets(self): 580 self.test_sequence.add_log_lines( 581 ["read packet: $qProcessInfo#dc", 582 { "direction":"send", "regex":r"^\$(.+)#[0-9a-fA-F]{2}$", "capture":{1:"process_info_raw"} }], 583 True) 584 585 _KNOWN_PROCESS_INFO_KEYS = [ 586 "pid", 587 "parent-pid", 588 "real-uid", 589 "real-gid", 590 "effective-uid", 591 "effective-gid", 592 "cputype", 593 "cpusubtype", 594 "ostype", 595 "triple", 596 "vendor", 597 "endian", 598 "ptrsize" 599 ] 600 601 def parse_process_info_response(self, context): 602 # Ensure we have a process info response. 603 self.assertIsNotNone(context) 604 process_info_raw = context.get("process_info_raw") 605 self.assertIsNotNone(process_info_raw) 606 607 # Pull out key:value; pairs. 608 process_info_dict = { match.group(1):match.group(2) for match in re.finditer(r"([^:]+):([^;]+);", process_info_raw) } 609 610 # Validate keys are known. 611 for (key, val) in list(process_info_dict.items()): 612 self.assertTrue(key in self._KNOWN_PROCESS_INFO_KEYS) 613 self.assertIsNotNone(val) 614 615 return process_info_dict 616 617 def add_register_info_collection_packets(self): 618 self.test_sequence.add_log_lines( 619 [ { "type":"multi_response", "query":"qRegisterInfo", "append_iteration_suffix":True, 620 "end_regex":re.compile(r"^\$(E\d+)?#[0-9a-fA-F]{2}$"), 621 "save_key":"reg_info_responses" } ], 622 True) 623 624 def parse_register_info_packets(self, context): 625 """Return an array of register info dictionaries, one per register info.""" 626 reg_info_responses = context.get("reg_info_responses") 627 self.assertIsNotNone(reg_info_responses) 628 629 # Parse register infos. 630 return [parse_reg_info_response(reg_info_response) for reg_info_response in reg_info_responses] 631 632 def expect_gdbremote_sequence(self, timeout_seconds=None): 633 if not timeout_seconds: 634 timeout_seconds = self._TIMEOUT_SECONDS 635 return expect_lldb_gdbserver_replay(self, self.sock, self.test_sequence, 636 self._pump_queues, timeout_seconds, self.logger) 637 638 _KNOWN_REGINFO_KEYS = [ 639 "name", 640 "alt-name", 641 "bitsize", 642 "offset", 643 "encoding", 644 "format", 645 "set", 646 "ehframe", 647 "dwarf", 648 "generic", 649 "container-regs", 650 "invalidate-regs" 651 ] 652 653 def assert_valid_reg_info(self, reg_info): 654 # Assert we know about all the reginfo keys parsed. 655 for key in reg_info: 656 self.assertTrue(key in self._KNOWN_REGINFO_KEYS) 657 658 # Check the bare-minimum expected set of register info keys. 659 self.assertTrue("name" in reg_info) 660 self.assertTrue("bitsize" in reg_info) 661 self.assertTrue("offset" in reg_info) 662 self.assertTrue("encoding" in reg_info) 663 self.assertTrue("format" in reg_info) 664 665 def find_pc_reg_info(self, reg_infos): 666 lldb_reg_index = 0 667 for reg_info in reg_infos: 668 if ("generic" in reg_info) and (reg_info["generic"] == "pc"): 669 return (lldb_reg_index, reg_info) 670 lldb_reg_index += 1 671 672 return (None, None) 673 674 def add_lldb_register_index(self, reg_infos): 675 """Add a "lldb_register_index" key containing the 0-baed index of each reg_infos entry. 676 677 We'll use this when we want to call packets like P/p with a register index but do so 678 on only a subset of the full register info set. 679 """ 680 self.assertIsNotNone(reg_infos) 681 682 reg_index = 0 683 for reg_info in reg_infos: 684 reg_info["lldb_register_index"] = reg_index 685 reg_index += 1 686 687 def add_query_memory_region_packets(self, address): 688 self.test_sequence.add_log_lines( 689 ["read packet: $qMemoryRegionInfo:{0:x}#00".format(address), 690 {"direction":"send", "regex":r"^\$(.+)#[0-9a-fA-F]{2}$", "capture":{1:"memory_region_response"} }], 691 True) 692 693 def parse_key_val_dict(self, key_val_text, allow_dupes=True): 694 self.assertIsNotNone(key_val_text) 695 kv_dict = {} 696 for match in re.finditer(r";?([^:]+):([^;]+)", key_val_text): 697 key = match.group(1) 698 val = match.group(2) 699 if key in kv_dict: 700 if allow_dupes: 701 if type(kv_dict[key]) == list: 702 kv_dict[key].append(val) 703 else: 704 # Promote to list 705 kv_dict[key] = [kv_dict[key], val] 706 else: 707 self.fail("key '{}' already present when attempting to add value '{}' (text='{}', dict={})".format(key, val, key_val_text, kv_dict)) 708 else: 709 kv_dict[key] = val 710 return kv_dict 711 712 def parse_memory_region_packet(self, context): 713 # Ensure we have a context. 714 self.assertIsNotNone(context.get("memory_region_response")) 715 716 # Pull out key:value; pairs. 717 mem_region_dict = self.parse_key_val_dict(context.get("memory_region_response")) 718 719 # Validate keys are known. 720 for (key, val) in list(mem_region_dict.items()): 721 self.assertTrue(key in ["start", "size", "permissions", "error"]) 722 self.assertIsNotNone(val) 723 724 # Return the dictionary of key-value pairs for the memory region. 725 return mem_region_dict 726 727 def assert_address_within_memory_region(self, test_address, mem_region_dict): 728 self.assertIsNotNone(mem_region_dict) 729 self.assertTrue("start" in mem_region_dict) 730 self.assertTrue("size" in mem_region_dict) 731 732 range_start = int(mem_region_dict["start"], 16) 733 range_size = int(mem_region_dict["size"], 16) 734 range_end = range_start + range_size 735 736 if test_address < range_start: 737 self.fail("address 0x{0:x} comes before range 0x{1:x} - 0x{2:x} (size 0x{3:x})".format(test_address, range_start, range_end, range_size)) 738 elif test_address >= range_end: 739 self.fail("address 0x{0:x} comes after range 0x{1:x} - 0x{2:x} (size 0x{3:x})".format(test_address, range_start, range_end, range_size)) 740 741 def add_threadinfo_collection_packets(self): 742 self.test_sequence.add_log_lines( 743 [ { "type":"multi_response", "first_query":"qfThreadInfo", "next_query":"qsThreadInfo", 744 "append_iteration_suffix":False, "end_regex":re.compile(r"^\$(l)?#[0-9a-fA-F]{2}$"), 745 "save_key":"threadinfo_responses" } ], 746 True) 747 748 def parse_threadinfo_packets(self, context): 749 """Return an array of thread ids (decimal ints), one per thread.""" 750 threadinfo_responses = context.get("threadinfo_responses") 751 self.assertIsNotNone(threadinfo_responses) 752 753 thread_ids = [] 754 for threadinfo_response in threadinfo_responses: 755 new_thread_infos = parse_threadinfo_response(threadinfo_response) 756 thread_ids.extend(new_thread_infos) 757 return thread_ids 758 759 def wait_for_thread_count(self, thread_count, timeout_seconds=3): 760 start_time = time.time() 761 timeout_time = start_time + timeout_seconds 762 763 actual_thread_count = 0 764 while actual_thread_count < thread_count: 765 self.reset_test_sequence() 766 self.add_threadinfo_collection_packets() 767 768 context = self.expect_gdbremote_sequence() 769 self.assertIsNotNone(context) 770 771 threads = self.parse_threadinfo_packets(context) 772 self.assertIsNotNone(threads) 773 774 actual_thread_count = len(threads) 775 776 if time.time() > timeout_time: 777 raise Exception( 778 'timed out after {} seconds while waiting for theads: waiting for at least {} threads, found {}'.format( 779 timeout_seconds, thread_count, actual_thread_count)) 780 781 return threads 782 783 def add_set_breakpoint_packets(self, address, do_continue=True, breakpoint_kind=1): 784 self.test_sequence.add_log_lines( 785 [# Set the breakpoint. 786 "read packet: $Z0,{0:x},{1}#00".format(address, breakpoint_kind), 787 # Verify the stub could set it. 788 "send packet: $OK#00", 789 ], True) 790 791 if (do_continue): 792 self.test_sequence.add_log_lines( 793 [# Continue the inferior. 794 "read packet: $c#63", 795 # Expect a breakpoint stop report. 796 {"direction":"send", "regex":r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture":{1:"stop_signo", 2:"stop_thread_id"} }, 797 ], True) 798 799 def add_remove_breakpoint_packets(self, address, breakpoint_kind=1): 800 self.test_sequence.add_log_lines( 801 [# Remove the breakpoint. 802 "read packet: $z0,{0:x},{1}#00".format(address, breakpoint_kind), 803 # Verify the stub could unset it. 804 "send packet: $OK#00", 805 ], True) 806 807 def add_qSupported_packets(self): 808 self.test_sequence.add_log_lines( 809 ["read packet: $qSupported#00", 810 {"direction":"send", "regex":r"^\$(.*)#[0-9a-fA-F]{2}", "capture":{1: "qSupported_response"}}, 811 ], True) 812 813 _KNOWN_QSUPPORTED_STUB_FEATURES = [ 814 "augmented-libraries-svr4-read", 815 "PacketSize", 816 "QStartNoAckMode", 817 "QThreadSuffixSupported", 818 "QListThreadsInStopReply", 819 "qXfer:auxv:read", 820 "qXfer:libraries:read", 821 "qXfer:libraries-svr4:read", 822 "qXfer:features:read", 823 "qEcho" 824 ] 825 826 def parse_qSupported_response(self, context): 827 self.assertIsNotNone(context) 828 829 raw_response = context.get("qSupported_response") 830 self.assertIsNotNone(raw_response) 831 832 # For values with key=val, the dict key and vals are set as expected. For feature+, feature- and feature?, the 833 # +,-,? is stripped from the key and set as the value. 834 supported_dict = {} 835 for match in re.finditer(r";?([^=;]+)(=([^;]+))?", raw_response): 836 key = match.group(1) 837 val = match.group(3) 838 839 # key=val: store as is 840 if val and len(val) > 0: 841 supported_dict[key] = val 842 else: 843 if len(key) < 2: 844 raise Exception("singular stub feature is too short: must be stub_feature{+,-,?}") 845 supported_type = key[-1] 846 key = key[:-1] 847 if not supported_type in ["+", "-", "?"]: 848 raise Exception("malformed stub feature: final character {} not in expected set (+,-,?)".format(supported_type)) 849 supported_dict[key] = supported_type 850 # Ensure we know the supported element 851 if not key in self._KNOWN_QSUPPORTED_STUB_FEATURES: 852 raise Exception("unknown qSupported stub feature reported: %s" % key) 853 854 return supported_dict 855 856 def run_process_then_stop(self, run_seconds=1): 857 # Tell the stub to continue. 858 self.test_sequence.add_log_lines( 859 ["read packet: $vCont;c#a8"], 860 True) 861 context = self.expect_gdbremote_sequence() 862 863 # Wait for run_seconds. 864 time.sleep(run_seconds) 865 866 # Send an interrupt, capture a T response. 867 self.reset_test_sequence() 868 self.test_sequence.add_log_lines( 869 ["read packet: {}".format(chr(3)), 870 {"direction":"send", "regex":r"^\$T([0-9a-fA-F]+)([^#]+)#[0-9a-fA-F]{2}$", "capture":{1:"stop_result"} }], 871 True) 872 context = self.expect_gdbremote_sequence() 873 self.assertIsNotNone(context) 874 self.assertIsNotNone(context.get("stop_result")) 875 876 return context 877 878 def select_modifiable_register(self, reg_infos): 879 """Find a register that can be read/written freely.""" 880 PREFERRED_REGISTER_NAMES = set(["rax",]) 881 882 # First check for the first register from the preferred register name set. 883 alternative_register_index = None 884 885 self.assertIsNotNone(reg_infos) 886 for reg_info in reg_infos: 887 if ("name" in reg_info) and (reg_info["name"] in PREFERRED_REGISTER_NAMES): 888 # We found a preferred register. Use it. 889 return reg_info["lldb_register_index"] 890 if ("generic" in reg_info) and (reg_info["generic"] == "fp"): 891 # A frame pointer register will do as a register to modify temporarily. 892 alternative_register_index = reg_info["lldb_register_index"] 893 894 # We didn't find a preferred register. Return whatever alternative register 895 # we found, if any. 896 return alternative_register_index 897 898 def extract_registers_from_stop_notification(self, stop_key_vals_text): 899 self.assertIsNotNone(stop_key_vals_text) 900 kv_dict = self.parse_key_val_dict(stop_key_vals_text) 901 902 registers = {} 903 for (key, val) in list(kv_dict.items()): 904 if re.match(r"^[0-9a-fA-F]+$", key): 905 registers[int(key, 16)] = val 906 return registers 907 908 def gather_register_infos(self): 909 self.reset_test_sequence() 910 self.add_register_info_collection_packets() 911 912 context = self.expect_gdbremote_sequence() 913 self.assertIsNotNone(context) 914 915 reg_infos = self.parse_register_info_packets(context) 916 self.assertIsNotNone(reg_infos) 917 self.add_lldb_register_index(reg_infos) 918 919 return reg_infos 920 921 def find_generic_register_with_name(self, reg_infos, generic_name): 922 self.assertIsNotNone(reg_infos) 923 for reg_info in reg_infos: 924 if ("generic" in reg_info) and (reg_info["generic"] == generic_name): 925 return reg_info 926 return None 927 928 def decode_gdbremote_binary(self, encoded_bytes): 929 decoded_bytes = "" 930 i = 0 931 while i < len(encoded_bytes): 932 if encoded_bytes[i] == "}": 933 # Handle escaped char. 934 self.assertTrue(i + 1 < len(encoded_bytes)) 935 decoded_bytes += chr(ord(encoded_bytes[i+1]) ^ 0x20) 936 i +=2 937 elif encoded_bytes[i] == "*": 938 # Handle run length encoding. 939 self.assertTrue(len(decoded_bytes) > 0) 940 self.assertTrue(i + 1 < len(encoded_bytes)) 941 repeat_count = ord(encoded_bytes[i+1]) - 29 942 decoded_bytes += decoded_bytes[-1] * repeat_count 943 i += 2 944 else: 945 decoded_bytes += encoded_bytes[i] 946 i += 1 947 return decoded_bytes 948 949 def build_auxv_dict(self, endian, word_size, auxv_data): 950 self.assertIsNotNone(endian) 951 self.assertIsNotNone(word_size) 952 self.assertIsNotNone(auxv_data) 953 954 auxv_dict = {} 955 956 while len(auxv_data) > 0: 957 # Chop off key. 958 raw_key = auxv_data[:word_size] 959 auxv_data = auxv_data[word_size:] 960 961 # Chop of value. 962 raw_value = auxv_data[:word_size] 963 auxv_data = auxv_data[word_size:] 964 965 # Convert raw text from target endian. 966 key = unpack_endian_binary_string(endian, raw_key) 967 value = unpack_endian_binary_string(endian, raw_value) 968 969 # Handle ending entry. 970 if key == 0: 971 self.assertEqual(value, 0) 972 return auxv_dict 973 974 # The key should not already be present. 975 self.assertFalse(key in auxv_dict) 976 auxv_dict[key] = value 977 978 self.fail("should not reach here - implies required double zero entry not found") 979 return auxv_dict 980 981 def read_binary_data_in_chunks(self, command_prefix, chunk_length): 982 """Collect command_prefix{offset:x},{chunk_length:x} until a single 'l' or 'l' with data is returned.""" 983 offset = 0 984 done = False 985 decoded_data = "" 986 987 while not done: 988 # Grab the next iteration of data. 989 self.reset_test_sequence() 990 self.test_sequence.add_log_lines([ 991 "read packet: ${}{:x},{:x}:#00".format(command_prefix, offset, chunk_length), 992 {"direction":"send", "regex":re.compile(r"^\$([^E])(.*)#[0-9a-fA-F]{2}$", re.MULTILINE|re.DOTALL), "capture":{1:"response_type", 2:"content_raw"} } 993 ], True) 994 995 context = self.expect_gdbremote_sequence() 996 self.assertIsNotNone(context) 997 998 response_type = context.get("response_type") 999 self.assertIsNotNone(response_type) 1000 self.assertTrue(response_type in ["l", "m"]) 1001 1002 # Move offset along. 1003 offset += chunk_length 1004 1005 # Figure out if we're done. We're done if the response type is l. 1006 done = response_type == "l" 1007 1008 # Decode binary data. 1009 content_raw = context.get("content_raw") 1010 if content_raw and len(content_raw) > 0: 1011 self.assertIsNotNone(content_raw) 1012 decoded_data += self.decode_gdbremote_binary(content_raw) 1013 return decoded_data 1014 1015 def add_interrupt_packets(self): 1016 self.test_sequence.add_log_lines([ 1017 # Send the intterupt. 1018 "read packet: {}".format(chr(3)), 1019 # And wait for the stop notification. 1020 {"direction":"send", "regex":r"^\$T([0-9a-fA-F]{2})(.*)#[0-9a-fA-F]{2}$", "capture":{1:"stop_signo", 2:"stop_key_val_text" } }, 1021 ], True) 1022 1023 def parse_interrupt_packets(self, context): 1024 self.assertIsNotNone(context.get("stop_signo")) 1025 self.assertIsNotNone(context.get("stop_key_val_text")) 1026 return (int(context["stop_signo"], 16), self.parse_key_val_dict(context["stop_key_val_text"])) 1027 1028 def add_QSaveRegisterState_packets(self, thread_id): 1029 if thread_id: 1030 # Use the thread suffix form. 1031 request = "read packet: $QSaveRegisterState;thread:{:x}#00".format(thread_id) 1032 else: 1033 request = "read packet: $QSaveRegisterState#00" 1034 1035 self.test_sequence.add_log_lines([ 1036 request, 1037 {"direction":"send", "regex":r"^\$(E?.*)#[0-9a-fA-F]{2}$", "capture":{1:"save_response" } }, 1038 ], True) 1039 1040 def parse_QSaveRegisterState_response(self, context): 1041 self.assertIsNotNone(context) 1042 1043 save_response = context.get("save_response") 1044 self.assertIsNotNone(save_response) 1045 1046 if len(save_response) < 1 or save_response[0] == "E": 1047 # error received 1048 return (False, None) 1049 else: 1050 return (True, int(save_response)) 1051 1052 def add_QRestoreRegisterState_packets(self, save_id, thread_id=None): 1053 if thread_id: 1054 # Use the thread suffix form. 1055 request = "read packet: $QRestoreRegisterState:{};thread:{:x}#00".format(save_id, thread_id) 1056 else: 1057 request = "read packet: $QRestoreRegisterState:{}#00".format(save_id) 1058 1059 self.test_sequence.add_log_lines([ 1060 request, 1061 "send packet: $OK#00" 1062 ], True) 1063 1064 def flip_all_bits_in_each_register_value(self, reg_infos, endian, thread_id=None): 1065 self.assertIsNotNone(reg_infos) 1066 1067 successful_writes = 0 1068 failed_writes = 0 1069 1070 for reg_info in reg_infos: 1071 # Use the lldb register index added to the reg info. We're not necessarily 1072 # working off a full set of register infos, so an inferred register index could be wrong. 1073 reg_index = reg_info["lldb_register_index"] 1074 self.assertIsNotNone(reg_index) 1075 1076 reg_byte_size = int(reg_info["bitsize"])/8 1077 self.assertTrue(reg_byte_size > 0) 1078 1079 # Handle thread suffix. 1080 if thread_id: 1081 p_request = "read packet: $p{:x};thread:{:x}#00".format(reg_index, thread_id) 1082 else: 1083 p_request = "read packet: $p{:x}#00".format(reg_index) 1084 1085 # Read the existing value. 1086 self.reset_test_sequence() 1087 self.test_sequence.add_log_lines([ 1088 p_request, 1089 { "direction":"send", "regex":r"^\$([0-9a-fA-F]+)#", "capture":{1:"p_response"} }, 1090 ], True) 1091 context = self.expect_gdbremote_sequence() 1092 self.assertIsNotNone(context) 1093 1094 # Verify the response length. 1095 p_response = context.get("p_response") 1096 self.assertIsNotNone(p_response) 1097 initial_reg_value = unpack_register_hex_unsigned(endian, p_response) 1098 1099 # Flip the value by xoring with all 1s 1100 all_one_bits_raw = "ff" * (int(reg_info["bitsize"]) / 8) 1101 flipped_bits_int = initial_reg_value ^ int(all_one_bits_raw, 16) 1102 # print("reg (index={}, name={}): val={}, flipped bits (int={}, hex={:x})".format(reg_index, reg_info["name"], initial_reg_value, flipped_bits_int, flipped_bits_int)) 1103 1104 # Handle thread suffix for P. 1105 if thread_id: 1106 P_request = "read packet: $P{:x}={};thread:{:x}#00".format(reg_index, pack_register_hex(endian, flipped_bits_int, byte_size=reg_byte_size), thread_id) 1107 else: 1108 P_request = "read packet: $P{:x}={}#00".format(reg_index, pack_register_hex(endian, flipped_bits_int, byte_size=reg_byte_size)) 1109 1110 # Write the flipped value to the register. 1111 self.reset_test_sequence() 1112 self.test_sequence.add_log_lines([ 1113 P_request, 1114 { "direction":"send", "regex":r"^\$(OK|E[0-9a-fA-F]+)#[0-9a-fA-F]{2}", "capture":{1:"P_response"} }, 1115 ], True) 1116 context = self.expect_gdbremote_sequence() 1117 self.assertIsNotNone(context) 1118 1119 # Determine if the write succeeded. There are a handful of registers that can fail, or partially fail 1120 # (e.g. flags, segment selectors, etc.) due to register value restrictions. Don't worry about them 1121 # all flipping perfectly. 1122 P_response = context.get("P_response") 1123 self.assertIsNotNone(P_response) 1124 if P_response == "OK": 1125 successful_writes += 1 1126 else: 1127 failed_writes += 1 1128 # print("reg (index={}, name={}) write FAILED (error: {})".format(reg_index, reg_info["name"], P_response)) 1129 1130 # Read back the register value, ensure it matches the flipped value. 1131 if P_response == "OK": 1132 self.reset_test_sequence() 1133 self.test_sequence.add_log_lines([ 1134 p_request, 1135 { "direction":"send", "regex":r"^\$([0-9a-fA-F]+)#", "capture":{1:"p_response"} }, 1136 ], True) 1137 context = self.expect_gdbremote_sequence() 1138 self.assertIsNotNone(context) 1139 1140 verify_p_response_raw = context.get("p_response") 1141 self.assertIsNotNone(verify_p_response_raw) 1142 verify_bits = unpack_register_hex_unsigned(endian, verify_p_response_raw) 1143 1144 if verify_bits != flipped_bits_int: 1145 # Some registers, like mxcsrmask and others, will permute what's written. Adjust succeed/fail counts. 1146 # print("reg (index={}, name={}): read verify FAILED: wrote {:x}, verify read back {:x}".format(reg_index, reg_info["name"], flipped_bits_int, verify_bits)) 1147 successful_writes -= 1 1148 failed_writes +=1 1149 1150 return (successful_writes, failed_writes) 1151 1152 def is_bit_flippable_register(self, reg_info): 1153 if not reg_info: 1154 return False 1155 if not "set" in reg_info: 1156 return False 1157 if reg_info["set"] != "General Purpose Registers": 1158 return False 1159 if ("container-regs" in reg_info) and (len(reg_info["container-regs"]) > 0): 1160 # Don't try to bit flip registers contained in another register. 1161 return False 1162 if re.match("^.s$", reg_info["name"]): 1163 # This is a 2-letter register name that ends in "s", like a segment register. 1164 # Don't try to bit flip these. 1165 return False 1166 if re.match("^(c|)psr$", reg_info["name"]): 1167 # This is an ARM program status register; don't flip it. 1168 return False 1169 # Okay, this looks fine-enough. 1170 return True 1171 1172 def read_register_values(self, reg_infos, endian, thread_id=None): 1173 self.assertIsNotNone(reg_infos) 1174 values = {} 1175 1176 for reg_info in reg_infos: 1177 # We append a register index when load reg infos so we can work with subsets. 1178 reg_index = reg_info.get("lldb_register_index") 1179 self.assertIsNotNone(reg_index) 1180 1181 # Handle thread suffix. 1182 if thread_id: 1183 p_request = "read packet: $p{:x};thread:{:x}#00".format(reg_index, thread_id) 1184 else: 1185 p_request = "read packet: $p{:x}#00".format(reg_index) 1186 1187 # Read it with p. 1188 self.reset_test_sequence() 1189 self.test_sequence.add_log_lines([ 1190 p_request, 1191 { "direction":"send", "regex":r"^\$([0-9a-fA-F]+)#", "capture":{1:"p_response"} }, 1192 ], True) 1193 context = self.expect_gdbremote_sequence() 1194 self.assertIsNotNone(context) 1195 1196 # Convert value from target endian to integral. 1197 p_response = context.get("p_response") 1198 self.assertIsNotNone(p_response) 1199 self.assertTrue(len(p_response) > 0) 1200 self.assertFalse(p_response[0] == "E") 1201 1202 values[reg_index] = unpack_register_hex_unsigned(endian, p_response) 1203 1204 return values 1205 1206 def add_vCont_query_packets(self): 1207 self.test_sequence.add_log_lines([ 1208 "read packet: $vCont?#49", 1209 {"direction":"send", "regex":r"^\$(vCont)?(.*)#[0-9a-fA-F]{2}$", "capture":{2:"vCont_query_response" } }, 1210 ], True) 1211 1212 def parse_vCont_query_response(self, context): 1213 self.assertIsNotNone(context) 1214 vCont_query_response = context.get("vCont_query_response") 1215 1216 # Handle case of no vCont support at all - in which case the capture group will be none or zero length. 1217 if not vCont_query_response or len(vCont_query_response) == 0: 1218 return {} 1219 1220 return {key:1 for key in vCont_query_response.split(";") if key and len(key) > 0} 1221 1222 def count_single_steps_until_true(self, thread_id, predicate, args, max_step_count=100, use_Hc_packet=True, step_instruction="s"): 1223 """Used by single step test that appears in a few different contexts.""" 1224 single_step_count = 0 1225 1226 while single_step_count < max_step_count: 1227 self.assertIsNotNone(thread_id) 1228 1229 # Build the packet for the single step instruction. We replace {thread}, if present, with the thread_id. 1230 step_packet = "read packet: ${}#00".format(re.sub(r"{thread}", "{:x}".format(thread_id), step_instruction)) 1231 # print("\nstep_packet created: {}\n".format(step_packet)) 1232 1233 # Single step. 1234 self.reset_test_sequence() 1235 if use_Hc_packet: 1236 self.test_sequence.add_log_lines( 1237 [# Set the continue thread. 1238 "read packet: $Hc{0:x}#00".format(thread_id), 1239 "send packet: $OK#00", 1240 ], True) 1241 self.test_sequence.add_log_lines([ 1242 # Single step. 1243 step_packet, 1244 # "read packet: $vCont;s:{0:x}#00".format(thread_id), 1245 # Expect a breakpoint stop report. 1246 {"direction":"send", "regex":r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture":{1:"stop_signo", 2:"stop_thread_id"} }, 1247 ], True) 1248 context = self.expect_gdbremote_sequence() 1249 self.assertIsNotNone(context) 1250 self.assertIsNotNone(context.get("stop_signo")) 1251 self.assertEqual(int(context.get("stop_signo"), 16), 1252 lldbutil.get_signal_number('SIGTRAP')) 1253 1254 single_step_count += 1 1255 1256 # See if the predicate is true. If so, we're done. 1257 if predicate(args): 1258 return (True, single_step_count) 1259 1260 # The predicate didn't return true within the runaway step count. 1261 return (False, single_step_count) 1262 1263 def g_c1_c2_contents_are(self, args): 1264 """Used by single step test that appears in a few different contexts.""" 1265 g_c1_address = args["g_c1_address"] 1266 g_c2_address = args["g_c2_address"] 1267 expected_g_c1 = args["expected_g_c1"] 1268 expected_g_c2 = args["expected_g_c2"] 1269 1270 # Read g_c1 and g_c2 contents. 1271 self.reset_test_sequence() 1272 self.test_sequence.add_log_lines( 1273 ["read packet: $m{0:x},{1:x}#00".format(g_c1_address, 1), 1274 {"direction":"send", "regex":r"^\$(.+)#[0-9a-fA-F]{2}$", "capture":{1:"g_c1_contents"} }, 1275 "read packet: $m{0:x},{1:x}#00".format(g_c2_address, 1), 1276 {"direction":"send", "regex":r"^\$(.+)#[0-9a-fA-F]{2}$", "capture":{1:"g_c2_contents"} }], 1277 True) 1278 1279 # Run the packet stream. 1280 context = self.expect_gdbremote_sequence() 1281 self.assertIsNotNone(context) 1282 1283 # Check if what we read from inferior memory is what we are expecting. 1284 self.assertIsNotNone(context.get("g_c1_contents")) 1285 self.assertIsNotNone(context.get("g_c2_contents")) 1286 1287 return (context.get("g_c1_contents").decode("hex") == expected_g_c1) and (context.get("g_c2_contents").decode("hex") == expected_g_c2) 1288 1289 def single_step_only_steps_one_instruction(self, use_Hc_packet=True, step_instruction="s"): 1290 """Used by single step test that appears in a few different contexts.""" 1291 # Start up the inferior. 1292 procs = self.prep_debug_monitor_and_inferior( 1293 inferior_args=["get-code-address-hex:swap_chars", "get-data-address-hex:g_c1", "get-data-address-hex:g_c2", "sleep:1", "call-function:swap_chars", "sleep:5"]) 1294 1295 # Run the process 1296 self.test_sequence.add_log_lines( 1297 [# Start running after initial stop. 1298 "read packet: $c#63", 1299 # Match output line that prints the memory address of the function call entry point. 1300 # Note we require launch-only testing so we can get inferior otuput. 1301 { "type":"output_match", "regex":r"^code address: 0x([0-9a-fA-F]+)\r\ndata address: 0x([0-9a-fA-F]+)\r\ndata address: 0x([0-9a-fA-F]+)\r\n$", 1302 "capture":{ 1:"function_address", 2:"g_c1_address", 3:"g_c2_address"} }, 1303 # Now stop the inferior. 1304 "read packet: {}".format(chr(3)), 1305 # And wait for the stop notification. 1306 {"direction":"send", "regex":r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture":{1:"stop_signo", 2:"stop_thread_id"} }], 1307 True) 1308 1309 # Run the packet stream. 1310 context = self.expect_gdbremote_sequence() 1311 self.assertIsNotNone(context) 1312 1313 # Grab the main thread id. 1314 self.assertIsNotNone(context.get("stop_thread_id")) 1315 main_thread_id = int(context.get("stop_thread_id"), 16) 1316 1317 # Grab the function address. 1318 self.assertIsNotNone(context.get("function_address")) 1319 function_address = int(context.get("function_address"), 16) 1320 1321 # Grab the data addresses. 1322 self.assertIsNotNone(context.get("g_c1_address")) 1323 g_c1_address = int(context.get("g_c1_address"), 16) 1324 1325 self.assertIsNotNone(context.get("g_c2_address")) 1326 g_c2_address = int(context.get("g_c2_address"), 16) 1327 1328 # Set a breakpoint at the given address. 1329 if self.getArchitecture() == "arm": 1330 # TODO: Handle case when setting breakpoint in thumb code 1331 BREAKPOINT_KIND = 4 1332 else: 1333 BREAKPOINT_KIND = 1 1334 self.reset_test_sequence() 1335 self.add_set_breakpoint_packets(function_address, do_continue=True, breakpoint_kind=BREAKPOINT_KIND) 1336 context = self.expect_gdbremote_sequence() 1337 self.assertIsNotNone(context) 1338 1339 # Remove the breakpoint. 1340 self.reset_test_sequence() 1341 self.add_remove_breakpoint_packets(function_address, breakpoint_kind=BREAKPOINT_KIND) 1342 context = self.expect_gdbremote_sequence() 1343 self.assertIsNotNone(context) 1344 1345 # Verify g_c1 and g_c2 match expected initial state. 1346 args = {} 1347 args["g_c1_address"] = g_c1_address 1348 args["g_c2_address"] = g_c2_address 1349 args["expected_g_c1"] = "0" 1350 args["expected_g_c2"] = "1" 1351 1352 self.assertTrue(self.g_c1_c2_contents_are(args)) 1353 1354 # Verify we take only a small number of steps to hit the first state. Might need to work through function entry prologue code. 1355 args["expected_g_c1"] = "1" 1356 args["expected_g_c2"] = "1" 1357 (state_reached, step_count) = self.count_single_steps_until_true(main_thread_id, self.g_c1_c2_contents_are, args, max_step_count=25, use_Hc_packet=use_Hc_packet, step_instruction=step_instruction) 1358 self.assertTrue(state_reached) 1359 1360 # Verify we hit the next state. 1361 args["expected_g_c1"] = "1" 1362 args["expected_g_c2"] = "0" 1363 (state_reached, step_count) = self.count_single_steps_until_true(main_thread_id, self.g_c1_c2_contents_are, args, max_step_count=5, use_Hc_packet=use_Hc_packet, step_instruction=step_instruction) 1364 self.assertTrue(state_reached) 1365 expected_step_count = 1 1366 arch = self.getArchitecture() 1367 1368 #MIPS required "3" (ADDIU, SB, LD) machine instructions for updation of variable value 1369 if re.match("mips",arch): 1370 expected_step_count = 3 1371 self.assertEqual(step_count, expected_step_count) 1372 1373 # Verify we hit the next state. 1374 args["expected_g_c1"] = "0" 1375 args["expected_g_c2"] = "0" 1376 (state_reached, step_count) = self.count_single_steps_until_true(main_thread_id, self.g_c1_c2_contents_are, args, max_step_count=5, use_Hc_packet=use_Hc_packet, step_instruction=step_instruction) 1377 self.assertTrue(state_reached) 1378 self.assertEqual(step_count, expected_step_count) 1379 1380 # Verify we hit the next state. 1381 args["expected_g_c1"] = "0" 1382 args["expected_g_c2"] = "1" 1383 (state_reached, step_count) = self.count_single_steps_until_true(main_thread_id, self.g_c1_c2_contents_are, args, max_step_count=5, use_Hc_packet=use_Hc_packet, step_instruction=step_instruction) 1384 self.assertTrue(state_reached) 1385 self.assertEqual(step_count, expected_step_count) 1386 1387