feat(session): answer bandwidth measurements during the session

crates/ironrdp-client/src/rdp.rs

@@ -3165,7 +3165,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);

crates/ironrdp-session/src/active_stage.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use ironrdp_bulk::{BulkCompressor, CompressionType as BulkCompressionType};
-use ironrdp_core::{ReadCursor, WriteBuf};
+use ironrdp_core::{MonotonicInstant, ReadCursor, WriteBuf};
 use ironrdp_displaycontrol::client::DisplayControlClient;
 use ironrdp_dvc::pdu::SoftSyncTunnelType;
 use ironrdp_dvc::{DrdynvcClient, DvcClientProcessor, DvcMessageBatch, DynamicChannelMut, DynamicChannelRef};
@@ -183,15 +183,37 @@ 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<Vec<ActiveStageOutput>> {
+        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<MonotonicInstant>,
     ) -> SessionResult<Vec<ActiveStageOutput>> {
         self.damage_regions.clear();
         let (mut stage_outputs, processor_updates) = match action {
             Action::FastPath => {
+                // 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
@@ -202,7 +224,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();
 

crates/ironrdp-session/src/autodetect.rs

@@ -0,0 +1,148 @@
+//! 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<BandwidthMeasurement>,
+}
+
+/// Size of the auto-detect header fields [MS-RDPBCGR] 3.2.5.14 counts along with payloadLength:
+/// headerLength, headerTypeId, sequenceNumber, requestType and payloadLength itself.
+///
+/// [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/16ffa852-8aa7-481c-99a0-36c1a9a198f6
+const AUTO_DETECT_HEADER_LEN: u32 = 8;
+
+/// Reported as `timeDelta` for a window that was not timed, and the floor for one that was:
+/// a server computing `byteCount * 8 / timeDelta` divides by it ([MS-RDPBCGR] 3.3.5.14).
+const UNMEASURABLE_INTERVAL_MS: u32 = 1;
+
+/// Bytes a connect-time Payload or Stop adds to the count: payloadLength plus the auto-detect
+/// header, but not the security header. This is the rule the connector applies to the same PDUs.
+fn counted_len(payload_len: usize) -> u32 {
+    u32::try_from(payload_len)
+        .unwrap_or(u32::MAX)
+        .saturating_add(AUTO_DETECT_HEADER_LEN)
+}
+
+#[derive(Debug)]
+struct BandwidthMeasurement {
+    started_at: MonotonicInstant,
+    bytes: u32,
+    continuous: bool,
+}
+
+impl AutoDetectResponder {
+    /// Counts received bytes while a continuous bandwidth window is open.
+    pub(crate) fn record_bytes(&mut self, bytes: usize) {
+        if let Some(measurement) = self.bandwidth.as_mut().filter(|measurement| measurement.continuous) {
+            measurement.bytes = measurement
+                .bytes
+                .saturating_add(u32::try_from(bytes).unwrap_or(u32::MAX));
+        }
+    }
+
+    /// Returns the response `request` calls for, if any.
+    ///
+    /// Without an arrival time, a bandwidth measurement is answered with an untimed, zero-byte
+    /// result.
+    pub(crate) fn respond(
+        &mut self,
+        request: AutoDetectRequest,
+        received_at: Option<MonotonicInstant>,
+    ) -> Option<AutoDetectResponse> {
+        match request {
+            AutoDetectRequest::RttRequest { sequence_number, .. } => {
+                Some(AutoDetectResponse::RttResponse { sequence_number })
+            }
+            AutoDetectRequest::BandwidthMeasureStart { request_type, .. }
+                if matches!(request_type, BW_START_CONNECT_TIME | BW_START_RELIABLE_UDP) =>
+            {
+                self.bandwidth = received_at.map(|started_at| BandwidthMeasurement {
+                    started_at,
+                    bytes: 0,
+                    continuous: request_type == BW_START_RELIABLE_UDP,
+                });
+                None
+            }
+            AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => {
+                if let Some(measurement) = self.bandwidth.as_mut().filter(|measurement| !measurement.continuous) {
+                    measurement.bytes = measurement.bytes.saturating_add(counted_len(payload.len()));
+                }
+                None
+            }
+            AutoDetectRequest::BandwidthMeasureStop {
+                sequence_number,
+                request_type,
+                payload,
+            } if matches!(request_type, BW_STOP_CONNECT_TIME | BW_STOP_RELIABLE_UDP) => {
+                let continuous = request_type == BW_STOP_RELIABLE_UDP;
+                let measurement = self
+                    .bandwidth
+                    .take()
+                    .filter(|measurement| measurement.continuous == continuous);
+                let stop_bytes = if continuous {
+                    0
+                } else {
+                    counted_len(payload.as_ref().map_or(0, Vec::len))
+                };
+                let (time_delta_ms, byte_count) = match (measurement, received_at) {
+                    (Some(measurement), Some(stopped_at)) => (
+                        u32::try_from(stopped_at.duration_since(measurement.started_at).as_millis())
+                            .unwrap_or(u32::MAX)
+                            .max(UNMEASURABLE_INTERVAL_MS),
+                        measurement.bytes.saturating_add(stop_bytes),
+                    ),
+                    (Some(measurement), None) => {
+                        // The window was timed but this Stop was not, so there is nothing to
+                        // divide the count by. Log the drop so it does not look like the
+                        // ordinary no-window case.
+                        debug!(
+                            dropped_bytes = measurement.bytes,
+                            "Bandwidth Measure Stop arrived with no arrival time although its window was open; \
+                             dropping the accumulated count"
+                        );
+                        (UNMEASURABLE_INTERVAL_MS, 0)
+                    }
+                    (None, _) => (UNMEASURABLE_INTERVAL_MS, 0),
+                };
+                Some(AutoDetectResponse::BandwidthMeasureResults {
+                    sequence_number,
+                    response_type: if continuous {
+                        BW_RESULTS_CONTINUOUS
+                    } else {
+                        BW_RESULTS_CONNECT_TIME
+                    },
+                    time_delta_ms,
+                    byte_count,
+                })
+            }
+            request @ AutoDetectRequest::NetworkCharacteristicsResult { .. } => {
+                // The 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
+            }
+            request => {
+                // Measurements for a lossy transport are not answered here.
+                debug!(?request, "Ignoring auto-detect request for another transport");
+                None
+            }
+        }
+    }
+}

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;