sccp_protocol/message/
wire.rs

1//! SCCP frame encoding and incremental stream decoding.
2//!
3//! Construct an outbound [`Frame`] and call [`Frame::encode`]. For inbound TCP
4//! data, retain one [`FrameDecoder`] per connection and feed each received
5//! chunk to [`FrameDecoder::push`]; it handles both fragmented and coalesced
6//! frames while enforcing [`MAX_FRAME_SIZE`]. Typed message decoding happens
7//! after framing through the message enums in the parent module.
8
9use std::fmt;
10use std::io::Cursor;
11
12use binrw::{BinRead, BinWrite};
13use thiserror::Error;
14
15use super::catalog::MessageId;
16use super::catalog::MessageRoute;
17
18/// Number of bytes in an SCCP frame header, including the message identifier.
19pub const HEADER_SIZE: usize = 12;
20/// Largest complete frame accepted or emitted by the framing layer.
21pub const MAX_FRAME_SIZE: usize = 8192;
22
23/// The fixed SCCP framing header. Keeping this separate from [`Frame`] lets
24/// the streaming decoder inspect a header before the complete payload has
25/// arrived, while still giving encoding and decoding one declarative layout.
26#[derive(BinRead, BinWrite, Clone, Copy, Debug, Eq, PartialEq)]
27#[brw(little)]
28struct WireHeader {
29    wire_len: u32,
30    protocol_version: u32,
31    message_id: u32,
32}
33
34#[derive(Clone, Eq, PartialEq)]
35/// One complete SCCP frame with an uninterpreted payload.
36///
37/// [`FrameDecoder`] produces frames from a byte stream. Call the appropriate
38/// typed message decoder afterward to validate the payload contract.
39pub struct Frame {
40    pub protocol_version: u32,
41    pub message_id: u32,
42    pub payload: Vec<u8>,
43}
44
45impl fmt::Debug for Frame {
46    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
47        f.debug_struct("Frame")
48            .field("protocol_version", &self.protocol_version)
49            .field("message_id", &format_args!("0x{:04x}", self.message_id))
50            .field("payload_len", &self.payload.len())
51            .finish()
52    }
53}
54
55impl Frame {
56    /// Creates a frame without validating its payload or total encoded size.
57    ///
58    /// Size validation occurs in [`Frame::encode`] or in typed message codecs.
59    pub fn new(protocol_version: u32, message_id: u32, payload: Vec<u8>) -> Self {
60        Self {
61            protocol_version,
62            message_id,
63            payload,
64        }
65    }
66
67    /// SCCP length includes the four-byte message ID, but excludes the first
68    /// two header words.
69    pub fn encode(&self) -> Result<Vec<u8>, CodecError> {
70        let wire_len = self
71            .payload
72            .len()
73            .checked_add(4)
74            .ok_or(CodecError::FrameTooLarge(usize::MAX))?;
75        let total = wire_len + 8;
76        if total > MAX_FRAME_SIZE {
77            return Err(CodecError::FrameTooLarge(total));
78        }
79        let header = WireHeader {
80            wire_len: u32::try_from(wire_len).map_err(|_| CodecError::FrameTooLarge(wire_len))?,
81            protocol_version: self.protocol_version,
82            message_id: self.message_id,
83        };
84        let mut output = Cursor::new(Vec::with_capacity(total));
85        header
86            .write(&mut output)
87            .map_err(|error| CodecError::wire("encode", self.message_id, &error))?;
88        output.get_mut().extend_from_slice(&self.payload);
89        Ok(output.into_inner())
90    }
91
92    /// Resolves the numeric identifier to its typed catalog value.
93    ///
94    /// Unrecognized identifiers become [`MessageId::Unknown`].
95    pub fn message_type(&self) -> MessageId {
96        MessageId::from(self.message_id)
97    }
98}
99
100#[derive(Debug, Error, Clone, Eq, PartialEq)]
101/// Validation and serialization failures produced by framing and message codecs.
102pub enum CodecError {
103    /// The header length cannot describe a valid frame.
104    #[error("SCCP frame length {0} is invalid")]
105    InvalidLength(u32),
106    /// The complete frame would exceed [`MAX_FRAME_SIZE`].
107    #[error("SCCP frame size {0} exceeds the configured maximum")]
108    FrameTooLarge(usize),
109    /// A fixed or mandatory payload prefix is incomplete.
110    #[error("message 0x{message_id:04x} is truncated: need {needed} bytes, got {actual}")]
111    Truncated {
112        message_id: u32,
113        needed: usize,
114        actual: usize,
115    },
116    /// A station identifier failed its textual or structural validation.
117    #[error("invalid SCCP device ID: {0}")]
118    InvalidDeviceId(String),
119    /// A runtime device definition violates a message-level constraint.
120    #[error("invalid SCCP device definition: {0}")]
121    InvalidDefinition(String),
122    /// A wire text field is not valid under the selected text policy.
123    #[error("invalid UTF-8-compatible SCCP text field")]
124    InvalidText,
125    /// The requested protocol version is outside the supported range.
126    #[error("unsupported SCCP protocol version {0}")]
127    UnsupportedProtocol(u32),
128    /// A known message was decoded through the wrong protocol role.
129    #[error("message 0x{message_id:04x} has route {actual:?}, expected {expected}")]
130    UnexpectedRoute {
131        message_id: u32,
132        actual: MessageRoute,
133        expected: &'static str,
134    },
135    /// A scalar field contains a value forbidden by its wire contract.
136    #[error("message 0x{message_id:04x} contains invalid {field}: {value}")]
137    InvalidValue {
138        message_id: u32,
139        field: &'static str,
140        value: u64,
141    },
142    /// A declared or decoded item count exceeds its bounded capacity.
143    #[error("message 0x{message_id:04x} {field} count {count} exceeds maximum {maximum}")]
144    CountTooLarge {
145        message_id: u32,
146        field: &'static str,
147        count: usize,
148        maximum: usize,
149    },
150    /// Reserved bytes in a fixed field are not zero-filled.
151    #[error("message 0x{message_id:04x} contains non-zero {field} padding")]
152    NonZeroPadding {
153        message_id: u32,
154        field: &'static str,
155    },
156    /// Bytes remain after decoding a layout that requires exact consumption.
157    #[error("message 0x{message_id:04x} has {count} unexpected trailing bytes")]
158    TrailingBytes { message_id: u32, count: usize },
159    /// A payload that requires 32-bit alignment has an invalid length.
160    #[error(
161        "message 0x{message_id:04x} payload length {actual} is not aligned to a four-byte boundary"
162    )]
163    InvalidAlignment { message_id: u32, actual: usize },
164    /// A text value exceeds the capacity of its wire field.
165    #[error(
166        "message 0x{message_id:04x} field {field} is too long: {actual} bytes, maximum {maximum}"
167    )]
168    TextTooLong {
169        message_id: u32,
170        field: &'static str,
171        actual: usize,
172        maximum: usize,
173    },
174    /// Secret key material exceeds its fixed wire capacity.
175    #[error("secret field {field} is too long: {actual} bytes, maximum {maximum}")]
176    SecretTooLong {
177        field: &'static str,
178        actual: usize,
179        maximum: usize,
180    },
181    /// The binary reader or writer failed at a known wire offset.
182    #[error("could not {operation} SCCP message 0x{message_id:04x} at byte {offset}: {detail}")]
183    Wire {
184        operation: &'static str,
185        message_id: u32,
186        offset: u64,
187        detail: String,
188    },
189}
190
191impl CodecError {
192    pub(crate) fn wire(operation: &'static str, message_id: u32, error: &binrw::Error) -> Self {
193        let offset = match error {
194            binrw::Error::BadMagic { pos, .. }
195            | binrw::Error::AssertFail { pos, .. }
196            | binrw::Error::Custom { pos, .. }
197            | binrw::Error::NoVariantMatch { pos }
198            | binrw::Error::EnumErrors { pos, .. } => *pos,
199            binrw::Error::Io(_) | binrw::Error::Backtrace(_) => 0,
200            _ => 0,
201        };
202        Self::Wire {
203            operation,
204            message_id,
205            offset,
206            detail: error.to_string(),
207        }
208    }
209}
210
211#[derive(Debug, Default)]
212/// Incremental decoder for an SCCP byte stream.
213///
214/// The decoder retains an incomplete trailing frame between calls. Complete
215/// frames are returned in stream order, including when a chunk contains more
216/// than one frame.
217pub struct FrameDecoder {
218    buffer: Vec<u8>,
219}
220
221impl FrameDecoder {
222    pub fn new() -> Self {
223        Self::default()
224    }
225
226    /// Appends a stream chunk and returns every newly completed frame.
227    ///
228    /// An empty vector means the buffered bytes do not yet complete a frame.
229    /// Framing errors are terminal for the current buffered stream; callers
230    /// should discard the decoder with the connection.
231    pub fn push(&mut self, bytes: &[u8]) -> Result<Vec<Frame>, CodecError> {
232        self.buffer.extend_from_slice(bytes);
233        let mut frames = Vec::new();
234        let mut consumed = 0_usize;
235        loop {
236            let retained = &self.buffer[consumed..];
237            if retained.len() < HEADER_SIZE {
238                break;
239            }
240            let header = WireHeader::read(&mut Cursor::new(&retained[..HEADER_SIZE]))
241                .map_err(|error| CodecError::wire("decode header for", 0, &error))?;
242            let length = header.wire_len;
243            if length < 4 {
244                return Err(CodecError::InvalidLength(length));
245            }
246            let total =
247                usize::try_from(length).map_err(|_| CodecError::FrameTooLarge(usize::MAX))? + 8;
248            if total > MAX_FRAME_SIZE {
249                return Err(CodecError::FrameTooLarge(total));
250            }
251            if retained.len() < total {
252                break;
253            }
254            let payload = retained[HEADER_SIZE..total].to_vec();
255            consumed += total;
256            frames.push(Frame {
257                protocol_version: header.protocol_version,
258                message_id: header.message_id,
259                payload,
260            });
261        }
262        // Compact once per input chunk. Draining after every decoded frame
263        // repeatedly shifted the retained coalesced stream and made a chunk
264        // containing many small frames quadratic in its byte length.
265        if consumed != 0 {
266            self.buffer.drain(..consumed);
267        }
268        debug_assert!(self.buffer.len() < MAX_FRAME_SIZE);
269        Ok(frames)
270    }
271
272    pub fn buffered_len(&self) -> usize {
273        self.buffer.len()
274    }
275}
276
277#[cfg(test)]
278mod tests {
279    use super::*;
280
281    #[test]
282    fn one_large_chunk_of_small_frames_is_drained_incrementally() {
283        let encoded = Frame::new(22, 0x0100, Vec::new()).encode().unwrap();
284        let frame_count = MAX_FRAME_SIZE * 2 / encoded.len() + 1;
285        let stream = encoded.repeat(frame_count);
286        assert!(stream.len() > MAX_FRAME_SIZE * 2);
287
288        let frames = FrameDecoder::new().push(&stream).unwrap();
289
290        assert_eq!(frames.len(), frame_count);
291        assert!(frames.iter().all(|frame| {
292            frame.protocol_version == 22 && frame.message_id == 0x0100 && frame.payload.is_empty()
293        }));
294    }
295
296    #[test]
297    fn frame_length_matches_skinny_header_definition() {
298        let frame = Frame::new(22, 0x0006, vec![1, 2, 3, 4]);
299        let encoded = frame.encode().unwrap();
300        assert_eq!(&encoded[..4], &8_u32.to_le_bytes());
301        assert_eq!(encoded.len(), 16);
302    }
303
304    #[test]
305    fn decoder_handles_fragmented_and_coalesced_frames() {
306        let first = Frame::new(0, 0, Vec::new()).encode().unwrap();
307        let second = Frame::new(22, 6, vec![1; 8]).encode().unwrap();
308        let mut decoder = FrameDecoder::new();
309        assert!(decoder.push(&first[..5]).unwrap().is_empty());
310        let mut rest = first[5..].to_vec();
311        rest.extend_from_slice(&second);
312        rest.extend_from_slice(&first);
313        rest.extend_from_slice(&second);
314        let frames = decoder.push(&rest).unwrap();
315        assert_eq!(
316            frames
317                .iter()
318                .map(|frame| frame.message_id)
319                .collect::<Vec<_>>(),
320            [0, 6, 0, 6]
321        );
322        assert_eq!(frames[1], frames[3], "duplicate frames changed in transit");
323        assert_eq!(frames[0], frames[2], "reordered frames changed in transit");
324    }
325
326    #[test]
327    fn decoder_retains_only_the_incomplete_tail_after_many_frames() {
328        let complete = Frame::new(22, 0x0100, vec![1, 2, 3, 4]).encode().unwrap();
329        let tail = Frame::new(22, 0x0101, vec![5; 32]).encode().unwrap();
330        let split = tail.len() - 7;
331        let mut chunk = complete.repeat(1_000);
332        chunk.extend_from_slice(&tail[..split]);
333
334        let mut decoder = FrameDecoder::new();
335        assert_eq!(decoder.push(&chunk).unwrap().len(), 1_000);
336        assert_eq!(decoder.buffered_len(), split);
337        let frames = decoder.push(&tail[split..]).unwrap();
338        assert_eq!(frames.len(), 1);
339        assert_eq!(frames[0].message_id, 0x0101);
340        assert_eq!(decoder.buffered_len(), 0);
341    }
342
343    #[test]
344    fn decoder_accepts_every_possible_single_fragment_boundary() {
345        let bytes = Frame::new(22, 0x22, (0_u8..64).collect()).encode().unwrap();
346        for split in 0..bytes.len() {
347            let mut decoder = FrameDecoder::new();
348            assert!(decoder.push(&bytes[..split]).unwrap().is_empty());
349            let frames = decoder.push(&bytes[split..]).unwrap();
350            assert_eq!(frames.len(), 1, "split at byte {split}");
351            assert_eq!(frames[0].payload, (0_u8..64).collect::<Vec<_>>());
352            assert_eq!(decoder.buffered_len(), 0);
353        }
354    }
355
356    #[test]
357    fn invalid_short_length_is_rejected() {
358        let mut decoder = FrameDecoder::new();
359        let mut bytes = vec![0; 12];
360        bytes[..4].copy_from_slice(&3_u32.to_le_bytes());
361        assert_eq!(decoder.push(&bytes), Err(CodecError::InvalidLength(3)));
362    }
363}
sccp_protocol/message/wire.rs

sccp_protocol/message/
wire.rs
