feat(ssl): opt-in OpenSSL content capture (#7) — prompts/responses via uprobes

RoyLin · RoyLin · commit db5e355ef189 · 2026-06-24T07:08:03.000+08:00
A3S_OBSERVER_SSL=1 attaches uprobes to SSL_write / SSL_read and emits AgentEvent::SslContent
with request (prompt) / response (completion) plaintext. Deliberately outside the
language-agnostic core (a uprobe binds to OpenSSL symbols, captures real content, off by
default) — a Rule 2 extension. eBPF uprobes + SslEvent + AgentEvent::SslContent + opt-in
collector attach/drain. Build-validated on KVM; runtime validation pending a non-prod VM
(content capture is gated off the shared prod node).
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,18 @@
 
 All notable changes to a3s-observer will be documented in this file.
 
+## [Unreleased]
+
+### Added
+
+- **SSL/TLS content capture (#7)** — the long-deferred opt-in OpenSSL uprobe extension.
+  `A3S_OBSERVER_SSL=1` attaches uprobes to `SSL_write` / `SSL_read` and emits `SslContent`
+  events with the request (prompt) / response (completion) **plaintext**. This is deliberately
+  outside the universal core (Rule 2): a uprobe binds to a library symbol, so it is **not**
+  language-agnostic (OpenSSL only — Python `requests`/`httpx`, Node, curl …, not Go
+  `crypto/tls`), and it captures real content, so it is **off by default**. Build-validated;
+  runtime validation pending a non-prod VM (content capture is gated off the shared prod node).
+
 ## [0.4.0] — file-access intervention
 
 ### Added
diff --git a/a3s-observer-collector/src/main.rs b/a3s-observer-collector/src/main.rs
@@ -10,11 +10,13 @@ use a3s_observer::{
     read_ppid, AgentEvent, EnrichedEvent, Exporter, Identity, IdentityResolver, JsonExporter,
     KubeResolver, LogExporter, Provider, ServiceClassifier, SniClassifier,
 };
-use a3s_observer_common::{ConnectEvent, DnsEvent, ExecEvent, FileEvent, LlmEvent, TlsEvent};
+use a3s_observer_common::{
+    ConnectEvent, DnsEvent, ExecEvent, FileEvent, LlmEvent, SslEvent, TlsEvent,
+};
 use anyhow::Context as _;
 use aya::{
     maps::{PerCpuArray, RingBuf},
-    programs::TracePoint,
+    programs::{TracePoint, UProbe},
     Ebpf,
 };
 use std::collections::HashMap;
@@ -33,7 +35,9 @@ async fn main() -> anyhow::Result<()> {
                 "a3s-observer-collector {} — language-agnostic eBPF observability for AI agents\n\n\
                  Run as root / CAP_BPF+CAP_PERFMON (Linux). Configure via env:\n  \
                  A3S_OBSERVER_JSON=1    emit NDJSON (default: human-readable log)\n  \
-                 A3S_OBSERVER_FILES=1   also capture file writes (high-volume; off by default)",
+                 A3S_OBSERVER_FILES=1   also capture file writes (high-volume; off by default)\n  \
+                 A3S_OBSERVER_SSL=1     also capture OpenSSL plaintext — prompts/responses \
+                 (uprobe, OpenSSL-only, off by default; or set a libssl path)",
                 env!("CARGO_PKG_VERSION")
             );
             return Ok(());
@@ -83,6 +87,34 @@ async fn main() -> anyhow::Result<()> {
         anyhow::bail!("no eBPF probes could be attached");
     }
 
+    // Opt-in OpenSSL content capture (uprobes). Off by default — it captures real plaintext
+    // (prompts/completions) and binds to OpenSSL only. A3S_OBSERVER_SSL=1, or set it to a
+    // libssl path on distros where the default below is wrong.
+    if let Some(val) = std::env::var("A3S_OBSERVER_SSL")
+        .ok()
+        .filter(|v| !v.is_empty())
+    {
+        let lib = if val.contains('/') {
+            val
+        } else {
+            "/usr/lib/x86_64-linux-gnu/libssl.so.3".to_string()
+        };
+        let mut ssl_ok = 0;
+        for (prog, sym) in [
+            ("ssl_write", "SSL_write"),
+            ("ssl_read_enter", "SSL_read"),
+            ("ssl_read_exit", "SSL_read"),
+        ] {
+            match attach_uprobe(&mut ebpf, prog, sym, &lib) {
+                Ok(()) => ssl_ok += 1,
+                Err(e) => {
+                    tracing::warn!(probe = prog, lib = %lib, error = %e, "SSL uprobe failed to attach")
+                }
+            }
+        }
+        tracing::info!(lib = %lib, attached = ssl_ok, "A3S_OBSERVER_SSL: OpenSSL content capture (uprobes)");
+    }
+
     // A3S_OBSERVER_JSON=1 → NDJSON (pipe to vector/Loki/jq); otherwise human-readable log.
     let exporter: Box<dyn Exporter> = if std::env::var_os("A3S_OBSERVER_JSON").is_some() {
         Box::new(JsonExporter)
@@ -117,6 +149,11 @@ async fn main() -> anyhow::Result<()> {
         ebpf.take_map("LLM_EVENTS")
             .context("`LLM_EVENTS` missing")?,
     )?;
+    // Opt-in OpenSSL content ring; stays empty unless A3S_OBSERVER_SSL attached the uprobes.
+    let mut ssl_ring = RingBuf::try_from(
+        ebpf.take_map("SSL_EVENTS")
+            .context("`SSL_EVENTS` missing")?,
+    )?;
     // Cumulative count of events dropped because a ring was full (data-loss visibility).
     let drops: PerCpuArray<_, u64> =
         PerCpuArray::try_from(ebpf.take_map("DROPS").context("`DROPS` missing")?)?;
@@ -149,6 +186,7 @@ async fn main() -> anyhow::Result<()> {
                     dns = stats.dns,
                     file = stats.file,
                     llm = stats.llm,
+                    ssl = stats.ssl,
                     dropped,
                     "a3s-observer: events in the last 60s (dropped = cumulative ring-full)"
                 );
@@ -275,6 +313,23 @@ async fn main() -> anyhow::Result<()> {
                         }
                     }
                 }
+                while let Some(item) = ssl_ring.next() {
+                    if let Some(ev) = read_pod::<SslEvent>(&item) {
+                        let len = (ev.len as usize).min(ev.data.len());
+                        let content = String::from_utf8_lossy(&ev.data[..len]).into_owned();
+                        if !content.is_empty() {
+                            emit(exporter.as_ref(), &mut stats, EnrichedEvent {
+                                identity: identity_for(&resolver, ev.pid, &ev.comm),
+                                provider: None,
+                                event: AgentEvent::SslContent {
+                                    pid: ev.pid,
+                                    is_read: ev.is_read != 0,
+                                    content,
+                                },
+                            });
+                        }
+                    }
+                }
             }
         }
     }
