From 2581487378898c873d4ab7484cd4e7303e901765 Mon Sep 17 00:00:00 2001 From: zhongyule Date: Tue, 7 Jul 2026 14:41:24 +0800 Subject: [PATCH] Emit process context with observer events --- a3s-observer-collector/src/main.rs | 39 +++++++++++++++++++++++++++++- src/lib.rs | 2 +- src/model.rs | 16 ++++++++++++ src/traits.rs | 1 + 4 files changed, 56 insertions(+), 2 deletions(-) diff --git a/a3s-observer-collector/src/main.rs b/a3s-observer-collector/src/main.rs index c7c7d0b..f380c54 100644 --- a/a3s-observer-collector/src/main.rs +++ b/a3s-observer-collector/src/main.rs @@ -8,7 +8,7 @@ use a3s_observer::{ read_ppid, AgentEvent, EnrichedEvent, Exporter, Identity, IdentityResolver, JsonExporter, - KubeResolver, LogExporter, Provider, ServiceClassifier, SniClassifier, + KubeResolver, LogExporter, ProcessContext, Provider, ServiceClassifier, SniClassifier, }; use a3s_observer_common::{ ConnectEvent, DnsEvent, ExecEvent, ExitEvent, FileEvent, LlmEvent, SecEvent, SslEvent, @@ -260,6 +260,7 @@ async fn main() -> anyhow::Result<()> { if let Some(ev) = read_pod::(&item) { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::ToolExec { pid: ev.pid, @@ -275,6 +276,7 @@ async fn main() -> anyhow::Result<()> { if let Some(ev) = read_pod::(&item) { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::ProcessExit { pid: ev.pid, @@ -294,6 +296,7 @@ async fn main() -> anyhow::Result<()> { }; emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::SecurityAction { pid: ev.pid, @@ -313,6 +316,7 @@ async fn main() -> anyhow::Result<()> { peers.insert(sock_key(ev.pid, ev.fd), (peer, ev.port)); emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::Egress { pid: ev.pid, @@ -343,6 +347,7 @@ async fn main() -> anyhow::Result<()> { llm_meta.insert(sock_key(ev.pid, ev.fd), (sni.clone(), provider.clone(), peer)); emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + process: Some(process_context(ev.pid, &ev.comm)), provider, event: AgentEvent::Egress { pid: ev.pid, @@ -360,6 +365,7 @@ async fn main() -> anyhow::Result<()> { if let Some(query) = parse_dns_qname(&ev.data[..len]) { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::Dns { pid: ev.pid, query }, }); @@ -382,6 +388,7 @@ async fn main() -> anyhow::Result<()> { }; emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + process: Some(process_context(ev.pid, &ev.comm)), provider: None, event, }); @@ -398,6 +405,7 @@ async fn main() -> anyhow::Result<()> { if provider.is_some() { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity_for(&resolver, ev.pid, &ev.comm), + process: Some(process_context(ev.pid, &ev.comm)), provider, event: AgentEvent::LlmCall { pid: ev.pid, @@ -426,6 +434,7 @@ async fn main() -> anyhow::Result<()> { { emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity: identity.clone(), + process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::LlmApi { pid: ev.pid, @@ -438,6 +447,7 @@ async fn main() -> anyhow::Result<()> { } emit(exporter.as_ref(), &mut stats, EnrichedEvent { identity, + process: Some(process_context(ev.pid, &ev.comm)), provider: None, event: AgentEvent::SslContent { pid: ev.pid, @@ -528,6 +538,32 @@ fn read_cwd(pid: u32) -> String { .unwrap_or_default() } +fn read_exe(pid: u32) -> Option { + std::fs::read_link(format!("/proc/{pid}/exe")) + .ok() + .map(|p| p.to_string_lossy().into_owned()) + .filter(|s| !s.is_empty()) +} + +fn read_cgroup(pid: u32) -> Option { + std::fs::read_to_string(format!("/proc/{pid}/cgroup")) + .ok() + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) +} + +fn process_context(pid: u32, comm: &[u8; 16]) -> ProcessContext { + let cwd = read_cwd(pid); + ProcessContext { + pid, + ppid: read_ppid(pid), + comm: cstr(comm), + exe: read_exe(pid), + cwd: (!cwd.is_empty()).then_some(cwd), + cgroup: read_cgroup(pid), + } +} + /// 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()); @@ -638,6 +674,7 @@ fn emit_collector_heartbeat( ) { exporter.export(&EnrichedEvent { identity: Identity::default(), + process: None, provider: None, event: AgentEvent::CollectorHeartbeat { collector_id: meta.collector_id.clone(), diff --git a/src/lib.rs b/src/lib.rs index fcd1341..9365be4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17,7 +17,7 @@ pub mod model; pub mod policy; pub mod traits; -pub use model::{AgentEvent, EnrichedEvent}; +pub use model::{AgentEvent, EnrichedEvent, ProcessContext}; pub use policy::{parse_egress_policy, AllowAll, Policy, ProviderPolicy, Verdict}; pub use traits::{ read_ppid, Exporter, Identity, IdentityResolver, JsonExporter, KubeResolver, LogExporter, diff --git a/src/model.rs b/src/model.rs index 718f746..ed31e89 100644 --- a/src/model.rs +++ b/src/model.rs @@ -6,6 +6,20 @@ use serde::Serialize; use std::net::IpAddr; use std::time::Duration; +/// Kernel-observed process context used by downstream attribution engines. +#[derive(Debug, Clone, Default, Serialize)] +pub struct ProcessContext { + pub pid: u32, + pub ppid: u32, + pub comm: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub exe: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cwd: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cgroup: Option, +} + /// A raw event captured by an eBPF probe, before identity enrichment. #[derive(Debug, Clone, Serialize)] pub enum AgentEvent { @@ -123,6 +137,8 @@ pub enum AgentEvent { #[derive(Debug, Clone, Serialize)] pub struct EnrichedEvent { pub identity: Identity, + #[serde(skip_serializing_if = "Option::is_none")] + pub process: Option, pub provider: Option, pub event: AgentEvent, } diff --git a/src/traits.rs b/src/traits.rs index 50fbc51..66e380c 100644 --- a/src/traits.rs +++ b/src/traits.rs @@ -346,6 +346,7 @@ mod tests { task: Some("1".into()), session: None, }, + process: None, provider: None, event: AgentEvent::ProcessExit { pid: 1,