feat: added the ability to run commands remotely

commands can now be sent and their outputs received

sparse-05/sparse-05-client/src/commands/connect/mod.rs

@@ -67,7 +67,7 @@ impl Connection {
             bail!("received packet from wrong computer");
         }
 
-        dbg!(self.decrypt_and_verify(&buffer[..read]))
+        self.decrypt_and_verify(&buffer[..read])
     }
 }
 

sparse-05/sparse-05-server/src/connection.rs

@@ -25,12 +25,13 @@ impl ConnectionHandle {
     }
 }
 
+#[derive(Clone)]
 struct ConnectionInformation {
     remote_sign_pubkey: ed25519_dalek::PublicKey,
     remote_enc_pubkey: ecies_ed25519::PublicKey,
-    local_sign_keypair: Keypair,
+    local_sign_keypair: Arc<Keypair>,
     local_enc_pubkey: ecies_ed25519::PublicKey,
-    local_enc_privkey: ecies_ed25519::SecretKey,
+    local_enc_privkey: Arc<ecies_ed25519::SecretKey>,
     srcmac: [u8; 6],
     dstmac: [u8; 6],
     srcip: Ipv4Addr,
@@ -131,8 +132,8 @@ pub fn spawn_connection_handler(
         ConnectionInformation {
             remote_sign_pubkey: (*sign_pubkey).clone(),
             remote_enc_pubkey: (*enc_pubkey).clone(),
-            local_sign_keypair,
+            local_sign_keypair: Arc::new(local_sign_keypair),
-            local_enc_privkey,
+            local_enc_privkey: Arc::new(local_enc_privkey),
             local_enc_pubkey,
             srcmac: *packet.source_address(),
             dstmac: *packet.destination_address(),
@@ -250,7 +251,7 @@ where
 
             match data {
                 Command::RunCommand(comm) => {
-                    let handler = match command::spawn_command(&s, comm, &conninfo) {
+                    let handler = match command::spawn_command(&s, comm, conninfo.clone()) {
                         Ok(handler) => handler,
                         Err(e) => {
                             eprintln!("error spawning command: {e:?}");

sparse-05/sparse-05-server/src/connection/command.rs

@@ -2,10 +2,11 @@ use std::{
     io::{Read, Write},
     process::{Command, Stdio},
     sync::{
-        atomic::AtomicU64,
+        atomic::{AtomicBool, AtomicU64, Ordering},
         mpsc::{channel, Sender},
+        Arc,
     },
-    thread::Scope,
+    thread::{scope, Scope},
 };
 
 use sparse_05_common::messages::Response;
@@ -22,7 +23,7 @@ pub(super) struct CommandHandler {
 pub(super) fn spawn_command<'a, 'b: 'a>(
     s: &'a Scope<'a, 'b>,
     command: String,
-    conninfo: &'a ConnectionInformation,
+    conninfo: ConnectionInformation,
 ) -> anyhow::Result<CommandHandler> {
     let (data_sender, data_receiver) = channel();
 
@@ -41,62 +42,120 @@ pub(super) fn spawn_command<'a, 'b: 'a>(
 
     let resp = Response::AckRunCommand(id);
     conninfo.send(conninfo.encrypt_and_sign_resp(resp)?)?;
+    let data_sender_2 = data_sender.clone();
 
     s.spawn(move || -> anyhow::Result<()> {
-        if let Some(mut stdout) = command.stdout.take() {
+        let should_die = AtomicBool::new(false);
-            s.spawn(move || {
+        scope(|s2| -> anyhow::Result<()> {
-                let mut stdout_buffer = [0u8; 1024];
+            let stdout_thread = if let Some(mut stdout) = command.stdout.take() {
+                Some(s2.spawn({
+                    let conninfo = conninfo.clone();
+                    let should_die = &should_die;
+                    move || {
+                        let mut stdout_buffer = [0u8; 1024];
 
-                loop {
+                        loop {
-                    let Ok(count) = stdout.read(&mut stdout_buffer) else {
+                            let Ok(count) = stdout.read(&mut stdout_buffer) else {
-                        continue;
+                                continue;
-                    };
+                            };
 
-                    let resp = Response::SendStdout(stdout_buffer[..count].to_vec(), id);
+                            if should_die.load(Ordering::Relaxed) {
-                    let Ok(packet) = conninfo.encrypt_and_sign_resp(resp) else {
+                                break;
-                        continue;
+                            }
-                    };
-                    _ = conninfo.send(packet);
-                }
-            });
-        }
 
-        if let Some(mut stderr) = command.stderr.take() {
+                            if count == 0 {
-            s.spawn(move || {
+                                continue;
-                let mut stderr_buffer = [0u8; 1024];
+                            }
 
-                loop {
+                            let resp = Response::SendStdout(stdout_buffer[..count].to_vec(), id);
-                    let Ok(count) = stderr.read(&mut stderr_buffer) else {
+                            let Ok(packet) = conninfo.encrypt_and_sign_resp(resp) else {
-                        continue;
+                                continue;
-                    };
+                            };
+                            _ = conninfo.send(packet);
+                        }
+                    }
+                }))
+            } else {
+                None
+            };
 
-                    let resp = Response::SendStderr(stderr_buffer[..count].to_vec(), id);
+            let stderr_thread = if let Some(mut stderr) = command.stderr.take() {
-                    let Ok(packet) = conninfo.encrypt_and_sign_resp(resp) else {
+                Some(s2.spawn({
-                        continue;
+                    let conninfo = conninfo.clone();
-                    };
+                    let should_die = &should_die;
-                    _ = conninfo.send(packet);
+                    move || {
-                }
+                        let mut stderr_buffer = [0u8; 1024];
-            });
-        }
 
-        if let Some(mut stdin) = command.stdin.take() {
+                        loop {
-            s.spawn(move || loop {
+                            let Ok(count) = stderr.read(&mut stderr_buffer) else {
-                let Ok(input): Result<Vec<u8>, _> = data_receiver.recv() else {
+                                continue;
-                    continue;
+                            };
-                };
 
-                _ = stdin.write(&input);
+                            if should_die.load(Ordering::Relaxed) {
-            });
+                                break;
-        }
+                            }
 
-        let code = match command.wait() {
+                            if count == 0 {
-            Ok(status) => status.code().unwrap_or(-1),
+                                continue;
-            Err(_) => -1,
+                            }
-        };
+
-        let resp = Response::CommandDone(id, code);
+                            let resp = Response::SendStderr(stderr_buffer[..count].to_vec(), id);
-        let resp = conninfo.encrypt_and_sign_resp(resp)?;
+                            let Ok(packet) = conninfo.encrypt_and_sign_resp(resp) else {
-        conninfo.send(resp)?;
+                                continue;
-        Ok(())
+                            };
+                            _ = conninfo.send(packet);
+                        }
+                    }
+                }))
+            } else {
+                None
+            };
+
+            let stdin_thread = if let Some(mut stdin) = command.stdin.take() {
+                Some(s2.spawn({
+                    let should_die = &should_die;
+                    move || loop {
+                        let Ok(input): Result<Vec<u8>, _> =
+                            data_receiver.recv_timeout(std::time::Duration::from_millis(250))
+                        else {
+                            if should_die.load(Ordering::Relaxed) {
+                                break;
+                            }
+
+                            continue;
+                        };
+
+                        if should_die.load(Ordering::Relaxed) {
+                            break;
+                        }
+
+                        _ = stdin.write(&input);
+                    }
+                }))
+            } else {
+                None
+            };
+
+            let code = match command.wait() {
+                Ok(status) => status.code().unwrap_or(-1),
+                Err(_) => -1,
+            };
+            drop(data_sender_2);
+            should_die.store(true, Ordering::Relaxed);
+            if let Some(thread) = stdout_thread {
+                _ = thread.join();
+            }
+            if let Some(thread) = stderr_thread {
+                _ = thread.join();
+            }
+            let resp = Response::CommandDone(id, code);
+            let resp = conninfo.encrypt_and_sign_resp(resp)?;
+            conninfo.send(resp)?;
+            if let Some(thread) = stdin_thread {
+                _ = thread.join();
+            }
+            Ok(())
+        })
     });
 
     Ok(CommandHandler { id, data_sender })