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