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