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

change src structure

Browse source
This commit is contained in:
Jun Kurihara 2022-07-26 15:30:50 +09:00
commit 09c6c4c102
9 changed files with 5 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

300
src/handler/handler_main.rs Normal file
View file

@ -0,0 +1,300 @@
// Highly motivated by https://github.com/felipenoris/hyper-reverse-proxy
use super::{utils_headers::*, utils_request::*, utils_response::ResLog, utils_synth_response::*};
use crate::{
backend::{ServerNameLC, Upstream},
error::*,
globals::Globals,
log::*,
};
use hyper::{
client::connect::Connect,
header::{self, HeaderValue},
http::uri::Scheme,
Body, Client, Request, Response, StatusCode, Uri, Version,
};
use std::{env, net::SocketAddr, sync::Arc};
use tokio::{io::copy_bidirectional, time::timeout};
#[derive(Clone)]
pub struct HttpMessageHandler<T>
where
T: Connect + Clone + Sync + Send + 'static,
{
pub forwarder: Arc<Client<T>>,
pub globals: Arc<Globals>,
}
impl<T> HttpMessageHandler<T>
where
T: Connect + Clone + Sync + Send + 'static,
{
fn return_with_error_log(&self, status_code: StatusCode, log_data: &mut MessageLog) -> Result<Response<Body>> {
log_data.status_code(&status_code).output();
http_error(status_code)
}
pub async fn handle_request(
self,
mut req: Request<Body>,
client_addr: SocketAddr, // アクセス制御用
listen_addr: SocketAddr,
tls_enabled: bool,
tls_server_name: Option<ServerNameLC>,
) -> Result<Response<Body>> {
req.log_debug(&client_addr, Some("(from Client)"));
////////
let mut log_data = MessageLog::from(&req);
log_data.client_addr(&client_addr);
//////
// Here we start to handle with server_name
let server_name_bytes = if let Ok(v) = req.parse_host() {
v.to_ascii_lowercase()
} else {
return self.return_with_error_log(StatusCode::BAD_REQUEST, &mut log_data);
};
// check consistency of between TLS SNI and HOST/Request URI Line.
#[allow(clippy::collapsible_if)]
if tls_enabled && self.globals.sni_consistency {
if !server_name_bytes.eq_ignore_ascii_case(&tls_server_name.unwrap_or_default()) {
return self.return_with_error_log(StatusCode::MISDIRECTED_REQUEST, &mut log_data);
}
}
// 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_bytes) {
be
} else if let Some(default_server_name) = &self.globals.backends.default_server_name {
debug!("Serving by default app");
self.globals.backends.apps.get(default_server_name).unwrap()
} else {
return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data);
};
// Redirect to https if !tls_enabled and redirect_to_https is true
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);
}
// Find reverse proxy for given path and choose one of upstream host
// Longest prefix match
let path = req.uri().path();
let upstream = if let Some(upstream) = backend.reverse_proxy.get(path) {
upstream
} else {
return self.return_with_error_log(StatusCode::NOT_FOUND, &mut log_data);
};
let upstream_scheme_host = if let Some(u) = upstream.get() {
u
} else {
return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data);
};
// Upgrade in request header
let upgrade_in_request = extract_upgrade(req.headers());
let request_upgraded = req.extensions_mut().remove::<hyper::upgrade::OnUpgrade>();
// Build request from destination information
if let Err(e) = self.generate_request_forwarded(
&client_addr,
&listen_addr,
&mut req,
upstream_scheme_host,
&upgrade_in_request,
upstream,
tls_enabled,
) {
error!("Failed to generate destination uri for reverse proxy: {}", e);
return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data);
};
// debug!("Request to be forwarded: {:?}", req_forwarded);
req.log_debug(&client_addr, Some("(to Backend)"));
log_data.xff(&req.headers().get("x-forwarded-for"));
log_data.upstream(&upstream_scheme_host.to_string());
//////
// Forward request to
let mut res_backend = {
match timeout(self.globals.upstream_timeout, self.forwarder.request(req)).await {
Err(_) => {
return self.return_with_error_log(StatusCode::GATEWAY_TIMEOUT, &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);
}
},
}
};
res_backend.log_debug(&backend.server_name, &client_addr, Some("(from Backend)"));
// let response_log = res_backend.status().to_string();
if res_backend.status() != StatusCode::SWITCHING_PROTOCOLS {
// Generate response to client
if self.generate_response_forwarded(&mut res_backend).is_ok() {
// info!("{} => {}", request_log, response_log);
res_backend.log_debug(&backend.server_name, &client_addr, Some("(to Client)"));
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);
}
}
// Handle StatusCode::SWITCHING_PROTOCOLS in response
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()) {
u_req.to_ascii_lowercase() == u_res.to_ascii_lowercase()
} else {
false
} {
if let Some(request_upgraded) = request_upgraded {
let onupgrade = if let Some(onupgrade) = res_backend.extensions_mut().remove::<hyper::upgrade::OnUpgrade>() {
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);
anyhow!("Failed to upgrade response: {}", e)
})?;
let mut request_upgraded = request_upgraded.await.map_err(|e| {
error!("Failed to upgrade request: {}", e);
anyhow!("Failed to upgrade request: {}", e)
})?;
copy_bidirectional(&mut response_upgraded, &mut request_upgraded)
.await
.map_err(|e| anyhow!("Coping between upgraded connections failed: {}", 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!(
"Backend tried to switch to protocol {:?} when {:?} was requested",
upgrade_in_response, upgrade_in_request
);
self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data)
}
}
////////////////////////////////////////////////////
// Functions to generate messages
fn generate_response_forwarded<B: core::fmt::Debug>(&self, response: &mut Response<B>) -> Result<()> {
let headers = response.headers_mut();
remove_connection_header(headers);
remove_hop_header(headers);
add_header_entry_overwrite_if_exist(headers, "server", env!("CARGO_PKG_NAME"))?;
#[cfg(feature = "http3")]
{
if self.globals.http3 {
if let Some(port) = self.globals.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
),
)?;
}
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn generate_request_forwarded<B: core::fmt::Debug>(
&self,
client_addr: &SocketAddr,
listen_addr: &SocketAddr,
req: &mut Request<B>,
upstream_scheme_host: &Uri,
upgrade: &Option<String>,
upstream: &Upstream,
tls_enabled: bool,
) -> Result<()> {
debug!("Generate request to be forwarded");
// Add te: trailer if contained in original request
let contains_te_trailers = {
if let Some(te) = req.headers().get(header::TE) {
te.as_bytes()
.split(|v| v == &b',' || v == &b' ')
.any(|x| x == "trailers".as_bytes())
} else {
false
}
};
let uri = req.uri().to_string();
let headers = req.headers_mut();
// delete headers specified in header.connection
remove_connection_header(headers);
// delete hop headers including header.connection
remove_hop_header(headers);
// X-Forwarded-For
add_forwarding_header(headers, client_addr, listen_addr, tls_enabled, &uri)?;
// Add te: trailer if te_trailer
if contains_te_trailers {
headers.insert(header::TE, HeaderValue::from_bytes("trailers".as_bytes()).unwrap());
}
// add "host" header of original server_name if not exist (default)
if req.headers().get(header::HOST).is_none() {
let org_host = req.uri().host().ok_or_else(|| anyhow!("Invalid request"))?.to_owned();
req
.headers_mut()
.insert(header::HOST, HeaderValue::from_str(&org_host)?);
};
// apply upstream-specific headers given in upstream_option
let headers = req.headers_mut();
apply_upstream_options_to_header(headers, client_addr, upstream_scheme_host, upstream)?;
// update uri in request
ensure!(upstream_scheme_host.authority().is_some() && upstream_scheme_host.scheme().is_some());
let new_uri = Uri::builder()
.scheme(upstream_scheme_host.scheme().unwrap().as_str())
.authority(upstream_scheme_host.authority().unwrap().as_str());
let pq = req.uri().path_and_query();
*req.uri_mut() = match pq {
None => new_uri,
Some(x) => new_uri.path_and_query(x.to_owned()),
}
.build()?;
// upgrade
if let Some(v) = upgrade {
req.headers_mut().insert(header::UPGRADE, v.parse()?);
req
.headers_mut()
.insert(header::CONNECTION, HeaderValue::from_str("upgrade")?);
}
// Change version to http/1.1 when destination scheme is http
if req.version() != Version::HTTP_11 && upstream_scheme_host.scheme() == Some(&Scheme::HTTP) {
*req.version_mut() = Version::HTTP_11;
} else if req.version() == Version::HTTP_3 {
debug!("HTTP/3 is currently unsupported for request to upstream. Use HTTP/2.");
*req.version_mut() = Version::HTTP_2;
}
Ok(())
}
}

