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