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

Merge pull request #63 from junkurihara/refactor

Browse source 
refactor http handler
This commit is contained in:
Jun Kurihara 2023-07-28 15:22:25 +09:00 • committed by GitHub
commit 7d1bac13bd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 92 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

2
quinn

@ -1 +1 @@
Subproject commit 0ae7c60b15637d7343410ba1e5cc3151e3814557 Subproject commit 532ba7d80405ad083fd05546fa71becbe5eff1a4

5
rpxy-bin/Cargo.toml
View file

@ -20,7 +20,8 @@ rpxy-lib = { path = "../rpxy-lib/", features = ["http3", "sticky-cookie"] }
anyhow = "1.0.72" anyhow = "1.0.72"
rustc-hash = "1.1.0" rustc-hash = "1.1.0"
serde = { version = "1.0.175", default-features = false, features = ["derive"] } serde = { version = "1.0.177", default-features = false, features = ["derive"] }
develop
derive_builder = "0.12.0" derive_builder = "0.12.0"
tokio = { version = "1.29.1", default-features = false, features = [ tokio = { version = "1.29.1", default-features = false, features = [
"net", "net",
@ -43,7 +44,7 @@ tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
[target.'cfg(not(target_env = "msvc"))'.dependencies] [target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemallocator = "0.5.0" tikv-jemallocator = "0.5.4"
[dev-dependencies] [dev-dependencies]

124
rpxy-lib/src/handler/handler_main.rs
View file

@ -19,6 +19,8 @@ use std::{env, net::SocketAddr, sync::Arc};
use tokio::{io::copy_bidirectional, time::timeout}; use tokio::{io::copy_bidirectional, time::timeout};
#[derive(Clone, Builder)] #[derive(Clone, Builder)]
/// HTTP message handler for requests from clients and responses from backend applications,
/// responsible to manipulate and forward messages to upstream backends and downstream clients.
pub struct HttpMessageHandler<T, U> pub struct HttpMessageHandler<T, U>
where where
T: Connect + Clone + Sync + Send + 'static, T: Connect + Clone + Sync + Send + 'static,
@ -33,11 +35,13 @@ where
T: Connect + Clone + Sync + Send + 'static, T: Connect + Clone + Sync + Send + 'static,
U: CryptoSource + Clone, U: CryptoSource + Clone,
{ {
/// Return with an arbitrary status code of error and log message
fn return_with_error_log(&self, status_code: StatusCode, log_data: &mut MessageLog) -> Result<Response<Body>> { fn return_with_error_log(&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)
} }
/// Handle incoming request message from a client
pub async fn handle_request( pub async fn handle_request(
self, self,
mut req: Request<Body>, mut req: Request<Body>,
@ -65,13 +69,15 @@ where
} }
} }
// Find backend application for given server_name, and drop if incoming request is invalid as request. // Find backend application for given server_name, and drop if incoming request is invalid as request.
let backend = if let Some(be) = self.globals.backends.apps.get(&server_name) { let backend = match self.globals.backends.apps.get(&server_name) {
be Some(be) => be,
} else if let Some(default_server_name) = &self.globals.backends.default_server_name_bytes { None => {
debug!("Serving by default app"); let Some(default_server_name) = &self.globals.backends.default_server_name_bytes else {
self.globals.backends.apps.get(default_server_name).unwrap() return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data);
} else { };
return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data); debug!("Serving by default app");
self.globals.backends.apps.get(default_server_name).unwrap()
}
}; };
// Redirect to https if !tls_enabled and redirect_to_https is true // Redirect to https if !tls_enabled and redirect_to_https is true
@ -84,9 +90,8 @@ where
// Find reverse proxy for given path and choose one of upstream host // Find reverse proxy for given path and choose one of upstream host
// Longest prefix match // Longest prefix match
let path = req.uri().path(); let path = req.uri().path();
let upstream_group = match backend.reverse_proxy.get(path) { let Some(upstream_group) = backend.reverse_proxy.get(path) else {
Some(ug) => ug, return self.return_with_error_log(StatusCode::NOT_FOUND, &mut log_data)
None => return self.return_with_error_log(StatusCode::NOT_FOUND, &mut log_data),
}; };
// Upgrade in request header // Upgrade in request header
@ -113,19 +118,17 @@ where
log_data.upstream(req.uri()); log_data.upstream(req.uri());
////// //////
// Forward request to // Forward request to a chosen backend
let mut res_backend = { let mut res_backend = {
match timeout(self.globals.proxy_config.upstream_timeout, self.forwarder.request(req)).await { let Ok(result) = timeout(self.globals.proxy_config.upstream_timeout, self.forwarder.request(req)).await else {
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); };
match result {
Ok(res) => res,
Err(e) => {
error!("Failed to get response from backend: {}", e);
return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data);
} }
Ok(x) => match x {
Ok(res) => res,
Err(e) => {
error!("Failed to get response from backend: {}", e);
return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data);
}
},
} }
}; };
@ -141,62 +144,63 @@ where
if res_backend.status() != StatusCode::SWITCHING_PROTOCOLS { if res_backend.status() != StatusCode::SWITCHING_PROTOCOLS {
// Generate response to client // Generate response to client
if self.generate_response_forwarded(&mut res_backend, backend).is_ok() { if self.generate_response_forwarded(&mut res_backend, backend).is_err() {
log_data.status_code(&res_backend.status()).output();
return Ok(res_backend);
} else {
return self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data); return self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data);
} }
log_data.status_code(&res_backend.status()).output();
return Ok(res_backend);
} }
// 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)) = (upgrade_in_request.as_ref(), upgrade_in_response.as_ref()) { let should_upgrade = if let (Some(u_req), Some(u_res)) = (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 !should_upgrade {
let Some(onupgrade) = res_backend.extensions_mut().remove::<hyper::upgrade::OnUpgrade>() else {
error!("Response does not have an upgrade extension");
return self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data);
};
self.globals.runtime_handle.spawn(async move {
let mut response_upgraded = onupgrade.await.map_err(|e| {
error!("Failed to upgrade response: {}", e);
RpxyError::Hyper(e)
})?;
let mut request_upgraded = request_upgraded.await.map_err(|e| {
error!("Failed to upgrade request: {}", e);
RpxyError::Hyper(e)
})?;
copy_bidirectional(&mut response_upgraded, &mut request_upgraded)
.await
.map_err(|e| {
error!("Coping between upgraded connections failed: {}", e);
RpxyError::Io(e)
})?;
Ok(()) as Result<()>
});
log_data.status_code(&res_backend.status()).output();
Ok(res_backend)
} else {
error!("Request does not have an upgrade extension");
self.return_with_error_log(StatusCode::BAD_REQUEST, &mut log_data)
}
} else {
error!( error!(
"Backend tried to switch to protocol {:?} when {:?} was requested", "Backend tried to switch to protocol {:?} when {:?} was requested",
upgrade_in_response, upgrade_in_request upgrade_in_response, upgrade_in_request
); );
self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data) return self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data);
} }
let Some(request_upgraded) = request_upgraded else {
error!("Request does not have an upgrade extension");
return self.return_with_error_log(StatusCode::BAD_REQUEST, &mut log_data);
};
let Some(onupgrade) = res_backend.extensions_mut().remove::<hyper::upgrade::OnUpgrade>() else {
error!("Response does not have an upgrade extension");
return self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data);
};
self.globals.runtime_handle.spawn(async move {
let mut response_upgraded = onupgrade.await.map_err(|e| {
error!("Failed to upgrade response: {}", e);
RpxyError::Hyper(e)
})?;
let mut request_upgraded = request_upgraded.await.map_err(|e| {
error!("Failed to upgrade request: {}", e);
RpxyError::Hyper(e)
})?;
copy_bidirectional(&mut response_upgraded, &mut request_upgraded)
.await
.map_err(|e| {
error!("Coping between upgraded connections failed: {}", e);
RpxyError::Io(e)
})?;
Ok(()) as Result<()>
});
log_data.status_code(&res_backend.status()).output();
Ok(res_backend)
} }
//////////////////////////////////////////////////// ////////////////////////////////////////////////////
// Functions to generate messages // Functions to generate messages
////////////////////////////////////////////////////
/// Manipulate a response message sent from a backend application to forward downstream to a client.
fn generate_response_forwarded<B>(&self, response: &mut Response<B>, chosen_backend: &Backend<U>) -> Result<()> fn generate_response_forwarded<B>(&self, response: &mut Response<B>, chosen_backend: &Backend<U>) -> Result<()>
where where
B: core::fmt::Debug, B: core::fmt::Debug,
@ -208,7 +212,8 @@ where
#[cfg(feature = "http3")] #[cfg(feature = "http3")]
{ {
// TODO: Workaround for avoid h3 for client authentication // Manipulate ALT_SVC allowing h3 in response message only when mutual TLS is not enabled
// TODO: This is a workaround for avoiding a client authentication in HTTP/3
if self.globals.proxy_config.http3 if self.globals.proxy_config.http3
&& chosen_backend && chosen_backend
.crypto_source .crypto_source
@ -241,6 +246,7 @@ where
} }
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
/// Manipulate a request message sent from a client to forward upstream to a backend application
fn generate_request_forwarded<B>( fn generate_request_forwarded<B>(
&self, &self,
client_addr: &SocketAddr, client_addr: &SocketAddr,

1
rpxy-lib/src/handler/mod.rs
View file

@ -9,6 +9,7 @@ pub use handler_main::{HttpMessageHandler, HttpMessageHandlerBuilder, HttpMessag
#[allow(dead_code)] #[allow(dead_code)]
#[derive(Debug)] #[derive(Debug)]
/// Context object to handle sticky cookies at HTTP message handler
struct HandlerContext { struct HandlerContext {
#[cfg(feature = "sticky-cookie")] #[cfg(feature = "sticky-cookie")]
context_lb: Option<LbContext>, context_lb: Option<LbContext>,

27
rpxy-lib/src/handler/utils_headers.rs
View file

@ -8,7 +8,7 @@ use hyper::{
header::{self, HeaderMap, HeaderName, HeaderValue}, header::{self, HeaderMap, HeaderName, HeaderValue},
Uri, Uri,
}; };
use std::net::SocketAddr; use std::{borrow::Cow, net::SocketAddr};
//////////////////////////////////////////////////// ////////////////////////////////////////////////////
// Functions to manipulate headers // Functions to manipulate headers
@ -83,6 +83,7 @@ pub(super) fn set_sticky_cookie_lb_context(headers: &mut HeaderMap, context_from
Ok(()) Ok(())
} }
/// Apply options to request header, which are specified in the configuration
pub(super) fn apply_upstream_options_to_header( pub(super) fn apply_upstream_options_to_header(
headers: &mut HeaderMap, headers: &mut HeaderMap,
_client_addr: &SocketAddr, _client_addr: &SocketAddr,
@ -113,7 +114,7 @@ pub(super) fn apply_upstream_options_to_header(
Ok(()) Ok(())
} }
// https://datatracker.ietf.org/doc/html/rfc9110 /// Append header entry with comma according to [RFC9110](https://datatracker.ietf.org/doc/html/rfc9110)
pub(super) fn append_header_entry_with_comma(headers: &mut HeaderMap, key: &str, value: &str) -> Result<()> { pub(super) fn append_header_entry_with_comma(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) => {
@ -132,10 +133,11 @@ pub(super) fn append_header_entry_with_comma(headers: &mut HeaderMap, key: &str,
Ok(()) Ok(())
} }
/// Add header entry if not exist
pub(super) fn add_header_entry_if_not_exist( pub(super) fn add_header_entry_if_not_exist(
headers: &mut HeaderMap, headers: &mut HeaderMap,
key: impl Into<std::borrow::Cow<'static, str>>, key: impl Into<Cow<'static, str>>,
value: impl Into<std::borrow::Cow<'static, str>>, value: impl Into<Cow<'static, str>>,
) -> Result<()> { ) -> Result<()> {
match headers.entry(HeaderName::from_bytes(key.into().as_bytes())?) { match headers.entry(HeaderName::from_bytes(key.into().as_bytes())?) {
header::Entry::Vacant(entry) => { header::Entry::Vacant(entry) => {
@ -147,10 +149,11 @@ pub(super) fn add_header_entry_if_not_exist(
Ok(()) Ok(())
} }
/// Overwrite header entry if exist
pub(super) fn add_header_entry_overwrite_if_exist( pub(super) fn add_header_entry_overwrite_if_exist(
headers: &mut HeaderMap, headers: &mut HeaderMap,
key: impl Into<std::borrow::Cow<'static, str>>, key: impl Into<Cow<'static, str>>,
value: impl Into<std::borrow::Cow<'static, str>>, value: impl Into<Cow<'static, str>>,
) -> Result<()> { ) -> Result<()> {
match headers.entry(HeaderName::from_bytes(key.into().as_bytes())?) { match headers.entry(HeaderName::from_bytes(key.into().as_bytes())?) {
header::Entry::Vacant(entry) => { header::Entry::Vacant(entry) => {
@ -164,11 +167,10 @@ pub(super) fn add_header_entry_overwrite_if_exist(
Ok(()) Ok(())
} }
/// Align cookie values in single line
/// Sometimes violates [RFC6265](https://www.rfc-editor.org/rfc/rfc6265#section-5.4) (for http/1.1).
/// This is allowed in RFC7540 (for http/2) as mentioned [here](https://stackoverflow.com/questions/4843556/in-http-specification-what-is-the-string-that-separates-cookies).
pub(super) fn make_cookie_single_line(headers: &mut HeaderMap) -> Result<()> { pub(super) fn make_cookie_single_line(headers: &mut HeaderMap) -> Result<()> {
// Sometimes violates RFC6265 (for http/1.1).
// https://www.rfc-editor.org/rfc/rfc6265#section-5.4
// This is allowed in RFC7540 (for http/2).
// https://stackoverflow.com/questions/4843556/in-http-specification-what-is-the-string-that-separates-cookies
let cookies = headers let cookies = headers
.iter() .iter()
.filter(|(k, _)| **k == hyper::header::COOKIE) .filter(|(k, _)| **k == hyper::header::COOKIE)
@ -182,6 +184,7 @@ pub(super) fn make_cookie_single_line(headers: &mut HeaderMap) -> Result<()> {
Ok(()) Ok(())
} }
/// Add forwarding headers like `x-forwarded-for`.
pub(super) fn add_forwarding_header( pub(super) fn add_forwarding_header(
headers: &mut HeaderMap, headers: &mut HeaderMap,
client_addr: &SocketAddr, client_addr: &SocketAddr,
@ -219,6 +222,7 @@ pub(super) fn add_forwarding_header(
Ok(()) Ok(())
} }
/// Remove connection header
pub(super) fn remove_connection_header(headers: &mut HeaderMap) { pub(super) fn remove_connection_header(headers: &mut HeaderMap) {
if let Some(values) = headers.get(header::CONNECTION) { if let Some(values) = headers.get(header::CONNECTION) {
if let Ok(v) = values.clone().to_str() { if let Ok(v) = values.clone().to_str() {
@ -231,6 +235,7 @@ pub(super) fn remove_connection_header(headers: &mut HeaderMap) {
} }
} }
/// Hop header values which are removed at proxy
const HOP_HEADERS: &[&str] = &[ const HOP_HEADERS: &[&str] = &[
"connection", "connection",
"te", "te",
@ -243,12 +248,14 @@ const HOP_HEADERS: &[&str] = &[
"upgrade", "upgrade",
]; ];
/// Remove hop headers
pub(super) fn remove_hop_header(headers: &mut HeaderMap) { pub(super) fn remove_hop_header(headers: &mut HeaderMap) {
HOP_HEADERS.iter().for_each(|key| { HOP_HEADERS.iter().for_each(|key| {
headers.remove(*key); headers.remove(*key);
}); });
} }
/// Extract upgrade header value if exist
pub(super) fn extract_upgrade(headers: &HeaderMap) -> Option<String> { pub(super) fn extract_upgrade(headers: &HeaderMap) -> Option<String> {
if let Some(c) = headers.get(header::CONNECTION) { if let Some(c) = headers.get(header::CONNECTION) {
if c if c

3
rpxy-lib/src/handler/utils_request.rs
View file

@ -7,6 +7,7 @@ use hyper::{header, Request};
//////////////////////////////////////////////////// ////////////////////////////////////////////////////
// Functions to manipulate request line // Functions to manipulate request line
/// Apply upstream options in request line, specified in the configuration
pub(super) fn apply_upstream_options_to_request_line<B>(req: &mut Request<B>, upstream: &UpstreamGroup) -> Result<()> { pub(super) fn apply_upstream_options_to_request_line<B>(req: &mut Request<B>, upstream: &UpstreamGroup) -> Result<()> {
for opt in upstream.opts.iter() { for opt in upstream.opts.iter() {
match opt { match opt {
@ -19,10 +20,12 @@ pub(super) fn apply_upstream_options_to_request_line<B>(req: &mut Request<B>, up
Ok(()) Ok(())
} }
/// Trait defining parser of hostname
pub trait ParseHost { pub trait ParseHost {
fn parse_host(&self) -> Result<&[u8]>; fn parse_host(&self) -> Result<&[u8]>;
} }
impl<B> ParseHost for Request<B> { impl<B> ParseHost for Request<B> {
/// Extract hostname from either the request HOST header or request line
fn parse_host(&self) -> Result<&[u8]> { fn parse_host(&self) -> Result<&[u8]> {
let headers_host = self.headers().get(header::HOST); let headers_host = self.headers().get(header::HOST);
let uri_host = self.uri().host(); let uri_host = self.uri().host();

2
rpxy-lib/src/handler/utils_synth_response.rs
View file

@ -5,11 +5,13 @@ use hyper::{Body, Request, Response, StatusCode, Uri};
//////////////////////////////////////////////////// ////////////////////////////////////////////////////
// Functions to create response (error or redirect) // Functions to create response (error or redirect)
/// Generate a synthetic response message of a certain error status code
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().status(status_code).body(Body::empty())?; let response = Response::builder().status(status_code).body(Body::empty())?;
Ok(response) Ok(response)
} }
/// Generate synthetic response message of a redirection to https host with 301
pub(super) fn secure_redirection<B>( pub(super) fn secure_redirection<B>(
server_name: &str, server_name: &str,
tls_port: Option<u16>, tls_port: Option<u16>,