diff --git a/Cargo.lock b/Cargo.lock index 74cfbeaf08..ee396f48c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2588,6 +2588,7 @@ dependencies = [ "ironrdp-core 0.2.1", "ironrdp-pdu", "tracing", + "web-time", ] [[package]] diff --git a/crates/ironrdp-acceptor/src/channel_connection.rs b/crates/ironrdp-acceptor/src/channel_connection.rs index 66a066823a..e531deb5e4 100644 --- a/crates/ironrdp-acceptor/src/channel_connection.rs +++ b/crates/ironrdp-acceptor/src/channel_connection.rs @@ -1,7 +1,7 @@ use std::collections::HashSet; use ironrdp_connector::{ - ConnectorError, ConnectorErrorExt as _, ConnectorResult, Sequence, State, Written, reason_err, + ConnectorError, ConnectorErrorExt as _, ConnectorResult, MonotonicInstant, Sequence, State, Written, reason_err, }; use ironrdp_core::WriteBuf; use ironrdp_pdu::mcs; @@ -72,7 +72,12 @@ impl Sequence for ChannelConnectionSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match core::mem::take(&mut self.state) { ChannelConnectionState::WaitErectDomainRequest => { let erect_domain_request = ironrdp_core::decode::>(input) diff --git a/crates/ironrdp-acceptor/src/connection.rs b/crates/ironrdp-acceptor/src/connection.rs index 7217bfab03..286dea151a 100644 --- a/crates/ironrdp-acceptor/src/connection.rs +++ b/crates/ironrdp-acceptor/src/connection.rs @@ -2,8 +2,8 @@ use core::any::TypeId; use core::mem; use ironrdp_connector::{ - ConnectorError, ConnectorErrorExt as _, ConnectorResult, DesktopSize, Sequence, State, Written, encode_x224_packet, - general_err, reason_err, + ConnectorError, ConnectorErrorExt as _, ConnectorResult, DesktopSize, MonotonicInstant, Sequence, State, Written, + encode_x224_packet, general_err, reason_err, }; use ironrdp_core::{WriteBuf, decode}; use ironrdp_pdu as pdu; @@ -287,7 +287,8 @@ impl Acceptor { /// Panics if state is not [AcceptorState::SecurityUpgrade]. pub fn mark_security_upgrade_as_done(&mut self) { assert!(self.reached_security_upgrade().is_some()); - self.step(&[], &mut WriteBuf::new()).expect("transition to next state"); + self.step(&[], None, &mut WriteBuf::new()) + .expect("transition to next state"); debug_assert!(self.reached_security_upgrade().is_none()); } @@ -300,7 +301,9 @@ impl Acceptor { /// Panics if state is not [AcceptorState::Credssp]. pub fn mark_credssp_as_done(&mut self) { assert!(self.should_perform_credssp()); - let res = self.step(&[], &mut WriteBuf::new()).expect("transition to next state"); + let res = self + .step(&[], None, &mut WriteBuf::new()) + .expect("transition to next state"); debug_assert!(!self.should_perform_credssp()); assert_eq!(res, Written::Nothing); } @@ -458,7 +461,12 @@ impl Sequence for Acceptor { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult { let prev_state = mem::take(&mut self.state); let (written, next_state) = match prev_state { @@ -724,7 +732,7 @@ impl Sequence for Acceptor { channels, mut connection, } => { - let written = connection.step(input, output)?; + let written = connection.step(input, received_at, output)?; let state = if connection.is_done() { AcceptorState::RdpSecurityCommencement { protocol, @@ -964,7 +972,7 @@ impl Sequence for Acceptor { channels, client_capabilities, } => { - let written = finalization.step(input, output)?; + let written = finalization.step(input, received_at, output)?; let state = if finalization.is_done() { AcceptorState::Accepted { diff --git a/crates/ironrdp-acceptor/src/finalization.rs b/crates/ironrdp-acceptor/src/finalization.rs index 9961e87ce7..3ad262a9d8 100644 --- a/crates/ironrdp-acceptor/src/finalization.rs +++ b/crates/ironrdp-acceptor/src/finalization.rs @@ -1,4 +1,6 @@ -use ironrdp_connector::{ConnectorError, ConnectorErrorExt as _, ConnectorResult, Sequence, State, Written}; +use ironrdp_connector::{ + ConnectorError, ConnectorErrorExt as _, ConnectorResult, MonotonicInstant, Sequence, State, Written, +}; use ironrdp_core::WriteBuf; use ironrdp_pdu::rdp; use ironrdp_pdu::x224::X224; @@ -78,7 +80,12 @@ impl Sequence for FinalizationSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match core::mem::take(&mut self.state) { FinalizationState::WaitSynchronize => { let synchronize = decode_share_control(input); diff --git a/crates/ironrdp-async/Cargo.toml b/crates/ironrdp-async/Cargo.toml index 749f3257c8..5423515e3e 100644 --- a/crates/ironrdp-async/Cargo.toml +++ b/crates/ironrdp-async/Cargo.toml @@ -22,6 +22,7 @@ ironrdp-core = { path = "../ironrdp-core", version = "0.2", features = ["alloc"] ironrdp-pdu = { path = "../ironrdp-pdu", version = "0.9" } # public tracing = { version = "0.1", features = ["log"] } bytes = "1" # public +web-time = "1.1" [lints] workspace = true diff --git a/crates/ironrdp-async/src/framed.rs b/crates/ironrdp-async/src/framed.rs index eff516456c..774858d988 100644 --- a/crates/ironrdp-async/src/framed.rs +++ b/crates/ironrdp-async/src/framed.rs @@ -1,6 +1,7 @@ use std::io; use bytes::{Bytes, BytesMut}; +use ironrdp_connector::MonotonicInstant; use ironrdp_connector::{ConnectorResult, Sequence, Written}; use ironrdp_core::WriteBuf; use ironrdp_pdu::PduHint; @@ -56,12 +57,25 @@ pub trait StreamWrapper: Sized { pub struct Framed { stream: S, buf: BytesMut, + /// When the most recent socket read completed, if this build observes time. + /// + /// A PDU served entirely from `buf` arrived at the socket read that filled it, + /// not at the moment the caller happened to drain it, so this is the honest + /// arrival time for anything `read_by_hint` returns. `None` until the first + /// read, and always `None` on a build with no clock. + last_read_at: Option, } impl Framed { pub fn peek(&self) -> &[u8] { &self.buf } + + /// When the bytes currently buffered last arrived from the socket, or `None` + /// if nothing has been read yet or this build cannot observe time. + pub fn last_read_at(&self) -> Option { + self.last_read_at + } } impl Framed @@ -76,6 +90,7 @@ where Self { stream: S::from_inner(stream), buf: leftover, + last_read_at: None, } } @@ -197,7 +212,9 @@ where /// `tokio::select!` statement and some other branch /// completes first, then it is guaranteed that no data was read. async fn read(&mut self) -> io::Result { - self.stream.read(&mut self.buf).await + let len = self.stream.read(&mut self.buf).await?; + self.last_read_at = Some(monotonic_now()); + Ok(len) } } @@ -261,7 +278,7 @@ where trace!(length = pdu.len(), "PDU received"); - sequence.step(&pdu, buf) + sequence.step(&pdu, framed.last_read_at(), buf) } else { sequence.step_no_input(buf) } @@ -287,3 +304,17 @@ where Ok(()) } + +/// Reads the driver-owned monotonic clock. +/// +/// The epoch is the first call; only differences are meaningful. +/// +/// `web_time::Instant` is `std::time::Instant` everywhere except +/// `wasm32-unknown-unknown`, where `std`'s panics and this one reads +/// `Performance.now()` instead. This crate is reached from `ironrdp-web` through +/// `ironrdp-futures`, so without it the browser build has no clock and every +/// measurement there is lost. +fn monotonic_now() -> MonotonicInstant { + static EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(web_time::Instant::now); + MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX)) +} diff --git a/crates/ironrdp-blocking/src/connector.rs b/crates/ironrdp-blocking/src/connector.rs index 8b24701f89..0e4e3ad585 100644 --- a/crates/ironrdp-blocking/src/connector.rs +++ b/crates/ironrdp-blocking/src/connector.rs @@ -230,7 +230,7 @@ where trace!(length = pdu.len(), "PDU received"); - connector.step(&pdu, buf)? + connector.step(&pdu, framed.last_read_at(), buf)? } else { connector.step_no_input(buf)? }; diff --git a/crates/ironrdp-blocking/src/framed.rs b/crates/ironrdp-blocking/src/framed.rs index a2964d4481..50bc5c8c5f 100644 --- a/crates/ironrdp-blocking/src/framed.rs +++ b/crates/ironrdp-blocking/src/framed.rs @@ -1,12 +1,17 @@ use std::io::{self, Read, Write}; use bytes::{Bytes, BytesMut}; +use ironrdp_connector::MonotonicInstant; use ironrdp_pdu::PduHint; use tracing::debug; pub struct Framed { stream: S, buf: BytesMut, + /// When the most recent socket read completed. A PDU served from `buf` + /// arrived at the read that filled it, not when the caller drained it. + /// `None` until the first read. + last_read_at: Option, } impl Framed { @@ -15,7 +20,11 @@ impl Framed { } pub fn new_with_leftover(stream: S, leftover: BytesMut) -> Self { - Self { stream, buf: leftover } + Self { + stream, + buf: leftover, + last_read_at: None, + } } pub fn into_inner(self) -> (S, BytesMut) { @@ -36,6 +45,11 @@ impl Framed { (&mut self.stream, &mut self.buf) } + /// When the bytes currently buffered last arrived from the socket. + pub fn last_read_at(&self) -> Option { + self.last_read_at + } + pub fn peek(&self) -> &[u8] { &self.buf } @@ -118,6 +132,7 @@ where let mut read_bytes = [0u8; 1024]; let len = self.stream.read(&mut read_bytes)?; + self.last_read_at = Some(monotonic_now()); self.buf.extend_from_slice(&read_bytes[..len]); Ok(len) @@ -133,3 +148,10 @@ where self.stream.write_all(buf) } } + +/// Reads the driver-owned monotonic clock. Epoch is the first call; only +/// differences are meaningful. +fn monotonic_now() -> MonotonicInstant { + static EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(std::time::Instant::now); + MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX)) +} diff --git a/crates/ironrdp-client/src/rdp.rs b/crates/ironrdp-client/src/rdp.rs index bd441342da..396bfffb80 100644 --- a/crates/ironrdp-client/src/rdp.rs +++ b/crates/ironrdp-client/src/rdp.rs @@ -1896,7 +1896,7 @@ where debug_assert!(connector.next_pdu_hint().is_some()); buf.clear(); - let written = connector.step(x224_connection_response.as_bytes(), &mut buf)?; + let written = connector.step(x224_connection_response.as_bytes(), None, &mut buf)?; debug_assert!(written.is_nothing()); let should_upgrade = ironrdp_tokio::skip_connect_begin(connector); diff --git a/crates/ironrdp-connector/src/channel_connection.rs b/crates/ironrdp-connector/src/channel_connection.rs index 199946aed4..abf55a1d06 100644 --- a/crates/ironrdp-connector/src/channel_connection.rs +++ b/crates/ironrdp-connector/src/channel_connection.rs @@ -7,7 +7,8 @@ use ironrdp_pdu::{PduHint, mcs}; use tracing::{debug, warn}; use crate::{ - ConnectorError, ConnectorErrorExt as _, ConnectorResult, Sequence, State, Written, general_err, reason_err, + ConnectorError, ConnectorErrorExt as _, ConnectorResult, MonotonicInstant, Sequence, State, Written, general_err, + reason_err, }; #[derive(Default, Debug)] @@ -94,7 +95,12 @@ impl Sequence for ChannelConnectionSequence { } } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { ChannelConnectionState::Consumed => { return Err(general_err!( diff --git a/crates/ironrdp-connector/src/connection.rs b/crates/ironrdp-connector/src/connection.rs index 82d5e2b0a0..7efee90317 100644 --- a/crates/ironrdp-connector/src/connection.rs +++ b/crates/ironrdp-connector/src/connection.rs @@ -18,7 +18,7 @@ use crate::connection_activation::{ }; use crate::license_exchange::{LicenseExchangeSequence, NoopLicenseCache}; use crate::{ - Config, ConnectorError, ConnectorErrorExt as _, ConnectorErrorKind, ConnectorResult, DesktopSize, + Config, ConnectorError, ConnectorErrorExt as _, ConnectorErrorKind, ConnectorResult, DesktopSize, MonotonicInstant, NegotiationFailure, Sequence, State, Written, encode_x224_packet, general_err, reason_err, }; @@ -27,6 +27,14 @@ use crate::{ /// transport protocol: reliable + lossy UDP). const MAX_MULTITRANSPORT_REQUESTS: usize = 2; +/// Reported as `timeDelta` when a connect-time bandwidth window was not timed. +/// +/// One millisecond rather than zero, because a server computing +/// `byteCount * 8 / timeDelta` divides by it ([MS-RDPBCGR] 3.3.5.14). It also +/// floors a window that was timed and elapsed in under a millisecond, where it is +/// a real bound rather than a stand-in. +const UNMEASURABLE_INTERVAL_MS: u32 = 1; + /// Outcome of a single multitransport bootstrapping request, passed to /// [`ClientConnector::complete_multitransport()`]. /// @@ -237,7 +245,7 @@ impl State for ClientConnectorState { #[expect( clippy::partial_pub_fields, - reason = "server response flags are negotiated internally and must not expand the public connector construction API" + reason = "server response flags are negotiated internally and must not expand the public connector construction API; the connect-time bandwidth accumulators are likewise internal to the measurement, and exposing them would let a caller break the Start/Payload/Stop invariant" )] #[derive(Debug)] pub struct ClientConnector { @@ -260,6 +268,18 @@ pub struct ClientConnector { /// /// Set via [`ClientConnector::with_auto_reconnect_cookie`]. pub auto_reconnect_cookie: Option, + /// Start of the in-flight connect-time bandwidth measurement window. + /// + /// Set when the server's Bandwidth Measure Start arrives, and only when the + /// driver reported an arrival time for it. Cleared when the matching Stop is + /// answered. `None` therefore means no window is open, whether because no Start + /// was seen or because this driver does not observe time at all. + connect_time_bw_started_at: Option, + /// Bytes seen in the open window, accumulated across Payload messages. + /// + /// Only accumulated while a window is open, since a total with no interval to + /// divide it by is not a measurement of anything. + connect_time_bw_bytes: u32, } impl ClientConnector { @@ -273,6 +293,8 @@ impl ClientConnector { response_flags: nego::ResponseFlags::empty(), server_multitransport_flags: None, auto_reconnect_cookie: None, + connect_time_bw_started_at: None, + connect_time_bw_bytes: 0, } } @@ -476,7 +498,8 @@ impl ClientConnector { /// Panics if state is not [ClientConnectorState::EnhancedSecurityUpgrade]. pub fn mark_security_upgrade_as_done(&mut self) { assert!(self.should_perform_security_upgrade()); - self.step(&[], &mut WriteBuf::new()).expect("transition to next state"); + self.step(&[], None, &mut WriteBuf::new()) + .expect("transition to next state"); debug_assert!(!self.should_perform_security_upgrade()); } @@ -491,7 +514,9 @@ impl ClientConnector { /// Panics if state is not [ClientConnectorState::Credssp]. pub fn mark_credssp_as_done(&mut self) { assert!(self.should_perform_credssp()); - let res = self.step(&[], &mut WriteBuf::new()).expect("transition to next state"); + let res = self + .step(&[], None, &mut WriteBuf::new()) + .expect("transition to next state"); debug_assert!(!self.should_perform_credssp()); assert_eq!(res, Written::Nothing); } @@ -685,6 +710,139 @@ impl ClientConnector { Written::from_size(total_written) } } + + fn respond_to_connect_time_autodetect( + &mut self, + request: rdp::autodetect::AutoDetectRequest, + received_at: Option, + message_channel_id: u16, + user_channel_id: u16, + output: &mut WriteBuf, + ) -> ConnectorResult { + use ironrdp_pdu::rdp::autodetect::{ + AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu, BW_RESULTS_CONNECT_TIME, + }; + + match request { + AutoDetectRequest::RttRequest { sequence_number, .. } => { + let response = AutoDetectRspPdu::new(AutoDetectResponse::RttResponse { sequence_number }); + let written = encode_send_data_request(user_channel_id, message_channel_id, &response, output)?; + Written::from_size(written) + } + // Start opens the measurement window ([MS-RDPBCGR] 2.2.14.1.2). No reply is + // due; we only note when it arrived. + // + // A driver that reports no arrival time cannot time this window, so it does + // not open one. That keeps the two unmeasurable situations distinct: a + // window that was timed and turned out to be short is still a measurement, + // while a driver with no clock never took one. + AutoDetectRequest::BandwidthMeasureStart { .. } => { + self.connect_time_bw_started_at = received_at; + self.connect_time_bw_bytes = 0; + Ok(Written::Nothing) + } + // Payload carries the bytes whose transfer is being timed ([MS-RDPBCGR] + // 2.2.14.1.3). No reply is due; accumulate so Stop can report the total. + // With no window open there is nothing for the total to be divided by, so + // there is nothing worth accumulating. + // + // [MS-RDPBCGR] 3.2.5.14 increments the Byte Count store by payloadLength + // plus the size of the header fields (8 bytes: headerLength, headerTypeId, + // sequenceNumber, requestType, and payloadLength itself), not by + // payloadLength alone. `payload.len()` is exactly payloadLength, since + // decode reads that many bytes into it after consuming the header fields. + AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => { + if self.connect_time_bw_started_at.is_some() { + let len = u32::try_from(payload.len()).unwrap_or(u32::MAX).saturating_add(8); + self.connect_time_bw_bytes = self.connect_time_bw_bytes.saturating_add(len); + } + Ok(Written::Nothing) + } + // A connect-time Bandwidth Measure Stop ([MS-RDPBCGR] 2.2.14.1.4) warrants a + // Bandwidth Measure Results reply ([MS-RDPBCGR] 2.2.14.2.2). This reply must + // be sent: FreeRDP-based servers (for example GNOME Remote Desktop) block in + // their AWAIT_BW_RESULT state until they receive it and never proceed to + // licensing without it, so omitting it stalls the whole connection. + // + // [MS-RDPBCGR] 3.2.5.14 has the client increment its Network Characteristics + // Byte Count store on each Payload and on this Stop, then send the store + // together with the elapsed timer. That is what a timed window reports here, + // measured from the Start that opened it using the times the I/O driver + // observed rather than any clock this sequence could read for itself. + // + // The spec does not contemplate a window it could not time, and two arise in + // practice. No Start was seen, so no window exists. Or the driver reports no + // arrival time for this particular step, as the FFI driver does for the whole + // connection and the wasm32 driver does for the single x224_connection_response + // step, so no window was opened to accumulate into. + // + // Both still owe the server a reply, and `timeDelta` of 0 divides out to an + // unbounded bandwidth for a server computing `byteCount * 8 / timeDelta` + // (3.2.5.14 again, and [MS-RDPBCGR] 3.3.5.14 for the server side). They + // report the floor against this Stop's payload alone, which is the smallest + // claim that answers the question asked. + // + // A window that was timed reports its full count even when the elapsed time + // rounds down to the floor, which happens when one socket read delivered the + // whole exchange. The bytes did arrive within that millisecond, so the floor + // is a real bound on a real measurement rather than a stand-in for a missing + // one, and the quotient it yields is honest. + AutoDetectRequest::BandwidthMeasureStop { + sequence_number, + payload, + .. + } => { + // Same 8-byte header addition as the Payload arm above, and for the + // same spec reason ([MS-RDPBCGR] 3.2.5.14): only applies when this Stop + // actually carries a payloadLength/payload pair. + let stop_bytes = payload + .as_ref() + .map_or(0, |p| u32::try_from(p.len()).unwrap_or(u32::MAX).saturating_add(8)); + + // A window normally opens and closes on the same driver, so the same + // driver stamps both Start and this Stop. Nothing enforces that: a + // `Framed` rebuilt between the two (leftover bytes handed to a fresh + // `Framed`, which starts with no arrival time of its own) would open a + // window on one driver and close it on another with no reading, landing + // in the `(Some, None)` arm below. That arm silently drops whatever this + // window had accumulated; the debug log makes the drop visible instead of + // leaving it indistinguishable from the ordinary no-window case. + let (time_delta_ms, byte_count) = match (self.connect_time_bw_started_at, received_at) { + (Some(started_at), Some(stopped_at)) => { + let measured_ms = + u32::try_from(stopped_at.duration_since(started_at).as_millis()).unwrap_or(u32::MAX); + ( + measured_ms.max(UNMEASURABLE_INTERVAL_MS), + self.connect_time_bw_bytes.saturating_add(stop_bytes), + ) + } + (Some(_), None) => { + debug!( + dropped_bytes = self.connect_time_bw_bytes, + "Bandwidth Measure Stop arrived with no arrival time although its window was open; \ + dropping the accumulated count" + ); + (UNMEASURABLE_INTERVAL_MS, stop_bytes) + } + (None, _) => (UNMEASURABLE_INTERVAL_MS, stop_bytes), + }; + + self.connect_time_bw_started_at = None; + self.connect_time_bw_bytes = 0; + + let response = AutoDetectRspPdu::new(AutoDetectResponse::BandwidthMeasureResults { + sequence_number, + response_type: BW_RESULTS_CONNECT_TIME, + time_delta_ms, + byte_count, + }); + let written = encode_send_data_request(user_channel_id, message_channel_id, &response, output)?; + Written::from_size(written) + } + // The Network Characteristics Result is informational; nothing to send. + _ => Ok(Written::Nothing), + } + } } /// Build an Initiate Multitransport Response carrying `hr_response`. @@ -708,9 +866,10 @@ fn advance_licensing_exchange( user_channel_id: u16, message_channel_id: Option, input: &[u8], + received_at: Option, output: &mut WriteBuf, ) -> ConnectorResult<(Written, ClientConnectorState)> { - let written = license_exchange.step(input, output)?; + let written = license_exchange.step(input, received_at, output)?; let next_state = if license_exchange.state.is_terminal() { ClientConnectorState::MultitransportBootstrapping { @@ -773,7 +932,12 @@ impl Sequence for ClientConnector { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { // Invalid state ClientConnectorState::Consumed => { @@ -954,7 +1118,7 @@ impl Sequence for ClientConnector { mut channel_connection, } => { debug!("Channel Connection"); - let written = channel_connection.step(input, output)?; + let written = channel_connection.step(input, received_at, output)?; let next_state = if let ChannelConnectionState::AllJoined { user_channel_id } = channel_connection.state { @@ -1035,8 +1199,9 @@ impl Sequence for ClientConnector { if let Some((message_channel_id, data)) = message_channel_pdu { if let Ok(autodetect) = decode::(&data.user_data) { - let written = respond_to_connect_time_autodetect( + let written = self.respond_to_connect_time_autodetect( autodetect.request, + received_at, message_channel_id, user_channel_id, output, @@ -1085,6 +1250,7 @@ impl Sequence for ClientConnector { user_channel_id, self.message_channel_id, input, + received_at, output, )? } else { @@ -1116,6 +1282,7 @@ impl Sequence for ClientConnector { user_channel_id, self.message_channel_id, input, + received_at, output, )? } @@ -1184,7 +1351,7 @@ impl Sequence for ClientConnector { // exchange with the PDU intact. let mut connection_activation = ConnectionActivationSequence::new(self.config.clone(), io_channel_id, user_channel_id); - let written = connection_activation.step(input, output)?; + let written = connection_activation.step(input, received_at, output)?; ( written, @@ -1221,7 +1388,7 @@ impl Sequence for ClientConnector { ClientConnectorState::CapabilitiesExchange { mut connection_activation, } => { - let written = connection_activation.step(input, output)?; + let written = connection_activation.step(input, received_at, output)?; match connection_activation.connection_activation_state() { ConnectionActivationState::ConnectionFinalization { .. } => ( written, @@ -1245,7 +1412,7 @@ impl Sequence for ClientConnector { ClientConnectorState::ConnectionFinalization { mut connection_activation, } => { - let written = connection_activation.step(input, output)?; + let written = connection_activation.step(input, received_at, output)?; let next_state = if !connection_activation.connection_activation_state().is_terminal() { ClientConnectorState::ConnectionFinalization { connection_activation } @@ -1328,54 +1495,6 @@ pub fn encode_send_data_request( Ok(written) } -fn respond_to_connect_time_autodetect( - request: rdp::autodetect::AutoDetectRequest, - message_channel_id: u16, - user_channel_id: u16, - output: &mut WriteBuf, -) -> ConnectorResult { - use ironrdp_pdu::rdp::autodetect::{ - AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu, BW_RESULTS_CONNECT_TIME, - }; - - match request { - AutoDetectRequest::RttRequest { sequence_number, .. } => { - let response = AutoDetectRspPdu::new(AutoDetectResponse::RttResponse { sequence_number }); - let written = encode_send_data_request(user_channel_id, message_channel_id, &response, output)?; - Written::from_size(written) - } - // A connect-time Bandwidth Measure Stop ([MS-RDPBCGR] 2.2.14.1.4) warrants a - // Bandwidth Measure Results reply ([MS-RDPBCGR] 2.2.14.2.2). This reply must - // be sent: FreeRDP-based servers (for example GNOME Remote Desktop) block in - // their AWAIT_BW_RESULT state until they receive it and never proceed to - // licensing without it, so omitting it stalls the whole connection. We do not - // run a stateful connect-time measurement, so we report the payload the - // server handed us over a nominal interval; the figure is an informational - // QoS hint and the server proceeds on receipt. A precise measurement (timing - // the Start/Payload/Stop window) can refine the reported bandwidth later. - AutoDetectRequest::BandwidthMeasureStop { - sequence_number, - payload, - .. - } => { - let byte_count = payload - .as_ref() - .map_or(0, |p| u32::try_from(p.len()).unwrap_or(u32::MAX)); - let response = AutoDetectRspPdu::new(AutoDetectResponse::BandwidthMeasureResults { - sequence_number, - response_type: BW_RESULTS_CONNECT_TIME, - time_delta_ms: 1, - byte_count, - }); - let written = encode_send_data_request(user_channel_id, message_channel_id, &response, output)?; - Written::from_size(written) - } - // Bandwidth Measure Start and Payload carry no client reply, and the Network - // Characteristics Result is informational; nothing to send for those. - _ => Ok(Written::Nothing), - } -} - #[expect(single_use_lifetimes)] // anonymous lifetimes in `impl Trait` are unstable fn create_gcc_blocks<'a>( config: &Config, diff --git a/crates/ironrdp-connector/src/connection_activation.rs b/crates/ironrdp-connector/src/connection_activation.rs index 9e606fff87..672c310e4b 100644 --- a/crates/ironrdp-connector/src/connection_activation.rs +++ b/crates/ironrdp-connector/src/connection_activation.rs @@ -8,7 +8,7 @@ use tracing::{debug, warn}; use crate::{ Config, ConnectionFinalizationSequence, ConnectorError, ConnectorErrorExt as _, ConnectorResult, DesktopSize, - Sequence, State, Written, general_err, reason_err, + MonotonicInstant, Sequence, State, Written, general_err, reason_err, }; /// Represents the Capability Exchange and Connection Finalization phases @@ -125,7 +125,12 @@ impl Sequence for ConnectionActivationSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut ironrdp_core::WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + received_at: Option, + output: &mut ironrdp_core::WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { ConnectionActivationState::Consumed | ConnectionActivationState::Finalized { .. } => { return Err(general_err!( @@ -323,7 +328,7 @@ impl Sequence for ConnectionActivationSequence { } => { debug!("Connection Finalization"); - let written = connection_finalization.step(input, output)?; + let written = connection_finalization.step(input, received_at, output)?; let next_state = if !connection_finalization.state.is_terminal() { ConnectionActivationState::ConnectionFinalization { diff --git a/crates/ironrdp-connector/src/connection_finalization.rs b/crates/ironrdp-connector/src/connection_finalization.rs index 06c53ec9da..672a1494ac 100644 --- a/crates/ironrdp-connector/src/connection_finalization.rs +++ b/crates/ironrdp-connector/src/connection_finalization.rs @@ -8,7 +8,8 @@ use ironrdp_pdu::rdp::{finalization_messages, server_error_info}; use tracing::{debug, warn}; use crate::{ - ConnectorError, ConnectorErrorExt as _, ConnectorResult, Sequence, State, Written, general_err, reason_err, + ConnectorError, ConnectorErrorExt as _, ConnectorResult, MonotonicInstant, Sequence, State, Written, general_err, + reason_err, }; #[derive(Default, Debug, Copy, Clone)] @@ -87,7 +88,12 @@ impl Sequence for ConnectionFinalizationSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { ConnectionFinalizationState::Consumed => { return Err(general_err!( diff --git a/crates/ironrdp-connector/src/lib.rs b/crates/ironrdp-connector/src/lib.rs index 585dafb787..5173485e9a 100644 --- a/crates/ironrdp-connector/src/lib.rs +++ b/crates/ironrdp-connector/src/lib.rs @@ -355,15 +355,39 @@ impl Written { } } +/// A point on a monotonic millisecond clock owned by the I/O driver. +/// +/// Lives in `ironrdp-core`, shared with `ironrdp-rdpeudp`, and re-exported +/// here. The clock deliberately lives outside the sans-I/O sequences, because +/// a sequence reading a clock itself would measure how quickly it drained an +/// already-filled buffer rather than how long the bytes took to arrive. Only +/// the driver that performed the read knows the latter, and a driver whose +/// caller owns the read loop does not know it either: that is what the `None` +/// in [`Sequence::step`] is for, and why it stays. +pub use ironrdp_core::MonotonicInstant; + pub trait Sequence: Send { fn next_pdu_hint(&self) -> Option<&dyn PduHint>; fn state(&self) -> &dyn State; - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult; + /// Advances the sequence. + /// + /// `received_at` is when `input` arrived on the wire, as observed by the I/O + /// driver, or `None` from a driver that does not observe arrival times. The + /// absence of a reading is deliberately not expressible as an instant: a + /// driver that cannot measure has taken no measurement, which is a different + /// thing from one that measured no elapsed time, and only the sequence + /// knows which of the two its reply may be derived from. + fn step( + &mut self, + input: &[u8], + received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult; fn step_no_input(&mut self, output: &mut WriteBuf) -> ConnectorResult { - self.step(&[], output) + self.step(&[], None, output) } } diff --git a/crates/ironrdp-connector/src/license_exchange.rs b/crates/ironrdp-connector/src/license_exchange.rs index 8b77ec76a0..4ced01a52f 100644 --- a/crates/ironrdp-connector/src/license_exchange.rs +++ b/crates/ironrdp-connector/src/license_exchange.rs @@ -11,7 +11,9 @@ use rand::RngCore as _; use tracing::{debug, error, info, trace}; use super::{ConnectorError, ConnectorErrorExt as _, custom_err, general_err}; -use crate::{ConnectorResult, ConnectorResultExt as _, Sequence, State, Written, encode_send_data_request}; +use crate::{ + ConnectorResult, ConnectorResultExt as _, MonotonicInstant, Sequence, State, Written, encode_send_data_request, +}; #[derive(Default, Debug)] #[non_exhaustive] @@ -117,7 +119,12 @@ impl Sequence for LicenseExchangeSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { LicenseExchangeState::Consumed => { return Err(general_err!( diff --git a/crates/ironrdp-core/src/lib.rs b/crates/ironrdp-core/src/lib.rs index bbef57bdda..35a36c37a1 100644 --- a/crates/ironrdp-core/src/lib.rs +++ b/crates/ironrdp-core/src/lib.rs @@ -19,6 +19,7 @@ mod into_owned; #[cfg(feature = "alloc")] mod non_empty; mod padding; +mod time; #[cfg(feature = "alloc")] mod write_buf; @@ -48,5 +49,6 @@ pub use self::into_owned::IntoOwned; #[cfg(feature = "alloc")] pub use self::non_empty::NonEmpty; pub use self::padding::{read_padding, write_padding}; +pub use self::time::MonotonicInstant; #[cfg(feature = "alloc")] pub use self::write_buf::WriteBuf; diff --git a/crates/ironrdp-core/src/time.rs b/crates/ironrdp-core/src/time.rs new file mode 100644 index 0000000000..1383d46fe4 --- /dev/null +++ b/crates/ironrdp-core/src/time.rs @@ -0,0 +1,65 @@ +//! Monotonic time readings supplied by the caller. +//! +//! Sans-I/O state machines never read a clock themselves. They receive the +//! current instant on every call that can advance time, and report when they +//! next want to be woken. The clock deliberately lives outside the sequence +//! or connection type, because a state machine reading a clock itself would +//! measure how quickly it drained an already-filled buffer rather than when +//! the bytes actually arrived on the wire; only the I/O driver that performed +//! the read knows that. +//! +//! `std::time::Instant` is deliberately not used here. `Instant::now` panics +//! on `wasm32-unknown-unknown`, which some drivers using this type compile +//! for, and this crate is `no_std` besides. + +use core::ops::Add; +use core::time::Duration; + +/// A monotonic instant, in milliseconds, from an epoch chosen by the caller. +/// +/// The epoch is arbitrary and carries no meaning; only differences between two +/// instants do. Millisecond resolution is well below the shortest interval any +/// current caller needs to measure. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct MonotonicInstant(u64); + +impl MonotonicInstant { + /// Builds an instant from a monotonic millisecond reading. + #[must_use] + pub fn from_millis(milliseconds: u64) -> Self { + Self(milliseconds) + } + + /// The reading this instant was built from. + #[must_use] + pub fn as_millis(self) -> u64 { + self.0 + } + + /// The time elapsed since `earlier`, saturating at zero if the clock went + /// backwards or the arguments were transposed. + #[must_use] + pub fn duration_since(self, earlier: Self) -> Duration { + Duration::from_millis(self.0.saturating_sub(earlier.0)) + } + + /// This instant advanced by `duration`, saturating at the end of the + /// representable range. + /// + /// Saturating is the right behavior for the only caller: a deadline that + /// cannot be represented is one that never fires, which is what a timer set + /// beyond the end of time should do. + #[must_use] + pub fn saturating_add(self, duration: Duration) -> Self { + let milliseconds = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX); + Self(self.0.saturating_add(milliseconds)) + } +} + +impl Add for MonotonicInstant { + type Output = Self; + + fn add(self, rhs: Duration) -> Self { + self.saturating_add(rhs) + } +} diff --git a/crates/ironrdp-rdpeudp/src/time.rs b/crates/ironrdp-rdpeudp/src/time.rs index 20b5ac5b2c..5be4464bd1 100644 --- a/crates/ironrdp-rdpeudp/src/time.rs +++ b/crates/ironrdp-rdpeudp/src/time.rs @@ -4,61 +4,10 @@ //! instant on every call that can advance time, and reports when it next wants //! to be woken through [`RdpeudpConnection::poll_timeout`]. //! -//! `std::time::Instant` is deliberately not used here. `Instant::now` panics -//! on `wasm32-unknown-unknown`, which this crate compiles for, and a state -//! machine that read a clock itself would measure how quickly it drained an -//! already-filled buffer rather than when the datagram actually arrived. +//! [`MonotonicInstant`] itself lives in `ironrdp-core`, shared with +//! `ironrdp-connector`'s `Sequence::step`, and is re-exported here so +//! existing callers of this crate are unaffected. //! //! [`RdpeudpConnection::poll_timeout`]: crate::RdpeudpConnection::poll_timeout -use core::ops::Add; -use core::time::Duration; - -/// A monotonic instant, in milliseconds, from an epoch chosen by the caller. -/// -/// The epoch is arbitrary and carries no meaning; only differences between two -/// instants do. Millisecond resolution is well below the shortest interval the -/// protocol asks us to measure, the delayed-ACK timer of [MS-RDPEUDP2] 3.1.1. -#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] -pub struct MonotonicInstant(u64); - -impl MonotonicInstant { - /// Builds an instant from a monotonic millisecond reading. - #[must_use] - pub fn from_millis(milliseconds: u64) -> Self { - Self(milliseconds) - } - - /// The reading this instant was built from. - #[must_use] - pub fn as_millis(self) -> u64 { - self.0 - } - - /// The time elapsed since `earlier`, saturating at zero if the clock went - /// backwards or the arguments were transposed. - #[must_use] - pub fn duration_since(self, earlier: Self) -> Duration { - Duration::from_millis(self.0.saturating_sub(earlier.0)) - } - - /// This instant advanced by `duration`, saturating at the end of the - /// representable range. - /// - /// Saturating is the right behavior for the only caller: a deadline that - /// cannot be represented is one that never fires, which is what a timer set - /// beyond the end of time should do. - #[must_use] - pub fn saturating_add(self, duration: Duration) -> Self { - let milliseconds = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX); - Self(self.0.saturating_add(milliseconds)) - } -} - -impl Add for MonotonicInstant { - type Output = Self; - - fn add(self, rhs: Duration) -> Self { - self.saturating_add(rhs) - } -} +pub use ironrdp_core::MonotonicInstant; diff --git a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs index ed4e62a618..804d1ad1d7 100644 --- a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs @@ -9,11 +9,12 @@ use std::borrow::Cow; +use ironrdp_connector::MonotonicInstant; use ironrdp_connector::{ClientConnector, ClientConnectorState, Credentials, DesktopSize, Sequence as _, Written}; use ironrdp_core::{WriteBuf, encode_vec}; use ironrdp_pdu::gcc; use ironrdp_pdu::mcs::{McsMessage, SendDataIndication}; -use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest}; +use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu}; use ironrdp_pdu::rdp::capability_sets::MajorPlatformType; use ironrdp_pdu::rdp::headers::{BasicSecurityHeader, BasicSecurityHeaderFlags}; use ironrdp_pdu::rdp::server_license::{ @@ -104,7 +105,7 @@ fn connect_time_autodetect_request_is_answered_and_phase_continues() { let frame = server_send_data_indication(MESSAGE_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - let written = connector.step(&frame, &mut output).unwrap(); + let written = connector.step(&frame, None, &mut output).unwrap(); assert!(written.size().is_some(), "an RTT request must produce a response frame"); assert!( @@ -127,7 +128,7 @@ fn unrelated_message_channel_pdu_is_ignored_and_phase_continues() { let frame = server_send_data_indication(MESSAGE_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - let written = connector.step(&frame, &mut output).unwrap(); + let written = connector.step(&frame, None, &mut output).unwrap(); assert_eq!( written, @@ -166,7 +167,7 @@ fn first_licensing_pdu_leaves_autodetect_for_the_licensing_path() { let frame = server_send_data_indication(IO_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); assert!( matches!( @@ -193,7 +194,7 @@ fn connect_time_bandwidth_measure_stop_is_answered_and_phase_continues() { let frame = server_send_data_indication(MESSAGE_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - let written = connector.step(&frame, &mut output).unwrap(); + let written = connector.step(&frame, None, &mut output).unwrap(); assert!( written.size().is_some(), @@ -204,3 +205,262 @@ fn connect_time_bandwidth_measure_stop_is_answered_and_phase_continues() { "the connector keeps listening after answering the bandwidth measurement" ); } + +/// Unwrap a Bandwidth Measure Results response and return `(time_delta_ms, byte_count)`. +/// +/// The response frame is X224 > MCS SendDataRequest > Auto-Detect Response PDU. +fn decode_bandwidth_results(output: &WriteBuf) -> (u32, u32) { + let X224(McsMessage::SendDataRequest(send_data)) = ironrdp_core::decode(output.filled()).unwrap() else { + panic!("expected a SendDataRequest in the response frame"); + }; + + let response = ironrdp_core::decode::(&send_data.user_data).unwrap(); + match response.response { + AutoDetectResponse::BandwidthMeasureResults { + time_delta_ms, + byte_count, + .. + } => (time_delta_ms, byte_count), + other => panic!("expected BandwidthMeasureResults, got {other:?}"), + } +} + +/// A Stop with no preceding Start has nothing to have measured. It is still +/// answered, because the server blocks without a reply, but the interval reported +/// is the unmeasurable floor rather than an invented figure. +#[test] +fn connect_time_bandwidth_stop_without_start_reports_the_floor() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + let stop = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_stop_connect_time( + 0x2222, + vec![0u8; 512], + ))) + .unwrap(); + let written = connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), + Some(MonotonicInstant::from_millis(9_999)), + &mut output, + ) + .unwrap(); + + assert!(written.size().is_some(), "the server still needs its reply"); + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 1, "no window was open, so no interval was measured"); +} + +/// Start, Payload and Stop delivered by one socket read carry the same arrival +/// time, so the elapsed time rounds down to nothing. TCP coalescing makes this the +/// common case on a fast link, since the connect-time payload is small. +/// +/// Reporting `timeDelta` of 0 would divide out to an unbounded bandwidth for a +/// server computing `byteCount * 8 / timeDelta`, so the floor is reported. Every +/// byte is still counted, per [MS-RDPBCGR] 3.2.5.14: the window was timed, and the +/// bytes really did arrive inside that millisecond, so the floor bounds a real +/// measurement instead of standing in for a missing one. +#[test] +fn connect_time_bandwidth_coalesced_into_one_read_reports_the_floor() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + // One instant for all three, which is what `Framed` hands down when a single + // read filled the buffer that all three PDUs were then extracted from. + let arrival = Some(MonotonicInstant::from_millis(7_000)); + + for request in [ + AutoDetectRequest::bw_start_connect_time(0x3333), + AutoDetectRequest::bw_payload(0x3333, vec![0u8; 1024]), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + arrival, + &mut output, + ) + .unwrap(); + } + + output.clear(); + let stop = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_stop_connect_time( + 0x3333, + vec![0u8; 512], + ))) + .unwrap(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), + arrival, + &mut output, + ) + .unwrap(); + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 1, "a window that arrived in one read floors to 1 ms"); + assert_eq!( + results.1, 1552, + "every byte in the timed window is still counted, plus the 8-byte header on each of the Payload and Stop" + ); +} + +/// A window can open on one driver and close on another, for example when a +/// `Framed` is rebuilt with leftover bytes between the Start and this Stop: the +/// rebuilt `Framed` starts with no arrival time of its own, so the Stop reports +/// `None` even though the Start that opened the window reported `Some`. Nothing +/// upstream of this function stops that from happening, so it is reachable even +/// though it did not use to be exercised by a test. +/// +/// The window's accumulated bytes are dropped in that case: reporting them +/// against `timeDelta = UNMEASURABLE_INTERVAL_MS` would pair a byte count that +/// arrived over the window's real duration with a floor timer that understates +/// it, inflating the reported bandwidth the same way an uncounted header would. +#[test] +fn connect_time_bandwidth_stop_with_no_arrival_time_drops_the_open_window() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + let arrival = Some(MonotonicInstant::from_millis(9_000)); + + for request in [ + AutoDetectRequest::bw_start_connect_time(0x7777), + AutoDetectRequest::bw_payload(0x7777, vec![0u8; 2048]), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + arrival, + &mut output, + ) + .unwrap(); + } + + output.clear(); + let stop = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_stop_connect_time( + 0x7777, + vec![0u8; 512], + ))) + .unwrap(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), + None, + &mut output, + ) + .unwrap(); + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 1, "an unmeasured window still floors to 1 ms"); + assert_eq!( + results.1, 520, + "the 2048-byte Payload is dropped along with the window; only the Stop's own 512 bytes plus its 8-byte header remain" + ); +} + +/// A driver with no clock reports no arrival time for this step, so no window is +/// ever opened and there is nothing to accumulate into. The FFI driver is in this +/// position for the whole connection; the wasm32 driver only for the single +/// x224_connection_response step. +/// +/// The server still blocks without a reply, so one is sent, but it claims no more +/// than this Stop's own payload. Counting the Payload messages here would pair a +/// full byte count with a `timeDelta` the client never measured, which yields a +/// bandwidth figure that grows with however much the server chose to send. +#[test] +fn connect_time_bandwidth_without_a_clock_reports_the_stop_payload_alone() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + for request in [ + AutoDetectRequest::bw_start_connect_time(0x5555), + AutoDetectRequest::bw_payload(0x5555, vec![0u8; 1024]), + AutoDetectRequest::bw_stop_connect_time(0x5555, vec![0u8; 512]), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + None, + &mut output, + ) + .unwrap(); + } + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 1, "a driver with no clock measured no interval"); + assert_eq!( + results.1, 520, + "the 1024-byte Payload is not counted, since no window was open; the Stop's own 512 bytes plus its 8-byte header are" + ); +} + +/// Payload messages accumulate across a timed window, and the total they reach is +/// what the Stop reports. This is the path the floor case deliberately skips, so it +/// needs its own coverage: without it, nothing would catch an accumulator that had +/// stopped adding. +#[test] +fn connect_time_bandwidth_measured_window_reports_every_payload() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + // Each message lands in its own read, a millisecond apart, so the window is + // timed and the accumulator is what decides the reported total. + for (request, arrival) in [ + (AutoDetectRequest::bw_start_connect_time(0x4444), 2_000), + (AutoDetectRequest::bw_payload(0x4444, vec![0u8; 2048]), 2_100), + (AutoDetectRequest::bw_payload(0x4444, vec![0u8; 1024]), 2_200), + (AutoDetectRequest::bw_stop_connect_time(0x4444, vec![0u8; 512]), 2_250), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + Some(MonotonicInstant::from_millis(arrival)), + &mut output, + ) + .unwrap(); + } + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 250, "interval is Stop arrival minus Start arrival"); + assert_eq!( + results.1, 3608, + "both Payload messages and the Stop payload are counted, each with its 8-byte header" + ); +} + +#[test] +fn connect_time_bandwidth_second_start_discards_the_first_window() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + // [MS-RDPBCGR] 3.2.5.14 has the client clear both stores and restart the + // timer on each Bandwidth Measure Start, so the 4096 bytes counted into the + // abandoned window must not survive into the reported total, and the + // interval must be measured from the second Start rather than the first. + for (request, arrival) in [ + (AutoDetectRequest::bw_start_connect_time(0x5555), 1_000), + (AutoDetectRequest::bw_payload(0x5555, vec![0u8; 4096]), 1_100), + (AutoDetectRequest::bw_start_connect_time(0x5555), 2_000), + (AutoDetectRequest::bw_payload(0x5555, vec![0u8; 1024]), 2_100), + (AutoDetectRequest::bw_stop_connect_time(0x5555, vec![0u8; 512]), 2_500), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + Some(MonotonicInstant::from_millis(arrival)), + &mut output, + ) + .unwrap(); + } + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 500, "interval runs from the second Start, not the first"); + assert_eq!( + results.1, 1552, + "the 4096 bytes counted before the second Start are discarded; only the second window's Payload and Stop, each with an 8-byte header, remain" + ); +} diff --git a/crates/ironrdp-testsuite-core/tests/server/acceptor.rs b/crates/ironrdp-testsuite-core/tests/server/acceptor.rs index 0d5d76f79b..ed95dd60f7 100644 --- a/crates/ironrdp-testsuite-core/tests/server/acceptor.rs +++ b/crates/ironrdp-testsuite-core/tests/server/acceptor.rs @@ -37,12 +37,12 @@ fn neg_failure_on_protocol_mismatch() { // Step 1: feed the connection request (HYBRID | HYBRID_EX, no SSL) let request_bytes = encode_connection_request(SecurityProtocol::HYBRID | SecurityProtocol::HYBRID_EX); let mut output = WriteBuf::new(); - let written = acceptor.step(&request_bytes, &mut output).unwrap(); + let written = acceptor.step(&request_bytes, None, &mut output).unwrap(); assert!(matches!(written, Written::Nothing)); // Step 2: acceptor tries to send confirm, finds no common protocol let mut output = WriteBuf::new(); - let result = acceptor.step(&[], &mut output); + let result = acceptor.step(&[], None, &mut output); // Must be an error assert!(result.is_err(), "expected error on protocol mismatch"); @@ -78,10 +78,10 @@ fn neg_success_when_protocols_match() { let request_bytes = encode_connection_request(SecurityProtocol::SSL | SecurityProtocol::HYBRID); let mut output = WriteBuf::new(); - acceptor.step(&request_bytes, &mut output).unwrap(); + acceptor.step(&request_bytes, None, &mut output).unwrap(); let mut output = WriteBuf::new(); - let written = acceptor.step(&[], &mut output).unwrap(); + let written = acceptor.step(&[], None, &mut output).unwrap(); assert!(!matches!(written, Written::Nothing)); let response_bytes = output.filled(); @@ -119,8 +119,8 @@ fn message_channel_advertised_when_client_requests_it() { // Connection request -> confirm -> (TLS upgrade) -> ready for ConnectInitial. let request_bytes = encode_connection_request(SecurityProtocol::SSL); - acceptor.step(&request_bytes, &mut WriteBuf::new()).unwrap(); - acceptor.step(&[], &mut WriteBuf::new()).unwrap(); + acceptor.step(&request_bytes, None, &mut WriteBuf::new()).unwrap(); + acceptor.step(&[], None, &mut WriteBuf::new()).unwrap(); acceptor.mark_security_upgrade_as_done(); // Client GCC with the message channel block and no network channels, so the @@ -132,10 +132,10 @@ fn message_channel_advertised_when_client_requests_it() { let mut initial_buf = WriteBuf::new(); encode_x224_packet(&connect_initial, &mut initial_buf).unwrap(); - acceptor.step(initial_buf.filled(), &mut WriteBuf::new()).unwrap(); + acceptor.step(initial_buf.filled(), None, &mut WriteBuf::new()).unwrap(); let mut output = WriteBuf::new(); - acceptor.step(&[], &mut output).unwrap(); + acceptor.step(&[], None, &mut output).unwrap(); let payload = decode::>>(output.filled()).unwrap().0; let response = decode::(payload.data.as_ref()).unwrap(); @@ -171,10 +171,10 @@ fn neg_failure_hybrid_required() { let request_bytes = encode_connection_request(SecurityProtocol::SSL); let mut output = WriteBuf::new(); - acceptor.step(&request_bytes, &mut output).unwrap(); + acceptor.step(&request_bytes, None, &mut output).unwrap(); let mut output = WriteBuf::new(); - let result = acceptor.step(&[], &mut output); + let result = acceptor.step(&[], None, &mut output); assert!(result.is_err()); let response_bytes = output.filled(); diff --git a/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs b/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs index 33c0da3cb9..70b05f57e6 100644 --- a/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs +++ b/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs @@ -107,7 +107,7 @@ fn demand_active_static_channel_chunk_size(chunk_size: Option) -> usize { let mut output = WriteBuf::new(); let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(demand_active)); sequence - .step(&frame, &mut output) + .step(&frame, None, &mut output) .expect("demand active should be accepted"); match sequence.connection_activation_state() { @@ -154,7 +154,7 @@ fn deactivate_all_during_capabilities_exchange_stays_in_same_state() { let frame = encode_server_share_control(ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll)); let mut output = WriteBuf::new(); - let written = seq.step(&frame, &mut output).unwrap(); + let written = seq.step(&frame, None, &mut output).unwrap(); assert_eq!(written, Written::Nothing); assert!( @@ -177,7 +177,7 @@ fn client_connector_stays_in_capabilities_exchange_on_deactivate_all() { let frame = encode_server_share_control(ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll)); let mut output = WriteBuf::new(); - let written = connector.step(&frame, &mut output).unwrap(); + let written = connector.step(&frame, None, &mut output).unwrap(); assert_eq!(written, Written::Nothing); assert!( @@ -207,7 +207,7 @@ fn set_error_info_during_capabilities_exchange_surfaces_the_disconnect_reason() let mut output = WriteBuf::new(); let err = seq - .step(&frame, &mut output) + .step(&frame, None, &mut output) .expect_err("a Set Error Info PDU during capabilities exchange must end the sequence with an error"); let message = err.to_string(); @@ -240,7 +240,7 @@ fn none_error_info_during_capabilities_exchange_is_skipped() { let frame = encode_server_share_control(none_error_info); let mut output = WriteBuf::new(); - let written = seq.step(&frame, &mut output).unwrap(); + let written = seq.step(&frame, None, &mut output).unwrap(); assert_eq!(written, Written::Nothing); assert!( @@ -260,13 +260,13 @@ fn demand_active_after_deactivate_all_transitions_to_connection_finalization() { // First: feed DeactivateAll let deactivate_frame = encode_server_share_control(ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll)); - let written = seq.step(&deactivate_frame, &mut output).unwrap(); + let written = seq.step(&deactivate_frame, None, &mut output).unwrap(); assert_eq!(written, Written::Nothing); // Then: feed ServerDemandActive let demand_active_frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(SERVER_DEMAND_ACTIVE.clone())); - let written = seq.step(&demand_active_frame, &mut output).unwrap(); + let written = seq.step(&demand_active_frame, None, &mut output).unwrap(); assert!(written != Written::Nothing, "should have written ClientConfirmActive"); assert!( @@ -287,7 +287,7 @@ fn demand_active_captures_server_input_flags() { let mut output = WriteBuf::new(); let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(SERVER_DEMAND_ACTIVE.clone())); - seq.step(&frame, &mut output).unwrap(); + seq.step(&frame, None, &mut output).unwrap(); match seq.connection_activation_state() { ConnectionActivationState::ConnectionFinalization { input_flags, .. } => { @@ -316,7 +316,7 @@ fn demand_active_without_input_capability_yields_empty_input_flags() { .retain(|c| !matches!(c, CapabilitySet::Input(_))); let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(demand_active)); - seq.step(&frame, &mut output).unwrap(); + seq.step(&frame, None, &mut output).unwrap(); match seq.connection_activation_state() { ConnectionActivationState::ConnectionFinalization { input_flags, .. } => { @@ -394,7 +394,7 @@ fn multitransport_request_is_surfaced_without_waiting_for_another_pdu() { ); let mut output = WriteBuf::new(); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); assert!( connector.should_perform_multitransport(), @@ -416,7 +416,7 @@ fn responding_returns_to_bootstrapping_for_the_next_request() { &multitransport_request(request_id, RequestedProtocol::UdpFecR), MESSAGE_CHANNEL_ID, ); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); assert!(connector.should_perform_multitransport()); assert_eq!(connector.multitransport_request().unwrap().request_id, request_id); @@ -445,7 +445,7 @@ fn third_multitransport_request_is_rejected() { &multitransport_request(request_id, RequestedProtocol::UdpFecR), MESSAGE_CHANNEL_ID, ); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); connector .complete_multitransport(MultitransportResult::Success, &mut output) .unwrap(); @@ -455,7 +455,7 @@ fn third_multitransport_request_is_rejected() { &multitransport_request(3, RequestedProtocol::UdpFecR), MESSAGE_CHANNEL_ID, ); - assert!(connector.step(&frame, &mut output).is_err()); + assert!(connector.step(&frame, None, &mut output).is_err()); } #[test] @@ -466,7 +466,7 @@ fn demand_active_on_the_io_channel_ends_bootstrapping() { let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(SERVER_DEMAND_ACTIVE.clone())); let mut output = WriteBuf::new(); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); assert!( matches!(connector.state, ClientConnectorState::ConnectionFinalization { .. }), diff --git a/crates/ironrdp-testsuite-extra/tests/async_framed.rs b/crates/ironrdp-testsuite-extra/tests/async_framed.rs new file mode 100644 index 0000000000..bcde3ac1e1 --- /dev/null +++ b/crates/ironrdp-testsuite-extra/tests/async_framed.rs @@ -0,0 +1,166 @@ +//! Arrival-time stamping on [`Framed`]. +//! +//! The connect-time bandwidth measurement in `ironrdp-connector` is only as +//! honest as `Framed::last_read_at`, and nothing else pins that down: the +//! connector tests drive `Sequence::step` with hand-picked instants, so a +//! regression that stopped stamping reads would leave them green while every +//! measured window silently collapsed to the unmeasurable floor. +//! +//! These live here rather than inline in `ironrdp-async` because that crate +//! sets `[lib] test = false`, so an inline `#[cfg(test)]` module would compile +//! and never run. + +use core::pin::Pin; +use core::task::{Context, Poll, Waker}; +use core::time::Duration; +use std::collections::VecDeque; +use std::io; + +use ironrdp_async::bytes::BytesMut; +use ironrdp_async::{Framed, FramedRead, StreamWrapper}; + +/// Drives a future to completion on the current thread. The mock below never +/// yields, so polling in a loop is enough and saves pulling in a runtime. +fn block_on(fut: F) -> F::Output { + let mut fut = core::pin::pin!(fut); + let mut cx = Context::from_waker(Waker::noop()); + loop { + if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { + return value; + } + } +} + +/// A stream that hands over pre-arranged chunks, one per read, so a test can +/// decide exactly which PDUs share a socket read and which get their own. +struct ChunkedStream { + chunks: VecDeque>, + /// Delay applied before each read completes, so consecutive reads land on + /// distinguishable instants rather than relying on clock resolution. + delay: Duration, +} + +impl ChunkedStream { + fn new(chunks: impl IntoIterator>) -> Self { + Self { + chunks: chunks.into_iter().collect(), + delay: Duration::from_millis(5), + } + } +} + +impl StreamWrapper for ChunkedStream { + type InnerStream = Self; + + fn from_inner(stream: Self::InnerStream) -> Self { + stream + } + + fn into_inner(self) -> Self::InnerStream { + self + } + + fn get_inner(&self) -> &Self::InnerStream { + self + } + + fn get_inner_mut(&mut self) -> &mut Self::InnerStream { + self + } +} + +impl FramedRead for ChunkedStream { + type ReadFut<'read> + = Pin> + 'read>> + where + Self: 'read; + + fn read<'a>(&'a mut self, buf: &'a mut BytesMut) -> Self::ReadFut<'a> { + Box::pin(async move { + std::thread::sleep(self.delay); + match self.chunks.pop_front() { + Some(chunk) => { + buf.extend_from_slice(&chunk); + Ok(chunk.len()) + } + None => Ok(0), + } + }) + } +} + +/// Smallest frame `ironrdp_pdu::find_size` will accept: a TPKT header whose +/// length field covers itself plus `payload`. +fn tpkt(payload: &[u8]) -> Vec { + let length = u16::try_from(4 + payload.len()).expect("frame fits"); + let mut frame = vec![0x03, 0x00]; + frame.extend_from_slice(&length.to_be_bytes()); + frame.extend_from_slice(payload); + frame +} + +#[test] +fn each_socket_read_advances_the_arrival_time() { + // One frame per read, so each PDU is stamped by the read that carried it. + let mut framed = Framed::::new(ChunkedStream::new([tpkt(&[0xAA; 8]), tpkt(&[0xBB; 8])])); + + assert!( + framed.last_read_at().is_none(), + "an unread Framed has observed no arrival" + ); + + block_on(framed.read_pdu()).expect("first frame"); + let first = framed.last_read_at().expect("host build observes time"); + + block_on(framed.read_pdu()).expect("second frame"); + let second = framed.last_read_at().expect("host build observes time"); + + assert!( + second.duration_since(first) >= Duration::from_millis(2), + "the second read must stamp a later arrival than the first, got {:?}", + second.duration_since(first) + ); +} + +#[test] +fn pdus_sharing_a_socket_read_share_its_arrival_time() { + // Both frames arrive in one read. The second is served from the buffer, so + // it arrived when that read completed, not when the caller drained it. + let mut chunk = tpkt(&[0xAA; 8]); + chunk.extend_from_slice(&tpkt(&[0xBB; 8])); + let mut framed = Framed::::new(ChunkedStream::new([chunk])); + + block_on(framed.read_pdu()).expect("first frame"); + let first = framed.last_read_at().expect("host build observes time"); + + // Wait before draining the second one. `MonotonicInstant` counts whole + // milliseconds, so without this the two drains would be indistinguishable + // and the assertion below would hold even if the stamp were taken on drain + // rather than on the read. + std::thread::sleep(Duration::from_millis(10)); + + block_on(framed.read_pdu()).expect("second frame"); + let second = framed.last_read_at().expect("host build observes time"); + + assert_eq!( + first, second, + "a PDU served from the buffer keeps the arrival time of the read that filled it" + ); +} + +#[test] +fn leftover_carried_into_a_new_framed_has_no_arrival_time() { + // `ironrdp-tokio`'s split/unsplit helpers rebuild a `Framed` around bytes + // that were read by the previous one. Those bytes did arrive, but not on + // this `Framed`, and it has no way to know when: reporting an arrival here + // would be inventing one. + let leftover = BytesMut::from(&*tpkt(&[0xDD; 8])); + let mut framed = Framed::::new_with_leftover(ChunkedStream::new([]), leftover); + + block_on(framed.read_pdu()).expect("frame served entirely from leftover"); + + assert!( + framed.last_read_at().is_none(), + "a frame served from leftover was never read by this Framed, so it has no arrival time" + ); +} diff --git a/crates/ironrdp-testsuite-extra/tests/main.rs b/crates/ironrdp-testsuite-extra/tests/main.rs index f24df9ccd4..7535c6bfc6 100644 --- a/crates/ironrdp-testsuite-extra/tests/main.rs +++ b/crates/ironrdp-testsuite-extra/tests/main.rs @@ -2,6 +2,7 @@ #![allow(clippy::unwrap_used, reason = "unwrap is fine in tests")] mod agent; +mod async_framed; mod capture_helpers; mod client_config; mod dvc_pipe_proxy; diff --git a/crates/ironrdp-web/src/session.rs b/crates/ironrdp-web/src/session.rs index 1093fe545a..c77670f9d5 100644 --- a/crates/ironrdp-web/src/session.rs +++ b/crates/ironrdp-web/src/session.rs @@ -1866,7 +1866,7 @@ where debug_assert!(connector.next_pdu_hint().is_some()); buf.clear(); - let written = connector.step(x224_connection_response.as_bytes(), &mut buf)?; + let written = connector.step(x224_connection_response.as_bytes(), None, &mut buf)?; debug_assert!(written.is_nothing()); let should_upgrade = ironrdp_futures::skip_connect_begin(connector); diff --git a/ffi/src/connector/activation.rs b/ffi/src/connector/activation.rs index bd1db7473d..d6f43aa9c0 100644 --- a/ffi/src/connector/activation.rs +++ b/ffi/src/connector/activation.rs @@ -27,7 +27,7 @@ pub mod ffi { } pub fn step(&mut self, pdu_hint: &[u8], buf: &mut WriteBuf) -> Result, Box> { - let res = self.0.step(pdu_hint, &mut buf.0).map(Written).map(Box::new)?; + let res = self.0.step(pdu_hint, None, &mut buf.0).map(Written).map(Box::new)?; Ok(res) } diff --git a/ffi/src/connector/mod.rs b/ffi/src/connector/mod.rs index 055ced5737..e13df8dc3a 100644 --- a/ffi/src/connector/mod.rs +++ b/ffi/src/connector/mod.rs @@ -193,7 +193,7 @@ pub mod ffi { let Some(connector) = self.0.as_mut() else { return Err(ValueConsumedError::for_item("connector").into()); }; - let written = connector.step(input, &mut write_buf.0)?; + let written = connector.step(input, None, &mut write_buf.0)?; Ok(Box::new(Written(written))) }