1pub mod parser;
29pub mod health;
30pub(crate) mod upstream;
31
32pub use upstream::{BreakerConfig, RetryConfig, StickyConfig};
33pub mod builder;
34pub mod validate;
35pub mod mcp_gateway;
36pub mod a2a_gateway;
37pub mod llm;
38#[cfg(feature = "sso")]
39#[doc(hidden)]
40pub mod oidc_route;
41
42#[cfg(test)]
43mod tests;
44
45use std::sync::atomic::{AtomicBool, Ordering};
46use std::sync::{Arc, RwLock};
47
48use crate::app::App;
49use crate::application::Application;
50use crate::core::New;
51use crate::request::Request;
52use crate::header::Header;
53use crate::response::{Response, STATUS_CODE_REASON_PHRASE};
54use crate::router::{host_matches, request_host};
55use crate::server::ConnectionInfo;
56use crate::server_config::ServerConfig;
57
58#[derive(Debug, Clone)]
61pub struct ProxyConfig {
62 pub upstreams: Vec<UpstreamConfig>,
63 pub routes: Vec<RouteConfig>,
64 pub tcp_proxies: Vec<TcpProxyConfig>,
65 pub udp_proxies: Vec<UdpProxyConfig>,
66 pub ws_proxies: Vec<WsProxyConfig>,
67 pub global_middleware: MiddlewareConfig,
68 pub authz: Option<AuthzConfig>,
70}
71
72#[derive(Debug, Clone, Default)]
74pub struct AuthzConfig {
75 pub store: String,
77 pub tuples: Vec<String>,
79 pub types: Vec<(String, Vec<(String, String)>)>,
81}
82
83#[derive(Debug, Clone)]
85pub struct RouteAuthzConfig {
86 pub relation: String,
87 pub object: String,
89}
90
91#[derive(Debug, Clone)]
92pub struct UpstreamConfig {
93 pub name: String,
94 pub backends: Vec<String>,
95 pub strategy: String, pub health_check: Option<HealthCheckConfig>,
97 pub tls: bool,
101 pub tls_ca_file: Option<String>,
105 pub tls_client_cert_file: Option<String>,
109 pub tls_client_key_file: Option<String>,
110 pub weights: Vec<u32>,
112 pub sticky: Option<StickyConfig>,
114 pub circuit_breaker: Option<BreakerConfig>,
117 pub retry: Option<RetryConfig>,
119 pub protocol: String,
124}
125
126#[derive(Debug, Clone)]
127pub struct HealthCheckConfig {
128 pub path: String,
129 pub interval_secs: u64,
130 pub timeout_ms: u64,
131 pub healthy_threshold: u32,
132 pub unhealthy_threshold: u32,
133}
134
135#[derive(Debug, Clone)]
136pub struct RouteConfig {
137 pub name: String,
138 pub match_: MatchConfig,
139 pub action: ActionConfig,
140 pub middleware: MiddlewareConfig,
141}
142
143#[derive(Debug, Clone, Default)]
144pub struct MatchConfig {
145 pub host: Option<String>,
146 pub path: Option<String>,
147 pub method: Option<String>,
148 pub content_type: Option<String>,
149}
150
151#[derive(Debug, Clone, Default, PartialEq, Eq)]
153pub enum UpstreamHost {
154 #[default]
156 Backend,
157 Client,
160 Fixed(String),
162}
163
164#[derive(Debug, Clone)]
165pub enum ActionConfig {
166 Proxy {
167 upstream: String,
168 connect_timeout_ms: u64,
169 read_timeout_ms: u64,
170 strip_path_prefix: Option<String>,
171 add_path_prefix: Option<String>,
172 upstream_host: UpstreamHost,
174 },
175 Grpc {
176 upstream: String,
177 connect_timeout_ms: u64,
178 read_timeout_ms: u64,
179 },
180 Static {
181 root: String,
182 index: Vec<String>,
183 },
184 Redirect {
185 location: String,
186 status: u16,
187 },
188 Respond {
189 status: u16,
190 body: String,
191 content_type: String,
192 },
193 FastCgi(Box<FastCgiAction>),
195 Mcp(Box<mcp_gateway::McpGatewayConfig>),
197 A2a(Box<a2a_gateway::A2aGatewayConfig>),
199 Llm(Box<llm::LlmConfig>),
201 Unknown(String),
202}
203
204#[derive(Debug, Clone, Default, PartialEq)]
206pub struct FastCgiAction {
207 pub address: String,
208 pub root: String,
209 pub index: Vec<String>,
210 pub split_path: String,
211 pub front_controller: String,
212 pub params: Vec<(String, String)>,
213 pub connect_timeout_ms: u64,
214 pub read_timeout_ms: u64,
215 pub max_idle: u64,
216}
217
218#[derive(Debug, Clone, PartialEq)]
222pub struct WafSettings {
223 pub config: crate::waf::WafConfig,
224 pub errors: Vec<String>,
225}
226
227#[derive(Debug, Clone, Default)]
228pub struct MiddlewareConfig {
229 pub rate_limit: Option<RateLimitConfig>,
230 pub cache: Option<CacheConfig>,
231 pub auth: Option<AuthConfig>,
232 pub rewrite_request: Vec<RewriteRuleConfig>,
233 pub rewrite_response: Vec<RewriteRuleConfig>,
234 pub ip_allow: Vec<String>,
235 pub ip_deny: Vec<String>,
236 pub timeout_ms: Option<u64>,
241 pub max_body_size: Option<u64>,
257 pub wasm: Option<WasmConfig>,
260 pub early_hints: Vec<String>,
263 pub authz: Option<RouteAuthzConfig>,
266 pub mirror: Option<MirrorConfig>,
269 pub waf: Option<WafSettings>,
271}
272
273#[derive(Debug, Clone, Default, PartialEq)]
275pub struct MirrorConfig {
276 pub upstream: String,
278 pub backends: Vec<String>,
280 pub percent: f64,
282}
283
284#[derive(Debug, Clone, Default, PartialEq)]
286pub struct WasmConfig {
287 pub module: String,
289 pub fuel: u64,
290 pub memory_mb: u64,
291 pub config: String,
293}
294
295#[derive(Debug, Clone)]
296pub struct RateLimitConfig {
297 pub max_requests: u32,
298 pub window_secs: u64,
299}
300
301#[derive(Debug, Clone, Default)]
302pub struct CacheConfig {
303 pub ttl_secs: u64,
304 pub vary_by: Vec<String>,
305 pub dir: String,
307 pub max_size: String,
309 pub purge_token_env: String,
312}
313
314#[derive(Debug, Clone, PartialEq)]
315pub enum AuthConfig {
316 Basic { htpasswd_file: String },
321 Jwt { secret_env: String, previous_secret_env: String },
326 Bearer { token_env: String },
327 Oidc(Box<OidcRouteAuth>),
329 Forward { url: String, copy_headers: Vec<String>, timeout_ms: u64 },
332 Unknown(String),
335}
336
337#[derive(Debug, Clone, Default, PartialEq)]
339pub struct OidcRouteAuth {
340 pub issuer: String,
342 pub provider: String,
344 pub tenant: String,
346 pub client_id: String,
347 pub client_secret_env: String,
348 pub redirect_uri: String,
350 pub scopes: Vec<String>,
351 pub allowed_emails: Vec<String>,
352 pub allowed_domains: Vec<String>,
353 pub allowed_groups: Vec<String>,
354 pub groups_claim: String,
355 pub cookie_secret_env: String,
357 pub session_ttl_secs: u64,
358 pub exclude: Vec<String>,
359 pub forward_headers: bool,
360 pub path_prefix: String,
363}
364
365#[derive(Debug, Clone, Default)]
366pub struct RewriteRuleConfig {
367 pub type_: String,
368 pub name: Option<String>,
369 pub value: Option<String>,
370 pub prefix: Option<String>,
371 pub from: Option<String>,
372 pub to: Option<String>,
373 pub code: Option<u16>,
374 pub reason: Option<String>,
375}
376
377#[derive(Debug, Clone)]
378pub struct TcpProxyConfig {
379 pub name: String,
380 pub listen: String,
381 pub backends: Vec<String>,
382 pub connect_timeout_ms: u64,
383}
384
385#[derive(Debug, Clone)]
386pub struct UdpProxyConfig {
387 pub name: String,
388 pub listen: String,
389 pub backends: Vec<String>,
390 pub reply_timeout_ms: u64,
391 pub buffer_size: usize,
392}
393
394#[derive(Debug, Clone)]
395pub struct WsProxyConfig {
396 pub name: String,
397 pub listen: String,
398 pub backends: Vec<String>,
399 pub connect_timeout_ms: u64,
400 pub read_timeout_ms: u64,
401 pub health_check: Option<HealthCheckConfig>,
408}
409
410fn merge_action_keys(map: &mut parser::SectionMap, action_section: &str, action_type: &str) {
419 if action_type.is_empty() {
420 return;
421 }
422 let short_form: Vec<(String, String)> = match map.get(action_section) {
423 Some(entries) => entries.iter().filter(|(k, _)| k != "type").cloned().collect(),
424 None => return,
425 };
426 let nested = map.entry(format!("{action_section}.{action_type}")).or_default();
427 for (key, value) in short_form {
428 if !nested.iter().any(|(k, _)| *k == key) {
429 nested.push((key, value));
430 }
431 }
432}
433
434impl ProxyConfig {
435 pub fn is_proxy_mode() -> bool {
439 let path = config_file_path();
440 match std::fs::read_to_string(&path) {
441 Ok(contents) => {
442 contents.contains("[[route]]") || contents.contains("[[upstream]]")
443 }
444 Err(_) => false,
445 }
446 }
447
448 pub fn load() -> Self {
450 let path = config_file_path();
451 let contents = std::fs::read_to_string(&path).unwrap_or_default();
452 Self::from_str(&contents)
453 }
454
455 pub fn from_str(toml: &str) -> Self {
457 use parser::{get_array, get_str, get_u32, get_u64, section_exists};
458
459 let mut map = parser::parse(toml);
460
461 let mut upstreams = Vec::new();
463 let mut i = 0;
464 loop {
465 let sec = format!("upstream[{}]", i);
466 if !section_exists(&map, &sec) {
467 break;
468 }
469 let name = get_str(&map, &sec, "name");
470 let backends = get_array(&map, &sec, "backends");
471 let strategy = {
472 let s = get_str(&map, &sec, "strategy");
473 if s.is_empty() { "round_robin".to_string() } else { s }
474 };
475 let hc_sec = format!("{}.health_check", sec);
476 let health_check = if section_exists(&map, &hc_sec) {
477 Some(HealthCheckConfig {
478 path: {
479 let p = get_str(&map, &hc_sec, "path");
480 if p.is_empty() { "/health".to_string() } else { p }
481 },
482 interval_secs: get_u64(&map, &hc_sec, "interval_secs", 30),
483 timeout_ms: get_u64(&map, &hc_sec, "timeout_ms", 5000),
484 healthy_threshold: get_u32(&map, &hc_sec, "healthy_threshold", 2),
485 unhealthy_threshold: get_u32(&map, &hc_sec, "unhealthy_threshold", 3),
486 })
487 } else {
488 None
489 };
490 let tls = backends.iter().any(|b| ["https://", "h2s://", "grpcs://"].iter().any(|s| b.starts_with(s)));
491 let tls_ca_file = Some(get_str(&map, &sec, "tls_ca_file")).filter(|v| !v.is_empty());
492 let tls_client_cert_file = Some(get_str(&map, &sec, "tls_client_cert_file")).filter(|v| !v.is_empty());
493 let tls_client_key_file = Some(get_str(&map, &sec, "tls_client_key_file")).filter(|v| !v.is_empty());
494 let weights: Vec<u32> = get_array(&map, &sec, "weights").iter().map(|w| w.trim().parse().unwrap_or(0)).collect();
495 let sticky_sec = format!("{}.sticky", sec);
496 let sticky = section_exists(&map, &sticky_sec).then(|| StickyConfig {
497 cookie: Some(get_str(&map, &sticky_sec, "cookie")).filter(|c| !c.is_empty()).unwrap_or_else(|| "rws_backend".to_string()),
498 max_age_secs: Some(get_u64(&map, &sticky_sec, "max_age_secs", 0)).filter(|s| *s > 0),
499 });
500 let cb_sec = format!("{}.circuit_breaker", sec);
501 let circuit_breaker = section_exists(&map, &cb_sec).then(|| BreakerConfig {
502 failure_threshold: get_u32(&map, &cb_sec, "failure_threshold", 5),
503 recovery_secs: get_u64(&map, &cb_sec, "recovery_secs", 30),
504 half_open_probes: get_u32(&map, &cb_sec, "half_open_probes", 1),
505 });
506 let retry_sec = format!("{}.retry", sec);
507 let retry = section_exists(&map, &retry_sec).then(|| {
508 let status: Vec<i16> = get_array(&map, &retry_sec, "status").iter().filter_map(|s| s.trim().parse().ok()).collect();
509 RetryConfig { attempts: get_u32(&map, &retry_sec, "attempts", 1), status: if status.is_empty() { vec![502, 503, 504] } else { status } }
510 });
511 upstreams.push(UpstreamConfig {
512 name, backends, strategy, health_check, tls, tls_ca_file, tls_client_cert_file, tls_client_key_file,
513 weights, sticky, circuit_breaker, retry,
514 protocol: get_str(&map, &sec, "protocol"),
515 });
516 i += 1;
517 }
518
519 let mut routes = Vec::new();
521 let mut i = 0;
522 loop {
523 let sec = format!("route[{}]", i);
524 if !section_exists(&map, &sec) {
525 break;
526 }
527 let name = get_str(&map, &sec, "name");
528
529 let m_sec = format!("{}.match", sec);
531 let match_ = MatchConfig {
532 host: {
533 let h = get_str(&map, &m_sec, "host");
534 if h.is_empty() { None } else { Some(h) }
535 },
536 path: {
537 let p = get_str(&map, &m_sec, "path");
538 if p.is_empty() { None } else { Some(p) }
539 },
540 method: {
541 let m = get_str(&map, &m_sec, "method");
542 if m.is_empty() { None } else { Some(m.to_uppercase()) }
543 },
544 content_type: {
545 let c = get_str(&map, &m_sec, "content_type");
546 if c.is_empty() { None } else { Some(c) }
547 },
548 };
549
550 let a_sec = format!("{}.action", sec);
552 let action_type = get_str(&map, &a_sec, "type");
553 merge_action_keys(&mut map, &a_sec, &action_type);
554 let action = match action_type.as_str() {
555 "proxy" => {
556 let p_sec = format!("{}.action.proxy", sec);
557 ActionConfig::Proxy {
558 upstream: get_str(&map, &p_sec, "upstream"),
559 connect_timeout_ms: get_u64(&map, &p_sec, "connect_timeout_ms", 5000),
560 read_timeout_ms: get_u64(&map, &p_sec, "read_timeout_ms", 30000),
561 strip_path_prefix: {
562 let v = get_str(&map, &p_sec, "strip_path_prefix");
563 if v.is_empty() { None } else { Some(v) }
564 },
565 add_path_prefix: {
566 let v = get_str(&map, &p_sec, "add_path_prefix");
567 if v.is_empty() { None } else { Some(v) }
568 },
569 upstream_host: {
570 let fixed = get_str(&map, &p_sec, "upstream_host");
571 if !fixed.is_empty() {
572 UpstreamHost::Fixed(fixed)
573 } else if get_str(&map, &p_sec, "preserve_host") == "true" {
574 UpstreamHost::Client
575 } else {
576 UpstreamHost::Backend
577 }
578 },
579 }
580 }
581 "grpc" => {
582 let p_sec = format!("{}.action.grpc", sec);
583 ActionConfig::Grpc {
584 upstream: get_str(&map, &p_sec, "upstream"),
585 connect_timeout_ms: get_u64(&map, &p_sec, "connect_timeout_ms", 5000),
586 read_timeout_ms: get_u64(&map, &p_sec, "read_timeout_ms", 30000),
587 }
588 }
589 "static" => {
590 let s_sec = format!("{}.action.static", sec);
591 ActionConfig::Static {
592 root: get_str(&map, &s_sec, "root"),
593 index: get_array(&map, &s_sec, "index"),
594 }
595 }
596 "redirect" => {
597 let r_sec = format!("{}.action.redirect", sec);
598 ActionConfig::Redirect {
599 location: get_str(&map, &r_sec, "location"),
600 status: get_u64(&map, &r_sec, "status", 301) as u16,
601 }
602 }
603 "respond" => {
604 let r_sec = format!("{}.action.respond", sec);
605 ActionConfig::Respond {
606 status: get_u64(&map, &r_sec, "status", 200) as u16,
607 body: get_str(&map, &r_sec, "body"),
608 content_type: {
609 let ct = get_str(&map, &r_sec, "content_type");
610 if ct.is_empty() { "text/plain".to_string() } else { ct }
611 },
612 }
613 }
614 "fastcgi" => {
615 let f_sec = format!("{}.action.fastcgi", sec);
616 let or = |key: &str, default: &str| parser::get(&map, &f_sec, key).unwrap_or_else(|| default.to_string());
617 let mut params: Vec<(String, String)> = Vec::new();
618 for table in [format!("{a_sec}.params"), format!("{f_sec}.params")] {
619 for (k, v) in map.get(&table).cloned().unwrap_or_default() {
620 params.retain(|(name, _)| *name != k);
621 params.push((k, v));
622 }
623 }
624 let index = get_array(&map, &f_sec, "index");
625 ActionConfig::FastCgi(Box::new(FastCgiAction {
626 address: get_str(&map, &f_sec, "address"),
627 root: get_str(&map, &f_sec, "root"),
628 index: if index.is_empty() { vec!["index.php".to_string()] } else { index },
629 split_path: or("split_path", ".php"),
630 front_controller: or("front_controller", "index.php"),
631 params,
632 connect_timeout_ms: get_u64(&map, &f_sec, "connect_timeout_ms", 5000),
633 read_timeout_ms: get_u64(&map, &f_sec, "read_timeout_ms", 60000),
634 max_idle: get_u64(&map, &f_sec, "max_idle", 16),
635 }))
636 }
637 "mcp" => ActionConfig::Mcp(Box::new(mcp_gateway::parse(&map, &a_sec))),
638 "a2a" => ActionConfig::A2a(Box::new(a2a_gateway::parse(&map, &a_sec))),
639 "llm" => ActionConfig::Llm(Box::new(llm::parse(&map, &a_sec))),
640 other => ActionConfig::Unknown(other.to_string()),
641 };
642
643 let mw_sec = format!("{}.middleware", sec);
645 let middleware = parse_middleware_config(&map, &mw_sec, i);
646
647 routes.push(RouteConfig { name, match_, action, middleware });
648 i += 1;
649 }
650
651 let mut tcp_proxies = Vec::new();
653 let mut i = 0;
654 loop {
655 let sec = format!("tcp_proxy[{}]", i);
656 if !section_exists(&map, &sec) {
657 break;
658 }
659 tcp_proxies.push(TcpProxyConfig {
660 name: get_str(&map, &sec, "name"),
661 listen: get_str(&map, &sec, "listen"),
662 backends: get_array(&map, &sec, "backends"),
663 connect_timeout_ms: get_u64(&map, &sec, "connect_timeout_ms", 5000),
664 });
665 i += 1;
666 }
667
668 let mut udp_proxies = Vec::new();
670 let mut i = 0;
671 loop {
672 let sec = format!("udp_proxy[{}]", i);
673 if !section_exists(&map, &sec) {
674 break;
675 }
676 udp_proxies.push(UdpProxyConfig {
677 name: get_str(&map, &sec, "name"),
678 listen: get_str(&map, &sec, "listen"),
679 backends: get_array(&map, &sec, "backends"),
680 reply_timeout_ms: get_u64(&map, &sec, "reply_timeout_ms", 5000),
681 buffer_size: get_u64(&map, &sec, "buffer_size", 65536) as usize,
682 });
683 i += 1;
684 }
685
686 let mut ws_proxies = Vec::new();
688 let mut i = 0;
689 loop {
690 let sec = format!("ws_proxy[{}]", i);
691 if !section_exists(&map, &sec) {
692 break;
693 }
694 let hc_sec = format!("{}.health_check", sec);
695 let health_check = if section_exists(&map, &hc_sec) {
696 Some(HealthCheckConfig {
697 path: {
698 let p = get_str(&map, &hc_sec, "path");
699 if p.is_empty() { "/health".to_string() } else { p }
700 },
701 interval_secs: get_u64(&map, &hc_sec, "interval_secs", 30),
702 timeout_ms: get_u64(&map, &hc_sec, "timeout_ms", 5000),
703 healthy_threshold: get_u32(&map, &hc_sec, "healthy_threshold", 2),
704 unhealthy_threshold: get_u32(&map, &hc_sec, "unhealthy_threshold", 3),
705 })
706 } else {
707 None
708 };
709 ws_proxies.push(WsProxyConfig {
710 name: get_str(&map, &sec, "name"),
711 listen: get_str(&map, &sec, "listen"),
712 backends: get_array(&map, &sec, "backends"),
713 connect_timeout_ms: get_u64(&map, &sec, "connect_timeout_ms", 5000),
714 read_timeout_ms: get_u64(&map, &sec, "read_timeout_ms", 30000),
715 health_check,
716 });
717 i += 1;
718 }
719
720 let global_middleware = parse_middleware_config(&map, "middleware", usize::MAX);
722
723 let authz = map.keys().any(|k| k == "authz" || k.starts_with("authz.")).then(|| {
725 let mut type_sections: Vec<&String> = map.keys().filter(|k| k.starts_with("authz.type.")).collect();
726 type_sections.sort();
727 AuthzConfig {
728 store: get_str(&map, "authz", "store"),
729 tuples: get_array(&map, "authz", "tuples"),
730 types: type_sections
731 .into_iter()
732 .map(|sec| {
733 let relations = map[sec].iter().map(|(k, _)| (k.clone(), get_str(&map, sec, k))).collect();
734 (sec["authz.type.".len()..].to_string(), relations)
735 })
736 .collect(),
737 }
738 });
739
740 ProxyConfig {
741 upstreams,
742 routes,
743 tcp_proxies,
744 udp_proxies,
745 ws_proxies,
746 global_middleware,
747 authz,
748 }
749 }
750}
751
752fn parse_middleware_config(
755 map: &parser::SectionMap,
756 mw_sec: &str,
757 route_idx: usize,
758) -> MiddlewareConfig {
759 use parser::{get_array, get_str, get_u32, get_u64, section_exists};
760
761 let rl_sec = format!("{}.rate_limit", mw_sec);
762 let rate_limit = if section_exists(map, &rl_sec) {
763 Some(RateLimitConfig {
764 max_requests: get_u32(map, &rl_sec, "max_requests", 1000),
765 window_secs: get_u64(map, &rl_sec, "window_secs", 60),
766 })
767 } else {
768 None
769 };
770
771 let c_sec = format!("{}.cache", mw_sec);
772 let cache = if section_exists(map, &c_sec) {
773 Some(CacheConfig {
774 ttl_secs: get_u64(map, &c_sec, "ttl_secs", 60),
775 vary_by: get_array(map, &c_sec, "vary_by"),
776 dir: get_str(map, &c_sec, "dir"),
777 max_size: get_str(map, &c_sec, "max_size"),
778 purge_token_env: get_str(map, &c_sec, "purge_token_env"),
779 })
780 } else {
781 None
782 };
783
784 let a_sec = format!("{}.auth", mw_sec);
785 let auth = if section_exists(map, &a_sec) {
786 let auth_type = get_str(map, &a_sec, "type");
787 match auth_type.as_str() {
788 "bearer" => Some(AuthConfig::Bearer {
789 token_env: get_str(map, &a_sec, "token_env"),
790 }),
791 "jwt" => Some(AuthConfig::Jwt {
792 secret_env: get_str(map, &a_sec, "secret_env"),
793 previous_secret_env: get_str(map, &a_sec, "previous_secret_env"),
794 }),
795 "basic" => Some(AuthConfig::Basic {
796 htpasswd_file: get_str(map, &a_sec, "htpasswd_file"),
797 }),
798 "oidc" => Some(AuthConfig::Oidc(Box::new(OidcRouteAuth {
799 issuer: get_str(map, &a_sec, "issuer").trim_end_matches('/').to_string(),
800 provider: get_str(map, &a_sec, "provider").to_ascii_lowercase(),
801 tenant: get_str(map, &a_sec, "tenant"),
802 client_id: get_str(map, &a_sec, "client_id"),
803 client_secret_env: get_str(map, &a_sec, "client_secret_env"),
804 redirect_uri: get_str(map, &a_sec, "redirect_uri"),
805 scopes: get_array(map, &a_sec, "scopes"),
806 allowed_emails: get_array(map, &a_sec, "allowed_emails"),
807 allowed_domains: get_array(map, &a_sec, "allowed_domains"),
808 allowed_groups: get_array(map, &a_sec, "allowed_groups"),
809 groups_claim: get_str(map, &a_sec, "groups_claim"),
810 cookie_secret_env: get_str(map, &a_sec, "cookie_secret_env"),
811 session_ttl_secs: get_u64(map, &a_sec, "session_ttl_secs", 86_400),
812 exclude: get_array(map, &a_sec, "exclude"),
813 forward_headers: get_str(map, &a_sec, "forward_headers") != "false",
814 path_prefix: String::new(),
815 }))),
816 "forward" => Some(AuthConfig::Forward {
817 url: get_str(map, &a_sec, "url"),
818 copy_headers: get_array(map, &a_sec, "copy_headers"),
819 timeout_ms: get_u64(map, &a_sec, "timeout_ms", 5000),
820 }),
821 other => Some(AuthConfig::Unknown(other.to_string())),
822 }
823 } else {
824 None
825 };
826
827 let rewrite_request = collect_rewrite_rules(map, mw_sec, "request");
833 let rewrite_response = collect_rewrite_rules(map, mw_sec, "response");
834
835 let ip_sec = format!("{}.ip_filter", mw_sec);
836 let ip_allow = if section_exists(map, &ip_sec) {
837 get_array(map, &ip_sec, "allow")
838 } else {
839 vec![]
840 };
841 let ip_deny = if section_exists(map, &ip_sec) {
842 get_array(map, &ip_sec, "deny")
843 } else {
844 vec![]
845 };
846
847 let _ = route_idx; let timeout_ms = match get_u64(map, mw_sec, "timeout_ms", 0) {
853 0 => None,
854 ms => Some(ms),
855 };
856
857 let max_body_size = match get_u64(map, mw_sec, "max_body_size", 0) {
861 0 => None,
862 n => Some(n),
863 };
864
865 let w_sec = format!("{}.wasm", mw_sec);
866 let wasm = section_exists(map, &w_sec).then(|| WasmConfig {
867 module: get_str(map, &w_sec, "module"),
868 fuel: get_u64(map, &w_sec, "fuel", 50_000_000),
869 memory_mb: get_u64(map, &w_sec, "memory_mb", 16),
870 config: get_str(map, &w_sec, "config"),
871 });
872
873 let m_sec = format!("{}.mirror", mw_sec);
874 let mirror = section_exists(map, &m_sec).then(|| MirrorConfig {
875 upstream: get_str(map, &m_sec, "upstream"),
876 backends: get_array(map, &m_sec, "backends"),
877 percent: {
878 let raw = get_str(map, &m_sec, "percent");
879 if raw.is_empty() { 100.0 } else { raw.parse::<f64>().unwrap_or(f64::NAN) }
880 },
881 });
882
883 let waf = parse_waf(map, &format!("{mw_sec}.waf"));
884
885 MiddlewareConfig {
886 rate_limit, cache, auth, rewrite_request, rewrite_response, ip_allow, ip_deny,
887 timeout_ms, max_body_size, wasm, mirror, waf,
888 early_hints: get_array(map, mw_sec, "early_hints"),
889 authz: section_exists(map, &format!("{mw_sec}.authz")).then(|| RouteAuthzConfig {
890 relation: get_str(map, &format!("{mw_sec}.authz"), "relation"),
891 object: get_str(map, &format!("{mw_sec}.authz"), "object"),
892 }),
893 }
894}
895
896fn parse_waf(map: &parser::SectionMap, sec: &str) -> Option<WafSettings> {
899 use crate::waf::{Action, CustomRule, Exclusion, Field, Mode, WafConfig};
900 use parser::{get_array, get_str, get_u64, section_exists};
901 if !section_exists(map, sec) {
902 return None;
903 }
904 let mut errors = Vec::new();
905 let defaults = WafConfig::default();
906 let mode_text = get_str(map, sec, "mode");
907 let mode = Mode::parse(&mode_text).unwrap_or_else(|| {
908 errors.push(format!("[waf] mode = {mode_text:?}: expected block, detect or off"));
909 Mode::Block
910 });
911 let paranoia = get_u64(map, sec, "paranoia", 1);
912 if !(1..=3).contains(¶noia) {
913 errors.push(format!("[waf] paranoia = {paranoia}: expected 1, 2 or 3"));
914 }
915 let mut rules = Vec::new();
916 let mut j = 0;
917 loop {
918 let rsec = format!("{sec}.rule[{j}]");
919 if !section_exists(map, &rsec) {
920 break;
921 }
922 j += 1;
923 let id = get_u64(map, &rsec, "id", 0) as u32;
924 let field_text = get_str(map, &rsec, "field");
925 let Some(field) = Field::parse(&field_text) else {
926 errors.push(format!("[waf] rule {id}: field = {field_text:?}: expected path, query, args, arg:<name>, header:<name>, method, ip or body"));
927 continue;
928 };
929 let action_text = get_str(map, &rsec, "action");
930 let action = match action_text.to_ascii_lowercase().as_str() {
931 "deny" | "" => Action::Deny,
932 "allow" => Action::Allow,
933 "log" => Action::Log,
934 "score" => Action::Score(get_u64(map, &rsec, "score", 5) as u32),
935 other => {
936 errors.push(format!("[waf] rule {id}: action = {other:?}: expected deny, allow, log or score"));
937 continue;
938 }
939 };
940 let rule = CustomRule {
941 id,
942 field,
943 contains: get_array(map, &rsec, "contains"),
944 prefix: get_array(map, &rsec, "prefix"),
945 equals: get_array(map, &rsec, "equals"),
946 cidr: get_array(map, &rsec, "cidr"),
947 action,
948 };
949 if rule.contains.is_empty() && rule.prefix.is_empty() && rule.equals.is_empty() && rule.cidr.is_empty() {
950 errors.push(format!("[waf] rule {id}: needs contains, prefix, equals or cidr"));
951 continue;
952 }
953 if let Some(bad) = rule.cidr.iter().find(|c| !crate::ip_filter::is_valid_entry(c)) {
954 errors.push(format!("[waf] rule {id}: cidr {bad:?} is not an address or range"));
955 }
956 rules.push(rule);
957 }
958 Some(WafSettings {
959 config: WafConfig {
960 mode,
961 paranoia: paranoia.clamp(1, 3) as u8,
962 threshold: get_u64(map, sec, "threshold", defaults.threshold as u64).max(1) as u32,
963 inspect_body_bytes: get_u64(map, sec, "inspect_body_bytes", defaults.inspect_body_bytes as u64) as usize,
964 max_args: get_u64(map, sec, "max_args", defaults.max_args as u64) as usize,
965 exclude: get_array(map, sec, "exclude").iter().map(|e| Exclusion::parse(e)).collect(),
966 rules,
967 },
968 errors,
969 })
970}
971
972fn collect_rewrite_rules(
973 map: &parser::SectionMap,
974 mw_sec: &str,
975 direction: &str,
976) -> Vec<RewriteRuleConfig> {
977 use parser::{get_str, get_u64};
978
979 let mut rules = Vec::new();
980 let mut j = 0;
981 loop {
982 let rsec = format!("{}.rewrite.{}[{}]", mw_sec, direction, j);
983 if !parser::section_exists(map, &rsec) {
984 break;
985 }
986 let code_val = get_u64(map, &rsec, "code", 0);
987 rules.push(RewriteRuleConfig {
988 type_: get_str(map, &rsec, "type"),
989 name: {
990 let v = get_str(map, &rsec, "name");
991 if v.is_empty() { None } else { Some(v) }
992 },
993 value: {
994 let v = get_str(map, &rsec, "value");
995 if v.is_empty() { None } else { Some(v) }
996 },
997 prefix: {
998 let v = get_str(map, &rsec, "prefix");
999 if v.is_empty() { None } else { Some(v) }
1000 },
1001 from: {
1002 let v = get_str(map, &rsec, "from");
1003 if v.is_empty() { None } else { Some(v) }
1004 },
1005 to: {
1006 let v = get_str(map, &rsec, "to");
1007 if v.is_empty() { None } else { Some(v) }
1008 },
1009 code: if code_val == 0 { None } else { Some(code_val as u16) },
1010 reason: {
1011 let v = get_str(map, &rsec, "reason");
1012 if v.is_empty() { None } else { Some(v) }
1013 },
1014 });
1015 j += 1;
1016 }
1017 rules
1018}
1019
1020#[doc(hidden)]
1021pub fn config_file_path() -> String {
1022 std::env::var("RWS_CONFIG_FILE").unwrap_or_else(|_| "rws.config.toml".to_string())
1023}
1024
1025#[doc(hidden)]
1029pub struct CompiledRoute {
1030 #[doc(hidden)]
1031 pub matcher: RouteMatcher,
1032 #[doc(hidden)]
1034 pub handler: Arc<dyn Application + Send + Sync>,
1035 #[cfg_attr(not(feature = "http2"), allow(dead_code))]
1039 #[doc(hidden)]
1040 pub async_proxy: Option<Arc<DynamicProxy>>,
1041}
1042
1043#[derive(Clone, Default)]
1045#[doc(hidden)]
1046pub struct RouteMatcher {
1047 #[doc(hidden)]
1049 pub host: Option<String>,
1050 #[doc(hidden)]
1052 pub path_prefix: Option<String>,
1053 #[doc(hidden)]
1055 pub path_exact: Option<String>,
1056 #[doc(hidden)]
1058 pub method: Option<String>,
1059 #[doc(hidden)]
1061 pub content_type_prefix: Option<String>,
1062}
1063
1064impl RouteMatcher {
1065 #[doc(hidden)]
1066 pub fn from_match_config(cfg: &MatchConfig) -> Self {
1067 let (path_prefix, path_exact) = match &cfg.path {
1068 Some(p) if p.ends_with('*') => {
1069 let stripped = p.trim_end_matches('*').to_string();
1071 (Some(stripped), None)
1072 }
1073 Some(p) => (None, Some(p.clone())),
1074 None => (None, None),
1075 };
1076 let content_type_prefix = cfg.content_type.as_ref().map(|ct| {
1077 if ct.ends_with('*') {
1078 ct.trim_end_matches('*').to_string()
1079 } else {
1080 ct.clone()
1081 }
1082 });
1083 RouteMatcher {
1084 host: cfg.host.clone(),
1085 path_prefix,
1086 path_exact,
1087 method: cfg.method.clone(),
1088 content_type_prefix,
1089 }
1090 }
1091
1092 #[doc(hidden)]
1094 pub fn matches(&self, request: &Request, conn: &ConnectionInfo) -> bool {
1095 if let Some(ref expected_host) = self.host {
1097 if !host_matches(expected_host, request_host(request, conn)) {
1098 return false;
1099 }
1100 }
1101
1102 if let Some(ref m) = self.method {
1104 if request.method.to_uppercase() != m.as_str() {
1105 return false;
1106 }
1107 }
1108
1109 let path = request.request_uri.split('?').next().unwrap_or(&request.request_uri);
1111 if let Some(ref prefix) = self.path_prefix {
1112 if !path.starts_with(prefix.as_str()) {
1113 return false;
1114 }
1115 } else if let Some(ref exact) = self.path_exact {
1116 if path != exact.as_str() {
1117 return false;
1118 }
1119 }
1120
1121 if let Some(ref ct_prefix) = self.content_type_prefix {
1123 let actual_ct = request
1124 .headers
1125 .iter()
1126 .find(|h| h.name.eq_ignore_ascii_case("content-type"))
1127 .map(|h| h.value.as_ref())
1128 .unwrap_or("");
1129 if !actual_ct.starts_with(ct_prefix.as_str()) {
1130 return false;
1131 }
1132 }
1133
1134 true
1135 }
1136}
1137
1138#[derive(Clone)]
1145pub struct ConfigDrivenApp {
1146 current: Arc<RwLock<Arc<Generation>>>,
1147 reload_lock: Arc<std::sync::Mutex<()>>,
1149}
1150
1151struct Generation {
1153 dispatch: RouteDispatch,
1154 entry: Option<Arc<dyn Application + Send + Sync>>,
1157 config: ProxyConfig,
1158 source: Option<String>,
1160 health_stop: Arc<AtomicBool>,
1162}
1163
1164impl Generation {
1165 fn new(routes: Arc<Vec<CompiledRoute>>, fallback: App, config: ProxyConfig, health_stop: Arc<AtomicBool>, source: Option<String>) -> Self {
1166 let dispatch = RouteDispatch { routes, fallback };
1167 let global = &config.global_middleware;
1168 let entry = if global.is_empty() && global.early_hints.is_empty() {
1169 None
1170 } else {
1171 let mut entry = builder::apply_middleware(arc_app(dispatch.clone()), global);
1172 if !global.early_hints.is_empty() {
1175 let layer = crate::early_hints::EarlyHintsLayer::new(global.early_hints.iter().cloned());
1176 entry = arc_app(crate::middleware::WithMiddleware::new(builder::ArcApp(entry)).wrap(layer));
1177 }
1178 Some(entry)
1179 };
1180 Generation { dispatch, entry, config, source, health_stop }
1181 }
1182}
1183
1184#[derive(Clone)]
1186struct RouteDispatch {
1187 routes: Arc<Vec<CompiledRoute>>,
1188 fallback: App,
1193}
1194
1195impl Application for RouteDispatch {
1196 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1197 for route in self.routes.iter() {
1198 if route.matcher.matches(request, conn) {
1199 return route.handler.execute(request, conn);
1200 }
1201 }
1202 self.fallback.execute(request, conn)
1203 }
1204
1205 fn streams_body(&self, request: &Request, conn: &ConnectionInfo) -> bool {
1206 match self.routes.iter().find(|r| r.matcher.matches(request, conn)) {
1207 Some(route) => route.handler.streams_body(request, conn),
1208 None => self.fallback.streams_body(request, conn),
1209 }
1210 }
1211}
1212
1213impl MiddlewareConfig {
1214 pub fn is_empty(&self) -> bool {
1218 self.rate_limit.is_none() && self.cache.is_none() && self.auth.is_none()
1219 && self.rewrite_request.is_empty() && self.rewrite_response.is_empty()
1220 && self.ip_allow.is_empty() && self.ip_deny.is_empty()
1221 && self.timeout_ms.is_none() && self.max_body_size.is_none() && self.wasm.is_none()
1222 && self.authz.is_none() && self.waf.is_none()
1223 }
1224}
1225
1226#[derive(Debug, Clone, PartialEq)]
1228pub enum ReloadOutcome {
1229 Unchanged,
1231 Reloaded { routes: usize, warnings: Vec<String> },
1234}
1235
1236impl ConfigDrivenApp {
1237 #[doc(hidden)]
1238 pub fn new(routes: Vec<CompiledRoute>, config: ProxyConfig, health_stop: Arc<AtomicBool>, source: Option<String>) -> Self {
1239 use crate::core::New;
1240 let generation = Generation::new(Arc::new(routes), App::new(), config, health_stop, source);
1241 ConfigDrivenApp {
1242 current: Arc::new(RwLock::new(Arc::new(generation))),
1243 reload_lock: Arc::new(std::sync::Mutex::new(())),
1244 }
1245 }
1246
1247 fn current(&self) -> Arc<Generation> {
1248 Arc::clone(&self.current.read().unwrap_or_else(|e| e.into_inner()))
1249 }
1250
1251 pub fn with_config(self, config: ServerConfig) -> Self {
1269 {
1270 let mut slot = self.current.write().unwrap_or_else(|e| e.into_inner());
1271 let old = Arc::clone(&slot);
1272 *slot = Arc::new(Generation::new(
1273 Arc::clone(&old.dispatch.routes),
1274 App::with_config(config),
1275 old.config.clone(),
1276 Arc::clone(&old.health_stop),
1277 old.source.clone(),
1278 ));
1279 }
1280 self
1281 }
1282
1283 pub fn reload_from_str(&self, toml: &str) -> Result<ReloadOutcome, crate::config_check::ConfigReport> {
1293 let _serialized = self.reload_lock.lock().unwrap_or_else(|e| e.into_inner());
1294 let old = self.current();
1295 if old.source.as_deref() == Some(toml) {
1296 return Ok(ReloadOutcome::Unchanged);
1297 }
1298 let config = ProxyConfig::from_str(toml);
1299 let mut report = crate::config_check::ConfigReport::default();
1300 validate::check_unknown_keys(toml, &mut report);
1301 config.validate(&mut report);
1302 if !report.is_ok() {
1303 return Err(report);
1304 }
1305
1306 let mut warnings = report.warnings;
1307 let listeners = |c: &ProxyConfig| format!("{:?}{:?}{:?}", c.tcp_proxies, c.udp_proxies, c.ws_proxies);
1308 if listeners(&config) != listeners(&old.config) {
1309 warnings.push("[[tcp_proxy]] / [[udp_proxy]] / [[ws_proxy]] changes need a restart; the old listeners keep running".to_string());
1310 }
1311
1312 let health_stop = Arc::new(AtomicBool::new(false));
1313 let routes = builder::compile_routes(&config, &health_stop);
1314 let route_count = routes.len();
1315 let generation = Generation::new(Arc::new(routes), old.dispatch.fallback.clone(), config, health_stop, Some(toml.to_string()));
1316 let replaced = {
1317 let mut slot = self.current.write().unwrap_or_else(|e| e.into_inner());
1318 std::mem::replace(&mut *slot, Arc::new(generation))
1319 };
1320 replaced.health_stop.store(true, Ordering::Relaxed);
1321 Ok(ReloadOutcome::Reloaded { routes: route_count, warnings })
1322 }
1323
1324 pub fn reload_from_file(&self) {
1329 let path = config_file_path();
1330 let toml = match std::fs::read_to_string(&path) {
1331 Ok(toml) => toml,
1332 Err(e) => {
1333 eprintln!("[proxy_config] reload: cannot read {path}: {e} — keeping the running routes");
1334 crate::config_reload::report_error(format!("cannot read {path}: {e}"));
1335 return;
1336 }
1337 };
1338 match self.reload_from_str(&toml) {
1339 Ok(ReloadOutcome::Unchanged) => {}
1340 Ok(ReloadOutcome::Reloaded { routes, warnings }) => {
1341 for w in &warnings {
1342 eprintln!("[proxy_config] Warning: {w}");
1343 }
1344 println!("[proxy_config] reloaded {path}: {routes} route{}", if routes == 1 { "" } else { "s" });
1345 }
1346 Err(report) => {
1347 for e in &report.errors {
1348 eprintln!("[proxy_config] reload ERROR: {e}");
1349 crate::config_reload::report_error(e.clone());
1350 }
1351 eprintln!("[proxy_config] {path} has errors — keeping the running routes");
1352 }
1353 }
1354 }
1355}
1356
1357fn normalized(request: &Request) -> std::borrow::Cow<'_, Request> {
1360 match normalize_request_target(&request.request_uri) {
1361 std::borrow::Cow::Borrowed(_) => std::borrow::Cow::Borrowed(request),
1362 std::borrow::Cow::Owned(target) => {
1363 let mut owned = request.clone();
1364 owned.request_uri = target;
1365 std::borrow::Cow::Owned(owned)
1366 }
1367 }
1368}
1369
1370#[doc(hidden)]
1378pub fn normalize_request_target(target: &str) -> std::borrow::Cow<'_, str> {
1379 let (path, query) = match target.find('?') {
1380 Some(i) => (&target[..i], &target[i..]),
1381 None => (target, ""),
1382 };
1383 if !path.starts_with('/') || !(path.contains('%') || path.contains("//") || path.contains("/.")) {
1384 return std::borrow::Cow::Borrowed(target);
1385 }
1386 let bytes = path.as_bytes();
1387 let mut decoded = String::with_capacity(path.len());
1388 let mut i = 0;
1389 while i < bytes.len() {
1390 let hex = |b: u8| (b as char).to_digit(16);
1391 if bytes[i] == b'%' && i + 2 < bytes.len() {
1392 if let (Some(h), Some(l)) = (hex(bytes[i + 1]), hex(bytes[i + 2])) {
1393 let value = (h * 16 + l) as u8;
1394 if value.is_ascii_alphanumeric() || matches!(value, b'-' | b'.' | b'_' | b'~') {
1395 decoded.push(value as char);
1396 } else {
1397 decoded.push('%');
1398 decoded.push(bytes[i + 1].to_ascii_uppercase() as char);
1399 decoded.push(bytes[i + 2].to_ascii_uppercase() as char);
1400 }
1401 i += 3;
1402 continue;
1403 }
1404 }
1405 if bytes[i] == b'%' {
1409 decoded.push_str("%25");
1410 i += 1;
1411 continue;
1412 }
1413 let ch = path[i..].chars().next().unwrap_or('\u{FFFD}');
1415 decoded.push(ch);
1416 i += ch.len_utf8().max(1);
1417 }
1418 let mut segments: Vec<&str> = Vec::new();
1419 let parts: Vec<&str> = decoded.split('/').collect();
1420 let last = parts.len().saturating_sub(1);
1421 let mut trailing_slash = false;
1422 for (n, segment) in parts.iter().enumerate() {
1423 match *segment {
1424 "" => trailing_slash = n == last,
1425 "." => trailing_slash = n == last,
1426 ".." => {
1427 segments.pop();
1428 trailing_slash = n == last;
1429 }
1430 s => {
1431 segments.push(s);
1432 trailing_slash = false;
1433 }
1434 }
1435 }
1436 let mut out = String::with_capacity(decoded.len() + query.len());
1437 out.push('/');
1438 out.push_str(&segments.join("/"));
1439 if trailing_slash && !segments.is_empty() {
1440 out.push('/');
1441 }
1442 out.push_str(query);
1443 if out == target {
1444 std::borrow::Cow::Borrowed(target)
1445 } else {
1446 std::borrow::Cow::Owned(out)
1447 }
1448}
1449
1450impl Application for ConfigDrivenApp {
1451 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1452 let request = normalized(request);
1453 let generation = self.current();
1454 match &generation.entry {
1455 Some(entry) => entry.execute(&request, conn),
1456 None => generation.dispatch.execute(&request, conn),
1457 }
1458 }
1459
1460 fn streams_body(&self, request: &Request, conn: &ConnectionInfo) -> bool {
1463 let request = normalized(request);
1464 let generation = self.current();
1465 match &generation.entry {
1466 Some(entry) => entry.streams_body(&request, conn),
1467 None => generation.dispatch.streams_body(&request, conn),
1468 }
1469 }
1470
1471 fn runs_inline(&self, request: &Request, conn: &ConnectionInfo) -> bool {
1475 let request = normalized(request);
1476 let generation = self.current();
1477 generation.entry.is_none()
1478 && !generation.dispatch.routes.iter().any(|r| r.matcher.matches(&request, conn))
1479 && App::is_inline_request(&request)
1480 }
1481
1482 #[cfg(feature = "http2")]
1486 fn execute_async<'a>(&'a self, request: &'a Request, conn: &'a ConnectionInfo) -> Option<crate::application::AsyncResponse<'a>> {
1487 let generation = self.current();
1488 if generation.entry.is_some() || crate::proxy::upgrade::is_upgrade(request) {
1490 return None;
1491 }
1492 let request = normalized(request);
1493 let route = generation.dispatch.routes.iter().find(|r| r.matcher.matches(&request, conn))?;
1494 let proxy = Arc::clone(route.async_proxy.as_ref()?);
1495 Some(Box::pin(async move { proxy.execute_async(&request, conn).await }))
1496 }
1497}
1498
1499#[cfg(feature = "http2")]
1505#[doc(hidden)]
1506pub struct H2Route {
1507 live: Arc<RwLock<Vec<String>>>,
1508 tls: bool,
1509 trust: Option<Arc<crate::proxy::UpstreamTls>>,
1510 connect_timeout: std::time::Duration,
1511 read_timeout: std::time::Duration,
1512 counter: std::sync::atomic::AtomicUsize,
1513}
1514
1515#[cfg(feature = "http2")]
1516impl H2Route {
1517 #[doc(hidden)]
1518 pub fn new(
1519 live: Arc<RwLock<Vec<String>>>,
1520 tls: bool,
1521 trust: Option<Arc<crate::proxy::UpstreamTls>>,
1522 connect_timeout_ms: u64,
1523 read_timeout_ms: u64,
1524 ) -> H2Route {
1525 H2Route {
1526 live,
1527 tls,
1528 trust,
1529 connect_timeout: std::time::Duration::from_millis(connect_timeout_ms),
1530 read_timeout: std::time::Duration::from_millis(read_timeout_ms),
1531 counter: std::sync::atomic::AtomicUsize::new(0),
1532 }
1533 }
1534}
1535
1536#[cfg(feature = "http2")]
1537impl Application for H2Route {
1538 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1539 use crate::middleware::Middleware;
1540 crate::proxy::GrpcWebLayer::new().handle(request, conn, &H2Forward(self))
1541 }
1542
1543 fn streams_body(&self, _request: &Request, _conn: &ConnectionInfo) -> bool {
1544 true
1545 }
1546}
1547
1548#[cfg(feature = "http2")]
1550struct H2Forward<'a>(&'a H2Route);
1551
1552#[cfg(feature = "http2")]
1553impl Application for H2Forward<'_> {
1554 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1555 let route = self.0;
1556 let backends = route.live.read().unwrap_or_else(|e| e.into_inner()).clone();
1557 if backends.is_empty() {
1558 let mut response = crate::proxy::bad_gateway();
1559 crate::proxy::status::failed(&mut response, crate::proxy::status::ProxyError::DestinationUnavailable, None, None);
1560 return Ok(response);
1561 }
1562 let attempts = if matches!(request.body, crate::request::Body::Stream(_)) { 1 } else { backends.len() };
1564 let start = route.counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1565 let mut failure: Option<(String, String)> = None;
1566 for attempt in 0..attempts {
1567 let url = &backends[(start + attempt) % backends.len()];
1568 let result = crate::proxy::forward_h2(
1569 request, crate::proxy::Peer::of(conn), url, route.tls, route.trust.as_deref(), route.connect_timeout, route.read_timeout,
1570 );
1571 match result {
1572 Ok(mut response) => {
1573 crate::proxy::status::received(&mut response, Some(url));
1574 return Ok(response);
1575 }
1576 Err(e) => {
1577 eprintln!("grpc route: {e}");
1578 failure = Some((e, url.clone()));
1579 }
1580 }
1581 }
1582 let mut response = crate::proxy::bad_gateway();
1583 if let Some((e, url)) = &failure {
1584 crate::proxy::status::failed_with(&mut response, e, Some(url));
1585 }
1586 Ok(response)
1587 }
1588}
1589
1590#[derive(Clone, Copy)]
1595#[doc(hidden)]
1596pub struct NullApp;
1597
1598impl Application for NullApp {
1599 fn execute(&self, _request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
1600 let mut r = Response::new();
1601 r.status_code = *STATUS_CODE_REASON_PHRASE.n404_not_found.status_code;
1602 r.reason_phrase = STATUS_CODE_REASON_PHRASE.n404_not_found.reason_phrase.to_string();
1603 Ok(r)
1604 }
1605}
1606
1607use std::collections::HashMap;
1610use std::collections::hash_map::DefaultHasher;
1611use std::hash::{Hash, Hasher};
1612use std::sync::atomic::AtomicUsize;
1613use std::time::{Duration, SystemTime, UNIX_EPOCH};
1614
1615#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1619#[doc(hidden)]
1620pub enum LoadBalanceStrategy {
1621 RoundRobin,
1622 Random,
1623 IpHash,
1624 LeastConnections,
1625}
1626
1627impl LoadBalanceStrategy {
1628 fn parse(s: &str) -> Self {
1629 match s {
1630 "random" => LoadBalanceStrategy::Random,
1631 "ip_hash" => LoadBalanceStrategy::IpHash,
1632 "least_connections" => LoadBalanceStrategy::LeastConnections,
1633 _ => LoadBalanceStrategy::RoundRobin,
1634 }
1635 }
1636}
1637
1638#[derive(Clone)]
1646#[doc(hidden)]
1647pub struct DynamicProxy {
1648 live: Arc<RwLock<Vec<String>>>,
1649 counter: Arc<AtomicUsize>,
1650 connect_timeout: Duration,
1651 read_timeout: Duration,
1652 strip_prefix: Option<Arc<String>>,
1653 add_prefix: Option<Arc<String>>,
1654 tls: bool,
1655 strategy: LoadBalanceStrategy,
1656 connections: Arc<RwLock<HashMap<String, Arc<AtomicUsize>>>>,
1657 trust: Option<Arc<crate::proxy::UpstreamTls>>,
1659 upstream: Arc<String>,
1661 policy: Option<Arc<upstream::UpstreamPolicy>>,
1663 mirror: Option<crate::proxy::mirror::Mirror>,
1666 early_hints: Option<Arc<Vec<String>>>,
1668 #[cfg(feature = "http3")]
1671 h3: Option<Arc<std::sync::Mutex<HashMap<String, std::time::Instant>>>>,
1672 upstream_host: UpstreamHost,
1674}
1675
1676impl DynamicProxy {
1677 #[doc(hidden)]
1679 pub fn with_upstream_host(mut self, host: UpstreamHost) -> Self {
1680 self.upstream_host = host;
1681 self
1682 }
1683
1684 fn host_override<'a>(&'a self, request: &'a Request) -> Option<&'a str> {
1686 match &self.upstream_host {
1687 UpstreamHost::Backend => None,
1688 UpstreamHost::Client => request.header(Header::_HOST).map(|h| h.value.as_ref()).filter(|v| !v.is_empty()),
1689 UpstreamHost::Fixed(host) => Some(host),
1690 }
1691 }
1692
1693 #[doc(hidden)]
1695 pub fn with_trust(mut self, trust: Option<Arc<crate::proxy::UpstreamTls>>) -> Self {
1696 self.trust = trust;
1697 self
1698 }
1699
1700 #[doc(hidden)]
1705 pub fn with_early_hints(mut self, links: &[String]) -> Self {
1706 self.early_hints = (!links.is_empty()).then(|| Arc::new(links.to_vec()));
1707 self
1708 }
1709
1710 #[cfg_attr(not(feature = "http3"), allow(unused_mut))]
1714 #[doc(hidden)]
1715 pub fn with_h3(mut self, enabled: bool) -> Self {
1716 #[cfg(feature = "http3")]
1717 {
1718 self.h3 = (enabled && self.tls).then(Default::default);
1719 }
1720 #[cfg(not(feature = "http3"))]
1721 let _ = enabled;
1722 self
1723 }
1724
1725 #[cfg(feature = "http3")]
1728 fn h3_target(&self, backend: &str) -> Option<Arc<rustls::ClientConfig>> {
1729 let tcp_until = self.h3.as_ref()?;
1730 let mut tcp_until = tcp_until.lock().unwrap_or_else(|e| e.into_inner());
1731 if let Some(until) = tcp_until.get(backend) {
1732 if std::time::Instant::now() < *until {
1733 return None;
1734 }
1735 tcp_until.remove(backend);
1736 }
1737 Some(match &self.trust {
1738 Some(trust) => trust.client_config(),
1739 None => crate::proxy::tls_client_config(),
1740 })
1741 }
1742
1743 #[cfg(feature = "http3")]
1745 fn fall_back_to_tcp(&self, backend: &str, error: &str) {
1746 if let Some(tcp_until) = &self.h3 {
1747 tcp_until.lock().unwrap_or_else(|e| e.into_inner()).insert(backend.to_string(), std::time::Instant::now() + H3_FALLBACK);
1748 }
1749 upstream::observe_h3_fallback(&self.upstream);
1750 eprintln!("upstream {}: HTTP/3 to {backend} failed ({error}); using TCP for {} s", self.upstream, H3_FALLBACK.as_secs());
1751 }
1752
1753 #[doc(hidden)]
1755 pub fn with_mirror(mut self, mirror: Option<crate::proxy::mirror::Mirror>) -> Self {
1756 self.mirror = mirror;
1757 self
1758 }
1759
1760 #[doc(hidden)]
1761 pub fn with_policy(mut self, upstream: &str, policy: Option<Arc<upstream::UpstreamPolicy>>) -> Self {
1762 self.upstream = Arc::new(upstream.to_string());
1763 self.policy = policy;
1764 self
1765 }
1766
1767 #[doc(hidden)]
1768 pub fn new(
1769 live: Arc<RwLock<Vec<String>>>,
1770 connect_timeout_ms: u64,
1771 read_timeout_ms: u64,
1772 strip_prefix: Option<String>,
1773 add_prefix: Option<String>,
1774 tls: bool,
1775 strategy: String,
1776 ) -> Self {
1777 DynamicProxy {
1778 live,
1779 counter: Arc::new(AtomicUsize::new(0)),
1780 connect_timeout: Duration::from_millis(connect_timeout_ms),
1781 read_timeout: Duration::from_millis(read_timeout_ms),
1782 strip_prefix: strip_prefix.map(Arc::new),
1783 add_prefix: add_prefix.map(Arc::new),
1784 tls,
1785 strategy: LoadBalanceStrategy::parse(&strategy),
1786 connections: Arc::new(RwLock::new(HashMap::new())),
1787 trust: None,
1788 upstream: Arc::new(String::new()),
1789 policy: None,
1790 mirror: None,
1791 early_hints: None,
1792 #[cfg(feature = "http3")]
1793 h3: None,
1794 upstream_host: UpstreamHost::Backend,
1795 }
1796 }
1797
1798 fn pick(&self, candidates: &[String], client_ip: &str) -> String {
1800 let policy = self.policy.as_deref().filter(|p| p.weighted());
1801 let idx = match self.strategy {
1802 LoadBalanceStrategy::RoundRobin => match policy {
1803 Some(policy) => return policy.weighted_round_robin(candidates).unwrap_or_else(|| candidates[0].clone()),
1804 None => self.counter.fetch_add(1, Ordering::Relaxed) % candidates.len(),
1805 },
1806 LoadBalanceStrategy::Random => {
1807 let nanos = SystemTime::now()
1808 .duration_since(UNIX_EPOCH)
1809 .map(|d| d.subsec_nanos())
1810 .unwrap_or(0) as usize;
1811 let salt = self.counter.fetch_add(1, Ordering::Relaxed);
1812 let roll = nanos.wrapping_add(salt.wrapping_mul(2_654_435_761));
1813 match policy {
1814 Some(policy) => policy.weighted_index(candidates, roll),
1815 None => roll % candidates.len(),
1816 }
1817 }
1818 LoadBalanceStrategy::IpHash => {
1819 let mut hasher = DefaultHasher::new();
1820 client_ip.hash(&mut hasher);
1821 (hasher.finish() as usize) % candidates.len()
1822 }
1823 LoadBalanceStrategy::LeastConnections => {
1824 let connections = self.connections.read().unwrap();
1825 let load = |backend: &String| connections.get(backend).map(|c| c.load(Ordering::Relaxed)).unwrap_or(0) as u64;
1826 let weight = |backend: &String| self.policy.as_deref().map_or(1, |p| p.weight(backend)) as u64;
1827 candidates
1829 .iter()
1830 .enumerate()
1831 .min_by(|(_, a), (_, b)| (load(a) * weight(b)).cmp(&(load(b) * weight(a))))
1832 .map(|(i, _)| i)
1833 .unwrap_or(0)
1834 }
1835 };
1836 candidates[idx].clone()
1837 }
1838
1839 fn next_backend(&self, request: &Request, client_ip: &str, tried: &[String]) -> Result<String, bool> {
1844 let mut candidates: Vec<String> = self.live.read().unwrap().iter().filter(|b| !tried.contains(b)).cloned().collect();
1845 if !candidates.is_empty() {
1846 candidates.retain(|b| !upstream::is_drained(&self.upstream, b));
1847 if candidates.is_empty() {
1848 return Err(true);
1850 }
1851 }
1852 let Some(policy) = self.policy.as_deref() else {
1853 return if candidates.is_empty() { Err(false) } else { Ok(self.pick(&candidates, client_ip)) };
1854 };
1855 if let Some(sticky) = &policy.sticky {
1856 if let Some(id) = upstream::cookie_value(request, &sticky.cookie) {
1857 if let Some(backend) = candidates.iter().find(|b| upstream::backend_id(b) == id).cloned() {
1858 if policy.admits(&backend) {
1859 return Ok(backend);
1860 }
1861 candidates.retain(|b| *b != backend);
1862 }
1863 }
1864 }
1865 let ejected = policy.any_open(&candidates);
1866 while !candidates.is_empty() {
1867 let backend = self.pick(&candidates, client_ip);
1868 if policy.admits(&backend) {
1869 return Ok(backend);
1870 }
1871 candidates.retain(|b| *b != backend);
1872 }
1873 Err(ejected)
1874 }
1875
1876 fn connection_counter(&self, backend: &str) -> Arc<AtomicUsize> {
1879 if let Some(counter) = self.connections.read().unwrap().get(backend) {
1880 return Arc::clone(counter);
1881 }
1882 let mut connections = self.connections.write().unwrap();
1883 Arc::clone(
1884 connections
1885 .entry(backend.to_string())
1886 .or_insert_with(|| Arc::new(AtomicUsize::new(0))),
1887 )
1888 }
1889}
1890
1891struct ConnectionGuard {
1894 counter: Arc<AtomicUsize>,
1895}
1896
1897impl Drop for ConnectionGuard {
1898 fn drop(&mut self) {
1899 self.counter.fetch_sub(1, Ordering::Relaxed);
1900 }
1901}
1902
1903struct Prepared<'a> {
1906 backend: String,
1907 host: String,
1908 port: u16,
1909 request: std::borrow::Cow<'a, Request>,
1910 _guard: Option<ConnectionGuard>,
1911}
1912
1913enum Attempt {
1915 Done(Response),
1917 Failed(Option<Response>),
1920}
1921
1922impl DynamicProxy {
1923 fn forward_upgrade(&self, request: &Request, conn: &ConnectionInfo) -> Response {
1926 let prepared = match self.prepare(request, conn, &[]) {
1927 Ok(p) => p,
1928 Err(response) => return response,
1929 };
1930 if self.tls || prepared.backend.starts_with("https://") || prepared.backend.starts_with("wss://") {
1931 eprintln!("[proxy] upgrade to {}: TLS backends can't take upgraded connections yet", prepared.backend);
1932 return bad_gateway();
1933 }
1934 let default_host = if prepared.port == 80 { prepared.host.clone() } else { format!("{}:{}", prepared.host, prepared.port) };
1935 let host_header = self.host_override(request).map(str::to_string).unwrap_or(default_host);
1936 match crate::proxy::upgrade::forward(&prepared.request, crate::proxy::Peer::of(conn), &prepared.host, prepared.port, &host_header, self.read_timeout) {
1937 Ok(response) => response,
1938 Err(e) => {
1939 eprintln!("[proxy] upgrade to {}: {e}", prepared.backend);
1940 let mut response = bad_gateway();
1941 crate::proxy::status::failed(&mut response, crate::proxy::status::ProxyError::ConnectionRefused, Some(&prepared.backend), None);
1942 response
1943 }
1944 }
1945 }
1946
1947 fn prepare<'a>(&self, request: &'a Request, conn: &ConnectionInfo, tried: &[String]) -> Result<Prepared<'a>, Response> {
1952 let backend = self.next_backend(request, &conn.client.ip, tried).map_err(|ejected| {
1953 let mut response = if ejected { service_unavailable() } else { bad_gateway() };
1954 crate::proxy::status::failed(&mut response, crate::proxy::status::ProxyError::DestinationUnavailable, None, None);
1955 response
1956 })?;
1957
1958 let guard = if self.strategy == LoadBalanceStrategy::LeastConnections {
1959 let counter = self.connection_counter(&backend);
1960 counter.fetch_add(1, Ordering::Relaxed);
1961 Some(ConnectionGuard { counter })
1962 } else {
1963 None
1964 };
1965
1966 let (host, port, _) = crate::proxy_config::health::parse_backend_url(&backend).ok_or_else(|| {
1967 let mut response = bad_gateway();
1968 crate::proxy::status::failed(&mut response, crate::proxy::status::ProxyError::ProxyConfigurationError, Some(&backend), None);
1969 response
1970 })?;
1971
1972 let request = if self.strip_prefix.is_some() || self.add_prefix.is_some() {
1974 let mut req_clone = request.clone();
1975 if let Some(ref sp) = self.strip_prefix {
1976 if let Some(stripped) = req_clone.request_uri.strip_prefix(sp.as_str()) {
1977 req_clone.request_uri = if stripped.is_empty() || !stripped.starts_with('/') {
1978 format!("/{}", stripped)
1979 } else {
1980 stripped.to_string()
1981 };
1982 }
1983 }
1984 if let Some(ref ap) = self.add_prefix {
1985 req_clone.request_uri = format!("{}{}", ap, req_clone.request_uri);
1986 }
1987 std::borrow::Cow::Owned(req_clone)
1988 } else {
1989 std::borrow::Cow::Borrowed(request)
1990 };
1991
1992 Ok(Prepared { backend, host, port, request, _guard: guard })
1993 }
1994
1995 fn settle(&self, backend: &str, result: Result<Response, String>, started: std::time::Instant, failure: &mut Option<(String, String)>) -> Attempt {
1997 let result = match result {
1998 Ok(mut response) => {
1999 crate::proxy::status::received(&mut response, Some(backend));
2000 Ok(response)
2001 }
2002 Err(e) => {
2003 *failure = Some((e, backend.to_string()));
2004 Err(())
2005 }
2006 };
2007 let status = result.as_ref().ok().map(|r| r.status_code);
2008 upstream::observe(&self.upstream, backend, status, started);
2009 let failed = status.is_none_or(|s| matches!(s, 502..=504));
2010 if let Some(policy) = &self.policy {
2011 policy.record(backend, failed);
2012 }
2013 match result {
2014 Ok(response) if self.policy.as_deref().is_some_and(|p| p.retries_status(response.status_code)) => Attempt::Failed(Some(response)),
2015 Ok(response) => Attempt::Done(response),
2016 Err(()) => Attempt::Failed(None),
2017 }
2018 }
2019
2020 fn attempts(&self, request: &Request) -> u32 {
2022 match self.policy.as_deref().and_then(|p| p.retry.as_ref()) {
2023 Some(retry) if upstream::replayable(request) => 1 + retry.attempts,
2024 _ => 1,
2025 }
2026 }
2027
2028 fn stick(&self, request: &Request, backend: &str, response: &mut Response) {
2030 let Some(sticky) = self.policy.as_deref().and_then(|p| p.sticky.as_ref()) else { return };
2031 let id = upstream::backend_id(backend);
2032 if upstream::cookie_value(request, &sticky.cookie).as_deref() == Some(id.as_str()) {
2033 return;
2034 }
2035 let max_age = sticky.max_age_secs.map(|s| format!("; Max-Age={s}")).unwrap_or_default();
2036 response.headers.push(crate::header::Header {
2037 name: "Set-Cookie".into(),
2038 value: format!("{}={id}; Path=/; HttpOnly; SameSite=Lax{max_age}", sticky.cookie).into(),
2039 });
2040 }
2041
2042 #[cfg(feature = "http2")]
2045 #[doc(hidden)]
2046 pub async fn execute_async(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
2047 if let Some(mirror) = &self.mirror {
2048 mirror.send(request, conn);
2049 }
2050 if let Some(links) = &self.early_hints {
2051 crate::early_hints::send_from(crate::early_hints::Source::Configured, links);
2052 }
2053 let mut tried: Vec<String> = Vec::new();
2054 let mut last: Option<Response> = None;
2055 let mut failure: Option<(String, String)> = None;
2056 for attempt in 0..self.attempts(request) {
2057 let prepared = match self.prepare(request, conn, &tried) {
2058 Ok(p) => p,
2059 Err(response) => return Ok(last.unwrap_or(response)),
2060 };
2061 if attempt > 0 {
2062 upstream::observe_retry(&self.upstream);
2063 }
2064 let tls = self.tls.then(|| match &self.trust {
2065 Some(trust) => trust.async_tls(),
2066 None => crate::proxy::async_upstream::AsyncTls {
2067 config: crate::proxy::tls_client_config(),
2068 pool: crate::proxy::async_upstream::default_tls_pool(),
2069 },
2070 });
2071 let started = std::time::Instant::now();
2072 #[cfg(feature = "http3")]
2073 if let Some(config) = self.h3_target(&prepared.backend) {
2074 let target = crate::proxy::h3_upstream::Target {
2075 host: &prepared.host,
2076 port: prepared.port,
2077 authority: self.host_override(request),
2078 tls: &config,
2079 connect_timeout: self.connect_timeout,
2080 read_timeout: self.read_timeout,
2081 };
2082 match crate::proxy::h3_upstream::forward_async(&prepared.request, crate::proxy::Peer::of(conn), &target).await {
2083 Err(crate::proxy::h3_upstream::H3Error::Connect(e)) => self.fall_back_to_tcp(&prepared.backend, &e),
2084 result => {
2085 match self.settle(&prepared.backend, result.map_err(|e| e.to_string()), started, &mut failure) {
2086 Attempt::Done(mut response) => {
2087 self.stick(request, &prepared.backend, &mut response);
2088 return Ok(response);
2089 }
2090 Attempt::Failed(response) => last = response,
2091 }
2092 tried.push(prepared.backend.clone());
2093 continue;
2094 }
2095 }
2096 }
2097 let result = crate::proxy::async_upstream::forward_as(
2098 &prepared.request,
2099 crate::proxy::Peer::of(conn),
2100 &prepared.host,
2101 prepared.port,
2102 self.host_override(request).unwrap_or(&prepared.host),
2103 tls,
2104 self.connect_timeout,
2105 self.read_timeout,
2106 )
2107 .await;
2108 match self.settle(&prepared.backend, result, started, &mut failure) {
2109 Attempt::Done(mut response) => {
2110 self.stick(request, &prepared.backend, &mut response);
2111 return Ok(response);
2112 }
2113 Attempt::Failed(response) => last = response,
2114 }
2115 tried.push(prepared.backend.clone());
2116 }
2117 Ok(last.unwrap_or_else(|| {
2118 let mut response = bad_gateway();
2119 match &failure {
2120 Some((e, backend)) => crate::proxy::status::failed_with(&mut response, e, Some(backend)),
2121 None => crate::proxy::status::failed(&mut response, crate::proxy::status::ProxyError::ProxyInternalError, None, None),
2122 }
2123 response
2124 }))
2125 }
2126
2127 #[doc(hidden)]
2129 pub fn supports_async(&self) -> bool {
2130 cfg!(feature = "http2")
2131 }
2132}
2133
2134impl Application for DynamicProxy {
2135 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
2136 if crate::proxy::upgrade::is_upgrade(request) {
2139 return Ok(self.forward_upgrade(request, conn));
2140 }
2141 if let Some(mirror) = &self.mirror {
2142 mirror.send(request, conn);
2143 }
2144 if let Some(links) = &self.early_hints {
2145 crate::early_hints::send_from(crate::early_hints::Source::Configured, links);
2146 }
2147 let mut tried: Vec<String> = Vec::new();
2148 let mut last: Option<Response> = None;
2149 let mut failure: Option<(String, String)> = None;
2150 for attempt in 0..self.attempts(request) {
2151 let prepared = match self.prepare(request, conn, &tried) {
2152 Ok(p) => p,
2153 Err(response) => return Ok(last.unwrap_or(response)),
2154 };
2155 if attempt > 0 {
2156 upstream::observe_retry(&self.upstream);
2157 }
2158 let started = std::time::Instant::now();
2161 #[cfg(feature = "http3")]
2162 if let Some(config) = self.h3_target(&prepared.backend) {
2163 let target = crate::proxy::h3_upstream::Target {
2164 host: &prepared.host,
2165 port: prepared.port,
2166 authority: self.host_override(request),
2167 tls: &config,
2168 connect_timeout: self.connect_timeout,
2169 read_timeout: self.read_timeout,
2170 };
2171 match crate::proxy::h3_upstream::forward(&prepared.request, crate::proxy::Peer::of(conn), &target) {
2172 Err(crate::proxy::h3_upstream::H3Error::Connect(e)) => self.fall_back_to_tcp(&prepared.backend, &e),
2173 result => {
2174 match self.settle(&prepared.backend, result.map_err(|e| e.to_string()), started, &mut failure) {
2175 Attempt::Done(mut response) => {
2176 self.stick(request, &prepared.backend, &mut response);
2177 return Ok(response);
2178 }
2179 Attempt::Failed(response) => last = response,
2180 }
2181 tried.push(prepared.backend.clone());
2182 continue;
2183 }
2184 }
2185 }
2186 let result = crate::proxy::forward_upstream_as(
2187 crate::proxy::shared_pool(),
2188 &prepared.request,
2189 crate::proxy::Peer::of(conn),
2190 &prepared.host,
2191 prepared.port,
2192 self.host_override(request).unwrap_or(&prepared.host),
2193 self.tls,
2194 self.trust.as_deref(),
2195 self.connect_timeout,
2196 self.read_timeout,
2197 );
2198 match self.settle(&prepared.backend, result, started, &mut failure) {
2199 Attempt::Done(mut response) => {
2200 self.stick(request, &prepared.backend, &mut response);
2201 return Ok(response);
2202 }
2203 Attempt::Failed(response) => last = response,
2204 }
2205 tried.push(prepared.backend.clone());
2206 }
2207 Ok(last.unwrap_or_else(|| {
2208 let mut response = bad_gateway();
2209 match &failure {
2210 Some((e, backend)) => crate::proxy::status::failed_with(&mut response, e, Some(backend)),
2211 None => crate::proxy::status::failed(&mut response, crate::proxy::status::ProxyError::ProxyInternalError, None, None),
2212 }
2213 response
2214 }))
2215 }
2216}
2217
2218#[cfg(feature = "http3")]
2220#[doc(hidden)]
2221pub const H3_FALLBACK: Duration = Duration::from_secs(300);
2222
2223fn service_unavailable() -> Response {
2224 let mut r = crate::error::problem(STATUS_CODE_REASON_PHRASE.n503_service_unavailable, "every backend is out of rotation");
2225 r.headers.push(crate::header::Header { name: "Retry-After".into(), value: "5".into() });
2226 r
2227}
2228
2229fn bad_gateway() -> Response {
2230 crate::error::problem(STATUS_CODE_REASON_PHRASE.n502_bad_gateway, "")
2231}
2232
2233#[derive(Clone)]
2239#[doc(hidden)]
2240pub struct RedirectAdapter {
2241 location_template: Arc<String>,
2242 status: i16,
2243 reason: Arc<String>,
2244}
2245
2246impl RedirectAdapter {
2247 #[doc(hidden)]
2248 pub fn new(location: String, status: u16) -> Self {
2249 let (code, reason) = redirect_status(status);
2250 RedirectAdapter {
2251 location_template: Arc::new(location),
2252 status: code,
2253 reason: Arc::new(reason),
2254 }
2255 }
2256}
2257
2258fn redirect_status(code: u16) -> (i16, String) {
2259 let phrase = match code {
2260 301 => STATUS_CODE_REASON_PHRASE.n301_moved_permanently.reason_phrase,
2261 302 => STATUS_CODE_REASON_PHRASE.n302_found.reason_phrase,
2262 307 => STATUS_CODE_REASON_PHRASE.n307_temporary_redirect.reason_phrase,
2263 308 => STATUS_CODE_REASON_PHRASE.n308_permanent_redirect.reason_phrase,
2264 _ => "Redirect",
2265 };
2266 (code as i16, phrase.to_string())
2267}
2268
2269impl Application for RedirectAdapter {
2270 fn execute(&self, request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
2271 let location = self
2272 .location_template
2273 .replace("$path", &request.request_uri);
2274 use crate::header::Header;
2275 let mut r = Response::new();
2276 r.status_code = self.status;
2277 r.reason_phrase = self.reason.as_ref().clone();
2278 r.headers.push(Header { name: "Location".into(), value: location.into() });
2279 Ok(r)
2280 }
2281}
2282
2283#[derive(Clone)]
2287#[doc(hidden)]
2288pub struct RespondAdapter {
2289 status: i16,
2290 reason: Arc<String>,
2291 body: Arc<Vec<u8>>,
2292 content_type: Arc<String>,
2293}
2294
2295impl RespondAdapter {
2296 #[doc(hidden)]
2297 pub fn new(status: u16, body: String, content_type: String) -> Self {
2298 use crate::response::STATUS_CODE_REASON_PHRASE;
2299 let reason = match status {
2300 200 => STATUS_CODE_REASON_PHRASE.n200_ok.reason_phrase.to_string(),
2301 201 => STATUS_CODE_REASON_PHRASE.n201_created.reason_phrase.to_string(),
2302 204 => STATUS_CODE_REASON_PHRASE.n204_no_content.reason_phrase.to_string(),
2303 400 => STATUS_CODE_REASON_PHRASE.n400_bad_request.reason_phrase.to_string(),
2304 401 => STATUS_CODE_REASON_PHRASE.n401_unauthorized.reason_phrase.to_string(),
2305 403 => STATUS_CODE_REASON_PHRASE.n403_forbidden.reason_phrase.to_string(),
2306 404 => STATUS_CODE_REASON_PHRASE.n404_not_found.reason_phrase.to_string(),
2307 500 => STATUS_CODE_REASON_PHRASE.n500_internal_server_error.reason_phrase.to_string(),
2308 _ => "OK".to_string(),
2309 };
2310 RespondAdapter {
2311 status: status as i16,
2312 reason: Arc::new(reason),
2313 body: Arc::new(body.into_bytes()),
2314 content_type: Arc::new(content_type),
2315 }
2316 }
2317}
2318
2319impl Application for RespondAdapter {
2320 fn execute(&self, _request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
2321 use crate::range::Range;
2322 let mut r = Response::new();
2323 r.status_code = self.status;
2324 r.reason_phrase = self.reason.as_ref().clone();
2325 if !self.body.is_empty() {
2326 r.content_range_list = vec![Range::get_content_range(
2327 self.body.as_ref().clone(),
2328 self.content_type.as_ref().clone(),
2329 )];
2330 }
2331 Ok(r)
2332 }
2333}
2334
2335#[derive(Clone)]
2344#[doc(hidden)]
2345pub struct StaticAdapter {
2346 root: Arc<std::path::PathBuf>,
2347 index: Arc<Vec<String>>,
2348}
2349
2350impl StaticAdapter {
2351 #[doc(hidden)]
2352 pub fn new(root: String, index: Vec<String>) -> Self {
2353 let index = if index.is_empty() { vec!["index.html".to_string()] } else { index };
2354 StaticAdapter {
2355 root: Arc::new(std::path::PathBuf::from(root)),
2356 index: Arc::new(index),
2357 }
2358 }
2359
2360 fn resolve(&self, request_uri: &str) -> Option<std::path::PathBuf> {
2364 let raw_path = request_uri.split('?').next().unwrap_or(request_uri);
2365 let decoded = crate::url::URL::percent_decode(raw_path);
2366
2367 if decoded.split('/').any(|segment| segment == "..") {
2368 return None;
2369 }
2370
2371 let relative = decoded.trim_start_matches('/');
2372 Some(self.root.join(relative))
2373 }
2374}
2375
2376impl Application for StaticAdapter {
2377 fn execute(&self, request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
2378 let mut response = Response::new();
2379
2380 let not_found = |mut response: Response| {
2381 response.status_code = *STATUS_CODE_REASON_PHRASE.n404_not_found.status_code;
2382 response.reason_phrase = STATUS_CODE_REASON_PHRASE.n404_not_found.reason_phrase.to_string();
2383 response
2384 };
2385
2386 let candidate = match self.resolve(&request.request_uri) {
2387 Some(p) => p,
2388 None => {
2389 response.status_code = *STATUS_CODE_REASON_PHRASE.n403_forbidden.status_code;
2390 response.reason_phrase = STATUS_CODE_REASON_PHRASE.n403_forbidden.reason_phrase.to_string();
2391 return Ok(response);
2392 }
2393 };
2394
2395 let mut file_path = candidate;
2396 if file_path.is_dir() {
2397 let indexed = self
2398 .index
2399 .iter()
2400 .map(|name| file_path.join(name))
2401 .find(|p| p.is_file());
2402
2403 file_path = match indexed {
2404 Some(p) => p,
2405 None => return Ok(not_found(response)),
2406 };
2407 }
2408
2409 if !file_path.is_file() {
2410 return Ok(not_found(response));
2411 }
2412
2413 if let (Ok(root_canon), Ok(file_canon)) =
2416 (self.root.canonicalize(), file_path.canonicalize())
2417 {
2418 if !file_canon.starts_with(&root_canon) {
2419 return Ok(not_found(response));
2420 }
2421 }
2422
2423 let path_str = match file_path.to_str() {
2424 Some(s) => s,
2425 None => return Ok(not_found(response)),
2426 };
2427
2428 match crate::range::Range::get_content_range_of_a_file(path_str) {
2429 Ok(content_range) => {
2430 response.status_code = *STATUS_CODE_REASON_PHRASE.n200_ok.status_code;
2431 response.reason_phrase = STATUS_CODE_REASON_PHRASE.n200_ok.reason_phrase.to_string();
2432 response.content_range_list = vec![content_range];
2433 Ok(response)
2434 }
2435 Err(_) => Ok(not_found(response)),
2436 }
2437 }
2438}
2439
2440#[doc(hidden)]
2444pub struct PerRouteRateLimit(pub(crate) Arc<crate::rate_limit::RateLimiter>);
2445
2446impl crate::middleware::Middleware for PerRouteRateLimit {
2447 fn handle(
2448 &self,
2449 request: &Request,
2450 conn: &ConnectionInfo,
2451 next: &dyn Application,
2452 ) -> Result<Response, String> {
2453 use crate::error::{AppError, IntoResponse};
2454 if self.0.check(&conn.client.ip) {
2455 next.execute(request, conn)
2456 } else {
2457 let mut response = AppError::TooManyRequests.into_response();
2458 let retry_after = self.0.retry_after(&conn.client.ip);
2459 response.headers.push(Header {
2460 name: Header::_RETRY_AFTER.into(),
2461 value: retry_after.to_string().into(),
2462 });
2463 Ok(response)
2464 }
2465 }
2466}
2467
2468#[doc(hidden)]
2483pub struct PerRouteMaxBodySize(pub(crate) u64);
2484
2485impl crate::middleware::Middleware for PerRouteMaxBodySize {
2486 fn handle(
2487 &self,
2488 request: &Request,
2489 conn: &ConnectionInfo,
2490 next: &dyn Application,
2491 ) -> Result<Response, String> {
2492 use crate::error::{AppError, IntoResponse};
2493 if request.body.len() as u64 > self.0 {
2494 Ok(AppError::PayloadTooLarge(format!(
2495 "request body of {} bytes exceeds this route's {}-byte limit",
2496 request.body.len(),
2497 self.0,
2498 ))
2499 .into_response())
2500 } else {
2501 next.execute(request, conn)
2502 }
2503 }
2504}
2505
2506#[doc(hidden)]
2510pub struct BearerAuthMiddleware {
2511 #[doc(hidden)]
2512 pub token: Arc<String>,
2513}
2514
2515impl crate::middleware::Middleware for BearerAuthMiddleware {
2516 fn handle(
2517 &self,
2518 request: &Request,
2519 conn: &ConnectionInfo,
2520 next: &dyn Application,
2521 ) -> Result<Response, String> {
2522 use crate::error::{AppError, IntoResponse};
2523 let expected = format!("Bearer {}", self.token);
2524 let authorized = request
2525 .headers
2526 .iter()
2527 .any(|h| h.name.eq_ignore_ascii_case("authorization") && constant_time_eq(h.value.as_bytes(), expected.as_bytes()));
2528 if authorized {
2529 next.execute(request, conn)
2530 } else {
2531 Ok(AppError::Unauthorized.into_response())
2532 }
2533 }
2534}
2535
2536pub(crate) use crate::core::constant_time_eq;
2537
2538#[doc(hidden)]
2541pub struct AuthMisconfigured;
2542
2543#[doc(hidden)]
2545pub fn cache_max_bytes(max_size: &str) -> Option<u64> {
2546 if max_size.trim().is_empty() {
2547 Some(256 * 1024 * 1024)
2548 } else {
2549 crate::log::access_sink::parse_size(max_size.trim()).filter(|n| *n > 0)
2550 }
2551}
2552
2553#[doc(hidden)]
2556pub struct PluginMisconfigured;
2557
2558impl crate::middleware::Middleware for PluginMisconfigured {
2559 fn handle(&self, _request: &Request, _conn: &ConnectionInfo, _next: &dyn Application) -> Result<Response, String> {
2560 Ok(crate::error::problem(STATUS_CODE_REASON_PHRASE.n503_service_unavailable, "this route's plugin is misconfigured"))
2561 }
2562}
2563
2564impl crate::middleware::Middleware for AuthMisconfigured {
2565 fn handle(
2566 &self,
2567 _request: &Request,
2568 _conn: &ConnectionInfo,
2569 _next: &dyn Application,
2570 ) -> Result<Response, String> {
2571 Ok(crate::error::problem(STATUS_CODE_REASON_PHRASE.n503_service_unavailable, "authentication for this route is misconfigured"))
2572 }
2573}
2574
2575#[doc(hidden)]
2579pub fn arc_app<A: Application + Send + Sync + 'static>(
2580 a: A,
2581) -> Arc<dyn Application + Send + Sync> {
2582 Arc::new(a)
2583}
2584
2585pub fn build_from_file() -> (ConfigDrivenApp, Vec<std::thread::JoinHandle<()>>) {
2590 builder::build_from_file()
2591}