Debug Control

def load_executable(self, target_exe: str, cmdline: str = "", current_dir: str = "", wait_timeout: int = 10) -> bool:
    """
    Loads a new executable into the debugger. This method will block until the debugee is ready to receive a command.

    Args:
        target_exe: Path to the executable to load
        cmdline: Command line arguments to pass to the executable
        current_dir: Current working directory for the executable
        wait_timeout: Max time to wait for the debugee to be ready

    Returns:
        True if successful, False otherwise
    """
    cmdline = cmdline.replace('"', r'\"')
    current_dir = current_dir.replace('"', r'\"')
    if len(current_dir) == 0:
        current_dir = "."
    if current_dir == "." or current_dir == "":
        target_exe = os.path.abspath(target_exe)
    else:
        target_exe = os.path.abspath(os.path.join(current_dir, target_exe))
    if not self.cmd_sync(f'init {target_exe}, "{cmdline}", "{current_dir}"'):
        return False
    return self.wait_cmd_ready(wait_timeout)

def unload_executable(self, wait_timeout: int = 10) -> bool:
    """
    Unloads the currently loaded executable. This method will block until the debugger is no longer debugging.

    Args:
        wait_timeout: Max time to wait for the debugger finish unloading

    Returns:
        True if successful, False otherwise
    """
    if not self.cmd_sync(f'stop'):
        return False
    return self.wait_until_not_debugging(wait_timeout)

def attach(self, pid: int, wait_timeout: int = 10) -> bool:
    """
    Attaches the debugger to a running process. This method does *NOT* block until a breakpoint has been reached.

    Args:
        pid (int): Process Identifier (PID) of the running process.
        wait_timeout (int): Maximum time (in seconds) to wait for the debugger to be ready.

    Returns:
        bool: True if the debugger attaches successfully, False otherwise.
    """
    if not self.cmd_sync(f"attach 0x{pid:x}"):
        return False
    return self.wait_until_debugging(wait_timeout)

def detach(self, wait_timeout: int = 10) -> bool:
    """
    Detaches the debugger from the currently debugged process.
    This method will block until we are no longer debugging.

    Args:
        wait_timeout (int): Maximum time (in seconds) to wait for the debugger to be ready.

    Returns:
        bool: True if the debugger detaches successfully, False otherwise.
    """
    if not self.cmd_sync("detach"):
        return False
    return self.wait_until_not_debugging(wait_timeout)

def go(self, pass_exceptions: bool = False, swallow_exceptions: bool = False) -> bool:
    """
    Resumes the debugee. This method will block until the debugee is in the running state.

    Args:
        pass_exceptions: Pass exceptions to the debugee
        swallow_exceptions: Swallow exceptions

    Returns:
        True if successful, False otherwise
    """
    if pass_exceptions == True and swallow_exceptions == True:
        raise ValueError("Cannot pass and swallow exceptions at the same time")
    if pass_exceptions:
        prefix = 'e'
    elif swallow_exceptions:
        prefix = 'se'
    else:
        prefix = ''
    ip_start = self.get_reg('cip')
    if not self.cmd_sync(f"{prefix}go"):
        return False
    if self.get_reg('cip') == ip_start:
        self.wait_until_running(timeout=1)
    return True

def pause(self) -> bool:
    """
    Pauses the debugee. This method will block until the debugee is in the stopped state.

    Returns:
        True if successful, False otherwise
    """
    if not self.cmd_sync(f"pause"):
        return False
    return self.wait_until_stopped()

def stepi(self, step_count: int = 1, pass_exceptions: bool = False, swallow_exceptions: bool = False, wait_for_ready: bool = True, wait_timeout: int = 2) -> bool:
    """
    Steps into N instructions.

    Args:
        step_count: Number of instructions to step through
        pass_exceptions: Pass exceptions to the debugee during step
        swallow_exceptions: Swallow exceptions during step
        wait_for_ready: Block until debugger is stopped
        wait_timeout: Maximum time in seconds to wait for debugger to stop
    Returns:
        bool: True if stepping operation was successful, False otherwise.
    Raises:
        ValueError: If both pass_exceptions and swallow_exceptions are True.
    """
    if pass_exceptions == True and swallow_exceptions == True:
        raise ValueError("Cannot pass and swallow exceptions at the same time")
    if pass_exceptions:
        prefix = 'e'
    elif swallow_exceptions:
        prefix = 'se'
    else:
        prefix = ''
    res = self.cmd_sync(f"{prefix}sti 0x{step_count:x}")
    if res and wait_for_ready:
        self.wait_until_stopped(wait_timeout)
    return res

