rtc_rtp/
header.rs

1use shared::{
2    error::{Error, Result},
3    marshal::{Marshal, MarshalSize, Unmarshal},
4};
5
6use bytes::{Buf, BufMut, Bytes};
7
8pub const HEADER_LENGTH: usize = 4;
9pub const VERSION_SHIFT: u8 = 6;
10pub const VERSION_MASK: u8 = 0x3;
11pub const PADDING_SHIFT: u8 = 5;
12pub const PADDING_MASK: u8 = 0x1;
13pub const EXTENSION_SHIFT: u8 = 4;
14pub const EXTENSION_MASK: u8 = 0x1;
15pub const EXTENSION_PROFILE_ONE_BYTE: u16 = 0xBEDE;
16pub const EXTENSION_PROFILE_TWO_BYTE: u16 = 0x1000;
17pub const EXTENSION_ID_RESERVED: u8 = 0xF;
18pub const CC_MASK: u8 = 0xF;
19pub const MARKER_SHIFT: u8 = 7;
20pub const MARKER_MASK: u8 = 0x1;
21pub const PT_MASK: u8 = 0x7F;
22pub const SEQ_NUM_OFFSET: usize = 2;
23pub const SEQ_NUM_LENGTH: usize = 2;
24pub const TIMESTAMP_OFFSET: usize = 4;
25pub const TIMESTAMP_LENGTH: usize = 4;
26pub const SSRC_OFFSET: usize = 8;
27pub const SSRC_LENGTH: usize = 4;
28pub const CSRC_OFFSET: usize = 12;
29pub const CSRC_LENGTH: usize = 4;
30
31#[derive(Debug, Eq, PartialEq, Default, Clone)]
32pub struct Extension {
33    pub id: u8,
34    pub payload: Bytes,
35}
36
37/// Header represents an RTP packet header
38/// NOTE: PayloadOffset is populated by Marshal/Unmarshal and should not be modified
39#[derive(Debug, Eq, PartialEq, Default, Clone)]
40pub struct Header {
41    pub version: u8,
42    pub padding: bool,
43    pub extension: bool,
44    pub marker: bool,
45    pub payload_type: u8,
46    pub sequence_number: u16,
47    pub timestamp: u32,
48    pub ssrc: u32,
49    pub csrc: Vec<u32>,
50    pub extension_profile: u16,
51    pub extensions: Vec<Extension>,
52    pub extensions_padding: usize,
53}
54
55impl Unmarshal for Header {
56    /// Unmarshal parses the passed byte slice and stores the result in the Header this method is called upon
57    fn unmarshal<B>(raw_packet: &mut B) -> Result<Self>
58    where
59        Self: Sized,
60        B: Buf,
61    {
62        let raw_packet_len = raw_packet.remaining();
63        if raw_packet_len < HEADER_LENGTH {
64            return Err(Error::ErrHeaderSizeInsufficient);
65        }
66        /*
67         *  0                   1                   2                   3
68         *  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
69         * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
70         * |V=2|P|X|  CC   |M|     PT      |       sequence number         |
71         * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
72         * |                           timestamp                           |
73         * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
74         * |           synchronization source (SSRC) identifier            |
75         * +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
76         * |            contributing source (CSRC) identifiers             |
77         * |                             ....                              |
78         * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
79         */
80        let b0 = raw_packet.get_u8();
81        let version = b0 >> VERSION_SHIFT & VERSION_MASK;
82        let padding = (b0 >> PADDING_SHIFT & PADDING_MASK) > 0;
83        let extension = (b0 >> EXTENSION_SHIFT & EXTENSION_MASK) > 0;
84        let cc = (b0 & CC_MASK) as usize;
85
86        let mut curr_offset = CSRC_OFFSET + (cc * CSRC_LENGTH);
87        if raw_packet_len < curr_offset {
88            return Err(Error::ErrHeaderSizeInsufficient);
89        }
90
91        let b1 = raw_packet.get_u8();
92        let marker = (b1 >> MARKER_SHIFT & MARKER_MASK) > 0;
93        let payload_type = b1 & PT_MASK;
94
95        let sequence_number = raw_packet.get_u16();
96        let timestamp = raw_packet.get_u32();
97        let ssrc = raw_packet.get_u32();
98
99        let mut csrc = Vec::with_capacity(cc);
100        for _ in 0..cc {
101            csrc.push(raw_packet.get_u32());
102        }
103        let mut extensions_padding: usize = 0;
104        let (extension_profile, extensions) = if extension {
105            let expected = curr_offset + 4;
106            if raw_packet_len < expected {
107                return Err(Error::ErrHeaderSizeInsufficientForExtension);
108            }
109            let extension_profile = raw_packet.get_u16();
110            curr_offset += 2;
111            let extension_length = raw_packet.get_u16() as usize * 4;
112            curr_offset += 2;
113
114            let expected = curr_offset + extension_length;
115            if raw_packet_len < expected {
116                return Err(Error::ErrHeaderSizeInsufficientForExtension);
117            }
118
119            let mut extensions = vec![];
120            match extension_profile {
121                // RFC 8285 RTP One Byte Header Extension
122                EXTENSION_PROFILE_ONE_BYTE => {
123                    let end = curr_offset + extension_length;
124                    while curr_offset < end {
125                        let b = raw_packet.get_u8();
126                        if b == 0x00 {
127                            // padding
128                            curr_offset += 1;
129                            extensions_padding += 1;
130                            continue;
131                        }
132
133                        let extid = b >> 4;
134                        let len = ((b & (0xFF ^ 0xF0)) + 1) as usize;
135                        curr_offset += 1;
136
137                        if extid == EXTENSION_ID_RESERVED {
138                            break;
139                        }
140
141                        if len > raw_packet.remaining() {
142                            return Err(Error::ErrHeaderSizeInsufficientForExtension);
143                        }
144
145                        extensions.push(Extension {
146                            id: extid,
147                            payload: raw_packet.copy_to_bytes(len),
148                        });
149                        curr_offset += len;
150                    }
151                }
152                // RFC 8285 RTP Two Byte Header Extension
153                EXTENSION_PROFILE_TWO_BYTE => {
154                    let end = curr_offset + extension_length;
155                    while curr_offset < end {
156                        let b = raw_packet.get_u8();
157                        if b == 0x00 {
158                            // padding
159                            curr_offset += 1;
160                            extensions_padding += 1;
161                            continue;
162                        }
163
164                        let extid = b;
165                        curr_offset += 1;
166
167                        if curr_offset >= end {
168                            return Err(Error::ErrHeaderSizeInsufficientForExtension);
169                        }
170
171                        let len = raw_packet.get_u8() as usize;
172                        curr_offset += 1;
173
174                        if len > raw_packet.remaining() {
175                            return Err(Error::ErrHeaderSizeInsufficientForExtension);
176                        }
177
178                        extensions.push(Extension {
179                            id: extid,
180                            payload: raw_packet.copy_to_bytes(len),
181                        });
182                        curr_offset += len;
183                    }
184                }
185                // RFC3550 Extension
186                _ => {
187                    if raw_packet_len < curr_offset + extension_length {
188                        return Err(Error::ErrHeaderSizeInsufficientForExtension);
189                    }
190                    extensions.push(Extension {
191                        id: 0,
192                        payload: raw_packet.copy_to_bytes(extension_length),
193                    });
194                }
195            };
196
197            (extension_profile, extensions)
198        } else {
199            (0, vec![])
200        };
201
202        Ok(Header {
203            version,
204            padding,
205            extension,
206            marker,
207            payload_type,
208            sequence_number,
209            timestamp,
210            ssrc,
211            csrc,
212            extension_profile,
213            extensions,
214            extensions_padding,
215        })
216    }
217}
218
219impl MarshalSize for Header {
220    /// MarshalSize returns the size of the packet once marshaled.
221    fn marshal_size(&self) -> usize {
222        let mut head_size = 12 + (self.csrc.len() * CSRC_LENGTH);
223        if self.extension {
224            let extension_payload_len = self.get_extension_payload_len() + self.extensions_padding;
225            let extension_payload_size = extension_payload_len.div_ceil(4);
226            head_size += 4 + extension_payload_size * 4;
227        }
228        head_size
229    }
230}
231
232impl Marshal for Header {
233    /// Marshal serializes the header and writes to the buffer.
234    fn marshal_to(&self, mut buf: &mut [u8]) -> Result<usize> {
235        /*
236         *  0                   1                   2                   3
237         *  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
238         * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
239         * |V=2|P|X|  CC   |M|     PT      |       sequence number         |
240         * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
241         * |                           timestamp                           |
242         * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
243         * |           synchronization source (SSRC) identifier            |
244         * +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
245         * |            contributing source (CSRC) identifiers             |
246         * |                             ....                              |
247         * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
248         */
249        let remaining_before = buf.remaining_mut();
250        if remaining_before < self.marshal_size() {
251            return Err(Error::ErrBufferTooSmall);
252        }
253
254        // The first byte contains the version, padding bit, extension bit, and csrc size
255        let mut b0 = (self.version << VERSION_SHIFT) | self.csrc.len() as u8;
256        if self.padding {
257            b0 |= 1 << PADDING_SHIFT;
258        }
259
260        if self.extension {
261            b0 |= 1 << EXTENSION_SHIFT;
262        }
263        buf.put_u8(b0);
264
265        // The second byte contains the marker bit and payload type.
266        let mut b1 = self.payload_type;
267        if self.marker {
268            b1 |= 1 << MARKER_SHIFT;
269        }
270        buf.put_u8(b1);
271
272        buf.put_u16(self.sequence_number);
273        buf.put_u32(self.timestamp);
274        buf.put_u32(self.ssrc);
275
276        for csrc in &self.csrc {
277            buf.put_u32(*csrc);
278        }
279
280        if self.extension {
281            buf.put_u16(self.extension_profile);
282
283            // calculate extensions size and round to 4 bytes boundaries
284            let extension_payload_len = self.get_extension_payload_len();
285            if self.extension_profile != EXTENSION_PROFILE_ONE_BYTE
286                && self.extension_profile != EXTENSION_PROFILE_TWO_BYTE
287                && !extension_payload_len.is_multiple_of(4)
288            {
289                //the payload must be in 32-bit words.
290                return Err(Error::HeaderExtensionPayloadNot32BitWords);
291            }
292            let extension_payload_size = (extension_payload_len as u16).div_ceil(4);
293            buf.put_u16(extension_payload_size);
294
295            match self.extension_profile {
296                // RFC 8285 RTP One Byte Header Extension
297                EXTENSION_PROFILE_ONE_BYTE => {
298                    for extension in &self.extensions {
299                        buf.put_u8((extension.id << 4) | (extension.payload.len() as u8 - 1));
300                        buf.put(&*extension.payload);
301                    }
302                }
303                // RFC 8285 RTP Two Byte Header Extension
304                EXTENSION_PROFILE_TWO_BYTE => {
305                    for extension in &self.extensions {
306                        buf.put_u8(extension.id);
307                        buf.put_u8(extension.payload.len() as u8);
308                        buf.put(&*extension.payload);
309                    }
310                }
311                // RFC3550 Extension
312                _ => {
313                    if self.extensions.len() != 1 {
314                        return Err(Error::ErrRfc3550headerIdrange);
315                    }
316
317                    if let Some(extension) = self.extensions.first() {
318                        let ext_len = extension.payload.len();
319                        if ext_len % 4 != 0 {
320                            return Err(Error::HeaderExtensionPayloadNot32BitWords);
321                        }
322                        buf.put(&*extension.payload);
323                    }
324                }
325            };
326
327            // add padding to reach 4 bytes boundaries
328            for _ in extension_payload_len..extension_payload_size as usize * 4 {
329                buf.put_u8(0);
330            }
331        }
332
333        let remaining_after = buf.remaining_mut();
334        Ok(remaining_before - remaining_after)
335    }
336}
337
338impl Header {
339    pub fn get_extension_payload_len(&self) -> usize {
340        let payload_len: usize = self
341            .extensions
342            .iter()
343            .map(|extension| extension.payload.len())
344            .sum();
345
346        let profile_len = self.extensions.len()
347            * match self.extension_profile {
348                EXTENSION_PROFILE_ONE_BYTE => 1,
349                EXTENSION_PROFILE_TWO_BYTE => 2,
350                _ => 0,
351            };
352
353        payload_len + profile_len
354    }
355
356    /// SetExtension sets an RTP header extension
357    pub fn set_extension(&mut self, id: u8, payload: Bytes) -> Result<()> {
358        let payload_len = payload.len() as isize;
359        if self.extension {
360            let extension_profile_len = match self.extension_profile {
361                EXTENSION_PROFILE_ONE_BYTE => {
362                    if !(1..=14).contains(&id) {
363                        return Err(Error::ErrRfc8285oneByteHeaderIdrange);
364                    }
365                    if payload_len > 16 {
366                        return Err(Error::ErrRfc8285oneByteHeaderSize);
367                    }
368                    1
369                }
370                EXTENSION_PROFILE_TWO_BYTE => {
371                    if id < 1 {
372                        return Err(Error::ErrRfc8285twoByteHeaderIdrange);
373                    }
374                    if payload_len > 255 {
375                        return Err(Error::ErrRfc8285twoByteHeaderSize);
376                    }
377                    2
378                }
379                _ => {
380                    if id != 0 {
381                        return Err(Error::ErrRfc3550headerIdrange);
382                    }
383                    0
384                }
385            };
386
387            let delta;
388            // Update existing if it exists else add new extension
389            if let Some(extension) = self
390                .extensions
391                .iter_mut()
392                .find(|extension| extension.id == id)
393            {
394                delta = payload_len - extension.payload.len() as isize;
395                extension.payload = payload;
396            } else {
397                delta = payload_len + extension_profile_len;
398                self.extensions.push(Extension { id, payload });
399            }
400
401            match delta.cmp(&0) {
402                std::cmp::Ordering::Less => {
403                    self.extensions_padding =
404                        ((self.extensions_padding as isize - delta) % 4) as usize;
405                }
406                std::cmp::Ordering::Greater => {
407                    let extension_padding = (delta % 4) as usize;
408                    if self.extensions_padding < extension_padding {
409                        self.extensions_padding = (self.extensions_padding + 4) - extension_padding;
410                    } else {
411                        self.extensions_padding -= extension_padding
412                    }
413                }
414                _ => {}
415            }
416        } else {
417            // No existing header extensions
418            self.extension = true;
419            let mut extension_profile_len = 0;
420            self.extension_profile = match payload_len {
421                0..=16 => {
422                    extension_profile_len = 1;
423                    EXTENSION_PROFILE_ONE_BYTE
424                }
425                17..=255 => {
426                    extension_profile_len = 2;
427                    EXTENSION_PROFILE_TWO_BYTE
428                }
429                _ => self.extension_profile,
430            };
431
432            let extension_padding = (payload.len() + extension_profile_len) % 4;
433            if self.extensions_padding < extension_padding {
434                self.extensions_padding = self.extensions_padding + 4 - extension_padding;
435            } else {
436                self.extensions_padding -= extension_padding
437            }
438            self.extensions.push(Extension { id, payload });
439        }
440        Ok(())
441    }
442
443    /// returns an extension id array
444    pub fn get_extension_ids(&self) -> Vec<u8> {
445        if self.extension {
446            self.extensions.iter().map(|e| e.id).collect()
447        } else {
448            vec![]
449        }
450    }
451
452    /// returns an RTP header extension
453    pub fn get_extension(&self, id: u8) -> Option<Bytes> {
454        if self.extension {
455            self.extensions
456                .iter()
457                .find(|extension| extension.id == id)
458                .map(|extension| extension.payload.clone())
459        } else {
460            None
461        }
462    }
463
464    /// Removes an RTP Header extension
465    pub fn del_extension(&mut self, id: u8) -> Result<()> {
466        if self.extension {
467            if let Some(index) = self
468                .extensions
469                .iter()
470                .position(|extension| extension.id == id)
471            {
472                let extension = self.extensions.remove(index);
473
474                let extension_profile_len = match self.extension_profile {
475                    EXTENSION_PROFILE_ONE_BYTE => 1,
476                    EXTENSION_PROFILE_TWO_BYTE => 2,
477                    _ => 0,
478                };
479
480                let extension_padding = (extension.payload.len() + extension_profile_len) % 4;
481                self.extensions_padding = (self.extensions_padding + extension_padding) % 4;
482
483                Ok(())
484            } else {
485                Err(Error::ErrHeaderExtensionNotFound)
486            }
487        } else {
488            Err(Error::ErrHeaderExtensionsNotEnabled)
489        }
490    }
491}
rtc_rtp/header.rs

rtc_rtp/
header.rs
