use std::time::Duration;
use serde_json::Value;
use crate::api::{Cancellation, types::ApiError};
use crate::cli::parse_number;
use crate::live::cli::{
DEFAULT_AFTER_WORDS, DEFAULT_ALIGN, DEFAULT_BEFORE_WORDS, DEFAULT_MAX_HITS, DEFAULT_PROBE_DURATION_MS,
MAX_CONTEXT_WORDS, MAX_PATTERN_WORDS, MAX_PROBES_PER_SESSION, MAX_WATCHES_PER_SESSION, ProbeBackend,
WatchAccessMode, WatchLen, encode_values_le, format_bytes_hex, validate_align, validate_context_words,
};
use crate::mcp::live_tool::{
LiveProbeBackend, LiveQuery, LiveWatchAccess, MAX_MCP_LIVE_ADDRS, MAX_MCP_LIVE_BYTES, MAX_MCP_LIVE_DURATION_MS,
MAX_MCP_LIVE_EVENTS, MAX_MCP_LIVE_HITS,
};
pub fn validate_live_query_budgets(query: &LiveQuery) -> Result<(), ApiError> {
match query {
LiveQuery::Maps {} => Ok(()),
LiveQuery::Resolve { addresses } => {
let _ = parse_addr_list(addresses, 1, MAX_MCP_LIVE_ADDRS, "resolve")?;
Ok(())
}
LiveQuery::Memscan { values, align, before_words, after_words, max_hits, max_bytes, timeout_ms } => {
let _ = parse_addr_list(values, 1, MAX_PATTERN_WORDS, "memscan values")?;
validate_align(align.unwrap_or(DEFAULT_ALIGN)).map_err(|e| ApiError::new("invalid_request", e))?;
validate_context_words(before_words.unwrap_or(DEFAULT_BEFORE_WORDS), "before_words")
.map_err(|e| ApiError::new("invalid_request", e))?;
validate_context_words(after_words.unwrap_or(DEFAULT_AFTER_WORDS), "after_words")
.map_err(|e| ApiError::new("invalid_request", e))?;
let hits = max_hits.unwrap_or(DEFAULT_MAX_HITS);
if hits == 0 || hits > MAX_MCP_LIVE_HITS {
return Err(ApiError::new(
"invalid_request",
format!("max_hits {hits} must be in 1..={MAX_MCP_LIVE_HITS} for MCP live_memscan"),
));
}
let bytes = max_bytes.unwrap_or(MAX_MCP_LIVE_BYTES);
if bytes == 0 || bytes > MAX_MCP_LIVE_BYTES {
return Err(ApiError::new(
"invalid_request",
format!("max_bytes {bytes} must be in 1..={MAX_MCP_LIVE_BYTES} for MCP live_memscan"),
));
}
let timeout = timeout_ms.unwrap_or(MAX_MCP_LIVE_DURATION_MS);
if timeout == 0 || timeout > MAX_MCP_LIVE_DURATION_MS {
return Err(ApiError::new(
"invalid_request",
format!("timeout_ms {timeout} must be in 1..={MAX_MCP_LIVE_DURATION_MS} for MCP live_memscan"),
));
}
Ok(())
}
LiveQuery::Probe { addresses, duration_ms, max_events, .. } => {
let _ = parse_addr_list(addresses, 1, MAX_PROBES_PER_SESSION, "probe")?;
validate_duration_and_events(*duration_ms, *max_events)
}
LiveQuery::Watch { addresses, follow, len_bytes, access, duration_ms, max_events, .. } => {
let len = parse_mcp_watch_len(*len_bytes)?;
let addrs = parse_addr_list(addresses, 0, MAX_WATCHES_PER_SESSION, "watch")?;
let mut total_slots = addrs.len();
for spec_str in follow {
let parsed =
crate::live::watch::rebind::parse_follow_spec(spec_str, len, WatchAccessMode::from(*access))
.map_err(|e| ApiError::new("invalid_request", e))?;
total_slots = total_slots.saturating_add(parsed.hardware_slots_required());
}
if total_slots == 0 || total_slots > MAX_WATCHES_PER_SESSION {
return Err(ApiError::new(
"invalid_request",
format!(
"watch requires 1..={MAX_WATCHES_PER_SESSION} total hardware watchpoint slots across addresses and follow (requested {total_slots})"
),
));
}
validate_duration_and_events(*duration_ms, *max_events)
}
}
}
fn validate_duration_and_events(duration_ms: Option<u64>, max_events: Option<usize>) -> Result<(), ApiError> {
let dur = duration_ms.unwrap_or(DEFAULT_PROBE_DURATION_MS);
if dur == 0 || dur > MAX_MCP_LIVE_DURATION_MS {
return Err(ApiError::new(
"invalid_request",
format!("duration_ms {dur} must be in 1..={MAX_MCP_LIVE_DURATION_MS} for MCP live tool"),
));
}
let ev = max_events.unwrap_or(MAX_MCP_LIVE_EVENTS);
if ev == 0 || ev > MAX_MCP_LIVE_EVENTS {
return Err(ApiError::new(
"invalid_request",
format!("max_events {ev} must be in 1..={MAX_MCP_LIVE_EVENTS} for MCP live tool"),
));
}
Ok(())
}
pub fn parse_mcp_watch_len(len_bytes: Option<u8>) -> Result<WatchLen, ApiError> {
match len_bytes.unwrap_or(8) {
1 => Ok(WatchLen::B1),
2 => Ok(WatchLen::B2),
4 => Ok(WatchLen::B4),
8 => Ok(WatchLen::B8),
other => Err(ApiError::new(
"invalid_request",
format!("invalid len_bytes {other}: allowed hardware watchpoint lengths are 1, 2, 4, or 8"),
)),
}
}
pub fn parse_addr_list(raw: &[String], min: usize, max: usize, label: &str) -> Result<Vec<u64>, ApiError> {
if raw.len() < min || raw.len() > max {
return Err(ApiError::new(
"invalid_request",
format!("{label} requires {min}..={max} entries (got {})", raw.len()),
));
}
let mut out = Vec::with_capacity(raw.len());
for item in raw {
let val = parse_number(item.trim())
.map_err(|e| ApiError::new("invalid_address", format!("invalid {label} entry {item:?}: {e}")))?;
out.push(val);
}
Ok(out)
}
pub fn classify_live_error(err: &str) -> ApiError {
for code in [
"cancelled",
"process_exited",
"process_changed",
"binary_mismatch",
"permission_denied",
"unsupported_platform",
"unsupported_uprobe_pmu",
] {
if let Some(rest) = err.strip_prefix(&format!("{code}:")) {
return ApiError::new(code, rest.trim());
}
}
ApiError::new("live_failed", err)
}
#[cfg(any(target_os = "linux", test))]
pub fn run_mcp_memscan(
session: &mut crate::live::procfs::LiveSession,
pattern: (&[String], Option<usize>, Option<usize>, Option<usize>),
budgets: (Option<usize>, Option<u64>, Option<u64>),
cancel: &Cancellation,
) -> Result<Value, ApiError> {
let (values_raw, align_opt, before_opt, after_opt) = pattern;
let (max_hits_opt, max_bytes_opt, timeout_opt) = budgets;
let values = parse_addr_list(values_raw, 1, MAX_PATTERN_WORDS, "memscan values")?;
let align = align_opt.unwrap_or(DEFAULT_ALIGN);
let before_words = before_opt.unwrap_or(DEFAULT_BEFORE_WORDS);
let after_words = after_opt.unwrap_or(DEFAULT_AFTER_WORDS);
if before_words > MAX_CONTEXT_WORDS || after_words > MAX_CONTEXT_WORDS {
return Err(ApiError::new("invalid_request", format!("context words must be in 0..={MAX_CONTEXT_WORDS}")));
}
let mem_file = crate::live::procfs::open_process_mem(session.process.pid, session.process.starttime_ticks)
.map_err(|e| classify_live_error(&e))?;
let needle = encode_values_le(&values);
let cfg = crate::live::memscan::ScanConfig {
align,
before_words,
after_words,
max_hits: max_hits_opt.unwrap_or(DEFAULT_MAX_HITS),
max_bytes: max_bytes_opt.unwrap_or(MAX_MCP_LIVE_BYTES),
timeout: Duration::from_millis(timeout_opt.unwrap_or(MAX_MCP_LIVE_DURATION_MS)),
};
let outcome = crate::live::memscan::scan_memory_cancellable(
&mem_file,
&session.image,
session.process.pie_slide,
&session.regions,
&needle,
&cfg,
Some(cancel.flag()),
)
.map_err(|e| classify_live_error(&e))?;
crate::live::procfs::refresh_session_identity(session);
let report = crate::live::types::MemscanReport {
schema_version: crate::live::types::SCHEMA_VERSION,
tool: crate::live::types::TOOL_NAME,
tool_version: crate::live::types::TOOL_VERSION,
operation: "live_memscan",
process: session.process.clone(),
needle_hex: format_bytes_hex(&needle),
align,
elapsed_ms: outcome.elapsed_ms,
regions_scanned: outcome.regions_scanned,
regions_skipped: outcome.regions_skipped,
bytes_scanned: outcome.bytes_scanned,
unreadable_ranges: outcome.unreadable_ranges,
stop_reason: outcome.stop_reason,
hits: outcome.hits,
};
serde_json::to_value(report).map_err(|e| ApiError::new("serialization_failed", e.to_string()))
}
#[cfg(any(target_os = "linux", test))]
pub fn run_mcp_probe(
session: &mut crate::live::procfs::LiveSession,
addresses: &[String],
backend: LiveProbeBackend,
capture: (bool, bool),
budgets: (Option<u64>, Option<usize>),
cancel: &Cancellation,
) -> Result<Value, ApiError> {
let addrs = parse_addr_list(addresses, 1, MAX_PROBES_PER_SESSION, "probe")?;
crate::live::probe::validate::validate_session_for_probe(session).map_err(|e| classify_live_error(&e))?;
let specs = crate::live::probe::validate::validate_probe_targets(
&session.image,
session.process.pie_slide,
&session.regions,
&addrs,
)
.map_err(|e| ApiError::new("invalid_request", e))?;
let cfg = crate::live::probe::tracefs::ProbeRunConfig {
duration: Duration::from_millis(budgets.0.unwrap_or(DEFAULT_PROBE_DURATION_MS)),
max_events: budgets.1.unwrap_or(MAX_MCP_LIVE_EVENTS),
want_callers: capture.0,
want_regs: capture.1,
spawn_watchdog: true,
cancel: Some(cancel.flag()),
};
let backend_enum = ProbeBackend::from(backend);
let (obs, probes) = match backend_enum {
ProbeBackend::Perf => crate::live::probe::perf::run_perf_probe(session, &specs, &cfg),
ProbeBackend::Tracefs => crate::live::probe::tracefs::run_tracefs_probe(session, &specs, &cfg),
}
.map_err(|e| classify_live_error(&e))?;
let report = crate::live::probe::build_probe_report(
backend_enum,
session.process.clone(),
obs,
probes,
&specs,
cfg.want_callers,
cfg.want_regs,
);
serde_json::to_value(report).map_err(|e| ApiError::new("serialization_failed", e.to_string()))
}
#[cfg(any(target_os = "linux", test))]
pub fn run_mcp_watch(
session: &mut crate::live::procfs::LiveSession,
targets: (&[String], &[String]),
mode: (Option<u8>, LiveWatchAccess, bool),
budgets: (Option<u64>, Option<usize>),
cancel: &Cancellation,
) -> Result<Value, ApiError> {
let len = parse_mcp_watch_len(mode.0)?;
let access = WatchAccessMode::from(mode.1);
let addrs = parse_addr_list(targets.0, 0, MAX_WATCHES_PER_SESSION, "watch")?;
crate::live::watch::validate::validate_session_for_watch(session).map_err(|e| classify_live_error(&e))?;
let mem_file = crate::live::procfs::open_process_mem(session.process.pid, session.process.starttime_ticks).ok();
let mem_reader: Option<&dyn crate::live::memscan::ReadAt> =
mem_file.as_ref().map(|f| f as &dyn crate::live::memscan::ReadAt);
let specs = crate::live::watch::validate::validate_watch_targets(
&session.image,
session.process.pie_slide,
&session.regions,
&addrs,
len,
access,
mem_reader,
)
.map_err(|e| ApiError::new("invalid_request", e))?;
let dependent = crate::live::watch::rebind_runtime::prepare_dependent_watches(
(&session.image, session.process.pie_slide, &session.regions),
(targets.1, len, access),
specs.len(),
mem_reader,
)
.map_err(|e| ApiError::new("invalid_request", e))?;
let cfg = crate::live::watch::perf::WatchRunConfig {
duration: Duration::from_millis(budgets.0.unwrap_or(DEFAULT_PROBE_DURATION_MS)),
max_events: budgets.1.unwrap_or(MAX_MCP_LIVE_EVENTS),
want_regs: mode.2,
cancel: Some(cancel.flag()),
};
let (obs, watches, dep_results) =
crate::live::watch::perf::run_perf_watch(session, &specs, dependent, &cfg, mem_reader)
.map_err(|e| classify_live_error(&e))?;
let report = crate::live::watch::build_watch_report_with_dependent(
session.process.clone(),
obs,
watches,
dep_results,
mode.2,
mem_file.is_some(),
);
serde_json::to_value(report).map_err(|e| ApiError::new("serialization_failed", e.to_string()))
}