pcap: 60s active / 15s inactive timeouts (0.4.0)

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

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"

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
 

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.

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<dyn std::error::Error>> {
 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 } => {

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<T> = std::result::Result<T, Box<dyn std::error::Error>>;
 
-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<Self> {
+        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<usize> {
+pub fn pcap_to_parquet(input: &str, output: &str, timeouts: FlowTimeouts) -> Result<usize> {
     let mut flows: Vec<FlowRecord> = Vec::new();
     let mut active: HashMap<FlowKey, FlowState> = HashMap::new();
 
@@ -74,7 +104,7 @@ pub fn pcap_to_parquet(input: &str, output: &str) -> Result<usize> {
                         };
                         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<usize> {
                         // 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<FlowKey, FlowState>,
     flows: &mut Vec<FlowRecord>,
+    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<Packet>, timeouts: FlowTimeouts) -> Vec<FlowRecord> {
+        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<Packet> = (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::<Vec<_>>();
+        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);
+    }
+}