LLDB remote protocol

lldb.py

"""
Implements Apple's customized version of the GDB/LLDB remote protocol, intended for use with debugserver on iOS.
A macOS implementation of debugserver can be found here: https://github.com/swiftlang/llvm-project/blob/next/lldb/tools/debugserver/source/debugserver.cpp

Use the script tools/proxy.py to proxy the real lldb implementation to discover new commands.
"""
from anyio.abc import ByteStream

class GDBRemote:
    """
    Basic GDB remote protocol as described by https://ftp.gnu.org/old-gnu/Manuals/gdb/html_node/gdb_129.html
                                          and https://sourceware.org/gdb/current/onlinedocs/gdb.html/Packets.html
    """
    def __init__(self, socket: ByteStream) -> None:
        self.socket = socket
        self._non_stop = False

    def _checksum(self, data: bytes) -> str:
        """
        Calculate the checksum of the given data.
        The checksum is the sum of all bytes in the data, modulo 256.
        """
        return f"{sum(data) % 256:02x}"
    
    def _escape(self, data: bytes) -> bytes:
        """
        Escape special characters }, #, $, and * (optional)
        """
        data = data.replace(b"}", b"}" + (ord(b"}") ^ 0x20).to_bytes())
        data = data.replace(b"#", b"}" + (ord(b"#") ^ 0x20).to_bytes())
        data = data.replace(b"$", b"}" + (ord(b"$") ^ 0x20).to_bytes())
        data = data.replace(b"*", b"}" + (ord(b"*") ^ 0x20).to_bytes())
        return data
    
    def _unescape(self, data: bytes) -> bytes:
        """
        Unescape anything starting with a }.
        Also should unescape *, which is runs of characters
        """
        if "*" in data.decode():
            raise ValueError("Unescaping * is not supported")
        data = data.replace(b"}" + (ord(b"}") ^ 0x20).to_bytes(), b"}")
        data = data.replace(b"}" + (ord(b"#") ^ 0x20).to_bytes(), b"#")
        data = data.replace(b"}" + (ord(b"$") ^ 0x20).to_bytes(), b"$")
        data = data.replace(b"}" + (ord(b"*") ^ 0x20).to_bytes(), b"*")
        return data
                
    def _packet(self, data: bytes) -> bytes:
        """
        Create a GDB remote packet.
        The packet format is: $<data>#<checksum>
        """
        checksum = self._checksum(data)
        return f"${self._escape(data).decode()}#{checksum}".encode()
    
    def _unpacket(self, packet: bytes) -> bytes:
        """
        Unpack a GDB remote packet.
        The packet format is: $<data>#<checksum>
        """
        if not packet.startswith(b"$") or b"#" not in packet:
            raise ValueError("Invalid packet format")
        
        data, checksum = packet[1:].split(b"#", 1)
        if self._checksum(data) != checksum.decode():
            raise ValueError("Checksum mismatch")
        
        return self._unescape(data)
    
    async def send(self, data: bytes, _require_ack = True, _response = True) -> bytes:
        """
        Send a GDB remote command.
        """
        packet = self._packet(data)
        await self.socket.send(packet)

        if _require_ack:
            ack = await self.socket.receive(1)
            if ack != b"+":
                raise ValueError("ACK not received")
        
        if _response:
            response = await self.socket.receive()
            if response.startswith(b"$"):
                while True:
                    if b"#" in response:
                        break
                    response += await self.socket.receive()
                resp = self._unpacket(response)
            else:
                raise ValueError("Invalid response format", response)
        else:
            resp = None
        
        if _require_ack:
            await self.socket.send(b"+")
        
        return resp
    
    async def continue_(self) -> None:
        """
        Continue the process.
        """
        await self.send(b"c", _response=False)
    
    async def interrupt(self) -> bytes:
        """
        Interrupt the process. (Sends Ctrl-C)
        """
        resp = await self.send(b"\x03")
        return resp

class LLDBRemote(GDBRemote):
    """
    LLDB remote protocol as described by https://github.com/llvm/llvm-project/blob/main/lldb/docs/resources/lldbgdbremote.md
                                     and https://github.com/llvm-mirror/lldb/blob/master/docs/lldb-gdb-remote.txt
    """
    def __init__(self, socket: ByteStream) -> None:
        super().__init__(socket)
        self._no_ack = False

    def _checksum(self, data: bytes) -> str:
        if self._no_ack:
            return "00" # In no-ack mode, the checksum is not used
        return super()._checksum(data)
    
    async def send(self, data: bytes, _response = True) -> bytes:
        """
        Send a LLDB remote command.
        """
        return await super().send(data, _require_ack=not self._no_ack, _response=_response)

    async def start_no_ack(self) -> None:
        """
        Start no-ack mode.
        """
        resp = await self.send(b"QStartNoAckMode")
        if resp != b"OK":
            raise ValueError("Failed to start no-ack mode")
        self._no_ack = True

    async def attach_by_name(self, name: str, wait: bool = False, include_existing: bool = False) -> bytes:
        """
        Attach to a process by name.
        """
        if wait and include_existing:
            resp = await self.send(f"vAttachOrWait;{name.encode().hex()}".encode())
        elif wait:
            resp = await self.send(f"qAttachWait:{name.encode().hex()}".encode())
        else:
            resp = await self.send(f"qAttachByName:{name}".encode())
        
        return resp
    
    async def memory_region_info(self, address: int) -> bytes:
        """
        Get memory region info for a given address.
        """
        resp = await self.send(f"qMemoryRegionInfo:{address:x}".encode())
        return resp

    async def dynamic_library_info(self) -> bytes:
        """
        Get dynamic library info.
        """
        resp = await self.send(b'jGetLoadedDynamicLibrariesInfos:{"fetch_all_solibs":true}')
        return resp