move more tools for the client into subcrates

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>

pbs-tools/Cargo.toml

@@ -9,6 +9,9 @@ description = "common tools used throughout pbs"
 [dependencies]
 anyhow = "1.0"
 base64 = "0.12"
+foreign-types = "0.3"
+futures = "0.3"
+lazy_static = "1.4"
 libc = "0.2"
 nix = "0.19.1"
 nom = "5.1"
@@ -17,6 +20,10 @@ percent-encoding = "2.1"
 regex = "1.2"
 serde = "1.0"
 serde_json = "1.0"
+# rt-multi-thread is required for block_in_place
+tokio = { version = "1.6", features = [ "rt", "rt-multi-thread", "sync" ] }
 url = "2.1"
 
 proxmox = { version = "0.11.5", default-features = false, features = [] }
+
+pbs-buildcfg = { path = "../pbs-buildcfg" }

pbs-tools/src/auth.rs

@@ -0,0 +1,26 @@
+//! Helpers for authentication used by both client and server.
+
+use anyhow::Error;
+use lazy_static::lazy_static;
+use openssl::pkey::{PKey, Private};
+use openssl::rsa::Rsa;
+
+use proxmox::tools::fs::file_get_contents;
+
+use pbs_buildcfg::configdir;
+
+fn load_private_auth_key() -> Result<PKey<Private>, Error> {
+    let pem = file_get_contents(configdir!("/authkey.key"))?;
+    let rsa = Rsa::private_key_from_pem(&pem)?;
+    let key = PKey::from_rsa(rsa)?;
+
+    Ok(key)
+}
+
+pub fn private_auth_key() -> &'static PKey<Private> {
+    lazy_static! {
+        static ref KEY: PKey<Private> = load_private_auth_key().unwrap();
+    }
+
+    &KEY
+}

pbs-tools/src/broadcast_future.rs

