1760 lines · python
1"""2Base class for gdb-remote test cases.3"""4 5import errno6import os7import os.path8import random9import re10import select11import socket12import subprocess13import sys14import tempfile15import time16from lldbsuite.test import configuration17from lldbsuite.test.lldbtest import *18from lldbsuite.support import seven19from lldbgdbserverutils import *20import logging21 22 23class _ConnectionRefused(IOError):24 pass25 26 27class GdbRemoteTestCaseFactory(type):28 def __new__(cls, name, bases, attrs):29 newattrs = {}30 for attrname, attrvalue in attrs.items():31 if not attrname.startswith("test"):32 newattrs[attrname] = attrvalue33 continue34 35 # If any debug server categories were explicitly tagged, assume36 # that list to be authoritative. If none were specified, try37 # all of them.38 all_categories = set(["debugserver", "llgs"])39 categories = set(getattr(attrvalue, "categories", [])) & all_categories40 if not categories:41 categories = all_categories42 43 for cat in categories:44 45 @decorators.add_test_categories([cat])46 @wraps(attrvalue)47 def test_method(self, attrvalue=attrvalue):48 return attrvalue(self)49 50 method_name = attrname + "_" + cat51 test_method.__name__ = method_name52 test_method.debug_server = cat53 newattrs[method_name] = test_method54 55 return super(GdbRemoteTestCaseFactory, cls).__new__(cls, name, bases, newattrs)56 57 58class GdbRemoteTestCaseBase(Base, metaclass=GdbRemoteTestCaseFactory):59 # Default time out in seconds. The timeout is increased tenfold under Asan.60 DEFAULT_TIMEOUT = 20 * (10 if ("ASAN_OPTIONS" in os.environ) else 1)61 # Default sleep time in seconds. The sleep time is doubled under Asan.62 DEFAULT_SLEEP = 5 * (2 if ("ASAN_OPTIONS" in os.environ) else 1)63 64 _GDBREMOTE_KILL_PACKET = b"$k#6b"65 66 # Start the inferior separately, attach to the inferior on the stub67 # command line.68 _STARTUP_ATTACH = "attach"69 # Start the inferior separately, start the stub without attaching, allow70 # the test to attach to the inferior however it wants (e.g. $vAttach;pid).71 _STARTUP_ATTACH_MANUALLY = "attach_manually"72 # Start the stub, and launch the inferior with an $A packet via the73 # initial packet stream.74 _STARTUP_LAUNCH = "launch"75 76 # GDB Signal numbers that are not target-specific used for common77 # exceptions78 TARGET_EXC_BAD_ACCESS = 0x9179 TARGET_EXC_BAD_INSTRUCTION = 0x9280 TARGET_EXC_ARITHMETIC = 0x9381 TARGET_EXC_EMULATION = 0x9482 TARGET_EXC_SOFTWARE = 0x9583 TARGET_EXC_BREAKPOINT = 0x9684 85 _verbose_log_handler = None86 _log_formatter = logging.Formatter(fmt="%(asctime)-15s %(levelname)-8s %(message)s")87 88 def setUpBaseLogging(self):89 self.logger = logging.getLogger(__name__)90 91 if len(self.logger.handlers) > 0:92 return # We have set up this handler already93 94 self.logger.propagate = False95 self.logger.setLevel(logging.DEBUG)96 97 # log all warnings to stderr98 handler = logging.StreamHandler()99 handler.setLevel(logging.WARNING)100 handler.setFormatter(self._log_formatter)101 self.logger.addHandler(handler)102 103 def isVerboseLoggingRequested(self):104 # We will report our detailed logs if the user requested that the "gdb-remote" channel is105 # logged.106 return any(("gdb-remote" in channel) for channel in lldbtest_config.channels)107 108 def getDebugServer(self):109 method = getattr(self, self.testMethodName)110 return getattr(method, "debug_server", None)111 112 def setUp(self):113 super(GdbRemoteTestCaseBase, self).setUp()114 115 self.setUpBaseLogging()116 self.debug_monitor_extra_args = []117 118 if self.isVerboseLoggingRequested():119 # If requested, full logs go to a log file120 self._verbose_log_handler = logging.FileHandler(121 self.getLogBasenameForCurrentTest() + "-host.log"122 )123 self._verbose_log_handler.setFormatter(self._log_formatter)124 self._verbose_log_handler.setLevel(logging.DEBUG)125 self.logger.addHandler(self._verbose_log_handler)126 127 self.test_sequence = GdbRemoteTestSequence(self.logger)128 self.set_inferior_startup_launch()129 self.port = self.get_next_port()130 self.stub_sends_two_stop_notifications_on_kill = False131 if configuration.lldb_platform_url:132 if configuration.lldb_platform_url.startswith("unix-"):133 url_pattern = r"(.+)://\[?(.+?)\]?/.*"134 else:135 url_pattern = r"(.+)://(.+):\d+"136 scheme, host = re.match(137 url_pattern, configuration.lldb_platform_url138 ).groups()139 if (140 configuration.lldb_platform_name == "remote-android"141 and host != "localhost"142 ):143 self.stub_device = host144 self.stub_hostname = "localhost"145 else:146 self.stub_device = None147 self.stub_hostname = host148 else:149 self.stub_hostname = "localhost"150 151 debug_server = self.getDebugServer()152 if debug_server == "debugserver":153 self._init_debugserver_test()154 else:155 self._init_llgs_test()156 157 def tearDown(self):158 self.logger.removeHandler(self._verbose_log_handler)159 self._verbose_log_handler = None160 TestBase.tearDown(self)161 162 def getLocalServerLogFile(self):163 return self.getLogBasenameForCurrentTest() + "-server.log"164 165 def setUpServerLogging(self, is_llgs):166 if len(lldbtest_config.channels) == 0:167 return # No logging requested168 169 if lldb.remote_platform:170 log_file = lldbutil.join_remote_paths(171 lldb.remote_platform.GetWorkingDirectory(), "server.log"172 )173 else:174 log_file = self.getLocalServerLogFile()175 176 if is_llgs:177 self.debug_monitor_extra_args.append("--log-file=" + log_file)178 self.debug_monitor_extra_args.append(179 "--log-channels={}".format(":".join(lldbtest_config.channels))180 )181 else:182 self.debug_monitor_extra_args = [183 "--log-file=" + log_file,184 "--log-flags=0x800000",185 ]186 187 def get_next_port(self):188 if available_ports := self.getPlatformAvailablePorts():189 return random.choice(available_ports)190 return 12000 + random.randint(0, 7999)191 192 def reset_test_sequence(self):193 self.test_sequence = GdbRemoteTestSequence(self.logger)194 195 def _init_llgs_test(self):196 reverse_connect = True197 if lldb.remote_platform:198 # Reverse connections may be tricky due to firewalls/NATs.199 reverse_connect = False200 201 # FIXME: This is extremely linux-oriented202 203 # Grab the ppid from /proc/[shell pid]/stat204 err, retcode, shell_stat = self.run_platform_command("cat /proc/$$/stat")205 self.assertTrue(206 err.Success() and retcode == 0,207 "Failed to read file /proc/$$/stat: %s, retcode: %d"208 % (err.GetCString(), retcode),209 )210 211 # [pid] ([executable]) [state] [*ppid*]212 pid = re.match(r"^\d+ \(.+\) . (\d+)", shell_stat).group(1)213 err, retcode, ls_output = self.run_platform_command(214 "ls -l /proc/%s/exe" % pid215 )216 self.assertTrue(217 err.Success() and retcode == 0,218 "Failed to read file /proc/%s/exe: %s, retcode: %d"219 % (pid, err.GetCString(), retcode),220 )221 exe = ls_output.split()[-1]222 223 # If the binary has been deleted, the link name has " (deleted)" appended.224 # Remove if it's there.225 self.debug_monitor_exe = re.sub(r" \(deleted\)$", "", exe)226 else:227 self.debug_monitor_exe = get_lldb_server_exe()228 229 self.debug_monitor_extra_args = ["gdbserver"]230 self.setUpServerLogging(is_llgs=True)231 232 self.reverse_connect = reverse_connect233 234 def _init_debugserver_test(self):235 self.debug_monitor_exe = get_debugserver_exe()236 self.setUpServerLogging(is_llgs=False)237 self.reverse_connect = True238 239 # The debugserver stub has a race on handling the 'k' command, so it sends an X09 right away, then sends the real X notification240 # when the process truly dies.241 self.stub_sends_two_stop_notifications_on_kill = True242 243 def forward_adb_port(self, source, target, direction, device):244 adb = ["adb"] + (["-s", device] if device else []) + [direction]245 246 def remove_port_forward():247 subprocess.call(adb + ["--remove", "tcp:%d" % source])248 249 subprocess.call(adb + ["tcp:%d" % source, "tcp:%d" % target])250 self.addTearDownHook(remove_port_forward)251 252 def _verify_socket(self, sock):253 # Normally, when the remote stub is not ready, we will get ECONNREFUSED during the254 # connect() attempt. However, due to the way how port forwarding can work, on some targets255 # the connect() will always be successful, but the connection will be immediately dropped256 # if we could not connect on the remote side. This function tries to detect this257 # situation, and report it as "connection refused" so that the upper layers attempt the258 # connection again.259 can_read, _, _ = select.select([sock], [], [], 0.1)260 if sock not in can_read:261 return # Data is not available, but the connection is alive.262 if len(sock.recv(1, socket.MSG_PEEK)) == 0:263 raise _ConnectionRefused() # Got EOF, connection dropped.264 265 def create_socket(self):266 try:267 sock = socket.socket(family=socket.AF_INET)268 except OSError as e:269 if e.errno != errno.EAFNOSUPPORT:270 raise271 sock = socket.socket(family=socket.AF_INET6)272 273 logger = self.logger274 275 triple = self.dbg.GetSelectedPlatform().GetTriple()276 if re.match(".*-.*-.*-android", triple):277 self.forward_adb_port(self.port, self.port, "forward", self.stub_device)278 279 logger.info(280 "Connecting to debug monitor on %s:%d", self.stub_hostname, self.port281 )282 connect_info = (self.stub_hostname, self.port)283 try:284 sock.connect(connect_info)285 except socket.error as serr:286 if serr.errno == errno.ECONNREFUSED:287 raise _ConnectionRefused()288 raise serr289 290 def shutdown_socket():291 if sock:292 try:293 # send the kill packet so lldb-server shuts down gracefully294 sock.sendall(GdbRemoteTestCaseBase._GDBREMOTE_KILL_PACKET)295 except:296 logger.warning(297 "failed to send kill packet to debug monitor: {}; ignoring".format(298 sys.exc_info()[0]299 )300 )301 302 try:303 sock.close()304 except:305 logger.warning(306 "failed to close socket to debug monitor: {}; ignoring".format(307 sys.exc_info()[0]308 )309 )310 311 self.addTearDownHook(shutdown_socket)312 313 self._verify_socket(sock)314 315 return sock316 317 def set_inferior_startup_launch(self):318 self._inferior_startup = self._STARTUP_LAUNCH319 320 def set_inferior_startup_attach(self):321 self._inferior_startup = self._STARTUP_ATTACH322 323 def set_inferior_startup_attach_manually(self):324 self._inferior_startup = self._STARTUP_ATTACH_MANUALLY325 326 def get_debug_monitor_command_line_args(self, attach_pid=None):327 commandline_args = self.debug_monitor_extra_args328 if attach_pid:329 commandline_args += ["--attach=%d" % attach_pid]330 if self.reverse_connect:331 commandline_args += ["--reverse-connect", self.connect_address]332 else:333 if lldb.remote_platform:334 commandline_args += ["*:{}".format(self.port)]335 else:336 commandline_args += ["localhost:{}".format(self.port)]337 338 return commandline_args339 340 def get_target_byte_order(self):341 inferior_exe_path = self.getBuildArtifact("a.out")342 target = self.dbg.CreateTarget(inferior_exe_path)343 return target.GetByteOrder()344 345 def launch_debug_monitor(self, attach_pid=None, logfile=None):346 if self.reverse_connect:347 family, type, proto, _, addr = socket.getaddrinfo(348 "localhost", 0, proto=socket.IPPROTO_TCP349 )[0]350 sock = socket.socket(family, type, proto)351 sock.settimeout(self.DEFAULT_TIMEOUT)352 353 sock.bind(addr)354 sock.listen(1)355 addr = sock.getsockname()356 self.connect_address = "[{}]:{}".format(*addr)357 358 # Create the command line.359 commandline_args = self.get_debug_monitor_command_line_args(360 attach_pid=attach_pid361 )362 363 # Start the server.364 server = self.spawnSubprocess(365 self.debug_monitor_exe, commandline_args, install_remote=False366 )367 self.assertIsNotNone(server)368 369 if self.reverse_connect:370 self.sock = sock.accept()[0]371 self.sock.settimeout(self.DEFAULT_TIMEOUT)372 373 return server374 375 def connect_to_debug_monitor(self, attach_pid=None):376 if self.reverse_connect:377 # Create the stub.378 server = self.launch_debug_monitor(attach_pid=attach_pid)379 self.assertIsNotNone(server)380 381 # Schedule debug monitor to be shut down during teardown.382 logger = self.logger383 384 self._server = Server(self.sock, server)385 return server386 387 # We're using a random port algorithm to try not to collide with other ports,388 # and retry a max # times.389 attempts = 0390 MAX_ATTEMPTS = 10391 attempt_wait = 3392 393 while attempts < MAX_ATTEMPTS:394 server = self.launch_debug_monitor(attach_pid=attach_pid)395 396 # Schedule debug monitor to be shut down during teardown.397 logger = self.logger398 399 connect_attempts = 0400 MAX_CONNECT_ATTEMPTS = 10401 402 while connect_attempts < MAX_CONNECT_ATTEMPTS:403 # Create a socket to talk to the server404 try:405 logger.info("Connect attempt %d", connect_attempts + 1)406 self.sock = self.create_socket()407 self._server = Server(self.sock, server)408 return server409 except _ConnectionRefused as serr:410 # Ignore, and try again.411 pass412 time.sleep(0.5)413 connect_attempts += 1414 415 # We should close the server here to be safe.416 server.terminate()417 418 # Increment attempts.419 print(420 "connect to debug monitor on port %d failed, attempt #%d of %d"421 % (self.port, attempts + 1, MAX_ATTEMPTS)422 )423 attempts += 1424 425 # And wait a random length of time before next attempt, to avoid426 # collisions.427 time.sleep(attempt_wait)428 attempt_wait *= 1.2429 430 # Now grab a new port number.431 self.port = self.get_next_port()432 433 raise Exception(434 "failed to create a socket to the launched debug monitor after %d tries"435 % attempts436 )437 438 def launch_process_for_attach(439 self, inferior_args=None, sleep_seconds=3, exe_path=None440 ):441 # We're going to start a child process that the debug monitor stub can later attach to.442 # This process needs to be started so that it just hangs around for a while. We'll443 # have it sleep.444 if not exe_path:445 exe_path = self.getBuildArtifact("a.out")446 447 # This file will be created once the inferior has enabled attaching.448 sync_file_path = lldbutil.append_to_process_working_directory(449 self, "process_ready"450 )451 args = [f"syncfile:{sync_file_path}"]452 if inferior_args:453 args.extend(inferior_args)454 if sleep_seconds:455 args.append("sleep:%d" % sleep_seconds)456 457 inferior = self.spawnSubprocess(exe_path, args)458 lldbutil.wait_for_file_on_target(self, sync_file_path)459 460 return inferior461 462 def prep_debug_monitor_and_inferior(463 self,464 inferior_args=None,465 inferior_sleep_seconds=3,466 inferior_exe_path=None,467 inferior_env=None,468 ):469 """Prep the debug monitor, the inferior, and the expected packet stream.470 471 Handle the separate cases of using the debug monitor in attach-to-inferior mode472 and in launch-inferior mode.473 474 For attach-to-inferior mode, the inferior process is first started, then475 the debug monitor is started in attach to pid mode (using --attach on the476 stub command line), and the no-ack-mode setup is appended to the packet477 stream. The packet stream is not yet executed, ready to have more expected478 packet entries added to it.479 480 For launch-inferior mode, the stub is first started, then no ack mode is481 setup on the expected packet stream, then the verified launch packets are added482 to the expected socket stream. The packet stream is not yet executed, ready483 to have more expected packet entries added to it.484 485 The return value is:486 {inferior:<inferior>, server:<server>}487 """488 inferior = None489 attach_pid = None490 491 if (492 self._inferior_startup == self._STARTUP_ATTACH493 or self._inferior_startup == self._STARTUP_ATTACH_MANUALLY494 ):495 # Launch the process that we'll use as the inferior.496 inferior = self.launch_process_for_attach(497 inferior_args=inferior_args,498 sleep_seconds=inferior_sleep_seconds,499 exe_path=inferior_exe_path,500 )501 self.assertIsNotNone(inferior)502 self.assertTrue(inferior.pid > 0)503 if self._inferior_startup == self._STARTUP_ATTACH:504 # In this case, we want the stub to attach via the command505 # line, so set the command line attach pid here.506 attach_pid = inferior.pid507 508 if self._inferior_startup == self._STARTUP_LAUNCH:509 # Build launch args510 if not inferior_exe_path:511 inferior_exe_path = self.getBuildArtifact("a.out")512 513 if lldb.remote_platform:514 remote_path = lldbutil.append_to_process_working_directory(515 self, os.path.basename(inferior_exe_path)516 )517 remote_file_spec = lldb.SBFileSpec(remote_path, False)518 err = lldb.remote_platform.Install(519 lldb.SBFileSpec(inferior_exe_path, True), remote_file_spec520 )521 if err.Fail():522 raise Exception(523 "remote_platform.Install('%s', '%s') failed: %s"524 % (inferior_exe_path, remote_path, err)525 )526 inferior_exe_path = remote_path527 528 launch_args = [inferior_exe_path]529 if inferior_args:530 launch_args.extend(inferior_args)531 532 # Launch the debug monitor stub, attaching to the inferior.533 server = self.connect_to_debug_monitor(attach_pid=attach_pid)534 self.assertIsNotNone(server)535 536 self.do_handshake()537 538 # Build the expected protocol stream539 if inferior_env:540 for name, value in inferior_env.items():541 self.add_set_environment_packets(name, value)542 if self._inferior_startup == self._STARTUP_LAUNCH:543 self.add_verified_launch_packets(launch_args)544 545 return {"inferior": inferior, "server": server}546 547 def do_handshake(self):548 server = self._server549 server.send_ack()550 server.send_packet(b"QStartNoAckMode")551 self.assertEqual(server.get_normal_packet(), b"+")552 self.assertEqual(server.get_normal_packet(), b"OK")553 server.send_ack()554 555 def add_verified_launch_packets(self, launch_args):556 self.test_sequence.add_log_lines(557 [558 "read packet: %s" % build_gdbremote_A_packet(launch_args),559 "send packet: $OK#00",560 "read packet: $qLaunchSuccess#a5",561 "send packet: $OK#00",562 ],563 True,564 )565 566 def add_thread_suffix_request_packets(self):567 self.test_sequence.add_log_lines(568 [569 "read packet: $QThreadSuffixSupported#e4",570 "send packet: $OK#00",571 ],572 True,573 )574 575 def add_process_info_collection_packets(self):576 self.test_sequence.add_log_lines(577 [578 "read packet: $qProcessInfo#dc",579 {580 "direction": "send",581 "regex": r"^\$(.+)#[0-9a-fA-F]{2}$",582 "capture": {1: "process_info_raw"},583 },584 ],585 True,586 )587 588 def add_set_environment_packets(self, name, value):589 self.test_sequence.add_log_lines(590 [591 "read packet: $QEnvironment:" + name + "=" + value + "#00",592 "send packet: $OK#00",593 ],594 True,595 )596 597 _KNOWN_PROCESS_INFO_KEYS = [598 "pid",599 "parent-pid",600 "real-uid",601 "real-gid",602 "effective-uid",603 "effective-gid",604 "cputype",605 "cpusubtype",606 "ostype",607 "triple",608 "vendor",609 "endian",610 "elf_abi",611 "ptrsize",612 ]613 614 def parse_process_info_response(self, context):615 # Ensure we have a process info response.616 self.assertIsNotNone(context)617 process_info_raw = context.get("process_info_raw")618 self.assertIsNotNone(process_info_raw)619 620 # Pull out key:value; pairs.621 process_info_dict = {622 match.group(1): match.group(2)623 for match in re.finditer(r"([^:]+):([^;]+);", process_info_raw)624 }625 626 # Validate keys are known.627 for key, val in list(process_info_dict.items()):628 self.assertTrue(key in self._KNOWN_PROCESS_INFO_KEYS)629 self.assertIsNotNone(val)630 631 return process_info_dict632 633 def add_register_info_collection_packets(self):634 self.test_sequence.add_log_lines(635 [636 {637 "type": "multi_response",638 "query": "qRegisterInfo",639 "append_iteration_suffix": True,640 "end_regex": re.compile(r"^\$(E\d+)?#[0-9a-fA-F]{2}$"),641 "save_key": "reg_info_responses",642 }643 ],644 True,645 )646 647 def parse_register_info_packets(self, context):648 """Return an array of register info dictionaries, one per register info."""649 reg_info_responses = context.get("reg_info_responses")650 self.assertIsNotNone(reg_info_responses)651 652 # Parse register infos.653 return [654 parse_reg_info_response(reg_info_response)655 for reg_info_response in reg_info_responses656 ]657 658 def expect_gdbremote_sequence(self):659 return expect_lldb_gdbserver_replay(660 self,661 self._server,662 self.test_sequence,663 self.DEFAULT_TIMEOUT * len(self.test_sequence),664 self.logger,665 )666 667 _KNOWN_REGINFO_KEYS = [668 "name",669 "alt-name",670 "bitsize",671 "offset",672 "encoding",673 "format",674 "set",675 "gcc",676 "ehframe",677 "dwarf",678 "generic",679 "container-regs",680 "invalidate-regs",681 "dynamic_size_dwarf_expr_bytes",682 "dynamic_size_dwarf_len",683 ]684 685 def assert_valid_reg_info(self, reg_info):686 # Assert we know about all the reginfo keys parsed.687 for key in reg_info:688 self.assertTrue(key in self._KNOWN_REGINFO_KEYS)689 690 # Check the bare-minimum expected set of register info keys.691 self.assertTrue("name" in reg_info)692 self.assertTrue("bitsize" in reg_info)693 694 if not (self.getArchitecture() == "aarch64" or self.isRISCV()):695 self.assertTrue("offset" in reg_info)696 697 self.assertTrue("encoding" in reg_info)698 self.assertTrue("format" in reg_info)699 700 def find_pc_reg_info(self, reg_infos):701 lldb_reg_index = 0702 for reg_info in reg_infos:703 if ("generic" in reg_info) and (reg_info["generic"] == "pc"):704 return (lldb_reg_index, reg_info)705 lldb_reg_index += 1706 707 return (None, None)708 709 def add_lldb_register_index(self, reg_infos):710 """Add a "lldb_register_index" key containing the 0-baed index of each reg_infos entry.711 712 We'll use this when we want to call packets like P/p with a register index but do so713 on only a subset of the full register info set.714 """715 self.assertIsNotNone(reg_infos)716 717 reg_index = 0718 for reg_info in reg_infos:719 reg_info["lldb_register_index"] = reg_index720 reg_index += 1721 722 def add_query_memory_region_packets(self, address):723 self.test_sequence.add_log_lines(724 [725 "read packet: $qMemoryRegionInfo:{0:x}#00".format(address),726 {727 "direction": "send",728 "regex": r"^\$(.+)#[0-9a-fA-F]{2}$",729 "capture": {1: "memory_region_response"},730 },731 ],732 True,733 )734 735 def parse_key_val_dict(self, key_val_text, allow_dupes=True):736 self.assertIsNotNone(key_val_text)737 kv_dict = {}738 for match in re.finditer(r";?([^:]+):([^;]+)", key_val_text):739 key = match.group(1)740 val = match.group(2)741 if key in kv_dict:742 if allow_dupes:743 if isinstance(kv_dict[key], list):744 kv_dict[key].append(val)745 else:746 # Promote to list747 kv_dict[key] = [kv_dict[key], val]748 else:749 self.fail(750 "key '{}' already present when attempting to add value '{}' (text='{}', dict={})".format(751 key, val, key_val_text, kv_dict752 )753 )754 else:755 kv_dict[key] = val756 return kv_dict757 758 def parse_memory_region_packet(self, context):759 # Ensure we have a context.760 self.assertIsNotNone(context.get("memory_region_response"))761 762 # Pull out key:value; pairs.763 mem_region_dict = self.parse_key_val_dict(context.get("memory_region_response"))764 765 # Validate keys are known.766 for key, val in list(mem_region_dict.items()):767 self.assertIn(768 key,769 [770 "start",771 "size",772 "permissions",773 "flags",774 "name",775 "error",776 "dirty-pages",777 "type",778 ],779 )780 self.assertIsNotNone(val)781 782 mem_region_dict["name"] = seven.unhexlify(mem_region_dict.get("name", ""))783 # Return the dictionary of key-value pairs for the memory region.784 return mem_region_dict785 786 def assert_address_within_memory_region(self, test_address, mem_region_dict):787 self.assertIsNotNone(mem_region_dict)788 self.assertTrue("start" in mem_region_dict)789 self.assertTrue("size" in mem_region_dict)790 791 range_start = int(mem_region_dict["start"], 16)792 range_size = int(mem_region_dict["size"], 16)793 range_end = range_start + range_size794 795 if test_address < range_start:796 self.fail(797 "address 0x{0:x} comes before range 0x{1:x} - 0x{2:x} (size 0x{3:x})".format(798 test_address, range_start, range_end, range_size799 )800 )801 elif test_address >= range_end:802 self.fail(803 "address 0x{0:x} comes after range 0x{1:x} - 0x{2:x} (size 0x{3:x})".format(804 test_address, range_start, range_end, range_size805 )806 )807 808 def add_threadinfo_collection_packets(self):809 self.test_sequence.add_log_lines(810 [811 {812 "type": "multi_response",813 "first_query": "qfThreadInfo",814 "next_query": "qsThreadInfo",815 "append_iteration_suffix": False,816 "end_regex": re.compile(r"^\$(l)?#[0-9a-fA-F]{2}$"),817 "save_key": "threadinfo_responses",818 }819 ],820 True,821 )822 823 def parse_threadinfo_packets(self, context):824 """Return an array of thread ids (decimal ints), one per thread."""825 threadinfo_responses = context.get("threadinfo_responses")826 self.assertIsNotNone(threadinfo_responses)827 828 thread_ids = []829 for threadinfo_response in threadinfo_responses:830 new_thread_infos = parse_threadinfo_response(threadinfo_response)831 thread_ids.extend(new_thread_infos)832 return thread_ids833 834 def launch_with_threads(self, thread_count):835 procs = self.prep_debug_monitor_and_inferior(836 inferior_args=["thread:new"] * (thread_count - 1) + ["trap"]837 )838 839 self.test_sequence.add_log_lines(840 [841 "read packet: $c#00",842 {843 "direction": "send",844 "regex": r"^\$T([0-9a-fA-F]{2})([^#]*)#..$",845 "capture": {1: "stop_signo", 2: "stop_reply_kv"},846 },847 ],848 True,849 )850 self.add_threadinfo_collection_packets()851 context = self.expect_gdbremote_sequence()852 threads = self.parse_threadinfo_packets(context)853 self.assertGreaterEqual(len(threads), thread_count)854 return context, threads855 856 def add_set_breakpoint_packets(857 self, address, z_packet_type=0, do_continue=True, breakpoint_kind=1858 ):859 self.test_sequence.add_log_lines(860 [ # Set the breakpoint.861 "read packet: $Z{2},{0:x},{1}#00".format(862 address, breakpoint_kind, z_packet_type863 ),864 # Verify the stub could set it.865 "send packet: $OK#00",866 ],867 True,868 )869 870 if do_continue:871 self.test_sequence.add_log_lines(872 [ # Continue the inferior.873 "read packet: $c#63",874 # Expect a breakpoint stop report.875 {876 "direction": "send",877 "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);",878 "capture": {1: "stop_signo", 2: "stop_thread_id"},879 },880 ],881 True,882 )883 884 def add_remove_breakpoint_packets(885 self, address, z_packet_type=0, breakpoint_kind=1886 ):887 self.test_sequence.add_log_lines(888 [ # Remove the breakpoint.889 "read packet: $z{2},{0:x},{1}#00".format(890 address, breakpoint_kind, z_packet_type891 ),892 # Verify the stub could unset it.893 "send packet: $OK#00",894 ],895 True,896 )897 898 def add_qSupported_packets(self, client_features=[]):899 features = "".join(";" + x for x in client_features)900 self.test_sequence.add_log_lines(901 [902 "read packet: $qSupported{}#00".format(features),903 {904 "direction": "send",905 "regex": r"^\$(.*)#[0-9a-fA-F]{2}",906 "capture": {1: "qSupported_response"},907 },908 ],909 True,910 )911 912 _KNOWN_QSUPPORTED_STUB_FEATURES = [913 "augmented-libraries-svr4-read",914 "PacketSize",915 "QStartNoAckMode",916 "QThreadSuffixSupported",917 "QListThreadsInStopReply",918 "qXfer:auxv:read",919 "qXfer:libraries:read",920 "qXfer:libraries-svr4:read",921 "qXfer:features:read",922 "qXfer:siginfo:read",923 "qEcho",924 "QPassSignals",925 "multiprocess",926 "fork-events",927 "vfork-events",928 "memory-tagging",929 "qSaveCore",930 "native-signals",931 "QNonStop",932 "SupportedWatchpointTypes",933 "SupportedCompressions",934 "MultiMemRead",935 ]936 937 def parse_qSupported_response(self, context):938 self.assertIsNotNone(context)939 940 raw_response = context.get("qSupported_response")941 self.assertIsNotNone(raw_response)942 943 # For values with key=val, the dict key and vals are set as expected. For feature+, feature- and feature?, the944 # +,-,? is stripped from the key and set as the value.945 supported_dict = {}946 for match in re.finditer(r";?([^=;]+)(=([^;]+))?", raw_response):947 key = match.group(1)948 val = match.group(3)949 950 # key=val: store as is951 if val and len(val) > 0:952 supported_dict[key] = val953 else:954 if len(key) < 2:955 raise Exception(956 "singular stub feature is too short: must be stub_feature{+,-,?}"957 )958 supported_type = key[-1]959 key = key[:-1]960 if not supported_type in ["+", "-", "?"]:961 raise Exception(962 "malformed stub feature: final character {} not in expected set (+,-,?)".format(963 supported_type964 )965 )966 supported_dict[key] = supported_type967 # Ensure we know the supported element968 if key not in self._KNOWN_QSUPPORTED_STUB_FEATURES:969 raise Exception("unknown qSupported stub feature reported: %s" % key)970 971 return supported_dict972 973 def continue_process_and_wait_for_stop(self):974 self.test_sequence.add_log_lines(975 [976 "read packet: $vCont;c#a8",977 {978 "direction": "send",979 "regex": r"^\$T([0-9a-fA-F]{2})(.*)#[0-9a-fA-F]{2}$",980 "capture": {1: "stop_signo", 2: "stop_key_val_text"},981 },982 ],983 True,984 )985 context = self.expect_gdbremote_sequence()986 self.assertIsNotNone(context)987 return self.parse_interrupt_packets(context)988 989 def select_modifiable_register(self, reg_infos):990 """Find a register that can be read/written freely."""991 PREFERRED_REGISTER_NAMES = set(992 [993 "rax",994 ]995 )996 997 # First check for the first register from the preferred register name998 # set.999 alternative_register_index = None1000 1001 self.assertIsNotNone(reg_infos)1002 for reg_info in reg_infos:1003 if ("name" in reg_info) and (reg_info["name"] in PREFERRED_REGISTER_NAMES):1004 # We found a preferred register. Use it.1005 return reg_info["lldb_register_index"]1006 if ("generic" in reg_info) and (1007 reg_info["generic"] == "fp" or reg_info["generic"] == "arg1"1008 ):1009 # A frame pointer or first arg register will do as a1010 # register to modify temporarily.1011 alternative_register_index = reg_info["lldb_register_index"]1012 1013 # We didn't find a preferred register. Return whatever alternative register1014 # we found, if any.1015 return alternative_register_index1016 1017 def extract_registers_from_stop_notification(self, stop_key_vals_text):1018 self.assertIsNotNone(stop_key_vals_text)1019 kv_dict = self.parse_key_val_dict(stop_key_vals_text)1020 1021 registers = {}1022 for key, val in list(kv_dict.items()):1023 if re.match(r"^[0-9a-fA-F]+$", key):1024 registers[int(key, 16)] = val1025 return registers1026 1027 def gather_register_infos(self):1028 self.reset_test_sequence()1029 self.add_register_info_collection_packets()1030 1031 context = self.expect_gdbremote_sequence()1032 self.assertIsNotNone(context)1033 1034 reg_infos = self.parse_register_info_packets(context)1035 self.assertIsNotNone(reg_infos)1036 self.add_lldb_register_index(reg_infos)1037 1038 return reg_infos1039 1040 def find_generic_register_with_name(self, reg_infos, generic_name):1041 self.assertIsNotNone(reg_infos)1042 for reg_info in reg_infos:1043 if ("generic" in reg_info) and (reg_info["generic"] == generic_name):1044 return reg_info1045 return None1046 1047 def find_register_with_name_and_dwarf_regnum(self, reg_infos, name, dwarf_num):1048 self.assertIsNotNone(reg_infos)1049 for reg_info in reg_infos:1050 if (reg_info["name"] == name) and (reg_info["dwarf"] == dwarf_num):1051 return reg_info1052 return None1053 1054 def decode_gdbremote_binary(self, encoded_bytes):1055 decoded_bytes = ""1056 i = 01057 while i < len(encoded_bytes):1058 if encoded_bytes[i] == "}":1059 # Handle escaped char.1060 self.assertTrue(i + 1 < len(encoded_bytes))1061 decoded_bytes += chr(ord(encoded_bytes[i + 1]) ^ 0x20)1062 i += 21063 elif encoded_bytes[i] == "*":1064 # Handle run length encoding.1065 self.assertTrue(len(decoded_bytes) > 0)1066 self.assertTrue(i + 1 < len(encoded_bytes))1067 repeat_count = ord(encoded_bytes[i + 1]) - 291068 decoded_bytes += decoded_bytes[-1] * repeat_count1069 i += 21070 else:1071 decoded_bytes += encoded_bytes[i]1072 i += 11073 return decoded_bytes1074 1075 def build_auxv_dict(self, endian, word_size, auxv_data):1076 self.assertIsNotNone(endian)1077 self.assertIsNotNone(word_size)1078 self.assertIsNotNone(auxv_data)1079 1080 auxv_dict = {}1081 1082 # PowerPC64le's auxvec has a special key that must be ignored.1083 # This special key may be used multiple times, resulting in1084 # multiple key/value pairs with the same key, which would otherwise1085 # break this test check for repeated keys.1086 #1087 # AT_IGNOREPPC = 221088 ignored_keys_for_arch = {"powerpc64le": [22]}1089 arch = self.getArchitecture()1090 ignore_keys = None1091 if arch in ignored_keys_for_arch:1092 ignore_keys = ignored_keys_for_arch[arch]1093 1094 while len(auxv_data) > 0:1095 # Chop off key.1096 raw_key = auxv_data[:word_size]1097 auxv_data = auxv_data[word_size:]1098 1099 # Chop of value.1100 raw_value = auxv_data[:word_size]1101 auxv_data = auxv_data[word_size:]1102 1103 # Convert raw text from target endian.1104 key = unpack_endian_binary_string(endian, raw_key)1105 value = unpack_endian_binary_string(endian, raw_value)1106 1107 if ignore_keys and key in ignore_keys:1108 continue1109 1110 # Handle ending entry.1111 if key == 0:1112 self.assertEqual(value, 0)1113 return auxv_dict1114 1115 # The key should not already be present.1116 self.assertFalse(key in auxv_dict)1117 auxv_dict[key] = value1118 1119 self.fail(1120 "should not reach here - implies required double zero entry not found"1121 )1122 return auxv_dict1123 1124 def read_binary_data_in_chunks(self, command_prefix, chunk_length):1125 """Collect command_prefix{offset:x},{chunk_length:x} until a single 'l' or 'l' with data is returned."""1126 offset = 01127 done = False1128 decoded_data = ""1129 1130 while not done:1131 # Grab the next iteration of data.1132 self.reset_test_sequence()1133 self.test_sequence.add_log_lines(1134 [1135 "read packet: ${}{:x},{:x}:#00".format(1136 command_prefix, offset, chunk_length1137 ),1138 {1139 "direction": "send",1140 "regex": re.compile(1141 r"^\$([^E])(.*)#[0-9a-fA-F]{2}$", re.MULTILINE | re.DOTALL1142 ),1143 "capture": {1: "response_type", 2: "content_raw"},1144 },1145 ],1146 True,1147 )1148 1149 context = self.expect_gdbremote_sequence()1150 self.assertIsNotNone(context)1151 1152 response_type = context.get("response_type")1153 self.assertIsNotNone(response_type)1154 self.assertTrue(response_type in ["l", "m"])1155 1156 # Move offset along.1157 offset += chunk_length1158 1159 # Figure out if we're done. We're done if the response type is l.1160 done = response_type == "l"1161 1162 # Decode binary data.1163 content_raw = context.get("content_raw")1164 if content_raw and len(content_raw) > 0:1165 self.assertIsNotNone(content_raw)1166 decoded_data += self.decode_gdbremote_binary(content_raw)1167 return decoded_data1168 1169 def add_interrupt_packets(self):1170 self.test_sequence.add_log_lines(1171 [1172 # Send the intterupt.1173 "read packet: {}".format(chr(3)),1174 # And wait for the stop notification.1175 {1176 "direction": "send",1177 "regex": r"^\$T([0-9a-fA-F]{2})(.*)#[0-9a-fA-F]{2}$",1178 "capture": {1: "stop_signo", 2: "stop_key_val_text"},1179 },1180 ],1181 True,1182 )1183 1184 def parse_interrupt_packets(self, context):1185 self.assertIsNotNone(context.get("stop_signo"))1186 self.assertIsNotNone(context.get("stop_key_val_text"))1187 return (1188 int(context["stop_signo"], 16),1189 self.parse_key_val_dict(context["stop_key_val_text"]),1190 )1191 1192 def add_QSaveRegisterState_packets(self, thread_id):1193 if thread_id:1194 # Use the thread suffix form.1195 request = "read packet: $QSaveRegisterState;thread:{:x}#00".format(1196 thread_id1197 )1198 else:1199 request = "read packet: $QSaveRegisterState#00"1200 1201 self.test_sequence.add_log_lines(1202 [1203 request,1204 {1205 "direction": "send",1206 "regex": r"^\$(E?.*)#[0-9a-fA-F]{2}$",1207 "capture": {1: "save_response"},1208 },1209 ],1210 True,1211 )1212 1213 def parse_QSaveRegisterState_response(self, context):1214 self.assertIsNotNone(context)1215 1216 save_response = context.get("save_response")1217 self.assertIsNotNone(save_response)1218 1219 if len(save_response) < 1 or save_response[0] == "E":1220 # error received1221 return (False, None)1222 else:1223 return (True, int(save_response))1224 1225 def add_QRestoreRegisterState_packets(self, save_id, thread_id=None):1226 if thread_id:1227 # Use the thread suffix form.1228 request = "read packet: $QRestoreRegisterState:{};thread:{:x}#00".format(1229 save_id, thread_id1230 )1231 else:1232 request = "read packet: $QRestoreRegisterState:{}#00".format(save_id)1233 1234 self.test_sequence.add_log_lines([request, "send packet: $OK#00"], True)1235 1236 def flip_all_bits_in_each_register_value(self, reg_infos, endian, thread_id=None):1237 self.assertIsNotNone(reg_infos)1238 1239 successful_writes = 01240 failed_writes = 01241 1242 for reg_info in reg_infos:1243 # Use the lldb register index added to the reg info. We're not necessarily1244 # working off a full set of register infos, so an inferred register1245 # index could be wrong.1246 reg_index = reg_info["lldb_register_index"]1247 self.assertIsNotNone(reg_index)1248 1249 reg_byte_size = int(reg_info["bitsize"]) // 81250 self.assertTrue(reg_byte_size > 0)1251 1252 # Handle thread suffix.1253 if thread_id:1254 p_request = "read packet: $p{:x};thread:{:x}#00".format(1255 reg_index, thread_id1256 )1257 else:1258 p_request = "read packet: $p{:x}#00".format(reg_index)1259 1260 # Read the existing value.1261 self.reset_test_sequence()1262 self.test_sequence.add_log_lines(1263 [1264 p_request,1265 {1266 "direction": "send",1267 "regex": r"^\$([0-9a-fA-F]+)#",1268 "capture": {1: "p_response"},1269 },1270 ],1271 True,1272 )1273 context = self.expect_gdbremote_sequence()1274 self.assertIsNotNone(context)1275 1276 # Verify the response length.1277 p_response = context.get("p_response")1278 self.assertIsNotNone(p_response)1279 initial_reg_value = unpack_register_hex_unsigned(endian, p_response)1280 1281 # Flip the value by xoring with all 1s1282 all_one_bits_raw = "ff" * (int(reg_info["bitsize"]) // 8)1283 flipped_bits_int = initial_reg_value ^ int(all_one_bits_raw, 16)1284 # 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))1285 1286 # Handle thread suffix for P.1287 if thread_id:1288 P_request = "read packet: $P{:x}={};thread:{:x}#00".format(1289 reg_index,1290 pack_register_hex(1291 endian, flipped_bits_int, byte_size=reg_byte_size1292 ),1293 thread_id,1294 )1295 else:1296 P_request = "read packet: $P{:x}={}#00".format(1297 reg_index,1298 pack_register_hex(1299 endian, flipped_bits_int, byte_size=reg_byte_size1300 ),1301 )1302 1303 # Write the flipped value to the register.1304 self.reset_test_sequence()1305 self.test_sequence.add_log_lines(1306 [1307 P_request,1308 {1309 "direction": "send",1310 "regex": r"^\$(OK|E[0-9a-fA-F]+)#[0-9a-fA-F]{2}",1311 "capture": {1: "P_response"},1312 },1313 ],1314 True,1315 )1316 context = self.expect_gdbremote_sequence()1317 self.assertIsNotNone(context)1318 1319 # Determine if the write succeeded. There are a handful of registers that can fail, or partially fail1320 # (e.g. flags, segment selectors, etc.) due to register value restrictions. Don't worry about them1321 # all flipping perfectly.1322 P_response = context.get("P_response")1323 self.assertIsNotNone(P_response)1324 if P_response == "OK":1325 successful_writes += 11326 else:1327 failed_writes += 11328 # print("reg (index={}, name={}) write FAILED (error: {})".format(reg_index, reg_info["name"], P_response))1329 1330 # Read back the register value, ensure it matches the flipped1331 # value.1332 if P_response == "OK":1333 self.reset_test_sequence()1334 self.test_sequence.add_log_lines(1335 [1336 p_request,1337 {1338 "direction": "send",1339 "regex": r"^\$([0-9a-fA-F]+)#",1340 "capture": {1: "p_response"},1341 },1342 ],1343 True,1344 )1345 context = self.expect_gdbremote_sequence()1346 self.assertIsNotNone(context)1347 1348 verify_p_response_raw = context.get("p_response")1349 self.assertIsNotNone(verify_p_response_raw)1350 verify_bits = unpack_register_hex_unsigned(1351 endian, verify_p_response_raw1352 )1353 1354 if verify_bits != flipped_bits_int:1355 # Some registers, like mxcsrmask and others, will permute what's written. Adjust succeed/fail counts.1356 # print("reg (index={}, name={}): read verify FAILED: wrote {:x}, verify read back {:x}".format(reg_index, reg_info["name"], flipped_bits_int, verify_bits))1357 successful_writes -= 11358 failed_writes += 11359 1360 return (successful_writes, failed_writes)1361 1362 def is_bit_flippable_register(self, reg_info):1363 if not reg_info:1364 return False1365 if not "set" in reg_info:1366 return False1367 if reg_info["set"] != "General Purpose Registers":1368 return False1369 if ("container-regs" in reg_info) and (len(reg_info["container-regs"]) > 0):1370 # Don't try to bit flip registers contained in another register.1371 return False1372 if re.match("^.s$", reg_info["name"]):1373 # This is a 2-letter register name that ends in "s", like a segment register.1374 # Don't try to bit flip these.1375 return False1376 if re.match("^(c|)psr$", reg_info["name"]):1377 # This is an ARM program status register; don't flip it.1378 return False1379 # Okay, this looks fine-enough.1380 return True1381 1382 def read_register_values(self, reg_infos, endian, thread_id=None):1383 self.assertIsNotNone(reg_infos)1384 values = {}1385 1386 for reg_info in reg_infos:1387 # We append a register index when load reg infos so we can work1388 # with subsets.1389 reg_index = reg_info.get("lldb_register_index")1390 self.assertIsNotNone(reg_index)1391 1392 # Handle thread suffix.1393 if thread_id:1394 p_request = "read packet: $p{:x};thread:{:x}#00".format(1395 reg_index, thread_id1396 )1397 else:1398 p_request = "read packet: $p{:x}#00".format(reg_index)1399 1400 # Read it with p.1401 self.reset_test_sequence()1402 self.test_sequence.add_log_lines(1403 [1404 p_request,1405 {1406 "direction": "send",1407 "regex": r"^\$([0-9a-fA-F]+)#",1408 "capture": {1: "p_response"},1409 },1410 ],1411 True,1412 )1413 context = self.expect_gdbremote_sequence()1414 self.assertIsNotNone(context)1415 1416 # Convert value from target endian to integral.1417 p_response = context.get("p_response")1418 self.assertIsNotNone(p_response)1419 self.assertTrue(len(p_response) > 0)1420 1421 # on x86 Darwin, 4 GPR registers are often1422 # unavailable, this is expected and correct.1423 if (1424 self.getArchitecture() == "x86_64"1425 and self.platformIsDarwin()1426 and p_response[0] == "E"1427 ):1428 values[reg_index] = 01429 else:1430 self.assertFalse(p_response[0] == "E")1431 1432 values[reg_index] = unpack_register_hex_unsigned(endian, p_response)1433 1434 return values1435 1436 def add_vCont_query_packets(self):1437 self.test_sequence.add_log_lines(1438 [1439 "read packet: $vCont?#49",1440 {1441 "direction": "send",1442 "regex": r"^\$(vCont)?(.*)#[0-9a-fA-F]{2}$",1443 "capture": {2: "vCont_query_response"},1444 },1445 ],1446 True,1447 )1448 1449 def parse_vCont_query_response(self, context):1450 self.assertIsNotNone(context)1451 vCont_query_response = context.get("vCont_query_response")1452 1453 # Handle case of no vCont support at all - in which case the capture1454 # group will be none or zero length.1455 if not vCont_query_response or len(vCont_query_response) == 0:1456 return {}1457 1458 return {1459 key: 1 for key in vCont_query_response.split(";") if key and len(key) > 01460 }1461 1462 def count_single_steps_until_true(1463 self,1464 thread_id,1465 predicate,1466 args,1467 max_step_count=100,1468 use_Hc_packet=True,1469 step_instruction="s",1470 ):1471 """Used by single step test that appears in a few different contexts."""1472 single_step_count = 01473 1474 while single_step_count < max_step_count:1475 self.assertIsNotNone(thread_id)1476 1477 # Build the packet for the single step instruction. We replace1478 # {thread}, if present, with the thread_id.1479 step_packet = "read packet: ${}#00".format(1480 re.sub(r"{thread}", "{:x}".format(thread_id), step_instruction)1481 )1482 # print("\nstep_packet created: {}\n".format(step_packet))1483 1484 # Single step.1485 self.reset_test_sequence()1486 if use_Hc_packet:1487 self.test_sequence.add_log_lines(1488 [ # Set the continue thread.1489 "read packet: $Hc{0:x}#00".format(thread_id),1490 "send packet: $OK#00",1491 ],1492 True,1493 )1494 self.test_sequence.add_log_lines(1495 [1496 # Single step.1497 step_packet,1498 # "read packet: $vCont;s:{0:x}#00".format(thread_id),1499 # Expect a breakpoint stop report.1500 {1501 "direction": "send",1502 "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);",1503 "capture": {1: "stop_signo", 2: "stop_thread_id"},1504 },1505 ],1506 True,1507 )1508 context = self.expect_gdbremote_sequence()1509 self.assertIsNotNone(context)1510 self.assertIsNotNone(context.get("stop_signo"))1511 self.assertEqual(1512 int(context.get("stop_signo"), 16),1513 lldbutil.get_signal_number("SIGTRAP"),1514 )1515 1516 single_step_count += 11517 1518 # See if the predicate is true. If so, we're done.1519 if predicate(args):1520 return (True, single_step_count)1521 1522 # The predicate didn't return true within the runaway step count.1523 return (False, single_step_count)1524 1525 def g_c1_c2_contents_are(self, args):1526 """Used by single step test that appears in a few different contexts."""1527 g_c1_address = args["g_c1_address"]1528 g_c2_address = args["g_c2_address"]1529 expected_g_c1 = args["expected_g_c1"]1530 expected_g_c2 = args["expected_g_c2"]1531 1532 # Read g_c1 and g_c2 contents.1533 self.reset_test_sequence()1534 self.test_sequence.add_log_lines(1535 [1536 "read packet: $m{0:x},{1:x}#00".format(g_c1_address, 1),1537 {1538 "direction": "send",1539 "regex": r"^\$(.+)#[0-9a-fA-F]{2}$",1540 "capture": {1: "g_c1_contents"},1541 },1542 "read packet: $m{0:x},{1:x}#00".format(g_c2_address, 1),1543 {1544 "direction": "send",1545 "regex": r"^\$(.+)#[0-9a-fA-F]{2}$",1546 "capture": {1: "g_c2_contents"},1547 },1548 ],1549 True,1550 )1551 1552 # Run the packet stream.1553 context = self.expect_gdbremote_sequence()1554 self.assertIsNotNone(context)1555 1556 # Check if what we read from inferior memory is what we are expecting.1557 self.assertIsNotNone(context.get("g_c1_contents"))1558 self.assertIsNotNone(context.get("g_c2_contents"))1559 1560 return (seven.unhexlify(context.get("g_c1_contents")) == expected_g_c1) and (1561 seven.unhexlify(context.get("g_c2_contents")) == expected_g_c21562 )1563 1564 def single_step_only_steps_one_instruction(1565 self, use_Hc_packet=True, step_instruction="s"1566 ):1567 """Used by single step test that appears in a few different contexts."""1568 # Start up the inferior.1569 procs = self.prep_debug_monitor_and_inferior(1570 inferior_args=[1571 "get-code-address-hex:swap_chars",1572 "get-data-address-hex:g_c1",1573 "get-data-address-hex:g_c2",1574 "sleep:1",1575 "call-function:swap_chars",1576 "sleep:5",1577 ]1578 )1579 1580 # Run the process1581 self.test_sequence.add_log_lines(1582 [ # Start running after initial stop.1583 "read packet: $c#63",1584 # Match output line that prints the memory address of the function call entry point.1585 # Note we require launch-only testing so we can get inferior otuput.1586 {1587 "type": "output_match",1588 "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$",1589 "capture": {1590 1: "function_address",1591 2: "g_c1_address",1592 3: "g_c2_address",1593 },1594 },1595 # Now stop the inferior.1596 "read packet: {}".format(chr(3)),1597 # And wait for the stop notification.1598 {1599 "direction": "send",1600 "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);",1601 "capture": {1: "stop_signo", 2: "stop_thread_id"},1602 },1603 ],1604 True,1605 )1606 1607 # Run the packet stream.1608 context = self.expect_gdbremote_sequence()1609 self.assertIsNotNone(context)1610 1611 # Grab the main thread id.1612 self.assertIsNotNone(context.get("stop_thread_id"))1613 main_thread_id = int(context.get("stop_thread_id"), 16)1614 1615 # Grab the function address.1616 self.assertIsNotNone(context.get("function_address"))1617 function_address = int(context.get("function_address"), 16)1618 1619 # Grab the data addresses.1620 self.assertIsNotNone(context.get("g_c1_address"))1621 g_c1_address = int(context.get("g_c1_address"), 16)1622 1623 self.assertIsNotNone(context.get("g_c2_address"))1624 g_c2_address = int(context.get("g_c2_address"), 16)1625 1626 # Set a breakpoint at the given address.1627 if self.getArchitecture().startswith("arm"):1628 # TODO: Handle case when setting breakpoint in thumb code1629 BREAKPOINT_KIND = 41630 else:1631 BREAKPOINT_KIND = 11632 self.reset_test_sequence()1633 self.add_set_breakpoint_packets(1634 function_address, do_continue=True, breakpoint_kind=BREAKPOINT_KIND1635 )1636 context = self.expect_gdbremote_sequence()1637 self.assertIsNotNone(context)1638 1639 # Remove the breakpoint.1640 self.reset_test_sequence()1641 self.add_remove_breakpoint_packets(1642 function_address, breakpoint_kind=BREAKPOINT_KIND1643 )1644 context = self.expect_gdbremote_sequence()1645 self.assertIsNotNone(context)1646 1647 # Verify g_c1 and g_c2 match expected initial state.1648 args = {}1649 args["g_c1_address"] = g_c1_address1650 args["g_c2_address"] = g_c2_address1651 args["expected_g_c1"] = "0"1652 args["expected_g_c2"] = "1"1653 1654 self.assertTrue(self.g_c1_c2_contents_are(args))1655 1656 # Verify we take only a small number of steps to hit the first state.1657 # Might need to work through function entry prologue code.1658 args["expected_g_c1"] = "1"1659 args["expected_g_c2"] = "1"1660 (state_reached, step_count) = self.count_single_steps_until_true(1661 main_thread_id,1662 self.g_c1_c2_contents_are,1663 args,1664 max_step_count=25,1665 use_Hc_packet=use_Hc_packet,1666 step_instruction=step_instruction,1667 )1668 self.assertTrue(state_reached)1669 1670 # Verify we hit the next state.1671 args["expected_g_c1"] = "1"1672 args["expected_g_c2"] = "0"1673 (state_reached, step_count) = self.count_single_steps_until_true(1674 main_thread_id,1675 self.g_c1_c2_contents_are,1676 args,1677 max_step_count=5,1678 use_Hc_packet=use_Hc_packet,1679 step_instruction=step_instruction,1680 )1681 self.assertTrue(state_reached)1682 expected_step_count = 11683 arch = self.getArchitecture()1684 1685 # MIPS required "3" (ADDIU, SB, LD) machine instructions for updation1686 # of variable value1687 if re.match("mips", arch):1688 expected_step_count = 31689 # S390X requires "2" (LARL, MVI) machine instructions for updation of1690 # variable value1691 if re.match("s390x", arch):1692 expected_step_count = 21693 # ARM64 requires "4" instructions: 2 to compute the address (adrp,1694 # add), one to materialize the constant (mov) and the store. Once1695 # addresses and constants are materialized, only one instruction is1696 # needed.1697 if re.match("arm64", arch):1698 before_materialization_step_count = 41699 after_matrialization_step_count = 11700 self.assertIn(1701 step_count,1702 [before_materialization_step_count, after_matrialization_step_count],1703 )1704 expected_step_count = after_matrialization_step_count1705 else:1706 self.assertEqual(step_count, expected_step_count)1707 1708 # Verify we hit the next state.1709 args["expected_g_c1"] = "0"1710 args["expected_g_c2"] = "0"1711 (state_reached, step_count) = self.count_single_steps_until_true(1712 main_thread_id,1713 self.g_c1_c2_contents_are,1714 args,1715 max_step_count=5,1716 use_Hc_packet=use_Hc_packet,1717 step_instruction=step_instruction,1718 )1719 self.assertTrue(state_reached)1720 self.assertEqual(step_count, expected_step_count)1721 1722 # Verify we hit the next state.1723 args["expected_g_c1"] = "0"1724 args["expected_g_c2"] = "1"1725 (state_reached, step_count) = self.count_single_steps_until_true(1726 main_thread_id,1727 self.g_c1_c2_contents_are,1728 args,1729 max_step_count=5,1730 use_Hc_packet=use_Hc_packet,1731 step_instruction=step_instruction,1732 )1733 self.assertTrue(state_reached)1734 self.assertEqual(step_count, expected_step_count)1735 1736 def maybe_strict_output_regex(self, regex):1737 return (1738 ".*" + regex + ".*"1739 if lldbplatformutil.hasChattyStderr(self)1740 else "^" + regex + "$"1741 )1742 1743 def install_and_create_launch_args(self):1744 exe_path = self.getBuildArtifact("a.out")1745 if not lldb.remote_platform:1746 return [exe_path]1747 remote_path = lldbutil.append_to_process_working_directory(1748 self, os.path.basename(exe_path)1749 )1750 remote_file_spec = lldb.SBFileSpec(remote_path, False)1751 err = lldb.remote_platform.Install(1752 lldb.SBFileSpec(exe_path, True), remote_file_spec1753 )1754 if err.Fail():1755 raise Exception(1756 "remote_platform.Install('%s', '%s') failed: %s"1757 % (exe_path, remote_path, err)1758 )1759 return [remote_path]1760