wip: implemented crypto reloader, as separated object from proxy itself

rpxy-lib/src/backend/backend_main.rs

@@ -1,5 +1,5 @@
 use crate::{
-  certs::CryptoSource,
+  crypto::CryptoSource,
   error::*,
   log::*,
   name_exp::{ByteName, ServerName},

rpxy-lib/src/backend/upstream.rs

@@ -7,7 +7,7 @@ use super::load_balance::{
 // use super::{BytesName, LbContext, PathNameBytesExp, UpstreamOption};
 use super::upstream_opts::UpstreamOption;
 use crate::{
-  certs::CryptoSource,
+  crypto::CryptoSource,
   error::RpxyError,
   globals::{AppConfig, UpstreamUri},
   log::*,

rpxy-lib/src/certs.rs

@@ -1,22 +0,0 @@
-use async_trait::async_trait;
-use rustls::{Certificate, PrivateKey};
-
-#[async_trait]
-// Trait to read certs and keys anywhere from KVS, file, sqlite, etc.
-pub trait CryptoSource {
-  type Error;
-
-  /// read crypto materials from source
-  async fn read(&self) -> Result<CertsAndKeys, Self::Error>;
-
-  /// Returns true when mutual tls is enabled
-  fn is_mutual_tls(&self) -> bool;
-}
-
-/// Certificates and private keys in rustls loaded from files
-#[derive(Debug, PartialEq, Eq, Clone)]
-pub struct CertsAndKeys {
-  pub certs: Vec<Certificate>,
-  pub cert_keys: Vec<PrivateKey>,
-  pub client_ca_certs: Option<Vec<Certificate>>,
-}

rpxy-lib/src/crypto/certs.rs

@@ -0,0 +1,91 @@
+use async_trait::async_trait;
+use rustc_hash::FxHashSet as HashSet;
+use rustls::{
+  sign::{any_supported_type, CertifiedKey},
+  Certificate, OwnedTrustAnchor, PrivateKey,
+};
+use std::io;
+use x509_parser::prelude::*;
+
+#[async_trait]
+// Trait to read certs and keys anywhere from KVS, file, sqlite, etc.
+pub trait CryptoSource {
+  type Error;
+
+  /// read crypto materials from source
+  async fn read(&self) -> Result<CertsAndKeys, Self::Error>;
+
+  /// Returns true when mutual tls is enabled
+  fn is_mutual_tls(&self) -> bool;
+}
+
+/// Certificates and private keys in rustls loaded from files
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct CertsAndKeys {
+  pub certs: Vec<Certificate>,
+  pub cert_keys: Vec<PrivateKey>,
+  pub client_ca_certs: Option<Vec<Certificate>>,
+}
+
+impl CertsAndKeys {
+  pub fn parse_server_certs_and_keys(&self) -> Result<CertifiedKey, anyhow::Error> {
+    // for (server_name_bytes_exp, certs_and_keys) in self.inner.iter() {
+    let signing_key = self
+      .cert_keys
+      .iter()
+      .find_map(|k| {
+        if let Ok(sk) = any_supported_type(k) {
+          Some(sk)
+        } else {
+          None
+        }
+      })
+      .ok_or_else(|| {
+        io::Error::new(
+          io::ErrorKind::InvalidInput,
+          "Unable to find a valid certificate and key",
+        )
+      })?;
+    Ok(CertifiedKey::new(self.certs.clone(), signing_key))
+  }
+
+  pub fn parse_client_ca_certs(&self) -> Result<(Vec<OwnedTrustAnchor>, HashSet<Vec<u8>>), anyhow::Error> {
+    let certs = self.client_ca_certs.as_ref().ok_or(anyhow::anyhow!("No client cert"))?;
+
+    let owned_trust_anchors: Vec<_> = certs
+      .iter()
+      .map(|v| {
+        // let trust_anchor = tokio_rustls::webpki::TrustAnchor::try_from_cert_der(&v.0).unwrap();
+        let trust_anchor = webpki::TrustAnchor::try_from_cert_der(&v.0).unwrap();
+        rustls::OwnedTrustAnchor::from_subject_spki_name_constraints(
+          trust_anchor.subject,
+          trust_anchor.spki,
+          trust_anchor.name_constraints,
+        )
+      })
+      .collect();
+
+    // TODO: SKID is not used currently
+    let subject_key_identifiers: HashSet<_> = certs
+      .iter()
+      .filter_map(|v| {
+        // retrieve ca key id (subject key id)
+        let cert = parse_x509_certificate(&v.0).unwrap().1;
+        let subject_key_ids = cert
+          .iter_extensions()
+          .filter_map(|ext| match ext.parsed_extension() {
+            ParsedExtension::SubjectKeyIdentifier(skid) => Some(skid),
+            _ => None,
+          })
+          .collect::<Vec<_>>();
+        if !subject_key_ids.is_empty() {
+          Some(subject_key_ids[0].0.to_owned())
+        } else {
+          None
+        }
+      })
+      .collect();
+
+    Ok((owned_trust_anchors, subject_key_identifiers))
+  }
+}

rpxy-lib/src/crypto/mod.rs

@@ -0,0 +1,36 @@
+mod certs;
+mod service;
+
+use crate::{
+  backend::BackendAppManager,
+  constants::{CERTS_WATCH_DELAY_SECS, LOAD_CERTS_ONLY_WHEN_UPDATED},
+  error::RpxyResult,
+};
+use hot_reload::{ReloaderReceiver, ReloaderService};
+use service::CryptoReloader;
+use std::sync::Arc;
+
+pub use certs::{CertsAndKeys, CryptoSource};
+pub use service::ServerCryptoBase;
+
+/// Result type inner of certificate reloader service
+type ReloaderServiceResultInner<T> = (
+  ReloaderService<CryptoReloader<T>, ServerCryptoBase>,
+  ReloaderReceiver<ServerCryptoBase>,
+);
+/// Build certificate reloader service
+pub(crate) async fn build_cert_reloader<T>(
+  app_manager: &Arc<BackendAppManager<T>>,
+) -> RpxyResult<ReloaderServiceResultInner<T>>
+where
+  T: CryptoSource + Clone + Send + Sync + 'static,
+{
+  let (cert_reloader_service, cert_reloader_rx) = ReloaderService::<
+    service::CryptoReloader<T>,
+    service::ServerCryptoBase,
+  >::new(
+    app_manager, CERTS_WATCH_DELAY_SECS, !LOAD_CERTS_ONLY_WHEN_UPDATED
+  )
+  .await?;
+  Ok((cert_reloader_service, cert_reloader_rx))
+}

rpxy-lib/src/crypto/service.rs

@@ -0,0 +1,272 @@
+use super::certs::{CertsAndKeys, CryptoSource};
+use crate::{backend::BackendAppManager, log::*, name_exp::ServerName};
+use async_trait::async_trait;
+use hot_reload::*;
+use rustc_hash::FxHashMap as HashMap;
+use rustls::{server::ResolvesServerCertUsingSni, sign::CertifiedKey, RootCertStore, ServerConfig};
+use std::sync::Arc;
+
+#[derive(Clone)]
+/// Reloader service for certificates and keys for TLS
+pub struct CryptoReloader<T>
+where
+  T: CryptoSource,
+{
+  inner: Arc<BackendAppManager<T>>,
+}
+
+/// SNI to ServerConfig map type
+pub type SniServerCryptoMap = HashMap<ServerName, Arc<ServerConfig>>;
+/// SNI to ServerConfig map
+pub struct ServerCrypto {
+  // For Quic/HTTP3, only servers with no client authentication
+  #[cfg(feature = "http3-quinn")]
+  pub inner_global_no_client_auth: Arc<ServerConfig>,
+  #[cfg(feature = "http3-s2n")]
+  pub inner_global_no_client_auth: s2n_quic_rustls::Server,
+  // For TLS over TCP/HTTP2 and 1.1, map of SNI to server_crypto for all given servers
+  pub inner_local_map: Arc<SniServerCryptoMap>,
+}
+
+/// Reloader target for the certificate reloader service
+#[derive(Debug, PartialEq, Eq, Clone, Default)]
+pub struct ServerCryptoBase {
+  inner: HashMap<ServerName, CertsAndKeys>,
+}
+
+#[async_trait]
+impl<T> Reload<ServerCryptoBase> for CryptoReloader<T>
+where
+  T: CryptoSource + Sync + Send,
+{
+  type Source = Arc<BackendAppManager<T>>;
+  async fn new(source: &Self::Source) -> Result<Self, ReloaderError<ServerCryptoBase>> {
+    Ok(Self { inner: source.clone() })
+  }
+
+  async fn reload(&self) -> Result<Option<ServerCryptoBase>, ReloaderError<ServerCryptoBase>> {
+    let mut certs_and_keys_map = ServerCryptoBase::default();
+
+    for (server_name_bytes_exp, backend) in self.inner.apps.iter() {
+      if let Some(crypto_source) = &backend.crypto_source {
+        let certs_and_keys = crypto_source
+          .read()
+          .await
+          .map_err(|_e| ReloaderError::<ServerCryptoBase>::Reload("Failed to reload cert, key or ca cert"))?;
+        certs_and_keys_map
+          .inner
+          .insert(server_name_bytes_exp.to_owned(), certs_and_keys);
+      }
+    }
+
+    Ok(Some(certs_and_keys_map))
+  }
+}
+
+impl TryInto<Arc<ServerCrypto>> for &ServerCryptoBase {
+  type Error = anyhow::Error;
+
+  fn try_into(self) -> Result<Arc<ServerCrypto>, Self::Error> {
+    #[cfg(any(feature = "http3-quinn", feature = "http3-s2n"))]
+    let server_crypto_global = self.build_server_crypto_global()?;
+    let server_crypto_local_map: SniServerCryptoMap = self.build_server_crypto_local_map()?;
+
+    Ok(Arc::new(ServerCrypto {
+      #[cfg(feature = "http3-quinn")]
+      inner_global_no_client_auth: Arc::new(server_crypto_global),
+      #[cfg(feature = "http3-s2n")]
+      inner_global_no_client_auth: server_crypto_global,
+      inner_local_map: Arc::new(server_crypto_local_map),
+    }))
+  }
+}
+
+impl ServerCryptoBase {
+  fn build_server_crypto_local_map(&self) -> Result<SniServerCryptoMap, ReloaderError<ServerCryptoBase>> {
+    let mut server_crypto_local_map: SniServerCryptoMap = HashMap::default();
+
+    for (server_name_bytes_exp, certs_and_keys) in self.inner.iter() {
+      let server_name: String = server_name_bytes_exp.try_into()?;
+
+      // Parse server certificates and private keys
+      let Ok(certified_key): Result<CertifiedKey, _> = certs_and_keys.parse_server_certs_and_keys() else {
+        warn!("Failed to add certificate for {}", server_name);
+        continue;
+      };
+
+      let mut resolver_local = ResolvesServerCertUsingSni::new();
+      let mut client_ca_roots_local = RootCertStore::empty();
+
+      // add server certificate and key
+      if let Err(e) = resolver_local.add(server_name.as_str(), certified_key.to_owned()) {
+        error!(
+          "{}: Failed to read some certificates and keys {}",
+          server_name.as_str(),
+          e
+        )
+      }
+
+      // add client certificate if specified
+      if certs_and_keys.client_ca_certs.is_some() {
+        // add client certificate if specified
+        match certs_and_keys.parse_client_ca_certs() {
+          Ok((owned_trust_anchors, _subject_key_ids)) => {
+            client_ca_roots_local.add_trust_anchors(owned_trust_anchors.into_iter());
+          }
+          Err(e) => {
+            warn!(
+              "Failed to add client CA certificate for {}: {}",
+              server_name.as_str(),
+              e
+            );
+          }
+        }
+      }
+
+      let mut server_config_local = if client_ca_roots_local.is_empty() {
+        // with no client auth, enable http1.1 -- 3
+        #[cfg(not(any(feature = "http3-quinn", feature = "http3-s2n")))]
+        {
+          ServerConfig::builder()
+            .with_safe_defaults()
+            .with_no_client_auth()
+            .with_cert_resolver(Arc::new(resolver_local))
+        }
+        #[cfg(any(feature = "http3-quinn", feature = "http3-s2n"))]
+        {
+          let mut sc = ServerConfig::builder()
+            .with_safe_defaults()
+            .with_no_client_auth()
+            .with_cert_resolver(Arc::new(resolver_local));
+          sc.alpn_protocols = vec![b"h3".to_vec(), b"hq-29".to_vec()]; // TODO: remove hq-29 later?
+          sc
+        }
+      } else {
+        // with client auth, enable only http1.1 and 2
+        // let client_certs_verifier = rustls::server::AllowAnyAnonymousOrAuthenticatedClient::new(client_ca_roots);
+        let client_certs_verifier = rustls::server::AllowAnyAuthenticatedClient::new(client_ca_roots_local);
+        ServerConfig::builder()
+          .with_safe_defaults()
+          .with_client_cert_verifier(Arc::new(client_certs_verifier))
+          .with_cert_resolver(Arc::new(resolver_local))
+      };
+      server_config_local.alpn_protocols.push(b"h2".to_vec());
+      server_config_local.alpn_protocols.push(b"http/1.1".to_vec());
+
+      server_crypto_local_map.insert(server_name_bytes_exp.to_owned(), Arc::new(server_config_local));
+    }
+    Ok(server_crypto_local_map)
+  }
+
+  #[cfg(feature = "http3-quinn")]
+  fn build_server_crypto_global(&self) -> Result<ServerConfig, ReloaderError<ServerCryptoBase>> {
+    let mut resolver_global = ResolvesServerCertUsingSni::new();
+
+    for (server_name_bytes_exp, certs_and_keys) in self.inner.iter() {
+      let server_name: String = server_name_bytes_exp.try_into()?;
+
+      // Parse server certificates and private keys
+      let Ok(certified_key): Result<CertifiedKey, _> = certs_and_keys.parse_server_certs_and_keys() else {
+        warn!("Failed to add certificate for {}", server_name);
+        continue;
+      };
+
+      if certs_and_keys.client_ca_certs.is_none() {
+        // aggregated server config for no client auth server for http3
+        if let Err(e) = resolver_global.add(server_name.as_str(), certified_key) {
+          error!(
+            "{}: Failed to read some certificates and keys {}",
+            server_name.as_str(),
+            e
+          )
+        }
+      }
+    }
+
+    //////////////
+    let mut server_crypto_global = ServerConfig::builder()
+      .with_safe_defaults()
+      .with_no_client_auth()
+      .with_cert_resolver(Arc::new(resolver_global));
+
+    //////////////////////////////
+
+    server_crypto_global.alpn_protocols = vec![
+      b"h3".to_vec(),
+      b"hq-29".to_vec(), // TODO: remove later?
+      b"h2".to_vec(),
+      b"http/1.1".to_vec(),
+    ];
+    Ok(server_crypto_global)
+  }
+
+  #[cfg(feature = "http3-s2n")]
+  fn build_server_crypto_global(&self) -> Result<s2n_quic_rustls::Server, ReloaderError<ServerCryptoBase>> {
+    let mut resolver_global = s2n_quic_rustls::rustls::server::ResolvesServerCertUsingSni::new();
+
+    for (server_name_bytes_exp, certs_and_keys) in self.inner.iter() {
+      let server_name: String = server_name_bytes_exp.try_into()?;
+
+      // Parse server certificates and private keys
+      let Ok(certified_key) = parse_server_certs_and_keys_s2n(certs_and_keys) else {
+        warn!("Failed to add certificate for {}", server_name);
+        continue;
+      };
+
+      if certs_and_keys.client_ca_certs.is_none() {
+        // aggregated server config for no client auth server for http3
+        if let Err(e) = resolver_global.add(server_name.as_str(), certified_key) {
+          error!(
+            "{}: Failed to read some certificates and keys {}",
+            server_name.as_str(),
+            e
+          )
+        }
+      }
+    }
+    let alpn = vec![
+      b"h3".to_vec(),
+      b"hq-29".to_vec(), // TODO: remove later?
+      b"h2".to_vec(),
+      b"http/1.1".to_vec(),
+    ];
+    let server_crypto_global = s2n_quic::provider::tls::rustls::Server::builder()
+      .with_cert_resolver(Arc::new(resolver_global))
+      .map_err(|e| anyhow::anyhow!(e))?
+      .with_application_protocols(alpn.iter())
+      .map_err(|e| anyhow::anyhow!(e))?
+      .build()
+      .map_err(|e| anyhow::anyhow!(e))?;
+    Ok(server_crypto_global)
+  }
+}
+
+#[cfg(feature = "http3-s2n")]
+/// This is workaround for the version difference between rustls and s2n-quic-rustls
+fn parse_server_certs_and_keys_s2n(
+  certs_and_keys: &CertsAndKeys,
+) -> Result<s2n_quic_rustls::rustls::sign::CertifiedKey, anyhow::Error> {
+  let signing_key = certs_and_keys
+    .cert_keys
+    .iter()
+    .find_map(|k| {
+      let s2n_private_key = s2n_quic_rustls::PrivateKey(k.0.clone());
+      if let Ok(sk) = s2n_quic_rustls::rustls::sign::any_supported_type(&s2n_private_key) {
+        Some(sk)
+      } else {
+        None
+      }
+    })
+    .ok_or_else(|| {
+      std::io::Error::new(
+        std::io::ErrorKind::InvalidInput,
+        "Unable to find a valid certificate and key",
+      )
+    })?;
+  let certs: Vec<_> = certs_and_keys
+    .certs
+    .iter()
+    .map(|c| s2n_quic_rustls::rustls::Certificate(c.0.clone()))
+    .collect();
+  Ok(s2n_quic_rustls::rustls::sign::CertifiedKey::new(certs, signing_key))
+}

rpxy-lib/src/error.rs

@@ -9,12 +9,15 @@ pub enum RpxyError {
   #[error("IO error: {0}")]
   Io(#[from] std::io::Error),
 
+  #[error("Certificate reload error: {0}")]
+  CertificateReloadError(#[from] hot_reload::ReloaderError<crate::crypto::ServerCryptoBase>),
+
   // backend errors
   #[error("Invalid reverse proxy setting")]
   InvalidReverseProxyConfig,
   #[error("Invalid upstream option setting")]
   InvalidUpstreamOptionSetting,
-  #[error("Failed to build backend app")]
+  #[error("Failed to build backend app: {0}")]
   FailedToBuildBackendApp(#[from] crate::backend::BackendAppBuilderError),
 
   #[error("Unsupported upstream option")]

rpxy-lib/src/globals.rs

@@ -1,9 +1,14 @@
-use crate::{certs::CryptoSource, constants::*, count::RequestCount};
+use crate::{
+  constants::*,
+  count::RequestCount,
+  crypto::{CryptoSource, ServerCryptoBase},
+};
+use hot_reload::ReloaderReceiver;
 use std::{net::SocketAddr, sync::Arc, time::Duration};
 
 /// Global object containing proxy configurations and shared object like counters.
 /// But note that in Globals, we do not have Mutex and RwLock. It is indeed, the context shared among async tasks.
-pub struct Globals {
+pub(crate) struct Globals {
   /// Configuration parameters for proxy transport and request handlers
   pub proxy_config: ProxyConfig,
   /// Shared context - Counter for serving requests
@@ -12,6 +17,8 @@ pub struct Globals {
   pub runtime_handle: tokio::runtime::Handle,
   /// Shared context - Notify object to stop async tasks
   pub term_notify: Option<Arc<tokio::sync::Notify>>,
+  /// Shared context - Certificate reloader service receiver
+  pub cert_reloader_rx: Option<ReloaderReceiver<ServerCryptoBase>>,
 }
 
 /// Configuration parameters for proxy transport and request handlers

rpxy-lib/src/lib.rs

@@ -1,7 +1,7 @@
 mod backend;
-mod certs;
 mod constants;
 mod count;
+mod crypto;
 mod error;
 mod globals;
 mod hyper_executor;
@@ -9,12 +9,12 @@ mod log;
 mod name_exp;
 mod proxy;
 
-use crate::{error::*, globals::Globals, log::*, proxy::Proxy};
+use crate::{crypto::build_cert_reloader, error::*, globals::Globals, log::*, proxy::Proxy};
 use futures::future::select_all;
 use std::sync::Arc;
 
 pub use crate::{
-  certs::{CertsAndKeys, CryptoSource},
+  crypto::{CertsAndKeys, CryptoSource},
   globals::{AppConfig, AppConfigList, ProxyConfig, ReverseProxyConfig, TlsConfig, UpstreamUri},
 };
 pub mod reexports {
@@ -64,17 +64,27 @@ where
     info!("Cache is disabled")
   }
 
-  // build global shared context
+  // 1. build backends, and make it contained in Arc
+  let app_manager = Arc::new(backend::BackendAppManager::try_from(app_config_list)?);
+
+  // 2. build crypto reloader service
+  let (cert_reloader_service, cert_reloader_rx) = match proxy_config.https_port {
+    Some(_) => {
+      let (s, r) = build_cert_reloader(&app_manager).await?;
+      (Some(s), Some(r))
+    }
+    None => (None, None),
+  };
+
+  // 3. build global shared context
   let globals = Arc::new(Globals {
     proxy_config: proxy_config.clone(),
     request_count: Default::default(),
     runtime_handle: runtime_handle.clone(),
     term_notify: term_notify.clone(),
+    cert_reloader_rx: cert_reloader_rx.clone(),
   });
 
-  // 1. build backends, and make it contained in Arc
-  let app_manager = Arc::new(backend::BackendAppManager::try_from(app_config_list)?);
-
   // TODO: 2. build message handler with Arc-ed http_client and backends, and make it contained in Arc as well
   // // build message handler including a request forwarder
   // let msg_handler = Arc::new(
@@ -106,9 +116,23 @@ where
   });
 
   // wait for all future
-  if let (Ok(Err(e)), _, _) = select_all(futures_iter).await {
-    error!("Some proxy services are down: {}", e);
-  };
+  match cert_reloader_service {
+    Some(cert_service) => {
+      tokio::select! {
+        _ = cert_service.start() => {
+          error!("Certificate reloader service got down");
+        }
+        _ = select_all(futures_iter) => {
+          error!("Some proxy services are down");
+        }
+      }
+    }
+    None => {
+      if let (Ok(Err(e)), _, _) = select_all(futures_iter).await {
+        error!("Some proxy services are down: {}", e);
+      }
+    }
+  }
 
   Ok(())
 }

rpxy-lib/src/proxy/mod.rs

@@ -1,6 +1,6 @@
 mod proxy_main;
-mod socket;
 mod proxy_tls;
+mod socket;
 
 use crate::{globals::Globals, hyper_executor::LocalExecutor};
 use hyper_util::server::{self, conn::auto::Builder as ConnectionBuilder};

rpxy-lib/src/proxy/proxy_main.rs

@@ -1,6 +1,5 @@
 use super::socket::bind_tcp_socket;
 use crate::{error::RpxyResult, globals::Globals, log::*};
-use hot_reload::{ReloaderReceiver, ReloaderService};
 use hyper_util::server::conn::auto::Builder as ConnectionBuilder;
 use std::{net::SocketAddr, sync::Arc};