Merge pull request #50 from kedars/feature/dedup

de-dup for RX packets

matter/src/data_model/sdm/noc.rs

@@ -30,7 +30,7 @@ use crate::interaction_model::messages::ib;
 use crate::tlv::{FromTLV, OctetStr, TLVElement, TLVWriter, TagType, ToTLV, UtfStr};
 use crate::transport::session::SessionMode;
 use crate::utils::writebuf::WriteBuf;
-use crate::{cmd_enter, error::*};
+use crate::{cmd_enter, error::*, secure_channel};
 use log::{error, info};
 use num_derive::FromPrimitive;
 
@@ -177,6 +177,15 @@ impl NocCluster {
             return Err(NocStatus::InsufficientPrivlege);
         }
 
+        // This command's processing may take longer, send a stand alone ACK to the peer to avoid any retranmissions
+        let ack_send = secure_channel::common::send_mrp_standalone_ack(
+            cmd_req.trans.exch,
+            cmd_req.trans.session,
+        );
+        if ack_send.is_err() {
+            error!("Error sending Standalone ACK, falling back to piggybacked ACK");
+        }
+
         let r = AddNocReq::from_tlv(&cmd_req.data).map_err(|_| NocStatus::InvalidNOC)?;
 
         let noc_value = Cert::new(r.noc_value.0).map_err(|_| NocStatus::InvalidNOC)?;
@@ -188,7 +197,6 @@ impl NocCluster {
         } else {
             None
         };
-
         let fabric = Fabric::new(
             noc_data.key_pair,
             noc_data.root_ca,

matter/src/error.rs

@@ -29,6 +29,7 @@ pub enum Error {
     BufferTooSmall,
     ClusterNotFound,
     CommandNotFound,
+    Duplicate,
     EndpointNotFound,
     Crypto,
     TLSStack,

matter/src/interaction_model/command.rs

@@ -37,11 +37,11 @@ macro_rules! cmd_enter {
     }};
 }
 
-pub struct CommandReq<'a, 'b, 'c, 'd> {
+pub struct CommandReq<'a, 'b, 'c, 'd, 'e> {
     pub cmd: ib::CmdPath,
     pub data: TLVElement<'a>,
     pub resp: &'a mut TLVWriter<'b, 'c>,
-    pub trans: &'a mut Transaction<'d>,
+    pub trans: &'a mut Transaction<'d, 'e>,
 }
 
 impl InteractionModel {

matter/src/interaction_model/core.rs

@@ -25,7 +25,7 @@ use crate::{
         exchange::Exchange,
         packet::Packet,
         proto_demux::{self, ProtoCtx, ResponseRequired},
-        session::Session,
+        session::SessionHandle,
     },
 };
 use colored::Colorize;
@@ -59,8 +59,8 @@ pub enum OpCode {
     TimedRequest = 10,
 }
 
