use std::collections::{HashMap, HashSet};
use std::io::{ErrorKind, Write};
use std::os::unix::net::UnixStream;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, RecvTimeoutError};
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use owo_colors::OwoColorize;
use crate::dbg_backend::{
BackendCapability, BugcheckInfo, DebugBackend, DebugCapability, StopEvent,
};
use crate::error::{Error, Result};
use crate::gdb::RegisterMap;
use crate::kd::framing::{BREAKIN_BYTE, KdFraming};
use crate::types::VirtAddr;
macro_rules! kd_trace {
($($arg:tt)*) => {
if crate::kd::trace_enabled() {
eprintln!($($arg)*);
}
};
}
macro_rules! kd_trace_bytes {
($($arg:tt)*) => {
if crate::kd::trace_bytes_enabled() {
eprint!($($arg)*);
}
};
}
pub fn trace_enabled() -> bool {
static ENABLED: OnceLock<bool> = OnceLock::new();
*ENABLED.get_or_init(|| std::env::var_os("NTOSEYE_KD_TRACE").is_some())
}
pub fn trace_bytes_enabled() -> bool {
static ENABLED: OnceLock<bool> = OnceLock::new();
*ENABLED.get_or_init(|| std::env::var_os("NTOSEYE_KD_TRACE_BYTES").is_some())
}
pub mod api;
pub mod context;
pub mod framing;
mod debug_io;
pub use debug_io::*;
mod event_loop;
pub use event_loop::*;
mod wire;
#[derive(Debug, Clone)]
pub struct StateChange {
processor: u16,
number_processors: u16,
new_state: u32,
exception_code: u32,
program_counter: u64,
kernel_base_hint: Option<VirtAddr>,
is_bugcheck: bool,
bugcheck: Option<BugcheckInfo>,
target_reloaded: bool,
assisted_breakin: bool,
}
const DBG_KD_EXCEPTION_STATE_CHANGE: u32 = 0x0000_3030;
const DBG_KD_LOAD_SYMBOLS_STATE_CHANGE: u32 = 0x0000_3031;
const DBG_KD_COMMAND_STRING_STATE_CHANGE: u32 = 0x0000_3032;
const AMD64_DEBUG_CONTROL_SPACE_KSPECIAL: u64 = 2;
const KSPECIAL_REGISTERS_CR0_OFFSET: usize = 0x00;
const KSPECIAL_REGISTERS_CR2_OFFSET: usize = 0x08;
const KSPECIAL_REGISTERS_CR3_OFFSET: usize = 0x10;
const KSPECIAL_REGISTERS_CR4_OFFSET: usize = 0x18;
const KSPECIAL_REGISTERS_CR8_OFFSET: usize = 0xA0;
const KSPECIAL_REGISTERS_MIN_SIZE: usize = KSPECIAL_REGISTERS_CR8_OFFSET + 8;
const STATUS_BREAKPOINT: u32 = 0x8000_0003;
const KD_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
const DBGKD_DEBUG_IO_HEADER_SIZE: usize = 16;
const DBGKD_DEBUG_IO_MIN_HEADER_SIZE: usize = 12;
const DBGKD_PRINT_STRING_API: u32 = 0x0000_3230;
const DBGKD_GET_STRING_API: u32 = 0x0000_3231;
const DBGKD_FILE_IO_HEADER_SIZE: usize = 64;
const DBGKD_CREATE_FILE_API: u32 = 0x0000_3430;
const DBGKD_READ_FILE_API: u32 = 0x0000_3431;
const DBGKD_WRITE_FILE_API: u32 = 0x0000_3432;
const DBGKD_CLOSE_FILE_API: u32 = 0x0000_3433;
const STATUS_UNSUCCESSFUL: u32 = 0xc000_0001;
const KD_REFRESH_MESSAGE: &[u8] = b"KDTARGET: Refreshing KD connection";
const KD_INITIAL_TIMEOUT_ENV: &str = "NTOSEYE_KD_TIMEOUT";
const KD_INITIAL_TIMEOUT_DEFAULT: Duration = Duration::from_secs(8);
const KD_INITIAL_PROGRESS_INTERVAL: Duration = Duration::from_secs(10);
const KD_REFRESH_BREAKIN_INTERVAL: Duration = Duration::from_millis(250);
const KD_REFRESH_BREAKIN_TRACE_EVERY: u32 = 8;
const BUGCHECK_REFRESH_ASSIST_GRACE: Duration = Duration::from_secs(2);
const POST_BUGCHECK_RECONNECT_ASSIST_DELAY: Duration = Duration::from_secs(20);
const BUGCHECK_MANUALLY_INITIATED_CRASH: u32 = 0x0000_00e2;
const KD_EXIT_STOP_POLL: Duration = Duration::from_secs(1);
const KD_EXIT_MAX_CONTINUES: u32 = 8;
const PUMP_POLL: Duration = Duration::from_millis(100);
fn thread_id_for(processor: u16) -> String {
format!("p1.{:x}", processor + 1)
}
fn parse_thread_id(tid: &str) -> Result<u16> {
let stripped = tid
.strip_prefix("p1.")
.ok_or_else(|| Error::Kd(format!("unrecognised thread id {tid}")))?;
let idx =
u16::from_str_radix(stripped, 16).map_err(|_| Error::Kd(format!("bad thread id {tid}")))?;
if idx == 0 {
return Err(Error::Kd(format!("thread id {tid} has zero index")));
}
Ok(idx - 1)
}
fn parse_thread_id_for_processor_count(tid: &str, processor_count: u16) -> Result<u16> {
let processor = parse_thread_id(tid)?;
if processor >= processor_count {
return Err(Error::Kd(format!(
"thread id {tid} selects processor {}, but guest reports {} processor(s)",
processor + 1,
processor_count
)));
}
Ok(processor)
}
fn should_advance_rip_before_continue(exception_code: u32, managed_breakpoint_stop: bool) -> bool {
exception_code == STATUS_BREAKPOINT && !managed_breakpoint_stop
}
fn append_control_registers_from_special(ctx: &mut Vec<u8>, special: &[u8]) -> Result<()> {
if special.len() < KSPECIAL_REGISTERS_MIN_SIZE {
return Err(Error::Kd(format!(
"KSPECIAL_REGISTERS buffer too short: {} bytes, expected at least {}",
special.len(),
KSPECIAL_REGISTERS_MIN_SIZE
)));
}
ctx.resize(context::REGISTER_BUFFER_SIZE, 0);
let copy_reg = |ctx: &mut [u8], ctx_offset: usize, special_offset: usize| {
ctx[ctx_offset..ctx_offset + 8]
.copy_from_slice(&special[special_offset..special_offset + 8]);
};
copy_reg(ctx, context::OFFSET_CR0, KSPECIAL_REGISTERS_CR0_OFFSET);
copy_reg(ctx, context::OFFSET_CR2, KSPECIAL_REGISTERS_CR2_OFFSET);
copy_reg(ctx, context::OFFSET_CR3, KSPECIAL_REGISTERS_CR3_OFFSET);
copy_reg(ctx, context::OFFSET_CR4, KSPECIAL_REGISTERS_CR4_OFFSET);
copy_reg(ctx, context::OFFSET_CR8, KSPECIAL_REGISTERS_CR8_OFFSET);
Ok(())
}
fn context_payload(data: &[u8]) -> Result<&[u8]> {
if data.len() < context::CONTEXT_SIZE {
return Err(Error::Kd(format!(
"CONTEXT buffer too short: {} bytes, expected {}",
data.len(),
context::CONTEXT_SIZE
)));
}
Ok(&data[..context::CONTEXT_SIZE])
}
fn stop_event(stop: StateChange) -> StopEvent {
StopEvent {
thread_id: Some(thread_id_for(stop.processor)),
exception_code: (stop.new_state == DBG_KD_EXCEPTION_STATE_CHANGE)
.then_some(stop.exception_code),
program_counter: Some(stop.program_counter),
is_bugcheck: stop.is_bugcheck,
bugcheck: stop.bugcheck,
target_reloaded: stop.target_reloaded,
target_kernel_base_hint: stop.kernel_base_hint,
assisted_breakin: stop.assisted_breakin,
}
}
pub struct KdBackend {
framing: Option<KdFraming<UnixStream>>,
breakin_clone: UnixStream,
pump: Option<PumpHandle>,
register_map: RegisterMap,
processor_count: u16,
current_processor: u16,
pending_stop: Option<StateChange>,
last_stop_processor: u16,
last_exception_code: u32,
last_rip: u64,
last_stop_was_managed_breakpoint: bool,
reconnect_assist_after_continue: Option<Duration>,
bp_handles: HashMap<u64, u32>,
managed_bp_addresses: HashSet<u64>,
special_register_cache: HashMap<u16, Vec<u8>>,
is_running: bool,
}
impl KdBackend {
pub fn connect(socket_path: &str) -> Result<Self> {
eprintln!(
"{} {}",
"kd: using KD backend on".bright_black(),
socket_path.cyan()
);
let stream = UnixStream::connect(socket_path)
.map_err(|err| kd_socket_connect_error(socket_path, err))?;
let mut framing = KdFraming::new(stream);
let initial_timeout = kd_initial_timeout()?;
eprintln!(
"{}",
format!(
"kd: serial connected; waiting for Windows KD target (timeout {}s)",
initial_timeout.as_secs()
)
.bright_black()
);
let mut initial_stop = poll_for_initial_break(&mut framing, initial_timeout)?;
if let Err(err) = probe_initial_request(&mut framing, initial_stop.processor) {
if !is_initial_resync_error(&err) {
return Err(err);
}
kd_trace!("kd: initial request probe failed ({err}); resetting KD packet stream");
framing.send_reset()?;
initial_stop = poll_for_initial_break(&mut framing, initial_timeout)?;
probe_initial_request(&mut framing, initial_stop.processor)?;
}
framing.take_peer_reset_seen();
kd_trace!(
"kd: initial state-change received: p{}/{}, exc={:#x}, rip={:#x}",
initial_stop.processor + 1,
initial_stop.number_processors,
initial_stop.exception_code,
initial_stop.program_counter
);
let breakin_clone = framing.transport_mut().try_clone()?;
Ok(Self {
framing: Some(framing),
breakin_clone,
pump: None,
register_map: context::build_register_map(),
processor_count: initial_stop.number_processors.max(1),
current_processor: initial_stop.processor,
last_stop_processor: initial_stop.processor,
last_exception_code: initial_stop.exception_code,
last_rip: initial_stop.program_counter,
pending_stop: None,
bp_handles: HashMap::new(),
managed_bp_addresses: HashSet::new(),
last_stop_was_managed_breakpoint: false,
reconnect_assist_after_continue: None,
special_register_cache: HashMap::new(),
is_running: false,
})
}
fn framing(&mut self) -> Result<&mut KdFraming<UnixStream>> {
self.framing
.as_mut()
.ok_or_else(|| Error::Kd("KD transport is busy: VM is running".into()))
}
fn start_pump(&mut self, reconnect_assist_delay: Option<Duration>) -> Result<()> {
if self.pump.is_some() {
return Ok(());
}
let framing = self
.framing
.take()
.ok_or_else(|| Error::Kd("cannot start KD pump: framing already taken".into()))?;
let (stop_tx, stop_rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let pump_shutdown = Arc::clone(&shutdown);
let join = std::thread::spawn(move || {
run_pump(framing, stop_tx, pump_shutdown, reconnect_assist_delay)
});
kd_trace!("kd: pump: spawned background servicing thread");
self.pump = Some(PumpHandle {
join,
stop_rx,
shutdown,
});
Ok(())
}
fn reclaim_framing(&mut self) {
if let Some(pump) = self.pump.take() {
match pump.join.join() {
Ok(framing) => self.framing = Some(framing),
Err(_) => {
kd_trace!("kd: pump: thread panicked, framing lost");
}
}
}
}
fn take_pump_stop(&mut self, wait: Option<Duration>) -> Result<Option<StateChange>> {
let Some(pump) = self.pump.as_ref() else {
return Ok(None);
};
let received = match wait {
None => pump
.stop_rx
.recv()
.map_err(|_| RecvTimeoutError::Disconnected),
Some(timeout) => pump.stop_rx.recv_timeout(timeout),
};
match received {
Ok(result) => {
self.reclaim_framing();
result.map(Some).map_err(Error::Kd)
}
Err(RecvTimeoutError::Timeout) => Ok(None),
Err(RecvTimeoutError::Disconnected) => {
self.reclaim_framing();
Err(Error::Kd("KD pump exited without reporting a stop".into()))
}
}
}
fn shutdown_pump(&mut self) {
let _ = self.shutdown_pump_with_stop();
}
fn try_recv_pump_stop(
stop_rx: &mpsc::Receiver<std::result::Result<StateChange, String>>,
) -> Result<Option<StateChange>> {
match stop_rx.try_recv() {
Ok(Ok(stop)) => Ok(Some(stop)),
Ok(Err(message)) => Err(Error::Kd(message)),
Err(mpsc::TryRecvError::Empty | mpsc::TryRecvError::Disconnected) => Ok(None),
}
}
fn shutdown_pump_with_stop(&mut self) -> Result<Option<StateChange>> {
let Some(pump) = self.pump.take() else {
return Ok(None);
};
let PumpHandle {
join,
stop_rx,
shutdown,
} = pump;
shutdown.store(true, Ordering::SeqCst);
let stop = Self::try_recv_pump_stop(&stop_rx)?;
match join.join() {
Ok(framing) => self.framing = Some(framing),
Err(_) => {
kd_trace!("kd: pump: thread panicked during shutdown, framing lost");
if stop.is_none() {
return Err(Error::Kd("KD pump thread panicked during shutdown".into()));
}
}
}
if stop.is_some() {
return Ok(stop);
}
Self::try_recv_pump_stop(&stop_rx)
}
fn send_raw_breakin(&mut self) -> Result<()> {
self.breakin_clone.write_all(&[BREAKIN_BYTE])?;
self.breakin_clone.flush()?;
Ok(())
}
fn record_stop(&mut self, stop: &StateChange) {
if stop.target_reloaded {
kd_trace!("kd: target reload detected; clearing target-owned breakpoint state");
self.bp_handles.clear();
self.managed_bp_addresses.clear();
} else if stop.is_bugcheck {
self.reconnect_assist_after_continue = Some(POST_BUGCHECK_RECONNECT_ASSIST_DELAY);
}
let managed_breakpoint_stop = stop.exception_code == STATUS_BREAKPOINT
&& self.managed_bp_addresses.contains(&stop.program_counter);
kd_trace!(
"kd: stop on p{}, new_state={:#x}, exception_code={:#x}, rip={:#x}, managed_bp={}",
stop.processor + 1,
stop.new_state,
stop.exception_code,
stop.program_counter,
managed_breakpoint_stop
);
self.current_processor = stop.processor;
self.processor_count = self.processor_count.max(stop.number_processors.max(1));
self.last_stop_processor = stop.processor;
self.last_exception_code = stop.exception_code;
self.last_rip = stop.program_counter;
self.last_stop_was_managed_breakpoint = managed_breakpoint_stop;
self.special_register_cache.clear();
self.is_running = false;
}
fn record_running(&mut self) {
self.is_running = true;
self.special_register_cache.clear();
}
fn advance_rip_past_int3(&mut self, processor: u16) -> Result<()> {
let mut ctx = with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::get_context(framing, processor)
})?;
let rip = self.register_map.read_u64("rip", &ctx)?;
kd_trace!(
"kd: advance_rip: p{} read rip={:#x}, ctx.len={}",
processor + 1,
rip,
ctx.len()
);
self.register_map
.write_u64("rip", &mut ctx, rip.wrapping_add(1))?;
with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::set_context(framing, processor, &ctx)
})?;
if trace_enabled() {
if let Ok(verify_ctx) =
with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::get_context(framing, processor)
})
&& let Ok(verify_rip) = self.register_map.read_u64("rip", &verify_ctx)
{
kd_trace!(
"kd: advance_rip: p{} wrote {:#x}, read back {:#x}",
processor + 1,
rip.wrapping_add(1),
verify_rip
);
}
}
Ok(())
}
fn read_special_registers(&mut self) -> Result<&[u8]> {
if !self
.special_register_cache
.contains_key(&self.current_processor)
{
let processor = self.current_processor;
let data = with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::read_control_space(
framing,
processor,
AMD64_DEBUG_CONTROL_SPACE_KSPECIAL,
KSPECIAL_REGISTERS_MIN_SIZE as u32,
)
})?;
self.special_register_cache.insert(processor, data);
}
self.special_register_cache
.get(&self.current_processor)
.map(Vec::as_slice)
.ok_or_else(|| Error::Kd("special-register cache lookup failed".into()))
}
fn append_control_registers(&mut self, ctx: &mut Vec<u8>) -> Result<()> {
let special = self.read_special_registers()?;
append_control_registers_from_special(ctx, special)
}
fn continue_stopped_for_exit(&mut self) -> Result<()> {
let processor = self.last_stop_processor;
if should_advance_rip_before_continue(
self.last_exception_code,
self.last_stop_was_managed_breakpoint,
) {
self.advance_rip_past_int3(processor)?;
}
with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::continue_api2(framing, processor, api::DBG_CONTINUE, false)
})?;
self.record_running();
Ok(())
}
fn finish_for_exit(&mut self, leave_running: bool) -> Result<()> {
if let Some(stop) = self.shutdown_pump_with_stop()? {
self.record_stop(&stop);
}
if !leave_running {
return Ok(());
}
for _ in 0..KD_EXIT_MAX_CONTINUES {
if self.is_running && self.try_wait_for_stop(KD_EXIT_STOP_POLL)?.is_none() {
return Ok(());
}
self.continue_stopped_for_exit()?;
if self.try_wait_for_stop(KD_EXIT_STOP_POLL)?.is_none() {
return Ok(());
}
}
Err(Error::Kd(format!(
"target kept stopping during debugger exit after {KD_EXIT_MAX_CONTINUES} continues"
)))
}
}
impl DebugBackend for KdBackend {
fn register_map(&self) -> &RegisterMap {
&self.register_map
}
fn read_registers(&mut self) -> Result<Vec<u8>> {
kd_trace!(
"kd: read_registers: GetContext on p{}",
self.current_processor + 1
);
let processor = self.current_processor;
let mut ctx = with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::get_context(framing, processor)
})?;
kd_trace!("kd: read_registers: got {} context bytes", ctx.len());
self.append_control_registers(&mut ctx)?;
kd_trace!("kd: read_registers: extended to {} bytes", ctx.len());
Ok(ctx)
}
fn write_registers(&mut self, data: &[u8]) -> Result<()> {
let context = context_payload(data)?;
let processor = self.current_processor;
with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::set_context(framing, processor, context)
})
}
fn set_breakpoint(&mut self, addr: u64) -> Result<()> {
let processor = self.current_processor;
let handle = with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::write_breakpoint(framing, processor, addr)
})?;
self.bp_handles.insert(addr, handle);
self.managed_bp_addresses.insert(addr);
Ok(())
}
fn remove_breakpoint(&mut self, addr: u64) -> Result<()> {
let handle = self
.bp_handles
.remove(&addr)
.ok_or_else(|| Error::Kd(format!("no breakpoint tracked at {addr:#x}")))?;
self.managed_bp_addresses.remove(&addr);
let processor = self.current_processor;
let result = with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::restore_breakpoint(framing, processor, handle)
});
if let Err(Error::KdStatus { ntstatus, api }) = &result
&& *ntstatus == STATUS_UNSUCCESSFUL
&& *api == api::DBGKD_RESTORE_BREAKPOINT
{
kd_trace!(
"kd: restore breakpoint handle {} at {:#x} was already consumed",
handle,
addr
);
return Ok(());
}
result
}
fn supports_user_mode_breakpoints(&self) -> bool {
true
}
fn optional_capabilities(&self) -> Vec<BackendCapability> {
vec![
BackendCapability::supported(DebugCapability::UserModeBreakpoints),
BackendCapability::supported(DebugCapability::TargetReloadDetection),
BackendCapability::supported(DebugCapability::KernelBaseHint),
BackendCapability::supported(DebugCapability::BugcheckDetection),
BackendCapability::supported(DebugCapability::BugcheckDetails),
BackendCapability::supported(DebugCapability::DebugOutput),
]
}
fn note_breakpoint_installed(&mut self, addr: u64) {
self.managed_bp_addresses.insert(addr);
}
fn note_breakpoint_uninstalled(&mut self, addr: u64) {
self.managed_bp_addresses.remove(&addr);
}
fn note_target_rediscovery_pending(&mut self) {
self.reconnect_assist_after_continue = Some(Duration::ZERO);
}
fn note_target_rediscovery_complete(&mut self) {
self.reconnect_assist_after_continue = None;
}
fn target_kernel_base_hint(&mut self) -> Result<Option<VirtAddr>> {
let processor = self.current_processor;
with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::get_version(framing, processor).map(|version| Some(VirtAddr(version.kern_base)))
})
}
fn continue_execution(&mut self) -> Result<()> {
const MAX_DRAIN_ITERATIONS: u32 = 64;
const DRAIN_POLL: Duration = Duration::from_millis(1000);
let mut drained = 0u32;
loop {
let resume_processor = self.last_stop_processor;
if should_advance_rip_before_continue(
self.last_exception_code,
self.last_stop_was_managed_breakpoint,
) {
kd_trace!(
"kd: continue: advancing p{} RIP past raw int3 (last_exception_code={:#x})",
resume_processor + 1,
self.last_exception_code,
);
self.advance_rip_past_int3(resume_processor)?;
} else {
kd_trace!(
"kd: continue: not advancing p{} (last_exception_code={:#x}, managed_bp={})",
resume_processor + 1,
self.last_exception_code,
self.last_stop_was_managed_breakpoint
);
}
let resumed_from_rip = self.last_rip;
let reconnect_assist_after_continue = self.reconnect_assist_after_continue;
kd_trace!(
"kd: continue: sending ContinueApi2 on p{}",
resume_processor + 1
);
with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::continue_api2(framing, resume_processor, api::DBG_CONTINUE, false)
})?;
kd_trace!("kd: continue: ContinueApi2 ACKed, VM should resume");
self.record_running();
self.framing()?
.transport_mut()
.set_read_timeout(Some(DRAIN_POLL))?;
let mut saw_kd_refresh = false;
let result = await_state_change(
self.framing()?,
Some(&mut saw_kd_refresh),
false,
None,
None,
);
let _ = self.framing()?.transport_mut().set_read_timeout(None);
match result {
Ok(stop) => {
let is_spurious = stop.exception_code == STATUS_BREAKPOINT
&& stop.program_counter == resumed_from_rip
&& !self.managed_bp_addresses.contains(&stop.program_counter);
if is_spurious && drained < MAX_DRAIN_ITERATIONS {
drained += 1;
kd_trace!(
"kd: continue: spurious re-break at {:#x} (drain {}/{})",
stop.program_counter,
drained,
MAX_DRAIN_ITERATIONS
);
self.record_stop(&stop);
continue;
}
kd_trace!(
"kd: continue: real stop at {:#x} (exc={:#x}), stashing as pending",
stop.program_counter,
stop.exception_code
);
self.record_stop(&stop);
self.pending_stop = Some(stop);
return Ok(());
}
Err(Error::Io(e))
if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut =>
{
if saw_kd_refresh {
kd_trace!("kd: continue: KD refresh observed; leaving pump to service it");
}
if drained > 0 {
kd_trace!(
"kd: continue: drained {} spurious break(s), VM now running",
drained
);
}
self.start_pump(reconnect_assist_after_continue)?;
return Ok(());
}
Err(e) => return Err(e),
}
}
}
fn step(&mut self) -> Result<()> {
let processor = self.current_processor;
with_framing_read_timeout(self.framing()?, KD_REQUEST_TIMEOUT, |framing| {
api::continue_api2(framing, processor, api::DBG_CONTINUE, true)
})?;
self.record_running();
Ok(())
}
fn interrupt(&mut self) -> Result<StopEvent> {
let stop = if self.pump.is_some() {
self.send_raw_breakin()?;
match self.take_pump_stop(Some(Duration::from_secs(10)))? {
Some(stop) => stop,
None => {
self.shutdown_pump();
return Err(Error::Kd("no break-in response within 10s".into()));
}
}
} else {
breakin_and_wait(self.framing()?, Duration::from_secs(10))?
};
self.record_stop(&stop);
Ok(stop_event(stop))
}
fn wait_for_stop(&mut self) -> Result<StopEvent> {
if let Some(stop) = self.pending_stop.take() {
self.record_stop(&stop);
return Ok(stop_event(stop));
}
if self.pump.is_some() {
let stop = self
.take_pump_stop(None)?
.ok_or_else(|| Error::Kd("KD pump returned no stop".into()))?;
self.record_stop(&stop);
return Ok(stop_event(stop));
}
let stop = await_state_change(self.framing()?, None, false, None, None)?;
self.record_stop(&stop);
Ok(stop_event(stop))
}
fn try_wait_for_stop(&mut self, timeout: Duration) -> Result<Option<StopEvent>> {
if let Some(stop) = self.pending_stop.take() {
kd_trace!(
"kd: try_wait: surfacing pending_stop rip={:#x} (bypassing spurious check)",
stop.program_counter
);
self.record_stop(&stop);
return Ok(Some(stop_event(stop)));
}
if self.pump.is_some() {
return match self.take_pump_stop(Some(timeout))? {
Some(stop) => {
kd_trace!(
"kd: try_wait: pump reported stop rip={:#x} exc={:#x}",
stop.program_counter,
stop.exception_code
);
self.record_stop(&stop);
Ok(Some(stop_event(stop)))
}
None => Ok(None),
};
}
self.framing()?
.transport_mut()
.set_read_timeout(Some(timeout))?;
let mut saw_kd_refresh = false;
let result = await_state_change(
self.framing()?,
Some(&mut saw_kd_refresh),
false,
None,
None,
);
let _ = self.framing()?.transport_mut().set_read_timeout(None);
let stop = match result {
Ok(stop) => stop,
Err(Error::Io(e))
if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut =>
{
if saw_kd_refresh {
kd_trace!("kd: try_wait: KD refresh observed while polling");
}
return Ok(None);
}
Err(e) => return Err(e),
};
kd_trace!(
"kd: try_wait: stop rip={:#x} exc={:#x} in_managed={}",
stop.program_counter,
stop.exception_code,
self.managed_bp_addresses.contains(&stop.program_counter)
);
self.record_stop(&stop);
Ok(Some(stop_event(stop)))
}
fn thread_list(&mut self) -> Result<Vec<String>> {
Ok((0..self.processor_count).map(thread_id_for).collect())
}
fn set_current_thread(&mut self, thread_id: &str) -> Result<()> {
self.current_processor =
parse_thread_id_for_processor_count(thread_id, self.processor_count)?;
Ok(())
}
fn stopped_thread_id(&mut self) -> Result<String> {
Ok(thread_id_for(self.current_processor))
}
fn is_running(&self) -> bool {
self.is_running
}
fn prepare_for_exit(&mut self, leave_running: bool) -> Result<()> {
self.finish_for_exit(leave_running)
}
fn take_modules_changed(&mut self) -> bool {
self.framing
.as_mut()
.map(KdFraming::take_modules_changed)
.unwrap_or(false)
}
}
impl Drop for KdBackend {
fn drop(&mut self) {
if self.pump.is_some() || !self.is_running {
let _ = self.finish_for_exit(true);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::kd::framing::{
PACKET_TYPE_KD_ACKNOWLEDGE, PACKET_TYPE_KD_DEBUG_IO, PACKET_TYPE_KD_FILE_IO,
PACKET_TYPE_KD_RESET, PACKET_TYPE_KD_STATE_CHANGE64, PACKET_TYPE_KD_STATE_MANIPULATE,
};
use std::io::{Cursor, Read, Write};
use std::time::Instant;
struct Loopback {
inbound: Cursor<Vec<u8>>,
outbound: Vec<u8>,
}
impl Loopback {
fn new() -> Self {
Self {
inbound: Cursor::new(Vec::new()),
outbound: Vec::new(),
}
}
fn with_inbound(inbound: Vec<u8>) -> Self {
Self {
inbound: Cursor::new(inbound),
outbound: Vec::new(),
}
}
}
impl Read for Loopback {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
Read::read(&mut self.inbound, buf)
}
}
impl Write for Loopback {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.outbound.extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
fn debug_io_print_payload(text: &[u8]) -> Vec<u8> {
let mut payload = vec![0u8; DBGKD_DEBUG_IO_HEADER_SIZE];
payload[0..4].copy_from_slice(&DBGKD_PRINT_STRING_API.to_le_bytes());
payload[8..12].copy_from_slice(&(text.len() as u32).to_le_bytes());
payload.extend_from_slice(text);
payload
}
#[test]
fn parse_state_change_extracts_processor_and_pc() {
let mut payload = vec![0u8; 64];
payload[0..4].copy_from_slice(&DBG_KD_EXCEPTION_STATE_CHANGE.to_le_bytes()); payload[6..8].copy_from_slice(&2u16.to_le_bytes()); payload[8..12].copy_from_slice(&4u32.to_le_bytes()); payload[24..32].copy_from_slice(&0xfffff800deadbeefu64.to_le_bytes());
payload[32..36].copy_from_slice(&STATUS_BREAKPOINT.to_le_bytes());
let s = parse_state_change(&payload).unwrap();
assert_eq!(s.processor, 2);
assert_eq!(s.number_processors, 4);
assert_eq!(s.new_state, DBG_KD_EXCEPTION_STATE_CHANGE);
assert_eq!(s.exception_code, STATUS_BREAKPOINT);
assert_eq!(s.program_counter, 0xfffff800deadbeef);
}
#[test]
fn parse_load_symbols_state_change_extracts_base_hint() {
let mut payload = vec![0u8; 64];
payload[0..4].copy_from_slice(&DBG_KD_LOAD_SYMBOLS_STATE_CHANGE.to_le_bytes());
payload[8..12].copy_from_slice(&1u32.to_le_bytes());
payload[24..32].copy_from_slice(&0xfffff800004f9325u64.to_le_bytes());
payload[40..48].copy_from_slice(&0xfffff80000000000u64.to_le_bytes());
let s = parse_state_change(&payload).unwrap();
assert_eq!(s.program_counter, 0xfffff800004f9325);
assert_eq!(s.kernel_base_hint, Some(VirtAddr(0xfffff80000000000)));
}
#[test]
fn stop_event_preserves_kd_exception_details() {
let stop = StateChange {
processor: 1,
number_processors: 2,
new_state: DBG_KD_EXCEPTION_STATE_CHANGE,
exception_code: STATUS_BREAKPOINT,
program_counter: 0xfffff800deadbeef,
kernel_base_hint: None,
is_bugcheck: false,
bugcheck: None,
target_reloaded: false,
assisted_breakin: false,
};
let event = stop_event(stop);
assert_eq!(event.thread_id.as_deref(), Some("p1.2"));
assert_eq!(event.exception_code, Some(STATUS_BREAKPOINT));
assert_eq!(event.program_counter, Some(0xfffff800deadbeef));
assert_eq!(event.target_kernel_base_hint, None);
assert!(!event.is_bugcheck);
assert!(event.bugcheck.is_none());
assert!(!event.target_reloaded);
assert!(!event.assisted_breakin);
}
#[test]
fn stop_event_preserves_target_reload_flag() {
let stop = StateChange {
processor: 0,
number_processors: 1,
new_state: DBG_KD_EXCEPTION_STATE_CHANGE,
exception_code: STATUS_BREAKPOINT,
program_counter: 0xfffff800deadbeef,
kernel_base_hint: None,
is_bugcheck: false,
bugcheck: None,
target_reloaded: true,
assisted_breakin: false,
};
let event = stop_event(stop);
assert!(event.target_reloaded);
assert!(!event.is_bugcheck);
}
#[test]
fn stop_event_preserves_assisted_breakin_flag() {
let stop = StateChange {
processor: 0,
number_processors: 1,
new_state: DBG_KD_EXCEPTION_STATE_CHANGE,
exception_code: STATUS_BREAKPOINT,
program_counter: 0xfffff800deadbeef,
kernel_base_hint: None,
is_bugcheck: false,
bugcheck: None,
target_reloaded: false,
assisted_breakin: true,
};
let event = stop_event(stop);
assert!(event.assisted_breakin);
}
#[test]
fn stop_event_flags_surfaced_load_symbols_as_bugcheck() {
let stop = StateChange {
processor: 0,
number_processors: 1,
new_state: DBG_KD_LOAD_SYMBOLS_STATE_CHANGE,
exception_code: 0,
program_counter: 0xfffff8007faf9325,
kernel_base_hint: Some(VirtAddr(0xfffff8007f600000)),
is_bugcheck: true,
bugcheck: None,
target_reloaded: false,
assisted_breakin: false,
};
let event = stop_event(stop);
assert!(event.is_bugcheck);
assert_eq!(event.exception_code, None);
assert_eq!(event.program_counter, Some(0xfffff8007faf9325));
assert_eq!(
event.target_kernel_base_hint,
Some(VirtAddr(0xfffff8007f600000))
);
assert!(event.bugcheck.is_none());
}
#[test]
fn stop_event_does_not_flag_reloaded_load_symbols_as_bugcheck() {
let stop = StateChange {
processor: 0,
number_processors: 1,
new_state: DBG_KD_LOAD_SYMBOLS_STATE_CHANGE,
exception_code: 0,
program_counter: 0xfffff8007faf9325,
kernel_base_hint: None,
is_bugcheck: false,
bugcheck: None,
target_reloaded: true,
assisted_breakin: false,
};
let event = stop_event(stop);
assert!(event.target_reloaded);
assert!(!event.is_bugcheck);
assert_eq!(event.exception_code, None);
}
#[test]
fn bugcheck_capture_extracts_fatal_error_and_driver() {
let mut capture = BugcheckCapture::default();
capture.observe_debug_text(
b"\r\n*** Fatal System Error: 0x000000d1\r\n (0xFFFFB90641184010,0x0000000000000002,0x0000000000000000,0xFFFFF8016E151730)\r\n",
);
capture.observe_debug_text(b"Driver at fault: myfault.sys.\r\n");
let info = capture.finish().unwrap();
assert_eq!(info.code, 0xd1);
assert_eq!(
info.parameters,
[
0xffff_b906_4118_4010,
0x0000_0000_0000_0002,
0x0000_0000_0000_0000,
0xffff_f801_6e15_1730,
]
);
assert_eq!(info.driver.as_deref(), Some("myfault.sys"));
}
#[test]
fn captured_bugcheck_debug_io_can_be_suppressed() {
let payload = debug_io_print_payload(
b"\r\n*** Fatal System Error: 0x000000d1\r\n (0x1,0x2,0x0,0x4)\r\n",
);
let mut framing = KdFraming::new(Loopback::new());
let mut capture = BugcheckCapture::default();
let mut output = Vec::new();
let saw_refresh = handle_debug_io_with_output(
&mut framing,
&payload,
true,
Some(&mut capture),
true,
&mut output,
)
.unwrap();
assert!(!saw_refresh);
assert!(output.is_empty());
assert_eq!(capture.finish().unwrap().code, 0xd1);
}
#[test]
fn parse_debug_io_print_extracts_string() {
let payload = debug_io_print_payload(b"hello");
match parse_debug_io(&payload).unwrap() {
DebugIo::PrintString { text } => assert_eq!(text, b"hello"),
DebugIo::GetString { .. } => panic!("expected print-string debug I/O"),
}
}
#[test]
fn debug_io_refresh_message_is_reported_when_waiting_for_stop() {
let payload = debug_io_print_payload(b"KDTARGET: Refreshing KD connection\n");
let mut framing = KdFraming::new(Loopback::new());
let mut output = Vec::new();
let saw_refresh =
handle_debug_io_with_output(&mut framing, &payload, true, None, false, &mut output)
.unwrap();
assert!(saw_refresh);
assert_eq!(output, b"KDTARGET: Refreshing KD connection\n");
assert!(framing.transport_ref().outbound.is_empty());
}
#[test]
fn debug_io_refresh_message_is_passive_during_manipulate_requests() {
let payload = debug_io_print_payload(b"KDTARGET: Refreshing KD connection\n");
let mut framing = KdFraming::new(Loopback::new());
let mut output = Vec::new();
let saw_refresh =
handle_debug_io_with_output(&mut framing, &payload, false, None, false, &mut output)
.unwrap();
assert!(!saw_refresh);
assert_eq!(output, b"KDTARGET: Refreshing KD connection\n");
assert!(framing.transport_ref().outbound.is_empty());
}
#[test]
fn parse_debug_io_print_accepts_legacy_short_header() {
let mut payload = vec![0u8; DBGKD_DEBUG_IO_MIN_HEADER_SIZE];
payload[0..4].copy_from_slice(&DBGKD_PRINT_STRING_API.to_le_bytes());
payload[8..12].copy_from_slice(&5u32.to_le_bytes());
payload.extend_from_slice(b"hello");
match parse_debug_io(&payload).unwrap() {
DebugIo::PrintString { text } => assert_eq!(text, b"hello"),
DebugIo::GetString { .. } => panic!("expected print-string debug I/O"),
}
}
#[test]
fn parse_debug_io_get_string_extracts_prompt() {
let mut payload = vec![0u8; DBGKD_DEBUG_IO_HEADER_SIZE];
payload[0..4].copy_from_slice(&DBGKD_GET_STRING_API.to_le_bytes());
payload[4..6].copy_from_slice(&0x33u16.to_le_bytes());
payload[6..8].copy_from_slice(&2u16.to_le_bytes());
payload[8..12].copy_from_slice(&7u32.to_le_bytes());
payload[12..16].copy_from_slice(&0x100u32.to_le_bytes());
payload.extend_from_slice(b"prompt>");
match parse_debug_io(&payload).unwrap() {
DebugIo::GetString {
processor_level,
processor,
prompt,
} => {
assert_eq!(processor_level, 0x33);
assert_eq!(processor, 2);
assert_eq!(prompt, b"prompt>");
}
DebugIo::PrintString { .. } => panic!("expected get-string debug I/O"),
}
}
#[test]
fn parse_debug_io_print_rejects_other_api() {
let mut payload = vec![0u8; DBGKD_DEBUG_IO_MIN_HEADER_SIZE];
payload[0..4].copy_from_slice(&0xdeadbeefu32.to_le_bytes());
assert!(parse_debug_io(&payload).is_none());
}
#[test]
fn parse_state_change_rejects_short_payload() {
let err = parse_state_change(&[0u8; 10]).unwrap_err();
match err {
Error::Kd(msg) => assert!(msg.contains("too short")),
other => panic!("unexpected error: {other:?}"),
}
}
#[test]
fn continue_advance_policy_skips_only_raw_int3() {
assert!(should_advance_rip_before_continue(STATUS_BREAKPOINT, false));
assert!(!should_advance_rip_before_continue(STATUS_BREAKPOINT, true));
assert!(!should_advance_rip_before_continue(0x8000_0004, false)); }
#[test]
fn initial_handshake_breaks_in_immediately_then_resets() {
assert_eq!(
initial_handshake_stimulus(0),
InitialHandshakeStimulus::BreakIn
);
assert_eq!(
initial_handshake_stimulus(1),
InitialHandshakeStimulus::Reset
);
assert_eq!(
initial_handshake_stimulus(2),
InitialHandshakeStimulus::BreakIn
);
assert_eq!(
initial_handshake_stimulus(3),
InitialHandshakeStimulus::Reset
);
}
#[test]
fn kd_initial_timeout_defaults_to_eight_seconds() {
assert_eq!(
parse_kd_initial_timeout(None).unwrap(),
Duration::from_secs(8)
);
}
#[test]
fn kd_initial_timeout_accepts_positive_seconds() {
assert_eq!(
parse_kd_initial_timeout(Some("12")).unwrap(),
Duration::from_secs(12)
);
}
#[test]
fn kd_initial_timeout_rejects_invalid_values() {
assert!(parse_kd_initial_timeout(Some("0")).is_err());
assert!(parse_kd_initial_timeout(Some("meow")).is_err());
}
#[test]
fn context_payload_accepts_synthetic_register_buffer() {
let synthetic = vec![0u8; context::REGISTER_BUFFER_SIZE];
assert_eq!(
context_payload(&synthetic).unwrap().len(),
context::CONTEXT_SIZE
);
}
#[test]
fn context_payload_rejects_short_buffers() {
let short = vec![0u8; context::CONTEXT_SIZE - 1];
assert!(context_payload(&short).is_err());
}
#[test]
fn append_control_registers_extends_context() {
let mut ctx = vec![0u8; context::CONTEXT_SIZE];
let mut special = vec![0u8; KSPECIAL_REGISTERS_MIN_SIZE];
special[KSPECIAL_REGISTERS_CR0_OFFSET..KSPECIAL_REGISTERS_CR0_OFFSET + 8]
.copy_from_slice(&0x8005_0033u64.to_le_bytes());
special[KSPECIAL_REGISTERS_CR2_OFFSET..KSPECIAL_REGISTERS_CR2_OFFSET + 8]
.copy_from_slice(&0x1111_2222u64.to_le_bytes());
special[KSPECIAL_REGISTERS_CR3_OFFSET..KSPECIAL_REGISTERS_CR3_OFFSET + 8]
.copy_from_slice(&0x1234_5000u64.to_le_bytes());
special[KSPECIAL_REGISTERS_CR4_OFFSET..KSPECIAL_REGISTERS_CR4_OFFSET + 8]
.copy_from_slice(&0x350ef8u64.to_le_bytes());
special[KSPECIAL_REGISTERS_CR8_OFFSET..KSPECIAL_REGISTERS_CR8_OFFSET + 8]
.copy_from_slice(&2u64.to_le_bytes());
append_control_registers_from_special(&mut ctx, &special).unwrap();
let map = context::build_register_map();
assert_eq!(ctx.len(), context::REGISTER_BUFFER_SIZE);
assert_eq!(map.read_u64("cr0", &ctx).unwrap(), 0x8005_0033);
assert_eq!(map.read_u64("cr2", &ctx).unwrap(), 0x1111_2222);
assert_eq!(map.read_u64("cr3", &ctx).unwrap(), 0x1234_5000);
assert_eq!(map.read_u64("cr4", &ctx).unwrap(), 0x350ef8);
assert_eq!(map.read_u64("cr8", &ctx).unwrap(), 2);
}
#[test]
fn thread_id_uses_one_based_hex() {
assert_eq!(thread_id_for(0), "p1.1");
assert_eq!(thread_id_for(3), "p1.4");
assert_eq!(thread_id_for(15), "p1.10");
}
#[test]
fn thread_id_round_trips() {
for proc in [0u16, 1, 7, 15, 31] {
let tid = thread_id_for(proc);
assert_eq!(parse_thread_id(&tid).unwrap(), proc);
}
}
#[test]
fn parse_thread_id_rejects_garbage() {
assert!(parse_thread_id("p2.1").is_err()); assert!(parse_thread_id("p1.zz").is_err()); assert!(parse_thread_id("garbage").is_err());
assert!(parse_thread_id("p1.0").is_err()); }
#[test]
fn parse_thread_id_for_processor_count_rejects_out_of_range() {
assert_eq!(parse_thread_id_for_processor_count("p1.4", 4).unwrap(), 3);
assert!(parse_thread_id_for_processor_count("p1.5", 4).is_err());
}
const WIRE_DATA_LEADER: u32 = 0x3030_3030;
const WIRE_CONTROL_LEADER: u32 = 0x6969_6969;
const WIRE_HEADER_SIZE: usize = 16;
const WIRE_TRAILER: u8 = 0xAA;
const WIRE_FIRST_PACKET_ID: u32 = 0x8080_0000;
fn wire_control_packet(packet_type: u16, packet_id: u32) -> Vec<u8> {
let mut pkt = Vec::new();
pkt.extend_from_slice(&WIRE_CONTROL_LEADER.to_le_bytes());
pkt.extend_from_slice(&packet_type.to_le_bytes());
pkt.extend_from_slice(&0u16.to_le_bytes());
pkt.extend_from_slice(&packet_id.to_le_bytes());
pkt.extend_from_slice(&0u32.to_le_bytes());
pkt
}
fn wire_data_packet(packet_type: u16, packet_id: u32, payload: &[u8]) -> Vec<u8> {
let checksum = payload.iter().fold(0u32, |a, &b| a.wrapping_add(b as u32));
let mut pkt = Vec::new();
pkt.extend_from_slice(&WIRE_DATA_LEADER.to_le_bytes());
pkt.extend_from_slice(&packet_type.to_le_bytes());
pkt.extend_from_slice(&(payload.len() as u16).to_le_bytes());
pkt.extend_from_slice(&packet_id.to_le_bytes());
pkt.extend_from_slice(&checksum.to_le_bytes());
pkt.extend_from_slice(payload);
pkt.push(WIRE_TRAILER);
pkt
}
fn read_wire_packet(stream: &mut UnixStream) -> Vec<u8> {
let mut header = [0u8; WIRE_HEADER_SIZE];
stream.read_exact(&mut header).unwrap();
let mut pkt = header.to_vec();
let leader = u32::from_le_bytes(header[0..4].try_into().unwrap());
if leader == WIRE_DATA_LEADER {
let len = u16::from_le_bytes(header[6..8].try_into().unwrap()) as usize;
let mut rest = vec![0u8; len + 1];
stream.read_exact(&mut rest).unwrap();
pkt.extend_from_slice(&rest);
}
pkt
}
fn state_change_payload(new_state: u32, pc: u64) -> Vec<u8> {
let mut payload = vec![0u8; 56];
payload[0..4].copy_from_slice(&new_state.to_le_bytes());
payload[8..12].copy_from_slice(&1u32.to_le_bytes()); payload[24..32].copy_from_slice(&pc.to_le_bytes());
payload[32..36].copy_from_slice(&STATUS_BREAKPOINT.to_le_bytes());
payload
}
fn exception_state_change_payload(pc: u64) -> Vec<u8> {
state_change_payload(DBG_KD_EXCEPTION_STATE_CHANGE, pc)
}
#[test]
fn file_io_create_file_gets_explicit_failure_reply() {
let mut payload = vec![0u8; DBGKD_FILE_IO_HEADER_SIZE];
payload[0..4].copy_from_slice(&DBGKD_CREATE_FILE_API.to_le_bytes());
let ack = wire_control_packet(PACKET_TYPE_KD_ACKNOWLEDGE, WIRE_FIRST_PACKET_ID);
let mut framing = KdFraming::new(Loopback::with_inbound(ack));
handle_file_io(&mut framing, &payload).unwrap();
let out = &framing.transport_ref().outbound;
assert_eq!(out.len(), WIRE_HEADER_SIZE + DBGKD_FILE_IO_HEADER_SIZE + 1);
assert_eq!(
u32::from_le_bytes(out[0..4].try_into().unwrap()),
WIRE_DATA_LEADER
);
assert_eq!(
u16::from_le_bytes(out[4..6].try_into().unwrap()),
PACKET_TYPE_KD_FILE_IO
);
assert_eq!(
u16::from_le_bytes(out[6..8].try_into().unwrap()) as usize,
DBGKD_FILE_IO_HEADER_SIZE
);
assert_eq!(
u32::from_le_bytes(out[8..12].try_into().unwrap()),
WIRE_FIRST_PACKET_ID
);
let reply = &out[WIRE_HEADER_SIZE..WIRE_HEADER_SIZE + DBGKD_FILE_IO_HEADER_SIZE];
assert_eq!(
u32::from_le_bytes(reply[0..4].try_into().unwrap()),
DBGKD_CREATE_FILE_API
);
assert_eq!(
u32::from_le_bytes(reply[4..8].try_into().unwrap()),
STATUS_UNSUCCESSFUL
);
assert_eq!(
out[WIRE_HEADER_SIZE + DBGKD_FILE_IO_HEADER_SIZE],
WIRE_TRAILER
);
}
fn kd_backend_with_pump(pump: PumpHandle, breakin_clone: UnixStream) -> KdBackend {
KdBackend {
framing: None,
breakin_clone,
pump: Some(pump),
register_map: context::build_register_map(),
processor_count: 1,
current_processor: 0,
pending_stop: None,
last_stop_processor: 0,
last_exception_code: 0,
last_rip: 0,
last_stop_was_managed_breakpoint: false,
reconnect_assist_after_continue: None,
bp_handles: HashMap::new(),
managed_bp_addresses: HashSet::new(),
special_register_cache: HashMap::new(),
is_running: true,
}
}
#[test]
fn pump_services_state_change_and_returns_framing() {
let (mut kernel, host) = UnixStream::pair().unwrap();
let framing = KdFraming::new(host);
let (tx, rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let handle = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || run_pump(framing, tx, shutdown, None))
};
let pc = 0xfffff800_deadbeef;
let pkt = wire_data_packet(
PACKET_TYPE_KD_STATE_CHANGE64,
WIRE_FIRST_PACKET_ID,
&exception_state_change_payload(pc),
);
kernel.write_all(&pkt).unwrap();
kernel.flush().unwrap();
let stop = rx
.recv_timeout(Duration::from_secs(5))
.expect("pump reported no stop")
.expect("pump reported an error");
assert_eq!(stop.program_counter, pc);
assert_eq!(stop.exception_code, STATUS_BREAKPOINT);
shutdown.store(true, Ordering::SeqCst);
let _framing = handle.join().expect("pump thread panicked");
}
#[test]
fn exit_resume_consumes_pump_stop_before_final_continue() {
let (mut kernel, host) = UnixStream::pair().unwrap();
let breakin_clone = host.try_clone().unwrap();
let framing = KdFraming::new(host);
let (tx, rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let join = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || run_pump(framing, tx, shutdown, None))
};
let pump = PumpHandle {
join,
stop_rx: rx,
shutdown,
};
let mut backend = kd_backend_with_pump(pump, breakin_clone);
let (continue_tx, continue_rx) = mpsc::channel();
let (done_tx, done_rx) = mpsc::channel();
let kernel_thread = std::thread::spawn(move || {
let pc = 0xfffff800_deadbeef;
let mut payload = exception_state_change_payload(pc);
payload[32..36].copy_from_slice(&0x8000_0004u32.to_le_bytes());
kernel
.write_all(&wire_data_packet(
PACKET_TYPE_KD_STATE_CHANGE64,
WIRE_FIRST_PACKET_ID,
&payload,
))
.unwrap();
kernel.flush().unwrap();
let ack = read_wire_packet(&mut kernel);
assert_eq!(
u32::from_le_bytes(ack[0..4].try_into().unwrap()),
WIRE_CONTROL_LEADER
);
assert_eq!(
u16::from_le_bytes(ack[4..6].try_into().unwrap()),
PACKET_TYPE_KD_ACKNOWLEDGE
);
let continue_packet = read_wire_packet(&mut kernel);
continue_tx.send(continue_packet).unwrap();
kernel
.write_all(&wire_control_packet(
PACKET_TYPE_KD_ACKNOWLEDGE,
WIRE_FIRST_PACKET_ID,
))
.unwrap();
kernel.flush().unwrap();
done_rx.recv_timeout(Duration::from_secs(5)).unwrap();
});
backend.prepare_for_exit(true).unwrap();
done_tx.send(()).unwrap();
kernel_thread.join().expect("kernel thread panicked");
let continue_packet = continue_rx
.recv_timeout(Duration::from_secs(5))
.expect("kernel thread did not capture continue packet");
assert!(backend.is_running);
assert!(backend.pump.is_none());
assert!(backend.framing.is_some());
assert_eq!(
u32::from_le_bytes(continue_packet[0..4].try_into().unwrap()),
WIRE_DATA_LEADER
);
assert_eq!(
u16::from_le_bytes(continue_packet[4..6].try_into().unwrap()),
PACKET_TYPE_KD_STATE_MANIPULATE
);
let request = &continue_packet[WIRE_HEADER_SIZE..];
assert_eq!(
u32::from_le_bytes(request[0..4].try_into().unwrap()),
api::DBGKD_CONTINUE_API2
);
assert_eq!(
u32::from_le_bytes(request[16..20].try_into().unwrap()),
api::DBG_CONTINUE
);
}
#[test]
fn pump_sends_breakin_after_peer_reset_while_waiting_for_reconnect() {
let (mut kernel, host) = UnixStream::pair().unwrap();
kernel
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let framing = KdFraming::new(host);
let (tx, rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let handle = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || run_pump(framing, tx, shutdown, None))
};
kernel
.write_all(&wire_control_packet(PACKET_TYPE_KD_RESET, 0))
.unwrap();
kernel.flush().unwrap();
let deadline = Instant::now() + Duration::from_secs(2);
let mut saw_breakin = false;
let mut buf = [0u8; 64];
while Instant::now() < deadline && !saw_breakin {
match kernel.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
saw_breakin = buf[..n].contains(&BREAKIN_BYTE);
}
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
Err(e) => panic!("failed to read pump output: {e}"),
}
}
assert!(saw_breakin, "pump should assist reboot reconnects");
shutdown.store(true, Ordering::SeqCst);
let _framing = handle.join().expect("pump thread panicked");
assert!(
rx.try_recv().is_err(),
"reset alone should not report a stop"
);
}
#[test]
fn pump_tags_stop_after_assisted_reconnect_breakin() {
let (mut kernel, host) = UnixStream::pair().unwrap();
kernel
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let framing = KdFraming::new(host);
let (tx, rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let handle = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || run_pump(framing, tx, shutdown, None))
};
kernel
.write_all(&wire_control_packet(PACKET_TYPE_KD_RESET, 0))
.unwrap();
kernel.flush().unwrap();
let deadline = Instant::now() + Duration::from_secs(2);
let mut saw_breakin = false;
let mut buf = [0u8; 64];
while Instant::now() < deadline && !saw_breakin {
match kernel.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
saw_breakin = buf[..n].contains(&BREAKIN_BYTE);
}
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
Err(e) => panic!("failed to read pump output: {e}"),
}
}
assert!(saw_breakin, "pump should send reconnect break-in");
let pc = 0xfffff800_deadbeef;
kernel
.write_all(&wire_data_packet(
PACKET_TYPE_KD_STATE_CHANGE64,
WIRE_FIRST_PACKET_ID,
&exception_state_change_payload(pc),
))
.unwrap();
kernel.flush().unwrap();
let stop = rx
.recv_timeout(Duration::from_secs(5))
.expect("pump reported no stop")
.expect("pump reported an error");
assert_eq!(stop.program_counter, pc);
assert!(stop.target_reloaded);
assert!(stop.assisted_breakin);
shutdown.store(true, Ordering::SeqCst);
let _framing = handle.join().expect("pump thread panicked");
}
#[test]
fn pump_surfaces_reloaded_transparent_state_change() {
let (mut kernel, host) = UnixStream::pair().unwrap();
let framing = KdFraming::new(host);
let (tx, rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let handle = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || run_pump(framing, tx, shutdown, None))
};
kernel
.write_all(&wire_control_packet(PACKET_TYPE_KD_RESET, 0))
.unwrap();
let pc = 0xfffff800_feedface;
kernel
.write_all(&wire_data_packet(
PACKET_TYPE_KD_STATE_CHANGE64,
WIRE_FIRST_PACKET_ID,
&state_change_payload(DBG_KD_LOAD_SYMBOLS_STATE_CHANGE, pc),
))
.unwrap();
kernel.flush().unwrap();
let stop = rx
.recv_timeout(Duration::from_secs(5))
.expect("pump reported no stop")
.expect("pump reported an error");
assert_eq!(stop.new_state, DBG_KD_LOAD_SYMBOLS_STATE_CHANGE);
assert_eq!(stop.program_counter, pc);
assert!(stop.target_reloaded);
shutdown.store(true, Ordering::SeqCst);
let _framing = handle.join().expect("pump thread panicked");
}
#[test]
fn pump_sends_breakin_when_started_in_reconnect_assist_mode() {
let (mut kernel, host) = UnixStream::pair().unwrap();
kernel
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let framing = KdFraming::new(host);
let (tx, rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let handle = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || run_pump(framing, tx, shutdown, Some(Duration::ZERO)))
};
let deadline = Instant::now() + Duration::from_secs(2);
let mut saw_breakin = false;
let mut buf = [0u8; 64];
while Instant::now() < deadline && !saw_breakin {
match kernel.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
saw_breakin = buf[..n].contains(&BREAKIN_BYTE);
}
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
Err(e) => panic!("failed to read pump output: {e}"),
}
}
assert!(
saw_breakin,
"post-bugcheck reconnect assist should not require a reset packet first"
);
shutdown.store(true, Ordering::SeqCst);
let _framing = handle.join().expect("pump thread panicked");
assert!(
rx.try_recv().is_err(),
"assist alone should not report a stop"
);
}
#[test]
fn pump_does_not_send_delayed_reconnect_assist_before_delay() {
let (mut kernel, host) = UnixStream::pair().unwrap();
kernel
.set_read_timeout(Some(Duration::from_millis(5)))
.unwrap();
let framing = KdFraming::new(host);
let (tx, _rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let handle = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || {
run_pump(framing, tx, shutdown, Some(Duration::from_secs(60)))
})
};
let deadline = Instant::now() + Duration::from_millis(200);
let mut saw_breakin = false;
let mut buf = [0u8; 64];
while Instant::now() < deadline {
match kernel.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
if buf[..n].contains(&BREAKIN_BYTE) {
saw_breakin = true;
break;
}
}
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
Err(e) => panic!("failed to read pump output: {e}"),
}
}
assert!(
!saw_breakin,
"delayed post-bugcheck reconnect assist should not fire immediately"
);
shutdown.store(true, Ordering::SeqCst);
let _framing = handle.join().expect("pump thread panicked");
}
#[test]
fn await_refresh_sets_flag_without_breakin() {
let (mut kernel, host) = UnixStream::pair().unwrap();
kernel
.set_read_timeout(Some(Duration::from_millis(5)))
.unwrap();
let handle = std::thread::spawn(move || {
let mut framing = KdFraming::new(host);
let mut saw_refresh = false;
let stop = await_state_change(&mut framing, Some(&mut saw_refresh), false, None, None)
.expect("await_state_change failed");
(saw_refresh, stop)
});
let refresh = debug_io_print_payload(KD_REFRESH_MESSAGE);
kernel
.write_all(&wire_data_packet(
PACKET_TYPE_KD_DEBUG_IO,
WIRE_FIRST_PACKET_ID,
&refresh,
))
.unwrap();
kernel.flush().unwrap();
let mut outbound = Vec::new();
let mut buf = [0u8; 64];
while outbound.len() < WIRE_HEADER_SIZE {
match kernel.read(&mut buf) {
Ok(0) => break,
Ok(n) => outbound.extend_from_slice(&buf[..n]),
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {
break;
}
Err(e) => panic!("failed to read ACK: {e}"),
}
}
assert!(
outbound.len() >= WIRE_HEADER_SIZE,
"refresh packet should be ACKed"
);
assert!(
!outbound.contains(&BREAKIN_BYTE),
"refresh ACK should not include a break-in"
);
let immediate_window = Instant::now() + Duration::from_millis(30);
while Instant::now() < immediate_window {
match kernel.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
assert!(
!buf[..n].contains(&BREAKIN_BYTE),
"plain KD refresh should not trigger an immediate break-in"
);
}
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {
break;
}
Err(e) => panic!("failed to read post-refresh output: {e}"),
}
}
let pc = 0xfffff800_deadbeef;
kernel
.write_all(&wire_data_packet(
PACKET_TYPE_KD_STATE_CHANGE64,
WIRE_FIRST_PACKET_ID ^ 1,
&exception_state_change_payload(pc),
))
.unwrap();
kernel.flush().unwrap();
let (saw_refresh, stop) = handle.join().expect("await thread panicked");
assert!(saw_refresh);
assert_eq!(stop.program_counter, pc);
}
#[test]
fn pump_does_not_breakin_immediately_on_bugcheck_refresh_print() {
let (mut kernel, host) = UnixStream::pair().unwrap();
kernel
.set_read_timeout(Some(Duration::from_millis(5)))
.unwrap();
let framing = KdFraming::new(host);
let (tx, _rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let handle = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || run_pump(framing, tx, shutdown, None))
};
let refresh = debug_io_print_payload(KD_REFRESH_MESSAGE);
kernel
.write_all(&wire_data_packet(
PACKET_TYPE_KD_DEBUG_IO,
WIRE_FIRST_PACKET_ID,
&refresh,
))
.unwrap();
kernel.flush().unwrap();
let deadline = Instant::now() + Duration::from_secs(2);
let mut buf = [0u8; 64];
let mut outbound = Vec::new();
while Instant::now() < deadline && outbound.len() < WIRE_HEADER_SIZE {
match kernel.read(&mut buf) {
Ok(0) => break,
Ok(n) => outbound.extend_from_slice(&buf[..n]),
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
Err(e) => panic!("failed to read pump output: {e}"),
}
}
assert!(
outbound.len() >= WIRE_HEADER_SIZE,
"pump should ACK the refresh print"
);
let mut saw_breakin = false;
let immediate_window = Instant::now() + Duration::from_millis(30);
while Instant::now() < immediate_window {
match kernel.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
outbound.extend_from_slice(&buf[..n]);
if buf[..n].contains(&BREAKIN_BYTE) {
saw_breakin = true;
break;
}
}
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {
break;
}
Err(e) => panic!("failed to read pump output: {e}"),
}
}
assert!(
!saw_breakin,
"bugcheck refresh should not interrupt the remaining debug text immediately"
);
shutdown.store(true, Ordering::SeqCst);
let _framing = handle.join().expect("pump thread panicked");
}
#[test]
fn pump_does_not_assist_non_e2_bugcheck_after_code_is_captured() {
let (mut kernel, host) = UnixStream::pair().unwrap();
kernel
.set_read_timeout(Some(Duration::from_millis(5)))
.unwrap();
let framing = KdFraming::new(host);
let (tx, _rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let handle = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || run_pump(framing, tx, shutdown, None))
};
let refresh = debug_io_print_payload(KD_REFRESH_MESSAGE);
kernel
.write_all(&wire_data_packet(
PACKET_TYPE_KD_DEBUG_IO,
WIRE_FIRST_PACKET_ID,
&refresh,
))
.unwrap();
let fatal = debug_io_print_payload(
b"\r\n*** Fatal System Error: 0x000000d1\r\n (0x1,0x2,0x0,0x4)\r\n",
);
kernel
.write_all(&wire_data_packet(
PACKET_TYPE_KD_DEBUG_IO,
WIRE_FIRST_PACKET_ID ^ 1,
&fatal,
))
.unwrap();
kernel.flush().unwrap();
let deadline = Instant::now() + Duration::from_millis(300);
let mut saw_breakin = false;
let mut buf = [0u8; 128];
while Instant::now() < deadline {
match kernel.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
if buf[..n].contains(&BREAKIN_BYTE) {
saw_breakin = true;
break;
}
}
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
Err(e) => panic!("failed to read pump output: {e}"),
}
}
assert!(
!saw_breakin,
"ordinary bugchecks should rely on the kernel-driven break once the code is known"
);
shutdown.store(true, Ordering::SeqCst);
let _framing = handle.join().expect("pump thread panicked");
}
#[test]
fn only_exception_state_changes_surface_as_breaks() {
assert!(!is_transparent_state_change(DBG_KD_EXCEPTION_STATE_CHANGE));
assert!(!is_transparent_state_change(0xdead_beef));
assert!(is_transparent_state_change(
DBG_KD_LOAD_SYMBOLS_STATE_CHANGE
));
assert!(is_transparent_state_change(
DBG_KD_COMMAND_STRING_STATE_CHANGE
));
}
#[test]
fn pump_exits_on_shutdown_when_idle() {
let (_kernel, host) = UnixStream::pair().unwrap();
let framing = KdFraming::new(host);
let (tx, rx) = mpsc::channel();
let shutdown = Arc::new(AtomicBool::new(false));
let handle = {
let shutdown = Arc::clone(&shutdown);
std::thread::spawn(move || run_pump(framing, tx, shutdown, None))
};
shutdown.store(true, Ordering::SeqCst);
let _framing = handle.join().expect("pump thread panicked");
assert!(rx.try_recv().is_err(), "idle pump should report no stop");
}
}