[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.