1use std::{
2 borrow::Cow,
3 collections::{
4 BTreeMap, BTreeSet, HashMap, HashSet, btree_map::Entry as BTreeMapEntry,
5 hash_map::DefaultHasher,
6 },
7 fmt,
8 fs::File,
9 hash::{Hash, Hasher},
10 io::Write,
11 iter::repeat,
12 net::SocketAddr,
13};
14
15use prost::{Message, UnknownEnumValue};
16
17use crate::{
18 ObjectKind,
19 certificate::{CertificateError, Fingerprint, calculate_fingerprint},
20 config::validate_sni_pattern,
21 listener_key::{InterfaceError, ListenerKey, validate_listener_interface},
22 proto::{
23 command::{
24 ActivateListener, AddBackend, AddCertificate, CertificateAndKey, Cluster,
25 ClusterInformation, CustomHttpAnswers, DeactivateListener, ForwardedHeaders,
26 FrontendFilters, HealthChecksList, HstsConfig, HttpListenerConfig, HttpsListenerConfig,
27 InitialState, ListedFrontends, ListenerType, ListenersList, PathRule,
28 QueryCertificatesFilters, RemoveBackend, RemoveCertificate, RemoveListener,
29 ReplaceCertificate, Request, RequestCounts, RequestHttpFrontend, RequestTcpFrontend,
30 RequestUdpFrontend, SetHealthCheck, SocketAddress, TcpListenerConfig,
31 UdpListenerConfig, UpdateHttpListenerConfig, UpdateHttpsListenerConfig,
32 UpdateTcpListenerConfig, UpdateUdpListenerConfig, WorkerRequest, request::RequestType,
33 },
34 display::format_request_type,
35 },
36 response::{Backend, HttpFrontend, TcpFrontend, UdpFrontend},
37};
38
39pub type ClusterId = std::sync::Arc<str>;
46
47const _: fn() = || {
51 fn assert_send_sync<T: Send + Sync>() {}
52 assert_send_sync::<ClusterId>();
53 assert_send_sync::<ConfigState>();
54 assert_send_sync::<StateError>();
55 assert_send_sync::<crate::response::HttpFrontend>();
56 assert_send_sync::<crate::response::TcpFrontend>();
57 assert_send_sync::<crate::response::UdpFrontend>();
58 assert_send_sync::<crate::response::Backend>();
59};
60
61pub fn rollback_request(request: &RequestType) -> Option<Request> {
69 let inverse = match request {
70 RequestType::AddHttpListener(config) => RequestType::RemoveListener(RemoveListener {
71 interface: config.interface.clone(),
72 address: config.address,
73 proxy: ListenerType::Http.into(),
74 }),
75 RequestType::AddHttpsListener(config)
76 | RequestType::AddHttpsListenerWithClientAuth(config) => {
77 RequestType::RemoveListener(RemoveListener {
78 interface: config.interface.clone(),
79 address: config.address,
80 proxy: ListenerType::Https.into(),
81 })
82 }
83 RequestType::AddTcpListener(config) => RequestType::RemoveListener(RemoveListener {
84 interface: config.interface.clone(),
85 address: config.address,
86 proxy: ListenerType::Tcp.into(),
87 }),
88 RequestType::AddUdpListener(config) => RequestType::RemoveListener(RemoveListener {
89 interface: config.interface.clone(),
90 address: config.address,
91 proxy: ListenerType::Udp.into(),
92 }),
93 RequestType::AddHttpFrontend(front) => RequestType::RemoveHttpFrontend(front.clone()),
94 RequestType::AddHttpsFrontend(front) => RequestType::RemoveHttpsFrontend(front.clone()),
95 RequestType::AddTcpFrontend(front) => RequestType::RemoveTcpFrontend(front.clone()),
96 RequestType::AddUdpFrontend(front) => RequestType::RemoveUdpFrontend(front.clone()),
97 _ => return None,
98 };
99 Some(inverse.into())
100}
101
102#[derive(thiserror::Error)]
103pub enum StateError {
104 #[error("Request came in empty")]
105 EmptyRequest,
106 #[error("dispatching this request did not bring any change to the state")]
107 NoChange,
108 #[error("State can not handle this request")]
109 UndispatchableRequest,
110 #[error("Did not find {kind:?} with address or id_bytes={}", .id.len())]
111 NotFound { kind: ObjectKind, id: String },
112 #[error(
113 "{kind:?} with id_bytes={} already exists; remove it first, or apply the corresponding update if the object supports one, instead of re-adding it",
114 .id.len()
115 )]
116 Exists { kind: ObjectKind, id: String },
117 #[error("Wrong field value: {0}")]
118 WrongFieldValue(UnknownEnumValue),
119 #[error("Could not add certificate: error_kind={}", certificate_error_kind(.0))]
120 AddCertificate(CertificateError),
121 #[error("Could not remove certificate: fingerprint_bytes={}", .0.len())]
122 RemoveCertificate(String),
123 #[error("Could not replace certificate: error_bytes={}", .0.len())]
124 ReplaceCertificate(String),
125 #[error("Could not convert frontend: frontend_bytes={} error_bytes={}", .frontend.len(), .error.len())]
126 FrontendConversion { frontend: String, error: String },
127 #[error("Could not write state to file: io_kind={:?} os_code={:?}", .0.kind(), .0.raw_os_error())]
128 FileError(std::io::Error),
129 #[error("Invalid value for field '{field}': {reason}")]
130 InvalidValue {
131 field: &'static str,
132 reason: &'static str,
133 },
134 #[error("invalid TCP frontend for address {address}: reason_bytes={}", .reason.len())]
140 InvalidTcpFrontend { address: SocketAddr, reason: String },
141 #[error(
159 "UDP frontend address {address} is already claimed by another frontend: \
160 cluster_id_bytes={} current_cluster_id_bytes={} current_tag_count={}",
161 .cluster_id.len(),
162 .current_cluster_id.len(),
163 .current_tags.len()
164 )]
165 UdpFrontendAddressTaken {
166 address: SocketAddr,
167 cluster_id: ClusterId,
168 current_cluster_id: ClusterId,
169 current_tags: BTreeMap<String, String>,
170 },
171 #[error("invalid listener interface: {0}")]
174 ListenerInterface(InterfaceError),
175 #[error(
178 "{count} listeners share the address {address} on different interfaces; an update \
179 patch cannot tell them apart: remove and re-add the one to change"
180 )]
181 AmbiguousListenerAddress { address: SocketAddr, count: usize },
182}
183
184fn validated_listener_key(key: ListenerKey) -> Result<ListenerKey, StateError> {
187 if let Some(interface) = &key.interface {
188 validate_listener_interface(interface).map_err(StateError::ListenerInterface)?;
189 }
190 Ok(key)
191}
192
193fn listener_key_for_update<V>(
197 listeners: &BTreeMap<ListenerKey, V>,
198 address: SocketAddr,
199 kind: ObjectKind,
200) -> Result<ListenerKey, StateError> {
201 let mut matching = listeners.keys().filter(|key| key.address == address);
202 let Some(first) = matching.next() else {
203 return Err(StateError::NotFound {
204 kind,
205 id: address.to_string(),
206 });
207 };
208 let others = matching.count();
209 if others > 0 {
210 return Err(StateError::AmbiguousListenerAddress {
211 address,
212 count: others + 1,
213 });
214 }
215 Ok(first.clone())
216}
217
218impl fmt::Debug for StateError {
219 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
220 write!(f, "StateError({self})")
221 }
222}
223
224fn certificate_error_kind(error: &CertificateError) -> &'static str {
225 match error {
226 CertificateError::ParsePEMCertificate(_) => "parse_pem_certificate",
227 CertificateError::ParseX509Certificate(_) => "parse_x509_certificate",
228 CertificateError::InvalidTlsVersion(_) => "invalid_tls_version",
229 CertificateError::InvalidFingerprint(_) => "invalid_fingerprint",
230 CertificateError::LoadFile { .. } => "load_file",
231 CertificateError::DecodeError(_) => "decode_error",
232 }
233}
234
235const TCP_CLUSTER_AFFINITY_KEY: &str = "a TCP frontend always keys on the client source IP: \
238 affinity_header and affinity_cookie apply to HTTP/HTTPS clusters only";
239
240#[derive(Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
249pub struct ConfigState {
250 pub clusters: BTreeMap<ClusterId, Cluster>,
251 pub backends: BTreeMap<ClusterId, Vec<Backend>>,
252 pub http_listeners: BTreeMap<ListenerKey, HttpListenerConfig>,
254 pub https_listeners: BTreeMap<ListenerKey, HttpsListenerConfig>,
256 pub tcp_listeners: BTreeMap<ListenerKey, TcpListenerConfig>,
258 pub udp_listeners: BTreeMap<ListenerKey, UdpListenerConfig>,
260 pub http_fronts: BTreeMap<String, HttpFrontend>,
263 pub https_fronts: BTreeMap<String, HttpFrontend>,
265 pub tcp_fronts: HashMap<ClusterId, Vec<TcpFrontend>>,
266 pub udp_fronts: HashMap<ClusterId, Vec<UdpFrontend>>,
267 pub certificates: HashMap<SocketAddr, HashMap<Fingerprint, CertificateAndKey>>,
268 pub request_counts: BTreeMap<String, i32>,
270}
271
272impl std::fmt::Debug for ConfigState {
273 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
274 let backends_count = self
275 .backends
276 .values()
277 .map(Vec::len)
278 .fold(0usize, usize::saturating_add);
279 let tcp_frontends_count = self
280 .tcp_fronts
281 .values()
282 .map(Vec::len)
283 .fold(0usize, usize::saturating_add);
284 let udp_frontends_count = self
285 .udp_fronts
286 .values()
287 .map(Vec::len)
288 .fold(0usize, usize::saturating_add);
289 let certificates_count = self
290 .certificates
291 .values()
292 .map(HashMap::len)
293 .fold(0usize, usize::saturating_add);
294
295 f.debug_struct("ConfigState")
296 .field("clusters_count", &self.clusters.len())
297 .field("backend_clusters_count", &self.backends.len())
298 .field("backends_count", &backends_count)
299 .field("http_listeners_count", &self.http_listeners.len())
300 .field("https_listeners_count", &self.https_listeners.len())
301 .field("tcp_listeners_count", &self.tcp_listeners.len())
302 .field("udp_listeners_count", &self.udp_listeners.len())
303 .field("http_frontends_count", &self.http_fronts.len())
304 .field("https_frontends_count", &self.https_fronts.len())
305 .field("tcp_frontend_clusters_count", &self.tcp_fronts.len())
306 .field("tcp_frontends_count", &tcp_frontends_count)
307 .field("udp_frontend_clusters_count", &self.udp_fronts.len())
308 .field("udp_frontends_count", &udp_frontends_count)
309 .field("certificate_addresses_count", &self.certificates.len())
310 .field("certificates_count", &certificates_count)
311 .field("request_counts_count", &self.request_counts.len())
312 .finish_non_exhaustive()
313 }
314}
315
316impl ConfigState {
317 pub fn new() -> Self {
318 Self::default()
319 }
320
321 pub fn rollback_for(&self, request: &RequestType) -> Option<Request> {
329 if let Some(rollback) = rollback_request(request) {
330 return Some(rollback);
331 }
332
333 let inverse = match request {
334 RequestType::RemoveHttpFrontend(selector) => self
335 .http_fronts
336 .get(&selector.to_string())
337 .filter(|stored| stored.cluster_id.as_deref() == selector.cluster_id.as_deref())
338 .cloned()
339 .map(RequestHttpFrontend::from)
340 .map(RequestType::AddHttpFrontend),
341 RequestType::RemoveHttpsFrontend(selector) => self
342 .https_fronts
343 .get(&selector.to_string())
344 .filter(|stored| stored.cluster_id.as_deref() == selector.cluster_id.as_deref())
345 .cloned()
346 .map(RequestHttpFrontend::from)
347 .map(RequestType::AddHttpsFrontend),
348 RequestType::RemoveTcpFrontend(selector) => {
349 let address: SocketAddr = selector.address.into();
350 let sni = selector.sni.as_deref().map(str::to_ascii_lowercase);
351 let alpn = canonical_tcp_alpn(&selector.alpn);
352 self.tcp_fronts
353 .get(selector.cluster_id.as_str())
354 .and_then(|frontends| {
355 frontends
356 .iter()
357 .find(|stored| tcp_frontend_matches(stored, address, &sni, &alpn))
358 })
359 .cloned()
360 .map(RequestTcpFrontend::from)
361 .map(RequestType::AddTcpFrontend)
362 }
363 RequestType::RemoveUdpFrontend(selector) => {
364 let address: SocketAddr = selector.address.into();
365 self.udp_fronts
366 .get(selector.cluster_id.as_str())
367 .and_then(|frontends| {
368 frontends.iter().find(|stored| {
369 stored.address == address && stored.tags == selector.tags
370 })
371 })
372 .cloned()
373 .map(RequestUdpFrontend::from)
374 .map(RequestType::AddUdpFrontend)
375 }
376 _ => None,
377 };
378
379 inverse.map(Into::into)
380 }
381
382 pub fn dispatch(&mut self, request: &Request) -> Result<(), StateError> {
383 let request_type = match &request.request_type {
384 Some(t) => t,
385 None => return Err(StateError::EmptyRequest),
386 };
387
388 self.increment_request_count(request);
389
390 let result = match request_type {
391 RequestType::AddCluster(cluster) => self.add_cluster(cluster),
392 RequestType::RemoveCluster(cluster_id) => self.remove_cluster(cluster_id),
393 RequestType::AddHttpListener(listener) => self.add_http_listener(listener),
394 RequestType::AddHttpsListener(listener)
395 | RequestType::AddHttpsListenerWithClientAuth(listener) => {
396 self.add_https_listener(listener)
397 }
398 RequestType::AddTcpListener(listener) => self.add_tcp_listener(listener),
399 RequestType::AddUdpListener(listener) => self.add_udp_listener(listener),
400 RequestType::RemoveListener(remove) => self.remove_listener(remove),
401 RequestType::ActivateListener(activate) => self.activate_listener(activate),
402 RequestType::DeactivateListener(deactivate) => self.deactivate_listener(deactivate),
403 RequestType::AddHttpFrontend(front) => self.add_http_frontend(front),
404 RequestType::RemoveHttpFrontend(front) => self.remove_http_frontend(front),
405 RequestType::AddCertificate(add) => self.add_certificate(add),
406 RequestType::RemoveCertificate(remove) => self.remove_certificate(remove),
407 RequestType::ReplaceCertificate(replace) => self.replace_certificate(replace),
408 RequestType::AddHttpsFrontend(front) => self.add_https_frontend(front),
409 RequestType::RemoveHttpsFrontend(front) => self.remove_https_frontend(front),
410 RequestType::AddTcpFrontend(front) => self.add_tcp_frontend(front),
411 RequestType::RemoveTcpFrontend(front) => self.remove_tcp_frontend(front),
412 RequestType::AddUdpFrontend(front) => self.add_udp_frontend(front),
413 RequestType::RemoveUdpFrontend(front) => self.remove_udp_frontend(front),
414 RequestType::AddBackend(add_backend) => self.add_backend(add_backend),
415 RequestType::RemoveBackend(backend) => self.remove_backend(backend),
416 RequestType::UpdateHttpListener(patch) => self.update_http_listener(patch),
417 RequestType::UpdateHttpsListener(patch)
418 | RequestType::UpdateHttpsListenerWithClientAuth(patch) => {
419 self.update_https_listener(patch)
420 }
421 RequestType::UpdateTcpListener(patch) => self.update_tcp_listener(patch),
422 RequestType::UpdateUdpListener(patch) => self.update_udp_listener(patch),
423 RequestType::SetHealthCheck(set) => self.set_health_check(set),
424 RequestType::RemoveHealthCheck(cluster_id) => self.remove_health_check(cluster_id),
425
426 RequestType::Logging(_)
433 | RequestType::CountRequests(_)
434 | RequestType::Status(_)
435 | RequestType::SoftStop(_)
436 | RequestType::QueryCertificatesFromWorkers(_)
437 | RequestType::QueryClusterById(_)
438 | RequestType::QueryClustersByDomain(_)
439 | RequestType::QueryMetrics(_)
440 | RequestType::QueryClustersHashes(_)
441 | RequestType::ConfigureMetrics(_)
442 | RequestType::SetMetricDetail(_)
443 | RequestType::ReturnListenSockets(_)
444 | RequestType::SetMaxConnectionsPerIp(_)
445 | RequestType::QueryMaxConnectionsPerIp(_)
446 | RequestType::SetMaxConnectionsPerSubnet(_)
447 | RequestType::QueryMaxConnectionsPerSubnet(_)
448 | RequestType::SetMaxConnectionAttempts(_)
449 | RequestType::HardStop(_) => Ok(()),
450
451 _other_request => Err(StateError::UndispatchableRequest),
452 };
453
454 #[cfg(debug_assertions)]
460 self.check_invariants();
461
462 result
463 }
464
465 #[cfg(debug_assertions)]
476 fn check_invariants(&self) {
477 for (addr, listener) in &self.http_listeners {
481 debug_assert_eq!(
482 listener.listener_key(),
483 *addr,
484 "http_listener value address must match its map key"
485 );
486 }
487 for (addr, listener) in &self.https_listeners {
488 debug_assert_eq!(
489 listener.listener_key(),
490 *addr,
491 "https_listener value address must match its map key"
492 );
493 }
494 for (addr, listener) in &self.tcp_listeners {
495 debug_assert_eq!(
496 listener.listener_key(),
497 *addr,
498 "tcp_listener value address must match its map key"
499 );
500 }
501
502 for (cluster_id, cluster) in &self.clusters {
505 debug_assert_eq!(
506 cluster.cluster_id.as_str(),
507 &**cluster_id,
508 "cluster value cluster_id must match its map key"
509 );
510 }
511
512 for (cluster_id, backends) in &self.backends {
517 for backend in backends {
518 debug_assert_eq!(
519 &backend.cluster_id, cluster_id,
520 "backend cluster_id must match its bucket key"
521 );
522 }
523 let unique: HashSet<(&String, &SocketAddr)> = backends
524 .iter()
525 .map(|b| (&b.backend_id, &b.address))
526 .collect();
527 debug_assert_eq!(
528 unique.len(),
529 backends.len(),
530 "backends within a cluster must be unique on (backend_id, address)"
531 );
532 }
533
534 for (cluster_id, fronts) in &self.tcp_fronts {
538 for front in fronts {
539 debug_assert_eq!(
540 &front.cluster_id, cluster_id,
541 "tcp_frontend cluster_id must match its bucket key"
542 );
543 }
544 let unique: HashSet<&TcpFrontend> = fronts.iter().collect();
545 debug_assert_eq!(
546 unique.len(),
547 fronts.len(),
548 "tcp frontends within a cluster must be unique"
549 );
550 }
551
552 for (address, certs) in &self.certificates {
574 debug_assert!(
575 !certs.is_empty(),
576 "certificate bucket for address {address} must never be empty (sozu-proxy/sozu#1404)"
577 );
578 }
579
580 let raw_frontends = self.http_fronts.len()
583 + self.https_fronts.len()
584 + self.count_tcp_frontends_raw()
585 + self.count_udp_frontends_raw();
586 debug_assert_eq!(
587 self.count_frontends(),
588 raw_frontends,
589 "count_frontends must equal the sum of all frontend map entries"
590 );
591 let raw_backends: usize = self.backends.values().map(|v| v.len()).sum();
592 debug_assert_eq!(
593 self.count_backends(),
594 raw_backends,
595 "count_backends must equal the sum of all backend Vec lengths"
596 );
597 }
598
599 fn count_tcp_frontends_raw(&self) -> usize {
607 self.tcp_fronts.values().map(|v| v.len()).sum()
608 }
609
610 fn count_udp_frontends_raw(&self) -> usize {
611 self.udp_fronts.values().map(|v| v.len()).sum()
612 }
613
614 fn count_objects_where(&self, owned: impl Fn(&str) -> bool) -> usize {
619 let keyed = |cluster_id: &ClusterId, len: usize| if owned(cluster_id) { len } else { 0 };
620 let routed = |front: &HttpFrontend| front.cluster_id.as_deref().is_some_and(&owned);
621 self.backends
622 .iter()
623 .map(|(k, v)| keyed(k, v.len()))
624 .sum::<usize>()
625 + self
626 .tcp_fronts
627 .iter()
628 .map(|(k, v)| keyed(k, v.len()))
629 .sum::<usize>()
630 + self
631 .udp_fronts
632 .iter()
633 .map(|(k, v)| keyed(k, v.len()))
634 .sum::<usize>()
635 + self.http_fronts.values().filter(|f| routed(f)).count()
636 + self.https_fronts.values().filter(|f| routed(f)).count()
637 }
638
639 fn count_objects_owned_by(&self, cluster_id: &str) -> usize {
640 self.count_objects_where(|owner| owner == cluster_id)
641 }
642
643 #[cfg(debug_assertions)]
646 fn count_objects_not_owned_by(&self, cluster_id: &str) -> usize {
647 self.count_objects_where(|owner| owner != cluster_id)
648 }
649
650 fn increment_request_count(&mut self, request: &Request) {
652 if let Some(request_type) = &request.request_type {
653 let count = self
654 .request_counts
655 .entry(format_request_type(request_type).to_owned())
656 .or_insert(1);
657 *count += 1;
658 }
659 }
660
661 pub fn get_request_counts(&self) -> RequestCounts {
662 RequestCounts {
663 map: self.request_counts.clone(),
664 }
665 }
666
667 fn add_cluster(&mut self, cluster: &Cluster) -> Result<(), StateError> {
668 if let Some(hc) = cluster.health_check.as_ref()
675 && let Err(reason) = crate::config::validate_health_check_config(hc)
676 {
677 return Err(StateError::InvalidValue {
678 field: "health_check",
679 reason,
680 });
681 }
682 if let Err(reason) = crate::config::validate_affinity_key(
686 cluster.affinity_header.as_deref(),
687 cluster.affinity_cookie.as_deref(),
688 ) {
689 return Err(StateError::InvalidValue {
690 field: "affinity_key",
691 reason,
692 });
693 }
694 if let Err(reason) = crate::config::validate_shuffle_sharding(
695 cluster.shard_percent,
696 cluster.shard_min_backends,
697 cluster.shard_mode,
698 ) {
699 return Err(StateError::InvalidValue {
700 field: "shuffle_sharding",
701 reason,
702 });
703 }
704 if let Some(Err(reason)) = cluster
705 .max_connection_attempts
706 .map(crate::config::validate_max_connection_attempts)
707 {
708 return Err(StateError::InvalidValue {
709 field: "max_connection_attempts",
710 reason,
711 });
712 }
713 if (cluster.affinity_header.is_some() || cluster.affinity_cookie.is_some())
718 && self
719 .tcp_fronts
720 .get(cluster.cluster_id.as_str())
721 .is_some_and(|fronts| !fronts.is_empty())
722 {
723 return Err(StateError::InvalidValue {
724 field: "affinity_key",
725 reason: TCP_CLUSTER_AFFINITY_KEY,
726 });
727 }
728 let cluster = cluster.clone();
729 let cluster_id = ClusterId::from(cluster.cluster_id.as_str());
733 self.clusters.insert(cluster_id.clone(), cluster);
734 debug_assert!(
735 self.clusters.contains_key(&cluster_id),
736 "add_cluster must leave the cluster present in the map"
737 );
738 debug_assert_eq!(
739 self.clusters
740 .get(&cluster_id)
741 .map(|c| c.cluster_id.as_str()),
742 Some(&*cluster_id),
743 "stored cluster must be keyed by its own cluster_id"
744 );
745 Ok(())
746 }
747
748 fn remove_cluster(&mut self, cluster_id: &str) -> Result<(), StateError> {
754 let before = self.clusters.len();
755 match self.clusters.remove(cluster_id) {
756 Some(_) => {
757 #[cfg(debug_assertions)]
758 let owned_by_others = self.count_objects_not_owned_by(cluster_id);
759
760 self.backends.remove(cluster_id);
761 self.tcp_fronts.remove(cluster_id);
762 self.udp_fronts.remove(cluster_id);
763 self.http_fronts
764 .retain(|_, front| front.cluster_id.as_deref() != Some(cluster_id));
765 self.https_fronts
766 .retain(|_, front| front.cluster_id.as_deref() != Some(cluster_id));
767
768 debug_assert!(
769 !self.clusters.contains_key(cluster_id),
770 "remove_cluster must evict the cluster"
771 );
772 debug_assert_eq!(
773 self.clusters.len(),
774 before - 1,
775 "remove_cluster must drop exactly one entry"
776 );
777 debug_assert_eq!(
778 self.count_objects_owned_by(cluster_id),
779 0,
780 "remove_cluster must leave no frontend or backend naming the cluster"
781 );
782 #[cfg(debug_assertions)]
783 debug_assert_eq!(
784 self.count_objects_not_owned_by(cluster_id),
785 owned_by_others,
786 "remove_cluster must not touch a frontend or backend of another cluster"
787 );
788 Ok(())
789 }
790 None => {
791 debug_assert_eq!(
792 self.clusters.len(),
793 before,
794 "a failed remove_cluster must not mutate the map"
795 );
796 Err(StateError::NotFound {
797 kind: ObjectKind::Cluster,
798 id: cluster_id.to_owned(),
799 })
800 }
801 }
802 }
803
804 pub fn cluster_removal_requests(&self, cluster_id: &str) -> Vec<Request> {
810 if !self.clusters.contains_key(cluster_id) {
811 return Vec::new();
812 }
813 let owned = |front: &&HttpFrontend| front.cluster_id.as_deref() == Some(cluster_id);
814 let mut requests: Vec<Request> = Vec::new();
815 requests.extend(
816 self.http_fronts
817 .values()
818 .filter(owned)
819 .map(|front| RequestType::RemoveHttpFrontend(front.clone().into()).into()),
820 );
821 requests.extend(
822 self.https_fronts
823 .values()
824 .filter(owned)
825 .map(|front| RequestType::RemoveHttpsFrontend(front.clone().into()).into()),
826 );
827 requests.extend(
828 self.tcp_fronts
829 .get(cluster_id)
830 .into_iter()
831 .flatten()
832 .map(|front| RequestType::RemoveTcpFrontend(front.clone().into()).into()),
833 );
834 requests.extend(
835 self.udp_fronts
836 .get(cluster_id)
837 .into_iter()
838 .flatten()
839 .map(|front| RequestType::RemoveUdpFrontend(front.clone().into()).into()),
840 );
841 requests.extend(
842 self.backends
843 .get(cluster_id)
844 .into_iter()
845 .flatten()
846 .map(|backend| {
847 RequestType::RemoveBackend(RemoveBackend {
848 cluster_id: backend.cluster_id.to_string(),
849 backend_id: backend.backend_id.clone(),
850 address: SocketAddress::from(backend.address),
851 })
852 .into()
853 }),
854 );
855 debug_assert_eq!(
856 requests.len(),
857 self.count_objects_owned_by(cluster_id),
858 "one removal order per frontend or backend the cluster owns"
859 );
860 requests
861 }
862
863 pub fn removed_with_its_cluster(&self, request: &Request, error: &StateError) -> bool {
876 if !matches!(error, StateError::NotFound { .. }) {
877 return false;
878 }
879 let cluster_id = match &request.request_type {
880 Some(
881 RequestType::RemoveHttpFrontend(front) | RequestType::RemoveHttpsFrontend(front),
882 ) => front.cluster_id.as_deref(),
883 Some(RequestType::RemoveTcpFrontend(front)) => Some(front.cluster_id.as_str()),
884 Some(RequestType::RemoveUdpFrontend(front)) => Some(front.cluster_id.as_str()),
885 Some(RequestType::RemoveBackend(backend)) => Some(backend.cluster_id.as_str()),
886 _ => None,
887 };
888 cluster_id.is_some_and(|cluster_id| !self.clusters.contains_key(cluster_id))
889 }
890
891 pub fn removed_with_its_cluster_message(&self, request: &Request) -> String {
897 let (cluster_id, owner) = match &request.request_type {
898 Some(RequestType::RemoveHttpFrontend(front)) => (
899 front.cluster_id.clone(),
900 self.http_fronts
901 .get(&front.to_string())
902 .map(|stored| stored.cluster_id.as_deref().map(str::to_owned)),
903 ),
904 Some(RequestType::RemoveHttpsFrontend(front)) => (
905 front.cluster_id.clone(),
906 self.https_fronts
907 .get(&front.to_string())
908 .map(|stored| stored.cluster_id.as_deref().map(str::to_owned)),
909 ),
910 Some(RequestType::RemoveTcpFrontend(front)) => {
911 let address: SocketAddr = front.address.into();
912 let sni = front.sni.as_deref().map(str::to_ascii_lowercase);
913 let alpn = canonical_tcp_alpn(&front.alpn);
914 (
915 Some(front.cluster_id.clone()),
916 self.tcp_fronts
917 .values()
918 .flatten()
919 .find(|stored| tcp_frontend_matches(stored, address, &sni, &alpn))
920 .map(|stored| Some(stored.cluster_id.to_string())),
921 )
922 }
923 Some(RequestType::RemoveUdpFrontend(front)) => {
924 let address: SocketAddr = front.address.into();
925 (
926 Some(front.cluster_id.clone()),
927 self.udp_fronts
928 .values()
929 .flatten()
930 .find(|stored| stored.address == address)
931 .map(|stored| Some(stored.cluster_id.to_string())),
932 )
933 }
934 Some(RequestType::RemoveBackend(backend)) => (Some(backend.cluster_id.clone()), None),
935 _ => (None, None),
936 };
937 let mut message = format!(
938 "nothing to remove: cluster {} does not exist (removed, or never added), \
939 and none of its frontends or backends either",
940 cluster_id.as_deref().unwrap_or("<none>")
941 );
942 match owner {
943 Some(Some(owner)) => message.push_str(&format!(
944 "; this key belongs to cluster {owner}, left in place"
945 )),
946 Some(None) => {
947 message.push_str("; this key belongs to a route with no cluster, left in place")
948 }
949 None => {}
950 }
951 message
952 }
953
954 fn set_health_check(&mut self, set: &SetHealthCheck) -> Result<(), StateError> {
955 if let Err(reason) = crate::config::validate_health_check_config(&set.config) {
962 return Err(StateError::InvalidValue {
963 field: "health_check",
964 reason,
965 });
966 }
967 match self.clusters.get_mut(set.cluster_id.as_str()) {
968 Some(cluster) => {
969 cluster.health_check = Some(set.config.to_owned());
970 Ok(())
971 }
972 None => Err(StateError::NotFound {
973 kind: ObjectKind::Cluster,
974 id: set.cluster_id.to_owned(),
975 }),
976 }
977 }
978
979 fn remove_health_check(&mut self, cluster_id: &str) -> Result<(), StateError> {
980 match self.clusters.get_mut(cluster_id) {
981 Some(cluster) => {
982 cluster.health_check = None;
983 Ok(())
984 }
985 None => Err(StateError::NotFound {
986 kind: ObjectKind::Cluster,
987 id: cluster_id.to_owned(),
988 }),
989 }
990 }
991
992 pub fn list_health_checks(&self, cluster_id: Option<&str>) -> HealthChecksList {
993 let map = self
994 .clusters
995 .iter()
996 .filter(|(id, _)| cluster_id.is_none_or(|filter| filter == &id[..]))
997 .filter_map(|(id, cluster)| {
998 cluster
999 .health_check
1000 .as_ref()
1001 .map(|hc| (id.to_string(), hc.to_owned()))
1002 })
1003 .collect();
1004 HealthChecksList { map }
1005 }
1006
1007 fn add_http_listener(&mut self, listener: &HttpListenerConfig) -> Result<(), StateError> {
1008 if let Some(v) = listener.forwarded_headers {
1011 validate_forwarded_headers(v)?;
1012 }
1013 let address = validated_listener_key(listener.listener_key())?;
1014 let before = self.http_listeners.len();
1015 match self.http_listeners.entry(address.clone()) {
1016 BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(listener.clone()),
1017 BTreeMapEntry::Occupied(_) => {
1018 debug_assert_eq!(
1019 self.http_listeners.len(),
1020 before,
1021 "a rejected duplicate add_http_listener must not mutate the map"
1022 );
1023 return Err(StateError::Exists {
1024 kind: ObjectKind::HttpListener,
1025 id: address.to_string(),
1026 });
1027 }
1028 };
1029 debug_assert!(
1030 self.http_listeners.contains_key(&address),
1031 "add_http_listener must insert the listener under its address"
1032 );
1033 debug_assert_eq!(
1034 self.http_listeners.len(),
1035 before + 1,
1036 "add_http_listener inserts exactly one entry on the vacant path"
1037 );
1038 Ok(())
1039 }
1040
1041 fn add_https_listener(&mut self, listener: &HttpsListenerConfig) -> Result<(), StateError> {
1042 if let Some(v) = listener.forwarded_headers {
1045 validate_forwarded_headers(v)?;
1046 }
1047 let address = validated_listener_key(listener.listener_key())?;
1048 let before = self.https_listeners.len();
1049 match self.https_listeners.entry(address.clone()) {
1050 BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(listener.clone()),
1051 BTreeMapEntry::Occupied(_) => {
1052 debug_assert_eq!(
1053 self.https_listeners.len(),
1054 before,
1055 "a rejected duplicate add_https_listener must not mutate the map"
1056 );
1057 return Err(StateError::Exists {
1058 kind: ObjectKind::HttpsListener,
1059 id: address.to_string(),
1060 });
1061 }
1062 };
1063 debug_assert!(
1064 self.https_listeners.contains_key(&address),
1065 "add_https_listener must insert the listener under its address"
1066 );
1067 debug_assert_eq!(
1068 self.https_listeners.len(),
1069 before + 1,
1070 "add_https_listener inserts exactly one entry on the vacant path"
1071 );
1072 Ok(())
1073 }
1074
1075 fn add_tcp_listener(&mut self, listener: &TcpListenerConfig) -> Result<(), StateError> {
1076 let address = validated_listener_key(listener.listener_key())?;
1077 let before = self.tcp_listeners.len();
1078 match self.tcp_listeners.entry(address.clone()) {
1079 BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(listener.clone()),
1080 BTreeMapEntry::Occupied(_) => {
1081 debug_assert_eq!(
1082 self.tcp_listeners.len(),
1083 before,
1084 "a rejected duplicate add_tcp_listener must not mutate the map"
1085 );
1086 return Err(StateError::Exists {
1087 kind: ObjectKind::TcpListener,
1088 id: address.to_string(),
1089 });
1090 }
1091 };
1092 debug_assert!(
1093 self.tcp_listeners.contains_key(&address),
1094 "add_tcp_listener must insert the listener under its address"
1095 );
1096 debug_assert_eq!(
1097 self.tcp_listeners.len(),
1098 before + 1,
1099 "add_tcp_listener inserts exactly one entry on the vacant path"
1100 );
1101 Ok(())
1102 }
1103
1104 fn add_udp_listener(&mut self, listener: &UdpListenerConfig) -> Result<(), StateError> {
1105 let address = validated_listener_key(listener.listener_key())?;
1106 match self.udp_listeners.entry(address.clone()) {
1107 BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(listener.clone()),
1108 BTreeMapEntry::Occupied(_) => {
1109 return Err(StateError::Exists {
1110 kind: ObjectKind::UdpListener,
1111 id: address.to_string(),
1112 });
1113 }
1114 };
1115 Ok(())
1116 }
1117
1118 fn remove_listener(&mut self, remove: &RemoveListener) -> Result<(), StateError> {
1119 match ListenerType::try_from(remove.proxy).map_err(StateError::WrongFieldValue)? {
1120 ListenerType::Http => self.remove_http_listener(&remove.listener_key()),
1121 ListenerType::Https => self.remove_https_listener(&remove.listener_key()),
1122 ListenerType::Tcp => self.remove_tcp_listener(&remove.listener_key()),
1123 ListenerType::Udp => self.remove_udp_listener(&remove.listener_key()),
1124 }
1125 }
1126
1127 fn remove_http_listener(&mut self, address: &ListenerKey) -> Result<(), StateError> {
1128 let before = self.http_listeners.len();
1129 if self.http_listeners.remove(address).is_none() {
1130 debug_assert_eq!(
1131 self.http_listeners.len(),
1132 before,
1133 "a failed remove_http_listener must not mutate the map"
1134 );
1135 return Err(StateError::NoChange);
1136 }
1137 debug_assert!(
1138 !self.http_listeners.contains_key(address),
1139 "remove_http_listener must evict the address"
1140 );
1141 debug_assert_eq!(
1142 self.http_listeners.len(),
1143 before - 1,
1144 "remove_http_listener drops exactly one entry"
1145 );
1146 Ok(())
1147 }
1148
1149 fn remove_https_listener(&mut self, address: &ListenerKey) -> Result<(), StateError> {
1150 let before = self.https_listeners.len();
1151 if self.https_listeners.remove(address).is_none() {
1152 debug_assert_eq!(
1153 self.https_listeners.len(),
1154 before,
1155 "a failed remove_https_listener must not mutate the map"
1156 );
1157 return Err(StateError::NoChange);
1158 }
1159 debug_assert!(
1160 !self.https_listeners.contains_key(address),
1161 "remove_https_listener must evict the address"
1162 );
1163 debug_assert_eq!(
1164 self.https_listeners.len(),
1165 before - 1,
1166 "remove_https_listener drops exactly one entry"
1167 );
1168 Ok(())
1169 }
1170
1171 fn remove_tcp_listener(&mut self, address: &ListenerKey) -> Result<(), StateError> {
1172 let before = self.tcp_listeners.len();
1173 if self.tcp_listeners.remove(address).is_none() {
1174 debug_assert_eq!(
1175 self.tcp_listeners.len(),
1176 before,
1177 "a failed remove_tcp_listener must not mutate the map"
1178 );
1179 return Err(StateError::NoChange);
1180 }
1181 debug_assert!(
1182 !self.tcp_listeners.contains_key(address),
1183 "remove_tcp_listener must evict the address"
1184 );
1185 debug_assert_eq!(
1186 self.tcp_listeners.len(),
1187 before - 1,
1188 "remove_tcp_listener drops exactly one entry"
1189 );
1190 Ok(())
1191 }
1192
1193 fn remove_udp_listener(&mut self, address: &ListenerKey) -> Result<(), StateError> {
1194 if self.udp_listeners.remove(address).is_none() {
1195 return Err(StateError::NoChange);
1196 }
1197 Ok(())
1198 }
1199
1200 fn update_http_listener(&mut self, patch: &UpdateHttpListenerConfig) -> Result<(), StateError> {
1212 validate_h2_flood_knobs_http(patch)?;
1213
1214 let key = listener_key_for_update(
1215 &self.http_listeners,
1216 patch.address.into(),
1217 ObjectKind::HttpListener,
1218 )?;
1219 if let Some(ref v) = patch.sozu_id_header {
1220 validate_sozu_id_header(v)?;
1221 }
1222 if let Some(v) = patch.forwarded_headers {
1223 validate_forwarded_headers(v)?;
1224 }
1225
1226 let listener = self
1227 .http_listeners
1228 .get_mut(&key)
1229 .expect("listener_key_for_update returns a key present in the map");
1230
1231 if let Some(v) = patch.public_address {
1233 listener.public_address = Some(v);
1234 }
1235 if let Some(v) = patch.expect_proxy {
1236 listener.expect_proxy = v;
1237 }
1238 if let Some(ref v) = patch.sticky_name {
1239 listener.sticky_name = v.to_owned();
1240 }
1241 if let Some(v) = patch.front_timeout {
1242 listener.front_timeout = v;
1243 }
1244 if let Some(v) = patch.back_timeout {
1245 listener.back_timeout = v;
1246 }
1247 if let Some(v) = patch.connect_timeout {
1248 listener.connect_timeout = v;
1249 }
1250 if let Some(v) = patch.request_timeout {
1251 listener.request_timeout = v;
1252 }
1253 if let Some(v) = patch.elide_x_real_ip {
1254 listener.elide_x_real_ip = Some(v);
1255 }
1256 if let Some(v) = patch.send_x_real_ip {
1257 listener.send_x_real_ip = Some(v);
1258 }
1259 if let Some(patch_answers) = patch.http_answers.as_ref() {
1260 merge_custom_http_answers(&mut listener.http_answers, patch_answers);
1261 }
1262 merge_answers(&mut listener.answers, &patch.answers);
1263 if let Some(v) = patch.h2_max_rst_stream_per_window {
1265 listener.h2_max_rst_stream_per_window = Some(v);
1266 }
1267 if let Some(v) = patch.h2_max_ping_per_window {
1268 listener.h2_max_ping_per_window = Some(v);
1269 }
1270 if let Some(v) = patch.h2_max_settings_per_window {
1271 listener.h2_max_settings_per_window = Some(v);
1272 }
1273 if let Some(v) = patch.h2_max_empty_data_per_window {
1274 listener.h2_max_empty_data_per_window = Some(v);
1275 }
1276 if let Some(v) = patch.h2_max_continuation_frames {
1277 listener.h2_max_continuation_frames = Some(v);
1278 }
1279 if let Some(v) = patch.h2_max_glitch_count {
1280 listener.h2_max_glitch_count = Some(v);
1281 }
1282 if let Some(v) = patch.h2_initial_connection_window {
1283 listener.h2_initial_connection_window = Some(v);
1284 }
1285 if let Some(v) = patch.h2_max_concurrent_streams {
1286 listener.h2_max_concurrent_streams = Some(v);
1287 }
1288 if let Some(v) = patch.h2_stream_shrink_ratio {
1289 listener.h2_stream_shrink_ratio = Some(v);
1290 }
1291 if let Some(v) = patch.h2_max_rst_stream_lifetime {
1292 listener.h2_max_rst_stream_lifetime = Some(v);
1293 }
1294 if let Some(v) = patch.h2_max_rst_stream_abusive_lifetime {
1295 listener.h2_max_rst_stream_abusive_lifetime = Some(v);
1296 }
1297 if let Some(v) = patch.h2_max_rst_stream_emitted_lifetime {
1298 listener.h2_max_rst_stream_emitted_lifetime = Some(v);
1299 }
1300 if let Some(v) = patch.h2_max_header_list_size {
1301 listener.h2_max_header_list_size = Some(v);
1302 }
1303 if let Some(v) = patch.h2_max_header_table_size {
1304 listener.h2_max_header_table_size = Some(v);
1305 }
1306 if let Some(v) = patch.h2_max_header_fields {
1307 listener.h2_max_header_fields = Some(v);
1308 }
1309 if let Some(v) = patch.h2_stream_refusal_percent {
1310 listener.h2_stream_refusal_percent = Some(v);
1311 }
1312 if let Some(v) = patch.h2_stream_idle_timeout_seconds {
1313 listener.h2_stream_idle_timeout_seconds = Some(v);
1314 }
1315 if let Some(v) = patch.h2_graceful_shutdown_deadline_seconds {
1317 listener.h2_graceful_shutdown_deadline_seconds = Some(v);
1318 }
1319 if let Some(v) = patch.h2_max_window_update_stream0_per_window {
1320 listener.h2_max_window_update_stream0_per_window = Some(v);
1321 }
1322 if let Some(ref v) = patch.sozu_id_header {
1323 listener.sozu_id_header = Some(v.to_owned());
1324 }
1325 if let Some(v) = patch.forwarded_headers {
1326 listener.forwarded_headers = Some(v);
1327 }
1328 Ok(())
1329 }
1330
1331 pub fn client_auth_refresh(
1341 &self,
1342 listener: &HttpsListenerConfig,
1343 ) -> Option<UpdateHttpsListenerConfig> {
1344 let running = self.https_listeners.get(&listener.listener_key())?;
1345 let policy = listener.client_auth_policy();
1346 (running.client_auth_policy() != policy).then(|| UpdateHttpsListenerConfig {
1347 address: listener.address,
1348 client_auth_policy: Some(policy),
1349 ..Default::default()
1350 })
1351 }
1352
1353 fn update_https_listener(
1373 &mut self,
1374 patch: &UpdateHttpsListenerConfig,
1375 ) -> Result<(), StateError> {
1376 validate_h2_flood_knobs_https(patch)?;
1377
1378 let key = listener_key_for_update(
1379 &self.https_listeners,
1380 patch.address.into(),
1381 ObjectKind::HttpsListener,
1382 )?;
1383 if let Some(ref alpn_wrapper) = patch.alpn_protocols {
1384 validate_alpn_protocols(&alpn_wrapper.values)?;
1385 }
1386 if let Some(ref v) = patch.sozu_id_header {
1387 validate_sozu_id_header(v)?;
1388 }
1389 if let Some(v) = patch.forwarded_headers {
1390 validate_forwarded_headers(v)?;
1391 }
1392 if let Some(ref hsts) = patch.hsts {
1393 validate_hsts_patch(hsts)?;
1394 }
1395
1396 let listener = self
1397 .https_listeners
1398 .get_mut(&key)
1399 .expect("listener_key_for_update returns a key present in the map");
1400
1401 if let Some(v) = patch.public_address {
1403 listener.public_address = Some(v);
1404 }
1405 if let Some(v) = patch.expect_proxy {
1406 listener.expect_proxy = v;
1407 }
1408 if let Some(ref v) = patch.sticky_name {
1409 listener.sticky_name = v.to_owned();
1410 }
1411 if let Some(v) = patch.front_timeout {
1412 listener.front_timeout = v;
1413 }
1414 if let Some(v) = patch.back_timeout {
1415 listener.back_timeout = v;
1416 }
1417 if let Some(v) = patch.connect_timeout {
1418 listener.connect_timeout = v;
1419 }
1420 if let Some(v) = patch.request_timeout {
1421 listener.request_timeout = v;
1422 }
1423 if let Some(v) = patch.elide_x_real_ip {
1424 listener.elide_x_real_ip = Some(v);
1425 }
1426 if let Some(v) = patch.send_x_real_ip {
1427 listener.send_x_real_ip = Some(v);
1428 }
1429 if let Some(patch_answers) = patch.http_answers.as_ref() {
1430 merge_custom_http_answers(&mut listener.http_answers, patch_answers);
1431 }
1432 merge_answers(&mut listener.answers, &patch.answers);
1433 if let Some(hsts) = patch.hsts {
1436 listener.hsts = Some(hsts);
1437 }
1438 if let Some(ref alpn_wrapper) = patch.alpn_protocols {
1439 listener.alpn_protocols = alpn_wrapper.values.clone();
1441 }
1442 if let Some(v) = patch.strict_sni_binding {
1443 listener.strict_sni_binding = Some(v);
1444 }
1445 if let Some(v) = patch.disable_http11 {
1446 listener.disable_http11 = Some(v);
1447 }
1448 if let Some(ref policy) = patch.client_auth_policy {
1449 listener.apply_client_auth_policy(policy);
1450 }
1451 if let Some(v) = patch.h2_max_rst_stream_per_window {
1453 listener.h2_max_rst_stream_per_window = Some(v);
1454 }
1455 if let Some(v) = patch.h2_max_ping_per_window {
1456 listener.h2_max_ping_per_window = Some(v);
1457 }
1458 if let Some(v) = patch.h2_max_settings_per_window {
1459 listener.h2_max_settings_per_window = Some(v);
1460 }
1461 if let Some(v) = patch.h2_max_empty_data_per_window {
1462 listener.h2_max_empty_data_per_window = Some(v);
1463 }
1464 if let Some(v) = patch.h2_max_continuation_frames {
1465 listener.h2_max_continuation_frames = Some(v);
1466 }
1467 if let Some(v) = patch.h2_max_glitch_count {
1468 listener.h2_max_glitch_count = Some(v);
1469 }
1470 if let Some(v) = patch.h2_initial_connection_window {
1471 listener.h2_initial_connection_window = Some(v);
1472 }
1473 if let Some(v) = patch.h2_max_concurrent_streams {
1474 listener.h2_max_concurrent_streams = Some(v);
1475 }
1476 if let Some(v) = patch.h2_stream_shrink_ratio {
1477 listener.h2_stream_shrink_ratio = Some(v);
1478 }
1479 if let Some(v) = patch.h2_max_rst_stream_lifetime {
1480 listener.h2_max_rst_stream_lifetime = Some(v);
1481 }
1482 if let Some(v) = patch.h2_max_rst_stream_abusive_lifetime {
1483 listener.h2_max_rst_stream_abusive_lifetime = Some(v);
1484 }
1485 if let Some(v) = patch.h2_max_rst_stream_emitted_lifetime {
1486 listener.h2_max_rst_stream_emitted_lifetime = Some(v);
1487 }
1488 if let Some(v) = patch.h2_max_header_list_size {
1489 listener.h2_max_header_list_size = Some(v);
1490 }
1491 if let Some(v) = patch.h2_max_header_table_size {
1492 listener.h2_max_header_table_size = Some(v);
1493 }
1494 if let Some(v) = patch.h2_max_header_fields {
1495 listener.h2_max_header_fields = Some(v);
1496 }
1497 if let Some(v) = patch.h2_stream_refusal_percent {
1498 listener.h2_stream_refusal_percent = Some(v);
1499 }
1500 if let Some(v) = patch.h2_stream_idle_timeout_seconds {
1501 listener.h2_stream_idle_timeout_seconds = Some(v);
1502 }
1503 if let Some(v) = patch.h2_graceful_shutdown_deadline_seconds {
1505 listener.h2_graceful_shutdown_deadline_seconds = Some(v);
1506 }
1507 if let Some(v) = patch.h2_max_window_update_stream0_per_window {
1508 listener.h2_max_window_update_stream0_per_window = Some(v);
1509 }
1510 if let Some(ref v) = patch.sozu_id_header {
1511 listener.sozu_id_header = Some(v.to_owned());
1512 }
1513 if let Some(v) = patch.forwarded_headers {
1514 listener.forwarded_headers = Some(v);
1515 }
1516 Ok(())
1517 }
1518
1519 fn update_tcp_listener(&mut self, patch: &UpdateTcpListenerConfig) -> Result<(), StateError> {
1524 let key = listener_key_for_update(
1525 &self.tcp_listeners,
1526 patch.address.into(),
1527 ObjectKind::TcpListener,
1528 )?;
1529 let listener = self
1530 .tcp_listeners
1531 .get_mut(&key)
1532 .expect("listener_key_for_update returns a key present in the map");
1533
1534 if let Some(v) = patch.public_address {
1535 listener.public_address = Some(v);
1536 }
1537 if let Some(v) = patch.expect_proxy {
1538 listener.expect_proxy = v;
1539 }
1540 if let Some(v) = patch.front_timeout {
1541 listener.front_timeout = v;
1542 }
1543 if let Some(v) = patch.back_timeout {
1544 listener.back_timeout = v;
1545 }
1546 if let Some(v) = patch.connect_timeout {
1547 listener.connect_timeout = v;
1548 }
1549 Ok(())
1550 }
1551
1552 fn update_udp_listener(&mut self, patch: &UpdateUdpListenerConfig) -> Result<(), StateError> {
1557 let key = listener_key_for_update(
1558 &self.udp_listeners,
1559 patch.address.into(),
1560 ObjectKind::UdpListener,
1561 )?;
1562 let listener = self
1563 .udp_listeners
1564 .get_mut(&key)
1565 .expect("listener_key_for_update returns a key present in the map");
1566
1567 if let Some(v) = patch.public_address {
1568 listener.public_address = Some(v);
1569 }
1570 if let Some(v) = patch.front_timeout {
1571 listener.front_timeout = v;
1572 }
1573 if let Some(v) = patch.back_timeout {
1574 listener.back_timeout = v;
1575 }
1576 if let Some(v) = patch.max_rx_datagram_size {
1577 listener.max_rx_datagram_size = v;
1578 }
1579 if let Some(v) = patch.max_flows {
1580 listener.max_flows = v;
1581 }
1582 Ok(())
1583 }
1584
1585 fn activate_listener(&mut self, activate: &ActivateListener) -> Result<(), StateError> {
1586 match ListenerType::try_from(activate.proxy).map_err(StateError::WrongFieldValue)? {
1587 ListenerType::Http => self
1588 .http_listeners
1589 .get_mut(&activate.listener_key())
1590 .map(|listener| listener.active = true)
1591 .ok_or(StateError::NotFound {
1592 kind: ObjectKind::HttpListener,
1593 id: activate.listener_key().to_string(),
1594 }),
1595 ListenerType::Https => self
1596 .https_listeners
1597 .get_mut(&activate.listener_key())
1598 .map(|listener| listener.active = true)
1599 .ok_or(StateError::NotFound {
1600 kind: ObjectKind::HttpsListener,
1601 id: activate.listener_key().to_string(),
1602 }),
1603 ListenerType::Tcp => self
1604 .tcp_listeners
1605 .get_mut(&activate.listener_key())
1606 .map(|listener| listener.active = true)
1607 .ok_or(StateError::NotFound {
1608 kind: ObjectKind::TcpListener,
1609 id: activate.listener_key().to_string(),
1610 }),
1611 ListenerType::Udp => self
1612 .udp_listeners
1613 .get_mut(&activate.listener_key())
1614 .map(|listener| listener.active = true)
1615 .ok_or(StateError::NotFound {
1616 kind: ObjectKind::UdpListener,
1617 id: activate.listener_key().to_string(),
1618 }),
1619 }
1620 }
1621
1622 fn deactivate_listener(&mut self, deactivate: &DeactivateListener) -> Result<(), StateError> {
1623 match ListenerType::try_from(deactivate.proxy).map_err(StateError::WrongFieldValue)? {
1624 ListenerType::Http => self
1625 .http_listeners
1626 .get_mut(&deactivate.listener_key())
1627 .map(|listener| listener.active = false)
1628 .ok_or(StateError::NotFound {
1629 kind: ObjectKind::HttpListener,
1630 id: deactivate.listener_key().to_string(),
1631 }),
1632 ListenerType::Https => self
1633 .https_listeners
1634 .get_mut(&deactivate.listener_key())
1635 .map(|listener| listener.active = false)
1636 .ok_or(StateError::NotFound {
1637 kind: ObjectKind::HttpsListener,
1638 id: deactivate.listener_key().to_string(),
1639 }),
1640 ListenerType::Tcp => self
1641 .tcp_listeners
1642 .get_mut(&deactivate.listener_key())
1643 .map(|listener| listener.active = false)
1644 .ok_or(StateError::NotFound {
1645 kind: ObjectKind::TcpListener,
1646 id: deactivate.listener_key().to_string(),
1647 }),
1648 ListenerType::Udp => self
1649 .udp_listeners
1650 .get_mut(&deactivate.listener_key())
1651 .map(|listener| listener.active = false)
1652 .ok_or(StateError::NotFound {
1653 kind: ObjectKind::UdpListener,
1654 id: deactivate.listener_key().to_string(),
1655 }),
1656 }
1657 }
1658
1659 fn add_frontend_to(
1666 fronts: &mut BTreeMap<String, HttpFrontend>,
1667 front: &RequestHttpFrontend,
1668 kind: ObjectKind,
1669 ) -> Result<(), StateError> {
1670 let front_as_key = front.to_string();
1671 let before = fronts.len();
1672
1673 match fronts.entry(front.to_string()) {
1674 BTreeMapEntry::Vacant(e) => {
1675 e.insert(front.clone().to_frontend().map_err(|into_error| {
1676 StateError::FrontendConversion {
1677 frontend: front_as_key,
1678 error: into_error.to_string(),
1679 }
1680 })?)
1681 }
1682 BTreeMapEntry::Occupied(_) => {
1683 debug_assert_eq!(
1684 fronts.len(),
1685 before,
1686 "a rejected duplicate {kind:?} add must not mutate the map"
1687 );
1688 return Err(StateError::Exists {
1689 kind,
1690 id: front.to_string(),
1691 });
1692 }
1693 };
1694 debug_assert!(
1697 fronts.contains_key(&front.to_string()),
1698 "adding a {kind:?} must insert the route key on success"
1699 );
1700 debug_assert_eq!(
1701 fronts.len(),
1702 before + 1,
1703 "adding a {kind:?} inserts exactly one entry on success"
1704 );
1705 Ok(())
1706 }
1707
1708 fn remove_frontend_from(
1711 fronts: &mut BTreeMap<String, HttpFrontend>,
1712 front: &RequestHttpFrontend,
1713 kind: ObjectKind,
1714 ) -> Result<(), StateError> {
1715 let key = front.to_string();
1716 let before = fronts.len();
1717 match fronts.get(&key) {
1722 Some(stored) if stored.cluster_id.as_deref() == front.cluster_id.as_deref() => {
1723 fronts.remove(&key);
1724 }
1725 _ => {
1726 debug_assert_eq!(fronts.len(), before, "a refused removal must not mutate");
1727 return Err(StateError::NotFound {
1728 kind,
1729 id: front.to_string(),
1730 });
1731 }
1732 }
1733 debug_assert!(
1734 !fronts.contains_key(&key),
1735 "removing a frontend must evict the route key"
1736 );
1737 debug_assert_eq!(
1738 fronts.len(),
1739 before - 1,
1740 "removing a frontend drops exactly one entry"
1741 );
1742 Ok(())
1743 }
1744
1745 fn add_http_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1746 Self::add_frontend_to(&mut self.http_fronts, front, ObjectKind::HttpFrontend)
1747 }
1748
1749 fn add_https_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1750 Self::add_frontend_to(&mut self.https_fronts, front, ObjectKind::HttpsFrontend)
1751 }
1752
1753 fn remove_http_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1754 Self::remove_frontend_from(&mut self.http_fronts, front, ObjectKind::HttpFrontend)
1755 }
1756
1757 fn remove_https_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1758 Self::remove_frontend_from(&mut self.https_fronts, front, ObjectKind::HttpsFrontend)
1759 }
1760
1761 fn add_certificate(&mut self, add: &AddCertificate) -> Result<(), StateError> {
1762 let fingerprint = add
1763 .certificate
1764 .fingerprint()
1765 .map_err(StateError::AddCertificate)?;
1766
1767 let mut add = add.clone();
1778 add.certificate
1779 .apply_overriding_names()
1780 .map_err(StateError::AddCertificate)?;
1781
1782 let entry = self.certificates.entry(add.address.into()).or_default();
1783
1784 if entry.contains_key(&fingerprint) {
1785 let names_bytes = add
1786 .certificate
1787 .names
1788 .iter()
1789 .fold(0usize, |total, name| total.saturating_add(name.len()));
1790 info!(
1791 "Skip loading certificate with fingerprint_bytes={} names_count={} names_bytes={} on listener '{}', the certificate is already present.",
1792 fingerprint.0.len(),
1793 add.certificate.names.len(),
1794 names_bytes,
1795 add.address
1796 );
1797 return Ok(());
1798 }
1799
1800 let before = entry.len();
1801 entry.insert(fingerprint.clone(), add.certificate);
1802 debug_assert!(
1803 entry.contains_key(&fingerprint),
1804 "add_certificate must insert the fingerprint under its address"
1805 );
1806 debug_assert_eq!(
1807 entry.len(),
1808 before + 1,
1809 "add_certificate inserts exactly one fingerprint on the new path"
1810 );
1811 Ok(())
1812 }
1813
1814 fn remove_certificate(&mut self, remove: &RemoveCertificate) -> Result<(), StateError> {
1815 let fingerprint = Fingerprint(
1816 hex::decode(&remove.fingerprint)
1817 .map_err(|decode_error| StateError::RemoveCertificate(decode_error.to_string()))?,
1818 );
1819 let address = remove.address.into();
1820
1821 let became_empty = if let Some(index) = self.certificates.get_mut(&address) {
1822 index.remove(&fingerprint);
1823 debug_assert!(
1824 !index.contains_key(&fingerprint),
1825 "remove_certificate must evict the fingerprint when the address is known"
1826 );
1827 index.is_empty()
1828 } else {
1829 false
1830 };
1831
1832 if became_empty {
1844 self.certificates.remove(&address);
1845 }
1846
1847 Ok(())
1848 }
1849
1850 fn replace_certificate(&mut self, replace: &ReplaceCertificate) -> Result<(), StateError> {
1868 let replace_address = replace.address.into();
1869 let old_fingerprint = Fingerprint(
1870 hex::decode(&replace.old_fingerprint)
1871 .map_err(|decode_error| StateError::RemoveCertificate(decode_error.to_string()))?,
1872 );
1873
1874 if !self.certificates.contains_key(&replace_address) {
1875 return Err(StateError::NotFound {
1876 kind: ObjectKind::Certificate,
1877 id: replace.address.to_string(),
1878 });
1879 }
1880
1881 let new_fingerprint = Fingerprint(
1882 calculate_fingerprint(replace.new_certificate.certificate.as_bytes()).map_err(
1883 |fingerprint_err| StateError::ReplaceCertificate(fingerprint_err.to_string()),
1884 )?,
1885 );
1886
1887 self.certificates
1888 .get_mut(&replace_address)
1889 .ok_or(StateError::ReplaceCertificate(
1890 "Unlikely error. This entry in the certificate hashmap should be present"
1891 .to_string(),
1892 ))?
1893 .insert(new_fingerprint.clone(), replace.new_certificate.clone());
1894
1895 if !self
1896 .certificates
1897 .get(&replace_address)
1898 .ok_or(StateError::ReplaceCertificate(
1899 "Unlikely error. This entry in the certificate hashmap should be present"
1900 .to_string(),
1901 ))?
1902 .contains_key(&new_fingerprint)
1903 {
1904 return Err(StateError::ReplaceCertificate(format!(
1905 "Failed to insert the new certificate for address {}",
1906 replace.address
1907 )));
1908 }
1909
1910 if new_fingerprint != old_fingerprint {
1915 self.certificates
1916 .get_mut(&replace_address)
1917 .map(|certs| certs.remove(&old_fingerprint));
1918 }
1919
1920 debug_assert!(
1923 self.certificates
1924 .get(&replace_address)
1925 .is_some_and(|certs| certs.contains_key(&new_fingerprint)),
1926 "replace_certificate must leave the new fingerprint present"
1927 );
1928 debug_assert!(
1929 new_fingerprint == old_fingerprint
1930 || self
1931 .certificates
1932 .get(&replace_address)
1933 .is_none_or(|certs| !certs.contains_key(&old_fingerprint)),
1934 "replace_certificate must evict the old fingerprint unless it equals the new one"
1935 );
1936 Ok(())
1937 }
1938
1939 fn add_tcp_frontend(&mut self, front: &RequestTcpFrontend) -> Result<(), StateError> {
1947 let address: SocketAddr = front.address.into();
1948
1949 if self
1952 .clusters
1953 .get(front.cluster_id.as_str())
1954 .is_some_and(|cluster| {
1955 cluster.affinity_header.is_some() || cluster.affinity_cookie.is_some()
1956 })
1957 {
1958 return Err(StateError::InvalidTcpFrontend {
1959 address,
1960 reason: TCP_CLUSTER_AFFINITY_KEY.to_owned(),
1961 });
1962 }
1963
1964 let normalized_sni = match &front.sni {
1972 Some(sni) => Some(validate_sni_pattern(sni).map_err(|config_error| {
1973 StateError::InvalidTcpFrontend {
1974 address,
1975 reason: format!("sni {sni:?} failed SNI shape validation: {config_error}"),
1976 }
1977 })?),
1978 None => None,
1979 };
1980 let alpn = canonical_tcp_alpn(&front.alpn);
1981
1982 if normalized_sni.is_none() && !alpn.is_empty() {
1988 return Err(StateError::InvalidTcpFrontend {
1989 address,
1990 reason: format!(
1991 "alpn = {alpn:?} set without sni: alpn only matches within an SNI-scoped \
1992 preread, so a frontend without sni would silently ignore its alpn list"
1993 ),
1994 });
1995 }
1996
1997 let total_before = self.count_tcp_frontends_raw();
1998
1999 let mut listener_has_no_sni_front = false;
2005 let mut listener_has_sni_front = false;
2006 for existing in self
2007 .tcp_fronts
2008 .values()
2009 .flatten()
2010 .filter(|existing| existing.address == address)
2011 {
2012 if tcp_frontend_matches(existing, address, &normalized_sni, &alpn) {
2013 return Err(StateError::Exists {
2014 kind: ObjectKind::TcpFrontend,
2015 id: format!(
2016 "TcpFrontend {{ address: {address}, sni: {normalized_sni:?}, \
2017 alpn: {alpn:?} }}"
2018 ),
2019 });
2020 }
2021 match &existing.sni {
2022 None => listener_has_no_sni_front = true,
2023 Some(existing_sni) => {
2024 listener_has_sni_front = true;
2025 if normalized_sni.as_deref() != Some(existing_sni.as_str()) {
2032 continue;
2033 }
2034 let existing_is_catch_all = existing.alpn.is_empty();
2035 let new_is_catch_all = alpn.is_empty();
2036 if existing_is_catch_all && new_is_catch_all {
2037 return Err(StateError::InvalidTcpFrontend {
2038 address,
2039 reason: format!(
2040 "sni {existing_sni:?} already has a catch-all (empty alpn) \
2041 frontend: at most one frontend per (address, sni) may omit \
2042 alpn"
2043 ),
2044 });
2045 }
2046 if !existing_is_catch_all
2051 && !new_is_catch_all
2052 && let Some(overlap) = alpn
2053 .iter()
2054 .find(|protocol| existing.alpn.contains(protocol))
2055 {
2056 return Err(StateError::InvalidTcpFrontend {
2057 address,
2058 reason: format!(
2059 "sni {existing_sni:?} already has a frontend matching ALPN \
2060 protocol {overlap:?}: ALPN matchers for the same (address, \
2061 sni) must not overlap"
2062 ),
2063 });
2064 }
2065 }
2066 }
2067 }
2068 match &normalized_sni {
2069 None if listener_has_sni_front => {
2070 return Err(StateError::InvalidTcpFrontend {
2071 address,
2072 reason: "a no-SNI frontend cannot be added to a listener that already has \
2073 SNI-scoped routes"
2074 .to_string(),
2075 });
2076 }
2077 Some(_) if listener_has_no_sni_front => {
2078 return Err(StateError::InvalidTcpFrontend {
2079 address,
2080 reason: "an SNI-scoped frontend cannot be added to a listener that already \
2081 has a no-SNI catch-all cluster"
2082 .to_string(),
2083 });
2084 }
2085 _ => {}
2086 }
2087 debug_assert_eq!(
2091 self.count_tcp_frontends_raw(),
2092 total_before,
2093 "add_tcp_frontend must not mutate tcp_fronts before every admission check has \
2094 passed"
2095 );
2096
2097 let tcp_frontend = TcpFrontend {
2098 cluster_id: front.cluster_id.as_str().into(),
2099 address,
2100 tags: front.tags.clone(),
2101 sni: normalized_sni,
2102 alpn,
2103 };
2104 let tcp_frontends = self
2105 .tcp_fronts
2106 .entry(tcp_frontend.cluster_id.clone())
2107 .or_default();
2108 let before = tcp_frontends.len();
2109 debug_assert_eq!(
2110 *tcp_frontend.cluster_id, *front.cluster_id,
2111 "the built frontend must carry its bucket's cluster_id"
2112 );
2113 tcp_frontends.push(tcp_frontend);
2114 debug_assert_eq!(
2115 tcp_frontends.len(),
2116 before + 1,
2117 "add_tcp_frontend appends exactly one entry"
2118 );
2119 Ok(())
2120 }
2121
2122 fn remove_tcp_frontend(
2123 &mut self,
2124 front_to_remove: &RequestTcpFrontend,
2125 ) -> Result<(), StateError> {
2126 let tcp_frontends =
2127 self.tcp_fronts
2128 .get_mut(front_to_remove.cluster_id.as_str())
2129 .ok_or(StateError::NotFound {
2130 kind: ObjectKind::TcpFrontend,
2131 id: format!(
2132 "RequestTcpFrontend {{ cluster_id: {:?}, address: {:?}, tags: {:?}, sni: {:?}, alpn: {:?} }}",
2133 front_to_remove.cluster_id,
2134 front_to_remove.address,
2135 front_to_remove.tags,
2136 front_to_remove.sni,
2137 front_to_remove.alpn,
2138 ),
2139 })?;
2140
2141 let len = tcp_frontends.len();
2142 let remove_address: SocketAddr = front_to_remove.address.into();
2143 let remove_sni = front_to_remove.sni.as_deref().map(str::to_ascii_lowercase);
2153 let remove_alpn = canonical_tcp_alpn(&front_to_remove.alpn);
2154 tcp_frontends.retain(|front| {
2159 !tcp_frontend_matches(front, remove_address, &remove_sni, &remove_alpn)
2160 });
2161 let after = tcp_frontends.len();
2162 if after == len {
2163 return Err(StateError::NoChange);
2164 }
2165 debug_assert_eq!(
2170 after,
2171 len - 1,
2172 "remove_tcp_frontend drops exactly one entry"
2173 );
2174 debug_assert!(
2175 !tcp_frontends.iter().any(|f| tcp_frontend_matches(
2176 f,
2177 remove_address,
2178 &remove_sni,
2179 &remove_alpn
2180 )),
2181 "remove_tcp_frontend must leave no frontend matching the removed (address, sni, alpn)"
2182 );
2183 Ok(())
2184 }
2185
2186 fn add_udp_frontend(&mut self, front: &RequestUdpFrontend) -> Result<(), StateError> {
2212 let address: SocketAddr = front.address.into();
2213 let udp_frontend = UdpFrontend {
2214 cluster_id: front.cluster_id.as_str().into(),
2215 address,
2216 tags: front.tags.clone(),
2217 };
2218
2219 let total_before = self.count_udp_frontends_raw();
2220
2221 if let Some(current) = self
2226 .udp_fronts
2227 .values()
2228 .flatten()
2229 .find(|existing| existing.address == address)
2230 {
2231 if *current == udp_frontend {
2236 return Err(StateError::Exists {
2237 kind: ObjectKind::UdpFrontend,
2238 id: format!("{udp_frontend:?}"),
2239 });
2240 }
2241 return Err(StateError::UdpFrontendAddressTaken {
2242 address,
2243 cluster_id: udp_frontend.cluster_id.clone(),
2244 current_cluster_id: current.cluster_id.clone(),
2245 current_tags: current.tags.clone(),
2246 });
2247 }
2248
2249 debug_assert_eq!(
2253 self.count_udp_frontends_raw(),
2254 total_before,
2255 "add_udp_frontend must not mutate udp_fronts before every admission check has passed"
2256 );
2257
2258 let udp_frontends = self
2259 .udp_fronts
2260 .entry(udp_frontend.cluster_id.clone())
2261 .or_default();
2262 let before = udp_frontends.len();
2263 udp_frontends.push(udp_frontend);
2264 debug_assert_eq!(
2265 udp_frontends.len(),
2266 before + 1,
2267 "add_udp_frontend appends exactly one entry"
2268 );
2269 debug_assert_eq!(
2273 self.udp_fronts
2274 .values()
2275 .flatten()
2276 .filter(|existing| existing.address == address)
2277 .count(),
2278 1,
2279 "add_udp_frontend must leave exactly one frontend on an address, across all clusters"
2280 );
2281 Ok(())
2282 }
2283
2284 fn remove_udp_frontend(
2285 &mut self,
2286 front_to_remove: &RequestUdpFrontend,
2287 ) -> Result<(), StateError> {
2288 let udp_frontends = self
2289 .udp_fronts
2290 .get_mut(front_to_remove.cluster_id.as_str())
2291 .ok_or(StateError::NotFound {
2292 kind: ObjectKind::UdpFrontend,
2293 id: format!("{front_to_remove:?}"),
2294 })?;
2295
2296 let len = udp_frontends.len();
2297 let remove_address: SocketAddr = front_to_remove.address.into();
2298 let matches_removal = |front: &UdpFrontend| {
2308 front.address == remove_address && front.tags == front_to_remove.tags
2309 };
2310 udp_frontends.retain(|front| !matches_removal(front));
2311 let after = udp_frontends.len();
2312 if after == len {
2313 return Err(StateError::NoChange);
2314 }
2315 debug_assert_eq!(
2320 after,
2321 len - 1,
2322 "remove_udp_frontend drops exactly one entry"
2323 );
2324 debug_assert!(
2325 !udp_frontends.iter().any(matches_removal),
2326 "remove_udp_frontend must leave no frontend matching the removed (address, tags)"
2327 );
2328 Ok(())
2329 }
2330
2331 fn add_backend(&mut self, add_backend: &AddBackend) -> Result<(), StateError> {
2332 let backend = Backend {
2333 address: add_backend.address.into(),
2334 cluster_id: add_backend.cluster_id.as_str().into(),
2335 backend_id: add_backend.backend_id.clone(),
2336 sticky_id: add_backend.sticky_id.clone(),
2337 load_balancing_parameters: add_backend.load_balancing_parameters,
2338 backup: add_backend.backup,
2339 };
2340 let backends = self.backends.entry(backend.cluster_id.clone()).or_default();
2341 let backend_id = backend.backend_id.clone();
2342 let backend_address = backend.address;
2343 let before = backends.len();
2344
2345 let was_present = backends
2350 .iter()
2351 .any(|b| b.backend_id == backend_id && b.address == backend_address);
2352 backends.retain(|b| b.backend_id != backend.backend_id || b.address != backend.address);
2353 debug_assert_eq!(
2354 backends.len(),
2355 before - was_present as usize,
2356 "the upsert retain must drop exactly the prior copy iff it existed"
2357 );
2358 backends.push(backend);
2359 backends.sort();
2360
2361 debug_assert_eq!(
2362 backends.len(),
2363 before + (!was_present) as usize,
2364 "add_backend grows the bucket by one iff the backend was new"
2365 );
2366 debug_assert_eq!(
2367 backends
2368 .iter()
2369 .filter(|b| b.backend_id == backend_id && b.address == backend_address)
2370 .count(),
2371 1,
2372 "exactly one copy of the upserted backend must remain"
2373 );
2374 Ok(())
2375 }
2376
2377 fn remove_backend(&mut self, backend: &RemoveBackend) -> Result<(), StateError> {
2378 let backend_list =
2379 self.backends
2380 .get_mut(backend.cluster_id.as_str())
2381 .ok_or(StateError::NotFound {
2382 kind: ObjectKind::Backend,
2383 id: backend.backend_id.to_owned(),
2384 })?;
2385
2386 let len = backend_list.len();
2387 let remove_address: SocketAddr = backend.address.into();
2388 backend_list.retain(|b| b.backend_id != backend.backend_id || b.address != remove_address);
2389 backend_list.sort();
2390 let after = backend_list.len();
2391 if after == len {
2392 return Err(StateError::NoChange);
2393 }
2394 debug_assert_eq!(after, len - 1, "remove_backend drops exactly one entry");
2397 debug_assert!(
2398 !backend_list
2399 .iter()
2400 .any(|b| b.backend_id == backend.backend_id && b.address == remove_address),
2401 "remove_backend must leave no backend matching (backend_id, address)"
2402 );
2403 Ok(())
2404 }
2405
2406 fn generate_requests(&self) -> Vec<Request> {
2408 let mut v: Vec<Request> = Vec::new();
2409
2410 for listener in self.http_listeners.values() {
2411 v.push(RequestType::AddHttpListener(listener.clone()).into());
2412 if listener.active {
2413 v.push(
2414 RequestType::ActivateListener(ActivateListener {
2415 interface: listener.interface.clone(),
2416 address: listener.address,
2417 proxy: ListenerType::Http.into(),
2418 from_scm: false,
2419 })
2420 .into(),
2421 );
2422 }
2423 }
2424
2425 for listener in self.https_listeners.values() {
2426 v.push(RequestType::add_https_listener(listener.clone()).into());
2427 if listener.active {
2428 v.push(
2429 RequestType::ActivateListener(ActivateListener {
2430 interface: listener.interface.clone(),
2431 address: listener.address,
2432 proxy: ListenerType::Https.into(),
2433 from_scm: false,
2434 })
2435 .into(),
2436 );
2437 }
2438 }
2439
2440 for listener in self.tcp_listeners.values() {
2441 v.push(RequestType::AddTcpListener(listener.clone()).into());
2442 if listener.active {
2443 v.push(
2444 RequestType::ActivateListener(ActivateListener {
2445 interface: listener.interface.clone(),
2446 address: listener.address,
2447 proxy: ListenerType::Tcp.into(),
2448 from_scm: false,
2449 })
2450 .into(),
2451 );
2452 }
2453 }
2454
2455 for listener in self.udp_listeners.values() {
2456 v.push(RequestType::AddUdpListener(listener.clone()).into());
2457 if listener.active {
2458 v.push(
2459 RequestType::ActivateListener(ActivateListener {
2460 interface: listener.interface.clone(),
2461 address: listener.address,
2462 proxy: ListenerType::Udp.into(),
2463 from_scm: false,
2464 })
2465 .into(),
2466 );
2467 }
2468 }
2469
2470 for cluster in self.clusters.values() {
2471 v.push(RequestType::AddCluster(cluster.clone()).into());
2472 }
2473
2474 for front in self.http_fronts.values() {
2475 v.push(RequestType::AddHttpFrontend(front.clone().into()).into());
2476 }
2477
2478 for (front, certs) in self.certificates.iter() {
2479 for certificate_and_key in certs.values() {
2480 v.push(
2481 RequestType::AddCertificate(AddCertificate {
2482 address: SocketAddress::from(*front),
2483 certificate: certificate_and_key.clone(),
2484 expired_at: None,
2485 })
2486 .into(),
2487 );
2488 }
2489 }
2490
2491 for front in self.https_fronts.values() {
2492 v.push(RequestType::AddHttpsFrontend(front.clone().into()).into());
2493 }
2494
2495 for front_list in self.tcp_fronts.values() {
2496 for front in front_list {
2497 v.push(RequestType::AddTcpFrontend(front.clone().into()).into());
2498 }
2499 }
2500
2501 for front_list in self.udp_fronts.values() {
2502 for front in front_list {
2503 v.push(RequestType::AddUdpFrontend(front.clone().into()).into());
2504 }
2505 }
2506
2507 for backend_list in self.backends.values() {
2508 for backend in backend_list {
2509 v.push(RequestType::AddBackend(backend.clone().to_add_backend()).into());
2510 }
2511 }
2512
2513 #[cfg(debug_assertions)]
2519 {
2520 let mut replayed = ConfigState::new();
2521 for request in &v {
2522 debug_assert!(
2523 replayed.dispatch(request).is_ok(),
2524 "every request from generate_requests must replay cleanly"
2525 );
2526 }
2527 debug_assert!(
2528 replayed.clusters == self.clusters
2529 && replayed.backends == self.backends
2530 && replayed.http_listeners == self.http_listeners
2531 && replayed.https_listeners == self.https_listeners
2532 && replayed.tcp_listeners == self.tcp_listeners
2533 && replayed.http_fronts == self.http_fronts
2534 && replayed.https_fronts == self.https_fronts
2535 && replayed.tcp_fronts == self.tcp_fronts
2536 && replayed.certificates == self.certificates,
2537 "replaying generate_requests into a fresh state must reproduce self"
2538 );
2539 }
2540
2541 v
2542 }
2543
2544 pub fn generate_activate_requests(&self) -> Vec<Request> {
2545 let mut v: Vec<Request> = Vec::new();
2546 for front in self
2547 .http_listeners
2548 .iter()
2549 .filter(|(_, listener)| listener.active)
2550 .map(|(k, _)| k)
2551 {
2552 v.push(
2553 RequestType::ActivateListener(ActivateListener {
2554 interface: front.interface.clone(),
2555 address: SocketAddress::from(front.address),
2556 proxy: ListenerType::Http.into(),
2557 from_scm: false,
2558 })
2559 .into(),
2560 );
2561 }
2562
2563 for front in self
2564 .https_listeners
2565 .iter()
2566 .filter(|(_, listener)| listener.active)
2567 .map(|(k, _)| k)
2568 {
2569 v.push(
2570 RequestType::ActivateListener(ActivateListener {
2571 interface: front.interface.clone(),
2572 address: SocketAddress::from(front.address),
2573 proxy: ListenerType::Https.into(),
2574 from_scm: false,
2575 })
2576 .into(),
2577 );
2578 }
2579 for front in self
2580 .tcp_listeners
2581 .iter()
2582 .filter(|(_, listener)| listener.active)
2583 .map(|(k, _)| k)
2584 {
2585 v.push(
2586 RequestType::ActivateListener(ActivateListener {
2587 interface: front.interface.clone(),
2588 address: SocketAddress::from(front.address),
2589 proxy: ListenerType::Tcp.into(),
2590 from_scm: false,
2591 })
2592 .into(),
2593 );
2594 }
2595 for front in self
2596 .udp_listeners
2597 .iter()
2598 .filter(|(_, listener)| listener.active)
2599 .map(|(k, _)| k)
2600 {
2601 v.push(
2602 RequestType::ActivateListener(ActivateListener {
2603 interface: front.interface.clone(),
2604 address: SocketAddress::from(front.address),
2605 proxy: ListenerType::Udp.into(),
2606 from_scm: false,
2607 })
2608 .into(),
2609 );
2610 }
2611
2612 #[cfg(debug_assertions)]
2615 {
2616 let active_listeners = self.http_listeners.values().filter(|l| l.active).count()
2617 + self.https_listeners.values().filter(|l| l.active).count()
2618 + self.tcp_listeners.values().filter(|l| l.active).count()
2619 + self.udp_listeners.values().filter(|l| l.active).count();
2620 debug_assert_eq!(
2621 v.len(),
2622 active_listeners,
2623 "generate_activate_requests emits one request per active listener"
2624 );
2625 debug_assert!(
2626 v.iter()
2627 .all(|r| matches!(r.request_type, Some(RequestType::ActivateListener(_)))),
2628 "generate_activate_requests must emit only ActivateListener requests"
2629 );
2630 }
2631
2632 v
2633 }
2634
2635 pub fn diff(&self, other: &ConfigState) -> Vec<Request> {
2636 let my_tcp_listeners: HashSet<&ListenerKey> = self.tcp_listeners.keys().collect();
2637 let their_tcp_listeners: HashSet<&ListenerKey> = other.tcp_listeners.keys().collect();
2638 let removed_tcp_listeners = my_tcp_listeners.difference(&their_tcp_listeners);
2639 let added_tcp_listeners = their_tcp_listeners.difference(&my_tcp_listeners);
2640
2641 let my_udp_listeners: HashSet<&ListenerKey> = self.udp_listeners.keys().collect();
2642 let their_udp_listeners: HashSet<&ListenerKey> = other.udp_listeners.keys().collect();
2643 let removed_udp_listeners = my_udp_listeners.difference(&their_udp_listeners);
2644 let added_udp_listeners = their_udp_listeners.difference(&my_udp_listeners);
2645
2646 let my_http_listeners: HashSet<&ListenerKey> = self.http_listeners.keys().collect();
2647 let their_http_listeners: HashSet<&ListenerKey> = other.http_listeners.keys().collect();
2648 let removed_http_listeners = my_http_listeners.difference(&their_http_listeners);
2649 let added_http_listeners = their_http_listeners.difference(&my_http_listeners);
2650
2651 let my_https_listeners: HashSet<&ListenerKey> = self.https_listeners.keys().collect();
2652 let their_https_listeners: HashSet<&ListenerKey> = other.https_listeners.keys().collect();
2653 let removed_https_listeners = my_https_listeners.difference(&their_https_listeners);
2654 let added_https_listeners = their_https_listeners.difference(&my_https_listeners);
2655
2656 let mut v: Vec<Request> = vec![];
2657
2658 for address in removed_tcp_listeners {
2659 if self.tcp_listeners[*address].active {
2660 v.push(
2661 RequestType::DeactivateListener(DeactivateListener {
2662 interface: address.interface.clone(),
2663 address: SocketAddress::from(address.address),
2664 proxy: ListenerType::Tcp.into(),
2665 to_scm: false,
2666 })
2667 .into(),
2668 );
2669 }
2670
2671 v.push(
2672 RequestType::RemoveListener(RemoveListener {
2673 interface: address.interface.clone(),
2674 address: SocketAddress::from(address.address),
2675 proxy: ListenerType::Tcp.into(),
2676 })
2677 .into(),
2678 );
2679 }
2680
2681 for address in added_tcp_listeners.clone() {
2682 v.push(RequestType::AddTcpListener(other.tcp_listeners[*address].clone()).into());
2683
2684 if other.tcp_listeners[*address].active {
2685 v.push(
2686 RequestType::ActivateListener(ActivateListener {
2687 interface: address.interface.clone(),
2688 address: SocketAddress::from(address.address),
2689 proxy: ListenerType::Tcp.into(),
2690 from_scm: false,
2691 })
2692 .into(),
2693 );
2694 }
2695 }
2696
2697 for address in removed_udp_listeners {
2698 if self.udp_listeners[*address].active {
2699 v.push(
2700 RequestType::DeactivateListener(DeactivateListener {
2701 interface: address.interface.clone(),
2702 address: SocketAddress::from(address.address),
2703 proxy: ListenerType::Udp.into(),
2704 to_scm: false,
2705 })
2706 .into(),
2707 );
2708 }
2709
2710 v.push(
2711 RequestType::RemoveListener(RemoveListener {
2712 interface: address.interface.clone(),
2713 address: SocketAddress::from(address.address),
2714 proxy: ListenerType::Udp.into(),
2715 })
2716 .into(),
2717 );
2718 }
2719
2720 for address in added_udp_listeners.clone() {
2721 v.push(RequestType::AddUdpListener(other.udp_listeners[*address].clone()).into());
2722
2723 if other.udp_listeners[*address].active {
2724 v.push(
2725 RequestType::ActivateListener(ActivateListener {
2726 interface: address.interface.clone(),
2727 address: SocketAddress::from(address.address),
2728 proxy: ListenerType::Udp.into(),
2729 from_scm: false,
2730 })
2731 .into(),
2732 );
2733 }
2734 }
2735
2736 for address in removed_http_listeners {
2737 if self.http_listeners[*address].active {
2738 v.push(
2739 RequestType::DeactivateListener(DeactivateListener {
2740 interface: address.interface.clone(),
2741 address: SocketAddress::from(address.address),
2742 proxy: ListenerType::Http.into(),
2743 to_scm: false,
2744 })
2745 .into(),
2746 );
2747 }
2748
2749 v.push(
2750 RequestType::RemoveListener(RemoveListener {
2751 interface: address.interface.clone(),
2752 address: SocketAddress::from(address.address),
2753 proxy: ListenerType::Http.into(),
2754 })
2755 .into(),
2756 );
2757 }
2758
2759 for address in added_http_listeners.clone() {
2760 v.push(RequestType::AddHttpListener(other.http_listeners[*address].clone()).into());
2761
2762 if other.http_listeners[*address].active {
2763 v.push(
2764 RequestType::ActivateListener(ActivateListener {
2765 interface: address.interface.clone(),
2766 address: SocketAddress::from(address.address),
2767 proxy: ListenerType::Http.into(),
2768 from_scm: false,
2769 })
2770 .into(),
2771 );
2772 }
2773 }
2774
2775 for address in removed_https_listeners {
2776 if self.https_listeners[*address].active {
2777 v.push(
2778 RequestType::DeactivateListener(DeactivateListener {
2779 interface: address.interface.clone(),
2780 address: SocketAddress::from(address.address),
2781 proxy: ListenerType::Https.into(),
2782 to_scm: false,
2783 })
2784 .into(),
2785 );
2786 }
2787
2788 v.push(
2789 RequestType::RemoveListener(RemoveListener {
2790 interface: address.interface.clone(),
2791 address: SocketAddress::from(address.address),
2792 proxy: ListenerType::Https.into(),
2793 })
2794 .into(),
2795 );
2796 }
2797
2798 for address in added_https_listeners.clone() {
2799 v.push(RequestType::add_https_listener(other.https_listeners[*address].clone()).into());
2800
2801 if other.https_listeners[*address].active {
2802 v.push(
2803 RequestType::ActivateListener(ActivateListener {
2804 interface: address.interface.clone(),
2805 address: SocketAddress::from(address.address),
2806 proxy: ListenerType::Https.into(),
2807 from_scm: false,
2808 })
2809 .into(),
2810 );
2811 }
2812 }
2813
2814 for addr in my_tcp_listeners.intersection(&their_tcp_listeners) {
2815 let my_listener = &self.tcp_listeners[*addr];
2816 let their_listener = &other.tcp_listeners[*addr];
2817
2818 if my_listener != their_listener {
2819 v.push(
2820 RequestType::RemoveListener(RemoveListener {
2821 interface: addr.interface.clone(),
2822 address: SocketAddress::from(addr.address),
2823 proxy: ListenerType::Tcp.into(),
2824 })
2825 .into(),
2826 );
2827 let mut listener_to_add = their_listener.clone();
2829 listener_to_add.active = false;
2830 v.push(RequestType::AddTcpListener(listener_to_add).into());
2831
2832 if their_listener.active {
2839 v.push(
2840 RequestType::ActivateListener(ActivateListener {
2841 interface: addr.interface.clone(),
2842 address: SocketAddress::from(addr.address),
2843 proxy: ListenerType::Tcp.into(),
2844 from_scm: false,
2845 })
2846 .into(),
2847 );
2848 }
2849 }
2850
2851 if my_listener.active && !their_listener.active {
2852 v.push(
2853 RequestType::DeactivateListener(DeactivateListener {
2854 interface: addr.interface.clone(),
2855 address: SocketAddress::from(addr.address),
2856 proxy: ListenerType::Tcp.into(),
2857 to_scm: false,
2858 })
2859 .into(),
2860 );
2861 }
2862 }
2863
2864 for addr in my_udp_listeners.intersection(&their_udp_listeners) {
2865 let my_listener = &self.udp_listeners[*addr];
2866 let their_listener = &other.udp_listeners[*addr];
2867
2868 if my_listener != their_listener {
2869 v.push(
2870 RequestType::RemoveListener(RemoveListener {
2871 interface: addr.interface.clone(),
2872 address: SocketAddress::from(addr.address),
2873 proxy: ListenerType::Udp.into(),
2874 })
2875 .into(),
2876 );
2877 let mut listener_to_add = their_listener.clone();
2879 listener_to_add.active = false;
2880 v.push(RequestType::AddUdpListener(listener_to_add).into());
2881
2882 if their_listener.active {
2889 v.push(
2890 RequestType::ActivateListener(ActivateListener {
2891 interface: addr.interface.clone(),
2892 address: SocketAddress::from(addr.address),
2893 proxy: ListenerType::Udp.into(),
2894 from_scm: false,
2895 })
2896 .into(),
2897 );
2898 }
2899 }
2900
2901 if my_listener.active && !their_listener.active {
2902 v.push(
2903 RequestType::DeactivateListener(DeactivateListener {
2904 interface: addr.interface.clone(),
2905 address: SocketAddress::from(addr.address),
2906 proxy: ListenerType::Udp.into(),
2907 to_scm: false,
2908 })
2909 .into(),
2910 );
2911 }
2912 }
2913
2914 for addr in my_http_listeners.intersection(&their_http_listeners) {
2915 let my_listener = &self.http_listeners[*addr];
2916 let their_listener = &other.http_listeners[*addr];
2917
2918 if my_listener != their_listener {
2919 v.push(
2920 RequestType::RemoveListener(RemoveListener {
2921 interface: addr.interface.clone(),
2922 address: SocketAddress::from(addr.address),
2923 proxy: ListenerType::Http.into(),
2924 })
2925 .into(),
2926 );
2927 let mut listener_to_add = their_listener.clone();
2929 listener_to_add.active = false;
2930 v.push(RequestType::AddHttpListener(listener_to_add).into());
2931
2932 if their_listener.active {
2939 v.push(
2940 RequestType::ActivateListener(ActivateListener {
2941 interface: addr.interface.clone(),
2942 address: SocketAddress::from(addr.address),
2943 proxy: ListenerType::Http.into(),
2944 from_scm: false,
2945 })
2946 .into(),
2947 );
2948 }
2949 }
2950
2951 if my_listener.active && !their_listener.active {
2952 v.push(
2953 RequestType::DeactivateListener(DeactivateListener {
2954 interface: addr.interface.clone(),
2955 address: SocketAddress::from(addr.address),
2956 proxy: ListenerType::Http.into(),
2957 to_scm: false,
2958 })
2959 .into(),
2960 );
2961 }
2962 }
2963
2964 for addr in my_https_listeners.intersection(&their_https_listeners) {
2965 let my_listener = &self.https_listeners[*addr];
2966 let their_listener = &other.https_listeners[*addr];
2967
2968 if my_listener != their_listener {
2969 v.push(
2970 RequestType::RemoveListener(RemoveListener {
2971 interface: addr.interface.clone(),
2972 address: SocketAddress::from(addr.address),
2973 proxy: ListenerType::Https.into(),
2974 })
2975 .into(),
2976 );
2977 let mut listener_to_add = their_listener.clone();
2979 listener_to_add.active = false;
2980 v.push(RequestType::add_https_listener(listener_to_add).into());
2981
2982 if their_listener.active {
2989 v.push(
2990 RequestType::ActivateListener(ActivateListener {
2991 interface: addr.interface.clone(),
2992 address: SocketAddress::from(addr.address),
2993 proxy: ListenerType::Https.into(),
2994 from_scm: false,
2995 })
2996 .into(),
2997 );
2998 }
2999 }
3000
3001 if my_listener.active && !their_listener.active {
3002 v.push(
3003 RequestType::DeactivateListener(DeactivateListener {
3004 interface: addr.interface.clone(),
3005 address: SocketAddress::from(addr.address),
3006 proxy: ListenerType::Https.into(),
3007 to_scm: false,
3008 })
3009 .into(),
3010 );
3011 }
3012 }
3013
3014 let mut removed_clusters: Vec<&ClusterId> = Vec::new();
3015 for (cluster_id, res) in diff_map(self.clusters.iter(), other.clusters.iter()) {
3016 match res {
3017 DiffResult::Added | DiffResult::Changed => v.push(
3018 RequestType::AddCluster(other.clusters.get(cluster_id).unwrap().clone()).into(),
3019 ),
3020 DiffResult::Removed => {
3021 removed_clusters.push(cluster_id);
3022 v.push(RequestType::RemoveCluster(cluster_id.to_string()).into())
3023 }
3024 }
3025 }
3026
3027 let base: Cow<'_, ConfigState> = if removed_clusters.is_empty() {
3034 Cow::Borrowed(self)
3035 } else {
3036 let mut cascaded = self.clone();
3037 for cluster_id in &removed_clusters {
3038 let removed = cascaded.remove_cluster(cluster_id);
3039 debug_assert!(
3040 removed.is_ok(),
3041 "a cluster diff_map reports removed must exist in self"
3042 );
3043 }
3044 Cow::Owned(cascaded)
3045 };
3046
3047 for ((cluster_id, backend_id), res) in diff_map(
3048 base.backends.iter().flat_map(|(cluster_id, v)| {
3049 v.iter()
3050 .map(move |backend| ((cluster_id, &backend.backend_id), backend))
3051 }),
3052 other.backends.iter().flat_map(|(cluster_id, v)| {
3053 v.iter()
3054 .map(move |backend| ((cluster_id, &backend.backend_id), backend))
3055 }),
3056 ) {
3057 match res {
3058 DiffResult::Added => {
3059 let backend = other
3060 .backends
3061 .get(cluster_id)
3062 .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
3063 .unwrap();
3064 v.push(RequestType::AddBackend(backend.clone().to_add_backend()).into());
3065 }
3066 DiffResult::Removed => {
3067 let backend = base
3068 .backends
3069 .get(cluster_id)
3070 .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
3071 .unwrap();
3072
3073 v.push(
3074 RequestType::RemoveBackend(RemoveBackend {
3075 cluster_id: backend.cluster_id.to_string(),
3076 backend_id: backend.backend_id.clone(),
3077 address: SocketAddress::from(backend.address),
3078 })
3079 .into(),
3080 );
3081 }
3082 DiffResult::Changed => {
3083 let backend = base
3084 .backends
3085 .get(cluster_id)
3086 .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
3087 .unwrap();
3088
3089 v.push(
3090 RequestType::RemoveBackend(RemoveBackend {
3091 cluster_id: backend.cluster_id.to_string(),
3092 backend_id: backend.backend_id.clone(),
3093 address: SocketAddress::from(backend.address),
3094 })
3095 .into(),
3096 );
3097
3098 let backend = other
3099 .backends
3100 .get(cluster_id)
3101 .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
3102 .unwrap();
3103 v.push(RequestType::AddBackend(backend.clone().to_add_backend()).into());
3104 }
3105 }
3106 }
3107
3108 let mut my_http_fronts: HashSet<(&str, &HttpFrontend)> = HashSet::new();
3109 for (route, front) in base.http_fronts.iter() {
3110 my_http_fronts.insert((route, front));
3111 }
3112 let mut their_http_fronts: HashSet<(&str, &HttpFrontend)> = HashSet::new();
3113 for (route, front) in other.http_fronts.iter() {
3114 their_http_fronts.insert((route, front));
3115 }
3116
3117 let removed_http_fronts = my_http_fronts.difference(&their_http_fronts);
3118 let added_http_fronts = their_http_fronts.difference(&my_http_fronts);
3119
3120 for &(_, front) in removed_http_fronts {
3121 v.push(RequestType::RemoveHttpFrontend(front.clone().into()).into());
3122 }
3123
3124 for &(_, front) in added_http_fronts {
3125 v.push(RequestType::AddHttpFrontend(front.clone().into()).into());
3126 }
3127
3128 let mut my_https_fronts: HashSet<(&String, &HttpFrontend)> = HashSet::new();
3129 for (route, front) in base.https_fronts.iter() {
3130 my_https_fronts.insert((route, front));
3131 }
3132 let mut their_https_fronts: HashSet<(&String, &HttpFrontend)> = HashSet::new();
3133 for (route, front) in other.https_fronts.iter() {
3134 their_https_fronts.insert((route, front));
3135 }
3136 let removed_https_fronts = my_https_fronts.difference(&their_https_fronts);
3137 let added_https_fronts = their_https_fronts.difference(&my_https_fronts);
3138
3139 for &(_, front) in removed_https_fronts {
3140 v.push(RequestType::RemoveHttpsFrontend(front.clone().into()).into());
3141 }
3142
3143 for &(_, front) in added_https_fronts {
3144 v.push(RequestType::AddHttpsFrontend(front.clone().into()).into());
3145 }
3146
3147 let mut my_tcp_fronts: HashSet<(&ClusterId, &TcpFrontend)> = HashSet::new();
3148 for (cluster_id, front_list) in base.tcp_fronts.iter() {
3149 for front in front_list.iter() {
3150 my_tcp_fronts.insert((cluster_id, front));
3151 }
3152 }
3153 let mut their_tcp_fronts: HashSet<(&ClusterId, &TcpFrontend)> = HashSet::new();
3154 for (cluster_id, front_list) in other.tcp_fronts.iter() {
3155 for front in front_list.iter() {
3156 their_tcp_fronts.insert((cluster_id, front));
3157 }
3158 }
3159
3160 let removed_tcp_fronts = my_tcp_fronts.difference(&their_tcp_fronts);
3161 let added_tcp_fronts = their_tcp_fronts.difference(&my_tcp_fronts);
3162
3163 for &(_, front) in removed_tcp_fronts {
3164 v.push(RequestType::RemoveTcpFrontend(front.clone().into()).into());
3165 }
3166
3167 for &(_, front) in added_tcp_fronts {
3168 v.push(RequestType::AddTcpFrontend(front.clone().into()).into());
3169 }
3170
3171 let mut my_udp_fronts: HashSet<(&ClusterId, &UdpFrontend)> = HashSet::new();
3172 for (cluster_id, front_list) in base.udp_fronts.iter() {
3173 for front in front_list.iter() {
3174 my_udp_fronts.insert((cluster_id, front));
3175 }
3176 }
3177 let mut their_udp_fronts: HashSet<(&ClusterId, &UdpFrontend)> = HashSet::new();
3178 for (cluster_id, front_list) in other.udp_fronts.iter() {
3179 for front in front_list.iter() {
3180 their_udp_fronts.insert((cluster_id, front));
3181 }
3182 }
3183
3184 let removed_udp_fronts = my_udp_fronts.difference(&their_udp_fronts);
3185 let added_udp_fronts = their_udp_fronts.difference(&my_udp_fronts);
3186
3187 for &(_, front) in removed_udp_fronts {
3188 v.push(RequestType::RemoveUdpFrontend(front.clone().into()).into());
3189 }
3190
3191 for &(_, front) in added_udp_fronts {
3192 v.push(RequestType::AddUdpFrontend(front.clone().into()).into());
3193 }
3194
3195 let my_certificates: HashSet<(SocketAddr, &Fingerprint)> = HashSet::from_iter(
3197 self.certificates
3198 .iter()
3199 .flat_map(|(addr, certs)| repeat(*addr).zip(certs.keys())),
3200 );
3201 let their_certificates: HashSet<(SocketAddr, &Fingerprint)> = HashSet::from_iter(
3202 other
3203 .certificates
3204 .iter()
3205 .flat_map(|(addr, certs)| repeat(*addr).zip(certs.keys())),
3206 );
3207
3208 let removed_certificates = my_certificates.difference(&their_certificates);
3209 let added_certificates = their_certificates.difference(&my_certificates);
3210
3211 for &(address, fingerprint) in removed_certificates {
3212 v.push(
3213 RequestType::RemoveCertificate(RemoveCertificate {
3214 address: SocketAddress::from(address),
3215 fingerprint: fingerprint.to_string(),
3216 })
3217 .into(),
3218 );
3219 }
3220
3221 for &(address, fingerprint) in added_certificates {
3222 if let Some(certificate_and_key) = other
3223 .certificates
3224 .get(&address)
3225 .and_then(|certs| certs.get(fingerprint))
3226 {
3227 v.push(
3228 RequestType::AddCertificate(AddCertificate {
3229 address: SocketAddress::from(address),
3230 certificate: certificate_and_key.clone(),
3231 expired_at: None,
3232 })
3233 .into(),
3234 );
3235 }
3236 }
3237
3238 for address in added_tcp_listeners {
3239 let listener = &other.tcp_listeners[*address];
3240 if listener.active {
3241 v.push(
3242 RequestType::ActivateListener(ActivateListener {
3243 interface: listener.interface.clone(),
3244 address: listener.address,
3245 proxy: ListenerType::Tcp.into(),
3246 from_scm: false,
3247 })
3248 .into(),
3249 );
3250 }
3251 }
3252
3253 for address in added_udp_listeners {
3254 let listener = &other.udp_listeners[*address];
3255 if listener.active {
3256 v.push(
3257 RequestType::ActivateListener(ActivateListener {
3258 interface: listener.interface.clone(),
3259 address: listener.address,
3260 proxy: ListenerType::Udp.into(),
3261 from_scm: false,
3262 })
3263 .into(),
3264 );
3265 }
3266 }
3267
3268 #[cfg(debug_assertions)]
3281 {
3282 let mut replayed = self.clone();
3283 for request in &v {
3284 debug_assert!(
3286 replayed.dispatch(request).is_ok(),
3287 "every request emitted by diff must replay cleanly onto self"
3288 );
3289 }
3290 let nonempty = |m: &BTreeMap<ClusterId, Vec<Backend>>| {
3291 m.iter()
3292 .filter(|(_, v)| !v.is_empty())
3293 .map(|(k, v)| (k.clone(), v.clone()))
3294 .collect::<BTreeMap<_, _>>()
3295 };
3296 let nonempty_tcp = |m: &HashMap<ClusterId, Vec<TcpFrontend>>| {
3297 m.iter()
3298 .filter(|(_, v)| !v.is_empty())
3299 .map(|(k, v)| (k.clone(), v.clone()))
3300 .collect::<HashMap<_, _>>()
3301 };
3302 debug_assert!(
3303 replayed.clusters == other.clusters
3304 && nonempty(&replayed.backends) == nonempty(&other.backends)
3305 && replayed.http_fronts == other.http_fronts
3306 && replayed.https_fronts == other.https_fronts
3307 && nonempty_tcp(&replayed.tcp_fronts) == nonempty_tcp(&other.tcp_fronts)
3308 && replayed.certificates == other.certificates
3309 && replayed.http_listeners == other.http_listeners
3310 && replayed.https_listeners == other.https_listeners
3311 && replayed.tcp_listeners == other.tcp_listeners
3312 && replayed.udp_listeners == other.udp_listeners,
3313 "replaying diff(self, other) onto self must reproduce other's clusters/backends/frontends/certificates/listeners"
3314 );
3315 }
3316
3317 v
3318 }
3319
3320 pub fn hash_state(&self) -> BTreeMap<ClusterId, u64> {
3322 let mut hm: HashMap<ClusterId, DefaultHasher> = self
3323 .clusters
3324 .keys()
3325 .map(|cluster_id| {
3326 let mut hasher = DefaultHasher::new();
3327 self.clusters.get(cluster_id).hash(&mut hasher);
3328 if let Some(backends) = self.backends.get(cluster_id) {
3329 backends.iter().collect::<BTreeSet<_>>().hash(&mut hasher)
3330 }
3331 if let Some(tcp_fronts) = self.tcp_fronts.get(cluster_id) {
3332 tcp_fronts.iter().collect::<BTreeSet<_>>().hash(&mut hasher)
3333 }
3334 (cluster_id.to_owned(), hasher)
3335 })
3336 .collect();
3337
3338 for front in self.http_fronts.values() {
3339 if let Some(cluster_id) = &front.cluster_id
3340 && let Some(hasher) = hm.get_mut(cluster_id)
3341 {
3342 front.hash(hasher);
3343 }
3344 }
3345
3346 for front in self.https_fronts.values() {
3347 if let Some(cluster_id) = &front.cluster_id
3348 && let Some(hasher) = hm.get_mut(cluster_id)
3349 {
3350 front.hash(hasher);
3351 }
3352 }
3353
3354 hm.drain()
3355 .map(|(cluster_id, hasher)| (cluster_id, hasher.finish()))
3356 .collect()
3357 }
3358
3359 pub fn cluster_state(&self, cluster_id: &str) -> Option<ClusterInformation> {
3362 let configuration = self.clusters.get(cluster_id).cloned()?;
3363 info!("{:?}", configuration);
3364
3365 let http_frontends: Vec<RequestHttpFrontend> = self
3366 .http_fronts
3367 .values()
3368 .filter(|front| front.cluster_id.as_deref() == Some(cluster_id))
3369 .map(|front| front.clone().into())
3370 .collect();
3371
3372 let https_frontends: Vec<RequestHttpFrontend> = self
3373 .https_fronts
3374 .values()
3375 .filter(|front| front.cluster_id.as_deref() == Some(cluster_id))
3376 .map(|front| front.clone().into())
3377 .collect();
3378
3379 let tcp_frontends: Vec<RequestTcpFrontend> = self
3380 .tcp_fronts
3381 .get(cluster_id)
3382 .cloned()
3383 .unwrap_or_default()
3384 .iter()
3385 .map(|front| front.clone().into())
3386 .collect();
3387
3388 let udp_frontends: Vec<RequestUdpFrontend> = self
3389 .udp_fronts
3390 .get(cluster_id)
3391 .cloned()
3392 .unwrap_or_default()
3393 .iter()
3394 .map(|front| front.clone().into())
3395 .collect();
3396
3397 let backends: Vec<AddBackend> = self
3398 .backends
3399 .get(cluster_id)
3400 .cloned()
3401 .unwrap_or_default()
3402 .iter()
3403 .map(|backend| backend.clone().into())
3404 .collect();
3405
3406 Some(ClusterInformation {
3407 configuration: Some(configuration),
3408 http_frontends,
3409 https_frontends,
3410 tcp_frontends,
3411 backends,
3412 udp_frontends,
3413 })
3414 }
3415
3416 pub fn count_backends(&self) -> usize {
3417 self.backends.values().fold(0, |acc, v| acc + v.len())
3418 }
3419
3420 pub fn count_frontends(&self) -> usize {
3421 self.http_fronts.values().count()
3422 + self.https_fronts.values().count()
3423 + self.tcp_fronts.values().fold(0, |acc, v| acc + v.len())
3424 + self.udp_fronts.values().fold(0, |acc, v| acc + v.len())
3425 }
3426
3427 pub fn get_cluster_ids_by_domain(
3428 &self,
3429 hostname: String,
3430 path: Option<String>,
3431 ) -> HashSet<ClusterId> {
3432 let mut cluster_ids: HashSet<ClusterId> = HashSet::new();
3433
3434 self.http_fronts.values().for_each(|front| {
3435 if domain_check(&front.hostname, &front.path, &hostname, &path)
3436 && let Some(id) = &front.cluster_id
3437 {
3438 cluster_ids.insert(id.clone());
3439 }
3440 });
3441
3442 self.https_fronts.values().for_each(|front| {
3443 if domain_check(&front.hostname, &front.path, &hostname, &path)
3444 && let Some(id) = &front.cluster_id
3445 {
3446 cluster_ids.insert(id.clone());
3447 }
3448 });
3449
3450 cluster_ids
3451 }
3452
3453 pub fn get_certificates(
3471 &self,
3472 filters: QueryCertificatesFilters,
3473 ) -> BTreeMap<String, CertificateAndKey> {
3474 self.certificates
3475 .values()
3476 .flat_map(|hash_map| hash_map.iter())
3477 .filter(|(fingerprint, cert)| {
3478 if let Some(domain) = &filters.domain {
3479 cert.names.contains(domain)
3480 } else if let Some(f) = &filters.fingerprint {
3481 fingerprint.to_string() == *f
3482 } else {
3483 true
3484 }
3485 })
3486 .map(|(fingerprint, cert)| (fingerprint.to_string(), cert.to_owned()))
3487 .collect()
3488 }
3489
3490 pub fn list_frontends(&self, filters: FrontendFilters) -> ListedFrontends {
3491 let list_all = !filters.http && !filters.https && !filters.tcp;
3493
3494 let matches_cluster = |cluster_id: Option<&str>| match &filters.cluster_id {
3497 Some(wanted) => cluster_id == Some(wanted.as_str()),
3498 None => true,
3499 };
3500 let matches_domain = |hostname: &str| match &filters.domain {
3501 Some(domain) => hostname.contains(domain),
3502 None => true,
3503 };
3504
3505 let mut listed_frontends = ListedFrontends::default();
3506
3507 if filters.http || list_all {
3508 for http_frontend in self.http_fronts.iter().filter(|f| {
3509 matches_domain(&f.1.hostname) && matches_cluster(f.1.cluster_id.as_deref())
3510 }) {
3511 listed_frontends
3512 .http_frontends
3513 .push(http_frontend.1.to_owned().into());
3514 }
3515 }
3516
3517 if filters.https || list_all {
3518 for https_frontend in self.https_fronts.iter().filter(|f| {
3519 matches_domain(&f.1.hostname) && matches_cluster(f.1.cluster_id.as_deref())
3520 }) {
3521 listed_frontends
3522 .https_frontends
3523 .push(https_frontend.1.to_owned().into());
3524 }
3525 }
3526
3527 if (filters.tcp || list_all) && filters.domain.is_none() {
3528 for tcp_frontend in fronts_of_cluster(&self.tcp_fronts, filters.cluster_id.as_deref()) {
3529 listed_frontends
3530 .tcp_frontends
3531 .push(tcp_frontend.to_owned().into())
3532 }
3533 }
3534
3535 if (filters.tcp || list_all) && filters.domain.is_none() {
3541 for udp_frontend in fronts_of_cluster(&self.udp_fronts, filters.cluster_id.as_deref()) {
3542 listed_frontends
3543 .udp_frontends
3544 .push(udp_frontend.to_owned().into())
3545 }
3546 }
3547
3548 listed_frontends
3549 }
3550
3551 pub fn list_listeners(&self) -> ListenersList {
3552 ListenersList {
3553 http_listeners: self
3554 .http_listeners
3555 .iter()
3556 .map(|(addr, listener)| (addr.to_string(), listener.clone()))
3557 .collect(),
3558 https_listeners: self
3559 .https_listeners
3560 .iter()
3561 .map(|(addr, listener)| (addr.to_string(), listener.clone()))
3562 .collect(),
3563 tcp_listeners: self
3564 .tcp_listeners
3565 .iter()
3566 .map(|(addr, listener)| (addr.to_string(), listener.clone()))
3567 .collect(),
3568 udp_listeners: self
3569 .udp_listeners
3570 .iter()
3571 .map(|(addr, listener)| (addr.to_string(), listener.clone()))
3572 .collect(),
3573 }
3574 }
3575
3576 pub fn produce_initial_state(&self) -> InitialState {
3578 let mut worker_requests = Vec::new();
3579 for (counter, request) in self.generate_requests().into_iter().enumerate() {
3580 worker_requests.push(WorkerRequest::new(format!("SAVE-{counter}"), request));
3581 }
3582 InitialState {
3583 requests: worker_requests,
3584 }
3585 }
3586
3587 pub fn write_initial_state_to_file(&self, file: &mut File) -> Result<usize, StateError> {
3590 let initial_state = self.produce_initial_state();
3591 let count = initial_state.requests.len();
3592
3593 let bytes_to_write = initial_state.encode_to_vec();
3594 println!("writing {} in the temp file", bytes_to_write.len());
3595 file.write_all(&bytes_to_write)
3596 .map_err(StateError::FileError)?;
3597
3598 file.sync_all().map_err(StateError::FileError)?;
3599
3600 Ok(count)
3601 }
3602
3603 pub fn write_requests_to_file(&self, file: &mut File) -> Result<usize, StateError> {
3607 let mut counter = 0usize;
3608 let requests = self.generate_requests();
3609
3610 for request in requests {
3611 let message = WorkerRequest::new(format!("SAVE-{counter}"), request);
3612
3613 file.write_all(
3614 &serde_json::to_string(&message)
3615 .map(|s| s.into_bytes())
3616 .unwrap_or_default(),
3617 )
3618 .map_err(StateError::FileError)?;
3619
3620 file.write_all(&b"\n\0"[..])
3621 .map_err(StateError::FileError)?;
3622
3623 if counter.is_multiple_of(1000) {
3624 info!("writing {} commands to file", counter);
3625 file.sync_all().map_err(StateError::FileError)?;
3626 }
3627 counter += 1;
3628 }
3629 file.sync_all().map_err(StateError::FileError)?;
3630
3631 Ok(counter)
3632 }
3633}
3634
3635macro_rules! require_h2_knob_floor {
3643 ($config:ident, $field:ident, $minimum:literal) => {
3644 if let Some(value) = $config.$field
3645 && u64::from(value) < $minimum
3646 {
3647 return Err(StateError::InvalidValue {
3648 field: stringify!($field),
3649 reason: concat!("must be >= ", stringify!($minimum)),
3650 });
3651 }
3652 };
3653}
3654
3655pub fn validate_h2_flood_knobs_http(patch: &UpdateHttpListenerConfig) -> Result<(), StateError> {
3673 crate::for_each_h2_knob_floor!(require_h2_knob_floor, patch);
3674 Ok(())
3675}
3676
3677pub fn validate_h2_flood_knobs_https(patch: &UpdateHttpsListenerConfig) -> Result<(), StateError> {
3679 crate::for_each_h2_knob_floor!(require_h2_knob_floor, patch);
3680 Ok(())
3681}
3682
3683pub fn validate_h2_flood_knobs_http_listener(
3707 config: &HttpListenerConfig,
3708) -> Result<(), StateError> {
3709 crate::for_each_h2_knob_floor!(require_h2_knob_floor, config);
3710 Ok(())
3711}
3712
3713pub fn validate_h2_flood_knobs_https_listener(
3717 config: &HttpsListenerConfig,
3718) -> Result<(), StateError> {
3719 crate::for_each_h2_knob_floor!(require_h2_knob_floor, config);
3720 Ok(())
3721}
3722
3723fn merge_answers(stored: &mut BTreeMap<String, String>, patch: &BTreeMap<String, String>) {
3733 for (code, body) in patch {
3734 if !body.is_empty() {
3735 stored.insert(code.to_owned(), body.to_owned());
3736 }
3737 }
3738}
3739
3740pub fn merge_custom_http_answers(
3748 target: &mut Option<CustomHttpAnswers>,
3749 patch: &CustomHttpAnswers,
3750) {
3751 let current = target.get_or_insert_with(CustomHttpAnswers::default);
3752 macro_rules! merge_field {
3753 ($field:ident) => {
3754 if let Some(ref v) = patch.$field {
3755 current.$field = Some(v.clone());
3756 }
3757 };
3758 }
3759 merge_field!(answer_301);
3760 merge_field!(answer_400);
3761 merge_field!(answer_401);
3762 merge_field!(answer_404);
3763 merge_field!(answer_408);
3764 merge_field!(answer_413);
3765 merge_field!(answer_421);
3766 merge_field!(answer_502);
3767 merge_field!(answer_503);
3768 merge_field!(answer_504);
3769 merge_field!(answer_507);
3770}
3771
3772pub fn validate_alpn_protocols(values: &[String]) -> Result<(), StateError> {
3775 for value in values {
3776 if value != "h2" && value != "http/1.1" {
3777 return Err(StateError::InvalidValue {
3778 field: "alpn_protocols",
3779 reason: "each value must be \"h2\" or \"http/1.1\"",
3780 });
3781 }
3782 }
3783 Ok(())
3784}
3785
3786pub fn validate_sozu_id_header(value: &str) -> Result<(), StateError> {
3797 if value.is_empty() {
3798 return Err(StateError::InvalidValue {
3799 field: "sozu_id_header",
3800 reason: "must not be empty",
3801 });
3802 }
3803 for b in value.bytes() {
3804 let is_tchar = b.is_ascii_alphanumeric()
3805 || matches!(
3806 b,
3807 b'!' | b'#'
3808 | b'$'
3809 | b'%'
3810 | b'&'
3811 | b'\''
3812 | b'*'
3813 | b'+'
3814 | b'-'
3815 | b'.'
3816 | b'^'
3817 | b'_'
3818 | b'`'
3819 | b'|'
3820 | b'~'
3821 );
3822 if !is_tchar {
3823 return Err(StateError::InvalidValue {
3824 field: "sozu_id_header",
3825 reason: "must be a valid HTTP header name (RFC 9110 §5.1 token: alphanumeric or one of !#$%&'*+-.^_`|~)",
3826 });
3827 }
3828 }
3829 Ok(())
3830}
3831
3832pub fn validate_hsts_patch(hsts: &HstsConfig) -> Result<(), StateError> {
3837 if hsts.enabled.is_none() {
3838 return Err(StateError::InvalidValue {
3839 field: "hsts",
3840 reason: "`enabled` is required whenever the hsts block is present",
3841 });
3842 }
3843 Ok(())
3844}
3845
3846pub fn validate_forwarded_headers(value: i32) -> Result<(), StateError> {
3851 ForwardedHeaders::try_from(value)
3852 .map(|_| ())
3853 .map_err(|_| StateError::InvalidValue {
3854 field: "forwarded_headers",
3855 reason: "must be one of both, x_forwarded, rfc7239, none",
3856 })
3857}
3858
3859fn fronts_of_cluster<'a, F>(
3863 fronts: &'a HashMap<ClusterId, Vec<F>>,
3864 cluster_id: Option<&str>,
3865) -> impl Iterator<Item = &'a F> {
3866 let (one, all) = match cluster_id {
3867 Some(cluster_id) => (fronts.get(cluster_id), None),
3868 None => (None, Some(fronts.values())),
3869 };
3870 one.into_iter().chain(all.into_iter().flatten()).flatten()
3871}
3872
3873fn domain_check(
3874 front_hostname: &str,
3875 front_path_rule: &PathRule,
3876 hostname: &str,
3877 path_prefix: &Option<String>,
3878) -> bool {
3879 if hostname != front_hostname {
3880 return false;
3881 }
3882
3883 if let Some(path) = &path_prefix {
3884 return path == &front_path_rule.value;
3885 }
3886
3887 true
3888}
3889
3890fn canonical_tcp_alpn(alpn: &[String]) -> Vec<String> {
3900 let mut alpn = alpn.to_vec();
3901 alpn.sort();
3902 alpn.dedup();
3903 debug_assert!(
3907 alpn.windows(2).all(|pair| pair[0] < pair[1]),
3908 "canonical_tcp_alpn must return a strictly sorted, duplicate-free list"
3909 );
3910 alpn
3911}
3912
3913fn tcp_frontend_matches(
3924 front: &TcpFrontend,
3925 address: SocketAddr,
3926 sni: &Option<String>,
3927 alpn: &[String],
3928) -> bool {
3929 front.address == address && front.sni == *sni && front.alpn.as_slice() == alpn
3930}
3931
3932struct DiffMap<'a, K: Ord, V, I1, I2> {
3933 my_it: I1,
3934 other_it: I2,
3935 my: Option<(K, &'a V)>,
3936 other: Option<(K, &'a V)>,
3937}
3938
3939fn diff_map<
3941 'a,
3942 K: Ord,
3943 V: PartialEq,
3944 I1: Iterator<Item = (K, &'a V)>,
3945 I2: Iterator<Item = (K, &'a V)>,
3946>(
3947 my: I1,
3948 other: I2,
3949) -> DiffMap<'a, K, V, I1, I2> {
3950 DiffMap {
3951 my_it: my,
3952 other_it: other,
3953 my: None,
3954 other: None,
3955 }
3956}
3957
3958enum DiffResult {
3959 Added,
3960 Removed,
3961 Changed,
3962}
3963
3964impl<'a, K: Ord, V: PartialEq, I1: Iterator<Item = (K, &'a V)>, I2: Iterator<Item = (K, &'a V)>>
3967 std::iter::Iterator for DiffMap<'a, K, V, I1, I2>
3968{
3969 type Item = (K, DiffResult);
3970
3971 fn next(&mut self) -> Option<Self::Item> {
3972 loop {
3973 if self.my.is_none() {
3974 self.my = self.my_it.next();
3975 }
3976 if self.other.is_none() {
3977 self.other = self.other_it.next();
3978 }
3979
3980 match (self.my.take(), self.other.take()) {
3981 (None, other) => return other.map(|(k, _)| (k, DiffResult::Added)),
3985 (Some((k, _)), None) => return Some((k, DiffResult::Removed)),
3988 (Some((k1, _v1)), Some((k2, v2))) if k1 < k2 => {
3990 self.other = Some((k2, v2));
3991 return Some((k1, DiffResult::Removed));
3992 }
3993 (Some((k1, v1)), Some((k2, _v2))) if k1 > k2 => {
3995 self.my = Some((k1, v1));
3996 return Some((k2, DiffResult::Added));
3997 }
3998 (Some((k1, v1)), Some((_k2, v2))) if v1 != v2 => {
3999 return Some((k1, DiffResult::Changed));
4002 }
4003 _ => {}
4004 }
4005 }
4006 }
4007}
4008
4009#[cfg(test)]
4010mod tests {
4011 use rand::{RngExt, rng, seq::SliceRandom};
4012
4013 use super::*;
4014 use crate::proto::command::{
4015 ClientAuthMode, CustomHttpAnswers, Header, HeaderPosition, HstsConfig, LoadBalancingParams,
4016 PathRuleKind, RedirectPolicy, RedirectScheme, RequestHttpFrontend, RequestTcpFrontend,
4017 RequestUdpFrontend, RulePosition, UdpListenerConfig, UpdateUdpListenerConfig,
4018 };
4019
4020 fn capture_test_logs(run: impl FnOnce() + Send + 'static) -> String {
4021 let receiver = std::net::UdpSocket::bind("127.0.0.1:0")
4022 .expect("test log receiver must bind to a loopback port");
4023 let target = format!(
4024 "udp://{}",
4025 receiver
4026 .local_addr()
4027 .expect("test log receiver must have a local address")
4028 );
4029
4030 std::thread::spawn(move || {
4031 crate::logging::Logger::init(
4032 "state-log-redaction-test".to_owned(),
4033 "info",
4034 &target,
4035 false,
4036 None,
4037 None,
4038 None,
4039 )
4040 .expect("test logger must initialize");
4041 run();
4042 })
4043 .join()
4044 .expect("log-producing test thread must not panic");
4045
4046 receiver
4047 .set_nonblocking(true)
4048 .expect("test log receiver must become nonblocking");
4049 let mut output = String::new();
4050 let mut datagram = vec![0; 65_507];
4051 loop {
4052 match receiver.recv(&mut datagram) {
4053 Ok(length) => output.push_str(&String::from_utf8_lossy(&datagram[..length])),
4054 Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => break,
4055 Err(error) => panic!("test log receiver failed: {error}"),
4056 }
4057 }
4058 assert!(!output.is_empty(), "test log capture received no datagrams");
4059 output
4060 }
4061
4062 #[test]
4063 fn duplicate_certificate_log_redacts_names() {
4064 const NAME_SECRET: &str = "DUPLICATE_CERTIFICATE_NAME_SECRET_SENTINEL";
4065
4066 let name = format!("{NAME_SECRET}{}", "x".repeat(4096));
4067 let name_len = name.len();
4068 let certificate = CertificateAndKey {
4069 certificate: include_str!("../assets/certificate.pem").to_owned(),
4070 key: include_str!("../assets/key.pem").to_owned(),
4071 certificate_chain: Vec::new(),
4072 versions: Vec::new(),
4073 names: vec![name],
4074 };
4075 let fingerprint = certificate
4076 .fingerprint()
4077 .expect("test certificate fingerprint must be computable")
4078 .to_string();
4079 let output = capture_test_logs(move || {
4080 let mut state = ConfigState::default();
4081 let add = AddCertificate {
4082 address: SocketAddress::new_v4(127, 0, 0, 1, 8443),
4083 certificate,
4084 expired_at: None,
4085 };
4086
4087 state
4088 .add_certificate(&add)
4089 .expect("first certificate insertion must succeed");
4090 state
4091 .add_certificate(&add)
4092 .expect("duplicate certificate insertion must be skipped");
4093 });
4094
4095 assert!(
4096 !output.contains(NAME_SECRET),
4097 "duplicate certificate log leaked certificate name {NAME_SECRET}"
4098 );
4099 assert!(
4100 !output.contains(&fingerprint),
4101 "duplicate certificate log leaked certificate fingerprint {fingerprint}"
4102 );
4103 for metadata in [
4104 "fingerprint_bytes=32".to_owned(),
4105 "names_count=1".to_owned(),
4106 format!("names_bytes={name_len}"),
4107 ] {
4108 assert!(
4109 output.contains(&metadata),
4110 "duplicate certificate log omitted bounded metadata {metadata}: {output}"
4111 );
4112 }
4113 assert!(
4114 output.len() <= 512,
4115 "duplicate certificate log is not bounded: {} bytes",
4116 output.len()
4117 );
4118 }
4119
4120 #[test]
4131 fn replace_certificate_with_malformed_pem_keeps_old_certificate() {
4132 let certificate = CertificateAndKey {
4133 certificate: include_str!("../assets/certificate.pem").to_owned(),
4134 key: include_str!("../assets/key.pem").to_owned(),
4135 certificate_chain: Vec::new(),
4136 versions: Vec::new(),
4137 names: Vec::new(),
4138 };
4139 let old_fingerprint = certificate
4140 .fingerprint()
4141 .expect("test certificate fingerprint must be computable");
4142 let address = SocketAddress::new_v4(127, 0, 0, 1, 8443);
4143
4144 let mut state = ConfigState::default();
4145 state
4146 .add_certificate(&AddCertificate {
4147 address,
4148 certificate,
4149 expired_at: None,
4150 })
4151 .expect("initial certificate insertion must succeed");
4152
4153 let malformed = CertificateAndKey {
4154 certificate: "not a valid pem certificate".to_owned(),
4155 key: include_str!("../assets/key.pem").to_owned(),
4156 certificate_chain: Vec::new(),
4157 versions: Vec::new(),
4158 names: Vec::new(),
4159 };
4160 let error = state
4161 .replace_certificate(&ReplaceCertificate {
4162 address,
4163 new_certificate: malformed,
4164 old_fingerprint: old_fingerprint.to_string(),
4165 new_expired_at: None,
4166 })
4167 .expect_err("replacing with a malformed PEM must fail");
4168 assert!(
4169 matches!(error, StateError::ReplaceCertificate(_)),
4170 "expected StateError::ReplaceCertificate, got {error:?}"
4171 );
4172
4173 let certificates_for_address = state
4174 .certificates
4175 .get(&SocketAddr::from(address))
4176 .expect("the address must still be present in the certificate map");
4177 assert!(
4178 certificates_for_address.contains_key(&old_fingerprint),
4179 "a failed replace must leave the original certificate in place, found {:?}",
4180 certificates_for_address.keys().collect::<Vec<_>>()
4181 );
4182 assert_eq!(
4183 certificates_for_address.len(),
4184 1,
4185 "a failed replace must not leave the address with zero (or extra) certificates"
4186 );
4187 }
4188
4189 #[test]
4196 fn replace_certificate_with_same_fingerprint_keeps_certificate() {
4197 let certificate = CertificateAndKey {
4198 certificate: include_str!("../assets/certificate.pem").to_owned(),
4199 key: include_str!("../assets/key.pem").to_owned(),
4200 certificate_chain: Vec::new(),
4201 versions: Vec::new(),
4202 names: Vec::new(),
4203 };
4204 let fingerprint = certificate
4205 .fingerprint()
4206 .expect("test certificate fingerprint must be computable");
4207 let address = SocketAddress::new_v4(127, 0, 0, 1, 8443);
4208
4209 let mut state = ConfigState::default();
4210 state
4211 .add_certificate(&AddCertificate {
4212 address,
4213 certificate: certificate.clone(),
4214 expired_at: None,
4215 })
4216 .expect("initial certificate insertion must succeed");
4217
4218 state
4219 .replace_certificate(&ReplaceCertificate {
4220 address,
4221 new_certificate: certificate,
4222 old_fingerprint: fingerprint.to_string(),
4223 new_expired_at: None,
4224 })
4225 .expect("idempotent replace with the identical certificate must succeed");
4226
4227 let certificates_for_address = state
4228 .certificates
4229 .get(&SocketAddr::from(address))
4230 .expect("the address must still be present in the certificate map");
4231 assert!(
4232 certificates_for_address.contains_key(&fingerprint),
4233 "an idempotent replace (old fingerprint == new fingerprint) must not delete \
4234 the entry the caller meant to retain, found {:?}",
4235 certificates_for_address.keys().collect::<Vec<_>>()
4236 );
4237 assert_eq!(
4238 certificates_for_address.len(),
4239 1,
4240 "an idempotent replace must not change the number of stored certificates"
4241 );
4242 }
4243
4244 #[test]
4259 fn remove_certificate_evicts_the_address_when_its_last_certificate_is_removed() {
4260 let certificate = CertificateAndKey {
4261 certificate: include_str!("../assets/certificate.pem").to_owned(),
4262 key: include_str!("../assets/key.pem").to_owned(),
4263 certificate_chain: Vec::new(),
4264 versions: Vec::new(),
4265 names: Vec::new(),
4266 };
4267 let fingerprint = certificate
4268 .fingerprint()
4269 .expect("test certificate fingerprint must be computable");
4270 let address = SocketAddress::new_v4(127, 0, 0, 1, 8443);
4271
4272 let mut state = ConfigState::default();
4273 state
4274 .add_certificate(&AddCertificate {
4275 address,
4276 certificate,
4277 expired_at: None,
4278 })
4279 .expect("initial certificate insertion must succeed");
4280
4281 state
4282 .remove_certificate(&RemoveCertificate {
4283 address,
4284 fingerprint: fingerprint.to_string(),
4285 })
4286 .expect("removing the only certificate must succeed");
4287
4288 assert!(
4289 !state.certificates.contains_key(&SocketAddr::from(address)),
4290 "removing an address's last certificate must evict the address key too, \
4291 found {:?}",
4292 state.certificates.get(&SocketAddr::from(address))
4293 );
4294
4295 let _ = state.produce_initial_state();
4299 }
4300
4301 #[test]
4313 fn add_certificate_with_valid_pem_armor_but_invalid_x509_leaves_no_bucket() {
4314 let junk_pem = "-----BEGIN CERTIFICATE-----\n\
4317 dGhpcyBpcyBub3QgYSB2YWxpZCB4NTA5IGRlciBzdHJ1Y3R1cmUgYXQgYWxsLCBq\n\
4318 dXN0IGZpbGxlciBieXRlcyBwYWRkZWQgb3V0\n\
4319 -----END CERTIFICATE-----\n";
4320 let certificate = CertificateAndKey {
4321 certificate: junk_pem.to_owned(),
4322 key: include_str!("../assets/key.pem").to_owned(),
4323 certificate_chain: Vec::new(),
4324 versions: Vec::new(),
4325 names: Vec::new(),
4326 };
4327 let address = SocketAddress::new_v4(127, 0, 0, 1, 8443);
4328
4329 let mut state = ConfigState::default();
4330 assert!(
4331 !state.certificates.contains_key(&SocketAddr::from(address)),
4332 "the address must be absent before the add attempt"
4333 );
4334
4335 let error = state
4336 .add_certificate(&AddCertificate {
4337 address,
4338 certificate,
4339 expired_at: None,
4340 })
4341 .expect_err("valid PEM armor around invalid X.509 content must be rejected");
4342 assert!(
4343 matches!(error, StateError::AddCertificate(_)),
4344 "expected StateError::AddCertificate, got {error:?}"
4345 );
4346
4347 assert!(
4348 !state.certificates.contains_key(&SocketAddr::from(address)),
4349 "a rejected add_certificate must leave no bucket for a previously-absent address, \
4350 found {:?}",
4351 state.certificates.get(&SocketAddr::from(address))
4352 );
4353 }
4354
4355 #[test]
4356 fn frontend_state_error_redacts_display_key() {
4357 const HOSTNAME_SECRET: &str = "STATE_ERROR_HOSTNAME_SECRET_SENTINEL";
4358 const PATH_SECRET: &str = "STATE_ERROR_PATH_SECRET_SENTINEL";
4359 const METHOD_SECRET: &str = "STATE_ERROR_METHOD_SECRET_SENTINEL";
4360
4361 let long_value = |marker: &str| format!("{marker}{}", "x".repeat(4096));
4362 let front = RequestHttpFrontend {
4363 cluster_id: Some("cluster".to_owned()),
4364 address: SocketAddress::new_v4(127, 0, 0, 1, 8080),
4365 hostname: long_value(HOSTNAME_SECRET),
4366 path: PathRule {
4367 kind: PathRuleKind::Prefix as i32,
4368 value: long_value(PATH_SECRET),
4369 },
4370 method: Some(long_value(METHOD_SECRET)),
4371 position: RulePosition::Tree as i32,
4372 tags: BTreeMap::new(),
4373 redirect: None,
4374 redirect_scheme: None,
4375 redirect_template: None,
4376 rewrite_host: None,
4377 rewrite_path: None,
4378 rewrite_port: None,
4379 required_auth: None,
4380 headers: Vec::new(),
4381 hsts: None,
4382 };
4383 let mut state = ConfigState::default();
4384 state
4385 .add_http_frontend(&front)
4386 .expect("first frontend insertion must succeed");
4387 assert!(
4388 state
4389 .http_fronts
4390 .keys()
4391 .next()
4392 .expect("raw state key must be retained")
4393 .contains(HOSTNAME_SECRET),
4394 "state storage must retain the raw operator-facing frontend key"
4395 );
4396
4397 let error = state
4398 .add_http_frontend(&front)
4399 .expect_err("duplicate frontend insertion must return StateError::Exists");
4400 let raw_frontend_key = front.to_string();
4401 match &error {
4402 StateError::Exists { id, .. } => assert_eq!(
4403 id, &raw_frontend_key,
4404 "StateError must preserve the public raw frontend identifier"
4405 ),
4406 other => panic!("expected StateError::Exists, got {other:?}"),
4407 }
4408 for (label, output) in [
4409 ("Display", error.to_string()),
4410 ("Debug", format!("{error:?}")),
4411 ] {
4412 for secret in [HOSTNAME_SECRET, PATH_SECRET, METHOD_SECRET] {
4413 assert!(
4414 !output.contains(secret),
4415 "StateError {label} leaked frontend marker {secret}"
4416 );
4417 }
4418 let metadata = format!("id_bytes={}", raw_frontend_key.len());
4419 assert!(
4420 output.contains(&metadata),
4421 "StateError {label} omitted bounded metadata {metadata}: {output}"
4422 );
4423 assert!(
4424 output.len() <= 512,
4425 "StateError {label} output is not bounded: {} bytes",
4426 output.len()
4427 );
4428 }
4429 }
4430
4431 #[test]
4432 fn state_error_formatting_bounds_every_string_bearing_variant() {
4433 const SECRET: &str = "STATE_ERROR_VARIANT_SECRET_SENTINEL";
4434
4435 let long_value = || format!("{SECRET}{}", "x".repeat(4096));
4436 let errors = [
4437 StateError::NotFound {
4438 kind: ObjectKind::Backend,
4439 id: long_value(),
4440 },
4441 StateError::Exists {
4442 kind: ObjectKind::Cluster,
4443 id: long_value(),
4444 },
4445 StateError::AddCertificate(CertificateError::ParsePEMCertificate(long_value())),
4446 StateError::RemoveCertificate(long_value()),
4447 StateError::ReplaceCertificate(long_value()),
4448 StateError::FrontendConversion {
4449 frontend: long_value(),
4450 error: long_value(),
4451 },
4452 StateError::FileError(std::io::Error::other(long_value())),
4453 StateError::InvalidTcpFrontend {
4454 address: "127.0.0.1:443"
4455 .parse()
4456 .expect("test TCP frontend address must parse"),
4457 reason: long_value(),
4458 },
4459 StateError::UdpFrontendAddressTaken {
4460 address: "127.0.0.1:5353"
4461 .parse()
4462 .expect("test UDP frontend address must parse"),
4463 cluster_id: long_value().into(),
4464 current_cluster_id: long_value().into(),
4465 current_tags: BTreeMap::from([(long_value(), long_value())]),
4466 },
4467 ];
4468
4469 for error in errors {
4470 for (label, output) in [
4471 ("Display", error.to_string()),
4472 ("Debug", format!("{error:?}")),
4473 ] {
4474 assert!(
4475 !output.contains(SECRET),
4476 "StateError {label} leaked a string-bearing variant: {output}"
4477 );
4478 assert!(
4479 output.len() <= 512,
4480 "StateError {label} is not bounded: {} bytes",
4481 output.len()
4482 );
4483 }
4484 }
4485 }
4486
4487 #[test]
4488 fn tcp_frontend_not_found_error_retains_the_raw_identity() {
4489 const TCP_SECRET: &str = "STATE_ERROR_TCP_IDENTITY_SECRET_SENTINEL";
4490
4491 let long_value = |suffix: &str| format!("{TCP_SECRET}_{suffix}{}", "x".repeat(1024));
4492 let frontend = RequestTcpFrontend {
4493 cluster_id: long_value("cluster"),
4494 address: SocketAddress::new_v4(127, 0, 0, 1, 443),
4495 tags: BTreeMap::from([(long_value("tag-key"), long_value("tag-value"))]),
4496 sni: Some(long_value("sni")),
4497 alpn: vec![long_value("alpn")],
4498 };
4499 let error = ConfigState::default()
4500 .remove_tcp_frontend(&frontend)
4501 .expect_err("removing from a missing cluster must return StateError::NotFound");
4502
4503 match &error {
4504 StateError::NotFound { id, .. } => assert!(
4505 id.contains(TCP_SECRET),
4506 "StateError must retain the raw TCP frontend identity: {id}"
4507 ),
4508 other => panic!("expected StateError::NotFound, got {other:?}"),
4509 }
4510 for output in [error.to_string(), format!("{error:?}")] {
4511 assert!(
4512 !output.contains(TCP_SECRET),
4513 "StateError formatting leaked the retained TCP identity: {output}"
4514 );
4515 assert!(
4516 output.len() <= 512,
4517 "StateError formatting is not bounded: {} bytes",
4518 output.len()
4519 );
4520 }
4521 }
4522
4523 #[test]
4524 fn retained_state_debug_redacts_http_frontend_and_collection_data() {
4525 const CLUSTER_SECRET: &str = "RETAINED_CLUSTER_SECRET_SENTINEL";
4526 const HOSTNAME_SECRET: &str = "RETAINED_HOSTNAME_SECRET_SENTINEL";
4527 const PATH_SECRET: &str = "RETAINED_PATH_SECRET_SENTINEL";
4528 const METHOD_SECRET: &str = "RETAINED_METHOD_SECRET_SENTINEL";
4529 const TAG_KEY_SECRET: &str = "RETAINED_TAG_KEY_SECRET_SENTINEL";
4530 const TAG_VALUE_SECRET: &str = "RETAINED_TAG_VALUE_SECRET_SENTINEL";
4531 const REDIRECT_TEMPLATE_SECRET: &str = "RETAINED_REDIRECT_TEMPLATE_SECRET_SENTINEL";
4532 const REWRITE_HOST_SECRET: &str = "RETAINED_REWRITE_HOST_SECRET_SENTINEL";
4533 const REWRITE_PATH_SECRET: &str = "RETAINED_REWRITE_PATH_SECRET_SENTINEL";
4534 const HEADER_KEY_SECRET: &str = "RETAINED_HEADER_KEY_SECRET_SENTINEL";
4535 const HEADER_VALUE_SECRET: &str = "RETAINED_HEADER_VALUE_SECRET_SENTINEL";
4536 const REQUEST_COUNT_SECRET: &str = "RETAINED_REQUEST_COUNT_SECRET_SENTINEL";
4537
4538 let long_value = |marker: &str| format!("{marker}{}", "x".repeat(4096));
4539 let request = RequestHttpFrontend {
4540 cluster_id: Some(long_value(CLUSTER_SECRET)),
4541 address: SocketAddress::new_v4(127, 0, 0, 1, 8443),
4542 hostname: long_value(HOSTNAME_SECRET),
4543 path: PathRule {
4544 kind: PathRuleKind::Regex as i32,
4545 value: long_value(PATH_SECRET),
4546 },
4547 method: Some(long_value(METHOD_SECRET)),
4548 position: RulePosition::Tree as i32,
4549 tags: BTreeMap::from([(long_value(TAG_KEY_SECRET), long_value(TAG_VALUE_SECRET))]),
4550 redirect: Some(RedirectPolicy::PermanentRedirect as i32),
4551 redirect_scheme: Some(RedirectScheme::UseHttps as i32),
4552 redirect_template: Some(long_value(REDIRECT_TEMPLATE_SECRET)),
4553 rewrite_host: Some(long_value(REWRITE_HOST_SECRET)),
4554 rewrite_path: Some(long_value(REWRITE_PATH_SECRET)),
4555 rewrite_port: Some(9443),
4556 required_auth: Some(true),
4557 headers: vec![Header {
4558 position: HeaderPosition::Both as i32,
4559 key: long_value(HEADER_KEY_SECRET),
4560 val: long_value(HEADER_VALUE_SECRET),
4561 }],
4562 hsts: Some(HstsConfig {
4563 enabled: Some(true),
4564 max_age: Some(31_536_000),
4565 include_subdomains: Some(true),
4566 preload: Some(false),
4567 force_replace_backend: Some(true),
4568 }),
4569 };
4570 let retained = request
4571 .clone()
4572 .to_frontend()
4573 .expect("adversarial frontend must convert");
4574 let retained_debug = format!("{retained:?}");
4575
4576 let mut state = ConfigState::default();
4577 state
4578 .dispatch(&RequestType::AddHttpFrontend(request).into())
4579 .expect("adversarial frontend must be retained");
4580 state
4581 .request_counts
4582 .insert(long_value(REQUEST_COUNT_SECRET), 7);
4583 let retained_key = state
4584 .http_fronts
4585 .keys()
4586 .next()
4587 .expect("dispatch must retain the frontend's Display key");
4588 assert!(retained_key.contains(HOSTNAME_SECRET));
4589 assert!(retained_key.contains(PATH_SECRET));
4590 assert!(retained_key.contains(METHOD_SECRET));
4591 let state_debug = format!("{state:?}");
4592
4593 let secrets = [
4594 CLUSTER_SECRET,
4595 HOSTNAME_SECRET,
4596 PATH_SECRET,
4597 METHOD_SECRET,
4598 TAG_KEY_SECRET,
4599 TAG_VALUE_SECRET,
4600 REDIRECT_TEMPLATE_SECRET,
4601 REWRITE_HOST_SECRET,
4602 REWRITE_PATH_SECRET,
4603 HEADER_KEY_SECRET,
4604 HEADER_VALUE_SECRET,
4605 REQUEST_COUNT_SECRET,
4606 ];
4607 for (label, output) in [
4608 ("HttpFrontend", &retained_debug),
4609 ("ConfigState", &state_debug),
4610 ] {
4611 for secret in secrets {
4612 assert!(
4613 !output.contains(secret),
4614 "{label} Debug leaked retained marker {secret}"
4615 );
4616 }
4617 assert!(
4618 output.len() <= 2048,
4619 "{label} Debug output is not bounded: {} bytes",
4620 output.len()
4621 );
4622 }
4623
4624 for metadata in [
4625 format!("cluster_id_len: Some({})", long_value(CLUSTER_SECRET).len()),
4626 "address: 127.0.0.1:8443".to_owned(),
4627 format!("hostname_len: {}", long_value(HOSTNAME_SECRET).len()),
4628 "path_kind: 1".to_owned(),
4629 format!("path_len: {}", long_value(PATH_SECRET).len()),
4630 format!("method_len: Some({})", long_value(METHOD_SECRET).len()),
4631 "position: Tree".to_owned(),
4632 "tags_count: Some(1)".to_owned(),
4633 format!(
4634 "tags_len: Some({})",
4635 long_value(TAG_KEY_SECRET).len() + long_value(TAG_VALUE_SECRET).len()
4636 ),
4637 "redirect: Some(4)".to_owned(),
4638 "redirect_scheme: Some(2)".to_owned(),
4639 format!(
4640 "redirect_template_len: Some({})",
4641 long_value(REDIRECT_TEMPLATE_SECRET).len()
4642 ),
4643 format!(
4644 "rewrite_host_len: Some({})",
4645 long_value(REWRITE_HOST_SECRET).len()
4646 ),
4647 format!(
4648 "rewrite_path_len: Some({})",
4649 long_value(REWRITE_PATH_SECRET).len()
4650 ),
4651 "rewrite_port: Some(9443)".to_owned(),
4652 "required_auth: Some(true)".to_owned(),
4653 "headers_count: 1".to_owned(),
4654 format!(
4655 "headers_len: {}",
4656 long_value(HEADER_KEY_SECRET).len() + long_value(HEADER_VALUE_SECRET).len()
4657 ),
4658 "hsts: Some(HstsConfig".to_owned(),
4659 ] {
4660 assert!(
4661 retained_debug.contains(&metadata),
4662 "HttpFrontend Debug omitted safe metadata {metadata}: {retained_debug}"
4663 );
4664 }
4665 for metadata in [
4666 "http_frontends_count: 1",
4667 "backends_count: 0",
4668 "tcp_frontends_count: 0",
4669 "udp_frontends_count: 0",
4670 "certificates_count: 0",
4671 "request_counts_count: 2",
4672 ] {
4673 assert!(
4674 state_debug.contains(metadata),
4675 "ConfigState Debug omitted count metadata {metadata}: {state_debug}"
4676 );
4677 }
4678 }
4679
4680 #[test]
4681 fn serialize() {
4682 let mut state: ConfigState = Default::default();
4683 state
4684 .dispatch(
4685 &RequestType::AddHttpFrontend(RequestHttpFrontend {
4686 cluster_id: Some(String::from("cluster_1")),
4687 hostname: String::from("lolcatho.st:8080"),
4688 path: PathRule::prefix(String::from("/")),
4689 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4690 position: RulePosition::Tree.into(),
4691 ..Default::default()
4692 })
4693 .into(),
4694 )
4695 .expect("Could not execute request");
4696 state
4697 .dispatch(
4698 &RequestType::AddHttpFrontend(RequestHttpFrontend {
4699 cluster_id: Some(String::from("cluster_2")),
4700 hostname: String::from("test.local"),
4701 path: PathRule::prefix(String::from("/abc")),
4702 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4703 position: RulePosition::Pre.into(),
4704 ..Default::default()
4705 })
4706 .into(),
4707 )
4708 .expect("Could not execute request");
4709 state
4710 .dispatch(
4711 &RequestType::AddBackend(AddBackend {
4712 cluster_id: String::from("cluster_1"),
4713 backend_id: String::from("cluster_1-0"),
4714 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4715 ..Default::default()
4716 })
4717 .into(),
4718 )
4719 .expect("Could not execute request");
4720 state
4721 .dispatch(
4722 &RequestType::AddBackend(AddBackend {
4723 cluster_id: String::from("cluster_1"),
4724 backend_id: String::from("cluster_1-1"),
4725 address: SocketAddress::new_v4(127, 0, 0, 1, 1027),
4726 ..Default::default()
4727 })
4728 .into(),
4729 )
4730 .expect("Could not execute request");
4731 state
4732 .dispatch(
4733 &RequestType::AddBackend(AddBackend {
4734 cluster_id: String::from("cluster_2"),
4735 backend_id: String::from("cluster_2-0"),
4736 address: SocketAddress::new_v4(192, 167, 1, 2, 1026),
4737 ..Default::default()
4738 })
4739 .into(),
4740 )
4741 .expect("Could not execute request");
4742 state
4743 .dispatch(
4744 &RequestType::AddBackend(AddBackend {
4745 cluster_id: String::from("cluster_1"),
4746 backend_id: String::from("cluster_1-3"),
4747 address: SocketAddress::new_v4(192, 168, 1, 3, 1027),
4748 ..Default::default()
4749 })
4750 .into(),
4751 )
4752 .expect("Could not execute request");
4753 state
4754 .dispatch(
4755 &RequestType::RemoveBackend(RemoveBackend {
4756 cluster_id: String::from("cluster_1"),
4757 backend_id: String::from("cluster_1-3"),
4758 address: SocketAddress::new_v4(192, 168, 1, 3, 1027),
4759 })
4760 .into(),
4761 )
4762 .expect("Could not execute request");
4763
4764 }
4774
4775 #[test]
4776 fn diff() {
4777 let mut state: ConfigState = Default::default();
4778 state
4779 .dispatch(
4780 &RequestType::AddHttpFrontend(RequestHttpFrontend {
4781 cluster_id: Some(String::from("cluster_1")),
4782 hostname: String::from("lolcatho.st:8080"),
4783 path: PathRule::prefix(String::from("/")),
4784 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4785 position: RulePosition::Post.into(),
4786 ..Default::default()
4787 })
4788 .into(),
4789 )
4790 .expect("Could not execute request");
4791 state
4792 .dispatch(
4793 &RequestType::AddHttpFrontend(RequestHttpFrontend {
4794 cluster_id: Some(String::from("cluster_2")),
4795 hostname: String::from("test.local"),
4796 path: PathRule::prefix(String::from("/abc")),
4797 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4798 ..Default::default()
4799 })
4800 .into(),
4801 )
4802 .expect("Could not execute request");
4803 state
4804 .dispatch(
4805 &RequestType::AddBackend(AddBackend {
4806 cluster_id: String::from("cluster_1"),
4807 backend_id: String::from("cluster_1-0"),
4808 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4809 load_balancing_parameters: Some(LoadBalancingParams::default()),
4810 ..Default::default()
4811 })
4812 .into(),
4813 )
4814 .expect("Could not execute request");
4815 state
4816 .dispatch(
4817 &RequestType::AddBackend(AddBackend {
4818 cluster_id: String::from("cluster_1"),
4819 backend_id: String::from("cluster_1-1"),
4820 address: SocketAddress::new_v4(127, 0, 0, 2, 1027),
4821 load_balancing_parameters: Some(LoadBalancingParams::default()),
4822 ..Default::default()
4823 })
4824 .into(),
4825 )
4826 .expect("Could not execute request");
4827 state
4828 .dispatch(
4829 &RequestType::AddBackend(AddBackend {
4830 cluster_id: String::from("cluster_2"),
4831 backend_id: String::from("cluster_2-0"),
4832 address: SocketAddress::new_v4(192, 167, 1, 2, 1026),
4833 load_balancing_parameters: Some(LoadBalancingParams::default()),
4834 ..Default::default()
4835 })
4836 .into(),
4837 )
4838 .expect("Could not execute request");
4839 state
4840 .dispatch(
4841 &RequestType::AddCluster(Cluster {
4842 cluster_id: String::from("cluster_2"),
4843 sticky_session: true,
4844 https_redirect: true,
4845 ..Default::default()
4846 })
4847 .into(),
4848 )
4849 .expect("Could not execute request");
4850
4851 let mut state2: ConfigState = Default::default();
4852 state2
4853 .dispatch(
4854 &RequestType::AddHttpFrontend(RequestHttpFrontend {
4855 cluster_id: Some(String::from("cluster_1")),
4856 hostname: String::from("lolcatho.st:8080"),
4857 path: PathRule::prefix(String::from("/")),
4858 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4859 position: RulePosition::Post.into(),
4860 ..Default::default()
4861 })
4862 .into(),
4863 )
4864 .expect("Could not execute request");
4865 state2
4866 .dispatch(
4867 &RequestType::AddBackend(AddBackend {
4868 cluster_id: String::from("cluster_1"),
4869 backend_id: String::from("cluster_1-0"),
4870 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4871 load_balancing_parameters: Some(LoadBalancingParams::default()),
4872 ..Default::default()
4873 })
4874 .into(),
4875 )
4876 .expect("Could not execute request");
4877 state2
4878 .dispatch(
4879 &RequestType::AddBackend(AddBackend {
4880 cluster_id: String::from("cluster_1"),
4881 backend_id: String::from("cluster_1-1"),
4882 address: SocketAddress::new_v4(127, 0, 0, 2, 1027),
4883 load_balancing_parameters: Some(LoadBalancingParams::default()),
4884 ..Default::default()
4885 })
4886 .into(),
4887 )
4888 .expect("Could not execute request");
4889 state2
4890 .dispatch(
4891 &RequestType::AddBackend(AddBackend {
4892 cluster_id: String::from("cluster_1"),
4893 backend_id: String::from("cluster_1-2"),
4894 address: SocketAddress::new_v4(127, 0, 0, 2, 1028),
4895 load_balancing_parameters: Some(LoadBalancingParams::default()),
4896 ..Default::default()
4897 })
4898 .into(),
4899 )
4900 .expect("Could not execute request");
4901 state2
4902 .dispatch(
4903 &RequestType::AddCluster(Cluster {
4904 cluster_id: String::from("cluster_3"),
4905 sticky_session: false,
4906 https_redirect: false,
4907 ..Default::default()
4908 })
4909 .into(),
4910 )
4911 .expect("Could not execute request");
4912
4913 let e: Vec<Request> = vec![
4917 RequestType::AddBackend(AddBackend {
4918 cluster_id: String::from("cluster_1"),
4919 backend_id: String::from("cluster_1-2"),
4920 address: SocketAddress::new_v4(127, 0, 0, 2, 1028),
4921 load_balancing_parameters: Some(LoadBalancingParams::default()),
4922 ..Default::default()
4923 })
4924 .into(),
4925 RequestType::RemoveCluster(String::from("cluster_2")).into(),
4926 RequestType::AddCluster(Cluster {
4927 cluster_id: String::from("cluster_3"),
4928 sticky_session: false,
4929 https_redirect: false,
4930 ..Default::default()
4931 })
4932 .into(),
4933 ];
4934 let expected_diff: HashSet<&Request> = HashSet::from_iter(e.iter());
4935
4936 let d = state.diff(&state2);
4937 let diff = HashSet::from_iter(d.iter());
4938 println!("diff requests:\n{diff:#?}\n");
4939 println!("expected diff requests:\n{expected_diff:#?}\n");
4940
4941 let hash1 = state.hash_state();
4942 let hash2 = state2.hash_state();
4943 let mut state3 = state.clone();
4944 state3
4945 .dispatch(
4946 &RequestType::AddBackend(AddBackend {
4947 cluster_id: String::from("cluster_1"),
4948 backend_id: String::from("cluster_1-2"),
4949 address: SocketAddress::new_v4(127, 0, 0, 2, 1028),
4950 load_balancing_parameters: Some(LoadBalancingParams::default()),
4951 ..Default::default()
4952 })
4953 .into(),
4954 )
4955 .expect("Could not execute request");
4956 let hash3 = state3.hash_state();
4957 println!("state 1 hashes: {hash1:#?}");
4958 println!("state 2 hashes: {hash2:#?}");
4959 println!("state 3 hashes: {hash3:#?}");
4960
4961 assert_eq!(diff, expected_diff);
4962 }
4963
4964 fn state_with_clusters_owning_every_object_kind(cluster_ids: &[&str]) -> ConfigState {
4970 let mut state = ConfigState::new();
4971 for cluster_id in cluster_ids {
4972 let port = match *cluster_id {
4973 "cascade_a" => 9200,
4974 "cascade_b" => 9300,
4975 other => panic!("no test ports reserved for {other}"),
4976 };
4977 let requests: Vec<Request> = vec![
4978 RequestType::AddCluster(Cluster {
4979 cluster_id: cluster_id.to_string(),
4980 ..Default::default()
4981 })
4982 .into(),
4983 RequestType::AddHttpFrontend(RequestHttpFrontend {
4984 cluster_id: Some(cluster_id.to_string()),
4985 hostname: format!("{cluster_id}.test"),
4986 path: PathRule::prefix(String::from("/")),
4987 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4988 ..Default::default()
4989 })
4990 .into(),
4991 RequestType::AddHttpsFrontend(RequestHttpFrontend {
4992 cluster_id: Some(cluster_id.to_string()),
4993 hostname: format!("{cluster_id}.test"),
4994 path: PathRule::prefix(String::from("/")),
4995 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4996 ..Default::default()
4997 })
4998 .into(),
4999 RequestType::AddTcpFrontend(RequestTcpFrontend {
5000 cluster_id: cluster_id.to_string(),
5001 address: SocketAddress::new_v4(127, 0, 0, 1, port),
5002 ..Default::default()
5003 })
5004 .into(),
5005 RequestType::AddUdpFrontend(RequestUdpFrontend {
5006 cluster_id: cluster_id.to_string(),
5007 address: SocketAddress::new_v4(127, 0, 0, 1, port + 1),
5008 ..Default::default()
5009 })
5010 .into(),
5011 RequestType::AddBackend(AddBackend {
5012 cluster_id: cluster_id.to_string(),
5013 backend_id: format!("{cluster_id}-0"),
5014 address: SocketAddress::new_v4(127, 0, 0, 1, port + 2),
5015 load_balancing_parameters: Some(LoadBalancingParams::default()),
5016 ..Default::default()
5017 })
5018 .into(),
5019 ];
5020 for request in &requests {
5021 state
5022 .dispatch(request)
5023 .unwrap_or_else(|e| panic!("could not apply {request:?}: {e}"));
5024 }
5025 }
5026 state
5027 .dispatch(&RequestType::AddHttpFrontend(cascade_deny_route()).into())
5028 .expect("could not add the deny route");
5029 state
5030 }
5031
5032 fn cascade_deny_route() -> RequestHttpFrontend {
5034 RequestHttpFrontend {
5035 cluster_id: None,
5036 hostname: String::from("cascade_a.test"),
5037 path: PathRule::prefix(String::from("/deny")),
5038 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5039 ..Default::default()
5040 }
5041 }
5042
5043 fn cluster_named_by(request: &Request) -> Option<&str> {
5045 match request.request_type.as_ref()? {
5046 RequestType::AddCluster(cluster) => Some(&cluster.cluster_id),
5047 RequestType::RemoveCluster(cluster_id) => Some(cluster_id),
5048 RequestType::AddHttpFrontend(front)
5049 | RequestType::RemoveHttpFrontend(front)
5050 | RequestType::AddHttpsFrontend(front)
5051 | RequestType::RemoveHttpsFrontend(front) => front.cluster_id.as_deref(),
5052 RequestType::AddTcpFrontend(front) | RequestType::RemoveTcpFrontend(front) => {
5053 Some(&front.cluster_id)
5054 }
5055 RequestType::AddUdpFrontend(front) | RequestType::RemoveUdpFrontend(front) => {
5056 Some(&front.cluster_id)
5057 }
5058 RequestType::AddBackend(backend) => Some(&backend.cluster_id),
5059 RequestType::RemoveBackend(backend) => Some(&backend.cluster_id),
5060 _ => None,
5061 }
5062 }
5063
5064 fn assert_same_routing(actual: &ConfigState, expected: &ConfigState, context: &str) {
5066 assert_eq!(actual.clusters, expected.clusters, "{context}: clusters");
5067 assert_eq!(actual.backends, expected.backends, "{context}: backends");
5068 assert_eq!(
5069 actual.http_fronts, expected.http_fronts,
5070 "{context}: http_fronts"
5071 );
5072 assert_eq!(
5073 actual.https_fronts, expected.https_fronts,
5074 "{context}: https_fronts"
5075 );
5076 assert_eq!(
5077 actual.tcp_fronts, expected.tcp_fronts,
5078 "{context}: tcp_fronts"
5079 );
5080 assert_eq!(
5081 actual.udp_fronts, expected.udp_fronts,
5082 "{context}: udp_fronts"
5083 );
5084 }
5085
5086 #[test]
5090 fn remove_cluster_removes_its_frontends_and_backends_for_every_protocol() {
5091 let mut state = state_with_clusters_owning_every_object_kind(&["cascade_a", "cascade_b"]);
5092 let expected = state_with_clusters_owning_every_object_kind(&["cascade_b"]);
5093
5094 state
5095 .dispatch(&RequestType::RemoveCluster(String::from("cascade_a")).into())
5096 .expect("removing an existing cluster must succeed");
5097
5098 assert_same_routing(&state, &expected, "after RemoveCluster(cascade_a)");
5099 assert!(
5100 state
5101 .http_fronts
5102 .values()
5103 .any(|front| front.cluster_id.is_none() && front.hostname == "cascade_a.test"),
5104 "a route that names no cluster must survive the removal of a cluster on its hostname"
5105 );
5106
5107 let late: Request = RequestType::RemoveHttpFrontend(RequestHttpFrontend {
5111 cluster_id: Some(String::from("cascade_a")),
5112 hostname: String::from("cascade_a.test"),
5113 path: PathRule::prefix(String::from("/")),
5114 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5115 ..Default::default()
5116 })
5117 .into();
5118 let err = state
5119 .dispatch(&late)
5120 .expect_err("the cascade must already have removed the frontend");
5121 assert!(
5122 matches!(err, StateError::NotFound { .. }),
5123 "expected NotFound, got: {err}"
5124 );
5125 assert!(
5126 state.removed_with_its_cluster(&late, &err),
5127 "a removal of an object that went with its cluster must be recognised"
5128 );
5129
5130 let mut typo = cascade_deny_route();
5134 typo.cluster_id = Some(String::from("cascade_typo"));
5135 let typo: Request = RequestType::RemoveHttpFrontend(typo).into();
5136 let err = state
5137 .dispatch(&typo)
5138 .expect_err("a removal naming another cluster must not match the deny route");
5139 assert!(state.removed_with_its_cluster(&typo, &err));
5140 let answer = state.removed_with_its_cluster_message(&typo);
5141 assert!(
5142 answer.contains("this key belongs to a route with no cluster, left in place"),
5143 "the ok answer must say the deny route holds the key: {answer}"
5144 );
5145 assert!(
5146 state
5147 .http_fronts
5148 .values()
5149 .any(|front| front.cluster_id.is_none() && front.hostname == "cascade_a.test"),
5150 "the deny route must survive a removal naming a cluster that never existed"
5151 );
5152 }
5153
5154 #[test]
5160 fn a_stale_frontend_removal_never_removes_another_clusters_route() {
5161 let route = |cluster_id: &str| RequestHttpFrontend {
5162 cluster_id: Some(cluster_id.to_owned()),
5163 hostname: String::from("reused.test"),
5164 path: PathRule::prefix(String::from("/")),
5165 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5166 ..Default::default()
5167 };
5168 let cluster = |cluster_id: &str| {
5169 RequestType::AddCluster(Cluster {
5170 cluster_id: cluster_id.to_owned(),
5171 ..Default::default()
5172 })
5173 };
5174 let owner =
5175 |state: &ConfigState, fronts: fn(&ConfigState) -> &BTreeMap<String, HttpFrontend>| {
5176 fronts(state)
5177 .get(&route("stale_b").to_string())
5178 .and_then(|front| front.cluster_id.as_deref().map(str::to_owned))
5179 };
5180
5181 for (add, remove, fronts) in [
5182 (
5183 RequestType::AddHttpFrontend as fn(RequestHttpFrontend) -> RequestType,
5184 RequestType::RemoveHttpFrontend as fn(RequestHttpFrontend) -> RequestType,
5185 (|state: &ConfigState| &state.http_fronts)
5186 as fn(&ConfigState) -> &BTreeMap<String, HttpFrontend>,
5187 ),
5188 (
5189 RequestType::AddHttpsFrontend,
5190 RequestType::RemoveHttpsFrontend,
5191 |state: &ConfigState| &state.https_fronts,
5192 ),
5193 ] {
5194 let mut state = ConfigState::new();
5197 for request in [
5198 cluster("stale_a"),
5199 add(route("stale_a")),
5200 RequestType::RemoveCluster(String::from("stale_a")),
5201 cluster("stale_b"),
5202 add(route("stale_b")),
5203 ] {
5204 state
5205 .dispatch(&request.clone().into())
5206 .unwrap_or_else(|e| panic!("could not apply {request:?}: {e}"));
5207 }
5208 let stale: Request = remove(route("stale_a")).into();
5209 let err = state
5210 .dispatch(&stale)
5211 .expect_err("a removal naming cluster A must not match cluster B's route");
5212 assert!(
5213 state.removed_with_its_cluster(&stale, &err),
5214 "A is gone, so its frontend went with it: {err}"
5215 );
5216 let answer = state.removed_with_its_cluster_message(&stale);
5217 assert!(
5218 answer.contains("cluster stale_a does not exist")
5219 && answer.contains("this key belongs to cluster stale_b, left in place"),
5220 "the ok answer must name the key's current owner: {answer}"
5221 );
5222 assert_eq!(
5223 owner(&state, fronts).as_deref(),
5224 Some("stale_b"),
5225 "cluster B's route must survive a stale removal naming cluster A"
5226 );
5227
5228 let mut state = ConfigState::new();
5231 for request in [
5232 cluster("stale_a"),
5233 cluster("stale_b"),
5234 add(route("stale_b")),
5235 ] {
5236 state
5237 .dispatch(&request.clone().into())
5238 .unwrap_or_else(|e| panic!("could not apply {request:?}: {e}"));
5239 }
5240 let err = state
5241 .dispatch(&stale)
5242 .expect_err("a removal naming cluster A must not match cluster B's route");
5243 assert!(
5244 !state.removed_with_its_cluster(&stale, &err),
5245 "A still exists: the removal is refused, not already done"
5246 );
5247 assert_eq!(
5248 owner(&state, fronts).as_deref(),
5249 Some("stale_b"),
5250 "cluster B's route must survive a removal naming cluster A"
5251 );
5252 }
5253 }
5254
5255 #[test]
5259 fn diff_removing_a_cluster_relies_on_the_cascade() {
5260 let before = state_with_clusters_owning_every_object_kind(&["cascade_a", "cascade_b"]);
5261 let after = state_with_clusters_owning_every_object_kind(&["cascade_b"]);
5262
5263 let diff = before.diff(&after);
5264 assert_eq!(
5265 diff,
5266 vec![Request::from(RequestType::RemoveCluster(String::from(
5267 "cascade_a"
5268 )))],
5269 "the cascade must be the only removal the diff emits"
5270 );
5271
5272 let mut replayed = before.clone();
5273 for request in &diff {
5274 replayed
5275 .dispatch(request)
5276 .unwrap_or_else(|e| panic!("could not replay {request:?}: {e}"));
5277 }
5278 assert_same_routing(&replayed, &after, "replaying diff(before, after)");
5279 }
5280
5281 #[test]
5285 fn diff_re_adds_a_frontend_the_target_keeps_for_a_removed_cluster() {
5286 let before = state_with_clusters_owning_every_object_kind(&["cascade_a", "cascade_b"]);
5287 let kept = RequestHttpFrontend {
5288 cluster_id: Some(String::from("cascade_a")),
5289 hostname: String::from("cascade_a.test"),
5290 path: PathRule::prefix(String::from("/")),
5291 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5292 ..Default::default()
5293 };
5294 let mut after = state_with_clusters_owning_every_object_kind(&["cascade_b"]);
5295 after
5296 .dispatch(&RequestType::AddHttpFrontend(kept.clone()).into())
5297 .expect("a frontend may name a cluster the state does not hold");
5298
5299 let diff = before.diff(&after);
5300 let removal = diff
5301 .iter()
5302 .position(|r| r.request_type == Some(RequestType::RemoveCluster("cascade_a".into())))
5303 .expect("the diff must remove cascade_a");
5304 let re_add = diff
5305 .iter()
5306 .position(|r| r.request_type == Some(RequestType::AddHttpFrontend(kept.clone())))
5307 .expect("the diff must add back the frontend the cascade removes");
5308 assert!(
5309 removal < re_add,
5310 "the frontend must be added back after its cluster is removed: {diff:?}"
5311 );
5312
5313 let mut replayed = before.clone();
5314 for request in &diff {
5315 replayed
5316 .dispatch(request)
5317 .unwrap_or_else(|e| panic!("could not replay {request:?}: {e}"));
5318 }
5319 assert_same_routing(&replayed, &after, "replaying diff(before, after)");
5320 }
5321
5322 #[test]
5326 fn a_removed_cluster_leaves_nothing_for_load_state_to_replay() {
5327 let mut state = state_with_clusters_owning_every_object_kind(&["cascade_a", "cascade_b"]);
5328 state
5329 .dispatch(&RequestType::RemoveCluster(String::from("cascade_a")).into())
5330 .expect("removing an existing cluster must succeed");
5331
5332 let requests = state.generate_requests();
5333 let leaked: Vec<&Request> = requests
5334 .iter()
5335 .filter(|request| cluster_named_by(request) == Some("cascade_a"))
5336 .collect();
5337 assert!(
5338 leaked.is_empty(),
5339 "LoadState would replay objects of the removed cluster: {leaked:?}"
5340 );
5341
5342 let mut replayed = ConfigState::new();
5343 for request in &requests {
5344 replayed
5345 .dispatch(request)
5346 .unwrap_or_else(|e| panic!("could not replay {request:?}: {e}"));
5347 }
5348 assert_same_routing(&replayed, &state, "replaying generate_requests");
5349 }
5350
5351 #[test]
5352 fn cluster_ids_by_domain() {
5353 let mut config = ConfigState::new();
5354 let http_front_cluster1 = RequestHttpFrontend {
5355 cluster_id: Some(String::from("MyCluster_1")),
5356 hostname: String::from("lolcatho.st"),
5357 path: PathRule::prefix(String::from("")),
5358 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5359 ..Default::default()
5360 };
5361
5362 let https_front_cluster1 = RequestHttpFrontend {
5363 cluster_id: Some(String::from("MyCluster_1")),
5364 hostname: String::from("lolcatho.st"),
5365 path: PathRule::prefix(String::from("")),
5366 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5367 ..Default::default()
5368 };
5369
5370 let http_front_cluster2 = RequestHttpFrontend {
5371 cluster_id: Some(String::from("MyCluster_2")),
5372 hostname: String::from("lolcatho.st"),
5373 path: PathRule::prefix(String::from("/api")),
5374 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5375 ..Default::default()
5376 };
5377
5378 let https_front_cluster2 = RequestHttpFrontend {
5379 cluster_id: Some(String::from("MyCluster_2")),
5380 hostname: String::from("lolcatho.st"),
5381 path: PathRule::prefix(String::from("/api")),
5382 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5383 ..Default::default()
5384 };
5385
5386 config
5387 .dispatch(&RequestType::AddHttpFrontend(http_front_cluster1).into())
5388 .expect("Could not execute request");
5389 config
5390 .dispatch(&RequestType::AddHttpFrontend(http_front_cluster2).into())
5391 .expect("Could not execute request");
5392 config
5393 .dispatch(&RequestType::AddHttpsFrontend(https_front_cluster1).into())
5394 .expect("Could not execute request");
5395 config
5396 .dispatch(&RequestType::AddHttpsFrontend(https_front_cluster2).into())
5397 .expect("Could not execute request");
5398
5399 let mut cluster1_cluster2: HashSet<ClusterId> = HashSet::new();
5400 cluster1_cluster2.insert(ClusterId::from("MyCluster_1"));
5401 cluster1_cluster2.insert(ClusterId::from("MyCluster_2"));
5402
5403 let mut cluster2: HashSet<ClusterId> = HashSet::new();
5404 cluster2.insert(ClusterId::from("MyCluster_2"));
5405
5406 let empty: HashSet<ClusterId> = HashSet::new();
5407 assert_eq!(
5408 config.get_cluster_ids_by_domain(String::from("lolcatho.st"), None),
5409 cluster1_cluster2
5410 );
5411 assert_eq!(
5412 config
5413 .get_cluster_ids_by_domain(String::from("lolcatho.st"), Some(String::from("/api"))),
5414 cluster2
5415 );
5416 assert_eq!(
5417 config.get_cluster_ids_by_domain(String::from("lolcathost"), None),
5418 empty
5419 );
5420 assert_eq!(
5421 config
5422 .get_cluster_ids_by_domain(String::from("lolcathost"), Some(String::from("/sozu"))),
5423 empty
5424 );
5425 }
5426
5427 #[test]
5428 fn duplicate_backends() {
5429 let mut state: ConfigState = Default::default();
5430 state
5431 .dispatch(
5432 &RequestType::AddBackend(AddBackend {
5433 cluster_id: String::from("cluster_1"),
5434 backend_id: String::from("cluster_1-0"),
5435 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
5436 load_balancing_parameters: Some(LoadBalancingParams::default()),
5437 ..Default::default()
5438 })
5439 .into(),
5440 )
5441 .expect("Could not execute request");
5442
5443 let b = Backend {
5444 cluster_id: ClusterId::from("cluster_1"),
5445 backend_id: String::from("cluster_1-0"),
5446 address: "127.0.0.1:1026".parse().unwrap(),
5447 load_balancing_parameters: Some(LoadBalancingParams::default()),
5448 sticky_id: Some("sticky".to_string()),
5449 backup: None,
5450 };
5451
5452 state
5453 .dispatch(&RequestType::AddBackend(b.clone().to_add_backend()).into())
5454 .expect("Could not execute order");
5455
5456 assert_eq!(state.backends.get("cluster_1").unwrap(), &vec![b]);
5457 }
5458
5459 #[test]
5460 fn remove_backend() {
5461 let mut state: ConfigState = Default::default();
5462 state
5463 .dispatch(
5464 &RequestType::AddCluster(Cluster {
5465 cluster_id: String::from("cluster_1"),
5466 ..Default::default()
5467 })
5468 .into(),
5469 )
5470 .expect("Could not execute request");
5471
5472 for i in 0..10 {
5473 state
5474 .dispatch(
5475 &RequestType::AddBackend(AddBackend {
5476 cluster_id: String::from("cluster_1"),
5477 backend_id: format!("cluster_1-{i}"),
5478 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
5479 ..Default::default()
5480 })
5481 .into(),
5482 )
5483 .expect("Could not execute request");
5484 }
5485
5486 assert_eq!(state.backends.get("cluster_1").unwrap().len(), 10);
5487
5488 let remove_backend_2 = RequestType::RemoveBackend(RemoveBackend {
5489 cluster_id: String::from("cluster_1"),
5490 backend_id: String::from("cluster_1-0"),
5491 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
5492 })
5493 .into();
5494
5495 let remove_backend_result = state.dispatch(&remove_backend_2);
5496
5497 assert!(remove_backend_result.is_ok());
5498 assert_eq!(state.backends.get("cluster_1").unwrap().len(), 9);
5499
5500 let redundant_remove = state.dispatch(&remove_backend_2);
5501 assert!(matches!(redundant_remove, Err(StateError::NoChange)));
5502 assert_eq!(state.backends.get("cluster_1").unwrap().len(), 9);
5503 }
5504
5505 #[test]
5506 fn remove_backends_randomly() {
5507 let mut state: ConfigState = Default::default();
5508 state
5509 .dispatch(
5510 &RequestType::AddCluster(Cluster {
5511 cluster_id: String::from("cluster_1"),
5512 ..Default::default()
5513 })
5514 .into(),
5515 )
5516 .expect("Could not execute request");
5517
5518 for _ in 0..1000 {
5519 for i in 0..10 {
5520 state
5521 .dispatch(
5522 &RequestType::AddBackend(AddBackend {
5523 cluster_id: String::from("cluster_1"),
5524 backend_id: format!("cluster_1-{i}"),
5525 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
5526 ..Default::default()
5527 })
5528 .into(),
5529 )
5530 .expect("Could not execute request");
5531 }
5532
5533 let mut rng = rng();
5534 let mut indexes = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9];
5535 indexes.shuffle(&mut rng);
5536 let random_count = rng.random_range(1..indexes.len());
5537 let random_indexes: Vec<i32> = indexes.into_iter().take(random_count).collect();
5538
5539 for j in random_indexes {
5540 let remove_backend_result = state.dispatch(
5541 &RequestType::RemoveBackend(RemoveBackend {
5542 cluster_id: String::from("cluster_1"),
5543 backend_id: format!("cluster_1-{j}"),
5544 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
5545 })
5546 .into(),
5547 );
5548 assert!(remove_backend_result.is_ok());
5549 }
5550 }
5551 }
5552
5553 #[test]
5554 fn listener_diff() {
5555 let mut state: ConfigState = Default::default();
5556 let custom_http_answers = Some(CustomHttpAnswers {
5557 answer_404: Some("test".to_string()),
5558 ..Default::default()
5559 });
5560 state
5561 .dispatch(
5562 &RequestType::AddTcpListener(TcpListenerConfig {
5563 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
5564 ..Default::default()
5565 })
5566 .into(),
5567 )
5568 .expect("Could not execute request");
5569 state
5570 .dispatch(
5571 &RequestType::ActivateListener(ActivateListener {
5572 interface: None,
5573 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
5574 proxy: ListenerType::Tcp.into(),
5575 from_scm: false,
5576 })
5577 .into(),
5578 )
5579 .expect("Could not execute request");
5580 state
5581 .dispatch(
5582 &RequestType::AddHttpListener(HttpListenerConfig {
5583 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5584 ..Default::default()
5585 })
5586 .into(),
5587 )
5588 .expect("Could not execute request");
5589 state
5590 .dispatch(
5591 &RequestType::AddHttpsListener(HttpsListenerConfig {
5592 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5593 ..Default::default()
5594 })
5595 .into(),
5596 )
5597 .expect("Could not execute request");
5598 state
5599 .dispatch(
5600 &RequestType::ActivateListener(ActivateListener {
5601 interface: None,
5602 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5603 proxy: ListenerType::Https.into(),
5604 from_scm: false,
5605 })
5606 .into(),
5607 )
5608 .expect("Could not execute request");
5609
5610 let mut state2: ConfigState = Default::default();
5611 state2
5612 .dispatch(
5613 &RequestType::AddTcpListener(TcpListenerConfig {
5614 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
5615 expect_proxy: true,
5616 ..Default::default()
5617 })
5618 .into(),
5619 )
5620 .expect("Could not execute request");
5621 state2
5622 .dispatch(
5623 &RequestType::AddHttpListener(HttpListenerConfig {
5624 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5625 http_answers: custom_http_answers.clone(),
5626 ..Default::default()
5627 })
5628 .into(),
5629 )
5630 .expect("Could not execute request");
5631 state2
5632 .dispatch(
5633 &RequestType::ActivateListener(ActivateListener {
5634 interface: None,
5635 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5636 proxy: ListenerType::Http.into(),
5637 from_scm: false,
5638 })
5639 .into(),
5640 )
5641 .expect("Could not execute request");
5642 state2
5643 .dispatch(
5644 &RequestType::AddHttpsListener(HttpsListenerConfig {
5645 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5646 http_answers: custom_http_answers.clone(),
5647 ..Default::default()
5648 })
5649 .into(),
5650 )
5651 .expect("Could not execute request");
5652 state2
5653 .dispatch(
5654 &RequestType::ActivateListener(ActivateListener {
5655 interface: None,
5656 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5657 proxy: ListenerType::Https.into(),
5658 from_scm: false,
5659 })
5660 .into(),
5661 )
5662 .expect("Could not execute request");
5663
5664 let e: Vec<Request> = vec![
5665 RequestType::RemoveListener(RemoveListener {
5666 interface: None,
5667 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
5668 proxy: ListenerType::Tcp.into(),
5669 })
5670 .into(),
5671 RequestType::AddTcpListener(TcpListenerConfig {
5672 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
5673 expect_proxy: true,
5674 ..Default::default()
5675 })
5676 .into(),
5677 RequestType::DeactivateListener(DeactivateListener {
5678 interface: None,
5679 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
5680 proxy: ListenerType::Tcp.into(),
5681 to_scm: false,
5682 })
5683 .into(),
5684 RequestType::RemoveListener(RemoveListener {
5685 interface: None,
5686 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5687 proxy: ListenerType::Http.into(),
5688 })
5689 .into(),
5690 RequestType::AddHttpListener(HttpListenerConfig {
5691 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5692 http_answers: custom_http_answers.clone(),
5693 ..Default::default()
5694 })
5695 .into(),
5696 RequestType::ActivateListener(ActivateListener {
5697 interface: None,
5698 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5699 proxy: ListenerType::Http.into(),
5700 from_scm: false,
5701 })
5702 .into(),
5703 RequestType::RemoveListener(RemoveListener {
5704 interface: None,
5705 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5706 proxy: ListenerType::Https.into(),
5707 })
5708 .into(),
5709 RequestType::AddHttpsListener(HttpsListenerConfig {
5710 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5711 http_answers: custom_http_answers.clone(),
5712 ..Default::default()
5713 })
5714 .into(),
5715 RequestType::ActivateListener(ActivateListener {
5721 interface: None,
5722 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5723 proxy: ListenerType::Https.into(),
5724 from_scm: false,
5725 })
5726 .into(),
5727 ];
5728
5729 let diff = state.diff(&state2);
5730 println!("expected diff requests:\n{e:#?}\n");
5732 println!("diff requests:\n{diff:#?}\n");
5733
5734 let _hash1 = state.hash_state();
5735 let _hash2 = state2.hash_state();
5736
5737 assert_eq!(diff, e);
5738
5739 let mut replayed = state.clone();
5746 for request in &diff {
5747 replayed
5748 .dispatch(request)
5749 .expect("every diff request must replay cleanly onto the source state");
5750 }
5751 assert_eq!(
5752 replayed.tcp_listeners, state2.tcp_listeners,
5753 "replayed tcp_listeners must match the target state"
5754 );
5755 assert_eq!(
5756 replayed.http_listeners, state2.http_listeners,
5757 "replayed http_listeners must match the target state"
5758 );
5759 assert_eq!(
5760 replayed.https_listeners, state2.https_listeners,
5761 "replayed https_listeners must match the target state"
5762 );
5763 let replayed_8443 = replayed
5766 .https_listeners
5767 .get(&ListenerKey::from(SocketAddress::new_v4(0, 0, 0, 0, 8443)))
5768 .expect("8443 HTTPS listener must exist after replay");
5769 assert!(
5770 replayed_8443.active,
5771 "the 8443 HTTPS listener must stay ACTIVE across a config change"
5772 );
5773 }
5774
5775 #[test]
5776 fn certificate_retrieval() {
5777 let mut state: ConfigState = Default::default();
5778 let certificate_and_key = CertificateAndKey {
5779 certificate: String::from(include_str!("../assets/certificate.pem")),
5780 key: String::from(include_str!("../assets/key.pem")),
5781 certificate_chain: vec![],
5782 versions: vec![],
5783 names: vec!["lolcatho.st".to_string()],
5784 };
5785 let add_certificate = AddCertificate {
5786 address: SocketAddress::new_v4(127, 0, 0, 1, 8080),
5787 certificate: certificate_and_key,
5788 expired_at: None,
5789 };
5790 state
5791 .dispatch(&RequestType::AddCertificate(add_certificate).into())
5792 .expect("Could not add certificate");
5793
5794 println!("state: {state:#?}");
5795
5796 let certificates_found_by_fingerprint = state.get_certificates(QueryCertificatesFilters {
5802 domain: None,
5803 fingerprint: Some(
5804 "ab2618b674e15243fd02a5618c66509e4840ba60e7d64cebec84cdbfeceee0c5".to_string(),
5805 ),
5806 });
5807
5808 println!("found certificate: {certificates_found_by_fingerprint:#?}");
5809
5810 assert!(!certificates_found_by_fingerprint.is_empty());
5811
5812 let certificate_found_by_domain_name = state.get_certificates(QueryCertificatesFilters {
5813 domain: Some("lolcatho.st".to_string()),
5814 fingerprint: None,
5815 });
5816
5817 assert!(!certificate_found_by_domain_name.is_empty());
5818 }
5819
5820 #[test]
5821 fn count_backends_across_clusters() {
5822 let mut state: ConfigState = Default::default();
5823
5824 assert_eq!(state.count_backends(), 0);
5825
5826 state
5827 .dispatch(
5828 &RequestType::AddBackend(AddBackend {
5829 cluster_id: String::from("cluster_1"),
5830 backend_id: String::from("cluster_1-0"),
5831 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
5832 ..Default::default()
5833 })
5834 .into(),
5835 )
5836 .expect("Could not execute request");
5837 assert_eq!(state.count_backends(), 1);
5838
5839 state
5840 .dispatch(
5841 &RequestType::AddBackend(AddBackend {
5842 cluster_id: String::from("cluster_1"),
5843 backend_id: String::from("cluster_1-1"),
5844 address: SocketAddress::new_v4(127, 0, 0, 1, 1027),
5845 ..Default::default()
5846 })
5847 .into(),
5848 )
5849 .expect("Could not execute request");
5850 assert_eq!(state.count_backends(), 2);
5851
5852 state
5854 .dispatch(
5855 &RequestType::AddBackend(AddBackend {
5856 cluster_id: String::from("cluster_2"),
5857 backend_id: String::from("cluster_2-0"),
5858 address: SocketAddress::new_v4(192, 168, 1, 1, 8080),
5859 ..Default::default()
5860 })
5861 .into(),
5862 )
5863 .expect("Could not execute request");
5864 assert_eq!(state.count_backends(), 3);
5865
5866 state
5868 .dispatch(
5869 &RequestType::RemoveBackend(RemoveBackend {
5870 cluster_id: String::from("cluster_1"),
5871 backend_id: String::from("cluster_1-0"),
5872 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
5873 })
5874 .into(),
5875 )
5876 .expect("Could not execute request");
5877 assert_eq!(state.count_backends(), 2);
5878 }
5879
5880 #[test]
5910 fn a_regex_hostname_is_always_replayed_before_the_exact_name_it_matches() {
5911 let mut state: ConfigState = Default::default();
5912
5913 for hostname in ["test4.example.com", "/test[0-9]/.example.com"] {
5916 state
5917 .dispatch(
5918 &RequestType::AddHttpFrontend(RequestHttpFrontend {
5919 cluster_id: Some(String::from("cluster_1")),
5920 hostname: String::from(hostname),
5921 path: PathRule::prefix(String::from("/")),
5922 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5923 position: RulePosition::Tree.into(),
5924 ..Default::default()
5925 })
5926 .into(),
5927 )
5928 .expect("Could not execute request");
5929 }
5930
5931 let replayed: Vec<String> = state
5934 .produce_initial_state()
5935 .requests
5936 .into_iter()
5937 .filter_map(|worker_request| match worker_request.content.request_type {
5938 Some(RequestType::AddHttpFrontend(front)) => Some(front.hostname),
5939 _ => None,
5940 })
5941 .collect();
5942
5943 assert_eq!(
5944 replayed,
5945 vec![
5946 String::from("/test[0-9]/.example.com"),
5947 String::from("test4.example.com"),
5948 ],
5949 "a worker replays the regex segment before the exact name it \
5950 matches, whatever order the operator added them in",
5951 );
5952 }
5953
5954 #[test]
5955 fn count_frontends_across_types() {
5956 let mut state: ConfigState = Default::default();
5957
5958 assert_eq!(state.count_frontends(), 0);
5959
5960 state
5962 .dispatch(
5963 &RequestType::AddHttpFrontend(RequestHttpFrontend {
5964 cluster_id: Some(String::from("cluster_1")),
5965 hostname: String::from("example.com"),
5966 path: PathRule::prefix(String::from("/")),
5967 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
5968 position: RulePosition::Tree.into(),
5969 ..Default::default()
5970 })
5971 .into(),
5972 )
5973 .expect("Could not execute request");
5974 assert_eq!(state.count_frontends(), 1);
5975
5976 state
5978 .dispatch(
5979 &RequestType::AddHttpsFrontend(RequestHttpFrontend {
5980 cluster_id: Some(String::from("cluster_1")),
5981 hostname: String::from("secure.example.com"),
5982 path: PathRule::prefix(String::from("/")),
5983 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
5984 position: RulePosition::Tree.into(),
5985 ..Default::default()
5986 })
5987 .into(),
5988 )
5989 .expect("Could not execute request");
5990 assert_eq!(state.count_frontends(), 2);
5991
5992 state
5994 .dispatch(
5995 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5996 cluster_id: String::from("cluster_2"),
5997 address: SocketAddress::new_v4(0, 0, 0, 0, 5432),
5998 ..Default::default()
5999 })
6000 .into(),
6001 )
6002 .expect("Could not execute request");
6003 assert_eq!(state.count_frontends(), 3);
6004
6005 state
6007 .dispatch(
6008 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6009 cluster_id: String::from("cluster_2"),
6010 address: SocketAddress::new_v4(0, 0, 0, 0, 5433),
6011 ..Default::default()
6012 })
6013 .into(),
6014 )
6015 .expect("Could not execute request");
6016 assert_eq!(state.count_frontends(), 4);
6017
6018 state
6020 .dispatch(
6021 &RequestType::RemoveHttpFrontend(RequestHttpFrontend {
6022 cluster_id: Some(String::from("cluster_1")),
6023 hostname: String::from("example.com"),
6024 path: PathRule::prefix(String::from("/")),
6025 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
6026 position: RulePosition::Tree.into(),
6027 ..Default::default()
6028 })
6029 .into(),
6030 )
6031 .expect("Could not execute request");
6032 assert_eq!(state.count_frontends(), 3);
6033 }
6034
6035 #[test]
6041 fn https_listener_requests_carry_client_auth_on_their_own_verb() {
6042 let mtls_address = SocketAddress::new_v4(127, 0, 0, 1, 8443);
6043 let plain_address = SocketAddress::new_v4(127, 0, 0, 1, 8444);
6044 let mtls = HttpsListenerConfig {
6045 client_auth: Some(ClientAuthMode::ClientAuthRequired as i32),
6046 client_ca_certificates: vec!["CLIENT CA PEM".to_owned()],
6047 ..make_https_listener(mtls_address)
6048 };
6049
6050 let mut state = ConfigState::new();
6051 state
6052 .dispatch(&RequestType::AddHttpsListenerWithClientAuth(mtls.clone()).into())
6053 .expect("the mTLS verb must add the listener to the state");
6054 state
6055 .dispatch(&RequestType::AddHttpsListener(make_https_listener(plain_address)).into())
6056 .expect("the historical verb must keep adding listeners without client auth");
6057 assert_eq!(
6058 state.https_listeners.len(),
6059 2,
6060 "both verbs add a listener to the same state"
6061 );
6062
6063 let https_adds = |requests: Vec<Request>| -> Vec<(SocketAddress, bool)> {
6064 requests
6065 .into_iter()
6066 .filter_map(|request| match request.request_type {
6067 Some(RequestType::AddHttpsListener(listener)) => {
6068 Some((listener.address, false))
6069 }
6070 Some(RequestType::AddHttpsListenerWithClientAuth(listener)) => {
6071 Some((listener.address, true))
6072 }
6073 _ => None,
6074 })
6075 .collect()
6076 };
6077 let mut generated = https_adds(state.generate_requests());
6078 generated.sort_by_key(|(address, _)| address.port);
6079 assert_eq!(
6080 generated,
6081 vec![(mtls_address, true), (plain_address, false)],
6082 "generate_requests must put only the listener asking for a client certificate on the mTLS verb"
6083 );
6084
6085 let diff = https_adds(ConfigState::new().diff(&state));
6086 assert!(
6087 diff.contains(&(mtls_address, true)) && !diff.contains(&(mtls_address, false)),
6088 "the reload diff must add the mTLS listener on the mTLS verb: {diff:?}"
6089 );
6090 }
6091
6092 #[test]
6097 fn a_reload_refreshes_the_client_auth_policy_of_a_running_listener() {
6098 let address = SocketAddress::new_v4(127, 0, 0, 1, 8443);
6099 let with_crl = |crl: &str| HttpsListenerConfig {
6100 client_auth: Some(ClientAuthMode::ClientAuthRequired as i32),
6101 client_ca_certificates: vec!["CLIENT CA PEM".to_owned()],
6102 client_ca_crls: vec![crl.to_owned()],
6103 ..make_https_listener(address)
6104 };
6105
6106 let mut state = ConfigState::new();
6107 assert_eq!(
6108 state.client_auth_refresh(&with_crl("CRL 1")),
6109 None,
6110 "a listener the state does not hold is added, not patched"
6111 );
6112 state
6113 .dispatch(&RequestType::add_https_listener(with_crl("CRL 1")).into())
6114 .expect("add the listener");
6115 assert_eq!(
6116 state.client_auth_refresh(&with_crl("CRL 1")),
6117 None,
6118 "an unchanged policy needs no patch"
6119 );
6120 assert!(
6121 matches!(
6122 state.dispatch(&RequestType::add_https_listener(with_crl("CRL 2")).into()),
6123 Err(StateError::Exists { .. })
6124 ),
6125 "precondition: replaying the add for a held address is refused"
6126 );
6127
6128 let patch = state
6129 .client_auth_refresh(&with_crl("CRL 2"))
6130 .expect("a refreshed CRL must produce a patch");
6131 assert_eq!(patch.address, address);
6132 assert_eq!(
6133 patch.client_auth_policy,
6134 Some(with_crl("CRL 2").client_auth_policy())
6135 );
6136 assert_eq!(
6137 UpdateHttpsListenerConfig {
6138 client_auth_policy: None,
6139 ..patch.clone()
6140 },
6141 UpdateHttpsListenerConfig {
6142 address,
6143 ..Default::default()
6144 },
6145 "the patch carries the policy and nothing else"
6146 );
6147
6148 state
6149 .dispatch(&RequestType::update_https_listener(patch).into())
6150 .expect("the state must apply the policy patch");
6151 let running = state
6152 .https_listeners
6153 .values()
6154 .next()
6155 .expect("the listener is still there");
6156 assert_eq!(running.client_ca_crls, vec!["CRL 2".to_owned()]);
6157 assert_eq!(
6158 state.client_auth_refresh(&with_crl("CRL 2")),
6159 None,
6160 "once applied, the same reload has nothing left to refresh"
6161 );
6162 }
6163
6164 fn make_https_listener(address: SocketAddress) -> HttpsListenerConfig {
6167 HttpsListenerConfig {
6168 address,
6169 sticky_name: "SOZUBALANCEID".to_owned(),
6170 front_timeout: 60,
6171 back_timeout: 30,
6172 connect_timeout: 3,
6173 request_timeout: 10,
6174 ..Default::default()
6175 }
6176 }
6177
6178 fn make_http_listener(address: SocketAddress) -> HttpListenerConfig {
6179 HttpListenerConfig {
6180 address,
6181 sticky_name: "SOZUBALANCEID".to_owned(),
6182 front_timeout: 60,
6183 back_timeout: 30,
6184 connect_timeout: 3,
6185 request_timeout: 10,
6186 ..Default::default()
6187 }
6188 }
6189
6190 fn make_tcp_listener(address: SocketAddress) -> TcpListenerConfig {
6191 TcpListenerConfig {
6192 address,
6193 front_timeout: 60,
6194 back_timeout: 30,
6195 connect_timeout: 3,
6196 ..Default::default()
6197 }
6198 }
6199
6200 fn make_udp_listener(address: SocketAddress, active: bool) -> UdpListenerConfig {
6201 UdpListenerConfig {
6202 interface: None,
6203 address,
6204 public_address: None,
6205 front_timeout: 30,
6206 back_timeout: 30,
6207 max_rx_datagram_size: 1500,
6208 max_flows: 0,
6209 active,
6210 }
6211 }
6212
6213 #[test]
6223 fn test_udp_state_roundtrip() {
6224 let address = SocketAddress::new_v4(127, 0, 0, 1, 5353);
6225
6226 let mut state = ConfigState::default();
6227 state
6228 .dispatch(&RequestType::AddUdpListener(make_udp_listener(address, true)).into())
6229 .expect("could not add udp listener");
6230 state
6231 .dispatch(
6232 &RequestType::ActivateListener(ActivateListener {
6233 interface: None,
6234 address,
6235 proxy: ListenerType::Udp.into(),
6236 from_scm: false,
6237 })
6238 .into(),
6239 )
6240 .expect("could not activate udp listener");
6241 state
6242 .dispatch(
6243 &RequestType::AddUdpFrontend(RequestUdpFrontend {
6244 cluster_id: "udp_cluster".to_string(),
6245 address,
6246 tags: BTreeMap::from([("owner".to_string(), "team".to_string())]),
6247 })
6248 .into(),
6249 )
6250 .expect("could not add udp frontend");
6251
6252 assert_eq!(state.udp_listeners.len(), 1);
6253 assert!(state.udp_listeners[&address.into()].active);
6254 assert_eq!(
6255 state.udp_fronts.get("udp_cluster").map(Vec::len),
6256 Some(1usize)
6257 );
6258
6259 let logical = |s: &ConfigState| {
6264 let mut c = s.clone();
6265 c.request_counts.clear();
6266 c
6267 };
6268
6269 let mut replayed = ConfigState::default();
6271 for request in state.generate_requests() {
6272 replayed
6273 .dispatch(&request)
6274 .expect("could not replay generated request");
6275 }
6276 assert_eq!(
6277 logical(&state),
6278 logical(&replayed),
6279 "UDP listener + frontend must survive generate_requests → replay"
6280 );
6281
6282 let empty = ConfigState::default();
6284 let mut from_diff = ConfigState::default();
6285 for request in empty.diff(&state) {
6286 from_diff
6287 .dispatch(&request)
6288 .expect("could not replay diff request");
6289 }
6290 assert_eq!(
6291 logical(&state),
6292 logical(&from_diff),
6293 "diff(empty -> state) must reconstruct the UDP listener + frontend"
6294 );
6295
6296 let mut torn_down = state.clone();
6298 for request in state.diff(&empty) {
6299 torn_down
6300 .dispatch(&request)
6301 .expect("could not replay teardown diff request");
6302 }
6303 assert!(
6304 torn_down.udp_listeners.is_empty(),
6305 "diff(state -> empty) must remove the UDP listener"
6306 );
6307 assert!(
6308 torn_down
6309 .udp_fronts
6310 .get("udp_cluster")
6311 .map(Vec::is_empty)
6312 .unwrap_or(true),
6313 "diff(state -> empty) must remove the UDP frontend"
6314 );
6315
6316 state
6318 .dispatch(
6319 &RequestType::UpdateUdpListener(UpdateUdpListenerConfig {
6320 address,
6321 max_flows: Some(4096),
6322 front_timeout: Some(15),
6323 ..Default::default()
6324 })
6325 .into(),
6326 )
6327 .expect("could not update udp listener");
6328 let updated = &state.udp_listeners[&address.into()];
6329 assert_eq!(updated.max_flows, 4096);
6330 assert_eq!(updated.front_timeout, 15);
6331 assert_eq!(
6332 updated.back_timeout, 30,
6333 "unpatched field must be preserved"
6334 );
6335 }
6336
6337 #[test]
6352 fn test_tcp_sni_alpn_state_roundtrip() {
6353 let address = SocketAddress::new_v4(127, 0, 0, 1, 6443);
6354
6355 let mut state = ConfigState::default();
6356 state
6357 .dispatch(
6358 &RequestType::AddTcpListener(TcpListenerConfig {
6359 address,
6360 front_timeout: 60,
6361 back_timeout: 30,
6362 connect_timeout: 3,
6363 active: true,
6364 sni_preread_timeout: Some(2),
6365 sni_preread_max_bytes: Some(8192),
6366 ..Default::default()
6367 })
6368 .into(),
6369 )
6370 .expect("could not add tcp listener");
6371 state
6372 .dispatch(
6373 &RequestType::ActivateListener(ActivateListener {
6374 interface: None,
6375 address,
6376 proxy: ListenerType::Tcp.into(),
6377 from_scm: false,
6378 })
6379 .into(),
6380 )
6381 .expect("could not activate tcp listener");
6382
6383 let frontend_a = RequestTcpFrontend {
6384 cluster_id: "cluster_sni_a".to_string(),
6385 address,
6386 tags: BTreeMap::new(),
6387 sni: Some("example.com".to_string()),
6388 alpn: vec!["h2".to_string()],
6389 };
6390 let frontend_b = RequestTcpFrontend {
6391 cluster_id: "cluster_sni_b".to_string(),
6392 address,
6393 tags: BTreeMap::new(),
6394 sni: Some("other.example.com".to_string()),
6395 alpn: vec![],
6396 };
6397 state
6398 .dispatch(&RequestType::AddTcpFrontend(frontend_a.clone()).into())
6399 .expect("could not add sni-a tcp frontend");
6400 state
6401 .dispatch(&RequestType::AddTcpFrontend(frontend_b.clone()).into())
6402 .expect("could not add sni-b tcp frontend");
6403
6404 assert_eq!(state.tcp_listeners.len(), 1);
6405 assert!(state.tcp_listeners[&address.into()].active);
6406 assert_eq!(
6407 state.tcp_listeners[&address.into()].sni_preread_timeout,
6408 Some(2)
6409 );
6410 assert_eq!(
6411 state.tcp_listeners[&address.into()].sni_preread_max_bytes,
6412 Some(8192)
6413 );
6414 assert_eq!(state.count_tcp_frontends_raw(), 2);
6415
6416 let logical = |s: &ConfigState| {
6421 let mut c = s.clone();
6422 c.request_counts.clear();
6423 c
6424 };
6425
6426 let mut replayed = ConfigState::default();
6428 for request in state.generate_requests() {
6429 replayed
6430 .dispatch(&request)
6431 .expect("could not replay generated request");
6432 }
6433 assert_eq!(
6434 logical(&state),
6435 logical(&replayed),
6436 "SNI/ALPN TCP listener + frontends must survive generate_requests → replay"
6437 );
6438
6439 let empty = ConfigState::default();
6441 let mut from_diff = ConfigState::default();
6442 for request in empty.diff(&state) {
6443 from_diff
6444 .dispatch(&request)
6445 .expect("could not replay diff request");
6446 }
6447 assert_eq!(
6448 logical(&state),
6449 logical(&from_diff),
6450 "diff(empty -> state) must reconstruct the tcp listener + both sni frontends"
6451 );
6452
6453 let mut torn_down = state.clone();
6455 for request in state.diff(&empty) {
6456 torn_down
6457 .dispatch(&request)
6458 .expect("could not replay teardown diff request");
6459 }
6460 assert!(
6461 torn_down.tcp_listeners.is_empty(),
6462 "diff(state -> empty) must remove the tcp listener"
6463 );
6464 assert_eq!(
6465 torn_down.count_tcp_frontends_raw(),
6466 0,
6467 "diff(state -> empty) must remove both sni frontends"
6468 );
6469
6470 let mismatched_removal = state.dispatch(
6474 &RequestType::RemoveTcpFrontend(RequestTcpFrontend {
6475 cluster_id: "cluster_sni_a".to_string(),
6476 address,
6477 tags: BTreeMap::new(),
6478 sni: None,
6479 alpn: vec![],
6480 })
6481 .into(),
6482 );
6483 assert!(
6484 mismatched_removal.is_err(),
6485 "removing by address alone (no matching sni/alpn) must be rejected"
6486 );
6487 assert_eq!(
6488 state.count_tcp_frontends_raw(),
6489 2,
6490 "a rejected mismatched removal must not change either sibling"
6491 );
6492
6493 state
6496 .dispatch(&RequestType::RemoveTcpFrontend(frontend_a).into())
6497 .expect("could not remove sni-a tcp frontend");
6498 assert_eq!(state.count_tcp_frontends_raw(), 1);
6499 let remaining = state
6500 .tcp_fronts
6501 .get("cluster_sni_b")
6502 .expect("cluster_sni_b bucket must survive");
6503 assert_eq!(remaining.len(), 1);
6504 assert_eq!(remaining[0].sni, Some("other.example.com".to_string()));
6505 }
6506
6507 fn add_tcp_test_cluster(state: &mut ConfigState, cluster_id: &str) {
6518 state
6519 .dispatch(
6520 &RequestType::AddCluster(Cluster {
6521 cluster_id: cluster_id.to_string(),
6522 sticky_session: false,
6523 https_redirect: false,
6524 ..Default::default()
6525 })
6526 .into(),
6527 )
6528 .expect("could not add cluster");
6529 }
6530
6531 #[test]
6536 fn add_tcp_frontend_rejects_cross_cluster_duplicate_identity() {
6537 let address = SocketAddress::new_v4(127, 0, 0, 1, 9001);
6538 let mut state = ConfigState::new();
6539 add_tcp_test_cluster(&mut state, "cluster_x");
6540 add_tcp_test_cluster(&mut state, "cluster_y");
6541
6542 state
6543 .dispatch(
6544 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6545 cluster_id: "cluster_x".to_string(),
6546 address,
6547 tags: BTreeMap::new(),
6548 sni: Some("example.com".to_string()),
6549 alpn: vec!["h2".to_string()],
6550 })
6551 .into(),
6552 )
6553 .expect("could not add cluster_x tcp frontend");
6554
6555 let hash_before = state.hash_state();
6556 let count_before = state.count_tcp_frontends_raw();
6557
6558 let result = state.dispatch(
6559 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6560 cluster_id: "cluster_y".to_string(),
6561 address,
6562 tags: BTreeMap::new(),
6563 sni: Some("example.com".to_string()),
6564 alpn: vec!["h2".to_string()],
6565 })
6566 .into(),
6567 );
6568 assert!(
6569 result.is_err(),
6570 "the same (address, sni, alpn) under a different cluster_id must be rejected, \
6571 got: {result:?}"
6572 );
6573 assert_eq!(
6574 state.hash_state(),
6575 hash_before,
6576 "a rejected cross-cluster duplicate must not change the state hash"
6577 );
6578 assert_eq!(
6579 state.count_tcp_frontends_raw(),
6580 count_before,
6581 "a rejected cross-cluster duplicate must not change the frontend count"
6582 );
6583 assert_eq!(state.tcp_fronts.get("cluster_x").map(Vec::len), Some(1));
6584 assert_eq!(
6585 state.tcp_fronts.get("cluster_y").map(Vec::len).unwrap_or(0),
6586 0
6587 );
6588 }
6589
6590 #[test]
6593 fn add_tcp_frontend_rejects_reordered_alpn_duplicate() {
6594 let address = SocketAddress::new_v4(127, 0, 0, 1, 9002);
6595 let mut state = ConfigState::new();
6596 add_tcp_test_cluster(&mut state, "cluster_reorder");
6597
6598 state
6599 .dispatch(
6600 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6601 cluster_id: "cluster_reorder".to_string(),
6602 address,
6603 tags: BTreeMap::new(),
6604 sni: Some("example.com".to_string()),
6605 alpn: vec!["h2".to_string(), "http/1.1".to_string()],
6606 })
6607 .into(),
6608 )
6609 .expect("could not add first tcp frontend");
6610
6611 let hash_before = state.hash_state();
6612 let count_before = state.count_tcp_frontends_raw();
6613
6614 let result = state.dispatch(
6615 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6616 cluster_id: "cluster_reorder".to_string(),
6617 address,
6618 tags: BTreeMap::new(),
6619 sni: Some("example.com".to_string()),
6620 alpn: vec!["http/1.1".to_string(), "h2".to_string()],
6621 })
6622 .into(),
6623 );
6624 assert!(
6625 result.is_err(),
6626 "the same ALPN set in a different order must be rejected as a duplicate \
6627 identity, got: {result:?}"
6628 );
6629 assert_eq!(state.hash_state(), hash_before);
6630 assert_eq!(state.count_tcp_frontends_raw(), count_before);
6631 assert_eq!(
6632 state.tcp_fronts.get("cluster_reorder").map(Vec::len),
6633 Some(1)
6634 );
6635 }
6636
6637 #[test]
6641 fn remove_tcp_frontend_matches_reordered_alpn() {
6642 let address = SocketAddress::new_v4(127, 0, 0, 1, 9003);
6643 let mut state = ConfigState::new();
6644 add_tcp_test_cluster(&mut state, "cluster_remove_reorder");
6645
6646 state
6647 .dispatch(
6648 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6649 cluster_id: "cluster_remove_reorder".to_string(),
6650 address,
6651 tags: BTreeMap::new(),
6652 sni: Some("example.com".to_string()),
6653 alpn: vec!["h2".to_string(), "http/1.1".to_string()],
6654 })
6655 .into(),
6656 )
6657 .expect("could not add tcp frontend");
6658
6659 let result = state.dispatch(
6660 &RequestType::RemoveTcpFrontend(RequestTcpFrontend {
6661 cluster_id: "cluster_remove_reorder".to_string(),
6662 address,
6663 tags: BTreeMap::new(),
6664 sni: Some("example.com".to_string()),
6665 alpn: vec!["http/1.1".to_string(), "h2".to_string()],
6666 })
6667 .into(),
6668 );
6669 assert!(
6670 result.is_ok(),
6671 "removing with the same ALPN set in a different order must succeed, got: {result:?}"
6672 );
6673 assert_eq!(state.count_tcp_frontends_raw(), 0);
6674 }
6675
6676 #[test]
6680 fn add_tcp_frontend_rejects_alpn_overlap_across_clusters_same_sni() {
6681 let address = SocketAddress::new_v4(127, 0, 0, 1, 9004);
6682 let mut state = ConfigState::new();
6683 add_tcp_test_cluster(&mut state, "cluster_overlap_a");
6684 add_tcp_test_cluster(&mut state, "cluster_overlap_b");
6685
6686 state
6687 .dispatch(
6688 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6689 cluster_id: "cluster_overlap_a".to_string(),
6690 address,
6691 tags: BTreeMap::new(),
6692 sni: Some("example.com".to_string()),
6693 alpn: vec!["h2".to_string()],
6694 })
6695 .into(),
6696 )
6697 .expect("could not add cluster_overlap_a tcp frontend");
6698
6699 let hash_before = state.hash_state();
6700 let count_before = state.count_tcp_frontends_raw();
6701
6702 let result = state.dispatch(
6703 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6704 cluster_id: "cluster_overlap_b".to_string(),
6705 address,
6706 tags: BTreeMap::new(),
6707 sni: Some("example.com".to_string()),
6708 alpn: vec!["h2".to_string(), "http/1.1".to_string()],
6709 })
6710 .into(),
6711 );
6712 assert!(
6713 result.is_err(),
6714 "an overlapping ALPN set on the same (address, sni) under a different cluster_id \
6715 must be rejected, got: {result:?}"
6716 );
6717 assert_eq!(state.hash_state(), hash_before);
6718 assert_eq!(state.count_tcp_frontends_raw(), count_before);
6719 assert_eq!(
6720 state
6721 .tcp_fronts
6722 .get("cluster_overlap_b")
6723 .map(Vec::len)
6724 .unwrap_or(0),
6725 0
6726 );
6727 }
6728
6729 #[test]
6732 fn add_tcp_frontend_rejects_no_sni_sni_mixing_both_directions() {
6733 let address_a = SocketAddress::new_v4(127, 0, 0, 1, 9005);
6734 let address_b = SocketAddress::new_v4(127, 0, 0, 1, 9006);
6735 let mut state = ConfigState::new();
6736 add_tcp_test_cluster(&mut state, "cluster_mix_1");
6737 add_tcp_test_cluster(&mut state, "cluster_mix_2");
6738 add_tcp_test_cluster(&mut state, "cluster_mix_3");
6739 add_tcp_test_cluster(&mut state, "cluster_mix_4");
6740
6741 state
6744 .dispatch(
6745 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6746 cluster_id: "cluster_mix_1".to_string(),
6747 address: address_a,
6748 tags: BTreeMap::new(),
6749 sni: None,
6750 alpn: vec![],
6751 })
6752 .into(),
6753 )
6754 .expect("could not add no-sni tcp frontend");
6755 let hash_before_a = state.hash_state();
6756 let count_before_a = state.count_tcp_frontends_raw();
6757 let result_a = state.dispatch(
6758 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6759 cluster_id: "cluster_mix_2".to_string(),
6760 address: address_a,
6761 tags: BTreeMap::new(),
6762 sni: Some("example.com".to_string()),
6763 alpn: vec![],
6764 })
6765 .into(),
6766 );
6767 assert!(
6768 result_a.is_err(),
6769 "an SNI-scoped frontend must not be added to a listener with an existing no-SNI \
6770 front, got: {result_a:?}"
6771 );
6772 assert_eq!(state.hash_state(), hash_before_a);
6773 assert_eq!(state.count_tcp_frontends_raw(), count_before_a);
6774
6775 state
6778 .dispatch(
6779 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6780 cluster_id: "cluster_mix_3".to_string(),
6781 address: address_b,
6782 tags: BTreeMap::new(),
6783 sni: Some("example.com".to_string()),
6784 alpn: vec![],
6785 })
6786 .into(),
6787 )
6788 .expect("could not add sni-scoped tcp frontend");
6789 let hash_before_b = state.hash_state();
6790 let count_before_b = state.count_tcp_frontends_raw();
6791 let result_b = state.dispatch(
6792 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6793 cluster_id: "cluster_mix_4".to_string(),
6794 address: address_b,
6795 tags: BTreeMap::new(),
6796 sni: None,
6797 alpn: vec![],
6798 })
6799 .into(),
6800 );
6801 assert!(
6802 result_b.is_err(),
6803 "a no-SNI frontend must not be added to a listener with existing SNI-scoped \
6804 fronts, got: {result_b:?}"
6805 );
6806 assert_eq!(state.hash_state(), hash_before_b);
6807 assert_eq!(state.count_tcp_frontends_raw(), count_before_b);
6808 }
6809
6810 #[test]
6815 fn add_tcp_frontend_rejects_malformed_sni_shape() {
6816 let address = SocketAddress::new_v4(127, 0, 0, 1, 9007);
6817 let mut state = ConfigState::new();
6818 add_tcp_test_cluster(&mut state, "cluster_malformed");
6819
6820 let hash_before = state.hash_state();
6821 let count_before = state.count_tcp_frontends_raw();
6822
6823 let result = state.dispatch(
6824 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6825 cluster_id: "cluster_malformed".to_string(),
6826 address,
6827 tags: BTreeMap::new(),
6828 sni: Some(".example.com".to_string()),
6830 alpn: vec![],
6831 })
6832 .into(),
6833 );
6834 assert!(
6835 result.is_err(),
6836 "a malformed SNI pattern must be rejected by the master, not merely by \
6837 config-load, got: {result:?}"
6838 );
6839 assert_eq!(state.hash_state(), hash_before);
6840 assert_eq!(state.count_tcp_frontends_raw(), count_before);
6841 assert_eq!(
6842 state
6843 .tcp_fronts
6844 .get("cluster_malformed")
6845 .map(Vec::len)
6846 .unwrap_or(0),
6847 0
6848 );
6849 }
6850
6851 #[test]
6857 fn add_tcp_frontend_rejects_alpn_without_sni() {
6858 let address = SocketAddress::new_v4(127, 0, 0, 1, 9009);
6859 let mut state = ConfigState::new();
6860 add_tcp_test_cluster(&mut state, "cluster_alpn_no_sni");
6861
6862 let hash_before = state.hash_state();
6863 let count_before = state.count_tcp_frontends_raw();
6864
6865 let result = state.dispatch(
6866 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6867 cluster_id: "cluster_alpn_no_sni".to_string(),
6868 address,
6869 tags: BTreeMap::new(),
6870 sni: None,
6871 alpn: vec!["h2".to_string()],
6872 })
6873 .into(),
6874 );
6875 assert!(
6876 result.is_err(),
6877 "a non-empty alpn without sni must be rejected by the master, got: {result:?}"
6878 );
6879 assert_eq!(
6880 state.hash_state(),
6881 hash_before,
6882 "a rejected alpn-without-sni add must not change the state hash"
6883 );
6884 assert_eq!(
6885 state.count_tcp_frontends_raw(),
6886 count_before,
6887 "a rejected alpn-without-sni add must not change the frontend count"
6888 );
6889 assert_eq!(
6890 state
6891 .tcp_fronts
6892 .get("cluster_alpn_no_sni")
6893 .map(Vec::len)
6894 .unwrap_or(0),
6895 0
6896 );
6897 }
6898
6899 #[test]
6905 fn add_tcp_frontend_accepts_exact_and_wildcard_catch_all_siblings() {
6906 let address = SocketAddress::new_v4(127, 0, 0, 1, 9008);
6907 let mut state = ConfigState::new();
6908 add_tcp_test_cluster(&mut state, "cluster_exact_catchall");
6909 add_tcp_test_cluster(&mut state, "cluster_wildcard_catchall");
6910
6911 state
6912 .dispatch(
6913 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6914 cluster_id: "cluster_exact_catchall".to_string(),
6915 address,
6916 tags: BTreeMap::new(),
6917 sni: Some("example.com".to_string()),
6918 alpn: vec![],
6919 })
6920 .into(),
6921 )
6922 .expect("could not add exact catch-all tcp frontend");
6923
6924 let result = state.dispatch(
6925 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6926 cluster_id: "cluster_wildcard_catchall".to_string(),
6927 address,
6928 tags: BTreeMap::new(),
6929 sni: Some("*.example.com".to_string()),
6930 alpn: vec![],
6931 })
6932 .into(),
6933 );
6934 assert!(
6935 result.is_ok(),
6936 "an exact catch-all and a wildcard catch-all on the same address must both be \
6937 accepted, got: {result:?}"
6938 );
6939 assert_eq!(state.count_tcp_frontends_raw(), 2);
6940 }
6941
6942 #[test]
6946 fn list_frontends_includes_udp() {
6947 let tcp_addr = SocketAddress::new_v4(0, 0, 0, 0, 6379);
6948 let udp_addr = SocketAddress::new_v4(0, 0, 0, 0, 5353);
6949
6950 let mut state = ConfigState::default();
6951 state
6952 .dispatch(
6953 &RequestType::AddTcpFrontend(RequestTcpFrontend {
6954 cluster_id: "tcp_cluster".to_string(),
6955 address: tcp_addr,
6956 ..Default::default()
6957 })
6958 .into(),
6959 )
6960 .expect("could not add tcp frontend");
6961 state
6962 .dispatch(
6963 &RequestType::AddUdpFrontend(RequestUdpFrontend {
6964 cluster_id: "udp_cluster".to_string(),
6965 address: udp_addr,
6966 ..Default::default()
6967 })
6968 .into(),
6969 )
6970 .expect("could not add udp frontend");
6971
6972 let all = state.list_frontends(FrontendFilters::default());
6974 assert_eq!(all.tcp_frontends.len(), 1, "tcp frontend must be listed");
6975 assert_eq!(
6976 all.udp_frontends.len(),
6977 1,
6978 "udp frontend must be listed under the default all-pass path"
6979 );
6980 assert_eq!(all.udp_frontends[0].cluster_id, "udp_cluster");
6981 assert_eq!(all.udp_frontends[0].address, udp_addr);
6982
6983 let tcp_only = state.list_frontends(FrontendFilters {
6985 tcp: true,
6986 ..Default::default()
6987 });
6988 assert_eq!(tcp_only.tcp_frontends.len(), 1);
6989 assert_eq!(
6990 tcp_only.udp_frontends.len(),
6991 1,
6992 "udp frontends ride the tcp filter"
6993 );
6994
6995 let http_only = state.list_frontends(FrontendFilters {
6997 http: true,
6998 ..Default::default()
6999 });
7000 assert!(http_only.tcp_frontends.is_empty());
7001 assert!(http_only.udp_frontends.is_empty());
7002
7003 let domain_filtered = state.list_frontends(FrontendFilters {
7005 domain: Some("example.com".to_string()),
7006 ..Default::default()
7007 });
7008 assert!(domain_filtered.tcp_frontends.is_empty());
7009 assert!(domain_filtered.udp_frontends.is_empty());
7010 }
7011
7012 #[test]
7016 fn list_frontends_filters_by_cluster_id() {
7017 let http_addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
7018 let https_addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7019 let mut state = ConfigState::default();
7020 for (cluster_id, hostname) in [
7021 (Some(String::from("wanted")), "wanted.example.com"),
7022 (Some(String::from("other")), "other.example.com"),
7023 (None, "denied.example.com"),
7025 ] {
7026 state
7027 .dispatch(
7028 &RequestType::AddHttpFrontend(RequestHttpFrontend {
7029 cluster_id: cluster_id.clone(),
7030 hostname: hostname.to_string(),
7031 path: PathRule::prefix(String::from("/")),
7032 address: http_addr,
7033 position: RulePosition::Tree.into(),
7034 ..Default::default()
7035 })
7036 .into(),
7037 )
7038 .expect("could not add http frontend");
7039 state
7040 .dispatch(
7041 &RequestType::AddHttpsFrontend(RequestHttpFrontend {
7042 cluster_id,
7043 hostname: hostname.to_string(),
7044 path: PathRule::prefix(String::from("/")),
7045 address: https_addr,
7046 position: RulePosition::Tree.into(),
7047 ..Default::default()
7048 })
7049 .into(),
7050 )
7051 .expect("could not add https frontend");
7052 }
7053 for (cluster_id, port) in [("wanted", 6379u16), ("other", 6380)] {
7055 state
7056 .dispatch(
7057 &RequestType::AddTcpFrontend(RequestTcpFrontend {
7058 cluster_id: cluster_id.to_string(),
7059 address: SocketAddress::new_v4(0, 0, 0, 0, port),
7060 ..Default::default()
7061 })
7062 .into(),
7063 )
7064 .expect("could not add tcp frontend");
7065 state
7066 .dispatch(
7067 &RequestType::AddUdpFrontend(RequestUdpFrontend {
7068 cluster_id: cluster_id.to_string(),
7069 address: SocketAddress::new_v4(0, 0, 0, 0, port + 1000),
7070 ..Default::default()
7071 })
7072 .into(),
7073 )
7074 .expect("could not add udp frontend");
7075 }
7076
7077 let all = state.list_frontends(FrontendFilters::default());
7079 assert_eq!(all.http_frontends.len(), 3);
7080 assert_eq!(all.https_frontends.len(), 3);
7081 assert_eq!(all.tcp_frontends.len(), 2);
7082 assert_eq!(all.udp_frontends.len(), 2);
7083
7084 let wanted = state.list_frontends(FrontendFilters {
7087 cluster_id: Some("wanted".to_string()),
7088 ..Default::default()
7089 });
7090 assert_eq!(wanted.http_frontends.len(), 1);
7091 assert_eq!(wanted.https_frontends.len(), 1);
7092 assert_eq!(wanted.tcp_frontends.len(), 1);
7093 assert_eq!(wanted.udp_frontends.len(), 1);
7094 assert_eq!(
7095 wanted.http_frontends[0].cluster_id.as_deref(),
7096 Some("wanted")
7097 );
7098 assert_eq!(wanted.tcp_frontends[0].cluster_id, "wanted");
7099 assert_eq!(wanted.udp_frontends[0].cluster_id, "wanted");
7100 assert!(
7101 wanted
7102 .http_frontends
7103 .iter()
7104 .chain(wanted.https_frontends.iter())
7105 .all(|f| f.hostname == "wanted.example.com"),
7106 "no frontend of another cluster may leak through the filter"
7107 );
7108
7109 let unknown = state.list_frontends(FrontendFilters {
7111 cluster_id: Some("nope".to_string()),
7112 ..Default::default()
7113 });
7114 assert!(unknown.http_frontends.is_empty());
7115 assert!(unknown.https_frontends.is_empty());
7116 assert!(unknown.tcp_frontends.is_empty());
7117 assert!(unknown.udp_frontends.is_empty());
7118
7119 let conflicting = state.list_frontends(FrontendFilters {
7121 cluster_id: Some("wanted".to_string()),
7122 domain: Some("other.example.com".to_string()),
7123 ..Default::default()
7124 });
7125 assert!(conflicting.http_frontends.is_empty());
7126 assert!(conflicting.https_frontends.is_empty());
7127
7128 let https_only = state.list_frontends(FrontendFilters {
7130 https: true,
7131 cluster_id: Some("wanted".to_string()),
7132 ..Default::default()
7133 });
7134 assert!(https_only.http_frontends.is_empty());
7135 assert_eq!(https_only.https_frontends.len(), 1);
7136 assert!(https_only.tcp_frontends.is_empty());
7137 assert!(https_only.udp_frontends.is_empty());
7138 }
7139
7140 #[test]
7145 fn update_https_listener_happy_path_h2_knobs() {
7146 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7147 let mut state = ConfigState::new();
7148 state
7149 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
7150 .unwrap();
7151
7152 let patch = UpdateHttpsListenerConfig {
7153 address: addr,
7154 h2_max_rst_stream_per_window: Some(50),
7155 h2_max_ping_per_window: Some(20),
7156 ..Default::default()
7157 };
7158 state
7159 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7160 .expect("update must succeed");
7161
7162 let listener = state
7163 .https_listeners
7164 .get(&ListenerKey::from(addr))
7165 .expect("listener must be present");
7166 assert_eq!(listener.h2_max_rst_stream_per_window, Some(50));
7167 assert_eq!(listener.h2_max_ping_per_window, Some(20));
7168 assert_eq!(listener.front_timeout, 60);
7170 assert_eq!(listener.h2_max_settings_per_window, None);
7171 }
7172
7173 #[test]
7176 fn update_https_listener_not_found() {
7177 let mut state = ConfigState::new();
7178 let patch = UpdateHttpsListenerConfig {
7179 address: SocketAddress::new_v4(1, 2, 3, 4, 9999),
7180 h2_max_rst_stream_per_window: Some(50),
7181 ..Default::default()
7182 };
7183 let err = state
7184 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7185 .unwrap_err();
7186 assert!(
7187 matches!(
7188 err,
7189 StateError::NotFound {
7190 kind: ObjectKind::HttpsListener,
7191 ..
7192 }
7193 ),
7194 "expected NotFound, got: {err}"
7195 );
7196 }
7197
7198 #[test]
7201 fn update_https_listener_noop_patch() {
7202 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7203 let mut state = ConfigState::new();
7204 let original = make_https_listener(addr);
7205 state
7206 .dispatch(&RequestType::AddHttpsListener(original.clone()).into())
7207 .unwrap();
7208
7209 let patch = UpdateHttpsListenerConfig {
7210 address: addr,
7211 ..Default::default()
7212 };
7213 state
7214 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7215 .expect("no-op patch must succeed");
7216
7217 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
7218 assert_eq!(listener.front_timeout, original.front_timeout);
7219 assert_eq!(
7220 listener.h2_max_rst_stream_per_window,
7221 original.h2_max_rst_stream_per_window
7222 );
7223 }
7224
7225 #[test]
7227 fn update_https_listener_invalid_value_flood_knob_zero() {
7228 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7229 let mut state = ConfigState::new();
7230 state
7231 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
7232 .unwrap();
7233
7234 let patch = UpdateHttpsListenerConfig {
7235 address: addr,
7236 h2_max_rst_stream_per_window: Some(0),
7237 ..Default::default()
7238 };
7239 let err = state
7240 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7241 .unwrap_err();
7242 assert!(
7243 matches!(
7244 err,
7245 StateError::InvalidValue {
7246 field: "h2_max_rst_stream_per_window",
7247 ..
7248 }
7249 ),
7250 "expected InvalidValue for flood knob 0, got: {err}"
7251 );
7252 }
7253
7254 #[test]
7260 fn a_tcp_frontend_and_an_affinity_key_never_share_a_cluster() {
7261 use crate::proto::command::RequestTcpFrontend;
7262
7263 let keyed = |cluster_id: &str| -> Request {
7264 RequestType::AddCluster(Cluster {
7265 cluster_id: cluster_id.to_owned(),
7266 affinity_header: Some(String::from("X-Tenant")),
7267 ..Default::default()
7268 })
7269 .into()
7270 };
7271 let tcp_front = |cluster_id: &str, port: u16| -> Request {
7272 RequestType::AddTcpFrontend(RequestTcpFrontend {
7273 cluster_id: cluster_id.to_owned(),
7274 address: SocketAddress::new_v4(127, 0, 0, 1, port),
7275 ..Default::default()
7276 })
7277 .into()
7278 };
7279
7280 let mut state = ConfigState::new();
7282 state
7283 .dispatch(&keyed("keyed"))
7284 .expect("a keyed cluster is valid");
7285 let err = state
7286 .dispatch(&tcp_front("keyed", 4000))
7287 .expect_err("a TCP frontend may not join a keyed cluster");
7288 assert!(
7289 matches!(err, StateError::InvalidTcpFrontend { .. }),
7290 "unexpected error: {err:?}"
7291 );
7292 assert!(state.tcp_fronts.get("keyed").is_none_or(Vec::is_empty));
7293
7294 let mut state = ConfigState::new();
7296 state
7297 .dispatch(
7298 &RequestType::AddCluster(Cluster {
7299 cluster_id: String::from("tcp"),
7300 ..Default::default()
7301 })
7302 .into(),
7303 )
7304 .expect("a plain cluster is valid");
7305 state
7306 .dispatch(&tcp_front("tcp", 4001))
7307 .expect("a TCP frontend may join a plain cluster");
7308 let err = state
7309 .dispatch(&keyed("tcp"))
7310 .expect_err("a cluster serving TCP may not name a key");
7311 assert!(
7312 matches!(
7313 err,
7314 StateError::InvalidValue {
7315 field: "affinity_key",
7316 ..
7317 }
7318 ),
7319 "unexpected error: {err:?}"
7320 );
7321 assert_eq!(state.clusters["tcp"].affinity_header, None);
7322 }
7323
7324 #[test]
7328 fn add_cluster_with_out_of_range_max_connection_attempts_rejected() {
7329 for attempts in [0, 256] {
7330 let mut state = ConfigState::new();
7331 let err = state
7332 .dispatch(
7333 &RequestType::AddCluster(Cluster {
7334 cluster_id: String::from("budget"),
7335 max_connection_attempts: Some(attempts),
7336 ..Default::default()
7337 })
7338 .into(),
7339 )
7340 .expect_err("an out-of-range budget must be refused");
7341 assert!(
7342 matches!(
7343 err,
7344 StateError::InvalidValue {
7345 field: "max_connection_attempts",
7346 ..
7347 }
7348 ),
7349 "{attempts}: unexpected error {err:?}"
7350 );
7351 assert!(state.clusters.is_empty());
7352 }
7353 let mut state = ConfigState::new();
7354 state
7355 .dispatch(
7356 &RequestType::AddCluster(Cluster {
7357 cluster_id: String::from("budget"),
7358 max_connection_attempts: Some(7),
7359 ..Default::default()
7360 })
7361 .into(),
7362 )
7363 .expect("a budget in range is accepted");
7364 assert_eq!(state.clusters["budget"].max_connection_attempts, Some(7));
7365 }
7366
7367 #[test]
7368 fn add_cluster_with_invalid_shuffle_sharding_rejected() {
7369 for (percent, minimum, mode) in [
7370 (Some(0), None, None),
7371 (Some(101), None, None),
7372 (Some(50), Some(1), None),
7373 (None, Some(8), None),
7374 (None, None, Some(1)),
7375 (Some(50), None, Some(7)),
7376 ] {
7377 let mut state = ConfigState::new();
7378 let err = state
7379 .dispatch(
7380 &RequestType::AddCluster(Cluster {
7381 cluster_id: String::from("sharded"),
7382 shard_percent: percent,
7383 shard_min_backends: minimum,
7384 shard_mode: mode,
7385 ..Default::default()
7386 })
7387 .into(),
7388 )
7389 .expect_err("invalid shuffle sharding must be refused");
7390 assert!(
7391 matches!(
7392 err,
7393 StateError::InvalidValue {
7394 field: "shuffle_sharding",
7395 ..
7396 }
7397 ),
7398 "({percent:?}, {minimum:?}, {mode:?}): unexpected error {err:?}"
7399 );
7400 assert!(state.clusters.is_empty());
7401 }
7402 let mut state = ConfigState::new();
7403 state
7404 .dispatch(
7405 &RequestType::AddCluster(Cluster {
7406 cluster_id: String::from("sharded"),
7407 shard_percent: Some(25),
7408 shard_min_backends: Some(4),
7409 shard_mode: Some(1),
7410 ..Default::default()
7411 })
7412 .into(),
7413 )
7414 .expect("a valid sharded cluster is accepted");
7415 }
7416
7417 #[test]
7418 fn add_cluster_with_two_affinity_key_sources_rejected() {
7419 let mut state = ConfigState::new();
7420 let err = state
7421 .dispatch(
7422 &RequestType::AddCluster(Cluster {
7423 cluster_id: String::from("keyed"),
7424 affinity_header: Some(String::from("X-Tenant")),
7425 affinity_cookie: Some(String::from("tenant")),
7426 ..Default::default()
7427 })
7428 .into(),
7429 )
7430 .expect_err("a cluster keyed on both a header and a cookie must be refused");
7431 assert!(
7432 matches!(
7433 err,
7434 StateError::InvalidValue {
7435 field: "affinity_key",
7436 ..
7437 }
7438 ),
7439 "unexpected error: {err:?}"
7440 );
7441 assert!(
7442 state.clusters.is_empty(),
7443 "a refused cluster must not be stored"
7444 );
7445 }
7446
7447 #[test]
7448 fn add_cluster_invalid_health_check_uri_rejected() {
7449 use crate::proto::command::HealthCheckConfig;
7450
7451 let mut state = ConfigState::new();
7452 let err = state
7453 .dispatch(
7454 &RequestType::AddCluster(Cluster {
7455 cluster_id: String::from("evil_cluster"),
7456 health_check: Some(HealthCheckConfig {
7457 uri: String::from("/foo\r\nGET /admin"),
7458 interval: 5_000,
7459 timeout: 1_000,
7460 healthy_threshold: 2,
7461 unhealthy_threshold: 2,
7462 ..Default::default()
7463 }),
7464 ..Default::default()
7465 })
7466 .into(),
7467 )
7468 .unwrap_err();
7469
7470 assert!(
7471 matches!(
7472 err,
7473 StateError::InvalidValue {
7474 field: "health_check",
7475 ..
7476 }
7477 ),
7478 "expected InvalidValue for CRLF-bearing health-check URI, got: {err}"
7479 );
7480 assert!(
7481 !state.clusters.contains_key("evil_cluster"),
7482 "cluster must not be inserted when health_check fails validation",
7483 );
7484 }
7485
7486 #[test]
7490 fn set_health_check_tcp_mode_with_status_field_rejected() {
7491 use crate::proto::command::{HealthCheckConfig, HealthCheckMode, SetHealthCheck};
7492
7493 let mut state = ConfigState::new();
7494 state
7495 .dispatch(
7496 &RequestType::AddCluster(Cluster {
7497 cluster_id: String::from("app"),
7498 ..Default::default()
7499 })
7500 .into(),
7501 )
7502 .expect("the cluster is added");
7503 let tcp = HealthCheckConfig {
7504 uri: String::new(),
7505 interval: 10,
7506 timeout: 5,
7507 healthy_threshold: 3,
7508 unhealthy_threshold: 3,
7509 mode: HealthCheckMode::Tcp as i32,
7510 ..Default::default()
7511 };
7512 let set = |config: HealthCheckConfig| {
7513 RequestType::SetHealthCheck(SetHealthCheck {
7514 cluster_id: String::from("app"),
7515 config,
7516 })
7517 .into()
7518 };
7519
7520 let err = state
7521 .dispatch(&set(HealthCheckConfig {
7522 expected_status: 200,
7523 ..tcp.clone()
7524 }))
7525 .unwrap_err();
7526 assert!(
7527 matches!(
7528 err,
7529 StateError::InvalidValue {
7530 field: "health_check",
7531 ..
7532 }
7533 ),
7534 "expected InvalidValue for a TCP probe with expected_status, got: {err}"
7535 );
7536 assert_eq!(state.clusters["app"].health_check, None);
7537
7538 state
7539 .dispatch(&set(tcp.clone()))
7540 .expect("a bare TCP probe is accepted");
7541 assert_eq!(state.clusters["app"].health_check, Some(tcp));
7542 }
7543
7544 #[test]
7546 fn update_https_listener_alpn_unknown_value_rejected() {
7547 use crate::proto::command::AlpnProtocols;
7548
7549 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7550 let mut state = ConfigState::new();
7551 state
7552 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
7553 .unwrap();
7554
7555 let patch = UpdateHttpsListenerConfig {
7556 address: addr,
7557 alpn_protocols: Some(AlpnProtocols {
7558 values: vec!["h3".to_owned()],
7559 }),
7560 ..Default::default()
7561 };
7562 let err = state
7563 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7564 .unwrap_err();
7565 assert!(
7566 matches!(
7567 err,
7568 StateError::InvalidValue {
7569 field: "alpn_protocols",
7570 ..
7571 }
7572 ),
7573 "expected InvalidValue for unknown ALPN, got: {err}"
7574 );
7575 }
7576
7577 #[test]
7579 fn update_https_listener_alpn_empty_reset_accepted() {
7580 use crate::proto::command::AlpnProtocols;
7581
7582 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7583 let mut state = ConfigState::new();
7584 let mut listener = make_https_listener(addr);
7585 listener.alpn_protocols = vec!["h2".to_owned()];
7586 state
7587 .dispatch(&RequestType::AddHttpsListener(listener).into())
7588 .unwrap();
7589
7590 let patch = UpdateHttpsListenerConfig {
7591 address: addr,
7592 alpn_protocols: Some(AlpnProtocols { values: vec![] }),
7593 ..Default::default()
7594 };
7595 state
7596 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7597 .expect("empty ALPN reset must succeed");
7598
7599 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
7600 assert!(
7601 listener.alpn_protocols.is_empty(),
7602 "ALPN must have been reset to empty"
7603 );
7604 }
7605
7606 #[test]
7608 fn update_https_listener_alpn_valid_values_accepted() {
7609 use crate::proto::command::AlpnProtocols;
7610
7611 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7612 let mut state = ConfigState::new();
7613 state
7614 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
7615 .unwrap();
7616
7617 let patch = UpdateHttpsListenerConfig {
7618 address: addr,
7619 alpn_protocols: Some(AlpnProtocols {
7620 values: vec!["h2".to_owned(), "http/1.1".to_owned()],
7621 }),
7622 ..Default::default()
7623 };
7624 state
7625 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7626 .expect("valid ALPN must be accepted");
7627
7628 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
7629 assert_eq!(listener.alpn_protocols, vec!["h2", "http/1.1"]);
7630 }
7631
7632 #[test]
7635 fn update_https_listener_alpn_absent_preserves_existing() {
7636 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7637 let mut state = ConfigState::new();
7638 let mut listener = make_https_listener(addr);
7639 listener.alpn_protocols = vec!["h2".to_owned()];
7640 state
7641 .dispatch(&RequestType::AddHttpsListener(listener).into())
7642 .unwrap();
7643
7644 let patch = UpdateHttpsListenerConfig {
7646 address: addr,
7647 front_timeout: Some(10),
7648 ..Default::default()
7649 };
7650 state
7651 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7652 .unwrap();
7653
7654 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
7655 assert_eq!(
7656 listener.alpn_protocols,
7657 vec!["h2"],
7658 "ALPN must be preserved when not patched"
7659 );
7660 }
7661
7662 #[test]
7664 fn update_https_listener_sozu_id_header_empty_rejected() {
7665 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7666 let mut state = ConfigState::new();
7667 state
7668 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
7669 .unwrap();
7670
7671 let patch = UpdateHttpsListenerConfig {
7672 address: addr,
7673 sozu_id_header: Some(String::new()),
7674 ..Default::default()
7675 };
7676 let err = state
7677 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7678 .unwrap_err();
7679 assert!(
7680 matches!(
7681 err,
7682 StateError::InvalidValue {
7683 field: "sozu_id_header",
7684 ..
7685 }
7686 ),
7687 "expected InvalidValue for empty header name, got: {err}"
7688 );
7689 }
7690
7691 #[test]
7693 fn update_https_listener_sozu_id_header_colon_rejected() {
7694 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7695 let mut state = ConfigState::new();
7696 state
7697 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
7698 .unwrap();
7699
7700 let patch = UpdateHttpsListenerConfig {
7701 address: addr,
7702 sozu_id_header: Some("bad: value".to_owned()),
7703 ..Default::default()
7704 };
7705 let err = state
7706 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7707 .unwrap_err();
7708 assert!(
7709 matches!(
7710 err,
7711 StateError::InvalidValue {
7712 field: "sozu_id_header",
7713 ..
7714 }
7715 ),
7716 "expected InvalidValue for header name with colon, got: {err}"
7717 );
7718 }
7719
7720 #[test]
7722 fn update_https_listener_sozu_id_header_valid_accepted() {
7723 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7724 let mut state = ConfigState::new();
7725 state
7726 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
7727 .unwrap();
7728
7729 let patch = UpdateHttpsListenerConfig {
7730 address: addr,
7731 sozu_id_header: Some("X-Edge-Id".to_owned()),
7732 ..Default::default()
7733 };
7734 state
7735 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7736 .expect("valid header name must be accepted");
7737
7738 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
7739 assert_eq!(listener.sozu_id_header.as_deref(), Some("X-Edge-Id"));
7740 }
7741
7742 #[test]
7748 fn add_listener_rejects_an_unknown_forwarded_headers_value() {
7749 let http = SocketAddress::new_v4(0, 0, 0, 0, 8080);
7750 let https = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7751 let mut state = ConfigState::new();
7752
7753 let http_error = state
7754 .dispatch(
7755 &RequestType::AddHttpListener(HttpListenerConfig {
7756 forwarded_headers: Some(42),
7757 ..make_http_listener(http)
7758 })
7759 .into(),
7760 )
7761 .unwrap_err();
7762 let https_error = state
7763 .dispatch(
7764 &RequestType::AddHttpsListener(HttpsListenerConfig {
7765 forwarded_headers: Some(-1),
7766 ..make_https_listener(https)
7767 })
7768 .into(),
7769 )
7770 .unwrap_err();
7771 for error in [http_error, https_error] {
7772 assert!(
7773 matches!(
7774 error,
7775 StateError::InvalidValue {
7776 field: "forwarded_headers",
7777 ..
7778 }
7779 ),
7780 "expected InvalidValue for an unknown mode, got: {error}"
7781 );
7782 }
7783 assert!(state.http_listeners.is_empty() && state.https_listeners.is_empty());
7784
7785 state
7786 .dispatch(
7787 &RequestType::AddHttpListener(HttpListenerConfig {
7788 forwarded_headers: Some(ForwardedHeaders::Rfc7239 as i32),
7789 ..make_http_listener(http)
7790 })
7791 .into(),
7792 )
7793 .expect("a known mode must be accepted on an HTTP listener");
7794 state
7795 .dispatch(
7796 &RequestType::AddHttpsListener(HttpsListenerConfig {
7797 forwarded_headers: Some(ForwardedHeaders::None as i32),
7798 ..make_https_listener(https)
7799 })
7800 .into(),
7801 )
7802 .expect("a known mode must be accepted on an HTTPS listener");
7803 }
7804
7805 #[test]
7810 fn update_listener_forwarded_headers_applies_and_rejects_unknown_values() {
7811 let http = SocketAddress::new_v4(0, 0, 0, 0, 8080);
7812 let https = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7813 let mut state = ConfigState::new();
7814 state
7815 .dispatch(&RequestType::AddHttpListener(make_http_listener(http)).into())
7816 .unwrap();
7817 state
7818 .dispatch(&RequestType::AddHttpsListener(make_https_listener(https)).into())
7819 .unwrap();
7820
7821 state
7822 .dispatch(
7823 &RequestType::UpdateHttpListener(UpdateHttpListenerConfig {
7824 address: http,
7825 forwarded_headers: Some(ForwardedHeaders::Rfc7239 as i32),
7826 ..Default::default()
7827 })
7828 .into(),
7829 )
7830 .expect("a known mode must be accepted on an HTTP listener");
7831 state
7832 .dispatch(
7833 &RequestType::UpdateHttpsListener(UpdateHttpsListenerConfig {
7834 address: https,
7835 forwarded_headers: Some(ForwardedHeaders::None as i32),
7836 ..Default::default()
7837 })
7838 .into(),
7839 )
7840 .expect("a known mode must be accepted on an HTTPS listener");
7841 assert_eq!(
7842 state.http_listeners[&http.into()].forwarded_headers(),
7843 ForwardedHeaders::Rfc7239
7844 );
7845 assert_eq!(
7846 state.https_listeners[&https.into()].forwarded_headers(),
7847 ForwardedHeaders::None
7848 );
7849
7850 let http_error = state
7851 .dispatch(
7852 &RequestType::UpdateHttpListener(UpdateHttpListenerConfig {
7853 address: http,
7854 forwarded_headers: Some(42),
7855 ..Default::default()
7856 })
7857 .into(),
7858 )
7859 .unwrap_err();
7860 let https_error = state
7861 .dispatch(
7862 &RequestType::UpdateHttpsListener(UpdateHttpsListenerConfig {
7863 address: https,
7864 forwarded_headers: Some(-1),
7865 ..Default::default()
7866 })
7867 .into(),
7868 )
7869 .unwrap_err();
7870 for error in [http_error, https_error] {
7871 assert!(
7872 matches!(
7873 error,
7874 StateError::InvalidValue {
7875 field: "forwarded_headers",
7876 ..
7877 }
7878 ),
7879 "expected InvalidValue for an unknown mode, got: {error}"
7880 );
7881 }
7882 assert_eq!(
7884 state.http_listeners[&http.into()].forwarded_headers(),
7885 ForwardedHeaders::Rfc7239
7886 );
7887 assert_eq!(
7888 state.https_listeners[&https.into()].forwarded_headers(),
7889 ForwardedHeaders::None
7890 );
7891 }
7892
7893 #[test]
7895 fn update_https_listener_graceful_shutdown_zero_allowed() {
7896 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
7897 let mut state = ConfigState::new();
7898 state
7899 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
7900 .unwrap();
7901
7902 let patch = UpdateHttpsListenerConfig {
7903 address: addr,
7904 h2_graceful_shutdown_deadline_seconds: Some(0),
7905 ..Default::default()
7906 };
7907 state
7908 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
7909 .expect("graceful_shutdown_deadline=0 must be allowed");
7910
7911 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
7912 assert_eq!(listener.h2_graceful_shutdown_deadline_seconds, Some(0));
7913 }
7914
7915 #[test]
7919 fn update_http_listener_happy_path() {
7920 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
7921 let mut state = ConfigState::new();
7922 state
7923 .dispatch(&RequestType::AddHttpListener(make_http_listener(addr)).into())
7924 .unwrap();
7925
7926 let patch = UpdateHttpListenerConfig {
7927 address: addr,
7928 front_timeout: Some(15),
7929 h2_max_rst_stream_per_window: Some(25),
7930 ..Default::default()
7931 };
7932 state
7933 .dispatch(&RequestType::UpdateHttpListener(patch).into())
7934 .expect("HTTP update must succeed");
7935
7936 let listener = state.http_listeners.get(&ListenerKey::from(addr)).unwrap();
7937 assert_eq!(listener.front_timeout, 15);
7938 assert_eq!(listener.h2_max_rst_stream_per_window, Some(25));
7939 assert_eq!(listener.back_timeout, 30);
7941 }
7942
7943 #[test]
7950 fn update_http_listener_records_the_answers_map() {
7951 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
7952 let mut state = ConfigState::new();
7953 let mut listener = make_http_listener(addr);
7954 listener.answers = [
7955 ("404".to_owned(), "stored 404".to_owned()),
7956 ("503".to_owned(), "stored 503".to_owned()),
7957 ]
7958 .into();
7959 state
7960 .dispatch(&RequestType::AddHttpListener(listener).into())
7961 .unwrap();
7962
7963 let patch = UpdateHttpListenerConfig {
7964 address: addr,
7965 answers: [
7966 ("404".to_owned(), String::new()),
7967 ("503".to_owned(), "patched 503".to_owned()),
7968 ("429".to_owned(), "patched 429".to_owned()),
7969 ]
7970 .into(),
7971 ..Default::default()
7972 };
7973 state
7974 .dispatch(&RequestType::UpdateHttpListener(patch).into())
7975 .expect("HTTP update must succeed");
7976
7977 let listener = state.http_listeners.get(&ListenerKey::from(addr)).unwrap();
7978 assert_eq!(
7979 listener.answers.get("404").map(String::as_str),
7980 Some("stored 404"),
7981 "an empty body preserves the stored template"
7982 );
7983 assert_eq!(
7984 listener.answers.get("503").map(String::as_str),
7985 Some("patched 503")
7986 );
7987 assert_eq!(
7988 listener.answers.get("429").map(String::as_str),
7989 Some("patched 429")
7990 );
7991 assert_eq!(listener.answers.len(), 3);
7992 }
7993
7994 #[test]
8000 fn update_https_listener_records_the_answers_map_and_hsts() {
8001 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
8002 let mut state = ConfigState::new();
8003 let mut listener = make_https_listener(addr);
8004 listener.answers = [("404".to_owned(), "stored 404".to_owned())].into();
8005 listener.hsts = Some(HstsConfig {
8006 enabled: Some(true),
8007 max_age: Some(300),
8008 include_subdomains: Some(true),
8009 ..Default::default()
8010 });
8011 state
8012 .dispatch(&RequestType::AddHttpsListener(listener).into())
8013 .unwrap();
8014
8015 let disabled = HstsConfig {
8016 enabled: Some(false),
8017 ..Default::default()
8018 };
8019 let patch = UpdateHttpsListenerConfig {
8020 address: addr,
8021 answers: [
8022 ("404".to_owned(), String::new()),
8023 ("503".to_owned(), "patched 503".to_owned()),
8024 ]
8025 .into(),
8026 hsts: Some(disabled),
8027 ..Default::default()
8028 };
8029 state
8030 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
8031 .expect("HTTPS update must succeed");
8032
8033 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
8034 assert_eq!(
8035 listener.answers.get("404").map(String::as_str),
8036 Some("stored 404"),
8037 "an empty body preserves the stored template"
8038 );
8039 assert_eq!(
8040 listener.answers.get("503").map(String::as_str),
8041 Some("patched 503")
8042 );
8043 assert_eq!(listener.answers.len(), 2);
8044 assert_eq!(
8045 listener.hsts,
8046 Some(disabled),
8047 "the patched hsts block replaces the stored one whole"
8048 );
8049
8050 let patch = UpdateHttpsListenerConfig {
8051 address: addr,
8052 front_timeout: Some(15),
8053 ..Default::default()
8054 };
8055 state
8056 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
8057 .expect("HTTPS update must succeed");
8058 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
8059 assert_eq!(
8060 listener.hsts,
8061 Some(disabled),
8062 "a patch without hsts keeps the recorded policy"
8063 );
8064 }
8065
8066 #[test]
8068 fn update_http_listener_flood_knob_zero_rejected() {
8069 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
8070 let mut state = ConfigState::new();
8071 state
8072 .dispatch(&RequestType::AddHttpListener(make_http_listener(addr)).into())
8073 .unwrap();
8074
8075 let patch = UpdateHttpListenerConfig {
8076 address: addr,
8077 h2_max_window_update_stream0_per_window: Some(0),
8078 ..Default::default()
8079 };
8080 let err = state
8081 .dispatch(&RequestType::UpdateHttpListener(patch).into())
8082 .unwrap_err();
8083 assert!(
8084 matches!(
8085 err,
8086 StateError::InvalidValue {
8087 field: "h2_max_window_update_stream0_per_window",
8088 ..
8089 }
8090 ),
8091 "expected InvalidValue, got: {err}"
8092 );
8093 }
8094
8095 #[test]
8101 fn update_http_listener_x_real_ip_patch_applied() {
8102 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
8103 let mut state = ConfigState::new();
8104 state
8105 .dispatch(&RequestType::AddHttpListener(make_http_listener(addr)).into())
8106 .unwrap();
8107
8108 let patch = UpdateHttpListenerConfig {
8109 address: addr,
8110 elide_x_real_ip: Some(true),
8111 send_x_real_ip: Some(true),
8112 ..Default::default()
8113 };
8114 state
8115 .dispatch(&RequestType::UpdateHttpListener(patch).into())
8116 .expect("HTTP update must succeed");
8117
8118 let listener = state.http_listeners.get(&ListenerKey::from(addr)).unwrap();
8119 assert_eq!(listener.elide_x_real_ip, Some(true));
8120 assert_eq!(listener.send_x_real_ip, Some(true));
8121 assert_eq!(listener.front_timeout, 60);
8123 }
8124
8125 #[test]
8128 fn update_http_listener_x_real_ip_absent_preserves_existing() {
8129 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
8130 let mut state = ConfigState::new();
8131 let listener = HttpListenerConfig {
8132 elide_x_real_ip: Some(true),
8133 send_x_real_ip: Some(false),
8134 ..make_http_listener(addr)
8135 };
8136 state
8137 .dispatch(&RequestType::AddHttpListener(listener).into())
8138 .unwrap();
8139
8140 let patch = UpdateHttpListenerConfig {
8141 address: addr,
8142 send_x_real_ip: Some(true),
8143 ..Default::default()
8144 };
8145 state
8146 .dispatch(&RequestType::UpdateHttpListener(patch).into())
8147 .expect("HTTP update must succeed");
8148
8149 let listener = state.http_listeners.get(&ListenerKey::from(addr)).unwrap();
8150 assert_eq!(listener.elide_x_real_ip, Some(true));
8151 assert_eq!(listener.send_x_real_ip, Some(true));
8152 }
8153
8154 #[test]
8157 fn update_https_listener_x_real_ip_patch_applied() {
8158 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
8159 let mut state = ConfigState::new();
8160 state
8161 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
8162 .unwrap();
8163
8164 let patch = UpdateHttpsListenerConfig {
8165 address: addr,
8166 elide_x_real_ip: Some(true),
8167 send_x_real_ip: Some(true),
8168 ..Default::default()
8169 };
8170 state
8171 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
8172 .expect("HTTPS update must succeed");
8173
8174 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
8175 assert_eq!(listener.elide_x_real_ip, Some(true));
8176 assert_eq!(listener.send_x_real_ip, Some(true));
8177 assert_eq!(listener.front_timeout, 60);
8179 }
8180
8181 #[test]
8184 fn update_https_listener_x_real_ip_absent_preserves_existing() {
8185 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
8186 let mut state = ConfigState::new();
8187 let listener = HttpsListenerConfig {
8188 elide_x_real_ip: Some(false),
8189 send_x_real_ip: Some(true),
8190 ..make_https_listener(addr)
8191 };
8192 state
8193 .dispatch(&RequestType::AddHttpsListener(listener).into())
8194 .unwrap();
8195
8196 let patch = UpdateHttpsListenerConfig {
8197 address: addr,
8198 elide_x_real_ip: Some(true),
8199 ..Default::default()
8200 };
8201 state
8202 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
8203 .expect("HTTPS update must succeed");
8204
8205 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
8206 assert_eq!(listener.elide_x_real_ip, Some(true));
8207 assert_eq!(listener.send_x_real_ip, Some(true));
8208 }
8209
8210 #[test]
8213 fn update_listener_x_real_ip_survives_generate_requests_replay() {
8214 let http_addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
8215 let https_addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
8216 let mut state = ConfigState::new();
8217 state
8218 .dispatch(&RequestType::AddHttpListener(make_http_listener(http_addr)).into())
8219 .unwrap();
8220 state
8221 .dispatch(&RequestType::AddHttpsListener(make_https_listener(https_addr)).into())
8222 .unwrap();
8223 state
8224 .dispatch(
8225 &RequestType::UpdateHttpListener(UpdateHttpListenerConfig {
8226 address: http_addr,
8227 elide_x_real_ip: Some(true),
8228 send_x_real_ip: Some(true),
8229 ..Default::default()
8230 })
8231 .into(),
8232 )
8233 .unwrap();
8234 state
8235 .dispatch(
8236 &RequestType::UpdateHttpsListener(UpdateHttpsListenerConfig {
8237 address: https_addr,
8238 elide_x_real_ip: Some(true),
8239 send_x_real_ip: Some(true),
8240 ..Default::default()
8241 })
8242 .into(),
8243 )
8244 .unwrap();
8245
8246 let mut replayed = ConfigState::new();
8247 for request in state.generate_requests() {
8248 replayed
8249 .dispatch(&request)
8250 .expect("could not replay generated request");
8251 }
8252
8253 let http = replayed
8254 .http_listeners
8255 .get(&ListenerKey::from(http_addr))
8256 .expect("replayed HTTP listener");
8257 assert_eq!(http.elide_x_real_ip, Some(true));
8258 assert_eq!(http.send_x_real_ip, Some(true));
8259 let https = replayed
8260 .https_listeners
8261 .get(&ListenerKey::from(https_addr))
8262 .expect("replayed HTTPS listener");
8263 assert_eq!(https.elide_x_real_ip, Some(true));
8264 assert_eq!(https.send_x_real_ip, Some(true));
8265 }
8266
8267 #[test]
8271 fn update_tcp_listener_happy_path() {
8272 let addr = SocketAddress::new_v4(0, 0, 0, 0, 9000);
8273 let mut state = ConfigState::new();
8274 state
8275 .dispatch(&RequestType::AddTcpListener(make_tcp_listener(addr)).into())
8276 .unwrap();
8277
8278 let patch = UpdateTcpListenerConfig {
8279 address: addr,
8280 front_timeout: Some(5),
8281 ..Default::default()
8282 };
8283 state
8284 .dispatch(&RequestType::UpdateTcpListener(patch).into())
8285 .expect("TCP update must succeed");
8286
8287 let listener = state.tcp_listeners.get(&ListenerKey::from(addr)).unwrap();
8288 assert_eq!(listener.front_timeout, 5);
8289 assert_eq!(listener.back_timeout, 30); }
8291
8292 #[test]
8294 fn update_tcp_listener_not_found() {
8295 let mut state = ConfigState::new();
8296 let patch = UpdateTcpListenerConfig {
8297 address: SocketAddress::new_v4(9, 9, 9, 9, 9999),
8298 front_timeout: Some(5),
8299 ..Default::default()
8300 };
8301 let err = state
8302 .dispatch(&RequestType::UpdateTcpListener(patch).into())
8303 .unwrap_err();
8304 assert!(
8305 matches!(
8306 err,
8307 StateError::NotFound {
8308 kind: ObjectKind::TcpListener,
8309 ..
8310 }
8311 ),
8312 "expected NotFound, got: {err}"
8313 );
8314 }
8315
8316 fn without_request_counts(state: &ConfigState) -> ConfigState {
8322 ConfigState {
8323 request_counts: BTreeMap::new(),
8324 ..state.clone()
8325 }
8326 }
8327
8328 #[test]
8339 fn update_http_listener_rejected_patch_leaves_the_state_unchanged() {
8340 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
8341 let mut state = ConfigState::new();
8342 state
8343 .dispatch(&RequestType::AddHttpListener(make_http_listener(addr)).into())
8344 .unwrap();
8345
8346 let valid_prefix = UpdateHttpListenerConfig {
8347 address: addr,
8348 front_timeout: Some(99),
8349 sticky_name: Some("PATCHED".to_owned()),
8350 elide_x_real_ip: Some(true),
8351 http_answers: Some(CustomHttpAnswers {
8352 answer_404: Some("HTTP/1.1 404 Not Found\r\n\r\n".to_owned()),
8353 ..Default::default()
8354 }),
8355 h2_max_rst_stream_per_window: Some(7),
8356 ..Default::default()
8357 };
8358 let rejected = [
8359 (
8360 "sozu_id_header",
8361 UpdateHttpListenerConfig {
8362 sozu_id_header: Some("bad header".to_owned()),
8363 ..valid_prefix.clone()
8364 },
8365 ),
8366 (
8367 "forwarded_headers",
8368 UpdateHttpListenerConfig {
8369 forwarded_headers: Some(i32::MAX),
8370 ..valid_prefix.clone()
8371 },
8372 ),
8373 ];
8374 for (field, patch) in rejected {
8375 let before = without_request_counts(&state);
8376 let err = state
8377 .dispatch(&RequestType::UpdateHttpListener(patch).into())
8378 .unwrap_err();
8379 assert!(
8380 matches!(err, StateError::InvalidValue { field: f, .. } if f == field),
8381 "expected InvalidValue on {field}, got: {err}"
8382 );
8383 assert!(
8384 without_request_counts(&state) == before,
8385 "a patch rejected on {field} must leave the main state unchanged"
8386 );
8387 }
8388
8389 state
8392 .dispatch(&RequestType::UpdateHttpListener(valid_prefix).into())
8393 .expect("the valid prefix alone must apply");
8394 let listener = state.http_listeners.get(&ListenerKey::from(addr)).unwrap();
8395 assert_eq!(listener.front_timeout, 99);
8396 assert_eq!(listener.sticky_name, "PATCHED");
8397 }
8398
8399 #[test]
8409 fn update_https_listener_rejected_patch_leaves_the_state_unchanged() {
8410 use crate::proto::command::AlpnProtocols;
8411
8412 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
8413 let mut state = ConfigState::new();
8414 state
8415 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
8416 .unwrap();
8417
8418 let valid_prefix = UpdateHttpsListenerConfig {
8419 address: addr,
8420 front_timeout: Some(99),
8421 sticky_name: Some("PATCHED".to_owned()),
8422 send_x_real_ip: Some(true),
8423 http_answers: Some(CustomHttpAnswers {
8424 answer_404: Some("HTTP/1.1 404 Not Found\r\n\r\n".to_owned()),
8425 ..Default::default()
8426 }),
8427 strict_sni_binding: Some(false),
8428 h2_max_rst_stream_per_window: Some(7),
8429 ..Default::default()
8430 };
8431 let rejected = [
8432 (
8433 "alpn_protocols",
8434 UpdateHttpsListenerConfig {
8435 alpn_protocols: Some(AlpnProtocols {
8436 values: vec!["h3".to_owned()],
8437 }),
8438 ..valid_prefix.clone()
8439 },
8440 ),
8441 (
8442 "sozu_id_header",
8443 UpdateHttpsListenerConfig {
8444 sozu_id_header: Some(String::new()),
8445 ..valid_prefix.clone()
8446 },
8447 ),
8448 (
8449 "forwarded_headers",
8450 UpdateHttpsListenerConfig {
8451 forwarded_headers: Some(-1),
8452 ..valid_prefix.clone()
8453 },
8454 ),
8455 (
8456 "hsts",
8457 UpdateHttpsListenerConfig {
8458 hsts: Some(HstsConfig {
8459 enabled: None,
8460 max_age: Some(60),
8461 ..Default::default()
8462 }),
8463 ..valid_prefix.clone()
8464 },
8465 ),
8466 ];
8467 for (field, patch) in rejected {
8468 let before = without_request_counts(&state);
8469 let err = state
8470 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
8471 .unwrap_err();
8472 assert!(
8473 matches!(err, StateError::InvalidValue { field: f, .. } if f == field),
8474 "expected InvalidValue on {field}, got: {err}"
8475 );
8476 assert!(
8477 without_request_counts(&state) == before,
8478 "a patch rejected on {field} must leave the main state unchanged"
8479 );
8480 }
8481
8482 state
8483 .dispatch(&RequestType::UpdateHttpsListener(valid_prefix).into())
8484 .expect("the valid prefix alone must apply");
8485 let listener = state.https_listeners.get(&ListenerKey::from(addr)).unwrap();
8486 assert_eq!(listener.front_timeout, 99);
8487 assert_eq!(listener.strict_sni_binding, Some(false));
8488 }
8489
8490 #[test]
8494 fn update_tcp_listener_rejected_patch_leaves_the_state_unchanged() {
8495 let addr = SocketAddress::new_v4(0, 0, 0, 0, 9000);
8496 let mut state = ConfigState::new();
8497 state
8498 .dispatch(&RequestType::AddTcpListener(make_tcp_listener(addr)).into())
8499 .unwrap();
8500
8501 let before = without_request_counts(&state);
8502 let err = state
8503 .dispatch(
8504 &RequestType::UpdateTcpListener(UpdateTcpListenerConfig {
8505 address: SocketAddress::new_v4(0, 0, 0, 0, 9001),
8506 front_timeout: Some(5),
8507 ..Default::default()
8508 })
8509 .into(),
8510 )
8511 .unwrap_err();
8512 assert!(matches!(err, StateError::NotFound { .. }), "got: {err}");
8513 assert!(
8514 without_request_counts(&state) == before,
8515 "a refused TCP patch must change nothing"
8516 );
8517 }
8518
8519 #[test]
8525 fn dispatch_passes_through_set_metric_detail() {
8526 use crate::proto::command::{MetricDetail, SetMetricDetail};
8527 let mut state = ConfigState::new();
8528 let req: Request = RequestType::SetMetricDetail(SetMetricDetail {
8529 client_id: "test:1".to_owned(),
8530 detail: Some(MetricDetail::DetailBackend as i32),
8531 ttl_seconds: Some(60),
8532 clear: Some(false),
8533 reason: Some("regression-guard".to_owned()),
8534 peer_pid: None,
8535 peer_session_ulid: None,
8536 })
8537 .into();
8538 state
8539 .dispatch(&req)
8540 .expect("SetMetricDetail must traverse dispatch without UndispatchableRequest");
8541 }
8542
8543 #[test]
8582 fn remove_udp_frontend_drops_exactly_the_frontend_its_tags_name() {
8583 let team_a_address = SocketAddress::new_v4(127, 0, 0, 1, 9100);
8584 let team_b_address = SocketAddress::new_v4(127, 0, 0, 1, 9101);
8585 let front = |owner: &str, address| RequestUdpFrontend {
8586 cluster_id: "udp_cluster".to_string(),
8587 address,
8588 tags: BTreeMap::from([("owner".to_string(), owner.to_string())]),
8589 };
8590 let mut state = ConfigState::new();
8591
8592 for (owner, address) in [("team-a", team_a_address), ("team-b", team_b_address)] {
8596 state
8597 .dispatch(&RequestType::AddUdpFrontend(front(owner, address)).into())
8598 .expect("one UDP frontend per address must be admitted");
8599 }
8600 assert_eq!(
8601 state.udp_fronts.get("udp_cluster").map(Vec::len),
8602 Some(2usize),
8603 "both tagged UDP frontends must be in the state before the removal"
8604 );
8605
8606 state
8607 .dispatch(&RequestType::RemoveUdpFrontend(front("team-a", team_a_address)).into())
8608 .expect("removing one of the two must succeed");
8609
8610 let surviving: Vec<&str> = state
8611 .udp_fronts
8612 .get("udp_cluster")
8613 .map(|fronts| {
8614 fronts
8615 .iter()
8616 .filter_map(|f| f.tags.get("owner").map(String::as_str))
8617 .collect()
8618 })
8619 .unwrap_or_default();
8620 assert_eq!(
8621 surviving,
8622 vec!["team-b"],
8623 "removing one UDP frontend must not evict the rest of its cluster's bucket"
8624 );
8625
8626 let tagless = RequestUdpFrontend {
8630 cluster_id: "udp_cluster".to_string(),
8631 address: team_b_address,
8632 tags: BTreeMap::new(),
8633 };
8634 let err = state
8635 .dispatch(&RequestType::RemoveUdpFrontend(tagless).into())
8636 .expect_err("a tagless remove must not match a tagged UDP frontend");
8637 assert!(
8638 matches!(err, StateError::NoChange),
8639 "expected NoChange, got: {err}"
8640 );
8641 assert_eq!(
8642 state.udp_fronts.get("udp_cluster").map(Vec::len),
8643 Some(1usize),
8644 "the tagless remove must leave the surviving tagged frontend in place"
8645 );
8646 }
8647
8648 fn udp_front(cluster_id: &str, address: SocketAddress, owner: &str) -> RequestUdpFrontend {
8649 RequestUdpFrontend {
8650 cluster_id: cluster_id.to_string(),
8651 address,
8652 tags: BTreeMap::from([("owner".to_string(), owner.to_string())]),
8653 }
8654 }
8655
8656 #[test]
8687 fn add_udp_frontend_refuses_a_second_frontend_on_a_taken_address() {
8688 let address = SocketAddress::new_v4(127, 0, 0, 1, 5353);
8689 let other_address = SocketAddress::new_v4(127, 0, 0, 1, 5354);
8690 let mut state = ConfigState::new();
8691
8692 state
8693 .dispatch(&RequestType::AddUdpFrontend(udp_front("dns", address, "team-a")).into())
8694 .expect("the first claim on an address must be admitted");
8695
8696 state
8698 .dispatch(&RequestType::AddUdpFrontend(udp_front("dns", address, "team-b")).into())
8699 .expect_err("a second UDP frontend on a taken address must be refused");
8700
8701 state
8704 .dispatch(&RequestType::AddUdpFrontend(udp_front("syslog", address, "team-c")).into())
8705 .expect_err("a second UDP frontend on a taken address must be refused");
8706
8707 assert_eq!(
8708 state.count_udp_frontends_raw(),
8709 1,
8710 "a refused claim must leave the state holding only the admitted frontend"
8711 );
8712 assert!(
8713 state.udp_fronts.get("syslog").is_none_or(Vec::is_empty),
8714 "a refused cross-cluster claim must not leave an opened bucket behind"
8715 );
8716
8717 state
8719 .dispatch(
8720 &RequestType::AddUdpFrontend(udp_front("syslog", other_address, "team-c")).into(),
8721 )
8722 .expect("a UDP frontend on a free address must still be admitted");
8723
8724 state
8727 .dispatch(&RequestType::RemoveUdpFrontend(udp_front("dns", address, "team-a")).into())
8728 .expect("removing the frontend holding the address must succeed");
8729 state
8730 .dispatch(&RequestType::AddUdpFrontend(udp_front("syslog", address, "team-c")).into())
8731 .expect("a freed address must be claimable again, by any cluster");
8732
8733 assert_eq!(
8734 state.count_udp_frontends_raw(),
8735 2,
8736 "the freed address must now be held by exactly one frontend, beside the other address"
8737 );
8738 }
8739
8740 #[test]
8757 fn a_refused_udp_frontend_names_the_cluster_already_holding_the_address() {
8758 const OWNER_SECRET: &str = "UDP_ADDRESS_OWNER_SECRET_SENTINEL";
8759
8760 let address = SocketAddress::new_v4(127, 0, 0, 1, 5353);
8761 let owner_cluster = format!("{OWNER_SECRET}-dns");
8762 let owner_tag = format!("{OWNER_SECRET}-team-a");
8763 let mut state = ConfigState::new();
8764
8765 state
8766 .dispatch(
8767 &RequestType::AddUdpFrontend(udp_front(&owner_cluster, address, &owner_tag)).into(),
8768 )
8769 .expect("the first claim on an address must be admitted");
8770
8771 let error = state
8772 .dispatch(&RequestType::AddUdpFrontend(udp_front("syslog", address, "team-c")).into())
8773 .expect_err("a second UDP frontend on a taken address must be refused");
8774
8775 match &error {
8776 StateError::UdpFrontendAddressTaken {
8777 address: refused_address,
8778 cluster_id,
8779 current_cluster_id,
8780 current_tags,
8781 } => {
8782 assert_eq!(
8783 *refused_address,
8784 SocketAddr::from(address),
8785 "the refusal must name the contested address"
8786 );
8787 assert_eq!(
8788 &**cluster_id, "syslog",
8789 "the refusal must name the claimant"
8790 );
8791 assert_eq!(
8792 **current_cluster_id, *owner_cluster,
8793 "the refusal must name the cluster already holding the address"
8794 );
8795 assert_eq!(
8796 current_tags.get("owner"),
8797 Some(&owner_tag),
8798 "the refusal must carry the current owner's access-log tags"
8799 );
8800 }
8801 other => panic!("expected UdpFrontendAddressTaken, got {other:?}"),
8802 }
8803
8804 for (label, output) in [
8807 ("Display", error.to_string()),
8808 ("Debug", format!("{error:?}")),
8809 ] {
8810 assert!(
8811 output.contains("127.0.0.1:5353"),
8812 "StateError {label} must name the contested address: {output}"
8813 );
8814 assert!(
8815 !output.contains(OWNER_SECRET),
8816 "StateError {label} leaked a protected identifier: {output}"
8817 );
8818 assert!(
8819 output.contains(&format!("current_cluster_id_bytes={}", owner_cluster.len())),
8820 "StateError {label} omitted the bounded owner metadata: {output}"
8821 );
8822 assert!(
8823 output.contains("current_tag_count=1"),
8824 "StateError {label} omitted the bounded owner tag count: {output}"
8825 );
8826 }
8827 }
8828
8829 #[test]
8852 fn a_saved_state_holding_one_udp_frontend_per_address_replays_unchanged() {
8853 use std::io::Read;
8854
8855 use crate::parser::parse_several_requests;
8856
8857 let first = SocketAddress::new_v4(127, 0, 0, 1, 5353);
8858 let second = SocketAddress::new_v4(127, 0, 0, 2, 5353);
8859
8860 let mut state = ConfigState::new();
8861 for (cluster_id, address) in [("dns", first), ("syslog", second)] {
8862 state
8863 .dispatch(&RequestType::AddUdpListener(make_udp_listener(address, true)).into())
8864 .expect("could not add udp listener");
8865 state
8866 .dispatch(
8867 &RequestType::ActivateListener(ActivateListener {
8868 interface: None,
8869 address,
8870 proxy: ListenerType::Udp.into(),
8871 from_scm: false,
8872 })
8873 .into(),
8874 )
8875 .expect("could not activate udp listener");
8876 state
8877 .dispatch(
8878 &RequestType::AddUdpFrontend(udp_front(cluster_id, address, "team-a")).into(),
8879 )
8880 .expect("one UDP frontend per address must be admitted");
8881 }
8882
8883 let directory = tempfile::tempdir().expect("a temporary directory for the state file");
8884 let path = directory.path().join("udpaddr-roundtrip.state");
8885 let mut file = File::create(&path).expect("the state file must be creatable");
8886 let written = state
8887 .write_requests_to_file(&mut file)
8888 .expect("SaveState must serialise the state");
8889 drop(file);
8890
8891 let mut saved = Vec::new();
8892 File::open(&path)
8893 .expect("the state file must be readable")
8894 .read_to_end(&mut saved)
8895 .expect("the state file must be readable");
8896
8897 let (remainder, requests) =
8898 parse_several_requests::<WorkerRequest>(&saved).expect("the state file must parse");
8899 assert!(
8900 remainder.is_empty(),
8901 "load_state would report leftover bytes: {} unparsed",
8902 remainder.len()
8903 );
8904 assert_eq!(
8905 requests.len(),
8906 written,
8907 "every record write_requests_to_file emitted must parse back"
8908 );
8909
8910 let mut replayed = ConfigState::new();
8911 for request in &requests {
8912 replayed
8913 .dispatch(&request.content)
8914 .expect("every saved UDP entry must replay into a fresh state");
8915 }
8916
8917 let logical = |s: &ConfigState| {
8918 let mut c = s.clone();
8919 c.request_counts.clear();
8920 c
8921 };
8922 assert_eq!(
8923 logical(&state),
8924 logical(&replayed),
8925 "a SaveState -> LoadState round trip must reconstruct the UDP configuration"
8926 );
8927 }
8928
8929 fn with_interface<T>(mut listener: T, set: impl FnOnce(&mut T)) -> T {
8932 set(&mut listener);
8933 listener
8934 }
8935
8936 #[test]
8940 fn listeners_on_one_address_with_different_interfaces_coexist() {
8941 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
8942 let mut state = ConfigState::new();
8943 for interface in [None, Some("lo"), Some("wg0")] {
8944 let iface = interface.map(str::to_owned);
8945 state
8946 .dispatch(
8947 &RequestType::AddHttpListener(with_interface(make_http_listener(addr), |l| {
8948 l.interface = iface.clone()
8949 }))
8950 .into(),
8951 )
8952 .expect("an HTTP listener on another interface must not collide");
8953 state
8954 .dispatch(
8955 &RequestType::AddHttpsListener(with_interface(
8956 make_https_listener(addr),
8957 |l| l.interface = iface.clone(),
8958 ))
8959 .into(),
8960 )
8961 .expect("an HTTPS listener on another interface must not collide");
8962 state
8963 .dispatch(
8964 &RequestType::AddTcpListener(with_interface(make_tcp_listener(addr), |l| {
8965 l.interface = iface.clone()
8966 }))
8967 .into(),
8968 )
8969 .expect("a TCP listener on another interface must not collide");
8970 state
8971 .dispatch(
8972 &RequestType::AddUdpListener(with_interface(
8973 make_udp_listener(addr, false),
8974 |l| l.interface = iface.clone(),
8975 ))
8976 .into(),
8977 )
8978 .expect("a UDP listener on another interface must not collide");
8979 }
8980
8981 let expected = vec![
8982 "0.0.0.0:8443".to_owned(),
8983 "0.0.0.0:8443%lo".to_owned(),
8984 "0.0.0.0:8443%wg0".to_owned(),
8985 ];
8986 let list = state.list_listeners();
8987 assert_eq!(
8988 list.http_listeners.keys().cloned().collect::<Vec<_>>(),
8989 expected
8990 );
8991 assert_eq!(
8992 list.https_listeners.keys().cloned().collect::<Vec<_>>(),
8993 expected
8994 );
8995 assert_eq!(
8996 list.tcp_listeners.keys().cloned().collect::<Vec<_>>(),
8997 expected
8998 );
8999 assert_eq!(
9000 list.udp_listeners.keys().cloned().collect::<Vec<_>>(),
9001 expected
9002 );
9003 assert_eq!(
9004 list.http_listeners["0.0.0.0:8443%wg0"].interface.as_deref(),
9005 Some("wg0")
9006 );
9007
9008 assert!(
9010 state
9011 .dispatch(
9012 &RequestType::AddHttpListener(with_interface(make_http_listener(addr), |l| {
9013 l.interface = Some("lo".to_owned())
9014 }))
9015 .into(),
9016 )
9017 .is_err(),
9018 "the same (address, interface) must still be refused as a duplicate"
9019 );
9020
9021 state
9023 .dispatch(
9024 &RequestType::ActivateListener(ActivateListener {
9025 address: addr,
9026 proxy: ListenerType::Http.into(),
9027 from_scm: false,
9028 interface: Some("wg0".to_owned()),
9029 })
9030 .into(),
9031 )
9032 .expect("activating the wg0 listener");
9033 let list = state.list_listeners();
9034 assert!(list.http_listeners["0.0.0.0:8443%wg0"].active);
9035 assert!(!list.http_listeners["0.0.0.0:8443%lo"].active);
9036 assert!(!list.http_listeners["0.0.0.0:8443"].active);
9037
9038 state
9039 .dispatch(
9040 &RequestType::RemoveListener(RemoveListener {
9041 address: addr,
9042 proxy: ListenerType::Tcp.into(),
9043 interface: Some("lo".to_owned()),
9044 })
9045 .into(),
9046 )
9047 .expect("removing the lo TCP listener");
9048 let list = state.list_listeners();
9049 assert_eq!(
9050 list.tcp_listeners.keys().cloned().collect::<Vec<_>>(),
9051 vec!["0.0.0.0:8443".to_owned(), "0.0.0.0:8443%wg0".to_owned()]
9052 );
9053
9054 assert!(
9057 state
9058 .dispatch(
9059 &RequestType::RemoveListener(RemoveListener {
9060 address: addr,
9061 proxy: ListenerType::Tcp.into(),
9062 interface: Some("eth9".to_owned()),
9063 })
9064 .into(),
9065 )
9066 .is_err()
9067 );
9068
9069 let mut replayed = ConfigState::new();
9072 for request in state.generate_requests() {
9073 replayed
9074 .dispatch(&request)
9075 .expect("replaying generated requests");
9076 }
9077 assert_eq!(replayed.list_listeners(), state.list_listeners());
9078 assert!(state.diff(&replayed).is_empty());
9079 }
9080
9081 #[test]
9085 fn listener_diff_distinguishes_interfaces() {
9086 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
9087 let mut before = ConfigState::new();
9088 before
9089 .dispatch(
9090 &RequestType::AddHttpListener(with_interface(make_http_listener(addr), |l| {
9091 l.interface = Some("eth0".to_owned())
9092 }))
9093 .into(),
9094 )
9095 .unwrap();
9096 let mut after = ConfigState::new();
9097 after
9098 .dispatch(
9099 &RequestType::AddHttpListener(with_interface(make_http_listener(addr), |l| {
9100 l.interface = Some("wg0".to_owned())
9101 }))
9102 .into(),
9103 )
9104 .unwrap();
9105
9106 let diff = before.diff(&after);
9107 let mut removed = Vec::new();
9108 let mut added = Vec::new();
9109 for request in &diff {
9110 match &request.request_type {
9111 Some(RequestType::RemoveListener(remove)) => removed.push(remove.interface.clone()),
9112 Some(RequestType::AddHttpListener(add)) => added.push(add.interface.clone()),
9113 _ => {}
9114 }
9115 }
9116 assert_eq!(removed, vec![Some("eth0".to_owned())]);
9117 assert_eq!(added, vec![Some("wg0".to_owned())]);
9118 }
9119
9120 #[test]
9126 fn reload_adding_an_interface_listener_sends_only_that_listener() {
9127 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
9128 let front = RequestHttpFrontend {
9129 cluster_id: Some("cluster_1".to_owned()),
9130 hostname: "example.com".to_owned(),
9131 path: PathRule::prefix("/".to_owned()),
9132 address: addr,
9133 ..Default::default()
9134 };
9135 let mut before = ConfigState::new();
9136 for request in [
9137 RequestType::AddHttpListener(make_http_listener(addr)),
9138 RequestType::AddHttpFrontend(front),
9139 ] {
9140 before.dispatch(&request.into()).unwrap();
9141 }
9142 let mut after = before.clone();
9143 after
9144 .dispatch(
9145 &RequestType::AddHttpListener(with_interface(make_http_listener(addr), |l| {
9146 l.interface = Some("wg0".to_owned())
9147 }))
9148 .into(),
9149 )
9150 .unwrap();
9151
9152 let diff = before.diff(&after);
9153 assert_eq!(diff.len(), 1, "{diff:?}");
9154 assert!(matches!(
9155 &diff[0].request_type,
9156 Some(RequestType::AddHttpListener(listener))
9157 if listener.interface.as_deref() == Some("wg0")
9158 ));
9159 }
9160
9161 #[test]
9165 fn update_listener_is_refused_when_the_address_is_ambiguous() {
9166 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
9167 let mut state = ConfigState::new();
9168 state
9169 .dispatch(
9170 &RequestType::AddHttpListener(with_interface(make_http_listener(addr), |l| {
9171 l.interface = Some("wg0".to_owned())
9172 }))
9173 .into(),
9174 )
9175 .unwrap();
9176
9177 state
9179 .dispatch(
9180 &RequestType::UpdateHttpListener(UpdateHttpListenerConfig {
9181 address: addr,
9182 front_timeout: Some(15),
9183 ..Default::default()
9184 })
9185 .into(),
9186 )
9187 .expect("an unambiguous patch applies to the interface-bound listener");
9188 assert_eq!(
9189 state.list_listeners().http_listeners["0.0.0.0:8080%wg0"].front_timeout,
9190 15
9191 );
9192
9193 state
9194 .dispatch(&RequestType::AddHttpListener(make_http_listener(addr)).into())
9195 .unwrap();
9196 let before = state.list_listeners();
9197 let error = state
9198 .dispatch(
9199 &RequestType::UpdateHttpListener(UpdateHttpListenerConfig {
9200 address: addr,
9201 front_timeout: Some(20),
9202 ..Default::default()
9203 })
9204 .into(),
9205 )
9206 .expect_err("a patch matching two listeners must be refused");
9207 assert!(
9208 matches!(error, StateError::AmbiguousListenerAddress { count: 2, .. }),
9209 "{error}"
9210 );
9211 assert_eq!(
9212 before,
9213 state.list_listeners(),
9214 "a refused patch must leave the listeners untouched"
9215 );
9216 }
9217}