flowscope/modbus/
session.rs

1//! [`ModbusParser`] — `SessionParser` over TCP/502.
2
3use bytes::BytesMut;
4
5use super::parser::drain_frames;
6use super::types::ModbusMessage;
7use crate::Timestamp;
8use crate::session::SessionParser;
9
10pub const PARSER_KIND: &str = "modbus";
11
12/// IANA-assigned Modbus/TCP port.
13pub const MODBUS_PORT: u16 = 502;
14
15/// `SessionParser` for Modbus/TCP. Drains zero or more
16/// [`ModbusMessage`]s per feed call. Defensively caps the
17/// internal buffer at 64 KiB per direction — a single
18/// pathological frame larger than 65 KiB would be the only
19/// thing that could occupy more.
20#[derive(Debug, Clone, Default)]
21pub struct ModbusParser {
22    init: BytesMut,
23    resp: BytesMut,
24    /// Bytes went missing on (initiator, responder): resync on the
25    /// next plausible MBAP header.
26    resync: [bool; 2],
27}
28
29/// Offset of the next plausible MBAP header: protocol id 0, a length
30/// of 2..=254 covering unit id + PDU, and a public function code
31/// (1..=127, or an exception response 0x81..=0xFF).
32fn find_mbap(buf: &[u8]) -> Option<usize> {
33    buf.windows(8).position(|w| {
34        let len = u16::from_be_bytes([w[4], w[5]]);
35        w[2] == 0 && w[3] == 0 && (2..=254).contains(&len) && w[7] != 0 && w[7] != 0x80
36    })
37}
38
39const MAX_BUFFER: usize = 65_536;
40
41impl ModbusParser {
42    pub fn new() -> Self {
43        Self::default()
44    }
45}
46
47impl SessionParser for ModbusParser {
48    type Message = ModbusMessage;
49
50    fn parser_kind(&self) -> crate::ParserKind {
51        crate::ParserKind::Modbus
52    }
53
54    fn feed_initiator(&mut self, bytes: &[u8], _ts: Timestamp, out: &mut Vec<Self::Message>) {
55        if self.init.len() + bytes.len() > MAX_BUFFER {
56            self.init.clear();
57        }
58        self.init.extend_from_slice(bytes);
59        crate::session::resync_frames(&mut self.resync[0], &mut self.init, 7, find_mbap);
60        if !self.resync[0] {
61            drain_frames(&mut self.init, out);
62        }
63    }
64
65    fn feed_responder(&mut self, bytes: &[u8], _ts: Timestamp, out: &mut Vec<Self::Message>) {
66        if self.resp.len() + bytes.len() > MAX_BUFFER {
67            self.resp.clear();
68        }
69        self.resp.extend_from_slice(bytes);
70        crate::session::resync_frames(&mut self.resync[1], &mut self.resp, 7, find_mbap);
71        if !self.resync[1] {
72            drain_frames(&mut self.resp, out);
73        }
74    }
75
76    /// Drop the partial frame and resume at the next plausible MBAP
77    /// header.
78    fn on_gap(
79        &mut self,
80        side: crate::FlowSide,
81        _missing: u64,
82        _ts: Timestamp,
83        _out: &mut Vec<ModbusMessage>,
84    ) -> crate::GapResponse {
85        let i = usize::from(side == crate::FlowSide::Responder);
86        match side {
87            crate::FlowSide::Initiator => self.init.clear(),
88            crate::FlowSide::Responder => self.resp.clear(),
89        }
90        self.resync[i] = true;
91        crate::GapResponse::Continue
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98    use crate::event::FlowSide;
99
100    fn ts() -> Timestamp {
101        Timestamp::default()
102    }
103
104    fn build_read_holding(txn: u16, unit: u8, start: u16, qty: u16) -> Vec<u8> {
105        let mut buf = Vec::new();
106        buf.extend_from_slice(&txn.to_be_bytes());
107        buf.extend_from_slice(&0u16.to_be_bytes());
108        buf.extend_from_slice(&6u16.to_be_bytes());
109        buf.push(unit);
110        buf.push(3);
111        buf.extend_from_slice(&start.to_be_bytes());
112        buf.extend_from_slice(&qty.to_be_bytes());
113        buf
114    }
115
116    #[test]
117    fn parser_kind_label() {
118        assert_eq!(ModbusParser::new().parser_kind().as_str(), "modbus");
119        assert_eq!(MODBUS_PORT, 502);
120    }
121
122    #[test]
123    fn split_feed_reassembles_one_frame() {
124        let buf = build_read_holding(0x42, 1, 0, 5);
125        let mut p = ModbusParser::new();
126        let mut out = Vec::new();
127        p.feed_initiator(&buf[..5], ts(), &mut out);
128        assert!(out.is_empty());
129        p.feed_initiator(&buf[5..], ts(), &mut out);
130        assert_eq!(out.len(), 1);
131        assert_eq!(out[0].transaction_id, 0x42);
132        let _ = FlowSide::Initiator;
133    }
134
135    #[test]
136    fn pipelined_frames_drain_in_order() {
137        let a = build_read_holding(0x01, 1, 0, 2);
138        let b = build_read_holding(0x02, 1, 10, 4);
139        let mut buf = a.clone();
140        buf.extend_from_slice(&b);
141        let mut p = ModbusParser::new();
142        let mut out = Vec::new();
143        p.feed_initiator(&buf, ts(), &mut out);
144        assert_eq!(out.len(), 2);
145        assert_eq!(out[0].transaction_id, 0x01);
146        assert_eq!(out[1].transaction_id, 0x02);
147    }
148
149    #[test]
150    fn a_gap_resyncs_on_the_next_mbap_header() {
151        let mut p = ModbusParser::new();
152        let mut out = Vec::new();
153        let f = build_read_holding(7, 1, 100, 2);
154        p.feed_initiator(&f[..4], ts(), &mut out);
155        let r = p.on_gap(FlowSide::Initiator, 3, ts(), &mut out);
156        assert_eq!(r, crate::GapResponse::Continue);
157        let mut tail = f[7..].to_vec();
158        tail.extend_from_slice(&f);
159        p.feed_initiator(&tail, ts(), &mut out);
160        assert_eq!(out.len(), 1);
161    }
162}
flowscope/modbus/session.rs

flowscope/modbus/
session.rs
