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