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