Implement TLS handling

examples/bffh.dhall

@@ -22,10 +22,10 @@
     { cmd = "./examples/fail-actor.sh" 
     }}
   }
-  --, init_connections = [] : List { machine : Text, initiator : Text }
-  , init_connections = [{ machine = "Testmachine", initiator = "Initiator" }]
-  , initiators = --{=}
-  { Initiator = { module = "Dummy", params = { uid = "Testuser" } } }
+  , init_connections = [] : List { machine : Text, initiator : Text }
+  --, init_connections = [{ machine = "Testmachine", initiator = "Initiator" }]
+  , initiators = {=}
+  --{ Initiator = { module = "Dummy", params = { uid = "Testuser" } } }
 , listens = 
   [ { address = "127.0.0.1", port = Some 59661 }
   , { address = "::1", port = Some 59661 }
@@ -74,4 +74,6 @@
       ]
     }
   }
+, certfile = "examples/self-signed-cert.pem"
+, keyfile = "examples/self-signed-key.pem"
 }

src/config.rs

@@ -52,6 +52,10 @@ pub struct Config {
     pub db_path: PathBuf,
 
     pub roles: HashMap<String, RoleConfig>,
+
+    /// Path to a certificate chain to be used
+    pub certfile: PathBuf,
+    pub keyfile: PathBuf,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -123,6 +127,9 @@ impl Default for Config {
 
             db_path: PathBuf::from("/run/bffh/database"),
             roles: HashMap::new(),
+
+            certfile: PathBuf::from("/etc/bffh/pub.crt"),
+            keyfile: PathBuf::from("/etc/bffh/priv.key"),
         }
     }
 }

src/connection.rs

@@ -1,20 +1,27 @@
-use futures::FutureExt;
+use std::fmt::Debug;
+use std::ops::DerefMut;
+use futures::{AsyncRead, AsyncWrite, FutureExt};
 use std::future::Future;
+use std::io::{IoSlice, IoSliceMut};
+use std::pin::Pin;
+use std::rc::Rc;
 use std::sync::Arc;
+use std::task::{Context, Poll};
+use async_rustls::server::TlsStream;
 
 use slog::Logger;
 
 use smol::lock::Mutex;
-use smol::net::TcpStream;
 
 use crate::api::Bootstrap;
 use crate::error::Result;
 
 use capnp_rpc::{rpc_twoparty_capnp, twoparty};
+use futures_util::{pin_mut, ready};
 
 use crate::schema::connection_capnp;
 
-use crate::db::access::{AccessControl, PermRule, RoleIdentifier};
+use crate::db::access::{PermRule, RoleIdentifier};
 use crate::db::user::UserId;
 use crate::db::Databases;
 use crate::network::Network;
@@ -75,14 +82,17 @@ impl ConnectionHandler {
         Self { log, db, network }
     }
 
-    pub fn handle(&mut self, stream: TcpStream) -> impl Future<Output = Result<()>> {
-        info!(self.log, "New connection from on {:?}", stream);
+    pub fn handle<IO: 'static + Unpin + AsyncWrite + AsyncRead>(&mut self, stream: TlsStream<IO>)
+        -> impl Future<Output=Result<()>>
+    {
+        let conn = Connection::new(stream);
+
         let boots = Bootstrap::new(self.log.new(o!()), self.db.clone(), self.network.clone());
         let rpc: connection_capnp::bootstrap::Client = capnp_rpc::new_client(boots);
 
         let network = twoparty::VatNetwork::new(
-            stream.clone(),
-            stream,
+            conn.clone(),
+            conn,
             rpc_twoparty_capnp::Side::Server,
             Default::default(),
         );
@@ -92,3 +102,76 @@ impl ConnectionHandler {
         rpc_system.map(|r| r.map_err(Into::into))
     }
 }
+
+struct Connection<IO> {
+    inner: Rc<Mutex<TlsStream<IO>>>,
+}
+
+impl<IO> Connection<IO> {
+    pub fn new(stream: TlsStream<IO>) -> Self {
+        Self {
+            inner: Rc::new(Mutex::new(stream)),
+        }
+    }
+}
+
+impl<IO> Clone for Connection<IO> {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone()
+        }
+    }
+}
+
+
+impl<IO: 'static + AsyncRead + AsyncWrite + Unpin> AsyncRead for Connection<IO> {
+    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<std::io::Result<usize>> {
+        let f = self.inner.lock();
+        pin_mut!(f);
+        let mut guard = ready!(f.poll(cx));
+        let stream = guard.deref_mut();
+        Pin::new(stream).poll_read(cx, buf)
+    }
+
+    fn poll_read_vectored(self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &mut [IoSliceMut<'_>]) -> Poll<std::io::Result<usize>> {
+        let f = self.inner.lock();
+        pin_mut!(f);
+        let mut guard = ready!(f.poll(cx));
+        let stream = guard.deref_mut();
+        Pin::new(stream).poll_read_vectored(cx, bufs)
+    }
+}
+
+impl<IO: 'static + AsyncWrite + AsyncRead + Unpin> AsyncWrite for Connection<IO> {
+    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
+        let f = self.inner.lock();
+        pin_mut!(f);
+        let mut guard = ready!(f.poll(cx));
+        let stream = guard.deref_mut();
+        Pin::new(stream).poll_write(cx, buf)
+    }
+
+    fn poll_write_vectored(self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[IoSlice<'_>]) -> Poll<std::io::Result<usize>> {
+        let f = self.inner.lock();
+        pin_mut!(f);
+        let mut guard = ready!(f.poll(cx));
+        let stream = guard.deref_mut();
+        Pin::new(stream).poll_write_vectored(cx, bufs)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        let f = self.inner.lock();
+        pin_mut!(f);
+        let mut guard = ready!(f.poll(cx));
+        let stream = guard.deref_mut();
+        Pin::new(stream).poll_flush(cx)
+    }
+
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        let f = self.inner.lock();
+        pin_mut!(f);
+        let mut guard = ready!(f.poll(cx));
+        let stream = guard.deref_mut();
+        Pin::new(stream).poll_close(cx)
+    }
+}

