Add `RtcpCompoundPacket`

src/rfc3550/mod.rs

@@ -2,7 +2,7 @@ pub use self::rtp::{RtpPacket, RtpFixedHeader, RtpHeaderExtension, RtpPacketRead
 
 pub use self::rtcp::{RtcpPacket, RtcpSenderReport, RtcpReceiverReport, RtcpPacketReader};
 pub use self::rtcp::{RtcpSourceDescription, RtcpGoodbye, RtcpApplicationDefined};
-pub use self::rtcp::{ReceptionReport, SdesChunk, SdesItem};
+pub use self::rtcp::{RtcpCompoundPacket, ReceptionReport, SdesChunk, SdesItem};
 
 pub use self::rtcp::{RTCP_PACKET_TYPE_SR, RTCP_PACKET_TYPE_RR, RTCP_PACKET_TYPE_SDES};
 pub use self::rtcp::{RTCP_PACKET_TYPE_BYE, RTCP_PACKET_TYPE_APP};

src/rfc3550/rtcp.rs

@@ -26,9 +26,17 @@ pub const SDES_ITEM_TYPE_PRIV: u8 = 8;
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RtcpPacketReader;
 impl traits::ReadPacket for RtcpPacketReader {
-    type Packet = RtcpPacket;
+    type Packet = RtcpCompoundPacket<RtcpPacket>;
     fn read_packet<R: Read>(&mut self, reader: &mut R) -> Result<Self::Packet> {
-        RtcpPacket::read_from(reader)
+        // TODO: optimize
+        let buf = track_try!(reader.read_all_bytes());
+        let mut packets = Vec::new();
+        let reader = &mut &buf[..];
+        while !reader.is_empty() {
+            let packet = track_try!(RtcpPacket::read_from(reader));
+            packets.push(packet);
+        }
+        Ok(RtcpCompoundPacket::new(packets))
     }
     fn supports_type(&self, ty: u8) -> bool {
         match ty {
@@ -45,12 +53,27 @@ impl traits::ReadPacket for RtcpPacketReader {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RtcpPacketWriter;
 impl traits::WritePacket for RtcpPacketWriter {
-    type Packet = RtcpPacket;
+    type Packet = RtcpCompoundPacket<RtcpPacket>;
     fn write_packet<W: Write>(&mut self, writer: &mut W, packet: &Self::Packet) -> Result<()> {
-        packet.write_to(writer)
+        for p in packet.packets.iter() {
+            track_try!(p.write_to(writer));
+        }
+        Ok(())
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RtcpCompoundPacket<T> {
+    pub packets: Vec<T>,
+}
+impl<T> RtcpCompoundPacket<T> {
+    pub fn new(packets: Vec<T>) -> Self {
+        RtcpCompoundPacket { packets: packets }
+    }
+}
+impl<T: Packet> Packet for RtcpCompoundPacket<T> {}
+impl<T: traits::RtcpPacket> traits::RtcpPacket for RtcpCompoundPacket<T> {}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum RtcpPacket {
     Sr(RtcpSenderReport),
@@ -145,6 +168,7 @@ fn read_sctp<R: Read>(reader: &mut R, expected_type: u8) -> Result<(U5, Vec<u8>)
 
     let word_count = track_try!(reader.read_u16be()) as usize;
     let mut payload = track_try!(reader.read_bytes(word_count * 4));
+
     if padding {
         let payload_len = payload.len();
         track_assert_ne!(payload_len, 0, ErrorKind::Invalid);
@@ -431,6 +455,7 @@ impl ReadFrom for SdesChunk {
             }
             let len = track_try!(reader.read_u8()) as usize;
             let text = track_try!(reader.read_string(len));
+
             read_bytes += 1 + len;
             let item = match ty {
                 SDES_ITEM_TYPE_CNAME => SdesItem::Cname(text),

src/rfc3711.rs

@@ -7,7 +7,7 @@ use handy_async::sync_io::{ReadExt, WriteExt};
 use {Result, ErrorKind};
 use io::{ReadFrom, WriteTo};
 use types::U48;
-use traits::ReadPacket;
+use traits::{ReadPacket, RtpPacket, RtcpPacket};
 use rfc3550;
 
 pub type PacketIndex = U48;
@@ -91,9 +91,10 @@ impl SrtpContext {
         let header = track_try!(rfc3550::RtpFixedHeader::read_from(reader));
         let encrypted_portion = &reader[0..reader.len() - self.auth_tag_len];
 
+        let index = ((self.rollover_counter as u64) << 32) + (header.seq_num as u64);
         let iv = BigUint::from_bytes_be(&self.session_salt_key) << 16;
         let iv = iv ^ (BigUint::from(header.ssrc) << 64);
-        let iv = iv ^ (BigUint::from(header.seq_num) << 16);
+        let iv = iv ^ (BigUint::from(index) << 16);
         let iv = &iv.to_bytes_be()[0..self.session_encr_key.len()];
 
         let mut ctr =
@@ -109,10 +110,104 @@ impl SrtpContext {
             let mut output = [0; 16];
             ctr.process(&input[..], &mut output[..]);
 
-            let mut temp_block = Vec::from(block);
+            for (a, b) in block.iter().zip(output.iter()) {
-            temp_block.resize(16, 0);
+                decrypted.push(*a ^ *b);
-            let temp = BigUint::from_bytes_be(&temp_block) ^ BigUint::from_bytes_be(&output);
+            }
-            decrypted.extend(&temp.to_bytes_be()[..block.len()]);
+        }
+
+        Ok(decrypted)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SrtcpContext {
+    // TODO: support other fields
+    pub master_key: Vec<u8>,
+    pub master_salt: Vec<u8>,
+    pub highest_recv_index: PacketIndex, // NOTE: 47-bits
+    pub encryption: EncryptionAlgorithm,
+    pub replay_list: SplaySet<PacketIndex>,
+    pub session_encr_key: Vec<u8>,
+    pub session_salt_key: Vec<u8>,
+    pub session_auth_key: Vec<u8>,
+    pub auth_tag_len: usize,
+}
+impl SrtcpContext {
+    pub fn new(master_key: &[u8], master_salt: &[u8]) -> Self {
+        // TODO: support MKI
+        SrtcpContext {
+            master_key: Vec::from(master_key),
+            master_salt: Vec::from(master_salt),
+            highest_recv_index: 0,
+            encryption: EncryptionAlgorithm::default(),
+            replay_list: SplaySet::new(),
+            session_encr_key: vec![0; 128 / 8],
+            session_salt_key: vec![0; 112 / 8],
+            session_auth_key: vec![0; 160 / 8],
+            auth_tag_len: 80 / 8,
+        }
+    }
+    pub fn update_session_keys(&mut self) {
+        // See: https://tools.ietf.org/html/rfc3711#section-4.3.2
+        let index = BigUint::from(self.highest_recv_index);
+
+        let enc_key_id = BigUint::from_bytes_be(&[3, 0, 0, 0, 0, 0, 0]) + index.clone();
+        let auth_key_id = BigUint::from_bytes_be(&[4, 0, 0, 0, 0, 0, 0]) + index.clone();
+        let salt_key_id = BigUint::from_bytes_be(&[5, 0, 0, 0, 0, 0, 0]) + index.clone();
+        let master_salt = BigUint::from_bytes_be(&self.master_salt);
+
+        self.session_encr_key = prf_n(&self.master_key,
+                                      enc_key_id ^ master_salt.clone(),
+                                      self.session_encr_key.len());
+        self.session_auth_key = prf_n(&self.master_key,
+                                      auth_key_id ^ master_salt.clone(),
+                                      self.session_auth_key.len());
+        self.session_salt_key = prf_n(&self.master_key,
+                                      salt_key_id ^ master_salt.clone(),
+                                      self.session_salt_key.len());
+    }
+    pub fn authenticate(&self, packet: &[u8]) -> Result<()> {
+        let auth_portion = &packet[..packet.len() - self.auth_tag_len];
+        let auth_tag = &packet[packet.len() - self.auth_tag_len..];
+
+        let mut expected_tag = hmac_hash_sha1(&self.session_auth_key, &auth_portion);
+        expected_tag.truncate(self.auth_tag_len);
+        track_assert_eq!(auth_tag, &expected_tag[..], ErrorKind::Invalid);
+        Ok(())
+    }
+    pub fn decrypt(&mut self, packet: &[u8]) -> Result<Vec<u8>> {
+        let index = track_try!((&mut &packet[packet.len() - self.auth_tag_len - 4..]).read_u32be());
+        let is_encrypted = index & 0x8000_0000 != 0;
+        if !is_encrypted {
+            return Ok(Vec::from(&packet[..packet.len() - self.auth_tag_len - 4]));
+        }
+        let index = index & 0x7FFF_FFFF;
+
+        let reader = &mut &packet[..];
+        let _ = track_try!(reader.read_u32be());
+        let ssrc = track_try!(reader.read_u32be());
+        let encrypted_portion = &reader[0..reader.len() - self.auth_tag_len - 4];
+
+        let iv = BigUint::from_bytes_be(&self.session_salt_key) << 16;
+        let iv = iv ^ (BigUint::from(ssrc) << 64);
+        let iv = iv ^ (BigUint::from(index) << 16);
+        let iv = &iv.to_bytes_be()[0..self.session_encr_key.len()];
+
+        let mut ctr =
+            crypto::aes::ctr(crypto::aes::KeySize::KeySize128, &self.session_encr_key, iv);
+        let block_size = self.session_encr_key.len();
+
+        let mut decrypted = Vec::from(&packet[..8]);
+
+        for (i, block) in encrypted_portion.chunks(block_size).enumerate() {
+            let mut input = [0; 16];
+            (&mut input[8..]).write_u64be(i as u64).unwrap();
+            let mut output = [0; 16];
+            ctr.process(&input[..], &mut output[..]);
+
+            for (a, b) in block.iter().zip(output.iter()) {
+                decrypted.push(*a ^ *b);
+            }
         }
 
         Ok(decrypted)
@@ -125,7 +220,8 @@ pub struct SrtpPacketReader<T> {
     inner: T,
 }
 impl<T> SrtpPacketReader<T>
-    where T: ReadPacket
+    where T: ReadPacket,
+          T::Packet: RtpPacket
 {
     pub fn new(mut context: SrtpContext, inner: T) -> Self {
         context.update_session_keys();
@@ -136,7 +232,42 @@ impl<T> SrtpPacketReader<T>
     }
 }
 impl<T> ReadPacket for SrtpPacketReader<T>
-    where T: ReadPacket
+    where T: ReadPacket,
+          T::Packet: RtpPacket
+{
+    type Packet = T::Packet;
+    fn read_packet<R: Read>(&mut self, reader: &mut R) -> Result<Self::Packet> {
+        let packet_bytes = track_try!(reader.read_all_bytes());
+        track_try!(self.context.authenticate(&packet_bytes));
+        let decrypted_packet_bytes = track_try!(self.context.decrypt(&packet_bytes));
+        track_err!(self.inner.read_packet(&mut &decrypted_packet_bytes[..]))
+    }
+
+    fn supports_type(&self, ty: u8) -> bool {
+        self.inner.supports_type(ty)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SrtcpPacketReader<T> {
+    context: SrtcpContext,
+    inner: T,
+}
+impl<T> SrtcpPacketReader<T>
+    where T: ReadPacket,
+          T::Packet: RtcpPacket
+{
+    pub fn new(mut context: SrtcpContext, inner: T) -> Self {
+        context.update_session_keys();
+        SrtcpPacketReader {
+            context: context,
+            inner: inner,
+        }
+    }
+}
+impl<T> ReadPacket for SrtcpPacketReader<T>
+    where T: ReadPacket,
+          T::Packet: RtcpPacket
 {
     type Packet = T::Packet;
     fn read_packet<R: Read>(&mut self, reader: &mut R) -> Result<Self::Packet> {
@@ -183,6 +314,7 @@ fn prf_n(master_key: &[u8], x: BigUint, n: usize) -> Vec<u8> {
 #[cfg(test)]
 mod test {
     use rfc3550;
+    use rfc4585;
     use super::*;
 
     #[test]
@@ -213,4 +345,20 @@ mod test {
         assert_eq!(&packet.payload[..expected_prefix.len()],
                    &expected_prefix[..]);
     }
+
+    #[test]
+    fn rtcp_decryption_works() {
+        let master_key = [254, 123, 44, 240, 174, 252, 53, 54, 2, 213, 123, 106, 85, 165, 5, 13];
+        let master_salt = [77, 202, 202, 112, 81, 101, 219, 232, 143, 131, 160, 89, 15, 141];
+        let packet = [128, 201, 0, 1, 194, 242, 138, 93, 67, 38, 193, 233, 60, 78, 188, 195, 230,
+                      90, 19, 196, 152, 235, 136, 164, 15, 177, 174, 217, 207, 115, 148, 223, 109,
+                      112, 71, 245, 16, 214, 216, 232, 87, 153, 5, 238, 72, 201, 223, 43, 69, 99,
+                      54, 211, 118, 28, 227, 100, 161, 216, 90, 203, 99, 167, 215, 130, 151, 16,
+                      128, 138, 128, 0, 0, 1, 126, 39, 201, 236, 161, 194, 6, 232, 194, 230];
+
+        let context = SrtcpContext::new(&master_key, &master_salt);
+        let mut rtcp_reader = SrtcpPacketReader::new(context, rfc4585::RtcpPacketReader);
+        let packet = track_try_unwrap!(rtcp_reader.read_packet(&mut &packet[..]));
+        println!("# {:?}", packet);
+    }
 }

src/rfc4585.rs

@@ -21,9 +21,17 @@ pub const PSFB_MESSAGE_TYPE_AFB: u8 = 15;
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RtcpPacketReader;
 impl traits::ReadPacket for RtcpPacketReader {
-    type Packet = RtcpPacket;
+    type Packet = rfc3550::RtcpCompoundPacket<RtcpPacket>;
     fn read_packet<R: Read>(&mut self, reader: &mut R) -> Result<Self::Packet> {
-        RtcpPacket::read_from(reader)
+        // TODO: optimize
+        let buf = track_try!(reader.read_all_bytes());
+        let mut packets = Vec::new();
+        let reader = &mut &buf[..];
+        while !reader.is_empty() {
+            let packet = track_try!(RtcpPacket::read_from(reader));
+            packets.push(packet);
+        }
+        Ok(rfc3550::RtcpCompoundPacket::new(packets))
     }
     fn supports_type(&self, ty: u8) -> bool {
         match ty {
@@ -42,9 +50,12 @@ impl traits::ReadPacket for RtcpPacketReader {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RtcpPacketWriter;
 impl traits::WritePacket for RtcpPacketWriter {
-    type Packet = RtcpPacket;
+    type Packet = rfc3550::RtcpCompoundPacket<RtcpPacket>;
     fn write_packet<W: Write>(&mut self, writer: &mut W, packet: &Self::Packet) -> Result<()> {
-        packet.write_to(writer)
+        for p in packet.packets.iter() {
+            track_try!(p.write_to(writer));
+        }
+        Ok(())
     }
 }