diff options
Diffstat (limited to 'third_party/rust/neqo-transport/src/connection/tests')
16 files changed, 8654 insertions, 0 deletions
diff --git a/third_party/rust/neqo-transport/src/connection/tests/ackrate.rs b/third_party/rust/neqo-transport/src/connection/tests/ackrate.rs new file mode 100644 index 0000000000..1b83d42acd --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/ackrate.rs @@ -0,0 +1,194 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{mem, time::Duration}; + +use test_fixture::{addr_v4, assertions}; + +use super::{ + super::{ConnectionParameters, ACK_RATIO_SCALE}, + ack_bytes, connect_rtt_idle, default_client, default_server, fill_cwnd, increase_cwnd, + induce_persistent_congestion, new_client, new_server, send_something, DEFAULT_RTT, +}; +use crate::stream_id::StreamType; + +/// With the default RTT here (100ms) and default ratio (4), endpoints won't send +/// `ACK_FREQUENCY` as the ACK delay isn't different enough from the default. +#[test] +fn ack_rate_default() { + let mut client = default_client(); + let mut server = default_server(); + _ = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + assert_eq!(client.stats().frame_tx.ack_frequency, 0); + assert_eq!(server.stats().frame_tx.ack_frequency, 0); +} + +/// When the congestion window increases, the rate doesn't change. +#[test] +fn ack_rate_slow_start() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Increase the congestion window a few times. + let stream = client.stream_create(StreamType::UniDi).unwrap(); + let now = increase_cwnd(&mut client, &mut server, stream, now); + let now = increase_cwnd(&mut client, &mut server, stream, now); + _ = increase_cwnd(&mut client, &mut server, stream, now); + + // The client should not have sent an ACK_FREQUENCY frame, even + // though the value would have updated. + assert_eq!(client.stats().frame_tx.ack_frequency, 0); + assert_eq!(server.stats().frame_rx.ack_frequency, 0); +} + +/// When the congestion window decreases, a frame is sent. +#[test] +fn ack_rate_exit_slow_start() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Increase the congestion window a few times, enough that after a loss, + // there are enough packets in the window to increase the packet + // count in ACK_FREQUENCY frames. + let stream = client.stream_create(StreamType::UniDi).unwrap(); + let now = increase_cwnd(&mut client, &mut server, stream, now); + let now = increase_cwnd(&mut client, &mut server, stream, now); + + // Now fill the congestion window and drop the first packet. + let (mut pkts, mut now) = fill_cwnd(&mut client, stream, now); + pkts.remove(0); + + // After acknowledging the other packets the client will notice the loss. + now += DEFAULT_RTT / 2; + let ack = ack_bytes(&mut server, stream, pkts, now); + + // Receiving the ACK will cause the client to reduce its congestion window + // and to send ACK_FREQUENCY. + now += DEFAULT_RTT / 2; + assert_eq!(client.stats().frame_tx.ack_frequency, 0); + let af = client.process(Some(&ack), now).dgram(); + assert!(af.is_some()); + assert_eq!(client.stats().frame_tx.ack_frequency, 1); +} + +/// When the congestion window collapses, `ACK_FREQUENCY` is updated. +#[test] +fn ack_rate_persistent_congestion() { + // Use a configuration that results in the value being set after exiting + // the handshake. + const RTT: Duration = Duration::from_millis(3); + let mut client = new_client(ConnectionParameters::default().ack_ratio(ACK_RATIO_SCALE)); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, RTT); + + // The client should have sent a frame. + assert_eq!(client.stats().frame_tx.ack_frequency, 1); + + // Now crash the congestion window. + let stream = client.stream_create(StreamType::UniDi).unwrap(); + let (dgrams, mut now) = fill_cwnd(&mut client, stream, now); + now += RTT / 2; + mem::drop(ack_bytes(&mut server, stream, dgrams, now)); + + let now = induce_persistent_congestion(&mut client, &mut server, stream, now); + + // The client sends a second ACK_FREQUENCY frame with an increased rate. + let af = client.process_output(now).dgram(); + assert!(af.is_some()); + assert_eq!(client.stats().frame_tx.ack_frequency, 2); +} + +/// Validate that the configuration works for the client. +#[test] +fn ack_rate_client_one_rtt() { + // This has to be chosen so that the resulting ACK delay is between 1ms and 50ms. + // We also have to avoid values between 20..30ms (approximately). The default + // maximum ACK delay is 25ms and an ACK_FREQUENCY frame won't be sent when the + // change to the maximum ACK delay is too small. + const RTT: Duration = Duration::from_millis(3); + let mut client = new_client(ConnectionParameters::default().ack_ratio(ACK_RATIO_SCALE)); + let mut server = default_server(); + let mut now = connect_rtt_idle(&mut client, &mut server, RTT); + + // A single packet from the client will cause the server to engage its delayed + // acknowledgment timer, which should now be equal to RTT. + // The first packet will elicit an immediate ACK however, so do this twice. + let d = send_something(&mut client, now); + now += RTT / 2; + let ack = server.process(Some(&d), now).dgram(); + assert!(ack.is_some()); + let d = send_something(&mut client, now); + now += RTT / 2; + let delay = server.process(Some(&d), now).callback(); + assert_eq!(delay, RTT); + + assert_eq!(client.stats().frame_tx.ack_frequency, 1); +} + +/// Validate that the configuration works for the server. +#[test] +fn ack_rate_server_half_rtt() { + const RTT: Duration = Duration::from_millis(10); + let mut client = default_client(); + let mut server = new_server(ConnectionParameters::default().ack_ratio(ACK_RATIO_SCALE * 2)); + let mut now = connect_rtt_idle(&mut client, &mut server, RTT); + + // The server now sends something. + let d = send_something(&mut server, now); + now += RTT / 2; + // The client now will acknowledge immediately because it has been more than + // an RTT since it last sent an acknowledgment. + let ack = client.process(Some(&d), now); + assert!(ack.as_dgram_ref().is_some()); + let d = send_something(&mut server, now); + now += RTT / 2; + let delay = client.process(Some(&d), now).callback(); + assert_eq!(delay, RTT / 2); + + assert_eq!(server.stats().frame_tx.ack_frequency, 1); +} + +/// ACK delay calculations are path-specific, +/// so check that they can be sent on new paths. +#[test] +fn migrate_ack_delay() { + // Have the client send ACK_FREQUENCY frames at a normal-ish rate. + let mut client = new_client(ConnectionParameters::default().ack_ratio(ACK_RATIO_SCALE)); + let mut server = default_server(); + let mut now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + client + .migrate(Some(addr_v4()), Some(addr_v4()), true, now) + .unwrap(); + + let client1 = send_something(&mut client, now); + assertions::assert_v4_path(&client1, true); // Contains PATH_CHALLENGE. + let client2 = send_something(&mut client, now); + assertions::assert_v4_path(&client2, false); // Doesn't. Is dropped. + now += DEFAULT_RTT / 2; + server.process_input(&client1, now); + + let stream = client.stream_create(StreamType::UniDi).unwrap(); + let now = increase_cwnd(&mut client, &mut server, stream, now); + let now = increase_cwnd(&mut client, &mut server, stream, now); + let now = increase_cwnd(&mut client, &mut server, stream, now); + + // Now lose a packet and force the client to update + let (mut pkts, mut now) = fill_cwnd(&mut client, stream, now); + pkts.remove(0); + now += DEFAULT_RTT / 2; + let ack = ack_bytes(&mut server, stream, pkts, now); + + // After noticing this new loss, the client sends ACK_FREQUENCY. + // It has sent a few before (as we dropped `client2`), so ignore those. + let ad_before = client.stats().frame_tx.ack_frequency; + let af = client.process(Some(&ack), now).dgram(); + assert!(af.is_some()); + assert_eq!(client.stats().frame_tx.ack_frequency, ad_before + 1); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/cc.rs b/third_party/rust/neqo-transport/src/connection/tests/cc.rs new file mode 100644 index 0000000000..b3467ea67c --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/cc.rs @@ -0,0 +1,429 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{convert::TryFrom, mem, time::Duration}; + +use neqo_common::{qdebug, qinfo, Datagram}; + +use super::{ + super::Output, ack_bytes, assert_full_cwnd, connect_rtt_idle, cwnd, cwnd_avail, cwnd_packets, + default_client, default_server, fill_cwnd, induce_persistent_congestion, send_something, + CLIENT_HANDSHAKE_1RTT_PACKETS, DEFAULT_RTT, POST_HANDSHAKE_CWND, +}; +use crate::{ + cc::MAX_DATAGRAM_SIZE, + packet::PacketNumber, + recovery::{ACK_ONLY_SIZE_LIMIT, PACKET_THRESHOLD}, + sender::PACING_BURST_SIZE, + stream_id::StreamType, + tracking::DEFAULT_ACK_PACKET_TOLERANCE, +}; + +#[test] +/// Verify initial CWND is honored. +fn cc_slow_start() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Try to send a lot of data + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + let (c_tx_dgrams, _) = fill_cwnd(&mut client, stream_id, now); + assert_full_cwnd(&c_tx_dgrams, POST_HANDSHAKE_CWND); + assert!(cwnd_avail(&client) < ACK_ONLY_SIZE_LIMIT); +} + +#[test] +/// Verify that CC moves to cong avoidance when a packet is marked lost. +fn cc_slow_start_to_cong_avoidance_recovery_period() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Create stream 0 + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + assert_eq!(stream_id, 0); + + // Buffer up lot of data and generate packets + let (c_tx_dgrams, mut now) = fill_cwnd(&mut client, stream_id, now); + assert_full_cwnd(&c_tx_dgrams, POST_HANDSHAKE_CWND); + // Predict the packet number of the last packet sent. + // We have already sent packets in `connect_rtt_idle`, + // so include a fudge factor. + let flight1_largest = + PacketNumber::try_from(c_tx_dgrams.len() + CLIENT_HANDSHAKE_1RTT_PACKETS).unwrap(); + + // Server: Receive and generate ack + now += DEFAULT_RTT / 2; + let s_ack = ack_bytes(&mut server, stream_id, c_tx_dgrams, now); + assert_eq!( + server.stats().frame_tx.largest_acknowledged, + flight1_largest + ); + + // Client: Process ack + now += DEFAULT_RTT / 2; + client.process_input(&s_ack, now); + assert_eq!( + client.stats().frame_rx.largest_acknowledged, + flight1_largest + ); + + // Client: send more + let (mut c_tx_dgrams, mut now) = fill_cwnd(&mut client, stream_id, now); + assert_full_cwnd(&c_tx_dgrams, POST_HANDSHAKE_CWND * 2); + let flight2_largest = flight1_largest + u64::try_from(c_tx_dgrams.len()).unwrap(); + + // Server: Receive and generate ack again, but drop first packet + now += DEFAULT_RTT / 2; + c_tx_dgrams.remove(0); + let s_ack = ack_bytes(&mut server, stream_id, c_tx_dgrams, now); + assert_eq!( + server.stats().frame_tx.largest_acknowledged, + flight2_largest + ); + + // Client: Process ack + now += DEFAULT_RTT / 2; + client.process_input(&s_ack, now); + assert_eq!( + client.stats().frame_rx.largest_acknowledged, + flight2_largest + ); +} + +#[test] +/// Verify that CC stays in recovery period when packet sent before start of +/// recovery period is acked. +fn cc_cong_avoidance_recovery_period_unchanged() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Create stream 0 + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + assert_eq!(stream_id, 0); + + // Buffer up lot of data and generate packets + let (mut c_tx_dgrams, now) = fill_cwnd(&mut client, stream_id, now); + assert_full_cwnd(&c_tx_dgrams, POST_HANDSHAKE_CWND); + + // Drop 0th packet. When acked, this should put client into CARP. + c_tx_dgrams.remove(0); + + let c_tx_dgrams2 = c_tx_dgrams.split_off(5); + + // Server: Receive and generate ack + let s_ack = ack_bytes(&mut server, stream_id, c_tx_dgrams, now); + client.process_input(&s_ack, now); + + let cwnd1 = cwnd(&client); + + // Generate ACK for more received packets + let s_ack = ack_bytes(&mut server, stream_id, c_tx_dgrams2, now); + + // ACK more packets but they were sent before end of recovery period + client.process_input(&s_ack, now); + + // cwnd should not have changed since ACKed packets were sent before + // recovery period expired + let cwnd2 = cwnd(&client); + assert_eq!(cwnd1, cwnd2); +} + +#[test] +/// Ensure that a single packet is sent after entering recovery, even +/// when that exceeds the available congestion window. +fn single_packet_on_recovery() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Drop a few packets, up to the reordering threshold. + for _ in 0..PACKET_THRESHOLD { + let _dropped = send_something(&mut client, now); + } + let delivered = send_something(&mut client, now); + + // Now fill the congestion window. + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + assert_eq!(stream_id, 0); + let (_, now) = fill_cwnd(&mut client, stream_id, now); + assert!(cwnd_avail(&client) < ACK_ONLY_SIZE_LIMIT); + + // Acknowledge just one packet and cause one packet to be declared lost. + // The length is the amount of credit the client should have. + let ack = server.process(Some(&delivered), now).dgram(); + assert!(ack.is_some()); + + // The client should see the loss and enter recovery. + // As there are many outstanding packets, there should be no available cwnd. + client.process_input(&ack.unwrap(), now); + assert_eq!(cwnd_avail(&client), 0); + + // The client should send one packet, ignoring the cwnd. + let dgram = client.process_output(now).dgram(); + assert!(dgram.is_some()); +} + +#[test] +/// Verify that CC moves out of recovery period when packet sent after start +/// of recovery period is acked. +fn cc_cong_avoidance_recovery_period_to_cong_avoidance() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Create stream 0 + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + assert_eq!(stream_id, 0); + + // Buffer up lot of data and generate packets + let (mut c_tx_dgrams, mut now) = fill_cwnd(&mut client, stream_id, now); + + // Drop 0th packet. When acked, this should put client into CARP. + c_tx_dgrams.remove(0); + + // Server: Receive and generate ack + now += DEFAULT_RTT / 2; + let s_ack = ack_bytes(&mut server, stream_id, c_tx_dgrams, now); + + // Client: Process ack + now += DEFAULT_RTT / 2; + client.process_input(&s_ack, now); + + // Should be in CARP now. + now += DEFAULT_RTT / 2; + qinfo!("moving to congestion avoidance {}", cwnd(&client)); + + // Now make sure that we increase congestion window according to the + // accurate byte counting version of congestion avoidance. + // Check over several increases to be sure. + let mut expected_cwnd = cwnd(&client); + // Fill cwnd. + let (mut c_tx_dgrams, next_now) = fill_cwnd(&mut client, stream_id, now); + now = next_now; + for i in 0..5 { + qinfo!("iteration {}", i); + + let c_tx_size: usize = c_tx_dgrams.iter().map(|d| d.len()).sum(); + qinfo!( + "client sending {} bytes into cwnd of {}", + c_tx_size, + cwnd(&client) + ); + assert_eq!(c_tx_size, expected_cwnd); + + // As acks arrive we will continue filling cwnd and save all packets + // from this cycle will be stored in next_c_tx_dgrams. + let mut next_c_tx_dgrams: Vec<Datagram> = Vec::new(); + + // Until we process all the packets, the congestion window remains the same. + // Note that we need the client to process ACK frames in stages, so split the + // datagrams into two, ensuring that we allow for an ACK for each batch. + let most = c_tx_dgrams.len() - usize::try_from(DEFAULT_ACK_PACKET_TOLERANCE).unwrap() - 1; + let s_ack = ack_bytes(&mut server, stream_id, c_tx_dgrams.drain(..most), now); + assert_eq!(cwnd(&client), expected_cwnd); + client.process_input(&s_ack, now); + // make sure to fill cwnd again. + let (mut new_pkts, next_now) = fill_cwnd(&mut client, stream_id, now); + now = next_now; + next_c_tx_dgrams.append(&mut new_pkts); + + let s_ack = ack_bytes(&mut server, stream_id, c_tx_dgrams, now); + assert_eq!(cwnd(&client), expected_cwnd); + client.process_input(&s_ack, now); + // make sure to fill cwnd again. + let (mut new_pkts, next_now) = fill_cwnd(&mut client, stream_id, now); + now = next_now; + next_c_tx_dgrams.append(&mut new_pkts); + + expected_cwnd += MAX_DATAGRAM_SIZE; + assert_eq!(cwnd(&client), expected_cwnd); + c_tx_dgrams = next_c_tx_dgrams; + } +} + +#[test] +/// Verify transition to persistent congestion state if conditions are met. +fn cc_slow_start_to_persistent_congestion_no_acks() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + let stream = client.stream_create(StreamType::BiDi).unwrap(); + + // Buffer up lot of data and generate packets + let (c_tx_dgrams, mut now) = fill_cwnd(&mut client, stream, now); + assert_full_cwnd(&c_tx_dgrams, POST_HANDSHAKE_CWND); + + // Server: Receive and generate ack + now += DEFAULT_RTT / 2; + mem::drop(ack_bytes(&mut server, stream, c_tx_dgrams, now)); + + // ACK lost. + induce_persistent_congestion(&mut client, &mut server, stream, now); +} + +#[test] +/// Verify transition to persistent congestion state if conditions are met. +fn cc_slow_start_to_persistent_congestion_some_acks() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Create stream 0 + let stream = client.stream_create(StreamType::BiDi).unwrap(); + + // Buffer up lot of data and generate packets + let (c_tx_dgrams, mut now) = fill_cwnd(&mut client, stream, now); + assert_full_cwnd(&c_tx_dgrams, POST_HANDSHAKE_CWND); + + // Server: Receive and generate ack + now += Duration::from_millis(100); + let s_ack = ack_bytes(&mut server, stream, c_tx_dgrams, now); + + now += Duration::from_millis(100); + client.process_input(&s_ack, now); + + // send bytes that will be lost + let (_, next_now) = fill_cwnd(&mut client, stream, now); + now = next_now + Duration::from_millis(100); + + induce_persistent_congestion(&mut client, &mut server, stream, now); +} + +#[test] +/// Verify persistent congestion moves to slow start after recovery period +/// ends. +fn cc_persistent_congestion_to_slow_start() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Create stream 0 + let stream = client.stream_create(StreamType::BiDi).unwrap(); + + // Buffer up lot of data and generate packets + let (c_tx_dgrams, mut now) = fill_cwnd(&mut client, stream, now); + assert_full_cwnd(&c_tx_dgrams, POST_HANDSHAKE_CWND); + + // Server: Receive and generate ack + now += Duration::from_millis(10); + mem::drop(ack_bytes(&mut server, stream, c_tx_dgrams, now)); + + // ACK lost. + + now = induce_persistent_congestion(&mut client, &mut server, stream, now); + + // New part of test starts here + + now += Duration::from_millis(10); + + // Send packets from after start of CARP + let (c_tx_dgrams, next_now) = fill_cwnd(&mut client, stream, now); + assert_eq!(c_tx_dgrams.len(), 2); + + // Server: Receive and generate ack + now = next_now + Duration::from_millis(100); + let s_ack = ack_bytes(&mut server, stream, c_tx_dgrams, now); + + // No longer in CARP. (pkts acked from after start of CARP) + // Should be in slow start now. + client.process_input(&s_ack, now); + + // ACKing 2 packets should let client send 4. + let (c_tx_dgrams, _) = fill_cwnd(&mut client, stream, now); + assert_eq!(c_tx_dgrams.len(), 4); +} + +#[test] +fn ack_are_not_cc() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Create a stream + let stream = client.stream_create(StreamType::BiDi).unwrap(); + assert_eq!(stream, 0); + + // Buffer up lot of data and generate packets, so that cc window is filled. + let (c_tx_dgrams, now) = fill_cwnd(&mut client, stream, now); + assert_full_cwnd(&c_tx_dgrams, POST_HANDSHAKE_CWND); + + // The server hasn't received any of these packets yet, the server + // won't ACK, but if it sends an ack-eliciting packet instead. + qdebug!([server], "Sending ack-eliciting"); + let other_stream = server.stream_create(StreamType::BiDi).unwrap(); + assert_eq!(other_stream, 1); + server.stream_send(other_stream, b"dropped").unwrap(); + let dropped_packet = server.process(None, now).dgram(); + assert!(dropped_packet.is_some()); // Now drop this one. + + // Now the server sends a packet that will force an ACK, + // because the client will detect a gap. + server.stream_send(other_stream, b"sent").unwrap(); + let ack_eliciting_packet = server.process(None, now).dgram(); + assert!(ack_eliciting_packet.is_some()); + + // The client can ack the server packet even if cc windows is full. + qdebug!([client], "Process ack-eliciting"); + let ack_pkt = client.process(ack_eliciting_packet.as_ref(), now).dgram(); + assert!(ack_pkt.is_some()); + qdebug!([server], "Handle ACK"); + let prev_ack_count = server.stats().frame_rx.ack; + server.process_input(&ack_pkt.unwrap(), now); + assert_eq!(server.stats().frame_rx.ack, prev_ack_count + 1); +} + +#[test] +fn pace() { + const DATA: &[u8] = &[0xcc; 4_096]; + let mut client = default_client(); + let mut server = default_server(); + let mut now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Now fill up the pipe and watch it trickle out. + let stream = client.stream_create(StreamType::BiDi).unwrap(); + loop { + let written = client.stream_send(stream, DATA).unwrap(); + if written < DATA.len() { + break; + } + } + let mut count = 0; + // We should get a burst at first. + // The first packet is not subject to pacing as there are no bytes in flight. + // After that we allow the burst to continue up to a number of packets (2). + for _ in 0..=PACING_BURST_SIZE { + let dgram = client.process_output(now).dgram(); + assert!(dgram.is_some()); + count += 1; + } + let gap = client.process_output(now).callback(); + assert_ne!(gap, Duration::new(0, 0)); + for _ in (1 + PACING_BURST_SIZE)..cwnd_packets(POST_HANDSHAKE_CWND) { + match client.process_output(now) { + Output::Callback(t) => assert_eq!(t, gap), + Output::Datagram(_) => { + // The last packet might not be paced. + count += 1; + break; + } + Output::None => panic!(), + } + now += gap; + let dgram = client.process_output(now).dgram(); + assert!(dgram.is_some()); + count += 1; + } + let dgram = client.process_output(now).dgram(); + assert!(dgram.is_none()); + assert_eq!(count, cwnd_packets(POST_HANDSHAKE_CWND)); + let fin = client.process_output(now).callback(); + assert_ne!(fin, Duration::new(0, 0)); + assert_ne!(fin, gap); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/close.rs b/third_party/rust/neqo-transport/src/connection/tests/close.rs new file mode 100644 index 0000000000..f45e77e549 --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/close.rs @@ -0,0 +1,210 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::time::Duration; + +use test_fixture::{self, datagram, now}; + +use super::{ + super::{Connection, Output, State}, + connect, connect_force_idle, default_client, default_server, send_something, +}; +use crate::{ + tparams::{self, TransportParameter}, + AppError, ConnectionError, Error, ERROR_APPLICATION_CLOSE, +}; + +fn assert_draining(c: &Connection, expected: &Error) { + assert!(c.state().closed()); + if let State::Draining { + error: ConnectionError::Transport(error), + .. + } = c.state() + { + assert_eq!(error, expected); + } else { + panic!(); + } +} + +#[test] +fn connection_close() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let now = now(); + + client.close(now, 42, ""); + + let out = client.process(None, now); + + server.process_input(&out.dgram().unwrap(), now); + assert_draining(&server, &Error::PeerApplicationError(42)); +} + +#[test] +fn connection_close_with_long_reason_string() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let now = now(); + // Create a long string and use it as the close reason. + let long_reason = String::from_utf8([0x61; 2048].to_vec()).unwrap(); + client.close(now, 42, long_reason); + + let out = client.process(None, now); + + server.process_input(&out.dgram().unwrap(), now); + assert_draining(&server, &Error::PeerApplicationError(42)); +} + +// During the handshake, an application close should be sanitized. +#[test] +fn early_application_close() { + let mut client = default_client(); + let mut server = default_server(); + + // One flight each. + let dgram = client.process(None, now()).dgram(); + assert!(dgram.is_some()); + let dgram = server.process(dgram.as_ref(), now()).dgram(); + assert!(dgram.is_some()); + + server.close(now(), 77, String::new()); + assert!(server.state().closed()); + let dgram = server.process(None, now()).dgram(); + assert!(dgram.is_some()); + + client.process_input(&dgram.unwrap(), now()); + assert_draining(&client, &Error::PeerError(ERROR_APPLICATION_CLOSE)); +} + +#[test] +fn bad_tls_version() { + let mut client = default_client(); + // Do a bad, bad thing. + client + .crypto + .tls + .set_option(neqo_crypto::Opt::Tls13CompatMode, true) + .unwrap(); + let mut server = default_server(); + + let dgram = client.process(None, now()).dgram(); + assert!(dgram.is_some()); + let dgram = server.process(dgram.as_ref(), now()).dgram(); + assert_eq!( + *server.state(), + State::Closed(ConnectionError::Transport(Error::ProtocolViolation)) + ); + assert!(dgram.is_some()); + client.process_input(&dgram.unwrap(), now()); + assert_draining(&client, &Error::PeerError(Error::ProtocolViolation.code())); +} + +/// Test the interaction between the loss recovery timer +/// and the closing timer. +#[test] +fn closing_timers_interation() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let mut now = now(); + + // We're going to induce time-based loss recovery so that timer is set. + let _p1 = send_something(&mut client, now); + let p2 = send_something(&mut client, now); + let ack = server.process(Some(&p2), now).dgram(); + assert!(ack.is_some()); // This is an ACK. + + // After processing the ACK, we should be on the loss recovery timer. + let cb = client.process(ack.as_ref(), now).callback(); + assert_ne!(cb, Duration::from_secs(0)); + now += cb; + + // Rather than let the timer pop, close the connection. + client.close(now, 0, ""); + let client_close = client.process(None, now).dgram(); + assert!(client_close.is_some()); + // This should now report the end of the closing period, not a + // zero-duration wait driven by the (now defunct) loss recovery timer. + let client_close_timer = client.process(None, now).callback(); + assert_ne!(client_close_timer, Duration::from_secs(0)); +} + +#[test] +fn closing_and_draining() { + const APP_ERROR: AppError = 7; + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + // Save a packet from the client for later. + let p1 = send_something(&mut client, now()); + + // Close the connection. + client.close(now(), APP_ERROR, ""); + let client_close = client.process(None, now()).dgram(); + assert!(client_close.is_some()); + let client_close_timer = client.process(None, now()).callback(); + assert_ne!(client_close_timer, Duration::from_secs(0)); + + // The client will spit out the same packet in response to anything it receives. + let p3 = send_something(&mut server, now()); + let client_close2 = client.process(Some(&p3), now()).dgram(); + assert_eq!( + client_close.as_ref().unwrap().len(), + client_close2.as_ref().unwrap().len() + ); + + // After this time, the client should transition to closed. + let end = client.process(None, now() + client_close_timer); + assert_eq!(end, Output::None); + assert_eq!( + *client.state(), + State::Closed(ConnectionError::Application(APP_ERROR)) + ); + + // When the server receives the close, it too should generate CONNECTION_CLOSE. + let server_close = server.process(client_close.as_ref(), now()).dgram(); + assert!(server.state().closed()); + assert!(server_close.is_some()); + // .. but it ignores any further close packets. + let server_close_timer = server.process(client_close2.as_ref(), now()).callback(); + assert_ne!(server_close_timer, Duration::from_secs(0)); + // Even a legitimate packet without a close in it. + let server_close_timer2 = server.process(Some(&p1), now()).callback(); + assert_eq!(server_close_timer, server_close_timer2); + + let end = server.process(None, now() + server_close_timer); + assert_eq!(end, Output::None); + assert_eq!( + *server.state(), + State::Closed(ConnectionError::Transport(Error::PeerApplicationError( + APP_ERROR + ))) + ); +} + +/// Test that a client can handle a stateless reset correctly. +#[test] +fn stateless_reset_client() { + let mut client = default_client(); + let mut server = default_server(); + server + .set_local_tparam( + tparams::STATELESS_RESET_TOKEN, + TransportParameter::Bytes(vec![77; 16]), + ) + .unwrap(); + connect_force_idle(&mut client, &mut server); + + client.process_input(&datagram(vec![77; 21]), now()); + assert_draining(&client, &Error::StatelessReset); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/datagram.rs b/third_party/rust/neqo-transport/src/connection/tests/datagram.rs new file mode 100644 index 0000000000..5b7b8dc0b4 --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/datagram.rs @@ -0,0 +1,620 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{cell::RefCell, convert::TryFrom, rc::Rc}; + +use neqo_common::event::Provider; +use test_fixture::now; + +use super::{ + assert_error, connect_force_idle, default_client, default_server, new_client, new_server, + AT_LEAST_PTO, +}; +use crate::{ + events::{ConnectionEvent, OutgoingDatagramOutcome}, + frame::FRAME_TYPE_DATAGRAM, + packet::PacketBuilder, + quic_datagrams::MAX_QUIC_DATAGRAM, + send_stream::{RetransmissionPriority, TransmissionPriority}, + Connection, ConnectionError, ConnectionParameters, Error, StreamType, +}; + +const DATAGRAM_LEN_MTU: u64 = 1310; +const DATA_MTU: &[u8] = &[1; 1310]; +const DATA_BIGGER_THAN_MTU: &[u8] = &[0; 2620]; +const DATAGRAM_LEN_SMALLER_THAN_MTU: u64 = 1200; +const DATA_SMALLER_THAN_MTU: &[u8] = &[0; 1200]; +const DATA_SMALLER_THAN_MTU_2: &[u8] = &[0; 600]; +const OUTGOING_QUEUE: usize = 2; + +struct InsertDatagram<'a> { + data: &'a [u8], +} + +impl crate::connection::test_internal::FrameWriter for InsertDatagram<'_> { + fn write_frames(&mut self, builder: &mut PacketBuilder) { + builder.encode_varint(FRAME_TYPE_DATAGRAM); + builder.encode(self.data); + } +} + +#[test] +fn datagram_disabled_both() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + assert_eq!(client.max_datagram_size(), Err(Error::NotAvailable)); + assert_eq!(server.max_datagram_size(), Err(Error::NotAvailable)); + assert_eq!( + client.send_datagram(DATA_SMALLER_THAN_MTU, None), + Err(Error::TooMuchData) + ); + assert_eq!(server.stats().frame_tx.datagram, 0); + assert_eq!( + server.send_datagram(DATA_SMALLER_THAN_MTU, None), + Err(Error::TooMuchData) + ); + assert_eq!(server.stats().frame_tx.datagram, 0); +} + +#[test] +fn datagram_enabled_on_client() { + let mut client = + new_client(ConnectionParameters::default().datagram_size(DATAGRAM_LEN_SMALLER_THAN_MTU)); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + assert_eq!(client.max_datagram_size(), Err(Error::NotAvailable)); + assert_eq!( + server.max_datagram_size(), + Ok(DATAGRAM_LEN_SMALLER_THAN_MTU) + ); + assert_eq!( + client.send_datagram(DATA_SMALLER_THAN_MTU, Some(1)), + Err(Error::TooMuchData) + ); + let dgram_sent = server.stats().frame_tx.datagram; + assert_eq!(server.send_datagram(DATA_SMALLER_THAN_MTU, Some(1)), Ok(())); + let out = server.process_output(now()).dgram().unwrap(); + assert_eq!(server.stats().frame_tx.datagram, dgram_sent + 1); + + client.process_input(&out, now()); + assert!(matches!( + client.next_event().unwrap(), + ConnectionEvent::Datagram(data) if data == DATA_SMALLER_THAN_MTU + )); +} + +#[test] +fn datagram_enabled_on_server() { + let mut client = default_client(); + let mut server = + new_server(ConnectionParameters::default().datagram_size(DATAGRAM_LEN_SMALLER_THAN_MTU)); + connect_force_idle(&mut client, &mut server); + + assert_eq!( + client.max_datagram_size(), + Ok(DATAGRAM_LEN_SMALLER_THAN_MTU) + ); + assert_eq!(server.max_datagram_size(), Err(Error::NotAvailable)); + assert_eq!( + server.send_datagram(DATA_SMALLER_THAN_MTU, Some(1)), + Err(Error::TooMuchData) + ); + let dgram_sent = client.stats().frame_tx.datagram; + assert_eq!(client.send_datagram(DATA_SMALLER_THAN_MTU, Some(1)), Ok(())); + let out = client.process_output(now()).dgram().unwrap(); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 1); + + server.process_input(&out, now()); + assert!(matches!( + server.next_event().unwrap(), + ConnectionEvent::Datagram(data) if data == DATA_SMALLER_THAN_MTU + )); +} + +fn connect_datagram() -> (Connection, Connection) { + let mut client = new_client( + ConnectionParameters::default() + .datagram_size(MAX_QUIC_DATAGRAM) + .outgoing_datagram_queue(OUTGOING_QUEUE), + ); + let mut server = new_server(ConnectionParameters::default().datagram_size(MAX_QUIC_DATAGRAM)); + connect_force_idle(&mut client, &mut server); + (client, server) +} + +#[test] +fn mtu_limit() { + let (client, server) = connect_datagram(); + + assert_eq!(client.max_datagram_size(), Ok(DATAGRAM_LEN_MTU)); + assert_eq!(server.max_datagram_size(), Ok(DATAGRAM_LEN_MTU)); +} + +#[test] +fn limit_data_size() { + let (mut client, mut server) = connect_datagram(); + + assert!(u64::try_from(DATA_BIGGER_THAN_MTU.len()).unwrap() > DATAGRAM_LEN_MTU); + // Datagram can be queued because they are smaller than allowed by the peer, + // but they cannot be sent. + assert_eq!(server.send_datagram(DATA_BIGGER_THAN_MTU, Some(1)), Ok(())); + + let dgram_dropped_s = server.stats().datagram_tx.dropped_too_big; + let dgram_sent_s = server.stats().frame_tx.datagram; + assert!(server.process_output(now()).dgram().is_none()); + assert_eq!( + server.stats().datagram_tx.dropped_too_big, + dgram_dropped_s + 1 + ); + assert_eq!(server.stats().frame_tx.datagram, dgram_sent_s); + assert!(matches!( + server.next_event().unwrap(), + ConnectionEvent::OutgoingDatagramOutcome { id, outcome } if id == 1 && outcome == OutgoingDatagramOutcome::DroppedTooBig + )); + + // The same test for the client side. + assert_eq!(client.send_datagram(DATA_BIGGER_THAN_MTU, Some(1)), Ok(())); + let dgram_sent_c = client.stats().frame_tx.datagram; + assert!(client.process_output(now()).dgram().is_none()); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent_c); + assert!(matches!( + client.next_event().unwrap(), + ConnectionEvent::OutgoingDatagramOutcome { id, outcome } if id == 1 && outcome == OutgoingDatagramOutcome::DroppedTooBig + )); +} + +#[test] +fn after_dgram_dropped_continue_writing_frames() { + let (mut client, _) = connect_datagram(); + + assert!(u64::try_from(DATA_BIGGER_THAN_MTU.len()).unwrap() > DATAGRAM_LEN_MTU); + // Datagram can be queued because they are smaller than allowed by the peer, + // but they cannot be sent. + assert_eq!(client.send_datagram(DATA_BIGGER_THAN_MTU, Some(1)), Ok(())); + assert_eq!(client.send_datagram(DATA_SMALLER_THAN_MTU, Some(2)), Ok(())); + + let datagram_dropped = |e| { + matches!( + e, + ConnectionEvent::OutgoingDatagramOutcome { id, outcome } if id == 1 && outcome == OutgoingDatagramOutcome::DroppedTooBig) + }; + + let dgram_dropped_c = client.stats().datagram_tx.dropped_too_big; + let dgram_sent_c = client.stats().frame_tx.datagram; + + assert!(client.process_output(now()).dgram().is_some()); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent_c + 1); + assert_eq!( + client.stats().datagram_tx.dropped_too_big, + dgram_dropped_c + 1 + ); + assert!(client.events().any(datagram_dropped)); +} + +#[test] +fn datagram_acked() { + let (mut client, mut server) = connect_datagram(); + + let dgram_sent = client.stats().frame_tx.datagram; + assert_eq!(client.send_datagram(DATA_SMALLER_THAN_MTU, Some(1)), Ok(())); + let out = client.process_output(now()).dgram(); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 1); + + let dgram_received = server.stats().frame_rx.datagram; + server.process_input(&out.unwrap(), now()); + assert_eq!(server.stats().frame_rx.datagram, dgram_received + 1); + let now = now() + AT_LEAST_PTO; + // Ack should be sent + let ack_sent = server.stats().frame_tx.ack; + let out = server.process_output(now).dgram(); + assert_eq!(server.stats().frame_tx.ack, ack_sent + 1); + + assert!(matches!( + server.next_event().unwrap(), + ConnectionEvent::Datagram(data) if data == DATA_SMALLER_THAN_MTU + )); + + client.process_input(&out.unwrap(), now); + assert!(matches!( + client.next_event().unwrap(), + ConnectionEvent::OutgoingDatagramOutcome { id, outcome } if id == 1 && outcome == OutgoingDatagramOutcome::Acked + )); +} + +fn send_packet_and_get_server_event( + client: &mut Connection, + server: &mut Connection, +) -> ConnectionEvent { + let out = client.process_output(now()).dgram(); + server.process_input(&out.unwrap(), now()); + let mut events: Vec<_> = server + .events() + .filter_map(|evt| match evt { + ConnectionEvent::RecvStreamReadable { .. } | ConnectionEvent::Datagram { .. } => { + Some(evt) + } + _ => None, + }) + .collect(); + // We should only get one event - either RecvStreamReadable or Datagram. + assert_eq!(events.len(), 1); + events.remove(0) +} + +/// Write a datagram that is big enough to fill a packet, but then see that +/// normal priority stream data is sent first. +#[test] +fn datagram_after_stream_data() { + let (mut client, mut server) = connect_datagram(); + + // Write a datagram first. + let dgram_sent = client.stats().frame_tx.datagram; + assert_eq!(client.send_datagram(DATA_MTU, Some(1)), Ok(())); + + // Create a stream with normal priority and send some data. + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + client.stream_send(stream_id, &[6; 1200]).unwrap(); + + assert!( + matches!(send_packet_and_get_server_event(&mut client, &mut server), ConnectionEvent::RecvStreamReadable { stream_id: s } if s == stream_id) + ); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent); + + if let ConnectionEvent::Datagram(data) = + &send_packet_and_get_server_event(&mut client, &mut server) + { + assert_eq!(data, DATA_MTU); + } else { + panic!(); + } + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 1); +} + +#[test] +fn datagram_before_stream_data() { + let (mut client, mut server) = connect_datagram(); + + // Create a stream with low priority and send some data before datagram. + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + client + .stream_priority( + stream_id, + TransmissionPriority::Low, + RetransmissionPriority::default(), + ) + .unwrap(); + client.stream_send(stream_id, &[6; 1200]).unwrap(); + + // Write a datagram. + let dgram_sent = client.stats().frame_tx.datagram; + assert_eq!(client.send_datagram(DATA_MTU, Some(1)), Ok(())); + + if let ConnectionEvent::Datagram(data) = + &send_packet_and_get_server_event(&mut client, &mut server) + { + assert_eq!(data, DATA_MTU); + } else { + panic!(); + } + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 1); + + assert!( + matches!(send_packet_and_get_server_event(&mut client, &mut server), ConnectionEvent::RecvStreamReadable { stream_id: s } if s == stream_id) + ); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 1); +} + +#[test] +fn datagram_lost() { + let (mut client, _) = connect_datagram(); + + let dgram_sent = client.stats().frame_tx.datagram; + assert_eq!(client.send_datagram(DATA_SMALLER_THAN_MTU, Some(1)), Ok(())); + let _out = client.process_output(now()).dgram(); // This packet will be lost. + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 1); + + // Wait for PTO + let now = now() + AT_LEAST_PTO; + let dgram_sent2 = client.stats().frame_tx.datagram; + let pings_sent = client.stats().frame_tx.ping; + let dgram_lost = client.stats().datagram_tx.lost; + let out = client.process_output(now).dgram(); + assert!(out.is_some()); // PING probing + // Datagram is not sent again. + assert_eq!(client.stats().frame_tx.ping, pings_sent + 1); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent2); + assert_eq!(client.stats().datagram_tx.lost, dgram_lost + 1); + + assert!(matches!( + client.next_event().unwrap(), + ConnectionEvent::OutgoingDatagramOutcome { id, outcome } if id == 1 && outcome == OutgoingDatagramOutcome::Lost + )); +} + +#[test] +fn datagram_sent_once() { + let (mut client, _) = connect_datagram(); + + let dgram_sent = client.stats().frame_tx.datagram; + assert_eq!(client.send_datagram(DATA_SMALLER_THAN_MTU, Some(1)), Ok(())); + let _out = client.process_output(now()).dgram(); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 1); + + // Call process_output again should not send any new Datagram. + assert!(client.process_output(now()).dgram().is_none()); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 1); +} + +#[test] +fn dgram_no_allowed() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + server.test_frame_writer = Some(Box::new(InsertDatagram { data: DATA_MTU })); + let out = server.process_output(now()).dgram().unwrap(); + server.test_frame_writer = None; + + client.process_input(&out, now()); + + assert_error( + &client, + &ConnectionError::Transport(Error::ProtocolViolation), + ); +} + +#[test] +#[allow(clippy::assertions_on_constants)] // this is a static assert, thanks +fn dgram_too_big() { + let mut client = + new_client(ConnectionParameters::default().datagram_size(DATAGRAM_LEN_SMALLER_THAN_MTU)); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + assert!(DATAGRAM_LEN_MTU > DATAGRAM_LEN_SMALLER_THAN_MTU); + server.test_frame_writer = Some(Box::new(InsertDatagram { data: DATA_MTU })); + let out = server.process_output(now()).dgram().unwrap(); + server.test_frame_writer = None; + + client.process_input(&out, now()); + + assert_error( + &client, + &ConnectionError::Transport(Error::ProtocolViolation), + ); +} + +#[test] +fn outgoing_datagram_queue_full() { + let (mut client, mut server) = connect_datagram(); + + let dgram_sent = client.stats().frame_tx.datagram; + assert_eq!(client.send_datagram(DATA_SMALLER_THAN_MTU, Some(1)), Ok(())); + assert_eq!( + client.send_datagram(DATA_SMALLER_THAN_MTU_2, Some(2)), + Ok(()) + ); + + // The outgoing datagram queue limit is 2, therefore the datagram with id 1 + // will be dropped after adding one more datagram. + let dgram_dropped = client.stats().datagram_tx.dropped_queue_full; + assert_eq!(client.send_datagram(DATA_MTU, Some(3)), Ok(())); + assert!(matches!( + client.next_event().unwrap(), + ConnectionEvent::OutgoingDatagramOutcome { id, outcome } if id == 1 && outcome == OutgoingDatagramOutcome::DroppedQueueFull + )); + assert_eq!( + client.stats().datagram_tx.dropped_queue_full, + dgram_dropped + 1 + ); + + // Send DATA_SMALLER_THAN_MTU_2 datagram + let out = client.process_output(now()).dgram(); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 1); + server.process_input(&out.unwrap(), now()); + assert!(matches!( + server.next_event().unwrap(), + ConnectionEvent::Datagram(data) if data == DATA_SMALLER_THAN_MTU_2 + )); + + // Send DATA_SMALLER_THAN_MTU_2 datagram + let dgram_sent2 = client.stats().frame_tx.datagram; + let out = client.process_output(now()).dgram(); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent2 + 1); + server.process_input(&out.unwrap(), now()); + assert!(matches!( + server.next_event().unwrap(), + ConnectionEvent::Datagram(data) if data == DATA_MTU + )); +} + +fn send_datagram(sender: &mut Connection, receiver: &mut Connection, data: &[u8]) { + let dgram_sent = sender.stats().frame_tx.datagram; + assert_eq!(sender.send_datagram(data, Some(1)), Ok(())); + let out = sender.process_output(now()).dgram().unwrap(); + assert_eq!(sender.stats().frame_tx.datagram, dgram_sent + 1); + + let dgram_received = receiver.stats().frame_rx.datagram; + receiver.process_input(&out, now()); + assert_eq!(receiver.stats().frame_rx.datagram, dgram_received + 1); +} + +#[test] +fn multiple_datagram_events() { + const DATA_SIZE: usize = 1200; + const MAX_QUEUE: usize = 3; + const FIRST_DATAGRAM: &[u8] = &[0; DATA_SIZE]; + const SECOND_DATAGRAM: &[u8] = &[1; DATA_SIZE]; + const THIRD_DATAGRAM: &[u8] = &[2; DATA_SIZE]; + const FOURTH_DATAGRAM: &[u8] = &[3; DATA_SIZE]; + + let mut client = new_client( + ConnectionParameters::default() + .datagram_size(u64::try_from(DATA_SIZE).unwrap()) + .incoming_datagram_queue(MAX_QUEUE), + ); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + send_datagram(&mut server, &mut client, FIRST_DATAGRAM); + send_datagram(&mut server, &mut client, SECOND_DATAGRAM); + send_datagram(&mut server, &mut client, THIRD_DATAGRAM); + + let mut datagrams = client.events().filter_map(|evt| { + if let ConnectionEvent::Datagram(d) = evt { + Some(d) + } else { + None + } + }); + assert_eq!(datagrams.next().unwrap(), FIRST_DATAGRAM); + assert_eq!(datagrams.next().unwrap(), SECOND_DATAGRAM); + assert_eq!(datagrams.next().unwrap(), THIRD_DATAGRAM); + assert!(datagrams.next().is_none()); + + // New events can be queued. + send_datagram(&mut server, &mut client, FOURTH_DATAGRAM); + let mut datagrams = client.events().filter_map(|evt| { + if let ConnectionEvent::Datagram(d) = evt { + Some(d) + } else { + None + } + }); + assert_eq!(datagrams.next().unwrap(), FOURTH_DATAGRAM); + assert!(datagrams.next().is_none()); +} + +#[test] +fn too_many_datagram_events() { + const DATA_SIZE: usize = 1200; + const MAX_QUEUE: usize = 2; + const FIRST_DATAGRAM: &[u8] = &[0; DATA_SIZE]; + const SECOND_DATAGRAM: &[u8] = &[1; DATA_SIZE]; + const THIRD_DATAGRAM: &[u8] = &[2; DATA_SIZE]; + const FOURTH_DATAGRAM: &[u8] = &[3; DATA_SIZE]; + + let mut client = new_client( + ConnectionParameters::default() + .datagram_size(u64::try_from(DATA_SIZE).unwrap()) + .incoming_datagram_queue(MAX_QUEUE), + ); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + send_datagram(&mut server, &mut client, FIRST_DATAGRAM); + send_datagram(&mut server, &mut client, SECOND_DATAGRAM); + send_datagram(&mut server, &mut client, THIRD_DATAGRAM); + + // Datagram with FIRST_DATAGRAM data will be dropped. + assert!(matches!( + client.next_event().unwrap(), + ConnectionEvent::Datagram(data) if data == SECOND_DATAGRAM + )); + assert!(matches!( + client.next_event().unwrap(), + ConnectionEvent::IncomingDatagramDropped + )); + assert!(matches!( + client.next_event().unwrap(), + ConnectionEvent::Datagram(data) if data == THIRD_DATAGRAM + )); + assert!(client.next_event().is_none()); + assert_eq!(client.stats().incoming_datagram_dropped, 1); + + // New events can be queued. + send_datagram(&mut server, &mut client, FOURTH_DATAGRAM); + assert!(matches!( + client.next_event().unwrap(), + ConnectionEvent::Datagram(data) if data == FOURTH_DATAGRAM + )); + assert!(client.next_event().is_none()); + assert_eq!(client.stats().incoming_datagram_dropped, 1); +} + +#[test] +fn multiple_quic_datagrams_in_one_packet() { + let (mut client, mut server) = connect_datagram(); + + let dgram_sent = client.stats().frame_tx.datagram; + // Enqueue 2 datagrams that can fit in a single packet. + assert_eq!( + client.send_datagram(DATA_SMALLER_THAN_MTU_2, Some(1)), + Ok(()) + ); + assert_eq!( + client.send_datagram(DATA_SMALLER_THAN_MTU_2, Some(2)), + Ok(()) + ); + + let out = client.process_output(now()).dgram(); + assert_eq!(client.stats().frame_tx.datagram, dgram_sent + 2); + server.process_input(&out.unwrap(), now()); + let datagram = |e: &_| matches!(e, ConnectionEvent::Datagram(..)); + assert_eq!(server.events().filter(datagram).count(), 2); +} + +/// Datagrams that are close to the capacity of the packet need special +/// handling. They need to use the packet-filling frame type and +/// they cannot allow other frames to follow. +#[test] +fn datagram_fill() { + struct PanickingFrameWriter {} + impl crate::connection::test_internal::FrameWriter for PanickingFrameWriter { + fn write_frames(&mut self, builder: &mut PacketBuilder) { + panic!( + "builder invoked with {} bytes remaining", + builder.remaining() + ); + } + } + struct TrackingFrameWriter { + called: Rc<RefCell<bool>>, + } + impl crate::connection::test_internal::FrameWriter for TrackingFrameWriter { + fn write_frames(&mut self, builder: &mut PacketBuilder) { + assert_eq!(builder.remaining(), 2); + *self.called.borrow_mut() = true; + } + } + + let (mut client, mut server) = connect_datagram(); + + // Work out how much space we have for a datagram. + let space = { + let p = client.paths.primary(); + let path = p.borrow(); + // Minimum overhead is connection ID length, 1 byte short header, 1 byte packet number, + // 1 byte for the DATAGRAM frame type, and 16 bytes for the AEAD. + path.mtu() - path.remote_cid().len() - 19 + }; + assert!(space >= 64); // Unlikely, but this test depends on the datagram being this large. + + // This should not be called. + client.test_frame_writer = Some(Box::new(PanickingFrameWriter {})); + + let buf = vec![9; space]; + // This will completely fill available space. + send_datagram(&mut client, &mut server, &buf); + // This will leave 1 byte free, but more frames won't be added in this space. + send_datagram(&mut client, &mut server, &buf[..buf.len() - 1]); + // This will leave 2 bytes free, which is enough space for a length field, + // but not enough space for another frame after that. + send_datagram(&mut client, &mut server, &buf[..buf.len() - 2]); + // Three bytes free will be space enough for a length frame, but not enough + // space left over for another frame (we need 2 bytes). + send_datagram(&mut client, &mut server, &buf[..buf.len() - 3]); + + // Four bytes free is enough space for another frame. + let called = Rc::new(RefCell::new(false)); + client.test_frame_writer = Some(Box::new(TrackingFrameWriter { + called: Rc::clone(&called), + })); + send_datagram(&mut client, &mut server, &buf[..buf.len() - 4]); + assert!(*called.borrow()); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/fuzzing.rs b/third_party/rust/neqo-transport/src/connection/tests/fuzzing.rs new file mode 100644 index 0000000000..5425e1a16e --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/fuzzing.rs @@ -0,0 +1,44 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +#![cfg_attr(feature = "deny-warnings", deny(warnings))] +#![warn(clippy::pedantic)] +#![cfg(feature = "fuzzing")] + +use neqo_crypto::FIXED_TAG_FUZZING; +use test_fixture::now; + +use super::{connect_force_idle, default_client, default_server}; +use crate::StreamType; + +#[test] +fn no_encryption() { + const DATA_CLIENT: &[u8] = &[2; 40]; + const DATA_SERVER: &[u8] = &[3; 50]; + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + + client.stream_send(stream_id, DATA_CLIENT).unwrap(); + let client_pkt = client.process_output(now()).dgram().unwrap(); + assert!(client_pkt[..client_pkt.len() - FIXED_TAG_FUZZING.len()].ends_with(DATA_CLIENT)); + + server.process_input(&client_pkt, now()); + let mut buf = vec![0; 100]; + let (len, _) = server.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(len, DATA_CLIENT.len()); + assert_eq!(&buf[..len], DATA_CLIENT); + server.stream_send(stream_id, DATA_SERVER).unwrap(); + let server_pkt = server.process_output(now()).dgram().unwrap(); + assert!(server_pkt[..server_pkt.len() - FIXED_TAG_FUZZING.len()].ends_with(DATA_SERVER)); + + client.process_input(&server_pkt, now()); + let (len, _) = client.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(len, DATA_SERVER.len()); + assert_eq!(&buf[..len], DATA_SERVER); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/handshake.rs b/third_party/rust/neqo-transport/src/connection/tests/handshake.rs new file mode 100644 index 0000000000..93385ac1bc --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/handshake.rs @@ -0,0 +1,1137 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{ + cell::RefCell, + convert::TryFrom, + mem, + net::{IpAddr, Ipv6Addr, SocketAddr}, + rc::Rc, + time::Duration, +}; + +use neqo_common::{event::Provider, qdebug, Datagram}; +use neqo_crypto::{ + constants::TLS_CHACHA20_POLY1305_SHA256, generate_ech_keys, AuthenticationStatus, +}; +use test_fixture::{ + self, addr, assertions, assertions::assert_coalesced_0rtt, datagram, fixture_init, now, + split_datagram, +}; + +use super::{ + super::{Connection, Output, State}, + assert_error, connect, connect_force_idle, connect_with_rtt, default_client, default_server, + get_tokens, handshake, maybe_authenticate, resumed_server, send_something, + CountingConnectionIdGenerator, AT_LEAST_PTO, DEFAULT_RTT, DEFAULT_STREAM_DATA, +}; +use crate::{ + connection::AddressValidation, + events::ConnectionEvent, + path::PATH_MTU_V6, + server::ValidateAddress, + tparams::{TransportParameter, MIN_ACK_DELAY}, + tracking::DEFAULT_ACK_DELAY, + ConnectionError, ConnectionParameters, EmptyConnectionIdGenerator, Error, StreamType, Version, +}; + +const ECH_CONFIG_ID: u8 = 7; +const ECH_PUBLIC_NAME: &str = "public.example"; + +#[test] +fn full_handshake() { + qdebug!("---- client: generate CH"); + let mut client = default_client(); + let out = client.process(None, now()); + assert!(out.as_dgram_ref().is_some()); + assert_eq!(out.as_dgram_ref().unwrap().len(), PATH_MTU_V6); + + qdebug!("---- server: CH -> SH, EE, CERT, CV, FIN"); + let mut server = default_server(); + let out = server.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_some()); + assert_eq!(out.as_dgram_ref().unwrap().len(), PATH_MTU_V6); + + qdebug!("---- client: cert verification"); + let out = client.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_some()); + + let out = server.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_none()); + + assert!(maybe_authenticate(&mut client)); + + qdebug!("---- client: SH..FIN -> FIN"); + let out = client.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_some()); + assert_eq!(*client.state(), State::Connected); + + qdebug!("---- server: FIN -> ACKS"); + let out = server.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_some()); + assert_eq!(*server.state(), State::Confirmed); + + qdebug!("---- client: ACKS -> 0"); + let out = client.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_none()); + assert_eq!(*client.state(), State::Confirmed); +} + +#[test] +fn handshake_failed_authentication() { + qdebug!("---- client: generate CH"); + let mut client = default_client(); + let out = client.process(None, now()); + assert!(out.as_dgram_ref().is_some()); + + qdebug!("---- server: CH -> SH, EE, CERT, CV, FIN"); + let mut server = default_server(); + let out = server.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_some()); + + qdebug!("---- client: cert verification"); + let out = client.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_some()); + + let out = server.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_none()); + + let authentication_needed = |e| matches!(e, ConnectionEvent::AuthenticationNeeded); + assert!(client.events().any(authentication_needed)); + qdebug!("---- client: Alert(certificate_revoked)"); + client.authenticated(AuthenticationStatus::CertRevoked, now()); + + qdebug!("---- client: -> Alert(certificate_revoked)"); + let out = client.process(None, now()); + assert!(out.as_dgram_ref().is_some()); + + qdebug!("---- server: Alert(certificate_revoked)"); + let out = server.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_some()); + assert_error(&client, &ConnectionError::Transport(Error::CryptoAlert(44))); + assert_error(&server, &ConnectionError::Transport(Error::PeerError(300))); +} + +#[test] +fn no_alpn() { + fixture_init(); + let mut client = Connection::new_client( + "example.com", + &["bad-alpn"], + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())), + addr(), + addr(), + ConnectionParameters::default(), + now(), + ) + .unwrap(); + let mut server = default_server(); + + handshake(&mut client, &mut server, now(), Duration::new(0, 0)); + // TODO (mt): errors are immediate, which means that we never send CONNECTION_CLOSE + // and the client never sees the server's rejection of its handshake. + // assert_error(&client, ConnectionError::Transport(Error::CryptoAlert(120))); + assert_error( + &server, + &ConnectionError::Transport(Error::CryptoAlert(120)), + ); +} + +#[test] +fn dup_server_flight1() { + qdebug!("---- client: generate CH"); + let mut client = default_client(); + let out = client.process(None, now()); + assert!(out.as_dgram_ref().is_some()); + assert_eq!(out.as_dgram_ref().unwrap().len(), PATH_MTU_V6); + qdebug!("Output={:0x?}", out.as_dgram_ref()); + + qdebug!("---- server: CH -> SH, EE, CERT, CV, FIN"); + let mut server = default_server(); + let out_to_rep = server.process(out.as_dgram_ref(), now()); + assert!(out_to_rep.as_dgram_ref().is_some()); + qdebug!("Output={:0x?}", out_to_rep.as_dgram_ref()); + + qdebug!("---- client: cert verification"); + let out = client.process(Some(out_to_rep.as_dgram_ref().unwrap()), now()); + assert!(out.as_dgram_ref().is_some()); + qdebug!("Output={:0x?}", out.as_dgram_ref()); + + let out = server.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_none()); + + assert!(maybe_authenticate(&mut client)); + + qdebug!("---- client: SH..FIN -> FIN"); + let out = client.process(None, now()); + assert!(out.as_dgram_ref().is_some()); + qdebug!("Output={:0x?}", out.as_dgram_ref()); + + assert_eq!(3, client.stats().packets_rx); + assert_eq!(0, client.stats().dups_rx); + assert_eq!(1, client.stats().dropped_rx); + + qdebug!("---- Dup, ignored"); + let out = client.process(out_to_rep.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_none()); + qdebug!("Output={:0x?}", out.as_dgram_ref()); + + // Four packets total received, 1 of them is a dup and one has been dropped because Initial keys + // are dropped. Add 2 counts of the padding that the server adds to Initial packets. + assert_eq!(6, client.stats().packets_rx); + assert_eq!(1, client.stats().dups_rx); + assert_eq!(3, client.stats().dropped_rx); +} + +// Test that we split crypto data if they cannot fit into one packet. +// To test this we will use a long server certificate. +#[test] +fn crypto_frame_split() { + let mut client = default_client(); + + let mut server = Connection::new_server( + test_fixture::LONG_CERT_KEYS, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())), + ConnectionParameters::default(), + ) + .expect("create a server"); + + let client1 = client.process(None, now()); + assert!(client1.as_dgram_ref().is_some()); + + // The entire server flight doesn't fit in a single packet because the + // certificate is large, therefore the server will produce 2 packets. + let server1 = server.process(client1.as_dgram_ref(), now()); + assert!(server1.as_dgram_ref().is_some()); + let server2 = server.process(None, now()); + assert!(server2.as_dgram_ref().is_some()); + + let client2 = client.process(server1.as_dgram_ref(), now()); + // This is an ack. + assert!(client2.as_dgram_ref().is_some()); + // The client might have the certificate now, so we can't guarantee that + // this will work. + let auth1 = maybe_authenticate(&mut client); + assert_eq!(*client.state(), State::Handshaking); + + // let server process the ack for the first packet. + let server3 = server.process(client2.as_dgram_ref(), now()); + assert!(server3.as_dgram_ref().is_none()); + + // Consume the second packet from the server. + let client3 = client.process(server2.as_dgram_ref(), now()); + + // Check authentication. + let auth2 = maybe_authenticate(&mut client); + assert!(auth1 ^ auth2); + // Now client has all data to finish handshake. + assert_eq!(*client.state(), State::Connected); + + let client4 = client.process(server3.as_dgram_ref(), now()); + // One of these will contain data depending on whether Authentication was completed + // after the first or second server packet. + assert!(client3.as_dgram_ref().is_some() ^ client4.as_dgram_ref().is_some()); + + mem::drop(server.process(client3.as_dgram_ref(), now())); + mem::drop(server.process(client4.as_dgram_ref(), now())); + + assert_eq!(*client.state(), State::Connected); + assert_eq!(*server.state(), State::Confirmed); +} + +/// Run a single ChaCha20-Poly1305 test and get a PTO. +#[test] +fn chacha20poly1305() { + let mut server = default_server(); + let mut client = Connection::new_client( + test_fixture::DEFAULT_SERVER_NAME, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(EmptyConnectionIdGenerator::default())), + addr(), + addr(), + ConnectionParameters::default(), + now(), + ) + .expect("create a default client"); + client.set_ciphers(&[TLS_CHACHA20_POLY1305_SHA256]).unwrap(); + connect_force_idle(&mut client, &mut server); +} + +/// Test that a server can send 0.5 RTT application data. +#[test] +fn send_05rtt() { + let mut client = default_client(); + let mut server = default_server(); + + let c1 = client.process(None, now()).dgram(); + assert!(c1.is_some()); + let s1 = server.process(c1.as_ref(), now()).dgram().unwrap(); + assert_eq!(s1.len(), PATH_MTU_V6); + + // The server should accept writes at this point. + let s2 = send_something(&mut server, now()); + + // Complete the handshake at the client. + client.process_input(&s1, now()); + maybe_authenticate(&mut client); + assert_eq!(*client.state(), State::Connected); + + // The client should receive the 0.5-RTT data now. + client.process_input(&s2, now()); + let mut buf = vec![0; DEFAULT_STREAM_DATA.len() + 1]; + let stream_id = client + .events() + .find_map(|e| { + if let ConnectionEvent::RecvStreamReadable { stream_id } = e { + Some(stream_id) + } else { + None + } + }) + .unwrap(); + let (l, ended) = client.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(&buf[..l], DEFAULT_STREAM_DATA); + assert!(ended); +} + +/// Test that a client buffers 0.5-RTT data when it arrives early. +#[test] +fn reorder_05rtt() { + let mut client = default_client(); + let mut server = default_server(); + + let c1 = client.process(None, now()).dgram(); + assert!(c1.is_some()); + let s1 = server.process(c1.as_ref(), now()).dgram().unwrap(); + + // The server should accept writes at this point. + let s2 = send_something(&mut server, now()); + + // We can't use the standard facility to complete the handshake, so + // drive it as aggressively as possible. + client.process_input(&s2, now()); + assert_eq!(client.stats().saved_datagrams, 1); + + // After processing the first packet, the client should go back and + // process the 0.5-RTT packet data, which should make data available. + client.process_input(&s1, now()); + // We can't use `maybe_authenticate` here as that consumes events. + client.authenticated(AuthenticationStatus::Ok, now()); + assert_eq!(*client.state(), State::Connected); + + let mut buf = vec![0; DEFAULT_STREAM_DATA.len() + 1]; + let stream_id = client + .events() + .find_map(|e| { + if let ConnectionEvent::RecvStreamReadable { stream_id } = e { + Some(stream_id) + } else { + None + } + }) + .unwrap(); + let (l, ended) = client.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(&buf[..l], DEFAULT_STREAM_DATA); + assert!(ended); +} + +#[test] +fn reorder_05rtt_with_0rtt() { + const RTT: Duration = Duration::from_millis(100); + + let mut client = default_client(); + let mut server = default_server(); + let validation = AddressValidation::new(now(), ValidateAddress::NoToken).unwrap(); + let validation = Rc::new(RefCell::new(validation)); + server.set_validation(Rc::clone(&validation)); + let mut now = connect_with_rtt(&mut client, &mut server, now(), RTT); + + // Include RTT in sending the ticket or the ticket age reported by the + // client is wrong, which causes the server to reject 0-RTT. + now += RTT / 2; + server.send_ticket(now, &[]).unwrap(); + let ticket = server.process_output(now).dgram().unwrap(); + now += RTT / 2; + client.process_input(&ticket, now); + + let token = get_tokens(&mut client).pop().unwrap(); + let mut client = default_client(); + client.enable_resumption(now, token).unwrap(); + let mut server = resumed_server(&client); + + // Send ClientHello and some 0-RTT. + let c1 = send_something(&mut client, now); + assertions::assert_coalesced_0rtt(&c1[..]); + // Drop the 0-RTT from the coalesced datagram, so that the server + // acknowledges the next 0-RTT packet. + let (c1, _) = split_datagram(&c1); + let c2 = send_something(&mut client, now); + + // Handle the first packet and send 0.5-RTT in response. Drop the response. + now += RTT / 2; + mem::drop(server.process(Some(&c1), now).dgram().unwrap()); + // The gap in 0-RTT will result in this 0.5 RTT containing an ACK. + server.process_input(&c2, now); + let s2 = send_something(&mut server, now); + + // Save the 0.5 RTT. + now += RTT / 2; + client.process_input(&s2, now); + assert_eq!(client.stats().saved_datagrams, 1); + + // Now PTO at the client and cause the server to re-send handshake packets. + now += AT_LEAST_PTO; + let c3 = client.process(None, now).dgram(); + assert_coalesced_0rtt(c3.as_ref().unwrap()); + + now += RTT / 2; + let s3 = server.process(c3.as_ref(), now).dgram().unwrap(); + + // The client should be able to process the 0.5 RTT now. + // This should contain an ACK, so we are processing an ACK from the past. + now += RTT / 2; + client.process_input(&s3, now); + maybe_authenticate(&mut client); + let c4 = client.process(None, now).dgram(); + assert_eq!(*client.state(), State::Connected); + assert_eq!(client.paths.rtt(), RTT); + + now += RTT / 2; + server.process_input(&c4.unwrap(), now); + assert_eq!(*server.state(), State::Confirmed); + // Don't check server RTT as it will be massively inflated by a + // poor initial estimate received when the server dropped the + // Initial packet number space. +} + +/// Test that a server that coalesces 0.5 RTT with handshake packets +/// doesn't cause the client to drop application data. +#[test] +fn coalesce_05rtt() { + const RTT: Duration = Duration::from_millis(100); + let mut client = default_client(); + let mut server = default_server(); + let mut now = now(); + + // The first exchange doesn't offer a chance for the server to send. + // So drop the server flight and wait for the PTO. + let c1 = client.process(None, now).dgram(); + assert!(c1.is_some()); + now += RTT / 2; + let s1 = server.process(c1.as_ref(), now).dgram(); + assert!(s1.is_some()); + + // Drop the server flight. Then send some data. + let stream_id = server.stream_create(StreamType::UniDi).unwrap(); + assert!(server.stream_send(stream_id, DEFAULT_STREAM_DATA).is_ok()); + assert!(server.stream_close_send(stream_id).is_ok()); + + // Now after a PTO the client can send another packet. + // The server should then send its entire flight again, + // including the application data, which it sends in a 1-RTT packet. + now += AT_LEAST_PTO; + let c2 = client.process(None, now).dgram(); + assert!(c2.is_some()); + now += RTT / 2; + let s2 = server.process(c2.as_ref(), now).dgram(); + // Even though there is a 1-RTT packet at the end of the datagram, the + // flight should be padded to full size. + assert_eq!(s2.as_ref().unwrap().len(), PATH_MTU_V6); + + // The client should process the datagram. It can't process the 1-RTT + // packet until authentication completes though. So it saves it. + now += RTT / 2; + assert_eq!(client.stats().dropped_rx, 0); + mem::drop(client.process(s2.as_ref(), now).dgram()); + // This packet will contain an ACK, but we can ignore it. + assert_eq!(client.stats().dropped_rx, 0); + assert_eq!(client.stats().packets_rx, 3); + assert_eq!(client.stats().saved_datagrams, 1); + + // After (successful) authentication, the packet is processed. + maybe_authenticate(&mut client); + let c3 = client.process(None, now).dgram(); + assert!(c3.is_some()); + assert_eq!(client.stats().dropped_rx, 0); // No Initial padding. + assert_eq!(client.stats().packets_rx, 4); + assert_eq!(client.stats().saved_datagrams, 1); + assert_eq!(client.stats().frame_rx.padding, 1); // Padding uses frames. + + // Allow the handshake to complete. + now += RTT / 2; + let s3 = server.process(c3.as_ref(), now).dgram(); + assert!(s3.is_some()); + assert_eq!(*server.state(), State::Confirmed); + now += RTT / 2; + mem::drop(client.process(s3.as_ref(), now).dgram()); + assert_eq!(*client.state(), State::Confirmed); + + assert_eq!(client.stats().dropped_rx, 0); // No dropped packets. +} + +#[test] +fn reorder_handshake() { + const RTT: Duration = Duration::from_millis(100); + let mut client = default_client(); + let mut server = default_server(); + let mut now = now(); + + let c1 = client.process(None, now).dgram(); + assert!(c1.is_some()); + + now += RTT / 2; + let s1 = server.process(c1.as_ref(), now).dgram(); + assert!(s1.is_some()); + + // Drop the Initial packet from this. + let (_, s_hs) = split_datagram(&s1.unwrap()); + assert!(s_hs.is_some()); + + // Pass just the handshake packet in and the client can't handle it yet. + // It can only send another Initial packet. + now += RTT / 2; + let dgram = client.process(s_hs.as_ref(), now).dgram(); + assertions::assert_initial(dgram.as_ref().unwrap(), false); + assert_eq!(client.stats().saved_datagrams, 1); + assert_eq!(client.stats().packets_rx, 1); + + // Get the server to try again. + // Though we currently allow the server to arm its PTO timer, use + // a second client Initial packet to cause it to send again. + now += AT_LEAST_PTO; + let c2 = client.process(None, now).dgram(); + now += RTT / 2; + let s2 = server.process(c2.as_ref(), now).dgram(); + assert!(s2.is_some()); + + let (s_init, s_hs) = split_datagram(&s2.unwrap()); + assert!(s_hs.is_some()); + + // Processing the Handshake packet first should save it. + now += RTT / 2; + client.process_input(&s_hs.unwrap(), now); + assert_eq!(client.stats().saved_datagrams, 2); + assert_eq!(client.stats().packets_rx, 2); + + client.process_input(&s_init, now); + // Each saved packet should now be "received" again. + assert_eq!(client.stats().packets_rx, 7); + maybe_authenticate(&mut client); + let c3 = client.process(None, now).dgram(); + assert!(c3.is_some()); + + // Note that though packets were saved and processed very late, + // they don't cause the RTT to change. + now += RTT / 2; + let s3 = server.process(c3.as_ref(), now).dgram(); + assert_eq!(*server.state(), State::Confirmed); + // Don't check server RTT estimate as it will be inflated due to + // it making a guess based on retransmissions when it dropped + // the Initial packet number space. + + now += RTT / 2; + client.process_input(&s3.unwrap(), now); + assert_eq!(*client.state(), State::Confirmed); + assert_eq!(client.paths.rtt(), RTT); +} + +#[test] +fn reorder_1rtt() { + const RTT: Duration = Duration::from_millis(100); + const PACKETS: usize = 4; // Many, but not enough to overflow cwnd. + let mut client = default_client(); + let mut server = default_server(); + let mut now = now(); + + let c1 = client.process(None, now).dgram(); + assert!(c1.is_some()); + + now += RTT / 2; + let s1 = server.process(c1.as_ref(), now).dgram(); + assert!(s1.is_some()); + + now += RTT / 2; + client.process_input(&s1.unwrap(), now); + maybe_authenticate(&mut client); + let c2 = client.process(None, now).dgram(); + assert!(c2.is_some()); + + // Now get a bunch of packets from the client. + // Give them to the server before giving it `c2`. + for _ in 0..PACKETS { + let d = send_something(&mut client, now); + server.process_input(&d, now + RTT / 2); + } + // The server has now received those packets, and saved them. + // The two extra received are Initial + the junk we use for padding. + assert_eq!(server.stats().packets_rx, PACKETS + 2); + assert_eq!(server.stats().saved_datagrams, PACKETS); + assert_eq!(server.stats().dropped_rx, 1); + + now += RTT / 2; + let s2 = server.process(c2.as_ref(), now).dgram(); + // The server has now received those packets, and saved them. + // The two additional are a Handshake and a 1-RTT (w/ NEW_CONNECTION_ID). + assert_eq!(server.stats().packets_rx, PACKETS * 2 + 4); + assert_eq!(server.stats().saved_datagrams, PACKETS); + assert_eq!(server.stats().dropped_rx, 1); + assert_eq!(*server.state(), State::Confirmed); + assert_eq!(server.paths.rtt(), RTT); + + now += RTT / 2; + client.process_input(&s2.unwrap(), now); + assert_eq!(client.paths.rtt(), RTT); + + // All the stream data that was sent should now be available. + let streams = server + .events() + .filter_map(|e| { + if let ConnectionEvent::RecvStreamReadable { stream_id } = e { + Some(stream_id) + } else { + None + } + }) + .collect::<Vec<_>>(); + assert_eq!(streams.len(), PACKETS); + for stream_id in streams { + let mut buf = vec![0; DEFAULT_STREAM_DATA.len() + 1]; + let (recvd, fin) = server.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(recvd, DEFAULT_STREAM_DATA.len()); + assert!(fin); + } +} + +#[cfg(not(feature = "fuzzing"))] +#[test] +fn corrupted_initial() { + let mut client = default_client(); + let mut server = default_server(); + let d = client.process(None, now()).dgram().unwrap(); + let mut corrupted = Vec::from(&d[..]); + // Find the last non-zero value and corrupt that. + let (idx, _) = corrupted + .iter() + .enumerate() + .rev() + .find(|(_, &v)| v != 0) + .unwrap(); + corrupted[idx] ^= 0x76; + let dgram = Datagram::new(d.source(), d.destination(), d.tos(), d.ttl(), corrupted); + server.process_input(&dgram, now()); + // The server should have received two packets, + // the first should be dropped, the second saved. + assert_eq!(server.stats().packets_rx, 2); + assert_eq!(server.stats().dropped_rx, 2); + assert_eq!(server.stats().saved_datagrams, 0); +} + +#[test] +// Absent path PTU discovery, max v6 packet size should be PATH_MTU_V6. +fn verify_pkt_honors_mtu() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + let now = now(); + + let res = client.process(None, now); + let idle_timeout = ConnectionParameters::default().get_idle_timeout(); + assert_eq!(res, Output::Callback(idle_timeout)); + + // Try to send a large stream and verify first packet is correctly sized + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(client.stream_send(stream_id, &[0xbb; 2000]).unwrap(), 2000); + let pkt0 = client.process(None, now); + assert!(matches!(pkt0, Output::Datagram(_))); + assert_eq!(pkt0.as_dgram_ref().unwrap().len(), PATH_MTU_V6); +} + +#[test] +fn extra_initial_hs() { + let mut client = default_client(); + let mut server = default_server(); + let mut now = now(); + + let c_init = client.process(None, now).dgram(); + assert!(c_init.is_some()); + now += DEFAULT_RTT / 2; + let s_init = server.process(c_init.as_ref(), now).dgram(); + assert!(s_init.is_some()); + now += DEFAULT_RTT / 2; + + // Drop the Initial packet, keep only the Handshake. + let (_, undecryptable) = split_datagram(&s_init.unwrap()); + assert!(undecryptable.is_some()); + + // Feed the same undecryptable packet into the client a few times. + // Do that EXTRA_INITIALS times and each time the client will emit + // another Initial packet. + for _ in 0..=super::super::EXTRA_INITIALS { + let c_init = client.process(undecryptable.as_ref(), now).dgram(); + assertions::assert_initial(c_init.as_ref().unwrap(), false); + now += DEFAULT_RTT / 10; + } + + // After EXTRA_INITIALS, the client stops sending Initial packets. + let nothing = client.process(undecryptable.as_ref(), now).dgram(); + assert!(nothing.is_none()); + + // Until PTO, where another Initial can be used to complete the handshake. + now += AT_LEAST_PTO; + let c_init = client.process(None, now).dgram(); + assertions::assert_initial(c_init.as_ref().unwrap(), false); + now += DEFAULT_RTT / 2; + let s_init = server.process(c_init.as_ref(), now).dgram(); + now += DEFAULT_RTT / 2; + client.process_input(&s_init.unwrap(), now); + maybe_authenticate(&mut client); + let c_fin = client.process_output(now).dgram(); + assert_eq!(*client.state(), State::Connected); + now += DEFAULT_RTT / 2; + server.process_input(&c_fin.unwrap(), now); + assert_eq!(*server.state(), State::Confirmed); +} + +#[test] +fn extra_initial_invalid_cid() { + let mut client = default_client(); + let mut server = default_server(); + let mut now = now(); + + let c_init = client.process(None, now).dgram(); + assert!(c_init.is_some()); + now += DEFAULT_RTT / 2; + let s_init = server.process(c_init.as_ref(), now).dgram(); + assert!(s_init.is_some()); + now += DEFAULT_RTT / 2; + + // If the client receives a packet that contains the wrong connection + // ID, it won't send another Initial. + let (_, hs) = split_datagram(&s_init.unwrap()); + let hs = hs.unwrap(); + let mut copy = hs.to_vec(); + assert_ne!(copy[5], 0); // The DCID should be non-zero length. + copy[6] ^= 0xc4; + let dgram_copy = Datagram::new(hs.destination(), hs.source(), hs.tos(), hs.ttl(), copy); + let nothing = client.process(Some(&dgram_copy), now).dgram(); + assert!(nothing.is_none()); +} + +#[test] +fn connect_one_version() { + fn connect_v(version: Version) { + fixture_init(); + let mut client = Connection::new_client( + test_fixture::DEFAULT_SERVER_NAME, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())), + addr(), + addr(), + ConnectionParameters::default().versions(version, vec![version]), + now(), + ) + .unwrap(); + let mut server = Connection::new_server( + test_fixture::DEFAULT_KEYS, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())), + ConnectionParameters::default().versions(version, vec![version]), + ) + .unwrap(); + connect_force_idle(&mut client, &mut server); + assert_eq!(client.version(), version); + assert_eq!(server.version(), version); + } + + for v in Version::all() { + println!("Connecting with {v:?}"); + connect_v(v); + } +} + +#[test] +fn anti_amplification() { + let mut client = default_client(); + let mut server = default_server(); + let mut now = now(); + + // With a gigantic transport parameter, the server is unable to complete + // the handshake within the amplification limit. + let very_big = TransportParameter::Bytes(vec![0; PATH_MTU_V6 * 3]); + server.set_local_tparam(0xce16, very_big).unwrap(); + + let c_init = client.process_output(now).dgram(); + now += DEFAULT_RTT / 2; + let s_init1 = server.process(c_init.as_ref(), now).dgram().unwrap(); + assert_eq!(s_init1.len(), PATH_MTU_V6); + let s_init2 = server.process_output(now).dgram().unwrap(); + assert_eq!(s_init2.len(), PATH_MTU_V6); + + // Skip the gap for pacing here. + let s_pacing = server.process_output(now).callback(); + assert_ne!(s_pacing, Duration::new(0, 0)); + now += s_pacing; + + let s_init3 = server.process_output(now).dgram().unwrap(); + assert_eq!(s_init3.len(), PATH_MTU_V6); + let cb = server.process_output(now).callback(); + assert_ne!(cb, Duration::new(0, 0)); + + now += DEFAULT_RTT / 2; + client.process_input(&s_init1, now); + client.process_input(&s_init2, now); + let ack_count = client.stats().frame_tx.ack; + let frame_count = client.stats().frame_tx.all; + let ack = client.process(Some(&s_init3), now).dgram().unwrap(); + assert!(!maybe_authenticate(&mut client)); // No need yet. + + // The client sends a padded datagram, with just ACK for Handshake. + assert_eq!(client.stats().frame_tx.ack, ack_count + 1); + assert_eq!(client.stats().frame_tx.all, frame_count + 1); + assert_ne!(ack.len(), PATH_MTU_V6); // Not padded (it includes Handshake). + + now += DEFAULT_RTT / 2; + let remainder = server.process(Some(&ack), now).dgram(); + + now += DEFAULT_RTT / 2; + client.process_input(&remainder.unwrap(), now); + assert!(maybe_authenticate(&mut client)); // OK, we have all of it. + let fin = client.process_output(now).dgram(); + assert_eq!(*client.state(), State::Connected); + + now += DEFAULT_RTT / 2; + server.process_input(&fin.unwrap(), now); + assert_eq!(*server.state(), State::Confirmed); +} + +#[cfg(not(feature = "fuzzing"))] +#[test] +fn garbage_initial() { + let mut client = default_client(); + let mut server = default_server(); + + let dgram = client.process_output(now()).dgram().unwrap(); + let (initial, rest) = split_datagram(&dgram); + let mut corrupted = Vec::from(&initial[..initial.len() - 1]); + corrupted.push(initial[initial.len() - 1] ^ 0xb7); + corrupted.extend_from_slice(rest.as_ref().map_or(&[], |r| &r[..])); + let garbage = datagram(corrupted); + assert_eq!(Output::None, server.process(Some(&garbage), now())); +} + +#[test] +fn drop_initial_packet_from_wrong_address() { + let mut client = default_client(); + let out = client.process(None, now()); + assert!(out.as_dgram_ref().is_some()); + + let mut server = default_server(); + let out = server.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_some()); + + let p = out.dgram().unwrap(); + let dgram = Datagram::new( + SocketAddr::new(IpAddr::V6(Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 2)), 443), + p.destination(), + p.tos(), + p.ttl(), + &p[..], + ); + + let out = client.process(Some(&dgram), now()); + assert!(out.as_dgram_ref().is_none()); +} + +#[test] +fn drop_handshake_packet_from_wrong_address() { + let mut client = default_client(); + let out = client.process(None, now()); + assert!(out.as_dgram_ref().is_some()); + + let mut server = default_server(); + let out = server.process(out.as_dgram_ref(), now()); + assert!(out.as_dgram_ref().is_some()); + + let (s_in, s_hs) = split_datagram(&out.dgram().unwrap()); + + // Pass the initial packet. + mem::drop(client.process(Some(&s_in), now()).dgram()); + + let p = s_hs.unwrap(); + let dgram = Datagram::new( + SocketAddr::new(IpAddr::V6(Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 2)), 443), + p.destination(), + p.tos(), + p.ttl(), + &p[..], + ); + + let out = client.process(Some(&dgram), now()); + assert!(out.as_dgram_ref().is_none()); +} + +#[test] +fn ech() { + let mut server = default_server(); + let (sk, pk) = generate_ech_keys().unwrap(); + server + .server_enable_ech(ECH_CONFIG_ID, ECH_PUBLIC_NAME, &sk, &pk) + .unwrap(); + + let mut client = default_client(); + client.client_enable_ech(server.ech_config()).unwrap(); + + connect(&mut client, &mut server); + + assert!(client.tls_info().unwrap().ech_accepted()); + assert!(server.tls_info().unwrap().ech_accepted()); + assert!(client.tls_preinfo().unwrap().ech_accepted().unwrap()); + assert!(server.tls_preinfo().unwrap().ech_accepted().unwrap()); +} + +fn damaged_ech_config(config: &[u8]) -> Vec<u8> { + let mut cfg = Vec::from(config); + // Ensure that the version and config_id is correct. + assert_eq!(cfg[2], 0xfe); + assert_eq!(cfg[3], 0x0d); + assert_eq!(cfg[6], ECH_CONFIG_ID); + // Change the config_id so that the server doesn't recognize it. + cfg[6] ^= 0x94; + cfg +} + +#[test] +fn ech_retry() { + fixture_init(); + let mut server = default_server(); + let (sk, pk) = generate_ech_keys().unwrap(); + server + .server_enable_ech(ECH_CONFIG_ID, ECH_PUBLIC_NAME, &sk, &pk) + .unwrap(); + + let mut client = default_client(); + client + .client_enable_ech(&damaged_ech_config(server.ech_config())) + .unwrap(); + + let dgram = client.process_output(now()).dgram(); + let dgram = server.process(dgram.as_ref(), now()).dgram(); + client.process_input(&dgram.unwrap(), now()); + let auth_event = ConnectionEvent::EchFallbackAuthenticationNeeded { + public_name: String::from(ECH_PUBLIC_NAME), + }; + assert!(client.events().any(|e| e == auth_event)); + client.authenticated(AuthenticationStatus::Ok, now()); + assert!(client.state().error().is_some()); + + // Tell the server about the error. + let dgram = client.process_output(now()).dgram(); + server.process_input(&dgram.unwrap(), now()); + assert_eq!( + server.state().error(), + Some(&ConnectionError::Transport(Error::PeerError(0x100 + 121))) + ); + + let Some(ConnectionError::Transport(Error::EchRetry(updated_config))) = client.state().error() + else { + panic!( + "Client state should be failed with EchRetry, is {:?}", + client.state() + ); + }; + + let mut server = default_server(); + server + .server_enable_ech(ECH_CONFIG_ID, ECH_PUBLIC_NAME, &sk, &pk) + .unwrap(); + let mut client = default_client(); + client.client_enable_ech(updated_config).unwrap(); + + connect(&mut client, &mut server); + + assert!(client.tls_info().unwrap().ech_accepted()); + assert!(server.tls_info().unwrap().ech_accepted()); + assert!(client.tls_preinfo().unwrap().ech_accepted().unwrap()); + assert!(server.tls_preinfo().unwrap().ech_accepted().unwrap()); +} + +#[test] +fn ech_retry_fallback_rejected() { + fixture_init(); + let mut server = default_server(); + let (sk, pk) = generate_ech_keys().unwrap(); + server + .server_enable_ech(ECH_CONFIG_ID, ECH_PUBLIC_NAME, &sk, &pk) + .unwrap(); + + let mut client = default_client(); + client + .client_enable_ech(&damaged_ech_config(server.ech_config())) + .unwrap(); + + let dgram = client.process_output(now()).dgram(); + let dgram = server.process(dgram.as_ref(), now()).dgram(); + client.process_input(&dgram.unwrap(), now()); + let auth_event = ConnectionEvent::EchFallbackAuthenticationNeeded { + public_name: String::from(ECH_PUBLIC_NAME), + }; + assert!(client.events().any(|e| e == auth_event)); + client.authenticated(AuthenticationStatus::PolicyRejection, now()); + assert!(client.state().error().is_some()); + + if let Some(ConnectionError::Transport(Error::EchRetry(_))) = client.state().error() { + panic!("Client should not get EchRetry error"); + } + + // Pass the error on. + let dgram = client.process_output(now()).dgram(); + server.process_input(&dgram.unwrap(), now()); + assert_eq!( + server.state().error(), + Some(&ConnectionError::Transport(Error::PeerError(298))) + ); // A bad_certificate alert. +} + +#[test] +fn bad_min_ack_delay() { + const EXPECTED_ERROR: ConnectionError = + ConnectionError::Transport(Error::TransportParameterError); + let mut server = default_server(); + let max_ad = u64::try_from(DEFAULT_ACK_DELAY.as_micros()).unwrap(); + server + .set_local_tparam(MIN_ACK_DELAY, TransportParameter::Integer(max_ad + 1)) + .unwrap(); + let mut client = default_client(); + + let dgram = client.process_output(now()).dgram(); + let dgram = server.process(dgram.as_ref(), now()).dgram(); + client.process_input(&dgram.unwrap(), now()); + client.authenticated(AuthenticationStatus::Ok, now()); + assert_eq!(client.state().error(), Some(&EXPECTED_ERROR)); + let dgram = client.process_output(now()).dgram(); + + server.process_input(&dgram.unwrap(), now()); + assert_eq!( + server.state().error(), + Some(&ConnectionError::Transport(Error::PeerError( + Error::TransportParameterError.code() + ))) + ); +} + +/// Ensure that the client probes correctly if it only receives Initial packets +/// from the server. +#[test] +fn only_server_initial() { + let mut server = default_server(); + let mut client = default_client(); + let mut now = now(); + + let client_dgram = client.process_output(now).dgram(); + + // Now fetch two flights of messages from the server. + let server_dgram1 = server.process(client_dgram.as_ref(), now).dgram(); + let server_dgram2 = server.process_output(now + AT_LEAST_PTO).dgram(); + + // Only pass on the Initial from the first. We should get a Handshake in return. + let (initial, handshake) = split_datagram(&server_dgram1.unwrap()); + assert!(handshake.is_some()); + + // The client will not acknowledge the Initial as it discards keys. + // It sends a Handshake probe instead, containing just a PING frame. + assert_eq!(client.stats().frame_tx.ping, 0); + let probe = client.process(Some(&initial), now).dgram(); + assertions::assert_handshake(&probe.unwrap()); + assert_eq!(client.stats().dropped_rx, 0); + assert_eq!(client.stats().frame_tx.ping, 1); + + let (initial, handshake) = split_datagram(&server_dgram2.unwrap()); + assert!(handshake.is_some()); + + // The same happens after a PTO, even though the client will discard the Initial packet. + now += AT_LEAST_PTO; + assert_eq!(client.stats().frame_tx.ping, 1); + let discarded = client.stats().dropped_rx; + let probe = client.process(Some(&initial), now).dgram(); + assertions::assert_handshake(&probe.unwrap()); + assert_eq!(client.stats().frame_tx.ping, 2); + assert_eq!(client.stats().dropped_rx, discarded + 1); + + // Pass the Handshake packet and complete the handshake. + client.process_input(&handshake.unwrap(), now); + maybe_authenticate(&mut client); + let dgram = client.process_output(now).dgram(); + let dgram = server.process(dgram.as_ref(), now).dgram(); + client.process_input(&dgram.unwrap(), now); + + assert_eq!(*client.state(), State::Confirmed); + assert_eq!(*server.state(), State::Confirmed); +} + +// Collect a few spare Initial packets as the handshake is exchanged. +// Later, replay those packets to see if they result in additional probes; they should not. +#[test] +fn no_extra_probes_after_confirmed() { + let mut server = default_server(); + let mut client = default_client(); + let mut now = now(); + + // First, collect a client Initial. + let spare_initial = client.process_output(now).dgram(); + assert!(spare_initial.is_some()); + + // Collect ANOTHER client Initial. + now += AT_LEAST_PTO; + let dgram = client.process_output(now).dgram(); + let (replay_initial, _) = split_datagram(dgram.as_ref().unwrap()); + + // Finally, run the handshake. + now += AT_LEAST_PTO * 2; + let dgram = client.process_output(now).dgram(); + let dgram = server.process(dgram.as_ref(), now).dgram(); + + // The server should have dropped the Initial keys now, so passing in the Initial + // should elicit a retransmit rather than having it completely ignored. + let spare_handshake = server.process(Some(&replay_initial), now).dgram(); + assert!(spare_handshake.is_some()); + + client.process_input(&dgram.unwrap(), now); + maybe_authenticate(&mut client); + let dgram = client.process_output(now).dgram(); + let dgram = server.process(dgram.as_ref(), now).dgram(); + client.process_input(&dgram.unwrap(), now); + + assert_eq!(*client.state(), State::Confirmed); + assert_eq!(*server.state(), State::Confirmed); + + let probe = server.process(spare_initial.as_ref(), now).dgram(); + assert!(probe.is_none()); + let probe = client.process(spare_handshake.as_ref(), now).dgram(); + assert!(probe.is_none()); +} + +#[test] +fn implicit_rtt_server() { + const RTT: Duration = Duration::from_secs(2); + let mut server = default_server(); + let mut client = default_client(); + let mut now = now(); + + let dgram = client.process_output(now).dgram(); + now += RTT / 2; + let dgram = server.process(dgram.as_ref(), now).dgram(); + now += RTT / 2; + let dgram = client.process(dgram.as_ref(), now).dgram(); + assertions::assert_handshake(dgram.as_ref().unwrap()); + now += RTT / 2; + server.process_input(&dgram.unwrap(), now); + + // The server doesn't receive any acknowledgments, but it can infer + // an RTT estimate from having discarded the Initial packet number space. + assert_eq!(server.stats().rtt, RTT); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/idle.rs b/third_party/rust/neqo-transport/src/connection/tests/idle.rs new file mode 100644 index 0000000000..c33726917a --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/idle.rs @@ -0,0 +1,752 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{ + mem, + time::{Duration, Instant}, +}; + +use neqo_common::{qtrace, Encoder}; +use test_fixture::{self, now, split_datagram}; + +use super::{ + super::{Connection, ConnectionParameters, IdleTimeout, Output, State}, + connect, connect_force_idle, connect_rtt_idle, connect_with_rtt, default_client, + default_server, maybe_authenticate, new_client, new_server, send_and_receive, send_something, + AT_LEAST_PTO, DEFAULT_STREAM_DATA, +}; +use crate::{ + packet::PacketBuilder, + stats::FrameStats, + stream_id::{StreamId, StreamType}, + tparams::{self, TransportParameter}, + tracking::PacketNumberSpace, +}; + +fn default_timeout() -> Duration { + ConnectionParameters::default().get_idle_timeout() +} + +fn test_idle_timeout(client: &mut Connection, server: &mut Connection, timeout: Duration) { + assert!(timeout > Duration::from_secs(1)); + connect_force_idle(client, server); + + let now = now(); + + let res = client.process(None, now); + assert_eq!(res, Output::Callback(timeout)); + + // Still connected after timeout-1 seconds. Idle timer not reset + mem::drop(client.process( + None, + now + timeout.checked_sub(Duration::from_secs(1)).unwrap(), + )); + assert!(matches!(client.state(), State::Confirmed)); + + mem::drop(client.process(None, now + timeout)); + + // Not connected after timeout. + assert!(matches!(client.state(), State::Closed(_))); +} + +#[test] +fn idle_timeout() { + let mut client = default_client(); + let mut server = default_server(); + test_idle_timeout(&mut client, &mut server, default_timeout()); +} + +#[test] +fn idle_timeout_custom_client() { + const IDLE_TIMEOUT: Duration = Duration::from_secs(5); + let mut client = new_client(ConnectionParameters::default().idle_timeout(IDLE_TIMEOUT)); + let mut server = default_server(); + test_idle_timeout(&mut client, &mut server, IDLE_TIMEOUT); +} + +#[test] +fn idle_timeout_custom_server() { + const IDLE_TIMEOUT: Duration = Duration::from_secs(5); + let mut client = default_client(); + let mut server = new_server(ConnectionParameters::default().idle_timeout(IDLE_TIMEOUT)); + test_idle_timeout(&mut client, &mut server, IDLE_TIMEOUT); +} + +#[test] +fn idle_timeout_custom_both() { + const LOWER_TIMEOUT: Duration = Duration::from_secs(5); + const HIGHER_TIMEOUT: Duration = Duration::from_secs(10); + let mut client = new_client(ConnectionParameters::default().idle_timeout(HIGHER_TIMEOUT)); + let mut server = new_server(ConnectionParameters::default().idle_timeout(LOWER_TIMEOUT)); + test_idle_timeout(&mut client, &mut server, LOWER_TIMEOUT); +} + +#[test] +fn asymmetric_idle_timeout() { + const LOWER_TIMEOUT_MS: u64 = 1000; + const LOWER_TIMEOUT: Duration = Duration::from_millis(LOWER_TIMEOUT_MS); + // Sanity check the constant. + assert!(LOWER_TIMEOUT < default_timeout()); + + let mut client = default_client(); + let mut server = default_server(); + + // Overwrite the default at the server. + server + .tps + .borrow_mut() + .local + .set_integer(tparams::IDLE_TIMEOUT, LOWER_TIMEOUT_MS); + server.idle_timeout = IdleTimeout::new(LOWER_TIMEOUT); + + // Now connect and force idleness manually. + // We do that by following what `force_idle` does and have each endpoint + // send two packets, which are delivered out of order. See `force_idle`. + connect(&mut client, &mut server); + let c1 = send_something(&mut client, now()); + let c2 = send_something(&mut client, now()); + server.process_input(&c2, now()); + server.process_input(&c1, now()); + let s1 = send_something(&mut server, now()); + let s2 = send_something(&mut server, now()); + client.process_input(&s2, now()); + let ack = client.process(Some(&s1), now()).dgram(); + assert!(ack.is_some()); + // Now both should have received ACK frames so should be idle. + assert_eq!( + server.process(ack.as_ref(), now()), + Output::Callback(LOWER_TIMEOUT) + ); + assert_eq!(client.process(None, now()), Output::Callback(LOWER_TIMEOUT)); +} + +#[test] +fn tiny_idle_timeout() { + const RTT: Duration = Duration::from_millis(500); + const LOWER_TIMEOUT_MS: u64 = 100; + const LOWER_TIMEOUT: Duration = Duration::from_millis(LOWER_TIMEOUT_MS); + // We won't respect a value that is lower than 3*PTO, sanity check. + assert!(LOWER_TIMEOUT < 3 * RTT); + + let mut client = default_client(); + let mut server = default_server(); + + // Overwrite the default at the server. + server + .set_local_tparam( + tparams::IDLE_TIMEOUT, + TransportParameter::Integer(LOWER_TIMEOUT_MS), + ) + .unwrap(); + server.idle_timeout = IdleTimeout::new(LOWER_TIMEOUT); + + // Now connect with an RTT and force idleness manually. + let mut now = connect_with_rtt(&mut client, &mut server, now(), RTT); + let c1 = send_something(&mut client, now); + let c2 = send_something(&mut client, now); + now += RTT / 2; + server.process_input(&c2, now); + server.process_input(&c1, now); + let s1 = send_something(&mut server, now); + let s2 = send_something(&mut server, now); + now += RTT / 2; + client.process_input(&s2, now); + let ack = client.process(Some(&s1), now).dgram(); + assert!(ack.is_some()); + + // The client should be idle now, but with a different timer. + if let Output::Callback(t) = client.process(None, now) { + assert!(t > LOWER_TIMEOUT); + } else { + panic!("Client not idle"); + } + + // The server should go idle after the ACK, but again with a larger timeout. + now += RTT / 2; + if let Output::Callback(t) = client.process(ack.as_ref(), now) { + assert!(t > LOWER_TIMEOUT); + } else { + panic!("Client not idle"); + } +} + +#[test] +fn idle_send_packet1() { + const DELTA: Duration = Duration::from_millis(10); + + let mut client = default_client(); + let mut server = default_server(); + let mut now = now(); + connect_force_idle(&mut client, &mut server); + + let timeout = client.process(None, now).callback(); + assert_eq!(timeout, default_timeout()); + + now += Duration::from_secs(10); + let dgram = send_and_receive(&mut client, &mut server, now); + assert!(dgram.is_some()); // the server will want to ACK, we can drop that. + + // Still connected after 39 seconds because idle timer reset by the + // outgoing packet. + now += default_timeout() - DELTA; + let dgram = client.process(None, now).dgram(); + assert!(dgram.is_some()); // PTO + assert!(client.state().connected()); + + // Not connected after 40 seconds. + now += DELTA; + let out = client.process(None, now); + assert!(matches!(out, Output::None)); + assert!(client.state().closed()); +} + +#[test] +fn idle_send_packet2() { + const GAP: Duration = Duration::from_secs(10); + const DELTA: Duration = Duration::from_millis(10); + + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + let mut now = now(); + + let timeout = client.process(None, now).callback(); + assert_eq!(timeout, default_timeout()); + + // First transmission at t=GAP. + now += GAP; + mem::drop(send_something(&mut client, now)); + + // Second transmission at t=2*GAP. + mem::drop(send_something(&mut client, now + GAP)); + assert!((GAP * 2 + DELTA) < default_timeout()); + + // Still connected just before GAP + default_timeout(). + now += default_timeout() - DELTA; + let dgram = client.process(None, now).dgram(); + assert!(dgram.is_some()); // PTO + assert!(matches!(client.state(), State::Confirmed)); + + // Not connected after 40 seconds because timer not reset by second + // outgoing packet + now += DELTA; + let out = client.process(None, now); + assert!(matches!(out, Output::None)); + assert!(matches!(client.state(), State::Closed(_))); +} + +#[test] +fn idle_recv_packet() { + const FUDGE: Duration = Duration::from_millis(10); + + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + let mut now = now(); + + let res = client.process(None, now); + assert_eq!(res, Output::Callback(default_timeout())); + + let stream = client.stream_create(StreamType::BiDi).unwrap(); + assert_eq!(stream, 0); + assert_eq!(client.stream_send(stream, b"hello").unwrap(), 5); + + // Respond with another packet. + // Note that it is important that this not result in the RTT increasing above 0. + // Otherwise, the eventual timeout will be extended (and we're not testing that). + now += Duration::from_secs(10); + let out = client.process(None, now); + server.process_input(&out.dgram().unwrap(), now); + assert_eq!(server.stream_send(stream, b"world").unwrap(), 5); + let out = server.process_output(now); + assert_ne!(out.as_dgram_ref(), None); + mem::drop(client.process(out.as_dgram_ref(), now)); + assert!(matches!(client.state(), State::Confirmed)); + + // Add a little less than the idle timeout and we're still connected. + now += default_timeout() - FUDGE; + mem::drop(client.process(None, now)); + assert!(matches!(client.state(), State::Confirmed)); + + now += FUDGE; + mem::drop(client.process(None, now)); + + assert!(matches!(client.state(), State::Closed(_))); +} + +/// Caching packets should not cause the connection to become idle. +/// This requires a few tricks to keep the connection from going +/// idle while preventing any progress on the handshake. +#[test] +fn idle_caching() { + let mut client = default_client(); + let mut server = default_server(); + let start = now(); + let mut builder = PacketBuilder::short(Encoder::new(), false, []); + + // Perform the first round trip, but drop the Initial from the server. + // The client then caches the Handshake packet. + let dgram = client.process_output(start).dgram(); + let dgram = server.process(dgram.as_ref(), start).dgram(); + let (_, handshake) = split_datagram(&dgram.unwrap()); + client.process_input(&handshake.unwrap(), start); + + // Perform an exchange and keep the connection alive. + // Only allow a packet containing a PING to pass. + let middle = start + AT_LEAST_PTO; + mem::drop(client.process_output(middle)); + let dgram = client.process_output(middle).dgram(); + + // Get the server to send its first probe and throw that away. + mem::drop(server.process_output(middle).dgram()); + // Now let the server process the client PING. This causes the server + // to send CRYPTO frames again, so manually extract and discard those. + let ping_before_s = server.stats().frame_rx.ping; + server.process_input(&dgram.unwrap(), middle); + assert_eq!(server.stats().frame_rx.ping, ping_before_s + 1); + let mut tokens = Vec::new(); + server + .crypto + .streams + .write_frame( + PacketNumberSpace::Initial, + &mut builder, + &mut tokens, + &mut FrameStats::default(), + ) + .unwrap(); + assert_eq!(tokens.len(), 1); + tokens.clear(); + server + .crypto + .streams + .write_frame( + PacketNumberSpace::Initial, + &mut builder, + &mut tokens, + &mut FrameStats::default(), + ) + .unwrap(); + assert!(tokens.is_empty()); + let dgram = server.process_output(middle).dgram(); + + // Now only allow the Initial packet from the server through; + // it shouldn't contain a CRYPTO frame. + let (initial, _) = split_datagram(&dgram.unwrap()); + let ping_before_c = client.stats().frame_rx.ping; + let ack_before = client.stats().frame_rx.ack; + client.process_input(&initial, middle); + assert_eq!(client.stats().frame_rx.ping, ping_before_c + 1); + assert_eq!(client.stats().frame_rx.ack, ack_before + 1); + + let end = start + default_timeout() + (AT_LEAST_PTO / 2); + // Now let the server Initial through, with the CRYPTO frame. + let dgram = server.process_output(end).dgram(); + let (initial, _) = split_datagram(&dgram.unwrap()); + neqo_common::qwarn!("client ingests initial, finally"); + mem::drop(client.process(Some(&initial), end)); + maybe_authenticate(&mut client); + let dgram = client.process_output(end).dgram(); + let dgram = server.process(dgram.as_ref(), end).dgram(); + client.process_input(&dgram.unwrap(), end); + assert_eq!(*client.state(), State::Confirmed); + assert_eq!(*server.state(), State::Confirmed); +} + +/// This function opens a bidirectional stream and leaves both endpoints +/// idle, with the stream left open. +/// The stream ID of that stream is returned (along with the new time). +fn create_stream_idle_rtt( + initiator: &mut Connection, + responder: &mut Connection, + mut now: Instant, + rtt: Duration, +) -> (Instant, StreamId) { + let check_idle = |endpoint: &mut Connection, now: Instant| { + let delay = endpoint.process_output(now).callback(); + qtrace!([endpoint], "idle timeout {:?}", delay); + if rtt < default_timeout() / 4 { + assert_eq!(default_timeout(), delay); + } else { + assert!(delay > default_timeout()); + } + }; + + // Exchange a message each way on a stream. + let stream = initiator.stream_create(StreamType::BiDi).unwrap(); + _ = initiator.stream_send(stream, DEFAULT_STREAM_DATA).unwrap(); + let req = initiator.process_output(now).dgram(); + now += rtt / 2; + responder.process_input(&req.unwrap(), now); + + // Reordering two packets from the responder forces the initiator to be idle. + _ = responder.stream_send(stream, DEFAULT_STREAM_DATA).unwrap(); + let resp1 = responder.process_output(now).dgram(); + _ = responder.stream_send(stream, DEFAULT_STREAM_DATA).unwrap(); + let resp2 = responder.process_output(now).dgram(); + + now += rtt / 2; + initiator.process_input(&resp2.unwrap(), now); + initiator.process_input(&resp1.unwrap(), now); + let ack = initiator.process_output(now).dgram(); + assert!(ack.is_some()); + check_idle(initiator, now); + + // Receiving the ACK should return the responder to idle too. + now += rtt / 2; + responder.process_input(&ack.unwrap(), now); + check_idle(responder, now); + + (now, stream) +} + +fn create_stream_idle(initiator: &mut Connection, responder: &mut Connection) -> StreamId { + let (_, stream) = create_stream_idle_rtt(initiator, responder, now(), Duration::new(0, 0)); + stream +} + +fn assert_idle(endpoint: &mut Connection, now: Instant, expected: Duration) { + assert_eq!(endpoint.process_output(now).callback(), expected); +} + +/// The creator of a stream marks it as important enough to use a keep-alive. +#[test] +fn keep_alive_initiator() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let stream = create_stream_idle(&mut server, &mut client); + let mut now = now(); + + // Marking the stream for keep-alive changes the idle timeout. + server.stream_keep_alive(stream, true).unwrap(); + assert_idle(&mut server, now, default_timeout() / 2); + + // Wait that long and the server should send a PING frame. + now += default_timeout() / 2; + let pings_before = server.stats().frame_tx.ping; + let ping = server.process_output(now).dgram(); + assert!(ping.is_some()); + assert_eq!(server.stats().frame_tx.ping, pings_before + 1); + + // Exchange ack for the PING. + let out = client.process(ping.as_ref(), now).dgram(); + let out = server.process(out.as_ref(), now).dgram(); + assert!(client.process(out.as_ref(), now).dgram().is_none()); + + // Check that there will be next keep-alive ping after default_timeout() / 2. + assert_idle(&mut server, now, default_timeout() / 2); + now += default_timeout() / 2; + let pings_before2 = server.stats().frame_tx.ping; + let ping = server.process_output(now).dgram(); + assert!(ping.is_some()); + assert_eq!(server.stats().frame_tx.ping, pings_before2 + 1); +} + +/// Test a keep-alive ping is retransmitted if lost. +#[test] +fn keep_alive_lost() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let stream = create_stream_idle(&mut server, &mut client); + let mut now = now(); + + // Marking the stream for keep-alive changes the idle timeout. + server.stream_keep_alive(stream, true).unwrap(); + assert_idle(&mut server, now, default_timeout() / 2); + + // Wait that long and the server should send a PING frame. + now += default_timeout() / 2; + let pings_before = server.stats().frame_tx.ping; + let ping = server.process_output(now).dgram(); + assert!(ping.is_some()); + assert_eq!(server.stats().frame_tx.ping, pings_before + 1); + + // Wait for ping to be marked lost. + assert!(server.process_output(now).callback() < AT_LEAST_PTO); + now += AT_LEAST_PTO; + let pings_before2 = server.stats().frame_tx.ping; + let ping = server.process_output(now).dgram(); + assert!(ping.is_some()); + assert_eq!(server.stats().frame_tx.ping, pings_before2 + 1); + + // Exchange ack for the PING. + let out = client.process(ping.as_ref(), now).dgram(); + + now += Duration::from_millis(20); + let out = server.process(out.as_ref(), now).dgram(); + + assert!(client.process(out.as_ref(), now).dgram().is_none()); + + // TODO: if we run server.process with current value of now, the server will + // return some small timeout for the recovry although it does not have + // any outstanding data. Therefore we call it after AT_LEAST_PTO. + now += AT_LEAST_PTO; + assert_idle(&mut server, now, default_timeout() / 2 - AT_LEAST_PTO); +} + +/// The other peer can also keep it alive. +#[test] +fn keep_alive_responder() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let stream = create_stream_idle(&mut server, &mut client); + let mut now = now(); + + // Marking the stream for keep-alive changes the idle timeout. + client.stream_keep_alive(stream, true).unwrap(); + assert_idle(&mut client, now, default_timeout() / 2); + + // Wait that long and the client should send a PING frame. + now += default_timeout() / 2; + let pings_before = client.stats().frame_tx.ping; + let ping = client.process_output(now).dgram(); + assert!(ping.is_some()); + assert_eq!(client.stats().frame_tx.ping, pings_before + 1); +} + +/// Unmark a stream as being keep-alive. +#[test] +fn keep_alive_unmark() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let stream = create_stream_idle(&mut client, &mut server); + + client.stream_keep_alive(stream, true).unwrap(); + assert_idle(&mut client, now(), default_timeout() / 2); + + client.stream_keep_alive(stream, false).unwrap(); + assert_idle(&mut client, now(), default_timeout()); +} + +/// The sender has something to send. Make it send it +/// and cause the receiver to become idle by sending something +/// else, reordering the packets, and consuming the ACK. +/// Note that the sender might not be idle if the thing that it +/// sends results in something in addition to an ACK. +fn transfer_force_idle(sender: &mut Connection, receiver: &mut Connection) { + let dgram = sender.process_output(now()).dgram(); + let chaff = send_something(sender, now()); + receiver.process_input(&chaff, now()); + receiver.process_input(&dgram.unwrap(), now()); + let ack = receiver.process_output(now()).dgram(); + sender.process_input(&ack.unwrap(), now()); +} + +/// Receiving the end of the stream stops keep-alives for that stream. +/// Even if that data hasn't been read. +#[test] +fn keep_alive_close() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let stream = create_stream_idle(&mut client, &mut server); + + client.stream_keep_alive(stream, true).unwrap(); + assert_idle(&mut client, now(), default_timeout() / 2); + + client.stream_close_send(stream).unwrap(); + transfer_force_idle(&mut client, &mut server); + assert_idle(&mut client, now(), default_timeout() / 2); + + server.stream_close_send(stream).unwrap(); + transfer_force_idle(&mut server, &mut client); + assert_idle(&mut client, now(), default_timeout()); +} + +/// Receiving `RESET_STREAM` stops keep-alives for that stream, but only once +/// the sending side is also closed. +#[test] +fn keep_alive_reset() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let stream = create_stream_idle(&mut client, &mut server); + + client.stream_keep_alive(stream, true).unwrap(); + assert_idle(&mut client, now(), default_timeout() / 2); + + client.stream_close_send(stream).unwrap(); + transfer_force_idle(&mut client, &mut server); + assert_idle(&mut client, now(), default_timeout() / 2); + + server.stream_reset_send(stream, 0).unwrap(); + transfer_force_idle(&mut server, &mut client); + assert_idle(&mut client, now(), default_timeout()); + + // The client will fade away from here. + let t = now() + (default_timeout() / 2); + assert_eq!(client.process_output(t).callback(), default_timeout() / 2); + let t = now() + default_timeout(); + assert_eq!(client.process_output(t), Output::None); +} + +/// Stopping sending also cancels the keep-alive. +#[test] +fn keep_alive_stop_sending() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let stream = create_stream_idle(&mut client, &mut server); + + client.stream_keep_alive(stream, true).unwrap(); + assert_idle(&mut client, now(), default_timeout() / 2); + + client.stream_close_send(stream).unwrap(); + client.stream_stop_sending(stream, 0).unwrap(); + transfer_force_idle(&mut client, &mut server); + // The server will have sent RESET_STREAM, which the client will + // want to acknowledge, so force that out. + let junk = send_something(&mut server, now()); + let ack = client.process(Some(&junk), now()).dgram(); + assert!(ack.is_some()); + + // Now the client should be idle. + assert_idle(&mut client, now(), default_timeout()); +} + +/// Multiple active streams are tracked properly. +#[test] +fn keep_alive_multiple_stop() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let stream = create_stream_idle(&mut client, &mut server); + + client.stream_keep_alive(stream, true).unwrap(); + assert_idle(&mut client, now(), default_timeout() / 2); + + let other = client.stream_create(StreamType::BiDi).unwrap(); + client.stream_keep_alive(other, true).unwrap(); + assert_idle(&mut client, now(), default_timeout() / 2); + + client.stream_keep_alive(stream, false).unwrap(); + assert_idle(&mut client, now(), default_timeout() / 2); + + client.stream_keep_alive(other, false).unwrap(); + assert_idle(&mut client, now(), default_timeout()); +} + +/// If the RTT is too long relative to the idle timeout, the keep-alive is large too. +#[test] +fn keep_alive_large_rtt() { + let mut client = default_client(); + let mut server = default_server(); + // Use an RTT that is large enough to cause the PTO timer to exceed half + // the idle timeout. + let rtt = default_timeout() * 3 / 4; + let now = connect_with_rtt(&mut client, &mut server, now(), rtt); + let (now, stream) = create_stream_idle_rtt(&mut server, &mut client, now, rtt); + + // Calculating PTO here is tricky as RTTvar has eroded after multiple round trips. + // Just check that the delay is larger than the baseline and the RTT. + for endpoint in &mut [client, server] { + endpoint.stream_keep_alive(stream, true).unwrap(); + let delay = endpoint.process_output(now).callback(); + qtrace!([endpoint], "new delay {:?}", delay); + assert!(delay > default_timeout() / 2); + assert!(delay > rtt); + } +} + +/// Only the recipient of a unidirectional stream can keep it alive. +#[test] +fn keep_alive_uni() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let stream = client.stream_create(StreamType::UniDi).unwrap(); + client.stream_keep_alive(stream, true).unwrap_err(); + _ = client.stream_send(stream, DEFAULT_STREAM_DATA).unwrap(); + let dgram = client.process_output(now()).dgram(); + + server.process_input(&dgram.unwrap(), now()); + server.stream_keep_alive(stream, true).unwrap(); +} + +/// Test a keep-alive ping is send if there are outstading ack-eliciting packets and that +/// the connection is closed after the idle timeout passes. +#[test] +fn keep_alive_with_ack_eliciting_packet_lost() { + const RTT: Duration = Duration::from_millis(500); // PTO will be ~1.1125s + + // The idle time out will be set to ~ 5 * PTO. (IDLE_TIMEOUT/2 > pto and IDLE_TIMEOUT/2 < pto + // + 2pto) After handshake all packets will be lost. The following steps will happen after + // the handshake: + // - data will be sent on a stream that is marked for keep-alive, (at start time) + // - PTO timer will trigger first, and the data will be retransmited toghether with a PING, (at + // the start time + pto) + // - keep-alive timer will trigger and a keep-alive PING will be sent, (at the start time + + // IDLE_TIMEOUT / 2) + // - PTO timer will trigger again. (at the start time + pto + 2*pto) + // - Idle time out will trigger (at the timeout + IDLE_TIMEOUT) + const IDLE_TIMEOUT: Duration = Duration::from_millis(6000); + + let mut client = new_client(ConnectionParameters::default().idle_timeout(IDLE_TIMEOUT)); + let mut server = default_server(); + let mut now = connect_rtt_idle(&mut client, &mut server, RTT); + // connect_rtt_idle increase now by RTT / 2; + now -= RTT / 2; + assert_idle(&mut client, now, IDLE_TIMEOUT); + + // Create a stream. + let stream = client.stream_create(StreamType::BiDi).unwrap(); + // Marking the stream for keep-alive changes the idle timeout. + client.stream_keep_alive(stream, true).unwrap(); + assert_idle(&mut client, now, IDLE_TIMEOUT / 2); + + // Send data on the stream that will be lost. + _ = client.stream_send(stream, DEFAULT_STREAM_DATA).unwrap(); + let _lost_packet = client.process_output(now).dgram(); + + let pto = client.process_output(now).callback(); + // Wait for packet to be marked lost. + assert!(pto < IDLE_TIMEOUT / 2); + now += pto; + let retransmit = client.process_output(now).dgram(); + assert!(retransmit.is_some()); + let retransmit = client.process_output(now).dgram(); + assert!(retransmit.is_some()); + + // The next callback should be for an idle PING. + assert_eq!( + client.process_output(now).callback(), + IDLE_TIMEOUT / 2 - pto + ); + + // Wait that long and the client should send a PING frame. + now += IDLE_TIMEOUT / 2 - pto; + let pings_before = client.stats().frame_tx.ping; + let ping = client.process_output(now).dgram(); + assert!(ping.is_some()); + assert_eq!(client.stats().frame_tx.ping, pings_before + 1); + + // The next callback is for a PTO, the PTO timer is 2 * pto now. + assert_eq!(client.process_output(now).callback(), pto * 2); + now += pto * 2; + // Now we will retransmit stream data. + let retransmit = client.process_output(now).dgram(); + assert!(retransmit.is_some()); + let retransmit = client.process_output(now).dgram(); + assert!(retransmit.is_some()); + + // The next callback will be an idle timeout. + assert_eq!( + client.process_output(now).callback(), + IDLE_TIMEOUT / 2 - 2 * pto + ); + + now += IDLE_TIMEOUT / 2 - 2 * pto; + let out = client.process_output(now); + assert!(matches!(out, Output::None)); + assert!(matches!(client.state(), State::Closed(_))); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/keys.rs b/third_party/rust/neqo-transport/src/connection/tests/keys.rs new file mode 100644 index 0000000000..c247bba670 --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/keys.rs @@ -0,0 +1,346 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::mem; + +use neqo_common::{qdebug, Datagram}; +use test_fixture::{self, now}; + +use super::{ + super::{ + super::{ConnectionError, ERROR_AEAD_LIMIT_REACHED}, + Connection, ConnectionParameters, Error, Output, State, StreamType, + }, + connect, connect_force_idle, default_client, default_server, maybe_authenticate, + send_and_receive, send_something, AT_LEAST_PTO, +}; +use crate::{ + crypto::{OVERWRITE_INVOCATIONS, UPDATE_WRITE_KEYS_AT}, + packet::PacketNumber, + path::PATH_MTU_V6, +}; + +fn check_discarded( + peer: &mut Connection, + pkt: &Datagram, + response: bool, + dropped: usize, + dups: usize, +) { + // Make sure to flush any saved datagrams before doing this. + mem::drop(peer.process_output(now())); + + let before = peer.stats(); + let out = peer.process(Some(pkt), now()); + assert_eq!(out.as_dgram_ref().is_some(), response); + let after = peer.stats(); + assert_eq!(dropped, after.dropped_rx - before.dropped_rx); + assert_eq!(dups, after.dups_rx - before.dups_rx); +} + +fn assert_update_blocked(c: &mut Connection) { + assert_eq!( + c.initiate_key_update().unwrap_err(), + Error::KeyUpdateBlocked + ); +} + +fn overwrite_invocations(n: PacketNumber) { + OVERWRITE_INVOCATIONS.with(|v| { + *v.borrow_mut() = Some(n); + }); +} + +#[test] +fn discarded_initial_keys() { + qdebug!("---- client: generate CH"); + let mut client = default_client(); + let init_pkt_c = client.process(None, now()).dgram(); + assert!(init_pkt_c.is_some()); + assert_eq!(init_pkt_c.as_ref().unwrap().len(), PATH_MTU_V6); + + qdebug!("---- server: CH -> SH, EE, CERT, CV, FIN"); + let mut server = default_server(); + let init_pkt_s = server.process(init_pkt_c.as_ref(), now()).dgram(); + assert!(init_pkt_s.is_some()); + + qdebug!("---- client: cert verification"); + let out = client.process(init_pkt_s.as_ref(), now()).dgram(); + assert!(out.is_some()); + + // The client has received a handshake packet. It will remove the Initial keys. + // We will check this by processing init_pkt_s a second time. + // The initial packet should be dropped. The packet contains a Handshake packet as well, which + // will be marked as dup. And it will contain padding, which will be "dropped". + // The client will generate a Handshake packet here to avoid stalling. + check_discarded(&mut client, &init_pkt_s.unwrap(), true, 2, 1); + + assert!(maybe_authenticate(&mut client)); + + // The server has not removed the Initial keys yet, because it has not yet received a Handshake + // packet from the client. + // We will check this by processing init_pkt_c a second time. + // The dropped packet is padding. The Initial packet has been mark dup. + check_discarded(&mut server, &init_pkt_c.clone().unwrap(), false, 1, 1); + + qdebug!("---- client: SH..FIN -> FIN"); + let out = client.process(None, now()).dgram(); + assert!(out.is_some()); + + // The server will process the first Handshake packet. + // After this the Initial keys will be dropped. + let out = server.process(out.as_ref(), now()).dgram(); + assert!(out.is_some()); + + // Check that the Initial keys are dropped at the server + // We will check this by processing init_pkt_c a third time. + // The Initial packet has been dropped and padding that follows it. + // There is no dups, everything has been dropped. + check_discarded(&mut server, &init_pkt_c.unwrap(), false, 1, 0); +} + +#[test] +fn key_update_client() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + let mut now = now(); + + assert_eq!(client.get_epochs(), (Some(3), Some(3))); // (write, read) + assert_eq!(server.get_epochs(), (Some(3), Some(3))); + + assert!(client.initiate_key_update().is_ok()); + assert_update_blocked(&mut client); + + // Initiating an update should only increase the write epoch. + let idle_timeout = ConnectionParameters::default().get_idle_timeout(); + assert_eq!(Output::Callback(idle_timeout), client.process(None, now)); + assert_eq!(client.get_epochs(), (Some(4), Some(3))); + + // Send something to propagate the update. + // Note that the server will acknowledge immediately when RTT is zero. + assert!(send_and_receive(&mut client, &mut server, now).is_some()); + + // The server should now be waiting to discharge read keys. + assert_eq!(server.get_epochs(), (Some(4), Some(3))); + let res = server.process(None, now); + if let Output::Callback(t) = res { + assert!(t < idle_timeout); + } else { + panic!("server should now be waiting to clear keys"); + } + + // Without having had time to purge old keys, more updates are blocked. + // The spec would permits it at this point, but we are more conservative. + assert_update_blocked(&mut client); + // The server can't update until it receives an ACK for a packet. + assert_update_blocked(&mut server); + + // Waiting now for at least a PTO should cause the server to drop old keys. + // But at this point the client hasn't received a key update from the server. + // It will be stuck with old keys. + now += AT_LEAST_PTO; + let dgram = client.process(None, now).dgram(); + assert!(dgram.is_some()); // Drop this packet. + assert_eq!(client.get_epochs(), (Some(4), Some(3))); + mem::drop(server.process(None, now)); + assert_eq!(server.get_epochs(), (Some(4), Some(4))); + + // Even though the server has updated, it hasn't received an ACK yet. + assert_update_blocked(&mut server); + + // Now get an ACK from the server. + // The previous PTO packet (see above) was dropped, so we should get an ACK here. + let dgram = send_and_receive(&mut client, &mut server, now); + assert!(dgram.is_some()); + let res = client.process(dgram.as_ref(), now); + // This is the first packet that the client has received from the server + // with new keys, so its read timer just started. + if let Output::Callback(t) = res { + assert!(t < ConnectionParameters::default().get_idle_timeout()); + } else { + panic!("client should now be waiting to clear keys"); + } + + assert_update_blocked(&mut client); + assert_eq!(client.get_epochs(), (Some(4), Some(3))); + // The server can't update until it gets something from the client. + assert_update_blocked(&mut server); + + now += AT_LEAST_PTO; + mem::drop(client.process(None, now)); + assert_eq!(client.get_epochs(), (Some(4), Some(4))); +} + +#[test] +fn key_update_consecutive() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let now = now(); + + assert!(server.initiate_key_update().is_ok()); + assert_eq!(server.get_epochs(), (Some(4), Some(3))); + + // Server sends something. + // Send twice and drop the first to induce an ACK from the client. + mem::drop(send_something(&mut server, now)); // Drop this. + + // Another packet from the server will cause the client to ACK and update keys. + let dgram = send_and_receive(&mut server, &mut client, now); + assert!(dgram.is_some()); + assert_eq!(client.get_epochs(), (Some(4), Some(3))); + + // Have the server process the ACK. + if let Output::Callback(_) = server.process(dgram.as_ref(), now) { + assert_eq!(server.get_epochs(), (Some(4), Some(3))); + // Now move the server temporarily into the future so that it + // rotates the keys. The client stays in the present. + mem::drop(server.process(None, now + AT_LEAST_PTO)); + assert_eq!(server.get_epochs(), (Some(4), Some(4))); + } else { + panic!("server should have a timer set"); + } + + // Now update keys on the server again. + assert!(server.initiate_key_update().is_ok()); + assert_eq!(server.get_epochs(), (Some(5), Some(4))); + + let dgram = send_something(&mut server, now + AT_LEAST_PTO); + + // However, as the server didn't wait long enough to update again, the + // client hasn't rotated its keys, so the packet gets dropped. + check_discarded(&mut client, &dgram, false, 1, 0); +} + +// Key updates can't be initiated too early. +#[test] +fn key_update_before_confirmed() { + let mut client = default_client(); + assert_update_blocked(&mut client); + let mut server = default_server(); + assert_update_blocked(&mut server); + + // Client Initial + let dgram = client.process(None, now()).dgram(); + assert!(dgram.is_some()); + assert_update_blocked(&mut client); + + // Server Initial + Handshake + let dgram = server.process(dgram.as_ref(), now()).dgram(); + assert!(dgram.is_some()); + assert_update_blocked(&mut server); + + // Client Handshake + client.process_input(&dgram.unwrap(), now()); + assert_update_blocked(&mut client); + + assert!(maybe_authenticate(&mut client)); + assert_update_blocked(&mut client); + + let dgram = client.process(None, now()).dgram(); + assert!(dgram.is_some()); + assert_update_blocked(&mut client); + + // Server HANDSHAKE_DONE + let dgram = server.process(dgram.as_ref(), now()).dgram(); + assert!(dgram.is_some()); + assert!(server.initiate_key_update().is_ok()); + + // Client receives HANDSHAKE_DONE + let dgram = client.process(dgram.as_ref(), now()).dgram(); + assert!(dgram.is_none()); + assert!(client.initiate_key_update().is_ok()); +} + +#[test] +fn exhaust_write_keys() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + overwrite_invocations(0); + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert!(client.stream_send(stream_id, b"explode!").is_ok()); + let dgram = client.process_output(now()).dgram(); + assert!(dgram.is_none()); + assert!(matches!( + client.state(), + State::Closed(ConnectionError::Transport(Error::KeysExhausted)) + )); +} + +#[test] +fn exhaust_read_keys() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + let dgram = send_something(&mut client, now()); + + overwrite_invocations(0); + let dgram = server.process(Some(&dgram), now()).dgram(); + assert!(matches!( + server.state(), + State::Closed(ConnectionError::Transport(Error::KeysExhausted)) + )); + + client.process_input(&dgram.unwrap(), now()); + assert!(matches!( + client.state(), + State::Draining { + error: ConnectionError::Transport(Error::PeerError(ERROR_AEAD_LIMIT_REACHED)), + .. + } + )); +} + +#[test] +fn automatic_update_write_keys() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + overwrite_invocations(UPDATE_WRITE_KEYS_AT); + mem::drop(send_something(&mut client, now())); + assert_eq!(client.get_epochs(), (Some(4), Some(3))); +} + +#[test] +fn automatic_update_write_keys_later() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + overwrite_invocations(UPDATE_WRITE_KEYS_AT + 2); + // No update after the first. + mem::drop(send_something(&mut client, now())); + assert_eq!(client.get_epochs(), (Some(3), Some(3))); + // The second will update though. + mem::drop(send_something(&mut client, now())); + assert_eq!(client.get_epochs(), (Some(4), Some(3))); +} + +#[test] +fn automatic_update_write_keys_blocked() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + // An outstanding key update will block the automatic update. + client.initiate_key_update().unwrap(); + + overwrite_invocations(UPDATE_WRITE_KEYS_AT); + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert!(client.stream_send(stream_id, b"explode!").is_ok()); + let dgram = client.process_output(now()).dgram(); + // Not being able to update is fatal. + assert!(dgram.is_none()); + assert!(matches!( + client.state(), + State::Closed(ConnectionError::Transport(Error::KeysExhausted)) + )); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/migration.rs b/third_party/rust/neqo-transport/src/connection/tests/migration.rs new file mode 100644 index 0000000000..8307a7dd84 --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/migration.rs @@ -0,0 +1,953 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{ + cell::RefCell, + net::{IpAddr, Ipv6Addr, SocketAddr}, + rc::Rc, + time::{Duration, Instant}, +}; + +use neqo_common::{Datagram, Decoder}; +use test_fixture::{ + self, addr, addr_v4, + assertions::{assert_v4_path, assert_v6_path}, + fixture_init, new_neqo_qlog, now, +}; + +use super::{ + super::{Connection, Output, State, StreamType}, + connect_fail, connect_force_idle, connect_rtt_idle, default_client, default_server, + maybe_authenticate, new_client, new_server, send_something, CountingConnectionIdGenerator, +}; +use crate::{ + cid::LOCAL_ACTIVE_CID_LIMIT, + connection::tests::send_something_paced, + frame::FRAME_TYPE_NEW_CONNECTION_ID, + packet::PacketBuilder, + path::{PATH_MTU_V4, PATH_MTU_V6}, + tparams::{self, PreferredAddress, TransportParameter}, + ConnectionError, ConnectionId, ConnectionIdDecoder, ConnectionIdGenerator, ConnectionIdRef, + ConnectionParameters, EmptyConnectionIdGenerator, Error, +}; + +/// This should be a valid-seeming transport parameter. +/// And it should have different values to `addr` and `addr_v4`. +const SAMPLE_PREFERRED_ADDRESS: &[u8] = &[ + 0xc0, 0x00, 0x02, 0x02, 0x01, 0xbb, 0xfe, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x01, 0xbb, 0x05, 0x01, 0x02, 0x03, 0x04, 0x05, 0x03, 0x03, + 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, +]; + +// These tests generally use two paths: +// The connection is established on a path with the same IPv6 address on both ends. +// Migrations move to a path with the same IPv4 address on both ends. +// This simplifies validation as the same assertions can be used for client and server. +// The risk is that there is a place where source/destination local/remote is inverted. + +fn loopback() -> SocketAddr { + SocketAddr::new(IpAddr::V6(Ipv6Addr::from(1)), 443) +} + +fn change_path(d: &Datagram, a: SocketAddr) -> Datagram { + Datagram::new(a, a, d.tos(), d.ttl(), &d[..]) +} + +fn new_port(a: SocketAddr) -> SocketAddr { + let (port, _) = a.port().overflowing_add(410); + SocketAddr::new(a.ip(), port) +} + +fn change_source_port(d: &Datagram) -> Datagram { + Datagram::new( + new_port(d.source()), + d.destination(), + d.tos(), + d.ttl(), + &d[..], + ) +} + +/// As these tests use a new path, that path often has a non-zero RTT. +/// Pacing can be a problem when testing that path. This skips time forward. +fn skip_pacing(c: &mut Connection, now: Instant) -> Instant { + let pacing = c.process_output(now).callback(); + assert_ne!(pacing, Duration::new(0, 0)); + now + pacing +} + +#[test] +fn rebinding_port() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + let dgram = send_something(&mut client, now()); + let dgram = change_source_port(&dgram); + + server.process_input(&dgram, now()); + // Have the server send something so that it generates a packet. + let stream_id = server.stream_create(StreamType::UniDi).unwrap(); + server.stream_close_send(stream_id).unwrap(); + let dgram = server.process_output(now()).dgram(); + let dgram = dgram.unwrap(); + assert_eq!(dgram.source(), addr()); + assert_eq!(dgram.destination(), new_port(addr())); +} + +/// This simulates an attack where a valid packet is forwarded on +/// a different path. This shows how both paths are probed and the +/// server eventually returns to the original path. +#[test] +fn path_forwarding_attack() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + let mut now = now(); + + let dgram = send_something(&mut client, now); + let dgram = change_path(&dgram, addr_v4()); + server.process_input(&dgram, now); + + // The server now probes the new (primary) path. + let new_probe = server.process_output(now).dgram().unwrap(); + assert_eq!(server.stats().frame_tx.path_challenge, 1); + assert_v4_path(&new_probe, false); // Can't be padded. + + // The server also probes the old path. + let old_probe = server.process_output(now).dgram().unwrap(); + assert_eq!(server.stats().frame_tx.path_challenge, 2); + assert_v6_path(&old_probe, true); + + // New data from the server is sent on the new path, but that is + // now constrained by the amplification limit. + let stream_id = server.stream_create(StreamType::UniDi).unwrap(); + server.stream_close_send(stream_id).unwrap(); + assert!(server.process_output(now).dgram().is_none()); + + // The client should respond to the challenge on the new path. + // The server couldn't pad, so the client is also amplification limited. + let new_resp = client.process(Some(&new_probe), now).dgram().unwrap(); + assert_eq!(client.stats().frame_rx.path_challenge, 1); + assert_eq!(client.stats().frame_tx.path_challenge, 1); + assert_eq!(client.stats().frame_tx.path_response, 1); + assert_v4_path(&new_resp, false); + + // The client also responds to probes on the old path. + let old_resp = client.process(Some(&old_probe), now).dgram().unwrap(); + assert_eq!(client.stats().frame_rx.path_challenge, 2); + assert_eq!(client.stats().frame_tx.path_challenge, 1); + assert_eq!(client.stats().frame_tx.path_response, 2); + assert_v6_path(&old_resp, true); + + // But the client still sends data on the old path. + let client_data1 = send_something(&mut client, now); + assert_v6_path(&client_data1, false); // Just data. + + // Receiving the PATH_RESPONSE from the client opens the amplification + // limit enough for the server to respond. + // This is padded because it includes PATH_CHALLENGE. + let server_data1 = server.process(Some(&new_resp), now).dgram().unwrap(); + assert_v4_path(&server_data1, true); + assert_eq!(server.stats().frame_tx.path_challenge, 3); + + // The client responds to this probe on the new path. + client.process_input(&server_data1, now); + let stream_before = client.stats().frame_tx.stream; + let padded_resp = send_something(&mut client, now); + assert_eq!(stream_before, client.stats().frame_tx.stream); + assert_v4_path(&padded_resp, true); // This is padded! + + // But new data from the client stays on the old path. + let client_data2 = client.process_output(now).dgram().unwrap(); + assert_v6_path(&client_data2, false); + + // The server keeps sending on the new path. + now = skip_pacing(&mut server, now); + let server_data2 = send_something(&mut server, now); + assert_v4_path(&server_data2, false); + + // Until new data is received from the client on the old path. + server.process_input(&client_data2, now); + // The server sends a probe on the "old" path. + let server_data3 = send_something(&mut server, now); + assert_v4_path(&server_data3, true); + // But switches data transmission to the "new" path. + let server_data4 = server.process_output(now).dgram().unwrap(); + assert_v6_path(&server_data4, false); +} + +#[test] +fn migrate_immediate() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + let now = now(); + + client + .migrate(Some(addr_v4()), Some(addr_v4()), true, now) + .unwrap(); + + let client1 = send_something(&mut client, now); + assert_v4_path(&client1, true); // Contains PATH_CHALLENGE. + let client2 = send_something(&mut client, now); + assert_v4_path(&client2, false); // Doesn't. + + let server_delayed = send_something(&mut server, now); + + // The server accepts the first packet and migrates (but probes). + let server1 = server.process(Some(&client1), now).dgram().unwrap(); + assert_v4_path(&server1, true); + let server2 = server.process_output(now).dgram().unwrap(); + assert_v6_path(&server2, true); + + // The second packet has no real effect, it just elicits an ACK. + let all_before = server.stats().frame_tx.all; + let ack_before = server.stats().frame_tx.ack; + let server3 = server.process(Some(&client2), now).dgram(); + assert!(server3.is_some()); + assert_eq!(server.stats().frame_tx.all, all_before + 1); + assert_eq!(server.stats().frame_tx.ack, ack_before + 1); + + // Receiving a packet sent by the server before migration doesn't change path. + client.process_input(&server_delayed, now); + // The client has sent two unpaced packets and this new path has no RTT estimate + // so this might be paced. + let (client3, _t) = send_something_paced(&mut client, now, true); + assert_v4_path(&client3, false); +} + +/// RTT estimates for paths should be preserved across migrations. +#[test] +fn migrate_rtt() { + const RTT: Duration = Duration::from_millis(20); + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, RTT); + + client + .migrate(Some(addr_v4()), Some(addr_v4()), true, now) + .unwrap(); + // The RTT might be increased for the new path, so allow a little flexibility. + let rtt = client.paths.rtt(); + assert!(rtt > RTT); + assert!(rtt < RTT * 2); +} + +#[test] +fn migrate_immediate_fail() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + let mut now = now(); + + client + .migrate(Some(addr_v4()), Some(addr_v4()), true, now) + .unwrap(); + + let probe = client.process_output(now).dgram().unwrap(); + assert_v4_path(&probe, true); // Contains PATH_CHALLENGE. + + for _ in 0..2 { + let cb = client.process_output(now).callback(); + assert_ne!(cb, Duration::new(0, 0)); + now += cb; + + let before = client.stats().frame_tx; + let probe = client.process_output(now).dgram().unwrap(); + assert_v4_path(&probe, true); // Contains PATH_CHALLENGE. + let after = client.stats().frame_tx; + assert_eq!(after.path_challenge, before.path_challenge + 1); + assert_eq!(after.padding, before.padding + 1); + assert_eq!(after.all, before.all + 2); + + // This might be a PTO, which will result in sending a probe. + if let Some(probe) = client.process_output(now).dgram() { + assert_v4_path(&probe, false); // Contains PATH_CHALLENGE. + let after = client.stats().frame_tx; + assert_eq!(after.ping, before.ping + 1); + assert_eq!(after.all, before.all + 3); + } + } + + let pto = client.process_output(now).callback(); + assert_ne!(pto, Duration::new(0, 0)); + now += pto; + + // The client should fall back to the original path and retire the connection ID. + let fallback = client.process_output(now).dgram(); + assert_v6_path(&fallback.unwrap(), false); + assert_eq!(client.stats().frame_tx.retire_connection_id, 1); +} + +/// Migrating to the same path shouldn't do anything special, +/// except that the path is probed. +#[test] +fn migrate_same() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + let now = now(); + + client + .migrate(Some(addr()), Some(addr()), true, now) + .unwrap(); + + let probe = client.process_output(now).dgram().unwrap(); + assert_v6_path(&probe, true); // Contains PATH_CHALLENGE. + assert_eq!(client.stats().frame_tx.path_challenge, 1); + + let resp = server.process(Some(&probe), now).dgram().unwrap(); + assert_v6_path(&resp, true); + assert_eq!(server.stats().frame_tx.path_response, 1); + assert_eq!(server.stats().frame_tx.path_challenge, 0); + + // Everything continues happily. + client.process_input(&resp, now); + let contd = send_something(&mut client, now); + assert_v6_path(&contd, false); +} + +/// Migrating to the same path, if it fails, causes the connection to fail. +#[test] +fn migrate_same_fail() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + let mut now = now(); + + client + .migrate(Some(addr()), Some(addr()), true, now) + .unwrap(); + + let probe = client.process_output(now).dgram().unwrap(); + assert_v6_path(&probe, true); // Contains PATH_CHALLENGE. + + for _ in 0..2 { + let cb = client.process_output(now).callback(); + assert_ne!(cb, Duration::new(0, 0)); + now += cb; + + let before = client.stats().frame_tx; + let probe = client.process_output(now).dgram().unwrap(); + assert_v6_path(&probe, true); // Contains PATH_CHALLENGE. + let after = client.stats().frame_tx; + assert_eq!(after.path_challenge, before.path_challenge + 1); + assert_eq!(after.padding, before.padding + 1); + assert_eq!(after.all, before.all + 2); + + // This might be a PTO, which will result in sending a probe. + if let Some(probe) = client.process_output(now).dgram() { + assert_v6_path(&probe, false); // Contains PATH_CHALLENGE. + let after = client.stats().frame_tx; + assert_eq!(after.ping, before.ping + 1); + assert_eq!(after.all, before.all + 3); + } + } + + let pto = client.process_output(now).callback(); + assert_ne!(pto, Duration::new(0, 0)); + now += pto; + + // The client should mark this path as failed and close immediately. + let res = client.process_output(now); + assert!(matches!(res, Output::None)); + assert!(matches!( + client.state(), + State::Closed(ConnectionError::Transport(Error::NoAvailablePath)) + )); +} + +/// This gets the connection ID from a datagram using the default +/// connection ID generator/decoder. +fn get_cid(d: &Datagram) -> ConnectionIdRef { + let gen = CountingConnectionIdGenerator::default(); + assert_eq!(d[0] & 0x80, 0); // Only support short packets for now. + gen.decode_cid(&mut Decoder::from(&d[1..])).unwrap() +} + +fn migration(mut client: Connection) { + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + let now = now(); + + client + .migrate(Some(addr_v4()), Some(addr_v4()), false, now) + .unwrap(); + + let probe = client.process_output(now).dgram().unwrap(); + assert_v4_path(&probe, true); // Contains PATH_CHALLENGE. + assert_eq!(client.stats().frame_tx.path_challenge, 1); + let probe_cid = ConnectionId::from(get_cid(&probe)); + + let resp = server.process(Some(&probe), now).dgram().unwrap(); + assert_v4_path(&resp, true); + assert_eq!(server.stats().frame_tx.path_response, 1); + assert_eq!(server.stats().frame_tx.path_challenge, 1); + + // Data continues to be exchanged on the new path. + let client_data = send_something(&mut client, now); + assert_ne!(get_cid(&client_data), probe_cid); + assert_v6_path(&client_data, false); + server.process_input(&client_data, now); + let server_data = send_something(&mut server, now); + assert_v6_path(&server_data, false); + + // Once the client receives the probe response, it migrates to the new path. + client.process_input(&resp, now); + assert_eq!(client.stats().frame_rx.path_challenge, 1); + let migrate_client = send_something(&mut client, now); + assert_v4_path(&migrate_client, true); // Responds to server probe. + + // The server now sees the migration and will switch over. + // However, it will probe the old path again, even though it has just + // received a response to its last probe, because it needs to verify + // that the migration is genuine. + server.process_input(&migrate_client, now); + let stream_before = server.stats().frame_tx.stream; + let probe_old_server = send_something(&mut server, now); + // This is just the double-check probe; no STREAM frames. + assert_v6_path(&probe_old_server, true); + assert_eq!(server.stats().frame_tx.path_challenge, 2); + assert_eq!(server.stats().frame_tx.stream, stream_before); + + // The server then sends data on the new path. + let migrate_server = server.process_output(now).dgram().unwrap(); + assert_v4_path(&migrate_server, false); + assert_eq!(server.stats().frame_tx.path_challenge, 2); + assert_eq!(server.stats().frame_tx.stream, stream_before + 1); + + // The client receives these checks and responds to the probe, but uses the new path. + client.process_input(&migrate_server, now); + client.process_input(&probe_old_server, now); + let old_probe_resp = send_something(&mut client, now); + assert_v6_path(&old_probe_resp, true); + let client_confirmation = client.process_output(now).dgram().unwrap(); + assert_v4_path(&client_confirmation, false); + + // The server has now sent 2 packets, so it is blocked on the pacer. Wait. + let server_pacing = server.process_output(now).callback(); + assert_ne!(server_pacing, Duration::new(0, 0)); + // ... then confirm that the server sends on the new path still. + let server_confirmation = send_something(&mut server, now + server_pacing); + assert_v4_path(&server_confirmation, false); +} + +#[test] +fn migration_graceful() { + migration(default_client()); +} + +/// A client should be able to migrate when it has a zero-length connection ID. +#[test] +fn migration_client_empty_cid() { + fixture_init(); + let client = Connection::new_client( + test_fixture::DEFAULT_SERVER_NAME, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(EmptyConnectionIdGenerator::default())), + addr(), + addr(), + ConnectionParameters::default(), + now(), + ) + .unwrap(); + migration(client); +} + +/// Drive the handshake in the most expeditious fashion. +/// Returns the packet containing `HANDSHAKE_DONE` from the server. +fn fast_handshake(client: &mut Connection, server: &mut Connection) -> Option<Datagram> { + let dgram = client.process_output(now()).dgram(); + let dgram = server.process(dgram.as_ref(), now()).dgram(); + client.process_input(&dgram.unwrap(), now()); + assert!(maybe_authenticate(client)); + let dgram = client.process_output(now()).dgram(); + server.process(dgram.as_ref(), now()).dgram() +} + +fn preferred_address(hs_client: SocketAddr, hs_server: SocketAddr, preferred: SocketAddr) { + let mtu = match hs_client.ip() { + IpAddr::V4(_) => PATH_MTU_V4, + IpAddr::V6(_) => PATH_MTU_V6, + }; + let assert_orig_path = |d: &Datagram, full_mtu: bool| { + assert_eq!( + d.destination(), + if d.source() == hs_client { + hs_server + } else if d.source() == hs_server { + hs_client + } else { + panic!(); + } + ); + if full_mtu { + assert_eq!(d.len(), mtu); + } + }; + let assert_toward_spa = |d: &Datagram, full_mtu: bool| { + assert_eq!(d.destination(), preferred); + assert_eq!(d.source(), hs_client); + if full_mtu { + assert_eq!(d.len(), mtu); + } + }; + let assert_from_spa = |d: &Datagram, full_mtu: bool| { + assert_eq!(d.destination(), hs_client); + assert_eq!(d.source(), preferred); + if full_mtu { + assert_eq!(d.len(), mtu); + } + }; + + fixture_init(); + let (log, _contents) = new_neqo_qlog(); + let mut client = Connection::new_client( + test_fixture::DEFAULT_SERVER_NAME, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(EmptyConnectionIdGenerator::default())), + hs_client, + hs_server, + ConnectionParameters::default(), + now(), + ) + .unwrap(); + client.set_qlog(log); + let spa = match preferred { + SocketAddr::V6(v6) => PreferredAddress::new(None, Some(v6)), + SocketAddr::V4(v4) => PreferredAddress::new(Some(v4), None), + }; + let mut server = new_server(ConnectionParameters::default().preferred_address(spa)); + + let dgram = fast_handshake(&mut client, &mut server); + + // The client is about to process HANDSHAKE_DONE. + // It should start probing toward the server's preferred address. + let probe = client.process(dgram.as_ref(), now()).dgram().unwrap(); + assert_toward_spa(&probe, true); + assert_eq!(client.stats().frame_tx.path_challenge, 1); + assert_ne!(client.process_output(now()).callback(), Duration::new(0, 0)); + + // Data continues on the main path for the client. + let data = send_something(&mut client, now()); + assert_orig_path(&data, false); + + // The server responds to the probe. + let resp = server.process(Some(&probe), now()).dgram().unwrap(); + assert_from_spa(&resp, true); + assert_eq!(server.stats().frame_tx.path_challenge, 1); + assert_eq!(server.stats().frame_tx.path_response, 1); + + // Data continues on the main path for the server. + server.process_input(&data, now()); + let data = send_something(&mut server, now()); + assert_orig_path(&data, false); + + // Client gets the probe response back and it migrates. + client.process_input(&resp, now()); + client.process_input(&data, now()); + let data = send_something(&mut client, now()); + assert_toward_spa(&data, true); + assert_eq!(client.stats().frame_tx.stream, 2); + assert_eq!(client.stats().frame_tx.path_response, 1); + + // The server sees the migration and probes the old path. + let probe = server.process(Some(&data), now()).dgram().unwrap(); + assert_orig_path(&probe, true); + assert_eq!(server.stats().frame_tx.path_challenge, 2); + + // But data now goes on the new path. + let data = send_something(&mut server, now()); + assert_from_spa(&data, false); +} + +/// Migration works for a new port number. +#[test] +fn preferred_address_new_port() { + let a = addr(); + preferred_address(a, a, new_port(a)); +} + +/// Migration works for a new address too. +#[test] +fn preferred_address_new_address() { + let mut preferred = addr(); + preferred.set_ip(IpAddr::V6(Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 2))); + preferred_address(addr(), addr(), preferred); +} + +/// Migration works for IPv4 addresses. +#[test] +fn preferred_address_new_port_v4() { + let a = addr_v4(); + preferred_address(a, a, new_port(a)); +} + +/// Migrating to a loopback address is OK if we started there. +#[test] +fn preferred_address_loopback() { + let a = loopback(); + preferred_address(a, a, new_port(a)); +} + +fn expect_no_migration(client: &mut Connection, server: &mut Connection) { + let dgram = fast_handshake(client, server); + + // The client won't probe now, though it could; it remains idle. + let out = client.process(dgram.as_ref(), now()); + assert_ne!(out.callback(), Duration::new(0, 0)); + + // Data continues on the main path for the client. + let data = send_something(client, now()); + assert_v6_path(&data, false); + assert_eq!(client.stats().frame_tx.path_challenge, 0); +} + +fn preferred_address_ignored(spa: PreferredAddress) { + let mut client = default_client(); + let mut server = new_server(ConnectionParameters::default().preferred_address(spa)); + + expect_no_migration(&mut client, &mut server); +} + +/// Using a loopback address in the preferred address is ignored. +#[test] +fn preferred_address_ignore_loopback() { + preferred_address_ignored(PreferredAddress::new_any(None, Some(loopback()))); +} + +/// A preferred address in the wrong address family is ignored. +#[test] +fn preferred_address_ignore_different_family() { + preferred_address_ignored(PreferredAddress::new_any(Some(addr_v4()), None)); +} + +/// Disabling preferred addresses at the client means that it ignores a perfectly +/// good preferred address. +#[test] +fn preferred_address_disabled_client() { + let mut client = new_client(ConnectionParameters::default().disable_preferred_address()); + let mut preferred = addr(); + preferred.set_ip(IpAddr::V6(Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 2))); + let spa = PreferredAddress::new_any(None, Some(preferred)); + let mut server = new_server(ConnectionParameters::default().preferred_address(spa)); + + expect_no_migration(&mut client, &mut server); +} + +#[test] +fn preferred_address_empty_cid() { + fixture_init(); + + let spa = PreferredAddress::new_any(None, Some(new_port(addr()))); + let res = Connection::new_server( + test_fixture::DEFAULT_KEYS, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(EmptyConnectionIdGenerator::default())), + ConnectionParameters::default().preferred_address(spa), + ); + assert_eq!(res.unwrap_err(), Error::ConnectionIdsExhausted); +} + +/// A server cannot include a preferred address if it chooses an empty connection ID. +#[test] +fn preferred_address_server_empty_cid() { + let mut client = default_client(); + let mut server = Connection::new_server( + test_fixture::DEFAULT_KEYS, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(EmptyConnectionIdGenerator::default())), + ConnectionParameters::default(), + ) + .unwrap(); + + server + .set_local_tparam( + tparams::PREFERRED_ADDRESS, + TransportParameter::Bytes(SAMPLE_PREFERRED_ADDRESS.to_vec()), + ) + .unwrap(); + + connect_fail( + &mut client, + &mut server, + Error::TransportParameterError, + Error::PeerError(Error::TransportParameterError.code()), + ); +} + +/// A client shouldn't send a preferred address transport parameter. +#[test] +fn preferred_address_client() { + let mut client = default_client(); + let mut server = default_server(); + + client + .set_local_tparam( + tparams::PREFERRED_ADDRESS, + TransportParameter::Bytes(SAMPLE_PREFERRED_ADDRESS.to_vec()), + ) + .unwrap(); + + connect_fail( + &mut client, + &mut server, + Error::PeerError(Error::TransportParameterError.code()), + Error::TransportParameterError, + ); +} + +/// Test that migration isn't permitted if the connection isn't in the right state. +#[test] +fn migration_invalid_state() { + let mut client = default_client(); + assert!(client + .migrate(Some(addr()), Some(addr()), false, now()) + .is_err()); + + let mut server = default_server(); + assert!(server + .migrate(Some(addr()), Some(addr()), false, now()) + .is_err()); + connect_force_idle(&mut client, &mut server); + + assert!(server + .migrate(Some(addr()), Some(addr()), false, now()) + .is_err()); + + client.close(now(), 0, "closing"); + assert!(client + .migrate(Some(addr()), Some(addr()), false, now()) + .is_err()); + let close = client.process(None, now()).dgram(); + + let dgram = server.process(close.as_ref(), now()).dgram(); + assert!(server + .migrate(Some(addr()), Some(addr()), false, now()) + .is_err()); + + client.process_input(&dgram.unwrap(), now()); + assert!(client + .migrate(Some(addr()), Some(addr()), false, now()) + .is_err()); +} + +#[test] +fn migration_invalid_address() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + let mut cant_migrate = |local, remote| { + assert_eq!( + client.migrate(local, remote, true, now()).unwrap_err(), + Error::InvalidMigration + ); + }; + + // Providing neither address is pointless and therefore an error. + cant_migrate(None, None); + + // Providing a zero port number isn't valid. + let mut zero_port = addr(); + zero_port.set_port(0); + cant_migrate(None, Some(zero_port)); + cant_migrate(Some(zero_port), None); + + // An unspecified remote address is bad. + let mut remote_unspecified = addr(); + remote_unspecified.set_ip(IpAddr::V6(Ipv6Addr::from(0))); + cant_migrate(None, Some(remote_unspecified)); + + // Mixed address families is bad. + cant_migrate(Some(addr()), Some(addr_v4())); + cant_migrate(Some(addr_v4()), Some(addr())); + + // Loopback to non-loopback is bad. + cant_migrate(Some(addr()), Some(loopback())); + cant_migrate(Some(loopback()), Some(addr())); + assert_eq!( + client + .migrate(Some(addr()), Some(loopback()), true, now()) + .unwrap_err(), + Error::InvalidMigration + ); + assert_eq!( + client + .migrate(Some(loopback()), Some(addr()), true, now()) + .unwrap_err(), + Error::InvalidMigration + ); +} + +/// This inserts a frame into packets that provides a single new +/// connection ID and retires all others. +struct RetireAll { + cid_gen: Rc<RefCell<dyn ConnectionIdGenerator>>, +} + +impl crate::connection::test_internal::FrameWriter for RetireAll { + fn write_frames(&mut self, builder: &mut PacketBuilder) { + // Use a sequence number that is large enough that all existing values + // will be lower (so they get retired). As the code doesn't care about + // gaps in sequence numbers, this is safe, even though the gap might + // hint that there are more outstanding connection IDs that are allowed. + const SEQNO: u64 = 100; + let cid = self.cid_gen.borrow_mut().generate_cid().unwrap(); + builder + .encode_varint(FRAME_TYPE_NEW_CONNECTION_ID) + .encode_varint(SEQNO) + .encode_varint(SEQNO) // Retire Prior To + .encode_vec(1, &cid) + .encode(&[0x7f; 16]); + } +} + +/// Test that forcing retirement of connection IDs forces retirement of all active +/// connection IDs and the use of of newer one. +#[test] +fn retire_all() { + let mut client = default_client(); + let cid_gen: Rc<RefCell<dyn ConnectionIdGenerator>> = + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())); + let mut server = Connection::new_server( + test_fixture::DEFAULT_KEYS, + test_fixture::DEFAULT_ALPN, + Rc::clone(&cid_gen), + ConnectionParameters::default(), + ) + .unwrap(); + connect_force_idle(&mut client, &mut server); + + let original_cid = ConnectionId::from(get_cid(&send_something(&mut client, now()))); + + server.test_frame_writer = Some(Box::new(RetireAll { cid_gen })); + let ncid = send_something(&mut server, now()); + server.test_frame_writer = None; + + let new_cid_before = client.stats().frame_rx.new_connection_id; + let retire_cid_before = client.stats().frame_tx.retire_connection_id; + client.process_input(&ncid, now()); + let retire = send_something(&mut client, now()); + assert_eq!( + client.stats().frame_rx.new_connection_id, + new_cid_before + 1 + ); + assert_eq!( + client.stats().frame_tx.retire_connection_id, + retire_cid_before + LOCAL_ACTIVE_CID_LIMIT + ); + + assert_ne!(get_cid(&retire), original_cid); +} + +/// During a graceful migration, if the probed path can't get a new connection ID due +/// to being forced to retire the one it is using, the migration will fail. +#[test] +fn retire_prior_to_migration_failure() { + let mut client = default_client(); + let cid_gen: Rc<RefCell<dyn ConnectionIdGenerator>> = + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())); + let mut server = Connection::new_server( + test_fixture::DEFAULT_KEYS, + test_fixture::DEFAULT_ALPN, + Rc::clone(&cid_gen), + ConnectionParameters::default(), + ) + .unwrap(); + connect_force_idle(&mut client, &mut server); + + let original_cid = ConnectionId::from(get_cid(&send_something(&mut client, now()))); + + client + .migrate(Some(addr_v4()), Some(addr_v4()), false, now()) + .unwrap(); + + // The client now probes the new path. + let probe = client.process_output(now()).dgram().unwrap(); + assert_v4_path(&probe, true); + assert_eq!(client.stats().frame_tx.path_challenge, 1); + let probe_cid = ConnectionId::from(get_cid(&probe)); + assert_ne!(original_cid, probe_cid); + + // Have the server receive the probe, but separately have it decide to + // retire all of the available connection IDs. + server.test_frame_writer = Some(Box::new(RetireAll { cid_gen })); + let retire_all = send_something(&mut server, now()); + server.test_frame_writer = None; + + let resp = server.process(Some(&probe), now()).dgram().unwrap(); + assert_v4_path(&resp, true); + assert_eq!(server.stats().frame_tx.path_response, 1); + assert_eq!(server.stats().frame_tx.path_challenge, 1); + + // Have the client receive the NEW_CONNECTION_ID with Retire Prior To. + client.process_input(&retire_all, now()); + // This packet contains the probe response, which should be fine, but it + // also includes PATH_CHALLENGE for the new path, and the client can't + // respond without a connection ID. We treat this as a connection error. + client.process_input(&resp, now()); + assert!(matches!( + client.state(), + State::Closing { + error: ConnectionError::Transport(Error::InvalidMigration), + .. + } + )); +} + +/// The timing of when frames arrive can mean that the migration path can +/// get the last available connection ID. +#[test] +fn retire_prior_to_migration_success() { + let mut client = default_client(); + let cid_gen: Rc<RefCell<dyn ConnectionIdGenerator>> = + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())); + let mut server = Connection::new_server( + test_fixture::DEFAULT_KEYS, + test_fixture::DEFAULT_ALPN, + Rc::clone(&cid_gen), + ConnectionParameters::default(), + ) + .unwrap(); + connect_force_idle(&mut client, &mut server); + + let original_cid = ConnectionId::from(get_cid(&send_something(&mut client, now()))); + + client + .migrate(Some(addr_v4()), Some(addr_v4()), false, now()) + .unwrap(); + + // The client now probes the new path. + let probe = client.process_output(now()).dgram().unwrap(); + assert_v4_path(&probe, true); + assert_eq!(client.stats().frame_tx.path_challenge, 1); + let probe_cid = ConnectionId::from(get_cid(&probe)); + assert_ne!(original_cid, probe_cid); + + // Have the server receive the probe, but separately have it decide to + // retire all of the available connection IDs. + server.test_frame_writer = Some(Box::new(RetireAll { cid_gen })); + let retire_all = send_something(&mut server, now()); + server.test_frame_writer = None; + + let resp = server.process(Some(&probe), now()).dgram().unwrap(); + assert_v4_path(&resp, true); + assert_eq!(server.stats().frame_tx.path_response, 1); + assert_eq!(server.stats().frame_tx.path_challenge, 1); + + // Have the client receive the NEW_CONNECTION_ID with Retire Prior To second. + // As this occurs in a very specific order, migration succeeds. + client.process_input(&resp, now()); + client.process_input(&retire_all, now()); + + // Migration succeeds and the new path gets the last connection ID. + let dgram = send_something(&mut client, now()); + assert_v4_path(&dgram, false); + assert_ne!(get_cid(&dgram), original_cid); + assert_ne!(get_cid(&dgram), probe_cid); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/mod.rs b/third_party/rust/neqo-transport/src/connection/tests/mod.rs new file mode 100644 index 0000000000..8a999f4048 --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/mod.rs @@ -0,0 +1,614 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +#![deny(clippy::pedantic)] + +use std::{ + cell::RefCell, + cmp::min, + convert::TryFrom, + mem, + rc::Rc, + time::{Duration, Instant}, +}; + +use enum_map::enum_map; +use neqo_common::{event::Provider, qdebug, qtrace, Datagram, Decoder, Role}; +use neqo_crypto::{random, AllowZeroRtt, AuthenticationStatus, ResumptionToken}; +use test_fixture::{self, addr, fixture_init, new_neqo_qlog, now}; + +use super::{Connection, ConnectionError, ConnectionId, Output, State}; +use crate::{ + addr_valid::{AddressValidation, ValidateAddress}, + cc::{CWND_INITIAL_PKTS, CWND_MIN}, + cid::ConnectionIdRef, + events::ConnectionEvent, + frame::FRAME_TYPE_PING, + packet::PacketBuilder, + path::PATH_MTU_V6, + recovery::ACK_ONLY_SIZE_LIMIT, + stats::{FrameStats, Stats, MAX_PTO_COUNTS}, + ConnectionIdDecoder, ConnectionIdGenerator, ConnectionParameters, Error, StreamId, StreamType, + Version, +}; + +// All the tests. +mod ackrate; +mod cc; +mod close; +mod datagram; +mod fuzzing; +mod handshake; +mod idle; +mod keys; +mod migration; +mod priority; +mod recovery; +mod resumption; +mod stream; +mod vn; +mod zerortt; + +const DEFAULT_RTT: Duration = Duration::from_millis(100); +const AT_LEAST_PTO: Duration = Duration::from_secs(1); +const DEFAULT_STREAM_DATA: &[u8] = b"message"; +/// The number of 1-RTT packets sent in `force_idle` by a client. +const CLIENT_HANDSHAKE_1RTT_PACKETS: usize = 1; + +/// WARNING! In this module, this version of the generator needs to be used. +/// This copies the implementation from +/// `test_fixture::CountingConnectionIdGenerator`, but it uses the different +/// types that are exposed to this module. See also `default_client`. +/// +/// This version doesn't randomize the length; as the congestion control tests +/// count the amount of data sent precisely. +#[derive(Debug, Default)] +pub struct CountingConnectionIdGenerator { + counter: u32, +} + +impl ConnectionIdDecoder for CountingConnectionIdGenerator { + fn decode_cid<'a>(&self, dec: &mut Decoder<'a>) -> Option<ConnectionIdRef<'a>> { + let len = usize::from(dec.peek_byte().unwrap()); + dec.decode(len).map(ConnectionIdRef::from) + } +} + +impl ConnectionIdGenerator for CountingConnectionIdGenerator { + fn generate_cid(&mut self) -> Option<ConnectionId> { + let mut r = random(20); + r[0] = 8; + r[1] = u8::try_from(self.counter >> 24).unwrap(); + r[2] = u8::try_from((self.counter >> 16) & 0xff).unwrap(); + r[3] = u8::try_from((self.counter >> 8) & 0xff).unwrap(); + r[4] = u8::try_from(self.counter & 0xff).unwrap(); + self.counter += 1; + Some(ConnectionId::from(&r[..8])) + } + + fn as_decoder(&self) -> &dyn ConnectionIdDecoder { + self + } +} + +// This is fabulous: because test_fixture uses the public API for Connection, +// it gets a different type to the ones that are referenced via super::super::*. +// Thus, this code can't use default_client() and default_server() from +// test_fixture because they produce different - and incompatible - types. +// +// These are a direct copy of those functions. +pub fn new_client(params: ConnectionParameters) -> Connection { + fixture_init(); + let (log, _contents) = new_neqo_qlog(); + let mut client = Connection::new_client( + test_fixture::DEFAULT_SERVER_NAME, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())), + addr(), + addr(), + params, + now(), + ) + .expect("create a default client"); + client.set_qlog(log); + client +} + +pub fn default_client() -> Connection { + new_client(ConnectionParameters::default()) +} + +pub fn new_server(params: ConnectionParameters) -> Connection { + fixture_init(); + let (log, _contents) = new_neqo_qlog(); + let mut c = Connection::new_server( + test_fixture::DEFAULT_KEYS, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())), + params, + ) + .expect("create a default server"); + c.set_qlog(log); + c.server_enable_0rtt(&test_fixture::anti_replay(), AllowZeroRtt {}) + .expect("enable 0-RTT"); + c +} +pub fn default_server() -> Connection { + new_server(ConnectionParameters::default()) +} +pub fn resumed_server(client: &Connection) -> Connection { + new_server(ConnectionParameters::default().versions(client.version(), Version::all())) +} + +/// If state is `AuthenticationNeeded` call `authenticated()`. This function will +/// consume all outstanding events on the connection. +pub fn maybe_authenticate(conn: &mut Connection) -> bool { + let authentication_needed = |e| matches!(e, ConnectionEvent::AuthenticationNeeded); + if conn.events().any(authentication_needed) { + conn.authenticated(AuthenticationStatus::Ok, now()); + return true; + } + false +} + +/// Compute the RTT variance after `n` ACKs or other RTT updates. +pub fn rttvar_after_n_updates(n: usize, rtt: Duration) -> Duration { + assert!(n > 0); + let mut rttvar = rtt / 2; + for _ in 1..n { + rttvar = rttvar * 3 / 4; + } + rttvar +} + +/// This inserts a PING frame into packets. +struct PingWriter {} + +impl crate::connection::test_internal::FrameWriter for PingWriter { + fn write_frames(&mut self, builder: &mut PacketBuilder) { + builder.encode_varint(FRAME_TYPE_PING); + } +} + +/// Drive the handshake between the client and server. +fn handshake( + client: &mut Connection, + server: &mut Connection, + now: Instant, + rtt: Duration, +) -> Instant { + let mut a = client; + let mut b = server; + let mut now = now; + + let mut input = None; + let is_done = |c: &mut Connection| { + matches!( + c.state(), + State::Confirmed | State::Closing { .. } | State::Closed(..) + ) + }; + + let mut did_ping = enum_map! {_ => false}; + while !is_done(a) { + _ = maybe_authenticate(a); + let had_input = input.is_some(); + // Insert a PING frame into the first application data packet an endpoint sends, + // in order to force the peer to ACK it. For the server, this is depending on the + // client's connection state, which is accessible during the tests. + // + // We're doing this to prevent packet loss from delaying ACKs, which would cause + // cwnd to shrink, and also to prevent the delayed ACK timer from being armed after + // the handshake, which is not something the tests are written to account for. + let should_ping = !did_ping[a.role()] + && (a.role() == Role::Client && *a.state() == State::Connected + || (a.role() == Role::Server && *b.state() == State::Connected)); + if should_ping { + a.test_frame_writer = Some(Box::new(PingWriter {})); + } + let output = a.process(input.as_ref(), now).dgram(); + if should_ping { + a.test_frame_writer = None; + did_ping[a.role()] = true; + } + assert!(had_input || output.is_some()); + input = output; + qtrace!("handshake: t += {:?}", rtt / 2); + now += rtt / 2; + mem::swap(&mut a, &mut b); + } + if let Some(d) = input { + a.process_input(&d, now); + } + now +} + +fn connect_fail( + client: &mut Connection, + server: &mut Connection, + client_error: Error, + server_error: Error, +) { + handshake(client, server, now(), Duration::new(0, 0)); + assert_error(client, &ConnectionError::Transport(client_error)); + assert_error(server, &ConnectionError::Transport(server_error)); +} + +fn connect_with_rtt( + client: &mut Connection, + server: &mut Connection, + now: Instant, + rtt: Duration, +) -> Instant { + fn check_rtt(stats: &Stats, rtt: Duration) { + assert_eq!(stats.rtt, rtt); + // Validate that rttvar has been computed correctly based on the number of RTT updates. + let n = stats.frame_rx.ack + usize::from(stats.rtt_init_guess); + assert_eq!(stats.rttvar, rttvar_after_n_updates(n, rtt)); + } + let now = handshake(client, server, now, rtt); + assert_eq!(*client.state(), State::Confirmed); + assert_eq!(*server.state(), State::Confirmed); + + check_rtt(&client.stats(), rtt); + check_rtt(&server.stats(), rtt); + now +} + +fn connect(client: &mut Connection, server: &mut Connection) { + connect_with_rtt(client, server, now(), Duration::new(0, 0)); +} + +fn assert_error(c: &Connection, expected: &ConnectionError) { + match c.state() { + State::Closing { error, .. } | State::Draining { error, .. } | State::Closed(error) => { + assert_eq!(*error, *expected, "{c} error mismatch"); + } + _ => panic!("bad state {:?}", c.state()), + } +} + +fn exchange_ticket( + client: &mut Connection, + server: &mut Connection, + now: Instant, +) -> ResumptionToken { + let validation = AddressValidation::new(now, ValidateAddress::NoToken).unwrap(); + let validation = Rc::new(RefCell::new(validation)); + server.set_validation(Rc::clone(&validation)); + server.send_ticket(now, &[]).expect("can send ticket"); + let ticket = server.process_output(now).dgram(); + assert!(ticket.is_some()); + client.process_input(&ticket.unwrap(), now); + assert_eq!(*client.state(), State::Confirmed); + get_tokens(client).pop().expect("should have token") +} + +/// The `handshake` method inserts PING frames into the first application data packets, +/// which forces each peer to ACK them. As a side effect, that causes both sides of the +/// connection to be idle aftwerwards. This method simply verifies that this is the case. +fn assert_idle(client: &mut Connection, server: &mut Connection, rtt: Duration, now: Instant) { + let idle_timeout = min( + client.conn_params.get_idle_timeout(), + server.conn_params.get_idle_timeout(), + ); + // Client started its idle period half an RTT before now. + assert_eq!( + client.process_output(now), + Output::Callback(idle_timeout - rtt / 2) + ); + assert_eq!(server.process_output(now), Output::Callback(idle_timeout)); +} + +/// Connect with an RTT and then force both peers to be idle. +fn connect_rtt_idle(client: &mut Connection, server: &mut Connection, rtt: Duration) -> Instant { + let now = connect_with_rtt(client, server, now(), rtt); + assert_idle(client, server, rtt, now); + // Drain events from both as well. + _ = client.events().count(); + _ = server.events().count(); + qtrace!("----- connected and idle with RTT {:?}", rtt); + now +} + +fn connect_force_idle(client: &mut Connection, server: &mut Connection) { + connect_rtt_idle(client, server, Duration::new(0, 0)); +} + +fn fill_stream(c: &mut Connection, stream: StreamId) { + const BLOCK_SIZE: usize = 4_096; + loop { + let bytes_sent = c.stream_send(stream, &[0x42; BLOCK_SIZE]).unwrap(); + qtrace!("fill_cwnd wrote {} bytes", bytes_sent); + if bytes_sent < BLOCK_SIZE { + break; + } + } +} + +/// This fills the congestion window from a single source. +/// As the pacer will interfere with this, this moves time forward +/// as `Output::Callback` is received. Because it is hard to tell +/// from the return value whether a timeout is an ACK delay, PTO, or +/// pacing, this looks at the congestion window to tell when to stop. +/// Returns a list of datagrams and the new time. +fn fill_cwnd(c: &mut Connection, stream: StreamId, mut now: Instant) -> (Vec<Datagram>, Instant) { + // Train wreck function to get the remaining congestion window on the primary path. + fn cwnd(c: &Connection) -> usize { + c.paths.primary().borrow().sender().cwnd_avail() + } + + qtrace!("fill_cwnd starting cwnd: {}", cwnd(c)); + fill_stream(c, stream); + + let mut total_dgrams = Vec::new(); + loop { + let pkt = c.process_output(now); + qtrace!("fill_cwnd cwnd remaining={}, output: {:?}", cwnd(c), pkt); + match pkt { + Output::Datagram(dgram) => { + total_dgrams.push(dgram); + } + Output::Callback(t) => { + if cwnd(c) < ACK_ONLY_SIZE_LIMIT { + break; + } + now += t; + } + Output::None => panic!(), + } + } + + qtrace!( + "fill_cwnd sent {} bytes", + total_dgrams.iter().map(|d| d.len()).sum::<usize>() + ); + (total_dgrams, now) +} + +/// This function is like the combination of `fill_cwnd` and `ack_bytes`. +/// However, it acknowledges everything inline and preserves an RTT of `DEFAULT_RTT`. +fn increase_cwnd( + sender: &mut Connection, + receiver: &mut Connection, + stream: StreamId, + mut now: Instant, +) -> Instant { + fill_stream(sender, stream); + loop { + let pkt = sender.process_output(now); + match pkt { + Output::Datagram(dgram) => { + receiver.process_input(&dgram, now + DEFAULT_RTT / 2); + } + Output::Callback(t) => { + if t < DEFAULT_RTT { + now += t; + } else { + break; // We're on PTO now. + } + } + Output::None => panic!(), + } + } + + // Now acknowledge all those packets at once. + now += DEFAULT_RTT / 2; + let ack = receiver.process_output(now).dgram(); + now += DEFAULT_RTT / 2; + sender.process_input(&ack.unwrap(), now); + now +} + +/// Receive multiple packets and generate an ack-only packet. +/// +/// # Panics +/// +/// The caller is responsible for ensuring that `dest` has received +/// enough data that it wants to generate an ACK. This panics if +/// no ACK frame is generated. +fn ack_bytes<D>(dest: &mut Connection, stream: StreamId, in_dgrams: D, now: Instant) -> Datagram +where + D: IntoIterator<Item = Datagram>, + D::IntoIter: ExactSizeIterator, +{ + let mut srv_buf = [0; 4_096]; + + let in_dgrams = in_dgrams.into_iter(); + qdebug!([dest], "ack_bytes {} datagrams", in_dgrams.len()); + for dgram in in_dgrams { + dest.process_input(&dgram, now); + } + + loop { + let (bytes_read, _fin) = dest.stream_recv(stream, &mut srv_buf).unwrap(); + qtrace!([dest], "ack_bytes read {} bytes", bytes_read); + if bytes_read == 0 { + break; + } + } + + dest.process_output(now).dgram().unwrap() +} + +// Get the current congestion window for the connection. +fn cwnd(c: &Connection) -> usize { + c.paths.primary().borrow().sender().cwnd() +} +fn cwnd_avail(c: &Connection) -> usize { + c.paths.primary().borrow().sender().cwnd_avail() +} + +fn induce_persistent_congestion( + client: &mut Connection, + server: &mut Connection, + stream: StreamId, + mut now: Instant, +) -> Instant { + // Note: wait some arbitrary time that should be longer than pto + // timer. This is rather brittle. + qtrace!([client], "induce_persistent_congestion"); + now += AT_LEAST_PTO; + + let mut pto_counts = [0; MAX_PTO_COUNTS]; + assert_eq!(client.stats.borrow().pto_counts, pto_counts); + + qtrace!([client], "first PTO"); + let (c_tx_dgrams, next_now) = fill_cwnd(client, stream, now); + now = next_now; + assert_eq!(c_tx_dgrams.len(), 2); // Two PTO packets + + pto_counts[0] = 1; + assert_eq!(client.stats.borrow().pto_counts, pto_counts); + + qtrace!([client], "second PTO"); + now += AT_LEAST_PTO * 2; + let (c_tx_dgrams, next_now) = fill_cwnd(client, stream, now); + now = next_now; + assert_eq!(c_tx_dgrams.len(), 2); // Two PTO packets + + pto_counts[0] = 0; + pto_counts[1] = 1; + assert_eq!(client.stats.borrow().pto_counts, pto_counts); + + qtrace!([client], "third PTO"); + now += AT_LEAST_PTO * 4; + let (c_tx_dgrams, next_now) = fill_cwnd(client, stream, now); + now = next_now; + assert_eq!(c_tx_dgrams.len(), 2); // Two PTO packets + + pto_counts[1] = 0; + pto_counts[2] = 1; + assert_eq!(client.stats.borrow().pto_counts, pto_counts); + + // An ACK for the third PTO causes persistent congestion. + let s_ack = ack_bytes(server, stream, c_tx_dgrams, now); + client.process_input(&s_ack, now); + assert_eq!(cwnd(client), CWND_MIN); + now +} + +/// This magic number is the size of the client's CWND after the handshake completes. +/// This is the same as the initial congestion window, because during the handshake +/// the cc is app limited and cwnd is not increased. +/// +/// As we change how we build packets, or even as NSS changes, +/// this number might be different. The tests that depend on this +/// value could fail as a result of variations, so it's OK to just +/// change this value, but it is good to first understand where the +/// change came from. +const POST_HANDSHAKE_CWND: usize = PATH_MTU_V6 * CWND_INITIAL_PKTS; + +/// Determine the number of packets required to fill the CWND. +const fn cwnd_packets(data: usize) -> usize { + // Add one if the last chunk is >= ACK_ONLY_SIZE_LIMIT. + (data + PATH_MTU_V6 - ACK_ONLY_SIZE_LIMIT) / PATH_MTU_V6 +} + +/// Determine the size of the last packet. +/// The minimal size of a packet is `ACK_ONLY_SIZE_LIMIT`. +fn last_packet(cwnd: usize) -> usize { + if (cwnd % PATH_MTU_V6) > ACK_ONLY_SIZE_LIMIT { + cwnd % PATH_MTU_V6 + } else { + PATH_MTU_V6 + } +} + +/// Assert that the set of packets fill the CWND. +fn assert_full_cwnd(packets: &[Datagram], cwnd: usize) { + assert_eq!(packets.len(), cwnd_packets(cwnd)); + let (last, rest) = packets.split_last().unwrap(); + assert!(rest.iter().all(|d| d.len() == PATH_MTU_V6)); + assert_eq!(last.len(), last_packet(cwnd)); +} + +/// Send something on a stream from `sender` to `receiver`, maybe allowing for pacing. +/// Return the resulting datagram and the new time. +#[must_use] +fn send_something_paced( + sender: &mut Connection, + mut now: Instant, + allow_pacing: bool, +) -> (Datagram, Instant) { + let stream_id = sender.stream_create(StreamType::UniDi).unwrap(); + assert!(sender.stream_send(stream_id, DEFAULT_STREAM_DATA).is_ok()); + assert!(sender.stream_close_send(stream_id).is_ok()); + qdebug!([sender], "send_something on {}", stream_id); + let dgram = match sender.process_output(now) { + Output::Callback(t) => { + assert!(allow_pacing, "send_something: unexpected delay"); + now += t; + sender + .process_output(now) + .dgram() + .expect("send_something: should have something to send") + } + Output::Datagram(d) => d, + Output::None => panic!("send_something: got Output::None"), + }; + (dgram, now) +} + +/// Send something on a stream from `sender` to `receiver`. +/// Return the resulting datagram. +fn send_something(sender: &mut Connection, now: Instant) -> Datagram { + send_something_paced(sender, now, false).0 +} + +/// Send something on a stream from `sender` to `receiver`. +/// Return any ACK that might result. +fn send_and_receive( + sender: &mut Connection, + receiver: &mut Connection, + now: Instant, +) -> Option<Datagram> { + let dgram = send_something(sender, now); + receiver.process(Some(&dgram), now).dgram() +} + +fn get_tokens(client: &mut Connection) -> Vec<ResumptionToken> { + client + .events() + .filter_map(|e| { + if let ConnectionEvent::ResumptionToken(token) = e { + Some(token) + } else { + None + } + }) + .collect() +} + +fn assert_default_stats(stats: &Stats) { + assert_eq!(stats.packets_rx, 0); + assert_eq!(stats.packets_tx, 0); + let dflt_frames = FrameStats::default(); + assert_eq!(stats.frame_rx, dflt_frames); + assert_eq!(stats.frame_tx, dflt_frames); +} + +#[test] +fn create_client() { + let client = default_client(); + assert_eq!(client.role(), Role::Client); + assert!(matches!(client.state(), State::Init)); + let stats = client.stats(); + assert_default_stats(&stats); + assert_eq!(stats.rtt, crate::rtt::INITIAL_RTT); + assert_eq!(stats.rttvar, crate::rtt::INITIAL_RTT / 2); +} + +#[test] +fn create_server() { + let server = default_server(); + assert_eq!(server.role(), Role::Server); + assert!(matches!(server.state(), State::Init)); + let stats = server.stats(); + assert_default_stats(&stats); + // Server won't have a default path, so no RTT. + assert_eq!(stats.rtt, Duration::from_secs(0)); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/priority.rs b/third_party/rust/neqo-transport/src/connection/tests/priority.rs new file mode 100644 index 0000000000..1f86aa22e5 --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/priority.rs @@ -0,0 +1,404 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{cell::RefCell, mem, rc::Rc}; + +use neqo_common::event::Provider; +use test_fixture::{self, now}; + +use super::{ + super::{Connection, Error, Output}, + connect, default_client, default_server, fill_cwnd, maybe_authenticate, +}; +use crate::{ + addr_valid::{AddressValidation, ValidateAddress}, + send_stream::{RetransmissionPriority, TransmissionPriority}, + ConnectionEvent, StreamId, StreamType, +}; + +const BLOCK_SIZE: usize = 4_096; + +fn fill_stream(c: &mut Connection, id: StreamId) { + loop { + if c.stream_send(id, &[0x42; BLOCK_SIZE]).unwrap() < BLOCK_SIZE { + return; + } + } +} + +/// A receive stream cannot be prioritized (yet). +#[test] +fn receive_stream() { + const MESSAGE: &[u8] = b"hello"; + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let id = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(MESSAGE.len(), client.stream_send(id, MESSAGE).unwrap()); + let dgram = client.process_output(now()).dgram(); + + server.process_input(&dgram.unwrap(), now()); + assert_eq!( + server + .stream_priority( + id, + TransmissionPriority::default(), + RetransmissionPriority::default() + ) + .unwrap_err(), + Error::InvalidStreamId, + "Priority doesn't apply to inbound unidirectional streams" + ); + + // But the stream does exist and can be read. + let mut buf = [0; 10]; + let (len, end) = server.stream_recv(id, &mut buf).unwrap(); + assert_eq!(MESSAGE, &buf[..len]); + assert!(!end); +} + +/// Higher priority streams get sent ahead of lower ones, even when +/// the higher priority stream is written to later. +#[test] +fn relative() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + // id_normal is created first, but it is lower priority. + let id_normal = client.stream_create(StreamType::UniDi).unwrap(); + fill_stream(&mut client, id_normal); + let high = client.stream_create(StreamType::UniDi).unwrap(); + fill_stream(&mut client, high); + client + .stream_priority( + high, + TransmissionPriority::High, + RetransmissionPriority::default(), + ) + .unwrap(); + + let dgram = client.process_output(now()).dgram(); + server.process_input(&dgram.unwrap(), now()); + + // The "id_normal" stream will get a `NewStream` event, but no data. + for e in server.events() { + if let ConnectionEvent::RecvStreamReadable { stream_id } = e { + assert_ne!(stream_id, id_normal); + } + } +} + +/// Check that changing priority has effect on the next packet that is sent. +#[test] +fn reprioritize() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + // id_normal is created first, but it is lower priority. + let id_normal = client.stream_create(StreamType::UniDi).unwrap(); + fill_stream(&mut client, id_normal); + let id_high = client.stream_create(StreamType::UniDi).unwrap(); + fill_stream(&mut client, id_high); + client + .stream_priority( + id_high, + TransmissionPriority::High, + RetransmissionPriority::default(), + ) + .unwrap(); + + let dgram = client.process_output(now()).dgram(); + server.process_input(&dgram.unwrap(), now()); + + // The "id_normal" stream will get a `NewStream` event, but no data. + for e in server.events() { + if let ConnectionEvent::RecvStreamReadable { stream_id } = e { + assert_ne!(stream_id, id_normal); + } + } + + // When the high priority stream drops in priority, the streams are equal + // priority and so their stream ID determines what is sent. + client + .stream_priority( + id_high, + TransmissionPriority::Normal, + RetransmissionPriority::default(), + ) + .unwrap(); + let dgram = client.process_output(now()).dgram(); + server.process_input(&dgram.unwrap(), now()); + + for e in server.events() { + if let ConnectionEvent::RecvStreamReadable { stream_id } = e { + assert_ne!(stream_id, id_high); + } + } +} + +/// Retransmission can be prioritized differently (usually higher). +#[test] +fn repairing_loss() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let mut now = now(); + + // Send a few packets at low priority, lose one. + let id_low = client.stream_create(StreamType::UniDi).unwrap(); + fill_stream(&mut client, id_low); + client + .stream_priority( + id_low, + TransmissionPriority::Low, + RetransmissionPriority::Higher, + ) + .unwrap(); + + let _lost = client.process_output(now).dgram(); + for _ in 0..5 { + match client.process_output(now) { + Output::Datagram(d) => server.process_input(&d, now), + Output::Callback(delay) => now += delay, + Output::None => unreachable!(), + } + } + + // Generate an ACK. The first packet is now considered lost. + let ack = server.process_output(now).dgram(); + _ = server.events().count(); // Drain events. + + let id_normal = client.stream_create(StreamType::UniDi).unwrap(); + fill_stream(&mut client, id_normal); + + let dgram = client.process(ack.as_ref(), now).dgram(); + assert_eq!(client.stats().lost, 1); // Client should have noticed the loss. + server.process_input(&dgram.unwrap(), now); + + // Only the low priority stream has data as the retransmission of the data from + // the lost packet is now more important than new data from the high priority stream. + for e in server.events() { + println!("Event: {e:?}"); + if let ConnectionEvent::RecvStreamReadable { stream_id } = e { + assert_eq!(stream_id, id_low); + } + } + + // However, only the retransmission is prioritized. + // Though this might contain some retransmitted data, as other frames might push + // the retransmitted data into a second packet, it will also contain data from the + // normal priority stream. + let dgram = client.process_output(now).dgram(); + server.process_input(&dgram.unwrap(), now); + assert!(server.events().any( + |e| matches!(e, ConnectionEvent::RecvStreamReadable { stream_id } if stream_id == id_normal), + )); +} + +#[test] +fn critical() { + let mut client = default_client(); + let mut server = default_server(); + let now = now(); + + // Rather than connect, send stream data in 0.5-RTT. + // That allows this to test that critical streams pre-empt most frame types. + let dgram = client.process_output(now).dgram(); + let dgram = server.process(dgram.as_ref(), now).dgram(); + client.process_input(&dgram.unwrap(), now); + maybe_authenticate(&mut client); + + let id = server.stream_create(StreamType::UniDi).unwrap(); + server + .stream_priority( + id, + TransmissionPriority::Critical, + RetransmissionPriority::default(), + ) + .unwrap(); + + // Can't use fill_cwnd here because the server is blocked on the amplification + // limit, so it can't fill the congestion window. + while server.stream_create(StreamType::UniDi).is_ok() {} + + fill_stream(&mut server, id); + let stats_before = server.stats().frame_tx; + let dgram = server.process_output(now).dgram(); + let stats_after = server.stats().frame_tx; + assert_eq!(stats_after.crypto, stats_before.crypto); + assert_eq!(stats_after.streams_blocked, 0); + assert_eq!(stats_after.new_connection_id, 0); + assert_eq!(stats_after.new_token, 0); + assert_eq!(stats_after.handshake_done, 0); + + // Complete the handshake. + let dgram = client.process(dgram.as_ref(), now).dgram(); + server.process_input(&dgram.unwrap(), now); + + // Critical beats everything but HANDSHAKE_DONE. + let stats_before = server.stats().frame_tx; + mem::drop(fill_cwnd(&mut server, id, now)); + let stats_after = server.stats().frame_tx; + assert_eq!(stats_after.crypto, stats_before.crypto); + assert_eq!(stats_after.streams_blocked, 0); + assert_eq!(stats_after.new_connection_id, 0); + assert_eq!(stats_after.new_token, 0); + assert_eq!(stats_after.handshake_done, 1); +} + +#[test] +fn important() { + let mut client = default_client(); + let mut server = default_server(); + let now = now(); + + // Rather than connect, send stream data in 0.5-RTT. + // That allows this to test that important streams pre-empt most frame types. + let dgram = client.process_output(now).dgram(); + let dgram = server.process(dgram.as_ref(), now).dgram(); + client.process_input(&dgram.unwrap(), now); + maybe_authenticate(&mut client); + + let id = server.stream_create(StreamType::UniDi).unwrap(); + server + .stream_priority( + id, + TransmissionPriority::Important, + RetransmissionPriority::default(), + ) + .unwrap(); + fill_stream(&mut server, id); + + // Important beats everything but flow control. + // Make enough streams to get a STREAMS_BLOCKED frame out. + while server.stream_create(StreamType::UniDi).is_ok() {} + + let stats_before = server.stats().frame_tx; + let dgram = server.process_output(now).dgram(); + let stats_after = server.stats().frame_tx; + assert_eq!(stats_after.crypto, stats_before.crypto); + assert_eq!(stats_after.streams_blocked, 1); + assert_eq!(stats_after.new_connection_id, 0); + assert_eq!(stats_after.new_token, 0); + assert_eq!(stats_after.handshake_done, 0); + assert_eq!(stats_after.stream, stats_before.stream + 1); + + // Complete the handshake. + let dgram = client.process(dgram.as_ref(), now).dgram(); + server.process_input(&dgram.unwrap(), now); + + // Important beats everything but flow control. + let stats_before = server.stats().frame_tx; + mem::drop(fill_cwnd(&mut server, id, now)); + let stats_after = server.stats().frame_tx; + assert_eq!(stats_after.crypto, stats_before.crypto); + assert_eq!(stats_after.streams_blocked, 1); + assert_eq!(stats_after.new_connection_id, 0); + assert_eq!(stats_after.new_token, 0); + assert_eq!(stats_after.handshake_done, 1); + assert!(stats_after.stream > stats_before.stream); +} + +#[test] +fn high_normal() { + let mut client = default_client(); + let mut server = default_server(); + let now = now(); + + // Rather than connect, send stream data in 0.5-RTT. + // That allows this to test that important streams pre-empt most frame types. + let dgram = client.process_output(now).dgram(); + let dgram = server.process(dgram.as_ref(), now).dgram(); + client.process_input(&dgram.unwrap(), now); + maybe_authenticate(&mut client); + + let id = server.stream_create(StreamType::UniDi).unwrap(); + server + .stream_priority( + id, + TransmissionPriority::High, + RetransmissionPriority::default(), + ) + .unwrap(); + fill_stream(&mut server, id); + + // Important beats everything but flow control. + // Make enough streams to get a STREAMS_BLOCKED frame out. + while server.stream_create(StreamType::UniDi).is_ok() {} + + let stats_before = server.stats().frame_tx; + let dgram = server.process_output(now).dgram(); + let stats_after = server.stats().frame_tx; + assert_eq!(stats_after.crypto, stats_before.crypto); + assert_eq!(stats_after.streams_blocked, 1); + assert_eq!(stats_after.new_connection_id, 0); + assert_eq!(stats_after.new_token, 0); + assert_eq!(stats_after.handshake_done, 0); + assert_eq!(stats_after.stream, stats_before.stream + 1); + + // Complete the handshake. + let dgram = client.process(dgram.as_ref(), now).dgram(); + server.process_input(&dgram.unwrap(), now); + + // High or Normal doesn't beat NEW_CONNECTION_ID, + // but they beat CRYPTO/NEW_TOKEN. + let stats_before = server.stats().frame_tx; + server.send_ticket(now, &[]).unwrap(); + mem::drop(fill_cwnd(&mut server, id, now)); + let stats_after = server.stats().frame_tx; + assert_eq!(stats_after.crypto, stats_before.crypto); + assert_eq!(stats_after.streams_blocked, 1); + assert_ne!(stats_after.new_connection_id, 0); // Note: > 0 + assert_eq!(stats_after.new_token, 0); + assert_eq!(stats_after.handshake_done, 1); + assert!(stats_after.stream > stats_before.stream); +} + +#[test] +fn low() { + let mut client = default_client(); + let mut server = default_server(); + let now = now(); + // Use address validation; note that we need to hold a strong reference + // as the server will only hold a weak reference. + let validation = Rc::new(RefCell::new( + AddressValidation::new(now, ValidateAddress::Never).unwrap(), + )); + server.set_validation(Rc::clone(&validation)); + connect(&mut client, &mut server); + + let id = server.stream_create(StreamType::UniDi).unwrap(); + server + .stream_priority( + id, + TransmissionPriority::Low, + RetransmissionPriority::default(), + ) + .unwrap(); + fill_stream(&mut server, id); + + // Send a session ticket and make it big enough to require a whole packet. + // The resulting CRYPTO frame beats out the stream data. + let stats_before = server.stats().frame_tx; + server.send_ticket(now, &[0; 2048]).unwrap(); + mem::drop(server.process_output(now)); + let stats_after = server.stats().frame_tx; + assert_eq!(stats_after.crypto, stats_before.crypto + 1); + assert_eq!(stats_after.stream, stats_before.stream); + + // The above can't test if NEW_TOKEN wins because once that fits in a packet, + // it is very hard to ensure that the STREAM frame won't also fit. + // However, we can ensure that the next packet doesn't consist of just STREAM. + let stats_before = server.stats().frame_tx; + mem::drop(server.process_output(now)); + let stats_after = server.stats().frame_tx; + assert_eq!(stats_after.crypto, stats_before.crypto + 1); + assert_eq!(stats_after.new_token, 1); + assert_eq!(stats_after.stream, stats_before.stream + 1); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/recovery.rs b/third_party/rust/neqo-transport/src/connection/tests/recovery.rs new file mode 100644 index 0000000000..0f12d03107 --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/recovery.rs @@ -0,0 +1,804 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{ + mem, + time::{Duration, Instant}, +}; + +use neqo_common::qdebug; +use neqo_crypto::AuthenticationStatus; +use test_fixture::{ + assertions::{assert_handshake, assert_initial}, + now, split_datagram, +}; + +use super::{ + super::{Connection, ConnectionParameters, Output, State}, + assert_full_cwnd, connect, connect_force_idle, connect_rtt_idle, connect_with_rtt, cwnd, + default_client, default_server, fill_cwnd, maybe_authenticate, new_client, send_and_receive, + send_something, AT_LEAST_PTO, DEFAULT_RTT, DEFAULT_STREAM_DATA, POST_HANDSHAKE_CWND, +}; +use crate::{ + cc::CWND_MIN, + path::PATH_MTU_V6, + recovery::{ + FAST_PTO_SCALE, MAX_OUTSTANDING_UNACK, MAX_PTO_PACKET_COUNT, MIN_OUTSTANDING_UNACK, + }, + rtt::GRANULARITY, + stats::MAX_PTO_COUNTS, + tparams::TransportParameter, + tracking::DEFAULT_ACK_DELAY, + StreamType, +}; + +#[test] +fn pto_works_basic() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + let mut now = now(); + + let res = client.process(None, now); + let idle_timeout = ConnectionParameters::default().get_idle_timeout(); + assert_eq!(res, Output::Callback(idle_timeout)); + + // Send data on two streams + let stream1 = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(client.stream_send(stream1, b"hello").unwrap(), 5); + assert_eq!(client.stream_send(stream1, b" world!").unwrap(), 7); + + let stream2 = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(client.stream_send(stream2, b"there!").unwrap(), 6); + + // Send a packet after some time. + now += Duration::from_secs(10); + let out = client.process(None, now); + assert!(out.dgram().is_some()); + + // Nothing to do, should return callback + let out = client.process(None, now); + assert!(matches!(out, Output::Callback(_))); + + // One second later, it should want to send PTO packet + now += AT_LEAST_PTO; + let out = client.process(None, now); + + let stream_before = server.stats().frame_rx.stream; + server.process_input(&out.dgram().unwrap(), now); + assert_eq!(server.stats().frame_rx.stream, stream_before + 2); +} + +#[test] +fn pto_works_full_cwnd() { + let mut client = default_client(); + let mut server = default_server(); + let now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + // Send lots of data. + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + let (dgrams, now) = fill_cwnd(&mut client, stream_id, now); + assert_full_cwnd(&dgrams, POST_HANDSHAKE_CWND); + + // Fill the CWND after waiting for a PTO. + let (dgrams, now) = fill_cwnd(&mut client, stream_id, now + AT_LEAST_PTO); + // Two packets in the PTO. + // The first should be full sized; the second might be small. + assert_eq!(dgrams.len(), 2); + assert_eq!(dgrams[0].len(), PATH_MTU_V6); + + // Both datagrams contain one or more STREAM frames. + for d in dgrams { + let stream_before = server.stats().frame_rx.stream; + server.process_input(&d, now); + assert!(server.stats().frame_rx.stream > stream_before); + } +} + +#[test] +fn pto_works_ping() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + let mut now = now() + Duration::from_secs(10); + + // Send a few packets from the client. + let pkt0 = send_something(&mut client, now); + let pkt1 = send_something(&mut client, now); + let pkt2 = send_something(&mut client, now); + let pkt3 = send_something(&mut client, now); + + // Nothing to do, should return callback + let cb = client.process(None, now).callback(); + // The PTO timer is calculated with: + // RTT + max(rttvar * 4, GRANULARITY) + max_ack_delay + // With zero RTT and rttvar, max_ack_delay is minimum too (GRANULARITY) + assert_eq!(cb, GRANULARITY * 2); + + // Process these by server, skipping pkt0 + let srv0 = server.process(Some(&pkt1), now).dgram(); + assert!(srv0.is_some()); // ooo, ack client pkt1 + + now += Duration::from_millis(20); + + // process pkt2 (immediate ack because last ack was more than an RTT ago; RTT=0) + let srv1 = server.process(Some(&pkt2), now).dgram(); + assert!(srv1.is_some()); // this is now dropped + + now += Duration::from_millis(20); + // process pkt3 (acked for same reason) + let srv2 = server.process(Some(&pkt3), now).dgram(); + // ack client pkt 2 & 3 + assert!(srv2.is_some()); + + // client processes ack + let pkt4 = client.process(srv2.as_ref(), now).dgram(); + // client resends data from pkt0 + assert!(pkt4.is_some()); + + // server sees ooo pkt0 and generates immediate ack + let srv3 = server.process(Some(&pkt0), now).dgram(); + assert!(srv3.is_some()); + + // Accept the acknowledgment. + let pkt5 = client.process(srv3.as_ref(), now).dgram(); + assert!(pkt5.is_none()); + + now += Duration::from_millis(70); + // PTO expires. No unacked data. Only send PING. + let client_pings = client.stats().frame_tx.ping; + let pkt6 = client.process(None, now).dgram(); + assert_eq!(client.stats().frame_tx.ping, client_pings + 1); + + let server_pings = server.stats().frame_rx.ping; + server.process_input(&pkt6.unwrap(), now); + assert_eq!(server.stats().frame_rx.ping, server_pings + 1); +} + +#[test] +fn pto_initial() { + const INITIAL_PTO: Duration = Duration::from_millis(300); + let mut now = now(); + + qdebug!("---- client: generate CH"); + let mut client = default_client(); + let pkt1 = client.process(None, now).dgram(); + assert!(pkt1.is_some()); + assert_eq!(pkt1.clone().unwrap().len(), PATH_MTU_V6); + + let delay = client.process(None, now).callback(); + assert_eq!(delay, INITIAL_PTO); + + // Resend initial after PTO. + now += delay; + let pkt2 = client.process(None, now).dgram(); + assert!(pkt2.is_some()); + assert_eq!(pkt2.unwrap().len(), PATH_MTU_V6); + + let delay = client.process(None, now).callback(); + // PTO has doubled. + assert_eq!(delay, INITIAL_PTO * 2); + + // Server process the first initial pkt. + let mut server = default_server(); + let out = server.process(pkt1.as_ref(), now).dgram(); + assert!(out.is_some()); + + // Client receives ack for the first initial packet as well a Handshake packet. + // After the handshake packet the initial keys and the crypto stream for the initial + // packet number space will be discarded. + // Here only an ack for the Handshake packet will be sent. + let out = client.process(out.as_ref(), now).dgram(); + assert!(out.is_some()); + + // We do not have PTO for the resent initial packet any more, but + // the Handshake PTO timer should be armed. As the RTT is apparently + // the same as the initial PTO value, and there is only one sample, + // the PTO will be 3x the INITIAL PTO. + let delay = client.process(None, now).callback(); + assert_eq!(delay, INITIAL_PTO * 3); +} + +/// A complete handshake that involves a PTO in the Handshake space. +#[test] +fn pto_handshake_complete() { + const HALF_RTT: Duration = Duration::from_millis(10); + + let mut now = now(); + // start handshake + let mut client = default_client(); + let mut server = default_server(); + + let pkt = client.process(None, now).dgram(); + assert_initial(pkt.as_ref().unwrap(), false); + let cb = client.process(None, now).callback(); + assert_eq!(cb, Duration::from_millis(300)); + + now += HALF_RTT; + let pkt = server.process(pkt.as_ref(), now).dgram(); + assert_initial(pkt.as_ref().unwrap(), false); + + now += HALF_RTT; + let pkt = client.process(pkt.as_ref(), now).dgram(); + assert_handshake(pkt.as_ref().unwrap()); + + let cb = client.process(None, now).callback(); + // The client now has a single RTT estimate (20ms), so + // the handshake PTO is set based on that. + assert_eq!(cb, HALF_RTT * 6); + + now += HALF_RTT; + let pkt = server.process(pkt.as_ref(), now).dgram(); + assert!(pkt.is_none()); + + now += HALF_RTT; + client.authenticated(AuthenticationStatus::Ok, now); + + qdebug!("---- client: SH..FIN -> FIN"); + let pkt1 = client.process(None, now).dgram(); + assert_handshake(pkt1.as_ref().unwrap()); + assert_eq!(*client.state(), State::Connected); + + let cb = client.process(None, now).callback(); + assert_eq!(cb, HALF_RTT * 6); + + let mut pto_counts = [0; MAX_PTO_COUNTS]; + assert_eq!(client.stats.borrow().pto_counts, pto_counts); + + // Wait for PTO to expire and resend a handshake packet. + // Wait long enough that the 1-RTT PTO also fires. + qdebug!("---- client: PTO"); + now += HALF_RTT * 6; + let pkt2 = client.process(None, now).dgram(); + assert_handshake(pkt2.as_ref().unwrap()); + + pto_counts[0] = 1; + assert_eq!(client.stats.borrow().pto_counts, pto_counts); + + // Get a second PTO packet. + // Add some application data to this datagram, then split the 1-RTT off. + // We'll use that packet to force the server to acknowledge 1-RTT. + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + client.stream_close_send(stream_id).unwrap(); + let pkt3 = client.process(None, now).dgram(); + assert_handshake(pkt3.as_ref().unwrap()); + let (pkt3_hs, pkt3_1rtt) = split_datagram(&pkt3.unwrap()); + assert_handshake(&pkt3_hs); + assert!(pkt3_1rtt.is_some()); + + // PTO has been doubled. + let cb = client.process(None, now).callback(); + assert_eq!(cb, HALF_RTT * 12); + + // We still have only a single PTO + assert_eq!(client.stats.borrow().pto_counts, pto_counts); + + qdebug!("---- server: receive FIN and send ACK"); + now += HALF_RTT; + // Now let the server have pkt1 and expect an immediate Handshake ACK. + // The output will be a Handshake packet with ACK and 1-RTT packet with + // HANDSHAKE_DONE and (because of pkt3_1rtt) an ACK. + // This should remove the 1-RTT PTO from messing this test up. + let server_acks = server.stats().frame_tx.ack; + let server_done = server.stats().frame_tx.handshake_done; + server.process_input(&pkt3_1rtt.unwrap(), now); + let ack = server.process(pkt1.as_ref(), now).dgram(); + assert!(ack.is_some()); + assert_eq!(server.stats().frame_tx.ack, server_acks + 2); + assert_eq!(server.stats().frame_tx.handshake_done, server_done + 1); + + // Check that the other packets (pkt2, pkt3) are Handshake packets. + // The server discarded the Handshake keys already, therefore they are dropped. + // Note that these don't include 1-RTT packets, because 1-RTT isn't send on PTO. + let (pkt2_hs, pkt2_1rtt) = split_datagram(&pkt2.unwrap()); + assert_handshake(&pkt2_hs); + assert!(pkt2_1rtt.is_some()); + let dropped_before1 = server.stats().dropped_rx; + let server_frames = server.stats().frame_rx.all; + server.process_input(&pkt2_hs, now); + assert_eq!(1, server.stats().dropped_rx - dropped_before1); + assert_eq!(server.stats().frame_rx.all, server_frames); + + server.process_input(&pkt2_1rtt.unwrap(), now); + let server_frames2 = server.stats().frame_rx.all; + let dropped_before2 = server.stats().dropped_rx; + server.process_input(&pkt3_hs, now); + assert_eq!(1, server.stats().dropped_rx - dropped_before2); + assert_eq!(server.stats().frame_rx.all, server_frames2); + + now += HALF_RTT; + + // Let the client receive the ACK. + // It should now be wait to acknowledge the HANDSHAKE_DONE. + let cb = client.process(ack.as_ref(), now).callback(); + // The default ack delay is the RTT divided by the default ACK ratio of 4. + let expected_ack_delay = HALF_RTT * 2 / 4; + assert_eq!(cb, expected_ack_delay); + + // Let the ACK delay timer expire. + now += cb; + let out = client.process(None, now).dgram(); + assert!(out.is_some()); +} + +/// Test that PTO in the Handshake space contains the right frames. +#[test] +fn pto_handshake_frames() { + let mut now = now(); + qdebug!("---- client: generate CH"); + let mut client = default_client(); + let pkt = client.process(None, now); + + now += Duration::from_millis(10); + qdebug!("---- server: CH -> SH, EE, CERT, CV, FIN"); + let mut server = default_server(); + let pkt = server.process(pkt.as_dgram_ref(), now); + + now += Duration::from_millis(10); + qdebug!("---- client: cert verification"); + let pkt = client.process(pkt.as_dgram_ref(), now); + + now += Duration::from_millis(10); + mem::drop(server.process(pkt.as_dgram_ref(), now)); + + now += Duration::from_millis(10); + client.authenticated(AuthenticationStatus::Ok, now); + + let stream = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(stream, 2); + assert_eq!(client.stream_send(stream, b"zero").unwrap(), 4); + qdebug!("---- client: SH..FIN -> FIN and 1RTT packet"); + let pkt1 = client.process(None, now).dgram(); + assert!(pkt1.is_some()); + + // Get PTO timer. + let out = client.process(None, now); + assert_eq!(out, Output::Callback(Duration::from_millis(60))); + + // Wait for PTO to expire and resend a handshake packet. + now += Duration::from_millis(60); + let pkt2 = client.process(None, now).dgram(); + assert!(pkt2.is_some()); + + now += Duration::from_millis(10); + let crypto_before = server.stats().frame_rx.crypto; + server.process_input(&pkt2.unwrap(), now); + assert_eq!(server.stats().frame_rx.crypto, crypto_before + 1); +} + +/// In the case that the Handshake takes too many packets, the server might +/// be stalled on the anti-amplification limit. If a Handshake ACK from the +/// client is lost, the client has to keep the PTO timer armed or the server +/// might be unable to send anything, causing a deadlock. +#[test] +fn handshake_ack_pto() { + const RTT: Duration = Duration::from_millis(10); + let mut now = now(); + let mut client = default_client(); + let mut server = default_server(); + // This is a greasing transport parameter, and large enough that the + // server needs to send two Handshake packets. + let big = TransportParameter::Bytes(vec![0; PATH_MTU_V6]); + server.set_local_tparam(0xce16, big).unwrap(); + + let c1 = client.process(None, now).dgram(); + + now += RTT / 2; + let s1 = server.process(c1.as_ref(), now).dgram(); + assert!(s1.is_some()); + let s2 = server.process(None, now).dgram(); + assert!(s1.is_some()); + + // Now let the client have the Initial, but drop the first coalesced Handshake packet. + now += RTT / 2; + let (initial, _) = split_datagram(&s1.unwrap()); + client.process_input(&initial, now); + let c2 = client.process(s2.as_ref(), now).dgram(); + assert!(c2.is_some()); // This is an ACK. Drop it. + let delay = client.process(None, now).callback(); + assert_eq!(delay, RTT * 3); + + let mut pto_counts = [0; MAX_PTO_COUNTS]; + assert_eq!(client.stats.borrow().pto_counts, pto_counts); + + // Wait for the PTO and ensure that the client generates a packet. + now += delay; + let c3 = client.process(None, now).dgram(); + assert!(c3.is_some()); + + now += RTT / 2; + let ping_before = server.stats().frame_rx.ping; + server.process_input(&c3.unwrap(), now); + assert_eq!(server.stats().frame_rx.ping, ping_before + 1); + + pto_counts[0] = 1; + assert_eq!(client.stats.borrow().pto_counts, pto_counts); + + // Now complete the handshake as cheaply as possible. + let dgram = server.process(None, now).dgram(); + client.process_input(&dgram.unwrap(), now); + maybe_authenticate(&mut client); + let dgram = client.process(None, now).dgram(); + assert_eq!(*client.state(), State::Connected); + let dgram = server.process(dgram.as_ref(), now).dgram(); + assert_eq!(*server.state(), State::Confirmed); + client.process_input(&dgram.unwrap(), now); + assert_eq!(*client.state(), State::Confirmed); + + assert_eq!(client.stats.borrow().pto_counts, pto_counts); +} + +#[test] +fn loss_recovery_crash() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let now = now(); + + // The server sends something, but we will drop this. + mem::drop(send_something(&mut server, now)); + + // Then send something again, but let it through. + let ack = send_and_receive(&mut server, &mut client, now); + assert!(ack.is_some()); + + // Have the server process the ACK. + let cb = server.process(ack.as_ref(), now).callback(); + assert!(cb > Duration::from_secs(0)); + + // Now we leap into the future. The server should regard the first + // packet as lost based on time alone. + let dgram = server.process(None, now + AT_LEAST_PTO).dgram(); + assert!(dgram.is_some()); + + // This crashes. + mem::drop(send_something(&mut server, now + AT_LEAST_PTO)); +} + +// If we receive packets after the PTO timer has fired, we won't clear +// the PTO state, but we might need to acknowledge those packets. +// This shouldn't happen, but we found that some implementations do this. +#[test] +fn ack_after_pto() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + let mut now = now(); + + // The client sends and is forced into a PTO. + mem::drop(send_something(&mut client, now)); + + // Jump forward to the PTO and drain the PTO packets. + now += AT_LEAST_PTO; + // We can use MAX_PTO_PACKET_COUNT, because we know the handshake is over. + for _ in 0..MAX_PTO_PACKET_COUNT { + let dgram = client.process(None, now).dgram(); + assert!(dgram.is_some()); + } + assert!(client.process(None, now).dgram().is_none()); + + // The server now needs to send something that will cause the + // client to want to acknowledge it. A little out of order + // delivery is just the thing. + // Note: The server can't ACK anything here, but none of what + // the client has sent so far has been transferred. + mem::drop(send_something(&mut server, now)); + let dgram = send_something(&mut server, now); + + // The client is now after a PTO, but if it receives something + // that demands acknowledgment, it will send just the ACK. + let ack = client.process(Some(&dgram), now).dgram(); + assert!(ack.is_some()); + + // Make sure that the packet only contained an ACK frame. + let all_frames_before = server.stats().frame_rx.all; + let ack_before = server.stats().frame_rx.ack; + server.process_input(&ack.unwrap(), now); + assert_eq!(server.stats().frame_rx.all, all_frames_before + 1); + assert_eq!(server.stats().frame_rx.ack, ack_before + 1); +} + +/// When we declare a packet as lost, we keep it around for a while for another loss period. +/// Those packets should not affect how we report the loss recovery timer. +/// As the loss recovery timer based on RTT we use that to drive the state. +#[test] +fn lost_but_kept_and_lr_timer() { + const RTT: Duration = Duration::from_secs(1); + let mut client = default_client(); + let mut server = default_server(); + let mut now = connect_with_rtt(&mut client, &mut server, now(), RTT); + + // Two packets (p1, p2) are sent at around t=0. The first is lost. + let _p1 = send_something(&mut client, now); + let p2 = send_something(&mut client, now); + + // At t=RTT/2 the server receives the packet and ACKs it. + now += RTT / 2; + let ack = server.process(Some(&p2), now).dgram(); + assert!(ack.is_some()); + // The client also sends another two packets (p3, p4), again losing the first. + let _p3 = send_something(&mut client, now); + let p4 = send_something(&mut client, now); + + // At t=RTT the client receives the ACK and goes into timed loss recovery. + // The client doesn't call p1 lost at this stage, but it will soon. + now += RTT / 2; + let res = client.process(ack.as_ref(), now); + // The client should be on a loss recovery timer as p1 is missing. + let lr_timer = res.callback(); + // Loss recovery timer should be RTT/8, but only check for 0 or >=RTT/2. + assert_ne!(lr_timer, Duration::from_secs(0)); + assert!(lr_timer < (RTT / 2)); + // The server also receives and acknowledges p4, again sending an ACK. + let ack = server.process(Some(&p4), now).dgram(); + assert!(ack.is_some()); + + // At t=RTT*3/2 the client should declare p1 to be lost. + now += RTT / 2; + // So the client will send the data from p1 again. + let res = client.process(None, now); + assert!(res.dgram().is_some()); + // When the client processes the ACK, it should engage the + // loss recovery timer for p3, not p1 (even though it still tracks p1). + let res = client.process(ack.as_ref(), now); + let lr_timer2 = res.callback(); + assert_eq!(lr_timer, lr_timer2); +} + +/// We should not be setting the loss recovery timer based on packets +/// that are sent prior to the largest acknowledged. +/// Testing this requires that we construct a case where one packet +/// number space causes the loss recovery timer to be engaged. At the same time, +/// there is a packet in another space that hasn't been acknowledged AND +/// that packet number space has not received acknowledgments for later packets. +#[test] +fn loss_time_past_largest_acked() { + const RTT: Duration = Duration::from_secs(10); + const INCR: Duration = Duration::from_millis(1); + let mut client = default_client(); + let mut server = default_server(); + + let mut now = now(); + + // Start the handshake. + let c_in = client.process(None, now).dgram(); + now += RTT / 2; + let s_hs1 = server.process(c_in.as_ref(), now).dgram(); + + // Get some spare server handshake packets for the client to ACK. + // This involves a time machine, so be a little cautious. + // This test uses an RTT of 10s, but our server starts + // with a much lower RTT estimate, so the PTO at this point should + // be much smaller than an RTT and so the server shouldn't see + // time go backwards. + let s_pto = server.process(None, now).callback(); + assert_ne!(s_pto, Duration::from_secs(0)); + assert!(s_pto < RTT); + let s_hs2 = server.process(None, now + s_pto).dgram(); + assert!(s_hs2.is_some()); + let s_hs3 = server.process(None, now + s_pto).dgram(); + assert!(s_hs3.is_some()); + + // Get some Handshake packets from the client. + // We need one to be left unacknowledged before one that is acknowledged. + // So that the client engages the loss recovery timer. + // This is complicated by the fact that it is hard to cause the client + // to generate an ack-eliciting packet. For that, we use the Finished message. + // Reordering delivery ensures that the later packet is also acknowledged. + now += RTT / 2; + let c_hs1 = client.process(s_hs1.as_ref(), now).dgram(); + assert!(c_hs1.is_some()); // This comes first, so it's useless. + maybe_authenticate(&mut client); + let c_hs2 = client.process(None, now).dgram(); + assert!(c_hs2.is_some()); // This one will elicit an ACK. + + // The we need the outstanding packet to be sent after the + // application data packet, so space these out a tiny bit. + let _p1 = send_something(&mut client, now + INCR); + let c_hs3 = client.process(s_hs2.as_ref(), now + (INCR * 2)).dgram(); + assert!(c_hs3.is_some()); // This will be left outstanding. + let c_hs4 = client.process(s_hs3.as_ref(), now + (INCR * 3)).dgram(); + assert!(c_hs4.is_some()); // This will be acknowledged. + + // Process c_hs2 and c_hs4, but skip c_hs3. + // Then get an ACK for the client. + now += RTT / 2; + // Deliver c_hs4 first, but don't generate a packet. + server.process_input(&c_hs4.unwrap(), now); + let s_ack = server.process(c_hs2.as_ref(), now).dgram(); + assert!(s_ack.is_some()); + // This includes an ACK, but it also includes HANDSHAKE_DONE, + // which we need to remove because that will cause the Handshake loss + // recovery state to be dropped. + let (s_hs_ack, _s_ap_ack) = split_datagram(&s_ack.unwrap()); + + // Now the client should start its loss recovery timer based on the ACK. + now += RTT / 2; + let c_ack = client.process(Some(&s_hs_ack), now).dgram(); + assert!(c_ack.is_none()); + // The client should now have the loss recovery timer active. + let lr_time = client.process(None, now).callback(); + assert_ne!(lr_time, Duration::from_secs(0)); + assert!(lr_time < (RTT / 2)); +} + +/// `sender` sends a little, `receiver` acknowledges it. +/// Repeat until `count` acknowledgements are sent. +/// Returns the last packet containing acknowledgements, if any. +fn trickle(sender: &mut Connection, receiver: &mut Connection, mut count: usize, now: Instant) { + let id = sender.stream_create(StreamType::UniDi).unwrap(); + let mut maybe_ack = None; + while count > 0 { + qdebug!("trickle: remaining={}", count); + assert_eq!(sender.stream_send(id, &[9]).unwrap(), 1); + let dgram = sender.process(maybe_ack.as_ref(), now).dgram(); + + maybe_ack = receiver.process(dgram.as_ref(), now).dgram(); + count -= usize::from(maybe_ack.is_some()); + } + sender.process_input(&maybe_ack.unwrap(), now); +} + +/// Ensure that a PING frame is sent with ACK sometimes. +/// `fast` allows testing of when `MAX_OUTSTANDING_UNACK` packets are +/// outstanding (`fast` is `true`) within 1 PTO and when only +/// `MIN_OUTSTANDING_UNACK` packets arrive after 2 PTOs (`fast` is `false`). +fn ping_with_ack(fast: bool) { + let mut sender = default_client(); + let mut receiver = default_server(); + let mut now = now(); + connect_force_idle(&mut sender, &mut receiver); + let sender_acks_before = sender.stats().frame_tx.ack; + let receiver_acks_before = receiver.stats().frame_tx.ack; + let count = if fast { + MAX_OUTSTANDING_UNACK + } else { + MIN_OUTSTANDING_UNACK + }; + trickle(&mut sender, &mut receiver, count, now); + assert_eq!(sender.stats().frame_tx.ack, sender_acks_before); + assert_eq!(receiver.stats().frame_tx.ack, receiver_acks_before + count); + assert_eq!(receiver.stats().frame_tx.ping, 0); + + if !fast { + // Wait at least one PTO, from the reciever's perspective. + // A receiver that hasn't received MAX_OUTSTANDING_UNACK won't send PING. + now += receiver.pto() + Duration::from_micros(1); + trickle(&mut sender, &mut receiver, 1, now); + assert_eq!(receiver.stats().frame_tx.ping, 0); + } + + // After a second PTO (or the first if fast), new acknowledgements come + // with a PING frame and cause an ACK to be sent by the sender. + now += receiver.pto() + Duration::from_micros(1); + trickle(&mut sender, &mut receiver, 1, now); + assert_eq!(receiver.stats().frame_tx.ping, 1); + if let Output::Callback(t) = sender.process_output(now) { + assert_eq!(t, DEFAULT_ACK_DELAY); + assert!(sender.process_output(now + t).dgram().is_some()); + } + assert_eq!(sender.stats().frame_tx.ack, sender_acks_before + 1); +} + +#[test] +fn ping_with_ack_fast() { + ping_with_ack(true); +} + +#[test] +fn ping_with_ack_slow() { + ping_with_ack(false); +} + +#[test] +fn ping_with_ack_min() { + const COUNT: usize = MIN_OUTSTANDING_UNACK - 2; + let mut sender = default_client(); + let mut receiver = default_server(); + let mut now = now(); + connect_force_idle(&mut sender, &mut receiver); + let sender_acks_before = sender.stats().frame_tx.ack; + let receiver_acks_before = receiver.stats().frame_tx.ack; + trickle(&mut sender, &mut receiver, COUNT, now); + assert_eq!(sender.stats().frame_tx.ack, sender_acks_before); + assert_eq!(receiver.stats().frame_tx.ack, receiver_acks_before + COUNT); + assert_eq!(receiver.stats().frame_tx.ping, 0); + + // After 3 PTO, no PING because there are too few outstanding packets. + now += receiver.pto() * 3 + Duration::from_micros(1); + trickle(&mut sender, &mut receiver, 1, now); + assert_eq!(receiver.stats().frame_tx.ping, 0); +} + +/// This calculates the PTO timer immediately after connection establishment. +/// It depends on there only being 2 RTT samples in the handshake. +fn expected_pto(rtt: Duration) -> Duration { + // PTO calculation is rtt + 4rttvar + ack delay. + // rttvar should be (rtt + 4 * (rtt / 2) * (3/4)^n + 25ms)/2 + // where n is the number of round trips + // This uses a 25ms ack delay as the ACK delay extension + // is negotiated and no ACK_DELAY frame has been received. + rtt + rtt * 9 / 8 + Duration::from_millis(25) +} + +#[test] +fn fast_pto() { + let mut client = new_client(ConnectionParameters::default().fast_pto(FAST_PTO_SCALE / 2)); + let mut server = default_server(); + let mut now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + let res = client.process(None, now); + let idle_timeout = ConnectionParameters::default().get_idle_timeout() - (DEFAULT_RTT / 2); + assert_eq!(res, Output::Callback(idle_timeout)); + + // Send data on two streams + let stream = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!( + client.stream_send(stream, DEFAULT_STREAM_DATA).unwrap(), + DEFAULT_STREAM_DATA.len() + ); + + // Send a packet after some time. + now += idle_timeout / 2; + let dgram = client.process_output(now).dgram(); + assert!(dgram.is_some()); + + // Nothing to do, should return a callback. + let cb = client.process_output(now).callback(); + assert_eq!(expected_pto(DEFAULT_RTT) / 2, cb); + + // Once the PTO timer expires, a PTO packet should be sent should want to send PTO packet. + now += cb; + let dgram = client.process(None, now).dgram(); + + let stream_before = server.stats().frame_rx.stream; + server.process_input(&dgram.unwrap(), now); + assert_eq!(server.stats().frame_rx.stream, stream_before + 1); +} + +/// Even if the PTO timer is slowed right down, persistent congestion is declared +/// based on the "true" value of the timer. +#[test] +fn fast_pto_persistent_congestion() { + let mut client = new_client(ConnectionParameters::default().fast_pto(FAST_PTO_SCALE * 2)); + let mut server = default_server(); + let mut now = connect_rtt_idle(&mut client, &mut server, DEFAULT_RTT); + + let res = client.process(None, now); + let idle_timeout = ConnectionParameters::default().get_idle_timeout() - (DEFAULT_RTT / 2); + assert_eq!(res, Output::Callback(idle_timeout)); + + // Send packets spaced by the PTO timer. And lose them. + // Note: This timing is a tiny bit higher than the client will use + // to determine persistent congestion. The ACK below adds another RTT + // estimate, which will reduce rttvar by 3/4, so persistent congestion + // will occur at `rtt + rtt*27/32 + 25ms`. + // That is OK as we're still showing that this interval is less than + // six times the PTO, which is what would be used if the scaling + // applied to the PTO used to determine persistent congestion. + let pc_interval = expected_pto(DEFAULT_RTT) * 3; + println!("pc_interval {pc_interval:?}"); + let _drop1 = send_something(&mut client, now); + + // Check that the PTO matches expectations. + let cb = client.process_output(now).callback(); + assert_eq!(expected_pto(DEFAULT_RTT) * 2, cb); + + now += pc_interval; + let _drop2 = send_something(&mut client, now); + let _drop3 = send_something(&mut client, now); + let _drop4 = send_something(&mut client, now); + let dgram = send_something(&mut client, now); + + // Now acknowledge the tail packet and enter persistent congestion. + now += DEFAULT_RTT / 2; + let ack = server.process(Some(&dgram), now).dgram(); + now += DEFAULT_RTT / 2; + client.process_input(&ack.unwrap(), now); + assert_eq!(cwnd(&client), CWND_MIN); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/resumption.rs b/third_party/rust/neqo-transport/src/connection/tests/resumption.rs new file mode 100644 index 0000000000..a8c45a9f06 --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/resumption.rs @@ -0,0 +1,246 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{cell::RefCell, mem, rc::Rc, time::Duration}; + +use test_fixture::{self, assertions, now}; + +use super::{ + connect, connect_with_rtt, default_client, default_server, exchange_ticket, get_tokens, + new_client, resumed_server, send_something, AT_LEAST_PTO, +}; +use crate::{ + addr_valid::{AddressValidation, ValidateAddress}, + ConnectionParameters, Error, Version, +}; + +#[test] +fn resume() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let token = exchange_ticket(&mut client, &mut server, now()); + let mut client = default_client(); + client + .enable_resumption(now(), token) + .expect("should set token"); + let mut server = resumed_server(&client); + connect(&mut client, &mut server); + assert!(client.tls_info().unwrap().resumed()); + assert!(server.tls_info().unwrap().resumed()); +} + +#[test] +fn remember_smoothed_rtt() { + const RTT1: Duration = Duration::from_millis(130); + const RTT2: Duration = Duration::from_millis(70); + + let mut client = default_client(); + let mut server = default_server(); + + let mut now = connect_with_rtt(&mut client, &mut server, now(), RTT1); + assert_eq!(client.paths.rtt(), RTT1); + + // We can't use exchange_ticket here because it doesn't respect RTT. + // Also, connect_with_rtt() ends with the server receiving a packet it + // wants to acknowledge; so the ticket will include an ACK frame too. + let validation = AddressValidation::new(now, ValidateAddress::NoToken).unwrap(); + let validation = Rc::new(RefCell::new(validation)); + server.set_validation(Rc::clone(&validation)); + server.send_ticket(now, &[]).expect("can send ticket"); + let ticket = server.process_output(now).dgram(); + assert!(ticket.is_some()); + now += RTT1 / 2; + client.process_input(&ticket.unwrap(), now); + let token = get_tokens(&mut client).pop().unwrap(); + + let mut client = default_client(); + client.enable_resumption(now, token).unwrap(); + assert_eq!( + client.paths.rtt(), + RTT1, + "client should remember previous RTT" + ); + let mut server = resumed_server(&client); + + connect_with_rtt(&mut client, &mut server, now, RTT2); + assert_eq!( + client.paths.rtt(), + RTT2, + "previous RTT should be completely erased" + ); +} + +/// Check that a resumed connection uses a token on Initial packets. +#[test] +fn address_validation_token_resume() { + const RTT: Duration = Duration::from_millis(10); + + let mut client = default_client(); + let mut server = default_server(); + let validation = AddressValidation::new(now(), ValidateAddress::Always).unwrap(); + let validation = Rc::new(RefCell::new(validation)); + server.set_validation(Rc::clone(&validation)); + let mut now = connect_with_rtt(&mut client, &mut server, now(), RTT); + + let token = exchange_ticket(&mut client, &mut server, now); + let mut client = default_client(); + client.enable_resumption(now, token).unwrap(); + let mut server = resumed_server(&client); + + // Grab an Initial packet from the client. + let dgram = client.process(None, now).dgram(); + assertions::assert_initial(dgram.as_ref().unwrap(), true); + + // Now try to complete the handshake after giving time for a client PTO. + now += AT_LEAST_PTO; + connect_with_rtt(&mut client, &mut server, now, RTT); + assert!(client.crypto.tls.info().unwrap().resumed()); + assert!(server.crypto.tls.info().unwrap().resumed()); +} + +fn can_resume(token: impl AsRef<[u8]>, initial_has_token: bool) { + let mut client = default_client(); + client.enable_resumption(now(), token).unwrap(); + let initial = client.process_output(now()).dgram(); + assertions::assert_initial(initial.as_ref().unwrap(), initial_has_token); +} + +#[test] +fn two_tickets_on_timer() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + // Send two tickets and then bundle those into a packet. + server.send_ticket(now(), &[]).expect("send ticket1"); + server.send_ticket(now(), &[]).expect("send ticket2"); + let pkt = send_something(&mut server, now()); + + // process() will return an ack first + assert!(client.process(Some(&pkt), now()).dgram().is_some()); + // We do not have a ResumptionToken event yet, because NEW_TOKEN was not sent. + assert_eq!(get_tokens(&mut client).len(), 0); + + // We need to wait for release_resumption_token_timer to expire. The timer will be + // set to 3 * PTO + let mut now = now() + 3 * client.pto(); + mem::drop(client.process(None, now)); + let mut recv_tokens = get_tokens(&mut client); + assert_eq!(recv_tokens.len(), 1); + let token1 = recv_tokens.pop().unwrap(); + // Wai for anottheer 3 * PTO to get the nex okeen. + now += 3 * client.pto(); + mem::drop(client.process(None, now)); + let mut recv_tokens = get_tokens(&mut client); + assert_eq!(recv_tokens.len(), 1); + let token2 = recv_tokens.pop().unwrap(); + // Wait for 3 * PTO, but now there are no more tokens. + now += 3 * client.pto(); + mem::drop(client.process(None, now)); + assert_eq!(get_tokens(&mut client).len(), 0); + assert_ne!(token1.as_ref(), token2.as_ref()); + + can_resume(token1, false); + can_resume(token2, false); +} + +#[test] +fn two_tickets_with_new_token() { + let mut client = default_client(); + let mut server = default_server(); + let validation = AddressValidation::new(now(), ValidateAddress::Always).unwrap(); + let validation = Rc::new(RefCell::new(validation)); + server.set_validation(Rc::clone(&validation)); + connect(&mut client, &mut server); + + // Send two tickets with tokens and then bundle those into a packet. + server.send_ticket(now(), &[]).expect("send ticket1"); + server.send_ticket(now(), &[]).expect("send ticket2"); + let pkt = send_something(&mut server, now()); + + client.process_input(&pkt, now()); + let mut all_tokens = get_tokens(&mut client); + assert_eq!(all_tokens.len(), 2); + let token1 = all_tokens.pop().unwrap(); + let token2 = all_tokens.pop().unwrap(); + assert_ne!(token1.as_ref(), token2.as_ref()); + + can_resume(token1, true); + can_resume(token2, true); +} + +/// By disabling address validation, the server won't send `NEW_TOKEN`, but +/// we can take the session ticket still. +#[test] +fn take_token() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + server.send_ticket(now(), &[]).unwrap(); + let dgram = server.process(None, now()).dgram(); + client.process_input(&dgram.unwrap(), now()); + + // There should be no ResumptionToken event here. + let tokens = get_tokens(&mut client); + assert_eq!(tokens.len(), 0); + + // But we should be able to get the token directly, and use it. + let token = client.take_resumption_token(now()).unwrap(); + can_resume(token, false); +} + +/// If a version is selected and subsequently disabled, resumption fails. +#[test] +fn resume_disabled_version() { + let mut client = new_client( + ConnectionParameters::default().versions(Version::Version1, vec![Version::Version1]), + ); + let mut server = default_server(); + connect(&mut client, &mut server); + let token = exchange_ticket(&mut client, &mut server, now()); + + let mut client = new_client( + ConnectionParameters::default().versions(Version::Version2, vec![Version::Version2]), + ); + assert_eq!( + client.enable_resumption(now(), token).unwrap_err(), + Error::DisabledVersion + ); +} + +/// It's not possible to resume once a packet has been sent. +#[test] +fn resume_after_packet() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let token = exchange_ticket(&mut client, &mut server, now()); + + let mut client = default_client(); + mem::drop(client.process_output(now()).dgram().unwrap()); + assert_eq!( + client.enable_resumption(now(), token).unwrap_err(), + Error::ConnectionState + ); +} + +/// It's not possible to resume at the server. +#[test] +fn resume_server() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + let token = exchange_ticket(&mut client, &mut server, now()); + + let mut server = default_server(); + assert_eq!( + server.enable_resumption(now(), token).unwrap_err(), + Error::ConnectionState + ); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/stream.rs b/third_party/rust/neqo-transport/src/connection/tests/stream.rs new file mode 100644 index 0000000000..586a537b9d --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/stream.rs @@ -0,0 +1,1162 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{cmp::max, collections::HashMap, convert::TryFrom, mem}; + +use neqo_common::{event::Provider, qdebug}; +use test_fixture::now; + +use super::{ + super::State, assert_error, connect, connect_force_idle, default_client, default_server, + maybe_authenticate, new_client, new_server, send_something, DEFAULT_STREAM_DATA, +}; +use crate::{ + events::ConnectionEvent, + recv_stream::RECV_BUFFER_SIZE, + send_stream::{OrderGroup, SendStreamState, SEND_BUFFER_SIZE}, + streams::{SendOrder, StreamOrder}, + tparams::{self, TransportParameter}, + // tracking::DEFAULT_ACK_PACKET_TOLERANCE, + Connection, + ConnectionError, + ConnectionParameters, + Error, + StreamId, + StreamType, +}; + +#[test] +fn stream_create() { + let mut client = default_client(); + + let out = client.process(None, now()); + let mut server = default_server(); + let out = server.process(out.as_dgram_ref(), now()); + + let out = client.process(out.as_dgram_ref(), now()); + mem::drop(server.process(out.as_dgram_ref(), now())); + assert!(maybe_authenticate(&mut client)); + let out = client.process(None, now()); + + // client now in State::Connected + assert_eq!(client.stream_create(StreamType::UniDi).unwrap(), 2); + assert_eq!(client.stream_create(StreamType::UniDi).unwrap(), 6); + assert_eq!(client.stream_create(StreamType::BiDi).unwrap(), 0); + assert_eq!(client.stream_create(StreamType::BiDi).unwrap(), 4); + + mem::drop(server.process(out.as_dgram_ref(), now())); + // server now in State::Connected + assert_eq!(server.stream_create(StreamType::UniDi).unwrap(), 3); + assert_eq!(server.stream_create(StreamType::UniDi).unwrap(), 7); + assert_eq!(server.stream_create(StreamType::BiDi).unwrap(), 1); + assert_eq!(server.stream_create(StreamType::BiDi).unwrap(), 5); +} + +#[test] +// tests stream send/recv after connection is established. +fn transfer() { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + qdebug!("---- client sends"); + // Send + let client_stream_id = client.stream_create(StreamType::UniDi).unwrap(); + client.stream_send(client_stream_id, &[6; 100]).unwrap(); + client.stream_send(client_stream_id, &[7; 40]).unwrap(); + client.stream_send(client_stream_id, &[8; 4000]).unwrap(); + + // Send to another stream but some data after fin has been set + let client_stream_id2 = client.stream_create(StreamType::UniDi).unwrap(); + client.stream_send(client_stream_id2, &[6; 60]).unwrap(); + client.stream_close_send(client_stream_id2).unwrap(); + client.stream_send(client_stream_id2, &[7; 50]).unwrap_err(); + // Sending this much takes a few datagrams. + let mut datagrams = vec![]; + let mut out = client.process_output(now()); + while let Some(d) = out.dgram() { + datagrams.push(d); + out = client.process_output(now()); + } + assert_eq!(datagrams.len(), 4); + assert_eq!(*client.state(), State::Confirmed); + + qdebug!("---- server receives"); + for d in datagrams { + let out = server.process(Some(&d), now()); + // With an RTT of zero, the server will acknowledge every packet immediately. + assert!(out.as_dgram_ref().is_some()); + qdebug!("Output={:0x?}", out.as_dgram_ref()); + } + assert_eq!(*server.state(), State::Confirmed); + + let mut buf = vec![0; 4000]; + + let mut stream_ids = server.events().filter_map(|evt| match evt { + ConnectionEvent::NewStream { stream_id, .. } => Some(stream_id), + _ => None, + }); + let first_stream = stream_ids.next().expect("should have a new stream event"); + let second_stream = stream_ids + .next() + .expect("should have a second new stream event"); + assert!(stream_ids.next().is_none()); + let (received1, fin1) = server.stream_recv(first_stream, &mut buf).unwrap(); + assert_eq!(received1, 4000); + assert!(!fin1); + let (received2, fin2) = server.stream_recv(first_stream, &mut buf).unwrap(); + assert_eq!(received2, 140); + assert!(!fin2); + + let (received3, fin3) = server.stream_recv(second_stream, &mut buf).unwrap(); + assert_eq!(received3, 60); + assert!(fin3); +} + +#[derive(PartialEq, Eq, PartialOrd, Ord)] +struct IdEntry { + sendorder: StreamOrder, + stream_id: StreamId, +} + +// tests stream sendorder priorization +fn sendorder_test(order_of_sendorder: &[Option<SendOrder>]) { + let mut client = default_client(); + let mut server = default_server(); + connect_force_idle(&mut client, &mut server); + + qdebug!("---- client sends"); + // open all streams and set the sendorders + let mut ordered = Vec::new(); + let mut streams = Vec::<StreamId>::new(); + for sendorder in order_of_sendorder { + let id = client.stream_create(StreamType::UniDi).unwrap(); + streams.push(id); + ordered.push((id, *sendorder)); + // must be set before sendorder + client.streams.set_fairness(id, true).ok(); + client.streams.set_sendorder(id, *sendorder).ok(); + } + // Write some data to all the streams + for stream_id in streams { + client.stream_send(stream_id, &[6; 100]).unwrap(); + } + + // Sending this much takes a few datagrams. + // Note: this test uses an RTT of 0 which simplifies things (no pacing) + let mut datagrams = Vec::new(); + let mut out = client.process_output(now()); + while let Some(d) = out.dgram() { + datagrams.push(d); + out = client.process_output(now()); + } + assert_eq!(*client.state(), State::Confirmed); + + qdebug!("---- server receives"); + for d in datagrams { + let out = server.process(Some(&d), now()); + qdebug!("Output={:0x?}", out.as_dgram_ref()); + } + assert_eq!(*server.state(), State::Confirmed); + + let stream_ids = server + .events() + .filter_map(|evt| match evt { + ConnectionEvent::RecvStreamReadable { stream_id, .. } => Some(stream_id), + _ => None, + }) + .enumerate() + .map(|(a, b)| (b, a)) + .collect::<HashMap<_, _>>(); + + // streams should arrive in priority order, not order of creation, if sendorder prioritization + // is working correctly + + // 'ordered' has the send order currently. Re-sort it by sendorder, but + // if two items from the same sendorder exist, secondarily sort by the ordering in + // the stream_ids vector (HashMap<StreamId, index: usize>) + ordered.sort_unstable_by_key(|(stream_id, sendorder)| { + ( + StreamOrder { + sendorder: *sendorder, + }, + stream_ids[stream_id], + ) + }); + // make sure everything now is in the same order, since we modified the order of + // same-sendorder items to match the ordering of those we saw in reception + for (i, (stream_id, _sendorder)) in ordered.iter().enumerate() { + assert_eq!(i, stream_ids[stream_id]); + } +} + +#[test] +fn sendorder_0() { + sendorder_test(&[None, Some(1), Some(2), Some(3)]); +} +#[test] +fn sendorder_1() { + sendorder_test(&[Some(3), Some(2), Some(1), None]); +} +#[test] +fn sendorder_2() { + sendorder_test(&[Some(3), None, Some(2), Some(1)]); +} +#[test] +fn sendorder_3() { + sendorder_test(&[Some(1), Some(2), None, Some(3)]); +} +#[test] +fn sendorder_4() { + sendorder_test(&[ + Some(1), + Some(2), + Some(1), + None, + Some(3), + Some(1), + Some(3), + None, + ]); +} + +// Tests stream sendorder priorization +// Converts Vecs of u64's into StreamIds +fn fairness_test<S, R>(source: S, number_iterates: usize, truncate_to: usize, result_array: &R) +where + S: IntoIterator, + S::Item: Into<StreamId>, + R: IntoIterator + std::fmt::Debug, + R::Item: Into<StreamId>, + Vec<u64>: PartialEq<R>, +{ + // test the OrderGroup code used for fairness + let mut group: OrderGroup = OrderGroup::default(); + for stream_id in source { + group.insert(stream_id.into()); + } + { + let mut iterator1 = group.iter(); + // advance_by() would help here + let mut n = number_iterates; + while n > 0 { + iterator1.next(); + n -= 1; + } + // let iterator1 go out of scope + } + group.truncate(truncate_to); + + let iterator2 = group.iter(); + let result: Vec<u64> = iterator2.map(StreamId::as_u64).collect(); + assert_eq!(result, *result_array); +} + +#[test] +fn ordergroup_0() { + let source: [u64; 0] = []; + let result: [u64; 0] = []; + fairness_test(source, 1, usize::MAX, &result); +} + +#[test] +fn ordergroup_1() { + let source: [u64; 6] = [0, 1, 2, 3, 4, 5]; + let result: [u64; 6] = [1, 2, 3, 4, 5, 0]; + fairness_test(source, 1, usize::MAX, &result); +} + +#[test] +fn ordergroup_2() { + let source: [u64; 6] = [0, 1, 2, 3, 4, 5]; + let result: [u64; 6] = [2, 3, 4, 5, 0, 1]; + fairness_test(source, 2, usize::MAX, &result); +} + +#[test] +fn ordergroup_3() { + let source: [u64; 6] = [0, 1, 2, 3, 4, 5]; + let result: [u64; 6] = [0, 1, 2, 3, 4, 5]; + fairness_test(source, 10, usize::MAX, &result); +} + +#[test] +fn ordergroup_4() { + let source: [u64; 6] = [0, 1, 2, 3, 4, 5]; + let result: [u64; 6] = [0, 1, 2, 3, 4, 5]; + fairness_test(source, 0, usize::MAX, &result); +} + +#[test] +fn ordergroup_5() { + let source: [u64; 1] = [0]; + let result: [u64; 1] = [0]; + fairness_test(source, 1, usize::MAX, &result); +} + +#[test] +fn ordergroup_6() { + let source: [u64; 6] = [0, 1, 2, 3, 4, 5]; + let result: [u64; 6] = [5, 0, 1, 2, 3, 4]; + fairness_test(source, 5, usize::MAX, &result); +} + +#[test] +fn ordergroup_7() { + let source: [u64; 6] = [0, 1, 2, 3, 4, 5]; + let result: [u64; 3] = [0, 1, 2]; + fairness_test(source, 5, 3, &result); +} + +#[test] +// Send fin even if a peer closes a reomte bidi send stream before sending any data. +fn report_fin_when_stream_closed_wo_data() { + // Note that the two servers in this test will get different anti-replay filters. + // That's OK because we aren't testing anti-replay. + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + // create a stream + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + client.stream_send(stream_id, &[0x00]).unwrap(); + let out = client.process(None, now()); + mem::drop(server.process(out.as_dgram_ref(), now())); + + server.stream_close_send(stream_id).unwrap(); + let out = server.process(None, now()); + mem::drop(client.process(out.as_dgram_ref(), now())); + let stream_readable = |e| matches!(e, ConnectionEvent::RecvStreamReadable { .. }); + assert!(client.events().any(stream_readable)); +} + +fn exchange_data(client: &mut Connection, server: &mut Connection) { + let mut input = None; + loop { + let out = client.process(input.as_ref(), now()).dgram(); + let c_done = out.is_none(); + let out = server.process(out.as_ref(), now()).dgram(); + if out.is_none() && c_done { + break; + } + input = out; + } +} + +#[test] +fn sending_max_data() { + const SMALL_MAX_DATA: usize = 2048; + + let mut client = default_client(); + let mut server = new_server( + ConnectionParameters::default().max_data(u64::try_from(SMALL_MAX_DATA).unwrap()), + ); + + connect(&mut client, &mut server); + + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(client.events().count(), 2); // SendStreamWritable, StateChange(connected) + assert_eq!(stream_id, 2); + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SMALL_MAX_DATA + ); + + assert_eq!( + client + .stream_send(stream_id, &[b'a'; SMALL_MAX_DATA + 1]) + .unwrap(), + SMALL_MAX_DATA + ); + + exchange_data(&mut client, &mut server); + + let mut buf = vec![0; 40000]; + let (received, fin) = server.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(received, SMALL_MAX_DATA); + assert!(!fin); + + let out = server.process(None, now()).dgram(); + client.process_input(&out.unwrap(), now()); + + assert_eq!( + client + .stream_send(stream_id, &[b'a'; SMALL_MAX_DATA + 1]) + .unwrap(), + SMALL_MAX_DATA + ); +} + +#[test] +fn max_data() { + const SMALL_MAX_DATA: usize = 16383; + + let mut client = default_client(); + let mut server = default_server(); + + server + .set_local_tparam( + tparams::INITIAL_MAX_DATA, + TransportParameter::Integer(u64::try_from(SMALL_MAX_DATA).unwrap()), + ) + .unwrap(); + + connect(&mut client, &mut server); + + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(client.events().count(), 2); // SendStreamWritable, StateChange(connected) + assert_eq!(stream_id, 2); + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SMALL_MAX_DATA + ); + assert_eq!( + client + .stream_send(stream_id, &[b'a'; SMALL_MAX_DATA + 1]) + .unwrap(), + SMALL_MAX_DATA + ); + assert_eq!(client.events().count(), 0); + + assert_eq!(client.stream_send(stream_id, b"hello").unwrap(), 0); + client + .streams + .get_send_stream_mut(stream_id) + .unwrap() + .mark_as_sent(0, 4096, false); + assert_eq!(client.events().count(), 0); + client + .streams + .get_send_stream_mut(stream_id) + .unwrap() + .mark_as_acked(0, 4096, false); + assert_eq!(client.events().count(), 0); + + assert_eq!(client.stream_send(stream_id, b"hello").unwrap(), 0); + // no event because still limited by conn max data + assert_eq!(client.events().count(), 0); + + // Increase max data. Avail space now limited by stream credit + client.streams.handle_max_data(100_000_000); + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SEND_BUFFER_SIZE - SMALL_MAX_DATA + ); + + // Increase max stream data. Avail space now limited by tx buffer + client + .streams + .get_send_stream_mut(stream_id) + .unwrap() + .set_max_stream_data(100_000_000); + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SEND_BUFFER_SIZE - SMALL_MAX_DATA + 4096 + ); + + let evts = client.events().collect::<Vec<_>>(); + assert_eq!(evts.len(), 1); + assert!(matches!( + evts[0], + ConnectionEvent::SendStreamWritable { .. } + )); +} + +#[test] +fn exceed_max_data() { + const SMALL_MAX_DATA: usize = 1024; + + let mut client = default_client(); + let mut server = new_server( + ConnectionParameters::default().max_data(u64::try_from(SMALL_MAX_DATA).unwrap()), + ); + + connect(&mut client, &mut server); + + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(client.events().count(), 2); // SendStreamWritable, StateChange(connected) + assert_eq!(stream_id, 2); + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SMALL_MAX_DATA + ); + assert_eq!( + client + .stream_send(stream_id, &[b'a'; SMALL_MAX_DATA + 1]) + .unwrap(), + SMALL_MAX_DATA + ); + + assert_eq!(client.stream_send(stream_id, b"hello").unwrap(), 0); + + // Artificially trick the client to think that it has more flow control credit. + client.streams.handle_max_data(100_000_000); + assert_eq!(client.stream_send(stream_id, b"h").unwrap(), 1); + + exchange_data(&mut client, &mut server); + + assert_error( + &client, + &ConnectionError::Transport(Error::PeerError(Error::FlowControlError.code())), + ); + assert_error( + &server, + &ConnectionError::Transport(Error::FlowControlError), + ); +} + +#[test] +// If we send a stop_sending to the peer, we should not accept more data from the peer. +fn do_not_accept_data_after_stop_sending() { + // Note that the two servers in this test will get different anti-replay filters. + // That's OK because we aren't testing anti-replay. + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + // create a stream + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + client.stream_send(stream_id, &[0x00]).unwrap(); + let out = client.process(None, now()); + mem::drop(server.process(out.as_dgram_ref(), now())); + + let stream_readable = |e| matches!(e, ConnectionEvent::RecvStreamReadable { .. }); + assert!(server.events().any(stream_readable)); + + // Send one more packet from client. The packet should arrive after the server + // has already requested stop_sending. + client.stream_send(stream_id, &[0x00]).unwrap(); + let out_second_data_frame = client.process(None, now()); + // Call stop sending. + assert_eq!( + Ok(()), + server.stream_stop_sending(stream_id, Error::NoError.code()) + ); + + // Receive the second data frame. The frame should be ignored and + // DataReadable events shouldn't be posted. + let out = server.process(out_second_data_frame.as_dgram_ref(), now()); + assert!(!server.events().any(stream_readable)); + + mem::drop(client.process(out.as_dgram_ref(), now())); + assert_eq!( + Err(Error::FinalSizeError), + client.stream_send(stream_id, &[0x00]) + ); +} + +#[test] +// Server sends stop_sending, the client simultaneous sends reset. +fn simultaneous_stop_sending_and_reset() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + // create a stream + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + client.stream_send(stream_id, &[0x00]).unwrap(); + let out = client.process(None, now()); + let ack = server.process(out.as_dgram_ref(), now()).dgram(); + + let stream_readable = + |e| matches!(e, ConnectionEvent::RecvStreamReadable { stream_id: id } if id == stream_id); + assert!(server.events().any(stream_readable)); + + // The client resets the stream. The packet with reset should arrive after the server + // has already requested stop_sending. + client.stream_reset_send(stream_id, 0).unwrap(); + let out_reset_frame = client.process(ack.as_ref(), now()).dgram(); + + // Send something out of order to force the server to generate an + // acknowledgment at the next opportunity. + let force_ack = send_something(&mut client, now()); + server.process_input(&force_ack, now()); + + // Call stop sending. + server.stream_stop_sending(stream_id, 0).unwrap(); + // Receive the second data frame. The frame should be ignored and + // DataReadable events shouldn't be posted. + let ack = server.process(out_reset_frame.as_ref(), now()).dgram(); + assert!(ack.is_some()); + assert!(!server.events().any(stream_readable)); + + // The client gets the STOP_SENDING frame. + client.process_input(&ack.unwrap(), now()); + assert_eq!( + Err(Error::InvalidStreamId), + client.stream_send(stream_id, &[0x00]) + ); +} + +#[test] +fn client_fin_reorder() { + let mut client = default_client(); + let mut server = default_server(); + + // Send ClientHello. + let client_hs = client.process(None, now()); + assert!(client_hs.as_dgram_ref().is_some()); + + let server_hs = server.process(client_hs.as_dgram_ref(), now()); + assert!(server_hs.as_dgram_ref().is_some()); // ServerHello, etc... + + let client_ack = client.process(server_hs.as_dgram_ref(), now()); + assert!(client_ack.as_dgram_ref().is_some()); + + let server_out = server.process(client_ack.as_dgram_ref(), now()); + assert!(server_out.as_dgram_ref().is_none()); + + assert!(maybe_authenticate(&mut client)); + assert_eq!(*client.state(), State::Connected); + + let client_fin = client.process(None, now()); + assert!(client_fin.as_dgram_ref().is_some()); + + let client_stream_id = client.stream_create(StreamType::UniDi).unwrap(); + client.stream_send(client_stream_id, &[1, 2, 3]).unwrap(); + let client_stream_data = client.process(None, now()); + assert!(client_stream_data.as_dgram_ref().is_some()); + + // Now stream data gets before client_fin + let server_out = server.process(client_stream_data.as_dgram_ref(), now()); + assert!(server_out.as_dgram_ref().is_none()); // the packet will be discarded + + assert_eq!(*server.state(), State::Handshaking); + let server_out = server.process(client_fin.as_dgram_ref(), now()); + assert!(server_out.as_dgram_ref().is_some()); +} + +#[test] +fn after_fin_is_read_conn_events_for_stream_should_be_removed() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let id = server.stream_create(StreamType::BiDi).unwrap(); + server.stream_send(id, &[6; 10]).unwrap(); + server.stream_close_send(id).unwrap(); + let out = server.process(None, now()).dgram(); + assert!(out.is_some()); + + mem::drop(client.process(out.as_ref(), now())); + + // read from the stream before checking connection events. + let mut buf = vec![0; 4000]; + let (_, fin) = client.stream_recv(id, &mut buf).unwrap(); + assert!(fin); + + // Make sure we do not have RecvStreamReadable events for the stream when fin has been read. + let readable_stream_evt = + |e| matches!(e, ConnectionEvent::RecvStreamReadable { stream_id } if stream_id == id); + assert!(!client.events().any(readable_stream_evt)); +} + +#[test] +fn after_stream_stop_sending_is_called_conn_events_for_stream_should_be_removed() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let id = server.stream_create(StreamType::BiDi).unwrap(); + server.stream_send(id, &[6; 10]).unwrap(); + server.stream_close_send(id).unwrap(); + let out = server.process(None, now()).dgram(); + assert!(out.is_some()); + + mem::drop(client.process(out.as_ref(), now())); + + // send stop seending. + client + .stream_stop_sending(id, Error::NoError.code()) + .unwrap(); + + // Make sure we do not have RecvStreamReadable events for the stream after stream_stop_sending + // has been called. + let readable_stream_evt = + |e| matches!(e, ConnectionEvent::RecvStreamReadable { stream_id } if stream_id == id); + assert!(!client.events().any(readable_stream_evt)); +} + +#[test] +fn stream_data_blocked_generates_max_stream_data() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let now = now(); + + // Send some data and consume some flow control. + let stream_id = server.stream_create(StreamType::UniDi).unwrap(); + _ = server.stream_send(stream_id, DEFAULT_STREAM_DATA).unwrap(); + let dgram = server.process(None, now).dgram(); + assert!(dgram.is_some()); + + // Consume the data. + client.process_input(&dgram.unwrap(), now); + let mut buf = [0; 10]; + let (count, end) = client.stream_recv(stream_id, &mut buf[..]).unwrap(); + assert_eq!(count, DEFAULT_STREAM_DATA.len()); + assert!(!end); + + // Now send `STREAM_DATA_BLOCKED`. + let internal_stream = server.streams.get_send_stream_mut(stream_id).unwrap(); + if let SendStreamState::Send { fc, .. } = internal_stream.state() { + fc.blocked(); + } else { + panic!("unexpected stream state"); + } + let dgram = server.process_output(now).dgram(); + assert!(dgram.is_some()); + + let sdb_before = client.stats().frame_rx.stream_data_blocked; + let dgram = client.process(dgram.as_ref(), now).dgram(); + assert_eq!(client.stats().frame_rx.stream_data_blocked, sdb_before + 1); + assert!(dgram.is_some()); + + // Client should have sent a MAX_STREAM_DATA frame with just a small increase + // on the default window size. + let msd_before = server.stats().frame_rx.max_stream_data; + server.process_input(&dgram.unwrap(), now); + assert_eq!(server.stats().frame_rx.max_stream_data, msd_before + 1); + + // Test that the entirety of the receive buffer is available now. + let mut written = 0; + loop { + const LARGE_BUFFER: &[u8] = &[0; 1024]; + let amount = server.stream_send(stream_id, LARGE_BUFFER).unwrap(); + if amount == 0 { + break; + } + written += amount; + } + assert_eq!(written, RECV_BUFFER_SIZE); +} + +/// See <https://github.com/mozilla/neqo/issues/871> +#[test] +fn max_streams_after_bidi_closed() { + const REQUEST: &[u8] = b"ping"; + const RESPONSE: &[u8] = b"pong"; + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + while client.stream_create(StreamType::BiDi).is_ok() { + // Exhaust the stream limit. + } + // Write on the one stream and send that out. + _ = client.stream_send(stream_id, REQUEST).unwrap(); + client.stream_close_send(stream_id).unwrap(); + let dgram = client.process(None, now()).dgram(); + + // Now handle the stream and send an incomplete response. + server.process_input(&dgram.unwrap(), now()); + server.stream_send(stream_id, RESPONSE).unwrap(); + let dgram = server.process_output(now()).dgram(); + + // The server shouldn't have released more stream credit. + client.process_input(&dgram.unwrap(), now()); + let e = client.stream_create(StreamType::BiDi).unwrap_err(); + assert!(matches!(e, Error::StreamLimitError)); + + // Closing the stream isn't enough. + server.stream_close_send(stream_id).unwrap(); + let dgram = server.process_output(now()).dgram(); + client.process_input(&dgram.unwrap(), now()); + assert!(client.stream_create(StreamType::BiDi).is_err()); + + // The server needs to see an acknowledgment from the client for its + // response AND the server has to read all of the request. + // and the server needs to read all the data. Read first. + let mut buf = [0; REQUEST.len()]; + let (count, fin) = server.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(&buf[..count], REQUEST); + assert!(fin); + + // We need an ACK from the client now, but that isn't guaranteed, + // so give the client one more packet just in case. + let dgram = send_something(&mut server, now()); + client.process_input(&dgram, now()); + + // Now get the client to send the ACK and have the server handle that. + let dgram = send_something(&mut client, now()); + let dgram = server.process(Some(&dgram), now()).dgram(); + client.process_input(&dgram.unwrap(), now()); + assert!(client.stream_create(StreamType::BiDi).is_ok()); + assert!(client.stream_create(StreamType::BiDi).is_err()); +} + +#[test] +fn no_dupdata_readable_events() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + // create a stream + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + client.stream_send(stream_id, &[0x00]).unwrap(); + let out = client.process(None, now()); + mem::drop(server.process(out.as_dgram_ref(), now())); + + // We have a data_readable event. + let stream_readable = |e| matches!(e, ConnectionEvent::RecvStreamReadable { .. }); + assert!(server.events().any(stream_readable)); + + // Send one more data frame from client. The previous stream data has not been read yet, + // therefore there should not be a new DataReadable event. + client.stream_send(stream_id, &[0x00]).unwrap(); + let out_second_data_frame = client.process(None, now()); + mem::drop(server.process(out_second_data_frame.as_dgram_ref(), now())); + assert!(!server.events().any(stream_readable)); + + // One more frame with a fin will not produce a new DataReadable event, because the + // previous stream data has not been read yet. + client.stream_send(stream_id, &[0x00]).unwrap(); + client.stream_close_send(stream_id).unwrap(); + let out_third_data_frame = client.process(None, now()); + mem::drop(server.process(out_third_data_frame.as_dgram_ref(), now())); + assert!(!server.events().any(stream_readable)); +} + +#[test] +fn no_dupdata_readable_events_empty_last_frame() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + // create a stream + let stream_id = client.stream_create(StreamType::BiDi).unwrap(); + client.stream_send(stream_id, &[0x00]).unwrap(); + let out = client.process(None, now()); + mem::drop(server.process(out.as_dgram_ref(), now())); + + // We have a data_readable event. + let stream_readable = |e| matches!(e, ConnectionEvent::RecvStreamReadable { .. }); + assert!(server.events().any(stream_readable)); + + // An empty frame with a fin will not produce a new DataReadable event, because + // the previous stream data has not been read yet. + client.stream_close_send(stream_id).unwrap(); + let out_second_data_frame = client.process(None, now()); + mem::drop(server.process(out_second_data_frame.as_dgram_ref(), now())); + assert!(!server.events().any(stream_readable)); +} + +fn change_flow_control(stream_type: StreamType, new_fc: u64) { + const RECV_BUFFER_START: u64 = 300; + + let mut client = new_client( + ConnectionParameters::default() + .max_stream_data(StreamType::BiDi, true, RECV_BUFFER_START) + .max_stream_data(StreamType::UniDi, true, RECV_BUFFER_START), + ); + let mut server = default_server(); + connect(&mut client, &mut server); + + // create a stream + let stream_id = server.stream_create(stream_type).unwrap(); + let written1 = server.stream_send(stream_id, &[0x0; 10000]).unwrap(); + assert_eq!(u64::try_from(written1).unwrap(), RECV_BUFFER_START); + + // Send the stream to the client. + let out = server.process(None, now()); + mem::drop(client.process(out.as_dgram_ref(), now())); + + // change max_stream_data for stream_id. + client.set_stream_max_data(stream_id, new_fc).unwrap(); + + // server should receive a MAX_SREAM_DATA frame if the flow control window is updated. + let out2 = client.process(None, now()); + let out3 = server.process(out2.as_dgram_ref(), now()); + let expected = usize::from(RECV_BUFFER_START < new_fc); + assert_eq!(server.stats().frame_rx.max_stream_data, expected); + + // If the flow control window has been increased, server can write more data. + let written2 = server.stream_send(stream_id, &[0x0; 10000]).unwrap(); + if RECV_BUFFER_START < new_fc { + assert_eq!(u64::try_from(written2).unwrap(), new_fc - RECV_BUFFER_START); + } else { + assert_eq!(written2, 0); + } + + // Exchange packets so that client gets all data. + let out4 = client.process(out3.as_dgram_ref(), now()); + let out5 = server.process(out4.as_dgram_ref(), now()); + mem::drop(client.process(out5.as_dgram_ref(), now())); + + // read all data by client + let mut buf = [0x0; 10000]; + let (read, _) = client.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(u64::try_from(read).unwrap(), max(RECV_BUFFER_START, new_fc)); + + let out4 = client.process(None, now()); + mem::drop(server.process(out4.as_dgram_ref(), now())); + + let written3 = server.stream_send(stream_id, &[0x0; 10000]).unwrap(); + assert_eq!(u64::try_from(written3).unwrap(), new_fc); +} + +#[test] +fn increase_decrease_flow_control() { + const RECV_BUFFER_NEW_BIGGER: u64 = 400; + const RECV_BUFFER_NEW_SMALLER: u64 = 200; + + change_flow_control(StreamType::UniDi, RECV_BUFFER_NEW_BIGGER); + change_flow_control(StreamType::BiDi, RECV_BUFFER_NEW_BIGGER); + + change_flow_control(StreamType::UniDi, RECV_BUFFER_NEW_SMALLER); + change_flow_control(StreamType::BiDi, RECV_BUFFER_NEW_SMALLER); +} + +#[test] +fn session_flow_control_stop_sending_state_recv() { + const SMALL_MAX_DATA: usize = 1024; + + let mut client = default_client(); + let mut server = new_server( + ConnectionParameters::default().max_data(u64::try_from(SMALL_MAX_DATA).unwrap()), + ); + + connect(&mut client, &mut server); + + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SMALL_MAX_DATA + ); + + // send 1 byte so that the server learns about the stream. + assert_eq!(client.stream_send(stream_id, b"a").unwrap(), 1); + + exchange_data(&mut client, &mut server); + + server + .stream_stop_sending(stream_id, Error::NoError.code()) + .unwrap(); + + assert_eq!( + client + .stream_send(stream_id, &[b'a'; SMALL_MAX_DATA]) + .unwrap(), + SMALL_MAX_DATA - 1 + ); + + // In this case the final size is only known after RESET frame is received. + // The server sends STOP_SENDING -> the client sends RESET -> the server + // sends MAX_DATA. + let out = server.process(None, now()).dgram(); + let out = client.process(out.as_ref(), now()).dgram(); + // the client is still limited. + let stream_id2 = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(client.stream_avail_send_space(stream_id2).unwrap(), 0); + let out = server.process(out.as_ref(), now()).dgram(); + client.process_input(&out.unwrap(), now()); + assert_eq!( + client.stream_avail_send_space(stream_id2).unwrap(), + SMALL_MAX_DATA + ); +} + +#[test] +fn session_flow_control_stop_sending_state_size_known() { + const SMALL_MAX_DATA: usize = 1024; + + let mut client = default_client(); + let mut server = new_server( + ConnectionParameters::default().max_data(u64::try_from(SMALL_MAX_DATA).unwrap()), + ); + + connect(&mut client, &mut server); + + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SMALL_MAX_DATA + ); + + // send 1 byte so that the server learns about the stream. + assert_eq!( + client + .stream_send(stream_id, &[b'a'; SMALL_MAX_DATA + 1]) + .unwrap(), + SMALL_MAX_DATA + ); + + let out1 = client.process(None, now()).dgram(); + // Delay this packet and let the server receive fin first (it will enter SizeKnown state). + client.stream_close_send(stream_id).unwrap(); + let out2 = client.process(None, now()).dgram(); + + server.process_input(&out2.unwrap(), now()); + + server + .stream_stop_sending(stream_id, Error::NoError.code()) + .unwrap(); + + // In this case the final size is known when stream_stop_sending is called + // and the server releases flow control immediately and sends STOP_SENDING and + // MAX_DATA in the same packet. + let out = server.process(out1.as_ref(), now()).dgram(); + client.process_input(&out.unwrap(), now()); + + // The flow control should have been updated and the client can again send + // SMALL_MAX_DATA. + let stream_id2 = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!( + client.stream_avail_send_space(stream_id2).unwrap(), + SMALL_MAX_DATA + ); +} + +#[test] +fn session_flow_control_stop_sending_state_data_recvd() { + const SMALL_MAX_DATA: usize = 1024; + + let mut client = default_client(); + let mut server = new_server( + ConnectionParameters::default().max_data(u64::try_from(SMALL_MAX_DATA).unwrap()), + ); + + connect(&mut client, &mut server); + + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SMALL_MAX_DATA + ); + + // send 1 byte so that the server learns about the stream. + assert_eq!( + client + .stream_send(stream_id, &[b'a'; SMALL_MAX_DATA + 1]) + .unwrap(), + SMALL_MAX_DATA + ); + + client.stream_close_send(stream_id).unwrap(); + + exchange_data(&mut client, &mut server); + + // The stream is DataRecvd state + server + .stream_stop_sending(stream_id, Error::NoError.code()) + .unwrap(); + + exchange_data(&mut client, &mut server); + + // The flow control should have been updated and the client can again send + // SMALL_MAX_DATA. + let stream_id2 = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!( + client.stream_avail_send_space(stream_id2).unwrap(), + SMALL_MAX_DATA + ); +} + +#[test] +fn session_flow_control_affects_all_streams() { + const SMALL_MAX_DATA: usize = 1024; + + let mut client = default_client(); + let mut server = new_server( + ConnectionParameters::default().max_data(u64::try_from(SMALL_MAX_DATA).unwrap()), + ); + + connect(&mut client, &mut server); + + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SMALL_MAX_DATA + ); + + let stream_id2 = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!( + client.stream_avail_send_space(stream_id2).unwrap(), + SMALL_MAX_DATA + ); + + assert_eq!( + client + .stream_send(stream_id, &[b'a'; SMALL_MAX_DATA / 2 + 1]) + .unwrap(), + SMALL_MAX_DATA / 2 + 1 + ); + + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SMALL_MAX_DATA / 2 - 1 + ); + assert_eq!( + client.stream_avail_send_space(stream_id2).unwrap(), + SMALL_MAX_DATA / 2 - 1 + ); + + exchange_data(&mut client, &mut server); + + let mut buf = [0x0; SMALL_MAX_DATA]; + let (read, _) = server.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(read, SMALL_MAX_DATA / 2 + 1); + + exchange_data(&mut client, &mut server); + + assert_eq!( + client.stream_avail_send_space(stream_id).unwrap(), + SMALL_MAX_DATA + ); + + assert_eq!( + client.stream_avail_send_space(stream_id2).unwrap(), + SMALL_MAX_DATA + ); +} + +fn connect_w_different_limit(bidi_limit: u64, unidi_limit: u64) { + let mut client = default_client(); + let out = client.process(None, now()); + let mut server = new_server( + ConnectionParameters::default() + .max_streams(StreamType::BiDi, bidi_limit) + .max_streams(StreamType::UniDi, unidi_limit), + ); + let out = server.process(out.as_dgram_ref(), now()); + + let out = client.process(out.as_dgram_ref(), now()); + mem::drop(server.process(out.as_dgram_ref(), now())); + + assert!(maybe_authenticate(&mut client)); + + let mut bidi_events = 0; + let mut unidi_events = 0; + let mut connected_events = 0; + for e in client.events() { + match e { + ConnectionEvent::SendStreamCreatable { stream_type } => { + if stream_type == StreamType::BiDi { + bidi_events += 1; + } else { + unidi_events += 1; + } + } + ConnectionEvent::StateChange(State::Connected) => { + connected_events += 1; + } + _ => {} + } + } + assert_eq!(bidi_events, usize::from(bidi_limit > 0)); + assert_eq!(unidi_events, usize::from(unidi_limit > 0)); + assert_eq!(connected_events, 1); +} + +#[test] +fn client_stream_creatable_event() { + connect_w_different_limit(0, 0); + connect_w_different_limit(0, 1); + connect_w_different_limit(1, 0); + connect_w_different_limit(1, 1); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/vn.rs b/third_party/rust/neqo-transport/src/connection/tests/vn.rs new file mode 100644 index 0000000000..22f15c991c --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/vn.rs @@ -0,0 +1,482 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{mem, time::Duration}; + +use neqo_common::{event::Provider, Decoder, Encoder}; +use test_fixture::{self, assertions, datagram, now}; + +use super::{ + super::{ConnectionError, ConnectionEvent, Output, State, ZeroRttState}, + connect, connect_fail, default_client, default_server, exchange_ticket, new_client, new_server, + send_something, +}; +use crate::{ + packet::PACKET_BIT_LONG, + tparams::{self, TransportParameter}, + ConnectionParameters, Error, Version, +}; + +// The expected PTO duration after the first Initial is sent. +const INITIAL_PTO: Duration = Duration::from_millis(300); + +#[test] +fn unknown_version() { + let mut client = default_client(); + // Start the handshake. + mem::drop(client.process(None, now()).dgram()); + + let mut unknown_version_packet = vec![0x80, 0x1a, 0x1a, 0x1a, 0x1a]; + unknown_version_packet.resize(1200, 0x0); + mem::drop(client.process(Some(&datagram(unknown_version_packet)), now())); + assert_eq!(1, client.stats().dropped_rx); +} + +#[test] +fn server_receive_unknown_first_packet() { + let mut server = default_server(); + + let mut unknown_version_packet = vec![0x80, 0x1a, 0x1a, 0x1a, 0x1a]; + unknown_version_packet.resize(1200, 0x0); + + assert_eq!( + server.process(Some(&datagram(unknown_version_packet,)), now(),), + Output::None + ); + + assert_eq!(1, server.stats().dropped_rx); +} + +fn create_vn(initial_pkt: &[u8], versions: &[u32]) -> Vec<u8> { + let mut dec = Decoder::from(&initial_pkt[5..]); // Skip past version. + let dst_cid = dec.decode_vec(1).expect("client DCID"); + let src_cid = dec.decode_vec(1).expect("client SCID"); + + let mut encoder = Encoder::default(); + encoder.encode_byte(PACKET_BIT_LONG); + encoder.encode(&[0; 4]); // Zero version == VN. + encoder.encode_vec(1, src_cid); + encoder.encode_vec(1, dst_cid); + + for v in versions { + encoder.encode_uint(4, *v); + } + encoder.into() +} + +#[test] +fn version_negotiation_current_version() { + let mut client = default_client(); + // Start the handshake. + let initial_pkt = client + .process(None, now()) + .dgram() + .expect("a datagram") + .to_vec(); + + let vn = create_vn( + &initial_pkt, + &[0x1a1a_1a1a, Version::default().wire_version()], + ); + + let dgram = datagram(vn); + let delay = client.process(Some(&dgram), now()).callback(); + assert_eq!(delay, INITIAL_PTO); + assert_eq!(*client.state(), State::WaitInitial); + assert_eq!(1, client.stats().dropped_rx); +} + +#[test] +fn version_negotiation_version0() { + let mut client = default_client(); + // Start the handshake. + let initial_pkt = client + .process(None, now()) + .dgram() + .expect("a datagram") + .to_vec(); + + let vn = create_vn(&initial_pkt, &[0, 0x1a1a_1a1a]); + + let dgram = datagram(vn); + let delay = client.process(Some(&dgram), now()).callback(); + assert_eq!(delay, INITIAL_PTO); + assert_eq!(*client.state(), State::WaitInitial); + assert_eq!(1, client.stats().dropped_rx); +} + +#[test] +fn version_negotiation_only_reserved() { + let mut client = default_client(); + // Start the handshake. + let initial_pkt = client + .process(None, now()) + .dgram() + .expect("a datagram") + .to_vec(); + + let vn = create_vn(&initial_pkt, &[0x1a1a_1a1a, 0x2a2a_2a2a]); + + let dgram = datagram(vn); + assert_eq!(client.process(Some(&dgram), now()), Output::None); + match client.state() { + State::Closed(err) => { + assert_eq!(*err, ConnectionError::Transport(Error::VersionNegotiation)); + } + _ => panic!("Invalid client state"), + } +} + +#[test] +fn version_negotiation_corrupted() { + let mut client = default_client(); + // Start the handshake. + let initial_pkt = client + .process(None, now()) + .dgram() + .expect("a datagram") + .to_vec(); + + let vn = create_vn(&initial_pkt, &[0x1a1a_1a1a, 0x2a2a_2a2a]); + + let dgram = datagram(vn[..vn.len() - 1].to_vec()); + let delay = client.process(Some(&dgram), now()).callback(); + assert_eq!(delay, INITIAL_PTO); + assert_eq!(*client.state(), State::WaitInitial); + assert_eq!(1, client.stats().dropped_rx); +} + +#[test] +fn version_negotiation_empty() { + let mut client = default_client(); + // Start the handshake. + let initial_pkt = client + .process(None, now()) + .dgram() + .expect("a datagram") + .to_vec(); + + let vn = create_vn(&initial_pkt, &[]); + + let dgram = datagram(vn); + let delay = client.process(Some(&dgram), now()).callback(); + assert_eq!(delay, INITIAL_PTO); + assert_eq!(*client.state(), State::WaitInitial); + assert_eq!(1, client.stats().dropped_rx); +} + +#[test] +fn version_negotiation_not_supported() { + let mut client = default_client(); + // Start the handshake. + let initial_pkt = client + .process(None, now()) + .dgram() + .expect("a datagram") + .to_vec(); + + let vn = create_vn(&initial_pkt, &[0x1a1a_1a1a, 0x2a2a_2a2a, 0xff00_0001]); + let dgram = datagram(vn); + assert_eq!(client.process(Some(&dgram), now()), Output::None); + match client.state() { + State::Closed(err) => { + assert_eq!(*err, ConnectionError::Transport(Error::VersionNegotiation)); + } + _ => panic!("Invalid client state"), + } +} + +#[test] +fn version_negotiation_bad_cid() { + let mut client = default_client(); + // Start the handshake. + let mut initial_pkt = client + .process(None, now()) + .dgram() + .expect("a datagram") + .to_vec(); + + initial_pkt[6] ^= 0xc4; + let vn = create_vn(&initial_pkt, &[0x1a1a_1a1a, 0x2a2a_2a2a, 0xff00_0001]); + + let dgram = datagram(vn); + let delay = client.process(Some(&dgram), now()).callback(); + assert_eq!(delay, INITIAL_PTO); + assert_eq!(*client.state(), State::WaitInitial); + assert_eq!(1, client.stats().dropped_rx); +} + +#[test] +fn compatible_upgrade() { + let mut client = default_client(); + let mut server = default_server(); + + connect(&mut client, &mut server); + assert_eq!(client.version(), Version::Version2); + assert_eq!(server.version(), Version::Version2); +} + +/// When the first packet from the client is gigantic, the server might generate acknowledgment +/// packets in version 1. Both client and server need to handle that gracefully. +#[test] +fn compatible_upgrade_large_initial() { + let params = ConnectionParameters::default().versions( + Version::Version1, + vec![Version::Version2, Version::Version1], + ); + let mut client = new_client(params.clone()); + client + .set_local_tparam( + 0x0845_de37_00ac_a5f9, + TransportParameter::Bytes(vec![0; 2048]), + ) + .unwrap(); + let mut server = new_server(params); + + // Client Initial should take 2 packets. + // Each should elicit a Version 1 ACK from the server. + let dgram = client.process_output(now()).dgram(); + assert!(dgram.is_some()); + let dgram = server.process(dgram.as_ref(), now()).dgram(); + assert!(dgram.is_some()); + // The following uses the Version from *outside* this crate. + assertions::assert_version(dgram.as_ref().unwrap(), Version::Version1.wire_version()); + client.process_input(&dgram.unwrap(), now()); + + connect(&mut client, &mut server); + assert_eq!(client.version(), Version::Version2); + assert_eq!(server.version(), Version::Version2); + // Only handshake padding is "dropped". + assert_eq!(client.stats().dropped_rx, 1); + assert_eq!(server.stats().dropped_rx, 1); +} + +/// A server that supports versions 1 and 2 might prefer version 1 and that's OK. +/// This one starts with version 1 and stays there. +#[test] +fn compatible_no_upgrade() { + let mut client = new_client(ConnectionParameters::default().versions( + Version::Version1, + vec![Version::Version2, Version::Version1], + )); + let mut server = new_server(ConnectionParameters::default().versions( + Version::Version1, + vec![Version::Version1, Version::Version2], + )); + + connect(&mut client, &mut server); + assert_eq!(client.version(), Version::Version1); + assert_eq!(server.version(), Version::Version1); +} + +/// A server that supports versions 1 and 2 might prefer version 1 and that's OK. +/// This one starts with version 2 and downgrades to version 1. +#[test] +fn compatible_downgrade() { + let mut client = new_client(ConnectionParameters::default().versions( + Version::Version2, + vec![Version::Version2, Version::Version1], + )); + let mut server = new_server(ConnectionParameters::default().versions( + Version::Version2, + vec![Version::Version1, Version::Version2], + )); + + connect(&mut client, &mut server); + assert_eq!(client.version(), Version::Version1); + assert_eq!(server.version(), Version::Version1); +} + +/// Inject a Version Negotiation packet, which the client detects when it validates the +/// server `version_negotiation` transport parameter. +#[test] +fn version_negotiation_downgrade() { + const DOWNGRADE: Version = Version::Draft29; + + let mut client = default_client(); + // The server sets the current version in the transport parameter and + // protects Initial packets with the version in its configuration. + // When a server `Connection` is created by a `Server`, the configuration is set + // to match the version of the packet it first receives. This replicates that. + let mut server = + new_server(ConnectionParameters::default().versions(DOWNGRADE, Version::all())); + + // Start the handshake and spoof a VN packet. + let initial = client.process_output(now()).dgram().unwrap(); + let vn = create_vn(&initial, &[DOWNGRADE.wire_version()]); + let dgram = datagram(vn); + client.process_input(&dgram, now()); + + connect_fail( + &mut client, + &mut server, + Error::VersionNegotiation, + Error::PeerError(Error::VersionNegotiation.code()), + ); +} + +/// A server connection needs to be configured with the version that the client attempts. +/// Otherwise, it will object to the client transport parameters and not do anything. +#[test] +fn invalid_server_version() { + let mut client = + new_client(ConnectionParameters::default().versions(Version::Version1, Version::all())); + let mut server = + new_server(ConnectionParameters::default().versions(Version::Version2, Version::all())); + + let dgram = client.process_output(now()).dgram(); + server.process_input(&dgram.unwrap(), now()); + + // One packet received. + assert_eq!(server.stats().packets_rx, 1); + // None dropped; the server will have decrypted it successfully. + assert_eq!(server.stats().dropped_rx, 0); + assert_eq!(server.stats().saved_datagrams, 0); + // The server effectively hasn't reacted here. + match server.state() { + State::Closed(err) => { + assert_eq!(*err, ConnectionError::Transport(Error::CryptoAlert(47))); + } + _ => panic!("invalid server state"), + } +} + +#[test] +fn invalid_current_version_client() { + const OTHER_VERSION: Version = Version::Draft29; + + let mut client = default_client(); + let mut server = default_server(); + + assert_ne!(OTHER_VERSION, client.version()); + client + .set_local_tparam( + tparams::VERSION_INFORMATION, + TransportParameter::Versions { + current: OTHER_VERSION.wire_version(), + other: Version::all() + .iter() + .copied() + .map(Version::wire_version) + .collect(), + }, + ) + .unwrap(); + + connect_fail( + &mut client, + &mut server, + Error::PeerError(Error::CryptoAlert(47).code()), + Error::CryptoAlert(47), + ); +} + +/// To test this, we need to disable compatible upgrade so that the server doesn't update +/// its transport parameters. Then, we can overwrite its transport parameters without +/// them being overwritten. Otherwise, it would be hard to find a window during which +/// the transport parameter can be modified. +#[test] +fn invalid_current_version_server() { + const OTHER_VERSION: Version = Version::Draft29; + + let mut client = default_client(); + let mut server = new_server( + ConnectionParameters::default().versions(Version::default(), vec![Version::default()]), + ); + + assert!(!Version::default().is_compatible(OTHER_VERSION)); + server + .set_local_tparam( + tparams::VERSION_INFORMATION, + TransportParameter::Versions { + current: OTHER_VERSION.wire_version(), + other: vec![OTHER_VERSION.wire_version()], + }, + ) + .unwrap(); + + connect_fail( + &mut client, + &mut server, + Error::CryptoAlert(47), + Error::PeerError(Error::CryptoAlert(47).code()), + ); +} + +#[test] +fn no_compatible_version() { + const OTHER_VERSION: Version = Version::Draft29; + + let mut client = default_client(); + let mut server = default_server(); + + assert_ne!(OTHER_VERSION, client.version()); + client + .set_local_tparam( + tparams::VERSION_INFORMATION, + TransportParameter::Versions { + current: Version::default().wire_version(), + other: vec![OTHER_VERSION.wire_version()], + }, + ) + .unwrap(); + + connect_fail( + &mut client, + &mut server, + Error::PeerError(Error::CryptoAlert(47).code()), + Error::CryptoAlert(47), + ); +} + +/// When a compatible upgrade chooses a different version, 0-RTT is rejected. +#[test] +fn compatible_upgrade_0rtt_rejected() { + // This is the baseline configuration where v1 is attempted and v2 preferred. + let prefer_v2 = ConnectionParameters::default().versions( + Version::Version1, + vec![Version::Version2, Version::Version1], + ); + let mut client = new_client(prefer_v2.clone()); + // The server will start with this so that the client resumes with v1. + let just_v1 = + ConnectionParameters::default().versions(Version::Version1, vec![Version::Version1]); + let mut server = new_server(just_v1); + + connect(&mut client, &mut server); + assert_eq!(client.version(), Version::Version1); + let token = exchange_ticket(&mut client, &mut server, now()); + + // Now upgrade the server to the preferred configuration. + let mut client = new_client(prefer_v2.clone()); + let mut server = new_server(prefer_v2); + client.enable_resumption(now(), token).unwrap(); + + // Create a packet with 0-RTT from the client. + let initial = send_something(&mut client, now()); + assertions::assert_version(&initial, Version::Version1.wire_version()); + assertions::assert_coalesced_0rtt(&initial); + server.process_input(&initial, now()); + assert!(!server + .events() + .any(|e| matches!(e, ConnectionEvent::NewStream { .. }))); + + // Finalize the connection. Don't use connect() because it uses + // maybe_authenticate() too liberally and that eats the events we want to check. + let dgram = server.process_output(now()).dgram(); // ServerHello flight + let dgram = client.process(dgram.as_ref(), now()).dgram(); // Client Finished (note: no authentication) + let dgram = server.process(dgram.as_ref(), now()).dgram(); // HANDSHAKE_DONE + client.process_input(&dgram.unwrap(), now()); + + assert!(matches!(client.state(), State::Confirmed)); + assert!(matches!(server.state(), State::Confirmed)); + + assert!(client.events().any(|e| { + println!(" client event: {e:?}"); + matches!(e, ConnectionEvent::ZeroRttRejected) + })); + assert_eq!(client.zero_rtt_state(), ZeroRttState::Rejected); +} diff --git a/third_party/rust/neqo-transport/src/connection/tests/zerortt.rs b/third_party/rust/neqo-transport/src/connection/tests/zerortt.rs new file mode 100644 index 0000000000..0aa5573c98 --- /dev/null +++ b/third_party/rust/neqo-transport/src/connection/tests/zerortt.rs @@ -0,0 +1,257 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{cell::RefCell, rc::Rc}; + +use neqo_common::event::Provider; +use neqo_crypto::{AllowZeroRtt, AntiReplay}; +use test_fixture::{self, assertions, now}; + +use super::{ + super::Connection, connect, default_client, default_server, exchange_ticket, new_server, + resumed_server, CountingConnectionIdGenerator, +}; +use crate::{events::ConnectionEvent, ConnectionParameters, Error, StreamType, Version}; + +#[test] +fn zero_rtt_negotiate() { + // Note that the two servers in this test will get different anti-replay filters. + // That's OK because we aren't testing anti-replay. + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let token = exchange_ticket(&mut client, &mut server, now()); + let mut client = default_client(); + client + .enable_resumption(now(), token) + .expect("should set token"); + let mut server = resumed_server(&client); + connect(&mut client, &mut server); + assert!(client.tls_info().unwrap().early_data_accepted()); + assert!(server.tls_info().unwrap().early_data_accepted()); +} + +#[test] +fn zero_rtt_send_recv() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let token = exchange_ticket(&mut client, &mut server, now()); + let mut client = default_client(); + client + .enable_resumption(now(), token) + .expect("should set token"); + let mut server = resumed_server(&client); + + // Send ClientHello. + let client_hs = client.process(None, now()); + assert!(client_hs.as_dgram_ref().is_some()); + + // Now send a 0-RTT packet. + let client_stream_id = client.stream_create(StreamType::UniDi).unwrap(); + client.stream_send(client_stream_id, &[1, 2, 3]).unwrap(); + let client_0rtt = client.process(None, now()); + assert!(client_0rtt.as_dgram_ref().is_some()); + // 0-RTT packets on their own shouldn't be padded to 1200. + assert!(client_0rtt.as_dgram_ref().unwrap().len() < 1200); + + let server_hs = server.process(client_hs.as_dgram_ref(), now()); + assert!(server_hs.as_dgram_ref().is_some()); // ServerHello, etc... + + let all_frames = server.stats().frame_tx.all; + let ack_frames = server.stats().frame_tx.ack; + let server_process_0rtt = server.process(client_0rtt.as_dgram_ref(), now()); + assert!(server_process_0rtt.as_dgram_ref().is_some()); + assert_eq!(server.stats().frame_tx.all, all_frames + 1); + assert_eq!(server.stats().frame_tx.ack, ack_frames + 1); + + let server_stream_id = server + .events() + .find_map(|evt| match evt { + ConnectionEvent::NewStream { stream_id, .. } => Some(stream_id), + _ => None, + }) + .expect("should have received a new stream event"); + assert_eq!(client_stream_id, server_stream_id.as_u64()); +} + +#[test] +fn zero_rtt_send_coalesce() { + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let token = exchange_ticket(&mut client, &mut server, now()); + let mut client = default_client(); + client + .enable_resumption(now(), token) + .expect("should set token"); + let mut server = resumed_server(&client); + + // Write 0-RTT before generating any packets. + // This should result in a datagram that coalesces Initial and 0-RTT. + let client_stream_id = client.stream_create(StreamType::UniDi).unwrap(); + client.stream_send(client_stream_id, &[1, 2, 3]).unwrap(); + let client_0rtt = client.process(None, now()); + assert!(client_0rtt.as_dgram_ref().is_some()); + + assertions::assert_coalesced_0rtt(&client_0rtt.as_dgram_ref().unwrap()[..]); + + let server_hs = server.process(client_0rtt.as_dgram_ref(), now()); + assert!(server_hs.as_dgram_ref().is_some()); // Should produce ServerHello etc... + + let server_stream_id = server + .events() + .find_map(|evt| match evt { + ConnectionEvent::NewStream { stream_id } => Some(stream_id), + _ => None, + }) + .expect("should have received a new stream event"); + assert_eq!(client_stream_id, server_stream_id.as_u64()); +} + +#[test] +fn zero_rtt_before_resumption_token() { + let mut client = default_client(); + assert!(client.stream_create(StreamType::BiDi).is_err()); +} + +#[test] +fn zero_rtt_send_reject() { + const MESSAGE: &[u8] = &[1, 2, 3]; + + let mut client = default_client(); + let mut server = default_server(); + connect(&mut client, &mut server); + + let token = exchange_ticket(&mut client, &mut server, now()); + let mut client = default_client(); + client + .enable_resumption(now(), token) + .expect("should set token"); + let mut server = Connection::new_server( + test_fixture::DEFAULT_KEYS, + test_fixture::DEFAULT_ALPN, + Rc::new(RefCell::new(CountingConnectionIdGenerator::default())), + ConnectionParameters::default().versions(client.version(), Version::all()), + ) + .unwrap(); + // Using a freshly initialized anti-replay context + // should result in the server rejecting 0-RTT. + let ar = + AntiReplay::new(now(), test_fixture::ANTI_REPLAY_WINDOW, 1, 3).expect("setup anti-replay"); + server + .server_enable_0rtt(&ar, AllowZeroRtt {}) + .expect("enable 0-RTT"); + + // Send ClientHello. + let client_hs = client.process(None, now()); + assert!(client_hs.as_dgram_ref().is_some()); + + // Write some data on the client. + let stream_id = client.stream_create(StreamType::UniDi).unwrap(); + client.stream_send(stream_id, MESSAGE).unwrap(); + let client_0rtt = client.process(None, now()); + assert!(client_0rtt.as_dgram_ref().is_some()); + + let server_hs = server.process(client_hs.as_dgram_ref(), now()); + assert!(server_hs.as_dgram_ref().is_some()); // Should produce ServerHello etc... + let server_ignored = server.process(client_0rtt.as_dgram_ref(), now()); + assert!(server_ignored.as_dgram_ref().is_none()); + + // The server shouldn't receive that 0-RTT data. + let recvd_stream_evt = |e| matches!(e, ConnectionEvent::NewStream { .. }); + assert!(!server.events().any(recvd_stream_evt)); + + // Client should get a rejection. + let client_fin = client.process(server_hs.as_dgram_ref(), now()); + let recvd_0rtt_reject = |e| e == ConnectionEvent::ZeroRttRejected; + assert!(client.events().any(recvd_0rtt_reject)); + + // Server consume client_fin + let server_ack = server.process(client_fin.as_dgram_ref(), now()); + assert!(server_ack.as_dgram_ref().is_some()); + let client_out = client.process(server_ack.as_dgram_ref(), now()); + assert!(client_out.as_dgram_ref().is_none()); + + // ...and the client stream should be gone. + let res = client.stream_send(stream_id, MESSAGE); + assert!(res.is_err()); + assert_eq!(res.unwrap_err(), Error::InvalidStreamId); + + // Open a new stream and send data. StreamId should start with 0. + let stream_id_after_reject = client.stream_create(StreamType::UniDi).unwrap(); + assert_eq!(stream_id, stream_id_after_reject); + client.stream_send(stream_id_after_reject, MESSAGE).unwrap(); + let client_after_reject = client.process(None, now()).dgram(); + assert!(client_after_reject.is_some()); + + // The server should receive new stream + server.process_input(&client_after_reject.unwrap(), now()); + assert!(server.events().any(recvd_stream_evt)); +} + +#[test] +fn zero_rtt_update_flow_control() { + const LOW: u64 = 3; + const HIGH: u64 = 10; + #[allow(clippy::cast_possible_truncation)] + const MESSAGE: &[u8] = &[0; HIGH as usize]; + + let mut client = default_client(); + let mut server = new_server( + ConnectionParameters::default() + .max_stream_data(StreamType::UniDi, true, LOW) + .max_stream_data(StreamType::BiDi, true, LOW), + ); + connect(&mut client, &mut server); + + let token = exchange_ticket(&mut client, &mut server, now()); + let mut client = default_client(); + client + .enable_resumption(now(), token) + .expect("should set token"); + let mut server = new_server( + ConnectionParameters::default() + .max_stream_data(StreamType::UniDi, true, HIGH) + .max_stream_data(StreamType::BiDi, true, HIGH) + .versions(client.version, Version::all()), + ); + + // Stream limits should be low for 0-RTT. + let client_hs = client.process(None, now()).dgram(); + let uni_stream = client.stream_create(StreamType::UniDi).unwrap(); + assert!(!client.stream_send_atomic(uni_stream, MESSAGE).unwrap()); + let bidi_stream = client.stream_create(StreamType::BiDi).unwrap(); + assert!(!client.stream_send_atomic(bidi_stream, MESSAGE).unwrap()); + + // Now get the server transport parameters. + let server_hs = server.process(client_hs.as_ref(), now()).dgram(); + client.process_input(&server_hs.unwrap(), now()); + + // The streams should report a writeable event. + let mut uni_stream_event = false; + let mut bidi_stream_event = false; + for e in client.events() { + if let ConnectionEvent::SendStreamWritable { stream_id } = e { + if stream_id.is_uni() { + uni_stream_event = true; + } else { + bidi_stream_event = true; + } + } + } + assert!(uni_stream_event); + assert!(bidi_stream_event); + // But no MAX_STREAM_DATA frame was received. + assert_eq!(client.stats().frame_rx.max_stream_data, 0); + + // And the new limit applies. + assert!(client.stream_send_atomic(uni_stream, MESSAGE).unwrap()); + assert!(client.stream_send_atomic(bidi_stream, MESSAGE).unwrap()); +} |