7
src/handler/mod.rs Normal file
View file

@ -0,0 +1,7 @@
mod handler_main;
mod utils_headers;
mod utils_request;
mod utils_response;
mod utils_synth_response;
pub use handler_main::HttpMessageHandler;

171
src/handler/utils_headers.rs Normal file
View file

@ -0,0 +1,171 @@
use crate::{backend::Upstream, backend_opt::UpstreamOption, error::*, log::*, utils::*};
use bytes::BufMut;
use hyper::{
header::{self, HeaderMap, HeaderName, HeaderValue},
Uri,
};
use std::net::SocketAddr;
////////////////////////////////////////////////////
// Functions to manipulate headers
pub(super) fn apply_upstream_options_to_header(
headers: &mut HeaderMap,
_client_addr: &SocketAddr,
upstream_scheme_host: &Uri,
upstream: &Upstream,
) -> Result<()> {
for opt in upstream.opts.iter() {
match opt {
UpstreamOption::OverrideHost => {
// overwrite HOST value with upstream hostname (like 192.168.xx.x seen from rpxy)
let upstream_host = upstream_scheme_host.host().ok_or_else(|| anyhow!("none"))?;
headers
.insert(header::HOST, HeaderValue::from_str(upstream_host)?)
.ok_or_else(|| anyhow!("none"))?;
}
UpstreamOption::UpgradeInsecureRequests => {
// add upgrade-insecure-requests in request header if not exist
headers
.entry(header::UPGRADE_INSECURE_REQUESTS)
.or_insert(HeaderValue::from_bytes(&[b'1']).unwrap());
}
}
}
Ok(())
}
// https://datatracker.ietf.org/doc/html/rfc9110
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())?) {
header::Entry::Vacant(entry) => {
entry.insert(value.parse::<HeaderValue>()?);
}
header::Entry::Occupied(mut entry) => {
// entry.append(value.parse::<HeaderValue>()?);
let mut new_value = Vec::<u8>::with_capacity(entry.get().as_bytes().len() + 2 + value.len());
new_value.put_slice(entry.get().as_bytes());
new_value.put_slice(&[b',', b' ']);
new_value.put_slice(value.as_bytes());
entry.insert(HeaderValue::from_bytes(&new_value)?);
}
}
Ok(())
}
pub(super) fn add_header_entry_if_not_exist(
headers: &mut HeaderMap,
key: impl Into<std::borrow::Cow<'static, str>>,
value: impl Into<std::borrow::Cow<'static, str>>,
) -> Result<()> {
match headers.entry(HeaderName::from_bytes(key.into().as_bytes())?) {
header::Entry::Vacant(entry) => {
entry.insert(value.into().parse::<HeaderValue>()?);
}
header::Entry::Occupied(_) => (),
};
Ok(())
}
pub(super) fn add_header_entry_overwrite_if_exist(
headers: &mut HeaderMap,
key: impl Into<std::borrow::Cow<'static, str>>,
value: impl Into<std::borrow::Cow<'static, str>>,
) -> Result<()> {
match headers.entry(HeaderName::from_bytes(key.into().as_bytes())?) {
header::Entry::Vacant(entry) => {
entry.insert(value.into().parse::<HeaderValue>()?);
}
header::Entry::Occupied(mut entry) => {
entry.insert(HeaderValue::from_bytes(value.into().as_bytes())?);
}
}
Ok(())
}
pub(super) fn add_forwarding_header(
headers: &mut HeaderMap,
client_addr: &SocketAddr,
listen_addr: &SocketAddr,
tls: bool,
uri_str: &str,
) -> Result<()> {
// default process
// optional process defined by upstream_option is applied in fn apply_upstream_options
let canonical_client_addr = client_addr.to_canonical().ip().to_string();
append_header_entry_with_comma(headers, "x-forwarded-for", &canonical_client_addr)?;
/////////// As Nginx
// If we receive X-Forwarded-Proto, pass it through; otherwise, pass along the
// scheme used to connect to this server
add_header_entry_if_not_exist(headers, "x-forwarded-proto", if tls { "https" } else { "http" })?;
// If we receive X-Forwarded-Port, pass it through; otherwise, pass along the
// server port the client connected to
add_header_entry_if_not_exist(headers, "x-forwarded-port", listen_addr.port().to_string())?;
/////////// As Nginx-Proxy
// x-real-ip
add_header_entry_overwrite_if_exist(headers, "x-real-ip", canonical_client_addr)?;
// x-forwarded-ssl
add_header_entry_overwrite_if_exist(headers, "x-forwarded-ssl", if tls { "on" } else { "off" })?;
// x-original-uri
add_header_entry_overwrite_if_exist(headers, "x-original-uri", uri_str.to_string())?;
// proxy
add_header_entry_overwrite_if_exist(headers, "proxy", "")?;
Ok(())
}
pub(super) fn remove_connection_header(headers: &mut HeaderMap) {
if let Some(values) = headers.get(header::CONNECTION) {
if let Ok(v) = values.clone().to_str() {
for m in v.split(',') {
if !m.is_empty() {
headers.remove(m.trim());
}
}
}
}
}
const HOP_HEADERS: &[&str] = &[
"connection",
"te",
"trailer",
"keep-alive",
"proxy-connection",
"proxy-authenticate",
"proxy-authorization",
"transfer-encoding",
"upgrade",
];
pub(super) fn remove_hop_header(headers: &mut HeaderMap) {
HOP_HEADERS.iter().for_each(|key| {
headers.remove(*key);
});
}
pub(super) fn extract_upgrade(headers: &HeaderMap) -> Option<String> {
if let Some(c) = headers.get(header::CONNECTION) {
if c
.to_str()
.unwrap_or("")
.split(',')
.into_iter()
.any(|w| w.trim().to_ascii_lowercase() == header::UPGRADE.as_str().to_ascii_lowercase())
{
if let Some(u) = headers.get(header::UPGRADE) {
if let Ok(m) = u.to_str() {
debug!("Upgrade in request header: {}", m);
return Some(m.to_owned());
}
}
}
}
None
}

