1use crate::dep::{AnyArc, DepEnv, DepFactory, DepResolver, TaskContext};
5use crate::error::{Error, Result};
6use crate::extract::{BodyLane, RequestCtx};
7use crate::handler::BoxHandlerFn;
8use crate::middleware::{Middleware, Next};
9use crate::module::{FlatRoute, Module};
10use crate::response::{IntoResponse, Response};
11use crate::router::{Endpoint, MethodRouter, RouteMatch, Trie};
12use crate::serve;
13#[cfg(test)]
14use crate::serve::is_transient_accept_error;
15#[cfg(test)]
16use bytes::Bytes;
17use std::any::TypeId;
18use std::collections::HashMap;
19use std::future::Future;
20use std::pin::Pin;
21use std::sync::Arc;
22
23pub(crate) type BackgroundFactory = Box<
27 dyn FnOnce(
28 TaskContext,
29 tokio::sync::watch::Receiver<bool>,
30 ) -> Pin<Box<dyn Future<Output = ()> + Send>>
31 + Send,
32>;
33
34pub type ErrorBodyMapper =
42 dyn Fn(http::StatusCode, &str, &str) -> Option<serde_json::Value> + Send + Sync;
43
44pub struct App {
47 routes: Vec<(String, MethodRouter)>,
48 mounts: Vec<(String, Module)>,
49 env: DepEnv,
50 middleware: Vec<Arc<dyn Middleware>>,
51 security_headers: bool,
52 cors: Option<std::sync::Arc<crate::cors::CorsConfig>>,
53 handler_timeout: std::time::Duration,
54 body_read_timeout: std::time::Duration,
55 write_stall_timeout: std::time::Duration,
56 error_mapper: Option<Arc<ErrorBodyMapper>>,
59 background: Vec<(&'static str, BackgroundFactory)>,
62}
63
64impl Default for App {
65 fn default() -> Self {
66 let mut env = DepEnv::default();
71 env.insert_value(crate::clock::Clock::system());
72 Self {
73 routes: Vec::new(),
74 mounts: Vec::new(),
75 env,
76 middleware: Vec::new(),
77 security_headers: true,
78 cors: None,
79 handler_timeout: std::time::Duration::from_secs(30),
80 body_read_timeout: std::time::Duration::from_secs(30),
81 write_stall_timeout: std::time::Duration::from_secs(30),
82 error_mapper: None,
83 background: Vec::new(),
84 }
85 }
86}
87
88pub trait Extension {
91 fn register(self, app: App) -> App;
92}
93
94impl App {
95 pub fn new() -> Self {
96 Self::default()
97 }
98
99 pub fn extend<E: Extension>(self, extension: E) -> App {
101 extension.register(self)
102 }
103
104 pub fn security_headers(mut self, on: bool) -> Self {
107 self.security_headers = on;
108 self
109 }
110
111 pub fn map_error_body<F>(mut self, f: F) -> Self
118 where
119 F: Fn(http::StatusCode, &str, &str) -> Option<serde_json::Value> + Send + Sync + 'static,
120 {
121 self.error_mapper = Some(Arc::new(f));
122 self
123 }
124
125 pub fn cors(mut self, config: crate::cors::CorsConfig) -> Self {
130 self.cors = Some(std::sync::Arc::new(config));
131 self
132 }
133
134 pub fn handler_timeout(mut self, budget: std::time::Duration) -> Self {
137 self.handler_timeout = budget;
138 self
139 }
140
141 pub fn body_read_timeout(mut self, budget: std::time::Duration) -> Self {
143 self.body_read_timeout = budget;
144 self
145 }
146
147 pub fn write_stall_timeout(mut self, budget: std::time::Duration) -> Self {
151 self.write_stall_timeout = budget;
152 self
153 }
154
155 pub fn route(mut self, path: &str, methods: MethodRouter) -> Self {
157 self.routes.push((path.to_string(), methods));
158 self
159 }
160
161 pub fn mount(mut self, prefix: &str, module: Module) -> Self {
163 self.mounts.push((prefix.to_string(), module));
164 self
165 }
166
167 pub fn provide<T: Send + Sync + 'static>(mut self, value: T) -> Self {
169 self.env.insert_value(value);
170 self
171 }
172
173 pub fn provide_dep<F, Args, T>(mut self, factory: F) -> Self
175 where
176 F: DepFactory<Args, T>,
177 T: Send + Sync + 'static,
178 {
179 self.env.insert_factory(factory);
180 self
181 }
182
183 pub fn middleware<M: Middleware>(mut self, mw: M) -> Self {
185 self.middleware.push(Arc::new(mw));
186 self
187 }
188
189 pub fn on_serve<F, Fut>(mut self, name: &'static str, f: F) -> App
202 where
203 F: FnOnce(TaskContext, tokio::sync::watch::Receiver<bool>) -> Fut + Send + 'static,
204 Fut: Future<Output = ()> + Send + 'static,
205 {
206 let factory: BackgroundFactory = Box::new(move |ctx, shutdown| Box::pin(f(ctx, shutdown)));
207 self.background.push((name, factory));
208 self
209 }
210
211 pub(crate) fn take_background(&mut self) -> Vec<(&'static str, BackgroundFactory)> {
215 std::mem::take(&mut self.background)
216 }
217
218 pub fn build(self) -> Result<BuiltApp> {
221 if let Some(c) = &self.cors {
222 c.validate()?;
223 }
224 let mut trie = Trie::default();
225 let app_env = Arc::new(self.env.clone());
226 let app_mw: Arc<[Arc<dyn Middleware>]> = Arc::from(self.middleware.clone());
227
228 for (path, methods) in self.routes {
229 let body_limit = methods.body_limit;
230 let handler_timeout = methods.handler_timeout;
231 let body_read_timeout = methods.body_read_timeout;
232 insert_flat(
233 &mut trie,
234 FlatRoute {
235 path,
236 methods,
237 env: app_env.clone(),
238 middleware: app_mw.clone(),
239 body_limit,
240 handler_timeout,
241 body_read_timeout,
242 },
243 )?;
244 }
245 for (prefix, module) in self.mounts {
246 for flat in module.flatten(&prefix, &self.env, &self.middleware) {
247 insert_flat(&mut trie, flat)?;
248 }
249 }
250 Ok(BuiltApp {
251 trie,
252 app_env,
253 overrides: Arc::new(HashMap::new()),
254 security_headers: self.security_headers,
255 cors: self.cors.clone(),
256 handler_timeout: self.handler_timeout,
257 body_read_timeout: self.body_read_timeout,
258 write_stall_timeout: self.write_stall_timeout,
259 error_mapper: self.error_mapper.clone(),
260 })
261 }
262
263 pub async fn serve(self) -> Result<()> {
267 let addr = std::env::var("JERRYCAN_ADDR").unwrap_or_else(|_| "127.0.0.1:8000".to_string());
268 let listener = tokio::net::TcpListener::bind(&addr)
269 .await
270 .map_err(|e| Error::internal(format!("failed to bind {addr}: {e}")))?;
271 self.serve_with_shutdown(listener, serve::shutdown_signal())
272 .await
273 }
274
275 pub async fn serve_with(self, listener: tokio::net::TcpListener) -> Result<()> {
277 self.serve_with_shutdown(listener, std::future::pending())
278 .await
279 }
280
281 pub async fn serve_with_shutdown(
284 self,
285 listener: tokio::net::TcpListener,
286 shutdown: impl std::future::Future<Output = ()> + Send,
287 ) -> Result<()> {
288 serve::run_with_shutdown(self, listener, shutdown).await
289 }
290}
291
292fn insert_flat(trie: &mut Trie, flat: FlatRoute) -> Result<()> {
293 let stream_body = flat.methods.stream_body;
294 let mut methods = HashMap::new();
295 for (m, h) in flat.methods.handlers {
296 if methods.insert(m.clone(), h).is_some() {
297 return Err(Error::internal(format!(
298 "duplicate method {m} for `{}`",
299 flat.path
300 )));
301 }
302 }
303 trie.insert(
304 &flat.path,
305 Endpoint {
306 methods,
307 env: flat.env,
308 middleware: flat.middleware,
309 body_limit: flat.body_limit,
310 stream_body,
311 handler_timeout: flat.handler_timeout,
312 body_read_timeout: flat.body_read_timeout,
313 },
314 )
315}
316
317pub struct BuiltApp {
319 pub(crate) trie: Trie,
320 pub(crate) app_env: Arc<DepEnv>,
324 pub(crate) overrides: Arc<HashMap<TypeId, AnyArc>>,
325 pub(crate) security_headers: bool,
326 pub(crate) cors: Option<std::sync::Arc<crate::cors::CorsConfig>>,
329 pub(crate) handler_timeout: std::time::Duration,
330 pub(crate) body_read_timeout: std::time::Duration,
331 pub(crate) write_stall_timeout: std::time::Duration,
332 pub(crate) error_mapper: Option<Arc<ErrorBodyMapper>>,
335}
336
337impl std::fmt::Debug for BuiltApp {
340 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
341 f.debug_struct("BuiltApp").finish_non_exhaustive()
342 }
343}
344
345pub(crate) const BODY_LIMIT: usize = 1024 * 1024; pub(crate) enum Policy {
354 Route {
361 limit: usize,
362 stream: bool,
363 body_read_timeout: std::time::Duration,
364 },
365 Reject(Response),
368}
369
370pub(crate) fn apply_security_headers(res: &mut Response) {
372 const DEFAULTS: [(&str, &str); 5] = [
373 ("x-content-type-options", "nosniff"),
374 ("x-frame-options", "DENY"),
375 ("referrer-policy", "no-referrer"),
376 ("content-security-policy", "default-src 'none'"),
377 ("cache-control", "no-store"),
378 ];
379 for (name, value) in DEFAULTS {
380 let header_name = http::HeaderName::from_static(name);
381 if !res.headers().contains_key(&header_name) {
382 res.headers_mut()
383 .insert(header_name, http::HeaderValue::from_static(value));
384 }
385 }
386}
387
388impl BuiltApp {
389 pub fn task_context(&self) -> crate::dep::TaskContext {
397 crate::dep::TaskContext::new(DepResolver::new(
398 self.app_env.clone(),
399 self.overrides.clone(),
400 ))
401 }
402
403 pub(crate) fn apply_error_mapper(&self, response: &mut Response) {
411 let Some(mapper) = &self.error_mapper else {
412 return;
413 };
414 let Some(parts) = response
415 .extensions()
416 .get::<crate::response::ErrorParts>()
417 .cloned()
418 else {
419 return;
420 };
421 if let Some(body) = mapper(response.status(), parts.code, &parts.message) {
422 let bytes = serde_json::to_vec(&body).unwrap_or_default();
423 *response.body_mut() = crate::response::JcBody::full(bytes);
424 }
425 }
426
427 pub(crate) fn route_policy(&self, parts: &http::request::Parts) -> Policy {
439 let path = parts.uri.path();
440 if let Some(config) = &self.cors
445 && crate::cors::is_preflight(parts)
446 {
447 let origin = parts
448 .headers
449 .get(http::header::ORIGIN)
450 .and_then(|v| v.to_str().ok())
451 .unwrap_or("");
452 if config.allows_origin(origin)
453 && let Some(methods) = self.trie.methods_for(path)
454 {
455 let acrh = parts
456 .headers
457 .get(http::header::ACCESS_CONTROL_REQUEST_HEADERS)
458 .and_then(|v| v.to_str().ok());
459 return Policy::Reject(crate::cors::preflight_response(
460 config, origin, acrh, &methods,
461 ));
462 }
463 let mut r = http::Response::new(crate::response::JcBody::empty());
466 *r.status_mut() = http::StatusCode::NO_CONTENT;
467 return Policy::Reject(r);
468 }
469 let reject = |response: Response| -> Policy {
470 let mut response = response;
471 self.apply_error_mapper(&mut response);
474 if self.security_headers {
475 apply_security_headers(&mut response);
476 }
477 if let Some(config) = &self.cors {
482 crate::cors::apply_cors(
483 &mut response,
484 parts.headers.get(http::header::ORIGIN),
485 config,
486 );
487 }
488 Policy::Reject(response)
489 };
490 match self.trie.find(path, &parts.method) {
491 RouteMatch::Found { endpoint, .. } => Policy::Route {
492 limit: endpoint.body_limit.unwrap_or(BODY_LIMIT),
493 stream: endpoint.stream_body,
494 body_read_timeout: endpoint.body_read_timeout.unwrap_or(self.body_read_timeout),
495 },
496 RouteMatch::NotFound => reject(Error::not_found().into_response()),
497 RouteMatch::MethodMissing => reject(Error::method_not_allowed().into_response()),
498 RouteMatch::Malformed => {
499 reject(Error::bad_request("malformed percent-encoding in path").into_response())
500 }
501 }
502 }
503
504 pub(crate) async fn dispatch(&self, parts: http::request::Parts, lane: BodyLane) -> Response {
509 let origin = parts.headers.get(http::header::ORIGIN).cloned();
513 let mut response = self.dispatch_inner(parts, lane).await;
514 self.apply_error_mapper(&mut response);
517 if self.security_headers {
518 apply_security_headers(&mut response);
519 }
520 if let Some(config) = &self.cors {
521 crate::cors::apply_cors(&mut response, origin.as_ref(), config);
522 }
523 response
524 }
525
526 async fn dispatch_inner(&self, parts: http::request::Parts, lane: BodyLane) -> Response {
527 let method = parts.method.clone();
528 let path = parts.uri.path().to_string();
529 match self.trie.find(&path, &method) {
530 RouteMatch::NotFound => Error::not_found().into_response(),
531 RouteMatch::MethodMissing => Error::method_not_allowed().into_response(),
532 RouteMatch::Malformed => {
533 Error::bad_request("malformed percent-encoding in path").into_response()
534 }
535 RouteMatch::Found { endpoint, params } => {
536 let budget = endpoint.handler_timeout.unwrap_or(self.handler_timeout);
539 let mut ctx = RequestCtx::with_lane(
540 parts,
541 lane,
542 DepResolver::new(endpoint.env.clone(), self.overrides.clone()),
543 );
544 ctx.params = params;
545 let handler: &BoxHandlerFn = endpoint
546 .methods
547 .get(&method)
548 .expect("find() checked the method");
549 let run = Next {
550 chain: &endpoint.middleware,
551 endpoint: handler,
552 }
553 .run(&mut ctx);
554 match tokio::time::timeout(budget, run).await {
555 Ok(response) => response,
556 Err(_) => Error::handler_timeout().into_response(),
557 }
558 }
559 }
560 }
561}
562
563#[cfg(test)]
564mod tests {
565 use super::*;
566 use crate::response::Json;
567 use crate::router::get;
568 use crate::{Dep, Path};
569 use std::sync::Mutex;
570
571 #[derive(Default)]
572 struct Store {
573 items: Mutex<Vec<String>>,
574 }
575
576 async fn list(store: Dep<Store>) -> Json<Vec<String>> {
577 Json(store.items.lock().unwrap().clone())
578 }
579
580 async fn create(store: Dep<Store>, Json(item): Json<String>) -> crate::Result<Json<usize>> {
581 let mut items = store.items.lock().unwrap();
582 items.push(item);
583 Ok(Json(items.len()))
584 }
585
586 async fn show(store: Dep<Store>, Path(ix): Path<usize>) -> crate::Result<Json<String>> {
587 store
588 .items
589 .lock()
590 .unwrap()
591 .get(ix)
592 .cloned()
593 .map(Json)
594 .ok_or_else(Error::not_found)
595 }
596
597 fn crud_app() -> App {
598 App::new().provide(Store::default()).mount(
599 "/todos",
600 Module::new("todos")
601 .route("/", get(list).post(create))
602 .route("/{ix}", get(show)),
603 )
604 }
605
606 async fn dispatch(built: &BuiltApp, method: http::Method, path: &str, body: &str) -> Response {
607 let req = http::Request::builder()
608 .method(method)
609 .uri(path)
610 .body(())
611 .unwrap();
612 let (parts, ()) = req.into_parts();
613 built
614 .dispatch(parts, BodyLane::Buffered(Bytes::from(body.to_string())))
615 .await
616 }
617
618 #[tokio::test]
619 async fn crud_round_trip_in_process() {
620 let built = crud_app().build().unwrap();
621 let r = dispatch(&built, http::Method::POST, "/todos/", r#""write spike""#).await;
622 assert_eq!(r.status(), http::StatusCode::OK);
623 let r = dispatch(&built, http::Method::GET, "/todos/0", "").await;
624 assert_eq!(r.status(), http::StatusCode::OK);
625 let r = dispatch(&built, http::Method::GET, "/todos/9", "").await;
626 assert_eq!(r.status(), http::StatusCode::NOT_FOUND);
627 let r = dispatch(&built, http::Method::PATCH, "/todos/", "").await;
628 assert_eq!(r.status(), http::StatusCode::METHOD_NOT_ALLOWED);
629 let r = dispatch(&built, http::Method::GET, "/nope", "").await;
630 assert_eq!(r.status(), http::StatusCode::NOT_FOUND);
631 }
632
633 async fn body_string(r: Response) -> String {
634 use http_body_util::BodyExt;
635 let bytes = r.into_body().collect().await.unwrap().to_bytes();
636 String::from_utf8(bytes.to_vec()).unwrap()
637 }
638
639 #[tokio::test]
646 async fn registered_error_mapper_reshapes_framework_error_bodies() {
647 async fn guarded() -> crate::Result<&'static str> {
648 Err(Error::unauthorized()) }
650 let secure = || Module::new("secure").route("/", get(guarded));
651
652 let mapped = App::new()
653 .map_error_body(|_status, code, message| {
654 Some(serde_json::json!({ "error": { "code": code, "message": message } }))
655 })
656 .mount("/secure", secure())
657 .build()
658 .unwrap();
659 let r = dispatch(&mapped, http::Method::GET, "/secure/", "").await;
661 assert_eq!(r.status(), http::StatusCode::UNAUTHORIZED);
662 assert_eq!(
663 body_string(r).await,
664 r#"{"error":{"code":"JC0401","message":"authentication required"}}"#
665 );
666 let r = dispatch(&mapped, http::Method::GET, "/nope", "").await;
668 let body = body_string(r).await;
669 assert!(
670 body.contains(r#""error""#) && body.contains("JC0404"),
671 "routing 404 must be reshaped: {body}"
672 );
673
674 let plain = App::new().mount("/secure", secure()).build().unwrap();
676 let r = dispatch(&plain, http::Method::GET, "/secure/", "").await;
677 assert_eq!(
678 body_string(r).await,
679 r#"{"code":"JC0401","message":"authentication required"}"#
680 );
681 }
682
683 #[tokio::test]
686 async fn error_mapper_returning_none_preserves_the_default_body_and_details() {
687 async fn detailed() -> crate::Result<&'static str> {
688 Err(Error::unprocessable("bad")
689 .with_details(serde_json::json!([{ "field": "x", "message": "required" }])))
690 }
691 let built = App::new()
692 .map_error_body(|_s, _c, _m| None)
693 .mount("/v", Module::new("v").route("/", get(detailed)))
694 .build()
695 .unwrap();
696 let body = body_string(dispatch(&built, http::Method::GET, "/v/", "").await).await;
697 assert!(
698 body.contains("JC0422") && body.contains("details") && body.contains("\"field\""),
699 "None must keep the default body with details: {body}"
700 );
701 }
702
703 #[test]
704 fn conflicting_routes_fail_at_build_not_at_request_time() {
705 let app = App::new()
706 .route("/x", get(|| async { "a" }))
707 .route("/x", get(|| async { "b" }));
708 let err = app.build().unwrap_err();
709 assert!(err.message().contains("/x"));
710 }
711
712 #[test]
713 fn wildcard_origin_with_credentials_is_a_build_error() {
714 let err = App::new()
715 .cors(
716 crate::cors::CorsConfig::new(crate::cors::CorsOrigins::any())
717 .allow_credentials(true),
718 )
719 .build()
720 .unwrap_err();
721 assert!(
722 err.to_string().to_lowercase().contains("credential"),
723 "{err}"
724 );
725 }
726
727 #[test]
728 fn allowlist_origin_with_credentials_builds() {
729 assert!(
730 App::new()
731 .cors(
732 crate::cors::CorsConfig::new(crate::cors::CorsOrigins::list([
733 "https://app.example"
734 ]))
735 .allow_credentials(true)
736 )
737 .build()
738 .is_ok()
739 );
740 }
741
742 #[tokio::test]
743 async fn extensions_register_through_extend() {
744 struct Greeting(&'static str);
745 struct GreetingExt;
746 impl Extension for GreetingExt {
747 fn register(self, app: App) -> App {
748 app.provide(Greeting("from-extension"))
749 }
750 }
751 async fn read(g: crate::Dep<Greeting>) -> String {
752 (*g).0.to_string()
755 }
756 let t = App::new()
757 .extend(GreetingExt)
758 .route("/", crate::router::get(read))
759 .into_test();
760 assert_eq!(t.get("/").await.text(), "from-extension");
761 }
762
763 #[test]
764 fn accept_error_classification_matches_unix_reality() {
765 use std::io::{Error as IoError, ErrorKind};
766 for transient in [
767 IoError::from(ErrorKind::ConnectionAborted),
768 IoError::from(ErrorKind::ConnectionReset),
769 IoError::from(ErrorKind::Interrupted),
770 IoError::from_raw_os_error(24), IoError::from_raw_os_error(23), ] {
773 assert!(is_transient_accept_error(&transient), "{transient:?}");
774 }
775 assert!(!is_transient_accept_error(&IoError::from(
776 ErrorKind::InvalidInput
777 )));
778 assert!(!is_transient_accept_error(&IoError::from(
779 ErrorKind::PermissionDenied
780 )));
781 }
782}