@@ -0,0 +1,180 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::{Arc, Mutex};
+
+use anyhow::{format_err, Error};
+use futures::future::{FutureExt, TryFutureExt};
+use tokio::sync::oneshot;
+
+/// Broadcast results to registered listeners using asnyc oneshot channels
+#[derive(Default)]
+pub struct BroadcastData<T> {
+    result: Option<Result<T, String>>,
+    listeners: Vec<oneshot::Sender<Result<T, Error>>>,
+}
+
+impl <T: Clone> BroadcastData<T> {
+
+    pub fn new() -> Self {
+        Self {
+            result: None,
+            listeners: vec![],
+        }
+    }
+
+    pub fn notify_listeners(&mut self, result: Result<T, String>) {
+
+        self.result = Some(result.clone());
+
+        loop {
+            match self.listeners.pop() {
+                None => { break; },
+                Some(ch) => {
+                    match &result {
+                        Ok(result) => { let _ = ch.send(Ok(result.clone())); },
+                        Err(err) => { let _ = ch.send(Err(format_err!("{}", err))); },
+                    }
+                },
+            }
+        }
+    }
+
+    pub fn listen(&mut self) -> impl Future<Output = Result<T, Error>> {
+        use futures::future::{ok, Either};
+
+        match &self.result {
+            None => {},
+            Some(Ok(result)) => return Either::Left(ok(result.clone())),
+            Some(Err(err)) => return Either::Left(futures::future::err(format_err!("{}", err))),
+        }
+
+        let (tx, rx) = oneshot::channel::<Result<T, Error>>();
+
+        self.listeners.push(tx);
+
+        Either::Right(rx
+            .map(|res| match res {
+                Ok(Ok(t)) => Ok(t),
+                Ok(Err(e)) => Err(e),
+                Err(e) => Err(Error::from(e)),
+            })
+        )
+    }
+}
+
+type SourceFuture<T> = Pin<Box<dyn Future<Output = Result<T, Error>> + Send>>;
+
+struct BroadCastFutureBinding<T> {
+    broadcast: BroadcastData<T>,
+    future: Option<SourceFuture<T>>,
+}
+
+/// Broadcast future results to registered listeners
+pub struct BroadcastFuture<T> {
+    inner: Arc<Mutex<BroadCastFutureBinding<T>>>,
+}
+
+impl<T: Clone + Send + 'static> BroadcastFuture<T> {
+    /// Create instance for specified source future.
+    ///
+    /// The result of the future is sent to all registered listeners.
+    pub fn new(source: Box<dyn Future<Output = Result<T, Error>> + Send>) -> Self {
+        let inner = BroadCastFutureBinding {
+            broadcast: BroadcastData::new(),
+            future: Some(Pin::from(source)),
+        };
+        Self { inner: Arc::new(Mutex::new(inner)) }
+    }
+
+    /// Creates a new instance with a oneshot channel as trigger
+    pub fn new_oneshot() -> (Self, oneshot::Sender<Result<T, Error>>) {
+
+        let (tx, rx) = oneshot::channel::<Result<T, Error>>();
+        let rx = rx
+            .map_err(Error::from)
+            .and_then(futures::future::ready);
+
+        (Self::new(Box::new(rx)), tx)
+    }
+
+    fn notify_listeners(
+        inner: Arc<Mutex<BroadCastFutureBinding<T>>>,
+        result: Result<T, String>,
+    ) {
+        let mut data = inner.lock().unwrap();
+        data.broadcast.notify_listeners(result);
+    }
+
+    fn spawn(inner: Arc<Mutex<BroadCastFutureBinding<T>>>) -> impl Future<Output = Result<T, Error>> {
+        let mut data = inner.lock().unwrap();
+
+        if let Some(source) = data.future.take() {
+
+            let inner1 = inner.clone();
+
+            let task = source.map(move |value| {
+                match value {
+                    Ok(value) => Self::notify_listeners(inner1, Ok(value)),
+                    Err(err) => Self::notify_listeners(inner1, Err(err.to_string())),
+                }
+            });
+            tokio::spawn(task);
+        }
+
+        data.broadcast.listen()
+    }
+
+    /// Register a listener
+    pub fn listen(&self) -> impl Future<Output = Result<T, Error>> {
+        let inner2 = self.inner.clone();
+        async move { Self::spawn(inner2).await }
+    }
+}
+
+#[test]
+fn test_broadcast_future() {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    static CHECKSUM: AtomicUsize  = AtomicUsize::new(0);
+
+    let (sender, trigger) = BroadcastFuture::new_oneshot();
+
+    let receiver1 = sender.listen()
+        .map_ok(|res| {
+            CHECKSUM.fetch_add(res, Ordering::SeqCst);
+        })
+        .map_err(|err| { panic!("got error {}", err); })
+        .map(|_| ());
+
+    let receiver2 = sender.listen()
+        .map_ok(|res| {
+            CHECKSUM.fetch_add(res*2, Ordering::SeqCst);
+        })
+        .map_err(|err| { panic!("got error {}", err); })
+        .map(|_| ());
+
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    rt.block_on(async move {
+        let r1 = tokio::spawn(receiver1);
+        let r2 = tokio::spawn(receiver2);
+
+        trigger.send(Ok(1)).unwrap();
+        let _ = r1.await;
+        let _ = r2.await;
+    });
+
+    let result = CHECKSUM.load(Ordering::SeqCst);
+
+    assert_eq!(result, 3);
+
+    // the result stays available until the BroadcastFuture is dropped
+    rt.block_on(sender.listen()
+        .map_ok(|res| {
+            CHECKSUM.fetch_add(res*4, Ordering::SeqCst);
+        })
+        .map_err(|err| { panic!("got error {}", err); })
+        .map(|_| ()));
+
+    let result = CHECKSUM.load(Ordering::SeqCst);
+    assert_eq!(result, 7);
+}

pbs-tools/src/cert.rs

