diff --git a/README.md b/README.md index b915a9c..1ef9ca9 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,7 @@ latency / TTFT, or plaintext) / **where** (peer IP / hostname). | signal | kernel hook | event | |---|---|---| -| `exec` | `sys_enter_execve` | `ToolExec` — tool / subprocess: **full argv** + cwd, comm, uid | +| `exec` | `sys_enter_execve` + `sched_process_exec` | `ToolExec` — bounded argv fragments, successful-exec confirmation, `/proc` supplementation + cwd, comm, uid | | `exit` | `do_exit` kprobe | `ProcessExit` — outcome: **exit code + signal** (clean / SIGSEGV crash / SIGKILL-OOM), one per process | | `connect` | `sys_enter_connect` | `Egress` — peer IP:port | | `sni` | TLS ClientHello (plaintext `server_name`) | LLM **provider** + endpoint | @@ -71,12 +71,25 @@ agent action and should not be fed into security policy decisions. "ttft":{"secs":0,"nanos":410000000}}}} {"identity":{"agent":"python3","task":"1903","session":null},"provider":null, "event":{"ToolExec":{"pid":1903,"ppid":1841, - "argv":["git","clone","https://github.com/acme/repo"],"cwd":"/home/agent/work"}}} + "argv":["git","clone","https://github.com/acme/repo"],"argv_truncated":false, + "argv_incomplete":false,"exec_confirmed":true,"argv_source":"kernel_fragments", + "captured_argc":3,"captured_bytes":36,"observed_argc":3,"observed_bytes":36, + "cwd":"/home/agent/work"}}} {"identity":{"agent":"python3","task":"1841","session":null},"provider":null, "event":{"SslContent":{"pid":1841,"is_read":false, "content":"POST /v1/messages HTTP/1.1\r\nHost: api.anthropic.com\r\n..."}}} ``` +`ToolExec.argv` is reconstructed in the collector from bounded 128-byte kernel records. The +collector captures up to 12 arguments and roughly 8 KiB across argv. `argv_truncated=true` means +the configured limit was reached; `argv_incomplete=true` means a chunk was lost or reassembly +timed out. After `sched_process_exec` confirms a successful exec, the collector attempts to replace +truncated/incomplete fragments with `/proc//cmdline` (bounded to 2 MiB). `argv_source` states +which source won and `exec_confirmed` distinguishes committed execs from failed attempts. Very +short-lived processes can exit before `/proc` is read, so the explicit truncation/incomplete flags +remain the authoritative evidence-quality signal. Neither condition is silent, and both counters +are included in collector heartbeats. + Filter with `jq`, e.g. every LLM call and its provider: `… | jq -c 'select(.event.LlmCall) | {agent:.identity.agent, provider, sni:.event.LlmCall.sni}'`. diff --git a/a3s-observer-collector/src/main.rs b/a3s-observer-collector/src/main.rs index 3d2e0e3..b5e1ab3 100644 --- a/a3s-observer-collector/src/main.rs +++ b/a3s-observer-collector/src/main.rs @@ -11,8 +11,10 @@ use a3s_observer::{ KubeResolver, LogExporter, ProcessContext, Provider, ServiceClassifier, SniClassifier, }; use a3s_observer_common::{ - ConnectEvent, DnsEvent, ExecEvent, ExitEvent, FileEvent, LlmEvent, SecEvent, SslEvent, - TlsEvent, FILE_DELETE_FLAG, SEC_BIND, SEC_PTRACE, SEC_SETUID, + ConnectEvent, DnsEvent, ExecRecord, ExitEvent, FileEvent, LlmEvent, SecEvent, SslEvent, + TlsEvent, ARGV_SLOTS, EXEC_ARG_CHUNK_LEN, EXEC_ARG_CHUNK_PAYLOAD, EXEC_FLAG_ARGV_INCOMPLETE, + EXEC_FLAG_ARGV_TRUNCATED, EXEC_MAX_CHUNKS, EXEC_RECORD_ARG_CHUNK, EXEC_RECORD_COMMIT, + EXEC_RECORD_END, EXEC_RECORD_HEADER, FILE_DELETE_FLAG, SEC_BIND, SEC_PTRACE, SEC_SETUID, }; use anyhow::Context as _; use aya::{ @@ -20,9 +22,243 @@ use aya::{ programs::{KProbe, TracePoint, UProbe}, Ebpf, }; -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::fs::File; +use std::io::Read; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; -use std::time::Duration; +use std::path::Path; +use std::time::{Duration, Instant}; + +const EXEC_REASSEMBLY_TIMEOUT: Duration = Duration::from_millis(500); +const EXEC_REASSEMBLY_LIMIT: usize = 4096; +const PROC_CMDLINE_MAX_BYTES: usize = 2 * 1024 * 1024; + +struct PendingExec { + first_seen: Instant, + pid: u32, + ppid: u32, + uid: u32, + comm: [u8; 16], + filename: [u8; EXEC_ARG_CHUNK_LEN], + saw_header: bool, + saw_commit: bool, + argc: Option, + captured_bytes: Option, + flags: u8, + chunks: HashMap>>, +} + +impl PendingExec { + fn new(record: &ExecRecord, now: Instant) -> Self { + Self { + first_seen: now, + pid: record.pid, + ppid: record.ppid, + uid: record.uid, + comm: record.comm, + filename: [0; EXEC_ARG_CHUNK_LEN], + saw_header: false, + saw_commit: false, + argc: None, + captured_bytes: None, + flags: 0, + chunks: HashMap::new(), + } + } + + fn apply(&mut self, record: &ExecRecord) { + self.flags |= record.flags; + match record.kind { + EXEC_RECORD_HEADER => { + self.saw_header = true; + let len = (record.data_len as usize).min(record.data.len()); + self.filename[..len].copy_from_slice(&record.data[..len]); + } + EXEC_RECORD_ARG_CHUNK => { + if record.arg_index as usize >= ARGV_SLOTS + || record.chunk_index as usize >= EXEC_MAX_CHUNKS + || record.data_len as usize > record.data.len() + { + self.flags |= EXEC_FLAG_ARGV_INCOMPLETE; + return; + } + let len = record.data_len as usize; + let value = record.data[..len].to_vec(); + let chunks = self.chunks.entry(record.arg_index).or_default(); + if let Some(existing) = chunks.get(&record.chunk_index) { + if existing != &value { + self.flags |= EXEC_FLAG_ARGV_INCOMPLETE; + } + } else { + chunks.insert(record.chunk_index, value); + } + } + EXEC_RECORD_END => { + self.argc = Some(record.argc.min(ARGV_SLOTS as u16)); + self.captured_bytes = Some(record.captured_bytes); + } + EXEC_RECORD_COMMIT => self.saw_commit = true, + _ => self.flags |= EXEC_FLAG_ARGV_INCOMPLETE, + } + } + + fn finish(mut self, timed_out: bool) -> CompletedExec { + let mut argv_incomplete = timed_out + || !self.saw_header + || self.argc.is_none() + || self.flags & EXEC_FLAG_ARGV_INCOMPLETE != 0; + let argv_truncated = self.flags & EXEC_FLAG_ARGV_TRUNCATED != 0; + let inferred_argc = self + .chunks + .keys() + .max() + .map(|index| index.saturating_add(1)) + .unwrap_or(0); + let captured_argc = self.argc.unwrap_or(inferred_argc).min(ARGV_SLOTS as u16); + let mut argv = Vec::with_capacity(captured_argc as usize); + let mut assembled_bytes = 0u32; + + for arg_index in 0..captured_argc { + let Some(chunks) = self.chunks.remove(&arg_index) else { + argv_incomplete = true; + argv.push(String::new()); + continue; + }; + let mut bytes = Vec::new(); + let mut expected_chunk = 0u16; + let mut saw_terminator = false; + for (chunk_index, chunk) in chunks { + if chunk_index != expected_chunk { + argv_incomplete = true; + } + expected_chunk = chunk_index.saturating_add(1); + assembled_bytes = assembled_bytes.saturating_add(chunk.len() as u32); + if chunk.len() < EXEC_ARG_CHUNK_PAYLOAD { + saw_terminator = true; + } + bytes.extend_from_slice(&chunk); + } + if !(saw_terminator || argv_truncated && arg_index + 1 == captured_argc) { + argv_incomplete = true; + } + argv.push(String::from_utf8_lossy(&bytes).into_owned()); + } + if !self.chunks.is_empty() { + argv_incomplete = true; + } + if let Some(expected_bytes) = self.captured_bytes { + if expected_bytes != assembled_bytes { + argv_incomplete = true; + } + } + if argv.is_empty() { + let filename = cstr(&self.filename); + if !filename.is_empty() { + argv.push(filename); + } + } + + CompletedExec { + pid: self.pid, + ppid: self.ppid, + uid: self.uid, + comm: self.comm, + filename: self.filename, + argv, + argv_truncated, + argv_incomplete, + captured_argc, + captured_bytes: self.captured_bytes.unwrap_or(assembled_bytes), + reassembly_timed_out: timed_out && (self.argc.is_none() || !self.saw_header), + exec_confirmed: self.saw_commit, + } + } +} + +struct CompletedExec { + pid: u32, + ppid: u32, + uid: u32, + comm: [u8; 16], + filename: [u8; EXEC_ARG_CHUNK_LEN], + argv: Vec, + argv_truncated: bool, + argv_incomplete: bool, + captured_argc: u16, + captured_bytes: u32, + reassembly_timed_out: bool, + exec_confirmed: bool, +} + +struct ExecAssembler { + pending: HashMap<(u64, u32), PendingExec>, + require_commit: bool, +} + +impl Default for ExecAssembler { + fn default() -> Self { + Self { + pending: HashMap::new(), + require_commit: true, + } + } +} + +impl ExecAssembler { + fn new(require_commit: bool) -> Self { + Self { + pending: HashMap::new(), + require_commit, + } + } + + fn push(&mut self, record: ExecRecord, now: Instant) -> Vec { + let mut completed = Vec::with_capacity(2); + let key = (record.exec_id, record.pid); + if !self.pending.contains_key(&key) && self.pending.len() >= EXEC_REASSEMBLY_LIMIT { + if let Some(oldest) = self + .pending + .iter() + .min_by_key(|(_, pending)| pending.first_seen) + .map(|(key, _)| *key) + { + if let Some(pending) = self.pending.remove(&oldest) { + completed.push(pending.finish(true)); + } + } + } + + self.pending + .entry(key) + .or_insert_with(|| PendingExec::new(&record, now)) + .apply(&record); + let ready = self.pending.get(&key).is_some_and(|pending| { + pending.argc.is_some() && (pending.saw_commit || !self.require_commit) + }); + if ready { + if let Some(pending) = self.pending.remove(&key) { + completed.push(pending.finish(false)); + } + } + completed + } + + fn expire(&mut self, now: Instant) -> Vec { + let expired: Vec<(u64, u32)> = self + .pending + .iter() + .filter(|(_, pending)| { + now.duration_since(pending.first_seen) >= EXEC_REASSEMBLY_TIMEOUT + }) + .map(|(key, _)| *key) + .collect(); + expired + .into_iter() + .filter_map(|key| self.pending.remove(&key)) + .map(|pending| pending.finish(true)) + .collect() + } +} #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -62,6 +298,10 @@ async fn main() -> anyhow::Result<()> { // unpacking images), and the agent's own writes need downstream identity filtering. let files = std::env::var_os("A3S_OBSERVER_FILES").is_some(); let mut probes = vec![ + ("track_clone", "sys_exit_clone"), + ("track_clone3", "sys_exit_clone3"), + ("track_fork", "sys_exit_fork"), + ("track_vfork", "sys_exit_vfork"), ("exec", "sys_enter_execve"), ("tls_write", "sys_enter_write"), ("tls_sendto", "sys_enter_sendto"), @@ -95,6 +335,33 @@ async fn main() -> anyhow::Result<()> { } } } + // sched_process_fork runs before the child can execute, so its parent PID is available even for short-lived tools. + match attach( + &mut ebpf, + "track_process_fork", + "sched", + "sched_process_fork", + ) { + Ok(()) => attached += 1, + Err(e) => { + tracing::warn!(error = %e, "sched_process_fork probe failed - using syscall-exit ancestry fallback") + } + } + let exec_commit_probe_attached = match attach( + &mut ebpf, + "track_process_exec", + "sched", + "sched_process_exec", + ) { + Ok(()) => { + attached += 1; + true + } + Err(e) => { + tracing::warn!(error = %e, "sched_process_exec probe failed - proc cmdline supplementation disabled"); + false + } + }; // proc_exit is a do_exit kprobe (not a tracepoint): do_exit fires for EVERY task exit, // including signal-kills (crash / OOM) that sys_enter_exit_group never sees. match attach_kprobe(&mut ebpf, "proc_exit", "do_exit") { @@ -148,6 +415,7 @@ async fn main() -> anyhow::Result<()> { // (pid,fd) -> (sni, provider, peer): recorded at ClientHello, read when the socket // closes (the in-kernel LlmEvent) to build the metric-bearing LlmCall. let mut llm_meta: HashMap, Option, IpAddr)> = HashMap::new(); + let mut exec_assembler = ExecAssembler::new(exec_commit_probe_attached); let mut exec_ring = RingBuf::try_from(ebpf.take_map("EVENTS").context("`EVENTS` missing")?)?; let mut exit_ring = RingBuf::try_from( ebpf.take_map("EXIT_EVENTS") @@ -188,7 +456,7 @@ async fn main() -> anyhow::Result<()> { tracing::info!( attached, - total = probes.len(), + total = probes.len() + 3, files, "a3s-observer-collector: probes attached (file-write capture: set A3S_OBSERVER_FILES=1); \ streaming (Ctrl-C to stop)" @@ -238,6 +506,9 @@ async fn main() -> anyhow::Result<()> { emit_collector_heartbeat(exporter.as_ref(), &collector, 60, &stats, dropped, output_dropped); tracing::info!( exec = stats.exec, + exec_truncated = stats.exec_truncated, + exec_incomplete = stats.exec_incomplete, + exec_reassembly_timeout = stats.exec_reassembly_timeout, exit = stats.exit, egress = stats.egress, dns = stats.dns, @@ -257,38 +528,19 @@ async fn main() -> anyhow::Result<()> { // switch to AsyncFd (epoll) on the ring fds and/or enlarge the rings. _ = tokio::time::sleep(Duration::from_millis(20)) => { while let Some(item) = exec_ring.next() { - if let Some(ev) = read_pod::(&item) { - emit(exporter.as_ref(), &mut stats, EnrichedEvent { - identity: identity_for(&resolver, ev.pid, &ev.comm), - workload: resolver.resolve_workload(ev.pid, 0, 0), - observation: None, - process: Some(process_context(ev.pid, &ev.comm)), - provider: None, - event: AgentEvent::ToolExec { - pid: ev.pid, - ppid: read_ppid(ev.pid), - uid: ev.uid, - argv: argv_of(&ev), - cwd: read_cwd(ev.pid), - }, - }); + if let Some(record) = read_pod::(&item) { + for completed in exec_assembler.push(record, Instant::now()) { + emit_completed_exec( + exporter.as_ref(), + &mut stats, + &resolver, + completed, + ); + } } } - while let Some(item) = exit_ring.next() { - if let Some(ev) = read_pod::(&item) { - emit(exporter.as_ref(), &mut stats, EnrichedEvent { - identity: identity_for(&resolver, ev.pid, &ev.comm), - workload: resolver.resolve_workload(ev.pid, 0, 0), - observation: None, - process: Some(process_context(ev.pid, &ev.comm)), - provider: None, - event: AgentEvent::ProcessExit { - pid: ev.pid, - exit_code: ev.exit_code, - signal: ev.signal, - }, - }); - } + for completed in exec_assembler.expire(Instant::now()) { + emit_completed_exec(exporter.as_ref(), &mut stats, &resolver, completed); } while let Some(item) = sec_ring.next() { if let Some(ev) = read_pod::(&item) { @@ -479,11 +731,37 @@ async fn main() -> anyhow::Result<()> { } } } + // Exit is drained last so downstream process caches can attribute all prior + // exec, file, network and security events before the PID lifecycle closes. + while let Some(item) = exit_ring.next() { + if let Some(ev) = read_pod::(&item) { + emit(exporter.as_ref(), &mut stats, EnrichedEvent { + identity: identity_for(&resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, + process: Some(process_context(ev.pid, &ev.comm)), + provider: None, + event: AgentEvent::ProcessExit { + pid: ev.pid, + exit_code: ev.exit_code, + signal: ev.signal, + }, + }); + } + } } } } + let dropped: u64 = drops + .get(&0, 0) + .map(|v| v.iter().copied().sum()) + .unwrap_or(0); + let output_dropped = exporter.output_drops(); tracing::info!( exec = stats.exec, + exec_truncated = stats.exec_truncated, + exec_incomplete = stats.exec_incomplete, + exec_reassembly_timeout = stats.exec_reassembly_timeout, exit = stats.exit, egress = stats.egress, dns = stats.dns, @@ -491,6 +769,8 @@ async fn main() -> anyhow::Result<()> { llm = stats.llm, ssl = stats.ssl, sec = stats.sec, + dropped, + output_dropped, "a3s-observer-collector: stopped (final window)" ); Ok(()) @@ -537,21 +817,6 @@ fn read_pod(item: &[u8]) -> Option { .then(|| unsafe { core::ptr::read_unaligned(item.as_ptr() as *const T) }) } -/// Full argv (argv[0..argc]) captured in-kernel; falls back to the binary path if none. -fn argv_of(ev: &ExecEvent) -> Vec { - let n = (ev.argc as usize).min(ev.args.len()); - let argv: Vec = ev.args[..n] - .iter() - .map(|a| cstr(a)) - .filter(|s| !s.is_empty()) - .collect(); - if argv.is_empty() { - vec![cstr(&ev.filename)] - } else { - argv - } -} - /// The process's current working directory (≈ exec-time for a fresh process). fn read_cwd(pid: u32) -> String { std::fs::read_link(format!("/proc/{pid}/cwd")) @@ -585,6 +850,142 @@ fn process_context(pid: u32, comm: &[u8; 16]) -> ProcessContext { } } +fn exec_ppid(ev: &CompletedExec) -> u32 { + if ev.ppid != 0 { + ev.ppid + } else { + read_ppid(ev.pid) + } +} + +fn exec_process_context(ev: &CompletedExec, ppid: u32) -> ProcessContext { + let cwd = read_cwd(ev.pid); + let captured_exe = cstr(&ev.filename); + ProcessContext { + pid: ev.pid, + ppid, + comm: cstr(&ev.comm), + exe: read_exe(ev.pid).or_else(|| (!captured_exe.is_empty()).then_some(captured_exe)), + cwd: (!cwd.is_empty()).then_some(cwd), + cgroup: read_cgroup(ev.pid), + } +} + +struct ProcCmdline { + argv: Vec, + observed_bytes: u32, + truncated: bool, +} + +fn read_proc_cmdline_at( + proc_root: &Path, + pid: u32, + max_bytes: usize, +) -> std::io::Result { + let file = File::open(proc_root.join(pid.to_string()).join("cmdline"))?; + let mut raw = Vec::new(); + file.take(max_bytes.saturating_add(1) as u64) + .read_to_end(&mut raw)?; + let truncated = raw.len() > max_bytes; + if truncated { + raw.truncate(max_bytes); + } + let observed_bytes = raw.len().min(u32::MAX as usize) as u32; + let argv = raw + .split(|byte| *byte == 0) + .filter(|arg| !arg.is_empty()) + .map(|arg| String::from_utf8_lossy(arg).into_owned()) + .collect(); + Ok(ProcCmdline { + argv, + observed_bytes, + truncated, + }) +} + +fn same_executable(ev: &CompletedExec, proc_argv: &[String]) -> bool { + let Some(proc_argv0) = proc_argv.first().filter(|value| !value.is_empty()) else { + return false; + }; + if ev.argv.first().is_some_and(|value| value == proc_argv0) { + return true; + } + let captured = cstr(&ev.filename); + let captured_name = Path::new(&captured).file_name(); + let proc_name = Path::new(proc_argv0).file_name(); + captured_name.is_some() && captured_name == proc_name +} + +fn supplement_exec_argv_at( + mut ev: CompletedExec, + proc_root: &Path, + max_bytes: usize, +) -> (CompletedExec, &'static str, u32, u32) { + let should_supplement = ev.exec_confirmed && (ev.argv_truncated || ev.argv_incomplete); + if should_supplement { + if let Ok(cmdline) = read_proc_cmdline_at(proc_root, ev.pid, max_bytes) { + if !cmdline.argv.is_empty() && same_executable(&ev, &cmdline.argv) { + ev.argv = cmdline.argv; + ev.argv_truncated = cmdline.truncated; + ev.argv_incomplete = false; + let argc = ev.argv.len().min(u32::MAX as usize) as u32; + return (ev, "proc_cmdline", argc, cmdline.observed_bytes); + } + } + } + let argc = ev.argv.len().min(u32::MAX as usize) as u32; + let bytes = ev + .argv + .iter() + .fold(0usize, |total, arg| total.saturating_add(arg.len())) + .min(u32::MAX as usize) as u32; + (ev, "kernel_fragments", argc, bytes) +} + +fn supplement_exec_argv(ev: CompletedExec) -> (CompletedExec, &'static str, u32, u32) { + supplement_exec_argv_at(ev, Path::new("/proc"), PROC_CMDLINE_MAX_BYTES) +} + +fn emit_completed_exec( + exporter: &dyn Exporter, + stats: &mut Stats, + resolver: &impl IdentityResolver, + ev: CompletedExec, +) { + if ev.reassembly_timed_out { + stats.exec_reassembly_timeout += 1; + } + let (ev, argv_source, observed_argc, observed_bytes) = supplement_exec_argv(ev); + let ppid = exec_ppid(&ev); + let cwd = read_cwd(ev.pid); + emit( + exporter, + stats, + EnrichedEvent { + identity: identity_for(resolver, ev.pid, &ev.comm), + workload: resolver.resolve_workload(ev.pid, 0, 0), + observation: None, + process: Some(exec_process_context(&ev, ppid)), + provider: None, + event: AgentEvent::ToolExec { + pid: ev.pid, + ppid, + uid: ev.uid, + argv: ev.argv, + argv_truncated: ev.argv_truncated, + argv_incomplete: ev.argv_incomplete, + exec_confirmed: ev.exec_confirmed, + argv_source: argv_source.to_string(), + captured_argc: ev.captured_argc, + captured_bytes: ev.captured_bytes, + observed_argc, + observed_bytes, + cwd, + }, + }, + ); +} + /// A NUL-terminated byte buffer (from a kernel copy) as a lossy String. fn cstr(buf: &[u8]) -> String { let end = buf.iter().position(|&b| b == 0).unwrap_or(buf.len()); @@ -608,6 +1009,9 @@ fn identity_for(r: &impl IdentityResolver, pid: u32, comm: &[u8; 16]) -> Identit #[derive(Default)] struct Stats { exec: u64, + exec_truncated: u64, + exec_incomplete: u64, + exec_reassembly_timeout: u64, exit: u64, egress: u64, dns: u64, @@ -718,6 +1122,9 @@ fn emit_collector_heartbeat( llm: stats.llm, ssl: stats.ssl, sec: stats.sec, + exec_truncated: stats.exec_truncated, + exec_incomplete: stats.exec_incomplete, + exec_reassembly_timeout: stats.exec_reassembly_timeout, dropped, output_dropped, }, @@ -727,7 +1134,15 @@ fn emit_collector_heartbeat( /// Export an event and count it by kind for the throughput report. fn emit(exporter: &dyn Exporter, stats: &mut Stats, ev: EnrichedEvent) { match &ev.event { - AgentEvent::ToolExec { .. } => stats.exec += 1, + AgentEvent::ToolExec { + argv_truncated, + argv_incomplete, + .. + } => { + stats.exec += 1; + stats.exec_truncated += u64::from(*argv_truncated); + stats.exec_incomplete += u64::from(*argv_incomplete); + } AgentEvent::ProcessExit { .. } => stats.exit += 1, AgentEvent::Egress { .. } => stats.egress += 1, AgentEvent::Dns { .. } => stats.dns += 1, @@ -843,7 +1258,210 @@ fn json_num_after(s: &str, key: &str) -> Option { #[cfg(test)] mod tests { - use super::{parse_dns_qname, parse_llm_meta, parse_sni}; + use super::{ + exec_ppid, exec_process_context, parse_dns_qname, parse_llm_meta, parse_sni, + supplement_exec_argv_at, CompletedExec, ExecAssembler, EXEC_REASSEMBLY_TIMEOUT, + }; + use a3s_observer_common::{ + ExecRecord, EXEC_ARG_CHUNK_PAYLOAD, EXEC_FLAG_ARGV_TRUNCATED, EXEC_RECORD_ARG_CHUNK, + EXEC_RECORD_COMMIT, EXEC_RECORD_END, EXEC_RECORD_HEADER, + }; + use std::fs; + use std::time::Instant; + + fn exec_record(exec_id: u64, kind: u8) -> ExecRecord { + let mut record: ExecRecord = unsafe { std::mem::zeroed() }; + record.exec_id = exec_id; + record.pid = u32::MAX; + record.ppid = 42; + record.uid = 1000; + record.kind = kind; + record + } + + fn chunk(exec_id: u64, arg_index: u16, chunk_index: u16, value: &[u8]) -> ExecRecord { + let mut record = exec_record(exec_id, EXEC_RECORD_ARG_CHUNK); + record.arg_index = arg_index; + record.chunk_index = chunk_index; + record.data_len = value.len() as u16; + record.data[..value.len()].copy_from_slice(value); + record + } + + #[test] + fn exec_uses_kernel_parent_snapshot_and_filename_fallback() { + let mut ev = CompletedExec { + pid: u32::MAX, + ppid: 42, + uid: 1000, + comm: [0; 16], + filename: [0; 128], + argv: vec!["bash".to_string()], + argv_truncated: false, + argv_incomplete: false, + captured_argc: 1, + captured_bytes: 4, + reassembly_timed_out: false, + exec_confirmed: true, + }; + let filename = b"/usr/bin/bash"; + ev.filename[..filename.len()].copy_from_slice(filename); + + assert_eq!(exec_ppid(&ev), 42); + let process = exec_process_context(&ev, exec_ppid(&ev)); + assert_eq!(process.ppid, 42); + assert_eq!(process.exe.as_deref(), Some("/usr/bin/bash")); + } + + #[test] + fn reassembles_long_arguments_without_silent_truncation() { + let now = Instant::now(); + let mut assembler = ExecAssembler::default(); + let mut header = exec_record(7, EXEC_RECORD_HEADER); + header.data_len = 9; + header.data[..9].copy_from_slice(b"/bin/bash"); + assert!(assembler.push(header, now).is_empty()); + assert!(assembler.push(chunk(7, 0, 0, b"bash"), now).is_empty()); + + let long = [b'x'; EXEC_ARG_CHUNK_PAYLOAD + 73]; + assert!(assembler + .push(chunk(7, 1, 0, &long[..EXEC_ARG_CHUNK_PAYLOAD]), now) + .is_empty()); + assert!(assembler + .push(chunk(7, 1, 1, &long[EXEC_ARG_CHUNK_PAYLOAD..]), now) + .is_empty()); + let mut end = exec_record(7, EXEC_RECORD_END); + end.argc = 2; + end.captured_bytes = (4 + long.len()) as u32; + assert!(assembler.push(end, now).is_empty()); + let completed = assembler + .push(exec_record(7, EXEC_RECORD_COMMIT), now) + .pop() + .unwrap(); + + assert_eq!(completed.argv[0], "bash"); + assert_eq!(completed.argv[1].len(), long.len()); + assert!(!completed.argv_truncated); + assert!(!completed.argv_incomplete); + } + + #[test] + fn marks_missing_chunks_and_timeouts_as_incomplete() { + let now = Instant::now(); + let mut assembler = ExecAssembler::default(); + let header = exec_record(8, EXEC_RECORD_HEADER); + assembler.push(header, now); + assembler.push(chunk(8, 0, 1, b"tail"), now); + let mut end = exec_record(8, EXEC_RECORD_END); + end.argc = 1; + end.captured_bytes = 4; + assembler.push(end, now); + let missing = assembler + .push(exec_record(8, EXEC_RECORD_COMMIT), now) + .pop() + .unwrap(); + assert!(missing.argv_incomplete); + + assembler.push(exec_record(9, EXEC_RECORD_HEADER), now); + let timed_out = assembler.expire(now + EXEC_REASSEMBLY_TIMEOUT); + assert_eq!(timed_out.len(), 1); + assert!(timed_out[0].argv_incomplete); + assert!(timed_out[0].reassembly_timed_out); + } + + #[test] + fn emits_without_waiting_when_exec_commit_probe_is_unavailable() { + let now = Instant::now(); + let mut assembler = ExecAssembler::new(false); + assembler.push(exec_record(11, EXEC_RECORD_HEADER), now); + assembler.push(chunk(11, 0, 0, b"echo"), now); + let mut end = exec_record(11, EXEC_RECORD_END); + end.argc = 1; + end.captured_bytes = 4; + + let completed = assembler.push(end, now).pop().unwrap(); + assert_eq!(completed.argv, ["echo"]); + assert!(!completed.exec_confirmed); + assert!(!completed.argv_incomplete); + assert!(!completed.reassembly_timed_out); + } + + #[test] + fn failed_exec_is_unconfirmed_but_not_a_reassembly_timeout() { + let now = Instant::now(); + let mut assembler = ExecAssembler::default(); + assembler.push(exec_record(12, EXEC_RECORD_HEADER), now); + assembler.push(chunk(12, 0, 0, b"missing-command"), now); + let mut end = exec_record(12, EXEC_RECORD_END); + end.argc = 1; + end.captured_bytes = 15; + assembler.push(end, now); + + let completed = assembler + .expire(now + EXEC_REASSEMBLY_TIMEOUT) + .pop() + .unwrap(); + assert!(!completed.exec_confirmed); + assert!(completed.argv_incomplete); + assert!(!completed.reassembly_timed_out); + } + + #[test] + fn preserves_explicit_kernel_truncation() { + let now = Instant::now(); + let mut assembler = ExecAssembler::default(); + assembler.push(exec_record(10, EXEC_RECORD_HEADER), now); + assembler.push(chunk(10, 0, 0, &[b'x'; EXEC_ARG_CHUNK_PAYLOAD]), now); + let mut end = exec_record(10, EXEC_RECORD_END); + end.argc = 1; + end.captured_bytes = EXEC_ARG_CHUNK_PAYLOAD as u32; + end.flags = EXEC_FLAG_ARGV_TRUNCATED; + assembler.push(end, now); + let completed = assembler + .push(exec_record(10, EXEC_RECORD_COMMIT), now) + .pop() + .unwrap(); + assert!(completed.argv_truncated); + assert!(!completed.argv_incomplete); + } + + #[test] + fn supplements_truncated_argv_from_matching_proc_cmdline() { + let pid = std::process::id(); + let root = std::env::temp_dir().join(format!("observer-proc-test-{pid}")); + let proc_dir = root.join(pid.to_string()); + fs::create_dir_all(&proc_dir).unwrap(); + fs::write( + proc_dir.join("cmdline"), + b"/usr/bin/bash\0-c\0echo complete-dangerous-tail\0", + ) + .unwrap(); + let mut filename = [0; 128]; + filename[..13].copy_from_slice(b"/usr/bin/bash"); + let event = CompletedExec { + pid, + ppid: 1, + uid: 1000, + comm: [0; 16], + filename, + argv: vec!["/usr/bin/bash".into(), "-c".into(), "echo complete".into()], + argv_truncated: true, + argv_incomplete: false, + captured_argc: 3, + captured_bytes: 32, + reassembly_timed_out: false, + exec_confirmed: true, + }; + + let (event, source, argc, bytes) = supplement_exec_argv_at(event, &root, 4096); + assert_eq!(source, "proc_cmdline"); + assert_eq!(event.argv[2], "echo complete-dangerous-tail"); + assert!(!event.argv_truncated); + assert!(!event.argv_incomplete); + assert_eq!(argc, 3); + assert!(bytes > event.captured_bytes); + fs::remove_dir_all(root).unwrap(); + } #[test] fn parses_sni_from_minimal_clienthello() { diff --git a/a3s-observer-common/src/lib.rs b/a3s-observer-common/src/lib.rs index 512314b..e88bdc2 100644 --- a/a3s-observer-common/src/lib.rs +++ b/a3s-observer-common/src/lib.rs @@ -6,25 +6,47 @@ /// A process / tool execution, captured at `sys_enter_execve`. /// -/// This is the language-agnostic "tool ran" signal — AgentSight's flagship is catching -/// subprocesses that bypass app-level instrumentation, and this is how we see them. -/// Up to `ARGV_SLOTS` arguments, each truncated to `ARG_LEN` bytes, are captured per exec — -/// the args carry the agent's intent (`curl `, `sh -c ""`), not just the binary. +/// Exec payloads are emitted as one header, zero or more argument chunks, and one end record. +/// Keeping each ring-buffer record small avoids the verifier/runtime failure caused by embedding +/// every argument in one large event while still allowing long shell commands to be reconstructed. pub const ARGV_SLOTS: usize = 12; -pub const ARG_LEN: usize = 128; +pub const EXEC_ARG_CHUNK_LEN: usize = 128; +/// `bpf_probe_read_user_str` reserves one byte for NUL in every chunk. +pub const EXEC_ARG_CHUNK_PAYLOAD: usize = EXEC_ARG_CHUNK_LEN - 1; +pub const EXEC_MAX_CHUNKS: usize = 64; +pub const EXEC_MAX_ARGV_BYTES: usize = EXEC_ARG_CHUNK_PAYLOAD * EXEC_MAX_CHUNKS; + +pub const EXEC_RECORD_HEADER: u8 = 1; +pub const EXEC_RECORD_ARG_CHUNK: u8 = 2; +pub const EXEC_RECORD_END: u8 = 3; +/// Emitted from `sched_process_exec` after the kernel successfully commits the exec. +pub const EXEC_RECORD_COMMIT: u8 = 4; + +pub const EXEC_FLAG_ARGV_TRUNCATED: u8 = 1 << 0; +pub const EXEC_FLAG_ARGV_INCOMPLETE: u8 = 1 << 1; #[repr(C)] #[derive(Clone, Copy)] -pub struct ExecEvent { +pub struct ExecRecord { + pub exec_id: u64, pub pid: u32, pub ppid: u32, pub uid: u32, - pub argc: u32, // number of args captured into `args` (may be < the real argc) + pub captured_bytes: u32, + pub argc: u16, + pub arg_index: u16, + pub chunk_index: u16, + pub data_len: u16, + pub kind: u8, + pub flags: u8, + pub _pad: [u8; 2], pub comm: [u8; 16], - pub filename: [u8; 128], - pub args: [[u8; ARG_LEN]; ARGV_SLOTS], // argv[0..argc], each NUL-terminated + /// Header: executable filename. Chunk: argument bytes. End: unused. + pub data: [u8; EXEC_ARG_CHUNK_LEN], } +const _: [(); 184] = [(); core::mem::size_of::()]; + /// A process exit (`sys_enter_exit_group`) — the other end of the tool lifecycle, carrying the /// exit status so tool *outcomes* are visible (did the command succeed?), not just that it ran. /// Captured via a `do_exit` kprobe, so it catches EVERY exit — clean exits and signal-kills diff --git a/a3s-observer-ebpf/src/main.rs b/a3s-observer-ebpf/src/main.rs index a9e25b6..8cd5715 100644 --- a/a3s-observer-ebpf/src/main.rs +++ b/a3s-observer-ebpf/src/main.rs @@ -2,23 +2,27 @@ #![no_main] use a3s_observer_common::{ - ConnectEvent, DnsEvent, ExecEvent, ExitEvent, FileEvent, LlmEvent, SecEvent, SslEvent, - TlsEvent, ARGV_SLOTS, ARG_LEN, DNS_SNAP_LEN, FILE_DELETE_FLAG, PATH_SNAP_LEN, SEC_BIND, - SEC_PTRACE, SEC_SETUID, SSL_SNAP_LEN, TLS_SNAP_LEN, + ConnectEvent, DnsEvent, ExecRecord, ExitEvent, FileEvent, LlmEvent, SecEvent, SslEvent, + TlsEvent, ARGV_SLOTS, DNS_SNAP_LEN, EXEC_ARG_CHUNK_PAYLOAD, EXEC_FLAG_ARGV_INCOMPLETE, + EXEC_FLAG_ARGV_TRUNCATED, EXEC_MAX_CHUNKS, EXEC_RECORD_ARG_CHUNK, EXEC_RECORD_COMMIT, + EXEC_RECORD_END, EXEC_RECORD_HEADER, FILE_DELETE_FLAG, PATH_SNAP_LEN, SEC_BIND, SEC_PTRACE, + SEC_SETUID, SSL_SNAP_LEN, TLS_SNAP_LEN, }; use aya_ebpf::{ + cty::c_void, helpers::gen::bpf_probe_read_user, helpers::{ bpf_get_current_comm, bpf_get_current_pid_tgid, bpf_get_current_uid_gid, bpf_ktime_get_ns, - bpf_probe_read_user_buf, bpf_probe_read_user_str_bytes, + bpf_loop, bpf_probe_read_user_buf, bpf_probe_read_user_str_bytes, }, macros::{cgroup_sock_addr, kprobe, map, tracepoint, uprobe, uretprobe}, - maps::{ring_buf::RingBufEntry, HashMap, PerCpuArray, RingBuf}, + maps::{ring_buf::RingBufEntry, HashMap, LruHashMap, PerCpuArray, RingBuf}, programs::{ProbeContext, RetProbeContext, SockAddrContext, TracePointContext}, }; -// Exec events now carry full argv (~928 B each), so this ring is larger than the others to -// keep a process burst (a build spawning many subprocesses) from dropping events. +// Exec records are fixed at 184 B. Typical commands need one header, a few argument chunks and +// one end record; long argv values can use up to EXEC_MAX_CHUNKS records without inflating every +// short exec event. #[map] static EVENTS: RingBuf = RingBuf::with_byte_size(512 * 1024, 0); @@ -49,6 +53,16 @@ static SEC_EVENTS: RingBuf = RingBuf::with_byte_size(64 * 1024, 0); #[map] static DROPS: PerCpuArray = PerCpuArray::with_max_entries(1, 0); +// child tgid -> parent tgid, captured before the child runs (with syscall-exit fallback). Exec must carry +// this in-kernel snapshot because short-lived tools can exit before userspace reads /proc. +#[map] +static PARENTS: LruHashMap = LruHashMap::with_max_entries(65_536, 0); + +// tgid -> latest syscall-entry capture. `sched_process_exec` consumes it only after a successful +// exec, so userspace can distinguish committed images from failed execve attempts. +#[map] +static EXEC_IDS: LruHashMap = LruHashMap::with_max_entries(65_536, 0); + // Egress deny-list (dest IPv4, host byte order). Populated by userspace from an external // policy; the cgroup/connect4 guard denies connect() to any IP present here. Cgroup-scoped. #[map] @@ -110,7 +124,199 @@ fn read_user_u64(addr: *const u8) -> Option { } } -// ---- tool / subprocess exec (sys_enter_execve) ---- +unsafe fn init_exec_record( + record: *mut ExecRecord, + exec_id: u64, + pid: u32, + ppid: u32, + uid: u32, + comm: [u8; 16], +) { + (*record).exec_id = exec_id; + (*record).pid = pid; + (*record).ppid = ppid; + (*record).uid = uid; + (*record).captured_bytes = 0; + (*record).argc = 0; + (*record).arg_index = 0; + (*record).chunk_index = 0; + (*record).data_len = 0; + (*record).kind = 0; + (*record).flags = 0; + (*record)._pad = [0; 2]; + (*record).comm = comm; + zero_exec_data(record); +} + +#[inline(always)] +unsafe fn zero_exec_data(record: *mut ExecRecord) { + // Avoid lowering `[0; 128]` to a bounded memset loop. This function runs inside the + // 64-iteration argv chunk loop; nesting those two loops makes the kernel verifier explore + // more than one million instructions and reject the exec probe. Fixed stores keep the same + // fully-initialized ring-buffer contract without adding another verifier loop. + let data = core::ptr::addr_of_mut!((*record).data) as *mut u64; + core::ptr::write_unaligned(data.add(0), 0); + core::ptr::write_unaligned(data.add(1), 0); + core::ptr::write_unaligned(data.add(2), 0); + core::ptr::write_unaligned(data.add(3), 0); + core::ptr::write_unaligned(data.add(4), 0); + core::ptr::write_unaligned(data.add(5), 0); + core::ptr::write_unaligned(data.add(6), 0); + core::ptr::write_unaligned(data.add(7), 0); + core::ptr::write_unaligned(data.add(8), 0); + core::ptr::write_unaligned(data.add(9), 0); + core::ptr::write_unaligned(data.add(10), 0); + core::ptr::write_unaligned(data.add(11), 0); + core::ptr::write_unaligned(data.add(12), 0); + core::ptr::write_unaligned(data.add(13), 0); + core::ptr::write_unaligned(data.add(14), 0); + core::ptr::write_unaligned(data.add(15), 0); +} + +#[repr(C)] +struct ExecLoopContext { + argv: u64, + exec_id: u64, + argp: u64, + arg_offset: u32, + captured_bytes: u32, + pid: u32, + ppid: u32, + uid: u32, + arg_index: u16, + chunk_index: u16, + captured_argc: u16, + flags: u8, + done: u8, + comm: [u8; 16], +} + +unsafe extern "C" fn capture_exec_chunk(_iteration: u32, raw_ctx: *mut c_void) -> i64 { + let state = &mut *(raw_ctx as *mut ExecLoopContext); + if state.done != 0 { + return 1; + } + + if state.argp == 0 { + if state.arg_index as usize >= ARGV_SLOTS { + match read_user_u64((state.argv as *const u8).add(ARGV_SLOTS * 8)) { + Some(0) => {} + Some(_) => state.flags |= EXEC_FLAG_ARGV_TRUNCATED, + None => state.flags |= EXEC_FLAG_ARGV_INCOMPLETE, + } + state.done = 1; + return 1; + } + let Some(next_arg) = + read_user_u64((state.argv as *const u8).add(state.arg_index as usize * 8)) + else { + state.flags |= EXEC_FLAG_ARGV_INCOMPLETE; + state.done = 1; + return 1; + }; + if next_arg == 0 { + state.done = 1; + return 1; + } + state.argp = next_arg; + state.captured_argc += 1; + } + + let Some(mut chunk_entry) = reserve_or_drop::(&EVENTS) else { + state.flags |= EXEC_FLAG_ARGV_INCOMPLETE; + state.done = 1; + return 1; + }; + let chunk = chunk_entry.as_mut_ptr(); + init_exec_record( + chunk, + state.exec_id, + state.pid, + state.ppid, + state.uid, + state.comm, + ); + (*chunk).kind = EXEC_RECORD_ARG_CHUNK; + (*chunk).arg_index = state.arg_index; + (*chunk).chunk_index = state.chunk_index; + + let len = match bpf_probe_read_user_str_bytes( + (state.argp as *const u8).add(state.arg_offset as usize), + &mut (*chunk).data, + ) { + Ok(bytes) => bytes.len(), + Err(_) => { + chunk_entry.discard(0); + state.flags |= EXEC_FLAG_ARGV_INCOMPLETE; + state.done = 1; + return 1; + } + }; + (*chunk).data_len = len as u16; + chunk_entry.submit(0); + state.captured_bytes += len as u32; + + if len < EXEC_ARG_CHUNK_PAYLOAD { + state.arg_index += 1; + state.chunk_index = 0; + state.arg_offset = 0; + state.argp = 0; + } else { + state.chunk_index += 1; + state.arg_offset += len as u32; + } + 0 +} + +// ---- process ancestry + tool exec ---- + +#[tracepoint] +pub fn track_process_fork(ctx: TracePointContext) -> u32 { + // Linux 6.17 tracepoint payload uses dynamic comm strings: parent_pid at offset 12, child_pid at offset 20. + let Ok(parent) = (unsafe { ctx.read_at::(12) }) else { + return 0; + }; + let Ok(child) = (unsafe { ctx.read_at::(20) }) else { + return 0; + }; + if parent > 0 && child > 0 { + let _ = PARENTS.insert(&(child as u32), &(parent as u32), 0); + } + 0 +} + +#[tracepoint] +pub fn track_clone(ctx: TracePointContext) -> u32 { + track_child(&ctx) +} + +#[tracepoint] +pub fn track_clone3(ctx: TracePointContext) -> u32 { + track_child(&ctx) +} + +#[tracepoint] +pub fn track_fork(ctx: TracePointContext) -> u32 { + track_child(&ctx) +} + +#[tracepoint] +pub fn track_vfork(ctx: TracePointContext) -> u32 { + track_child(&ctx) +} + +fn track_child(ctx: &TracePointContext) -> u32 { + // sys_exit_clone/fork/vfork: positive return value is the child PID in the parent process. + let Ok(child) = (unsafe { ctx.read_at::(16) }) else { + return 0; + }; + if child <= 0 || child > u32::MAX as i64 { + return 0; + } + let parent = (bpf_get_current_pid_tgid() >> 32) as u32; + let _ = PARENTS.insert(&(child as u32), &parent, 0); + 0 +} #[tracepoint] pub fn exec(ctx: TracePointContext) -> u32 { @@ -118,42 +324,115 @@ pub fn exec(ctx: TracePointContext) -> u32 { } fn try_exec(ctx: &TracePointContext) -> Result { - let Some(mut entry) = reserve_or_drop::(&EVENTS) else { + let pid_tgid = bpf_get_current_pid_tgid(); + let pid = (pid_tgid >> 32) as u32; + let uid = bpf_get_current_uid_gid() as u32; + let ppid = unsafe { PARENTS.get(&pid).copied().unwrap_or(0) }; + let comm = bpf_get_current_comm().unwrap_or_default(); + let exec_id = unsafe { bpf_ktime_get_ns() } ^ pid_tgid; + let mut flags = 0u8; + let _ = EXEC_IDS.insert(&pid, &exec_id, 0); + + let Some(mut header_entry) = reserve_or_drop::(&EVENTS) else { return Ok(0); }; - let ev = entry.as_mut_ptr(); + let header = header_entry.as_mut_ptr(); unsafe { - let pid_tgid = bpf_get_current_pid_tgid(); - (*ev).pid = (pid_tgid >> 32) as u32; - (*ev).uid = bpf_get_current_uid_gid() as u32; - (*ev).ppid = 0; - (*ev).comm = bpf_get_current_comm().unwrap_or_default(); - (*ev).filename = [0u8; 128]; + init_exec_record(header, exec_id, pid, ppid, uid, comm); + (*header).kind = EXEC_RECORD_HEADER; // sys_enter_execve: `const char *filename` at offset 16. if let Ok(filename_ptr) = ctx.read_at::<*const u8>(16) { - let _ = bpf_probe_read_user_str_bytes(filename_ptr, &mut (*ev).filename); + match bpf_probe_read_user_str_bytes(filename_ptr, &mut (*header).data) { + Ok(bytes) => (*header).data_len = bytes.len() as u16, + Err(_) => flags |= EXEC_FLAG_ARGV_INCOMPLETE, + } + } else { + flags |= EXEC_FLAG_ARGV_INCOMPLETE; } - // `const char *const *argv` at offset 24 — capture up to ARGV_SLOTS args (the intent: - // which URL / which command, not just the binary). Bounded for the verifier. - (*ev).argc = 0; - (*ev).args = [[0u8; ARG_LEN]; ARGV_SLOTS]; + (*header).flags = flags; + } + header_entry.submit(0); + + let captured_argc: u16; + let captured_bytes: u32; + + unsafe { + // `const char *const *argv` at offset 24. bpf_loop verifies the callback once instead of + // exploring every state transition through a 64-iteration in-program loop. if let Ok(argv) = ctx.read_at::<*const u8>(24) { - for i in 0..ARGV_SLOTS { - let Some(argp) = read_user_u64(argv.add(i * 8)) else { - break; - }; - if argp == 0 { - break; // end of argv - } - let _ = bpf_probe_read_user_str_bytes(argp as *const u8, &mut (*ev).args[i]); - (*ev).argc += 1; + let mut loop_ctx = ExecLoopContext { + argv: argv as u64, + exec_id, + argp: 0, + arg_offset: 0, + captured_bytes: 0, + pid, + ppid, + uid, + arg_index: 0, + chunk_index: 0, + captured_argc: 0, + flags, + done: 0, + comm, + }; + let iterations = bpf_loop( + EXEC_MAX_CHUNKS as u32, + capture_exec_chunk as *mut c_void, + &mut loop_ctx as *mut ExecLoopContext as *mut c_void, + 0, + ); + if iterations < 0 { + loop_ctx.flags |= EXEC_FLAG_ARGV_INCOMPLETE; + } else if loop_ctx.done == 0 { + loop_ctx.flags |= EXEC_FLAG_ARGV_TRUNCATED; } + flags = loop_ctx.flags; + captured_argc = loop_ctx.captured_argc; + captured_bytes = loop_ctx.captured_bytes; + } else { + flags |= EXEC_FLAG_ARGV_INCOMPLETE; + captured_argc = 0; + captured_bytes = 0; } + + let Some(mut end_entry) = reserve_or_drop::(&EVENTS) else { + return Ok(0); + }; + let end = end_entry.as_mut_ptr(); + init_exec_record(end, exec_id, pid, ppid, uid, comm); + (*end).kind = EXEC_RECORD_END; + (*end).flags = flags; + (*end).argc = captured_argc; + (*end).captured_bytes = captured_bytes; + end_entry.submit(0); } - entry.submit(0); Ok(0) } +/// Successful exec commit. Userspace correlates this small record with the bounded syscall-entry +/// fragments and can then read `/proc//cmdline` while the committed image is still alive. +#[tracepoint] +pub fn track_process_exec(_ctx: TracePointContext) -> u32 { + let pid = (bpf_get_current_pid_tgid() >> 32) as u32; + let Some(exec_id) = (unsafe { EXEC_IDS.get(&pid).copied() }) else { + return 0; + }; + let uid = bpf_get_current_uid_gid() as u32; + let ppid = unsafe { PARENTS.get(&pid).copied().unwrap_or(0) }; + let comm = bpf_get_current_comm().unwrap_or_default(); + if let Some(mut entry) = reserve_or_drop::(&EVENTS) { + let record = entry.as_mut_ptr(); + unsafe { + init_exec_record(record, exec_id, pid, ppid, uid, comm); + (*record).kind = EXEC_RECORD_COMMIT; + } + entry.submit(0); + } + let _ = EXEC_IDS.remove(&pid); + 0 +} + // ---- process exit (do_exit kprobe) — the tool's outcome: exit code AND terminating signal ---- #[kprobe] @@ -183,6 +462,9 @@ fn try_proc_exit(ctx: &ProbeContext) -> Result { (*ev).signal = (code & 0x7f) as u32; // & 0x7f intentionally drops the 0x80 core-dump bit } entry.submit(0); + let pid = (id >> 32) as u32; + let _ = PARENTS.remove(&pid); + let _ = EXEC_IDS.remove(&pid); Ok(0) } diff --git a/scripts/smoke.sh b/scripts/smoke.sh index 959922e..3270922 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -4,10 +4,17 @@ # and checks an event flows. Exits non-zero on failure. CI can't run this (needs a real # kernel + root); it makes the manual validation reproducible. # -# sudo ./scripts/smoke.sh +# ./scripts/smoke.sh set -euo pipefail cd "$(dirname "$0")/.." +# Older instructions ran the whole script through sudo, which hides the invoking user's Rust +# toolchain and leaves root-owned build artifacts. Drop back to the original user; only the +# collector process below needs privileges to load eBPF programs. +if [ "$(id -u)" -eq 0 ] && [ -n "${SUDO_USER:-}" ]; then + exec sudo -u "$SUDO_USER" -H "$0" "$@" +fi + # rustup installs to ~/.cargo/bin; make this work in a non-login shell too. command -v cargo >/dev/null || { [ -f "$HOME/.cargo/env" ] && . "$HOME/.cargo/env"; } @@ -22,15 +29,98 @@ echo "== --version ==" "$bin" --version echo "== run + drive one LLM call (need root) ==" -log=$(mktemp) -RUST_LOG=info A3S_OBSERVER_JSON=1 timeout -s INT 8 "$bin" >"$log" 2>&1 & +log=/tmp/a3s-observer-smoke.log +: >"$log" +fail() { + echo "FAIL: $1" + tail -200 "$log" + echo "Full log: $log" + exit 1 +} +sudo -E env RUST_LOG=info A3S_OBSERVER_JSON=1 timeout -s INT 30 "$bin" >"$log" 2>&1 & pid=$! -sleep 3 +for _ in $(seq 1 25); do + grep -q "probes attached" "$log" && break + kill -0 "$pid" 2>/dev/null || break + sleep 1 +done +grep -q "probes attached" "$log" || { + wait "$pid" 2>/dev/null || true + fail "probes did not attach" +} +marker="observer-exec-v2-$$" +short_marker="$marker-short" +semantic_marker="$marker-semantic" +many_marker="$marker-many" +huge_marker="$marker-huge" +stress_marker="$marker-stress" +proc_tail_marker="$marker-proc-tail" + +# Short argv remains compatible with the old capture path. +/usr/bin/printf '%s\n' "$short_marker" >/dev/null + +# A normal long argument must be reconstructed losslessly. +long_arg=$(printf 'x%.0s' $(seq 1 600)) +/usr/bin/printf '%s\n' "$marker-$long_arg" >/dev/null + +# Put a security-relevant tail beyond the old 64-byte boundary without executing it. +semantic_prefix=$(printf 'A%.0s' $(seq 1 96)) +/usr/bin/printf '%s\n' "$semantic_marker-$semantic_prefix-base64 -d | sh" >/dev/null + +# Exceed the argument-count cap. The event must be emitted and explicitly marked truncated. +/usr/bin/printf '%s' "$many_marker" a02 a03 a04 a05 a06 a07 a08 a09 a10 a11 a12 a13 a14 >/dev/null + +# Exceed the total argv-byte cap. Again, emit an explicitly truncated event, never a silent cut. +printf -v huge_arg '%*s' 9000 '' +huge_arg=${huge_arg// /z} +/usr/bin/printf '%s' "$huge_marker-$huge_arg" >/dev/null + +# Keep a process alive long enough for the successful-exec commit to trigger `/proc` argv +# supplementation. The marker is beyond the 12-argument kernel cap, so it cannot appear unless the +# userspace supplement recovered the complete command line. +/usr/bin/python3 -c 'import time; time.sleep(1)' a03 a04 a05 a06 a07 a08 a09 a10 a11 a12 "$proc_tail_marker" + +# Exercise reassembly under a short burst of concurrent execs. +stress_pids=() +for i in $(seq 1 100); do + /usr/bin/printf '%s\n' "$stress_marker-$i" >/dev/null & + stress_pids+=("$!") +done +for child in "${stress_pids[@]}"; do + wait "$child" +done curl -s -o /dev/null --max-time 4 https://api.anthropic.com/ 2>/dev/null || true wait "$pid" 2>/dev/null || true echo "== checks ==" -grep -q "probes attached" "$log" || { echo "FAIL: probes did not attach"; cat "$log"; rm -f "$log"; exit 1; } -grep -aq '"event"' "$log" || { echo "FAIL: no events captured"; rm -f "$log"; exit 1; } -echo "PASS: probes attached and events flowed" -rm -f "$log" +# Tracing may force ANSI formatting even when stderr is redirected. Strip it before checking +# structured counter fields so color codes cannot create false failures. +plain_log="${log}.plain" +sed -E $'s/\\x1B\\[[0-9;]*[[:alpha:]]//g' "$log" >"$plain_log" +grep -q "probes attached" "$log" || fail "probes did not attach" +grep -aq '"event"' "$log" || fail "no events captured" +grep -aF "$short_marker" "$log" | grep -q '"argv_truncated":false,"argv_incomplete":false' || { + fail "short argv regressed" +} +grep -aFq "$marker-$long_arg" "$log" || fail "600-byte argv was missing or truncated" +grep -aF "$marker-$long_arg" "$log" | grep -q '"argv_truncated":false,"argv_incomplete":false' || { + fail "long argv was marked truncated or incomplete" +} +grep -aFq "$semantic_marker-$semantic_prefix-base64 -d | sh" "$log" || { + fail "content beyond the old 64-byte boundary was missing" +} +grep -aF "$many_marker" "$log" | grep -q '"argv_truncated":true,"argv_incomplete":false' || { + fail "argument-count overflow was not explicitly marked truncated" +} +grep -aF "$huge_marker" "$log" | grep -q '"argv_truncated":true,"argv_incomplete":false' || { + fail "argv-byte overflow was not explicitly marked truncated" +} +grep -aF "$proc_tail_marker" "$log" | grep -q '"argv_truncated":false,"argv_incomplete":false,"exec_confirmed":true,"argv_source":"proc_cmdline"' || { + fail "successful exec did not recover arguments beyond the kernel cap from /proc" +} +stress_count=$(grep -aFc "$stress_marker-" "$log" || true) +[ "$stress_count" -eq 100 ] || fail "concurrent exec capture mismatch: expected 100, got $stress_count" +grep -q 'exec_reassembly_timeout=0' "$plain_log" || fail "exec reassembly timed out" +grep -q 'dropped=0 output_dropped=0' "$plain_log" || fail "collector dropped events" +echo "PASS: argv boundaries, exec confirmation, /proc supplementation, explicit truncation, concurrent reassembly, and drop counters" +rm -f "$log" "$plain_log" diff --git a/src/model.rs b/src/model.rs index 9c84724..fa45784 100644 --- a/src/model.rs +++ b/src/model.rs @@ -31,6 +31,19 @@ pub enum AgentEvent { /// Real UID the tool runs as (0 = root) — surfaces privilege / privesc. uid: u32, argv: Vec, + /// True when the configured argument-count or total-byte budget was exceeded. + argv_truncated: bool, + /// True when one or more kernel chunks were missing or reassembly timed out. + argv_incomplete: bool, + /// True when `sched_process_exec` confirmed that the kernel committed this exec. + exec_confirmed: bool, + /// `kernel_fragments` or `proc_cmdline` when a successful exec was supplemented. + argv_source: String, + captured_argc: u16, + captured_bytes: u32, + /// Argument count and bytes in the final exported argv after best-effort supplementation. + observed_argc: u32, + observed_bytes: u32, cwd: String, }, /// A process exited (`do_exit` kprobe) — the tool's outcome: exit code AND terminating signal @@ -128,6 +141,9 @@ pub enum AgentEvent { llm: u64, ssl: u64, sec: u64, + exec_truncated: u64, + exec_incomplete: u64, + exec_reassembly_timeout: u64, dropped: u64, output_dropped: u64, },