1use std::collections::HashMap;
2use std::net::SocketAddr;
3use std::str::FromStr;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_channel::Sender;
8use chrono::{DateTime, Local};
9use ferron_common::logging::{ErrorLogger, LogMessage};
10use ferron_common::observability::{MetricsMultiSender, TraceSignal};
11#[cfg(feature = "runtime-vibeio")]
12use ferron_common::util::FileStream;
13use futures_util::stream::TryStreamExt;
14use http_body_util::combinators::BoxBody;
15use http_body_util::{BodyExt, Empty, Full, StreamBody};
16use hyper::body::{Body, Bytes, Frame};
17use hyper::header::{HeaderName, HeaderValue};
18use hyper::{header, HeaderMap, Method, Request, Response, StatusCode};
19#[cfg(feature = "runtime-tokio")]
20use tokio::io::BufReader;
21#[cfg(feature = "runtime-tokio")]
22use tokio_util::io::ReaderStream;
23
24use crate::config::{ServerConfiguration, ServerConfigurationValue, ServerConfigurations};
25use crate::get_value;
26use crate::runtime::timeout;
27#[cfg(feature = "runtime-monoio")]
28use crate::util::MonoioFileStreamNoSpawn;
29use crate::util::{
30 generate_access_log_message, generate_default_error_page, replace_header_placeholders, sanitize_url, SERVER_SOFTWARE,
31};
32
33use ferron_common::modules::{ModuleHandlers, RequestData, SocketData};
34use ferron_common::{get_entries, get_entry};
35
36async fn generate_error_response(
38 status_code: StatusCode,
39 config: &ServerConfiguration,
40 headers: &Option<HeaderMap>,
41) -> Response<BoxBody<Bytes, std::io::Error>> {
42 let bare_body = generate_default_error_page(
43 status_code,
44 get_value!("server_administrator_email", config).and_then(|v| v.as_str()),
45 );
46 let mut content_length: Option<u64> = bare_body.len().try_into().ok();
47 let mut response_body = Full::new(Bytes::from(bare_body)).map_err(|e| match e {}).boxed();
48
49 if let Some(error_pages) = get_entries!("error_page", config) {
50 for error_page in &error_pages.inner {
51 if let Some(page_status_code) = error_page.values.first().and_then(|v| v.as_i128()) {
52 if page_status_code
53 .try_into()
54 .ok()
55 .and_then(|s| StatusCode::from_u16(s).ok())
56 .is_none_or(|s| s != status_code)
57 {
58 continue;
59 }
60 if let Some(page_path) = error_page.values.get(1).and_then(|v| v.as_str()) {
61 #[cfg(feature = "runtime-monoio")]
62 let file = monoio::fs::File::open(page_path).await;
63 #[cfg(feature = "runtime-tokio")]
64 let file = tokio::fs::File::open(page_path).await;
65 #[cfg(feature = "runtime-vibeio")]
66 let file = vibeio::fs::File::open(page_path).await;
67
68 let Ok(file) = file else {
69 continue;
70 };
71
72 #[cfg(any(
74 feature = "runtime-tokio",
75 feature = "runtime-vibeio",
76 all(feature = "runtime-monoio", unix)
77 ))]
78 let metadata = file.metadata().await;
79 #[cfg(all(feature = "runtime-monoio", windows))]
80 let metadata = {
81 let page_path = page_path.to_owned();
82 monoio::spawn_blocking(move || std::fs::metadata(page_path))
83 .await
84 .unwrap_or(Err(std::io::Error::other(
85 "Can't spawn a blocking task to obtain the file metadata",
86 )))
87 };
88
89 content_length = metadata.ok().map(|m| m.len());
90
91 #[cfg(feature = "runtime-monoio")]
92 let file_stream = MonoioFileStreamNoSpawn::new(file, None, content_length);
93 #[cfg(feature = "runtime-vibeio")]
94 let file_stream = FileStream::new(file, None, content_length);
95 #[cfg(feature = "runtime-tokio")]
96 let file_stream = ReaderStream::new(BufReader::with_capacity(12800, file));
97
98 let stream_body = StreamBody::new(file_stream.map_ok(Frame::data));
99 let boxed_body = stream_body.boxed();
100
101 response_body = boxed_body;
102
103 break;
104 }
105 }
106 }
107 }
108
109 let mut response_builder = Response::builder().status(status_code);
110
111 if let Some(headers) = headers {
112 let headers_iter = headers.iter();
113 for (name, value) in headers_iter {
114 if name != header::CONTENT_TYPE && name != header::CONTENT_LENGTH {
115 response_builder = response_builder.header(name, value);
116 }
117 }
118 }
119
120 if let Some(content_length) = content_length {
121 response_builder = response_builder.header(header::CONTENT_LENGTH, content_length);
122 }
123 response_builder = response_builder.header(header::CONTENT_TYPE, HeaderValue::from_static("text/html"));
124
125 response_builder.body(response_body).unwrap_or_default()
126}
127
128#[allow(clippy::too_many_arguments)]
130async fn log_access(
131 loggers: &[Sender<LogMessage>],
132 request_parts: &hyper::http::request::Parts,
133 socket_data: &SocketData,
134 auth_user: Option<&str>,
135 status_code: u16,
136 content_length: Option<u64>,
137 date_format: Option<&str>,
138 log_format: Option<&str>,
139 log_json_props: Option<&HashMap<String, ServerConfigurationValue>>,
140) {
141 let now: DateTime<Local> = Local::now();
142 let formatted_time = now.format(date_format.unwrap_or("%d/%b/%Y:%H:%M:%S %z")).to_string();
143 let log_message_string = generate_access_log_message(
144 request_parts,
145 socket_data,
146 auth_user,
147 &formatted_time,
148 status_code,
149 content_length,
150 log_format,
151 log_json_props,
152 );
153 for logger in loggers {
154 logger
155 .send(LogMessage::new(log_message_string.clone(), false))
156 .await
157 .unwrap_or_default();
158 }
159}
160
161#[inline]
163fn add_custom_headers(
164 response_parts: &mut hyper::http::response::Parts,
165 headers_to_add: &HeaderMap,
166 headers_to_replace: &HeaderMap,
167 headers_to_remove: &[HeaderName],
168) {
169 for (header_name, header_value) in headers_to_add {
170 if !response_parts.headers.contains_key(header_name) {
171 response_parts.headers.insert(header_name, header_value.to_owned());
172 }
173 }
174
175 for (header_name, header_value) in headers_to_replace {
176 response_parts.headers.insert(header_name, header_value.to_owned());
177 }
178
179 for header_to_remove in headers_to_remove.iter().rev() {
180 if response_parts.headers.contains_key(header_to_remove) {
181 while response_parts.headers.remove(header_to_remove).is_some() {}
182 }
183 }
184}
185
186#[inline]
188fn add_http3_alt_svc_header(response_parts: &mut hyper::http::response::Parts, http3_alt_port: Option<u16>) {
189 if let Some(http3_alt_port) = http3_alt_port {
190 if let Ok(header_value) = match response_parts.headers.get(header::ALT_SVC) {
191 Some(value) => {
192 let header_value_old = String::from_utf8_lossy(value.as_bytes());
193 let header_value_new = format!("h3=\":{http3_alt_port}\", h3-29=\":{http3_alt_port}\"");
194
195 if header_value_old != header_value_new {
196 HeaderValue::from_bytes(format!("{header_value_old}, {header_value_new}").as_bytes())
197 } else {
198 HeaderValue::from_bytes(header_value_old.as_bytes())
199 }
200 }
201 None => HeaderValue::from_bytes(format!("h3=\":{http3_alt_port}\", h3-29=\":{http3_alt_port}\"").as_bytes()),
202 } {
203 response_parts.headers.insert(header::ALT_SVC, header_value);
204 }
205 }
206}
207
208#[inline]
210fn add_server_header(response_parts: &mut hyper::http::response::Parts) {
211 response_parts
212 .headers
213 .insert(header::SERVER, HeaderValue::from_static(SERVER_SOFTWARE));
214}
215
216fn extract_content_length(response: &Response<BoxBody<Bytes, std::io::Error>>) -> Option<u64> {
218 response
219 .headers()
220 .get(header::CONTENT_LENGTH)
221 .and_then(|header_value| {
222 header_value.to_str().ok().and_then(|header_value| {
223 header_value
224 .parse::<u64>()
225 .ok()
226 .or_else(|| response.body().size_hint().exact())
227 })
228 })
229 .or_else(|| response.body().size_hint().exact())
230}
231
232fn basic_error_response(status_code: StatusCode) -> Response<BoxBody<Bytes, std::io::Error>> {
234 Response::builder()
235 .status(status_code)
236 .header(header::CONTENT_TYPE, HeaderValue::from_static("text/html"))
237 .body(
238 Full::new(Bytes::from(generate_default_error_page(status_code, None)))
239 .map_err(|e| match e {})
240 .boxed(),
241 )
242 .unwrap_or_default()
243}
244
245fn build_custom_headers(
247 configuration: &ServerConfiguration,
248 request_parts: &hyper::http::request::Parts,
249) -> (HeaderMap, HeaderMap, Vec<HeaderName>) {
250 let mut headers_to_add = HeaderMap::new();
251 let mut headers_to_replace = HeaderMap::new();
252 let mut headers_to_remove = Vec::new();
253
254 if let Some(custom_headers) = get_entries!("header", configuration) {
255 for custom_header in custom_headers.inner.iter().rev() {
256 if let Some(header_name) = custom_header.values.first().and_then(|v| v.as_str()) {
257 if let Some(header_value) = custom_header.values.get(1).and_then(|v| v.as_str()) {
258 if !headers_to_add.contains_key(header_name) {
259 if let Ok(header_name) = HeaderName::from_str(header_name) {
260 if let Ok(header_value) =
261 HeaderValue::from_str(&replace_header_placeholders(header_value, request_parts, None))
262 {
263 headers_to_add.insert(header_name, header_value);
264 }
265 }
266 }
267 }
268 }
269 }
270 }
271
272 if let Some(custom_headers) = get_entries!("header_replace", configuration) {
273 for custom_header in custom_headers.inner.iter().rev() {
274 if let Some(header_name) = custom_header.values.first().and_then(|v| v.as_str()) {
275 if let Some(header_value) = custom_header.values.get(1).and_then(|v| v.as_str()) {
276 if let Ok(header_name) = HeaderName::from_str(header_name) {
277 if let Ok(header_value) =
278 HeaderValue::from_str(&replace_header_placeholders(header_value, request_parts, None))
279 {
280 headers_to_replace.insert(header_name, header_value);
281 }
282 }
283 }
284 }
285 }
286 }
287
288 if let Some(custom_headers_to_remove) = get_entries!("header_remove", configuration) {
289 for custom_header in custom_headers_to_remove.inner.iter().rev() {
290 if let Some(header_name) = custom_header.values.first().and_then(|v| v.as_str()) {
291 if let Ok(header_name) = HeaderName::from_str(header_name) {
292 headers_to_remove.push(header_name);
293 }
294 }
295 }
296 }
297
298 (headers_to_add, headers_to_replace, headers_to_remove)
299}
300
301#[allow(clippy::too_many_arguments)]
303async fn finalize_response_and_log(
304 response: Response<BoxBody<Bytes, std::io::Error>>,
305 http3_alt_port: Option<u16>,
306 headers_to_add: HeaderMap,
307 headers_to_replace: HeaderMap,
308 headers_to_remove: Vec<HeaderName>,
309 loggers: &[Sender<LogMessage>],
310 request_parts: &Option<hyper::http::request::Parts>,
311 socket_data: &SocketData,
312 latest_auth_data: Option<&str>,
313 date_format: Option<&str>,
314 log_format: Option<&str>,
315 log_json_props: Option<&HashMap<String, ServerConfigurationValue>>,
316) -> Response<BoxBody<Bytes, std::io::Error>> {
317 let (mut response_parts, response_body) = response.into_parts();
318
319 add_http3_alt_svc_header(&mut response_parts, http3_alt_port);
320 add_server_header(&mut response_parts);
321 add_custom_headers(
322 &mut response_parts,
323 &headers_to_add,
324 &headers_to_replace,
325 &headers_to_remove,
326 );
327
328 let response = Response::from_parts(response_parts, response_body);
329
330 if let Some(request_parts) = request_parts {
331 if !loggers.is_empty() {
332 log_access(
333 loggers,
334 request_parts,
335 socket_data,
336 latest_auth_data,
337 response.status().as_u16(),
338 extract_content_length(&response),
339 date_format,
340 log_format,
341 log_json_props,
342 )
343 .await;
344 }
345 }
346
347 response
348}
349
350#[allow(clippy::too_many_arguments)]
352async fn finalize_basic_error_response(
353 response: Response<BoxBody<Bytes, std::io::Error>>,
354 request_parts: hyper::http::request::Parts,
355 http3_alt_port: Option<u16>,
356 loggers: &[Sender<LogMessage>],
357 socket_data: &SocketData,
358 date_format: Option<&str>,
359 log_format: Option<&str>,
360 log_json_props: Option<&HashMap<String, ServerConfigurationValue>>,
361) -> Response<BoxBody<Bytes, std::io::Error>> {
362 let request_parts = Some(request_parts);
363 finalize_response_and_log(
364 response,
365 http3_alt_port,
366 HeaderMap::new(),
367 HeaderMap::new(),
368 Vec::new(),
369 loggers,
370 &request_parts,
371 socket_data,
372 None,
373 date_format,
374 log_format,
375 log_json_props,
376 )
377 .await
378}
379
380#[allow(clippy::too_many_arguments)]
382#[inline]
383async fn execute_response_modifying_handlers(
384 mut response: Response<BoxBody<Bytes, std::io::Error>>,
385 mut executed_handlers: Vec<Box<dyn ModuleHandlers>>,
386 configuration: &ServerConfiguration,
387 http3_alt_port: Option<u16>,
388 headers_to_add: HeaderMap,
389 headers_to_replace: HeaderMap,
390 headers_to_remove: Vec<HeaderName>,
391 loggers: &[Sender<LogMessage>],
392 request_parts: &Option<hyper::http::request::Parts>,
393 socket_data: &SocketData,
394 latest_auth_data: Option<&str>,
395 date_format: Option<&str>,
396 log_format: Option<&str>,
397 log_json_props: Option<&HashMap<String, ServerConfigurationValue>>,
398 metrics_sender: MetricsMultiSender,
399 metrics_enabled: bool,
400 traces_senders: Vec<Sender<TraceSignal>>,
401 traces_enabled: bool,
402 timeout_instant: std::time::Instant,
403 timeout_duration: Option<std::time::Duration>,
404) -> Result<Result<Response<BoxBody<Bytes, std::io::Error>>, Response<BoxBody<Bytes, std::io::Error>>>, anyhow::Error> {
405 while let Some(mut executed_handler) = executed_handlers.pop() {
406 if traces_enabled {
407 for trace_sender in &traces_senders {
408 trace_sender
409 .send(TraceSignal::StartSpan(format!(
410 "{}::response_modifying_handler",
411 executed_handler.get_name()
412 )))
413 .await
414 .unwrap_or_default();
415 }
416 }
417 let (response_status, is_timeout) = if let Some(timeout_duration) = &timeout_duration {
418 let elapsed = timeout_instant.elapsed();
419 if let Some(timeout_cur_duration) = timeout_duration.checked_sub(elapsed) {
420 match timeout(
421 timeout_cur_duration,
422 executed_handler.response_modifying_handler(response),
423 )
424 .await
425 {
426 Ok(result) => (result, false),
427 Err(_) => (Err(anyhow::anyhow!("The client or server has timed out").into()), true),
428 }
429 } else {
430 (Err(anyhow::anyhow!("The client or server has timed out").into()), true)
431 }
432 } else {
433 (executed_handler.response_modifying_handler(response).await, false)
434 };
435 if traces_enabled {
436 for trace_sender in &traces_senders {
437 trace_sender
438 .send(TraceSignal::EndSpan(
439 format!("{}::response_modifying_handler", executed_handler.get_name()),
440 response_status.as_ref().err().map(|e| e.to_string()),
441 ))
442 .await
443 .unwrap_or_default();
444 }
445 }
446 if is_timeout {
447 if metrics_enabled {
448 while let Some(mut executed_handler) = executed_handlers.pop() {
449 executed_handler.metric_data_after_handler(&metrics_sender).await;
450 }
451 }
452 Err(anyhow::anyhow!("The client or server has timed out"))?;
453 }
454 response = match response_status {
455 Ok(response) => response,
456 Err(err) => {
457 for logger in loggers {
458 logger
459 .send(LogMessage::new(
460 format!("Unexpected error while serving a request: {err}"),
461 true,
462 ))
463 .await
464 .unwrap_or_default();
465 }
466
467 let error_response = generate_error_response(StatusCode::INTERNAL_SERVER_ERROR, configuration, &None).await;
468
469 let final_response = finalize_response_and_log(
470 error_response,
471 http3_alt_port,
472 headers_to_add,
473 headers_to_replace,
474 headers_to_remove,
475 loggers,
476 request_parts,
477 socket_data,
478 latest_auth_data,
479 date_format,
480 log_format,
481 log_json_props,
482 )
483 .await;
484
485 if metrics_enabled {
486 while let Some(mut executed_handler) = executed_handlers.pop() {
487 executed_handler.metric_data_after_handler(&metrics_sender).await;
488 }
489 }
490
491 return Ok(Err(final_response));
492 }
493 };
494 if metrics_enabled {
495 executed_handler.metric_data_after_handler(&metrics_sender).await;
496 }
497 }
498 Ok(Ok(response))
499}
500
501#[allow(clippy::too_many_arguments)]
503async fn finalize_with_modifying_handlers(
504 response: Response<BoxBody<Bytes, std::io::Error>>,
505 executed_handlers: Vec<Box<dyn ModuleHandlers>>,
506 configuration: &ServerConfiguration,
507 http3_alt_port: Option<u16>,
508 headers_to_add: HeaderMap,
509 headers_to_replace: HeaderMap,
510 headers_to_remove: Vec<HeaderName>,
511 log_request_parts: &Option<hyper::http::request::Parts>,
512 socket_data: &SocketData,
513 latest_auth_data: Option<&str>,
514 log_date_format: Option<&str>,
515 log_format: Option<&str>,
516 log_json_props: Option<&HashMap<String, ServerConfigurationValue>>,
517 metrics_sender: MetricsMultiSender,
518 metrics_enabled: bool,
519 traces_senders: Vec<Sender<TraceSignal>>,
520 traces_enabled: bool,
521 timeout_instant: std::time::Instant,
522 timeout_duration: Option<std::time::Duration>,
523) -> Result<Response<BoxBody<Bytes, std::io::Error>>, anyhow::Error> {
524 let (mut response_parts, response_body) = response.into_parts();
525
526 add_http3_alt_svc_header(&mut response_parts, http3_alt_port);
527 add_server_header(&mut response_parts);
528 add_custom_headers(
529 &mut response_parts,
530 &headers_to_add,
531 &headers_to_replace,
532 &headers_to_remove,
533 );
534
535 let response = Response::from_parts(response_parts, response_body);
536
537 match execute_response_modifying_handlers(
538 response,
539 executed_handlers,
540 configuration,
541 http3_alt_port,
542 headers_to_add,
543 headers_to_replace,
544 headers_to_remove,
545 &configuration.observability.log_channels,
546 log_request_parts,
547 socket_data,
548 latest_auth_data,
549 log_date_format,
550 log_format,
551 log_json_props,
552 metrics_sender,
553 metrics_enabled,
554 traces_senders,
555 traces_enabled,
556 timeout_instant,
557 timeout_duration,
558 )
559 .await?
560 {
561 Ok(response) => {
562 if let Some(request_parts) = log_request_parts.as_ref() {
563 log_access(
564 &configuration.observability.log_channels,
565 request_parts,
566 socket_data,
567 latest_auth_data,
568 response.status().as_u16(),
569 extract_content_length(&response),
570 log_date_format,
571 log_format,
572 log_json_props,
573 )
574 .await;
575 }
576 Ok(response)
577 }
578 Err(error_response) => Ok(error_response),
579 }
580}
581
582#[allow(clippy::type_complexity)]
583#[allow(clippy::result_large_err)]
584fn sanitize_url_request(
585 configuration: &ServerConfiguration,
586 mut request: Request<BoxBody<Bytes, std::io::Error>>,
587) -> Result<
588 (Request<BoxBody<Bytes, std::io::Error>>, bool),
589 (
590 Request<BoxBody<Bytes, std::io::Error>>,
591 Box<dyn std::error::Error + Send + Sync>,
592 ),
593> {
594 let url_pathname = request.uri().path();
595 let sanitized_url_pathname = match sanitize_url(
596 url_pathname,
597 get_value!("allow_double_slashes", configuration)
598 .and_then(|v| v.as_bool())
599 .unwrap_or(false),
600 ) {
601 Ok(sanitized_url_pathname) => sanitized_url_pathname,
602 Err(err) => {
603 return Err((request, err.into()));
604 }
605 };
606
607 if sanitized_url_pathname != url_pathname {
608 let (mut parts, body) = request.into_parts();
609 let orig_uri = parts.uri.clone();
610 let mut url_parts = parts.uri.into_parts();
611 url_parts.path_and_query = Some(
612 match format!(
613 "{}{}",
614 sanitized_url_pathname,
615 url_parts
616 .path_and_query
617 .as_ref()
618 .and_then(|pq| pq.query())
619 .map_or("".to_string(), |q| format!("?{q}"))
620 )
621 .parse()
622 {
623 Ok(path_and_query) => path_and_query,
624 Err(e) => {
625 parts.uri = orig_uri;
626 request = Request::from_parts(parts, body);
627 return Err((request, e.into()));
628 }
629 },
630 );
631 parts.uri = match hyper::Uri::from_parts(url_parts) {
632 Ok(uri) => uri,
633 Err(e) => {
634 parts.uri = orig_uri;
635 request = Request::from_parts(parts, body);
636 return Err((request, e.into()));
637 }
638 };
639 request = Request::from_parts(parts, body);
640 return Ok((request, true));
641 }
642 Ok((request, false))
643}
644
645#[allow(clippy::too_many_arguments)]
647pub async fn request_handler(
648 mut request: Request<BoxBody<Bytes, std::io::Error>>,
649 client_address: SocketAddr,
650 server_address: SocketAddr,
651 encrypted: bool,
652 configurations: Arc<ServerConfigurations>,
653 http3_alt_port: Option<u16>,
654 acme_http_01_resolvers: Arc<tokio::sync::RwLock<Vec<crate::acme::Http01DataLock>>>,
655 proxy_protocol_client_address: Option<SocketAddr>,
656 proxy_protocol_server_address: Option<SocketAddr>,
657) -> Result<Response<BoxBody<Bytes, std::io::Error>>, anyhow::Error> {
658 let global_configuration = configurations.find_global_configuration();
660 let global_loggers: &[Sender<LogMessage>] = global_configuration
661 .as_ref()
662 .map_or(&[], |c| &*c.observability.log_channels);
663 let global_log_date_format = global_configuration
664 .as_deref()
665 .and_then(|c| get_value!("log_date_format", c))
666 .and_then(|v| v.as_str());
667 let global_log_format = global_configuration
668 .as_deref()
669 .and_then(|c| get_value!("log_format", c))
670 .and_then(|v| v.as_str());
671 let global_log_json = global_configuration
672 .as_deref()
673 .and_then(|c| c.entries.get("log_json"))
674 .and_then(|entries| entries.get_entry())
675 .map(|entry| entry.props.clone());
676
677 let timeout_from_config = global_configuration
679 .as_deref()
680 .and_then(|c| get_entry!("timeout", c))
681 .and_then(|e| e.values.last());
682 let timeout_duration = if timeout_from_config.is_some_and(|v| v.is_null()) {
683 None
684 } else {
685 let timeout_millis = timeout_from_config.and_then(|v| v.as_i128()).unwrap_or(300000) as u64;
686 Some(Duration::from_millis(timeout_millis))
687 };
688 let timeout_instant = std::time::Instant::now();
689
690 if matches!(request.version(), hyper::Version::HTTP_2 | hyper::Version::HTTP_3) {
692 if let Some(authority) = request.uri().authority() {
694 let authority = authority.to_owned();
695 let headers = request.headers_mut();
696 if !headers.contains_key(header::HOST) {
697 if let Ok(authority_value) = HeaderValue::from_bytes(authority.as_str().as_bytes()) {
698 headers.append(header::HOST, authority_value);
699 }
700 }
701 }
702
703 let mut cookie_normalized = String::new();
705 let mut cookie_set = false;
706 let headers = request.headers_mut();
707 for cookie in headers.get_all(header::COOKIE) {
708 if let Ok(cookie) = cookie.to_str() {
709 if cookie_set {
710 cookie_normalized.push_str("; ");
711 }
712 cookie_set = true;
713 cookie_normalized.push_str(cookie);
714 }
715 }
716 if cookie_set {
717 if let Ok(cookie_value) = HeaderValue::from_bytes(cookie_normalized.as_bytes()) {
718 headers.insert(header::COOKIE, cookie_value);
719 }
720 }
721 }
722
723 let mut socket_data = SocketData {
725 remote_addr: proxy_protocol_client_address.unwrap_or(client_address),
726 local_addr: proxy_protocol_server_address.unwrap_or(server_address),
727 encrypted,
728 };
729
730 if request.method() == Method::CONNECT && !request.uri().path().is_empty() {
732 for logger in global_loggers {
733 logger
734 .send(LogMessage::new("CONNECT request with non-empty path".to_string(), true))
735 .await
736 .unwrap_or_default();
737 }
738 let response = basic_error_response(StatusCode::BAD_REQUEST);
739 let (request_parts, _) = request.into_parts();
740 return Ok(
741 finalize_basic_error_response(
742 response,
743 request_parts,
744 http3_alt_port,
745 global_loggers,
746 &socket_data,
747 global_log_date_format,
748 global_log_format,
749 global_log_json.as_ref(),
750 )
751 .await,
752 );
753 }
754
755 let host_header_option = request.headers().get(header::HOST);
757 if let Some(header_data) = host_header_option {
758 match header_data.to_str() {
759 Ok(host_header) => {
760 let host_header_lower_case = host_header.to_lowercase();
761 let host_header_without_dot = host_header_lower_case
762 .strip_suffix('.')
763 .unwrap_or(host_header_lower_case.as_str());
764 if host_header_without_dot != host_header {
765 let host_header_value = match HeaderValue::from_str(host_header_without_dot) {
766 Ok(host_header_value) => host_header_value,
767 Err(err) => {
768 for logger in global_loggers {
769 logger
770 .send(LogMessage::new(format!("Host header sanitation error: {err}"), true))
771 .await
772 .unwrap_or_default();
773 }
774 let response = basic_error_response(StatusCode::BAD_REQUEST);
775 let (request_parts, _) = request.into_parts();
776 return Ok(
777 finalize_basic_error_response(
778 response,
779 request_parts,
780 http3_alt_port,
781 global_loggers,
782 &socket_data,
783 global_log_date_format,
784 global_log_format,
785 global_log_json.as_ref(),
786 )
787 .await,
788 );
789 }
790 };
791
792 request.headers_mut().insert(header::HOST, host_header_value);
793 }
794 }
795 Err(err) => {
796 for logger in global_loggers {
797 logger
798 .send(LogMessage::new(format!("Host header sanitation error: {err}"), true))
799 .await
800 .unwrap_or_default();
801 }
802 let response = basic_error_response(StatusCode::BAD_REQUEST);
803 let (request_parts, _) = request.into_parts();
804 return Ok(
805 finalize_basic_error_response(
806 response,
807 request_parts,
808 http3_alt_port,
809 global_loggers,
810 &socket_data,
811 global_log_date_format,
812 global_log_format,
813 global_log_json.as_ref(),
814 )
815 .await,
816 );
817 }
818 }
819 };
820
821 let hostname_determinant = request.headers().get(header::HOST).and_then(|value| {
822 value.to_str().ok().map(|h| {
823 h.rsplit_once(':')
824 .and_then(|(left, right)| {
825 if right.parse::<u16>().is_ok() {
826 Some(left.to_string())
827 } else {
828 None
829 }
830 })
831 .unwrap_or_else(|| h.to_string())
832 })
833 });
834
835 let (request_parts, request_body) = request.into_parts();
836 let mut log_request_parts = if global_configuration
837 .as_ref()
838 .is_some_and(|c| !c.observability.log_channels.is_empty())
839 {
840 Some(request_parts.clone())
841 } else {
842 None
843 };
844 let request = Request::from_parts(request_parts, request_body);
845
846 let (request_parts, request_body) = request.into_parts();
848 let configuration_option =
849 configurations.find_configuration(&request_parts, hostname_determinant.as_deref(), &socket_data);
850 let mut request = Request::from_parts(request_parts, request_body);
851 let mut configuration_error_handler_lookup = match configuration_option
852 .and_then(|c| c.ok_or_else(|| anyhow::anyhow!("No matching configuration found").into_boxed_dyn_error()))
853 {
854 Ok(configuration) => configuration,
855 Err(err) => {
856 for logger in global_loggers {
857 logger
858 .send(LogMessage::new(
859 format!("Cannot determine server configuration: {err}"),
860 true,
861 ))
862 .await
863 .unwrap_or_default()
864 }
865 let response = basic_error_response(StatusCode::INTERNAL_SERVER_ERROR);
866 let (request_parts, _) = request.into_parts();
867 return Ok(
868 finalize_basic_error_response(
869 response,
870 request_parts,
871 http3_alt_port,
872 global_loggers,
873 &socket_data,
874 global_log_date_format,
875 global_log_format,
876 global_log_json.as_ref(),
877 )
878 .await,
879 );
880 }
881 };
882 let mut configuration = match configuration_error_handler_lookup.get_default().cloned() {
883 Some(configuration) => configuration,
884 None => {
885 for logger in global_loggers {
886 logger
887 .send(LogMessage::new(
888 "Cannot determine server configuration: No matching configuration found".to_string(),
889 true,
890 ))
891 .await
892 .unwrap_or_default()
893 }
894 let response = basic_error_response(StatusCode::INTERNAL_SERVER_ERROR);
895 let (request_parts, _) = request.into_parts();
896 return Ok(
897 finalize_basic_error_response(
898 response,
899 request_parts,
900 http3_alt_port,
901 global_loggers,
902 &socket_data,
903 global_log_date_format,
904 global_log_format,
905 global_log_json.as_ref(),
906 )
907 .await,
908 );
909 }
910 };
911
912 let mut log_date_format = get_value!("log_date_format", configuration).and_then(|v| v.as_str());
914 let mut log_format = get_value!("log_format", configuration).and_then(|v| v.as_str());
915 let mut log_json_props = configuration
916 .entries
917 .get("log_json")
918 .and_then(|entries| entries.get_entry())
919 .map(|entry| entry.props.clone());
920
921 if !configuration.observability.log_channels.is_empty() && log_request_parts.is_none() {
923 let (request_parts, request_body) = request.into_parts();
924 log_request_parts = Some(request_parts.clone());
925 request = Request::from_parts(request_parts, request_body);
926 }
927
928 if !get_value!("disable_url_sanitizer", configuration)
930 .and_then(|v| v.as_bool())
931 .unwrap_or(false)
932 {
933 request = match sanitize_url_request(&configuration, request) {
934 Ok((mut request, was_dirty)) => {
935 if was_dirty {
936 let (parts, body) = request.into_parts();
937 let configuration_option =
938 configurations.find_configuration(&parts, hostname_determinant.as_deref(), &socket_data);
939 request = Request::from_parts(parts, body);
940 match configuration_option {
941 Ok(Some(new_configuration)) => {
942 if let Some(new_config2) = new_configuration.get_default().cloned() {
943 configuration_error_handler_lookup = new_configuration;
944 configuration = new_config2;
945 log_date_format = get_value!("log_date_format", configuration).and_then(|v| v.as_str());
946 log_format = get_value!("log_format", configuration).and_then(|v| v.as_str());
947 log_json_props = configuration
948 .entries
949 .get("log_json")
950 .and_then(|entries| entries.get_entry())
951 .map(|entry| entry.props.clone());
952 }
953 }
954 Ok(None) => {}
955 Err(err) => {
956 for logger in &configuration.observability.log_channels {
957 logger
958 .send(LogMessage::new(
959 format!("Cannot determine server configuration: {err}"),
960 true,
961 ))
962 .await
963 .unwrap_or_default();
964 }
965 let response = generate_error_response(StatusCode::BAD_REQUEST, &configuration, &None).await;
966
967 let (request_parts, _) = request.into_parts();
968 let (headers_to_add, headers_to_replace, headers_to_remove) =
969 build_custom_headers(&configuration, &request_parts);
970
971 return Ok(
972 finalize_response_and_log(
973 response,
974 http3_alt_port,
975 headers_to_add,
976 headers_to_replace,
977 headers_to_remove,
978 &configuration.observability.log_channels,
979 &log_request_parts,
980 &socket_data,
981 None,
982 log_date_format,
983 log_format,
984 log_json_props.as_ref(),
985 )
986 .await,
987 );
988 }
989 }
990 }
991 request
992 }
993 Err((request, error)) => {
994 for logger in &configuration.observability.log_channels {
995 logger
996 .send(LogMessage::new(format!("URL sanitation error: {error}"), true))
997 .await
998 .unwrap_or_default();
999 }
1000
1001 let response = generate_error_response(StatusCode::BAD_REQUEST, &configuration, &None).await;
1002
1003 let (parts, _) = request.into_parts();
1004 let (headers_to_add, headers_to_replace, headers_to_remove) = build_custom_headers(&configuration, &parts);
1005
1006 return Ok(
1007 finalize_response_and_log(
1008 response,
1009 http3_alt_port,
1010 headers_to_add,
1011 headers_to_replace,
1012 headers_to_remove,
1013 &configuration.observability.log_channels,
1014 &log_request_parts,
1015 &socket_data,
1016 None,
1017 log_date_format,
1018 log_format,
1019 log_json_props.as_ref(),
1020 )
1021 .await,
1022 );
1023 }
1024 };
1025 }
1026
1027 let (request_parts, request_body) = request.into_parts();
1028 let (headers_to_add, headers_to_replace, headers_to_remove) = build_custom_headers(&configuration, &request_parts);
1029 let mut request = Request::from_parts(request_parts, request_body);
1030
1031 if request.uri().path() == "*" {
1032 let response = match request.method() {
1033 &Method::OPTIONS => Response::builder()
1034 .status(StatusCode::NO_CONTENT)
1035 .header(header::ALLOW, HeaderValue::from_static("GET, POST, HEAD, OPTIONS"))
1036 .body(Empty::new().map_err(|e| match e {}).boxed())
1037 .unwrap_or_default(),
1038 _ => {
1039 let mut header_map = HeaderMap::new();
1040 header_map.insert(header::ALLOW, HeaderValue::from_static("GET, POST, HEAD, OPTIONS"));
1041 generate_error_response(StatusCode::BAD_REQUEST, &configuration, &Some(header_map)).await
1042 }
1043 };
1044 return Ok(
1045 finalize_response_and_log(
1046 response,
1047 http3_alt_port,
1048 headers_to_add,
1049 headers_to_replace,
1050 headers_to_remove,
1051 &configuration.observability.log_channels,
1052 &log_request_parts,
1053 &socket_data,
1054 None,
1055 log_date_format,
1056 log_format,
1057 log_json_props.as_ref(),
1058 )
1059 .await,
1060 );
1061 }
1062
1063 let acme_http_01_resolvers_inner = acme_http_01_resolvers.read().await;
1065 if !acme_http_01_resolvers_inner.is_empty() {
1066 if let Some(challenge_token) = request.uri().path().strip_prefix("/.well-known/acme-challenge/") {
1067 for acme_http01_resolver in &*acme_http_01_resolvers_inner {
1068 if let Some(http01_acme_data) = &*acme_http01_resolver.read().await {
1069 let acme_response = http01_acme_data.1.clone();
1070 if challenge_token == http01_acme_data.0 {
1071 let response = Response::builder()
1072 .status(StatusCode::OK)
1073 .header(
1074 header::CONTENT_TYPE,
1075 HeaderValue::from_static("application/octet-stream"),
1076 )
1077 .body(Full::new(Bytes::from(acme_response)).map_err(|e| match e {}).boxed())
1078 .unwrap_or_default();
1079
1080 return Ok(
1081 finalize_response_and_log(
1082 response,
1083 http3_alt_port,
1084 headers_to_add,
1085 headers_to_replace,
1086 headers_to_remove,
1087 &configuration.observability.log_channels,
1088 &log_request_parts,
1089 &socket_data,
1090 None,
1091 log_date_format,
1092 log_format,
1093 log_json_props.as_ref(),
1094 )
1095 .await,
1096 );
1097 }
1098 }
1099 }
1100 }
1101 };
1102 drop(acme_http_01_resolvers_inner);
1103
1104 let mut error_logger = if !configuration.observability.log_channels.is_empty() {
1105 ErrorLogger::new_multiple(configuration.observability.log_channels.clone())
1106 } else {
1107 ErrorLogger::without_logger()
1108 };
1109 let mut metrics_enabled = !configuration.observability.metric_channels.is_empty();
1110 let mut metrics_sender = if metrics_enabled {
1111 MetricsMultiSender::new_multiple(configuration.observability.metric_channels.clone())
1112 } else {
1113 MetricsMultiSender::without_sender()
1114 };
1115 let mut traces_enabled = !configuration.observability.trace_channels.is_empty();
1116 let mut traces_senders = if traces_enabled {
1117 let mut traces_senders = Vec::with_capacity(configuration.observability.trace_channels.len());
1118 for channel in &configuration.observability.trace_channels {
1119 channel.0.send(()).await.unwrap_or_default();
1120 if let Ok(channel2) = channel.1.recv().await {
1121 traces_senders.push(channel2);
1122 }
1123 }
1124 traces_senders
1125 } else {
1126 vec![]
1127 };
1128
1129 let mut module_handlers = Vec::with_capacity(configuration.modules.len());
1131 for module in &configuration.modules {
1132 module_handlers.push(module.get_module_handlers());
1133 }
1134
1135 request.extensions_mut().insert(RequestData {
1137 auth_user: None,
1138 original_url: None,
1139 error_status_code: None,
1140 });
1141 let mut executed_handlers = Vec::new();
1142 let (request_parts, request_body) = request.into_parts();
1143 let request_parts_cloned = if configuration_error_handler_lookup.has_status_codes() {
1144 let mut request_parts_cloned = request_parts.clone();
1145 request_parts_cloned
1146 .headers
1147 .insert(header::CONTENT_LENGTH, HeaderValue::from_static("0"));
1148 Some(request_parts_cloned)
1149 } else {
1150 None
1152 };
1153 let mut request = Request::from_parts(request_parts, request_body);
1154 let mut latest_auth_data = None;
1155 let mut is_error_handler = false;
1156 let mut handlers_iter: Box<dyn Iterator<Item = Box<dyn ModuleHandlers>>> = Box::new(module_handlers.into_iter());
1157 while let Some(mut handlers) = handlers_iter.next() {
1158 if metrics_enabled {
1159 handlers
1160 .metric_data_before_handler(&request, &socket_data, &metrics_sender)
1161 .await;
1162 }
1163
1164 if traces_enabled {
1165 for trace_sender in &traces_senders {
1166 trace_sender
1167 .send(TraceSignal::StartSpan(format!(
1168 "{}::request_handler",
1169 handlers.get_name()
1170 )))
1171 .await
1172 .unwrap_or_default();
1173 }
1174 }
1175
1176 let (response_result, is_timeout) = if let Some(timeout_duration) = &timeout_duration {
1177 let elapsed = timeout_instant.elapsed();
1178 if let Some(timeout_cur_duration) = timeout_duration.checked_sub(elapsed) {
1179 match timeout(
1180 timeout_cur_duration,
1181 handlers.request_handler(request, &configuration, &socket_data, &error_logger),
1182 )
1183 .await
1184 {
1185 Ok(result) => (result, false),
1186 Err(_) => (Err(anyhow::anyhow!("The client or server has timed out").into()), true),
1187 }
1188 } else {
1189 (Err(anyhow::anyhow!("The client or server has timed out").into()), true)
1190 }
1191 } else {
1192 (
1193 handlers
1194 .request_handler(request, &configuration, &socket_data, &error_logger)
1195 .await,
1196 false,
1197 )
1198 };
1199
1200 if traces_enabled {
1201 for trace_sender in &traces_senders {
1202 trace_sender
1203 .send(TraceSignal::EndSpan(
1204 format!("{}::request_handler", handlers.get_name()),
1205 response_result.as_ref().err().map(|e| e.to_string()),
1206 ))
1207 .await
1208 .unwrap_or_default();
1209 }
1210 }
1211
1212 executed_handlers.push(handlers);
1213
1214 if is_timeout {
1215 if metrics_enabled {
1216 while let Some(mut executed_handler) = executed_handlers.pop() {
1217 executed_handler.metric_data_after_handler(&metrics_sender).await;
1218 }
1219 }
1220 Err(anyhow::anyhow!("The client or server has timed out"))?;
1221 }
1222
1223 match response_result {
1224 Ok(response) => {
1225 let status = response.response_status;
1226 let headers = response.response_headers;
1227 let new_remote_address = response.new_remote_address;
1228 let request_option = response.request;
1229 let response = response.response;
1230 let request_extensions = request_option
1231 .as_ref()
1232 .and_then(|r| r.extensions().get::<RequestData>());
1233 if let Some(request_extensions) = request_extensions {
1234 latest_auth_data = request_extensions.auth_user.clone();
1235 }
1236 if let Some(new_remote_address) = new_remote_address {
1237 socket_data.remote_addr = new_remote_address;
1238 };
1239
1240 match response {
1241 Some(response) => {
1242 return finalize_with_modifying_handlers(
1243 response,
1244 executed_handlers,
1245 &configuration,
1246 http3_alt_port,
1247 headers_to_add,
1248 headers_to_replace,
1249 headers_to_remove,
1250 &log_request_parts,
1251 &socket_data,
1252 latest_auth_data.as_deref(),
1253 log_date_format,
1254 log_format,
1255 log_json_props.as_ref(),
1256 metrics_sender,
1257 metrics_enabled,
1258 traces_senders,
1259 traces_enabled,
1260 timeout_instant,
1261 timeout_duration,
1262 )
1263 .await;
1264 }
1265 None => match status {
1266 Some(status) => {
1267 if !is_error_handler {
1268 if let Some(error_configuration) = configuration_error_handler_lookup.get(status.as_u16()).cloned() {
1269 let request_option = if let Some(request) = request_option {
1270 Some(request)
1271 } else {
1272 request_parts_cloned.clone().map(|request_parts_cloned| {
1273 Request::from_parts(request_parts_cloned, Empty::new().map_err(|e| match e {}).boxed())
1274 })
1275 };
1276 if let Some(request_cloned) = request_option {
1277 configuration = error_configuration;
1278 let mut module_handlers = Vec::with_capacity(configuration.modules.len());
1279 for module in &configuration.modules {
1280 module_handlers.push(module.get_module_handlers());
1281 }
1282 handlers_iter = Box::new(module_handlers.into_iter());
1283 if metrics_enabled {
1284 while let Some(mut executed_handler) = executed_handlers.pop() {
1285 executed_handler.metric_data_after_handler(&metrics_sender).await;
1286 }
1287 }
1288 executed_handlers = Vec::new();
1289 request = request_cloned;
1290 if let Some(request_data) = request.extensions_mut().get_mut::<RequestData>() {
1291 request_data.error_status_code = Some(status);
1292 }
1293 is_error_handler = true;
1294 log_date_format = get_value!("log_date_format", configuration).and_then(|v| v.as_str());
1295 log_format = get_value!("log_format", configuration).and_then(|v| v.as_str());
1296 log_json_props = configuration
1297 .entries
1298 .get("log_json")
1299 .and_then(|entries| entries.get_entry())
1300 .map(|entry| entry.props.clone());
1301 error_logger = if !configuration.observability.log_channels.is_empty() {
1302 ErrorLogger::new_multiple(configuration.observability.log_channels.clone())
1303 } else {
1304 ErrorLogger::without_logger()
1305 };
1306 metrics_enabled = !configuration.observability.metric_channels.is_empty();
1307 metrics_sender = if metrics_enabled {
1308 MetricsMultiSender::new_multiple(configuration.observability.metric_channels.clone())
1309 } else {
1310 MetricsMultiSender::without_sender()
1311 };
1312 traces_enabled = !configuration.observability.trace_channels.is_empty();
1313 traces_senders = if traces_enabled {
1314 let mut traces_senders = Vec::with_capacity(configuration.observability.trace_channels.len());
1315 for channel in &configuration.observability.trace_channels {
1316 channel.0.send(()).await.unwrap_or_default();
1317 if let Ok(channel2) = channel.1.recv().await {
1318 traces_senders.push(channel2);
1319 }
1320 }
1321 traces_senders
1322 } else {
1323 vec![]
1324 };
1325 continue;
1326 }
1327 }
1328 }
1329 let response = generate_error_response(status, &configuration, &headers).await;
1330 return finalize_with_modifying_handlers(
1331 response,
1332 executed_handlers,
1333 &configuration,
1334 http3_alt_port,
1335 headers_to_add,
1336 headers_to_replace,
1337 headers_to_remove,
1338 &log_request_parts,
1339 &socket_data,
1340 latest_auth_data.as_deref(),
1341 log_date_format,
1342 log_format,
1343 log_json_props.as_ref(),
1344 metrics_sender,
1345 metrics_enabled,
1346 traces_senders,
1347 traces_enabled,
1348 timeout_instant,
1349 timeout_duration,
1350 )
1351 .await;
1352 }
1353 None => match request_option {
1354 Some(request_obtained) => {
1355 request = request_obtained;
1356 continue;
1357 }
1358 None => {
1359 break;
1360 }
1361 },
1362 },
1363 }
1364 }
1365 Err(err) => {
1366 let response = generate_error_response(StatusCode::INTERNAL_SERVER_ERROR, &configuration, &None).await;
1367 let err_string = err.to_string();
1368
1369 if !err_string.is_empty() {
1370 for logger in &configuration.observability.log_channels {
1371 logger
1372 .send(LogMessage::new(
1373 format!("Unexpected error while serving a request: {err}"),
1374 true,
1375 ))
1376 .await
1377 .unwrap_or_default();
1378 }
1379 }
1380
1381 let response_result = finalize_with_modifying_handlers(
1382 response,
1383 executed_handlers,
1384 &configuration,
1385 http3_alt_port,
1386 headers_to_add,
1387 headers_to_replace,
1388 headers_to_remove,
1389 &log_request_parts,
1390 &socket_data,
1391 latest_auth_data.as_deref(),
1392 log_date_format,
1393 log_format,
1394 log_json_props.as_ref(),
1395 metrics_sender,
1396 metrics_enabled,
1397 traces_senders,
1398 traces_enabled,
1399 timeout_instant,
1400 timeout_duration,
1401 )
1402 .await;
1403
1404 if err_string.is_empty() {
1405 return Err(anyhow::anyhow!("HTTP request aborted"));
1407 }
1408
1409 return response_result;
1410 }
1411 }
1412 }
1413
1414 let response = generate_error_response(StatusCode::NOT_FOUND, &configuration, &None).await;
1415
1416 finalize_with_modifying_handlers(
1417 response,
1418 executed_handlers,
1419 &configuration,
1420 http3_alt_port,
1421 headers_to_add,
1422 headers_to_replace,
1423 headers_to_remove,
1424 &log_request_parts,
1425 &socket_data,
1426 latest_auth_data.as_deref(),
1427 log_date_format,
1428 log_format,
1429 log_json_props.as_ref(),
1430 metrics_sender,
1431 metrics_enabled,
1432 traces_senders,
1433 traces_enabled,
1434 timeout_instant,
1435 timeout_duration,
1436 )
1437 .await
1438}