1use std::collections::BTreeSet;
10use std::collections::HashMap;
11use std::sync::Arc;
12use std::sync::atomic::{AtomicBool, Ordering};
13use std::time::Duration;
14
15use axum::Router;
16use axum::extract::{DefaultBodyLimit, State};
17use axum::http::{HeaderMap, Method, StatusCode, header};
18use axum::middleware::{Next, from_fn_with_state};
19use axum::response::{IntoResponse, Response};
20use axum::routing::{get, post};
21use bytes::Bytes;
22use serde_json::json;
23use tracing::error;
24
25use crate::admin::require_admin;
26use crate::config::Config;
27use crate::config::ConfigError;
28use crate::events::EventBus;
29use crate::http::{
30 Json, MAX_BODY_BYTES, ScopeState, cors_layer, rate_limited, scope_layer,
31 security_headers_layer, token_response_layer,
32};
33use crate::module::{HARNESS_API, Module, ModuleContext, harness_api_mismatch};
34use crate::ports::Dispatcher;
35use crate::ports::{
36 Clock, Database, Port, Ports, RateLimiter, Statement, SystemClock, warn_undeclared_ports,
37};
38use crate::problem::Problem;
39use crate::problems::SLUGS;
40use crate::rate_limit::{RateLimit, RateLimitFailure};
41use crate::route_policy::deployed_env;
42use crate::scheduled::ScheduledBudget;
43use crate::scope::Scope;
44use crate::sidecar::{
45 GatewayGuard, SIDECAR_REQUIRE_GATEWAY, SidecarMount, X_HARNESS_GATEWAY, gateway_guard,
46 gateway_signer, mint_gateway, truthy,
47};
48use crate::signer::HmacSigner;
49use crate::stream::{RequestStream, StreamRoute};
50use crate::surface::{
51 MAX_SIDECAR_SURFACE_BYTES, RenderedSurface, SurfaceDocument, SurfaceSource, UiContext, UiMount,
52 sanitize_sidecar_document, strip_unguarded_captcha_actions,
53};
54use crate::template::{Template, TemplateRegistry};
55use crate::venture::{Venture, VentureEnv};
56
57fn check_tables(
66 module: &dyn Module,
67 tables: &mut HashMap<&'static str, &'static str>,
68 errors: &mut ConfigError,
69) {
70 let name = module.name();
71 for table in module.tables() {
72 match tables.get(table) {
73 Some(owner) => errors.push(format!(
74 "duplicate table `{table}` claimed by modules `{owner}` and `{name}`"
75 )),
76 None => {
77 tables.insert(table, name);
78 }
79 }
80 }
81
82 let owns = module.tables();
86 let mut declared: Vec<&'static str> = Vec::new();
87 for set in module.personal_data() {
88 for problem in set.validate(name, owns) {
89 errors.push(problem);
90 }
91 if declared.contains(&set.table) {
92 errors.push(format!(
93 "module `{name}` declares table `{}` twice in personal_data()",
94 set.table
95 ));
96 }
97 declared.push(set.table);
98 }
99}
100
101pub trait Runtime: Send + Sync + 'static {
106 fn provides(&self) -> Vec<Port>;
107 fn effectively_configured(&self, port: Port) -> bool {
113 self.provides().contains(&port)
114 }
115}
116
117pub struct Harness {
119 venture: Arc<Venture>,
120 modules: Vec<Arc<dyn Module>>,
121 templates: Arc<TemplateRegistry>,
122 events: EventBus,
123 runtime: Option<Arc<dyn Runtime>>,
124 well_known: Option<Router>,
127 surface: Arc<SurfaceVariants>,
130 ui: Option<Arc<dyn UiMount>>,
132 personal_data: Arc<crate::PersonalDataCatalog>,
135 unprotected_acceptance_recorded: AtomicBool,
145 unlimited_acceptance_recorded: AtomicBool,
146}
147
148struct SurfaceVariants {
149 document: Arc<SurfaceDocument>,
150 full: RenderedSurface,
151 public: RenderedSurface,
152}
153
154impl SurfaceVariants {
155 fn compose(
156 venture: &Venture,
157 modules: &[Arc<dyn Module>],
158 ui: Option<&Arc<dyn UiMount>>,
159 ) -> Self {
160 let mut document = SurfaceDocument::compose(venture, modules);
161 document.ui = ui.and_then(|ui| ui.describe());
162 Self {
163 full: RenderedSurface::render(&document),
164 public: RenderedSurface::render(&document.public()),
165 document: Arc::new(document),
166 }
167 }
168}
169
170impl std::fmt::Debug for Harness {
171 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
172 let modules: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
173 f.debug_struct("Harness")
174 .field("venture", &self.venture.name)
175 .field("modules", &modules)
176 .finish_non_exhaustive()
177 }
178}
179
180impl Harness {
181 #[must_use]
185 pub fn clone_arc(self: &std::sync::Arc<Self>) -> std::sync::Arc<Self> {
186 std::sync::Arc::clone(self)
187 }
188 pub fn builder() -> HarnessBuilder {
189 HarnessBuilder::default()
190 }
191
192 pub fn venture(&self) -> &Arc<Venture> {
193 &self.venture
194 }
195
196 pub fn modules(&self) -> &[Arc<dyn Module>] {
197 &self.modules
198 }
199
200 pub fn max_body_bytes(&self, path: &str, cfg: &dyn Config) -> usize {
223 let Some((name, _)) = module_and_segments(path) else {
224 return MAX_BODY_BYTES;
225 };
226 self.modules
227 .iter()
228 .find(|module| module.name() == name)
229 .map_or(MAX_BODY_BYTES, |module| {
230 MAX_BODY_BYTES.max(module.max_body_bytes(cfg))
231 })
232 }
233
234 pub fn streaming_route(&self, path: &str, method: &axum::http::Method) -> Option<usize> {
252 let (name, segments) = exact_module_and_segments(path)?;
253 let module = self.modules.iter().find(|module| module.name() == name)?;
254 match_streaming(module.streaming_routes(), method, &segments)
255 }
256
257 pub fn templates(&self) -> &Arc<TemplateRegistry> {
258 &self.templates
259 }
260
261 pub fn events(&self) -> &EventBus {
262 &self.events
263 }
264
265 pub fn module_context(&self, module: &dyn Module, ports: &Ports) -> ModuleContext {
270 let venture = self.venture_as_deployed(ports.config.as_ref());
285 ModuleContext {
286 config: Arc::clone(&ports.config),
287 ports: ports.view_for(module),
288 events: self.events_for(ports),
289 templates: Arc::clone(&self.templates),
290 venture,
291 unprotected_writes_accepted: crate::route_policy::unprotected_writes_override(
292 ports.config.as_ref(),
293 )
294 .is_some(),
295 personal_data: Arc::clone(&self.personal_data),
296 ui_mounted: self.ui.is_some(),
297 scheduled: Arc::new(ScheduledBudget::unbounded()),
301 }
302 }
303
304 fn gateway_guard_state(ports: &Ports, gateway: Option<&Arc<HmacSigner>>) -> GatewayGuard {
314 GatewayGuard {
315 require: truthy(ports.config.as_ref(), SIDECAR_REQUIRE_GATEWAY),
316 signer: gateway.map(Arc::clone),
317 admin_token: ports
318 .config
319 .get("ADMIN_TOKEN")
320 .filter(|token| !token.is_empty()),
321 }
322 }
323
324 fn merged_surface(
327 &self,
328 mounts: Vec<SidecarMount>,
329 ports: &Ports,
330 gateway: Option<Arc<HmacSigner>>,
331 env: VentureEnv,
332 ) -> Arc<dyn SurfaceSource> {
333 Arc::new(MergedSurface {
334 base: Arc::clone(&self.surface),
335 mounts,
336 dispatcher: ports.dispatcher.clone(),
337 gateway,
338 env,
339 captcha_configured: ports.captcha.is_some(),
340 })
341 }
342
343 fn venture_as_deployed(&self, config: &dyn Config) -> Arc<Venture> {
349 let env = deployed_env(self.venture.env, config);
350 if env == self.venture.env {
351 return Arc::clone(&self.venture);
352 }
353 let mut deployed = (*self.venture).clone();
354 deployed.env = env;
355 Arc::new(deployed)
356 }
357
358 fn nest_modules(&self, ports: &Ports) -> Router {
364 let mut api = Router::new();
365 for module in &self.modules {
366 let ctx = self.module_context(module.as_ref(), ports);
367 api = api.nest(&format!("/v1/{}", module.name()), module.router(ctx));
368 }
369 api
370 }
371
372 fn production_readiness_now(
388 &self,
389 env: VentureEnv,
390 config: &dyn Config,
391 rate_limiter_ready: bool,
392 signer_ready: bool,
393 ) -> Vec<String> {
394 if let Some(note) = crate::route_policy::env_disagreement(self.venture.env, env) {
395 tracing::warn!("{note}");
396 crate::logging::forward_control_event(crate::logging::ControlLevel::Warn, ¬e);
397 }
398 let unprotected = crate::route_policy::unprotected_writes_override(config);
402 let unlimited = crate::route_policy::unlimited_public_routes_override(config);
403 let guards = crate::route_policy::WriteGuards::collect(&self.modules);
404 let mut problems = crate::route_policy::production_readiness(
405 env,
406 &guards,
407 self.runtime.as_ref(),
408 rate_limiter_ready,
409 signer_ready,
410 unprotected.as_deref(),
411 unlimited.as_deref(),
412 );
413 let hmac_problems = crate::route_policy::webhook_secret_readiness(env, &guards, config);
418 if problems.is_empty() && hmac_problems.is_empty() {
419 let accepted = crate::route_policy::production_readiness(
428 env,
429 &guards,
430 self.runtime.as_ref(),
431 rate_limiter_ready,
432 signer_ready,
433 None,
434 None,
435 );
436 self.record_acceptances_once(
437 &guards,
438 rate_limiter_ready,
439 unprotected.as_deref(),
440 unlimited.as_deref(),
441 &accepted,
442 );
443 return problems;
444 }
445 for problem in &problems {
453 let detail = format!(
454 "refusing guarded routes: {problem} — set {} to a reason to accept this \
455 explicitly while the port is wired",
456 crate::route_policy::ALLOW_UNPROTECTED_WRITES
457 );
458 error!("{detail}");
459 crate::logging::forward_control_event(crate::logging::ControlLevel::Error, &detail);
460 }
461 problems.extend(hmac_problems);
462 problems
463 }
464
465 fn record_acceptances_once(
471 &self,
472 guards: &crate::route_policy::WriteGuards,
473 rate_limiter_ready: bool,
474 unprotected: Option<&str>,
475 unlimited: Option<&str>,
476 accepted: &[String],
477 ) {
478 if guards.needs_captcha()
479 && !crate::route_policy::captcha_effective(self.runtime.as_ref())
480 && let Some(reason) = unprotected
481 && !self
482 .unprotected_acceptance_recorded
483 .swap(true, Ordering::Relaxed)
484 {
485 let detail = format!(
486 "serving guarded routes unprotected on an operator's recorded acceptance \
487 (reason: {reason}; problems: {})",
488 accepted.join("; ")
489 );
490 tracing::warn!(
491 control = "production-readiness",
492 acceptance = crate::route_policy::ALLOW_UNPROTECTED_WRITES,
493 reason,
494 "serving captcha-guarded routes unprotected on an operator's recorded acceptance"
495 );
496 crate::logging::forward_control_event(crate::logging::ControlLevel::Warn, &detail);
500 }
501 if guards.needs_rate_limiter()
502 && !rate_limiter_ready
503 && let Some(reason) = unlimited
504 && !self
505 .unlimited_acceptance_recorded
506 .swap(true, Ordering::Relaxed)
507 {
508 let detail = format!(
509 "serving public writes and admin routes without a rate limiter on an \
510 operator's recorded acceptance (reason: {reason}; problems: {})",
511 accepted.join("; ")
512 );
513 tracing::warn!(
514 control = "production-readiness",
515 acceptance = crate::route_policy::ALLOW_UNLIMITED_PUBLIC_ROUTES,
516 reason,
517 "serving public writes and admin routes without a rate limiter on an \
518 operator's recorded acceptance"
519 );
520 crate::logging::forward_control_event(crate::logging::ControlLevel::Warn, &detail);
522 }
523 }
524
525 fn nest_sidecars(
528 mut api: Router,
529 mounted: &[SidecarMount],
530 ports: &Ports,
531 gateway: Option<&Arc<HmacSigner>>,
532 ) -> Router {
533 for mount in mounted {
534 api = api.nest(
535 &format!("/v1/{}", mount.name),
536 crate::sidecar::router(
537 mount.clone(),
538 ports.dispatcher.clone(),
539 Arc::clone(&ports.config),
540 ports.rate_limiter.clone(),
541 gateway.map(Arc::clone),
542 ),
543 );
544 }
545 api
546 }
547
548 fn sidecar_mounts(&self, ports: &Ports) -> Vec<SidecarMount> {
549 let mounts = match crate::sidecar::SidecarMounts::from_config(ports.config.as_ref()) {
550 Ok(mounts) => mounts,
551 Err(errors) => {
552 for error in errors {
553 let detail = format!("ignoring the sidecar mount table: {error}");
554 tracing::error!(error, "ignoring the sidecar mount table");
555 crate::logging::forward_control_event(
558 crate::logging::ControlLevel::Error,
559 &detail,
560 );
561 }
562 crate::sidecar::SidecarMounts::default()
563 }
564 };
565 let module_names: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
566 for collision in mounts.collisions(&module_names) {
567 let detail = format!("ignoring the colliding sidecar mount: {collision}");
568 tracing::error!(error = collision, "ignoring the colliding sidecar mount");
569 crate::logging::forward_control_event(crate::logging::ControlLevel::Error, &detail);
570 }
571 mounts
572 .iter()
573 .filter(|m| !module_names.contains(&m.name.as_str()))
574 .cloned()
575 .collect()
576 }
577
578 fn resolved_mounts(&self, ports: &Ports) -> Vec<SidecarMount> {
582 let mounts =
583 crate::sidecar::SidecarMounts::from_config(ports.config.as_ref()).unwrap_or_default();
584 let module_names: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
585 mounts
586 .iter()
587 .filter(|m| !module_names.contains(&m.name.as_str()))
588 .cloned()
589 .collect()
590 }
591
592 fn events_for(&self, ports: &Ports) -> EventBus {
600 let mounts = self.resolved_mounts(ports);
601 if mounts.is_empty() {
602 return self.events.clone();
603 }
604 let Some(dispatcher) = ports.dispatcher.clone() else {
605 return self.events.clone();
606 };
607 let gateway = gateway_signer(
608 ports.config.as_ref(),
609 ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
610 );
611 self.events
612 .clone()
613 .forwarding_to(crate::sidecar::EventForwarder::new(
614 mounts, dispatcher, gateway,
615 ))
616 }
617
618 fn ui_router(
623 &self,
624 api: &Router,
625 ports: &Ports,
626 surface: &Arc<dyn SurfaceSource>,
627 ) -> Option<Router> {
628 self.ui.as_ref().map(|ui| {
629 ui.router(UiContext {
630 surface: Arc::clone(surface),
631 api: api.clone(),
632 config: Arc::clone(&ports.config),
633 venture: Arc::clone(&self.venture),
634 captcha_configured: ports.captcha.is_some(),
635 signer: ports.signer.clone(),
636 rate_limiter: ports.rate_limiter.clone(),
637 })
638 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
639 })
640 }
641
642 fn root_router(
648 &self,
649 api: Router,
650 ui: Option<Router>,
651 gateway: Option<Arc<HmacSigner>>,
652 health_state: HealthState,
653 ready_state: ReadyState,
654 surface_state: SurfaceState,
655 ) -> Router {
656 let root = Router::new()
657 .route("/__health", get(health_handler))
658 .with_state(health_state)
659 .route("/__ready", get(ready_handler))
660 .with_state(ready_state)
661 .route("/__surface", get(surface_handler))
662 .with_state(surface_state)
663 .merge(self.inbound_events_route(gateway).unwrap_or_default())
666 .merge(api);
667 let root = match &self.well_known {
668 Some(well_known) => root.nest(
669 "/.well-known",
670 well_known
671 .clone()
672 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)),
673 ),
674 None => root,
675 };
676 match ui {
677 Some(ui) => root.nest("/ui", ui),
678 None => root,
679 }
680 }
681
682 fn inbound_events_route(&self, gateway: Option<Arc<HmacSigner>>) -> Option<Router> {
696 gateway.as_ref()?;
697 Some(
698 Router::new()
699 .route(
700 "/__events",
701 post(crate::sidecar::events_inbound)
702 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)),
703 )
704 .with_state(Arc::new(crate::sidecar::InboundEvents {
705 bus: self.events.clone(),
706 gateway,
707 })),
708 )
709 }
710
711 pub fn runtime(&self) -> Option<&Arc<dyn Runtime>> {
713 self.runtime.as_ref()
714 }
715
716 #[must_use]
720 pub fn surface(&self) -> &SurfaceDocument {
721 &self.surface.document
722 }
723
724 fn health_state(&self, ports: &Ports, sidecars: Vec<SidecarMount>) -> HealthState {
738 HealthState {
739 venture: self.venture_as_deployed(ports.config.as_ref()),
744 modules: self.modules.clone(),
745 harness_build: ports.config.get("HARNESS_BUILD").filter(|b| !b.is_empty()),
746 mailer_configured: ports.mailer.is_some(),
747 captcha_configured: ports.captcha.is_some(),
748 sidecars,
749 dispatcher: ports.dispatcher.clone(),
750 clock: ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
751 probe_cache: new_probe_cache(),
752 }
753 }
754
755 pub fn router(&self, ports: Ports) -> Router {
756 let env = deployed_env(self.venture.env, ports.config.as_ref());
761 let (limiter, signer) = (ports.rate_limiter.is_some(), ports.signer.is_some());
762 let readiness = self.production_readiness_now(env, ports.config.as_ref(), limiter, signer);
763 let problem_type_base: Arc<str> = self.venture.problem_type_base().into();
767 let mut api = self.nest_modules(&ports);
768 let gateway = gateway_signer(
771 ports.config.as_ref(),
772 ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
773 );
774 let mounted = self.sidecar_mounts(&ports);
775 api = Self::nest_sidecars(api, &mounted, &ports, gateway.as_ref());
776 let gateway_state = Self::gateway_guard_state(&ports, gateway.as_ref());
777 let mounts_for_health = mounted.clone();
778 let tenant_layer = Arc::new(TenantLayer {
780 routing: ports.tenants.clone(),
781 fallback: ports.db.clone(),
782 });
783 let streaming = StreamingState::build(&self.modules);
805 let api = api.layer(from_fn_with_state(streaming, streaming_request_layer));
806 let api = with_abuse_floor(api, ports.rate_limiter.clone(), tenant_layer);
807
808 let surface_source = self.merged_surface(mounted, &ports, gateway.clone(), env);
809
810 let ui = self.ui_router(&api, &ports, &surface_source);
811
812 let health_state = self.health_state(&ports, mounts_for_health);
813 let missing_ports = missing_module_ports(&self.modules, &ports);
817 let Ports {
818 config,
819 db,
820 auth: _,
821 mailer: _,
822 captcha: _,
823 rate_limiter: _,
824 signer: _,
825 kv: _,
826 blob: _,
827 push: _,
828 payments: _,
829 tracker: _,
830 realtime: _,
831 text_model: _,
832 classifier: _,
833 vector_index: _,
834 embedder: _,
835 custom_hostnames: _,
836 http: _,
837 clock,
838 id_gen,
839 defer,
840 dispatcher: _,
841 tenants: _,
842 } = ports;
843
844 let scope_state = ScopeState {
845 defer: defer.unwrap_or_else(|| Arc::new(crate::ports::NoopDefer)),
846 id_gen: id_gen.unwrap_or_else(|| Arc::new(crate::ports::UlidIdGen)),
847 module: (self.modules.len() == 1).then(|| self.modules[0].name()),
850 };
851 let ready_state = ReadyState {
852 db,
853 clock: clock.unwrap_or_else(|| Arc::new(SystemClock)),
854 missing_ports,
855 };
856
857 let surface_state = SurfaceState {
858 config,
859 source: surface_source,
860 };
861
862 let root = self.root_router(api, ui, gateway, health_state, ready_state, surface_state);
863
864 root.layer(from_fn_with_state(
873 Arc::new(readiness),
874 production_readiness_guard,
875 ))
876 .layer(from_fn_with_state(gateway_state, gateway_guard))
877 .layer(from_fn_with_state(scope_state, scope_layer))
878 .layer(axum::middleware::from_fn(token_response_layer))
879 .layer(cors_layer(&self.venture.cors_origins))
880 .layer(from_fn_with_state(problem_type_base, problem_type_layer))
889 }
890}
891
892fn module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
900 let path = match path.split_once(['?', '#']) {
901 Some((path, _)) => path,
902 None => path,
903 };
904 let mut segments = path.split('/').filter(|segment| !segment.is_empty());
905 if segments.next() != Some("v1") {
906 return None;
907 }
908 let name = segments.next()?;
909 Some((name, segments.collect()))
910}
911
912fn exact_module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
919 let path = match path.split_once(['?', '#']) {
920 Some((path, _)) => path,
921 None => path,
922 };
923 let mut segments = path.split('/');
924 if segments.next() != Some("") || segments.next() != Some("v1") {
926 return None;
927 }
928 let name = segments.next()?;
929 if name.is_empty() {
930 return None;
931 }
932 Some((name, segments.collect()))
933}
934
935fn match_streaming(routes: &[StreamRoute], method: &Method, segments: &[&str]) -> Option<usize> {
938 routes.iter().find_map(|route| {
939 (route.method == *method && pattern_matches(route.path, segments))
940 .then_some(route.max_bytes)
941 })
942}
943
944fn pattern_matches(pattern: &str, segments: &[&str]) -> bool {
948 let mut pattern_segments = pattern.split('/').filter(|segment| !segment.is_empty());
949 let mut rest = segments.iter();
950 loop {
951 match pattern_segments.next() {
952 Some("{*rest}") => {
953 return pattern_segments.next().is_none() && rest.next().is_some();
954 }
955 Some(param) if param.starts_with('{') && param.ends_with('}') => {
956 if rest.next().is_none() {
957 return false;
958 }
959 }
960 Some(literal) => {
961 if rest.next() != Some(&literal) {
962 return false;
963 }
964 }
965 None => return rest.next().is_none(),
966 }
967 }
968}
969
970#[derive(Clone)]
973struct StreamingState {
974 modules: Arc<Vec<(&'static str, &'static [StreamRoute])>>,
975}
976
977impl StreamingState {
978 fn build(modules: &[Arc<dyn Module>]) -> Self {
979 let routes = modules
982 .iter()
983 .filter(|module| !module.streaming_routes().is_empty())
984 .map(|module| (module.name(), module.streaming_routes()))
985 .collect();
986 Self {
987 modules: Arc::new(routes),
988 }
989 }
990
991 fn max_bytes(&self, path: &str, method: &Method) -> Option<usize> {
992 let (name, segments) = exact_module_and_segments(path)?;
993 let (_, routes) = self.modules.iter().find(|(module, _)| *module == name)?;
994 match_streaming(routes, method, &segments)
995 }
996}
997
998async fn streaming_request_layer(
1010 State(state): State<StreamingState>,
1011 mut request: axum::extract::Request,
1012 next: Next,
1013) -> Response {
1014 if state.modules.is_empty() {
1015 return next.run(request).await;
1016 }
1017 let Some(max_bytes) = state.max_bytes(request.uri().path(), request.method()) else {
1018 return next.run(request).await;
1019 };
1020 if request.extensions().get::<RequestStream>().is_some() {
1021 return next.run(request).await;
1022 }
1023 if let Some(declared) = crate::ports::declared_content_length(request.headers())
1026 && declared > max_bytes
1027 {
1028 let problem = Problem::request_too_large();
1029 let problem = match request.extensions().get::<Scope>() {
1030 Some(scope) => problem.instance(&scope.request_id),
1031 None => problem,
1032 };
1033 return problem.into_response();
1034 }
1035 let body = std::mem::replace(request.body_mut(), axum::body::Body::empty());
1036 request
1037 .extensions_mut()
1038 .insert(RequestStream::new(body.into_data_stream(), max_bytes));
1039 next.run(request).await
1040}
1041
1042fn with_abuse_floor(
1047 api: Router,
1048 rate_limiter: Option<Arc<dyn RateLimiter>>,
1049 tenant_layer: Arc<TenantLayer>,
1050) -> Router {
1051 api.layer(from_fn_with_state(rate_limiter, admin_rate_limit_layer))
1052 .layer(axum::middleware::from_fn_with_state(
1053 tenant_layer,
1054 resolve_tenant_layer,
1055 ))
1056 .layer(axum::middleware::from_fn(security_headers_layer))
1057 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
1058}
1059
1060async fn problem_type_layer(
1065 State(base): State<Arc<str>>,
1066 request: axum::extract::Request,
1067 next: Next,
1068) -> Response {
1069 let mut response = next.run(request).await;
1070 let Some(problem) = response.extensions().get::<Problem>().cloned() else {
1071 return response;
1072 };
1073 let body = problem.body(&base).to_string();
1074 *response.body_mut() = axum::body::Body::from(body);
1075 response.headers_mut().remove(header::CONTENT_LENGTH);
1076 response.extensions_mut().remove::<Problem>();
1080 response
1081}
1082
1083type ProbeCache = Option<(i64, Vec<serde_json::Value>)>;
1086
1087#[allow(clippy::disallowed_types)]
1090fn new_probe_cache() -> Arc<std::sync::Mutex<ProbeCache>> {
1091 Arc::new(std::sync::Mutex::new(None))
1092}
1093
1094#[derive(Clone)]
1095struct HealthState {
1096 venture: Arc<Venture>,
1097 modules: Vec<Arc<dyn Module>>,
1098 harness_build: Option<String>,
1100 mailer_configured: bool,
1101 captcha_configured: bool,
1102 sidecars: Vec<SidecarMount>,
1103 dispatcher: Option<Arc<dyn Dispatcher>>,
1104 clock: Arc<dyn Clock>,
1105 #[allow(clippy::disallowed_types)]
1113 probe_cache: Arc<std::sync::Mutex<ProbeCache>>,
1114}
1115
1116const SIDECAR_PROBE_TTL_SECS: i64 = 30;
1119
1120const SIDECAR_PROBE_TIMEOUT: Duration = Duration::from_secs(2);
1123
1124async fn health_handler(State(state): State<HealthState>) -> impl IntoResponse {
1125 let modules: Vec<serde_json::Value> = state
1126 .modules
1127 .iter()
1128 .map(|module| {
1129 json!({
1130 "name": module.name(),
1131 "version": module.version(),
1132 "requires": module.requires().iter().map(Port::name).collect::<Vec<_>>(),
1133 "optional": module.optional().iter().map(Port::name).collect::<Vec<_>>(),
1134 "tables": module.tables(),
1135 "emits": module.emits(),
1136 })
1137 })
1138 .collect();
1139 let sidecars = probe_sidecars(&state).await;
1140 Json(json!({
1141 "venture": state.venture.name,
1142 "env": state.venture.env.as_str(),
1143 "harness_api": HARNESS_API,
1144 "harness_build": state.harness_build,
1145 "mailer": if state.mailer_configured { "configured" } else { "not_configured" },
1148 "captcha": if state.captcha_configured { "configured" } else { "absent" },
1149 "modules": modules,
1150 "sidecars": sidecars,
1151 }))
1152}
1153
1154async fn probe_sidecars(state: &HealthState) -> Vec<serde_json::Value> {
1160 if state.sidecars.is_empty() {
1161 return Vec::new();
1162 }
1163 let now_ms =
1164 i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1165 if let Some((at_ms, entries)) = state.probe_cache.lock().unwrap().as_ref()
1166 && now_ms - at_ms < SIDECAR_PROBE_TTL_SECS * 1000
1167 {
1168 return entries.clone();
1169 }
1170
1171 let mut entries = Vec::new();
1172 for mount in &state.sidecars {
1173 entries.push(probe_one_sidecar(state, mount).await);
1174 }
1175 *state.probe_cache.lock().unwrap() = Some((now_ms, entries.clone()));
1176 entries
1177}
1178
1179async fn probe_one_sidecar(state: &HealthState, mount: &SidecarMount) -> serde_json::Value {
1180 let mut missing = json!({
1181 "name": mount.name,
1182 "binding": mount.binding,
1183 "probe": "unreachable",
1184 "contract": serde_json::Value::Null,
1185 "module": serde_json::Value::Null,
1186 "version": serde_json::Value::Null,
1187 });
1188 let Some(dispatcher) = state.dispatcher.as_ref() else {
1189 return missing;
1190 };
1191 let request = http::Request::builder()
1192 .method(http::Method::GET)
1193 .uri("/__health")
1194 .body(Bytes::new());
1195 let Ok(request) = request else {
1196 return missing;
1197 };
1198 let dispatcher = Arc::clone(dispatcher);
1203 let binding = mount.binding.clone();
1204 let probe = async move { dispatcher.dispatch(&binding, request).await };
1205 let Some(Ok(response)) =
1206 crate::ports::timeout(&*state.clock, probe, SIDECAR_PROBE_TIMEOUT).await
1207 else {
1208 return missing;
1209 };
1210 let contract = response
1211 .headers()
1212 .get(crate::sidecar::X_HARNESS_API)
1213 .and_then(|v| v.to_str().ok())
1214 .and_then(|v| v.parse::<u32>().ok());
1215 let body = serde_json::from_slice::<serde_json::Value>(response.body()).ok();
1216 let module = response
1217 .headers()
1218 .get(crate::sidecar::X_HARNESS_MODULE)
1219 .and_then(|v| v.to_str().ok())
1220 .map(str::to_owned)
1221 .or_else(|| first_module_string(body.as_ref(), "name"));
1222 let version = first_module_string(body.as_ref(), "version");
1223 let tables = declared_tables(body.as_ref());
1224 let contract = contract.or_else(|| {
1225 body.as_ref()
1226 .and_then(|b| b.get("harness_api"))
1227 .and_then(serde_json::Value::as_u64)
1228 .and_then(|api| u32::try_from(api).ok())
1229 });
1230 missing["contract"] = contract.map_or(serde_json::Value::Null, |api| json!(api));
1231 missing["module"] = module.map_or(serde_json::Value::Null, |name| json!(name));
1232 missing["version"] = version.map_or(serde_json::Value::Null, |version| json!(version));
1233 missing["tables"] = json!(tables);
1234 match contract {
1235 Some(api) if api != HARNESS_API => missing["probe"] = json!("mismatch"),
1236 Some(_) => missing["probe"] = json!("ok"),
1237 None => missing["probe"] = json!("unreachable"),
1238 }
1239 let clashes = collisions_with(&tables, state, &mount.name);
1251 if !clashes.is_empty() {
1252 let detail = format!(
1253 "sidecar `{}` claims table(s) {} already claimed in this deployment",
1254 mount.name,
1255 clashes.join(", ")
1256 );
1257 tracing::error!(sidecar = %mount.name, tables = %clashes.join(","), "{detail}");
1258 crate::logging::forward_internal_error(&detail);
1259 missing["probe"] = json!("table-collision");
1260 missing["table_collisions"] = json!(clashes);
1261 }
1262 missing
1263}
1264
1265fn declared_tables(body: Option<&serde_json::Value>) -> Vec<String> {
1269 let Some(modules) = body
1270 .and_then(|b| b.get("modules"))
1271 .and_then(|m| m.as_array())
1272 else {
1273 return Vec::new();
1274 };
1275 let mut tables: Vec<String> = modules
1276 .iter()
1277 .filter_map(|module| module.get("tables")?.as_array())
1278 .flatten()
1279 .filter_map(|table| table.as_str().map(str::to_owned))
1280 .collect();
1281 tables.sort_unstable();
1282 tables.dedup();
1283 tables
1284}
1285
1286fn collisions_with(claimed: &[String], state: &HealthState, mount: &str) -> Vec<String> {
1289 let mut clashes: Vec<String> = Vec::new();
1290 for module in &state.modules {
1291 if module.name() == mount {
1292 continue;
1293 }
1294 for table in module.tables() {
1295 if claimed.iter().any(|claimed| claimed == table)
1296 && !clashes.iter().any(|seen| seen == table)
1297 {
1298 clashes.push(format!("`{table}` (also module `{}`)", module.name()));
1299 }
1300 }
1301 }
1302 clashes
1303}
1304
1305fn first_module_string(body: Option<&serde_json::Value>, key: &str) -> Option<String> {
1308 body?
1309 .get("modules")?
1310 .get(0)?
1311 .get(key)?
1312 .as_str()
1313 .map(str::to_owned)
1314}
1315
1316struct TenantLayer {
1319 routing: Option<Arc<dyn crate::tenant::TenantRouting>>,
1320 fallback: Option<Arc<dyn Database>>,
1326}
1327
1328async fn resolve_tenant_layer(
1335 State(layer): State<Arc<TenantLayer>>,
1336 scope: Scope,
1337 mut request: axum::extract::Request,
1338 next: Next,
1339) -> Response {
1340 let host = request
1341 .headers()
1342 .get(header::HOST)
1343 .and_then(|value| value.to_str().ok())
1344 .unwrap_or_default()
1345 .to_owned();
1346
1347 let resolution = match layer.routing.as_ref() {
1352 Some(routing) => routing.resolve(&host),
1353 None => <crate::tenant::ImplicitTenant as crate::tenant::ResolveTenant>::resolve(
1354 &crate::tenant::ImplicitTenant,
1355 &host,
1356 ),
1357 };
1358
1359 let tenant = match resolution.admit() {
1360 Ok(tenant) => tenant,
1361 Err(problem) => {
1362 return Problem::new(problem)
1363 .instance(&scope.request_id)
1364 .into_response();
1365 }
1366 };
1367
1368 let db = match layer.routing.as_ref() {
1369 Some(routing) => match routing.database(&tenant).await {
1370 Ok(db) => Some(db),
1371 Err(err) => {
1372 tracing::error!(tenant = %tenant.id(), "{err}");
1375 crate::logging::forward_internal_error(&err.to_string());
1376 return Problem::new(&SLUGS.tenant_degraded)
1377 .instance(&scope.request_id)
1378 .into_response();
1379 }
1380 },
1381 None => layer.fallback.clone(),
1382 };
1383
1384 if let Some(db) = db {
1385 let tenancy = if layer.routing.is_some() {
1390 crate::tenant::Tenancy::FromRegistry
1391 } else {
1392 crate::tenant::Tenancy::Sole
1393 };
1394 request
1395 .extensions_mut()
1396 .insert(crate::tenant_conn::ResolvedTenant {
1397 tenant,
1398 tenancy,
1399 db,
1400 });
1401 }
1402 next.run(request).await
1403}
1404
1405#[derive(Clone)]
1406struct ReadyState {
1407 db: Option<Arc<dyn Database>>,
1408 clock: Arc<dyn Clock>,
1409 missing_ports: Vec<Port>,
1413}
1414
1415fn missing_module_ports(modules: &[Arc<dyn Module>], ports: &Ports) -> Vec<Port> {
1420 const PROBED: &[Port] = &[Port::VectorIndex, Port::Embedder];
1421 PROBED
1422 .iter()
1423 .copied()
1424 .filter(|port| {
1425 modules
1426 .iter()
1427 .any(|module| module.requires().contains(port))
1428 && !ports.has(*port)
1429 })
1430 .collect()
1431}
1432
1433async fn ready_handler(State(state): State<ReadyState>) -> impl IntoResponse {
1437 if !state.missing_ports.is_empty() {
1438 let detail = state
1439 .missing_ports
1440 .iter()
1441 .map(|port| format!("{} port is not configured", port.name()))
1442 .collect::<Vec<_>>()
1443 .join("; ");
1444 return Problem::not_ready(detail).into_response();
1445 }
1446 let Some(db) = state.db else {
1447 return Problem::not_ready("database port is not configured").into_response();
1448 };
1449 let stmt = Statement::new("SELECT 1");
1450 let query = async move { db.query(&stmt).await };
1451 match crate::ports::timeout(&*state.clock, query, Duration::from_secs(2)).await {
1452 Some(Ok(_rows)) => Json(json!({ "ok": true })).into_response(),
1453 Some(Err(err)) => {
1454 error!(error = %err, "readiness probe query failed");
1455 crate::logging::forward_internal_error(&format!("readiness probe query failed: {err}"));
1456 Problem::not_ready("database query failed").into_response()
1457 }
1458 None => Problem::not_ready("database did not answer within 2 s").into_response(),
1459 }
1460}
1461
1462#[derive(Clone)]
1463struct SurfaceState {
1464 config: Arc<dyn Config>,
1465 source: Arc<dyn SurfaceSource>,
1466}
1467
1468struct MergedSurface {
1475 base: Arc<SurfaceVariants>,
1476 mounts: Vec<SidecarMount>,
1477 dispatcher: Option<Arc<dyn Dispatcher>>,
1478 gateway: Option<Arc<HmacSigner>>,
1481 env: VentureEnv,
1482 captcha_configured: bool,
1483}
1484
1485impl MergedSurface {
1486 async fn sidecar_modules(&self) -> Vec<crate::surface::ModuleSurface> {
1488 let mut extra = Vec::new();
1489 let Some(dispatcher) = &self.dispatcher else {
1490 return extra;
1491 };
1492 for mount in &self.mounts {
1493 if !dispatcher.has(&mount.binding) {
1494 continue;
1495 }
1496 let mut builder = axum::http::Request::builder()
1497 .method(axum::http::Method::GET)
1498 .uri("/__surface")
1499 .header(header::ACCEPT, "application/json");
1500 if let Some(signer) = &self.gateway {
1501 builder =
1503 builder.header(X_HARNESS_GATEWAY, mint_gateway(signer, &mount.name, false));
1504 }
1505 let request = builder
1506 .body(bytes::Bytes::new())
1507 .expect("static request builds");
1508 let answer = match dispatcher.dispatch(&mount.binding, request).await {
1509 Ok(response) if response.status().is_success() => response,
1510 Ok(response) => {
1511 tracing::warn!(module = mount.name, status = %response.status(), "sidecar surface not available");
1512 continue;
1513 }
1514 Err(err) => {
1515 tracing::warn!(module = mount.name, error = %err, "sidecar surface fetch failed");
1516 continue;
1517 }
1518 };
1519 if answer.body().len() > MAX_SIDECAR_SURFACE_BYTES {
1522 tracing::warn!(
1523 module = mount.name,
1524 "sidecar surface exceeds {MAX_SIDECAR_SURFACE_BYTES} bytes and was not merged"
1525 );
1526 continue;
1527 }
1528 match serde_json::from_slice::<SurfaceDocument>(answer.body()) {
1529 Ok(document) => match sanitize_sidecar_document(&document, &mount.name) {
1530 Ok(entries) => extra.extend(entries.into_iter().filter_map(|mut entry| {
1531 if self.env == VentureEnv::Production
1535 && !self.captcha_configured
1536 && strip_unguarded_captcha_actions(&mut entry.surface) > 0
1537 {
1538 tracing::warn!(
1539 module = entry.name,
1540 "merged sidecar actions requiring a captcha were dropped: no Captcha port here"
1541 );
1542 }
1543 (!entry.surface.is_empty()).then_some(entry)
1544 })),
1545 Err(problems) => {
1546 for problem in problems {
1547 tracing::warn!(module = mount.name, error = problem, "sidecar surface rejected at merge");
1548 }
1549 }
1550 },
1551 Err(err) => {
1552 tracing::warn!(module = mount.name, error = %err, "sidecar surface is not a surface document");
1553 }
1554 }
1555 }
1556 extra
1557 }
1558}
1559
1560#[async_trait::async_trait]
1561impl SurfaceSource for MergedSurface {
1562 async fn current(&self) -> Arc<SurfaceDocument> {
1563 if self.mounts.is_empty() {
1564 return Arc::clone(&self.base.document);
1565 }
1566 let mut document = (*self.base.document).clone();
1567 document.modules.extend(self.sidecar_modules().await);
1568 Arc::new(document)
1569 }
1570
1571 fn built(&self) -> Arc<SurfaceDocument> {
1572 Arc::clone(&self.base.document)
1573 }
1574
1575 fn rendered(&self, admin: bool) -> Option<&RenderedSurface> {
1576 Some(if admin {
1577 &self.base.full
1578 } else {
1579 &self.base.public
1580 })
1581 }
1582}
1583
1584async fn surface_handler(
1600 State(state): State<SurfaceState>,
1601 headers: HeaderMap,
1602) -> impl IntoResponse {
1603 let admin = require_admin(&*state.config, &headers).is_ok();
1604 let current = state.source.current().await;
1608 let built = state.source.built();
1609 let prerendered = Arc::ptr_eq(¤t, &built).then(|| state.source.rendered(admin));
1610 let fresh;
1611 let rendered: &RenderedSurface = if let Some(rendered) = prerendered.flatten() {
1612 rendered
1613 } else {
1614 fresh = if admin {
1615 RenderedSurface::render(¤t)
1616 } else {
1617 RenderedSurface::render(¤t.public())
1618 };
1619 &fresh
1620 };
1621 let matches = headers
1622 .get(header::IF_NONE_MATCH)
1623 .and_then(|value| value.to_str().ok())
1624 .is_some_and(|value| {
1625 value
1626 .split(',')
1627 .map(str::trim)
1628 .any(|tag| tag == "*" || tag == rendered.etag)
1629 });
1630 let mut response = if matches {
1631 StatusCode::NOT_MODIFIED.into_response()
1632 } else {
1633 (
1634 [(header::CONTENT_TYPE, "application/json")],
1635 rendered.json.clone(),
1636 )
1637 .into_response()
1638 };
1639 let response_headers = response.headers_mut();
1640 response_headers.insert(
1641 header::ETAG,
1642 header::HeaderValue::from_str(&rendered.etag).expect("hex etag is a valid header"),
1643 );
1644 response_headers.insert(
1645 header::CACHE_CONTROL,
1646 if admin {
1647 header::HeaderValue::from_static("private, no-store")
1648 } else {
1649 header::HeaderValue::from_static("no-cache")
1650 },
1651 );
1652 response_headers.insert(
1653 header::VARY,
1654 header::HeaderValue::from_static("Authorization"),
1655 );
1656 response
1657}
1658
1659#[derive(Default)]
1662pub struct HarnessBuilder {
1663 venture: Option<Venture>,
1664 modules: Vec<Arc<dyn Module>>,
1665 provides: Vec<Port>,
1666 runtime: Option<Arc<dyn Runtime>>,
1667 module_templates: Vec<(String, Box<dyn Template>)>,
1668 overrides: Vec<(String, Box<dyn Template>)>,
1669 ui: Option<Arc<dyn UiMount>>,
1670}
1671
1672impl HarnessBuilder {
1673 #[must_use]
1674 pub fn venture(mut self, venture: Venture) -> Self {
1675 self.venture = Some(venture);
1676 self
1677 }
1678
1679 #[must_use]
1682 pub fn module(mut self, module: impl Module) -> Self {
1683 self.modules.push(Arc::new(module));
1684 self
1685 }
1686
1687 #[must_use]
1690 pub fn module_arc(mut self, module: Arc<dyn Module>) -> Self {
1691 self.modules.push(module);
1692 self
1693 }
1694
1695 #[must_use]
1699 pub fn runtime(mut self, runtime: impl Runtime) -> Self {
1700 let runtime: Arc<dyn Runtime> = Arc::new(runtime);
1701 self.provides = runtime.provides();
1702 self.runtime = Some(runtime);
1703 self
1704 }
1705
1706 #[must_use]
1709 pub fn templates(
1710 mut self,
1711 templates: impl IntoIterator<Item = (String, Box<dyn Template>)>,
1712 ) -> Self {
1713 self.module_templates.extend(templates);
1714 self
1715 }
1716
1717 #[must_use]
1720 pub fn ui(mut self, ui: impl UiMount) -> Self {
1721 self.ui = Some(Arc::new(ui));
1722 self
1723 }
1724
1725 #[must_use]
1728 pub fn template(mut self, id: impl Into<String>, template: Box<dyn Template>) -> Self {
1729 self.overrides.push((id.into(), template));
1730 self
1731 }
1732
1733 pub fn build(self) -> Result<Harness, ConfigError> {
1744 let mut errors = ConfigError::default();
1745
1746 let venture = if let Some(venture) = self.venture {
1747 venture.validate(&mut errors);
1748 venture
1749 } else {
1750 errors.push("missing venture: call .venture(Venture::new(..)) before .build()");
1751 Venture::new("invalid", "invalid.invalid")
1752 };
1753
1754 let mut names: HashMap<&'static str, usize> = HashMap::new();
1755 let mut tables: HashMap<&'static str, &'static str> = HashMap::new();
1756
1757 for module in &self.modules {
1758 if module.harness_api() != HARNESS_API {
1759 errors.push(harness_api_mismatch(module.as_ref()));
1760 }
1761
1762 let name = module.name();
1763 if name.is_empty() || !is_module_name(name) {
1764 errors.push(format!(
1765 "module name `{name}` must be kebab-case ([a-z0-9]+ separated by '-')"
1766 ));
1767 }
1768 if names.insert(name, 1).is_some() {
1769 errors.push(format!("duplicate module name `{name}`"));
1770 }
1771
1772 for port in module.requires().iter().chain(module.optional()) {
1773 if !Port::ALL.contains(port) {
1774 errors.push(format!(
1775 "module `{name}` declares unknown port {}",
1776 port.name()
1777 ));
1778 }
1779 }
1780 for port in module.requires() {
1781 if module.optional().contains(port) {
1782 errors.push(format!(
1783 "module `{name}` lists port {} in both requires() and optional()",
1784 port.name()
1785 ));
1786 }
1787 }
1788
1789 module.surface().validate(name, &mut errors);
1790
1791 check_tables(module.as_ref(), &mut tables, &mut errors);
1792 }
1793
1794 let well_known = collect_well_known(&self.modules, &mut errors);
1795
1796 for module in &self.modules {
1797 for port in module.requires() {
1798 if !self.provides.contains(port) {
1799 errors.push(format!(
1800 "module `{}` requires port {} which the runtime does not provide",
1801 module.name(),
1802 port.name()
1803 ));
1804 }
1805 }
1806 warn_undeclared_ports(module.as_ref(), &self.provides);
1807 }
1808
1809 let ordered = resolve_dependency_order(&self.modules, &mut errors);
1810
1811 append_production_readiness(&venture, &self.modules, self.runtime.as_ref(), &mut errors);
1812
1813 check_template_ids(
1814 self.overrides.iter().chain(self.module_templates.iter()),
1815 &names,
1816 &mut errors,
1817 );
1818
1819 let surface = Arc::new(SurfaceVariants::compose(
1820 &venture,
1821 &self.modules,
1822 self.ui.as_ref(),
1823 ));
1824 if let Some(ui) = &self.ui {
1825 ui.validate(&surface.document, &mut errors);
1826 }
1827
1828 errors.into_result()?;
1829
1830 let mut registry = TemplateRegistry::new();
1831 registry.register_all(self.module_templates);
1832 registry.register_all(self.overrides);
1833
1834 let mut events = EventBus::new();
1835 for module in &self.modules {
1836 for (name, handler) in module.events() {
1837 events = events.on(name, handler);
1838 }
1839 }
1840
1841 Ok(Harness {
1842 venture: Arc::new(venture),
1843 unprotected_acceptance_recorded: AtomicBool::new(false),
1844 unlimited_acceptance_recorded: AtomicBool::new(false),
1845 personal_data: Arc::new(crate::PersonalDataCatalog::compose(
1851 ordered.iter().map(|m| (m.name(), m.personal_data())),
1852 )),
1853 modules: ordered,
1854 templates: Arc::new(registry),
1855 events,
1856 runtime: self.runtime,
1857 well_known,
1858 surface,
1859 ui: self.ui,
1860 })
1861 }
1862}
1863
1864fn resolve_dependency_order(
1877 modules: &[Arc<dyn Module>],
1878 errors: &mut ConfigError,
1879) -> Vec<Arc<dyn Module>> {
1880 let names: BTreeSet<&str> = modules.iter().map(|module| module.name()).collect();
1881 let mut unknown = false;
1882 for module in modules {
1883 for needed in module.depends_on() {
1884 if !names.contains(needed) {
1885 unknown = true;
1886 errors.push(format!(
1887 "module `{}` depends on `{needed}`, which this venture does not compose: \
1888 add the module or drop the dependency",
1889 module.name()
1890 ));
1891 }
1892 }
1893 }
1894 if unknown {
1895 return modules.to_vec();
1896 }
1897
1898 let mut remaining: Vec<Option<Arc<dyn Module>>> = modules
1902 .iter()
1903 .map(|module| Some(Arc::clone(module)))
1904 .collect();
1905 let mut placed: BTreeSet<&str> = BTreeSet::new();
1906 let mut ordered: Vec<Arc<dyn Module>> = Vec::with_capacity(modules.len());
1907
1908 while ordered.len() < modules.len() {
1909 let mut progressed = false;
1910 for slot in &mut remaining {
1911 let ready = slot.as_ref().is_some_and(|module| {
1912 module
1913 .depends_on()
1914 .iter()
1915 .all(|needed| placed.contains(needed))
1916 });
1917 if ready {
1918 let module = slot.take().expect("checked above");
1919 placed.insert(module.name());
1920 ordered.push(module);
1921 progressed = true;
1922 }
1923 }
1924 if !progressed {
1925 let stuck: Vec<&str> = remaining
1927 .iter()
1928 .filter_map(|slot| slot.as_ref().map(|module| module.name()))
1929 .collect();
1930 errors.push(format!(
1931 "modules [{}] depend on each other in a cycle: migrations cannot be ordered so \
1932 that every module follows the ones it sits on top of",
1933 stuck.join(", ")
1934 ));
1935 return modules.to_vec();
1936 }
1937 }
1938 ordered
1939}
1940
1941fn check_template_ids<'a>(
1942 ids: impl Iterator<Item = &'a (String, Box<dyn Template>)>,
1943 names: &HashMap<&'static str, usize>,
1944 errors: &mut ConfigError,
1945) {
1946 for (id, _) in ids {
1947 let Some(module_name) = id.split('/').next() else {
1948 continue;
1949 };
1950 if !names.contains_key(module_name) {
1951 errors.push(format!(
1952 "template `{id}` names module `{module_name}` which is not registered"
1953 ));
1954 }
1955 }
1956}
1957
1958fn is_module_name(name: &str) -> bool {
1959 name.split('-').all(|part| {
1960 !part.is_empty()
1961 && part
1962 .chars()
1963 .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
1964 })
1965}
1966
1967async fn production_readiness_guard(
1977 State(problems): State<Arc<Vec<String>>>,
1978 request: axum::extract::Request,
1979 next: Next,
1980) -> Response {
1981 if problems.is_empty() || !request.uri().path().starts_with("/v1/") {
1982 return next.run(request).await;
1983 }
1984 let problem = Problem::new(&SLUGS.not_production_ready).with_detail(problems.join("; "));
1985 match request.extensions().get::<Scope>() {
1986 Some(scope) => problem.instance(&scope.request_id.clone()),
1987 None => problem,
1988 }
1989 .into_response()
1990}
1991
1992async fn admin_rate_limit_layer(
2001 State(limiter): State<Option<Arc<dyn RateLimiter>>>,
2002 request: axum::extract::Request,
2003 next: Next,
2004) -> Response {
2005 if !crate::sidecar::is_admin_path(request.uri().path()) {
2006 return next.run(request).await;
2007 }
2008 let keys: Vec<String> = crate::rate_limit::rate_limit_keys(
2009 crate::rate_limit::client_ip(request.headers()).as_deref(),
2010 None,
2011 )
2012 .into_iter()
2013 .map(|key| format!("admin:{key}"))
2014 .collect();
2015 if let RateLimit::Denied { decision } =
2016 crate::rate_limit::check_rate_limit(limiter.as_ref(), &keys, RateLimitFailure::FailClosed)
2017 .await
2018 {
2019 return rate_limited(&decision);
2020 }
2021 next.run(request).await
2022}
2023
2024fn append_production_readiness(
2036 venture: &Venture,
2037 modules: &[Arc<dyn Module>],
2038 runtime: Option<&Arc<dyn Runtime>>,
2039 errors: &mut ConfigError,
2040) {
2041 for error in crate::route_policy::production_readiness(
2042 venture.env,
2043 &crate::route_policy::WriteGuards::collect(modules),
2044 runtime,
2045 crate::route_policy::rate_limiter_effective(runtime),
2051 crate::route_policy::signer_effective(runtime),
2052 None,
2053 None,
2054 ) {
2055 errors.push(error);
2056 }
2057}
2058
2059fn collect_well_known(modules: &[Arc<dyn Module>], errors: &mut ConfigError) -> Option<Router> {
2060 let mut providers: Vec<&'static str> = Vec::new();
2061 let mut well_known = None;
2062 for module in modules {
2063 if let Some(router) = module.well_known() {
2064 providers.push(module.name());
2065 well_known = Some(router);
2066 }
2067 }
2068 if providers.len() > 1 {
2069 let listed = providers
2070 .iter()
2071 .map(|name| format!("`{name}`"))
2072 .collect::<Vec<_>>()
2073 .join(", ");
2074 errors.push(format!(
2075 "modules {listed} all provide a well-known router; at most one module may occupy /.well-known"
2076 ));
2077 }
2078 well_known
2079}
2080
2081#[cfg(test)]
2082mod acceptance_recording {
2083 use super::*;
2089 use crate::Migrations;
2090 use crate::config::MapConfig;
2091
2092 struct PubWriter;
2095
2096 impl Module for PubWriter {
2097 fn name(&self) -> &'static str {
2098 "writer"
2099 }
2100 fn version(&self) -> &'static str {
2101 "0.0.0-test"
2102 }
2103 fn requires(&self) -> &'static [Port] {
2104 &[]
2105 }
2106 fn public_writes(&self) -> bool {
2107 true
2108 }
2109 fn migrations(&self) -> Migrations {
2110 Migrations::default()
2111 }
2112 fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2113 Ok(())
2114 }
2115 fn router(&self, _: ModuleContext) -> Router {
2116 Router::new()
2117 }
2118 }
2119
2120 struct AllPorts;
2123
2124 impl Runtime for AllPorts {
2125 fn provides(&self) -> Vec<Port> {
2126 Port::ALL.to_vec()
2127 }
2128 }
2129
2130 fn harness() -> Harness {
2131 Harness::builder()
2132 .venture(
2133 Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2134 )
2135 .module(PubWriter)
2136 .runtime(AllPorts)
2137 .build()
2138 .expect("a development venture builds")
2139 }
2140
2141 fn deployed(unlimited: Option<&str>) -> MapConfig {
2142 let pairs = [("ENV", "production")].into_iter().chain(
2143 unlimited.map(|reason| (crate::route_policy::ALLOW_UNLIMITED_PUBLIC_ROUTES, reason)),
2144 );
2145 MapConfig::from_pairs(pairs)
2146 }
2147
2148 #[test]
2149 fn the_waiver_is_recorded_once_and_only_when_it_waived() {
2150 let harness = harness();
2151 let config = deployed(Some("issue #437: limiter binding pending"));
2152 assert!(
2153 harness
2154 .production_readiness_now(VentureEnv::Production, &config, false, true)
2155 .is_empty(),
2156 "the recorded acceptance serves"
2157 );
2158 assert!(
2159 harness
2160 .unlimited_acceptance_recorded
2161 .load(Ordering::Relaxed)
2162 );
2163
2164 assert!(
2167 harness
2168 .production_readiness_now(VentureEnv::Production, &config, false, true)
2169 .is_empty()
2170 );
2171 }
2172
2173 #[test]
2174 fn nothing_is_recorded_when_the_port_was_fine_all_along() {
2175 let harness = harness();
2176 let config = deployed(Some("unused: the limiter resolved"));
2177 assert!(
2178 harness
2179 .production_readiness_now(VentureEnv::Production, &config, true, true)
2180 .is_empty()
2181 );
2182 assert!(
2183 !harness
2184 .unlimited_acceptance_recorded
2185 .load(Ordering::Relaxed),
2186 "a waiver that removed no refusal is not a recorded acceptance"
2187 );
2188 }
2189
2190 #[test]
2191 fn a_refusing_deployment_records_no_acceptance_at_all() {
2192 let harness = harness();
2193 let config = MapConfig::from_pairs([
2197 ("ENV", "production"),
2198 (
2199 crate::route_policy::ALLOW_UNPROTECTED_WRITES,
2200 "captcha pending",
2201 ),
2202 ]);
2203 assert_eq!(
2204 harness
2205 .production_readiness_now(VentureEnv::Production, &config, false, true)
2206 .len(),
2207 1
2208 );
2209 assert!(
2210 !harness
2211 .unprotected_acceptance_recorded
2212 .load(Ordering::Relaxed)
2213 );
2214 assert!(
2215 !harness
2216 .unlimited_acceptance_recorded
2217 .load(Ordering::Relaxed)
2218 );
2219 }
2220
2221 struct HmacWriter;
2225
2226 impl Module for HmacWriter {
2227 fn name(&self) -> &'static str {
2228 "webhooks"
2229 }
2230 fn version(&self) -> &'static str {
2231 "0.0.0-test"
2232 }
2233 fn requires(&self) -> &'static [Port] {
2234 &[]
2235 }
2236 fn public_writes(&self) -> bool {
2237 true
2238 }
2239 fn public_write_policy(&self) -> crate::route_policy::RoutePolicy {
2240 crate::route_policy::RoutePolicy::Signature
2241 }
2242 fn signature_verification(&self) -> crate::route_policy::SignatureVerification {
2243 crate::route_policy::SignatureVerification::Hmac {
2244 secret: "WEBHOOK_SECRET",
2245 }
2246 }
2247 fn migrations(&self) -> Migrations {
2248 Migrations::default()
2249 }
2250 fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2251 Ok(())
2252 }
2253 fn router(&self, _: ModuleContext) -> Router {
2254 Router::new()
2255 }
2256 }
2257
2258 #[test]
2259 fn a_non_payments_signature_venture_boots_on_its_webhook_secret_alone() {
2260 let harness = Harness::builder()
2261 .venture(
2262 Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2263 )
2264 .module(HmacWriter)
2265 .runtime(AllPorts)
2266 .build()
2267 .expect("a development venture builds");
2268 let deployed = MapConfig::from_pairs([("ENV", "production")]);
2269 let problems =
2270 harness.production_readiness_now(VentureEnv::Production, &deployed, true, true);
2271 assert_eq!(problems.len(), 1, "{problems:?}");
2272 assert!(
2273 problems[0].contains("WEBHOOKS_WEBHOOK_SECRET"),
2274 "{}",
2275 problems[0]
2276 );
2277 assert!(
2278 !problems[0].contains("Payments"),
2279 "a module verifying through the HMAC scheme is not the Payments leg's business: {}",
2280 problems[0]
2281 );
2282
2283 let ready = MapConfig::from_pairs([
2285 ("ENV", "production"),
2286 ("WEBHOOKS_WEBHOOK_SECRET", "whsec-test-dummy"),
2287 ]);
2288 assert!(
2289 harness
2290 .production_readiness_now(VentureEnv::Production, &ready, true, true)
2291 .is_empty()
2292 );
2293 }
2294}