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