zero-out timer

4 files changed, 67 insertions, 30 deletions

diff --git a/src/consts.rs b/src/consts.rs
index 8a3839d..6889ac7 100644
--- a/src/consts.rs
+++ b/src/consts.rs
@@ -6,14 +6,17 @@ use std::time::Duration;
 lazy_static! {
     pub static ref REKEY_ATTEMPT_TIME: Duration = Duration::new(90, 0);
     pub static ref REKEY_AFTER_TIME: Duration = Duration::new(120, 0);
+    pub static ref REKEY_AFTER_TIME_RECV: Duration = *REJECT_AFTER_TIME - *KEEPALIVE_TIMEOUT - *REKEY_TIMEOUT;
     pub static ref REJECT_AFTER_TIME: Duration = Duration::new(180, 0);
-    pub static ref REJECT_ALL_AFTER_TIME: Duration = *REJECT_AFTER_TIME * 3;
+    pub static ref WIPE_AFTER_TIME: Duration = *REJECT_AFTER_TIME * 3;
+
     pub static ref REKEY_TIMEOUT: Duration = Duration::new(5, 0);
     pub static ref KEEPALIVE_TIMEOUT: Duration = Duration::new(10, 0);
     pub static ref STALE_SESSION_TIMEOUT: Duration = *KEEPALIVE_TIMEOUT + *REKEY_TIMEOUT;
-    pub static ref RECV_REKEY_AFTER_TIME: Duration = *REJECT_AFTER_TIME - *KEEPALIVE_TIMEOUT - *REKEY_TIMEOUT;
+
     pub static ref TIMER_RESOLUTION: Duration = Duration::from_millis(100);
     pub static ref COOKIE_REFRESH_TIME: Duration = Duration::new(120, 0);
+
 }
 
 // transport ratcheting message limits, in seconds
diff --git a/src/interface/peer_server.rs b/src/interface/peer_server.rs
index eac343b..c1d6c44 100644
--- a/src/interface/peer_server.rs
+++ b/src/interface/peer_server.rs
@@ -1,5 +1,5 @@
-use consts::{REKEY_TIMEOUT, REKEY_AFTER_TIME, REJECT_AFTER_TIME, REKEY_ATTEMPT_TIME,
-             KEEPALIVE_TIMEOUT, STALE_SESSION_TIMEOUT, MAX_CONTENT_SIZE, TIMER_RESOLUTION};
+use consts::{REKEY_TIMEOUT, REJECT_AFTER_TIME, REKEY_ATTEMPT_TIME, KEEPALIVE_TIMEOUT,
+             STALE_SESSION_TIMEOUT, MAX_CONTENT_SIZE, WIPE_AFTER_TIME};
 use cookie;
 use interface::{SharedPeer, SharedState, UtunPacket, config};
 use peer::{Peer, SessionType};
@@ -213,11 +213,9 @@ impl PeerServer {
         }
         info!("handshake response received, current session now {}", our_index);
 
