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

reformat

Browse source
This commit is contained in:
Jun Kurihara 2022-07-22 22:28:26 +09:00
commit 45895ca7bc
No known key found for this signature in database
GPG key ID: 48ADFD173ED22B03
12 changed files with 36 additions and 151 deletions
Download patch file Download diff file Expand all files Collapse all files

38
src/backend.rs
View file

@ -135,28 +135,18 @@ impl Backend {
if let (Some(c), Some(k)) = (self.tls_cert_path.as_ref(), self.tls_cert_key_path.as_ref()) { if let (Some(c), Some(k)) = (self.tls_cert_path.as_ref(), self.tls_cert_key_path.as_ref()) {
(c, k) (c, k)
} else { } else {
return Err(io::Error::new( return Err(io::Error::new(io::ErrorKind::Other, "Invalid certs and keys paths"));
io::ErrorKind::Other,
"Invalid certs and keys paths",
));
}; };
let certs: Vec<_> = { let certs: Vec<_> = {
let certs_path_str = certs_path.display().to_string(); let certs_path_str = certs_path.display().to_string();
let mut reader = BufReader::new(File::open(certs_path).map_err(|e| { let mut reader = BufReader::new(File::open(certs_path).map_err(|e| {
io::Error::new( io::Error::new(
e.kind(), e.kind(),
format!( format!("Unable to load the certificates [{}]: {}", certs_path_str, e),
"Unable to load the certificates [{}]: {}",
certs_path_str, e
),
) )
})?); })?);
rustls_pemfile::certs(&mut reader).map_err(|_| { rustls_pemfile::certs(&mut reader)
io::Error::new( .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "Unable to parse the certificates"))?
io::ErrorKind::InvalidInput,
"Unable to parse the certificates",
)
})?
} }
.drain(..) .drain(..)
.map(Certificate) .map(Certificate)
@ -169,10 +159,7 @@ impl Backend {
.map_err(|e| { .map_err(|e| {
io::Error::new( io::Error::new(
e.kind(), e.kind(),
format!( format!("Unable to load the certificate keys [{}]: {}", certs_keys_path_str, e),
"Unable to load the certificate keys [{}]: {}",
certs_keys_path_str, e
),
) )
})? })?
.read_to_end(&mut encoded_keys)?; .read_to_end(&mut encoded_keys)?;
@ -217,9 +204,7 @@ impl Backend {
} }
impl Backends { impl Backends {
pub async fn generate_server_crypto_with_cert_resolver( pub async fn generate_server_crypto_with_cert_resolver(&self) -> Result<ServerConfig, anyhow::Error> {
&self,
) -> Result<ServerConfig, anyhow::Error> {
let mut resolver = ResolvesServerCertUsingSni::new(); let mut resolver = ResolvesServerCertUsingSni::new();
let mut cnt = 0; let mut cnt = 0;
@ -234,19 +219,12 @@ impl Backends {
e e
) )
} else { } else {
debug!( debug!("Add certificate for server_name: {}", backend.server_name.as_str());
"Add certificate for server_name: {}",
backend.server_name.as_str()
);
cnt += 1; cnt += 1;
} }
} }
Err(e) => { Err(e) => {
warn!( warn!("Failed to add certificate for {}: {}", backend.server_name.as_str(), e);
"Failed to add certificate for {}: {}",
backend.server_name.as_str(),
e
);
} }
} }
} }

5
src/config/parse.rs
View file

@ -207,10 +207,7 @@ fn get_reverse_proxy(rp_settings: &[ReverseProxyOption]) -> Result<ReverseProxy>
}; };
if rpo.path.is_some() { if rpo.path.is_some() {
upstream.insert( upstream.insert(rpo.path.as_ref().unwrap().as_bytes().to_ascii_lowercase(), elem);
rpo.path.as_ref().unwrap().as_bytes().to_ascii_lowercase(),
elem,
);
} else { } else {
upstream.insert("/".as_bytes().to_ascii_lowercase(), elem); upstream.insert("/".as_bytes().to_ascii_lowercase(), elem);
} }

4
src/log.rs
View file

