diff --git a/AGENTS.md b/AGENTS.md index 016a5c2..fa671e9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -57,7 +57,7 @@ Tokio must remain optional for `irtt-client`; default builds must remain runtime ### Server architecture (`irtt-server`) -`irtt-server` is a first-class reusable server crate. It currently contains the deterministic OPEN/session core — packet admission, authentication policy, open negotiation, the bounded session table, open reply construction and client-initiated close. ECHO processing, server-initiated close and session lifetime remain unimplemented, as does the Tokio runtime; see `crates/irtt-server/AGENTS.md`. The rest of this section describes the boundary the crate is built to, including the parts not yet written. +`irtt-server` is a first-class reusable server crate. It currently contains the deterministic OPEN/ECHO/session core — packet admission, authentication policy, open negotiation, the bounded session table, open reply construction, echo processing with its per-session receive state and timestamps, and client-initiated close. Rate limiting, session lifetime, server-initiated close, DSCP application and the server fill policy remain unimplemented, as does the Tokio runtime; see `crates/irtt-server/AGENTS.md`. The rest of this section describes the boundary the crate is built to, including the parts not yet written. - It shares wire semantics with the client through `irtt-proto` and follows the same clean-room boundary. - It must **not** depend on `irtt-client` merely to reuse protocol behavior. diff --git a/crates/irtt-server/AGENTS.md b/crates/irtt-server/AGENTS.md index 871b4b8..333311a 100644 --- a/crates/irtt-server/AGENTS.md +++ b/crates/irtt-server/AGENTS.md @@ -103,9 +103,50 @@ effective parameters must be safe for the server to run. - Zero is a valid maximum and is not a synonym for unlimited: it refuses every session, and no-test opens with it, since both are validated against the same effective session. -- The echo slice must reuse this same configured maximum to admit inbound echo - request datagrams by their received length, before any receive state is - mutated. +- Echo request admission reuses this same configured maximum, comparing it + against the received datagram length before any receive state is mutated. + +## Echo processing + +- An accepted echo owns per-session `ReceiveState` — count, window and the + previous accepted sequence number. Admission order is authentication, then + received datagram length against the configured maximum, then token, then + endpoint; a request failing any of them mutates nothing, in this session or + any other. +- The count and window transitions follow the clean specification's Section 10 + and the Section 1 vectors, including the parts that read as defects: a + duplicate advances the count and leaves the window alone, and a late, + reordered or far-gap request resets the window to `0x1` rather than setting a + historical bit, because the distance is unsigned. Do not replace this with a + conventional selective-acknowledgement bitmap. Reference the vectors rather + than restating the tables here. +- Only the token and sequence number mean anything in a request. The tail is + opaque, and the request's length never sizes the reply — layout and length + come from the session's negotiated params through `irtt-proto` alone. +- The transition is pure and is committed only after the reply has encoded, so + an internal encoding failure cannot leave a session claiming to have answered + a request it never did. +- Clock sampling is a private injected seam, like token generation, not a + runtime abstraction: the receive instant is taken as soon as a datagram is + classified as an echo and the send instant just before the reply is built, so + they bracket the server's own handling. The monotonic origin belongs to the + clock source and is stable for its life. +- A reply's receive instant is held back to its send instant where the wall + clock stepped backwards between the two readings, so the required ordering + holds. That correction is per reply and stateless on purpose: latching a + pre-step wall value across packets would keep reporting a time the host has + corrected away, which is the smoothing the specification forbids, and + anchoring the wall clock to the monotonic origin would drift and never pick up + a legitimate correction. The server reports its own wall clock honestly. +- For a single-clock midpoint, `irtt-rs` emits the one negotiated field. It does + not reproduce upstream 0.9.1's dual-field midpoint; `irtt-proto` still decodes + that form from a peer. +- Reply payload bytes are zero, which is the initial safe fill policy: payload + carries no protocol meaning, and a server must never emit residue from another + request or client. Request payload bytes never reach a reply. +- Rate limiting, idle expiry, maximum duration, server-initiated close, DSCP + socket application and the full `ServerFill` policy are later slices. Do not + add placeholder session fields for them. ## Resource policy @@ -130,9 +171,10 @@ half-created session. maintain shadow encoders for well-formed wire data. - Hand-build packets only where the point is a payload a compliant encoder cannot produce — truncated varints, out-of-range enums, corrupted MACs. -- Token generation is the only nondeterministic part; tests inject a scripted - source so identity, collisions and allocation failure are assertable. Keep - that seam private. +- Token generation and clock sampling are the only nondeterministic parts; tests + inject a scripted source for each, so identity, collisions, allocation failure + and timestamp values are all assertable. Keep both seams private, and keep + timestamp tests free of sleeps, tolerances and real wall-clock assertions. - As the server gains behavior, prefer it over fake peers for normal compliant behavior in other crates' tests. Keep small raw/adversarial peers for intentionally malformed or non-compliant wire behavior. diff --git a/crates/irtt-server/src/clock.rs b/crates/irtt-server/src/clock.rs new file mode 100644 index 0000000..8a5b424 --- /dev/null +++ b/crates/irtt-server/src/clock.rs @@ -0,0 +1,211 @@ +//! Clock sampling for echo timestamps. +//! +//! An echo reply reports when the server received the request and when it sent +//! the answer, so the core needs a clock. It takes one through this private +//! seam rather than reading the system clock inline, so that timestamp behavior +//! is deterministically testable without a runtime, sleeps or timing +//! tolerances. +//! +//! This is the clock counterpart of [`TokenSource`](crate::token::TokenSource) +//! and nothing more. It is **not** a runtime abstraction, not a transport +//! boundary and not a pluggable product API: it is crate-private, the public +//! constructor still takes only a [`ServerConfig`](crate::ServerConfig), and +//! callers never pass timestamps into the core. + +use std::{ + fmt, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; + +/// One instant, read from both clock domains together. +/// +/// Both fields are signed nanoseconds, matching the wire encoding: `wall_ns` +/// counts from the Unix epoch, and `mono_ns` from an origin the clock source +/// owns. The specification lets a server pick any monotonic origin as long as +/// it is stable for as long as any session it stamps may live, so `mono_ns` is +/// deliberately not process uptime or anything else externally meaningful — +/// only the difference between two samples from the same source has meaning. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct ClockSample { + pub(crate) wall_ns: i64, + pub(crate) mono_ns: i64, +} + +impl ClockSample { + /// The arithmetic mean of two samples, taken separately per clock domain. + /// + /// This is what a midpoint timestamp is: the mean of one reply's receive + /// and send instants. The two domains never mix, and the mean goes through + /// `i128` because `a + b` can overflow `i64` while the mean itself never + /// can. + pub(crate) fn midpoint(self, other: Self) -> Self { + Self { + wall_ns: mean_ns(self.wall_ns, other.wall_ns), + mono_ns: mean_ns(self.mono_ns, other.mono_ns), + } + } + + /// This sample, held back in any domain where it runs ahead of `later`. + /// + /// A reply's receive instant must not be later than its send instant. The + /// monotonic domain gets that from its source, which only moves forward. + /// The wall clock does not: it can be stepped backwards between two + /// readings by NTP, a hypervisor or an administrator, which would otherwise + /// invert the pair. + /// + /// The *earlier* reading is the one that moves, so the pair settles on the + /// clock as it now stands rather than on a value the clock has already + /// disowned. Nothing is remembered between calls: a latch that carried a + /// pre-step value forward would keep reporting a wall time the host has + /// corrected away, which is the smoothing across packets the specification + /// forbids — and it would make one-way delays wrong for as long as the + /// latch held, rather than for one reply. + pub(crate) fn not_after(self, later: Self) -> Self { + Self { + wall_ns: self.wall_ns.min(later.wall_ns), + mono_ns: self.mono_ns.min(later.mono_ns), + } + } +} + +/// A source of clock samples. +/// +/// Deliberately private, like [`TokenSource`](crate::token::TokenSource): it +/// exists so tests can script the receive and send instants of a reply, not as +/// an extension point. The production implementation is [`SystemClock`]. +/// +/// Sampling is infallible. A clock that could refuse would make an admitted +/// echo fail for a reason that is neither the peer's fault nor recoverable, so +/// the production source saturates rather than reporting an error. +pub(crate) trait ClockSource: fmt::Debug + Send { + fn sample(&mut self) -> ClockSample; +} + +/// Reads the wall clock from [`SystemTime`] and the monotonic clock from an +/// [`Instant`] captured when the source was created. +/// +/// That instant is the monotonic origin, so it is fixed for the life of the +/// source and shared by every session the core using it holds — which is +/// exactly the stability the specification asks for. +#[derive(Debug)] +pub(crate) struct SystemClock { + origin: Instant, +} + +impl SystemClock { + pub(crate) fn new() -> Self { + Self { + origin: Instant::now(), + } + } +} + +impl Default for SystemClock { + fn default() -> Self { + Self::new() + } +} + +impl ClockSource for SystemClock { + fn sample(&mut self) -> ClockSample { + ClockSample { + wall_ns: wall_ns(), + mono_ns: saturating_ns(self.origin.elapsed()), + } + } +} + +/// Nanoseconds since the Unix epoch, negative before it. +fn wall_ns() -> i64 { + match SystemTime::now().duration_since(UNIX_EPOCH) { + Ok(since_epoch) => saturating_ns(since_epoch), + Err(before_epoch) => saturating_ns(before_epoch.duration()).saturating_neg(), + } +} + +/// A duration as nanoseconds, saturating rather than panicking. +/// +/// [`Duration`] counts nanoseconds in `u128` and reaches far beyond `i64`, +/// which the wire field is. A host clock set past the year 2262 is a local +/// misconfiguration; it must not become a panic that a remote peer can reach by +/// sending an echo request. +fn saturating_ns(duration: Duration) -> i64 { + i64::try_from(duration.as_nanos()).unwrap_or(i64::MAX) +} + +/// The mean of two nanosecond values, computed without overflowing. +fn mean_ns(a: i64, b: i64) -> i64 { + let mean = (i128::from(a) + i128::from(b)) / 2; + // The mean of two `i64` values is always an `i64`, so this conversion + // cannot actually fail. Clamping rather than unwrapping keeps timestamp + // arithmetic free of any panic at all. + i64::try_from(mean).unwrap_or(if mean.is_negative() { + i64::MIN + } else { + i64::MAX + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Constructed values only: the production clock's real readings are not + /// assertable without a timing tolerance, and the behavioral echo tests use + /// a scripted source instead. + #[test] + fn a_midpoint_is_the_per_domain_mean_and_cannot_overflow() { + let receive = ClockSample { + wall_ns: 1_000, + mono_ns: 10_000, + }; + let send = ClockSample { + wall_ns: 1_200, + mono_ns: 10_300, + }; + assert_eq!( + receive.midpoint(send), + ClockSample { + wall_ns: 1_100, + mono_ns: 10_150, + } + ); + + // The sum overflows `i64`; the mean does not. + let high = ClockSample { + wall_ns: i64::MAX, + mono_ns: i64::MAX, + }; + assert_eq!( + high.midpoint(high), + ClockSample { + wall_ns: i64::MAX, + mono_ns: i64::MAX, + } + ); + let low = ClockSample { + wall_ns: i64::MIN, + mono_ns: i64::MIN, + }; + assert_eq!( + low.midpoint(low), + ClockSample { + wall_ns: i64::MIN, + mono_ns: i64::MIN, + } + ); + assert_eq!( + high.midpoint(low), + ClockSample { + wall_ns: 0, + mono_ns: 0, + } + ); + } + + #[test] + fn a_duration_beyond_the_wire_field_saturates_rather_than_panicking() { + assert_eq!(saturating_ns(Duration::from_nanos(1_500)), 1_500); + assert_eq!(saturating_ns(Duration::MAX), i64::MAX); + } +} diff --git a/crates/irtt-server/src/config.rs b/crates/irtt-server/src/config.rs index 50b689e..abd9213 100644 --- a/crates/irtt-server/src/config.rs +++ b/crates/irtt-server/src/config.rs @@ -71,8 +71,8 @@ impl ServerConfig { /// reduced to it during negotiation, and an open whose mandatory echo field /// block would not fit is silently refused, since the server could not emit /// a compliant reply for it. Authentication counts toward that field block. - /// The echo slice will additionally use this limit to admit inbound echo - /// request datagrams by their received length. + /// The same limit additionally admits inbound echo request datagrams by + /// their received length, before any receive state moves. /// /// The value is trusted local configuration, like /// [`with_max_sessions`](Self::with_max_sessions): an operator may diff --git a/crates/irtt-server/src/core.rs b/crates/irtt-server/src/core.rs index 593b761..3f5977f 100644 --- a/crates/irtt-server/src/core.rs +++ b/crates/irtt-server/src/core.rs @@ -1,11 +1,13 @@ use std::{collections::HashMap, net::SocketAddr}; use irtt_proto::{ - decode_request, echo_packet_len, encode_open_reply, verify_packet_hmac, Clock, DecodedParams, - DecodedRequestKind, OpenReply, Params, StampAt, FLAG_CLOSE, FLAG_OPEN, FLAG_REPLY, + decode_request, echo_packet_len, encode_echo_reply, encode_open_reply, verify_packet_hmac, + Clock, DecodedParams, DecodedRequestKind, EchoReply, OpenReply, Params, StampAt, + TimestampFields, FLAG_CLOSE, FLAG_OPEN, FLAG_REPLY, }; use crate::{ + clock::{ClockSample, ClockSource, SystemClock}, config::ServerConfig, error::ServerError, negotiate::negotiate_params, @@ -16,10 +18,12 @@ use crate::{ /// The deterministic protocol and session engine of the server. /// /// `ServerCore` owns admission, negotiation, the session table and reply -/// construction. It performs no I/O and keeps no clock: a caller hands it a -/// source endpoint and the bytes of one received datagram, and gets back the -/// bytes to send to that endpoint, or nothing. Everything it does is a pure -/// function of its state and that input, except for drawing session tokens. +/// construction. It performs no I/O: a caller hands it a source endpoint and +/// the bytes of one received datagram, and gets back the bytes to send to that +/// endpoint, or nothing. Everything it does is a pure function of its state and +/// that input, except for drawing session tokens and sampling the clock an echo +/// reply's timestamps come from — both of which are private injected seams, so +/// the whole engine stays deterministically testable. /// /// The Tokio runtime slice will wrap it directly: /// @@ -35,29 +39,45 @@ use crate::{ /// /// # Scope /// -/// This slice implements open handling, session creation and client-initiated -/// close. Echo requests are structurally recognized and then deliberately -/// ignored — no reply, no session mutation — until their own slice. +/// This slice implements open handling, session creation, normal echo +/// processing and client-initiated close. Rate limiting, session lifetime and +/// expiry, server-initiated close, DSCP application and the server fill policy +/// are separate slices, so every otherwise admissible echo is answered and no +/// session ever ages out. #[derive(Debug)] pub struct ServerCore { config: ServerConfig, sessions: HashMap, tokens: Box, + clock: Box, } impl ServerCore { /// Creates a server core drawing session tokens from the operating system's - /// random source. + /// random source and timestamps from the system clock. #[must_use] pub fn new(config: ServerConfig) -> Self { Self::with_token_source(config, Box::new(OsTokenSource)) } + /// A core with a chosen token source and the production clock, which is + /// what tests that care about session identity but not about timestamp + /// values want. pub(crate) fn with_token_source(config: ServerConfig, tokens: Box) -> Self { + Self::with_sources(config, tokens, Box::new(SystemClock::new())) + } + + /// A core with both nondeterministic sources chosen. + pub(crate) fn with_sources( + config: ServerConfig, + tokens: Box, + clock: Box, + ) -> Self { Self { config, sessions: HashMap::new(), tokens, + clock, } } @@ -96,17 +116,19 @@ impl ServerCore { /// flag, disagrees with this server's authentication configuration, fails /// authentication, carries a malformed or out-of-range open parameter /// payload, negotiates to parameters the server could not execute, would - /// exceed the session bound, or is a request kind this slice does not - /// implement. None of that mutates any live session. + /// exceed the session bound, is longer than the configured maximum packet + /// length, or names no live session from the endpoint that opened it. None + /// of that mutates any live session. /// /// `Ok(None)` is not only a rejection, though. A close request that does /// release its session is answered with nothing as well, because protocol /// version 1 defines no acknowledgement for one. /// - /// Authentication is checked before open parameters are decoded. That - /// ordering is deliberate: a packet that fails authentication is discarded - /// without parsing untrusted parameter data, and no reply behavior may - /// reveal which stage rejected it. + /// Authentication is checked before open parameters are decoded and before + /// any session is looked up. That ordering is deliberate: a packet that + /// fails authentication is discarded without parsing untrusted parameter + /// data or consulting the session table, and no reply behavior may reveal + /// which stage rejected it. /// /// # Errors /// @@ -122,6 +144,16 @@ impl ServerCore { let Ok(request) = decode_request(packet) else { return Ok(None); }; + // An echo's arrival instant is sampled the moment the datagram is known + // to be one, so that everything the reply's receive timestamp should + // bracket — authentication, the session lookup, the receive-state + // transition and encoding — happens after it. A datagram rejected later + // has consumed one clock sample and nothing else; sampling is not + // protocol state, and it is a strictly better trade than looking a + // session up before authentication to discover whether it even wanted + // timestamps. + let kind = self.classify(&request.kind, packet.len()); + // The HMAC flag must agree with this server's configuration in both // directions of mismatch, and the MAC itself is verified separately. if request.hmac_present != self.config.hmac_key().is_some() { @@ -133,22 +165,116 @@ impl ServerCore { } } - match request.kind { - DecodedRequestKind::Open { no_test, params } => self.handle_open(peer, no_test, params), - DecodedRequestKind::Close { token } => { + match kind { + ClassifiedKind::Open { no_test, params } => self.handle_open(peer, no_test, params), + ClassifiedKind::Close { token } => { self.handle_close(peer, token); Ok(None) } - // Not implemented in this slice. An echo must not advance receive - // state until the slice that implements it; emitting a placeholder - // reply now would be worse than silence. That slice must admit an - // echo by its received datagram length against - // `max_packet_length` before it touches any receive state; adding - // that branch now would only be dead code guarding nothing. - DecodedRequestKind::Echo { .. } => Ok(None), + ClassifiedKind::Echo(echo) => self.handle_echo(peer, echo), } } + /// Pairs a structurally decoded request with the clock sample an echo + /// needs, leaving every other kind untouched. + fn classify<'a>( + &mut self, + kind: &DecodedRequestKind<'a>, + datagram_len: usize, + ) -> ClassifiedKind<'a> { + match *kind { + DecodedRequestKind::Open { no_test, params } => { + ClassifiedKind::Open { no_test, params } + } + DecodedRequestKind::Close { token } => ClassifiedKind::Close { token }, + // Everything after the sequence number is opaque, so the tail is + // read no further than its length. + DecodedRequestKind::Echo { + token, sequence, .. + } => ClassifiedKind::Echo(EchoRequest { + token, + sequence, + datagram_len, + received_at: self.clock.sample(), + }), + } + } + + /// Answers an echo request that names one of this server's sessions. + /// + /// Admission runs before any state moves, and each step is a silent + /// discard: the datagram must not be longer than + /// [`max_packet_length`](ServerConfig::max_packet_length), its token must + /// name a live session, and it must have arrived from that session's own + /// endpoint. Authentication was already settled by the caller. A request + /// failing any of these leaves the receive state — of this session and of + /// every other — exactly where it was. + /// + /// Only the token and the sequence number mean anything. The request's own + /// length is judged against the configured maximum and then forgotten: it + /// does not have to equal the negotiated length, and it never sizes the + /// reply, which is laid out and sized from the session's negotiated + /// parameters alone. + /// + /// The next receive state is computed but not committed until the reply has + /// been encoded, so a reply this server failed to build cannot leave a + /// session claiming to have answered a request it never did. + fn handle_echo( + &mut self, + peer: SocketAddr, + request: EchoRequest, + ) -> Result>, ServerError> { + // Resource policy, applied to the length actually received. This is + // `irtt-rs` policy rather than upstream-mechanism emulation: there is + // no interface-MTU effective length here, just the datagram the caller + // handed over. The comparison is strict — a request of exactly the + // maximum is served. + if request.datagram_len > self.config.max_packet_length() { + return Ok(None); + } + let Some(session) = self.sessions.get_mut(&request.token) else { + return Ok(None); + }; + if !same_endpoint(session.peer(), peer) { + return Ok(None); + } + + let next = session.receive_state().accepted(request.sequence); + // The departure side, sampled as late as the core can: after admission + // and the transition, immediately before the reply is built. There is + // no socket here, so this is the core's departure instant rather than a + // kernel transmission time; the runtime slice does not need to rewrite + // it. + let sent_at = self.clock.sample(); + let params = session.params(); + let reply = EchoReply { + // Reply only. Open must be clear or upstream clients abort, and + // Close belongs to the server-initiated close slice. The + // authentication flag and field are the encoder's, decided by the + // key it is given. + flags: FLAG_REPLY, + // Both are copied through unchanged: no token is allocated for an + // echo, and no sequence number is normalized or range-checked. + token: request.token, + sequence: request.sequence, + recv_count: params.received_stats.has_count().then_some(next.count), + recv_window: params.received_stats.has_window().then_some(next.window), + timestamps: timestamp_fields(params, request.received_at, sent_at), + // No payload bytes of our own: the encoder zero-fills the whole + // region between the field block and the negotiated length. That is + // the initial safe fill policy — payload bytes carry no protocol + // meaning, and zeroes cannot disclose residue from another request + // or another client. Request payload bytes never reach a reply. + payload: Vec::new(), + }; + let packet = encode_echo_reply(&reply, params, self.config.hmac_key()) + .map_err(|source| ServerError::ReplyEncoding { source })?; + + // The last fallible step is behind us, so the transition can be made. + session.commit_receive_state(next); + Ok(Some(packet)) + } + fn handle_open( &mut self, peer: SocketAddr, @@ -252,6 +378,80 @@ impl ServerCore { } } +/// A structurally decoded request kind, with an echo's arrival instant already +/// sampled. +/// +/// This mirrors [`DecodedRequestKind`] rather than replacing it: the decoder +/// classifies the packet, and this records what the core had to capture at that +/// moment. Keeping the two apart is what lets the clock be read before +/// authentication without any handler having to ask whether a sample exists. +enum ClassifiedKind<'a> { + Open { no_test: bool, params: &'a [u8] }, + Close { token: u64 }, + Echo(EchoRequest), +} + +/// The whole of what an echo request means to this server. +/// +/// Everything after the sequence number is opaque, so nothing beyond these +/// fields is carried forward. `datagram_len` is the length the datagram +/// actually arrived with, which the configured maximum judges; it is not +/// required to match the negotiated length, and it never sizes the reply. +#[derive(Debug, Clone, Copy)] +struct EchoRequest { + token: u64, + sequence: u32, + datagram_len: usize, + received_at: ClockSample, +} + +/// Builds the timestamp fields a session negotiated, and only those. +/// +/// [`StampAt`] selects which instants are reported and [`Clock`] selects which +/// domains, exactly as the reply layout does, so a field this returns as `Some` +/// is a field the encoder has a slot for. A midpoint is the per-domain mean of +/// the two instants. +/// +/// A single-clock midpoint therefore produces exactly one midpoint field. That +/// is the conforming representation; upstream 0.9.1's habit of emitting both +/// midpoint fields regardless of the negotiated clock is a defect this server +/// deliberately does not reproduce. `irtt-proto` still *decodes* that form, so +/// interoperating with a peer that sends it is unaffected. +/// +/// [`Clock::Unspecified`] selects no domain and so produces no field. It cannot +/// be reached with a non-none `StampAt` on an acknowledged session, because +/// that combination is refused at open; nothing here needs to repair it. +/// +/// The pair is ordered before anything is built from it, because a reply's +/// receive instant must not be later than its send instant and the wall clock +/// can be stepped backwards between the two readings. See +/// [`ClockSample::not_after`]; the correction is confined to this one reply. +fn timestamp_fields(params: &Params, received: ClockSample, sent: ClockSample) -> TimestampFields { + let received = received.not_after(sent); + let per_clock = |sample: ClockSample| { + ( + params.clock.has_wall().then_some(sample.wall_ns), + params.clock.has_mono().then_some(sample.mono_ns), + ) + }; + let absent = (None, None); + let (recv, midpoint, send) = match params.stamp_at { + StampAt::None => (absent, absent, absent), + StampAt::Receive => (per_clock(received), absent, absent), + StampAt::Send => (absent, absent, per_clock(sent)), + StampAt::Both => (per_clock(received), absent, per_clock(sent)), + StampAt::Midpoint => (absent, per_clock(received.midpoint(sent)), absent), + }; + TimestampFields { + recv_wall: recv.0, + recv_mono: recv.1, + midpoint_wall: midpoint.0, + midpoint_mono: midpoint.1, + send_wall: send.0, + send_mono: send.1, + } +} + /// Whether two source endpoints identify the same session endpoint. /// /// The address family, the IP address and the UDP source port are established @@ -352,7 +552,7 @@ fn open_params_are_admissible(decoded: &DecodedParams) -> bool { /// open rather than acknowledged and then found unserviceable. /// /// The size comes from [`echo_packet_len`], the one packet-size authority, so -/// this cannot drift from what the echo slice will actually encode. Its +/// this cannot drift from what an echo reply actually encodes. Its /// unrepresentable-length error is a rejection here, not a [`ServerError`]: a /// pathological requested length is remote input, not an internal failure. The /// default policy makes it unreachable for an acknowledged session anyway, since diff --git a/crates/irtt-server/src/lib.rs b/crates/irtt-server/src/lib.rs index bb551c6..eec7b7c 100644 --- a/crates/irtt-server/src/lib.rs +++ b/crates/irtt-server/src/lib.rs @@ -7,7 +7,9 @@ //! //! [`ServerCore`] is the deterministic protocol and session engine: packet //! admission, authentication policy, open negotiation, the session table, -//! resource decisions and reply construction. It performs no I/O. Socket and +//! echo receive state, resource decisions and reply construction. It performs +//! no I/O; the clock its echo timestamps come from is a private injected seam, +//! not a runtime abstraction, so the engine stays testable. Socket and //! runtime orchestration — the UDP listener, address handling, timers, expiry //! and shutdown — will live around it and is intentionally Tokio-native. There //! is no blocking or alternate-runtime counterpart and no transport @@ -33,12 +35,14 @@ //! //! # Current scope //! -//! Open handling, session creation and client-initiated close. Echo -//! processing, server-initiated close, session lifetime and expiry, and the -//! Tokio runtime are separate slices; a structurally valid echo request is -//! currently accepted by the admission boundary and then ignored. +//! Open handling, session creation, normal echo processing and +//! client-initiated close. Rate limiting, session lifetime and expiry, +//! server-initiated close, DSCP application, the server fill policy and the +//! Tokio runtime are separate slices: every otherwise admissible echo is +//! answered, and a session lives until its client closes it. #![forbid(unsafe_code)] +mod clock; mod config; mod core; mod error; diff --git a/crates/irtt-server/src/session.rs b/crates/irtt-server/src/session.rs index 8411184..69b970c 100644 --- a/crates/irtt-server/src/session.rs +++ b/crates/irtt-server/src/session.rs @@ -4,25 +4,29 @@ use irtt_proto::Params; /// A live session created by an accepted open request. /// -/// This slice records only what identifies the session and what was negotiated -/// for it. Echo receive state (counts, windows, sequence tracking) and lifetime -/// state (activity, deadlines, expiry) belong to the slices that implement -/// those behaviors, and are deliberately absent rather than present as -/// placeholders. +/// A session records what identifies it, what was negotiated for it, and what +/// its accepted echo requests have added up to. Lifetime state (activity, +/// deadlines, expiry) and rate state belong to the slices that implement those +/// behaviors, and are deliberately absent rather than present as placeholders. /// -/// Deliberately crate-private. The next slices are all but certain to add and -/// restructure fields here, and nothing outside the crate needs a session yet; -/// publishing internal session state before ECHO even exists would buy nothing -/// and commit us to a shape we are about to change. +/// Deliberately crate-private. The remaining slices are all but certain to add +/// and restructure fields here, and nothing outside the crate needs a session; +/// publishing internal session state would buy nothing and commit us to a shape +/// we are about to change. #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct Session { peer: SocketAddr, params: Params, + receive: ReceiveState, } impl Session { pub(crate) fn new(peer: SocketAddr, params: Params) -> Self { - Self { peer, params } + Self { + peer, + params, + receive: ReceiveState::default(), + } } /// The exact source endpoint the open request arrived from. @@ -30,10 +34,10 @@ impl Session { /// A session is bound to this endpoint, not merely to its address: the /// address family, the address and the UDP source port all form part of its /// identity, as — by `irtt-rs` policy rather than observed behavior — does - /// the IPv6 scope identifier. Close processing requires both a token match - /// and an endpoint match before removing a session, and echo processing - /// will require the same; `same_endpoint` in `core` is where the comparison - /// and its reasoning live. + /// the IPv6 scope identifier. Close and echo processing both require a + /// token match *and* an endpoint match before touching the session; + /// `same_endpoint` in `core` is where the comparison and its reasoning + /// live. /// /// The address is stored exactly as received, including the flow label that /// identity ignores, so a reply goes back to the endpoint the request @@ -42,13 +46,95 @@ impl Session { self.peer } - /// The negotiated parameters the server returned for this session and will - /// enforce for it. - // - // Also awaiting its production consumer: echo replies are built from the - // negotiated params. - #[allow(dead_code)] + /// The negotiated parameters the server returned for this session and + /// enforces for it. + /// + /// Echo replies are laid out and sized from these, never from the request + /// that provoked them. pub(crate) fn params(&self) -> &Params { &self.params } + + /// What this session's accepted echo requests have added up to so far. + pub(crate) fn receive_state(&self) -> ReceiveState { + self.receive + } + + /// Adopts the state computed for an accepted echo request. + /// + /// Kept separate from [`ReceiveState::accepted`] on purpose. The transition + /// is pure and this is the irreversible step, so echo handling runs every + /// fallible part — building and encoding the reply — before calling it, and + /// a reply that fails to encode leaves the session exactly as it was. + pub(crate) fn commit_receive_state(&mut self, next: ReceiveState) { + self.receive = next; + } +} + +/// What a session's accepted echo requests have added up to. +/// +/// The semantics are those of the clean server specification, Section 10, and +/// the input → output vectors in `test-vectors/SERVER_BEHAVIORAL_VECTORS.md` +/// Section 1. Both values are per session, start at zero, and move only for a +/// request that reaches the reply stage: a datagram dropped for its size, its +/// authentication, an unknown token or a foreign endpoint leaves them alone. +/// +/// The state is maintained whatever the session negotiated. Which of the two +/// values actually appears in a reply is a separate question, answered by the +/// negotiated `ReceivedStats` when the reply is built. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub(crate) struct ReceiveState { + /// Accepted echo requests, including the one currently being answered. + pub(crate) count: u32, + /// The received window: bit 0 is the request being answered and bit *k* is + /// the request with sequence number (current − *k*), for 1 ≤ *k* ≤ 63. + pub(crate) window: u64, + /// The sequence number the previous accepted request carried, which the + /// next transition is measured against. `None` before the session's first. + last_sequence: Option, +} + +impl ReceiveState { + /// The state after accepting a request carrying `sequence`. + /// + /// Pure, so the caller can compute the next state, build and encode a reply + /// from it, and only then commit it. + /// + /// The window transition is the observed one, and it is **not** a + /// selective-acknowledgement bitmap: history is shifted along by the + /// unsigned distance from the previous accepted sequence number, and a + /// distance it cannot represent throws all of it away. The consequence is + /// the distinctive part — because the distance is unsigned, a *late or + /// reordered* request produces a very large one, so it resets the window to + /// `0x1` instead of setting an old bit, and the history it discards is + /// never recovered. That late sequence number then becomes the reference + /// for the next transition like any other. + pub(crate) fn accepted(self, sequence: u32) -> Self { + let window = match self.last_sequence { + // The first accepted request of a session reports only itself, + // whatever its sequence number. + None => 0x1, + Some(last) => match sequence.wrapping_sub(last) { + // A duplicate of the most recent sequence number leaves the + // window unchanged, since bit 0 is already set for it. + 0 => self.window | 1, + // In order, or a gap the window can still represent: shift the + // history along by the distance and record this request. + distance @ 1..=63 => (self.window << distance) | 1, + // A gap of 64 or more discards every earlier bit — and so does + // any late or reordered request, which lands here because the + // unsigned distance wrapped. + _ => 0x1, + }, + }; + Self { + // The wire field is 32 bits. No clean evidence reaches 2^32 + // requests in one session, so the wrap is not verified behavior: it + // is simply what maintaining a fixed-width counter means, and it is + // chosen over a debug-build overflow panic a peer could drive. + count: self.count.wrapping_add(1), + window, + last_sequence: Some(sequence), + } + } } diff --git a/crates/irtt-server/src/tests/admission.rs b/crates/irtt-server/src/tests/admission.rs new file mode 100644 index 0000000..917f685 --- /dev/null +++ b/crates/irtt-server/src/tests/admission.rs @@ -0,0 +1,42 @@ +//! The structural admission boundary. +//! +//! A datagram that never decodes as a request is discarded before any request +//! kind is chosen, so it can reach neither the session table nor an echo's +//! receive state. + +use super::support::{client_params, core_for, open_request, open_session, peer, ScriptedTokens}; + +const TOKEN_A: u64 = 0x0102_0304_0506_0708; + +#[test] +fn structurally_invalid_datagrams_are_discarded() { + let mut core = core_for(None, ScriptedTokens::new([TOKEN_A])); + open_session(&mut core, peer(), None); + let session = core.session(TOKEN_A).cloned().expect("session must exist"); + + let mut bad_magic = open_request(&client_params(), None); + bad_magic[0] ^= 0x01; + + let mut reserved_flag = open_request(&client_params(), None); + reserved_flag[3] |= 0x10; + + let mut reply_flag = open_request(&client_params(), None); + reply_flag[3] |= irtt_proto::FLAG_REPLY; + + for (name, packet) in [ + ("empty datagram", Vec::new()), + ("datagram shorter than the header", vec![0x14, 0xa7, 0x5b]), + ("bad magic", bad_magic), + ("reserved flag bit", reserved_flag), + ("reply flag", reply_flag), + ] { + assert_eq!( + core.handle_datagram(peer(), &packet).unwrap(), + None, + "{name} must be discarded" + ); + } + + assert_eq!(core.session_count(), 1); + assert_eq!(core.session(TOKEN_A), Some(&session)); +} diff --git a/crates/irtt-server/src/tests/close.rs b/crates/irtt-server/src/tests/close.rs index 081d70b..8a27e1a 100644 --- a/crates/irtt-server/src/tests/close.rs +++ b/crates/irtt-server/src/tests/close.rs @@ -63,8 +63,8 @@ fn a_repeated_close_is_a_stale_token_and_changes_nothing() { #[test] fn an_echo_after_a_close_is_a_stale_token() { - // Echo processing does not exist yet; the point here is that the token is - // gone, so the request has no session to reach even once it does. + // The same request would have been answered a moment earlier: releasing the + // session takes its token with it, so there is nothing left to reach. for hmac_key in [None, Some(KEY)] { let mut core = core_for(hmac_key, ScriptedTokens::new([TOKEN_A])); let token = open_session(&mut core, peer(), hmac_key); diff --git a/crates/irtt-server/src/tests/echo.rs b/crates/irtt-server/src/tests/echo.rs new file mode 100644 index 0000000..26e2faa --- /dev/null +++ b/crates/irtt-server/src/tests/echo.rs @@ -0,0 +1,729 @@ +//! Normal echo processing. +//! +//! An echo request that authenticates, fits the configured maximum, names a +//! live session and arrives from that session's endpoint is answered from the +//! session's negotiated parameters. Everything else is silence that leaves the +//! receive state alone, which these tests prove from the wire — the count and +//! window in the *next* accepted reply are what show that a rejected datagram +//! changed nothing. +//! +//! The count and window vectors come from the clean +//! `test-vectors/SERVER_BEHAVIORAL_VECTORS.md` Section 1 and are hard-coded +//! here rather than recomputed, so that replacing the observed semantics with a +//! conventional selective-acknowledgement bitmap fails these tests. + +use std::net::{SocketAddr, SocketAddrV6}; + +use irtt_proto::{ + echo_packet_len, verify_packet_hmac, Clock, Params, ReceivedStats, StampAt, TimestampFields, +}; + +use super::support::{ + core_for, core_with_sources, core_with_tokens, echo_params, echo_request, expect_echo_reply, + open_negotiated, other_peer, peer, sample, ScriptedClock, ScriptedTokens, KEY, OTHER_KEY, +}; +use crate::ServerConfig; + +const TOKEN_A: u64 = 0x0102_0304_0506_0708; +const UNKNOWN_TOKEN: u64 = 0x2122_2324_2526_2728; + +/// Params selecting no optional field at all, so the echo layout is the +/// mandatory field block and nothing else. +fn bare_params() -> Params { + echo_params( + ReceivedStats::None, + StampAt::None, + Clock::Unspecified, + /* length */ 0, + ) +} + +/// Params reporting both statistics, which is what most vectors are stated in. +fn counting_params(length: i64) -> Params { + echo_params( + ReceivedStats::Both, + StampAt::None, + Clock::Unspecified, + length, + ) +} + +/// Opens one session on a default core and returns the core, the token and the +/// negotiated params. +fn session_with(params: &Params) -> (crate::ServerCore, u64, Params) { + let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A])); + let (token, negotiated) = open_negotiated(&mut core, peer(), params, None); + (core, token, negotiated) +} + +#[test] +fn a_valid_echo_is_answered_from_the_negotiated_params() { + let (mut core, token, negotiated) = session_with(&bare_params()); + + let packet = core + .handle_datagram(peer(), &echo_request(token, 42, &negotiated, &[], None)) + .unwrap() + .expect("an admissible echo must be answered"); + + let reply = expect_echo_reply(&packet, &negotiated, None); + assert_eq!(reply.token, token, "the session token is copied through"); + assert_eq!( + reply.sequence, 42, + "the sequence number is copied unchanged" + ); + assert_eq!(reply.recv_count, None, "no statistics were negotiated"); + assert_eq!(reply.recv_window, None); + assert_eq!( + reply.timestamps, + TimestampFields::default(), + "no timestamps were negotiated, so none are emitted" + ); + assert!(reply.payload.is_empty()); + assert_eq!(packet.len(), echo_packet_len(false, &negotiated).unwrap()); + assert_eq!(packet.len(), 16, "magic, flags, token and sequence only"); + + assert_eq!( + core.session_count(), + 1, + "an echo neither creates nor releases a session" + ); +} + +#[test] +fn a_reply_is_sized_by_the_session_not_by_the_request() { + // A request may be shorter or longer than the negotiated length, and its + // bytes past the sequence number are opaque either way. + let (mut core, token, negotiated) = session_with(&counting_params(64)); + assert_eq!(negotiated.length, 64); + + let short = Params { + length: 0, + ..negotiated.clone() + }; + // Requesting no statistics moves this request's payload region up to where + // the *session's* reply keeps its count and window, so the opaque bytes + // land squarely on fields a server must not read from a request. + let long = Params { + length: 100, + received_stats: ReceivedStats::None, + ..negotiated.clone() + }; + + for (sequence, name, sizing, payload, expected) in [ + ( + 0, + "a request shorter than the negotiated length", + &short, + &[][..], + (1, 0x1), + ), + ( + 1, + "a request longer than the negotiated length, with opaque bytes", + &long, + &[0xa5; 40][..], + (2, 0x3), + ), + ] { + let request = echo_request(token, sequence, sizing, payload, None); + assert_ne!(request.len(), 64, "{name} must not be the reply's length"); + + let packet = core + .handle_datagram(peer(), &request) + .unwrap() + .unwrap_or_else(|| panic!("{name} must be answered")); + let reply = expect_echo_reply(&packet, &negotiated, None); + + assert_eq!( + packet.len(), + 64, + "{name}: the reply is the negotiated length" + ); + assert_eq!(reply.sequence, sequence); + assert_eq!( + (reply.recv_count, reply.recv_window), + (Some(expected.0), Some(expected.1)), + "{name}: the opaque tail must not influence the reply" + ); + } +} + +#[test] +fn an_echo_over_the_configured_maximum_is_dropped_before_anything_advances() { + let config = ServerConfig::default().with_max_packet_length(64); + let mut core = core_with_tokens(config, ScriptedTokens::new([TOKEN_A])); + let (token, negotiated) = open_negotiated(&mut core, peer(), &counting_params(64), None); + assert_eq!(negotiated.length, 64); + + let at_maximum = echo_request(token, 0, &negotiated, &[], None); + assert_eq!(at_maximum.len(), 64); + let over_maximum = echo_request( + token, + 1, + &Params { + length: 65, + ..negotiated.clone() + }, + &[], + None, + ); + assert_eq!(over_maximum.len(), 65); + + // The comparison is strict, so exactly the maximum is served. + let packet = core + .handle_datagram(peer(), &at_maximum) + .unwrap() + .expect("a request of exactly the maximum must be answered"); + let reply = expect_echo_reply(&packet, &negotiated, None); + assert_eq!((reply.recv_count, reply.recv_window), (Some(1), Some(0x1))); + + assert_eq!( + core.handle_datagram(peer(), &over_maximum).unwrap(), + None, + "one byte over the maximum must be answered with silence" + ); + + // The next admissible request is what proves the drop advanced nothing: had + // the oversized one counted, this would report 3 rather than 2. + let packet = core + .handle_datagram(peer(), &echo_request(token, 1, &negotiated, &[], None)) + .unwrap() + .expect("an admissible echo after a dropped one must be answered"); + let reply = expect_echo_reply(&packet, &negotiated, None); + assert_eq!((reply.recv_count, reply.recv_window), (Some(2), Some(0x3))); +} + +#[test] +fn an_echo_that_owns_no_session_here_is_answered_with_silence() { + let (mut core, token, negotiated) = session_with(&counting_params(0)); + + for (name, endpoint, request_token) in [ + ("an unknown token", peer(), UNKNOWN_TOKEN), + ("a zero token", peer(), 0), + ( + "a valid token from another source port", + other_peer(), + token, + ), + ] { + let request = echo_request(request_token, 0, &negotiated, &[], None); + assert_eq!( + core.handle_datagram(endpoint, &request).unwrap(), + None, + "{name} must be answered with silence" + ); + } + + // Count 1 and window 0x1 are the first-request values, so none of the three + // rejected datagrams touched this session's receive state. + let packet = core + .handle_datagram(peer(), &echo_request(token, 0, &negotiated, &[], None)) + .unwrap() + .expect("the session's own echo must still be answered"); + let reply = expect_echo_reply(&packet, &negotiated, None); + assert_eq!((reply.recv_count, reply.recv_window), (Some(1), Some(0x1))); + assert_eq!(core.session_count(), 1); +} + +#[test] +fn an_echo_resolves_ipv6_endpoints_by_the_same_identity_rule_as_a_close() { + // The comparator itself is pinned by the close tests; this is the echo path + // reaching the same one rather than growing a second rule. + let global = "2001:db8::1".parse().unwrap(); + let link_local = "fe80::1".parse().unwrap(); + let unlabeled = SocketAddr::V6(SocketAddrV6::new(global, 2112, 0, 0)); + let labeled = SocketAddr::V6(SocketAddrV6::new(global, 2112, 0x000f_1234, 0)); + let zone_three = SocketAddr::V6(SocketAddrV6::new(link_local, 2112, 0, 3)); + let zone_four = SocketAddr::V6(SocketAddrV6::new(link_local, 2112, 0, 4)); + + for (bound, sender, answered) in [ + // Flow information is routing metadata that identifies no endpoint. + (unlabeled, labeled, true), + (labeled, unlabeled, true), + // The zone is identity by `irtt-rs` policy, not by observed behavior. + (zone_three, zone_four, false), + ] { + let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A])); + let (token, negotiated) = open_negotiated(&mut core, bound, &counting_params(0), None); + + let reply = core + .handle_datagram(sender, &echo_request(token, 0, &negotiated, &[], None)) + .unwrap(); + + assert_eq!( + reply.is_some(), + answered, + "an echo from {sender} on a session opened from {bound}" + ); + } +} + +#[test] +fn received_count_and_window_follow_the_observed_vectors() { + // Every row is `(request sequence, reply count, reply window)` exactly as + // measured in Section 1 of the behavioral vectors. + for (name, vectors) in [ + ( + "1.1 sequential, no loss", + &[ + (0, 1, 0x1), + (1, 2, 0x3), + (2, 3, 0x7), + (3, 4, 0xf), + (4, 5, 0x1f), + (5, 6, 0x3f), + ][..], + ), + ( + "1.2 one gap", + &[(0, 1, 0x1), (1, 2, 0x3), (3, 3, 0xd), (4, 4, 0x1b)][..], + ), + ( + "1.3 multiple gaps", + &[(0, 1, 0x1), (3, 2, 0x9), (7, 3, 0x91), (8, 4, 0x123)][..], + ), + ( + // Filling a gap late discards the history rather than setting the + // bit the gap left clear. + "1.4 later gap fill", + &[(0, 1, 0x1), (1, 2, 0x3), (5, 3, 0x31), (3, 4, 0x1)][..], + ), + ( + "1.5 duplicate sequence number", + &[(0, 1, 0x1), (1, 2, 0x3), (1, 3, 0x3), (2, 4, 0x7)][..], + ), + ( + // Row 5 rebuilds from the reset: bit 1 stays clear even though + // sequence 2 was received before the reordering. + "1.6 reordering", + &[ + (0, 1, 0x1), + (1, 2, 0x3), + (2, 3, 0x7), + (1, 4, 0x1), + (3, 5, 0x5), + ][..], + ), + ( + "1.7 strictly descending sequence numbers", + &[ + (5, 1, 0x1), + (4, 2, 0x1), + (3, 3, 0x1), + (2, 4, 0x1), + (1, 5, 0x1), + (0, 6, 0x1), + ][..], + ), + ( + "1.8 a gap of exactly 63", + &[(0, 1, 0x1), (63, 2, 0x8000_0000_0000_0001)][..], + ), + ("1.8 a gap of 64", &[(0, 1, 0x1), (64, 2, 0x1)][..]), + ("1.8 a gap of 65", &[(0, 1, 0x1), (65, 2, 0x1)][..]), + ("1.8 a gap of 1000", &[(0, 1, 0x1), (1000, 2, 0x1)][..]), + ( + "1.9 a non-zero starting sequence number", + &[(100, 1, 0x1), (101, 2, 0x3), (102, 3, 0x7)][..], + ), + ( + "1.10 32-bit sequence wraparound", + &[ + (0xffff_fffd, 1, 0x1), + (0xffff_fffe, 2, 0x3), + (0xffff_ffff, 3, 0x7), + (0, 4, 0xf), + (1, 5, 0x1f), + ][..], + ), + ] { + let (mut core, token, negotiated) = session_with(&counting_params(0)); + + for &(sequence, count, window) in vectors { + let packet = core + .handle_datagram( + peer(), + &echo_request(token, sequence, &negotiated, &[], None), + ) + .unwrap() + .unwrap_or_else(|| panic!("{name}: sequence {sequence} must be answered")); + let reply = expect_echo_reply(&packet, &negotiated, None); + + assert_eq!( + reply.recv_count, + Some(count), + "{name}: count after sequence {sequence}" + ); + assert_eq!( + reply.recv_window, + Some(window), + "{name}: window after sequence {sequence}" + ); + } + } +} + +#[test] +fn only_the_negotiated_received_statistics_are_reported() { + // Field presence only; the transition itself is pinned by the vectors + // above, so two in-order requests are enough here. + for (stats, count, window) in [ + (ReceivedStats::None, None, None), + (ReceivedStats::Count, Some(2), None), + (ReceivedStats::Window, None, Some(0x3)), + (ReceivedStats::Both, Some(2), Some(0x3)), + ] { + let params = echo_params(stats, StampAt::None, Clock::Unspecified, 0); + let (mut core, token, negotiated) = session_with(¶ms); + + let mut reply = None; + for sequence in 0..2 { + let packet = core + .handle_datagram( + peer(), + &echo_request(token, sequence, &negotiated, &[], None), + ) + .unwrap() + .unwrap_or_else(|| panic!("{stats:?}: sequence {sequence} must be answered")); + reply = Some(expect_echo_reply(&packet, &negotiated, None)); + } + + let reply = reply.expect("two requests were sent"); + assert_eq!(reply.recv_count, count, "count presence for {stats:?}"); + assert_eq!(reply.recv_window, window, "window presence for {stats:?}"); + } +} + +#[test] +fn an_authenticated_session_answers_only_authenticated_echoes() { + let mut core = core_for(Some(KEY), ScriptedTokens::new([TOKEN_A])); + let (token, negotiated) = open_negotiated(&mut core, peer(), &counting_params(0), Some(KEY)); + + let packet = core + .handle_datagram(peer(), &echo_request(token, 3, &negotiated, &[], Some(KEY))) + .unwrap() + .expect("a correctly authenticated echo must be answered"); + assert!( + verify_packet_hmac(KEY, &packet).is_ok(), + "the reply must carry a MAC that verifies with the server's key" + ); + // `expect_echo_reply` pins the HMAC flag and verifies the MAC again while + // decoding, so a reply that authenticated but omitted the flag would fail + // here — the failure mode that silently breaks upstream clients. + let reply = expect_echo_reply(&packet, &negotiated, Some(KEY)); + assert_eq!(reply.token, token); + assert_eq!(reply.sequence, 3); + assert_eq!((reply.recv_count, reply.recv_window), (Some(1), Some(0x1))); + + let mut corrupted = echo_request(token, 4, &negotiated, &[], Some(KEY)); + corrupted[5] ^= 0x80; + for (name, packet) in [ + ( + "an echo with no MAC field at all", + echo_request(token, 4, &negotiated, &[], None), + ), + ( + "an echo signed with a different key", + echo_request(token, 4, &negotiated, &[], Some(OTHER_KEY)), + ), + ("an echo with one bit flipped in the MAC", corrupted), + ] { + assert_eq!( + core.handle_datagram(peer(), &packet).unwrap(), + None, + "{name} must be answered with silence" + ); + } + + let packet = core + .handle_datagram(peer(), &echo_request(token, 4, &negotiated, &[], Some(KEY))) + .unwrap() + .expect("an authenticated echo after failed ones must be answered"); + let reply = expect_echo_reply(&packet, &negotiated, Some(KEY)); + assert_eq!( + (reply.recv_count, reply.recv_window), + (Some(2), Some(0x3)), + "none of the packets that failed authentication may be counted" + ); +} + +#[test] +fn only_the_negotiated_timestamps_are_emitted() { + let received = sample(1_000, 10_000); + let sent = sample(1_200, 10_300); + + for (stamp_at, clock, expected) in [ + ( + StampAt::Send, + Clock::Wall, + TimestampFields { + send_wall: Some(1_200), + ..TimestampFields::default() + }, + ), + ( + StampAt::Send, + Clock::Monotonic, + TimestampFields { + send_mono: Some(10_300), + ..TimestampFields::default() + }, + ), + ( + StampAt::Send, + Clock::Both, + TimestampFields { + send_wall: Some(1_200), + send_mono: Some(10_300), + ..TimestampFields::default() + }, + ), + ( + StampAt::Receive, + Clock::Wall, + TimestampFields { + recv_wall: Some(1_000), + ..TimestampFields::default() + }, + ), + ( + StampAt::Receive, + Clock::Monotonic, + TimestampFields { + recv_mono: Some(10_000), + ..TimestampFields::default() + }, + ), + ( + StampAt::Receive, + Clock::Both, + TimestampFields { + recv_wall: Some(1_000), + recv_mono: Some(10_000), + ..TimestampFields::default() + }, + ), + ( + StampAt::Both, + Clock::Wall, + TimestampFields { + recv_wall: Some(1_000), + send_wall: Some(1_200), + ..TimestampFields::default() + }, + ), + ( + StampAt::Both, + Clock::Monotonic, + TimestampFields { + recv_mono: Some(10_000), + send_mono: Some(10_300), + ..TimestampFields::default() + }, + ), + ( + StampAt::Both, + Clock::Both, + TimestampFields { + recv_wall: Some(1_000), + recv_mono: Some(10_000), + send_wall: Some(1_200), + send_mono: Some(10_300), + ..TimestampFields::default() + }, + ), + ( + StampAt::Midpoint, + Clock::Wall, + TimestampFields { + midpoint_wall: Some(1_100), + ..TimestampFields::default() + }, + ), + ( + StampAt::Midpoint, + Clock::Monotonic, + TimestampFields { + midpoint_mono: Some(10_150), + ..TimestampFields::default() + }, + ), + ( + StampAt::Midpoint, + Clock::Both, + TimestampFields { + midpoint_wall: Some(1_100), + midpoint_mono: Some(10_150), + ..TimestampFields::default() + }, + ), + ] { + let clock_source = ScriptedClock::new([received, sent]); + let mut core = core_with_sources( + ServerConfig::default(), + ScriptedTokens::new([TOKEN_A]), + clock_source.clone(), + ); + let params = echo_params(ReceivedStats::None, stamp_at, clock, 0); + let (token, negotiated) = open_negotiated(&mut core, peer(), ¶ms, None); + assert_eq!( + clock_source.remaining(), + 2, + "an open must not consume a clock sample" + ); + + let packet = core + .handle_datagram(peer(), &echo_request(token, 0, &negotiated, &[], None)) + .unwrap() + .unwrap_or_else(|| panic!("{stamp_at:?} on {clock:?} must be answered")); + let reply = expect_echo_reply(&packet, &negotiated, None); + + // Equality over the whole struct, so an unselected field appearing is a + // failure as much as a selected one carrying the wrong instant. + assert_eq!(reply.timestamps, expected, "{stamp_at:?} on {clock:?}"); + assert_eq!( + packet.len(), + echo_packet_len(false, &negotiated).unwrap(), + "{stamp_at:?} on {clock:?} must be the negotiated layout's length" + ); + assert_eq!( + clock_source.remaining(), + 0, + "{stamp_at:?} on {clock:?}: one receive and one send sample" + ); + } +} + +#[test] +fn a_backward_wall_step_never_reports_receive_after_send() { + // A reply's receive instant must not be later than its send instant, and + // the wall clock can be stepped backwards between the two readings by NTP, + // a hypervisor or an administrator. The pair settles on the later reading, + // and only the wall domain moves: the monotonic one was already ordered. + let mut core = core_with_sources( + ServerConfig::default(), + ScriptedTokens::new([TOKEN_A]), + ScriptedClock::new([sample(1_000, 10_000), sample(990, 10_300)]), + ); + let params = echo_params(ReceivedStats::None, StampAt::Both, Clock::Both, 0); + let (token, negotiated) = open_negotiated(&mut core, peer(), ¶ms, None); + + let packet = core + .handle_datagram(peer(), &echo_request(token, 0, &negotiated, &[], None)) + .unwrap() + .expect("a backward wall step must not stop the echo being answered"); + let reply = expect_echo_reply(&packet, &negotiated, None); + + assert_eq!( + reply.timestamps, + TimestampFields { + recv_wall: Some(990), + recv_mono: Some(10_000), + send_wall: Some(990), + send_mono: Some(10_300), + ..TimestampFields::default() + }, + "the receive wall value is held back to the send reading, and only it" + ); + assert!(reply.timestamps.recv_wall <= reply.timestamps.send_wall); + assert!(reply.timestamps.recv_mono <= reply.timestamps.send_mono); + + // A midpoint drawn from the same stepped pair is ordered for the same + // reason: it is the mean of two values that are no longer inverted. + let mut core = core_with_sources( + ServerConfig::default(), + ScriptedTokens::new([TOKEN_A]), + ScriptedClock::new([sample(1_000, 10_000), sample(990, 10_300)]), + ); + let params = echo_params(ReceivedStats::None, StampAt::Midpoint, Clock::Wall, 0); + let (token, negotiated) = open_negotiated(&mut core, peer(), ¶ms, None); + + let packet = core + .handle_datagram(peer(), &echo_request(token, 0, &negotiated, &[], None)) + .unwrap() + .expect("a backward wall step must not stop the echo being answered"); + let reply = expect_echo_reply(&packet, &negotiated, None); + + assert_eq!( + reply.timestamps, + TimestampFields { + midpoint_wall: Some(990), + ..TimestampFields::default() + } + ); +} + +#[test] +fn a_single_clock_midpoint_emits_exactly_one_field() { + // A guard against reproducing upstream 0.9.1's dual-field midpoint, which + // emits both midpoint fields whatever the negotiated clock. `irtt-proto` + // still decodes that form from a peer; this is about what this server + // sends. + for (clock, expected) in [ + ( + Clock::Wall, + TimestampFields { + midpoint_wall: Some(1_100), + ..TimestampFields::default() + }, + ), + ( + Clock::Monotonic, + TimestampFields { + midpoint_mono: Some(10_150), + ..TimestampFields::default() + }, + ), + ] { + let mut core = core_with_sources( + ServerConfig::default(), + ScriptedTokens::new([TOKEN_A]), + ScriptedClock::new([sample(1_000, 10_000), sample(1_200, 10_300)]), + ); + let params = echo_params(ReceivedStats::None, StampAt::Midpoint, clock, 0); + let (token, negotiated) = open_negotiated(&mut core, peer(), ¶ms, None); + + let packet = core + .handle_datagram(peer(), &echo_request(token, 0, &negotiated, &[], None)) + .unwrap() + .unwrap_or_else(|| panic!("a midpoint echo on {clock:?} must be answered")); + let reply = expect_echo_reply(&packet, &negotiated, None); + + assert_eq!( + packet.len(), + echo_packet_len(false, &negotiated).unwrap(), + "{clock:?}: the negotiated length, not the compatibility one" + ); + assert_eq!( + packet.len(), + 24, + "{clock:?}: one 8-byte midpoint field, not the upstream two" + ); + assert_eq!(reply.timestamps, expected); + } +} + +#[test] +fn a_reply_payload_is_zero_filled_and_never_reflects_the_request() { + let params = echo_params(ReceivedStats::None, StampAt::None, Clock::Unspecified, 64); + let (mut core, token, negotiated) = session_with(¶ms); + + let request = echo_request(token, 0, &negotiated, &[0xa5; 48], None); + assert!( + request[16..].iter().all(|byte| *byte == 0xa5), + "the request must carry non-zero payload bytes to reflect" + ); + + let packet = core + .handle_datagram(peer(), &request) + .unwrap() + .expect("an admissible echo must be answered"); + let reply = expect_echo_reply(&packet, &negotiated, None); + + assert_eq!(packet.len(), 64); + assert_eq!(reply.payload.len(), 48); + assert!( + reply.payload.iter().all(|byte| *byte == 0), + "the payload region must be zero-filled, never request or buffer residue" + ); +} diff --git a/crates/irtt-server/src/tests/mod.rs b/crates/irtt-server/src/tests/mod.rs index 96687ed..f9a4f60 100644 --- a/crates/irtt-server/src/tests/mod.rs +++ b/crates/irtt-server/src/tests/mod.rs @@ -5,10 +5,15 @@ //! production encoders wherever the wire form is well-formed, and by hand only //! where the point of the test is a payload a compliant encoder cannot produce. //! -//! Tokens come from a scripted source so session identity, collisions and -//! allocation failure are all assertable. +//! The two nondeterministic parts are injected. Tokens come from a scripted +//! source so session identity, collisions and allocation failure are all +//! assertable; the clock an echo reply's timestamps come from is scripted the +//! same way, so timestamp behavior is pinned exactly, with no sleeps, no +//! runtime and no timing tolerances. +mod admission; mod close; +mod echo; mod hmac; mod no_test; mod open; @@ -16,4 +21,3 @@ mod params; mod sessions; mod support; mod tokens; -mod unsupported; diff --git a/crates/irtt-server/src/tests/support.rs b/crates/irtt-server/src/tests/support.rs index 73c8d66..e780cee 100644 --- a/crates/irtt-server/src/tests/support.rs +++ b/crates/irtt-server/src/tests/support.rs @@ -5,11 +5,17 @@ use std::{ }; use irtt_proto::{ - decode_open_reply, encode_request, varint, OpenReply, PacketLayout, Params, RequestToEncode, - FLAG_CLOSE, FLAG_HMAC, FLAG_OPEN, FLAG_REPLY, + decode_echo_reply, decode_open_reply, encode_request, varint, EchoReply, OpenReply, + PacketLayout, Params, RequestToEncode, FLAG_CLOSE, FLAG_HMAC, FLAG_OPEN, FLAG_REPLY, }; -use crate::{core::ServerCore, error::ServerError, token::TokenSource, ServerConfig}; +use crate::{ + clock::{ClockSample, ClockSource}, + core::ServerCore, + error::ServerError, + token::TokenSource, + ServerConfig, +}; pub(crate) const KEY: &[u8] = b"correctkey"; pub(crate) const OTHER_KEY: &[u8] = b"wrongkey"; @@ -67,11 +73,63 @@ impl TokenSource for ScriptedTokens { } } -/// A core with deterministic tokens, so session identity is assertable. +/// A clock handing out a fixed script of samples, then failing loudly. +/// +/// Running the script dry is a test bug, not a wrap-around: a test that asserts +/// timestamps must know exactly which sample each field came from. The core +/// takes one sample when it classifies a datagram as an echo, and a second when +/// an admitted echo is about to be answered. +#[derive(Debug, Clone)] +pub(crate) struct ScriptedClock { + samples: Arc>>, +} + +impl ScriptedClock { + pub(crate) fn new(samples: I) -> Self + where + I: IntoIterator, + { + Self { + samples: Arc::new(Mutex::new(samples.into_iter().collect())), + } + } + + /// How many scripted samples are still unused. + pub(crate) fn remaining(&self) -> usize { + self.samples.lock().unwrap().len() + } +} + +impl ClockSource for ScriptedClock { + fn sample(&mut self) -> ClockSample { + self.samples + .lock() + .unwrap() + .pop_front() + .expect("the scripted clock ran out: script one sample per echo received, and one more per echo answered") + } +} + +pub(crate) fn sample(wall_ns: i64, mono_ns: i64) -> ClockSample { + ClockSample { wall_ns, mono_ns } +} + +/// A core with deterministic tokens, so session identity is assertable. Its +/// clock is the production one, which is fine for everything that does not +/// assert timestamp values. pub(crate) fn core_with_tokens(config: ServerConfig, tokens: ScriptedTokens) -> ServerCore { ServerCore::with_token_source(config, Box::new(tokens)) } +/// A core with both nondeterministic sources scripted, for timestamp tests. +pub(crate) fn core_with_sources( + config: ServerConfig, + tokens: ScriptedTokens, + clock: ScriptedClock, +) -> ServerCore { + ServerCore::with_sources(config, Box::new(tokens), Box::new(clock)) +} + /// An otherwise default core that authenticates iff `hmac_key` is set. /// /// Most behavior has to hold identically with and without authentication, so @@ -97,6 +155,80 @@ pub(crate) fn open_session( expect_normal_open_reply(&packet, hmac_key).token } +/// Opens one session carrying `params` and returns its token together with the +/// parameters the server negotiated for it. +/// +/// Echo tests need both: the token addresses the session, and the negotiated +/// params are what the reply is laid out from and what decoding it requires. +pub(crate) fn open_negotiated( + core: &mut ServerCore, + endpoint: SocketAddr, + params: &Params, + hmac_key: Option<&[u8]>, +) -> (u64, Params) { + let packet = core + .handle_datagram(endpoint, &open_request(params, hmac_key)) + .unwrap() + .expect("the session-creating open must be answered"); + let reply = expect_normal_open_reply(&packet, hmac_key); + (reply.token, reply.params) +} + +/// Params requesting exactly the echo layout a test cares about, with the +/// duration and interval a well-behaved client sends. +pub(crate) fn echo_params( + received_stats: irtt_proto::ReceivedStats, + stamp_at: irtt_proto::StampAt, + clock: irtt_proto::Clock, + length: i64, +) -> Params { + Params { + received_stats, + stamp_at, + clock, + length, + ..client_params() + } +} + +/// Encodes an echo request. `params` supplies only the *request's* own sizing; +/// it is deliberately a separate argument from the session's negotiated params +/// so tests can send requests shorter and longer than the session negotiated. +pub(crate) fn echo_request( + token: u64, + sequence: u32, + params: &Params, + payload: &[u8], + hmac_key: Option<&[u8]>, +) -> Vec { + encode_request( + RequestToEncode::Echo { + token, + sequence, + params, + payload, + }, + hmac_key, + ) + .unwrap() +} + +/// Decodes an echo reply the server produced and checks the flags every reply +/// must carry. Decoding verifies the MAC against `hmac_key` on the way. +pub(crate) fn expect_echo_reply( + packet: &[u8], + params: &Params, + hmac_key: Option<&[u8]>, +) -> EchoReply { + let reply = decode_echo_reply(packet, params, hmac_key).expect("server echo reply must decode"); + // Exact equality, so Open and Close are pinned clear as well: Open makes an + // upstream client abort, and Close belongs to the server-initiated close + // slice. + let expected = FLAG_REPLY | if hmac_key.is_some() { FLAG_HMAC } else { 0 }; + assert_eq!(reply.flags, expected, "echo reply flags"); + reply +} + /// Encodes a normal open request carrying `params`. pub(crate) fn open_request(params: &Params, hmac_key: Option<&[u8]>) -> Vec { encode_request( diff --git a/crates/irtt-server/src/tests/unsupported.rs b/crates/irtt-server/src/tests/unsupported.rs deleted file mode 100644 index 4b69d46..0000000 --- a/crates/irtt-server/src/tests/unsupported.rs +++ /dev/null @@ -1,74 +0,0 @@ -//! Requests this slice deliberately does not implement. -//! -//! Echo is recognized by the admission boundary and then ignored. These tests -//! exist to keep the scope boundary honest: receive state must not appear by -//! accident before the slice that owns it. - -use irtt_proto::{encode_request, RequestToEncode}; - -use super::support::{ - client_params, core_for, open_request, open_session, peer, ScriptedTokens, KEY, -}; - -const TOKEN_A: u64 = 0x0102_0304_0506_0708; - -fn open_one_session(hmac_key: Option<&[u8]>) -> crate::ServerCore { - let mut core = core_for(hmac_key, ScriptedTokens::new([TOKEN_A])); - open_session(&mut core, peer(), hmac_key); - core -} - -#[test] -fn a_valid_echo_is_neither_answered_nor_counted_yet() { - for hmac_key in [None, Some(KEY)] { - let mut core = open_one_session(hmac_key); - let session = core.session(TOKEN_A).cloned().expect("session must exist"); - - let echo = encode_request( - RequestToEncode::Echo { - token: TOKEN_A, - sequence: 0, - params: session.params(), - payload: &[], - }, - hmac_key, - ) - .unwrap(); - - assert_eq!(core.handle_datagram(peer(), &echo).unwrap(), None); - assert_eq!(core.session_count(), 1); - assert_eq!(core.session(TOKEN_A), Some(&session)); - } -} - -#[test] -fn structurally_invalid_datagrams_are_discarded() { - let mut core = open_one_session(None); - let session = core.session(TOKEN_A).cloned().expect("session must exist"); - - let mut bad_magic = open_request(&client_params(), None); - bad_magic[0] ^= 0x01; - - let mut reserved_flag = open_request(&client_params(), None); - reserved_flag[3] |= 0x10; - - let mut reply_flag = open_request(&client_params(), None); - reply_flag[3] |= irtt_proto::FLAG_REPLY; - - for (name, packet) in [ - ("empty datagram", Vec::new()), - ("datagram shorter than the header", vec![0x14, 0xa7, 0x5b]), - ("bad magic", bad_magic), - ("reserved flag bit", reserved_flag), - ("reply flag", reply_flag), - ] { - assert_eq!( - core.handle_datagram(peer(), &packet).unwrap(), - None, - "{name} must be discarded" - ); - } - - assert_eq!(core.session_count(), 1); - assert_eq!(core.session(TOKEN_A), Some(&session)); -}