use rmcp::{
ErrorData as McpError, ServiceExt,
handler::server::router::tool::ToolRouter,
handler::server::wrapper::Parameters,
model::{CallToolResult, Implementation, ProtocolVersion, ServerCapabilities, ServerInfo},
tool, tool_handler, tool_router,
transport::stdio,
};
use serde::Deserialize;
use serde_json::Value;
use tokio::sync::{mpsc, oneshot};
use tokio_util::sync::CancellationToken;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use crate::backend::MemoryOps;
use crate::bugchecks::{analyze_bugcheck, current_bugcheck};
use crate::dbg_backend::DebugBackend;
use crate::error::Error;
use crate::expr::Expr;
use crate::gdb::GdbClient;
use crate::kd::KdBackend;
use crate::memory_backend::MemoryBackend;
use crate::session::{ContinueOutcome, Session};
use crate::symbols::{FieldValue, TypeInfo};
use crate::target::{kthread_state_name, wait_reason_name};
use crate::types::VirtAddr;
use crate::view;
// MCP handlers are async/Send, but `Session` is not. A dedicated actor thread
// owns the session; handlers send closures and await one reply.
/// A unit of work run on the actor thread that owns the (`!Send`) session.
/// Returns JSON on success or a display string on error.
type SessionJob = Box<dyn FnOnce(&mut Session) -> Result<Value, ToolError> + Send>;
/// What the async handlers ask the actor to do.
enum Command {
/// Run a job against the session and reply with its result.
Run {
job: SessionJob,
reply: oneshot::Sender<Result<Value, ToolError>>,
},
/// Clean up the session (remove our breakpoints, resume the VM) and stop the
/// actor. `ack` fires once cleanup is done so the caller can exit knowing the
/// guest isn't left frozen with `int3`s installed.
Shutdown { ack: oneshot::Sender<()> },
/// A periodic nudge (from the background ticker) for the actor to service the
/// guest while otherwise idle; absorb wrong-process hits on a shared-page
/// breakpoint so they don't leave it frozen between tool calls. Carries no
/// reply; the actor just runs `Session::service_idle`.
Service,
}
/// How often the background ticker nudges the actor to service the guest (see
/// [`Command::Service`]). Small enough that a wrong-process breakpoint hit is
/// absorbed within a frame, large enough that the idle no-op is negligible.
const SERVICE_TICK: Duration = Duration::from_millis(20);
/// Tear down so the guest isn't left frozen at a stop with our `int3`s patched
/// in: remove breakpoints and leave the VM running (the halt-then-resume detail
/// lives in `Session::cleanup_for_exit`).
fn cleanup_session(ctx: &mut Session) {
let _ = ctx.cleanup_for_exit();
}
/// Spawn the actor thread that owns the (`!Send`) debugging session. The backend
/// is constructed *on this thread* so the non-`Send` state never crosses a
/// thread boundary. Returns the sender the MCP handlers use to reach it.
fn spawn_session(
backend: String,
connect: Option<String>,
) -> anyhow::Result<(mpsc::UnboundedSender<Command>, Arc<AtomicBool>)> {
let (ready_tx, ready_rx) = std::sync::mpsc::channel::<Result<(), String>>();
let (tx, mut rx) = mpsc::unbounded_channel::<Command>();
// Coalesces the background `Service` nudges: the ticker only enqueues one when
// this is false (and sets it), the actor clears it as it services. Caps queued
// `Service` commands at one, so a long `wait_for_stop` can't let a burst of
// them pile up in the unbounded channel and drain all at once afterward.
let service_pending = Arc::new(AtomicBool::new(false));
let service_pending_actor = service_pending.clone();
std::thread::spawn(move || {
// `connect` takes the single-instance lock (on this actor thread, where
// the `!Send` session lives) before building the backend, so the MCP
// server refuses to attach if another ntoseye already owns the VM.
let built = Session::connect(|| {
let backend: Box<dyn DebugBackend> = match backend.as_str() {
"gdb" => Box::new(GdbClient::connect(
connect.as_deref().unwrap_or("127.0.0.1:1234"),
)?),
"kd" => Box::new(KdBackend::connect(
connect.as_deref().unwrap_or("/tmp/ntoseye-kd.sock"),
)?),
"memory" => Box::new(MemoryBackend::new()),
other => return Err(Error::DebugInfo(format!("unknown backend '{other}'"))),
};
Ok(backend)
})
.map_err(|e| e.to_string());
let mut ctx = match built {
Ok(ctx) => {
let _ = ready_tx.send(Ok(()));
ctx
}
Err(e) => {
let _ = ready_tx.send(Err(e));
return;
}
};
// Unlike the REPL (which pauses at its prompt), the MCP keeps the guest
// running for the session; the connect handshake broke the target in, so
// resume now. Live introspection reads go through /dev/kvm; tools that need
// a stopped target say so and ask the client to call `interrupt` first.
if !ctx.backend.is_running() {
let _ = ctx.backend.continue_execution();
}
// Synchronous job loop. `blocking_recv` is valid here because this is a
// plain std thread with no tokio runtime on it. We clean up on an explicit
// `Shutdown` (Ctrl+C / client disconnect drives one) and also if the
// channel closes outright, so the VM is never left frozen.
loop {
match rx.blocking_recv() {
Some(Command::Run { job, reply }) => {
// DIAGNOSTIC: every tool call submits exactly one Run job, so
// this logs whether the actor runs anything while you believe
// the session is idle (e.g. a client auto-probing tools).
// Gated on the same env as the kd traces so it interleaves.
if std::env::var_os("NTOSEYE_KD_TRACE").is_some() {
eprintln!("mcp: actor: running a tool job");
}
let _ = reply.send(job(&mut ctx));
}
Some(Command::Service) => {
service_pending_actor.store(false, Ordering::Release);
ctx.service_idle();
}
Some(Command::Shutdown { ack }) => {
cleanup_session(&mut ctx);
let _ = ack.send(());
break;
}
None => {
cleanup_session(&mut ctx);
break;
}
}
}
});
match ready_rx.recv() {
Ok(Ok(())) => Ok((tx, service_pending)),
Ok(Err(e)) => Err(anyhow::anyhow!("failed to attach: {e}")),
Err(_) => Err(anyhow::anyhow!("session thread exited before attaching")),
}
}
#[derive(Clone)]
struct NtoseyeMcp {
tx: mpsc::UnboundedSender<Command>,
tool_router: ToolRouter<Self>,
/// Flipped on shutdown so an in-flight `wait_for_stop` bails out promptly and
/// the actor can run cleanup (resume the VM) before exit.
interrupt: Arc<AtomicBool>,
}
// --- argument schemas ---
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct AddressArgs {
#[schemars(
description = "Address as a debugger expression (symbol, register, hex, arithmetic)"
)]
address: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct ReadMemoryArgs {
#[schemars(description = "Start address as a debugger expression")]
address: String,
#[schemars(
range(min = 1, max = 4096),
description = "Number of bytes to read (range 1-4096)"
)]
length: usize,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct ReadCStringArgs {
#[schemars(
description = "Address of the NUL-terminated CHAR* string as a debugger expression"
)]
address: String,
#[schemars(
range(min = 1, max = 4096),
description = "Maximum bytes to scan for the NUL terminator (default 260, range 1-4096)"
)]
max_len: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct DisassembleArgs {
#[schemars(description = "Start address as a debugger expression")]
address: String,
#[schemars(
range(min = 1, max = 128),
description = "Number of instructions to decode (range 1-128)"
)]
count: usize,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct BacktraceArgs {
#[schemars(
range(min = 1, max = 256),
description = "Maximum number of frames to walk (default 64, range 1-256)"
)]
limit: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct TypeArgs {
#[serde(rename = "type")]
#[schemars(description = "Type name to look up across loaded modules, e.g. _EPROCESS")]
type_name: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct SymbolSearchArgs {
#[schemars(
description = "Fuzzy symbol query; `module!query` (e.g. `nt!KiSwap`) scopes to one module"
)]
query: String,
#[schemars(range(min = 1, max = 500), description = "Max results (default 50)")]
limit: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct TypeSearchArgs {
#[schemars(description = "Fuzzy type-name query, e.g. `_EPROCESS` or `KTHREAD`")]
query: String,
#[schemars(range(min = 1, max = 500), description = "Max results (default 50)")]
limit: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct EnumSearchArgs {
#[schemars(description = "Fuzzy enum-name query, e.g. `_KWAIT_REASON` or `_POOL_TYPE`")]
query: String,
#[schemars(range(min = 1, max = 500), description = "Max results (default 50)")]
limit: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct ReadStructArgs {
#[serde(rename = "type")]
#[schemars(description = "Type name, e.g. _EPROCESS")]
type_name: String,
#[schemars(description = "Address of the struct as a debugger expression")]
address: String,
}
/// Shared paging+filter args for the naturally-bounded enumerations
/// (processes/modules/drivers). `offset`/`limit` window the list like reading a
/// file by line range; `filter` is the ergonomic win for finding one entry.
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct ListArgs {
#[schemars(
description = "Case-insensitive substring matched against the name (find a specific entry instead of paging all)"
)]
filter: Option<String>,
#[schemars(
description = "0-based index to start at (default 0). To page, pass the previous page's `next_offset`; absent means done."
)]
offset: Option<usize>,
#[schemars(
range(min = 1, max = 500),
description = "Window size: how many to return from offset (default 100, range 1-500)"
)]
limit: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct ThreadsArgs {
#[schemars(description = "Only threads belonging to this process id")]
pid: Option<u64>,
#[schemars(
description = "0-based index to start at (default 0). To page, pass the previous page's `next_offset`; absent means done."
)]
offset: Option<usize>,
#[schemars(
range(min = 1, max = 200),
description = "Window size (default 50, range 1-200). A live system has 1000+ threads; filter by pid and/or page via offset."
)]
limit: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct MemoryMapArgs {
#[schemars(description = "Process id whose VAD tree to enumerate")]
pid: u64,
#[schemars(
description = "0-based index to start at (default 0). To page, pass the previous page's `next_offset`; absent means done."
)]
offset: Option<usize>,
#[schemars(
range(min = 1, max = 500),
description = "Window size (default 200, range 1-500)."
)]
limit: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct PidArgs {
#[schemars(description = "Process id to attach the inspection context to")]
pid: u64,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct SetBreakpointArgs {
#[schemars(
description = "Breakpoint address as a debugger expression, e.g. user32!PostQuitMessage"
)]
address: String,
#[schemars(
description = "Optional break condition, re-evaluated each hit; the breakpoint only surfaces when it holds (e.g. \"$rcx == 0x4\" or a bare expression treated as non-zero). Comparison ops: == != < <= > >="
)]
condition: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct BreakpointIdArgs {
#[schemars(description = "Breakpoint id")]
id: u32,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct SearchArgs {
#[schemars(description = "Start address as a debugger expression")]
start: String,
#[schemars(
description = "Byte pattern to find, as contiguous lowercase hex, e.g. \"48895c24\""
)]
pattern: String,
#[schemars(
range(min = 1, max = 1048576),
description = "How many bytes to scan from start (range 1-1048576)"
)]
length: usize,
#[schemars(
description = "0-based index into the match list to start at (default 0). To page, pass the previous page's `next_offset`; absent means done."
)]
offset: Option<usize>,
#[schemars(
range(min = 1, max = 4096),
description = "Max matches to return from offset (default 256, range 1-4096). A short pattern over a dense region can match millions of times."
)]
limit: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct WalkListArgs {
#[schemars(
description = "List-head address as a debugger expression, e.g. PsLoadedModuleList"
)]
head: String,
#[schemars(description = "Record struct type the list links into, e.g. _LDR_DATA_TABLE_ENTRY")]
record_type: String,
#[schemars(
description = "Name of the _LIST_ENTRY field inside the record, e.g. InLoadOrderLinks"
)]
link_field: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct WriteMemoryArgs {
#[schemars(description = "Start address as a debugger expression")]
address: String,
#[schemars(
description = "Bytes to write as contiguous lowercase hex (same form read_memory returns), e.g. \"90909090\". 1-4096 bytes."
)]
hex: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct ThreadIdArgs {
#[schemars(description = "Thread/vCPU id to select as the current inspection thread")]
thread: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct SetRegisterArgs {
#[schemars(description = "Register name, e.g. rip, rax, rflags")]
name: String,
#[schemars(
description = "New value as a debugger expression (symbol, register, hex, arithmetic)"
)]
value: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct WaitArgs {
#[schemars(
range(min = 0, max = 20000),
description = "How long to wait for a stop before returning {stop:\"running\"} (default 10000, max 20000; 0 means use the default). Bounded by design: poll by calling again while it returns running. There is no indefinite wait; your MCP client times out the request on its own, and a long wait blocks every other tool on the single debugger session."
)]
timeout_ms: Option<u64>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct DebugLogArgs {
#[schemars(
description = "Resume cursor: pass the previous call's `next_seq` to get only lines captured since then. Default 0 returns from the oldest retained line."
)]
since_seq: Option<u64>,
}
const LIST_DEFAULT_LIMIT: usize = 100;
const LIST_MAX_LIMIT: usize = 500;
const THREADS_DEFAULT_LIMIT: usize = 50;
const THREADS_MAX_LIMIT: usize = 200;
const MEMORY_MAP_DEFAULT_LIMIT: usize = 200;
const MEMORY_MAP_MAX_LIMIT: usize = 500;
const READ_MEMORY_MAX_LENGTH: usize = 4096;
const READ_C_STRING_MAX_LENGTH: usize = 4096;
const READ_C_STRING_DEFAULT_LENGTH: usize = 260; // MAX_PATH, the common case
const WRITE_MEMORY_MAX_LENGTH: usize = 4096;
const SEARCH_MAX_LENGTH: usize = 1 << 20; // 1 MiB scanned per call
const SEARCH_DEFAULT_LIMIT: usize = 256;
const SEARCH_MAX_LIMIT: usize = 4096;
const DISASSEMBLE_MAX_COUNT: usize = 128;
const BACKTRACE_DEFAULT_LIMIT: usize = 64;
const BACKTRACE_MAX_LIMIT: usize = 256;
const CONTINUE_DEFAULT_TIMEOUT_MS: u64 = 10_000;
// Capped well under common MCP client request timeouts (e.g. oh-my-pi defaults to
// 30s) so a wait returns {stop:"running"} and frees the single-session actor
// before the client gives up. No indefinite wait is offered over MCP.
const CONTINUE_MAX_TIMEOUT_MS: u64 = 20_000;
/// A tool failure carrying enough structure to map to the right MCP error class.
/// Keeps guest memory faults distinguishable from bad arguments and internal bugs.
enum ToolError {
/// A bad argument value (JSON-RPC invalid params).
Params(String),
/// Well-formed but not valid in the current state (invalid request).
Request(String),
/// A guest memory access fault (unmapped page, partial read, ...).
Memory(String),
/// Anything else (internal error).
Internal(String),
}
impl From<Error> for ToolError {
fn from(e: Error) -> Self {
let msg = e.to_string();
match e {
Error::BadVirtualAddress(_)
| Error::BadPhysicalAddress(_)
| Error::PartialRead(_)
| Error::PartialWrite(_)
| Error::BufferNotEnough
| Error::InvalidRange => ToolError::Memory(msg),
_ => ToolError::Internal(msg),
}
}
}
impl ToolError {
fn into_mcp(self) -> McpError {
match self {
ToolError::Params(m) => McpError::invalid_params(m, None),
ToolError::Request(m) => McpError::invalid_request(m, None),
// No dedicated JSON-RPC code for a guest fault; surface it as an
// invalid request tagged with `kind` so a client can tell "this
// address isn't readable" apart from an internal bug.
ToolError::Memory(m) => {
McpError::invalid_request(m, Some(serde_json::json!({ "kind": "memory_access" })))
}
ToolError::Internal(m) => McpError::internal_error(m, None),
}
}
}
fn invalid_params(message: impl Into<String>) -> McpError {
McpError::invalid_params(message.into(), None)
}
fn invalid_request(message: impl Into<String>) -> ToolError {
ToolError::Request(message.into())
}
/// Turn a kernel-enumeration failure into an actionable directive when the kernel
/// lists aren't up yet (early boot / mid-rediscovery): the process and module
/// lists don't exist, so the raw "bad virtual address" is noise. Only rewrites
/// when the target is known-incoherent, so a genuine bad address still surfaces
/// as itself (classified by `ToolError::from`).
fn enumeration_error(ctx: &Session, e: Error) -> ToolError {
if !ctx.kernel_coherent() {
invalid_request(
"guest is rebooting or at early boot: kernel process/module lists are not up yet, \
so enumeration is unavailable. resume, then wait_for_stop until status reports \
coherent:true, then retry.",
)
} else {
ToolError::from(e)
}
}
fn required_range(name: &str, value: usize, max: usize) -> Result<usize, McpError> {
if value == 0 || value > max {
Err(invalid_params(format!("{name} must be in range 1..={max}")))
} else {
Ok(value)
}
}
fn optional_limit(
name: &str,
value: Option<usize>,
default: usize,
max: usize,
) -> Result<usize, McpError> {
match value {
Some(v) => required_range(name, v, max),
None => Ok(default),
}
}
fn optional_timeout_ms(value: Option<u64>) -> Result<u64, McpError> {
// 0 or omitted means "use the default". There is deliberately no indefinite
// wait over MCP: the client owns the request timeout, and a long wait pins the
// single-session actor, so callers poll instead (a bounded wait returns
// {stop:"running"} and the caller calls again).
let ms = match value {
None | Some(0) => CONTINUE_DEFAULT_TIMEOUT_MS,
Some(ms) => ms,
};
if ms > CONTINUE_MAX_TIMEOUT_MS {
Err(invalid_params(format!(
"timeout_ms must be 0 (use default) or in range 1..={CONTINUE_MAX_TIMEOUT_MS}"
)))
} else {
Ok(ms)
}
}
fn require_halted(ctx: &mut Session, tool: &str) -> Result<(), ToolError> {
// A stop the servicer already caught (e.g. an early-boot break) leaves the VM
// physically halted while `is_running()` still reads stale-true. Ingest it
// first so a genuinely halted target isn't turned away with "call interrupt";
// a debugger-noise break-in is absorbed (resume), so only a VM that is truly
// running, or stopped on a real, inspectable site, reaches the guard.
ctx.settle_pending_stop().map_err(ToolError::from)?;
// This op runs against the (already-rebuilt) current kernel, so the host has
// engaged the target post-reboot: finish any memory-completable rediscovery
// and stop deferring the reboot notification, so a later `wait_for_stop`
// doesn't flush a stale `target_reloaded` into the middle of the agent's flow
// (e.g. right after it set a kernel breakpoint). Matches what `status` does.
ctx.try_finish_rediscovery_from_memory();
ctx.clear_deferred_reload_surface();
if ctx.backend.is_running() {
Err(invalid_request(format!(
"{tool} requires the VM to be halted; call interrupt first (resume afterwards with resume)"
)))
} else {
Ok(())
}
}
/// Render a [`ContinueOutcome`] as JSON, enriching breakpoint/exception stops with
/// the current process and resolved symbol from `ctx`.
fn continue_outcome_json(ctx: &Session, outcome: ContinueOutcome) -> Value {
let process = ctx
.target
.current_process_info
.as_ref()
.map(|p| serde_json::json!({ "pid": p.pid, "name": p.name }));
let symbol_at = |rip: u64| ctx.target.closest_symbol_current_context(VirtAddr(rip));
match outcome {
ContinueOutcome::Breakpoint {
id,
address,
symbol,
temporary,
rip,
} => serde_json::json!({
"stop": "breakpoint",
"id": id,
"address": hex(address),
"symbol": symbol.or_else(|| symbol_at(rip)),
"temporary": temporary,
"rip": hex(rip),
"process": process,
}),
ContinueOutcome::Bugcheck { rip, info } => {
let analysis = info
.map(|i| analyze_bugcheck(&ctx.target, &i))
.or_else(|| current_bugcheck(&ctx.target));
serde_json::json!({
"stop": "bugcheck",
"rip": rip.map(hex),
"bugcheck": analysis.as_ref().map(|a| view::to_json(&view::bugcheck(a))),
})
}
ContinueOutcome::Stopped {
rip,
exception_code,
} => serde_json::json!({
"stop": "exception",
"rip": hex(rip),
"exception_code": exception_code,
"symbol": symbol_at(rip),
"process": process,
}),
ContinueOutcome::Step { rip } => serde_json::json!({
"stop": "step",
"rip": hex(rip),
"symbol": symbol_at(rip),
"process": process,
}),
ContinueOutcome::TargetReloaded {
kernel_base,
coherent,
} => {
let note = if coherent {
"The guest rebooted and debugger state is now fully rebuilt against \
the new kernel: the loaded-module list is available and the system \
is introspectable. The VM is halted at an internal KD break-in (an \
arbitrary landing site, not a meaningful location). Every prior \
address (eprocess, ethread/kthread, module base, dtb) from before \
the reboot is invalid; re-enumerate (processes/threads/modules) \
before acting."
} else {
"The guest rebooted and the VM is halted at the earliest post-reboot \
stop, before kernel initialization: kernel symbols are loaded, but \
the loaded-module list does not exist yet, so process, thread, and \
module enumeration are UNAVAILABLE at this stop and most reads will \
fail or mislead. Every prior address (eprocess, ethread/kthread, \
module base, dtb) is now invalid; do not reuse results from before \
the reload. Use this stop to debug early boot (e.g. set breakpoints \
on init paths; hits are reported even while the system is coming \
up). Otherwise to reach a usable system: resume, then poll \
wait_for_stop, and only enumerate once status reports \
coherent:true (rediscovery finishes silently as the system comes \
up; this is the only reload stop reported). Enumerating while \
coherent:false returns a directive error, not data."
};
serde_json::json!({
"stop": "target_reloaded",
"kernel_base": kernel_base.map(hex),
"coherent": coherent,
"note": note,
})
}
ContinueOutcome::Running => serde_json::json!({ "stop": "running" }),
ContinueOutcome::Halted { rip } => serde_json::json!({
"stop": "halted",
// Not a new event; the VM was already parked here, so don't treat it
// like a fresh breakpoint/exception stop.
"event": false,
"rip": hex(rip),
"symbol": symbol_at(rip),
"process": process,
"coherent": ctx.kernel_coherent(),
}),
}
}
// --- value decoding helpers (run on the actor thread) ---
/// Resolve an address-expression argument against the live session.
fn eval_addr(ctx: &Session, expr: &str) -> Result<u64, ToolError> {
Expr::eval(expr, &ctx.target)
.map(|v| v.0)
.map_err(|e| ToolError::Params(e.to_string()))
}
/// Lowercase contiguous hex for a byte slice.
fn hex_of(slice: &[u8]) -> String {
slice.iter().map(|b| format!("{b:02x}")).collect()
}
/// Decode contiguous hex (the form `hex_of`/`read_memory` emit) back to bytes.
/// Rejects odd length and non-hex digits as invalid params.
fn bytes_of_hex(s: &str) -> Result<Vec<u8>, ToolError> {
let s = s.strip_prefix("0x").unwrap_or(s);
if s.len() % 2 != 0 {
return Err(invalid_request("hex must have an even number of digits"));
}
(0..s.len())
.step_by(2)
.map(|i| {
u8::from_str_radix(&s[i..i + 2], 16)
.map_err(|_| invalid_request("hex contains non-hex digits"))
})
.collect()
}
/// Format an address/value as a `0x` hex string. JSON numbers are decimal-only,
/// so addresses (which debugger users always read in hex) are emitted as strings.
fn hex(v: u64) -> String {
format!("{v:#x}")
}
/// Decode a struct's fields the same way the Python SDK does: scalars (1/2/4/8
/// bytes) and pointers become ints, bitfields are masked to their value, and
/// larger aggregates become a hex string. Nested-struct fields the PDB reports
/// with size 0 are omitted; read those separately with their own type.
fn decode_struct(info: &TypeInfo, buf: &[u8]) -> Value {
let mut map = serde_json::Map::new();
for (name, value) in info.decode_fields(buf) {
// A struct dump renders every scalar in hex (WinDbg `dt` convention):
// type-stable and predictable, so the same field never flips between
// decimal and hex across instances. Most 8-byte fields are addresses
// anyway (ULONG_PTR/ULONGLONG the PDB types as ints, e.g.
// _LOADER_PARAMETER_BLOCK.KernelStack/Prcb/Process); counts, sizes and
// versions read as hex too, which a consumer converts trivially.
// Aggregates stay contiguous bytes.
let v = match value {
FieldValue::Int(n) | FieldValue::Bitfield(n) | FieldValue::Pointer(n) => {
Value::from(hex(n))
}
FieldValue::Bytes(b) => Value::from(hex_of(&b)),
};
map.insert(name, v);
}
Value::Object(map)
}
/// Build a paged list envelope: `{total, offset, returned, has_more, next_offset?, <key>: [...]}`.
/// `next_offset` is the offset of the next page, present only when more pages remain.
/// `total` is the count after filtering; `items` is the windowed slice.
fn paged(total: usize, offset: usize, key: &str, items: Vec<Value>) -> Value {
let returned = items.len();
let end = offset.saturating_add(returned);
let mut obj = serde_json::json!({
"total": total,
"offset": offset,
"returned": returned,
"has_more": end < total,
});
if end < total {
obj["next_offset"] = end.into();
}
obj[key] = Value::Array(items);
obj
}
/// Wrap a JSON value as a successful tool result (pretty-printed text content).
// Structured tool result. rmcp's `structured()` also embeds a compact-text copy
// in `content`, so clients that ignore structuredContent still get JSON.
fn json_result(v: Value) -> Result<CallToolResult, McpError> {
Ok(CallToolResult::structured(v))
}
#[tool_router]
impl NtoseyeMcp {
fn new(tx: mpsc::UnboundedSender<Command>, interrupt: Arc<AtomicBool>) -> Self {
let tool_router = Self::tool_router();
Self {
tx,
tool_router,
interrupt,
}
}
/// Ship a job to the session actor and await its JSON reply.
async fn run<F>(&self, job: F) -> Result<Value, McpError>
where
F: FnOnce(&mut Session) -> Result<Value, ToolError> + Send + 'static,
{
let (reply_tx, reply_rx) = oneshot::channel();
self.tx
.send(Command::Run {
job: Box::new(job),
reply: reply_tx,
})
.map_err(|_| McpError::internal_error("debugger session is gone", None))?;
reply_rx
.await
.map_err(|_| McpError::internal_error("debugger session dropped the request", None))?
.map_err(ToolError::into_mcp)
}
#[tool(
description = "List running processes (optional name filter; paged via offset/limit) as {total, offset, returned, has_more, next_offset?, processes:[{pid, name, eprocess, dtb}]}"
)]
async fn processes(
&self,
Parameters(ListArgs {
filter,
offset,
limit,
}): Parameters<ListArgs>,
) -> Result<CallToolResult, McpError> {
let offset = offset.unwrap_or(0);
let limit = optional_limit("limit", limit, LIST_DEFAULT_LIMIT, LIST_MAX_LIMIT)?;
let v = self
.run(move |ctx| {
let matched = ctx
.target
.matching_processes(filter.as_deref())
.map_err(|e| enumeration_error(ctx, e))?;
let total = matched.len();
let items: Vec<Value> = matched
.iter()
.skip(offset)
.take(limit)
.map(|p| {
serde_json::json!({
"pid": p.pid,
"name": p.name,
"eprocess": hex(p.eprocess_va.0),
"dtb": hex(p.dtb),
})
})
.collect();
Ok(paged(total, offset, "processes", items))
})
.await?;
json_result(v)
}
#[tool(
description = "List loaded kernel modules (optional name filter; paged via offset/limit) as {total, offset, returned, has_more, next_offset?, modules:[{name, short_name, base, end, size}]}"
)]
async fn kernel_modules(
&self,
Parameters(ListArgs {
filter,
offset,
limit,
}): Parameters<ListArgs>,
) -> Result<CallToolResult, McpError> {
let offset = offset.unwrap_or(0);
let limit = optional_limit("limit", limit, LIST_DEFAULT_LIMIT, LIST_MAX_LIMIT)?;
let v = self
.run(move |ctx| {
let mods = ctx
.target
.guest
.kernel_modules()
.map_err(|e| enumeration_error(ctx, e))?;
let f = filter.as_deref().map(str::to_ascii_lowercase);
let matched: Vec<_> = mods
.iter()
.filter(|m| {
f.as_deref().is_none_or(|f| {
m.name.to_ascii_lowercase().contains(f)
|| m.short_name.to_ascii_lowercase().contains(f)
})
})
.collect();
let total = matched.len();
let items: Vec<Value> = matched
.iter()
.skip(offset)
.take(limit)
.map(|m| {
serde_json::json!({
"name": m.name,
"short_name": m.short_name,
"base": hex(m.base_address.0),
"end": hex(m.end_address().0),
"size": m.size,
})
})
.collect();
Ok(paged(total, offset, "modules", items))
})
.await?;
json_result(v)
}
#[tool(
description = "List driver objects (optional name filter; paged via offset/limit) as {total, offset, returned, has_more, next_offset?, drivers:[{name, object, driver_start, driver_size, device_object, driver_unload}]}"
)]
async fn driver_objects(
&self,
Parameters(ListArgs {
filter,
offset,
limit,
}): Parameters<ListArgs>,
) -> Result<CallToolResult, McpError> {
let offset = offset.unwrap_or(0);
let limit = optional_limit("limit", limit, LIST_DEFAULT_LIMIT, LIST_MAX_LIMIT)?;
let v = self
.run(move |ctx| {
let drivers = ctx
.target
.enumerate_driver_objects()
.map_err(|e| enumeration_error(ctx, e))?;
let f = filter.as_deref().map(str::to_ascii_lowercase);
let matched: Vec<_> = drivers
.iter()
.filter(|d| {
f.as_deref()
.is_none_or(|f| d.name.to_ascii_lowercase().contains(f))
})
.collect();
let total = matched.len();
let items: Vec<Value> = matched
.iter()
.skip(offset)
.take(limit)
.map(|d| {
serde_json::json!({
"name": d.name,
"object": hex(d.object.0),
"driver_start": hex(d.driver_start.0),
"driver_size": d.driver_size,
"device_object": hex(d.device_object.0),
"driver_unload": hex(d.driver_unload.0),
})
})
.collect();
Ok(paged(total, offset, "drivers", items))
})
.await?;
json_result(v)
}
#[tool(
description = "List Windows threads (filter by pid; paged via offset/limit) as {total, offset, returned, has_more, next_offset?, threads:[{tid, pid, process_name, ethread, kthread, eprocess, state, state_name, wait_reason, wait_reason_name, active}]}. `active` is the vCPU id currently running the thread (only resolved when the VM is halted), else null. A live system has 1000+ threads; pass pid to scope to one process."
)]
async fn threads(
&self,
Parameters(ThreadsArgs { pid, offset, limit }): Parameters<ThreadsArgs>,
) -> Result<CallToolResult, McpError> {
let offset = offset.unwrap_or(0);
let limit = optional_limit("limit", limit, THREADS_DEFAULT_LIMIT, THREADS_MAX_LIMIT)?;
let v = self
.run(move |ctx| {
let (threads, active) = ctx
.windows_threads()
.map_err(|e| enumeration_error(ctx, e))?;
let matched: Vec<_> = threads
.iter()
.filter(|t| pid.is_none_or(|p| t.pid == Some(p)))
.collect();
let total = matched.len();
let items: Vec<Value> = matched
.iter()
.skip(offset)
.take(limit)
.map(|t| {
serde_json::json!({
"tid": t.tid,
"pid": t.pid,
"process_name": t.process_name,
"ethread": hex(t.ethread.0),
"kthread": hex(t.kthread.0),
"eprocess": t.eprocess.map(|a| hex(a.0)),
"state": t.state,
"state_name": t.state.map(kthread_state_name),
"wait_reason": t.wait_reason,
"wait_reason_name": t.wait_reason.map(wait_reason_name),
"active": active.get(&t.ethread.0),
})
})
.collect();
Ok(paged(total, offset, "threads", items))
})
.await?;
json_result(v)
}
#[tool(
description = "Inspect every vCPU as {vcpus:[{id, rip, context, symbol, error}]}: the address space each is executing in (\"kernel\", a process name, or \"unknown\") and the nearest symbol. Requires the VM halted (call interrupt first, or be stopped at a breakpoint)."
)]
async fn vcpus(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
require_halted(ctx, "vcpus")?;
let vcpus = ctx.vcpus().map_err(ToolError::from)?;
let items: Vec<Value> = vcpus
.into_iter()
.map(|v| {
serde_json::json!({
"id": v.id,
"rip": v.rip.map(hex),
"context": v.context,
"symbol": v.symbol,
"error": v.error,
})
})
.collect();
Ok(serde_json::json!({ "vcpus": items }))
})
.await?;
json_result(v)
}
#[tool(
description = "Read all registers of the current thread as a {name: value} object. Requires the VM halted (call interrupt first, or be stopped at a breakpoint)."
)]
async fn registers(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
require_halted(ctx, "registers")?;
let regs = ctx.backend.read_registers().map_err(ToolError::from)?;
let map: serde_json::Map<String, Value> = ctx
.register_map
.to_hashmap(®s)
.into_iter()
.map(|(name, value)| (name, Value::from(hex(value))))
.collect();
Ok(Value::Object(map))
})
.await?;
json_result(v)
}
#[tool(
description = "Read guest virtual memory in the current address space (attached process or kernel); length must be 1-4096. Returns {address, length, hex} (hex is contiguous bytes)"
)]
async fn read_memory(
&self,
Parameters(ReadMemoryArgs { address, length }): Parameters<ReadMemoryArgs>,
) -> Result<CallToolResult, McpError> {
let length = required_range("length", length, READ_MEMORY_MAX_LENGTH)?;
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let len = length;
let mut buf = vec![0u8; len];
let process = ctx.target.current_process();
process
.memory()
.read_bytes(VirtAddr(addr), &mut buf)
.map_err(ToolError::from)?;
ctx.breakpoints
.mask_breakpoint_bytes(VirtAddr(addr), &mut buf, process.dtb());
Ok(serde_json::json!({
"address": hex(addr),
"length": len,
"hex": hex_of(&buf),
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Disassemble 1-128 instructions at an address in the current address space; returns {ip, hex, asm, comment} rows"
)]
async fn disassemble(
&self,
Parameters(DisassembleArgs { address, count }): Parameters<DisassembleArgs>,
) -> Result<CallToolResult, McpError> {
let count = required_range("count", count, DISASSEMBLE_MAX_COUNT)?;
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let rows = ctx
.disassemble(VirtAddr(addr), count)
.map_err(ToolError::from)?;
let arr = rows
.iter()
.map(|r| {
serde_json::json!({
"ip": hex(r.ip),
"hex": r.hex,
"asm": r.asm,
"comment": r.comment,
})
})
.collect();
Ok(Value::Array(arr))
})
.await?;
json_result(v)
}
#[tool(
description = "Walk the current thread's call stack; limit default 64, range 1-256. Returns {ip, sp, symbol, source} frames (source: current/unwind/scan). Requires the VM halted (call interrupt first, or be stopped at a breakpoint)."
)]
async fn backtrace(
&self,
Parameters(BacktraceArgs { limit }): Parameters<BacktraceArgs>,
) -> Result<CallToolResult, McpError> {
let limit = optional_limit("limit", limit, BACKTRACE_DEFAULT_LIMIT, BACKTRACE_MAX_LIMIT)?;
let v = self
.run(move |ctx| {
require_halted(ctx, "backtrace")?;
let trace = ctx.backtrace(limit).map_err(ToolError::from)?;
let arr = trace
.frames
.iter()
.map(|f| {
serde_json::json!({
"ip": hex(f.ip),
"sp": hex(f.sp),
"symbol": f.symbol,
"source": f.source.as_str(),
})
})
.collect();
Ok(Value::Array(arr))
})
.await?;
json_result(v)
}
#[tool(
description = "Walk the page tables for an address; returns {address, levels:[{level, address, value, pfn, present, large_page, writable, user, nx, flags}]}"
)]
async fn pte_walk(
&self,
Parameters(AddressArgs { address }): Parameters<AddressArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let t = ctx
.target
.pte_traverse(VirtAddr(addr))
.map_err(ToolError::from)?;
let mut levels = vec![
view::to_json(&view::pte_level(&t.pxe)),
view::to_json(&view::pte_level(&t.ppe)),
];
if let Some(pde) = &t.pde {
levels.push(view::to_json(&view::pte_level(pde)));
}
if let Some(pte) = &t.pte {
levels.push(view::to_json(&view::pte_level(pte)));
}
Ok(serde_json::json!({
"address": hex(t.address.0),
"dtb": hex(t.dtb),
"levels": levels,
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Inspect an _IRP at an address (expression) in the current address space. Returns {address, type, size, stack_count, current_location, pending_returned, requestor_mode, io_status, user_event, user_buffer, mdl_address, thread, current_stack:{address, major_function, major_function_name, minor_function, device_object, file_object, completion_routine, context}|null}. current_stack is null when CurrentLocation is out of range or the slot is unreadable."
)]
async fn inspect_irp(
&self,
Parameters(AddressArgs { address }): Parameters<AddressArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let irp = ctx
.target
.inspect_irp(VirtAddr(addr))
.map_err(ToolError::from)?;
Ok(view::to_json(&view::irp(&irp)))
})
.await?;
json_result(v)
}
#[tool(
description = "Inspect a _DRIVER_OBJECT at an address (expression). Accepts a pointer to one. Returns {object, via_pointer, name, driver_start, driver_size, driver_section, driver_unload, devices:[{device, device_type, flags, characteristics, attached, next}], dispatch:[{index, name, routine, symbol}]} where dispatch is the 28-entry IRP_MJ_* table."
)]
async fn inspect_driver_object(
&self,
Parameters(AddressArgs { address }): Parameters<AddressArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let d = ctx
.target
.inspect_driver_object(VirtAddr(addr))
.map_err(ToolError::from)?;
Ok(view::to_json(&view::driver_object(&ctx.target, &d)))
})
.await?;
json_result(v)
}
#[tool(
description = "Inspect a _DEVICE_OBJECT at an address (expression). Accepts a pointer to one. Returns {object, via_pointer, device_type, flags, characteristics, driver_object, attached_device, next_device, current_irp, device_extension, attached_stack:[{device, driver_object, device_type, flags}]}."
)]
async fn inspect_device_object(
&self,
Parameters(AddressArgs { address }): Parameters<AddressArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let d = ctx
.target
.inspect_device_object(VirtAddr(addr))
.map_err(ToolError::from)?;
Ok(view::to_json(&view::device_object(&d)))
})
.await?;
json_result(v)
}
#[tool(
description = "Inspect an executive _OBJECT_HEADER for an address (expression), accepting either the object body or the header. Returns {input, mode, header, body, pointer_count, handle_count, type_index, type_object, type_name, info_mask, name_info, name}. mode is 'body' or 'header'; type_name/name are null when unresolved."
)]
async fn inspect_object_header(
&self,
Parameters(AddressArgs { address }): Parameters<AddressArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let o = ctx
.target
.inspect_object_header(VirtAddr(addr))
.map_err(ToolError::from)?;
Ok(view::to_json(&view::object_header(&o)))
})
.await?;
json_result(v)
}
#[tool(
description = "Enumerate process/thread/image notification callbacks (Psp*NotifyRoutine arrays). `filter` is a case-insensitive substring matched against the resolved routine symbol. Paged via offset/limit. Returns {total, offset, returned, has_more, next_offset?, callbacks:[{kind, index, function, symbol, block, raw, context}]}."
)]
async fn notify_callbacks(
&self,
Parameters(ListArgs {
filter,
offset,
limit,
}): Parameters<ListArgs>,
) -> Result<CallToolResult, McpError> {
let offset = offset.unwrap_or(0);
let limit = optional_limit("limit", limit, LIST_DEFAULT_LIMIT, LIST_MAX_LIMIT)?;
let v = self
.run(move |ctx| {
let dtb = ctx.target.guest.ntoskrnl.dtb();
let cbs = ctx
.target
.enumerate_notify_callbacks()
.map_err(ToolError::from)?;
let f = filter.as_deref().map(str::to_ascii_lowercase);
let rows: Vec<Value> = cbs
.iter()
.map(|c| {
let symbol = ctx
.target
.symbols
.format_closest_symbol_for_address(dtb, c.function);
(c, symbol)
})
.filter(|(_, symbol)| {
f.as_deref().is_none_or(|f| {
symbol
.as_deref()
.is_some_and(|s| s.to_ascii_lowercase().contains(f))
})
})
.map(|(c, symbol)| view::to_json(&view::notify_callback(c, symbol)))
.collect();
let total = rows.len();
let items: Vec<Value> = rows.into_iter().skip(offset).take(limit).collect();
Ok(paged(total, offset, "callbacks", items))
})
.await?;
json_result(v)
}
#[tool(
description = "Dump the kernel SSDT (KiServiceTable) and, when initialized, the win32k shadow table. Returns {tables:[{label, base, limit, entries:[{index, target, symbol, module}]}]}. An entry whose module differs from the expected owner (ntoskrnl/win32k) is a candidate hook."
)]
async fn ssdt(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let tables = ctx.target.dump_ssdt().map_err(ToolError::from)?;
let tables: Vec<Value> = tables
.iter()
.map(|t| view::to_json(&view::ssdt_table(t)))
.collect();
Ok(serde_json::json!({ "tables": tables }))
})
.await?;
json_result(v)
}
#[tool(
description = "Discover in-flight IRPs from each thread's _ETHREAD.IrpList and each device's _DEVICE_OBJECT.CurrentIrp. `filter` scopes processes (pid or name substring) and driver names; a numeric/pid filter skips the device sweep. Paged via offset/limit. Returns {total, offset, returned, has_more, next_offset?, irps:[{irp, source, stack_count, current_location, pid, tid, ethread, state, wait_reason, driver, device}]}."
)]
async fn discover_irps(
&self,
Parameters(ListArgs {
filter,
offset,
limit,
}): Parameters<ListArgs>,
) -> Result<CallToolResult, McpError> {
let offset = offset.unwrap_or(0);
let limit = optional_limit("limit", limit, LIST_DEFAULT_LIMIT, LIST_MAX_LIMIT)?;
let v = self
.run(move |ctx| {
let hits = ctx
.target
.discover_irps(filter.as_deref())
.map_err(ToolError::from)?;
let total = hits.len();
let items: Vec<Value> = hits
.iter()
.skip(offset)
.take(limit)
.map(|h| view::to_json(&view::irp_hit(h)))
.collect();
Ok(paged(total, offset, "irps", items))
})
.await?;
json_result(v)
}
#[tool(
description = "Describe what an address belongs to (complements pte_walk's how-it's-mapped with where-it-lives). Returns {address, dtb, kind, module:{name, base, size, offset}|null, section|null, va_type|null, region:{start, end, protection, vad_type, private_memory, commit_charge, details}|null}. kind is kernel-module/user-image/kernel-region/private/mapped/unknown; section is the PE section for a module hit; va_type is the MM region name (e.g. KernelStacks, PagedPool, SystemPtes) for a kernel address; region is the VAD entry for a process address."
)]
async fn describe_address(
&self,
Parameters(AddressArgs { address }): Parameters<AddressArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let d = ctx
.target
.describe_address(VirtAddr(addr))
.map_err(ToolError::from)?;
Ok(view::to_json(&view::address_description(&d)))
})
.await?;
json_result(v)
}
#[tool(
description = "Resolve the nearest symbol to an address as module!name+0x..; returns {address, symbol}"
)]
async fn closest_symbol(
&self,
Parameters(AddressArgs { address }): Parameters<AddressArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let symbol = ctx.target.closest_symbol_current_context(VirtAddr(addr));
Ok(serde_json::json!({ "address": hex(addr), "symbol": symbol }))
})
.await?;
json_result(v)
}
#[tool(
description = "Get a struct/class type's layout; returns {name, size, fields:[{name, offset, size, type}]} sorted by offset. Use enum_values for PDB enums."
)]
async fn type_layout(
&self,
Parameters(TypeArgs { type_name }): Parameters<TypeArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let dtb = ctx.target.current_dtb();
let info = ctx
.target
.symbols
.find_type_across_modules(dtb, &type_name)
.ok_or_else(|| {
ToolError::Request(ctx.target.symbols.unresolved_type_message(dtb, &type_name))
})?;
let fields: Vec<Value> = {
let mut fields: Vec<_> = info.fields.iter().collect();
fields.sort_by_key(|(_, f)| f.offset);
fields
}
.into_iter()
.map(|(n, f)| {
serde_json::json!({
"name": n,
"offset": hex(f.offset as u64),
"size": f.size,
"type": format!("{}", f.type_data),
})
})
.collect();
Ok(serde_json::json!({
"name": type_name,
"size": info.size,
"fields": fields,
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Fuzzy-search symbols by name (use `module!query` to scope to one module). Returns {query, total, symbols:[{name, address, module}]}; address/module are null when a name doesn't resolve in the current context."
)]
async fn search_symbols(
&self,
Parameters(SymbolSearchArgs { query, limit }): Parameters<SymbolSearchArgs>,
) -> Result<CallToolResult, McpError> {
let limit = optional_limit("limit", limit, 50, 500)?;
let v = self
.run(move |ctx| {
let dtb = ctx.target.current_dtb();
let (module, names) = match query.split_once('!') {
Some((m, q)) => (
Some(m.to_string()),
ctx.target
.symbols
.search_symbols_in_module(dtb, m, q, limit),
),
None => (
None,
ctx.target.current_symbol_index().search(&query, limit),
),
};
let symbols: Vec<Value> = names
.iter()
.map(|name| {
let lookup = match &module {
Some(m) => format!("{m}!{name}"),
None => name.clone(),
};
let resolved = ctx.target.symbols.find_symbol_with_module(dtb, &lookup);
serde_json::json!({
"name": name,
"address": resolved.as_ref().map(|(a, _)| hex(a.0)),
"module": resolved.map(|(_, m)| m),
})
})
.collect();
Ok(serde_json::json!({
"query": query,
"total": symbols.len(),
"symbols": symbols,
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Fuzzy-search struct/class type names across loaded modules. Returns {query, total, types:[name]}; feed a name to type_layout for its fields. Use search_enums for PDB enums."
)]
async fn search_types(
&self,
Parameters(TypeSearchArgs { query, limit }): Parameters<TypeSearchArgs>,
) -> Result<CallToolResult, McpError> {
let limit = optional_limit("limit", limit, 50, 500)?;
let v = self
.run(move |ctx| {
let names = ctx.target.current_types_index().search(&query, limit);
Ok(serde_json::json!({
"query": query,
"total": names.len(),
"types": names,
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Fuzzy-search PDB enum names across loaded modules. Returns {query, total, enums:[name]}; feed a name to enum_values."
)]
async fn search_enums(
&self,
Parameters(EnumSearchArgs { query, limit }): Parameters<EnumSearchArgs>,
) -> Result<CallToolResult, McpError> {
let limit = optional_limit("limit", limit, 50, 500)?;
let v = self
.run(move |ctx| {
let names = ctx.target.current_enums_index().search(&query, limit);
Ok(serde_json::json!({
"query": query,
"total": names.len(),
"enums": names,
}))
})
.await?;
json_result(v)
}
#[tool(
description = "List a PDB enum's variants (in declaration order) as {name, values:[{name, value}]}. Enums aren't structs, so use this rather than type_layout (e.g. _MI_SYSTEM_VA_TYPE, _KWAIT_REASON, _POOL_TYPE)."
)]
async fn enum_values(
&self,
Parameters(TypeArgs { type_name }): Parameters<TypeArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let variants = ctx
.target
.symbols
.find_enum_across_modules(ctx.target.current_dtb(), &type_name)
.ok_or_else(|| ToolError::Request(format!("unknown enum: {type_name}")))?;
let values: Vec<Value> = variants
.into_iter()
.map(|(name, value)| serde_json::json!({ "name": name, "value": value }))
.collect();
Ok(serde_json::json!({ "name": type_name, "values": values }))
})
.await?;
json_result(v)
}
#[tool(
description = "Read a typed struct at an address in the current address space; returns {type, address, fields:{name: value}}. Nested structs are omitted; read them separately at their own address."
)]
async fn read_struct(
&self,
Parameters(ReadStructArgs { type_name, address }): Parameters<ReadStructArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let dtb = ctx.target.current_dtb();
let info = ctx
.target
.symbols
.find_type_across_modules(dtb, &type_name)
.ok_or_else(|| {
ToolError::Request(ctx.target.symbols.unresolved_type_message(dtb, &type_name))
})?;
let mut buf = vec![0u8; info.size];
ctx.target
.current_process()
.memory()
.read_bytes(VirtAddr(addr), &mut buf)
.map_err(ToolError::from)?;
Ok(serde_json::json!({
"type": type_name,
"address": hex(addr),
"fields": decode_struct(&info, &buf),
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Enumerate a process's virtual address map (VAD tree) by pid, paged via offset/limit; returns {total, offset, returned, has_more, next_offset?, regions:[{start, end, size, protection, vad_type, private, commit, details}]}"
)]
async fn memory_map(
&self,
Parameters(MemoryMapArgs { pid, offset, limit }): Parameters<MemoryMapArgs>,
) -> Result<CallToolResult, McpError> {
let offset = offset.unwrap_or(0);
let limit = optional_limit(
"limit",
limit,
MEMORY_MAP_DEFAULT_LIMIT,
MEMORY_MAP_MAX_LIMIT,
)?;
let v = self
.run(move |ctx| {
let process = ctx
.target
.guest
.enumerate_processes()
.map_err(|e| enumeration_error(ctx, e))?
.into_iter()
.find(|p| p.pid == pid)
.ok_or_else(|| ToolError::Request(format!("no process with pid {pid}")))?;
let all = ctx
.target
.enumerate_vad_regions_for_process_info(&process)
.map_err(ToolError::from)?;
let total = all.len();
let items: Vec<Value> = all
.iter()
.skip(offset)
.take(limit)
.map(|r| {
serde_json::json!({
"start": hex(r.start.0),
"end": hex(r.end.0),
"size": r.size(),
"protection": r.protection,
"vad_type": r.vad_type,
"private": r.private_memory,
"commit": r.commit_charge,
"details": r.details,
})
})
.collect();
Ok(paged(total, offset, "regions", items))
})
.await?;
json_result(v)
}
#[tool(
description = "Current inspection context: {dtb, current_thread, process:{pid, name, eprocess}|null}"
)]
async fn context(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
let process = ctx.target.current_process_info.as_ref().map(|p| {
serde_json::json!({
"pid": p.pid,
"name": p.name,
"eprocess": hex(p.eprocess_va.0),
})
});
Ok(serde_json::json!({
"dtb": hex(ctx.target.current_dtb()),
"current_thread": ctx.current_thread,
"process": process,
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Analyze the current bugcheck (BSOD): {code, code_hex, name, description, driver, args:[{index, value, description}], fault:{ip, symbol, driver}, source}, or null if the guest is not bugchecking"
)]
async fn bugcheck(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
Ok(match current_bugcheck(&ctx.target) {
Some(analysis) => view::to_json(&view::bugcheck(&analysis)),
None => Value::Null,
})
})
.await?;
json_result(v)
}
#[tool(
description = "Read captured guest debug output (DbgPrint, e.g. \"DriverEntry failed 0x...\"). Snapshot+cursor: pass since_seq (the prior call's next_seq) to poll only new lines; returns {lines:[{seq, timestamp_ms, text}], next_seq, dropped}. dropped=true means older lines were evicted before you read them (poll more often). Output is captured only while the target runs, so resume to let it accumulate; an empty result is not proof the guest is silent. Empty on backends without a debug stream."
)]
async fn debug_log(
&self,
Parameters(DebugLogArgs { since_seq }): Parameters<DebugLogArgs>,
) -> Result<CallToolResult, McpError> {
let since = since_seq.unwrap_or(0);
let v = self
.run(move |ctx| {
let page = ctx.read_debug_output(since);
let lines: Vec<Value> = page
.lines
.iter()
.map(|l| {
serde_json::json!({
"seq": l.seq,
"timestamp_ms": l.timestamp_ms,
"text": l.text,
})
})
.collect();
Ok(serde_json::json!({
"lines": lines,
"next_seq": page.next_seq,
"dropped": page.dropped,
}))
})
.await?;
json_result(v)
}
#[tool(
description = "List the backend's capability matrix - which debug operations the current transport (kd/gdb/memory) supports - as [{capability, label, supported}]. Check before a state-changing op (e.g. usermode breakpoints)."
)]
async fn capabilities(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
let arr = ctx
.capabilities()
.iter()
.map(|c| {
serde_json::json!({
"capability": format!("{:?}", c.capability),
"label": c.capability.label(),
"supported": c.supported,
})
})
.collect();
Ok(Value::Array(arr))
})
.await?;
json_result(v)
}
#[tool(
description = "List loaded modules for the current inspection scope: the attached process's user-mode modules when attached (attach_process), otherwise the kernel module list. Optional name filter; paged via offset/limit. Returns {total, offset, returned, has_more, next_offset?, modules:[{name, short_name, base, end, size}]}. Use kernel_modules to list kernel modules regardless of attach state."
)]
async fn modules(
&self,
Parameters(ListArgs {
filter,
offset,
limit,
}): Parameters<ListArgs>,
) -> Result<CallToolResult, McpError> {
let offset = offset.unwrap_or(0);
let limit = optional_limit("limit", limit, LIST_DEFAULT_LIMIT, LIST_MAX_LIMIT)?;
let v = self
.run(move |ctx| {
let mods = ctx
.target
.modules()
.map_err(|e| enumeration_error(ctx, e))?;
let f = filter.as_deref().map(str::to_ascii_lowercase);
let matched: Vec<_> = mods
.iter()
.filter(|m| {
f.as_deref().is_none_or(|f| {
m.name.to_ascii_lowercase().contains(f)
|| m.short_name.to_ascii_lowercase().contains(f)
})
})
.collect();
let total = matched.len();
let items: Vec<Value> = matched
.iter()
.skip(offset)
.take(limit)
.map(|m| {
serde_json::json!({
"name": m.name,
"short_name": m.short_name,
"base": hex(m.base_address.0),
"end": hex(m.end_address().0),
"size": m.size,
})
})
.collect();
Ok(paged(total, offset, "modules", items))
})
.await?;
json_result(v)
}
#[tool(
description = "Search guest memory in the current address space: scan `length` bytes from `start` for a byte `pattern` (contiguous lowercase hex). Returns {start, length, total, offset, returned, has_more, next_offset?, matches:[addr...]}; `total` is the full match count and `matches` is the offset/limit window."
)]
async fn search(
&self,
Parameters(SearchArgs {
start,
pattern,
length,
offset,
limit,
}): Parameters<SearchArgs>,
) -> Result<CallToolResult, McpError> {
let length = required_range("length", length, SEARCH_MAX_LENGTH)?;
let offset = offset.unwrap_or(0);
let limit = optional_limit("limit", limit, SEARCH_DEFAULT_LIMIT, SEARCH_MAX_LIMIT)?;
let v = self
.run(move |ctx| {
let start_addr = eval_addr(ctx, &start)?;
let needle = bytes_of_hex(&pattern)?;
if needle.is_empty() {
return Err(invalid_request("pattern must be at least one byte"));
}
let matches = ctx
.target
.search(VirtAddr(start_addr), &needle, length)
.map_err(ToolError::from)?;
let total = matches.len();
let items: Vec<Value> = matches
.into_iter()
.skip(offset)
.take(limit)
.map(|a| Value::from(hex(a)))
.collect();
let mut out = paged(total, offset, "matches", items);
out["start"] = Value::from(hex(start_addr));
out["length"] = Value::from(length);
Ok(out)
})
.await?;
json_result(v)
}
#[tool(
description = "Walk an intrusive _LIST_ENTRY list in the current address space. Given the list-head address, the record struct type, and the name of the _LIST_ENTRY field embedded in the record, returns each record's base address: {head, record_type, total, records:[addr...]}. Bounded to 1000 records. E.g. head=PsLoadedModuleList, record_type=_LDR_DATA_TABLE_ENTRY, link_field=InLoadOrderLinks."
)]
async fn walk_list(
&self,
Parameters(WalkListArgs {
head,
record_type,
link_field,
}): Parameters<WalkListArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let head_addr = eval_addr(ctx, &head)?;
let dtb = ctx.target.current_dtb();
let info = ctx
.target
.symbols
.find_type_across_modules(dtb, &record_type)
.ok_or_else(|| invalid_request(format!("unknown type: {record_type}")))?;
let link_offset = info.field_offset(&link_field).map_err(ToolError::from)?;
let records = ctx
.target
.walk_list(VirtAddr(head_addr), link_offset)
.map_err(ToolError::from)?;
Ok(serde_json::json!({
"head": hex(head_addr),
"record_type": record_type,
"total": records.len(),
"records": records.into_iter().map(hex).collect::<Vec<_>>(),
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Decode the _UNICODE_STRING at an address in the current address space to a string. Returns {address, value}."
)]
async fn read_unicode_string(
&self,
Parameters(AddressArgs { address }): Parameters<AddressArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let value = ctx
.target
.read_unicode_string(VirtAddr(addr))
.map_err(ToolError::from)?;
Ok(serde_json::json!({ "address": hex(addr), "value": value }))
})
.await?;
json_result(v)
}
#[tool(
description = "Read a NUL-terminated CHAR* (ASCII/UTF-8) string at an address in the current address space, up to max_len bytes (default 260). Returns {address, value}. The CHAR* counterpart to read_unicode_string; use it for ARC paths, boot LoadOptions, and other plain C strings."
)]
async fn read_c_string(
&self,
Parameters(ReadCStringArgs { address, max_len }): Parameters<ReadCStringArgs>,
) -> Result<CallToolResult, McpError> {
let max_len = optional_limit(
"max_len",
max_len,
READ_C_STRING_DEFAULT_LENGTH,
READ_C_STRING_MAX_LENGTH,
)?;
let v = self
.run(move |ctx| {
let addr = eval_addr(ctx, &address)?;
let value = ctx
.target
.read_c_string(VirtAddr(addr), max_len)
.map_err(ToolError::from)?;
Ok(serde_json::json!({ "address": hex(addr), "value": value }))
})
.await?;
json_result(v)
}
// --- context / run-control (state-changing) ---
#[tool(
description = "Switch the inspection context to a process by pid (scopes memory reads and breakpoints to it and loads its module symbols); returns {pid, name}"
)]
async fn attach_process(
&self,
Parameters(PidArgs { pid }): Parameters<PidArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let report = ctx.target.attach(pid).map_err(ToolError::from)?;
Ok(serde_json::json!({ "pid": pid, "name": report.name }))
})
.await?;
json_result(v)
}
#[tool(description = "Return the inspection context to the default kernel scope")]
async fn detach(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
ctx.target.detach();
Ok(serde_json::json!({ "detached": true }))
})
.await?;
json_result(v)
}
#[tool(
description = "Set a code breakpoint at an address (scoped to the attached process, or kernel-wide if not attached), with an optional break condition. Requires the VM halted (call interrupt first, or be stopped at a breakpoint); resume then wait_for_stop to run to a hit. Returns {id, address, condition}"
)]
async fn set_breakpoint(
&self,
Parameters(SetBreakpointArgs { address, condition }): Parameters<SetBreakpointArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
require_halted(ctx, "set_breakpoint")?;
let addr = eval_addr(ctx, &address)?;
let id = ctx
.add_breakpoint_with_condition(VirtAddr(addr), condition.clone())
.map_err(ToolError::from)?;
Ok(serde_json::json!({ "id": id, "address": hex(addr), "condition": condition }))
})
.await?;
json_result(v)
}
#[tool(
description = "Remove a breakpoint by id. Requires the VM halted (call interrupt first, or be stopped at a breakpoint)."
)]
async fn clear_breakpoint(
&self,
Parameters(BreakpointIdArgs { id }): Parameters<BreakpointIdArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
require_halted(ctx, "clear_breakpoint")?;
ctx.remove_breakpoint(id).map_err(ToolError::from)?;
Ok(serde_json::json!({ "cleared": id }))
})
.await?;
json_result(v)
}
#[tool(
description = "Re-arm a disabled breakpoint by id (re-patch its int3). Requires the VM halted (call interrupt first, or be stopped at a breakpoint)."
)]
async fn enable_breakpoint(
&self,
Parameters(BreakpointIdArgs { id }): Parameters<BreakpointIdArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
require_halted(ctx, "enable_breakpoint")?;
ctx.enable_breakpoint(id).map_err(ToolError::from)?;
Ok(serde_json::json!({ "id": id, "enabled": true }))
})
.await?;
json_result(v)
}
#[tool(
description = "Disable a breakpoint by id (restore the original byte) without forgetting it, so it can be re-enabled later. Requires the VM halted (call interrupt first, or be stopped at a breakpoint)."
)]
async fn disable_breakpoint(
&self,
Parameters(BreakpointIdArgs { id }): Parameters<BreakpointIdArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
require_halted(ctx, "disable_breakpoint")?;
ctx.disable_breakpoint(id).map_err(ToolError::from)?;
Ok(serde_json::json!({ "id": id, "enabled": false }))
})
.await?;
json_result(v)
}
#[tool(
description = "List breakpoints as {id, address, enabled, symbol, scope, condition, temporary} objects (scope is \"global\" for kernel-wide or \"name (pid)\" for a process-scoped breakpoint)"
)]
async fn list_breakpoints(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
let arr = ctx
.list_breakpoints()
.iter()
.map(|b| {
serde_json::json!({
"id": b.id,
"address": hex(b.address.0),
"enabled": b.enabled,
"symbol": b.symbol,
"scope": b.scope.label(),
"condition": b.condition,
"temporary": b.temporary,
})
})
.collect();
Ok(Value::Array(arr))
})
.await?;
json_result(v)
}
#[tool(
description = "Resume the VM (go). Non-blocking: returns immediately as {running:true, already_running:bool}. To wait for the next stop, call wait_for_stop; to see where it is now without resuming, call status."
)]
async fn resume(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
// Drain any stop the servicer caught so a real halt that already
// surfaced is reflected as `already_running:false` and the resume
// below actually advances past it, rather than the stale running
// value reporting `already_running:true` and doing nothing. A
// debugger-noise break-in is absorbed (already running, no-op).
ctx.settle_pending_stop().map_err(ToolError::from)?;
let already_running = ctx.backend.is_running();
if !already_running {
ctx.resume().map_err(ToolError::from)?;
}
Ok(serde_json::json!({ "running": true, "already_running": already_running }))
})
.await?;
json_result(v)
}
#[tool(
description = "Wait up to timeout_ms for the next stop WITHOUT resuming (default 10000, max 20000; 0 = default). Returns {stop:\"breakpoint\"|\"exception\"|\"bugcheck\"|\"target_reloaded\"} with context, {stop:\"running\"} if the wait elapsed (call again to keep waiting; no stops lost between calls), or {stop:\"halted\"} immediately if the VM is already parked with nothing pending. Does not resume; call resume to advance. Use short timeouts and poll; there is no indefinite wait."
)]
async fn wait_for_stop(
&self,
Parameters(WaitArgs { timeout_ms }): Parameters<WaitArgs>,
ct: CancellationToken,
) -> Result<CallToolResult, McpError> {
let timeout_ms = optional_timeout_ms(timeout_ms)?;
// Per-request cancel flag the actor's wait loop polls. Set when the
// client cancels/times out this request (`ct`) or the server is shutting
// down (`self.interrupt`), so an in-flight wait returns and frees the
// single-threaded actor instead of pinning it for the whole timeout.
let cancel = Arc::new(AtomicBool::new(false));
let watcher = {
let cancel = cancel.clone();
let shutdown = self.interrupt.clone();
tokio::spawn(async move {
loop {
if shutdown.load(Ordering::Relaxed) {
cancel.store(true, Ordering::Relaxed);
return;
}
tokio::select! {
_ = ct.cancelled() => {
cancel.store(true, Ordering::Relaxed);
return;
}
_ = tokio::time::sleep(Duration::from_millis(200)) => {}
}
}
})
};
let result = self
.run(move |ctx| {
// A stop the background `service_idle` caught and parked while the
// host was idle (e.g. a breakpoint that fired between tool calls)
// is the proper event for this wait; return it before waiting for
// a new one.
if let Some(parked) = ctx.take_parked_stop() {
return Ok(continue_outcome_json(ctx, parked));
}
// Always bounded over MCP (see optional_timeout_ms): the caller
// polls by calling again rather than waiting indefinitely. Never
// resumes, so a held stop (e.g. the early-boot reload) surfaces in
// place instead of being run past.
let timeout = Some(Duration::from_millis(timeout_ms));
let outcome = ctx
.wait_for_stop_bounded(timeout, &cancel)
.map_err(ToolError::from)?;
Ok(continue_outcome_json(ctx, outcome))
})
.await;
watcher.abort();
let v = result?;
json_result(v)
}
#[tool(
description = "Read-only run-control state (where am I): {running, current_thread, rip, symbol, process:{pid,name,eprocess}|null, coherent}. rip/symbol are null while running. coherent=false means the guest rebooted and rediscovery is still in progress, so process/module enumeration is not yet meaningful; resume + wait_for_stop to advance rather than reading stale state."
)]
async fn status(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
let s = ctx.run_status();
let process = s.process.map(|(pid, name, eprocess)| {
serde_json::json!({ "pid": pid, "name": name, "eprocess": hex(eprocess) })
});
Ok(serde_json::json!({
"running": s.running,
"current_thread": s.current_thread,
"rip": s.rip.map(hex),
"symbol": s.symbol,
"process": process,
"coherent": s.coherent,
"kernel_base": hex(s.kernel_base),
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Pause a running VM (e.g. before reading registers/backtrace); returns {already_halted, rip}. If already halted, no action is taken. Resume with resume."
)]
async fn interrupt(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
// A stop the servicer already caught means the VM is halted now;
// ingest it so `already_halted` is truthful at a real stop and we
// don't send a redundant break-in over it (a debugger-noise break-in
// is absorbed, so already_halted is false and we interrupt for real).
ctx.settle_pending_stop().map_err(ToolError::from)?;
let already_halted = !ctx.backend.is_running();
let event_rip = if already_halted {
None
} else {
// Route through the context so the selected thread tracks
// the halt (rather than calling the backend raw).
ctx.interrupt().map_err(ToolError::from)?.program_counter
};
let rip = event_rip.or_else(|| {
ctx.backend
.read_registers()
.ok()
.and_then(|r| ctx.register_map.read_u64("rip", &r).ok())
});
Ok(serde_json::json!({
"already_halted": already_halted,
"rip": rip.map(hex),
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Single-step one instruction on the current thread. Requires the VM halted (call interrupt first, or be stopped at a breakpoint). Returns {rip, symbol}"
)]
async fn step(&self) -> Result<CallToolResult, McpError> {
let v = self
.run(|ctx| {
require_halted(ctx, "step")?;
ctx.step().map_err(ToolError::from)?;
let regs = ctx.backend.read_registers().map_err(ToolError::from)?;
let rip = ctx.register_map.read_u64("rip", ®s).unwrap_or(0);
let symbol = ctx.target.closest_symbol_current_context(VirtAddr(rip));
Ok(serde_json::json!({ "rip": hex(rip), "symbol": symbol }))
})
.await?;
json_result(v)
}
#[tool(
description = "Step over the current instruction on the current thread: if it's a call, run to its return site, otherwise single-step. Requires the VM halted (call interrupt first, or be stopped at a breakpoint). Returns the same shape as wait_for_stop ({stop:\"step\"} on completion, or breakpoint/bugcheck/exception if one is hit en route)."
)]
async fn step_over(&self) -> Result<CallToolResult, McpError> {
let cancel = self.interrupt.clone();
let v = self
.run(move |ctx| {
require_halted(ctx, "step_over")?;
let outcome = ctx.step_over(&cancel).map_err(ToolError::from)?;
Ok(continue_outcome_json(ctx, outcome))
})
.await?;
json_result(v)
}
#[tool(
description = "Step out of the current function: run to the caller's return address. Requires the VM halted (call interrupt first, or be stopped at a breakpoint). Returns the same shape as wait_for_stop ({stop:\"step\"} on completion, or breakpoint/bugcheck/exception if one is hit en route)."
)]
async fn step_out(&self) -> Result<CallToolResult, McpError> {
let cancel = self.interrupt.clone();
let v = self
.run(move |ctx| {
require_halted(ctx, "step_out")?;
let outcome = ctx.step_out(&cancel).map_err(ToolError::from)?;
Ok(continue_outcome_json(ctx, outcome))
})
.await?;
json_result(v)
}
#[tool(
description = "Select the current inspection thread (a vCPU id, as listed by vcpus or in threads' `active` field) so registers/backtrace/step operate on it. Returns {current_thread}."
)]
async fn set_current_thread(
&self,
Parameters(ThreadIdArgs { thread }): Parameters<ThreadIdArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
ctx.set_current_thread(&thread).map_err(ToolError::from)?;
Ok(serde_json::json!({ "current_thread": thread }))
})
.await?;
json_result(v)
}
// --- guest writes (state-changing) ---
#[tool(
description = "Write bytes to guest virtual memory (scoped to the attached process, or kernel-wide if not attached). `hex` is contiguous lowercase hex, 1-4096 bytes (the form read_memory returns). Works while the guest runs; if the guest may be concurrently touching the same bytes, interrupt first to avoid a torn write. Returns {address, length}."
)]
async fn write_memory(
&self,
Parameters(WriteMemoryArgs { address, hex: data }): Parameters<WriteMemoryArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
let bytes = bytes_of_hex(&data)?;
if bytes.is_empty() || bytes.len() > WRITE_MEMORY_MAX_LENGTH {
return Err(invalid_request(format!(
"hex must decode to 1..={WRITE_MEMORY_MAX_LENGTH} bytes"
)));
}
let addr = eval_addr(ctx, &address)?;
ctx.target
.current_process()
.memory()
.write_bytes(VirtAddr(addr), &bytes)
.map_err(ToolError::from)?;
Ok(serde_json::json!({
"address": hex(addr),
"length": bytes.len(),
}))
})
.await?;
json_result(v)
}
#[tool(
description = "Set a single register on the current thread. `value` is a debugger expression. Requires the VM halted (call interrupt first, or be stopped at a breakpoint). Returns {name, value}."
)]
async fn set_register(
&self,
Parameters(SetRegisterArgs { name, value }): Parameters<SetRegisterArgs>,
) -> Result<CallToolResult, McpError> {
let v = self
.run(move |ctx| {
require_halted(ctx, "set_register")?;
let value = eval_addr(ctx, &value)?;
ctx.write_register(&name, value)
.map_err(ToolError::from)?;
Ok(serde_json::json!({ "name": name, "value": hex(value) }))
})
.await?;
json_result(v)
}
}
#[tool_handler]
impl rmcp::ServerHandler for NtoseyeMcp {
fn get_info(&self) -> ServerInfo {
ServerInfo {
protocol_version: ProtocolVersion::V_2025_06_18,
capabilities: ServerCapabilities::builder().enable_tools().build(),
server_info: Implementation {
name: env!("CARGO_PKG_NAME").to_string(),
version: env!("CARGO_PKG_VERSION").to_string(),
..Implementation::from_build_env()
},
instructions: Some(
"ntoseye: introspect and control a live Windows kernel running under \
KVM/QEMU. Read-only tools for process/thread/module/driver \
enumeration, memory and struct reads, disassembly, backtraces, \
page-table walks, and symbol/type lookup. State-changing tools: \
attach_process/detach (scope the context to a process), \
set_breakpoint/clear_breakpoint/list_breakpoints, run-control \
(resume, wait_for_stop, interrupt, step, step_over, step_out, \
set_current_thread), and guest writes \
(write_memory, set_register). Run-control is split: resume goes \
(non-blocking), wait_for_stop polls for the next stop without \
resuming, status reports where it is now, interrupt pauses. \
Typical flow for a user-mode breakpoint: attach_process(pid) → \
interrupt → set_breakpoint(expr) → resume → wait_for_stop(timeout) \
(poll until stop:\"breakpoint\"). \
The guest runs freely by default (memory/process/struct reads work \
live). registers/backtrace need the CPU halted: call interrupt \
first (or be stopped at a breakpoint), inspect, then resume to let \
the guest run again. Breakpoint mutation, step, and set_register \
require the VM halted; write_memory works live. After a reboot, \
status reports coherent:false until rediscovery finishes; \
wait_for_stop for it rather than enumerating stale state. \
Tools that take an address accept a debugger expression (symbol, \
register, hex, arithmetic). Addresses in tool output are 0x hex \
strings (JSON has no hex numbers); ids/pids/sizes stay decimal."
.to_string(),
),
}
}
}
fn is_loopback_http_bind(addr: &str) -> bool {
if let Ok(socket) = addr.parse::<std::net::SocketAddr>() {
return socket.ip().is_loopback();
}
let Some((host, _port)) = addr.rsplit_once(':') else {
return false;
};
let host = host.trim_start_matches('[').trim_end_matches(']');
host.eq_ignore_ascii_case("localhost")
|| host
.parse::<std::net::IpAddr>()
.map(|ip| ip.is_loopback())
.unwrap_or(false)
}
/// Whether a browser `Origin` header (`scheme://host[:port]`, no path) names a
/// loopback host. Used to gate cross-origin access to the loopback HTTP bind so a
/// website the user merely visits can't reach the debugger via 127.0.0.1.
fn is_loopback_origin(origin: &str) -> bool {
let Some((_scheme, rest)) = origin.split_once("://") else {
return false;
};
// host[:port]; a bracketed IPv6 literal ([::1]:port) keeps the colons inside
// the brackets, so peel those off before splitting on the port colon.
let host = if let Some(after_bracket) = rest.strip_prefix('[') {
match after_bracket.split_once(']') {
Some((host, _port)) => host,
None => return false,
}
} else {
rest.split(':').next().unwrap_or(rest)
};
host.eq_ignore_ascii_case("localhost")
|| host
.parse::<std::net::IpAddr>()
.map(|ip| ip.is_loopback())
.unwrap_or(false)
}
fn check_http_bind_policy(addr: &str, unsafe_http: bool) -> anyhow::Result<()> {
if is_loopback_http_bind(addr) {
return Ok(());
}
if unsafe_http {
eprintln!(
"ntoseye-mcp: warning: HTTP bind {addr} is not loopback; debugger control tools are reachable by clients that can access this address"
);
Ok(())
} else {
Err(anyhow::anyhow!(
"refusing non-loopback MCP HTTP bind {addr}; use 127.0.0.1:PORT for local browser clients or pass --unsafe-http to expose debugger control tools on the network"
))
}
}
/// Attach (on a dedicated thread) and serve the MCP protocol until the client
/// disconnects. Synchronous entry point; it owns its own tokio runtime, so the
/// rest of the binary stays runtime-free.
///
/// `http` selects the transport: `None` serves over **stdio** (the client
/// launches this binary as a subprocess, e.g. Claude Desktop / Claude Code), while
/// `Some(addr)` serves the **Streamable HTTP** transport on `addr` (for web
/// clients like llama.cpp's webui that connect over the network and can't spawn a
/// subprocess). HTTP binds are loopback-only unless `unsafe_http` is set. Both
/// transports drive the same single session actor.
pub fn run(
backend: String,
connect: Option<String>,
http: Option<String>,
unsafe_http: bool,
) -> anyhow::Result<()> {
if let Some(addr) = http.as_deref() {
check_http_bind_policy(addr, unsafe_http)?;
}
// The stdio transport speaks MCP on stdout, so all logging goes to stderr.
eprintln!("ntoseye-mcp: attaching ({backend})...");
let (tx, service_pending) = spawn_session(backend, connect)?;
// A sender kept aside so we can drive teardown even after the service (which
// owns its own sender) is dropped.
let shutdown_tx = tx.clone();
// Shared with the handlers so shutdown can interrupt an in-flight
// `wait_for_stop` (otherwise the actor stays busy and never reaches
// cleanup, leaving the VM frozen).
let interrupt = Arc::new(AtomicBool::new(false));
let interrupt_for_signal = interrupt.clone();
let runtime = tokio::runtime::Runtime::new()?;
runtime.block_on(async move {
// Background servicing ticker: periodically nudge the actor to service the
// guest while idle (see `Command::Service`), so a wrong-process hit on a
// shared-page breakpoint doesn't leave it frozen between tool calls.
// `Skip` keeps it from flooding the queue if the actor is busy; the task
// exits once the actor's channel closes (send fails).
let service_tx = tx.clone();
tokio::spawn(async move {
let mut tick = tokio::time::interval(SERVICE_TICK);
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
tick.tick().await;
// Only enqueue when none is outstanding (the actor clears the flag
// as it services), so at most one `Service` is ever queued even if
// the actor is busy in a long wait.
if service_pending.swap(true, Ordering::AcqRel) {
continue;
}
if service_tx.send(Command::Service).is_err() {
break;
}
}
});
let serve = async {
match http {
Some(addr) => {
eprintln!(
"ntoseye-mcp: attached; serving Streamable HTTP at http://{addr}/mcp"
);
serve_http(tx, addr, unsafe_http, interrupt).await
}
None => {
eprintln!("ntoseye-mcp: attached; serving over stdio");
let service = NtoseyeMcp::new(tx, interrupt).serve(stdio()).await?;
service.waiting().await?;
Ok(())
}
}
};
// Serve until the client disconnects (or the server errors), or until
// Ctrl+C; either way fall through to teardown.
let result = tokio::select! {
r = serve => r,
_ = tokio::signal::ctrl_c() => {
eprintln!("ntoseye-mcp: interrupted");
Ok(())
}
};
// Ask the actor to remove our breakpoints and resume the VM before we
// exit, so Ctrl+C doesn't leave the guest frozen with int3s installed.
// Set the interrupt first so any in-flight wait returns and the actor is
// free to process the Shutdown.
eprintln!("ntoseye-mcp: resuming VM and cleaning up...");
interrupt_for_signal.store(true, Ordering::Relaxed);
let (ack_tx, ack_rx) = oneshot::channel();
if shutdown_tx.send(Command::Shutdown { ack: ack_tx }).is_ok() {
let _ = tokio::time::timeout(std::time::Duration::from_secs(5), ack_rx).await;
}
result
})
}
/// Serve the Streamable HTTP transport on `addr`, mounting the MCP service at
/// `/mcp`. Every HTTP session gets a clone of the handler (cheap; it holds only
/// the actor's channel sender), so all connections funnel to the one live
/// debugger session.
async fn serve_http(
tx: mpsc::UnboundedSender<Command>,
addr: String,
unsafe_http: bool,
interrupt: Arc<AtomicBool>,
) -> anyhow::Result<()> {
use rmcp::transport::StreamableHttpService;
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use tower_http::cors::{AllowOrigin, Any, CorsLayer};
let template = NtoseyeMcp::new(tx, interrupt);
let service = StreamableHttpService::new(
move || Ok(template.clone()),
LocalSessionManager::default().into(),
Default::default(),
);
// Methods/headers stay permissive for the Streamable HTTP handshake. Origin
// is the exposure that matters: loopback binds trust only loopback browser
// origins, while `--unsafe-http` widens it to any origin.
let allow_origin = if unsafe_http {
AllowOrigin::any()
} else {
AllowOrigin::predicate(|origin, _parts| {
origin.to_str().is_ok_and(is_loopback_origin)
})
};
let cors = CorsLayer::new()
.allow_origin(allow_origin)
.allow_methods(Any)
.allow_headers(Any)
.expose_headers(Any);
let router = axum::Router::new()
.nest_service("/mcp", service)
.layer(cors);
let listener = tokio::net::TcpListener::bind(&addr).await?;
axum::serve(listener, router).await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn loopback_origins_are_trusted() {
assert!(is_loopback_origin("http://localhost"));
assert!(is_loopback_origin("http://localhost:8080"));
assert!(is_loopback_origin("http://127.0.0.1:3000"));
assert!(is_loopback_origin("https://127.0.0.1"));
assert!(is_loopback_origin("http://[::1]:9000"));
assert!(is_loopback_origin("http://LOCALHOST:1234"));
}
#[test]
fn non_loopback_origins_are_rejected() {
assert!(!is_loopback_origin("http://meow.example.com"));
assert!(!is_loopback_origin("https://meow.test:443"));
assert!(!is_loopback_origin("http://10.0.0.5:8080"));
// `null` (sandboxed iframes, file://) is not a loopback host.
assert!(!is_loopback_origin("null"));
// Malformed / no scheme.
assert!(!is_loopback_origin("127.0.0.1"));
assert!(!is_loopback_origin(""));
}
}