255 lines
9.1 KiB
Rust
255 lines
9.1 KiB
Rust
use super::{
|
|
http_log::HttpMessageLog,
|
|
http_result::{HttpError, HttpResult},
|
|
synthetic_response::{secure_redirection_response, synthetic_error_response},
|
|
utils_headers::*,
|
|
utils_request::InspectParseHost,
|
|
};
|
|
use crate::{
|
|
backend::{BackendAppManager, LoadBalanceContext},
|
|
crypto::CryptoSource,
|
|
error::*,
|
|
globals::Globals,
|
|
hyper_ext::body::{BoxBody, IncomingLike, IncomingOr},
|
|
log::*,
|
|
name_exp::ServerName,
|
|
};
|
|
use derive_builder::Builder;
|
|
use http::{Request, Response, StatusCode};
|
|
use std::{net::SocketAddr, sync::Arc};
|
|
|
|
#[allow(dead_code)]
|
|
#[derive(Debug)]
|
|
/// Context object to handle sticky cookies at HTTP message handler
|
|
pub(super) struct HandlerContext {
|
|
#[cfg(feature = "sticky-cookie")]
|
|
pub(super) context_lb: Option<LoadBalanceContext>,
|
|
#[cfg(not(feature = "sticky-cookie"))]
|
|
pub(super) context_lb: Option<()>,
|
|
}
|
|
|
|
#[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<U>
|
|
where
|
|
// T: Connect + Clone + Sync + Send + 'static,
|
|
U: CryptoSource + Clone,
|
|
{
|
|
// forwarder: Arc<Forwarder<T>>,
|
|
globals: Arc<Globals>,
|
|
app_manager: Arc<BackendAppManager<U>>,
|
|
}
|
|
|
|
impl<U> HttpMessageHandler<U>
|
|
where
|
|
// T: Connect + Clone + Sync + Send + 'static,
|
|
U: CryptoSource + Clone,
|
|
{
|
|
/// Handle incoming request message from a client.
|
|
/// Responsible to passthrough responses from backend applications or generate synthetic error responses.
|
|
pub async fn handle_request(
|
|
&self,
|
|
mut req: Request<IncomingOr<IncomingLike>>,
|
|
client_addr: SocketAddr, // For access control
|
|
listen_addr: SocketAddr,
|
|
tls_enabled: bool,
|
|
tls_server_name: Option<ServerName>,
|
|
) -> RpxyResult<Response<IncomingOr<BoxBody>>> {
|
|
let mut log_data = HttpMessageLog::from(&req);
|
|
|
|
let http_result = self
|
|
.handle_request_inner(
|
|
&mut log_data,
|
|
req,
|
|
client_addr,
|
|
listen_addr,
|
|
tls_enabled,
|
|
tls_server_name,
|
|
)
|
|
.await;
|
|
|
|
// passthrough or synthetic response
|
|
match http_result {
|
|
Ok(v) => {
|
|
log_data.status_code(&v.status()).output();
|
|
Ok(v)
|
|
}
|
|
Err(e) => {
|
|
debug!("{e}");
|
|
let code = StatusCode::from(e);
|
|
log_data.status_code(&code).output();
|
|
synthetic_error_response(code)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Handle inner with no synthetic error response.
|
|
/// Synthetic response is generated by caller.
|
|
async fn handle_request_inner(
|
|
&self,
|
|
log_data: &mut HttpMessageLog,
|
|
mut req: Request<IncomingOr<IncomingLike>>,
|
|
client_addr: SocketAddr, // For access control
|
|
listen_addr: SocketAddr,
|
|
tls_enabled: bool,
|
|
tls_server_name: Option<ServerName>,
|
|
) -> HttpResult<Response<IncomingOr<BoxBody>>> {
|
|
// preparing log data
|
|
let mut log_data = HttpMessageLog::from(&req);
|
|
log_data.client_addr(&client_addr);
|
|
|
|
// Here we start to inspect and parse with server_name
|
|
let server_name = req
|
|
.inspect_parse_host()
|
|
.map(|v| ServerName::from(v.as_slice()))
|
|
.map_err(|_e| HttpError::InvalidHostInRequestHeader)?;
|
|
|
|
// check consistency of between TLS SNI and HOST/Request URI Line.
|
|
#[allow(clippy::collapsible_if)]
|
|
if tls_enabled && self.globals.proxy_config.sni_consistency {
|
|
if server_name != tls_server_name.unwrap_or_default() {
|
|
return Err(HttpError::SniHostInconsistency);
|
|
}
|
|
}
|
|
// Find backend application for given server_name, and drop if incoming request is invalid as request.
|
|
let backend_app = match self.app_manager.apps.get(&server_name) {
|
|
Some(backend_app) => backend_app,
|
|
None => {
|
|
let Some(default_server_name) = &self.app_manager.default_server_name else {
|
|
return Err(HttpError::NoMatchingBackendApp);
|
|
};
|
|
debug!("Serving by default app");
|
|
self.app_manager.apps.get(default_server_name).unwrap()
|
|
}
|
|
};
|
|
|
|
// Redirect to https if !tls_enabled and redirect_to_https is true
|
|
if !tls_enabled && backend_app.https_redirection.unwrap_or(false) {
|
|
debug!(
|
|
"Redirect to secure connection: {}",
|
|
<&ServerName as TryInto<String>>::try_into(&backend_app.server_name).unwrap_or_default()
|
|
);
|
|
return secure_redirection_response(&backend_app.server_name, self.globals.proxy_config.https_port, &req);
|
|
}
|
|
|
|
// Find reverse proxy for given path and choose one of upstream host
|
|
// Longest prefix match
|
|
let path = req.uri().path();
|
|
let Some(upstream_candidates) = backend_app.path_manager.get(path) else {
|
|
return Err(HttpError::NoUpstreamCandidates);
|
|
};
|
|
|
|
// 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
|
|
let _context = match self.generate_request_forwarded(
|
|
&client_addr,
|
|
&listen_addr,
|
|
&mut req,
|
|
&upgrade_in_request,
|
|
upstream_candidates,
|
|
tls_enabled,
|
|
) {
|
|
Err(e) => {
|
|
error!("Failed to generate destination uri for backend application: {}", e);
|
|
return Err(HttpError::FailedToGenerateUpstreamRequest(e.to_string()));
|
|
}
|
|
Ok(v) => v,
|
|
};
|
|
debug!(
|
|
"Request to be forwarded: uri {}, version {:?}, headers {:?}",
|
|
req.uri(),
|
|
req.version(),
|
|
req.headers()
|
|
);
|
|
log_data.xff(&req.headers().get("x-forwarded-for"));
|
|
log_data.upstream(req.uri());
|
|
//////
|
|
|
|
//////////////
|
|
// // TODO: remove later
|
|
let body = crate::hyper_ext::body::full(hyper::body::Bytes::from("not yet implemented"));
|
|
let mut res_backend = super::synthetic_response::synthetic_response(Response::builder().body(body).unwrap());
|
|
// // Forward request to a chosen backend
|
|
// let mut res_backend = {
|
|
// let Ok(result) = timeout(self.globals.proxy_config.upstream_timeout, self.forwarder.request(req)).await else {
|
|
// 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);
|
|
// }
|
|
// }
|
|
// };
|
|
//////////////
|
|
// Process reverse proxy context generated during the forwarding request generation.
|
|
#[cfg(feature = "sticky-cookie")]
|
|
if let Some(context_from_lb) = _context.context_lb {
|
|
let res_headers = res_backend.headers_mut();
|
|
if let Err(e) = set_sticky_cookie_lb_context(res_headers, &context_from_lb) {
|
|
error!("Failed to append context to the response given from backend: {}", e);
|
|
return Err(HttpError::FailedToAddSetCookeInResponse);
|
|
}
|
|
}
|
|
|
|
if res_backend.status() != StatusCode::SWITCHING_PROTOCOLS {
|
|
// // Generate response to client
|
|
// if self.generate_response_forwarded(&mut res_backend, backend).is_err() {
|
|
// 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
|
|
// let upgrade_in_response = extract_upgrade(res_backend.headers());
|
|
// 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()
|
|
// } else {
|
|
// false
|
|
// };
|
|
// if !should_upgrade {
|
|
// error!(
|
|
// "Backend tried to switch to protocol {:?} when {:?} was requested",
|
|
// upgrade_in_response, upgrade_in_request
|
|
// );
|
|
// 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)
|
|
}
|
|
}
|