From f90253ed99857c96130760ec4562e956c8c251e6 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Thu, 10 Sep 2026 12:49:41 -0500 Subject: [PATCH 1/2] feat(rdpeai): add RdpeaiServer orchestration for AUDIO_INPUT ironrdp-rdpeai only had a client processor (PR #1645); the PDU layer (pdu.rs) was already bidirectional, but nothing drove the server side of the MS-RDPEAI state machine. RdpeaiServer fills that gap, mirroring RdpeaiClient's structure and the RdpdrServer precedent (PR #1783): sends Version on channel start and Formats once the client replies, automatically; exposes open()/change_format() for the consuming application to call once it wants to record; and delivers decoded audio to an injected RdpeaiServerBackend. Out-of-sequence, unrecognized, and mismatched PDUs are logged and ignored rather than treated as errors, per MS-RDPEAI 3.1.5's explicit MUST-ignore requirement. change_format() honors 3.3.5.3.1's SHOULD NOT for AAC formats when the client only advertised protocol version 1. 14 new tests in ironrdp-testsuite-core cover the full handshake, open failure/retry, format-change round trip and the AAC/V1 skip rule, out-of-sequence handling, and channel close. ironrdp-server integration (a factory trait plus attach_channels wiring, matching RdpeiServerFactory's shape) is a separate follow-up, same split RdpdrServer took. --- crates/ironrdp-rdpeai/src/lib.rs | 1 + crates/ironrdp-rdpeai/src/server.rs | 395 +++++++++++++++++ .../tests/rdpeai/mod.rs | 1 + .../tests/rdpeai/server.rs | 404 ++++++++++++++++++ 4 files changed, 801 insertions(+) create mode 100644 crates/ironrdp-rdpeai/src/server.rs create mode 100644 crates/ironrdp-testsuite-core/tests/rdpeai/server.rs diff --git a/crates/ironrdp-rdpeai/src/lib.rs b/crates/ironrdp-rdpeai/src/lib.rs index 659161514f..2a51aac0b6 100644 --- a/crates/ironrdp-rdpeai/src/lib.rs +++ b/crates/ironrdp-rdpeai/src/lib.rs @@ -6,5 +6,6 @@ pub const CHANNEL_NAME: &str = "AUDIO_INPUT"; pub mod client; pub mod pdu; +pub mod server; pub use ironrdp_rdpsnd::pdu::{AudioFormat, WaveFormat}; diff --git a/crates/ironrdp-rdpeai/src/server.rs b/crates/ironrdp-rdpeai/src/server.rs new file mode 100644 index 0000000000..b1d136a74b --- /dev/null +++ b/crates/ironrdp-rdpeai/src/server.rs @@ -0,0 +1,395 @@ +//! AUDIO_INPUT dynamic virtual channel server (MS-RDPEAI). + +use ironrdp_core::{decode, impl_as_any}; +use ironrdp_dvc::{DvcMessage, DvcProcessor, DvcServerProcessor}; +use ironrdp_pdu::{PduResult, decode_err, pdu_other_err}; +use ironrdp_rdpsnd::pdu::{AudioFormat, WaveFormat}; +use tracing::{debug, trace, warn}; + +use crate::CHANNEL_NAME; +use crate::pdu::{DataPdu, FormatChangePdu, FormatsPdu, OpenPdu, OpenReplyPdu, RdpeaiPdu, Version, VersionPdu}; + +/// Handler for the server side of the Audio Input Redirection Virtual Channel (`AUDIO_INPUT`). +/// +/// Implementations supply the list of audio formats the server offers and receive the +/// client's captured audio once [`RdpeaiServer::open`] succeeds. `RdpeaiServer` owns the +/// MS-RDPEAI state machine; the backend only reacts to negotiated state and produces raw +/// audio bytes to consumers (e.g. a PipeWire virtual microphone source). +pub trait RdpeaiServerBackend: Send { + /// Formats this server offers, in preferred order. Sent verbatim in the Sound Formats PDU + /// (MS-RDPEAI 2.2.2.2) once the client's Version PDU is processed. + fn supported_formats(&self) -> &[AudioFormat]; + + /// Called once the client has replied with its own Sound Formats PDU, establishing the + /// final negotiated list (MS-RDPEAI 3.3.5.1.5). `negotiated` is the client's subset, in the + /// client's order — the same list [`RdpeaiServer::open`]'s `initial_format` and + /// [`RdpeaiServer::change_format`]'s `new_format` index into. May be empty if the client + /// advertised no format the server also offers, in which case recording is unavailable + /// for the session. + fn on_formats_negotiated(&mut self, negotiated: &[AudioFormat]) { + let _ = negotiated; + } + + /// Called when the client answers an Open PDU (MS-RDPEAI 3.3.5.1.8). `result` is the + /// client's raw HRESULT; `result == 0` ([`OpenReplyPdu::S_OK`]) means capture started. + fn on_open_reply(&mut self, result: i32) { + let _ = result; + } + + /// Called for every decoded audio packet the client sends while opened + /// (MS-RDPEAI 3.3.5.2.2). `format` is the format currently in effect. `data` is the raw + /// payload from the Data PDU, encoded per `format` — this crate does not decode compressed + /// formats; PCM formats are already raw samples. + fn on_audio_data(&mut self, format: &AudioFormat, data: &[u8]); + + /// Called once the client confirms a server-initiated format change + /// (MS-RDPEAI 3.3.5.3.2). Not called for the initial format confirm that precedes the + /// first Open Reply — see [`Self::on_open_reply`] for that. + fn on_format_change_confirmed(&mut self, format: &AudioFormat) { + let _ = format; + } + + /// Called when the dynamic virtual channel is torn down. + fn on_close(&mut self) {} +} + +/// A [`RdpeaiServerBackend`] that offers no formats and drops all audio. +#[derive(Debug, Default)] +pub struct NoopRdpeaiServerBackend; + +impl RdpeaiServerBackend for NoopRdpeaiServerBackend { + fn supported_formats(&self) -> &[AudioFormat] { + &[] + } + + fn on_audio_data(&mut self, _format: &AudioFormat, _data: &[u8]) {} +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum State { + /// Channel not yet started. + Start, + /// Server Version PDU sent; waiting for the client's Version PDU. + AwaitingClientVersion, + /// Server Formats PDU sent; waiting for the client's Formats PDU. + AwaitingClientFormats, + /// Formats negotiated; no Open outstanding. + Ready, + /// Open PDU sent; waiting for the client's FormatChange confirming the initial format. + AwaitingFormatConfirm, + /// Initial FormatChange confirmed; waiting for the client's Open Reply. + AwaitingOpenReply, + /// Capture device open; Data PDUs expected. + Opened, +} + +/// Server processor for the `AUDIO_INPUT` dynamic virtual channel (MS-RDPEAI). +/// +/// Drives the initialization sequence automatically on channel start (Version, then Formats, +/// per 3.3.5.1.1-3.3.5.1.5): the server always advertises its formats immediately, since +/// nothing in the protocol depends on a higher-layer "start recording" decision until +/// [`Self::open`] is called. Recording itself is caller-initiated: call [`Self::open`] once +/// the negotiated formats are known ([`RdpeaiServerBackend::on_formats_negotiated`]) and the +/// consuming application actually wants to record, then forward the returned PDUs to the +/// client over the `channel_id` this processor was started with. +/// +/// Malformed, unrecognized, or out-of-sequence PDUs are logged and ignored rather than +/// treated as errors, per MS-RDPEAI 3.1.5's explicit MUST-ignore requirement. +pub struct RdpeaiServer { + backend: Box, + state: State, + channel_id: Option, + client_version: Option, + /// Negotiated format list, in the client's reply order — Open/FormatChange indices refer + /// here (MS-RDPEAI 3.1.1: this is the list the client establishes as final). + negotiated_formats: Vec, + /// Index into `negotiated_formats` currently in effect once opened. + current_format: Option, + /// `initialFormat` from the outstanding Open PDU, pending the client's confirming + /// FormatChange. + pending_open_format: Option, + /// `NewFormat` from an outstanding server-initiated FormatChange, pending confirm. + pending_format_change: Option, +} + +impl RdpeaiServer { + pub fn new(backend: Box) -> Self { + Self { + backend, + state: State::Start, + channel_id: None, + client_version: None, + negotiated_formats: Vec::new(), + current_format: None, + pending_open_format: None, + pending_format_change: None, + } + } + + /// Formats negotiated with the client (MS-RDPEAI 3.3.5.1.5); empty before that completes + /// or if no common format exists. + pub fn negotiated_formats(&self) -> &[AudioFormat] { + &self.negotiated_formats + } + + /// The format currently in effect, once [`Self::open`] has succeeded. + pub fn current_format(&self) -> Option<&AudioFormat> { + self.current_format.and_then(|idx| self.resolve_format(idx)) + } + + /// The protocol version the client advertised, once its Version PDU has been processed. + pub fn client_version(&self) -> Option { + self.client_version + } + + fn resolve_format(&self, index: u32) -> Option<&AudioFormat> { + let idx = usize::try_from(index).ok()?; + self.negotiated_formats.get(idx) + } + + /// Request the client start recording (MS-RDPEAI 3.3.5.1.6 / 3.1.4.1). + /// + /// `initial_format` indexes [`Self::negotiated_formats`] and selects the encoding used for + /// captured Data PDUs; `capture_format` is the WAVEFORMATEX the client's capture device is + /// asked to open with (MS-RDPEAI 2.2.2.3) and MAY differ from the encoding format when the + /// client captures PCM and encodes afterward. `frames_per_packet` bounds each Data PDU to + /// `nChannels * 2 * FramesPerPacket` bytes (2.2.2.3 / 2.2.3.2). + /// + /// Valid only once formats are negotiated and no Open is already outstanding or active; + /// returns an error otherwise rather than sending a second, contract-violating Open PDU. + /// Errors are returned, not panics, since the caller may race the negotiation completing. + pub fn open( + &mut self, + frames_per_packet: u32, + initial_format: u32, + capture_format: AudioFormat, + ) -> PduResult> { + if self.state != State::Ready { + return Err(pdu_other_err!("RdpeaiServer::open called outside the Ready state")); + } + if self.resolve_format(initial_format).is_none() { + return Err(pdu_other_err!("RdpeaiServer::open initial_format out of range")); + } + + self.pending_open_format = Some(initial_format); + self.state = State::AwaitingFormatConfirm; + + Ok(vec![Box::new(RdpeaiPdu::Open(OpenPdu { + frames_per_packet, + initial_format, + capture_format, + }))]) + } + + /// Request the client switch to a different negotiated format while streaming + /// (MS-RDPEAI 3.3.5.3.1). `new_format` indexes [`Self::negotiated_formats`]. + /// + /// Per 3.3.5.3.1, when the currently active format is AAC and the client only advertised + /// protocol version 1, the server SHOULD NOT send this PDU: that case logs a warning and + /// returns no messages rather than sending a PDU the spec discourages. Valid only while + /// [`Self::current_format`] is set (i.e. after a successful [`Self::open`]). + pub fn change_format(&mut self, new_format: u32) -> PduResult> { + if self.state != State::Opened { + return Err(pdu_other_err!( + "RdpeaiServer::change_format called outside the Opened state" + )); + } + if self.resolve_format(new_format).is_none() { + return Err(pdu_other_err!("RdpeaiServer::change_format new_format out of range")); + } + + let active_is_aac = self + .current_format() + .is_some_and(|fmt| fmt.format == WaveFormat::AAC_MS); + if active_is_aac && self.client_version == Some(Version::V1) { + warn!( + "Skipping AUDIO_INPUT FormatChange: active format is AAC and the client only \ + advertised protocol version 1 (MS-RDPEAI 3.3.5.3.1 SHOULD NOT)" + ); + return Ok(Vec::new()); + } + + self.pending_format_change = Some(new_format); + Ok(vec![Box::new(RdpeaiPdu::FormatChange(FormatChangePdu::new( + new_format, + )))]) + } + + fn handle_version(&mut self, pdu: VersionPdu) -> PduResult> { + if self.state != State::AwaitingClientVersion { + warn!(?self.state, "Ignoring out-of-sequence AUDIO_INPUT Version PDU"); + return Ok(Vec::new()); + } + self.client_version = Some(pdu.version); + self.state = State::AwaitingClientFormats; + debug!(client_version = ?pdu.version, "AUDIO_INPUT version received"); + // MS-RDPEAI 3.3.5.1.3: the server MUST send a Sound Formats PDU after processing the + // client's Version PDU. + Ok(vec![Box::new(RdpeaiPdu::Formats(FormatsPdu::server( + self.backend.supported_formats().to_vec(), + )))]) + } + + fn handle_formats(&mut self, client: FormatsPdu) -> PduResult> { + if self.state != State::AwaitingClientFormats { + warn!(?self.state, "Ignoring out-of-sequence AUDIO_INPUT Formats PDU"); + return Ok(Vec::new()); + } + + // MS-RDPEAI 3.2.5.1.5: the client's list MUST be a subset of what the server offered. + // Enforce defensively rather than trusting the client's claim verbatim. + let offered = self.backend.supported_formats(); + let negotiated: Vec = client + .formats + .into_iter() + .filter(|fmt| offered.iter().any(|server_fmt| fmt.matches_for_negotiation(server_fmt))) + .collect(); + + if negotiated.is_empty() { + warn!("No common AUDIO_INPUT formats after client reply; recording is unavailable this session"); + } else { + debug!(count = negotiated.len(), "AUDIO_INPUT formats negotiated"); + } + + self.negotiated_formats = negotiated; + self.state = State::Ready; + self.backend.on_formats_negotiated(&self.negotiated_formats); + Ok(Vec::new()) + } + + fn handle_format_change(&mut self, pdu: FormatChangePdu) -> PduResult> { + match self.state { + State::AwaitingFormatConfirm => { + // MS-RDPEAI 3.3.5.1.7: before the Open Reply, the client confirms the initial + // format chosen by the server's Open PDU. + let Some(pending) = self.pending_open_format else { + warn!("AwaitingFormatConfirm with no pending Open format; ignoring"); + return Ok(Vec::new()); + }; + if pdu.new_format != pending { + warn!( + expected = pending, + got = pdu.new_format, + "Initial AUDIO_INPUT FormatChange confirm does not match the format from Open" + ); + } + self.current_format = Some(pdu.new_format); + self.pending_open_format = None; + self.state = State::AwaitingOpenReply; + Ok(Vec::new()) + } + State::Opened => { + // MS-RDPEAI 3.3.5.3.2: client confirms a server-initiated format change. + let Some(pending) = self.pending_format_change.take() else { + warn!("Unsolicited AUDIO_INPUT FormatChange confirm while Opened; ignoring"); + return Ok(Vec::new()); + }; + if pdu.new_format != pending { + warn!( + expected = pending, + got = pdu.new_format, + "AUDIO_INPUT FormatChange confirm does not match the requested format" + ); + } + self.current_format = Some(pdu.new_format); + if let Some(fmt) = self.current_format().cloned() { + self.backend.on_format_change_confirmed(&fmt); + } + Ok(Vec::new()) + } + _ => { + warn!(?self.state, "Ignoring out-of-sequence AUDIO_INPUT FormatChange PDU"); + Ok(Vec::new()) + } + } + } + + fn handle_open_reply(&mut self, pdu: OpenReplyPdu) -> PduResult> { + if self.state != State::AwaitingOpenReply { + warn!(?self.state, "Ignoring out-of-sequence AUDIO_INPUT OpenReply PDU"); + return Ok(Vec::new()); + } + + self.backend.on_open_reply(pdu.result); + if pdu.result == OpenReplyPdu::S_OK { + self.state = State::Opened; + debug!("AUDIO_INPUT capture opened"); + } else { + // MS-RDPEAI 3.3.5.1.8: on failure the server MAY send another Open PDU; leave that + // to the caller by returning to Ready rather than retrying automatically. + self.current_format = None; + self.state = State::Ready; + warn!( + result = pdu.result, + "AUDIO_INPUT client failed to open its capture device" + ); + } + Ok(Vec::new()) + } + + fn handle_data(&mut self, pdu: DataPdu) -> PduResult> { + if self.state != State::Opened { + warn!(?self.state, "Ignoring AUDIO_INPUT Data PDU outside the Opened state"); + return Ok(Vec::new()); + } + let Some(format) = self.current_format().cloned() else { + warn!("Opened with no current format resolved; dropping audio data"); + return Ok(Vec::new()); + }; + self.backend.on_audio_data(&format, &pdu.data); + Ok(Vec::new()) + } +} + +impl_as_any!(RdpeaiServer); + +impl DvcProcessor for RdpeaiServer { + fn channel_name(&self) -> &str { + CHANNEL_NAME + } + + fn start(&mut self, channel_id: u32) -> PduResult> { + self.channel_id = Some(channel_id); + self.state = State::AwaitingClientVersion; + self.client_version = None; + self.negotiated_formats.clear(); + self.current_format = None; + self.pending_open_format = None; + self.pending_format_change = None; + debug!(channel_id, "AUDIO_INPUT channel started"); + // MS-RDPEAI 3.3.5.1.1: the Version PDU MUST be the first PDU sent by the server. + Ok(vec![Box::new(RdpeaiPdu::Version(VersionPdu::new(Version::V2)))]) + } + + fn process(&mut self, _channel_id: u32, payload: &[u8]) -> PduResult> { + let pdu: RdpeaiPdu = decode(payload).map_err(|e| decode_err!(e))?; + trace!(?pdu, "AUDIO_INPUT PDU received"); + match pdu { + RdpeaiPdu::Version(v) => self.handle_version(v), + RdpeaiPdu::Formats(f) => self.handle_formats(f), + RdpeaiPdu::FormatChange(c) => self.handle_format_change(c), + RdpeaiPdu::OpenReply(r) => self.handle_open_reply(r), + // MS-RDPEAI 3.3.5.1.4 / 3.3.5.2.1: diagnostic only, precedes Formats or Data. + RdpeaiPdu::DataIncoming => Ok(Vec::new()), + RdpeaiPdu::Data(d) => self.handle_data(d), + RdpeaiPdu::Open(_) => { + warn!("Ignoring server-originated AUDIO_INPUT Open PDU received from the client"); + Ok(Vec::new()) + } + } + } + + fn close(&mut self, _channel_id: u32) { + self.backend.on_close(); + self.channel_id = None; + self.state = State::Start; + self.client_version = None; + self.negotiated_formats.clear(); + self.current_format = None; + self.pending_open_format = None; + self.pending_format_change = None; + debug!("AUDIO_INPUT channel closed"); + } +} + +impl DvcServerProcessor for RdpeaiServer {} diff --git a/crates/ironrdp-testsuite-core/tests/rdpeai/mod.rs b/crates/ironrdp-testsuite-core/tests/rdpeai/mod.rs index 8ef9d10c99..b180621aa1 100644 --- a/crates/ironrdp-testsuite-core/tests/rdpeai/mod.rs +++ b/crates/ironrdp-testsuite-core/tests/rdpeai/mod.rs @@ -1,2 +1,3 @@ mod client; mod pdu; +mod server; diff --git a/crates/ironrdp-testsuite-core/tests/rdpeai/server.rs b/crates/ironrdp-testsuite-core/tests/rdpeai/server.rs new file mode 100644 index 0000000000..ec07797f7f --- /dev/null +++ b/crates/ironrdp-testsuite-core/tests/rdpeai/server.rs @@ -0,0 +1,404 @@ +use std::sync::{Arc, Mutex}; + +use ironrdp_core::{decode, encode_vec}; +use ironrdp_dvc::DvcProcessor as _; +use ironrdp_rdpeai::pdu::{ + DataPdu, FormatChangePdu, FormatsPdu, OpenPdu, OpenReplyPdu, RdpeaiPdu, Version, VersionPdu, pcm_format, +}; +use ironrdp_rdpeai::server::{NoopRdpeaiServerBackend, RdpeaiServer, RdpeaiServerBackend}; +use ironrdp_rdpsnd::pdu::{AudioFormat, WaveFormat}; + +#[derive(Default)] +struct MockBackendState { + negotiated: Vec, + open_replies: Vec, + audio: Vec<(AudioFormat, Vec)>, + format_changes: Vec, + closed: bool, +} + +#[derive(Clone, Default)] +struct MockBackend { + formats: Vec, + state: Arc>, +} + +impl MockBackend { + fn new(formats: Vec) -> Self { + Self { + formats, + state: Arc::new(Mutex::new(MockBackendState::default())), + } + } +} + +impl RdpeaiServerBackend for MockBackend { + fn supported_formats(&self) -> &[AudioFormat] { + &self.formats + } + + fn on_formats_negotiated(&mut self, negotiated: &[AudioFormat]) { + self.state.lock().unwrap().negotiated = negotiated.to_vec(); + } + + fn on_open_reply(&mut self, result: i32) { + self.state.lock().unwrap().open_replies.push(result); + } + + fn on_audio_data(&mut self, format: &AudioFormat, data: &[u8]) { + self.state.lock().unwrap().audio.push((format.clone(), data.to_vec())); + } + + fn on_format_change_confirmed(&mut self, format: &AudioFormat) { + self.state.lock().unwrap().format_changes.push(format.clone()); + } + + fn on_close(&mut self) { + self.state.lock().unwrap().closed = true; + } +} + +fn decode_dvc(msg: &ironrdp_dvc::DvcMessage) -> RdpeaiPdu { + let bytes = encode_vec(msg.as_ref()).expect("encode dvc message"); + decode(&bytes).expect("decode dvc message") +} + +fn process_encoded(server: &mut RdpeaiServer, channel_id: u32, pdu: RdpeaiPdu) -> Vec { + server + .process(channel_id, &encode_vec(&pdu).expect("encode")) + .expect("process") +} + +/// Drive the server through Version -> Formats -> Open -> FormatChange confirm -> OpenReply, +/// returning the backend so callers can inspect what it observed. +fn negotiate_and_open( + server: &mut RdpeaiServer, + channel_id: u32, + client_formats: Vec, + frames_per_packet: u32, + initial_format: u32, + capture_format: AudioFormat, + open_result: i32, +) { + let start_out = server.start(channel_id).expect("start"); + assert_eq!(start_out.len(), 1); + assert!(matches!(decode_dvc(&start_out[0]), RdpeaiPdu::Version(_))); + + let version_out = process_encoded(server, channel_id, RdpeaiPdu::Version(VersionPdu::new(Version::V1))); + assert_eq!(version_out.len(), 1); + assert!(matches!(decode_dvc(&version_out[0]), RdpeaiPdu::Formats(_))); + + let formats_out = process_encoded( + server, + channel_id, + RdpeaiPdu::Formats(FormatsPdu::client(client_formats)), + ); + assert!(formats_out.is_empty()); + + let open_out = server + .open(frames_per_packet, initial_format, capture_format) + .expect("open"); + assert_eq!(open_out.len(), 1); + assert!(matches!(decode_dvc(&open_out[0]), RdpeaiPdu::Open(_))); + + let confirm_out = process_encoded( + server, + channel_id, + RdpeaiPdu::FormatChange(FormatChangePdu::new(initial_format)), + ); + assert!(confirm_out.is_empty()); + + let reply_out = process_encoded( + server, + channel_id, + RdpeaiPdu::OpenReply(OpenReplyPdu { result: open_result }), + ); + assert!(reply_out.is_empty()); +} + +#[test] +fn version_and_formats_negotiate_correctly() { + let fmt_a = pcm_format(1, 16000, 16); + let fmt_b = pcm_format(2, 48000, 16); + let backend = MockBackend::new(vec![fmt_a.clone(), fmt_b.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend.clone())); + + let start_out = server.start(1).expect("start"); + match decode_dvc(&start_out[0]) { + RdpeaiPdu::Version(v) => assert_eq!(v.version, Version::V2), + other => panic!("expected server Version PDU, got {other:?}"), + } + + let version_out = process_encoded(&mut server, 1, RdpeaiPdu::Version(VersionPdu::new(Version::V1))); + match decode_dvc(&version_out[0]) { + RdpeaiPdu::Formats(f) => assert_eq!(f.formats, vec![fmt_a.clone(), fmt_b]), + other => panic!("expected server Formats PDU, got {other:?}"), + } + + // Client accepts only fmt_a. + let _ = process_encoded( + &mut server, + 1, + RdpeaiPdu::Formats(FormatsPdu::client(vec![fmt_a.clone()])), + ); + assert_eq!(server.negotiated_formats(), &[fmt_a]); + assert_eq!(backend.state.lock().unwrap().negotiated, server.negotiated_formats()); + assert_eq!(server.client_version(), Some(Version::V1)); +} + +#[test] +fn formats_reply_is_filtered_to_the_servers_offered_set() { + // MS-RDPEAI 3.2.5.1.5: the client's list MUST be a subset of the server's; a + // non-conformant client claiming a format the server never offered must not poison + // the negotiated list. + let fmt_a = pcm_format(1, 16000, 16); + let unoffered = pcm_format(2, 96000, 24); + let backend = MockBackend::new(vec![fmt_a.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend)); + + server.start(1).unwrap(); + let _ = process_encoded(&mut server, 1, RdpeaiPdu::Version(VersionPdu::new(Version::V1))); + let _ = process_encoded( + &mut server, + 1, + RdpeaiPdu::Formats(FormatsPdu::client(vec![fmt_a.clone(), unoffered])), + ); + + assert_eq!(server.negotiated_formats(), &[fmt_a]); +} + +#[test] +fn open_through_data_delivers_audio_to_the_backend() { + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend.clone())); + + negotiate_and_open( + &mut server, + 1, + vec![fmt.clone()], + 320, + 0, + fmt.clone(), + OpenReplyPdu::S_OK, + ); + + assert_eq!(backend.state.lock().unwrap().open_replies, vec![OpenReplyPdu::S_OK]); + assert_eq!(server.current_format(), Some(&fmt)); + + let data_out = process_encoded(&mut server, 1, RdpeaiPdu::Data(DataPdu::new(vec![1, 2, 3, 4]))); + assert!(data_out.is_empty()); + + let audio = backend.state.lock().unwrap().audio.clone(); + assert_eq!(audio, vec![(fmt, vec![1, 2, 3, 4])]); +} + +#[test] +fn data_before_open_is_ignored() { + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend.clone())); + + server.start(1).unwrap(); + let _ = process_encoded(&mut server, 1, RdpeaiPdu::Version(VersionPdu::new(Version::V1))); + let _ = process_encoded(&mut server, 1, RdpeaiPdu::Formats(FormatsPdu::client(vec![fmt]))); + + // Still Ready: no Open in flight yet. + let _ = process_encoded(&mut server, 1, RdpeaiPdu::Data(DataPdu::new(vec![9, 9, 9]))); + assert!(backend.state.lock().unwrap().audio.is_empty()); +} + +#[test] +fn open_rejects_out_of_range_initial_format() { + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend)); + + server.start(1).unwrap(); + let _ = process_encoded(&mut server, 1, RdpeaiPdu::Version(VersionPdu::new(Version::V1))); + let _ = process_encoded( + &mut server, + 1, + RdpeaiPdu::Formats(FormatsPdu::client(vec![fmt.clone()])), + ); + + assert!(server.open(320, 7, fmt).is_err()); +} + +#[test] +fn open_rejects_when_not_yet_negotiated() { + let backend = NoopRdpeaiServerBackend; + let mut server = RdpeaiServer::new(Box::new(backend)); + server.start(1).unwrap(); + // Version/Formats round trip has not happened: state is not Ready. + assert!(server.open(320, 0, pcm_format(1, 16000, 16)).is_err()); +} + +#[test] +fn open_reply_failure_returns_to_ready_and_allows_retry() { + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend.clone())); + + negotiate_and_open( + &mut server, + 1, + vec![fmt.clone()], + 320, + 0, + fmt.clone(), + OpenReplyPdu::E_FAIL, + ); + + assert_eq!(backend.state.lock().unwrap().open_replies, vec![OpenReplyPdu::E_FAIL]); + assert_eq!(server.current_format(), None); + + // MS-RDPEAI 3.3.5.1.8: on failure the server MAY retry Open; the state machine must + // allow it rather than being stuck. + assert!(server.open(320, 0, fmt).is_ok()); +} + +#[test] +fn change_format_round_trips_and_notifies_backend() { + let fmt_a = pcm_format(1, 16000, 16); + let fmt_b = pcm_format(1, 48000, 16); + let backend = MockBackend::new(vec![fmt_a.clone(), fmt_b.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend.clone())); + + negotiate_and_open( + &mut server, + 1, + vec![fmt_a.clone(), fmt_b.clone()], + 320, + 0, + fmt_a, + OpenReplyPdu::S_OK, + ); + + let change_out = server.change_format(1).expect("change_format"); + assert_eq!(change_out.len(), 1); + assert!(matches!(decode_dvc(&change_out[0]), RdpeaiPdu::FormatChange(_))); + + let confirm_out = process_encoded(&mut server, 1, RdpeaiPdu::FormatChange(FormatChangePdu::new(1))); + assert!(confirm_out.is_empty()); + + assert_eq!(server.current_format(), Some(&fmt_b)); + assert_eq!(backend.state.lock().unwrap().format_changes, vec![fmt_b]); +} + +#[test] +fn change_format_rejects_out_of_range() { + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend)); + + negotiate_and_open(&mut server, 1, vec![fmt.clone()], 320, 0, fmt, OpenReplyPdu::S_OK); + + assert!(server.change_format(5).is_err()); +} + +#[test] +fn change_format_rejects_before_open() { + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend)); + + server.start(1).unwrap(); + let _ = process_encoded(&mut server, 1, RdpeaiPdu::Version(VersionPdu::new(Version::V1))); + let _ = process_encoded(&mut server, 1, RdpeaiPdu::Formats(FormatsPdu::client(vec![fmt]))); + + assert!(server.change_format(0).is_err()); +} + +#[test] +fn change_format_skips_when_active_format_is_aac_and_client_is_version_one() { + // MS-RDPEAI 3.3.5.3.1: if the client advertised only version 1 and the active format is + // AAC, the server SHOULD NOT send FormatChange. + let aac = AudioFormat { + format: WaveFormat::AAC_MS, + n_channels: 2, + n_samples_per_sec: 48000, + n_avg_bytes_per_sec: 12000, + n_block_align: 1, + bits_per_sample: 0, + data: None, + }; + let pcm = pcm_format(2, 48000, 16); + let backend = MockBackend::new(vec![aac.clone(), pcm.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend.clone())); + + negotiate_and_open( + &mut server, + 1, + vec![aac.clone(), pcm], + 320, + 0, // aac is index 0 + aac, + OpenReplyPdu::S_OK, + ); + assert_eq!(server.client_version(), Some(Version::V1)); + + let out = server + .change_format(1) + .expect("change_format should not error, only skip"); + assert!( + out.is_empty(), + "FormatChange must not be sent per 3.3.5.3.1's SHOULD NOT" + ); + assert!(backend.state.lock().unwrap().format_changes.is_empty()); +} + +#[test] +fn out_of_sequence_open_reply_before_open_is_ignored() { + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend.clone())); + + server.start(1).unwrap(); + let _ = process_encoded(&mut server, 1, RdpeaiPdu::Version(VersionPdu::new(Version::V1))); + let _ = process_encoded(&mut server, 1, RdpeaiPdu::Formats(FormatsPdu::client(vec![fmt]))); + + let out = process_encoded(&mut server, 1, RdpeaiPdu::OpenReply(OpenReplyPdu::ok())); + assert!(out.is_empty()); + assert!(backend.state.lock().unwrap().open_replies.is_empty()); +} + +#[test] +fn client_originated_open_pdu_is_ignored() { + // Open is server-to-client only; a client sending one is non-conformant and must not + // be treated as a valid transition (MS-RDPEAI 3.1.5: unrecognized/out-of-sequence + // packets MUST be ignored). + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend)); + + server.start(1).unwrap(); + let out = process_encoded( + &mut server, + 1, + RdpeaiPdu::Open(OpenPdu { + frames_per_packet: 320, + initial_format: 0, + capture_format: fmt, + }), + ); + assert!(out.is_empty()); +} + +#[test] +fn close_resets_state_and_notifies_backend() { + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend.clone())); + + negotiate_and_open(&mut server, 1, vec![fmt.clone()], 320, 0, fmt, OpenReplyPdu::S_OK); + assert!(server.current_format().is_some()); + + server.close(1); + + assert!(backend.state.lock().unwrap().closed); + assert!(server.current_format().is_none()); + assert!(server.negotiated_formats().is_empty()); + // Opened state was reset to Start, so Open must be rejected again until re-negotiated. + assert!(server.open(320, 0, pcm_format(1, 16000, 16)).is_err()); +} From 41179855f181e54f1fde79faab2ac9db8b0d5406 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Thu, 10 Sep 2026 22:21:43 -0500 Subject: [PATCH 2/2] fix(rdpeai): ignore malformed PDUs and divergent FormatChange echoes process() propagated a decode failure as an Err through the shared DRDYNVC SVC processor instead of ignoring it, contradicting the module's own MS-RDPEAI 3.1.5 MUST-ignore documentation and risking the whole dynamic-channel loop on a single malformed AUDIO_INPUT PDU. It now warns and returns an empty message list, matching every other ignore case in the same function. handle_format_change also trusted the client's echoed format index even when it did not match what the server requested, warning but still applying it. A non-conformant confirm now keeps the server-requested index instead, since an out-of-range echo silently drops all subsequent Data PDUs and an in-range-but-wrong one mislabels the format. Also drops a channel_id field that was set but never read (integrators recover the id via DrdynvcServer::get_channel_id_by_type, as the stacked wiring PR already does), folds the start()/close() reset sequence into a shared helper, and trims three constant-valued parameters plus a stale doc comment from the negotiate_and_open test helper. --- crates/ironrdp-rdpeai/src/server.rs | 62 ++++++---- .../tests/rdpeai/server.rs | 115 +++++++++--------- 2 files changed, 97 insertions(+), 80 deletions(-) diff --git a/crates/ironrdp-rdpeai/src/server.rs b/crates/ironrdp-rdpeai/src/server.rs index b1d136a74b..11f250cfbc 100644 --- a/crates/ironrdp-rdpeai/src/server.rs +++ b/crates/ironrdp-rdpeai/src/server.rs @@ -2,7 +2,7 @@ use ironrdp_core::{decode, impl_as_any}; use ironrdp_dvc::{DvcMessage, DvcProcessor, DvcServerProcessor}; -use ironrdp_pdu::{PduResult, decode_err, pdu_other_err}; +use ironrdp_pdu::{PduResult, pdu_other_err}; use ironrdp_rdpsnd::pdu::{AudioFormat, WaveFormat}; use tracing::{debug, trace, warn}; @@ -91,14 +91,14 @@ enum State { /// [`Self::open`] is called. Recording itself is caller-initiated: call [`Self::open`] once /// the negotiated formats are known ([`RdpeaiServerBackend::on_formats_negotiated`]) and the /// consuming application actually wants to record, then forward the returned PDUs to the -/// client over the `channel_id` this processor was started with. +/// client over the channel id this processor was created for (recover it with +/// [`ironrdp_dvc::DrdynvcServer::get_channel_id_by_type`]). /// /// Malformed, unrecognized, or out-of-sequence PDUs are logged and ignored rather than /// treated as errors, per MS-RDPEAI 3.1.5's explicit MUST-ignore requirement. pub struct RdpeaiServer { backend: Box, state: State, - channel_id: Option, client_version: Option, /// Negotiated format list, in the client's reply order — Open/FormatChange indices refer /// here (MS-RDPEAI 3.1.1: this is the list the client establishes as final). @@ -117,7 +117,6 @@ impl RdpeaiServer { Self { backend, state: State::Start, - channel_id: None, client_version: None, negotiated_formats: Vec::new(), current_format: None, @@ -126,6 +125,19 @@ impl RdpeaiServer { } } + /// Clear all per-connection state, transitioning to `state`. Shared by + /// [`DvcProcessor::start`] (transitions to `AwaitingClientVersion`) and + /// [`DvcProcessor::close`] (transitions to `Start`), so the two reset sequences can't + /// drift apart as fields are added. + fn reset(&mut self, state: State) { + self.state = state; + self.client_version = None; + self.negotiated_formats.clear(); + self.current_format = None; + self.pending_open_format = None; + self.pending_format_change = None; + } + /// Formats negotiated with the client (MS-RDPEAI 3.3.5.1.5); empty before that completes /// or if no common format exists. pub fn negotiated_formats(&self) -> &[AudioFormat] { @@ -267,13 +279,16 @@ impl RdpeaiServer { return Ok(Vec::new()); }; if pdu.new_format != pending { + // MS-RDPEAI 3.1.5: a non-conformant confirm MUST be ignored, not trusted. + // Keep the server-requested index rather than the client's divergent echo. warn!( expected = pending, got = pdu.new_format, - "Initial AUDIO_INPUT FormatChange confirm does not match the format from Open" + "Initial AUDIO_INPUT FormatChange confirm does not match the format from Open; \ + keeping the requested format" ); } - self.current_format = Some(pdu.new_format); + self.current_format = Some(pending); self.pending_open_format = None; self.state = State::AwaitingOpenReply; Ok(Vec::new()) @@ -285,13 +300,16 @@ impl RdpeaiServer { return Ok(Vec::new()); }; if pdu.new_format != pending { + // MS-RDPEAI 3.1.5: a non-conformant confirm MUST be ignored, not trusted. + // Keep the server-requested index rather than the client's divergent echo. warn!( expected = pending, got = pdu.new_format, - "AUDIO_INPUT FormatChange confirm does not match the requested format" + "AUDIO_INPUT FormatChange confirm does not match the requested format; \ + keeping the requested format" ); } - self.current_format = Some(pdu.new_format); + self.current_format = Some(pending); if let Some(fmt) = self.current_format().cloned() { self.backend.on_format_change_confirmed(&fmt); } @@ -349,20 +367,24 @@ impl DvcProcessor for RdpeaiServer { } fn start(&mut self, channel_id: u32) -> PduResult> { - self.channel_id = Some(channel_id); - self.state = State::AwaitingClientVersion; - self.client_version = None; - self.negotiated_formats.clear(); - self.current_format = None; - self.pending_open_format = None; - self.pending_format_change = None; + self.reset(State::AwaitingClientVersion); debug!(channel_id, "AUDIO_INPUT channel started"); // MS-RDPEAI 3.3.5.1.1: the Version PDU MUST be the first PDU sent by the server. Ok(vec![Box::new(RdpeaiPdu::Version(VersionPdu::new(Version::V2)))]) } fn process(&mut self, _channel_id: u32, payload: &[u8]) -> PduResult> { - let pdu: RdpeaiPdu = decode(payload).map_err(|e| decode_err!(e))?; + // MS-RDPEAI 3.1.5: malformed or unrecognized PDUs MUST be ignored rather than + // treated as errors. A decode failure must not propagate: this processor is + // reached through the shared DRDYNVC SVC processor, so an Err here would fail + // the whole dynamic-channel message loop, not just this channel. + let pdu: RdpeaiPdu = match decode(payload) { + Ok(pdu) => pdu, + Err(e) => { + warn!(error = %e, "Ignoring malformed AUDIO_INPUT PDU"); + return Ok(Vec::new()); + } + }; trace!(?pdu, "AUDIO_INPUT PDU received"); match pdu { RdpeaiPdu::Version(v) => self.handle_version(v), @@ -381,13 +403,7 @@ impl DvcProcessor for RdpeaiServer { fn close(&mut self, _channel_id: u32) { self.backend.on_close(); - self.channel_id = None; - self.state = State::Start; - self.client_version = None; - self.negotiated_formats.clear(); - self.current_format = None; - self.pending_open_format = None; - self.pending_format_change = None; + self.reset(State::Start); debug!("AUDIO_INPUT channel closed"); } } diff --git a/crates/ironrdp-testsuite-core/tests/rdpeai/server.rs b/crates/ironrdp-testsuite-core/tests/rdpeai/server.rs index ec07797f7f..0b2f4bff36 100644 --- a/crates/ironrdp-testsuite-core/tests/rdpeai/server.rs +++ b/crates/ironrdp-testsuite-core/tests/rdpeai/server.rs @@ -69,50 +69,34 @@ fn process_encoded(server: &mut RdpeaiServer, channel_id: u32, pdu: RdpeaiPdu) - .expect("process") } -/// Drive the server through Version -> Formats -> Open -> FormatChange confirm -> OpenReply, -/// returning the backend so callers can inspect what it observed. +/// Drive the server through Version -> Formats -> Open -> FormatChange confirm -> OpenReply. +/// Every call site uses the same channel id, frames-per-packet, and initial format index, so +/// those are fixed here rather than threaded through as parameters. fn negotiate_and_open( server: &mut RdpeaiServer, - channel_id: u32, client_formats: Vec, - frames_per_packet: u32, - initial_format: u32, capture_format: AudioFormat, open_result: i32, ) { - let start_out = server.start(channel_id).expect("start"); + let start_out = server.start(1).expect("start"); assert_eq!(start_out.len(), 1); assert!(matches!(decode_dvc(&start_out[0]), RdpeaiPdu::Version(_))); - let version_out = process_encoded(server, channel_id, RdpeaiPdu::Version(VersionPdu::new(Version::V1))); + let version_out = process_encoded(server, 1, RdpeaiPdu::Version(VersionPdu::new(Version::V1))); assert_eq!(version_out.len(), 1); assert!(matches!(decode_dvc(&version_out[0]), RdpeaiPdu::Formats(_))); - let formats_out = process_encoded( - server, - channel_id, - RdpeaiPdu::Formats(FormatsPdu::client(client_formats)), - ); + let formats_out = process_encoded(server, 1, RdpeaiPdu::Formats(FormatsPdu::client(client_formats))); assert!(formats_out.is_empty()); - let open_out = server - .open(frames_per_packet, initial_format, capture_format) - .expect("open"); + let open_out = server.open(320, 0, capture_format).expect("open"); assert_eq!(open_out.len(), 1); assert!(matches!(decode_dvc(&open_out[0]), RdpeaiPdu::Open(_))); - let confirm_out = process_encoded( - server, - channel_id, - RdpeaiPdu::FormatChange(FormatChangePdu::new(initial_format)), - ); + let confirm_out = process_encoded(server, 1, RdpeaiPdu::FormatChange(FormatChangePdu::new(0))); assert!(confirm_out.is_empty()); - let reply_out = process_encoded( - server, - channel_id, - RdpeaiPdu::OpenReply(OpenReplyPdu { result: open_result }), - ); + let reply_out = process_encoded(server, 1, RdpeaiPdu::OpenReply(OpenReplyPdu { result: open_result })); assert!(reply_out.is_empty()); } @@ -173,15 +157,7 @@ fn open_through_data_delivers_audio_to_the_backend() { let backend = MockBackend::new(vec![fmt.clone()]); let mut server = RdpeaiServer::new(Box::new(backend.clone())); - negotiate_and_open( - &mut server, - 1, - vec![fmt.clone()], - 320, - 0, - fmt.clone(), - OpenReplyPdu::S_OK, - ); + negotiate_and_open(&mut server, vec![fmt.clone()], fmt.clone(), OpenReplyPdu::S_OK); assert_eq!(backend.state.lock().unwrap().open_replies, vec![OpenReplyPdu::S_OK]); assert_eq!(server.current_format(), Some(&fmt)); @@ -240,15 +216,7 @@ fn open_reply_failure_returns_to_ready_and_allows_retry() { let backend = MockBackend::new(vec![fmt.clone()]); let mut server = RdpeaiServer::new(Box::new(backend.clone())); - negotiate_and_open( - &mut server, - 1, - vec![fmt.clone()], - 320, - 0, - fmt.clone(), - OpenReplyPdu::E_FAIL, - ); + negotiate_and_open(&mut server, vec![fmt.clone()], fmt.clone(), OpenReplyPdu::E_FAIL); assert_eq!(backend.state.lock().unwrap().open_replies, vec![OpenReplyPdu::E_FAIL]); assert_eq!(server.current_format(), None); @@ -267,10 +235,7 @@ fn change_format_round_trips_and_notifies_backend() { negotiate_and_open( &mut server, - 1, vec![fmt_a.clone(), fmt_b.clone()], - 320, - 0, fmt_a, OpenReplyPdu::S_OK, ); @@ -286,13 +251,40 @@ fn change_format_round_trips_and_notifies_backend() { assert_eq!(backend.state.lock().unwrap().format_changes, vec![fmt_b]); } +#[test] +fn change_format_confirm_ignores_divergent_echo() { + // MS-RDPEAI 3.1.5: a confirm that echoes a different index than the server requested is + // non-conformant and MUST be ignored, not trusted, keeping the server-requested format. + let fmt_a = pcm_format(1, 16000, 16); + let fmt_b = pcm_format(1, 48000, 16); + let backend = MockBackend::new(vec![fmt_a.clone(), fmt_b.clone()]); + let mut server = RdpeaiServer::new(Box::new(backend.clone())); + + negotiate_and_open( + &mut server, + vec![fmt_a.clone(), fmt_b.clone()], + fmt_a, + OpenReplyPdu::S_OK, + ); + + let change_out = server.change_format(1).expect("change_format"); + assert_eq!(change_out.len(), 1); + + // Client echoes index 0 instead of the requested index 1. + let confirm_out = process_encoded(&mut server, 1, RdpeaiPdu::FormatChange(FormatChangePdu::new(0))); + assert!(confirm_out.is_empty()); + + assert_eq!(server.current_format(), Some(&fmt_b)); + assert_eq!(backend.state.lock().unwrap().format_changes, vec![fmt_b]); +} + #[test] fn change_format_rejects_out_of_range() { let fmt = pcm_format(1, 16000, 16); let backend = MockBackend::new(vec![fmt.clone()]); let mut server = RdpeaiServer::new(Box::new(backend)); - negotiate_and_open(&mut server, 1, vec![fmt.clone()], 320, 0, fmt, OpenReplyPdu::S_OK); + negotiate_and_open(&mut server, vec![fmt.clone()], fmt, OpenReplyPdu::S_OK); assert!(server.change_format(5).is_err()); } @@ -327,15 +319,8 @@ fn change_format_skips_when_active_format_is_aac_and_client_is_version_one() { let backend = MockBackend::new(vec![aac.clone(), pcm.clone()]); let mut server = RdpeaiServer::new(Box::new(backend.clone())); - negotiate_and_open( - &mut server, - 1, - vec![aac.clone(), pcm], - 320, - 0, // aac is index 0 - aac, - OpenReplyPdu::S_OK, - ); + // aac is index 0 + negotiate_and_open(&mut server, vec![aac.clone(), pcm], aac, OpenReplyPdu::S_OK); assert_eq!(server.client_version(), Some(Version::V1)); let out = server @@ -385,13 +370,29 @@ fn client_originated_open_pdu_is_ignored() { assert!(out.is_empty()); } +#[test] +fn malformed_pdu_is_ignored_not_errored() { + // MS-RDPEAI 3.1.5: malformed or unrecognized PDUs MUST be ignored. A decode failure + // must not surface as an Err, since process() is reached through the shared DRDYNVC + // SVC processor and an Err there would fail the whole dynamic-channel message loop. + let fmt = pcm_format(1, 16000, 16); + let backend = MockBackend::new(vec![fmt]); + let mut server = RdpeaiServer::new(Box::new(backend)); + + server.start(1).unwrap(); + let out = server + .process(1, &[0xFF]) + .expect("malformed PDU must be ignored, not errored"); + assert!(out.is_empty()); +} + #[test] fn close_resets_state_and_notifies_backend() { let fmt = pcm_format(1, 16000, 16); let backend = MockBackend::new(vec![fmt.clone()]); let mut server = RdpeaiServer::new(Box::new(backend.clone())); - negotiate_and_open(&mut server, 1, vec![fmt.clone()], 320, 0, fmt, OpenReplyPdu::S_OK); + negotiate_and_open(&mut server, vec![fmt.clone()], fmt, OpenReplyPdu::S_OK); assert!(server.current_format().is_some()); server.close(1);