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