1use apimock_config::Config;
17use apimock_config::config::constant::{
18 SERVICE_DEFAULT_MAX_REQUEST_BODY_BYTES, SERVICE_DEFAULT_MIDDLEWARE_MAX_OPERATIONS,
19 TLS_DEFAULT_HANDSHAKE_TIMEOUT_SECONDS, TLS_DEFAULT_MAX_CONNECTIONS,
20};
21use apimock_routing::ParsedRequest;
22use console::style;
23use http_body_util::{BodyExt, Empty};
24use hyper::{
25 HeaderMap, Response, body,
26 header::{CONTENT_LENGTH, HeaderValue},
27 service::service_fn,
28};
29use hyper_util::{
30 rt::{TokioExecutor, TokioIo},
31 server::conn::auto::Builder,
32};
33use tokio::net::TcpListener;
34use tokio::sync::Semaphore;
35use tokio_rustls::TlsAcceptor;
36
37use std::net::{SocketAddr, ToSocketAddrs};
38use std::path::PathBuf;
39use std::sync::Arc;
40
41use crate::{
42 dyn_route::dyn_route_content_traced,
43 error::{ServerError, ServerResult},
44 middleware::LoadedMiddlewares,
45 parsed_request::{ParsedRequestError, capture_in_log_with_trace_config, parsed_request_from},
46 respond_response::respond_response,
47 response::{
48 confine::canonical_dir,
49 error_response::{internal_server_error_response, payload_too_large_response},
50 },
51 response_handler::default_response_headers,
52 tls::{build_server_config_reloadable, load_certs, load_private_key},
53 types::BoxBody,
54};
55
56pub use crate::control::{ReloadHint, ServerControl, ServerHandle, ServerState};
57use crate::trace::{Outcome, RequestSummary, TraceEmitter};
58
59#[non_exhaustive]
73pub struct AppState {
74 pub config: Config,
75 pub middlewares: LoadedMiddlewares,
76 pub tracer: TraceEmitter,
78 canonical_fallback_respond_dir: Option<PathBuf>,
82 canonical_rule_set_respond_dirs: Vec<Option<PathBuf>>,
84}
85
86impl AppState {
87 pub fn new(config: Config, middlewares: LoadedMiddlewares, tracer: TraceEmitter) -> Self {
88 let canonical_fallback_respond_dir =
89 canonical_dir(config.service.fallback_respond_dir.as_str());
90 let canonical_rule_set_respond_dirs = config
91 .service
92 .rule_sets
93 .iter()
94 .map(|rule_set| canonical_dir(rule_set.dir_prefix().as_str()))
95 .collect();
96 Self {
97 config,
98 middlewares,
99 tracer,
100 canonical_fallback_respond_dir,
101 canonical_rule_set_respond_dirs,
102 }
103 }
104}
105
106#[non_exhaustive]
108pub struct Server {
109 pub app_state: Arc<AppState>,
114 pub http_addr: Option<SocketAddr>,
115 pub https_addr: Option<SocketAddr>,
116 https_tls: Option<HttpsTls>,
121}
122
123#[derive(Clone)]
126struct HttpsTls {
127 acceptor: TlsAcceptor,
128 handshake_timeout: std::time::Duration,
129 max_connections: usize,
130}
131
132impl Server {
133 pub async fn new(config: Config) -> ServerResult<Self> {
156 let http_addr = resolve_listener(config.listener_http_addr().as_deref())?;
157 let https_addr = resolve_listener(config.listener_https_addr().as_deref())?;
158
159 let https_tls = match https_addr {
160 Some(addr) => Some(build_https_tls(&config, addr)?),
161 None => None,
162 };
163
164 let relative_dir_path = config
166 .current_dir_to_parent_dir_relative_path()
167 .map_err(ServerError::Config)?;
168
169 let middleware_max_operations = config
170 .service
171 .middleware_max_operations
172 .unwrap_or(SERVICE_DEFAULT_MIDDLEWARE_MAX_OPERATIONS);
173 let middlewares = LoadedMiddlewares::compile(
174 config
175 .service
176 .middlewares_file_paths
177 .as_deref()
178 .unwrap_or(&[]),
179 relative_dir_path.as_str(),
180 middleware_max_operations,
181 )?;
182 if !middlewares.is_empty() {
183 log::info!("middleware is activated: {} file(s)", middlewares.len());
184 }
185
186 Ok(Server {
187 http_addr,
188 https_addr,
189 https_tls,
190 app_state: Arc::new(AppState::new(config, middlewares, TraceEmitter::new())),
191 })
192 }
193
194 pub async fn start(&self) {
196 let http = self.http_start();
197 let https = self.https_start();
198 tokio::join!(http, https);
199 }
200
201 pub async fn bind_http(&self) -> ServerResult<Option<TcpListener>> {
216 let Some(addr) = self.http_addr else {
217 return Ok(None);
218 };
219
220 let listener =
221 TcpListener::bind(addr)
222 .await
223 .map_err(|err| ServerError::ListenerAddress {
224 addr: addr.to_string(),
225 reason: err.to_string(),
226 })?;
227
228 Ok(Some(listener))
229 }
230
231 pub async fn serve_http(&self, listener: TcpListener) {
233 if let Ok(addr) = listener.local_addr() {
234 log::info!(
235 "Greetings from apimock-rs (API Mock) !!\nListening on {} ...\n",
236 style(format!("http://{}", addr)).cyan()
237 );
238 }
239
240 let app_state = Arc::clone(&self.app_state);
241 loop {
242 let (stream, _) = match listener.accept().await {
243 Ok(pair) => pair,
244 Err(err) => {
245 log::error!("HTTP accept failed: {}", err);
246 continue;
247 }
248 };
249 let io = TokioIo::new(stream);
250
251 let app_state = Arc::clone(&app_state);
252 tokio::task::spawn(async move {
253 if let Err(err) = Builder::new(TokioExecutor::new())
254 .serve_connection(
255 io,
256 service_fn(move |request: hyper::Request<body::Incoming>| {
257 service(request, app_state.clone())
258 }),
259 )
260 .await
261 {
262 log::error!("{} to build connection: {:?}", style("failed").red(), err);
263 }
264 });
265 }
266 }
267
268 async fn http_start(&self) {
269 match self.bind_http().await {
270 Ok(Some(listener)) => self.serve_http(listener).await,
271 Ok(None) => (),
272 Err(err) => log::error!("{}", err),
273 }
274 }
275
276 pub async fn bind_https(&self) -> ServerResult<Option<(TcpListener, TlsAcceptor)>> {
283 let (Some(addr), Some(https_tls)) = (self.https_addr, self.https_tls.as_ref()) else {
284 return Ok(None);
285 };
286
287 let listener =
288 TcpListener::bind(addr)
289 .await
290 .map_err(|err| ServerError::ListenerAddress {
291 addr: addr.to_string(),
292 reason: err.to_string(),
293 })?;
294
295 Ok(Some((listener, https_tls.acceptor.clone())))
296 }
297
298 pub async fn serve_https(&self, listener: TcpListener, acceptor: TlsAcceptor) {
313 if let Ok(addr) = listener.local_addr() {
314 log::info!(
315 "Greetings from apimock-rs (API Mock) !!\nListening on {} ...\n",
316 style(format!("https://{}", addr)).cyan()
317 );
318 }
319
320 let (handshake_timeout, max_connections) = self
321 .https_tls
322 .as_ref()
323 .map(|t| (t.handshake_timeout, t.max_connections))
324 .unwrap_or((
325 std::time::Duration::from_secs(TLS_DEFAULT_HANDSHAKE_TIMEOUT_SECONDS),
326 TLS_DEFAULT_MAX_CONNECTIONS,
327 ));
328 let connection_slots = Arc::new(Semaphore::new(max_connections));
329
330 let app_state = Arc::clone(&self.app_state);
331 loop {
332 let (stream, _) = match listener.accept().await {
333 Ok(pair) => pair,
334 Err(err) => {
335 log::error!("HTTPS accept failed: {}", err);
336 continue;
337 }
338 };
339 let Ok(permit) = Arc::clone(&connection_slots).acquire_owned().await else {
340 continue;
344 };
345 let acceptor = acceptor.clone();
346 let app_state = Arc::clone(&app_state);
347
348 tokio::spawn(async move {
349 let _permit = permit; let tls_stream =
351 match tokio::time::timeout(handshake_timeout, acceptor.accept(stream)).await {
352 Ok(Ok(s)) => s,
353 Ok(Err(e)) => {
354 log::error!("TLS handshake failed: {:?}", e);
355 return;
356 }
357 Err(_elapsed) => {
358 log::error!(
359 "TLS handshake timed out after {:?}; dropping connection",
360 handshake_timeout
361 );
362 return;
363 }
364 };
365 let io = TokioIo::new(tls_stream);
366 let app_state = app_state.clone();
367 tokio::task::spawn(async move {
368 if let Err(err) = Builder::new(TokioExecutor::new())
369 .serve_connection(
370 io,
371 service_fn(move |request: hyper::Request<body::Incoming>| {
372 service(request, app_state.clone())
373 }),
374 )
375 .await
376 {
377 log::error!("{} to build connection: {:?}", style("failed").red(), err);
378 }
379 });
380 });
381 }
382 }
383
384 async fn https_start(&self) {
385 match self.bind_https().await {
386 Ok(Some((listener, acceptor))) => self.serve_https(listener, acceptor).await,
387 Ok(None) => (),
388 Err(err) => log::error!("{}", err),
389 }
390 }
391}
392
393fn build_https_tls(config: &Config, addr: SocketAddr) -> ServerResult<HttpsTls> {
398 let tls = config
399 .listener
400 .as_ref()
401 .and_then(|l| l.tls.as_ref())
402 .cloned()
403 .ok_or_else(|| ServerError::ListenerAddress {
404 addr: addr.to_string(),
405 reason: "internal: HTTPS listener scheduled without TLS config".to_owned(),
406 })?;
407
408 let certs = load_certs(tls.cert.as_str())?;
409 let key = load_private_key(tls.key.as_str())?;
410
411 let (tls_config, resolver) =
414 build_server_config_reloadable(certs, key).map_err(|err| ServerError::ListenerAddress {
415 addr: addr.to_string(),
416 reason: format!("failed to build TLS config: {}", err),
417 })?;
418 let acceptor = TlsAcceptor::from(Arc::new(tls_config));
419 drop(resolver); let handshake_timeout = std::time::Duration::from_secs(
422 tls.handshake_timeout_seconds
423 .unwrap_or(TLS_DEFAULT_HANDSHAKE_TIMEOUT_SECONDS),
424 );
425 let max_connections = tls.max_connections.unwrap_or(TLS_DEFAULT_MAX_CONNECTIONS);
426
427 Ok(HttpsTls {
428 acceptor,
429 handshake_timeout,
430 max_connections,
431 })
432}
433
434fn resolve_listener(addr_str: Option<&str>) -> ServerResult<Option<SocketAddr>> {
436 let Some(addr_str) = addr_str else {
437 return Ok(None);
438 };
439
440 let mut addrs = addr_str
441 .to_socket_addrs()
442 .map_err(|e| ServerError::ListenerAddress {
443 addr: addr_str.to_owned(),
444 reason: e.to_string(),
445 })?;
446
447 addrs
448 .next()
449 .map(Some)
450 .ok_or_else(|| ServerError::ListenerAddress {
451 addr: addr_str.to_owned(),
452 reason: "address resolved to no socket addresses".to_owned(),
453 })
454}
455
456pub async fn service(
463 request: hyper::Request<body::Incoming>,
464 app_state: Arc<AppState>,
465) -> Result<hyper::Response<BoxBody>, hyper::http::Error> {
466 let request_headers = request.headers().clone();
467
468 let config = &app_state.config;
475 let middlewares = &app_state.middlewares;
476 let tracer = &app_state.tracer;
477
478 let cors_allow_credentials_origins = config
479 .service
480 .cors_allow_credentials_origins
481 .clone()
482 .unwrap_or_default();
483
484 if request.method() == hyper::Method::OPTIONS {
485 return handle_options(&request_headers, &cors_allow_credentials_origins);
486 }
487
488 let max_request_body_bytes = config
489 .service
490 .max_request_body_bytes
491 .unwrap_or(SERVICE_DEFAULT_MAX_REQUEST_BODY_BYTES);
492 let max_request_body_bytes = usize::try_from(max_request_body_bytes).unwrap_or(usize::MAX);
493
494 let parsed_request = match parsed_request_from(request, max_request_body_bytes).await {
495 Ok(x) => x,
496 Err(ParsedRequestError::BodyTooLarge) => {
497 return payload_too_large_response(
498 &format!(
499 "request body exceeds the configured limit ({} bytes)",
500 max_request_body_bytes
501 ),
502 &request_headers,
503 &cors_allow_credentials_origins,
504 );
505 }
506 Err(ParsedRequestError::Other(err)) => {
507 return internal_server_error_response(
508 err.as_str(),
509 &request_headers,
510 &cors_allow_credentials_origins,
511 );
512 }
513 };
514
515 let received_at_ms = std::time::SystemTime::now()
516 .duration_since(std::time::UNIX_EPOCH)
517 .unwrap_or_default()
518 .as_millis() as u64;
519 let start = std::time::Instant::now();
520
521 capture_in_log_with_trace_config(
522 &parsed_request,
523 config.log.clone().unwrap_or_default().verbose,
524 &tracer.config,
525 );
526
527 let trace_summary = if tracer.has_subscribers() {
533 let headers = parsed_request
534 .component_parts
535 .headers
536 .iter()
537 .filter_map(|(k, v)| v.to_str().ok().map(|v| (k.to_string(), v.to_owned())))
538 .collect();
539 let mut summary = RequestSummary::new(
540 parsed_request.component_parts.method.to_string(),
541 parsed_request.url_path.clone(),
542 headers,
543 parsed_request.body_len,
544 &tracer.config,
545 );
546 tracer.enrich_with_body(&mut summary, parsed_request.body_json.as_ref());
547 Some(summary)
548 } else {
549 None
550 };
551
552 if let Some((middleware_file_path, response)) = middleware_response(
553 middlewares,
554 &parsed_request,
555 &cors_allow_credentials_origins,
556 )
557 .await
558 {
559 let status = response_status_or(&response, 500);
560 emit_trace_event(
561 tracer,
562 received_at_ms,
563 start,
564 trace_summary,
565 Outcome::Middleware {
566 file_path: middleware_file_path,
567 status,
568 },
569 );
570 return response;
571 }
572
573 if let Some((rule_set_idx, rule_idx, response)) = rule_set_response(
574 config,
575 &parsed_request,
576 app_state.canonical_rule_set_respond_dirs.as_slice(),
577 &cors_allow_credentials_origins,
578 )
579 .await
580 {
581 emit_trace_event(
582 tracer,
583 received_at_ms,
584 start,
585 trace_summary,
586 Outcome::Matched {
587 rule_set_index: rule_set_idx,
588 rule_index: rule_idx,
589 },
590 );
591 return response;
592 }
593
594 let (dyn_route_response, resolved_file_path) = dyn_route_content_traced(
595 parsed_request.url_path.as_str(),
596 config.service.fallback_respond_dir.as_str(),
597 &request_headers,
598 app_state.canonical_fallback_respond_dir.as_deref(),
599 &cors_allow_credentials_origins,
600 )
601 .await;
602
603 let status = response_status_or(&dyn_route_response, 500);
604 let outcome = match resolved_file_path {
610 Some(file_path) if status != 404 => Outcome::Fallback { file_path, status },
611 _ => Outcome::Miss { status },
612 };
613 emit_trace_event(tracer, received_at_ms, start, trace_summary, outcome);
614
615 dyn_route_response
616}
617
618fn response_status_or(
624 response: &Result<hyper::Response<BoxBody>, hyper::http::Error>,
625 fallback: u16,
626) -> u16 {
627 response
628 .as_ref()
629 .map(|r| r.status().as_u16())
630 .unwrap_or(fallback)
631}
632
633fn emit_trace_event(
638 tracer: &TraceEmitter,
639 received_at_ms: u64,
640 start: std::time::Instant,
641 trace_summary: Option<RequestSummary>,
642 outcome: Outcome,
643) {
644 if let Some(summary) = trace_summary {
645 tracer.emit(
646 received_at_ms,
647 start.elapsed().as_millis() as u32,
648 summary,
649 outcome,
650 );
651 }
652}
653
654async fn middleware_response(
662 middlewares: &LoadedMiddlewares,
663 parsed_request: &ParsedRequest,
664 cors_allow_credentials_origins: &[String],
665) -> Option<(String, Result<hyper::Response<BoxBody>, hyper::http::Error>)> {
666 for handler in middlewares.iter() {
667 match handler
668 .handle(
669 parsed_request.url_path.as_str(),
670 parsed_request.body_json.as_ref(),
671 &parsed_request.component_parts.headers,
672 cors_allow_credentials_origins,
673 )
674 .await
675 {
676 Some(x) => return Some((handler.file_path.clone(), x)),
677 None => continue,
678 }
679 }
680 None
681}
682
683async fn rule_set_response(
693 config: &Config,
694 parsed_request: &ParsedRequest,
695 canonical_rule_set_respond_dirs: &[Option<PathBuf>],
696 cors_allow_credentials_origins: &[String],
697) -> Option<(
698 usize,
699 usize,
700 Result<hyper::Response<BoxBody>, hyper::http::Error>,
701)> {
702 for (rule_set_idx, rule_set) in config.service.rule_sets.iter().enumerate() {
703 if let Some((rule_idx, respond)) = rule_set.find_matched(
704 parsed_request,
705 config.service.strategy.as_ref(),
706 rule_set_idx,
707 ) {
708 let dir_prefix = rule_set.dir_prefix();
709 let rule_set_default_delay_ms = rule_set
710 .default
711 .as_ref()
712 .and_then(|default| default.delay_response_milliseconds);
713 let confine_to = canonical_rule_set_respond_dirs
714 .get(rule_set_idx)
715 .and_then(|dir| dir.as_deref());
716 return Some((
717 rule_set_idx,
718 rule_idx,
719 respond_response(
720 &respond,
721 dir_prefix.as_str(),
722 parsed_request,
723 rule_set_default_delay_ms,
724 confine_to,
725 cors_allow_credentials_origins,
726 )
727 .await,
728 ));
729 }
730 }
731 None
732}
733
734pub fn handle_options(
738 request_headers: &HeaderMap,
739 cors_allow_credentials_origins: &[String],
740) -> Result<hyper::Response<BoxBody>, hyper::http::Error> {
741 let mut response = Response::new(Empty::new().boxed());
742 *response.status_mut() = hyper::StatusCode::NO_CONTENT;
743 response
744 .headers_mut()
745 .insert(CONTENT_LENGTH, HeaderValue::from_static("0"));
746
747 for (header_key, header_value) in
748 default_response_headers(request_headers, cors_allow_credentials_origins).into_iter()
749 {
750 if let Some(header_key) = header_key {
751 response.headers_mut().insert(header_key, header_value);
752 }
753 }
754
755 Ok(response)
756}