94
src/handler/utils_request.rs Normal file
View file

@ -0,0 +1,94 @@
use crate::{error::*, log::*, utils::*};
use hyper::{header, Request};
use std::fmt::Display;
////////////////////////////////////////////////////
// Functions of utils for request messages
pub trait ReqLog {
fn log<T: Display + ToCanonical>(self, src: &T, extra: Option<&str>);
fn log_debug<T: Display + ToCanonical>(self, src: &T, extra: Option<&str>);
fn build_message<T: Display + ToCanonical>(self, src: &T, extra: Option<&str>) -> String;
}
impl<B> ReqLog for &Request<B> {
fn log<T: Display + ToCanonical>(self, src: &T, extra: Option<&str>) {
info!("{}", &self.build_message(src, extra));
}
fn log_debug<T: Display + ToCanonical>(self, src: &T, extra: Option<&str>) {
debug!("{}", &self.build_message(src, extra));
}
fn build_message<T: Display + ToCanonical>(self, src: &T, extra: Option<&str>) -> String {
let canonical_src = src.to_canonical();
let host = self
.headers()
.get(header::HOST)
.map_or_else(|| "", |v| v.to_str().unwrap_or(""));
let uri_scheme = self
.uri()
.scheme_str()
.map_or_else(|| "".to_string(), |v| format!("{}://", v));
let uri_host = self.uri().host().unwrap_or("");
let uri_pq = self.uri().path_and_query().map_or_else(|| "", |v| v.as_str());
let ua = self
.headers()
.get(header::USER_AGENT)
.map_or_else(|| "", |v| v.to_str().unwrap_or(""));
let xff = self
.headers()
.get("x-forwarded-for")
.map_or_else(|| "", |v| v.to_str().unwrap_or(""));
format!(
"{} <- {} -- {} {} {:?} -- ({}{}) \"{}\" \"{}\" {}",
host,
canonical_src,
self.method(),
uri_pq,
self.version(),
uri_scheme,
uri_host,
ua,
xff,
extra.unwrap_or("")
)
}
}
pub trait ParseHost {
fn parse_host(&self) -> Result<&[u8]>;
}
impl<B> ParseHost for Request<B> {
fn parse_host(&self) -> Result<&[u8]> {
let headers_host = self.headers().get(header::HOST);
let uri_host = self.uri().host();
// let uri_port = self.uri().port_u16();
ensure!(
!(headers_host.is_none() && uri_host.is_none()),
"No host in request header"
);
// prioritize server_name in uri
uri_host.map_or_else(
|| {
let m = headers_host.unwrap().as_bytes();
if m.starts_with(&[b'[']) {
// v6 address with bracket case. if port is specified, always it is in this case.
let mut iter = m.split(|ptr| ptr == &b'[' || ptr == &b']');
iter.next().ok_or_else(|| anyhow!("Invalid Host"))?; // first item is always blank
iter.next().ok_or_else(|| anyhow!("Invalid Host"))
} else if m.len() - m.split(|v| v == &b':').fold(0, |acc, s| acc + s.len()) >= 2 {
// v6 address case, if 2 or more ':' is contained
Ok(m)
} else {
// v4 address or hostname
m.split(|colon| colon == &b':')
.into_iter()
.next()
.ok_or_else(|| anyhow!("Invalid Host"))
}
},
|v| Ok(v.as_bytes()),
)
}
}

