Re: [pbs-devel] [PATCH backup 5/7] proxy: implement 'reload-certificate' command

Re: [pbs-devel] [PATCH backup 5/7] proxy: implement 'reload-certificate' command

[Wolfgang Bumiller]

> On 05/12/2021 10:37 AM Dietmar Maurer <dietmar@proxmox.com> wrote:
> 
>  
> > I wish there was some nice form of a `select_loop!`-like helper...
> 
> Another way would be to avoid the select inside the loop, for example
> by using an Atomic counter (cert is loaded on next accept, not immediately)

That would also work. Should the reload command handler also log then, so it is visible that there's a pending reload?

Another possibility is wrapping the Arc<SslAcceptor> in an AtomicBox (though that's not in the std lib, only AtomicPtr, but it should be easy to write one, (and there's an atomic_box crate)).
The advantage there being that the command handler could do the certificate loading and provide feedback to its client about whether it was successful, and it can just replace the pointer atomically.

Re: [pbs-devel] [PATCH backup 5/7] proxy: implement 'reload-certificate' command

[Wolfgang Bumiller]

> On 05/12/2021 11:13 AM Dietmar Maurer <dietmar@proxmox.com> wrote:
> 
>  
> > > Another way would be to avoid the select inside the loop, for example
> > > by using an Atomic counter (cert is loaded on next accept, not immediately)
> > 
> > That would also work. Should the reload command handler also log then, so it is visible that there's a pending reload?
> > 
> > Another possibility is wrapping the Arc<SslAcceptor> in an AtomicBox (though that's not in the std lib, only AtomicPtr, but it should be easy to write one, (and there's an atomic_box crate)).
> 
> Instead, I use Arc<Mutex<SslAcceptor>>:

right we only need to get the context from the acceptor

lgtm

Re: [pbs-devel] [PATCH backup 5/7] proxy: implement 'reload-certificate' command

[Dietmar Maurer]

> > Another way would be to avoid the select inside the loop, for example
> > by using an Atomic counter (cert is loaded on next accept, not immediately)
> 
> That would also work. Should the reload command handler also log then, so it is visible that there's a pending reload?
> 
> Another possibility is wrapping the Arc<SslAcceptor> in an AtomicBox (though that's not in the std lib, only AtomicPtr, but it should be easy to write one, (and there's an atomic_box crate)).

Instead, I use Arc<Mutex<SslAcceptor>>:

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index fc773459..fb4d43ce 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -1,6 +1,5 @@
-use std::sync::Arc;
+use std::sync::{Mutex, Arc};
 use std::path::{Path, PathBuf};
-use std::pin::Pin;
 use std::os::unix::io::AsRawFd;
 
 use anyhow::{bail, format_err, Error};