@@ -284,6 +339,7 @@ async fn main() -> anyhow::Result<()> {
         dns = stats.dns,
         file = stats.file,
         llm = stats.llm,
+        ssl = stats.ssl,
         "a3s-observer-collector: stopped (final window)"
     );
     Ok(())
@@ -303,6 +359,17 @@ fn attach(ebpf: &mut Ebpf, prog: &str, category: &str, name: &str) -> anyhow::Re
     Ok(())
 }
 
+fn attach_uprobe(ebpf: &mut Ebpf, prog: &str, sym: &str, target: &str) -> anyhow::Result<()> {
+    let p: &mut UProbe = ebpf
+        .program_mut(prog)
+        .with_context(|| format!("`{prog}` program not found"))?
+        .try_into()?;
+    p.load()?;
+    p.attach(Some(sym), 0, target, None)
+        .with_context(|| format!("attach uprobe {sym} in {target}"))?;
+    Ok(())
+}
+
 fn read_pod<T: Copy>(item: &[u8]) -> Option<T> {
     (item.len() >= core::mem::size_of::<T>())
         .then(|| unsafe { core::ptr::read_unaligned(item.as_ptr() as *const T) })
@@ -339,6 +406,7 @@ struct Stats {
     dns: u64,
     file: u64,
     llm: u64,
+    ssl: u64,
 }
 
 /// Export an event and count it by kind for the throughput report.
@@ -349,6 +417,7 @@ fn emit(exporter: &dyn Exporter, stats: &mut Stats, ev: EnrichedEvent) {
         AgentEvent::Dns { .. } => stats.dns += 1,
         AgentEvent::FileAccess { .. } => stats.file += 1,
         AgentEvent::LlmCall { .. } => stats.llm += 1,
+        AgentEvent::SslContent { .. } => stats.ssl += 1,
     }
     exporter.export(&ev);
 }
