.

distant-net/src/common/authentication/handler.rs

@@ -107,11 +107,11 @@ where
 
             // Get an answer from user input, or use a blank string as an answer
             // if we fail to get input from the user
-            let answer = self.password_prompt(line).unwrap_or_default();
+            let answer = (self.password_prompt)(line).unwrap_or_default();
 
             answers.push(answer);
         }
-        Ok(answers)
+        Ok(ChallengeResponse { answers })
     }
 
     async fn on_verification(
@@ -123,13 +123,15 @@ where
             VerificationKind::Host => {
                 eprintln!("{}", verification.text);
 
-                let answer = self.text_prompt("Enter [y/N]> ")?;
+                let answer = (self.text_prompt)("Enter [y/N]> ")?;
                 trace!("Verify? Answer = '{answer}'");
-                Ok(matches!(answer.trim(), "y" | "Y" | "yes" | "YES"))
+                Ok(VerificationResponse {
+                    valid: matches!(answer.trim(), "y" | "Y" | "yes" | "YES"),
+                })
             }
             x => {
                 error!("Unsupported verify kind: {x}");
-                Ok(false)
+                Ok(VerificationResponse { valid: false })
             }
         }
     }

distant-net/src/manager/data.rs

@@ -1,4 +1,5 @@
 pub type ManagerChannelId = u32;
+pub type ManagerAuthenticationId = u32;
 
 mod capabilities;
 pub use capabilities::*;

distant-net/src/manager/data/request.rs

@@ -1,4 +1,4 @@
-use super::ManagerChannelId;
+use super::{ManagerAuthenticationId, ManagerChannelId};
 use crate::common::{authentication::msg::AuthenticationResponse, ConnectionId, Destination, Map};
 use derive_more::IsVariant;
 use serde::{Deserialize, Serialize};
