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