//! MCP server: the debugger's REPL command language over the Model Context
//! Protocol, for clients that cannot run Python themselves.
//!
//! One `command` tool runs a REPL line and returns its text; `resume`
//! (non-blocking), `wait_for_stop` (bounded), `interrupt`, and `status` cover
//! run control; `open`/`close` manage the single session slot.
use rmcp::{
ErrorData as McpError, ServiceExt,
handler::server::router::tool::ToolRouter,
handler::server::wrapper::Parameters,
model::{
CallToolResult, ContentBlock, 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::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
use crate::bugchecks::{analyze_bugcheck, bugcheck_from_dump_info, current_bugcheck};
use crate::dbg_backend::ContinueDisposition;
use crate::diagnostics;
use crate::error::Error;
use crate::kd::KdMemorySource;
use crate::output;
use crate::repl::{DispatchContext, Flow, RemoteClient, ReplState, ReplStore};
use crate::session::{ContinueOutcome, Session};
use crate::types::VirtAddr;
use crate::view;
use crate::{Backend, TargetSpec};
/// The session actor's state: the (`!Send`) session plus the REPL state the
/// `command` tool keeps between calls (built on first use).
struct Actor {
ctx: Session,
repl: Option<ReplStore>,
}
/// A unit of work run on the actor thread. MCP handlers are async/`Send`
/// but `Session` is not, so a dedicated thread owns it; handlers send
/// closures and await one reply.
type Job = Box<dyn FnOnce(&mut Actor) -> Result<CallToolResult, ToolError> + Send>;
enum Command {
Run {
job: Job,
reply: oneshot::Sender<Result<CallToolResult, ToolError>>,
},
/// Clean up the session and stop the actor. The guest resumes only after
/// every debugger-owned breakpoint is restored; cleanup failure is reported
/// and leaves the target halted rather than running with an orphaned int3.
Shutdown { ack: oneshot::Sender<()> },
/// A periodic nudge (from the background ticker) for the actor to service
/// the guest while otherwise idle, absorbing wrong-process hits on a
/// shared-page breakpoint so they don't leave it frozen between calls.
Service,
}
/// How often the background ticker nudges the actor (see [`Command::Service`]).
const SERVICE_TICK: Duration = Duration::from_millis(20);
const CONTINUE_DEFAULT_TIMEOUT_MS: u64 = 10_000;
/// Capped well under common MCP client request timeouts (30 s is typical) 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;
fn cleanup_session(ctx: &mut Session) {
if let Err(error) = ctx.cleanup_for_exit() {
diagnostics::eprint_warning(format!(
"debugger cleanup failed; target was not resumed: {error}"
));
}
}
/// Spawn the actor thread that owns the session. The backend is constructed
/// *on this thread* so the non-`Send` state never crosses a thread boundary.
/// Returns the sender the handlers use to reach it and the ticker's coalescing
/// flag.
fn spawn_session(
spec: TargetSpec,
) -> 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, so
// a long wait can't let a burst of them pile up in the unbounded channel.
let service_pending = Arc::new(AtomicBool::new(false));
let service_pending_actor = service_pending.clone();
std::thread::spawn(move || {
let is_dump = matches!(spec, TargetSpec::Dump(_));
let mut actor = match Session::open(&spec) {
Ok(ctx) => {
let _ = ready_tx.send(Ok(()));
Actor { ctx, repl: None }
}
Err(e) => {
let _ = ready_tx.send(Err(e.to_string()));
return;
}
};
// For live targets the MCP keeps the guest running between calls;
// tools that need a stopped target ask the client to `interrupt`.
// Dumps are always halted.
if !is_dump && !actor.ctx.backend.is_running() {
let _ = actor.ctx.backend.continue_execution();
}
// `blocking_recv` is valid here: a plain std thread, no runtime. Clean
// up on an explicit `Shutdown` (Ctrl+C / client disconnect) and if the
// channel closes outright, so the VM is never left frozen.
loop {
match rx.blocking_recv() {
Some(Command::Run { job, reply }) => {
let _ = reply.send(job(&mut actor));
}
Some(Command::Service) => {
service_pending_actor.store(false, Ordering::Release);
actor.ctx.service_idle();
}
Some(Command::Shutdown { ack }) => {
cleanup_session(&mut actor.ctx);
drop(actor);
let _ = ack.send(());
break;
}
None => {
cleanup_session(&mut actor.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")),
}
}
/// The one shared session slot all transports funnel into. `Opening(gen)`
/// reserves the slot while `open` builds a session off-thread, so a concurrent
/// open can't race the vacancy check and leak a second actor. The generation
/// lets [`OpeningGuard`] detect that its reservation was canceled by a
/// concurrent `close`.
enum SessionSlot {
Vacant,
Opening(u64),
Active(mpsc::UnboundedSender<Command>),
}
static OPENING_GENERATION: AtomicU64 = AtomicU64::new(0);
type SharedSession = Arc<std::sync::Mutex<SessionSlot>>;
/// RAII guard that rolls `SessionSlot` back to `Vacant` if the opening future
/// is canceled (e.g. MCP client timeout). [`OpeningGuard::promote`] installs
/// the active sender and defuses the rollback.
struct OpeningGuard {
session: SharedSession,
generation: u64,
}
impl OpeningGuard {
fn claim(session: &SharedSession) -> Result<Self, McpError> {
let mut guard = session.lock().unwrap();
match *guard {
SessionSlot::Vacant => {
let id = OPENING_GENERATION.fetch_add(1, Ordering::Relaxed);
*guard = SessionSlot::Opening(id);
Ok(Self {
session: session.clone(),
generation: id,
})
}
_ => Err(McpError::invalid_request(
"a debugger session is already active or opening",
None,
)),
}
}
fn promote(self, tx: mpsc::UnboundedSender<Command>) -> Result<(), McpError> {
let mut guard = self.session.lock().unwrap();
match *guard {
SessionSlot::Opening(id) if id == self.generation => {
*guard = SessionSlot::Active(tx);
Ok(())
}
_ => Err(McpError::internal_error("session open was canceled", None)),
}
}
}
impl Drop for OpeningGuard {
fn drop(&mut self) {
let mut guard = self.session.lock().unwrap();
if matches!(*guard, SessionSlot::Opening(id) if id == self.generation) {
*guard = SessionSlot::Vacant;
}
}
}
struct InterruptResetGuard(Arc<AtomicBool>);
impl Drop for InterruptResetGuard {
fn drop(&mut self) {
self.0.store(false, Ordering::Relaxed);
}
}
#[derive(Clone)]
struct NtoseyeMcp {
session: SharedSession,
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>,
}
#[derive(Clone, Copy, Debug, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
enum BackendArg {
Kd,
#[serde(rename = "kdnet")]
KdNet,
Gdb,
Memory,
Dump,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct OpenArgs {
#[schemars(
description = "What to attach to: kd (KD over Unix socket), kdnet (KDNET over encrypted UDP), gdb (GDB remote stub), memory (physical memory only, no debug transport), or dump (a Windows kernel crash dump file)"
)]
backend: BackendArg,
#[schemars(
description = "Connection target: Unix socket path for kd (default /tmp/ntoseye-kd.sock), listen address for kdnet (default 0.0.0.0:50000), host:port for gdb (default 127.0.0.1:1234), absolute .dmp path for dump (required). Not used by memory."
)]
connect: Option<String>,
#[schemars(
description = "KDNET encryption key as four base-36 components. Required for kdnet only."
)]
key: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct CommandArgs {
#[schemars(
description = "A REPL command line in ntoseye's WinDbg-style syntax, e.g. `!process 0 0`, `dt nt!_EPROCESS ffff...`, `k`, `bp nt!NtCreateFile`, `dq rsp l8`, `u rip`, `lm`. Several commands may be separated by `;`. Run `help` for the list and `help <cmd>` for one command."
)]
line: String,
}
#[derive(Clone, Copy, Debug, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
enum ContinueDispositionArg {
Handled,
NotHandled,
}
impl From<ContinueDispositionArg> for ContinueDisposition {
fn from(disposition: ContinueDispositionArg) -> Self {
match disposition {
ContinueDispositionArg::Handled => Self::Handled,
ContinueDispositionArg::NotHandled => Self::NotHandled,
}
}
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct ResumeArgs {
#[schemars(
description = "Exception acknowledgment: handled (default) or not_handled. not_handled requires native transport support (currently KD) and otherwise returns an error."
)]
disposition: Option<ContinueDispositionArg>,
}
#[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 the default). Bounded by design: poll by calling again while it returns running. A long wait blocks every other tool on the single debugger session."
)]
timeout_ms: Option<u64>,
}
/// A tool failure from the session, classified so a guest memory fault stays
/// distinguishable from an internal bug. Argument errors never reach the
/// actor; handlers reject them as `McpError` params before dispatch.
enum ToolError {
/// 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::AddressNotInDump(_)
| Error::BadPhysicalAddress(_)
| Error::PartialRead(_)
| Error::PartialWrite(_)
| Error::BufferNotEnough
| Error::InvalidRange => ToolError::Memory(msg),
_ => ToolError::Internal(msg),
}
}
}
impl From<ToolError> for McpError {
fn from(e: ToolError) -> Self {
match e {
// 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 optional_timeout_ms(value: Option<u64>) -> Result<u64, McpError> {
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)
}
}
/// Structured tool result. rmcp's `structured()` also embeds a compact-text
/// copy in `content`, so clients that ignore `structuredContent` still get
/// the JSON.
fn json(v: Value) -> Result<CallToolResult, ToolError> {
Ok(CallToolResult::structured(v))
}
/// 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}")
}
/// Render a [`ContinueOutcome`] as JSON, enriching breakpoint/exception stops
/// with the resolved symbol and the stop's context from `ctx`.
///
/// The context is three separate answers, because collapsing them is how a
/// client ends up trusting an attached scope many stops old as the process
/// that is executing: `attached_process` is the operator's inspection scope,
/// `stopped_process` owns the page tables the stopped vCPU has loaded, and
/// `stopped_thread` is the Windows thread it is running.
fn continue_outcome_json(ctx: &mut Session, outcome: ContinueOutcome) -> Value {
let (stopped_process, stopped_thread) = ctx.stopped_context();
let stopped_process = stopped_process.map(|p| view::to_json(&view::process(&p)));
let stopped_thread = stopped_thread.map(|t| view::to_json(&view::thread(&t, None)));
let attached_process = ctx
.target
.current_process_info
.as_ref()
.map(|p| view::to_json(&view::process(p)));
let symbol_at = |rip: u64| ctx.target.closest_symbol_current_context(VirtAddr(rip));
match outcome {
ContinueOutcome::Breakpoint {
id,
address,
symbol,
temporary,
rip,
condition_error,
..
} => {
let bp = ctx.breakpoint(id);
let watch_access = bp.and_then(|bp| bp.watch_access_name());
serde_json::json!({
"stop": if watch_access.is_some() { "watchpoint" } else { "breakpoint" },
"id": id,
"address": hex(address),
"symbol": symbol.or_else(|| symbol_at(rip)),
"temporary": temporary,
"rip": hex(rip),
"attached_process": attached_process,
"stopped_process": stopped_process,
"stopped_thread": stopped_thread,
"watch_access": watch_access,
"watch_length": bp.and_then(|bp| bp.watch_length()),
"condition_error": condition_error,
})
}
ContinueOutcome::Bugcheck { rip, info } => {
let analysis = info
.map(|i| analyze_bugcheck(&ctx.target, &i))
.or_else(|| current_bugcheck(&ctx.target))
.or_else(|| bugcheck_from_dump_info(&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,
first_chance,
exception_address,
} => serde_json::json!({
"stop": "exception",
"rip": hex(rip),
"exception_code": exception_code,
"first_chance": first_chance,
"exception_address": exception_address.map(hex),
"symbol": symbol_at(rip),
"attached_process": attached_process,
"stopped_process": stopped_process,
"stopped_thread": stopped_thread,
}),
ContinueOutcome::Step { rip } => serde_json::json!({
"stop": "step",
"rip": hex(rip),
"symbol": symbol_at(rip),
"attached_process": attached_process,
"stopped_process": stopped_process,
"stopped_thread": stopped_thread,
}),
ContinueOutcome::TargetReloaded {
kernel_base,
coherent,
} => {
let note = if coherent {
"The guest rebooted and debugger state is now fully rebuilt against \
the new kernel. The VM is halted at an internal KD break-in (an \
arbitrary landing site). Every prior address (eprocess, ethread, \
module base, dtb) is invalid; re-enumerate 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/module \
enumeration is UNAVAILABLE at this stop. Every prior address is now \
invalid. Use this stop to debug early boot (breakpoints on init paths \
work); otherwise resume, poll wait_for_stop, and enumerate only once \
status reports coherent:true."
};
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.
"event": false,
"rip": hex(rip),
"symbol": symbol_at(rip),
"attached_process": attached_process,
"stopped_process": stopped_process,
"stopped_thread": stopped_thread,
"coherent": ctx.kernel_coherent(),
}),
}
}
#[tool_router]
impl NtoseyeMcp {
fn new(session: SharedSession, interrupt: Arc<AtomicBool>) -> Self {
Self {
session,
tool_router: Self::tool_router(),
interrupt,
}
}
/// Ship a job to the session actor and await its reply.
async fn run<F>(&self, job: F) -> Result<CallToolResult, McpError>
where
F: FnOnce(&mut Actor) -> Result<CallToolResult, ToolError> + Send + 'static,
{
let tx = {
let guard = self.session.lock().unwrap();
match &*guard {
SessionSlot::Active(tx) => tx.clone(),
SessionSlot::Opening(_) => {
return Err(McpError::invalid_request(
"a debugger session is still opening; retry shortly",
None,
));
}
SessionSlot::Vacant => {
return Err(McpError::invalid_request(
"no debugger session is active; call open to attach",
None,
));
}
}
};
let (reply_tx, reply_rx) = oneshot::channel();
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(McpError::from)
}
#[tool(
description = "Run one line of ntoseye's WinDbg-style REPL and return its text output (styling stripped). This is the whole debugger: `help` lists every command; `help <cmd>` explains one. Common: `!process 0 0` / `!process <pid|name>` (processes), `.process /p <pid>` / `.process 0` (address-space scope; `attach`/`detach` aliases), `lm` (modules), `dt <type> [addr]` (struct layout/read), `x <mod>!<pat>` (symbols), `dq/dd/db <addr> [l<n>]` (memory), `u <addr>` (disassemble), `k` (backtrace; VM must be halted), `r` (registers; halted), `bp/bl/bc/bd/be` (breakpoints; halted), `!pte <addr>`, `!analyze`. Commands that resume until the next stop (g, gh, gn, p, pa, pc, pt, ph, ta, tc, tt, th, gu, wt, .reboot, .crash) are refused here because they would block the session: use the resume tool then poll wait_for_stop. `t`/`si` (one instruction) is allowed. Addresses accept expressions (symbols, registers, hex, arithmetic, poi())."
)]
async fn command(
&self,
Parameters(CommandArgs { line }): Parameters<CommandArgs>,
) -> Result<CallToolResult, McpError> {
self.run(move |actor| {
let store = actor.repl.take().unwrap_or_else(|| {
ReplStore::new(&actor.ctx, DispatchContext::Remote(RemoteClient::Mcp))
});
let mut state = ReplState::attach(&mut actor.ctx, store);
state.line = line.trim().to_string();
let (result, mut text) = output::capture(|| state.dispatch_line(&line));
actor.repl = Some(state.detach());
let ok = match result {
Ok(Flow::Continue | Flow::Quit) => true,
Ok(Flow::Denied) => false,
Err(e) => {
text.push_str(&format!("error: {e}\n"));
false
}
};
let content = vec![ContentBlock::text(text)];
Ok(if ok {
CallToolResult::success(content)
} else {
CallToolResult::error(content)
})
})
.await
}
#[tool(
description = "Read-only run-control state (where am I): {running, current_thread, rip, symbol, attached_process:{pid,name,eprocess}|null, stopped_process:{pid,name,eprocess}|null, stopped_thread|null, coherent, kernel_base}. The context fields are three different answers: attached_process is the inspection scope memory commands read through (set with `.process`, persists across resumes), stopped_process owns the page tables the stopped vCPU has loaded, and stopped_thread is the Windows thread it is running - a thread attached to another address space runs on borrowed page tables, so the last two can legitimately differ. rip/symbol are null while running. coherent=false means the guest rebooted and rediscovery is still in progress, so enumeration is not yet meaningful; resume + wait_for_stop rather than reading stale state."
)]
async fn status(&self) -> Result<CallToolResult, McpError> {
self.run(|actor| json(view::to_json(&view::run_status(&actor.ctx.run_status()))))
.await
}
#[tool(
description = "Resume the VM with an optional exception acknowledgment (handled by default, or not_handled; KD only). Non-blocking: returns {running:true, already_running, disposition}. To wait for the next stop, call wait_for_stop."
)]
async fn resume(
&self,
Parameters(ResumeArgs { disposition }): Parameters<ResumeArgs>,
) -> Result<CallToolResult, McpError> {
let disposition =
disposition.map_or(ContinueDisposition::Handled, ContinueDisposition::from);
self.run(move |actor| {
let ctx = &mut actor.ctx;
// Drain any stop the servicer caught so a real halt that already
// surfaced is reflected as `already_running:false` and the resume
// actually advances past it.
ctx.settle_pending_stop()?;
let already_running = ctx.backend.is_running();
if !already_running {
ctx.resume_with_disposition(disposition)?;
}
json(serde_json::json!({
"running": true,
"already_running": already_running,
"disposition": disposition.name(),
}))
})
.await
}
#[tool(
description = "Wait up to timeout_ms for the next stop WITHOUT resuming (default 10000, max 20000; 0 = default). Returns {stop:\"breakpoint\"|\"watchpoint\"|\"exception\"|\"bugcheck\"|\"step\"|\"target_reloaded\"} with context, {stop:\"running\"} if the wait elapsed (call again; no stops are lost between calls), or {stop:\"halted\"} immediately if the VM is already parked with nothing pending. Does not resume; call resume to advance. Poll with short timeouts; 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 this request (`ct`) or the server is shutting down
// (`self.interrupt`), so an in-flight wait returns and frees the 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 |actor| {
let ctx = &mut actor.ctx;
let outcome =
ctx.wait_for_stop_bounded(Some(Duration::from_millis(timeout_ms)), &cancel)?;
json(continue_outcome_json(ctx, outcome))
})
.await;
watcher.abort();
result
}
#[tool(
description = "Pause a running VM (needed before k, r, bp, t and other halted-only commands); returns {already_halted, rip}. If already halted, no action is taken. Resume with resume."
)]
async fn interrupt(&self) -> Result<CallToolResult, McpError> {
self.run(|actor| {
let ctx = &mut actor.ctx;
// A stop the servicer already caught means the VM is halted now;
// ingest it so `already_halted` is truthful and we don't send a
// redundant break-in over it.
ctx.settle_pending_stop()?;
let already_halted = !ctx.backend.is_running();
let event_rip = if already_halted {
None
} else {
ctx.interrupt()?.program_counter
};
let rip = event_rip.or_else(|| {
ctx.read_registers()
.ok()
.and_then(|r| ctx.register_map.read_u64("rip", &r).ok())
});
json(serde_json::json!({
"already_halted": already_halted,
"rip": rip.map(hex),
}))
})
.await
}
#[tool(
description = "Attach to a target: a live Windows VM over kd/kdnet/gdb/memory, or a crash dump (backend=dump, connect=<path>). Must be called before other tools when the server was started without --connect/--dump. Only one session can be active at a time. Returns {status:\"connected\", backend, connect, processors}."
)]
async fn open(
&self,
Parameters(OpenArgs {
backend,
connect,
key,
}): Parameters<OpenArgs>,
) -> Result<CallToolResult, McpError> {
let spec = match backend {
BackendArg::Dump => {
if key.is_some() {
return Err(invalid_params("dump does not use 'key'"));
}
let path = connect
.clone()
.ok_or_else(|| invalid_params("dump requires 'connect' (the .dmp path)"))?;
TargetSpec::Dump(PathBuf::from(path))
}
live => {
// The memory source is the operator's call (`--memory-source`
// on the CLI); `auto` validates host memory and falls back to
// KD, which is right whenever nobody knows better.
TargetSpec::Live {
backend: match live {
BackendArg::Kd => Backend::Kd,
BackendArg::KdNet => Backend::KdNet,
BackendArg::Gdb => Backend::Gdb,
BackendArg::Memory => Backend::Memory,
BackendArg::Dump => unreachable!("handled above"),
},
connect: connect.clone(),
kdnet_key: key,
memory_source: KdMemorySource::Auto,
}
}
};
spec.validate().map_err(invalid_params)?;
let label = match &spec {
TargetSpec::Dump(_) => "dump".to_string(),
TargetSpec::Live { backend, .. } => backend.to_string(),
};
let needs_ticker = !matches!(
spec,
TargetSpec::Dump(_)
| TargetSpec::Live {
backend: Backend::Memory,
..
}
);
let opening = OpeningGuard::claim(&self.session)?;
let (tx, service_pending) = tokio::task::spawn_blocking(move || spawn_session(spec))
.await
.map_err(|e| McpError::internal_error(format!("spawn_blocking failed: {e}"), None))?
.map_err(|e| {
McpError::internal_error(format!("failed to open ({label}): {e}"), None)
})?;
let tx_for_ticker = tx.clone();
opening.promote(tx)?;
if needs_ticker {
spawn_service_ticker(tx_for_ticker, service_pending);
}
self.run(move |actor| {
let processors = actor
.ctx
.backend
.thread_list()
.map(|t| t.len())
.unwrap_or(1);
json(serde_json::json!({
"status": "connected",
"backend": label,
"connect": connect,
"processors": processors,
}))
})
.await
}
#[tool(
description = "Close the active debugger session (restores breakpoints and resumes the guest) so a new one can be opened. Returns {status:\"closed\"}, or {status:\"pending\", warning} if the shutdown timed out (retry shortly). Cancels an in-progress open if one is pending."
)]
async fn close(&self) -> Result<CallToolResult, McpError> {
let tx = {
let mut guard = self.session.lock().unwrap();
match &*guard {
SessionSlot::Active(tx) => tx.clone(),
SessionSlot::Opening(_) => {
*guard = SessionSlot::Vacant;
return Ok(CallToolResult::structured(serde_json::json!({
"status": "closed",
"note": "canceled a pending open; the old connection may take a moment to release, so retry if the next open reports AlreadyRunning",
})));
}
SessionSlot::Vacant => {
return Err(McpError::invalid_request(
"no debugger session is active; nothing to close",
None,
));
}
}
};
self.interrupt.store(true, Ordering::Relaxed);
let _reset = InterruptResetGuard(self.interrupt.clone());
let (ack_tx, ack_rx) = oneshot::channel();
let clean = if tx.send(Command::Shutdown { ack: ack_tx }).is_ok() {
tokio::time::timeout(Duration::from_secs(5), ack_rx)
.await
.is_ok_and(|r| r.is_ok())
} else {
true
};
drop(tx);
Ok(CallToolResult::structured(if clean {
// Vacate only after the actor has acked (and released its
// instance lock), so a subsequent open() can acquire the same
// target.
*self.session.lock().unwrap() = SessionSlot::Vacant;
serde_json::json!({ "status": "closed" })
} else {
// The actor is still running (and still holds the instance lock),
// so leave the slot Active; vacating now would let a concurrent
// open() past `claim` only to fail on the lock.
serde_json::json!({
"status": "pending",
"warning": "shutdown timed out; the session is still closing. Retry close shortly",
})
}))
}
}
#[tool_handler(router = self.tool_router)]
impl rmcp::ServerHandler for NtoseyeMcp {
fn get_info(&self) -> ServerInfo {
ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
.with_protocol_version(ProtocolVersion::LATEST)
.with_server_info(Implementation::new(
env!("CARGO_PKG_NAME"),
env!("CARGO_PKG_VERSION"),
))
.with_instructions(
"ntoseye: a WinDbg-like kernel debugger for a Windows VM (KVM/QEMU, \
VMware, UTM) or a crash dump. Drive it with the `command` tool, which \
runs one REPL line in WinDbg-style syntax and returns its text; `help` \
lists commands. The guest runs freely by default: memory, process, \
module, and struct commands work live, while registers, backtraces, \
stepping, and breakpoint changes need the VM halted (call `interrupt` \
first, or be stopped at a breakpoint). Run-control is split so no \
request blocks: `resume` returns immediately, `wait_for_stop` polls \
(bounded) for the next stop, `status` reports where the target is now. \
Typical breakpoint flow: interrupt, command(\"bp nt!NtCreateFile\"), \
resume, wait_for_stop until stop:\"breakpoint\", then command(\"k\"). \
After a reboot, status reports coherent:false until rediscovery \
finishes; wait for it rather than enumerating stale state. Addresses \
in JSON results are 0x hex strings. If no session is open and the \
user has not said how the VM is exposed (kd socket path, kdnet key, \
gdb address, or a dump file), ask them before calling `open` rather \
than guessing; the defaults only fit the documented QEMU setup.",
)
}
}
/// Whether a host string (`localhost`, an IP literal, optionally bracketed)
/// names the loopback interface.
fn is_loopback_host(host: &str) -> bool {
let host = host.trim_start_matches('[').trim_end_matches(']');
host.eq_ignore_ascii_case("localhost")
|| host
.parse::<std::net::IpAddr>()
.is_ok_and(|ip| ip.is_loopback())
}
fn is_loopback_http_bind(addr: &str) -> bool {
if let Ok(socket) = addr.parse::<std::net::SocketAddr>() {
return socket.ip().is_loopback();
}
addr.rsplit_once(':')
.is_some_and(|(host, _port)| is_loopback_host(host))
}
/// Whether a browser `Origin` header (`scheme://host[:port]`, no path) names a
/// loopback host. Gates 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.
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)
};
is_loopback_host(host)
}
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) per `spec`, if given, 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), `Some(addr)` serves **Streamable
/// HTTP** on `addr` for web clients. HTTP binds are loopback-only unless
/// `unsafe_http` is set. Both transports drive the same single session actor.
pub fn run(
spec: Option<TargetSpec>,
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.
let session: SharedSession = Arc::new(std::sync::Mutex::new(SessionSlot::Vacant));
match spec {
Some(spec) => {
let (label, needs_ticker) = match &spec {
TargetSpec::Dump(_) => ("dump".to_string(), false),
TargetSpec::Live { backend, .. } => {
(backend.to_string(), *backend != Backend::Memory)
}
};
eprintln!("ntoseye-mcp: attaching ({label})...");
let (tx, service_pending) = spawn_session(spec)?;
*session.lock().unwrap() = SessionSlot::Active(tx.clone());
if needs_ticker {
spawn_service_ticker(tx, service_pending);
}
}
None => eprintln!("ntoseye-mcp: starting without a session (use open to attach)"),
}
// 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 session_for_shutdown = session.clone();
let runtime = tokio::runtime::Runtime::new()?;
let result = runtime.block_on(async move {
let serve = async {
match http {
Some(addr) => {
eprintln!("ntoseye-mcp: serving Streamable HTTP at http://{addr}/mcp");
serve_http(session, addr, unsafe_http, interrupt).await
}
None => {
eprintln!("ntoseye-mcp: serving over stdio");
let service = NtoseyeMcp::new(session, 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 a live guest frozen with int3s
// installed (a no-op for dumps). Set the interrupt first so any
// in-flight wait returns and the actor is free to process the Shutdown.
eprintln!("ntoseye-mcp: cleaning up...");
interrupt_for_signal.store(true, Ordering::Relaxed);
// Clone the sender out so the slot's mutex isn't held across the await.
let shutdown_tx = match &*session_for_shutdown.lock().unwrap() {
SessionSlot::Active(tx) => Some(tx.clone()),
_ => None,
};
if let Some(tx) = shutdown_tx {
let (ack_tx, ack_rx) = oneshot::channel();
if tx.send(Command::Shutdown { ack: ack_tx }).is_ok() {
let _ = tokio::time::timeout(Duration::from_secs(5), ack_rx).await;
}
}
result
});
runtime.shutdown_background();
result
}
/// Background servicing ticker: periodically nudge the actor to service the
/// guest while idle (see [`Command::Service`]). `service_pending` keeps at
/// most one `Service` queued even if the actor is busy in a long wait; the
/// thread exits once the actor's channel closes (send fails).
fn spawn_service_ticker(tx: mpsc::UnboundedSender<Command>, service_pending: Arc<AtomicBool>) {
std::thread::spawn(move || {
loop {
std::thread::sleep(SERVICE_TICK);
if service_pending.swap(true, Ordering::AcqRel) {
continue;
}
if tx.send(Command::Service).is_err() {
break;
}
}
});
}
/// 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(
session: SharedSession,
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(session, 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::*;
fn empty_mcp() -> NtoseyeMcp {
NtoseyeMcp::new(
Arc::new(std::sync::Mutex::new(SessionSlot::Vacant)),
Arc::new(AtomicBool::new(false)),
)
}
fn fake_active_mcp() -> (NtoseyeMcp, mpsc::UnboundedReceiver<Command>) {
let (tx, rx) = mpsc::unbounded_channel::<Command>();
let mcp = NtoseyeMcp::new(
Arc::new(std::sync::Mutex::new(SessionSlot::Active(tx))),
Arc::new(AtomicBool::new(false)),
);
(mcp, rx)
}
fn open_kd(
mcp: &NtoseyeMcp,
connect: &str,
) -> impl Future<Output = Result<CallToolResult, McpError>> {
mcp.open(Parameters(OpenArgs {
backend: BackendArg::Kd,
connect: Some(connect.into()),
key: None,
}))
}
#[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"));
assert!(!is_loopback_origin("null"));
assert!(!is_loopback_origin("127.0.0.1"));
assert!(!is_loopback_origin(""));
}
#[test]
fn loopback_binds_are_recognized() {
assert!(is_loopback_http_bind("127.0.0.1:8080"));
assert!(is_loopback_http_bind("[::1]:8080"));
assert!(is_loopback_http_bind("localhost:8080"));
assert!(!is_loopback_http_bind("0.0.0.0:8080"));
assert!(!is_loopback_http_bind("192.168.1.2:8080"));
}
#[tokio::test]
async fn open_rejects_when_session_active() {
let (mcp, _rx) = fake_active_mcp();
let err = open_kd(&mcp, "/tmp/fake.sock").await.unwrap_err();
assert!(
err.message.contains("already active"),
"unexpected error: {err:?}"
);
}
#[tokio::test]
async fn open_validates_backend_arguments() {
let mcp = empty_mcp();
let err = mcp
.open(Parameters(OpenArgs {
backend: BackendArg::KdNet,
connect: None,
key: None,
}))
.await
.unwrap_err();
assert!(err.message.contains("requires a key"), "{err:?}");
let err = mcp
.open(Parameters(OpenArgs {
backend: BackendArg::Memory,
connect: Some("127.0.0.1:1234".into()),
key: None,
}))
.await
.unwrap_err();
assert!(err.message.contains("does not use"), "{err:?}");
let err = mcp
.open(Parameters(OpenArgs {
backend: BackendArg::Dump,
connect: None,
key: None,
}))
.await
.unwrap_err();
assert!(err.message.contains("requires 'connect'"), "{err:?}");
}
#[tokio::test]
async fn open_reports_connect_failure_and_frees_the_slot() {
let mcp = empty_mcp();
let err = open_kd(&mcp, "/tmp/ntoseye-test-does-not-exist.sock")
.await
.unwrap_err();
assert!(err.message.contains("failed to open"), "{err:?}");
assert!(matches!(&*mcp.session.lock().unwrap(), SessionSlot::Vacant));
}
#[tokio::test]
async fn close_when_no_session() {
let mcp = empty_mcp();
let err = mcp.close().await.unwrap_err();
assert!(err.message.contains("no debugger session"), "{err:?}");
}
#[tokio::test]
async fn close_active_session_then_reopen_allowed() {
let (mcp, mut rx) = fake_active_mcp();
tokio::spawn(async move {
while let Some(cmd) = rx.recv().await {
if let Command::Shutdown { ack } = cmd {
let _ = ack.send(());
break;
}
}
});
let result = mcp.close().await.expect("close should succeed");
assert_eq!(result.structured_content.unwrap()["status"], "closed");
assert!(matches!(&*mcp.session.lock().unwrap(), SessionSlot::Vacant));
assert!(!mcp.interrupt.load(Ordering::Relaxed));
let err = open_kd(&mcp, "/tmp/ntoseye-test-close-reopen-does-not-exist.sock")
.await
.unwrap_err();
assert!(!err.message.contains("already active"), "{err:?}");
}
#[tokio::test]
async fn close_cancels_opening_session() {
let mcp = NtoseyeMcp::new(
Arc::new(std::sync::Mutex::new(SessionSlot::Opening(0))),
Arc::new(AtomicBool::new(false)),
);
let result = mcp.close().await.expect("close of Opening should succeed");
assert_eq!(result.structured_content.unwrap()["status"], "closed");
assert!(matches!(&*mcp.session.lock().unwrap(), SessionSlot::Vacant));
}
#[tokio::test]
async fn command_requires_session() {
let mcp = empty_mcp();
let err = mcp
.command(Parameters(CommandArgs { line: "lm".into() }))
.await
.unwrap_err();
assert!(err.message.contains("no debugger session"), "{err:?}");
}
}