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