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