41
src/handler/utils_response.rs Normal file
View file

@ -0,0 +1,41 @@
use crate::{log::*, utils::*};
use hyper::Response;
use std::fmt::Display;
////////////////////////////////////////////////////
// Functions of utils for request messages
pub trait ResLog {
fn log<T1: Display, T2: Display + ToCanonical>(self, 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>(
self,
server_name: &T1,
client_addr: &T2,
extra: Option<&str>,
) -> String;
}
impl<B> ResLog for &Response<B> {
fn log<T1: Display, T2: Display + ToCanonical>(self, server_name: &T1, client_addr: &T2, extra: Option<&str>) {
info!("{}", &self.build_message(server_name, client_addr, extra));
}
fn log_debug<T1: Display, T2: Display + ToCanonical>(self, server_name: &T1, client_addr: &T2, extra: Option<&str>) {
debug!("{}", &self.build_message(server_name, client_addr, extra));
}
fn build_message<T1: Display, T2: Display + ToCanonical>(
self,
server_name: &T1,
client_addr: &T2,
extra: Option<&str>,
) -> String {
let canonical_client_addr = client_addr.to_canonical();
format!(
"{} <- {} -- {} {:?} {}",
canonical_client_addr,
server_name,
self.status(),
self.version(),
// self.headers(),
extra.map_or_else(|| "", |v| v)
)
}
}

33
src/handler/utils_synth_response.rs Normal file
View file

@ -0,0 +1,33 @@
// Highly motivated by https://github.com/felipenoris/hyper-reverse-proxy
use crate::error::*;
use hyper::{Body, Request, Response, StatusCode, Uri};
////////////////////////////////////////////////////
// Functions to create response (error or redirect)
pub(super) fn http_error(status_code: StatusCode) -> Result<Response<Body>> {
let response = Response::builder().status(status_code).body(Body::empty())?;
Ok(response)
}
pub(super) fn secure_redirection<B>(
server_name: &str,
tls_port: Option<u16>,
req: &Request<B>,
) -> Result<Response<Body>> {
let pq = match req.uri().path_and_query() {
Some(x) => x.as_str(),
_ => "",
};
let new_uri = Uri::builder().scheme("https").path_and_query(pq);
let dest_uri = match tls_port {
Some(443) | None => new_uri.authority(server_name),
Some(p) => new_uri.authority(format!("{}:{}", server_name, p)),
}
.build()?;
let response = Response::builder()
.status(StatusCode::MOVED_PERMANENTLY)
.header("Location", dest_uri.to_string())
.body(Body::empty())?;
Ok(response)
}