diff --git a/src/log.rs b/src/log.rs index cb4253e..7a20d42 100644 --- a/src/log.rs +++ b/src/log.rs @@ -1 +1,101 @@ +use std::net::SocketAddr; + pub use log::{debug, error, info, warn}; + +use crate::utils::ToCanonical; + +#[derive(Debug, Clone)] +pub struct MessageLog { + // pub tls_server_name: String, + pub client_addr: String, + pub method: String, + pub host: String, + pub p_and_q: String, + pub version: hyper::Version, + pub uri_scheme: String, + pub uri_host: String, + pub ua: String, + pub xff: String, + pub status: String, + pub upstream: String, +} + +impl From<&hyper::Request> for MessageLog { + fn from(req: &hyper::Request) -> Self { + let header_mapper = |v: hyper::header::HeaderName| { + req + .headers() + .get(v) + .map_or_else(|| "", |s| s.to_str().unwrap_or("")) + .to_string() + }; + Self { + // tls_server_name: "".to_string(), + client_addr: "".to_string(), + method: req.method().to_string(), + host: header_mapper(hyper::header::HOST), + p_and_q: req + .uri() + .path_and_query() + .map_or_else(|| "", |v| v.as_str()) + .to_string(), + version: req.version(), + uri_scheme: req.uri().scheme_str().unwrap_or("").to_string(), + uri_host: req.uri().host().unwrap_or("").to_string(), + ua: header_mapper(hyper::header::USER_AGENT), + xff: header_mapper(hyper::header::HeaderName::from_static("x-forwarded-for")), + status: "".to_string(), + upstream: "".to_string(), + } + } +} + +impl MessageLog { + pub fn client_addr(&mut self, client_addr: &SocketAddr) -> &mut Self { + self.client_addr = client_addr.to_canonical().to_string(); + self + } + // pub fn tls_server_name(&mut self, tls_server_name: &str) -> &mut Self { + // self.tls_server_name = tls_server_name.to_string(); + // self + // } + pub fn status_code(&mut self, status_code: &hyper::StatusCode) -> &mut Self { + self.status = status_code.to_string(); + self + } + pub fn xff(&mut self, xff: &Option<&hyper::header::HeaderValue>) -> &mut Self { + self.xff = xff + .map_or_else(|| "", |v| v.to_str().unwrap_or("")) + .to_string(); + self + } + pub fn upstream(&mut self, upstream: &str) -> &mut Self { + self.upstream = upstream.to_string(); + self + } + + pub fn output(&self) { + info!( + "{} <- {} -- {} {} {:?} -- {} -- {} \"{}\", \"{}\" \"{}\"", + if !self.host.is_empty() { + self.host.as_str() + } else { + self.uri_host.as_str() + }, + self.client_addr, + self.method, + self.p_and_q, + self.version, + self.status, + if !self.uri_scheme.is_empty() && !self.uri_host.is_empty() { + format!("{}://{}", self.uri_scheme, self.uri_host) + } else { + "".to_string() + }, + self.ua, + self.xff, + self.upstream, + // self.tls_server_name + ); + } +} diff --git a/src/msg_handler/handler.rs b/src/msg_handler/handler.rs index 135b971..22ac66a 100644 --- a/src/msg_handler/handler.rs +++ b/src/msg_handler/handler.rs @@ -32,6 +32,15 @@ impl HttpMessageHandler where T: Connect + Clone + Sync + Send + 'static, { + fn return_with_error_log( + &self, + status_code: StatusCode, + log_data: &mut MessageLog, + ) -> Result> { + log_data.status_code(&status_code).output(); + http_error(status_code) + } + pub async fn handle_request( self, mut req: Request, @@ -40,15 +49,24 @@ where tls_enabled: bool, tls_server_name: Option, ) -> Result> { - req.log_debug(&client_addr, Some("(Request from Client)")); + 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 = req.parse_host()?.to_ascii_lowercase(); + 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. - if self.globals.sni_consistency - && !server_name_bytes.eq_ignore_ascii_case(&tls_server_name.unwrap()) - { - return http_error(StatusCode::MISDIRECTED_REQUEST); + #[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) { @@ -57,14 +75,15 @@ where debug!("Serving by default app"); self.globals.backends.apps.get(default_server_name).unwrap() } else { - // info!("{} => {}", request_log, StatusCode::SERVICE_UNAVAILABLE); - return http_error(StatusCode::SERVICE_UNAVAILABLE); + 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); - // info!("{} => {}", request_log, StatusCode::PERMANENT_REDIRECT); + log_data + .status_code(&StatusCode::PERMANENT_REDIRECT) + .output(); return secure_redirection(&backend.server_name, self.globals.https_port, &req); } @@ -74,12 +93,12 @@ where let upstream = if let Some(upstream) = backend.reverse_proxy.get(path) { upstream } else { - return http_error(StatusCode::NOT_FOUND); + 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 http_error(StatusCode::INTERNAL_SERVER_ERROR); + return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data); }; // Upgrade in request header @@ -99,10 +118,13 @@ where req } else { error!("Failed to generate destination uri for reverse proxy"); - return http_error(StatusCode::SERVICE_UNAVAILABLE); + return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data); }; // debug!("Request to be forwarded: {:?}", req_forwarded); - req_forwarded.log_debug(&client_addr, Some("(Request to Backend)")); + req_forwarded.log_debug(&client_addr, Some("(to Backend)")); + log_data.xff(&req_forwarded.headers().get("x-forwarded-for")); + log_data.upstream(&upstream_scheme_host.to_string()); + ////// // Forward request to let mut res_backend = { @@ -113,23 +135,19 @@ where .await { Err(_) => { - return http_error(StatusCode::GATEWAY_TIMEOUT); + 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 http_error(StatusCode::BAD_REQUEST); + return self.return_with_error_log(StatusCode::SERVICE_UNAVAILABLE, &mut log_data); } }, } }; - res_backend.log_debug( - &backend.server_name, - &client_addr, - Some("(Response from Backend)"), - ); + 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 { @@ -143,51 +161,54 @@ where false } { if let Some(request_upgraded) = request_upgraded { - let mut response_upgraded = res_backend + let onupgrade = if let Some(onupgrade) = res_backend .extensions_mut() .remove::() - .ok_or_else(|| anyhow!("Response does not have an upgrade extension"))? // TODO: any response code? - .await?; - // TODO: H3で死ぬことがある - // thread 'rpxy' panicked at 'Failed to upgrade request: hyper::Error(User(ManualUpgrade))', src/proxy/proxy_handler.rs:124:63 + { + onupgrade + } else { + error!("Response does not have an upgrade extension"); + return self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data); + }; + let mut response_upgraded = if let Ok(res) = onupgrade.await { + res + } else { + error!("No upgrade is available"); + return self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data); + }; + self.globals.runtime_handle.spawn(async move { let mut request_upgraded = request_upgraded.await.map_err(|e| { error!("Failed to upgrade request: {}", e); anyhow!("Failed to upgrade request: {}", e) - })?; // TODO: any response code? + })?; copy_bidirectional(&mut response_upgraded, &mut request_upgraded) .await - .map_err(|e| anyhow!("Coping between upgraded connections failed: {}", e))?; // TODO: any response code? + .map_err(|e| anyhow!("Coping between upgraded connections failed: {}", e))?; Ok(()) as Result<()> }); // info!("{} => {}", request_log, response_log); Ok(res_backend) } else { error!("Request does not have an upgrade extension"); - // info!("{} => {}", request_log, StatusCode::BAD_REQUEST); - http_error(StatusCode::BAD_REQUEST) + 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 ); - // info!("{} => {}", request_log, StatusCode::SERVICE_UNAVAILABLE); - http_error(StatusCode::SERVICE_UNAVAILABLE) + self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data) } } else { // 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("(Response to Client)"), - ); + res_backend.log_debug(&backend.server_name, &client_addr, Some("(to Client)")); + log_data.status_code(&res_backend.status()).output(); Ok(res_backend) } else { - // info!("{} => {}", request_log, StatusCode::BAD_GATEWAY); - http_error(StatusCode::BAD_GATEWAY) + self.return_with_error_log(StatusCode::INTERNAL_SERVER_ERROR, &mut log_data) } } } diff --git a/src/proxy/proxy_tls.rs b/src/proxy/proxy_tls.rs index bc38b3c..4d2c571 100644 --- a/src/proxy/proxy_tls.rs +++ b/src/proxy/proxy_tls.rs @@ -101,7 +101,7 @@ where } #[cfg(feature = "h3")] - async fn parse_sni_and_get_crypto_h3<'a>( + async fn parse_sni_and_get_crypto_h3<'a, 'b>( &self, peeked_conn: &mut quinn::Connecting, server_crypto_map: &'a ServerCryptoMap,