-impl<'a> Transaction<'a> {
+impl<'a, 'b> Transaction<'a, 'b> {
-    pub fn new(session: &'a mut Session, exch: &'a mut Exchange) -> Self {
+    pub fn new(session: &'a mut SessionHandle<'b>, exch: &'a mut Exchange) -> Self {
         Self {
             state: TransactionState::Ongoing,
             session,

matter/src/interaction_model/mod.rs

@@ -18,7 +18,7 @@
 use crate::{
     error::Error,
     tlv::TLVWriter,
-    transport::{exchange::Exchange, proto_demux::ResponseRequired, session::Session},
+    transport::{exchange::Exchange, proto_demux::ResponseRequired, session::SessionHandle},
 };
 
 use self::{
@@ -32,9 +32,9 @@ pub enum TransactionState {
     Complete,
     Terminate,
 }
-pub struct Transaction<'a> {
+pub struct Transaction<'a, 'b> {
     pub state: TransactionState,
-    pub session: &'a mut Session,
+    pub session: &'a mut SessionHandle<'b>,
     pub exch: &'a mut Exchange,
 }
 

matter/src/secure_channel/common.rs

@@ -15,9 +15,18 @@
  *    limitations under the License.
  */
 
+use boxslab::Slab;
+use log::info;
 use num_derive::FromPrimitive;
 
-use crate::{error::Error, transport::packet::Packet};
+use crate::{
+    error::Error,
+    transport::{
+        exchange::Exchange,
+        packet::{Packet, PacketPool},
+        session::SessionHandle,
+    },
+};
 
 use super::status_report::{create_status_report, GeneralCode};
 
@@ -83,3 +92,10 @@ pub fn create_mrp_standalone_ack(proto_tx: &mut Packet) {
     proto_tx.set_proto_opcode(OpCode::MRPStandAloneAck as u8);
     proto_tx.unset_reliable();
 }
+
+pub fn send_mrp_standalone_ack(exch: &mut Exchange, sess: &mut SessionHandle) -> Result<(), Error> {
+    info!("Sending standalone ACK");
+    let mut ack_packet = Slab::<PacketPool>::try_new(Packet::new_tx()?).ok_or(Error::NoMemory)?;
+    create_mrp_standalone_ack(&mut ack_packet);
+    exch.send(ack_packet, sess)
+}

matter/src/transport/dedup.rs

@@ -0,0 +1,206 @@
+/*
+ *
+ *    Copyright (c) 2023 Project CHIP Authors
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+const MSG_RX_STATE_BITMAP_LEN: u32 = 16;
+
+#[derive(Debug)]
+pub struct RxCtrState {
+    max_ctr: u32,
+    ctr_bitmap: u16,
+}
+
+impl RxCtrState {
+    pub fn new(max_ctr: u32) -> Self {
+        Self {
+            max_ctr,
+            ctr_bitmap: 0xffff,
+        }
+    }
+
+    fn contains(&self, bit_number: u32) -> bool {
+        (self.ctr_bitmap & (1 << bit_number)) != 0
+    }
+
+    fn insert(&mut self, bit_number: u32) {
+        self.ctr_bitmap |= 1 << bit_number;
+    }
+
+    /// Receive a message and update Rx State accordingly
+    /// Returns a bool indicating whether the message is a duplicate
+    pub fn recv(&mut self, msg_ctr: u32, is_encrypted: bool) -> bool {
+        let idiff = (msg_ctr as i32) - (self.max_ctr as i32);
+        let udiff = idiff.unsigned_abs();
+
+        if msg_ctr == self.max_ctr {
+            // Duplicate
+            true
+        } else if (-(MSG_RX_STATE_BITMAP_LEN as i32)..0).contains(&idiff) {
+            // In Rx Bitmap
+            let index = udiff - 1;
+            if self.contains(index) {
+                // Duplicate
+                true
+            } else {
+                self.insert(index);
+                false
+            }
+        }
+        // Now the leftover cases are the new counter is outside of the bitmap as well as max_ctr
+        // in either direction. Encrypted only allows in forward direction
+        else if msg_ctr > self.max_ctr {
+            self.max_ctr = msg_ctr;
+            if udiff < MSG_RX_STATE_BITMAP_LEN {
+                // The previous max_ctr is now the actual counter
+                self.ctr_bitmap <<= udiff;
+                self.insert(udiff - 1);
+            } else {
+                self.ctr_bitmap = 0xffff;
+            }
+            false
+        } else if !is_encrypted {
+            // This is the case where the peer possibly rebooted and chose a different
+            // random counter
+            self.max_ctr = msg_ctr;
+            self.ctr_bitmap = 0xffff;
+            false
+        } else {
+            true
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use super::RxCtrState;
+
+    const ENCRYPTED: bool = true;
+    const NOT_ENCRYPTED: bool = false;
+
+    fn assert_ndup(b: bool) {
+        assert!(!b);
+    }
+
+    fn assert_dup(b: bool) {
+        assert!(b);
+    }
+
+    #[test]
+    fn new_msg_ctr() {
+        let mut s = RxCtrState::new(101);
+
+        assert_ndup(s.recv(103, ENCRYPTED));
+        assert_ndup(s.recv(104, ENCRYPTED));
+        assert_ndup(s.recv(106, ENCRYPTED));
+        assert_eq!(s.max_ctr, 106);
+        assert_eq!(s.ctr_bitmap, 0b1111_1111_1111_0110);
+
+        assert_ndup(s.recv(118, NOT_ENCRYPTED));
+        assert_eq!(s.ctr_bitmap, 0b0110_1000_0000_0000);
+        assert_ndup(s.recv(119, NOT_ENCRYPTED));
+        assert_ndup(s.recv(121, NOT_ENCRYPTED));
+        assert_eq!(s.ctr_bitmap, 0b0100_0000_0000_0110);
+    }
+
+    #[test]
+    fn dup_max_ctr() {
+        let mut s = RxCtrState::new(101);
+
+        assert_ndup(s.recv(103, ENCRYPTED));
+        assert_dup(s.recv(103, ENCRYPTED));
+        assert_dup(s.recv(103, NOT_ENCRYPTED));
+
+        assert_eq!(s.max_ctr, 103);
+        assert_eq!(s.ctr_bitmap, 0b1111_1111_1111_1110);
+    }
+
+    #[test]
+    fn dup_in_rx_bitmap() {
+        let mut ctr = 101;
+        let mut s = RxCtrState::new(101);
+        for _ in 1..8 {
+            ctr += 2;
+            assert_ndup(s.recv(ctr, ENCRYPTED));
+        }
+        assert_ndup(s.recv(116, ENCRYPTED));
+        assert_ndup(s.recv(117, ENCRYPTED));
+        assert_eq!(s.max_ctr, 117);
+        assert_eq!(s.ctr_bitmap, 0b1010_1010_1010_1011);
+
+        // duplicate on the left corner
+        assert_dup(s.recv(101, ENCRYPTED));
+        assert_dup(s.recv(101, NOT_ENCRYPTED));
+
+        // duplicate on the right corner
+        assert_dup(s.recv(116, ENCRYPTED));
+        assert_dup(s.recv(116, NOT_ENCRYPTED));
+
+        // valid insert
+        assert_ndup(s.recv(102, ENCRYPTED));
+        assert_dup(s.recv(102, ENCRYPTED));
+        assert_eq!(s.ctr_bitmap, 0b1110_1010_1010_1011);
+    }
+
+    #[test]
+    fn valid_corners_in_rx_bitmap() {
+        let mut ctr = 102;
+        let mut s = RxCtrState::new(101);
+        for _ in 1..9 {
+            ctr += 2;
+            assert_ndup(s.recv(ctr, ENCRYPTED));
+        }
+        assert_eq!(s.max_ctr, 118);
+        assert_eq!(s.ctr_bitmap, 0b0010_1010_1010_1010);
+
+        // valid insert on the left corner
+        assert_ndup(s.recv(102, ENCRYPTED));
+        assert_eq!(s.ctr_bitmap, 0b1010_1010_1010_1010);
+
+        // valid insert on the right corner
+        assert_ndup(s.recv(117, ENCRYPTED));
+        assert_eq!(s.ctr_bitmap, 0b1010_1010_1010_1011);
+    }
+
+    #[test]
+    fn encrypted_wraparound() {
+        let mut s = RxCtrState::new(65534);
+
+        assert_ndup(s.recv(65535, ENCRYPTED));
+        assert_ndup(s.recv(65536, ENCRYPTED));
+        assert_dup(s.recv(0, ENCRYPTED));
+    }
+
+    #[test]
+    fn unencrypted_wraparound() {
+        let mut s = RxCtrState::new(65534);
+
+        assert_ndup(s.recv(65536, NOT_ENCRYPTED));
+        assert_ndup(s.recv(0, NOT_ENCRYPTED));
+    }
+
+    #[test]
+    fn unencrypted_device_reboot() {
+        println!("Sub 65532 is {:?}", 1_u16.overflowing_sub(65532));
+        println!("Sub 65535 is {:?}", 1_u16.overflowing_sub(65535));
+        println!("Sub 11-13 is {:?}", 11_u32.wrapping_sub(13_u32) as i32);
+        println!("Sub regular is {:?}", 2000_u16.overflowing_sub(1998));
+        let mut s = RxCtrState::new(20010);
+
+        assert_ndup(s.recv(20011, NOT_ENCRYPTED));
+        assert_ndup(s.recv(0, NOT_ENCRYPTED));
+    }
+}

matter/src/transport/exchange.rs

@@ -175,7 +175,7 @@ impl Exchange {
         }
     }
 
-    fn send(
+    pub fn send(
         &mut self,
         mut proto_tx: BoxSlab<PacketPool>,
         session: &mut SessionHandle,

matter/src/transport/mod.rs

@@ -15,6 +15,7 @@
  *    limitations under the License.
  */
 
+mod dedup;
 pub mod exchange;
 pub mod mgr;
 pub mod mrp;

matter/src/transport/session.rs

@@ -33,6 +33,7 @@ use log::{info, trace};
 use rand::Rng;
 
 use super::{
+    dedup::RxCtrState,
     network::{Address, NetworkInterface},
     packet::{Packet, PacketPool},
 };
@@ -84,6 +85,7 @@ pub struct Session {
     local_sess_id: u16,
     peer_sess_id: u16,
     msg_ctr: u32,
+    rx_ctr_state: RxCtrState,
     mode: SessionMode,
     data: Option<Box<dyn Any>>,
     last_use: SystemTime,
@@ -138,6 +140,7 @@ impl Session {
             peer_sess_id: 0,
             local_sess_id: 0,
             msg_ctr: rand::thread_rng().gen_range(0..MATTER_MSG_CTR_RANGE),
+            rx_ctr_state: RxCtrState::new(0),
             mode: SessionMode::PlainText,
             data: None,
             last_use: SystemTime::now(),
@@ -156,6 +159,7 @@ impl Session {
             local_sess_id: clone_from.local_sess_id,
             peer_sess_id: clone_from.peer_sess_id,
             msg_ctr: rand::thread_rng().gen_range(0..MATTER_MSG_CTR_RANGE),
+            rx_ctr_state: RxCtrState::new(0),
             mode: clone_from.mode,
             data: None,
             last_use: SystemTime::now(),
@@ -481,7 +485,17 @@ impl SessionMgr {
             rx.plain.get_src_u64(),
             rx.plain.is_encrypted(),
         ) {
-            Ok(s) => Some(s),
+            Ok(s) => {
+                let session = self.sessions[s].as_mut().unwrap();
+                let is_encrypted = session.is_encrypted();
+                let duplicate = session.rx_ctr_state.recv(rx.plain.ctr, is_encrypted);
+                if duplicate {
+                    info!("Dropping duplicate packet");
+                    return Err(Error::Duplicate);
+                } else {
+                    Some(s)
+                }
+            }
             Err(Error::NoSpace) => None,
             Err(e) => {
                 return Err(e);
@@ -508,6 +522,7 @@ impl SessionMgr {
 
         // Get session
         let sess_handle = self.post_recv(&rx)?;
+
         Ok((rx, sess_handle))
     }