Handle timeouts during established connection.

I think we still need to ensure that we can timeout during the handshakes.

Author: markasoftware

src/config_cli.rs

@@ -1,7 +1,10 @@
 use std::time::Duration;
 
+use crate::wire_config::{MIN_DEFAULT_TIMEOUT, MIN_DEFAULT_TIMEOUT_PACKETS};
+
 use clap::{Arg, ArgAction, ArgGroup, ArgMatches, Command, value_parser};
 use declarative_enum_dispatch::enum_dispatch;
+use humantime::format_duration;
 
 const DEFAULT_TUN_NAME: &str = "tun-i405";
 // TODO ensure there are no issues when this is greater than the inter-packet interval
@@ -30,9 +33,11 @@ pub(crate) struct WireConfigCli {
     pub(crate) outgoing_packet_length: Option<u64>,
     pub(crate) outgoing_finalize_delta: u64,
     pub(crate) outgoing_interval: WireIntervalCli,
+    pub(crate) client_timeout: Option<Duration>,
     pub(crate) incoming_packet_length: Option<u64>,
     pub(crate) incoming_finalize_delta: u64,
     pub(crate) incoming_interval: WireIntervalCli,
+    pub(crate) server_timeout: Option<Duration>,
 }
 
 impl EzClap for WireConfigCli {
@@ -64,6 +69,10 @@ impl EzClap for WireConfigCli {
                 .default_value(DEFAULT_FINALIZE_DELTA)
                 .value_parser(humantime::parse_duration)
                 .help("How long to finalize the contents of a packet to be sent before actually sending it. If too short, then packets may frequently be delayed and not sent at the intended times. This is most likely to happen under system load, and those delayed packets could indicate to an attacker that your system is under load. Warnings are logged whenever this happens, so increase this value if you see those warnings frequently."),
+            Arg::new("client_timeout")
+                .long("client-timeout")
+                .value_parser(humantime::parse_duration)
+                .help(format!("The client will disconnect if no packet is received from the server within this amount of time. Defaults to max({}, {MIN_DEFAULT_TIMEOUT_PACKETS}*incoming_packet_interval), ie, will disconnect after at least 10 seconds have elapsed and at least {MIN_DEFAULT_TIMEOUT_PACKETS} packets are dropped/delayed.", format_duration(MIN_DEFAULT_TIMEOUT))),
             Arg::new("incoming_packet_length")
                 .long("incoming-packet-length")
                 .visible_alias("down-packet-length")
@@ -91,6 +100,10 @@ impl EzClap for WireConfigCli {
                 // TODO make sure we print warnings for delayed finalization on the server, on the
                 // client (via statistics).
                 .help("See the documentation for --outgoing-finalize-delta"),
+            Arg::new("server_timeout")
+                .long("server-timeout")
+                .value_parser(humantime::parse_duration)
+                .help("See --client-timeout"),
         ]
     }
 
@@ -148,6 +161,7 @@ impl EzClap for WireConfigCli {
                 .as_nanos()
                 .try_into()
                 .unwrap(),
+            client_timeout: matches.get_one::<Duration>("client_timeout").cloned(),
             incoming_packet_length: matches
                 .get_one::<bytesize::ByteSize>("incoming_packet_length")
                 .map(|x| x.as_u64()),
@@ -158,6 +172,7 @@ impl EzClap for WireConfigCli {
                 .as_nanos()
                 .try_into()
                 .unwrap(),
+            server_timeout: matches.get_one::<Duration>("server_timeout").cloned(),
         }
     }
 }

src/core/client.rs

@@ -263,15 +263,16 @@ impl C2SHandshakeSent {
 
     fn send_one_handshake(&mut self, config: &Config, hardware: &mut impl Hardware) -> Result<()> {
         let mut builder = messages::PacketBuilder::new(
-            ip_to_i405_length(config.c2s_wire_config.packet_length, config.peer_address).into(),
+            ip_to_i405_length(config.client_wire_config.packet_length, config.peer_address).into(),
         );
         let c2s_handshake = messages::ClientToServerHandshake {
             protocol_version: PROTOCOL_VERSION,
             oldest_compatible_protocol_version: OLDEST_COMPATIBLE_PROTOCOL_VERSION,
-            s2c_packet_length: config.s2c_wire_config.packet_length,
-            s2c_packet_interval_min: config.s2c_wire_config.packet_interval_min,
-            s2c_packet_interval_max: config.s2c_wire_config.packet_interval_max,
-            s2c_packet_finalize_delta: config.s2c_wire_config.packet_finalize_delta,
+            s2c_packet_length: config.server_wire_config.packet_length,
+            s2c_packet_interval_min: config.server_wire_config.packet_interval_min,
+            s2c_packet_interval_max: config.server_wire_config.packet_interval_max,
+            s2c_packet_finalize_delta: config.server_wire_config.packet_finalize_delta,
+            server_timeout: config.server_wire_config.timeout,
         };
         let did_add =
             builder.try_add_message(&messages::Message::ClientToServerHandshake(c2s_handshake));
@@ -393,7 +394,7 @@ impl ConnectionStateTrait for C2SHandshakeSent {
                         hardware,
                         self.session,
                         config.peer_address,
-                        config.c2s_wire_config.clone(),
+                        config.client_wire_config.clone(),
                     )?,
                 ))
             }
