1use std::collections::HashMap;
2use std::future::Future;
3use std::net::SocketAddr;
4use std::pin::Pin;
5use std::sync::Arc;
6use std::time::Duration;
7
8use anyhow::anyhow;
9use axum::Router;
10use axum::handler::HandlerWithoutStateExt;
11use axum::middleware::{from_fn, from_fn_with_state};
12use tokio::net::TcpListener;
13use tokio::sync::watch;
14use tower_http::services::ServeDir;
15use tower_http::trace::TraceLayer;
16use tracing_subscriber::EnvFilter;
17
18use crate::auth::{Gate, LoginThrottle, User};
19use crate::crypto::parse_key;
20use crate::db::{Db, Migration, MigrationStatus, Migrator};
21use crate::events::Event;
22use crate::mail::Mailer;
23use crate::queue::{Handlers, Job, Queue, Worker};
24use crate::routing::RouteInfo;
25use crate::schedule::Schedule;
26use crate::timezone::Zone;
27use crate::{
28 AppState, Config, Environment, Error, Module, Registry, Result, RouteTable, Views, assets,
29 auth, csrf, session, view,
30};
31
32type Seeder = Box<dyn Fn(AppState) -> Pin<Box<dyn Future<Output = Result> + Send>> + Send + Sync>;
33
34const SHUTDOWN_GRACE: Duration = Duration::from_secs(30);
36
37const USAGE: &str = "\
38Usage: <app> [command]
39
40Commands:
41 serve Start the web server, queue workers and scheduler (default)
42 migrate Run pending migrations
43 migrate:rollback [--step N]
44 Undo the last N batches of migrations (default 1)
45 migrate:fresh [--seed] Drop all tables, run every migration, optionally seed
46 migrate:status List migrations and whether they have run
47 db:seed Run the seeders
48 queue:work [--queue a,b] [--workers N] [--once]
49 Run queued jobs (until stopped, or --once for what's there)
50 queue:failed List failed jobs
51 queue:retry <id|all> Put failed jobs back on the queue
52 queue:flush Delete failed jobs
53 queue:forget <id> Delete one failed job
54 queue:prune-failed [--hours N]
55 Delete failed jobs older than N hours (default 168)
56 queue:prune-batches [--hours N]
57 Delete batches finished more than N hours ago (default 24)
58 webhook:failed List webhook calls whose processing failed
59 webhook:retry <id> Process a stored webhook call again
60 ui:publish [--force] Copy the UI kit (renox/ui.html and its CSS) into the app to change it
61 cache:prune Delete expired rows of the database cache store
62 session:prune Delete expired sessions (SESSION_DRIVER=database)
63 schedule:list List scheduled tasks and when they run next
64 schedule:run <task> Run one scheduled task now
65 schedule:work Run scheduled tasks (when SCHEDULER=false for serve)
66 route:list List every route with its name, module and guards
67 db:shell Run SQL against the database (`.tables`, `.quit`)
68 down [--secret S] [--retry N]
69 Maintenance mode: answer 503 (visit /S to bypass it)
70 up Leave maintenance mode
71 help Show this message";
72
73pub struct App {
98 config: Option<Config>,
99 modules: Vec<Box<dyn Module>>,
100 migrations: Vec<Migration>,
101 seeders: Vec<Seeder>,
102 gates: HashMap<String, Gate>,
103 async_gates: HashMap<String, crate::auth::AsyncGate>,
104 gate_before: Option<crate::auth::GateBefore>,
105 registry: Registry,
106 embedded: Option<crate::Embedded>,
107 csp: crate::security::Csp,
108 provided: HashMap<std::any::TypeId, Arc<dyn std::any::Any + Send + Sync>>,
109 layers: Vec<AppLayer>,
110 limiters: HashMap<String, crate::rate_limit::LimitRule>,
111 detect_locale: bool,
112 disks: Vec<(String, DiskSettings)>,
113 mailers: Vec<(String, MailerSettings)>,
114 xsrf_cookie: bool,
115}
116
117type MailerSettings = Box<dyn Fn(&Config) -> Result<crate::mail::MailConfig> + Send + Sync>;
119
120type DiskSettings = Box<dyn Fn(&Config) -> Result<crate::storage::StorageConfig> + Send + Sync>;
122
123type AppLayer = Box<dyn Fn(Router<AppState>) -> Router<AppState> + Send + Sync>;
126
127impl App {
128 pub fn new() -> Self {
130 Self {
131 config: None,
132 modules: Vec::new(),
133 migrations: Vec::new(),
134 seeders: Vec::new(),
135 gates: HashMap::new(),
136 async_gates: HashMap::new(),
137 gate_before: None,
138 registry: Registry::default(),
139 embedded: None,
140 csp: crate::security::Csp::default(),
141 provided: HashMap::new(),
142 layers: Vec::new(),
143 limiters: HashMap::new(),
144 detect_locale: false,
145 disks: Vec::new(),
146 mailers: Vec::new(),
147 xsrf_cookie: false,
148 }
149 }
150
151 pub fn detect_locale(mut self) -> Self {
165 self.detect_locale = true;
166 self
167 }
168
169 pub fn layer<L>(mut self, layer: L) -> Self
193 where
194 L: tower::Layer<axum::routing::Route> + Clone + Send + Sync + 'static,
195 L::Service: tower::Service<axum::extract::Request> + Clone + Send + Sync + 'static,
196 <L::Service as tower::Service<axum::extract::Request>>::Response:
197 axum::response::IntoResponse + 'static,
198 <L::Service as tower::Service<axum::extract::Request>>::Error:
199 Into<std::convert::Infallible> + 'static,
200 <L::Service as tower::Service<axum::extract::Request>>::Future: Send + 'static,
201 {
202 self.layers
203 .push(Box::new(move |router| router.layer(layer.clone())));
204 self
205 }
206
207 pub fn csp(mut self, allow: impl FnOnce(&mut crate::security::Csp)) -> Self {
210 allow(&mut self.csp);
211 self
212 }
213
214 pub fn embed(mut self, embedded: crate::Embedded) -> Self {
218 self.embedded = Some(embedded);
219 self
220 }
221
222 pub fn with_config(config: Config) -> Self {
224 Self {
225 config: Some(config),
226 ..Self::new()
227 }
228 }
229
230 pub fn config(mut self, config: Config) -> Self {
232 self.config = Some(config);
233 self
234 }
235
236 pub fn module(mut self, module: impl Module) -> Self {
238 self.modules.push(Box::new(module));
239 self
240 }
241
242 pub fn migrations(mut self, migrations: &[Migration]) -> Self {
244 self.migrations.extend_from_slice(migrations);
245 self
246 }
247
248 pub fn seeder<F, Fut>(mut self, seeder: F) -> Self
264 where
265 F: Fn(AppState) -> Fut + Send + Sync + 'static,
266 Fut: Future<Output = Result> + Send + 'static,
267 {
268 self.seeders
269 .push(Box::new(move |state| Box::pin(seeder(state))));
270 self
271 }
272
273 pub fn gate(
283 mut self,
284 name: &str,
285 check: impl Fn(&User) -> bool + Send + Sync + 'static,
286 ) -> Self {
287 self.gates.insert(name.to_owned(), Arc::new(check));
288 self
289 }
290
291 pub fn gate_before(
322 mut self,
323 check: impl Fn(&User, &str) -> Option<bool> + Send + Sync + 'static,
324 ) -> Self {
325 self.gate_before = Some(Arc::new(check));
326 self
327 }
328
329 pub fn gate_async<F, Fut>(mut self, name: &str, check: F) -> Self
332 where
333 F: Fn(User, AppState) -> Fut + Send + Sync + 'static,
334 Fut: Future<Output = Result<bool>> + Send + 'static,
335 {
336 self.async_gates.insert(
337 name.to_owned(),
338 Arc::new(move |user, state| Box::pin(check(user, state))),
339 );
340 self
341 }
342
343 pub fn webhook<W: crate::webhook::Webhook>(mut self) -> Self {
346 self.registry.webhook::<W>();
347 self
348 }
349
350 pub fn job<J: Job>(mut self) -> Self {
352 self.registry.job::<J>();
353 self
354 }
355
356 pub fn listen<E, F, Fut>(mut self, listener: F) -> Self
358 where
359 E: Event,
360 F: Fn(E, AppState) -> Fut + Send + Sync + 'static,
361 Fut: Future<Output = Result> + Send + 'static,
362 {
363 self.registry.listen(listener);
364 self
365 }
366
367 pub fn command<F, Fut>(mut self, name: &str, about: &str, run: F) -> Self
370 where
371 F: Fn(crate::command::Args, AppState) -> Fut + Send + Sync + 'static,
372 Fut: Future<Output = Result> + Send + 'static,
373 {
374 self.registry.command(name, about, run);
375 self
376 }
377
378 pub fn typed_command<T: crate::command::AppCommand>(mut self) -> Self {
381 self.registry.typed_command::<T>();
382 self
383 }
384
385 pub fn rate_limiter(
389 mut self,
390 name: &str,
391 rule: impl Fn(&crate::rate_limit::LimitRequest) -> crate::rate_limit::Limit
392 + Send
393 + Sync
394 + 'static,
395 ) -> Self {
396 self.limiters
397 .insert(name.to_owned(), std::sync::Arc::new(rule));
398 self
399 }
400
401 pub fn report<F, Fut>(mut self, reporter: F) -> Self
405 where
406 F: Fn(crate::report::ErrorReport, AppState) -> Fut + Send + Sync + 'static,
407 Fut: Future<Output = ()> + Send + 'static,
408 {
409 self.registry.report(reporter);
410 self
411 }
412
413 pub fn channel<F, Fut>(mut self, name: &str, send: F) -> Self
416 where
417 F: Fn(crate::auth::Recipient, serde_json::Value, AppState) -> Fut + Send + Sync + 'static,
418 Fut: Future<Output = Result> + Send + 'static,
419 {
420 self.registry.channel(name, send);
421 self
422 }
423
424 pub fn xsrf_cookie(mut self) -> Self {
430 self.xsrf_cookie = true;
431 self
432 }
433
434 pub fn mailer(
452 mut self,
453 name: &str,
454 settings: impl Fn(&Config) -> Result<crate::mail::MailConfig> + Send + Sync + 'static,
455 ) -> Self {
456 self.mailers.push((name.to_owned(), Box::new(settings)));
457 self
458 }
459
460 pub fn disk(
479 mut self,
480 name: &str,
481 settings: impl Fn(&Config) -> Result<crate::storage::StorageConfig> + Send + Sync + 'static,
482 ) -> Self {
483 self.disks.push((name.to_owned(), Box::new(settings)));
484 self
485 }
486
487 pub fn provide<T: Send + Sync + 'static>(mut self, value: T) -> Self {
493 self.provided
494 .insert(std::any::TypeId::of::<T>(), Arc::new(value));
495 self
496 }
497
498 pub fn share<F, Fut, T>(mut self, key: &str, compute: F) -> Self
500 where
501 F: Fn(crate::view::ViewContext) -> Fut + Send + Sync + 'static,
502 Fut: Future<Output = Result<T>> + Send + 'static,
503 T: serde::Serialize,
504 {
505 self.registry.share(key, compute);
506 self
507 }
508
509 pub fn templates(
511 mut self,
512 hook: impl Fn(&mut minijinja::Environment<'static>) + Send + Sync + 'static,
513 ) -> Self {
514 self.registry.templates(hook);
515 self
516 }
517
518 pub fn schedule(mut self, define: impl FnOnce(&mut Schedule)) -> Self {
520 define(self.registry.schedule());
521 self
522 }
523
524 pub async fn boot(mut self) -> Result<Kernel> {
526 let config = match self.config {
527 Some(config) => config,
528 None => Config::load()?,
529 };
530
531 self.registry.job::<crate::mail::SendMail>();
532 self.registry.job::<crate::mail::SendMailVia>();
533 self.registry
534 .job::<crate::auth::notifications::SendToChannel>();
535 self.registry.job::<crate::webhook::ProcessWebhook>();
536 #[cfg(feature = "server-events")]
537 self.registry.job::<crate::analytics::ServerEvent>();
538 let mut migrations = crate::queue::MIGRATIONS.to_vec();
539 migrations.push(crate::cache::MIGRATION);
540 migrations.push(crate::session::MIGRATION);
541 migrations.push(crate::grid::MIGRATION);
542 migrations.extend(crate::webhook::MIGRATIONS);
543 migrations.extend(self.migrations);
544 for module in &self.modules {
545 migrations.extend_from_slice(module.migrations());
546 module.register(&mut self.registry);
547 }
548 let Registry {
549 jobs,
550 listeners,
551 schedule,
552 duplicate_job,
553 webhooks,
554 commands,
555 templates,
556 shares,
557 channels,
558 reporters,
559 permissions,
560 second_factor,
561 duplicate_second_factor,
562 mut account_sections,
563 auth,
564 assets: static_assets,
565 provided: mut module_provided,
566 } = self.registry;
567 module_provided.extend(self.provided);
569 check_assets(&static_assets)?;
570 let static_assets: Arc<[crate::registry::StaticAsset]> = static_assets.into();
571 if let Some(name) = duplicate_job {
572 return Err(anyhow!("job `{name}` is registered twice").into());
573 }
574 if duplicate_second_factor {
575 return Err(
576 anyhow!("two modules set a second login step (`Registry::second_factor`)").into(),
577 );
578 }
579 check_commands(&commands)?;
580 schedule.check()?;
581 let zone: Zone = config.timezone;
582
583 let migrator = Migrator::new(migrations)?;
584 let key = match &config.key {
585 Some(key) => parse_key(key)?,
586 None => parse_key(&crate::generate_key())?,
587 };
588 let db = crate::db::connect(&config).await?.with_key(key.clone());
590
591 let mut router = Router::new();
592 let mut fallback: Option<axum::routing::MethodRouter<AppState>> = None;
593 let mut domains: Vec<(
595 crate::domain::DomainPattern,
596 Router<AppState>,
597 Option<axum::routing::MethodRouter<AppState>>,
598 )> = Vec::new();
599 let mut routes = RouteTable::default();
600 let mut listing: Vec<RouteInfo> = Vec::new();
601 for module in &self.modules {
602 tracing::debug!(module = module.name(), "registering module");
603 let parts = module.routes().into_parts();
604 check_clashes(&listing, &parts.listing, None, module.name())?;
605 router = merge_routes(router, parts.router, module.name())?;
606 for (name, path) in parts.names {
607 routes.insert(name, path)?;
608 }
609 listing.extend(parts.listing.into_iter().map(|info| RouteInfo {
610 module: module.name().to_owned(),
611 ..info
612 }));
613 if let Some(handler) = parts.fallback {
614 if fallback.is_some() {
615 return Err(anyhow!(
616 "two modules set a fallback route (`{}` is the second)",
617 module.name()
618 )
619 .into());
620 }
621 fallback = Some(handler);
622 }
623 for (text, domain_routes) in parts.domains {
624 let pattern = crate::domain::DomainPattern::parse(&text)?;
625 let inner = domain_routes.into_parts();
626 if !inner.domains.is_empty() {
627 return Err(anyhow!(
628 "Routes::domain(\"{text}\", …) inside another domain (module `{}`)",
629 module.name()
630 )
631 .into());
632 }
633 check_clashes(
634 &listing,
635 &inner.listing,
636 Some(pattern.as_str()),
637 module.name(),
638 )?;
639 let at = match domains.iter().position(|(p, _, _)| *p == pattern) {
640 Some(at) => at,
641 None => {
642 domains.push((pattern.clone(), Router::new(), None));
643 domains.len() - 1
644 }
645 };
646 let (_, domain_router, domain_fallback) = &mut domains[at];
647 *domain_router =
648 merge_routes(std::mem::take(domain_router), inner.router, module.name())?;
649 if let Some(handler) = inner.fallback {
650 if domain_fallback.is_some() {
651 return Err(anyhow!(
652 "two fallbacks for the domain `{text}` (module `{}`)",
653 module.name()
654 )
655 .into());
656 }
657 *domain_fallback = Some(handler);
658 }
659 for (name, path) in inner.names {
660 routes.set_domain(&name, pattern.as_str());
661 routes.insert(name, path)?;
662 }
663 listing.extend(inner.listing.into_iter().map(|info| RouteInfo {
664 module: module.name().to_owned(),
665 domain: Some(pattern.as_str().to_owned()),
666 ..info
667 }));
668 }
669 }
670 for layer in self.layers.iter().rev() {
672 router = layer(router);
673 for (_, domain_router, _) in &mut domains {
674 *domain_router = layer(std::mem::take(domain_router));
675 }
676 }
677 listing.extend(framework_routes(&config));
678 listing.sort_by(|a, b| (&a.path, &a.method).cmp(&(&b.path, &b.method)));
679 for info in &listing {
681 if let Some(name) = &info.name {
682 routes.set_method(name, &info.method);
683 }
684 }
685 for route in &listing {
687 for mark in &route.middleware {
688 if let Some(name) = mark.strip_prefix("throttle:")
689 && !name.contains('/')
690 && !self.limiters.contains_key(name)
691 {
692 return Err(anyhow!(
693 "{} {} uses throttle_by(\"{name}\"), but there's no App::rate_limiter(\"{name}\", …)",
694 route.method,
695 route.path
696 )
697 .into());
698 }
699 }
700 }
701 for route in &listing {
702 for provider in route
703 .middleware
704 .iter()
705 .filter_map(|m| m.strip_prefix("webhook:"))
706 {
707 if !webhooks.contains_key(provider) {
708 return Err(anyhow!(
709 "the route {} receives `{provider}` webhooks, but they're not registered: \
710 add `app.webhook::<…>()` in the module's `register`",
711 route.path
712 )
713 .into());
714 }
715 }
716 }
717 let routes = Arc::new(routes);
718
719 let storage = crate::storage::Storage::from_config(&config)?;
720 let mut disks = HashMap::new();
721 for (name, settings) in &self.disks {
722 let valid = !name.is_empty()
723 && name
724 .chars()
725 .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_');
726 if !valid {
727 return Err(anyhow!(
728 "disk name `{name}`: use letters, digits, `-` and `_` (App::disk)"
729 )
730 .into());
731 }
732 let disk = crate::storage::Storage::named(&config, name, &settings(&config)?)?;
733 if disks.insert(name.clone(), disk).is_some() {
734 return Err(anyhow!("two disks are named `{name}` (App::disk)").into());
735 }
736 }
737 let embedded = self.embedded.filter(|_| !config.debug);
739 let shared_counters =
741 (config.cache_store == crate::CacheStore::Database).then(|| db.clone());
742 let versions = Arc::new(crate::embedded::AssetVersions::new(
743 &config.public_path,
744 embedded.map(|e| e.public),
745 ));
746 let views = Views::new(
747 &config,
748 routes.clone(),
749 storage.clone(),
750 embedded.map(|e| e.views),
751 Arc::new(templates),
752 zone,
753 versions,
754 );
755 let mut security = crate::security::Security::new(&config, &self.csp, &listing);
756 security.xsrf_cookie = self.xsrf_cookie;
757 let security = Arc::new(security);
758 let mut mailers = HashMap::new();
759 for (name, settings) in &self.mailers {
760 let mailer = Mailer::open(&settings(&config)?, &config)?;
761 if mailers.insert(name.clone(), mailer).is_some() {
762 return Err(anyhow!("two mailers are named `{name}` (App::mailer)").into());
763 }
764 }
765 let mut failover = Vec::new();
766 for name in &config.mail.failover {
767 let mailer = mailers.get(name).ok_or_else(|| {
768 anyhow!("MAIL_FAILOVER names `{name}`, but no mailer has that name: add it with App::mailer")
769 })?;
770 failover.push((name.clone(), mailer.clone()));
771 }
772 let mailer = Mailer::from_config(&config)?.with_failover(failover);
773 let state = AppState::new(crate::state::AppStateInner {
774 security,
775 webhooks: Arc::new(webhooks),
776 mailer,
777 mailers: Arc::new(mailers),
778 queue: Queue::new(db.clone(), key.clone()),
779 cache: crate::cache::Cache::new(config.cache_store, db.clone())?,
780 storage,
781 disks: Arc::new(disks),
782 http: crate::http::Http::default(),
783 fakes: Arc::default(),
784 translator: Arc::new(match embedded {
785 Some(files) => crate::i18n::Translator::embedded(files.lang)?,
786 None => crate::i18n::Translator::load(&config.lang_path, config.debug)?,
787 }),
788 session_mirror: (config.session_driver == crate::SessionDriver::Database
790 && config.env == Environment::Testing)
791 .then(Default::default),
792 live: (config.debug && config.env == Environment::Local).then(|| {
793 crate::live::Live::start(vec![
794 config.views_path.clone(),
795 config.public_path.clone(),
796 config.lang_path.clone(),
797 ])
798 }),
799 notification_hub: Arc::new(crate::auth::notifications::Hub::new()),
800 listeners: Arc::new(listeners),
801 inspector: (config.debug && config.env == Environment::Local)
802 .then(|| Arc::new(crate::inspector::Inspector::default())),
803 config: Arc::new(config),
804 routes,
805 views,
806 db,
807 key,
808 gates: Arc::new(crate::auth::Access {
809 gates: self.gates,
810 before: self.gate_before,
811 permissions,
812 }),
813 async_gates: Arc::new(self.async_gates),
814 shares: Arc::new(shares),
815 channels: Arc::new(channels),
816 reporters: Arc::new(reporters),
817 second_factor: second_factor.map(Arc::new),
818 account_sections: {
819 account_sections.sort_by_key(|section| section.order);
820 Arc::new(account_sections)
821 },
822 auth,
823 limiters: Arc::new(
824 self.limiters
825 .into_iter()
826 .map(|(name, rule)| (name, crate::rate_limit::NamedLimiter::new(rule)))
827 .collect(),
828 ),
829 provided: Arc::new(module_provided),
830 throttle: Arc::new(LoginThrottle::new(shared_counters.clone())),
831 detect_locale: self.detect_locale,
832 });
833 if let Some(second) = &state.second_factor
834 && state.url(&second.challenge, &[]).is_err()
835 {
836 return Err(anyhow!(
837 "the second login step's challenge route `{}` doesn't exist",
838 second.challenge
839 )
840 .into());
841 }
842
843 let public = embedded.map(|e| e.public);
844 let default = build_router(router, state.clone(), public, fallback, &static_assets);
845 let router = if domains.is_empty() {
846 default
847 } else {
848 let hosts = domains
849 .into_iter()
850 .map(|(pattern, router, fallback)| {
851 (
852 pattern,
853 build_router(router, state.clone(), public, fallback, &static_assets),
854 )
855 })
856 .collect();
857 crate::domain::dispatch(hosts, default)
858 };
859 Ok(Kernel {
860 listing,
861 router,
862 state,
863 migrator,
864 seeders: self.seeders,
865 handlers: Arc::new(jobs),
866 schedule,
867 zone,
868 commands,
869 })
870 }
871
872 pub async fn into_router(self) -> Result<Router> {
874 Ok(self.boot().await?.router)
875 }
876
877 pub fn run(self) -> Result {
880 let args: Vec<String> = std::env::args().skip(1).collect();
881 tokio::runtime::Builder::new_multi_thread()
882 .enable_all()
883 .build()?
884 .block_on(self.run_command(&args))
885 }
886
887 pub async fn run_args(self, args: impl IntoIterator<Item = impl Into<String>>) -> Result {
898 let args: Vec<String> = args.into_iter().map(Into::into).collect();
899 self.run_command(&args).await
900 }
901
902 pub async fn serve(self) -> Result {
904 self.run_command(&[]).await
905 }
906
907 async fn run_command(self, args: &[String]) -> Result {
908 let command = args.first().map(String::as_str).unwrap_or("serve");
909 if matches!(command, "help" | "--help" | "-h") {
910 let mut modules = Registry::default();
912 for module in &self.modules {
913 module.register(&mut modules);
914 }
915 let mut commands = self.registry.commands.clone();
916 commands.extend(modules.commands);
917 println!("{USAGE}{}", app_commands_help(&commands));
918 return Ok(());
919 }
920
921 let config = match self.config.clone() {
922 Some(config) => config,
923 None => Config::load()?,
924 };
925 let long_running = matches!(command, "serve" | "queue:work" | "schedule:work");
926 init_tracing(&config, long_running);
927 if command == "serve" && config.key.is_none() && config.env != Environment::Testing {
928 tracing::warn!(
929 "APP_KEY is not set; using a temporary key, so sessions end on restart. \
930 Run `rnx key:generate`."
931 );
932 }
933 let kernel = App {
934 config: Some(config),
935 ..self
936 }
937 .boot()
938 .await?;
939
940 match command {
941 "serve" => kernel.serve().await?,
942 "migrate" => print_done("Migrated", &kernel.migrate().await?),
943 "migrate:rollback" => {
944 let steps = flag_value(args, "--step")?.unwrap_or(1);
945 print_done("Rolled back", &kernel.rollback(steps).await?);
946 }
947 "migrate:fresh" => {
948 println!("Dropped all tables.");
949 print_done("Migrated", &kernel.fresh().await?);
950 if args.iter().any(|a| a == "--seed") {
951 kernel.seed().await?;
952 println!("Seeded.");
953 }
954 }
955 "migrate:status" => {
956 for m in kernel.migration_status().await? {
957 let note = if m.missing {
958 " (applied, but its file is gone)"
959 } else if m.changed {
960 " (edited after it ran; the edit won't run)"
961 } else {
962 ""
963 };
964 match m.batch {
965 Some(batch) => println!(" ran (batch {batch}) {}{note}", m.name),
966 None => println!(" pending {}", m.name),
967 }
968 }
969 }
970 "db:seed" => {
971 kernel.seed().await?;
972 println!("Seeded.");
973 }
974 "queue:work" => {
975 let queues: Vec<String> = flag_text(args, "--queue")
976 .map(|q| q.split(',').map(|s| s.trim().to_owned()).collect())
977 .unwrap_or_default();
978 let worker = kernel.worker(queues);
979 if args.iter().any(|a| a == "--once") {
980 println!("Ran {} job(s).", worker.drain().await?);
981 } else {
982 let workers = flag_value(args, "--workers")?.unwrap_or(1) as usize;
983 let (stop, stopped) = watch::channel(false);
984 let running = tokio::spawn(worker.run(workers, stopped));
985 shutdown_signal().await;
986 let _ = stop.send(true);
987 let _ = running.await;
988 }
989 }
990 "webhook:failed" => {
991 let failed = crate::webhook::WebhookCall::failed(&kernel.state.db).await?;
992 if failed.is_empty() {
993 println!("No failed webhook calls.");
994 }
995 for call in failed {
996 println!(
997 " #{} {} {}: {}",
998 call.id,
999 call.provider,
1000 call.event_id,
1001 call.error
1002 .unwrap_or_default()
1003 .lines()
1004 .next()
1005 .unwrap_or_default()
1006 );
1007 }
1008 }
1009 "webhook:retry" => {
1010 let id: i64 = args
1011 .get(1)
1012 .and_then(|id| id.parse().ok())
1013 .ok_or_else(|| anyhow!("usage: webhook:retry <id>"))?;
1014 if crate::webhook::retry(&kernel.state, id).await? {
1015 println!("Webhook call #{id} queued again.");
1016 } else {
1017 return Err(anyhow!("there is no webhook call #{id}").into());
1018 }
1019 }
1020 "queue:failed" => {
1021 let failed = kernel.state.queue.failed().await?;
1022 if failed.is_empty() {
1023 println!("No failed jobs.");
1024 }
1025 for job in failed {
1026 println!(" #{} {} ({}): {}", job.id, job.job, job.queue, job.error);
1027 }
1028 }
1029 "queue:retry" => {
1030 let id = match args.get(1).map(String::as_str) {
1031 Some("all") => None,
1032 Some(id) => Some(
1033 id.parse()
1034 .map_err(|_| anyhow!("expected a job id or `all`"))?,
1035 ),
1036 None => return Err(anyhow!("usage: queue:retry <id|all>").into()),
1037 };
1038 let queued = match id {
1039 Some(id) => u64::from(kernel.state.queue.retry(id).await?),
1040 None => kernel.state.queue.retry_all().await?,
1041 };
1042 println!("Queued {queued} job(s) again.");
1043 }
1044 "queue:forget" => {
1045 let id: i64 = args
1046 .get(1)
1047 .and_then(|id| id.parse().ok())
1048 .ok_or_else(|| anyhow!("usage: queue:forget <id>"))?;
1049 if kernel.state.queue.forget_failed(id).await? {
1050 println!("Deleted failed job {id}.");
1051 } else {
1052 return Err(anyhow!("no failed job {id}").into());
1053 }
1054 }
1055 "queue:prune-failed" => {
1056 let hours = flag_value(args, "--hours")?.unwrap_or(168);
1057 let age = Duration::from_secs(u64::from(hours) * 3600);
1058 println!(
1059 "Deleted {} failed job(s) older than {hours} hour(s).",
1060 kernel.state.queue.prune_failed(age).await?
1061 );
1062 }
1063 "queue:prune-batches" => {
1064 let hours = flag_value(args, "--hours")?.unwrap_or(24);
1065 let age = Duration::from_secs(u64::from(hours) * 3600);
1066 println!(
1067 "Deleted {} batch(es) finished more than {hours} hour(s) ago.",
1068 kernel.state.queue.prune_batches(age).await?
1069 );
1070 }
1071 "queue:flush" => println!(
1072 "Deleted {} failed job(s).",
1073 kernel.state.queue.flush_failed().await?
1074 ),
1075 "ui:publish" => {
1076 let force = args.iter().any(|a| a == "--force");
1077 crate::assets::publish_ui(&kernel.state.config, force)?;
1078 }
1079 "session:prune" => println!(
1080 "Deleted {} expired session(s).",
1081 crate::Session::prune_expired(&kernel.state.db).await?
1082 ),
1083 "cache:prune" => println!(
1084 "Deleted {} expired cache row(s).",
1085 kernel.state.cache.prune().await?
1086 ),
1087 "schedule:list" => {
1088 if kernel.schedule.is_empty() {
1089 println!("No scheduled tasks.");
1090 }
1091 for run in kernel.schedule.upcoming(kernel.zone) {
1092 let when = match run.at {
1093 Some(at) => run
1094 .zone
1095 .local(at.timestamp())
1096 .format("%Y-%m-%d %H:%M")
1097 .to_string(),
1098 None => "never".to_owned(),
1099 };
1100 println!(" {when:<16} {:<18} {}", run.zone, run.name);
1101 }
1102 }
1103 "schedule:run" => {
1104 let Some(name) = args.get(1).filter(|a| !a.starts_with('-')) else {
1105 return Err(anyhow!("usage: schedule:run <task>").into());
1106 };
1107 kernel.run_scheduled(name).await?;
1108 println!("Ran `{name}`.");
1109 }
1110 "schedule:work" => {
1111 let (stop, stopped) = watch::channel(false);
1112 let running = tokio::spawn(kernel.schedule.clone().run(
1113 kernel.state.clone(),
1114 kernel.zone,
1115 stopped,
1116 ));
1117 shutdown_signal().await;
1118 let _ = stop.send(true);
1119 let _ = running.await;
1120 }
1121 "route:list" => print_routes(kernel.routes()),
1122 "db:shell" => crate::shell::run(kernel.db()).await?,
1123 "down" => {
1124 let secret = flag_text(args, "--secret").map(str::to_owned);
1125 let retry = flag_value(args, "--retry")?.map(u64::from);
1126 let mut options = crate::maintenance::DownOptions::new();
1127 if let Some(secret) = &secret {
1128 options = options.secret(secret.clone());
1129 }
1130 if let Some(retry) = retry {
1131 options = options.retry(retry);
1132 }
1133 crate::maintenance::down(&kernel.state.config.storage_path, options)?;
1134 match secret {
1135 Some(secret) => println!("The app is down. Visit /{secret} to bypass it."),
1136 None => println!("The app is down."),
1137 }
1138 }
1139 "up" => match crate::maintenance::up(&kernel.state.config.storage_path)? {
1140 true => println!("The app is up."),
1141 false => println!("The app was not down."),
1142 },
1143 other if kernel.commands.iter().any(|c| c.name == other) => {
1144 kernel.call(other, args[1..].iter().cloned()).await?;
1145 }
1146 other => {
1147 return Err(anyhow!(
1148 "unknown command `{other}`\n\n{USAGE}{}",
1149 app_commands_help(&kernel.commands)
1150 )
1151 .into());
1152 }
1153 }
1154 Ok(())
1155 }
1156}
1157
1158fn inherited_listener() -> Result<Option<TcpListener>> {
1162 let mut fds = listenfd::ListenFd::from_env();
1163 let Some(listener) = fds
1164 .take_tcp_listener(0)
1165 .map_err(|err| anyhow!("the socket systemd passed isn't a TCP listener: {err}"))?
1166 else {
1167 return Ok(None);
1168 };
1169 listener
1170 .set_nonblocking(true)
1171 .map_err(anyhow::Error::from)?;
1172 Ok(Some(
1173 TcpListener::from_std(listener).map_err(anyhow::Error::from)?,
1174 ))
1175}
1176
1177const BUILT_IN_COMMANDS: &[&str] = &[
1184 "serve",
1185 "migrate",
1186 "migrate:rollback",
1187 "migrate:fresh",
1188 "migrate:status",
1189 "db:seed",
1190 "queue:work",
1191 "queue:failed",
1192 "queue:retry",
1193 "queue:flush",
1194 "queue:forget",
1195 "queue:prune-failed",
1196 "queue:prune-batches",
1197 "webhook:failed",
1198 "webhook:retry",
1199 "cache:prune",
1200 "session:prune",
1201 "ui:publish",
1202 "schedule:list",
1203 "schedule:run",
1204 "schedule:work",
1205 "route:list",
1206 "db:shell",
1207 "down",
1208 "up",
1209 "help",
1210];
1211
1212fn check_commands(commands: &[crate::command::Command]) -> Result {
1213 let mut seen = std::collections::HashSet::new();
1214 for command in commands {
1215 let name = command.name.as_str();
1216 if name.is_empty() || name.starts_with('-') || name.contains(char::is_whitespace) {
1217 return Err(anyhow!("`{name}` is not a valid command name").into());
1218 }
1219 if BUILT_IN_COMMANDS.contains(&name) {
1220 return Err(anyhow!("the command `{name}` is built in; choose another name").into());
1221 }
1222 if !seen.insert(name) {
1223 return Err(anyhow!("the command `{name}` is registered twice").into());
1224 }
1225 }
1226 Ok(())
1227}
1228
1229fn app_commands_help(commands: &[crate::command::Command]) -> String {
1230 if commands.is_empty() {
1231 return String::new();
1232 }
1233 let mut out = String::from("\n\nApp commands:\n");
1234 for command in commands {
1235 out.push_str(&format!(" {:<26}{}\n", command.name, command.about));
1236 }
1237 out.trim_end().to_owned()
1238}
1239
1240impl Default for App {
1241 fn default() -> Self {
1242 Self::new()
1243 }
1244}
1245
1246pub struct Kernel {
1248 router: Router,
1249 listing: Vec<RouteInfo>,
1250 state: AppState,
1251 migrator: Migrator,
1252 seeders: Vec<Seeder>,
1253 handlers: Handlers,
1254 schedule: Schedule,
1255 zone: Zone,
1256 commands: Vec<crate::command::Command>,
1257}
1258
1259impl Kernel {
1260 pub async fn call(
1262 &self,
1263 name: &str,
1264 args: impl IntoIterator<Item = impl Into<String>>,
1265 ) -> Result {
1266 let command = self
1267 .commands
1268 .iter()
1269 .find(|c| c.name == name)
1270 .ok_or_else(|| anyhow!("unknown command `{name}`"))?;
1271 let run = (command.run)(self.state.clone(), crate::command::Args::new(args));
1272 crate::context::scope_app(self.state.clone(), run).await
1273 }
1274
1275 pub async fn run_scheduled(&self, name: &str) -> Result {
1279 self.schedule.run_now(self.state.clone(), name).await
1280 }
1281
1282 pub fn router(&self) -> Router {
1284 self.router.clone()
1285 }
1286
1287 pub fn routes(&self) -> &[RouteInfo] {
1289 &self.listing
1290 }
1291
1292 pub fn state(&self) -> &AppState {
1294 &self.state
1295 }
1296
1297 pub fn db(&self) -> &Db {
1299 &self.state.db
1300 }
1301
1302 pub fn mailer(&self) -> &Mailer {
1304 &self.state.mailer
1305 }
1306
1307 pub fn worker(&self, queues: Vec<String>) -> Worker {
1309 Worker::new(self.state.clone(), self.handlers.clone(), queues)
1310 }
1311
1312 pub async fn run_jobs(&self) -> Result<usize> {
1314 self.worker(Vec::new()).drain().await
1315 }
1316
1317 pub async fn serve(self) -> Result {
1320 let config = self.state.config.clone();
1321 let listener = match inherited_listener()? {
1322 Some(listener) => {
1323 tracing::info!("using the socket systemd passed (socket activation)");
1324 listener
1325 }
1326 None => TcpListener::bind(config.addr()).await?,
1327 };
1328 tracing::info!(
1329 "{} listening on http://{}",
1330 config.name,
1331 listener.local_addr()?
1332 );
1333
1334 let (stop, stopped) = watch::channel(false);
1335 let mut background = Vec::new();
1336 if config.queue_workers > 0 {
1337 tracing::info!(workers = config.queue_workers, "queue workers started");
1338 let worker = self.worker(Vec::new());
1339 background.push(tokio::spawn(
1340 worker.run(config.queue_workers, stopped.clone()),
1341 ));
1342 }
1343 if config.scheduler && !self.schedule.is_empty() {
1344 tracing::info!("scheduler started");
1345 background.push(tokio::spawn(self.schedule.clone().run(
1346 self.state.clone(),
1347 self.zone,
1348 stopped,
1349 )));
1350 }
1351
1352 let service = self
1353 .router
1354 .into_make_service_with_connect_info::<SocketAddr>();
1355 let live = self.state.live.clone();
1356 let notification_hub = self.state.notification_hub.clone();
1357 axum::serve(listener, service)
1358 .with_graceful_shutdown(async move {
1359 shutdown_signal().await;
1360 if let Some(live) = live {
1363 live.stop();
1364 }
1365 notification_hub.stop();
1366 })
1367 .await?;
1368
1369 let _ = stop.send(true);
1370 let finished = async {
1371 for task in background {
1372 let _ = task.await;
1373 }
1374 };
1375 if tokio::time::timeout(SHUTDOWN_GRACE, finished)
1376 .await
1377 .is_err()
1378 {
1379 tracing::warn!("background work was still running after {SHUTDOWN_GRACE:?}");
1380 }
1381 tracing::info!("{} stopped", config.name);
1382 Ok(())
1383 }
1384
1385 pub async fn migrate(&self) -> Result<Vec<String>> {
1387 let done = self.migrator.run(self.db()).await?;
1388 Ok(done.into_iter().map(str::to_owned).collect())
1389 }
1390
1391 pub async fn rollback(&self, batches: u32) -> Result<Vec<String>> {
1393 Ok(self.migrator.rollback(self.db(), batches).await?)
1394 }
1395
1396 pub async fn fresh(&self) -> Result<Vec<String>> {
1398 let done = self.migrator.fresh(self.db()).await?;
1399 Ok(done.into_iter().map(str::to_owned).collect())
1400 }
1401
1402 pub async fn migration_status(&self) -> Result<Vec<MigrationStatus>> {
1404 Ok(self.migrator.status(self.db()).await?)
1405 }
1406
1407 pub async fn seed(&self) -> Result {
1409 for seeder in &self.seeders {
1410 crate::context::scope_app(self.state.clone(), seeder(self.state.clone())).await?;
1413 }
1414 Ok(())
1415 }
1416}
1417
1418fn framework_routes(config: &Config) -> Vec<RouteInfo> {
1420 let route = |method: &str, path: &str| RouteInfo {
1421 method: method.to_owned(),
1422 path: path.to_owned(),
1423 name: None,
1424 module: "renox".to_owned(),
1425 middleware: Vec::new(),
1426 domain: None,
1427 };
1428 let mut routes = vec![
1429 route("GET", "/health"),
1430 route("GET", "/robots.txt"),
1431 route("GET", "/favicon.ico"),
1432 route("GET", "/_renox/{asset}"),
1433 route("GET", "/_renox/files/{*key}"),
1434 route("POST", "/_renox/grid/{grid}/prefs"),
1435 route("DELETE", "/_renox/grid/{grid}/prefs"),
1436 route("GET", "/storage/{*path}"),
1437 ];
1438 if config.debug {
1439 routes.push(route("GET", "/_renox/mail"));
1440 routes.push(route("GET", "/_renox/mail/{id}"));
1441 }
1442 if config.debug && config.env == Environment::Local {
1443 routes.push(route("GET", "/_renox/live"));
1444 }
1445 routes
1446}
1447
1448fn print_routes(routes: &[RouteInfo]) {
1449 print!("{}", route_table(routes));
1450}
1451
1452fn route_table(routes: &[RouteInfo]) -> String {
1454 use std::fmt::Write as _;
1455 let mut out = String::new();
1456 let with_domains = routes.iter().any(|r| r.domain.is_some());
1457 let rows: Vec<Vec<String>> = routes
1458 .iter()
1459 .map(|r| {
1460 let mut row = vec![
1461 r.method.clone(),
1462 r.path.clone(),
1463 r.name.clone().unwrap_or_default(),
1464 r.module.clone(),
1465 r.middleware
1466 .iter()
1467 .map(|guard| {
1468 if is_role_guard(guard) {
1469 format!("{guard}*")
1470 } else {
1471 guard.clone()
1472 }
1473 })
1474 .collect::<Vec<_>>()
1475 .join(", "),
1476 ];
1477 if with_domains {
1478 row.insert(0, r.domain.clone().unwrap_or_default());
1479 }
1480 row
1481 })
1482 .collect();
1483 let mut header: Vec<String> = ["METHOD", "PATH", "NAME", "MODULE", "MIDDLEWARE"]
1484 .map(str::to_owned)
1485 .to_vec();
1486 if with_domains {
1487 header.insert(0, "DOMAIN".into());
1488 }
1489 let mut widths: Vec<usize> = header.iter().map(String::len).collect();
1490 for row in &rows {
1491 for (width, cell) in widths.iter_mut().zip(row) {
1492 *width = (*width).max(cell.chars().count());
1493 }
1494 }
1495 for row in std::iter::once(&header).chain(&rows) {
1496 let line: Vec<String> = row
1497 .iter()
1498 .zip(&widths)
1499 .map(|(cell, width)| format!("{cell:<width$}", width = *width))
1500 .collect();
1501 let _ = writeln!(out, "{}", line.join(" ").trim_end());
1502 }
1503 if routes
1504 .iter()
1505 .flat_map(|r| &r.middleware)
1506 .any(|g| is_role_guard(g))
1507 {
1508 out.push_str(
1509 "\n* role and permission guards count the user's global roles plus the roles \
1510 given in the request's active scope (permissions::set_scope).\n",
1511 );
1512 }
1513 out
1514}
1515
1516fn is_role_guard(guard: &str) -> bool {
1518 guard.starts_with("role:") || guard.starts_with("permission:")
1519}
1520
1521fn print_done(verb: &str, names: &[String]) {
1522 if names.is_empty() {
1523 println!("Nothing to do.");
1524 }
1525 for name in names {
1526 println!("{verb}: {name}");
1527 }
1528}
1529
1530fn flag_text<'a>(args: &'a [String], flag: &str) -> Option<&'a str> {
1531 let i = args.iter().position(|a| a == flag)?;
1532 args.get(i + 1).map(String::as_str)
1533}
1534
1535fn flag_value(args: &[String], flag: &str) -> Result<Option<u32>> {
1536 if !args.iter().any(|a| a == flag) {
1537 return Ok(None);
1538 }
1539 match flag_text(args, flag).and_then(|v| v.parse().ok()) {
1540 Some(value) => Ok(Some(value)),
1541 None => Err(anyhow!("{flag} needs a number").into()),
1542 }
1543}
1544
1545async fn guard(
1548 limit: Option<Duration>,
1549 req: axum::extract::Request,
1550 next: axum::middleware::Next,
1551) -> axum::response::Response {
1552 use axum::response::IntoResponse;
1553 let run = futures_util::FutureExt::catch_unwind(std::panic::AssertUnwindSafe(next.run(req)));
1554 let outcome = match limit {
1555 Some(limit) => match tokio::time::timeout(limit, run).await {
1556 Ok(outcome) => outcome,
1557 Err(_) => {
1558 return Error::Internal(anyhow!(
1559 "the request took longer than REQUEST_TIMEOUT ({}s)",
1560 limit.as_secs()
1561 ))
1562 .into_response();
1563 }
1564 },
1565 None => run.await,
1566 };
1567 outcome.unwrap_or_else(|panic| {
1568 let message = crate::error::panic_message(&*panic);
1569 Error::Internal(anyhow!("the handler panicked: {message}")).into_response()
1570 })
1571}
1572
1573const BACKSTOP_GRACE: Duration = Duration::from_secs(1);
1576
1577async fn backstop(
1582 limit: Option<Duration>,
1583 req: axum::extract::Request,
1584 next: axum::middleware::Next,
1585) -> axum::response::Response {
1586 use axum::response::IntoResponse;
1587 let Some(limit) = limit else {
1588 return next.run(req).await;
1589 };
1590 match tokio::time::timeout(limit + BACKSTOP_GRACE, next.run(req)).await {
1591 Ok(res) => res,
1592 Err(_) => Error::Internal(anyhow!(
1593 "the request took longer than REQUEST_TIMEOUT ({}s) before reaching its handler",
1594 limit.as_secs()
1595 ))
1596 .into_response(),
1597 }
1598}
1599
1600fn merge_routes(
1602 router: Router<AppState>,
1603 other: Router<AppState>,
1604 module: &str,
1605) -> Result<Router<AppState>> {
1606 std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| router.merge(other))).map_err(
1607 |panic| {
1608 let message = crate::error::panic_message(&*panic);
1609 anyhow!("the routes of the `{module}` module clash: {message}").into()
1610 },
1611 )
1612}
1613
1614fn check_clashes(
1617 listing: &[RouteInfo],
1618 new: &[RouteInfo],
1619 domain: Option<&str>,
1620 module: &str,
1621) -> Result {
1622 for info in new {
1623 let clash = listing.iter().find(|other| {
1624 other.domain.as_deref() == domain
1625 && other.path == info.path
1626 && (other.method == info.method || other.method == "*" || info.method == "*")
1627 });
1628 if let Some(other) = clash {
1629 let on = domain.map(|d| format!(" on {d}")).unwrap_or_default();
1630 return Err(anyhow!(
1631 "{} {}{on} is defined by both the `{}` and the `{module}` module",
1632 info.method,
1633 info.path,
1634 other.module,
1635 )
1636 .into());
1637 }
1638 }
1639 Ok(())
1640}
1641
1642fn check_assets(assets: &[crate::registry::StaticAsset]) -> Result<()> {
1644 let mut seen = std::collections::HashSet::new();
1645 for asset in assets {
1646 if !asset.path.starts_with('/') || asset.path.contains(['{', '}']) {
1647 return Err(anyhow!(
1648 "Registry::asset(\"{}\", …): the path must start with `/` and have no `{{…}}` parameters",
1649 asset.path
1650 )
1651 .into());
1652 }
1653 if !seen.insert(asset.path) {
1654 return Err(anyhow!(
1655 "two modules serve a file at `{}` (Registry::asset)",
1656 asset.path
1657 )
1658 .into());
1659 }
1660 }
1661 Ok(())
1662}
1663
1664fn build_router(
1665 router: Router<AppState>,
1666 state: AppState,
1667 embedded_public: Option<&'static [(&'static str, &'static [u8])]>,
1668 fallback: Option<axum::routing::MethodRouter<AppState>>,
1669 static_assets: &[crate::registry::StaticAsset],
1670) -> Router {
1671 let fallback = fallback.map(|handler| handler.with_state(state.clone()));
1673 let not_found = move |req: axum::extract::Request| {
1674 let fallback = fallback.clone();
1675 async move {
1676 match fallback {
1677 Some(handler) => match tower::ServiceExt::oneshot(handler, req).await {
1678 Ok(res) => res,
1679 Err(never) => match never {},
1680 },
1681 None => axum::response::IntoResponse::into_response(Error::NotFound),
1682 }
1683 }
1684 };
1685 let router = router
1686 .merge(crate::storage::router())
1687 .merge(crate::grid::router());
1688 let router = if state.config.debug {
1689 router
1690 .merge(crate::mail::preview_router())
1691 .merge(crate::inspector::router())
1692 } else {
1693 router
1694 };
1695 let public = state.config.public_path.clone();
1696 let router = if let Some(files) = embedded_public {
1697 let files = crate::embedded::public_map(files);
1698 let not_found = not_found.clone();
1699 router.fallback(move |req: axum::extract::Request| {
1700 let (files, not_found) = (files.clone(), not_found.clone());
1701 async move {
1702 let res = crate::embedded::serve(&files, req.uri());
1703 if res.status() == axum::http::StatusCode::NOT_FOUND {
1704 not_found(req).await
1705 } else {
1706 res
1707 }
1708 }
1709 })
1710 } else if public.is_dir() {
1711 let files = ServeDir::new(public).fallback(not_found.clone().into_service());
1714 router.fallback(move |req: axum::extract::Request| {
1715 let files = files.clone();
1716 async move {
1717 if req
1720 .uri()
1721 .path()
1722 .split('/')
1723 .any(|segment| segment.len() > 255)
1724 {
1725 return axum::response::IntoResponse::into_response(Error::NotFound);
1726 }
1727 let versioned = crate::embedded::is_versioned(req.uri());
1728 match tower::ServiceExt::oneshot(files, req).await {
1729 Ok(res) => {
1730 let mut res = axum::response::IntoResponse::into_response(res);
1731 if versioned && res.status().is_success() {
1732 res.headers_mut().insert(
1733 axum::http::header::CACHE_CONTROL,
1734 axum::http::HeaderValue::from_static(
1735 "public, max-age=31536000, immutable",
1736 ),
1737 );
1738 }
1739 res
1740 }
1741 Err(never) => match never {},
1742 }
1743 }
1744 })
1745 } else {
1746 router.fallback(not_found)
1747 };
1748
1749 let request_timeout = state.config.request_timeout;
1750 let router: Router = router
1751 .layer(from_fn(
1752 move |req: axum::extract::Request, next: axum::middleware::Next| {
1753 guard(request_timeout, req, next)
1754 },
1755 ))
1756 .layer(from_fn_with_state(
1757 state.clone(),
1758 crate::maintenance::middleware,
1759 ))
1760 .layer(from_fn_with_state(state.clone(), view::middleware))
1761 .layer(from_fn(csrf::middleware))
1762 .layer(from_fn_with_state(state.clone(), auth::middleware))
1763 .layer(from_fn_with_state(state.clone(), crate::i18n::middleware))
1764 .layer(from_fn_with_state(state.clone(), session::middleware))
1765 .layer(from_fn_with_state(
1767 state.clone(),
1768 crate::context::middleware,
1769 ))
1770 .layer(from_fn(
1771 move |req: axum::extract::Request, next: axum::middleware::Next| {
1772 backstop(request_timeout, req, next)
1773 },
1774 ))
1775 .layer(from_fn_with_state(
1777 state.clone(),
1778 crate::inspector::middleware,
1779 ))
1780 .merge(assets::router())
1781 .merge(assets::module_router(static_assets))
1782 .merge(crate::health::router())
1783 .merge(robots(&state, embedded_public))
1784 .merge(favicon(&state, embedded_public))
1785 .merge(crate::live::router())
1786 .merge(public_files(&state))
1787 .layer(axum::extract::DefaultBodyLimit::max(
1788 state.config.upload_max_size,
1789 ))
1790 .layer(
1791 TraceLayer::new_for_http().make_span_with(|req: &axum::extract::Request| {
1792 let ip = crate::ClientIp::of(req).map(|ip| ip.to_string());
1793 let id = req
1794 .headers()
1795 .get(crate::request_id::HEADER)
1796 .and_then(|v| v.to_str().ok())
1797 .unwrap_or_default();
1798 tracing::info_span!(
1799 "request",
1800 id = %id,
1801 method = %req.method(),
1802 uri = %req.uri(),
1803 ip = ip.as_deref().unwrap_or("unknown"),
1804 )
1805 }),
1806 )
1807 .layer(from_fn(crate::request_id::middleware))
1808 .with_state(state.clone());
1809 let limit = state.config.upload_max_size;
1811 let spoofing = from_fn(
1812 move |req: axum::extract::Request, next: axum::middleware::Next| {
1813 crate::method::middleware(req, next, limit)
1814 },
1815 );
1816 Router::new()
1819 .fallback_service(tower::Layer::layer(&spoofing, router))
1820 .layer(from_fn_with_state(state, crate::security::middleware))
1821}
1822
1823fn has_public_file(
1825 state: &AppState,
1826 embedded_public: Option<&'static [(&'static str, &'static [u8])]>,
1827 name: &str,
1828) -> bool {
1829 match embedded_public {
1830 Some(files) => files.iter().any(|(path, _)| *path == name),
1831 None => state.config.public_path.join(name).is_file(),
1832 }
1833}
1834
1835fn robots(
1837 state: &AppState,
1838 embedded_public: Option<&'static [(&'static str, &'static [u8])]>,
1839) -> Router<AppState> {
1840 if has_public_file(state, embedded_public, "robots.txt") {
1841 Router::new()
1842 } else {
1843 crate::seo::robots_router(state)
1844 }
1845}
1846
1847fn favicon(
1851 state: &AppState,
1852 embedded_public: Option<&'static [(&'static str, &'static [u8])]>,
1853) -> Router<AppState> {
1854 if has_public_file(state, embedded_public, "favicon.ico") {
1855 return Router::new();
1856 }
1857 Router::new().route(
1858 "/favicon.ico",
1859 axum::routing::get(|| async {
1860 (
1861 axum::http::StatusCode::NO_CONTENT,
1862 [(axum::http::header::CACHE_CONTROL, "public, max-age=86400")],
1863 )
1864 }),
1865 )
1866}
1867
1868fn public_files(state: &AppState) -> Router<AppState> {
1870 match state.storage.public_root() {
1871 Some(root) => Router::new()
1872 .nest_service("/storage", ServeDir::new(root))
1873 .layer(axum::middleware::map_response(user_file_headers)),
1874 None => Router::new(),
1875 }
1876}
1877
1878pub(crate) async fn user_file_headers(
1883 mut res: axum::response::Response,
1884) -> axum::response::Response {
1885 use axum::http::HeaderValue;
1886 use axum::http::header::{
1887 CONTENT_DISPOSITION, CONTENT_SECURITY_POLICY, CONTENT_TYPE, X_CONTENT_TYPE_OPTIONS,
1888 };
1889
1890 let content_type = res
1891 .headers()
1892 .get(CONTENT_TYPE)
1893 .and_then(|v| v.to_str().ok())
1894 .unwrap_or_default()
1895 .to_ascii_lowercase();
1896 let headers = res.headers_mut();
1897 headers.insert(
1898 CONTENT_SECURITY_POLICY,
1899 HeaderValue::from_static(
1900 "default-src 'none'; img-src 'self' data:; style-src 'unsafe-inline'; sandbox",
1901 ),
1902 );
1903 headers.insert(X_CONTENT_TYPE_OPTIONS, HeaderValue::from_static("nosniff"));
1904 let document = ["html", "xml", "javascript", "ecmascript"]
1905 .iter()
1906 .any(|kind| content_type.contains(kind))
1907 && !content_type.contains("svg");
1908 if document {
1909 headers.insert(CONTENT_DISPOSITION, HeaderValue::from_static("attachment"));
1910 }
1911 res
1912}
1913
1914fn init_tracing(config: &Config, long_running: bool) {
1916 let default = match (long_running, config.debug) {
1917 (true, true) => "info,renox=debug",
1918 (true, false) => "info",
1919 (false, _) => "warn",
1920 };
1921 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(default));
1922 let file = config.log_file.as_ref().and_then(|path| {
1926 match std::fs::OpenOptions::new()
1927 .create(true)
1928 .append(true)
1929 .open(path)
1930 {
1931 Ok(file) => Some(std::sync::Mutex::new(file)),
1932 Err(err) => {
1933 eprintln!("LOG_FILE {}: {err}; logging to stdout", path.display());
1934 None
1935 }
1936 }
1937 });
1938 let json = config.log_format == crate::LogFormat::Json;
1939 let builder = tracing_subscriber::fmt().with_env_filter(filter);
1940 let _ = match (json, file) {
1941 (true, Some(file)) => builder
1942 .json()
1943 .with_current_span(true)
1944 .with_span_list(false)
1945 .with_writer(file)
1946 .try_init(),
1947 (true, None) => builder
1948 .json()
1949 .with_current_span(true)
1950 .with_span_list(false)
1951 .try_init(),
1952 (false, Some(file)) => builder.with_ansi(false).with_writer(file).try_init(),
1953 (false, None) => builder.try_init(),
1954 };
1955}
1956
1957async fn shutdown_signal() {
1958 let ctrl_c = async {
1959 let _ = tokio::signal::ctrl_c().await;
1960 };
1961
1962 #[cfg(unix)]
1963 let terminate = async {
1964 match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
1965 Ok(mut signal) => {
1966 signal.recv().await;
1967 }
1968 Err(_) => std::future::pending().await,
1969 }
1970 };
1971 #[cfg(not(unix))]
1972 let terminate = std::future::pending::<()>();
1973
1974 tokio::select! {
1975 _ = ctrl_c => {},
1976 _ = terminate => {},
1977 }
1978 tracing::info!("shutdown signal received");
1979}
1980
1981#[cfg(test)]
1982mod tests {
1983 use super::*;
1984
1985 fn route(path: &str, middleware: &[&str]) -> RouteInfo {
1986 RouteInfo {
1987 method: "GET".into(),
1988 path: path.into(),
1989 name: None,
1990 module: "app".into(),
1991 middleware: middleware.iter().map(|m| (*m).to_owned()).collect(),
1992 domain: None,
1993 }
1994 }
1995
1996 #[test]
1997 fn route_list_says_role_guards_count_the_active_scope() {
1998 let plain = route_table(&[route("/", &["auth"])]);
1999 assert!(!plain.contains('*'), "{plain}");
2000 let table = route_table(&[
2001 route("/orders", &["auth", "permission:orders.view"]),
2002 route("/drafts", &["role:editor"]),
2003 route("/admin", &["gate:admin"]),
2004 ]);
2005 assert!(table.contains("auth, permission:orders.view*"), "{table}");
2006 assert!(table.contains("role:editor*"), "{table}");
2007 assert!(table.contains("gate:admin\n"), "{table}");
2008 assert!(
2009 table.contains("active scope (permissions::set_scope)"),
2010 "{table}"
2011 );
2012 }
2013}