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

refine deps

Browse source
This commit is contained in:
Jun Kurihara 2022-07-22 23:32:14 +09:00
commit 66e791f34b
No known key found for this signature in database
GPG key ID: 48ADFD173ED22B03
11 changed files with 85 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

24
Cargo.toml
View file

@ -12,14 +12,14 @@ publish = false
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[features]
default = ["h3"]
h3 = ["quinn"]
default = ["http3"]
http3 = ["quinn", "h3", "h3-quinn"]
[dependencies]
env_logger = "0.9.0"
anyhow = "1.0.58"
clap = { version = "3.2.13", features = ["std", "cargo", "wrap_help"] }
env_logger = "0.9.0"
futures = "0.3.21"
futures = { version = "0.3.21", features = ["alloc", "async-await"] }
hyper = { version = "0.14.20", default-features = false, features = [
"server",
"http1",
@ -36,26 +36,28 @@ tokio = { version = "1.20.0", default-features = false, features = [
"macros",
] }
tokio-rustls = { version = "0.23.4", features = ["early-data"] }
rustls-pemfile = { version = "1.0.0" }
rustls = "0.20.6"
rustls-pemfile = "1.0.0"
rustls = { version = "0.20.6", default-features = false }
rand = "0.8.5"
toml = "0.5.9"
toml = { version = "0.5.9", default-features = false }
rustc-hash = "1.1.0"
serde = { version = "1.0.139", features = ["derive"] }
serde = { version = "1.0.140", default-features = false, features = ["derive"] }
hyper-rustls = { version = "0.23.0", default-features = false, features = [
"tokio-runtime",
"webpki-tokio",
"http1",
"http2",
] }
quinn = { version = "0.8.3", optional = true }
h3 = { git = "https://github.com/hyperium/h3.git" }
h3-quinn = { git = "https://github.com/hyperium/h3.git" }
bytes = "1.2.0"
quinn = { version = "0.8.3", optional = true }
h3 = { git = "https://github.com/hyperium/h3.git", optional = true }
h3-quinn = { git = "https://github.com/hyperium/h3.git", optional = true }
[target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemallocator = "0.5.0"
[dev-dependencies]

6
src/backend.rs
View file

@ -236,16 +236,16 @@ impl Backends {
.with_no_client_auth()
.with_cert_resolver(Arc::new(resolver));
#[cfg(feature = "h3")]
#[cfg(feature = "http3")]
{
server_config.alpn_protocols = vec![
b"h3".to_vec(),
b"hq-29".to_vec(), // quinn draft example TODO: remove later
b"hq-29".to_vec(), // TODO: remove later?
b"h2".to_vec(),
b"http/1.1".to_vec(),
];
}
#[cfg(not(feature = "h3"))]
#[cfg(not(feature = "http3"))]
{
server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
}

38
src/config/parse.rs
View file

@ -157,25 +157,29 @@ pub fn parse_opts(globals: &mut Globals) -> Result<()> {
// experimental
if let Some(exp) = config.experimental {
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();
#[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(b) = exp.ignore_sni_consistency {
globals.sni_consistency = !b;
if b {

2
src/config/toml.rs
View file

@ -15,6 +15,7 @@ pub struct ConfigToml {
pub experimental: Option<Experimental>,
}
#[cfg(feature = "http3")]
#[derive(Deserialize, Debug, Default)]
pub struct Http3Option {
pub alt_svc_max_age: Option<u32>,
@ -26,6 +27,7 @@ pub struct Http3Option {
#[derive(Deserialize, Debug, Default)]
pub struct Experimental {
#[cfg(feature = "http3")]
pub h3: Option<Http3Option>,
pub ignore_sni_consistency: Option<bool>,
}

6
src/constants.rs
View file

@ -9,13 +9,13 @@ pub const MAX_CONCURRENT_STREAMS: u32 = 64;
// #[cfg(feature = "tls")]
pub const CERTS_WATCH_DELAY_SECS: u32 = 30;
// #[cfg(feature = "h3")]
// #[cfg(feature = "http3")]
// pub const H3_RESPONSE_BUF_SIZE: usize = 65_536; // 64KB
// #[cfg(feature = "h3")]
// #[cfg(feature = "http3")]
// pub const H3_REQUEST_BUF_SIZE: usize = 65_536; // 64KB // handled by quinn
#[allow(non_snake_case)]
#[cfg(feature = "h3")]
#[cfg(feature = "http3")]
pub mod H3 {
pub const ALT_SVC_MAX_AGE: u32 = 3600;
pub const REQUEST_MAX_BODY_SIZE: usize = 268_435_456; // 256MB

7
src/globals.rs
View file

@ -24,11 +24,18 @@ pub struct Globals {
// experimentals
pub sni_consistency: bool,
#[cfg(feature = "http3")]
pub http3: bool,
#[cfg(feature = "http3")]
pub h3_alt_svc_max_age: u32,
#[cfg(feature = "http3")]
pub h3_request_max_body_size: usize,
#[cfg(feature = "http3")]
pub h3_max_concurrent_bidistream: quinn::VarInt,
#[cfg(feature = "http3")]
pub h3_max_concurrent_unistream: quinn::VarInt,
#[cfg(feature = "http3")]
pub h3_max_concurrent_connections: u32,
}

7
src/main.rs
View file

@ -77,11 +77,18 @@ fn main() {
},
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(),
};

2
src/msg_handler/handler.rs
View file

@ -200,7 +200,7 @@ where
remove_hop_header(headers);
add_header_entry_overwrite_if_exist(headers, "server", env!("CARGO_PKG_NAME"))?;
#[cfg(feature = "h3")]
#[cfg(feature = "http3")]
{
if self.globals.http3 {
if let Some(port) = self.globals.https_port {

2
src/proxy/mod.rs
View file

@ -1,4 +1,4 @@
#[cfg(feature = "h3")]
#[cfg(feature = "http3")]
mod proxy_h3;
mod proxy_main;
mod proxy_tls;

12
src/proxy/proxy_h3.rs
View file

@ -10,17 +10,7 @@ impl<T> Proxy<T>
where
T: Connect + Clone + Sync + Send + 'static,
{
pub(super) fn connection_serve_h3(&self, conn: quinn::Connecting, tls_server_name: &[u8]) {
let fut = self.clone().handle_connection_h3(conn, tls_server_name.to_vec());
self.globals.runtime_handle.spawn(async move {
// Timeout is based on underlying quic
if let Err(e) = fut.await {
warn!("QUIC or HTTP/3 connection failed: {}", e)
}
});
}
async fn handle_connection_h3(self, conn: quinn::Connecting, tls_server_name: ServerNameLC) -> Result<()> {
pub(super) async fn connection_serve_h3(self, conn: quinn::Connecting, tls_server_name: ServerNameLC) -> Result<()> {
let client_addr = conn.remote_address();
match conn.await {

44
src/proxy/proxy_tls.rs
View file

@ -1,7 +1,5 @@
use super::proxy_main::{LocalExecutor, Proxy};
use crate::{constants::*, error::*, log::*};
#[cfg(feature = "h3")]
use futures::StreamExt;
use futures::{future::FutureExt, select};
use hyper::{client::connect::Connect, server::conn::Http};
use rustls::ServerConfig;
@ -13,6 +11,11 @@ use tokio::{
};
use tokio_rustls::TlsAcceptor;
#[cfg(feature = "http3")]
use futures::StreamExt;
#[cfg(feature = "http3")]
use quinn::{crypto::rustls::HandshakeData, Endpoint, ServerConfig as QuicServerConfig, TransportConfig};
impl<T> Proxy<T>
where
T: Connect + Clone + Sync + Send + 'static,
@ -76,9 +79,9 @@ where
Ok(()) as Result<()>
}
#[cfg(feature = "h3")]
#[cfg(feature = "http3")]
async fn listener_service_h3(&self, mut server_crypto_rx: watch::Receiver<Option<Arc<ServerConfig>>>) -> Result<()> {
let mut transport_config_quic = quinn::TransportConfig::default();
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);
@ -89,10 +92,10 @@ where
.generate_server_crypto_with_cert_resolver()
.await?;
let mut server_config_h3 = quinn::ServerConfig::with_crypto(Arc::new(server_crypto));
let mut server_config_h3 = QuicServerConfig::with_crypto(Arc::new(server_crypto));
server_config_h3.transport = Arc::new(transport_config_quic);
server_config_h3.concurrent_connections(self.globals.h3_max_concurrent_connections);
let (endpoint, mut incoming) = quinn::Endpoint::server(server_config_h3, self.listening_on)?;
let (endpoint, mut incoming) = Endpoint::server(server_config_h3, self.listening_on)?;
info!("Start UDP proxy serving with HTTP/3 request for configured host names");
let mut server_crypto: Option<Arc<ServerConfig>> = None;
@ -103,15 +106,14 @@ where
continue;
}
let mut conn = new_conn.unwrap();
let hsd = if let Ok(h) = conn.handshake_data().await {
h
} else {
continue
let hsd = match conn.handshake_data().await {
Ok(h) => h,
Err(_) => continue
};
let hsd_downcast = if let Ok(d) = hsd.downcast::<quinn::crypto::rustls::HandshakeData>() {
d
} else {
continue;
let hsd_downcast = match hsd.downcast::<HandshakeData>() {
Ok(d) => d,
Err(_) => continue
};
let new_server_name = if let Some(sn) = hsd_downcast.server_name {
sn.as_bytes().to_ascii_lowercase()
@ -125,7 +127,13 @@ where
);
// TODO: server_nameをここで出してどんどん深く投げていくのは効率が悪い。connecting -> connectionsの後でいいのでは？
// TODO: 通常のTLSと同じenumか何かにまとめたい
self.clone().connection_serve_h3(conn, new_server_name.as_ref());
let fut = self.clone().connection_serve_h3(conn, new_server_name);
self.globals.runtime_handle.spawn(async move {
// Timeout is based on underlying quic
if let Err(e) = fut.await {
warn!("QUIC or HTTP/3 connection failed: {}", e)
}
});
}
_ = server_crypto_rx.changed().fuse() => {
if server_crypto_rx.borrow().is_none() {
@ -134,7 +142,7 @@ where
server_crypto = server_crypto_rx.borrow().clone();
if server_crypto.is_some(){
debug!("Reload server crypto");
endpoint.set_server_config(Some(quinn::ServerConfig::with_crypto(server_crypto.clone().unwrap())));
endpoint.set_server_config(Some(QuicServerConfig::with_crypto(server_crypto.clone().unwrap())));
}
}
complete => break
@ -146,7 +154,7 @@ where
pub async fn start_with_tls(self, server: Http<LocalExecutor>) -> Result<()> {
let (tx, rx) = watch::channel::<Option<Arc<ServerConfig>>>(None);
#[cfg(not(feature = "h3"))]
#[cfg(not(feature = "http3"))]
{
select! {
_= self.cert_service(tx).fuse() => {
@ -162,7 +170,7 @@ where
};
Ok(())
}
#[cfg(feature = "h3")]
#[cfg(feature = "http3")]
{
if self.globals.http3 {
select! {