@@ -0,0 +1,104 @@
+use std::path::PathBuf;
+use std::mem::MaybeUninit;
+
+use anyhow::{bail, format_err, Error};
+use foreign_types::ForeignTypeRef;
+use openssl::x509::{X509, GeneralName};
+use openssl::stack::Stack;
+use openssl::pkey::{Public, PKey};
+
+use pbs_buildcfg::configdir;
+
+// C type:
+#[allow(non_camel_case_types)]
+type ASN1_TIME = <openssl::asn1::Asn1TimeRef as ForeignTypeRef>::CType;
+
+extern "C" {
+    fn ASN1_TIME_to_tm(s: *const ASN1_TIME, tm: *mut libc::tm) -> libc::c_int;
+}
+
+fn asn1_time_to_unix(time: &openssl::asn1::Asn1TimeRef) -> Result<i64, Error> {
+    let mut c_tm = MaybeUninit::<libc::tm>::uninit();
+    let rc = unsafe { ASN1_TIME_to_tm(time.as_ptr(), c_tm.as_mut_ptr()) };
+    if rc != 1 {
+        bail!("failed to parse ASN1 time");
+    }
+    let mut c_tm = unsafe { c_tm.assume_init() };
+    proxmox::tools::time::timegm(&mut c_tm)
+}
+
+pub struct CertInfo {
+    x509: X509,
+}
+
+fn x509name_to_string(name: &openssl::x509::X509NameRef) -> Result<String, Error> {
+    let mut parts = Vec::new();
+    for entry in name.entries() {
+        parts.push(format!("{} = {}", entry.object().nid().short_name()?, entry.data().as_utf8()?));
+    }
+    Ok(parts.join(", "))
+}
+
+impl CertInfo {
+    pub fn new() -> Result<Self, Error> {
+        Self::from_path(PathBuf::from(configdir!("/proxy.pem")))
+    }
+
+    pub fn from_path(path: PathBuf) -> Result<Self, Error> {
+        Self::from_pem(&proxmox::tools::fs::file_get_contents(&path)?)
+            .map_err(|err| format_err!("failed to load certificate from {:?} - {}", path, err))
+    }
+
+    pub fn from_pem(cert_pem: &[u8]) -> Result<Self, Error> {
+        let x509 = openssl::x509::X509::from_pem(&cert_pem)?;
+        Ok(Self{
+            x509
+        })
+    }
+
+    pub fn subject_alt_names(&self) -> Option<Stack<GeneralName>> {
+        self.x509.subject_alt_names()
+    }
+
+    pub fn subject_name(&self) -> Result<String, Error> {
+        Ok(x509name_to_string(self.x509.subject_name())?)
+    }
+
+    pub fn issuer_name(&self) -> Result<String, Error> {
+        Ok(x509name_to_string(self.x509.issuer_name())?)
+    }
+
+    pub fn fingerprint(&self) -> Result<String, Error> {
+        let fp = self.x509.digest(openssl::hash::MessageDigest::sha256())?;
+        let fp_string = proxmox::tools::digest_to_hex(&fp);
+        let fp_string = fp_string.as_bytes().chunks(2).map(|v| std::str::from_utf8(v).unwrap())
+            .collect::<Vec<&str>>().join(":");
+        Ok(fp_string)
+    }
+
+    pub fn public_key(&self) -> Result<PKey<Public>, Error> {
+        let pubkey = self.x509.public_key()?;
+        Ok(pubkey)
+    }
+
+    pub fn not_before(&self) -> &openssl::asn1::Asn1TimeRef {
+        self.x509.not_before()
+    }
+
+    pub fn not_after(&self) -> &openssl::asn1::Asn1TimeRef {
+        self.x509.not_after()
+    }
+
+    pub fn not_before_unix(&self) -> Result<i64, Error> {
+        asn1_time_to_unix(&self.not_before())
+    }
+
+    pub fn not_after_unix(&self) -> Result<i64, Error> {
+        asn1_time_to_unix(&self.not_after())
+    }
+
+    /// Check if the certificate is expired at or after a specific unix epoch.
+    pub fn is_expired_after_epoch(&self, epoch: i64) -> Result<bool, Error> {
+        Ok(self.not_after_unix()? < epoch)
+    }
+}