@@ -55,7 +55,7 @@ pub enum ManagerRequest {
     #[strum_discriminants(strum(message = "Supports authenticating with a remote server"))]
     Authenticate {
         /// Id of the authentication request that is being responded to
-        id: String,
+        id: ManagerAuthenticationId,
 
         /// Response being sent to some active connection
         msg: AuthenticationResponse,

distant-net/src/manager/data/response.rs

@@ -1,4 +1,7 @@
-use super::{ConnectionInfo, ConnectionList, Error, ManagerCapabilities, ManagerChannelId};
+use super::{
+    ConnectionInfo, ConnectionList, Error, ManagerAuthenticationId, ManagerCapabilities,
+    ManagerChannelId,
+};
 use crate::common::{authentication::msg::Authentication, ConnectionId, Destination};
 use serde::{Deserialize, Serialize};
 
@@ -29,7 +32,7 @@ pub enum ManagerResponse {
     /// Authentication information being sent to a client
     Authenticate {
         /// Id tied to authentication information in case a response is needed
-        id: String,
+        id: ManagerAuthenticationId,
 
         /// Authentication message
         msg: Authentication,

distant-net/src/manager/server.rs

@@ -1,11 +1,12 @@
 use crate::{
     client::Client,
     common::{
-        authentication::msg::AuthenticationResponse, Listener, Map, MpscListener, Request, Response,
+        authentication::msg::AuthenticationResponse, ConnectionId, Destination, Listener, Map,
+        MpscListener, Request, Response,
     },
     manager::{
-        ChannelId, ConnectionId, ConnectionInfo, ConnectionList, Destination, ManagerCapabilities,
-        ManagerRequest, ManagerResponse,
+        ConnectionInfo, ConnectionList, ManagerAuthenticationId, ManagerCapabilities,
+        ManagerChannelId, ManagerRequest, ManagerResponse,
     },
     server::{Server, ServerCtx, ServerHandler},
 };
@@ -17,6 +18,9 @@ use tokio::{
     task::JoinHandle,
 };
 
+mod authentication;
+pub use authentication::*;
+
 mod config;
 pub use config::*;
 
@@ -35,7 +39,8 @@ pub struct ManagerServer {
     connections: RwLock<HashMap<ConnectionId, ManagerConnection>>,
 
     /// Mapping of auth id -> callback
-    authentication: RwLock<HashMap<String, oneshot::Sender<AuthenticationResponse>>>,
+    registry:
+        Arc<RwLock<HashMap<ManagerAuthenticationId, oneshot::Sender<AuthenticationResponse>>>>,
 }
 
 impl ManagerServer {
@@ -46,7 +51,7 @@ impl ManagerServer {
         Server::new().handler(Self {
             config,
             connections: RwLock::new(HashMap::new()),
-            authentication: RwLock::new(HashMap::new()),
+            registry: Arc::new(RwLock::new(HashMap::new())),
         })
     }
 
@@ -54,7 +59,12 @@ impl ManagerServer {
     /// and authentication client (if needed) to retrieve additional information needed to
     /// enter the destination prior to starting the server, returning the destination of the
     /// launched server
-    async fn launch(&self, destination: Destination, options: Map) -> io::Result<Destination> {
+    async fn launch(
+        &self,
+        destination: Destination,
+        options: Map,
+        mut authenticator: ManagerAuthenticator,
+    ) -> io::Result<Destination> {
         let scheme = match destination.scheme.as_deref() {
             Some(scheme) => {
                 trace!("Using scheme {}", scheme);
@@ -77,7 +87,9 @@ impl ManagerServer {
                     format!("No launch handler registered for {}", scheme),
                 )
             })?;
-            handler.launch(&destination, &options, auth).await?
+            handler
+                .launch(&destination, &options, &mut authenticator)
+                .await?
         };
 
         Ok(credentials)
@@ -86,7 +98,12 @@ impl ManagerServer {
     /// Connects to a new server at the specified `destination` using the given `options` information
     /// and authentication client (if needed) to retrieve additional information needed to
     /// establish the connection to the server
-    async fn connect(&self, destination: Destination, options: Map) -> io::Result<ConnectionId> {
+    async fn connect(
+        &self,
+        destination: Destination,
+        options: Map,
+        mut authenticator: ManagerAuthenticator,
+    ) -> io::Result<ConnectionId> {
         let scheme = match destination.scheme.as_deref() {
             Some(scheme) => {
                 trace!("Using scheme {}", scheme);
@@ -109,7 +126,9 @@ impl ManagerServer {
                     format!("No connect handler registered for {}", scheme),
                 )
             })?;
-            handler.connect(&destination, &options, auth).await?
+            handler
+                .connect(&destination, &options, &mut authenticator)
+                .await?
         };
 
         let connection = ManagerConnection::new(destination, options, transport);
@@ -191,25 +210,45 @@ impl ServerHandler for ManagerServer {
             ManagerRequest::Launch {
                 destination,
                 options,
-            } => match self.launch(*destination, options).await {
+            } => match self
+                .launch(
+                    *destination,
+                    options,
+                    ManagerAuthenticator {
+                        reply,
+                        registry: Arc::clone(&self.registry),
+                    },
+                )
+                .await
+            {
                 Ok(destination) => ManagerResponse::Launched { destination },
                 Err(x) => ManagerResponse::Error(x.into()),
             },
             ManagerRequest::Connect {
                 destination,
                 options,
-            } => match self.connect(*destination, options).await {
+            } => match self
+                .connect(
+                    *destination,
+                    options,
+                    ManagerAuthenticator {
+                        reply,
+                        registry: Arc::clone(&self.registry),
+                    },
+                )
+                .await
+            {
                 Ok(id) => ManagerResponse::Connected { id },
                 Err(x) => ManagerResponse::Error(x.into()),
             },
             ManagerRequest::Authenticate { id, msg } => {
-                match self.authentication.write().await.remove(&id) {
+                match self.registry.write().await.remove(&id) {
                     Some(cb) => match cb.send(msg) {
                         Ok(_) => ManagerResponse::Ok,
                         Err(x) => ManagerResponse::Error(x.into()),
                     },
                     None => ManagerResponse::Error(
-                        io::Error::new(io::ErrorKind::InvalidInput, "Invalid authenticate id")
+                        io::Error::new(io::ErrorKind::InvalidInput, "Invalid authentication id")
                             .into(),
                     ),
                 }
@@ -283,65 +322,48 @@ impl ServerHandler for ManagerServer {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{
-        common::{FramedTransport, InmemoryTransport, MappedListener, OneshotListener, PlainCodec},
-        server::ServerRef,
-    };
-
-    /// Create a new server, bypassing the start loop
-    fn setup() -> ManagerServer {
-        ManagerServer {
-            config: Default::default(),
-            connections: RwLock::new(HashMap::new()),
-            launch_handlers: Arc::new(RwLock::new(HashMap::new())),
-            connect_handlers: Arc::new(RwLock::new(HashMap::new())),
-            task: tokio::spawn(async move {}),
+    use crate::server::ServerReply;
+
+    fn test_config() -> Config {
+        Config {
+            launch_fallback_scheme: "ssh".to_string(),
+            connect_fallback_scheme: "distant".to_string(),
+            connection_buffer_size: 100,
+            user: false,
+            launch_handlers: HashMap::new(),
+            connect_handlers: HashMap::new(),
         }
     }
 
-    /// Creates a connected [`AuthClient`] with a launched auth server that blindly responds
-    fn auth_client_server() -> (AuthClient, Box<dyn ServerRef>) {
-        let (t1, t2) = FramedTransport::pair(1);
-        let client = AuthClient::from(Client::from_framed_transport(t1).unwrap());
-
-        // Create a server that does nothing, but will support
-        let server = HeapAuthServer {
-            on_challenge: Box::new(|_, _| Vec::new()),
-            on_verify: Box::new(|_, _| false),
-            on_info: Box::new(|_| ()),
-            on_error: Box::new(|_, _| ()),
-        }
-        .start(MappedListener::new(OneshotListener::from_value(t2), |t| {
-            t.into_split()
-        }))
-        .unwrap();
+    /// Create a new server and authenticator
+    fn setup(config: Config) -> (ManagerServer, ManagerAuthenticator) {
+        let registry = Arc::new(RwLock::new(HashMap::new()));
 
-        (client, server)
-    }
+        let authenticator = ManagerAuthenticator {
+            reply: ServerReply {
+                origin_id: format!("{}", rand::random::<u8>()),
+                tx: mpsc::channel(1).0,
+            },
+            registry: Arc::clone(&registry),
+        };
 
-    fn dummy_distant_writer_reader() -> (BoxedDistantWriter, BoxedDistantReader) {
-        setup_distant_writer_reader().0
-    }
+        let server = ManagerServer {
+            config,
+            connections: RwLock::new(HashMap::new()),
+            registry,
+        };
 
-    /// Creates a writer & reader with a connected transport
-    fn setup_distant_writer_reader() -> (
-        (BoxedDistantWriter, BoxedDistantReader),
-        FramedTransport<InmemoryTransport, PlainCodec>,
-    ) {
-        let (t1, t2) = FramedTransport::pair(1);
-        let (writer, reader) = t1.into_split();
-        ((Box::new(writer), Box::new(reader)), t2)
+        (server, authenticator)
     }
 
     #[tokio::test]
     async fn launch_should_fail_if_destination_scheme_is_unsupported() {
-        let server = setup();
+        let (server, authenticator) = setup(test_config());
 
         let destination = "scheme://host".parse::<Destination>().unwrap();
         let options = "".parse::<Map>().unwrap();
-        let (mut auth, _auth_server) = auth_client_server();
         let err = server
-            .launch(destination, options, Some(&mut auth))
+            .launch(destination, options, authenticator)
             .await
             .unwrap_err();
         assert_eq!(err.kind(), io::ErrorKind::InvalidInput, "{:?}", err);
@@ -349,23 +371,19 @@ mod tests {
 
     #[tokio::test]
     async fn launch_should_fail_if_handler_tied_to_scheme_fails() {
-        let server = setup();
+        let mut config = test_config();
 
         let handler: Box<dyn LaunchHandler> = Box::new(|_: &_, _: &_, _: &mut _| async {
             Err(io::Error::new(io::ErrorKind::Other, "test failure"))
         });
 
-        server
-            .launch_handlers
-            .write()
-            .await
-            .insert("scheme".to_string(), handler);
+        config.launch_handlers.insert("scheme".to_string(), handler);
 
+        let (server, authenticator) = setup(config);
         let destination = "scheme://host".parse::<Destination>().unwrap();
         let options = "".parse::<Map>().unwrap();
-        let (mut auth, _auth_server) = auth_client_server();
         let err = server
-            .launch(destination, options, Some(&mut auth))
+            .launch(destination, options, authenticator)
             .await
             .unwrap_err();
         assert_eq!(err.kind(), io::ErrorKind::Other);
@@ -374,7 +392,7 @@ mod tests {
 
     #[tokio::test]
     async fn launch_should_return_new_destination_on_success() {
-        let server = setup();
+        let mut config = test_config();
 
         let handler: Box<dyn LaunchHandler> = {
             Box::new(|_: &_, _: &_, _: &mut _| async {
@@ -382,17 +400,13 @@ mod tests {
             })
         };
 
-        server
-            .launch_handlers
-            .write()
-            .await
-            .insert("scheme".to_string(), handler);
+        config.launch_handlers.insert("scheme".to_string(), handler);
 
+        let (server, authenticator) = setup(config);
         let destination = "scheme://host".parse::<Destination>().unwrap();
         let options = "key=value".parse::<Map>().unwrap();
-        let (mut auth, _auth_server) = auth_client_server();
         let destination = server
-            .launch(destination, options, Some(&mut auth))
+            .launch(destination, options, authenticator)
             .await
             .unwrap();
 
@@ -404,13 +418,12 @@ mod tests {
 
     #[tokio::test]
     async fn connect_should_fail_if_destination_scheme_is_unsupported() {
-        let server = setup();
+        let (server, authenticator) = setup(test_config());
 
         let destination = "scheme://host".parse::<Destination>().unwrap();
         let options = "".parse::<Map>().unwrap();
-        let (mut auth, _auth_server) = auth_client_server();
         let err = server
-            .connect(destination, options, Some(&mut auth))
+            .connect(destination, options, authenticator)
             .await
             .unwrap_err();
         assert_eq!(err.kind(), io::ErrorKind::InvalidInput, "{:?}", err);
@@ -418,23 +431,21 @@ mod tests {
 
     #[tokio::test]
     async fn connect_should_fail_if_handler_tied_to_scheme_fails() {
-        let server = setup();
+        let mut config = test_config();
 
         let handler: Box<dyn ConnectHandler> = Box::new(|_: &_, _: &_, _: &mut _| async {
             Err(io::Error::new(io::ErrorKind::Other, "test failure"))
         });
 
-        server
+        config
             .connect_handlers
-            .write()
-            .await
             .insert("scheme".to_string(), handler);
 
+        let (server, authenticator) = setup(config);
         let destination = "scheme://host".parse::<Destination>().unwrap();
         let options = "".parse::<Map>().unwrap();
-        let (mut auth, _auth_server) = auth_client_server();
         let err = server
-            .connect(destination, options, Some(&mut auth))
+            .connect(destination, options, authenticator)
             .await
             .unwrap_err();
         assert_eq!(err.kind(), io::ErrorKind::Other);
@@ -443,22 +454,20 @@ mod tests {
 
     #[tokio::test]
     async fn connect_should_return_id_of_new_connection_on_success() {
-        let server = setup();
+        let mut config = test_config();
 
         let handler: Box<dyn ConnectHandler> =
             Box::new(|_: &_, _: &_, _: &mut _| async { Ok(dummy_distant_writer_reader()) });
 
-        server
+        config
             .connect_handlers
-            .write()
-            .await
             .insert("scheme".to_string(), handler);
 
+        let (server, authenticator) = setup(config);
         let destination = "scheme://host".parse::<Destination>().unwrap();
         let options = "key=value".parse::<Map>().unwrap();
-        let (mut auth, _auth_server) = auth_client_server();
         let id = server
-            .connect(destination, options, Some(&mut auth))
+            .connect(destination, options, authenticator)
             .await
             .unwrap();
 
@@ -471,7 +480,7 @@ mod tests {
 
     #[tokio::test]
     async fn info_should_fail_if_no_connection_found_for_specified_id() {
-        let server = setup();
+        let (server, _) = setup(test_config());
 
         let err = server.info(999).await.unwrap_err();
         assert_eq!(err.kind(), io::ErrorKind::NotConnected, "{:?}", err);
@@ -479,7 +488,7 @@ mod tests {
 
     #[tokio::test]
     async fn info_should_return_information_about_established_connection() {
-        let server = setup();
+        let (server, _) = setup(test_config());
 
         let (writer, reader) = dummy_distant_writer_reader();
         let connection = ManagerConnection::new(
@@ -504,7 +513,7 @@ mod tests {
 
     #[tokio::test]
     async fn list_should_return_empty_connection_list_if_no_established_connections() {
-        let server = setup();
+        let (server, _) = setup(test_config());
 
         let list = server.list().await.unwrap();
         assert_eq!(list, ConnectionList(HashMap::new()));
@@ -512,7 +521,7 @@ mod tests {
 
     #[tokio::test]
     async fn list_should_return_a_list_of_established_connections() {
-        let server = setup();
+        let (server, _) = setup(test_config());
 
         let (writer, reader) = dummy_distant_writer_reader();
         let connection = ManagerConnection::new(
@@ -547,7 +556,7 @@ mod tests {
 
     #[tokio::test]
     async fn kill_should_fail_if_no_connection_found_for_specified_id() {
-        let server = setup();
+        let (server, _) = setup(test_config());
 
         let err = server.kill(999).await.unwrap_err();
         assert_eq!(err.kind(), io::ErrorKind::NotConnected, "{:?}", err);
@@ -555,7 +564,7 @@ mod tests {
 
     #[tokio::test]
     async fn kill_should_terminate_established_connection_and_remove_it_from_the_list() {
-        let server = setup();
+        let (server, _) = setup(test_config());
 
         let (writer, reader) = dummy_distant_writer_reader();
         let connection = ManagerConnection::new(

distant-net/src/manager/server/authentication.rs

@@ -0,0 +1,103 @@
+use crate::{
+    common::authentication::{msg::*, Authenticator},
+    manager::data::{ManagerAuthenticationId, ManagerResponse},
+    server::ServerReply,
+};
+use async_trait::async_trait;
+use std::{collections::HashMap, io, sync::Arc};
+use tokio::sync::{oneshot, RwLock};
+
+/// Implementation of [`Authenticator`] used by a manger to perform authentication with
+/// remote servers it is managing.
+#[derive(Clone)]
+pub struct ManagerAuthenticator {
+    /// Used to communicate authentication requests
+    pub(super) reply: ServerReply<ManagerResponse>,
+
+    /// Used to store one-way response senders that are used to return callbacks
+    pub(super) registry:
+        Arc<RwLock<HashMap<ManagerAuthenticationId, oneshot::Sender<AuthenticationResponse>>>>,
+}
+
+impl ManagerAuthenticator {
+    /// Sends an [`Authentication`] `msg` that expects a reply, storing a callback.
+    async fn send(&self, msg: Authentication) -> io::Result<AuthenticationResponse> {
+        let (tx, rx) = oneshot::channel();
+        let id = rand::random();
+
+        self.registry.write().await.insert(id, tx);
+        self.reply
+            .send(ManagerResponse::Authenticate { id, msg })
+            .await?;
+        rx.await
+            .map_err(|x| io::Error::new(io::ErrorKind::Other, x))
+    }
+
+    /// Sends an [`Authentication`] `msg` without expecting a reply. No callback is stored.
+    async fn fire(&self, msg: Authentication) -> io::Result<()> {
+        let id = rand::random();
+        self.reply
+            .send(ManagerResponse::Authenticate { id, msg })
+            .await?;
+        Ok(())
+    }
+}
+
+/// Represents an interface for submitting challenges for authentication.
+#[async_trait]
+impl Authenticator for ManagerAuthenticator {
+    async fn initialize(
+        &mut self,
+        initialization: Initialization,
+    ) -> io::Result<InitializationResponse> {
+        match self
+            .send(Authentication::Initialization(initialization))
+            .await
+        {
+            Ok(AuthenticationResponse::Initialization(x)) => Ok(x),
+            Ok(x) => Err(io::Error::new(
+                io::ErrorKind::Other,
+                "Unexpected response: {x:?}",
+            )),
+            Err(x) => Err(x),
+        }
+    }
+
+    async fn challenge(&mut self, challenge: Challenge) -> io::Result<ChallengeResponse> {
+        match self.send(Authentication::Challenge(challenge)).await {
+            Ok(AuthenticationResponse::Challenge(x)) => Ok(x),
+            Ok(x) => Err(io::Error::new(
+                io::ErrorKind::Other,
+                "Unexpected response: {x:?}",
+            )),
+            Err(x) => Err(x),
+        }
+    }
+
+    async fn verify(&mut self, verification: Verification) -> io::Result<VerificationResponse> {
+        match self.send(Authentication::Verification(verification)).await {
+            Ok(AuthenticationResponse::Verification(x)) => Ok(x),
+            Ok(x) => Err(io::Error::new(
+                io::ErrorKind::Other,
+                "Unexpected response: {x:?}",
+            )),
+            Err(x) => Err(x),
+        }
+    }
+
+    async fn info(&mut self, info: Info) -> io::Result<()> {
+        self.fire(Authentication::Info(info)).await
+    }
+
+    async fn error(&mut self, error: Error) -> io::Result<()> {
+        self.fire(Authentication::Error(error)).await
+    }
+
+    async fn start_method(&mut self, start_method: StartMethod) -> io::Result<()> {
+        self.fire(Authentication::StartMethod(start_method)).await
+    }
+
+    async fn finished(&mut self) -> io::Result<()> {
+        self.fire(Authentication::Finished).await
+    }
+}

distant-net/src/manager/server/connection.rs

@@ -4,7 +4,7 @@ use crate::{
         UntypedRequest, UntypedResponse,
     },
     manager::data::{ManagerChannelId, ManagerResponse},
-    server::{ServerRef, ServerReply},
+    server::ServerReply,
 };
 use log::*;
 use serde::Serialize;

distant-net/src/manager/server/handler.rs

@@ -53,21 +53,23 @@ pub trait ConnectHandler: Send + Sync {
         destination: &Destination,
         options: &Map,
         authenticator: &mut dyn Authenticator,
-    ) -> io::Result<FramedTransport<Box<dyn Transport>>>;
+    ) -> io::Result<FramedTransport<Box<dyn Transport + Send + Sync>>>;
 }
 
 #[async_trait]
 impl<F, R> ConnectHandler for F
 where
     F: for<'a> Fn(&'a Destination, &'a Map, &'a mut dyn Authenticator) -> R + Send + Sync + 'static,
-    R: Future<Output = io::Result<FramedTransport<Box<dyn Transport>>>> + Send + 'static,
+    R: Future<Output = io::Result<FramedTransport<Box<dyn Transport + Send + Sync>>>>
+        + Send
+        + 'static,
 {
     async fn connect(
         &self,
         destination: &Destination,
         options: &Map,
         authenticator: &mut dyn Authenticator,
-    ) -> io::Result<FramedTransport<Box<dyn Transport>>> {
+    ) -> io::Result<FramedTransport<Box<dyn Transport + Send + Sync>>> {
         self(destination, options, authenticator).await
     }
 }