feat: add passive Modbus TCP decoding

.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

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

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<boolean>", "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<int32>", "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"
+  ]
+}

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 } => {