rtc_shared/
turn_framing.rs

1//! Splitting a TURN byte stream back into messages (RFC 8656 §12.5).
2//!
3//! Over TCP or TLS, TURN sends STUN messages and ChannelData messages back to
4//! back with no framing of their own: each is self-delimiting. This is not the
5//! RFC 4571 length prefix that ICE-TCP uses — see [`tcp_framing`](crate::tcp_framing)
6//! for that — and a stream carries one or the other, never both.
7//!
8//! ```text
9//! STUN         00xxxxxx ........ | length (16) | magic cookie, id ... | attributes
10//!              total = 20 + length
11//! ChannelData  01xxxxxx ........ | length (16) | data | padding to a multiple of 4
12//!              total = 4 + length rounded up to 4
13//! ```
14//!
15//! The first two bits tell them apart. Anything else cannot be TURN, and since
16//! a stream has no resynchronisation point it is an error that ends the stream.
17//!
18//! ```rust
19//! use rtc_shared::turn_framing::TurnStreamDecoder;
20//!
21//! // A ChannelData message on channel 0x4000 carrying three bytes, padded to 8.
22//! let wire = [0x40, 0x00, 0x00, 0x03, b'a', b'b', b'c', 0x00];
23//! let mut decoder = TurnStreamDecoder::new();
24//! decoder.extend_from_slice(&wire[..5]);
25//! assert_eq!(decoder.next_message().unwrap(), None); // not all of it yet
26//! decoder.extend_from_slice(&wire[5..]);
27//! assert_eq!(decoder.next_message().unwrap(), Some(wire.to_vec()));
28//! ```
29
30use crate::error::{Error, Result};
31
32/// Sans-IO splitter for a TURN stream: push bytes as they arrive, pop whole
33/// messages. A message comes out exactly as it was on the wire, ChannelData
34/// padding included, which TURN's decoders accept.
35#[derive(Debug, Default)]
36pub struct TurnStreamDecoder {
37    buffer: Vec<u8>,
38}
39
40impl TurnStreamDecoder {
41    /// Creates a decoder with an empty buffer.
42    pub fn new() -> Self {
43        Self::default()
44    }
45
46    /// Appends bytes read from the stream.
47    pub fn extend_from_slice(&mut self, data: &[u8]) {
48        self.buffer.extend_from_slice(data);
49    }
50
51    /// The next complete message, `Ok(None)` while one is still arriving, or an
52    /// error when the stream holds something that is not TURN.
53    pub fn next_message(&mut self) -> Result<Option<Vec<u8>>> {
54        // Both kinds carry their length in bytes 2..4, so four bytes decide it.
55        if self.buffer.len() < CHANNEL_DATA_HEADER_LEN {
56            return Ok(None);
57        }
58        let length = u16::from_be_bytes([self.buffer[2], self.buffer[3]]) as usize;
59        let total = match self.buffer[0] >> 6 {
60            0b00 => STUN_HEADER_LEN + length,
61            0b01 => CHANNEL_DATA_HEADER_LEN + length.next_multiple_of(4),
62            _ => {
63                return Err(Error::OtherTurnErr(format!(
64                    "stream byte {:#04x} starts neither a STUN nor a ChannelData message",
65                    self.buffer[0]
66                )));
67            }
68        };
69        if self.buffer.len() < total {
70            return Ok(None);
71        }
72        let message = self.buffer[..total].to_vec();
73        self.buffer.drain(..total);
74        Ok(Some(message))
75    }
76
77    /// Bytes held that do not yet make a whole message.
78    pub fn buffered_len(&self) -> usize {
79        self.buffer.len()
80    }
81}
82
83/// A STUN header: type, length, magic cookie and transaction id (RFC 8489 §5).
84const STUN_HEADER_LEN: usize = 20;
85/// A ChannelData header: channel number and length (RFC 8656 §12.4).
86const CHANNEL_DATA_HEADER_LEN: usize = 4;
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91
92    /// A STUN Binding request header with `length` bytes of attributes (zeros).
93    fn stun(length: u16) -> Vec<u8> {
94        let mut msg = vec![0x00, 0x01];
95        msg.extend_from_slice(&length.to_be_bytes());
96        msg.extend_from_slice(&[0x21, 0x12, 0xa4, 0x42]);
97        msg.extend_from_slice(&[7; 12]);
98        msg.extend(std::iter::repeat_n(0u8, length as usize));
99        msg
100    }
101
102    /// A ChannelData message on 0x4000 carrying `data`, padded to a multiple of 4.
103    fn channel_data(data: &[u8]) -> Vec<u8> {
104        let mut msg = vec![0x40, 0x00];
105        msg.extend_from_slice(&(data.len() as u16).to_be_bytes());
106        msg.extend_from_slice(data);
107        while msg.len() % 4 != 0 {
108            msg.push(0);
109        }
110        msg
111    }
112
113    fn drain(decoder: &mut TurnStreamDecoder) -> Vec<Vec<u8>> {
114        let mut out = Vec::new();
115        while let Some(msg) = decoder.next_message().unwrap() {
116            out.push(msg);
117        }
118        out
119    }
120
121    #[test]
122    fn a_stun_message_is_twenty_bytes_plus_its_length() {
123        let msg = stun(8);
124        let mut decoder = TurnStreamDecoder::new();
125        decoder.extend_from_slice(&msg);
126        assert_eq!(drain(&mut decoder), vec![msg]);
127        assert_eq!(decoder.buffered_len(), 0);
128    }
129
130    #[test]
131    fn channel_data_comes_out_with_its_padding() {
132        for data in [&b""[..], b"a", b"ab", b"abc", b"abcd", b"abcde"] {
133            let msg = channel_data(data);
134            let mut decoder = TurnStreamDecoder::new();
135            decoder.extend_from_slice(&msg);
136            assert_eq!(drain(&mut decoder), vec![msg], "{} data bytes", data.len());
137            assert_eq!(decoder.buffered_len(), 0);
138        }
139    }
140
141    #[test]
142    fn messages_arriving_together_come_out_one_by_one() {
143        let (a, b, c) = (stun(4), channel_data(b"xyz"), stun(0));
144        let mut decoder = TurnStreamDecoder::new();
145        decoder.extend_from_slice(&[a.clone(), b.clone(), c.clone()].concat());
146        assert_eq!(drain(&mut decoder), vec![a, b, c]);
147    }
148
149    #[test]
150    fn a_message_split_across_reads_waits_until_it_is_whole() {
151        let msg = channel_data(b"hello");
152        let mut decoder = TurnStreamDecoder::new();
153        // One byte at a time, including splits inside the four-byte header.
154        for (i, byte) in msg.iter().enumerate() {
155            decoder.extend_from_slice(&[*byte]);
156            if i + 1 < msg.len() {
157                assert_eq!(decoder.next_message().unwrap(), None, "early at byte {i}");
158            }
159        }
160        assert_eq!(drain(&mut decoder), vec![msg]);
161    }
162
163    #[test]
164    fn a_partial_message_after_a_whole_one_stays_buffered() {
165        let whole = stun(0);
166        let next = channel_data(b"later");
167        let mut decoder = TurnStreamDecoder::new();
168        decoder.extend_from_slice(&[whole.clone(), next[..3].to_vec()].concat());
169        assert_eq!(drain(&mut decoder), vec![whole]);
170        assert_eq!(decoder.buffered_len(), 3);
171        decoder.extend_from_slice(&next[3..]);
172        assert_eq!(drain(&mut decoder), vec![next]);
173    }
174
175    #[test]
176    fn bytes_that_are_not_turn_end_the_stream() {
177        // 10xxxxxx and 11xxxxxx are neither STUN nor ChannelData.
178        for first in [0x80u8, 0xbf, 0xc0, 0xff] {
179            let mut decoder = TurnStreamDecoder::new();
180            decoder.extend_from_slice(&[first, 0, 0, 4, 1, 2, 3, 4]);
181            assert!(
182                matches!(decoder.next_message(), Err(Error::OtherTurnErr(_))),
183                "first byte {first:#04x} was accepted"
184            );
185        }
186    }
187}
rtc_shared/turn_framing.rs

rtc_shared/
turn_framing.rs