def stepo(self, step_count: int = 1, pass_exceptions: bool = False, swallow_exceptions: bool = False, wait_for_ready: bool = True, wait_timeout: int = 2) -> bool:
    """
    Steps over N instructions.

    Args:
        step_count: Number of instructions to step through
        pass_exceptions: Pass exceptions to the debugee during step
        swallow_exceptions: Swallow exceptions during step
        wait_for_ready: Block until debugger is stopped
        wait_timeout: Maximum time in seconds to wait for debugger to stop
    Returns:
        bool: True if stepping operation was successful, False otherwise.
    Raises:
        ValueError: If both pass_exceptions and swallow_exceptions are True.
    """
    if pass_exceptions == True and swallow_exceptions == True:
        raise ValueError("Cannot pass and swallow exceptions at the same time")
    if pass_exceptions:
        prefix = 'e'
    elif swallow_exceptions:
        prefix = 'se'
    else:
        prefix = ''
    res = self.cmd_sync(f"{prefix}sto 0x{step_count:x}")
    if res and wait_for_ready:
        self.wait_until_stopped(wait_timeout)
    return res

def skip(self, skip_count: int = 1, wait_for_ready: bool = True, wait_timeout: int = 2) -> bool:
    """
    Skips over N instructions.

    Args:
        skip_count: Number of instructions to skip
        wait_for_ready: Block until debugger is stopped
        wait_timeout: Maximum time in seconds to wait for debugger to stop
    Returns:
        bool: True if stepping operation was successful, False otherwise.
    """
    res = self.cmd_sync(f"skip {skip_count}")
    if res and wait_for_ready:
        self.wait_until_stopped(wait_timeout)
    return res

def ret(self, frames: int = 1, wait_timeout: int = 10) -> bool:
    """
    Steps until a ret instruction is encountered.

    Args:
        frames: Number of ret instructions to seek
        wait_timeout: Maximum time in seconds to wait for debugger to stop
    Returns:
        bool: True if stepping operation was successful, False otherwise.
    """
    if not self.cmd_sync(f"rtr {frames}"):
        return False
    return self.wait_cmd_ready(wait_timeout)

def thread_create(self, addr: int, arg: int = 0) -> int | None:
    """
    Create a new thread in the debugee.

    Args:
        addr: Address of the thread entry point
        arg: Argument to pass to the thread
    """
    success = self.cmd_sync(f'createthread 0x{addr:x}, 0x{arg:x}')
    if not success:
        return None
    tid, success = self.eval_sync('$result')
    if not success:
        return None
    return tid

def thread_terminate(self, tid: int):
    """
    Kills a thread in the debugee.

    Args:
        tid: Thread ID to kill
    """
    return self.cmd_sync(f'killthread 0x{tid:x}')

def thread_pause(self, tid: int):
    """
    Pauses a thread in the debugee.

    Args:
        tid: Thread ID to kill
    """
    return self.cmd_sync(f'suspendthread 0x{tid:x}')

def thread_resume(self, tid: int):
    """
    Resumes a thread in the debugee.

    Args:
        tid: Thread ID to kill
    """
    return self.cmd_sync(f'resumethread 0x{tid:x}')

def switch_thread(self, tid: int):
    """
    Switches the currently observed debugger thread.

    Args:
        tid: Thread ID to kill
    """
    return self.cmd_sync(f'switchthread 0x{tid:x}')

def wait_for_debug_event(self, event_type: EventType, timeout: int = 5) -> DbgEvent | None:
    """
    Wait for a debug event of a specific type to occur. This method returns the latest event of the specified type, which may have occurred before the method was called.

    Returned events are removed from the queue. If the event has not occurred within the timeout, None is returned.

    `clear_debug_events` can be used to ensure an empty debug event queue before calling this method.

    Args:
        event_type: The type of event to wait for
        timeout: The maximum time to wait for the event in seconds

    Returns:
        DbgEvent | None: The latest event of the specified type, or None if the event did not occur within the timeout.
    """
    while timeout > 0:
        for ix, event in enumerate(self._debug_events_q):
            if event.event_type == event_type:
                return self._debug_events_q.pop(ix)
        time.sleep(0.25)
        timeout -= 0.25
    return None

def clear_debug_events(self, event_type: EventType | None = None) -> None:
    """
    Clear the debug event queue. If `event_type` is specified, only events of that type will be removed.

    Args:
        event_type: The type of event to clear. If None, all events will be cleared.
    """
    filtered = []
    for _ in range(len(self._debug_events_q)):
        event = self._debug_events_q.pop(0)
        if event.event_type != event_type and event_type is not None:
            filtered.append(event)
    self._debug_events_q = filtered