diff --git a/a3s-observer-common/src/lib.rs b/a3s-observer-common/src/lib.rs
@@ -89,3 +89,23 @@ pub struct LlmEvent {
     pub ttft_ns: u64,    // ClientHello → first response byte; 0 = no response seen
     pub comm: [u8; 16],
 }
+
+/// A plaintext snapshot from a TLS connection, captured by **uprobes** on OpenSSL
+/// `SSL_write` / `SSL_read` — the OPT-IN content extension (LLM prompt / completion bodies).
+///
+/// Unlike every other probe this is **not** language-agnostic: a uprobe binds to a specific
+/// library symbol, so this covers **OpenSSL only** (Python `requests`/`httpx`, Node, curl, …),
+/// not Go's `crypto/tls` or BoringSSL. It also captures real request/response content, so it is
+/// **off by default** (`A3S_OBSERVER_SSL=1`) and must run where that's acceptable. This is why
+/// it lives outside the universal core — see Rule 2 (minimal core + external extensions).
+pub const SSL_SNAP_LEN: usize = 1024;
+
+#[repr(C)]
+#[derive(Clone, Copy)]
+pub struct SslEvent {
+    pub pid: u32,
+    pub is_read: u32, // 0 = SSL_write (request / prompt), 1 = SSL_read (response / completion)
+    pub len: u32,     // bytes captured into `data` (<= SSL_SNAP_LEN)
+    pub comm: [u8; 16],
+    pub data: [u8; SSL_SNAP_LEN],
+}
diff --git a/a3s-observer-ebpf/src/main.rs b/a3s-observer-ebpf/src/main.rs
@@ -2,18 +2,18 @@
 #![no_main]
 
 use a3s_observer_common::{
-    ConnectEvent, DnsEvent, ExecEvent, FileEvent, LlmEvent, TlsEvent, DNS_SNAP_LEN, PATH_SNAP_LEN,
-    TLS_SNAP_LEN,
+    ConnectEvent, DnsEvent, ExecEvent, FileEvent, LlmEvent, SslEvent, TlsEvent, DNS_SNAP_LEN,
+    PATH_SNAP_LEN, SSL_SNAP_LEN, TLS_SNAP_LEN,
 };
 use aya_ebpf::{
     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,
     },
-    macros::{cgroup_sock_addr, map, tracepoint},
+    macros::{cgroup_sock_addr, map, tracepoint, uprobe, uretprobe},
     maps::{ring_buf::RingBufEntry, HashMap, PerCpuArray, RingBuf},
-    programs::{SockAddrContext, TracePointContext},
+    programs::{ProbeContext, RetProbeContext, SockAddrContext, TracePointContext},
 };
 
 #[map]
@@ -53,6 +53,15 @@ static LLM_SOCKS: HashMap<u64, LlmStat> = HashMap::with_max_entries(4096, 0);
 #[map]
 static READ_FD: HashMap<u64, u32> = HashMap::with_max_entries(10240, 0);
 
