diff --git a/Cargo.toml b/Cargo.toml index 70a46b19..533b079b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,9 @@ package.version = "0.11.0" [workspace.dependencies] beam-lib = { path = "./beam-lib", features = [ "strict-ids" ] } +rsa = "0.10.0-rc.18" +chacha20poly1305 = "0.11" +rand = "0.10" # Command Line Interface clap = { version = "4", features = ["env", "derive"] } reqwest = { version = "0.13", default-features = false } diff --git a/proxy/Cargo.toml b/proxy/Cargo.toml index f6b051f9..c30835e7 100644 --- a/proxy/Cargo.toml +++ b/proxy/Cargo.toml @@ -30,7 +30,7 @@ serde = "1" serde_json = "1" # Encryption handling -rsa = "0.9" +rsa.workspace = true # Server-sent Events (SSE) support tokio-util = "0.7" @@ -39,16 +39,17 @@ sse-stream = "0.2.3" async-stream = "0.3" # Socket dependencies -chacha20poly1305 = { version = "0.10", features = ["stream"], optional = true } +aead-stream = { version = "0.6", features = ["alloc"], optional = true } +chacha20poly1305 = { workspace = true, features = ["bytes"], optional = true } dashmap = { version = "6.0", optional = true} hyper = { version = "1", default-features = false, optional = true } hyper-util = { version = "0.1", default-features = false, features = ["tokio"], optional = true} [features] -sockets = ["dep:chacha20poly1305", "dep:dashmap", "tokio-util/codec", "tokio-util/compat", "shared/sockets", "shared/expire_map", "dep:hyper", "dep:hyper-util"] +sockets = ["dep:aead-stream", "dep:chacha20poly1305", "dep:dashmap", "tokio-util/codec", "tokio-util/compat", "shared/sockets", "shared/expire_map", "dep:hyper", "dep:hyper-util"] [build-dependencies] build-data = "0" [dev-dependencies] -rand = "0.8.5" +rand.workspace = true diff --git a/proxy/src/serve_sockets.rs b/proxy/src/serve_sockets.rs index 060e38bd..13cf8d44 100644 --- a/proxy/src/serve_sockets.rs +++ b/proxy/src/serve_sockets.rs @@ -1,25 +1,16 @@ use std::{ - io::{self, Write}, ops::{Deref, DerefMut}, pin::Pin, sync::{atomic::AtomicUsize, Arc}, task::Poll, time::{Duration, Instant, SystemTime} + io, mem::size_of, pin::Pin, sync::{atomic::AtomicUsize, Arc}, time::{Duration, Instant, SystemTime} }; use axum::{ extract::{Path, Request, State}, http::{self, header, HeaderValue, StatusCode}, response::{IntoResponse, Response}, routing::{get, post}, Extension, Json, RequestPartsExt, Router }; +use aead_stream::{DecryptorLE31, EncryptorLE31, Nonce, StreamLE31}; use bytes::{Buf, BufMut, BytesMut}; -use chacha20poly1305::{ - aead::{ - self, - generic_array::{typenum::Unsigned, GenericArray}, - stream::{DecryptorLE31, EncryptorLE31, NewStream, StreamLE31, StreamPrimitive}, - Buffer, Nonce, OsRng, - }, - consts::{U20, U32}, - AeadCore, AeadInPlace, ChaCha20Poly1305, KeyInit, XChaCha20Poly1305, -}; +use chacha20poly1305::{aead::{array::typenum::Unsigned, AeadCore, Generate}, Key, KeyInit, XChaCha20Poly1305}; use dashmap::DashMap; use futures::{stream::IntoAsyncRead, FutureExt, SinkExt, StreamExt, TryStreamExt}; use hyper_util::rt::TokioIo; -use rsa::rand_core::RngCore; use serde::{Deserialize, Serialize}; use serde_json::Value; use beam_lib::AppOrProxyId; @@ -38,6 +29,10 @@ use crate::{ }; type MsgSecretMap = Arc>; +type StreamNonce = Nonce>; +type FrameLen = u32; +const FRAME_LEN_SIZE: usize = size_of::(); +const TAG_SIZE: usize = ::TagSize::USIZE; const TASK_SECRET_CLEANUP_INTERVAL: Duration = Duration::from_secs(5 * 60); pub(crate) fn router(client: SamplyHttpClient, config: &'static Config) -> Router { @@ -236,17 +231,15 @@ async fn connect_socket( } #[derive(Debug, Clone, Copy)] -struct SocketEncKey(GenericArray); +struct SocketEncKey(Key); impl SocketEncKey { fn generate() -> Self { - let mut arr = GenericArray::default(); - OsRng.fill_bytes(&mut arr); - SocketEncKey(arr) + SocketEncKey(Key::generate()) } fn to_b64_str(&self) -> Result { - Base64UrlSafeNoPadding::encode_to_string(self.as_slice()) + Base64UrlSafeNoPadding::encode_to_string(self.0.as_slice()) } } @@ -265,25 +258,9 @@ impl<'de> Deserialize<'de> for SocketEncKey { D: serde::Deserializer<'de>, { let bytes = Base64UrlSafeNoPadding::decode_to_vec(String::deserialize(deserializer)?, None).map_err(serde::de::Error::custom)?; - if bytes.len() != U32::to_usize() { - return Err(serde::de::Error::custom("Key does not match required key length")); - } else { - Ok(SocketEncKey(GenericArray::clone_from_slice(bytes.as_slice()))) - } - } -} - -impl Deref for SocketEncKey { - type Target = GenericArray; - - fn deref(&self) -> &GenericArray { - &self.0 - } -} - -impl DerefMut for SocketEncKey { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.0 + let key = Key::try_from(bytes.as_slice()) + .map_err(|_| serde::de::Error::custom("Key does not match required key length"))?; + Ok(SocketEncKey(key)) } } @@ -301,111 +278,71 @@ struct DecryptorCodec { decryptor: DecryptorLE31, } -impl DecryptorCodec { - const SIZE_OVERHEAD: usize = 4; -} - -impl EncryptorCodec { - const TAG_SIZE: usize = ::TagSize::USIZE; -} - impl Encoder<&[u8]> for EncryptorCodec { type Error = io::Error; fn encode(&mut self, item: &[u8], dst: &mut BytesMut) -> Result<(), Self::Error> { - let mut enc_buf = EncBuffer::new(dst, (item.len() + Self::TAG_SIZE).try_into().expect("item to large")); - enc_buf.extend_from_slice(item).expect("Infallible"); - self.encryptor - .encrypt_next_in_place(b"", &mut enc_buf) - .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Encryption failed")) + let frame_len: FrameLen = item + .len() + .checked_add(TAG_SIZE) + .and_then(|l| l.try_into().ok()) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "item too large"))?; + + dst.reserve(FRAME_LEN_SIZE + frame_len as usize); + let frame_start = dst.len(); + dst.put_u32_le(frame_len); + dst.extend_from_slice(item); + let mut frame = dst.split_off(frame_start + FRAME_LEN_SIZE); + self + .encryptor + .encrypt_next_in_place(&[], &mut frame) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Encryption failed"))?; + dst.unsplit(frame); + Ok(()) } } impl Decoder for DecryptorCodec { - type Item = Vec; + type Item = BytesMut; type Error = io::Error; fn decode(&mut self, src: &mut BytesMut) -> Result, Self::Error> { - if src.len() < Self::SIZE_OVERHEAD { + if src.len() < FRAME_LEN_SIZE { return Ok(None); } - let mut size_slice = [0; Self::SIZE_OVERHEAD]; - size_slice.clone_from_slice(&src[..Self::SIZE_OVERHEAD]); - let size = u32::from_le_bytes(size_slice); - let total_frame_size = size as usize + Self::SIZE_OVERHEAD; + let mut size_slice = [0; FRAME_LEN_SIZE]; + size_slice.copy_from_slice(&src[..FRAME_LEN_SIZE]); + let size = FrameLen::from_le_bytes(size_slice) as usize; + let total_frame_size = size + FRAME_LEN_SIZE; if src.len() < total_frame_size { return Ok(None); } - let plain = self + src.advance(FRAME_LEN_SIZE); + let mut frame = src.split_to(size); + self .decryptor - .decrypt_next(&src[Self::SIZE_OVERHEAD..(Self::SIZE_OVERHEAD + size as usize)]) + .decrypt_next_in_place(&[], &mut frame) .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Decryption failed"))?; - src.advance(total_frame_size); - Ok(Some(plain)) - } -} - -struct EncBuffer<'a> { - buf: &'a mut BytesMut, - /// The index from which the inplace encryption takes place - enc_idx: usize, -} - -impl<'a> EncBuffer<'a> { - fn new(buffer: &'a mut BytesMut, content_len: u32) -> Self { - let enc_idx = buffer.len() + Self::SIZE_OVERHEAD; - buffer.reserve(content_len as usize + Self::SIZE_OVERHEAD); - buffer.extend_from_slice(&u32::to_le_bytes(content_len)); - Self { buf: buffer, enc_idx } - } - - /// Reserved for size of msg - const SIZE_OVERHEAD: usize = (u32::BITS / 8) as usize; -} - -impl<'a> Buffer for EncBuffer<'a> { - // This should only be called to append the tag to the buffer - fn extend_from_slice(&mut self, other: &[u8]) -> aead::Result<()> { - self.buf.extend_from_slice(other); - Ok(()) - } - - // This should only be called when decrypting - fn truncate(&mut self, len: usize) { - warn!("Buffer got truncated. This should never happen as it should be perfectly sized"); - self.buf.truncate(self.enc_idx + len) - } -} - -impl<'a> AsRef<[u8]> for EncBuffer<'a> { - fn as_ref(&self) -> &[u8] { - &self.buf[self.enc_idx..] - } -} - -impl<'a> AsMut<[u8]> for EncBuffer<'a> { - fn as_mut(&mut self) -> &mut [u8] { - &mut self.buf[self.enc_idx..] + Ok(Some(frame)) } } impl EncryptedSocket { - /// Creates a new cipher stream with a 32 byte key and a 16 + 4 byte Nonce - async fn new(mut inner: S, key: &GenericArray) -> io::Result { - let aead = XChaCha20Poly1305::new(key); + /// Creates a new cipher stream from a shared AEAD key and per-direction STREAM nonces. + async fn new(mut inner: S, key: &SocketEncKey) -> io::Result { + let cipher = XChaCha20Poly1305::new(&key.0); // Encryption - let mut enc_nonce = GenericArray::default(); - OsRng.fill_bytes(&mut enc_nonce); - let encryptor = EncryptorLE31::from_aead(aead.clone(), &enc_nonce); + let enc_nonce = StreamNonce::generate(); + let encryptor = EncryptorLE31::from_aead(cipher.clone(), &enc_nonce); inner.write_all(&enc_nonce).await?; // Decryption - let mut dec_nonce = GenericArray::default(); + let mut dec_nonce = StreamNonce::default(); inner.read_exact(dec_nonce.as_mut_slice()).await?; - let decryptor = DecryptorLE31::from_aead(aead, &dec_nonce); + let decryptor = DecryptorLE31::from_aead(cipher, &dec_nonce); let (r, w) = tokio::io::split(inner); let read = FramedRead::new(r, DecryptorCodec { decryptor }); @@ -452,19 +389,17 @@ impl AsyncWrite for EncryptedSocket { #[cfg(test)] mod tests { - use chacha20poly1305::aead::stream::{Decryptor, Encryptor, EncryptorLE31}; - use rand::Rng; + use rand::RngExt; use tokio::net::{TcpListener, TcpStream}; use super::*; #[tokio::test] async fn test_encryption() { - let mut key = GenericArray::default(); - OsRng.fill_bytes(&mut key); + let key = SocketEncKey::generate(); let data: Arc> = (0..13337).map(|_| { - let mut chunk = vec![0; OsRng.gen_range(1..9999)]; - OsRng.fill_bytes(&mut chunk); + let mut chunk = vec![0; rand::rng().random_range(1..9999)]; + rand::rng().fill(&mut chunk[..]); chunk }).collect::>().into(); @@ -501,7 +436,7 @@ mod tests { }); } - async fn client(key: &GenericArray) -> impl AsyncRead + AsyncWrite { + async fn client(key: &SocketEncKey) -> impl AsyncRead + AsyncWrite { // Wait for server to start tokio::time::sleep(Duration::from_millis(100)).await; let stream = TcpStream::connect("127.0.0.1:1337").await.unwrap(); @@ -510,27 +445,17 @@ mod tests { #[test] fn normal_enc() { - let mut key = GenericArray::default(); - OsRng.fill_bytes(&mut key); const N: usize = 2_usize.pow(10); let test_data: &mut [u8; N] = &mut [0; N]; - OsRng.fill_bytes(test_data); - - let mut nonce = GenericArray::::default(); - OsRng.fill_bytes(&mut nonce); - let aead = XChaCha20Poly1305::new(&key); - let client1 = StreamLE31::from_aead(aead, &nonce); - let mut encryptor = Encryptor::from_stream_primitive(client1); - - // let mut nonce = GenericArray::::default(); - // OsRng.fill_bytes(&mut nonce); - let aead = XChaCha20Poly1305::new(&key); - let client2 = StreamLE31::from_aead(aead, &nonce); - let mut decrypter = Decryptor::from_stream_primitive(client2); + rand::rng().fill(&mut test_data[..]); + let key = SocketEncKey::generate(); + let cipher = XChaCha20Poly1305::new(&key.0); + let nonce = StreamNonce::generate(); + let mut encryptor = EncryptorLE31::from_aead(cipher.clone(), &nonce); + let mut decryptor = DecryptorLE31::from_aead(cipher, &nonce); let cipher_text = encryptor.encrypt_next(test_data.as_slice()).unwrap(); - dbg!(cipher_text.len()); - let a = decrypter.decrypt_next(cipher_text.as_slice()).unwrap(); - assert_eq!(test_data, a.as_slice()); + let plain = decryptor.decrypt_next(cipher_text.as_slice()).unwrap(); + assert_eq!(test_data.as_slice(), plain); } } diff --git a/shared/Cargo.toml b/shared/Cargo.toml index 9302fd6b..4187c680 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -30,11 +30,11 @@ tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } # Crypto -rand = "0.8" -rsa = "0.9" -sha2 = "0.10" +rand.workspace = true +rsa.workspace = true +sha2 = "0.11" +chacha20poly1305.workspace = true openssl = "0.10" -chacha20poly1305 = "0.10" itertools = "0.14.0" jwt-simple = "0.11" diff --git a/shared/src/crypto.rs b/shared/src/crypto.rs index 2a60960b..2dca8de7 100644 --- a/shared/src/crypto.rs +++ b/shared/src/crypto.rs @@ -825,10 +825,8 @@ pub fn load_certificates_from_dir(ca_dir: Option) -> Result Result { let cert_rsa = cert.public_key()?.rsa()?; - let cert_mod = cert_rsa.n(); - let key_mod = key.n(); - let key_mod_bignum = openssl::bn::BigNum::from_slice(&key_mod.to_bytes_be())?; - let is_equal = cert_mod.ucmp(&key_mod_bignum) == std::cmp::Ordering::Equal; + let cert_mod = rsa::BoxedUint::from_be_slice_vartime(&cert_rsa.n().to_vec()); + let is_equal = cert_mod.cmp(&key.n()) == std::cmp::Ordering::Equal; if !is_equal { match ProxyCertInfo::try_from(cert) { Ok(x) => { diff --git a/shared/src/lib.rs b/shared/src/lib.rs index 25f692ad..e16f459d 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -2,8 +2,7 @@ use beam_lib::{AppId, AppOrProxyId, ProxyId, FailureStrategy, WorkStatus}; use chacha20poly1305::{ - aead::{Aead, AeadCore, KeyInit, OsRng}, - XChaCha20Poly1305, XNonce, + aead, XChaCha20Poly1305, XNonce, aead::{Aead, AeadCore, Generate, KeyInit, array::typenum::Unsigned} }; use crypto_jwt::extract_jwt; use errors::SamplyBeamError; @@ -21,7 +20,7 @@ use std::{ time::{Duration, Instant, SystemTime}, net::SocketAddr, error::Error, }; -use rand::Rng; +use rand::{rng, Rng}; use serde::{ de::{DeserializeOwned, Visitor}, Deserialize, Serialize, @@ -256,18 +255,27 @@ pub trait DecryptableMsg: Msg + Serialize + Sized { let encrypted_decryption_key = &encryption_keys[to_array_index]; // Cryptographic Operations - let cipher_engine = XChaCha20Poly1305::new_from_slice(&my_priv_key.decrypt( - Oaep::new::(), + let symmetric_key = my_priv_key.decrypt( + Oaep::::new(), &encrypted_decryption_key, - )?) - .map_err(|e| { - SamplyBeamError::SignEncryptError(format!( - "Decryption error: Cannot initialize stream cipher because {}", - e - )) - })?; - let nonce: XNonce = XNonce::clone_from_slice(&encrypted[0..24]); - let ciphertext = &encrypted[24..]; + )?; + let symmetric_key = aead::Key::::try_from(symmetric_key.as_slice()) + .map_err(|_| { + SamplyBeamError::SignEncryptError( + "Decryption error: Invalid symmetric key length".into(), + ) + })?; + let cipher_engine = XChaCha20Poly1305::new(&symmetric_key); + const NONCE_SIZE: usize = ::NonceSize::USIZE; + let nonce = encrypted + .get(..NONCE_SIZE) + .and_then(|nonce| XNonce::try_from(nonce).ok()) + .ok_or(SamplyBeamError::SignEncryptError( + "Decryption error: Missing or invalid nonce".into(), + ))?; + let ciphertext = encrypted.get(NONCE_SIZE..).ok_or(SamplyBeamError::SignEncryptError( + "Decryption error: Missing ciphertext".into(), + ))?; let plaintext = String::from_utf8( cipher_engine .decrypt(&nonce, ciphertext.as_ref()) @@ -302,9 +310,9 @@ pub trait EncryptableMsg: Msg + Serialize + Sized { receivers_public_keys: &Vec, ) -> Result { // Generate Symmetric Key and Nonce - let mut rng = rand::thread_rng(); - let symmetric_key = XChaCha20Poly1305::generate_key(&mut rng); - let nonce = XChaCha20Poly1305::generate_nonce(&mut rng); + let mut rng = rng(); + let symmetric_key = aead::Key::::generate_from_rng(&mut rng); + let nonce = aead::Nonce::::generate_from_rng(&mut rng); // Encrypt symmetric key with receivers' public keys let Ok(encrypted_keys) = receivers_public_keys @@ -312,7 +320,7 @@ pub trait EncryptableMsg: Msg + Serialize + Sized { .map(|key| { key.encrypt( &mut rng, - Oaep::new::(), + Oaep::::new(), symmetric_key.as_slice(), ) }) @@ -767,7 +775,7 @@ mod tests { }; //Setup Keypairs - let mut rng = rand::thread_rng(); + let mut rng = rand::rng(); let rsa_length: usize = 2048; let p1_private = RsaPrivateKey::new(&mut rng, rsa_length) .expect("Failed to generate private key for proxy 1"); @@ -813,7 +821,7 @@ mod tests { }; //Setup Keypairs - let mut rng = rand::thread_rng(); + let mut rng = rand::rng(); let rsa_length: usize = 2048; let p1_private = RsaPrivateKey::new(&mut rng, rsa_length) .expect("Failed to generate private key for proxy 1"); diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 8e159918..6fb3c1d5 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -11,7 +11,7 @@ beam-lib = { workspace = true, features = ["http-util"] } once_cell = "1" serde_json = "1" anyhow = "1" -rand = "0.8" +rand.workspace = true serde = { version = "1", features = ["derive"] } reqwest = { workspace = true, features = ["stream"], default-features = false } futures.workspace = true diff --git a/tests/src/socket_test.rs b/tests/src/socket_test.rs index e49723a6..1df50d25 100644 --- a/tests/src/socket_test.rs +++ b/tests/src/socket_test.rs @@ -1,7 +1,7 @@ use std::time::Duration; use beam_lib::{BlockingOptions, MsgId}; -use rand::RngCore; +use rand::Rng; use tokio::io::{AsyncWriteExt, AsyncReadExt, AsyncRead, AsyncWrite}; use anyhow::Result; use crate::*; @@ -10,7 +10,7 @@ async fn test_connection(mut a: T, mut b: T) const N: usize = 2_usize.pow(8); for _ in 0..10 { let test_data: &mut [u8; N] = &mut [0; N]; - rand::thread_rng().fill_bytes(test_data); + rand::rng().fill_bytes(test_data); let mut read_buf = [0; N]; a.write_all(test_data).await?; a.flush().await?;