def watch_debug_event(self, event_type: EventType, callback: callable):
    """
    Register a callback to be invoked when a debug event of a specific type occurs.

    Args:
        event_type: The type of event to watch
        callback: The callback to invoke when the event occurs. The callback should accept a single argument of type `DbgEvent`.
    """
    self.listeners[event_type] = self.listeners.get(event_type, []) + [callback]

def unwatch_debug_event(self, event_type: EventType, callback: callable):
    """
    Remove a callback registered with `watch_debug_event`

    Args:
        event_type: The type of event to unwatch
        callback: The callback instance to remove
    """
    self.listeners[event_type] = [x for x in self.listeners.get(event_type, []) if x != callback]

def get_latest_debug_event(self) -> DbgEvent | None:
    """
    Get the latest debug event that occurred in the debugee. The event is removed from the queue.
    """
    if len(self._debug_events_q) == 0:
        return None
    return self._debug_events_q.pop()

def peek_latest_debug_event(self) -> DbgEvent | None:
    """
    Get the latest debug event that occurred in the debugee. The event is not removed from the queue.
    """
    if len(self._debug_events_q) == 0:
        return None
    return self._debug_events_q[-1]

def wait_until_debugging(self, timeout: int = 10) -> bool:
    """
    Blocks until the debugger enters a debugging state

    Args:
        timeout: The maximum time to wait in seconds

    Returns:
        True if the debugger is debugging, False otherwise
    """
    slept = 0
    while True:
        if self.is_debugging():
            return True
        time.sleep(0.2)
        slept += 0.2
        if slept >= timeout:
            return False

def wait_until_not_debugging(self, timeout: int = 10) -> bool:
    """
    Blocks until the debugger enters a not-debugging state

    Args:
        timeout: The maximum time to wait in seconds

    Returns:
        True if the debugger is not-debugging, False otherwise
    """
    slept = 0
    while True:
        if not self.is_debugging():
            return True
        time.sleep(0.2)
        slept += 0.2
        if slept >= timeout:
            return False

def wait_until_running(self, timeout: int = 10) -> bool:
    """
    Blocks until the debugger enters a running state

    Args:
        timeout: The maximum time to wait in seconds

    Returns:
        True if the debugger is running, False otherwise
    """
    slept = 0
    while True:
        if self.is_running():
            return True
        time.sleep(0.08)
        slept += 0.08
        if slept >= timeout:
            return False

def wait_until_stopped(self, timeout: int = 10) -> bool:
    """
    Blocks until the debugger enters a stopped state

    Args:
        timeout: The maximum time to wait in seconds

    Returns:
        True if the debugger is stopped, False otherwise
    """
    slept = 0
    while True:
        if not self.is_running() or not self.is_debugging():
            return True
        time.sleep(0.2)
        slept += 0.2
        if slept >= timeout:
            return False

def wait_cmd_ready(self, timeout: int = 10) -> bool:
    """
    Blocks until the debugger is ready to accept debug control commands (debugging + stopped)

    Args:
        timeout: The maximum time to wait in seconds

    Returns:
        True if the debugger is ready, False otherwise
    """
    return self.wait_until_debugging(timeout) and self.wait_until_stopped(timeout)

def hide_debugger_peb(self) -> bool:
    """
    Hides the debugger in the debugee's PEB

    Returns:
        Success
    """
    return self.cmd_sync(f'hide')

def debugee_pid(self) -> int | None:
    """
    Retrieves the PID of the debugee

    Returns:
        PID of the debugee, or None if the debugger is not debugging
    """
    if self.is_debugging():
        pid, res = self.eval_sync(f'pid')
        if res:
            return pid
    return None

def get_debugger_pid(self) -> int:
    """
    Retrieves the PID of the x64dbg process

    Returns:
        The PID of the x64dbg process
    """
    return self._send_request(XAutoCommand.XAUTO_REQ_DEBUGGER_PID)

def is_running(self) -> bool:
    """
    Checks if the debugee's state is "running"

    Returns:
        True if the debugee is running, False otherwise
    """
    return self._send_request(XAutoCommand.XAUTO_REQ_DBG_IS_RUNNING)

def is_debugging(self) -> bool:
    """
    Checks if the debugee's state is "debugging"

    Returns:
        True if the debugee is running, False otherwise
    """
    return self._send_request(XAutoCommand.XAUTO_REQ_DBG_IS_DEBUGGING)