Add support for encryption using derived common secret and introduce authentication using an auth secret

Cargo.toml

@@ -19,7 +19,7 @@ hex = "0.4.3"
 k256 = { version = "0.9.6", features = ["ecdh"] }
 log = "0.4.14"
 orion = "0.16.0"
-rand = "0.8.4"
+rand = { version = "0.8.4", features = ["getrandom"] }
 serde = { version = "1.0.126", features = ["derive"] }
 serde_cbor = "0.11.1"
 serde_json = "1.0.64"

src/constants.rs

@@ -1,3 +1,6 @@
 /// Capacity associated with a client broadcasting its received messages that
 /// do not have a callback associated
 pub static CLIENT_BROADCAST_CHANNEL_CAPACITY: usize = 100;
+
+/// Represents the length of the salt to use for encryption
+pub static SALT_LEN: usize = 16;

src/data.rs

@@ -139,12 +139,6 @@ pub enum RequestPayload {
         detach: bool,
     },
 
-    /// Re-connects to a detached process on the remote machine (to receive stdout/stderr)
-    ProcConnect {
-        /// Id of the actively-running process
-        id: usize,
-    },
-
     /// Kills a process running on the remote machine
     ProcKill {
         /// Id of the actively-running process

src/net/transport/codec.rs

@@ -1,5 +1,4 @@
-use bytes::{Buf, BufMut, Bytes, BytesMut};
-use derive_more::{Display, Error, From};
+use bytes::{Buf, BufMut, BytesMut};
 use std::convert::TryInto;
 use tokio::io;
 use tokio_util::codec::{Decoder, Encoder};
@@ -13,20 +12,12 @@ fn frame_size(msg_size: usize) -> usize {
     LEN_SIZE + msg_size
 }
 
-/// Possible errors that can occur during encoding and decoding
-#[derive(Debug, Display, Error, From)]
-pub enum DistantCodecError {
-    #[display(fmt = "Corrupt Marker: {:?}", _0)]
-    CorruptMarker(#[error(not(source))] Bytes),
-    IoError(io::Error),
-}
-
 /// Represents the codec to encode and decode data for transmission
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 pub struct DistantCodec;
 
 impl<'a> Encoder<&'a [u8]> for DistantCodec {
-    type Error = DistantCodecError;
+    type Error = io::Error;
 
     fn encode(&mut self, item: &'a [u8], dst: &mut BytesMut) -> Result<(), Self::Error> {
         // Add our full frame to the bytes
@@ -40,7 +31,7 @@ impl<'a> Encoder<&'a [u8]> for DistantCodec {
 
 impl Decoder for DistantCodec {
     type Item = Vec<u8>;
-    type Error = DistantCodecError;
+    type Error = io::Error;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
         // First, check if we have more data than just our frame's message length

src/net/transport/mod.rs

@@ -1,10 +1,14 @@
-use crate::utils::Session;
-use codec::{DistantCodec, DistantCodecError};
+use crate::{constants::SALT_LEN, utils::Session};
+use codec::DistantCodec;
 use derive_more::{Display, Error, From};
 use futures::SinkExt;
+use k256::{ecdh::EphemeralSecret, EncodedPoint, PublicKey};
 use orion::{
     aead::{self, SecretKey},
+    auth::{self, Tag},
     errors::UnknownCryptoError,
+    kdf::{self, Salt},
+    pwhash::Password,
 };
 use serde::{de::DeserializeOwned, Serialize};
 use std::{net::SocketAddr, sync::Arc};
@@ -19,7 +23,9 @@ mod codec;
 
 #[derive(Debug, Display, Error, From)]
 pub enum TransportError {
-    CodecError(DistantCodecError),
+    #[from(ignore)]
+    AuthError(UnknownCryptoError),
+    #[from(ignore)]
     EncryptError(UnknownCryptoError),
     IoError(io::Error),
     SerializeError(serde_cbor::Error),
@@ -27,66 +33,108 @@ pub enum TransportError {
 
 /// Represents a transport of data across the network
 pub struct Transport {
-    inner: Framed<TcpStream, DistantCodec>,
-    key: Arc<SecretKey>,
+    /// Underlying connection to some remote system
+    conn: Framed<TcpStream, DistantCodec>,
+
+    /// Used to sign and validate messages
+    auth_key: Arc<SecretKey>,
+
+    /// Used to encrypt and decrypt messages
+    crypt_key: Arc<SecretKey>,
 }
 
 impl Transport {
-    /// Wraps a `TcpStream` and associated credentials in a transport layer
-    pub fn new(stream: TcpStream, key: Arc<SecretKey>) -> Self {
-        Self {
-            inner: Framed::new(stream, DistantCodec),
-            key,
-        }
+    /// Takes a pre-existing connection and performs a handshake to build out the encryption key
+    /// with the remote system, returning a transport ready to communicate with the other side
+    pub async fn from_handshake(stream: TcpStream, auth_key: Arc<SecretKey>) -> io::Result<Self> {
+        // First, wrap the raw stream in our framed codec
+        let mut conn = Framed::new(stream, DistantCodec);
+
+        // Second, generate a private key that will be used to eventually derive a shared secret
+        let private_key = EphemeralSecret::random(&mut rand::rngs::OsRng);
+
+        // Third, produce a private key that will be shared unencrypted to the other side
+        let public_key = EncodedPoint::from(private_key.public_key());
+
+        // Fourth, share a random salt and the public key with the server as our first message
+        let salt = Salt::generate(SALT_LEN).map_err(|x| io::Error::new(io::ErrorKind::Other, x))?;
+        let mut data = Vec::new();
+        data.extend_from_slice(salt.as_ref());
+        data.extend_from_slice(public_key.as_bytes());
+        conn.send(&data)
+            .await
+            .map_err(|x| io::Error::new(io::ErrorKind::Other, x))?;
+
+        // Fifth, wait for a response that we will assume is the other side's salt & public key
+        let data = conn.next().await.ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::UnexpectedEof,
+                "Stream ended before handshake completed",
+            )
+        })??;
+        let (salt_bytes, other_public_key_bytes) = data.split_at(SALT_LEN);
+        let other_salt = Salt::from_slice(salt_bytes)
+            .map_err(|x| io::Error::new(io::ErrorKind::InvalidData, x))?;
+
+        // Sixth, decode the serialized public key from the other side
+        let other_public_key = PublicKey::from_sec1_bytes(other_public_key_bytes)
+            .map_err(|x| io::Error::new(io::ErrorKind::InvalidData, x))?;
+
+        // Seventh, establish a shared secret that is NOT uniformly random, so we can't
+        // directly use it as our encryption key (32 bytes in length)
+        let shared_secret = private_key.diffie_hellman(&other_public_key);
+
+        // Eighth, convert our secret key into an orion password that we'll use to derive
+        // a new key; need to ensure that the secret is at least 32 bytes!
+        let password = Password::from_slice(shared_secret.as_bytes())
+            .map_err(|x| io::Error::new(io::ErrorKind::InvalidData, x))?;
+
+        // Ninth, take our salt and the salt from the other side and combine them in a consistent
+        // manner such that both sides derive the same salt
+        let mixed_salt = Salt::from_slice(
+            &salt
+                .as_ref()
+                .iter()
+                .zip(other_salt.as_ref().iter())
+                .map(|(x, y)| x ^ y)
+                .collect::<Vec<u8>>(),
+        )
+        .map_err(|x| io::Error::new(io::ErrorKind::InvalidData, x))?;
+
+        // Tenth, derive a higher-entropy key from our shared secret
+        let derived_key = kdf::derive_key(&password, &mixed_salt, 3, 1 << 16, 32)
+            .map_err(|x| io::Error::new(io::ErrorKind::Other, x))?;
+
+        let crypt_key = Arc::new(derived_key);
+        log::trace!(
+            "Handshake complete: {}",
+            hex::encode(crypt_key.unprotected_as_bytes())
+        );
+
+        Ok(Self {
+            conn,
+            auth_key,
+            crypt_key,
+        })
     }
 
-    /// Establishes a connection using the provided session
+    /// Establishes a connection using the provided session and performs a handshake to establish
+    /// means of encryption, returning a transport ready to communicate with the other side
     pub async fn connect(session: Session) -> io::Result<Self> {
         let stream = TcpStream::connect(session.to_socket_addr().await?).await?;
-        Ok(Self::new(stream, Arc::new(session.key)))
+        Self::from_handshake(stream, Arc::new(session.auth_key)).await
     }
 
     /// Returns the address of the peer the transport is connected to
     pub fn peer_addr(&self) -> io::Result<SocketAddr> {
-        self.inner.get_ref().peer_addr()
-    }
-
-    /// Sends some data across the wire
-    #[allow(dead_code)]
-    pub async fn send<T: Serialize>(&mut self, data: T) -> Result<(), TransportError> {
-        // Serialize, encrypt, and then (TODO) sign
-        // NOTE: Cannot used packed implementation for now due to issues with deserialization
-        let data = serde_cbor::to_vec(&data)?;
-        let data = aead::seal(&self.key, &data)?;
-
-        self.inner
-            .send(&data)
-            .await
-            .map_err(TransportError::CodecError)
-    }
-
-    /// Receives some data from out on the wire, waiting until it's available,
-    /// returning none if the transport is now closed
-    #[allow(dead_code)]
-    pub async fn receive<T: DeserializeOwned>(&mut self) -> Result<Option<T>, TransportError> {
-        // If data is received, we process like usual
-        if let Some(data) = self.inner.next().await {
-            // Validate (TODO) signature, decrypt, and then deserialize
-            let data = data?;
-            let data = aead::open(&self.key, &data)?;
-            let data = serde_cbor::from_slice(&data)?;
-            Ok(Some(data))
-
-        // Otherwise, if no data is received, this means that our socket has closed
-        } else {
-            Ok(None)
-        }
+        self.conn.get_ref().peer_addr()
     }
 
     /// Splits transport into read and write halves
     pub fn into_split(self) -> (TransportReadHalf, TransportWriteHalf) {
-        let key = self.key;
-        let parts = self.inner.into_parts();
+        let auth_key = self.auth_key;
+        let crypt_key = self.crypt_key;
+        let parts = self.conn.into_parts();
         let (read_half, write_half) = parts.io.into_split();
 
         // Create our split read half and populate its buffer with existing contents
@@ -98,12 +146,14 @@ impl Transport {
         *f_write.write_buffer_mut() = parts.write_buf;
 
         let t_read = TransportReadHalf {
-            inner: f_read,
-            key: Arc::clone(&key),
+            conn: f_read,
+            auth_key: Arc::clone(&auth_key),
+            crypt_key: Arc::clone(&crypt_key),
         };
         let t_write = TransportWriteHalf {
-            inner: f_write,
-            key,
+            conn: f_write,
+            auth_key,
+            crypt_key,
         };
 
         (t_read, t_write)
@@ -112,29 +162,44 @@ impl Transport {
 
 /// Represents a transport of data out to the network
 pub struct TransportWriteHalf {
-    inner: FramedWrite<tcp::OwnedWriteHalf, DistantCodec>,
-    key: Arc<SecretKey>,
+    /// Underlying connection to some remote system
+    conn: FramedWrite<tcp::OwnedWriteHalf, DistantCodec>,
+
+    /// Used to sign and validate messages
+    auth_key: Arc<SecretKey>,
+
+    /// Used to encrypt and decrypt messages
+    crypt_key: Arc<SecretKey>,
 }
 
 impl TransportWriteHalf {
     /// Sends some data across the wire, waiting for it to completely send
     pub async fn send<T: Serialize>(&mut self, data: T) -> Result<(), TransportError> {
-        // Serialize, encrypt, and then (TODO) sign
+        // Serialize, encrypt, and then sign
         // NOTE: Cannot used packed implementation for now due to issues with deserialization
         let data = serde_cbor::to_vec(&data)?;
-        let data = aead::seal(&self.key, &data)?;
-
-        self.inner
-            .send(&data)
-            .await
-            .map_err(TransportError::CodecError)
+        let data = aead::seal(&self.crypt_key, &data).map_err(TransportError::EncryptError)?;
+        let tag = auth::authenticate(&self.auth_key, &data).map_err(TransportError::AuthError)?;
+
+        // Send {TAG LEN}{TAG}{ENCRYPTED DATA}
+        let mut out: Vec<u8> = Vec::new();
+        out.push(tag.unprotected_as_bytes().len() as u8);
+        out.extend_from_slice(tag.unprotected_as_bytes());
+        out.extend(data);
+        self.conn.send(&out).await.map_err(TransportError::from)
     }
 }
 
 /// Represents a transport of data in from the network
 pub struct TransportReadHalf {
-    inner: FramedRead<tcp::OwnedReadHalf, DistantCodec>,
-    key: Arc<SecretKey>,
+    /// Underlying connection to some remote system
+    conn: FramedRead<tcp::OwnedReadHalf, DistantCodec>,
+
+    /// Used to sign and validate messages
+    auth_key: Arc<SecretKey>,
+
+    /// Used to encrypt and decrypt messages
+    crypt_key: Arc<SecretKey>,
 }
 
 impl TransportReadHalf {
@@ -142,10 +207,25 @@ impl TransportReadHalf {
     /// returning none if the transport is now closed
     pub async fn receive<T: DeserializeOwned>(&mut self) -> Result<Option<T>, TransportError> {
         // If data is received, we process like usual
-        if let Some(data) = self.inner.next().await {
-            // Validate (TODO) signature, decrypt, and then deserialize
-            let data = data?;
-            let data = aead::open(&self.key, &data)?;
+        if let Some(data) = self.conn.next().await {
+            let mut data = data?;
+            if data.is_empty() {
+                return Err(TransportError::from(io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    "Received data is empty",
+                )));
+            }
+
+            // Retrieve in form {TAG LEN}{TAG}{ENCRYPTED DATA}
+            let tag_len = data[0];
+            let tag =
+                Tag::from_slice(&data[1..=tag_len as usize]).map_err(TransportError::AuthError)?;
+            let data = data.split_off(tag_len as usize + 1);
+
+            // Validate signature, decrypt, and then deserialize
+            auth::authenticate_verify(&tag, &self.auth_key, &data)
+                .map_err(TransportError::AuthError)?;
+            let data = aead::open(&self.crypt_key, &data).map_err(TransportError::EncryptError)?;
             let data = serde_cbor::from_slice(&data)?;
             Ok(Some(data))
 

src/subcommand/launch.rs

@@ -79,17 +79,21 @@ async fn run_async(cmd: LaunchSubcommand, _opt: CommonOpt) -> Result<(), Error>
         .unwrap_or(Err(Error::MissingSessionData));
 
     // Write a session file containing our data for use in subsequent calls
-    let (port, key) = result?;
+    let (port, auth_key) = result?;
     let session = Session {
         host: cmd.host,
         port,
-        key,
+        auth_key,
     };
 
     session.save().await?;
 
     if cmd.print_startup_data {
-        println!("DISTANT DATA {} {}", port, session.to_unprotected_hex_key());
+        println!(
+            "DISTANT DATA {} {}",
+            port,
+            session.to_unprotected_hex_auth_key()
+        );
     }
 
     Ok(())

src/subcommand/listen/handler.rs

@@ -40,7 +40,6 @@ pub(super) async fn process(
             RequestPayload::ProcRun { cmd, args, detach } => {
                 proc_run(client_id, state, tx, cmd, args, detach).await
             }
-            RequestPayload::ProcConnect { id } => proc_connect(id).await,
             RequestPayload::ProcKill { id } => proc_kill(state, id).await,
             RequestPayload::ProcStdin { id, data } => proc_stdin(state, id, data).await,
             RequestPayload::ProcList {} => proc_list(state).await,
@@ -342,10 +341,6 @@ async fn proc_run(
     Ok(ResponsePayload::ProcStart { id })
 }
 
-async fn proc_connect(id: usize) -> Result<ResponsePayload, Box<dyn Error>> {
-    todo!();
-}
-
 async fn proc_kill(state: HState, id: usize) -> Result<ResponsePayload, Box<dyn Error>> {
     if let Some(process) = state.lock().await.processes.remove(&id) {
         process.kill_tx.send(()).map_err(|_| {

src/subcommand/listen/mod.rs

@@ -136,9 +136,18 @@ async fn run_async(cmd: ListenSubcommand, _opt: CommonOpt, is_forked: bool) -> R
         // Create a unique id for the client
         let id = rand::random();
 
-        // Build a transport around the client, splitting into read and write halves so we can
-        // handle input and output concurrently
-        let (t_read, t_write) = Transport::new(client, Arc::clone(&key)).into_split();
+        // Establish a proper connection via a handshake, discarding the connection otherwise
+        let transport = match Transport::from_handshake(client, Arc::clone(&key)).await {
+            Ok(transport) => transport,
+            Err(x) => {
+                error!("<Client @ {}> Failed handshake: {}", addr_string, x);
+                continue;
+            }
+        };
+
+        // Split the transport into read and write halves so we can handle input
+        // and output concurrently
+        let (t_read, t_write) = transport.into_split();
         let (tx, rx) = mpsc::channel(cmd.max_msg_capacity as usize);
 
         // Spawn a new task that loops to handle requests from the client

src/subcommand/send.rs

@@ -29,7 +29,7 @@ async fn run_async(cmd: SendSubcommand, _opt: CommonOpt) -> Result<(), Error> {
     let req = Request::from(cmd.operation);
 
     // Special conditions for continuing to process responses
-    let is_proc_req = req.payload.is_proc_run() || req.payload.is_proc_connect();
+    let is_proc_req = req.payload.is_proc_run();
     let not_detach = if let RequestPayload::ProcRun { detach, .. } = req.payload {
         !detach
     } else {

src/utils.rs

@@ -37,13 +37,13 @@ pub enum SessionError {
 pub struct Session {
     pub host: String,
     pub port: u16,
-    pub key: SecretKey,
+    pub auth_key: SecretKey,
 }
 
 impl Session {
     /// Returns a string representing the secret key as hex
-    pub fn to_unprotected_hex_key(&self) -> String {
-        hex::encode(self.key.unprotected_as_bytes())
+    pub fn to_unprotected_hex_auth_key(&self) -> String {
+        hex::encode(self.auth_key.unprotected_as_bytes())
     }
 
     /// Returns the ip address associated with the session based on the host
@@ -75,7 +75,7 @@ impl Session {
 
     /// Saves a session to disk
     pub async fn save(&self) -> io::Result<()> {
-        let key_hex_str = self.to_unprotected_hex_key();
+        let key_hex_str = self.to_unprotected_hex_auth_key();
 
         // Ensure our cache directory exists
         let cache_dir = PROJECT_DIRS.cache_dir();
@@ -115,12 +115,16 @@ impl Session {
             .map_err(|_| SessionError::InvalidSessionPort)?;
 
         // Third, load up the key and convert it back into a secret key from a hex slice
-        let key = SecretKey::from_slice(
+        let auth_key = SecretKey::from_slice(
             &hex::decode(tokens.next().ok_or(SessionError::MissingSessionKey)?.trim())
                 .map_err(|_| SessionError::BadSessionHexKey)?,
         )
         .map_err(|_| SessionError::InvalidSessionKey)?;
 
-        Ok(Session { host, port, key })
+        Ok(Session {
+            host,
+            port,
+            auth_key,
+        })
     }
 }