path: root/lldb/packages/Python/lldbsuite/test/lldbreverse.py

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")