add tests and fixes

Cargo.lock

@@ -451,6 +451,7 @@ dependencies = [
 name = "cluelesshd"
 version = "0.1.0"
 dependencies = [
+ "cluelessh-format",
  "cluelessh-keys",
  "cluelessh-protocol",
  "cluelessh-tokio",

bin/cluelesshd/Cargo.toml

@@ -4,6 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+cluelessh-format = { path = "../../lib/cluelessh-format" }
 cluelessh-protocol = { path = "../../lib/cluelessh-protocol" }
 cluelessh-tokio = { path = "../../lib/cluelessh-tokio" }
 cluelessh-transport = { path = "../../lib/cluelessh-transport" }

bin/cluelesshd/src/main.rs

@@ -1,7 +1,13 @@
 mod auth;
 mod pty;
 
-use std::{io, net::SocketAddr, process::ExitStatus, sync::Arc};
+use std::{
+    io,
+    net::SocketAddr,
+    pin::Pin,
+    process::{ExitStatus, Stdio},
+    sync::Arc,
+};
 
 use auth::AuthError;
 use cluelessh_keys::{private::EncryptedPrivateKeys, public::PublicKey};
@@ -12,7 +18,7 @@ use pty::Pty;
 use rustix::termios::Winsize;
 use tokio::{
     fs::File,
-    io::{AsyncReadExt, AsyncWriteExt},
+    io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
     net::{TcpListener, TcpStream},
     process::Command,
     sync::mpsc,
@@ -224,8 +230,12 @@ struct SessionState {
 
     envs: Vec<(String, String)>,
 
-    writer: Option<File>,
+    //// stdin
-    reader: Option<File>,
+    writer: Option<Pin<Box<dyn AsyncWrite + Send + Sync>>>,
+    /// stdout
+    reader: Option<Pin<Box<dyn AsyncRead + Send + Sync>>>,
+    /// stderr
+    reader_ext: Option<Pin<Box<dyn AsyncRead + Send + Sync>>>,
 }
 
 async fn handle_session_channel(user: String, channel: Channel) -> Result<()> {
@@ -240,9 +250,11 @@ async fn handle_session_channel(user: String, channel: Channel) -> Result<()> {
         envs: Vec::new(),
         writer: None,
         reader: None,
+        reader_ext: None,
     };
 
     let mut read_buf = [0; 1024];
+    let mut read_ext_buf = [0; 1024];
 
     loop {
         let read = async {
@@ -254,6 +266,15 @@ async fn handle_session_channel(user: String, channel: Channel) -> Result<()> {
                 },
             }
         };
+        let read_ext = async {
+            match &mut state.reader_ext {
+                Some(file) => file.read(&mut read_ext_buf).await,
+                // Ensure that if this is None, the future never finishes so the state update and process exit can progress.
+                None => loop {
+                    tokio::task::yield_now().await;
+                },
+            }
+        };
         tokio::select! {
             update = state.channel.next_update() => {
                 match update {
@@ -281,6 +302,12 @@ async fn handle_session_channel(user: String, channel: Channel) -> Result<()> {
                 };
                 let _ = state.channel.send(ChannelOperationKind::Data(read_buf[..read].to_vec())).await;
             }
+            read = read_ext => {
+                let Ok(read) = read else {
+                    bail!("failed to read");
+                };
+                let _ = state.channel.send(ChannelOperationKind::ExtendedData(1, read_ext_buf[..read].to_vec())).await;
+            }
         }
     }
 }
@@ -325,7 +352,7 @@ impl SessionState {
                             }
                         }
                     }
-                    ChannelRequest::Shell { want_reply } => match self.shell().await {
+                    ChannelRequest::Shell { want_reply } => match self.shell(None).await {
                         Ok(()) => {
                             if want_reply {
                                 self.channel.send(ChannelOperationKind::Success).await?;
@@ -338,9 +365,31 @@ impl SessionState {
                             }
                         }
                     },
-                    ChannelRequest::Exec { .. } => {
+                    ChannelRequest::Exec {
-                        todo!()
+                        want_reply,
+                        command,
+                    } => match String::from_utf8(command) {
+                        Ok(command) => match self.shell(Some(&command)).await {
+                            Ok(()) => {
+                                if want_reply {
+                                    self.channel.send(ChannelOperationKind::Success).await?;
                                 }
+                            }
+                            Err(err) => {
+                                debug!(%err, "Failed to spawn shell");
+                                if want_reply {
+                                    self.channel.send(ChannelOperationKind::Failure).await?;
+                                }
+                            }
+                        },
+                        Err(err) => {
+                            debug!(%err, "Exec command is invalid UTF-8");
+
+                            if want_reply {
+                                self.channel.send(ChannelOperationKind::Failure).await?;
+                            }
+                        }
+                    },
                     ChannelRequest::Env {
                         name,
                         value,
@@ -368,10 +417,15 @@ impl SessionState {
                     writer.write_all(&data).await?;
                 }
             }
+            ChannelUpdateKind::Eof => {
+                if let Some(writer) = &mut self.writer {
+                    writer.shutdown().await?;
+                }
+                self.writer = None;
+            }
             ChannelUpdateKind::Open(_)
             | ChannelUpdateKind::Closed
             | ChannelUpdateKind::ExtendedData { .. }
-            | ChannelUpdateKind::Eof
             | ChannelUpdateKind::Success
             | ChannelUpdateKind::Failure => { /* ignore */ }
         }
@@ -383,12 +437,14 @@ impl SessionState {
         let controller = pty.controller().try_clone_to_owned()?;
 
         self.pty = Some(pty);
-        self.writer = Some(File::from_std(std::fs::File::from(controller.try_clone()?)));
+        self.writer = Some(Box::pin(File::from_std(std::fs::File::from(
-        self.reader = Some(File::from_std(std::fs::File::from(controller)));
+            controller.try_clone()?,
+        ))));
+        self.reader = Some(Box::pin(File::from_std(std::fs::File::from(controller))));
         Ok(())
     }
 
-    async fn shell(&mut self) -> Result<()> {
+    async fn shell(&mut self, shell_command: Option<&str>) -> Result<()> {
         let user = self.user.clone();
         let user = tokio::task::spawn_blocking(move || users::get_user_by_name(&user))
             .await?
@@ -397,10 +453,18 @@ impl SessionState {
         let shell = user.shell();
 
         let mut cmd = Command::new(shell);
+        if let Some(shell_command) = shell_command {
+            cmd.arg("-c");
+            cmd.arg(shell_command);
+        }
         cmd.env_clear();
 
         if let Some(pty) = &self.pty {
             pty.start_session_for_command(&mut cmd)?;
+        } else {
+            cmd.stdin(Stdio::piped());
+            cmd.stdout(Stdio::piped());
+            cmd.stderr(Stdio::piped());
         }
 
         // TODO: **user** home directory
@@ -417,6 +481,16 @@ impl SessionState {
 
         let mut shell = cmd.spawn()?;
 
+        if self.pty.is_none() {
+            let stdin = shell.stdin.take().unwrap();
+            let stdout = shell.stdout.take().unwrap();
+            let stderr = shell.stderr.take().unwrap();
+
+            self.writer = Some(Box::pin(stdin));
+            self.reader = Some(Box::pin(stdout));
+            self.reader_ext = Some(Box::pin(stderr));
+        }
+
         let process_exit_send = self.process_exit_send.clone();
         tokio::spawn(async move {
             let result = shell.wait().await;

bin/cluelesshd/tests/openssh-client/command.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+ssh -p "$PORT" "$HOST" echo jdklfsjdöklfd | grep "jdklfsjdöklfd"

bin/cluelesshd/tests/openssh-client/different-algorithms.sh

@@ -1,12 +1,12 @@
 #!/usr/bin/env bash
 
 # KEX
-printf $"exit\r" | ssh -oKexAlgorithms=curve25519-sha256 -p "$PORT" "$HOST"
+ssh -oKexAlgorithms=curve25519-sha256 -p "$PORT" "$HOST" true
-printf $"exit\r" | ssh -oKexAlgorithms=ecdh-sha2-nistp256 -p "$PORT" "$HOST"
+ssh -oKexAlgorithms=ecdh-sha2-nistp256 -p "$PORT" "$HOST" true
 
 # Encryption
-printf $"exit\r" | ssh -oCiphers=chacha20-poly1305@openssh.com -p "$PORT" "$HOST"
+ssh -oCiphers=chacha20-poly1305@openssh.com -p "$PORT" "$HOST" true
-printf $"exit\r" | ssh -oCiphers=aes256-gcm@openssh.com -p "$PORT" "$HOST"
+ssh -oCiphers=aes256-gcm@openssh.com -p "$PORT" "$HOST" true
 
 # Host Key
-printf $"exit\r" | ssh -oHostKeyAlgorithms=ssh-ed25519 -p "$PORT" "$HOST"
+ssh -oHostKeyAlgorithms=ssh-ed25519 -p "$PORT" "$HOST" true

bin/cluelesshd/tests/openssh-client/interactive-no-pty.sh

@@ -1,6 +1,3 @@
 #!/usr/bin/env bash
 
-# disabled, TODO
+printf $"echo jdklfsjdöklfd" | ssh -p "$PORT" "$HOST" | grep jdklfsjdöklfd
-exit 0
-
-echo | ssh -p "$PORT" "$HOST"

bin/cluelesshd/tests/run.sh

@@ -19,7 +19,30 @@ kill_server() {
 
 trap kill_server EXIT
 
+failures=()
+
+export PORT=2223
+export HOST=localhost
+
 for script in "$script_dir"/openssh-client/*.sh; do
-    echo "-------------- Running $script"
+
-    PORT=2223 HOST=localhost bash -euo pipefail "$script"
+    echo "-------------- Running PORT=$PORT HOST=$HOST bash $script"
+
+    set +e
+    bash -euo pipefail "$script"
+    result=$?
+    set -e
+    if [ "$result" -ne "0" ]; then
+        echo "Test $script failed!"
+
+        failures+=("$script")
+    fi 
 done
+
+if (( ${#failures[@]} )); then
+    echo "FAILED"
+    for failure in "${failures[@]}"; do
+        echo " failed: PORT=$PORT HOST=$HOST bash $failure"
+    done
+    exit 1
+fi

lib/cluelessh-connection/src/lib.rs

@@ -58,7 +58,8 @@ struct Channel {
 
     /// Queued data that we want to send, but have not been able to because of the window limits.
     /// Whenever we get more window space, we will send this data.
-    queued_data: Vec<u8>,
+    queued_data_default: Vec<u8>,
+    queued_data_extended: HashMap<u32, Vec<u8>>,
 }
 
 /// An update from a channel.
@@ -133,6 +134,7 @@ pub enum ChannelOperationKind {
     Success,
     Failure,
     Data(Vec<u8>),
+    ExtendedData(u32, Vec<u8>),
     Request(ChannelRequest),
     Eof,
     Close,
@@ -212,7 +214,8 @@ impl ChannelsState {
                         our_window_size: initial_window_size,
                         our_window_size_increase_step: initial_window_size,
 
-                        queued_data: Vec::new(),
+                        queued_data_default: Vec::new(),
+                        queued_data_extended: HashMap::new(),
                     }),
                 );
 
@@ -255,7 +258,8 @@ impl ChannelsState {
                         our_window_size,
                         our_window_size_increase_step: our_window_size,
 
-                        queued_data: Vec::new(),
+                        queued_data_default: Vec::new(),
+                        queued_data_extended: HashMap::new(),
                     }),
                 );
 
@@ -297,11 +301,38 @@ impl ChannelsState {
                     .checked_add(bytes_to_add)
                     .ok_or_else(|| peer_error!("window size larger than 2^32"))?;
 
-                if !channel.queued_data.is_empty() {
+                if !channel.queued_data_default.is_empty() {
-                    let limit =
+                    let limit = cmp::min(
-                        cmp::min(channel.queued_data.len(), channel.peer_window_size as usize);
+                        channel.queued_data_default.len(),
-                    let data_to_send = channel.queued_data.splice(..limit, []).collect::<Vec<_>>();
+                        channel.peer_window_size as usize,
-                    self.send_data(our_channel, &data_to_send);
+                    );
+                    let data_to_send = channel
+                        .queued_data_default
+                        .splice(..limit, [])
+                        .collect::<Vec<_>>();
+                    self.send_data(our_channel, &data_to_send, None);
+                }
+
+                // After potentially sending default data, see if we can send some extended data too.
+                let channel = self.channel(our_channel)?;
+                let data_keys = channel
+                    .queued_data_extended
+                    .keys()
+                    .copied()
+                    .collect::<Vec<_>>();
+
+                for number in data_keys {
+                    let channel = self.channel(our_channel)?;
+                    let peer_window_size = channel.peer_window_size;
+                    let queued_data_extended =
+                        channel.queued_data_extended.get_mut(&number).unwrap();
+
+                    if !queued_data_extended.is_empty() {
+                        let limit = cmp::min(queued_data_extended.len(), peer_window_size as usize);
+                        let data_to_send =
+                            queued_data_extended.splice(..limit, []).collect::<Vec<_>>();
+                        self.send_data(our_channel, &data_to_send, Some(number));
+                    }
                 }
             }
             numbers::SSH_MSG_CHANNEL_DATA => {
@@ -569,7 +600,10 @@ impl ChannelsState {
             ChannelOperationKind::Success => self.send_channel_success(peer),
             ChannelOperationKind::Failure => self.send_channel_failure(peer),
             ChannelOperationKind::Data(data) => {
-                self.send_data(op.number, &data);
+                self.send_data(op.number, &data, None);
+            }
+            ChannelOperationKind::ExtendedData(code, data) => {
+                self.send_data(op.number, &data, Some(code));
             }
             ChannelOperationKind::Request(req) => {
                 let packet = match req {
@@ -623,7 +657,12 @@ impl ChannelsState {
         }
     }
 
-    fn send_data(&mut self, channel_number: ChannelNumber, data: &[u8]) {
+    fn send_data(
+        &mut self,
+        channel_number: ChannelNumber,
+        data: &[u8],
+        extended_code: Option<u32>,
+    ) {
         let channel = self.channel(channel_number).unwrap();
 
         let mut chunks = data.chunks(channel.peer_max_packet_size as usize);
@@ -647,11 +686,24 @@ impl ChannelsState {
                     // It's over, we have exhausted all window space.
                     // Queue the rest of the bytes.
                     let channel = self.channel(channel_number).unwrap();
-                    channel.queued_data.extend_from_slice(to_keep);
+                    match extended_code {
+                        Some(extended) => {
+                            let queued_data_extended =
+                                channel.queued_data_extended.entry(extended).or_default();
+                            queued_data_extended.extend_from_slice(to_keep);
                             for data in chunks {
-                        channel.queued_data.extend_from_slice(data);
+                                queued_data_extended.extend_from_slice(data);
+                            }
+                            debug!(channel = %channel_number, queue_len = %channel.queued_data_extended.len(), "Exhausted window space, queueing the rest of the data");
+                        }
+                        None => {
+                            channel.queued_data_default.extend_from_slice(to_keep);
+                            for data in chunks {
+                                channel.queued_data_default.extend_from_slice(data);
+                            }
+                            debug!(channel = %channel_number, queue_len = %channel.queued_data_default.len(), "Exhausted window space, queueing the rest of the data");
+                        }
                     }
-                    debug!(channel = %channel_number, queue_len = %channel.queued_data.len(), "Exhausted window space, queueing the rest of the data");
                     return;
                 }
                 Some(space) => channel.peer_window_size = space,
@@ -713,6 +765,7 @@ impl ChannelOperation {
             ChannelOperationKind::Success => "success",
             ChannelOperationKind::Failure => "failure",
             ChannelOperationKind::Data(_) => "data",
+            ChannelOperationKind::ExtendedData(_, _) => "extended-data",
             ChannelOperationKind::Request(req) => match req {
                 ChannelRequest::PtyReq { .. } => "pty-req",
                 ChannelRequest::Shell { .. } => "shell",
@@ -824,6 +877,7 @@ mod tests {
         assert_response_types(state, &[]);
     }
 
+    // TODO: test with extended data
     #[test]
     fn respect_peer_windowing() {
         let state = &mut ChannelsState::new(true);

testing.rs

@@ -0,0 +1,12 @@
+use std::process::{Command, Stdio};
+
+fn main() {
+    let mut cmd = Command::new("fish");
+    cmd.stderr(Stdio::piped());
+    cmd.stdout(Stdio::piped());
+    cmd.stdin(Stdio::piped());
+
+    let mut child = cmd.spawn().unwrap();
+
+    child.wait().unwrap();
+}