diff --git a/crates/irtt-client/src/client/tests/no_test.rs b/crates/irtt-client/src/client/tests/no_test.rs index 2af264f..6b64075 100644 --- a/crates/irtt-client/src/client/tests/no_test.rs +++ b/crates/irtt-client/src/client/tests/no_test.rs @@ -22,13 +22,55 @@ fn no_test_rejects_non_close_open_reply() { let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); config.run_mode = RunMode::NoTest; let params = params_from_config(&config).unwrap(); - let server = open_success_server(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); + }); config.server_addr = server.addr.to_string(); let mut client = Client::connect(config).unwrap(); assert!(matches!( client.open(), Err(ClientError::UnexpectedNoTestReply) )); + 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 no_test_cleanup_send_failure_preserves_unexpected_reply() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + let params = params_from_config(&config).unwrap(); + 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(); + }); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + client.test_hooks.fail_cleanup_send.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::UnexpectedNoTestReply) + )); + assert!(client.runtime.prepare_open_request().is_ok()); server.join(); } diff --git a/crates/irtt-client/src/client/tests/open.rs b/crates/irtt-client/src/client/tests/open.rs index 2b75380..f7f4c21 100644 --- a/crates/irtt-client/src/client/tests/open.rs +++ b/crates/irtt-client/src/client/tests/open.rs @@ -29,10 +29,6 @@ fn successful_open_restores_configured_receive_timeout() { 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(); } diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 06d10ad..bb11dc3 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -1092,10 +1092,10 @@ fn prepare_group_open( machine: PreparedOpenAcceptance, opened_at: ClientTimestamp, ) -> Result> { - if let Err(primary) = validate_group_negotiation(shared, machine.negotiated()) { - return Err(Box::new(PreparedGroupOpenFailure { primary, machine })); - } let schedule = if let Some(negotiated) = machine.normal_negotiated() { + if let Err(primary) = validate_group_negotiation(shared, negotiated) { + return Err(Box::new(PreparedGroupOpenFailure { primary, machine })); + } match ProbeSchedule::new(opened_at.mono, negotiated) { Ok(schedule) => Some(schedule), Err(primary) => return Err(Box::new(PreparedGroupOpenFailure { primary, machine })), @@ -3379,6 +3379,162 @@ mod tests { server.join(); } + #[test] + fn no_test_non_close_reply_sends_cleanup_and_fails_without_retry() { + let interval = Duration::from_millis(10); + let params = test_params(None, interval); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let (tx, rx) = mpsc::channel(); + let done = thread::spawn(move || { + let (open, peer) = recv_request_timeout(&socket).expect("missing open request"); + assert_ne!(open[3] & flags::FLAG_CLOSE, 0); + tx.send(ServerObservation::Open { at: Instant::now() }) + .unwrap(); + socket + .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), peer) + .unwrap(); + let (cleanup, _) = + recv_request_timeout(&socket).expect("missing no-test cleanup close"); + assert_eq!(cleanup[3], flags::FLAG_CLOSE); + assert_eq!( + u64::from_le_bytes(cleanup[4..12].try_into().unwrap()), + TOKEN + ); + tx.send(ServerObservation::Close { at: Instant::now() }) + .unwrap(); + }); + let server = FakeServer { + addr, + _observations: rx, + done, + }; + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered); + config.client.run_mode = crate::RunMode::NoTest; + + let outcome = ManagedClientGroup::start(config, vec![target("peer", server.addr)]) + .unwrap() + .join() + .unwrap(); + + assert_eq!(outcome.failed_target_outcomes, 1); + assert!(matches!( + outcome.targets[0].end_reason, + ManagedTargetEndReason::OpenFailed(ManagedTargetFailure { + kind: ManagedTargetFailureKind::OpeningProtocol, + .. + }) + )); + assert_eq!(server._observations.try_iter().collect::>().len(), 2); + server.join(); + } + + #[test] + fn loose_no_test_group_skips_active_interval_and_dscp_policy() { + let interval = Duration::from_millis(10); + let mut changed_interval = test_params(None, interval); + changed_interval.interval_ns += 1; + let changed_dscp = test_params(None, interval); + + for (requested_dscp, params) in [(0, changed_interval), (1, changed_dscp)] { + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.run_mode = crate::RunMode::NoTest; + config.negotiation_policy = NegotiationPolicy::Loose; + config.dscp = requested_dscp; + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target.clone()); + let opened_at = ClientTimestamp::now(); + + let mut target = target.lock().expect("target mutex poisoned"); + let packet = open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, ¶ms); + let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() { + OpenDatagramDisposition::Trusted(reply) => reply, + OpenDatagramDisposition::Ignore => panic!("no-test reply must be trusted"), + }; + let machine = target + .runtime + .prepare_open_acceptance(reply, opened_at) + .unwrap(); + let prepared = prepare_group_open(&shared, machine, opened_at).unwrap(); + assert!(prepared.schedule.is_none()); + let outcome = target.runtime.commit_open(prepared.machine); + target.schedule = prepared.schedule; + publish_open_outcome(&shared.hub, &mut target, outcome); + + assert!(matches!(target.status, TargetStatus::Finished)); + assert_eq!( + target.final_reason, + Some(ManagedTargetEndReason::NoTestComplete) + ); + } + } + + #[test] + fn normal_and_strict_sessions_preserve_group_negotiation_checks() { + let interval = Duration::from_millis(10); + let mut changed_interval = test_params(None, interval); + changed_interval.interval_ns += 1; + let changed_dscp = test_params(None, interval); + + for (requested_dscp, params) in [(0, changed_interval.clone()), (1, changed_dscp)] { + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.negotiation_policy = NegotiationPolicy::Loose; + config.dscp = requested_dscp; + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target.clone()); + let opened_at = ClientTimestamp::now(); + let target = target.lock().expect("target mutex poisoned"); + let packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms); + let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() { + OpenDatagramDisposition::Trusted(reply) => reply, + OpenDatagramDisposition::Ignore => panic!("normal reply must be trusted"), + }; + let machine = target + .runtime + .prepare_open_acceptance(reply, opened_at) + .unwrap(); + let failure = prepare_group_open(&shared, machine, opened_at).unwrap_err(); + assert!(matches!( + failure.primary, + ClientError::NegotiationRejected { .. } + )); + } + + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.run_mode = crate::RunMode::NoTest; + let target = TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(); + let packet = open_reply( + FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, + 0, + &changed_interval, + ); + let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() { + OpenDatagramDisposition::Trusted(reply) => reply, + OpenDatagramDisposition::Ignore => panic!("strict no-test reply must be trusted"), + }; + let failure = target + .runtime + .prepare_open_acceptance(reply, ClientTimestamp::now()) + .unwrap_err(); + assert!(matches!( + failure.primary, + ClientError::NegotiationRejected { .. } + )); + } + #[test] fn group_open_deadline_overflow_fails_before_send() { let silent_peer = UdpSocket::bind("127.0.0.1:0").unwrap(); diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index e8c1c96..74cced0 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -80,13 +80,6 @@ pub(crate) struct PreparedOpenAcceptance { } impl PreparedOpenAcceptance { - pub(crate) fn negotiated(&self) -> &NegotiatedParams { - match &self.outcome { - OpenOutcome::Started { negotiated, .. } - | OpenOutcome::NoTestCompleted { negotiated, .. } => negotiated, - } - } - pub(crate) fn normal_negotiated(&self) -> Option<&NegotiatedParams> { match &self.next_state { MachineState::Open(session) => Some(&session.negotiated), @@ -228,18 +221,17 @@ impl SessionMachine { .map_err(OpenAcceptanceFailure::without_cleanup)?; let reply_is_close = flags::has(reply.flags, flags::FLAG_CLOSE); - let cleanup_close = - if self.config.run_mode == RunMode::Normal && !reply_is_close && reply.token != 0 { - let bytes = encode_close_request( - &CloseRequest { token: reply.token }, - self.config.hmac_key.as_deref(), - ) - .map_err(ClientError::from) - .map_err(OpenAcceptanceFailure::without_cleanup)?; - Some(bytes.into_boxed_slice()) - } else { - None - }; + let cleanup_close = if !reply_is_close && reply.token != 0 { + let bytes = encode_close_request( + &CloseRequest { token: reply.token }, + self.config.hmac_key.as_deref(), + ) + .map_err(ClientError::from) + .map_err(OpenAcceptanceFailure::without_cleanup)?; + Some(bytes.into_boxed_slice()) + } else { + None + }; if reply.params.protocol_version != PROTOCOL_VERSION { return Err(OpenAcceptanceFailure::new( @@ -1186,6 +1178,69 @@ mod tests { assert!(matches!(no_test.state, MachineState::Connected)); } + #[test] + fn no_test_non_close_reply_prepares_authenticated_cleanup_without_state_change() { + let key = b"cleanup-key".to_vec(); + let machine = connected_machine(ClientConfig { + run_mode: RunMode::NoTest, + hmac_key: Some(key.clone()), + ..ClientConfig::default() + }); + let token = 0x1020_3040_5060_7080; + + let failure = machine + .prepare_open_acceptance( + normal_open_reply(&machine, token), + timestamp(Instant::now()), + ) + .unwrap_err(); + + assert!(matches!( + failure.primary, + ClientError::UnexpectedNoTestReply + )); + let cleanup = failure.cleanup_close.as_deref().unwrap(); + assert_eq!(cleanup[3], flags::FLAG_CLOSE | flags::FLAG_HMAC); + irtt_proto::verify_hmac(&key, cleanup, 4).unwrap(); + assert_eq!( + u64::from_le_bytes( + cleanup[4 + irtt_proto::HMAC_SIZE..12 + irtt_proto::HMAC_SIZE] + .try_into() + .unwrap() + ), + token + ); + drop(failure); + assert!(matches!(machine.state, MachineState::Connected)); + } + + #[test] + fn no_test_close_replies_do_not_prepare_cleanup() { + let machine = connected_machine(ClientConfig { + run_mode: RunMode::NoTest, + ..ClientConfig::default() + }); + let reply = |token| OpenReply { + flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE, + token, + params: machine.requested.clone(), + }; + + let prepared = machine + .prepare_open_acceptance(reply(0), timestamp(Instant::now())) + .unwrap(); + assert!(prepared.cleanup_close_packet().is_none()); + + let failure = machine + .prepare_open_acceptance(reply(0x1020_3040_5060_7080), timestamp(Instant::now())) + .unwrap_err(); + assert!(matches!( + failure.primary, + ClientError::NonZeroNoTestToken { .. } + )); + assert!(failure.cleanup_close.is_none()); + } + #[test] fn dropping_prepared_acceptance_leaves_connected_state() { let machine = connected_machine(ClientConfig::default());