From 27bddf6fb21123476e59fc05c836c048d98df765 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Mon, 28 Sep 2026 00:21:36 -0500 Subject: [PATCH 01/12] feat(rdpeudp): carry tunnel sub-headers through the UDP transport The RDPEMT codec already handled sub-headers, but UdpTransport exposed only a PDU's data, so the Continuous Auto-Detection messages MS-RDPBCGR 1.3.9 carries there could be neither sent nor received. TunnelMessage carries both; send_message and recv_message use it, and send/recv keep working on data alone. --- crates/ironrdp-rdpeudp-tokio/src/error.rs | 10 +- crates/ironrdp-rdpeudp-tokio/src/framed.rs | 21 +- crates/ironrdp-rdpeudp-tokio/src/lib.rs | 3 +- crates/ironrdp-rdpeudp-tokio/src/transport.rs | 204 +++++++++++++++--- crates/ironrdp-rdpeudp-tokio/src/tunnel.rs | 59 ++++- .../tests/rdpeudp_tokio.rs | 41 +++- 6 files changed, 290 insertions(+), 48 deletions(-) diff --git a/crates/ironrdp-rdpeudp-tokio/src/error.rs b/crates/ironrdp-rdpeudp-tokio/src/error.rs index 2ebe906487..2018af9bcc 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/error.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/error.rs @@ -114,6 +114,11 @@ pub enum UdpTransportErrorKind { /// A `send()` payload exceeds the wire `PayloadLength` field's 65535-byte /// capacity ([MS-RDPEMT] 2.2.2.3, `RDP_TUNNEL_DATA`). PayloadTooLarge { len: usize }, + + /// A `send_message()` message's sub-headers, `len` bytes encoded, do not + /// fit beside the 4-byte tunnel header in the one-byte `HeaderLength` + /// field ([MS-RDPEMT] 2.2.1.1). + SubHeadersTooLarge { len: usize }, } impl fmt::Display for UdpTransportErrorKind { @@ -140,6 +145,9 @@ impl fmt::Display for UdpTransportErrorKind { "send payload of {len} bytes exceeds the 65535-byte tunnel data limit" ) } + Self::SubHeadersTooLarge { len } => { + write!(f, "{len} bytes of sub-headers exceed the 251 a tunnel header holds") + } } } } @@ -157,7 +165,7 @@ impl core::error::Error for UdpTransportErrorKind { | Self::TunnelTimeout | Self::TunnelRejected { .. } | Self::DriverPanic => None, - Self::UnsupportedProtocol { .. } | Self::PayloadTooLarge { .. } => None, + Self::UnsupportedProtocol { .. } | Self::PayloadTooLarge { .. } | Self::SubHeadersTooLarge { .. } => None, } } } diff --git a/crates/ironrdp-rdpeudp-tokio/src/framed.rs b/crates/ironrdp-rdpeudp-tokio/src/framed.rs index f38f5ef361..4fff17426d 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/framed.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/framed.rs @@ -74,11 +74,12 @@ mod tests { use tokio::sync::mpsc; use super::*; + use crate::transport::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); - let (outgoing_tx, outgoing_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); @@ -88,7 +89,7 @@ mod tests { #[tokio::test] async fn framed_read_delivers_one_message() { let (mut transport, feeder, _) = test_transport(); - feeder.send(vec![0xDE, 0xAD, 0xBE, 0xEF]).await.unwrap(); + feeder.send(vec![0xDE, 0xAD, 0xBE, 0xEF].into()).await.unwrap(); let mut buf = BytesMut::new(); let n = FramedRead::read(&mut transport, &mut buf).await.unwrap(); @@ -120,8 +121,8 @@ mod tests { async fn framed_read_does_not_mistake_an_empty_message_for_eof() { let (mut transport, feeder, _) = test_transport(); - feeder.send(Vec::new()).await.unwrap(); - feeder.send(vec![0x11, 0x22]).await.unwrap(); + feeder.send(Vec::new().into()).await.unwrap(); + feeder.send(vec![0x11, 0x22].into()).await.unwrap(); let mut buf = BytesMut::new(); let n = FramedRead::read(&mut transport, &mut buf).await.unwrap(); @@ -135,7 +136,7 @@ mod tests { async fn framed_read_still_reports_eof_after_an_empty_message() { let (mut transport, feeder, _) = test_transport(); - feeder.send(Vec::new()).await.unwrap(); + feeder.send(Vec::new().into()).await.unwrap(); drop(feeder); let mut buf = BytesMut::new(); @@ -154,7 +155,7 @@ mod tests { .unwrap(); let data = receiver.recv().await.unwrap(); - assert_eq!(data, vec![0x01, 0x02, 0x03]); + assert_eq!(data, TunnelMessage::from(vec![0x01, 0x02, 0x03])); } #[tokio::test] @@ -171,8 +172,8 @@ mod tests { async fn framed_read_multiple_messages_accumulate() { let (mut transport, feeder, _) = test_transport(); - feeder.send(vec![0xAA, 0xBB]).await.unwrap(); - feeder.send(vec![0xCC, 0xDD]).await.unwrap(); + feeder.send(vec![0xAA, 0xBB].into()).await.unwrap(); + feeder.send(vec![0xCC, 0xDD].into()).await.unwrap(); let mut buf = BytesMut::new(); diff --git a/crates/ironrdp-rdpeudp-tokio/src/lib.rs b/crates/ironrdp-rdpeudp-tokio/src/lib.rs index 517975ccdd..20fece5f1a 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/lib.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/lib.rs @@ -15,5 +15,6 @@ pub(crate) mod tunnel; pub use self::error::{DriverError, DriverErrorKind, UdpTransportError, UdpTransportErrorKind}; pub use self::multitransport::MultitransportBootstrap; pub use self::transport::{ - UdpAcceptConfig, UdpTlsConfig, UdpTransport, UdpTransportConfig, UdpTransportSender, accept_udp, connect_udp, + TunnelMessage, UdpAcceptConfig, UdpTlsConfig, UdpTransport, UdpTransportConfig, UdpTransportSender, accept_udp, + connect_udp, }; diff --git a/crates/ironrdp-rdpeudp-tokio/src/transport.rs b/crates/ironrdp-rdpeudp-tokio/src/transport.rs index d5b9c79605..f3d42d184d 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/transport.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/transport.rs @@ -29,7 +29,10 @@ use tokio::task::JoinHandle; use tracing::{debug, info, trace, warn}; use crate::driver::Driver; -use crate::error::{DriverError, DriverErrorExt as _, DriverErrorKind, UdpTransportError, UdpTransportErrorExt as _}; +use crate::error::{ + DriverError, DriverErrorExt as _, DriverErrorKind, UdpTransportError, UdpTransportErrorExt as _, + UdpTransportErrorKind, +}; use crate::stream::{RdpeudpStream, SharedIo}; use crate::tls::{tls_accept, tls_upgrade}; use crate::tunnel::{read_tunnel_pdu, tunnel_data_loop, write_tunnel_pdu}; @@ -211,6 +214,28 @@ impl Drop for AbortOnDrop { } } +/// One RDPEMT Tunnel Data PDU's content: the higher-layer data and the +/// sub-headers carried beside it ([MS-RDPEMT] 2.2.2.3). +/// +/// The sub-headers carry the Continuous Auto-Detection messages that +/// [MS-RDPBCGR] 1.3.9 sends over a sideband channel in use, such as a +/// bandwidth measurement's Start and Stop and the client's results. `data` +/// may be empty when a message carries only sub-headers. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct TunnelMessage { + pub sub_headers: Vec, + pub data: Vec, +} + +impl From> for TunnelMessage { + fn from(data: Vec) -> Self { + Self { + sub_headers: Vec::new(), + data, + } + } +} + /// A cloneable handle for sending data over an established UDP transport, /// obtained from [`UdpTransport::sender`]. /// @@ -218,7 +243,7 @@ impl Drop for AbortOnDrop { /// and receiving already run over separate channels fed by separate /// background tasks, so this never contends with a concurrent `recv()`. #[derive(Clone)] -pub struct UdpTransportSender(mpsc::Sender>); +pub struct UdpTransportSender(mpsc::Sender); impl UdpTransportSender { /// Send a higher-layer data frame through the tunnel. @@ -231,13 +256,36 @@ impl UdpTransportSender { /// Returns `PayloadTooLarge` if `data` exceeds 65535 bytes, the wire /// `PayloadLength` field's capacity ([MS-RDPEMT] 2.2.2.3). pub async fn send(&self, data: Vec) -> Result<(), UdpTransportError> { - if data.len() > usize::from(u16::MAX) { - debug!(len = data.len(), "Rejected oversized tunnel payload"); - return Err(UdpTransportError::payload_too_large("send", data.len())); + self.send_message(TunnelMessage::from(data)).await + } + + /// Send higher-layer data through the tunnel with sub-headers beside it + /// in the same Tunnel Data PDU. + /// + /// # Errors + /// + /// As [`Self::send`], for `message.data`, and `SubHeadersTooLarge` if + /// the sub-headers do not fit the one-byte `HeaderLength` field + /// ([MS-RDPEMT] 2.2.1.1), checked here for the same reason. + pub async fn send_message(&self, message: TunnelMessage) -> Result<(), UdpTransportError> { + if message.data.len() > usize::from(u16::MAX) { + debug!(len = message.data.len(), "Rejected oversized tunnel payload"); + return Err(UdpTransportError::payload_too_large("send", message.data.len())); + } + let sub_headers_len = message + .sub_headers + .iter() + .map(ironrdp_rdpemt::TunnelSubHeader::wire_size) + .sum::(); + if 4 /* RDP_TUNNEL_HEADER */ + sub_headers_len > usize::from(u8::MAX) { + return Err(UdpTransportError::new( + "send", + UdpTransportErrorKind::SubHeadersTooLarge { len: sub_headers_len }, + )); } self.0 - .send(data) + .send(message) .await .map_err(|_| UdpTransportError::driver("send", DriverError::connection_closed("send"))) } @@ -252,10 +300,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, /// Sends higher-layer data into the tunnel for encryption and transmission. - data_tx: mpsc::Sender>, + data_tx: mpsc::Sender, /// Shared I/O bridge between the driver and the TLS/RDPEMT layer. /// Held here so `shutdown()` can signal closure to both the @@ -275,8 +323,16 @@ pub struct UdpTransport { impl UdpTransport { /// Receive the next higher-layer data frame from the tunnel. /// - /// Returns `None` when the tunnel is closed. + /// Returns `None` when the tunnel is closed. Any sub-headers that came + /// with the frame are dropped; use [`Self::recv_message`] to keep them. pub async fn recv(&mut self) -> Option> { + self.recv_message().await.map(|message| message.data) + } + + /// Receive the next Tunnel Data PDU's content, sub-headers included. + /// + /// Returns `None` when the tunnel is closed. + pub async fn recv_message(&mut self) -> Option { self.data_rx.recv().await } @@ -296,6 +352,16 @@ impl UdpTransport { self.sender().send(data).await } + /// Send higher-layer data through the tunnel with sub-headers beside it + /// in the same Tunnel Data PDU. + /// + /// # Errors + /// + /// As [`UdpTransportSender::send_message`]. + pub async fn send_message(&self, message: TunnelMessage) -> Result<(), UdpTransportError> { + self.sender().send_message(message).await + } + /// Returns a cloneable handle for sending data, independent of this /// object's `&mut self`-requiring [`Self::recv`]. /// @@ -380,7 +446,7 @@ 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, data_tx, @@ -548,8 +614,8 @@ pub async fn connect_udp(config: UdpTransportConfig) -> Result>(64); - let (outgoing_tx, mut outgoing_rx) = mpsc::channel::>(64); + let (incoming_tx, incoming_rx) = mpsc::channel::(64); + let (outgoing_tx, mut outgoing_rx) = mpsc::channel::(64); // Read pump: TLS → RDPEMT decode → channel let pump_handle = AbortOnDrop::new(tokio::spawn(async move { @@ -740,8 +806,8 @@ async fn accept_udp_inner(socket: UdpSocket, config: UdpAcceptConfig) -> 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 (outgoing_tx, mut outgoing_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 { tunnel_data_loop(&mut tls_read, &mut tunnel, &incoming_tx).await @@ -818,15 +884,19 @@ where /// Shared by both `connect_udp` and `accept_udp`. Returns the first failure /// encountered rather than only logging it, so `shutdown()` can surface it /// to the caller instead of the pump silently going quiet. -async fn write_pump(tls_write: &mut W, outgoing_rx: &mut mpsc::Receiver>) -> Result<(), UdpTransportError> +async fn write_pump( + tls_write: &mut W, + outgoing_rx: &mut mpsc::Receiver, +) -> Result<(), UdpTransportError> where W: tokio::io::AsyncWrite + Unpin, { - while let Some(data) = outgoing_rx.recv().await { - let len = data.len(); + while let Some(message) = outgoing_rx.recv().await { + let len = message.data.len(); + let sub_headers = message.sub_headers.len(); let pdu = ironrdp_rdpemt::TunnelData { - sub_headers: Vec::new(), - higher_layer_data: data, + sub_headers: message.sub_headers, + higher_layer_data: message.data, }; let encoded = ironrdp_core::encode_vec(&pdu) .map_err(|error| UdpTransportError::rdpemt("write pump", ironrdp_rdpemt::RdpemtError::encode(error)))?; @@ -838,7 +908,7 @@ where debug!(%error, "Write pump failed to flush tunnel data"); UdpTransportError::tls("write pump", error) })?; - trace!(len, encoded_len = encoded.len(), "Sent tunnel data"); + trace!(len, sub_headers, encoded_len = encoded.len(), "Sent tunnel data"); } debug!("Write pump stopped, send channel closed"); Ok(()) @@ -917,8 +987,8 @@ mod tests { static DRIVER_RUNNING: AtomicBool = AtomicBool::new(false); DRIVER_RUNNING.store(false, Ordering::SeqCst); - let (_incoming_tx, incoming_rx) = mpsc::channel::>(4); - let (outgoing_tx, _outgoing_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, @@ -960,7 +1030,7 @@ mod tests { let error = driver_exit_during_handshake("test", Ok(Err(driver_error))); assert!( - matches!(error.kind(), crate::error::UdpTransportErrorKind::Handshake(_)), + matches!(error.kind(), UdpTransportErrorKind::Handshake(_)), "got {error:?}, expected a Handshake error carrying the driver's own cause" ); } @@ -973,7 +1043,7 @@ mod tests { let error = driver_exit_during_handshake("test", Ok(Ok(()))); assert!( - matches!(error.kind(), crate::error::UdpTransportErrorKind::Handshake(_)), + matches!(error.kind(), UdpTransportErrorKind::Handshake(_)), "got {error:?}, expected a Handshake error for an unexpectedly-clean driver exit" ); } @@ -988,7 +1058,7 @@ mod tests { let error = driver_exit_during_handshake("test", join_result); assert!( - matches!(error.kind(), crate::error::UdpTransportErrorKind::DriverPanic), + matches!(error.kind(), UdpTransportErrorKind::DriverPanic), "got {error:?}, expected DriverPanic" ); } @@ -1031,8 +1101,90 @@ mod tests { .expect("the driver branch should resolve almost immediately, well inside 5 seconds"); assert!( - matches!(outcome.kind(), crate::error::UdpTransportErrorKind::Handshake(_)), + matches!(outcome.kind(), UdpTransportErrorKind::Handshake(_)), "got {outcome:?}, expected the driver's real error to surface" ); } + + /// A message's sub-headers go into the Tunnel Data PDU beside its data. + #[tokio::test] + async fn the_write_pump_encodes_sub_headers() { + use ironrdp_rdpemt::{SubHeaderType, TunnelData, TunnelSubHeader}; + use tokio::io::AsyncReadExt as _; + + let sub_header = TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectRequest, + data: vec![0x06, 0x00, 0x07, 0x00, 0x14, 0x00], + }; + let (tx, mut rx) = mpsc::channel(4); + tx.send(TunnelMessage { + sub_headers: vec![sub_header.clone()], + data: vec![0xaa, 0xbb], + }) + .await + .expect("queue message"); + drop(tx); + + let (mut writer, mut reader) = tokio::io::duplex(1024); + write_pump(&mut writer, &mut rx).await.expect("pump drains"); + drop(writer); + + let mut wire = Vec::new(); + reader.read_to_end(&mut wire).await.expect("read back"); + let pdu: TunnelData = ironrdp_core::decode(&wire).expect("decode"); + assert_eq!(pdu.sub_headers, vec![sub_header]); + assert_eq!(pdu.higher_layer_data, vec![0xaa, 0xbb]); + } + + /// `recv` keeps its old shape and drops sub-headers; `recv_message` keeps them. + #[tokio::test] + async fn recv_message_keeps_what_recv_drops() { + use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader}; + + 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); + let message = TunnelMessage { + sub_headers: vec![TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectResponse, + data: vec![0x01], + }], + data: vec![0x02], + }; + + incoming_tx.send(message.clone()).await.expect("queue"); + incoming_tx.send(message.clone()).await.expect("queue"); + + assert_eq!(transport.recv().await, Some(vec![0x02])); + assert_eq!(transport.recv_message().await, Some(message)); + } + + /// Sub-headers that would overflow `HeaderLength` are refused at the + /// call, not left to fail the write pump; the largest that fit go out. + #[tokio::test] + async fn send_message_refuses_sub_headers_the_header_cannot_hold() { + use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader}; + + let (_incoming_tx, incoming_rx) = mpsc::channel(4); + let (outgoing_tx, mut outgoing_rx) = mpsc::channel(4); + let transport = UdpTransport::from_channels(incoming_rx, outgoing_tx); + let message = |data_len| TunnelMessage { + sub_headers: vec![TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectRequest, + data: vec![0; data_len], + }], + data: Vec::new(), + }; + + // 4 (tunnel header) + 2 (sub-header header) + 249 = 255. + transport.send_message(message(249)).await.expect("fits"); + let error = transport.send_message(message(250)).await.expect_err("one byte over"); + assert!(matches!( + error.kind(), + UdpTransportErrorKind::SubHeadersTooLarge { len: 252 } + )); + + assert_eq!(outgoing_rx.recv().await, Some(message(249))); + assert!(outgoing_rx.try_recv().is_err(), "the refused message was not queued"); + } } diff --git a/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs b/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs index 89c682bcbe..654c66b4ab 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs @@ -15,6 +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; /// Read a complete RDPEMT PDU from the stream using self-framing. /// @@ -128,7 +129,7 @@ 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, @@ -154,13 +155,16 @@ where while let Some(event) = tunnel.poll_event() { match event { - // `sub_headers` (e.g. auto-detect bandwidth measurement, MS-RDPBCGR - // 2.2.14) are not consumed here; this driver only wires the DVC - // payload through. A future auto-detect integration would need to - // dispatch them instead of discarding them. - TunnelEvent::Data { data, .. } => { - trace!(len = data.len(), "Forwarding tunnel data"); - if data_tx.send(data).await.is_err() { + // Sub-headers go through with the data: they carry the + // auto-detect messages ([MS-RDPBCGR] 1.3.9) of a sideband + // channel in use, which the application answers. + TunnelEvent::Data { sub_headers, data } => { + trace!( + len = data.len(), + sub_headers = sub_headers.len(), + "Forwarding tunnel data" + ); + if data_tx.send(TunnelMessage { sub_headers, data }).await.is_err() { debug!("Tunnel data receiver dropped, stopping read pump"); // Application dropped the receiver return Ok(()); @@ -263,4 +267,43 @@ mod tests { let mut cursor = io::Cursor::new(vec![0x02, 0x05, 0x00, 0x04, 0x48, 0x65]); assert!(read_tunnel_pdu(&mut cursor).await.is_err()); } + + /// Sub-headers reach the application with the data they came with. They + /// carry the auto-detect messages of a sideband channel in use + /// ([MS-RDPBCGR] 1.3.9), which used to be dropped here. + #[tokio::test] + async fn the_data_loop_forwards_sub_headers() { + use ironrdp_rdpemt::{SubHeaderType, TunnelConfig, TunnelCreateResponse, TunnelData, TunnelSubHeader}; + + let mut tunnel = RdpemtTunnel::client(TunnelConfig { + request_id: 1, + security_cookie: [0; 16], + }); + let response = ironrdp_core::encode_vec(&TunnelCreateResponse { + hr_response: TunnelCreateResponse::S_OK, + }) + .expect("encode response"); + tunnel.handle_pdu(&response).expect("tunnel established"); + while tunnel.poll_event().is_some() {} + + let sub_header = TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectResponse, + data: vec![0x0e, 0x00, 0x05, 0x00], + }; + let wire = ironrdp_core::encode_vec(&TunnelData { + sub_headers: vec![sub_header.clone()], + higher_layer_data: Vec::new(), + }) + .expect("encode data"); + + let (tx, mut rx) = tokio::sync::mpsc::channel(4); + let mut stream = io::Cursor::new(wire); + tunnel_data_loop(&mut stream, &mut tunnel, &tx) + .await + .expect("clean EOF"); + + let message = rx.recv().await.expect("one message"); + assert_eq!(message.sub_headers, vec![sub_header]); + assert!(message.data.is_empty()); + } } diff --git a/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs b/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs index b207ba923e..421f38e97a 100644 --- a/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs +++ b/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs @@ -11,10 +11,11 @@ use core::sync::atomic::{AtomicUsize, Ordering}; use core::time::Duration; use std::sync::Arc; -use ironrdp_rdpemt::TunnelConfig; +use ironrdp_rdpemt::{SubHeaderType, TunnelConfig, TunnelSubHeader}; use ironrdp_rdpeudp::ConnectionConfig; use ironrdp_rdpeudp_tokio::{ - MultitransportBootstrap, UdpAcceptConfig, UdpTlsConfig, UdpTransport, UdpTransportConfig, accept_udp, connect_udp, + MultitransportBootstrap, TunnelMessage, UdpAcceptConfig, UdpTlsConfig, UdpTransport, UdpTransportConfig, + accept_udp, connect_udp, }; use ironrdp_tls::{CertificateValidation, CertificateValidationCallback}; use tokio::net::UdpSocket; @@ -222,6 +223,42 @@ async fn full_stack_bidirectional_data() { server.shutdown().await.expect("server shutdown"); } +/// Sub-headers cross the full stack beside their data, and a message may +/// carry sub-headers alone, as an auto-detect Start or Stop does. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn full_stack_sub_headers() { + let (mut client, server) = establish_loopback_pair_on("127.0.0.1:0").await; + + let start = TunnelMessage { + sub_headers: vec![TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectRequest, + data: vec![0x06, 0x00, 0x07, 0x00, 0x14, 0x00], + }], + data: Vec::new(), + }; + let graphics = TunnelMessage { + sub_headers: vec![TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectRequest, + data: vec![0x08], + }], + data: vec![0x01, 0x02, 0x03], + }; + server + .send_message(start.clone()) + .await + .expect("send sub-headers alone"); + server + .send_message(graphics.clone()) + .await + .expect("send sub-headers with data"); + + assert_eq!(client.recv_message().await.expect("recv"), start); + assert_eq!(client.recv_message().await.expect("recv"), graphics); + + client.shutdown().await.expect("client shutdown"); + server.shutdown().await.expect("server shutdown"); +} + /// A payload over the wire `PayloadLength` field's 65535-byte capacity must /// be rejected synchronously by `send()`, and must not take the write pump /// down with it: a normal-sized send right after must still succeed. From 505914bb67099333a6ced623363fa60110c5286b Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Mon, 28 Sep 2026 01:26:57 -0500 Subject: [PATCH 02/12] test(rdpeudp): use a real Bandwidth Measure Start's sub-header data The sub-header examples carried a whole Bandwidth Measure Start, header bytes included. An auto-detect sub-header is the auto-detect structure itself (MS-RDPEMT 2.2.1.1.1): its SubHeaderLength and SubHeaderType are the structure's headerLength and headerTypeId, so on the wire those examples would repeat the header. The transport treats the data as opaque and the tests passed either way; the examples now carry only what follows the header. --- crates/ironrdp-rdpeudp-tokio/src/transport.rs | 4 +++- crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/crates/ironrdp-rdpeudp-tokio/src/transport.rs b/crates/ironrdp-rdpeudp-tokio/src/transport.rs index f3d42d184d..1ceb0b9d88 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/transport.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/transport.rs @@ -1114,7 +1114,9 @@ mod tests { let sub_header = TunnelSubHeader { sub_header_type: SubHeaderType::AutoDetectRequest, - data: vec![0x06, 0x00, 0x07, 0x00, 0x14, 0x00], + // A Bandwidth Measure Start: sequenceNumber 7, requestType 0x0014. The + // sub-header's own two bytes are its headerLength and headerTypeId. + data: vec![0x07, 0x00, 0x14, 0x00], }; let (tx, mut rx) = mpsc::channel(4); tx.send(TunnelMessage { diff --git a/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs b/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs index 421f38e97a..a0069ef8c0 100644 --- a/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs +++ b/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs @@ -232,7 +232,9 @@ async fn full_stack_sub_headers() { let start = TunnelMessage { sub_headers: vec![TunnelSubHeader { sub_header_type: SubHeaderType::AutoDetectRequest, - data: vec![0x06, 0x00, 0x07, 0x00, 0x14, 0x00], + // A Bandwidth Measure Start: sequenceNumber 7, requestType 0x0014. The + // sub-header's own two bytes are its headerLength and headerTypeId. + data: vec![0x07, 0x00, 0x14, 0x00], }], data: Vec::new(), }; From 72705d90be8831ff84a5e1ff5764851bf8e9746e Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Mon, 28 Sep 2026 09:58:20 -0500 Subject: [PATCH 03/12] review: drop the unused Default derive and correct the FramedRead comment Nothing builds a default TunnelMessage, and a default one would be an empty Tunnel Data PDU. The FramedRead comment said the tunnel consumes the sub-headers before this point; recv drops them, and a caller that needs them uses recv_message. --- crates/ironrdp-rdpeudp-tokio/src/framed.rs | 6 +++--- crates/ironrdp-rdpeudp-tokio/src/transport.rs | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/ironrdp-rdpeudp-tokio/src/framed.rs b/crates/ironrdp-rdpeudp-tokio/src/framed.rs index 4fff17426d..ea74fbcd8f 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/framed.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/framed.rs @@ -33,9 +33,9 @@ impl FramedRead for UdpTransport { // puts no minimum on HigherLayerData, and [MS-RDPBCGR] 1.3.9 sends // the four Continuous Auto-Detection messages "encapsulated in the // RDP_TUNNEL_SUBHEADER structure ... over the sideband channels - // that are in active use". The tunnel has already taken what it - // needs from those subheaders by the time we get here, leaving a - // payload of nothing to pass on. + // that are in active use". `recv` drops those subheaders (a caller + // that needs them uses `recv_message`), leaving a payload of + // nothing to pass on. loop { match self.recv().await { Some(data) if data.is_empty() => { diff --git a/crates/ironrdp-rdpeudp-tokio/src/transport.rs b/crates/ironrdp-rdpeudp-tokio/src/transport.rs index 1ceb0b9d88..f171da5928 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/transport.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/transport.rs @@ -221,7 +221,7 @@ impl Drop for AbortOnDrop { /// [MS-RDPBCGR] 1.3.9 sends over a sideband channel in use, such as a /// bandwidth measurement's Start and Stop and the client's results. `data` /// may be empty when a message carries only sub-headers. -#[derive(Debug, Clone, Default, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct TunnelMessage { pub sub_headers: Vec, pub data: Vec, From efe177750c9b0a40c00e944e13edf448602fb3d6 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Tue, 29 Sep 2026 16:22:06 -0500 Subject: [PATCH 04/12] review: pin the send_message header bound to the TunnelData encoder The send_message precheck restates the HeaderLength bound that ironrdp-rdpemt's TunnelData encoder owns. The boundary test now also runs the largest accepted message and the smallest refused one through the real encoder, so the two cannot drift apart unnoticed, and the comment on the check cites MS-RDPEMT 2.2.1.1. --- crates/ironrdp-rdpeudp-tokio/src/transport.rs | 24 +++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/crates/ironrdp-rdpeudp-tokio/src/transport.rs b/crates/ironrdp-rdpeudp-tokio/src/transport.rs index f171da5928..0a2b86be15 100644 --- a/crates/ironrdp-rdpeudp-tokio/src/transport.rs +++ b/crates/ironrdp-rdpeudp-tokio/src/transport.rs @@ -277,6 +277,9 @@ impl UdpTransportSender { .iter() .map(ironrdp_rdpemt::TunnelSubHeader::wire_size) .sum::(); + // HeaderLength is one byte and counts the 4 byte fixed part of the tunnel + // header plus the sub-headers ([MS-RDPEMT] 2.2.1.1). The write pump's + // TunnelData encoder enforces the same bound; a test pins the two together. if 4 /* RDP_TUNNEL_HEADER */ + sub_headers_len > usize::from(u8::MAX) { return Err(UdpTransportError::new( "send", @@ -1162,10 +1165,12 @@ mod tests { } /// Sub-headers that would overflow `HeaderLength` are refused at the - /// call, not left to fail the write pump; the largest that fit go out. + /// 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 + /// precheck cannot drift from the bound the write pump enforces. #[tokio::test] async fn send_message_refuses_sub_headers_the_header_cannot_hold() { - use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader}; + use ironrdp_rdpemt::{SubHeaderType, TunnelData, TunnelSubHeader}; let (_incoming_tx, incoming_rx) = mpsc::channel(4); let (outgoing_tx, mut outgoing_rx) = mpsc::channel(4); @@ -1188,5 +1193,20 @@ mod tests { assert_eq!(outgoing_rx.recv().await, Some(message(249))); assert!(outgoing_rx.try_recv().is_err(), "the refused message was not queued"); + + let encode = |message: TunnelMessage| { + ironrdp_core::encode_vec(&TunnelData { + sub_headers: message.sub_headers, + higher_layer_data: message.data, + }) + }; + assert!( + encode(message(249)).is_ok(), + "the encoder accepts what the precheck accepts" + ); + assert!( + encode(message(250)).is_err(), + "the encoder refuses what the precheck refuses" + ); } } From f4ca57fe87055c2c908bb62634873c5422085d2f Mon Sep 17 00:00:00 2001 From: AKolenda Date: Fri, 25 Sep 2026 23:41:09 -0600 Subject: [PATCH 05/12] feat(session): answer bandwidth measurements during the session The session answered RTT requests on the message channel but ignored Bandwidth Measure Start and Stop, so a server running continuous network auto-detection (MS-RDPBCGR 2.2.14, 3.2.5.14) never got a Bandwidth Measure Results PDU back. Windows runs continuous detection by default. The x224 processor now keeps a bandwidth window: - A continuous Start (0x0014) opens a window, and every server byte received until the Stop (0x0429) is counted. Following 3.2.5.14, only the bytes after a Basic Security Header count on the message channel. On the IO and static channels the MCS user data is counted, and on the fast path the data after the fast-path header, which is what FreeRDP counts too. - A connect-time Start (0x1014) counts Bandwidth Measure Payload PDUs and the 0x002B Stop the way the connector does: payloadLength plus the eight header bytes. - A repeated Start restarts the window, and the Results report the elapsed time floored at 1 ms (the server divides by it) and the byte count. - Lossy (0x0114/0x0629) requests belong to a lossy tunnel and are not answered on this channel. Timing follows the connector: the caller passes the frame's arrival time through the new `ActiveStage::process_with_timestamp` and `x224::Processor::process_with_timestamp`, and the client passes the time its framed reader filled the buffer. Without a timestamp no window opens and a Stop is answered with a zero-byte, 1 ms result instead of a figure the client never measured. `process` keeps working that way. --- crates/ironrdp-client/src/rdp.rs | 3 +- crates/ironrdp-session/src/active_stage.rs | 32 +++- crates/ironrdp-session/src/x224/mod.rs | 159 +++++++++++++++--- .../tests/session/autodetect.rs | 126 +++++++++++++- 4 files changed, 288 insertions(+), 32 deletions(-) diff --git a/crates/ironrdp-client/src/rdp.rs b/crates/ironrdp-client/src/rdp.rs index d184f13b11..7ab66aed1c 100644 --- a/crates/ironrdp-client/src/rdp.rs +++ b/crates/ironrdp-client/src/rdp.rs @@ -3184,7 +3184,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); diff --git a/crates/ironrdp-session/src/active_stage.rs b/crates/ironrdp-session/src/active_stage.rs index 050111d751..dffa561f59 100644 --- a/crates/ironrdp-session/src/active_stage.rs +++ b/crates/ironrdp-session/src/active_stage.rs @@ -1,12 +1,13 @@ use std::sync::Arc; use ironrdp_bulk::{BulkCompressor, CompressionType as BulkCompressionType}; -use ironrdp_core::{ReadCursor, WriteBuf}; +use ironrdp_core::{Decode as _, MonotonicInstant, ReadCursor, WriteBuf}; use ironrdp_displaycontrol::client::DisplayControlClient; use ironrdp_dvc::pdu::SoftSyncTunnelType; use ironrdp_dvc::{DrdynvcClient, DvcClientProcessor, DvcMessageBatch, DynamicChannelMut, DynamicChannelRef}; use ironrdp_egfx::client::GraphicsPipelineClient; use ironrdp_graphics::pointer::DecodedPointer; +use ironrdp_pdu::fast_path::FastPathHeader; use ironrdp_pdu::gcc::{ChannelName, Monitor}; use ironrdp_pdu::geometry::{ExclusiveRectangle, InclusiveRectangle, Rectangle as _}; use ironrdp_pdu::input::fast_path::{FastPathInput, FastPathInputEvent}; @@ -183,15 +184,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 => { + // A continuous bandwidth measurement counts what follows the fast-path header. + let mut header = ReadCursor::new(frame); + FastPathHeader::decode(&mut header).map_err(SessionError::decode)?; + self.x224_processor.record_bandwidth_bytes(header.len()); let mut output = WriteBuf::new(); let processor_updates = self.fast_path_processor @@ -202,7 +226,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(); @@ -1066,7 +1092,7 @@ mod tests { use core::any::TypeId; use super::*; - use ironrdp_core::{Decode as _, encode_vec}; + use ironrdp_core::encode_vec; use ironrdp_displaycontrol::pdu::{DisplayControlCapabilities, DisplayControlPdu}; use ironrdp_dvc::pdu::{ ClosePdu, CreateRequestPdu, DataPdu, DrdynvcDataPdu, DrdynvcServerPdu, SoftSyncChannelList, SoftSyncRequestPdu, diff --git a/crates/ironrdp-session/src/x224/mod.rs b/crates/ironrdp-session/src/x224/mod.rs index d006393ba4..08f3076807 100644 --- a/crates/ironrdp-session/src/x224/mod.rs +++ b/crates/ironrdp-session/src/x224/mod.rs @@ -1,9 +1,12 @@ 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}; -use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu}; +use ironrdp_pdu::rdp::autodetect::{ + AutoDetectReqPdu, AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu, BW_RESULTS_CONNECT_TIME, + BW_RESULTS_CONTINUOUS, BW_START_CONNECT_TIME, BW_START_RELIABLE_UDP, BW_STOP_CONNECT_TIME, BW_STOP_RELIABLE_UDP, +}; use ironrdp_pdu::rdp::client_info::CompressionType; use ironrdp_pdu::rdp::headers::{ BasicSecurityHeader, BasicSecurityHeaderFlags, CompressionFlags, IoChannelPdu, ShareDataCtx, ShareDataPdu, @@ -72,7 +75,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 +110,13 @@ pub struct Processor { io_channel_id: u16, message_channel_id: Option, share_id: u32, + bandwidth: Option, +} + +struct BandwidthMeasurement { + started_at: MonotonicInstant, + bytes: u32, + continuous: bool, } impl Processor { @@ -123,6 +133,7 @@ impl Processor { io_channel_id, message_channel_id, share_id, + bandwidth: None, } } @@ -212,6 +223,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, @@ -232,10 +256,15 @@ impl Processor { }; let channel_id = data_ctx.channel_id; + // Message-channel PDUs carry a Basic Security Header, excluded below. + // Ordinary session data on TLS-protected IO/SVC channels does not. + if self.message_channel_id != Some(channel_id) { + self.record_bandwidth_bytes(data_ctx.user_data.len()); + } if channel_id == self.io_channel_id { self.process_io_channel_data_indication(data_ctx, 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 { let maximum_chunk_size = self.static_channels.maximum_chunk_size(); if let Some(svc) = self.static_channels.get_by_channel_id_mut(channel_id) { @@ -445,6 +474,16 @@ impl Processor { Ok(decompressed) } + /// Counts session bytes after transport/security headers while a continuous + /// bandwidth window is open ([MS-RDPBCGR] 3.2.5.14). + pub(crate) fn record_bandwidth_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)); + } + } + /// 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 +496,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 +539,87 @@ impl Processor { let req = decode::(data_ctx.user_data).map_err(SessionError::decode)?; - match req.request { + let response = 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())]) + 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, + }); + return Ok(Vec::new()); + } + AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => { + if let Some(measurement) = self.bandwidth.as_mut().filter(|measurement| !measurement.continuous) { + // [MS-RDPBCGR] 3.2.5.14 counts the eight-byte auto-detect + // header as well as payloadLength, but not the security header. + measurement.bytes = measurement + .bytes + .saturating_add(u32::try_from(payload.len()).unwrap_or(u32::MAX).saturating_add(8)); + } + return Ok(Vec::new()); + } + 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; + let measurement = self + .bandwidth + .take() + .filter(|measurement| measurement.continuous == continuous); + let stop_bytes = if continuous { + 0 + } else { + u32::try_from(payload.as_ref().map_or(0, Vec::len)) + .unwrap_or(u32::MAX) + .saturating_add(8) + }; + 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(1), + measurement.bytes.saturating_add(stop_bytes), + ), + _ => (1, stop_bytes), + }; + AutoDetectResponse::BandwidthMeasureResults { + sequence_number, + response_type: if continuous { + BW_RESULTS_CONTINUOUS + } else { + BW_RESULTS_CONNECT_TIME + }, + time_delta_ms, + byte_count, + } } req @ AutoDetectRequest::NetworkCharacteristicsResult { .. } => { debug!(?req, "Received network characteristics from server"); - Ok(vec![ProcessorOutput::AutoDetect(req)]) + return Ok(vec![ProcessorOutput::AutoDetect(req)]); } req => { - debug!(?req, "Auto-detect request not yet implemented"); - Ok(Vec::new()) + // Lossy variants belong to a lossy tunnel, never this channel. + debug!(?req, "Ignoring auto-detect request for another transport"); + 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 +740,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!( diff --git a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs index 4591f3fe92..00f93d0285 100644 --- a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs @@ -130,7 +130,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 +142,127 @@ 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"); + }; + 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_counts_data_without_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 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 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)); +} From 82d81037003d58f86e9dc5193831cb673f1f0978 Mon Sep 17 00:00:00 2001 From: AKolenda Date: Mon, 28 Sep 2026 00:33:42 -0600 Subject: [PATCH 06/12] refactor(session): answer auto-detect requests through one responder Move the handling of RTT and bandwidth measurement requests out of the x224 processor into an `AutoDetectResponder`, so the message channel and a multitransport tunnel can answer them with the same code. Each transport keeps its own responder, because a continuous measurement counts the data received on its own transport ([MS-RDPBCGR] 3.2.5.14). No behavior change on the message channel. --- crates/ironrdp-session/src/autodetect.rs | 122 +++++++++++++++++++++++ crates/ironrdp-session/src/lib.rs | 1 + crates/ironrdp-session/src/x224/mod.rs | 98 ++---------------- 3 files changed, 134 insertions(+), 87 deletions(-) create mode 100644 crates/ironrdp-session/src/autodetect.rs diff --git a/crates/ironrdp-session/src/autodetect.rs b/crates/ironrdp-session/src/autodetect.rs new file mode 100644 index 0000000000..ccc0a3105a --- /dev/null +++ b/crates/ironrdp-session/src/autodetect.rs @@ -0,0 +1,122 @@ +//! 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 that arrive on one transport. +/// +/// Each transport keeps its own, because a continuous measurement counts the data received on +/// the transport it runs on ([MS-RDPBCGR] 3.2.5.14). Timestamps come from the caller, from the +/// same monotonic clock for every request on that 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, +} + +#[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) { + // [MS-RDPBCGR] 3.2.5.14 counts the eight-byte auto-detect + // header as well as payloadLength, but not the security header. + measurement.bytes = measurement + .bytes + .saturating_add(u32::try_from(payload.len()).unwrap_or(u32::MAX).saturating_add(8)); + } + 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; + let measurement = self + .bandwidth + .take() + .filter(|measurement| measurement.continuous == continuous); + let stop_bytes = if continuous { + 0 + } else { + u32::try_from(payload.as_ref().map_or(0, Vec::len)) + .unwrap_or(u32::MAX) + .saturating_add(8) + }; + 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(1), + measurement.bytes.saturating_add(stop_bytes), + ), + _ => (1, stop_bytes), + }; + 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 { .. } => { + 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 08f3076807..a85b55ac6b 100644 --- a/crates/ironrdp-session/src/x224/mod.rs +++ b/crates/ironrdp-session/src/x224/mod.rs @@ -3,10 +3,7 @@ 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}; -use ironrdp_pdu::rdp::autodetect::{ - AutoDetectReqPdu, AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu, BW_RESULTS_CONNECT_TIME, - BW_RESULTS_CONTINUOUS, BW_START_CONNECT_TIME, BW_START_RELIABLE_UDP, BW_STOP_CONNECT_TIME, BW_STOP_RELIABLE_UDP, -}; +use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest, AutoDetectRspPdu}; use ironrdp_pdu::rdp::client_info::CompressionType; use ironrdp_pdu::rdp::headers::{ BasicSecurityHeader, BasicSecurityHeaderFlags, CompressionFlags, IoChannelPdu, ShareDataCtx, ShareDataPdu, @@ -21,6 +18,7 @@ use ironrdp_svc::{ }; use tracing::debug; +use crate::autodetect::AutoDetectResponder; use crate::{SessionError, SessionErrorExt as _, SessionResult, reason_err}; /// X224 Processor output @@ -110,13 +108,7 @@ pub struct Processor { io_channel_id: u16, message_channel_id: Option, share_id: u32, - bandwidth: Option, -} - -struct BandwidthMeasurement { - started_at: MonotonicInstant, - bytes: u32, - continuous: bool, + auto_detect: AutoDetectResponder, } impl Processor { @@ -133,7 +125,7 @@ impl Processor { io_channel_id, message_channel_id, share_id, - bandwidth: None, + auto_detect: AutoDetectResponder::default(), } } @@ -477,11 +469,7 @@ impl Processor { /// Counts session bytes after transport/security headers while a continuous /// bandwidth window is open ([MS-RDPBCGR] 3.2.5.14). pub(crate) fn record_bandwidth_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)); - } + self.auto_detect.record_bytes(bytes); } /// Process a PDU received on the MCS message channel: auto-detect @@ -539,76 +527,12 @@ impl Processor { let req = decode::(data_ctx.user_data).map_err(SessionError::decode)?; - let response = match req.request { - AutoDetectRequest::RttRequest { sequence_number, .. } => { - 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, - }); - return Ok(Vec::new()); - } - AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => { - if let Some(measurement) = self.bandwidth.as_mut().filter(|measurement| !measurement.continuous) { - // [MS-RDPBCGR] 3.2.5.14 counts the eight-byte auto-detect - // header as well as payloadLength, but not the security header. - measurement.bytes = measurement - .bytes - .saturating_add(u32::try_from(payload.len()).unwrap_or(u32::MAX).saturating_add(8)); - } - return Ok(Vec::new()); - } - 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; - let measurement = self - .bandwidth - .take() - .filter(|measurement| measurement.continuous == continuous); - let stop_bytes = if continuous { - 0 - } else { - u32::try_from(payload.as_ref().map_or(0, Vec::len)) - .unwrap_or(u32::MAX) - .saturating_add(8) - }; - 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(1), - measurement.bytes.saturating_add(stop_bytes), - ), - _ => (1, stop_bytes), - }; - AutoDetectResponse::BandwidthMeasureResults { - sequence_number, - response_type: if continuous { - BW_RESULTS_CONTINUOUS - } else { - BW_RESULTS_CONNECT_TIME - }, - time_delta_ms, - byte_count, - } - } - req @ AutoDetectRequest::NetworkCharacteristicsResult { .. } => { - debug!(?req, "Received network characteristics from server"); - return Ok(vec![ProcessorOutput::AutoDetect(req)]); - } - req => { - // Lossy variants belong to a lossy tunnel, never this channel. - debug!(?req, "Ignoring auto-detect request for another transport"); - return 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(); From 2898221b3fda1780aa38b31b55407b469d0ebda3 Mon Sep 17 00:00:00 2001 From: AKolenda Date: Mon, 28 Sep 2026 13:17:37 -0600 Subject: [PATCH 07/12] refactor(session): tidy the bandwidth responder Decode the fast-path header for the byte count only while a continuous measurement is running, since `fast_path::Processor::process` decodes it again. Keep the payloadLength plus header rule in one `counted_len` helper, and log a timed window that is dropped because its Stop has no arrival time, as the connector does. Adds a test for fast-path counting. --- crates/ironrdp-session/src/active_stage.rs | 10 ++- crates/ironrdp-session/src/autodetect.rs | 50 +++++++++--- crates/ironrdp-session/src/x224/mod.rs | 5 ++ .../tests/session/autodetect.rs | 79 +++++++++++++++++++ 4 files changed, 131 insertions(+), 13 deletions(-) diff --git a/crates/ironrdp-session/src/active_stage.rs b/crates/ironrdp-session/src/active_stage.rs index dffa561f59..6a5660c662 100644 --- a/crates/ironrdp-session/src/active_stage.rs +++ b/crates/ironrdp-session/src/active_stage.rs @@ -213,9 +213,13 @@ impl ActiveStage { let (mut stage_outputs, processor_updates) = match action { Action::FastPath => { // A continuous bandwidth measurement counts what follows the fast-path header. - let mut header = ReadCursor::new(frame); - FastPathHeader::decode(&mut header).map_err(SessionError::decode)?; - self.x224_processor.record_bandwidth_bytes(header.len()); + // The header is decoded again by `process`, so only pay for it here while a + // measurement is running. + if self.x224_processor.is_counting_bandwidth() { + let mut header = ReadCursor::new(frame); + FastPathHeader::decode(&mut header).map_err(SessionError::decode)?; + self.x224_processor.record_bandwidth_bytes(header.len()); + } let mut output = WriteBuf::new(); let processor_updates = self.fast_path_processor diff --git a/crates/ironrdp-session/src/autodetect.rs b/crates/ironrdp-session/src/autodetect.rs index ccc0a3105a..e1f59a5f3d 100644 --- a/crates/ironrdp-session/src/autodetect.rs +++ b/crates/ironrdp-session/src/autodetect.rs @@ -21,6 +21,24 @@ 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, @@ -29,6 +47,13 @@ struct BandwidthMeasurement { } impl AutoDetectResponder { + /// Whether a continuous bandwidth window is open, so received bytes are being counted. + pub(crate) fn is_counting(&self) -> bool { + self.bandwidth + .as_ref() + .is_some_and(|measurement| measurement.continuous) + } + /// 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) { @@ -63,11 +88,7 @@ impl AutoDetectResponder { } AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => { if let Some(measurement) = self.bandwidth.as_mut().filter(|measurement| !measurement.continuous) { - // [MS-RDPBCGR] 3.2.5.14 counts the eight-byte auto-detect - // header as well as payloadLength, but not the security header. - measurement.bytes = measurement - .bytes - .saturating_add(u32::try_from(payload.len()).unwrap_or(u32::MAX).saturating_add(8)); + measurement.bytes = measurement.bytes.saturating_add(counted_len(payload.len())); } None } @@ -84,18 +105,27 @@ impl AutoDetectResponder { let stop_bytes = if continuous { 0 } else { - u32::try_from(payload.as_ref().map_or(0, Vec::len)) - .unwrap_or(u32::MAX) - .saturating_add(8) + 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(1), + .max(UNMEASURABLE_INTERVAL_MS), measurement.bytes.saturating_add(stop_bytes), ), - _ => (1, 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, stop_bytes) + } + (None, _) => (UNMEASURABLE_INTERVAL_MS, stop_bytes), }; Some(AutoDetectResponse::BandwidthMeasureResults { sequence_number, diff --git a/crates/ironrdp-session/src/x224/mod.rs b/crates/ironrdp-session/src/x224/mod.rs index a85b55ac6b..735a2968cf 100644 --- a/crates/ironrdp-session/src/x224/mod.rs +++ b/crates/ironrdp-session/src/x224/mod.rs @@ -472,6 +472,11 @@ impl Processor { self.auto_detect.record_bytes(bytes); } + /// Whether a continuous bandwidth window is open, so session bytes are being counted. + pub(crate) fn is_counting_bandwidth(&self) -> bool { + self.auto_detect.is_counting() + } + /// 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). diff --git a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs index 00f93d0285..ef531d3475 100644 --- a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs @@ -1,10 +1,15 @@ use std::borrow::Cow; use ironrdp_core::encode_vec; +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::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; @@ -161,6 +166,10 @@ fn bandwidth_result(outputs: &[ironrdp_session::x224::ProcessorOutput]) -> (u16, 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"); @@ -266,3 +275,73 @@ fn untimed_driver_does_not_report_accumulated_bytes_as_a_real_measurement() { ); assert_eq!(bandwidth_result(&outputs), (3, 0x000b, 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 frame between Start and Stop counts without its fast-path header: a + // three-byte update header and 100 bytes of update data. The Stop adds its + // six-byte auto-detect header, as in the message-channel tests above. + assert_eq!(bandwidth_result_frame(response), (2, 0x000b, 30, 3 + 100 + 6)); +} From 3fc5c2713c8946d9c6cfce1aa0d4847e5edeec36 Mon Sep 17 00:00:00 2001 From: AKolenda Date: Mon, 28 Sep 2026 00:33:42 -0600 Subject: [PATCH 08/12] fix(client): answer auto-detect requests on the UDP tunnel The client dropped the sub-headers of every Tunnel Data PDU, so the auto-detect requests Windows sends on the UDP tunnel once it is established (MS-RDPEMT 2.2.1.1.1, MS-RDPBCGR 2.2.14) were never answered. The client now reads the tunnel with `recv_message` and hands those requests to the session, which answers them with the same code as the message channel: - The request handling moves out of the x224 processor into an `AutoDetectResponder`. The message channel and the tunnel each have one, because a continuous measurement counts the data of its own transport. - `ActiveStage::process_tunnel_auto_detect` answers the requests of one Tunnel Data PDU and counts its higher-layer data. MS-RDPBCGR 3.2.5.14 counts only the data after the tunnel PDU header, and that data follows the sub-headers, so the data of the PDU carrying a Start is counted and the data of the one carrying a Stop is not. - The client times the tunnel against its own monotonic clock and sends each set of responses back with `send_message`, in a Tunnel Data PDU with no data. A tunnel sub-header is the auto-detect structure itself: its SubHeaderLength and SubHeaderType are the structure's headerLength and headerTypeId. Requests are decoded from the whole sub-header, and responses are encoded the same way. --- Cargo.lock | 1 + crates/ironrdp-client/Cargo.toml | 2 + crates/ironrdp-client/src/rdp.rs | 155 +++++++++++++++++++-- crates/ironrdp-session/src/active_stage.rs | 103 +++++++++++++- 4 files changed, 247 insertions(+), 14 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7e3280ae94..70a77fffed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2771,6 +2771,7 @@ dependencies = [ "ironrdp-rdpeai", "ironrdp-rdpei", "ironrdp-rdpel", + "ironrdp-rdpemt", "ironrdp-rdpeudp", "ironrdp-rdpeudp-tokio", "ironrdp-rdpewa", diff --git a/crates/ironrdp-client/Cargo.toml b/crates/ironrdp-client/Cargo.toml index d3b50eb465..19a5d49735 100644 --- a/crates/ironrdp-client/Cargo.toml +++ b/crates/ironrdp-client/Cargo.toml @@ -50,6 +50,7 @@ webauthn = ["dep:ironrdp-rdpewa", "dep:ironrdp-rdpewa-native", "dvc-com-plugin"] vmconnect = ["dep:ironrdp-vmconnect"] location = ["dep:ironrdp-rdpel"] udp = [ + "dep:ironrdp-rdpemt", "dep:ironrdp-rdpeudp", "dep:ironrdp-rdpeudp-tokio", "ironrdp-rdpeudp-tokio/rustls-aws-lc-rs", @@ -82,6 +83,7 @@ ironrdp-displaycontrol = { path = "../ironrdp-displaycontrol", version = "0.8" } ironrdp-echo = { path = "../ironrdp-echo", version = "0.4" } ironrdp-egfx = { path = "../ironrdp-egfx", version = "0.3" } ironrdp-rdpei = { path = "../ironrdp-rdpei", version = "0.1" } +ironrdp-rdpemt = { path = "../ironrdp-rdpemt", version = "0.1", optional = true } ironrdp-rdpeudp = { path = "../ironrdp-rdpeudp", version = "0.1", optional = true } ironrdp-rdpeudp-tokio = { path = "../ironrdp-rdpeudp-tokio", version = "0.1", optional = true } ironrdp-tls = { path = "../ironrdp-tls", version = "0.2" } # public diff --git a/crates/ironrdp-client/src/rdp.rs b/crates/ironrdp-client/src/rdp.rs index 7ab66aed1c..56cc762b22 100644 --- a/crates/ironrdp-client/src/rdp.rs +++ b/crates/ironrdp-client/src/rdp.rs @@ -9,7 +9,7 @@ use std::io; use std::sync::Arc; #[cfg(feature = "location")] use std::sync::mpsc as std_mpsc; -#[cfg(feature = "location")] +#[cfg(any(feature = "location", feature = "udp"))] use std::time::Instant; #[cfg(feature = "clipboard")] @@ -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")] @@ -2855,6 +2856,56 @@ fn build_ordinary_rdcleanpath_request( // ── Active session ──────────────────────────────────────────────────────────── +/// Decodes the auto-detect requests among the sub-headers of a Tunnel Data PDU. +/// +/// Such a sub-header is the request structure itself: its SubHeaderLength and SubHeaderType +/// are the request's headerLength and headerTypeId ([MS-RDPEMT] 2.2.1.1.1), so the request is +/// decoded from the whole sub-header. Windows also sends request types the decoder does not +/// model; those are skipped. +/// +/// [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad +#[cfg(feature = "udp")] +fn tunnel_auto_detect_requests(sub_headers: &[ironrdp_rdpemt::TunnelSubHeader]) -> Vec { + sub_headers + .iter() + .filter(|sub_header| sub_header.sub_header_type == ironrdp_rdpemt::SubHeaderType::AutoDetectRequest) + .filter_map(|sub_header| { + let decoded = ironrdp_core::encode_vec(sub_header) + .map_err(|error| error.to_string()) + .and_then(|structure| { + ironrdp_core::decode::(&structure).map_err(|error| error.to_string()) + }); + match decoded { + Ok(request) => Some(request), + Err(error) => { + debug!(%error, data = ?sub_header.data, "Ignoring an undecodable auto-detect request on the tunnel"); + None + } + } + }) + .collect() +} + +/// Encodes an auto-detect response as the tunnel sub-header it is on the wire (see +/// [`tunnel_auto_detect_requests`]). +#[cfg(feature = "udp")] +fn tunnel_auto_detect_sub_header( + response: &ironrdp_pdu::rdp::autodetect::AutoDetectResponse, +) -> Option { + let encoded = ironrdp_core::encode_vec(response) + .map_err(|error| error.to_string()) + .and_then(|structure| { + ironrdp_core::decode::(&structure).map_err(|error| error.to_string()) + }); + match encoded { + Ok(sub_header) => Some(sub_header), + Err(error) => { + debug!(%error, ?response, "Could not encode an auto-detect response for the tunnel"); + None + } + } +} + enum RdpControlFlow { ReconnectWithNewSize { width: u16, @@ -3103,6 +3154,9 @@ async fn active_session( let mut graceful_shutdown_sent = false; let mut post_logon_redraw_requested = false; let mut pending_udp_payload: Option> = None; + // Auto-detect on the tunnel is timed against its own monotonic clock. + #[cfg(feature = "udp")] + let tunnel_clock = Instant::now(); let mut initial_outputs = if *graceful_close_receiver.borrow_and_update() { graceful_shutdown_sent = true; Some(active_stage.graceful_shutdown()?) @@ -3247,22 +3301,25 @@ 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) => transport + .recv_message() + .await + .map(|message| (tunnel_auto_detect_requests(&message.sub_headers), message.data)), (Some(_), false) | (None, _) => { - core::future::pending::>>().await + core::future::pending::, Vec)>>().await } } } #[cfg(not(feature = "udp"))] { - core::future::pending::>>().await + core::future::pending::, Vec)>>().await } } => { - match udp_payload { + match udp_message { None => { if active_stage.reliable_udp_dvc_tunnel_in_use() { return Ok(RdpControlFlow::TransportFailure( @@ -3277,12 +3334,45 @@ async fn active_session( 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((auto_detect_requests, payload)) => { + #[cfg(feature = "udp")] + { + let received_at = ironrdp_core::MonotonicInstant::from_millis( + u64::try_from(tunnel_clock.elapsed().as_millis()).unwrap_or(u64::MAX), + ); + let responses = active_stage.process_tunnel_auto_detect( + auto_detect_requests, + payload.len(), + Some(received_at), + ); + let sub_headers: Vec<_> = responses.iter().filter_map(tunnel_auto_detect_sub_header).collect(); + if let (false, Some(transport)) = (sub_headers.is_empty(), udp_tunnel.transport.as_ref()) { + let reply = ironrdp_rdpeudp_tokio::TunnelMessage { + sub_headers, + data: Vec::new(), + }; + let Some(result) = + cancelable_operation(transport.send_message(reply), close_receiver).await + else { + return Ok(RdpControlFlow::TerminatedGracefully( + GracefulDisconnectReason::UserInitiated, + )); + }; + if let Err(error) = result { + return Ok(RdpControlFlow::TransportFailure(ironrdp_session::custom_err!( + "answer reliable UDP tunnel auto-detect", + error + ))); + } + } + } + #[cfg(not(feature = "udp"))] + let _ = auto_detect_requests; + + 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( @@ -4410,12 +4500,51 @@ fn process_clipboard_message( mod tests { use super::*; + /// Windows frames an RTT Measure Request on the tunnel as `06 00 01 00`: the + /// sub-header's own two bytes are the request's header. + #[cfg(feature = "udp")] + #[test] + fn tunnel_sub_headers_decode_as_auto_detect_requests() { + use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader}; + + 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 = tunnel_auto_detect_requests(&[rtt, unmodeled, response]); + 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. + #[cfg(feature = "udp")] + #[test] + fn auto_detect_responses_encode_as_tunnel_sub_headers() { + let response = ironrdp_pdu::rdp::autodetect::AutoDetectResponse::RttResponse { sequence_number: 7 }; + let sub_header = tunnel_auto_detect_sub_header(&response).expect("sub-header"); + let pdu = encode_vec(&ironrdp_rdpemt::TunnelData { + sub_headers: vec![sub_header], + higher_layer_data: Vec::new(), + }) + .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]); + } + #[cfg(feature = "rdpdr")] use core::any::TypeId; #[cfg(feature = "rdpdr")] use core::sync::atomic::AtomicUsize; - #[cfg(feature = "rdpdr")] + #[cfg(any(feature = "rdpdr", feature = "udp"))] use ironrdp_core::encode_vec; use ironrdp_pdu::input::fast_path::KeyboardFlags; #[cfg(feature = "rdpdr")] diff --git a/crates/ironrdp-session/src/active_stage.rs b/crates/ironrdp-session/src/active_stage.rs index 6a5660c662..1981e6feaf 100644 --- a/crates/ironrdp-session/src/active_stage.rs +++ b/crates/ironrdp-session/src/active_stage.rs @@ -11,7 +11,7 @@ use ironrdp_pdu::fast_path::FastPathHeader; 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; @@ -28,6 +28,7 @@ use ironrdp_rdpei::RdpeiClient; use ironrdp_svc::{StaticChannelSet, SvcMessage, SvcProcessor, SvcProcessorMessages}; use tracing::{debug, warn}; +use crate::autodetect::AutoDetectResponder; use crate::fast_path::UpdateKind; use crate::image::DecodedImage; use crate::{SessionError, SessionErrorExt as _, SessionResult, fast_path, x224}; @@ -50,6 +51,8 @@ pub struct ActiveStage { window_support_level: Option, graphics_output_needs_full_refresh: bool, damage_regions: Vec, + /// Answers the auto-detect requests carried on the reliable UDP tunnel. + tunnel_auto_detect: AutoDetectResponder, } /// Builder for [`ActiveStage`]. @@ -108,6 +111,7 @@ impl ActiveStageBuilder { window_support_level: None, graphics_output_needs_full_refresh: false, damage_regions: Vec::new(), + tunnel_auto_detect: AutoDetectResponder::default(), } } } @@ -592,6 +596,33 @@ 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 keeps its own measurement, apart from the message channel's, because a + /// measurement on the tunnel counts only the data that follows the tunnel PDU header + /// ([MS-RDPBCGR] 3.2.5.14). That data follows the sub-headers on the wire, so the data of the + /// PDU that carries a Start is counted and the data of the one that carries a Stop is not. + /// `received_at` is the time the PDU arrived, from one monotonic clock for the whole tunnel. + /// + /// [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 { + let responses = requests + .into_iter() + .filter_map(|request| self.tunnel_auto_detect.respond(request, received_at)) + .collect(); + self.tunnel_auto_detect.record_bytes(data_len); + responses + } + /// 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 @@ -1109,6 +1140,76 @@ mod tests { use ironrdp_pdu::pointer::{ColorPointerAttribute, Point16, PointerAttribute, PointerUpdateData}; use ironrdp_rdpei::pdu::{PenEventPdu, RdpInputProtocolVersion, RdpeiPdu, ScReadyPdu, TouchEventPdu}; + fn tunnel_test_stage() -> ActiveStage { + ActiveStageBuilder { + static_channels: StaticChannelSet::new(), + user_channel_id: 1001, + io_channel_id: 1003, + message_channel_id: None, + share_id: 1, + compression_type: None, + enable_server_pointer: true, + pointer_software_rendering: false, + } + .build() + } + + #[test] + fn tunnel_auto_detect_answers_an_rtt_request() { + let mut stage = tunnel_test_stage(); + let responses = stage.process_tunnel_auto_detect( + vec![AutoDetectRequest::rtt_continuous(7)], + 10, + Some(MonotonicInstant::from_millis(1_000)), + ); + assert_eq!(responses, [AutoDetectResponse::RttResponse { sequence_number: 7 }]); + } + + /// The data of the PDU that carries the Start follows it on the wire and is counted; the data + /// of the one that carries the Stop is not. + #[test] + fn tunnel_auto_detect_counts_the_data_between_start_and_stop() { + let mut stage = tunnel_test_stage(); + let at = |millis| Some(MonotonicInstant::from_millis(millis)); + + assert!( + stage + .process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 7, at(1_000)) + .is_empty() + ); + assert!(stage.process_tunnel_auto_detect(Vec::new(), 100, at(1_010)).is_empty()); + let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(3)], 50, 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, 7 + 100)); + } + + /// The tunnel's measurement is separate from the message channel's, so data that arrives over + /// TCP is not counted in it. + #[test] + fn tunnel_auto_detect_keeps_its_own_window() { + let mut stage = tunnel_test_stage(); + let at = |millis| Some(MonotonicInstant::from_millis(millis)); + + stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 0, at(1_000)); + stage.x224_processor.record_bandwidth_bytes(500); + let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(2)], 0, at(1_020)); + assert!(matches!( + responses.as_slice(), + [AutoDetectResponse::BandwidthMeasureResults { byte_count: 0, .. }] + )); + } + #[test] fn full_redraw_prefers_suppress_output_toggle_when_supported() { let stage = ActiveStageBuilder { From aca65e9afe8530e0ad22d4bf5180e2236b0144a1 Mon Sep 17 00:00:00 2001 From: AKolenda Date: Mon, 28 Sep 2026 13:24:52 -0600 Subject: [PATCH 09/12] test(client): run the tunnel auto-detect tests in CI `ironrdp-session` and `ironrdp-client` set `test = false`, so their inline tests never ran under `cargo test --workspace`. Moves them to the testsuites and exposes the two sub-header helpers as `#[doc(hidden)]`, as the client already does for `connect_preferring_direct`. The separate-window test now opens both windows and feeds data to each. --- crates/ironrdp-client/src/rdp.rs | 47 ++---------- crates/ironrdp-session/src/active_stage.rs | 70 ------------------ .../tests/session/autodetect.rs | 71 +++++++++++++++++++ .../tests/client/mod.rs | 1 + .../tests/client/tunnel_auto_detect.rs | 45 ++++++++++++ 5 files changed, 122 insertions(+), 112 deletions(-) create mode 100644 crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs diff --git a/crates/ironrdp-client/src/rdp.rs b/crates/ironrdp-client/src/rdp.rs index 56cc762b22..1d8dee4525 100644 --- a/crates/ironrdp-client/src/rdp.rs +++ b/crates/ironrdp-client/src/rdp.rs @@ -2865,7 +2865,8 @@ fn build_ordinary_rdcleanpath_request( /// /// [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad #[cfg(feature = "udp")] -fn tunnel_auto_detect_requests(sub_headers: &[ironrdp_rdpemt::TunnelSubHeader]) -> Vec { +#[doc(hidden)] +pub fn tunnel_auto_detect_requests(sub_headers: &[ironrdp_rdpemt::TunnelSubHeader]) -> Vec { sub_headers .iter() .filter(|sub_header| sub_header.sub_header_type == ironrdp_rdpemt::SubHeaderType::AutoDetectRequest) @@ -2889,7 +2890,8 @@ fn tunnel_auto_detect_requests(sub_headers: &[ironrdp_rdpemt::TunnelSubHeader]) /// Encodes an auto-detect response as the tunnel sub-header it is on the wire (see /// [`tunnel_auto_detect_requests`]). #[cfg(feature = "udp")] -fn tunnel_auto_detect_sub_header( +#[doc(hidden)] +pub fn tunnel_auto_detect_sub_header( response: &ironrdp_pdu::rdp::autodetect::AutoDetectResponse, ) -> Option { let encoded = ironrdp_core::encode_vec(response) @@ -4500,51 +4502,12 @@ fn process_clipboard_message( mod tests { use super::*; - /// Windows frames an RTT Measure Request on the tunnel as `06 00 01 00`: the - /// sub-header's own two bytes are the request's header. - #[cfg(feature = "udp")] - #[test] - fn tunnel_sub_headers_decode_as_auto_detect_requests() { - use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader}; - - 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 = tunnel_auto_detect_requests(&[rtt, unmodeled, response]); - 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. - #[cfg(feature = "udp")] - #[test] - fn auto_detect_responses_encode_as_tunnel_sub_headers() { - let response = ironrdp_pdu::rdp::autodetect::AutoDetectResponse::RttResponse { sequence_number: 7 }; - let sub_header = tunnel_auto_detect_sub_header(&response).expect("sub-header"); - let pdu = encode_vec(&ironrdp_rdpemt::TunnelData { - sub_headers: vec![sub_header], - higher_layer_data: Vec::new(), - }) - .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]); - } - #[cfg(feature = "rdpdr")] use core::any::TypeId; #[cfg(feature = "rdpdr")] use core::sync::atomic::AtomicUsize; - #[cfg(any(feature = "rdpdr", feature = "udp"))] + #[cfg(feature = "rdpdr")] use ironrdp_core::encode_vec; use ironrdp_pdu::input::fast_path::KeyboardFlags; #[cfg(feature = "rdpdr")] diff --git a/crates/ironrdp-session/src/active_stage.rs b/crates/ironrdp-session/src/active_stage.rs index 1981e6feaf..060b4fe96b 100644 --- a/crates/ironrdp-session/src/active_stage.rs +++ b/crates/ironrdp-session/src/active_stage.rs @@ -1140,76 +1140,6 @@ mod tests { use ironrdp_pdu::pointer::{ColorPointerAttribute, Point16, PointerAttribute, PointerUpdateData}; use ironrdp_rdpei::pdu::{PenEventPdu, RdpInputProtocolVersion, RdpeiPdu, ScReadyPdu, TouchEventPdu}; - fn tunnel_test_stage() -> ActiveStage { - ActiveStageBuilder { - static_channels: StaticChannelSet::new(), - user_channel_id: 1001, - io_channel_id: 1003, - message_channel_id: None, - share_id: 1, - compression_type: None, - enable_server_pointer: true, - pointer_software_rendering: false, - } - .build() - } - - #[test] - fn tunnel_auto_detect_answers_an_rtt_request() { - let mut stage = tunnel_test_stage(); - let responses = stage.process_tunnel_auto_detect( - vec![AutoDetectRequest::rtt_continuous(7)], - 10, - Some(MonotonicInstant::from_millis(1_000)), - ); - assert_eq!(responses, [AutoDetectResponse::RttResponse { sequence_number: 7 }]); - } - - /// The data of the PDU that carries the Start follows it on the wire and is counted; the data - /// of the one that carries the Stop is not. - #[test] - fn tunnel_auto_detect_counts_the_data_between_start_and_stop() { - let mut stage = tunnel_test_stage(); - let at = |millis| Some(MonotonicInstant::from_millis(millis)); - - assert!( - stage - .process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 7, at(1_000)) - .is_empty() - ); - assert!(stage.process_tunnel_auto_detect(Vec::new(), 100, at(1_010)).is_empty()); - let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(3)], 50, 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, 7 + 100)); - } - - /// The tunnel's measurement is separate from the message channel's, so data that arrives over - /// TCP is not counted in it. - #[test] - fn tunnel_auto_detect_keeps_its_own_window() { - let mut stage = tunnel_test_stage(); - let at = |millis| Some(MonotonicInstant::from_millis(millis)); - - stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 0, at(1_000)); - stage.x224_processor.record_bandwidth_bytes(500); - let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(2)], 0, at(1_020)); - assert!(matches!( - responses.as_slice(), - [AutoDetectResponse::BandwidthMeasureResults { byte_count: 0, .. }] - )); - } - #[test] fn full_redraw_prefers_suppress_output_toggle_when_supported() { let stage = ActiveStageBuilder { diff --git a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs index ef531d3475..22084265a4 100644 --- a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs @@ -345,3 +345,74 @@ fn continuous_measurement_counts_fast_path_data_only_inside_the_window() { // six-byte auto-detect header, as in the message-channel tests above. assert_eq!(bandwidth_result_frame(response), (2, 0x000b, 30, 3 + 100 + 6)); } + +fn at(millis: u64) -> Option { + Some(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, at(1_000)); + assert_eq!(responses, [AutoDetectResponse::RttResponse { sequence_number: 7 }]); +} + +/// The data of the PDU that carries the Start follows it on the wire and is counted; the data +/// of the one that carries the Stop is not. +#[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, at(1_000)) + .is_empty() + ); + assert!(stage.process_tunnel_auto_detect(Vec::new(), 100, at(1_010)).is_empty()); + let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(3)], 50, 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, 7 + 100)); +} + +/// The tunnel and the main connection each keep their own measurement, so data received on one +/// is not counted in the other's, even while both windows are open. +#[test] +fn tunnel_and_main_connection_measure_separately() { + 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); + stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 0, at(1_000)); + + process_stage_frame(&mut stage, &mut image, Action::FastPath, &fast_path_frame(100), 1_010); + stage.process_tunnel_auto_detect(Vec::new(), 500, at(1_010)); + + let tunnel = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(2)], 0, at(1_020)); + assert!( + matches!( + tunnel.as_slice(), + [AutoDetectResponse::BandwidthMeasureResults { byte_count: 500, .. }] + ), + "{tunnel:?}" + ); + + let stop = encode_server_autodetect(AutoDetectRequest::bw_stop_continuous(2)); + let outputs = process_stage_frame(&mut stage, &mut image, Action::X224, &stop, 1_020); + let [ActiveStageOutput::ResponseFrame(response)] = outputs.as_slice() else { + panic!("expected exactly one bandwidth response, got {outputs:?}"); + }; + // The fast-path data and the Stop's own six bytes, but none of the tunnel's 500. + assert_eq!(bandwidth_result_frame(response), (2, 0x000b, 20, 3 + 100 + 6)); +} diff --git a/crates/ironrdp-testsuite-extra/tests/client/mod.rs b/crates/ironrdp-testsuite-extra/tests/client/mod.rs index d4172e224b..b147cf0f12 100644 --- a/crates/ironrdp-testsuite-extra/tests/client/mod.rs +++ b/crates/ironrdp-testsuite-extra/tests/client/mod.rs @@ -2,3 +2,4 @@ mod config; mod input; mod output_channel; mod rail; +mod tunnel_auto_detect; diff --git a/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs b/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs new file mode 100644 index 0000000000..35a01c0300 --- /dev/null +++ b/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs @@ -0,0 +1,45 @@ +//! 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_client::rdp::{tunnel_auto_detect_requests, tunnel_auto_detect_sub_header}; +use ironrdp_core::encode_vec; +use ironrdp_pdu::rdp::autodetect::{AutoDetectRequest, AutoDetectResponse}; +use ironrdp_rdpemt::{SubHeaderType, TunnelData, TunnelSubHeader}; + +/// 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 = tunnel_auto_detect_requests(&[rtt, unmodeled, response]); + 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 response = AutoDetectResponse::RttResponse { sequence_number: 7 }; + let sub_header = tunnel_auto_detect_sub_header(&response).expect("sub-header"); + let pdu = encode_vec(&TunnelData { + sub_headers: vec![sub_header], + higher_layer_data: Vec::new(), + }) + .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]); +} From 4ff2b8e3756b761bcba8ec08519d1f5ab2869e18 Mon Sep 17 00:00:00 2001 From: AKolenda Date: Fri, 2 Oct 2026 00:13:53 -0600 Subject: [PATCH 10/12] fix(client): handle tunnel auto-detect failures and byte accounting --- crates/ironrdp-client/Cargo.toml | 3 + crates/ironrdp-client/src/lib.rs | 10 ++ crates/ironrdp-client/src/rdp.rs | 88 ++++---------- crates/ironrdp-client/src/udp.rs | 63 +++++++++++ crates/ironrdp-session/src/active_stage.rs | 16 +-- .../tests/session/autodetect.rs | 10 +- crates/ironrdp-testsuite-extra/Cargo.toml | 2 +- .../tests/client/tunnel_auto_detect.rs | 107 +++++++++++++++++- 8 files changed, 218 insertions(+), 81 deletions(-) create mode 100644 crates/ironrdp-client/src/udp.rs diff --git a/crates/ironrdp-client/Cargo.toml b/crates/ironrdp-client/Cargo.toml index 19a5d49735..d2c9a182ee 100644 --- a/crates/ironrdp-client/Cargo.toml +++ b/crates/ironrdp-client/Cargo.toml @@ -18,6 +18,9 @@ test = false [features] default = [] +# Internal integration-test access; not part of the supported public API. +__test = [] + rustls = [ "ironrdp-tls/rustls", "tokio-tungstenite/rustls-tls-native-roots", diff --git a/crates/ironrdp-client/src/lib.rs b/crates/ironrdp-client/src/lib.rs index fe4b112898..0459c35972 100644 --- a/crates/ironrdp-client/src/lib.rs +++ b/crates/ironrdp-client/src/lib.rs @@ -16,3 +16,13 @@ pub mod rdp; mod clipboard; mod ws; + +#[cfg(all(feature = "udp", feature = "__test"))] +#[doc(hidden)] +pub mod udp; +#[cfg(all(feature = "udp", not(feature = "__test")))] +#[expect( + unreachable_pub, + reason = "the __test feature exposes this module to the shared integration tests" +)] +mod udp; diff --git a/crates/ironrdp-client/src/rdp.rs b/crates/ironrdp-client/src/rdp.rs index 1d8dee4525..676f085084 100644 --- a/crates/ironrdp-client/src/rdp.rs +++ b/crates/ironrdp-client/src/rdp.rs @@ -84,6 +84,8 @@ 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, tunnel_auto_detect_requests, tunnel_auto_detect_sub_header}; use ironrdp_rail::pdu::{ExecutePdu, ExecuteResultPdu}; // ── Public event types ──────────────────────────────────────────────────────── @@ -2856,58 +2858,6 @@ fn build_ordinary_rdcleanpath_request( // ── Active session ──────────────────────────────────────────────────────────── -/// Decodes the auto-detect requests among the sub-headers of a Tunnel Data PDU. -/// -/// Such a sub-header is the request structure itself: its SubHeaderLength and SubHeaderType -/// are the request's headerLength and headerTypeId ([MS-RDPEMT] 2.2.1.1.1), so the request is -/// decoded from the whole sub-header. Windows also sends request types the decoder does not -/// model; those are skipped. -/// -/// [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad -#[cfg(feature = "udp")] -#[doc(hidden)] -pub fn tunnel_auto_detect_requests(sub_headers: &[ironrdp_rdpemt::TunnelSubHeader]) -> Vec { - sub_headers - .iter() - .filter(|sub_header| sub_header.sub_header_type == ironrdp_rdpemt::SubHeaderType::AutoDetectRequest) - .filter_map(|sub_header| { - let decoded = ironrdp_core::encode_vec(sub_header) - .map_err(|error| error.to_string()) - .and_then(|structure| { - ironrdp_core::decode::(&structure).map_err(|error| error.to_string()) - }); - match decoded { - Ok(request) => Some(request), - Err(error) => { - debug!(%error, data = ?sub_header.data, "Ignoring an undecodable auto-detect request on the tunnel"); - None - } - } - }) - .collect() -} - -/// Encodes an auto-detect response as the tunnel sub-header it is on the wire (see -/// [`tunnel_auto_detect_requests`]). -#[cfg(feature = "udp")] -#[doc(hidden)] -pub fn tunnel_auto_detect_sub_header( - response: &ironrdp_pdu::rdp::autodetect::AutoDetectResponse, -) -> Option { - let encoded = ironrdp_core::encode_vec(response) - .map_err(|error| error.to_string()) - .and_then(|structure| { - ironrdp_core::decode::(&structure).map_err(|error| error.to_string()) - }); - match encoded { - Ok(sub_header) => Some(sub_header), - Err(error) => { - debug!(%error, ?response, "Could not encode an auto-detect response for the tunnel"); - None - } - } -} - enum RdpControlFlow { ReconnectWithNewSize { width: u16, @@ -3323,17 +3273,16 @@ async fn active_session( } => { 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((auto_detect_requests, payload)) => { @@ -3345,10 +3294,10 @@ async fn active_session( let responses = active_stage.process_tunnel_auto_detect( auto_detect_requests, payload.len(), - Some(received_at), + received_at, ); let sub_headers: Vec<_> = responses.iter().filter_map(tunnel_auto_detect_sub_header).collect(); - if let (false, Some(transport)) = (sub_headers.is_empty(), udp_tunnel.transport.as_ref()) { + if !sub_headers.is_empty() && let Some(transport) = udp_tunnel.transport.as_ref() { let reply = ironrdp_rdpeudp_tokio::TunnelMessage { sub_headers, data: Vec::new(), @@ -3361,10 +3310,17 @@ async fn active_session( )); }; if let Err(error) = result { - return Ok(RdpControlFlow::TransportFailure(ironrdp_session::custom_err!( - "answer reliable UDP tunnel auto-detect", - error - ))); + 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)); + } + // This PDU arrived before any channel migrated. Once the + // tunnel fails, it cannot retain data for a future Soft-Sync. + pending_udp_payload = None; + continue; } } } diff --git a/crates/ironrdp-client/src/udp.rs b/crates/ironrdp-client/src/udp.rs new file mode 100644 index 0000000000..edef32363a --- /dev/null +++ b/crates/ironrdp-client/src/udp.rs @@ -0,0 +1,63 @@ +//! Internal client handling for a reliable UDP tunnel. + +use ironrdp_core::{Decode, Encode}; +use ironrdp_pdu::rdp::autodetect::{AutoDetectRequest, AutoDetectResponse}; +use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader}; +use ironrdp_rdpeudp_tokio::UdpTransport; +use ironrdp_session::{ActiveStage, SessionError, SessionResult}; +use tracing::{debug, warn}; + +/// Decodes the auto-detect requests among the sub-headers of a Tunnel Data PDU. +/// +/// Each sub-header is the request structure itself: SubHeaderLength and SubHeaderType +/// are headerLength and headerTypeId ([MS-RDPEMT] 2.2.1.1.1). Unknown requests are skipped. +/// RTT requests are accepted for Windows interoperability in addition to the bandwidth +/// requests listed by MS-RDPEMT; RTT encapsulation on a tunnel is not specified there. +/// +/// [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad +pub fn tunnel_auto_detect_requests(sub_headers: &[TunnelSubHeader]) -> Vec { + 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() +} + +/// Encodes a response as its wire-compatible tunnel sub-header. +pub fn tunnel_auto_detect_sub_header(response: &AutoDetectResponse) -> Option { + 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 + } + } +} + +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()) +} + +/// 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 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-session/src/active_stage.rs b/crates/ironrdp-session/src/active_stage.rs index 060b4fe96b..2cdea70472 100644 --- a/crates/ironrdp-session/src/active_stage.rs +++ b/crates/ironrdp-session/src/active_stage.rs @@ -603,8 +603,9 @@ impl ActiveStage { /// /// The tunnel keeps its own measurement, apart from the message channel's, because a /// measurement on the tunnel counts only the data that follows the tunnel PDU header - /// ([MS-RDPBCGR] 3.2.5.14). That data follows the sub-headers on the wire, so the data of the - /// PDU that carries a Start is counted and the data of the one that carries a Stop is not. + /// ([MS-RDPBCGR] 3.2.5.14). 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. Sub-header bytes themselves never count on the tunnel. /// `received_at` is the time the PDU arrived, from one monotonic clock for the whole tunnel. /// /// [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad @@ -613,14 +614,13 @@ impl ActiveStage { &mut self, requests: Vec, data_len: usize, - received_at: Option, + received_at: MonotonicInstant, ) -> Vec { - let responses = requests - .into_iter() - .filter_map(|request| self.tunnel_auto_detect.respond(request, received_at)) - .collect(); self.tunnel_auto_detect.record_bytes(data_len); - responses + requests + .into_iter() + .filter_map(|request| self.tunnel_auto_detect.respond(request, Some(received_at))) + .collect() } /// Returns the Soft-Sync tunnel selected for client messages on `channel_id`. diff --git a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs index 22084265a4..2f35ed2f7c 100644 --- a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs @@ -346,8 +346,8 @@ fn continuous_measurement_counts_fast_path_data_only_inside_the_window() { assert_eq!(bandwidth_result_frame(response), (2, 0x000b, 30, 3 + 100 + 6)); } -fn at(millis: u64) -> Option { - Some(ironrdp_core::MonotonicInstant::from_millis(millis)) +fn at(millis: u64) -> ironrdp_core::MonotonicInstant { + ironrdp_core::MonotonicInstant::from_millis(millis) } #[test] @@ -357,8 +357,8 @@ fn tunnel_auto_detect_answers_an_rtt_request() { assert_eq!(responses, [AutoDetectResponse::RttResponse { sequence_number: 7 }]); } -/// The data of the PDU that carries the Start follows it on the wire and is counted; the data -/// of the one that carries the Stop is not. +/// 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(); @@ -382,7 +382,7 @@ fn tunnel_auto_detect_counts_the_data_between_start_and_stop() { else { panic!("expected one bandwidth result, got {responses:?}"); }; - assert_eq!((*sequence_number, *time_delta_ms, *byte_count), (3, 40, 7 + 100)); + assert_eq!((*sequence_number, *time_delta_ms, *byte_count), (3, 40, 100 + 50)); } /// The tunnel and the main connection each keep their own measurement, so data received on one diff --git a/crates/ironrdp-testsuite-extra/Cargo.toml b/crates/ironrdp-testsuite-extra/Cargo.toml index 3f5a348296..bc4719bae2 100644 --- a/crates/ironrdp-testsuite-extra/Cargo.toml +++ b/crates/ironrdp-testsuite-extra/Cargo.toml @@ -30,7 +30,7 @@ ironrdp-async.path = "../ironrdp-async" ironrdp-agent.path = "../ironrdp-agent" ironrdp-daemon.path = "../ironrdp-daemon" ironrdp-bulk.path = "../ironrdp-bulk" -ironrdp-client = { path = "../ironrdp-client", features = ["sound", "udp"] } +ironrdp-client = { path = "../ironrdp-client", features = ["sound", "udp", "__test"] } ironrdp-core.path = "../ironrdp-core" ironrdp-dvc.path = "../ironrdp-dvc" ironrdp-dvc-pipe-proxy.path = "../ironrdp-dvc-pipe-proxy" diff --git a/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs b/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs index 35a01c0300..43098761e7 100644 --- a/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs +++ b/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs @@ -4,7 +4,7 @@ //! //! [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad -use ironrdp_client::rdp::{tunnel_auto_detect_requests, tunnel_auto_detect_sub_header}; +use ironrdp_client::udp::{tunnel_auto_detect_requests, tunnel_auto_detect_sub_header}; use ironrdp_core::encode_vec; use ironrdp_pdu::rdp::autodetect::{AutoDetectRequest, AutoDetectResponse}; use ironrdp_rdpemt::{SubHeaderType, TunnelData, TunnelSubHeader}; @@ -43,3 +43,108 @@ fn auto_detect_responses_encode_as_tunnel_sub_headers() { // 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 bandwidth_sub_headers_preserve_the_wire_layout() { + let start = TunnelSubHeader { + sub_header_type: SubHeaderType::AutoDetectRequest, + data: vec![9, 0, 0x14, 0], + }; + assert_eq!( + tunnel_auto_detect_requests(&[start]), + [AutoDetectRequest::bw_start_continuous(9)] + ); + let response = AutoDetectResponse::BandwidthMeasureResults { + sequence_number: 9, + response_type: 0x000b, + time_delta_ms: 40, + byte_count: 150, + }; + let sub_header = tunnel_auto_detect_sub_header(&response).expect("bandwidth sub-header"); + assert_eq!( + encode_vec(&sub_header).unwrap(), + [14, 1, 9, 0, 11, 0, 40, 0, 0, 0, 150, 0, 0, 0] + ); +} + +#[test] +fn tunnel_failure_before_migration_withdraws_soft_sync_without_ending_the_session() { + let mut stage = make_stage(false); + let mut transport = None; + ironrdp_client::udp::disable_failed_tunnel( + &mut stage, + &mut transport, + ironrdp_session::general_err!("auto-detect reply send failed"), + ) + .expect("unused tunnel must fall back to TCP"); + assert!(transport.is_none()); + assert!(!stage.reliable_udp_dvc_tunnel_in_use()); + + // The failed tunnel must no longer be offered to a later Soft-Sync request. + use ironrdp_svc::SvcProcessor as _; + let drdynvc = stage.get_svc_processor_mut::().unwrap(); + assert!(drdynvc.process(&soft_sync_request()).is_err()); +} + +#[test] +fn tunnel_failure_after_migration_requires_reconnection() { + let mut stage = make_stage(true); + let mut transport = None; + let error = ironrdp_client::udp::disable_failed_tunnel( + &mut stage, + &mut transport, + ironrdp_session::general_err!("auto-detect reply send failed"), + ) + .expect_err("migrated channels cannot silently resume on TCP"); + assert!(error.to_string().contains("auto-detect reply send failed")); + assert!(stage.reliable_udp_dvc_tunnel_in_use()); +} + +fn make_stage(migrated: bool) -> ironrdp_session::ActiveStage { + use ironrdp_dvc::pdu::SoftSyncTunnelType; + use ironrdp_svc::SvcProcessor as _; + + let mut drdynvc = ironrdp_dvc::DrdynvcClient::new(); + drdynvc.attach_established_dynamic_channel(7, TestChannel).unwrap(); + drdynvc.enable_soft_sync_tunnel(SoftSyncTunnelType::RELIABLE_UDP); + if migrated { + drdynvc.process(&soft_sync_request()).unwrap(); + } + let mut static_channels = ironrdp_svc::StaticChannelSet::new(); + static_channels.insert(drdynvc); + ironrdp_session::ActiveStageBuilder { + static_channels, + user_channel_id: 1001, + io_channel_id: 1003, + message_channel_id: Some(1005), + share_id: 1, + compression_type: None, + enable_server_pointer: false, + pointer_software_rendering: false, + } + .build() +} + +fn soft_sync_request() -> Vec { + use ironrdp_dvc::pdu::{DrdynvcServerPdu, SoftSyncChannelList, SoftSyncRequestPdu, SoftSyncTunnelType}; + encode_vec(&DrdynvcServerPdu::SoftSyncRequest(SoftSyncRequestPdu::new(vec![ + SoftSyncChannelList::new(SoftSyncTunnelType::RELIABLE_UDP, vec![7]), + ]))) + .unwrap() +} + +struct TestChannel; +ironrdp_core::impl_as_any!(TestChannel); + +impl ironrdp_dvc::DvcProcessor for TestChannel { + fn channel_name(&self) -> &str { + "test-channel" + } + fn start(&mut self, _channel_id: u32) -> ironrdp_pdu::PduResult> { + Ok(Vec::new()) + } + fn process(&mut self, _channel_id: u32, _payload: &[u8]) -> ironrdp_pdu::PduResult> { + Ok(Vec::new()) + } +} +impl ironrdp_dvc::DvcClientProcessor for TestChannel {} From a97dde1a1f1d734b533d43ffdf56ba9aef179473 Mon Sep 17 00:00:00 2001 From: AKolenda Date: Fri, 2 Oct 2026 00:11:40 -0600 Subject: [PATCH 11/12] fix(session): count complete bandwidth frames and reject untimed estimates --- crates/ironrdp-session/src/active_stage.rs | 16 +-- crates/ironrdp-session/src/autodetect.rs | 14 +- crates/ironrdp-session/src/x224/mod.rs | 29 ++-- .../tests/session/autodetect.rs | 130 +++++++++++++++++- 4 files changed, 150 insertions(+), 39 deletions(-) diff --git a/crates/ironrdp-session/src/active_stage.rs b/crates/ironrdp-session/src/active_stage.rs index 2cdea70472..1f028d8619 100644 --- a/crates/ironrdp-session/src/active_stage.rs +++ b/crates/ironrdp-session/src/active_stage.rs @@ -1,13 +1,12 @@ use std::sync::Arc; use ironrdp_bulk::{BulkCompressor, CompressionType as BulkCompressionType}; -use ironrdp_core::{Decode as _, MonotonicInstant, 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}; use ironrdp_egfx::client::GraphicsPipelineClient; use ironrdp_graphics::pointer::DecodedPointer; -use ironrdp_pdu::fast_path::FastPathHeader; use ironrdp_pdu::gcc::{ChannelName, Monitor}; use ironrdp_pdu::geometry::{ExclusiveRectangle, InclusiveRectangle, Rectangle as _}; use ironrdp_pdu::input::fast_path::{FastPathInput, FastPathInputEvent}; @@ -216,14 +215,9 @@ impl ActiveStage { self.damage_regions.clear(); let (mut stage_outputs, processor_updates) = match action { Action::FastPath => { - // A continuous bandwidth measurement counts what follows the fast-path header. - // The header is decoded again by `process`, so only pay for it here while a - // measurement is running. - if self.x224_processor.is_counting_bandwidth() { - let mut header = ReadCursor::new(frame); - FastPathHeader::decode(&mut header).map_err(SessionError::decode)?; - self.x224_processor.record_bandwidth_bytes(header.len()); - } + // TLS-protected fast-path frames have no RDP Security Header, so the + // continuous bandwidth count includes the entire frame. + self.x224_processor.record_bandwidth_bytes(frame.len()); let mut output = WriteBuf::new(); let processor_updates = self.fast_path_processor @@ -1127,7 +1121,7 @@ mod tests { use core::any::TypeId; use super::*; - use ironrdp_core::encode_vec; + use ironrdp_core::{Decode as _, encode_vec}; use ironrdp_displaycontrol::pdu::{DisplayControlCapabilities, DisplayControlPdu}; use ironrdp_dvc::pdu::{ ClosePdu, CreateRequestPdu, DataPdu, DrdynvcDataPdu, DrdynvcServerPdu, SoftSyncChannelList, SoftSyncRequestPdu, diff --git a/crates/ironrdp-session/src/autodetect.rs b/crates/ironrdp-session/src/autodetect.rs index e1f59a5f3d..8c43f8df31 100644 --- a/crates/ironrdp-session/src/autodetect.rs +++ b/crates/ironrdp-session/src/autodetect.rs @@ -47,13 +47,6 @@ struct BandwidthMeasurement { } impl AutoDetectResponder { - /// Whether a continuous bandwidth window is open, so received bytes are being counted. - pub(crate) fn is_counting(&self) -> bool { - self.bandwidth - .as_ref() - .is_some_and(|measurement| measurement.continuous) - } - /// 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) { @@ -123,9 +116,9 @@ impl AutoDetectResponder { "Bandwidth Measure Stop arrived with no arrival time although its window was open; \ dropping the accumulated count" ); - (UNMEASURABLE_INTERVAL_MS, stop_bytes) + (UNMEASURABLE_INTERVAL_MS, 0) } - (None, _) => (UNMEASURABLE_INTERVAL_MS, stop_bytes), + (None, _) => (UNMEASURABLE_INTERVAL_MS, 0), }; Some(AutoDetectResponse::BandwidthMeasureResults { sequence_number, @@ -139,6 +132,9 @@ impl AutoDetectResponder { }) } request @ AutoDetectRequest::NetworkCharacteristicsResult { .. } => { + // The TCP message-channel processor surfaces this request itself. Keep this + // arm for the UDP tunnel responder introduced in PR #2009, which handles + // auto-detect requests without passing through that processor. debug!(?request, "Received network characteristics from server"); None } diff --git a/crates/ironrdp-session/src/x224/mod.rs b/crates/ironrdp-session/src/x224/mod.rs index 735a2968cf..7e64c7ebf8 100644 --- a/crates/ironrdp-session/src/x224/mod.rs +++ b/crates/ironrdp-session/src/x224/mod.rs @@ -248,16 +248,13 @@ impl Processor { }; let channel_id = data_ctx.channel_id; - // Message-channel PDUs carry a Basic Security Header, excluded below. - // Ordinary session data on TLS-protected IO/SVC channels does not. - if self.message_channel_id != Some(channel_id) { - self.record_bandwidth_bytes(data_ctx.user_data.len()); - } 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, received_at) } else { + // TLS-protected SVC data has no RDP Security Header: 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)?; @@ -272,6 +269,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); @@ -282,9 +280,13 @@ 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 TLS-protected IO data has no RDP Security Header. 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; @@ -466,17 +468,13 @@ impl Processor { Ok(decompressed) } - /// Counts session bytes after transport/security headers while a continuous - /// bandwidth window is open ([MS-RDPBCGR] 3.2.5.14). + /// 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); } - /// Whether a continuous bandwidth window is open, so session bytes are being counted. - pub(crate) fn is_counting_bandwidth(&self) -> bool { - self.auto_detect.is_counting() - } - /// 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). @@ -701,6 +699,7 @@ mod tests { channel_id: 1003, user_data: &encoded, }, + encoded.len(), &mut None, ) .expect("ignore a misrouted optional multitransport request"); @@ -918,6 +917,7 @@ mod tests { channel_id: 1003, user_data: &user_data, }, + user_data.len(), &mut None, ) .expect("concatenated Share Control PDUs should be split"); @@ -939,6 +939,7 @@ mod tests { channel_id: 1003, user_data: &[0x06], }, + 1, &mut None, ) .expect_err("a truncated totalLength field is invalid"); @@ -959,6 +960,7 @@ mod tests { channel_id: 1003, user_data: &user_data, }, + user_data.len(), &mut None, ) .expect_err("an overrunning concatenated totalLength is invalid"); @@ -984,6 +986,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 2f35ed2f7c..f3c58de136 100644 --- a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs @@ -1,11 +1,18 @@ 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; @@ -38,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), }); @@ -189,7 +199,7 @@ fn bandwidth_result_frame(frame: &[u8]) -> (u16, u16, u32, u32) { } #[test] -fn continuous_measurement_counts_data_without_security_headers() { +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); @@ -199,6 +209,82 @@ fn continuous_measurement_counts_data_without_security_headers() { 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(); @@ -276,6 +362,36 @@ fn untimed_driver_does_not_report_accumulated_bytes_as_a_real_measurement() { 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(), @@ -340,10 +456,12 @@ fn continuous_measurement_counts_fast_path_data_only_inside_the_window() { let [ActiveStageOutput::ResponseFrame(response)] = outputs.as_slice() else { panic!("expected exactly one bandwidth response, got {outputs:?}"); }; - // The frame between Start and Stop counts without its fast-path header: a - // three-byte update header and 100 bytes of update data. The Stop adds its - // six-byte auto-detect header, as in the message-channel tests above. - assert_eq!(bandwidth_result_frame(response), (2, 0x000b, 30, 3 + 100 + 6)); + // 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 { From 33080ea761de0b3d89c691689d78e4ebbaf59607 Mon Sep 17 00:00:00 2001 From: AKolenda Date: Fri, 2 Oct 2026 00:33:51 -0600 Subject: [PATCH 12/12] test(client): verify tunnel fallback and complete frame accounting --- crates/ironrdp-testsuite-core/tests/session/autodetect.rs | 5 +++-- .../tests/client/tunnel_auto_detect.rs | 8 ++++---- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs index f3c58de136..302e0981f9 100644 --- a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs @@ -531,6 +531,7 @@ fn tunnel_and_main_connection_measure_separately() { let [ActiveStageOutput::ResponseFrame(response)] = outputs.as_slice() else { panic!("expected exactly one bandwidth response, got {outputs:?}"); }; - // The fast-path data and the Stop's own six bytes, but none of the tunnel's 500. - assert_eq!(bandwidth_result_frame(response), (2, 0x000b, 20, 3 + 100 + 6)); + // The complete fast-path frame and the Stop's own six bytes, but none of the tunnel's 500. + let tcp_bytes = u32::try_from(fast_path_frame(100).len()).unwrap(); + assert_eq!(bandwidth_result_frame(response), (2, 0x000b, 20, tcp_bytes + 6)); } diff --git a/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs b/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs index 43098761e7..4dc6345d8d 100644 --- a/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs +++ b/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs @@ -74,7 +74,7 @@ fn tunnel_failure_before_migration_withdraws_soft_sync_without_ending_the_sessio ironrdp_client::udp::disable_failed_tunnel( &mut stage, &mut transport, - ironrdp_session::general_err!("auto-detect reply send failed"), + ironrdp::session::general_err!("auto-detect reply send failed"), ) .expect("unused tunnel must fall back to TCP"); assert!(transport.is_none()); @@ -93,14 +93,14 @@ fn tunnel_failure_after_migration_requires_reconnection() { let error = ironrdp_client::udp::disable_failed_tunnel( &mut stage, &mut transport, - ironrdp_session::general_err!("auto-detect reply send failed"), + ironrdp::session::general_err!("auto-detect reply send failed"), ) .expect_err("migrated channels cannot silently resume on TCP"); assert!(error.to_string().contains("auto-detect reply send failed")); assert!(stage.reliable_udp_dvc_tunnel_in_use()); } -fn make_stage(migrated: bool) -> ironrdp_session::ActiveStage { +fn make_stage(migrated: bool) -> ironrdp::session::ActiveStage { use ironrdp_dvc::pdu::SoftSyncTunnelType; use ironrdp_svc::SvcProcessor as _; @@ -112,7 +112,7 @@ fn make_stage(migrated: bool) -> ironrdp_session::ActiveStage { } let mut static_channels = ironrdp_svc::StaticChannelSet::new(); static_channels.insert(drdynvc); - ironrdp_session::ActiveStageBuilder { + ironrdp::session::ActiveStageBuilder { static_channels, user_channel_id: 1001, io_channel_id: 1003,