src/db/user.rs

@@ -134,7 +134,7 @@ pub fn load_file<P: AsRef<Path>>(path: P) -> Result<HashMap<String, User>> {
 }
 
 pub fn init(log: Logger, _config: &Config, env: Arc<lmdb::Environment>) -> Result<Internal> {
-    let mut flags = lmdb::DatabaseFlags::empty();
+    let flags = lmdb::DatabaseFlags::empty();
     let db = env.create_db(Some("userdb"), flags)?;
     debug!(&log, "Opened user db successfully.");
 

src/error.rs

@@ -29,6 +29,7 @@ pub enum Error {
     BadVersion((u32,u32)),
     Argon2(argon2::Error),
     EventNetwork(network::Error),
+    RustTls(rustls::TLSError),
     Denied,
 }
 
@@ -83,6 +84,9 @@ impl fmt::Display for Error {
             Error::EventNetwork(e) => {
                 e.fmt(f)
             }
+            Error::RustTls(e) => {
+                write!(f, "TLS Error: {}", e)
+            }
         }
     }
 }
@@ -171,4 +175,10 @@ impl From<argon2::Error> for Error {
     }
 }
 
+impl From<rustls::TLSError> for Error {
+    fn from(e: rustls::TLSError) -> Error {
+        Error::RustTls(e)
+    }
+}
+
 pub(crate) type Result<T> = std::result::Result<T, Error>;

src/server.rs

@@ -1,3 +1,4 @@
+use std::fs::File;
 use slog::Logger;
 
 use crate::config;
@@ -13,10 +14,15 @@ use smol::Executor;
 use futures::prelude::*;
 
 use std::io;
+use std::io::BufReader;
 
 use std::sync::Arc;
 
 use std::os::unix::io::AsRawFd;
+use std::path::Path;
+use async_rustls::TlsAcceptor;
+use rustls::{Certificate, KeyLogFile, NoClientAuth, PrivateKey, ServerConfig};
+use rustls_pemfile::Item;
 use signal_hook::low_level::pipe as sigpipe;
 
 use crate::db::Databases;
@@ -46,6 +52,33 @@ pub fn serve_api_connections(log: Arc<Logger>, config: Config, db: Databases, nw
         io::Result::Ok(LoopResult::Stop)
     });
 
+    info!(log, "Reading certificate chain file");
+    let mut certfp = BufReader::new(File::open(&config.certfile)?);
+    let certs = rustls_pemfile::certs(&mut certfp)?
+        .into_iter()
+        .map(Certificate)
+        .collect();
+    info!(log, "Reading private key file");
+    let mut keyfp = BufReader::new(File::open(&config.keyfile)?);
+    let mut tls_config = ServerConfig::new(Arc::new(NoClientAuth));
+    tls_config.key_log = Arc::new(KeyLogFile::new());
+    if let Some(path) = std::env::var_os("SSLKEYLOGFILE") {
+        let path = Path::new(&path);
+        warn!(log, "TLS SECRET LOGGING ENABLED! This will write all connection secrets to file {}!",
+            path.display());
+    }
+    match rustls_pemfile::read_one(&mut keyfp)? {
+        Some(rustls_pemfile::Item::PKCS8Key(key) | rustls_pemfile::Item::RSAKey(key)) => {
+            let key = PrivateKey(key);
+            tls_config.set_single_cert(certs, key)?;
+        }
+        _ => {
+            error!(log, "private key file must contain a PEM-encoded private key");
+            return Ok(());
+        }
+    }
+    let tls_acceptor: TlsAcceptor = Arc::new(tls_config).into();
+
     // Bind to each address in config.listens.
     // This is a Stream over Futures so it will do absolutely nothing unless polled to completion
     let listeners_s: futures::stream::Collect<_, Vec<TcpListener>> 
@@ -97,18 +130,29 @@ pub fn serve_api_connections(log: Arc<Logger>, config: Config, db: Databases, nw
                             inner_log.new(o!())
                         };
 
+
                     let db = db.clone();
                     let network = network.clone();
                     let tlog = inner_log.new(o!());
+
+                    let tls_acceptor_clone = tls_acceptor.clone();
                     std::thread::spawn(move || {
+                        let tls_acceptor = tls_acceptor_clone;
                         let local_ex = LocalExecutor::new();
 
+                        info!(tlog, "New connection from on {:?}", socket);
                         let mut handler = connection::ConnectionHandler::new(tlog, db, network);
                         // We handle the error using map_err
-                        let f = handler.handle(socket)
+                        let log2 = log.clone();
+                        let f = tls_acceptor.accept(socket)
                             .map_err(move |e| {
                                 error!(log, "Error occured during protocol handling: {}", e);
                             })
+                            .and_then(|stream| {
+                                handler.handle(stream).map_err(move |e| {
+                                   error!(log2, "Error occured during protocol handling: {}", e);
+                                })
+                        })
                         // Void any and all results since pool.spawn allows no return value.
                         .map(|_| ());