import os import os.path import lldb from lldbsuite.test.lldbtest import * from lldbsuite.test.gdbclientutils import * from lldbsuite.test.lldbgdbproxy import * import lldbgdbserverutils import re class ThreadSnapshot: def __init__(self, thread_id, registers): self.thread_id = thread_id self.registers = registers class MemoryBlockSnapshot: def __init__(self, address, data): self.address = address self.data = data class StateSnapshot: def __init__(self, thread_snapshots, memory): self.thread_snapshots = thread_snapshots self.memory = memory self.thread_id = None class RegisterInfo: def __init__(self, lldb_index, name, bitsize, little_endian): self.lldb_index = lldb_index self.name = name self.bitsize = bitsize self.little_endian = little_endian BELOW_STACK_POINTER = 16384 ABOVE_STACK_POINTER = 4096 BLOCK_SIZE = 1024 SOFTWARE_BREAKPOINTS = 0 HARDWARE_BREAKPOINTS = 1 WRITE_WATCHPOINTS = 2 class ReverseTestBase(GDBProxyTestBase): """ Base class for tests that need reverse execution. This class uses a gdbserver proxy to add very limited reverse- execution capability to lldb-server/debugserver for testing purposes only. To use this class, run the inferior forward until some stopping point. Then call `start_recording()` and execute forward again until reaching a software breakpoint; this class records the state before each execution executes. At that point, the server will accept "bc" and "bs" packets to step backwards through the state. When executing during recording, we only allow single-step and continue without delivering a signal, and only software breakpoint stops are allowed. We assume that while recording is enabled, the only effects of instructions are on general-purpose registers (read/written by the 'g' and 'G' packets) and on memory bytes between [SP - BELOW_STACK_POINTER, SP + ABOVE_STACK_POINTER). """ NO_DEBUG_INFO_TESTCASE = True """ A list of StateSnapshots in time order. There is one snapshot per single-stepped instruction, representing the state before that instruction was executed. The last snapshot in the list is the snapshot before the last instruction was executed. This is an undo log; we snapshot a superset of the state that may have been changed by the instruction's execution. """ snapshots = None recording_enabled = False breakpoints = None pc_register_info = None sp_register_info = None general_purpose_register_info = None def __init__(self, *args, **kwargs): GDBProxyTestBase.__init__(self, *args, **kwargs) self.breakpoints = [set(), set(), set(), set(), set()] def respond(self, packet): if not packet: raise ValueError("Invalid empty packet") if packet == self.server.PACKET_INTERRUPT: # Don't send a response. We'll just run to completion. return [] if self.is_command(packet, "qSupported", ":"): # Disable multiprocess support in the server and in LLDB # since Mac debugserver doesn't support it and we want lldb-server to # be consistent with that reply = self.pass_through(packet.replace(";multiprocess", "")) return reply.replace(";multiprocess", "") + ";ReverseStep+;ReverseContinue+" if packet == "c" or packet == "s": packet = "vCont;" + packet elif ( packet[0] == "c" or packet[0] == "s" or packet[0] == "C" or packet[0] == "S" ): raise ValueError( "Old-style continuation packets with address or signal not supported yet" ) if self.is_command(packet, "vCont", ";"): if self.recording_enabled: return self.continue_with_recording(packet) snapshots = [] if packet == "bc": return self.reverse_continue() if packet == "bs": return self.reverse_step() if packet == "jThreadsInfo": # Suppress this because it contains thread stop reasons which we might # need to modify, and we don't want to have to implement that. return "" if packet[0] == "x": # Suppress *binary* reads as results starting with "O" can be mistaken for an output packet # by the test server code return "" if packet[0] == "z" or packet[0] == "Z": reply = self.pass_through(packet) if reply == "OK": self.update_breakpoints(packet) return reply return GDBProxyTestBase.respond(self, packet) def start_recording(self): self.recording_enabled = True self.snapshots = [] def stop_recording(self): """ Don't record when executing foward. Reverse execution is still supported until the next forward continue. """ self.recording_enabled = False def is_command(self, packet, cmd, follow_token): return packet == cmd or packet[0 : len(cmd) + 1] == cmd + follow_token def update_breakpoints(self, packet): m = re.match("([zZ])([01234]),([0-9a-f]+),([0-9a-f]+)", packet) if m is None: raise ValueError("Invalid breakpoint packet: " + packet) t = int(m.group(2)) addr = int(m.group(3), 16) kind = int(m.group(4), 16) if m.group(1) == "Z": self.breakpoints[t].add((addr, kind)) else: self.breakpoints[t].discard((addr, kind)) def breakpoint_triggered_at(self, pc): if any(addr == pc for addr, kind in self.breakpoints[SOFTWARE_BREAKPOINTS]): return True if any(addr == pc for addr, kind in self.breakpoints[HARDWARE_BREAKPOINTS]): return True return False def watchpoint_triggered(self, new_value_block, current_contents): """Returns the address or None.""" for watch_addr, kind in self.breakpoints[WRITE_WATCHPOINTS]: for offset in range(0, kind): addr = watch_addr + offset if ( addr >= new_value_block.address and addr < new_value_block.address + len(new_value_block.data) ): index = addr - new_value_block.address if ( new_value_block.data[index * 2 : (index + 1) * 2] != current_contents[index * 2 : (index + 1) * 2] ): return watch_addr return None def continue_with_recording(self, packet): self.logger.debug("Continue with recording enabled") step_packet = "vCont;s" if packet == "vCont": requested_step = False else: m = re.match("vCont;(c|s)(.*)", packet) if m is None: raise ValueError("Unsupported vCont packet: " + packet) requested_step = m.group(1) == "s" step_packet += m.group(2) while True: snapshot = self.capture_snapshot() reply = self.pass_through(step_packet) (stop_signal, stop_pairs) = self.parse_stop_reply(reply) if stop_signal != 5: raise ValueError("Unexpected stop signal: " + reply) is_swbreak = False thread_id = None for key, value in stop_pairs.items(): if key == "thread": thread_id = self.parse_thread_id(value) continue if re.match("[0-9a-f]+", key): continue if key == "swbreak" or (key == "reason" and value == "breakpoint"): is_swbreak = True continue if key == "metype": reason = self.stop_reason_from_mach_exception(stop_pairs) if reason == "breakpoint": is_swbreak = True elif reason != "singlestep": raise ValueError(f"Unsupported stop reason in {reply}") continue if key in [ "name", "threads", "thread-pcs", "reason", "mecount", "medata", "memory", ]: continue raise ValueError(f"Unknown stop key '{key}' in {reply}") if is_swbreak: self.logger.debug("Recording stopped") return reply if thread_id is None: return ValueError("Expected thread ID: " + reply) snapshot.thread_id = thread_id self.snapshots.append(snapshot) if requested_step: self.logger.debug("Recording stopped for step") return reply def stop_reason_from_mach_exception(self, stop_pairs): # See StopInfoMachException::CreateStopReasonWithMachException. if int(stop_pairs["metype"]) != 6: # EXC_BREAKPOINT raise ValueError(f"Unsupported exception type {value} in {reply}") medata = stop_pairs["medata"] arch = self.getArchitecture() if arch in ["amd64", "i386", "x86_64"]: if int(medata[0], 16) == 2: return "breakpoint" if int(medata[0], 16) == 1 and int(medata[1], 16) == 0: return "singlestep" elif arch in ["arm64", "arm64e"]: if int(medata[0], 16) == 1 and int(medata[1], 16) != 0: return "breakpoint" elif int(medata[0], 16) == 1 and int(medata[1], 16) == 0: return "singlestep" else: raise ValueError(f"Unsupported architecture '{arch}'") raise ValueError(f"Unsupported exception details in {reply}") def parse_stop_reply(self, reply): if not reply: raise ValueError("Invalid empty packet") if reply[0] == "T" and len(reply) >= 3: result = {} for k, v in self.parse_pairs(reply[3:]): if k in ["medata", "memory"]: if k in result: result[k].append(v) else: result[k] = [v] else: result[k] = v return (int(reply[1:3], 16), result) raise ValueError("Unsupported stop reply: " + reply) def parse_pairs(self, text): for pair in text.split(";"): if not pair: continue m = re.match("([^:]+):(.*)", pair) if m is None: raise ValueError("Invalid pair text: " + text) yield (m.group(1), m.group(2)) def capture_snapshot(self): """Snapshot all threads and their stack memories.""" self.ensure_register_info() current_thread = self.get_current_thread() thread_snapshots = [] memory = [] for thread_id in self.get_thread_list(): registers = {} for index in sorted(self.general_purpose_register_info.keys()): reply = self.pass_through(f"p{index:x};thread:{thread_id:x};") if reply == "" or reply[0] == "E": # Mac debugserver tells us about registers that it won't let # us actually read. Ignore those registers. self.logger.debug(f"Failed to read register {index:x}") continue registers[index] = reply thread_snapshot = ThreadSnapshot(thread_id, registers) thread_sp = self.get_register( self.sp_register_info, thread_snapshot.registers ) # The memory above or below the stack pointer may be mapped, but not # both readable and writeable. For example on Arm 32-bit Linux, there # is a "[vectors]" mapping above the stack, which can be read but not # written to. # # Therefore, we should limit any reads to the stack region, which we # know is readable and writeable. region_info = self.get_memory_region_info(thread_sp) lower = max(thread_sp - BELOW_STACK_POINTER, region_info["start"]) upper = min( thread_sp + ABOVE_STACK_POINTER, region_info["start"] + region_info["size"], ) memory += self.read_memory(lower, upper) thread_snapshots.append(thread_snapshot) self.set_current_thread(current_thread) return StateSnapshot(thread_snapshots, memory) def restore_snapshot(self, snapshot): """ Restore the snapshot during reverse execution. If this triggers a breakpoint or watchpoint, return the stop reply, otherwise None. """ current_thread = self.get_current_thread() stop_reasons = [] for thread_snapshot in snapshot.thread_snapshots: thread_id = thread_snapshot.thread_id for lldb_index in sorted(thread_snapshot.registers.keys()): data = thread_snapshot.registers[lldb_index] reply = self.pass_through( f"P{lldb_index:x}={data};thread:{thread_id:x};" ) if reply != "OK": try: reg_name = self.general_purpose_register_info[lldb_index].name except KeyError: reg_name = f"with index {lldb_index}" raise ValueError(f"Can't restore thread register {reg_name}") if thread_id == snapshot.thread_id: new_pc = self.get_register( self.pc_register_info, thread_snapshot.registers ) if self.breakpoint_triggered_at(new_pc): stop_reasons.append([("reason", "breakpoint")]) self.set_current_thread(current_thread) for block in snapshot.memory: current_memory = self.pass_through( f"m{block.address:x},{(len(block.data)//2):x}" ) if not current_memory or current_memory[0] == "E": raise ValueError("Can't read back memory") reply = self.pass_through( f"M{block.address:x},{len(block.data)//2:x}:" + block.data ) if reply != "OK": raise ValueError( f"Can't restore memory block ranging from 0x{block.address:x} to 0x{block.address+len(block.data):x}." ) watch_addr = self.watchpoint_triggered(block, current_memory) if watch_addr is not None: stop_reasons.append( [("reason", "watchpoint"), ("watch", f"{watch_addr:x}")] ) if stop_reasons: pairs = ";".join(f"{key}:{value}" for key, value in stop_reasons[0]) return f"T05thread:{snapshot.thread_id:x};{pairs};" return None def reverse_step(self): if not self.snapshots: self.logger.debug("Reverse-step at history boundary") return self.history_boundary_reply(self.get_current_thread()) self.logger.debug("Reverse-step started") snapshot = self.snapshots.pop() stop_reply = self.restore_snapshot(snapshot) self.set_current_thread(snapshot.thread_id) self.logger.debug("Reverse-step stopped") if stop_reply is None: return self.singlestep_stop_reply(snapshot.thread_id) return stop_reply def reverse_continue(self): self.logger.debug("Reverse-continue started") thread_id = None while self.snapshots: snapshot = self.snapshots.pop() stop_reply = self.restore_snapshot(snapshot) thread_id = snapshot.thread_id if stop_reply is not None: self.set_current_thread(thread_id) self.logger.debug("Reverse-continue stopped") return stop_reply if thread_id is None: thread_id = self.get_current_thread() else: self.set_current_thread(snapshot.thread_id) self.logger.debug("Reverse-continue stopped at history boundary") return self.history_boundary_reply(thread_id) def get_current_thread(self): reply = self.pass_through("qC") return self.parse_thread_id(reply[2:]) def parse_thread_id(self, thread_id): m = re.match("([0-9a-f]+)", thread_id) if m is None: raise ValueError("Invalid thread ID: " + thread_id) return int(m.group(1), 16) def history_boundary_reply(self, thread_id): return f"T00thread:{thread_id:x};replaylog:begin;" def singlestep_stop_reply(self, thread_id): return f"T05thread:{thread_id:x};" def set_current_thread(self, thread_id): """ Set current thread in inner gdbserver. """ if thread_id >= 0: self.pass_through(f"Hg{thread_id:x}") self.pass_through(f"Hc{thread_id:x}") else: self.pass_through(f"Hc-1") self.pass_through(f"Hg-1") def get_register(self, register_info, registers): if register_info.bitsize % 8 != 0: raise ValueError("Register size must be a multiple of 8 bits") if register_info.lldb_index not in registers: raise ValueError("Register value not captured") data = registers[register_info.lldb_index] num_bytes = register_info.bitsize // 8 bytes = [] for i in range(0, num_bytes): bytes.append(int(data[i * 2 : (i + 1) * 2], 16)) if register_info.little_endian: bytes.reverse() result = 0 for byte in bytes: result = (result << 8) + byte return result def get_memory_region_info(self, addr): reply = self.pass_through(f"qMemoryRegionInfo:{addr:x}") if not reply or reply[0] == "E": raise RuntimeError("Failed to get memory region info.") # Valid reply looks like: # start:fffcf000;size:21000;permissions:rw;flags:;name:5b737461636b5d; values = [v for v in reply.strip().split(";") if v] region_info = {} for value in values: key, value = value.split( ":", ) region_info[key] = value if not ("start" in region_info and "size" in region_info): raise RuntimeError("Did not get extent of memory region.") region_info["start"] = int(region_info["start"], 16) region_info["size"] = int(region_info["size"], 16) return region_info def read_memory(self, start_addr, end_addr): """ Read a region of memory from the target. Some of the addresses may extend into memory we cannot read, skip those. Return a list of blocks containing the valid area(s) in the requested range. """ regions = [] start_addr = start_addr - (start_addr % BLOCK_SIZE) if end_addr % BLOCK_SIZE > 0: end_addr = end_addr - (end_addr % BLOCK_SIZE) + BLOCK_SIZE for addr in range(start_addr, end_addr, BLOCK_SIZE): reply = self.pass_through(f"m{addr:x},{(BLOCK_SIZE - 1):x}") if reply and reply[0] != "E": block = MemoryBlockSnapshot(addr, reply) regions.append(block) return regions def ensure_register_info(self): if self.general_purpose_register_info is not None: return reply = self.pass_through("qHostInfo") little_endian = any( kv == ("endian", "little") for kv in self.parse_pairs(reply) ) self.general_purpose_register_info = {} lldb_index = 0 while True: reply = self.pass_through(f"qRegisterInfo{lldb_index:x}") if not reply or reply[0] == "E": break info = {k: v for k, v in self.parse_pairs(reply)} reg_info = RegisterInfo( lldb_index, info["name"], int(info["bitsize"]), little_endian ) if ( info["set"] == "General Purpose Registers" and not "container-regs" in info ): self.general_purpose_register_info[lldb_index] = reg_info if "generic" in info: if info["generic"] == "pc": self.pc_register_info = reg_info elif info["generic"] == "sp": self.sp_register_info = reg_info lldb_index += 1 if self.pc_register_info is None or self.sp_register_info is None: raise ValueError("Can't find generic pc or sp register") def get_thread_list(self): threads = [] reply = self.pass_through("qfThreadInfo") while True: if not reply: raise ValueError("Missing reply packet") if reply[0] == "m": for id in reply[1:].split(","): threads.append(self.parse_thread_id(id)) elif reply[0] == "l": return threads reply = self.pass_through("qsThreadInfo")