diff --git a/CHANGELOG.md b/CHANGELOG.md index 925d989d..04d4a75a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ # Unreleased + * Start handshake clocks at ClientHello and honor timing settings across DTLS 1.2/1.3 and Auto #161 + # 0.7.3 * Fix DTLS 1.2 ClientHello retransmissions #160 diff --git a/README.md b/README.md index cbc924a3..9b29bab5 100644 --- a/README.md +++ b/README.md @@ -102,12 +102,15 @@ use std::time::Instant; use dimpl::{certificate, Config, Dtls, Output}; // Stub I/O to keep the example focused on the state machine -enum Event { Udp(Vec), Timer(Instant) } -fn wait_next_event(_next_wake: Option) -> Event { Event::Udp(Vec::new()) } +enum Event { Udp(Vec, Instant), Timer(Instant) } +fn wait_next_event(_next_wake: Option) -> Event { + Event::Udp(Vec::new(), Instant::now()) +} fn send_udp(_bytes: &[u8]) {} fn example_event_loop(mut dtls: Dtls) -> Result<(), dimpl::Error> { let mut next_wake: Option = None; + let mut received_packet: Option> = None; loop { // Drain engine output until we have to wait for I/O or a timer let mut out_buf = vec![0u8; 2048]; @@ -138,9 +141,17 @@ fn example_event_loop(mut dtls: Dtls) -> Result<(), dimpl::Error> { } } + if let Some(packet) = received_packet.take() { + dtls.handle_packet(&packet)?; + continue; + } + // Block waiting for either UDP input or the scheduled timeout match wait_next_event(next_wake) { - Event::Udp(pkt) => dtls.handle_packet(&pkt)?, + Event::Udp(pkt, now) => { + dtls.handle_timeout(now)?; + received_packet = Some(pkt); + } Event::Timer(now) => dtls.handle_timeout(now)?, } } diff --git a/src/auto.rs b/src/auto.rs index c32219cc..04fd3a9c 100644 --- a/src/auto.rs +++ b/src/auto.rs @@ -16,7 +16,7 @@ /// and falls back to DTLS 1.2 via [`Error::Dtls12Fallback`] if the /// reassembled ClientHello does not offer DTLS 1.3. use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::Instant; use arrayvec::ArrayVec; @@ -28,6 +28,7 @@ use crate::dtls13::message::Random; use crate::dtls13::message::SignatureAlgorithmsExtension; use crate::dtls13::message::SupportedGroupsExtension; use crate::dtls13::message::UseSrtpExtension; +use crate::timer::HandshakeTimers; use crate::types::NamedGroup; use crate::{Config, CryptoError, DtlsCertificate, Error, Output, SeededRng, TimeoutError}; // Extension type constants @@ -265,10 +266,8 @@ pub(crate) struct ClientPending { needs_send: bool, /// Last time handle_timeout was called. last_now: Instant, - /// When to retransmit the wire_packet. - retransmit_at: Option, - /// How many retransmits have occurred. - retransmit_count: usize, + timers: HandshakeTimers, + rng: SeededRng, } impl ClientPending { @@ -279,6 +278,9 @@ impl ClientPending { ) -> Result { let hybrid = HybridClientHello::new(&config)?; let wire_packet = hybrid.wire_packet(); + let mut rng = SeededRng::new(config.rng_seed()); + let mut timers = HandshakeTimers::new(&config, &mut rng); + timers.begin_flight(&mut rng); Ok(ClientPending { hybrid, config, @@ -286,30 +288,24 @@ impl ClientPending { wire_packet, needs_send: true, last_now: now, - retransmit_at: None, - retransmit_count: 0, + timers, + rng, }) } pub fn handle_timeout(&mut self, now: Instant) -> Result<(), Error> { self.last_now = now; - // Arm initial retransmit timer on first call - if self.retransmit_at.is_none() { - self.retransmit_at = Some(now + Duration::from_secs(1)); - return Ok(()); - } - if let Some(deadline) = self.retransmit_at { - if now >= deadline { - if self.retransmit_count >= self.config.flight_retries() { - return Err(Error::Timeout(TimeoutError::HybridClientHello)); - } - self.retransmit_count += 1; - self.needs_send = true; - // Exponential backoff: 2s, 4s, 8s, ... - let shift = self.retransmit_count.min(5) as u32; - let rto = Duration::from_secs(1u64 << shift); - self.retransmit_at = Some(now + rto); - } + let resend = self + .timers + .handle_timeout(now, &mut self.rng) + .map_err(|error| { + Error::Timeout(match error { + TimeoutError::Handshake => TimeoutError::HybridClientHello, + other => other, + }) + })?; + if resend { + self.needs_send = true; } Ok(()) } @@ -323,16 +319,30 @@ impl ClientPending { } self.needs_send = false; buf[..len].copy_from_slice(&self.wire_packet); + self.timers.start_handshake(self.last_now); + self.timers.flight_sent(self.last_now); return Output::Packet(&buf[..len]); } - let next = self - .retransmit_at - .unwrap_or(self.last_now + Duration::from_secs(1)); + let next = self.timers.poll_timeout(self.last_now); Output::Timeout(next) } - pub fn into_parts(self) -> (HybridClientHello, Arc, DtlsCertificate, Instant) { - (self.hybrid, self.config, self.certificate, self.last_now) + pub fn into_parts( + self, + ) -> ( + HybridClientHello, + Arc, + DtlsCertificate, + Instant, + HandshakeTimers, + ) { + ( + self.hybrid, + self.config, + self.certificate, + self.last_now, + self.timers, + ) } } @@ -450,6 +460,9 @@ fn server_hello_version_inner(packet: &[u8]) -> Option { #[cfg(test)] mod tests { + #[cfg(feature = "rcgen")] + use std::time::Duration; + use super::*; use crate::PskResolver; use crate::dtls12::message::Dtls12CipherSuite; @@ -481,6 +494,75 @@ mod tests { } } + #[test] + #[cfg(feature = "rcgen")] + fn timing_expired_deadline_is_fatal_during_either_client_handoff() { + use crate::certificate::generate_self_signed_certificate; + use crate::{Dtls, Inner}; + + let now = Instant::now(); + let budget = Duration::from_millis(10); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .handshake_timeout(budget) + .build() + .expect("valid config"), + ); + let certificate = generate_self_signed_certificate().expect("certificate"); + for dtls13 in [false, true] { + let mut pending = ClientPending::new(config.clone(), certificate.clone(), now) + .expect("pending client"); + let mut buffer = [0; 2048]; + let Output::Packet(hello) = pending.poll_output(&mut buffer) else { + panic!("expected hybrid ClientHello"); + }; + let hello = hello.to_vec(); + assert!(matches!( + pending.poll_output(&mut buffer), + Output::Timeout(_) + )); + let mut server = if dtls13 { + Dtls::new_13(config.clone(), certificate.clone(), now) + } else { + Dtls::new_12(config.clone(), certificate.clone(), now) + }; + server.handle_timeout(now).expect("server clock"); + assert!(matches!( + server.poll_output(&mut buffer), + Output::Timeout(_) + )); + server.handle_packet(&hello).expect("accept ClientHello"); + let mut response = None; + loop { + match server.poll_output(&mut buffer) { + Output::Packet(packet) => response = Some(packet.to_vec()), + Output::Timeout(_) => break, + Output::BufferTooSmall { .. } => { + panic!("unexpected server output: buffer too small") + } + Output::Connected => panic!("unexpected server output: connected"), + Output::PeerCert(_) => panic!("unexpected server output: peer certificate"), + Output::KeyingMaterial(_, _) => { + panic!("unexpected server output: keying material") + } + Output::ApplicationData(_) => { + panic!("unexpected server output: application data") + } + Output::CloseNotify => panic!("unexpected server output: close notify"), + } + } + pending.last_now = now + budget; + let mut client = Dtls { + inner: Some(Inner::ClientPending(pending)), + }; + assert_eq!( + client.handle_packet(&response.expect("server response")), + Err(Error::Timeout(TimeoutError::Connect)) + ); + } + } + #[test] fn hello_verify_request_is_dtls12() { // Minimal HelloVerifyRequest packet diff --git a/src/config.rs b/src/config.rs index b84dc122..af84cc3d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -9,6 +9,8 @@ use crate::dtls12::message::Dtls12CipherSuite; use crate::types::{Dtls13CipherSuite, NamedGroup}; use crate::{ConfigError, Error}; +const MAX_TIMING_DURATION: Duration = Duration::from_secs(10 * 365 * 24 * 60 * 60); + /// Callback for resolving PSK identities to shared secrets. /// /// Implement this trait and provide it via [`ConfigBuilder::with_psk_client`] @@ -148,20 +150,23 @@ impl Config { /// Time of first retry. /// - /// Every flight restarts with this value. - /// Doubled for every retry with a ±25% jitter. + /// See [`ConfigBuilder::flight_start_rto`] for the backoff policy. #[inline(always)] pub fn flight_start_rto(&self) -> Duration { self.flight_start_rto } /// Max number of retries per flight. + /// + /// Excludes the initial transmission. Also applies to DTLS 1.3 KeyUpdate. #[inline(always)] pub fn flight_retries(&self) -> usize { self.flight_retries } /// Timeout for the entire handshake, regardless of flights. + /// + /// See [`ConfigBuilder::handshake_timeout`] for the clock-start events. #[inline(always)] pub fn handshake_timeout(&self) -> Duration { self.handshake_timeout @@ -362,9 +367,13 @@ impl ConfigBuilder { /// Set the time of first retry. /// - /// Every flight restarts with this value. - /// Doubled for every retry with a ±25% jitter. - /// Defaults to 1 second. + /// Each flight starts its timer when its first packet is emitted. The base + /// interval doubles after every retry, with independent +/-25% jitter on + /// each interval. This is proportional even for sub-second values; the + /// minimum jittered interval is 1 nanosecond. Backoff arithmetic saturates + /// on overflow. Also governs DTLS 1.3 KeyUpdate flights. + /// + /// Defaults to 1 second. Must be nonzero and at most ten years (3650 days). pub fn flight_start_rto(mut self, rto: Duration) -> Self { self.flight_start_rto = rto; self @@ -372,7 +381,9 @@ impl ConfigBuilder { /// Set the max number of retries per flight. /// - /// Defaults to 4. + /// Excludes the initial transmission: zero disables retransmissions. + /// Timer-driven and duplicate-triggered retransmissions share this budget. + /// Also governs DTLS 1.3 KeyUpdate flights. Defaults to 4. pub fn flight_retries(mut self, retries: usize) -> Self { self.flight_retries = retries; self @@ -380,7 +391,20 @@ impl ConfigBuilder { /// Set the timeout for the entire handshake, regardless of flights. /// - /// Defaults to 40 seconds. + /// Starts when the client emits its first ClientHello packet through + /// [`crate::Output::Packet`], or when the server accepts its first + /// ClientHello fragment. Construction and idle time do not consume this + /// budget. Uses the logical time supplied to [`crate::Dtls::handle_timeout`]. + /// + /// The absolute deadline survives Auto version selection, cookie exchanges, + /// later flights, and retransmissions. It is disabled after completion and + /// does not apply to application traffic or DTLS 1.3 KeyUpdate. + /// + /// Retry exhaustion may fail earlier; increasing this timeout does not + /// increase [`Self::flight_retries`]. With the defaults, an unanswered flight + /// exhausts retries after roughly 31 seconds, before the 40-second deadline. + /// + /// Defaults to 40 seconds. Must be nonzero and at most ten years (3650 days). pub fn handshake_timeout(mut self, timeout: Duration) -> Self { self.handshake_timeout = timeout; self @@ -476,8 +500,8 @@ impl ConfigBuilder { /// Build the configuration. /// - /// This validates the crypto provider before returning the configuration. - /// Returns `Error::ConfigError` if the provider is invalid. + /// Validates timing settings and the crypto provider before returning the + /// configuration. Returns `Error::ConfigError` for invalid settings. /// /// The crypto provider is selected in the following priority order: /// 1. Explicit provider set via `with_crypto_provider()` @@ -486,6 +510,13 @@ impl ConfigBuilder { /// 4. RustCrypto provider (if `rust-crypto` feature is enabled) /// 5. Panic if no provider is available pub fn build(self) -> Result { + if self.handshake_timeout.is_zero() || self.handshake_timeout > MAX_TIMING_DURATION { + return Err(ConfigError::InvalidHandshakeTimeout.into()); + } + if self.flight_start_rto.is_zero() || self.flight_start_rto > MAX_TIMING_DURATION { + return Err(ConfigError::InvalidFlightStartRto.into()); + } + let crypto_provider = self .crypto_provider .or_else(|| CryptoProvider::get_default().cloned()); @@ -693,6 +724,61 @@ impl fmt::Debug for ConfigBuilder { mod tests { use super::*; + #[test] + fn timing_defaults() { + let config = Config::default(); + assert_eq!(config.handshake_timeout(), Duration::from_secs(40)); + assert_eq!(config.flight_start_rto(), Duration::from_secs(1)); + assert_eq!(config.flight_retries(), 4); + } + + #[test] + fn timing_duration_bounds() { + for duration in [ + Duration::ZERO, + MAX_TIMING_DURATION + Duration::from_nanos(1), + Duration::MAX, + ] { + let error = Config::builder() + .handshake_timeout(duration) + .build() + .expect_err("unsupported handshake timeout"); + assert_eq!( + error, + Error::ConfigError(ConfigError::InvalidHandshakeTimeout) + ); + assert_eq!( + error.to_string(), + "config error: handshake_timeout must be nonzero and at most 3650 days" + ); + let error = Config::builder() + .flight_start_rto(duration) + .build() + .expect_err("unsupported initial RTO"); + assert_eq!( + error, + Error::ConfigError(ConfigError::InvalidFlightStartRto) + ); + assert_eq!( + error.to_string(), + "config error: flight_start_rto must be nonzero and at most 3650 days" + ); + } + for duration in [Duration::from_nanos(1), MAX_TIMING_DURATION] { + for retries in [0, usize::MAX] { + let config = Config::builder() + .handshake_timeout(duration) + .flight_start_rto(duration) + .flight_retries(retries) + .build() + .expect("supported timing boundaries"); + assert_eq!(config.handshake_timeout(), duration); + assert_eq!(config.flight_start_rto(), duration); + assert_eq!(config.flight_retries(), retries); + } + } + } + #[test] fn rejects_zero_mtu() { match Config::builder().mtu(0).build() { diff --git a/src/crypto/provider.rs b/src/crypto/provider.rs index a4f0af08..a1a4c15d 100644 --- a/src/crypto/provider.rs +++ b/src/crypto/provider.rs @@ -146,11 +146,13 @@ use std::fmt::Debug; use std::panic::{RefUnwindSafe, UnwindSafe}; use std::sync::OnceLock; +#[cfg(feature = "_crypto-common")] +use crate::CertificateError; +use crate::CryptoError; use crate::buffer::{Buf, TmpBuf}; use crate::crypto::{Aad, Nonce}; use crate::dtls12::message::Dtls12CipherSuite; use crate::types::{Dtls13CipherSuite, HashAlgorithm, NamedGroup, SignatureAlgorithm}; -use crate::{CertificateError, CryptoError}; /// OID for the P-256 elliptic curve (secp256r1 / prime256v1). #[cfg(feature = "_crypto-common")] @@ -626,7 +628,7 @@ impl CryptoProvider { } } -#[cfg(test)] +#[cfg(all(test, feature = "rcgen"))] mod tests { use super::*; diff --git a/src/dtls12/client.rs b/src/dtls12/client.rs index 4ec92a60..b5363e36 100644 --- a/src/dtls12/client.rs +++ b/src/dtls12/client.rs @@ -29,6 +29,7 @@ use crate::dtls12::message::{CompressionMethod, ContentType, Cookie}; use crate::dtls12::message::{DigitallySigned, Dtls12CipherSuite}; use crate::dtls12::message::{ExtensionType, KeyExchangeAlgorithm, MessageType, ProtocolVersion}; use crate::dtls12::message::{Random, SessionId, SignatureAndHashAlgorithm, UseSrtpExtension}; +use crate::timer::HandshakeTimers; use crate::{Config, DtlsCertificate, Error, InternalError, KeyingMaterial, Output}; /// DTLS client @@ -123,10 +124,11 @@ impl Client { /// clears the transcript anyway, so the injected bytes are harmless. pub(crate) fn new_from_hybrid( random: Random, - handshake_fragment: &[u8], + handshake_fragment: Buf, config: std::sync::Arc, certificate: DtlsCertificate, now: Instant, + timers: HandshakeTimers, ) -> Result { assert!( !certificate.certificate.is_empty(), @@ -147,15 +149,7 @@ impl Client { }; let mut engine = Engine::new(config, auth); engine.set_client(true); - // The hybrid ClientHello was sent with message_seq=0 outside this - // engine. Advance the counter so the with-cookie CH gets message_seq=1 - // per RFC 6347 §4.2.2. - engine.set_next_handshake_seq_no(1); - // Inject the hybrid CH into the transcript so it matches the server's - // transcript when the server skips HelloVerifyRequest. - engine.transcript.extend_from_slice(handshake_fragment); - // Advance epoch-0 record sequence past the hybrid CH record. - engine.advance_epoch_0_sequence(); + engine.inject_hybrid_client_hello(handshake_fragment, timers); let mut client = Client { state: State::AwaitHelloVerifyRequest, @@ -199,7 +193,7 @@ impl Client { pub fn handle_packet(&mut self, packet: &[u8]) -> Result<(), Error> { match self .engine - .parse_packet(packet) + .handle_packet(packet, self.last_now) .and_then(|_| self.make_progress()) { Ok(()) => Ok(()), @@ -208,6 +202,9 @@ impl Client { } pub fn poll_output<'a>(&mut self, buf: &'a mut [u8]) -> Output<'a> { + if self.state == State::SendClientHello { + return Output::Timeout(self.last_now); + } if let Some(event) = self.local_events.pop_front() { return event.into_output(buf, &self.server_certificates); } diff --git a/src/dtls12/engine.rs b/src/dtls12/engine.rs index 9310eb7f..1e478430 100644 --- a/src/dtls12/engine.rs +++ b/src/dtls12/engine.rs @@ -1,17 +1,19 @@ use std::mem; use std::sync::Arc; use std::sync::atomic::AtomicBool; -use std::time::{Duration, Instant}; +use std::time::Instant; use super::queue::{QueueRx, QueueTx}; use crate::buffer::{Buf, BufferPool, TmpBuf}; use crate::crypto::{Aad, Iv, Nonce}; use crate::dtls12::context::{AuthMode, CryptoContext}; use crate::dtls12::incoming::{Incoming, Record, RecordHandler}; -use crate::dtls12::message::{Body, HashAlgorithm, Header, MessageType, ProtocolVersion, Sequence}; -use crate::dtls12::message::{ContentType, DTLSRecord, Dtls12CipherSuite, Handshake}; +use crate::dtls12::message::{Body, ClientHello}; +use crate::dtls12::message::{ContentType, DTLSRecord, Dtls12CipherSuite, Handshake, Header}; +use crate::dtls12::message::{HashAlgorithm, MessageType, ProtocolVersion, Sequence}; use crate::error::bounded_error_len; -use crate::timer::ExponentialBackoff; +use crate::timer::{HandshakeTimers, Timeout}; +use crate::util::accepts_client_hello_fragment; use crate::window::ReplayWindow; use crate::{Config, Error, InternalError, Output, SeededRng}; @@ -83,14 +85,7 @@ pub struct Engine { /// The records that have been sent in the current flight. flight_saved_records: Vec, - /// Flight backoff - flight_backoff: ExponentialBackoff, - - /// Timeout for the current flight - flight_timeout: Timeout, - - /// Global timeout for the entire connect operation. - connect_timeout: Timeout, + timers: HandshakeTimers, /// Whether we are ready to release application data from poll_output. release_app_data: bool, @@ -114,13 +109,6 @@ pub struct Engine { close_notify_reported: bool, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum Timeout { - Disabled, - Unarmed, - Armed(Instant), -} - #[derive(Debug)] struct Entry { content_type: ContentType, @@ -138,8 +126,7 @@ impl Engine { pub fn new(config: Arc, auth: AuthMode) -> Self { let mut rng = SeededRng::new(config.rng_seed()); - let flight_backoff = - ExponentialBackoff::new(config.flight_start_rto(), config.flight_retries(), &mut rng); + let timers = HandshakeTimers::new(&config, &mut rng); let crypto_context = CryptoContext::new(auth, Arc::clone(&config)); @@ -162,9 +149,7 @@ impl Engine { transcript: Buf::new(), replay: ReplayWindow::new(), flight_saved_records: Vec::new(), - flight_backoff, - flight_timeout: Timeout::Unarmed, - connect_timeout: Timeout::Unarmed, + timers, release_app_data: false, peer_handshake_confirmed: false, close_notify_received: false, @@ -176,19 +161,30 @@ impl Engine { self.is_client = is_client; } - /// Set the next outgoing handshake message sequence number. - /// - /// Used by `Client::new_from_hybrid` to account for the hybrid - /// ClientHello (message_seq=0) that was already sent outside this engine. - pub fn set_next_handshake_seq_no(&mut self, seq: u16) { - self.next_handshake_seq_no = seq; + pub fn set_handshake_deadline(&mut self, deadline: Timeout) { + self.timers.set_handshake_deadline(deadline); } - /// Advance the epoch-0 record sequence number by one. - /// - /// Used by `Client::new_from_hybrid` so subsequent epoch-0 records - /// don't reuse the sequence number of the hybrid ClientHello record. - pub fn advance_epoch_0_sequence(&mut self) { + pub fn handshake_deadline(&self) -> Timeout { + self.timers.handshake_deadline() + } + + pub fn discard_initial_client_hello(&mut self) { + self.timers.set_handshake_deadline(Timeout::Unarmed); + self.peer_handshake_seq_no = 0; + self.transcript.clear(); + } + + /// Restore an already emitted hybrid ClientHello and its outstanding timers. + pub fn inject_hybrid_client_hello(&mut self, fragment: Buf, timers: HandshakeTimers) { + self.timers = timers; + self.transcript.extend_from_slice(&fragment); + self.flight_saved_records.push(Entry { + content_type: ContentType::Handshake, + epoch: 0, + fragment, + }); + self.next_handshake_seq_no = 1; self.sequence_epoch_0.sequence_number += 1; } @@ -225,6 +221,42 @@ impl Engine { &mut self.crypto_context } + pub fn handle_packet(&mut self, packet: &[u8], now: Instant) -> Result<(), InternalError> { + self.parse_packet(packet)?; + if !self.is_client + && self.timers.handshake_deadline() == Timeout::Unarmed + && self + .queue_rx + .iter() + .flat_map(|incoming| incoming.records().iter()) + .any(|record| { + record.record().sequence.epoch == 0 + && record.handshakes().iter().any(|handshake| { + let header = &handshake.header; + match &handshake.body { + Body::Fragment(range) + if header.msg_type == MessageType::ClientHello + && header.message_seq == self.peer_handshake_seq_no => + { + accepts_client_hello_fragment( + header.length, + header.fragment_offset, + header.fragment_length, + record.buffer(), + range.clone(), + ClientHello::parse, + ) + } + _ => false, + } + }) + }) + { + self.timers.start_handshake(now); + } + Ok(()) + } + pub fn parse_packet(&mut self, packet: &[u8]) -> Result<(), InternalError> { let cs = self.cipher_suite; let incoming = Incoming::parse_packet(packet, self, cs)?; @@ -279,7 +311,11 @@ impl Engine { // unauthenticated noise (or a replay/amplification attempt) and must not // drive a resend. if let Some(dupe_seq) = maybe_dupe_seq { - if dupe_seq < self.peer_handshake_seq_no && !self.peer_handshake_confirmed { + if dupe_seq < self.peer_handshake_seq_no + && !self.peer_handshake_confirmed + && !self.flight_saved_records.is_empty() + && self.timers.request_resend(&mut self.rng) + { self.flight_resend("dupe triggers resend")?; } } @@ -369,48 +405,12 @@ impl Engine { } pub fn handle_timeout(&mut self, now: Instant) -> Result<(), Error> { - if self.connect_timeout == Timeout::Unarmed { - debug!( - "Connect timeout in: {:.03}s", - self.config.handshake_timeout().as_secs_f32() - ); - let timeout = now + self.config.handshake_timeout(); - self.connect_timeout = Timeout::Armed(timeout); - } - if self.flight_timeout == Timeout::Unarmed { - debug!( - "Flight timeout in: {:.03}s", - self.flight_backoff.rto().as_secs_f32() - ); - let timeout = now + self.flight_backoff.rto(); - self.flight_timeout = Timeout::Armed(timeout); - } - - // The connect timeout is the overall timeout for establishing the connection - if let Timeout::Armed(connect_timeout) = self.connect_timeout { - if now >= connect_timeout { - return Err(Error::Timeout(crate::TimeoutError::Connect)); - } - } - - // If there is no flight timeout, we have already checked the global connect timeout. - let Timeout::Armed(flight_timeout) = self.flight_timeout else { - return Ok(()); - }; - - if now >= flight_timeout { - if self.flight_backoff.can_retry() { - self.flight_backoff.attempt(&mut self.rng); - debug!( - "Re-arm flight timeout due to resend in {}", - self.flight_backoff.rto().as_secs_f32() - ); - let timeout = now + self.flight_backoff.rto(); - self.flight_timeout = Timeout::Armed(timeout); - self.flight_resend("flight timeout")?; - } else { - return Err(Error::Timeout(crate::TimeoutError::Handshake)); - } + if self + .timers + .handle_timeout(now, &mut self.rng) + .map_err(Error::Timeout)? + { + self.flight_resend("flight timeout")?; } Ok(()) @@ -427,7 +427,13 @@ impl Engine { }; match self.poll_packet_tx(buf) { - PollOutput::Data(p) => return Output::Packet(p), + PollOutput::Data(p) => { + if self.is_client { + self.timers.start_handshake(now); + } + self.timers.flight_sent(now); + return Output::Packet(p); + } PollOutput::BufferTooSmall { needed } => return Output::BufferTooSmall { needed }, PollOutput::None(_) => {} } @@ -437,7 +443,7 @@ impl Engine { return Output::CloseNotify; } - let next_timeout = self.poll_timeout(now); + let next_timeout = self.timers.poll_timeout(now); Output::Timeout(next_timeout) } @@ -506,45 +512,15 @@ impl Engine { PollOutput::Data(&buf[..len]) } - fn poll_timeout(&self, now: Instant) -> Instant { - // No timeouts, return a distant future - if self.connect_timeout == Timeout::Disabled && self.flight_timeout == Timeout::Disabled { - const DISTANT_FUTURE: Duration = Duration::from_secs(10 * 365 * 24 * 60 * 60); - return now + DISTANT_FUTURE; - } - - match (self.connect_timeout, self.flight_timeout) { - // Keep this before the `(Armed, _)` arms. Starting a new flight resets its timer to - // `Unarmed`, but leaves the overall connection timer armed. If that mixed state - // returned the connection deadline, the caller would not drive `handle_timeout` to - // arm the flight timer until the whole handshake expired, so the flight would never - // be retransmitted. Returning `now` requests that immediate drive; the next poll sees - // both concrete deadlines and can return the earlier one. - (Timeout::Unarmed, _) | (_, Timeout::Unarmed) => now, - (Timeout::Armed(c), Timeout::Armed(f)) => { - if c < f { - c - } else { - f - } - } - (Timeout::Armed(c), _) => c, - (_, Timeout::Armed(f)) => f, - _ => now, - } - } - pub fn flight_begin(&mut self, flight_no: u8) { debug!("Begin flight {}", flight_no); - self.flight_backoff.reset(&mut self.rng); + self.timers.begin_flight(&mut self.rng); self.flight_clear_resends(); - self.flight_timeout = Timeout::Unarmed; } pub fn flight_stop_resend_timers(&mut self) { debug!("Stop connect and flight timeouts"); - self.flight_timeout = Timeout::Disabled; - self.connect_timeout = Timeout::Disabled; + self.timers.stop(); // The client stops its resend timer only once it has received the // server's final flight, which proves the server received the client's @@ -1045,8 +1021,7 @@ impl Engine { pub fn abort(&mut self) { self.queue_tx.clear(); self.flight_saved_records.clear(); - self.flight_timeout = Timeout::Disabled; - self.connect_timeout = Timeout::Disabled; + self.timers.stop(); } /// Pop a buffer from the buffer pool for temporary use @@ -1398,3 +1373,42 @@ impl RecordHandler for Engine { self.release_app_data } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn timing_resend_queue_failure_is_fatal() { + let now = Instant::now(); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .build() + .expect("valid config"), + ); + let mut engine = Engine::new(config, AuthMode::Psk); + engine.flight_begin(1); + engine + .create_record(ContentType::Handshake, 0, true, |fragment| fragment.push(1)) + .expect("queue original flight"); + let mut buffer = [0; 64]; + assert!(matches!( + engine.poll_output(&mut buffer, now), + Output::Packet(_) + )); + let Output::Timeout(retry_at) = engine.poll_output(&mut buffer, now) else { + panic!("expected flight timer"); + }; + engine.config = Arc::new( + Config::builder() + .max_queue_tx(0) + .build() + .expect("inject exhausted transmit capacity"), + ); + assert_eq!( + engine.handle_timeout(retry_at), + Err(Error::TransmitQueueFull) + ); + } +} diff --git a/src/dtls12/server.rs b/src/dtls12/server.rs index 1e0a765a..7ba1cf19 100644 --- a/src/dtls12/server.rs +++ b/src/dtls12/server.rs @@ -36,6 +36,7 @@ use crate::dtls12::message::{ServerHello, SessionId, SignatureAlgorithm}; use crate::dtls12::message::{SignatureAlgorithmsExtension, SignatureAndHashAlgorithm}; use crate::dtls12::message::{SignatureAndHashAlgorithmVec, SrtpProfileId}; use crate::dtls12::message::{SrtpProfileVec, SupportedGroupsExtension, UseSrtpExtension}; +use crate::timer::Timeout; use crate::{Config, Error, InternalError, Output}; /// Length of the random dummy PSK used when identity resolution fails. @@ -184,6 +185,10 @@ impl Server { Client::new_with_engine(self.engine, self.last_now) } + pub fn set_handshake_deadline(&mut self, deadline: Timeout) { + self.engine.set_handshake_deadline(deadline); + } + pub(crate) fn state_name(&self) -> &'static str { self.state.name() } @@ -199,13 +204,22 @@ impl Server { } pub fn handle_packet(&mut self, packet: &[u8]) -> Result<(), Error> { + let handshake_was_idle = self.engine.handshake_deadline() == Timeout::Unarmed; match self .engine - .parse_packet(packet) + .handle_packet(packet, self.last_now) .and_then(|_| self.make_progress()) { Ok(()) => Ok(()), - Err(e) => e.into_public_error().map_or(Ok(()), Err), + Err(error) => match error.into_public_error() { + Some(error) => Err(error), + None => { + if handshake_was_idle && self.state == State::AwaitClientHello { + self.engine.discard_initial_client_hello(); + } + Ok(()) + } + }, } } diff --git a/src/dtls13/client.rs b/src/dtls13/client.rs index 0f5193d5..9c960f9d 100644 --- a/src/dtls13/client.rs +++ b/src/dtls13/client.rs @@ -59,6 +59,7 @@ use crate::dtls13::message::SupportedVersionsClientHello; use crate::dtls13::message::SupportedVersionsServerHello; use crate::dtls13::message::UseSrtpExtension; use crate::dtls13::message::parse_cookie_extension; +use crate::timer::HandshakeTimers; use crate::{Error, InternalError, KeyingMaterial, Output}; /// DTLS 1.3 client @@ -175,13 +176,18 @@ impl Client { config: std::sync::Arc, certificate: crate::DtlsCertificate, now: Instant, + timers: HandshakeTimers, ) -> Result { let mut engine = Engine::new(config, certificate); engine.set_client(true); // Inject transcript + sequence state from the hybrid CH that was // already sent on the wire by ClientPending. - engine.inject_hybrid_client_hello(&hybrid.transcript_bytes); + engine.inject_hybrid_client_hello( + &hybrid.transcript_bytes, + hybrid.handshake_fragment, + timers, + ); let mut client = Client { state: State::AwaitServerHello, @@ -232,7 +238,7 @@ impl Client { pub fn handle_packet(&mut self, packet: &[u8]) -> Result<(), Error> { match self .engine - .parse_packet(packet) + .handle_packet(packet, self.last_now) .and_then(|_| self.make_progress()) { Ok(()) => Ok(()), @@ -241,6 +247,9 @@ impl Client { } pub fn poll_output<'a>(&mut self, buf: &'a mut [u8]) -> Output<'a> { + if self.state == State::SendClientHello { + return Output::Timeout(self.last_now); + } if let Some(event) = self.local_events.pop_front() { return event.into_output(buf, &self.server_certificates); } diff --git a/src/dtls13/engine.rs b/src/dtls13/engine.rs index 6d822164..3553cb4e 100644 --- a/src/dtls13/engine.rs +++ b/src/dtls13/engine.rs @@ -17,6 +17,7 @@ use crate::crypto::SupportedKxGroup; use crate::crypto::prf_hkdf; use crate::dtls13::incoming::{Incoming, Record, RecordHandler}; use crate::dtls13::message::Body; +use crate::dtls13::message::ClientHello; use crate::dtls13::message::ContentType; use crate::dtls13::message::Dtls13CipherSuite; use crate::dtls13::message::Dtls13Record; @@ -25,8 +26,9 @@ use crate::dtls13::message::Header; use crate::dtls13::message::KeyUpdateRequest; use crate::dtls13::message::MessageType; use crate::dtls13::message::Sequence; -use crate::timer::ExponentialBackoff; +use crate::timer::{HandshakeTimers, Timeout, deadline}; use crate::types::{HashAlgorithm, Random}; +use crate::util::accepts_client_hello_fragment; use crate::window::ReplayWindow; use crate::{Config, DtlsCertificate, Error, InternalError, Output, SeededRng}; @@ -131,14 +133,7 @@ pub struct Engine { /// The records that have been sent in the current flight. flight_saved_records: ArrayVec, - /// Flight backoff - flight_backoff: ExponentialBackoff, - - /// Timeout for the current flight - flight_timeout: Timeout, - - /// Global timeout for the entire connect operation. - connect_timeout: Timeout, + timers: HandshakeTimers, /// Whether we are ready to release application data from poll_output. release_app_data: bool, @@ -185,13 +180,6 @@ struct RecvEpochEntry { replay: ReplayWindow, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum Timeout { - Disabled, - Unarmed, - Armed(Instant), -} - #[derive(Debug)] struct Entry { content_type: ContentType, @@ -211,8 +199,7 @@ impl Engine { pub fn new(config: Arc, certificate: DtlsCertificate) -> Self { let mut rng = SeededRng::new(config.rng_seed()); - let flight_backoff = - ExponentialBackoff::new(config.flight_start_rto(), config.flight_retries(), &mut rng); + let timers = HandshakeTimers::new(&config, &mut rng); let signing_key = config .crypto_provider() @@ -254,9 +241,7 @@ impl Engine { handshake_ack_deadline: None, datagram_sealed: false, flight_saved_records: ArrayVec::new(), - flight_backoff, - flight_timeout: Timeout::Unarmed, - connect_timeout: Timeout::Unarmed, + timers, release_app_data: false, exporter_master_secret: None, app_send_record_count: 0, @@ -268,14 +253,28 @@ impl Engine { } } - pub fn into_fallback(self) -> (Arc, DtlsCertificate) { - (self.config, self.certificate) + pub fn into_fallback(self) -> (Arc, DtlsCertificate, Timeout) { + ( + self.config, + self.certificate, + self.timers.handshake_deadline(), + ) } pub fn set_client(&mut self, is_client: bool) { self.is_client = is_client; } + pub fn handshake_deadline(&self) -> Timeout { + self.timers.handshake_deadline() + } + + pub fn discard_initial_client_hello(&mut self) { + self.timers.set_handshake_deadline(Timeout::Unarmed); + self.peer_handshake_seq_no = 0; + self.transcript.clear(); + } + /// Inject a pre-built hybrid ClientHello into this engine. /// /// Inject the transcript and state from a hybrid ClientHello that was @@ -284,9 +283,22 @@ impl Engine { /// Sets the transcript, advances the handshake sequence number to 1, /// and bumps the epoch-0 record sequence so subsequent records don't /// collide. Does **not** enqueue the record for output — the hybrid - /// CH was already transmitted. - pub fn inject_hybrid_client_hello(&mut self, transcript_bytes: &[u8]) { + /// CH was already transmitted. Retains its outstanding retry state. + pub fn inject_hybrid_client_hello( + &mut self, + transcript_bytes: &[u8], + fragment: Buf, + timers: HandshakeTimers, + ) { + self.timers = timers; self.transcript.extend_from_slice(transcript_bytes); + self.flight_saved_records.push(Entry { + content_type: ContentType::Handshake, + epoch: 0, + send_seq: 0, + fragment, + acked: false, + }); self.next_handshake_seq_no = 1; // Advance past the record sequence used by the hybrid CH. // Defense-in-depth: guard against epoch-0 sequence overflow. @@ -340,6 +352,42 @@ impl Engine { &mut *self.signing_key } + pub fn handle_packet(&mut self, packet: &[u8], now: Instant) -> Result<(), InternalError> { + self.parse_packet(packet)?; + if !self.is_client + && self.timers.handshake_deadline() == Timeout::Unarmed + && self + .queue_rx + .iter() + .flat_map(|incoming| incoming.records().iter()) + .any(|record| { + record.record().sequence.epoch == 0 + && record.handshakes().iter().any(|handshake| { + let header = &handshake.header; + match &handshake.body { + Body::Fragment(range) + if header.msg_type == MessageType::ClientHello + && header.message_seq == self.peer_handshake_seq_no => + { + accepts_client_hello_fragment( + header.length, + header.fragment_offset, + header.fragment_length, + record.buffer(), + range.clone(), + ClientHello::parse_allow_unknown_suites, + ) + } + _ => false, + } + }) + }) + { + self.timers.start_handshake(now); + } + Ok(()) + } + pub fn parse_packet(&mut self, packet: &[u8]) -> Result<(), InternalError> { let cs = self.cipher_suite; let incoming = Incoming::parse_packet(packet, self, cs)?; @@ -352,11 +400,7 @@ impl Engine { fn insert_incoming(&mut self, incoming: Incoming) -> Result<(), Error> { if self.queue_rx.len() >= self.config.max_queue_rx() { - warn!( - "Receive queue full (max {}): {:?}", - self.config.max_queue_rx(), - self.queue_rx - ); + warn!("Receive queue full"); return Err(Error::ReceiveQueueFull); } @@ -386,7 +430,10 @@ impl Engine { .next(); if let Some(dupe_seq) = maybe_dupe_seq { - if dupe_seq < self.peer_handshake_seq_no { + if dupe_seq < self.peer_handshake_seq_no + && !self.flight_saved_records.is_empty() + && self.timers.request_resend(&mut self.rng) + { self.flight_resend("dupe triggers resend")?; } } @@ -417,15 +464,7 @@ impl Engine { match search_result { Err(index) => { - // Track received record numbers for ACK generation - for record in incoming.records().iter() { - let seq = record.record().sequence; - if seq.epoch >= 2 && record.record().content_type == ContentType::Handshake { - let _ = self - .received_record_numbers - .try_push((seq.epoch as u64, seq.sequence_number)); - } - } + self.track_handshake_records(&incoming); self.queue_rx.insert(index, incoming); } Ok(index) => { @@ -450,15 +489,7 @@ impl Engine { existing_corrupt && incoming_ok }; if should_replace { - for record in incoming.records().iter() { - let seq = record.record().sequence; - if seq.epoch >= 2 && record.record().content_type == ContentType::Handshake - { - let _ = self - .received_record_numbers - .try_push((seq.epoch as u64, seq.sequence_number)); - } - } + self.track_handshake_records(&incoming); self.queue_rx[index] = incoming; } } @@ -476,7 +507,10 @@ impl Engine { .binary_search_by_key(&seq_current, |item| item.first().record().sequence); match search_result { - Err(index) => self.queue_rx.insert(index, incoming), + Err(index) => { + self.track_handshake_records(&incoming); + self.queue_rx.insert(index, incoming); + } Ok(_) => { // Duplicate - silently drop. For encrypted records (epoch >= 2) the replay // window filters most duplicates, but undecrypted ciphertext records can @@ -487,47 +521,24 @@ impl Engine { Ok(()) } - pub fn handle_timeout(&mut self, now: Instant) -> Result<(), Error> { - if self.connect_timeout == Timeout::Unarmed { - debug!( - "Connect timeout in: {:.03}s", - self.config.handshake_timeout().as_secs_f32() - ); - let timeout = now + self.config.handshake_timeout(); - self.connect_timeout = Timeout::Armed(timeout); - } - if self.flight_timeout == Timeout::Unarmed { - debug!( - "Flight timeout in: {:.03}s", - self.flight_backoff.rto().as_secs_f32() - ); - let timeout = now + self.flight_backoff.rto(); - self.flight_timeout = Timeout::Armed(timeout); - } - - if let Timeout::Armed(connect_timeout) = self.connect_timeout { - if now >= connect_timeout { - return Err(Error::Timeout(crate::TimeoutError::Connect)); + fn track_handshake_records(&mut self, incoming: &Incoming) { + for record in incoming.records().iter() { + let sequence = record.record().sequence; + if sequence.epoch >= 2 && record.record().content_type == ContentType::Handshake { + let _ = self + .received_record_numbers + .try_push((sequence.epoch as u64, sequence.sequence_number)); } } + } - let Timeout::Armed(flight_timeout) = self.flight_timeout else { - return Ok(()); - }; - - if now >= flight_timeout { - if self.flight_backoff.can_retry() { - self.flight_backoff.attempt(&mut self.rng); - debug!( - "Re-arm flight timeout due to resend in {}", - self.flight_backoff.rto().as_secs_f32() - ); - let timeout = now + self.flight_backoff.rto(); - self.flight_timeout = Timeout::Armed(timeout); - self.flight_resend("flight timeout")?; - } else { - return Err(Error::Timeout(crate::TimeoutError::Handshake)); - } + pub fn handle_timeout(&mut self, now: Instant) -> Result<(), Error> { + if self + .timers + .handle_timeout(now, &mut self.rng) + .map_err(Error::Timeout)? + { + self.flight_resend("flight timeout")?; } // During handshake, schedule/flush ACKs to help peer with selective retransmission @@ -549,7 +560,13 @@ impl Engine { self.maybe_schedule_handshake_ack(now); match self.poll_packet_tx(buf) { - PollOutput::Data(p) => return Output::Packet(p), + PollOutput::Data(p) => { + if self.is_client { + self.timers.start_handshake(now); + } + self.timers.flight_sent(now); + return Output::Packet(p); + } PollOutput::BufferTooSmall { needed } => return Output::BufferTooSmall { needed }, PollOutput::None(_) => {} } @@ -635,26 +652,7 @@ impl Engine { } fn poll_timeout(&self, now: Instant) -> Instant { - if self.connect_timeout == Timeout::Disabled - && self.flight_timeout == Timeout::Disabled - && self.handshake_ack_deadline.is_none() - { - const DISTANT_FUTURE: Duration = Duration::from_secs(10 * 365 * 24 * 60 * 60); - return now + DISTANT_FUTURE; - } - - let mut timeout = match (self.connect_timeout, self.flight_timeout) { - (Timeout::Armed(c), Timeout::Armed(f)) => { - if c < f { - c - } else { - f - } - } - (Timeout::Armed(c), _) => c, - (_, Timeout::Armed(f)) => f, - _ => now + Duration::from_secs(10 * 365 * 24 * 60 * 60), - }; + let mut timeout = self.timers.poll_timeout(now); if let Some(deadline) = self.handshake_ack_deadline { if deadline < timeout { @@ -667,15 +665,13 @@ impl Engine { pub fn flight_begin(&mut self, flight_no: u8) { debug!("Begin flight {}", flight_no); - self.flight_backoff.reset(&mut self.rng); + self.timers.begin_flight(&mut self.rng); self.flight_clear_resends(); - self.flight_timeout = Timeout::Unarmed; } pub fn flight_stop_resend_timers(&mut self) { - debug!("Stop connect and flight timeouts"); - self.flight_timeout = Timeout::Disabled; - self.connect_timeout = Timeout::Disabled; + debug!("Stop flight timeout"); + self.timers.stop_flight(); } fn flight_clear_resends(&mut self) { @@ -917,11 +913,7 @@ impl Engine { .unwrap_or(false); if !can_append && self.queue_tx.len() >= self.config.max_queue_tx() { - warn!( - "Transmit queue full (max {}): {:?}", - self.config.max_queue_tx(), - self.queue_tx - ); + warn!("Transmit queue full"); return Err(Error::TransmitQueueFull); } @@ -1069,11 +1061,7 @@ impl Engine { .unwrap_or(false); if !can_append && self.queue_tx.len() >= self.config.max_queue_tx() { - warn!( - "Transmit queue full (max {}): {:?}", - self.config.max_queue_tx(), - self.queue_tx - ); + warn!("Transmit queue full"); return Err(Error::TransmitQueueFull); } @@ -1259,10 +1247,12 @@ impl Engine { pub fn release_application_data(&mut self) { self.release_app_data = true; self.hs_recv_keys = None; + self.timers.finish_handshake(); } pub fn release_application_data_retaining_handshake_keys(&mut self) { self.release_app_data = true; + self.timers.finish_handshake(); } /// Whether a close_notify alert has been received from the peer. @@ -1285,8 +1275,7 @@ impl Engine { /// allowing the queued close_notify alert to be sent. pub fn cancel_flights(&mut self) { self.flight_saved_records.clear(); - self.flight_timeout = Timeout::Disabled; - self.connect_timeout = Timeout::Disabled; + self.timers.stop(); self.handshake_ack_deadline = None; } @@ -1295,8 +1284,7 @@ impl Engine { pub fn abort(&mut self) { self.queue_tx.clear(); self.flight_saved_records.clear(); - self.flight_timeout = Timeout::Disabled; - self.connect_timeout = Timeout::Disabled; + self.timers.stop(); self.handshake_ack_deadline = None; } @@ -1309,6 +1297,10 @@ impl Engine { pub fn send_ack_retransmittable(&mut self) -> Result<(), Error> { if !self.received_record_numbers.is_empty() { + // This ACK is a new courtesy-resend flight: reset its duplicate + // budget, but keep timer-driven retransmission disabled. + self.timers.begin_flight(&mut self.rng); + self.timers.stop_flight(); self.flight_clear_resends(); } self.send_ack_inner(true) @@ -1385,7 +1377,7 @@ impl Engine { .all(|e| e.acked); if has_epoch2 && all_epoch2_acked { debug!("Handshake flight ACKed; stopping retransmission"); - self.flight_timeout = Timeout::Disabled; + self.timers.stop_flight(); self.flight_clear_resends(); } @@ -1400,7 +1392,7 @@ impl Engine { self.prev_app_send_keys = None; self.key_update_in_flight = false; self.flight_clear_resends(); - self.flight_timeout = Timeout::Disabled; + self.timers.stop_flight(); } Ok(()) @@ -1531,15 +1523,10 @@ impl Engine { let delay = if self.has_gap_in_incoming_handshake() { Duration::from_millis(0) } else { - let rto = self.flight_backoff.rto(); - if rto > Duration::from_millis(0) { - rto / 4 - } else { - Duration::from_millis(0) - } + self.timers.rto() / 4 }; - self.handshake_ack_deadline = Some(now + delay); + self.handshake_ack_deadline = Some(deadline(now, delay)); } /// Flush a scheduled handshake ACK if the deadline has passed. @@ -1921,9 +1908,8 @@ impl Engine { /// the current app epoch. Send keys rotate only after its ACK arrives. pub fn create_key_update(&mut self, request: KeyUpdateRequest) -> Result<(), Error> { // Set up retransmission - self.flight_backoff.reset(&mut self.rng); + self.timers.begin_flight(&mut self.rng); self.flight_clear_resends(); - self.flight_timeout = Timeout::Unarmed; let msg_seq = self.next_handshake_seq_no; self.next_handshake_seq_no += 1; @@ -2539,7 +2525,7 @@ impl RecordHandler for Engine { } } -#[cfg(test)] +#[cfg(all(test, feature = "rcgen"))] mod tests { use super::*; @@ -2553,6 +2539,42 @@ mod tests { Engine::new(config, cert) } + #[test] + #[cfg(feature = "rcgen")] + fn timing_resend_queue_failure_is_fatal() { + let now = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .build() + .expect("valid config"), + ); + let mut engine = Engine::new(config, certificate); + engine.flight_begin(1); + engine + .create_plaintext_record(ContentType::Handshake, true, |fragment| fragment.push(1)) + .expect("queue original flight"); + let mut buffer = [0; 64]; + assert!(matches!( + engine.poll_output(&mut buffer, now), + Output::Packet(_) + )); + let Output::Timeout(retry_at) = engine.poll_output(&mut buffer, now) else { + panic!("expected flight timer"); + }; + engine.config = Arc::new( + Config::builder() + .max_queue_tx(0) + .build() + .expect("inject exhausted transmit capacity"), + ); + assert_eq!( + engine.handle_timeout(retry_at), + Err(Error::TransmitQueueFull) + ); + } + struct PassthroughRecordHandler; impl RecordHandler for PassthroughRecordHandler { diff --git a/src/dtls13/server.rs b/src/dtls13/server.rs index ebe3ed5b..e8d87f83 100644 --- a/src/dtls13/server.rs +++ b/src/dtls13/server.rs @@ -66,6 +66,7 @@ use crate::dtls13::message::SupportedVersionsClientHello; use crate::dtls13::message::SupportedVersionsServerHello; use crate::dtls13::message::UseSrtpExtension; use crate::dtls13::message::parse_cookie_extension; +use crate::timer::Timeout; use crate::{Config, DtlsCertificate, Error, InternalError, Output}; /// Magic random value indicating HelloRetryRequest (RFC 8446 Section 4.1.3). @@ -227,9 +228,17 @@ impl Server { /// /// 1. Switching a server pending (auto-mode) to dtls12 server /// 2. set_active(true), turning a server pending (auto-mode) to a ClientPending - pub fn into_parts(self) -> (Arc, DtlsCertificate, Instant, VecDeque) { - let (config, cert) = self.engine.into_fallback(); - (config, cert, self.last_now, self.retained_hello) + pub fn into_parts( + self, + ) -> ( + Arc, + DtlsCertificate, + Instant, + VecDeque, + Timeout, + ) { + let (config, cert, deadline) = self.engine.into_fallback(); + (config, cert, self.last_now, self.retained_hello, deadline) } pub(crate) fn state_name(&self) -> &'static str { @@ -248,6 +257,7 @@ impl Server { } pub fn handle_packet(&mut self, packet: &[u8]) -> Result<(), Error> { + let handshake_was_idle = self.engine.handshake_deadline() == Timeout::Unarmed; // In auto-sense mode, buffer raw packets while still waiting for // the ClientHello so they can be replayed to Server12 on fallback. if self.auto_mode && self.state == State::AwaitClientHello { @@ -260,7 +270,7 @@ impl Server { match self .engine - .parse_packet(packet) + .handle_packet(packet, self.last_now) .and_then(|_| self.make_progress()) { Ok(()) => {} @@ -268,6 +278,10 @@ impl Server { if let Some(err) = e.into_public_error() { return Err(err); } + if handshake_was_idle && self.state == State::AwaitClientHello { + self.engine.discard_initial_client_hello(); + self.retained_hello.clear(); + } return Ok(()); } } diff --git a/src/error.rs b/src/error.rs index c22663f7..76828450 100644 --- a/src/error.rs +++ b/src/error.rs @@ -547,6 +547,10 @@ pub enum ConfigError { }, /// The configured AEAD encryption limit is too small. AeadEncryptionLimitTooSmall, + /// The handshake timeout is zero or greater than 3650 days. + InvalidHandshakeTimeout, + /// The initial flight RTO is zero or greater than 3650 days. + InvalidFlightStartRto, /// Cipher-suite filtering removed every available suite. NoCipherSuitesAfterFiltering, /// A PSK resolver is configured but no PSK cipher suite remains enabled. @@ -1229,6 +1233,12 @@ impl fmt::Display for ConfigError { Self::AeadEncryptionLimitTooSmall => { write!(f, "aead_encryption_limit must be at least 1") } + Self::InvalidHandshakeTimeout => { + write!(f, "handshake_timeout must be nonzero and at most 3650 days") + } + Self::InvalidFlightStartRto => { + write!(f, "flight_start_rto must be nonzero and at most 3650 days") + } Self::NoCipherSuitesAfterFiltering => write!( f, concat!( diff --git a/src/lib.rs b/src/lib.rs index 8356df11..684db24e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -102,12 +102,13 @@ //! use dimpl::{certificate, Config, Dtls, Output}; //! //! // Stub I/O to keep the example focused on the state machine -//! enum Event { Udp(Vec), Timer(Instant) } -//! fn wait_next_event(_next_wake: Option) -> Event { Event::Udp(Vec::new()) } +//! enum Event { Udp(Vec, Instant), Timer(Instant) } +//! fn wait_next_event(_next_wake: Option) -> Event { Event::Udp(Vec::new(), Instant::now()) } //! fn send_udp(_bytes: &[u8]) {} //! //! fn example_event_loop(mut dtls: Dtls) -> Result<(), dimpl::Error> { //! let mut next_wake: Option = None; +//! let mut received_packet: Option> = None; //! loop { //! // Drain engine output until we have to wait for I/O or a timer //! let mut out_buf = vec![0u8; 2048]; @@ -138,9 +139,17 @@ //! } //! } //! +//! if let Some(packet) = received_packet.take() { +//! dtls.handle_packet(&packet)?; +//! continue; +//! } +//! //! // Block waiting for either UDP input or the scheduled timeout //! match wait_next_event(next_wake) { -//! Event::Udp(pkt) => dtls.handle_packet(&pkt)?, +//! Event::Udp(pkt, now) => { +//! dtls.handle_timeout(now)?; +//! received_packet = Some(pkt); +//! } //! Event::Timer(now) => dtls.handle_timeout(now)?, //! } //! } @@ -651,7 +660,7 @@ impl Dtls { } Inner::Server13(s) => { if s.is_auto_mode() { - let (config, certificate, now, _) = s.into_parts(); + let (config, certificate, now, _, _) = s.into_parts(); let cp = ClientPending::new(config, certificate, now) .expect("failed to build hybrid ClientHello"); self.inner = Some(Inner::ClientPending(cp)); @@ -668,6 +677,12 @@ impl Dtls { } /// Process an incoming DTLS datagram. + /// + /// Refresh the logical clock with [`Self::handle_timeout`] before receiving + /// packets at a new instant, even if no advertised timer is due. A server's + /// overall handshake deadline starts at that time when the first valid + /// ClientHello fragment is accepted. Rejected or unrelated input does not + /// start the deadline. Poll until [`Output::Timeout`] after each mutation. pub fn handle_packet(&mut self, packet: &[u8]) -> Result<(), Error> { // unwrap is ok. The inner is only Option to work around borrowing // issues when doing auto-sensing of DTLS version. @@ -724,15 +739,16 @@ impl Dtls { let Inner::ClientPending(cp) = inner else { unreachable!() }; - let (hybrid, config, certificate, now) = cp.into_parts(); + let (hybrid, config, certificate, now, timers) = cp.into_parts(); match version { auto::DetectedVersion::Dtls12 => { let mut client12 = Client12::new_from_hybrid( hybrid.random, - &hybrid.handshake_fragment, + hybrid.handshake_fragment, config, certificate, now, + timers, )?; // Feed the HVR to Client12 — it enters // AwaitHelloVerifyRequest and processes the cookie. @@ -744,7 +760,8 @@ impl Dtls { Ok(()) } auto::DetectedVersion::Dtls13 => { - let mut client13 = Client13::new_from_hybrid(hybrid, config, certificate, now)?; + let mut client13 = + Client13::new_from_hybrid(hybrid, config, certificate, now, timers)?; if let Err(e) = client13.handle_packet(packet) { self.inner = Some(Inner::Client13(client13)); return Err(e); @@ -767,7 +784,7 @@ impl Dtls { _ => unreachable!(), }; - let (config, cert, now, buffered) = server.into_parts(); + let (config, cert, now, buffered, deadline) = server.into_parts(); // A Server12 instance is either cert-auth or PSK-auth — the auth // mode must be chosen before construction. Peek at the buffered @@ -781,6 +798,7 @@ impl Dtls { } else { Server12::new(config, cert, now) }; + server12.set_handshake_deadline(deadline); server12.handle_timeout(now)?; self.inner = Some(Inner::Server12(server12)); @@ -792,6 +810,16 @@ impl Dtls { } /// Poll for pending output from the DTLS engine. + /// + /// A client's overall handshake deadline starts when the first ClientHello + /// packet (or fragment) is returned as [`Output::Packet`], using the latest + /// logical time supplied to [`Self::handle_timeout`]. Refresh that time + /// before polling if time has advanced. Queuing a packet internally or + /// returning [`Output::BufferTooSmall`] does not start the deadline. + /// + /// Emission is the Sans-IO send boundary, not a socket-write timestamp. + /// Any subsequent caller-side queuing or transport delay is outside dimpl. + /// Continue polling until [`Output::Timeout`]. pub fn poll_output<'a>(&mut self, buf: &'a mut [u8]) -> Output<'a> { match self.inner.as_mut().unwrap() { Inner::Client12(client) => client.poll_output(buf), @@ -803,6 +831,16 @@ impl Dtls { } /// Handle time-based events such as retransmission timers. + /// + /// Also advances the logical clock used by subsequent send/receive calls; + /// it may be called before an advertised timeout. Advancing time alone does + /// not activate the overall handshake timer or an unsent flight's retry + /// timer. Supply monotonically increasing instants and poll until + /// [`Output::Timeout`] after driving the endpoint. + /// + /// The overall deadline is absolute across flights and Auto version + /// transitions. Retry exhaustion can fail earlier. Timeout errors are + /// terminal: stop driving the failed instance. pub fn handle_timeout(&mut self, now: Instant) -> Result<(), Error> { match self.inner.as_mut().unwrap() { Inner::Client12(client) => client.handle_timeout(now), diff --git a/src/timer.rs b/src/timer.rs index ad5fe5fe..1c7f3093 100644 --- a/src/timer.rs +++ b/src/timer.rs @@ -1,10 +1,23 @@ -use std::ops::Mul; -use std::time::Duration; +use std::time::{Duration, Instant}; -use crate::SeededRng; +use crate::{Config, SeededRng, TimeoutError}; -// In seconds. const JITTER_RANGE: f32 = 0.5; +const IDLE_INTERVAL: Duration = Duration::from_secs(10 * 365 * 24 * 60 * 60); + +pub struct HandshakeTimers { + handshake_timeout: Duration, + handshake: Timeout, + flight: Timeout, + backoff: ExponentialBackoff, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Timeout { + Disabled, + Unarmed, + Armed(Instant), +} pub struct ExponentialBackoff { start_rto: Duration, @@ -14,6 +27,104 @@ pub struct ExponentialBackoff { left: usize, } +impl HandshakeTimers { + pub fn new(config: &Config, rng: &mut SeededRng) -> Self { + Self { + handshake_timeout: config.handshake_timeout(), + handshake: Timeout::Unarmed, + flight: Timeout::Disabled, + backoff: ExponentialBackoff::new( + config.flight_start_rto(), + config.flight_retries(), + rng, + ), + } + } + + pub fn start_handshake(&mut self, now: Instant) { + if self.handshake == Timeout::Unarmed { + self.handshake = Timeout::Armed(deadline(now, self.handshake_timeout)); + } + } + + pub fn handshake_deadline(&self) -> Timeout { + self.handshake + } + + pub fn set_handshake_deadline(&mut self, timeout: Timeout) { + self.handshake = timeout; + } + + pub fn begin_flight(&mut self, rng: &mut SeededRng) { + self.backoff.reset(rng); + self.flight = Timeout::Unarmed; + } + + pub fn flight_sent(&mut self, now: Instant) { + if self.flight == Timeout::Unarmed { + self.flight = Timeout::Armed(deadline(now, self.backoff.rto())); + } + } + + pub fn stop_flight(&mut self) { + self.flight = Timeout::Disabled; + } + + pub fn finish_handshake(&mut self) { + self.handshake = Timeout::Disabled; + } + + pub fn stop(&mut self) { + self.finish_handshake(); + self.stop_flight(); + } + + pub fn rto(&self) -> Duration { + self.backoff.rto() + } + + /// Reserve a retry for a saved flight. Emission arms its next timeout. + pub fn request_resend(&mut self, rng: &mut SeededRng) -> bool { + if self.flight == Timeout::Unarmed || !self.backoff.can_retry() { + return false; + } + self.backoff.attempt(rng); + if matches!(self.flight, Timeout::Armed(_)) { + self.flight = Timeout::Unarmed; + } + true + } + + pub fn handle_timeout( + &mut self, + now: Instant, + rng: &mut SeededRng, + ) -> Result { + if let Timeout::Armed(timeout) = self.handshake { + if now >= timeout { + return Err(TimeoutError::Connect); + } + } + if let Timeout::Armed(timeout) = self.flight { + if now >= timeout { + if !self.request_resend(rng) { + return Err(TimeoutError::Handshake); + } + return Ok(true); + } + } + Ok(false) + } + + pub fn poll_timeout(&self, now: Instant) -> Instant { + match (self.handshake, self.flight) { + (Timeout::Armed(connect), Timeout::Armed(flight)) => connect.min(flight), + (Timeout::Armed(timeout), _) | (_, Timeout::Armed(timeout)) => timeout, + _ => deadline(now, IDLE_INTERVAL), + } + } +} + impl ExponentialBackoff { pub fn new(start_rto: Duration, retries: usize, rng: &mut SeededRng) -> Self { Self { @@ -32,16 +143,15 @@ impl ExponentialBackoff { } pub fn rto(&self) -> Duration { + let jitter = self.rto.mul_f64(f64::from(self.jitter.abs())); if self.jitter < 0.0 { - let duration = Duration::from_secs_f32(self.jitter.abs()); - self.rto.saturating_sub(duration) + self.rto.saturating_sub(jitter) } else { - self.rto + Duration::from_secs_f32(self.jitter) + self.rto.saturating_add(jitter) } - .max(Duration::from_millis(50)) + .max(Duration::from_nanos(1)) } - // A value between -0.25s and 0.25s fn jitter(rng: &mut SeededRng) -> f32 { rng.random::() * JITTER_RANGE - (JITTER_RANGE / 2.0) } @@ -55,7 +165,7 @@ impl ExponentialBackoff { self.left = n; self.jitter = Self::jitter(rng); - self.rto = self.rto.mul(2); + self.rto = self.rto.saturating_mul(2); } pub fn can_retry(&self) -> bool { @@ -63,10 +173,288 @@ impl ExponentialBackoff { } } +/// Add a logical delay, reducing unrepresentable intervals until they fit. +pub fn deadline(now: Instant, mut delay: Duration) -> Instant { + loop { + if let Some(deadline) = now.checked_add(delay) { + return deadline; + } + delay /= 2; + } +} + #[cfg(test)] mod test { use super::*; + fn timers(retries: usize) -> (HandshakeTimers, SeededRng) { + let config = Config::builder() + .handshake_timeout(Duration::from_secs(2)) + .flight_start_rto(Duration::from_millis(20)) + .flight_retries(retries) + .build() + .expect("valid timing config"); + let mut rng = SeededRng::new(Some(42)); + (HandshakeTimers::new(&config, &mut rng), rng) + } + + #[test] + fn idle_and_unsent_flights_do_not_run_clocks() { + let (mut timers, mut rng) = timers(0); + let now = Instant::now(); + assert!(timers.poll_timeout(now) > now); + assert_eq!(timers.handle_timeout(now, &mut rng), Ok(false)); + timers.flight_sent(now); + assert_eq!(timers.flight, Timeout::Disabled); + timers.begin_flight(&mut rng); + let late = now + Duration::from_secs(100); + assert_eq!(timers.handle_timeout(late, &mut rng), Ok(false)); + assert!(timers.poll_timeout(late) > late); + assert_eq!(timers.handshake_deadline(), Timeout::Unarmed); + timers.start_handshake(late); + assert_eq!(timers.poll_timeout(late), late + Duration::from_secs(2)); + timers.flight_sent(late); + let retry_at = late + timers.rto(); + assert_eq!(timers.poll_timeout(late), retry_at); + assert_eq!(timers.handle_timeout(late, &mut rng), Ok(false)); + assert_eq!( + timers.handle_timeout(retry_at, &mut rng), + Err(TimeoutError::Handshake) + ); + } + + #[test] + fn flights_fragments_and_handoffs_preserve_deadline() { + let (mut timers, mut rng) = timers(1); + let now = Instant::now(); + timers.start_handshake(now); + let original = timers.handshake_deadline(); + timers.begin_flight(&mut rng); + timers.flight_sent(now); + let first_retry = timers.poll_timeout(now); + timers.flight_sent(now + Duration::from_millis(1)); + assert_eq!(timers.poll_timeout(now), first_retry); + assert_eq!(timers.handle_timeout(first_retry, &mut rng), Ok(true)); + timers.flight_sent(first_retry); + let last_retry = timers.poll_timeout(first_retry); + assert_eq!( + timers.handle_timeout(last_retry, &mut rng), + Err(TimeoutError::Handshake) + ); + timers.begin_flight(&mut rng); + assert_eq!( + timers.poll_timeout(last_retry), + now + Duration::from_secs(2) + ); + timers.flight_sent(last_retry); + assert!(timers.poll_timeout(last_retry) < now + Duration::from_secs(2)); + timers.start_handshake(last_retry); + assert_eq!(timers.handshake_deadline(), original); + timers.stop_flight(); + assert_eq!( + timers.poll_timeout(last_retry), + now + Duration::from_secs(2) + ); + + let config = Config::default(); + let mut inherited = HandshakeTimers::new(&config, &mut rng); + inherited.set_handshake_deadline(original); + inherited.start_handshake(last_retry); + assert_eq!( + inherited.poll_timeout(last_retry), + now + Duration::from_secs(2) + ); + assert_eq!( + inherited.handle_timeout(now + Duration::from_secs(2), &mut rng), + Err(TimeoutError::Connect) + ); + } + + #[test] + fn overall_deadline_wins_over_flight_deadline() { + let (mut timers, mut rng) = timers(usize::MAX); + let now = Instant::now(); + timers.start_handshake(now); + timers.begin_flight(&mut rng); + timers.backoff.rto = Duration::from_secs(10); + timers.flight_sent(now); + assert_eq!(timers.poll_timeout(now), now + Duration::from_secs(2)); + assert_eq!( + timers.handle_timeout(now + Duration::from_secs(2), &mut rng), + Err(TimeoutError::Connect) + ); + } + + #[test] + fn completion_disables_handshake_but_allows_key_update_flights() { + let (mut timers, mut rng) = timers(1); + let now = Instant::now(); + timers.start_handshake(now); + timers.begin_flight(&mut rng); + timers.flight_sent(now); + timers.finish_handshake(); + assert_eq!(timers.poll_timeout(now), now + timers.rto()); + timers.stop(); + let later = now + Duration::from_secs(100); + timers.start_handshake(later); + timers.flight_sent(later); + assert_eq!(timers.handshake_deadline(), Timeout::Disabled); + assert!(timers.poll_timeout(later) > later); + assert_eq!(timers.handle_timeout(later, &mut rng), Ok(false)); + timers.begin_flight(&mut rng); + timers.flight_sent(later); + assert_eq!(timers.poll_timeout(later), later + timers.rto()); + assert_eq!( + timers.handle_timeout(later + timers.rto(), &mut rng), + Ok(true) + ); + timers.stop_flight(); + assert!(timers.poll_timeout(later) > later); + } + + #[test] + fn deadline_overflow_does_not_panic() { + let now = Instant::now(); + assert_eq!(deadline(now, Duration::ZERO), now); + assert_eq!( + deadline(now, Duration::from_nanos(1)), + now + Duration::from_nanos(1) + ); + assert!(now.checked_add(Duration::MAX).is_none()); + assert!(deadline(now, Duration::MAX) > now); + } + + #[test] + fn duplicate_and_timed_resends_share_attempts() { + let (mut timers, mut rng) = timers(2); + let now = Instant::now(); + timers.start_handshake(now); + timers.begin_flight(&mut rng); + assert!(!timers.request_resend(&mut rng)); + assert_eq!(timers.backoff.left, 2); + timers.flight_sent(now); + let first_retry = timers.poll_timeout(now); + assert_eq!(timers.handle_timeout(first_retry, &mut rng), Ok(true)); + assert_eq!(timers.backoff.left, 1); + assert_eq!(timers.flight, Timeout::Unarmed); + timers.flight_sent(first_retry); + assert!(timers.request_resend(&mut rng)); + assert_eq!(timers.backoff.left, 0); + assert!(!timers.request_resend(&mut rng)); + assert_eq!( + timers.poll_timeout(first_retry), + now + Duration::from_secs(2) + ); + timers.flight_sent(first_retry); + let exhausted = timers.poll_timeout(first_retry); + assert_eq!( + timers.handle_timeout(exhausted, &mut rng), + Err(TimeoutError::Handshake) + ); + } + + #[test] + fn courtesy_resends_do_not_restart_disabled_timers() { + let (mut timers, mut rng) = timers(1); + let now = Instant::now(); + timers.start_handshake(now); + timers.begin_flight(&mut rng); + timers.flight_sent(now); + timers.stop(); + assert!(timers.request_resend(&mut rng)); + timers.flight_sent(now); + assert_eq!(timers.handshake_deadline(), Timeout::Disabled); + assert_eq!(timers.flight, Timeout::Disabled); + assert!(!timers.request_resend(&mut rng)); + assert_eq!( + timers.handle_timeout(now + Duration::from_secs(100), &mut rng), + Ok(false) + ); + } + + #[test] + fn proportional_jitter() { + let mut rng = SeededRng::new(Some(42)); + for rto in [ + Duration::from_nanos(4), + Duration::from_micros(100), + Duration::from_millis(20), + Duration::from_secs(1), + Duration::from_secs(100), + ] { + let mut exp = ExponentialBackoff::new(rto, 1, &mut rng); + exp.jitter = -0.25; + assert_eq!(exp.rto(), rto - rto / 4); + exp.jitter = 0.0; + assert_eq!(exp.rto(), rto); + exp.jitter = 0.25; + assert_eq!(exp.rto(), rto + rto / 4); + } + } + + #[test] + fn smallest_rto_stays_positive() { + let mut rng = SeededRng::new(Some(42)); + let mut exp = ExponentialBackoff::new(Duration::from_nanos(1), 1, &mut rng); + for jitter in [-0.25, 0.0, 0.25] { + exp.jitter = jitter; + assert_eq!(exp.rto(), Duration::from_nanos(1)); + } + } + + #[test] + fn overflow_saturates() { + let mut rng = SeededRng::new(Some(42)); + let mut exp = ExponentialBackoff::new(Duration::MAX, usize::MAX, &mut rng); + exp.jitter = 0.25; + assert_eq!(exp.rto(), Duration::MAX); + exp.attempt(&mut rng); + assert_eq!(exp.rto, Duration::MAX); + assert_eq!(exp.left, usize::MAX - 1); + exp.jitter = -0.25; + assert!(exp.rto() < Duration::MAX); + assert!(exp.rto() > Duration::MAX / 2); + } + + #[test] + fn zero_retries_and_reset() { + let mut rng = SeededRng::new(Some(42)); + let start = Duration::from_millis(20); + let mut exp = ExponentialBackoff::new(start, 0, &mut rng); + let initial = exp.rto(); + assert!(!exp.can_retry()); + exp.attempt(&mut rng); + assert_eq!(exp.rto(), initial); + exp.reset(&mut rng); + assert!(!exp.can_retry()); + assert_eq!(exp.rto, start); + + let mut exp = ExponentialBackoff::new(start, 2, &mut rng); + exp.attempt(&mut rng); + exp.attempt(&mut rng); + assert!(!exp.can_retry()); + exp.reset(&mut rng); + assert!(exp.can_retry()); + assert_eq!(exp.left, 2); + assert_eq!(exp.rto, start); + } + + #[test] + fn seeded_jitter_is_repeatable_and_bounded() { + let mut first_rng = SeededRng::new(Some(42)); + let mut second_rng = SeededRng::new(Some(42)); + let mut first = ExponentialBackoff::new(Duration::from_millis(20), 10, &mut first_rng); + let mut second = ExponentialBackoff::new(Duration::from_millis(20), 10, &mut second_rng); + for _ in 0..10 { + assert_eq!(first.rto(), second.rto()); + assert!((-0.25..0.25).contains(&first.jitter)); + assert!(first.rto() >= first.rto - first.rto / 4); + assert!(first.rto() <= first.rto + first.rto / 4); + first.attempt(&mut first_rng); + second.attempt(&mut second_rng); + } + } + #[test] fn attempts() { let mut rng = SeededRng::new(Some(42)); diff --git a/src/util.rs b/src/util.rs index 68efa90f..5331b809 100644 --- a/src/util.rs +++ b/src/util.rs @@ -1,7 +1,25 @@ +use std::ops::Range; + use arrayvec::ArrayVec; use nom::error::{ErrorKind, ParseError, make_error}; use nom::{Err, IResult, Input, Parser}; +/// Check parsed 24-bit fragment bounds, parsing the body only when complete. +pub fn accepts_client_hello_fragment<'a, Message>( + length: u32, + fragment_offset: u32, + fragment_length: u32, + buffer: &'a [u8], + range: Range, + parser: impl FnOnce(&'a [u8], usize) -> IResult<&'a [u8], Message>, +) -> bool { + length >= 42 + && fragment_length > 0 + && fragment_offset + fragment_length <= length + && (fragment_length < length + || parser(&buffer[range.clone()], range.start).is_ok_and(|(rest, _)| rest.is_empty())) +} + /// A combinator that parses items using the provided parser but only collects /// items that pass a filter predicate. Allows zero matches. #[inline(always)] @@ -117,3 +135,84 @@ where Ok((input.take_from(bound), res)) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn client_hello_invalid_bounds_skip_parser() { + for (length, offset, fragment_length) in [ + (0, 0, 0), + (41, 0, 41), + (42, 0, 0), + (42, 42, 1), + (42, 1, 42), + (0x00ff_ffff, 0x00ff_ffff, 0x00ff_ffff), + ] { + assert!(!accepts_client_hello_fragment::<()>( + length, + offset, + fragment_length, + &[], + 0..0, + |_, _| panic!("invalid fragment must not reach the body parser"), + )); + } + } + + #[test] + fn client_hello_incomplete_fragments_skip_parser() { + for (length, offset, fragment_length) in [ + (42, 0, 1), + (42, 41, 1), + (84, 21, 42), + (0x00ff_ffff, 0x00ff_fffe, 1), + ] { + assert!(accepts_client_hello_fragment::<()>( + length, + offset, + fragment_length, + &[], + 0..0, + |_, _| panic!("incomplete fragment must not reach the body parser"), + )); + } + } + + #[test] + fn client_hello_complete_body_requires_successful_full_parse() { + let buffer = [7; 50]; + let mut parsed = false; + assert!(accepts_client_hello_fragment( + 42, + 0, + 42, + &buffer, + 3..45, + |input, offset| { + parsed = true; + assert_eq!(input, &buffer[3..45]); + assert_eq!(offset, 3); + Ok((&input[input.len()..], ())) + }, + )); + assert!(parsed); + assert!(!accepts_client_hello_fragment( + 42, + 0, + 42, + &buffer, + 3..45, + |input, _| Ok((&input[1..], ())), + )); + assert!(!accepts_client_hello_fragment::<()>( + 42, + 0, + 42, + &buffer, + 3..45, + |input, _| Err(Err::Error(make_error(input, ErrorKind::Verify))), + )); + } +} diff --git a/tests/auto/main.rs b/tests/auto/main.rs index 195d12bd..dda594ab 100644 --- a/tests/auto/main.rs +++ b/tests/auto/main.rs @@ -1,4 +1,6 @@ mod common; mod cross_matrix; +#[cfg(feature = "rcgen")] mod handshake; +#[cfg(feature = "rcgen")] mod server_fallback; diff --git a/tests/dtls12/fragmentation.rs b/tests/dtls12/fragmentation.rs index b7a2de78..4b74de0d 100644 --- a/tests/dtls12/fragmentation.rs +++ b/tests/dtls12/fragmentation.rs @@ -2,10 +2,13 @@ use std::collections::VecDeque; use std::sync::Arc; -use std::time::{Duration, Instant}; +#[cfg(feature = "rcgen")] +use std::time::Duration; +use std::time::Instant; use dimpl::{Config, Dtls, Output}; +#[cfg(feature = "rcgen")] use crate::common::*; use crate::ossl_helper::{DtlsCertOptions, DtlsEvent, OsslDtlsCert}; diff --git a/tests/dtls12/handshake.rs b/tests/dtls12/handshake.rs index a6f19907..601d05ac 100644 --- a/tests/dtls12/handshake.rs +++ b/tests/dtls12/handshake.rs @@ -3,7 +3,9 @@ use std::sync::Arc; use std::time::{Duration, Instant}; -use dimpl::{Config, Dtls, SrtpProfile}; +use dimpl::Dtls; +#[cfg(feature = "rcgen")] +use dimpl::{Config, SrtpProfile}; use crate::common::*; diff --git a/tests/dtls12/main.rs b/tests/dtls12/main.rs index c77bc49d..f681d341 100644 --- a/tests/dtls12/main.rs +++ b/tests/dtls12/main.rs @@ -3,11 +3,14 @@ mod ossl_helper; mod common; mod crypto; +#[cfg(feature = "rcgen")] mod data; +#[cfg(feature = "rcgen")] mod edge; mod fragmentation; mod handshake; mod ossl; mod psk; +#[cfg(feature = "rcgen")] mod reorder; mod retransmit; diff --git a/tests/dtls12/retransmit.rs b/tests/dtls12/retransmit.rs index b1c9dde8..2c475340 100644 --- a/tests/dtls12/retransmit.rs +++ b/tests/dtls12/retransmit.rs @@ -172,22 +172,20 @@ fn dtls12_client_hello_retransmits_using_advertised_deadlines() { "client should emit ClientHello" ); - // Drop the initial flight and handle the timeout. - let arm_at = initial_output + let retransmit_at = initial_output .timeout - .expect("client should advertise a deadline"); + .expect("client should advertise a retransmission deadline"); + assert!(retransmit_at >= now + Duration::from_millis(750)); + assert!(retransmit_at <= now + Duration::from_millis(1250)); client - .handle_timeout(arm_at) - .expect("arm ClientHello retransmission"); - let after_timer_arming = drain_outputs(&mut client); + .handle_timeout(retransmit_at - Duration::from_nanos(1)) + .expect("before ClientHello retransmission"); + let before_retransmission = drain_outputs(&mut client); assert!( - after_timer_arming.packets.is_empty(), - "arming should not retransmit early" + before_retransmission.packets.is_empty(), + "must not retransmit early" ); - - let retransmit_at = after_timer_arming - .timeout - .expect("client should advertise a retransmission deadline"); + assert_eq!(before_retransmission.timeout, Some(retransmit_at)); client .handle_timeout(retransmit_at) .expect("retransmit ClientHello"); diff --git a/tests/dtls13/conformance.rs b/tests/dtls13/conformance.rs index 3719dc6e..1eac9fb2 100644 --- a/tests/dtls13/conformance.rs +++ b/tests/dtls13/conformance.rs @@ -526,3 +526,87 @@ fn server_retransmits_final_ack_for_retransmitted_client_final_flight() { "server must retransmit its final ACK when the client final flight is retransmitted" ); } + +#[test] +#[cfg(feature = "rcgen")] +fn server_final_ack_gets_fresh_duplicate_resend_budget() { + let client_config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(false) + .require_client_certificate(false) + .handshake_timeout(Duration::from_secs(60)) + .flight_start_rto(Duration::from_secs(5)) + .flight_retries(2) + .build() + .expect("client config"), + ); + let server_config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(false) + .require_client_certificate(false) + .handshake_timeout(Duration::from_secs(60)) + .flight_start_rto(Duration::from_millis(20)) + .flight_retries(1) + .build() + .expect("server config"), + ); + let client_cert = generate_self_signed_certificate().expect("gen client cert"); + let server_cert = generate_self_signed_certificate().expect("gen server cert"); + let now = Instant::now(); + let mut client = Dtls::new_13(client_config, client_cert, now); + client.set_active(true); + let mut server = Dtls::new_13(server_config, server_cert, now); + + client.handle_timeout(now).expect("start client"); + server.handle_timeout(now).expect("initialize server clock"); + let client_hello = drain_outputs(&mut client).packets; + deliver_packets(&client_hello, &mut server); + let first_server_flight = drain_outputs(&mut server); + assert!(!first_server_flight.packets.is_empty()); + + let server_retry_at = first_server_flight.timeout.expect("server retry"); + server + .handle_timeout(server_retry_at) + .expect("use the only server-flight retry"); + let retried_server_flight = drain_outputs(&mut server).packets; + assert!(!retried_server_flight.is_empty()); + + client + .handle_timeout(server_retry_at) + .expect("advance client clock"); + deliver_packets(&retried_server_flight, &mut client); + let client_final = drain_outputs(&mut client); + assert!(client_final.connected); + assert!(!client_final.packets.is_empty()); + + server + .handle_timeout(server_retry_at) + .expect("advance server clock"); + deliver_packets(&client_final.packets, &mut server); + let completion = drain_outputs(&mut server); + assert!(completion.connected); + assert!(!completion.packets.is_empty(), "server completion ACK"); + + let client_retry_at = client_final.timeout.expect("client final-flight retry"); + client + .handle_timeout(client_retry_at) + .expect("retransmit unacknowledged final flight"); + let retransmitted_final = drain_outputs(&mut client).packets; + assert!(!retransmitted_final.is_empty()); + + server + .handle_timeout(client_retry_at) + .expect("advance completed server clock"); + assert!( + drain_outputs(&mut server).packets.is_empty(), + "completion ACK must not retransmit on a timer" + ); + deliver_packets(&retransmitted_final, &mut server); + let replacement_ack = drain_outputs(&mut server).packets; + assert!( + !replacement_ack.is_empty(), + "server must retransmit its completion ACK with a fresh flight budget" + ); +} diff --git a/tests/dtls13/main.rs b/tests/dtls13/main.rs index 162e8215..ecbd94ef 100644 --- a/tests/dtls13/main.rs +++ b/tests/dtls13/main.rs @@ -1,3 +1,5 @@ +#![cfg(feature = "rcgen")] + #[cfg(not(windows))] #[path = "../wolfssl/mod.rs"] mod wolfssl_helper; diff --git a/tests/dtls13/retransmit.rs b/tests/dtls13/retransmit.rs index ad0226be..d1215857 100644 --- a/tests/dtls13/retransmit.rs +++ b/tests/dtls13/retransmit.rs @@ -628,6 +628,7 @@ fn dtls13_retransmit_exponential_backoff() { // Use enough retries to observe several backoff steps let config = Arc::new( Config::builder() + .dangerously_set_rng_seed(42) .flight_retries(6) .handshake_timeout(Duration::from_secs(300)) .build() @@ -686,16 +687,11 @@ fn dtls13_retransmit_exponential_backoff() { ); } - // Verify rough doubling: each timeout should be at least 1.5x the previous - // (accounting for jitter of +/- 0.25s) - for i in 1..timeouts.len() { - let ratio = timeouts[i].as_secs_f64() / timeouts[i - 1].as_secs_f64(); + for (attempt, timeout) in timeouts.iter().enumerate() { + let nominal = Duration::from_secs(1 << attempt); assert!( - ratio > 1.4, - "Timeout ratio {}/{} = {:.2} should be > 1.4 (exponential backoff)", - i, - i - 1, - ratio + *timeout >= nominal.mul_f64(0.75) && *timeout <= nominal.mul_f64(1.25), + "Timeout {attempt} ({timeout:?}) must be within +/-25% of {nominal:?}" ); } diff --git a/tests/ossl/io_buf.rs b/tests/ossl/io_buf.rs index f84daead..f81d6926 100644 --- a/tests/ossl/io_buf.rs +++ b/tests/ossl/io_buf.rs @@ -52,7 +52,7 @@ impl io::Read for IoBuffer { if max == self.incoming.len() { // The typical case is that the entire input is consumed at once, // which means the happy path is cheap. - self.incoming.truncate(0); + self.incoming.clear(); } else { // Shifting data inside a vector is not good. This should be rare. self.incoming.drain(..max); diff --git a/tests/timing.rs b/tests/timing.rs new file mode 100644 index 00000000..88c3ea8e --- /dev/null +++ b/tests/timing.rs @@ -0,0 +1,1373 @@ +#![cfg(feature = "rcgen")] + +use std::mem; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use dimpl::certificate::generate_self_signed_certificate; +use dimpl::{Config, Dtls, DtlsCertificate, Error, Output, ProtocolVersion, TimeoutError}; + +#[path = "dtls13/common.rs"] +mod common; + +use common::{DrainedOutputs, drain_outputs}; + +const VERSIONS: &[Version] = &[Version::Dtls12, Version::Dtls13, Version::Auto]; +const BUDGET: Duration = Duration::from_millis(100); +const LONG_RTO: Duration = Duration::from_secs(1); +const PAIRS: &[(Version, Version)] = &[ + (Version::Dtls12, Version::Dtls12), + (Version::Dtls13, Version::Dtls13), + (Version::Auto, Version::Dtls12), + (Version::Auto, Version::Dtls13), + (Version::Dtls12, Version::Auto), + (Version::Dtls13, Version::Auto), + (Version::Auto, Version::Auto), +]; + +#[derive(Clone, Copy, Debug)] +enum Version { + Dtls12, + Dtls13, + Auto, +} + +impl Version { + fn endpoint(self, config: Arc, certificate: DtlsCertificate, now: Instant) -> Dtls { + match self { + Self::Dtls12 => Dtls::new_12(config, certificate, now), + Self::Dtls13 => Dtls::new_13(config, certificate, now), + Self::Auto => Dtls::new_auto(config, certificate, now), + } + } +} + +fn config() -> Arc { + Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .handshake_timeout(BUDGET) + .flight_start_rto(LONG_RTO) + .flight_retries(0) + .build() + .expect("valid timing config"), + ) +} + +fn drain(endpoint: &mut Dtls) -> (Vec>, Instant) { + let mut packets = Vec::new(); + let mut buffer = vec![0; 65536]; + for _ in 0..100 { + match endpoint.poll_output(&mut buffer) { + Output::Packet(packet) => packets.push(packet.to_vec()), + Output::Timeout(deadline) => return (packets, deadline), + Output::BufferTooSmall { .. } => panic!("unexpected buffer requirement"), + _ => {} + } + } + panic!("poll cycle did not reach Timeout"); +} + +fn start_endpoint( + version: Version, + active: bool, + config: Arc, + certificate: &DtlsCertificate, + now: Instant, +) -> (Dtls, Vec>, Instant) { + let mut endpoint = version.endpoint(config.clone(), certificate.clone(), now); + endpoint.set_active(active); + endpoint.handle_timeout(now).expect("initialize clock"); + let (mut packets, mut deadline) = drain(&mut endpoint); + if !active { + let mut peer = version.endpoint(config, certificate.clone(), now); + peer.set_active(true); + peer.handle_timeout(now).expect("queue peer ClientHello"); + for packet in drain(&mut peer).0 { + endpoint.handle_packet(&packet).expect("accept ClientHello"); + let output = drain(&mut endpoint); + packets.extend(output.0); + deadline = output.1; + } + } + assert!(!packets.is_empty(), "{version:?}, active={active}"); + (endpoint, packets, deadline) +} + +fn merge_output(received: &mut DrainedOutputs, output: DrainedOutputs) { + received.connected |= output.connected; + received.app_data.extend(output.app_data); + received.packets.extend(output.packets); + received.timeout = output.timeout; +} + +fn deliver_queued(source: &mut DrainedOutputs, target: &mut Dtls, received: &mut DrainedOutputs) { + for packet in mem::take(&mut source.packets) { + target.handle_packet(&packet).expect("deliver packet"); + merge_output(received, drain_outputs(target)); + } +} + +fn exchange( + client: &mut Dtls, + server: &mut Dtls, + now: Instant, +) -> (DrainedOutputs, DrainedOutputs) { + let mut client_output = drain_outputs(client); + let mut server_output = drain_outputs(server); + for _ in 0..100 { + deliver_queued(&mut client_output, server, &mut server_output); + deliver_queued(&mut server_output, client, &mut client_output); + client.handle_timeout(now).expect("client pending progress"); + merge_output(&mut client_output, drain_outputs(client)); + server.handle_timeout(now).expect("server pending progress"); + merge_output(&mut server_output, drain_outputs(server)); + if client_output.packets.is_empty() && server_output.packets.is_empty() { + return (client_output, server_output); + } + } + panic!("packet exchange did not settle"); +} + +fn connected_pair( + versions: (Version, Version), + config: Arc, + certificate: &DtlsCertificate, + now: Instant, +) -> (Dtls, Dtls) { + let mut client = versions + .0 + .endpoint(config.clone(), certificate.clone(), now); + let mut server = versions.1.endpoint(config, certificate.clone(), now); + client.set_active(true); + client.handle_timeout(now).expect("start client"); + server.handle_timeout(now).expect("server clock"); + let output = exchange(&mut client, &mut server, now); + assert!(output.0.connected, "client {versions:?}"); + assert!(output.1.connected, "server {versions:?}"); + assert_eq!(client.protocol_version(), server.protocol_version()); + (client, server) +} + +fn assert_expires_at(endpoint: &mut Dtls, expected: Instant) { + for _ in 0..20 { + let deadline = drain(endpoint).1; + if deadline == expected { + assert_eq!( + endpoint.handle_timeout(deadline), + Err(Error::Timeout(TimeoutError::Connect)) + ); + return; + } + assert!(deadline < expected, "overall deadline moved later"); + endpoint + .handle_timeout(deadline) + .expect("retry budget remains"); + } + panic!("overall deadline was not advertised"); +} + +fn corrupt_client_hello_extension(hello: &mut [u8], extension_type: u16) { + let mut cursor = 25 + 34; + cursor += 1 + hello[cursor] as usize; + cursor += 1 + hello[cursor] as usize; + cursor += 2 + u16::from_be_bytes([hello[cursor], hello[cursor + 1]]) as usize; + cursor += 1 + hello[cursor] as usize; + cursor += 2; + while cursor + 4 <= hello.len() { + let kind = u16::from_be_bytes([hello[cursor], hello[cursor + 1]]); + let length = u16::from_be_bytes([hello[cursor + 2], hello[cursor + 3]]) as usize; + if kind == extension_type { + assert!(length >= 2); + hello[cursor + 4..cursor + 6].copy_from_slice(&u16::MAX.to_be_bytes()); + return; + } + cursor += 4 + length; + } + panic!("missing ClientHello extension {extension_type}"); +} + +#[test] +fn rejected_extensions_leave_server_idle_until_a_valid_client_hello() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(false) + .handshake_timeout(BUDGET) + .flight_start_rto(LONG_RTO) + .build() + .expect("config"), + ); + for &(client_version, server_version) in PAIRS { + for extension_type in [10, 14] { + let mut client = client_version.endpoint(config.clone(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("client clock"); + let (hello, _) = drain(&mut client); + assert_eq!(hello.len(), 1); + let mut malformed = hello[0].clone(); + corrupt_client_hello_extension(&mut malformed, extension_type); + + let mut server = server_version.endpoint(config.clone(), certificate.clone(), base); + server.handle_timeout(base).expect("server clock"); + drain(&mut server); + server + .handle_packet(&malformed) + .expect("discard malformed extension"); + let rejected = drain(&mut server); + assert!(rejected.0.is_empty()); + assert!( + rejected.1 > base + BUDGET, + "{client_version:?} -> {server_version:?}" + ); + + let reception = base + Duration::from_secs(1); + server + .handle_timeout(reception) + .expect("rejected input did not start the clock"); + drain(&mut server); + server + .handle_packet(&hello[0]) + .expect("valid ClientHello after rejected input"); + let accepted = drain(&mut server); + assert!( + !accepted.0.is_empty(), + "{client_version:?} -> {server_version:?}, extension {extension_type}: {server:?}" + ); + assert_eq!(accepted.1, reception + BUDGET); + assert_expires_at(&mut server, reception + BUDGET); + } + } +} + +#[test] +fn initial_poll_schedules_client_progress_without_spinning_idle_servers() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for &version in VERSIONS { + let mut endpoint = version.endpoint(config(), certificate.clone(), base); + let idle = drain(&mut endpoint); + assert!(idle.0.is_empty()); + assert!(idle.1 > base); + endpoint.set_active(true); + if !matches!(version, Version::Auto) { + let initial = drain(&mut endpoint); + assert!(initial.0.is_empty()); + assert_eq!(initial.1, base); + endpoint + .handle_timeout(initial.1) + .expect("initial client progress"); + } + let initial = drain(&mut endpoint); + assert!(!initial.0.is_empty()); + assert_eq!(initial.1, base + BUDGET); + endpoint.handle_timeout(base).expect("same logical instant"); + let after = drain(&mut endpoint); + assert!(after.0.is_empty()); + assert_eq!(after.1, initial.1); + } +} + +#[test] +fn handshake_completes_after_rejected_client_hello() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(false) + .handshake_timeout(BUDGET) + .build() + .expect("config"), + ); + for &(client_version, server_version) in PAIRS { + for extension_type in [10, 14] { + let mut client = client_version.endpoint(config.clone(), certificate.clone(), base); + let mut server = server_version.endpoint(config.clone(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("client clock"); + let (hello, _) = drain(&mut client); + assert_eq!(hello.len(), 1); + server.handle_timeout(base).expect("server clock"); + drain(&mut server); + let mut malformed = hello[0].clone(); + corrupt_client_hello_extension(&mut malformed, extension_type); + server + .handle_packet(&malformed) + .expect("rejected ClientHello"); + assert!(drain(&mut server).0.is_empty()); + server + .handle_packet(&hello[0]) + .expect("valid ClientHello retry"); + let output = exchange(&mut client, &mut server, base); + assert!( + output.0.connected, + "{client_version:?} -> {server_version:?}" + ); + assert!( + output.1.connected, + "{client_version:?} -> {server_version:?}" + ); + let later = base + Duration::from_secs(1); + client.handle_timeout(later).expect("completed client"); + server.handle_timeout(later).expect("completed server"); + exchange(&mut client, &mut server, later); + } + } +} + +#[test] +fn rejected_later_fragments_keep_first_reception_deadline() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(false) + .max_queue_rx(50) + .handshake_timeout(BUDGET) + .flight_start_rto(LONG_RTO) + .build() + .expect("config"), + ); + for &(client_version, server_version) in PAIRS { + let mut client = client_version.endpoint(config.clone(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("client clock"); + let (hello, _) = drain(&mut client); + assert_eq!(hello.len(), 1); + let mut malformed = hello[0].clone(); + corrupt_client_hello_extension(&mut malformed, 14); + let mut server = server_version.endpoint(config.clone(), certificate.clone(), base); + server.handle_timeout(base).expect("server clock"); + drain(&mut server); + assert!(malformed.len() > 25 + 32); + for (index, body) in malformed[25..].chunks(32).enumerate() { + if index == 1 { + server + .handle_timeout(base + Duration::from_millis(70)) + .expect("partial budget"); + drain(&mut server); + } + let mut fragment = malformed[..25].to_vec(); + fragment.extend_from_slice(body); + fragment[5..11].copy_from_slice(&(index as u64).to_be_bytes()[2..]); + fragment[11..13].copy_from_slice(&((12 + body.len()) as u16).to_be_bytes()); + fragment[19..22].copy_from_slice(&((index * 32) as u32).to_be_bytes()[1..]); + fragment[22..25].copy_from_slice(&(body.len() as u32).to_be_bytes()[1..]); + server + .handle_packet(&fragment) + .expect("accept fragment or reject completed extension"); + let output = drain(&mut server); + assert!(output.0.is_empty()); + assert_eq!( + output.1, + base + BUDGET, + "{client_version:?} -> {server_version:?}, fragment {index}" + ); + } + assert_expires_at(&mut server, base + BUDGET); + } +} + +#[test] +fn later_message_rejection_does_not_undo_accepted_client_hello() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(false) + .handshake_timeout(BUDGET) + .flight_start_rto(LONG_RTO) + .build() + .expect("config"), + ); + for &(client_version, server_version) in PAIRS { + let mut client = client_version.endpoint(config.clone(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("client clock"); + let (hello, _) = drain(&mut client); + assert_eq!(hello.len(), 1); + let mut packet = hello[0].clone(); + let mut malformed_certificate = packet[..13].to_vec(); + malformed_certificate[5..11].copy_from_slice(&1u64.to_be_bytes()[2..]); + malformed_certificate[11..13].copy_from_slice(&13u16.to_be_bytes()); + malformed_certificate.extend_from_slice(&[11, 0, 0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 0]); + packet.extend_from_slice(&malformed_certificate); + let mut server = server_version.endpoint(config.clone(), certificate.clone(), base); + server.handle_timeout(base).expect("server clock"); + drain(&mut server); + server + .handle_packet(&packet) + .expect("accept ClientHello and discard malformed certificate"); + let output = drain(&mut server); + assert!(!output.0.is_empty()); + assert_eq!( + output.1, + base + BUDGET, + "{client_version:?} -> {server_version:?}" + ); + assert_expires_at(&mut server, base + BUDGET); + } +} + +#[test] +fn clients_start_at_emission_not_construction_or_timeout() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for &version in VERSIONS { + let mut client = version.endpoint(config(), certificate.clone(), base); + let late = base + Duration::from_secs(60); + client + .handle_timeout(late) + .expect("idle server has no deadline"); + assert!(drain(&mut client).1 > late); + client.set_active(true); + client + .handle_timeout(late) + .expect("queue initial ClientHello"); + // Test-only polling exception (#161): keep ClientHello queued across a + // clock update to prove its timers start at emission, not construction. + let emission = late + Duration::from_secs(60); + client + .handle_timeout(emission) + .expect("unsent flight has no timer"); + let (packets, deadline) = drain(&mut client); + assert!(!packets.is_empty(), "{version:?}"); + assert_eq!(deadline, emission + BUDGET, "{version:?}"); + client + .handle_timeout(deadline - Duration::from_nanos(1)) + .expect("full budget remains"); + assert_eq!(drain(&mut client).1, deadline); + assert_eq!( + client.handle_timeout(deadline), + Err(Error::Timeout(TimeoutError::Connect)) + ); + } +} + +#[test] +fn too_small_output_does_not_start_or_exhaust_client_timers() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for &version in VERSIONS { + let mut client = version.endpoint(config(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("queue ClientHello"); + // Test-only polling exception (#161): retain the packet after each + // BufferTooSmall to verify neither timer runs before successful output. + for seconds in [1, 10, 100] { + client + .handle_timeout(base + Duration::from_secs(seconds)) + .expect("not emitted"); + assert!(matches!( + client.poll_output(&mut []), + Output::BufferTooSmall { .. } + )); + } + let emission = base + Duration::from_secs(100); + let (packets, deadline) = drain(&mut client); + assert_eq!(packets.len(), 1, "{version:?}"); + assert_eq!(deadline, emission + BUDGET, "{version:?}"); + assert_eq!( + client.handle_timeout(deadline), + Err(Error::Timeout(TimeoutError::Connect)) + ); + } +} + +#[test] +fn too_small_retry_output_does_not_start_or_exhaust_flight_timers() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let initial_rto = Duration::from_millis(20); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .handshake_timeout(Duration::from_secs(10)) + .flight_start_rto(initial_rto) + .flight_retries(1) + .build() + .expect("retry configuration"), + ); + for &version in VERSIONS { + for active in [false, true] { + let (mut endpoint, original, retry_at) = + start_endpoint(version, active, config.clone(), &certificate, base); + endpoint.handle_timeout(retry_at).expect("queue retry"); + assert!(matches!( + endpoint.poll_output(&mut []), + Output::BufferTooSmall { .. } + )); + + let emission = retry_at + Duration::from_secs(1); + endpoint + .handle_timeout(emission) + .expect("unemitted retry has no running timer"); + assert!(matches!( + endpoint.poll_output(&mut []), + Output::BufferTooSmall { .. } + )); + + let (retried, deadline) = drain(&mut endpoint); + assert_eq!( + retried.len(), + original.len(), + "{version:?}, active={active}" + ); + let nominal = initial_rto * 2; + assert!(deadline >= emission + nominal.mul_f64(0.75)); + assert!(deadline <= emission + nominal.mul_f64(1.25)); + } + } +} + +#[test] +fn servers_start_at_late_client_hello_reception() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for &version in VERSIONS { + let mut server = version.endpoint(config(), certificate.clone(), base); + for seconds in [1, 10, 100] { + let now = base + Duration::from_secs(seconds); + server.handle_timeout(now).expect("passive server"); + assert!(drain(&mut server).1 > now); + server + .handle_packet(&[0, 1, 2]) + .expect("malformed input is discarded"); + assert!(drain(&mut server).1 > now); + } + let reception = base + Duration::from_secs(100); + let mut client = version.endpoint(config(), certificate.clone(), reception); + client.set_active(true); + client.handle_timeout(reception).expect("queue ClientHello"); + for packet in drain(&mut client).0 { + server.handle_packet(&packet).expect("accept ClientHello"); + drain(&mut server); + } + let deadline = drain(&mut server).1; + assert_eq!(deadline, reception + BUDGET, "{version:?}"); + assert_eq!( + server.handle_timeout(deadline), + Err(Error::Timeout(TimeoutError::Connect)) + ); + } +} + +#[test] +fn configured_rto_jitter_and_exact_retry_counts_apply_to_every_role() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for initial_rto in [ + Duration::from_nanos(1), + Duration::from_micros(80), + Duration::from_millis(20), + ] { + for retries in [0, 1, 3] { + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .handshake_timeout(Duration::from_secs(3600)) + .flight_start_rto(initial_rto) + .flight_retries(retries) + .build() + .expect("valid retry configuration"), + ); + for &version in VERSIONS { + for active in [false, true] { + let (mut endpoint, original, mut deadline) = + start_endpoint(version, active, config.clone(), &certificate, base); + let mut now = base; + for attempt in 0..=retries { + let nominal = initial_rto * (1 << attempt); + let interval = deadline.duration_since(now); + assert!(interval >= nominal.mul_f64(0.75).max(Duration::from_nanos(1))); + assert!(interval <= nominal.mul_f64(1.25).max(Duration::from_nanos(1))); + endpoint + .handle_timeout(deadline - Duration::from_nanos(1)) + .expect("not yet due"); + let before = drain(&mut endpoint); + assert!(before.0.is_empty()); + assert_eq!(before.1, deadline); + if attempt == retries { + let reason = if active && matches!(version, Version::Auto) { + TimeoutError::HybridClientHello + } else { + TimeoutError::Handshake + }; + assert_eq!( + endpoint.handle_timeout(deadline), + Err(Error::Timeout(reason)) + ); + } else { + endpoint.handle_timeout(deadline).expect("retry available"); + now = deadline; + let output = drain(&mut endpoint); + assert_eq!( + output.0.len(), + original.len(), + "{version:?}, active={active}" + ); + deadline = output.1; + } + } + } + } + } + } +} + +#[test] +fn overall_deadline_precedes_long_rto_with_generous_retries() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .handshake_timeout(BUDGET) + .flight_start_rto(LONG_RTO) + .flight_retries(100) + .build() + .expect("valid timing config"), + ); + for &version in VERSIONS { + for active in [false, true] { + let (mut endpoint, _, deadline) = + start_endpoint(version, active, config.clone(), &certificate, base); + assert_eq!(deadline, base + BUDGET, "{version:?}, active={active}"); + assert_eq!( + endpoint.handle_timeout(deadline), + Err(Error::Timeout(TimeoutError::Connect)) + ); + } + } +} + +#[test] +fn duplicate_client_hellos_share_the_flight_retry_budget() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let initial_rto = Duration::from_millis(20); + for use_cookie in [false, true] { + for (retries, timed_retries) in [(0, 0), (1, 0), (2, 1)] { + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(use_cookie) + .flight_start_rto(initial_rto) + .flight_retries(retries) + .handshake_timeout(Duration::from_secs(1)) + .build() + .expect("retry config"), + ); + for &(client_version, server_version) in PAIRS { + let mut client = client_version.endpoint(config.clone(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("client clock"); + let (hello, _) = drain(&mut client); + assert_eq!(hello.len(), 1); + let mut server = server_version.endpoint(config.clone(), certificate.clone(), base); + server.handle_timeout(base).expect("server clock"); + drain(&mut server); + server + .handle_packet(&hello[0]) + .expect("initial ClientHello"); + let (original, mut retry_at) = drain(&mut server); + assert!(!original.is_empty()); + let mut now = base; + for _ in 0..timed_retries { + server.handle_timeout(retry_at).expect("timer retry"); + now = retry_at; + let output = drain(&mut server); + assert_eq!(output.0.len(), original.len()); + retry_at = output.1; + } + let remaining = retries - timed_retries; + for duplicate in 0..remaining + 2 { + now += Duration::from_millis(1); + server.handle_timeout(now).expect("before retry timer"); + assert!(drain(&mut server).0.is_empty()); + server + .handle_packet(&hello[0]) + .expect("duplicate ClientHello"); + let output = drain(&mut server); + let expected_packets = if duplicate < remaining { + original.len() + } else { + 0 + }; + assert_eq!( + output.0.len(), + expected_packets, + "{client_version:?} -> {server_version:?}, retries={retries}, duplicate={duplicate}" + ); + if duplicate < remaining { + let nominal = initial_rto * (1 << (timed_retries + duplicate + 1)); + assert!(output.1 >= now + nominal.mul_f64(0.75)); + assert!(output.1 <= now + nominal.mul_f64(1.25)); + retry_at = output.1; + } else { + assert_eq!(output.1, retry_at); + } + } + assert_eq!( + server.handle_timeout(retry_at), + Err(Error::Timeout(TimeoutError::Handshake)) + ); + } + } + } +} + +#[test] +fn delayed_auto_client_selection_keeps_only_the_original_remaining_budget() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for peer_version in [Version::Dtls12, Version::Dtls13] { + let mut client = Version::Auto.endpoint(config(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("initial clock"); + let (hello, original_deadline) = drain(&mut client); + assert_eq!(client.protocol_version(), None); + let selection = base + Duration::from_millis(70); + client + .handle_timeout(selection) + .expect("remaining initial budget"); + assert_eq!(drain(&mut client).1, original_deadline); + let mut server = peer_version.endpoint(config(), certificate.clone(), selection); + server.handle_timeout(selection).expect("server clock"); + drain(&mut server); + let mut responses = Vec::new(); + for packet in hello { + server + .handle_packet(&packet) + .expect("peer accepts hybrid ClientHello"); + responses.extend(drain(&mut server).0); + } + assert!(!responses.is_empty()); + let mut second_hello = Vec::new(); + for packet in responses { + client.handle_packet(&packet).expect("version handoff"); + let output = drain(&mut client); + second_hello.extend(output.0); + assert_eq!(output.1, original_deadline); + } + assert!(!second_hello.is_empty()); + let expected = match peer_version { + Version::Dtls12 => ProtocolVersion::DTLS1_2, + Version::Dtls13 => ProtocolVersion::DTLS1_3, + Version::Auto => unreachable!(), + }; + assert_eq!(client.protocol_version(), Some(expected)); + assert_eq!( + client.handle_timeout(original_deadline), + Err(Error::Timeout(TimeoutError::Connect)) + ); + } +} + +#[test] +fn delayed_auto_client_without_cookie_keeps_deadline_while_waiting_for_server_flight() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for (peer_version, expected_version) in [ + (Version::Dtls12, ProtocolVersion::DTLS1_2), + (Version::Dtls13, ProtocolVersion::DTLS1_3), + ] { + let mut client = Version::Auto.endpoint(config(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("client clock"); + let (hello, original_deadline) = drain(&mut client); + let selection = base + Duration::from_millis(70); + client + .handle_timeout(selection) + .expect("remaining initial budget"); + assert_eq!(drain(&mut client).1, original_deadline); + + let server_config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(false) + .mtu(128) + .max_queue_tx(30) + .build() + .expect("fragmented server flight"), + ); + let mut server = peer_version.endpoint(server_config, certificate.clone(), selection); + server.handle_timeout(selection).expect("server clock"); + drain(&mut server); + let mut responses = Vec::new(); + for packet in hello { + server.handle_packet(&packet).expect("hybrid ClientHello"); + responses.extend(drain(&mut server).0); + } + assert!(responses.len() > 1, "withhold part of the server flight"); + client + .handle_packet(&responses[0]) + .expect("version selection without a cookie"); + let output = drain_outputs(&mut client); + assert!(!output.connected); + assert!(output.packets.is_empty()); + assert_eq!(client.protocol_version(), Some(expected_version)); + assert_eq!(output.timeout, Some(original_deadline)); + assert_expires_at(&mut client, original_deadline); + } +} + +#[test] +fn auto_handoff_preserves_outstanding_flight_retries() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let initial_rto = Duration::from_millis(20); + for peer_version in [Version::Dtls12, Version::Dtls13] { + for (retries, used) in [(0, 0), (2, 0), (2, 1)] { + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .flight_start_rto(initial_rto) + .flight_retries(retries) + .handshake_timeout(Duration::from_secs(1)) + .build() + .expect("client config"), + ); + let mut client = Version::Auto.endpoint(config, certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("client clock"); + let (mut hello, mut retry_at) = drain(&mut client); + let mut now = base; + for _ in 0..used { + client.handle_timeout(retry_at).expect("pending Auto retry"); + now = retry_at; + (hello, retry_at) = drain(&mut client); + } + assert_eq!(hello.len(), 1); + let selection = now + Duration::from_millis(1); + client + .handle_timeout(selection) + .expect("before outstanding retry"); + assert_eq!(drain(&mut client).1, retry_at); + let server_config = Arc::new( + Config::builder() + .use_server_cookie(false) + .mtu(128) + .max_queue_tx(30) + .build() + .expect("fragmented server config"), + ); + let mut server = peer_version.endpoint(server_config, certificate.clone(), selection); + server.handle_timeout(selection).expect("server clock"); + drain(&mut server); + server.handle_packet(&hello[0]).expect("hybrid ClientHello"); + let (responses, _) = drain(&mut server); + assert!(responses.len() > 1); + client + .handle_packet(&responses[0]) + .expect("partial server flight handoff"); + assert_eq!( + drain(&mut client).1, + retry_at, + "{peer_version:?}, retries={retries}, used={used}" + ); + for attempt in used..retries { + client + .handle_timeout(retry_at) + .expect("remaining retry after handoff"); + let sent_at = retry_at; + let output = drain(&mut client); + assert_eq!(output.0.len(), 1); + assert_eq!(&output.0[0][13..], &hello[0][13..]); + let nominal = initial_rto * (1 << (attempt + 1)); + assert!(output.1 >= sent_at + nominal.mul_f64(0.75)); + assert!(output.1 <= sent_at + nominal.mul_f64(1.25)); + retry_at = output.1; + } + assert_eq!( + client.handle_timeout(retry_at), + Err(Error::Timeout(TimeoutError::Handshake)) + ); + } + } +} + +#[test] +fn first_client_fragment_starts_deadline_and_later_emissions_do_not_refresh_it() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .mtu(64) + .handshake_timeout(BUDGET) + .flight_start_rto(LONG_RTO) + .flight_retries(0) + .build() + .expect("fragmented configuration"), + ); + for version in [Version::Dtls12, Version::Dtls13] { + let mut client = version.endpoint(config.clone(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("queue fragments"); + // Test-only polling exception (#161): leave the first fragment pending + // after BufferTooSmall while advancing the logical clock. + assert!(matches!( + client.poll_output(&mut []), + Output::BufferTooSmall { .. } + )); + let emission = base + Duration::from_secs(60); + client + .handle_timeout(emission) + .expect("no emitted fragments yet"); + let mut buffer = [0; 64]; + assert!(matches!(client.poll_output(&mut buffer), Output::Packet(_))); + // Test-only polling exception (#161): pause after the first fragment + // to verify later fragments cannot refresh the original deadline. + client + .handle_timeout(emission + Duration::from_millis(70)) + .expect("first fragment has full budget"); + let (remaining, deadline) = drain(&mut client); + assert!(!remaining.is_empty()); + assert_eq!(deadline, emission + BUDGET); + assert_eq!( + client.handle_timeout(deadline), + Err(Error::Timeout(TimeoutError::Connect)) + ); + } +} + +#[test] +fn server_fragments_duplicates_and_auto_fallback_preserve_first_reception() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .mtu(64) + .handshake_timeout(BUDGET) + .flight_start_rto(LONG_RTO) + .flight_retries(0) + .build() + .expect("fragmented configuration"), + ); + for (client_version, server_version) in [ + (Version::Dtls12, Version::Dtls12), + (Version::Dtls13, Version::Dtls13), + (Version::Dtls12, Version::Auto), + (Version::Dtls13, Version::Auto), + ] { + for reverse in [false, true] { + for complete in [false, true] { + let reception = base + Duration::from_secs(60); + let mut client = + client_version.endpoint(config.clone(), certificate.clone(), reception); + client.set_active(true); + client + .handle_timeout(reception) + .expect("queue fragmented ClientHello"); + let mut fragments = drain(&mut client).0; + assert!(fragments.len() > 1); + if reverse { + fragments.reverse(); + } + let mut server = server_version.endpoint(config.clone(), certificate.clone(), base); + server + .handle_timeout(reception) + .expect("late passive server"); + assert!(drain(&mut server).1 > reception + BUDGET); + server + .handle_packet(&fragments[0]) + .expect("accept first arriving fragment"); + let first_output = drain(&mut server); + assert!(first_output.0.is_empty()); + assert_eq!(first_output.1, reception + BUDGET); + server + .handle_timeout(reception + Duration::from_millis(20)) + .expect("incomplete assembly"); + drain(&mut server); + server + .handle_packet(&fragments[0]) + .expect("duplicate fragment"); + assert_eq!(drain(&mut server).1, first_output.1); + if complete { + server + .handle_timeout(reception + Duration::from_millis(70)) + .expect("remaining assembly budget"); + drain(&mut server); + let mut responses = Vec::new(); + for fragment in &fragments[1..] { + server + .handle_packet(fragment) + .expect("complete ClientHello"); + let output = drain(&mut server); + responses.extend(output.0); + assert_eq!(output.1, first_output.1); + } + assert!(!responses.is_empty()); + if matches!(server_version, Version::Auto) + && matches!(client_version, Version::Dtls13) + { + assert_eq!(server.protocol_version(), None); + } else { + let expected = if matches!(client_version, Version::Dtls12) { + ProtocolVersion::DTLS1_2 + } else { + ProtocolVersion::DTLS1_3 + }; + assert_eq!(server.protocol_version(), Some(expected)); + } + } + assert_eq!( + server.handle_timeout(first_output.1), + Err(Error::Timeout(TimeoutError::Connect)) + ); + } + } + } +} + +#[test] +fn rejected_client_hellos_and_unrelated_input_do_not_start_server_clocks() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for &version in VERSIONS { + let (_, hello, _) = start_endpoint(version, true, config(), &certificate, base); + assert_eq!(hello.len(), 1); + let hello = &hello[0]; + let mut inputs = vec![("unrelated bytes", vec![0, 1, 2])]; + + let mut empty = hello[..25].to_vec(); + empty[11..13].copy_from_slice(&12u16.to_be_bytes()); + empty[14..17].fill(0); + empty[22..25].fill(0); + inputs.push(("empty ClientHello", empty)); + + let mut zero_fragment = hello[..25].to_vec(); + zero_fragment[11..13].copy_from_slice(&12u16.to_be_bytes()); + zero_fragment[22..25].fill(0); + inputs.push(("zero-length fragment", zero_fragment)); + + let mut outside_message = hello.clone(); + outside_message[19..22].fill(0xff); + inputs.push(("fragment outside message", outside_message)); + + let mut malformed_body = hello.clone(); + assert_eq!(&malformed_body[59..61], &[0, 0]); + malformed_body[61..63].fill(0xff); + inputs.push(("malformed complete body", malformed_body)); + + let mut future_message = hello.clone(); + future_message[17..19].copy_from_slice(&1u16.to_be_bytes()); + inputs.push(("unexpected handshake sequence", future_message)); + + let mut encrypted = hello.clone(); + encrypted[0] = 23; + encrypted[3..5].copy_from_slice(&1u16.to_be_bytes()); + inputs.push(("unrelated encrypted application data", encrypted)); + + let mut ccs = hello[..13].to_vec(); + ccs[0] = 20; + ccs[11..13].copy_from_slice(&1u16.to_be_bytes()); + ccs.push(1); + inputs.push(("unrelated ChangeCipherSpec", ccs)); + + for (label, input) in inputs { + let mut server = version.endpoint(config(), certificate.clone(), base); + server.handle_timeout(base).expect("passive server"); + drain(&mut server); + server + .handle_packet(&input) + .expect("discard or defer unrelated input"); + let output = drain(&mut server); + assert!(output.0.is_empty(), "{version:?}: {label}"); + assert!( + output.1 > base + BUDGET, + "{version:?}: {label} started the clock" + ); + let reception = base + Duration::from_secs(60); + server + .handle_timeout(reception) + .expect("rejected input did not arm timers"); + drain(&mut server); + } + } +} + +#[test] +fn cookie_exchanges_and_fresh_flights_reset_only_retry_state() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let rto = Duration::from_millis(20); + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .handshake_timeout(BUDGET) + .flight_start_rto(rto) + .flight_retries(8) + .build() + .expect("valid cookie configuration"), + ); + for &(client_version, server_version) in PAIRS { + let mut client = client_version.endpoint(config.clone(), certificate.clone(), base); + let mut server = server_version.endpoint(config.clone(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("start client"); + server.handle_timeout(base).expect("server clock"); + drain(&mut server); + let (hello, client_retry) = drain(&mut client); + let mut cookie_packets = Vec::new(); + let mut server_retry = base; + for packet in hello { + server.handle_packet(&packet).expect("initial ClientHello"); + let output = drain(&mut server); + cookie_packets.extend(output.0); + server_retry = output.1; + } + assert!(!cookie_packets.is_empty()); + client + .handle_timeout(client_retry) + .expect("retry ClientHello"); + assert!(!drain(&mut client).0.is_empty()); + server + .handle_timeout(server_retry) + .expect("retry cookie challenge"); + assert!(!drain(&mut server).0.is_empty()); + let now = client_retry.max(server_retry) + Duration::from_millis(1); + client + .handle_timeout(now) + .expect("client clock before challenge"); + server + .handle_timeout(now) + .expect("server clock before challenge response"); + drain(&mut client); + drain(&mut server); + + let mut cookie_hello = Vec::new(); + let mut next_client = now; + for packet in &cookie_packets { + client.handle_packet(packet).expect("cookie challenge"); + let output = drain(&mut client); + cookie_hello.extend(output.0); + next_client = output.1; + } + assert!(!cookie_hello.is_empty()); + assert!(next_client >= now + rto.mul_f64(0.75)); + assert!(next_client <= now + rto.mul_f64(1.25)); + for packet in cookie_packets { + client + .handle_packet(&packet) + .expect("duplicate cookie challenge"); + let output = drain(&mut client); + if client.protocol_version() == Some(ProtocolVersion::DTLS1_2) { + assert!(!output.0.is_empty()); + assert!(output.1 >= now + (rto * 2).mul_f64(0.75)); + assert!(output.1 <= now + (rto * 2).mul_f64(1.25)); + } else { + assert!(output.0.is_empty()); + assert_eq!(output.1, next_client); + } + } + + let mut server_flight = Vec::new(); + let mut next_server = now; + for packet in cookie_hello { + server + .handle_packet(&packet) + .expect("ClientHello with cookie"); + let output = drain(&mut server); + server_flight.extend(output.0); + next_server = output.1; + } + assert!(!server_flight.is_empty()); + assert!(next_server >= now + rto.mul_f64(0.75)); + assert!(next_server <= now + rto.mul_f64(1.25)); + assert_expires_at(&mut client, base + BUDGET); + assert_expires_at(&mut server, base + BUDGET); + } +} + +#[test] +fn completed_handshakes_stay_connected_after_deadline_and_key_updates() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for use_cookie in [false, true] { + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(use_cookie) + .handshake_timeout(BUDGET) + .flight_start_rto(Duration::from_millis(20)) + .flight_retries(3) + .aead_encryption_limit(3) + .build() + .expect("small key-update threshold"), + ); + for &versions in PAIRS { + let (mut client, mut server) = + connected_pair(versions, config.clone(), &certificate, base); + for round in 0..6 { + let now = base + Duration::from_secs(60 + round); + client + .handle_timeout(now) + .expect("completed client deadline disabled"); + server + .handle_timeout(now) + .expect("completed server deadline disabled"); + let idle = exchange(&mut client, &mut server, now); + assert!(idle.0.timeout.expect("client timeout") > now); + assert!(idle.1.timeout.expect("server timeout") > now); + + client + .send_application_data(b"client traffic") + .expect("client send"); + let received = exchange(&mut client, &mut server, now); + assert_eq!(received.1.app_data, [b"client traffic".to_vec()]); + server + .send_application_data(b"server traffic") + .expect("server send"); + let received = exchange(&mut client, &mut server, now); + assert_eq!(received.0.app_data, [b"server traffic".to_vec()]); + assert!(received.0.timeout.expect("client timeout") > now); + assert!(received.1.timeout.expect("server timeout") > now); + } + } + } +} + +#[test] +fn key_update_uses_configured_retries_without_restarting_handshake_budget() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + let rto = Duration::from_millis(20); + for retries in [0, 2] { + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .handshake_timeout(Duration::from_millis(5)) + .flight_start_rto(rto) + .flight_retries(retries) + .aead_encryption_limit(3) + .build() + .expect("small handshake budget and key-update threshold"), + ); + for versions in [ + (Version::Dtls13, Version::Dtls13), + (Version::Auto, Version::Dtls13), + (Version::Dtls13, Version::Auto), + (Version::Auto, Version::Auto), + ] { + for client_sends in [false, true] { + let (mut client, mut server) = + connected_pair(versions, config.clone(), &certificate, base); + let now = base + Duration::from_secs(60); + client + .handle_timeout(now) + .expect("client idle after completion"); + server + .handle_timeout(now) + .expect("server idle after completion"); + exchange(&mut client, &mut server, now); + let sender = if client_sends { + &mut client + } else { + &mut server + }; + let mut next_retry = None; + for _ in 0..3 { + sender + .send_application_data(b"trigger key update") + .expect("application data"); + assert!(!drain(sender).0.is_empty()); + sender.handle_timeout(now).expect("trigger KeyUpdate"); + let output = drain(sender); + if output.1 < now + Duration::from_secs(1) { + assert!(!output.0.is_empty()); + next_retry = Some(output.1); + break; + } + } + let mut deadline = + next_retry.expect("AEAD threshold must trigger a KeyUpdate flight"); + let mut sent_at = now; + for attempt in 0..=retries { + let nominal = rto * (1 << attempt); + assert!(deadline >= sent_at + nominal.mul_f64(0.75)); + assert!(deadline <= sent_at + nominal.mul_f64(1.25)); + if attempt == retries { + assert_eq!( + sender.handle_timeout(deadline), + Err(Error::Timeout(TimeoutError::Handshake)) + ); + } else { + sender.handle_timeout(deadline).expect("KeyUpdate retry"); + sent_at = deadline; + let output = drain(sender); + assert!(!output.0.is_empty()); + deadline = output.1; + } + } + } + } + } +} + +#[test] +fn server_deadline_survives_client_certificate_until_finished() { + let base = Instant::now(); + let certificate = generate_self_signed_certificate().expect("certificate"); + for use_cookie in [false, true] { + let config = Arc::new( + Config::builder() + .dangerously_set_rng_seed(42) + .use_server_cookie(use_cookie) + .mtu(128) + .max_queue_tx(30) + .handshake_timeout(BUDGET) + .flight_start_rto(LONG_RTO) + .flight_retries(100) + .build() + .expect("fragmented certificate configuration"), + ); + for (client_version, server_version) in [ + (Version::Dtls13, Version::Dtls13), + (Version::Auto, Version::Dtls13), + (Version::Dtls13, Version::Auto), + (Version::Auto, Version::Auto), + ] { + let mut client = client_version.endpoint(config.clone(), certificate.clone(), base); + let mut server = server_version.endpoint(config.clone(), certificate.clone(), base); + client.set_active(true); + client.handle_timeout(base).expect("start client"); + server.handle_timeout(base).expect("server clock"); + let mut client_output = drain_outputs(&mut client); + let mut server_output = drain_outputs(&mut server); + let mut saw_certificate = false; + 'handshake: for _ in 0..20 { + for packet in mem::take(&mut client_output.packets) { + server + .handle_packet(&packet) + .expect("receive client flight"); + let output = drain_outputs(&mut server); + if output.peer_cert.is_some() { + assert!(!output.connected, "Finished must still be withheld"); + assert_eq!(output.timeout, Some(base + BUDGET)); + saw_certificate = true; + break 'handshake; + } + merge_output(&mut server_output, output); + } + deliver_queued(&mut server_output, &mut client, &mut client_output); + } + assert!( + saw_certificate, + "must stop between Certificate and Finished" + ); + server + .handle_timeout(base + Duration::from_millis(70)) + .expect("remaining handshake budget"); + assert_eq!(drain(&mut server).1, base + BUDGET); + assert_expires_at(&mut server, base + BUDGET); + } + } +}