sipx_rtp/
packet.rs

1//! RTP packets (RFC 3550 §5.1).
2//!
3//! The header is twelve fixed bytes plus optional contributing sources and an optional
4//! extension, and the whole of it is bit-packed. Every field here has been the cause of a
5//! decoder reading someone else's audio as its own, so each is parsed explicitly rather than
6//! by casting a struct over the buffer.
7
8use bytes::{BufMut, Bytes, BytesMut};
9
10/// The fixed header size, before CSRCs or extensions.
11pub const HEADER_LEN: usize = 12;
12
13/// What can go wrong reading a packet.
14#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
15#[non_exhaustive]
16pub enum RtpError {
17    /// Fewer bytes than a header.
18    #[error("packet is {0} bytes; an RTP header is {HEADER_LEN}")]
19    TooShort(usize),
20    /// A version other than 2.
21    #[error("RTP version {0}; only version 2 exists")]
22    BadVersion(u8),
23    /// The header claims more content than the packet holds.
24    #[error("header claims more bytes than the packet contains")]
25    Truncated,
26    /// The padding length is impossible.
27    #[error("padding of {0} bytes does not fit")]
28    BadPadding(usize),
29}
30
31/// An RTP packet.
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct Packet {
34    /// Whether this packet marks a significant event — the start of a talkspurt, or the end of
35    /// a DTMF tone.
36    pub marker: bool,
37    /// Which codec the payload is in.
38    pub payload_type: u8,
39    /// Increases by one per packet, and wraps.
40    pub sequence: u16,
41    /// The sampling instant of the first byte of payload.
42    pub timestamp: u32,
43    /// Who sent it.
44    pub ssrc: u32,
45    /// Sources that contributed, when a mixer combined streams.
46    pub csrc: Vec<u32>,
47    /// The media.
48    pub payload: Bytes,
49}
50
51impl Packet {
52    /// A packet carrying a payload.
53    #[must_use]
54    pub fn new(payload_type: u8, sequence: u16, timestamp: u32, ssrc: u32, payload: Bytes) -> Self {
55        Self {
56            marker: false,
57            payload_type,
58            sequence,
59            timestamp,
60            ssrc,
61            csrc: Vec::new(),
62            payload,
63        }
64    }
65
66    /// Serialize to the wire.
67    #[must_use]
68    pub fn encode(&self) -> Bytes {
69        let csrc_count = self.csrc.len().min(15);
70        let mut out = BytesMut::with_capacity(HEADER_LEN + csrc_count * 4 + self.payload.len());
71
72        // Version 2, no padding, no extension, and the CSRC count in the low nibble.
73        let first = 0b1000_0000 | u8::try_from(csrc_count).unwrap_or(0);
74        out.put_u8(first);
75        out.put_u8((u8::from(self.marker) << 7) | (self.payload_type & 0x7F));
76        out.put_u16(self.sequence);
77        out.put_u32(self.timestamp);
78        out.put_u32(self.ssrc);
79        for csrc in self.csrc.iter().take(csrc_count) {
80            out.put_u32(*csrc);
81        }
82        out.put_slice(&self.payload);
83        out.freeze()
84    }
85
86    /// Parse a packet.
87    ///
88    /// Rejects rather than guesses. A decoder that reads a malformed packet optimistically
89    /// ends up playing header bytes as audio, which is heard as a loud click.
90    pub fn decode(bytes: &Bytes) -> Result<Self, RtpError> {
91        if bytes.len() < HEADER_LEN {
92            return Err(RtpError::TooShort(bytes.len()));
93        }
94
95        let first = bytes.first().copied().unwrap_or(0);
96        let version = first >> 6;
97        if version != 2 {
98            return Err(RtpError::BadVersion(version));
99        }
100        let has_padding = first & 0b0010_0000 != 0;
101        let has_extension = first & 0b0001_0000 != 0;
102        let csrc_count = usize::from(first & 0x0F);
103
104        let second = bytes.get(1).copied().unwrap_or(0);
105        let marker = second & 0x80 != 0;
106        let payload_type = second & 0x7F;
107
108        let sequence = u16::from_be_bytes([
109            bytes.get(2).copied().unwrap_or(0),
110            bytes.get(3).copied().unwrap_or(0),
111        ]);
112        let timestamp = read_u32(bytes, 4)?;
113        let ssrc = read_u32(bytes, 8)?;
114
115        let mut offset = HEADER_LEN;
116        let mut csrc = Vec::with_capacity(csrc_count);
117        for _ in 0..csrc_count {
118            csrc.push(read_u32(bytes, offset)?);
119            offset += 4;
120        }
121
122        if has_extension {
123            // The extension is a 16-bit profile field, a 16-bit length in 32-bit words, then
124            // that many words. The length excludes the four bytes of the header itself, which
125            // is the detail that makes off-by-one errors here so easy.
126            let words = usize::from(u16::from_be_bytes([
127                bytes.get(offset + 2).copied().ok_or(RtpError::Truncated)?,
128                bytes.get(offset + 3).copied().ok_or(RtpError::Truncated)?,
129            ]));
130            offset = offset
131                .checked_add(4 + words * 4)
132                .ok_or(RtpError::Truncated)?;
133        }
134
135        if offset > bytes.len() {
136            return Err(RtpError::Truncated);
137        }
138        let mut end = bytes.len();
139
140        if has_padding {
141            // The last byte says how many bytes of padding there are, *including itself*.
142            let pad = usize::from(bytes.last().copied().unwrap_or(0));
143            if pad == 0 || pad > end - offset {
144                return Err(RtpError::BadPadding(pad));
145            }
146            end -= pad;
147        }
148
149        Ok(Self {
150            marker,
151            payload_type,
152            sequence,
153            timestamp,
154            ssrc,
155            csrc,
156            payload: bytes.slice(offset..end),
157        })
158    }
159}
160
161fn read_u32(bytes: &Bytes, at: usize) -> Result<u32, RtpError> {
162    Ok(u32::from_be_bytes([
163        bytes.get(at).copied().ok_or(RtpError::Truncated)?,
164        bytes.get(at + 1).copied().ok_or(RtpError::Truncated)?,
165        bytes.get(at + 2).copied().ok_or(RtpError::Truncated)?,
166        bytes.get(at + 3).copied().ok_or(RtpError::Truncated)?,
167    ]))
168}
169
170/// Compare two sequence numbers across the 16-bit wrap.
171///
172/// The counter wraps every ~22 minutes at 50 packets per second, so this is an ordinary event
173/// in any call worth having, not an edge case. Comparing with `<` instead treats the wrap as a
174/// 65535-packet jump backwards and throws away a minute of audio while the buffer resyncs.
175#[must_use]
176pub fn sequence_is_newer(candidate: u16, current: u16) -> bool {
177    // RFC 1982 serial number arithmetic: the difference, read as signed, is the distance.
178    candidate != current && candidate.wrapping_sub(current) < 0x8000
179}
180
181/// The forward distance from one sequence number to another, across the wrap.
182#[must_use]
183pub fn sequence_distance(from: u16, to: u16) -> u16 {
184    to.wrapping_sub(from)
185}
186
187#[cfg(test)]
188#[allow(
189    clippy::unwrap_used,
190    clippy::expect_used,
191    clippy::panic,
192    clippy::indexing_slicing
193)]
194mod tests {
195    use super::*;
196
197    fn sample_packet() -> Packet {
198        Packet::new(
199            0,
200            1000,
201            160_000,
202            0xDEAD_BEEF,
203            Bytes::from_static(&[0xFF; 160]),
204        )
205    }
206
207    #[test]
208    fn a_packet_round_trips_through_the_wire_format() {
209        let original = sample_packet();
210        let decoded = Packet::decode(&original.encode()).expect("decodes");
211        assert_eq!(decoded, original);
212    }
213
214    #[test]
215    fn the_header_is_twelve_bytes_before_the_payload() {
216        let encoded = sample_packet().encode();
217        assert_eq!(encoded.len(), HEADER_LEN + 160);
218        assert_eq!(encoded[0] >> 6, 2, "version 2");
219        assert_eq!(encoded[1] & 0x7F, 0, "payload type 0");
220        assert_eq!(u16::from_be_bytes([encoded[2], encoded[3]]), 1000);
221    }
222
223    #[test]
224    fn the_marker_bit_survives_a_round_trip() {
225        let mut packet = sample_packet();
226        packet.marker = true;
227        let decoded = Packet::decode(&packet.encode()).expect("decodes");
228        assert!(decoded.marker);
229        assert_eq!(
230            decoded.payload_type, 0,
231            "the marker must not bleed into the type"
232        );
233    }
234
235    /// A payload type of 127 sets every bit the marker does not. Packing them into one byte is
236    /// where a decoder starts reading type 127 as a marker with type 0.
237    #[test]
238    fn a_high_payload_type_does_not_collide_with_the_marker() {
239        let mut packet = sample_packet();
240        packet.payload_type = 127;
241        packet.marker = false;
242        let decoded = Packet::decode(&packet.encode()).expect("decodes");
243        assert_eq!(decoded.payload_type, 127);
244        assert!(!decoded.marker);
245    }
246
247    #[test]
248    fn contributing_sources_survive() {
249        let mut packet = sample_packet();
250        packet.csrc = vec![1, 2, 3];
251        let decoded = Packet::decode(&packet.encode()).expect("decodes");
252        assert_eq!(decoded.csrc, vec![1, 2, 3]);
253        assert_eq!(
254            decoded.payload, packet.payload,
255            "the payload starts after them"
256        );
257    }
258
259    /// The padding count includes itself and must be removed from the payload. Leaving it in
260    /// plays padding as audio.
261    #[test]
262    fn padding_is_stripped_from_the_payload() {
263        let mut raw = BytesMut::new();
264        raw.put_u8(0b1010_0000); // version 2, padding set
265        raw.put_u8(0);
266        raw.put_u16(1);
267        raw.put_u32(0);
268        raw.put_u32(0);
269        raw.put_slice(&[1, 2, 3, 4]);
270        raw.put_slice(&[0, 0, 0, 4]); // four bytes of padding, the last being the count
271
272        let decoded = Packet::decode(&raw.freeze()).expect("decodes");
273        assert_eq!(decoded.payload.as_ref(), &[1, 2, 3, 4]);
274    }
275
276    #[test]
277    fn impossible_padding_is_rejected() {
278        let mut raw = BytesMut::new();
279        raw.put_u8(0b1010_0000);
280        raw.put_u8(0);
281        raw.put_u16(1);
282        raw.put_u32(0);
283        raw.put_u32(0);
284        raw.put_slice(&[1, 2, 200]); // claims 200 bytes of padding in a 3-byte payload
285        assert!(matches!(
286            Packet::decode(&raw.freeze()),
287            Err(RtpError::BadPadding(200))
288        ));
289    }
290
291    /// The extension length counts 32-bit words and excludes its own four-byte header, which
292    /// is where off-by-one errors here come from.
293    #[test]
294    fn a_header_extension_is_skipped_not_played() {
295        let mut raw = BytesMut::new();
296        raw.put_u8(0b1001_0000); // version 2, extension set
297        raw.put_u8(0);
298        raw.put_u16(7);
299        raw.put_u32(0);
300        raw.put_u32(0);
301        raw.put_u16(0xBEDE); // profile
302        raw.put_u16(2); // two words follow
303        raw.put_slice(&[9; 8]);
304        raw.put_slice(&[1, 2, 3]);
305
306        let decoded = Packet::decode(&raw.freeze()).expect("decodes");
307        assert_eq!(
308            decoded.payload.as_ref(),
309            &[1, 2, 3],
310            "the extension is not payload"
311        );
312    }
313
314    #[test]
315    fn a_short_packet_is_rejected() {
316        assert!(matches!(
317            Packet::decode(&Bytes::from_static(&[0x80, 0, 0])),
318            Err(RtpError::TooShort(3))
319        ));
320    }
321
322    /// Version 1 does not exist in the wild and version 0 is usually a stray STUN packet on the
323    /// RTP port. Either way it is not audio.
324    #[test]
325    fn a_wrong_version_is_rejected() {
326        let mut raw = BytesMut::from(&[0u8; 12][..]);
327        raw[0] = 0b0100_0000; // version 1
328        assert!(matches!(
329            Packet::decode(&raw.freeze()),
330            Err(RtpError::BadVersion(1))
331        ));
332    }
333
334    #[test]
335    fn a_truncated_csrc_list_is_rejected() {
336        let mut raw = BytesMut::from(&[0u8; 12][..]);
337        raw[0] = 0b1000_0011; // claims three CSRCs that are not there
338        assert!(matches!(
339            Packet::decode(&raw.freeze()),
340            Err(RtpError::Truncated)
341        ));
342    }
343
344    /// The failing-first test for this story. The counter wraps every ~22 minutes at 50 packets
345    /// per second; treating that as a jump backwards throws away audio while the buffer
346    /// resynchronises.
347    #[test]
348    fn sequence_wraparound_is_ordered_correctly() {
349        assert!(sequence_is_newer(1, 0));
350        assert!(!sequence_is_newer(0, 1));
351
352        // Across the wrap: 0 follows 65535.
353        assert!(sequence_is_newer(0, 65_535));
354        assert!(!sequence_is_newer(65_535, 0));
355        assert!(sequence_is_newer(5, 65_530));
356        assert!(!sequence_is_newer(65_530, 5));
357
358        // A number is never newer than itself.
359        assert!(!sequence_is_newer(42, 42));
360
361        // Half the space away is the boundary where "newer" stops meaning anything; the
362        // convention is that it counts as older.
363        assert!(!sequence_is_newer(0x8000, 0));
364        assert!(sequence_is_newer(0x7FFF, 0));
365    }
366
367    #[test]
368    fn sequence_distance_counts_forward_across_the_wrap() {
369        assert_eq!(sequence_distance(0, 1), 1);
370        assert_eq!(sequence_distance(65_535, 0), 1);
371        assert_eq!(sequence_distance(65_530, 5), 11);
372        assert_eq!(sequence_distance(10, 10), 0);
373    }
374}
sipx_rtp/packet.rs

sipx_rtp/
packet.rs
