[pbs-devel] [PATCH backup 0/7] hot-reload proxy certificates

[pbs-devel] [PATCH backup 0/7] hot-reload proxy certificates

[Wolfgang Bumiller]

This adds the ability to tell a running proxy to just reload the TLS
cert certificates via the command-socket.

Starts off with some cleanup/refactoring to get rid of all that heavy
indentation...

Wolfgang Bumiller (7):
  proxy: factor out accept_connection
  proxy: "continue on error" for the accept call, too
  proxy: Arc usage cleanup
  proxy: factor out tls acceptor creation
  proxy: implement 'reload-certificate' command
  refactor send_command
  hot-reload proxy certificate when updating via the API

 src/api2/node/certificates.rs   |  26 ++--
 src/bin/proxmox-backup-proxy.rs | 220 ++++++++++++++++++++------------
 src/config.rs                   |  17 +--
 src/server.rs                   |   9 ++
 src/server/command_socket.rs    |  71 ++++++-----
 src/server/worker_task.rs       |   4 +-
 6 files changed, 204 insertions(+), 143 deletions(-)

-- 
2.20.1

[pbs-devel] [PATCH backup 1/7] proxy: factor out accept_connection

[Wolfgang Bumiller]

no functional changes, moved code and named the channel's
type for more readability

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
---
diff -w for a quicker view:
    diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
    index 31dc8332..27d1cbeb 100644
    --- a/src/bin/proxmox-backup-proxy.rs
    +++ b/src/bin/proxmox-backup-proxy.rs
    @@ -170,19 +170,31 @@ async fn run() -> Result<(), Error> {
         Ok(())
     }

    +type ClientStreamResult =
    +    Result<std::pin::Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>, Error>;
    +const MAX_PENDING_ACCEPTS: usize = 1024;
    +
     fn accept_connections(
         listener: tokio::net::TcpListener,
         acceptor: Arc<openssl::ssl::SslAcceptor>,
         debug: bool,
    -) -> tokio::sync::mpsc::Receiver<Result<std::pin::Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>, Error>> {
    -
    -    const MAX_PENDING_ACCEPTS: usize = 1024;
    +) -> tokio::sync::mpsc::Receiver<ClientStreamResult> {

         let (sender, receiver) = tokio::sync::mpsc::channel(MAX_PENDING_ACCEPTS);

    +    tokio::spawn(accept_connection(listener, acceptor, debug, sender));
    +
    +    receiver
    +}
    +
    +async fn accept_connection(
    +    listener: tokio::net::TcpListener,
    +    acceptor: Arc<openssl::ssl::SslAcceptor>,
    +    debug: bool,
    +    sender: tokio::sync::mpsc::Sender<ClientStreamResult>,
    +) {
         let accept_counter = Arc::new(());

    -    tokio::spawn(async move {
         loop {
             match listener.accept().await {
                 Err(err) => {
    @@ -246,9 +258,6 @@ fn accept_connections(
                 }
             }
         }
    -    });
    -
    -    receiver
     }


 src/bin/proxmox-backup-proxy.rs | 127 +++++++++++++++++---------------
 1 file changed, 68 insertions(+), 59 deletions(-)

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index 31dc8332..27d1cbeb 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -170,85 +170,94 @@ async fn run() -> Result<(), Error> {
     Ok(())
 }
 
+type ClientStreamResult =
+    Result<std::pin::Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>, Error>;
+const MAX_PENDING_ACCEPTS: usize = 1024;
+
 fn accept_connections(
     listener: tokio::net::TcpListener,
     acceptor: Arc<openssl::ssl::SslAcceptor>,
     debug: bool,
-) -> tokio::sync::mpsc::Receiver<Result<std::pin::Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>, Error>> {
-
-    const MAX_PENDING_ACCEPTS: usize = 1024;
+) -> tokio::sync::mpsc::Receiver<ClientStreamResult> {
 
     let (sender, receiver) = tokio::sync::mpsc::channel(MAX_PENDING_ACCEPTS);
 
+    tokio::spawn(accept_connection(listener, acceptor, debug, sender));
+
+    receiver
+}
+
+async fn accept_connection(
+    listener: tokio::net::TcpListener,
+    acceptor: Arc<openssl::ssl::SslAcceptor>,
+    debug: bool,
+    sender: tokio::sync::mpsc::Sender<ClientStreamResult>,
+) {
     let accept_counter = Arc::new(());
 
-    tokio::spawn(async move {
-        loop {
-            match listener.accept().await {
-                Err(err) => {
-                    eprintln!("error accepting tcp connection: {}", err);
-                }
-                Ok((sock, _addr)) =>  {
-                    sock.set_nodelay(true).unwrap();
-                    let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
-                    let acceptor = Arc::clone(&acceptor);
-
-                    let ssl = match openssl::ssl::Ssl::new(acceptor.context()) {
-                        Ok(ssl) => ssl,
-                        Err(err) => {
-                            eprintln!("failed to create Ssl object from Acceptor context - {}", err);
-                            continue;
-                        },
-                    };
-                    let stream = match tokio_openssl::SslStream::new(ssl, sock) {
-                        Ok(stream) => stream,
-                        Err(err) => {
-                            eprintln!("failed to create SslStream using ssl and connection socket - {}", err);
-                            continue;
-                        },
-                    };
-
-                    let mut stream = Box::pin(stream);
-                    let sender = sender.clone();
-
-                    if Arc::strong_count(&accept_counter) > MAX_PENDING_ACCEPTS {
-                        eprintln!("connection rejected - to many open connections");
+    loop {
+        match listener.accept().await {
+            Err(err) => {
+                eprintln!("error accepting tcp connection: {}", err);
+            }
+            Ok((sock, _addr)) =>  {
+                sock.set_nodelay(true).unwrap();
+                let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
+                let acceptor = Arc::clone(&acceptor);
+
+                let ssl = match openssl::ssl::Ssl::new(acceptor.context()) {
+                    Ok(ssl) => ssl,
+                    Err(err) => {
+                        eprintln!("failed to create Ssl object from Acceptor context - {}", err);
                         continue;
-                    }
+                    },
+                };
+                let stream = match tokio_openssl::SslStream::new(ssl, sock) {
+                    Ok(stream) => stream,
+                    Err(err) => {
+                        eprintln!("failed to create SslStream using ssl and connection socket - {}", err);
+                        continue;
+                    },
+                };
+
+                let mut stream = Box::pin(stream);
+                let sender = sender.clone();
+
+                if Arc::strong_count(&accept_counter) > MAX_PENDING_ACCEPTS {
+                    eprintln!("connection rejected - to many open connections");
+                    continue;
+                }
 
-                    let accept_counter = accept_counter.clone();
-                    tokio::spawn(async move {
-                        let accept_future = tokio::time::timeout(
-                            Duration::new(10, 0), stream.as_mut().accept());
+                let accept_counter = accept_counter.clone();
+                tokio::spawn(async move {
+                    let accept_future = tokio::time::timeout(
+                        Duration::new(10, 0), stream.as_mut().accept());
 
-                        let result = accept_future.await;
+                    let result = accept_future.await;
 
-                        match result {
-                            Ok(Ok(())) => {
-                                if sender.send(Ok(stream)).await.is_err() && debug {
-                                    eprintln!("detect closed connection channel");
-                                }
+                    match result {
+                        Ok(Ok(())) => {
+                            if sender.send(Ok(stream)).await.is_err() && debug {
+                                eprintln!("detect closed connection channel");
                             }
-                            Ok(Err(err)) => {
-                                if debug {
-                                    eprintln!("https handshake failed - {}", err);
-                                }
+                        }
+                        Ok(Err(err)) => {
+                            if debug {
+                                eprintln!("https handshake failed - {}", err);
                             }
-                            Err(_) => {
-                                if debug {
-                                    eprintln!("https handshake timeout");
-                                }
+                        }
+                        Err(_) => {
+                            if debug {
+                                eprintln!("https handshake timeout");
                             }
                         }
+                    }
 
-                        drop(accept_counter); // decrease reference count
-                    });
-                }
+                    drop(accept_counter); // decrease reference count
+                });
             }
         }
-    });
-
-    receiver
+    }
 }
 
 fn start_stat_generator() {
-- 
2.20.1

[pbs-devel] [PATCH backup 2/7] proxy: "continue on error" for the accept call, too

[Wolfgang Bumiller]

as this gets rid of 2 levels of indentation

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
---
diff -w for a quicker view:
    diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
    index 27d1cbeb..b28ab75b 100644
    --- a/src/bin/proxmox-backup-proxy.rs
    +++ b/src/bin/proxmox-backup-proxy.rs
    @@ -196,11 +196,14 @@ async fn accept_connection(
         let accept_counter = Arc::new(());

         loop {
    -        match listener.accept().await {
    +        let (sock, _addr) = match listener.accept().await {
    +            Ok(conn) => conn,
                 Err(err) => {
                     eprintln!("error accepting tcp connection: {}", err);
    +                continue;
                 }
    -            Ok((sock, _addr)) =>  {
    +        };
    +
             sock.set_nodelay(true).unwrap();
             let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
             let acceptor = Arc::clone(&acceptor);
    @@ -257,8 +260,6 @@ async fn accept_connection(
             });
         }
     }
    -    }
    -}

     fn start_stat_generator() {
         let abort_future = server::shutdown_future();


 src/bin/proxmox-backup-proxy.rs | 107 ++++++++++++++++----------------
 1 file changed, 54 insertions(+), 53 deletions(-)

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index 27d1cbeb..b28ab75b 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -196,67 +196,68 @@ async fn accept_connection(
     let accept_counter = Arc::new(());
 
     loop {
-        match listener.accept().await {
+        let (sock, _addr) = match listener.accept().await {
+            Ok(conn) => conn,
             Err(err) => {
                 eprintln!("error accepting tcp connection: {}", err);
+                continue;
             }
-            Ok((sock, _addr)) =>  {
-                sock.set_nodelay(true).unwrap();
-                let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
-                let acceptor = Arc::clone(&acceptor);
-
-                let ssl = match openssl::ssl::Ssl::new(acceptor.context()) {
-                    Ok(ssl) => ssl,
-                    Err(err) => {
-                        eprintln!("failed to create Ssl object from Acceptor context - {}", err);
-                        continue;
-                    },
-                };
-                let stream = match tokio_openssl::SslStream::new(ssl, sock) {
-                    Ok(stream) => stream,
-                    Err(err) => {
-                        eprintln!("failed to create SslStream using ssl and connection socket - {}", err);
-                        continue;
-                    },
-                };
-
-                let mut stream = Box::pin(stream);
-                let sender = sender.clone();
-
-                if Arc::strong_count(&accept_counter) > MAX_PENDING_ACCEPTS {
-                    eprintln!("connection rejected - to many open connections");
-                    continue;
-                }
+        };
 
-                let accept_counter = accept_counter.clone();
-                tokio::spawn(async move {
-                    let accept_future = tokio::time::timeout(
-                        Duration::new(10, 0), stream.as_mut().accept());
+        sock.set_nodelay(true).unwrap();
+        let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
+        let acceptor = Arc::clone(&acceptor);
 
-                    let result = accept_future.await;
+        let ssl = match openssl::ssl::Ssl::new(acceptor.context()) {
+            Ok(ssl) => ssl,
+            Err(err) => {
+                eprintln!("failed to create Ssl object from Acceptor context - {}", err);
+                continue;
+            },
+        };
+        let stream = match tokio_openssl::SslStream::new(ssl, sock) {
+            Ok(stream) => stream,
+            Err(err) => {
+                eprintln!("failed to create SslStream using ssl and connection socket - {}", err);
+                continue;
+            },
+        };
 
-                    match result {
-                        Ok(Ok(())) => {
-                            if sender.send(Ok(stream)).await.is_err() && debug {
-                                eprintln!("detect closed connection channel");
-                            }
-                        }
-                        Ok(Err(err)) => {
-                            if debug {
-                                eprintln!("https handshake failed - {}", err);
-                            }
-                        }
-                        Err(_) => {
-                            if debug {
-                                eprintln!("https handshake timeout");
-                            }
-                        }
-                    }
+        let mut stream = Box::pin(stream);
+        let sender = sender.clone();
 
-                    drop(accept_counter); // decrease reference count
-                });
-            }
+        if Arc::strong_count(&accept_counter) > MAX_PENDING_ACCEPTS {
+            eprintln!("connection rejected - to many open connections");
+            continue;
         }
+
+        let accept_counter = accept_counter.clone();
+        tokio::spawn(async move {
+            let accept_future = tokio::time::timeout(
+                Duration::new(10, 0), stream.as_mut().accept());
+
+            let result = accept_future.await;
+
+            match result {
+                Ok(Ok(())) => {
+                    if sender.send(Ok(stream)).await.is_err() && debug {
+                        eprintln!("detect closed connection channel");
+                    }
+                }
+                Ok(Err(err)) => {
+                    if debug {
+                        eprintln!("https handshake failed - {}", err);
+                    }
+                }
+                Err(_) => {
+                    if debug {
+                        eprintln!("https handshake timeout");
+                    }
+                }
+            }
+
+            drop(accept_counter); // decrease reference count
+        });
     }
 }
 
-- 
2.20.1

[pbs-devel] [PATCH backup 3/7] proxy: Arc usage cleanup

[Wolfgang Bumiller]

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
---
 src/bin/proxmox-backup-proxy.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index b28ab75b..0b4b5586 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -231,7 +231,7 @@ async fn accept_connection(
             continue;
         }
 
-        let accept_counter = accept_counter.clone();
+        let accept_counter = Arc::clone(&accept_counter);
         tokio::spawn(async move {
             let accept_future = tokio::time::timeout(
                 Duration::new(10, 0), stream.as_mut().accept());
-- 
2.20.1

[pbs-devel] [PATCH backup 4/7] proxy: factor out tls acceptor creation

[Wolfgang Bumiller]

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

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index 0b4b5586..c05bcd20 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -113,15 +113,11 @@ async fn run() -> Result<(), Error> {
     let rest_server = RestServer::new(config);
 
     //openssl req -x509 -newkey rsa:4096 -keyout /etc/proxmox-backup/proxy.key -out /etc/proxmox-backup/proxy.pem -nodes
-    let key_path = configdir!("/proxy.key");
-    let cert_path = configdir!("/proxy.pem");
 
-    let mut acceptor = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls()).unwrap();
-    acceptor.set_private_key_file(key_path, SslFiletype::PEM)
-        .map_err(|err| format_err!("unable to read proxy key {} - {}", key_path, err))?;
-    acceptor.set_certificate_chain_file(cert_path)
-        .map_err(|err| format_err!("unable to read proxy cert {} - {}", cert_path, err))?;
-    acceptor.check_private_key().unwrap();
+    // 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!
+    let acceptor = make_tls_acceptor()?;
 
     let acceptor = Arc::new(acceptor.build());
 
@@ -170,6 +166,20 @@ async fn run() -> Result<(), Error> {
     Ok(())
 }
 
+fn make_tls_acceptor() -> Result<Arc<SslAcceptor>, Error> {
+    let key_path = configdir!("/proxy.key");
+    let cert_path = configdir!("/proxy.pem");
+
+    let mut acceptor = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls()).unwrap();
+    acceptor.set_private_key_file(key_path, SslFiletype::PEM)
+        .map_err(|err| format_err!("unable to read proxy key {} - {}", key_path, err))?;
+    acceptor.set_certificate_chain_file(cert_path)
+        .map_err(|err| format_err!("unable to read proxy cert {} - {}", cert_path, err))?;
+    acceptor.check_private_key().unwrap();
+
+    Ok(Arc::new(acceptor.build()))
+}
+
 type ClientStreamResult =
     Result<std::pin::Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>, Error>;
 const MAX_PENDING_ACCEPTS: usize = 1024;
-- 
2.20.1

[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

[pbs-devel] [PATCH backup 6/7] refactor send_command

[Wolfgang Bumiller]

- refactor the combinators,
- make it take a `&T: Serialize` instead of a Value, and
  allow sending the raw string via `send_raw_command`.

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
---
 src/bin/proxmox-backup-proxy.rs |  8 +---
 src/server/command_socket.rs    | 71 ++++++++++++++++++---------------
 src/server/worker_task.rs       |  4 +-
 3 files changed, 42 insertions(+), 41 deletions(-)

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index 793ba67d..fc773459 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -750,15 +750,11 @@ async fn command_reopen_logfiles() -> Result<(), Error> {
     // only care about the most recent daemon instance for each, proxy & api, as other older ones
     // should not respond to new requests anyway, but only finish their current one and then exit.
     let sock = server::our_ctrl_sock();
-    let f1 = server::send_command(sock, serde_json::json!({
-        "command": "api-access-log-reopen",
-    }));
+    let f1 = server::send_command(sock, "{\"command\":\"api-access-log-reopen\"}\n");
 
     let pid = server::read_pid(buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
     let sock = server::ctrl_sock_from_pid(pid);
-    let f2 = server::send_command(sock, serde_json::json!({
-        "command": "api-access-log-reopen",
-    }));
+    let f2 = server::send_command(sock, "{\"command\":\"api-access-log-reopen\"}\n");
 
     match futures::join!(f1, f2) {
         (Err(e1), Err(e2)) => Err(format_err!("reopen commands failed, proxy: {}; api: {}", e1, e2)),
diff --git a/src/server/command_socket.rs b/src/server/command_socket.rs
index 89c77585..af41dd16 100644
--- a/src/server/command_socket.rs
+++ b/src/server/command_socket.rs
@@ -2,11 +2,12 @@ use anyhow::{bail, format_err, Error};
 
 use std::collections::HashMap;
 use std::os::unix::io::AsRawFd;
-use std::path::PathBuf;
+use std::path::{PathBuf, Path};
 use std::sync::Arc;
 
 use futures::*;
 use tokio::net::UnixListener;
+use serde::Serialize;
 use serde_json::Value;
 use nix::sys::socket;
 
@@ -102,43 +103,47 @@ where
 }
 
 
-pub async fn send_command<P>(
-    path: P,
-    params: Value
-) -> Result<Value, Error>
-    where P: Into<PathBuf>,
+pub async fn send_command<P, T>(path: P, params: &T) -> Result<Value, Error>
+where
+    P: AsRef<Path>,
+    T: ?Sized + Serialize,
 {
-    let path: PathBuf = path.into();
+    let mut command_string = serde_json::to_string(params)?;
+    command_string.push('\n');
+    send_raw_command(path.as_ref(), &command_string).await
+}
 
-    tokio::net::UnixStream::connect(path)
-        .map_err(move |err| format_err!("control socket connect failed - {}", err))
-        .and_then(move |mut conn| {
+pub async fn send_raw_command<P>(path: P, command_string: &str) -> Result<Value, Error>
+where
+    P: AsRef<Path>,
+{
+    use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
 
-            let mut command_string = params.to_string();
-            command_string.push('\n');
+    let mut conn = tokio::net::UnixStream::connect(path)
+        .map_err(move |err| format_err!("control socket connect failed - {}", err))
+        .await?;
 
-            async move {
-                use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
+    conn.write_all(command_string.as_bytes()).await?;
+    if !command_string.as_bytes().ends_with(b"\n") {
+        conn.write_all(b"\n").await?;
+    }
 
-                conn.write_all(command_string.as_bytes()).await?;
-                AsyncWriteExt::shutdown(&mut conn).await?;
-                let mut rx = tokio::io::BufReader::new(conn);
-                let mut data = String::new();
-                if rx.read_line(&mut data).await? == 0 {
-                    bail!("no response");
-                }
-                if let Some(res) = data.strip_prefix("OK: ") {
-                    match res.parse::<Value>() {
-                        Ok(v) => Ok(v),
-                        Err(err) => bail!("unable to parse json response - {}", err),
-                    }
-                } else if let Some(err) = data.strip_prefix("ERROR: ") {
-                    bail!("{}", err);
-                } else {
-                    bail!("unable to parse response: {}", data);
-                }
-            }
-        }).await
+    AsyncWriteExt::shutdown(&mut conn).await?;
+    let mut rx = tokio::io::BufReader::new(conn);
+    let mut data = String::new();
+    if rx.read_line(&mut data).await? == 0 {
+        bail!("no response");
+    }
+    if let Some(res) = data.strip_prefix("OK: ") {
+        match res.parse::<Value>() {
+            Ok(v) => Ok(v),
+            Err(err) => bail!("unable to parse json response - {}", err),
+        }
+    } else if let Some(err) = data.strip_prefix("ERROR: ") {
+        bail!("{}", err);
+    } else {
+        bail!("unable to parse response: {}", data);
+    }
 }
 
 /// A callback for a specific commando socket.
diff --git a/src/server/worker_task.rs b/src/server/worker_task.rs
index 6c5456c9..84019fef 100644
--- a/src/server/worker_task.rs
+++ b/src/server/worker_task.rs
@@ -59,7 +59,7 @@ pub async fn worker_is_active(upid: &UPID) -> Result<bool, Error> {
             "upid": upid.to_string(),
         },
     });
-    let status = super::send_command(sock, cmd).await?;
+    let status = super::send_command(sock, &cmd).await?;
 
     if let Some(active) = status.as_bool() {
         Ok(active)
@@ -133,7 +133,7 @@ pub async fn abort_worker(upid: UPID) -> Result<(), Error> {
             "upid": upid.to_string(),
         },
     });
-    super::send_command(sock, cmd).map_ok(|_| ()).await
+    super::send_command(sock, &cmd).map_ok(|_| ()).await
 }
 
 fn parse_worker_status_line(line: &str) -> Result<(String, UPID, Option<TaskState>), Error> {
-- 
2.20.1

[pbs-devel] [PATCH backup 7/7] hot-reload proxy certificate when updating via the API

[Wolfgang Bumiller]

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
---
 src/api2/node/certificates.rs | 26 +++++++++++++-------------
 src/config.rs                 | 17 ++---------------
 src/server.rs                 |  9 +++++++++
 3 files changed, 24 insertions(+), 28 deletions(-)

diff --git a/src/api2/node/certificates.rs b/src/api2/node/certificates.rs
index e6ad59b3..79df5d0f 100644
--- a/src/api2/node/certificates.rs
+++ b/src/api2/node/certificates.rs
@@ -175,12 +175,13 @@ pub fn get_info() -> Result<Vec<CertificateInfo>, Error> {
             node: { schema: NODE_SCHEMA },
             certificates: { description: "PEM encoded certificate (chain)." },
             key: { description: "PEM encoded private key." },
+            // FIXME: widget-toolkit should have an option to disable using these 2 parameters...
             restart: {
-                description: "Restart proxmox-backup-proxy",
+                description: "UI compatibility parameter, ignored",
+                type: Boolean,
                 optional: true,
                 default: false,
             },
-            // FIXME: widget-toolkit should have an option to disable using this parameter...
             force: {
                 description: "Force replacement of existing files.",
                 type: Boolean,
@@ -200,10 +201,9 @@ pub fn get_info() -> Result<Vec<CertificateInfo>, Error> {
     protected: true,
 )]
 /// Upload a custom certificate.
-pub fn upload_custom_certificate(
+pub async fn upload_custom_certificate(
     certificates: String,
     key: String,
-    restart: bool,
 ) -> Result<Vec<CertificateInfo>, Error> {
     let certificates = X509::stack_from_pem(certificates.as_bytes())
         .map_err(|err| format_err!("failed to decode certificate chain: {}", err))?;
@@ -223,7 +223,8 @@ pub fn upload_custom_certificate(
 
     let key = key.private_key_to_pem_pkcs8()?;
 
-    crate::config::set_proxy_certificate(&certificates, &key, restart)?;
+    crate::config::set_proxy_certificate(&certificates, &key)?;
+    crate::server::reload_proxy_certificate().await?;
 
     get_info()
 }
@@ -233,7 +234,8 @@ pub fn upload_custom_certificate(
         properties: {
             node: { schema: NODE_SCHEMA },
             restart: {
-                description: "Restart proxmox-backup-proxy",
+                description: "UI compatibility parameter, ignored",
+                type: Boolean,
                 optional: true,
                 default: false,
             },
@@ -245,7 +247,7 @@ pub fn upload_custom_certificate(
     protected: true,
 )]
 /// Delete the current certificate and regenerate a self signed one.
-pub fn delete_custom_certificate(restart: bool) -> Result<(), Error> {
+pub async fn delete_custom_certificate() -> Result<(), Error> {
     let cert_path = configdir!("/proxy.pem");
     // Here we fail since if this fails nothing else breaks anyway
     std::fs::remove_file(&cert_path)
@@ -263,10 +265,7 @@ pub fn delete_custom_certificate(restart: bool) -> Result<(), Error> {
     }
 
     crate::config::update_self_signed_cert(true)?;
-
-    if restart {
-        crate::config::reload_proxy()?;
-    }
+    crate::server::reload_proxy_certificate().await?;
 
     Ok(())
 }
@@ -535,7 +534,8 @@ fn spawn_certificate_worker(
 
     WorkerTask::spawn(name, None, auth_id, true, move |worker| async move {
         if let Some(cert) = order_certificate(worker, &node_config).await? {
-            crate::config::set_proxy_certificate(&cert.certificate, &cert.private_key_pem, true)?;
+            crate::config::set_proxy_certificate(&cert.certificate, &cert.private_key_pem)?;
+            crate::server::reload_proxy_certificate().await?;
         }
         Ok(())
     })
@@ -572,7 +572,7 @@ pub fn revoke_acme_cert(rpcenv: &mut dyn RpcEnvironment) -> Result<String, Error
             worker.log("Revoking old certificate");
             acme.revoke_certificate(cert_pem.as_bytes(), None).await?;
             worker.log("Deleting certificate and regenerating a self-signed one");
-            delete_custom_certificate(true)?;
+            delete_custom_certificate().await?;
             Ok(())
         },
     )
diff --git a/src/config.rs b/src/config.rs
index 22c293c9..b9cd6281 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -187,16 +187,12 @@ pub fn update_self_signed_cert(force: bool) -> Result<(), Error> {
     let x509 = x509.build();
     let cert_pem = x509.to_pem()?;
 
-    set_proxy_certificate(&cert_pem, &priv_pem, false)?;
+    set_proxy_certificate(&cert_pem, &priv_pem)?;
 
     Ok(())
 }
 
-pub(crate) fn set_proxy_certificate(
-    cert_pem: &[u8],
-    key_pem: &[u8],
-    reload: bool,
-) -> Result<(), Error> {
+pub(crate) fn set_proxy_certificate(cert_pem: &[u8], key_pem: &[u8]) -> Result<(), Error> {
     let backup_user = crate::backup::backup_user()?;
     let options = CreateOptions::new()
         .perm(Mode::from_bits_truncate(0o0640))
@@ -211,14 +207,5 @@ pub(crate) fn set_proxy_certificate(
     replace_file(&cert_path, &cert_pem, options)
         .map_err(|err| format_err!("error writing certificate file - {}", err))?;
 
-    if reload {
-        reload_proxy()?;
-    }
-
     Ok(())
 }
-
-pub(crate) fn reload_proxy() -> Result<(), Error> {
-    crate::tools::systemd::reload_unit("proxmox-backup-proxy")
-        .map_err(|err| format_err!("error signaling reload to pbs proxy: {}", err))
-}
diff --git a/src/server.rs b/src/server.rs
index b6a37b92..ba25617d 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -7,6 +7,7 @@
 use anyhow::{format_err, Error};
 use lazy_static::lazy_static;
 use nix::unistd::Pid;
+use serde_json::Value;
 
 use proxmox::sys::linux::procfs::PidStat;
 
@@ -91,3 +92,11 @@ pub use report::*;
 pub mod ticket;
 
 pub mod auth;
+
+pub(crate) async fn reload_proxy_certificate() -> Result<(), Error> {
+    let proxy_pid = crate::server::read_pid(buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
+    let sock = crate::server::ctrl_sock_from_pid(proxy_pid);
+    let _: Value = crate::server::send_raw_command(sock, "{\"command\":\"reload-certificate\"}\n")
+        .await?;
+    Ok(())
+}
-- 
2.20.1

[pbs-devel] applied-series: [PATCH backup 0/7] hot-reload proxy certificates

[Thomas Lamprecht]

On 11.05.21 15:53, Wolfgang Bumiller wrote:
> This adds the ability to tell a running proxy to just reload the TLS
> cert certificates via the command-socket.
> 
> Starts off with some cleanup/refactoring to get rid of all that heavy
> indentation...
> 
> Wolfgang Bumiller (7):
>   proxy: factor out accept_connection
>   proxy: "continue on error" for the accept call, too
>   proxy: Arc usage cleanup
>   proxy: factor out tls acceptor creation
>   proxy: implement 'reload-certificate' command
>   refactor send_command
>   hot-reload proxy certificate when updating via the API
> 
>  src/api2/node/certificates.rs   |  26 ++--
>  src/bin/proxmox-backup-proxy.rs | 220 ++++++++++++++++++++------------
>  src/config.rs                   |  17 +--
>  src/server.rs                   |   9 ++
>  src/server/command_socket.rs    |  71 ++++++-----
>  src/server/worker_task.rs       |   4 +-
>  6 files changed, 204 insertions(+), 143 deletions(-)
> 

applied whole series, thanks!

I followed this up with:
* fallback to "default" account on order (check commit message, was really confusing else)
* add UI task-description entries for acme related tasks, mostly based on the ones from
  PVE, but as the worker type is spelled slightly different anyway here I took the chance
  to use a slightly nicer version there too.
* set account name as worker ID so that they can be used in by the task-descriptions
  - note: all account actions done before that commit miss it and will display "default",
    even if the account was named something else, I ignored this as we did not really
    rolled this out externally yet.

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

[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

[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