1use axum::{
2 extract::State,
3 http::Request,
4 middleware::Next,
5 response::{IntoResponse, Response},
6};
7use http_body::Body as HttpBody;
8use pin_project_lite::pin_project;
9use secrecy::ExposeSecret;
10use std::future::Future;
11use std::pin::Pin;
12use std::task::{Context, Poll};
13
14tokio::task_local! {
16 pub static CURRENT_TENANT: Option<String>;
17}
18
19#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct Tenant(pub String);
22
23impl axum::extract::FromRequestParts<crate::AppState> for Tenant {
24 type Rejection = crate::AutumnError;
25
26 async fn from_request_parts(
27 parts: &mut axum::http::request::Parts,
28 state: &crate::AppState,
29 ) -> Result<Self, Self::Rejection> {
30 if let Ok(Some(tenant_id)) = CURRENT_TENANT.try_with(Clone::clone) {
34 return Ok(Self(tenant_id));
35 }
36
37 let config = state
38 .extension::<crate::config::AutumnConfig>()
39 .ok_or_else(|| {
40 crate::AutumnError::service_unavailable_msg("Config is not available")
41 })?;
42 let domains =
43 state.extension::<std::sync::Arc<crate::custom_domain::CustomDomainRegistry>>();
44 let tenant_id = extract_tenant_from_parts_with_domains(
45 parts,
46 &config,
47 domains.as_deref().map(AsRef::as_ref),
48 )
49 .await?;
50 Ok(Self(tenant_id))
51 }
52}
53
54pub(crate) fn current_tenant_id() -> String {
60 CURRENT_TENANT
61 .try_with(Clone::clone)
62 .ok()
63 .flatten()
64 .unwrap_or_default()
65}
66
67pub async fn with_tenant<F, R>(tenant_id: String, future: F) -> R
69where
70 F: Future<Output = R>,
71{
72 CURRENT_TENANT.scope(Some(tenant_id), future).await
73}
74
75#[allow(clippy::missing_errors_doc, clippy::too_many_lines)]
89pub async fn extract_tenant_from_parts(
90 parts: &mut axum::http::request::Parts,
91 config: &crate::config::AutumnConfig,
92) -> Result<String, crate::AutumnError> {
93 extract_tenant_from_parts_with_domains(parts, config, None).await
94}
95
96pub async fn extract_tenant_from_parts_with_domains(
117 parts: &mut axum::http::request::Parts,
118 config: &crate::config::AutumnConfig,
119 domains: Option<&crate::custom_domain::CustomDomainRegistry>,
120) -> Result<String, crate::AutumnError> {
121 let replayed = replayed_tenant();
122 match &replayed {
123 ReplayedTenant::NoTape | ReplayedTenant::RecordedFailure | ReplayedTenant::Unchecked => {}
124 ReplayedTenant::Resolved(tenant_id) => return Ok(tenant_id.clone()),
125 ReplayedTenant::Unrecorded => {
128 return Err(crate::AutumnError::internal_server_error(
129 std::io::Error::other(
130 "the replayed run resolved a tenant, which the capsule has no recording for; \
131 live tenant resolution was not consulted",
132 ),
133 ));
134 }
135 }
136 if config.tenancy.enabled
142 && config.tenancy.source == "subdomain"
143 && let Some(registry) = domains
144 && let Some(host) = request_host(parts)
145 && let Some(tenant_id) = registry.tenant_for_host(&host)
146 {
147 if matches!(replayed, ReplayedTenant::RecordedFailure) {
148 return Err(tenant_resolved_after_recorded_failure());
149 }
150 record_tenant(Some(&tenant_id));
151 return Ok(tenant_id);
152 }
153 let resolved = extract_tenant_from_parts_inner(parts, config).await;
157 if matches!(replayed, ReplayedTenant::RecordedFailure) && resolved.is_ok() {
160 return Err(tenant_resolved_after_recorded_failure());
161 }
162 record_tenant(resolved.as_ref().ok().map(String::as_str));
165 resolved
166}
167
168fn tenant_resolved_after_recorded_failure() -> crate::AutumnError {
174 note_tenant_resolved_after_recorded_failure();
175 crate::AutumnError::internal_server_error(std::io::Error::other(
176 "the recorded tenant lookup failed, but the replayed run resolved a tenant",
177 ))
178}
179
180enum ReplayedTenant {
182 NoTape,
184 Resolved(String),
186 RecordedFailure,
188 Unrecorded,
190 Unchecked,
192}
193
194#[cfg(feature = "reporting")]
196fn replayed_tenant() -> ReplayedTenant {
197 use crate::capsule::effects::TenantVerdict;
198 let Some(tape) = crate::capsule::effects::current_tape() else {
199 return ReplayedTenant::NoTape;
200 };
201 match tape.tenant() {
202 TenantVerdict::Resolved(id) => ReplayedTenant::Resolved(id),
203 TenantVerdict::RecordedFailure => ReplayedTenant::RecordedFailure,
204 TenantVerdict::Unrecorded => ReplayedTenant::Unrecorded,
205 TenantVerdict::Unchecked => ReplayedTenant::Unchecked,
206 }
207}
208
209#[cfg(feature = "reporting")]
211fn note_tenant_resolved_after_recorded_failure() {
212 if let Some(tape) = crate::capsule::effects::current_tape() {
213 tape.tenant_resolved_after_recorded_failure();
214 }
215}
216
217#[cfg(not(feature = "reporting"))]
219const fn note_tenant_resolved_after_recorded_failure() {}
220
221#[cfg(not(feature = "reporting"))]
223const fn replayed_tenant() -> ReplayedTenant {
224 ReplayedTenant::NoTape
225}
226
227#[cfg(feature = "reporting")]
230fn record_tenant(tenant_id: Option<&str>) {
231 if let Some(scope) = crate::capsule::current_scope() {
232 scope.record_tenant(crate::capsule::TenantEffect {
233 id: tenant_id.map(ToOwned::to_owned),
234 });
235 }
236}
237
238#[cfg(not(feature = "reporting"))]
240const fn record_tenant(_tenant_id: Option<&str>) {}
241
242fn request_host(parts: &axum::http::request::Parts) -> Option<String> {
250 parts
251 .extensions
252 .get::<crate::security::ResolvedClientIdentity>()
253 .and_then(|id| id.host.clone())
254 .or_else(|| {
255 parts
256 .headers
257 .get(axum::http::header::HOST)
258 .and_then(|h| h.to_str().ok())
259 .map(ToOwned::to_owned)
260 })
261}
262
263#[allow(
265 clippy::too_many_lines,
266 clippy::needless_pass_by_ref_mut,
267 reason = "the `&mut Parts` signature mirrors the public wrapper and axum's \
268 `FromRequestParts` contract; narrowing it here would force the \
269 caller to reborrow"
270)]
271async fn extract_tenant_from_parts_inner(
272 parts: &mut axum::http::request::Parts,
273 config: &crate::config::AutumnConfig,
274) -> Result<String, crate::AutumnError> {
275 if !config.tenancy.enabled {
276 return Err(crate::AutumnError::service_unavailable_msg(
277 "Tenancy is not enabled; set [tenancy] enabled = true in autumn.toml",
278 ));
279 }
280
281 match config.tenancy.source.as_str() {
282 "header" => {
283 let header_value = parts
284 .headers
285 .get(&config.tenancy.header_name)
286 .ok_or_else(|| {
287 crate::AutumnError::bad_request_msg(format!(
288 "Missing required tenant header: {}",
289 config.tenancy.header_name
290 ))
291 })?;
292 let val = header_value
293 .to_str()
294 .map_err(|_| {
295 crate::AutumnError::bad_request_msg(format!(
296 "Invalid UTF-8 in tenant header: {}",
297 config.tenancy.header_name
298 ))
299 })?
300 .to_string();
301 if val.trim().is_empty() {
302 return Err(crate::AutumnError::bad_request_msg(format!(
303 "Tenant header {} is empty",
304 config.tenancy.header_name
305 )));
306 }
307 Ok(val)
308 }
309 "subdomain" => {
310 let host_owned = match request_host(parts) {
311 Some(host) => host,
312 None if parts.headers.contains_key(axum::http::header::HOST) => {
315 return Err(crate::AutumnError::bad_request_msg(
316 "Invalid UTF-8 in Host header",
317 ));
318 }
319 None => {
320 return Err(crate::AutumnError::bad_request_msg(
321 "Missing Host header for subdomain tenancy",
322 ));
323 }
324 };
325
326 let host = host_owned.as_str();
327 let host_only = host.split(':').next().unwrap_or(host).trim();
328
329 if host_only.parse::<std::net::IpAddr>().is_ok() {
330 return Err(crate::AutumnError::bad_request_msg(
331 "IP address host not allowed in subdomain mode",
332 ));
333 }
334
335 let host_lower = host_only.to_lowercase();
338
339 if let Some(ref base_domain) = config.tenancy.base_domain {
340 let base_domain_clean = base_domain.trim().to_lowercase();
341 if !host_lower.ends_with(base_domain_clean.as_str()) {
342 return Err(crate::AutumnError::bad_request_msg(format!(
343 "Host does not match base domain: {base_domain_clean}"
344 )));
345 }
346 if host_lower.len() <= base_domain_clean.len() {
347 return Err(crate::AutumnError::bad_request_msg(
348 "Apex domain not allowed in subdomain mode",
349 ));
350 }
351 let prefix_len = host_lower.len() - base_domain_clean.len();
352 if !host_lower[..prefix_len].ends_with('.') {
353 return Err(crate::AutumnError::bad_request_msg(
354 "Invalid subdomain format",
355 ));
356 }
357 let subdomain_part = &host_lower[..prefix_len - 1];
358 let tenant = subdomain_part.split('.').next().ok_or_else(|| {
359 crate::AutumnError::bad_request_msg("Unable to extract subdomain tenant")
360 })?;
361 if tenant.trim().is_empty() {
362 return Err(crate::AutumnError::bad_request_msg(
363 "Extracted subdomain tenant is empty",
364 ));
365 }
366 Ok(tenant.to_string())
367 } else {
368 let labels: Vec<&str> = host_lower.split('.').filter(|s| !s.is_empty()).collect();
369 if labels.is_empty() {
370 return Err(crate::AutumnError::bad_request_msg("Empty host header"));
371 }
372
373 if labels.len() < 2 {
374 return Err(crate::AutumnError::bad_request_msg(
375 "Apex or local host without subdomain not allowed",
376 ));
377 }
378
379 if labels.len() == 2 && labels[1] != "localhost" {
380 return Err(crate::AutumnError::bad_request_msg(
381 "Apex domain not allowed in subdomain mode",
382 ));
383 }
384
385 let tenant = labels[0].to_string();
386 if tenant.trim().is_empty() {
387 return Err(crate::AutumnError::bad_request_msg(
388 "Extracted subdomain tenant is empty",
389 ));
390 }
391 Ok(tenant)
392 }
393 }
394 "session" => {
395 let session = parts
396 .extensions
397 .get::<crate::session::Session>()
398 .ok_or_else(|| {
399 crate::AutumnError::internal_server_error_msg(
400 "SessionLayer not installed but session tenancy source is configured",
401 )
402 })?;
403 let tenant = session
404 .get(&config.tenancy.session_key)
405 .await
406 .ok_or_else(|| {
407 crate::AutumnError::unauthorized_msg(format!(
408 "Tenant ID missing from session key: {}",
409 config.tenancy.session_key
410 ))
411 })?;
412 if tenant.trim().is_empty() {
413 return Err(crate::AutumnError::unauthorized_msg(format!(
414 "Tenant ID in session key {} is empty",
415 config.tenancy.session_key
416 )));
417 }
418 Ok(tenant)
419 }
420 "jwt" => {
421 let auth_header = parts
422 .headers
423 .get(axum::http::header::AUTHORIZATION)
424 .ok_or_else(|| {
425 crate::AutumnError::unauthorized_msg(
426 "Missing Authorization header for JWT tenancy",
427 )
428 })?;
429 let auth_str = auth_header.to_str().map_err(|_| {
430 crate::AutumnError::unauthorized_msg("Invalid UTF-8 in Authorization header")
431 })?;
432
433 if auth_str.len() < 7
434 || !auth_str.is_char_boundary(7)
435 || !auth_str[..7].eq_ignore_ascii_case("bearer ")
436 {
437 return Err(crate::AutumnError::unauthorized_msg(
438 "Invalid Authorization header format. Expected Bearer <token>",
439 ));
440 }
441 let token = &auth_str[7..];
442
443 let secret = config.tenancy.jwt_secret.as_ref().ok_or_else(|| {
444 crate::AutumnError::unauthorized_msg("JWT secret is not configured")
445 })?;
446
447 let mut validation = ::jsonwebtoken::Validation::default();
448 if let Some(ref iss) = config.tenancy.jwt_issuer {
449 validation.set_issuer(::std::slice::from_ref(iss));
450 }
451 if let Some(ref aud) = config.tenancy.jwt_audience {
452 validation.set_audience(&[aud.as_str()]);
453 } else {
454 validation.validate_aud = false;
455 }
456
457 let token_data = ::jsonwebtoken::decode::<serde_json::Value>(
458 token,
459 &::jsonwebtoken::DecodingKey::from_secret(secret.expose_secret().as_bytes()),
460 &validation,
461 )
462 .map_err(|e| {
463 crate::AutumnError::unauthorized_msg(format!("JWT validation failed: {e}"))
464 })?;
465
466 if let Some(ref expected_aud) = config.tenancy.jwt_audience {
472 let aud_ok = token_data.claims.get("aud").is_some_and(|v| match v {
473 serde_json::Value::String(s) => s == expected_aud,
474 serde_json::Value::Array(arr) => arr
475 .iter()
476 .any(|e| e.as_str() == Some(expected_aud.as_str())),
477 _ => false,
478 });
479 if !aud_ok {
480 return Err(crate::AutumnError::unauthorized_msg(
481 "JWT audience validation failed: missing or invalid aud claim",
482 ));
483 }
484 }
485
486 let tenant = token_data
487 .claims
488 .get(&config.tenancy.jwt_claim)
489 .and_then(|v| v.as_str())
490 .ok_or_else(|| {
491 crate::AutumnError::unauthorized_msg(format!(
492 "Tenant claim '{}' missing from JWT payload",
493 config.tenancy.jwt_claim
494 ))
495 })?
496 .to_string();
497
498 if tenant.trim().is_empty() {
499 return Err(crate::AutumnError::unauthorized_msg(format!(
500 "Tenant claim '{}' in JWT payload is empty",
501 config.tenancy.jwt_claim
502 )));
503 }
504 Ok(tenant)
505 }
506 other => Err(crate::AutumnError::internal_server_error_msg(format!(
507 "Unsupported tenancy source: {other}"
508 ))),
509 }
510}
511
512fn is_public_path(path: &str, config: &crate::config::AutumnConfig) -> bool {
535 let path = strip_locale_prefix_for_tenancy(path, config);
545
546 let matches =
551 |prefix: &str| !prefix.is_empty() && crate::router::path_matches_route_prefix(path, prefix);
552
553 let user_paths_match = config.tenancy.public_paths.iter().any(|p| {
557 let p = if p.len() > 1 {
558 p.trim_end_matches('/')
559 } else {
560 p.as_str()
561 };
562 matches(p)
563 });
564
565 let redirect_match = config
576 .tenancy
577 .login_redirect
578 .as_deref()
579 .and_then(|target| target.parse::<axum::http::Uri>().ok())
580 .filter(|uri| uri.authority().is_none())
581 .is_some_and(|uri| uri.path() == path);
582
583 let actuator_prefix = crate::actuator::normalize_actuator_prefix(&config.actuator.prefix);
589 let actuator_match = if actuator_prefix.is_empty() {
590 crate::actuator::actuator_endpoint_paths(
594 &actuator_prefix,
595 config.actuator.sensitive,
596 config.actuator.prometheus,
597 )
598 .iter()
599 .any(|p| matches(p))
600 } else {
601 matches(&actuator_prefix)
602 };
603
604 let probe_match = path == config.health.path
609 || path == config.health.live_path
610 || path == config.health.ready_path
611 || path == config.health.startup_path;
612
613 user_paths_match || redirect_match || probe_match || actuator_match
614}
615
616#[cfg(feature = "i18n")]
646fn strip_locale_prefix_for_tenancy<'a>(
647 path: &'a str,
648 config: &crate::config::AutumnConfig,
649) -> &'a str {
650 if !config.i18n.locale_prefix_enabled {
651 return path;
652 }
653 if config
654 .i18n
655 .locale_prefix_exclude_exact
656 .iter()
657 .any(|p| p == path)
658 || crate::router::matches_locale_exclude_prefix(path, &config.i18n.locale_prefix_exclude)
659 {
660 return path;
661 }
662 let Some(rest) = path.strip_prefix('/') else {
663 return path;
664 };
665 let (segment, remainder) = rest
666 .find('/')
667 .map_or((rest, ""), |idx| (&rest[..idx], &rest[idx..]));
668 if config.i18n.supported_locales.iter().any(|l| l == segment) {
669 if remainder.is_empty() { "/" } else { remainder }
670 } else {
671 path
672 }
673}
674
675#[cfg(not(feature = "i18n"))]
676const fn strip_locale_prefix_for_tenancy<'a>(
677 path: &'a str,
678 _config: &crate::config::AutumnConfig,
679) -> &'a str {
680 path
681}
682
683pub async fn tenancy_middleware(
685 State(state): State<crate::AppState>,
686 request: Request<axum::body::Body>,
687 next: Next,
688) -> Response {
689 let Some(config) = state.extension::<crate::config::AutumnConfig>() else {
690 return crate::AutumnError::internal_server_error_msg("AutumnConfig not found in AppState")
691 .into_response();
692 };
693
694 if !config.tenancy.enabled {
695 return next.run(request).await;
696 }
697
698 let (mut parts, body) = request.into_parts();
699
700 if is_public_path(parts.uri.path(), &config) {
704 return next.run(Request::from_parts(parts, body)).await;
705 }
706
707 let domains = crate::custom_domain::CustomDomainRegistry::from_state(&state);
708 let tenant_id =
709 match extract_tenant_from_parts_with_domains(&mut parts, &config, domains.as_deref()).await
710 {
711 Ok(t) => t,
712 Err(e) => {
713 if e.status() == axum::http::StatusCode::UNAUTHORIZED
721 && let Some(target) = &config.tenancy.login_redirect
722 && parts
723 .headers
724 .get(axum::http::header::ACCEPT)
725 .and_then(|v| v.to_str().ok())
726 .is_some_and(|accept| accept.contains("text/html"))
727 {
728 return axum::response::Redirect::to(target).into_response();
729 }
730 return e.into_response();
731 }
732 };
733
734 crate::log::context::set_tenant_id(&tenant_id);
737
738 let bulkheads =
741 state.extension_or_insert_with(|| TenantBulkheads::from_config(&config.tenancy));
742 let acquired = bulkheads
743 .requests
744 .as_ref()
745 .map(|bulkhead| bulkhead.try_acquire(&tenant_id));
746 if matches!(acquired, Some(None)) {
747 state.metrics.record_tenant_request_rejection();
748 let origin = parts.headers.get(axum::http::header::ORIGIN);
749 return tenant_bulkhead_rejection(&tenant_id, &config.cors, origin);
750 }
751 let _request_permit = acquired.flatten();
752
753 let request = Request::from_parts(parts, body);
754 let tenant_id_clone = tenant_id.clone();
755
756 let registry = state.extension_or_insert_with(|| {
764 crate::tenant_cell::TenantCellRegistry::with_limits(
765 config.tenancy.max_cells,
766 (config.tenancy.idle_ttl_secs > 0)
767 .then(|| std::time::Duration::from_secs(config.tenancy.idle_ttl_secs)),
768 )
769 });
770 let handle = crate::tenant_cell::TenantCellHandle::from_shared(
773 (*registry).clone(),
774 std::sync::Arc::from(tenant_id.as_str()),
775 config.tenancy.quota_bytes,
776 );
777 let handle_for_body = Some(handle.clone());
778
779 let run = crate::tenant_cell::CURRENT_TENANT_CELL.scope(Some(handle), next.run(request));
780 let response = match bulkheads.db.clone() {
781 Some(db) => {
782 CURRENT_TENANT
783 .scope(
784 Some(tenant_id),
785 crate::bulkhead::TENANT_DB_BULKHEAD.scope(db, run),
786 )
787 .await
788 }
789 None => CURRENT_TENANT.scope(Some(tenant_id), run).await,
790 };
791
792 let (parts, body) = response.into_parts();
793 let wrapped = TenantPropagatingBody {
794 inner: body,
795 tenant_id: tenant_id_clone,
796 handle: handle_for_body,
797 db_bulkhead: bulkheads.db.clone(),
798 };
799 Response::from_parts(parts, axum::body::Body::new(wrapped))
800}
801
802struct TenantBulkheads {
805 requests: Option<std::sync::Arc<crate::bulkhead::TenantBulkhead>>,
807 db: Option<std::sync::Arc<crate::bulkhead::TenantBulkhead>>,
809}
810
811impl TenantBulkheads {
812 fn from_config(config: &crate::config::TenancyConfig) -> Self {
813 let make = |max: usize| (max > 0).then(|| crate::bulkhead::TenantBulkhead::new(max));
814 Self {
815 requests: make(config.max_concurrent_requests),
816 db: make(config.max_db_connections),
817 }
818 }
819}
820
821fn tenant_bulkhead_rejection(
825 tenant_id: &str,
826 cors: &crate::config::CorsConfig,
827 origin: Option<&axum::http::HeaderValue>,
828) -> Response {
829 tracing::debug!(tenant = %tenant_id, "tenant request cap reached");
830 let mut response = crate::AutumnError::service_unavailable_msg(
831 "Too many concurrent requests for this tenant; try again shortly.",
832 )
833 .into_response();
834 response.headers_mut().insert(
835 axum::http::header::RETRY_AFTER,
836 axum::http::HeaderValue::from_static("1"),
837 );
838 if !cors.allowed_origins.is_empty() {
839 crate::router::mirror_cors_headers(cors, origin, &mut response);
840 }
841 response
842}
843
844pin_project! {
845 pub struct TenantPropagatingBody<B> {
849 #[pin]
850 pub inner: B,
851 pub tenant_id: String,
852 pub handle: Option<crate::tenant_cell::TenantCellHandle>,
853 pub db_bulkhead: Option<std::sync::Arc<crate::bulkhead::TenantBulkhead>>,
856 }
857}
858
859impl<B> HttpBody for TenantPropagatingBody<B>
860where
861 B: HttpBody,
862{
863 type Data = B::Data;
864 type Error = B::Error;
865
866 fn poll_frame(
867 self: Pin<&mut Self>,
868 cx: &mut Context<'_>,
869 ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
870 let this = self.project();
871 let tenant_id = this.tenant_id.clone();
872 let handle = this.handle.clone();
873 let inner = this.inner;
874 CURRENT_TENANT.sync_scope(Some(tenant_id), || {
875 crate::tenant_cell::CURRENT_TENANT_CELL.sync_scope(handle, || match this.db_bulkhead {
876 Some(db) => crate::bulkhead::TENANT_DB_BULKHEAD
877 .sync_scope(std::sync::Arc::clone(db), || inner.poll_frame(cx)),
878 None => inner.poll_frame(cx),
879 })
880 })
881 }
882
883 fn is_end_stream(&self) -> bool {
884 self.inner.is_end_stream()
885 }
886
887 fn size_hint(&self) -> http_body::SizeHint {
888 self.inner.size_hint()
889 }
890}
891
892#[cfg(feature = "db")]
897pub trait TenantInsertable<'a, Table> {
898 type Values;
899 fn tenant_values(self, tenant_id: &'a str) -> Self::Values;
900}
901
902#[cfg(feature = "db")]
904pub trait ModelTenantIdMeta {
905 const HAS_MANUAL_TENANT_ID: bool;
907 fn try_set_tenant_id(&mut self, tenant_id: &str);
909}
910
911#[cfg(feature = "db")]
913pub trait HasTenantIdColumn {
914 type Column: ::diesel::Expression;
915 fn column() -> Self::Column;
916}
917
918#[cfg(feature = "db")]
920pub struct TenantInsertableValuesSelector<'a, T, Table, const HAS_MANUAL: bool> {
921 pub inner: T,
922 pub tenant_id: &'a str,
923 pub _marker: std::marker::PhantomData<Table>,
924}
925
926#[cfg(feature = "db")]
928pub trait GetInsertableValues {
929 type Values;
930 fn get_values(self) -> Self::Values;
931}
932
933#[cfg(feature = "db")]
934impl<T, Table> GetInsertableValues for TenantInsertableValuesSelector<'_, T, Table, true>
935where
936 T: ModelTenantIdMeta,
937{
938 type Values = T;
939 fn get_values(mut self) -> Self::Values {
940 self.inner.try_set_tenant_id(self.tenant_id);
941 self.inner
942 }
943}
944
945#[cfg(feature = "db")]
946impl<'a, T, Table> GetInsertableValues for TenantInsertableValuesSelector<'a, T, Table, false>
947where
948 Table: HasTenantIdColumn,
949 Table::Column: ::diesel::ExpressionMethods,
950 <Table::Column as ::diesel::Expression>::SqlType: ::diesel::sql_types::SqlType,
951 &'a str: ::diesel::expression::AsExpression<<Table::Column as ::diesel::Expression>::SqlType>,
952{
953 type Values = (T, ::diesel::dsl::Eq<Table::Column, &'a str>);
954 fn get_values(self) -> Self::Values {
955 use ::diesel::ExpressionMethods;
956 (self.inner, Table::column().eq(self.tenant_id))
957 }
958}
959
960pub trait DisplayTenantId {
962 fn tenant_id_str(&self) -> &str;
964}
965
966impl DisplayTenantId for String {
967 fn tenant_id_str(&self) -> &str {
968 self
969 }
970}
971
972impl DisplayTenantId for Option<String> {
973 fn tenant_id_str(&self) -> &str {
974 self.as_deref().unwrap_or("default")
975 }
976}
977
978#[cfg(test)]
979mod tests {
980 use super::*;
981 use crate::security::ResolvedClientIdentity;
982
983 struct PermitProbeBody {
985 outcome: std::sync::Arc<std::sync::Mutex<Option<bool>>>,
986 }
987
988 impl HttpBody for PermitProbeBody {
989 type Data = bytes::Bytes;
990 type Error = std::convert::Infallible;
991
992 fn poll_frame(
993 self: Pin<&mut Self>,
994 _cx: &mut Context<'_>,
995 ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
996 let rejected = crate::bulkhead::acquire_db_permit().is_err();
997 *self.outcome.lock().unwrap() = Some(rejected);
998 Poll::Ready(None)
999 }
1000 }
1001
1002 #[test]
1004 fn a_streaming_body_poll_is_inside_the_db_bulkhead() {
1005 let bulkhead = crate::bulkhead::TenantBulkhead::new(1);
1006 let _held = bulkhead.try_acquire("acme").expect("the one permit");
1007 let outcome = std::sync::Arc::new(std::sync::Mutex::new(None));
1008 let mut wrapped = TenantPropagatingBody {
1009 inner: PermitProbeBody {
1010 outcome: std::sync::Arc::clone(&outcome),
1011 },
1012 tenant_id: "acme".to_owned(),
1013 handle: None,
1014 db_bulkhead: Some(bulkhead),
1015 };
1016 let waker = futures::task::noop_waker();
1017 let mut cx = Context::from_waker(&waker);
1018 let _ = Pin::new(&mut wrapped).poll_frame(&mut cx);
1019 assert_eq!(
1020 *outcome.lock().unwrap(),
1021 Some(true),
1022 "the body's checkout counts against the tenant's cap"
1023 );
1024 }
1025
1026 #[test]
1029 fn tenant_cell_handle_clones_share_one_tenant_id() {
1030 let reg = crate::tenant_cell::TenantCellRegistry::new();
1031 let handle =
1032 crate::tenant_cell::TenantCellHandle::new(reg.clone(), "acme".to_owned(), 1_000);
1033 let clone = handle.clone();
1034 assert_eq!(clone.tenant_id(), "acme");
1035 assert!(
1036 std::ptr::eq(handle.tenant_id(), clone.tenant_id()),
1037 "clones share the id"
1038 );
1039 let debug = format!("{clone:?}");
1040 assert!(debug.contains("tenant_id: \"acme\""), "{debug}");
1041
1042 let shared: std::sync::Arc<str> = std::sync::Arc::from("globex");
1044 let from_shared = crate::tenant_cell::TenantCellHandle::from_shared(
1045 reg,
1046 std::sync::Arc::clone(&shared),
1047 1_000,
1048 );
1049 assert!(std::ptr::eq(
1050 from_shared.tenant_id().as_ptr(),
1051 shared.as_ptr()
1052 ));
1053 assert_eq!(from_shared.cell().tenant_id(), "globex");
1054 }
1055
1056 #[tokio::test]
1058 async fn a_handler_is_inside_the_db_bulkhead() {
1059 use tower::ServiceExt as _;
1060 let state = crate::AppState::for_test();
1061 let mut config = crate::config::AutumnConfig::default();
1062 config.tenancy.enabled = true;
1063 config.tenancy.source = "header".to_string();
1064 config.tenancy.header_name = "x-tenant-id".to_string();
1065 config.tenancy.max_db_connections = 1;
1066 state.insert_extension(config);
1067 let app = axum::Router::new()
1068 .route(
1069 "/",
1070 axum::routing::get(|| async {
1071 let first = crate::bulkhead::acquire_db_permit();
1072 let second = crate::bulkhead::acquire_db_permit();
1073 let held = matches!(first, Ok(Some(_)));
1074 if held && second.is_err() {
1075 axum::http::StatusCode::OK
1076 } else {
1077 axum::http::StatusCode::INTERNAL_SERVER_ERROR
1078 }
1079 }),
1080 )
1081 .layer(axum::middleware::from_fn_with_state(
1082 state.clone(),
1083 tenancy_middleware,
1084 ))
1085 .with_state(state);
1086 let request = axum::http::Request::builder()
1087 .uri("/")
1088 .header("x-tenant-id", "acme")
1089 .body(axum::body::Body::empty())
1090 .unwrap();
1091 let response = app.oneshot(request).await.unwrap();
1092 assert_eq!(
1093 response.status(),
1094 axum::http::StatusCode::OK,
1095 "one permit, then a 503 for the same tenant"
1096 );
1097 }
1098
1099 fn subdomain_config() -> crate::config::AutumnConfig {
1100 let mut c = crate::config::AutumnConfig::default();
1101 c.tenancy.enabled = true;
1102 c.tenancy.source = "subdomain".to_string();
1103 c
1104 }
1105
1106 fn subdomain_config_with_base(base: &str) -> crate::config::AutumnConfig {
1107 let mut c = subdomain_config();
1108 c.tenancy.base_domain = Some(base.to_string());
1109 c
1110 }
1111
1112 fn make_parts(host: &str) -> axum::http::request::Parts {
1113 let (parts, ()) = axum::http::Request::builder()
1114 .uri("http://ignored/")
1115 .header(axum::http::header::HOST, host)
1116 .body(())
1117 .unwrap()
1118 .into_parts();
1119 parts
1120 }
1121
1122 fn make_parts_with_identity(
1123 host_header: &str,
1124 resolved_host: &str,
1125 ) -> axum::http::request::Parts {
1126 let (mut parts, ()) = axum::http::Request::builder()
1127 .uri("http://ignored/")
1128 .header(axum::http::header::HOST, host_header)
1129 .body(())
1130 .unwrap()
1131 .into_parts();
1132 parts.extensions.insert(ResolvedClientIdentity {
1133 addr: None,
1134 host: Some(resolved_host.to_string()),
1135 scheme: None,
1136 });
1137 parts
1138 }
1139
1140 #[tokio::test]
1143 async fn subdomain_falls_back_to_host_header_without_extension() {
1144 let config = subdomain_config();
1145 let mut parts = make_parts("tenant1.example.com");
1146 let result = extract_tenant_from_parts(&mut parts, &config).await;
1147 assert_eq!(result.unwrap(), "tenant1");
1148 }
1149
1150 #[cfg(feature = "reporting")]
1154 #[tokio::test]
1155 async fn a_replay_with_no_recorded_tenant_fails_closed() {
1156 let config = subdomain_config();
1157 let mut parts = make_parts("tenant1.example.com");
1158 let tape = std::sync::Arc::new(crate::capsule::ReplayEffects::new(
1159 crate::capsule::CapsuleEffects::default(),
1160 ));
1161 let result = crate::capsule::with_effect_tape(
1162 std::sync::Arc::clone(&tape),
1163 extract_tenant_from_parts(&mut parts, &config),
1164 )
1165 .await;
1166 assert!(result.is_err(), "{result:?}");
1167 assert_eq!(tape.divergences().len(), 1);
1168 }
1169
1170 #[cfg(feature = "reporting")]
1172 #[tokio::test]
1173 async fn a_recorded_failure_that_now_resolves_fails_closed() {
1174 let config = subdomain_config();
1175 let mut parts = make_parts("tenant1.example.com");
1176 let tape = std::sync::Arc::new(crate::capsule::ReplayEffects::new(
1177 crate::capsule::CapsuleEffects {
1178 tenant: Some(crate::capsule::TenantEffect { id: None }),
1179 ..crate::capsule::CapsuleEffects::default()
1180 },
1181 ));
1182 let result = crate::capsule::with_effect_tape(
1183 std::sync::Arc::clone(&tape),
1184 extract_tenant_from_parts(&mut parts, &config),
1185 )
1186 .await;
1187 assert!(result.is_err(), "{result:?}");
1188 assert_eq!(tape.divergences().len(), 1);
1189 }
1190
1191 #[cfg(feature = "reporting")]
1194 #[tokio::test]
1195 async fn a_failed_tenant_lookup_is_recorded() {
1196 let config = subdomain_config();
1197 let mut parts = make_parts("example.com");
1198 let scope = std::sync::Arc::new(crate::capsule::CaptureScope::new(
1199 "tenant-test".to_owned(),
1200 std::sync::Arc::new(crate::capsule::CaptureSettings::default()),
1201 std::sync::Arc::new(crate::log::filter::ParameterFilter::default()),
1202 ));
1203 let result = crate::capsule::capture::with_capture_scope(
1204 std::sync::Arc::clone(&scope),
1205 extract_tenant_from_parts(&mut parts, &config),
1206 )
1207 .await;
1208 assert!(result.is_err(), "{result:?}");
1209 assert_eq!(
1210 scope.effects_snapshot().tenant,
1211 Some(crate::capsule::TenantEffect { id: None })
1212 );
1213 }
1214
1215 #[tokio::test]
1218 async fn subdomain_uses_resolved_host_from_extension() {
1219 let config = subdomain_config();
1220 let mut parts = make_parts_with_identity("internal.cluster.local", "tenant1.example.com");
1222 let result = extract_tenant_from_parts(&mut parts, &config).await;
1223 assert_eq!(result.unwrap(), "tenant1");
1224 }
1225
1226 #[tokio::test]
1228 async fn subdomain_uses_resolved_host_with_base_domain() {
1229 let config = subdomain_config_with_base("example.com");
1230 let mut parts = make_parts_with_identity("internal.cluster.local", "acme.example.com");
1231 let result = extract_tenant_from_parts(&mut parts, &config).await;
1232 assert_eq!(result.unwrap(), "acme");
1233 }
1234
1235 #[tokio::test]
1237 async fn subdomain_strips_port_from_resolved_host() {
1238 let config = subdomain_config_with_base("example.com");
1239 let mut parts =
1240 make_parts_with_identity("internal.cluster.local", "tenant2.example.com:8080");
1241 let result = extract_tenant_from_parts(&mut parts, &config).await;
1242 assert_eq!(result.unwrap(), "tenant2");
1243 }
1244
1245 #[tokio::test]
1248 async fn subdomain_falls_back_when_resolved_host_is_none() {
1249 let config = subdomain_config();
1250 let (mut parts, ()) = axum::http::Request::builder()
1251 .uri("http://ignored/")
1252 .header(axum::http::header::HOST, "tenant3.example.com")
1253 .body(())
1254 .unwrap()
1255 .into_parts();
1256 parts.extensions.insert(ResolvedClientIdentity {
1257 addr: None,
1258 host: None,
1259 scheme: None,
1260 });
1261 let result = extract_tenant_from_parts(&mut parts, &config).await;
1262 assert_eq!(result.unwrap(), "tenant3");
1263 }
1264
1265 fn public_paths_config(paths: &[&str]) -> crate::config::AutumnConfig {
1266 let mut c = crate::config::AutumnConfig::default();
1267 c.tenancy.public_paths = paths.iter().map(|s| (*s).to_string()).collect();
1268 c
1269 }
1270
1271 #[test]
1273 fn public_path_exact_and_subtree_match() {
1274 let c = public_paths_config(&["/login", "/static"]);
1275 assert!(is_public_path("/login", &c));
1276 assert!(is_public_path("/login/sso", &c));
1277 assert!(is_public_path("/static/css/app.css", &c));
1278 }
1279
1280 #[test]
1282 fn public_path_does_not_bleed_to_adjacent_prefix() {
1283 let c = public_paths_config(&["/login"]);
1284 assert!(!is_public_path("/login-admin", &c));
1285 assert!(!is_public_path("/dashboard", &c));
1286 }
1287
1288 #[cfg(feature = "i18n")]
1295 #[test]
1296 fn public_path_exemption_follows_the_locale_prefix() {
1297 let mut c = public_paths_config(&["/login"]);
1298 c.i18n.locale_prefix_enabled = true;
1299 c.i18n.supported_locales = vec!["en".to_owned(), "es".to_owned()];
1300
1301 assert!(is_public_path("/login", &c), "bare path must stay exempt");
1302 assert!(is_public_path("/en/login", &c));
1303 assert!(is_public_path("/es/login/sso", &c));
1304 assert!(
1305 !is_public_path("/en/dashboard", &c),
1306 "the locale prefix must not exempt an otherwise non-public route"
1307 );
1308 assert!(
1309 !is_public_path("/fr/login", &c),
1310 "an unsupported locale segment must not be stripped"
1311 );
1312 }
1313
1314 #[cfg(feature = "i18n")]
1321 #[test]
1322 fn locale_lookalike_prefix_is_not_stripped_when_it_matches_an_exclusion() {
1323 let mut c = public_paths_config(&["/login"]);
1324 c.i18n.locale_prefix_enabled = true;
1325 c.i18n.supported_locales = vec!["en".to_owned()];
1326 c.i18n.locale_prefix_exclude = vec!["/en".to_owned()];
1327
1328 assert!(
1329 !is_public_path("/en/login", &c),
1330 "/en/login matches the /en exclusion, so it must stay tenant-scoped, \
1331 not be treated as the locale-prefixed /login"
1332 );
1333 }
1334
1335 #[cfg(feature = "i18n")]
1338 #[test]
1339 fn login_redirect_exemption_follows_the_locale_prefix() {
1340 let mut c = crate::config::AutumnConfig::default();
1341 c.tenancy.login_redirect = Some("/login".to_owned());
1342 c.i18n.locale_prefix_enabled = true;
1343 c.i18n.supported_locales = vec!["en".to_owned()];
1344
1345 assert!(is_public_path("/en/login", &c));
1346 }
1347
1348 #[cfg(feature = "i18n")]
1351 #[test]
1352 fn locale_prefix_stripping_is_a_noop_when_routing_disabled() {
1353 let c = public_paths_config(&["/login"]);
1354 assert!(!c.i18n.locale_prefix_enabled);
1355 assert!(!is_public_path("/en/login", &c));
1356 }
1357
1358 #[test]
1360 fn health_paths_are_always_public() {
1361 let c = crate::config::AutumnConfig::default();
1362 assert!(c.tenancy.public_paths.is_empty());
1363 assert!(is_public_path(&c.health.path, &c));
1364 assert!(is_public_path(&c.health.live_path, &c));
1365 assert!(is_public_path(&c.health.ready_path, &c));
1366 assert!(is_public_path(&c.health.startup_path, &c));
1367 }
1368
1369 #[test]
1373 fn probe_paths_match_exactly_not_as_prefix() {
1374 let c = crate::config::AutumnConfig::default();
1375 assert!(is_public_path("/health", &c));
1376 assert!(!is_public_path("/health/history", &c));
1377 assert!(!is_public_path("/live/details", &c));
1378 }
1379
1380 #[test]
1382 fn empty_public_path_entry_does_not_exempt_all() {
1383 let c = public_paths_config(&[""]);
1384 assert!(!is_public_path("/dashboard", &c));
1385 assert!(!is_public_path("/secret", &c));
1386 }
1387
1388 #[test]
1390 fn trailing_slash_entry_matches_subtree() {
1391 let c = public_paths_config(&["/static/"]);
1392 assert!(is_public_path("/static", &c));
1393 assert!(is_public_path("/static/css/app.css", &c));
1394 assert!(!is_public_path("/dashboard", &c));
1395 }
1396
1397 #[test]
1399 fn actuator_prefix_is_always_public() {
1400 let c = crate::config::AutumnConfig::default();
1401 assert!(is_public_path(&c.actuator.prefix, &c));
1402 assert!(is_public_path(
1403 &format!("{}/prometheus", c.actuator.prefix),
1404 &c
1405 ));
1406 assert!(is_public_path(&format!("{}/health", c.actuator.prefix), &c));
1407 }
1408
1409 #[test]
1412 fn actuator_prefix_is_normalized_before_exemption() {
1413 for raw in ["ops/", "/ops/", "/ops", "ops"] {
1414 let mut c = crate::config::AutumnConfig::default();
1415 c.actuator.prefix = raw.to_string();
1416 assert!(
1417 is_public_path("/ops/prometheus", &c),
1418 "prefix {raw:?} should exempt the mounted /ops/prometheus path"
1419 );
1420 assert!(
1421 is_public_path("/ops", &c),
1422 "prefix {raw:?} should exempt the mounted /ops base path"
1423 );
1424 assert!(!is_public_path("/dashboard", &c));
1425 }
1426 }
1427
1428 #[test]
1431 fn root_mounted_actuator_endpoints_are_public() {
1432 let mut c = crate::config::AutumnConfig::default();
1433 c.actuator.prefix = "/".to_string();
1434 c.actuator.prometheus = true;
1435 assert!(is_public_path("/prometheus", &c));
1437 assert!(is_public_path("/metrics", &c));
1438 assert!(!is_public_path("/dashboard", &c));
1440 }
1441
1442 #[test]
1446 fn openapi_path_is_not_auto_exempt() {
1447 let c = crate::config::AutumnConfig::default();
1448 assert!(c.openapi_runtime.enabled);
1449 assert!(!is_public_path(&c.openapi_runtime.path, &c));
1450 let c = public_paths_config(&["/openapi.json"]);
1452 assert!(is_public_path("/openapi.json", &c));
1453 }
1454
1455 #[test]
1458 fn login_redirect_target_is_always_public() {
1459 let mut c = crate::config::AutumnConfig::default();
1460 c.tenancy.login_redirect = Some("/auth/login".to_string());
1461 assert!(is_public_path("/auth/login", &c));
1463 assert!(!is_public_path("/auth/login/sso", &c));
1465 }
1466
1467 #[test]
1470 fn login_redirect_target_with_query_is_public_by_path() {
1471 let mut c = crate::config::AutumnConfig::default();
1472 c.tenancy.login_redirect = Some("/login?next=/dashboard".to_string());
1473 assert!(is_public_path("/login", &c));
1475 assert!(!is_public_path("/dashboard", &c));
1476 }
1477
1478 #[test]
1483 fn login_redirect_absolute_url_is_not_auto_exempt() {
1484 let mut c = crate::config::AutumnConfig::default();
1485 c.tenancy.login_redirect = Some("https://idp.example.com/login".to_string());
1486 assert!(!is_public_path("/login", &c));
1487
1488 c.tenancy.login_redirect = Some("https://app.example.com/login".to_string());
1490 assert!(!is_public_path("/login", &c));
1491
1492 c.tenancy.public_paths = vec!["/login".to_string()];
1494 assert!(is_public_path("/login", &c));
1495 }
1496
1497 #[test]
1500 fn root_public_path_is_preserved() {
1501 let c = public_paths_config(&["/"]);
1502 assert!(is_public_path("/", &c));
1503 assert!(!is_public_path("/dashboard", &c));
1506 }
1507
1508 #[test]
1511 fn empty_builtin_path_does_not_exempt_all() {
1512 let mut c = crate::config::AutumnConfig::default();
1513 c.health.path = String::new();
1514 c.health.live_path = String::new();
1515 c.health.ready_path = String::new();
1516 c.health.startup_path = String::new();
1517 c.actuator.prefix = String::new();
1518 assert!(!is_public_path("/dashboard", &c));
1519 assert!(!is_public_path("/", &c));
1520 }
1521}