diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 0000000..4738693 --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,21 @@ +# Flowprep + +Canonicalization of network telemetry into ML-ready NetFlow parquet. + +## Language + +**Active timeout**: +The maximum age of an open flow record before it is closed and a new record for the same key is started, even if packets are still arriving. +_Avoid_: max flow duration, max duration, flow lifetime + +**Inactive timeout**: +The idle gap after the last packet on a flow key before that open record is closed. +_Avoid_: idle timeout, idle gap, silence timeout + +**Flow key**: +The direction-normalized 5-tuple used to aggregate packets into one bidirectional flow record (src/dest IP and port ordered so both halves of a conversation share one key, plus protocol). +_Avoid_: connection, session, conversation (those may span multiple flow records after timeout splits) + +**Flow record**: +One closed aggregation for a flow key over a contiguous packet window bounded by active and inactive timeouts (or end of capture). +_Avoid_: flow (alone — ambiguous between key, record, and session), biflow diff --git a/Cargo.lock b/Cargo.lock index 70a5d99..c24435f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -601,7 +601,7 @@ dependencies = [ [[package]] name = "flowprep" -version = "0.3.0" +version = "0.4.0" dependencies = [ "arrow", "byteorder", diff --git a/Cargo.toml b/Cargo.toml index 1d77555..bcc982e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "flowprep" -version = "0.3.0" +version = "0.4.0" edition = "2024" rust-version = "1.85" description = "Convert network telemetry (pcap, flow CSVs, vendor exports) into ML-ready canonical NetFlow parquet" diff --git a/README.md b/README.md index 1c29708..ec48ec5 100644 --- a/README.md +++ b/README.md @@ -70,7 +70,9 @@ Five subcommands: ```bash # 1. Raw packet captures -> bidirectional flow records +# Defaults: --active-timeout 60 --inactive-timeout 15 (integer seconds) flowprep pcap capture.pcap flows.parquet +flowprep pcap capture.pcap flows.parquet --active-timeout 60 --inactive-timeout 15 # 2. Any aliased flow table (CSV, parquet, Zeek TSV log, Argus .binetflow) # -> the canonical schema @@ -159,10 +161,13 @@ flows and are reported as an error rather than a silent empty file.) Packets are grouped by a direction-normalized 5-tuple, so both halves of a conversation aggregate into a single flow record with separate -forward/backward byte and packet counters. Flows split on a 60s idle -timeout and a 1h maximum duration. The reader streams pcap and pcapng, -keeps constant memory on the packet path, and is robust to the -slightly-out-of-order packets real captures contain. +forward/backward byte and packet counters. Flows split on an **inactive +timeout** (default 15s idle gap) and an **active timeout** (default 60s max +age), matching common NetFlow exporter practice. Both are overridable as +integer seconds on `flowprep pcap` (`--active-timeout`, +`--inactive-timeout`; both must be `> 0` and inactive `<=` active). The +reader streams pcap and pcapng, keeps constant memory on the packet path, +and is robust to the slightly-out-of-order packets real captures contain. ### Zeek logs and research exports diff --git a/docs/adr/0001-pcap-active-inactive-timeouts.md b/docs/adr/0001-pcap-active-inactive-timeouts.md new file mode 100644 index 0000000..3d05960 --- /dev/null +++ b/docs/adr/0001-pcap-active-inactive-timeouts.md @@ -0,0 +1,26 @@ +# PCAP active/inactive timeout defaults (60s / 15s), CLI-overridable + +Status: accepted + +PCAP flow aggregation previously used a 60s idle split and a 1h max-duration +split, hardcoded and named unlike NetFlow exporter practice. We change the +defaults to an **active timeout** of 60s and an **inactive timeout** of 15s — +common exporter values — and expose both as integer-second CLI flags on +`flowprep pcap`, keeping strict `>` comparisons. This is an intentional +breaking change for default pcap→parquet cardinality (version bump to 0.4.0). + +## Considered options + +- **Keep 1h active / 60s inactive, hardcoded** — rejected; mismatches common + NetFlow exporter timeouts and cannot be tuned per capture. +- **New defaults only, still hardcoded** — rejected; operators still cannot + match a specific exporter profile without a rebuild. +- **Defaults 60s/15s + CLI overrides (chosen)** — matches exporter language and + lets callers align with their collector without forking flowprep. + +## Consequences + +- Long or chatty conversations produce more flow records under defaults. +- `inactive > active` and non-positive timeouts are rejected at CLI parse/validate time. +- Other subcommands (`canonicalize`, `ocsf`, `nfcapd`) are unchanged — they + ingest already-closed flow records. diff --git a/src/main.rs b/src/main.rs index bb6c57c..328f401 100644 --- a/src/main.rs +++ b/src/main.rs @@ -21,7 +21,16 @@ struct Cli { #[derive(Subcommand)] enum Command { /// pcap/pcapng -> canonical flow parquet - Pcap { input: String, output: String }, + Pcap { + input: String, + output: String, + /// Max age of an open flow record before it is closed (seconds). + #[arg(long, default_value_t = pcap::DEFAULT_ACTIVE_TIMEOUT_SECS)] + active_timeout: u64, + /// Idle gap after the last packet before a flow record is closed (seconds). + #[arg(long, default_value_t = pcap::DEFAULT_INACTIVE_TIMEOUT_SECS)] + inactive_timeout: u64, + }, /// aliased parquet/CSV flow table -> canonical parquet Canonicalize { input: String, output: String }, /// OCSF Network Activity JSON/NDJSON -> canonical parquet @@ -50,9 +59,16 @@ fn peek(input: &str, rows: usize) -> Result<(), Box> { fn main() { let cli = Cli::parse(); let result = match &cli.command { - Command::Pcap { input, output } => { - pcap::pcap_to_parquet(input, output).map(|n| println!("Wrote {n} flows to {output}")) - } + Command::Pcap { + input, + output, + active_timeout, + inactive_timeout, + } => match pcap::FlowTimeouts::from_secs(*active_timeout, *inactive_timeout) { + Ok(timeouts) => pcap::pcap_to_parquet(input, output, timeouts) + .map(|n| println!("Wrote {n} flows to {output}")), + Err(e) => Err(e), + }, Command::Canonicalize { input, output } => canonicalize::canonicalize_file(input, output) .map(|n| println!("Wrote {n} flows to {output}")), Command::Ocsf { input, output } => { diff --git a/src/pcap.rs b/src/pcap.rs index b42f318..3e5e997 100644 --- a/src/pcap.rs +++ b/src/pcap.rs @@ -4,7 +4,8 @@ //! bounded by active-flow count). Flows are bidirectional: keys are //! direction-normalized so both halves of a conversation aggregate into one //! record, with fwd_*/bwd_* counters split by which side matches the key. -//! Flows split on idle timeout (60s) and max duration (1h). +//! Flows split on inactive timeout (default 15s) and active timeout +//! (default 60s). See CONTEXT.md and docs/adr/0001-pcap-active-inactive-timeouts.md. use std::collections::HashMap; use std::fs::File; @@ -18,11 +19,40 @@ use crate::writer::write_parquet; type Result = std::result::Result>; -const IDLE_TIMEOUT_USEC: i64 = 60 * 1_000_000; -const MAX_FLOW_DURATION_USEC: i64 = 3600 * 1_000_000; +/// Default active timeout (seconds): max age of an open flow record. +pub const DEFAULT_ACTIVE_TIMEOUT_SECS: u64 = 60; +/// Default inactive timeout (seconds): idle gap before closing a flow record. +pub const DEFAULT_INACTIVE_TIMEOUT_SECS: u64 = 15; const LINKTYPE_ETHERNET: u16 = 1; +/// Active/inactive timeouts for PCAP flow aggregation (microseconds). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct FlowTimeouts { + pub active_usec: i64, + pub inactive_usec: i64, +} + +impl FlowTimeouts { + /// Build timeouts from integer seconds. Both must be `> 0` and + /// `inactive_secs <= active_secs`. + pub fn from_secs(active_secs: u64, inactive_secs: u64) -> Result { + if active_secs == 0 || inactive_secs == 0 { + return Err("active and inactive timeouts must be > 0 seconds".into()); + } + if inactive_secs > active_secs { + return Err(format!( + "inactive timeout ({inactive_secs}s) must be <= active timeout ({active_secs}s)" + ) + .into()); + } + Ok(Self { + active_usec: (active_secs as i64) * 1_000_000, + inactive_usec: (inactive_secs as i64) * 1_000_000, + }) + } +} + struct Packet { timestamp: i64, // epoch microseconds src_ip: String, @@ -49,7 +79,7 @@ struct FlowRecord { state: FlowState, } -pub fn pcap_to_parquet(input: &str, output: &str) -> Result { +pub fn pcap_to_parquet(input: &str, output: &str, timeouts: FlowTimeouts) -> Result { let mut flows: Vec = Vec::new(); let mut active: HashMap = HashMap::new(); @@ -74,7 +104,7 @@ pub fn pcap_to_parquet(input: &str, output: &str) -> Result { }; let ts = b.ts_sec as i64 * 1_000_000 + frac_usec; if let Some(p) = parse_packet(b.data, linktype, ts, b.origlen as i64) { - ingest_packet(p, &mut active, &mut flows); + ingest_packet(p, &mut active, &mut flows, timeouts); } } PcapBlockOwned::NG(Block::InterfaceDescription(idb)) => { @@ -85,7 +115,7 @@ pub fn pcap_to_parquet(input: &str, output: &str) -> Result { // are out of spike scope. let ts = ((epb.ts_high as i64) << 32) | epb.ts_low as i64; if let Some(p) = parse_packet(epb.data, linktype, ts, epb.origlen as i64) { - ingest_packet(p, &mut active, &mut flows); + ingest_packet(p, &mut active, &mut flows, timeouts); } } _ => {} @@ -207,13 +237,14 @@ fn ingest_packet( packet: Packet, active: &mut HashMap, flows: &mut Vec, + timeouts: FlowTimeouts, ) { let key = make_flow_key(&packet); let ts = packet.timestamp; if let Some(state) = active.get(&key) { - if ts - state.last_timestamp > IDLE_TIMEOUT_USEC - || ts - state.first_timestamp > MAX_FLOW_DURATION_USEC + if ts - state.last_timestamp > timeouts.inactive_usec + || ts - state.first_timestamp > timeouts.active_usec { let state = active.remove(&key).unwrap(); flows.push(FlowRecord { @@ -244,3 +275,91 @@ fn ingest_packet( state.bwd_pkts += 1; } } + +#[cfg(test)] +mod tests { + use super::*; + + fn pkt(ts_usec: i64) -> Packet { + Packet { + timestamp: ts_usec, + src_ip: "10.0.0.1".into(), + dest_ip: "10.0.0.2".into(), + src_port: 12345, + dest_port: 80, + protocol: PROTOCOL_TCP, + packet_bytes: 100, + } + } + + fn aggregate(packets: Vec, timeouts: FlowTimeouts) -> Vec { + let mut flows = Vec::new(); + let mut active = HashMap::new(); + for p in packets { + ingest_packet(p, &mut active, &mut flows, timeouts); + } + flows.extend( + active + .into_iter() + .map(|(key, state)| FlowRecord { key, state }), + ); + flows.sort_by_key(|f| (f.state.first_timestamp, f.key.clone())); + flows + } + + fn defaults() -> FlowTimeouts { + FlowTimeouts::from_secs(DEFAULT_ACTIVE_TIMEOUT_SECS, DEFAULT_INACTIVE_TIMEOUT_SECS) + .expect("default timeouts are valid") + } + + #[test] + fn default_timeouts_are_60s_active_15s_inactive() { + let t = defaults(); + assert_eq!(t.active_usec, 60 * 1_000_000); + assert_eq!(t.inactive_usec, 15 * 1_000_000); + assert_eq!(DEFAULT_ACTIVE_TIMEOUT_SECS, 60); + assert_eq!(DEFAULT_INACTIVE_TIMEOUT_SECS, 15); + } + + #[test] + fn from_secs_rejects_zero_and_inactive_gt_active() { + assert!(FlowTimeouts::from_secs(0, 15).is_err()); + assert!(FlowTimeouts::from_secs(60, 0).is_err()); + assert!(FlowTimeouts::from_secs(15, 60).is_err()); + assert!(FlowTimeouts::from_secs(60, 60).is_ok()); + } + + #[test] + fn inactive_timeout_splits_after_idle_gap() { + // Defaults: inactive 15s. Gap of exactly 15s must NOT split (`>`); + // gap of 15s + 1µs must. + let timeouts = defaults(); + let no_split = aggregate(vec![pkt(0), pkt(15_000_000)], timeouts); + assert_eq!(no_split.len(), 1); + assert_eq!(no_split[0].state.fwd_pkts, 2); + + let split = aggregate(vec![pkt(0), pkt(15_000_001)], timeouts); + assert_eq!(split.len(), 2); + assert_eq!(split[0].state.fwd_pkts, 1); + assert_eq!(split[1].state.fwd_pkts, 1); + } + + #[test] + fn active_timeout_splits_long_continuous_flow() { + // Defaults: active 60s. Keep inter-packet gaps under inactive (15s) + // so only the active axis can fire. Age exactly 60s must NOT split; + // age of 60s + 1µs must. + let timeouts = defaults(); + let continuous: Vec = (0..=6).map(|i| pkt(i * 10_000_000)).collect(); + let no_split = aggregate(continuous, timeouts); + assert_eq!(no_split.len(), 1); + assert_eq!(no_split[0].state.fwd_pkts, 7); + + let mut continuous_then_over = (0..=5).map(|i| pkt(i * 10_000_000)).collect::>(); + continuous_then_over.push(pkt(60_000_001)); + let split = aggregate(continuous_then_over, timeouts); + assert_eq!(split.len(), 2); + assert_eq!(split[0].state.fwd_pkts, 6); + assert_eq!(split[1].state.fwd_pkts, 1); + } +}