fix(server): drop unexpected PDUs during deactivation-reactivation

ARCHITECTURE.md

@@ -209,8 +209,6 @@ This is to keep iteration time short.
 
 Contains all integration tests for code living in the extra tier, in a single binary, organized in modules.
 
-(WIP: this crate does not exist yet.)
-
 #### [`crates/ironrdp-fuzzing`](./crates/ironrdp-fuzzing)
 
 Provides test case generators and oracles for use with fuzzing.

Cargo.toml

@@ -52,6 +52,7 @@ ironrdp-session-generators = { path = "crates/ironrdp-session-generators" }
 ironrdp-session = { version = "0.2", path = "crates/ironrdp-session" }
 ironrdp-svc = { version = "0.1", path = "crates/ironrdp-svc" }
 ironrdp-testsuite-core = { path = "crates/ironrdp-testsuite-core" }
+ironrdp-testsuite-extra = { path = "crates/ironrdp-testsuite-extra" }
 ironrdp-tls = { version = "0.1", path = "crates/ironrdp-tls" }
 ironrdp-tokio = { version = "0.2", path = "crates/ironrdp-tokio" }
 ironrdp = { version = "0.7", path = "crates/ironrdp" }

crates/ironrdp-acceptor/src/connection.rs

@@ -1,4 +1,3 @@
-use core::any::TypeId;
 use core::mem;
 
 use ironrdp_connector::{
@@ -34,6 +33,7 @@ pub struct Acceptor {
     static_channels: StaticChannelSet,
     saved_for_reactivation: AcceptorState,
     pub(crate) creds: Option<Credentials>,
+    reactivation: bool,
 }
 
 #[derive(Debug)]
@@ -43,6 +43,7 @@ pub struct AcceptorResult {
     pub input_events: Vec<Vec<u8>>,
     pub user_channel_id: u16,
     pub io_channel_id: u16,
+    pub reactivation: bool,
 }
 
 impl Acceptor {
@@ -62,10 +63,15 @@ impl Acceptor {
             static_channels: StaticChannelSet::new(),
             saved_for_reactivation: Default::default(),
             creds,
+            reactivation: false,
         }
     }
 
-    pub fn new_deactivation_reactivation(mut consumed: Acceptor, desktop_size: DesktopSize) -> Self {
+    pub fn new_deactivation_reactivation(
+        mut consumed: Acceptor,
+        static_channels: StaticChannelSet,
+        desktop_size: DesktopSize,
+    ) -> Self {
         let AcceptorState::CapabilitiesSendServer {
             early_capability,
             channels,
@@ -95,28 +101,18 @@ impl Acceptor {
             io_channel_id: consumed.io_channel_id,
             desktop_size,
             server_capabilities: consumed.server_capabilities,
-            static_channels: StaticChannelSet::new(),
+            static_channels,
             saved_for_reactivation,
             creds: consumed.creds,
+            reactivation: true,
         }
     }
 
     pub fn attach_static_channel<T>(&mut self, channel: T)
     where
         T: SvcServerProcessor + 'static,
     {
-        let channel_name = channel.channel_name();
-
         self.static_channels.insert(channel);
-
-        // Restore channel id if it was already attached.
-        if let AcceptorState::CapabilitiesSendServer { channels, .. } = &self.state {
-            for (channel_id, c) in channels {
-                if c.name == channel_name {
-                    self.static_channels.attach_channel_id(TypeId::of::<T>(), *channel_id);
-                }
-            }
-        }
     }
 
     pub fn reached_security_upgrade(&self) -> Option<SecurityProtocol> {
@@ -155,6 +151,7 @@ impl Acceptor {
                 input_events,
                 user_channel_id: self.user_channel_id,
                 io_channel_id: self.io_channel_id,
+                reactivation: self.reactivation,
             }),
             previous_state => {
                 self.state = previous_state;
@@ -291,7 +288,9 @@ impl Sequence for Acceptor {
     }
 
     fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult<Written> {
-        let (written, next_state) = match mem::take(&mut self.state) {
+        let prev_state = mem::take(&mut self.state);
+
+        let (written, next_state) = match prev_state {
             AcceptorState::InitiationWaitRequest => {
                 let connection_request = decode::<X224<nego::ConnectionRequest>>(input)
                     .map_err(ConnectorError::decode)
@@ -315,6 +314,8 @@ impl Sequence for Acceptor {
                     SecurityProtocol::HYBRID
                 } else if protocols.intersects(SecurityProtocol::SSL) {
                     SecurityProtocol::SSL
+                } else if self.security.is_empty() {
+                    SecurityProtocol::empty()
                 } else {
                     return Err(ConnectorError::general("failed to negotiate security protocol"));
                 };
@@ -639,15 +640,38 @@ impl Sequence for Acceptor {
                 )
             }
 
-            AcceptorState::CapabilitiesWaitConfirm { channels } => {
+            AcceptorState::CapabilitiesWaitConfirm { ref channels } => {
                 let message = decode::<X224<mcs::McsMessage<'_>>>(input)
                     .map_err(ConnectorError::decode)
-                    .map(|p| p.0)?;
-
+                    .map(|p| p.0);
+                let message = match message {
+                    Ok(msg) => msg,
+                    Err(e) => {
+                        if self.reactivation {
+                            debug!("Dropping unexpected PDU during reactivation");
+                            self.state = prev_state;
+                            return Ok(Written::Nothing);
+                        } else {
+                            return Err(e);
+                        }
+                    }
+                };
                 match message {
                     mcs::McsMessage::SendDataRequest(data) => {
                         let capabilities_confirm = decode::<rdp::headers::ShareControlHeader>(data.user_data.as_ref())
-                            .map_err(ConnectorError::decode)?;
+                            .map_err(ConnectorError::decode);
+                        let capabilities_confirm = match capabilities_confirm {
+                            Ok(capabilities_confirm) => capabilities_confirm,
+                            Err(e) => {
+                                if self.reactivation {
+                                    debug!("Dropping unexpected PDU during reactivation");
+                                    self.state = prev_state;
+                                    return Ok(Written::Nothing);
+                                } else {
+                                    return Err(e);
+                                }
+                            }
+                        };
 
                         debug!(message = ?capabilities_confirm, "Received");
 
@@ -659,7 +683,7 @@ impl Sequence for Acceptor {
                         (
                             Written::Nothing,
                             AcceptorState::ConnectionFinalization {
-                                channels,
+                                channels: channels.clone(),
                                 finalization: FinalizationSequence::new(self.user_channel_id, self.io_channel_id),
                                 client_capabilities: confirm.pdu.capability_sets,
                             },
@@ -673,7 +697,7 @@ impl Sequence for Acceptor {
                     _ => {
                         warn!(?message, "Unexpected MCS message received");
 
-                        (Written::Nothing, AcceptorState::CapabilitiesWaitConfirm { channels })
+                        (Written::Nothing, prev_state)
                     }
                 }
             }
@@ -684,6 +708,7 @@ impl Sequence for Acceptor {
                 client_capabilities,
             } => {
                 let written = finalization.step(input, output)?;
+
                 let state = if finalization.is_done() {
                     AcceptorState::Accepted {
                         channels,

crates/ironrdp-acceptor/src/lib.rs

@@ -4,7 +4,6 @@
 #[macro_use]
 extern crate tracing;
 
-use ironrdp_async::bytes::Bytes;
 use ironrdp_async::{single_sequence_step, Framed, FramedRead, FramedWrite, StreamWrapper};
 use ironrdp_connector::credssp::KerberosConfig;
 use ironrdp_connector::sspi::credssp::EarlyUserAuthResult;
@@ -50,7 +49,7 @@ where
             return Ok(result);
         }
 
-        single_sequence_step(&mut framed, acceptor, &mut buf, None).await?;
+        single_sequence_step(&mut framed, acceptor, &mut buf).await?;
     }
 }
 
@@ -84,7 +83,6 @@ where
 pub async fn accept_finalize<S>(
     mut framed: Framed<S>,
     acceptor: &mut Acceptor,
-    mut unmatched: Option<&mut Vec<Bytes>>,
 ) -> ConnectorResult<(Framed<S>, AcceptorResult)>
 where
     S: FramedRead + FramedWrite,
@@ -95,7 +93,7 @@ where
         if let Some(result) = acceptor.get_result() {
             return Ok((framed, result));
         }
-        single_sequence_step(&mut framed, acceptor, &mut buf, unmatched.as_deref_mut()).await?;
+        single_sequence_step(&mut framed, acceptor, &mut buf).await?;
     }
 }
 
@@ -152,7 +150,7 @@ where
             );
 
             let pdu = framed
-                .read_by_hint(next_pdu_hint, None)
+                .read_by_hint(next_pdu_hint)
                 .await
                 .map_err(|e| ironrdp_connector::custom_err!("read frame by hint", e))?;
 

crates/ironrdp-async/src/connector.rs

@@ -23,7 +23,7 @@ where
     info!("Begin connection procedure");
 
     while !connector.should_perform_security_upgrade() {
-        single_sequence_step(framed, connector, &mut buf, None).await?;
+        single_sequence_step(framed, connector, &mut buf).await?;
     }
 
     Ok(ShouldUpgrade)
@@ -73,7 +73,7 @@ where
     }
 
     let result = loop {
-        single_sequence_step(framed, &mut connector, &mut buf, None).await?;
+        single_sequence_step(framed, &mut connector, &mut buf).await?;
 
         if let ClientConnectorState::Connected { result } = connector.state {
             break result;
@@ -171,7 +171,7 @@ where
         );
 
         let pdu = framed
-            .read_by_hint(next_pdu_hint, None)
+            .read_by_hint(next_pdu_hint)
             .await
             .map_err(|e| ironrdp_connector::custom_err!("read frame by hint", e))?;
 

crates/ironrdp-async/src/framed.rs

@@ -165,11 +165,7 @@ where
     /// `tokio::select!` statement and some other branch
     /// completes first, then it is safe to drop the future and re-create it later.
     /// Data may have been read, but it will be stored in the internal buffer.
-    pub async fn read_by_hint(
-        &mut self,
-        hint: &dyn PduHint,
-        mut unmatched: Option<&mut Vec<Bytes>>,
-    ) -> io::Result<Bytes> {
+    pub async fn read_by_hint(&mut self, hint: &dyn PduHint) -> io::Result<Bytes> {
         loop {
             match hint
                 .find_size(self.peek())
@@ -179,10 +175,8 @@ where
                     let bytes = self.read_exact(length).await?.freeze();
                     if matched {
                         return Ok(bytes);
-                    } else if let Some(ref mut unmatched) = unmatched {
-                        unmatched.push(bytes);
                     } else {
-                        warn!("Received and lost an unexpected PDU");
+                        debug!("Received and lost an unexpected PDU");
                     }
                 }
                 None => {
@@ -236,21 +230,19 @@ pub async fn single_sequence_step<S>(
     framed: &mut Framed<S>,
     sequence: &mut dyn Sequence,
     buf: &mut WriteBuf,
-    unmatched: Option<&mut Vec<Bytes>>,
 ) -> ConnectorResult<()>
 where
     S: FramedWrite + FramedRead,
 {
     buf.clear();
-    let written = single_sequence_step_read(framed, sequence, buf, unmatched).await?;
+    let written = single_sequence_step_read(framed, sequence, buf).await?;
     single_sequence_step_write(framed, buf, written).await
 }
 
 pub async fn single_sequence_step_read<S>(
     framed: &mut Framed<S>,
     sequence: &mut dyn Sequence,
     buf: &mut WriteBuf,
-    unmatched: Option<&mut Vec<Bytes>>,
 ) -> ConnectorResult<Written>
 where
     S: FramedRead,
@@ -265,7 +257,7 @@ where
         );
 
         let pdu = framed
-            .read_by_hint(next_pdu_hint, unmatched)
+            .read_by_hint(next_pdu_hint)
             .await
             .map_err(|e| ironrdp_connector::custom_err!("read frame by hint", e))?;
 

crates/ironrdp-blocking/src/connector.rs

@@ -1,6 +1,5 @@
 use std::io::{Read, Write};
 
-use bytes::Bytes;
 use ironrdp_connector::credssp::{CredsspProcessGenerator, CredsspSequence, KerberosConfig};
 use ironrdp_connector::sspi::credssp::ClientState;
 use ironrdp_connector::sspi::generator::GeneratorState;
@@ -26,7 +25,7 @@ where
     info!("Begin connection procedure");
 
     while !connector.should_perform_security_upgrade() {
-        single_sequence_step(framed, connector, &mut buf, None)?;
+        single_sequence_step(framed, connector, &mut buf)?;
     }
 
     Ok(ShouldUpgrade)
@@ -79,7 +78,7 @@ where
     debug!("Remaining of connection sequence");
 
     let result = loop {
-        single_sequence_step(framed, &mut connector, &mut buf, None)?;
+        single_sequence_step(framed, &mut connector, &mut buf)?;
 
         if let ClientConnectorState::Connected { result } = connector.state {
             break result;
@@ -168,7 +167,7 @@ where
         );
 
         let pdu = framed
-            .read_by_hint(next_pdu_hint, None)
+            .read_by_hint(next_pdu_hint)
             .map_err(|e| ironrdp_connector::custom_err!("read frame by hint", e))?;
 
         trace!(length = pdu.len(), "PDU received");
@@ -189,7 +188,6 @@ pub fn single_sequence_step<S>(
     framed: &mut Framed<S>,
     connector: &mut ClientConnector,
     buf: &mut WriteBuf,
-    unmatched: Option<&mut Vec<Bytes>>,
 ) -> ConnectorResult<()>
 where
     S: Read + Write,
@@ -204,7 +202,7 @@ where
         );
 
         let pdu = framed
-            .read_by_hint(next_pdu_hint, unmatched)
+            .read_by_hint(next_pdu_hint)
             .map_err(|e| ironrdp_connector::custom_err!("read frame by hint", e))?;
 
         trace!(length = pdu.len(), "PDU received");