Skip to content

Commit 2581487

Browse files
zhongyulezhongyule
zhongyule
authored and
zhongyule
committed
Emit process context with observer events
1 parent 4e2db38 commit 2581487

4 files changed

Lines changed: 56 additions & 2 deletions

File tree

‎a3s-observer-collector/src/main.rs‎

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
99
use a3s_observer::{
1010
read_ppid, AgentEvent, EnrichedEvent, Exporter, Identity, IdentityResolver, JsonExporter,
11-
KubeResolver, LogExporter, Provider, ServiceClassifier, SniClassifier,
11+
KubeResolver, LogExporter, ProcessContext, Provider, ServiceClassifier, SniClassifier,
1212
};
1313
use a3s_observer_common::{
1414
ConnectEvent, DnsEvent, ExecEvent, ExitEvent, FileEvent, LlmEvent, SecEvent, SslEvent,
@@ -260,6 +260,7 @@ async fn main() -> anyhow::Result<()> {
260260
if let Some(ev) = read_pod::<ExecEvent>(&item) {
261261
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
262262
identity: identity_for(&resolver, ev.pid, &ev.comm),
263+
process: Some(process_context(ev.pid, &ev.comm)),
263264
provider: None,
264265
event: AgentEvent::ToolExec {
265266
pid: ev.pid,
@@ -275,6 +276,7 @@ async fn main() -> anyhow::Result<()> {
275276
if let Some(ev) = read_pod::<ExitEvent>(&item) {
276277
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
277278
identity: identity_for(&resolver, ev.pid, &ev.comm),
279+
process: Some(process_context(ev.pid, &ev.comm)),
278280
provider: None,
279281
event: AgentEvent::ProcessExit {
280282
pid: ev.pid,
@@ -294,6 +296,7 @@ async fn main() -> anyhow::Result<()> {
294296
};
295297
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
296298
identity: identity_for(&resolver, ev.pid, &ev.comm),
299+
process: Some(process_context(ev.pid, &ev.comm)),
297300
provider: None,
298301
event: AgentEvent::SecurityAction {
299302
pid: ev.pid,
@@ -313,6 +316,7 @@ async fn main() -> anyhow::Result<()> {
313316
peers.insert(sock_key(ev.pid, ev.fd), (peer, ev.port));
314317
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
315318
identity: identity_for(&resolver, ev.pid, &ev.comm),
319+
process: Some(process_context(ev.pid, &ev.comm)),
316320
provider: None,
317321
event: AgentEvent::Egress {
318322
pid: ev.pid,
@@ -343,6 +347,7 @@ async fn main() -> anyhow::Result<()> {
343347
llm_meta.insert(sock_key(ev.pid, ev.fd), (sni.clone(), provider.clone(), peer));
344348
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
345349
identity: identity_for(&resolver, ev.pid, &ev.comm),
350+
process: Some(process_context(ev.pid, &ev.comm)),
346351
provider,
347352
event: AgentEvent::Egress {
348353
pid: ev.pid,
@@ -360,6 +365,7 @@ async fn main() -> anyhow::Result<()> {
360365
if let Some(query) = parse_dns_qname(&ev.data[..len]) {
361366
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
362367
identity: identity_for(&resolver, ev.pid, &ev.comm),
368+
process: Some(process_context(ev.pid, &ev.comm)),
363369
provider: None,
364370
event: AgentEvent::Dns { pid: ev.pid, query },
365371
});
@@ -382,6 +388,7 @@ async fn main() -> anyhow::Result<()> {
382388
};
383389
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
384390
identity: identity_for(&resolver, ev.pid, &ev.comm),
391+
process: Some(process_context(ev.pid, &ev.comm)),
385392
provider: None,
386393
event,
387394
});
@@ -398,6 +405,7 @@ async fn main() -> anyhow::Result<()> {
398405
if provider.is_some() {
399406
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
400407
identity: identity_for(&resolver, ev.pid, &ev.comm),
408+
process: Some(process_context(ev.pid, &ev.comm)),
401409
provider,
402410
event: AgentEvent::LlmCall {
403411
pid: ev.pid,
@@ -426,6 +434,7 @@ async fn main() -> anyhow::Result<()> {
426434
{
427435
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
428436
identity: identity.clone(),
437+
process: Some(process_context(ev.pid, &ev.comm)),
429438
provider: None,
430439
event: AgentEvent::LlmApi {
431440
pid: ev.pid,
@@ -438,6 +447,7 @@ async fn main() -> anyhow::Result<()> {
438447
}
439448
emit(exporter.as_ref(), &mut stats, EnrichedEvent {
440449
identity,
450+
process: Some(process_context(ev.pid, &ev.comm)),
441451
provider: None,
442452
event: AgentEvent::SslContent {
443453
pid: ev.pid,
@@ -528,6 +538,32 @@ fn read_cwd(pid: u32) -> String {
528538
.unwrap_or_default()
529539
}
530540

541+
fn read_exe(pid: u32) -> Option<String> {
542+
std::fs::read_link(format!("/proc/{pid}/exe"))
543+
.ok()
544+
.map(|p| p.to_string_lossy().into_owned())
545+
.filter(|s| !s.is_empty())
546+
}
547+
548+
fn read_cgroup(pid: u32) -> Option<String> {
549+
std::fs::read_to_string(format!("/proc/{pid}/cgroup"))
550+
.ok()
551+
.map(|s| s.trim().to_string())
552+
.filter(|s| !s.is_empty())
553+
}
554+
555+
fn process_context(pid: u32, comm: &[u8; 16]) -> ProcessContext {
556+
let cwd = read_cwd(pid);
557+
ProcessContext {
558+
pid,
559+
ppid: read_ppid(pid),
560+
comm: cstr(comm),
561+
exe: read_exe(pid),
562+
cwd: (!cwd.is_empty()).then_some(cwd),
563+
cgroup: read_cgroup(pid),
564+
}
565+
}
566+
531567
/// A NUL-terminated byte buffer (from a kernel copy) as a lossy String.
532568
fn cstr(buf: &[u8]) -> String {
533569
let end = buf.iter().position(|&b| b == 0).unwrap_or(buf.len());
@@ -638,6 +674,7 @@ fn emit_collector_heartbeat(
638674
) {
639675
exporter.export(&EnrichedEvent {
640676
identity: Identity::default(),
677+
process: None,
641678
provider: None,
642679
event: AgentEvent::CollectorHeartbeat {
643680
collector_id: meta.collector_id.clone(),

‎src/lib.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ pub mod model;
1717
pub mod policy;
1818
pub mod traits;
1919

20-
pub use model::{AgentEvent, EnrichedEvent};
20+
pub use model::{AgentEvent, EnrichedEvent, ProcessContext};
2121
pub use policy::{parse_egress_policy, AllowAll, Policy, ProviderPolicy, Verdict};
2222
pub use traits::{
2323
read_ppid, Exporter, Identity, IdentityResolver, JsonExporter, KubeResolver, LogExporter,

‎src/model.rs‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,20 @@ use serde::Serialize;
66
use std::net::IpAddr;
77
use std::time::Duration;
88

9+
/// Kernel-observed process context used by downstream attribution engines.
10+
#[derive(Debug, Clone, Default, Serialize)]
11+
pub struct ProcessContext {
12+
pub pid: u32,
13+
pub ppid: u32,
14+
pub comm: String,
15+
#[serde(skip_serializing_if = "Option::is_none")]
16+
pub exe: Option<String>,
17+
#[serde(skip_serializing_if = "Option::is_none")]
18+
pub cwd: Option<String>,
19+
#[serde(skip_serializing_if = "Option::is_none")]
20+
pub cgroup: Option<String>,
21+
}
22+
923
/// A raw event captured by an eBPF probe, before identity enrichment.
1024
#[derive(Debug, Clone, Serialize)]
1125
pub enum AgentEvent {
@@ -123,6 +137,8 @@ pub enum AgentEvent {
123137
#[derive(Debug, Clone, Serialize)]
124138
pub struct EnrichedEvent {
125139
pub identity: Identity,
140+
#[serde(skip_serializing_if = "Option::is_none")]
141+
pub process: Option<ProcessContext>,
126142
pub provider: Option<Provider>,
127143
pub event: AgentEvent,
128144
}

‎src/traits.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -346,6 +346,7 @@ mod tests {
346346
task: Some("1".into()),
347347
session: None,
348348
},
349+
process: None,
349350
provider: None,
350351
event: AgentEvent::ProcessExit {
351352
pid: 1,

0 commit comments

Comments
 (0)