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