feat: Add Send bounds to AsyncNetworkClient (fixes #542)

crates/ironrdp-async/src/connector.rs

@@ -8,7 +8,7 @@ use ironrdp_connector::{
 use ironrdp_core::WriteBuf;
 
 use crate::framed::{Framed, FramedRead, FramedWrite};
-use crate::{single_sequence_step, AsyncNetworkClient};
+use crate::{single_sequence_step, AsyncNetworkClient, WasmAsyncNetworkClient};
 
 #[non_exhaustive]
 pub struct ShouldUpgrade;
@@ -85,6 +85,47 @@ where
     Ok(result)
 }
 
+#[instrument(skip_all)]
+pub async fn wasm_connect_finalize<S>(
+    _: Upgraded,
+    framed: &mut Framed<S>,
+    mut connector: ClientConnector,
+    server_name: ServerName,
+    server_public_key: Vec<u8>,
+    network_client: Option<&mut dyn WasmAsyncNetworkClient>,
+    kerberos_config: Option<KerberosConfig>,
+) -> ConnectorResult<ConnectionResult>
+where
+    S: FramedRead + FramedWrite,
+{
+    let mut buf = WriteBuf::new();
+
+    if connector.should_perform_credssp() {
+        wasm_perform_credssp_step(
+            framed,
+            &mut connector,
+            &mut buf,
+            server_name,
+            server_public_key,
+            network_client,
+            kerberos_config,
+        )
+        .await?;
+    }
+
+    let result = loop {
+        single_sequence_step(framed, &mut connector, &mut buf).await?;
+
+        if let ClientConnectorState::Connected { result } = connector.state {
+            break result;
+        }
+    };
+
+    info!("Connected with success");
+
+    Ok(result)
+}
+
 async fn resolve_generator(
     generator: &mut CredsspProcessGenerator<'_>,
     network_client: &mut dyn AsyncNetworkClient,
@@ -105,6 +146,26 @@ async fn resolve_generator(
     }
 }
 
+async fn wasm_resolve_generator(
+    generator: &mut CredsspProcessGenerator<'_>,
+    network_client: &mut dyn WasmAsyncNetworkClient,
+) -> ConnectorResult<ClientState> {
+    let mut state = generator.start();
+
+    loop {
+        match state {
+            GeneratorState::Suspended(request) => {
+                let response = network_client.send(&request).await?;
+                state = generator.resume(Ok(response));
+            }
+            GeneratorState::Completed(client_state) => {
+                break client_state
+                    .map_err(|e| ConnectorError::new("CredSSP", ironrdp_connector::ConnectorErrorKind::Credssp(e)))
+            }
+        }
+    }
+}
+
 #[instrument(level = "trace", skip_all)]
 async fn perform_credssp_step<S>(
     framed: &mut Framed<S>,
@@ -188,3 +249,87 @@ where
 
     Ok(())
 }
+
+#[instrument(level = "trace", skip_all)]
+async fn wasm_perform_credssp_step<S>(
+    framed: &mut Framed<S>,
+    connector: &mut ClientConnector,
+    buf: &mut WriteBuf,
+    server_name: ServerName,
+    server_public_key: Vec<u8>,
+    mut network_client: Option<&mut dyn WasmAsyncNetworkClient>,
+    kerberos_config: Option<KerberosConfig>,
+) -> ConnectorResult<()>
+where
+    S: FramedRead + FramedWrite,
+{
+    assert!(connector.should_perform_credssp());
+
+    let selected_protocol = match connector.state {
+        ClientConnectorState::Credssp { selected_protocol, .. } => selected_protocol,
+        _ => return Err(general_err!("invalid connector state for CredSSP sequence")),
+    };
+
+    let (mut sequence, mut ts_request) = CredsspSequence::init(
+        connector.config.credentials.clone(),
+        connector.config.domain.as_deref(),
+        selected_protocol,
+        server_name,
+        server_public_key,
+        kerberos_config,
+    )?;
+
+    loop {
+        let client_state = {
+            let mut generator = sequence.process_ts_request(ts_request);
+
+            if let Some(network_client_ref) = network_client.as_deref_mut() {
+                trace!("resolving network");
+                wasm_resolve_generator(&mut generator, network_client_ref).await?
+            } else {
+                generator
+                    .resolve_to_result()
+                    .map_err(|e| custom_err!("resolve without network client", e))?
+            }
+        }; // drop generator
+
+        buf.clear();
+        let written = sequence.handle_process_result(client_state, buf)?;
+
+        if let Some(response_len) = written.size() {
+            let response = &buf[..response_len];
+            trace!(response_len, "Send response");
+            framed
+                .write_all(response)
+                .await
+                .map_err(|e| ironrdp_connector::custom_err!("write all", e))?;
+        }
+
+        let Some(next_pdu_hint) = sequence.next_pdu_hint() else {
+            break;
+        };
+
+        debug!(
+            connector.state = connector.state.name(),
+            hint = ?next_pdu_hint,
+            "Wait for PDU"
+        );
+
+        let pdu = framed
+            .read_by_hint(next_pdu_hint)
+            .await
+            .map_err(|e| ironrdp_connector::custom_err!("read frame by hint", e))?;
+
+        trace!(length = pdu.len(), "PDU received");
+
+        if let Some(next_request) = sequence.decode_server_message(&pdu)? {
+            ts_request = next_request;
+        } else {
+            break;
+        }
+    }
+
+    connector.mark_credssp_as_done();
+
+    Ok(())
+}

crates/ironrdp-async/src/lib.rs

@@ -21,7 +21,14 @@ pub use self::connector::*;
 pub use self::framed::*;
 // pub use self::session::*;
 
-pub trait AsyncNetworkClient {
+pub trait AsyncNetworkClient: Send {
+    fn send<'a>(
+        &'a mut self,
+        network_request: &'a NetworkRequest,
+    ) -> Pin<Box<dyn Future<Output = ConnectorResult<Vec<u8>>> + Send + 'a>>;
+}
+
+pub trait WasmAsyncNetworkClient {
     fn send<'a>(
         &'a mut self,
         network_request: &'a NetworkRequest,

crates/ironrdp-tokio/src/reqwest.rs

@@ -19,7 +19,7 @@ impl AsyncNetworkClient for ReqwestNetworkClient {
     fn send<'a>(
         &'a mut self,
         request: &'a sspi::generator::NetworkRequest,
-    ) -> Pin<Box<dyn Future<Output = ConnectorResult<Vec<u8>>> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ConnectorResult<Vec<u8>>> + Send + 'a>> {
         Box::pin(async move {
             match &request.protocol {
                 sspi::network_client::NetworkProtocol::Tcp => self.send_tcp(&request.url, &request.data).await,

crates/ironrdp-web/src/network_client.rs

@@ -4,12 +4,12 @@ use futures_util::Future;
 use ironrdp::connector::sspi::generator::NetworkRequest;
 use ironrdp::connector::sspi::network_client::NetworkProtocol;
 use ironrdp::connector::{custom_err, reason_err, ConnectorResult};
-use ironrdp_futures::AsyncNetworkClient;
+use ironrdp_futures::WasmAsyncNetworkClient;
 
 #[derive(Debug)]
 pub(crate) struct WasmNetworkClient;
 
-impl AsyncNetworkClient for WasmNetworkClient {
+impl WasmAsyncNetworkClient for WasmNetworkClient {
     fn send<'a>(
         &'a mut self,
         network_request: &'a NetworkRequest,

crates/ironrdp-web/src/session.rs

@@ -940,7 +940,7 @@ async fn connect(
     let (upgraded, server_public_key) =
         connect_rdcleanpath(&mut framed, &mut connector, destination.clone(), proxy_auth_token, pcb).await?;
 
-    let connection_result = ironrdp_futures::connect_finalize(
+    let connection_result = ironrdp_futures::wasm_connect_finalize(
         upgraded,
         &mut framed,
         connector,