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