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,