ibsr: record-incident — CF-style sampled packet capture - #3
Conversation
…capture Lands the load-bearing design choices before any code per project principle 1: per-CPU decrement counter, classic-pcap microsecond format, filesystem-permission-gated trigger socket, shadow-mode safety carryover, encryption-at-rest as runbook responsibility.
TC ingress/egress program with per-CPU sampling counter and a 4-entry config_map (CFG_SAMPLE_RATE / CFG_SAMPLING_ACTIVE / CFG_INCIDENT_TAG_HASH / CFG_TRIGGER_TIMESTAMP). Snaplen-256 events emitted to a ringbuf using the same bucketed constant-size load pattern proven on tc_payload. Userspace side: RawPacketEvent + decoder, libbpf-rs adapter with attach_with_config + set_config / get_config for runtime mutation, fnv1a64 helper for tag hashing, ConfigKey enum. Safety profile re-uses ShadowPayload — TC + ringbuf, no drops, no redirect, ringbuf pressure cannot backpressure the network stack. Always TC_ACT_OK.
Phase 1 PcapWriter<W>: classic-pcap microsecond format with the
exact byte layout tcpdump -r expects (linktype Ethernet, snaplen
256). Boot-anchor helper translates kernel ktime_get_ns into
pcap (sec, usec).
Phase 4 PacketSink trait + SimplePacketSink (no-rotate, used by
tests + static-rate mode) + RotatingPcapSink (closes + reopens at
{out-dir}/{tag}-{ts}/packets.pcap on each trigger). Writer factory
is injectable so tests pin the rotation layout against an
in-memory registry.
Newline-delimited JSON protocol: set-sample-rate / trigger / stop / status. Pure parse_command + apply_command pieces unit- tested in isolation; channel-based listener thread sends commands to the orchestrator and awaits responses. apply_command writes config_map slots in the order rate → tag-hash → trigger-ts → active so the BPF program never reads 'active=1, rate=stale'. Status command is read-only (no kernel write). check_auto_stop fires Stop semantics at deadline. Trigger socket bound mode 0660 per design decision #4 — non- group-members cannot connect. Pins the invariant in trigger_socket_server_binds_with_0660_perms.
Pure apply_scrub(pkt, cfg) returns Pass(bytes) or Drop. IPv4 src/dst at offsets 26..34 are hashed with FNV-1a-64 keyed on a per-customer salt; packets where BOTH endpoints lie within a configured internal subnet are dropped. Subnet check runs BEFORE hashing — if both endpoints were in the internal range, hashing would push them out and we'd never drop real internal traffic. Pinned in scrub_subnet_check_runs_before_hashing. Limitations documented in the module header: IPv4 only, IP/TCP checksums not recomputed (tcpdump shows 'incorrect' but parses), hash is FNV-keyed not cryptographic.
Stateless sweeper that scans out_dir for *.pcap older than
min_age_sec, gzips them into archive_dir, and removes the source
only after the gzip completes (partial-archive failures leave
the source intact for the next sweep).
archive_target_path preserves the {tag-ts}/packets.pcap relative
layout under archive_dir. archive_pass recurses one level into
the phase-4 partition directories. Designed for the
orchestrator to call on a periodic tick.
Activates record-incident as the third top-level subcommand
alongside collect / collect-payload. Wires:
- RecordIncidentArgs with --tag, --sample-rate, --duration-sec,
--trigger-socket, --scrub-ip-salt, --scrub-internal-subnet,
--max-pcap-bytes, --archive-dir, --archive-after-sec.
- is_valid_incident_tag charset gate ([a-zA-Z0-9_-], 1..=64)
so trigger-socket inputs can't sneak path components into
the partitioned output dir.
- record_incident_loop: ringbuf pump → decode → scrub →
rotation-aware sink. Drains trigger-socket requests via
channel, applies via process_request, rotates pcap on
Trigger commands. Phase 6 size-driven rotation +
archive-sweeper tick. status.jsonl heartbeat per
--status-interval-sec via FsStatusEmitter.
- LibbpfRecordIncidentSource adapter implements
PacketEventSource + ConfigMutator over the libbpf collector.
- main.rs run_record_incident: opens initial pcap (Simple or
Rotating sink based on --trigger-socket), spawns
TriggerSocketServer, computes boot anchor, runs the loop.
Cargo deps: nix (clock_gettime CLOCK_MONOTONIC for boot anchor)
+ flate2 (gzip warm-tier archives).
Test surface: 416 ibsr-collector tests covering the full path
from CLI parse → orchestrator → loop → sink, all backed by
trait-abstracted I/O.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements the record-incident operating mode, enabling sampled packet capture via BPF with pcap output and a Unix socket for runtime control. The review identifies critical performance and security concerns, specifically the blocking of the main recording loop by synchronous archiving and the risk of memory exhaustion from unbounded socket reads. Additionally, the feedback highlights issues with resource leaks during initialization, excessive memory allocations in the scrubbing path, and vulnerabilities related to socket permissions and ownership.
| if let Some(archive_cfg) = &retention.archive { | ||
| let now = clock.now_unix_sec(); | ||
| if now.saturating_sub(last_archive_sec) >= archive_cfg.sweep_interval_sec { | ||
| let pass = crate::archive::archive_pass( | ||
| &archive_cfg.out_dir, | ||
| &archive_cfg.archive_dir, | ||
| archive_cfg.min_age_sec, | ||
| now, | ||
| ); | ||
| stats.archived += pass.archived; | ||
| stats.archive_errors += pass.errors; | ||
| if pass.archived > 0 || pass.errors > 0 { | ||
| logger.info(&format!( | ||
| "archive sweep: {} files moved to {}, {} errors", | ||
| pass.archived, | ||
| archive_cfg.archive_dir.display(), | ||
| pass.errors, | ||
| )); | ||
| } | ||
| last_archive_sec = now; | ||
| } | ||
| } |
There was a problem hiding this comment.
The archive sweep, which includes synchronous directory traversal and Gzip compression, is performed directly within the main recording loop. This blocks the draining of the BPF ringbuf. During periods of high traffic, this blocking operation will likely cause the ringbuf to overflow, leading to packet loss in the capture. Archiving should be moved to a background thread or triggered via a channel to a dedicated worker.
| { | ||
| let mut line = String::new(); | ||
| let mut buf = reader; | ||
| buf.read_line(&mut line)?; |
There was a problem hiding this comment.
The use of read_line on the control socket is unbounded. A malicious local client could send an extremely long stream of data without a newline character, causing the process to consume excessive memory and potentially leading to an OOM crash. Since this process runs as root, this represents a significant local Denial of Service vector. Please implement a limit on the maximum command length.
| iface: &str, | ||
| sample_rate: u64, | ||
| sampling_active: bool, | ||
| incident_tag: &str, | ||
| trigger_timestamp_unix_sec: u64, | ||
| resolver: &dyn InterfaceResolver, | ||
| ) -> Result<Self, TcPayloadLoaderError> { | ||
| let ifindex = resolver.ifindex(iface)?; | ||
|
|
||
| let open_object: &'static mut MaybeUninit<OpenObject> = | ||
| Box::leak(Box::new(MaybeUninit::<OpenObject>::uninit())); | ||
|
|
||
| let skel_builder = RecordIncidentSkelBuilder::default(); | ||
| let open_skel = skel_builder | ||
| .open(open_object) | ||
| .map_err(|e| TcPayloadLoaderError::BpfLoad(e.to_string()))?; | ||
|
|
||
| let skel = open_skel | ||
| .load() | ||
| .map_err(|e| TcPayloadLoaderError::BpfLoad(e.to_string()))?; | ||
|
|
||
| // Initialise the per-CPU sample counter — one slot, n_cpus | ||
| // values, each set to (rate - 1) so the rate-th packet on each | ||
| // CPU is the first sample. | ||
| let n_cpus = libbpf_rs::num_possible_cpus() | ||
| .map_err(|e| TcPayloadLoaderError::MapProgram(format!("num_possible_cpus: {}", e)))?; | ||
| let (counter_key, counter_values) = build_sample_counter_init(sample_rate, n_cpus); | ||
| skel.maps | ||
| .sample_counter | ||
| .update_percpu(&counter_key, &counter_values, libbpf_rs::MapFlags::ANY) | ||
| .map_err(|e| { | ||
| TcPayloadLoaderError::MapProgram(format!("sample_counter init: {}", e)) | ||
| })?; | ||
|
|
||
| // Initialise the 4-entry config_map. | ||
| let cfg_entries = build_config_map_entries( | ||
| sample_rate, | ||
| sampling_active, | ||
| incident_tag, | ||
| trigger_timestamp_unix_sec, | ||
| ); | ||
| for (key, value) in &cfg_entries { | ||
| skel.maps | ||
| .config_map | ||
| .update(key, value, libbpf_rs::MapFlags::ANY) | ||
| .map_err(|e| { | ||
| TcPayloadLoaderError::MapProgram(format!("config_map update: {}", e)) | ||
| })?; | ||
| } | ||
|
|
||
| // Create clsact qdisc on the interface. | ||
| let ingress_fd = skel.progs.tc_record_ingress.as_fd(); | ||
| let mut qdisc_builder = TcHookBuilder::new(ingress_fd); | ||
| qdisc_builder.ifindex(ifindex as i32).replace(true); | ||
| let qdisc = qdisc_builder | ||
| .hook(TC_INGRESS | TC_EGRESS) | ||
| .create() | ||
| .map_err(|e| TcPayloadLoaderError::Qdisc { | ||
| iface: iface.to_string(), | ||
| reason: e.to_string(), | ||
| })?; | ||
|
|
||
| // Attach ingress filter. | ||
| let mut ingress_builder = TcHookBuilder::new(ingress_fd); | ||
| ingress_builder | ||
| .ifindex(ifindex as i32) | ||
| .replace(true) | ||
| .handle(1) | ||
| .priority(1); | ||
| let mut ingress_hook = ingress_builder.hook(TC_INGRESS); | ||
| let ingress_hook = ingress_hook | ||
| .attach() | ||
| .map_err(|e| TcPayloadLoaderError::Attach { | ||
| direction: "ingress", | ||
| reason: e.to_string(), | ||
| })?; | ||
|
|
||
| // Attach egress filter. | ||
| let egress_fd = skel.progs.tc_record_egress.as_fd(); | ||
| let mut egress_builder = TcHookBuilder::new(egress_fd); | ||
| egress_builder | ||
| .ifindex(ifindex as i32) | ||
| .replace(true) | ||
| .handle(1) | ||
| .priority(1); | ||
| let mut egress_hook = egress_builder.hook(TC_EGRESS); | ||
| let egress_hook = egress_hook | ||
| .attach() | ||
| .map_err(|e| TcPayloadLoaderError::Attach { | ||
| direction: "egress", | ||
| reason: e.to_string(), | ||
| })?; | ||
|
|
||
| // Set up the ringbuf consumer. | ||
| let pending = PendingEvents::new(); | ||
| let pending_for_callback = pending.shared(); | ||
| let mut rb_builder = RingBufferBuilder::new(); | ||
| rb_builder | ||
| .add(&skel.maps.packet_rb, move |bytes: &[u8]| { | ||
| let mut guard = pending_for_callback | ||
| .lock() | ||
| .expect("ringbuf callback: pending mutex poisoned"); | ||
| guard.push(bytes.to_vec()); | ||
| 0 | ||
| }) | ||
| .map_err(|e| TcPayloadLoaderError::Ringbuf(e.to_string()))?; | ||
| let ringbuf = rb_builder | ||
| .build() | ||
| .map_err(|e| TcPayloadLoaderError::Ringbuf(e.to_string()))?; | ||
|
|
||
| Ok(Self { | ||
| ringbuf, | ||
| pending, | ||
| ingress_hook: Some(ingress_hook), | ||
| egress_hook: Some(egress_hook), | ||
| qdisc: Some(qdisc), | ||
| _skel: skel, | ||
| interface: iface.to_string(), | ||
| }) | ||
| } |
There was a problem hiding this comment.
The attach_with_config function lacks comprehensive cleanup logic for partial failures. If an error occurs after the qdisc or ingress_hook have been created/attached (e.g., if the egress hook attachment fails at line 345), these resources are leaked and will remain active in the kernel. The function should ensure that all successfully attached hooks and created qdiscs are detached/destroyed if the overall attachment process fails.
| /// else passes through unchanged. | ||
| /// | ||
| /// Pure function — no I/O. | ||
| pub fn apply_scrub(pkt: &[u8], cfg: &ScrubConfig) -> ScrubOutcome { |
There was a problem hiding this comment.
The apply_scrub function takes a slice and returns a Vec<u8>, which forces a new allocation for every packet that passes the scrubbing check, even if no modifications (like IP hashing) are required. Given that the caller already owns a Vec<u8> for the packet data, it would be much more efficient to pass the Vec by value, modify it in place if needed, and return it. This would significantly reduce allocation pressure in the hot path.
| // intended caller (API gateway / inference / operator CLI) | ||
| // into the owning group; without that, the socket is | ||
| // root-only. | ||
| std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o660))?; |
There was a problem hiding this comment.
While the code correctly sets the socket permissions to 0660, it does not set the group ownership to ibsr-trigger as specified in the design document (Decision #4). Without explicitly setting the group (e.g., using nix::unistd::chown), the socket will be owned by the process's primary group (likely root), which may prevent authorized non-root users in the ibsr-trigger group from accessing it.
| for entry in entries.flatten() { | ||
| let path = entry.path(); | ||
| if path.is_dir() { | ||
| // Recurse one level to handle phase-4 tag-partitioned | ||
| // sub-directories. We don't recurse arbitrarily deep — | ||
| // the layout is at most {out-dir}/{tag-ts}/packets.pcap. | ||
| if let Ok(inner) = std::fs::read_dir(&path) { | ||
| for sub in inner.flatten() { | ||
| let sub_path = sub.path(); | ||
| process_one(&sub_path, out_dir, archive_dir, min_age_sec, now_unix_sec, &mut result); | ||
| } | ||
| } | ||
| continue; | ||
| } | ||
| process_one(&path, out_dir, archive_dir, min_age_sec, now_unix_sec, &mut result); | ||
| } |
There was a problem hiding this comment.
The archive_pass function does not check if a directory it encounters is the archive_dir itself. If the archive directory is a sub-directory of the output directory, the sweeper will attempt to process its contents recursively, which is inefficient and could lead to unexpected behavior if .pcap files are present in the archive.
for entry in entries.flatten() {
let path = entry.path();
if path == archive_dir {
continue;
}
if path.is_dir() {
// Recurse one level to handle phase-4 tag-partitioned
// sub-directories. We don't recurse arbitrarily deep —
// the layout is at most {out-dir}/{tag-ts}/packets.pcap.
if let Ok(inner) = std::fs::read_dir(&path) {
for sub in inner.flatten() {
let sub_path = sub.path();
process_one(&sub_path, out_dir, archive_dir, min_age_sec, now_unix_sec, &mut result);
}
}
continue;
}
process_one(&path, out_dir, archive_dir, min_age_sec, now_unix_sec, &mut result);
}| let listener = UnixListener::bind(&path)?; | ||
| listener.set_nonblocking(true)?; | ||
|
|
||
| // Per docs/CF-INCIDENT-RECORDING-DESIGN-V1.md decision #4: | ||
| // socket access is gated by filesystem permissions. 0660 = | ||
| // owner + group RW, world none. The deployment must put the | ||
| // intended caller (API gateway / inference / operator CLI) | ||
| // into the owning group; without that, the socket is | ||
| // root-only. | ||
| std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o660))?; |
There was a problem hiding this comment.
There is a race condition between UnixListener::bind and fs::set_permissions. The socket is initially created with default permissions (governed by the process umask, typically resulting in 0755), allowing any local user to connect before the permissions are restricted to 0660. For a control socket, this window of vulnerability should be avoided, for example by setting the umask before binding or creating the socket in a restricted directory.
1 participant
Summary
ibsr record-incidentalongsidecollect/collect-payload. Implements PLAN-CF-INCIDENT-RECORDING-2026-05-09 end-to-end (Phases 0–6).--max-pcap-bytesrotation + warm-tier--archive-dirgzip sweeper.Phase walk
Test plan
Out-of-scope (documented)
🤖 Generated with Claude Code