-        self.timer.spawn_delayed(*KEEPALIVE_TIMEOUT,
-                                 TimerMessage::PassiveKeepAlive(peer_ref.clone(), our_index));
-
-        self.timer.spawn_delayed(*REJECT_AFTER_TIME,
-                                 TimerMessage::Reject(peer_ref.clone(), our_index));
+        self.timer.spawn_delayed(*KEEPALIVE_TIMEOUT, TimerMessage::PassiveKeepAlive(peer_ref.clone(), our_index));
+        self.timer.spawn_delayed(*REJECT_AFTER_TIME, TimerMessage::Reject(peer_ref.clone(), our_index));
+        self.timer.spawn_delayed(*WIPE_AFTER_TIME,   TimerMessage::Wipe(peer_ref.clone()));
 
         match peer.info.keepalive {
             Some(keepalive) if keepalive > 0 => {
@@ -246,6 +244,7 @@ impl PeerServer {
             let mut state = self.shared_state.borrow_mut();
             let (raw_packet, transition) = peer.handle_incoming_transport(addr, packet)?;
 
+            // If a new session has been set to current (TODO make this more clear)
             if let Some(possible_dead_index) = transition {
                 if let Some(index) = possible_dead_index {
                     let _ = state.index_map.remove(&index);
@@ -259,6 +258,11 @@ impl PeerServer {
                         Err(e) => warn!("failed to encrypt packet: {}", e)
                     }
                 }
+
+                let our_new_index = peer.sessions.current.as_ref().unwrap().our_index;
+                self.timer.spawn_delayed(*KEEPALIVE_TIMEOUT, TimerMessage::PassiveKeepAlive(peer_ref.clone(), our_new_index));
+                self.timer.spawn_delayed(*REJECT_AFTER_TIME, TimerMessage::Reject(peer_ref.clone(), our_new_index));
+                self.timer.spawn_delayed(*WIPE_AFTER_TIME,   TimerMessage::Wipe(peer_ref.clone()));
             }
             (raw_packet, peer.needs_new_handshake(false))
         };
@@ -323,22 +327,23 @@ impl PeerServer {
 
         self.send_to_peer((endpoint, init_packet))?;
         peer.last_sent_init = Timestamp::now();
-        let when = *REKEY_TIMEOUT + *TIMER_RESOLUTION * 2;
+        let when = *REKEY_TIMEOUT;
         self.timer.spawn_delayed(when, TimerMessage::Rekey(peer_ref.clone(), new_index));
         Ok(new_index)
     }
 
     fn handle_timer(&mut self, message: TimerMessage) -> Result<(), Error> {
+        use self::TimerMessage::*;
         match message {
-            TimerMessage::Rekey(peer_ref, our_index) => {
+            Rekey(peer_ref, our_index) => {
                 {
                     let mut peer = peer_ref.borrow_mut();
 
                     match peer.find_session(our_index) {
                         Some((_, SessionType::Next)) => {
                             if peer.last_sent_init.elapsed() < *REKEY_TIMEOUT {
-                                let wait = *REKEY_TIMEOUT - peer.last_sent_init.elapsed() + *TIMER_RESOLUTION * 2;
-                                self.timer.spawn_delayed(wait, TimerMessage::Rekey(peer_ref.clone(), our_index));
+                                let wait = *REKEY_TIMEOUT - peer.last_sent_init.elapsed();
+                                self.timer.spawn_delayed(wait, Rekey(peer_ref.clone(), our_index));
                                 bail!("too soon since last init sent, waiting {:?} ({})", wait, our_index);
                             } else if peer.last_tun_queue.elapsed() > *REKEY_ATTEMPT_TIME {
                                 peer.sessions.next = None;
@@ -348,19 +353,19 @@ impl PeerServer {
                         Some((_, SessionType::Current)) => {
                             let since_last_recv = peer.sessions.current.as_ref().unwrap().last_received.elapsed(); // gross
                             if since_last_recv <= *STALE_SESSION_TIMEOUT {
-                                let wait = *STALE_SESSION_TIMEOUT - since_last_recv + *TIMER_RESOLUTION * 2;
-                                self.timer.spawn_delayed(wait, TimerMessage::Rekey(peer_ref.clone(), our_index));
+                                let wait = *STALE_SESSION_TIMEOUT - since_last_recv;
+                                self.timer.spawn_delayed(wait, Rekey(peer_ref.clone(), our_index));
                                 bail!("rekey tick (waiting ~{}s due to stale session check)", wait.as_secs());
                             }
                         },
-                        _ => bail!("index is linked to a dead session, bailing.")
+                        _ => bail!("index is linked to a dead session, bailing ({})", our_index)
                     }
                 }
 
                 let new_index = self.send_handshake_init(&peer_ref)?;
                 debug!("sent handshake init (Rekey timer) ({} -> {})", our_index, new_index);
             },
-            TimerMessage::Reject(peer_ref, our_index) => {
+            Reject(peer_ref, our_index) => {
                 let mut peer  = peer_ref.borrow_mut();
                 let mut state = self.shared_state.borrow_mut();
 
@@ -374,7 +379,7 @@ impl PeerServer {
                 }
                 let _ = state.index_map.remove(&our_index);
             },
-            TimerMessage::PassiveKeepAlive(peer_ref, our_index) => {
+            PassiveKeepAlive(peer_ref, our_index) => {
                 let mut peer = peer_ref.borrow_mut();
                 {
                     let (session, session_type) = peer.find_session(our_index).ok_or_else(|| err_msg("missing session for timer"))?;
@@ -383,15 +388,15 @@ impl PeerServer {
                     let since_last_recv = session.last_received.elapsed();
                     let since_last_send = session.last_sent.elapsed();
                     if since_last_recv < *KEEPALIVE_TIMEOUT {
-                        let wait = *KEEPALIVE_TIMEOUT - since_last_recv + *TIMER_RESOLUTION;
-                        self.timer.spawn_delayed(wait, TimerMessage::PassiveKeepAlive(peer_ref.clone(), our_index));
+                        let wait = *KEEPALIVE_TIMEOUT - since_last_recv;
+                        self.timer.spawn_delayed(wait, PassiveKeepAlive(peer_ref.clone(), our_index));
                         bail!("passive keepalive tick (waiting ~{}s due to last recv time)", wait.as_secs());
                     } else if since_last_send < *KEEPALIVE_TIMEOUT {
-                        let wait = *KEEPALIVE_TIMEOUT - since_last_send + *TIMER_RESOLUTION;
-                        self.timer.spawn_delayed(wait, TimerMessage::PassiveKeepAlive(peer_ref.clone(), our_index));
+                        let wait = *KEEPALIVE_TIMEOUT - since_last_send;
+                        self.timer.spawn_delayed(wait, PassiveKeepAlive(peer_ref.clone(), our_index));
                         bail!("passive keepalive tick (waiting ~{}s due to last send time)", wait.as_secs());
                     } else if session.keepalive_sent {
-                        self.timer.spawn_delayed(*KEEPALIVE_TIMEOUT, TimerMessage::PassiveKeepAlive(peer_ref.clone(), our_index));
+                        self.timer.spawn_delayed(*KEEPALIVE_TIMEOUT, PassiveKeepAlive(peer_ref.clone(), our_index));
                         bail!("passive keepalive already sent (waiting ~{}s to see if session survives)", KEEPALIVE_TIMEOUT.as_secs());
                     } else {
                         session.keepalive_sent = true;
@@ -401,9 +406,9 @@ impl PeerServer {
                 self.send_to_peer(peer.handle_outgoing_transport(&[])?)?;
                 debug!("sent passive keepalive packet ({})", our_index);
 
-                self.timer.spawn_delayed(*KEEPALIVE_TIMEOUT, TimerMessage::PassiveKeepAlive(peer_ref.clone(), our_index));
+                self.timer.spawn_delayed(*KEEPALIVE_TIMEOUT, PassiveKeepAlive(peer_ref.clone(), our_index));
             },
-            TimerMessage::PersistentKeepAlive(peer_ref, our_index) => {
+            PersistentKeepAlive(peer_ref, our_index) => {
                 let mut peer = peer_ref.borrow_mut();
                 {
                     let (_, session_type) = peer.find_session(our_index).ok_or_else(|| err_msg("missing session for timer"))?;
@@ -415,9 +420,21 @@ impl PeerServer {
 
                 if let Some(keepalive) = peer.info.keepalive {
                     self.timer.spawn_delayed(Duration::from_secs(u64::from(keepalive)),
-                                             TimerMessage::PersistentKeepAlive(peer_ref.clone(), our_index));
+                                             PersistentKeepAlive(peer_ref.clone(), our_index));
 
                 }
+            },
+            Wipe(peer_ref) => {
+                let mut peer = peer_ref.borrow_mut();
+                let mut state = self.shared_state.borrow_mut();
+                if peer.last_handshake.elapsed() >= *WIPE_AFTER_TIME {
+                    info!("wiping all old sessions due to staleness timeout for peer {}", peer.info);
+                    for index in peer.sessions.wipe() {
+                        let _ = state.index_map.remove(&index);
+                    }
+                } else {
+                    debug!("skipping wipe timer for since activity has happened since triggered. ({})", peer.info);
+                }
             }
         }
         Ok(())
diff --git a/src/peer.rs b/src/peer.rs
index 5721081..1773d6b 100644
--- a/src/peer.rs
+++ b/src/peer.rs
@@ -1,7 +1,7 @@
 use anti_replay::AntiReplay;
 use byteorder::{ByteOrder, LittleEndian};
 use consts::{TRANSPORT_OVERHEAD, TRANSPORT_HEADER_SIZE, MAX_SEGMENT_SIZE, REKEY_AFTER_MESSAGES,
-             REKEY_AFTER_TIME, RECV_REKEY_AFTER_TIME, REJECT_AFTER_MESSAGES, PADDING_MULTIPLE};
+             REKEY_AFTER_TIME, REKEY_AFTER_TIME_RECV, REJECT_AFTER_MESSAGES, PADDING_MULTIPLE};
 use cookie;
 use failure::{Error, err_msg};
 use interface::UtunPacket;
@@ -112,6 +112,18 @@ pub struct Sessions {
     pub next    : Option<Session>,
 }
 
+impl Sessions {
+    /// Remove all stored sessions from memory, returning all of our indices that were stored
+    /// in order to clear out caches/maps.
+    pub fn wipe(&mut self) -> Vec<u32> {
+        let indices = vec![mem::replace(&mut self.past,    None),
+                           mem::replace(&mut self.current, None),
+                           mem::replace(&mut self.next,    None)];
+
+        indices.into_iter().filter_map(|sesh| sesh.map(|s| s.our_index)).collect()
+    }
+}
+
 impl Display for Peer {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         write!(f, "Peer({})", self.info)
@@ -169,7 +181,7 @@ impl Peer {
             debug!("needs new handshake: sending after REKEY_AFTER_TIME");
             return true;
         }
-        if !sending && self.last_handshake.elapsed() > *RECV_REKEY_AFTER_TIME {
+        if !sending && self.last_handshake.elapsed() > *REKEY_AFTER_TIME_RECV {
             debug!("needs new handshake: receiving after RECV_REKEY_AFTER_TIME");
             return true;
         }
diff --git a/src/timer.rs b/src/timer.rs
index ee544fb..254fa22 100644
--- a/src/timer.rs
+++ b/src/timer.rs
@@ -1,3 +1,4 @@
+use consts::TIMER_RESOLUTION;
 use futures::{Future, Stream, Sink, Poll, unsync};
 use std::time::Duration;
 use tokio_core::reactor::Handle;
@@ -10,6 +11,7 @@ pub enum TimerMessage {
     PassiveKeepAlive(SharedPeer, u32),
     Rekey(SharedPeer, u32),
     Reject(SharedPeer, u32),
+    Wipe(SharedPeer),
 }
 
 pub struct Timer {
@@ -22,13 +24,16 @@ pub struct Timer {
 impl Timer {
     pub fn new(handle: Handle) -> Self {
         let (tx, rx) = unsync::mpsc::channel::<TimerMessage>(1024);
-        let timer = tokio_timer::Timer::default();
+        let timer = tokio_timer::wheel()
+            .tick_duration(*TIMER_RESOLUTION)
+            .num_slots(1 << 14)
+            .build();
         Self { handle, timer, tx, rx }
     }
 
     pub fn spawn_delayed(&mut self, delay: Duration, message: TimerMessage) {
         trace!("queuing timer message {:?}", &message);
-        let timer = self.timer.sleep(delay);
+        let timer = self.timer.sleep(delay + (*TIMER_RESOLUTION * 2));
         let future = timer.and_then({
             let tx = self.tx.clone();
             move |_| {