+// Opt-in OpenSSL content (uprobe). Bigger ring — payloads are up to SSL_SNAP_LEN each.
+#[map]
+static SSL_EVENTS: RingBuf = RingBuf::with_byte_size(512 * 1024, 0);
+
+// SSL_read entry saves the caller's buffer ptr by tid so the uretprobe (which knows how many
+// bytes were decrypted into it) can snapshot the plaintext.
+#[map]
+static SSL_READ_BUF: HashMap<u64, u64> = HashMap::with_max_entries(10240, 0);
+
 #[repr(C)]
 #[derive(Clone, Copy)]
 struct LlmStat {
@@ -196,6 +205,76 @@ fn try_tls(ctx: &TracePointContext) -> Result<u32, i64> {
     Ok(0)
 }
 
+// ---- OPT-IN OpenSSL content (uprobes on SSL_write / SSL_read) ----
+//
+// NOT language-agnostic (a uprobe binds to OpenSSL symbols) and captures real plaintext, so the
+// collector only attaches these when A3S_OBSERVER_SSL=1. SSL_write(ssl, buf, num): the request
+// plaintext is in `buf` at entry. SSL_read(ssl, buf, num): `buf` is filled during the call, so
+// snapshot it at return, with the byte count from the return value.
+
+#[uprobe]
+pub fn ssl_write(ctx: ProbeContext) -> u32 {
+    // SSL_write(ssl, buf, num): read args as raw register values (pointers carried as u64).
+    let buf = ctx.arg::<u64>(1).unwrap_or(0);
+    let num = ctx.arg::<u64>(2).unwrap_or(0);
+    emit_ssl(buf as *const u8, num, 0)
+}
+
+#[uprobe]
+pub fn ssl_read_enter(ctx: ProbeContext) -> u32 {
+    let buf = ctx.arg::<u64>(1).unwrap_or(0);
+    if buf != 0 {
+        let tid = unsafe { bpf_get_current_pid_tgid() };
+        let _ = SSL_READ_BUF.insert(&tid, &buf, 0);
+    }
+    0
+}
+
+#[uretprobe]
+pub fn ssl_read_exit(ctx: RetProbeContext) -> u32 {
+    let tid = unsafe { bpf_get_current_pid_tgid() };
+    let ret = ctx.ret::<i32>().unwrap_or(0) as i64; // SSL_read return value = bytes decrypted
+    let buf = unsafe { SSL_READ_BUF.get(&tid) }.copied();
+    let _ = SSL_READ_BUF.remove(&tid);
+    if ret <= 0 {
+        return 0;
+    }
+    match buf {
+        Some(addr) => emit_ssl(addr as *const u8, ret as u64, 1),
+        None => 0,
+    }
+}
+
+fn emit_ssl(buf: *const u8, len: u64, is_read: u32) -> u32 {
+    if buf.is_null() || len == 0 {
+        return 0;
+    }
+    let Some(mut entry) = reserve_or_drop::<SslEvent>(&SSL_EVENTS) else {
+        return 0;
+    };
+    let ev = entry.as_mut_ptr();
+    unsafe {
+        (*ev).pid = (bpf_get_current_pid_tgid() >> 32) as u32;
+        (*ev).is_read = is_read;
+        (*ev).comm = bpf_get_current_comm().unwrap_or_default();
+        // n <= SSL_SNAP_LEN (data capacity) and n <= len (bytes actually written/read).
+        let n: u32 = if len > SSL_SNAP_LEN as u64 {
+            SSL_SNAP_LEN as u32
+        } else {
+            len as u32
+        };
+        (*ev).len = n;
+        (*ev).data = [0u8; SSL_SNAP_LEN];
+        let _ = bpf_probe_read_user(
+            (*ev).data.as_mut_ptr() as *mut core::ffi::c_void,
+            n,
+            buf as *const core::ffi::c_void,
+        );
+    }
+    entry.submit(0);
+    0
+}
+
 // ---- outbound connection peer (sys_enter_connect) ----
 
 #[tracepoint]
diff --git a/src/model.rs b/src/model.rs
@@ -44,6 +44,16 @@ pub enum AgentEvent {
     },
     /// A DNS query — a hostname the process resolved (`sys_enter_sendto` to :53).
     Dns { pid: u32, query: String },
+    /// Plaintext from a TLS connection, captured by the **opt-in** OpenSSL uprobe extension
+    /// (`A3S_OBSERVER_SSL=1`): the request (prompt) or response (completion) body. OpenSSL
+    /// only (not language-agnostic), off by default. `content` is a UTF-8-lossy snapshot,
+    /// truncated to the kernel snapshot length.
+    SslContent {
+        pid: u32,
+        /// true = response (`SSL_read`, completion); false = request (`SSL_write`, prompt).
+        is_read: bool,
+        content: String,
+    },
 }
 
 /// An [`AgentEvent`] tagged with the resolved [`Identity`] and, for LLM calls, the
