1pub mod parser;
29pub mod health;
30pub(crate) mod upstream;
31
32pub use upstream::{BreakerConfig, RetryConfig, StickyConfig};
33pub mod builder;
34pub mod validate;
35
36#[cfg(test)]
37mod tests;
38
39use std::sync::atomic::{AtomicBool, Ordering};
40use std::sync::{Arc, RwLock};
41
42use crate::app::App;
43use crate::application::Application;
44use crate::core::New;
45use crate::request::Request;
46use crate::header::Header;
47use crate::response::{Response, STATUS_CODE_REASON_PHRASE};
48use crate::server::ConnectionInfo;
49use crate::server_config::ServerConfig;
50
51#[derive(Debug, Clone)]
54pub struct ProxyConfig {
55 pub upstreams: Vec<UpstreamConfig>,
56 pub routes: Vec<RouteConfig>,
57 pub tcp_proxies: Vec<TcpProxyConfig>,
58 pub udp_proxies: Vec<UdpProxyConfig>,
59 pub ws_proxies: Vec<WsProxyConfig>,
60 pub global_middleware: MiddlewareConfig,
61}
62
63#[derive(Debug, Clone)]
64pub struct UpstreamConfig {
65 pub name: String,
66 pub backends: Vec<String>,
67 pub strategy: String, pub health_check: Option<HealthCheckConfig>,
69 pub tls: bool,
73 pub tls_ca_file: Option<String>,
77 pub tls_client_cert_file: Option<String>,
81 pub tls_client_key_file: Option<String>,
82 pub weights: Vec<u32>,
84 pub sticky: Option<StickyConfig>,
86 pub circuit_breaker: Option<BreakerConfig>,
89 pub retry: Option<RetryConfig>,
91}
92
93#[derive(Debug, Clone)]
94pub struct HealthCheckConfig {
95 pub path: String,
96 pub interval_secs: u64,
97 pub timeout_ms: u64,
98 pub healthy_threshold: u32,
99 pub unhealthy_threshold: u32,
100}
101
102#[derive(Debug, Clone)]
103pub struct RouteConfig {
104 pub name: String,
105 pub match_: MatchConfig,
106 pub action: ActionConfig,
107 pub middleware: MiddlewareConfig,
108}
109
110#[derive(Debug, Clone, Default)]
111pub struct MatchConfig {
112 pub host: Option<String>,
113 pub path: Option<String>,
114 pub method: Option<String>,
115 pub content_type: Option<String>,
116}
117
118#[derive(Debug, Clone)]
119pub enum ActionConfig {
120 Proxy {
121 upstream: String,
122 connect_timeout_ms: u64,
123 read_timeout_ms: u64,
124 strip_path_prefix: Option<String>,
125 add_path_prefix: Option<String>,
126 },
127 Grpc {
128 upstream: String,
129 connect_timeout_ms: u64,
130 read_timeout_ms: u64,
131 },
132 Static {
133 root: String,
134 index: Vec<String>,
135 },
136 Redirect {
137 location: String,
138 status: u16,
139 },
140 Respond {
141 status: u16,
142 body: String,
143 content_type: String,
144 },
145 Mcp,
146 Unknown(String),
147}
148
149#[derive(Debug, Clone, Default)]
150pub struct MiddlewareConfig {
151 pub rate_limit: Option<RateLimitConfig>,
152 pub cache: Option<CacheConfig>,
153 pub auth: Option<AuthConfig>,
154 pub rewrite_request: Vec<RewriteRuleConfig>,
155 pub rewrite_response: Vec<RewriteRuleConfig>,
156 pub ip_allow: Vec<String>,
157 pub ip_deny: Vec<String>,
158 pub timeout_ms: Option<u64>,
163 pub max_body_size: Option<u64>,
179}
180
181#[derive(Debug, Clone)]
182pub struct RateLimitConfig {
183 pub max_requests: u32,
184 pub window_secs: u64,
185}
186
187#[derive(Debug, Clone)]
188pub struct CacheConfig {
189 pub ttl_secs: u64,
190 pub vary_by: Vec<String>,
191}
192
193#[derive(Debug, Clone)]
194pub enum AuthConfig {
195 Basic { htpasswd_file: String },
200 Jwt { secret_env: String },
203 Bearer { token_env: String },
204}
205
206#[derive(Debug, Clone, Default)]
207pub struct RewriteRuleConfig {
208 pub type_: String,
209 pub name: Option<String>,
210 pub value: Option<String>,
211 pub prefix: Option<String>,
212 pub from: Option<String>,
213 pub to: Option<String>,
214 pub code: Option<u16>,
215 pub reason: Option<String>,
216}
217
218#[derive(Debug, Clone)]
219pub struct TcpProxyConfig {
220 pub name: String,
221 pub listen: String,
222 pub backends: Vec<String>,
223 pub connect_timeout_ms: u64,
224}
225
226#[derive(Debug, Clone)]
227pub struct UdpProxyConfig {
228 pub name: String,
229 pub listen: String,
230 pub backends: Vec<String>,
231 pub reply_timeout_ms: u64,
232 pub buffer_size: usize,
233}
234
235#[derive(Debug, Clone)]
236pub struct WsProxyConfig {
237 pub name: String,
238 pub listen: String,
239 pub backends: Vec<String>,
240 pub connect_timeout_ms: u64,
241 pub read_timeout_ms: u64,
242 pub health_check: Option<HealthCheckConfig>,
249}
250
251fn merge_action_keys(map: &mut parser::SectionMap, action_section: &str, action_type: &str) {
260 if action_type.is_empty() {
261 return;
262 }
263 let short_form: Vec<(String, String)> = match map.get(action_section) {
264 Some(entries) => entries.iter().filter(|(k, _)| k != "type").cloned().collect(),
265 None => return,
266 };
267 let nested = map.entry(format!("{action_section}.{action_type}")).or_default();
268 for (key, value) in short_form {
269 if !nested.iter().any(|(k, _)| *k == key) {
270 nested.push((key, value));
271 }
272 }
273}
274
275impl ProxyConfig {
276 pub fn is_proxy_mode() -> bool {
280 let path = config_file_path();
281 match std::fs::read_to_string(&path) {
282 Ok(contents) => {
283 contents.contains("[[route]]") || contents.contains("[[upstream]]")
284 }
285 Err(_) => false,
286 }
287 }
288
289 pub fn load() -> Self {
291 let path = config_file_path();
292 let contents = std::fs::read_to_string(&path).unwrap_or_default();
293 Self::from_str(&contents)
294 }
295
296 pub fn from_str(toml: &str) -> Self {
298 use parser::{get_array, get_str, get_u32, get_u64, section_exists};
299
300 let mut map = parser::parse(toml);
301
302 let mut upstreams = Vec::new();
304 let mut i = 0;
305 loop {
306 let sec = format!("upstream[{}]", i);
307 if !section_exists(&map, &sec) {
308 break;
309 }
310 let name = get_str(&map, &sec, "name");
311 let backends = get_array(&map, &sec, "backends");
312 let strategy = {
313 let s = get_str(&map, &sec, "strategy");
314 if s.is_empty() { "round_robin".to_string() } else { s }
315 };
316 let hc_sec = format!("{}.health_check", sec);
317 let health_check = if section_exists(&map, &hc_sec) {
318 Some(HealthCheckConfig {
319 path: {
320 let p = get_str(&map, &hc_sec, "path");
321 if p.is_empty() { "/health".to_string() } else { p }
322 },
323 interval_secs: get_u64(&map, &hc_sec, "interval_secs", 30),
324 timeout_ms: get_u64(&map, &hc_sec, "timeout_ms", 5000),
325 healthy_threshold: get_u32(&map, &hc_sec, "healthy_threshold", 2),
326 unhealthy_threshold: get_u32(&map, &hc_sec, "unhealthy_threshold", 3),
327 })
328 } else {
329 None
330 };
331 let tls = backends.iter().any(|b| ["https://", "h2s://", "grpcs://"].iter().any(|s| b.starts_with(s)));
332 let tls_ca_file = Some(get_str(&map, &sec, "tls_ca_file")).filter(|v| !v.is_empty());
333 let tls_client_cert_file = Some(get_str(&map, &sec, "tls_client_cert_file")).filter(|v| !v.is_empty());
334 let tls_client_key_file = Some(get_str(&map, &sec, "tls_client_key_file")).filter(|v| !v.is_empty());
335 let weights: Vec<u32> = get_array(&map, &sec, "weights").iter().map(|w| w.trim().parse().unwrap_or(0)).collect();
336 let sticky_sec = format!("{}.sticky", sec);
337 let sticky = section_exists(&map, &sticky_sec).then(|| StickyConfig {
338 cookie: Some(get_str(&map, &sticky_sec, "cookie")).filter(|c| !c.is_empty()).unwrap_or_else(|| "rws_backend".to_string()),
339 max_age_secs: Some(get_u64(&map, &sticky_sec, "max_age_secs", 0)).filter(|s| *s > 0),
340 });
341 let cb_sec = format!("{}.circuit_breaker", sec);
342 let circuit_breaker = section_exists(&map, &cb_sec).then(|| BreakerConfig {
343 failure_threshold: get_u32(&map, &cb_sec, "failure_threshold", 5),
344 recovery_secs: get_u64(&map, &cb_sec, "recovery_secs", 30),
345 half_open_probes: get_u32(&map, &cb_sec, "half_open_probes", 1),
346 });
347 let retry_sec = format!("{}.retry", sec);
348 let retry = section_exists(&map, &retry_sec).then(|| {
349 let status: Vec<i16> = get_array(&map, &retry_sec, "status").iter().filter_map(|s| s.trim().parse().ok()).collect();
350 RetryConfig { attempts: get_u32(&map, &retry_sec, "attempts", 1), status: if status.is_empty() { vec![502, 503, 504] } else { status } }
351 });
352 upstreams.push(UpstreamConfig {
353 name, backends, strategy, health_check, tls, tls_ca_file, tls_client_cert_file, tls_client_key_file,
354 weights, sticky, circuit_breaker, retry,
355 });
356 i += 1;
357 }
358
359 let mut routes = Vec::new();
361 let mut i = 0;
362 loop {
363 let sec = format!("route[{}]", i);
364 if !section_exists(&map, &sec) {
365 break;
366 }
367 let name = get_str(&map, &sec, "name");
368
369 let m_sec = format!("{}.match", sec);
371 let match_ = MatchConfig {
372 host: {
373 let h = get_str(&map, &m_sec, "host");
374 if h.is_empty() { None } else { Some(h) }
375 },
376 path: {
377 let p = get_str(&map, &m_sec, "path");
378 if p.is_empty() { None } else { Some(p) }
379 },
380 method: {
381 let m = get_str(&map, &m_sec, "method");
382 if m.is_empty() { None } else { Some(m.to_uppercase()) }
383 },
384 content_type: {
385 let c = get_str(&map, &m_sec, "content_type");
386 if c.is_empty() { None } else { Some(c) }
387 },
388 };
389
390 let a_sec = format!("{}.action", sec);
392 let action_type = get_str(&map, &a_sec, "type");
393 merge_action_keys(&mut map, &a_sec, &action_type);
394 let action = match action_type.as_str() {
395 "proxy" => {
396 let p_sec = format!("{}.action.proxy", sec);
397 ActionConfig::Proxy {
398 upstream: get_str(&map, &p_sec, "upstream"),
399 connect_timeout_ms: get_u64(&map, &p_sec, "connect_timeout_ms", 5000),
400 read_timeout_ms: get_u64(&map, &p_sec, "read_timeout_ms", 30000),
401 strip_path_prefix: {
402 let v = get_str(&map, &p_sec, "strip_path_prefix");
403 if v.is_empty() { None } else { Some(v) }
404 },
405 add_path_prefix: {
406 let v = get_str(&map, &p_sec, "add_path_prefix");
407 if v.is_empty() { None } else { Some(v) }
408 },
409 }
410 }
411 "grpc" => {
412 let p_sec = format!("{}.action.grpc", sec);
413 ActionConfig::Grpc {
414 upstream: get_str(&map, &p_sec, "upstream"),
415 connect_timeout_ms: get_u64(&map, &p_sec, "connect_timeout_ms", 5000),
416 read_timeout_ms: get_u64(&map, &p_sec, "read_timeout_ms", 30000),
417 }
418 }
419 "static" => {
420 let s_sec = format!("{}.action.static", sec);
421 ActionConfig::Static {
422 root: get_str(&map, &s_sec, "root"),
423 index: get_array(&map, &s_sec, "index"),
424 }
425 }
426 "redirect" => {
427 let r_sec = format!("{}.action.redirect", sec);
428 ActionConfig::Redirect {
429 location: get_str(&map, &r_sec, "location"),
430 status: get_u64(&map, &r_sec, "status", 301) as u16,
431 }
432 }
433 "respond" => {
434 let r_sec = format!("{}.action.respond", sec);
435 ActionConfig::Respond {
436 status: get_u64(&map, &r_sec, "status", 200) as u16,
437 body: get_str(&map, &r_sec, "body"),
438 content_type: {
439 let ct = get_str(&map, &r_sec, "content_type");
440 if ct.is_empty() { "text/plain".to_string() } else { ct }
441 },
442 }
443 }
444 "mcp" => ActionConfig::Mcp,
445 other => ActionConfig::Unknown(other.to_string()),
446 };
447
448 let mw_sec = format!("{}.middleware", sec);
450 let middleware = parse_middleware_config(&map, &mw_sec, i);
451
452 routes.push(RouteConfig { name, match_, action, middleware });
453 i += 1;
454 }
455
456 let mut tcp_proxies = Vec::new();
458 let mut i = 0;
459 loop {
460 let sec = format!("tcp_proxy[{}]", i);
461 if !section_exists(&map, &sec) {
462 break;
463 }
464 tcp_proxies.push(TcpProxyConfig {
465 name: get_str(&map, &sec, "name"),
466 listen: get_str(&map, &sec, "listen"),
467 backends: get_array(&map, &sec, "backends"),
468 connect_timeout_ms: get_u64(&map, &sec, "connect_timeout_ms", 5000),
469 });
470 i += 1;
471 }
472
473 let mut udp_proxies = Vec::new();
475 let mut i = 0;
476 loop {
477 let sec = format!("udp_proxy[{}]", i);
478 if !section_exists(&map, &sec) {
479 break;
480 }
481 udp_proxies.push(UdpProxyConfig {
482 name: get_str(&map, &sec, "name"),
483 listen: get_str(&map, &sec, "listen"),
484 backends: get_array(&map, &sec, "backends"),
485 reply_timeout_ms: get_u64(&map, &sec, "reply_timeout_ms", 5000),
486 buffer_size: get_u64(&map, &sec, "buffer_size", 65536) as usize,
487 });
488 i += 1;
489 }
490
491 let mut ws_proxies = Vec::new();
493 let mut i = 0;
494 loop {
495 let sec = format!("ws_proxy[{}]", i);
496 if !section_exists(&map, &sec) {
497 break;
498 }
499 let hc_sec = format!("{}.health_check", sec);
500 let health_check = if section_exists(&map, &hc_sec) {
501 Some(HealthCheckConfig {
502 path: {
503 let p = get_str(&map, &hc_sec, "path");
504 if p.is_empty() { "/health".to_string() } else { p }
505 },
506 interval_secs: get_u64(&map, &hc_sec, "interval_secs", 30),
507 timeout_ms: get_u64(&map, &hc_sec, "timeout_ms", 5000),
508 healthy_threshold: get_u32(&map, &hc_sec, "healthy_threshold", 2),
509 unhealthy_threshold: get_u32(&map, &hc_sec, "unhealthy_threshold", 3),
510 })
511 } else {
512 None
513 };
514 ws_proxies.push(WsProxyConfig {
515 name: get_str(&map, &sec, "name"),
516 listen: get_str(&map, &sec, "listen"),
517 backends: get_array(&map, &sec, "backends"),
518 connect_timeout_ms: get_u64(&map, &sec, "connect_timeout_ms", 5000),
519 read_timeout_ms: get_u64(&map, &sec, "read_timeout_ms", 30000),
520 health_check,
521 });
522 i += 1;
523 }
524
525 let global_middleware = parse_middleware_config(&map, "middleware", usize::MAX);
527
528 ProxyConfig {
529 upstreams,
530 routes,
531 tcp_proxies,
532 udp_proxies,
533 ws_proxies,
534 global_middleware,
535 }
536 }
537}
538
539fn parse_middleware_config(
542 map: &parser::SectionMap,
543 mw_sec: &str,
544 route_idx: usize,
545) -> MiddlewareConfig {
546 use parser::{get_array, get_str, get_u32, get_u64, section_exists};
547
548 let rl_sec = format!("{}.rate_limit", mw_sec);
549 let rate_limit = if section_exists(map, &rl_sec) {
550 Some(RateLimitConfig {
551 max_requests: get_u32(map, &rl_sec, "max_requests", 1000),
552 window_secs: get_u64(map, &rl_sec, "window_secs", 60),
553 })
554 } else {
555 None
556 };
557
558 let c_sec = format!("{}.cache", mw_sec);
559 let cache = if section_exists(map, &c_sec) {
560 Some(CacheConfig {
561 ttl_secs: get_u64(map, &c_sec, "ttl_secs", 60),
562 vary_by: get_array(map, &c_sec, "vary_by"),
563 })
564 } else {
565 None
566 };
567
568 let a_sec = format!("{}.auth", mw_sec);
569 let auth = if section_exists(map, &a_sec) {
570 let auth_type = get_str(map, &a_sec, "type");
571 match auth_type.as_str() {
572 "bearer" => Some(AuthConfig::Bearer {
573 token_env: get_str(map, &a_sec, "token_env"),
574 }),
575 "jwt" => Some(AuthConfig::Jwt {
576 secret_env: get_str(map, &a_sec, "secret_env"),
577 }),
578 "basic" => Some(AuthConfig::Basic {
579 htpasswd_file: get_str(map, &a_sec, "htpasswd_file"),
580 }),
581 _ => None,
582 }
583 } else {
584 None
585 };
586
587 let rewrite_request = collect_rewrite_rules(map, mw_sec, "request");
593 let rewrite_response = collect_rewrite_rules(map, mw_sec, "response");
594
595 let ip_sec = format!("{}.ip_filter", mw_sec);
596 let ip_allow = if section_exists(map, &ip_sec) {
597 get_array(map, &ip_sec, "allow")
598 } else {
599 vec![]
600 };
601 let ip_deny = if section_exists(map, &ip_sec) {
602 get_array(map, &ip_sec, "deny")
603 } else {
604 vec![]
605 };
606
607 let _ = route_idx; let timeout_ms = match get_u64(map, mw_sec, "timeout_ms", 0) {
613 0 => None,
614 ms => Some(ms),
615 };
616
617 let max_body_size = match get_u64(map, mw_sec, "max_body_size", 0) {
621 0 => None,
622 n => Some(n),
623 };
624
625 MiddlewareConfig {
626 rate_limit, cache, auth, rewrite_request, rewrite_response, ip_allow, ip_deny,
627 timeout_ms, max_body_size,
628 }
629}
630
631fn collect_rewrite_rules(
633 map: &parser::SectionMap,
634 mw_sec: &str,
635 direction: &str,
636) -> Vec<RewriteRuleConfig> {
637 use parser::{get_str, get_u64};
638
639 let mut rules = Vec::new();
640 let mut j = 0;
641 loop {
642 let rsec = format!("{}.rewrite.{}[{}]", mw_sec, direction, j);
643 if !parser::section_exists(map, &rsec) {
644 break;
645 }
646 let code_val = get_u64(map, &rsec, "code", 0);
647 rules.push(RewriteRuleConfig {
648 type_: get_str(map, &rsec, "type"),
649 name: {
650 let v = get_str(map, &rsec, "name");
651 if v.is_empty() { None } else { Some(v) }
652 },
653 value: {
654 let v = get_str(map, &rsec, "value");
655 if v.is_empty() { None } else { Some(v) }
656 },
657 prefix: {
658 let v = get_str(map, &rsec, "prefix");
659 if v.is_empty() { None } else { Some(v) }
660 },
661 from: {
662 let v = get_str(map, &rsec, "from");
663 if v.is_empty() { None } else { Some(v) }
664 },
665 to: {
666 let v = get_str(map, &rsec, "to");
667 if v.is_empty() { None } else { Some(v) }
668 },
669 code: if code_val == 0 { None } else { Some(code_val as u16) },
670 reason: {
671 let v = get_str(map, &rsec, "reason");
672 if v.is_empty() { None } else { Some(v) }
673 },
674 });
675 j += 1;
676 }
677 rules
678}
679
680pub(crate) fn config_file_path() -> String {
681 std::env::var("RWS_CONFIG_FILE").unwrap_or_else(|_| "rws.config.toml".to_string())
682}
683
684pub(crate) struct CompiledRoute {
688 pub(crate) matcher: RouteMatcher,
689 pub(crate) handler: Arc<dyn Application + Send + Sync>,
691 #[cfg_attr(not(feature = "http2"), allow(dead_code))]
695 pub(crate) async_proxy: Option<Arc<DynamicProxy>>,
696}
697
698#[derive(Clone, Default)]
700pub(crate) struct RouteMatcher {
701 pub(crate) host: Option<String>,
703 pub(crate) path_prefix: Option<String>,
705 pub(crate) path_exact: Option<String>,
707 pub(crate) method: Option<String>,
709 pub(crate) content_type_prefix: Option<String>,
711}
712
713impl RouteMatcher {
714 pub(crate) fn from_match_config(cfg: &MatchConfig) -> Self {
715 let (path_prefix, path_exact) = match &cfg.path {
716 Some(p) if p.ends_with('*') => {
717 let stripped = p.trim_end_matches('*').to_string();
719 (Some(stripped), None)
720 }
721 Some(p) => (None, Some(p.clone())),
722 None => (None, None),
723 };
724 let content_type_prefix = cfg.content_type.as_ref().map(|ct| {
725 if ct.ends_with('*') {
726 ct.trim_end_matches('*').to_string()
727 } else {
728 ct.clone()
729 }
730 });
731 RouteMatcher {
732 host: cfg.host.clone(),
733 path_prefix,
734 path_exact,
735 method: cfg.method.clone(),
736 content_type_prefix,
737 }
738 }
739
740 pub(crate) fn matches(&self, request: &Request, conn: &ConnectionInfo) -> bool {
742 if let Some(ref expected_host) = self.host {
744 let actual_host = conn
745 .sni_hostname
746 .as_deref()
747 .or_else(|| {
748 request
749 .headers
750 .iter()
751 .find(|h| h.name.eq_ignore_ascii_case("host"))
752 .map(|h| h.value.as_ref())
753 })
754 .unwrap_or("");
755 if !host_matches(expected_host, actual_host) {
756 return false;
757 }
758 }
759
760 if let Some(ref m) = self.method {
762 if request.method.to_uppercase() != m.as_str() {
763 return false;
764 }
765 }
766
767 let path = request.request_uri.split('?').next().unwrap_or(&request.request_uri);
769 if let Some(ref prefix) = self.path_prefix {
770 if !path.starts_with(prefix.as_str()) {
771 return false;
772 }
773 } else if let Some(ref exact) = self.path_exact {
774 if path != exact.as_str() {
775 return false;
776 }
777 }
778
779 if let Some(ref ct_prefix) = self.content_type_prefix {
781 let actual_ct = request
782 .headers
783 .iter()
784 .find(|h| h.name.eq_ignore_ascii_case("content-type"))
785 .map(|h| h.value.as_ref())
786 .unwrap_or("");
787 if !actual_ct.starts_with(ct_prefix.as_str()) {
788 return false;
789 }
790 }
791
792 true
793 }
794}
795
796#[derive(Clone)]
803pub struct ConfigDrivenApp {
804 current: Arc<RwLock<Arc<Generation>>>,
805 reload_lock: Arc<std::sync::Mutex<()>>,
807}
808
809struct Generation {
811 dispatch: RouteDispatch,
812 entry: Option<Arc<dyn Application + Send + Sync>>,
815 config: ProxyConfig,
816 source: Option<String>,
818 health_stop: Arc<AtomicBool>,
820}
821
822impl Generation {
823 fn new(routes: Arc<Vec<CompiledRoute>>, fallback: App, config: ProxyConfig, health_stop: Arc<AtomicBool>, source: Option<String>) -> Self {
824 let dispatch = RouteDispatch { routes, fallback };
825 let entry = if config.global_middleware.is_empty() {
826 None
827 } else {
828 Some(builder::apply_middleware(arc_app(dispatch.clone()), &config.global_middleware))
829 };
830 Generation { dispatch, entry, config, source, health_stop }
831 }
832}
833
834#[derive(Clone)]
836struct RouteDispatch {
837 routes: Arc<Vec<CompiledRoute>>,
838 fallback: App,
843}
844
845impl Application for RouteDispatch {
846 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
847 for route in self.routes.iter() {
848 if route.matcher.matches(request, conn) {
849 return route.handler.execute(request, conn);
850 }
851 }
852 self.fallback.execute(request, conn)
853 }
854
855 fn streams_body(&self, request: &Request, conn: &ConnectionInfo) -> bool {
856 match self.routes.iter().find(|r| r.matcher.matches(request, conn)) {
857 Some(route) => route.handler.streams_body(request, conn),
858 None => self.fallback.streams_body(request, conn),
859 }
860 }
861}
862
863impl MiddlewareConfig {
864 pub fn is_empty(&self) -> bool {
866 self.rate_limit.is_none() && self.cache.is_none() && self.auth.is_none()
867 && self.rewrite_request.is_empty() && self.rewrite_response.is_empty()
868 && self.ip_allow.is_empty() && self.ip_deny.is_empty()
869 && self.timeout_ms.is_none() && self.max_body_size.is_none()
870 }
871}
872
873#[derive(Debug, Clone, PartialEq)]
875pub enum ReloadOutcome {
876 Unchanged,
878 Reloaded { routes: usize, warnings: Vec<String> },
881}
882
883impl ConfigDrivenApp {
884 pub(crate) fn new(routes: Vec<CompiledRoute>, config: ProxyConfig, health_stop: Arc<AtomicBool>, source: Option<String>) -> Self {
885 use crate::core::New;
886 let generation = Generation::new(Arc::new(routes), App::new(), config, health_stop, source);
887 ConfigDrivenApp {
888 current: Arc::new(RwLock::new(Arc::new(generation))),
889 reload_lock: Arc::new(std::sync::Mutex::new(())),
890 }
891 }
892
893 fn current(&self) -> Arc<Generation> {
894 Arc::clone(&self.current.read().unwrap_or_else(|e| e.into_inner()))
895 }
896
897 pub fn with_config(self, config: ServerConfig) -> Self {
914 {
915 let mut slot = self.current.write().unwrap_or_else(|e| e.into_inner());
916 let old = Arc::clone(&slot);
917 *slot = Arc::new(Generation::new(
918 Arc::clone(&old.dispatch.routes),
919 App::with_config(config),
920 old.config.clone(),
921 Arc::clone(&old.health_stop),
922 old.source.clone(),
923 ));
924 }
925 self
926 }
927
928 pub fn reload_from_str(&self, toml: &str) -> Result<ReloadOutcome, crate::config_check::ConfigReport> {
938 let _serialized = self.reload_lock.lock().unwrap_or_else(|e| e.into_inner());
939 let old = self.current();
940 if old.source.as_deref() == Some(toml) {
941 return Ok(ReloadOutcome::Unchanged);
942 }
943 let config = ProxyConfig::from_str(toml);
944 let mut report = crate::config_check::ConfigReport::default();
945 validate::check_unknown_keys(toml, &mut report);
946 config.validate(&mut report);
947 if !report.is_ok() {
948 return Err(report);
949 }
950
951 let mut warnings = report.warnings;
952 let listeners = |c: &ProxyConfig| format!("{:?}{:?}{:?}", c.tcp_proxies, c.udp_proxies, c.ws_proxies);
953 if listeners(&config) != listeners(&old.config) {
954 warnings.push("[[tcp_proxy]] / [[udp_proxy]] / [[ws_proxy]] changes need a restart; the old listeners keep running".to_string());
955 }
956
957 let health_stop = Arc::new(AtomicBool::new(false));
958 let routes = builder::compile_routes(&config, &health_stop);
959 let route_count = routes.len();
960 let generation = Generation::new(Arc::new(routes), old.dispatch.fallback.clone(), config, health_stop, Some(toml.to_string()));
961 let replaced = {
962 let mut slot = self.current.write().unwrap_or_else(|e| e.into_inner());
963 std::mem::replace(&mut *slot, Arc::new(generation))
964 };
965 replaced.health_stop.store(true, Ordering::Relaxed);
966 Ok(ReloadOutcome::Reloaded { routes: route_count, warnings })
967 }
968
969 pub fn reload_from_file(&self) {
974 let path = config_file_path();
975 let toml = match std::fs::read_to_string(&path) {
976 Ok(toml) => toml,
977 Err(e) => {
978 eprintln!("[proxy_config] reload: cannot read {path}: {e} — keeping the running routes");
979 return;
980 }
981 };
982 match self.reload_from_str(&toml) {
983 Ok(ReloadOutcome::Unchanged) => {}
984 Ok(ReloadOutcome::Reloaded { routes, warnings }) => {
985 for w in &warnings {
986 eprintln!("[proxy_config] Warning: {w}");
987 }
988 println!("[proxy_config] reloaded {path}: {routes} route{}", if routes == 1 { "" } else { "s" });
989 }
990 Err(report) => {
991 for e in &report.errors {
992 eprintln!("[proxy_config] reload ERROR: {e}");
993 }
994 eprintln!("[proxy_config] {path} has errors — keeping the running routes");
995 }
996 }
997 }
998}
999
1000fn normalized(request: &Request) -> std::borrow::Cow<'_, Request> {
1003 match normalize_request_target(&request.request_uri) {
1004 std::borrow::Cow::Borrowed(_) => std::borrow::Cow::Borrowed(request),
1005 std::borrow::Cow::Owned(target) => {
1006 let mut owned = request.clone();
1007 owned.request_uri = target;
1008 std::borrow::Cow::Owned(owned)
1009 }
1010 }
1011}
1012
1013pub(crate) fn normalize_request_target(target: &str) -> std::borrow::Cow<'_, str> {
1021 let (path, query) = match target.find('?') {
1022 Some(i) => (&target[..i], &target[i..]),
1023 None => (target, ""),
1024 };
1025 if !path.starts_with('/') || !(path.contains('%') || path.contains("//") || path.contains("/.")) {
1026 return std::borrow::Cow::Borrowed(target);
1027 }
1028 let bytes = path.as_bytes();
1029 let mut decoded = String::with_capacity(path.len());
1030 let mut i = 0;
1031 while i < bytes.len() {
1032 let hex = |b: u8| (b as char).to_digit(16);
1033 if bytes[i] == b'%' && i + 2 < bytes.len() {
1034 if let (Some(h), Some(l)) = (hex(bytes[i + 1]), hex(bytes[i + 2])) {
1035 let value = (h * 16 + l) as u8;
1036 if value.is_ascii_alphanumeric() || matches!(value, b'-' | b'.' | b'_' | b'~') {
1037 decoded.push(value as char);
1038 } else {
1039 decoded.push('%');
1040 decoded.push(bytes[i + 1].to_ascii_uppercase() as char);
1041 decoded.push(bytes[i + 2].to_ascii_uppercase() as char);
1042 }
1043 i += 3;
1044 continue;
1045 }
1046 }
1047 if bytes[i] == b'%' {
1051 decoded.push_str("%25");
1052 i += 1;
1053 continue;
1054 }
1055 let ch = path[i..].chars().next().unwrap_or('\u{FFFD}');
1057 decoded.push(ch);
1058 i += ch.len_utf8().max(1);
1059 }
1060 let mut segments: Vec<&str> = Vec::new();
1061 let parts: Vec<&str> = decoded.split('/').collect();
1062 let last = parts.len().saturating_sub(1);
1063 let mut trailing_slash = false;
1064 for (n, segment) in parts.iter().enumerate() {
1065 match *segment {
1066 "" => trailing_slash = n == last,
1067 "." => trailing_slash = n == last,
1068 ".." => {
1069 segments.pop();
1070 trailing_slash = n == last;
1071 }
1072 s => {
1073 segments.push(s);
1074 trailing_slash = false;
1075 }
1076 }
1077 }
1078 let mut out = String::with_capacity(decoded.len() + query.len());
1079 out.push('/');
1080 out.push_str(&segments.join("/"));
1081 if trailing_slash && !segments.is_empty() {
1082 out.push('/');
1083 }
1084 out.push_str(query);
1085 if out == target {
1086 std::borrow::Cow::Borrowed(target)
1087 } else {
1088 std::borrow::Cow::Owned(out)
1089 }
1090}
1091
1092fn host_matches(expected: &str, actual: &str) -> bool {
1095 let actual = if expected.contains(':') {
1096 actual
1097 } else if let Some(rest) = actual.strip_prefix('[') {
1098 rest.split(']').next().unwrap_or(rest)
1100 } else {
1101 actual.rsplit_once(':').map_or(actual, |(host, port)| if port.bytes().all(|b| b.is_ascii_digit()) { host } else { actual })
1102 };
1103 let expected = expected.trim_start_matches('[').trim_end_matches(']');
1104 actual.eq_ignore_ascii_case(expected)
1105}
1106
1107impl Application for ConfigDrivenApp {
1108 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1109 let request = normalized(request);
1110 let generation = self.current();
1111 match &generation.entry {
1112 Some(entry) => entry.execute(&request, conn),
1113 None => generation.dispatch.execute(&request, conn),
1114 }
1115 }
1116
1117 fn streams_body(&self, request: &Request, conn: &ConnectionInfo) -> bool {
1120 let request = normalized(request);
1121 let generation = self.current();
1122 match &generation.entry {
1123 Some(entry) => entry.streams_body(&request, conn),
1124 None => generation.dispatch.streams_body(&request, conn),
1125 }
1126 }
1127
1128 fn runs_inline(&self, request: &Request, conn: &ConnectionInfo) -> bool {
1132 let request = normalized(request);
1133 let generation = self.current();
1134 generation.entry.is_none()
1135 && !generation.dispatch.routes.iter().any(|r| r.matcher.matches(&request, conn))
1136 && App::is_inline_request(&request)
1137 }
1138
1139 #[cfg(feature = "http2")]
1143 fn execute_async<'a>(&'a self, request: &'a Request, conn: &'a ConnectionInfo) -> Option<crate::application::AsyncResponse<'a>> {
1144 let generation = self.current();
1145 if generation.entry.is_some() {
1146 return None;
1147 }
1148 let request = normalized(request);
1149 let route = generation.dispatch.routes.iter().find(|r| r.matcher.matches(&request, conn))?;
1150 let proxy = Arc::clone(route.async_proxy.as_ref()?);
1151 Some(Box::pin(async move { proxy.execute_async(&request, conn).await }))
1152 }
1153}
1154
1155#[cfg(feature = "http2")]
1161pub(crate) struct H2Route {
1162 live: Arc<RwLock<Vec<String>>>,
1163 tls: bool,
1164 trust: Option<Arc<crate::proxy::UpstreamTls>>,
1165 connect_timeout: std::time::Duration,
1166 read_timeout: std::time::Duration,
1167 counter: std::sync::atomic::AtomicUsize,
1168}
1169
1170#[cfg(feature = "http2")]
1171impl H2Route {
1172 pub(crate) fn new(
1173 live: Arc<RwLock<Vec<String>>>,
1174 tls: bool,
1175 trust: Option<Arc<crate::proxy::UpstreamTls>>,
1176 connect_timeout_ms: u64,
1177 read_timeout_ms: u64,
1178 ) -> H2Route {
1179 H2Route {
1180 live,
1181 tls,
1182 trust,
1183 connect_timeout: std::time::Duration::from_millis(connect_timeout_ms),
1184 read_timeout: std::time::Duration::from_millis(read_timeout_ms),
1185 counter: std::sync::atomic::AtomicUsize::new(0),
1186 }
1187 }
1188}
1189
1190#[cfg(feature = "http2")]
1191impl Application for H2Route {
1192 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1193 use crate::middleware::Middleware;
1194 crate::proxy::GrpcWebLayer::new().handle(request, conn, &H2Forward(self))
1195 }
1196
1197 fn streams_body(&self, _request: &Request, _conn: &ConnectionInfo) -> bool {
1198 true
1199 }
1200}
1201
1202#[cfg(feature = "http2")]
1204struct H2Forward<'a>(&'a H2Route);
1205
1206#[cfg(feature = "http2")]
1207impl Application for H2Forward<'_> {
1208 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1209 let route = self.0;
1210 let backends = route.live.read().unwrap_or_else(|e| e.into_inner()).clone();
1211 if backends.is_empty() {
1212 return Ok(crate::proxy::bad_gateway());
1213 }
1214 let attempts = if matches!(request.body, crate::request::Body::Stream(_)) { 1 } else { backends.len() };
1216 let start = route.counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1217 for attempt in 0..attempts {
1218 let url = &backends[(start + attempt) % backends.len()];
1219 let result = crate::proxy::forward_h2(
1220 request, &conn.client.ip, url, route.tls, route.trust.as_deref(), route.connect_timeout, route.read_timeout,
1221 );
1222 match result {
1223 Ok(response) => return Ok(response),
1224 Err(e) => eprintln!("grpc route: {e}"),
1225 }
1226 }
1227 Ok(crate::proxy::bad_gateway())
1228 }
1229}
1230
1231#[derive(Clone, Copy)]
1236pub(crate) struct NullApp;
1237
1238impl Application for NullApp {
1239 fn execute(&self, _request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
1240 let mut r = Response::new();
1241 r.status_code = *STATUS_CODE_REASON_PHRASE.n404_not_found.status_code;
1242 r.reason_phrase = STATUS_CODE_REASON_PHRASE.n404_not_found.reason_phrase.to_string();
1243 Ok(r)
1244 }
1245}
1246
1247use std::collections::HashMap;
1250use std::collections::hash_map::DefaultHasher;
1251use std::hash::{Hash, Hasher};
1252use std::sync::atomic::AtomicUsize;
1253use std::time::{Duration, SystemTime, UNIX_EPOCH};
1254
1255#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1259pub(crate) enum LoadBalanceStrategy {
1260 RoundRobin,
1261 Random,
1262 IpHash,
1263 LeastConnections,
1264}
1265
1266impl LoadBalanceStrategy {
1267 fn parse(s: &str) -> Self {
1268 match s {
1269 "random" => LoadBalanceStrategy::Random,
1270 "ip_hash" => LoadBalanceStrategy::IpHash,
1271 "least_connections" => LoadBalanceStrategy::LeastConnections,
1272 _ => LoadBalanceStrategy::RoundRobin,
1273 }
1274 }
1275}
1276
1277#[derive(Clone)]
1285pub(crate) struct DynamicProxy {
1286 live: Arc<RwLock<Vec<String>>>,
1287 counter: Arc<AtomicUsize>,
1288 connect_timeout: Duration,
1289 read_timeout: Duration,
1290 strip_prefix: Option<Arc<String>>,
1291 add_prefix: Option<Arc<String>>,
1292 tls: bool,
1293 strategy: LoadBalanceStrategy,
1294 connections: Arc<RwLock<HashMap<String, Arc<AtomicUsize>>>>,
1295 trust: Option<Arc<crate::proxy::UpstreamTls>>,
1297 upstream: Arc<String>,
1299 policy: Option<Arc<upstream::UpstreamPolicy>>,
1301}
1302
1303impl DynamicProxy {
1304 pub(crate) fn with_trust(mut self, trust: Option<Arc<crate::proxy::UpstreamTls>>) -> Self {
1306 self.trust = trust;
1307 self
1308 }
1309
1310 pub(crate) fn with_policy(mut self, upstream: &str, policy: Option<Arc<upstream::UpstreamPolicy>>) -> Self {
1313 self.upstream = Arc::new(upstream.to_string());
1314 self.policy = policy;
1315 self
1316 }
1317
1318 pub(crate) fn new(
1319 live: Arc<RwLock<Vec<String>>>,
1320 connect_timeout_ms: u64,
1321 read_timeout_ms: u64,
1322 strip_prefix: Option<String>,
1323 add_prefix: Option<String>,
1324 tls: bool,
1325 strategy: String,
1326 ) -> Self {
1327 DynamicProxy {
1328 live,
1329 counter: Arc::new(AtomicUsize::new(0)),
1330 connect_timeout: Duration::from_millis(connect_timeout_ms),
1331 read_timeout: Duration::from_millis(read_timeout_ms),
1332 strip_prefix: strip_prefix.map(Arc::new),
1333 add_prefix: add_prefix.map(Arc::new),
1334 tls,
1335 strategy: LoadBalanceStrategy::parse(&strategy),
1336 connections: Arc::new(RwLock::new(HashMap::new())),
1337 trust: None,
1338 upstream: Arc::new(String::new()),
1339 policy: None,
1340 }
1341 }
1342
1343 fn pick(&self, candidates: &[String], client_ip: &str) -> String {
1345 let policy = self.policy.as_deref().filter(|p| p.weighted());
1346 let idx = match self.strategy {
1347 LoadBalanceStrategy::RoundRobin => match policy {
1348 Some(policy) => return policy.weighted_round_robin(candidates).unwrap_or_else(|| candidates[0].clone()),
1349 None => self.counter.fetch_add(1, Ordering::Relaxed) % candidates.len(),
1350 },
1351 LoadBalanceStrategy::Random => {
1352 let nanos = SystemTime::now()
1353 .duration_since(UNIX_EPOCH)
1354 .map(|d| d.subsec_nanos())
1355 .unwrap_or(0) as usize;
1356 let salt = self.counter.fetch_add(1, Ordering::Relaxed);
1357 let roll = nanos.wrapping_add(salt.wrapping_mul(2_654_435_761));
1358 match policy {
1359 Some(policy) => policy.weighted_index(candidates, roll),
1360 None => roll % candidates.len(),
1361 }
1362 }
1363 LoadBalanceStrategy::IpHash => {
1364 let mut hasher = DefaultHasher::new();
1365 client_ip.hash(&mut hasher);
1366 (hasher.finish() as usize) % candidates.len()
1367 }
1368 LoadBalanceStrategy::LeastConnections => {
1369 let connections = self.connections.read().unwrap();
1370 let load = |backend: &String| connections.get(backend).map(|c| c.load(Ordering::Relaxed)).unwrap_or(0) as u64;
1371 let weight = |backend: &String| self.policy.as_deref().map_or(1, |p| p.weight(backend)) as u64;
1372 candidates
1374 .iter()
1375 .enumerate()
1376 .min_by(|(_, a), (_, b)| (load(a) * weight(b)).cmp(&(load(b) * weight(a))))
1377 .map(|(i, _)| i)
1378 .unwrap_or(0)
1379 }
1380 };
1381 candidates[idx].clone()
1382 }
1383
1384 fn next_backend(&self, request: &Request, client_ip: &str, tried: &[String]) -> Result<String, bool> {
1389 let mut candidates: Vec<String> = self.live.read().unwrap().iter().filter(|b| !tried.contains(b)).cloned().collect();
1390 let Some(policy) = self.policy.as_deref() else {
1391 return if candidates.is_empty() { Err(false) } else { Ok(self.pick(&candidates, client_ip)) };
1392 };
1393 if let Some(sticky) = &policy.sticky {
1394 if let Some(id) = upstream::cookie_value(request, &sticky.cookie) {
1395 if let Some(backend) = candidates.iter().find(|b| upstream::backend_id(b) == id).cloned() {
1396 if policy.admits(&backend) {
1397 return Ok(backend);
1398 }
1399 candidates.retain(|b| *b != backend);
1400 }
1401 }
1402 }
1403 let ejected = policy.any_open(&candidates);
1404 while !candidates.is_empty() {
1405 let backend = self.pick(&candidates, client_ip);
1406 if policy.admits(&backend) {
1407 return Ok(backend);
1408 }
1409 candidates.retain(|b| *b != backend);
1410 }
1411 Err(ejected)
1412 }
1413
1414 fn connection_counter(&self, backend: &str) -> Arc<AtomicUsize> {
1417 if let Some(counter) = self.connections.read().unwrap().get(backend) {
1418 return Arc::clone(counter);
1419 }
1420 let mut connections = self.connections.write().unwrap();
1421 Arc::clone(
1422 connections
1423 .entry(backend.to_string())
1424 .or_insert_with(|| Arc::new(AtomicUsize::new(0))),
1425 )
1426 }
1427}
1428
1429struct ConnectionGuard {
1432 counter: Arc<AtomicUsize>,
1433}
1434
1435impl Drop for ConnectionGuard {
1436 fn drop(&mut self) {
1437 self.counter.fetch_sub(1, Ordering::Relaxed);
1438 }
1439}
1440
1441struct Prepared<'a> {
1444 backend: String,
1445 host: String,
1446 port: u16,
1447 request: std::borrow::Cow<'a, Request>,
1448 _guard: Option<ConnectionGuard>,
1449}
1450
1451enum Attempt {
1453 Done(Response),
1455 Failed(Option<Response>),
1458}
1459
1460impl DynamicProxy {
1461 fn prepare<'a>(&self, request: &'a Request, conn: &ConnectionInfo, tried: &[String]) -> Result<Prepared<'a>, Response> {
1466 let backend = self.next_backend(request, &conn.client.ip, tried).map_err(|ejected| if ejected { service_unavailable() } else { bad_gateway() })?;
1467
1468 let guard = if self.strategy == LoadBalanceStrategy::LeastConnections {
1469 let counter = self.connection_counter(&backend);
1470 counter.fetch_add(1, Ordering::Relaxed);
1471 Some(ConnectionGuard { counter })
1472 } else {
1473 None
1474 };
1475
1476 let (host, port, _) = crate::proxy_config::health::parse_backend_url(&backend).ok_or_else(bad_gateway)?;
1477
1478 let request = if self.strip_prefix.is_some() || self.add_prefix.is_some() {
1480 let mut req_clone = request.clone();
1481 if let Some(ref sp) = self.strip_prefix {
1482 if let Some(stripped) = req_clone.request_uri.strip_prefix(sp.as_str()) {
1483 req_clone.request_uri = if stripped.is_empty() || !stripped.starts_with('/') {
1484 format!("/{}", stripped)
1485 } else {
1486 stripped.to_string()
1487 };
1488 }
1489 }
1490 if let Some(ref ap) = self.add_prefix {
1491 req_clone.request_uri = format!("{}{}", ap, req_clone.request_uri);
1492 }
1493 std::borrow::Cow::Owned(req_clone)
1494 } else {
1495 std::borrow::Cow::Borrowed(request)
1496 };
1497
1498 Ok(Prepared { backend, host, port, request, _guard: guard })
1499 }
1500
1501 fn settle(&self, backend: &str, result: Result<Response, ()>, started: std::time::Instant) -> Attempt {
1503 let status = result.as_ref().ok().map(|r| r.status_code);
1504 upstream::observe(&self.upstream, backend, status, started);
1505 let failed = status.is_none_or(|s| matches!(s, 502..=504));
1506 if let Some(policy) = &self.policy {
1507 policy.record(backend, failed);
1508 }
1509 match result {
1510 Ok(response) if self.policy.as_deref().is_some_and(|p| p.retries_status(response.status_code)) => Attempt::Failed(Some(response)),
1511 Ok(response) => Attempt::Done(response),
1512 Err(()) => Attempt::Failed(None),
1513 }
1514 }
1515
1516 fn attempts(&self, request: &Request) -> u32 {
1518 match self.policy.as_deref().and_then(|p| p.retry.as_ref()) {
1519 Some(retry) if upstream::replayable(request) => 1 + retry.attempts,
1520 _ => 1,
1521 }
1522 }
1523
1524 fn stick(&self, request: &Request, backend: &str, response: &mut Response) {
1526 let Some(sticky) = self.policy.as_deref().and_then(|p| p.sticky.as_ref()) else { return };
1527 let id = upstream::backend_id(backend);
1528 if upstream::cookie_value(request, &sticky.cookie).as_deref() == Some(id.as_str()) {
1529 return;
1530 }
1531 let max_age = sticky.max_age_secs.map(|s| format!("; Max-Age={s}")).unwrap_or_default();
1532 response.headers.push(crate::header::Header {
1533 name: "Set-Cookie".into(),
1534 value: format!("{}={id}; Path=/; HttpOnly; SameSite=Lax{max_age}", sticky.cookie).into(),
1535 });
1536 }
1537
1538 #[cfg(feature = "http2")]
1541 pub(crate) async fn execute_async(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1542 let mut tried: Vec<String> = Vec::new();
1543 let mut last: Option<Response> = None;
1544 for attempt in 0..self.attempts(request) {
1545 let prepared = match self.prepare(request, conn, &tried) {
1546 Ok(p) => p,
1547 Err(response) => return Ok(last.unwrap_or(response)),
1548 };
1549 if attempt > 0 {
1550 upstream::observe_retry(&self.upstream);
1551 }
1552 let tls = self.tls.then(|| match &self.trust {
1553 Some(trust) => trust.async_tls(),
1554 None => crate::proxy::async_upstream::AsyncTls {
1555 config: crate::proxy::tls_client_config(),
1556 pool: crate::proxy::async_upstream::default_tls_pool(),
1557 },
1558 });
1559 let started = std::time::Instant::now();
1560 let result = crate::proxy::async_upstream::forward(
1561 &prepared.request,
1562 &conn.client.ip,
1563 &prepared.host,
1564 prepared.port,
1565 tls,
1566 self.connect_timeout,
1567 self.read_timeout,
1568 )
1569 .await
1570 .map_err(|_| ());
1571 match self.settle(&prepared.backend, result, started) {
1572 Attempt::Done(mut response) => {
1573 self.stick(request, &prepared.backend, &mut response);
1574 return Ok(response);
1575 }
1576 Attempt::Failed(response) => last = response,
1577 }
1578 tried.push(prepared.backend.clone());
1579 }
1580 Ok(last.unwrap_or_else(bad_gateway))
1581 }
1582
1583 pub(crate) fn supports_async(&self) -> bool {
1585 cfg!(feature = "http2")
1586 }
1587}
1588
1589impl Application for DynamicProxy {
1590 fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1591 let mut tried: Vec<String> = Vec::new();
1592 let mut last: Option<Response> = None;
1593 for attempt in 0..self.attempts(request) {
1594 let prepared = match self.prepare(request, conn, &tried) {
1595 Ok(p) => p,
1596 Err(response) => return Ok(last.unwrap_or(response)),
1597 };
1598 if attempt > 0 {
1599 upstream::observe_retry(&self.upstream);
1600 }
1601 let started = std::time::Instant::now();
1604 let result = crate::proxy::forward_upstream_trusting(
1605 crate::proxy::shared_pool(),
1606 &prepared.request,
1607 &conn.client.ip,
1608 &prepared.host,
1609 prepared.port,
1610 self.tls,
1611 self.trust.as_deref(),
1612 self.connect_timeout,
1613 self.read_timeout,
1614 )
1615 .map_err(|_| ());
1616 match self.settle(&prepared.backend, result, started) {
1617 Attempt::Done(mut response) => {
1618 self.stick(request, &prepared.backend, &mut response);
1619 return Ok(response);
1620 }
1621 Attempt::Failed(response) => last = response,
1622 }
1623 tried.push(prepared.backend.clone());
1624 }
1625 Ok(last.unwrap_or_else(bad_gateway))
1626 }
1627}
1628
1629fn service_unavailable() -> Response {
1630 use crate::mime_type::MimeType;
1631 use crate::range::Range;
1632 let mut r = Response::new();
1633 r.status_code = *STATUS_CODE_REASON_PHRASE.n503_service_unavailable.status_code;
1634 r.reason_phrase = STATUS_CODE_REASON_PHRASE.n503_service_unavailable.reason_phrase.to_string();
1635 r.headers.push(crate::header::Header { name: "Retry-After".into(), value: "5".into() });
1636 r.content_range_list = vec![Range::get_content_range(b"503 Service Unavailable: every backend is out of rotation".to_vec(), MimeType::TEXT_PLAIN.to_string())];
1637 r
1638}
1639
1640fn bad_gateway() -> Response {
1641 use crate::mime_type::MimeType;
1642 use crate::range::Range;
1643 let cr = Range::get_content_range(
1644 b"502 Bad Gateway".to_vec(),
1645 MimeType::TEXT_PLAIN.to_string(),
1646 );
1647 let mut r = Response::new();
1648 r.status_code = *STATUS_CODE_REASON_PHRASE.n502_bad_gateway.status_code;
1649 r.reason_phrase = STATUS_CODE_REASON_PHRASE.n502_bad_gateway.reason_phrase.to_string();
1650 r.content_range_list = vec![cr];
1651 r
1652}
1653
1654#[derive(Clone)]
1660pub(crate) struct RedirectAdapter {
1661 location_template: Arc<String>,
1662 status: i16,
1663 reason: Arc<String>,
1664}
1665
1666impl RedirectAdapter {
1667 pub(crate) fn new(location: String, status: u16) -> Self {
1668 let (code, reason) = redirect_status(status);
1669 RedirectAdapter {
1670 location_template: Arc::new(location),
1671 status: code,
1672 reason: Arc::new(reason),
1673 }
1674 }
1675}
1676
1677fn redirect_status(code: u16) -> (i16, String) {
1678 let phrase = match code {
1679 301 => STATUS_CODE_REASON_PHRASE.n301_moved_permanently.reason_phrase,
1680 302 => STATUS_CODE_REASON_PHRASE.n302_found.reason_phrase,
1681 307 => STATUS_CODE_REASON_PHRASE.n307_temporary_redirect.reason_phrase,
1682 308 => STATUS_CODE_REASON_PHRASE.n308_permanent_redirect.reason_phrase,
1683 _ => "Redirect",
1684 };
1685 (code as i16, phrase.to_string())
1686}
1687
1688impl Application for RedirectAdapter {
1689 fn execute(&self, request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
1690 let location = self
1691 .location_template
1692 .replace("$path", &request.request_uri);
1693 use crate::header::Header;
1694 let mut r = Response::new();
1695 r.status_code = self.status;
1696 r.reason_phrase = self.reason.as_ref().clone();
1697 r.headers.push(Header { name: "Location".into(), value: location.into() });
1698 Ok(r)
1699 }
1700}
1701
1702#[derive(Clone)]
1706pub(crate) struct RespondAdapter {
1707 status: i16,
1708 reason: Arc<String>,
1709 body: Arc<Vec<u8>>,
1710 content_type: Arc<String>,
1711}
1712
1713impl RespondAdapter {
1714 pub(crate) fn new(status: u16, body: String, content_type: String) -> Self {
1715 use crate::response::STATUS_CODE_REASON_PHRASE;
1716 let reason = match status {
1717 200 => STATUS_CODE_REASON_PHRASE.n200_ok.reason_phrase.to_string(),
1718 201 => STATUS_CODE_REASON_PHRASE.n201_created.reason_phrase.to_string(),
1719 204 => STATUS_CODE_REASON_PHRASE.n204_no_content.reason_phrase.to_string(),
1720 400 => STATUS_CODE_REASON_PHRASE.n400_bad_request.reason_phrase.to_string(),
1721 401 => STATUS_CODE_REASON_PHRASE.n401_unauthorized.reason_phrase.to_string(),
1722 403 => STATUS_CODE_REASON_PHRASE.n403_forbidden.reason_phrase.to_string(),
1723 404 => STATUS_CODE_REASON_PHRASE.n404_not_found.reason_phrase.to_string(),
1724 500 => STATUS_CODE_REASON_PHRASE.n500_internal_server_error.reason_phrase.to_string(),
1725 _ => "OK".to_string(),
1726 };
1727 RespondAdapter {
1728 status: status as i16,
1729 reason: Arc::new(reason),
1730 body: Arc::new(body.into_bytes()),
1731 content_type: Arc::new(content_type),
1732 }
1733 }
1734}
1735
1736impl Application for RespondAdapter {
1737 fn execute(&self, _request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
1738 use crate::range::Range;
1739 let mut r = Response::new();
1740 r.status_code = self.status;
1741 r.reason_phrase = self.reason.as_ref().clone();
1742 if !self.body.is_empty() {
1743 r.content_range_list = vec![Range::get_content_range(
1744 self.body.as_ref().clone(),
1745 self.content_type.as_ref().clone(),
1746 )];
1747 }
1748 Ok(r)
1749 }
1750}
1751
1752#[derive(Clone)]
1761pub(crate) struct StaticAdapter {
1762 root: Arc<std::path::PathBuf>,
1763 index: Arc<Vec<String>>,
1764}
1765
1766impl StaticAdapter {
1767 pub(crate) fn new(root: String, index: Vec<String>) -> Self {
1768 let index = if index.is_empty() { vec!["index.html".to_string()] } else { index };
1769 StaticAdapter {
1770 root: Arc::new(std::path::PathBuf::from(root)),
1771 index: Arc::new(index),
1772 }
1773 }
1774
1775 fn resolve(&self, request_uri: &str) -> Option<std::path::PathBuf> {
1779 let raw_path = request_uri.split('?').next().unwrap_or(request_uri);
1780 let decoded = crate::url::URL::percent_decode(raw_path);
1781
1782 if decoded.split('/').any(|segment| segment == "..") {
1783 return None;
1784 }
1785
1786 let relative = decoded.trim_start_matches('/');
1787 Some(self.root.join(relative))
1788 }
1789}
1790
1791impl Application for StaticAdapter {
1792 fn execute(&self, request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
1793 let mut response = Response::new();
1794
1795 let not_found = |mut response: Response| {
1796 response.status_code = *STATUS_CODE_REASON_PHRASE.n404_not_found.status_code;
1797 response.reason_phrase = STATUS_CODE_REASON_PHRASE.n404_not_found.reason_phrase.to_string();
1798 response
1799 };
1800
1801 let candidate = match self.resolve(&request.request_uri) {
1802 Some(p) => p,
1803 None => {
1804 response.status_code = *STATUS_CODE_REASON_PHRASE.n403_forbidden.status_code;
1805 response.reason_phrase = STATUS_CODE_REASON_PHRASE.n403_forbidden.reason_phrase.to_string();
1806 return Ok(response);
1807 }
1808 };
1809
1810 let mut file_path = candidate;
1811 if file_path.is_dir() {
1812 let indexed = self
1813 .index
1814 .iter()
1815 .map(|name| file_path.join(name))
1816 .find(|p| p.is_file());
1817
1818 file_path = match indexed {
1819 Some(p) => p,
1820 None => return Ok(not_found(response)),
1821 };
1822 }
1823
1824 if !file_path.is_file() {
1825 return Ok(not_found(response));
1826 }
1827
1828 if let (Ok(root_canon), Ok(file_canon)) =
1831 (self.root.canonicalize(), file_path.canonicalize())
1832 {
1833 if !file_canon.starts_with(&root_canon) {
1834 return Ok(not_found(response));
1835 }
1836 }
1837
1838 let path_str = match file_path.to_str() {
1839 Some(s) => s,
1840 None => return Ok(not_found(response)),
1841 };
1842
1843 match crate::range::Range::get_content_range_of_a_file(path_str) {
1844 Ok(content_range) => {
1845 response.status_code = *STATUS_CODE_REASON_PHRASE.n200_ok.status_code;
1846 response.reason_phrase = STATUS_CODE_REASON_PHRASE.n200_ok.reason_phrase.to_string();
1847 response.content_range_list = vec![content_range];
1848 Ok(response)
1849 }
1850 Err(_) => Ok(not_found(response)),
1851 }
1852 }
1853}
1854
1855pub(crate) struct PerRouteRateLimit(pub(crate) Arc<crate::rate_limit::RateLimiter>);
1859
1860impl crate::middleware::Middleware for PerRouteRateLimit {
1861 fn handle(
1862 &self,
1863 request: &Request,
1864 conn: &ConnectionInfo,
1865 next: &dyn Application,
1866 ) -> Result<Response, String> {
1867 use crate::error::{AppError, IntoResponse};
1868 if self.0.check(&conn.client.ip) {
1869 next.execute(request, conn)
1870 } else {
1871 let mut response = AppError::TooManyRequests.into_response();
1872 let retry_after = self.0.retry_after(&conn.client.ip);
1873 response.headers.push(Header {
1874 name: Header::_RETRY_AFTER.into(),
1875 value: retry_after.to_string().into(),
1876 });
1877 Ok(response)
1878 }
1879 }
1880}
1881
1882pub(crate) struct PerRouteMaxBodySize(pub(crate) u64);
1897
1898impl crate::middleware::Middleware for PerRouteMaxBodySize {
1899 fn handle(
1900 &self,
1901 request: &Request,
1902 conn: &ConnectionInfo,
1903 next: &dyn Application,
1904 ) -> Result<Response, String> {
1905 use crate::error::{AppError, IntoResponse};
1906 if request.body.len() as u64 > self.0 {
1907 Ok(AppError::PayloadTooLarge(format!(
1908 "request body of {} bytes exceeds this route's {}-byte limit",
1909 request.body.len(),
1910 self.0,
1911 ))
1912 .into_response())
1913 } else {
1914 next.execute(request, conn)
1915 }
1916 }
1917}
1918
1919pub(crate) struct BearerAuthMiddleware {
1923 pub(crate) token: Arc<String>,
1924}
1925
1926impl crate::middleware::Middleware for BearerAuthMiddleware {
1927 fn handle(
1928 &self,
1929 request: &Request,
1930 conn: &ConnectionInfo,
1931 next: &dyn Application,
1932 ) -> Result<Response, String> {
1933 use crate::error::{AppError, IntoResponse};
1934 let expected = format!("Bearer {}", self.token);
1935 let authorized = request
1936 .headers
1937 .iter()
1938 .any(|h| h.name.eq_ignore_ascii_case("authorization") && constant_time_eq(h.value.as_bytes(), expected.as_bytes()));
1939 if authorized {
1940 next.execute(request, conn)
1941 } else {
1942 Ok(AppError::Unauthorized.into_response())
1943 }
1944 }
1945}
1946
1947pub(crate) fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
1950 if a.len() != b.len() {
1951 return false;
1952 }
1953 a.iter().zip(b).fold(0u8, |diff, (x, y)| diff | (x ^ y)) == 0
1954}
1955
1956pub(crate) struct AuthMisconfigured;
1959
1960impl crate::middleware::Middleware for AuthMisconfigured {
1961 fn handle(
1962 &self,
1963 _request: &Request,
1964 _conn: &ConnectionInfo,
1965 _next: &dyn Application,
1966 ) -> Result<Response, String> {
1967 Ok(Response::text(
1968 STATUS_CODE_REASON_PHRASE.n503_service_unavailable,
1969 "503 Service Unavailable: authentication for this route is misconfigured",
1970 ))
1971 }
1972}
1973
1974pub(crate) fn arc_app<A: Application + Send + Sync + 'static>(
1978 a: A,
1979) -> Arc<dyn Application + Send + Sync> {
1980 Arc::new(a)
1981}
1982
1983pub fn build_from_file() -> (ConfigDrivenApp, Vec<std::thread::JoinHandle<()>>) {
1988 builder::build_from_file()
1989}