@ -62,9 +62,7 @@ impl MessageLog {
self self
} }
pub fn xff(&mut self, xff: &Option<&hyper::header::HeaderValue>) -> &mut Self { pub fn xff(&mut self, xff: &Option<&hyper::header::HeaderValue>) -> &mut Self {
self.xff = xff self.xff = xff.map_or_else(|| "", |v| v.to_str().unwrap_or("")).to_string();
.map_or_else(|| "", |v| v.to_str().unwrap_or(""))
.to_string();
self self
} }
pub fn upstream(&mut self, upstream: &str) -> &mut Self { pub fn upstream(&mut self, upstream: &str) -> &mut Self {

42
src/msg_handler/handler.rs
View file

@ -28,11 +28,7 @@ impl<T> HttpMessageHandler<T>
where where
T: Connect + Clone + Sync + Send + 'static, T: Connect + Clone + Sync + Send + 'static,
{ {
fn return_with_error_log( fn return_with_error_log(&self, status_code: StatusCode, log_data: &mut MessageLog) -> Result<Response<Body>> {
&self,
status_code: StatusCode,
log_data: &mut MessageLog,
) -> Result<Response<Body>> {
log_data.status_code(&status_code).output(); log_data.status_code(&status_code).output();
http_error(status_code) http_error(status_code)
} }
@ -77,9 +73,7 @@ where
// Redirect to https if !tls_enabled and redirect_to_https is true // Redirect to https if !tls_enabled and redirect_to_https is true
if !tls_enabled && backend.https_redirection.unwrap_or(false) { if !tls_enabled && backend.https_redirection.unwrap_or(false) {
debug!("Redirect to secure connection: {}", &backend.server_name); debug!("Redirect to secure connection: {}", &backend.server_name);
log_data log_data.status_code(&StatusCode::PERMANENT_REDIRECT).output();
.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.https_port, &req);
} }
@ -124,12 +118,7 @@ where
// Forward request to // Forward request to
let mut res_backend = { let mut res_backend = {
match timeout( match timeout(self.globals.upstream_timeout, self.forwarder.request(req_forwarded)).await {
self.globals.upstream_timeout,
self.forwarder.request(req_forwarded),
)
.await
{
Err(_) => { Err(_) => {
return self.return_with_error_log(StatusCode::GATEWAY_TIMEOUT, &mut log_data); return self.return_with_error_log(StatusCode::GATEWAY_TIMEOUT, &mut log_data);
} }
@ -160,18 +149,13 @@ where
// Handle StatusCode::SWITCHING_PROTOCOLS in response // Handle StatusCode::SWITCHING_PROTOCOLS in response
let upgrade_in_response = extract_upgrade(res_backend.headers()); let upgrade_in_response = extract_upgrade(res_backend.headers());
if if let (Some(u_req), Some(u_res)) = if if let (Some(u_req), Some(u_res)) = (upgrade_in_request.as_ref(), upgrade_in_response.as_ref()) {
(upgrade_in_request.as_ref(), upgrade_in_response.as_ref())
{
u_req.to_ascii_lowercase() == u_res.to_ascii_lowercase() u_req.to_ascii_lowercase() == u_res.to_ascii_lowercase()
} else { } else {
false false
} { } {
if let Some(request_upgraded) = request_upgraded { if let Some(request_upgraded) = request_upgraded {
let onupgrade = if let Some(onupgrade) = res_backend let onupgrade = if let Some(onupgrade) = res_backend.extensions_mut().remove::<hyper::upgrade::OnUpgrade>() {
.extensions_mut()
.remove::<hyper::upgrade::OnUpgrade>()
{
onupgrade onupgrade
} else { } else {
error!("Response does not have an upgrade extension"); error!("Response does not have an upgrade extension");
@ -210,10 +194,7 @@ where
//////////////////////////////////////////////////// ////////////////////////////////////////////////////
// Functions to generate messages // Functions to generate messages
fn generate_response_forwarded<B: core::fmt::Debug>( fn generate_response_forwarded<B: core::fmt::Debug>(&self, response: &mut Response<B>) -> Result<()> {
&self,
response: &mut Response<B>,
) -> Result<()> {
let headers = response.headers_mut(); let headers = response.headers_mut();
remove_connection_header(headers); remove_connection_header(headers);
remove_hop_header(headers); remove_hop_header(headers);
@ -272,19 +253,12 @@ where
// Add te: trailer if te_trailer // Add te: trailer if te_trailer
if contains_te_trailers { if contains_te_trailers {
headers.insert( headers.insert(header::TE, HeaderValue::from_bytes("trailers".as_bytes()).unwrap());
header::TE,
HeaderValue::from_bytes("trailers".as_bytes()).unwrap(),
);
} }
// add "host" header of original server_name if not exist (default) // add "host" header of original server_name if not exist (default)
if req.headers().get(header::HOST).is_none() { if req.headers().get(header::HOST).is_none() {
let org_host = req let org_host = req.uri().host().ok_or_else(|| anyhow!("Invalid request"))?.to_owned();
.uri()
.host()
.ok_or_else(|| anyhow!("Invalid request"))?
.to_owned();
req req
.headers_mut() .headers_mut()
.insert(header::HOST, HeaderValue::from_str(&org_host)?); .insert(header::HOST, HeaderValue::from_str(&org_host)?);

18
src/msg_handler/utils_headers.rs
View file

@ -37,11 +37,7 @@ pub(super) fn apply_upstream_options_to_header(
} }
// https://datatracker.ietf.org/doc/html/rfc9110 // https://datatracker.ietf.org/doc/html/rfc9110
pub(super) fn append_header_entry_with_comma( pub(super) fn append_header_entry_with_comma(headers: &mut HeaderMap, key: &str, value: &str) -> Result<()> {
headers: &mut HeaderMap,
key: &str,
value: &str,
) -> Result<()> {
match headers.entry(HeaderName::from_bytes(key.as_bytes())?) { match headers.entry(HeaderName::from_bytes(key.as_bytes())?) {
header::Entry::Vacant(entry) => { header::Entry::Vacant(entry) => {
entry.insert(value.parse::<HeaderValue>()?); entry.insert(value.parse::<HeaderValue>()?);
@ -99,20 +95,12 @@ pub(super) fn add_forwarding_header(
) -> Result<()> { ) -> Result<()> {
// default process // default process
// optional process defined by upstream_option is applied in fn apply_upstream_options // optional process defined by upstream_option is applied in fn apply_upstream_options
append_header_entry_with_comma( append_header_entry_with_comma(headers, "x-forwarded-for", &client_addr.to_canonical().ip().to_string())?;
headers,
"x-forwarded-for",
&client_addr.to_canonical().ip().to_string(),
)?;
/////////// As Nginx /////////// As Nginx
// If we receive X-Forwarded-Proto, pass it through; otherwise, pass along the // If we receive X-Forwarded-Proto, pass it through; otherwise, pass along the
// scheme used to connect to this server // scheme used to connect to this server
add_header_entry_if_not_exist( add_header_entry_if_not_exist(headers, "x-forwarded-proto", if tls { "https" } else { "http" })?;
headers,
"x-forwarded-proto",
if tls { "https" } else { "http" },
)?;
// If we receive X-Forwarded-Port, pass it through; otherwise, pass along the // If we receive X-Forwarded-Port, pass it through; otherwise, pass along the
// server port the client connected to // server port the client connected to
add_header_entry_if_not_exist(headers, "x-forwarded-port", listen_addr.port().to_string())?; add_header_entry_if_not_exist(headers, "x-forwarded-port", listen_addr.port().to_string())?;

5
src/msg_handler/utils_request.rs
View file

@ -28,10 +28,7 @@ impl<B> ReqLog for &Request<B> {
.scheme_str() .scheme_str()
.map_or_else(|| "".to_string(), |v| format!("{}://", v)); .map_or_else(|| "".to_string(), |v| format!("{}://", v));
let uri_host = self.uri().host().unwrap_or(""); let uri_host = self.uri().host().unwrap_or("");
let uri_pq = self let uri_pq = self.uri().path_and_query().map_or_else(|| "", |v| v.as_str());
.uri()
.path_and_query()
.map_or_else(|| "", |v| v.as_str());
let ua = self let ua = self
.headers() .headers()
.get(header::USER_AGENT) .get(header::USER_AGENT)

28
src/msg_handler/utils_response.rs
View file

@ -5,18 +5,8 @@ use std::fmt::Display;
//////////////////////////////////////////////////// ////////////////////////////////////////////////////
// Functions of utils for request messages // Functions of utils for request messages
pub trait ResLog { pub trait ResLog {
fn log<T1: Display, T2: Display + ToCanonical>( fn log<T1: Display, T2: Display + ToCanonical>(self, server_name: &T1, client_addr: &T2, extra: Option<&str>);
self, fn log_debug<T1: Display, T2: Display + ToCanonical>(self, server_name: &T1, client_addr: &T2, extra: Option<&str>);
server_name: &T1,
client_addr: &T2,
extra: Option<&str>,
);
fn log_debug<T1: Display, T2: Display + ToCanonical>(
self,
server_name: &T1,
client_addr: &T2,
extra: Option<&str>,
);
fn build_message<T1: Display, T2: Display + ToCanonical>( fn build_message<T1: Display, T2: Display + ToCanonical>(
self, self,
server_name: &T1, server_name: &T1,
@ -25,20 +15,10 @@ pub trait ResLog {
) -> String; ) -> String;
} }
impl<B> ResLog for &Response<B> { impl<B> ResLog for &Response<B> {
fn log<T1: Display, T2: Display + ToCanonical>( fn log<T1: Display, T2: Display + ToCanonical>(self, server_name: &T1, client_addr: &T2, extra: Option<&str>) {
self,
server_name: &T1,
client_addr: &T2,
extra: Option<&str>,
) {
info!("{}", &self.build_message(server_name, client_addr, extra)); info!("{}", &self.build_message(server_name, client_addr, extra));
} }
fn log_debug<T1: Display, T2: Display + ToCanonical>( fn log_debug<T1: Display, T2: Display + ToCanonical>(self, server_name: &T1, client_addr: &T2, extra: Option<&str>) {
self,
server_name: &T1,
client_addr: &T2,
extra: Option<&str>,
) {
debug!("{}", &self.build_message(server_name, client_addr, extra)); debug!("{}", &self.build_message(server_name, client_addr, extra));
} }
fn build_message<T1: Display, T2: Display + ToCanonical>( fn build_message<T1: Display, T2: Display + ToCanonical>(

4
src/msg_handler/utils_synth_response.rs
View file

@ -6,9 +6,7 @@ use hyper::{Body, Request, Response, StatusCode, Uri};
// Functions to create response (error or redirect) // Functions to create response (error or redirect)
pub(super) fn http_error(status_code: StatusCode) -> Result<Response<Body>> { pub(super) fn http_error(status_code: StatusCode) -> Result<Response<Body>> {
let response = Response::builder() let response = Response::builder().status(status_code).body(Body::empty())?;
.status(status_code)
.body(Body::empty())?;
Ok(response) Ok(response)
} }

18
src/proxy/proxy_h3.rs
View file

@ -11,9 +11,7 @@ where
T: Connect + Clone + Sync + Send + 'static, T: Connect + Clone + Sync + Send + 'static,
{ {
pub(super) fn connection_serve_h3(&self, conn: quinn::Connecting, tls_server_name: &[u8]) { pub(super) fn connection_serve_h3(&self, conn: quinn::Connecting, tls_server_name: &[u8]) {
let fut = self let fut = self.clone().handle_connection_h3(conn, tls_server_name.to_vec());
.clone()
.handle_connection_h3(conn, tls_server_name.to_vec());
self.globals.runtime_handle.spawn(async move { self.globals.runtime_handle.spawn(async move {
// Timeout is based on underlying quic // Timeout is based on underlying quic
if let Err(e) = fut.await { if let Err(e) = fut.await {
@ -22,18 +20,12 @@ where
}); });
} }
async fn handle_connection_h3( async fn handle_connection_h3(self, conn: quinn::Connecting, tls_server_name: ServerNameLC) -> Result<()> {
self,
conn: quinn::Connecting,
tls_server_name: ServerNameLC,
) -> Result<()> {
let client_addr = conn.remote_address(); let client_addr = conn.remote_address();
match conn.await { match conn.await {
Ok(new_conn) => { Ok(new_conn) => {
let mut h3_conn = let mut h3_conn = h3::server::Connection::<_, bytes::Bytes>::new(h3_quinn::Connection::new(new_conn)).await?;
h3::server::Connection::<_, bytes::Bytes>::new(h3_quinn::Connection::new(new_conn))
.await?;
info!( info!(
"QUIC/HTTP3 connection established from {:?} {:?}", "QUIC/HTTP3 connection established from {:?} {:?}",
client_addr, tls_server_name client_addr, tls_server_name
@ -113,9 +105,7 @@ where
return Err(anyhow!("Exceeds max request body size for HTTP/3")); return Err(anyhow!("Exceeds max request body size for HTTP/3"));
} }
// create stream body to save memory, shallow copy (increment of ref-count) to Bytes using copy_to_bytes // create stream body to save memory, shallow copy (increment of ref-count) to Bytes using copy_to_bytes
sender sender.send_data(body.copy_to_bytes(body.remaining())).await?;
.send_data(body.copy_to_bytes(body.remaining()))
.await?;
} }
// trailers: use inner for work around. (directly get trailer) // trailers: use inner for work around. (directly get trailer)

8
src/proxy/proxy_main.rs
View file

@ -1,7 +1,5 @@
// use super::proxy_handler::handle_request; // use super::proxy_handler::handle_request;
use crate::{ use crate::{backend::ServerNameLC, error::*, globals::Globals, log::*, msg_handler::HttpMessageHandler};
backend::ServerNameLC, error::*, globals::Globals, log::*, msg_handler::HttpMessageHandler,
};
use hyper::{client::connect::Connect, server::conn::Http, service::service_fn, Body, Request}; use hyper::{client::connect::Connect, server::conn::Http, service::service_fn, Body, Request};
use std::{net::SocketAddr, sync::Arc}; use std::{net::SocketAddr, sync::Arc};
use tokio::{ use tokio::{
@ -95,9 +93,7 @@ where
let tcp_listener = TcpListener::bind(&self.listening_on).await?; let tcp_listener = TcpListener::bind(&self.listening_on).await?;
info!("Start TCP proxy serving with HTTP request for configured host names"); info!("Start TCP proxy serving with HTTP request for configured host names");
while let Ok((stream, _client_addr)) = tcp_listener.accept().await { while let Ok((stream, _client_addr)) = tcp_listener.accept().await {
self self.clone().client_serve(stream, server.clone(), _client_addr, None);
.clone()
.client_serve(stream, server.clone(), _client_addr, None);
} }
Ok(()) as Result<()> Ok(()) as Result<()>
}; };

12
src/proxy/proxy_tls.rs
View file

@ -20,12 +20,7 @@ where
async fn cert_service(&self, server_crypto_tx: watch::Sender<Option<Arc<ServerConfig>>>) { async fn cert_service(&self, server_crypto_tx: watch::Sender<Option<Arc<ServerConfig>>>) {
info!("Start cert watch service"); info!("Start cert watch service");
loop { loop {
if let Ok(server_crypto) = self if let Ok(server_crypto) = self.globals.backends.generate_server_crypto_with_cert_resolver().await {
.globals
.backends
.generate_server_crypto_with_cert_resolver()
.await
{
if let Err(_e) = server_crypto_tx.send(Some(Arc::new(server_crypto))) { if let Err(_e) = server_crypto_tx.send(Some(Arc::new(server_crypto))) {
error!("Failed to populate server crypto"); error!("Failed to populate server crypto");
break; break;
@ -82,10 +77,7 @@ where
} }
#[cfg(feature = "h3")] #[cfg(feature = "h3")]
async fn listener_service_h3( async fn listener_service_h3(&self, mut server_crypto_rx: watch::Receiver<Option<Arc<ServerConfig>>>) -> Result<()> {
&self,
mut server_crypto_rx: watch::Receiver<Option<Arc<ServerConfig>>>,
) -> Result<()> {
let mut transport_config_quic = quinn::TransportConfig::default(); let mut transport_config_quic = quinn::TransportConfig::default();
transport_config_quic transport_config_quic
.max_concurrent_bidi_streams(self.globals.h3_max_concurrent_bidistream) .max_concurrent_bidi_streams(self.globals.h3_max_concurrent_bidistream)

5
src/utils/socket_addr.rs
View file

@ -51,10 +51,7 @@ mod tests {
} }
#[test] #[test]
fn ipv4_mapped_to_ipv6_to_canonical() { fn ipv4_mapped_to_ipv6_to_canonical() {
let socket = SocketAddr::new( let socket = SocketAddr::new(IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0xffff, 0xc00a, 0x2ff)), 8080);
IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0xffff, 0xc00a, 0x2ff)),
8080,
);
assert_eq!( assert_eq!(
socket.to_canonical(), socket.to_canonical(),
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 10, 2, 255)), 8080) SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 10, 2, 255)), 8080)