feat: support tls in batcher connections [wip]

Author: JuArce

Makefile

@@ -304,10 +304,15 @@ batcher_start: ./batcher/aligned-batcher/.env user_fund_payment_service
 	@echo "Starting Batcher..."
 	@cargo run --manifest-path ./batcher/aligned-batcher/Cargo.toml --release -- --config ./config-files/config-batcher.yaml --env-file ./batcher/aligned-batcher/.env
 
-batcher_start_local: user_fund_payment_service
+batcher_create_self_signed_cert:
+	@echo "Creating TLS certificate for localhost"
+	@openssl req -x509 -days 1825 -newkey rsa:2048 -keyout rootCA.key -out rootCA.crt -nodes -subj '/CN=localhost'
+	@echo "TLS certificate created"
+
+batcher_start_local: user_fund_payment_service batcher_create_self_signed_cert
 	@echo "Starting Batcher..."
 	@$(MAKE) run_storage &
-	@cargo run --manifest-path ./batcher/aligned-batcher/Cargo.toml --release -- --config ./config-files/config-batcher.yaml --env-file ./batcher/aligned-batcher/.env.dev
+	@cargo run --manifest-path ./batcher/aligned-batcher/Cargo.toml --release -- --config ./config-files/config-batcher.yaml --env-file ./batcher/aligned-batcher/.env.dev --cert ./rootCA.crt --key ./rootCA.key
 
 batcher_start_local_no_fund:
 	@echo "Starting Batcher..."

batcher/Cargo.lock

@@ -34,3 +34,5 @@ once_cell = "1.20.2"
 warp = "0.3.7"
 prometheus = { version = "0.13.4", features = ["process"] }
 backon = "1.2.0"
+tokio-boring = "4.13.0"
+boring = "4.13.0"

batcher/aligned-batcher/Cargo.toml

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use boring::ssl::{SslStream};
 
 use crate::types::{batch_queue::BatchQueueEntry, errors::BatcherError};
 use aligned_sdk::{
@@ -15,7 +16,7 @@ use tokio_tungstenite::{
     WebSocketStream,
 };
 
-pub(crate) type WsMessageSink = Arc<RwLock<SplitSink<WebSocketStream<TcpStream>, Message>>>;
+pub(crate) type WsMessageSink = Arc<RwLock<SplitSink<WebSocketStream<SslStream<TcpStream>>, Message>>>;
 
 pub(crate) async fn send_batch_inclusion_data_responses(
     finalized_batch: Vec<BatchQueueEntry>,

batcher/aligned-batcher/src/connection.rs

@@ -15,10 +15,11 @@ use retry::{retry_function, RetryError};
 use tokio::time::{timeout, Instant};
 use types::batch_state::BatchState;
 use types::user_state::UserState;
-
+use boring::ssl::{SslMethod, SslAcceptor, SslStream, SslFiletype};
 use std::collections::HashMap;
 use std::env;
 use std::net::SocketAddr;
+use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -261,7 +262,14 @@ impl Batcher {
         }
     }
 
-    pub async fn listen_connections(self: Arc<Self>, address: &str) -> Result<(), BatcherError> {
+    pub async fn listen_connections(self: Arc<Self>, address: &str, cert: PathBuf, key: PathBuf) -> Result<(), BatcherError> {
+        let mut acceptor;
+        let mut acceptor_builder = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls()).unwrap();
+        acceptor_builder.set_private_key_file(key, SslFiletype::PEM).unwrap();
+        acceptor_builder.set_certificate_chain_file(cert).unwrap();
+        acceptor_builder.check_private_key().unwrap();
+        acceptor = Arc::new(acceptor_builder.build());
+
         // Create the event loop and TCP listener we'll accept connections on.
         let listener = TcpListener::bind(address)
             .await
@@ -273,7 +281,7 @@ impl Batcher {
                 Ok((stream, addr)) => {
                     let batcher = self.clone();
                     // Let's spawn the handling of each connection in a separate task.
-                    tokio::spawn(batcher.handle_connection(stream, addr));
+                    tokio::spawn(batcher.handle_connection(stream, addr, acceptor.clone()));
                 }
                 Err(e) => {
                     self.metrics.user_error(&["connection_accept_error", ""]);
@@ -367,11 +375,14 @@ impl Batcher {
         self: Arc<Self>,
         raw_stream: TcpStream,
         addr: SocketAddr,
+        acceptor: Arc<SslAcceptor>,
     ) -> Result<(), BatcherError> {
         info!("Incoming TCP connection from: {}", addr);
         self.metrics.open_connections.inc();
-
-        let ws_stream_future = tokio_tungstenite::accept_async(raw_stream);
+        let tls_stream = tokio_boring::accept(&acceptor, raw_stream)
+            .await
+            .map_err(|e | BatcherError::TlsError(e.to_string()))?;
+        let ws_stream_future = tokio_tungstenite::accept_async(tls_stream);
         let ws_stream =
             match timeout(Duration::from_secs(CONNECTION_TIMEOUT), ws_stream_future).await {
                 Ok(Ok(stream)) => stream,

batcher/aligned-batcher/src/lib.rs

@@ -1,7 +1,7 @@
 extern crate dotenvy;
 
+use std::path::PathBuf;
 use std::sync::Arc;
-
 use clap::Parser;
 use env_logger::Env;
 
@@ -24,6 +24,12 @@ struct Cli {
     env_file: Option<String>,
     #[arg(short, long)]
     port: Option<u16>,
+    /// cert file
+    #[argh(option, short = 'c')]
+    cert: PathBuf,
+    /// key file
+    #[argh(option, short = 'k')]
+    key: PathBuf,
 }
 
 #[tokio::main]
@@ -40,8 +46,6 @@ async fn main() -> Result<(), BatcherError> {
     let batcher = Batcher::new(cli.config).await;
     let batcher = Arc::new(batcher);
 
-    let addr = format!("localhost:{}", port);
-
     // spawn task to listening for incoming blocks
     tokio::spawn({
         let app = batcher.clone();
@@ -54,7 +58,8 @@ async fn main() -> Result<(), BatcherError> {
 
     batcher.metrics.inc_batcher_restart();
 
-    batcher.listen_connections(&addr).await?;
+    let addr = format!("localhost:{}", port);
+    batcher.listen_connections(&addr, cli.cert, cli.key).await?;
 
     Ok(())
 }

batcher/aligned-batcher/src/main.rs

@@ -41,6 +41,7 @@ impl From<Bytes> for TransactionSendError {
 }
 
 pub enum BatcherError {
+    TlsError(String),
     TcpListenerError(String),
     ConnectionError(tungstenite::Error),
     BatchVerifiedEventStreamError(String),
@@ -75,6 +76,9 @@ impl From<SignatureError> for BatcherError {
 impl fmt::Debug for BatcherError {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
+            BatcherError::TlsError(e) => {
+                write!(f, "TLS Handshake error: {}", e)
+            }
             BatcherError::TcpListenerError(e) => {
                 write!(f, "TCP Listener error: {}", e)
             }