From 6182bc537898406b5dd64ec80b7e262bdc1f9661 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Apr 2026 16:32:48 +0700 Subject: [PATCH] refactor(irtt-client): update lib.rs for modularized client --- crates/irtt-client/src/lib.rs | 849 +--------------------------------- 1 file changed, 14 insertions(+), 835 deletions(-) diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index 77df58d..8e9250c 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -1,837 +1,16 @@ #![forbid(unsafe_code)] -use std::{ - io, - net::{SocketAddr, ToSocketAddrs, UdpSocket}, - time::{Duration, Instant, SystemTime}, -}; - -use irtt_proto::{ - close::CloseRequest, decode_open_reply, encode_close_request, encode_open_request, flags, - Clock, OpenRequest, Params, ReceivedStats, ServerFill, StampAt, PROTOCOL_VERSION, -}; -use socket2::{Domain, Protocol, Socket, Type}; -use thiserror::Error; - -const DEFAULT_PORT: u16 = 2112; -const DEFAULT_DURATION: Duration = Duration::from_secs(3); -const DEFAULT_INTERVAL: Duration = Duration::from_secs(1); -const DEFAULT_OPEN_TIMEOUTS: [Duration; 4] = [ - Duration::from_secs(1), - Duration::from_secs(2), - Duration::from_secs(4), - Duration::from_secs(8), -]; -const MIN_OPEN_TIMEOUT: Duration = Duration::from_millis(200); -const MAX_OPEN_PACKET_SIZE: usize = 512; - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ClientConfig { - pub server_addr: String, - pub duration: Option, - pub interval: Duration, - pub length: u32, - pub received_stats: ReceivedStats, - pub stamp_at: StampAt, - pub clock: Clock, - pub dscp: u8, - pub hmac_key: Option>, - pub server_fill: Option, - pub open_timeouts: Vec, - pub run_mode: RunMode, - pub negotiation_policy: NegotiationPolicy, - pub socket_config: SocketConfig, -} - -impl Default for ClientConfig { - fn default() -> Self { - Self { - server_addr: format!("127.0.0.1:{DEFAULT_PORT}"), - duration: Some(DEFAULT_DURATION), - interval: DEFAULT_INTERVAL, - length: 0, - received_stats: ReceivedStats::Both, - stamp_at: StampAt::Both, - clock: Clock::Both, - dscp: 0, - hmac_key: None, - server_fill: None, - open_timeouts: DEFAULT_OPEN_TIMEOUTS.to_vec(), - run_mode: RunMode::Normal, - negotiation_policy: NegotiationPolicy::Strict, - socket_config: SocketConfig::default(), - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Default)] -pub struct SocketConfig { - pub bind_addr: Option, - pub ttl: Option, - pub ipv4_only: bool, - pub ipv6_only: bool, - pub recv_timeout: Option, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum NegotiationPolicy { - Strict, - Loose, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum RunMode { - Normal, - NoTest, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct NegotiatedParams { - pub params: Params, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum OpenOutcome { - Started { - remote: SocketAddr, - token: u64, - negotiated: NegotiatedParams, - event: ClientEvent, - }, - NoTestCompleted { - remote: SocketAddr, - event: ClientEvent, - }, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum ClientEvent { - SessionStarted { - remote: SocketAddr, - token: u64, - negotiated: NegotiatedParams, - at: ClientTimestamp, - }, - NoTestCompleted { - remote: SocketAddr, - at: ClientTimestamp, - }, - SessionClosed { - remote: SocketAddr, - token: u64, - at: ClientTimestamp, - }, - Warning { - message: String, - }, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct ClientTimestamp { - pub wall: SystemTime, - pub mono: Instant, -} - -impl ClientTimestamp { - pub fn now() -> Self { - Self { - wall: SystemTime::now(), - mono: Instant::now(), - } - } -} - -#[derive(Debug, Error)] -pub enum ClientError { - #[error("failed to resolve server address {addr:?}")] - Resolve { addr: String }, - #[error("socket error: {0}")] - Socket(#[from] io::Error), - #[error("protocol error: {0}")] - Protocol(#[from] irtt_proto::ProtoError), - #[error("all open requests timed out")] - OpenTimeout, - #[error("open timeout {timeout:?} is below the minimum {minimum:?}")] - OpenTimeoutTooSmall { - timeout: Duration, - minimum: Duration, - }, - #[error("server rejected the open request")] - ServerRejected, - #[error("protocol version mismatch: requested {requested}, received {received}")] - ProtocolVersionMismatch { requested: i64, received: i64 }, - #[error("server returned a zero connection token")] - ZeroToken, - #[error("strict negotiation rejected changed params: {reason}")] - NegotiationRejected { reason: String }, - #[error("client is not open")] - NotOpen, - #[error("client session is already closed")] - AlreadyClosed, - #[error("operation is not implemented for this milestone")] - NotImplementedForMilestone, - #[error("duration is too large to encode as nanoseconds")] - DurationOverflow, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum ClientState { - Connected, - Open { token: u64 }, - NoTestCompleted, - Closed, -} - -#[derive(Debug)] -pub struct Client { - config: ClientConfig, - socket: UdpSocket, - remote: SocketAddr, - requested: Params, - negotiated: Option, - state: ClientState, -} - -impl Client { - pub fn connect(config: ClientConfig) -> Result { - validate_open_timeouts(&config.open_timeouts)?; - let remote = resolve_remote(&config)?; - let socket = connect_udp_socket(&config.socket_config, remote)?; - let requested = params_from_config(&config)?; - - Ok(Self { - config, - socket, - remote, - requested, - negotiated: None, - state: ClientState::Connected, - }) - } - - pub fn open(&mut self, now: ClientTimestamp) -> Result { - validate_open_timeouts(&self.config.open_timeouts)?; - let request = OpenRequest { - params: self.requested.clone(), - close: self.config.run_mode == RunMode::NoTest, - }; - let packet = encode_open_request(&request, self.config.hmac_key.as_deref())?; - let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE]; - - for timeout in &self.config.open_timeouts { - self.socket.set_read_timeout(Some(*timeout))?; - self.socket.send(&packet)?; - - match self.socket.recv(&mut buf) { - Ok(size) => { - let reply = decode_open_reply(&buf[..size], self.config.hmac_key.as_deref())?; - return self.accept_open_reply(reply, now); - } - Err(err) - if matches!( - err.kind(), - io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut - ) => {} - Err(err) => return Err(ClientError::Socket(err)), - } - } - - Err(ClientError::OpenTimeout) - } - - pub fn close(&mut self, now: ClientTimestamp) -> Result, ClientError> { - let token = match self.state { - ClientState::Open { token } => token, - ClientState::Closed => return Err(ClientError::AlreadyClosed), - ClientState::Connected | ClientState::NoTestCompleted => { - return Err(ClientError::NotOpen) - } - }; - - let packet = - encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?; - self.socket.send(&packet)?; - self.state = ClientState::Closed; - - Ok(vec![ClientEvent::SessionClosed { - remote: self.remote, - token, - at: now, - }]) - } - - fn accept_open_reply( - &mut self, - reply: irtt_proto::OpenReply, - now: ClientTimestamp, - ) -> Result { - if reply.params.protocol_version != PROTOCOL_VERSION { - return Err(ClientError::ProtocolVersionMismatch { - requested: PROTOCOL_VERSION, - received: reply.params.protocol_version, - }); - } - - let reply_is_close = flags::has(reply.flags, flags::FLAG_CLOSE); - match (self.config.run_mode, reply_is_close) { - (RunMode::Normal, true) => return Err(ClientError::ServerRejected), - (RunMode::Normal, false) if reply.token == 0 => return Err(ClientError::ZeroToken), - (RunMode::NoTest, true) => { - self.state = ClientState::NoTestCompleted; - let event = ClientEvent::NoTestCompleted { - remote: self.remote, - at: now, - }; - return Ok(OpenOutcome::NoTestCompleted { - remote: self.remote, - event, - }); - } - (RunMode::NoTest, false) if reply.token == 0 => return Err(ClientError::ZeroToken), - (RunMode::NoTest, false) => {} - (RunMode::Normal, false) => {} - } - - validate_negotiated_params( - &self.requested, - &reply.params, - self.config.negotiation_policy, - )?; - let negotiated = NegotiatedParams { - params: reply.params.clone(), - }; - self.negotiated = Some(negotiated.clone()); - self.state = ClientState::Open { token: reply.token }; - let event = ClientEvent::SessionStarted { - remote: self.remote, - token: reply.token, - negotiated: negotiated.clone(), - at: now, - }; - - Ok(OpenOutcome::Started { - remote: self.remote, - token: reply.token, - negotiated, - event, - }) - } -} - -fn validate_open_timeouts(timeouts: &[Duration]) -> Result<(), ClientError> { - for timeout in timeouts { - if *timeout < MIN_OPEN_TIMEOUT { - return Err(ClientError::OpenTimeoutTooSmall { - timeout: *timeout, - minimum: MIN_OPEN_TIMEOUT, - }); - } - } - Ok(()) -} - -fn resolve_remote(config: &ClientConfig) -> Result { - let addr = with_default_port(&config.server_addr); - let mut addrs = addr - .to_socket_addrs() - .map_err(|_| ClientError::Resolve { addr: addr.clone() })?; - addrs - .find(|addr| { - (!config.socket_config.ipv4_only || addr.is_ipv4()) - && (!config.socket_config.ipv6_only || addr.is_ipv6()) - }) - .ok_or(ClientError::Resolve { addr }) -} - -fn with_default_port(addr: &str) -> String { - if addr.parse::().is_ok() { - return addr.to_owned(); - } - if addr - .rsplit_once(':') - .is_some_and(|(_, port)| port.parse::().is_ok()) - { - return addr.to_owned(); - } - format!("{addr}:{DEFAULT_PORT}") -} - -fn connect_udp_socket(config: &SocketConfig, remote: SocketAddr) -> Result { - let domain = if remote.is_ipv4() { - Domain::IPV4 - } else { - Domain::IPV6 - }; - let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?; - - if config.ipv6_only && remote.is_ipv6() { - socket.set_only_v6(true)?; - } - if let Some(ttl) = config.ttl { - socket.set_ttl(ttl)?; - } - - let bind_addr = config.bind_addr.unwrap_or_else(|| { - if remote.is_ipv4() { - SocketAddr::from(([0, 0, 0, 0], 0)) - } else { - SocketAddr::from(([0_u16; 8], 0)) - } - }); - socket.bind(&bind_addr.into())?; - socket.connect(&remote.into())?; - - let socket: UdpSocket = socket.into(); - socket.set_read_timeout(config.recv_timeout)?; - Ok(socket) -} - -fn params_from_config(config: &ClientConfig) -> Result { - Ok(Params { - protocol_version: PROTOCOL_VERSION, - duration_ns: duration_to_ns(config.duration.unwrap_or_default())?, - interval_ns: duration_to_ns(config.interval)?, - length: i64::from(config.length), - received_stats: config.received_stats, - stamp_at: config.stamp_at, - clock: config.clock, - dscp: i64::from(config.dscp), - server_fill: config.server_fill.clone().map(|value| ServerFill { value }), - }) -} - -fn duration_to_ns(duration: Duration) -> Result { - i64::try_from(duration.as_nanos()).map_err(|_| ClientError::DurationOverflow) -} - -fn validate_negotiated_params( - requested: &Params, - returned: &Params, - policy: NegotiationPolicy, -) -> Result<(), ClientError> { - if returned.protocol_version != PROTOCOL_VERSION { - return Err(ClientError::ProtocolVersionMismatch { - requested: PROTOCOL_VERSION, - received: returned.protocol_version, - }); - } - if returned.duration_ns > requested.duration_ns { - return Err(ClientError::NegotiationRejected { - reason: "duration increased".to_owned(), - }); - } - if returned.length > requested.length { - return Err(ClientError::NegotiationRejected { - reason: "length increased".to_owned(), - }); - } - if returned.interval_ns <= 0 { - return Err(ClientError::NegotiationRejected { - reason: "interval must be positive".to_owned(), - }); - } - - if policy == NegotiationPolicy::Strict && returned != requested { - return Err(ClientError::NegotiationRejected { - reason: "returned params differ from requested params".to_owned(), - }); - } - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::*; - use irtt_proto::{ - compute_hmac_in_place, flags::FLAG_HMAC, flags::FLAG_OPEN, flags::FLAG_REPLY, verify_hmac, - HMAC_SIZE, MAGIC, - }; - use std::{ - net::UdpSocket, - sync::mpsc, - thread::{self, JoinHandle}, - }; - - const TOKEN: u64 = 0x1234_5678_90ab_cdef; - const HMAC_OFFSET: usize = 4; - - struct FakeServer { - addr: SocketAddr, - rx: mpsc::Receiver>, - done: JoinHandle<()>, - } - - impl FakeServer { - fn join(self) { - self.done.join().unwrap(); - } - } - - fn default_test_config(addr: SocketAddr) -> ClientConfig { - ClientConfig { - server_addr: addr.to_string(), - open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], - ..ClientConfig::default() - } - } - - fn start_fake_server(handler: F) -> FakeServer - where - F: FnOnce(UdpSocket, mpsc::Sender>) + Send + 'static, - { - let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); - let addr = socket.local_addr().unwrap(); - let (tx, rx) = mpsc::channel(); - let done = thread::spawn(move || handler(socket, tx)); - FakeServer { addr, rx, done } - } - - fn recv_request(socket: &UdpSocket, tx: &mpsc::Sender>) -> (Vec, SocketAddr) { - let mut buf = [0_u8; 512]; - let (size, peer) = socket.recv_from(&mut buf).unwrap(); - let packet = buf[..size].to_vec(); - tx.send(packet.clone()).unwrap(); - (packet, peer) - } - - fn open_reply(flags: u8, token: u64, params: &Params, hmac_key: Option<&[u8]>) -> Vec { - let mut packet = Vec::new(); - packet.extend_from_slice(&MAGIC); - packet.push(flags | hmac_key.map_or(0, |_| FLAG_HMAC)); - if hmac_key.is_some() { - packet.extend_from_slice(&[0_u8; HMAC_SIZE]); - } - packet.extend_from_slice(&token.to_le_bytes()); - packet.extend_from_slice(¶ms.encode()); - if let Some(key) = hmac_key { - compute_hmac_in_place(key, &mut packet, HMAC_OFFSET).unwrap(); - } - packet - } - - fn open_success_server(params: Params) -> FakeServer { - start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - }) - } - - fn assert_open_started(outcome: OpenOutcome) -> NegotiatedParams { - match outcome { - OpenOutcome::Started { - token, negotiated, .. - } => { - assert_eq!(token, TOKEN); - negotiated - } - OpenOutcome::NoTestCompleted { .. } => panic!("unexpected no-test outcome"), - } - } - - #[test] - fn client_config_default() { - let config = ClientConfig::default(); - assert_eq!(config.duration, Some(Duration::from_secs(3))); - assert_eq!(config.interval, Duration::from_secs(1)); - assert_eq!(config.length, 0); - assert_eq!(config.received_stats, ReceivedStats::Both); - assert_eq!(config.stamp_at, StampAt::Both); - assert_eq!(config.clock, Clock::Both); - assert_eq!(config.dscp, 0); - assert_eq!(config.hmac_key, None); - assert_eq!(config.server_fill, None); - assert_eq!(config.open_timeouts, DEFAULT_OPEN_TIMEOUTS); - assert_eq!(config.run_mode, RunMode::Normal); - assert_eq!(config.negotiation_policy, NegotiationPolicy::Strict); - } - - #[test] - fn address_resolution_connects_to_local_fake_server() { - let server = start_fake_server(|socket, tx| { - let _ = recv_request(&socket, &tx); - }); - let client = Client::connect(default_test_config(server.addr)).unwrap(); - client.socket.send(b"ping").unwrap(); - assert_eq!(server.rx.recv().unwrap(), b"ping"); - server.join(); - } - - #[test] - fn successful_open_handshake() { - let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - let params = params_from_config(&config).unwrap(); - let server = open_success_server(params.clone()); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - - let negotiated = assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - assert_eq!(negotiated.params, params); - assert!(matches!(client.state, ClientState::Open { token: TOKEN })); - server.join(); - } - - #[test] - fn open_retries_after_first_timeout() { - let server = start_fake_server(|socket, tx| { - let (first, _) = recv_request(&socket, &tx); - let (_, peer) = recv_request(&socket, &tx); - let params = Params::decode(&first[4..]).unwrap(); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - assert_eq!(first[3] & FLAG_OPEN, FLAG_OPEN); - socket.send_to(&reply, peer).unwrap(); - }); - let mut config = default_test_config(server.addr); - config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(500)]; - let params = params_from_config(&config).unwrap(); - let reply_params = params.clone(); - drop(reply_params); - let mut client = Client::connect(config).unwrap(); - let outcome = client.open(ClientTimestamp::now()).unwrap(); - assert_open_started(outcome); - assert_eq!(server.rx.iter().take(2).count(), 2); - server.join(); - } - - #[test] - fn open_timeout_after_all_timeouts() { - let server = start_fake_server(|socket, tx| { - socket - .set_read_timeout(Some(Duration::from_millis(700))) - .unwrap(); - while recv_request_timeout(&socket, &tx).is_some() {} - }); - let mut config = default_test_config(server.addr); - config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::OpenTimeout) - )); - assert_eq!(server.rx.iter().take(2).count(), 2); - server.join(); - } - - fn recv_request_timeout( - socket: &UdpSocket, - tx: &mpsc::Sender>, - ) -> Option<(Vec, SocketAddr)> { - let mut buf = [0_u8; 512]; - match socket.recv_from(&mut buf) { - Ok((size, peer)) => { - let packet = buf[..size].to_vec(); - tx.send(packet.clone()).unwrap(); - Some((packet, peer)) - } - Err(_) => None, - } - } - - #[test] - fn strict_negotiation_accepts_identical_params() { - let config = ClientConfig::default(); - let params = params_from_config(&config).unwrap(); - assert!(validate_negotiated_params(¶ms, ¶ms, NegotiationPolicy::Strict).is_ok()); - } - - #[test] - fn strict_negotiation_rejects_changed_params() { - let config = ClientConfig::default(); - let requested = params_from_config(&config).unwrap(); - let mut returned = requested.clone(); - returned.dscp = 1; - assert!(matches!( - validate_negotiated_params(&requested, &returned, NegotiationPolicy::Strict), - Err(ClientError::NegotiationRejected { .. }) - )); - } - - #[test] - fn loose_negotiation_accepts_server_restricted_params() { - let config = ClientConfig::default(); - let requested = params_from_config(&config).unwrap(); - let mut returned = requested.clone(); - returned.duration_ns /= 2; - returned.length = 0; - assert!( - validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose).is_ok() - ); - } - - #[test] - fn protocol_version_mismatch_fails() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.negotiation_policy = NegotiationPolicy::Loose; - let mut params = params_from_config(&config).unwrap(); - params.protocol_version = 2; - let server = open_success_server(params); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::ProtocolVersionMismatch { received: 2, .. }) - )); - server.join(); - } - - #[test] - fn server_rejection_fails_in_normal_mode() { - let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::ServerRejected) - )); - server.join(); - } - - #[test] - fn no_test_open_close_succeeds_on_open_reply_close() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (request, peer) = recv_request(&socket, &tx); - assert_eq!(request[3] & flags::FLAG_CLOSE, flags::FLAG_CLOSE); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()).unwrap(), - OpenOutcome::NoTestCompleted { .. } - )); - assert!(matches!( - client.close(ClientTimestamp::now()), - Err(ClientError::NotOpen) - )); - server.join(); - } - - #[test] - fn close_sends_one_close_packet_with_negotiated_token() { - let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - let _ = recv_request(&socket, &tx); - }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - let events = client.close(ClientTimestamp::now()).unwrap(); - assert_eq!(events.len(), 1); - let packets: Vec<_> = server.rx.iter().take(2).collect(); - let close = &packets[1]; - assert_eq!(close[3], flags::FLAG_CLOSE); - assert_eq!(u64::from_le_bytes(close[4..12].try_into().unwrap()), TOKEN); - server.join(); - } - - #[test] - fn hmac_open_success() { - let key = b"secret".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key.clone()); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (request, peer) = recv_request(&socket, &tx); - verify_hmac(&key, &request, HMAC_OFFSET).unwrap(); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&key)); - socket.send_to(&reply, peer).unwrap(); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - server.join(); - } - - #[test] - fn hmac_open_rejects_missing_hmac() { - let key = b"secret".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::Protocol( - irtt_proto::ProtoError::HmacPresenceMismatch - )) - )); - server.join(); - } - - #[test] - fn hmac_open_rejects_bad_hmac() { - let key = b"secret".to_vec(); - let wrong_key = b"wrong".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&wrong_key)); - socket.send_to(&reply, peer).unwrap(); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::Protocol(irtt_proto::ProtoError::BadHmac)) - )); - server.join(); - } - - #[test] - fn hmac_close_packet_includes_valid_hmac() { - let key = b"secret".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key.clone()); - let params = params_from_config(&config).unwrap(); - let server_key = key.clone(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&server_key)); - socket.send_to(&reply, peer).unwrap(); - let _ = recv_request(&socket, &tx); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.close(ClientTimestamp::now()).unwrap(); - let packets: Vec<_> = server.rx.iter().take(2).collect(); - let close = &packets[1]; - assert_eq!(close[3], flags::FLAG_CLOSE | FLAG_HMAC); - verify_hmac(&key, close, HMAC_OFFSET).unwrap(); - assert_eq!( - u64::from_le_bytes(close[4 + HMAC_SIZE..12 + HMAC_SIZE].try_into().unwrap()), - TOKEN - ); - server.join(); - } - - #[test] - fn minimum_open_timeout_under_200ms_is_rejected() { - let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(199)], - ..ClientConfig::default() - }; - assert!(matches!( - Client::connect(config), - Err(ClientError::OpenTimeoutTooSmall { .. }) - )); - } -} +mod client; +mod config; +mod error; +mod event; +mod session; +mod socket; +mod timing; + +pub use client::Client; +pub use config::{ClientConfig, NegotiationPolicy, RunMode, SocketConfig}; +pub use error::ClientError; +pub use event::{ClientEvent, OpenOutcome}; +pub use session::NegotiatedParams; +pub use timing::ClientTimestamp;