diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index a674ad8..592e3ce 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -4,6 +4,9 @@ use std::{ time::{Duration, Instant}, }; +#[cfg(test)] +use std::cell::Cell; + #[cfg(test)] use irtt_proto::{flags, Params, TimestampFields, PROTOCOL_VERSION}; @@ -12,7 +15,10 @@ use crate::{ error::ClientError, event::{ClientEvent, OpenOutcome}, receive::recv_datagram, - session::machine::{recv_buffer_size, ProbeSent, SessionMachine, MAX_OPEN_PACKET_SIZE}, + session::machine::{ + recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, ProbeSent, + SessionMachine, MAX_OPEN_PACKET_SIZE, + }, socket::{connect_udp_socket, resolve_remote, validate_open_timeouts}, socket_options::{apply_dscp_to_socket, clear_dscp_on_socket}, timing::ClientTimestamp, @@ -28,6 +34,34 @@ enum ProbeScheduleMode { Managed { scheduled_at: Instant }, } +#[derive(Debug)] +struct PreparedClientOpen { + machine: PreparedOpenAcceptance, + schedule: Option, + recv_buffer_len: Option, + negotiated_dscp: Option, + previous_dscp: Option, + post_open_recv_timeout: Option, +} + +#[derive(Debug)] +struct PreparedClientOpenFailure { + primary: ClientError, + machine: PreparedOpenAcceptance, +} + +#[cfg(test)] +#[derive(Debug, Default)] +struct ClientTestHooks { + fail_open_dscp: Cell, + fail_open_timeout_restore: Cell, + fail_cleanup_send: Cell, + recv_buffer_len_override: Cell>, + fail_dscp_restore: Cell, + last_restored_read_timeout: Cell>, + prepared_active_session_before_dscp: Cell, +} + #[cfg(test)] #[derive(Debug, Clone, Copy)] pub(crate) struct ProbeSendTimestamps { @@ -60,6 +94,9 @@ pub struct Client { socket: UdpSocket, remote: SocketAddr, recv_buffer: Vec, + applied_dscp: Option, + #[cfg(test)] + test_hooks: ClientTestHooks, #[cfg(test)] probe_reported_len: Option, #[cfg(test)] @@ -86,6 +123,9 @@ impl Client { socket, remote, recv_buffer: vec![0_u8; recv_buffer_size(false, None)?], + applied_dscp: None, + #[cfg(test)] + test_hooks: ClientTestHooks::default(), #[cfg(test)] probe_reported_len: None, #[cfg(test)] @@ -99,66 +139,24 @@ impl Client { /// /// On success, returns the negotiated open outcome and transitions the /// client into either an open probe session or completed no-test state. - /// Open attempts use [`ClientConfig::open_timeouts`]. + /// Open attempts use [`ClientConfig::open_timeouts`]. Malformed, unrelated, + /// or unauthenticated datagrams are ignored until the current attempt's + /// absolute deadline, so one attempt may consume several datagrams without + /// retransmitting. Silence or ignored traffic eventually produces + /// [`ClientError::OpenTimeout`], while authenticated incompatibility remains + /// terminal. + /// + /// When a trusted reply allocates a token but later negotiation or socket + /// preparation fails, the client sends a best-effort cleanup close and + /// preserves the original failure. Failed opening never leaves the session + /// machine open. pub fn open(&mut self) -> Result { - let outcome = (|| -> Result { - let packet = self.runtime.open_packet()?; - let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE]; - - for timeout in &self.runtime.config().open_timeouts { - self.socket.set_read_timeout(Some(*timeout))?; - self.socket.send(&packet)?; - - match self.socket.recv(&mut buf) { - Ok(size) => { - let reply = self.runtime.decode_open_reply(&buf[..size])?; - let opened_at = ClientTimestamp::now(); - let remote = self.remote; - let has_hmac = self.runtime.has_hmac(); - let socket = &self.socket; - let recv_buffer = &mut self.recv_buffer; - let mut schedule = None; - - let outcome = - self.runtime - .accept_open_reply(reply, opened_at, |negotiated| { - let size = recv_buffer_size(has_hmac, Some(negotiated))?; - let negotiated_dscp = u8::try_from(negotiated.params.dscp) - .map_err(|_| ClientError::InvalidConfig { - reason: "negotiated dscp must be in range 0..=63" - .to_owned(), - })?; - apply_dscp_to_socket(socket, remote, negotiated_dscp)?; - recv_buffer.resize(size, 0); - schedule = - Some(ProbeSchedule::new(opened_at.mono, negotiated)?); - Ok(()) - })?; - self.schedule = schedule; - return Ok(outcome); - } - Err(err) - if matches!( - err.kind(), - io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut - ) => {} - Err(err) => return Err(ClientError::Socket(err)), - } - } - - Err(ClientError::OpenTimeout) - })(); - - let restore = self - .socket - .set_read_timeout(self.runtime.config().socket_config.recv_timeout); - - match (outcome, restore) { - (Ok(outcome), Ok(())) => Ok(outcome), - (Ok(_), Err(source)) => Err(ClientError::ReadTimeoutRestore { source }), - (Err(err), Ok(())) => Err(err), - (Err(err), Err(_)) => Err(err), + let result = self.open_transaction(); + if result.is_err() { + let _ = + self.restore_open_read_timeout(self.runtime.config().socket_config.recv_timeout); } + result } /// Send a close request and emit a [`ClientEvent::SessionClosed`] event. @@ -417,6 +415,231 @@ impl Client { )); Ok(events) } + + fn open_transaction(&mut self) -> Result { + let request = self.runtime.prepare_open_request()?; + let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE]; + let attempt_count = self.runtime.config().open_timeouts.len(); + + for attempt in 0..attempt_count { + let timeout = self.runtime.config().open_timeouts[attempt]; + let deadline = Instant::now() + .checked_add(timeout) + .ok_or(ClientError::DurationOverflow)?; + self.socket.set_read_timeout(Some(timeout))?; + self.socket.send(&request.bytes)?; + + loop { + let now = Instant::now(); + let Some(remaining) = deadline.checked_duration_since(now) else { + break; + }; + if remaining.is_zero() { + break; + } + self.socket.set_read_timeout(Some(remaining))?; + + let datagram = match recv_datagram(&self.socket, &mut buf) { + Ok(datagram) => datagram, + Err(err) + if matches!( + err.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + if Instant::now() >= deadline { + break; + } + continue; + } + Err(err) => return Err(ClientError::Socket(err)), + }; + if datagram.received_at.mono > deadline { + break; + } + + let reply = match self.runtime.inspect_open_datagram(&buf[..datagram.len])? { + OpenDatagramDisposition::Ignore => continue, + OpenDatagramDisposition::Trusted(reply) => reply, + }; + let machine = match self + .runtime + .prepare_open_acceptance(reply, datagram.received_at) + { + Ok(machine) => machine, + Err(failure) => { + self.send_cleanup_close_best_effort(failure.cleanup_close.as_deref()); + return Err(failure.primary); + } + }; + let prepared = match self.prepare_client_open(machine, datagram.received_at) { + Ok(prepared) => prepared, + Err(failure) => { + self.send_cleanup_close_best_effort(failure.machine.cleanup_close_packet()); + return Err(failure.primary); + } + }; + return self.apply_prepared_open(prepared); + } + } + + Err(ClientError::OpenTimeout) + } + + fn prepare_client_open( + &self, + machine: PreparedOpenAcceptance, + opened_at: ClientTimestamp, + ) -> Result> { + let Some(negotiated) = machine.normal_negotiated() else { + return Ok(PreparedClientOpen { + machine, + schedule: None, + recv_buffer_len: None, + negotiated_dscp: None, + previous_dscp: self.applied_dscp, + post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout, + }); + }; + let schedule = match ProbeSchedule::new(opened_at.mono, negotiated) { + Ok(schedule) => schedule, + Err(primary) => return Err(Box::new(PreparedClientOpenFailure { primary, machine })), + }; + let recv_buffer_len = match recv_buffer_size(self.runtime.has_hmac(), Some(negotiated)) { + Ok(size) => size, + Err(primary) => return Err(Box::new(PreparedClientOpenFailure { primary, machine })), + }; + #[cfg(test)] + let recv_buffer_len = self + .test_hooks + .recv_buffer_len_override + .get() + .unwrap_or(recv_buffer_len); + let negotiated_dscp = match u8::try_from(negotiated.params.dscp) { + Ok(dscp) => dscp, + Err(_) => { + return Err(Box::new(PreparedClientOpenFailure { + primary: ClientError::InvalidConfig { + reason: "negotiated dscp must be in range 0..=63".to_owned(), + }, + machine, + })); + } + }; + + Ok(PreparedClientOpen { + machine, + schedule: Some(schedule), + recv_buffer_len: Some(recv_buffer_len), + negotiated_dscp: Some(negotiated_dscp), + previous_dscp: self.applied_dscp, + post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout, + }) + } + + fn apply_prepared_open( + &mut self, + prepared: PreparedClientOpen, + ) -> Result { + let PreparedClientOpen { + machine, + schedule, + recv_buffer_len, + negotiated_dscp, + previous_dscp, + post_open_recv_timeout, + } = prepared; + + if let (Some(recv_buffer_len), Some(negotiated_dscp)) = (recv_buffer_len, negotiated_dscp) { + let previous_len = self.recv_buffer.len(); + let additional = recv_buffer_len.saturating_sub(previous_len); + if let Err(source) = self.recv_buffer.try_reserve(additional) { + self.send_cleanup_close_best_effort(machine.cleanup_close_packet()); + return Err(ClientError::AllocationFailed { + operation: "negotiated receive buffer", + source, + }); + } + self.recv_buffer.resize(recv_buffer_len, 0); + + #[cfg(test)] + self.test_hooks + .prepared_active_session_before_dscp + .set(machine.has_prepared_active_session()); + if let Err(primary) = self.apply_open_dscp(negotiated_dscp) { + self.recv_buffer.truncate(previous_len); + self.restore_dscp_best_effort(previous_dscp); + self.send_cleanup_close_best_effort(machine.cleanup_close_packet()); + return Err(primary); + } + if let Err(source) = self.restore_open_read_timeout(post_open_recv_timeout) { + let primary = ClientError::ReadTimeoutRestore { source }; + self.recv_buffer.truncate(previous_len); + self.restore_dscp_best_effort(previous_dscp); + self.send_cleanup_close_best_effort(machine.cleanup_close_packet()); + return Err(primary); + } + + let outcome = self.runtime.commit_open(machine); + self.schedule = schedule; + self.applied_dscp = Some(negotiated_dscp); + Ok(outcome) + } else { + debug_assert!(schedule.is_none()); + if let Err(source) = self.restore_open_read_timeout(post_open_recv_timeout) { + return Err(ClientError::ReadTimeoutRestore { source }); + } + let outcome = self.runtime.commit_open(machine); + self.schedule = None; + self.applied_dscp = None; + Ok(outcome) + } + } + + fn restore_dscp_best_effort(&self, previous_dscp: Option) { + #[cfg(test)] + if self.test_hooks.fail_dscp_restore.replace(false) { + return; + } + let _ = match previous_dscp { + Some(dscp) => apply_dscp_to_socket(&self.socket, self.remote, dscp), + None => clear_dscp_on_socket(&self.socket, self.remote), + }; + } + + fn send_cleanup_close_best_effort(&self, packet: Option<&[u8]>) { + if let Some(packet) = packet { + #[cfg(test)] + if self.test_hooks.fail_cleanup_send.replace(false) { + return; + } + let _ = self.socket.send(packet); + } + } + + fn apply_open_dscp(&self, dscp: u8) -> Result<(), ClientError> { + #[cfg(test)] + if self.test_hooks.fail_open_dscp.replace(false) { + return Err(ClientError::SocketOption { + operation: "set negotiated DSCP", + remote: self.remote, + source: io::Error::other("injected negotiated DSCP failure"), + }); + } + apply_dscp_to_socket(&self.socket, self.remote, dscp) + } + + fn restore_open_read_timeout(&self, timeout: Option) -> io::Result<()> { + #[cfg(test)] + self.test_hooks.last_restored_read_timeout.set(timeout); + #[cfg(test)] + if self.test_hooks.fail_open_timeout_restore.replace(false) { + return Err(io::Error::other( + "injected configured read timeout restoration failure", + )); + } + self.socket.set_read_timeout(timeout) + } } pub(crate) fn echo_sent_event( diff --git a/crates/irtt-client/src/client/tests/hmac.rs b/crates/irtt-client/src/client/tests/hmac.rs index a49d106..dc36a99 100644 --- a/crates/irtt-client/src/client/tests/hmac.rs +++ b/crates/irtt-client/src/client/tests/hmac.rs @@ -19,45 +19,77 @@ fn hmac_open_success() { } #[test] -fn hmac_open_rejects_missing_hmac() { +fn hmac_open_ignores_missing_hmac_before_valid_reply() { let key = b"secret".to_vec(); let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key); + config.hmac_key = Some(key.clone()); let params = params_from_config(&config).unwrap(); let server = start_fake_server(move |socket, tx| { let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); + let missing = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&key)); + socket.send_to(&missing, peer).unwrap(); + socket.send_to(&valid, peer).unwrap(); }); config.server_addr = server.addr.to_string(); let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(), - Err(ClientError::Protocol( - irtt_proto::ProtoError::HmacPresenceMismatch - )) - )); + assert_open_started(client.open().unwrap()); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(server.rx.try_recv().is_err()); server.join(); } #[test] -fn hmac_open_rejects_bad_hmac() { +fn hmac_open_ignores_bad_hmac_before_valid_reply() { let key = b"secret".to_vec(); let wrong_key = b"wrong".to_vec(); let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key); + config.hmac_key = Some(key.clone()); let params = params_from_config(&config).unwrap(); let server = start_fake_server(move |socket, tx| { let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&wrong_key)); - socket.send_to(&reply, peer).unwrap(); + let bad = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&wrong_key)); + let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&key)); + socket.send_to(&bad, peer).unwrap(); + socket.send_to(&valid, peer).unwrap(); }); config.server_addr = server.addr.to_string(); let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open().unwrap()); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(server.rx.try_recv().is_err()); + server.join(); +} + +#[test] +fn post_token_hmac_negotiation_failure_sends_authenticated_cleanup_close() { + let key = b"secret".to_vec(); + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.hmac_key = Some(key.clone()); + let mut returned = params_from_config(&config).unwrap(); + returned.interval_ns += 1; + let server_key = key.clone(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, Some(&server_key)); + socket.send_to(&reply, peer).unwrap(); + let _ = recv_request(&socket, &tx); + }); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert!(matches!( client.open(), - Err(ClientError::Protocol(irtt_proto::ProtoError::BadHmac)) + Err(ClientError::NegotiationRejected { .. }) )); + let packets: Vec<_> = server.rx.iter().take(2).collect(); + let cleanup = &packets[1]; + assert_eq!(cleanup[3], flags::FLAG_CLOSE | FLAG_HMAC); + verify_hmac(&key, cleanup, HMAC_OFFSET).unwrap(); + assert_eq!( + u64::from_le_bytes(cleanup[4 + HMAC_SIZE..12 + HMAC_SIZE].try_into().unwrap()), + TOKEN + ); server.join(); } diff --git a/crates/irtt-client/src/client/tests/no_test.rs b/crates/irtt-client/src/client/tests/no_test.rs index d1fc893..2af264f 100644 --- a/crates/irtt-client/src/client/tests/no_test.rs +++ b/crates/irtt-client/src/client/tests/no_test.rs @@ -13,6 +13,7 @@ fn no_test_success_validates_params() { assert_eq!(negotiated.params, params); assert!(client.next_send_deadline().is_none()); assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); server.join(); } diff --git a/crates/irtt-client/src/client/tests/open.rs b/crates/irtt-client/src/client/tests/open.rs index ae19c5f..12b8859 100644 --- a/crates/irtt-client/src/client/tests/open.rs +++ b/crates/irtt-client/src/client/tests/open.rs @@ -9,6 +9,30 @@ fn successful_open_handshake() { let negotiated = assert_open_started(client.open().unwrap()); assert_eq!(negotiated.params, params); + assert!(client.schedule.is_some()); + assert_eq!(client.applied_dscp, Some(0)); + server.join(); +} + +#[test] +fn successful_open_restores_configured_receive_timeout() { + let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let params = params_from_config(&config).unwrap(); + let server = open_success_server(params); + let mut config = default_test_config(server.addr); + config.socket_config.recv_timeout = Some(Duration::from_millis(75)); + let mut client = Client::connect(config).unwrap(); + + assert_open_started(client.open().unwrap()); + + assert_eq!( + client.test_hooks.last_restored_read_timeout.get(), + Some(Duration::from_millis(75)) + ); + assert_eq!( + client.socket.read_timeout().unwrap(), + Some(Duration::from_millis(75)) + ); server.join(); } @@ -49,11 +73,19 @@ fn open_timeout_after_all_timeouts() { let server = timeout_server(Duration::from_millis(700)); let config = ClientConfig { open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, ..default_test_config(server.addr) }; let mut client = Client::connect(config).unwrap(); assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); assert_eq!(server.rx.iter().take(2).count(), 2); + assert_eq!( + client.test_hooks.last_restored_read_timeout.get(), + Some(Duration::from_millis(50)) + ); server.join(); } @@ -69,6 +101,8 @@ fn protocol_version_mismatch_fails() { client.open(), Err(ClientError::ProtocolVersionMismatch { received: 2, .. }) )); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(client.runtime.prepare_open_request().is_ok()); server.join(); } @@ -83,5 +117,345 @@ fn server_rejection_fails_in_normal_mode() { }); let mut client = Client::connect(default_test_config(server.addr)).unwrap(); assert!(matches!(client.open(), Err(ClientError::ServerRejected))); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(client.runtime.prepare_open_request().is_ok()); + server.join(); +} + +#[test] +fn multiple_untrusted_datagrams_before_valid_reply_use_one_attempt() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + let mut wrong_type = valid.clone(); + wrong_type[3] = FLAG_REPLY; + let mut reserved_flags = valid.clone(); + reserved_flags[3] |= 0x10; + let mut bad_magic = valid.clone(); + bad_magic[0] ^= 0xff; + let unexpected_hmac = + open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(b"unexpected")); + + for packet in [ + vec![0_u8], + bad_magic, + wrong_type, + reserved_flags, + unexpected_hmac, + valid, + ] { + socket.send_to(&packet, peer).unwrap(); + } + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + + assert_open_started(client.open().unwrap()); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(server.rx.try_recv().is_err()); + assert!(client.schedule.is_some()); + server.join(); +} + +#[test] +fn ignored_datagrams_do_not_restart_the_attempt_deadline() { + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + thread::sleep(Duration::from_millis(150)); + socket.send_to(&[0_u8], peer).unwrap(); + thread::sleep(Duration::from_millis(150)); + socket.send_to(&[0_u8], peer).unwrap(); + let _ = recv_request(&socket, &tx); + }); + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(250), Duration::from_millis(250)], + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + let started = Instant::now(); + + assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); + let requests: Vec<_> = server.rx.iter().take(2).collect(); + assert_eq!(requests.len(), 2); + assert!( + started.elapsed() < Duration::from_millis(700), + "ignored datagrams restarted an attempt deadline" + ); + server.join(); +} + +#[test] +fn only_ignored_open_datagrams_eventually_time_out() { + let server = start_fake_server(move |socket, tx| { + for _ in 0..2 { + let (_, peer) = recv_request(&socket, &tx); + socket.send_to(&[0_u8], peer).unwrap(); + socket.send_to(&MAGIC, peer).unwrap(); + } + thread::sleep(Duration::from_millis(300)); + }); + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + + assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); + assert_eq!(server.rx.iter().take(2).count(), 2); + assert!(server.rx.try_recv().is_err()); + assert_eq!( + client.test_hooks.last_restored_read_timeout.get(), + Some(Duration::from_millis(50)) + ); + server.join(); +} + +#[test] +fn trusted_zero_token_normal_reply_is_terminal() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, 0, ¶ms, None), peer) + .unwrap(); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + + assert!(matches!(client.open(), Err(ClientError::ZeroToken))); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + server.join(); +} + +#[test] +fn opening_deadline_overflow_occurs_before_send() { + let server = timeout_server(Duration::from_millis(250)); + let config = ClientConfig { + open_timeouts: vec![Duration::MAX], + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + + assert!(matches!(client.open(), Err(ClientError::DurationOverflow))); + assert!(server.rx.try_recv().is_err()); + server.join(); +} + +#[test] +fn schedule_duration_overflow_cannot_partially_open() { + let server = start_fake_server(|_socket, _tx| {}); + let client = Client::connect(default_test_config(server.addr)).unwrap(); + let reply = irtt_proto::OpenReply { + flags: FLAG_OPEN | FLAG_REPLY, + token: TOKEN, + params: params_from_config(client.runtime.config()).unwrap(), + }; + let mut latest = Instant::now(); + let mut step = Duration::MAX; + while !step.is_zero() { + if let Some(candidate) = latest.checked_add(step) { + latest = candidate; + } + step /= 2; + } + let opened_at = ClientTimestamp { + mono: latest, + wall: SystemTime::now(), + }; + let machine = client + .runtime + .prepare_open_acceptance(reply, opened_at) + .unwrap(); + let previous_recv_len = client.recv_buffer.len(); + + let failure = client.prepare_client_open(machine, opened_at).unwrap_err(); + + assert!(matches!( + failure.primary, + ClientError::NegotiationRejected { .. } + )); + assert!(failure.machine.cleanup_close_packet().is_some()); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + assert_eq!(client.recv_buffer.len(), previous_recv_len); + server.join(); +} + +#[test] +fn post_token_negotiation_failure_sends_cleanup_close() { + let mut returned = default_params(); + returned.interval_ns += 1; + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + + assert!(matches!( + client.open(), + Err(ClientError::NegotiationRejected { .. }) + )); + let packets: Vec<_> = server.rx.iter().take(2).collect(); + assert_eq!(packets[1][3], flags::FLAG_CLOSE); + assert_eq!( + u64::from_le_bytes(packets[1][4..12].try_into().unwrap()), + TOKEN + ); + assert!(client.runtime.prepare_open_request().is_ok()); + server.join(); +} + +#[test] +fn cleanup_send_failure_does_not_replace_primary_open_error() { + let mut returned = default_params(); + returned.interval_ns += 1; + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None), + peer, + ) + .unwrap(); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + client.test_hooks.fail_cleanup_send.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::NegotiationRejected { .. }) + )); + assert!(client.runtime.prepare_open_request().is_ok()); + server.join(); +} + +#[test] +fn receive_buffer_reservation_failure_precedes_dscp_application() { + let mut returned = default_params(); + returned.dscp = 46; + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut config = default_test_config(server.addr); + config.dscp = 46; + let mut client = Client::connect(config).unwrap(); + client + .test_hooks + .recv_buffer_len_override + .set(Some(usize::MAX)); + client.test_hooks.fail_open_dscp.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::AllocationFailed { + operation: "negotiated receive buffer", + .. + }) + )); + assert!(client.test_hooks.fail_open_dscp.get()); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + server.join(); +} + +#[test] +fn dscp_application_failure_leaves_machine_connected() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + client.test_hooks.fail_open_dscp.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::SocketOption { + operation: "set negotiated DSCP", + .. + }) + )); + assert!(client.test_hooks.prepared_active_session_before_dscp.get()); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + server.join(); +} + +#[test] +fn read_timeout_restoration_failure_leaves_machine_connected() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + client.test_hooks.fail_open_timeout_restore.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::ReadTimeoutRestore { .. }) + )); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + server.join(); +} + +#[test] +fn dscp_rollback_failure_keeps_timeout_restoration_primary() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + client.test_hooks.fail_open_timeout_restore.set(true); + client.test_hooks.fail_dscp_restore.set(true); + + let error = client.open().unwrap_err(); + + assert!(matches!(error, ClientError::ReadTimeoutRestore { .. })); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); server.join(); } diff --git a/crates/irtt-client/src/error.rs b/crates/irtt-client/src/error.rs index 325946c..ee03b2a 100644 --- a/crates/irtt-client/src/error.rs +++ b/crates/irtt-client/src/error.rs @@ -38,8 +38,8 @@ pub enum ClientError { }, /// Restoring the configured read timeout after open negotiation failed. /// - /// The session may have opened successfully, but the socket could not be - /// returned to the caller's requested receive-timeout behavior. + /// The prepared session was not committed, and the client remains + /// connected rather than partially open. #[error("failed to restore configured socket read timeout after open negotiation: {source}")] ReadTimeoutRestore { source: io::Error }, /// A packet failed IRTT protocol parsing or authentication. diff --git a/crates/irtt-client/tests/hmac.rs b/crates/irtt-client/tests/hmac.rs index fbd099e..1ccd0a9 100644 --- a/crates/irtt-client/tests/hmac.rs +++ b/crates/irtt-client/tests/hmac.rs @@ -5,7 +5,7 @@ use std::time::Duration; use irtt_client::{ Client, ClientConfig, ClientError, ClientEvent, OpenOutcome, SocketConfig, WarningKind, }; -use irtt_proto::{ProtoError, TimestampFields}; +use irtt_proto::TimestampFields; use support::{ config_for_params, default_params, params_for_modes, run_one_probe_with_config, @@ -192,7 +192,7 @@ fn bad_hmac_echo_reply_is_rejected_without_echo_reply_event() { } #[test] -fn hmac_open_reply_with_bad_hmac_fails_with_protocol_error() { +fn hmac_open_reply_with_bad_hmac_is_ignored_until_timeout() { let key = b"compat-secret".to_vec(); let wrong_key = b"wrong-secret".to_vec(); let params = default_params(); @@ -201,10 +201,7 @@ fn hmac_open_reply_with_bad_hmac_fails_with_protocol_error() { config.hmac_key = Some(key); let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(), - Err(ClientError::Protocol(ProtoError::BadHmac)) - )); + assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); server.join(); }