diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 599e19b..6bbd951 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,5 +59,7 @@ jobs: python -m pip install --upgrade pip python -m pip install dpkt pyarrow - - name: Run e2e smoke test - run: python tests/test_e2e.py + - name: Run e2e smoke tests + run: | + python tests/test_e2e.py + python tests/test_modbus_e2e.py diff --git a/README.md b/README.md index 1c29708..cb2a69b 100644 --- a/README.md +++ b/README.md @@ -66,25 +66,30 @@ cargo build --release ./demo.sh ``` -Five subcommands: +Six subcommands: ```bash # 1. Raw packet captures -> bidirectional flow records flowprep pcap capture.pcap flows.parquet -# 2. Any aliased flow table (CSV, parquet, Zeek TSV log, Argus .binetflow) +# 2. Passively decode Modbus/TCP from a packet capture -> protocol observations +flowprep modbus capture.pcap modbus.parquet +# Non-standard server ports are explicit, never guessed +flowprep modbus capture.pcap modbus.parquet --server-port 1502 + +# 3. Any aliased flow table (CSV, parquet, Zeek TSV log, Argus .binetflow) # -> the canonical schema flowprep canonicalize cic_export.csv flows.parquet flowprep canonicalize conn.log.labeled flows.parquet flowprep canonicalize capture.binetflow flows.parquet -# 3. OCSF Network Activity events (JSON/NDJSON) -> the canonical schema +# 4. OCSF Network Activity events (JSON/NDJSON) -> the canonical schema flowprep ocsf network_activity.ndjson flows.parquet -# 4. nfdump/nfcapd binary flow files -> the canonical schema +# 5. nfdump/nfcapd binary flow files -> the canonical schema flowprep nfcapd nfcapd.202401011200 flows.parquet -# 5. Inspect any parquet file from the terminal, no Python required +# 6. Inspect any parquet file from the terminal, no Python required flowprep peek flows.parquet -n 20 ``` @@ -164,6 +169,35 @@ timeout and a 1h maximum duration. The reader streams pcap and pcapng, keeps constant memory on the packet path, and is robust to the slightly-out-of-order packets real captures contain. +### Passive Modbus/TCP decoding + +`modbus` reads pcap or pcapng offline and writes a separate, versioned +`modbus_observation/v1` table. It does not poll devices or put traffic on an OT +network. The decoder reassembles in-order and modestly out-of-order TCP segments, +suppresses retransmissions, splits coalesced application data units, and pairs +requests with responses using the TCP conversation, transaction identifier, and +unit identifier. Missing requests or responses remain explicit observations; +protocol exceptions and parser recovery warnings are not collapsed into success. + +The output includes client/server endpoints, unit and function identifiers, +read/write operation, address and quantity fields where the function defines +them, response latency/status, and capture packet references. Requested write +values are retained as ordered `coil_values` for functions 5 and 15 and ordered +`register_values` for functions 6, 16, and 23. These values record attempted +intent; use `response_status` to distinguish acknowledged writes from exceptions +or missing responses. Function 23 values start at `write_address`; the other +supported writes start at `address`. + +Read Device Identification responses (function 43/MEI 14) populate vendor, +product, revision, and model fields when those objects are actually present on +the wire. Device identity is evidence, not an inference: absent objects stay +null. Read-response process values and raw PDUs are deliberately excluded from +v1. + +Direction is based only on the configured server port (502 by default). Set +`--server-port` for a known non-standard deployment; the decoder does not guess +roles from payload content. + ### Zeek logs and research exports `canonicalize` also reads **Zeek TSV logs** (`conn.log`, including labeled @@ -197,6 +231,9 @@ convert once, train immediately. Plus any passthrough label columns present in the source. +The Modbus protocol-observation contract is separate from canonical NetFlow and +lives at [`schemas/modbus/v1/schema.json`](schemas/modbus/v1/schema.json). + ## Example: a real research dataset The repo ships a ~100k-row slice of **CIC-IDS-2017** with its original @@ -278,6 +315,7 @@ cargo build --release # end-to-end tests (python harness generates fixtures; needs dpkt + pyarrow) python3 -m venv .venv && .venv/bin/pip install dpkt pyarrow .venv/bin/python tests/test_e2e.py +.venv/bin/python tests/test_modbus_e2e.py # throughput benchmark .venv/bin/python tests/bench_pcap.py diff --git a/schemas/modbus/v1/schema.json b/schemas/modbus/v1/schema.json new file mode 100644 index 0000000..3a5d7ef --- /dev/null +++ b/schemas/modbus/v1/schema.json @@ -0,0 +1,55 @@ +{ + "schema_version": "modbus_observation/v1", + "description": "Passive Modbus/TCP request-response observations decoded from packet captures. Requested write values are retained; read-response values and raw PDUs are intentionally excluded from v1.", + "direction": { + "basis": "configured_server_port", + "default_server_port": 502, + "notes": "Traffic to the configured port is a request; traffic from it is a response. No role inference is presented as discovered device identity." + }, + "timestamp_unit": "epoch_microseconds", + "fields": [ + { "name": "schema_version", "type": "utf8", "nullable": false }, + { "name": "timestamp", "type": "int64", "nullable": false, "description": "Timestamp of the packet that completed the first observed ADU for this transaction." }, + { "name": "client_ip", "type": "utf8", "nullable": false }, + { "name": "client_port", "type": "int32", "nullable": false }, + { "name": "server_ip", "type": "utf8", "nullable": false }, + { "name": "server_port", "type": "int32", "nullable": false }, + { "name": "transaction_id", "type": "int32", "nullable": false }, + { "name": "unit_id", "type": "int32", "nullable": false, "description": "Modbus unit identifier; important when a TCP endpoint fronts multiple downstream devices." }, + { "name": "function_code", "type": "int32", "nullable": false }, + { "name": "function_name", "type": "utf8", "nullable": false }, + { "name": "operation", "type": "utf8", "nullable": false, "values": ["read", "write", "read_write", "diagnostic", "device_identification", "encapsulated", "other"] }, + { "name": "address", "type": "int32", "nullable": true, "description": "Primary zero-based protocol address when the request function carries one." }, + { "name": "quantity", "type": "int32", "nullable": true }, + { "name": "write_address", "type": "int32", "nullable": true, "description": "Write-side address for function 23." }, + { "name": "write_quantity", "type": "int32", "nullable": true, "description": "Write-side quantity for function 23." }, + { "name": "coil_values", "type": "list", "nullable": true, "description": "Requested coil values in address order for functions 5 and 15. The base is address. Null means not applicable or not decoded; presence records requested intent, not confirmed application." }, + { "name": "register_values", "type": "list", "nullable": true, "description": "Requested unsigned 16-bit register values in address order for functions 6, 16, and 23. The base is address, except function 23 uses write_address. Null means not applicable or not decoded; presence records requested intent, not confirmed application." }, + { "name": "diagnostic_subfunction", "type": "int32", "nullable": true }, + { "name": "device_id_code", "type": "int32", "nullable": true }, + { "name": "device_id_object", "type": "int32", "nullable": true }, + { "name": "request_seen", "type": "boolean", "nullable": false }, + { "name": "response_seen", "type": "boolean", "nullable": false }, + { "name": "request_timestamp", "type": "int64", "nullable": true }, + { "name": "response_timestamp", "type": "int64", "nullable": true }, + { "name": "latency_usec", "type": "int64", "nullable": true }, + { "name": "response_status", "type": "utf8", "nullable": false, "values": ["ok", "exception", "missing_response", "orphan_response", "function_mismatch"] }, + { "name": "exception_code", "type": "int32", "nullable": true }, + { "name": "exception_name", "type": "utf8", "nullable": true }, + { "name": "vendor_name", "type": "utf8", "nullable": true, "source": "FC 43/14 object 0" }, + { "name": "product_code", "type": "utf8", "nullable": true, "source": "FC 43/14 object 1" }, + { "name": "revision", "type": "utf8", "nullable": true, "source": "FC 43/14 object 2" }, + { "name": "vendor_url", "type": "utf8", "nullable": true, "source": "FC 43/14 object 3" }, + { "name": "product_name", "type": "utf8", "nullable": true, "source": "FC 43/14 object 4" }, + { "name": "model_name", "type": "utf8", "nullable": true, "source": "FC 43/14 object 5" }, + { "name": "user_application_name", "type": "utf8", "nullable": true, "source": "FC 43/14 object 6" }, + { "name": "request_packet", "type": "int64", "nullable": true, "description": "One-based capture packet number in which the request ADU became decodable." }, + { "name": "response_packet", "type": "int64", "nullable": true, "description": "One-based capture packet number in which the response ADU became decodable." }, + { "name": "direction_basis", "type": "utf8", "nullable": false }, + { "name": "parser_warning", "type": "utf8", "nullable": true, "description": "Semicolon-delimited, machine-stable parser caveats attached to this observation." } + ], + "excluded_from_v1": [ + "read_response_values", + "raw_pdu" + ] +} diff --git a/src/main.rs b/src/main.rs index bb6c57c..e162e2b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,5 @@ mod canonicalize; +mod modbus; mod nfcapd; mod nfdump; mod ocsf; @@ -22,6 +23,14 @@ struct Cli { enum Command { /// pcap/pcapng -> canonical flow parquet Pcap { input: String, output: String }, + /// passively decode Modbus/TCP from pcap/pcapng -> protocol observations + Modbus { + input: String, + output: String, + /// TCP port used by Modbus servers in this capture + #[arg(long, default_value_t = 502)] + server_port: u16, + }, /// aliased parquet/CSV flow table -> canonical parquet Canonicalize { input: String, output: String }, /// OCSF Network Activity JSON/NDJSON -> canonical parquet @@ -53,6 +62,12 @@ fn main() { Command::Pcap { input, output } => { pcap::pcap_to_parquet(input, output).map(|n| println!("Wrote {n} flows to {output}")) } + Command::Modbus { + input, + output, + server_port, + } => modbus::modbus_to_parquet(input, output, *server_port) + .map(|summary| println!("Wrote Modbus {summary} to {output}")), Command::Canonicalize { input, output } => canonicalize::canonicalize_file(input, output) .map(|n| println!("Wrote {n} flows to {output}")), Command::Ocsf { input, output } => { diff --git a/src/modbus.rs b/src/modbus.rs new file mode 100644 index 0000000..8ca7869 --- /dev/null +++ b/src/modbus.rs @@ -0,0 +1,1715 @@ +//! Passive Modbus/TCP decoding from offline PCAP/PCAPNG captures. +//! +//! This module deliberately writes a protocol-observation schema rather than +//! adding application fields to canonical NetFlow. Direction is inferred only +//! from a configured server port (502 by default); the decoder never connects +//! to, polls, or otherwise interacts with an OT device. + +use std::collections::{BTreeMap, HashMap}; +use std::fs::File; +use std::net::{Ipv4Addr, Ipv6Addr}; +use std::sync::Arc; + +use arrow::array::{ + ArrayRef, BooleanArray, BooleanBuilder, Int32Array, Int32Builder, Int64Array, ListBuilder, + StringArray, +}; +use arrow::datatypes::{DataType, Field, Schema}; +use arrow::error::ArrowError; +use arrow::record_batch::RecordBatch; +use etherparse::{NetSlice, SlicedPacket, TransportSlice}; +use pcap_parser::{Block, PcapBlockOwned, PcapError, create_reader}; +use serde_json::Value; + +use crate::writer::write_parquet; + +type Result = std::result::Result>; + +const LINKTYPE_ETHERNET: u16 = 1; +const MAX_MODBUS_LENGTH: usize = 254; // unit identifier + PDU +const MAX_PENDING_STREAM_BYTES: usize = 1024 * 1024; +const SCHEMA_VERSION: &str = "modbus_observation/v1"; +const DIRECTION_BASIS: &str = "configured_server_port"; + +const SCHEMA_JSON: &str = include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/schemas/modbus/v1/schema.json" +)); + +#[derive(Debug, Default, Clone)] +pub struct DecodeSummary { + pub observations: usize, + pub complete: usize, + pub request_only: usize, + pub response_only: usize, + pub exceptions: usize, + pub tcp_payload_packets: usize, + pub request_adus: usize, + pub response_adus: usize, + pub malformed_bytes: usize, + pub retransmitted_segments: usize, + pub out_of_order_segments: usize, + pub forced_stream_gaps: usize, + pub incomplete_streams: usize, +} + +impl std::fmt::Display for DecodeSummary { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "{} observations ({} complete, {} request-only, {} response-only, {} exceptions); \ + {} request/{} response ADUs; {} malformed bytes skipped; {} retransmissions, \ + {} out-of-order segments, {} forced gaps, {} incomplete streams", + self.observations, + self.complete, + self.request_only, + self.response_only, + self.exceptions, + self.request_adus, + self.response_adus, + self.malformed_bytes, + self.retransmitted_segments, + self.out_of_order_segments, + self.forced_stream_gaps, + self.incomplete_streams, + ) + } +} + +#[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord)] +struct ConversationKey { + client_ip: String, + client_port: u16, + server_ip: String, + server_port: u16, +} + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] +enum Direction { + Request, + Response, +} + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct StreamKey { + conversation: ConversationKey, + direction: Direction, +} + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct TransactionKey { + conversation: ConversationKey, + transaction_id: u16, + unit_id: u8, +} + +struct TcpPayloadPacket { + timestamp: i64, + packet_number: i64, + src_ip: String, + dest_ip: String, + src_port: u16, + dest_port: u16, + sequence_number: u32, + syn: bool, + fin: bool, + rst: bool, + payload: Vec, +} + +#[derive(Debug)] +struct RawAdu { + transaction_id: u16, + unit_id: u8, + pdu: Vec, + warning: Option, +} + +#[derive(Default)] +struct FeedReport { + retransmitted_segments: usize, + out_of_order_segments: usize, + forced_stream_gaps: usize, +} + +#[derive(Default)] +struct ExtractReport { + adus: Vec, + malformed_bytes: usize, +} + +#[derive(Default)] +struct TcpStreamState { + next_sequence: Option, + pending_segments: BTreeMap>, + decoded_bytes: Vec, + pending_warning: Option, +} + +impl TcpStreamState { + fn has_incomplete_data(&self) -> bool { + !self.pending_segments.is_empty() || !self.decoded_bytes.is_empty() + } + + fn feed(&mut self, sequence: u32, payload: &[u8]) -> FeedReport { + let mut report = FeedReport::default(); + if payload.is_empty() { + return report; + } + + let expected = *self.next_sequence.get_or_insert(sequence as u64); + let mut start = unwrap_sequence(sequence, expected); + let mut data = payload; + let end = start.saturating_add(data.len() as u64); + + if end <= expected { + report.retransmitted_segments += 1; + return report; + } + if start < expected { + report.retransmitted_segments += 1; + let overlap = (expected - start) as usize; + data = &data[overlap..]; + start = expected; + } else if start > expected { + report.out_of_order_segments += 1; + } + + self.pending_segments + .entry(start) + .and_modify(|existing| { + if data.len() > existing.len() { + existing.clear(); + existing.extend_from_slice(data); + } + }) + .or_insert_with(|| data.to_vec()); + self.drain_contiguous(); + + let pending_bytes: usize = self.pending_segments.values().map(Vec::len).sum(); + if pending_bytes > MAX_PENDING_STREAM_BYTES { + if let Some((&first_start, _)) = self.pending_segments.first_key_value() { + // A missing TCP segment must not allow unbounded memory growth. If + // the cap is reached, discard the incomplete ADU and resume at the + // next observed segment while marking the next record explicitly. + self.decoded_bytes.clear(); + self.next_sequence = Some(first_start); + add_warning(&mut self.pending_warning, "tcp_gap_forced_resync"); + report.forced_stream_gaps += 1; + self.drain_contiguous(); + } + } + report + } + + fn drain_contiguous(&mut self) { + while let Some(expected) = self.next_sequence { + let Some((&start, _)) = self.pending_segments.range(..=expected).next_back() else { + break; + }; + let data = self.pending_segments.remove(&start).unwrap(); + let end = start + data.len() as u64; + if end <= expected { + continue; + } + let offset = (expected - start) as usize; + self.decoded_bytes.extend_from_slice(&data[offset..]); + self.next_sequence = Some(end); + } + } + + fn extract_adus(&mut self) -> ExtractReport { + let mut report = ExtractReport::default(); + loop { + if self.decoded_bytes.len() < 7 { + break; + } + + if !plausible_mbap_header(&self.decoded_bytes, 0) { + if let Some(offset) = find_next_mbap_header(&self.decoded_bytes) { + self.decoded_bytes.drain(..offset); + report.malformed_bytes += offset; + add_warning(&mut self.pending_warning, "mbap_resynchronized"); + continue; + } + + // Preserve the final six bytes because they may be the start + // of a header split across TCP segments. + let discard = self.decoded_bytes.len().saturating_sub(6); + if discard > 0 { + self.decoded_bytes.drain(..discard); + report.malformed_bytes += discard; + add_warning(&mut self.pending_warning, "mbap_resynchronized"); + } + break; + } + + let length = + u16::from_be_bytes([self.decoded_bytes[4], self.decoded_bytes[5]]) as usize; + let frame_length = 6 + length; + if self.decoded_bytes.len() < frame_length { + break; + } + + let frame: Vec = self.decoded_bytes.drain(..frame_length).collect(); + report.adus.push(RawAdu { + transaction_id: u16::from_be_bytes([frame[0], frame[1]]), + unit_id: frame[6], + pdu: frame[7..].to_vec(), + warning: self.pending_warning.take(), + }); + } + report + } +} + +fn unwrap_sequence(sequence: u32, reference: u64) -> u64 { + const SPACE: u64 = 1_u64 << 32; + const HALF: u64 = 1_u64 << 31; + let base = reference & !(SPACE - 1); + let mut candidate = base | sequence as u64; + if candidate.saturating_add(HALF) < reference { + candidate = candidate.saturating_add(SPACE); + } else if candidate > reference.saturating_add(HALF) && candidate >= SPACE { + candidate -= SPACE; + } + candidate +} + +fn plausible_mbap_header(bytes: &[u8], offset: usize) -> bool { + if bytes.len() < offset + 7 { + return false; + } + let protocol_id = u16::from_be_bytes([bytes[offset + 2], bytes[offset + 3]]); + let length = u16::from_be_bytes([bytes[offset + 4], bytes[offset + 5]]) as usize; + if protocol_id != 0 || !(2..=MAX_MODBUS_LENGTH).contains(&length) { + return false; + } + // When the function byte is already present, zero cannot be a valid + // Modbus function and is useful protection against false resynchronization. + bytes.get(offset + 7).is_none_or(|function| *function != 0) +} + +fn find_next_mbap_header(bytes: &[u8]) -> Option { + (1..bytes.len().saturating_sub(6)).find(|offset| plausible_mbap_header(bytes, *offset)) +} + +#[derive(Clone, Debug, Default)] +struct Identity { + vendor_name: Option, + product_code: Option, + revision: Option, + vendor_url: Option, + product_name: Option, + model_name: Option, + user_application_name: Option, +} + +#[derive(Debug)] +struct DecodedRequest { + function_code: u8, + function_name: &'static str, + operation: &'static str, + address: Option, + quantity: Option, + write_address: Option, + write_quantity: Option, + coil_values: Option>, + register_values: Option>, + diagnostic_subfunction: Option, + device_id_code: Option, + device_id_object: Option, + warning: Option, +} + +fn decode_request(pdu: &[u8]) -> DecodedRequest { + let raw_function = pdu.first().copied().unwrap_or_default(); + let function_code = raw_function & 0x7f; + let mut decoded = DecodedRequest { + function_code, + function_name: function_name(function_code, pdu), + operation: operation_name(function_code, pdu), + address: None, + quantity: None, + write_address: None, + write_quantity: None, + coil_values: None, + register_values: None, + diagnostic_subfunction: None, + device_id_code: None, + device_id_object: None, + warning: None, + }; + + if raw_function & 0x80 != 0 { + add_warning( + &mut decoded.warning, + "exception_function_in_request_direction", + ); + } + + match function_code { + 1..=4 => { + decoded.address = read_u16(pdu, 1); + decoded.quantity = read_u16(pdu, 3); + require_pdu_len(pdu, 5, &mut decoded.warning); + } + 5 => { + decoded.address = read_u16(pdu, 1); + decoded.quantity = Some(1); + require_pdu_len(pdu, 5, &mut decoded.warning); + decoded.coil_values = decode_single_coil_value(pdu, &mut decoded.warning); + } + 6 => { + decoded.address = read_u16(pdu, 1); + decoded.quantity = Some(1); + require_pdu_len(pdu, 5, &mut decoded.warning); + decoded.register_values = read_u16(pdu, 3).map(|value| vec![value]); + } + 8 => { + decoded.diagnostic_subfunction = read_u16(pdu, 1); + require_pdu_len(pdu, 3, &mut decoded.warning); + } + 15 => { + decoded.address = read_u16(pdu, 1); + decoded.quantity = read_u16(pdu, 3); + decoded.coil_values = + decode_coil_values(pdu, decoded.quantity, 5, 6, &mut decoded.warning); + } + 16 => { + decoded.address = read_u16(pdu, 1); + decoded.quantity = read_u16(pdu, 3); + decoded.register_values = + decode_register_values(pdu, decoded.quantity, 5, 6, &mut decoded.warning); + } + 22 => { + decoded.address = read_u16(pdu, 1); + decoded.quantity = Some(1); + require_pdu_len(pdu, 7, &mut decoded.warning); + } + 23 => { + decoded.address = read_u16(pdu, 1); + decoded.quantity = read_u16(pdu, 3); + decoded.write_address = read_u16(pdu, 5); + decoded.write_quantity = read_u16(pdu, 7); + decoded.register_values = + decode_register_values(pdu, decoded.write_quantity, 9, 10, &mut decoded.warning); + } + 24 => { + decoded.address = read_u16(pdu, 1); + require_pdu_len(pdu, 3, &mut decoded.warning); + } + 43 if pdu.get(1) == Some(&0x0e) => { + decoded.device_id_code = pdu.get(2).copied(); + decoded.device_id_object = pdu.get(3).copied(); + require_pdu_len(pdu, 4, &mut decoded.warning); + } + _ => {} + } + decoded +} + +fn require_pdu_len(pdu: &[u8], required: usize, warning: &mut Option) { + if pdu.len() < required { + add_warning(warning, "truncated_function_payload"); + } +} + +fn read_u16(bytes: &[u8], offset: usize) -> Option { + let pair = bytes.get(offset..offset + 2)?; + Some(u16::from_be_bytes([pair[0], pair[1]])) +} + +fn decode_single_coil_value(pdu: &[u8], warning: &mut Option) -> Option> { + match read_u16(pdu, 3)? { + 0x0000 => Some(vec![false]), + 0xff00 => Some(vec![true]), + _ => { + add_warning(warning, "invalid_single_coil_value"); + None + } + } +} + +fn decode_coil_values( + pdu: &[u8], + quantity: Option, + byte_count_offset: usize, + values_offset: usize, + warning: &mut Option, +) -> Option> { + require_pdu_len(pdu, values_offset, warning); + let quantity = quantity? as usize; + let byte_count = *pdu.get(byte_count_offset)? as usize; + let expected_byte_count = quantity.div_ceil(8); + if byte_count != expected_byte_count { + add_warning(warning, "coil_write_byte_count_mismatch"); + return None; + } + + let required = values_offset + byte_count; + require_pdu_len(pdu, required, warning); + let packed = pdu.get(values_offset..required)?; + Some( + (0..quantity) + .map(|index| packed[index / 8] & (1_u8 << (index % 8)) != 0) + .collect(), + ) +} + +fn decode_register_values( + pdu: &[u8], + quantity: Option, + byte_count_offset: usize, + values_offset: usize, + warning: &mut Option, +) -> Option> { + require_pdu_len(pdu, values_offset, warning); + let quantity = quantity? as usize; + let byte_count = *pdu.get(byte_count_offset)? as usize; + let expected_byte_count = quantity * 2; + if byte_count != expected_byte_count { + add_warning(warning, "register_write_byte_count_mismatch"); + return None; + } + + let required = values_offset + byte_count; + require_pdu_len(pdu, required, warning); + let encoded = pdu.get(values_offset..required)?; + Some( + encoded + .chunks_exact(2) + .map(|pair| u16::from_be_bytes([pair[0], pair[1]])) + .collect(), + ) +} + +fn validate_response_pdu(pdu: &[u8]) -> Option { + let mut warning = None; + let raw_function = pdu.first().copied().unwrap_or_default(); + if raw_function & 0x80 != 0 { + require_pdu_len(pdu, 2, &mut warning); + return warning; + } + let function = raw_function & 0x7f; + match function { + 1..=4 | 23 => { + require_pdu_len(pdu, 2, &mut warning); + if let Some(byte_count) = pdu.get(1) { + require_pdu_len(pdu, 2 + *byte_count as usize, &mut warning); + } + } + 5 | 6 | 15 | 16 => require_pdu_len(pdu, 5, &mut warning), + 7 => require_pdu_len(pdu, 2, &mut warning), + 8 => require_pdu_len(pdu, 3, &mut warning), + 22 => require_pdu_len(pdu, 7, &mut warning), + _ => {} + } + warning +} + +fn function_name(function: u8, pdu: &[u8]) -> &'static str { + match function { + 1 => "read_coils", + 2 => "read_discrete_inputs", + 3 => "read_holding_registers", + 4 => "read_input_registers", + 5 => "write_single_coil", + 6 => "write_single_register", + 7 => "read_exception_status", + 8 => "diagnostics", + 11 => "get_communication_event_counter", + 12 => "get_communication_event_log", + 15 => "write_multiple_coils", + 16 => "write_multiple_registers", + 17 => "report_server_id", + 20 => "read_file_record", + 21 => "write_file_record", + 22 => "mask_write_register", + 23 => "read_write_multiple_registers", + 24 => "read_fifo_queue", + 43 if pdu.get(1) == Some(&0x0e) => "read_device_identification", + 43 => "encapsulated_interface_transport", + _ => "unknown", + } +} + +fn operation_name(function: u8, pdu: &[u8]) -> &'static str { + match function { + 1..=4 | 7 | 11 | 12 | 17 | 20 | 24 => "read", + 5 | 6 | 15 | 16 | 21 | 22 => "write", + 23 => "read_write", + 8 => "diagnostic", + 43 if pdu.get(1) == Some(&0x0e) => "device_identification", + 43 => "encapsulated", + _ => "other", + } +} + +fn parse_identity_response(pdu: &[u8]) -> (Identity, Option) { + let mut identity = Identity::default(); + let mut warning = None; + if pdu.len() < 7 || pdu.first() != Some(&43) || pdu.get(1) != Some(&0x0e) { + add_warning(&mut warning, "truncated_device_identification_response"); + return (identity, warning); + } + + let object_count = pdu[6] as usize; + let mut offset = 7; + for _ in 0..object_count { + if pdu.len() < offset + 2 { + add_warning(&mut warning, "truncated_device_identification_object"); + break; + } + let object_id = pdu[offset]; + let object_length = pdu[offset + 1] as usize; + offset += 2; + let Some(raw_value) = pdu.get(offset..offset + object_length) else { + add_warning(&mut warning, "truncated_device_identification_object"); + break; + }; + offset += object_length; + let value = sanitize_identity_value(raw_value); + if value.is_empty() { + continue; + } + match object_id { + 0x00 => identity.vendor_name = Some(value), + 0x01 => identity.product_code = Some(value), + 0x02 => identity.revision = Some(value), + 0x03 => identity.vendor_url = Some(value), + 0x04 => identity.product_name = Some(value), + 0x05 => identity.model_name = Some(value), + 0x06 => identity.user_application_name = Some(value), + _ => {} + } + } + (identity, warning) +} + +fn sanitize_identity_value(raw: &[u8]) -> String { + String::from_utf8_lossy(raw) + .chars() + .map(|character| { + if character.is_control() { + ' ' + } else { + character + } + }) + .collect::() + .trim() + .to_string() +} + +fn exception_name(code: u8) -> &'static str { + match code { + 1 => "illegal_function", + 2 => "illegal_data_address", + 3 => "illegal_data_value", + 4 => "server_device_failure", + 5 => "acknowledge", + 6 => "server_device_busy", + 8 => "memory_parity_error", + 10 => "gateway_path_unavailable", + 11 => "gateway_target_failed_to_respond", + _ => "unknown", + } +} + +#[derive(Clone, Debug)] +struct Observation { + timestamp: i64, + conversation: ConversationKey, + transaction_id: u16, + unit_id: u8, + function_code: u8, + function_name: String, + operation: String, + address: Option, + quantity: Option, + write_address: Option, + write_quantity: Option, + coil_values: Option>, + register_values: Option>, + diagnostic_subfunction: Option, + device_id_code: Option, + device_id_object: Option, + request_seen: bool, + response_seen: bool, + request_timestamp: Option, + response_timestamp: Option, + latency_usec: Option, + response_status: String, + exception_code: Option, + exception_name: Option, + identity: Identity, + request_packet: Option, + response_packet: Option, + parser_warning: Option, +} + +impl Observation { + fn from_request( + conversation: ConversationKey, + adu: &RawAdu, + decoded: DecodedRequest, + timestamp: i64, + packet_number: i64, + ) -> Self { + let mut parser_warning = adu.warning.clone(); + if let Some(warning) = decoded.warning { + add_warning(&mut parser_warning, &warning); + } + Self { + timestamp, + conversation, + transaction_id: adu.transaction_id, + unit_id: adu.unit_id, + function_code: decoded.function_code, + function_name: decoded.function_name.to_string(), + operation: decoded.operation.to_string(), + address: decoded.address, + quantity: decoded.quantity, + write_address: decoded.write_address, + write_quantity: decoded.write_quantity, + coil_values: decoded.coil_values, + register_values: decoded.register_values, + diagnostic_subfunction: decoded.diagnostic_subfunction, + device_id_code: decoded.device_id_code, + device_id_object: decoded.device_id_object, + request_seen: true, + response_seen: false, + request_timestamp: Some(timestamp), + response_timestamp: None, + latency_usec: None, + response_status: "missing_response".to_string(), + exception_code: None, + exception_name: None, + identity: Identity::default(), + request_packet: Some(packet_number), + response_packet: None, + parser_warning, + } + } + + fn from_orphan_response( + conversation: ConversationKey, + adu: &RawAdu, + timestamp: i64, + packet_number: i64, + ) -> Self { + let raw_function = adu.pdu[0]; + let function_code = raw_function & 0x7f; + let (identity, identity_warning) = if function_code == 43 && raw_function & 0x80 == 0 { + parse_identity_response(&adu.pdu) + } else { + (Identity::default(), None) + }; + let mut parser_warning = adu.warning.clone(); + if let Some(warning) = validate_response_pdu(&adu.pdu) { + add_warning(&mut parser_warning, &warning); + } + if let Some(warning) = identity_warning { + add_warning(&mut parser_warning, &warning); + } + let exception_code = (raw_function & 0x80 != 0) + .then(|| adu.pdu.get(1).copied()) + .flatten(); + if raw_function & 0x80 != 0 && exception_code.is_none() { + add_warning(&mut parser_warning, "truncated_exception_response"); + } + Self { + timestamp, + conversation, + transaction_id: adu.transaction_id, + unit_id: adu.unit_id, + function_code, + function_name: function_name(function_code, &adu.pdu).to_string(), + operation: operation_name(function_code, &adu.pdu).to_string(), + address: None, + quantity: None, + write_address: None, + write_quantity: None, + coil_values: None, + register_values: None, + diagnostic_subfunction: None, + device_id_code: None, + device_id_object: None, + request_seen: false, + response_seen: true, + request_timestamp: None, + response_timestamp: Some(timestamp), + latency_usec: None, + response_status: "orphan_response".to_string(), + exception_code, + exception_name: exception_code.map(exception_name).map(str::to_string), + identity, + request_packet: None, + response_packet: Some(packet_number), + parser_warning, + } + } +} + +struct Decoder { + server_port: u16, + streams: HashMap, + pending: HashMap, + observations: Vec, + summary: DecodeSummary, +} + +impl Decoder { + fn new(server_port: u16) -> Self { + Self { + server_port, + streams: HashMap::new(), + pending: HashMap::new(), + observations: Vec::new(), + summary: DecodeSummary::default(), + } + } + + fn ingest(&mut self, packet: TcpPayloadPacket) { + let direction = match ( + packet.src_port == self.server_port, + packet.dest_port == self.server_port, + ) { + (false, true) => Direction::Request, + (true, false) => Direction::Response, + _ => return, + }; + let conversation = match direction { + Direction::Request => ConversationKey { + client_ip: packet.src_ip.clone(), + client_port: packet.src_port, + server_ip: packet.dest_ip.clone(), + server_port: packet.dest_port, + }, + Direction::Response => ConversationKey { + client_ip: packet.dest_ip.clone(), + client_port: packet.dest_port, + server_ip: packet.src_ip.clone(), + server_port: packet.src_port, + }, + }; + let stream_key = StreamKey { + conversation: conversation.clone(), + direction, + }; + + if packet.syn { + if self + .streams + .get(&stream_key) + .is_some_and(TcpStreamState::has_incomplete_data) + { + self.summary.incomplete_streams += 1; + } + self.streams.remove(&stream_key); + } + + let payload_sequence = packet.sequence_number.wrapping_add(u32::from(packet.syn)); + let extract = if packet.payload.is_empty() { + ExtractReport::default() + } else { + self.summary.tcp_payload_packets += 1; + let stream = self.streams.entry(stream_key.clone()).or_default(); + let feed = stream.feed(payload_sequence, &packet.payload); + self.summary.retransmitted_segments += feed.retransmitted_segments; + self.summary.out_of_order_segments += feed.out_of_order_segments; + self.summary.forced_stream_gaps += feed.forced_stream_gaps; + stream.extract_adus() + }; + self.summary.malformed_bytes += extract.malformed_bytes; + + for adu in extract.adus { + match direction { + Direction::Request => { + self.summary.request_adus += 1; + self.handle_request( + conversation.clone(), + adu, + packet.timestamp, + packet.packet_number, + ); + } + Direction::Response => { + self.summary.response_adus += 1; + self.handle_response( + conversation.clone(), + adu, + packet.timestamp, + packet.packet_number, + ); + } + } + } + + if packet.fin || packet.rst { + let incomplete = self + .streams + .remove(&stream_key) + .is_some_and(|stream| stream.has_incomplete_data()); + self.summary.incomplete_streams += usize::from(incomplete); + } + } + + fn handle_request( + &mut self, + conversation: ConversationKey, + adu: RawAdu, + timestamp: i64, + packet_number: i64, + ) { + let decoded = decode_request(&adu.pdu); + let key = TransactionKey { + conversation: conversation.clone(), + transaction_id: adu.transaction_id, + unit_id: adu.unit_id, + }; + let observation = + Observation::from_request(conversation, &adu, decoded, timestamp, packet_number); + if let Some(mut replaced) = self.pending.insert(key, observation) { + add_warning( + &mut replaced.parser_warning, + "transaction_id_reused_before_response", + ); + self.observations.push(replaced); + } + } + + fn handle_response( + &mut self, + conversation: ConversationKey, + adu: RawAdu, + timestamp: i64, + packet_number: i64, + ) { + let key = TransactionKey { + conversation: conversation.clone(), + transaction_id: adu.transaction_id, + unit_id: adu.unit_id, + }; + let Some(mut observation) = self.pending.remove(&key) else { + self.observations.push(Observation::from_orphan_response( + conversation, + &adu, + timestamp, + packet_number, + )); + return; + }; + + let raw_function = adu.pdu[0]; + let response_function = raw_function & 0x7f; + observation.response_seen = true; + observation.response_timestamp = Some(timestamp); + observation.response_packet = Some(packet_number); + observation.latency_usec = observation + .request_timestamp + .and_then(|request| timestamp.checked_sub(request)) + .filter(|latency| *latency >= 0); + if observation + .request_timestamp + .is_some_and(|request| timestamp < request) + { + add_warning( + &mut observation.parser_warning, + "response_timestamp_precedes_request", + ); + } + if let Some(warning) = adu.warning.as_deref() { + add_warning(&mut observation.parser_warning, warning); + } + if let Some(warning) = validate_response_pdu(&adu.pdu) { + add_warning(&mut observation.parser_warning, &warning); + } + + if response_function != observation.function_code { + observation.response_status = "function_mismatch".to_string(); + add_warning( + &mut observation.parser_warning, + "response_function_does_not_match_request", + ); + } else if raw_function & 0x80 != 0 { + observation.response_status = "exception".to_string(); + observation.exception_code = adu.pdu.get(1).copied(); + observation.exception_name = observation + .exception_code + .map(exception_name) + .map(str::to_string); + if observation.exception_code.is_none() { + add_warning( + &mut observation.parser_warning, + "truncated_exception_response", + ); + } + } else { + observation.response_status = "ok".to_string(); + } + + if response_function == 43 && raw_function & 0x80 == 0 { + let (identity, warning) = parse_identity_response(&adu.pdu); + observation.identity = identity; + if let Some(warning) = warning { + add_warning(&mut observation.parser_warning, &warning); + } + } + self.observations.push(observation); + } + + fn finish(mut self) -> (Vec, DecodeSummary) { + self.summary.incomplete_streams += self + .streams + .values() + .filter(|stream| stream.has_incomplete_data()) + .count(); + self.observations.extend(self.pending.into_values()); + self.observations.sort_by_key(|observation| { + ( + observation.timestamp, + observation.conversation.clone(), + observation.transaction_id, + observation.unit_id, + ) + }); + + self.summary.observations = self.observations.len(); + self.summary.complete = self + .observations + .iter() + .filter(|observation| observation.request_seen && observation.response_seen) + .count(); + self.summary.request_only = self + .observations + .iter() + .filter(|observation| observation.request_seen && !observation.response_seen) + .count(); + self.summary.response_only = self + .observations + .iter() + .filter(|observation| !observation.request_seen && observation.response_seen) + .count(); + self.summary.exceptions = self + .observations + .iter() + .filter(|observation| observation.response_status == "exception") + .count(); + (self.observations, self.summary) + } +} + +pub fn modbus_to_parquet(input: &str, output: &str, server_port: u16) -> Result { + if server_port == 0 { + return Err("Modbus server port must be between 1 and 65535".into()); + } + + let file = File::open(input)?; + let mut reader = create_reader(1 << 20, file)?; + let mut decoder = Decoder::new(server_port); + let mut linktype = LINKTYPE_ETHERNET; + let mut legacy_nanos = false; + let mut packet_number = 0_i64; + + loop { + match reader.next() { + Ok((offset, block)) => { + match block { + PcapBlockOwned::LegacyHeader(header) => { + linktype = header.network.0 as u16; + legacy_nanos = header.magic_number == 0xa1b2_3c4d; + } + PcapBlockOwned::Legacy(packet) => { + packet_number += 1; + let fractional_usec = if legacy_nanos { + (packet.ts_usec / 1000) as i64 + } else { + packet.ts_usec as i64 + }; + let timestamp = packet.ts_sec as i64 * 1_000_000 + fractional_usec; + if let Some(packet) = + parse_tcp_payload(packet.data, linktype, timestamp, packet_number) + { + decoder.ingest(packet); + } + } + PcapBlockOwned::NG(Block::InterfaceDescription(description)) => { + linktype = description.linktype.0 as u16; + } + PcapBlockOwned::NG(Block::EnhancedPacket(packet)) => { + packet_number += 1; + // pcapng's default if_tsresol is microseconds, matching + // the existing flow reader. Interface-specific options + // are intentionally left for a later capture-layer pass. + let timestamp = ((packet.ts_high as i64) << 32) | packet.ts_low as i64; + if let Some(packet) = + parse_tcp_payload(packet.data, linktype, timestamp, packet_number) + { + decoder.ingest(packet); + } + } + _ => {} + } + reader.consume(offset); + } + Err(PcapError::Eof) => break, + Err(PcapError::Incomplete(_)) => { + if reader.refill().is_err() { + return Err("pcap refill failed (truncated capture?)".into()); + } + } + Err(error) => return Err(format!("pcap parse error: {error:?}").into()), + } + } + + let (observations, summary) = decoder.finish(); + if observations.is_empty() { + return Err(format!( + "no decodable Modbus/TCP observations found on server port {server_port} \ + ({} payload packets, {} malformed bytes, {} incomplete streams)", + summary.tcp_payload_packets, summary.malformed_bytes, summary.incomplete_streams, + ) + .into()); + } + let batch = observations_to_batch(&observations)?; + write_parquet(&batch, output)?; + Ok(summary) +} + +fn parse_tcp_payload( + data: &[u8], + linktype: u16, + timestamp: i64, + packet_number: i64, +) -> Option { + let sliced = if linktype == LINKTYPE_ETHERNET { + SlicedPacket::from_ethernet(data).ok()? + } else { + SlicedPacket::from_ip(data).ok()? + }; + let (src_ip, dest_ip) = match sliced.net.as_ref()? { + NetSlice::Ipv4(ipv4) => ( + Ipv4Addr::from(ipv4.header().source()).to_string(), + Ipv4Addr::from(ipv4.header().destination()).to_string(), + ), + NetSlice::Ipv6(ipv6) => ( + Ipv6Addr::from(ipv6.header().source()).to_string(), + Ipv6Addr::from(ipv6.header().destination()).to_string(), + ), + _ => return None, + }; + let Some(TransportSlice::Tcp(tcp)) = sliced.transport else { + return None; + }; + Some(TcpPayloadPacket { + timestamp, + packet_number, + src_ip, + dest_ip, + src_port: tcp.source_port(), + dest_port: tcp.destination_port(), + sequence_number: tcp.sequence_number(), + syn: tcp.syn(), + fin: tcp.fin(), + rst: tcp.rst(), + payload: tcp.payload().to_vec(), + }) +} + +fn modbus_schema() -> Arc { + let spec: Value = serde_json::from_str(SCHEMA_JSON).expect("embedded Modbus schema is valid"); + let declared_version = spec["schema_version"] + .as_str() + .expect("Modbus schema declares schema_version"); + assert_eq!(declared_version, SCHEMA_VERSION); + let metadata = HashMap::from([ + ("deeptempo.schema".to_string(), SCHEMA_VERSION.to_string()), + ("deeptempo.decoder".to_string(), "passive".to_string()), + ( + "deeptempo.direction_basis".to_string(), + DIRECTION_BASIS.to_string(), + ), + ]); + Arc::new(Schema::new_with_metadata( + vec![ + Field::new("schema_version", DataType::Utf8, false), + Field::new("timestamp", DataType::Int64, false), + Field::new("client_ip", DataType::Utf8, false), + Field::new("client_port", DataType::Int32, false), + Field::new("server_ip", DataType::Utf8, false), + Field::new("server_port", DataType::Int32, false), + Field::new("transaction_id", DataType::Int32, false), + Field::new("unit_id", DataType::Int32, false), + Field::new("function_code", DataType::Int32, false), + Field::new("function_name", DataType::Utf8, false), + Field::new("operation", DataType::Utf8, false), + Field::new("address", DataType::Int32, true), + Field::new("quantity", DataType::Int32, true), + Field::new("write_address", DataType::Int32, true), + Field::new("write_quantity", DataType::Int32, true), + Field::new( + "coil_values", + DataType::List(Arc::new(Field::new("item", DataType::Boolean, false))), + true, + ), + Field::new( + "register_values", + DataType::List(Arc::new(Field::new("item", DataType::Int32, false))), + true, + ), + Field::new("diagnostic_subfunction", DataType::Int32, true), + Field::new("device_id_code", DataType::Int32, true), + Field::new("device_id_object", DataType::Int32, true), + Field::new("request_seen", DataType::Boolean, false), + Field::new("response_seen", DataType::Boolean, false), + Field::new("request_timestamp", DataType::Int64, true), + Field::new("response_timestamp", DataType::Int64, true), + Field::new("latency_usec", DataType::Int64, true), + Field::new("response_status", DataType::Utf8, false), + Field::new("exception_code", DataType::Int32, true), + Field::new("exception_name", DataType::Utf8, true), + Field::new("vendor_name", DataType::Utf8, true), + Field::new("product_code", DataType::Utf8, true), + Field::new("revision", DataType::Utf8, true), + Field::new("vendor_url", DataType::Utf8, true), + Field::new("product_name", DataType::Utf8, true), + Field::new("model_name", DataType::Utf8, true), + Field::new("user_application_name", DataType::Utf8, true), + Field::new("request_packet", DataType::Int64, true), + Field::new("response_packet", DataType::Int64, true), + Field::new("direction_basis", DataType::Utf8, false), + Field::new("parser_warning", DataType::Utf8, true), + ], + metadata, + )) +} + +fn observations_to_batch( + observations: &[Observation], +) -> std::result::Result { + let columns: Vec = vec![ + Arc::new(StringArray::from_iter_values( + observations.iter().map(|_| SCHEMA_VERSION), + )), + Arc::new(Int64Array::from_iter_values( + observations.iter().map(|observation| observation.timestamp), + )), + Arc::new(StringArray::from_iter_values( + observations + .iter() + .map(|observation| observation.conversation.client_ip.as_str()), + )), + Arc::new(Int32Array::from_iter_values( + observations + .iter() + .map(|observation| observation.conversation.client_port as i32), + )), + Arc::new(StringArray::from_iter_values( + observations + .iter() + .map(|observation| observation.conversation.server_ip.as_str()), + )), + Arc::new(Int32Array::from_iter_values( + observations + .iter() + .map(|observation| observation.conversation.server_port as i32), + )), + Arc::new(Int32Array::from_iter_values( + observations + .iter() + .map(|observation| observation.transaction_id as i32), + )), + Arc::new(Int32Array::from_iter_values( + observations + .iter() + .map(|observation| observation.unit_id as i32), + )), + Arc::new(Int32Array::from_iter_values( + observations + .iter() + .map(|observation| observation.function_code as i32), + )), + Arc::new(StringArray::from_iter_values( + observations + .iter() + .map(|observation| observation.function_name.as_str()), + )), + Arc::new(StringArray::from_iter_values( + observations + .iter() + .map(|observation| observation.operation.as_str()), + )), + optional_u16_array(observations.iter().map(|observation| observation.address)), + optional_u16_array(observations.iter().map(|observation| observation.quantity)), + optional_u16_array( + observations + .iter() + .map(|observation| observation.write_address), + ), + optional_u16_array( + observations + .iter() + .map(|observation| observation.write_quantity), + ), + optional_bool_list_array( + observations + .iter() + .map(|observation| observation.coil_values.as_deref()), + ), + optional_u16_list_array( + observations + .iter() + .map(|observation| observation.register_values.as_deref()), + ), + optional_u16_array( + observations + .iter() + .map(|observation| observation.diagnostic_subfunction), + ), + optional_u8_array( + observations + .iter() + .map(|observation| observation.device_id_code), + ), + optional_u8_array( + observations + .iter() + .map(|observation| observation.device_id_object), + ), + Arc::new(BooleanArray::from( + observations + .iter() + .map(|observation| observation.request_seen) + .collect::>(), + )), + Arc::new(BooleanArray::from( + observations + .iter() + .map(|observation| observation.response_seen) + .collect::>(), + )), + Arc::new(Int64Array::from( + observations + .iter() + .map(|observation| observation.request_timestamp) + .collect::>(), + )), + Arc::new(Int64Array::from( + observations + .iter() + .map(|observation| observation.response_timestamp) + .collect::>(), + )), + Arc::new(Int64Array::from( + observations + .iter() + .map(|observation| observation.latency_usec) + .collect::>(), + )), + Arc::new(StringArray::from_iter_values( + observations + .iter() + .map(|observation| observation.response_status.as_str()), + )), + optional_u8_array( + observations + .iter() + .map(|observation| observation.exception_code), + ), + optional_string_array( + observations + .iter() + .map(|observation| observation.exception_name.as_deref()), + ), + optional_string_array( + observations + .iter() + .map(|observation| observation.identity.vendor_name.as_deref()), + ), + optional_string_array( + observations + .iter() + .map(|observation| observation.identity.product_code.as_deref()), + ), + optional_string_array( + observations + .iter() + .map(|observation| observation.identity.revision.as_deref()), + ), + optional_string_array( + observations + .iter() + .map(|observation| observation.identity.vendor_url.as_deref()), + ), + optional_string_array( + observations + .iter() + .map(|observation| observation.identity.product_name.as_deref()), + ), + optional_string_array( + observations + .iter() + .map(|observation| observation.identity.model_name.as_deref()), + ), + optional_string_array( + observations + .iter() + .map(|observation| observation.identity.user_application_name.as_deref()), + ), + Arc::new(Int64Array::from( + observations + .iter() + .map(|observation| observation.request_packet) + .collect::>(), + )), + Arc::new(Int64Array::from( + observations + .iter() + .map(|observation| observation.response_packet) + .collect::>(), + )), + Arc::new(StringArray::from_iter_values( + observations.iter().map(|_| DIRECTION_BASIS), + )), + optional_string_array( + observations + .iter() + .map(|observation| observation.parser_warning.as_deref()), + ), + ]; + RecordBatch::try_new(modbus_schema(), columns) +} + +fn optional_u16_array(values: impl Iterator>) -> ArrayRef { + Arc::new(Int32Array::from( + values.map(|value| value.map(i32::from)).collect::>(), + )) +} + +fn optional_u8_array(values: impl Iterator>) -> ArrayRef { + Arc::new(Int32Array::from( + values.map(|value| value.map(i32::from)).collect::>(), + )) +} + +fn optional_bool_list_array<'a>(values: impl Iterator>) -> ArrayRef { + let mut builder = ListBuilder::new(BooleanBuilder::new()).with_field(Field::new( + "item", + DataType::Boolean, + false, + )); + for values in values { + match values { + Some(values) => { + for value in values { + builder.values().append_value(*value); + } + builder.append(true); + } + None => builder.append(false), + } + } + Arc::new(builder.finish()) +} + +fn optional_u16_list_array<'a>(values: impl Iterator>) -> ArrayRef { + let mut builder = ListBuilder::new(Int32Builder::new()).with_field(Field::new( + "item", + DataType::Int32, + false, + )); + for values in values { + match values { + Some(values) => { + for value in values { + builder.values().append_value(i32::from(*value)); + } + builder.append(true); + } + None => builder.append(false), + } + } + Arc::new(builder.finish()) +} + +fn optional_string_array<'a>(values: impl Iterator>) -> ArrayRef { + Arc::new(StringArray::from_iter(values)) +} + +fn add_warning(target: &mut Option, warning: &str) { + if warning.is_empty() { + return; + } + match target { + Some(existing) if !existing.split(';').any(|part| part == warning) => { + existing.push(';'); + existing.push_str(warning); + } + None => *target = Some(warning.to_string()), + _ => {} + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn adu(transaction_id: u16, unit_id: u8, pdu: &[u8]) -> Vec { + let length = (1 + pdu.len()) as u16; + let mut bytes = Vec::new(); + bytes.extend_from_slice(&transaction_id.to_be_bytes()); + bytes.extend_from_slice(&0_u16.to_be_bytes()); + bytes.extend_from_slice(&length.to_be_bytes()); + bytes.push(unit_id); + bytes.extend_from_slice(pdu); + bytes + } + + #[test] + fn reassembles_fragmented_and_coalesced_adus() { + let first = adu(1, 7, &[3, 0, 10, 0, 2]); + let second = adu(2, 7, &[6, 0, 20, 0, 1]); + let mut stream = TcpStreamState::default(); + + stream.feed(1000, &first[..8]); + assert!(stream.extract_adus().adus.is_empty()); + stream.feed(1008, &[first[8..].to_vec(), second.clone()].concat()); + let extracted = stream.extract_adus(); + assert_eq!(extracted.adus.len(), 2); + assert_eq!(extracted.adus[0].transaction_id, 1); + assert_eq!(extracted.adus[1].transaction_id, 2); + } + + #[test] + fn handles_out_of_order_tail_and_retransmission() { + let frame = adu(9, 1, &[3, 0, 1, 0, 4]); + let mut stream = TcpStreamState::default(); + stream.feed(50, &frame[..6]); + let out_of_order = stream.feed(59, &frame[9..]); + assert_eq!(out_of_order.out_of_order_segments, 1); + assert!(stream.extract_adus().adus.is_empty()); + stream.feed(56, &frame[6..9]); + let extracted = stream.extract_adus(); + assert_eq!(extracted.adus.len(), 1); + let retransmission = stream.feed(50, &frame); + assert_eq!(retransmission.retransmitted_segments, 1); + assert!(stream.extract_adus().adus.is_empty()); + } + + #[test] + fn resynchronizes_after_malformed_prefix() { + let frame = adu(5, 1, &[16, 0, 100, 0, 2, 4, 0, 1, 0, 2]); + let mut stream = TcpStreamState::default(); + stream.feed(0, &[vec![0xff, 0xaa], frame].concat()); + let extracted = stream.extract_adus(); + assert_eq!(extracted.malformed_bytes, 2); + assert_eq!(extracted.adus.len(), 1); + assert_eq!( + extracted.adus[0].warning.as_deref(), + Some("mbap_resynchronized") + ); + } + + #[test] + fn decodes_read_write_and_device_identification_requests() { + let read = decode_request(&[3, 0x12, 0x34, 0, 8]); + assert_eq!(read.function_name, "read_holding_registers"); + assert_eq!(read.address, Some(0x1234)); + assert_eq!(read.quantity, Some(8)); + + let read_write = decode_request(&[23, 0, 10, 0, 2, 0, 20, 0, 3, 6, 0, 1, 0, 2, 0, 3]); + assert_eq!(read_write.operation, "read_write"); + assert_eq!(read_write.address, Some(10)); + assert_eq!(read_write.write_address, Some(20)); + assert_eq!(read_write.write_quantity, Some(3)); + assert_eq!(read_write.register_values, Some(vec![1, 2, 3])); + + let identity = decode_request(&[43, 14, 1, 0]); + assert_eq!(identity.operation, "device_identification"); + assert_eq!(identity.device_id_code, Some(1)); + assert_eq!(identity.device_id_object, Some(0)); + } + + #[test] + fn decodes_write_values_without_treating_mask_writes_as_final_values() { + let single_coil_on = decode_request(&[5, 0, 7, 0xff, 0]); + assert_eq!(single_coil_on.coil_values, Some(vec![true])); + + let single_coil_off = decode_request(&[5, 0, 7, 0, 0]); + assert_eq!(single_coil_off.coil_values, Some(vec![false])); + + let invalid_single_coil = decode_request(&[5, 0, 7, 0x12, 0x34]); + assert_eq!(invalid_single_coil.coil_values, None); + assert_eq!( + invalid_single_coil.warning.as_deref(), + Some("invalid_single_coil_value") + ); + + let multiple_coils = decode_request(&[15, 0, 20, 0, 10, 2, 0x55, 0x03]); + assert_eq!( + multiple_coils.coil_values, + Some(vec![ + true, false, true, false, true, false, true, false, true, true, + ]) + ); + + let single_register = decode_request(&[6, 0x04, 0x01, 0, 10]); + assert_eq!(single_register.address, Some(1025)); + assert_eq!(single_register.register_values, Some(vec![10])); + + let multiple_registers = decode_request(&[16, 0, 30, 0, 2, 4, 0, 10, 0, 20]); + assert_eq!(multiple_registers.register_values, Some(vec![10, 20])); + + let mask_write = decode_request(&[22, 0, 40, 0xff, 0, 0, 0xff]); + assert_eq!(mask_write.register_values, None); + } + + #[test] + fn rejects_incomplete_or_inconsistent_write_value_lists() { + let truncated = decode_request(&[16, 0, 10, 0, 2, 4, 0, 1]); + assert_eq!(truncated.register_values, None); + assert_eq!( + truncated.warning.as_deref(), + Some("truncated_function_payload") + ); + + let register_mismatch = decode_request(&[16, 0, 10, 0, 2, 2, 0, 1]); + assert_eq!(register_mismatch.register_values, None); + assert_eq!( + register_mismatch.warning.as_deref(), + Some("register_write_byte_count_mismatch") + ); + + let coil_mismatch = decode_request(&[15, 0, 10, 0, 9, 1, 0xff]); + assert_eq!(coil_mismatch.coil_values, None); + assert_eq!( + coil_mismatch.warning.as_deref(), + Some("coil_write_byte_count_mismatch") + ); + } + + #[test] + fn extracts_vendor_identity_objects() { + let mut pdu = vec![43, 14, 1, 1, 0, 0, 3]; + pdu.extend_from_slice(&[0, 7]); + pdu.extend_from_slice(b"Siemens"); + pdu.extend_from_slice(&[1, 7]); + pdu.extend_from_slice(b"S7-1500"); + pdu.extend_from_slice(&[4, 11]); + pdu.extend_from_slice(b"SIMATIC PLC"); + let (identity, warning) = parse_identity_response(&pdu); + assert!(warning.is_none()); + assert_eq!(identity.vendor_name.as_deref(), Some("Siemens")); + assert_eq!(identity.product_code.as_deref(), Some("S7-1500")); + assert_eq!(identity.product_name.as_deref(), Some("SIMATIC PLC")); + } + + #[test] + fn batch_schema_has_version_metadata_and_requested_write_values() { + let conversation = ConversationKey { + client_ip: "10.0.0.1".to_string(), + client_port: 40000, + server_ip: "10.0.0.2".to_string(), + server_port: 502, + }; + let raw = RawAdu { + transaction_id: 1, + unit_id: 1, + pdu: vec![6, 0x04, 0x01, 0, 10], + warning: None, + }; + let observation = + Observation::from_request(conversation, &raw, decode_request(&raw.pdu), 1, 2); + let batch = observations_to_batch(&[observation]).unwrap(); + assert_eq!(batch.num_rows(), 1); + assert_eq!( + batch + .schema() + .metadata() + .get("deeptempo.schema") + .map(String::as_str), + Some(SCHEMA_VERSION) + ); + assert_eq!( + batch + .schema() + .field_with_name("register_values") + .unwrap() + .data_type(), + &DataType::List(Arc::new(Field::new("item", DataType::Int32, false))) + ); + assert!(batch.schema().field_with_name("raw_pdu").is_err()); + } + + #[test] + fn executable_schema_matches_declared_json_contract() { + let spec: Value = serde_json::from_str(SCHEMA_JSON).unwrap(); + let declared = spec["fields"].as_array().unwrap(); + let arrow = modbus_schema(); + assert_eq!(declared.len(), arrow.fields().len()); + for (json_field, arrow_field) in declared.iter().zip(arrow.fields()) { + assert_eq!( + json_field["name"].as_str(), + Some(arrow_field.name().as_str()) + ); + assert_eq!( + json_field["nullable"].as_bool(), + Some(arrow_field.is_nullable()) + ); + let expected_type = match json_field["type"].as_str().unwrap() { + "utf8" => DataType::Utf8, + "int32" => DataType::Int32, + "int64" => DataType::Int64, + "boolean" => DataType::Boolean, + "list" => { + DataType::List(Arc::new(Field::new("item", DataType::Boolean, false))) + } + "list" => { + DataType::List(Arc::new(Field::new("item", DataType::Int32, false))) + } + other => panic!("unexpected declared type: {other}"), + }; + assert_eq!(arrow_field.data_type(), &expected_type); + } + } + + #[test] + fn configured_nonstandard_port_sets_roles_without_guessing() { + let payload = adu(1, 4, &[3, 0, 0, 0, 1]); + let mut decoder = Decoder::new(1502); + decoder.ingest(TcpPayloadPacket { + timestamp: 10, + packet_number: 1, + src_ip: "10.0.0.1".to_string(), + dest_ip: "10.0.0.2".to_string(), + src_port: 40000, + dest_port: 1502, + sequence_number: 100, + syn: false, + fin: false, + rst: false, + payload, + }); + let (observations, summary) = decoder.finish(); + assert_eq!(summary.request_only, 1); + assert_eq!(observations[0].conversation.client_port, 40000); + assert_eq!(observations[0].conversation.server_port, 1502); + assert_eq!(observations[0].unit_id, 4); + } + + #[test] + fn truncated_byte_count_is_visible_but_still_decoded() { + let request = decode_request(&[16, 0, 10, 0, 2, 4, 0, 1]); + assert_eq!(request.address, Some(10)); + assert_eq!( + request.warning.as_deref(), + Some("truncated_function_payload") + ); + assert_eq!( + validate_response_pdu(&[3, 4, 0, 1]).as_deref(), + Some("truncated_function_payload") + ); + } +} diff --git a/tests/test_modbus_e2e.py b/tests/test_modbus_e2e.py new file mode 100644 index 0000000..77801a0 --- /dev/null +++ b/tests/test_modbus_e2e.py @@ -0,0 +1,377 @@ +"""Golden end-to-end test for passive Modbus/TCP PCAP decoding. + +Requires ``dpkt`` and ``pyarrow`` plus a built flowprep binary. The fixture +exercises split and coalesced ADUs, a retransmission, MBAP resynchronization, +request/response pairing, an exception, missing/orphan halves, and FC 43/14 +device identity extraction. +""" + +import os +import subprocess +import sys +import tempfile + +import dpkt +import pyarrow.parquet as pq + + +_REPO = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") +_DEFAULT_BIN = os.path.join(_REPO, "target", "release", "flowprep") +FLOWPREP_BIN = os.environ.get("FLOWPREP_BIN", _DEFAULT_BIN) + + +def modbus_adu(transaction_id, unit_id, pdu): + length = 1 + len(pdu) + return ( + transaction_id.to_bytes(2, "big") + + b"\x00\x00" + + length.to_bytes(2, "big") + + bytes([unit_id]) + + pdu + ) + + +def identity_response(objects): + pdu = bytes([43, 14, 1, 1, 0, 0, len(objects)]) + for object_id, value in objects: + encoded = value.encode("utf-8") + pdu += bytes([object_id, len(encoded)]) + encoded + return pdu + + +def tcp_packet(src, dst, sport, dport, sequence, payload): + ethernet = dpkt.ethernet.Ethernet( + src=b"\x00\x01\x02\x03\x04\x05", + dst=b"\x06\x07\x08\x09\x0a\x0b", + ) + ip = dpkt.ip.IP( + src=bytes(map(int, src.split("."))), + dst=bytes(map(int, dst.split("."))), + p=dpkt.ip.IP_PROTO_TCP, + ) + tcp = dpkt.tcp.TCP( + sport=sport, + dport=dport, + seq=sequence, + flags=dpkt.tcp.TH_ACK | dpkt.tcp.TH_PUSH, + data=payload, + ) + ip.data = tcp + ip.len = len(bytes(ip)) + ethernet.data = ip + ethernet.type = dpkt.ethernet.ETH_TYPE_IP + return bytes(ethernet) + + +def build_modbus_pcap(path): + client = "10.20.0.10" + server = "10.20.0.20" + client_port = 40000 + server_port = 502 + client_sequence = 1000 + server_sequence = 5000 + base = 1_750_000_000.0 + + read_request = modbus_adu(1, 1, bytes([3, 0, 0, 0, 2])) + read_response = modbus_adu(1, 1, bytes([3, 4, 0, 10, 0, 20])) + write_request = modbus_adu(2, 1, bytes([6, 0, 16, 0, 123])) + write_response = modbus_adu(2, 1, bytes([6, 0, 16, 0, 123])) + id_request = modbus_adu(3, 1, bytes([43, 14, 1, 0])) + id_response = modbus_adu( + 3, + 1, + identity_response( + [ + (0, "Siemens"), + (1, "S7-1500"), + (2, "V3.1"), + (4, "SIMATIC PLC"), + ] + ), + ) + multi_write_request = modbus_adu( + 4, 1, bytes([16, 0, 100, 0, 2, 4, 0, 1, 0, 2]) + ) + exception_response = modbus_adu(4, 1, bytes([0x90, 2])) + coil_write_request = modbus_adu( + 6, 1, bytes([15, 0, 32, 0, 10, 2, 0x55, 0x03]) + ) + coil_write_response = modbus_adu(6, 1, bytes([15, 0, 32, 0, 10])) + missing_response_request = modbus_adu(5, 7, bytes([1, 0, 32, 0, 8])) + orphan_response = modbus_adu(99, 1, bytes([3, 2, 0, 42])) + + with open(path, "wb") as capture: + writer = dpkt.pcap.Writer(capture) + + # A request split across two TCP segments, followed by a full duplicate + # of the second segment. Only one ADU should be emitted. + first_part = read_request[:8] + second_part = read_request[8:] + writer.writepkt( + tcp_packet( + client, + server, + client_port, + server_port, + client_sequence, + first_part, + ), + ts=base, + ) + writer.writepkt( + tcp_packet( + client, + server, + client_port, + server_port, + client_sequence + len(first_part), + second_part, + ), + ts=base + 0.001, + ) + writer.writepkt( + tcp_packet( + client, + server, + client_port, + server_port, + client_sequence + len(first_part), + second_part, + ), + ts=base + 0.002, + ) + client_sequence += len(read_request) + + writer.writepkt( + tcp_packet( + server, + client, + server_port, + client_port, + server_sequence, + read_response, + ), + ts=base + 0.003, + ) + server_sequence += len(read_response) + + # Two complete requests and two responses coalesced into one segment in + # each direction. + coalesced_requests = write_request + id_request + writer.writepkt( + tcp_packet( + client, + server, + client_port, + server_port, + client_sequence, + coalesced_requests, + ), + ts=base + 0.004, + ) + client_sequence += len(coalesced_requests) + + coalesced_responses = write_response + id_response + writer.writepkt( + tcp_packet( + server, + client, + server_port, + client_port, + server_sequence, + coalesced_responses, + ), + ts=base + 0.005, + ) + server_sequence += len(coalesced_responses) + + # Malformed bytes before a valid request force a bounded MBAP resync; + # the protocol exception still pairs with the recovered request. + prefixed_request = b"\xff\xaa" + multi_write_request + writer.writepkt( + tcp_packet( + client, + server, + client_port, + server_port, + client_sequence, + prefixed_request, + ), + ts=base + 0.006, + ) + client_sequence += len(prefixed_request) + writer.writepkt( + tcp_packet( + server, + client, + server_port, + client_port, + server_sequence, + exception_response, + ), + ts=base + 0.007, + ) + server_sequence += len(exception_response) + + writer.writepkt( + tcp_packet( + client, + server, + client_port, + server_port, + client_sequence, + coil_write_request, + ), + ts=base + 0.008, + ) + client_sequence += len(coil_write_request) + writer.writepkt( + tcp_packet( + server, + client, + server_port, + client_port, + server_sequence, + coil_write_response, + ), + ts=base + 0.009, + ) + server_sequence += len(coil_write_response) + + writer.writepkt( + tcp_packet( + client, + server, + client_port, + server_port, + client_sequence, + missing_response_request, + ), + ts=base + 0.010, + ) + writer.writepkt( + tcp_packet( + server, + client, + server_port, + client_port, + server_sequence, + orphan_response, + ), + ts=base + 0.011, + ) + + +def build_malformed_only_pcap(path): + with open(path, "wb") as capture: + writer = dpkt.pcap.Writer(capture) + writer.writepkt( + tcp_packet( + "10.20.0.10", + "10.20.0.20", + 40000, + 502, + 1, + b"\xff" * 32, + ), + ts=1_750_000_000.0, + ) + + +def main(): + with tempfile.TemporaryDirectory(prefix="flowprep_modbus_") as tempdir: + capture = os.path.join(tempdir, "modbus.pcap") + output = os.path.join(tempdir, "modbus.parquet") + build_modbus_pcap(capture) + result = subprocess.run( + [FLOWPREP_BIN, "modbus", capture, output], + capture_output=True, + text=True, + ) + print(result.stdout.strip(), result.stderr.strip()) + assert result.returncode == 0, f"Modbus conversion failed: {result.stderr}" + assert "7 observations" in result.stdout + assert "1 retransmissions" in result.stdout + assert "2 malformed bytes skipped" in result.stdout + + table = pq.read_table(output) + assert table.num_rows == 7 + assert table.schema.metadata[b"deeptempo.schema"] == b"modbus_observation/v1" + assert table.schema.metadata[b"deeptempo.decoder"] == b"passive" + assert "coil_values" in table.column_names + assert "register_values" in table.column_names + assert "raw_pdu" not in table.column_names + + rows = {row["transaction_id"]: row for row in table.to_pylist()} + read = rows[1] + assert read["function_name"] == "read_holding_registers" + assert read["address"] == 0 and read["quantity"] == 2 + assert read["coil_values"] is None and read["register_values"] is None + assert read["response_status"] == "ok" + assert read["request_packet"] == 2 and read["response_packet"] == 4 + assert read["latency_usec"] == 2000 + + write = rows[2] + assert write["operation"] == "write" and write["address"] == 16 + assert write["register_values"] == [123] + assert write["response_status"] == "ok" + + identity = rows[3] + assert identity["operation"] == "device_identification" + assert identity["vendor_name"] == "Siemens" + assert identity["product_code"] == "S7-1500" + assert identity["revision"] == "V3.1" + assert identity["product_name"] == "SIMATIC PLC" + + exception = rows[4] + assert exception["function_name"] == "write_multiple_registers" + assert exception["register_values"] == [1, 2] + assert exception["response_status"] == "exception" + assert exception["exception_code"] == 2 + assert exception["exception_name"] == "illegal_data_address" + assert "mbap_resynchronized" in exception["parser_warning"] + + coil_write = rows[6] + assert coil_write["function_name"] == "write_multiple_coils" + assert coil_write["address"] == 32 and coil_write["quantity"] == 10 + assert coil_write["coil_values"] == [ + True, + False, + True, + False, + True, + False, + True, + False, + True, + True, + ] + assert coil_write["response_status"] == "ok" + + missing = rows[5] + assert missing["request_seen"] is True and missing["response_seen"] is False + assert missing["response_status"] == "missing_response" + assert missing["unit_id"] == 7 + + orphan = rows[99] + assert orphan["request_seen"] is False and orphan["response_seen"] is True + assert orphan["response_status"] == "orphan_response" + + malformed_capture = os.path.join(tempdir, "malformed.pcap") + malformed_output = os.path.join(tempdir, "malformed.parquet") + build_malformed_only_pcap(malformed_capture) + malformed = subprocess.run( + [FLOWPREP_BIN, "modbus", malformed_capture, malformed_output], + capture_output=True, + text=True, + ) + print(malformed.stdout.strip(), malformed.stderr.strip()) + assert malformed.returncode != 0 + assert "no decodable Modbus/TCP observations" in malformed.stderr + assert not os.path.exists(malformed_output) + + print("MODBUS E2E TEST PASSED") + + +if __name__ == "__main__": + sys.exit(main())