diff --git a/crates/irtt-client/src/config.rs b/crates/irtt-client/src/config.rs new file mode 100644 index 0000000..8e4f634 --- /dev/null +++ b/crates/irtt-client/src/config.rs @@ -0,0 +1,74 @@ +use std::{net::SocketAddr, time::Duration}; + +use irtt_proto::{Clock, ReceivedStats, StampAt}; + +pub(crate) const DEFAULT_PORT: u16 = 2112; +pub(crate) const DEFAULT_DURATION: Duration = Duration::from_secs(3); +pub(crate) const DEFAULT_INTERVAL: Duration = Duration::from_secs(1); +pub(crate) const DEFAULT_OPEN_TIMEOUTS: [Duration; 4] = [ + Duration::from_secs(1), + Duration::from_secs(2), + Duration::from_secs(4), + Duration::from_secs(8), +]; +pub(crate) const MIN_OPEN_TIMEOUT: Duration = Duration::from_millis(200); + +#[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, +} diff --git a/crates/irtt-client/src/error.rs b/crates/irtt-client/src/error.rs new file mode 100644 index 0000000..c769ec3 --- /dev/null +++ b/crates/irtt-client/src/error.rs @@ -0,0 +1,40 @@ +use std::{io, time::Duration}; + +use thiserror::Error; + +#[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("unexpected no-test open reply")] + UnexpectedNoTestReply, + #[error("no-test reply included a non-zero connection token: {token}")] + NonZeroNoTestToken { token: u64 }, + #[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, +} diff --git a/crates/irtt-client/src/event.rs b/crates/irtt-client/src/event.rs new file mode 100644 index 0000000..7e3c458 --- /dev/null +++ b/crates/irtt-client/src/event.rs @@ -0,0 +1,41 @@ +use std::net::SocketAddr; + +use crate::{session::NegotiatedParams, timing::ClientTimestamp}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum OpenOutcome { + Started { + remote: SocketAddr, + token: u64, + negotiated: NegotiatedParams, + event: ClientEvent, + }, + NoTestCompleted { + remote: SocketAddr, + negotiated: NegotiatedParams, + event: ClientEvent, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ClientEvent { + SessionStarted { + remote: SocketAddr, + token: u64, + negotiated: NegotiatedParams, + at: ClientTimestamp, + }, + NoTestCompleted { + remote: SocketAddr, + negotiated: NegotiatedParams, + at: ClientTimestamp, + }, + SessionClosed { + remote: SocketAddr, + token: u64, + at: ClientTimestamp, + }, + Warning { + message: String, + }, +} diff --git a/crates/irtt-client/src/session.rs b/crates/irtt-client/src/session.rs new file mode 100644 index 0000000..d7226ba --- /dev/null +++ b/crates/irtt-client/src/session.rs @@ -0,0 +1,54 @@ +use irtt_proto::{Params, PROTOCOL_VERSION}; + +use crate::{config::NegotiationPolicy, error::ClientError}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NegotiatedParams { + pub params: Params, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ClientState { + Connected, + Open { token: u64 }, + NoTestCompleted, + Closed, +} + +pub(crate) 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(), + }); + } + // TODO/OPEN: apply a documented negotiated interval safety floor once the + // clean spec or black-box report defines one. For now, only reject + // impossible non-positive intervals. + 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(()) +} diff --git a/crates/irtt-client/src/socket.rs b/crates/irtt-client/src/socket.rs new file mode 100644 index 0000000..acf23f7 --- /dev/null +++ b/crates/irtt-client/src/socket.rs @@ -0,0 +1,103 @@ +use std::{ + net::{SocketAddr, ToSocketAddrs, UdpSocket}, + time::Duration, +}; + +use socket2::{Domain, Protocol, Socket, Type}; + +use crate::{ + config::{ClientConfig, SocketConfig, DEFAULT_PORT, MIN_OPEN_TIMEOUT}, + error::ClientError, +}; + +pub(crate) 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(()) +} + +pub(crate) fn resolve_remote(config: &ClientConfig) -> Result { + let addr = normalize_server_addr(&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 }) +} + +pub(crate) fn normalize_server_addr(addr: &str) -> String { + if addr.parse::().is_ok() { + return addr.to_owned(); + } + if addr.starts_with('[') { + return addr.to_owned(); + } + if addr.parse::().is_ok() { + return format!("[{addr}]:{DEFAULT_PORT}"); + } + if addr + .rsplit_once(':') + .is_some_and(|(_, port)| port.parse::().is_ok()) + { + return addr.to_owned(); + } + format!("{addr}:{DEFAULT_PORT}") +} + +pub(crate) 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) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn normalize_server_addr_adds_default_ports() { + assert_eq!(normalize_server_addr("127.0.0.1"), "127.0.0.1:2112"); + assert_eq!(normalize_server_addr("127.0.0.1:1234"), "127.0.0.1:1234"); + assert_eq!(normalize_server_addr("localhost"), "localhost:2112"); + assert_eq!(normalize_server_addr("localhost:1234"), "localhost:1234"); + assert_eq!(normalize_server_addr("::1"), "[::1]:2112"); + assert_eq!(normalize_server_addr("[::1]:1234"), "[::1]:1234"); + } +} diff --git a/crates/irtt-client/src/timing.rs b/crates/irtt-client/src/timing.rs new file mode 100644 index 0000000..ec145e9 --- /dev/null +++ b/crates/irtt-client/src/timing.rs @@ -0,0 +1,16 @@ +use std::time::{Instant, SystemTime}; + +#[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(), + } + } +}