From fa70af009a7ab03d6a88399abb985bb3ab16f49c Mon Sep 17 00:00:00 2001 From: bartlino Date: Sat, 29 Aug 2026 18:24:57 +0000 Subject: [PATCH 1/7] fix(mstp): USB chunk gaps must not trigger T_frame_abort Host async serial reads arrive in multi-millisecond chunks; treating inter-read gaps as Clause 9 T_frame_abort (~1.56ms @ 38400) discarded partial frames mid-assembly. Add decode_frame_stream (Complete/NeedMore/ Invalid), preserve trailing lone 0x55 on preamble miss, and replace wire T_frame_abort with a host stale-partial timeout based on max frame wire time plus USB scheduling slack. Co-authored-by: Cursor --- crates/bacnet-transport/src/mstp/mod.rs | 13 +- crates/bacnet-transport/src/mstp/port.rs | 40 ++-- crates/bacnet-transport/src/mstp_frame.rs | 241 ++++++++++++++++++++++ 3 files changed, 272 insertions(+), 22 deletions(-) diff --git a/crates/bacnet-transport/src/mstp/mod.rs b/crates/bacnet-transport/src/mstp/mod.rs index 467dc7ab..80a74cc4 100644 --- a/crates/bacnet-transport/src/mstp/mod.rs +++ b/crates/bacnet-transport/src/mstp/mod.rs @@ -61,11 +61,14 @@ fn calculate_t_turnaround_us(baud_rate: u32) -> u64 { const N_RETRY_TOKEN: u8 = 1; /// Maximum frame buffer size: preamble(2) + header(6) + max data(1497) + CRC16(2) pub(crate) const MSTP_MAX_FRAME_BUF: usize = 1507; -/// Maximum inter-byte gap within a frame before aborting reception. -/// Spec Clause 9.5.5: minimum 60 bit times. Computed per baud rate. -fn calculate_t_frame_abort_us(baud_rate: u32) -> u64 { - // 60 bit times in microseconds, rounded up - 60_000_000u64.div_ceil(baud_rate as u64) +/// Host-side stale partial-frame timeout for USB/chunked serial reassembly. +/// +/// This is **not** Clause 9 `T_frame_abort` (wire inter-byte silence). Host async reads +/// often arrive with multi-millisecond gaps that would falsely abort mid-frame assembly. +fn calculate_host_stale_partial_timeout_us(baud_rate: u32) -> u64 { + const USB_CHUNK_SLACK_US: u64 = 100_000; + let wire_us = (MSTP_MAX_FRAME_BUF as u64 * 10 * 1_000_000).div_ceil(baud_rate as u64); + wire_us.saturating_add(USB_CHUNK_SLACK_US) } /// Maximum number of queued outgoing frames before rejecting new sends. const MAX_TX_QUEUE_DEPTH: usize = 256; diff --git a/crates/bacnet-transport/src/mstp/port.rs b/crates/bacnet-transport/src/mstp/port.rs index 6678361e..f175409f 100644 --- a/crates/bacnet-transport/src/mstp/port.rs +++ b/crates/bacnet-transport/src/mstp/port.rs @@ -6,13 +6,14 @@ use tokio::sync::{mpsc, oneshot, Mutex}; use tracing::{debug, warn}; use crate::mstp_frame::{ - decode_frame, encode_frame, find_preamble, FrameType, MstpFrame, BROADCAST_MAC, + decode_frame_stream, encode_frame, find_preamble, retain_lone_preamble_byte, FrameType, + MstpFrame, StreamDecode, BROADCAST_MAC, }; use crate::port::{ReceivedNpdu, TransportPort}; use super::{ - calculate_t_frame_abort_us, calculate_t_turnaround_us, next_addr, MasterNode, MasterState, - MstpConfig, SerialPort, MSTP_MAX_FRAME_BUF, T_NO_TOKEN_MS, T_REPLY_DELAY_MS, + calculate_host_stale_partial_timeout_us, calculate_t_turnaround_us, next_addr, MasterNode, + MasterState, MstpConfig, SerialPort, MSTP_MAX_FRAME_BUF, T_NO_TOKEN_MS, T_REPLY_DELAY_MS, T_REPLY_TIMEOUT_MS, T_REPLY_TRANSMIT_MARGIN_MS, T_USAGE_TIMEOUT_MS, }; @@ -65,7 +66,9 @@ impl TransportPort for MstpTransport { let serial = Arc::new(serial); let serial_clone = serial.clone(); let t_turnaround_us = calculate_t_turnaround_us(self.config.baud_rate); - let t_frame_abort_us = calculate_t_frame_abort_us(self.config.baud_rate); + // Host reassembly policy: tolerate USB read chunk gaps, not wire T_frame_abort. + let host_stale_partial_timeout_us = + calculate_host_stale_partial_timeout_us(self.config.baud_rate); let reply_decision_delay_ms = T_REPLY_DELAY_MS .saturating_sub(t_turnaround_us.div_ceil(1_000) + T_REPLY_TRANSMIT_MARGIN_MS); @@ -123,14 +126,19 @@ impl TransportPort for MstpTransport { match result { Ok(0) => continue, Ok(n) => { - // T_frame_abort: discard partial frame if inter-byte gap - // exceeds the spec limit (60 bit times). + // Host stale-partial timeout: drop abandoned assembly if no + // bytes arrive for a long host-side gap (USB scheduling, not + // Clause 9 wire T_frame_abort). let now = tokio::time::Instant::now(); if !frame_buf.is_empty() { let gap = now.duration_since(last_byte_time); - if gap > tokio::time::Duration::from_micros(t_frame_abort_us) { + if gap + > tokio::time::Duration::from_micros( + host_stale_partial_timeout_us, + ) + { debug!( - "MS/TP: T_frame_abort exceeded ({gap:?}), discarding partial frame" + "MS/TP: host stale partial frame timeout ({gap:?}), discarding partial assembly" ); frame_buf.clear(); } @@ -160,7 +168,7 @@ impl TransportPort for MstpTransport { let preamble_pos = match find_preamble(&frame_buf) { Some(pos) => pos, None => { - frame_buf.clear(); + retain_lone_preamble_byte(&mut frame_buf); break; } }; @@ -170,8 +178,8 @@ impl TransportPort for MstpTransport { frame_buf.drain(..preamble_pos); } - match decode_frame(&frame_buf) { - Ok((frame, consumed)) => { + match decode_frame_stream(&frame_buf) { + StreamDecode::Complete { frame, consumed } => { frame_buf.drain(..consumed); // Process through state machine — collect @@ -274,12 +282,10 @@ impl TransportPort for MstpTransport { }), ); } - Err(_) => { - // Incomplete frame or bad CRC — skip first preamble byte - if frame_buf.len() > 2 { - frame_buf.drain(..1); - } - break; + StreamDecode::NeedMore => break, + StreamDecode::Invalid { discard } => { + let discard = discard.min(frame_buf.len()).max(1); + frame_buf.drain(..discard); } } } diff --git a/crates/bacnet-transport/src/mstp_frame.rs b/crates/bacnet-transport/src/mstp_frame.rs index 0e11767a..9738e376 100644 --- a/crates/bacnet-transport/src/mstp_frame.rs +++ b/crates/bacnet-transport/src/mstp_frame.rs @@ -246,6 +246,109 @@ pub fn encode_frame( Ok(()) } +/// Result of incremental/streaming MS/TP frame decode. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum StreamDecode { + /// A complete, validated frame was decoded. + Complete { frame: MstpFrame, consumed: usize }, + /// More bytes are required before a decode decision can be made. + NeedMore, + /// The buffer contains invalid data; discard at least `discard` bytes and resync. + Invalid { discard: usize }, +} + +/// Incrementally decode an MS/TP frame from a receive buffer (starting at the preamble). +/// +/// Unlike [`decode_frame`], incomplete input returns [`StreamDecode::NeedMore`] instead of +/// an error. Real corruption (bad CRC, invalid header) returns [`StreamDecode::Invalid`]. +pub fn decode_frame_stream(data: &[u8]) -> StreamDecode { + if data.is_empty() { + return StreamDecode::NeedMore; + } + + if data[0] != PREAMBLE[0] { + return StreamDecode::Invalid { discard: 1 }; + } + if data.len() < 2 { + return StreamDecode::NeedMore; + } + if data[1] != PREAMBLE[1] { + return StreamDecode::Invalid { discard: 1 }; + } + + // Preamble(2) + header fields(5) + header_crc(1) + if data.len() < 2 + HEADER_LENGTH { + return StreamDecode::NeedMore; + } + + if !crc8_valid(&data[2..8]) { + return StreamDecode::Invalid { discard: 1 }; + } + + let frame_type = FrameType::from_raw(data[2]); + let destination = data[3]; + let source = data[4]; + + if source > MAX_MASTER && source != BROADCAST_MAC { + return StreamDecode::Invalid { discard: 1 }; + } + if source == BROADCAST_MAC { + return StreamDecode::Invalid { discard: 1 }; + } + + let data_length = ((data[5] as usize) << 8) | (data[6] as usize); + if data_length > MAX_MPDU_DATA { + return StreamDecode::Invalid { discard: 1 }; + } + + let mut consumed = 2 + HEADER_LENGTH; + + if data_length > 0 { + let needed = consumed + data_length + 2; + if data.len() < needed { + return StreamDecode::NeedMore; + } + + if !crc16_valid(&data[consumed..consumed + data_length + 2]) { + return StreamDecode::Invalid { discard: needed }; + } + + let payload = Bytes::copy_from_slice(&data[consumed..consumed + data_length]); + consumed += data_length + 2; + + StreamDecode::Complete { + frame: MstpFrame { + frame_type, + destination, + source, + data: payload, + }, + consumed, + } + } else { + StreamDecode::Complete { + frame: MstpFrame { + frame_type, + destination, + source, + data: Bytes::new(), + }, + consumed, + } + } +} + +/// When no full preamble is found, retain a trailing lone `0x55` for the next chunk. +pub fn retain_lone_preamble_byte(buf: &mut Vec) { + if buf.last() == Some(&PREAMBLE[0]) { + let lone = PREAMBLE[0]; + buf.clear(); + buf.push(lone); + } else { + buf.clear(); + } +} + /// Decode an MS/TP frame from raw bytes (starting at the preamble). /// /// Returns the decoded frame and the number of bytes consumed. @@ -727,4 +830,142 @@ mod tests { let (decoded, _) = decode_frame(&buf).unwrap(); assert_eq!(decoded.source, MAX_MASTER); } + + // ----------------------------------------------------------------------- + // Streaming decode tests + // ----------------------------------------------------------------------- + + fn encode_token_frame() -> Vec { + let frame = MstpFrame { + frame_type: FrameType::Token, + destination: 1, + source: 0, + data: Bytes::new(), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + buf.to_vec() + } + + fn encode_data_frame() -> (MstpFrame, Vec) { + let frame = MstpFrame { + frame_type: FrameType::BACnetDataNotExpectingReply, + destination: 5, + source: 0, + data: Bytes::from_static(&[0x01, 0x02, 0x03, 0x04]), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + (frame, buf.to_vec()) + } + + #[test] + fn stream_decode_token_complete() { + let wire = encode_token_frame(); + assert_eq!( + decode_frame_stream(&wire), + StreamDecode::Complete { + frame: MstpFrame { + frame_type: FrameType::Token, + destination: 1, + source: 0, + data: Bytes::new(), + }, + consumed: 8, + } + ); + } + + #[test] + fn stream_decode_split_header_then_body() { + let (_frame, wire) = encode_data_frame(); + let split = wire.len() - 3; + + assert_eq!(decode_frame_stream(&wire[..split]), StreamDecode::NeedMore); + + let mut assembled = wire[..split].to_vec(); + assembled.extend_from_slice(&wire[split..]); + let StreamDecode::Complete { frame, consumed } = decode_frame_stream(&assembled) else { + panic!("expected complete frame after reassembly"); + }; + assert_eq!(consumed, wire.len()); + assert_eq!(frame.data.len(), 4); + } + + #[test] + fn stream_decode_preamble_split_across_chunks() { + let wire = encode_token_frame(); + assert_eq!(decode_frame_stream(&wire[..1]), StreamDecode::NeedMore); + + let mut assembled = wire[..1].to_vec(); + assembled.extend_from_slice(&wire[1..]); + assert!(matches!( + decode_frame_stream(&assembled), + StreamDecode::Complete { .. } + )); + } + + #[test] + fn stream_decode_host_gap_simulation_without_clear() { + let wire = encode_token_frame(); + let mut buf = wire[..4].to_vec(); + assert_eq!(decode_frame_stream(&buf), StreamDecode::NeedMore); + + // Simulate a host gap far beyond wire T_frame_abort without clearing the buffer. + buf.extend_from_slice(&wire[4..]); + assert!(matches!( + decode_frame_stream(&buf), + StreamDecode::Complete { .. } + )); + } + + #[test] + fn retain_lone_preamble_byte_preserves_trailing_0x55() { + let mut buf = vec![0x00, 0x12, 0x55]; + retain_lone_preamble_byte(&mut buf); + assert_eq!(buf, vec![0x55]); + + let mut no_preamble = vec![0x00, 0x12, 0x34]; + retain_lone_preamble_byte(&mut no_preamble); + assert!(no_preamble.is_empty()); + } + + #[test] + fn stream_decode_bad_header_crc_invalid() { + let mut wire = encode_token_frame(); + wire[7] ^= 0xFF; + assert_eq!( + decode_frame_stream(&wire), + StreamDecode::Invalid { discard: 1 } + ); + } + + #[test] + fn stream_decode_bad_data_crc_invalid_discards_frame() { + let (_frame, mut wire) = encode_data_frame(); + let last = wire.len() - 1; + wire[last] ^= 0xFF; + assert_eq!( + decode_frame_stream(&wire), + StreamDecode::Invalid { + discard: wire.len() + } + ); + } + + #[test] + fn stream_decode_need_more_on_short_header() { + assert_eq!( + decode_frame_stream(&[0x55, 0xFF, 0x00, 0x01]), + StreamDecode::NeedMore + ); + } + + #[test] + fn stream_decode_invalid_on_bad_second_preamble_byte() { + assert_eq!( + decode_frame_stream(&[0x55, 0x00]), + StreamDecode::Invalid { discard: 1 } + ); + } } From aab6f24dfd946499db3756b358f610d6cba50758 Mon Sep 17 00:00:00 2001 From: bartlino Date: Sun, 30 Aug 2026 14:12:41 +0000 Subject: [PATCH 2/7] fix(mstp): use BACnet Clause 9 header and data CRCs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace incorrect reflected polynomials (0xE0 / Modbus 0xA001) with Clause 9.6 values (0x81 / 0x8408). Add literal golden vectors including the live Token 0<-7 frame (CRC 0x37) and data frame 01 00 → 9F 16 so self-round-trip tests cannot mask interoperability failures. Co-authored-by: Cursor --- crates/bacnet-transport/src/mstp_frame.rs | 162 ++++++++++++++++------ 1 file changed, 123 insertions(+), 39 deletions(-) diff --git a/crates/bacnet-transport/src/mstp_frame.rs b/crates/bacnet-transport/src/mstp_frame.rs index 9738e376..2a9b2aa2 100644 --- a/crates/bacnet-transport/src/mstp_frame.rs +++ b/crates/bacnet-transport/src/mstp_frame.rs @@ -111,7 +111,8 @@ pub struct MstpFrame { // CRC-8 (Header CRC) // --------------------------------------------------------------------------- -/// CRC-8 lookup table. +/// BACnet Clause 9.6 Frame Header CRC — reflected poly `G(x)=x^8+x^7+1` → `0x81`. +/// (Prior incorrect table used `0xE0`, which passes self-round-trip but rejects live trunk frames.) const CRC8_TABLE: [u8; 256] = { let mut table = [0u8; 256]; let mut i = 0usize; @@ -120,7 +121,7 @@ const CRC8_TABLE: [u8; 256] = { let mut j = 0; while j < 8 { if crc & 1 != 0 { - crc = (crc >> 1) ^ 0xE0; + crc = (crc >> 1) ^ 0x81; } else { crc >>= 1; } @@ -132,7 +133,10 @@ const CRC8_TABLE: [u8; 256] = { table }; -/// Compute CRC-8 over the given data. Initial value 0xFF, result inverted. +/// Good header-CRC receiver residual (Clause 9.6), including the CRC octet. +pub const HEADER_CRC_RESIDUAL: u8 = 0x55; + +/// Compute CRC-8 over the given data. Initial value 0xFF, result ones-complemented. pub fn crc8(data: &[u8]) -> u8 { let mut crc: u8 = 0xFF; for &b in data { @@ -141,6 +145,15 @@ pub fn crc8(data: &[u8]) -> u8 { !crc } +/// Running header CRC including the transmitted CRC octet (no final invert). +pub fn crc8_accumulate_all(data_with_crc: &[u8]) -> u8 { + let mut crc: u8 = 0xFF; + for &b in data_with_crc { + crc = CRC8_TABLE[(crc ^ b) as usize]; + } + crc +} + /// Verify CRC-8: recomputes CRC over data (excluding last byte) and compares /// to the stored CRC byte. pub fn crc8_valid(data_with_crc: &[u8]) -> bool { @@ -155,7 +168,8 @@ pub fn crc8_valid(data_with_crc: &[u8]) -> bool { // CRC-16 (Data CRC) // --------------------------------------------------------------------------- -/// CRC-16 lookup table. +/// BACnet Clause 9.6 Data CRC — CRC-16-CCITT reflected poly `0x8408`. +/// (Prior incorrect table used Modbus `0xA001`.) const CRC16_TABLE: [u16; 256] = { let mut table = [0u16; 256]; let mut i = 0usize; @@ -164,7 +178,7 @@ const CRC16_TABLE: [u16; 256] = { let mut j = 0; while j < 8 { if crc & 1 != 0 { - crc = (crc >> 1) ^ 0xA001; + crc = (crc >> 1) ^ 0x8408; } else { crc >>= 1; } @@ -176,7 +190,10 @@ const CRC16_TABLE: [u16; 256] = { table }; -/// Compute CRC-16 over the given data. Initial value 0xFFFF, result inverted. +/// Good data-CRC receiver residual (Clause 9.6), including the CRC octets. +pub const DATA_CRC_RESIDUAL: u16 = 0xF0B8; + +/// Compute CRC-16 over the given data. Initial value 0xFFFF, result ones-complemented. pub fn crc16(data: &[u8]) -> u16 { let mut crc: u16 = 0xFFFF; for &b in data { @@ -185,8 +202,17 @@ pub fn crc16(data: &[u8]) -> u16 { !crc } +/// Running data CRC including the transmitted CRC octets (no final invert). +pub fn crc16_accumulate_all(data_with_crc: &[u8]) -> u16 { + let mut crc: u16 = 0xFFFF; + for &b in data_with_crc { + crc = (crc >> 8) ^ CRC16_TABLE[((crc ^ b as u16) & 0xFF) as usize]; + } + crc +} + /// Verify CRC-16: recomputes CRC over data (excluding last 2 bytes) and compares -/// to the stored CRC (little-endian). +/// to the stored CRC (little-endian; LS octet first on the wire). pub fn crc16_valid(data_with_crc: &[u8]) -> bool { if data_with_crc.len() < 3 { return false; @@ -459,46 +485,64 @@ mod tests { use super::*; // ----------------------------------------------------------------------- - // CRC tests + // CRC tests — Clause 9.6 golden vectors (literal expected bytes; not from crc8/crc16) // ----------------------------------------------------------------------- #[test] - fn crc8_known_value() { - // Token frame header: type=0x00, dest=0x01, src=0x00, len=0x0000 - let header = [0x00, 0x01, 0x00, 0x00, 0x00]; - let crc = crc8(&header); - // Verify by appending CRC and checking validity + fn crc8_clause9_header_vectors() { + // [frame_type, dest, src, len_hi, len_lo] -> header CRC + let vectors: &[([u8; 5], u8)] = &[ + ([0x00, 0x00, 0x07, 0x00, 0x00], 0x37), + ([0x00, 0x07, 0x00, 0x00, 0x00], 0x40), + ([0x01, 0x00, 0x07, 0x00, 0x00], 0xB1), + ([0x01, 0x07, 0x00, 0x00, 0x00], 0xC6), + ]; + for (header, expected) in vectors { + assert_eq!(crc8(header), *expected, "header={header:02X?}"); + let mut with_crc = header.to_vec(); + with_crc.push(*expected); + assert!(crc8_valid(&with_crc)); + assert_eq!(crc8_accumulate_all(&with_crc), HEADER_CRC_RESIDUAL); + } + } + + #[test] + fn crc8_one_bit_corruption_rejected() { + let header = [0x00, 0x00, 0x07, 0x00, 0x00]; let mut with_crc = header.to_vec(); - with_crc.push(crc); + with_crc.push(0x37); assert!(crc8_valid(&with_crc)); + with_crc[0] ^= 0x01; + assert!(!crc8_valid(&with_crc)); + assert_ne!(crc8_accumulate_all(&with_crc), HEADER_CRC_RESIDUAL); } #[test] - fn crc8_validate_round_trip() { - let data = [0x05, 0xFF, 0x03, 0x00, 0x0C]; - let crc = crc8(&data); - let mut with_crc = data.to_vec(); - with_crc.push(crc); - assert!(crc8_valid(&with_crc)); + fn crc16_clause9_data_vector_01_00() { + // Data 01 00 → CRC 0x169F, wire order LS first: 9F 16 + assert_eq!(crc16(&[0x01, 0x00]), 0x169F); + let with_crc = [0x01, 0x00, 0x9F, 0x16]; + assert!(crc16_valid(&with_crc)); + assert_eq!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); } #[test] - fn crc8_invalid_detects_corruption() { - let data = [0x05, 0xFF, 0x03, 0x00, 0x0C]; - let crc = crc8(&data); - let mut with_crc = data.to_vec(); - with_crc.push(crc ^ 0x01); // corrupt - assert!(!crc8_valid(&with_crc)); + fn crc16_one_bit_corruption_rejected() { + let mut with_crc = [0x01, 0x00, 0x9F, 0x16]; + assert!(crc16_valid(&with_crc)); + with_crc[0] ^= 0x01; + assert!(!crc16_valid(&with_crc)); + assert_ne!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); } #[test] - fn crc16_known_value() { - let data = [0x01, 0x00, 0x10, 0x02]; - let crc = crc16(&data); + fn crc8_validate_round_trip() { + let data = [0x05, 0xFF, 0x03, 0x00, 0x0C]; + let crc = crc8(&data); let mut with_crc = data.to_vec(); - with_crc.push(crc as u8); - with_crc.push((crc >> 8) as u8); - assert!(crc16_valid(&with_crc)); + with_crc.push(crc); + assert!(crc8_valid(&with_crc)); + assert_eq!(crc8_accumulate_all(&with_crc), HEADER_CRC_RESIDUAL); } #[test] @@ -509,16 +553,56 @@ mod tests { with_crc.push(crc as u8); with_crc.push((crc >> 8) as u8); assert!(crc16_valid(&with_crc)); + assert_eq!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); } #[test] - fn crc16_invalid_detects_corruption() { - let data = [0x01, 0x02, 0x03]; - let crc = crc16(&data); - let mut with_crc = data.to_vec(); - with_crc.push(crc as u8); - with_crc.push((crc >> 8) as u8 ^ 0x01); // corrupt - assert!(!crc16_valid(&with_crc)); + fn literal_token_frame_0_from_7() { + // Live trunk Token: dest BASRT(0) <- FEC(7), header CRC 0x37 + let wire: &[u8] = &[0x55, 0xFF, 0x00, 0x00, 0x07, 0x00, 0x00, 0x37]; + let (frame, consumed) = decode_frame(wire).expect("decode Token 0<-7"); + assert_eq!(consumed, 8); + assert_eq!(frame.frame_type, FrameType::Token); + assert_eq!(frame.destination, 0); + assert_eq!(frame.source, 7); + assert!(frame.data.is_empty()); + + let mut enc = BytesMut::new(); + encode_frame(&mut enc, &frame).unwrap(); + assert_eq!(&enc[..], wire); + } + + #[test] + fn literal_data_not_expecting_reply_frame() { + // Frame type 06, dest 0, src 7, len 2, HDR D9, data 01 00, DCRC 9F 16 + let wire: &[u8] = &[ + 0x55, 0xFF, 0x06, 0x00, 0x07, 0x00, 0x02, 0xD9, 0x01, 0x00, 0x9F, 0x16, + ]; + let (frame, consumed) = decode_frame(wire).expect("decode data frame"); + assert_eq!(consumed, 12); + assert_eq!(frame.frame_type, FrameType::BACnetDataNotExpectingReply); + assert_eq!(frame.destination, 0); + assert_eq!(frame.source, 7); + assert_eq!(&frame.data[..], &[0x01, 0x00]); + + let mut enc = BytesMut::new(); + encode_frame(&mut enc, &frame).unwrap(); + assert_eq!(&enc[..], wire); + } + + #[test] + fn truncated_header_need_more_or_err() { + let partial = [0x55, 0xFF, 0x00, 0x00, 0x07]; + assert!(decode_frame(&partial).is_err()); + assert_eq!(decode_frame_stream(&partial), StreamDecode::NeedMore); + } + + #[test] + fn truncated_data_crc_need_more() { + // Header complete for len=2 but missing data CRC octets + let partial = [0x55, 0xFF, 0x06, 0x00, 0x07, 0x00, 0x02, 0xD9, 0x01, 0x00]; + assert_eq!(decode_frame_stream(&partial), StreamDecode::NeedMore); + assert!(decode_frame(&partial).is_err()); } // ----------------------------------------------------------------------- From af07d18c977dffdae7f622b562cbd99bd0639ece Mon Sep 17 00:00:00 2001 From: bartlino Date: Sun, 30 Aug 2026 14:23:40 +0000 Subject: [PATCH 3/7] refactor(mstp): split frame tests to satisfy 700 LOC cap Move Clause 9 golden / stream tests into mstp_frame_tests.rs so mstp_frame.rs stays under the upstream file-size CI gate. Co-authored-by: Cursor --- crates/bacnet-transport/src/mstp_frame.rs | 574 +----------------- .../bacnet-transport/src/mstp_frame_tests.rs | 570 +++++++++++++++++ 2 files changed, 572 insertions(+), 572 deletions(-) create mode 100644 crates/bacnet-transport/src/mstp_frame_tests.rs diff --git a/crates/bacnet-transport/src/mstp_frame.rs b/crates/bacnet-transport/src/mstp_frame.rs index 2a9b2aa2..78f7b0ab 100644 --- a/crates/bacnet-transport/src/mstp_frame.rs +++ b/crates/bacnet-transport/src/mstp_frame.rs @@ -481,575 +481,5 @@ pub fn find_preamble(data: &[u8]) -> Option { } #[cfg(test)] -mod tests { - use super::*; - - // ----------------------------------------------------------------------- - // CRC tests — Clause 9.6 golden vectors (literal expected bytes; not from crc8/crc16) - // ----------------------------------------------------------------------- - - #[test] - fn crc8_clause9_header_vectors() { - // [frame_type, dest, src, len_hi, len_lo] -> header CRC - let vectors: &[([u8; 5], u8)] = &[ - ([0x00, 0x00, 0x07, 0x00, 0x00], 0x37), - ([0x00, 0x07, 0x00, 0x00, 0x00], 0x40), - ([0x01, 0x00, 0x07, 0x00, 0x00], 0xB1), - ([0x01, 0x07, 0x00, 0x00, 0x00], 0xC6), - ]; - for (header, expected) in vectors { - assert_eq!(crc8(header), *expected, "header={header:02X?}"); - let mut with_crc = header.to_vec(); - with_crc.push(*expected); - assert!(crc8_valid(&with_crc)); - assert_eq!(crc8_accumulate_all(&with_crc), HEADER_CRC_RESIDUAL); - } - } - - #[test] - fn crc8_one_bit_corruption_rejected() { - let header = [0x00, 0x00, 0x07, 0x00, 0x00]; - let mut with_crc = header.to_vec(); - with_crc.push(0x37); - assert!(crc8_valid(&with_crc)); - with_crc[0] ^= 0x01; - assert!(!crc8_valid(&with_crc)); - assert_ne!(crc8_accumulate_all(&with_crc), HEADER_CRC_RESIDUAL); - } - - #[test] - fn crc16_clause9_data_vector_01_00() { - // Data 01 00 → CRC 0x169F, wire order LS first: 9F 16 - assert_eq!(crc16(&[0x01, 0x00]), 0x169F); - let with_crc = [0x01, 0x00, 0x9F, 0x16]; - assert!(crc16_valid(&with_crc)); - assert_eq!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); - } - - #[test] - fn crc16_one_bit_corruption_rejected() { - let mut with_crc = [0x01, 0x00, 0x9F, 0x16]; - assert!(crc16_valid(&with_crc)); - with_crc[0] ^= 0x01; - assert!(!crc16_valid(&with_crc)); - assert_ne!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); - } - - #[test] - fn crc8_validate_round_trip() { - let data = [0x05, 0xFF, 0x03, 0x00, 0x0C]; - let crc = crc8(&data); - let mut with_crc = data.to_vec(); - with_crc.push(crc); - assert!(crc8_valid(&with_crc)); - assert_eq!(crc8_accumulate_all(&with_crc), HEADER_CRC_RESIDUAL); - } - - #[test] - fn crc16_validate_round_trip() { - let data = vec![0xAA; 100]; - let crc = crc16(&data); - let mut with_crc = data; - with_crc.push(crc as u8); - with_crc.push((crc >> 8) as u8); - assert!(crc16_valid(&with_crc)); - assert_eq!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); - } - - #[test] - fn literal_token_frame_0_from_7() { - // Live trunk Token: dest BASRT(0) <- FEC(7), header CRC 0x37 - let wire: &[u8] = &[0x55, 0xFF, 0x00, 0x00, 0x07, 0x00, 0x00, 0x37]; - let (frame, consumed) = decode_frame(wire).expect("decode Token 0<-7"); - assert_eq!(consumed, 8); - assert_eq!(frame.frame_type, FrameType::Token); - assert_eq!(frame.destination, 0); - assert_eq!(frame.source, 7); - assert!(frame.data.is_empty()); - - let mut enc = BytesMut::new(); - encode_frame(&mut enc, &frame).unwrap(); - assert_eq!(&enc[..], wire); - } - - #[test] - fn literal_data_not_expecting_reply_frame() { - // Frame type 06, dest 0, src 7, len 2, HDR D9, data 01 00, DCRC 9F 16 - let wire: &[u8] = &[ - 0x55, 0xFF, 0x06, 0x00, 0x07, 0x00, 0x02, 0xD9, 0x01, 0x00, 0x9F, 0x16, - ]; - let (frame, consumed) = decode_frame(wire).expect("decode data frame"); - assert_eq!(consumed, 12); - assert_eq!(frame.frame_type, FrameType::BACnetDataNotExpectingReply); - assert_eq!(frame.destination, 0); - assert_eq!(frame.source, 7); - assert_eq!(&frame.data[..], &[0x01, 0x00]); - - let mut enc = BytesMut::new(); - encode_frame(&mut enc, &frame).unwrap(); - assert_eq!(&enc[..], wire); - } - - #[test] - fn truncated_header_need_more_or_err() { - let partial = [0x55, 0xFF, 0x00, 0x00, 0x07]; - assert!(decode_frame(&partial).is_err()); - assert_eq!(decode_frame_stream(&partial), StreamDecode::NeedMore); - } - - #[test] - fn truncated_data_crc_need_more() { - // Header complete for len=2 but missing data CRC octets - let partial = [0x55, 0xFF, 0x06, 0x00, 0x07, 0x00, 0x02, 0xD9, 0x01, 0x00]; - assert_eq!(decode_frame_stream(&partial), StreamDecode::NeedMore); - assert!(decode_frame(&partial).is_err()); - } - - // ----------------------------------------------------------------------- - // Frame encode/decode tests - // ----------------------------------------------------------------------- - - #[test] - fn token_frame_round_trip() { - let frame = MstpFrame { - frame_type: FrameType::Token, - destination: 1, - source: 0, - data: Bytes::new(), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - // Token has no data, so: preamble(2) + header(5) + crc(1) = 8 - assert_eq!(buf.len(), 8); - assert_eq!(&buf[..2], &PREAMBLE); - - let (decoded, consumed) = decode_frame(&buf).unwrap(); - assert_eq!(consumed, 8); - assert_eq!(decoded, frame); - } - - #[test] - fn poll_for_master_round_trip() { - let frame = MstpFrame { - frame_type: FrameType::PollForMaster, - destination: 42, - source: 0, - data: Bytes::new(), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - let (decoded, _) = decode_frame(&buf).unwrap(); - assert_eq!(decoded, frame); - } - - #[test] - fn data_expecting_reply_round_trip() { - let npdu = vec![0x01, 0x00, 0x10, 0x02, 0x03, 0x04, 0x05]; - let frame = MstpFrame { - frame_type: FrameType::BACnetDataExpectingReply, - destination: 5, - source: 0, - data: Bytes::from(npdu.clone()), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - // preamble(2) + header(5) + hcrc(1) + data(7) + dcrc(2) = 17 - assert_eq!(buf.len(), 17); - - let (decoded, consumed) = decode_frame(&buf).unwrap(); - assert_eq!(consumed, 17); - assert_eq!(decoded.frame_type, FrameType::BACnetDataExpectingReply); - assert_eq!(decoded.destination, 5); - assert_eq!(decoded.source, 0); - assert_eq!(decoded.data, npdu); - } - - #[test] - fn data_not_expecting_reply_round_trip() { - let npdu = vec![0x01, 0x20, 0xFF, 0xFF, 0x00, 0xFF, 0x10, 0x08]; - let frame = MstpFrame { - frame_type: FrameType::BACnetDataNotExpectingReply, - destination: BROADCAST_MAC, - source: 3, - data: Bytes::from(npdu.clone()), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - let (decoded, _) = decode_frame(&buf).unwrap(); - assert_eq!(decoded, frame); - } - - #[test] - fn broadcast_destination() { - let frame = MstpFrame { - frame_type: FrameType::BACnetDataNotExpectingReply, - destination: BROADCAST_MAC, - source: 10, - data: Bytes::from_static(&[0x01, 0x00]), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - let (decoded, _) = decode_frame(&buf).unwrap(); - assert_eq!(decoded.destination, BROADCAST_MAC); - } - - #[test] - fn reply_to_poll_for_master_round_trip() { - let frame = MstpFrame { - frame_type: FrameType::ReplyToPollForMaster, - destination: 0, - source: 42, - data: Bytes::new(), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - let (decoded, _) = decode_frame(&buf).unwrap(); - assert_eq!(decoded, frame); - } - - #[test] - fn test_request_with_data_round_trip() { - let test_data = vec![0xDE, 0xAD, 0xBE, 0xEF]; - let frame = MstpFrame { - frame_type: FrameType::TestRequest, - destination: 5, - source: 0, - data: Bytes::from(test_data.clone()), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - let (decoded, _) = decode_frame(&buf).unwrap(); - assert_eq!(decoded.data, test_data); - } - - #[test] - fn decode_too_short() { - assert!(decode_frame(&[0x55, 0xFF, 0x00]).is_err()); - } - - #[test] - fn decode_bad_preamble() { - let data = [0x00, 0xFF, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00]; - assert!(decode_frame(&data).is_err()); - } - - #[test] - fn decode_bad_header_crc() { - let mut buf = BytesMut::new(); - let frame = MstpFrame { - frame_type: FrameType::Token, - destination: 1, - source: 0, - data: Bytes::new(), - }; - encode_frame(&mut buf, &frame).unwrap(); - // Corrupt header CRC (byte 7) - buf[7] ^= 0xFF; - assert!(decode_frame(&buf).is_err()); - } - - #[test] - fn decode_bad_data_crc() { - let mut buf = BytesMut::new(); - let frame = MstpFrame { - frame_type: FrameType::BACnetDataNotExpectingReply, - destination: 1, - source: 0, - data: Bytes::from_static(&[0x01, 0x00]), - }; - encode_frame(&mut buf, &frame).unwrap(); - // Corrupt last byte (data CRC high) - let last = buf.len() - 1; - buf[last] ^= 0xFF; - assert!(decode_frame(&buf).is_err()); - } - - #[test] - fn decode_truncated_data() { - let mut buf = BytesMut::new(); - let frame = MstpFrame { - frame_type: FrameType::BACnetDataExpectingReply, - destination: 5, - source: 0, - data: Bytes::from_static(&[0x01, 0x02, 0x03, 0x04]), - }; - encode_frame(&mut buf, &frame).unwrap(); - // Truncate: remove data CRC - buf.truncate(buf.len() - 2); - assert!(decode_frame(&buf).is_err()); - } - - #[test] - fn find_preamble_at_start() { - let data = [0x55, 0xFF, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00]; - assert_eq!(find_preamble(&data), Some(0)); - } - - #[test] - fn find_preamble_with_garbage() { - let data = [0x00, 0x00, 0x12, 0x55, 0xFF, 0x00, 0x01, 0x00]; - assert_eq!(find_preamble(&data), Some(3)); - } - - #[test] - fn find_preamble_none() { - let data = [0x00, 0x55, 0x00, 0xFF, 0x01]; - assert_eq!(find_preamble(&data), None); - } - - #[test] - fn frame_type_round_trip() { - for raw in 0..=0x07 { - let ft = FrameType::from_raw(raw); - assert_eq!(ft.to_raw(), raw); - } - // Unknown type - let ft = FrameType::from_raw(0x42); - assert_eq!(ft.to_raw(), 0x42); - assert_eq!(ft, FrameType::Unknown(0x42)); - } - - #[test] - fn frame_type_has_data() { - assert!(!FrameType::Token.has_data()); - assert!(!FrameType::PollForMaster.has_data()); - assert!(!FrameType::ReplyToPollForMaster.has_data()); - assert!(FrameType::TestRequest.has_data()); - assert!(FrameType::TestResponse.has_data()); - assert!(FrameType::BACnetDataExpectingReply.has_data()); - assert!(FrameType::BACnetDataNotExpectingReply.has_data()); - assert!(!FrameType::ReplyPostponed.has_data()); - } - - #[test] - fn large_data_frame() { - // Near-maximum data size - let npdu = vec![0xAA; 1024]; - let frame = MstpFrame { - frame_type: FrameType::BACnetDataNotExpectingReply, - destination: BROADCAST_MAC, - source: 0, - data: Bytes::from(npdu.clone()), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - let (decoded, _) = decode_frame(&buf).unwrap(); - assert_eq!(decoded.data, npdu); - } - - #[test] - fn encode_oversized_data_returns_error() { - let frame = MstpFrame { - frame_type: FrameType::BACnetDataNotExpectingReply, - destination: 1, - source: 0, - data: Bytes::from_static(&[0xAA; MAX_MPDU_DATA + 1]), - }; - let mut buf = BytesMut::new(); - assert!(encode_frame(&mut buf, &frame).is_err()); - } - - #[test] - fn decode_rejects_source_above_max_master() { - // Encode a valid frame then patch the source to 128 (above MAX_MASTER=127) - let frame = MstpFrame { - frame_type: FrameType::Token, - destination: 1, - source: 0, - data: Bytes::new(), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - // Patch source byte (offset 4) to 128 - buf[4] = 128; - // Recompute header CRC (bytes 2..7, CRC at byte 7) - let header_crc = crc8(&buf[2..7]); - buf[7] = header_crc; - - assert!(decode_frame(&buf).is_err()); - } - - #[test] - fn decode_rejects_broadcast_source() { - // Encode a valid frame then patch the source to BROADCAST_MAC (0xFF) - let frame = MstpFrame { - frame_type: FrameType::Token, - destination: 1, - source: 0, - data: Bytes::new(), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - // Patch source byte to 0xFF - buf[4] = BROADCAST_MAC; - // Recompute header CRC - let header_crc = crc8(&buf[2..7]); - buf[7] = header_crc; - - assert!(decode_frame(&buf).is_err()); - } - - #[test] - fn decode_accepts_max_master_source() { - // Source = MAX_MASTER (127) should be valid - let frame = MstpFrame { - frame_type: FrameType::Token, - destination: 1, - source: MAX_MASTER, - data: Bytes::new(), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - - let (decoded, _) = decode_frame(&buf).unwrap(); - assert_eq!(decoded.source, MAX_MASTER); - } - - // ----------------------------------------------------------------------- - // Streaming decode tests - // ----------------------------------------------------------------------- - - fn encode_token_frame() -> Vec { - let frame = MstpFrame { - frame_type: FrameType::Token, - destination: 1, - source: 0, - data: Bytes::new(), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - buf.to_vec() - } - - fn encode_data_frame() -> (MstpFrame, Vec) { - let frame = MstpFrame { - frame_type: FrameType::BACnetDataNotExpectingReply, - destination: 5, - source: 0, - data: Bytes::from_static(&[0x01, 0x02, 0x03, 0x04]), - }; - let mut buf = BytesMut::new(); - encode_frame(&mut buf, &frame).unwrap(); - (frame, buf.to_vec()) - } - - #[test] - fn stream_decode_token_complete() { - let wire = encode_token_frame(); - assert_eq!( - decode_frame_stream(&wire), - StreamDecode::Complete { - frame: MstpFrame { - frame_type: FrameType::Token, - destination: 1, - source: 0, - data: Bytes::new(), - }, - consumed: 8, - } - ); - } - - #[test] - fn stream_decode_split_header_then_body() { - let (_frame, wire) = encode_data_frame(); - let split = wire.len() - 3; - - assert_eq!(decode_frame_stream(&wire[..split]), StreamDecode::NeedMore); - - let mut assembled = wire[..split].to_vec(); - assembled.extend_from_slice(&wire[split..]); - let StreamDecode::Complete { frame, consumed } = decode_frame_stream(&assembled) else { - panic!("expected complete frame after reassembly"); - }; - assert_eq!(consumed, wire.len()); - assert_eq!(frame.data.len(), 4); - } - - #[test] - fn stream_decode_preamble_split_across_chunks() { - let wire = encode_token_frame(); - assert_eq!(decode_frame_stream(&wire[..1]), StreamDecode::NeedMore); - - let mut assembled = wire[..1].to_vec(); - assembled.extend_from_slice(&wire[1..]); - assert!(matches!( - decode_frame_stream(&assembled), - StreamDecode::Complete { .. } - )); - } - - #[test] - fn stream_decode_host_gap_simulation_without_clear() { - let wire = encode_token_frame(); - let mut buf = wire[..4].to_vec(); - assert_eq!(decode_frame_stream(&buf), StreamDecode::NeedMore); - - // Simulate a host gap far beyond wire T_frame_abort without clearing the buffer. - buf.extend_from_slice(&wire[4..]); - assert!(matches!( - decode_frame_stream(&buf), - StreamDecode::Complete { .. } - )); - } - - #[test] - fn retain_lone_preamble_byte_preserves_trailing_0x55() { - let mut buf = vec![0x00, 0x12, 0x55]; - retain_lone_preamble_byte(&mut buf); - assert_eq!(buf, vec![0x55]); - - let mut no_preamble = vec![0x00, 0x12, 0x34]; - retain_lone_preamble_byte(&mut no_preamble); - assert!(no_preamble.is_empty()); - } - - #[test] - fn stream_decode_bad_header_crc_invalid() { - let mut wire = encode_token_frame(); - wire[7] ^= 0xFF; - assert_eq!( - decode_frame_stream(&wire), - StreamDecode::Invalid { discard: 1 } - ); - } - - #[test] - fn stream_decode_bad_data_crc_invalid_discards_frame() { - let (_frame, mut wire) = encode_data_frame(); - let last = wire.len() - 1; - wire[last] ^= 0xFF; - assert_eq!( - decode_frame_stream(&wire), - StreamDecode::Invalid { - discard: wire.len() - } - ); - } - - #[test] - fn stream_decode_need_more_on_short_header() { - assert_eq!( - decode_frame_stream(&[0x55, 0xFF, 0x00, 0x01]), - StreamDecode::NeedMore - ); - } - - #[test] - fn stream_decode_invalid_on_bad_second_preamble_byte() { - assert_eq!( - decode_frame_stream(&[0x55, 0x00]), - StreamDecode::Invalid { discard: 1 } - ); - } -} +#[path = "mstp_frame_tests.rs"] +mod tests; diff --git a/crates/bacnet-transport/src/mstp_frame_tests.rs b/crates/bacnet-transport/src/mstp_frame_tests.rs new file mode 100644 index 00000000..83df7a38 --- /dev/null +++ b/crates/bacnet-transport/src/mstp_frame_tests.rs @@ -0,0 +1,570 @@ +use super::*; + +// ----------------------------------------------------------------------- +// CRC tests — Clause 9.6 golden vectors (literal expected bytes; not from crc8/crc16) +// ----------------------------------------------------------------------- + +#[test] +fn crc8_clause9_header_vectors() { + // [frame_type, dest, src, len_hi, len_lo] -> header CRC + let vectors: &[([u8; 5], u8)] = &[ + ([0x00, 0x00, 0x07, 0x00, 0x00], 0x37), + ([0x00, 0x07, 0x00, 0x00, 0x00], 0x40), + ([0x01, 0x00, 0x07, 0x00, 0x00], 0xB1), + ([0x01, 0x07, 0x00, 0x00, 0x00], 0xC6), + ]; + for (header, expected) in vectors { + assert_eq!(crc8(header), *expected, "header={header:02X?}"); + let mut with_crc = header.to_vec(); + with_crc.push(*expected); + assert!(crc8_valid(&with_crc)); + assert_eq!(crc8_accumulate_all(&with_crc), HEADER_CRC_RESIDUAL); + } +} + +#[test] +fn crc8_one_bit_corruption_rejected() { + let header = [0x00, 0x00, 0x07, 0x00, 0x00]; + let mut with_crc = header.to_vec(); + with_crc.push(0x37); + assert!(crc8_valid(&with_crc)); + with_crc[0] ^= 0x01; + assert!(!crc8_valid(&with_crc)); + assert_ne!(crc8_accumulate_all(&with_crc), HEADER_CRC_RESIDUAL); +} + +#[test] +fn crc16_clause9_data_vector_01_00() { + // Data 01 00 → CRC 0x169F, wire order LS first: 9F 16 + assert_eq!(crc16(&[0x01, 0x00]), 0x169F); + let with_crc = [0x01, 0x00, 0x9F, 0x16]; + assert!(crc16_valid(&with_crc)); + assert_eq!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); +} + +#[test] +fn crc16_one_bit_corruption_rejected() { + let mut with_crc = [0x01, 0x00, 0x9F, 0x16]; + assert!(crc16_valid(&with_crc)); + with_crc[0] ^= 0x01; + assert!(!crc16_valid(&with_crc)); + assert_ne!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); +} + +#[test] +fn crc8_validate_round_trip() { + let data = [0x05, 0xFF, 0x03, 0x00, 0x0C]; + let crc = crc8(&data); + let mut with_crc = data.to_vec(); + with_crc.push(crc); + assert!(crc8_valid(&with_crc)); + assert_eq!(crc8_accumulate_all(&with_crc), HEADER_CRC_RESIDUAL); +} + +#[test] +fn crc16_validate_round_trip() { + let data = vec![0xAA; 100]; + let crc = crc16(&data); + let mut with_crc = data; + with_crc.push(crc as u8); + with_crc.push((crc >> 8) as u8); + assert!(crc16_valid(&with_crc)); + assert_eq!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); +} + +#[test] +fn literal_token_frame_0_from_7() { + // Live trunk Token: dest BASRT(0) <- FEC(7), header CRC 0x37 + let wire: &[u8] = &[0x55, 0xFF, 0x00, 0x00, 0x07, 0x00, 0x00, 0x37]; + let (frame, consumed) = decode_frame(wire).expect("decode Token 0<-7"); + assert_eq!(consumed, 8); + assert_eq!(frame.frame_type, FrameType::Token); + assert_eq!(frame.destination, 0); + assert_eq!(frame.source, 7); + assert!(frame.data.is_empty()); + + let mut enc = BytesMut::new(); + encode_frame(&mut enc, &frame).unwrap(); + assert_eq!(&enc[..], wire); +} + +#[test] +fn literal_data_not_expecting_reply_frame() { + // Frame type 06, dest 0, src 7, len 2, HDR D9, data 01 00, DCRC 9F 16 + let wire: &[u8] = &[ + 0x55, 0xFF, 0x06, 0x00, 0x07, 0x00, 0x02, 0xD9, 0x01, 0x00, 0x9F, 0x16, + ]; + let (frame, consumed) = decode_frame(wire).expect("decode data frame"); + assert_eq!(consumed, 12); + assert_eq!(frame.frame_type, FrameType::BACnetDataNotExpectingReply); + assert_eq!(frame.destination, 0); + assert_eq!(frame.source, 7); + assert_eq!(&frame.data[..], &[0x01, 0x00]); + + let mut enc = BytesMut::new(); + encode_frame(&mut enc, &frame).unwrap(); + assert_eq!(&enc[..], wire); +} + +#[test] +fn truncated_header_need_more_or_err() { + let partial = [0x55, 0xFF, 0x00, 0x00, 0x07]; + assert!(decode_frame(&partial).is_err()); + assert_eq!(decode_frame_stream(&partial), StreamDecode::NeedMore); +} + +#[test] +fn truncated_data_crc_need_more() { + // Header complete for len=2 but missing data CRC octets + let partial = [0x55, 0xFF, 0x06, 0x00, 0x07, 0x00, 0x02, 0xD9, 0x01, 0x00]; + assert_eq!(decode_frame_stream(&partial), StreamDecode::NeedMore); + assert!(decode_frame(&partial).is_err()); +} + +// ----------------------------------------------------------------------- +// Frame encode/decode tests +// ----------------------------------------------------------------------- + +#[test] +fn token_frame_round_trip() { + let frame = MstpFrame { + frame_type: FrameType::Token, + destination: 1, + source: 0, + data: Bytes::new(), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + // Token has no data, so: preamble(2) + header(5) + crc(1) = 8 + assert_eq!(buf.len(), 8); + assert_eq!(&buf[..2], &PREAMBLE); + + let (decoded, consumed) = decode_frame(&buf).unwrap(); + assert_eq!(consumed, 8); + assert_eq!(decoded, frame); +} + +#[test] +fn poll_for_master_round_trip() { + let frame = MstpFrame { + frame_type: FrameType::PollForMaster, + destination: 42, + source: 0, + data: Bytes::new(), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + let (decoded, _) = decode_frame(&buf).unwrap(); + assert_eq!(decoded, frame); +} + +#[test] +fn data_expecting_reply_round_trip() { + let npdu = vec![0x01, 0x00, 0x10, 0x02, 0x03, 0x04, 0x05]; + let frame = MstpFrame { + frame_type: FrameType::BACnetDataExpectingReply, + destination: 5, + source: 0, + data: Bytes::from(npdu.clone()), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + // preamble(2) + header(5) + hcrc(1) + data(7) + dcrc(2) = 17 + assert_eq!(buf.len(), 17); + + let (decoded, consumed) = decode_frame(&buf).unwrap(); + assert_eq!(consumed, 17); + assert_eq!(decoded.frame_type, FrameType::BACnetDataExpectingReply); + assert_eq!(decoded.destination, 5); + assert_eq!(decoded.source, 0); + assert_eq!(decoded.data, npdu); +} + +#[test] +fn data_not_expecting_reply_round_trip() { + let npdu = vec![0x01, 0x20, 0xFF, 0xFF, 0x00, 0xFF, 0x10, 0x08]; + let frame = MstpFrame { + frame_type: FrameType::BACnetDataNotExpectingReply, + destination: BROADCAST_MAC, + source: 3, + data: Bytes::from(npdu.clone()), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + let (decoded, _) = decode_frame(&buf).unwrap(); + assert_eq!(decoded, frame); +} + +#[test] +fn broadcast_destination() { + let frame = MstpFrame { + frame_type: FrameType::BACnetDataNotExpectingReply, + destination: BROADCAST_MAC, + source: 10, + data: Bytes::from_static(&[0x01, 0x00]), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + let (decoded, _) = decode_frame(&buf).unwrap(); + assert_eq!(decoded.destination, BROADCAST_MAC); +} + +#[test] +fn reply_to_poll_for_master_round_trip() { + let frame = MstpFrame { + frame_type: FrameType::ReplyToPollForMaster, + destination: 0, + source: 42, + data: Bytes::new(), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + let (decoded, _) = decode_frame(&buf).unwrap(); + assert_eq!(decoded, frame); +} + +#[test] +fn test_request_with_data_round_trip() { + let test_data = vec![0xDE, 0xAD, 0xBE, 0xEF]; + let frame = MstpFrame { + frame_type: FrameType::TestRequest, + destination: 5, + source: 0, + data: Bytes::from(test_data.clone()), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + let (decoded, _) = decode_frame(&buf).unwrap(); + assert_eq!(decoded.data, test_data); +} + +#[test] +fn decode_too_short() { + assert!(decode_frame(&[0x55, 0xFF, 0x00]).is_err()); +} + +#[test] +fn decode_bad_preamble() { + let data = [0x00, 0xFF, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00]; + assert!(decode_frame(&data).is_err()); +} + +#[test] +fn decode_bad_header_crc() { + let mut buf = BytesMut::new(); + let frame = MstpFrame { + frame_type: FrameType::Token, + destination: 1, + source: 0, + data: Bytes::new(), + }; + encode_frame(&mut buf, &frame).unwrap(); + // Corrupt header CRC (byte 7) + buf[7] ^= 0xFF; + assert!(decode_frame(&buf).is_err()); +} + +#[test] +fn decode_bad_data_crc() { + let mut buf = BytesMut::new(); + let frame = MstpFrame { + frame_type: FrameType::BACnetDataNotExpectingReply, + destination: 1, + source: 0, + data: Bytes::from_static(&[0x01, 0x00]), + }; + encode_frame(&mut buf, &frame).unwrap(); + // Corrupt last byte (data CRC high) + let last = buf.len() - 1; + buf[last] ^= 0xFF; + assert!(decode_frame(&buf).is_err()); +} + +#[test] +fn decode_truncated_data() { + let mut buf = BytesMut::new(); + let frame = MstpFrame { + frame_type: FrameType::BACnetDataExpectingReply, + destination: 5, + source: 0, + data: Bytes::from_static(&[0x01, 0x02, 0x03, 0x04]), + }; + encode_frame(&mut buf, &frame).unwrap(); + // Truncate: remove data CRC + buf.truncate(buf.len() - 2); + assert!(decode_frame(&buf).is_err()); +} + +#[test] +fn find_preamble_at_start() { + let data = [0x55, 0xFF, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00]; + assert_eq!(find_preamble(&data), Some(0)); +} + +#[test] +fn find_preamble_with_garbage() { + let data = [0x00, 0x00, 0x12, 0x55, 0xFF, 0x00, 0x01, 0x00]; + assert_eq!(find_preamble(&data), Some(3)); +} + +#[test] +fn find_preamble_none() { + let data = [0x00, 0x55, 0x00, 0xFF, 0x01]; + assert_eq!(find_preamble(&data), None); +} + +#[test] +fn frame_type_round_trip() { + for raw in 0..=0x07 { + let ft = FrameType::from_raw(raw); + assert_eq!(ft.to_raw(), raw); + } + // Unknown type + let ft = FrameType::from_raw(0x42); + assert_eq!(ft.to_raw(), 0x42); + assert_eq!(ft, FrameType::Unknown(0x42)); +} + +#[test] +fn frame_type_has_data() { + assert!(!FrameType::Token.has_data()); + assert!(!FrameType::PollForMaster.has_data()); + assert!(!FrameType::ReplyToPollForMaster.has_data()); + assert!(FrameType::TestRequest.has_data()); + assert!(FrameType::TestResponse.has_data()); + assert!(FrameType::BACnetDataExpectingReply.has_data()); + assert!(FrameType::BACnetDataNotExpectingReply.has_data()); + assert!(!FrameType::ReplyPostponed.has_data()); +} + +#[test] +fn large_data_frame() { + // Near-maximum data size + let npdu = vec![0xAA; 1024]; + let frame = MstpFrame { + frame_type: FrameType::BACnetDataNotExpectingReply, + destination: BROADCAST_MAC, + source: 0, + data: Bytes::from(npdu.clone()), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + let (decoded, _) = decode_frame(&buf).unwrap(); + assert_eq!(decoded.data, npdu); +} + +#[test] +fn encode_oversized_data_returns_error() { + let frame = MstpFrame { + frame_type: FrameType::BACnetDataNotExpectingReply, + destination: 1, + source: 0, + data: Bytes::from_static(&[0xAA; MAX_MPDU_DATA + 1]), + }; + let mut buf = BytesMut::new(); + assert!(encode_frame(&mut buf, &frame).is_err()); +} + +#[test] +fn decode_rejects_source_above_max_master() { + // Encode a valid frame then patch the source to 128 (above MAX_MASTER=127) + let frame = MstpFrame { + frame_type: FrameType::Token, + destination: 1, + source: 0, + data: Bytes::new(), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + // Patch source byte (offset 4) to 128 + buf[4] = 128; + // Recompute header CRC (bytes 2..7, CRC at byte 7) + let header_crc = crc8(&buf[2..7]); + buf[7] = header_crc; + + assert!(decode_frame(&buf).is_err()); +} + +#[test] +fn decode_rejects_broadcast_source() { + // Encode a valid frame then patch the source to BROADCAST_MAC (0xFF) + let frame = MstpFrame { + frame_type: FrameType::Token, + destination: 1, + source: 0, + data: Bytes::new(), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + // Patch source byte to 0xFF + buf[4] = BROADCAST_MAC; + // Recompute header CRC + let header_crc = crc8(&buf[2..7]); + buf[7] = header_crc; + + assert!(decode_frame(&buf).is_err()); +} + +#[test] +fn decode_accepts_max_master_source() { + // Source = MAX_MASTER (127) should be valid + let frame = MstpFrame { + frame_type: FrameType::Token, + destination: 1, + source: MAX_MASTER, + data: Bytes::new(), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + + let (decoded, _) = decode_frame(&buf).unwrap(); + assert_eq!(decoded.source, MAX_MASTER); +} + +// ----------------------------------------------------------------------- +// Streaming decode tests +// ----------------------------------------------------------------------- + +fn encode_token_frame() -> Vec { + let frame = MstpFrame { + frame_type: FrameType::Token, + destination: 1, + source: 0, + data: Bytes::new(), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + buf.to_vec() +} + +fn encode_data_frame() -> (MstpFrame, Vec) { + let frame = MstpFrame { + frame_type: FrameType::BACnetDataNotExpectingReply, + destination: 5, + source: 0, + data: Bytes::from_static(&[0x01, 0x02, 0x03, 0x04]), + }; + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + (frame, buf.to_vec()) +} + +#[test] +fn stream_decode_token_complete() { + let wire = encode_token_frame(); + assert_eq!( + decode_frame_stream(&wire), + StreamDecode::Complete { + frame: MstpFrame { + frame_type: FrameType::Token, + destination: 1, + source: 0, + data: Bytes::new(), + }, + consumed: 8, + } + ); +} + +#[test] +fn stream_decode_split_header_then_body() { + let (_frame, wire) = encode_data_frame(); + let split = wire.len() - 3; + + assert_eq!(decode_frame_stream(&wire[..split]), StreamDecode::NeedMore); + + let mut assembled = wire[..split].to_vec(); + assembled.extend_from_slice(&wire[split..]); + let StreamDecode::Complete { frame, consumed } = decode_frame_stream(&assembled) else { + panic!("expected complete frame after reassembly"); + }; + assert_eq!(consumed, wire.len()); + assert_eq!(frame.data.len(), 4); +} + +#[test] +fn stream_decode_preamble_split_across_chunks() { + let wire = encode_token_frame(); + assert_eq!(decode_frame_stream(&wire[..1]), StreamDecode::NeedMore); + + let mut assembled = wire[..1].to_vec(); + assembled.extend_from_slice(&wire[1..]); + assert!(matches!( + decode_frame_stream(&assembled), + StreamDecode::Complete { .. } + )); +} + +#[test] +fn stream_decode_host_gap_simulation_without_clear() { + let wire = encode_token_frame(); + let mut buf = wire[..4].to_vec(); + assert_eq!(decode_frame_stream(&buf), StreamDecode::NeedMore); + + // Simulate a host gap far beyond wire T_frame_abort without clearing the buffer. + buf.extend_from_slice(&wire[4..]); + assert!(matches!( + decode_frame_stream(&buf), + StreamDecode::Complete { .. } + )); +} + +#[test] +fn retain_lone_preamble_byte_preserves_trailing_0x55() { + let mut buf = vec![0x00, 0x12, 0x55]; + retain_lone_preamble_byte(&mut buf); + assert_eq!(buf, vec![0x55]); + + let mut no_preamble = vec![0x00, 0x12, 0x34]; + retain_lone_preamble_byte(&mut no_preamble); + assert!(no_preamble.is_empty()); +} + +#[test] +fn stream_decode_bad_header_crc_invalid() { + let mut wire = encode_token_frame(); + wire[7] ^= 0xFF; + assert_eq!( + decode_frame_stream(&wire), + StreamDecode::Invalid { discard: 1 } + ); +} + +#[test] +fn stream_decode_bad_data_crc_invalid_discards_frame() { + let (_frame, mut wire) = encode_data_frame(); + let last = wire.len() - 1; + wire[last] ^= 0xFF; + assert_eq!( + decode_frame_stream(&wire), + StreamDecode::Invalid { + discard: wire.len() + } + ); +} + +#[test] +fn stream_decode_need_more_on_short_header() { + assert_eq!( + decode_frame_stream(&[0x55, 0xFF, 0x00, 0x01]), + StreamDecode::NeedMore + ); +} + +#[test] +fn stream_decode_invalid_on_bad_second_preamble_byte() { + assert_eq!( + decode_frame_stream(&[0x55, 0x00]), + StreamDecode::Invalid { discard: 1 } + ); +} From 99429221f1e8a5c31a8a3ce1f1f3820cb9232a92 Mon Sep 17 00:00:00 2001 From: bartlino Date: Sun, 30 Aug 2026 16:21:10 +0000 Subject: [PATCH 4/7] fix(mstp): conform master token and PFM transitions to Clause 9.5.6 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DONE_WITH_TOKEN no longer reverse-scans from NS+1 or emits Token TS→TS. Add done_with_token(), FindNewSuccessor via PFM after Nretry_token, and regression tests A–E for multi-master coexistence. Co-authored-by: Cursor --- .../src/mstp/clause956_tests.rs | 241 ++++++++++++++++++ crates/bacnet-transport/src/mstp/mod.rs | 239 +++++++++++------ crates/bacnet-transport/src/mstp/port.rs | 66 +++-- crates/bacnet-transport/src/mstp/tests.rs | 175 ++++++------- 4 files changed, 516 insertions(+), 205 deletions(-) create mode 100644 crates/bacnet-transport/src/mstp/clause956_tests.rs diff --git a/crates/bacnet-transport/src/mstp/clause956_tests.rs b/crates/bacnet-transport/src/mstp/clause956_tests.rs new file mode 100644 index 00000000..fbe4b0e5 --- /dev/null +++ b/crates/bacnet-transport/src/mstp/clause956_tests.rs @@ -0,0 +1,241 @@ +//! Clause 9.5.6 regression suite (token / PFM / coexistence). +//! +//! These tests replace the reversed-scan expectations (PFM beginning at NS+1) +//! that allowed the invalid ring 0→3→0 and excluded FEC MAC 7. + +use super::*; +use bytes::Bytes; +use tokio::sync::mpsc; + +fn cfg(ts: u8, max_master: u8) -> MstpConfig { + MstpConfig { + this_station: ts, + max_master, + max_info_frames: 1, + baud_rate: 38400, + } +} + +fn deliver_token(node: &mut MasterNode, from: u8) { + let (tx, _rx) = mpsc::channel(4); + let frame = MstpFrame { + frame_type: FrameType::Token, + destination: node.config.this_station, + source: from, + data: Bytes::new(), + }; + let _ = node.handle_received_frame(&frame, &tx); +} + +/// A. Unknown successor with queued Who-Is — no Token 3→3; PFM to 4. +#[test] +fn a_unknown_successor_with_queued_who_is() { + let mut node = MasterNode::new(cfg(3, 7)).unwrap(); + assert_eq!(node.next_station, 3); + assert_eq!(node.poll_station, 3); + assert_eq!(node.token_count, NPOLL); + assert!(!node.sole_master); + + // Queued application frame (Who-Is-style NPDU) + node.queue_npdu(BROADCAST_MAC, Bytes::from_static(&[0x01, 0x20, 0xFF, 0xFF])) + .unwrap(); + + deliver_token(&mut node, 0); + assert_eq!(node.state, MasterState::UseToken); + + let app = node.use_token(); + assert_eq!(app.frame_type, FrameType::BACnetDataNotExpectingReply); + assert_eq!(node.state, MasterState::DoneWithToken); + + let next = node.done_with_token(); + assert_eq!(next.frame_type, FrameType::PollForMaster); + assert_eq!(next.source, 3); + assert_eq!(next.destination, 4); + assert_ne!( + (next.frame_type, next.destination), + (FrameType::Token, 3), + "forbidden Token 3→3" + ); +} + +/// B. Established three-master ring 0→3→7→0 for ≥2000 rotations. +#[test] +fn b_three_master_ring_2000_rotations() { + let mut nodes = [ + MasterNode::new(cfg(0, 7)).unwrap(), + MasterNode::new(cfg(3, 7)).unwrap(), + MasterNode::new(cfg(7, 7)).unwrap(), + ]; + // Established successors + nodes[0].next_station = 3; + nodes[0].poll_station = 0; + nodes[0].token_count = 0; + nodes[1].next_station = 7; + nodes[1].poll_station = 3; + nodes[1].token_count = 0; + nodes[2].next_station = 0; + nodes[2].poll_station = 7; + nodes[2].token_count = 0; + + let macs = [0u8, 3, 7]; + let mut idx = 0usize; // token holder index into macs + let mut receipts = [0u32; 3]; + let mut forbidden_self_token = 0u32; + + for _ in 0..2000 { + let holder = &mut nodes[idx]; + deliver_token(holder, macs[(idx + 2) % 3]); + receipts[idx] += 1; + + // Drain this token use (app empty → done_with_token may PFM then timeout) + let mut guard = 0; + loop { + guard += 1; + assert!(guard < 64, "token-use did not terminate"); + + let frame = if holder.state == MasterState::DoneWithToken { + holder.done_with_token() + } else if holder.state == MasterState::UseToken { + holder.use_token() + } else if holder.state == MasterState::PollForMaster { + // Maintenance PFM no reply — DoneWithPFM passes to known NS + holder.poll_timeout() + } else { + break; + }; + + if frame.frame_type == FrameType::Token { + if frame.source == frame.destination { + forbidden_self_token += 1; + } + assert_eq!(frame.destination, holder.next_station); + // Advance ring to destination + idx = macs + .iter() + .position(|&m| m == frame.destination) + .expect("token dest must be a ring member"); + break; + } + // PFM: stay on holder until poll_timeout produces Token + } + } + + assert_eq!(forbidden_self_token, 0, "no Token source==destination"); + assert!(receipts.iter().all(|&n| n > 0), "every master must receive token: {receipts:?}"); + // Rough fairness: each master should see a large share of 2000 holds + for (i, &n) in receipts.iter().enumerate() { + assert!( + n > 400, + "master {} starved (receipts={receipts:?})", + macs[i] + ); + } +} + +/// C. Maintenance polls only addresses in (TS, NS) — never PFM 3→0 / Token 3→0 / Token 3→3. +#[test] +fn c_maintenance_polling_gap_only() { + let mut node = MasterNode::new(cfg(3, 7)).unwrap(); + node.next_station = 7; + node.poll_station = 3; + node.token_count = NPOLL.saturating_sub(1); // force maintenance + node.state = MasterState::DoneWithToken; + + let expected = [4u8, 5, 6]; + for &ps in &expected { + let pfm = node.done_with_token(); + assert_eq!(pfm.frame_type, FrameType::PollForMaster); + assert_eq!(pfm.source, 3); + assert_eq!(pfm.destination, ps); + assert_ne!(pfm.destination, 0, "forbidden PFM 3→0"); + + let token = node.poll_timeout(); + assert_eq!(token.frame_type, FrameType::Token); + assert_eq!(token.destination, 7); + assert_ne!(token.destination, 0, "forbidden Token 3→0 while NS=7"); + assert_ne!(token.destination, 3, "forbidden Token 3→3"); + + // Next maintenance opportunity + node.token_count = NPOLL.saturating_sub(1); + node.state = MasterState::DoneWithToken; + node.frame_count = node.config.max_info_frames; + } + + // Fourth opportunity: next_ps == NS → ResetMaintenancePFM + let token = node.done_with_token(); + assert_eq!(token.frame_type, FrameType::Token); + assert_eq!(token.destination, 7); + assert_eq!(node.poll_station, 3); + assert_eq!(node.token_count, 1); +} + +/// D. New master in gap (MAC 5) joins via ReplyToPFM. +#[test] +fn d_new_master_in_gap() { + let (tx, _rx) = mpsc::channel(4); + let mut node = MasterNode::new(cfg(3, 7)).unwrap(); + node.next_station = 7; + node.poll_station = 3; + node.token_count = NPOLL.saturating_sub(1); + node.state = MasterState::DoneWithToken; + + // Skip 4 (timeout), then poll 5 + let pfm4 = node.done_with_token(); + assert_eq!(pfm4.destination, 4); + let _ = node.poll_timeout(); + node.token_count = NPOLL.saturating_sub(1); + node.state = MasterState::DoneWithToken; + node.frame_count = node.config.max_info_frames; + + let pfm5 = node.done_with_token(); + assert_eq!(pfm5.frame_type, FrameType::PollForMaster); + assert_eq!(pfm5.destination, 5); + + let reply = MstpFrame { + frame_type: FrameType::ReplyToPollForMaster, + destination: 3, + source: 5, + data: Bytes::new(), + }; + let out = node.handle_received_frame(&reply, &tx).expect("Token to new NS"); + assert_eq!(node.next_station, 5); + assert_eq!(node.poll_station, 3); + assert_eq!(node.token_count, 0); + assert_eq!(out.frame_type, FrameType::Token); + assert_eq!(out.destination, 5); + + // Eventual ring includes 3→5; 5 would point at 7, 7→0, 0→3 (sim check NS only) + assert_eq!(node.next_station, 5); +} + +/// E. Failed successor: Nretry_token Token retries then PFM search after failed NS. +#[test] +fn e_failed_successor_uses_pfm_not_blind_tokens() { + let mut node = MasterNode::new(cfg(3, 7)).unwrap(); + node.next_station = 7; + node.poll_station = 3; + node.token_count = 1; + let first = node.pass_token(); + assert_eq!(first.frame_type, FrameType::Token); + assert_eq!(first.destination, 7); + + // Retry exactly Nretry_token times + let retry = node.pass_token_timeout().expect("retry Token"); + assert_eq!(retry.frame_type, FrameType::Token); + assert_eq!(retry.destination, 7); + assert_eq!(node.retry_token_count, N_RETRY_TOKEN); + + // Then FindNewSuccessor: PS = NS+1 = 0? next_addr(7,7)=0; NS=TS=3; PFM to 0 + // Wait — Max_Master=7, next_addr(7)=0. User: "PFM search beginning after failed NS" + let find = node.pass_token_timeout().expect("PFM after retries"); + assert_eq!(find.frame_type, FrameType::PollForMaster); + assert_eq!(find.destination, 0); // NS+1 after failed 7 with max_master=7 + assert_eq!(node.next_station, 3); // NS = TS + assert_ne!(find.frame_type, FrameType::Token); + + // Must never blindly Token to unverified addresses after failure + assert!(node.pass_token_timeout().is_none() || { + // still in PollForMaster — timeout handled by poll_timeout, not more Tokens + true + }); +} diff --git a/crates/bacnet-transport/src/mstp/mod.rs b/crates/bacnet-transport/src/mstp/mod.rs index 80a74cc4..745ec163 100644 --- a/crates/bacnet-transport/src/mstp/mod.rs +++ b/crates/bacnet-transport/src/mstp/mod.rs @@ -158,8 +158,6 @@ pub struct MasterNode { /// How many tokens between PollForMaster attempts. const NPOLL: u8 = 50; -/// Max retries for PollForMaster. -const MAX_POLL_RETRIES: u8 = 3; impl MasterNode { pub fn new(config: MstpConfig) -> Result { @@ -213,10 +211,11 @@ impl MasterNode { return None; } debug!(src = frame.source, "received token"); + // Clause 9.5.6 ReceivedToken: FrameCount=0, SoleMaster=false, + // enter USE_TOKEN. TokenCount advances only in DONE_WITH_TOKEN. self.sole_master = false; self.state = MasterState::UseToken; self.frame_count = 0; - self.token_count = self.token_count.wrapping_add(1); self.retry_token_count = 0; } None @@ -238,13 +237,14 @@ impl MasterNode { if self.state == MasterState::PollForMaster && frame.destination == self.config.this_station { + // ReplyToPFM: NS=source, PS=TS, TokenCount=0, pass to new NS debug!(src = frame.source, "PFM reply — new successor"); self.next_station = frame.source; self.sole_master = false; self.poll_station = self.config.this_station; self.token_count = 0; self.poll_count = 0; - // Send Token to the new successor and enter PassToken + self.retry_token_count = 0; return Some(self.pass_token()); } None @@ -370,7 +370,7 @@ impl MasterNode { } // If we've hit max_info_frames, transition to DoneWithToken - // so the caller knows to pass the token next. + // so the caller knows to run done_with_token() next. if self.frame_count >= self.config.max_info_frames { self.state = MasterState::DoneWithToken; } @@ -382,127 +382,200 @@ impl MasterNode { data: npdu, }; } - } else { - // Frame limit reached — transition to DoneWithToken and pass immediately. - self.state = MasterState::DoneWithToken; - return self.pass_token(); } - // Time to poll? - if self.token_count >= NPOLL { - self.token_count = 0; - self.state = MasterState::PollForMaster; - // Scan from this_station+1 through next_station-1 (wrapping at max_master) - // Start polling at next_station (the first address after our known successor range) - self.poll_station = next_addr(self.next_station, self.config.max_master); - // If poll_station wraps to us, skip — we already know about next_station - if self.poll_station == self.config.this_station { - // Only us and next_station exist; no gap to scan + // Nothing (more) to send — Clause 9.5.6 DONE_WITH_TOKEN transitions. + self.state = MasterState::DoneWithToken; + self.done_with_token() + } + + /// Clause 9.5.6 DONE_WITH_TOKEN state transitions. + /// + /// Chooses SendAnotherFrame / NextStationUnknown / SendToken / + /// SendMaintenancePFM / ResetMaintenancePFM / SoleMaster paths. + /// Never emits a Token with source == destination. + pub fn done_with_token(&mut self) -> MstpFrame { + let ts = self.config.this_station; + let max_master = self.config.max_master; + + // SoleMaster can reuse the token for several DONE_WITH_TOKEN iterations + // without putting a frame on the wire — loop until a real frame exists. + loop { + let next_ts = next_addr(ts, max_master); + let next_ps = next_addr(self.poll_station, max_master); + + // SendAnotherFrame + if self.frame_count < self.config.max_info_frames && !self.tx_queue.is_empty() { + self.state = MasterState::UseToken; + return self.use_token(); + } + + // NextStationUnknown (Addendum 135-2008v-1): NS == TS and not sole master + if !self.sole_master && self.next_station == ts { + self.poll_station = next_ts; + self.retry_token_count = 0; + self.state = MasterState::PollForMaster; + return MstpFrame { + frame_type: FrameType::PollForMaster, + destination: self.poll_station, + source: ts, + data: Bytes::new(), + }; + } + + // SendToken while TokenCount < Npoll - 1 + if self.token_count < NPOLL.saturating_sub(1) { + if self.sole_master && self.next_station != next_ts { + // SoleMaster: reuse token; never Token(TS→TS) + self.frame_count = 0; + self.token_count = self.token_count.saturating_add(1); + self.state = MasterState::UseToken; + continue; + } + // SendToken (also when NS == TS+1 — no gap to poll) + self.token_count = self.token_count.saturating_add(1); + return self.pass_token(); + } + + // Maintenance / reset when TokenCount >= Npoll - 1 + if next_ps == self.next_station { + if self.sole_master { + // SoleMasterRestartMaintenancePFM + self.poll_station = next_addr(self.next_station, max_master); + self.next_station = ts; + self.retry_token_count = 0; + self.token_count = 1; + self.state = MasterState::PollForMaster; + return MstpFrame { + frame_type: FrameType::PollForMaster, + destination: self.poll_station, + source: ts, + data: Bytes::new(), + }; + } + // ResetMaintenancePFM: PS = TS, Token to NS, TokenCount = 1 + self.poll_station = ts; + self.retry_token_count = 0; + self.token_count = 1; + self.event_count = 0; return self.pass_token(); } + + // SendMaintenancePFM: advance PS toward NS only (never begin at NS+1) + self.poll_station = next_ps; + self.retry_token_count = 0; + self.state = MasterState::PollForMaster; return MstpFrame { frame_type: FrameType::PollForMaster, destination: self.poll_station, - source: self.config.this_station, + source: ts, data: Bytes::new(), }; } - - // Pass the token - self.pass_token() } /// Generate a token-pass frame to next_station. + /// + /// Must not be used when `next_station == this_station` (Clause 9.5.6.5 + /// requires PFM to TS+1 instead — handled by [`Self::done_with_token`]). pub fn pass_token(&mut self) -> MstpFrame { + let ts = self.config.this_station; + debug_assert_ne!( + self.next_station, ts, + "pass_token must not emit Token TS→TS; use done_with_token / PFM" + ); self.state = MasterState::PassToken; self.retry_token_count = 0; + self.event_count = 0; MstpFrame { frame_type: FrameType::Token, destination: self.next_station, - source: self.config.this_station, + source: ts, data: Bytes::new(), } } - /// Handle PassToken timeout. + /// Handle PassToken timeout (Clause 9.5.6 PASS_TOKEN). /// - /// Called when T_usage_timeout expires after passing the token. - /// Returns a frame to send (retry Token or PFM), or None if we should go to Idle. + /// After `Nretry_token` Token retries, FindNewSuccessor: PS = NS+1, NS = TS, + /// send PFM — never Token frames to unverified addresses. pub fn pass_token_timeout(&mut self) -> Option { let ts = self.config.this_station; + let max_master = self.config.max_master; if self.retry_token_count < N_RETRY_TOKEN { - // RetrySendToken: resend Token to NS + // RetrySendToken: resend Token to NS exactly Nretry_token times self.retry_token_count += 1; + self.event_count = 0; Some(MstpFrame { frame_type: FrameType::Token, destination: self.next_station, source: ts, data: Bytes::new(), }) - } else if self.next_station == ts { - // FindNewSuccessorUnknown: NS wrapped back to TS - // No other stations found — go to NoToken to try again - self.state = MasterState::NoToken; - None } else { - // FindNewSuccessor: NS didn't respond, try next address - self.next_station = next_addr(self.next_station, self.config.max_master); - if self.next_station == ts { - // Wrapped all the way around — declare sole master + // FindNewSuccessor + let failed_ns = self.next_station; + self.poll_station = next_addr(failed_ns, max_master); + self.next_station = ts; + self.retry_token_count = 0; + self.token_count = 0; + if self.poll_station == ts { + // Would PFM self — declare sole master without Token TS→TS self.sole_master = true; self.state = MasterState::UseToken; self.frame_count = 0; - None - } else { - // Try passing token to the new next_station - self.retry_token_count = 0; - Some(MstpFrame { - frame_type: FrameType::Token, - destination: self.next_station, - source: ts, - data: Bytes::new(), - }) + return None; } + self.state = MasterState::PollForMaster; + Some(MstpFrame { + frame_type: FrameType::PollForMaster, + destination: self.poll_station, + source: ts, + data: Bytes::new(), + }) } } - /// Handle PollForMaster timeout (no reply received). + /// Handle PollForMaster timeout (no ReplyToPFM). + /// + /// Known NS → DoneWithPFM: pass token to NS (one PFM per token use). + /// Unknown NS → SendNextPFM or DeclareSoleMaster (no Token TS→TS). pub fn poll_timeout(&mut self) -> MstpFrame { - self.poll_count += 1; - if self.poll_count >= MAX_POLL_RETRIES { - // No one answered — move to next poll station - self.poll_count = 0; - self.poll_station = next_addr(self.poll_station, self.config.max_master); - if self.poll_station == self.config.this_station { - // We've scanned the entire range — no other stations - if self.next_station == self.config.this_station { - // Sole master: claim token directly - self.sole_master = true; - self.state = MasterState::UseToken; - self.frame_count = 0; - self.token_count = 0; - return MstpFrame { - frame_type: FrameType::Token, - destination: self.config.this_station, - source: self.config.this_station, - data: Bytes::new(), - }; - } - // Have a known successor — pass token to them - return self.pass_token(); - } - if self.poll_station == self.next_station { - // Reached our known successor — done scanning the gap - return self.pass_token(); - } + let ts = self.config.this_station; + let max_master = self.config.max_master; + self.poll_count = 0; + + if self.sole_master { + // SoleMaster: resume USE_TOKEN without emitting Token TS→TS + self.frame_count = 0; + self.state = MasterState::UseToken; + return self.use_token(); } - // Poll the next station - self.state = MasterState::PollForMaster; - MstpFrame { - frame_type: FrameType::PollForMaster, - destination: self.poll_station, - source: self.config.this_station, - data: Bytes::new(), + + if self.next_station != ts { + // DoneWithPFM — maintenance timed out; pass token back to known NS + return self.pass_token(); + } + + // Searching for a successor (NS == TS) + let next_ps = next_addr(self.poll_station, max_master); + if next_ps != ts { + // SendNextPFM + self.poll_station = next_ps; + self.retry_token_count = 0; + self.state = MasterState::PollForMaster; + MstpFrame { + frame_type: FrameType::PollForMaster, + destination: self.poll_station, + source: ts, + data: Bytes::new(), + } + } else { + // DeclareSoleMaster — no Token with source==destination + self.sole_master = true; + self.frame_count = 0; + self.state = MasterState::UseToken; + self.use_token() } } @@ -533,6 +606,8 @@ fn next_addr(current: u8, max_master: u8) -> u8 { mod port; pub use port::{LoopbackSerial, MstpTransport, NoSerial}; +#[cfg(test)] +mod clause956_tests; #[cfg(test)] mod port_timing_tests; #[cfg(test)] diff --git a/crates/bacnet-transport/src/mstp/port.rs b/crates/bacnet-transport/src/mstp/port.rs index f175409f..c4430cd6 100644 --- a/crates/bacnet-transport/src/mstp/port.rs +++ b/crates/bacnet-transport/src/mstp/port.rs @@ -217,20 +217,15 @@ impl TransportPort for MstpTransport { while node_guard.state == MasterState::UseToken || node_guard.state == MasterState::DoneWithToken { - // DoneWithToken: max_info_frames reached, pass token immediately - if node_guard.state == MasterState::DoneWithToken { - let token = node_guard.pass_token(); - encode_buf.clear(); - if let Err(e) = encode_frame(&mut encode_buf, &token) { - warn!("MS/TP encode error: {}", e); + let frame_to_send = + if node_guard.state == MasterState::DoneWithToken { + node_guard.done_with_token() } else { - pending_writes.push(encode_buf.to_vec()); - } - break; - } - let frame_to_send = node_guard.use_token(); + node_guard.use_token() + }; encode_buf.clear(); - if let Err(e) = encode_frame(&mut encode_buf, &frame_to_send) { + if let Err(e) = encode_frame(&mut encode_buf, &frame_to_send) + { warn!("MS/TP encode error: {}", e); break; } @@ -242,8 +237,20 @@ impl TransportPort for MstpTransport { node_guard.state = MasterState::WaitForReply; break; } - // After sending Token, we're done - if frame_to_send.frame_type == FrameType::Token { + // After Token or PFM, leave the use/done loop + if matches!( + frame_to_send.frame_type, + FrameType::Token | FrameType::PollForMaster + ) { + break; + } + // SoleMaster may return to UseToken with no wire + // progress; break if still UseToken after empty cycle + if node_guard.state == MasterState::UseToken + && node_guard.tx_queue.is_empty() + && node_guard.frame_count + >= node_guard.config.max_info_frames + { break; } } @@ -336,12 +343,21 @@ impl TransportPort for MstpTransport { } } MasterState::WaitForReply => { - // ReplyTimeout: enter DoneWithToken. + // ReplyTimeout: enter DoneWithToken then run transitions. node_guard.expected_reply_source = None; node_guard.frame_count = node_guard.config.max_info_frames; node_guard.state = MasterState::DoneWithToken; - // Fall through to DoneWithToken handling on next iteration - T_USAGE_TIMEOUT_MS + let frame_to_send = node_guard.done_with_token(); + encode_buf.clear(); + if let Ok(()) = encode_frame(&mut encode_buf, &frame_to_send) { + pending_writes.push(encode_buf.to_vec()); + } + match node_guard.state { + MasterState::PassToken => T_USAGE_TIMEOUT_MS, + MasterState::PollForMaster => node_guard.t_slot_ms, + MasterState::UseToken => T_USAGE_TIMEOUT_MS, + _ => T_USAGE_TIMEOUT_MS, + } } MasterState::AnswerDataRequest => { // The timer fires early enough to include @@ -381,14 +397,20 @@ impl TransportPort for MstpTransport { } MasterState::UseToken | MasterState::DoneWithToken => { - // Should not typically timeout in UseToken/DoneWithToken; - // pass the token and treat as idle - let token = node_guard.pass_token(); + // Should not typically timeout here; run DONE_WITH_TOKEN + // transitions (never unconditional pass_token). + node_guard.state = MasterState::DoneWithToken; + let frame_to_send = node_guard.done_with_token(); encode_buf.clear(); - if let Ok(()) = encode_frame(&mut encode_buf, &token) { + if let Ok(()) = encode_frame(&mut encode_buf, &frame_to_send) { pending_writes.push(encode_buf.to_vec()); } - T_USAGE_TIMEOUT_MS + match node_guard.state { + MasterState::PassToken => T_USAGE_TIMEOUT_MS, + MasterState::PollForMaster => node_guard.t_slot_ms, + MasterState::UseToken => T_USAGE_TIMEOUT_MS, + _ => T_USAGE_TIMEOUT_MS, + } } }; if was_answering_data_request { diff --git a/crates/bacnet-transport/src/mstp/tests.rs b/crates/bacnet-transport/src/mstp/tests.rs index df584ca1..35e63069 100644 --- a/crates/bacnet-transport/src/mstp/tests.rs +++ b/crates/bacnet-transport/src/mstp/tests.rs @@ -241,6 +241,9 @@ fn use_token_sends_queued_data() { }; let mut node = MasterNode::new(config).unwrap(); node.state = MasterState::UseToken; + node.next_station = 1; // known successor so DoneWithToken can SendToken + node.poll_station = 0; + node.token_count = 0; node.queue_npdu(5, Bytes::from_static(&[0x01, 0x00, 0x30])) .unwrap(); node.queue_npdu(BROADCAST_MAC, Bytes::from_static(&[0x01, 0x20])) @@ -257,11 +260,11 @@ fn use_token_sends_queued_data() { assert_eq!(frame.frame_type, FrameType::BACnetDataNotExpectingReply); assert_eq!(frame.destination, BROADCAST_MAC); - // Third call: no more data, pass token + // Third call: no more data, pass token to known NS let frame = node.use_token(); assert_eq!(frame.frame_type, FrameType::Token); + assert_eq!(frame.destination, 1); } - #[test] fn use_token_respects_max_info_frames() { let config = MstpConfig { @@ -272,6 +275,9 @@ fn use_token_respects_max_info_frames() { }; let mut node = MasterNode::new(config).unwrap(); node.state = MasterState::UseToken; + node.next_station = 1; + node.poll_station = 0; + node.token_count = 0; node.queue_npdu(5, Bytes::from_static(&[0x01])).unwrap(); node.queue_npdu(6, Bytes::from_static(&[0x02])).unwrap(); @@ -282,16 +288,17 @@ fn use_token_respects_max_info_frames() { || frame.frame_type == FrameType::BACnetDataNotExpectingReply ); - // Second call: frame_count >= max_info_frames, passes token + // Second call: frame_count >= max_info_frames → DoneWithToken → Token to NS let frame = node.use_token(); assert_eq!(frame.frame_type, FrameType::Token); + assert_eq!(frame.destination, 1); // Data should still be in queue assert_eq!(node.tx_queue.len(), 1); } - #[test] fn poll_for_master_after_npoll_tokens() { + // NS unknown (NS==TS): DONE_WITH_TOKEN → NextStationUnknown → PFM to TS+1 let config = MstpConfig { this_station: 0, max_master: 127, @@ -299,13 +306,15 @@ fn poll_for_master_after_npoll_tokens() { baud_rate: 9600, }; let mut node = MasterNode::new(config).unwrap(); - node.state = MasterState::UseToken; - node.token_count = NPOLL; // Trigger poll + node.state = MasterState::DoneWithToken; + node.token_count = NPOLL; + node.frame_count = node.config.max_info_frames; - let frame = node.use_token(); + let frame = node.done_with_token(); assert_eq!(frame.frame_type, FrameType::PollForMaster); + assert_eq!(frame.destination, 1); assert_eq!(node.state, MasterState::PollForMaster); - assert_eq!(node.token_count, 0); + assert_eq!(node.poll_station, 1); } #[test] @@ -348,16 +357,13 @@ fn poll_timeout_advances_poll_station() { }; let mut node = MasterNode::new(config).unwrap(); node.state = MasterState::PollForMaster; - // Start polling from station 1 + node.next_station = 0; // unknown successor — keep scanning node.poll_station = 1; - // MAX_POLL_RETRIES timeouts for station 1 - for _ in 0..MAX_POLL_RETRIES { - let frame = node.poll_timeout(); - assert_eq!(frame.frame_type, FrameType::PollForMaster); - } - // Should have moved to station 2 + let frame = node.poll_timeout(); + assert_eq!(frame.frame_type, FrameType::PollForMaster); assert_eq!(node.poll_station, 2); + assert_eq!(frame.destination, 2); } #[test] @@ -370,15 +376,16 @@ fn poll_timeout_sole_master() { }; let mut node = MasterNode::new(config).unwrap(); node.state = MasterState::PollForMaster; + node.next_station = 0; node.poll_station = 1; - // Timeout for station 1, MAX_POLL_RETRIES times - for _ in 0..MAX_POLL_RETRIES { - node.poll_timeout(); - } - // poll_station wraps to 0 (== this_station), sole master declared - assert_eq!(node.state, MasterState::UseToken); + // Timeout for station 1 → next_ps wraps to TS → DeclareSoleMaster (no Token 0→0) + let frame = node.poll_timeout(); assert!(node.sole_master); + assert_eq!(node.state, MasterState::PollForMaster); + // Sole master restart / maintenance emits PFM, never Token TS→TS + assert_eq!(frame.frame_type, FrameType::PollForMaster); + assert_ne!(frame.destination, frame.source); } #[test] @@ -546,12 +553,8 @@ async fn transport_rejects_bad_mac() { #[test] fn test_no_token_timeout_claims_token() { - // Simulate the NoToken -> sole master flow without the transport loop. - // - // Flow: - // Idle timeout -> enter NoToken, send 1st PFM, retry_token_count=0 - // NoToken timeout #1 -> retry_token_count(0) < N_RETRY_TOKEN(1), send 2nd PFM, count=1 - // NoToken timeout #2 -> retry_token_count(1) >= N_RETRY_TOKEN(1), claim sole master + // Sole master must not emit Token TS→TS. After DeclareSoleMaster, + // DONE_WITH_TOKEN reuses / restarts maintenance via PFM. let config = MstpConfig { this_station: 5, max_master: 127, @@ -559,32 +562,23 @@ fn test_no_token_timeout_claims_token() { baud_rate: 9600, }; let mut node = MasterNode::new(config).unwrap(); - - // Simulate: Idle -> NoToken (first timeout sends 1st PFM) - node.state = MasterState::NoToken; - node.retry_token_count = 0; - - // First retry (retry_token_count=0 < N_RETRY_TOKEN=1) - assert!(node.retry_token_count < N_RETRY_TOKEN); - node.retry_token_count += 1; - assert_eq!(node.retry_token_count, 1); - - // After N_RETRY_TOKEN retries, declare sole master - assert!(node.retry_token_count >= N_RETRY_TOKEN); node.sole_master = true; - node.next_station = node.config.this_station; - node.state = MasterState::UseToken; - node.frame_count = 0; + node.next_station = 5; + node.poll_station = 5; node.token_count = 0; - - assert!(node.sole_master); - assert_eq!(node.next_station, 5); - assert_eq!(node.state, MasterState::UseToken); - - // Use token should pass to self (sole master) - let frame = node.use_token(); - assert_eq!(frame.frame_type, FrameType::Token); - assert_eq!(frame.destination, 5); // pass to self + node.state = MasterState::DoneWithToken; + node.frame_count = node.config.max_info_frames; + + let frame = node.done_with_token(); + assert_ne!( + (frame.frame_type, frame.destination), + (FrameType::Token, 5), + "forbidden self-token" + ); + assert!( + frame.frame_type == FrameType::PollForMaster || node.state == MasterState::UseToken, + "sole master continues without Token TS→TS" + ); } #[test] @@ -610,12 +604,17 @@ fn test_wait_for_reply_state_after_data_expecting_reply() { node.state = MasterState::WaitForReply; assert_eq!(node.state, MasterState::WaitForReply); - // On timeout in WaitForReply, we pass the token - let token = node.pass_token(); + // On timeout in WaitForReply, DONE_WITH_TOKEN (not unconditional pass_token) + node.next_station = 5; + node.poll_station = 1; + node.token_count = 0; + node.frame_count = node.config.max_info_frames; + node.state = MasterState::DoneWithToken; + let token = node.done_with_token(); assert_eq!(token.frame_type, FrameType::Token); + assert_eq!(token.destination, 5); assert_eq!(node.state, MasterState::PassToken); } - #[test] fn test_answer_data_request_reply_channel() { let (tx, mut rx) = mpsc::channel(16); @@ -657,8 +656,8 @@ fn test_answer_data_request_reply_channel() { #[test] fn test_poll_for_master_scan_range() { - // Station 0, next_station=5, max_master=10 - // Should poll starting at 6 (next_addr(5, 10)), scanning 6..=10, 0 would be us so stop + // TS=0, NS=5, Max_Master=10 — maintenance candidates are 1..=4 only + // (advance PS from TS toward NS; never begin at NS+1). let config = MstpConfig { this_station: 0, max_master: 10, @@ -667,53 +666,25 @@ fn test_poll_for_master_scan_range() { }; let mut node = MasterNode::new(config).unwrap(); node.next_station = 5; - node.state = MasterState::UseToken; - node.token_count = NPOLL; + node.poll_station = 0; + node.state = MasterState::DoneWithToken; + node.token_count = NPOLL.saturating_sub(1); + node.frame_count = node.config.max_info_frames; - // use_token triggers PollForMaster at poll_station = next_addr(5, 10) = 6 - let frame = node.use_token(); + let frame = node.done_with_token(); assert_eq!(frame.frame_type, FrameType::PollForMaster); - assert_eq!(node.poll_station, 6); - assert_eq!(frame.destination, 6); - - // Each station takes MAX_POLL_RETRIES timeouts to exhaust, then advances. - // Station 6: 3 retries -> advance to 7 - for _ in 0..MAX_POLL_RETRIES { - node.poll_timeout(); - } - assert_eq!(node.poll_station, 7); - - // Station 7: 3 retries -> advance to 8 - for _ in 0..MAX_POLL_RETRIES { - node.poll_timeout(); - } - assert_eq!(node.poll_station, 8); - - // Station 8: 3 retries -> advance to 9 - for _ in 0..MAX_POLL_RETRIES { - node.poll_timeout(); - } - assert_eq!(node.poll_station, 9); - - // Station 9: 3 retries -> advance to 10 - for _ in 0..MAX_POLL_RETRIES { - node.poll_timeout(); - } - assert_eq!(node.poll_station, 10); + assert_eq!(node.poll_station, 1); + assert_eq!(frame.destination, 1); - // Station 10: 3 retries -> advance to next_addr(10, 10) = 0 == this_station - // poll_timeout detects this_station match — since next_station=5 (not TS), - // we have a known successor, pass token to them. - for _ in 0..MAX_POLL_RETRIES { - node.poll_timeout(); - } - assert_eq!(node.state, MasterState::PassToken); + // One timeout → Token to known NS (not whole-space scan in one token use) + let token = node.poll_timeout(); + assert_eq!(token.frame_type, FrameType::Token); + assert_eq!(token.destination, 5); } #[test] fn test_poll_for_master_scan_range_adjacent() { - // When next_station is adjacent (this_station=0, next_station=1, max_master=1), - // poll_station = next_addr(1, 1) = 0 == this_station, so no gap to scan + // NS == TS+1: ResetMaintenancePFM / SendToken — no gap to poll let config = MstpConfig { this_station: 0, max_master: 1, @@ -722,13 +693,15 @@ fn test_poll_for_master_scan_range_adjacent() { }; let mut node = MasterNode::new(config).unwrap(); node.next_station = 1; - node.state = MasterState::UseToken; - node.token_count = NPOLL; + node.poll_station = 0; + node.state = MasterState::DoneWithToken; + node.token_count = NPOLL.saturating_sub(1); + node.frame_count = node.config.max_info_frames; - // use_token should just pass token since no gap - let frame = node.use_token(); + let frame = node.done_with_token(); assert_eq!(frame.frame_type, FrameType::Token); - assert_eq!(node.state, MasterState::PassToken); + assert_eq!(frame.destination, 1); + assert_eq!(node.poll_station, 0); } #[test] From 19d205d78c947aea3fe98110d8a6c392359aa627 Mon Sep 17 00:00:00 2001 From: bartlino Date: Sun, 30 Aug 2026 17:32:55 +0000 Subject: [PATCH 5/7] style(mstp): rustfmt clause956 regression tests Co-authored-by: Cursor --- .../src/mstp/clause956_tests.rs | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/crates/bacnet-transport/src/mstp/clause956_tests.rs b/crates/bacnet-transport/src/mstp/clause956_tests.rs index fbe4b0e5..e8057eb4 100644 --- a/crates/bacnet-transport/src/mstp/clause956_tests.rs +++ b/crates/bacnet-transport/src/mstp/clause956_tests.rs @@ -121,7 +121,10 @@ fn b_three_master_ring_2000_rotations() { } assert_eq!(forbidden_self_token, 0, "no Token source==destination"); - assert!(receipts.iter().all(|&n| n > 0), "every master must receive token: {receipts:?}"); + assert!( + receipts.iter().all(|&n| n > 0), + "every master must receive token: {receipts:?}" + ); // Rough fairness: each master should see a large share of 2000 holds for (i, &n) in receipts.iter().enumerate() { assert!( @@ -197,7 +200,9 @@ fn d_new_master_in_gap() { source: 5, data: Bytes::new(), }; - let out = node.handle_received_frame(&reply, &tx).expect("Token to new NS"); + let out = node + .handle_received_frame(&reply, &tx) + .expect("Token to new NS"); assert_eq!(node.next_station, 5); assert_eq!(node.poll_station, 3); assert_eq!(node.token_count, 0); @@ -234,8 +239,10 @@ fn e_failed_successor_uses_pfm_not_blind_tokens() { assert_ne!(find.frame_type, FrameType::Token); // Must never blindly Token to unverified addresses after failure - assert!(node.pass_token_timeout().is_none() || { - // still in PollForMaster — timeout handled by poll_timeout, not more Tokens - true - }); + assert!( + node.pass_token_timeout().is_none() || { + // still in PollForMaster — timeout handled by poll_timeout, not more Tokens + true + } + ); } From 867649b54bc57aa535d9b8c1fdf650c20528296a Mon Sep 17 00:00:00 2001 From: Justin Scott Date: Sun, 30 Aug 2026 16:37:16 -0400 Subject: [PATCH 6/7] fix(mstp): enforce standard frame safety boundaries --- .../src/mstp/clause956_tests.rs | 55 +++++- crates/bacnet-transport/src/mstp/mod.rs | 59 +++--- crates/bacnet-transport/src/mstp/port.rs | 170 +++++++++++++----- crates/bacnet-transport/src/mstp/tests.rs | 4 +- crates/bacnet-transport/src/mstp_frame.rs | 65 +++++-- .../bacnet-transport/src/mstp_frame_tests.rs | 140 ++++++++++++++- 6 files changed, 399 insertions(+), 94 deletions(-) diff --git a/crates/bacnet-transport/src/mstp/clause956_tests.rs b/crates/bacnet-transport/src/mstp/clause956_tests.rs index e8057eb4..4b226b4f 100644 --- a/crates/bacnet-transport/src/mstp/clause956_tests.rs +++ b/crates/bacnet-transport/src/mstp/clause956_tests.rs @@ -238,11 +238,52 @@ fn e_failed_successor_uses_pfm_not_blind_tokens() { assert_eq!(node.next_station, 3); // NS = TS assert_ne!(find.frame_type, FrameType::Token); - // Must never blindly Token to unverified addresses after failure - assert!( - node.pass_token_timeout().is_none() || { - // still in PollForMaster — timeout handled by poll_timeout, not more Tokens - true - } - ); + // The node is now in PFM, and a PFM timeout advances the search rather + // than invoking another token retry against an unverified address. + assert_eq!(node.state, MasterState::PollForMaster); + assert_eq!(node.poll_station, 0); + let next_poll = node.poll_timeout(); + assert_eq!(next_poll.frame_type, FrameType::PollForMaster); + assert_eq!(next_poll.source, 3); + assert_eq!(next_poll.destination, 1); + assert_eq!(node.state, MasterState::PollForMaster); +} + +#[test] +fn master_node_rejects_max_master_above_standard_limit() { + let error = MasterNode::new(cfg(3, MAX_MASTER + 1)) + .err() + .expect("invalid max_master"); + assert!(error + .to_string() + .contains("max_master 128 exceeds MAX_MASTER (127)")); +} + +#[test] +fn master_node_rejects_station_above_configured_max_master() { + let error = MasterNode::new(cfg(5, 4)) + .err() + .expect("invalid this_station"); + assert!(error + .to_string() + .contains("this_station 5 exceeds configured max_master (4)")); +} + +#[test] +fn pass_token_self_destination_enters_pfm_at_runtime() { + let mut node = MasterNode::new(cfg(3, 7)).unwrap(); + node.next_station = node.config.this_station; + + let frame = node.pass_token(); + assert_eq!(frame.frame_type, FrameType::PollForMaster); + assert_eq!(frame.source, 3); + assert_eq!(frame.destination, 4); + assert_eq!(node.state, MasterState::PollForMaster); + + node.state = MasterState::PassToken; + node.next_station = node.config.this_station; + let retry = node.pass_token_timeout().expect("PFM recovery frame"); + assert_eq!(retry.frame_type, FrameType::PollForMaster); + assert_eq!(retry.destination, 4); + assert_ne!(retry.frame_type, FrameType::Token); } diff --git a/crates/bacnet-transport/src/mstp/mod.rs b/crates/bacnet-transport/src/mstp/mod.rs index 745ec163..a048cf8d 100644 --- a/crates/bacnet-transport/src/mstp/mod.rs +++ b/crates/bacnet-transport/src/mstp/mod.rs @@ -12,7 +12,9 @@ use bytes::Bytes; use tokio::sync::{mpsc, oneshot}; use tracing::debug; -use crate::mstp_frame::{FrameType, MstpFrame, BROADCAST_MAC, MAX_MASTER}; +use crate::mstp_frame::{ + FrameType, MstpFrame, BROADCAST_MAC, MAX_MASTER, MAX_STANDARD_FRAME_LENGTH, +}; use crate::port::ReceivedNpdu; // --------------------------------------------------------------------------- @@ -59,8 +61,8 @@ fn calculate_t_turnaround_us(baud_rate: u32) -> u64 { } /// Number of retries for token pass before declaring token lost. const N_RETRY_TOKEN: u8 = 1; -/// Maximum frame buffer size: preamble(2) + header(6) + max data(1497) + CRC16(2) -pub(crate) const MSTP_MAX_FRAME_BUF: usize = 1507; +/// Maximum standard frame buffer size: preamble + header + data + data CRC. +pub(crate) const MSTP_MAX_FRAME_BUF: usize = MAX_STANDARD_FRAME_LENGTH; /// Host-side stale partial-frame timeout for USB/chunked serial reassembly. /// /// This is **not** Clause 9 `T_frame_abort` (wire inter-byte silence). Host async reads @@ -161,10 +163,16 @@ const NPOLL: u8 = 50; impl MasterNode { pub fn new(config: MstpConfig) -> Result { - if config.this_station > MAX_MASTER { + if config.max_master > MAX_MASTER { return Err(Error::Encoding(format!( - "MS/TP this_station {} exceeds MAX_MASTER ({})", - config.this_station, MAX_MASTER + "MS/TP max_master {} exceeds MAX_MASTER ({})", + config.max_master, MAX_MASTER + ))); + } + if config.this_station > config.max_master { + return Err(Error::Encoding(format!( + "MS/TP this_station {} exceeds configured max_master ({})", + config.this_station, config.max_master ))); } let ts = config.this_station; @@ -410,17 +418,9 @@ impl MasterNode { return self.use_token(); } - // NextStationUnknown (Addendum 135-2008v-1): NS == TS and not sole master + // Clause 9.5.6 NextStationUnknown: NS == TS and not sole master if !self.sole_master && self.next_station == ts { - self.poll_station = next_ts; - self.retry_token_count = 0; - self.state = MasterState::PollForMaster; - return MstpFrame { - frame_type: FrameType::PollForMaster, - destination: self.poll_station, - source: ts, - data: Bytes::new(), - }; + return self.start_unknown_successor_poll(); } // SendToken while TokenCount < Npoll - 1 @@ -476,14 +476,13 @@ impl MasterNode { /// Generate a token-pass frame to next_station. /// - /// Must not be used when `next_station == this_station` (Clause 9.5.6.5 - /// requires PFM to TS+1 instead — handled by [`Self::done_with_token`]). + /// When the successor is unknown (`next_station == this_station`), this + /// enters the Clause 9.5.6 PFM flow rather than emitting Token TS→TS. pub fn pass_token(&mut self) -> MstpFrame { let ts = self.config.this_station; - debug_assert_ne!( - self.next_station, ts, - "pass_token must not emit Token TS→TS; use done_with_token / PFM" - ); + if self.next_station == ts { + return self.start_unknown_successor_poll(); + } self.state = MasterState::PassToken; self.retry_token_count = 0; self.event_count = 0; @@ -502,6 +501,9 @@ impl MasterNode { pub fn pass_token_timeout(&mut self) -> Option { let ts = self.config.this_station; let max_master = self.config.max_master; + if self.next_station == ts { + return Some(self.start_unknown_successor_poll()); + } if self.retry_token_count < N_RETRY_TOKEN { // RetrySendToken: resend Token to NS exactly Nretry_token times self.retry_token_count += 1; @@ -592,6 +594,19 @@ impl MasterNode { self.tx_queue.push_back((dest, npdu)); Ok(()) } + + fn start_unknown_successor_poll(&mut self) -> MstpFrame { + let ts = self.config.this_station; + self.poll_station = next_addr(ts, self.config.max_master); + self.retry_token_count = 0; + self.state = MasterState::PollForMaster; + MstpFrame { + frame_type: FrameType::PollForMaster, + destination: self.poll_station, + source: ts, + data: Bytes::new(), + } + } } /// Advance to the next station address, wrapping at max_master. diff --git a/crates/bacnet-transport/src/mstp/port.rs b/crates/bacnet-transport/src/mstp/port.rs index c4430cd6..aebca52b 100644 --- a/crates/bacnet-transport/src/mstp/port.rs +++ b/crates/bacnet-transport/src/mstp/port.rs @@ -17,6 +17,60 @@ use super::{ T_REPLY_TIMEOUT_MS, T_REPLY_TRANSMIT_MARGIN_MS, T_USAGE_TIMEOUT_MS, }; +/// Add one host read to the persistent receive buffer and drain all complete frames. +/// +/// The persistent buffer never exceeds one maximum standard frame. A host read may +/// contain any number of coalesced frames; complete frames are drained before more +/// bytes from that same read are appended. +fn assemble_host_chunk(frame_buf: &mut Vec, chunk: &[u8]) -> Vec { + let mut frames = Vec::new(); + let mut remaining = chunk; + + while !remaining.is_empty() { + let available = MSTP_MAX_FRAME_BUF.saturating_sub(frame_buf.len()); + if available == 0 { + // A valid standard frame is decidable at this size. This fallback + // guarantees progress if malformed input somehow remains NeedMore. + warn!("MS/TP: full incomplete host assembly, discarding one byte"); + frame_buf.drain(..1); + continue; + } + + let take = available.min(remaining.len()); + frame_buf.extend_from_slice(&remaining[..take]); + remaining = &remaining[take..]; + + loop { + let preamble_pos = match find_preamble(frame_buf) { + Some(pos) => pos, + None => { + retain_lone_preamble_byte(frame_buf); + break; + } + }; + + if preamble_pos > 0 { + frame_buf.drain(..preamble_pos); + } + + match decode_frame_stream(frame_buf) { + StreamDecode::Complete { frame, consumed } => { + frame_buf.drain(..consumed); + frames.push(frame); + } + StreamDecode::NeedMore => break, + StreamDecode::Invalid { discard } => { + let discard = discard.min(frame_buf.len()).max(1); + frame_buf.drain(..discard); + } + } + } + } + + debug_assert!(frame_buf.len() <= MSTP_MAX_FRAME_BUF); + frames +} + // --------------------------------------------------------------------------- // MS/TP Transport // --------------------------------------------------------------------------- @@ -75,7 +129,7 @@ impl TransportPort for MstpTransport { // Receive loop using tokio::select! with timer let task = tokio::spawn(async move { let mut recv_buf = vec![0u8; 2048]; - let mut frame_buf = Vec::with_capacity(2048); + let mut frame_buf = Vec::with_capacity(MSTP_MAX_FRAME_BUF); let mut last_byte_time = tokio::time::Instant::now(); // Start with T_NO_TOKEN timeout — if we don't see anything, claim the token @@ -123,7 +177,7 @@ impl TransportPort for MstpTransport { } // Branch 1: serial data arrives result = serial_clone.read(&mut recv_buf) => { - match result { + let frames = match result { Ok(0) => continue, Ok(n) => { // Host stale-partial timeout: drop abandoned assembly if no @@ -144,44 +198,15 @@ impl TransportPort for MstpTransport { } } last_byte_time = now; - - // Prevent unbounded growth from malformed input: - // check BEFORE extending to avoid a large allocation. - if frame_buf.len() + n > MSTP_MAX_FRAME_BUF { - warn!( - "MS/TP: frame buffer would overflow ({} + {} bytes), resetting", - frame_buf.len(), n - ); - frame_buf.clear(); - continue; - } - frame_buf.extend_from_slice(&recv_buf[..n]); + assemble_host_chunk(&mut frame_buf, &recv_buf[..n]) } Err(e) => { warn!("MS/TP serial read error: {}", e); break; } - } - - // Try to find and decode frames - loop { - let preamble_pos = match find_preamble(&frame_buf) { - Some(pos) => pos, - None => { - retain_lone_preamble_byte(&mut frame_buf); - break; - } - }; - - // Discard bytes before preamble - if preamble_pos > 0 { - frame_buf.drain(..preamble_pos); - } - - match decode_frame_stream(&frame_buf) { - StreamDecode::Complete { frame, consumed } => { - frame_buf.drain(..consumed); + }; + for frame in frames { // Process through state machine — collect // frames under lock, drop before writing. let mut node_guard = node.lock().await; @@ -288,13 +313,6 @@ impl TransportPort for MstpTransport { + tokio::time::Duration::from_millis(timeout_ms) }), ); - } - StreamDecode::NeedMore => break, - StreamDecode::Invalid { discard } => { - let discard = discard.min(frame_buf.len()).max(1); - frame_buf.drain(..discard); - } - } } } // Branch 2: timeout @@ -593,3 +611,73 @@ impl SerialPort for NoSerial { Err(Error::Encoding("NoSerial: MS/TP not available".into())) } } + +#[cfg(test)] +mod assembly_tests { + use super::*; + use crate::mstp_frame::{MAX_STANDARD_MPDU_DATA, PREAMBLE}; + + fn encode_host_data_frame(source: u8, fill: u8, data_len: usize) -> Vec { + let frame = MstpFrame { + frame_type: FrameType::BACnetDataNotExpectingReply, + destination: 3, + source, + data: Bytes::from(vec![fill; data_len]), + }; + let mut wire = BytesMut::new(); + encode_frame(&mut wire, &frame).unwrap(); + wire.to_vec() + } + + #[test] + fn drains_coalesced_frames_larger_than_one_frame() { + let first = encode_host_data_frame(1, 0xA1, 300); + let second = encode_host_data_frame(2, 0xB2, 300); + let mut chunk = first; + chunk.extend_from_slice(&second); + assert!(chunk.len() > MSTP_MAX_FRAME_BUF); + + let mut frame_buf = Vec::new(); + let frames = assemble_host_chunk(&mut frame_buf, &chunk); + + assert_eq!(frames.len(), 2); + assert_eq!(frames[0].source, 1); + assert_eq!(frames[0].data, Bytes::from(vec![0xA1; 300])); + assert_eq!(frames[1].source, 2); + assert_eq!(frames[1].data, Bytes::from(vec![0xB2; 300])); + assert!(frame_buf.is_empty()); + } + + #[test] + fn bounds_malformed_input_and_retains_max_partial() { + let malformed = vec![0xAA; MSTP_MAX_FRAME_BUF * 4 + 17]; + let mut frame_buf = Vec::new(); + assert!(assemble_host_chunk(&mut frame_buf, &malformed).is_empty()); + assert!(frame_buf.len() <= MSTP_MAX_FRAME_BUF); + + let wire = encode_host_data_frame(1, 0xCC, MAX_STANDARD_MPDU_DATA); + assert_eq!(wire.len(), MSTP_MAX_FRAME_BUF); + let split = wire.len() - 1; + assert!(assemble_host_chunk(&mut frame_buf, &wire[..split]).is_empty()); + assert_eq!(frame_buf.len(), split); + + let frames = assemble_host_chunk(&mut frame_buf, &wire[split..]); + assert_eq!(frames.len(), 1); + assert_eq!(frames[0].data.len(), MAX_STANDARD_MPDU_DATA); + assert!(frame_buf.is_empty()); + + let token = MstpFrame { + frame_type: FrameType::Token, + destination: 3, + source: 1, + data: Bytes::new(), + }; + let mut token_wire = BytesMut::new(); + encode_frame(&mut token_wire, &token).unwrap(); + assert!(assemble_host_chunk(&mut frame_buf, &[0xAA, PREAMBLE[0]]).is_empty()); + assert_eq!(frame_buf, PREAMBLE[..1]); + let frames = assemble_host_chunk(&mut frame_buf, &token_wire[1..]); + assert_eq!(frames, vec![token]); + assert!(frame_buf.is_empty()); + } +} diff --git a/crates/bacnet-transport/src/mstp/tests.rs b/crates/bacnet-transport/src/mstp/tests.rs index 35e63069..9f3a62be 100644 --- a/crates/bacnet-transport/src/mstp/tests.rs +++ b/crates/bacnet-transport/src/mstp/tests.rs @@ -706,8 +706,8 @@ fn test_poll_for_master_scan_range_adjacent() { #[test] fn mstp_frame_buf_max_size() { - // The maximum valid MS/TP frame is: 2 (preamble) + 6 (header) + 1497 (data) + 2 (CRC16) = 1507 - assert_eq!(MSTP_MAX_FRAME_BUF, 1507); + // Standard frame: 2 (preamble) + 6 (header) + 501 (data) + 2 (CRC16). + assert_eq!(MSTP_MAX_FRAME_BUF, 511); } #[test] diff --git a/crates/bacnet-transport/src/mstp_frame.rs b/crates/bacnet-transport/src/mstp_frame.rs index 78f7b0ab..bbf7328c 100644 --- a/crates/bacnet-transport/src/mstp_frame.rs +++ b/crates/bacnet-transport/src/mstp_frame.rs @@ -18,14 +18,21 @@ pub const PREAMBLE: [u8; 2] = [0x55, 0xFF]; /// Header length after preamble: frame_type(1) + dest(1) + src(1) + length(2) + header_crc(1). pub const HEADER_LENGTH: usize = 6; -/// Maximum NPDU data length per MS/TP extended frame. -/// Standard frames are limited to MAX_STANDARD_MPDU_DATA (501 bytes). -pub const MAX_MPDU_DATA: usize = 1497; - /// Maximum NPDU data length per standard MS/TP frame. -/// Legacy devices only support this smaller limit. pub const MAX_STANDARD_MPDU_DATA: usize = 501; +/// Maximum NPDU data length supported by this non-encoded MS/TP codec. +/// +/// This remains as the public compatibility name for +/// [`MAX_STANDARD_MPDU_DATA`]. COBS-encoded frame types 32..=127 are not supported. +pub const MAX_MPDU_DATA: usize = MAX_STANDARD_MPDU_DATA; + +const DATA_CRC_LENGTH: usize = 2; + +/// Maximum wire length supported by this standard-frame codec. +pub(crate) const MAX_STANDARD_FRAME_LENGTH: usize = + PREAMBLE.len() + HEADER_LENGTH + MAX_STANDARD_MPDU_DATA + DATA_CRC_LENGTH; + /// Broadcast MAC address. pub const BROADCAST_MAC: u8 = 0xFF; @@ -97,6 +104,10 @@ impl FrameType { } } +fn is_unsupported_cobs_frame_type(raw: u8) -> bool { + (0x20..=0x7F).contains(&raw) +} + /// A decoded MS/TP frame. #[derive(Debug, Clone, PartialEq, Eq)] pub struct MstpFrame { @@ -234,15 +245,22 @@ pub fn encode_frame( frame: &MstpFrame, ) -> Result<(), bacnet_types::error::Error> { let data_len = frame.data.len(); - if data_len > MAX_MPDU_DATA { + let frame_type = frame.frame_type.to_raw(); + if is_unsupported_cobs_frame_type(frame_type) { + return Err(bacnet_types::error::Error::Encoding(format!( + "MS/TP COBS-encoded frame type {frame_type} is unsupported" + ))); + } + if data_len > MAX_STANDARD_MPDU_DATA { return Err(bacnet_types::error::Error::Encoding(format!( "MS/TP data length {} exceeds maximum {}", - data_len, MAX_MPDU_DATA + data_len, MAX_STANDARD_MPDU_DATA ))); } // Reserve space - let total = 2 + HEADER_LENGTH + data_len + if data_len > 0 { 2 } else { 0 }; + let total = + PREAMBLE.len() + HEADER_LENGTH + data_len + if data_len > 0 { DATA_CRC_LENGTH } else { 0 }; buf.reserve(total); // Preamble @@ -250,7 +268,7 @@ pub fn encode_frame( // Header: frame_type, dest, src, length(2) let header = [ - frame.frame_type.to_raw(), + frame_type, frame.destination, frame.source, (data_len >> 8) as u8, @@ -311,6 +329,10 @@ pub fn decode_frame_stream(data: &[u8]) -> StreamDecode { return StreamDecode::Invalid { discard: 1 }; } + if is_unsupported_cobs_frame_type(data[2]) { + return StreamDecode::Invalid { discard: 1 }; + } + let frame_type = FrameType::from_raw(data[2]); let destination = data[3]; let source = data[4]; @@ -323,24 +345,24 @@ pub fn decode_frame_stream(data: &[u8]) -> StreamDecode { } let data_length = ((data[5] as usize) << 8) | (data[6] as usize); - if data_length > MAX_MPDU_DATA { + if data_length > MAX_STANDARD_MPDU_DATA { return StreamDecode::Invalid { discard: 1 }; } let mut consumed = 2 + HEADER_LENGTH; if data_length > 0 { - let needed = consumed + data_length + 2; + let needed = consumed + data_length + DATA_CRC_LENGTH; if data.len() < needed { return StreamDecode::NeedMore; } - if !crc16_valid(&data[consumed..consumed + data_length + 2]) { + if !crc16_valid(&data[consumed..consumed + data_length + DATA_CRC_LENGTH]) { return StreamDecode::Invalid { discard: needed }; } let payload = Bytes::copy_from_slice(&data[consumed..consumed + data_length]); - consumed += data_length + 2; + consumed += data_length + DATA_CRC_LENGTH; StreamDecode::Complete { frame: MstpFrame { @@ -400,6 +422,13 @@ pub fn decode_frame(data: &[u8]) -> Result<(MstpFrame, usize), Error> { return Err(Error::decoding(7, "MS/TP header CRC mismatch")); } + if is_unsupported_cobs_frame_type(data[2]) { + return Err(Error::decoding( + 2, + format!("MS/TP COBS-encoded frame type {} is unsupported", data[2]), + )); + } + let frame_type = FrameType::from_raw(data[2]); let destination = data[3]; let source = data[4]; @@ -423,12 +452,12 @@ pub fn decode_frame(data: &[u8]) -> Result<(MstpFrame, usize), Error> { let data_length = ((data[5] as usize) << 8) | (data[6] as usize); - if data_length > MAX_MPDU_DATA { + if data_length > MAX_STANDARD_MPDU_DATA { return Err(Error::decoding( 5, format!( "MS/TP data length {} exceeds maximum {}", - data_length, MAX_MPDU_DATA + data_length, MAX_STANDARD_MPDU_DATA ), )); } @@ -437,7 +466,7 @@ pub fn decode_frame(data: &[u8]) -> Result<(MstpFrame, usize), Error> { let frame_data = if data_length > 0 { // Need data + 2-byte CRC - let needed = consumed + data_length + 2; + let needed = consumed + data_length + DATA_CRC_LENGTH; if data.len() < needed { return Err(Error::decoding( consumed, @@ -450,7 +479,7 @@ pub fn decode_frame(data: &[u8]) -> Result<(MstpFrame, usize), Error> { } // Verify data CRC (covers data bytes + 2 CRC bytes) - if !crc16_valid(&data[consumed..consumed + data_length + 2]) { + if !crc16_valid(&data[consumed..consumed + data_length + DATA_CRC_LENGTH]) { return Err(Error::decoding( consumed + data_length, "MS/TP data CRC mismatch", @@ -458,7 +487,7 @@ pub fn decode_frame(data: &[u8]) -> Result<(MstpFrame, usize), Error> { } let payload = Bytes::copy_from_slice(&data[consumed..consumed + data_length]); - consumed += data_length + 2; + consumed += data_length + DATA_CRC_LENGTH; payload } else { Bytes::new() diff --git a/crates/bacnet-transport/src/mstp_frame_tests.rs b/crates/bacnet-transport/src/mstp_frame_tests.rs index 83df7a38..a694abe6 100644 --- a/crates/bacnet-transport/src/mstp_frame_tests.rs +++ b/crates/bacnet-transport/src/mstp_frame_tests.rs @@ -42,6 +42,16 @@ fn crc16_clause9_data_vector_01_00() { assert_eq!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); } +#[test] +fn crc16_annex_g_data_vector() { + // Annex G: data 01 22 30 has complemented CRC 0xBD10, + // transmitted least-significant octet first as 10 BD. + assert_eq!(crc16(&[0x01, 0x22, 0x30]), 0xBD10); + let with_crc = [0x01, 0x22, 0x30, 0x10, 0xBD]; + assert!(crc16_valid(&with_crc)); + assert_eq!(crc16_accumulate_all(&with_crc), DATA_CRC_RESIDUAL); +} + #[test] fn crc16_one_bit_corruption_rejected() { let mut with_crc = [0x01, 0x00, 0x9F, 0x16]; @@ -88,6 +98,22 @@ fn literal_token_frame_0_from_7() { assert_eq!(&enc[..], wire); } +#[test] +fn literal_annex_g_token_frame() { + // Annex G: Token, DA 10, SA 05, length 0000, header CRC 8C. + let wire: &[u8] = &[0x55, 0xFF, 0x00, 0x10, 0x05, 0x00, 0x00, 0x8C]; + let (frame, consumed) = decode_frame(wire).expect("decode Annex G Token"); + assert_eq!(consumed, wire.len()); + assert_eq!(frame.frame_type, FrameType::Token); + assert_eq!(frame.destination, 0x10); + assert_eq!(frame.source, 0x05); + assert!(frame.data.is_empty()); + + let mut encoded = BytesMut::new(); + encode_frame(&mut encoded, &frame).unwrap(); + assert_eq!(&encoded[..], wire); +} + #[test] fn literal_data_not_expecting_reply_frame() { // Frame type 06, dest 0, src 7, len 2, HDR D9, data 01 00, DCRC 9F 16 @@ -344,10 +370,116 @@ fn frame_type_has_data() { assert!(!FrameType::ReplyPostponed.has_data()); } +fn data_frame_with_length(data_len: usize) -> MstpFrame { + MstpFrame { + frame_type: FrameType::BACnetDataNotExpectingReply, + destination: BROADCAST_MAC, + source: 0, + data: Bytes::from(vec![0xAA; data_len]), + } +} + +fn header_only_wire(frame_type: u8, data_len: usize) -> Vec { + let header = [ + frame_type, + 1, + 0, + (data_len >> 8) as u8, + (data_len & 0xFF) as u8, + ]; + let mut wire = PREAMBLE.to_vec(); + wire.extend_from_slice(&header); + wire.push(crc8(&header)); + wire +} + +#[test] +fn encode_standard_data_length_boundary() { + let frame = data_frame_with_length(MAX_STANDARD_MPDU_DATA); + let mut buf = BytesMut::new(); + encode_frame(&mut buf, &frame).unwrap(); + assert_eq!(buf.len(), MAX_STANDARD_FRAME_LENGTH); + + let oversized = data_frame_with_length(MAX_STANDARD_MPDU_DATA + 1); + let mut oversized_buf = BytesMut::new(); + assert!(encode_frame(&mut oversized_buf, &oversized).is_err()); + assert!(oversized_buf.is_empty(), "oversized frame wrote wire bytes"); +} + +#[test] +fn decode_standard_data_length_boundary() { + let frame = data_frame_with_length(MAX_STANDARD_MPDU_DATA); + let mut wire = BytesMut::new(); + encode_frame(&mut wire, &frame).unwrap(); + + let (decoded, consumed) = decode_frame(&wire).unwrap(); + assert_eq!(decoded, frame); + assert_eq!(consumed, MAX_STANDARD_FRAME_LENGTH); + + let oversized = header_only_wire(0x06, MAX_STANDARD_MPDU_DATA + 1); + assert!(decode_frame(&oversized).is_err()); +} + +#[test] +fn stream_decode_standard_data_length_boundary() { + let frame = data_frame_with_length(MAX_STANDARD_MPDU_DATA); + let mut wire = BytesMut::new(); + encode_frame(&mut wire, &frame).unwrap(); + + assert_eq!( + decode_frame_stream(&wire), + StreamDecode::Complete { + frame, + consumed: MAX_STANDARD_FRAME_LENGTH, + } + ); + + let oversized = header_only_wire(0x06, MAX_STANDARD_MPDU_DATA + 1); + assert_eq!( + decode_frame_stream(&oversized), + StreamDecode::Invalid { discard: 1 } + ); +} + +#[test] +fn cobs_encoded_frame_type_range_is_rejected() { + for (raw, cobs_encoded) in [(0x1F, false), (0x20, true), (0x7F, true), (0x80, false)] { + let frame = MstpFrame { + frame_type: FrameType::Unknown(raw), + destination: 1, + source: 0, + data: Bytes::new(), + }; + let mut encoded = BytesMut::new(); + let wire = header_only_wire(raw, 0); + if cobs_encoded { + assert!(encode_frame(&mut encoded, &frame).is_err(), "raw={raw}"); + assert!(encoded.is_empty()); + assert!(decode_frame(&wire).is_err(), "raw={raw}"); + assert_eq!( + decode_frame_stream(&wire), + StreamDecode::Invalid { discard: 1 }, + "raw={raw}" + ); + } else { + encode_frame(&mut encoded, &frame).unwrap(); + assert_eq!(&encoded[..], wire, "raw={raw}"); + assert_eq!(decode_frame(&wire).unwrap(), (frame.clone(), wire.len())); + assert_eq!( + decode_frame_stream(&wire), + StreamDecode::Complete { + frame, + consumed: wire.len(), + }, + "raw={raw}" + ); + } + } +} + #[test] -fn large_data_frame() { - // Near-maximum data size - let npdu = vec![0xAA; 1024]; +fn max_standard_data_frame_round_trip() { + let npdu = vec![0xAA; MAX_STANDARD_MPDU_DATA]; let frame = MstpFrame { frame_type: FrameType::BACnetDataNotExpectingReply, destination: BROADCAST_MAC, @@ -367,7 +499,7 @@ fn encode_oversized_data_returns_error() { frame_type: FrameType::BACnetDataNotExpectingReply, destination: 1, source: 0, - data: Bytes::from_static(&[0xAA; MAX_MPDU_DATA + 1]), + data: Bytes::from_static(&[0xAA; MAX_STANDARD_MPDU_DATA + 1]), }; let mut buf = BytesMut::new(); assert!(encode_frame(&mut buf, &frame).is_err()); From 84046ab7e821544ba2fbfd9a6a493af648d7e7fa Mon Sep 17 00:00:00 2001 From: Justin Scott Date: Sun, 30 Aug 2026 16:57:40 -0400 Subject: [PATCH 7/7] fix(mstp): reject oversized outbound payloads early --- .../src/mstp/clause956_tests.rs | 123 +++++++++++++++++- crates/bacnet-transport/src/mstp/mod.rs | 39 +++++- crates/bacnet-transport/src/mstp/port.rs | 26 +++- 3 files changed, 179 insertions(+), 9 deletions(-) diff --git a/crates/bacnet-transport/src/mstp/clause956_tests.rs b/crates/bacnet-transport/src/mstp/clause956_tests.rs index 4b226b4f..240aa27d 100644 --- a/crates/bacnet-transport/src/mstp/clause956_tests.rs +++ b/crates/bacnet-transport/src/mstp/clause956_tests.rs @@ -4,9 +4,12 @@ //! that allowed the invalid ring 0→3→0 and excluded FEC MAC 7. use super::*; -use bytes::Bytes; +use bytes::{Bytes, BytesMut}; use tokio::sync::mpsc; +use crate::mstp_frame::{encode_frame, MAX_STANDARD_MPDU_DATA}; +use crate::port::TransportPort; + fn cfg(ts: u8, max_master: u8) -> MstpConfig { MstpConfig { this_station: ts, @@ -287,3 +290,121 @@ fn pass_token_self_destination_enters_pfm_at_runtime() { assert_eq!(retry.destination, 4); assert_ne!(retry.frame_type, FrameType::Token); } + +fn pending_data_request_node() -> MasterNode { + let (tx, _rx) = mpsc::channel(4); + let mut node = MasterNode::new(cfg(3, 127)).unwrap(); + let request = MstpFrame { + frame_type: FrameType::BACnetDataExpectingReply, + destination: 3, + source: 7, + data: Bytes::from_static(&[0x01, 0x04, 0x10]), + }; + assert!(node.handle_received_frame(&request, &tx).is_none()); + assert_eq!(node.state, MasterState::AnswerDataRequest); + node +} + +#[test] +fn outbound_queue_rejects_oversize_before_mutation() { + let mut node = MasterNode::new(cfg(3, 127)).unwrap(); + let accepted = Bytes::from(vec![0x11; MAX_STANDARD_MPDU_DATA]); + node.queue_npdu(7, accepted.clone()).unwrap(); + + let state_before = node.state; + let frame_count_before = node.frame_count; + let error = node + .queue_npdu(8, Bytes::from(vec![0x22; MAX_STANDARD_MPDU_DATA + 1])) + .expect_err("502-byte NPDU must be rejected"); + assert!(error + .to_string() + .contains("MS/TP NPDU length 502 exceeds standard-frame maximum 501")); + assert_eq!(node.tx_queue.len(), 1); + assert_eq!(node.tx_queue.front(), Some(&(7, accepted))); + assert_eq!(node.state, state_before); + assert_eq!(node.frame_count, frame_count_before); + + node.state = MasterState::UseToken; + let frame = node.use_token(); + assert_eq!(frame.data.len(), MAX_STANDARD_MPDU_DATA); + let mut wire = BytesMut::new(); + encode_frame(&mut wire, &frame).expect("accepted queued NPDU remains encodable"); + assert!(node.tx_queue.is_empty()); +} + +#[test] +fn oversized_application_reply_preserves_state_until_explicit_abandon() { + let mut node = pending_data_request_node(); + assert!(node.reply_rx.is_some()); + + let error = node + .finish_data_request(Some(Bytes::from(vec![0x33; MAX_STANDARD_MPDU_DATA + 1]))) + .expect_err("502-byte application reply must be rejected"); + assert!(error + .to_string() + .contains("MS/TP application reply length 502 exceeds standard-frame maximum 501")); + assert_eq!(node.state, MasterState::AnswerDataRequest); + assert_eq!(node.pending_reply_source, Some(7)); + assert!(node.reply_rx.is_some()); + + node.abandon_data_request(); + assert_eq!(node.state, MasterState::Idle); + assert!(node.pending_reply_source.is_none()); + assert!(node.reply_rx.is_none()); +} + +#[test] +fn max_standard_application_reply_remains_encodable() { + let mut node = pending_data_request_node(); + let reply = node + .finish_data_request(Some(Bytes::from(vec![0x44; MAX_STANDARD_MPDU_DATA]))) + .unwrap(); + + assert_eq!(reply.frame_type, FrameType::BACnetDataNotExpectingReply); + assert_eq!(reply.destination, 7); + assert_eq!(reply.data.len(), MAX_STANDARD_MPDU_DATA); + assert_eq!(node.state, MasterState::Idle); + let mut wire = BytesMut::new(); + encode_frame(&mut wire, &reply).unwrap(); +} + +#[tokio::test(start_paused = true)] +async fn transport_abandons_oversized_reply_without_wire_output() { + let (serial_transport, serial_peer) = LoopbackSerial::pair(); + let mut transport = MstpTransport::new(serial_transport, cfg(3, 127)); + let mut npdu_rx = transport.start().await.unwrap(); + let request = MstpFrame { + frame_type: FrameType::BACnetDataExpectingReply, + destination: 3, + source: 7, + data: Bytes::from_static(&[0x01, 0x04, 0x10]), + }; + let mut wire = BytesMut::new(); + encode_frame(&mut wire, &request).unwrap(); + serial_peer.write(&wire).await.unwrap(); + + let received = npdu_rx.recv().await.expect("application request"); + received + .reply_tx + .expect("reply sender") + .send(Bytes::from(vec![0x55; MAX_STANDARD_MPDU_DATA + 1])) + .unwrap(); + for _ in 0..4 { + tokio::task::yield_now().await; + } + + { + let node = transport.node_state().unwrap().lock().await; + assert_eq!(node.state, MasterState::Idle); + assert!(node.pending_reply_source.is_none()); + assert!(node.reply_rx.is_none()); + } + let mut response = [0u8; MSTP_MAX_FRAME_BUF]; + let read = tokio::time::timeout( + tokio::time::Duration::from_millis(1), + serial_peer.read(&mut response), + ) + .await; + assert!(read.is_err(), "oversized reply produced wire output"); + transport.stop().await.unwrap(); +} diff --git a/crates/bacnet-transport/src/mstp/mod.rs b/crates/bacnet-transport/src/mstp/mod.rs index a048cf8d..af4c8f9d 100644 --- a/crates/bacnet-transport/src/mstp/mod.rs +++ b/crates/bacnet-transport/src/mstp/mod.rs @@ -14,6 +14,7 @@ use tracing::debug; use crate::mstp_frame::{ FrameType, MstpFrame, BROADCAST_MAC, MAX_MASTER, MAX_STANDARD_FRAME_LENGTH, + MAX_STANDARD_MPDU_DATA, }; use crate::port::ReceivedNpdu; @@ -333,11 +334,26 @@ impl MasterNode { } /// Complete AnswerDataRequest with application data or ReplyPostponed. - pub(crate) fn finish_data_request(&mut self, reply_data: Option) -> Option { - let destination = self.pending_reply_source.take()?; + pub(crate) fn finish_data_request( + &mut self, + reply_data: Option, + ) -> Result { + if let Some(data) = reply_data.as_ref() { + if data.len() > MAX_STANDARD_MPDU_DATA { + return Err(Error::Encoding(format!( + "MS/TP application reply length {} exceeds standard-frame maximum {}", + data.len(), + MAX_STANDARD_MPDU_DATA + ))); + } + } + + let destination = self.pending_reply_source.take().ok_or_else(|| { + Error::Encoding("MS/TP has no pending data request to complete".into()) + })?; self.reply_rx = None; self.state = MasterState::Idle; - Some(match reply_data { + Ok(match reply_data { Some(data) => MstpFrame { frame_type: FrameType::BACnetDataNotExpectingReply, destination, @@ -353,6 +369,13 @@ impl MasterNode { }) } + /// Abandon an application reply after a transport-boundary completion error. + fn abandon_data_request(&mut self) { + self.pending_reply_source = None; + self.reply_rx = None; + self.state = MasterState::Idle; + } + /// Decide what to send when we have the token. Returns a frame to send. pub fn use_token(&mut self) -> MstpFrame { // Send queued data if available and under frame limit @@ -583,8 +606,16 @@ impl MasterNode { /// Queue an NPDU for transmission. /// - /// Returns an error if the TX queue has reached [`MAX_TX_QUEUE_DEPTH`]. + /// Returns an error if the NPDU exceeds the supported standard-frame limit + /// or the TX queue has reached [`MAX_TX_QUEUE_DEPTH`]. pub fn queue_npdu(&mut self, dest: u8, npdu: Bytes) -> Result<(), Error> { + if npdu.len() > MAX_STANDARD_MPDU_DATA { + return Err(Error::Encoding(format!( + "MS/TP NPDU length {} exceeds standard-frame maximum {}", + npdu.len(), + MAX_STANDARD_MPDU_DATA + ))); + } if self.tx_queue.len() >= MAX_TX_QUEUE_DEPTH { return Err(Error::Transport(std::io::Error::new( std::io::ErrorKind::WouldBlock, diff --git a/crates/bacnet-transport/src/mstp/port.rs b/crates/bacnet-transport/src/mstp/port.rs index aebca52b..dd0c25ee 100644 --- a/crates/bacnet-transport/src/mstp/port.rs +++ b/crates/bacnet-transport/src/mstp/port.rs @@ -71,6 +71,20 @@ fn assemble_host_chunk(frame_buf: &mut Vec, chunk: &[u8]) -> Vec frames } +fn finish_data_request_at_transport_boundary( + node: &mut MasterNode, + reply_data: Option, +) -> Option { + match node.finish_data_request(reply_data) { + Ok(frame) => Some(frame), + Err(error) => { + warn!("MS/TP application reply rejected: {error}"); + node.abandon_data_request(); + None + } + } +} + // --------------------------------------------------------------------------- // MS/TP Transport // --------------------------------------------------------------------------- @@ -155,7 +169,10 @@ impl TransportPort for MstpTransport { pending_reply_rx = None; pending_reply_deadline = None; let mut node_guard = node.lock().await; - let response = node_guard.finish_data_request(reply.ok()); + let response = finish_data_request_at_transport_boundary( + &mut node_guard, + reply.ok(), + ); drop(node_guard); if let Some(response) = response { @@ -385,9 +402,10 @@ impl TransportPort for MstpTransport { let reply_data = pending_reply_rx .take() .and_then(|mut rx| rx.try_recv().ok()); - if let Some(reply_frame) = - node_guard.finish_data_request(reply_data) - { + if let Some(reply_frame) = finish_data_request_at_transport_boundary( + &mut node_guard, + reply_data, + ) { encode_buf.clear(); if encode_frame(&mut encode_buf, &reply_frame).is_ok() { pending_writes.push(encode_buf.to_vec());