@@ -116,19 +115,24 @@ async fn run() -> Result<(), Error> {
 
     //openssl req -x509 -newkey rsa:4096 -keyout /etc/proxmox-backup/proxy.key -out /etc/proxmox-backup/proxy.pem -nodes
 
-    // we build the initial acceptor here as we cannot start if this fails - certificate reloads
-    // will be handled inside the accept loop and simply log an error if we cannot load the new
-    // certificate!
+    // we build the initial acceptor here as we cannot start if this fails -
     let acceptor = make_tls_acceptor()?;
+    let acceptor = Arc::new(Mutex::new(acceptor));
 
-    // to renew the acceptor we just let a command-socket handler trigger a Notify:
-    let notify_tls_cert_reload = Arc::new(tokio::sync::Notify::new());
+    // to renew the acceptor we just add a command-socket handler
     commando_sock.register_command(
         "reload-certificate".to_string(),
         {
-            let notify_tls_cert_reload = Arc::clone(&notify_tls_cert_reload);
+            let acceptor = Arc::clone(&acceptor);
             move |_value| -> Result<_, Error> {
-                notify_tls_cert_reload.notify_one();
+                log::info!("reloading certificate");
+                match make_tls_acceptor() {
+                    Err(err) =>  eprintln!("error reloading certificate: {}", err),
+                    Ok(new_acceptor) => {
+                        let mut guard = acceptor.lock().unwrap();
+                        *guard = new_acceptor;
+                    }
+                }
                 Ok(Value::Null)
             }
         },
@@ -138,7 +142,7 @@ async fn run() -> Result<(), Error> {
         ([0,0,0,0,0,0,0,0], 8007).into(),
         move |listener, ready| {
 
-            let connections = accept_connections(listener, acceptor, debug, notify_tls_cert_reload);
+            let connections = accept_connections(listener, acceptor, debug);
             let connections = hyper::server::accept::from_stream(ReceiverStream::new(connections));
 
             Ok(ready
@@ -179,7 +183,7 @@ async fn run() -> Result<(), Error> {
     Ok(())
 }
 
-fn make_tls_acceptor() -> Result<Arc<SslAcceptor>, Error> {
+fn make_tls_acceptor() -> Result<SslAcceptor, Error> {
     let key_path = configdir!("/proxy.key");
     let cert_path = configdir!("/proxy.pem");
 
@@ -190,7 +194,7 @@ fn make_tls_acceptor() -> Result<Arc<SslAcceptor>, Error> {
         .map_err(|err| format_err!("unable to read proxy cert {} - {}", cert_path, err))?;
     acceptor.check_private_key().unwrap();
 
-    Ok(Arc::new(acceptor.build()))
+    Ok(acceptor.build())
 }
 
 type ClientStreamResult =
@@ -199,76 +203,47 @@ const MAX_PENDING_ACCEPTS: usize = 1024;
 
 fn accept_connections(
     listener: tokio::net::TcpListener,
-    acceptor: Arc<openssl::ssl::SslAcceptor>,
+    acceptor: Arc<Mutex<openssl::ssl::SslAcceptor>>,
     debug: bool,
-    notify_tls_cert_reload: Arc<tokio::sync::Notify>,
 ) -> tokio::sync::mpsc::Receiver<ClientStreamResult> {
 
     let (sender, receiver) = tokio::sync::mpsc::channel(MAX_PENDING_ACCEPTS);
 
-    tokio::spawn(accept_connection(listener, acceptor, debug, sender, notify_tls_cert_reload));
+    tokio::spawn(accept_connection(listener, acceptor, debug, sender));
 
     receiver
 }
 
 async fn accept_connection(
     listener: tokio::net::TcpListener,
-    mut acceptor: Arc<openssl::ssl::SslAcceptor>,
+    acceptor: Arc<Mutex<openssl::ssl::SslAcceptor>>,
     debug: bool,
     sender: tokio::sync::mpsc::Sender<ClientStreamResult>,
-    notify_tls_cert_reload: Arc<tokio::sync::Notify>,
 ) {
     let accept_counter = Arc::new(());
 
-    // Note that these must not be moved out/modified directly, they get pinned in the loop and
-    // "rearmed" after waking up:
-    let mut reload_tls = notify_tls_cert_reload.notified();
-    let mut accept = listener.accept();
-
     loop {
-        let sock;
-
-        // normally we'd use `tokio::pin!()` but we need this to happen outside the loop and we
-        // need to be able to "rearm" the futures:
-        let reload_tls_pin = unsafe { Pin::new_unchecked(&mut reload_tls) };
-        let accept_pin = unsafe { Pin::new_unchecked(&mut accept) };
-        tokio::select! {
-            _ = reload_tls_pin => {
-                // rearm the notification:
-                reload_tls = notify_tls_cert_reload.notified();
-
-                log::info!("reloading certificate");
-                match make_tls_acceptor() {
-                    Err(err) => eprintln!("error reloading certificate: {}", err),
-                    Ok(new_acceptor) => acceptor = new_acceptor,
-                }
+        let (sock, _addr) = match listener.accept().await {
+            Ok(conn) => conn,
+            Err(err) =>  {
+                eprintln!("error accepting tcp connection: {}", err);
                 continue;
             }
-            res = accept_pin => match res {
-                Err(err) => {
-                    eprintln!("error accepting tcp connection: {}", err);
-                    continue;
-                }
-                Ok((new_sock, _addr)) => {
-                    // rearm the accept future:
-                    accept = listener.accept();
-
-                    sock = new_sock;
-                }
-            }
         };
 
         sock.set_nodelay(true).unwrap();
         let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
-        let acceptor = Arc::clone(&acceptor);
+        let acceptor_guard = acceptor.lock().unwrap();
 
-        let ssl = match openssl::ssl::Ssl::new(acceptor.context()) {
+        let ssl = match openssl::ssl::Ssl::new(acceptor_guard.context()) {
             Ok(ssl) => ssl,
             Err(err) => {
                 eprintln!("failed to create Ssl object from Acceptor context - {}", err);
                 continue;
             },
         };
+        drop(acceptor_guard);
+
         let stream = match tokio_openssl::SslStream::new(ssl, sock) {
             Ok(stream) => stream,
             Err(err) => {

Re: [pbs-devel] [PATCH backup 5/7] proxy: implement 'reload-certificate' command

[Dietmar Maurer]

> I wish there was some nice form of a `select_loop!`-like helper...

Another way would be to avoid the select inside the loop, for example
by using an Atomic counter (cert is loaded on next accept, not immediately)

--- Use AtomicUsize ---

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index fc773459..8ecdacec 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -2,6 +2,7 @@ use std::sync::Arc;
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
 use std::os::unix::io::AsRawFd;
+use std::sync::atomic::{AtomicUsize, Ordering};
 
 use anyhow::{bail, format_err, Error};
 use futures::*;
@@ -122,13 +123,13 @@ async fn run() -> Result<(), Error> {
     let acceptor = make_tls_acceptor()?;
 
     // to renew the acceptor we just let a command-socket handler trigger a Notify:
-    let notify_tls_cert_reload = Arc::new(tokio::sync::Notify::new());
+    let notify_tls_cert_reload = Arc::new(AtomicUsize::new(0));
     commando_sock.register_command(
         "reload-certificate".to_string(),
         {
             let notify_tls_cert_reload = Arc::clone(&notify_tls_cert_reload);
             move |_value| -> Result<_, Error> {
-                notify_tls_cert_reload.notify_one();
+                notify_tls_cert_reload.fetch_add(1, Ordering::SeqCst);
                 Ok(Value::Null)
             }
         },
@@ -201,7 +202,7 @@ fn accept_connections(
     listener: tokio::net::TcpListener,
     acceptor: Arc<openssl::ssl::SslAcceptor>,
     debug: bool,
-    notify_tls_cert_reload: Arc<tokio::sync::Notify>,
+    notify_tls_cert_reload: Arc<AtomicUsize>,
 ) -> tokio::sync::mpsc::Receiver<ClientStreamResult> {
 
     let (sender, receiver) = tokio::sync::mpsc::channel(MAX_PENDING_ACCEPTS);
@@ -216,47 +217,26 @@ async fn accept_connection(
     mut acceptor: Arc<openssl::ssl::SslAcceptor>,
     debug: bool,
     sender: tokio::sync::mpsc::Sender<ClientStreamResult>,
-    notify_tls_cert_reload: Arc<tokio::sync::Notify>,
+    notify_tls_cert_reload: Arc<AtomicUsize>,
 ) {
     let accept_counter = Arc::new(());
 
-    // Note that these must not be moved out/modified directly, they get pinned in the loop and
-    // "rearmed" after waking up:
-    let mut reload_tls = notify_tls_cert_reload.notified();
-    let mut accept = listener.accept();
-
     loop {
-        let sock;
-
-        // normally we'd use `tokio::pin!()` but we need this to happen outside the loop and we
-        // need to be able to "rearm" the futures:
-        let reload_tls_pin = unsafe { Pin::new_unchecked(&mut reload_tls) };
-        let accept_pin = unsafe { Pin::new_unchecked(&mut accept) };
-        tokio::select! {
-            _ = reload_tls_pin => {
-                // rearm the notification:
-                reload_tls = notify_tls_cert_reload.notified();
-
-                log::info!("reloading certificate");
-                match make_tls_acceptor() {
-                    Err(err) => eprintln!("error reloading certificate: {}", err),
-                    Ok(new_acceptor) => acceptor = new_acceptor,
-                }
+        let (sock, _addr) = match listener.accept().await {
+            Ok(conn) => conn,
+            Err(err) =>  {
+                eprintln!("error accepting tcp connection: {}", err);
                 continue;
             }
-            res = accept_pin => match res {
-                Err(err) => {
-                    eprintln!("error accepting tcp connection: {}", err);
-                    continue;
-                }
-                Ok((new_sock, _addr)) => {
-                    // rearm the accept future:
-                    accept = listener.accept();
+        };
 
-                    sock = new_sock;
-                }
+        if notify_tls_cert_reload.swap(0, Ordering::SeqCst) > 0 {
+            log::info!("reloading certificate");
+            match make_tls_acceptor() {
+                Err(err) =>  eprintln!("error reloading certificate: {}", err),
+                Ok(new_acceptor) => acceptor = new_acceptor,
             }
-        };
+        }
 
         sock.set_nodelay(true).unwrap();
         let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);

Re: [pbs-devel] [PATCH backup 5/7] proxy: implement 'reload-certificate' command

[Wolfgang Bumiller]

> On 05/12/2021 9:42 AM Dietmar Maurer <dietmar@proxmox.com> wrote:
> 
>  
> Stupid questzioon, but why cant we do:
> 
> diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
> index fc773459..29298a22 100644
> --- a/src/bin/proxmox-backup-proxy.rs
> +++ b/src/bin/proxmox-backup-proxy.rs
> @@ -223,7 +223,6 @@ async fn accept_connection(
>      // Note that these must not be moved out/modified directly, they get pinned in the loop and
>      // "rearmed" after waking up:
>      let mut reload_tls = notify_tls_cert_reload.notified();
> -    let mut accept = listener.accept();
>  
>      loop {
>          let sock;
> @@ -231,7 +230,9 @@ async fn accept_connection(
>          // normally we'd use `tokio::pin!()` but we need this to happen outside the loop and we
>          // need to be able to "rearm" the futures:
>          let reload_tls_pin = unsafe { Pin::new_unchecked(&mut reload_tls) };
> -        let accept_pin = unsafe { Pin::new_unchecked(&mut accept) };
> +        //let accept_pin = unsafe { Pin::new_unchecked(&mut accept) };
> +        let accept = listener.accept();
> +

For the `accept` call this can even work, but not for the notify one
(won't compile), which is a bit weird, since looking up the Notify code,
it *does* have code in the Drop handler to forward wakeups in case it
gets dropped before being received, however, both need to have correct
code to handle this somehow, and I felt like dropping and requerying is
less "nice" and allows for more "weird" errors (similar to the
DuplexStream issue) happening, and I didn't feel comfortable leaving
more room for our main client accept loop to end up in an unexplained
"hanging" state somehow.
(I have trust issues sometimes...)
But sure, we could do this for accept.
(With the patch set I just sent out that would just need the Box::pin()
calls removed)

I wish there was some nice form of a `select_loop!`-like helper...

Re: [pbs-devel] [PATCH backup 5/7] proxy: implement 'reload-certificate' command

[Dietmar Maurer]

Stupid questzioon, but why cant we do:

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index fc773459..29298a22 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -223,7 +223,6 @@ async fn accept_connection(
     // Note that these must not be moved out/modified directly, they get pinned in the loop and
     // "rearmed" after waking up:
     let mut reload_tls = notify_tls_cert_reload.notified();
-    let mut accept = listener.accept();
 
     loop {
         let sock;
@@ -231,7 +230,9 @@ async fn accept_connection(
         // normally we'd use `tokio::pin!()` but we need this to happen outside the loop and we
         // need to be able to "rearm" the futures:
         let reload_tls_pin = unsafe { Pin::new_unchecked(&mut reload_tls) };
-        let accept_pin = unsafe { Pin::new_unchecked(&mut accept) };
+        //let accept_pin = unsafe { Pin::new_unchecked(&mut accept) };
+        let accept = listener.accept();
+
         tokio::select! {
             _ = reload_tls_pin => {
                 // rearm the notification:
@@ -244,14 +245,14 @@ async fn accept_connection(
                 }
                 continue;
             }
-            res = accept_pin => match res {
+            res = accept => match res {
                 Err(err) => {
                     eprintln!("error accepting tcp connection: {}", err);
                     continue;
                 }
                 Ok((new_sock, _addr)) => {
                     // rearm the accept future:
-                    accept = listener.accept();
+                    //accept = listener.accept();
 
                     sock = new_sock;
                 }

>  async fn accept_connection(
>      listener: tokio::net::TcpListener,
> -    acceptor: Arc<openssl::ssl::SslAcceptor>,
> +    mut acceptor: Arc<openssl::ssl::SslAcceptor>,
>      debug: bool,
>      sender: tokio::sync::mpsc::Sender<ClientStreamResult>,
> +    notify_tls_cert_reload: Arc<tokio::sync::Notify>,
>  ) {
>      let accept_counter = Arc::new(());
>  
> +    // Note that these must not be moved out/modified directly, they get pinned in the loop and
> +    // "rearmed" after waking up:
> +    let mut reload_tls = notify_tls_cert_reload.notified();
> +    let mut accept = listener.accept();
> +
>      loop {
> -        let (sock, _addr) = match listener.accept().await {
> -            Ok(conn) => conn,
> -            Err(err) => {
> -                eprintln!("error accepting tcp connection: {}", err);
> +        let sock;
> +
> +        // normally we'd use `tokio::pin!()` but we need this to happen outside the loop and we
> +        // need to be able to "rearm" the futures:
> +        let reload_tls_pin = unsafe { Pin::new_unchecked(&mut reload_tls) };
> +        let accept_pin = unsafe { Pin::new_unchecked(&mut accept) };
> +        tokio::select! {
> +            _ = reload_tls_pin => {
> +                // rearm the notification:
> +                reload_tls = notify_tls_cert_reload.notified();
> +
> +                log::info!("reloading certificate");
> +                match make_tls_acceptor() {
> +                    Err(err) => eprintln!("error reloading certificate: {}", err),
> +                    Ok(new_acceptor) => acceptor = new_acceptor,
> +                }
>                  continue;
>              }
> +            res = accept_pin => match res {
> +                Err(err) => {
> +                    eprintln!("error accepting tcp connection: {}", err);
> +                    continue;
> +                }
> +                Ok((new_sock, _addr)) => {
> +                    // rearm the accept future:
> +                    accept = listener.accept();
> +
> +                    sock = new_sock;
> +                }
> +            }
>          };
>  
>          sock.set_nodelay(true).unwrap();
> -- 
> 2.20.1
> 
> 
> 
> _______________________________________________
> pbs-devel mailing list
> pbs-devel@lists.proxmox.com
> https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH backup 5/7] proxy: implement 'reload-certificate' command

[Wolfgang Bumiller]

to be used via the command socket

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
---
 src/bin/proxmox-backup-proxy.rs | 62 ++++++++++++++++++++++++++++-----
 1 file changed, 53 insertions(+), 9 deletions(-)

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index c05bcd20..793ba67d 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -1,5 +1,6 @@
 use std::sync::Arc;
 use std::path::{Path, PathBuf};
+use std::pin::Pin;
 use std::os::unix::io::AsRawFd;
 
 use anyhow::{bail, format_err, Error};
@@ -7,6 +8,7 @@ use futures::*;
 
 use openssl::ssl::{SslMethod, SslAcceptor, SslFiletype};
 use tokio_stream::wrappers::ReceiverStream;
+use serde_json::Value;
 
 use proxmox::try_block;
 use proxmox::api::RpcEnvironmentType;
@@ -119,13 +121,24 @@ async fn run() -> Result<(), Error> {
     // certificate!
     let acceptor = make_tls_acceptor()?;
 
-    let acceptor = Arc::new(acceptor.build());
+    // to renew the acceptor we just let a command-socket handler trigger a Notify:
+    let notify_tls_cert_reload = Arc::new(tokio::sync::Notify::new());
+    commando_sock.register_command(
+        "reload-certificate".to_string(),
+        {
+            let notify_tls_cert_reload = Arc::clone(&notify_tls_cert_reload);
+            move |_value| -> Result<_, Error> {
+                notify_tls_cert_reload.notify_one();
+                Ok(Value::Null)
+            }
+        },
+    )?;
 
     let server = daemon::create_daemon(
         ([0,0,0,0,0,0,0,0], 8007).into(),
-        |listener, ready| {
+        move |listener, ready| {
 
-            let connections = accept_connections(listener, acceptor, debug);
+            let connections = accept_connections(listener, acceptor, debug, notify_tls_cert_reload);
             let connections = hyper::server::accept::from_stream(ReceiverStream::new(connections));
 
             Ok(ready
@@ -188,30 +201,61 @@ fn accept_connections(
     listener: tokio::net::TcpListener,
     acceptor: Arc<openssl::ssl::SslAcceptor>,
     debug: bool,
+    notify_tls_cert_reload: Arc<tokio::sync::Notify>,
 ) -> tokio::sync::mpsc::Receiver<ClientStreamResult> {
 
     let (sender, receiver) = tokio::sync::mpsc::channel(MAX_PENDING_ACCEPTS);
 
-    tokio::spawn(accept_connection(listener, acceptor, debug, sender));
+    tokio::spawn(accept_connection(listener, acceptor, debug, sender, notify_tls_cert_reload));
 
     receiver
 }
 
 async fn accept_connection(
     listener: tokio::net::TcpListener,
-    acceptor: Arc<openssl::ssl::SslAcceptor>,
+    mut acceptor: Arc<openssl::ssl::SslAcceptor>,
     debug: bool,
     sender: tokio::sync::mpsc::Sender<ClientStreamResult>,
+    notify_tls_cert_reload: Arc<tokio::sync::Notify>,
 ) {
     let accept_counter = Arc::new(());
 
+    // Note that these must not be moved out/modified directly, they get pinned in the loop and
+    // "rearmed" after waking up:
+    let mut reload_tls = notify_tls_cert_reload.notified();
+    let mut accept = listener.accept();
+
     loop {
-        let (sock, _addr) = match listener.accept().await {
-            Ok(conn) => conn,
-            Err(err) => {
-                eprintln!("error accepting tcp connection: {}", err);
+        let sock;
+
+        // normally we'd use `tokio::pin!()` but we need this to happen outside the loop and we
+        // need to be able to "rearm" the futures:
+        let reload_tls_pin = unsafe { Pin::new_unchecked(&mut reload_tls) };
+        let accept_pin = unsafe { Pin::new_unchecked(&mut accept) };
+        tokio::select! {
+            _ = reload_tls_pin => {
+                // rearm the notification:
+                reload_tls = notify_tls_cert_reload.notified();
+
+                log::info!("reloading certificate");
+                match make_tls_acceptor() {
+                    Err(err) => eprintln!("error reloading certificate: {}", err),
+                    Ok(new_acceptor) => acceptor = new_acceptor,
+                }
                 continue;
             }
+            res = accept_pin => match res {
+                Err(err) => {
+                    eprintln!("error accepting tcp connection: {}", err);
+                    continue;
+                }
+                Ok((new_sock, _addr)) => {
+                    // rearm the accept future:
+                    accept = listener.accept();
+
+                    sock = new_sock;
+                }
+            }
         };
 
         sock.set_nodelay(true).unwrap();
-- 
2.20.1