From 5195be3dd6f1d5997b860b1c20a9d6c8390c63e6 Mon Sep 17 00:00:00 2001 From: Praise Date: Tue, 28 Jul 2026 11:47:50 -0700 Subject: [PATCH] canonicalize: read four more real flow formats (Argus, CIDDS, CIC, Zeek) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds support for four widely-used public flow formats. Verified end to end against real files rather than synthetic fixtures — none of the four converted before this change. Argus / .binetflow ctu-13, iot-23 107,251 rows CIDDS-001/002 traffic week1, 1.09 GB 8,185,992 rows CIC (two variants) CICIDS2017 + v4 225,745 + 18,965 rows Zeek conn.log ctu-sme-11 labeled 35,672 rows schema.json gains 25 aliases and 3 label fields, all APPENDED. resolve_columns is first-match-wins in list order, so appending gives the new entries lowest precedence and cannot change how any existing dataset resolves. Confirmed: the CIDDS and Argus outputs are byte-identical across every later change. Deliberately NOT aliased: Argus TotBytes/TotPkts. Both are totals, not directional, so mapping them onto a fwd_* field would inflate forward volume and corrupt byte-conservation and direction checks. SrcBytes is genuinely forward. Recovering Argus backward bytes needs a derived field (TotBytes - SrcBytes), which has no mechanism here. Guarded by a regression test. Two changes affect every canonicalize input, not just the new formats: * Ports that are not numbers become 0, with the count logged. Real exports routinely have no usable port — protocols that carry none leave the field empty, Argus packs hex ICMP type/code there (0x0303), CIDDS uses a decimal (8.0). src_port/dest_port are non-nullable, so the alternative is failing the file. 0 is the sentinel the nfcapd reader already emits for ICMP, so this makes the two readers consistent. Purely additive: a file that converts today has no unparseable ports by definition. * CSV schema inference now reads the whole file instead of the first 1000 rows. Sampling made reads fail outright when a column's type widened later: a real CICFlowMeter export types column 75 as Int64 from its first 1000 rows, then hits 4308037.666666667 at line 5130 and dies — before canonicalize runs, on a column flowprep does not use. Costs one extra sequential pass (1.09 GB file: 9.7s -> 15.6s, peak RSS +10%). Other reader work: * Reader choice is now content-based (PAR1 magic at both ends of the file) rather than keyed on a .csv extension, which sent .binetflow and every extensionless flow CSV to the parquet reader to fail confusingly. * Whitespace is stripped from text columns on the CSV path. nfdump-derived exports right-align fields (" 44870", " 0.003"); arrow's casts do not trim, so a padded duration failed the file on non-nullable flow_dur. * nfdump magnitude suffixes are expanded (K/M/G/T). 10,748 rows of one real CIDDS file write Bytes as " M" mixed in with plain integers. Unparseable counters still stay null rather than becoming 0 — a byte count of 0 is meaningful data, so it fails loudly instead. * Slash dates are handled. YYYY/MM/DD is unambiguous and rewritten directly. For A/B/YYYY the column is scanned: a component over 12 proves the ordering (CICFlowMeter's 20/11/2020 proves itself), contradictory evidence is an error, and an unprovable column falls back to month-first with a loud warning, since CICIDS2017 per-day exports contain a single date such as 7/7/2017 and cannot be resolved from the data. * New Zeek TSV reader driven by the log's own preamble. Zeek declares its schema out of band, so there is no header row; #fields supplies the names and #types the types, which keeps addr as text so IPv4 and IPv6 both survive. Fields Zeek did not measure ("-") become 0 in numeric columns, counted and logged — 15% of real conn.log rows, where packet counts are present but duration and bytes are not. Non-conn Zeek logs correctly reject for want of byte and duration columns, which covers 168 of the 181 such files we see. * protocol_number gains four IP protocols that were missing: igmp (2), ipv6-icmp (58), ipv6 (41) and pim (103). On the real Argus capture that takes null protocols from 255 to 159 with every other column unchanged. Protocol 58 is matched under all three spellings seen in the wild — IANA's official ipv6-icmp plus the colloquial icmpv6/icmp6. The remaining 159 are correctly null: arp and rarp are layer 2, while rtp, rtcp and udt ride over UDP, so none has an IANA IP protocol number. A comment now records that, so they are not "fixed" with invented values later. Note protocol_number lowercases and trims but does NOT apply normalize_name, so keys keep their hyphens — now covered by a test. Tests: 17 -> 45. The e2e CIC fixture was replaced: it claimed to be "CICIDS-style" but invented Total Fwd Bytes and Flow Duration_Milliseconds, names no CIC export emits, which both happened to already resolve. That fixture is why the CIC gap went unnoticed. Both real spellings are now covered verbatim. Known and intentional: CIC Flow Duration is microseconds but resolves as seconds, so downstream data-quality checks flag the 10^6 inflation rather than flowprep silently rescaling an ambiguous column name. Pinned by test. Co-Authored-By: Claude Opus 5 (1M context) --- schemas/netflow/v1/schema.json | 50 +- src/canonicalize.rs | 1010 +++++++++++++++++++++++++++++++- src/schema.rs | 74 ++- tests/test_e2e.py | 248 +++++++- 4 files changed, 1355 insertions(+), 27 deletions(-) diff --git a/schemas/netflow/v1/schema.json b/schemas/netflow/v1/schema.json index d64dfb7..0ce32ad 100644 --- a/schemas/netflow/v1/schema.json +++ b/schemas/netflow/v1/schema.json @@ -17,7 +17,9 @@ "first_seen", "firstseen", "_time", - "flow_start_milliseconds" + "flow_start_milliseconds", + "starttime", + "date_first_seen" ] }, "src_ip": { @@ -31,7 +33,9 @@ "ipv4_address_cli", "srcaddr", "src_addr", - "ipv4_src_addr" + "ipv4_src_addr", + "src_ip_addr", + "id.orig_h" ] }, "dest_ip": { @@ -47,7 +51,9 @@ "ipv4_address_serv", "dstaddr", "dst_addr", - "ipv4_dst_addr" + "ipv4_dst_addr", + "dst_ip_addr", + "id.resp_h" ] }, "src_port": { @@ -60,7 +66,9 @@ "client_port", "srcport", "sport", - "l4_src_port" + "l4_src_port", + "src_pt", + "id.orig_p" ] }, "dest_port": { @@ -75,7 +83,9 @@ "server_port", "dstport", "dport", - "l4_dst_port" + "l4_dst_port", + "dst_pt", + "id.resp_p" ] }, "fwd_bytes": { @@ -93,7 +103,11 @@ "bytes_sent", "octets_fwd", "total_fwd_bytes", - "octets" + "octets", + "srcbytes", + "total_length_of_fwd_packets", + "totlen_fwd_pkts", + "orig_bytes" ] }, "bwd_bytes": { @@ -109,7 +123,10 @@ "s_bytes_all", "bytes_received", "octets_bwd", - "total_bwd_bytes" + "total_bwd_bytes", + "total_length_of_bwd_packets", + "totlen_bwd_pkts", + "resp_bytes" ] }, "fwd_pkts": { @@ -128,7 +145,10 @@ "packets_sent", "total_fwd_pkts", "total_fwd_packets", - "packets" + "packets", + "tot_fwd_pkts", + "total_forward_packets", + "orig_pkts" ] }, "bwd_pkts": { @@ -146,7 +166,10 @@ "packets_out", "packets_received", "total_bwd_pkts", - "total_bwd_packets" + "total_bwd_packets", + "tot_bwd_pkts", + "total_backward_packets", + "resp_pkts" ] }, "flow_dur": { @@ -167,7 +190,8 @@ "flow_duration_microseconds", "duration_ns", "duration_nanoseconds", - "flow_duration_nanoseconds" + "flow_duration_nanoseconds", + "dur" ], "unit_detection": { "seconds": [ @@ -176,7 +200,8 @@ "durat", "flow_duration", "duration_sec", - "elapsed_time" + "elapsed_time", + "dur" ], "milliseconds": [ "duration_ms", @@ -223,6 +248,9 @@ "label_fields": [ "attack", "attack_type", + "attacktype", + "class", + "detailedlabel", "src_tags", "dst_tags" ], diff --git a/src/canonicalize.rs b/src/canonicalize.rs index b4efea0..e887ee3 100644 --- a/src/canonicalize.rs +++ b/src/canonicalize.rs @@ -5,7 +5,9 @@ //! name), and timestamp encoding (epoch s/ms/us/ns or string datetimes, //! inferred from magnitude/type). +use std::borrow::Cow; use std::fs::File; +use std::io::{BufRead, Read, Seek, SeekFrom}; use std::sync::Arc; use arrow::array::{Array, ArrayRef, Float64Array, Int32Array, Int64Array, StringArray}; @@ -27,26 +29,285 @@ pub fn canonicalize_file(input: &str, output: &str) -> Result { Ok(out.num_rows()) } +/// Parquet's file magic, written at both the start and the end of the file. +const PARQUET_MAGIC: &[u8; 4] = b"PAR1"; + +/// Detect parquet from file content rather than from the file extension. +/// +/// The extension is not a reliable signal for flow exports: Argus writes +/// `.binetflow`, plenty of flow CSVs arrive as `.txt` or with no extension at +/// all, and keying on `.csv` alone sent every one of them to the parquet reader +/// to fail on a confusing error. Both the leading and the trailing magic are +/// checked, so a delimited-text file whose first column happens to be named +/// `PAR1` cannot be mistaken for parquet. +fn is_parquet(reader: &mut R) -> Result { + let magic_len = PARQUET_MAGIC.len(); + // A valid parquet file carries the magic twice, so it cannot be shorter. + if reader.seek(SeekFrom::End(0))? < 2 * magic_len as u64 { + return Ok(false); + } + let mut buf = [0u8; 4]; + reader.rewind()?; + reader.read_exact(&mut buf)?; + if &buf != PARQUET_MAGIC { + return Ok(false); + } + reader.seek(SeekFrom::End(-(magic_len as i64)))?; + reader.read_exact(&mut buf)?; + Ok(&buf == PARQUET_MAGIC) +} + +/// A Zeek log's `#`-prefixed preamble. Zeek keeps the column names and types +/// out of band, so without reading it the columns have no names to resolve +/// aliases against. +struct ZeekPreamble { + separator: u8, + fields: Vec, + types: Vec, + unset: String, + empty: String, +} + +/// Decode `#separator \x09` (the escape is written literally) into a byte. +fn decode_zeek_separator(line: &str) -> u8 { + let spec = line.trim_end_matches(['\n', '\r']); + // Strip the key, keeping any literal separator byte that follows it. + let spec = spec + .strip_prefix("#separator ") + .or_else(|| spec.strip_prefix("#separator")) + .unwrap_or(""); + if let Some(hex) = spec.strip_prefix("\\x") { + // An escape that does not decode is not a literal backslash separator — + // fall back to Zeek's default rather than picking up the '\'. + return u8::from_str_radix(hex.trim(), 16).unwrap_or(b'\t'); + } + spec.as_bytes().first().copied().unwrap_or(b'\t') +} + +/// Read a Zeek log preamble, or `None` when the file is not a Zeek log. +/// +/// Zeek writes its schema as directives before the data: +/// #separator \x09 +/// #unset_field - +/// #fields ts uid id.orig_h id.orig_p ... +/// #types time string addr port ... +fn read_zeek_preamble(path: &str) -> Result> { + let mut reader = std::io::BufReader::new(File::open(path)?); + let mut line = String::new(); + if reader.read_line(&mut line)? == 0 || !line.starts_with("#separator") { + return Ok(None); + } + let separator = decode_zeek_separator(&line); + let mut preamble = ZeekPreamble { + separator, + fields: Vec::new(), + types: Vec::new(), + unset: "-".to_string(), + empty: "(empty)".to_string(), + }; + loop { + line.clear(); + if reader.read_line(&mut line)? == 0 { + break; + } + if !line.starts_with('#') { + break; + } + let cleaned = line.trim_end_matches(['\n', '\r']); + let mut parts = cleaned.split(separator as char); + match parts.next().unwrap_or("") { + "#fields" => preamble.fields = parts.map(str::to_string).collect(), + "#types" => preamble.types = parts.map(str::to_string).collect(), + "#unset_field" => { + if let Some(v) = parts.next() { + preamble.unset = v.to_string(); + } + } + "#empty_field" => { + if let Some(v) = parts.next() { + preamble.empty = v.to_string(); + } + } + _ => {} + } + } + if preamble.fields.is_empty() { + return Err("Zeek log has a #separator directive but no #fields line".into()); + } + Ok(Some(preamble)) +} + +/// Map a Zeek type name to the arrow type it should become. `None` means leave +/// it as text (`string`, `addr`, `enum`, `bool`, `set[..]`, `vector[..]`). +fn zeek_arrow_type(declared: &str) -> Option { + match declared { + "time" | "interval" | "double" => Some(DataType::Float64), + "count" | "int" | "port" => Some(DataType::Int64), + _ => None, + } +} + +/// Apply Zeek's declared types, substituting 0 for unset numeric values. +/// +/// Zeek marks a field it did not measure with `-` (and an empty one with +/// `(empty)`). On roughly 15% of real conn.log rows that covers `duration`, +/// `orig_bytes` and `resp_bytes` together — connections it saw but did not fully +/// analyse — while the packet counts stay populated. `flow_dur` and `fwd_bytes` +/// are non-nullable canonically, so the alternative to a substitute is failing +/// every such file. +/// +/// 0 is close to the truth for these rows (they are single-packet or unanalysed +/// connections) and, more importantly, it stays visible: `flow_check`'s +/// `bytes.zero_with_packets` check exists for exactly this signature — zero bytes +/// against non-zero packets — so the substituted rows get flagged rather than +/// passing silently. Scoped to the Zeek reader, so no other format's behaviour +/// changes. +fn apply_zeek_types(batch: &RecordBatch, preamble: &ZeekPreamble) -> Result { + let mut substituted = 0usize; + let mut fields: Vec = Vec::with_capacity(batch.num_columns()); + let mut columns: Vec = Vec::with_capacity(batch.num_columns()); + + for (i, field) in batch.schema().fields().iter().enumerate() { + let declared = preamble.types.get(i).map(String::as_str).unwrap_or(""); + let Some(target) = zeek_arrow_type(declared) else { + fields.push(field.as_ref().clone()); + columns.push(batch.column(i).clone()); + continue; + }; + let text = batch + .column(i) + .as_any() + .downcast_ref::() + .ok_or("Zeek columns are read as text")?; + let mut values: Vec> = Vec::with_capacity(text.len()); + for value in text.iter() { + match value { + Some(s) if s == preamble.unset || s == preamble.empty => { + substituted += 1; + values.push(Some("0")); + } + other => values.push(other), + } + } + let filled: ArrayRef = Arc::new(StringArray::from(values)); + fields.push(Field::new(field.name(), target.clone(), true)); + columns.push(cast(&filled, &target)?); + } + + if substituted > 0 { + eprintln!( + "canonicalize: zeek — substituted 0 for {substituted} unset numeric value(s) \ + (Zeek writes '{}' where it did not measure a field; flow_check's \ + bytes.zero_with_packets surfaces the affected rows)", + preamble.unset + ); + } + Ok(RecordBatch::try_new( + Arc::new(Schema::new(fields)), + columns, + )?) +} + +/// Read a Zeek TSV log into a batch, using its own declared schema. +fn read_zeek_table(path: &str, preamble: &ZeekPreamble) -> Result { + // Every column is read as text first: Zeek's unset marker `-` sits in + // numeric columns, and arrow's typed CSV parser rejects it outright. + let schema = Arc::new(Schema::new( + preamble + .fields + .iter() + .map(|name| Field::new(name, DataType::Utf8, true)) + .collect::>(), + )); + let format = arrow::csv::reader::Format::default() + .with_header(false) + .with_delimiter(preamble.separator) + .with_comment(b'#'); + let reader = arrow::csv::ReaderBuilder::new(schema) + .with_format(format) + .build(File::open(path)?)?; + let batches: Vec = reader.collect::>()?; + if batches.is_empty() { + return Err("input file contains no rows".into()); + } + let batch = concat_batches(&batches[0].schema(), &batches)?; + apply_zeek_types(&trim_text_columns(&batch)?, preamble) +} + fn read_table(path: &str) -> Result { - let batches: Vec = if path.ends_with(".csv") { + let parquet = { + let mut probe = File::open(path)?; + is_parquet(&mut probe)? + }; + // Zeek logs are delimited text but carry their schema in a `#` preamble, so + // they need reading before the generic CSV path guesses at a header row. + if !parquet { + if let Some(preamble) = read_zeek_preamble(path)? { + return read_zeek_table(path, &preamble); + } + } + let batches: Vec = if parquet { + let file = File::open(path)?; + let reader = parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(file)? + .build()?; + reader.collect::>()? + } else { let mut file = File::open(path)?; let format = arrow::csv::reader::Format::default().with_header(true); - let (schema, _) = format.infer_schema(&mut file, Some(1000))?; + // Infer over the WHOLE file, not a leading sample. Sampling the first N + // rows makes the read fail outright when a column's type widens later: + // a real CICFlowMeter export types column 75 as Int64 from its first + // 1000 rows, then hits `4308037.666666667` at line 5130 and dies — + // before canonicalize ever runs, on a column flowprep does not even use. + // Costs one extra sequential pass, which is cheap next to being unable + // to read the file at all. + let (schema, _) = format.infer_schema(&mut file, None)?; let file = File::open(path)?; let reader = arrow::csv::ReaderBuilder::new(Arc::new(schema)) .with_format(format) .build(file)?; reader.collect::>()? - } else { - let file = File::open(path)?; - let reader = parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(file)? - .build()?; - reader.collect::>()? }; if batches.is_empty() { return Err("input file contains no rows".into()); } - Ok(concat_batches(&batches[0].schema(), &batches)?) + let batch = concat_batches(&batches[0].schema(), &batches)?; + // Padding is a text-format artifact, so only delimited input needs it + // stripped; parquet columns arrive already typed. + if parquet { + Ok(batch) + } else { + trim_text_columns(&batch) + } +} + +/// Strip surrounding whitespace from every text column. +/// +/// nfdump-derived exports right-align their fields: CIDDS writes `Src Pt` as +/// `" 44870"`, `Duration` as `" 0.003"` and `Proto` as `"TCP "`. Arrow's +/// casts do not trim, so `" 44870"` casts to null — and with `flow_dur` +/// non-nullable a padded duration fails the whole file. +/// +/// Done once, centrally, rather than at each use site, because whitespace is +/// never meaningful in any canonical field. It matters for labels too: `"normal "` +/// and `"normal"` would otherwise count as two distinct classes downstream. +fn trim_text_columns(batch: &RecordBatch) -> Result { + let columns: Vec = batch + .columns() + .iter() + .map( + |column| match column.as_any().downcast_ref::() { + Some(strings) => Arc::new( + strings + .iter() + .map(|v| v.map(str::trim)) + .collect::(), + ) as ArrayRef, + None => column.clone(), + }, + ) + .collect(); + Ok(RecordBatch::try_new(batch.schema(), columns)?) } pub fn canonicalize(batch: &RecordBatch) -> Result { @@ -77,12 +338,23 @@ pub fn canonicalize(batch: &RecordBatch) -> Result { .clone() }; let n = batch.num_rows(); + + let (src_port, src_coerced) = port_to_i32(&col("src_port"))?; + let (dest_port, dest_coerced) = port_to_i32(&col("dest_port"))?; + if src_coerced + dest_coerced > 0 { + eprintln!( + "canonicalize: coerced {src_coerced} src_port and {dest_coerced} dest_port \ + value(s) to 0 — absent, or not a number (e.g. Argus writes hex ICMP \ + type/code such as 0x0303 into Sport/Dport)" + ); + } + let mut columns: Vec = vec![ timestamp_to_micros(&col("timestamp"))?, cast(&col("src_ip"), &DataType::Utf8)?, cast(&col("dest_ip"), &DataType::Utf8)?, - cast(&col("src_port"), &DataType::Int32)?, - cast(&col("dest_port"), &DataType::Int32)?, + src_port, + dest_port, to_rounded_i64(&col("fwd_bytes"))?, ]; @@ -94,7 +366,9 @@ pub fn canonicalize(batch: &RecordBatch) -> Result { for pkts in ["fwd_pkts", "bwd_pkts"] { if resolved.contains_key(pkts) { - columns.push(cast(&col(pkts), &DataType::Int64)?); + // Same reader as bytes: nfdump suffixes packet counts too, and a + // text column of "2.0"-style values must not silently become null. + columns.push(to_rounded_i64(&col(pkts))?); } else { columns.push(Arc::new(Int64Array::from(vec![None::; n]))); } @@ -143,13 +417,171 @@ pub fn canonicalize(batch: &RecordBatch) -> Result { )?) } +/// Rewrite a leading `YYYY/MM/DD` date as `YYYY-MM-DD`. +/// +/// Argus writes `StartTime` as `2011/08/18 10:21:46.633335`. Arrow's +/// string→timestamp cast accepts only dash-separated dates, so the slash form +/// casts to null instead of erroring — and because `timestamp` is non-nullable +/// in the canonical schema the failure surfaces much later as a confusing +/// "declared as non-nullable but contains null values" write error. +/// +/// Only the unambiguous `dddd/dd/dd` shape is rewritten. Day-first and +/// month-first slash formats (`18/08/2011`, `08/18/2011`) are left alone: they +/// cannot be told apart without external knowledge, so guessing would risk +/// silently transposing month and day. +fn normalize_slash_date(value: &str) -> Cow<'_, str> { + let b = value.as_bytes(); + let is_ymd_slash = b.len() >= 10 + && b[..4].iter().all(u8::is_ascii_digit) + && b[4] == b'/' + && b[5..7].iter().all(u8::is_ascii_digit) + && b[7] == b'/' + && b[8..10].iter().all(u8::is_ascii_digit); + if !is_ymd_slash { + return Cow::Borrowed(value); + } + let mut owned = value.to_string(); + owned.replace_range(4..5, "-"); + owned.replace_range(7..8, "-"); + Cow::Owned(owned) +} + +/// Which component of an `A/B/YYYY` date is the month. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SlashDateOrder { + MonthFirst, + DayFirst, +} + +/// Parse the leading `A/B/YYYY` of a value into `(a, b, year_and_remainder)`. +/// +/// Returns `None` for anything else — including the `YYYY/MM/DD` shape, whose +/// first component is four digits, so that form stays with +/// `normalize_slash_date` where it is unambiguous. +fn slash_date_parts(value: &str) -> Option<(u32, u32, &str)> { + let mut it = value.trim().splitn(3, '/'); + let a = it.next()?; + let b = it.next()?; + let tail = it.next()?; + let two_digit = |s: &str| (1..=2).contains(&s.len()) && s.bytes().all(|c| c.is_ascii_digit()); + if !two_digit(a) || !two_digit(b) || tail.len() < 4 { + return None; + } + if !tail.as_bytes()[..4].iter().all(u8::is_ascii_digit) { + return None; + } + Some((a.parse().ok()?, b.parse().ok()?, tail)) +} + +/// Decide whether a column of `A/B/YYYY` dates is month-first or day-first. +/// +/// Evidence beats assumption: a first component over 12 can only be a day, and a +/// second component over 12 can only be a day, so one such value anywhere in the +/// column proves the ordering. CICFlowMeter v4 writes `20/11/2020`, which proves +/// itself. +/// +/// When nothing proves it the file is irreducibly ambiguous — a CICIDS2017 +/// per-day export contains the single date `7/7/2017` and no more — and we assume +/// month-first, which is what CICIDS2017 and most CIC tooling emit. Either way +/// the decision is logged, so a wrong assumption is visible rather than silent. +/// +/// `Ok(None)` means the column holds no such dates at all, so nothing is logged. +fn infer_slash_date_order(values: &StringArray) -> Result> { + let (mut seen, mut day_first, mut month_first) = (false, false, false); + for value in values.iter().flatten() { + if let Some((a, b, _)) = slash_date_parts(value) { + seen = true; + day_first |= a > 12; + month_first |= b > 12; + } + } + if !seen { + return Ok(None); + } + match (day_first, month_first) { + (true, true) => Err("timestamp column mixes D/M/YYYY and M/D/YYYY dates; \ + no single ordering can be correct" + .into()), + (true, false) => { + eprintln!("canonicalize: timestamp dates read as D/M/YYYY (proven: a day over 12)"); + Ok(Some(SlashDateOrder::DayFirst)) + } + (false, true) => { + eprintln!("canonicalize: timestamp dates read as M/D/YYYY (proven: a day over 12)"); + Ok(Some(SlashDateOrder::MonthFirst)) + } + (false, false) => { + eprintln!( + "canonicalize: timestamp dates ASSUMED M/D/YYYY — no value in the column has a \ + component over 12, so the ordering cannot be proven (CICIDS2017 per-day exports \ + look like this). If this source is day-first, every date is wrong." + ); + Ok(Some(SlashDateOrder::MonthFirst)) + } + } +} + +/// Rewrite `A/B/YYYY[ H:M[:S]]` as `YYYY-MM-DD HH:MM:SS`, which arrow can cast. +/// +/// The time fragment is zero-padded because sources are inconsistent about it: +/// CICIDS2017 writes `7/7/2017 3:30` (single-digit hour, no seconds) while +/// CICFlowMeter v4 writes `20/11/2020 09:50:19`. Note CICIDS2017 omits AM/PM +/// entirely, so `3:30` is taken literally as 03:30 — a known flaw of that export, +/// not something this can recover. +fn rewrite_slash_date(value: &str, order: SlashDateOrder) -> Option { + let (a, b, tail) = slash_date_parts(value)?; + let (month, day) = match order { + SlashDateOrder::MonthFirst => (a, b), + SlashDateOrder::DayFirst => (b, a), + }; + let year = &tail[..4]; + let mut time = tail[4..].trim().split(':'); + let parse_or_zero = |part: Option<&str>| -> Option { + match part.map(str::trim).filter(|s| !s.is_empty()) { + Some(s) => s.parse().ok(), + None => Some(0), + } + }; + let hour = parse_or_zero(time.next())?; + let minute = parse_or_zero(time.next())?; + let second = time + .next() + .map(str::trim) + .filter(|s| !s.is_empty()) + .unwrap_or("00"); + Some(format!( + "{year}-{month:02}-{day:02} {hour:02}:{minute:02}:{second:0>2}" + )) +} + /// Coerce a timestamp column to epoch microseconds (int64). fn timestamp_to_micros(column: &ArrayRef) -> Result { match column.data_type() { - DataType::Timestamp(_, _) | DataType::Utf8 | DataType::LargeUtf8 => { + DataType::Timestamp(_, _) => { let ts = cast(column, &DataType::Timestamp(TimeUnit::Microsecond, None))?; Ok(cast(&ts, &DataType::Int64)?) } + DataType::Utf8 | DataType::LargeUtf8 => { + let strings = cast(column, &DataType::Utf8)?; + let strings = strings.as_any().downcast_ref::().unwrap(); + let order = infer_slash_date_order(strings)?; + let normalized: ArrayRef = Arc::new( + strings + .iter() + .map(|v| { + v.map(|s| match order.and_then(|o| rewrite_slash_date(s, o)) { + Some(rewritten) => rewritten, + None => normalize_slash_date(s).into_owned(), + }) + }) + .collect::(), + ); + let ts = cast( + &normalized, + &DataType::Timestamp(TimeUnit::Microsecond, None), + )?; + Ok(cast(&ts, &DataType::Int64)?) + } _ => { let floats = cast(column, &DataType::Float64)?; let floats = floats.as_any().downcast_ref::().unwrap(); @@ -171,7 +603,72 @@ fn timestamp_to_micros(column: &ArrayRef) -> Result { } } +/// Cast a port column to `Int32`, substituting 0 for anything that does not +/// parse. Returns the array plus how many values were substituted. +/// +/// `src_port`/`dest_port` are non-nullable in the canonical schema, so the only +/// alternative to a sentinel is failing the whole file. Real exports routinely +/// have no usable port: protocols that carry none (arp, igmp, ipx/spx) leave the +/// field empty, and Argus packs ICMP type/code into it as hex (`0x0303`), which +/// is not a port at all. 0 is the sentinel flowprep's own nfcapd reader already +/// emits for ICMP, so this keeps the two readers consistent. +/// +/// Note this applies to every `canonicalize` input, not just Argus. It is purely +/// additive: a file that converts today has no unparseable ports by definition, +/// so no existing behaviour changes. Downstream, a flood of zero ports is +/// visible rather than silent — `flow_check` has `structure.port_zero` and +/// `integrity.port_range` for exactly this. +fn port_to_i32(column: &ArrayRef) -> Result<(ArrayRef, usize)> { + // arrow's default cast is safe: unparseable input becomes null, not an error. + let ints = cast(column, &DataType::Int32)?; + let coerced = ints.null_count(); + if coerced == 0 { + return Ok((ints, coerced)); + } + let typed = ints.as_any().downcast_ref::().unwrap(); + let filled = Int32Array::from_iter_values(typed.iter().map(|v| v.unwrap_or(0))); + Ok((Arc::new(filled), coerced)) +} + +/// Expand an nfdump-style magnitude suffix into a plain number. +/// +/// nfdump writes large counters human-readably, and CIDDS — which is derived +/// from it — inherits that: a `Bytes` column holds plain integers and values +/// like `"1.4 M"` side by side. Arrow types such a column as Utf8 and casts the +/// suffixed entries to null, so with `fwd_bytes` non-nullable a single suffixed +/// row fails the entire file. +fn parse_magnitude(value: &str) -> Option { + let trimmed = value.trim(); + // Slicing at len-1 is safe: we only do it when the last byte is ASCII, + // which is always a char boundary. + let (digits, multiplier) = match trimmed.as_bytes().last()?.to_ascii_uppercase() { + b'K' => (&trimmed[..trimmed.len() - 1], 1e3), + b'M' => (&trimmed[..trimmed.len() - 1], 1e6), + b'G' => (&trimmed[..trimmed.len() - 1], 1e9), + b'T' => (&trimmed[..trimmed.len() - 1], 1e12), + _ => (trimmed, 1.0), + }; + digits.trim().parse::().ok().map(|v| v * multiplier) +} + +/// Cast a counter column (bytes, packets) to `Int64`, rounding, and expanding +/// nfdump magnitude suffixes when the column arrived as text. +/// +/// Deliberately does NOT substitute a sentinel for unparseable input, unlike +/// `port_to_i32`: a byte or packet count of 0 is meaningful data, so quietly +/// inventing one would corrupt a measure rather than fill in a field the source +/// genuinely lacks. Truly unparseable counters stay null — which fails the file +/// for non-nullable `fwd_bytes`, loudly and on purpose. fn to_rounded_i64(column: &ArrayRef) -> Result { + if matches!(column.data_type(), DataType::Utf8 | DataType::LargeUtf8) { + let strings = cast(column, &DataType::Utf8)?; + let strings = strings.as_any().downcast_ref::().unwrap(); + return Ok(Arc::new(Int64Array::from_iter( + strings + .iter() + .map(|v| v.and_then(parse_magnitude).map(|x| x.round() as i64)), + ))); + } let floats = cast(column, &DataType::Float64)?; let floats = floats.as_any().downcast_ref::().unwrap(); Ok(Arc::new(Int64Array::from_iter( @@ -189,3 +686,492 @@ fn protocol_to_number(column: &ArrayRef) -> Result { strings.iter().map(|v| v.and_then(protocol_number)), ))) } + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Cursor; + + fn sniff(bytes: &[u8]) -> bool { + is_parquet(&mut Cursor::new(bytes.to_vec())).expect("sniff succeeds") + } + + /// The Argus/CTU-13 header, verbatim. `.binetflow` files are comma-delimited + /// text, so content detection must route them to the CSV reader. + const ARGUS_HEADER: &str = "StartTime,Dur,Proto,SrcAddr,Sport,Dir,DstAddr,Dport,State,\ + sTos,dTos,TotPkts,TotBytes,SrcBytes,Label"; + + #[test] + fn detects_parquet_by_magic_not_extension() { + // Magic at both ends is parquet, whatever the file is called. + assert!(sniff(b"PAR1....payload....PAR1")); + // Argus text named `.binetflow` is not parquet. + assert!(!sniff(ARGUS_HEADER.as_bytes())); + } + + #[test] + fn rejects_partial_parquet_magic() { + // Leading magic only: a CSV whose first column is literally `PAR1` + // must not be handed to the parquet reader. + assert!(!sniff(b"PAR1,src_ip,dest_ip\n1,2,3\n")); + // Trailing magic only. + assert!(!sniff(b"not parquet at all PAR1")); + // Too short to carry the magic twice. + assert!(!sniff(b"PAR1")); + assert!(!sniff(b"")); + } + + #[test] + fn argus_header_resolves_every_required_field() { + let names: Vec = ARGUS_HEADER.split(',').map(|s| s.to_string()).collect(); + let resolved = load_schema_spec().resolve_columns(&names); + + for field in REQUIRED_FIELDS { + assert!( + resolved.contains_key(*field), + "Argus header should resolve {field}, got {resolved:?}" + ); + } + assert_eq!( + resolved.get("timestamp").map(String::as_str), + Some("StartTime") + ); + assert_eq!(resolved.get("flow_dur").map(String::as_str), Some("Dur")); + assert_eq!(resolved.get("src_ip").map(String::as_str), Some("SrcAddr")); + assert_eq!(resolved.get("dest_ip").map(String::as_str), Some("DstAddr")); + } + + /// Argus `TotBytes`/`TotPkts` are *totals*, not directional measures. + /// Mapping either onto a fwd_* field would inflate forward volume and + /// corrupt downstream byte-conservation and direction checks, so + /// `fwd_bytes` must come from `SrcBytes` and the packet counts must stay + /// unresolved (canonicalize null-fills them) until a derived-field + /// mechanism can compute `TotBytes - SrcBytes`. + #[test] + fn argus_totals_are_not_mistaken_for_directional_counts() { + let names: Vec = ARGUS_HEADER.split(',').map(|s| s.to_string()).collect(); + let resolved = load_schema_spec().resolve_columns(&names); + + assert_eq!( + resolved.get("fwd_bytes").map(String::as_str), + Some("SrcBytes") + ); + assert!( + !resolved.contains_key("bwd_bytes"), + "TotBytes is not bwd_bytes" + ); + assert!( + !resolved.contains_key("fwd_pkts"), + "TotPkts is not fwd_pkts" + ); + assert!(!resolved.contains_key("bwd_pkts")); + } + + /// `Dur` is seconds in Argus. Guard the divisor explicitly rather than + /// relying on the unmatched-name fallback. + #[test] + fn argus_duration_is_seconds() { + let spec = load_schema_spec(); + assert_eq!(spec.duration_divisors.get("dur").copied(), Some(1.0)); + } + + /// CIDDS-001/002, as produced by nfdump. Note the label column is named + /// `class` in the OpenStack/ExternalServer captures and `label` in the + /// `traffic__week*` ones — both must survive. + const CIDDS_HEADER: &str = "Date first seen,Duration,Proto,Src IP Addr,Src Pt,Dst IP Addr,\ + Dst Pt,Packets,Bytes,Flows,Flags,Tos,class,attackType,attackID,\ + attackDescription"; + + #[test] + fn cidds_header_resolves_every_required_field() { + let names: Vec = CIDDS_HEADER.split(',').map(|s| s.to_string()).collect(); + let resolved = load_schema_spec().resolve_columns(&names); + + for field in REQUIRED_FIELDS { + assert!( + resolved.contains_key(*field), + "CIDDS header should resolve {field}, got {resolved:?}" + ); + } + assert_eq!( + resolved.get("timestamp").map(String::as_str), + Some("Date first seen") + ); + assert_eq!( + resolved.get("src_ip").map(String::as_str), + Some("Src IP Addr") + ); + assert_eq!(resolved.get("src_port").map(String::as_str), Some("Src Pt")); + assert_eq!( + resolved.get("dest_port").map(String::as_str), + Some("Dst Pt") + ); + // CIDDS rows are unidirectional, so its single Bytes/Packets totals are + // genuinely the forward direction for that row. + assert_eq!(resolved.get("fwd_bytes").map(String::as_str), Some("Bytes")); + } + + #[test] + fn cidds_label_columns_are_passthrough() { + let spec = load_schema_spec(); + for col in ["class", "attacktype", "label"] { + assert!( + spec.passthrough.iter().any(|p| p == col), + "{col} should be preserved as a ground-truth label" + ); + } + } + + #[test] + fn expands_nfdump_magnitude_suffixes() { + // Real CIDDS shapes: 10,748 rows of one 1.09 GB file carry " M". + assert_eq!(parse_magnitude("1.4 M"), Some(1_400_000.0)); + assert_eq!(parse_magnitude("999.9 M"), Some(999_900_000.0)); + assert_eq!(parse_magnitude("2 K"), Some(2_000.0)); + assert_eq!(parse_magnitude("1.5G"), Some(1_500_000_000.0)); + assert_eq!(parse_magnitude("3 T"), Some(3e12)); + // Plain numbers are untouched. + assert_eq!(parse_magnitude("174"), Some(174.0)); + assert_eq!(parse_magnitude(" 46 "), Some(46.0)); + assert_eq!(parse_magnitude("8.0"), Some(8.0)); + // Junk yields None rather than a wrong number. + assert_eq!(parse_magnitude(""), None); + assert_eq!(parse_magnitude("---"), None); + assert_eq!(parse_magnitude("M"), None); + } + + #[test] + fn counter_columns_expand_suffixes_but_keep_junk_null() { + let raw: ArrayRef = Arc::new(StringArray::from(vec![ + Some("174"), + Some("1.4 M"), + Some("---"), + None, + ])); + let out = to_rounded_i64(&raw).expect("counter cast succeeds"); + let out = out.as_any().downcast_ref::().unwrap(); + assert_eq!(out.value(0), 174); + assert_eq!(out.value(1), 1_400_000); + // A byte count of 0 is meaningful data, so junk must NOT become 0 — + // it stays null and fails the file loudly for non-nullable fwd_bytes. + assert!(out.is_null(2) && out.is_null(3)); + } + + #[test] + fn decodes_zeek_separator_directive() { + // Zeek writes the escape literally, as four characters. + assert_eq!(decode_zeek_separator("#separator \\x09\n"), b'\t'); + assert_eq!(decode_zeek_separator("#separator \\x2c"), b','); + // A writer that emits the literal byte instead. + assert_eq!(decode_zeek_separator("#separator\t"), b'\t'); + // Nothing usable falls back to tab, Zeek's default. + assert_eq!(decode_zeek_separator("#separator"), b'\t'); + assert_eq!(decode_zeek_separator("#separator \\xZZ"), b'\t'); + } + + #[test] + fn maps_zeek_declared_types() { + // Numeric: these are the ones an unset `-` must be substituted in. + assert_eq!(zeek_arrow_type("time"), Some(DataType::Float64)); + assert_eq!(zeek_arrow_type("interval"), Some(DataType::Float64)); + assert_eq!(zeek_arrow_type("double"), Some(DataType::Float64)); + assert_eq!(zeek_arrow_type("count"), Some(DataType::Int64)); + assert_eq!(zeek_arrow_type("port"), Some(DataType::Int64)); + assert_eq!(zeek_arrow_type("int"), Some(DataType::Int64)); + // Text: addr stays a string so IPv4 and IPv6 both survive. + assert_eq!(zeek_arrow_type("addr"), None); + assert_eq!(zeek_arrow_type("string"), None); + assert_eq!(zeek_arrow_type("enum"), None); + assert_eq!(zeek_arrow_type("bool"), None); + assert_eq!(zeek_arrow_type("set[string]"), None); + assert_eq!(zeek_arrow_type("vector[interval]"), None); + } + + /// The real ctu-sme-11 `conn.log.labeled` field list, verbatim. + const ZEEK_CONN_FIELDS: &str = "ts,uid,id.orig_h,id.orig_p,id.resp_h,id.resp_p,proto,service,\ + duration,orig_bytes,resp_bytes,conn_state,local_orig,\ + local_resp,missed_bytes,history,orig_pkts,orig_ip_bytes,\ + resp_pkts,resp_ip_bytes,tunnel_parents,label,detailedlabel"; + + #[test] + fn zeek_conn_log_resolves_every_required_field() { + let names: Vec = ZEEK_CONN_FIELDS.split(',').map(|s| s.to_string()).collect(); + let resolved = load_schema_spec().resolve_columns(&names); + for field in REQUIRED_FIELDS { + assert!( + resolved.contains_key(*field), + "Zeek conn.log should resolve {field}, got {resolved:?}" + ); + } + // Dots are not normalized away, so these aliases must match literally. + assert_eq!( + resolved.get("src_ip").map(String::as_str), + Some("id.orig_h") + ); + assert_eq!( + resolved.get("dest_ip").map(String::as_str), + Some("id.resp_h") + ); + assert_eq!( + resolved.get("src_port").map(String::as_str), + Some("id.orig_p") + ); + assert_eq!( + resolved.get("dest_port").map(String::as_str), + Some("id.resp_p") + ); + // orig_bytes/resp_bytes are payload counts and genuinely directional. + assert_eq!( + resolved.get("fwd_bytes").map(String::as_str), + Some("orig_bytes") + ); + assert_eq!( + resolved.get("bwd_bytes").map(String::as_str), + Some("resp_bytes") + ); + assert_eq!( + resolved.get("fwd_pkts").map(String::as_str), + Some("orig_pkts") + ); + } + + /// The sibling Zeek logs (dns, ssl, x509, weird, …) carry no bytes, packets + /// or duration. 168 of the 181 `.labeled` files in the corpus are these, and + /// they must reject rather than convert into junk flows. + #[test] + fn zeek_non_conn_logs_lack_required_fields() { + let dns = "ts,uid,id.orig_h,id.orig_p,id.resp_h,id.resp_p,proto,trans_id,rtt,query,\ + qclass,qclass_name,qtype,qtype_name,rcode,rcode_name,AA,TC,RD,RA,Z,answers,\ + TTLs,rejected,label,detailedlabel"; + let names: Vec = dns.split(',').map(|s| s.to_string()).collect(); + let resolved = load_schema_spec().resolve_columns(&names); + let missing: Vec<&&str> = REQUIRED_FIELDS + .iter() + .filter(|f| !resolved.contains_key(**f)) + .collect(); + assert_eq!( + missing, + vec![&"fwd_bytes", &"flow_dur"], + "a Zeek dns.log must be rejected for want of byte and duration columns" + ); + } + + #[test] + fn zeek_detailedlabel_is_passthrough() { + assert!( + load_schema_spec() + .passthrough + .iter() + .any(|p| p == "detailedlabel") + ); + } + + #[test] + fn coerces_unusable_ports_to_zero_and_counts_them() { + // Argus reality: hex ICMP type/code, an empty field for a protocol with + // no ports, and ordinary numeric ports side by side. + let raw: ArrayRef = Arc::new(StringArray::from(vec![ + Some("1611"), + Some("0x0303"), + Some(""), + None, + Some("443"), + ])); + let (ports, coerced) = port_to_i32(&raw).expect("port cast succeeds"); + let ports = ports.as_any().downcast_ref::().unwrap(); + + assert_eq!(coerced, 3, "0x0303, empty and null should all be coerced"); + assert_eq!(ports.values(), &[1611, 0, 0, 0, 443]); + assert_eq!(ports.null_count(), 0, "canonical ports are non-nullable"); + } + + #[test] + fn leaves_clean_port_columns_untouched() { + let raw: ArrayRef = Arc::new(StringArray::from(vec!["1611", "443", "0"])); + let (ports, coerced) = port_to_i32(&raw).expect("port cast succeeds"); + assert_eq!(coerced, 0, "nothing to coerce means no reported coercions"); + let ports = ports.as_any().downcast_ref::().unwrap(); + assert_eq!(ports.values(), &[1611, 443, 0]); + } + + /// Real CIC headers, both spellings. Only the columns that must resolve. + const CIC_A_HEADER: &str = "Flow ID,Source IP,Source Port,Destination IP,Destination Port,\ + Protocol,Timestamp,Flow Duration,Total Fwd Packets,\ + Total Backward Packets,Total Length of Fwd Packets,\ + Total Length of Bwd Packets,Label"; + const CIC_B_HEADER: &str = "Flow ID,Src IP,Src Port,Dst IP,Dst Port,Protocol,Timestamp,\ + Flow Duration,Tot Fwd Pkts,Tot Bwd Pkts,TotLen Fwd Pkts,\ + TotLen Bwd Pkts,Label"; + + #[test] + fn both_cic_header_variants_resolve_every_required_field() { + for (name, header) in [("variant A", CIC_A_HEADER), ("variant B", CIC_B_HEADER)] { + let names: Vec = header.split(',').map(|s| s.to_string()).collect(); + let resolved = load_schema_spec().resolve_columns(&names); + for field in REQUIRED_FIELDS { + assert!( + resolved.contains_key(*field), + "CIC {name} should resolve {field}, got {resolved:?}" + ); + } + } + } + + #[test] + fn cic_byte_and_packet_columns_map_directionally() { + let a: Vec = CIC_A_HEADER.split(',').map(|s| s.to_string()).collect(); + let a = load_schema_spec().resolve_columns(&a); + assert_eq!( + a.get("fwd_bytes").map(String::as_str), + Some("Total Length of Fwd Packets") + ); + assert_eq!( + a.get("bwd_bytes").map(String::as_str), + Some("Total Length of Bwd Packets") + ); + + let b: Vec = CIC_B_HEADER.split(',').map(|s| s.to_string()).collect(); + let b = load_schema_spec().resolve_columns(&b); + assert_eq!( + b.get("fwd_bytes").map(String::as_str), + Some("TotLen Fwd Pkts") + ); + assert_eq!( + b.get("bwd_bytes").map(String::as_str), + Some("TotLen Bwd Pkts") + ); + assert_eq!(b.get("fwd_pkts").map(String::as_str), Some("Tot Fwd Pkts")); + assert_eq!(b.get("bwd_pkts").map(String::as_str), Some("Tot Bwd Pkts")); + } + + /// Variant A spells the packet counts `Total Fwd Packets` and `Total + /// Backward Packets` — *Backward*, not *Bwd*. Missing the second one left + /// every bwd_pkts null across 225,745 real rows. That resolves to a null + /// column rather than an error, so only inspecting converted data catches it. + #[test] + fn cic_variant_a_resolves_both_packet_directions() { + let names: Vec = CIC_A_HEADER.split(',').map(|s| s.to_string()).collect(); + let resolved = load_schema_spec().resolve_columns(&names); + assert_eq!( + resolved.get("fwd_pkts").map(String::as_str), + Some("Total Fwd Packets") + ); + assert_eq!( + resolved.get("bwd_pkts").map(String::as_str), + Some("Total Backward Packets") + ); + } + + /// `Flow Duration` is microseconds in CIC, but by decision it resolves as + /// seconds and `flow_check` catches the 10^6 inflation downstream. Pinning it + /// here so the behaviour is deliberate rather than incidental. + #[test] + fn cic_flow_duration_is_intentionally_treated_as_seconds() { + let spec = load_schema_spec(); + assert_eq!( + spec.duration_divisors.get("flow_duration").copied(), + Some(1.0), + "CIC Flow Duration is microseconds; treating it as seconds is a \ + deliberate decision, with flow_check's duration.implausible_magnitude \ + as the safety net" + ); + } + + fn order_of(values: &[&str]) -> Result> { + infer_slash_date_order(&StringArray::from(values.to_vec())) + } + + #[test] + fn proves_slash_date_order_from_evidence() { + // 20 > 12 can only be a day -> day-first. + assert_eq!( + order_of(&["20/11/2020 09:50:19"]).unwrap(), + Some(SlashDateOrder::DayFirst) + ); + // 31 in the second position can only be a day -> month-first. + assert_eq!( + order_of(&["7/31/2017 3:30"]).unwrap(), + Some(SlashDateOrder::MonthFirst) + ); + // Evidence anywhere in the column settles the whole column. + assert_eq!( + order_of(&["1/2/2020", "3/4/2020", "20/5/2020"]).unwrap(), + Some(SlashDateOrder::DayFirst) + ); + } + + #[test] + fn assumes_month_first_only_when_unprovable() { + // A CICIDS2017 per-day export: one date, nothing over 12. + assert_eq!( + order_of(&["7/7/2017 3:30", "7/7/2017 15:45:09"]).unwrap(), + Some(SlashDateOrder::MonthFirst) + ); + // No slash dates at all -> nothing inferred, nothing logged. + assert_eq!( + order_of(&["1750000000", "2011-08-18 10:00:00"]).unwrap(), + None + ); + // Contradictory evidence is an error, not a coin flip. + assert!(order_of(&["20/11/2020", "7/31/2017"]).is_err()); + } + + #[test] + fn rewrites_ambiguous_slash_dates_with_padding() { + use SlashDateOrder::{DayFirst, MonthFirst}; + // Single-digit hour, no seconds (CICIDS2017). + assert_eq!( + rewrite_slash_date("7/7/2017 3:30", MonthFirst).unwrap(), + "2017-07-07 03:30:00" + ); + // Full time (CICFlowMeter v4), day-first. + assert_eq!( + rewrite_slash_date("20/11/2020 09:50:19", DayFirst).unwrap(), + "2020-11-20 09:50:19" + ); + // The same string under the two orderings must differ. + assert_eq!( + rewrite_slash_date("7/5/2017", MonthFirst).unwrap(), + "2017-07-05 00:00:00" + ); + assert_eq!( + rewrite_slash_date("7/5/2017", DayFirst).unwrap(), + "2017-05-07 00:00:00" + ); + // Year-first is NOT claimed here — normalize_slash_date owns that shape. + assert!(rewrite_slash_date("2011/08/18 10:21:46.633335", MonthFirst).is_none()); + assert!(rewrite_slash_date("1750000000", MonthFirst).is_none()); + } + + #[test] + fn rewrites_year_first_slash_dates() { + // Argus StartTime, microsecond precision preserved. + assert_eq!( + normalize_slash_date("2011/08/18 10:21:46.633335"), + "2011-08-18 10:21:46.633335" + ); + assert_eq!(normalize_slash_date("2011/08/18"), "2011-08-18"); + } + + #[test] + fn leaves_ambiguous_and_already_valid_dates_alone() { + // Already dash-separated. + assert_eq!( + normalize_slash_date("2011-08-18 10:21:46"), + "2011-08-18 10:21:46" + ); + // RFC3339. + assert_eq!( + normalize_slash_date("2011-08-18T10:21:46Z"), + "2011-08-18T10:21:46Z" + ); + // Day-first / month-first are ambiguous — must not be rewritten, or we + // would risk silently transposing month and day. + assert_eq!(normalize_slash_date("18/08/2011"), "18/08/2011"); + assert_eq!(normalize_slash_date("08/18/2011"), "08/18/2011"); + // Epoch strings and junk pass through untouched. + assert_eq!(normalize_slash_date("1750000000"), "1750000000"); + assert_eq!(normalize_slash_date(""), ""); + assert_eq!(normalize_slash_date("2011/8/1"), "2011/8/1"); + } +} diff --git a/src/schema.rs b/src/schema.rs index 8198c9f..8663baa 100644 --- a/src/schema.rs +++ b/src/schema.rs @@ -180,12 +180,31 @@ pub fn normalize_name(name: &str) -> String { name.trim().to_lowercase().replace([' ', '-'], "_") } +/// Resolve a protocol name (or a numeric string) to its IANA protocol number. +/// +/// NOTE: this lowercases and trims but deliberately does NOT use +/// `normalize_name`, so hyphens survive. Keys must be written as the exporter +/// spells them — `"ipv6-icmp"`, not `"ipv6_icmp"`, which would never match. +/// +/// `None` means "no IANA IP protocol number", which covers two different cases: +/// an unrecognized string, and a value that is genuinely not an IP protocol. +/// Argus reports link-layer and application protocols in the same column — +/// `arp` and `rarp` are layer 2 (EtherType, not IP), while `rtp`, `rtcp` and +/// `udt` ride over UDP and have no protocol number of their own. Those are +/// deliberately left unmapped; do not invent numbers for them. They surface +/// downstream as a null protocol alongside a zero port, which is a detectable +/// signature for a data-quality check rather than something to paper over here. pub fn protocol_number(name: &str) -> Option { match name.trim().to_lowercase().as_str() { "tcp" => Some(6), "udp" => Some(17), "icmp" => Some(1), - "icmpv6" => Some(58), + // 58 has three spellings in the wild: IANA's official `ipv6-icmp`, and + // the colloquial `icmpv6`/`icmp6`. Real Argus captures use the first. + "ipv6-icmp" | "icmpv6" | "icmp6" => Some(58), + "igmp" => Some(2), + "ipv6" => Some(41), + "pim" => Some(103), "gre" => Some(47), "esp" => Some(50), other => other.parse::().ok(), @@ -211,3 +230,56 @@ impl SchemaSpec { resolved } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resolves_ip_protocol_names_and_spellings() { + assert_eq!(protocol_number("tcp"), Some(6)); + assert_eq!(protocol_number("UDP"), Some(17)); + assert_eq!(protocol_number(" TCP "), Some(6), "values arrive padded"); + assert_eq!(protocol_number("igmp"), Some(2)); + assert_eq!(protocol_number("ipv6"), Some(41)); + assert_eq!(protocol_number("pim"), Some(103)); + // All three spellings of 58, including IANA's hyphenated official name. + for spelling in ["ipv6-icmp", "icmpv6", "icmp6", "IPv6-ICMP"] { + assert_eq!(protocol_number(spelling), Some(58), "{spelling}"); + } + // Numeric passthrough for exporters that already emit the number. + assert_eq!(protocol_number("6"), Some(6)); + assert_eq!(protocol_number("132"), Some(132)); + } + + /// Argus reports link-layer and application protocols in the same column. + /// None of these has an IANA IP protocol number, so `None` is the correct + /// answer — not a gap to be filled with an invented value. + #[test] + fn leaves_non_ip_protocols_unmapped() { + for not_ip in ["arp", "rarp", "rtp", "rtcp", "udt", "ipx/spx"] { + assert_eq!( + protocol_number(not_ip), + None, + "{not_ip} is not an IP protocol and must not be given a number" + ); + } + assert_eq!(protocol_number(""), None); + assert_eq!(protocol_number("nonsense"), None); + } + + /// `protocol_number` lowercases and trims but does NOT apply + /// `normalize_name`, so hyphens survive. A key written with an underscore + /// would never match what an exporter emits. + #[test] + fn protocol_lookup_does_not_normalize_hyphens() { + assert_eq!(normalize_name("ipv6-icmp"), "ipv6_icmp"); + assert_eq!(protocol_number("ipv6-icmp"), Some(58)); + assert_eq!( + protocol_number("ipv6_icmp"), + None, + "the underscored form is not what exporters write; \ + new entries must use the hyphenated spelling" + ); + } +} diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 9839767..fd5fdf7 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -62,8 +62,11 @@ def packet(src, dst, sport, dport, proto_cls, payload): ) -def build_cicids_csv(path): - """CSV with CICIDS-style headers: spaces, mixed case, ms durations.""" +def build_generic_flow_csv(path): + """Generic aliased flow CSV: leading spaces in header names, an explicitly + ms-named duration, and epoch-seconds timestamps. Not a real vendor format — + it guards the alias/trim/unit machinery in isolation. Real CIC headers are + covered by build_cic_csv below.""" rows = [ "Source IP, Destination IP, Source Port, Destination Port, Flow Duration_Milliseconds, Total Fwd Bytes, Total Bwd Bytes, Protocol, Timestamp, Label", "192.168.1.5,10.9.9.9,51000,80,2500,1200,34000,tcp,1750000000,BENIGN", @@ -73,6 +76,90 @@ def build_cicids_csv(path): f.write("\n".join(rows)) +def build_cic_csv(path, variant): + """Real CICFlowMeter headers, verbatim, in both spellings that exist in the + wild. The previous fixture claimed to be "CICIDS-style" but invented + `Total Fwd Bytes` and `Flow Duration_Milliseconds` — names no CIC export + actually emits, which both happened to already resolve. That is why the real + CIC gap went unnoticed. + + variant "A" = original CICIDS2017/2018 (`Source IP`, `Total Length of Fwd + Packets`, M/D/Y date with a single-digit hour and no seconds). + variant "B" = CICFlowMeter v4 (`Src IP`, `TotLen Fwd Pkts`, D/M/Y date that + proves its own ordering because 20 > 12, float byte counts). + + Flow Duration is MICROSECONDS in both. By decision it is passed through as + seconds, so flow_check catches the 10^6 inflation downstream — see the + assertions, which pin that on purpose. + """ + if variant == "A": + rows = [ + "Flow ID,Source IP,Source Port,Destination IP,Destination Port,Protocol,Timestamp,Flow Duration,Total Fwd Packets,Total Backward Packets,Total Length of Fwd Packets,Total Length of Bwd Packets,Label", + "x-1,104.16.207.165,443,192.168.10.5,54865,6,7/7/2017 3:30,3,2,0,12,0,BENIGN", + "x-2,192.168.10.5,54865,104.16.207.165,443,6,7/7/2017 15:45:09,5000000,4,2,600,1200,DDoS", + ] + else: + rows = [ + "Flow ID,Src IP,Src Port,Dst IP,Dst Port,Protocol,Timestamp,Flow Duration,Tot Fwd Pkts,Tot Bwd Pkts,TotLen Fwd Pkts,TotLen Bwd Pkts,Label", + "y-1,10.0.0.1,60602,10.0.1.2,5051,6,20/11/2020 09:50:19,38031,3,4,117.0,353.0,No Label", + "y-2,10.0.0.2,443,10.0.1.3,51000,17,20/11/2020 09:50:20,120,1,0,64.0,0.0,Attack", + ] + with open(path, "w") as f: + f.write("\n".join(rows)) + + +def build_argus_binetflow(path): + """Argus/CTU-13 `.binetflow`: verbatim real header, `YYYY/MM/DD` StartTime, + only a *forward* byte counter (SrcBytes) alongside two totals, plus the two + port shapes real Argus emits that are not numbers — hex ICMP type/code and + an empty field for a protocol that carries no ports.""" + rows = [ + "StartTime,Dur,Proto,SrcAddr,Sport,Dir,DstAddr,Dport,State,sTos,dTos,TotPkts,TotBytes,SrcBytes,Label", + "2011/08/18 10:21:46.633335,1.060248,tcp,93.45.239.29,1611, ->,147.32.84.118,6881,S_RA,0,0,4,252,132,flow=Background-TCP-Attempt", + "2011/08/18 10:19:49.027650,279.349152,udp,62.240.166.118,1031, ,147.32.84.229,13363,SRPA_PA,0,0,15,1318,955,flow=From-Botnet-V42-TCP", + "2011/08/18 10:22:07.160628,0.000000,icmp,147.32.84.59,0x0303, ->,147.32.80.9,0x4fa8,URP,0,0,1,70,70,flow=Background", + "2011/08/18 10:23:01.000000,0.000000,arp,147.32.84.59,, ->,147.32.85.1,,CON,0,0,1,42,42,flow=Background", + ] + with open(path, "w") as f: + f.write("\n".join(rows)) + + +def build_cidds_csv(path): + """CIDDS-001/002 as emitted by nfdump: space-padded fields, `Date first seen`, + a `class` label column (the OpenStack/ExternalServer captures use `class`, the + traffic__week* ones use `label`), an nfdump magnitude suffix in Bytes, and an + ICMP row that puts type.code in Dst Pt as a decimal.""" + rows = [ + "Date first seen,Duration,Proto,Src IP Addr,Src Pt,Dst IP Addr,Dst Pt,Packets,Bytes,Flows,Flags,Tos,class,attackType,attackID,attackDescription", + "2017-08-02 00:00:00.419, 0.003,TCP ,192.168.210.55, 44870,192.168.100.11, 445, 2, 174, 1,.AP..., 0,normal,---,---,---", + "2017-08-02 00:00:01.000, 12.500,TCP ,192.168.220.47, 55101,192.168.100.11, 445, 1500, 1.4 M, 1,.AP..., 0,attacker,dos,1,---", + "2017-08-02 00:00:02.000, 0.000,ICMP ,192.168.220.16, 0,192.168.100.5, 8.0, 1, 92, 1,......, 0,victim,---,---,---", + ] + with open(path, "w") as f: + f.write("\n".join(rows)) + + +def build_zeek_conn_log(path): + """Zeek `conn.log.labeled` (ctu-sme-11 shape): tab-separated, schema declared + in a `#` preamble rather than a header row, `-` for fields Zeek did not + measure, and dotted field names (`id.orig_h`) that normalization leaves + intact. Row 2 is the unset case — 15% of real rows look like this, with + packet counts present but duration and bytes absent.""" + rows = [ + "#separator \\x09", + "#set_separator\t,", + "#empty_field\t(empty)", + "#unset_field\t-", + "#path\tconn", + "#fields\tts\tuid\tid.orig_h\tid.orig_p\tid.resp_h\tid.resp_p\tproto\tservice\tduration\torig_bytes\tresp_bytes\tconn_state\torig_pkts\tresp_pkts\tlabel\tdetailedlabel", + "#types\ttime\tstring\taddr\tport\taddr\tport\tenum\tstring\tinterval\tcount\tcount\tstring\tcount\tcount\tstring\tstring", + "1677110378.922531\tCUfDVH\t192.168.1.108\t138\t192.168.1.255\t138\tudp\t-\t2.5\t1200\t340\tSF\t8\t4\tBenign\tFrom_benign-To_benign", + "1677110379.100000\tCAwxab\t192.168.1.108\t54517\t1.1.1.1\t53\tudp\tdns\t-\t-\t-\tS0\t1\t0\tMalicious\tFrom_malicious", + ] + with open(path, "w") as f: + f.write("\n".join(rows) + "\n") + + def build_ocsf_ndjson(path): """OCSF Network Activity NDJSON: nested fields, ms units, a non-close event.""" rows = [ @@ -98,7 +185,12 @@ def build_ocsf_ndjson(path): def main(): build_test_pcap("/tmp/flowprep_test.pcap") - build_cicids_csv("/tmp/flowprep_test.csv") + build_generic_flow_csv("/tmp/flowprep_test.csv") + build_argus_binetflow("/tmp/flowprep_test.binetflow") + build_cidds_csv("/tmp/flowprep_cidds.csv") + build_cic_csv("/tmp/flowprep_cic_a.csv", "A") + build_cic_csv("/tmp/flowprep_cic_b.csv", "B") + build_zeek_conn_log("/tmp/flowprep_zeek_conn.log") build_ocsf_ndjson("/tmp/flowprep_test.ndjson") r = subprocess.run( @@ -131,6 +223,156 @@ def main(): assert rows[1]["timestamp"] == 1750000060_000000, "epoch-seconds detection wrong" assert rows[0]["label"] == "BENIGN", "label passthrough wrong" + # Argus `.binetflow`: exercises content-based reader detection (the file is + # comma-delimited text but is NOT named .csv), the YYYY/MM/DD StartTime + # rewrite, and the deliberate choice to map only SrcBytes. + r = subprocess.run( + [FLOWPREP_BIN, "canonicalize", "/tmp/flowprep_test.binetflow", "/tmp/flowprep_argus.parquet"], + capture_output=True, text=True, + ) + print(r.stdout.strip(), r.stderr.strip()) + assert r.returncode == 0, f"binetflow conversion failed: {r.stderr.strip()}" + + t = pq.read_table("/tmp/flowprep_argus.parquet") + print(t.to_pydict()) + rows = t.to_pylist() + assert t.num_rows == 4, f"expected 4 flows, got {t.num_rows}" + # StartTime "2011/08/18 10:21:46.633335" -> epoch us, sub-second preserved. + assert rows[0]["timestamp"] == 1313662906633335, f"slash-date parse wrong: {rows[0]['timestamp']}" + assert rows[0]["src_ip"] == "93.45.239.29" and rows[0]["dest_ip"] == "147.32.84.118" + assert rows[0]["src_port"] == 1611 and rows[0]["dest_port"] == 6881 + assert rows[0]["flow_dur"] == 1.060248, f"Dur should be seconds: {rows[0]['flow_dur']}" + assert rows[0]["protocol"] == 6 and rows[1]["protocol"] == 17, "Proto name mapping wrong" + # Argus label values keep their `flow=` prefix; interpreting them is the + # consumer's job, not flowprep's. + assert rows[0]["label"] == "flow=Background-TCP-Attempt", "label passthrough wrong" + # fwd_bytes comes from SrcBytes (132), NOT TotBytes (252). TotBytes/TotPkts + # are totals, not directional, so they are deliberately unmapped: bwd_bytes + # zero-fills and the packet counts stay null rather than carrying a wrong + # value. Recovering backward bytes needs TotBytes - SrcBytes, which requires + # a derived-field mechanism flowprep does not have. + assert rows[0]["fwd_bytes"] == 132, f"fwd_bytes should be SrcBytes: {rows[0]['fwd_bytes']}" + assert rows[0]["bwd_bytes"] == 0, "TotBytes must not be read as bwd_bytes" + assert rows[0]["fwd_pkts"] is None, "TotPkts must not be read as fwd_pkts" + assert rows[0]["bwd_pkts"] is None + # Ports that are not numbers become 0 rather than failing the file: hex ICMP + # type/code (row 3) and an empty field on a portless protocol (row 4). + # canonical src_port/dest_port are non-nullable, so 0 is the sentinel — the + # same one the nfcapd reader already emits for ICMP. + assert rows[2]["src_port"] == 0 and rows[2]["dest_port"] == 0, "hex ICMP port should coerce to 0" + assert rows[3]["src_port"] == 0 and rows[3]["dest_port"] == 0, "empty port should coerce to 0" + assert all(r["src_port"] is not None for r in rows), "ports are non-nullable" + + # CIDDS: nfdump-style export. Exercises the 5 CIDDS aliases, the `class` + # label passthrough, nfdump magnitude suffixes in Bytes, and a decimal + # ICMP type.code in Dst Pt. + r = subprocess.run( + [FLOWPREP_BIN, "canonicalize", "/tmp/flowprep_cidds.csv", "/tmp/flowprep_cidds.parquet"], + capture_output=True, text=True, + ) + print(r.stdout.strip(), r.stderr.strip()) + assert r.returncode == 0, f"CIDDS conversion failed: {r.stderr.strip()}" + + t = pq.read_table("/tmp/flowprep_cidds.parquet") + print(t.to_pydict()) + rows = t.to_pylist() + assert t.num_rows == 3, f"expected 3 flows, got {t.num_rows}" + assert rows[0]["timestamp"] == 1501632000419000, f"Date first seen wrong: {rows[0]['timestamp']}" + assert rows[0]["src_ip"] == "192.168.210.55" and rows[0]["dest_ip"] == "192.168.100.11" + assert rows[0]["src_port"] == 44870 and rows[0]["dest_port"] == 445, "Src Pt/Dst Pt wrong" + assert rows[0]["flow_dur"] == 0.003, f"Duration should be seconds: {rows[0]['flow_dur']}" + assert rows[0]["protocol"] == 6, "space-padded 'TCP ' should map to 6" + # CIDDS rows are unidirectional, so its single Bytes total IS that row's + # forward volume; bwd zero-fills correctly. + assert rows[0]["fwd_bytes"] == 174 and rows[0]["bwd_bytes"] == 0 + # nfdump magnitude suffix: "1.4 M" -> 1_400_000, not null. + assert rows[1]["fwd_bytes"] == 1_400_000, f"suffix expansion wrong: {rows[1]['fwd_bytes']}" + # ICMP type.code "8.0" in Dst Pt is not a port -> coerced to 0. + assert rows[2]["dest_port"] == 0, "decimal ICMP type.code should coerce to 0" + # `class` survives as ground truth. + assert rows[0]["class"] == "normal" and rows[1]["class"] == "attacker" + assert rows[1]["attacktype"] == "dos", "attackType should survive as attacktype" + + # CIC variant A — original CICIDS2017/2018 spelling. Date order cannot be + # proven from the data (7/7/2017 has no component over 12), so it is assumed + # M/D/Y and that assumption is logged. + r = subprocess.run( + [FLOWPREP_BIN, "canonicalize", "/tmp/flowprep_cic_a.csv", "/tmp/flowprep_cic_a.parquet"], + capture_output=True, text=True, + ) + print(r.stdout.strip(), r.stderr.strip()) + assert r.returncode == 0, f"CIC variant A failed: {r.stderr.strip()}" + assert "ASSUMED M/D/YYYY" in r.stderr, "an unprovable date order must be logged as assumed" + + t = pq.read_table("/tmp/flowprep_cic_a.parquet") + print(t.to_pydict()) + rows = t.to_pylist() + assert rows[0]["src_ip"] == "104.16.207.165" and rows[0]["dest_port"] == 54865 + assert rows[0]["fwd_bytes"] == 12, "Total Length of Fwd Packets -> fwd_bytes" + assert rows[1]["fwd_bytes"] == 600 and rows[1]["bwd_bytes"] == 1200 + assert rows[0]["fwd_pkts"] == 2, "Total Fwd Packets -> fwd_pkts" + # 7/7/2017 3:30 read month-first, single-digit hour zero-padded, no seconds. + assert rows[0]["timestamp"] == 1499398200000000, f"date rewrite wrong: {rows[0]['timestamp']}" + assert rows[1]["timestamp"] == 1499442309000000, "HH:MM:SS form should parse too" + # Flow Duration is MICROSECONDS but is deliberately passed through as + # seconds; flow_check's duration.implausible_magnitude catches the 10^6 + # inflation downstream. This assertion pins that decision on purpose — + # do NOT "fix" it to 5.0 without revisiting it. + assert rows[1]["flow_dur"] == 5_000_000.0, f"duration must pass through: {rows[1]['flow_dur']}" + assert rows[0]["label"] == "BENIGN" + + # CIC variant B — CICFlowMeter v4 spelling. 20/11/2020 proves D/M/Y. + r = subprocess.run( + [FLOWPREP_BIN, "canonicalize", "/tmp/flowprep_cic_b.csv", "/tmp/flowprep_cic_b.parquet"], + capture_output=True, text=True, + ) + print(r.stdout.strip(), r.stderr.strip()) + assert r.returncode == 0, f"CIC variant B failed: {r.stderr.strip()}" + assert "D/M/YYYY (proven" in r.stderr, "20/11/2020 should prove day-first ordering" + + t = pq.read_table("/tmp/flowprep_cic_b.parquet") + print(t.to_pydict()) + rows = t.to_pylist() + assert rows[0]["src_ip"] == "10.0.0.1" and rows[0]["src_port"] == 60602 + # Float byte counts round to i64. + assert rows[0]["fwd_bytes"] == 117 and rows[0]["bwd_bytes"] == 353 + assert rows[0]["fwd_pkts"] == 3 and rows[0]["bwd_pkts"] == 4, "Tot Fwd/Bwd Pkts" + # 20 November 2020 09:50:19 UTC, NOT 11 August (which month-first would give). + assert rows[0]["timestamp"] == 1605865819000000, f"D/M/Y parse wrong: {rows[0]['timestamp']}" + assert rows[0]["label"] == "No Label" + + # Zeek conn.log.labeled: schema comes from the `#` preamble, not a header + # row, and unset numeric fields ("-") are substituted with 0. + r = subprocess.run( + [FLOWPREP_BIN, "canonicalize", "/tmp/flowprep_zeek_conn.log", "/tmp/flowprep_zeek.parquet"], + capture_output=True, text=True, + ) + print(r.stdout.strip(), r.stderr.strip()) + assert r.returncode == 0, f"Zeek conn.log conversion failed: {r.stderr.strip()}" + assert "substituted 0 for 3 unset" in r.stderr, "unset duration/orig_bytes/resp_bytes counted" + + t = pq.read_table("/tmp/flowprep_zeek.parquet") + print(t.to_pydict()) + rows = t.to_pylist() + assert t.num_rows == 2, f"expected 2 flows, got {t.num_rows}" + # ts is an epoch double, so the magnitude path converts it to microseconds. + assert rows[0]["timestamp"] == 1677110378922531, f"epoch ts wrong: {rows[0]['timestamp']}" + # Dotted field names resolve literally: id.orig_h/id.resp_p etc. + assert rows[0]["src_ip"] == "192.168.1.108" and rows[0]["dest_ip"] == "192.168.1.255" + assert rows[0]["src_port"] == 138 and rows[0]["dest_port"] == 138 + assert rows[0]["flow_dur"] == 2.5, "Zeek duration is already seconds" + assert rows[0]["fwd_bytes"] == 1200 and rows[0]["bwd_bytes"] == 340 + assert rows[0]["fwd_pkts"] == 8 and rows[0]["bwd_pkts"] == 4 + assert rows[0]["protocol"] == 17 + # Row 2: duration/orig_bytes/resp_bytes were "-" -> 0, packets still real. + # That leaves zero bytes against non-zero packets, which is exactly what + # flow_check's bytes.zero_with_packets reports — visible, not silent. + assert rows[1]["flow_dur"] == 0.0 and rows[1]["fwd_bytes"] == 0 + assert rows[1]["fwd_pkts"] == 1, "packet counts must survive the substitution" + # Both label columns come through. + assert rows[1]["label"] == "Malicious" + assert rows[0]["detailedlabel"] == "From_benign-To_benign" + r = subprocess.run( [FLOWPREP_BIN, "ocsf", "/tmp/flowprep_test.ndjson", "/tmp/flowprep_ocsf.parquet"], capture_output=True, text=True,