@@ -412,15 +413,46 @@ impl ConnectionStateTrait for C2SHandshakeSent {
     }
 }
 
+impl EstablishedConnection {
+    // stupid naming _s because enum_dispatch won't let us use the same name here and in server
+    fn handle_is_connection_open_c(
+        self,
+        config: &Config,
+        hardware: &mut impl Hardware,
+        is_connection_open: IsConnectionOpen,
+    ) -> Result<ConnectionState> {
+        match is_connection_open {
+            IsConnectionOpen::Yes => Ok(ConnectionState::EstablishedConnection(self)),
+            IsConnectionOpen::TimedOut => {
+                log::warn!(
+                    "Received no packets from server in a while -- returning to NoConnection state"
+                );
+                Ok(ConnectionState::NoConnection(NoConnection::new(
+                    config, hardware,
+                )?))
+            }
+            IsConnectionOpen::TerminatedNormally => {
+                log::info!(
+                    "Client terminated connection normally -- returning to NoConnection state"
+                );
+                Ok(ConnectionState::NoConnection(NoConnection::new(
+                    config, hardware,
+                )?))
+            }
+        }
+    }
+}
+
 impl ConnectionStateTrait for EstablishedConnection {
     fn on_timer(
         mut self,
-        _config: &Config,
+        config: &Config,
         hardware: &mut impl Hardware,
         timer_timestamp: u64,
     ) -> Result<ConnectionState> {
-        EstablishedConnection::on_timer(&mut self, hardware, timer_timestamp)?;
-        Ok(ConnectionState::EstablishedConnection(self))
+        let is_connection_open =
+            EstablishedConnection::on_timer(&mut self, hardware, timer_timestamp)?;
+        return self.handle_is_connection_open_c(config, hardware, is_connection_open);
     }
 
     fn on_read_outgoing_packet(
@@ -440,19 +472,9 @@ impl ConnectionStateTrait for EstablishedConnection {
         hardware: &mut impl Hardware,
         packet: &[u8],
     ) -> Result<ConnectionState> {
-        match EstablishedConnection::on_read_incoming_packet(&mut self, hardware, packet)? {
-            IsConnectionOpen::Yes => Ok(ConnectionState::EstablishedConnection(self)),
-            IsConnectionOpen::No => {
-                log::info!(
-                    "Server informed us that it shut down normally -- will attempt to reconnect, in case it's restarting."
-                );
-                // TODO this should be handled by some global policy on whether to retry or shut
-                // down after errors, rather than always retrying by resetting to NoConnection.
-                return Ok(ConnectionState::NoConnection(NoConnection::new(
-                    config, hardware,
-                )?));
-            }
-        }
+        let is_connection_open =
+            EstablishedConnection::on_read_incoming_packet(&mut self, hardware, packet)?;
+        self.handle_is_connection_open_c(config, hardware, is_connection_open)
     }
 
     fn on_terminate(self, config: &Config, hardware: &mut impl Hardware) -> Result<()> {
@@ -465,8 +487,8 @@ impl ConnectionStateTrait for EstablishedConnection {
 
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub(crate) struct Config {
-    pub(crate) c2s_wire_config: WireConfig,
-    pub(crate) s2c_wire_config: WireConfig,
+    pub(crate) client_wire_config: WireConfig,
+    pub(crate) server_wire_config: WireConfig,
     pub(crate) peer_address: SocketAddr,
     pub(crate) pre_shared_key: Vec<u8>,
 }

src/core/established_connection.rs

@@ -16,28 +16,31 @@ use anyhow::Result;
 pub(crate) struct EstablishedConnection {
     session: dtls::EstablishedSession,
     peer: std::net::SocketAddr,
-    outgoing_wire_config: WireConfig,
+    wire_config: WireConfig,
+    last_incoming_packet_timestamp: u64,
     jitterator: Jitterator,
     outgoing_connection: OutgoingConnection,
     partial_outgoing_packet: messages::WriteCursor<IpPacketBuffer>,
     defragger: Defragger,
 }
 
+// OutgoingConnection is sort of historical cruft, there was once a grand plan for a "scheduled"
+// mode where outbound packets would be read off the TUN at predetermined times also, in order to
+// hide even from userspace applications the details of the tunnel, and there would be both
+// scheduled and unscheduled OutgoingConnection implementations.
 /// The OutgoingConnection is responsible for reading outgoing packets.
 #[derive(Debug)]
-enum OutgoingConnection {
-    Unscheduled {
-        queued_packet: Box<Option<QueuedIpPacket>>,
-        fragmentation_id: u16,
-    },
+struct OutgoingConnection {
+    queued_packet: Box<Option<QueuedIpPacket>>,
+    fragmentation_id: u16,
 }
 
 impl OutgoingConnection {
     fn new(hardware: &mut impl Hardware) -> Self {
         // unless we have a queued packet (which we don't yet), we want the hardware to always know
         // we are ready to read an outgoing packet.
         hardware.read_outgoing_packet();
-        Self::Unscheduled {
+        Self {
             queued_packet: Box::new(None),
             fragmentation_id: 0,
         }
@@ -50,33 +53,26 @@ impl OutgoingConnection {
         hardware: &mut impl Hardware,
         write_cursor: &mut messages::WriteCursor<IpPacketBuffer>,
     ) {
-        match self {
-            OutgoingConnection::Unscheduled {
-                queued_packet,
-                fragmentation_id,
-            } => {
-                // this indirection is me being afraid that we're going to accidentally reallocate
-                // the box if we try to `take` the `queued_packet` directly (it might try to `take`
-                // the `Box` and then `expect` will still work due to deref coercion?).
-                let queued_packet_mut: &mut Option<QueuedIpPacket> = queued_packet;
-                if let Some(old_queued_packet) = std::mem::take(queued_packet_mut) {
-                    let bytes_left = write_cursor.num_bytes_left();
-                    *fragmentation_id = fragmentation_id.wrapping_add(1);
-                    let fragment = old_queued_packet.fragment(bytes_left, *fragmentation_id);
-                    match fragment {
-                        FragmentResult::Done(msg) => {
-                            msg.serialize(write_cursor);
-                            // queue is empty, let's ask for another packet!
-                            hardware.read_outgoing_packet();
-                        }
-                        FragmentResult::Partial(msg, new_queued_packet) => {
-                            msg.serialize(write_cursor);
-                            *queued_packet_mut = Some(new_queued_packet);
-                        }
-                        FragmentResult::MaxLengthTooShort(new_queued_packet) => {
-                            *queued_packet_mut = Some(new_queued_packet);
-                        }
-                    }
+        // this indirection is me being afraid that we're going to accidentally reallocate
+        // the box if we try to `take` the `queued_packet` directly (it might try to `take`
+        // the `Box` and then `expect` will still work due to deref coercion?).
+        let queued_packet_mut: &mut Option<QueuedIpPacket> = &mut self.queued_packet;
+        if let Some(old_queued_packet) = std::mem::take(queued_packet_mut) {
+            let bytes_left = write_cursor.num_bytes_left();
+            self.fragmentation_id = self.fragmentation_id.wrapping_add(1);
+            let fragment = old_queued_packet.fragment(bytes_left, self.fragmentation_id);
+            match fragment {
+                FragmentResult::Done(msg) => {
+                    msg.serialize(write_cursor);
+                    // queue is empty, let's ask for another packet!
+                    hardware.read_outgoing_packet();
+                }
+                FragmentResult::Partial(msg, new_queued_packet) => {
+                    msg.serialize(write_cursor);
+                    *queued_packet_mut = Some(new_queued_packet);
+                }
+                FragmentResult::MaxLengthTooShort(new_queued_packet) => {
+                    *queued_packet_mut = Some(new_queued_packet);
                 }
             }
         }
@@ -89,20 +85,13 @@ impl OutgoingConnection {
         packet: &[u8],
         _recv_timestamp: u64,
     ) {
-        match self {
-            OutgoingConnection::Unscheduled {
-                queued_packet,
-                fragmentation_id: _,
-            } => {
-                // strategy: queue the packet, then dequeue it!
-                assert!(
-                    queued_packet.is_none(),
-                    "We never request to read an outgoing packet while we have a queued packet"
-                );
-                **queued_packet = Some(QueuedIpPacket::new(packet));
-                self.try_to_dequeue(hardware, write_cursor);
-            }
-        }
+        // strategy: queue the packet, then dequeue it!
+        assert!(
+            self.queued_packet.is_none(),
+            "We never request to read an outgoing packet while we have a queued packet"
+        );
+        *self.queued_packet = Some(QueuedIpPacket::new(packet));
+        self.try_to_dequeue(hardware, write_cursor);
     }
 }
 
@@ -128,7 +117,14 @@ impl EstablishedConnection {
                 ip_to_i405_length(outgoing_wire_config.packet_length, peer).into(),
             )),
             defragger: Defragger::new(),
-            outgoing_wire_config,
+            wire_config: outgoing_wire_config,
+            // this is a tiny bit jank in the client case, because the server won't start sending us
+            // packets until it receives our first post-handshake packet. If we have fast incoming
+            // intervals but long roundtrip time, it's possible that quite a few incoming intervals
+            // will elapse before we start receiving anything from the server. We can't just set
+            // this to None though and wait for the first server packet, because the server could
+            // theoretically crash even now!
+            last_incoming_packet_timestamp: hardware.timestamp(),
             jitterator,
         })
     }
@@ -141,14 +137,14 @@ impl EstablishedConnection {
         &mut self,
         hardware: &mut impl Hardware,
         timer_timestamp: u64,
-    ) -> Result<()> {
+    ) -> Result<IsConnectionOpen> {
         // timer means that it's about time to send a packet -- let's finalize the packet and send
         // it to the hardware!
-        let send_timestamp = timer_timestamp + self.outgoing_wire_config.packet_finalize_delta;
+        let send_timestamp = timer_timestamp + self.wire_config.packet_finalize_delta;
         let outgoing_cleartext_packet = std::mem::replace(
             &mut self.partial_outgoing_packet,
             messages::WriteCursor::new(IpPacketBuffer::new_empty(
-                ip_to_i405_length(self.outgoing_wire_config.packet_length, self.peer).into(),
+                ip_to_i405_length(self.wire_config.packet_length, self.peer).into(),
             )),
         )
         .into_inner();
@@ -165,10 +161,14 @@ impl EstablishedConnection {
         hardware.register_interval(next_interval);
         self.outgoing_connection
             .try_to_dequeue(hardware, &mut self.partial_outgoing_packet);
-        hardware.set_timer(
-            send_timestamp + next_interval - self.outgoing_wire_config.packet_finalize_delta,
-        );
-        Ok(())
+        hardware.set_timer(send_timestamp + next_interval - self.wire_config.packet_finalize_delta);
+
+        // check if the incoming connection timed out
+        if hardware.timestamp() > self.last_incoming_packet_timestamp + self.wire_config.timeout {
+            return Ok(IsConnectionOpen::TimedOut);
+        }
+
+        Ok(IsConnectionOpen::Yes)
     }
 
     pub(crate) fn on_read_outgoing_packet<H: Hardware>(
@@ -194,15 +194,20 @@ impl EstablishedConnection {
     ) -> Result<IsConnectionOpen> {
         match self.session.decrypt_datagram(packet) {
             dtls::DecryptResult::Decrypted(cleartext_packet) => {
-                self.on_read_incoming_cleartext_packet(hardware, &cleartext_packet)
+                // notice how we only update the last_incoming_packet_timestamp on a successful
+                // decryption. Otherwise, eg if the client restarts unexpectedly, it might keep
+                // attempting a handshake so we won't time out.
+                self.last_incoming_packet_timestamp = hardware.timestamp();
+                self.on_read_incoming_cleartext_packet(hardware, &cleartext_packet)?;
+                Ok(IsConnectionOpen::Yes)
             }
             dtls::DecryptResult::SendThese(send_these) => {
                 for packet in send_these {
                     hardware.send_outgoing_packet(&packet, self.peer, None)?;
                 }
                 Ok(IsConnectionOpen::Yes)
             }
-            dtls::DecryptResult::Terminated => Ok(IsConnectionOpen::No),
+            dtls::DecryptResult::Terminated => Ok(IsConnectionOpen::TerminatedNormally),
             dtls::DecryptResult::Err(err) => Err(err.into()),
         }
     }
@@ -211,7 +216,7 @@ impl EstablishedConnection {
         &mut self,
         hardware: &mut H,
         packet: &[u8],
-    ) -> Result<IsConnectionOpen> {
+    ) -> Result<()> {
         let mut cursor = messages::ReadCursor::new(packet);
         while messages::has_message(&cursor) {
             let msg = cursor.read()?;
@@ -237,7 +242,7 @@ impl EstablishedConnection {
                 }
             }
         }
-        Ok(IsConnectionOpen::Yes)
+        Ok(())
     }
 
     // When I name this just `on_terminate`, there's a conflict with the name of the same method
@@ -251,5 +256,6 @@ impl EstablishedConnection {
 #[must_use]
 pub(crate) enum IsConnectionOpen {
     Yes,
-    No,
+    TimedOut,
+    TerminatedNormally,
 }