diff --git a/crates/ironrdp-async/src/framed.rs b/crates/ironrdp-async/src/framed.rs index ee3a91d773..c25a86b513 100644 --- a/crates/ironrdp-async/src/framed.rs +++ b/crates/ironrdp-async/src/framed.rs @@ -317,16 +317,19 @@ where Ok(()) } -/// Reads the driver-owned monotonic clock. +/// Reads the monotonic clock behind [`Framed::last_read_at`]. /// -/// The epoch is the first call; only differences are meaningful. +/// The epoch is the first call; only differences are meaningful. Transports that +/// carry part of the same session elsewhere, such as a UDP sideband, can stamp +/// their reads with this clock so the session compares them with frames read +/// through [`Framed`]. /// /// `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 { +pub 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-client/src/lib.rs b/crates/ironrdp-client/src/lib.rs index fe4b112898..532ca29fc7 100644 --- a/crates/ironrdp-client/src/lib.rs +++ b/crates/ironrdp-client/src/lib.rs @@ -16,3 +16,6 @@ pub mod rdp; mod clipboard; mod ws; + +#[cfg(feature = "udp")] +mod udp; diff --git a/crates/ironrdp-client/src/rdp.rs b/crates/ironrdp-client/src/rdp.rs index 6e81254c76..04c7b48b5e 100644 --- a/crates/ironrdp-client/src/rdp.rs +++ b/crates/ironrdp-client/src/rdp.rs @@ -39,6 +39,7 @@ use ironrdp_pdu::input::mouse::PointerFlags; all(windows, feature = "webauthn") ))] use ironrdp_pdu::pdu_other_err; +use ironrdp_pdu::rdp::autodetect::AutoDetectRequest; use ironrdp_pdu::rdp::multitransport::MultitransportResponsePdu; use ironrdp_pdu::rdp::session_info::ServerAutoReconnect; #[cfg(feature = "rdpdr")] @@ -83,8 +84,18 @@ use ironrdp_rdpsnd_native::{RdpeaiCaptureBackend, cpal}; use crate::config::{Config, RDCleanPathConfig, Transport}; use crate::rail::{RailClient, RailControlEvent, RailEvent, RailInputEvent}; +#[cfg(feature = "udp")] +use crate::udp::disable_failed_tunnel; use ironrdp_rail::pdu::{ExecutePdu, ExecuteResultPdu}; +/// A Tunnel Data PDU received on the reliable UDP tunnel, ready for the session. +struct ReceivedTunnelMessage { + auto_detect_requests: Vec, + payload: Vec, + /// When the transport read the PDU, on the clock of the TCP reader's `last_read_at`. + received_at: Option, +} + // ── Public event types ──────────────────────────────────────────────────────── /// Explains why a dynamic display update must reconnect instead of completing in-session. @@ -3189,7 +3200,8 @@ async fn active_session( Err(error) => return Err(ironrdp_session::custom_err!("read frame", error)), }; trace!(?action, frame_length = payload.len(), "Frame received"); - let mut outputs = active_stage.process(&mut image, action, &payload)?; + let mut outputs = + active_stage.process_with_timestamp(&mut image, action, &payload, reader.last_read_at())?; #[cfg(feature = "rdpdr")] if let Some(output) = poll_deferred_rdpdr_output(&mut active_stage)? { outputs.push(output); @@ -3251,42 +3263,84 @@ async fn active_session( } ActiveSessionIteration::outputs(outputs) } - udp_payload = async { + udp_message = async { #[cfg(feature = "udp")] { match (udp_tunnel.transport.as_mut(), pending_udp_payload.is_none()) { - (Some(transport), true) => transport.recv().await, + (Some(transport), true) => match transport.recv_message().await { + // The read time comes from the same clock as the TCP reader's + // `last_read_at`, so a bandwidth window can span both transports. + Some(message) => Some(ReceivedTunnelMessage { + auto_detect_requests: message.auto_detect_requests(), + payload: message.data, + received_at: transport.last_received_at(), + }), + None => None, + }, (Some(_), false) | (None, _) => { - core::future::pending::>>().await + core::future::pending::>().await } } } #[cfg(not(feature = "udp"))] { - core::future::pending::>>().await + core::future::pending::>().await } } => { - match udp_payload { + match udp_message { None => { - if active_stage.reliable_udp_dvc_tunnel_in_use() { - return Ok(RdpControlFlow::TransportFailure( - ironrdp_session::general_err!("reliable UDP tunnel closed"), - )); - } #[cfg(feature = "udp")] { - udp_tunnel.transport = None; + if let Err(error) = disable_failed_tunnel( + &mut active_stage, + &mut udp_tunnel.transport, + ironrdp_session::general_err!("reliable UDP tunnel closed"), + ) { + return Ok(RdpControlFlow::TransportFailure(error)); + } } - active_stage.disable_reliable_udp_dvc_tunnel()?; - warn!("Reliable UDP tunnel closed before Soft-Sync; continuing with TCP"); ActiveSessionIteration::outputs(Vec::new()) } - Some(payload) if payload.is_empty() => { - trace!("Ignoring reliable UDP tunnel PDU without higher-layer data"); - ActiveSessionIteration::outputs(Vec::new()) - } - Some(payload) => { - if active_stage.reliable_udp_dvc_tunnel_in_use() { + Some(ReceivedTunnelMessage { auto_detect_requests, payload, received_at }) => { + #[cfg(feature = "udp")] + { + let responses = active_stage.process_tunnel_auto_detect( + auto_detect_requests, + payload.len(), + received_at, + ); + let reply = ironrdp_rdpeudp_tokio::TunnelMessage::from_auto_detect_responses(&responses); + if let Some(reply) = reply + && let Some(transport) = udp_tunnel.transport.as_ref() + { + let Some(result) = + cancelable_operation(transport.send_message(reply), close_receiver).await + else { + return Ok(RdpControlFlow::TerminatedGracefully( + GracefulDisconnectReason::UserInitiated, + )); + }; + if let Err(error) = result { + if let Err(error) = disable_failed_tunnel( + &mut active_stage, + &mut udp_tunnel.transport, + ironrdp_session::custom_err!("answer reliable UDP tunnel auto-detect", error), + ) { + return Ok(RdpControlFlow::TransportFailure(error)); + } + // No channel migrated, and the tunnel is now withdrawn, so + // nothing will read this PDU's data. + continue; + } + } + } + #[cfg(not(feature = "udp"))] + let _ = (auto_detect_requests, received_at); + + if payload.is_empty() { + trace!("Reliable UDP tunnel PDU without higher-layer data"); + ActiveSessionIteration::outputs(Vec::new()) + } else if active_stage.reliable_udp_dvc_tunnel_in_use() { ActiveSessionIteration::tunnel( SoftSyncTunnelType::RELIABLE_UDP, active_stage.process_dvc_tunnel( diff --git a/crates/ironrdp-client/src/udp.rs b/crates/ironrdp-client/src/udp.rs new file mode 100644 index 0000000000..aec2dea977 --- /dev/null +++ b/crates/ironrdp-client/src/udp.rs @@ -0,0 +1,22 @@ +//! Internal client handling for a reliable UDP tunnel. + +use ironrdp_rdpeudp_tokio::UdpTransport; +use ironrdp_session::{ActiveStage, SessionError, SessionResult}; +use tracing::warn; + +/// Drops a failed sideband and withdraws it from future Soft-Sync negotiation. +/// A channel already moved to the tunnel cannot resume over TCP without reconnecting. +/// Both receive closure and auto-detect send failures use this policy. +pub(crate) fn disable_failed_tunnel( + stage: &mut ActiveStage, + transport: &mut Option, + error: SessionError, +) -> SessionResult<()> { + if stage.reliable_udp_dvc_tunnel_in_use() { + return Err(error); + } + *transport = None; + stage.disable_reliable_udp_dvc_tunnel()?; + warn!(%error, "Reliable UDP tunnel failed before channel migration; continuing with TCP"); + Ok(()) +} diff --git a/crates/ironrdp-rdpeudp-tokio/src/autodetect.rs b/crates/ironrdp-rdpeudp-tokio/src/autodetect.rs new file mode 100644 index 0000000000..bae5097f2f --- /dev/null +++ b/crates/ironrdp-rdpeudp-tokio/src/autodetect.rs @@ -0,0 +1,64 @@ +//! Auto-detect messages carried in tunnel sub-headers. +//! +//! On a sideband channel in use, the server sends its auto-detect requests in the sub-headers of +//! Tunnel Data PDUs, and the client answers there ([MS-RDPBCGR] 3.2.5.14). Each sub-header is the +//! auto-detect structure itself: SubHeaderLength and SubHeaderType are its headerLength and +//! headerTypeId ([MS-RDPEMT] 2.2.1.1.1). +//! +//! [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/16ffa852-8aa7-481c-99a0-36c1a9a198f6 +//! [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad + +use ironrdp_core::{Decode, Encode}; +use ironrdp_pdu::rdp::autodetect::{AutoDetectRequest, AutoDetectResponse}; +use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader}; +use tracing::debug; + +use crate::transport::TunnelMessage; + +impl TunnelMessage { + /// Decodes the auto-detect requests among this message's sub-headers. + /// + /// Sub-headers of other types and requests that do not decode are skipped. RTT requests + /// are returned along with the bandwidth requests MS-RDPEMT lists, because Windows sends + /// them on the tunnel too, although MS-RDPEMT does not specify that encapsulation. + pub fn auto_detect_requests(&self) -> Vec { + self.sub_headers + .iter() + .filter(|sub_header| sub_header.sub_header_type == SubHeaderType::AutoDetectRequest) + .filter_map(|sub_header| match redecode(sub_header) { + Ok(request) => Some(request), + Err(error) => { + debug!(%error, data = ?sub_header.data, "Ignoring an undecodable auto-detect request on the tunnel"); + None + } + }) + .collect() + } + + /// Builds a message with no higher-layer data that carries `responses` as sub-headers. + /// + /// A response that cannot be encoded is skipped. Returns `None` when nothing is left to + /// send. + pub fn from_auto_detect_responses(responses: &[AutoDetectResponse]) -> Option { + let sub_headers: Vec = responses + .iter() + .filter_map(|response| match redecode(response) { + Ok(sub_header) => Some(sub_header), + Err(error) => { + debug!(%error, ?response, "Could not encode an auto-detect response for the tunnel"); + None + } + }) + .collect(); + (!sub_headers.is_empty()).then_some(Self { + sub_headers, + data: Vec::new(), + }) + } +} + +/// Re-reads one wire structure as another with the same bytes. +fn redecode Decode<'de>>(value: &dyn Encode) -> Result { + let bytes = ironrdp_core::encode_vec(value).map_err(|error| error.to_string())?; + ironrdp_core::decode(&bytes).map_err(|error| error.to_string()) +} diff --git a/crates/ironrdp-rdpeudp-tokio/src/framed.rs b/crates/ironrdp-rdpeudp-tokio/src/framed.rs index ea74fbcd8f..77892e2e80 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/framed.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/framed.rs @@ -74,11 +74,15 @@ mod tests { use tokio::sync::mpsc; use super::*; - use crate::transport::TunnelMessage; + use crate::transport::{ReceivedMessage, TunnelMessage}; /// Build a `UdpTransport` backed by test channels (no real network). - fn test_transport() -> (UdpTransport, mpsc::Sender, mpsc::Receiver) { - let (incoming_tx, incoming_rx) = mpsc::channel::(16); + fn test_transport() -> ( + UdpTransport, + mpsc::Sender, + mpsc::Receiver, + ) { + let (incoming_tx, incoming_rx) = mpsc::channel::(16); let (outgoing_tx, outgoing_rx) = mpsc::channel::(16); let transport = UdpTransport::from_channels(incoming_rx, outgoing_tx); diff --git a/crates/ironrdp-rdpeudp-tokio/src/lib.rs b/crates/ironrdp-rdpeudp-tokio/src/lib.rs index 20fece5f1a..53332ec922 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/lib.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/lib.rs @@ -6,6 +6,7 @@ pub mod framed; pub mod multitransport; pub mod transport; +pub(crate) mod autodetect; pub(crate) mod clock; pub(crate) mod driver; pub(crate) mod stream; diff --git a/crates/ironrdp-rdpeudp-tokio/src/transport.rs b/crates/ironrdp-rdpeudp-tokio/src/transport.rs index 0a2b86be15..983febbd8a 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/transport.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/transport.rs @@ -17,6 +17,7 @@ use core::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use core::time::Duration; use std::sync::{Arc, Mutex}; +use ironrdp_core::MonotonicInstant; use ironrdp_rdpemt::{RdpemtErrorExt as _, TunnelConfig}; use ironrdp_rdpeudp::pdu::V1Datagram; use ironrdp_rdpeudp::{ConnectionConfig, RdpeudpConnection, RdpeudpErrorExt as _}; @@ -236,6 +237,23 @@ impl From> for TunnelMessage { } } +/// A [`TunnelMessage`] with the time the read pump read its Tunnel Data PDU. +#[derive(Debug)] +pub(crate) struct ReceivedMessage { + pub(crate) message: TunnelMessage, + pub(crate) received_at: MonotonicInstant, +} + +#[cfg(test)] +impl> From for ReceivedMessage { + fn from(message: T) -> Self { + Self { + message: message.into(), + received_at: ironrdp_async::monotonic_now(), + } + } +} + /// A cloneable handle for sending data over an established UDP transport, /// obtained from [`UdpTransport::sender`]. /// @@ -303,7 +321,10 @@ impl UdpTransportSender { /// Drop this handle to initiate shutdown of the background tasks. pub struct UdpTransport { /// Receives higher-layer data (DVC frames) from the tunnel. - data_rx: mpsc::Receiver, + data_rx: mpsc::Receiver, + + /// When the read pump read the PDU most recently returned by `recv_message`. + last_received_at: Option, /// Sends higher-layer data into the tunnel for encryption and transmission. data_tx: mpsc::Sender, @@ -336,7 +357,20 @@ impl UdpTransport { /// /// Returns `None` when the tunnel is closed. pub async fn recv_message(&mut self) -> Option { - self.data_rx.recv().await + let received = self.data_rx.recv().await?; + self.last_received_at = Some(received.received_at); + Some(received.message) + } + + /// When the read pump read the Tunnel Data PDU most recently returned by + /// [`Self::recv`] or [`Self::recv_message`], or `None` before the first one. + /// + /// This is the time the PDU came off the tunnel, not the time the caller + /// dequeued it. The reading comes from [`ironrdp_async::monotonic_now`], the + /// clock behind [`ironrdp_async::Framed::last_read_at`], so it can be compared + /// with the arrival times of frames read on the main connection. + pub fn last_received_at(&self) -> Option { + self.last_received_at } /// Send a higher-layer data frame through the tunnel. @@ -449,9 +483,13 @@ impl UdpTransport { /// For unit tests that exercise the channel-based API (FramedRead, /// FramedWrite) without needing a real UDP socket or TLS stack. #[cfg(test)] - pub(crate) fn from_channels(data_rx: mpsc::Receiver, data_tx: mpsc::Sender) -> Self { + pub(crate) fn from_channels( + data_rx: mpsc::Receiver, + data_tx: mpsc::Sender, + ) -> Self { Self { data_rx, + last_received_at: None, data_tx, shared: Arc::new(Mutex::new(SharedIo::new())), driver_handle: AbortOnDrop::new(tokio::spawn(async { Ok(()) })), @@ -617,7 +655,7 @@ pub async fn connect_udp(config: UdpTransportConfig) -> Result(64); + let (incoming_tx, incoming_rx) = mpsc::channel::(64); let (outgoing_tx, mut outgoing_rx) = mpsc::channel::(64); // Read pump: TLS → RDPEMT decode → channel @@ -634,6 +672,7 @@ pub async fn connect_udp(config: UdpTransportConfig) -> Result Result< debug!("RDPEMT server tunnel established, starting data pump"); // Phase 6: Set up data channels and spawn pumps (identical to client side) - let (incoming_tx, incoming_rx) = mpsc::channel::(64); + let (incoming_tx, incoming_rx) = mpsc::channel::(64); let (outgoing_tx, mut outgoing_rx) = mpsc::channel::(64); let pump_handle = AbortOnDrop::new(tokio::spawn(async move { @@ -825,6 +864,7 @@ async fn accept_udp_inner(socket: UdpSocket, config: UdpAcceptConfig) -> Result< Ok(UdpTransport { data_rx: incoming_rx, + last_received_at: None, data_tx: outgoing_tx, shared, driver_handle, @@ -990,11 +1030,12 @@ mod tests { static DRIVER_RUNNING: AtomicBool = AtomicBool::new(false); DRIVER_RUNNING.store(false, Ordering::SeqCst); - let (_incoming_tx, incoming_rx) = mpsc::channel::(4); + let (_incoming_tx, incoming_rx) = mpsc::channel::(4); let (outgoing_tx, _outgoing_rx) = mpsc::channel::(4); let transport = UdpTransport { data_rx: incoming_rx, + last_received_at: None, data_tx: outgoing_tx, shared: Arc::new(Mutex::new(SharedIo::new())), driver_handle: AbortOnDrop::new(tokio::spawn(async { @@ -1157,13 +1198,37 @@ mod tests { data: vec![0x02], }; - incoming_tx.send(message.clone()).await.expect("queue"); - incoming_tx.send(message.clone()).await.expect("queue"); + incoming_tx.send(message.clone().into()).await.expect("queue"); + incoming_tx.send(message.clone().into()).await.expect("queue"); assert_eq!(transport.recv().await, Some(vec![0x02])); assert_eq!(transport.recv_message().await, Some(message)); } + /// The arrival time is the one the read pump stamped, not the time of the dequeue. + #[tokio::test] + async fn last_received_at_reports_the_pump_read_time() { + let (incoming_tx, incoming_rx) = mpsc::channel(4); + let (outgoing_tx, _outgoing_rx) = mpsc::channel(4); + let mut transport = UdpTransport::from_channels(incoming_rx, outgoing_tx); + assert_eq!(transport.last_received_at(), None); + + for millis in [10, 25] { + incoming_tx + .send(ReceivedMessage { + message: TunnelMessage::from(vec![0x01]), + received_at: MonotonicInstant::from_millis(millis), + }) + .await + .expect("queue"); + } + + transport.recv_message().await.expect("first message"); + assert_eq!(transport.last_received_at(), Some(MonotonicInstant::from_millis(10))); + transport.recv().await.expect("second message"); + assert_eq!(transport.last_received_at(), Some(MonotonicInstant::from_millis(25))); + } + /// Sub-headers that would overflow `HeaderLength` are refused at the /// call, not left to fail the write pump; the largest that fit go out. The /// boundary is also checked against the real `TunnelData` encoder, so the diff --git a/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs b/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs index 654c66b4ab..b694bab46d 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs @@ -15,7 +15,7 @@ use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _}; use tracing::{debug, trace}; use crate::error::{UdpTransportError, UdpTransportErrorExt as _}; -use crate::transport::TunnelMessage; +use crate::transport::{ReceivedMessage, TunnelMessage}; /// Read a complete RDPEMT PDU from the stream using self-framing. /// @@ -129,13 +129,18 @@ where pub(crate) async fn tunnel_data_loop( stream: &mut S, tunnel: &mut RdpemtTunnel, - data_tx: &tokio::sync::mpsc::Sender, + data_tx: &tokio::sync::mpsc::Sender, ) -> Result<(), UdpTransportError> where S: AsyncRead + Unpin, { loop { - let pdu = match read_tunnel_pdu(stream).await { + let read = read_tunnel_pdu(stream).await; + // Stamp the PDU now, before it waits in the channel, so a bandwidth + // measurement sees when it came off the tunnel rather than when the + // application got to it. + let received_at = ironrdp_async::monotonic_now(); + let pdu = match read { Ok(Some(pdu)) => pdu, // Clean shutdown: EOF before any byte of a new PDU. Ok(None) => { @@ -164,7 +169,11 @@ where sub_headers = sub_headers.len(), "Forwarding tunnel data" ); - if data_tx.send(TunnelMessage { sub_headers, data }).await.is_err() { + let received = ReceivedMessage { + message: TunnelMessage { sub_headers, data }, + received_at, + }; + if data_tx.send(received).await.is_err() { debug!("Tunnel data receiver dropped, stopping read pump"); // Application dropped the receiver return Ok(()); @@ -298,12 +307,16 @@ mod tests { let (tx, mut rx) = tokio::sync::mpsc::channel(4); let mut stream = io::Cursor::new(wire); + let before = ironrdp_async::monotonic_now(); tunnel_data_loop(&mut stream, &mut tunnel, &tx) .await .expect("clean EOF"); + let after = ironrdp_async::monotonic_now(); - let message = rx.recv().await.expect("one message"); - assert_eq!(message.sub_headers, vec![sub_header]); - assert!(message.data.is_empty()); + let received = rx.recv().await.expect("one message"); + assert_eq!(received.message.sub_headers, vec![sub_header]); + assert!(received.message.data.is_empty()); + // The pump stamps the read on the clock the main connection's reader uses. + assert!(before <= received.received_at && received.received_at <= after); } } diff --git a/crates/ironrdp-session/src/active_stage.rs b/crates/ironrdp-session/src/active_stage.rs index 050111d751..a6e348e3f9 100644 --- a/crates/ironrdp-session/src/active_stage.rs +++ b/crates/ironrdp-session/src/active_stage.rs @@ -1,7 +1,7 @@ use std::sync::Arc; use ironrdp_bulk::{BulkCompressor, CompressionType as BulkCompressionType}; -use ironrdp_core::{ReadCursor, WriteBuf}; +use ironrdp_core::{MonotonicInstant, ReadCursor, WriteBuf}; use ironrdp_displaycontrol::client::DisplayControlClient; use ironrdp_dvc::pdu::SoftSyncTunnelType; use ironrdp_dvc::{DrdynvcClient, DvcClientProcessor, DvcMessageBatch, DynamicChannelMut, DynamicChannelRef}; @@ -10,7 +10,7 @@ use ironrdp_graphics::pointer::DecodedPointer; use ironrdp_pdu::gcc::{ChannelName, Monitor}; use ironrdp_pdu::geometry::{ExclusiveRectangle, InclusiveRectangle, Rectangle as _}; use ironrdp_pdu::input::fast_path::{FastPathInput, FastPathInputEvent}; -use ironrdp_pdu::rdp::autodetect::AutoDetectRequest; +use ironrdp_pdu::rdp::autodetect::{AutoDetectRequest, AutoDetectResponse}; use ironrdp_pdu::rdp::capability_sets::WindowSupportLevel; use ironrdp_pdu::rdp::client_info::CompressionType; use ironrdp_pdu::rdp::headers::ShareDataPdu; @@ -183,15 +183,38 @@ impl ActiveStage { } /// Process a frame received from the server. + /// + /// Without an arrival time, bandwidth measurements are answered with an untimed, + /// zero-byte result; see [`Self::process_with_timestamp`]. pub fn process( &mut self, image: &mut DecodedImage, action: Action, frame: &[u8], + ) -> SessionResult> { + self.process_with_timestamp(image, action, frame, None) + } + + /// Process a frame received from the server, together with its arrival time. + /// + /// The clock stays outside this state machine: `received_at` is the time the transport + /// read the frame, from the same monotonic clock for every frame. Frames that were + /// buffered must keep their read time so a bandwidth measurement reflects network arrival + /// rather than the time spent decoding earlier frames. + pub fn process_with_timestamp( + &mut self, + image: &mut DecodedImage, + action: Action, + frame: &[u8], + received_at: Option, ) -> SessionResult> { self.damage_regions.clear(); let (mut stage_outputs, processor_updates) = match action { Action::FastPath => { + // Fast-path frames carry no security flags or data signature: IronRDP supports + // Enhanced RDP Security, and Standard RDP Security only at ENCRYPTION_LEVEL_NONE. + // The continuous bandwidth count therefore includes the entire frame. + self.x224_processor.record_bandwidth_bytes(frame.len()); let mut output = WriteBuf::new(); let processor_updates = self.fast_path_processor @@ -202,7 +225,9 @@ impl ActiveStage { ) } Action::X224 => { - let x224_outputs = self.x224_processor.process(frame, &mut self.bulk_decompressor)?; + let x224_outputs = + self.x224_processor + .process_with_timestamp(frame, &mut self.bulk_decompressor, received_at)?; let mut stage_outputs = Vec::new(); let mut processor_updates = Vec::new(); @@ -562,6 +587,37 @@ impl ActiveStage { .is_some_and(|drdynvc| drdynvc.switched_to_tunnel(SoftSyncTunnelType::RELIABLE_UDP)) } + /// Answers the auto-detect requests the server sends in the sub-headers of one Tunnel Data + /// PDU ([MS-RDPEMT] 2.2.1.1.1), and counts the `data_len` bytes of higher-layer data the PDU + /// carries for an open bandwidth measurement. Returns the responses to send back on the + /// tunnel. + /// + /// The tunnel and the main connection share one measurement, as [MS-RDPBCGR] 3.2.5.14 + /// defines one byte count and one timer: a window opened on either transport counts the + /// data received on both, and a Stop on either transport closes it. On the tunnel only the + /// data after the tunnel PDU header counts, so sub-header bytes never do. As on the message + /// channel, received bytes are counted before handling control messages: a Start resets the + /// count, and a Stop includes the carrying PDU's data in the result. + /// + /// `received_at` is the time the transport read the PDU, from the same monotonic clock as + /// the timestamps given to [`Self::process_with_timestamp`]. Without it, a bandwidth + /// measurement is answered with an untimed, zero-byte result. + /// + /// [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad + /// [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/16ffa852-8aa7-481c-99a0-36c1a9a198f6 + pub fn process_tunnel_auto_detect( + &mut self, + requests: Vec, + data_len: usize, + received_at: Option, + ) -> Vec { + self.x224_processor.record_bandwidth_bytes(data_len); + requests + .into_iter() + .filter_map(|request| self.x224_processor.respond_auto_detect(request, received_at)) + .collect() + } + /// Returns the Soft-Sync tunnel selected for client messages on `channel_id`. pub fn dvc_tunnel_for_channel(&self, channel_id: u32) -> Option { self.x224_processor diff --git a/crates/ironrdp-session/src/autodetect.rs b/crates/ironrdp-session/src/autodetect.rs new file mode 100644 index 0000000000..0b94893c49 --- /dev/null +++ b/crates/ironrdp-session/src/autodetect.rs @@ -0,0 +1,162 @@ +//! Client answers to the network auto-detect requests of [MS-RDPBCGR] 2.2.14. +//! +//! [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/dc672839-4f4e-40b1-a71c-cd6a959baa38 + +use ironrdp_core::MonotonicInstant; +use ironrdp_pdu::rdp::autodetect::{ + AutoDetectRequest, AutoDetectResponse, BW_RESULTS_CONNECT_TIME, BW_RESULTS_CONTINUOUS, BW_START_CONNECT_TIME, + BW_START_RELIABLE_UDP, BW_STOP_CONNECT_TIME, BW_STOP_RELIABLE_UDP, +}; +use tracing::debug; + +/// Answers the RTT and bandwidth measurement requests of a session. +/// +/// A session has one Network Characteristics Byte Count store and one timer, whichever +/// transport carries the Start and the Stop: a continuous measurement counts the data received +/// from the server on every transport ([MS-RDPBCGR] 3.2.5.14). The caller sends each response +/// back on the transport its request arrived on. Timestamps come from the caller, from one +/// monotonic clock for every transport. +/// +/// [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/16ffa852-8aa7-481c-99a0-36c1a9a198f6 +#[derive(Debug, Default)] +pub(crate) struct AutoDetectResponder { + bandwidth: Option, +} + +/// Size of the auto-detect header fields [MS-RDPBCGR] 3.2.5.14 counts along with payloadLength: +/// headerLength, headerTypeId, sequenceNumber, requestType and payloadLength itself. +/// +/// [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/16ffa852-8aa7-481c-99a0-36c1a9a198f6 +const AUTO_DETECT_HEADER_LEN: u32 = 8; + +/// Reported as `timeDelta` for a window that was not timed, and the floor for one that was: +/// a server computing `byteCount * 8 / timeDelta` divides by it ([MS-RDPBCGR] 3.3.5.14). +const UNMEASURABLE_INTERVAL_MS: u32 = 1; + +/// Bytes a connect-time Payload or Stop adds to the count: payloadLength plus the auto-detect +/// header, but not the security header. This is the rule the connector applies to the same PDUs. +fn counted_len(payload_len: usize) -> u32 { + u32::try_from(payload_len) + .unwrap_or(u32::MAX) + .saturating_add(AUTO_DETECT_HEADER_LEN) +} + +#[derive(Debug)] +struct BandwidthMeasurement { + started_at: MonotonicInstant, + bytes: u32, + continuous: bool, +} + +impl AutoDetectResponder { + /// Counts received bytes while a continuous bandwidth window is open. + pub(crate) fn record_bytes(&mut self, bytes: usize) { + if let Some(measurement) = self.bandwidth.as_mut().filter(|measurement| measurement.continuous) { + measurement.bytes = measurement + .bytes + .saturating_add(u32::try_from(bytes).unwrap_or(u32::MAX)); + } + } + + /// Returns the response `request` calls for, if any. + /// + /// Without an arrival time, a bandwidth measurement is answered with an untimed, zero-byte + /// result. + pub(crate) fn respond( + &mut self, + request: AutoDetectRequest, + received_at: Option, + ) -> Option { + match request { + AutoDetectRequest::RttRequest { sequence_number, .. } => { + Some(AutoDetectResponse::RttResponse { sequence_number }) + } + AutoDetectRequest::BandwidthMeasureStart { request_type, .. } + if matches!(request_type, BW_START_CONNECT_TIME | BW_START_RELIABLE_UDP) => + { + self.bandwidth = received_at.map(|started_at| BandwidthMeasurement { + started_at, + bytes: 0, + continuous: request_type == BW_START_RELIABLE_UDP, + }); + None + } + AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => { + if let Some(measurement) = self.bandwidth.as_mut().filter(|measurement| !measurement.continuous) { + measurement.bytes = measurement.bytes.saturating_add(counted_len(payload.len())); + } + None + } + AutoDetectRequest::BandwidthMeasureStop { + sequence_number, + request_type, + payload, + } if matches!(request_type, BW_STOP_CONNECT_TIME | BW_STOP_RELIABLE_UDP) => { + let continuous = request_type == BW_STOP_RELIABLE_UDP; + // A Stop of the other measurement type leaves the open window in place, so a + // later Stop of the matching type still reports it. + let measurement = if self + .bandwidth + .as_ref() + .is_some_and(|measurement| measurement.continuous == continuous) + { + self.bandwidth.take() + } else { + None + }; + let stop_bytes = if continuous { + 0 + } else { + counted_len(payload.as_ref().map_or(0, Vec::len)) + }; + let (time_delta_ms, byte_count) = match (measurement, received_at) { + (Some(measurement), Some(stopped_at)) => ( + u32::try_from(stopped_at.duration_since(measurement.started_at).as_millis()) + .unwrap_or(u32::MAX) + .max(UNMEASURABLE_INTERVAL_MS), + measurement.bytes.saturating_add(stop_bytes), + ), + (Some(measurement), None) => { + // The window was timed but this Stop was not, so there is nothing to + // divide the count by. Log the drop so it does not look like the + // ordinary no-window case. + debug!( + dropped_bytes = measurement.bytes, + "Bandwidth Measure Stop arrived with no arrival time although its window was open; \ + dropping the accumulated count" + ); + (UNMEASURABLE_INTERVAL_MS, 0) + } + (None, _) => { + debug!( + continuous, + "Bandwidth Measure Stop arrived with no matching open window; reporting no measurement" + ); + (UNMEASURABLE_INTERVAL_MS, 0) + } + }; + Some(AutoDetectResponse::BandwidthMeasureResults { + sequence_number, + response_type: if continuous { + BW_RESULTS_CONTINUOUS + } else { + BW_RESULTS_CONNECT_TIME + }, + time_delta_ms, + byte_count, + }) + } + request @ AutoDetectRequest::NetworkCharacteristicsResult { .. } => { + // The message-channel processor surfaces this request itself; requests from + // the UDP tunnel reach this responder directly. + debug!(?request, "Received network characteristics from server"); + None + } + request => { + // Measurements for a lossy transport are not answered here. + debug!(?request, "Ignoring auto-detect request for another transport"); + None + } + } + } +} diff --git a/crates/ironrdp-session/src/lib.rs b/crates/ironrdp-session/src/lib.rs index 3103da0a37..84420484ed 100644 --- a/crates/ironrdp-session/src/lib.rs +++ b/crates/ironrdp-session/src/lib.rs @@ -11,6 +11,7 @@ pub mod rfx; // FIXME: maybe this module should not be in this crate pub mod x224; mod active_stage; +mod autodetect; mod palette; use core::fmt; diff --git a/crates/ironrdp-session/src/x224/mod.rs b/crates/ironrdp-session/src/x224/mod.rs index d006393ba4..570222404f 100644 --- a/crates/ironrdp-session/src/x224/mod.rs +++ b/crates/ironrdp-session/src/x224/mod.rs @@ -1,5 +1,5 @@ use ironrdp_bulk::BulkCompressor; -use ironrdp_core::{Decode as _, ReadCursor, WriteBuf, decode}; +use ironrdp_core::{Decode as _, MonotonicInstant, ReadCursor, WriteBuf, decode}; use ironrdp_dvc::{DrdynvcClient, DvcClientProcessor, DynamicChannelMut, DynamicChannelRef}; use ironrdp_pdu::gcc::{ChannelName, Monitor}; use ironrdp_pdu::mcs::{DisconnectProviderUltimatum, DisconnectReason, McsMessage, SendDataIndicationCtx}; @@ -18,6 +18,7 @@ use ironrdp_svc::{ }; use tracing::debug; +use crate::autodetect::AutoDetectResponder; use crate::{SessionError, SessionErrorExt as _, SessionResult, reason_err}; /// X224 Processor output @@ -72,7 +73,7 @@ pub enum ProcessorOutput { /// Auto-detect network characteristics from server ([\[MS-RDPBCGR\] 2.2.14]). /// /// Currently only surfaces [`AutoDetectRequest::NetworkCharacteristicsResult`]. - /// RTT requests are handled internally with automatic responses. + /// RTT and bandwidth measurement requests are answered internally. /// /// [\[MS-RDPBCGR\] 2.2.14]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/dc672839-4f4e-40b1-a71c-cd6a959baa38 AutoDetect(AutoDetectRequest), @@ -107,6 +108,7 @@ pub struct Processor { io_channel_id: u16, message_channel_id: Option, share_id: u32, + auto_detect: AutoDetectResponder, } impl Processor { @@ -123,6 +125,7 @@ impl Processor { io_channel_id, message_channel_id, share_id, + auto_detect: AutoDetectResponder::default(), } } @@ -212,6 +215,19 @@ impl Processor { &mut self, frame: &[u8], bulk_decompressor: &mut Option, + ) -> SessionResult> { + self.process_with_timestamp(frame, bulk_decompressor, None) + } + + /// Processes a frame with its driver-observed arrival time. + /// + /// Supply the same monotonic clock for every frame. Without a timestamp, + /// bandwidth replies use a conservative zero-byte measurement. + pub fn process_with_timestamp( + &mut self, + frame: &[u8], + bulk_decompressor: &mut Option, + received_at: Option, ) -> SessionResult> { let data_ctx: SendDataIndicationCtx<'_> = match ironrdp_pdu::mcs::decode_send_data_indication(frame) { Ok(data_ctx) => data_ctx, @@ -233,10 +249,13 @@ impl Processor { let channel_id = data_ctx.channel_id; if channel_id == self.io_channel_id { - self.process_io_channel_data_indication(data_ctx, bulk_decompressor) + self.process_io_channel_data_indication(data_ctx, frame.len(), bulk_decompressor) } else if self.message_channel_id == Some(channel_id) { - self.process_message_channel(data_ctx) + self.process_message_channel(data_ctx, received_at) } else { + // SVC data has no RDP Security Header (Enhanced RDP Security, or Standard RDP + // Security at ENCRYPTION_LEVEL_NONE): include TPKT/X224/MCS. + self.record_bandwidth_bytes(frame.len()); let maximum_chunk_size = self.static_channels.maximum_chunk_size(); if let Some(svc) = self.static_channels.get_by_channel_id_mut(channel_id) { let response_pdus = svc.process(data_ctx.user_data).map_err(SessionError::pdu)?; @@ -251,6 +270,7 @@ impl Processor { fn process_io_channel_data_indication( &mut self, data_ctx: SendDataIndicationCtx<'_>, + frame_len: usize, bulk_decompressor: &mut Option, ) -> SessionResult> { debug_assert_eq!(data_ctx.channel_id, self.io_channel_id); @@ -261,9 +281,15 @@ impl Processor { ironrdp_pdu::rdp::headers::decode_io_channel(data_ctx), Ok(IoChannelPdu::MultitransportRequest(_)) ) { + self.record_bandwidth_bytes(data_ctx.user_data.len() - BasicSecurityHeader::FIXED_PART_SIZE); return self.process_io_channel(data_ctx, bulk_decompressor); } + // Ordinary IO data has no RDP Security Header (Enhanced RDP Security, or Standard RDP + // Security at ENCRYPTION_LEVEL_NONE), and the Share Control decode below relies on + // that. Count the entire frame once, including framing and any concatenated + // Share Control PDUs. + self.record_bandwidth_bytes(frame_len); let mut outputs = Vec::new(); let mut offset = 0usize; let data = data_ctx.user_data; @@ -445,6 +471,23 @@ impl Processor { Ok(decompressed) } + /// Counts received session bytes while a continuous bandwidth window is open. + /// Exclude framing only when an RDP Security Header is present, counting just + /// the bytes after that header ([MS-RDPBCGR] 3.2.5.14). + pub(crate) fn record_bandwidth_bytes(&mut self, bytes: usize) { + self.auto_detect.record_bytes(bytes); + } + + /// Answers an auto-detect request that arrived on another transport, using the same + /// measurement state as the message channel. + pub(crate) fn respond_auto_detect( + &mut self, + request: AutoDetectRequest, + received_at: Option, + ) -> Option { + self.auto_detect.respond(request, received_at) + } + /// Process a PDU received on the MCS message channel: auto-detect /// ([MS-RDPBCGR] 2.2.14), multitransport ([MS-RDPBCGR] 2.2.15), or /// Heartbeat ([MS-RDPBCGR] 2.2.16.1). @@ -457,13 +500,18 @@ impl Processor { /// session-fatal decode error: this channel is forward-safe for future /// message-channel PDU types the same way the connect-time demux /// (`ironrdp-connector`) already is. - fn process_message_channel(&self, data_ctx: SendDataIndicationCtx<'_>) -> SessionResult> { + fn process_message_channel( + &mut self, + data_ctx: SendDataIndicationCtx<'_>, + received_at: Option, + ) -> SessionResult> { let Some(message_channel_id) = self.message_channel_id else { return Err(reason_err!("message channel", "no message channel negotiated")); }; let mut peek = ReadCursor::new(data_ctx.user_data); let security_header = BasicSecurityHeader::decode(&mut peek).map_err(SessionError::decode)?; + self.record_bandwidth_bytes(peek.len()); let flags = security_header .flags .difference(BasicSecurityHeaderFlags::RESET_SEQNO | BasicSecurityHeaderFlags::IGNORE_SEQNO); @@ -495,29 +543,23 @@ impl Processor { let req = decode::(data_ctx.user_data).map_err(SessionError::decode)?; - match req.request { - AutoDetectRequest::RttRequest { sequence_number, .. } => { - let response = AutoDetectRspPdu::new(AutoDetectResponse::RttResponse { sequence_number }); - let mut frame = WriteBuf::new(); - ironrdp_pdu::mcs::encode_send_data_request( - self.user_channel_id, - message_channel_id, - &response, - &mut frame, - ) - .map_err(SessionError::encode)?; - debug!(sequence_number, "Responded to auto-detect RTT request"); - Ok(vec![ProcessorOutput::ResponseFrame(frame.into_inner())]) - } - req @ AutoDetectRequest::NetworkCharacteristicsResult { .. } => { - debug!(?req, "Received network characteristics from server"); - Ok(vec![ProcessorOutput::AutoDetect(req)]) - } - req => { - debug!(?req, "Auto-detect request not yet implemented"); - Ok(Vec::new()) - } + if let request @ AutoDetectRequest::NetworkCharacteristicsResult { .. } = req.request { + debug!(?request, "Received network characteristics from server"); + return Ok(vec![ProcessorOutput::AutoDetect(request)]); } + let Some(response) = self.auto_detect.respond(req.request, received_at) else { + return Ok(Vec::new()); + }; + debug!(?response, "Responding to an auto-detect request"); + let mut frame = WriteBuf::new(); + ironrdp_pdu::mcs::encode_send_data_request( + self.user_channel_id, + message_channel_id, + &AutoDetectRspPdu::new(response), + &mut frame, + ) + .map_err(SessionError::encode)?; + Ok(vec![ProcessorOutput::ResponseFrame(frame.into_inner())]) } /// Encodes an Initiate Multitransport Response on the MCS message channel. @@ -638,14 +680,17 @@ mod tests { fn processor_surfaces_multitransport_request_on_message_channel() { let request = multitransport_request(); let encoded = encode_vec(&request).expect("encode multitransport request"); - let processor = Processor::new(StaticChannelSet::new(), 1002, 1003, Some(1004), 0); + let mut processor = Processor::new(StaticChannelSet::new(), 1002, 1003, Some(1004), 0); let outputs = processor - .process_message_channel(SendDataIndicationCtx { - initiator_id: 1002, - channel_id: 1004, - user_data: &encoded, - }) + .process_message_channel( + SendDataIndicationCtx { + initiator_id: 1002, + channel_id: 1004, + user_data: &encoded, + }, + None, + ) .expect("surface multitransport request"); assert!(matches!( @@ -667,6 +712,7 @@ mod tests { channel_id: 1003, user_data: &encoded, }, + encoded.len(), &mut None, ) .expect("ignore a misrouted optional multitransport request"); @@ -884,6 +930,7 @@ mod tests { channel_id: 1003, user_data: &user_data, }, + user_data.len(), &mut None, ) .expect("concatenated Share Control PDUs should be split"); @@ -905,6 +952,7 @@ mod tests { channel_id: 1003, user_data: &[0x06], }, + 1, &mut None, ) .expect_err("a truncated totalLength field is invalid"); @@ -925,6 +973,7 @@ mod tests { channel_id: 1003, user_data: &user_data, }, + user_data.len(), &mut None, ) .expect_err("an overrunning concatenated totalLength is invalid"); @@ -950,6 +999,7 @@ mod tests { channel_id: 1003, user_data: &user_data, }, + user_data.len(), &mut None, ) .expect("invalid first totalLength should fall back to whole-buffer decode"); diff --git a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs index 4591f3fe92..5a97e01e3d 100644 --- a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs @@ -1,10 +1,22 @@ use std::borrow::Cow; use ironrdp_core::encode_vec; +use ironrdp_dvc::DrdynvcClient; +use ironrdp_dvc::pdu::{CapabilitiesRequestPdu, CapsVersion, DrdynvcServerPdu}; +use ironrdp_graphics::image_processing::PixelFormat; +use ironrdp_pdu::Action; +use ironrdp_pdu::fast_path::{EncryptionFlags, FastPathHeader, FastPathUpdatePdu, Fragmentation, UpdateCode}; use ironrdp_pdu::mcs::{McsMessage, SendDataIndication}; use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu}; +use ironrdp_pdu::rdp::headers::{ + BasicSecurityHeader, BasicSecurityHeaderFlags, ServerDeactivateAll, ShareControlHeader, ShareControlPdu, +}; +use ironrdp_pdu::rdp::multitransport::{MultitransportRequestPdu, RequestedProtocol}; +use ironrdp_pdu::rdp::vc::{ChannelControlFlags, ChannelPduHeader}; use ironrdp_pdu::x224::X224; +use ironrdp_session::image::DecodedImage; use ironrdp_session::x224::Processor; +use ironrdp_session::{ActiveStage, ActiveStageBuilder, ActiveStageOutput}; use ironrdp_svc::StaticChannelSet; const USER_CHANNEL_ID: u16 = 1002; @@ -33,10 +45,13 @@ fn process_frame(processor: &mut Processor, frame: &[u8]) -> Vec Vec { let pdu = AutoDetectReqPdu::new(request); let user_data = encode_vec(&pdu).unwrap(); + encode_server_data(MESSAGE_CHANNEL_ID, user_data) +} +fn encode_server_data(channel_id: u16, user_data: Vec) -> Vec { let indication = McsMessage::SendDataIndication(SendDataIndication { initiator_id: USER_CHANNEL_ID, - channel_id: MESSAGE_CHANNEL_ID, + channel_id, user_data: Cow::Owned(user_data), }); @@ -130,7 +145,7 @@ fn bandwidth_measure_stop_does_not_crash() { let frame = encode_server_autodetect(request); let outputs = process_frame(&mut processor, &frame); - assert!(outputs.is_empty(), "BW stop should produce no output"); + assert_eq!(bandwidth_result(&outputs), (200, 0x000b, 1, 0)); } #[test] @@ -142,3 +157,457 @@ fn bandwidth_measure_payload_does_not_crash() { let outputs = process_frame(&mut processor, &frame); assert!(outputs.is_empty(), "BW payload should produce no output"); } + +fn timed_request( + processor: &mut Processor, + request: AutoDetectRequest, + millis: u64, +) -> Vec { + processor + .process_with_timestamp( + &encode_server_autodetect(request), + &mut None, + Some(ironrdp_core::MonotonicInstant::from_millis(millis)), + ) + .expect("process timed auto-detect request") +} + +fn bandwidth_result(outputs: &[ironrdp_session::x224::ProcessorOutput]) -> (u16, u16, u32, u32) { + let [ironrdp_session::x224::ProcessorOutput::ResponseFrame(frame)] = outputs else { + panic!("expected exactly one bandwidth response"); + }; + bandwidth_result_frame(frame) +} + +fn bandwidth_result_frame(frame: &[u8]) -> (u16, u16, u32, u32) { + let X224(McsMessage::SendDataRequest(message)) = ironrdp_core::decode::>>(frame).unwrap() + else { + panic!("expected main-channel response"); + }; + assert_eq!(message.channel_id, MESSAGE_CHANNEL_ID); + let response = ironrdp_core::decode::(&message.user_data).unwrap(); + let AutoDetectResponse::BandwidthMeasureResults { + sequence_number, + response_type, + time_delta_ms, + byte_count, + } = response.response + else { + panic!("expected bandwidth results"); + }; + (sequence_number, response_type, time_delta_ms, byte_count) +} + +#[test] +fn continuous_measurement_excludes_message_channel_security_headers() { + let mut processor = make_processor(); + assert!(timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10).is_empty()); + timed_request(&mut processor, AutoDetectRequest::rtt_continuous(2), 20); + let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(3), 35); + // RTT and Stop have six-byte auto-detect headers. Neither four-byte + // security header, nor TPKT/X224/MCS framing, belongs to the count. + assert_eq!(bandwidth_result(&result), (3, 0x000b, 25, 12)); +} + +#[test] +fn continuous_measurement_counts_io_framing_once_for_concatenated_pdus() { + let mut processor = make_processor(); + let pdu = encode_vec(&ShareControlHeader { + share_control_pdu: ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll), + pdu_source: USER_CHANNEL_ID, + share_id: SHARE_ID, + }) + .unwrap(); + let frame = encode_server_data(IO_CHANNEL_ID, [pdu.as_slice(), pdu.as_slice()].concat()); + + timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10); + assert_eq!(process_frame(&mut processor, &frame).len(), 2); + let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(2), 35); + // There is no RDP Security Header: include TPKT, X224, MCS and both PDUs. + assert_eq!( + bandwidth_result(&result), + (2, 0x000b, 25, u32::try_from(frame.len()).unwrap() + 6) + ); +} + +#[test] +fn continuous_measurement_counts_svc_framing() { + let mut channels = StaticChannelSet::new(); + channels.insert(DrdynvcClient::new()); + channels.attach_channel_id(core::any::TypeId::of::(), 1005); + let mut processor = Processor::new( + channels, + USER_CHANNEL_ID, + IO_CHANNEL_ID, + Some(MESSAGE_CHANNEL_ID), + SHARE_ID, + ); + let capabilities = encode_vec(&DrdynvcServerPdu::Capabilities(CapabilitiesRequestPdu::new( + CapsVersion::V1, + None, + ))) + .unwrap(); + let mut chunk = encode_vec(&ChannelPduHeader { + length: u32::try_from(capabilities.len()).unwrap(), + flags: ChannelControlFlags::FLAG_FIRST | ChannelControlFlags::FLAG_LAST, + }) + .unwrap(); + chunk.extend_from_slice(&capabilities); + let frame = encode_server_data(1005, chunk); + + timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10); + process_frame(&mut processor, &frame); + let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(2), 35); + assert_eq!( + bandwidth_result(&result), + (2, 0x000b, 25, u32::try_from(frame.len()).unwrap() + 6) + ); +} + +#[test] +fn continuous_measurement_excludes_io_multitransport_security_header() { + let mut processor = make_processor(); + let data = encode_vec(&MultitransportRequestPdu { + security_header: BasicSecurityHeader { + flags: BasicSecurityHeaderFlags::TRANSPORT_REQ, + }, + request_id: 42, + requested_protocol: RequestedProtocol::UdpFecR, + security_cookie: [0xab; 16], + }) + .unwrap(); + let counted = u32::try_from(data.len() - BasicSecurityHeader::FIXED_PART_SIZE).unwrap(); + let frame = encode_server_data(IO_CHANNEL_ID, data); + + timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10); + assert!(process_frame(&mut processor, &frame).is_empty()); + let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(2), 35); + assert_eq!(bandwidth_result(&result), (2, 0x000b, 25, counted + 6)); +} + +#[test] +fn connect_time_measurement_includes_payload_headers_once() { + let mut processor = make_processor(); + timed_request(&mut processor, AutoDetectRequest::bw_start_connect_time(1), 100); + timed_request(&mut processor, AutoDetectRequest::bw_payload(2, vec![0xaa; 64]), 110); + timed_request(&mut processor, AutoDetectRequest::rtt_continuous(3), 112); + let result = timed_request( + &mut processor, + AutoDetectRequest::bw_stop_connect_time(4, vec![0xbb; 16]), + 140, + ); + assert_eq!(bandwidth_result(&result), (4, 0x0003, 40, 64 + 8 + 16 + 8)); +} + +#[test] +fn repeated_start_resets_count_and_timer_and_stop_ends_window() { + let mut processor = make_processor(); + timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 100); + timed_request(&mut processor, AutoDetectRequest::rtt_continuous(2), 110); + timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(3), 200); + let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(4), 210); + assert_eq!(bandwidth_result(&result), (4, 0x000b, 10, 6)); + let repeated = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(5), 220); + assert_eq!(bandwidth_result(&repeated), (5, 0x000b, 1, 0)); +} + +#[test] +fn stop_of_the_other_type_keeps_the_open_window() { + let mut processor = make_processor(); + timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 100); + let mismatched = timed_request( + &mut processor, + AutoDetectRequest::bw_stop_connect_time(2, vec![0xbb; 16]), + 110, + ); + assert_eq!(bandwidth_result(&mismatched), (2, 0x0003, 1, 0)); + let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(3), 130); + // The mismatched connect-time Stop arrived inside the continuous window, so its bytes + // after the Security Header count, along with the matching Stop's six. + assert_eq!(bandwidth_result(&result), (3, 0x000b, 30, 16 + 8 + 6)); + + let mut processor = make_processor(); + timed_request(&mut processor, AutoDetectRequest::bw_start_connect_time(1), 100); + let mismatched = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(2), 110); + assert_eq!(bandwidth_result(&mismatched), (2, 0x000b, 1, 0)); + let result = timed_request( + &mut processor, + AutoDetectRequest::bw_stop_connect_time(3, vec![0xbb; 16]), + 130, + ); + assert_eq!(bandwidth_result(&result), (3, 0x0003, 30, 16 + 8)); +} + +#[test] +fn measurement_timing_saturates_and_never_reports_zero_divisor() { + for (start, stop, expected) in [(100, 100, 1), (100, 90, 1), (0, u64::MAX, u32::MAX)] { + let mut processor = make_processor(); + timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), start); + let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(2), stop); + assert_eq!(bandwidth_result(&result), (2, 0x000b, expected, 6)); + } +} + +#[test] +fn lossy_requests_on_main_channel_are_not_answered() { + use ironrdp_pdu::rdp::autodetect::{BW_START_LOSSY_UDP, BW_STOP_LOSSY_UDP}; + let mut processor = make_processor(); + assert!( + timed_request( + &mut processor, + AutoDetectRequest::BandwidthMeasureStart { + sequence_number: 1, + request_type: BW_START_LOSSY_UDP, + }, + 10 + ) + .is_empty() + ); + assert!( + timed_request( + &mut processor, + AutoDetectRequest::BandwidthMeasureStop { + sequence_number: 1, + request_type: BW_STOP_LOSSY_UDP, + payload: None, + }, + 20 + ) + .is_empty() + ); +} + +#[test] +fn untimed_driver_does_not_report_accumulated_bytes_as_a_real_measurement() { + let mut processor = make_processor(); + timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10); + timed_request(&mut processor, AutoDetectRequest::rtt_continuous(2), 20); + let outputs = process_frame( + &mut processor, + &encode_server_autodetect(AutoDetectRequest::bw_stop_continuous(3)), + ); + assert_eq!(bandwidth_result(&outputs), (3, 0x000b, 1, 0)); +} + +#[test] +fn untimed_connect_time_stop_does_not_report_stop_payload_bytes() { + let mut processor = make_processor(); + timed_request(&mut processor, AutoDetectRequest::bw_start_connect_time(1), 10); + timed_request(&mut processor, AutoDetectRequest::bw_payload(2, vec![0xaa; 64]), 20); + let outputs = process_frame( + &mut processor, + &encode_server_autodetect(AutoDetectRequest::bw_stop_connect_time(3, vec![0xbb; 16])), + ); + assert_eq!(bandwidth_result(&outputs), (3, 0x0003, 1, 0)); +} + +#[test] +fn connect_time_stop_without_a_timed_start_reports_zero_bytes() { + for send_untimed_start in [false, true] { + for stop_time in [None, Some(ironrdp_core::MonotonicInstant::from_millis(20))] { + let mut processor = make_processor(); + if send_untimed_start { + process_frame( + &mut processor, + &encode_server_autodetect(AutoDetectRequest::bw_start_connect_time(1)), + ); + } + let frame = encode_server_autodetect(AutoDetectRequest::bw_stop_connect_time(2, vec![0xbb; 16])); + let outputs = processor.process_with_timestamp(&frame, &mut None, stop_time).unwrap(); + assert_eq!(bandwidth_result(&outputs), (2, 0x0003, 1, 0)); + } + } +} + +fn make_active_stage() -> ActiveStage { + ActiveStageBuilder { + static_channels: StaticChannelSet::new(), + user_channel_id: USER_CHANNEL_ID, + io_channel_id: IO_CHANNEL_ID, + message_channel_id: Some(MESSAGE_CHANNEL_ID), + share_id: SHARE_ID, + compression_type: None, + enable_server_pointer: false, + pointer_software_rendering: false, + } + .build() +} + +/// A fast-path frame carrying one Synchronize update with `data_len` bytes of data. +fn fast_path_frame(data_len: usize) -> Vec { + let data = vec![0; data_len]; + let update = encode_vec(&FastPathUpdatePdu { + fragmentation: Fragmentation::Single, + update_code: UpdateCode::Synchronize, + compression_flags: None, + compression_type: None, + data: &data, + }) + .unwrap(); + let mut frame = encode_vec(&FastPathHeader::new(EncryptionFlags::empty(), update.len())).unwrap(); + frame.extend_from_slice(&update); + frame +} + +fn process_stage_frame( + stage: &mut ActiveStage, + image: &mut DecodedImage, + action: Action, + frame: &[u8], + millis: u64, +) -> Vec { + stage + .process_with_timestamp( + image, + action, + frame, + Some(ironrdp_core::MonotonicInstant::from_millis(millis)), + ) + .expect("process timed frame") +} + +#[test] +fn continuous_measurement_counts_fast_path_data_only_inside_the_window() { + let mut stage = make_active_stage(); + let mut image = DecodedImage::new(PixelFormat::RgbA32, 64, 64); + let frame = fast_path_frame(100); + + process_stage_frame(&mut stage, &mut image, Action::FastPath, &frame, 5); + let start = encode_server_autodetect(AutoDetectRequest::bw_start_continuous(1)); + process_stage_frame(&mut stage, &mut image, Action::X224, &start, 10); + process_stage_frame(&mut stage, &mut image, Action::FastPath, &frame, 20); + let stop = encode_server_autodetect(AutoDetectRequest::bw_stop_continuous(2)); + let outputs = process_stage_frame(&mut stage, &mut image, Action::X224, &stop, 40); + process_stage_frame(&mut stage, &mut image, Action::FastPath, &frame, 50); + + let [ActiveStageOutput::ResponseFrame(response)] = outputs.as_slice() else { + panic!("expected exactly one bandwidth response, got {outputs:?}"); + }; + // The whole fast-path frame counts because it has no RDP Security Header. + // The Stop adds only its six-byte auto-detect header, after its Security Header. + assert_eq!( + bandwidth_result_frame(response), + (2, 0x000b, 30, u32::try_from(frame.len()).unwrap() + 6) + ); +} + +fn at(millis: u64) -> ironrdp_core::MonotonicInstant { + ironrdp_core::MonotonicInstant::from_millis(millis) +} + +#[test] +fn tunnel_auto_detect_answers_an_rtt_request() { + let mut stage = make_active_stage(); + let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::rtt_continuous(7)], 10, Some(at(1_000))); + assert_eq!(responses, [AutoDetectResponse::RttResponse { sequence_number: 7 }]); +} + +/// Count received payload before processing control messages, as on the main connection: +/// the Start resets its carrying PDU's count, while the Stop includes its carrying PDU. +#[test] +fn tunnel_auto_detect_counts_the_data_between_start_and_stop() { + let mut stage = make_active_stage(); + + assert!( + stage + .process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 7, Some(at(1_000))) + .is_empty() + ); + assert!( + stage + .process_tunnel_auto_detect(Vec::new(), 100, Some(at(1_010))) + .is_empty() + ); + let responses = + stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(3)], 50, Some(at(1_040))); + + let [ + AutoDetectResponse::BandwidthMeasureResults { + sequence_number, + time_delta_ms, + byte_count, + .. + }, + ] = responses.as_slice() + else { + panic!("expected one bandwidth result, got {responses:?}"); + }; + assert_eq!((*sequence_number, *time_delta_ms, *byte_count), (3, 40, 100 + 50)); +} + +/// The tunnel and the main connection share one byte count and one timer ([MS-RDPBCGR] +/// 3.2.5.14): a window opened on the main connection counts tunnel data too, a Stop on the +/// tunnel closes it, and the result goes back on the tunnel. +/// +/// [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/16ffa852-8aa7-481c-99a0-36c1a9a198f6 +#[test] +fn tunnel_and_main_connection_share_one_measurement() { + let mut stage = make_active_stage(); + let mut image = DecodedImage::new(PixelFormat::RgbA32, 64, 64); + + let start = encode_server_autodetect(AutoDetectRequest::bw_start_continuous(1)); + process_stage_frame(&mut stage, &mut image, Action::X224, &start, 1_000); + process_stage_frame(&mut stage, &mut image, Action::FastPath, &fast_path_frame(100), 1_010); + assert!( + stage + .process_tunnel_auto_detect(Vec::new(), 500, Some(at(1_015))) + .is_empty() + ); + + let tunnel = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(2)], 0, Some(at(1_020))); + let tcp_bytes = u32::try_from(fast_path_frame(100).len()).unwrap(); + assert_eq!( + tunnel, + [AutoDetectResponse::BandwidthMeasureResults { + sequence_number: 2, + response_type: 0x000b, + time_delta_ms: 20, + byte_count: tcp_bytes + 500, + }] + ); + + // The tunnel's Stop closed the only window, so a later Stop on the main connection has + // nothing to report. + let stop = encode_server_autodetect(AutoDetectRequest::bw_stop_continuous(3)); + let outputs = process_stage_frame(&mut stage, &mut image, Action::X224, &stop, 1_030); + let [ActiveStageOutput::ResponseFrame(response)] = outputs.as_slice() else { + panic!("expected exactly one bandwidth response, got {outputs:?}"); + }; + assert_eq!(bandwidth_result_frame(response), (3, 0x000b, 1, 0)); +} + +/// A window opened on the tunnel is closed by a Stop on the main connection, and the result +/// goes back there. +#[test] +fn tunnel_start_and_main_connection_stop_measure_one_window() { + let mut stage = make_active_stage(); + let mut image = DecodedImage::new(PixelFormat::RgbA32, 64, 64); + + stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 0, Some(at(1_000))); + stage.process_tunnel_auto_detect(Vec::new(), 300, Some(at(1_010))); + + let stop = encode_server_autodetect(AutoDetectRequest::bw_stop_continuous(2)); + let outputs = process_stage_frame(&mut stage, &mut image, Action::X224, &stop, 1_025); + let [ActiveStageOutput::ResponseFrame(response)] = outputs.as_slice() else { + panic!("expected exactly one bandwidth response, got {outputs:?}"); + }; + // The tunnel's 300 bytes and the Stop's own six bytes after its Security Header. + assert_eq!(bandwidth_result_frame(response), (2, 0x000b, 25, 300 + 6)); +} + +/// Without an arrival time, a tunnel bandwidth measurement is answered as untimed. +#[test] +fn untimed_tunnel_stop_reports_no_measurement() { + let mut stage = make_active_stage(); + stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 0, Some(at(1_000))); + stage.process_tunnel_auto_detect(Vec::new(), 300, Some(at(1_010))); + let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(2)], 0, None); + assert!(matches!( + responses.as_slice(), + [AutoDetectResponse::BandwidthMeasureResults { + time_delta_ms: 1, + byte_count: 0, + .. + }] + )); +} diff --git a/crates/ironrdp-testsuite-extra/tests/main.rs b/crates/ironrdp-testsuite-extra/tests/main.rs index cd9ce0044b..77881e57f5 100644 --- a/crates/ironrdp-testsuite-extra/tests/main.rs +++ b/crates/ironrdp-testsuite-extra/tests/main.rs @@ -11,5 +11,6 @@ mod dvc_pipe_proxy; mod e2e; mod gateway_detect; mod rdpeudp_tokio; +mod rdpeudp_tokio_auto_detect; mod vmconnect; mod volume; diff --git a/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio_auto_detect.rs b/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio_auto_detect.rs new file mode 100644 index 0000000000..b449da012f --- /dev/null +++ b/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio_auto_detect.rs @@ -0,0 +1,83 @@ +//! Auto-detect requests and responses travel on the UDP tunnel as tunnel sub-headers, whose +//! own two bytes are the auto-detect structure's headerLength and headerTypeId ([MS-RDPEMT] +//! 2.2.1.1.1). +//! +//! [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad + +use ironrdp_core::encode_vec; +use ironrdp_pdu::rdp::autodetect::{AutoDetectRequest, AutoDetectResponse}; +use ironrdp_rdpemt::{SubHeaderType, TunnelData, TunnelSubHeader}; +use ironrdp_rdpeudp_tokio::TunnelMessage; + +fn message(sub_headers: Vec) -> TunnelMessage { + TunnelMessage { + sub_headers, + data: Vec::new(), + } +} + +/// Windows frames an RTT Measure Request on the tunnel as `06 00 01 00`. Request types the +/// decoder does not model, and sub-headers that are not requests, are skipped. +#[test] +fn tunnel_sub_headers_decode_as_auto_detect_requests() { + let rtt: TunnelSubHeader = ironrdp_core::decode(&[0x06, 0x00, 0x07, 0x00, 0x01, 0x00]).expect("sub-header"); + let unmodeled = TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectRequest, + data: vec![0x01, 0x00, 0xFF, 0x7F], + }; + let response = TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectResponse, + data: vec![0x07, 0x00, 0x00, 0x00], + }; + + let requests = message(vec![rtt, unmodeled, response]).auto_detect_requests(); + assert!(matches!( + requests.as_slice(), + [AutoDetectRequest::RttRequest { sequence_number: 7, .. }] + )); +} + +/// The RTT response goes back as `06 01 00 00`, alone in a Tunnel Data PDU with no data. +#[test] +fn auto_detect_responses_encode_as_tunnel_sub_headers() { + let reply = TunnelMessage::from_auto_detect_responses(&[AutoDetectResponse::RttResponse { sequence_number: 7 }]) + .expect("reply"); + let pdu = encode_vec(&TunnelData { + sub_headers: reply.sub_headers, + higher_layer_data: reply.data, + }) + .expect("encode"); + // TunnelData: action 2, payloadLength 0, headerLength 4 + 6, then the sub-header. + assert_eq!(pdu, [0x02, 0x00, 0x00, 0x0A, 0x06, 0x01, 0x07, 0x00, 0x00, 0x00]); +} + +#[test] +fn no_responses_make_no_reply() { + assert_eq!(TunnelMessage::from_auto_detect_responses(&[]), None); +} + +#[test] +fn bandwidth_sub_headers_preserve_the_wire_layout() { + let start = TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectRequest, + data: vec![9, 0, 0x14, 0], + }; + assert_eq!( + message(vec![start]).auto_detect_requests(), + [AutoDetectRequest::bw_start_continuous(9)] + ); + let response = AutoDetectResponse::BandwidthMeasureResults { + sequence_number: 9, + response_type: 0x000b, + time_delta_ms: 40, + byte_count: 150, + }; + let reply = TunnelMessage::from_auto_detect_responses(&[response]).expect("bandwidth reply"); + let [sub_header] = reply.sub_headers.as_slice() else { + panic!("expected one sub-header, got {:?}", reply.sub_headers); + }; + assert_eq!( + encode_vec(sub_header).unwrap(), + [14, 1, 9, 0, 11, 0, 40, 0, 0, 0, 150, 0, 0, 0] + ); +}