1use std::net::SocketAddr;
11use std::sync::Arc;
12
13use axum::Router;
14use axum::body::Body;
15use axum::extract::{Request, State};
16use axum::http::{HeaderMap, HeaderValue, StatusCode, Uri};
17use axum::response::{IntoResponse, Response};
18use hyper::header::{COOKIE, HOST};
19
20const PITCHFORK_HEADER: &str = "x-pitchfork";
22
23const PROXY_HOPS_HEADER: &str = "x-pitchfork-hops";
26
27const MAX_PROXY_HOPS: u64 = 5;
29
30const HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
37
38const MAX_HOST_CERTS: usize = 256;
46
47const REFUSAL_LOG_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);
49
50static REFUSED_HANDSHAKE: crate::proxy::LogThrottle = crate::proxy::LogThrottle::new();
52
53#[cfg(feature = "proxy-tls")]
55static REFUSED_SNI: crate::proxy::LogThrottle = crate::proxy::LogThrottle::new();
56
57pub(crate) const SHUTDOWN_DRAIN_BUDGET: std::time::Duration = std::time::Duration::from_secs(10);
62
63const MAX_TUNNELS: usize = 256;
70
71static REFUSED_TUNNEL: crate::proxy::LogThrottle = crate::proxy::LogThrottle::new();
73
74static ABANDONED_HANDSHAKE: crate::proxy::LogThrottle = crate::proxy::LogThrottle::new();
82
83static ABANDONED_TUNNEL: crate::proxy::LogThrottle = crate::proxy::LogThrottle::new();
85
86const MAX_PENDING_HANDSHAKES: usize = 512;
93
94const HOP_BY_HOP_HEADERS: &[&str] = &[
97 "connection",
98 "keep-alive",
99 "proxy-connection",
100 "transfer-encoding",
101 "upgrade",
102];
103
104use hyper_util::client::legacy::Client;
105use hyper_util::client::legacy::connect::HttpConnector;
106use hyper_util::rt::TokioExecutor;
107use tokio::net::{TcpListener, TcpStream};
108
109use crate::daemon_id::DaemonId;
110use crate::pitchfork_toml::{ProxyProtocol, ProxyTlsMode};
111use crate::proxy::activity::{ACTIVITY, ActivityGuard, GuardedBody};
112use crate::settings::settings;
113use crate::supervisor::SUPERVISOR;
114
115const SLUG_CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(2);
128
129#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
135pub struct ProxyTlsRoute {
136 pub mode: ProxyTlsMode,
138 pub protocol: ProxyProtocol,
140 pub port: Option<u16>,
143}
144
145#[derive(Clone, Debug)]
147pub struct CachedSlugEntry {
148 pub slug: String,
150 pub namespace: Option<String>,
152 pub daemon_name: String,
154 pub dir: std::path::PathBuf,
156 pub worktrees: Vec<crate::proxy::worktree::WorktreeEntry>,
158 pub rejected_worktree_prefixes: std::collections::HashSet<String>,
162 pub tls: ProxyTlsRoute,
164 pub worktree_tls: std::collections::HashMap<String, ProxyTlsRoute>,
168}
169
170impl CachedSlugEntry {
171 fn known_route(
176 &self,
177 dir: &std::path::Path,
178 namespace: Option<&str>,
179 daemon_name: &str,
180 ) -> Option<ProxyTlsRoute> {
181 (self.dir == dir
182 && self.namespace.as_deref() == namespace
183 && self.daemon_name == daemon_name)
184 .then_some(self.tls)
185 }
186
187 fn known_worktree_route(
190 &self,
191 wt: &crate::proxy::worktree::WorktreeEntry,
192 daemon_name: &str,
193 ) -> Option<ProxyTlsRoute> {
194 if self.daemon_name != daemon_name {
195 return None;
196 }
197 let branch = wt.sanitized_branch.to_ascii_lowercase();
198 self.worktrees
199 .iter()
200 .any(|known| {
201 known.sanitized_branch.eq_ignore_ascii_case(&branch)
202 && known.path == wt.path
203 && known.namespace == wt.namespace
204 })
205 .then(|| self.worktree_tls.get(&branch).copied())
206 .flatten()
207 }
208}
209
210struct SlugCache {
212 entries: Arc<std::collections::HashMap<String, CachedSlugEntry>>,
213 expires_at: std::time::Instant,
214}
215
216static SLUG_CACHE: once_cell::sync::Lazy<std::sync::RwLock<SlugCache>> =
226 once_cell::sync::Lazy::new(|| {
227 std::sync::RwLock::new(SlugCache {
228 entries: Arc::new(std::collections::HashMap::new()),
229 expires_at: std::time::Instant::now(), })
231 });
232
233fn slug_snapshot() -> Arc<std::collections::HashMap<String, CachedSlugEntry>> {
239 let guard = SLUG_CACHE
240 .read()
241 .unwrap_or_else(std::sync::PoisonError::into_inner);
242 Arc::clone(&guard.entries)
243}
244
245static SLUG_REFRESH: once_cell::sync::Lazy<tokio::sync::Mutex<()>> =
254 once_cell::sync::Lazy::new(|| tokio::sync::Mutex::new(()));
255
256fn fresh_slugs() -> Option<Arc<std::collections::HashMap<String, CachedSlugEntry>>> {
258 let cache = SLUG_CACHE
259 .read()
260 .unwrap_or_else(std::sync::PoisonError::into_inner);
261 (std::time::Instant::now() < cache.expires_at).then(|| Arc::clone(&cache.entries))
262}
263
264fn reject_case_colliding_worktrees(
271 wts: Vec<crate::proxy::worktree::WorktreeEntry>,
272) -> (
273 Vec<crate::proxy::worktree::WorktreeEntry>,
274 std::collections::HashSet<String>,
275) {
276 let collisions =
277 crate::proxy::ascii_case_collisions(wts.iter().map(|w| w.sanitized_branch.as_str()));
278 if collisions.is_empty() {
279 return (wts, collisions);
280 }
281
282 let (dropped, kept): (Vec<_>, Vec<_>) = wts
283 .into_iter()
284 .partition(|w| collisions.contains(&w.sanitized_branch.to_ascii_lowercase()));
285
286 let mut folded: Vec<&String> = collisions.iter().collect();
287 folded.sort();
288 for key in folded {
289 let mut branches: Vec<&str> = dropped
290 .iter()
291 .filter(|w| w.sanitized_branch.eq_ignore_ascii_case(key))
292 .map(|w| w.branch.as_str())
293 .collect();
294 branches.sort();
295 log::warn!(
296 "Worktree slug collision: branches [{}] all route to '{key}' under \
297 case-insensitive host matching. None of them will be routed; \
298 rename a branch to disambiguate.",
299 branches.join(", "),
300 );
301 }
302
303 (kept, collisions)
304}
305
306pub(crate) fn read_proxy_tls_route(
329 dir: &std::path::Path,
330 namespace: Option<&str>,
331 daemon_name: &str,
332) -> miette::Result<Option<ProxyTlsRoute>> {
333 let Some(id) = namespace.and_then(|ns| DaemonId::try_new(ns, daemon_name).ok()) else {
334 return Ok(None);
335 };
336 let pt = crate::pitchfork_toml::PitchforkToml::all_merged_from(dir)?;
337 Ok(pt.daemons.get(&id).map(|cfg| ProxyTlsRoute {
338 mode: cfg.proxy_tls.unwrap_or_default(),
339 protocol: cfg.proxy_protocol.unwrap_or_default(),
340 port: cfg.effective_proxy_tls_port(),
341 }))
342}
343
344fn route_or_last_known(
350 read: miette::Result<Option<ProxyTlsRoute>>,
351 known: Option<ProxyTlsRoute>,
352 dir: &std::path::Path,
353 daemon_name: &str,
354) -> Option<ProxyTlsRoute> {
355 read.unwrap_or_else(|e| {
356 crate::proxy::hostname::warn_once(&format!(
357 "Proxy TLS route for daemon '{daemon_name}': could not read config in {}; \
358 keeping its last known TLS mode until the config is fixed: {e}",
359 dir.display()
360 ));
361 known
362 })
363}
364
365fn build_slug_entries(
377 previous: &std::collections::HashMap<String, CachedSlugEntry>,
378) -> std::collections::HashMap<String, CachedSlugEntry> {
379 let global_slugs = crate::pitchfork_toml::PitchforkToml::read_global_slugs();
380 let collisions = crate::proxy::ascii_case_collisions(global_slugs.keys().map(String::as_str));
381 let mut folded: Vec<&String> = collisions.iter().collect();
382 folded.sort();
383 for key in folded {
384 let mut spellings: Vec<&str> = global_slugs
385 .keys()
386 .filter(|s| s.eq_ignore_ascii_case(key))
387 .map(String::as_str)
388 .collect();
389 spellings.sort();
390 log::warn!(
391 "Slug collision: [{}] differ only by case and host names are case-insensitive. \
392 None of them will be routed; remove or rename all but one.",
393 spellings.join(", "),
394 );
395 }
396
397 let mut entries: std::collections::HashMap<String, CachedSlugEntry> =
398 std::collections::HashMap::with_capacity(global_slugs.len());
399 let worktree_enabled = crate::settings::settings().general.worktree;
400 for (slug, entry) in &global_slugs {
401 let key = slug.to_ascii_lowercase();
402 if collisions.contains(&key) {
403 continue;
404 }
405 let ns = entry.resolve_namespace();
406 let daemon_name = entry.daemon.as_deref().unwrap_or(slug).to_string();
407 let (worktrees, rejected_worktree_prefixes) = if worktree_enabled {
408 let wts = match entry.resolve_dir() {
409 Some(dir) => crate::proxy::worktree::discover_worktrees(&dir),
410 None => vec![],
411 };
412 let wts = wts
413 .into_iter()
414 .map(|mut wt| {
415 wt.namespace =
416 crate::pitchfork_toml::PitchforkToml::namespace_for_dir(&wt.path).ok();
417 wt
418 })
419 .collect();
420 reject_case_colliding_worktrees(wts)
421 } else {
422 (vec![], std::collections::HashSet::new())
423 };
424 let dir = entry.resolve_dir().unwrap_or_default();
425 let prev = previous.get(&key);
426 let tls = route_or_last_known(
430 read_proxy_tls_route(&dir, ns.as_deref(), &daemon_name),
431 prev.and_then(|p| p.known_route(&dir, ns.as_deref(), &daemon_name)),
432 &dir,
433 &daemon_name,
434 )
435 .unwrap_or_default();
436 let worktree_tls = worktrees
440 .iter()
441 .filter_map(|wt| {
442 let branch = wt.sanitized_branch.to_ascii_lowercase();
443 route_or_last_known(
444 read_proxy_tls_route(&wt.path, wt.namespace.as_deref(), &daemon_name),
445 prev.and_then(|p| p.known_worktree_route(wt, &daemon_name)),
446 &wt.path,
447 &daemon_name,
448 )
449 .map(|route| (branch, route))
450 })
451 .collect();
452 entries.insert(
453 key,
454 CachedSlugEntry {
455 slug: slug.clone(),
456 namespace: ns,
457 daemon_name,
458 dir,
459 worktrees,
460 rejected_worktree_prefixes,
461 tls,
462 worktree_tls,
463 },
464 );
465 }
466 entries
467}
468
469pub async fn get_cached_slugs() -> Arc<std::collections::HashMap<String, CachedSlugEntry>> {
479 if let Some(entries) = fresh_slugs() {
481 return entries;
482 }
483
484 let _refreshing = SLUG_REFRESH.lock().await;
486
487 if let Some(entries) = fresh_slugs() {
489 return entries;
490 }
491
492 let previous = slug_snapshot();
494 let new_entries = Arc::new(
495 tokio::task::spawn_blocking(move || build_slug_entries(&previous))
496 .await
497 .unwrap_or_else(|e| {
498 log::warn!("Failed to refresh slug cache: {e}");
499 std::collections::HashMap::new()
500 }),
501 );
502
503 let mut cache = SLUG_CACHE
504 .write()
505 .unwrap_or_else(std::sync::PoisonError::into_inner);
506 cache.entries = Arc::clone(&new_entries);
507 cache.expires_at = std::time::Instant::now() + SLUG_CACHE_TTL;
508 new_entries
509}
510
511struct RegistryCache {
519 registry: Arc<crate::proxy::hostname::HostRegistry>,
520 expires_at: std::time::Instant,
521}
522
523static HOST_REGISTRY: once_cell::sync::Lazy<std::sync::RwLock<RegistryCache>> =
530 once_cell::sync::Lazy::new(|| {
531 std::sync::RwLock::new(RegistryCache {
532 registry: Arc::new(crate::proxy::hostname::HostRegistry::default()),
533 expires_at: std::time::Instant::now(), })
535 });
536
537static REGISTRY_REFRESH: once_cell::sync::Lazy<tokio::sync::Mutex<()>> =
540 once_cell::sync::Lazy::new(|| tokio::sync::Mutex::new(()));
541
542fn fresh_registry() -> Option<Arc<crate::proxy::hostname::HostRegistry>> {
544 let cache = HOST_REGISTRY
545 .read()
546 .unwrap_or_else(std::sync::PoisonError::into_inner);
547 (std::time::Instant::now() < cache.expires_at).then(|| Arc::clone(&cache.registry))
548}
549
550fn registry_snapshot() -> Arc<crate::proxy::hostname::HostRegistry> {
556 let cache = HOST_REGISTRY
557 .read()
558 .unwrap_or_else(std::sync::PoisonError::into_inner);
559 Arc::clone(&cache.registry)
560}
561
562pub async fn get_cached_host_registry() -> Arc<crate::proxy::hostname::HostRegistry> {
564 if let Some(registry) = fresh_registry() {
565 return registry;
566 }
567
568 let _refreshing = REGISTRY_REFRESH.lock().await;
570 if let Some(registry) = fresh_registry() {
571 return registry;
572 }
573
574 let registry = Arc::new(
575 tokio::task::spawn_blocking(crate::proxy::hostname::HostRegistry::build)
576 .await
577 .unwrap_or_else(|e| {
578 log::warn!("Failed to refresh hostname registry: {e}");
579 crate::proxy::hostname::HostRegistry::default()
580 }),
581 );
582 for err in ®istry.errors {
583 crate::proxy::hostname::warn_once(err);
584 }
585
586 let mut cache = HOST_REGISTRY
587 .write()
588 .unwrap_or_else(std::sync::PoisonError::into_inner);
589 cache.registry = Arc::clone(®istry);
590 cache.expires_at = std::time::Instant::now() + SLUG_CACHE_TTL;
591 registry
592}
593
594fn wildcard_slug_lookup<'a>(
604 subdomain: &str,
605 entries: &'a std::collections::HashMap<String, CachedSlugEntry>,
606 wildcard: bool,
607) -> Option<&'a CachedSlugEntry> {
608 let subdomain = subdomain.to_ascii_lowercase();
609
610 entries.get(&subdomain).or_else(|| {
611 if !wildcard {
612 return None;
613 }
614 subdomain
616 .match_indices('.')
617 .map(|(i, _)| &subdomain[i + 1..])
618 .find_map(|candidate| entries.get(candidate))
619 })
620}
621
622#[derive(Debug)]
624enum PrefixMatch<'a> {
625 Worktree(&'a crate::proxy::worktree::WorktreeEntry),
627 Unknown,
630 Ambiguous,
634}
635
636fn match_worktree_prefix<'a>(cached: &'a CachedSlugEntry, prefix: &str) -> PrefixMatch<'a> {
638 if let Some(wt) = cached
639 .worktrees
640 .iter()
641 .find(|w| w.sanitized_branch.eq_ignore_ascii_case(prefix))
642 {
643 return PrefixMatch::Worktree(wt);
644 }
645 if cached
646 .rejected_worktree_prefixes
647 .contains(&prefix.to_ascii_lowercase())
648 {
649 return PrefixMatch::Ambiguous;
650 }
651 PrefixMatch::Unknown
652}
653
654fn worktree_route(cached: &CachedSlugEntry, sanitized_branch: &str) -> ProxyTlsRoute {
668 cached
669 .worktree_tls
670 .get(&sanitized_branch.to_ascii_lowercase())
671 .copied()
672 .unwrap_or(ProxyTlsRoute {
673 mode: cached.tls.mode,
674 protocol: cached.tls.protocol,
675 port: None,
676 })
677}
678
679fn strip_dot_suffix_ignore_case(s: &str, suffix: &str) -> Option<String> {
683 let needle_len = suffix.len() + 1;
684 if s.len() <= needle_len {
685 return None;
686 }
687 let split = s.len() - needle_len;
688 if !s.is_char_boundary(split) {
689 return None;
690 }
691 let (head, tail) = s.split_at(split);
692 if tail.starts_with('.') && tail[1..].eq_ignore_ascii_case(suffix) {
693 Some(head.to_string())
694 } else {
695 None
696 }
697}
698
699async fn cached_slug_lookup(subdomain: &str) -> Option<CachedSlugEntry> {
706 let entries = get_cached_slugs().await;
707 wildcard_slug_lookup(subdomain, &entries, settings().proxy.wildcard).cloned()
708}
709
710static AUTO_START_IN_PROGRESS: once_cell::sync::Lazy<
717 tokio::sync::Mutex<std::collections::HashSet<DaemonId>>,
718> = once_cell::sync::Lazy::new(|| tokio::sync::Mutex::new(std::collections::HashSet::new()));
719
720enum ResolveResult {
722 Ready(u16, Option<ActivityGuard>, ProxyProtocol),
728 Starting { slug: String },
730 NotFound,
732 Page {
735 project: String,
736 worktree: Option<String>,
737 daemons: Vec<String>,
738 dir: Option<std::path::PathBuf>,
742 },
743 Unknown { heading: String, known: Vec<String> },
745 Error(String),
747}
748
749type OnErrorFn = Arc<dyn Fn(&str) + Send + Sync>;
752
753#[derive(Clone)]
754struct ProxyState {
755 client: Arc<Client<HttpConnector, Body>>,
757 h2c_client: Arc<Client<HttpConnector, Body>>,
760 tld: String,
762 is_tls: bool,
764 connect_target: Option<SocketAddr>,
766 contact_ip: std::net::IpAddr,
769 cancel: tokio_util::sync::CancellationToken,
771 tunnel_slots: Arc<tokio::sync::Semaphore>,
773 tunnels: tokio_util::task::TaskTracker,
780 on_error: Option<OnErrorFn>,
782}
783
784pub async fn serve(
792 bind_tx: tokio::sync::oneshot::Sender<std::result::Result<(), String>>,
793 cancel: tokio_util::sync::CancellationToken,
794) -> crate::Result<()> {
795 let s = settings();
796 let lan_enabled = s.proxy.lan || !s.proxy.lan_ip.is_empty();
797
798 let effective_tld = crate::proxy::effective_tld(&s).to_string();
799
800 let Some(effective_port) = u16::try_from(s.proxy.port).ok().filter(|&p| p > 0) else {
801 let msg = format!(
802 "proxy.port {} is out of valid port range (1-65535), proxy server cannot start",
803 s.proxy.port
804 );
805 let _ = bind_tx.send(Err(msg.clone()));
806 miette::bail!("{msg}");
807 };
808
809 let mut connector = HttpConnector::new();
810 connector.set_connect_timeout(Some(std::time::Duration::from_secs(10)));
814
815 let client = Client::builder(TokioExecutor::new())
816 .pool_idle_timeout(std::time::Duration::from_secs(30))
819 .build(connector.clone());
820
821 let h2c_client = Client::builder(TokioExecutor::new())
822 .pool_idle_timeout(std::time::Duration::from_secs(30))
823 .http2_only(true)
824 .build(connector);
825
826 let bind_ip: std::net::IpAddr = if lan_enabled && s.proxy.host == "127.0.0.1" {
830 std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED)
831 } else {
832 match s.proxy.host.parse() {
833 Ok(ip) => ip,
834 Err(_) => {
835 log::warn!(
836 "proxy.host {:?} is not a valid IP address — falling back to 127.0.0.1. \
837 The proxy will only be reachable on the loopback interface.",
838 s.proxy.host
839 );
840 std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)
841 }
842 }
843 };
844 let addr = SocketAddr::from((bind_ip, effective_port));
845 let contact_ip = local_contact_ip(bind_ip);
850 let tunnels = tokio_util::task::TaskTracker::new();
851
852 let state = ProxyState {
853 client: Arc::new(client),
854 h2c_client: Arc::new(h2c_client),
855 tld: effective_tld.clone(),
856 is_tls: s.proxy.https,
857 connect_target: Some(SocketAddr::from((contact_ip, effective_port))),
860 contact_ip,
861 cancel: cancel.clone(),
862 tunnel_slots: Arc::new(tokio::sync::Semaphore::new(MAX_TUNNELS)),
863 tunnels: tunnels.clone(),
864 on_error: None,
865 };
866
867 let plain_state = state.clone();
872 let app = Router::new()
873 .route(crate::proxy::pac::PAC_PATH, axum::routing::any(pac_handler))
879 .fallback(proxy_handler)
880 .with_state(state);
881
882 if s.proxy.https {
883 serve_https_with_http_fallback(
884 app,
885 addr,
886 &s,
887 effective_port,
888 effective_tld,
889 plain_state,
890 bind_tx,
891 cancel,
892 )
893 .await
894 } else {
895 serve_http(app, addr, effective_port, plain_state, bind_tx, cancel).await
899 }
900}
901
902macro_rules! serve_conn_until_cancelled {
909 ($io:expr, $svc:expr, $cancel:expr) => {{
910 let builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
911 let conn = builder.serve_connection_with_upgrades($io, $svc);
912 tokio::pin!(conn);
913 tokio::select! {
914 r = conn.as_mut() => r,
915 _ = $cancel.cancelled() => {
916 conn.as_mut().graceful_shutdown();
917 conn.await
918 }
919 }
920 }};
921}
922
923async fn serve_http(
925 app: Router,
926 addr: SocketAddr,
927 effective_port: u16,
928 proxy_state: ProxyState,
929 bind_tx: tokio::sync::oneshot::Sender<std::result::Result<(), String>>,
930 cancel: tokio_util::sync::CancellationToken,
931) -> crate::Result<()> {
932 let listener = match TcpListener::bind(addr).await {
933 Ok(l) => {
934 if settings().proxy.sync_hosts {
935 crate::proxy::hosts::sync_hosts_from_settings();
936 }
937 let _ = bind_tx.send(Ok(()));
938 l
939 }
940 Err(e) => {
941 let msg = bind_error_message(effective_port, &e);
942 let _ = bind_tx.send(Err(msg.clone()));
943 return Err(miette::miette!("{msg}"));
944 }
945 };
946
947 log::info!("Proxy server listening on http://{addr}");
948 if effective_port < 1024 {
949 log::info!(
950 "Note: port {effective_port} is a privileged port. \
951 The supervisor must be started with sudo to bind to this port."
952 );
953 }
954 let mut conn_tasks: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
961 loop {
962 while conn_tasks.try_join_next().is_some() {}
963 tokio::select! {
964 accept_result = listener.accept() => {
965 let (stream, peer_addr) = match accept_result {
966 Ok(conn) => conn,
967 Err(e) => {
968 log::warn!("Accept error (will retry): {e}");
969 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
970 continue;
971 }
972 };
973 let app = app
976 .clone()
977 .layer(axum::Extension(axum::extract::ConnectInfo(peer_addr)));
978 let cancel = cancel.clone();
979 conn_tasks.spawn(async move {
980 let io = hyper_util::rt::TokioIo::new(stream);
981 let svc = hyper_util::service::TowerToHyperService::new(app);
982 if let Err(e) = serve_conn_until_cancelled!(io, svc, cancel) {
983 log::debug!("Connection error: {e}");
984 }
985 });
986 }
987 _ = cancel.cancelled() => break,
988 }
989 }
990
991 let deadline = tokio::time::Instant::now() + SHUTDOWN_DRAIN_BUDGET;
994 if tokio::time::timeout_at(deadline, async {
995 while conn_tasks.join_next().await.is_some() {}
996 })
997 .await
998 .is_err()
999 {
1000 log::debug!("Proxy connections still open after {SHUTDOWN_DRAIN_BUDGET:?}; aborting them");
1001 }
1002 drop(conn_tasks);
1004
1005 proxy_state.tunnels.close();
1009 let _ = tokio::time::timeout_at(deadline, proxy_state.tunnels.wait()).await;
1010 Ok(())
1011}
1012
1013#[cfg(feature = "proxy-tls")]
1019#[allow(clippy::too_many_arguments)]
1020async fn serve_https_with_http_fallback(
1021 app: Router,
1022 addr: SocketAddr,
1023 s: &crate::settings::Settings,
1024 effective_port: u16,
1025 effective_tld: String,
1026 plain_proxy_state: ProxyState,
1027 bind_tx: tokio::sync::oneshot::Sender<std::result::Result<(), String>>,
1028 cancel: tokio_util::sync::CancellationToken,
1029) -> crate::Result<()> {
1030 use rustls::ServerConfig;
1031 use tokio_rustls::TlsAcceptor;
1032
1033 let (cert_path, key_path) = resolve_tls_paths(s)?;
1034
1035 let _ = rustls::crypto::ring::default_provider().install_default();
1037
1038 let resolver: Arc<dyn rustls::server::ResolvesServerCert> = if s.proxy.tls_cert.is_empty() {
1043 if ensure_ca(&cert_path, &key_path, || {
1044 cert_path.exists() && key_path.exists()
1045 })? {
1046 log::info!("Generated local CA certificate at {}", cert_path.display());
1047 log::info!("To trust the CA in your browser, run: pitchfork proxy trust");
1048 }
1049 Arc::new(SniCertResolver::new(
1050 &cert_path,
1051 &key_path,
1052 effective_tld.clone(),
1053 )?)
1054 } else {
1055 log::info!(
1056 "Serving the configured certificate {} (no certificates are minted)",
1057 cert_path.display()
1058 );
1059 Arc::new(StaticCertResolver::new(
1060 &cert_path,
1061 &key_path,
1062 effective_tld.clone(),
1063 )?)
1064 };
1065
1066 let mut tls_config = ServerConfig::builder()
1067 .with_no_client_auth()
1068 .with_cert_resolver(resolver);
1069 tls_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
1072
1073 let acceptor = TlsAcceptor::from(Arc::new(tls_config));
1074
1075 let listener = match TcpListener::bind(addr).await {
1076 Ok(l) => {
1077 if settings().proxy.sync_hosts {
1078 crate::proxy::hosts::sync_hosts_from_settings();
1079 }
1080 let _ = bind_tx.send(Ok(()));
1081 l
1082 }
1083 Err(e) => {
1084 let msg = bind_error_message(effective_port, &e);
1085 let _ = bind_tx.send(Err(msg.clone()));
1086 return Err(miette::miette!("{msg}"));
1087 }
1088 };
1089
1090 log::info!("Proxy server listening on https://{addr} (HTTP also accepted)");
1091 if effective_port < 1024 {
1092 log::info!(
1093 "Note: port {effective_port} is a privileged port. \
1094 The supervisor must be started with sudo to bind to this port."
1095 );
1096 }
1097
1098 let redirect_app = Router::new()
1106 .route(
1107 crate::proxy::pac::PAC_PATH,
1110 axum::routing::any(plain_pac_handler),
1111 )
1112 .fallback(plain_fallback_handler)
1113 .with_state(PlainState {
1114 tld: effective_tld.clone(),
1115 port: effective_port,
1116 proxy: plain_proxy_state.clone(),
1117 });
1118
1119 let mut conn_tasks: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
1121 let handshake_slots = Arc::new(tokio::sync::Semaphore::new(MAX_PENDING_HANDSHAKES));
1122 loop {
1123 while conn_tasks.try_join_next().is_some() {}
1126
1127 tokio::select! {
1128 accept_result = listener.accept() => {
1129 let (stream, peer_addr) = match accept_result {
1130 Ok(conn) => conn,
1131 Err(e) => {
1132 log::warn!("Accept error (will retry): {e}");
1133 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1134 continue;
1135 }
1136 };
1137
1138 let Ok(handshake_permit) = Arc::clone(&handshake_slots).try_acquire_owned() else {
1146 if let Some(suppressed) = REFUSED_HANDSHAKE.allow(REFUSAL_LOG_INTERVAL) {
1149 log::warn!(
1150 "Proxy refused a connection: {MAX_PENDING_HANDSHAKES} \
1151 still negotiating \
1152 ({suppressed} similar refusals since the last message)"
1153 );
1154 }
1155 drop(stream);
1156 continue;
1157 };
1158
1159 let acceptor = acceptor.clone();
1160 let app = app
1165 .clone()
1166 .layer(axum::Extension(axum::extract::ConnectInfo(peer_addr)));
1167 let redirect_app = redirect_app.clone();
1168 let tld = effective_tld.clone();
1169 let cancel = cancel.clone();
1170
1171 conn_tasks.spawn(async move {
1172 let handshake_deadline = tokio::time::Instant::now() + HANDSHAKE_TIMEOUT;
1179 let mut peek_buf = [0u8; 1];
1180 match tokio::time::timeout_at(handshake_deadline, stream.peek(&mut peek_buf)).await {
1181 Ok(Ok(0)) | Ok(Err(_)) => return,
1182 Err(_) => {
1183 if let Some(suppressed) =
1184 ABANDONED_HANDSHAKE.allow(REFUSAL_LOG_INTERVAL)
1185 {
1186 log::debug!(
1187 "Connection sent nothing within the handshake timeout \
1188 ({suppressed} similar since the last message)"
1189 );
1190 }
1191 return;
1192 }
1193 Ok(Ok(_)) => {}
1194 }
1195
1196 if peek_buf[0] == 0x16 {
1197 let sni_budget = SNI_PEEK_TIMEOUT.min(
1205 handshake_deadline.saturating_duration_since(tokio::time::Instant::now()),
1206 );
1207 match peek_sni_host(&stream, sni_budget).await {
1208 SniProbe::Host(host) => {
1209 let Ok(mode) = tokio::time::timeout_at(
1214 handshake_deadline,
1215 resolve_tls_mode(&host, &tld),
1216 )
1217 .await
1218 else {
1219 if let Some(suppressed) =
1220 ABANDONED_HANDSHAKE.allow(REFUSAL_LOG_INTERVAL)
1221 {
1222 log::debug!(
1223 "Routing lookup for '{host}' did not finish within the \
1224 handshake timeout ({suppressed} similar since the \
1225 last message)"
1226 );
1227 }
1228 return;
1229 };
1230 if mode.is_passthrough() {
1231 drop(handshake_permit);
1235 tokio::select! {
1239 _ = serve_passthrough(stream, &host, &tld) => {}
1240 _ = cancel.cancelled() => {}
1241 }
1242 return;
1243 }
1244 }
1245 SniProbe::NoHost => {}
1246 SniProbe::Undetermined => {
1255 log::debug!(
1256 "Could not read the ClientHello of a TLS connection; \
1257 handing it to the TLS acceptor, which refuses to terminate \
1258 a passthrough hostname."
1259 );
1260 let _ = tokio::time::timeout_at(handshake_deadline, async {
1265 let _ = get_cached_slugs().await;
1266 let _ = get_cached_host_registry().await;
1267 })
1268 .await;
1269 }
1270 }
1271
1272 let accepted = match tokio::time::timeout_at(
1274 handshake_deadline,
1275 acceptor.accept(stream),
1276 )
1277 .await
1278 {
1279 Ok(r) => r,
1280 Err(_) => {
1281 if let Some(suppressed) =
1282 ABANDONED_HANDSHAKE.allow(REFUSAL_LOG_INTERVAL)
1283 {
1284 log::debug!(
1285 "TLS handshake did not complete in time \
1286 ({suppressed} similar since the last message)"
1287 );
1288 }
1289 return;
1290 }
1291 };
1292 drop(handshake_permit);
1295 match accepted {
1296 Ok(tls_stream) => {
1297 let io = hyper_util::rt::TokioIo::new(tls_stream);
1298 let svc = hyper_util::service::TowerToHyperService::new(app);
1299 if let Err(e) = serve_conn_until_cancelled!(io, svc, cancel) {
1300 log::debug!("Connection error: {e}");
1303 }
1304 }
1305 Err(e) => {
1306 log::debug!("TLS handshake error: {e}");
1307 }
1308 }
1309 } else {
1310 drop(handshake_permit);
1313 let io = hyper_util::rt::TokioIo::new(stream);
1314 let svc = hyper_util::service::TowerToHyperService::new(redirect_app);
1315 let _ = serve_conn_until_cancelled!(io, svc, cancel);
1316 }
1317 });
1318
1319 while conn_tasks.try_join_next().is_some() {}
1320 }
1321 _ = cancel.cancelled() => {
1322 log::info!("Proxy server shutting down (cancel signal received)");
1323 break;
1324 }
1325 }
1326 }
1327
1328 let deadline = tokio::time::Instant::now() + SHUTDOWN_DRAIN_BUDGET;
1332
1333 let _ = tokio::time::timeout_at(deadline, async {
1335 while conn_tasks.join_next().await.is_some() {}
1336 })
1337 .await;
1338
1339 plain_proxy_state.tunnels.close();
1343 let _ = tokio::time::timeout_at(deadline, plain_proxy_state.tunnels.wait()).await;
1344
1345 Ok(())
1346}
1347
1348#[cfg(feature = "proxy-tls")]
1353const SNI_PEEK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
1354
1355#[cfg(feature = "proxy-tls")]
1360const SNI_PEEK_MAX_BYTES: usize = 16 * 1024;
1361
1362#[cfg(feature = "proxy-tls")]
1368#[derive(Debug, PartialEq, Eq)]
1369enum SniProbe {
1370 Host(String),
1372 NoHost,
1375 Undetermined,
1378}
1379
1380#[cfg(feature = "proxy-tls")]
1392async fn peek_sni_host(stream: &TcpStream, timeout: std::time::Duration) -> SniProbe {
1393 use crate::proxy::sni::{SniPeek, parse_sni};
1394
1395 const MIN_PAUSE: std::time::Duration = std::time::Duration::from_millis(10);
1396 const MAX_PAUSE: std::time::Duration = std::time::Duration::from_millis(200);
1397
1398 let deadline = tokio::time::Instant::now() + timeout;
1399 let mut buf = vec![0u8; 2048];
1400 let mut last_n = 0;
1401 let mut pause = MIN_PAUSE;
1402
1403 loop {
1404 let n = match tokio::time::timeout_at(deadline, stream.peek(&mut buf)).await {
1408 Ok(Ok(0)) => return SniProbe::NoHost,
1412 Ok(Ok(n)) => n,
1413 Ok(Err(e)) => {
1414 log::debug!("Failed to peek at a TLS connection: {e}");
1415 return SniProbe::Undetermined;
1416 }
1417 Err(_elapsed) => {
1418 log::debug!("Timed out waiting for a client that sent no ClientHello");
1419 return SniProbe::Undetermined;
1420 }
1421 };
1422 match parse_sni(&buf[..n]) {
1423 SniPeek::Found(host) => return SniProbe::Host(host),
1424 SniPeek::Absent | SniPeek::NotTls => return SniProbe::NoHost,
1425 SniPeek::Incomplete => {}
1426 }
1427
1428 if n == buf.len() && buf.len() < SNI_PEEK_MAX_BYTES {
1430 buf.resize((buf.len() * 2).min(SNI_PEEK_MAX_BYTES), 0);
1431 continue;
1432 }
1433 if n >= SNI_PEEK_MAX_BYTES {
1434 log::debug!("Giving up on SNI after {n} bytes without a complete ClientHello");
1435 return SniProbe::Undetermined;
1436 }
1437 if tokio::time::Instant::now() >= deadline {
1438 log::debug!("Timed out waiting for a complete ClientHello ({n} bytes read)");
1439 return SniProbe::Undetermined;
1440 }
1441 match stream.ready(tokio::io::Interest::READABLE).await {
1447 Ok(ready) if ready.is_read_closed() => {
1448 log::debug!("Client closed after {n} bytes of an incomplete ClientHello");
1449 return SniProbe::Undetermined;
1450 }
1451 Ok(_) => {}
1452 Err(e) => {
1453 log::debug!("Failed to poll a TLS connection: {e}");
1454 return SniProbe::Undetermined;
1455 }
1456 }
1457 pause = if n > last_n {
1460 MIN_PAUSE
1461 } else {
1462 (pause * 2).min(MAX_PAUSE)
1463 };
1464 last_n = n;
1465 tokio::time::sleep_until((tokio::time::Instant::now() + pause).min(deadline)).await;
1466 }
1467}
1468
1469#[cfg(feature = "proxy-tls")]
1482async fn serve_passthrough(mut stream: TcpStream, host: &str, tld: &str) {
1483 let (port, _activity) = match resolve_passthrough_port(host, tld).await {
1487 Ok(ready) => ready,
1488 Err(msg) => {
1489 log::warn!("TLS passthrough for '{host}' failed: {msg}");
1490 return;
1491 }
1492 };
1493
1494 let addr = SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port));
1495 let mut backend = match connect_backend(addr).await {
1496 Ok(b) => b,
1497 Err(e) => {
1498 log::warn!("TLS passthrough for '{host}': failed to connect to {addr}: {e}");
1499 return;
1500 }
1501 };
1502
1503 log::debug!("TLS passthrough: splicing '{host}' to {addr}");
1504 if let Err(e) = tokio::io::copy_bidirectional(&mut stream, &mut backend).await {
1505 log::debug!("TLS passthrough for '{host}' ended: {e}");
1507 }
1508}
1509
1510#[cfg(feature = "proxy-tls")]
1517const PASSTHROUGH_CONNECT_GRACE: std::time::Duration = std::time::Duration::from_secs(2);
1518
1519#[cfg(feature = "proxy-tls")]
1521async fn connect_backend(addr: SocketAddr) -> std::io::Result<TcpStream> {
1522 let deadline = tokio::time::Instant::now() + PASSTHROUGH_CONNECT_GRACE;
1523 loop {
1524 match TcpStream::connect(addr).await {
1525 Ok(stream) => return Ok(stream),
1526 Err(e) => {
1527 let retryable = e.kind() == std::io::ErrorKind::ConnectionRefused;
1528 if !retryable || tokio::time::Instant::now() >= deadline {
1529 return Err(e);
1530 }
1531 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1532 }
1533 }
1534 }
1535}
1536
1537#[cfg(feature = "proxy-tls")]
1545async fn resolve_passthrough_port(
1546 host: &str,
1547 tld: &str,
1548) -> std::result::Result<(u16, Option<ActivityGuard>), String> {
1549 let budget = settings().proxy_auto_start_timeout();
1550 match tokio::time::timeout(budget, resolve_passthrough_port_inner(host, tld)).await {
1551 Ok(result) => result,
1552 Err(_elapsed) => Err(format!(
1553 "no daemon was ready for '{host}' within proxy.auto_start_timeout ({budget:?})"
1554 )),
1555 }
1556}
1557
1558#[cfg(feature = "proxy-tls")]
1561async fn resolve_passthrough_port_inner(
1562 host: &str,
1563 tld: &str,
1564) -> std::result::Result<(u16, Option<ActivityGuard>), String> {
1565 loop {
1566 match resolve_target(host, tld).await {
1567 ResolveResult::Ready(port, activity, _) => return Ok((port, activity)),
1568 ResolveResult::Starting { slug: _ } => {
1573 tokio::time::sleep(std::time::Duration::from_millis(250)).await;
1574 }
1575 ResolveResult::NotFound => {
1576 return Err(
1577 "no running daemon with a port matched this hostname, and it could not \
1578 be auto-started"
1579 .to_string(),
1580 );
1581 }
1582 ResolveResult::Page {
1587 project, worktree, ..
1588 } => {
1589 return Err(match worktree {
1590 Some(worktree) => {
1591 format!("'{worktree}' of project '{project}' is a stack page, not a daemon")
1592 }
1593 None => format!("'{project}' is a project page, not a daemon"),
1594 });
1595 }
1596 ResolveResult::Unknown { heading, .. } => return Err(heading),
1597 ResolveResult::Error(msg) => return Err(msg),
1598 }
1599 }
1600}
1601
1602#[cfg(not(feature = "proxy-tls"))]
1604#[allow(clippy::too_many_arguments)]
1605async fn serve_https_with_http_fallback(
1606 _app: Router,
1607 _addr: SocketAddr,
1608 _s: &crate::settings::Settings,
1609 _effective_port: u16,
1610 _effective_tld: String,
1611 _plain_proxy_state: ProxyState,
1612 bind_tx: tokio::sync::oneshot::Sender<std::result::Result<(), String>>,
1613 _cancel: tokio_util::sync::CancellationToken,
1614) -> crate::Result<()> {
1615 let msg = "HTTPS proxy support requires the `proxy-tls` feature.\n\
1616 Rebuild pitchfork with: cargo build --features proxy-tls"
1617 .to_string();
1618 let _ = bind_tx.send(Err(msg.clone()));
1619 miette::bail!("{msg}");
1620}
1621
1622#[cfg(feature = "proxy-tls")]
1632fn resolve_tls_paths(
1633 s: &crate::settings::Settings,
1634) -> crate::Result<(std::path::PathBuf, std::path::PathBuf)> {
1635 if let Some(problem) = tls_pair_problem(&s.proxy.tls_cert, &s.proxy.tls_key) {
1636 miette::bail!("{problem}");
1637 }
1638 let proxy_dir = crate::env::PITCHFORK_STATE_DIR.join("proxy");
1639 let resolve = |configured: &str, default: &str| {
1640 if configured.is_empty() {
1641 proxy_dir.join(default)
1642 } else {
1643 std::path::PathBuf::from(configured)
1644 }
1645 };
1646 Ok((
1647 resolve(&s.proxy.tls_cert, "ca.pem"),
1648 resolve(&s.proxy.tls_key, "ca-key.pem"),
1649 ))
1650}
1651
1652pub(crate) fn tls_pair_problem(cert: &str, key: &str) -> Option<String> {
1655 match (cert.is_empty(), key.is_empty()) {
1656 (false, true) => Some(
1657 "proxy.tls_cert is set but proxy.tls_key is empty; set both, or neither to use \
1658 the generated CA"
1659 .to_string(),
1660 ),
1661 (true, false) => Some(
1662 "proxy.tls_key is set but proxy.tls_cert is empty; set both, or neither to use \
1663 the generated CA"
1664 .to_string(),
1665 ),
1666 _ => None,
1667 }
1668}
1669
1670#[cfg(feature = "proxy-tls")]
1680pub fn ensure_ca(
1681 cert_path: &std::path::Path,
1682 key_path: &std::path::Path,
1683 usable: impl FnOnce() -> bool,
1684) -> crate::Result<bool> {
1685 let _lock = xx::fslock::get(cert_path, false)
1686 .map_err(|e| miette::miette!("Failed to lock {}: {e}", cert_path.display()))?;
1687 if usable() {
1688 return Ok(false);
1689 }
1690 generate_ca(cert_path, key_path)?;
1691 clear_host_certs(&host_certs_dir_for(cert_path), None);
1695 Ok(true)
1696}
1697
1698#[cfg(feature = "proxy-tls")]
1700fn host_certs_dir_for(ca_cert_path: &std::path::Path) -> std::path::PathBuf {
1701 ca_cert_path
1702 .parent()
1703 .unwrap_or(std::path::Path::new("."))
1704 .join("host-certs")
1705}
1706
1707#[cfg(feature = "proxy-tls")]
1715fn ca_cache_id(ca_cert_pem: &str) -> String {
1716 let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
1717 for b in ca_cert_pem.bytes() {
1718 hash ^= u64::from(b);
1719 hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
1720 }
1721 format!("ca-{hash:016x}")
1722}
1723
1724#[cfg(feature = "proxy-tls")]
1728fn clear_host_certs(root: &std::path::Path, keep: Option<&str>) {
1729 let Ok(entries) = std::fs::read_dir(root) else {
1730 return;
1731 };
1732 for entry in entries.filter_map(|e| e.ok()) {
1733 let path = entry.path();
1734 let name = entry.file_name();
1735 let result = if path.is_dir() {
1736 let is_ca_dir = name.to_str().is_some_and(|n| n.starts_with("ca-"));
1737 if !is_ca_dir || keep.is_some_and(|k| name == k) {
1738 continue;
1739 }
1740 std::fs::remove_dir_all(&path)
1741 } else if path.extension().is_some_and(|x| x == "pem") {
1742 std::fs::remove_file(&path)
1743 } else {
1744 continue;
1745 };
1746 if let Err(e) = result {
1747 log::debug!(
1748 "Could not remove stale cached certs {}: {e}",
1749 path.display()
1750 );
1751 }
1752 }
1753}
1754
1755#[cfg(feature = "proxy-tls")]
1762fn generate_ca(cert_path: &std::path::Path, key_path: &std::path::Path) -> crate::Result<()> {
1763 use rcgen::{
1764 BasicConstraints, CertificateParams, DistinguishedName, DnType, IsCa, KeyUsagePurpose,
1765 };
1766
1767 if let Some(parent) = cert_path.parent() {
1769 std::fs::create_dir_all(parent)
1770 .map_err(|e| miette::miette!("Failed to create proxy cert directory: {e}"))?;
1771 }
1772
1773 let mut params = CertificateParams::default();
1774 let mut dn = DistinguishedName::new();
1775 dn.push(DnType::CommonName, "Pitchfork Local CA");
1776 dn.push(DnType::OrganizationName, "Pitchfork");
1777 params.distinguished_name = dn;
1778 params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
1779 params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign];
1780
1781 let key_pair = rcgen::KeyPair::generate()
1782 .map_err(|e| miette::miette!("Failed to generate CA key pair: {e}"))?;
1783 let ca_cert = params
1784 .self_signed(&key_pair)
1785 .map_err(|e| miette::miette!("Failed to self-sign CA certificate: {e}"))?;
1786
1787 std::fs::write(cert_path, ca_cert.pem()).map_err(|e| {
1789 miette::miette!(
1790 "Failed to write CA certificate to {}: {e}",
1791 cert_path.display()
1792 )
1793 })?;
1794
1795 {
1799 #[cfg(unix)]
1800 {
1801 use std::io::Write;
1802 use std::os::unix::fs::OpenOptionsExt;
1803 std::fs::OpenOptions::new()
1804 .write(true)
1805 .create(true)
1806 .truncate(true)
1807 .mode(0o600)
1808 .open(key_path)
1809 .and_then(|mut f| f.write_all(key_pair.serialize_pem().as_bytes()))
1810 .map_err(|e| {
1811 miette::miette!("Failed to write CA key to {}: {e}", key_path.display())
1812 })?;
1813 }
1814 #[cfg(not(unix))]
1815 {
1816 std::fs::write(key_path, key_pair.serialize_pem()).map_err(|e| {
1817 miette::miette!("Failed to write CA key to {}: {e}", key_path.display())
1818 })?;
1819 log::debug!(
1820 "CA private key written to {} (file permissions are not restricted \
1821 on non-Unix platforms — consider restricting access manually)",
1822 key_path.display()
1823 );
1824 }
1825 }
1826
1827 Ok(())
1828}
1829
1830#[cfg(feature = "proxy-tls")]
1836struct StaticCertResolver {
1837 certified: Arc<rustls::sign::CertifiedKey>,
1838 tld: String,
1841}
1842
1843#[cfg(feature = "proxy-tls")]
1844impl std::fmt::Debug for StaticCertResolver {
1845 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1846 f.debug_struct("StaticCertResolver").finish_non_exhaustive()
1847 }
1848}
1849
1850#[cfg(feature = "proxy-tls")]
1851impl StaticCertResolver {
1852 fn new(
1853 cert_path: &std::path::Path,
1854 key_path: &std::path::Path,
1855 tld: String,
1856 ) -> crate::Result<Self> {
1857 use rustls::pki_types::CertificateDer;
1858 use rustls_pemfile::{certs, private_key};
1859
1860 let cert_pem = std::fs::read(cert_path).map_err(|e| {
1861 miette::miette!("Failed to read proxy.tls_cert {}: {e}", cert_path.display())
1862 })?;
1863 let key_pem = std::fs::read(key_path).map_err(|e| {
1864 miette::miette!("Failed to read proxy.tls_key {}: {e}", key_path.display())
1865 })?;
1866
1867 let cert_ders: Vec<CertificateDer<'static>> = certs(&mut cert_pem.as_slice())
1868 .collect::<Result<Vec<_>, _>>()
1869 .map_err(|e| miette::miette!("Failed to parse {}: {e}", cert_path.display()))?;
1870 if cert_ders.is_empty() {
1871 miette::bail!("No certificates found in {}", cert_path.display());
1872 }
1873
1874 let key_der = private_key(&mut key_pem.as_slice())
1875 .map_err(|e| miette::miette!("Failed to parse {}: {e}", key_path.display()))?
1876 .ok_or_else(|| miette::miette!("No private key found in {}", key_path.display()))?;
1877 let signing_key = rustls::crypto::ring::sign::any_supported_type(&key_der)
1878 .map_err(|e| miette::miette!("Failed to use the configured private key: {e}"))?;
1879
1880 let certified = rustls::sign::CertifiedKey::new(cert_ders, signing_key);
1881 certified.keys_match().map_err(|e| {
1885 miette::miette!(
1886 "proxy.tls_key {} does not match proxy.tls_cert {}: {e}",
1887 key_path.display(),
1888 cert_path.display()
1889 )
1890 })?;
1891
1892 Ok(Self {
1893 certified: Arc::new(certified),
1894 tld,
1895 })
1896 }
1897}
1898
1899#[cfg(feature = "proxy-tls")]
1900impl rustls::server::ResolvesServerCert for StaticCertResolver {
1901 fn resolve(
1902 &self,
1903 client_hello: rustls::server::ClientHello<'_>,
1904 ) -> Option<Arc<rustls::sign::CertifiedKey>> {
1905 if let Some(domain) = client_hello.server_name()
1909 && resolve_tls_mode_in(domain, &self.tld, &slug_snapshot(), ®istry_snapshot())
1910 .is_passthrough()
1911 {
1912 log::warn!(
1913 "Refusing to terminate TLS for '{domain}', which is configured for \
1914 proxy_tls = \"passthrough\": its ClientHello could not be inspected before the \
1915 handshake, so the stream could not be spliced to the daemon."
1916 );
1917 return None;
1918 }
1919 Some(Arc::clone(&self.certified))
1920 }
1921}
1922
1923#[cfg(feature = "proxy-tls")]
1943struct SniCertResolver {
1944 issuer: rcgen::Issuer<'static, rcgen::KeyPair>,
1946 tld: String,
1950 host_certs_dir: std::path::PathBuf,
1952 cache: std::sync::Mutex<CertCache>,
1955 pending: std::sync::Mutex<std::collections::HashSet<String>>,
1959 pending_cv: std::sync::Condvar,
1961}
1962
1963#[cfg(feature = "proxy-tls")]
1969fn prune_host_certs(dir: &std::path::Path) {
1970 let Ok(entries) = std::fs::read_dir(dir) else {
1971 return;
1972 };
1973 let mut files: Vec<(std::time::SystemTime, std::path::PathBuf)> = entries
1974 .filter_map(|e| e.ok())
1975 .filter(|e| e.path().extension().is_some_and(|x| x == "pem"))
1976 .filter_map(|e| {
1977 let modified = e.metadata().and_then(|m| m.modified()).ok()?;
1978 Some((modified, e.path()))
1979 })
1980 .collect();
1981 if files.len() <= MAX_HOST_CERTS {
1982 return;
1983 }
1984 files.sort_by_key(|(t, _)| *t);
1985 let excess = files.len() - MAX_HOST_CERTS;
1986 for (_, path) in files.into_iter().take(excess) {
1987 if let Err(e) = std::fs::remove_file(&path) {
1988 log::debug!("Could not prune cached cert {}: {e}", path.display());
1989 }
1990 }
1991}
1992
1993#[cfg(feature = "proxy-tls")]
1995#[derive(Default)]
1996struct CertCache {
1997 by_domain: std::collections::HashMap<String, Arc<rustls::sign::CertifiedKey>>,
1998 order: std::collections::VecDeque<String>,
2000}
2001
2002#[cfg(feature = "proxy-tls")]
2003impl CertCache {
2004 fn get(&self, domain: &str) -> Option<&Arc<rustls::sign::CertifiedKey>> {
2005 self.by_domain.get(domain)
2006 }
2007
2008 fn insert(&mut self, domain: String, key: Arc<rustls::sign::CertifiedKey>) -> Option<String> {
2012 if self.by_domain.insert(domain.clone(), key).is_none() {
2013 self.order.push_back(domain);
2014 }
2015 if self.by_domain.len() <= MAX_HOST_CERTS {
2016 return None;
2017 }
2018 let evicted = self.order.pop_front()?;
2019 self.by_domain.remove(&evicted);
2020 Some(evicted)
2021 }
2022}
2023
2024#[cfg(feature = "proxy-tls")]
2025impl std::fmt::Debug for SniCertResolver {
2026 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2027 f.debug_struct("SniCertResolver").finish_non_exhaustive()
2028 }
2029}
2030
2031#[cfg(feature = "proxy-tls")]
2032impl SniCertResolver {
2033 fn new(
2035 ca_cert_path: &std::path::Path,
2036 ca_key_path: &std::path::Path,
2037 tld: String,
2038 ) -> crate::Result<Self> {
2039 let ca_key_pem = std::fs::read_to_string(ca_key_path)
2040 .map_err(|e| miette::miette!("Failed to read CA key {}: {e}", ca_key_path.display()))?;
2041 let ca_cert_pem = std::fs::read_to_string(ca_cert_path).map_err(|e| {
2042 miette::miette!("Failed to read CA cert {}: {e}", ca_cert_path.display())
2043 })?;
2044
2045 if !ca_cert_pem.contains("BEGIN CERTIFICATE") {
2047 miette::bail!("CA cert file does not contain a valid PEM certificate");
2048 }
2049
2050 let ca_key = rcgen::KeyPair::from_pem(&ca_key_pem)
2051 .map_err(|e| miette::miette!("Failed to parse CA key: {e}"))?;
2052
2053 let issuer = rcgen::Issuer::from_ca_cert_pem(&ca_cert_pem, ca_key)
2055 .map_err(|e| miette::miette!("Failed to parse CA cert: {e}"))?;
2056
2057 let host_certs_root = host_certs_dir_for(ca_cert_path);
2060 let cache_id = ca_cache_id(&ca_cert_pem);
2061 clear_host_certs(&host_certs_root, Some(&cache_id));
2062 let host_certs_dir = host_certs_root.join(&cache_id);
2063 std::fs::create_dir_all(&host_certs_dir)
2064 .map_err(|e| miette::miette!("Failed to create host-certs dir: {e}"))?;
2065 prune_host_certs(&host_certs_dir);
2069
2070 Ok(Self {
2071 issuer,
2072 tld,
2073 host_certs_dir,
2074 cache: std::sync::Mutex::new(CertCache::default()),
2075 pending: std::sync::Mutex::new(std::collections::HashSet::new()),
2076 pending_cv: std::sync::Condvar::new(),
2077 })
2078 }
2079
2080 fn get_or_create_checked(&self, domain: &str) -> Option<Arc<rustls::sign::CertifiedKey>> {
2089 if !crate::proxy::owns_name(&self.tld, domain) {
2090 if let Some(suppressed) = REFUSED_SNI.allow(REFUSAL_LOG_INTERVAL) {
2093 log::warn!(
2094 "Refusing to issue a certificate for {domain:?}: \
2095 the pitchfork CA only signs names under .{} \
2096 ({suppressed} similar refusals since the last message)",
2097 self.tld
2098 );
2099 }
2100 return None;
2101 }
2102 self.get_or_create(domain)
2103 }
2104
2105 fn get_or_create(&self, domain: &str) -> Option<Arc<rustls::sign::CertifiedKey>> {
2128 {
2130 let cache = self.cache.lock().ok()?;
2131 if let Some(ck) = cache.get(domain) {
2132 return Some(Arc::clone(ck));
2133 }
2134 } loop {
2147 {
2148 let mut pending = self.pending.lock().ok()?;
2149 if pending.contains(domain) {
2150 pending = self.pending_cv.wait(pending).ok()?;
2152 drop(pending);
2154 } else {
2155 pending.insert(domain.to_string());
2157 break;
2158 }
2159 } {
2165 let cache = self.cache.lock().ok()?;
2166 if let Some(ck) = cache.get(domain) {
2167 return Some(Arc::clone(ck));
2168 }
2169 } } let result = self.get_or_create_inner(domain);
2173
2174 {
2180 let mut pending = match self.pending.lock() {
2181 Ok(g) => g,
2182 Err(e) => e.into_inner(),
2183 };
2184 pending.remove(domain);
2185 self.pending_cv.notify_all();
2186 }
2187
2188 result
2189 }
2190
2191 fn get_or_create_inner(&self, domain: &str) -> Option<Arc<rustls::sign::CertifiedKey>> {
2193 let disk_path = self.disk_path(domain);
2194
2195 if disk_path.exists() {
2197 if let Ok(ck) = self.load_from_disk(&disk_path) {
2198 let ck = Arc::new(ck);
2199 self.remember(domain, &ck);
2200 return Some(ck);
2201 }
2202 let _ = std::fs::remove_file(&disk_path);
2204 }
2205
2206 let ck = self.sign_for_domain(domain).ok()?;
2208
2209 let ck = Arc::new(ck);
2210 self.remember(domain, &ck);
2211 Some(ck)
2212 }
2213
2214 fn remember(&self, domain: &str, ck: &Arc<rustls::sign::CertifiedKey>) {
2219 let evicted = match self.cache.lock() {
2220 Ok(mut cache) => cache.insert(domain.to_string(), Arc::clone(ck)),
2221 Err(_) => return,
2222 };
2223 if let Some(evicted) = evicted {
2224 let path = self.disk_path(&evicted);
2225 if let Err(e) = std::fs::remove_file(&path)
2226 && e.kind() != std::io::ErrorKind::NotFound
2227 {
2228 log::debug!("Could not evict cached cert {}: {e}", path.display());
2229 }
2230 }
2231 }
2232
2233 fn disk_path(&self, domain: &str) -> std::path::PathBuf {
2235 self.host_certs_dir
2236 .join(format!("{}.pem", cert_cache_file_stem(domain)))
2237 }
2238
2239 fn load_from_disk(&self, path: &std::path::Path) -> crate::Result<rustls::sign::CertifiedKey> {
2244 use rustls::pki_types::CertificateDer;
2245 use rustls_pemfile::{certs, private_key};
2246
2247 let pem = std::fs::read_to_string(path)
2248 .map_err(|e| miette::miette!("Failed to read disk cert {}: {e}", path.display()))?;
2249
2250 let cert_ders: Vec<CertificateDer<'static>> = certs(&mut pem.as_bytes())
2251 .collect::<Result<Vec<_>, _>>()
2252 .map_err(|e| miette::miette!("Failed to parse certs from {}: {e}", path.display()))?;
2253
2254 if cert_ders.is_empty() {
2255 miette::bail!("No certificates found in {}", path.display());
2256 }
2257
2258 {
2260 let (_, cert) = x509_parser::parse_x509_certificate(&cert_ders[0]).map_err(|e| {
2261 miette::miette!("Failed to parse certificate from {}: {e}", path.display())
2262 })?;
2263 use chrono::Utc;
2264 let now_ts = Utc::now().timestamp();
2265 let not_after_ts = cert.validity().not_after.timestamp();
2266 if not_after_ts < now_ts {
2267 miette::bail!(
2268 "Cached certificate at {} has expired — will regenerate",
2269 path.display()
2270 );
2271 }
2272 }
2273
2274 let key_der = private_key(&mut pem.as_bytes())
2275 .map_err(|e| miette::miette!("Failed to parse key from {}: {e}", path.display()))?
2276 .ok_or_else(|| miette::miette!("No private key found in {}", path.display()))?;
2277
2278 let signing_key = rustls::crypto::ring::sign::any_supported_type(&key_der)
2279 .map_err(|e| miette::miette!("Failed to create signing key from disk: {e}"))?;
2280
2281 Ok(rustls::sign::CertifiedKey::new(cert_ders, signing_key))
2282 }
2283
2284 fn sign_for_domain(&self, domain: &str) -> crate::Result<rustls::sign::CertifiedKey> {
2292 use rcgen::date_time_ymd;
2293 use rcgen::{CertificateParams, DistinguishedName, DnType, SanType};
2294 use rustls::pki_types::CertificateDer;
2295 use rustls_pemfile::private_key;
2296
2297 let mut params = CertificateParams::default();
2298 let mut dn = DistinguishedName::new();
2299 dn.push(DnType::CommonName, domain);
2300 params.distinguished_name = dn;
2301
2302 {
2304 use chrono::{Datelike, Duration, Utc};
2305 let yesterday = Utc::now() - Duration::days(1);
2306 let expiry = Utc::now() + Duration::days(397);
2309 params.not_before = date_time_ymd(
2310 yesterday.year(),
2311 yesterday.month() as u8,
2312 yesterday.day() as u8,
2313 );
2314 params.not_after =
2315 date_time_ymd(expiry.year(), expiry.month() as u8, expiry.day() as u8);
2316 }
2317
2318 let mut sans =
2320 vec![SanType::DnsName(domain.to_string().try_into().map_err(
2321 |e| miette::miette!("Invalid domain name '{domain}': {e}"),
2322 )?)];
2323 if let Some(dot_pos) = domain.find('.') {
2330 let parent = &domain[dot_pos + 1..];
2331 if crate::proxy::is_strictly_under_tld(&self.tld, parent) {
2332 let wildcard = format!("*.{parent}");
2333 if let Ok(wc) = wildcard.try_into() {
2334 sans.push(SanType::DnsName(wc));
2335 }
2336 }
2337 }
2338 params.subject_alt_names = sans;
2339
2340 let leaf_key = rcgen::KeyPair::generate()
2341 .map_err(|e| miette::miette!("Failed to generate leaf key: {e}"))?;
2342 let leaf_cert = params
2343 .signed_by(&leaf_key, &self.issuer)
2344 .map_err(|e| miette::miette!("Failed to sign leaf cert for '{domain}': {e}"))?;
2345
2346 let cert_der = CertificateDer::from(leaf_cert.der().to_vec());
2348 let key_pem = leaf_key.serialize_pem();
2349 let key_der = private_key(&mut key_pem.as_bytes())
2350 .map_err(|e| miette::miette!("Failed to parse leaf key PEM: {e}"))?
2351 .ok_or_else(|| miette::miette!("No private key found in generated PEM"))?;
2352
2353 let signing_key = rustls::crypto::ring::sign::any_supported_type(&key_der)
2354 .map_err(|e| miette::miette!("Failed to create signing key: {e}"))?;
2355
2356 let disk_path = self.disk_path(domain);
2359 let combined_pem = format!("{}{}", leaf_cert.pem(), key_pem);
2360 {
2361 #[cfg(unix)]
2362 {
2363 use std::io::Write;
2364 use std::os::unix::fs::OpenOptionsExt;
2365 if let Err(e) = std::fs::OpenOptions::new()
2366 .write(true)
2367 .create(true)
2368 .truncate(true)
2369 .mode(0o600)
2370 .open(&disk_path)
2371 .and_then(|mut f| f.write_all(combined_pem.as_bytes()))
2372 {
2373 log::warn!(
2374 "Failed to persist cert for '{domain}' to {}: {e}",
2375 disk_path.display()
2376 );
2377 }
2378 }
2379 #[cfg(not(unix))]
2380 {
2381 if let Err(e) = std::fs::write(&disk_path, combined_pem) {
2382 log::warn!(
2383 "Failed to persist cert for '{domain}' to {}: {e}",
2384 disk_path.display()
2385 );
2386 } else {
2387 log::debug!(
2388 "Leaf cert for '{domain}' written to {} (file permissions are not \
2389 restricted on non-Unix platforms — consider restricting access manually)",
2390 disk_path.display()
2391 );
2392 }
2393 }
2394 }
2395
2396 Ok(rustls::sign::CertifiedKey::new(vec![cert_der], signing_key))
2397 }
2398}
2399
2400#[cfg(feature = "proxy-tls")]
2401impl rustls::server::ResolvesServerCert for SniCertResolver {
2402 fn resolve(
2403 &self,
2404 client_hello: rustls::server::ClientHello<'_>,
2405 ) -> Option<Arc<rustls::sign::CertifiedKey>> {
2406 let domain = client_hello.server_name()?;
2407 if resolve_tls_mode_in(domain, &self.tld, &slug_snapshot(), ®istry_snapshot())
2414 .is_passthrough()
2415 {
2416 log::warn!(
2417 "Refusing to terminate TLS for '{domain}', which is configured for \
2418 proxy_tls = \"passthrough\": its ClientHello could not be inspected before the \
2419 handshake, so the stream could not be spliced to the daemon."
2420 );
2421 return None;
2422 }
2423
2424 self.get_or_create_checked(domain)
2433 }
2434}
2435
2436#[cfg(feature = "proxy-tls")]
2442pub(crate) fn ca_pair_problem(cert: &std::path::Path, key: &std::path::Path) -> Option<String> {
2443 use rcgen::PublicKeyData;
2444
2445 let cert_pem = match std::fs::read_to_string(cert) {
2446 Ok(p) => p,
2447 Err(e) => return Some(format!("cannot read {}: {e}", cert.display())),
2448 };
2449 let key_pem = match std::fs::read_to_string(key) {
2450 Ok(p) => p,
2451 Err(e) => return Some(format!("cannot read {}: {e}", key.display())),
2452 };
2453 let key_pair = match rcgen::KeyPair::from_pem(&key_pem) {
2454 Ok(k) => k,
2455 Err(e) => return Some(format!("cannot parse {}: {e}", key.display())),
2456 };
2457 let Some(Ok(der)) = rustls_pemfile::certs(&mut cert_pem.as_bytes()).next() else {
2458 return Some(format!("no certificate in {}", cert.display()));
2459 };
2460 let parsed = match x509_parser::parse_x509_certificate(&der) {
2461 Ok((_, c)) => c,
2462 Err(e) => return Some(format!("cannot parse {}: {e}", cert.display())),
2463 };
2464 if parsed.public_key().subject_public_key.data.as_ref() != key_pair.der_bytes() {
2465 return Some(format!(
2466 "{} is not the key for {}",
2467 key.display(),
2468 cert.display()
2469 ));
2470 }
2471 if let Err(e) = rcgen::Issuer::from_ca_cert_pem(&cert_pem, key_pair) {
2472 return Some(format!("{} cannot sign certificates: {e}", cert.display()));
2473 }
2474 None
2475}
2476
2477#[cfg(feature = "proxy-tls")]
2484fn cert_cache_file_stem(domain: &str) -> String {
2485 let mut out = String::with_capacity(domain.len());
2486 for b in domain.bytes() {
2487 if b.is_ascii_alphanumeric() || b == b'-' || b == b'.' {
2488 out.push(char::from(b));
2489 } else {
2490 out.push_str(&format!("%{b:02X}"));
2491 }
2492 }
2493 out
2494}
2495
2496#[derive(Clone)]
2498struct PlainState {
2499 tld: String,
2500 port: u16,
2501 proxy: ProxyState,
2504}
2505
2506async fn plain_fallback_handler(State(state): State<PlainState>, req: Request) -> Response {
2511 if req.method() == axum::http::Method::CONNECT {
2512 let raw_host = get_request_host(&req).unwrap_or_default();
2513 return connect_handler(&state.proxy, req, &raw_host).await;
2514 }
2515 redirect_to_https_handler(req).await
2516}
2517
2518fn reject_non_read(method: &axum::http::Method) -> Option<Response> {
2525 if matches!(*method, axum::http::Method::GET | axum::http::Method::HEAD) {
2526 return None;
2527 }
2528 let mut res = error_response(
2529 StatusCode::METHOD_NOT_ALLOWED,
2530 "the proxy auto-config file is read-only\n",
2531 );
2532 res.headers_mut().insert(
2533 axum::http::header::ALLOW,
2534 HeaderValue::from_static("GET, HEAD"),
2535 );
2536 Some(res)
2537}
2538
2539fn pac_response(tld: &str, host: &str, port: u16) -> Response {
2540 match crate::proxy::pac::generate(tld, host, port) {
2541 Ok(body) => (
2542 StatusCode::OK,
2543 [(
2544 axum::http::header::CONTENT_TYPE,
2545 "application/x-ns-proxy-autoconfig",
2546 )],
2547 body,
2548 )
2549 .into_response(),
2550 Err(e) => error_response(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
2551 }
2552}
2553
2554async fn pac_handler(State(state): State<ProxyState>, req: Request) -> Response {
2561 let host = get_request_host(&req).unwrap_or_default();
2562 let bare = host.split(':').next().unwrap_or("");
2563 if !bare.is_empty() && crate::proxy::is_strictly_under_tld(&state.tld, bare) {
2564 return proxy_handler(State(state), req).await;
2565 }
2566 if let Some(deny) = reject_non_read(req.method()) {
2569 return deny;
2570 }
2571 let port = req
2572 .uri()
2573 .authority()
2574 .and_then(|a| a.port_u16())
2575 .or_else(|| host.rsplit(':').next().and_then(|p| p.parse().ok()))
2576 .unwrap_or(if state.is_tls { 443 } else { 80 });
2577 pac_response(&state.tld, &url_host(state.contact_ip), port)
2578}
2579
2580async fn plain_pac_handler(State(state): State<PlainState>, req: Request) -> Response {
2582 let host = get_request_host(&req).unwrap_or_default();
2586 let bare = host.split(':').next().unwrap_or("");
2587 if !bare.is_empty() && crate::proxy::is_strictly_under_tld(&state.tld, bare) {
2588 return redirect_to_https_handler(req).await;
2589 }
2590 if let Some(deny) = reject_non_read(req.method()) {
2591 return deny;
2592 }
2593 pac_response(&state.tld, &url_host(state.proxy.contact_ip), state.port)
2594}
2595
2596fn local_contact_ip(bind_ip: std::net::IpAddr) -> std::net::IpAddr {
2602 match bind_ip {
2603 std::net::IpAddr::V4(ip) if ip.is_unspecified() => {
2604 std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)
2605 }
2606 std::net::IpAddr::V6(ip) if ip.is_unspecified() => {
2607 std::net::IpAddr::V6(std::net::Ipv6Addr::LOCALHOST)
2608 }
2609 ip => ip,
2610 }
2611}
2612
2613fn url_host(ip: std::net::IpAddr) -> String {
2615 match ip {
2616 std::net::IpAddr::V6(ip) => format!("[{ip}]"),
2617 std::net::IpAddr::V4(ip) => ip.to_string(),
2618 }
2619}
2620
2621fn connect_port(authority: &str) -> Option<u16> {
2623 let (host, port) = authority.rsplit_once(':')?;
2624 if host.contains(':') && !host.ends_with(']') {
2626 return None;
2627 }
2628 port.parse().ok()
2629}
2630
2631async fn connect_handler(state: &ProxyState, req: Request, raw_host: &str) -> Response {
2646 let authority = req
2647 .uri()
2648 .authority()
2649 .map(|a| a.as_str().to_string())
2650 .unwrap_or_else(|| raw_host.to_string());
2651 let host = authority
2652 .rsplit_once(':')
2653 .map(|(h, _)| h)
2654 .unwrap_or(&authority)
2655 .trim_start_matches('[')
2656 .trim_end_matches(']')
2657 .to_string();
2658
2659 if !crate::proxy::owns_name(&state.tld, &host) {
2662 return error_response(
2663 StatusCode::FORBIDDEN,
2664 &format!(
2665 "pitchfork only tunnels CONNECT for names under .{} — refusing {host}",
2666 state.tld
2667 ),
2668 );
2669 }
2670
2671 if !state.is_tls && connect_port(&authority) == Some(443) {
2676 return error_response(
2677 StatusCode::BAD_GATEWAY,
2678 &format!(
2679 "proxy.https is false, so pitchfork cannot serve https://{host}; \
2680 use http:// or enable proxy.https"
2681 ),
2682 );
2683 }
2684
2685 let Some(target) = state.connect_target else {
2686 return error_response(
2687 StatusCode::SERVICE_UNAVAILABLE,
2688 "The proxy listener address is unknown, so CONNECT cannot be tunnelled",
2689 );
2690 };
2691
2692 if state.cancel.is_cancelled() {
2695 return error_response(
2696 StatusCode::SERVICE_UNAVAILABLE,
2697 "The proxy is shutting down",
2698 );
2699 }
2700
2701 let Ok(permit) = Arc::clone(&state.tunnel_slots).try_acquire_owned() else {
2704 if let Some(suppressed) = REFUSED_TUNNEL.allow(REFUSAL_LOG_INTERVAL) {
2705 log::warn!(
2706 "Proxy refused a CONNECT tunnel: {MAX_TUNNELS} already open \
2707 ({suppressed} similar refusals since the last message)"
2708 );
2709 }
2710 return error_response(
2711 StatusCode::SERVICE_UNAVAILABLE,
2712 "Too many CONNECT tunnels are open",
2713 );
2714 };
2715 let cancel = state.cancel.clone();
2716
2717 state.tunnels.spawn(async move {
2723 let _permit = permit;
2725
2726 macro_rules! setup_step {
2739 ($what:literal, $fut:expr) => {
2740 tokio::select! {
2741 r = tokio::time::timeout(HANDSHAKE_TIMEOUT, $fut) => match r {
2742 Ok(Ok(v)) => v,
2743 Ok(Err(e)) => {
2744 if let Some(n) = ABANDONED_TUNNEL.allow(REFUSAL_LOG_INTERVAL) {
2745 log::debug!(
2746 concat!(
2747 "CONNECT {} for {target} failed: {e}",
2748 " ({n} similar since the last message)"
2749 ),
2750 $what,
2751 target = target,
2752 e = e,
2753 n = n
2754 );
2755 }
2756 return;
2757 }
2758 Err(_) => {
2759 if let Some(n) = ABANDONED_TUNNEL.allow(REFUSAL_LOG_INTERVAL) {
2760 log::debug!(
2761 concat!(
2762 "CONNECT {} for {target} did not finish within",
2763 " {timeout:?}",
2764 " ({n} similar since the last message)"
2765 ),
2766 $what,
2767 target = target,
2768 timeout = HANDSHAKE_TIMEOUT,
2769 n = n
2770 );
2771 }
2772 return;
2773 }
2774 },
2775 _ = cancel.cancelled() => {
2778 log::debug!("CONNECT {} for {target} abandoned by shutdown", $what);
2779 return;
2780 }
2781 }
2782 };
2783 }
2784
2785 let upgraded = setup_step!("upgrade", hyper::upgrade::on(req));
2786 let mut client = hyper_util::rt::TokioIo::new(upgraded);
2787 let mut server = setup_step!("connect", tokio::net::TcpStream::connect(target));
2788
2789 tokio::select! {
2794 r = tokio::io::copy_bidirectional(&mut client, &mut server) => {
2795 if let Err(e) = r
2798 && let Some(n) = ABANDONED_TUNNEL.allow(REFUSAL_LOG_INTERVAL)
2799 {
2800 log::debug!(
2801 "CONNECT tunnel to {target} ended: {e} ({n} similar since the last message)"
2802 );
2803 }
2804 }
2805 _ = cancel.cancelled() => {
2806 log::debug!("CONNECT tunnel to {target} closed by shutdown");
2807 }
2808 }
2809 });
2810
2811 Response::builder()
2812 .status(StatusCode::OK)
2813 .header(PITCHFORK_HEADER, "1")
2814 .body(Body::empty())
2815 .unwrap_or_else(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response())
2816}
2817
2818fn get_request_host(req: &Request) -> Option<String> {
2824 let authority = req
2826 .uri()
2827 .authority()
2828 .map(|a| a.as_str().to_string())
2829 .filter(|s| !s.is_empty());
2830
2831 authority.or_else(|| {
2832 req.headers()
2833 .get(HOST)
2834 .and_then(|h| h.to_str().ok())
2835 .map(str::to_string)
2836 })
2837}
2838
2839fn join_cookie_fields(headers: &mut HeaderMap) {
2845 let fields: Vec<&[u8]> = headers
2846 .get_all(COOKIE)
2847 .iter()
2848 .map(HeaderValue::as_bytes)
2849 .collect();
2850 if fields.len() < 2 {
2851 return;
2852 }
2853
2854 let joined = HeaderValue::from_bytes(&fields.join(b"; ".as_slice()))
2855 .expect("valid header values joined with \"; \" form a valid header value");
2856 headers.insert(COOKIE, joined);
2857}
2858
2859fn inject_forwarded_headers(req: &mut Request, is_tls: bool, host_header: &str) {
2870 let remote_addr = req
2871 .extensions()
2872 .get::<axum::extract::ConnectInfo<SocketAddr>>()
2873 .map(|ci| ci.0.ip().to_string())
2874 .unwrap_or_else(|| "127.0.0.1".to_string());
2875
2876 let proto = if is_tls { "https" } else { "http" };
2877 let default_port = if is_tls { "443" } else { "80" };
2878
2879 let forwarded_for = remote_addr.clone();
2882 let forwarded_proto = proto.to_string();
2883 let forwarded_host = host_header.to_string();
2884 let forwarded_port = host_header
2885 .rsplit_once(':')
2886 .map(|(_, port)| port.to_string())
2887 .unwrap_or_else(|| default_port.to_string());
2888
2889 for name in [
2895 "x-forwarded-for",
2896 "x-forwarded-proto",
2897 "x-forwarded-host",
2898 "x-forwarded-port",
2899 "forwarded",
2900 ] {
2901 if let Ok(header_name) = axum::http::HeaderName::from_bytes(name.as_bytes()) {
2902 req.headers_mut().remove(&header_name);
2903 }
2904 }
2905
2906 let headers = [
2907 ("x-forwarded-for", forwarded_for),
2908 ("x-forwarded-proto", forwarded_proto),
2909 ("x-forwarded-host", forwarded_host),
2910 ("x-forwarded-port", forwarded_port),
2911 ];
2912
2913 for (name, value) in headers {
2914 if let Ok(v) = HeaderValue::from_str(&value) {
2915 let header_name = axum::http::HeaderName::from_static(name);
2916 req.headers_mut().insert(header_name, v);
2917 }
2918 }
2919}
2920
2921async fn proxy_handler(State(state): State<ProxyState>, mut req: Request) -> Response {
2926 let Some(raw_host) = get_request_host(&req) else {
2928 return error_response(StatusCode::BAD_REQUEST, "Missing Host header");
2929 };
2930
2931 if req.method() == axum::http::Method::CONNECT {
2935 return connect_handler(&state, req, &raw_host).await;
2936 }
2937 let host = if raw_host.starts_with('[') {
2941 raw_host
2943 .split("]:")
2944 .next()
2945 .unwrap_or(&raw_host)
2946 .trim_start_matches('[')
2947 .trim_end_matches(']')
2948 .to_string()
2949 } else {
2950 raw_host.split(':').next().unwrap_or(&raw_host).to_string()
2952 };
2953 let host = host.trim_end_matches('.').to_string();
2958
2959 let is_from_pitchfork = req.headers().contains_key(PROXY_HOPS_HEADER);
2970 let hops: u64 = if is_from_pitchfork {
2971 req.headers()
2972 .get(PROXY_HOPS_HEADER)
2973 .and_then(|v| v.to_str().ok())
2974 .and_then(|s| s.parse().ok())
2975 .unwrap_or(0)
2976 } else {
2977 0
2979 };
2980 if hops >= MAX_PROXY_HOPS {
2981 return error_response(
2982 StatusCode::LOOP_DETECTED,
2983 &format!(
2984 "Loop detected for '{host}': request has passed through the proxy {hops} times.\n\
2985 This usually means a backend is proxying back through pitchfork without rewriting \n\
2986 the Host header. If you use Vite/webpack proxy, set changeOrigin: true."
2987 ),
2988 );
2989 }
2990
2991 let local_client = is_local_client(&req);
2992
2993 let target_port = if let Some(subdomain) = strip_tld(&host, &state.tld) {
2995 if subdomain.eq_ignore_ascii_case("pitchfork") {
2996 crate::web::port()
2997 } else {
2998 None
2999 }
3000 } else {
3001 None
3002 };
3003
3004 let (target_port, activity, protocol) = if let Some(port) = target_port {
3005 (port, None, ProxyProtocol::Http1)
3006 } else {
3007 if resolve_tls_mode(&host, &state.tld).await.is_passthrough() {
3011 return error_response(
3012 StatusCode::BAD_GATEWAY,
3013 &passthrough_unroutable_message(&host, state.is_tls),
3014 );
3015 }
3016 match resolve_target(&host, &state.tld).await {
3017 ResolveResult::Ready(port, activity, protocol) => (port, activity, protocol),
3018 ResolveResult::Starting { slug } => {
3019 return starting_html_response(&slug, &raw_host);
3020 }
3021 ResolveResult::Page {
3022 project,
3023 worktree,
3024 daemons,
3025 dir,
3026 } => {
3027 if !local_client {
3031 return unknown_host_response(&host, "Not found", &[]);
3032 }
3033 if let Some(base) = crate::web::url()
3040 && let Some(resolved) = dir.clone()
3041 && let Some(path) = tokio::task::spawn_blocking(move || {
3042 crate::web::routes::api::projects::page_path_for_dir(&resolved)
3043 })
3044 .await
3045 .ok()
3046 .flatten()
3047 {
3048 return page_redirect_response(&base, &path);
3049 }
3050 return page_placeholder_response(
3051 &project,
3052 worktree.as_deref(),
3053 &daemons,
3054 &state.tld,
3055 &host_port_suffix(&raw_host),
3056 crate::web::url().as_deref(),
3057 );
3058 }
3059 ResolveResult::Unknown { heading, known } => {
3060 return unknown_host_response(
3063 &host,
3064 if local_client { &heading } else { "Not found" },
3065 if local_client { &known } else { &[] },
3066 );
3067 }
3068 ResolveResult::NotFound => {
3069 return error_response(
3070 StatusCode::BAD_GATEWAY,
3071 &format!(
3072 "No daemon found for host '{host}'.\n\
3073 A daemon is reachable once it configures a `port` and its project is \
3074 known to pitchfork; run `pitchfork proxy status` to see the hostnames \
3075 it serves.\n\
3076 Expected format: <daemon>.<project>.{tld}",
3077 tld = state.tld
3078 ),
3079 );
3080 }
3081 ResolveResult::Error(msg) => {
3082 if local_client {
3083 return error_response(StatusCode::BAD_GATEWAY, &msg);
3084 }
3085 log::warn!("Refused '{host}' for a non-local client: {msg}");
3087 return error_response(
3088 StatusCode::BAD_GATEWAY,
3089 &format!("'{host}' is not available."),
3090 );
3091 }
3092 }
3093 };
3094 let path_and_query = req
3096 .uri()
3097 .path_and_query()
3098 .map(|pq| pq.as_str())
3099 .unwrap_or("/");
3100
3101 let forward_uri = match Uri::builder()
3102 .scheme("http")
3103 .authority(format!("localhost:{target_port}"))
3104 .path_and_query(path_and_query)
3105 .build()
3106 {
3107 Ok(uri) => uri,
3108 Err(e) => {
3109 return error_response(
3110 StatusCode::INTERNAL_SERVER_ERROR,
3111 &format!("Failed to build forward URI: {e}"),
3112 );
3113 }
3114 };
3115
3116 let h2c = protocol.is_h2c();
3117 if h2c && req.headers().contains_key(axum::http::header::UPGRADE) {
3118 return error_response(
3119 StatusCode::NOT_IMPLEMENTED,
3120 &format!(
3121 "'{host}' uses proxy_protocol = \"h2c\", which cannot carry protocol upgrades \
3122 such as WebSocket. Remove proxy_protocol, or connect without an Upgrade header."
3123 ),
3124 );
3125 }
3126 if h2c {
3127 let nominated: Vec<axum::http::HeaderName> = req
3131 .headers()
3132 .get_all(axum::http::header::CONNECTION)
3133 .iter()
3134 .filter_map(|v| v.to_str().ok())
3135 .flat_map(|v| v.split(','))
3136 .filter_map(|t| axum::http::HeaderName::from_bytes(t.trim().as_bytes()).ok())
3137 .collect();
3138 for name in nominated {
3139 if name != axum::http::header::TE {
3140 req.headers_mut().remove(&name);
3141 }
3142 }
3143 for h in HOP_BY_HOP_HEADERS {
3144 if let Ok(name) = axum::http::HeaderName::from_bytes(h.as_bytes()) {
3145 req.headers_mut().remove(&name);
3146 }
3147 }
3148 }
3149
3150 *req.uri_mut() = forward_uri;
3152 req.headers_mut().insert(
3153 HOST,
3154 HeaderValue::from_str(&format!("localhost:{target_port}"))
3155 .unwrap_or_else(|_| HeaderValue::from_static("localhost")),
3156 );
3157
3158 inject_forwarded_headers(&mut req, state.is_tls, &raw_host);
3160
3161 if let Ok(v) = HeaderValue::from_str(&(hops + 1).to_string()) {
3163 req.headers_mut()
3164 .insert(axum::http::HeaderName::from_static(PROXY_HOPS_HEADER), v);
3165 }
3166
3167 let pseudo_headers: Vec<_> = req
3172 .headers()
3173 .keys()
3174 .filter(|k| k.as_str().starts_with(':'))
3175 .cloned()
3176 .collect();
3177 for key in pseudo_headers {
3178 req.headers_mut().remove(&key);
3179 }
3180
3181 join_cookie_fields(req.headers_mut());
3182
3183 *req.version_mut() = if h2c {
3188 axum::http::Version::HTTP_2
3189 } else {
3190 axum::http::Version::HTTP_11
3191 };
3192
3193 let client_upgrade = hyper::upgrade::on(&mut req);
3195
3196 let result = match tokio::time::timeout(std::time::Duration::from_secs(120), async {
3204 if h2c {
3205 state.h2c_client.request(req).await
3206 } else {
3207 state.client.request(req).await
3208 }
3209 })
3210 .await
3211 {
3212 Ok(r) => r,
3213 Err(_elapsed) => {
3214 let msg = format!(
3215 "Request to daemon on port {target_port} timed out after 120 s.\n\
3216 The daemon accepted the connection but did not respond in time."
3217 );
3218 log::warn!("{msg}");
3219 if let Some(ref on_error) = state.on_error {
3220 on_error(&msg);
3221 }
3222 return error_response(StatusCode::GATEWAY_TIMEOUT, &msg);
3223 }
3224 };
3225 match result {
3226 Ok(mut resp) => {
3227 let backend_upgrade = hyper::upgrade::on(&mut resp);
3229 let (mut parts, body) = resp.into_parts();
3230
3231 parts.headers.insert(
3233 axum::http::HeaderName::from_static(PITCHFORK_HEADER),
3234 HeaderValue::from_static("1"),
3235 );
3236
3237 parts.headers.remove(PROXY_HOPS_HEADER);
3239
3240 if state.is_tls && parts.status != StatusCode::SWITCHING_PROTOCOLS {
3245 for h in HOP_BY_HOP_HEADERS {
3246 if let Ok(name) = axum::http::HeaderName::from_bytes(h.as_bytes()) {
3247 parts.headers.remove(&name);
3248 }
3249 }
3250 }
3251
3252 if parts.status == StatusCode::SWITCHING_PROTOCOLS {
3254 let cancel = state.cancel.clone();
3262 state.tunnels.spawn(async move {
3263 let _activity = activity;
3266 let splice = async move {
3267 if let (Ok(client_upgraded), Ok(backend_upgraded)) =
3268 (client_upgrade.await, backend_upgrade.await)
3269 {
3270 let mut client_io = hyper_util::rt::TokioIo::new(client_upgraded);
3271 let mut backend_io = hyper_util::rt::TokioIo::new(backend_upgraded);
3272 let _ = tokio::io::copy_bidirectional(&mut client_io, &mut backend_io)
3280 .await;
3281 }
3282 };
3283 tokio::select! {
3287 _ = splice => {}
3288 _ = cancel.cancelled() => {}
3289 }
3290 });
3291 return Response::from_parts(parts, Body::empty());
3292 }
3293
3294 Response::from_parts(parts, Body::new(GuardedBody::new(body, activity)))
3300 }
3301 Err(e) => {
3302 let msg = format!(
3303 "Failed to connect to daemon on port {target_port}: {e}\n\
3304 The daemon may have stopped or is not yet ready."
3305 );
3306 if let Some(ref on_error) = state.on_error {
3307 on_error(&msg);
3308 } else {
3309 log::warn!("{msg}");
3310 }
3311 error_response(StatusCode::BAD_GATEWAY, &msg)
3312 }
3313 }
3314}
3315
3316fn passthrough_unroutable_message(host: &str, is_tls: bool) -> String {
3326 if is_tls {
3327 format!(
3328 "'{host}' uses proxy_tls = \"passthrough\", which routes on the host name \
3329 in the TLS ClientHello.\n\
3330 This connection's TLS handshake named a different host, so it was \
3331 terminated here and the request inside it cannot be spliced to the \
3332 daemon.\n\
3333 Make the connection itself name '{host}' rather than overriding the Host \
3334 header of a connection opened to something else."
3335 )
3336 } else {
3337 format!(
3338 "'{host}' uses proxy_tls = \"passthrough\", but the proxy is serving \
3339 plain HTTP.\n\
3340 Passthrough splices a TLS stream to the daemon, so it requires \
3341 settings.proxy.https = true.\n\
3342 Enable HTTPS on the proxy, or set proxy_tls = \"terminate\" on the daemon."
3343 )
3344 }
3345}
3346
3347async fn resolve_target(host: &str, tld: &str) -> ResolveResult {
3366 let deadline = auto_start_deadline();
3369 let ctx = match resolve_route_context(host, tld).await {
3370 Ok(ctx) => ctx,
3371 Err(result) => return result,
3372 };
3373
3374 let daemons = {
3375 let state_file = SUPERVISOR.state_file.lock().await;
3376 state_file.daemons.clone()
3377 };
3378
3379 let daemon_name = &ctx.cached.daemon_name;
3380 let running_matches: Vec<(&DaemonId, &crate::daemon::Daemon)> = daemons
3381 .iter()
3382 .filter(|(id, d)| {
3383 id.name() == daemon_name
3384 && d.status.is_running()
3385 && match &ctx.expected_namespace {
3386 Some(ns) => id.namespace() == ns,
3387 None => true,
3388 }
3389 })
3390 .collect();
3391
3392 match running_matches.as_slice() {
3393 [] => {
3394 try_auto_start(
3395 &ctx.cached.slug,
3396 &ctx.cached,
3397 ctx.worktree_dir.as_deref(),
3398 ctx.expected_namespace.as_deref(),
3399 &ctx.route,
3400 deadline,
3401 )
3402 .await
3403 }
3404 [(id, d), ..] => match select_daemon_port(&ctx.route, d) {
3407 Some(port) => match begin_running(id, deadline).await {
3408 Running::Yes(activity) => {
3409 ResolveResult::Ready(port, Some(activity), ctx.route.protocol)
3410 }
3411 Running::Stopping => ResolveResult::Starting {
3412 slug: ctx.cached.slug.clone(),
3413 },
3414 Running::No => {
3415 try_auto_start(
3416 &ctx.cached.slug,
3417 &ctx.cached,
3418 ctx.worktree_dir.as_deref(),
3419 ctx.expected_namespace.as_deref(),
3420 &ctx.route,
3421 deadline,
3422 )
3423 .await
3424 }
3425 },
3426 None => ResolveResult::NotFound,
3427 },
3428 }
3429}
3430
3431enum Running {
3434 Yes(ActivityGuard),
3436 Stopping,
3438 No,
3441}
3442
3443fn auto_start_deadline() -> tokio::time::Instant {
3446 tokio::time::Instant::now() + settings().proxy_auto_start_timeout()
3447}
3448
3449async fn wait_out_idle_stops(ids: &[DaemonId], deadline: tokio::time::Instant) -> bool {
3456 while ids.iter().any(|id| ACTIVITY.is_idle_stopping(id)) {
3457 if tokio::time::Instant::now() >= deadline {
3458 return false;
3459 }
3460 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
3461 }
3462 true
3463}
3464
3465async fn begin_running(id: &DaemonId, deadline: tokio::time::Instant) -> Running {
3476 let activity = match ACTIVITY.begin(id) {
3477 Some(activity) => activity,
3478 None => {
3479 if !wait_out_idle_stops(std::slice::from_ref(id), deadline).await {
3480 return Running::Stopping;
3481 }
3482 match ACTIVITY.begin(id) {
3483 Some(activity) => activity,
3484 None => return Running::Stopping,
3486 }
3487 }
3488 };
3489 let running = {
3490 let state_file = SUPERVISOR.state_file.lock().await;
3491 state_file
3492 .daemons
3493 .get(id)
3494 .is_some_and(|d| d.status.is_running())
3495 };
3496 if running {
3497 Running::Yes(activity)
3498 } else {
3499 Running::No
3500 }
3501}
3502
3503struct RouteContext {
3507 cached: CachedSlugEntry,
3508 expected_namespace: Option<String>,
3509 worktree_dir: Option<std::path::PathBuf>,
3510 route: ProxyTlsRoute,
3511}
3512
3513async fn resolve_route_context(host: &str, tld: &str) -> Result<RouteContext, ResolveResult> {
3522 let Some(subdomain) = strip_tld(host, tld) else {
3523 return Err(ResolveResult::NotFound);
3524 };
3525
3526 let cached = cached_slug_lookup(&subdomain).await.filter(|cached| {
3527 if crate::proxy::hostname::hostname_fits(&cached.slug) {
3531 return true;
3532 }
3533 crate::proxy::hostname::warn_once(&format!(
3534 "Slug '{}' plus the configured proxy.tld is over the DNS length limit, \
3535 so it is not routed.",
3536 cached.slug
3537 ));
3538 false
3539 });
3540 let Some(cached) = cached else {
3541 return Err(resolve_registry_target(&subdomain).await);
3544 };
3545
3546 let (expected_namespace, worktree_dir, route) = if !subdomain.eq_ignore_ascii_case(&cached.slug)
3550 {
3551 let prefix = strip_dot_suffix_ignore_case(&subdomain, &cached.slug);
3552 match prefix {
3553 Some(ref p) => match match_worktree_prefix(&cached, p) {
3554 PrefixMatch::Worktree(wt) => {
3555 let ns = wt.namespace.clone().or_else(|| {
3556 log::warn!(
3557 "Worktree '{}' has no cached namespace; \
3558 falling back to parent slug namespace.",
3559 wt.path.display()
3560 );
3561 cached.namespace.clone()
3562 });
3563 let route = worktree_route(&cached, &wt.sanitized_branch);
3564 (ns, Some(wt.path.clone()), route)
3565 }
3566 PrefixMatch::Ambiguous => {
3567 return Err(ResolveResult::Error(format!(
3568 "'{host}' is ambiguous: more than one branch or workspace of '{slug}' \
3569 sanitizes to the prefix '{p}', and host names are case-insensitive.\n\
3570 Rename one of them so the prefixes differ by more than case, then \
3571 reload.\n\
3572 The supervisor log lists the colliding branches.",
3573 slug = cached.slug,
3574 )));
3575 }
3576 PrefixMatch::Unknown => (cached.namespace.clone(), None, cached.tls),
3577 },
3578 None => (cached.namespace.clone(), None, cached.tls),
3579 }
3580 } else {
3581 (cached.namespace.clone(), None, cached.tls)
3582 };
3583
3584 Ok(RouteContext {
3585 cached,
3586 expected_namespace,
3587 worktree_dir,
3588 route,
3589 })
3590}
3591
3592pub(crate) async fn resolve_tls_mode(host: &str, tld: &str) -> ProxyTlsMode {
3599 let entries = get_cached_slugs().await;
3600 let registry = get_cached_host_registry().await;
3601 resolve_tls_mode_in(host, tld, &entries, ®istry)
3602}
3603
3604fn resolve_tls_mode_in(
3609 host: &str,
3610 tld: &str,
3611 entries: &std::collections::HashMap<String, CachedSlugEntry>,
3612 registry: &crate::proxy::hostname::HostRegistry,
3613) -> ProxyTlsMode {
3614 let Some(subdomain) = strip_tld(host, tld) else {
3615 return ProxyTlsMode::Terminate;
3616 };
3617 let wildcard = settings().proxy.wildcard;
3618
3619 if let Some(cached) = wildcard_slug_lookup(&subdomain, entries, wildcard)
3621 && crate::proxy::hostname::hostname_fits(&cached.slug)
3622 {
3623 if !subdomain.eq_ignore_ascii_case(&cached.slug)
3625 && let Some(prefix) = strip_dot_suffix_ignore_case(&subdomain, &cached.slug)
3626 && let PrefixMatch::Worktree(wt) = match_worktree_prefix(cached, &prefix)
3627 {
3628 return worktree_route(cached, &wt.sanitized_branch).mode;
3629 }
3630 return cached.tls.mode;
3631 }
3632
3633 match registry.resolve(&subdomain, wildcard) {
3636 crate::proxy::hostname::HostTarget::Daemon { proxy_tls, .. } => {
3637 proxy_tls.unwrap_or_default()
3638 }
3639 _ => ProxyTlsMode::Terminate,
3642 }
3643}
3644
3645fn select_daemon_port(route: &ProxyTlsRoute, daemon: &crate::daemon::Daemon) -> Option<u16> {
3662 let Some(want) = route.port else {
3663 let detected = daemon.active_port.filter(|&p| p != 0);
3672 let first_declared = daemon.resolved_port.iter().copied().find(|&p| p != 0);
3673 return if route.mode.is_passthrough() {
3674 first_declared.or(detected)
3675 } else {
3676 detected.or(first_declared)
3677 };
3678 };
3679
3680 let configured = daemon
3681 .port
3682 .as_ref()
3683 .map(|p| p.expect.as_slice())
3684 .unwrap_or(&[]);
3685 if let Some(idx) = configured.iter().position(|&p| p == want)
3686 && let Some(&resolved) = daemon.resolved_port.get(idx)
3687 {
3688 return (resolved != 0).then_some(resolved);
3694 }
3695 if daemon.resolved_port.contains(&want) {
3696 return Some(want);
3697 }
3698 if daemon.resolved_port.is_empty() {
3699 return Some(want);
3702 }
3703 crate::proxy::hostname::warn_once(&format!(
3706 "Daemon {} has proxy_tls_port {want}, which is not among its resolved ports {:?}; \
3707 refusing to route rather than forwarding to a port it never bound. \
3708 Restart the daemon if its ports changed.",
3709 daemon.id, daemon.resolved_port,
3710 ));
3711 None
3712}
3713
3714struct AutoStartGuard {
3721 daemon_id: DaemonId,
3722}
3723
3724impl Drop for AutoStartGuard {
3725 fn drop(&mut self) {
3726 let daemon_id = self.daemon_id.clone();
3727 tokio::spawn(async move {
3731 AUTO_START_IN_PROGRESS.lock().await.remove(&daemon_id);
3732 });
3733 }
3734}
3735
3736static STARTUP_LOCKS: once_cell::sync::Lazy<
3750 std::sync::Mutex<std::collections::HashMap<DaemonId, std::sync::Weak<tokio::sync::Mutex<()>>>>,
3751> = once_cell::sync::Lazy::new(Default::default);
3752
3753async fn lock_startup_graph(mut ids: Vec<DaemonId>) -> Vec<tokio::sync::OwnedMutexGuard<()>> {
3756 ids.sort();
3757 ids.dedup();
3758 let locks: Vec<Arc<tokio::sync::Mutex<()>>> = {
3759 let mut map = STARTUP_LOCKS.lock().unwrap_or_else(|e| e.into_inner());
3760 map.retain(|_, lock| lock.strong_count() > 0);
3761 ids.into_iter()
3762 .map(|id| {
3763 let entry = map.entry(id).or_default();
3764 entry.upgrade().unwrap_or_else(|| {
3765 let lock = Arc::new(tokio::sync::Mutex::new(()));
3766 *entry = Arc::downgrade(&lock);
3767 lock
3768 })
3769 })
3770 .collect()
3771 };
3772 let mut guards = Vec::with_capacity(locks.len());
3773 for lock in locks {
3774 guards.push(lock.lock_owned().await);
3775 }
3776 guards
3777}
3778
3779async fn try_auto_start(
3793 slug: &str,
3794 cached: &CachedSlugEntry,
3795 worktree_dir: Option<&std::path::Path>,
3796 expected_namespace: Option<&str>,
3797 route: &ProxyTlsRoute,
3798 deadline: tokio::time::Instant,
3799) -> ResolveResult {
3800 let s = settings();
3801 if !s.proxy.auto_start {
3802 return ResolveResult::NotFound;
3803 }
3804
3805 let ns = expected_namespace
3806 .map(|s| s.to_string())
3807 .or_else(|| cached.namespace.clone())
3808 .unwrap_or_else(|| "global".to_string());
3809 let daemon_id = match DaemonId::try_new(&ns, &cached.daemon_name) {
3810 Ok(id) => id,
3811 Err(_) => return ResolveResult::NotFound,
3812 };
3813
3814 {
3815 let mut in_progress = AUTO_START_IN_PROGRESS.lock().await;
3816 if !in_progress.insert(daemon_id.clone()) {
3817 return ResolveResult::Starting {
3818 slug: slug.to_string(),
3819 };
3820 }
3821 }
3822
3823 let guard = Arc::new(AutoStartGuard {
3826 daemon_id: daemon_id.clone(),
3827 });
3828
3829 let timeout = s.proxy_auto_start_timeout();
3832 let timed_out = || {
3833 log::warn!("Auto-start: total timeout ({timeout:?}) exceeded for daemon {daemon_id}");
3834 ResolveResult::Error(format!(
3835 "Auto-start for '{daemon_id}' timed out after {timeout:?}.\n\
3836 The daemon and its dependencies did not all become ready, with the daemon \
3837 bound to a port, within the configured proxy_auto_start_timeout.\n\
3838 Startup continues in the background; reload to check again.\n\
3839 Increase the timeout or check the logs of the daemon and its dependencies \
3840 for slow startup."
3841 ))
3842 };
3843
3844 log::info!("Auto-start: starting daemon {daemon_id} for slug '{slug}'");
3845 let start = tokio::spawn({
3846 let guard = guard.clone();
3847 let daemon_id = daemon_id.clone();
3848 let config_dir = worktree_dir.unwrap_or(&cached.dir).to_path_buf();
3849 async move {
3850 let _guard = guard;
3851 start_with_dependencies(&daemon_id, &config_dir).await
3852 }
3853 });
3854 match tokio::time::timeout_at(deadline, start).await {
3855 Ok(Ok(Ok(()))) => {}
3856 Ok(Ok(Err(result))) => return result,
3857 Ok(Err(e)) => {
3858 log::warn!("Auto-start: start task for {daemon_id} failed: {e}");
3859 return ResolveResult::Error(format!("Failed to start daemon '{daemon_id}': {e}"));
3860 }
3861 Err(_elapsed) => return timed_out(),
3862 }
3863
3864 let result =
3865 tokio::time::timeout_at(deadline, wait_for_active_port(slug, &daemon_id, route)).await;
3866 drop(guard);
3867 result.unwrap_or_else(|_elapsed| timed_out())
3868}
3869
3870async fn start_with_dependencies(
3880 daemon_id: &DaemonId,
3881 config_dir: &std::path::Path,
3882) -> std::result::Result<(), ResolveResult> {
3883 let loaded =
3884 crate::pitchfork_toml::PitchforkToml::all_merged_all_namespaces_from(config_dir).await;
3885 let pt = match loaded {
3886 Ok(pt) => pt,
3887 Err(e) => {
3888 log::warn!(
3889 "Auto-start: failed to load config from {}: {e}",
3890 config_dir.display()
3891 );
3892 return Err(ResolveResult::NotFound);
3893 }
3894 };
3895
3896 if !pt.daemons.contains_key(daemon_id) {
3897 log::debug!(
3898 "Auto-start: daemon {daemon_id} not found in config at {}",
3899 config_dir.display()
3900 );
3901 return Err(ResolveResult::NotFound);
3902 }
3903
3904 if SUPERVISOR
3906 .state_file
3907 .lock()
3908 .await
3909 .disabled
3910 .contains(daemon_id)
3911 {
3912 return Err(ResolveResult::Error(format!(
3913 "Daemon '{daemon_id}' is disabled, so it is not started.\n\
3914 Enable it with: pitchfork enable {daemon_id}"
3915 )));
3916 }
3917
3918 let graph: Vec<DaemonId> =
3919 match crate::deps::resolve_dependencies(std::slice::from_ref(daemon_id), &pt.daemons) {
3920 Ok(order) => order.levels.into_iter().flatten().collect(),
3921 Err(e) => {
3922 log::warn!("Auto-start: cannot resolve dependencies of {daemon_id}: {e}");
3923 return Err(ResolveResult::Error(format!(
3924 "Cannot start '{daemon_id}': {e}"
3925 )));
3926 }
3927 };
3928 let _activity = match ACTIVITY.begin_all(&graph) {
3936 Some(activity) => activity,
3937 None => {
3938 wait_out_idle_stops(&graph, auto_start_deadline()).await;
3939 match ACTIVITY.begin_all(&graph) {
3940 Some(activity) => activity,
3941 None => {
3942 return Err(ResolveResult::Error(format!(
3943 "A dependency of '{daemon_id}' is still being stopped for inactivity.\n\
3944 Reload to start it again."
3945 )));
3946 }
3947 }
3948 }
3949 };
3950 let proxy_idle = proxy_idle_timeouts(daemon_id, &graph, &pt).await;
3951 let _locks = lock_startup_graph(graph).await;
3952
3953 let ipc = match crate::ipc::client::IpcClient::connect(false).await {
3956 Ok(ipc) => Arc::new(ipc),
3957 Err(e) => {
3958 log::warn!("Auto-start: failed to connect to the supervisor: {e}");
3959 return Err(ResolveResult::Error(format!(
3960 "Failed to start daemon '{daemon_id}': {e}"
3961 )));
3962 }
3963 };
3964 let opts = crate::ipc::batch::StartOptions {
3965 quiet: true,
3966 proxy_idle: Some(proxy_idle),
3967 ..Default::default()
3968 };
3969 let result = match ipc
3970 .start_daemons_with_config(std::slice::from_ref(daemon_id), opts, pt)
3971 .await
3972 {
3973 Ok(result) => result,
3974 Err(e) => {
3975 log::warn!("Auto-start: failed to start {daemon_id}: {e}");
3976 return Err(ResolveResult::Error(format!(
3977 "Failed to start daemon '{daemon_id}': {e}"
3978 )));
3979 }
3980 };
3981 if !result.any_failed {
3982 return Ok(());
3983 }
3984
3985 let message = match result.failed.first() {
3986 Some((id, reason)) if id == daemon_id => {
3987 format!("Daemon '{daemon_id}' failed to start: {reason}\nCheck its logs for errors.")
3988 }
3989 Some((dep, reason)) => format!(
3990 "Daemon '{daemon_id}' was not started because its dependency '{dep}' failed: \
3991 {reason}\n\
3992 Check the logs of '{dep}' for errors."
3993 ),
3994 None => format!(
3995 "Daemon '{daemon_id}' or one of its dependencies failed to start.\n\
3996 Check the supervisor log for errors."
3997 ),
3998 };
3999 log::warn!("Auto-start: {message}");
4000 Err(ResolveResult::Error(message))
4001}
4002
4003async fn wait_for_active_port(
4005 slug: &str,
4006 daemon_id: &DaemonId,
4007 route: &ProxyTlsRoute,
4008) -> ResolveResult {
4009 let poll_interval = std::time::Duration::from_millis(250);
4010
4011 loop {
4012 let daemons = {
4013 let sf = SUPERVISOR.state_file.lock().await;
4014 sf.daemons.clone()
4015 };
4016
4017 if let Some(d) = daemons.get(daemon_id) {
4018 if d.status.is_running() {
4019 if let Some(port) = select_daemon_port(route, d) {
4024 return match ACTIVITY.begin(daemon_id) {
4032 Some(activity) => {
4033 log::info!("Auto-start: daemon {daemon_id} is ready on port {port}");
4034 ResolveResult::Ready(port, Some(activity), route.protocol)
4035 }
4036 None => ResolveResult::Starting {
4037 slug: slug.to_string(),
4038 },
4039 };
4040 }
4041 } else {
4042 log::warn!(
4043 "Auto-start: daemon {daemon_id} is no longer running (status: {})",
4044 d.status
4045 );
4046 return ResolveResult::Error(format!(
4047 "Daemon '{daemon_id}' started but exited unexpectedly.\n\
4048 Check its logs for errors."
4049 ));
4050 }
4051 } else {
4052 log::warn!("Auto-start: daemon {daemon_id} not found in state file after start");
4053 return ResolveResult::Error(format!(
4054 "Daemon '{daemon_id}' started but disappeared from the state file.\n\
4055 Check its logs for errors."
4056 ));
4057 }
4058
4059 tokio::time::sleep(poll_interval).await;
4060 }
4061}
4062
4063async fn proxy_idle_timeouts(
4075 daemon_id: &DaemonId,
4076 closure: &[DaemonId],
4077 pt: &crate::pitchfork_toml::PitchforkToml,
4078) -> std::collections::HashMap<DaemonId, u64> {
4079 let own = |id: &DaemonId| pt.daemons.get(id).and_then(|d| d.proxy_idle_timeout);
4080 let target = match own(daemon_id) {
4081 Some(configured) => configured.duration(),
4082 None => {
4083 let project_dir = crate::ipc::batch::resolve_config_base_dir(
4085 pt.daemons.get(daemon_id).and_then(|d| d.path.as_deref()),
4086 );
4087 tokio::task::spawn_blocking(move || {
4088 crate::settings::Settings::load_from_dir(&project_dir).proxy_idle_timeout()
4089 })
4090 .await
4091 .ok()
4092 .flatten()
4093 }
4094 };
4095 let millis = |d: std::time::Duration| u64::try_from(d.as_millis()).unwrap_or(u64::MAX);
4096 closure
4097 .iter()
4098 .filter_map(|id| {
4099 let grace = if id == daemon_id {
4100 target
4101 } else {
4102 own(id).map_or(target, |configured| configured.duration())
4103 };
4104 grace.map(|g| (id.clone(), millis(g)))
4105 })
4106 .collect()
4107}
4108
4109async fn resolve_registry_target(subdomain: &str) -> ResolveResult {
4114 let registry = get_cached_host_registry().await;
4115 if !crate::proxy::hostname::hostname_fits(subdomain) {
4116 return ResolveResult::Unknown {
4118 heading: "Host name too long".to_string(),
4119 known: registry.project_labels(),
4120 };
4121 }
4122 match registry.resolve(subdomain, settings().proxy.wildcard) {
4123 crate::proxy::hostname::HostTarget::Daemon {
4124 ref dir,
4125 ref namespace,
4126 ref daemon,
4127 proxy_tls,
4128 proxy_protocol,
4129 proxy_tls_port,
4130 ..
4131 } => {
4132 let route = ProxyTlsRoute {
4137 mode: proxy_tls.unwrap_or_default(),
4138 protocol: proxy_protocol.unwrap_or_default(),
4139 port: proxy_tls_port,
4140 };
4141 let per_checkout = registry.shares_daemon_id(namespace, daemon);
4146 resolve_registry_daemon(subdomain, dir, namespace, daemon, per_checkout, &route).await
4147 }
4148 crate::proxy::hostname::HostTarget::ProjectPage { project } => {
4149 let entry = registry.projects.get(&project);
4150 ResolveResult::Page {
4151 daemons: entry.map(|p| p.primary.labels()).unwrap_or_default(),
4152 dir: entry.map(|p| p.primary.dir.clone()),
4153 project,
4154 worktree: None,
4155 }
4156 }
4157 crate::proxy::hostname::HostTarget::WorktreePage { project, worktree } => {
4158 let checkout = registry
4159 .projects
4160 .get(&project)
4161 .and_then(|p| p.worktrees.get(&worktree));
4162 ResolveResult::Page {
4163 daemons: checkout.map(|c| c.labels()).unwrap_or_default(),
4164 dir: checkout.map(|c| c.dir.clone()),
4165 project,
4166 worktree: Some(worktree),
4167 }
4168 }
4169 crate::proxy::hostname::HostTarget::UnknownProject { known } => ResolveResult::Unknown {
4170 heading: "Unknown project".to_string(),
4171 known,
4172 },
4173 crate::proxy::hostname::HostTarget::UnknownDaemon {
4174 project,
4175 worktree,
4176 known,
4177 } => ResolveResult::Unknown {
4178 heading: match worktree {
4179 Some(wt) => format!("Unknown daemon in '{wt}' of project '{project}'"),
4180 None => format!("Unknown daemon in project '{project}'"),
4181 },
4182 known,
4183 },
4184 }
4185}
4186
4187async fn resolve_registry_daemon(
4194 host: &str,
4195 dir: &std::path::Path,
4196 namespace: &str,
4197 daemon: &str,
4198 per_checkout: bool,
4199 route: &ProxyTlsRoute,
4200) -> ResolveResult {
4201 let deadline = auto_start_deadline();
4203 let daemons = {
4204 let state_file = SUPERVISOR.state_file.lock().await;
4205 state_file.daemons.clone()
4206 };
4207
4208 let mut matches: Vec<crate::daemon::Daemon> = daemons
4209 .iter()
4210 .filter(|(id, d)| {
4211 id.name() == daemon && id.namespace() == namespace && d.status.is_running()
4212 })
4213 .map(|(_, d)| d.clone())
4214 .collect();
4215 matches = sort_by_checkout(matches, dir).await;
4218
4219 if let Some(d) = matches.first() {
4220 if per_checkout && !runs_in_checkout(d.clone(), dir).await {
4223 return ResolveResult::Error(format!(
4224 "'{host}' belongs to the checkout at {}, but daemon '{namespace}/{daemon}' is \
4225 running from {}.\n\
4226 These checkouts share the namespace '{namespace}', so pitchfork cannot run \
4227 both copies at once.\n\
4228 Give each checkout its own top-level `namespace`, or stop the other one first.",
4229 dir.display(),
4230 d.dir
4231 .as_deref()
4232 .map(|p| p.display().to_string())
4233 .unwrap_or_else(|| "an unknown directory".to_string()),
4234 ));
4235 }
4236 let Some(port) = select_daemon_port(route, d) else {
4237 return ResolveResult::NotFound;
4238 };
4239 match begin_running(&d.id, deadline).await {
4240 Running::Yes(activity) => {
4241 return ResolveResult::Ready(port, Some(activity), route.protocol);
4242 }
4243 Running::Stopping => {
4244 return ResolveResult::Starting {
4245 slug: host.to_string(),
4246 };
4247 }
4248 Running::No => {}
4250 }
4251 }
4252
4253 let cached = CachedSlugEntry {
4254 slug: host.to_string(),
4255 namespace: Some(namespace.to_string()),
4256 daemon_name: daemon.to_string(),
4257 dir: dir.to_path_buf(),
4258 worktrees: vec![],
4259 rejected_worktree_prefixes: std::collections::HashSet::new(),
4260 tls: *route,
4261 worktree_tls: std::collections::HashMap::new(),
4262 };
4263 let result = try_auto_start(host, &cached, None, Some(namespace), route, deadline).await;
4264
4265 if per_checkout && let ResolveResult::Ready(..) = result {
4269 let started = {
4270 let state_file = SUPERVISOR.state_file.lock().await;
4271 state_file
4272 .daemons
4273 .iter()
4274 .find(|(id, _)| id.name() == daemon && id.namespace() == namespace)
4275 .map(|(_, d)| d.clone())
4276 };
4277 if let Some(d) = started
4278 && !runs_in_checkout(d.clone(), dir).await
4279 {
4280 return ResolveResult::Error(format!(
4281 "'{host}' belongs to the checkout at {}, but daemon '{namespace}/{daemon}' is \
4282 running from {}.\n\
4283 These checkouts share the namespace '{namespace}', so pitchfork cannot run \
4284 both copies at once.\n\
4285 Give each checkout its own top-level `namespace`, or stop the other one first.",
4286 dir.display(),
4287 d.dir
4288 .as_deref()
4289 .map(|p| p.display().to_string())
4290 .unwrap_or_else(|| "an unknown directory".to_string()),
4291 ));
4292 }
4293 }
4294
4295 result
4296}
4297
4298fn is_local_client(req: &Request) -> bool {
4305 req.extensions()
4308 .get::<axum::extract::ConnectInfo<SocketAddr>>()
4309 .is_some_and(|ci| ci.0.ip().is_loopback())
4310}
4311
4312fn daemon_runs_in(daemon: &crate::daemon::Daemon, checkout: &std::path::Path) -> bool {
4321 daemon
4322 .dir
4323 .as_deref()
4324 .is_some_and(|d| crate::proxy::hostname::checkout_root_of(d) == checkout)
4325}
4326
4327async fn runs_in_checkout(daemon: crate::daemon::Daemon, checkout: &std::path::Path) -> bool {
4329 let checkout = checkout.to_path_buf();
4330 tokio::task::spawn_blocking(move || daemon_runs_in(&daemon, &checkout))
4331 .await
4332 .unwrap_or(false)
4333}
4334
4335async fn sort_by_checkout(
4341 daemons: Vec<crate::daemon::Daemon>,
4342 checkout: &std::path::Path,
4343) -> Vec<crate::daemon::Daemon> {
4344 if daemons.len() < 2 {
4345 return daemons;
4346 }
4347 let dirs: Vec<Option<std::path::PathBuf>> = daemons.iter().map(|d| d.dir.clone()).collect();
4348 let checkout = checkout.to_path_buf();
4349 let here = tokio::task::spawn_blocking(move || {
4350 dirs.iter()
4351 .map(|dir| {
4352 dir.as_deref()
4353 .is_some_and(|d| crate::proxy::hostname::checkout_root_of(d) == checkout)
4354 })
4355 .collect::<Vec<bool>>()
4356 })
4357 .await;
4358
4359 match here {
4360 Ok(here) => {
4361 let mut ordered: Vec<(bool, crate::daemon::Daemon)> =
4362 here.into_iter().zip(daemons).collect();
4363 ordered.sort_by_key(|(here, _)| !here);
4364 ordered.into_iter().map(|(_, d)| d).collect()
4365 }
4366 Err(e) => {
4367 log::warn!("Checkout attribution task failed: {e}");
4368 daemons
4369 }
4370 }
4371}
4372
4373fn escape_html(s: &str) -> String {
4375 s.replace('&', "&")
4376 .replace('<', "<")
4377 .replace('>', ">")
4378 .replace('"', """)
4379 .replace('\'', "'")
4380}
4381
4382fn html_page(status: StatusCode, title: &str, body: String) -> Response {
4384 let html = format!(
4385 r##"<!DOCTYPE html>
4386<html lang="en">
4387<head>
4388 <meta charset="UTF-8">
4389 <meta name="viewport" content="width=device-width, initial-scale=1">
4390 <title>{title} — pitchfork</title>
4391 <style>
4392 * {{ margin: 0; padding: 0; box-sizing: border-box; }}
4393 body {{
4394 font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
4395 background: #0f1117;
4396 color: #e1e4e8;
4397 display: flex;
4398 align-items: center;
4399 justify-content: center;
4400 min-height: 100vh;
4401 }}
4402 .container {{ max-width: 640px; padding: 2rem; }}
4403 h1 {{ font-size: 1.5rem; font-weight: 600; margin-bottom: 0.75rem; }}
4404 p {{ color: #8b949e; font-size: 0.9rem; margin-bottom: 0.75rem; }}
4405 ul {{ list-style: none; margin: 0.5rem 0 1rem; }}
4406 li {{ margin: 0.25rem 0; }}
4407 code, a {{
4408 color: #58a6ff;
4409 font-family: "SFMono-Regular", Consolas, "Liberation Mono", Menlo, monospace;
4410 text-decoration: none;
4411 }}
4412 </style>
4413</head>
4414<body>
4415 <div class="container">{body}</div>
4416</body>
4417</html>"##
4418 );
4419 Response::builder()
4420 .status(status)
4421 .header("content-type", "text/html; charset=utf-8")
4422 .body(Body::from(html))
4423 .unwrap_or_else(|_| (status, title.to_string()).into_response())
4424}
4425
4426fn host_port_suffix(raw_host: &str) -> String {
4431 let port = if raw_host.starts_with('[') {
4432 raw_host.split_once("]:").map(|(_, port)| port)
4433 } else {
4434 raw_host.rsplit_once(':').map(|(_, port)| port)
4435 };
4436 port.filter(|p| p.chars().all(|c| c.is_ascii_digit()) && !p.is_empty())
4437 .map(|p| format!(":{p}"))
4438 .unwrap_or_default()
4439}
4440
4441fn page_redirect_response(base: &str, path: &str) -> Response {
4447 let target = format!("{base}{path}");
4448 Response::builder()
4449 .status(StatusCode::FOUND)
4450 .header(axum::http::header::LOCATION, &target)
4451 .header(axum::http::header::CACHE_CONTROL, "no-store")
4452 .body(axum::body::Body::from(format!(
4453 "This page is at {target}\n"
4454 )))
4455 .unwrap_or_else(|_| {
4456 html_page(
4457 StatusCode::INTERNAL_SERVER_ERROR,
4458 "pitchfork",
4459 String::new(),
4460 )
4461 })
4462}
4463
4464fn page_placeholder_response(
4471 project: &str,
4472 worktree: Option<&str>,
4473 daemons: &[String],
4474 tld: &str,
4475 port_suffix: &str,
4476 web_url: Option<&str>,
4477) -> Response {
4478 let heading = match worktree {
4479 Some(wt) => format!("{} · {}", escape_html(project), escape_html(wt)),
4480 None => escape_html(project),
4481 };
4482 let suffix = match worktree {
4483 Some(wt) => format!(
4484 "{}.{}.{}",
4485 escape_html(wt),
4486 escape_html(project),
4487 escape_html(tld)
4488 ),
4489 None => format!("{}.{}", escape_html(project), escape_html(tld)),
4490 };
4491 let list = if daemons.is_empty() {
4492 "<p>No daemon in this checkout has a port configured.</p>".to_string()
4493 } else {
4494 let items: String = daemons
4495 .iter()
4496 .map(|d| {
4497 let d = escape_html(d);
4498 format!("<li><a href=\"//{d}.{suffix}{port_suffix}\">{d}.{suffix}</a></li>")
4499 })
4500 .collect();
4501 format!("<p>Daemons here:</p><ul>{items}</ul>")
4502 };
4503 let page = if worktree.is_some() {
4504 "stack"
4505 } else {
4506 "project"
4507 };
4508 let explanation = match web_url {
4513 Some(url) => format!(
4514 "<p>This address is reserved for the {page} page, which the web UI serves. No registered project covers this checkout, so it has no page yet: add its directory under <code>[namespaces]</code> in your user config, or run <code>pitchfork proxy add</code> from it. <a href=\"{url}/projects\">Open the project list</a>.</p>",
4515 url = escape_html(url),
4516 ),
4517 None => format!(
4518 "<p>This address is reserved for the {page} page, which the web UI serves. Enable it with <code>[settings.web] auto_start = true</code> to open this address.</p>"
4519 ),
4520 };
4521 let body = format!("<h1>{heading}</h1>{explanation}{list}");
4522 html_page(StatusCode::OK, "pitchfork", body)
4523}
4524
4525fn unknown_host_response(host: &str, heading: &str, known: &[String]) -> Response {
4527 let list = if known.is_empty() {
4528 "<p>Nothing is registered under this name yet.</p>".to_string()
4529 } else {
4530 let items: String = known
4531 .iter()
4532 .map(|k| format!("<li><code>{}</code></li>", escape_html(k)))
4533 .collect();
4534 format!("<p>Known names:</p><ul>{items}</ul>")
4535 };
4536 let body = format!(
4537 "<h1>{heading}</h1><p>No route for <code>{host}</code>.</p>{list}",
4538 heading = escape_html(heading),
4539 host = escape_html(host),
4540 );
4541 html_page(StatusCode::NOT_FOUND, "Not found", body)
4542}
4543
4544fn strip_tld(host: &str, tld: &str) -> Option<String> {
4559 strip_dot_suffix_ignore_case(host.trim_end_matches('.'), tld)
4560}
4561
4562fn bind_error_message(port: u16, err: &std::io::Error) -> String {
4564 if port < 1024 {
4565 format!(
4566 "Failed to bind proxy server to port {port}: {err}\n\
4567 Hint: ports below 1024 require elevated privileges. Run \
4568 `pitchfork proxy setup`, which grants the bind capability on Linux, \
4569 or set an unprivileged proxy.port and let setup redirect {port} to it."
4570 )
4571 } else {
4572 format!(
4573 "Failed to bind proxy server to port {port}: {err}\n\
4574 Hint: another process may already be using this port."
4575 )
4576 }
4577}
4578
4579fn starting_html_response(slug: &str, raw_host: &str) -> Response {
4584 let escaped_slug = slug
4585 .replace('&', "&")
4586 .replace('<', "<")
4587 .replace('>', ">")
4588 .replace('"', """)
4589 .replace('\'', "'");
4590 let escaped_host = raw_host
4591 .replace('&', "&")
4592 .replace('<', "<")
4593 .replace('>', ">")
4594 .replace('"', """)
4595 .replace('\'', "'");
4596
4597 let html = format!(
4598 r##"<!DOCTYPE html>
4599<html lang="en">
4600<head>
4601 <meta charset="UTF-8">
4602 <meta name="viewport" content="width=device-width, initial-scale=1">
4603 <meta http-equiv="refresh" content="2">
4604 <title>Starting {escaped_slug}… — pitchfork</title>
4605 <style>
4606 * {{ margin: 0; padding: 0; box-sizing: border-box; }}
4607 body {{
4608 font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
4609 background: #0f1117;
4610 color: #e1e4e8;
4611 display: flex;
4612 align-items: center;
4613 justify-content: center;
4614 min-height: 100vh;
4615 }}
4616 .container {{
4617 text-align: center;
4618 max-width: 480px;
4619 padding: 2rem;
4620 }}
4621 .spinner {{
4622 width: 48px;
4623 height: 48px;
4624 border: 4px solid rgba(255, 255, 255, 0.1);
4625 border-top-color: #58a6ff;
4626 border-radius: 50%;
4627 animation: spin 0.8s linear infinite;
4628 margin: 0 auto 1.5rem;
4629 }}
4630 @keyframes spin {{
4631 to {{ transform: rotate(360deg); }}
4632 }}
4633 h1 {{
4634 font-size: 1.5rem;
4635 font-weight: 600;
4636 margin-bottom: 0.5rem;
4637 }}
4638 .slug {{
4639 color: #58a6ff;
4640 font-family: "SFMono-Regular", Consolas, "Liberation Mono", Menlo, monospace;
4641 }}
4642 .host {{
4643 color: #8b949e;
4644 font-size: 0.875rem;
4645 margin-top: 0.25rem;
4646 }}
4647 .hint {{
4648 color: #8b949e;
4649 font-size: 0.8rem;
4650 margin-top: 1.5rem;
4651 }}
4652 </style>
4653</head>
4654<body>
4655 <div class="container">
4656 <div class="spinner"></div>
4657 <h1>Starting <span class="slug">{escaped_slug}</span>…</h1>
4658 <p class="host">{escaped_host}</p>
4659 <p class="hint">This page will refresh automatically when the daemon is ready.</p>
4660 </div>
4661</body>
4662</html>"##
4663 );
4664
4665 Response::builder()
4666 .status(StatusCode::SERVICE_UNAVAILABLE)
4667 .header("content-type", "text/html; charset=utf-8")
4668 .header("retry-after", "2")
4669 .body(Body::from(html))
4670 .unwrap_or_else(|_| (StatusCode::SERVICE_UNAVAILABLE, "Starting…").into_response())
4671}
4672
4673async fn redirect_to_https_handler(req: Request) -> Response {
4682 if req.headers().contains_key("upgrade") {
4684 log::warn!("Dropping plain-HTTP WebSocket upgrade attempt — use wss:// instead of ws://");
4685 return (
4686 StatusCode::BAD_REQUEST,
4687 "WebSocket over plain HTTP is not supported on the HTTPS port. Use wss:// instead.",
4688 )
4689 .into_response();
4690 }
4691
4692 let raw_host = get_request_host(&req);
4693 let Some(raw_host) = raw_host else {
4694 return (StatusCode::BAD_REQUEST, "Missing Host header").into_response();
4695 };
4696
4697 let hostname = if raw_host.starts_with('[') {
4699 raw_host
4701 .split_once("]:")
4702 .map(|(host, _)| host)
4703 .unwrap_or(&raw_host)
4704 .trim_start_matches('[')
4705 .trim_end_matches(']')
4706 } else {
4707 let mut parts = raw_host.rsplitn(2, ':');
4709 let last = parts.next().unwrap_or(&raw_host);
4710 parts.next().unwrap_or(last)
4711 };
4712
4713 let path = req
4714 .uri()
4715 .path_and_query()
4716 .map(|pq| pq.as_str())
4717 .unwrap_or("/");
4718
4719 let https_port = match u16::try_from(settings().proxy.port).ok().filter(|&p| p > 0) {
4720 Some(443) | None => String::new(),
4721 Some(port) => format!(":{port}"),
4722 };
4723
4724 let host_for_url = if raw_host.starts_with('[') {
4725 format!("[{hostname}]")
4726 } else {
4727 hostname.to_string()
4728 };
4729
4730 let location = format!("https://{host_for_url}{https_port}{path}");
4731 (
4732 StatusCode::FOUND,
4733 [
4734 (axum::http::header::LOCATION, location),
4735 (
4738 axum::http::HeaderName::from_static(PITCHFORK_HEADER),
4739 "1".to_string(),
4740 ),
4741 ],
4742 )
4743 .into_response()
4744}
4745
4746fn error_response(status: StatusCode, message: &str) -> Response {
4748 (
4752 status,
4753 [(
4754 axum::http::HeaderName::from_static(PITCHFORK_HEADER),
4755 HeaderValue::from_static("1"),
4756 )],
4757 message.to_string(),
4758 )
4759 .into_response()
4760}
4761
4762#[cfg(test)]
4763mod tests {
4764 use super::*;
4765
4766 #[tokio::test]
4770 async fn a_request_forwards_after_an_idle_stop_is_called_off() {
4771 let id = DaemonId::new("calledoff", "api");
4772 SUPERVISOR.state_file.lock().await.daemons.insert(
4773 id.clone(),
4774 crate::daemon::Daemon {
4775 id: id.clone(),
4776 status: crate::daemon_status::DaemonStatus::Running,
4777 ..Default::default()
4778 },
4779 );
4780 assert!(ACTIVITY.claim_idle_stop(&id, std::time::Duration::ZERO));
4781 let releaser = {
4782 let id = id.clone();
4783 tokio::spawn(async move {
4784 tokio::time::sleep(std::time::Duration::from_millis(300)).await;
4785 ACTIVITY.release_idle_stop(&id);
4786 })
4787 };
4788 assert!(matches!(
4789 begin_running(&id, auto_start_deadline()).await,
4790 Running::Yes(_)
4791 ));
4792 releaser.await.unwrap();
4793 SUPERVISOR.state_file.lock().await.daemons.remove(&id);
4794 }
4795
4796 #[tokio::test]
4799 async fn a_request_waits_out_an_idle_stop() {
4800 let id = DaemonId::new("waitproj", "api");
4801 assert!(ACTIVITY.claim_idle_stop(&id, std::time::Duration::ZERO));
4802 let releaser = {
4803 let id = id.clone();
4804 tokio::spawn(async move {
4805 tokio::time::sleep(std::time::Duration::from_millis(300)).await;
4806 ACTIVITY.release_idle_stop(&id);
4807 })
4808 };
4809 assert!(matches!(
4810 begin_running(&id, auto_start_deadline()).await,
4811 Running::No
4812 ));
4813 releaser.await.unwrap();
4814 assert!(ACTIVITY.begin(&id).is_some());
4815 }
4816
4817 #[tokio::test]
4823 async fn a_get_only_route_never_reaches_the_fallback() {
4824 async fn routed() -> &'static str {
4825 "routed"
4826 }
4827 async fn fell_through() -> &'static str {
4828 "fell-through"
4829 }
4830
4831 async fn post_to(app: Router) -> String {
4834 use tokio::io::{AsyncReadExt, AsyncWriteExt};
4835
4836 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4837 let addr = listener.local_addr().unwrap();
4838 let server = tokio::spawn(async move {
4839 let _ = axum::serve(listener, app).await;
4840 });
4841
4842 let mut sock = tokio::net::TcpStream::connect(addr).await.unwrap();
4843 sock.write_all(
4844 b"POST /proxy.pac HTTP/1.1\r\nHost: example\r\nConnection: close\r\n\r\n",
4845 )
4846 .await
4847 .unwrap();
4848 let mut raw = Vec::new();
4849 sock.read_to_end(&mut raw).await.unwrap();
4850 server.abort();
4851
4852 let text = String::from_utf8_lossy(&raw).into_owned();
4853 let status = text.split_whitespace().nth(1).unwrap_or("").to_string();
4854 if status == "405" {
4855 return status;
4856 }
4857 text.rsplit("\r\n").next().unwrap_or("").to_string()
4858 }
4859
4860 let get_only = Router::new()
4861 .route("/proxy.pac", axum::routing::get(routed))
4862 .fallback(fell_through);
4863 assert_eq!(
4864 post_to(get_only).await,
4865 "405",
4866 "axum reached the fallback on a method mismatch, so `any` is unnecessary"
4867 );
4868
4869 let any_method = Router::new()
4870 .route("/proxy.pac", axum::routing::any(routed))
4871 .fallback(fell_through);
4872 assert_eq!(
4873 post_to(any_method).await,
4874 "routed",
4875 "`any` did not deliver the POST to the handler"
4876 );
4877 }
4878
4879 #[cfg(feature = "proxy-tls")]
4880 #[test]
4881 fn a_ca_pair_is_checked_not_just_found() {
4882 let dir = tempfile::tempdir().unwrap();
4883 let (cert, key) = (dir.path().join("ca.pem"), dir.path().join("ca-key.pem"));
4884 generate_ca(&cert, &key).unwrap();
4885 assert_eq!(ca_pair_problem(&cert, &key), None);
4886
4887 let (other_cert, other_key) = (dir.path().join("b.pem"), dir.path().join("b-key.pem"));
4889 generate_ca(&other_cert, &other_key).unwrap();
4890 assert!(ca_pair_problem(&cert, &other_key).is_some());
4891
4892 std::fs::write(&cert, "-----BEGIN CERTIFICATE-----\nAAAA\n").unwrap();
4894 assert!(ca_pair_problem(&cert, &key).is_some());
4895 }
4896
4897 #[cfg(feature = "proxy-tls")]
4898 #[test]
4899 fn a_new_ca_clears_leaves_signed_by_the_old_one() {
4900 let dir = tempfile::tempdir().unwrap();
4903 let (cert, key) = (dir.path().join("ca.pem"), dir.path().join("ca-key.pem"));
4904 let host_certs = host_certs_dir_for(&cert);
4905 std::fs::create_dir_all(&host_certs).unwrap();
4906 std::fs::write(host_certs.join("api.localhost.pem"), "old leaf").unwrap();
4907 std::fs::write(host_certs.join("notes.txt"), "keep me").unwrap();
4908
4909 assert!(!ensure_ca(&cert, &key, || true).unwrap());
4911 assert!(host_certs.join("api.localhost.pem").exists());
4912
4913 assert!(ensure_ca(&cert, &key, || false).unwrap());
4915 assert!(!host_certs.join("api.localhost.pem").exists());
4916 assert!(host_certs.join("notes.txt").exists());
4917 }
4918
4919 #[cfg(feature = "proxy-tls")]
4920 #[test]
4921 fn leaves_are_cached_per_ca_so_a_replaced_ca_is_never_served() {
4922 let dir = tempfile::tempdir().unwrap();
4926 let (cert, key) = (dir.path().join("ca.pem"), dir.path().join("ca-key.pem"));
4927 let _ = rustls::crypto::ring::default_provider().install_default();
4928 ensure_ca(&cert, &key, || false).unwrap();
4929 let first = SniCertResolver::new(&cert, &key, "localhost".into()).unwrap();
4930 assert!(first.get_or_create_checked("api.localhost").is_some());
4931 let old_dir = first.host_certs_dir.clone();
4932 assert!(std::fs::read_dir(&old_dir).unwrap().count() > 0);
4933
4934 ensure_ca(&cert, &key, || false).unwrap();
4935 std::fs::create_dir_all(&old_dir).unwrap();
4937 std::fs::write(old_dir.join("late.localhost.pem"), "old leaf").unwrap();
4938
4939 let second = SniCertResolver::new(&cert, &key, "localhost".into()).unwrap();
4940 assert_ne!(second.host_certs_dir, old_dir);
4941 assert!(!old_dir.exists());
4943 }
4944
4945 #[cfg(feature = "proxy-tls")]
4946 #[test]
4947 fn cert_cache_file_names_do_not_collide() {
4948 assert_ne!(
4950 cert_cache_file_stem("a_b.localhost"),
4951 cert_cache_file_stem("a.b.localhost")
4952 );
4953 assert_eq!(cert_cache_file_stem("api.localhost"), "api.localhost");
4954 assert_eq!(cert_cache_file_stem("a_b.localhost"), "a%5Fb.localhost");
4955 assert_eq!(
4956 cert_cache_file_stem("*.proj.localhost"),
4957 "%2A.proj.localhost"
4958 );
4959 assert!(!cert_cache_file_stem("../x").contains('/'));
4961 }
4962
4963 #[test]
4964 fn connect_port_reads_the_authority() {
4965 assert_eq!(connect_port("api.localhost:443"), Some(443));
4966 assert_eq!(connect_port("api.localhost:80"), Some(80));
4967 assert_eq!(connect_port("[::1]:443"), Some(443));
4968 assert_eq!(connect_port("api.localhost"), None);
4969 assert_eq!(connect_port("::1"), None);
4970 }
4971
4972 #[test]
4973 fn a_half_configured_certificate_pair_is_refused() {
4974 assert!(tls_pair_problem("", "/k.pem").is_some());
4978 assert!(tls_pair_problem("/c.pem", "").is_some());
4979 assert!(tls_pair_problem("", "").is_none());
4980 assert!(tls_pair_problem("/c.pem", "/k.pem").is_none());
4981 }
4982
4983 #[cfg(feature = "proxy-tls")]
4985 fn test_resolver(tld: &str) -> (SniCertResolver, tempfile::TempDir) {
4986 let dir = tempfile::tempdir().unwrap();
4987 let cert = dir.path().join("ca.pem");
4988 let key = dir.path().join("ca-key.pem");
4989 generate_ca(&cert, &key).unwrap();
4990 let _ = rustls::crypto::ring::default_provider().install_default();
4991 (
4992 SniCertResolver::new(&cert, &key, tld.to_string()).unwrap(),
4993 dir,
4994 )
4995 }
4996
4997 #[cfg(feature = "proxy-tls")]
4999 fn sans_for(resolver: &SniCertResolver, domain: &str) -> Vec<String> {
5000 let ck = resolver.get_or_create(domain).expect("a certificate");
5001 let (_, cert) = x509_parser::parse_x509_certificate(&ck.cert[0]).unwrap();
5002 cert.subject_alternative_name()
5003 .unwrap()
5004 .map(|ext| {
5005 ext.value
5006 .general_names
5007 .iter()
5008 .filter_map(|n| match n {
5009 x509_parser::extensions::GeneralName::DNSName(d) => Some(d.to_string()),
5010 _ => None,
5011 })
5012 .collect()
5013 })
5014 .unwrap_or_default()
5015 }
5016
5017 #[cfg(feature = "proxy-tls")]
5018 #[test]
5019 fn the_ca_refuses_to_sign_for_a_name_outside_the_tld() {
5020 use rustls::server::ResolvesServerCert;
5021
5022 let (resolver, _dir) = test_resolver("localhost");
5023 let cache = resolver.host_certs_dir.clone();
5024
5025 for foreign in [
5029 "login.microsoftonline.com",
5030 "example.com",
5031 "notlocalhost",
5032 "localhost.evil.com",
5033 ] {
5034 assert!(
5035 resolver.get_or_create_checked(foreign).is_none(),
5036 "expected {foreign:?} to be refused"
5037 );
5038 }
5039
5040 let cached: Vec<String> = std::fs::read_dir(&cache)
5042 .map(|rd| {
5043 rd.filter_map(|e| e.ok())
5044 .map(|e| e.file_name().to_string_lossy().into_owned())
5045 .collect()
5046 })
5047 .unwrap_or_default();
5048 assert!(
5049 cached.is_empty(),
5050 "unexpected cached certificates: {cached:?}"
5051 );
5052
5053 for ours in ["localhost", "api.localhost", "core.wt.proj.localhost"] {
5055 assert!(
5056 resolver.get_or_create_checked(ours).is_some(),
5057 "expected {ours:?} to be issued"
5058 );
5059 }
5060 let _ = &resolver as &dyn ResolvesServerCert;
5062 }
5063
5064 #[cfg(feature = "proxy-tls")]
5065 #[test]
5066 fn the_certificate_cache_is_bounded_and_evicts_the_oldest() {
5067 let (resolver, _dir) = test_resolver("localhost");
5070 let host_certs = resolver.host_certs_dir.clone();
5071
5072 for i in 0..MAX_HOST_CERTS + 8 {
5073 assert!(
5074 resolver
5075 .get_or_create_checked(&format!("h{i}.localhost"))
5076 .is_some()
5077 );
5078 }
5079
5080 let cached = resolver.cache.lock().unwrap();
5081 assert_eq!(cached.by_domain.len(), MAX_HOST_CERTS);
5082 assert_eq!(cached.order.len(), MAX_HOST_CERTS);
5083 assert!(cached.get("h0.localhost").is_none());
5085 assert!(
5086 cached
5087 .get(&format!("h{}.localhost", MAX_HOST_CERTS + 7))
5088 .is_some()
5089 );
5090 drop(cached);
5091
5092 let on_disk = std::fs::read_dir(&host_certs).unwrap().count();
5094 assert!(
5095 on_disk <= MAX_HOST_CERTS,
5096 "{on_disk} files cached, expected at most {MAX_HOST_CERTS}"
5097 );
5098 }
5099
5100 #[cfg(feature = "proxy-tls")]
5101 #[test]
5102 fn the_disk_cache_is_pruned_at_startup() {
5103 let dir = tempfile::tempdir().unwrap();
5107 let host_certs = dir.path().join("host-certs");
5108 std::fs::create_dir_all(&host_certs).unwrap();
5109 for i in 0..MAX_HOST_CERTS + 20 {
5110 std::fs::write(host_certs.join(format!("old{i}.pem")), "stale").unwrap();
5111 }
5112 std::fs::write(host_certs.join("notes.txt"), "keep me").unwrap();
5114
5115 prune_host_certs(&host_certs);
5116
5117 let pems = std::fs::read_dir(&host_certs)
5118 .unwrap()
5119 .filter_map(|e| e.ok())
5120 .filter(|e| e.path().extension().is_some_and(|x| x == "pem"))
5121 .count();
5122 assert_eq!(pems, MAX_HOST_CERTS);
5123 assert!(host_certs.join("notes.txt").exists());
5124
5125 let small = dir.path().join("small");
5127 std::fs::create_dir_all(&small).unwrap();
5128 std::fs::write(small.join("a.pem"), "x").unwrap();
5129 prune_host_certs(&small);
5130 assert!(small.join("a.pem").exists());
5131 }
5132
5133 #[cfg(feature = "proxy-tls")]
5134 #[test]
5135 fn a_minted_certificate_never_wildcards_the_whole_tld() {
5136 let (resolver, _dir) = test_resolver("localhost");
5137
5138 let sans = sans_for(&resolver, "api.localhost");
5140 assert!(sans.contains(&"api.localhost".to_string()));
5141 assert!(
5142 !sans.iter().any(|s| s.starts_with('*')),
5143 "unexpected wildcard in {sans:?}"
5144 );
5145
5146 let sans = sans_for(&resolver, "core.wt.proj.localhost");
5148 assert!(sans.contains(&"*.wt.proj.localhost".to_string()));
5149
5150 let (resolver, _dir) = test_resolver("dev.internal");
5152 let sans = sans_for(&resolver, "api.dev.internal");
5153 assert!(
5154 !sans.iter().any(|s| s.starts_with('*')),
5155 "unexpected wildcard in {sans:?}"
5156 );
5157 }
5158
5159 #[tokio::test]
5163 async fn test_page_placeholder_explains_which_step_is_missing() {
5164 async fn body_of(response: Response) -> String {
5165 let (_, body) = response.into_parts();
5166 let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap();
5167 String::from_utf8(bytes.to_vec()).unwrap()
5168 }
5169
5170 let disabled = page_placeholder_response("shop", None, &[], "localhost", "", None);
5171 assert_eq!(disabled.status(), StatusCode::OK);
5172 let disabled = body_of(disabled).await;
5173 assert!(disabled.contains("auto_start"), "{disabled}");
5174 assert!(!disabled.contains("[namespaces]"));
5175
5176 let running = page_placeholder_response(
5177 "shop",
5178 Some("feature-a"),
5179 &[],
5180 "localhost",
5181 "",
5182 Some("http://127.0.0.1:3120"),
5183 );
5184 let running = body_of(running).await;
5185 assert!(running.contains("[namespaces]"), "{running}");
5186 assert!(running.contains("http://127.0.0.1:3120/projects"));
5187 assert!(!running.contains("auto_start"));
5188 }
5189
5190 #[test]
5193 fn test_page_redirect_targets_the_web_ui() {
5194 let response = page_redirect_response("http://127.0.0.1:3120", "/projects/shop/feature-a");
5195 assert_eq!(response.status(), StatusCode::FOUND);
5196 assert_eq!(
5197 response
5198 .headers()
5199 .get(axum::http::header::LOCATION)
5200 .unwrap(),
5201 "http://127.0.0.1:3120/projects/shop/feature-a"
5202 );
5203
5204 let project = page_redirect_response("http://127.0.0.1:3120/ps", "/projects/shop");
5205 assert_eq!(
5206 project.headers().get(axum::http::header::LOCATION).unwrap(),
5207 "http://127.0.0.1:3120/ps/projects/shop"
5209 );
5210 }
5211
5212 #[tokio::test]
5213 async fn overlapping_startup_graphs_wait_for_each_other() {
5214 use std::time::Duration;
5215 let id = |name: &str| DaemonId::try_new("startup-lock-test", name).unwrap();
5216
5217 let first = lock_startup_graph(vec![id("shared"), id("app-a")]).await;
5218
5219 let second = tokio::spawn(lock_startup_graph(vec![id("shared"), id("app-b")]));
5222 tokio::time::sleep(Duration::from_millis(50)).await;
5223 assert!(
5224 !second.is_finished(),
5225 "a graph sharing a daemon must wait for the graph starting it"
5226 );
5227
5228 let disjoint = tokio::time::timeout(
5230 Duration::from_secs(1),
5231 lock_startup_graph(vec![id("other")]),
5232 )
5233 .await;
5234 assert!(disjoint.is_ok());
5235
5236 drop(first);
5237 let second = tokio::time::timeout(Duration::from_secs(1), second)
5238 .await
5239 .expect("the waiting graph proceeds once the first releases")
5240 .unwrap();
5241 assert_eq!(second.len(), 2);
5242 drop(second);
5243 drop(disjoint);
5244
5245 drop(lock_startup_graph(vec![id("unrelated")]).await);
5247 let map = STARTUP_LOCKS.lock().unwrap();
5248 assert!(
5249 !map.keys()
5250 .any(|k| k.namespace() == "startup-lock-test" && k.name() != "unrelated"),
5251 "released startup locks must be pruned"
5252 );
5253 }
5254
5255 #[test]
5256 fn test_strip_tld() {
5257 assert_eq!(
5258 strip_tld("api.myproject.localhost", "localhost"),
5259 Some("api.myproject".to_string())
5260 );
5261 assert_eq!(
5263 strip_tld("API.MyProject.LOCALHOST", "localhost"),
5264 Some("API.MyProject".to_string())
5265 );
5266 assert_eq!(
5267 strip_tld("api.localhost", "LOCALHOST"),
5268 Some("api".to_string())
5269 );
5270 assert_eq!(
5271 strip_tld("api.localhost", "localhost"),
5272 Some("api".to_string())
5273 );
5274 assert_eq!(strip_tld("localhost", "localhost"), None);
5275 assert_eq!(
5280 strip_tld("API.LocalHost", "localhost"),
5281 Some("API".to_string())
5282 );
5283 assert_eq!(
5284 strip_tld("api.localhost.", "localhost"),
5285 Some("api".to_string())
5286 );
5287 assert_eq!(
5288 strip_tld("API.MyProject.LOCALHOST.", "localhost"),
5289 Some("API.MyProject".to_string())
5290 );
5291 assert_eq!(strip_tld("localhost.", "localhost"), None);
5292 assert_eq!(
5293 strip_tld("api.myproject.test", "test"),
5294 Some("api.myproject".to_string())
5295 );
5296 assert_eq!(strip_tld("other.com", "localhost"), None);
5297 }
5298
5299 fn make_entry(name: &str) -> CachedSlugEntry {
5300 CachedSlugEntry {
5301 slug: name.to_string(),
5302 namespace: None,
5303 daemon_name: name.to_string(),
5304 dir: std::path::PathBuf::from(format!("/tmp/{name}")),
5305 worktrees: vec![],
5306 rejected_worktree_prefixes: std::collections::HashSet::new(),
5307 tls: ProxyTlsRoute::default(),
5308 worktree_tls: std::collections::HashMap::new(),
5309 }
5310 }
5311
5312 fn make_daemon(
5314 configured: &[u16],
5315 resolved: &[u16],
5316 active: Option<u16>,
5317 ) -> crate::daemon::Daemon {
5318 crate::daemon::Daemon {
5319 id: DaemonId::try_new("proj", "api").unwrap(),
5320 port: crate::config_types::PortConfig::from_parts(
5321 configured.to_vec(),
5322 crate::config_types::PortBump(0),
5323 ),
5324 resolved_port: resolved.to_vec(),
5325 active_port: active,
5326 ..crate::daemon::Daemon::default()
5327 }
5328 }
5329
5330 #[test]
5333 fn test_select_daemon_port_prefers_active_port() {
5334 let route = ProxyTlsRoute::default();
5335 let d = make_daemon(&[8443, 9443], &[8443, 9443], Some(8443));
5336 assert_eq!(select_daemon_port(&route, &d), Some(8443));
5337 }
5338
5339 #[test]
5343 fn test_select_daemon_port_skips_port_zero() {
5344 for mode in [ProxyTlsMode::Passthrough, ProxyTlsMode::Terminate] {
5345 let route = ProxyTlsRoute {
5346 mode,
5347 protocol: Default::default(),
5348 port: None,
5349 };
5350
5351 let mixed = make_daemon(&[0, 8443], &[0, 8443], None);
5353 assert_eq!(select_daemon_port(&route, &mixed), Some(8443));
5354
5355 let detected_placeholder = make_daemon(&[0, 8443], &[0, 8443], Some(0));
5358 assert_eq!(
5359 select_daemon_port(&route, &detected_placeholder),
5360 Some(8443)
5361 );
5362
5363 let unresolved = make_daemon(&[0], &[0], Some(0));
5365 assert_eq!(select_daemon_port(&route, &unresolved), None);
5366 }
5367 }
5368
5369 #[test]
5374 fn test_select_daemon_port_passthrough_prefers_declared_first_port() {
5375 let route = ProxyTlsRoute {
5376 protocol: Default::default(),
5377 mode: ProxyTlsMode::Passthrough,
5378 port: None,
5379 };
5380 let d = make_daemon(&[8443, 9080], &[8443, 9080], Some(9080));
5381 assert_eq!(select_daemon_port(&route, &d), Some(8443));
5382
5383 let detected_only = make_daemon(&[], &[], Some(9080));
5386 assert_eq!(select_daemon_port(&route, &detected_only), Some(9080));
5387 }
5388
5389 #[test]
5391 fn test_select_daemon_port_falls_back_to_first_resolved() {
5392 let route = ProxyTlsRoute::default();
5393 let d = make_daemon(&[8443, 9443], &[8443, 9443], None);
5394 assert_eq!(select_daemon_port(&route, &d), Some(8443));
5395
5396 let none = make_daemon(&[], &[], None);
5398 assert_eq!(select_daemon_port(&route, &none), None);
5399 }
5400
5401 #[test]
5404 fn test_select_daemon_port_honors_configured_port() {
5405 let route = ProxyTlsRoute {
5406 protocol: Default::default(),
5407 mode: ProxyTlsMode::Passthrough,
5408 port: Some(9443),
5409 };
5410 let d = make_daemon(&[8443, 9443], &[8443, 9443], Some(8443));
5411 assert_eq!(select_daemon_port(&route, &d), Some(9443));
5412 }
5413
5414 #[test]
5418 fn test_select_daemon_port_follows_auto_bump() {
5419 let route = ProxyTlsRoute {
5420 protocol: Default::default(),
5421 mode: ProxyTlsMode::Passthrough,
5422 port: Some(9443),
5423 };
5424 let d = make_daemon(&[8443, 9443], &[8444, 9444], Some(8444));
5425 assert_eq!(select_daemon_port(&route, &d), Some(9444));
5426 }
5427
5428 #[test]
5432 fn test_select_daemon_port_skips_a_configured_port_resolved_to_zero() {
5433 let route = ProxyTlsRoute {
5434 protocol: Default::default(),
5435 mode: ProxyTlsMode::Passthrough,
5436 port: Some(9443),
5437 };
5438 let pending = make_daemon(&[8443, 9443], &[8443, 0], Some(8443));
5441 assert_eq!(select_daemon_port(&route, &pending), None);
5442
5443 let ready = make_daemon(&[8443, 9443], &[8443, 9444], Some(8443));
5445 assert_eq!(select_daemon_port(&route, &ready), Some(9444));
5446 }
5447
5448 #[test]
5453 fn test_select_daemon_port_refuses_a_port_the_daemon_never_bound() {
5454 let route = ProxyTlsRoute {
5455 protocol: Default::default(),
5456 mode: ProxyTlsMode::Passthrough,
5457 port: Some(9443),
5458 };
5459 let stale = make_daemon(&[8443], &[8443], Some(8443));
5460 assert_eq!(select_daemon_port(&route, &stale), None);
5461
5462 let bare = make_daemon(&[], &[], None);
5465 assert_eq!(select_daemon_port(&route, &bare), Some(9443));
5466 }
5467
5468 #[test]
5471 fn test_select_daemon_port_honors_configured_port_when_terminating() {
5472 let route = ProxyTlsRoute {
5473 protocol: Default::default(),
5474 mode: ProxyTlsMode::Terminate,
5475 port: Some(9080),
5476 };
5477 let d = make_daemon(&[8080, 9080], &[8080, 9080], Some(8080));
5478 assert_eq!(select_daemon_port(&route, &d), Some(9080));
5479
5480 let bumped = make_daemon(&[8080, 9080], &[8081, 9081], Some(8081));
5482 assert_eq!(select_daemon_port(&route, &bumped), Some(9081));
5483 }
5484
5485 #[test]
5487 fn test_read_proxy_tls_route_reads_protocol() {
5488 let dir = tempfile::tempdir().unwrap();
5489 std::fs::write(
5490 dir.path().join("pitchfork.toml"),
5491 "[daemons.grpc]\nrun = \"serve\"\nport = 50051\nproxy_protocol = \"h2c\"\n\n\
5492 [daemons.web]\nrun = \"serve\"\nport = 3000\n",
5493 )
5494 .unwrap();
5495 let ns = crate::pitchfork_toml::PitchforkToml::namespace_for_dir(dir.path()).unwrap();
5496 let grpc = read_proxy_tls_route(dir.path(), Some(&ns), "grpc")
5497 .unwrap()
5498 .unwrap();
5499 assert_eq!(grpc.protocol, ProxyProtocol::H2c);
5500 let web = read_proxy_tls_route(dir.path(), Some(&ns), "web")
5501 .unwrap()
5502 .unwrap();
5503 assert_eq!(web.protocol, ProxyProtocol::Http1);
5504 }
5505
5506 #[tokio::test]
5509 async fn test_h2c_client_reaches_http2_only_daemon() {
5510 use http_body_util::BodyExt;
5511 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
5512 let port = listener.local_addr().unwrap().port();
5513 tokio::spawn(async move {
5514 let (stream, _) = listener.accept().await.unwrap();
5515 let svc = hyper::service::service_fn(
5516 |req: axum::http::Request<hyper::body::Incoming>| async move {
5517 assert_eq!(req.version(), axum::http::Version::HTTP_2);
5518 let body = http_body_util::Full::new(axum::body::Bytes::from_static(b"ok"));
5519 Ok::<_, std::convert::Infallible>(axum::http::Response::new(body))
5520 },
5521 );
5522 let _ = hyper::server::conn::http2::Builder::new(TokioExecutor::new())
5523 .serve_connection(hyper_util::rt::TokioIo::new(stream), svc)
5524 .await;
5525 });
5526 let client: Client<HttpConnector, Body> = Client::builder(TokioExecutor::new())
5527 .http2_only(true)
5528 .build(HttpConnector::new());
5529 let mut req = axum::http::Request::new(Body::empty());
5530 *req.uri_mut() = format!("http://127.0.0.1:{port}/svc/Method")
5531 .parse()
5532 .unwrap();
5533 *req.version_mut() = axum::http::Version::HTTP_2;
5534 let resp = client.request(req).await.unwrap();
5535 assert_eq!(resp.version(), axum::http::Version::HTTP_2);
5536 let collected = resp.into_body().collect().await.unwrap();
5537 assert_eq!(collected.to_bytes(), "ok");
5538 }
5539
5540 #[test]
5544 fn test_read_proxy_tls_route_absent_without_config() {
5545 let dir = tempfile::tempdir().unwrap();
5546
5547 assert_eq!(
5549 read_proxy_tls_route(dir.path(), Some("proj"), "api").unwrap(),
5550 None
5551 );
5552
5553 std::fs::write(
5555 dir.path().join("pitchfork.toml"),
5556 "[daemons.other]\nrun = \"serve\"\n",
5557 )
5558 .unwrap();
5559 assert_eq!(
5560 read_proxy_tls_route(dir.path(), Some("proj"), "api").unwrap(),
5561 None,
5562 "a config without this daemon says nothing about it"
5563 );
5564
5565 assert_eq!(read_proxy_tls_route(dir.path(), None, "api").unwrap(), None);
5567 }
5568
5569 #[test]
5573 fn test_unreadable_config_keeps_the_last_known_route() {
5574 let dir = tempfile::tempdir().unwrap();
5575 std::fs::write(
5576 dir.path().join("pitchfork.toml"),
5577 "[daemons.api]\nrun = \"serve\"\nport = 8443\nproxy_tls = \"passthrough\"\n\n\
5578 [daemons.broken]\nrun = \"serve\"\nproxy_tls = \"passthrough\"\n",
5579 )
5580 .unwrap();
5581 let read = read_proxy_tls_route(dir.path(), Some("proj"), "api");
5582 assert!(
5583 read.is_err(),
5584 "an invalid sibling makes the config unreadable"
5585 );
5586
5587 let known = ProxyTlsRoute {
5588 protocol: Default::default(),
5589 mode: ProxyTlsMode::Passthrough,
5590 port: Some(8443),
5591 };
5592 assert_eq!(
5593 route_or_last_known(read, Some(known), dir.path(), "api"),
5594 Some(known)
5595 );
5596
5597 assert_eq!(
5599 route_or_last_known(Ok(None), Some(known), dir.path(), "api"),
5600 None
5601 );
5602 }
5603
5604 #[test]
5607 fn test_known_route_requires_the_same_target() {
5608 let known = ProxyTlsRoute {
5609 protocol: Default::default(),
5610 mode: ProxyTlsMode::Passthrough,
5611 port: Some(8443),
5612 };
5613 let mut entry = make_entry("api");
5614 entry.namespace = Some("proj".to_string());
5615 entry.tls = known;
5616 let dir = entry.dir.clone();
5617
5618 assert_eq!(entry.known_route(&dir, Some("proj"), "api"), Some(known));
5619 assert_eq!(entry.known_route(&dir, Some("proj"), "web"), None);
5620 assert_eq!(entry.known_route(&dir, Some("other"), "api"), None);
5621 assert_eq!(
5622 entry.known_route(std::path::Path::new("/elsewhere"), Some("proj"), "api"),
5623 None
5624 );
5625
5626 let wt = make_worktree("feature/x", "feature-x");
5627 entry.worktrees = vec![wt.clone()];
5628 entry.worktree_tls.insert("feature-x".to_string(), known);
5629 assert_eq!(entry.known_worktree_route(&wt, "api"), Some(known));
5630 assert_eq!(entry.known_worktree_route(&wt, "web"), None);
5631 let moved = crate::proxy::worktree::WorktreeEntry {
5632 path: std::path::PathBuf::from("/elsewhere/feature-x"),
5633 ..wt
5634 };
5635 assert_eq!(entry.known_worktree_route(&moved, "api"), None);
5636 }
5637
5638 #[test]
5642 fn test_worktree_route_inherits_when_unknown() {
5643 let mut entry = make_entry("spliced");
5644 entry.tls = ProxyTlsRoute {
5645 protocol: Default::default(),
5646 mode: ProxyTlsMode::Passthrough,
5647 port: Some(8443),
5648 };
5649 entry.worktrees = vec![
5650 make_worktree("feature/known", "feature-known"),
5651 make_worktree("feature/unknown", "feature-unknown"),
5652 ];
5653 entry.worktree_tls.insert(
5655 "feature-known".to_string(),
5656 ProxyTlsRoute {
5657 protocol: Default::default(),
5658 mode: ProxyTlsMode::Terminate,
5659 port: None,
5660 },
5661 );
5662 let mut entries = std::collections::HashMap::new();
5663 entries.insert("spliced".to_string(), entry);
5664
5665 let mode =
5666 |host: &str| resolve_tls_mode_in(host, "localhost", &entries, &Default::default());
5667 assert_eq!(
5668 mode("feature-known.spliced.localhost"),
5669 ProxyTlsMode::Terminate,
5670 "an explicit worktree setting wins"
5671 );
5672 assert_eq!(
5673 mode("feature-unknown.spliced.localhost"),
5674 ProxyTlsMode::Passthrough,
5675 "a worktree with nothing recorded inherits the slug"
5676 );
5677
5678 let cached = entries.get("spliced").unwrap();
5682 assert_eq!(
5683 worktree_route(cached, "feature-unknown"),
5684 ProxyTlsRoute {
5685 protocol: Default::default(),
5686 mode: ProxyTlsMode::Passthrough,
5687 port: None,
5688 }
5689 );
5690 assert_eq!(
5692 worktree_route(cached, "feature-known"),
5693 ProxyTlsRoute {
5694 protocol: Default::default(),
5695 mode: ProxyTlsMode::Terminate,
5696 port: None,
5697 }
5698 );
5699 }
5700
5701 #[test]
5704 fn test_worktree_route_lookup() {
5705 let mut entry = make_entry("myapp");
5706 entry.tls = ProxyTlsRoute {
5707 protocol: Default::default(),
5708 mode: ProxyTlsMode::Terminate,
5709 port: None,
5710 };
5711 entry.worktrees = vec![
5712 make_worktree("feature/b", "feature-b"),
5713 make_worktree("feature/c", "feature-c"),
5714 ];
5715 entry.worktree_tls.insert(
5716 "feature-b".to_string(),
5717 ProxyTlsRoute {
5718 protocol: Default::default(),
5719 mode: ProxyTlsMode::Passthrough,
5720 port: Some(9443),
5721 },
5722 );
5723
5724 let lookup = |prefix: &str| match match_worktree_prefix(&entry, prefix) {
5725 PrefixMatch::Worktree(wt) => entry
5726 .worktree_tls
5727 .get(&wt.sanitized_branch.to_ascii_lowercase())
5728 .copied()
5729 .unwrap_or(entry.tls),
5730 _ => entry.tls,
5731 };
5732
5733 assert_eq!(lookup("feature-b").mode, ProxyTlsMode::Passthrough);
5734 assert_eq!(lookup("feature-b").port, Some(9443));
5735 assert_eq!(lookup("feature-c").mode, ProxyTlsMode::Terminate);
5737 assert_eq!(lookup("tenant").mode, ProxyTlsMode::Terminate);
5739 }
5740
5741 #[test]
5742 fn test_wildcard_slug_lookup_exact_match() {
5743 let mut entries = std::collections::HashMap::new();
5744 entries.insert("myapp".to_string(), make_entry("myapp"));
5745 let result = wildcard_slug_lookup("myapp", &entries, true);
5747 assert!(result.is_some());
5748 assert_eq!(result.unwrap().daemon_name, "myapp");
5749 }
5750
5751 #[test]
5752 fn test_wildcard_slug_lookup_subdomain_fallback() {
5753 let mut entries = std::collections::HashMap::new();
5754 entries.insert("myapp".to_string(), make_entry("myapp"));
5755 let result = wildcard_slug_lookup("tenant.myapp", &entries, true);
5757 assert!(result.is_some());
5758 assert_eq!(result.unwrap().daemon_name, "myapp");
5759 }
5760
5761 #[test]
5762 fn test_wildcard_slug_lookup_nested_fallback() {
5763 let mut entries = std::collections::HashMap::new();
5764 entries.insert("myapp".to_string(), make_entry("myapp"));
5765 let result = wildcard_slug_lookup("a.b.myapp", &entries, true);
5767 assert!(result.is_some());
5768 assert_eq!(result.unwrap().daemon_name, "myapp");
5769 }
5770
5771 #[test]
5772 fn test_wildcard_slug_lookup_no_match() {
5773 let entries = std::collections::HashMap::new();
5774 let result = wildcard_slug_lookup("tenant.myapp", &entries, true);
5776 assert!(result.is_none());
5777 }
5778
5779 #[test]
5780 fn test_wildcard_slug_lookup_disabled() {
5781 let mut entries = std::collections::HashMap::new();
5782 entries.insert("myapp".to_string(), make_entry("myapp"));
5783 let result = wildcard_slug_lookup("tenant.myapp", &entries, false);
5785 assert!(result.is_none());
5786 let result = wildcard_slug_lookup("myapp", &entries, false);
5788 assert!(result.is_some());
5789 }
5790
5791 #[test]
5792 fn test_wildcard_slug_lookup_exact_beats_wildcard() {
5793 let mut entries = std::collections::HashMap::new();
5794 entries.insert("myapp".to_string(), make_entry("myapp"));
5795 let mut tenant_entry = make_entry("tenant-daemon");
5796 tenant_entry.slug = "tenant.myapp".to_string();
5797 entries.insert("tenant.myapp".to_string(), tenant_entry);
5798 let result = wildcard_slug_lookup("tenant.myapp", &entries, true);
5800 assert!(result.is_some());
5801 assert_eq!(result.unwrap().daemon_name, "tenant-daemon");
5802 }
5803
5804 #[test]
5805 fn test_wildcard_slug_lookup_ignores_case() {
5806 let mut entries = std::collections::HashMap::new();
5807 entries.insert("myapp".to_string(), make_entry("myapp"));
5808 for host in ["MyApp", "MYAPP", "myapp"] {
5810 let result = wildcard_slug_lookup(host, &entries, true);
5811 assert!(result.is_some(), "exact lookup failed for {host}");
5812 assert_eq!(result.unwrap().daemon_name, "myapp");
5813 }
5814 for host in ["Tenant.MyApp", "tenant.MYAPP", "A.B.MyApp"] {
5816 let result = wildcard_slug_lookup(host, &entries, true);
5817 assert!(result.is_some(), "wildcard lookup failed for {host}");
5818 assert_eq!(result.unwrap().daemon_name, "myapp");
5819 }
5820 }
5821
5822 #[test]
5823 fn test_wildcard_slug_lookup_case_insensitive_registration() {
5824 let mut entries = std::collections::HashMap::new();
5825 let mut entry = make_entry("upper");
5826 entry.slug = "MyApp".to_string();
5827 entries.insert("myapp".to_string(), entry);
5830 for host in ["myapp", "MyApp", "tenant.MYAPP"] {
5831 let result = wildcard_slug_lookup(host, &entries, true);
5832 assert!(result.is_some(), "lookup failed for {host}");
5833 assert_eq!(result.unwrap().daemon_name, "upper");
5834 }
5835 }
5836
5837 fn make_worktree(branch: &str, sanitized: &str) -> crate::proxy::worktree::WorktreeEntry {
5838 crate::proxy::worktree::WorktreeEntry {
5839 path: std::path::PathBuf::from(format!("/tmp/{sanitized}")),
5840 branch: branch.to_string(),
5841 sanitized_branch: sanitized.to_string(),
5842 namespace: Some(sanitized.to_string()),
5843 }
5844 }
5845
5846 #[test]
5847 fn test_reject_case_colliding_worktrees_drops_both_sides() {
5848 let wts = vec![
5849 make_worktree("Feature-A", "Feature-A"),
5850 make_worktree("feature-a", "feature-a"),
5851 make_worktree("main", "main"),
5852 ];
5853 let (kept, rejected) = reject_case_colliding_worktrees(wts);
5854 assert_eq!(kept.len(), 1);
5857 assert_eq!(kept[0].sanitized_branch, "main");
5858 assert!(rejected.contains("feature-a"));
5861 }
5862
5863 #[test]
5864 fn test_reject_case_colliding_worktrees_keeps_unambiguous() {
5865 let wts = vec![
5866 make_worktree("main", "main"),
5867 make_worktree("feature/a", "feature-a"),
5868 ];
5869 let (kept, rejected) = reject_case_colliding_worktrees(wts);
5870 assert_eq!(kept.len(), 2);
5871 assert!(rejected.is_empty());
5872 }
5873
5874 #[test]
5875 fn test_reject_case_colliding_worktrees_drops_sanitize_duplicates() {
5876 let wts = vec![
5879 make_worktree("feature/a", "feature-a"),
5880 make_worktree("feature.a", "feature-a"),
5881 ];
5882 let (kept, rejected) = reject_case_colliding_worktrees(wts);
5883 assert!(kept.is_empty());
5884 assert!(rejected.contains("feature-a"));
5885 }
5886
5887 #[test]
5888 fn test_match_worktree_prefix() {
5889 let mut entry = make_entry("myapp");
5890 entry.worktrees = vec![make_worktree("feature/b", "feature-b")];
5891 entry
5892 .rejected_worktree_prefixes
5893 .insert("feature-a".to_string());
5894
5895 assert!(matches!(
5896 match_worktree_prefix(&entry, "feature-b"),
5897 PrefixMatch::Worktree(_)
5898 ));
5899 assert!(matches!(
5901 match_worktree_prefix(&entry, "Feature-B"),
5902 PrefixMatch::Worktree(_)
5903 ));
5904 assert!(matches!(
5906 match_worktree_prefix(&entry, "feature-a"),
5907 PrefixMatch::Ambiguous
5908 ));
5909 assert!(matches!(
5910 match_worktree_prefix(&entry, "FEATURE-A"),
5911 PrefixMatch::Ambiguous
5912 ));
5913 assert!(matches!(
5915 match_worktree_prefix(&entry, "tenant"),
5916 PrefixMatch::Unknown
5917 ));
5918 }
5919
5920 #[test]
5923 fn test_passthrough_unroutable_message() {
5924 let no_tls = passthrough_unroutable_message("api.localhost", false);
5925 assert!(no_tls.contains("api.localhost"), "{no_tls}");
5926 assert!(no_tls.contains("settings.proxy.https = true"), "{no_tls}");
5927
5928 let mismatch = passthrough_unroutable_message("api.localhost", true);
5932 assert!(mismatch.contains("named a different host"), "{mismatch}");
5933 assert!(
5934 !mismatch.contains("settings.proxy.https"),
5935 "a host mismatch is not an HTTPS configuration problem: {mismatch}"
5936 );
5937 }
5938
5939 #[tokio::test]
5943 async fn test_slug_snapshot_is_the_cached_table() {
5944 let from_async = get_cached_slugs().await;
5945 let from_sync = slug_snapshot();
5946 assert!(
5947 Arc::ptr_eq(&from_async, &from_sync),
5948 "the synchronous read must see the same table routing does"
5949 );
5950 }
5951
5952 #[tokio::test]
5957 async fn test_concurrent_refresh_returns_the_published_table() {
5958 let (first, second) = tokio::join!(get_cached_slugs(), get_cached_slugs());
5961 assert!(
5962 Arc::ptr_eq(&first, &second),
5963 "overlapping refreshes must agree on one table"
5964 );
5965 assert!(
5966 Arc::ptr_eq(&first, &slug_snapshot()),
5967 "and it must be the published one"
5968 );
5969 }
5970
5971 #[test]
5975 fn test_resolve_tls_mode_in() {
5976 let mut entries = std::collections::HashMap::new();
5977
5978 let mut spliced = make_entry("spliced");
5979 spliced.tls = ProxyTlsRoute {
5980 protocol: Default::default(),
5981 mode: ProxyTlsMode::Passthrough,
5982 port: Some(8443),
5983 };
5984 spliced.worktrees = vec![
5985 make_worktree("feature/b", "feature-b"),
5986 make_worktree("feature/c", "feature-c"),
5987 ];
5988 spliced.worktree_tls.insert(
5989 "feature-b".to_string(),
5990 ProxyTlsRoute {
5991 protocol: Default::default(),
5992 mode: ProxyTlsMode::Terminate,
5993 port: None,
5994 },
5995 );
5996 entries.insert("spliced".to_string(), spliced);
5997 entries.insert("plain".to_string(), make_entry("plain"));
5998
5999 let mode =
6000 |host: &str| resolve_tls_mode_in(host, "localhost", &entries, &Default::default());
6001
6002 assert_eq!(mode("spliced.localhost"), ProxyTlsMode::Passthrough);
6003 assert_eq!(mode("SPLICED.localhost"), ProxyTlsMode::Passthrough);
6007 assert_eq!(mode("Spliced.LocalHost"), ProxyTlsMode::Passthrough);
6008 assert_eq!(mode("spliced.localhost."), ProxyTlsMode::Passthrough);
6009 assert_eq!(
6010 mode("FEATURE-C.Spliced.LOCALHOST."),
6011 ProxyTlsMode::Passthrough
6012 );
6013 assert_eq!(mode("tenant.spliced.localhost"), ProxyTlsMode::Passthrough);
6015 assert_eq!(mode("feature-b.spliced.localhost"), ProxyTlsMode::Terminate);
6017 assert_eq!(
6019 mode("feature-c.spliced.localhost"),
6020 ProxyTlsMode::Passthrough
6021 );
6022 assert_eq!(mode("plain.localhost"), ProxyTlsMode::Terminate);
6025 assert_eq!(mode("unknown.localhost"), ProxyTlsMode::Terminate);
6026 assert_eq!(mode("localhost"), ProxyTlsMode::Terminate);
6027 assert_eq!(mode("spliced.example.com"), ProxyTlsMode::Terminate);
6028 }
6029
6030 #[test]
6035 fn test_resolve_tls_mode_in_uses_the_hostname_registry() {
6036 let dir = tempfile::tempdir().unwrap();
6037 let project = dir.path().join("autoproj");
6038 std::fs::create_dir_all(&project).unwrap();
6039 std::fs::write(
6040 project.join("pitchfork.toml"),
6041 "[daemons.secure]\nrun = \"serve\"\nport = 8443\nproxy_tls = \"passthrough\"\n\
6042 [daemons.plain]\nrun = \"serve\"\nport = 8080\n",
6043 )
6044 .unwrap();
6045
6046 let registry =
6047 crate::proxy::hostname::HostRegistry::from_dirs(std::slice::from_ref(&project));
6048 let slugs = std::collections::HashMap::new();
6049 let mode = |host: &str| resolve_tls_mode_in(host, "localhost", &slugs, ®istry);
6050
6051 assert_eq!(mode("secure.autoproj.localhost"), ProxyTlsMode::Passthrough);
6052 assert_eq!(mode("plain.autoproj.localhost"), ProxyTlsMode::Terminate);
6053
6054 std::fs::write(project.join("pitchfork.toml"), "this is not toml = [\n").unwrap();
6058 assert_eq!(mode("secure.autoproj.localhost"), ProxyTlsMode::Passthrough);
6059 assert_eq!(mode("autoproj.localhost"), ProxyTlsMode::Terminate);
6062 assert_eq!(mode("nothing.autoproj.localhost"), ProxyTlsMode::Terminate);
6063 assert_eq!(mode("unknown.localhost"), ProxyTlsMode::Terminate);
6064 }
6065
6066 #[test]
6069 fn test_resolve_tls_mode_in_empty_table() {
6070 let entries = std::collections::HashMap::new();
6071 assert_eq!(
6072 resolve_tls_mode_in(
6073 "spliced.localhost",
6074 "localhost",
6075 &entries,
6076 &Default::default()
6077 ),
6078 ProxyTlsMode::Terminate
6079 );
6080 }
6081
6082 #[test]
6083 fn test_strip_dot_suffix_ignore_case() {
6084 assert_eq!(
6085 strip_dot_suffix_ignore_case("feature-a.myapp", "myapp"),
6086 Some("feature-a".to_string())
6087 );
6088 assert_eq!(
6089 strip_dot_suffix_ignore_case("Feature-A.MyApp", "myapp"),
6090 Some("Feature-A".to_string())
6091 );
6092 assert_eq!(
6093 strip_dot_suffix_ignore_case("feature-a.myapp", "MYAPP"),
6094 Some("feature-a".to_string())
6095 );
6096 assert_eq!(strip_dot_suffix_ignore_case("xmyapp", "myapp"), None);
6098 assert_eq!(strip_dot_suffix_ignore_case(".myapp", "myapp"), None);
6099 assert_eq!(strip_dot_suffix_ignore_case("myapp", "myapp"), None);
6100 assert_eq!(
6101 strip_dot_suffix_ignore_case("feature-a.other", "myapp"),
6102 None
6103 );
6104 assert_eq!(
6106 strip_dot_suffix_ignore_case("café.myapp", "myapp"),
6107 Some("café".to_string())
6108 );
6109 assert_eq!(strip_dot_suffix_ignore_case("café", "afé"), None);
6110 }
6111
6112 #[cfg(feature = "proxy-tls")]
6114 fn client_hello_wire(host: &str) -> Vec<u8> {
6115 let mut entry = vec![0u8];
6116 entry.extend_from_slice(&(host.len() as u16).to_be_bytes());
6117 entry.extend_from_slice(host.as_bytes());
6118 let mut sni = (entry.len() as u16).to_be_bytes().to_vec();
6119 sni.extend_from_slice(&entry);
6120
6121 let mut ext = vec![0x00, 0x00];
6122 ext.extend_from_slice(&(sni.len() as u16).to_be_bytes());
6123 ext.extend_from_slice(&sni);
6124
6125 let mut body = vec![0x03, 0x03];
6126 body.extend_from_slice(&[0x22; 32]);
6127 body.push(0);
6128 body.extend_from_slice(&[0x00, 0x02, 0x13, 0x01]);
6129 body.extend_from_slice(&[0x01, 0x00]);
6130 body.extend_from_slice(&(ext.len() as u16).to_be_bytes());
6131 body.extend_from_slice(&ext);
6132
6133 let mut msg = vec![0x01];
6134 let len = body.len() as u32;
6135 msg.extend_from_slice(&[(len >> 16) as u8, (len >> 8) as u8, len as u8]);
6136 msg.extend_from_slice(&body);
6137
6138 let mut record = vec![0x16, 0x03, 0x01];
6139 record.extend_from_slice(&(msg.len() as u16).to_be_bytes());
6140 record.extend_from_slice(&msg);
6141 record
6142 }
6143
6144 #[cfg(feature = "proxy-tls")]
6147 async fn probe_over_socket(
6148 writes: Vec<Vec<u8>>,
6149 gap: std::time::Duration,
6150 timeout: std::time::Duration,
6151 ) -> (SniProbe, Vec<u8>) {
6152 probe_over_socket_then(writes, gap, timeout, false).await
6153 }
6154
6155 #[cfg(feature = "proxy-tls")]
6158 async fn probe_over_socket_then(
6159 writes: Vec<Vec<u8>>,
6160 gap: std::time::Duration,
6161 timeout: std::time::Duration,
6162 close_after: bool,
6163 ) -> (SniProbe, Vec<u8>) {
6164 use tokio::io::{AsyncReadExt, AsyncWriteExt};
6165
6166 let listener = TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
6167 .await
6168 .unwrap();
6169 let addr = listener.local_addr().unwrap();
6170 let total: usize = writes.iter().map(Vec::len).sum();
6171
6172 let client = tokio::spawn(async move {
6173 let mut sock = TcpStream::connect(addr).await.unwrap();
6174 for chunk in writes {
6175 sock.write_all(&chunk).await.unwrap();
6176 sock.flush().await.unwrap();
6177 tokio::time::sleep(gap).await;
6178 }
6179 if close_after {
6180 sock.shutdown().await.unwrap();
6181 }
6182 tokio::time::sleep(std::time::Duration::from_secs(2)).await;
6185 });
6186
6187 let (stream, _) = listener.accept().await.unwrap();
6188 let probe = peek_sni_host(&stream, timeout).await;
6189
6190 let mut replayed = vec![0u8; total];
6192 let mut stream = stream;
6193 let read = tokio::time::timeout(
6194 std::time::Duration::from_secs(2),
6195 stream.read_exact(&mut replayed),
6196 )
6197 .await;
6198 let replayed = match read {
6199 Ok(Ok(_)) => replayed,
6200 _ => vec![],
6201 };
6202 client.abort();
6203 (probe, replayed)
6204 }
6205
6206 #[cfg(feature = "proxy-tls")]
6210 #[tokio::test]
6211 async fn test_peek_sni_host_reads_a_hello_split_across_writes() {
6212 let wire = client_hello_wire("api.localhost");
6213 let writes: Vec<Vec<u8>> = wire.chunks(3).map(<[u8]>::to_vec).collect();
6214 let (probe, replayed) = probe_over_socket(
6215 writes,
6216 std::time::Duration::from_millis(5),
6217 std::time::Duration::from_secs(5),
6218 )
6219 .await;
6220
6221 assert_eq!(probe, SniProbe::Host("api.localhost".to_string()));
6222 assert_eq!(replayed, wire, "peeked bytes must still be readable");
6223 }
6224
6225 #[cfg(feature = "proxy-tls")]
6229 #[tokio::test]
6230 async fn test_peek_sni_host_undetermined_when_a_hello_stalls() {
6231 let wire = client_hello_wire("api.localhost");
6232 let truncated = wire[..wire.len() / 2].to_vec();
6233 let (probe, _) = probe_over_socket(
6234 vec![truncated],
6235 std::time::Duration::ZERO,
6236 std::time::Duration::from_millis(150),
6237 )
6238 .await;
6239
6240 assert_eq!(probe, SniProbe::Undetermined);
6241 }
6242
6243 #[cfg(feature = "proxy-tls")]
6247 #[tokio::test]
6248 async fn test_peek_sni_host_notices_a_client_that_closes_mid_hello() {
6249 let wire = client_hello_wire("api.localhost");
6250 let truncated = wire[..wire.len() / 2].to_vec();
6251 let started = std::time::Instant::now();
6252 let (probe, _) = probe_over_socket_then(
6253 vec![truncated],
6254 std::time::Duration::ZERO,
6255 std::time::Duration::from_secs(5),
6256 true,
6257 )
6258 .await;
6259
6260 assert_eq!(probe, SniProbe::Undetermined);
6261 assert!(
6262 started.elapsed() < std::time::Duration::from_secs(1),
6263 "the probe must end when the client closes, not at its deadline"
6264 );
6265 }
6266
6267 #[cfg(feature = "proxy-tls")]
6271 #[tokio::test]
6272 async fn test_peek_sni_host_gives_up_on_a_silent_peer() {
6273 let started = std::time::Instant::now();
6274 let (probe, _) = probe_over_socket(
6275 vec![],
6276 std::time::Duration::ZERO,
6277 std::time::Duration::from_millis(150),
6278 )
6279 .await;
6280
6281 assert_eq!(probe, SniProbe::Undetermined);
6282 assert!(
6283 started.elapsed() < std::time::Duration::from_secs(1),
6284 "the probe must end at its deadline, not wait on the peer"
6285 );
6286 }
6287
6288 #[cfg(feature = "proxy-tls")]
6291 #[tokio::test]
6292 async fn test_peek_sni_host_reports_no_host_for_non_tls() {
6293 let (probe, _) = probe_over_socket(
6294 vec![b"GET / HTTP/1.1\r\n\r\n".to_vec()],
6295 std::time::Duration::ZERO,
6296 std::time::Duration::from_secs(5),
6297 )
6298 .await;
6299
6300 assert_eq!(probe, SniProbe::NoHost);
6301 }
6302
6303 #[cfg(feature = "proxy-tls")]
6304 #[test]
6305 fn test_generate_ca() {
6306 let dir = tempfile::tempdir().unwrap();
6307 let cert_path = dir.path().join("ca.pem");
6308 let key_path = dir.path().join("ca-key.pem");
6309
6310 generate_ca(&cert_path, &key_path).unwrap();
6311
6312 assert!(cert_path.exists(), "ca.pem should be created");
6313 assert!(key_path.exists(), "ca-key.pem should be created");
6314
6315 let cert_pem = std::fs::read_to_string(&cert_path).unwrap();
6316 let key_pem = std::fs::read_to_string(&key_path).unwrap();
6317
6318 assert!(cert_pem.contains("BEGIN CERTIFICATE"), "should be PEM cert");
6319 assert!(
6320 key_pem.contains("BEGIN") && key_pem.contains("PRIVATE KEY"),
6321 "should be PEM key"
6322 );
6323 }
6324
6325 fn cookie_fields(headers: &HeaderMap) -> Vec<&[u8]> {
6327 headers
6328 .get_all(COOKIE)
6329 .iter()
6330 .map(HeaderValue::as_bytes)
6331 .collect()
6332 }
6333
6334 #[test]
6337 fn test_join_cookie_fields_joins_with_semicolon_space() {
6338 let mut headers = HeaderMap::new();
6339 headers.append(COOKIE, HeaderValue::from_static("_session=abc123"));
6340 headers.append(COOKIE, HeaderValue::from_static("consent=ads,stats"));
6341 headers.append(COOKIE, HeaderValue::from_static("theme=dark"));
6342
6343 join_cookie_fields(&mut headers);
6344
6345 assert_eq!(
6346 cookie_fields(&headers),
6347 vec![&b"_session=abc123; consent=ads,stats; theme=dark"[..]]
6348 );
6349 }
6350
6351 #[test]
6354 fn test_join_cookie_fields_joins_bytes_outside_ascii() {
6355 let mut headers = HeaderMap::new();
6356 headers.append(COOKIE, HeaderValue::from_static("_session=abc123"));
6357 headers.append(
6358 COOKIE,
6359 HeaderValue::from_bytes(b"name=Jos\xc3\xa9").unwrap(),
6360 );
6361
6362 join_cookie_fields(&mut headers);
6363
6364 assert_eq!(
6365 cookie_fields(&headers),
6366 vec![&b"_session=abc123; name=Jos\xc3\xa9"[..]]
6367 );
6368 }
6369
6370 #[test]
6372 fn test_join_cookie_fields_without_cookies() {
6373 let mut headers = HeaderMap::new();
6374 headers.insert(HOST, HeaderValue::from_static("app.localhost"));
6375
6376 join_cookie_fields(&mut headers);
6377
6378 assert!(headers.get(COOKIE).is_none());
6379 }
6380
6381 #[test]
6382 fn test_host_port_suffix() {
6383 assert_eq!(host_port_suffix("api.myproj.localhost:8088"), ":8088");
6384 assert_eq!(host_port_suffix("api.myproj.localhost"), "");
6385 assert_eq!(host_port_suffix("[::1]:8088"), ":8088");
6386 assert_eq!(host_port_suffix("[::1]"), "");
6387 assert_eq!(host_port_suffix("host:notaport"), "");
6389 }
6390
6391 #[test]
6395 fn test_daemon_runs_in() {
6396 let temp = tempfile::tempdir().unwrap();
6397 let repo = temp.path().join("my-repo");
6398 std::fs::create_dir_all(repo.join(".git/worktrees/feature")).unwrap();
6399 std::fs::create_dir_all(repo.join("sub")).unwrap();
6400 let nested = repo.join(".worktrees/feature");
6402 std::fs::create_dir_all(&nested).unwrap();
6403 std::fs::write(
6404 nested.join(".git"),
6405 format!(
6406 "gitdir: {}\n",
6407 repo.join(".git/worktrees/feature").display()
6408 ),
6409 )
6410 .unwrap();
6411
6412 let root = |p: &std::path::Path| crate::proxy::hostname::checkout_root_of(p);
6413 let repo_root = root(&repo);
6414 let nested_root = root(&nested);
6415
6416 let mut daemon = crate::daemon::Daemon {
6417 dir: Some(repo.join("sub")),
6418 ..Default::default()
6419 };
6420 assert!(daemon_runs_in(&daemon, &repo_root));
6421 assert!(!daemon_runs_in(&daemon, &nested_root));
6422
6423 daemon.dir = Some(nested.clone());
6426 assert!(daemon_runs_in(&daemon, &nested_root));
6427 assert!(!daemon_runs_in(&daemon, &repo_root));
6428
6429 daemon.dir = Some(temp.path().join("elsewhere"));
6431 assert!(!daemon_runs_in(&daemon, &repo_root));
6432
6433 daemon.dir = None;
6434 assert!(!daemon_runs_in(&daemon, &repo_root));
6435 }
6436
6437 #[test]
6440 fn test_is_local_client() {
6441 let build = |info: Option<SocketAddr>| {
6442 let mut req = Request::new(Body::empty());
6443 if let Some(addr) = info {
6444 req.extensions_mut()
6445 .insert(axum::extract::ConnectInfo(addr));
6446 }
6447 req
6448 };
6449
6450 assert!(is_local_client(&build(Some(
6451 "127.0.0.1:5000".parse().unwrap()
6452 ))));
6453 assert!(is_local_client(&build(Some("[::1]:5000".parse().unwrap()))));
6454 assert!(!is_local_client(&build(Some(
6455 "192.168.1.42:5000".parse().unwrap()
6456 ))));
6457 assert!(!is_local_client(&build(None)));
6458 }
6459}