tuxmain/rust-rpxy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: cleanup codes

Browse source
This commit is contained in:
Jun Kurihara 2023-07-21 22:07:36 +09:00
commit f6c4032f83
No known key found for this signature in database
GPG key ID: 6D3FEE70E498C15B
10 changed files with 280 additions and 124 deletions
Download patch file Download diff file Expand all files Collapse all files

8
rpxy-bin/Cargo.toml
View file

@ -12,6 +12,8 @@ publish = false
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[features]
default = ["http3"]
http3 = []
[dependencies]
rpxy-lib = { path = "../rpxy-lib/", features = ["http3", "sticky-cookie"] }
@ -28,14 +30,12 @@ tokio = { version = "1.29.1", default-features = false, features = [
"macros",
] }
async-trait = "0.1.72"
rustls-pemfile = "1.0.3"
# config
clap = { version = "4.3.17", features = ["std", "cargo", "wrap_help"] }
toml = { version = "0.7.6", default-features = false, features = ["parse"] }
# reloading certs
hot_reload = "0.1.2"
rustls-pemfile = "1.0.3"
# hot_reload = "0.1.2"
# logging
tracing = { version = "0.1.37" }

10
rpxy-bin/src/cert_file_reader.rs
View file

@ -150,8 +150,8 @@ mod tests {
use super::*;
#[tokio::test]
async fn read_server_crt_key_files() {
let tls_cert_path = "example-certs/server.crt";
let tls_cert_key_path = "example-certs/server.key";
let tls_cert_path = "../example-certs/server.crt";
let tls_cert_key_path = "../example-certs/server.key";
let crypto_file_source = CryptoFileSourceBuilder::default()
.tls_cert_key_path(tls_cert_key_path)
.tls_cert_path(tls_cert_path)
@ -165,9 +165,9 @@ mod tests {
#[tokio::test]
async fn read_server_crt_key_files_with_client_ca_crt() {
let tls_cert_path = "example-certs/server.crt";
let tls_cert_key_path = "example-certs/server.key";
let client_ca_cert_path = Some("example-certs/client.ca.crt".to_string());
let tls_cert_path = "../example-certs/server.crt";
let tls_cert_key_path = "../example-certs/server.key";
let client_ca_cert_path = Some("../example-certs/client.ca.crt".to_string());
let crypto_file_source = CryptoFileSourceBuilder::default()
.tls_cert_key_path(tls_cert_key_path)
.tls_cert_path(tls_cert_path)

2
rpxy-bin/src/config/mod.rs
View file

@ -1,4 +1,4 @@
mod parse;
mod toml;
pub use parse::build_globals;
pub use parse::build_settings;

44
rpxy-bin/src/config/parse.rs
View file

@ -5,10 +5,9 @@ use crate::{
log::*,
};
use clap::Arg;
use rpxy_lib::{Backends, BytesName, Globals, ProxyConfig};
use tokio::runtime::Handle;
use rpxy_lib::{AppConfig, AppConfigList, ProxyConfig};
pub fn build_globals(runtime_handle: Handle) -> std::result::Result<Globals<CryptoFileSource>, anyhow::Error> {
pub fn build_settings() -> std::result::Result<(ProxyConfig, AppConfigList<CryptoFileSource>), anyhow::Error> {
let _ = include_str!("../../Cargo.toml");
let options = clap::command!().arg(
Arg::new("config_file")
@ -76,39 +75,22 @@ pub fn build_globals(runtime_handle: Handle) -> std::result::Result<Globals<Cryp
);
}
// build backends
let mut backends = Backends::new();
// build applications
let mut app_config_list_inner = Vec::<AppConfig<CryptoFileSource>>::new();
// let mut backends = Backends::new();
for (app_name, app) in apps.0.iter() {
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);
let app_config = app.try_into()?;
app_config_list_inner.push(app_config);
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> = backends
.apps
.iter()
.filter(|(_k, v)| v.app_name == d)
.map(|(_, v)| v.server_name.as_ref())
.collect();
if !d_sn.is_empty() {
info!(
"Serving plaintext http for requests to unconfigured server_name by app {} (server_name: {}).",
d, d_sn[0]
);
backends.default_server_name_bytes = Some(d_sn[0].to_server_name_vec());
}
}
///////////////////////////////////
let globals = Globals {
proxy_config,
backends,
request_count: Default::default(),
runtime_handle,
let app_config_list = AppConfigList {
inner: app_config_list_inner,
default_app: config.default_app, // default backend application for plaintext http requests
};
Ok(globals)
Ok((proxy_config, app_config_list))
// todo!()
}

116
rpxy-bin/src/config/toml.rs
View file

@ -3,10 +3,7 @@ use crate::{
constants::*,
error::{anyhow, ensure},
};
use rpxy_lib::{
reexports::Uri, Backend, BackendBuilder, PathNameBytesExp, ProxyConfig, ReverseProxy, Upstream, UpstreamGroup,
UpstreamGroupBuilder, UpstreamOption,
};
use rpxy_lib::{reexports::Uri, AppConfig, ProxyConfig, ReverseProxyConfig, TlsConfig, UpstreamUri};
use rustc_hash::FxHashMap as HashMap;
use serde::Deserialize;
use std::{fs, net::SocketAddr};
@ -148,8 +145,7 @@ impl TryInto<ProxyConfig> for &ConfigToml {
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())
proxy_config.h3_max_idle_timeout = Some(tokio::time::Duration::from_secs(x))
}
}
}
@ -172,101 +168,87 @@ impl ConfigToml {
}
}
impl TryInto<Backend<CryptoFileSource>> for &Application {
impl TryInto<AppConfig<CryptoFileSource>> for &Application {
type Error = anyhow::Error;
fn try_into(self) -> std::result::Result<Backend<CryptoFileSource>, Self::Error> {
fn try_into(self) -> std::result::Result<AppConfig<CryptoFileSource>, Self::Error> {
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()?;
let reverse_proxy_config: Vec<ReverseProxyConfig> = 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 {
// tls settings
let tls_config = if self.tls.is_some() {
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
};
let crypto_source = CryptoFileSourceBuilder::default()
let inner = CryptoFileSourceBuilder::default()
.tls_cert_path(tls.tls_cert_path.as_ref().unwrap())
.tls_cert_key_path(tls.tls_cert_key_path.as_ref().unwrap())
.client_ca_cert_path(&tls.client_ca_cert_path)
.build()?;
backend_builder
.https_redirection(https_redirection)
.crypto_source(Some(crypto_source))
.build()?
let https_redirection = if tls.https_redirection.is_none() {
true // Default true
} else {
tls.https_redirection.unwrap()
};
Ok(backend)
Some(TlsConfig {
inner,
https_redirection,
})
} else {
None
};
Ok(AppConfig {
server_name: server_name_string.to_owned(),
reverse_proxy: reverse_proxy_config,
tls: tls_config,
})
}
}
impl TryInto<ReverseProxy> for &Application {
impl TryInto<Vec<ReverseProxyConfig>> for &Application {
type Error = anyhow::Error;
fn try_into(self) -> std::result::Result<ReverseProxy, Self::Error> {
let server_name_string = self.server_name.as_ref().ok_or(anyhow!("Missing server_name"))?;
fn try_into(self) -> std::result::Result<Vec<ReverseProxyConfig>, Self::Error> {
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<PathNameBytesExp, UpstreamGroup> = HashMap::default();
let mut reverse_proxies: Vec<ReverseProxyConfig> = Vec::new();
rp_settings.iter().for_each(|rpo| {
let upstream_vec: Vec<Upstream> = 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();
for rpo in rp_settings.iter() {
let upstream_res: Vec<Option<UpstreamUri>> = rpo.upstream.iter().map(|v| v.try_into().ok()).collect();
if !upstream_res.iter().all(|v| v.is_some()) {
return Err(anyhow!("[{}] Upstream uri is invalid", &_server_name_string));
}
let upstream = upstream_res.into_iter().map(|v| v.unwrap()).collect();
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"
);
reverse_proxies.push(ReverseProxyConfig {
path: rpo.path.clone(),
replace_path: rpo.replace_path.clone(),
upstream,
upstream_options: rpo.upstream_options.clone(),
load_balance: rpo.load_balance.clone(),
})
}
Ok(ReverseProxy { upstream })
Ok(reverse_proxies)
}
}
impl TryInto<Upstream> for &UpstreamParams {
impl TryInto<UpstreamUri> for &UpstreamParams {
type Error = anyhow::Error;
fn try_into(self) -> std::result::Result<Upstream, Self::Error> {
fn try_into(self) -> std::result::Result<UpstreamUri, Self::Error> {
let scheme = match self.tls {
Some(true) => "https",
_ => "http",
};
let location = format!("{}://{}", scheme, self.location);
Ok(Upstream {
uri: location.parse::<Uri>().map_err(|e| anyhow!("{}", e))?,
Ok(UpstreamUri {
inner: location.parse::<Uri>().map_err(|e| anyhow!("{}", e))?,
})
}
}

11
rpxy-bin/src/main.rs
View file

@ -11,9 +11,8 @@ mod constants;
mod error;
mod log;
use crate::{cert_file_reader::CryptoFileSource, config::build_globals, log::*};
use rpxy_lib::{entrypoint, Globals};
use std::sync::Arc;
use crate::{config::build_settings, log::*};
use rpxy_lib::entrypoint;
fn main() {
init_logger();
@ -24,7 +23,7 @@ fn main() {
let runtime = runtime_builder.build().unwrap();
runtime.block_on(async {
let globals: Globals<CryptoFileSource> = match build_globals(runtime.handle().clone()) {
let (proxy_conf, app_conf) = match build_settings() {
Ok(g) => g,
Err(e) => {
error!("Invalid configuration: {}", e);
@ -32,7 +31,9 @@ fn main() {
}
};
entrypoint(Arc::new(globals)).await.unwrap()
entrypoint(proxy_conf, app_conf, runtime.handle().clone())
.await
.unwrap()
});
warn!("rpxy exited!");
}

6
rpxy-lib/src/error.rs
View file

@ -10,9 +10,15 @@ pub enum RpxyError {
#[error("Proxy build error")]
ProxyBuild(#[from] crate::proxy::ProxyBuilderError),
#[error("Backend build error")]
BackendBuild(#[from] crate::backend::BackendBuilderError),
#[error("MessageHandler build error")]
HandlerBuild(#[from] crate::handler::HttpMessageHandlerBuilderError),
#[error("Config builder error: {0}")]
ConfigBuild(&'static str),
#[error("Http Message Handler Error: {0}")]
Handler(&'static str),

178
rpxy-lib/src/globals.rs
View file

@ -1,4 +1,14 @@
use crate::{backend::Backends, certs::CryptoSource, constants::*};
use crate::{
backend::{
Backend, BackendBuilder, Backends, ReverseProxy, Upstream, UpstreamGroup, UpstreamGroupBuilder, UpstreamOption,
},
certs::CryptoSource,
constants::*,
error::RpxyError,
log::*,
utils::{BytesName, PathNameBytesExp},
};
use rustc_hash::FxHashMap as HashMap;
use std::net::SocketAddr;
use std::sync::{
atomic::{AtomicUsize, Ordering},
@ -26,6 +36,7 @@ where
}
/// Configuration parameters for proxy transport and request handlers
#[derive(PartialEq, Eq)]
pub struct ProxyConfig {
pub listen_sockets: Vec<SocketAddr>, // when instantiate server
pub http_port: Option<u16>, // when instantiate server
@ -54,7 +65,7 @@ pub struct ProxyConfig {
#[cfg(feature = "http3")]
pub h3_max_concurrent_connections: u32,
#[cfg(feature = "http3")]
pub h3_max_idle_timeout: Option<quinn::IdleTimeout>,
pub h3_max_idle_timeout: Option<Duration>,
}
impl Default for ProxyConfig {
@ -87,11 +98,172 @@ impl Default for ProxyConfig {
#[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()),
h3_max_idle_timeout: Some(Duration::from_secs(H3::MAX_IDLE_TIMEOUT)),
}
}
}
/// Configuration parameters for backend applications
#[derive(PartialEq, Eq)]
pub struct AppConfigList<T>
where
T: CryptoSource,
{
pub inner: Vec<AppConfig<T>>,
pub default_app: Option<String>,
}
impl<T> TryInto<Backends<T>> for AppConfigList<T>
where
T: CryptoSource + Clone,
{
type Error = RpxyError;
fn try_into(self) -> Result<Backends<T>, Self::Error> {
let mut backends = Backends::new();
for app_config in self.inner.iter() {
let backend = app_config.try_into()?;
backends
.apps
.insert(app_config.server_name.clone().to_server_name_vec(), backend);
info!("Registering application: ({})", &app_config.server_name);
}
// default backend application for plaintext http requests
if let Some(d) = self.default_app {
let d_sn: Vec<&str> = backends
.apps
.iter()
.filter(|(_k, v)| v.app_name == d)
.map(|(_, v)| v.server_name.as_ref())
.collect();
if !d_sn.is_empty() {
info!(
"Serving plaintext http for requests to unconfigured server_name by app {} (server_name: {}).",
d, d_sn[0]
);
backends.default_server_name_bytes = Some(d_sn[0].to_server_name_vec());
}
}
Ok(backends)
}
}
/// Configuration parameters for single backend application
#[derive(PartialEq, Eq)]
pub struct AppConfig<T>
where
T: CryptoSource,
{
pub server_name: String,
pub reverse_proxy: Vec<ReverseProxyConfig>,
pub tls: Option<TlsConfig<T>>,
}
impl<T> TryInto<Backend<T>> for &AppConfig<T>
where
T: CryptoSource + Clone,
{
type Error = RpxyError;
fn try_into(self) -> Result<Backend<T>, Self::Error> {
// backend builder
let mut backend_builder = BackendBuilder::default();
// reverse proxy settings
let reverse_proxy = self.try_into()?;
backend_builder
.app_name(self.server_name.clone())
.server_name(self.server_name.clone())
.reverse_proxy(reverse_proxy);
// TLS settings and build backend instance
let backend = if self.tls.is_none() {
backend_builder.build().map_err(RpxyError::BackendBuild)?
} else {
let tls = self.tls.as_ref().unwrap();
backend_builder
.https_redirection(Some(tls.https_redirection))
.crypto_source(Some(tls.inner.clone()))
.build()?
};
Ok(backend)
}
}
impl<T> TryInto<ReverseProxy> for &AppConfig<T>
where
T: CryptoSource + Clone,
{
type Error = RpxyError;
fn try_into(self) -> Result<ReverseProxy, Self::Error> {
let mut upstream: HashMap<PathNameBytesExp, UpstreamGroup> = HashMap::default();
self.reverse_proxy.iter().for_each(|rpo| {
let upstream_vec: Vec<Upstream> = 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, &self.server_name, &rpo.path)
.opts(&rpo.upstream_options)
.build()
.unwrap();
upstream.insert(elem.path.clone(), elem);
});
if self.reverse_proxy.iter().filter(|rpo| rpo.path.is_none()).count() >= 2 {
error!("Multiple default reverse proxy setting");
return Err(RpxyError::ConfigBuild("Invalid reverse proxy setting"));
}
if !(upstream.iter().all(|(_, elem)| {
!(elem.opts.contains(&UpstreamOption::ConvertHttpsTo11) && elem.opts.contains(&UpstreamOption::ConvertHttpsTo2))
})) {
error!("Either one of force_http11 or force_http2 can be enabled");
return Err(RpxyError::ConfigBuild("Invalid upstream option setting"));
}
Ok(ReverseProxy { upstream })
}
}
/// Configuration parameters for single reverse proxy corresponding to the path
#[derive(PartialEq, Eq)]
pub struct ReverseProxyConfig {
pub path: Option<String>,
pub replace_path: Option<String>,
pub upstream: Vec<UpstreamUri>,
pub upstream_options: Option<Vec<String>>,
pub load_balance: Option<String>,
}
/// Configuration parameters for single upstream destination from a reverse proxy
#[derive(PartialEq, Eq)]
pub struct UpstreamUri {
pub inner: hyper::Uri,
}
impl TryInto<Upstream> for &UpstreamUri {
type Error = anyhow::Error;
fn try_into(self) -> std::result::Result<Upstream, Self::Error> {
Ok(Upstream {
uri: self.inner.clone(),
})
}
}
/// Configuration parameters on TLS for a single backend application
#[derive(PartialEq, Eq)]
pub struct TlsConfig<T>
where
T: CryptoSource,
{
pub inner: T,
pub https_redirection: bool,
}
#[derive(Debug, Clone, Default)]
/// Counter for serving requests
pub struct RequestCount(Arc<AtomicUsize>);

21
rpxy-lib/src/lib.rs
View file

@ -8,19 +8,15 @@ mod log;
mod proxy;
mod utils;
use crate::{error::*, handler::HttpMessageHandlerBuilder, log::*, proxy::ProxyBuilder};
use crate::{error::*, globals::Globals, handler::HttpMessageHandlerBuilder, log::*, proxy::ProxyBuilder};
use futures::future::select_all;
use hyper::Client;
// use hyper_trust_dns::TrustDnsResolver;
use std::sync::Arc;
pub use crate::{
backend::{
Backend, BackendBuilder, Backends, ReverseProxy, Upstream, UpstreamGroup, UpstreamGroupBuilder, UpstreamOption,
},
certs::{CertsAndKeys, CryptoSource},
globals::{Globals, ProxyConfig}, // TODO: BackendConfigに変える
utils::{BytesName, PathNameBytesExp},
globals::{AppConfig, AppConfigList, ProxyConfig, ReverseProxyConfig, TlsConfig, UpstreamUri},
};
pub mod reexports {
pub use hyper::Uri;
@ -28,10 +24,21 @@ pub mod reexports {
}
/// Entrypoint that creates and spawns tasks of reverse proxy services
pub async fn entrypoint<T>(globals: Arc<Globals<T>>) -> Result<()>
pub async fn entrypoint<T>(
proxy_config: ProxyConfig,
app_config_list: AppConfigList<T>,
runtime_handle: tokio::runtime::Handle,
) -> Result<()>
where
T: CryptoSource + Clone + Send + Sync + 'static,
{
// build global
let globals = Arc::new(Globals {
proxy_config,
backends: app_config_list.try_into()?,
request_count: Default::default(),
runtime_handle,
});
// let connector = TrustDnsResolver::default().into_rustls_webpki_https_connector();
let connector = hyper_rustls::HttpsConnectorBuilder::new()
.with_webpki_roots()

8
rpxy-lib/src/proxy/proxy_tls.rs
View file

@ -119,7 +119,13 @@ where
transport_config_quic
.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);
.max_idle_timeout(
self
.globals
.proxy_config
.h3_max_idle_timeout
.map(|v| quinn::IdleTimeout::try_from(v).unwrap()),
);
let mut server_config_h3 = QuicServerConfig::with_crypto(Arc::new(rustls_server_config));
server_config_h3.transport = Arc::new(transport_config_quic);