diff --git a/Cargo.toml b/Cargo.toml index c73533e..5f7b0ea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,7 @@ clap = { version = "4.3.11", features = ["std", "cargo", "wrap_help"] } rand = "0.8.5" toml = { version = "0.7.6", default-features = false, features = ["parse"] } rustc-hash = "1.1.0" -serde = { version = "1.0.167", default-features = false, features = ["derive"] } +serde = { version = "1.0.171", default-features = false, features = ["derive"] } bytes = "1.4.0" thiserror = "1.0.43" x509-parser = "0.15.0" @@ -31,11 +31,12 @@ futures = { version = "0.3.28", features = ["alloc", "async-await"] } tokio = { version = "1.29.1", default-features = false, features = [ "net", "rt-multi-thread", - "parking_lot", "time", "sync", "macros", ] } +async-trait = "0.1.71" +hot_reload = "0.1.2" # reloading certs # http and tls hyper = { version = "0.14.27", default-features = false, features = [ diff --git a/quinn b/quinn index b30711f..e652b6d 160000 --- a/quinn +++ b/quinn @@ -1 +1 @@ -Subproject commit b30711f5595983989b60bbbad0ac3f067be7a596 +Subproject commit e652b6d999f053ffe21eeea247854882ae480281 diff --git a/src/backend/mod.rs b/src/backend/mod.rs index b7923c5..c8298c3 100644 --- a/src/backend/mod.rs +++ b/src/backend/mod.rs @@ -13,26 +13,10 @@ pub use self::{ upstream::{ReverseProxy, Upstream, UpstreamGroup, UpstreamGroupBuilder}, upstream_opts::UpstreamOption, }; -use crate::{ - log::*, - utils::{BytesName, PathNameBytesExp, ServerNameBytesExp}, -}; +use crate::utils::{BytesName, PathNameBytesExp, ServerNameBytesExp}; use derive_builder::Builder; -use rustc_hash::{FxHashMap as HashMap, FxHashSet as HashSet}; -use rustls::{OwnedTrustAnchor, RootCertStore}; -use std::{ - borrow::Cow, - fs::File, - io::{self, BufReader, Cursor, Read}, - path::PathBuf, - sync::Arc, -}; -use tokio_rustls::rustls::{ - server::ResolvesServerCertUsingSni, - sign::{any_supported_type, CertifiedKey}, - Certificate, PrivateKey, ServerConfig, -}; -use x509_parser::prelude::*; +use rustc_hash::FxHashMap as HashMap; +use std::{borrow::Cow, path::PathBuf}; /// Struct serving information to route incoming connections, like server name to be handled and tls certs/keys settings. #[derive(Builder)] @@ -79,264 +63,9 @@ fn opt_string_to_opt_pathbuf(input: &Option) -> Option { input.to_owned().as_ref().map(PathBuf::from) } -impl Backend { - pub fn read_certs_and_key(&self) -> io::Result { - debug!("Read TLS server certificates and private key"); - let (Some(certs_path), Some(certs_keys_path)) = (self.tls_cert_path.as_ref(), self.tls_cert_key_path.as_ref()) else { - return Err(io::Error::new(io::ErrorKind::Other, "Invalid certs and keys paths")); - }; - let certs: Vec<_> = { - let certs_path_str = certs_path.display().to_string(); - let mut reader = BufReader::new(File::open(certs_path).map_err(|e| { - io::Error::new( - e.kind(), - format!("Unable to load the certificates [{certs_path_str}]: {e}"), - ) - })?); - rustls_pemfile::certs(&mut reader) - .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "Unable to parse the certificates"))? - } - .drain(..) - .map(Certificate) - .collect(); - let certs_keys: Vec<_> = { - let certs_keys_path_str = certs_keys_path.display().to_string(); - let encoded_keys = { - let mut encoded_keys = vec![]; - File::open(certs_keys_path) - .map_err(|e| { - io::Error::new( - e.kind(), - format!("Unable to load the certificate keys [{certs_keys_path_str}]: {e}"), - ) - })? - .read_to_end(&mut encoded_keys)?; - encoded_keys - }; - let mut reader = Cursor::new(encoded_keys); - let pkcs8_keys = rustls_pemfile::pkcs8_private_keys(&mut reader).map_err(|_| { - io::Error::new( - io::ErrorKind::InvalidInput, - "Unable to parse the certificates private keys (PKCS8)", - ) - })?; - reader.set_position(0); - let mut rsa_keys = rustls_pemfile::rsa_private_keys(&mut reader)?; - let mut keys = pkcs8_keys; - keys.append(&mut rsa_keys); - if keys.is_empty() { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "No private keys found - Make sure that they are in PKCS#8/PEM format", - )); - } - keys.drain(..).map(PrivateKey).collect() - }; - let signing_key = certs_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(certs, signing_key)) - } - - fn read_client_ca_certs(&self) -> io::Result<(Vec, HashSet>)> { - debug!("Read CA certificates for client authentication"); - // Reads client certificate and returns client - let client_ca_cert_path = { - let Some(c) = self.client_ca_cert_path.as_ref() else { - return Err(io::Error::new(io::ErrorKind::Other, "Invalid certs and keys paths")); - }; - c - }; - let certs: Vec<_> = { - let certs_path_str = client_ca_cert_path.display().to_string(); - let mut reader = BufReader::new(File::open(client_ca_cert_path).map_err(|e| { - io::Error::new( - e.kind(), - format!("Unable to load the client certificates [{certs_path_str}]: {e}"), - ) - })?); - rustls_pemfile::certs(&mut reader) - .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "Unable to parse the client certificates"))? - } - .drain(..) - .map(Certificate) - .collect(); - - 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::>(); - if !subject_key_ids.is_empty() { - Some(subject_key_ids[0].0.to_owned()) - } else { - None - } - }) - .collect(); - - Ok((owned_trust_anchors, subject_key_identifiers)) - } -} - +#[derive(Default)] /// HashMap and some meta information for multiple Backend structs. pub struct Backends { pub apps: HashMap, // hyper::uriで抜いたhostで引っ掛ける pub default_server_name_bytes: Option, // for plaintext http } - -pub type SniServerCryptoMap = HashMap>; -pub struct ServerCrypto { - // For Quic/HTTP3, only servers with no client authentication - pub inner_global_no_client_auth: Arc, - // For TLS over TCP/HTTP2 and 1.1, map of SNI to server_crypto for all given servers - pub inner_local_map: Arc, -} - -impl Backends { - pub async fn generate_server_crypto(&self) -> Result { - let mut resolver_global = ResolvesServerCertUsingSni::new(); - let mut server_crypto_local_map: SniServerCryptoMap = HashMap::default(); - - for (server_name_bytes_exp, backend) in self.apps.iter() { - if backend.tls_cert_key_path.is_some() && backend.tls_cert_path.is_some() { - match backend.read_certs_and_key() { - Ok(certified_key) => { - 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(backend.server_name.as_str(), certified_key.to_owned()) { - error!( - "{}: Failed to read some certificates and keys {}", - backend.server_name.as_str(), - e - ) - } - - if backend.client_ca_cert_path.is_none() { - // aggregated server config for no client auth server for http3 - if let Err(e) = resolver_global.add(backend.server_name.as_str(), certified_key) { - error!( - "{}: Failed to read some certificates and keys {}", - backend.server_name.as_str(), - e - ) - } - } else { - // add client certificate if specified - match backend.read_client_ca_certs() { - Ok((owned_trust_anchors, _subject_key_ids)) => { - client_ca_roots_local.add_server_trust_anchors(owned_trust_anchors.into_iter()); - } - Err(e) => { - warn!( - "Failed to add client CA certificate for {}: {}", - backend.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(feature = "http3"))] - { - ServerConfig::builder() - .with_safe_defaults() - .with_no_client_auth() - .with_cert_resolver(Arc::new(resolver_local)) - } - #[cfg(feature = "http3")] - { - 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)); - } - Err(e) => { - warn!("Failed to add certificate for {}: {}", backend.server_name.as_str(), e); - } - } - } - } - // debug!("Load certificate chain for {} server_name's", cnt); - - ////////////// - let mut server_crypto_global = ServerConfig::builder() - .with_safe_defaults() - .with_no_client_auth() - .with_cert_resolver(Arc::new(resolver_global)); - - ////////////////////////////// - - #[cfg(feature = "http3")] - { - 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(), - ]; - } - #[cfg(not(feature = "http3"))] - { - server_crypto_global.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()]; - } - - Ok(ServerCrypto { - inner_global_no_client_auth: Arc::new(server_crypto_global), - inner_local_map: Arc::new(server_crypto_local_map), - }) - } -} diff --git a/src/cert_reader.rs b/src/cert_reader.rs new file mode 100644 index 0000000..a52f2e2 --- /dev/null +++ b/src/cert_reader.rs @@ -0,0 +1,93 @@ +use crate::{log::*, proxy::CertsAndKeys}; +use rustls::{Certificate, PrivateKey}; +use std::{ + fs::File, + io::{self, BufReader, Cursor, Read}, + path::PathBuf, +}; + +/// Read certificates and private keys from file +pub(crate) fn read_certs_and_keys( + cert_path: &PathBuf, + cert_key_path: &PathBuf, + client_ca_cert_path: Option<&PathBuf>, +) -> Result { + debug!("Read TLS server certificates and private key"); + + let certs: Vec<_> = { + let certs_path_str = cert_path.display().to_string(); + let mut reader = BufReader::new(File::open(cert_path).map_err(|e| { + io::Error::new( + e.kind(), + format!("Unable to load the certificates [{certs_path_str}]: {e}"), + ) + })?); + rustls_pemfile::certs(&mut reader) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "Unable to parse the certificates"))? + } + .drain(..) + .map(Certificate) + .collect(); + + let cert_keys: Vec<_> = { + let cert_key_path_str = cert_key_path.display().to_string(); + let encoded_keys = { + let mut encoded_keys = vec![]; + File::open(cert_key_path) + .map_err(|e| { + io::Error::new( + e.kind(), + format!("Unable to load the certificate keys [{cert_key_path_str}]: {e}"), + ) + })? + .read_to_end(&mut encoded_keys)?; + encoded_keys + }; + let mut reader = Cursor::new(encoded_keys); + let pkcs8_keys = rustls_pemfile::pkcs8_private_keys(&mut reader).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidInput, + "Unable to parse the certificates private keys (PKCS8)", + ) + })?; + reader.set_position(0); + let mut rsa_keys = rustls_pemfile::rsa_private_keys(&mut reader)?; + let mut keys = pkcs8_keys; + keys.append(&mut rsa_keys); + if keys.is_empty() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "No private keys found - Make sure that they are in PKCS#8/PEM format", + )); + } + keys.drain(..).map(PrivateKey).collect() + }; + + let client_ca_certs = if let Some(path) = client_ca_cert_path { + debug!("Read CA certificates for client authentication"); + // Reads client certificate and returns client + let certs: Vec<_> = { + let certs_path_str = path.display().to_string(); + let mut reader = BufReader::new(File::open(path).map_err(|e| { + io::Error::new( + e.kind(), + format!("Unable to load the client certificates [{certs_path_str}]: {e}"), + ) + })?); + rustls_pemfile::certs(&mut reader) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "Unable to parse the client certificates"))? + } + .drain(..) + .map(Certificate) + .collect(); + Some(certs) + } else { + None + }; + + Ok(CertsAndKeys { + certs, + cert_keys, + client_ca_certs, + }) +} diff --git a/src/config/mod.rs b/src/config/mod.rs index 6e8123c..54b2600 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -1,4 +1,4 @@ mod parse; mod toml; -pub use parse::parse_opts; +pub use parse::build_globals; diff --git a/src/config/parse.rs b/src/config/parse.rs index 8e4ddf7..1593aba 100644 --- a/src/config/parse.rs +++ b/src/config/parse.rs @@ -1,17 +1,9 @@ -use super::toml::{ConfigToml, ReverseProxyOption}; -use crate::{ - backend::{BackendBuilder, ReverseProxy, Upstream, UpstreamGroup, UpstreamGroupBuilder, UpstreamOption}, - constants::*, - error::*, - globals::*, - log::*, - utils::{BytesName, PathNameBytesExp}, -}; +use super::toml::ConfigToml; +use crate::{backend::Backends, error::*, globals::*, log::*, utils::BytesName}; use clap::Arg; -use rustc_hash::FxHashMap as HashMap; -use std::net::SocketAddr; +use tokio::runtime::Handle; -pub fn parse_opts(globals: &mut Globals) -> std::result::Result<(), anyhow::Error> { +pub fn build_globals(runtime_handle: Handle) -> std::result::Result { let _ = include_str!("../../Cargo.toml"); let options = clap::command!().arg( Arg::new("config_file") @@ -22,6 +14,7 @@ pub fn parse_opts(globals: &mut Globals) -> std::result::Result<(), anyhow::Erro ); let matches = options.get_matches(); + /////////////////////////////////// let config = if let Some(config_file_path) = matches.get_one::("config_file") { ConfigToml::new(config_file_path)? } else { @@ -29,117 +22,67 @@ pub fn parse_opts(globals: &mut Globals) -> std::result::Result<(), anyhow::Erro ConfigToml::default() }; - // listen port and socket - globals.http_port = config.listen_port; - globals.https_port = config.listen_port_tls; - ensure!( - { globals.http_port.is_some() || globals.https_port.is_some() } && { - if let (Some(p), Some(t)) = (globals.http_port, globals.https_port) { - p != t - } else { - true - } - }, - anyhow!("Wrong port spec.") - ); - // NOTE: when [::]:xx is bound, both v4 and v6 listeners are enabled. - let listen_addresses: Vec<&str> = match config.listen_ipv6 { - Some(true) => { - info!("Listen both IPv4 and IPv6"); - LISTEN_ADDRESSES_V6.to_vec() - } - Some(false) | None => { - info!("Listen IPv4"); - LISTEN_ADDRESSES_V4.to_vec() - } - }; - globals.listen_sockets = listen_addresses - .iter() - .flat_map(|x| { - let mut v: Vec = vec![]; - if let Some(p) = globals.http_port { - v.push(format!("{x}:{p}").parse().unwrap()); - } - if let Some(p) = globals.https_port { - v.push(format!("{x}:{p}").parse().unwrap()); - } - v - }) - .collect(); - if globals.http_port.is_some() { - info!("Listen port: {}", globals.http_port.unwrap()); + /////////////////////////////////// + // build proxy config + let proxy_config: ProxyConfig = (&config).try_into()?; + // For loggings + if proxy_config.listen_sockets.iter().any(|addr| addr.is_ipv6()) { + info!("Listen both IPv4 and IPv6") + } else { + info!("Listen IPv4") } - if globals.https_port.is_some() { - info!("Listen port: {} (for TLS)", globals.https_port.unwrap()); + if proxy_config.http_port.is_some() { + info!("Listen port: {}", proxy_config.http_port.unwrap()); + } + if proxy_config.https_port.is_some() { + info!("Listen port: {} (for TLS)", proxy_config.https_port.unwrap()); + } + if proxy_config.http3 { + info!("Experimental HTTP/3.0 is enabled. Note it is still very unstable."); + } + if !proxy_config.sni_consistency { + info!("Ignore consistency between TLS SNI and Host header (or Request line). Note it violates RFC."); } - // max values - if let Some(c) = config.max_clients { - globals.max_clients = c as usize; - } - if let Some(c) = config.max_concurrent_streams { - globals.max_concurrent_streams = c; - } + /////////////////////////////////// + // backend_apps + let apps = config.apps.ok_or(anyhow!("Missing application spec"))?; - // backend apps - ensure!(config.apps.is_some(), "Missing application spec."); - let apps = config.apps.unwrap(); + // assertions for all backend apps ensure!(!apps.0.is_empty(), "Wrong application spec."); + // if only https_port is specified, tls must be configured for all apps + if proxy_config.http_port.is_none() { + ensure!( + apps.0.iter().all(|(_, app)| app.tls.is_some()), + "Some apps serves only plaintext HTTP" + ); + } + // https redirection can be configured if both ports are active + if !(proxy_config.https_port.is_some() && proxy_config.http_port.is_some()) { + ensure!( + apps.0.iter().all(|(_, app)| { + if let Some(tls) = app.tls.as_ref() { + tls.https_redirection.is_none() + } else { + true + } + }), + "https_redirection can be specified only when both http_port and https_port are specified" + ); + } - // each app + // build backends + let mut backends = Backends::default(); for (app_name, app) in apps.0.iter() { - ensure!(app.server_name.is_some(), "Missing server_name"); - let server_name_string = app.server_name.as_ref().unwrap(); - if globals.http_port.is_none() { - // if only https_port is specified, tls must be configured - ensure!(app.tls.is_some()) - } - - // backend builder - let mut backend_builder = BackendBuilder::default(); - // reverse proxy settings - ensure!(app.reverse_proxy.is_some(), "Missing reverse_proxy"); - let reverse_proxy = get_reverse_proxy(server_name_string, app.reverse_proxy.as_ref().unwrap())?; - - backend_builder - .app_name(server_name_string) - .server_name(server_name_string) - .reverse_proxy(reverse_proxy); - - // TLS settings and build backend instance - let backend = if app.tls.is_none() { - ensure!(globals.http_port.is_some(), "Required HTTP port"); - backend_builder.build()? - } else { - let tls = app.tls.as_ref().unwrap(); - ensure!(tls.tls_cert_key_path.is_some() && tls.tls_cert_path.is_some()); - - let https_redirection = if tls.https_redirection.is_none() { - Some(true) // Default true - } else { - ensure!(globals.https_port.is_some()); // only when both https ports are configured. - tls.https_redirection - }; - - backend_builder - .tls_cert_path(&tls.tls_cert_path) - .tls_cert_key_path(&tls.tls_cert_key_path) - .https_redirection(https_redirection) - .client_ca_cert_path(&tls.client_ca_cert_path) - .build()? - }; - - globals - .backends - .apps - .insert(server_name_string.to_server_name_vec(), backend); + let server_name_string = app.server_name.as_ref().ok_or(anyhow!("No server name"))?; + let backend = app.try_into()?; + backends.apps.insert(server_name_string.to_server_name_vec(), backend); info!("Registering application: {} ({})", app_name, server_name_string); } // default backend application for plaintext http requests if let Some(d) = config.default_app { - let d_sn: Vec<&str> = globals - .backends + let d_sn: Vec<&str> = backends .apps .iter() .filter(|(_k, v)| v.app_name == d) @@ -150,86 +93,17 @@ pub fn parse_opts(globals: &mut Globals) -> std::result::Result<(), anyhow::Erro "Serving plaintext http for requests to unconfigured server_name by app {} (server_name: {}).", d, d_sn[0] ); - globals.backends.default_server_name_bytes = Some(d_sn[0].to_server_name_vec()); + backends.default_server_name_bytes = Some(d_sn[0].to_server_name_vec()); } } - // experimental - if let Some(exp) = config.experimental { - #[cfg(feature = "http3")] - { - if let Some(h3option) = exp.h3 { - globals.http3 = true; - info!("Experimental HTTP/3.0 is enabled. Note it is still very unstable."); - if let Some(x) = h3option.alt_svc_max_age { - globals.h3_alt_svc_max_age = x; - } - if let Some(x) = h3option.request_max_body_size { - globals.h3_request_max_body_size = x; - } - if let Some(x) = h3option.max_concurrent_connections { - globals.h3_max_concurrent_connections = x; - } - if let Some(x) = h3option.max_concurrent_bidistream { - globals.h3_max_concurrent_bidistream = x.into(); - } - if let Some(x) = h3option.max_concurrent_unistream { - globals.h3_max_concurrent_unistream = x.into(); - } - if let Some(x) = h3option.max_idle_timeout { - if x == 0u64 { - globals.h3_max_idle_timeout = None; - } else { - globals.h3_max_idle_timeout = - Some(quinn::IdleTimeout::try_from(tokio::time::Duration::from_secs(x)).unwrap()) - } - } - } - } + /////////////////////////////////// + let globals = Globals { + proxy_config, + backends, + request_count: Default::default(), + runtime_handle, + }; - if let Some(b) = exp.ignore_sni_consistency { - globals.sni_consistency = !b; - if b { - info!("Ignore consistency between TLS SNI and Host header (or Request line). Note it violates RFC."); - } - } - } - - Ok(()) -} - -fn get_reverse_proxy( - server_name_string: &str, - rp_settings: &[ReverseProxyOption], -) -> std::result::Result { - let mut upstream: HashMap = HashMap::default(); - - rp_settings.iter().for_each(|rpo| { - let upstream_vec: Vec = rpo.upstream.iter().map(|x| x.to_upstream().unwrap()).collect(); - // let upstream_iter = rpo.upstream.iter().map(|x| x.to_upstream().unwrap()); - // let lb_upstream_num = vec_upstream.len(); - let elem = UpstreamGroupBuilder::default() - .upstream(&upstream_vec) - .path(&rpo.path) - .replace_path(&rpo.replace_path) - .lb(&rpo.load_balance, &upstream_vec, server_name_string, &rpo.path) - .opts(&rpo.upstream_options) - .build() - .unwrap(); - - upstream.insert(elem.path.clone(), elem); - }); - ensure!( - rp_settings.iter().filter(|rpo| rpo.path.is_none()).count() < 2, - "Multiple default reverse proxy setting" - ); - ensure!( - upstream - .iter() - .all(|(_, elem)| !(elem.opts.contains(&UpstreamOption::ConvertHttpsTo11) - && elem.opts.contains(&UpstreamOption::ConvertHttpsTo2))), - "either one of force_http11 or force_http2 can be enabled" - ); - - Ok(ReverseProxy { upstream }) + Ok(globals) } diff --git a/src/config/toml.rs b/src/config/toml.rs index 6ce48b2..b883f6a 100644 --- a/src/config/toml.rs +++ b/src/config/toml.rs @@ -1,7 +1,13 @@ -use crate::{backend::Upstream, error::*}; +use crate::{ + backend::{Backend, BackendBuilder, ReverseProxy, Upstream, UpstreamGroup, UpstreamGroupBuilder, UpstreamOption}, + constants::*, + error::*, + globals::ProxyConfig, + utils::PathNameBytesExp, +}; use rustc_hash::FxHashMap as HashMap; use serde::Deserialize; -use std::fs; +use std::{fs, net::SocketAddr}; #[derive(Deserialize, Debug, Default)] pub struct ConfigToml { @@ -65,25 +71,196 @@ pub struct UpstreamParams { pub location: String, pub tls: Option, } -impl UpstreamParams { - pub fn to_upstream(&self) -> Result { - let mut scheme = "http"; - if let Some(t) = self.tls { - if t { - scheme = "https"; + +impl TryInto for &ConfigToml { + type Error = anyhow::Error; + + fn try_into(self) -> std::result::Result { + let mut proxy_config = ProxyConfig { + // listen port and socket + http_port: self.listen_port, + https_port: self.listen_port_tls, + ..Default::default() + }; + ensure!( + proxy_config.http_port.is_some() || proxy_config.https_port.is_some(), + anyhow!("Either/Both of http_port or https_port must be specified") + ); + if proxy_config.http_port.is_some() && proxy_config.https_port.is_some() { + ensure!( + proxy_config.http_port.unwrap() != proxy_config.https_port.unwrap(), + anyhow!("http_port and https_port must be different") + ); + } + + // NOTE: when [::]:xx is bound, both v4 and v6 listeners are enabled. + let listen_addresses: Vec<&str> = if let Some(true) = self.listen_ipv6 { + LISTEN_ADDRESSES_V6.to_vec() + } else { + LISTEN_ADDRESSES_V4.to_vec() + }; + proxy_config.listen_sockets = listen_addresses + .iter() + .flat_map(|addr| { + let mut v: Vec = vec![]; + if let Some(port) = proxy_config.http_port { + v.push(format!("{addr}:{port}").parse().unwrap()); + } + if let Some(port) = proxy_config.https_port { + v.push(format!("{addr}:{port}").parse().unwrap()); + } + v + }) + .collect(); + + // max values + if let Some(c) = self.max_clients { + proxy_config.max_clients = c as usize; + } + if let Some(c) = self.max_concurrent_streams { + proxy_config.max_concurrent_streams = c; + } + + // experimental + if let Some(exp) = &self.experimental { + #[cfg(feature = "http3")] + { + if let Some(h3option) = &exp.h3 { + proxy_config.http3 = true; + if let Some(x) = h3option.alt_svc_max_age { + proxy_config.h3_alt_svc_max_age = x; + } + if let Some(x) = h3option.request_max_body_size { + proxy_config.h3_request_max_body_size = x; + } + if let Some(x) = h3option.max_concurrent_connections { + proxy_config.h3_max_concurrent_connections = x; + } + if let Some(x) = h3option.max_concurrent_bidistream { + proxy_config.h3_max_concurrent_bidistream = x.into(); + } + if let Some(x) = h3option.max_concurrent_unistream { + proxy_config.h3_max_concurrent_unistream = x.into(); + } + if let Some(x) = h3option.max_idle_timeout { + if x == 0u64 { + proxy_config.h3_max_idle_timeout = None; + } else { + proxy_config.h3_max_idle_timeout = + Some(quinn::IdleTimeout::try_from(tokio::time::Duration::from_secs(x)).unwrap()) + } + } + } + } + + if let Some(ignore) = exp.ignore_sni_consistency { + proxy_config.sni_consistency = !ignore; } } + + Ok(proxy_config) + } +} + +impl ConfigToml { + pub fn new(config_file: &str) -> std::result::Result { + let config_str = fs::read_to_string(config_file).map_err(RpxyError::Io)?; + + toml::from_str(&config_str).map_err(RpxyError::TomlDe) + } +} + +impl TryInto for &Application { + type Error = anyhow::Error; + + fn try_into(self) -> std::result::Result { + let server_name_string = self.server_name.as_ref().ok_or(anyhow!("Missing server_name"))?; + + // backend builder + let mut backend_builder = BackendBuilder::default(); + // reverse proxy settings + let reverse_proxy = self.try_into()?; + + backend_builder + .app_name(server_name_string) + .server_name(server_name_string) + .reverse_proxy(reverse_proxy); + + // TLS settings and build backend instance + let backend = if self.tls.is_none() { + backend_builder.build()? + } else { + let tls = self.tls.as_ref().unwrap(); + ensure!(tls.tls_cert_key_path.is_some() && tls.tls_cert_path.is_some()); + + let https_redirection = if tls.https_redirection.is_none() { + Some(true) // Default true + } else { + tls.https_redirection + }; + + backend_builder + .tls_cert_path(&tls.tls_cert_path) + .tls_cert_key_path(&tls.tls_cert_key_path) + .https_redirection(https_redirection) + .client_ca_cert_path(&tls.client_ca_cert_path) + .build()? + }; + Ok(backend) + } +} + +impl TryInto for &Application { + type Error = anyhow::Error; + + fn try_into(self) -> std::result::Result { + let server_name_string = self.server_name.as_ref().ok_or(anyhow!("Missing server_name"))?; + let rp_settings = self.reverse_proxy.as_ref().ok_or(anyhow!("Missing reverse_proxy"))?; + + let mut upstream: HashMap = HashMap::default(); + + rp_settings.iter().for_each(|rpo| { + let upstream_vec: Vec = rpo.upstream.iter().map(|x| x.try_into().unwrap()).collect(); + // let upstream_iter = rpo.upstream.iter().map(|x| x.to_upstream().unwrap()); + // let lb_upstream_num = vec_upstream.len(); + let elem = UpstreamGroupBuilder::default() + .upstream(&upstream_vec) + .path(&rpo.path) + .replace_path(&rpo.replace_path) + .lb(&rpo.load_balance, &upstream_vec, server_name_string, &rpo.path) + .opts(&rpo.upstream_options) + .build() + .unwrap(); + + upstream.insert(elem.path.clone(), elem); + }); + ensure!( + rp_settings.iter().filter(|rpo| rpo.path.is_none()).count() < 2, + "Multiple default reverse proxy setting" + ); + ensure!( + upstream + .iter() + .all(|(_, elem)| !(elem.opts.contains(&UpstreamOption::ConvertHttpsTo11) + && elem.opts.contains(&UpstreamOption::ConvertHttpsTo2))), + "either one of force_http11 or force_http2 can be enabled" + ); + + Ok(ReverseProxy { upstream }) + } +} + +impl TryInto for &UpstreamParams { + type Error = RpxyError; + + fn try_into(self) -> std::result::Result { + let scheme = match self.tls { + Some(true) => "https", + _ => "http", + }; let location = format!("{}://{}", scheme, self.location); Ok(Upstream { uri: location.parse::().map_err(|e| anyhow!("{}", e))?, }) } } - -impl ConfigToml { - pub fn new(config_file: &str) -> std::result::Result { - let config_str = fs::read_to_string(config_file).context("Failed to read config file")?; - - toml::from_str(&config_str).context("Failed to parse toml config") - } -} diff --git a/src/constants.rs b/src/constants.rs index a29be29..2ed14d1 100644 --- a/src/constants.rs +++ b/src/constants.rs @@ -8,6 +8,7 @@ pub const TLS_HANDSHAKE_TIMEOUT_SEC: u64 = 15; // default as with firefox browse pub const MAX_CLIENTS: usize = 512; pub const MAX_CONCURRENT_STREAMS: u32 = 64; pub const CERTS_WATCH_DELAY_SECS: u32 = 60; +pub const LOAD_CERTS_ONLY_WHEN_UPDATED: bool = true; // #[cfg(feature = "http3")] // pub const H3_RESPONSE_BUF_SIZE: usize = 65_536; // 64KB diff --git a/src/error.rs b/src/error.rs index 3fb3474..18b4307 100644 --- a/src/error.rs +++ b/src/error.rs @@ -29,6 +29,9 @@ pub enum RpxyError { #[error("I/O Error")] Io(#[from] io::Error), + #[error("Toml Deserialization Error")] + TomlDe(#[from] toml::de::Error), + #[cfg(feature = "http3")] #[error("Quic Connection Error")] QuicConn(#[from] quinn::ConnectionError), diff --git a/src/globals.rs b/src/globals.rs index 0bd06a6..64f9d8d 100644 --- a/src/globals.rs +++ b/src/globals.rs @@ -1,4 +1,4 @@ -use crate::backend::Backends; +use crate::{backend::Backends, constants::*}; use std::net::SocketAddr; use std::sync::{ atomic::{AtomicUsize, Ordering}, @@ -6,25 +6,38 @@ use std::sync::{ }; use tokio::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 listen_sockets: Vec, - pub http_port: Option, - pub https_port: Option, + /// Configuration parameters for proxy transport and request handlers + pub proxy_config: ProxyConfig, // TODO: proxy configはarcに包んでこいつだけ使いまわせばいいように変えていく。backendsも? - pub proxy_timeout: Duration, - pub upstream_timeout: Duration, - - pub max_clients: usize, - pub request_count: RequestCount, - pub max_concurrent_streams: u32, - pub keepalive: bool, - - pub runtime_handle: tokio::runtime::Handle, + /// Backend application objects to which http request handler forward incoming requests pub backends: Backends, - // experimentals - pub sni_consistency: bool, + /// Shared context - Counter for serving requests + pub request_count: RequestCount, + /// Shared context - Async task runtime handler + pub runtime_handle: tokio::runtime::Handle, +} + +/// Configuration parameters for proxy transport and request handlers +pub struct ProxyConfig { + pub listen_sockets: Vec, // when instantiate server + pub http_port: Option, // when instantiate server + pub https_port: Option, // when instantiate server + + pub proxy_timeout: Duration, // when serving requests at Proxy + pub upstream_timeout: Duration, // when serving requests at Handler + + pub max_clients: usize, // when serving requests + pub max_concurrent_streams: u32, // when instantiate server + pub keepalive: bool, // when instantiate server + + // experimentals + pub sni_consistency: bool, // Handler + // All need to make packet acceptor #[cfg(feature = "http3")] pub http3: bool, #[cfg(feature = "http3")] @@ -41,7 +54,43 @@ pub struct Globals { pub h3_max_idle_timeout: Option, } +impl Default for ProxyConfig { + fn default() -> Self { + Self { + listen_sockets: Vec::new(), + http_port: None, + https_port: None, + + // TODO: Reconsider each timeout values + proxy_timeout: Duration::from_secs(PROXY_TIMEOUT_SEC), + upstream_timeout: Duration::from_secs(UPSTREAM_TIMEOUT_SEC), + + max_clients: MAX_CLIENTS, + max_concurrent_streams: MAX_CONCURRENT_STREAMS, + keepalive: true, + + sni_consistency: true, + + #[cfg(feature = "http3")] + http3: false, + #[cfg(feature = "http3")] + h3_alt_svc_max_age: H3::ALT_SVC_MAX_AGE, + #[cfg(feature = "http3")] + h3_request_max_body_size: H3::REQUEST_MAX_BODY_SIZE, + #[cfg(feature = "http3")] + h3_max_concurrent_connections: H3::MAX_CONCURRENT_CONNECTIONS, + #[cfg(feature = "http3")] + h3_max_concurrent_bidistream: H3::MAX_CONCURRENT_BIDISTREAM.into(), + #[cfg(feature = "http3")] + h3_max_concurrent_unistream: H3::MAX_CONCURRENT_UNISTREAM.into(), + #[cfg(feature = "http3")] + h3_max_idle_timeout: Some(quinn::IdleTimeout::try_from(Duration::from_secs(H3::MAX_IDLE_TIMEOUT)).unwrap()), + } + } +} + #[derive(Debug, Clone, Default)] +/// Counter for serving requests pub struct RequestCount(Arc); impl RequestCount { diff --git a/src/handler/handler_main.rs b/src/handler/handler_main.rs index a73dcbc..d2a47be 100644 --- a/src/handler/handler_main.rs +++ b/src/handler/handler_main.rs @@ -56,7 +56,7 @@ where }; // check consistency of between TLS SNI and HOST/Request URI Line. #[allow(clippy::collapsible_if)] - if tls_enabled && self.globals.sni_consistency { + if tls_enabled && self.globals.proxy_config.sni_consistency { if server_name != tls_server_name.unwrap_or_default() { return self.return_with_error_log(StatusCode::MISDIRECTED_REQUEST, &mut log_data); } @@ -75,7 +75,7 @@ where if !tls_enabled && backend.https_redirection.unwrap_or(false) { debug!("Redirect to secure connection: {}", &backend.server_name); log_data.status_code(&StatusCode::PERMANENT_REDIRECT).output(); - return secure_redirection(&backend.server_name, self.globals.https_port, &req); + return secure_redirection(&backend.server_name, self.globals.proxy_config.https_port, &req); } // Find reverse proxy for given path and choose one of upstream host @@ -112,7 +112,7 @@ where // Forward request to let mut res_backend = { - match timeout(self.globals.upstream_timeout, self.forwarder.request(req)).await { + match timeout(self.globals.proxy_config.upstream_timeout, self.forwarder.request(req)).await { Err(_) => { return self.return_with_error_log(StatusCode::GATEWAY_TIMEOUT, &mut log_data); } @@ -207,14 +207,14 @@ where #[cfg(feature = "http3")] { // TODO: Workaround for avoid h3 for client authentication - if self.globals.http3 && chosen_backend.client_ca_cert_path.is_none() { - if let Some(port) = self.globals.https_port { + if self.globals.proxy_config.http3 && chosen_backend.client_ca_cert_path.is_none() { + if let Some(port) = self.globals.proxy_config.https_port { add_header_entry_overwrite_if_exist( headers, header::ALT_SVC.as_str(), format!( "h3=\":{}\"; ma={}, h3-29=\":{}\"; ma={}", - port, self.globals.h3_alt_svc_max_age, port, self.globals.h3_alt_svc_max_age + port, self.globals.proxy_config.h3_alt_svc_max_age, port, self.globals.proxy_config.h3_alt_svc_max_age ), )?; } @@ -225,7 +225,7 @@ where } #[cfg(not(feature = "http3"))] { - if let Some(port) = self.globals.https_port { + if let Some(port) = self.globals.proxy_config.https_port { headers.remove(header::ALT_SVC.as_str()); } } diff --git a/src/main.rs b/src/main.rs index ea60fa7..ea47e57 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,6 +6,7 @@ use tikv_jemallocator::Jemalloc; static GLOBAL: Jemalloc = Jemalloc; mod backend; +mod cert_reader; mod config; mod constants; mod error; @@ -16,22 +17,12 @@ mod proxy; mod utils; use crate::{ - backend::{Backend, Backends}, - config::parse_opts, - constants::*, - error::*, - globals::*, - handler::HttpMessageHandlerBuilder, - log::*, - proxy::ProxyBuilder, - utils::ServerNameBytesExp, + config::build_globals, error::*, globals::*, handler::HttpMessageHandlerBuilder, log::*, proxy::ProxyBuilder, }; use futures::future::select_all; use hyper::Client; // use hyper_trust_dns::TrustDnsResolver; -use rustc_hash::FxHashMap as HashMap; use std::sync::Arc; -use tokio::time::Duration; fn main() { init_logger(); @@ -42,52 +33,17 @@ fn main() { let runtime = runtime_builder.build().unwrap(); runtime.block_on(async { - let mut globals = Globals { - listen_sockets: Vec::new(), - http_port: None, - https_port: None, - - // TODO: Reconsider each timeout values - proxy_timeout: Duration::from_secs(PROXY_TIMEOUT_SEC), - upstream_timeout: Duration::from_secs(UPSTREAM_TIMEOUT_SEC), - - max_clients: MAX_CLIENTS, - request_count: Default::default(), - max_concurrent_streams: MAX_CONCURRENT_STREAMS, - keepalive: true, - - runtime_handle: runtime.handle().clone(), - backends: Backends { - default_server_name_bytes: None, - apps: HashMap::::default(), - }, - - sni_consistency: true, - - #[cfg(feature = "http3")] - http3: false, - #[cfg(feature = "http3")] - h3_alt_svc_max_age: H3::ALT_SVC_MAX_AGE, - #[cfg(feature = "http3")] - h3_request_max_body_size: H3::REQUEST_MAX_BODY_SIZE, - #[cfg(feature = "http3")] - h3_max_concurrent_connections: H3::MAX_CONCURRENT_CONNECTIONS, - #[cfg(feature = "http3")] - h3_max_concurrent_bidistream: H3::MAX_CONCURRENT_BIDISTREAM.into(), - #[cfg(feature = "http3")] - h3_max_concurrent_unistream: H3::MAX_CONCURRENT_UNISTREAM.into(), - #[cfg(feature = "http3")] - h3_max_idle_timeout: Some(quinn::IdleTimeout::try_from(Duration::from_secs(H3::MAX_IDLE_TIMEOUT)).unwrap()), - }; - - if let Err(e) = parse_opts(&mut globals) { - error!("Invalid configuration: {}", e); - std::process::exit(1); + let globals = match build_globals(runtime.handle().clone()) { + Ok(g) => g, + Err(e) => { + error!("Invalid configuration: {}", e); + std::process::exit(1); + } }; entrypoint(Arc::new(globals)).await.unwrap() }); - warn!("Exit the program"); + warn!("rpxy exited!"); } // entrypoint creates and spawns tasks of proxy services @@ -105,10 +61,10 @@ async fn entrypoint(globals: Arc) -> Result<()> { .globals(globals.clone()) .build()?; - let addresses = globals.listen_sockets.clone(); + let addresses = globals.proxy_config.listen_sockets.clone(); let futures = select_all(addresses.into_iter().map(|addr| { let mut tls_enabled = false; - if let Some(https_port) = globals.https_port { + if let Some(https_port) = globals.proxy_config.https_port { tls_enabled = https_port == addr.port() } diff --git a/src/proxy/crypto_service.rs b/src/proxy/crypto_service.rs new file mode 100644 index 0000000..728a531 --- /dev/null +++ b/src/proxy/crypto_service.rs @@ -0,0 +1,253 @@ +use crate::{ + cert_reader::read_certs_and_keys, // TODO: Trait defining read_certs_and_keys and add struct implementing the trait to backend when build backend + globals::Globals, + log::*, + utils::ServerNameBytesExp, +}; +use async_trait::async_trait; +use hot_reload::*; +use rustc_hash::{FxHashMap as HashMap, FxHashSet as HashSet}; +use rustls::{ + server::ResolvesServerCertUsingSni, + sign::{any_supported_type, CertifiedKey}, + Certificate, OwnedTrustAnchor, PrivateKey, RootCertStore, ServerConfig, +}; +use std::{io, sync::Arc}; +use x509_parser::prelude::*; + +#[derive(Clone)] +/// Reloader service for certificates and keys for TLS +pub struct CryptoReloader { + globals: Arc, +} + +/// Certificates and private keys in rustls loaded from files +#[derive(Debug, PartialEq, Eq, Clone)] +pub struct CertsAndKeys { + pub certs: Vec, + pub cert_keys: Vec, + pub client_ca_certs: Option>, +} + +pub type SniServerCryptoMap = HashMap>; +pub struct ServerCrypto { + // For Quic/HTTP3, only servers with no client authentication + pub inner_global_no_client_auth: Arc, + // For TLS over TCP/HTTP2 and 1.1, map of SNI to server_crypto for all given servers + pub inner_local_map: Arc, +} + +/// Reloader target for the certificate reloader service +#[derive(Debug, PartialEq, Eq, Clone, Default)] +pub struct ServerCryptoBase { + inner: HashMap, +} + +#[async_trait] +impl Reload for CryptoReloader { + type Source = Arc; + async fn new(source: &Self::Source) -> Result> { + Ok(Self { + globals: source.clone(), + }) + } + + async fn reload(&self) -> Result, ReloaderError> { + let mut certs_and_keys_map = ServerCryptoBase::default(); + + for (server_name_bytes_exp, backend) in self.globals.backends.apps.iter() { + if backend.tls_cert_key_path.is_some() && backend.tls_cert_path.is_some() { + let tls_cert_key_path = backend.tls_cert_key_path.as_ref().unwrap(); + let tls_cert_path = backend.tls_cert_path.as_ref().unwrap(); + let tls_client_ca_cert_path = backend.client_ca_cert_path.as_ref(); + let certs_and_keys = read_certs_and_keys(tls_cert_path, tls_cert_key_path, tls_client_ca_cert_path) + .map_err(|_e| ReloaderError::::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 CertsAndKeys { + fn parse_server_certs_and_keys(&self) -> Result { + // 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, HashSet>), 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::>(); + if !subject_key_ids.is_empty() { + Some(subject_key_ids[0].0.to_owned()) + } else { + None + } + }) + .collect(); + + Ok((owned_trust_anchors, subject_key_identifiers)) + } +} + +impl TryInto> for &ServerCryptoBase { + type Error = anyhow::Error; + + fn try_into(self) -> Result, Self::Error> { + let mut resolver_global = ResolvesServerCertUsingSni::new(); + 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 = 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_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 + ) + } + } else { + // 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_server_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(feature = "http3"))] + { + ServerConfig::builder() + .with_safe_defaults() + .with_no_client_auth() + .with_cert_resolver(Arc::new(resolver_local)) + } + #[cfg(feature = "http3")] + { + 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)); + } + + ////////////// + let mut server_crypto_global = ServerConfig::builder() + .with_safe_defaults() + .with_no_client_auth() + .with_cert_resolver(Arc::new(resolver_global)); + + ////////////////////////////// + + #[cfg(feature = "http3")] + { + 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(), + ]; + } + #[cfg(not(feature = "http3"))] + { + server_crypto_global.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()]; + } + + Ok(Arc::new(ServerCrypto { + inner_global_no_client_auth: Arc::new(server_crypto_global), + inner_local_map: Arc::new(server_crypto_local_map), + })) + } +} diff --git a/src/proxy/mod.rs b/src/proxy/mod.rs index 04413f5..d8fdc83 100644 --- a/src/proxy/mod.rs +++ b/src/proxy/mod.rs @@ -1,7 +1,9 @@ +mod crypto_service; mod proxy_client_cert; #[cfg(feature = "http3")] mod proxy_h3; mod proxy_main; mod proxy_tls; +pub use crypto_service::CertsAndKeys; pub use proxy_main::{Proxy, ProxyBuilder, ProxyBuilderError}; diff --git a/src/proxy/proxy_client_cert.rs b/src/proxy/proxy_client_cert.rs index adac4b7..dfba4ce 100644 --- a/src/proxy/proxy_client_cert.rs +++ b/src/proxy/proxy_client_cert.rs @@ -10,26 +10,18 @@ pub(super) fn check_client_authentication( client_certs: Option<&[Certificate]>, client_ca_keyids_set_for_sni: Option<&HashSet>>, ) -> std::result::Result<(), ClientCertsError> { - let client_ca_keyids_set = match client_ca_keyids_set_for_sni { - Some(c) => c, - None => { - // No client cert settings for given server name - return Ok(()); - } + let Some(client_ca_keyids_set) = client_ca_keyids_set_for_sni else { + // No client cert settings for given server name + return Ok(()); }; - let client_certs = match client_certs { - Some(c) => { - debug!("Incoming TLS client is (temporarily) authenticated via client cert"); - c - } - None => { - error!("Client certificate is needed for given server name"); - return Err(ClientCertsError::ClientCertRequired( - "Client certificate is needed for given server name".to_string(), - )); - } + let Some(client_certs) = client_certs else { + error!("Client certificate is needed for given server name"); + return Err(ClientCertsError::ClientCertRequired( + "Client certificate is needed for given server name".to_string(), + )); }; + debug!("Incoming TLS client is (temporarily) authenticated via client cert"); // Check client certificate key ids let mut client_certs_parsed_iter = client_certs.iter().filter_map(|d| parse_x509_certificate(&d.0).ok()); diff --git a/src/proxy/proxy_h3.rs b/src/proxy/proxy_h3.rs index d5a6c88..12ebd7d 100644 --- a/src/proxy/proxy_h3.rs +++ b/src/proxy/proxy_h3.rs @@ -43,7 +43,7 @@ where // We consider the connection count separately from the stream count. // Max clients for h1/h2 = max 'stream' for h3. let request_count = self.globals.request_count.clone(); - if request_count.increment() > self.globals.max_clients { + if request_count.increment() > self.globals.proxy_config.max_clients { request_count.decrement(); h3_conn.shutdown(0).await?; break; @@ -54,7 +54,7 @@ where let tls_server_name_inner = tls_server_name.clone(); self.globals.runtime_handle.spawn(async move { if let Err(e) = timeout( - self_inner.globals.proxy_timeout + Duration::from_secs(1), // timeout per stream are considered as same as one in http2 + self_inner.globals.proxy_config.proxy_timeout + Duration::from_secs(1), // timeout per stream are considered as same as one in http2 self_inner.stream_serve_h3(req, stream, client_addr, tls_server_name_inner), ) .await @@ -97,7 +97,7 @@ where // Buffering and sending body through channel for protocol conversion like h3 -> h2/http1.1 // The underling buffering, i.e., buffer given by the API recv_data.await?, is handled by quinn. - let max_body_size = self.globals.h3_request_max_body_size; + let max_body_size = self.globals.proxy_config.h3_request_max_body_size; self.globals.runtime_handle.spawn(async move { let mut sender = body_sender; let mut size = 0usize; diff --git a/src/proxy/proxy_main.rs b/src/proxy/proxy_main.rs index 722ef3c..a0f9660 100644 --- a/src/proxy/proxy_main.rs +++ b/src/proxy/proxy_main.rs @@ -56,7 +56,7 @@ where I: AsyncRead + AsyncWrite + Send + Unpin + 'static, { let request_count = self.globals.request_count.clone(); - if request_count.increment() > self.globals.max_clients { + if request_count.increment() > self.globals.proxy_config.max_clients { request_count.decrement(); return; } @@ -64,7 +64,7 @@ where self.globals.runtime_handle.clone().spawn(async move { timeout( - self.globals.proxy_timeout + Duration::from_secs(1), + self.globals.proxy_config.proxy_timeout + Duration::from_secs(1), server .serve_connection( stream, @@ -103,8 +103,8 @@ where pub async fn start(self) -> Result<()> { let mut server = Http::new(); - server.http1_keep_alive(self.globals.keepalive); - server.http2_max_concurrent_streams(self.globals.max_concurrent_streams); + server.http1_keep_alive(self.globals.proxy_config.keepalive); + server.http2_max_concurrent_streams(self.globals.proxy_config.max_concurrent_streams); server.pipeline_flush(true); let executor = LocalExecutor::new(self.globals.runtime_handle.clone()); let server = server.with_executor(executor); diff --git a/src/proxy/proxy_tls.rs b/src/proxy/proxy_tls.rs index de18e0c..e01f9d3 100644 --- a/src/proxy/proxy_tls.rs +++ b/src/proxy/proxy_tls.rs @@ -1,11 +1,9 @@ -use super::proxy_main::{LocalExecutor, Proxy}; -use crate::{ - backend::{ServerCrypto, SniServerCryptoMap}, - constants::*, - error::*, - log::*, - utils::BytesName, +use super::{ + crypto_service::{CryptoReloader, ServerCrypto, ServerCryptoBase, SniServerCryptoMap}, + proxy_main::{LocalExecutor, Proxy}, }; +use crate::{constants::*, error::*, log::*, utils::BytesName}; +use hot_reload::{ReloaderReceiver, ReloaderService}; use hyper::{client::connect::Connect, server::conn::Http}; #[cfg(feature = "http3")] use quinn::{crypto::rustls::HandshakeData, Endpoint, ServerConfig as QuicServerConfig, TransportConfig}; @@ -14,34 +12,18 @@ use rustls::ServerConfig; use std::sync::Arc; use tokio::{ net::TcpListener, - sync::watch, - time::{sleep, timeout, Duration}, + time::{timeout, Duration}, }; impl Proxy where T: Connect + Clone + Sync + Send + 'static, { - async fn cert_service(&self, server_crypto_tx: watch::Sender>>) { - info!("Start cert watch service"); - loop { - if let Ok(server_crypto) = self.globals.backends.generate_server_crypto().await { - if let Err(_e) = server_crypto_tx.send(Some(Arc::new(server_crypto))) { - error!("Failed to populate server crypto"); - break; - } - } else { - error!("Failed to update certs"); - } - sleep(Duration::from_secs(CERTS_WATCH_DELAY_SECS.into())).await; - } - } - // TCP Listener Service, i.e., http/2 and http/1.1 async fn listener_service( &self, server: Http, - mut server_crypto_rx: watch::Receiver>>, + mut server_crypto_rx: ReloaderReceiver, ) -> Result<()> { let tcp_listener = TcpListener::bind(&self.listening_on).await?; info!("Start TCP proxy serving with HTTPS request for configured host names"); @@ -105,9 +87,14 @@ where } _ = server_crypto_rx.changed() => { if server_crypto_rx.borrow().is_none() { + error!("Reloader is broken"); break; } - let server_crypto = server_crypto_rx.borrow().clone().unwrap(); + let cert_keys_map = server_crypto_rx.borrow().clone().unwrap(); + let Some(server_crypto): Option> = (&cert_keys_map).try_into().ok() else { + error!("Failed to update server crypto"); + break; + }; server_crypto_map = Some(server_crypto.inner_local_map.clone()); } else => break @@ -117,7 +104,7 @@ where } #[cfg(feature = "http3")] - async fn listener_service_h3(&self, mut server_crypto_rx: watch::Receiver>>) -> Result<()> { + async fn listener_service_h3(&self, mut server_crypto_rx: ReloaderReceiver) -> Result<()> { info!("Start UDP proxy serving with HTTP/3 request for configured host names"); // first set as null config server let rustls_server_config = ServerConfig::builder() @@ -129,13 +116,13 @@ where let mut transport_config_quic = TransportConfig::default(); transport_config_quic - .max_concurrent_bidi_streams(self.globals.h3_max_concurrent_bidistream) - .max_concurrent_uni_streams(self.globals.h3_max_concurrent_unistream) - .max_idle_timeout(self.globals.h3_max_idle_timeout); + .max_concurrent_bidi_streams(self.globals.proxy_config.h3_max_concurrent_bidistream) + .max_concurrent_uni_streams(self.globals.proxy_config.h3_max_concurrent_unistream) + .max_idle_timeout(self.globals.proxy_config.h3_max_idle_timeout); let mut server_config_h3 = QuicServerConfig::with_crypto(Arc::new(rustls_server_config)); server_config_h3.transport = Arc::new(transport_config_quic); - server_config_h3.concurrent_connections(self.globals.h3_max_concurrent_connections); + server_config_h3.concurrent_connections(self.globals.proxy_config.h3_max_concurrent_connections); let endpoint = Endpoint::server(server_config_h3, self.listening_on)?; let mut server_crypto: Option> = None; @@ -146,29 +133,24 @@ where continue; } let mut conn: quinn::Connecting = new_conn.unwrap(); - let hsd = match conn.handshake_data().await { - Ok(h) => h, - Err(_) => continue + let Ok(hsd) = conn.handshake_data().await else { + continue }; - let hsd_downcast = match hsd.downcast::() { - Ok(d) => d, - Err(_) => continue + let Ok(hsd_downcast) = hsd.downcast::() else { + continue }; - let new_server_name = match hsd_downcast.server_name { - Some(sn) => sn.to_server_name_vec(), - None => { - warn!("HTTP/3 no SNI is given"); - continue; - } + let Some(new_server_name) = hsd_downcast.server_name else { + warn!("HTTP/3 no SNI is given"); + continue; }; debug!( "HTTP/3 connection incoming (SNI {:?})", - new_server_name.0 + new_server_name ); // TODO: server_nameをここで出してどんどん深く投げていくのは効率が悪い。connecting -> connectionsの後でいいのでは? // TODO: 通常のTLSと同じenumか何かにまとめたい - let fut = self.clone().connection_serve_h3(conn, new_server_name); + let fut = self.clone().connection_serve_h3(conn, new_server_name.to_server_name_vec()); self.globals.runtime_handle.spawn(async move { // Timeout is based on underlying quic if let Err(e) = fut.await { @@ -178,12 +160,18 @@ where } _ = server_crypto_rx.changed() => { if server_crypto_rx.borrow().is_none() { + error!("Reloader is broken"); break; } - server_crypto = server_crypto_rx.borrow().clone(); - if server_crypto.is_some(){ - endpoint.set_server_config(Some(QuicServerConfig::with_crypto(server_crypto.clone().unwrap().inner_global_no_client_auth.clone()))); - } + let cert_keys_map = server_crypto_rx.borrow().clone().unwrap(); + + server_crypto = (&cert_keys_map).try_into().ok(); + let Some(inner) = server_crypto.clone() else { + error!("Failed to update server crypto for h3"); + break; + }; + endpoint.set_server_config(Some(QuicServerConfig::with_crypto(inner.clone().inner_global_no_client_auth.clone()))); + } else => break } @@ -193,7 +181,14 @@ where } pub async fn start_with_tls(self, server: Http) -> Result<()> { - let (tx, rx) = watch::channel::>>(None); + let (cert_reloader_service, cert_reloader_rx) = ReloaderService::::new( + &self.globals.clone(), + CERTS_WATCH_DELAY_SECS, + !LOAD_CERTS_ONLY_WHEN_UPDATED, + ) + .await + .map_err(|e| anyhow::anyhow!(e))?; + #[cfg(not(feature = "http3"))] { tokio::select! { @@ -212,15 +207,15 @@ where } #[cfg(feature = "http3")] { - if self.globals.http3 { + if self.globals.proxy_config.http3 { tokio::select! { - _= self.cert_service(tx) => { + _= cert_reloader_service.start() => { error!("Cert service for TLS exited"); }, - _ = self.listener_service(server, rx.clone()) => { + _ = self.listener_service(server, cert_reloader_rx.clone()) => { error!("TCP proxy service for TLS exited"); }, - _= self.listener_service_h3(rx) => { + _= self.listener_service_h3(cert_reloader_rx) => { error!("UDP proxy service for QUIC exited"); }, else => { @@ -231,10 +226,10 @@ where Ok(()) } else { tokio::select! { - _= self.cert_service(tx) => { + _= cert_reloader_service.start() => { error!("Cert service for TLS exited"); }, - _ = self.listener_service(server, rx) => { + _ = self.listener_service(server, cert_reloader_rx) => { error!("TCP proxy service for TLS exited"); }, else => { diff --git a/src/utils/bytes_name.rs b/src/utils/bytes_name.rs index 16ec7ab..a093c41 100644 --- a/src/utils/bytes_name.rs +++ b/src/utils/bytes_name.rs @@ -7,6 +7,13 @@ impl From<&[u8]> for ServerNameBytesExp { Self(b.to_ascii_lowercase()) } } +impl TryInto for &ServerNameBytesExp { + type Error = anyhow::Error; + fn try_into(self) -> Result { + let s = std::str::from_utf8(&self.0)?; + Ok(s.to_string()) + } +} /// Path name, like "/path/ok", represented in bytes-based struct /// for searching hashmap or key list by exact or longest-prefix matching