pbs-tools/src/lib.rs

@@ -1,12 +1,18 @@
+pub mod auth;
 pub mod borrow;
+pub mod broadcast_future;
+pub mod cert;
 pub mod format;
 pub mod fs;
 pub mod json;
 pub mod nom;
+pub mod percent_encoding;
 pub mod process_locker;
 pub mod sha;
 pub mod str;
+pub mod sync;
 pub mod ticket;
+pub mod tokio;
 
 mod command;
 pub use command::{command_output, command_output_as_string, run_command};

pbs-tools/src/percent_encoding.rs

@@ -0,0 +1,22 @@
+use percent_encoding::{utf8_percent_encode, AsciiSet};
+
+/// This used to be: `SIMPLE_ENCODE_SET` plus space, `"`, `#`, `<`, `>`, backtick, `?`, `{`, `}`
+pub const DEFAULT_ENCODE_SET: &AsciiSet = &percent_encoding::CONTROLS // 0..1f and 7e
+    // The SIMPLE_ENCODE_SET adds space and anything >= 0x7e (7e itself is already included above)
+    .add(0x20)
+    .add(0x7f)
+    // the DEFAULT_ENCODE_SET added:
+    .add(b' ')
+    .add(b'"')
+    .add(b'#')
+    .add(b'<')
+    .add(b'>')
+    .add(b'`')
+    .add(b'?')
+    .add(b'{')
+    .add(b'}');
+
+/// percent encode a url component
+pub fn percent_encode_component(comp: &str) -> String {
+    utf8_percent_encode(comp, percent_encoding::NON_ALPHANUMERIC).to_string()
+}

pbs-tools/src/sync/mod.rs

@@ -0,0 +1,2 @@
+mod std_channel_writer;
+pub use std_channel_writer::StdChannelWriter;

pbs-tools/src/sync/std_channel_writer.rs

@@ -0,0 +1,28 @@
+use std::io::Write;
+use std::sync::mpsc::SyncSender;
+
+use anyhow::{Error};
+
+/// Wrapper around SyncSender, which implements Write
+///
+/// Each write in translated into a send(Vec<u8>).
+pub struct StdChannelWriter(SyncSender<Result<Vec<u8>, Error>>);
+
+impl StdChannelWriter {
+    pub fn new(sender: SyncSender<Result<Vec<u8>, Error>>) -> Self {
+        Self(sender)
+    }
+}
+
+impl Write for StdChannelWriter {
+    fn write(&mut self, buf: &[u8]) -> Result<usize, std::io::Error> {
+        self.0
+            .send(Ok(buf.to_vec()))
+            .map_err(proxmox::sys::error::io_err_other)
+            .and(Ok(buf.len()))
+    }
+
+    fn flush(&mut self) -> Result<(), std::io::Error> {
+        Ok(())
+    }
+}

pbs-tools/src/tokio/mod.rs

@@ -0,0 +1,2 @@
+pub mod tokio_writer_adapter;
+pub use tokio_writer_adapter::TokioWriterAdapter;

pbs-tools/src/tokio/tokio_writer_adapter.rs

@@ -0,0 +1,26 @@
+use std::io::Write;
+
+use tokio::task::block_in_place;
+
+/// Wrapper around a writer which implements Write
+///
+/// wraps each write with a 'block_in_place' so that
+/// any (blocking) writer can be safely used in async context in a
+/// tokio runtime
+pub struct TokioWriterAdapter<W: Write>(W);
+
+impl<W: Write> TokioWriterAdapter<W> {
+    pub fn new(writer: W) -> Self {
+        Self(writer)
+    }
+}
+
+impl<W: Write> Write for TokioWriterAdapter<W> {
+    fn write(&mut self, buf: &[u8]) -> Result<usize, std::io::Error> {
+        block_in_place(|| self.0.write(buf))
+    }
+
+    fn flush(&mut self) -> Result<(), std::io::Error> {
+        block_in_place(|| self.0.flush())
+    }
+}