retina/
rtp.rs

1// Copyright (C) The Retina Authors
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! Handles RTP data as described in
5//! [RFC 3550 section 5.1](https://datatracker.ietf.org/doc/html/rfc3550#section-5.1).
6
7use std::ops::Range;
8
9use bytes::{Buf, Bytes};
10
11use crate::inputs::Input;
12use crate::{PacketContext, Timestamp};
13
14/// Fixed RTP packet header (no CSRCs, payload, or extensions).
15///
16/// This is a thin wrapper around the raw 12-byte header, with accessor
17/// methods for the individual fields. It can be constructed on-the-fly
18/// from a [`ReceivedPacket`] or via [`PacketHeader::validate`].
19#[derive(Clone, Copy, Eq, PartialEq)]
20pub struct PacketHeader(
21    /// ```text
22    ///  0                   1                   2                   3
23    ///  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
24    /// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
25    /// |V=2|P|X|  CC   |M|     PT      |       sequence number         |
26    /// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
27    /// |                           timestamp                           |
28    /// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
29    /// |           synchronization source (SSRC) identifier            |
30    /// +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
31    /// ```
32    pub(crate) [u8; Self::LEN as usize],
33);
34
35impl PacketHeader {
36    const LEN: u16 = 12;
37
38    /// Validates an RTP packet, returning the fixed header and payload range.
39    ///
40    /// Accepts any [`Input`] implementation: a `&[u8]` or a
41    /// [`Split`](crate::inputs::Split) pair of ring-buffer slices.
42    pub fn validate<'i>(input: impl Input<'i>) -> Result<(Self, Range<u16>), &'static str> {
43        let len = u16::try_from(input.len()).map_err(|_| "too long")?;
44        if len < Self::LEN {
45            return Err("too short");
46        }
47        let header = Self(input.peek_array::<{ Self::LEN as usize }>());
48        if (header.0[0] & 0b1100_0000) != 2 << 6 {
49            return Err("must be version 2");
50        }
51        let has_padding = (header.0[0] & 0b0010_0000) != 0;
52        let has_extension = (header.0[0] & 0b0001_0000) != 0;
53        let csrc_count = header.0[0] & 0b0000_1111;
54        let csrc_end = Self::LEN + (4 * u16::from(csrc_count));
55        let payload_start = if has_extension {
56            if input.len() < usize::from(csrc_end + 4) {
57                return Err("extension is after end of packet");
58            }
59            let extension_len = u16::from_be_bytes([
60                input.byte_at(usize::from(csrc_end) + 2),
61                input.byte_at(usize::from(csrc_end) + 3),
62            ]);
63            extension_len
64                .checked_mul(4)
65                .and_then(|e| e.checked_add(csrc_end + 4))
66                .ok_or("extension extends beyond maximum packet size")?
67        } else {
68            csrc_end
69        };
70        if len < payload_start {
71            return Err("payload start is after end of packet");
72        }
73        let payload_end = if has_padding {
74            if len == payload_start {
75                return Err("missing padding");
76            }
77            let padding_len = u16::from(input.byte_at(input.len() - 1));
78            if padding_len == 0 {
79                return Err("invalid padding length 0");
80            }
81            let payload_end = len
82                .checked_sub(padding_len)
83                .ok_or("padding larger than packet")?;
84            if payload_end < payload_start {
85                return Err("bad padding");
86            }
87            payload_end
88        } else {
89            len
90        };
91        Ok((header, payload_start..payload_end))
92    }
93
94    #[inline]
95    pub fn mark(&self) -> bool {
96        (self.0[1] & 0b1000_0000) != 0
97    }
98
99    #[inline]
100    pub fn payload_type(&self) -> u8 {
101        self.0[1] & 0b0111_1111
102    }
103
104    #[inline]
105    pub fn sequence_number(&self) -> u16 {
106        u16::from_be_bytes([self.0[2], self.0[3]])
107    }
108
109    #[inline]
110    pub fn timestamp(&self) -> u32 {
111        u32::from_be_bytes([self.0[4], self.0[5], self.0[6], self.0[7]])
112    }
113
114    #[inline]
115    pub fn ssrc(&self) -> u32 {
116        u32::from_be_bytes([self.0[8], self.0[9], self.0[10], self.0[11]])
117    }
118}
119
120impl std::fmt::Debug for PacketHeader {
121    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
122        f.debug_struct("PacketHeader")
123            .field("mark", &self.mark())
124            .field("payload_type", &self.payload_type())
125            .field("sequence_number", &self.sequence_number())
126            .field("timestamp", &self.timestamp())
127            .field("ssrc", &self.ssrc())
128            .finish()
129    }
130}
131
132/// A received RTP packet.
133///
134/// This holds more information than the packet itself: also a
135/// [`PacketContext`], the stream, and extended timestamp.
136#[derive(Eq, PartialEq)]
137pub struct ReceivedPacket {
138    pub(crate) ctx: PacketContext,
139    pub(crate) stream_id: usize,
140    pub(crate) timestamp: crate::Timestamp,
141
142    /// Full packet data, including headers.
143    pub(crate) data: Bytes,
144    pub(crate) payload_range: Range<u16>,
145
146    // TODO: consider dropping this field in favor of a PacketItem::Loss.
147    // https://github.com/scottlamb/retina/issues/47
148    pub(crate) loss: u16,
149}
150
151impl std::fmt::Debug for ReceivedPacket {
152    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
153        f.debug_struct("ReceivedPacket")
154            .field("ctx", &self.ctx)
155            .field("stream_id", &self.stream_id)
156            .field("timestamp", &self.timestamp)
157            .field("ssrc", &self.ssrc())
158            .field("sequence_number", &self.sequence_number())
159            .field("mark", &self.mark())
160            .field("payload", &crate::hex::LimitedHex::new(self.payload(), 64))
161            .finish()
162    }
163}
164
165impl ReceivedPacket {
166    /// Returns the fixed 12-byte RTP header.
167    ///
168    /// This constructs a [`PacketHeader`] on-the-fly from the packet data;
169    /// inlining should avoid an actual copy in most cases.
170    #[inline]
171    pub fn header(&self) -> PacketHeader {
172        PacketHeader(self.data[..PacketHeader::LEN as usize].try_into().unwrap())
173    }
174
175    #[inline]
176    pub fn timestamp(&self) -> crate::Timestamp {
177        self.timestamp
178    }
179
180    #[inline]
181    pub fn mark(&self) -> bool {
182        self.header().mark()
183    }
184
185    #[inline]
186    pub fn ctx(&self) -> &PacketContext {
187        &self.ctx
188    }
189
190    #[inline]
191    pub fn stream_id(&self) -> usize {
192        self.stream_id
193    }
194
195    #[inline]
196    pub fn ssrc(&self) -> u32 {
197        self.header().ssrc()
198    }
199
200    #[inline]
201    pub fn sequence_number(&self) -> u16 {
202        self.header().sequence_number()
203    }
204
205    /// Returns the raw bytes, including the RTP headers.
206    #[inline]
207    pub fn raw(&self) -> &[u8] {
208        &self.data[..]
209    }
210
211    /// Returns only the payload bytes.
212    #[inline]
213    pub fn payload(&self) -> &[u8] {
214        &self.data[usize::from(self.payload_range.start)..usize::from(self.payload_range.end)]
215    }
216
217    #[inline]
218    pub fn loss(&self) -> u16 {
219        self.loss
220    }
221
222    /// Consumes the `ReceivedPacket` and returns the `Payload` as a [`Bytes`].
223    ///
224    /// This is currently very efficient (no copying or reference-counting),
225    /// although that is not an API guarantee.
226    #[inline]
227    pub fn into_payload_bytes(self) -> Bytes {
228        let mut data = self.data;
229        data.truncate(usize::from(self.payload_range.end));
230        data.advance(usize::from(self.payload_range.start));
231        data
232    }
233}
234
235/// Metadata extracted from a received RTP packet, for use by depacketizers.
236///
237/// This allows depacketizers to work without depending on [`ReceivedPacket`],
238/// enabling the `Demuxed` path to pass data directly from the ring buffer.
239#[derive(Clone, Copy, Debug)]
240#[doc(hidden)]
241pub struct PacketMeta {
242    pub ctx: crate::PacketContext,
243    pub stream_id: usize,
244    pub timestamp: crate::Timestamp,
245    pub sequence_number: u16,
246    pub ssrc: u32,
247    pub mark: bool,
248    pub loss: u16,
249}
250
251impl PacketMeta {
252    pub fn from_received(pkt: &ReceivedPacket) -> Self {
253        let header = pkt.header();
254        PacketMeta {
255            ctx: *pkt.ctx(),
256            stream_id: pkt.stream_id(),
257            timestamp: pkt.timestamp(),
258            sequence_number: header.sequence_number(),
259            ssrc: header.ssrc(),
260            mark: header.mark(),
261            loss: pkt.loss(),
262        }
263    }
264}
265
266/// Builds raw RTP packet bytes for testing.
267pub(crate) fn build_raw_rtp<P: IntoIterator<Item = u8>>(
268    sequence_number: u16,
269    timestamp: u32,
270    payload_type: u8,
271    ssrc: u32,
272    mark: bool,
273    payload: P,
274) -> Result<Bytes, &'static str> {
275    if payload_type >= 0x80 {
276        return Err("payload type too large");
277    }
278    let data: Bytes = [
279        2 << 6, // version=2, no padding, no extensions, no CSRCs.
280        if mark { 0b1000_0000 } else { 0 } | payload_type,
281    ]
282    .into_iter()
283    .chain(sequence_number.to_be_bytes())
284    .chain(timestamp.to_be_bytes())
285    .chain(ssrc.to_be_bytes())
286    .chain(payload)
287    .collect();
288    let _ = u16::try_from(data.len()).map_err(|_| "payload too long")?;
289    Ok(data)
290}
291
292/// Testing API; exposed for fuzz tests.
293#[doc(hidden)]
294pub struct ReceivedPacketBuilder {
295    pub ctx: PacketContext,
296    pub stream_id: usize,
297    pub sequence_number: u16,
298    pub timestamp: Timestamp,
299    pub payload_type: u8,
300    pub ssrc: u32,
301    pub mark: bool,
302    pub loss: u16,
303}
304
305impl ReceivedPacketBuilder {
306    pub fn build<P: IntoIterator<Item = u8>>(
307        self,
308        payload: P,
309    ) -> Result<ReceivedPacket, &'static str> {
310        let data = build_raw_rtp(
311            self.sequence_number,
312            self.timestamp.timestamp as u32,
313            self.payload_type,
314            self.ssrc,
315            self.mark,
316            payload,
317        )?;
318        let len = data.len() as u16;
319        Ok(ReceivedPacket {
320            ctx: self.ctx,
321            stream_id: self.stream_id,
322            timestamp: self.timestamp,
323            data,
324            payload_range: PacketHeader::LEN..len,
325            loss: self.loss,
326        })
327    }
328}
329
330#[cfg(test)]
331mod tests {
332    use super::*;
333    use crate::testutil::init_logging;
334
335    #[test]
336    pub fn pkt_with_extension() {
337        init_logging();
338        let data = b"\x90\x60\x4c\x62\x01\xbb\x3c\xb5\x1c\x04\x15\xb1\xab\xac\x00\x03\
339                     \x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x67\x64\x00\x32\
340                     \xac\x3c\x6b\x81\x7c\x05\x46\x9b\x82\x80\x82\xa0\x00\x00\x03\x00\
341                     \x20\x00\x00\x07\x90\x80\x00";
342        let (header, payload_range) = PacketHeader::validate(&data[..]).unwrap();
343        assert_eq!(payload_range, 28..55);
344        assert_eq!(data[payload_range.start as usize], 0x67);
345        assert!(!header.mark());
346    }
347}
retina/rtp.rs

retina/
rtp.rs
