1use std::{
2 collections::{
3 BTreeMap, BTreeSet, HashMap, HashSet, btree_map::Entry as BTreeMapEntry,
4 hash_map::DefaultHasher,
5 },
6 fmt,
7 fs::File,
8 hash::{Hash, Hasher},
9 io::Write,
10 iter::repeat,
11 net::SocketAddr,
12};
13
14use prost::{Message, UnknownEnumValue};
15
16use crate::{
17 ObjectKind,
18 certificate::{CertificateError, Fingerprint, calculate_fingerprint},
19 config::validate_sni_pattern,
20 proto::{
21 command::{
22 ActivateListener, AddBackend, AddCertificate, CertificateAndKey, Cluster,
23 ClusterInformation, CustomHttpAnswers, DeactivateListener, FrontendFilters,
24 HealthChecksList, HttpListenerConfig, HttpsListenerConfig, InitialState,
25 ListedFrontends, ListenerType, ListenersList, PathRule, QueryCertificatesFilters,
26 RemoveBackend, RemoveCertificate, RemoveListener, ReplaceCertificate, Request,
27 RequestCounts, RequestHttpFrontend, RequestTcpFrontend, RequestUdpFrontend,
28 SetHealthCheck, SocketAddress, TcpListenerConfig, UdpListenerConfig,
29 UpdateHttpListenerConfig, UpdateHttpsListenerConfig, UpdateTcpListenerConfig,
30 UpdateUdpListenerConfig, WorkerRequest, request::RequestType,
31 },
32 display::format_request_type,
33 },
34 response::{Backend, HttpFrontend, TcpFrontend, UdpFrontend},
35};
36
37pub type ClusterId = String;
39
40#[derive(thiserror::Error)]
41pub enum StateError {
42 #[error("Request came in empty")]
43 EmptyRequest,
44 #[error("dispatching this request did not bring any change to the state")]
45 NoChange,
46 #[error("State can not handle this request")]
47 UndispatchableRequest,
48 #[error("Did not find {kind:?} with address or id_bytes={}", .id.len())]
49 NotFound { kind: ObjectKind, id: String },
50 #[error(
51 "{kind:?} with id_bytes={} already exists; remove it first, or apply the corresponding update if the object supports one, instead of re-adding it",
52 .id.len()
53 )]
54 Exists { kind: ObjectKind, id: String },
55 #[error("Wrong field value: {0}")]
56 WrongFieldValue(UnknownEnumValue),
57 #[error("Could not add certificate: error_kind={}", certificate_error_kind(.0))]
58 AddCertificate(CertificateError),
59 #[error("Could not remove certificate: fingerprint_bytes={}", .0.len())]
60 RemoveCertificate(String),
61 #[error("Could not replace certificate: error_bytes={}", .0.len())]
62 ReplaceCertificate(String),
63 #[error("Could not convert frontend: frontend_bytes={} error_bytes={}", .frontend.len(), .error.len())]
64 FrontendConversion { frontend: String, error: String },
65 #[error("Could not write state to file: io_kind={:?} os_code={:?}", .0.kind(), .0.raw_os_error())]
66 FileError(std::io::Error),
67 #[error("Invalid value for field '{field}': {reason}")]
68 InvalidValue {
69 field: &'static str,
70 reason: &'static str,
71 },
72 #[error("invalid TCP frontend for address {address}: reason_bytes={}", .reason.len())]
78 InvalidTcpFrontend { address: SocketAddr, reason: String },
79}
80
81impl fmt::Debug for StateError {
82 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
83 write!(f, "StateError({self})")
84 }
85}
86
87fn certificate_error_kind(error: &CertificateError) -> &'static str {
88 match error {
89 CertificateError::ParsePEMCertificate(_) => "parse_pem_certificate",
90 CertificateError::ParseX509Certificate(_) => "parse_x509_certificate",
91 CertificateError::InvalidTlsVersion(_) => "invalid_tls_version",
92 CertificateError::InvalidFingerprint(_) => "invalid_fingerprint",
93 CertificateError::LoadFile { .. } => "load_file",
94 CertificateError::DecodeError(_) => "decode_error",
95 }
96}
97
98#[derive(Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
107pub struct ConfigState {
108 pub clusters: BTreeMap<ClusterId, Cluster>,
109 pub backends: BTreeMap<ClusterId, Vec<Backend>>,
110 pub http_listeners: BTreeMap<SocketAddr, HttpListenerConfig>,
112 pub https_listeners: BTreeMap<SocketAddr, HttpsListenerConfig>,
114 pub tcp_listeners: BTreeMap<SocketAddr, TcpListenerConfig>,
116 pub udp_listeners: BTreeMap<SocketAddr, UdpListenerConfig>,
118 pub http_fronts: BTreeMap<String, HttpFrontend>,
121 pub https_fronts: BTreeMap<String, HttpFrontend>,
123 pub tcp_fronts: HashMap<ClusterId, Vec<TcpFrontend>>,
124 pub udp_fronts: HashMap<ClusterId, Vec<UdpFrontend>>,
125 pub certificates: HashMap<SocketAddr, HashMap<Fingerprint, CertificateAndKey>>,
126 pub request_counts: BTreeMap<String, i32>,
128}
129
130impl std::fmt::Debug for ConfigState {
131 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
132 let backends_count = self
133 .backends
134 .values()
135 .map(Vec::len)
136 .fold(0usize, usize::saturating_add);
137 let tcp_frontends_count = self
138 .tcp_fronts
139 .values()
140 .map(Vec::len)
141 .fold(0usize, usize::saturating_add);
142 let udp_frontends_count = self
143 .udp_fronts
144 .values()
145 .map(Vec::len)
146 .fold(0usize, usize::saturating_add);
147 let certificates_count = self
148 .certificates
149 .values()
150 .map(HashMap::len)
151 .fold(0usize, usize::saturating_add);
152
153 f.debug_struct("ConfigState")
154 .field("clusters_count", &self.clusters.len())
155 .field("backend_clusters_count", &self.backends.len())
156 .field("backends_count", &backends_count)
157 .field("http_listeners_count", &self.http_listeners.len())
158 .field("https_listeners_count", &self.https_listeners.len())
159 .field("tcp_listeners_count", &self.tcp_listeners.len())
160 .field("udp_listeners_count", &self.udp_listeners.len())
161 .field("http_frontends_count", &self.http_fronts.len())
162 .field("https_frontends_count", &self.https_fronts.len())
163 .field("tcp_frontend_clusters_count", &self.tcp_fronts.len())
164 .field("tcp_frontends_count", &tcp_frontends_count)
165 .field("udp_frontend_clusters_count", &self.udp_fronts.len())
166 .field("udp_frontends_count", &udp_frontends_count)
167 .field("certificate_addresses_count", &self.certificates.len())
168 .field("certificates_count", &certificates_count)
169 .field("request_counts_count", &self.request_counts.len())
170 .finish_non_exhaustive()
171 }
172}
173
174impl ConfigState {
175 pub fn new() -> Self {
176 Self::default()
177 }
178
179 pub fn dispatch(&mut self, request: &Request) -> Result<(), StateError> {
180 let request_type = match &request.request_type {
181 Some(t) => t,
182 None => return Err(StateError::EmptyRequest),
183 };
184
185 self.increment_request_count(request);
186
187 let result = match request_type {
188 RequestType::AddCluster(cluster) => self.add_cluster(cluster),
189 RequestType::RemoveCluster(cluster_id) => self.remove_cluster(cluster_id),
190 RequestType::AddHttpListener(listener) => self.add_http_listener(listener),
191 RequestType::AddHttpsListener(listener) => self.add_https_listener(listener),
192 RequestType::AddTcpListener(listener) => self.add_tcp_listener(listener),
193 RequestType::AddUdpListener(listener) => self.add_udp_listener(listener),
194 RequestType::RemoveListener(remove) => self.remove_listener(remove),
195 RequestType::ActivateListener(activate) => self.activate_listener(activate),
196 RequestType::DeactivateListener(deactivate) => self.deactivate_listener(deactivate),
197 RequestType::AddHttpFrontend(front) => self.add_http_frontend(front),
198 RequestType::RemoveHttpFrontend(front) => self.remove_http_frontend(front),
199 RequestType::AddCertificate(add) => self.add_certificate(add),
200 RequestType::RemoveCertificate(remove) => self.remove_certificate(remove),
201 RequestType::ReplaceCertificate(replace) => self.replace_certificate(replace),
202 RequestType::AddHttpsFrontend(front) => self.add_https_frontend(front),
203 RequestType::RemoveHttpsFrontend(front) => self.remove_https_frontend(front),
204 RequestType::AddTcpFrontend(front) => self.add_tcp_frontend(front),
205 RequestType::RemoveTcpFrontend(front) => self.remove_tcp_frontend(front),
206 RequestType::AddUdpFrontend(front) => self.add_udp_frontend(front),
207 RequestType::RemoveUdpFrontend(front) => self.remove_udp_frontend(front),
208 RequestType::AddBackend(add_backend) => self.add_backend(add_backend),
209 RequestType::RemoveBackend(backend) => self.remove_backend(backend),
210 RequestType::UpdateHttpListener(patch) => self.update_http_listener(patch),
211 RequestType::UpdateHttpsListener(patch) => self.update_https_listener(patch),
212 RequestType::UpdateTcpListener(patch) => self.update_tcp_listener(patch),
213 RequestType::UpdateUdpListener(patch) => self.update_udp_listener(patch),
214 RequestType::SetHealthCheck(set) => self.set_health_check(set),
215 RequestType::RemoveHealthCheck(cluster_id) => self.remove_health_check(cluster_id),
216
217 RequestType::Logging(_)
224 | RequestType::CountRequests(_)
225 | RequestType::Status(_)
226 | RequestType::SoftStop(_)
227 | RequestType::QueryCertificatesFromWorkers(_)
228 | RequestType::QueryClusterById(_)
229 | RequestType::QueryClustersByDomain(_)
230 | RequestType::QueryMetrics(_)
231 | RequestType::QueryClustersHashes(_)
232 | RequestType::ConfigureMetrics(_)
233 | RequestType::SetMetricDetail(_)
234 | RequestType::ReturnListenSockets(_)
235 | RequestType::SetMaxConnectionsPerIp(_)
236 | RequestType::QueryMaxConnectionsPerIp(_)
237 | RequestType::HardStop(_) => Ok(()),
238
239 _other_request => Err(StateError::UndispatchableRequest),
240 };
241
242 #[cfg(debug_assertions)]
248 self.check_invariants();
249
250 result
251 }
252
253 #[cfg(debug_assertions)]
264 fn check_invariants(&self) {
265 for (addr, listener) in &self.http_listeners {
269 debug_assert_eq!(
270 SocketAddr::from(listener.address),
271 *addr,
272 "http_listener value address must match its map key"
273 );
274 }
275 for (addr, listener) in &self.https_listeners {
276 debug_assert_eq!(
277 SocketAddr::from(listener.address),
278 *addr,
279 "https_listener value address must match its map key"
280 );
281 }
282 for (addr, listener) in &self.tcp_listeners {
283 debug_assert_eq!(
284 SocketAddr::from(listener.address),
285 *addr,
286 "tcp_listener value address must match its map key"
287 );
288 }
289
290 for (cluster_id, cluster) in &self.clusters {
293 debug_assert_eq!(
294 &cluster.cluster_id, cluster_id,
295 "cluster value cluster_id must match its map key"
296 );
297 }
298
299 for (cluster_id, backends) in &self.backends {
304 for backend in backends {
305 debug_assert_eq!(
306 &backend.cluster_id, cluster_id,
307 "backend cluster_id must match its bucket key"
308 );
309 }
310 let unique: HashSet<(&String, &SocketAddr)> = backends
311 .iter()
312 .map(|b| (&b.backend_id, &b.address))
313 .collect();
314 debug_assert_eq!(
315 unique.len(),
316 backends.len(),
317 "backends within a cluster must be unique on (backend_id, address)"
318 );
319 }
320
321 for (cluster_id, fronts) in &self.tcp_fronts {
325 for front in fronts {
326 debug_assert_eq!(
327 &front.cluster_id, cluster_id,
328 "tcp_frontend cluster_id must match its bucket key"
329 );
330 }
331 let unique: HashSet<&TcpFrontend> = fronts.iter().collect();
332 debug_assert_eq!(
333 unique.len(),
334 fronts.len(),
335 "tcp frontends within a cluster must be unique"
336 );
337 }
338
339 let raw_frontends = self.http_fronts.len()
350 + self.https_fronts.len()
351 + self.count_tcp_frontends_raw()
352 + self.udp_fronts.values().map(|v| v.len()).sum::<usize>();
353 debug_assert_eq!(
354 self.count_frontends(),
355 raw_frontends,
356 "count_frontends must equal the sum of all frontend map entries"
357 );
358 let raw_backends: usize = self.backends.values().map(|v| v.len()).sum();
359 debug_assert_eq!(
360 self.count_backends(),
361 raw_backends,
362 "count_backends must equal the sum of all backend Vec lengths"
363 );
364 }
365
366 fn count_tcp_frontends_raw(&self) -> usize {
374 self.tcp_fronts.values().map(|v| v.len()).sum()
375 }
376
377 fn increment_request_count(&mut self, request: &Request) {
379 if let Some(request_type) = &request.request_type {
380 let count = self
381 .request_counts
382 .entry(format_request_type(request_type).to_owned())
383 .or_insert(1);
384 *count += 1;
385 }
386 }
387
388 pub fn get_request_counts(&self) -> RequestCounts {
389 RequestCounts {
390 map: self.request_counts.clone(),
391 }
392 }
393
394 fn add_cluster(&mut self, cluster: &Cluster) -> Result<(), StateError> {
395 if let Some(hc) = cluster.health_check.as_ref()
402 && let Err(reason) = crate::config::validate_health_check_config(hc)
403 {
404 return Err(StateError::InvalidValue {
405 field: "health_check",
406 reason,
407 });
408 }
409 let cluster = cluster.clone();
410 let cluster_id = cluster.cluster_id.clone();
414 self.clusters.insert(cluster_id.clone(), cluster);
415 debug_assert!(
416 self.clusters.contains_key(&cluster_id),
417 "add_cluster must leave the cluster present in the map"
418 );
419 debug_assert_eq!(
420 self.clusters.get(&cluster_id).map(|c| &c.cluster_id),
421 Some(&cluster_id),
422 "stored cluster must be keyed by its own cluster_id"
423 );
424 Ok(())
425 }
426
427 fn remove_cluster(&mut self, cluster_id: &str) -> Result<(), StateError> {
428 let before = self.clusters.len();
429 match self.clusters.remove(cluster_id) {
430 Some(_) => {
431 debug_assert!(
432 !self.clusters.contains_key(cluster_id),
433 "remove_cluster must evict the cluster"
434 );
435 debug_assert_eq!(
436 self.clusters.len(),
437 before - 1,
438 "remove_cluster must drop exactly one entry"
439 );
440 Ok(())
441 }
442 None => {
443 debug_assert_eq!(
444 self.clusters.len(),
445 before,
446 "a failed remove_cluster must not mutate the map"
447 );
448 Err(StateError::NotFound {
449 kind: ObjectKind::Cluster,
450 id: cluster_id.to_owned(),
451 })
452 }
453 }
454 }
455
456 fn set_health_check(&mut self, set: &SetHealthCheck) -> Result<(), StateError> {
457 if let Err(reason) = crate::config::validate_health_check_config(&set.config) {
464 return Err(StateError::InvalidValue {
465 field: "health_check",
466 reason,
467 });
468 }
469 match self.clusters.get_mut(&set.cluster_id) {
470 Some(cluster) => {
471 cluster.health_check = Some(set.config.to_owned());
472 Ok(())
473 }
474 None => Err(StateError::NotFound {
475 kind: ObjectKind::Cluster,
476 id: set.cluster_id.to_owned(),
477 }),
478 }
479 }
480
481 fn remove_health_check(&mut self, cluster_id: &str) -> Result<(), StateError> {
482 match self.clusters.get_mut(cluster_id) {
483 Some(cluster) => {
484 cluster.health_check = None;
485 Ok(())
486 }
487 None => Err(StateError::NotFound {
488 kind: ObjectKind::Cluster,
489 id: cluster_id.to_owned(),
490 }),
491 }
492 }
493
494 pub fn list_health_checks(&self, cluster_id: Option<&str>) -> HealthChecksList {
495 let map = self
496 .clusters
497 .iter()
498 .filter(|(id, _)| cluster_id.is_none_or(|filter| filter == id.as_str()))
499 .filter_map(|(id, cluster)| {
500 cluster
501 .health_check
502 .as_ref()
503 .map(|hc| (id.to_owned(), hc.to_owned()))
504 })
505 .collect();
506 HealthChecksList { map }
507 }
508
509 fn add_http_listener(&mut self, listener: &HttpListenerConfig) -> Result<(), StateError> {
510 let address: SocketAddr = listener.address.into();
511 let before = self.http_listeners.len();
512 match self.http_listeners.entry(address) {
513 BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(listener.clone()),
514 BTreeMapEntry::Occupied(_) => {
515 debug_assert_eq!(
516 self.http_listeners.len(),
517 before,
518 "a rejected duplicate add_http_listener must not mutate the map"
519 );
520 return Err(StateError::Exists {
521 kind: ObjectKind::HttpListener,
522 id: address.to_string(),
523 });
524 }
525 };
526 debug_assert!(
527 self.http_listeners.contains_key(&address),
528 "add_http_listener must insert the listener under its address"
529 );
530 debug_assert_eq!(
531 self.http_listeners.len(),
532 before + 1,
533 "add_http_listener inserts exactly one entry on the vacant path"
534 );
535 Ok(())
536 }
537
538 fn add_https_listener(&mut self, listener: &HttpsListenerConfig) -> Result<(), StateError> {
539 let address: SocketAddr = listener.address.into();
540 let before = self.https_listeners.len();
541 match self.https_listeners.entry(address) {
542 BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(listener.clone()),
543 BTreeMapEntry::Occupied(_) => {
544 debug_assert_eq!(
545 self.https_listeners.len(),
546 before,
547 "a rejected duplicate add_https_listener must not mutate the map"
548 );
549 return Err(StateError::Exists {
550 kind: ObjectKind::HttpsListener,
551 id: address.to_string(),
552 });
553 }
554 };
555 debug_assert!(
556 self.https_listeners.contains_key(&address),
557 "add_https_listener must insert the listener under its address"
558 );
559 debug_assert_eq!(
560 self.https_listeners.len(),
561 before + 1,
562 "add_https_listener inserts exactly one entry on the vacant path"
563 );
564 Ok(())
565 }
566
567 fn add_tcp_listener(&mut self, listener: &TcpListenerConfig) -> Result<(), StateError> {
568 let address: SocketAddr = listener.address.into();
569 let before = self.tcp_listeners.len();
570 match self.tcp_listeners.entry(address) {
571 BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(*listener),
572 BTreeMapEntry::Occupied(_) => {
573 debug_assert_eq!(
574 self.tcp_listeners.len(),
575 before,
576 "a rejected duplicate add_tcp_listener must not mutate the map"
577 );
578 return Err(StateError::Exists {
579 kind: ObjectKind::TcpListener,
580 id: address.to_string(),
581 });
582 }
583 };
584 debug_assert!(
585 self.tcp_listeners.contains_key(&address),
586 "add_tcp_listener must insert the listener under its address"
587 );
588 debug_assert_eq!(
589 self.tcp_listeners.len(),
590 before + 1,
591 "add_tcp_listener inserts exactly one entry on the vacant path"
592 );
593 Ok(())
594 }
595
596 fn add_udp_listener(&mut self, listener: &UdpListenerConfig) -> Result<(), StateError> {
597 let address: SocketAddr = listener.address.into();
598 match self.udp_listeners.entry(address) {
599 BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(*listener),
600 BTreeMapEntry::Occupied(_) => {
601 return Err(StateError::Exists {
602 kind: ObjectKind::UdpListener,
603 id: address.to_string(),
604 });
605 }
606 };
607 Ok(())
608 }
609
610 fn remove_listener(&mut self, remove: &RemoveListener) -> Result<(), StateError> {
611 match ListenerType::try_from(remove.proxy).map_err(StateError::WrongFieldValue)? {
612 ListenerType::Http => self.remove_http_listener(&remove.address.into()),
613 ListenerType::Https => self.remove_https_listener(&remove.address.into()),
614 ListenerType::Tcp => self.remove_tcp_listener(&remove.address.into()),
615 ListenerType::Udp => self.remove_udp_listener(&remove.address.into()),
616 }
617 }
618
619 fn remove_http_listener(&mut self, address: &SocketAddr) -> Result<(), StateError> {
620 let before = self.http_listeners.len();
621 if self.http_listeners.remove(address).is_none() {
622 debug_assert_eq!(
623 self.http_listeners.len(),
624 before,
625 "a failed remove_http_listener must not mutate the map"
626 );
627 return Err(StateError::NoChange);
628 }
629 debug_assert!(
630 !self.http_listeners.contains_key(address),
631 "remove_http_listener must evict the address"
632 );
633 debug_assert_eq!(
634 self.http_listeners.len(),
635 before - 1,
636 "remove_http_listener drops exactly one entry"
637 );
638 Ok(())
639 }
640
641 fn remove_https_listener(&mut self, address: &SocketAddr) -> Result<(), StateError> {
642 let before = self.https_listeners.len();
643 if self.https_listeners.remove(address).is_none() {
644 debug_assert_eq!(
645 self.https_listeners.len(),
646 before,
647 "a failed remove_https_listener must not mutate the map"
648 );
649 return Err(StateError::NoChange);
650 }
651 debug_assert!(
652 !self.https_listeners.contains_key(address),
653 "remove_https_listener must evict the address"
654 );
655 debug_assert_eq!(
656 self.https_listeners.len(),
657 before - 1,
658 "remove_https_listener drops exactly one entry"
659 );
660 Ok(())
661 }
662
663 fn remove_tcp_listener(&mut self, address: &SocketAddr) -> Result<(), StateError> {
664 let before = self.tcp_listeners.len();
665 if self.tcp_listeners.remove(address).is_none() {
666 debug_assert_eq!(
667 self.tcp_listeners.len(),
668 before,
669 "a failed remove_tcp_listener must not mutate the map"
670 );
671 return Err(StateError::NoChange);
672 }
673 debug_assert!(
674 !self.tcp_listeners.contains_key(address),
675 "remove_tcp_listener must evict the address"
676 );
677 debug_assert_eq!(
678 self.tcp_listeners.len(),
679 before - 1,
680 "remove_tcp_listener drops exactly one entry"
681 );
682 Ok(())
683 }
684
685 fn remove_udp_listener(&mut self, address: &SocketAddr) -> Result<(), StateError> {
686 if self.udp_listeners.remove(address).is_none() {
687 return Err(StateError::NoChange);
688 }
689 Ok(())
690 }
691
692 fn update_http_listener(&mut self, patch: &UpdateHttpListenerConfig) -> Result<(), StateError> {
699 validate_h2_flood_knobs_http(patch)?;
700
701 let address: SocketAddr = patch.address.into();
702 let listener =
703 self.http_listeners
704 .get_mut(&address)
705 .ok_or_else(|| StateError::NotFound {
706 kind: ObjectKind::HttpListener,
707 id: address.to_string(),
708 })?;
709
710 if let Some(v) = patch.public_address {
712 listener.public_address = Some(v);
713 }
714 if let Some(v) = patch.expect_proxy {
715 listener.expect_proxy = v;
716 }
717 if let Some(ref v) = patch.sticky_name {
718 listener.sticky_name = v.to_owned();
719 }
720 if let Some(v) = patch.front_timeout {
721 listener.front_timeout = v;
722 }
723 if let Some(v) = patch.back_timeout {
724 listener.back_timeout = v;
725 }
726 if let Some(v) = patch.connect_timeout {
727 listener.connect_timeout = v;
728 }
729 if let Some(v) = patch.request_timeout {
730 listener.request_timeout = v;
731 }
732 if let Some(patch_answers) = patch.http_answers.as_ref() {
733 merge_custom_http_answers(&mut listener.http_answers, patch_answers);
734 }
735 if let Some(v) = patch.h2_max_rst_stream_per_window {
737 listener.h2_max_rst_stream_per_window = Some(v);
738 }
739 if let Some(v) = patch.h2_max_ping_per_window {
740 listener.h2_max_ping_per_window = Some(v);
741 }
742 if let Some(v) = patch.h2_max_settings_per_window {
743 listener.h2_max_settings_per_window = Some(v);
744 }
745 if let Some(v) = patch.h2_max_empty_data_per_window {
746 listener.h2_max_empty_data_per_window = Some(v);
747 }
748 if let Some(v) = patch.h2_max_continuation_frames {
749 listener.h2_max_continuation_frames = Some(v);
750 }
751 if let Some(v) = patch.h2_max_glitch_count {
752 listener.h2_max_glitch_count = Some(v);
753 }
754 if let Some(v) = patch.h2_initial_connection_window {
755 listener.h2_initial_connection_window = Some(v);
756 }
757 if let Some(v) = patch.h2_max_concurrent_streams {
758 listener.h2_max_concurrent_streams = Some(v);
759 }
760 if let Some(v) = patch.h2_stream_shrink_ratio {
761 listener.h2_stream_shrink_ratio = Some(v);
762 }
763 if let Some(v) = patch.h2_max_rst_stream_lifetime {
764 listener.h2_max_rst_stream_lifetime = Some(v);
765 }
766 if let Some(v) = patch.h2_max_rst_stream_abusive_lifetime {
767 listener.h2_max_rst_stream_abusive_lifetime = Some(v);
768 }
769 if let Some(v) = patch.h2_max_rst_stream_emitted_lifetime {
770 listener.h2_max_rst_stream_emitted_lifetime = Some(v);
771 }
772 if let Some(v) = patch.h2_max_header_list_size {
773 listener.h2_max_header_list_size = Some(v);
774 }
775 if let Some(v) = patch.h2_max_header_table_size {
776 listener.h2_max_header_table_size = Some(v);
777 }
778 if let Some(v) = patch.h2_max_header_fields {
779 listener.h2_max_header_fields = Some(v);
780 }
781 if let Some(v) = patch.h2_stream_idle_timeout_seconds {
782 listener.h2_stream_idle_timeout_seconds = Some(v);
783 }
784 if let Some(v) = patch.h2_graceful_shutdown_deadline_seconds {
786 listener.h2_graceful_shutdown_deadline_seconds = Some(v);
787 }
788 if let Some(v) = patch.h2_max_window_update_stream0_per_window {
789 listener.h2_max_window_update_stream0_per_window = Some(v);
790 }
791 if let Some(ref v) = patch.sozu_id_header {
792 validate_sozu_id_header(v)?;
793 listener.sozu_id_header = Some(v.to_owned());
794 }
795 Ok(())
796 }
797
798 fn update_https_listener(
805 &mut self,
806 patch: &UpdateHttpsListenerConfig,
807 ) -> Result<(), StateError> {
808 validate_h2_flood_knobs_https(patch)?;
809
810 let address: SocketAddr = patch.address.into();
811 let listener =
812 self.https_listeners
813 .get_mut(&address)
814 .ok_or_else(|| StateError::NotFound {
815 kind: ObjectKind::HttpsListener,
816 id: address.to_string(),
817 })?;
818
819 if let Some(v) = patch.public_address {
821 listener.public_address = Some(v);
822 }
823 if let Some(v) = patch.expect_proxy {
824 listener.expect_proxy = v;
825 }
826 if let Some(ref v) = patch.sticky_name {
827 listener.sticky_name = v.to_owned();
828 }
829 if let Some(v) = patch.front_timeout {
830 listener.front_timeout = v;
831 }
832 if let Some(v) = patch.back_timeout {
833 listener.back_timeout = v;
834 }
835 if let Some(v) = patch.connect_timeout {
836 listener.connect_timeout = v;
837 }
838 if let Some(v) = patch.request_timeout {
839 listener.request_timeout = v;
840 }
841 if let Some(patch_answers) = patch.http_answers.as_ref() {
842 merge_custom_http_answers(&mut listener.http_answers, patch_answers);
843 }
844 if let Some(ref alpn_wrapper) = patch.alpn_protocols {
846 validate_alpn_protocols(&alpn_wrapper.values)?;
847 listener.alpn_protocols = alpn_wrapper.values.clone();
849 }
850 if let Some(v) = patch.strict_sni_binding {
851 listener.strict_sni_binding = Some(v);
852 }
853 if let Some(v) = patch.disable_http11 {
854 listener.disable_http11 = Some(v);
855 }
856 if let Some(v) = patch.h2_max_rst_stream_per_window {
858 listener.h2_max_rst_stream_per_window = Some(v);
859 }
860 if let Some(v) = patch.h2_max_ping_per_window {
861 listener.h2_max_ping_per_window = Some(v);
862 }
863 if let Some(v) = patch.h2_max_settings_per_window {
864 listener.h2_max_settings_per_window = Some(v);
865 }
866 if let Some(v) = patch.h2_max_empty_data_per_window {
867 listener.h2_max_empty_data_per_window = Some(v);
868 }
869 if let Some(v) = patch.h2_max_continuation_frames {
870 listener.h2_max_continuation_frames = Some(v);
871 }
872 if let Some(v) = patch.h2_max_glitch_count {
873 listener.h2_max_glitch_count = Some(v);
874 }
875 if let Some(v) = patch.h2_initial_connection_window {
876 listener.h2_initial_connection_window = Some(v);
877 }
878 if let Some(v) = patch.h2_max_concurrent_streams {
879 listener.h2_max_concurrent_streams = Some(v);
880 }
881 if let Some(v) = patch.h2_stream_shrink_ratio {
882 listener.h2_stream_shrink_ratio = Some(v);
883 }
884 if let Some(v) = patch.h2_max_rst_stream_lifetime {
885 listener.h2_max_rst_stream_lifetime = Some(v);
886 }
887 if let Some(v) = patch.h2_max_rst_stream_abusive_lifetime {
888 listener.h2_max_rst_stream_abusive_lifetime = Some(v);
889 }
890 if let Some(v) = patch.h2_max_rst_stream_emitted_lifetime {
891 listener.h2_max_rst_stream_emitted_lifetime = Some(v);
892 }
893 if let Some(v) = patch.h2_max_header_list_size {
894 listener.h2_max_header_list_size = Some(v);
895 }
896 if let Some(v) = patch.h2_max_header_table_size {
897 listener.h2_max_header_table_size = Some(v);
898 }
899 if let Some(v) = patch.h2_max_header_fields {
900 listener.h2_max_header_fields = Some(v);
901 }
902 if let Some(v) = patch.h2_stream_idle_timeout_seconds {
903 listener.h2_stream_idle_timeout_seconds = Some(v);
904 }
905 if let Some(v) = patch.h2_graceful_shutdown_deadline_seconds {
907 listener.h2_graceful_shutdown_deadline_seconds = Some(v);
908 }
909 if let Some(v) = patch.h2_max_window_update_stream0_per_window {
910 listener.h2_max_window_update_stream0_per_window = Some(v);
911 }
912 if let Some(ref v) = patch.sozu_id_header {
913 validate_sozu_id_header(v)?;
914 listener.sozu_id_header = Some(v.to_owned());
915 }
916 Ok(())
917 }
918
919 fn update_tcp_listener(&mut self, patch: &UpdateTcpListenerConfig) -> Result<(), StateError> {
924 let address: SocketAddr = patch.address.into();
925 let listener =
926 self.tcp_listeners
927 .get_mut(&address)
928 .ok_or_else(|| StateError::NotFound {
929 kind: ObjectKind::TcpListener,
930 id: address.to_string(),
931 })?;
932
933 if let Some(v) = patch.public_address {
934 listener.public_address = Some(v);
935 }
936 if let Some(v) = patch.expect_proxy {
937 listener.expect_proxy = v;
938 }
939 if let Some(v) = patch.front_timeout {
940 listener.front_timeout = v;
941 }
942 if let Some(v) = patch.back_timeout {
943 listener.back_timeout = v;
944 }
945 if let Some(v) = patch.connect_timeout {
946 listener.connect_timeout = v;
947 }
948 Ok(())
949 }
950
951 fn update_udp_listener(&mut self, patch: &UpdateUdpListenerConfig) -> Result<(), StateError> {
956 let address: SocketAddr = patch.address.into();
957 let listener =
958 self.udp_listeners
959 .get_mut(&address)
960 .ok_or_else(|| StateError::NotFound {
961 kind: ObjectKind::UdpListener,
962 id: address.to_string(),
963 })?;
964
965 if let Some(v) = patch.public_address {
966 listener.public_address = Some(v);
967 }
968 if let Some(v) = patch.front_timeout {
969 listener.front_timeout = v;
970 }
971 if let Some(v) = patch.back_timeout {
972 listener.back_timeout = v;
973 }
974 if let Some(v) = patch.max_rx_datagram_size {
975 listener.max_rx_datagram_size = v;
976 }
977 if let Some(v) = patch.max_flows {
978 listener.max_flows = v;
979 }
980 Ok(())
981 }
982
983 fn activate_listener(&mut self, activate: &ActivateListener) -> Result<(), StateError> {
984 match ListenerType::try_from(activate.proxy).map_err(StateError::WrongFieldValue)? {
985 ListenerType::Http => self
986 .http_listeners
987 .get_mut(&activate.address.into())
988 .map(|listener| listener.active = true)
989 .ok_or(StateError::NotFound {
990 kind: ObjectKind::HttpListener,
991 id: activate.address.to_string(),
992 }),
993 ListenerType::Https => self
994 .https_listeners
995 .get_mut(&activate.address.into())
996 .map(|listener| listener.active = true)
997 .ok_or(StateError::NotFound {
998 kind: ObjectKind::HttpsListener,
999 id: activate.address.to_string(),
1000 }),
1001 ListenerType::Tcp => self
1002 .tcp_listeners
1003 .get_mut(&activate.address.into())
1004 .map(|listener| listener.active = true)
1005 .ok_or(StateError::NotFound {
1006 kind: ObjectKind::TcpListener,
1007 id: activate.address.to_string(),
1008 }),
1009 ListenerType::Udp => self
1010 .udp_listeners
1011 .get_mut(&activate.address.into())
1012 .map(|listener| listener.active = true)
1013 .ok_or(StateError::NotFound {
1014 kind: ObjectKind::UdpListener,
1015 id: activate.address.to_string(),
1016 }),
1017 }
1018 }
1019
1020 fn deactivate_listener(&mut self, deactivate: &DeactivateListener) -> Result<(), StateError> {
1021 match ListenerType::try_from(deactivate.proxy).map_err(StateError::WrongFieldValue)? {
1022 ListenerType::Http => self
1023 .http_listeners
1024 .get_mut(&deactivate.address.into())
1025 .map(|listener| listener.active = false)
1026 .ok_or(StateError::NotFound {
1027 kind: ObjectKind::HttpListener,
1028 id: deactivate.address.to_string(),
1029 }),
1030 ListenerType::Https => self
1031 .https_listeners
1032 .get_mut(&deactivate.address.into())
1033 .map(|listener| listener.active = false)
1034 .ok_or(StateError::NotFound {
1035 kind: ObjectKind::HttpsListener,
1036 id: deactivate.address.to_string(),
1037 }),
1038 ListenerType::Tcp => self
1039 .tcp_listeners
1040 .get_mut(&deactivate.address.into())
1041 .map(|listener| listener.active = false)
1042 .ok_or(StateError::NotFound {
1043 kind: ObjectKind::TcpListener,
1044 id: deactivate.address.to_string(),
1045 }),
1046 ListenerType::Udp => self
1047 .udp_listeners
1048 .get_mut(&deactivate.address.into())
1049 .map(|listener| listener.active = false)
1050 .ok_or(StateError::NotFound {
1051 kind: ObjectKind::UdpListener,
1052 id: deactivate.address.to_string(),
1053 }),
1054 }
1055 }
1056
1057 fn add_http_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1058 let front_as_key = front.to_string();
1059 let before = self.http_fronts.len();
1060
1061 match self.http_fronts.entry(front.to_string()) {
1062 BTreeMapEntry::Vacant(e) => {
1063 e.insert(front.clone().to_frontend().map_err(|into_error| {
1064 StateError::FrontendConversion {
1065 frontend: front_as_key,
1066 error: into_error.to_string(),
1067 }
1068 })?)
1069 }
1070 BTreeMapEntry::Occupied(_) => {
1071 debug_assert_eq!(
1072 self.http_fronts.len(),
1073 before,
1074 "a rejected duplicate add_http_frontend must not mutate the map"
1075 );
1076 return Err(StateError::Exists {
1077 kind: ObjectKind::HttpFrontend,
1078 id: front.to_string(),
1079 });
1080 }
1081 };
1082 debug_assert!(
1085 self.http_fronts.contains_key(&front.to_string()),
1086 "add_http_frontend must insert the route key on success"
1087 );
1088 debug_assert_eq!(
1089 self.http_fronts.len(),
1090 before + 1,
1091 "add_http_frontend inserts exactly one entry on success"
1092 );
1093 Ok(())
1094 }
1095
1096 fn add_https_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1097 let front_as_key = front.to_string();
1098 let before = self.https_fronts.len();
1099
1100 match self.https_fronts.entry(front.to_string()) {
1101 BTreeMapEntry::Vacant(e) => {
1102 e.insert(front.clone().to_frontend().map_err(|into_error| {
1103 StateError::FrontendConversion {
1104 frontend: front_as_key,
1105 error: into_error.to_string(),
1106 }
1107 })?)
1108 }
1109 BTreeMapEntry::Occupied(_) => {
1110 debug_assert_eq!(
1111 self.https_fronts.len(),
1112 before,
1113 "a rejected duplicate add_https_frontend must not mutate the map"
1114 );
1115 return Err(StateError::Exists {
1116 kind: ObjectKind::HttpsFrontend,
1117 id: front.to_string(),
1118 });
1119 }
1120 };
1121 debug_assert!(
1122 self.https_fronts.contains_key(&front.to_string()),
1123 "add_https_frontend must insert the route key on success"
1124 );
1125 debug_assert_eq!(
1126 self.https_fronts.len(),
1127 before + 1,
1128 "add_https_frontend inserts exactly one entry on success"
1129 );
1130 Ok(())
1131 }
1132
1133 fn remove_http_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1134 let key = front.to_string();
1135 let before = self.http_fronts.len();
1136 self.http_fronts.remove(&key).ok_or(StateError::NotFound {
1137 kind: ObjectKind::HttpFrontend,
1138 id: front.to_string(),
1139 })?;
1140 debug_assert!(
1141 !self.http_fronts.contains_key(&key),
1142 "remove_http_frontend must evict the route key"
1143 );
1144 debug_assert_eq!(
1145 self.http_fronts.len(),
1146 before - 1,
1147 "remove_http_frontend drops exactly one entry"
1148 );
1149 Ok(())
1150 }
1151
1152 fn remove_https_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1153 let key = front.to_string();
1154 let before = self.https_fronts.len();
1155 self.https_fronts.remove(&key).ok_or(StateError::NotFound {
1156 kind: ObjectKind::HttpsFrontend,
1157 id: front.to_string(),
1158 })?;
1159 debug_assert!(
1160 !self.https_fronts.contains_key(&key),
1161 "remove_https_frontend must evict the route key"
1162 );
1163 debug_assert_eq!(
1164 self.https_fronts.len(),
1165 before - 1,
1166 "remove_https_frontend drops exactly one entry"
1167 );
1168 Ok(())
1169 }
1170
1171 fn add_certificate(&mut self, add: &AddCertificate) -> Result<(), StateError> {
1172 let fingerprint = add
1173 .certificate
1174 .fingerprint()
1175 .map_err(StateError::AddCertificate)?;
1176
1177 let entry = self.certificates.entry(add.address.into()).or_default();
1178
1179 let mut add = add.clone();
1180 add.certificate
1181 .apply_overriding_names()
1182 .map_err(StateError::AddCertificate)?;
1183
1184 if entry.contains_key(&fingerprint) {
1185 let names_bytes = add
1186 .certificate
1187 .names
1188 .iter()
1189 .fold(0usize, |total, name| total.saturating_add(name.len()));
1190 info!(
1191 "Skip loading certificate with fingerprint_bytes={} names_count={} names_bytes={} on listener '{}', the certificate is already present.",
1192 fingerprint.0.len(),
1193 add.certificate.names.len(),
1194 names_bytes,
1195 add.address
1196 );
1197 return Ok(());
1198 }
1199
1200 let before = entry.len();
1201 entry.insert(fingerprint.clone(), add.certificate);
1202 debug_assert!(
1203 entry.contains_key(&fingerprint),
1204 "add_certificate must insert the fingerprint under its address"
1205 );
1206 debug_assert_eq!(
1207 entry.len(),
1208 before + 1,
1209 "add_certificate inserts exactly one fingerprint on the new path"
1210 );
1211 Ok(())
1212 }
1213
1214 fn remove_certificate(&mut self, remove: &RemoveCertificate) -> Result<(), StateError> {
1215 let fingerprint = Fingerprint(
1216 hex::decode(&remove.fingerprint)
1217 .map_err(|decode_error| StateError::RemoveCertificate(decode_error.to_string()))?,
1218 );
1219
1220 if let Some(index) = self.certificates.get_mut(&remove.address.into()) {
1221 index.remove(&fingerprint);
1222 debug_assert!(
1223 !index.contains_key(&fingerprint),
1224 "remove_certificate must evict the fingerprint when the address is known"
1225 );
1226 }
1227
1228 Ok(())
1229 }
1230
1231 fn replace_certificate(&mut self, replace: &ReplaceCertificate) -> Result<(), StateError> {
1236 let replace_address = replace.address.into();
1237 let old_fingerprint = Fingerprint(
1238 hex::decode(&replace.old_fingerprint)
1239 .map_err(|decode_error| StateError::RemoveCertificate(decode_error.to_string()))?,
1240 );
1241
1242 self.certificates
1243 .get_mut(&replace_address)
1244 .ok_or(StateError::NotFound {
1245 kind: ObjectKind::Certificate,
1246 id: replace.address.to_string(),
1247 })?
1248 .remove(&old_fingerprint);
1249
1250 let new_fingerprint = Fingerprint(
1251 calculate_fingerprint(replace.new_certificate.certificate.as_bytes()).map_err(
1252 |fingerprint_err| StateError::ReplaceCertificate(fingerprint_err.to_string()),
1253 )?,
1254 );
1255
1256 self.certificates
1257 .get_mut(&replace_address)
1258 .map(|certs| certs.insert(new_fingerprint.clone(), replace.new_certificate.clone()));
1259
1260 if !self
1261 .certificates
1262 .get(&replace_address)
1263 .ok_or(StateError::ReplaceCertificate(
1264 "Unlikely error. This entry in the certificate hashmap should be present"
1265 .to_string(),
1266 ))?
1267 .contains_key(&new_fingerprint)
1268 {
1269 return Err(StateError::ReplaceCertificate(format!(
1270 "Failed to insert the new certificate for address {}",
1271 replace.address
1272 )));
1273 }
1274 debug_assert!(
1277 self.certificates
1278 .get(&replace_address)
1279 .is_some_and(|certs| certs.contains_key(&new_fingerprint)),
1280 "replace_certificate must leave the new fingerprint present"
1281 );
1282 debug_assert!(
1283 new_fingerprint == old_fingerprint
1284 || self
1285 .certificates
1286 .get(&replace_address)
1287 .is_none_or(|certs| !certs.contains_key(&old_fingerprint)),
1288 "replace_certificate must evict the old fingerprint unless it equals the new one"
1289 );
1290 Ok(())
1291 }
1292
1293 fn add_tcp_frontend(&mut self, front: &RequestTcpFrontend) -> Result<(), StateError> {
1301 let address: SocketAddr = front.address.into();
1302
1303 let normalized_sni = match &front.sni {
1311 Some(sni) => Some(validate_sni_pattern(sni).map_err(|config_error| {
1312 StateError::InvalidTcpFrontend {
1313 address,
1314 reason: format!("sni {sni:?} failed SNI shape validation: {config_error}"),
1315 }
1316 })?),
1317 None => None,
1318 };
1319 let alpn = canonical_tcp_alpn(&front.alpn);
1320
1321 if normalized_sni.is_none() && !alpn.is_empty() {
1327 return Err(StateError::InvalidTcpFrontend {
1328 address,
1329 reason: format!(
1330 "alpn = {alpn:?} set without sni: alpn only matches within an SNI-scoped \
1331 preread, so a frontend without sni would silently ignore its alpn list"
1332 ),
1333 });
1334 }
1335
1336 let total_before = self.count_tcp_frontends_raw();
1337
1338 let mut listener_has_no_sni_front = false;
1344 let mut listener_has_sni_front = false;
1345 for existing in self
1346 .tcp_fronts
1347 .values()
1348 .flatten()
1349 .filter(|existing| existing.address == address)
1350 {
1351 if tcp_frontend_matches(existing, address, &normalized_sni, &alpn) {
1352 return Err(StateError::Exists {
1353 kind: ObjectKind::TcpFrontend,
1354 id: format!(
1355 "TcpFrontend {{ address: {address}, sni: {normalized_sni:?}, \
1356 alpn: {alpn:?} }}"
1357 ),
1358 });
1359 }
1360 match &existing.sni {
1361 None => listener_has_no_sni_front = true,
1362 Some(existing_sni) => {
1363 listener_has_sni_front = true;
1364 if normalized_sni.as_deref() != Some(existing_sni.as_str()) {
1371 continue;
1372 }
1373 let existing_is_catch_all = existing.alpn.is_empty();
1374 let new_is_catch_all = alpn.is_empty();
1375 if existing_is_catch_all && new_is_catch_all {
1376 return Err(StateError::InvalidTcpFrontend {
1377 address,
1378 reason: format!(
1379 "sni {existing_sni:?} already has a catch-all (empty alpn) \
1380 frontend: at most one frontend per (address, sni) may omit \
1381 alpn"
1382 ),
1383 });
1384 }
1385 if !existing_is_catch_all
1390 && !new_is_catch_all
1391 && let Some(overlap) = alpn
1392 .iter()
1393 .find(|protocol| existing.alpn.contains(protocol))
1394 {
1395 return Err(StateError::InvalidTcpFrontend {
1396 address,
1397 reason: format!(
1398 "sni {existing_sni:?} already has a frontend matching ALPN \
1399 protocol {overlap:?}: ALPN matchers for the same (address, \
1400 sni) must not overlap"
1401 ),
1402 });
1403 }
1404 }
1405 }
1406 }
1407 match &normalized_sni {
1408 None if listener_has_sni_front => {
1409 return Err(StateError::InvalidTcpFrontend {
1410 address,
1411 reason: "a no-SNI frontend cannot be added to a listener that already has \
1412 SNI-scoped routes"
1413 .to_string(),
1414 });
1415 }
1416 Some(_) if listener_has_no_sni_front => {
1417 return Err(StateError::InvalidTcpFrontend {
1418 address,
1419 reason: "an SNI-scoped frontend cannot be added to a listener that already \
1420 has a no-SNI catch-all cluster"
1421 .to_string(),
1422 });
1423 }
1424 _ => {}
1425 }
1426 debug_assert_eq!(
1430 self.count_tcp_frontends_raw(),
1431 total_before,
1432 "add_tcp_frontend must not mutate tcp_fronts before every admission check has \
1433 passed"
1434 );
1435
1436 let tcp_frontend = TcpFrontend {
1437 cluster_id: front.cluster_id.clone(),
1438 address,
1439 tags: front.tags.clone(),
1440 sni: normalized_sni,
1441 alpn,
1442 };
1443 let tcp_frontends = self.tcp_fronts.entry(front.cluster_id.clone()).or_default();
1444 let before = tcp_frontends.len();
1445 debug_assert_eq!(
1446 tcp_frontend.cluster_id, front.cluster_id,
1447 "the built frontend must carry its bucket's cluster_id"
1448 );
1449 tcp_frontends.push(tcp_frontend);
1450 debug_assert_eq!(
1451 tcp_frontends.len(),
1452 before + 1,
1453 "add_tcp_frontend appends exactly one entry"
1454 );
1455 Ok(())
1456 }
1457
1458 fn remove_tcp_frontend(
1459 &mut self,
1460 front_to_remove: &RequestTcpFrontend,
1461 ) -> Result<(), StateError> {
1462 let tcp_frontends =
1463 self.tcp_fronts
1464 .get_mut(&front_to_remove.cluster_id)
1465 .ok_or(StateError::NotFound {
1466 kind: ObjectKind::TcpFrontend,
1467 id: format!(
1468 "RequestTcpFrontend {{ cluster_id: {:?}, address: {:?}, tags: {:?}, sni: {:?}, alpn: {:?} }}",
1469 front_to_remove.cluster_id,
1470 front_to_remove.address,
1471 front_to_remove.tags,
1472 front_to_remove.sni,
1473 front_to_remove.alpn,
1474 ),
1475 })?;
1476
1477 let len = tcp_frontends.len();
1478 let remove_address: SocketAddr = front_to_remove.address.into();
1479 let remove_sni = front_to_remove.sni.as_deref().map(str::to_ascii_lowercase);
1489 let remove_alpn = canonical_tcp_alpn(&front_to_remove.alpn);
1490 tcp_frontends.retain(|front| {
1495 !tcp_frontend_matches(front, remove_address, &remove_sni, &remove_alpn)
1496 });
1497 let after = tcp_frontends.len();
1498 if after == len {
1499 return Err(StateError::NoChange);
1500 }
1501 debug_assert_eq!(
1506 after,
1507 len - 1,
1508 "remove_tcp_frontend drops exactly one entry"
1509 );
1510 debug_assert!(
1511 !tcp_frontends.iter().any(|f| tcp_frontend_matches(
1512 f,
1513 remove_address,
1514 &remove_sni,
1515 &remove_alpn
1516 )),
1517 "remove_tcp_frontend must leave no frontend matching the removed (address, sni, alpn)"
1518 );
1519 Ok(())
1520 }
1521
1522 fn add_udp_frontend(&mut self, front: &RequestUdpFrontend) -> Result<(), StateError> {
1523 let udp_frontends = self.udp_fronts.entry(front.cluster_id.clone()).or_default();
1524
1525 let udp_frontend = UdpFrontend {
1526 cluster_id: front.cluster_id.clone(),
1527 address: front.address.into(),
1528 tags: front.tags.clone(),
1529 };
1530 if udp_frontends.contains(&udp_frontend) {
1531 return Err(StateError::Exists {
1532 kind: ObjectKind::UdpFrontend,
1533 id: format!("{udp_frontend:?}"),
1534 });
1535 }
1536
1537 udp_frontends.push(udp_frontend);
1538 Ok(())
1539 }
1540
1541 fn remove_udp_frontend(
1542 &mut self,
1543 front_to_remove: &RequestUdpFrontend,
1544 ) -> Result<(), StateError> {
1545 let udp_frontends =
1546 self.udp_fronts
1547 .get_mut(&front_to_remove.cluster_id)
1548 .ok_or(StateError::NotFound {
1549 kind: ObjectKind::UdpFrontend,
1550 id: format!("{front_to_remove:?}"),
1551 })?;
1552
1553 let len = udp_frontends.len();
1554 udp_frontends.retain(|front| front.address != front_to_remove.address.into());
1555 if udp_frontends.len() == len {
1556 return Err(StateError::NoChange);
1557 }
1558 Ok(())
1559 }
1560
1561 fn add_backend(&mut self, add_backend: &AddBackend) -> Result<(), StateError> {
1562 let backend = Backend {
1563 address: add_backend.address.into(),
1564 cluster_id: add_backend.cluster_id.clone(),
1565 backend_id: add_backend.backend_id.clone(),
1566 sticky_id: add_backend.sticky_id.clone(),
1567 load_balancing_parameters: add_backend.load_balancing_parameters,
1568 backup: add_backend.backup,
1569 };
1570 let backends = self.backends.entry(backend.cluster_id.clone()).or_default();
1571 let backend_id = backend.backend_id.clone();
1572 let backend_address = backend.address;
1573 let before = backends.len();
1574
1575 let was_present = backends
1580 .iter()
1581 .any(|b| b.backend_id == backend_id && b.address == backend_address);
1582 backends.retain(|b| b.backend_id != backend.backend_id || b.address != backend.address);
1583 debug_assert_eq!(
1584 backends.len(),
1585 before - was_present as usize,
1586 "the upsert retain must drop exactly the prior copy iff it existed"
1587 );
1588 backends.push(backend);
1589 backends.sort();
1590
1591 debug_assert_eq!(
1592 backends.len(),
1593 before + (!was_present) as usize,
1594 "add_backend grows the bucket by one iff the backend was new"
1595 );
1596 debug_assert_eq!(
1597 backends
1598 .iter()
1599 .filter(|b| b.backend_id == backend_id && b.address == backend_address)
1600 .count(),
1601 1,
1602 "exactly one copy of the upserted backend must remain"
1603 );
1604 Ok(())
1605 }
1606
1607 fn remove_backend(&mut self, backend: &RemoveBackend) -> Result<(), StateError> {
1608 let backend_list =
1609 self.backends
1610 .get_mut(&backend.cluster_id)
1611 .ok_or(StateError::NotFound {
1612 kind: ObjectKind::Backend,
1613 id: backend.backend_id.to_owned(),
1614 })?;
1615
1616 let len = backend_list.len();
1617 let remove_address: SocketAddr = backend.address.into();
1618 backend_list.retain(|b| b.backend_id != backend.backend_id || b.address != remove_address);
1619 backend_list.sort();
1620 let after = backend_list.len();
1621 if after == len {
1622 return Err(StateError::NoChange);
1623 }
1624 debug_assert_eq!(after, len - 1, "remove_backend drops exactly one entry");
1627 debug_assert!(
1628 !backend_list
1629 .iter()
1630 .any(|b| b.backend_id == backend.backend_id && b.address == remove_address),
1631 "remove_backend must leave no backend matching (backend_id, address)"
1632 );
1633 Ok(())
1634 }
1635
1636 fn generate_requests(&self) -> Vec<Request> {
1638 let mut v: Vec<Request> = Vec::new();
1639
1640 for listener in self.http_listeners.values() {
1641 v.push(RequestType::AddHttpListener(listener.clone()).into());
1642 if listener.active {
1643 v.push(
1644 RequestType::ActivateListener(ActivateListener {
1645 address: listener.address,
1646 proxy: ListenerType::Http.into(),
1647 from_scm: false,
1648 })
1649 .into(),
1650 );
1651 }
1652 }
1653
1654 for listener in self.https_listeners.values() {
1655 v.push(RequestType::AddHttpsListener(listener.clone()).into());
1656 if listener.active {
1657 v.push(
1658 RequestType::ActivateListener(ActivateListener {
1659 address: listener.address,
1660 proxy: ListenerType::Https.into(),
1661 from_scm: false,
1662 })
1663 .into(),
1664 );
1665 }
1666 }
1667
1668 for listener in self.tcp_listeners.values() {
1669 v.push(RequestType::AddTcpListener(*listener).into());
1670 if listener.active {
1671 v.push(
1672 RequestType::ActivateListener(ActivateListener {
1673 address: listener.address,
1674 proxy: ListenerType::Tcp.into(),
1675 from_scm: false,
1676 })
1677 .into(),
1678 );
1679 }
1680 }
1681
1682 for listener in self.udp_listeners.values() {
1683 v.push(RequestType::AddUdpListener(*listener).into());
1684 if listener.active {
1685 v.push(
1686 RequestType::ActivateListener(ActivateListener {
1687 address: listener.address,
1688 proxy: ListenerType::Udp.into(),
1689 from_scm: false,
1690 })
1691 .into(),
1692 );
1693 }
1694 }
1695
1696 for cluster in self.clusters.values() {
1697 v.push(RequestType::AddCluster(cluster.clone()).into());
1698 }
1699
1700 for front in self.http_fronts.values() {
1701 v.push(RequestType::AddHttpFrontend(front.clone().into()).into());
1702 }
1703
1704 for (front, certs) in self.certificates.iter() {
1705 for certificate_and_key in certs.values() {
1706 v.push(
1707 RequestType::AddCertificate(AddCertificate {
1708 address: SocketAddress::from(*front),
1709 certificate: certificate_and_key.clone(),
1710 expired_at: None,
1711 })
1712 .into(),
1713 );
1714 }
1715 }
1716
1717 for front in self.https_fronts.values() {
1718 v.push(RequestType::AddHttpsFrontend(front.clone().into()).into());
1719 }
1720
1721 for front_list in self.tcp_fronts.values() {
1722 for front in front_list {
1723 v.push(RequestType::AddTcpFrontend(front.clone().into()).into());
1724 }
1725 }
1726
1727 for front_list in self.udp_fronts.values() {
1728 for front in front_list {
1729 v.push(RequestType::AddUdpFrontend(front.clone().into()).into());
1730 }
1731 }
1732
1733 for backend_list in self.backends.values() {
1734 for backend in backend_list {
1735 v.push(RequestType::AddBackend(backend.clone().to_add_backend()).into());
1736 }
1737 }
1738
1739 #[cfg(debug_assertions)]
1745 {
1746 let mut replayed = ConfigState::new();
1747 for request in &v {
1748 debug_assert!(
1749 replayed.dispatch(request).is_ok(),
1750 "every request from generate_requests must replay cleanly"
1751 );
1752 }
1753 debug_assert!(
1754 replayed.clusters == self.clusters
1755 && replayed.backends == self.backends
1756 && replayed.http_listeners == self.http_listeners
1757 && replayed.https_listeners == self.https_listeners
1758 && replayed.tcp_listeners == self.tcp_listeners
1759 && replayed.http_fronts == self.http_fronts
1760 && replayed.https_fronts == self.https_fronts
1761 && replayed.tcp_fronts == self.tcp_fronts
1762 && replayed.certificates == self.certificates,
1763 "replaying generate_requests into a fresh state must reproduce self"
1764 );
1765 }
1766
1767 v
1768 }
1769
1770 pub fn generate_activate_requests(&self) -> Vec<Request> {
1771 let mut v: Vec<Request> = Vec::new();
1772 for front in self
1773 .http_listeners
1774 .iter()
1775 .filter(|(_, listener)| listener.active)
1776 .map(|(k, _)| k)
1777 {
1778 v.push(
1779 RequestType::ActivateListener(ActivateListener {
1780 address: SocketAddress::from(*front),
1781 proxy: ListenerType::Http.into(),
1782 from_scm: false,
1783 })
1784 .into(),
1785 );
1786 }
1787
1788 for front in self
1789 .https_listeners
1790 .iter()
1791 .filter(|(_, listener)| listener.active)
1792 .map(|(k, _)| k)
1793 {
1794 v.push(
1795 RequestType::ActivateListener(ActivateListener {
1796 address: SocketAddress::from(*front),
1797 proxy: ListenerType::Https.into(),
1798 from_scm: false,
1799 })
1800 .into(),
1801 );
1802 }
1803 for front in self
1804 .tcp_listeners
1805 .iter()
1806 .filter(|(_, listener)| listener.active)
1807 .map(|(k, _)| k)
1808 {
1809 v.push(
1810 RequestType::ActivateListener(ActivateListener {
1811 address: SocketAddress::from(*front),
1812 proxy: ListenerType::Tcp.into(),
1813 from_scm: false,
1814 })
1815 .into(),
1816 );
1817 }
1818 for front in self
1819 .udp_listeners
1820 .iter()
1821 .filter(|(_, listener)| listener.active)
1822 .map(|(k, _)| k)
1823 {
1824 v.push(
1825 RequestType::ActivateListener(ActivateListener {
1826 address: SocketAddress::from(*front),
1827 proxy: ListenerType::Udp.into(),
1828 from_scm: false,
1829 })
1830 .into(),
1831 );
1832 }
1833
1834 #[cfg(debug_assertions)]
1837 {
1838 let active_listeners = self.http_listeners.values().filter(|l| l.active).count()
1839 + self.https_listeners.values().filter(|l| l.active).count()
1840 + self.tcp_listeners.values().filter(|l| l.active).count()
1841 + self.udp_listeners.values().filter(|l| l.active).count();
1842 debug_assert_eq!(
1843 v.len(),
1844 active_listeners,
1845 "generate_activate_requests emits one request per active listener"
1846 );
1847 debug_assert!(
1848 v.iter()
1849 .all(|r| matches!(r.request_type, Some(RequestType::ActivateListener(_)))),
1850 "generate_activate_requests must emit only ActivateListener requests"
1851 );
1852 }
1853
1854 v
1855 }
1856
1857 pub fn diff(&self, other: &ConfigState) -> Vec<Request> {
1858 let my_tcp_listeners: HashSet<&SocketAddr> = self.tcp_listeners.keys().collect();
1860 let their_tcp_listeners: HashSet<&SocketAddr> = other.tcp_listeners.keys().collect();
1861 let removed_tcp_listeners = my_tcp_listeners.difference(&their_tcp_listeners);
1862 let added_tcp_listeners = their_tcp_listeners.difference(&my_tcp_listeners);
1863
1864 let my_udp_listeners: HashSet<&SocketAddr> = self.udp_listeners.keys().collect();
1865 let their_udp_listeners: HashSet<&SocketAddr> = other.udp_listeners.keys().collect();
1866 let removed_udp_listeners = my_udp_listeners.difference(&their_udp_listeners);
1867 let added_udp_listeners = their_udp_listeners.difference(&my_udp_listeners);
1868
1869 let my_http_listeners: HashSet<&SocketAddr> = self.http_listeners.keys().collect();
1870 let their_http_listeners: HashSet<&SocketAddr> = other.http_listeners.keys().collect();
1871 let removed_http_listeners = my_http_listeners.difference(&their_http_listeners);
1872 let added_http_listeners = their_http_listeners.difference(&my_http_listeners);
1873
1874 let my_https_listeners: HashSet<&SocketAddr> = self.https_listeners.keys().collect();
1875 let their_https_listeners: HashSet<&SocketAddr> = other.https_listeners.keys().collect();
1876 let removed_https_listeners = my_https_listeners.difference(&their_https_listeners);
1877 let added_https_listeners = their_https_listeners.difference(&my_https_listeners);
1878
1879 let mut v: Vec<Request> = vec![];
1880
1881 for address in removed_tcp_listeners {
1882 if self.tcp_listeners[*address].active {
1883 v.push(
1884 RequestType::DeactivateListener(DeactivateListener {
1885 address: SocketAddress::from(**address),
1886 proxy: ListenerType::Tcp.into(),
1887 to_scm: false,
1888 })
1889 .into(),
1890 );
1891 }
1892
1893 v.push(
1894 RequestType::RemoveListener(RemoveListener {
1895 address: SocketAddress::from(**address),
1896 proxy: ListenerType::Tcp.into(),
1897 })
1898 .into(),
1899 );
1900 }
1901
1902 for address in added_tcp_listeners.clone() {
1903 v.push(RequestType::AddTcpListener(other.tcp_listeners[*address]).into());
1904
1905 if other.tcp_listeners[*address].active {
1906 v.push(
1907 RequestType::ActivateListener(ActivateListener {
1908 address: SocketAddress::from(**address),
1909 proxy: ListenerType::Tcp.into(),
1910 from_scm: false,
1911 })
1912 .into(),
1913 );
1914 }
1915 }
1916
1917 for address in removed_udp_listeners {
1918 if self.udp_listeners[*address].active {
1919 v.push(
1920 RequestType::DeactivateListener(DeactivateListener {
1921 address: SocketAddress::from(**address),
1922 proxy: ListenerType::Udp.into(),
1923 to_scm: false,
1924 })
1925 .into(),
1926 );
1927 }
1928
1929 v.push(
1930 RequestType::RemoveListener(RemoveListener {
1931 address: SocketAddress::from(**address),
1932 proxy: ListenerType::Udp.into(),
1933 })
1934 .into(),
1935 );
1936 }
1937
1938 for address in added_udp_listeners.clone() {
1939 v.push(RequestType::AddUdpListener(other.udp_listeners[*address]).into());
1940
1941 if other.udp_listeners[*address].active {
1942 v.push(
1943 RequestType::ActivateListener(ActivateListener {
1944 address: SocketAddress::from(**address),
1945 proxy: ListenerType::Udp.into(),
1946 from_scm: false,
1947 })
1948 .into(),
1949 );
1950 }
1951 }
1952
1953 for address in removed_http_listeners {
1954 if self.http_listeners[*address].active {
1955 v.push(
1956 RequestType::DeactivateListener(DeactivateListener {
1957 address: SocketAddress::from(**address),
1958 proxy: ListenerType::Http.into(),
1959 to_scm: false,
1960 })
1961 .into(),
1962 );
1963 }
1964
1965 v.push(
1966 RequestType::RemoveListener(RemoveListener {
1967 address: SocketAddress::from(**address),
1968 proxy: ListenerType::Http.into(),
1969 })
1970 .into(),
1971 );
1972 }
1973
1974 for address in added_http_listeners.clone() {
1975 v.push(RequestType::AddHttpListener(other.http_listeners[*address].clone()).into());
1976
1977 if other.http_listeners[*address].active {
1978 v.push(
1979 RequestType::ActivateListener(ActivateListener {
1980 address: SocketAddress::from(**address),
1981 proxy: ListenerType::Http.into(),
1982 from_scm: false,
1983 })
1984 .into(),
1985 );
1986 }
1987 }
1988
1989 for address in removed_https_listeners {
1990 if self.https_listeners[*address].active {
1991 v.push(
1992 RequestType::DeactivateListener(DeactivateListener {
1993 address: SocketAddress::from(**address),
1994 proxy: ListenerType::Https.into(),
1995 to_scm: false,
1996 })
1997 .into(),
1998 );
1999 }
2000
2001 v.push(
2002 RequestType::RemoveListener(RemoveListener {
2003 address: SocketAddress::from(**address),
2004 proxy: ListenerType::Https.into(),
2005 })
2006 .into(),
2007 );
2008 }
2009
2010 for address in added_https_listeners.clone() {
2011 v.push(RequestType::AddHttpsListener(other.https_listeners[*address].clone()).into());
2012
2013 if other.https_listeners[*address].active {
2014 v.push(
2015 RequestType::ActivateListener(ActivateListener {
2016 address: SocketAddress::from(**address),
2017 proxy: ListenerType::Https.into(),
2018 from_scm: false,
2019 })
2020 .into(),
2021 );
2022 }
2023 }
2024
2025 for addr in my_tcp_listeners.intersection(&their_tcp_listeners) {
2026 let my_listener = &self.tcp_listeners[*addr];
2027 let their_listener = &other.tcp_listeners[*addr];
2028
2029 if my_listener != their_listener {
2030 v.push(
2031 RequestType::RemoveListener(RemoveListener {
2032 address: SocketAddress::from(**addr),
2033 proxy: ListenerType::Tcp.into(),
2034 })
2035 .into(),
2036 );
2037 let mut listener_to_add = *their_listener;
2039 listener_to_add.active = false;
2040 v.push(RequestType::AddTcpListener(listener_to_add).into());
2041
2042 if their_listener.active {
2049 v.push(
2050 RequestType::ActivateListener(ActivateListener {
2051 address: SocketAddress::from(**addr),
2052 proxy: ListenerType::Tcp.into(),
2053 from_scm: false,
2054 })
2055 .into(),
2056 );
2057 }
2058 }
2059
2060 if my_listener.active && !their_listener.active {
2061 v.push(
2062 RequestType::DeactivateListener(DeactivateListener {
2063 address: SocketAddress::from(**addr),
2064 proxy: ListenerType::Tcp.into(),
2065 to_scm: false,
2066 })
2067 .into(),
2068 );
2069 }
2070 }
2071
2072 for addr in my_udp_listeners.intersection(&their_udp_listeners) {
2073 let my_listener = &self.udp_listeners[*addr];
2074 let their_listener = &other.udp_listeners[*addr];
2075
2076 if my_listener != their_listener {
2077 v.push(
2078 RequestType::RemoveListener(RemoveListener {
2079 address: SocketAddress::from(**addr),
2080 proxy: ListenerType::Udp.into(),
2081 })
2082 .into(),
2083 );
2084 let mut listener_to_add = *their_listener;
2086 listener_to_add.active = false;
2087 v.push(RequestType::AddUdpListener(listener_to_add).into());
2088
2089 if their_listener.active {
2096 v.push(
2097 RequestType::ActivateListener(ActivateListener {
2098 address: SocketAddress::from(**addr),
2099 proxy: ListenerType::Udp.into(),
2100 from_scm: false,
2101 })
2102 .into(),
2103 );
2104 }
2105 }
2106
2107 if my_listener.active && !their_listener.active {
2108 v.push(
2109 RequestType::DeactivateListener(DeactivateListener {
2110 address: SocketAddress::from(**addr),
2111 proxy: ListenerType::Udp.into(),
2112 to_scm: false,
2113 })
2114 .into(),
2115 );
2116 }
2117 }
2118
2119 for addr in my_http_listeners.intersection(&their_http_listeners) {
2120 let my_listener = &self.http_listeners[*addr];
2121 let their_listener = &other.http_listeners[*addr];
2122
2123 if my_listener != their_listener {
2124 v.push(
2125 RequestType::RemoveListener(RemoveListener {
2126 address: SocketAddress::from(**addr),
2127 proxy: ListenerType::Http.into(),
2128 })
2129 .into(),
2130 );
2131 let mut listener_to_add = their_listener.clone();
2133 listener_to_add.active = false;
2134 v.push(RequestType::AddHttpListener(listener_to_add).into());
2135
2136 if their_listener.active {
2143 v.push(
2144 RequestType::ActivateListener(ActivateListener {
2145 address: SocketAddress::from(**addr),
2146 proxy: ListenerType::Http.into(),
2147 from_scm: false,
2148 })
2149 .into(),
2150 );
2151 }
2152 }
2153
2154 if my_listener.active && !their_listener.active {
2155 v.push(
2156 RequestType::DeactivateListener(DeactivateListener {
2157 address: SocketAddress::from(**addr),
2158 proxy: ListenerType::Http.into(),
2159 to_scm: false,
2160 })
2161 .into(),
2162 );
2163 }
2164 }
2165
2166 for addr in my_https_listeners.intersection(&their_https_listeners) {
2167 let my_listener = &self.https_listeners[*addr];
2168 let their_listener = &other.https_listeners[*addr];
2169
2170 if my_listener != their_listener {
2171 v.push(
2172 RequestType::RemoveListener(RemoveListener {
2173 address: SocketAddress::from(**addr),
2174 proxy: ListenerType::Https.into(),
2175 })
2176 .into(),
2177 );
2178 let mut listener_to_add = their_listener.clone();
2180 listener_to_add.active = false;
2181 v.push(RequestType::AddHttpsListener(listener_to_add).into());
2182
2183 if their_listener.active {
2190 v.push(
2191 RequestType::ActivateListener(ActivateListener {
2192 address: SocketAddress::from(**addr),
2193 proxy: ListenerType::Https.into(),
2194 from_scm: false,
2195 })
2196 .into(),
2197 );
2198 }
2199 }
2200
2201 if my_listener.active && !their_listener.active {
2202 v.push(
2203 RequestType::DeactivateListener(DeactivateListener {
2204 address: SocketAddress::from(**addr),
2205 proxy: ListenerType::Https.into(),
2206 to_scm: false,
2207 })
2208 .into(),
2209 );
2210 }
2211 }
2212
2213 for (cluster_id, res) in diff_map(self.clusters.iter(), other.clusters.iter()) {
2214 match res {
2215 DiffResult::Added | DiffResult::Changed => v.push(
2216 RequestType::AddCluster(other.clusters.get(cluster_id).unwrap().clone()).into(),
2217 ),
2218 DiffResult::Removed => {
2219 v.push(RequestType::RemoveCluster(cluster_id.to_string()).into())
2220 }
2221 }
2222 }
2223
2224 for ((cluster_id, backend_id), res) in diff_map(
2225 self.backends.iter().flat_map(|(cluster_id, v)| {
2226 v.iter()
2227 .map(move |backend| ((cluster_id, &backend.backend_id), backend))
2228 }),
2229 other.backends.iter().flat_map(|(cluster_id, v)| {
2230 v.iter()
2231 .map(move |backend| ((cluster_id, &backend.backend_id), backend))
2232 }),
2233 ) {
2234 match res {
2235 DiffResult::Added => {
2236 let backend = other
2237 .backends
2238 .get(cluster_id)
2239 .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
2240 .unwrap();
2241 v.push(RequestType::AddBackend(backend.clone().to_add_backend()).into());
2242 }
2243 DiffResult::Removed => {
2244 let backend = self
2245 .backends
2246 .get(cluster_id)
2247 .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
2248 .unwrap();
2249
2250 v.push(
2251 RequestType::RemoveBackend(RemoveBackend {
2252 cluster_id: backend.cluster_id.clone(),
2253 backend_id: backend.backend_id.clone(),
2254 address: SocketAddress::from(backend.address),
2255 })
2256 .into(),
2257 );
2258 }
2259 DiffResult::Changed => {
2260 let backend = self
2261 .backends
2262 .get(cluster_id)
2263 .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
2264 .unwrap();
2265
2266 v.push(
2267 RequestType::RemoveBackend(RemoveBackend {
2268 cluster_id: backend.cluster_id.clone(),
2269 backend_id: backend.backend_id.clone(),
2270 address: SocketAddress::from(backend.address),
2271 })
2272 .into(),
2273 );
2274
2275 let backend = other
2276 .backends
2277 .get(cluster_id)
2278 .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
2279 .unwrap();
2280 v.push(RequestType::AddBackend(backend.clone().to_add_backend()).into());
2281 }
2282 }
2283 }
2284
2285 let mut my_http_fronts: HashSet<(&str, &HttpFrontend)> = HashSet::new();
2286 for (route, front) in self.http_fronts.iter() {
2287 my_http_fronts.insert((route, front));
2288 }
2289 let mut their_http_fronts: HashSet<(&str, &HttpFrontend)> = HashSet::new();
2290 for (route, front) in other.http_fronts.iter() {
2291 their_http_fronts.insert((route, front));
2292 }
2293
2294 let removed_http_fronts = my_http_fronts.difference(&their_http_fronts);
2295 let added_http_fronts = their_http_fronts.difference(&my_http_fronts);
2296
2297 for &(_, front) in removed_http_fronts {
2298 v.push(RequestType::RemoveHttpFrontend(front.clone().into()).into());
2299 }
2300
2301 for &(_, front) in added_http_fronts {
2302 v.push(RequestType::AddHttpFrontend(front.clone().into()).into());
2303 }
2304
2305 let mut my_https_fronts: HashSet<(&String, &HttpFrontend)> = HashSet::new();
2306 for (route, front) in self.https_fronts.iter() {
2307 my_https_fronts.insert((route, front));
2308 }
2309 let mut their_https_fronts: HashSet<(&String, &HttpFrontend)> = HashSet::new();
2310 for (route, front) in other.https_fronts.iter() {
2311 their_https_fronts.insert((route, front));
2312 }
2313 let removed_https_fronts = my_https_fronts.difference(&their_https_fronts);
2314 let added_https_fronts = their_https_fronts.difference(&my_https_fronts);
2315
2316 for &(_, front) in removed_https_fronts {
2317 v.push(RequestType::RemoveHttpsFrontend(front.clone().into()).into());
2318 }
2319
2320 for &(_, front) in added_https_fronts {
2321 v.push(RequestType::AddHttpsFrontend(front.clone().into()).into());
2322 }
2323
2324 let mut my_tcp_fronts: HashSet<(&ClusterId, &TcpFrontend)> = HashSet::new();
2325 for (cluster_id, front_list) in self.tcp_fronts.iter() {
2326 for front in front_list.iter() {
2327 my_tcp_fronts.insert((cluster_id, front));
2328 }
2329 }
2330 let mut their_tcp_fronts: HashSet<(&ClusterId, &TcpFrontend)> = HashSet::new();
2331 for (cluster_id, front_list) in other.tcp_fronts.iter() {
2332 for front in front_list.iter() {
2333 their_tcp_fronts.insert((cluster_id, front));
2334 }
2335 }
2336
2337 let removed_tcp_fronts = my_tcp_fronts.difference(&their_tcp_fronts);
2338 let added_tcp_fronts = their_tcp_fronts.difference(&my_tcp_fronts);
2339
2340 for &(_, front) in removed_tcp_fronts {
2341 v.push(RequestType::RemoveTcpFrontend(front.clone().into()).into());
2342 }
2343
2344 for &(_, front) in added_tcp_fronts {
2345 v.push(RequestType::AddTcpFrontend(front.clone().into()).into());
2346 }
2347
2348 let mut my_udp_fronts: HashSet<(&ClusterId, &UdpFrontend)> = HashSet::new();
2349 for (cluster_id, front_list) in self.udp_fronts.iter() {
2350 for front in front_list.iter() {
2351 my_udp_fronts.insert((cluster_id, front));
2352 }
2353 }
2354 let mut their_udp_fronts: HashSet<(&ClusterId, &UdpFrontend)> = HashSet::new();
2355 for (cluster_id, front_list) in other.udp_fronts.iter() {
2356 for front in front_list.iter() {
2357 their_udp_fronts.insert((cluster_id, front));
2358 }
2359 }
2360
2361 let removed_udp_fronts = my_udp_fronts.difference(&their_udp_fronts);
2362 let added_udp_fronts = their_udp_fronts.difference(&my_udp_fronts);
2363
2364 for &(_, front) in removed_udp_fronts {
2365 v.push(RequestType::RemoveUdpFrontend(front.clone().into()).into());
2366 }
2367
2368 for &(_, front) in added_udp_fronts {
2369 v.push(RequestType::AddUdpFrontend(front.clone().into()).into());
2370 }
2371
2372 let my_certificates: HashSet<(SocketAddr, &Fingerprint)> = HashSet::from_iter(
2374 self.certificates
2375 .iter()
2376 .flat_map(|(addr, certs)| repeat(*addr).zip(certs.keys())),
2377 );
2378 let their_certificates: HashSet<(SocketAddr, &Fingerprint)> = HashSet::from_iter(
2379 other
2380 .certificates
2381 .iter()
2382 .flat_map(|(addr, certs)| repeat(*addr).zip(certs.keys())),
2383 );
2384
2385 let removed_certificates = my_certificates.difference(&their_certificates);
2386 let added_certificates = their_certificates.difference(&my_certificates);
2387
2388 for &(address, fingerprint) in removed_certificates {
2389 v.push(
2390 RequestType::RemoveCertificate(RemoveCertificate {
2391 address: SocketAddress::from(address),
2392 fingerprint: fingerprint.to_string(),
2393 })
2394 .into(),
2395 );
2396 }
2397
2398 for &(address, fingerprint) in added_certificates {
2399 if let Some(certificate_and_key) = other
2400 .certificates
2401 .get(&address)
2402 .and_then(|certs| certs.get(fingerprint))
2403 {
2404 v.push(
2405 RequestType::AddCertificate(AddCertificate {
2406 address: SocketAddress::from(address),
2407 certificate: certificate_and_key.clone(),
2408 expired_at: None,
2409 })
2410 .into(),
2411 );
2412 }
2413 }
2414
2415 for address in added_tcp_listeners {
2416 let listener = &other.tcp_listeners[*address];
2417 if listener.active {
2418 v.push(
2419 RequestType::ActivateListener(ActivateListener {
2420 address: listener.address,
2421 proxy: ListenerType::Tcp.into(),
2422 from_scm: false,
2423 })
2424 .into(),
2425 );
2426 }
2427 }
2428
2429 for address in added_udp_listeners {
2430 let listener = &other.udp_listeners[*address];
2431 if listener.active {
2432 v.push(
2433 RequestType::ActivateListener(ActivateListener {
2434 address: listener.address,
2435 proxy: ListenerType::Udp.into(),
2436 from_scm: false,
2437 })
2438 .into(),
2439 );
2440 }
2441 }
2442
2443 #[cfg(debug_assertions)]
2456 {
2457 let mut replayed = self.clone();
2458 for request in &v {
2459 debug_assert!(
2461 replayed.dispatch(request).is_ok(),
2462 "every request emitted by diff must replay cleanly onto self"
2463 );
2464 }
2465 let nonempty = |m: &BTreeMap<ClusterId, Vec<Backend>>| {
2466 m.iter()
2467 .filter(|(_, v)| !v.is_empty())
2468 .map(|(k, v)| (k.clone(), v.clone()))
2469 .collect::<BTreeMap<_, _>>()
2470 };
2471 let nonempty_tcp = |m: &HashMap<ClusterId, Vec<TcpFrontend>>| {
2472 m.iter()
2473 .filter(|(_, v)| !v.is_empty())
2474 .map(|(k, v)| (k.clone(), v.clone()))
2475 .collect::<HashMap<_, _>>()
2476 };
2477 debug_assert!(
2478 replayed.clusters == other.clusters
2479 && nonempty(&replayed.backends) == nonempty(&other.backends)
2480 && replayed.http_fronts == other.http_fronts
2481 && replayed.https_fronts == other.https_fronts
2482 && nonempty_tcp(&replayed.tcp_fronts) == nonempty_tcp(&other.tcp_fronts)
2483 && replayed.certificates == other.certificates
2484 && replayed.http_listeners == other.http_listeners
2485 && replayed.https_listeners == other.https_listeners
2486 && replayed.tcp_listeners == other.tcp_listeners
2487 && replayed.udp_listeners == other.udp_listeners,
2488 "replaying diff(self, other) onto self must reproduce other's clusters/backends/frontends/certificates/listeners"
2489 );
2490 }
2491
2492 v
2493 }
2494
2495 pub fn hash_state(&self) -> BTreeMap<ClusterId, u64> {
2497 let mut hm: HashMap<ClusterId, DefaultHasher> = self
2498 .clusters
2499 .keys()
2500 .map(|cluster_id| {
2501 let mut hasher = DefaultHasher::new();
2502 self.clusters.get(cluster_id).hash(&mut hasher);
2503 if let Some(backends) = self.backends.get(cluster_id) {
2504 backends.iter().collect::<BTreeSet<_>>().hash(&mut hasher)
2505 }
2506 if let Some(tcp_fronts) = self.tcp_fronts.get(cluster_id) {
2507 tcp_fronts.iter().collect::<BTreeSet<_>>().hash(&mut hasher)
2508 }
2509 (cluster_id.to_owned(), hasher)
2510 })
2511 .collect();
2512
2513 for front in self.http_fronts.values() {
2514 if let Some(cluster_id) = &front.cluster_id
2515 && let Some(hasher) = hm.get_mut(cluster_id)
2516 {
2517 front.hash(hasher);
2518 }
2519 }
2520
2521 for front in self.https_fronts.values() {
2522 if let Some(cluster_id) = &front.cluster_id
2523 && let Some(hasher) = hm.get_mut(cluster_id)
2524 {
2525 front.hash(hasher);
2526 }
2527 }
2528
2529 hm.drain()
2530 .map(|(cluster_id, hasher)| (cluster_id, hasher.finish()))
2531 .collect()
2532 }
2533
2534 pub fn cluster_state(&self, cluster_id: &str) -> Option<ClusterInformation> {
2537 let configuration = self.clusters.get(cluster_id).cloned()?;
2538 info!("{:?}", configuration);
2539
2540 let http_frontends: Vec<RequestHttpFrontend> = self
2541 .http_fronts
2542 .values()
2543 .filter(|front| front.cluster_id.as_deref() == Some(cluster_id))
2544 .map(|front| front.clone().into())
2545 .collect();
2546
2547 let https_frontends: Vec<RequestHttpFrontend> = self
2548 .https_fronts
2549 .values()
2550 .filter(|front| front.cluster_id.as_deref() == Some(cluster_id))
2551 .map(|front| front.clone().into())
2552 .collect();
2553
2554 let tcp_frontends: Vec<RequestTcpFrontend> = self
2555 .tcp_fronts
2556 .get(cluster_id)
2557 .cloned()
2558 .unwrap_or_default()
2559 .iter()
2560 .map(|front| front.clone().into())
2561 .collect();
2562
2563 let udp_frontends: Vec<RequestUdpFrontend> = self
2564 .udp_fronts
2565 .get(cluster_id)
2566 .cloned()
2567 .unwrap_or_default()
2568 .iter()
2569 .map(|front| front.clone().into())
2570 .collect();
2571
2572 let backends: Vec<AddBackend> = self
2573 .backends
2574 .get(cluster_id)
2575 .cloned()
2576 .unwrap_or_default()
2577 .iter()
2578 .map(|backend| backend.clone().into())
2579 .collect();
2580
2581 Some(ClusterInformation {
2582 configuration: Some(configuration),
2583 http_frontends,
2584 https_frontends,
2585 tcp_frontends,
2586 backends,
2587 udp_frontends,
2588 })
2589 }
2590
2591 pub fn count_backends(&self) -> usize {
2592 self.backends.values().fold(0, |acc, v| acc + v.len())
2593 }
2594
2595 pub fn count_frontends(&self) -> usize {
2596 self.http_fronts.values().count()
2597 + self.https_fronts.values().count()
2598 + self.tcp_fronts.values().fold(0, |acc, v| acc + v.len())
2599 + self.udp_fronts.values().fold(0, |acc, v| acc + v.len())
2600 }
2601
2602 pub fn get_cluster_ids_by_domain(
2603 &self,
2604 hostname: String,
2605 path: Option<String>,
2606 ) -> HashSet<ClusterId> {
2607 let mut cluster_ids: HashSet<ClusterId> = HashSet::new();
2608
2609 self.http_fronts.values().for_each(|front| {
2610 if domain_check(&front.hostname, &front.path, &hostname, &path)
2611 && let Some(id) = &front.cluster_id
2612 {
2613 cluster_ids.insert(id.to_string());
2614 }
2615 });
2616
2617 self.https_fronts.values().for_each(|front| {
2618 if domain_check(&front.hostname, &front.path, &hostname, &path)
2619 && let Some(id) = &front.cluster_id
2620 {
2621 cluster_ids.insert(id.to_string());
2622 }
2623 });
2624
2625 cluster_ids
2626 }
2627
2628 pub fn get_certificates(
2629 &self,
2630 filters: QueryCertificatesFilters,
2631 ) -> BTreeMap<String, CertificateAndKey> {
2632 self.certificates
2633 .values()
2634 .flat_map(|hash_map| hash_map.iter())
2635 .filter(|(fingerprint, cert)| {
2636 if let Some(domain) = &filters.domain {
2637 cert.names.contains(domain)
2638 } else if let Some(f) = &filters.fingerprint {
2639 fingerprint.to_string() == *f
2640 } else {
2641 true
2642 }
2643 })
2644 .map(|(fingerprint, cert)| (fingerprint.to_string(), cert.to_owned()))
2645 .collect()
2646 }
2647
2648 pub fn list_frontends(&self, filters: FrontendFilters) -> ListedFrontends {
2649 let list_all = !filters.http && !filters.https && !filters.tcp;
2651
2652 let mut listed_frontends = ListedFrontends::default();
2653
2654 if filters.http || list_all {
2655 for http_frontend in self.http_fronts.iter().filter(|f| {
2656 if let Some(domain) = &filters.domain {
2657 f.1.hostname.contains(domain)
2658 } else {
2659 true
2660 }
2661 }) {
2662 listed_frontends
2663 .http_frontends
2664 .push(http_frontend.1.to_owned().into());
2665 }
2666 }
2667
2668 if filters.https || list_all {
2669 for https_frontend in self.https_fronts.iter().filter(|f| {
2670 if let Some(domain) = &filters.domain {
2671 f.1.hostname.contains(domain)
2672 } else {
2673 true
2674 }
2675 }) {
2676 listed_frontends
2677 .https_frontends
2678 .push(https_frontend.1.to_owned().into());
2679 }
2680 }
2681
2682 if (filters.tcp || list_all) && filters.domain.is_none() {
2683 for tcp_frontend in self.tcp_fronts.values().flat_map(|v| v.iter()) {
2684 listed_frontends
2685 .tcp_frontends
2686 .push(tcp_frontend.to_owned().into())
2687 }
2688 }
2689
2690 if (filters.tcp || list_all) && filters.domain.is_none() {
2696 for udp_frontend in self.udp_fronts.values().flat_map(|v| v.iter()) {
2697 listed_frontends
2698 .udp_frontends
2699 .push(udp_frontend.to_owned().into())
2700 }
2701 }
2702
2703 listed_frontends
2704 }
2705
2706 pub fn list_listeners(&self) -> ListenersList {
2707 ListenersList {
2708 http_listeners: self
2709 .http_listeners
2710 .iter()
2711 .map(|(addr, listener)| (addr.to_string(), listener.clone()))
2712 .collect(),
2713 https_listeners: self
2714 .https_listeners
2715 .iter()
2716 .map(|(addr, listener)| (addr.to_string(), listener.clone()))
2717 .collect(),
2718 tcp_listeners: self
2719 .tcp_listeners
2720 .iter()
2721 .map(|(addr, listener)| (addr.to_string(), *listener))
2722 .collect(),
2723 udp_listeners: self
2724 .udp_listeners
2725 .iter()
2726 .map(|(addr, listener)| (addr.to_string(), *listener))
2727 .collect(),
2728 }
2729 }
2730
2731 pub fn produce_initial_state(&self) -> InitialState {
2733 let mut worker_requests = Vec::new();
2734 for (counter, request) in self.generate_requests().into_iter().enumerate() {
2735 worker_requests.push(WorkerRequest::new(format!("SAVE-{counter}"), request));
2736 }
2737 InitialState {
2738 requests: worker_requests,
2739 }
2740 }
2741
2742 pub fn write_initial_state_to_file(&self, file: &mut File) -> Result<usize, StateError> {
2745 let initial_state = self.produce_initial_state();
2746 let count = initial_state.requests.len();
2747
2748 let bytes_to_write = initial_state.encode_to_vec();
2749 println!("writing {} in the temp file", bytes_to_write.len());
2750 file.write_all(&bytes_to_write)
2751 .map_err(StateError::FileError)?;
2752
2753 file.sync_all().map_err(StateError::FileError)?;
2754
2755 Ok(count)
2756 }
2757
2758 pub fn write_requests_to_file(&self, file: &mut File) -> Result<usize, StateError> {
2762 let mut counter = 0usize;
2763 let requests = self.generate_requests();
2764
2765 for request in requests {
2766 let message = WorkerRequest::new(format!("SAVE-{counter}"), request);
2767
2768 file.write_all(
2769 &serde_json::to_string(&message)
2770 .map(|s| s.into_bytes())
2771 .unwrap_or_default(),
2772 )
2773 .map_err(StateError::FileError)?;
2774
2775 file.write_all(&b"\n\0"[..])
2776 .map_err(StateError::FileError)?;
2777
2778 if counter.is_multiple_of(1000) {
2779 info!("writing {} commands to file", counter);
2780 file.sync_all().map_err(StateError::FileError)?;
2781 }
2782 counter += 1;
2783 }
2784 file.sync_all().map_err(StateError::FileError)?;
2785
2786 Ok(counter)
2787 }
2788}
2789
2790pub fn validate_h2_flood_knobs_http(patch: &UpdateHttpListenerConfig) -> Result<(), StateError> {
2804 macro_rules! require_ge1 {
2805 ($field:expr, $name:literal) => {
2806 if let Some(0) = $field {
2807 return Err(StateError::InvalidValue {
2808 field: $name,
2809 reason: "must be >= 1",
2810 });
2811 }
2812 };
2813 }
2814 require_ge1!(
2815 patch.h2_max_rst_stream_per_window,
2816 "h2_max_rst_stream_per_window"
2817 );
2818 require_ge1!(patch.h2_max_ping_per_window, "h2_max_ping_per_window");
2819 require_ge1!(
2820 patch.h2_max_settings_per_window,
2821 "h2_max_settings_per_window"
2822 );
2823 require_ge1!(
2824 patch.h2_max_empty_data_per_window,
2825 "h2_max_empty_data_per_window"
2826 );
2827 require_ge1!(
2828 patch.h2_max_continuation_frames,
2829 "h2_max_continuation_frames"
2830 );
2831 require_ge1!(patch.h2_max_glitch_count, "h2_max_glitch_count");
2832 require_ge1!(
2833 patch.h2_max_window_update_stream0_per_window,
2834 "h2_max_window_update_stream0_per_window"
2835 );
2836 require_ge1!(patch.h2_max_concurrent_streams, "h2_max_concurrent_streams");
2837 if let Some(v) = patch.h2_stream_shrink_ratio
2840 && v < 2
2841 {
2842 return Err(StateError::InvalidValue {
2843 field: "h2_stream_shrink_ratio",
2844 reason: "must be >= 2",
2845 });
2846 }
2847 require_ge1!(
2850 patch.h2_max_rst_stream_lifetime,
2851 "h2_max_rst_stream_lifetime"
2852 );
2853 require_ge1!(
2854 patch.h2_max_rst_stream_abusive_lifetime,
2855 "h2_max_rst_stream_abusive_lifetime"
2856 );
2857 require_ge1!(
2858 patch.h2_max_rst_stream_emitted_lifetime,
2859 "h2_max_rst_stream_emitted_lifetime"
2860 );
2861 require_ge1!(patch.h2_max_header_list_size, "h2_max_header_list_size");
2862 require_ge1!(patch.h2_max_header_table_size, "h2_max_header_table_size");
2863 require_ge1!(patch.h2_max_header_fields, "h2_max_header_fields");
2864 Ok(())
2865}
2866
2867pub fn validate_h2_flood_knobs_https(patch: &UpdateHttpsListenerConfig) -> Result<(), StateError> {
2869 macro_rules! require_ge1 {
2870 ($field:expr, $name:literal) => {
2871 if let Some(0) = $field {
2872 return Err(StateError::InvalidValue {
2873 field: $name,
2874 reason: "must be >= 1",
2875 });
2876 }
2877 };
2878 }
2879 require_ge1!(
2880 patch.h2_max_rst_stream_per_window,
2881 "h2_max_rst_stream_per_window"
2882 );
2883 require_ge1!(patch.h2_max_ping_per_window, "h2_max_ping_per_window");
2884 require_ge1!(
2885 patch.h2_max_settings_per_window,
2886 "h2_max_settings_per_window"
2887 );
2888 require_ge1!(
2889 patch.h2_max_empty_data_per_window,
2890 "h2_max_empty_data_per_window"
2891 );
2892 require_ge1!(
2893 patch.h2_max_continuation_frames,
2894 "h2_max_continuation_frames"
2895 );
2896 require_ge1!(patch.h2_max_glitch_count, "h2_max_glitch_count");
2897 require_ge1!(
2898 patch.h2_max_window_update_stream0_per_window,
2899 "h2_max_window_update_stream0_per_window"
2900 );
2901 require_ge1!(patch.h2_max_concurrent_streams, "h2_max_concurrent_streams");
2902 if let Some(v) = patch.h2_stream_shrink_ratio
2903 && v < 2
2904 {
2905 return Err(StateError::InvalidValue {
2906 field: "h2_stream_shrink_ratio",
2907 reason: "must be >= 2",
2908 });
2909 }
2910 require_ge1!(
2911 patch.h2_max_rst_stream_lifetime,
2912 "h2_max_rst_stream_lifetime"
2913 );
2914 require_ge1!(
2915 patch.h2_max_rst_stream_abusive_lifetime,
2916 "h2_max_rst_stream_abusive_lifetime"
2917 );
2918 require_ge1!(
2919 patch.h2_max_rst_stream_emitted_lifetime,
2920 "h2_max_rst_stream_emitted_lifetime"
2921 );
2922 require_ge1!(patch.h2_max_header_list_size, "h2_max_header_list_size");
2923 require_ge1!(patch.h2_max_header_table_size, "h2_max_header_table_size");
2924 require_ge1!(patch.h2_max_header_fields, "h2_max_header_fields");
2925 Ok(())
2926}
2927
2928pub fn merge_custom_http_answers(
2941 target: &mut Option<CustomHttpAnswers>,
2942 patch: &CustomHttpAnswers,
2943) {
2944 let current = target.get_or_insert_with(CustomHttpAnswers::default);
2945 macro_rules! merge_field {
2946 ($field:ident) => {
2947 if let Some(ref v) = patch.$field {
2948 current.$field = Some(v.clone());
2949 }
2950 };
2951 }
2952 merge_field!(answer_301);
2953 merge_field!(answer_400);
2954 merge_field!(answer_401);
2955 merge_field!(answer_404);
2956 merge_field!(answer_408);
2957 merge_field!(answer_413);
2958 merge_field!(answer_421);
2959 merge_field!(answer_502);
2960 merge_field!(answer_503);
2961 merge_field!(answer_504);
2962 merge_field!(answer_507);
2963}
2964
2965pub fn validate_alpn_protocols(values: &[String]) -> Result<(), StateError> {
2968 for value in values {
2969 if value != "h2" && value != "http/1.1" {
2970 return Err(StateError::InvalidValue {
2971 field: "alpn_protocols",
2972 reason: "each value must be \"h2\" or \"http/1.1\"",
2973 });
2974 }
2975 }
2976 Ok(())
2977}
2978
2979pub fn validate_sozu_id_header(value: &str) -> Result<(), StateError> {
2990 if value.is_empty() {
2991 return Err(StateError::InvalidValue {
2992 field: "sozu_id_header",
2993 reason: "must not be empty",
2994 });
2995 }
2996 for b in value.bytes() {
2997 let is_tchar = b.is_ascii_alphanumeric()
2998 || matches!(
2999 b,
3000 b'!' | b'#'
3001 | b'$'
3002 | b'%'
3003 | b'&'
3004 | b'\''
3005 | b'*'
3006 | b'+'
3007 | b'-'
3008 | b'.'
3009 | b'^'
3010 | b'_'
3011 | b'`'
3012 | b'|'
3013 | b'~'
3014 );
3015 if !is_tchar {
3016 return Err(StateError::InvalidValue {
3017 field: "sozu_id_header",
3018 reason: "must be a valid HTTP header name (RFC 9110 §5.1 token: alphanumeric or one of !#$%&'*+-.^_`|~)",
3019 });
3020 }
3021 }
3022 Ok(())
3023}
3024
3025fn domain_check(
3026 front_hostname: &str,
3027 front_path_rule: &PathRule,
3028 hostname: &str,
3029 path_prefix: &Option<String>,
3030) -> bool {
3031 if hostname != front_hostname {
3032 return false;
3033 }
3034
3035 if let Some(path) = &path_prefix {
3036 return path == &front_path_rule.value;
3037 }
3038
3039 true
3040}
3041
3042fn canonical_tcp_alpn(alpn: &[String]) -> Vec<String> {
3052 let mut alpn = alpn.to_vec();
3053 alpn.sort();
3054 alpn.dedup();
3055 debug_assert!(
3059 alpn.windows(2).all(|pair| pair[0] < pair[1]),
3060 "canonical_tcp_alpn must return a strictly sorted, duplicate-free list"
3061 );
3062 alpn
3063}
3064
3065fn tcp_frontend_matches(
3076 front: &TcpFrontend,
3077 address: SocketAddr,
3078 sni: &Option<String>,
3079 alpn: &[String],
3080) -> bool {
3081 front.address == address && front.sni == *sni && front.alpn.as_slice() == alpn
3082}
3083
3084struct DiffMap<'a, K: Ord, V, I1, I2> {
3085 my_it: I1,
3086 other_it: I2,
3087 my: Option<(K, &'a V)>,
3088 other: Option<(K, &'a V)>,
3089}
3090
3091fn diff_map<
3093 'a,
3094 K: Ord,
3095 V: PartialEq,
3096 I1: Iterator<Item = (K, &'a V)>,
3097 I2: Iterator<Item = (K, &'a V)>,
3098>(
3099 my: I1,
3100 other: I2,
3101) -> DiffMap<'a, K, V, I1, I2> {
3102 DiffMap {
3103 my_it: my,
3104 other_it: other,
3105 my: None,
3106 other: None,
3107 }
3108}
3109
3110enum DiffResult {
3111 Added,
3112 Removed,
3113 Changed,
3114}
3115
3116impl<'a, K: Ord, V: PartialEq, I1: Iterator<Item = (K, &'a V)>, I2: Iterator<Item = (K, &'a V)>>
3119 std::iter::Iterator for DiffMap<'a, K, V, I1, I2>
3120{
3121 type Item = (K, DiffResult);
3122
3123 fn next(&mut self) -> Option<Self::Item> {
3124 loop {
3125 if self.my.is_none() {
3126 self.my = self.my_it.next();
3127 }
3128 if self.other.is_none() {
3129 self.other = self.other_it.next();
3130 }
3131
3132 match (self.my.take(), self.other.take()) {
3133 (None, other) => return other.map(|(k, _)| (k, DiffResult::Added)),
3137 (Some((k, _)), None) => return Some((k, DiffResult::Removed)),
3140 (Some((k1, _v1)), Some((k2, v2))) if k1 < k2 => {
3142 self.other = Some((k2, v2));
3143 return Some((k1, DiffResult::Removed));
3144 }
3145 (Some((k1, v1)), Some((k2, _v2))) if k1 > k2 => {
3147 self.my = Some((k1, v1));
3148 return Some((k2, DiffResult::Added));
3149 }
3150 (Some((k1, v1)), Some((_k2, v2))) if v1 != v2 => {
3151 return Some((k1, DiffResult::Changed));
3154 }
3155 _ => {}
3156 }
3157 }
3158 }
3159}
3160
3161#[cfg(test)]
3162mod tests {
3163 use rand::{RngExt, rng, seq::SliceRandom};
3164
3165 use super::*;
3166 use crate::proto::command::{
3167 CustomHttpAnswers, Header, HeaderPosition, HstsConfig, LoadBalancingParams, PathRuleKind,
3168 RedirectPolicy, RedirectScheme, RequestHttpFrontend, RequestTcpFrontend,
3169 RequestUdpFrontend, RulePosition, UdpListenerConfig, UpdateUdpListenerConfig,
3170 };
3171
3172 fn capture_test_logs(run: impl FnOnce() + Send + 'static) -> String {
3173 let receiver = std::net::UdpSocket::bind("127.0.0.1:0")
3174 .expect("test log receiver must bind to a loopback port");
3175 let target = format!(
3176 "udp://{}",
3177 receiver
3178 .local_addr()
3179 .expect("test log receiver must have a local address")
3180 );
3181
3182 std::thread::spawn(move || {
3183 crate::logging::Logger::init(
3184 "state-log-redaction-test".to_owned(),
3185 "info",
3186 &target,
3187 false,
3188 None,
3189 None,
3190 None,
3191 )
3192 .expect("test logger must initialize");
3193 run();
3194 })
3195 .join()
3196 .expect("log-producing test thread must not panic");
3197
3198 receiver
3199 .set_nonblocking(true)
3200 .expect("test log receiver must become nonblocking");
3201 let mut output = String::new();
3202 let mut datagram = vec![0; 65_507];
3203 loop {
3204 match receiver.recv(&mut datagram) {
3205 Ok(length) => output.push_str(&String::from_utf8_lossy(&datagram[..length])),
3206 Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => break,
3207 Err(error) => panic!("test log receiver failed: {error}"),
3208 }
3209 }
3210 assert!(!output.is_empty(), "test log capture received no datagrams");
3211 output
3212 }
3213
3214 #[test]
3215 fn duplicate_certificate_log_redacts_names() {
3216 const NAME_SECRET: &str = "DUPLICATE_CERTIFICATE_NAME_SECRET_SENTINEL";
3217
3218 let name = format!("{NAME_SECRET}{}", "x".repeat(4096));
3219 let name_len = name.len();
3220 let certificate = CertificateAndKey {
3221 certificate: include_str!("../assets/certificate.pem").to_owned(),
3222 key: include_str!("../assets/key.pem").to_owned(),
3223 certificate_chain: Vec::new(),
3224 versions: Vec::new(),
3225 names: vec![name],
3226 };
3227 let fingerprint = certificate
3228 .fingerprint()
3229 .expect("test certificate fingerprint must be computable")
3230 .to_string();
3231 let output = capture_test_logs(move || {
3232 let mut state = ConfigState::default();
3233 let add = AddCertificate {
3234 address: SocketAddress::new_v4(127, 0, 0, 1, 8443),
3235 certificate,
3236 expired_at: None,
3237 };
3238
3239 state
3240 .add_certificate(&add)
3241 .expect("first certificate insertion must succeed");
3242 state
3243 .add_certificate(&add)
3244 .expect("duplicate certificate insertion must be skipped");
3245 });
3246
3247 assert!(
3248 !output.contains(NAME_SECRET),
3249 "duplicate certificate log leaked certificate name {NAME_SECRET}"
3250 );
3251 assert!(
3252 !output.contains(&fingerprint),
3253 "duplicate certificate log leaked certificate fingerprint {fingerprint}"
3254 );
3255 for metadata in [
3256 "fingerprint_bytes=32".to_owned(),
3257 "names_count=1".to_owned(),
3258 format!("names_bytes={name_len}"),
3259 ] {
3260 assert!(
3261 output.contains(&metadata),
3262 "duplicate certificate log omitted bounded metadata {metadata}: {output}"
3263 );
3264 }
3265 assert!(
3266 output.len() <= 512,
3267 "duplicate certificate log is not bounded: {} bytes",
3268 output.len()
3269 );
3270 }
3271
3272 #[test]
3273 fn frontend_state_error_redacts_display_key() {
3274 const HOSTNAME_SECRET: &str = "STATE_ERROR_HOSTNAME_SECRET_SENTINEL";
3275 const PATH_SECRET: &str = "STATE_ERROR_PATH_SECRET_SENTINEL";
3276 const METHOD_SECRET: &str = "STATE_ERROR_METHOD_SECRET_SENTINEL";
3277
3278 let long_value = |marker: &str| format!("{marker}{}", "x".repeat(4096));
3279 let front = RequestHttpFrontend {
3280 cluster_id: Some("cluster".to_owned()),
3281 address: SocketAddress::new_v4(127, 0, 0, 1, 8080),
3282 hostname: long_value(HOSTNAME_SECRET),
3283 path: PathRule {
3284 kind: PathRuleKind::Prefix as i32,
3285 value: long_value(PATH_SECRET),
3286 },
3287 method: Some(long_value(METHOD_SECRET)),
3288 position: RulePosition::Tree as i32,
3289 tags: BTreeMap::new(),
3290 redirect: None,
3291 redirect_scheme: None,
3292 redirect_template: None,
3293 rewrite_host: None,
3294 rewrite_path: None,
3295 rewrite_port: None,
3296 required_auth: None,
3297 headers: Vec::new(),
3298 hsts: None,
3299 };
3300 let mut state = ConfigState::default();
3301 state
3302 .add_http_frontend(&front)
3303 .expect("first frontend insertion must succeed");
3304 assert!(
3305 state
3306 .http_fronts
3307 .keys()
3308 .next()
3309 .expect("raw state key must be retained")
3310 .contains(HOSTNAME_SECRET),
3311 "state storage must retain the raw operator-facing frontend key"
3312 );
3313
3314 let error = state
3315 .add_http_frontend(&front)
3316 .expect_err("duplicate frontend insertion must return StateError::Exists");
3317 let raw_frontend_key = front.to_string();
3318 match &error {
3319 StateError::Exists { id, .. } => assert_eq!(
3320 id, &raw_frontend_key,
3321 "StateError must preserve the public raw frontend identifier"
3322 ),
3323 other => panic!("expected StateError::Exists, got {other:?}"),
3324 }
3325 for (label, output) in [
3326 ("Display", error.to_string()),
3327 ("Debug", format!("{error:?}")),
3328 ] {
3329 for secret in [HOSTNAME_SECRET, PATH_SECRET, METHOD_SECRET] {
3330 assert!(
3331 !output.contains(secret),
3332 "StateError {label} leaked frontend marker {secret}"
3333 );
3334 }
3335 let metadata = format!("id_bytes={}", raw_frontend_key.len());
3336 assert!(
3337 output.contains(&metadata),
3338 "StateError {label} omitted bounded metadata {metadata}: {output}"
3339 );
3340 assert!(
3341 output.len() <= 512,
3342 "StateError {label} output is not bounded: {} bytes",
3343 output.len()
3344 );
3345 }
3346 }
3347
3348 #[test]
3349 fn state_error_formatting_bounds_every_string_bearing_variant() {
3350 const SECRET: &str = "STATE_ERROR_VARIANT_SECRET_SENTINEL";
3351
3352 let long_value = || format!("{SECRET}{}", "x".repeat(4096));
3353 let errors = [
3354 StateError::NotFound {
3355 kind: ObjectKind::Backend,
3356 id: long_value(),
3357 },
3358 StateError::Exists {
3359 kind: ObjectKind::Cluster,
3360 id: long_value(),
3361 },
3362 StateError::AddCertificate(CertificateError::ParsePEMCertificate(long_value())),
3363 StateError::RemoveCertificate(long_value()),
3364 StateError::ReplaceCertificate(long_value()),
3365 StateError::FrontendConversion {
3366 frontend: long_value(),
3367 error: long_value(),
3368 },
3369 StateError::FileError(std::io::Error::other(long_value())),
3370 StateError::InvalidTcpFrontend {
3371 address: "127.0.0.1:443"
3372 .parse()
3373 .expect("test TCP frontend address must parse"),
3374 reason: long_value(),
3375 },
3376 ];
3377
3378 for error in errors {
3379 for (label, output) in [
3380 ("Display", error.to_string()),
3381 ("Debug", format!("{error:?}")),
3382 ] {
3383 assert!(
3384 !output.contains(SECRET),
3385 "StateError {label} leaked a string-bearing variant: {output}"
3386 );
3387 assert!(
3388 output.len() <= 512,
3389 "StateError {label} is not bounded: {} bytes",
3390 output.len()
3391 );
3392 }
3393 }
3394 }
3395
3396 #[test]
3397 fn tcp_frontend_not_found_error_retains_the_raw_identity() {
3398 const TCP_SECRET: &str = "STATE_ERROR_TCP_IDENTITY_SECRET_SENTINEL";
3399
3400 let long_value = |suffix: &str| format!("{TCP_SECRET}_{suffix}{}", "x".repeat(1024));
3401 let frontend = RequestTcpFrontend {
3402 cluster_id: long_value("cluster"),
3403 address: SocketAddress::new_v4(127, 0, 0, 1, 443),
3404 tags: BTreeMap::from([(long_value("tag-key"), long_value("tag-value"))]),
3405 sni: Some(long_value("sni")),
3406 alpn: vec![long_value("alpn")],
3407 };
3408 let error = ConfigState::default()
3409 .remove_tcp_frontend(&frontend)
3410 .expect_err("removing from a missing cluster must return StateError::NotFound");
3411
3412 match &error {
3413 StateError::NotFound { id, .. } => assert!(
3414 id.contains(TCP_SECRET),
3415 "StateError must retain the raw TCP frontend identity: {id}"
3416 ),
3417 other => panic!("expected StateError::NotFound, got {other:?}"),
3418 }
3419 for output in [error.to_string(), format!("{error:?}")] {
3420 assert!(
3421 !output.contains(TCP_SECRET),
3422 "StateError formatting leaked the retained TCP identity: {output}"
3423 );
3424 assert!(
3425 output.len() <= 512,
3426 "StateError formatting is not bounded: {} bytes",
3427 output.len()
3428 );
3429 }
3430 }
3431
3432 #[test]
3433 fn retained_state_debug_redacts_http_frontend_and_collection_data() {
3434 const CLUSTER_SECRET: &str = "RETAINED_CLUSTER_SECRET_SENTINEL";
3435 const HOSTNAME_SECRET: &str = "RETAINED_HOSTNAME_SECRET_SENTINEL";
3436 const PATH_SECRET: &str = "RETAINED_PATH_SECRET_SENTINEL";
3437 const METHOD_SECRET: &str = "RETAINED_METHOD_SECRET_SENTINEL";
3438 const TAG_KEY_SECRET: &str = "RETAINED_TAG_KEY_SECRET_SENTINEL";
3439 const TAG_VALUE_SECRET: &str = "RETAINED_TAG_VALUE_SECRET_SENTINEL";
3440 const REDIRECT_TEMPLATE_SECRET: &str = "RETAINED_REDIRECT_TEMPLATE_SECRET_SENTINEL";
3441 const REWRITE_HOST_SECRET: &str = "RETAINED_REWRITE_HOST_SECRET_SENTINEL";
3442 const REWRITE_PATH_SECRET: &str = "RETAINED_REWRITE_PATH_SECRET_SENTINEL";
3443 const HEADER_KEY_SECRET: &str = "RETAINED_HEADER_KEY_SECRET_SENTINEL";
3444 const HEADER_VALUE_SECRET: &str = "RETAINED_HEADER_VALUE_SECRET_SENTINEL";
3445 const REQUEST_COUNT_SECRET: &str = "RETAINED_REQUEST_COUNT_SECRET_SENTINEL";
3446
3447 let long_value = |marker: &str| format!("{marker}{}", "x".repeat(4096));
3448 let request = RequestHttpFrontend {
3449 cluster_id: Some(long_value(CLUSTER_SECRET)),
3450 address: SocketAddress::new_v4(127, 0, 0, 1, 8443),
3451 hostname: long_value(HOSTNAME_SECRET),
3452 path: PathRule {
3453 kind: PathRuleKind::Regex as i32,
3454 value: long_value(PATH_SECRET),
3455 },
3456 method: Some(long_value(METHOD_SECRET)),
3457 position: RulePosition::Tree as i32,
3458 tags: BTreeMap::from([(long_value(TAG_KEY_SECRET), long_value(TAG_VALUE_SECRET))]),
3459 redirect: Some(RedirectPolicy::PermanentRedirect as i32),
3460 redirect_scheme: Some(RedirectScheme::UseHttps as i32),
3461 redirect_template: Some(long_value(REDIRECT_TEMPLATE_SECRET)),
3462 rewrite_host: Some(long_value(REWRITE_HOST_SECRET)),
3463 rewrite_path: Some(long_value(REWRITE_PATH_SECRET)),
3464 rewrite_port: Some(9443),
3465 required_auth: Some(true),
3466 headers: vec![Header {
3467 position: HeaderPosition::Both as i32,
3468 key: long_value(HEADER_KEY_SECRET),
3469 val: long_value(HEADER_VALUE_SECRET),
3470 }],
3471 hsts: Some(HstsConfig {
3472 enabled: Some(true),
3473 max_age: Some(31_536_000),
3474 include_subdomains: Some(true),
3475 preload: Some(false),
3476 force_replace_backend: Some(true),
3477 }),
3478 };
3479 let retained = request
3480 .clone()
3481 .to_frontend()
3482 .expect("adversarial frontend must convert");
3483 let retained_debug = format!("{retained:?}");
3484
3485 let mut state = ConfigState::default();
3486 state
3487 .dispatch(&RequestType::AddHttpFrontend(request).into())
3488 .expect("adversarial frontend must be retained");
3489 state
3490 .request_counts
3491 .insert(long_value(REQUEST_COUNT_SECRET), 7);
3492 let retained_key = state
3493 .http_fronts
3494 .keys()
3495 .next()
3496 .expect("dispatch must retain the frontend's Display key");
3497 assert!(retained_key.contains(HOSTNAME_SECRET));
3498 assert!(retained_key.contains(PATH_SECRET));
3499 assert!(retained_key.contains(METHOD_SECRET));
3500 let state_debug = format!("{state:?}");
3501
3502 let secrets = [
3503 CLUSTER_SECRET,
3504 HOSTNAME_SECRET,
3505 PATH_SECRET,
3506 METHOD_SECRET,
3507 TAG_KEY_SECRET,
3508 TAG_VALUE_SECRET,
3509 REDIRECT_TEMPLATE_SECRET,
3510 REWRITE_HOST_SECRET,
3511 REWRITE_PATH_SECRET,
3512 HEADER_KEY_SECRET,
3513 HEADER_VALUE_SECRET,
3514 REQUEST_COUNT_SECRET,
3515 ];
3516 for (label, output) in [
3517 ("HttpFrontend", &retained_debug),
3518 ("ConfigState", &state_debug),
3519 ] {
3520 for secret in secrets {
3521 assert!(
3522 !output.contains(secret),
3523 "{label} Debug leaked retained marker {secret}"
3524 );
3525 }
3526 assert!(
3527 output.len() <= 2048,
3528 "{label} Debug output is not bounded: {} bytes",
3529 output.len()
3530 );
3531 }
3532
3533 for metadata in [
3534 format!("cluster_id_len: Some({})", long_value(CLUSTER_SECRET).len()),
3535 "address: 127.0.0.1:8443".to_owned(),
3536 format!("hostname_len: {}", long_value(HOSTNAME_SECRET).len()),
3537 "path_kind: 1".to_owned(),
3538 format!("path_len: {}", long_value(PATH_SECRET).len()),
3539 format!("method_len: Some({})", long_value(METHOD_SECRET).len()),
3540 "position: Tree".to_owned(),
3541 "tags_count: Some(1)".to_owned(),
3542 format!(
3543 "tags_len: Some({})",
3544 long_value(TAG_KEY_SECRET).len() + long_value(TAG_VALUE_SECRET).len()
3545 ),
3546 "redirect: Some(4)".to_owned(),
3547 "redirect_scheme: Some(2)".to_owned(),
3548 format!(
3549 "redirect_template_len: Some({})",
3550 long_value(REDIRECT_TEMPLATE_SECRET).len()
3551 ),
3552 format!(
3553 "rewrite_host_len: Some({})",
3554 long_value(REWRITE_HOST_SECRET).len()
3555 ),
3556 format!(
3557 "rewrite_path_len: Some({})",
3558 long_value(REWRITE_PATH_SECRET).len()
3559 ),
3560 "rewrite_port: Some(9443)".to_owned(),
3561 "required_auth: Some(true)".to_owned(),
3562 "headers_count: 1".to_owned(),
3563 format!(
3564 "headers_len: {}",
3565 long_value(HEADER_KEY_SECRET).len() + long_value(HEADER_VALUE_SECRET).len()
3566 ),
3567 "hsts: Some(HstsConfig".to_owned(),
3568 ] {
3569 assert!(
3570 retained_debug.contains(&metadata),
3571 "HttpFrontend Debug omitted safe metadata {metadata}: {retained_debug}"
3572 );
3573 }
3574 for metadata in [
3575 "http_frontends_count: 1",
3576 "backends_count: 0",
3577 "tcp_frontends_count: 0",
3578 "udp_frontends_count: 0",
3579 "certificates_count: 0",
3580 "request_counts_count: 2",
3581 ] {
3582 assert!(
3583 state_debug.contains(metadata),
3584 "ConfigState Debug omitted count metadata {metadata}: {state_debug}"
3585 );
3586 }
3587 }
3588
3589 #[test]
3590 fn serialize() {
3591 let mut state: ConfigState = Default::default();
3592 state
3593 .dispatch(
3594 &RequestType::AddHttpFrontend(RequestHttpFrontend {
3595 cluster_id: Some(String::from("cluster_1")),
3596 hostname: String::from("lolcatho.st:8080"),
3597 path: PathRule::prefix(String::from("/")),
3598 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3599 position: RulePosition::Tree.into(),
3600 ..Default::default()
3601 })
3602 .into(),
3603 )
3604 .expect("Could not execute request");
3605 state
3606 .dispatch(
3607 &RequestType::AddHttpFrontend(RequestHttpFrontend {
3608 cluster_id: Some(String::from("cluster_2")),
3609 hostname: String::from("test.local"),
3610 path: PathRule::prefix(String::from("/abc")),
3611 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3612 position: RulePosition::Pre.into(),
3613 ..Default::default()
3614 })
3615 .into(),
3616 )
3617 .expect("Could not execute request");
3618 state
3619 .dispatch(
3620 &RequestType::AddBackend(AddBackend {
3621 cluster_id: String::from("cluster_1"),
3622 backend_id: String::from("cluster_1-0"),
3623 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
3624 ..Default::default()
3625 })
3626 .into(),
3627 )
3628 .expect("Could not execute request");
3629 state
3630 .dispatch(
3631 &RequestType::AddBackend(AddBackend {
3632 cluster_id: String::from("cluster_1"),
3633 backend_id: String::from("cluster_1-1"),
3634 address: SocketAddress::new_v4(127, 0, 0, 1, 1027),
3635 ..Default::default()
3636 })
3637 .into(),
3638 )
3639 .expect("Could not execute request");
3640 state
3641 .dispatch(
3642 &RequestType::AddBackend(AddBackend {
3643 cluster_id: String::from("cluster_2"),
3644 backend_id: String::from("cluster_2-0"),
3645 address: SocketAddress::new_v4(192, 167, 1, 2, 1026),
3646 ..Default::default()
3647 })
3648 .into(),
3649 )
3650 .expect("Could not execute request");
3651 state
3652 .dispatch(
3653 &RequestType::AddBackend(AddBackend {
3654 cluster_id: String::from("cluster_1"),
3655 backend_id: String::from("cluster_1-3"),
3656 address: SocketAddress::new_v4(192, 168, 1, 3, 1027),
3657 ..Default::default()
3658 })
3659 .into(),
3660 )
3661 .expect("Could not execute request");
3662 state
3663 .dispatch(
3664 &RequestType::RemoveBackend(RemoveBackend {
3665 cluster_id: String::from("cluster_1"),
3666 backend_id: String::from("cluster_1-3"),
3667 address: SocketAddress::new_v4(192, 168, 1, 3, 1027),
3668 })
3669 .into(),
3670 )
3671 .expect("Could not execute request");
3672
3673 }
3683
3684 #[test]
3685 fn diff() {
3686 let mut state: ConfigState = Default::default();
3687 state
3688 .dispatch(
3689 &RequestType::AddHttpFrontend(RequestHttpFrontend {
3690 cluster_id: Some(String::from("cluster_1")),
3691 hostname: String::from("lolcatho.st:8080"),
3692 path: PathRule::prefix(String::from("/")),
3693 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3694 position: RulePosition::Post.into(),
3695 ..Default::default()
3696 })
3697 .into(),
3698 )
3699 .expect("Could not execute request");
3700 state
3701 .dispatch(
3702 &RequestType::AddHttpFrontend(RequestHttpFrontend {
3703 cluster_id: Some(String::from("cluster_2")),
3704 hostname: String::from("test.local"),
3705 path: PathRule::prefix(String::from("/abc")),
3706 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3707 ..Default::default()
3708 })
3709 .into(),
3710 )
3711 .expect("Could not execute request");
3712 state
3713 .dispatch(
3714 &RequestType::AddBackend(AddBackend {
3715 cluster_id: String::from("cluster_1"),
3716 backend_id: String::from("cluster_1-0"),
3717 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
3718 load_balancing_parameters: Some(LoadBalancingParams::default()),
3719 ..Default::default()
3720 })
3721 .into(),
3722 )
3723 .expect("Could not execute request");
3724 state
3725 .dispatch(
3726 &RequestType::AddBackend(AddBackend {
3727 cluster_id: String::from("cluster_1"),
3728 backend_id: String::from("cluster_1-1"),
3729 address: SocketAddress::new_v4(127, 0, 0, 2, 1027),
3730 load_balancing_parameters: Some(LoadBalancingParams::default()),
3731 ..Default::default()
3732 })
3733 .into(),
3734 )
3735 .expect("Could not execute request");
3736 state
3737 .dispatch(
3738 &RequestType::AddBackend(AddBackend {
3739 cluster_id: String::from("cluster_2"),
3740 backend_id: String::from("cluster_2-0"),
3741 address: SocketAddress::new_v4(192, 167, 1, 2, 1026),
3742 load_balancing_parameters: Some(LoadBalancingParams::default()),
3743 ..Default::default()
3744 })
3745 .into(),
3746 )
3747 .expect("Could not execute request");
3748 state
3749 .dispatch(
3750 &RequestType::AddCluster(Cluster {
3751 cluster_id: String::from("cluster_2"),
3752 sticky_session: true,
3753 https_redirect: true,
3754 ..Default::default()
3755 })
3756 .into(),
3757 )
3758 .expect("Could not execute request");
3759
3760 let mut state2: ConfigState = Default::default();
3761 state2
3762 .dispatch(
3763 &RequestType::AddHttpFrontend(RequestHttpFrontend {
3764 cluster_id: Some(String::from("cluster_1")),
3765 hostname: String::from("lolcatho.st:8080"),
3766 path: PathRule::prefix(String::from("/")),
3767 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3768 position: RulePosition::Post.into(),
3769 ..Default::default()
3770 })
3771 .into(),
3772 )
3773 .expect("Could not execute request");
3774 state2
3775 .dispatch(
3776 &RequestType::AddBackend(AddBackend {
3777 cluster_id: String::from("cluster_1"),
3778 backend_id: String::from("cluster_1-0"),
3779 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
3780 load_balancing_parameters: Some(LoadBalancingParams::default()),
3781 ..Default::default()
3782 })
3783 .into(),
3784 )
3785 .expect("Could not execute request");
3786 state2
3787 .dispatch(
3788 &RequestType::AddBackend(AddBackend {
3789 cluster_id: String::from("cluster_1"),
3790 backend_id: String::from("cluster_1-1"),
3791 address: SocketAddress::new_v4(127, 0, 0, 2, 1027),
3792 load_balancing_parameters: Some(LoadBalancingParams::default()),
3793 ..Default::default()
3794 })
3795 .into(),
3796 )
3797 .expect("Could not execute request");
3798 state2
3799 .dispatch(
3800 &RequestType::AddBackend(AddBackend {
3801 cluster_id: String::from("cluster_1"),
3802 backend_id: String::from("cluster_1-2"),
3803 address: SocketAddress::new_v4(127, 0, 0, 2, 1028),
3804 load_balancing_parameters: Some(LoadBalancingParams::default()),
3805 ..Default::default()
3806 })
3807 .into(),
3808 )
3809 .expect("Could not execute request");
3810 state2
3811 .dispatch(
3812 &RequestType::AddCluster(Cluster {
3813 cluster_id: String::from("cluster_3"),
3814 sticky_session: false,
3815 https_redirect: false,
3816 ..Default::default()
3817 })
3818 .into(),
3819 )
3820 .expect("Could not execute request");
3821
3822 let e: Vec<Request> = vec![
3823 RequestType::RemoveHttpFrontend(RequestHttpFrontend {
3824 cluster_id: Some(String::from("cluster_2")),
3825 hostname: String::from("test.local"),
3826 path: PathRule::prefix(String::from("/abc")),
3827 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3828 ..Default::default()
3829 })
3830 .into(),
3831 RequestType::RemoveBackend(RemoveBackend {
3832 cluster_id: String::from("cluster_2"),
3833 backend_id: String::from("cluster_2-0"),
3834 address: SocketAddress::new_v4(192, 167, 1, 2, 1026),
3835 })
3836 .into(),
3837 RequestType::AddBackend(AddBackend {
3838 cluster_id: String::from("cluster_1"),
3839 backend_id: String::from("cluster_1-2"),
3840 address: SocketAddress::new_v4(127, 0, 0, 2, 1028),
3841 load_balancing_parameters: Some(LoadBalancingParams::default()),
3842 ..Default::default()
3843 })
3844 .into(),
3845 RequestType::RemoveCluster(String::from("cluster_2")).into(),
3846 RequestType::AddCluster(Cluster {
3847 cluster_id: String::from("cluster_3"),
3848 sticky_session: false,
3849 https_redirect: false,
3850 ..Default::default()
3851 })
3852 .into(),
3853 ];
3854 let expected_diff: HashSet<&Request> = HashSet::from_iter(e.iter());
3855
3856 let d = state.diff(&state2);
3857 let diff = HashSet::from_iter(d.iter());
3858 println!("diff requests:\n{diff:#?}\n");
3859 println!("expected diff requests:\n{expected_diff:#?}\n");
3860
3861 let hash1 = state.hash_state();
3862 let hash2 = state2.hash_state();
3863 let mut state3 = state.clone();
3864 state3
3865 .dispatch(
3866 &RequestType::AddBackend(AddBackend {
3867 cluster_id: String::from("cluster_1"),
3868 backend_id: String::from("cluster_1-2"),
3869 address: SocketAddress::new_v4(127, 0, 0, 2, 1028),
3870 load_balancing_parameters: Some(LoadBalancingParams::default()),
3871 ..Default::default()
3872 })
3873 .into(),
3874 )
3875 .expect("Could not execute request");
3876 let hash3 = state3.hash_state();
3877 println!("state 1 hashes: {hash1:#?}");
3878 println!("state 2 hashes: {hash2:#?}");
3879 println!("state 3 hashes: {hash3:#?}");
3880
3881 assert_eq!(diff, expected_diff);
3882 }
3883
3884 #[test]
3885 fn cluster_ids_by_domain() {
3886 let mut config = ConfigState::new();
3887 let http_front_cluster1 = RequestHttpFrontend {
3888 cluster_id: Some(String::from("MyCluster_1")),
3889 hostname: String::from("lolcatho.st"),
3890 path: PathRule::prefix(String::from("")),
3891 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3892 ..Default::default()
3893 };
3894
3895 let https_front_cluster1 = RequestHttpFrontend {
3896 cluster_id: Some(String::from("MyCluster_1")),
3897 hostname: String::from("lolcatho.st"),
3898 path: PathRule::prefix(String::from("")),
3899 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
3900 ..Default::default()
3901 };
3902
3903 let http_front_cluster2 = RequestHttpFrontend {
3904 cluster_id: Some(String::from("MyCluster_2")),
3905 hostname: String::from("lolcatho.st"),
3906 path: PathRule::prefix(String::from("/api")),
3907 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3908 ..Default::default()
3909 };
3910
3911 let https_front_cluster2 = RequestHttpFrontend {
3912 cluster_id: Some(String::from("MyCluster_2")),
3913 hostname: String::from("lolcatho.st"),
3914 path: PathRule::prefix(String::from("/api")),
3915 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
3916 ..Default::default()
3917 };
3918
3919 config
3920 .dispatch(&RequestType::AddHttpFrontend(http_front_cluster1).into())
3921 .expect("Could not execute request");
3922 config
3923 .dispatch(&RequestType::AddHttpFrontend(http_front_cluster2).into())
3924 .expect("Could not execute request");
3925 config
3926 .dispatch(&RequestType::AddHttpsFrontend(https_front_cluster1).into())
3927 .expect("Could not execute request");
3928 config
3929 .dispatch(&RequestType::AddHttpsFrontend(https_front_cluster2).into())
3930 .expect("Could not execute request");
3931
3932 let mut cluster1_cluster2: HashSet<ClusterId> = HashSet::new();
3933 cluster1_cluster2.insert(String::from("MyCluster_1"));
3934 cluster1_cluster2.insert(String::from("MyCluster_2"));
3935
3936 let mut cluster2: HashSet<ClusterId> = HashSet::new();
3937 cluster2.insert(String::from("MyCluster_2"));
3938
3939 let empty: HashSet<ClusterId> = HashSet::new();
3940 assert_eq!(
3941 config.get_cluster_ids_by_domain(String::from("lolcatho.st"), None),
3942 cluster1_cluster2
3943 );
3944 assert_eq!(
3945 config
3946 .get_cluster_ids_by_domain(String::from("lolcatho.st"), Some(String::from("/api"))),
3947 cluster2
3948 );
3949 assert_eq!(
3950 config.get_cluster_ids_by_domain(String::from("lolcathost"), None),
3951 empty
3952 );
3953 assert_eq!(
3954 config
3955 .get_cluster_ids_by_domain(String::from("lolcathost"), Some(String::from("/sozu"))),
3956 empty
3957 );
3958 }
3959
3960 #[test]
3961 fn duplicate_backends() {
3962 let mut state: ConfigState = Default::default();
3963 state
3964 .dispatch(
3965 &RequestType::AddBackend(AddBackend {
3966 cluster_id: String::from("cluster_1"),
3967 backend_id: String::from("cluster_1-0"),
3968 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
3969 load_balancing_parameters: Some(LoadBalancingParams::default()),
3970 ..Default::default()
3971 })
3972 .into(),
3973 )
3974 .expect("Could not execute request");
3975
3976 let b = Backend {
3977 cluster_id: String::from("cluster_1"),
3978 backend_id: String::from("cluster_1-0"),
3979 address: "127.0.0.1:1026".parse().unwrap(),
3980 load_balancing_parameters: Some(LoadBalancingParams::default()),
3981 sticky_id: Some("sticky".to_string()),
3982 backup: None,
3983 };
3984
3985 state
3986 .dispatch(&RequestType::AddBackend(b.clone().to_add_backend()).into())
3987 .expect("Could not execute order");
3988
3989 assert_eq!(state.backends.get("cluster_1").unwrap(), &vec![b]);
3990 }
3991
3992 #[test]
3993 fn remove_backend() {
3994 let mut state: ConfigState = Default::default();
3995 state
3996 .dispatch(
3997 &RequestType::AddCluster(Cluster {
3998 cluster_id: String::from("cluster_1"),
3999 ..Default::default()
4000 })
4001 .into(),
4002 )
4003 .expect("Could not execute request");
4004
4005 for i in 0..10 {
4006 state
4007 .dispatch(
4008 &RequestType::AddBackend(AddBackend {
4009 cluster_id: String::from("cluster_1"),
4010 backend_id: format!("cluster_1-{i}"),
4011 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4012 ..Default::default()
4013 })
4014 .into(),
4015 )
4016 .expect("Could not execute request");
4017 }
4018
4019 assert_eq!(state.backends.get("cluster_1").unwrap().len(), 10);
4020
4021 let remove_backend_2 = RequestType::RemoveBackend(RemoveBackend {
4022 cluster_id: String::from("cluster_1"),
4023 backend_id: String::from("cluster_1-0"),
4024 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4025 })
4026 .into();
4027
4028 let remove_backend_result = state.dispatch(&remove_backend_2);
4029
4030 assert!(remove_backend_result.is_ok());
4031 assert_eq!(state.backends.get("cluster_1").unwrap().len(), 9);
4032
4033 let redundant_remove = state.dispatch(&remove_backend_2);
4034 assert!(matches!(redundant_remove, Err(StateError::NoChange)));
4035 assert_eq!(state.backends.get("cluster_1").unwrap().len(), 9);
4036 }
4037
4038 #[test]
4039 fn remove_backends_randomly() {
4040 let mut state: ConfigState = Default::default();
4041 state
4042 .dispatch(
4043 &RequestType::AddCluster(Cluster {
4044 cluster_id: String::from("cluster_1"),
4045 ..Default::default()
4046 })
4047 .into(),
4048 )
4049 .expect("Could not execute request");
4050
4051 for _ in 0..1000 {
4052 for i in 0..10 {
4053 state
4054 .dispatch(
4055 &RequestType::AddBackend(AddBackend {
4056 cluster_id: String::from("cluster_1"),
4057 backend_id: format!("cluster_1-{i}"),
4058 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4059 ..Default::default()
4060 })
4061 .into(),
4062 )
4063 .expect("Could not execute request");
4064 }
4065
4066 let mut rng = rng();
4067 let mut indexes = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9];
4068 indexes.shuffle(&mut rng);
4069 let random_count = rng.random_range(1..indexes.len());
4070 let random_indexes: Vec<i32> = indexes.into_iter().take(random_count).collect();
4071
4072 for j in random_indexes {
4073 let remove_backend_result = state.dispatch(
4074 &RequestType::RemoveBackend(RemoveBackend {
4075 cluster_id: String::from("cluster_1"),
4076 backend_id: format!("cluster_1-{j}"),
4077 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4078 })
4079 .into(),
4080 );
4081 assert!(remove_backend_result.is_ok());
4082 }
4083 }
4084 }
4085
4086 #[test]
4087 fn listener_diff() {
4088 let mut state: ConfigState = Default::default();
4089 let custom_http_answers = Some(CustomHttpAnswers {
4090 answer_404: Some("test".to_string()),
4091 ..Default::default()
4092 });
4093 state
4094 .dispatch(
4095 &RequestType::AddTcpListener(TcpListenerConfig {
4096 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4097 ..Default::default()
4098 })
4099 .into(),
4100 )
4101 .expect("Could not execute request");
4102 state
4103 .dispatch(
4104 &RequestType::ActivateListener(ActivateListener {
4105 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4106 proxy: ListenerType::Tcp.into(),
4107 from_scm: false,
4108 })
4109 .into(),
4110 )
4111 .expect("Could not execute request");
4112 state
4113 .dispatch(
4114 &RequestType::AddHttpListener(HttpListenerConfig {
4115 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4116 ..Default::default()
4117 })
4118 .into(),
4119 )
4120 .expect("Could not execute request");
4121 state
4122 .dispatch(
4123 &RequestType::AddHttpsListener(HttpsListenerConfig {
4124 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4125 ..Default::default()
4126 })
4127 .into(),
4128 )
4129 .expect("Could not execute request");
4130 state
4131 .dispatch(
4132 &RequestType::ActivateListener(ActivateListener {
4133 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4134 proxy: ListenerType::Https.into(),
4135 from_scm: false,
4136 })
4137 .into(),
4138 )
4139 .expect("Could not execute request");
4140
4141 let mut state2: ConfigState = Default::default();
4142 state2
4143 .dispatch(
4144 &RequestType::AddTcpListener(TcpListenerConfig {
4145 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4146 expect_proxy: true,
4147 ..Default::default()
4148 })
4149 .into(),
4150 )
4151 .expect("Could not execute request");
4152 state2
4153 .dispatch(
4154 &RequestType::AddHttpListener(HttpListenerConfig {
4155 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4156 http_answers: custom_http_answers.clone(),
4157 ..Default::default()
4158 })
4159 .into(),
4160 )
4161 .expect("Could not execute request");
4162 state2
4163 .dispatch(
4164 &RequestType::ActivateListener(ActivateListener {
4165 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4166 proxy: ListenerType::Http.into(),
4167 from_scm: false,
4168 })
4169 .into(),
4170 )
4171 .expect("Could not execute request");
4172 state2
4173 .dispatch(
4174 &RequestType::AddHttpsListener(HttpsListenerConfig {
4175 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4176 http_answers: custom_http_answers.clone(),
4177 ..Default::default()
4178 })
4179 .into(),
4180 )
4181 .expect("Could not execute request");
4182 state2
4183 .dispatch(
4184 &RequestType::ActivateListener(ActivateListener {
4185 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4186 proxy: ListenerType::Https.into(),
4187 from_scm: false,
4188 })
4189 .into(),
4190 )
4191 .expect("Could not execute request");
4192
4193 let e: Vec<Request> = vec![
4194 RequestType::RemoveListener(RemoveListener {
4195 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4196 proxy: ListenerType::Tcp.into(),
4197 })
4198 .into(),
4199 RequestType::AddTcpListener(TcpListenerConfig {
4200 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4201 expect_proxy: true,
4202 ..Default::default()
4203 })
4204 .into(),
4205 RequestType::DeactivateListener(DeactivateListener {
4206 address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4207 proxy: ListenerType::Tcp.into(),
4208 to_scm: false,
4209 })
4210 .into(),
4211 RequestType::RemoveListener(RemoveListener {
4212 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4213 proxy: ListenerType::Http.into(),
4214 })
4215 .into(),
4216 RequestType::AddHttpListener(HttpListenerConfig {
4217 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4218 http_answers: custom_http_answers.clone(),
4219 ..Default::default()
4220 })
4221 .into(),
4222 RequestType::ActivateListener(ActivateListener {
4223 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4224 proxy: ListenerType::Http.into(),
4225 from_scm: false,
4226 })
4227 .into(),
4228 RequestType::RemoveListener(RemoveListener {
4229 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4230 proxy: ListenerType::Https.into(),
4231 })
4232 .into(),
4233 RequestType::AddHttpsListener(HttpsListenerConfig {
4234 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4235 http_answers: custom_http_answers.clone(),
4236 ..Default::default()
4237 })
4238 .into(),
4239 RequestType::ActivateListener(ActivateListener {
4245 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4246 proxy: ListenerType::Https.into(),
4247 from_scm: false,
4248 })
4249 .into(),
4250 ];
4251
4252 let diff = state.diff(&state2);
4253 println!("expected diff requests:\n{e:#?}\n");
4255 println!("diff requests:\n{diff:#?}\n");
4256
4257 let _hash1 = state.hash_state();
4258 let _hash2 = state2.hash_state();
4259
4260 assert_eq!(diff, e);
4261
4262 let mut replayed = state.clone();
4269 for request in &diff {
4270 replayed
4271 .dispatch(request)
4272 .expect("every diff request must replay cleanly onto the source state");
4273 }
4274 assert_eq!(
4275 replayed.tcp_listeners, state2.tcp_listeners,
4276 "replayed tcp_listeners must match the target state"
4277 );
4278 assert_eq!(
4279 replayed.http_listeners, state2.http_listeners,
4280 "replayed http_listeners must match the target state"
4281 );
4282 assert_eq!(
4283 replayed.https_listeners, state2.https_listeners,
4284 "replayed https_listeners must match the target state"
4285 );
4286 let replayed_8443 = replayed
4289 .https_listeners
4290 .get(&SocketAddr::from(SocketAddress::new_v4(0, 0, 0, 0, 8443)))
4291 .expect("8443 HTTPS listener must exist after replay");
4292 assert!(
4293 replayed_8443.active,
4294 "the 8443 HTTPS listener must stay ACTIVE across a config change"
4295 );
4296 }
4297
4298 #[test]
4299 fn certificate_retrieval() {
4300 let mut state: ConfigState = Default::default();
4301 let certificate_and_key = CertificateAndKey {
4302 certificate: String::from(include_str!("../assets/certificate.pem")),
4303 key: String::from(include_str!("../assets/key.pem")),
4304 certificate_chain: vec![],
4305 versions: vec![],
4306 names: vec!["lolcatho.st".to_string()],
4307 };
4308 let add_certificate = AddCertificate {
4309 address: SocketAddress::new_v4(127, 0, 0, 1, 8080),
4310 certificate: certificate_and_key,
4311 expired_at: None,
4312 };
4313 state
4314 .dispatch(&RequestType::AddCertificate(add_certificate).into())
4315 .expect("Could not add certificate");
4316
4317 println!("state: {state:#?}");
4318
4319 let certificates_found_by_fingerprint = state.get_certificates(QueryCertificatesFilters {
4325 domain: None,
4326 fingerprint: Some(
4327 "ab2618b674e15243fd02a5618c66509e4840ba60e7d64cebec84cdbfeceee0c5".to_string(),
4328 ),
4329 });
4330
4331 println!("found certificate: {certificates_found_by_fingerprint:#?}");
4332
4333 assert!(!certificates_found_by_fingerprint.is_empty());
4334
4335 let certificate_found_by_domain_name = state.get_certificates(QueryCertificatesFilters {
4336 domain: Some("lolcatho.st".to_string()),
4337 fingerprint: None,
4338 });
4339
4340 assert!(!certificate_found_by_domain_name.is_empty());
4341 }
4342
4343 #[test]
4344 fn count_backends_across_clusters() {
4345 let mut state: ConfigState = Default::default();
4346
4347 assert_eq!(state.count_backends(), 0);
4348
4349 state
4350 .dispatch(
4351 &RequestType::AddBackend(AddBackend {
4352 cluster_id: String::from("cluster_1"),
4353 backend_id: String::from("cluster_1-0"),
4354 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4355 ..Default::default()
4356 })
4357 .into(),
4358 )
4359 .expect("Could not execute request");
4360 assert_eq!(state.count_backends(), 1);
4361
4362 state
4363 .dispatch(
4364 &RequestType::AddBackend(AddBackend {
4365 cluster_id: String::from("cluster_1"),
4366 backend_id: String::from("cluster_1-1"),
4367 address: SocketAddress::new_v4(127, 0, 0, 1, 1027),
4368 ..Default::default()
4369 })
4370 .into(),
4371 )
4372 .expect("Could not execute request");
4373 assert_eq!(state.count_backends(), 2);
4374
4375 state
4377 .dispatch(
4378 &RequestType::AddBackend(AddBackend {
4379 cluster_id: String::from("cluster_2"),
4380 backend_id: String::from("cluster_2-0"),
4381 address: SocketAddress::new_v4(192, 168, 1, 1, 8080),
4382 ..Default::default()
4383 })
4384 .into(),
4385 )
4386 .expect("Could not execute request");
4387 assert_eq!(state.count_backends(), 3);
4388
4389 state
4391 .dispatch(
4392 &RequestType::RemoveBackend(RemoveBackend {
4393 cluster_id: String::from("cluster_1"),
4394 backend_id: String::from("cluster_1-0"),
4395 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4396 })
4397 .into(),
4398 )
4399 .expect("Could not execute request");
4400 assert_eq!(state.count_backends(), 2);
4401 }
4402
4403 #[test]
4404 fn count_frontends_across_types() {
4405 let mut state: ConfigState = Default::default();
4406
4407 assert_eq!(state.count_frontends(), 0);
4408
4409 state
4411 .dispatch(
4412 &RequestType::AddHttpFrontend(RequestHttpFrontend {
4413 cluster_id: Some(String::from("cluster_1")),
4414 hostname: String::from("example.com"),
4415 path: PathRule::prefix(String::from("/")),
4416 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4417 position: RulePosition::Tree.into(),
4418 ..Default::default()
4419 })
4420 .into(),
4421 )
4422 .expect("Could not execute request");
4423 assert_eq!(state.count_frontends(), 1);
4424
4425 state
4427 .dispatch(
4428 &RequestType::AddHttpsFrontend(RequestHttpFrontend {
4429 cluster_id: Some(String::from("cluster_1")),
4430 hostname: String::from("secure.example.com"),
4431 path: PathRule::prefix(String::from("/")),
4432 address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4433 position: RulePosition::Tree.into(),
4434 ..Default::default()
4435 })
4436 .into(),
4437 )
4438 .expect("Could not execute request");
4439 assert_eq!(state.count_frontends(), 2);
4440
4441 state
4443 .dispatch(
4444 &RequestType::AddTcpFrontend(RequestTcpFrontend {
4445 cluster_id: String::from("cluster_2"),
4446 address: SocketAddress::new_v4(0, 0, 0, 0, 5432),
4447 ..Default::default()
4448 })
4449 .into(),
4450 )
4451 .expect("Could not execute request");
4452 assert_eq!(state.count_frontends(), 3);
4453
4454 state
4456 .dispatch(
4457 &RequestType::AddTcpFrontend(RequestTcpFrontend {
4458 cluster_id: String::from("cluster_2"),
4459 address: SocketAddress::new_v4(0, 0, 0, 0, 5433),
4460 ..Default::default()
4461 })
4462 .into(),
4463 )
4464 .expect("Could not execute request");
4465 assert_eq!(state.count_frontends(), 4);
4466
4467 state
4469 .dispatch(
4470 &RequestType::RemoveHttpFrontend(RequestHttpFrontend {
4471 cluster_id: Some(String::from("cluster_1")),
4472 hostname: String::from("example.com"),
4473 path: PathRule::prefix(String::from("/")),
4474 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4475 position: RulePosition::Tree.into(),
4476 ..Default::default()
4477 })
4478 .into(),
4479 )
4480 .expect("Could not execute request");
4481 assert_eq!(state.count_frontends(), 3);
4482 }
4483
4484 fn make_https_listener(address: SocketAddress) -> HttpsListenerConfig {
4487 HttpsListenerConfig {
4488 address,
4489 sticky_name: "SOZUBALANCEID".to_owned(),
4490 front_timeout: 60,
4491 back_timeout: 30,
4492 connect_timeout: 3,
4493 request_timeout: 10,
4494 ..Default::default()
4495 }
4496 }
4497
4498 fn make_http_listener(address: SocketAddress) -> HttpListenerConfig {
4499 HttpListenerConfig {
4500 address,
4501 sticky_name: "SOZUBALANCEID".to_owned(),
4502 front_timeout: 60,
4503 back_timeout: 30,
4504 connect_timeout: 3,
4505 request_timeout: 10,
4506 ..Default::default()
4507 }
4508 }
4509
4510 fn make_tcp_listener(address: SocketAddress) -> TcpListenerConfig {
4511 TcpListenerConfig {
4512 address,
4513 front_timeout: 60,
4514 back_timeout: 30,
4515 connect_timeout: 3,
4516 ..Default::default()
4517 }
4518 }
4519
4520 fn make_udp_listener(address: SocketAddress, active: bool) -> UdpListenerConfig {
4521 UdpListenerConfig {
4522 address,
4523 public_address: None,
4524 front_timeout: 30,
4525 back_timeout: 30,
4526 max_rx_datagram_size: 1500,
4527 max_flows: 0,
4528 active,
4529 }
4530 }
4531
4532 #[test]
4542 fn test_udp_state_roundtrip() {
4543 let address = SocketAddress::new_v4(127, 0, 0, 1, 5353);
4544
4545 let mut state = ConfigState::default();
4546 state
4547 .dispatch(&RequestType::AddUdpListener(make_udp_listener(address, true)).into())
4548 .expect("could not add udp listener");
4549 state
4550 .dispatch(
4551 &RequestType::ActivateListener(ActivateListener {
4552 address,
4553 proxy: ListenerType::Udp.into(),
4554 from_scm: false,
4555 })
4556 .into(),
4557 )
4558 .expect("could not activate udp listener");
4559 state
4560 .dispatch(
4561 &RequestType::AddUdpFrontend(RequestUdpFrontend {
4562 cluster_id: "udp_cluster".to_string(),
4563 address,
4564 tags: BTreeMap::from([("owner".to_string(), "team".to_string())]),
4565 })
4566 .into(),
4567 )
4568 .expect("could not add udp frontend");
4569
4570 assert_eq!(state.udp_listeners.len(), 1);
4571 assert!(state.udp_listeners[&address.into()].active);
4572 assert_eq!(
4573 state.udp_fronts.get("udp_cluster").map(Vec::len),
4574 Some(1usize)
4575 );
4576
4577 let logical = |s: &ConfigState| {
4582 let mut c = s.clone();
4583 c.request_counts.clear();
4584 c
4585 };
4586
4587 let mut replayed = ConfigState::default();
4589 for request in state.generate_requests() {
4590 replayed
4591 .dispatch(&request)
4592 .expect("could not replay generated request");
4593 }
4594 assert_eq!(
4595 logical(&state),
4596 logical(&replayed),
4597 "UDP listener + frontend must survive generate_requests → replay"
4598 );
4599
4600 let empty = ConfigState::default();
4602 let mut from_diff = ConfigState::default();
4603 for request in empty.diff(&state) {
4604 from_diff
4605 .dispatch(&request)
4606 .expect("could not replay diff request");
4607 }
4608 assert_eq!(
4609 logical(&state),
4610 logical(&from_diff),
4611 "diff(empty -> state) must reconstruct the UDP listener + frontend"
4612 );
4613
4614 let mut torn_down = state.clone();
4616 for request in state.diff(&empty) {
4617 torn_down
4618 .dispatch(&request)
4619 .expect("could not replay teardown diff request");
4620 }
4621 assert!(
4622 torn_down.udp_listeners.is_empty(),
4623 "diff(state -> empty) must remove the UDP listener"
4624 );
4625 assert!(
4626 torn_down
4627 .udp_fronts
4628 .get("udp_cluster")
4629 .map(Vec::is_empty)
4630 .unwrap_or(true),
4631 "diff(state -> empty) must remove the UDP frontend"
4632 );
4633
4634 state
4636 .dispatch(
4637 &RequestType::UpdateUdpListener(UpdateUdpListenerConfig {
4638 address,
4639 max_flows: Some(4096),
4640 front_timeout: Some(15),
4641 ..Default::default()
4642 })
4643 .into(),
4644 )
4645 .expect("could not update udp listener");
4646 let updated = &state.udp_listeners[&address.into()];
4647 assert_eq!(updated.max_flows, 4096);
4648 assert_eq!(updated.front_timeout, 15);
4649 assert_eq!(
4650 updated.back_timeout, 30,
4651 "unpatched field must be preserved"
4652 );
4653 }
4654
4655 #[test]
4670 fn test_tcp_sni_alpn_state_roundtrip() {
4671 let address = SocketAddress::new_v4(127, 0, 0, 1, 6443);
4672
4673 let mut state = ConfigState::default();
4674 state
4675 .dispatch(
4676 &RequestType::AddTcpListener(TcpListenerConfig {
4677 address,
4678 front_timeout: 60,
4679 back_timeout: 30,
4680 connect_timeout: 3,
4681 active: true,
4682 sni_preread_timeout: Some(2),
4683 sni_preread_max_bytes: Some(8192),
4684 ..Default::default()
4685 })
4686 .into(),
4687 )
4688 .expect("could not add tcp listener");
4689 state
4690 .dispatch(
4691 &RequestType::ActivateListener(ActivateListener {
4692 address,
4693 proxy: ListenerType::Tcp.into(),
4694 from_scm: false,
4695 })
4696 .into(),
4697 )
4698 .expect("could not activate tcp listener");
4699
4700 let frontend_a = RequestTcpFrontend {
4701 cluster_id: "cluster_sni_a".to_string(),
4702 address,
4703 tags: BTreeMap::new(),
4704 sni: Some("example.com".to_string()),
4705 alpn: vec!["h2".to_string()],
4706 };
4707 let frontend_b = RequestTcpFrontend {
4708 cluster_id: "cluster_sni_b".to_string(),
4709 address,
4710 tags: BTreeMap::new(),
4711 sni: Some("other.example.com".to_string()),
4712 alpn: vec![],
4713 };
4714 state
4715 .dispatch(&RequestType::AddTcpFrontend(frontend_a.clone()).into())
4716 .expect("could not add sni-a tcp frontend");
4717 state
4718 .dispatch(&RequestType::AddTcpFrontend(frontend_b.clone()).into())
4719 .expect("could not add sni-b tcp frontend");
4720
4721 assert_eq!(state.tcp_listeners.len(), 1);
4722 assert!(state.tcp_listeners[&address.into()].active);
4723 assert_eq!(
4724 state.tcp_listeners[&address.into()].sni_preread_timeout,
4725 Some(2)
4726 );
4727 assert_eq!(
4728 state.tcp_listeners[&address.into()].sni_preread_max_bytes,
4729 Some(8192)
4730 );
4731 assert_eq!(state.count_tcp_frontends_raw(), 2);
4732
4733 let logical = |s: &ConfigState| {
4738 let mut c = s.clone();
4739 c.request_counts.clear();
4740 c
4741 };
4742
4743 let mut replayed = ConfigState::default();
4745 for request in state.generate_requests() {
4746 replayed
4747 .dispatch(&request)
4748 .expect("could not replay generated request");
4749 }
4750 assert_eq!(
4751 logical(&state),
4752 logical(&replayed),
4753 "SNI/ALPN TCP listener + frontends must survive generate_requests → replay"
4754 );
4755
4756 let empty = ConfigState::default();
4758 let mut from_diff = ConfigState::default();
4759 for request in empty.diff(&state) {
4760 from_diff
4761 .dispatch(&request)
4762 .expect("could not replay diff request");
4763 }
4764 assert_eq!(
4765 logical(&state),
4766 logical(&from_diff),
4767 "diff(empty -> state) must reconstruct the tcp listener + both sni frontends"
4768 );
4769
4770 let mut torn_down = state.clone();
4772 for request in state.diff(&empty) {
4773 torn_down
4774 .dispatch(&request)
4775 .expect("could not replay teardown diff request");
4776 }
4777 assert!(
4778 torn_down.tcp_listeners.is_empty(),
4779 "diff(state -> empty) must remove the tcp listener"
4780 );
4781 assert_eq!(
4782 torn_down.count_tcp_frontends_raw(),
4783 0,
4784 "diff(state -> empty) must remove both sni frontends"
4785 );
4786
4787 let mismatched_removal = state.dispatch(
4791 &RequestType::RemoveTcpFrontend(RequestTcpFrontend {
4792 cluster_id: "cluster_sni_a".to_string(),
4793 address,
4794 tags: BTreeMap::new(),
4795 sni: None,
4796 alpn: vec![],
4797 })
4798 .into(),
4799 );
4800 assert!(
4801 mismatched_removal.is_err(),
4802 "removing by address alone (no matching sni/alpn) must be rejected"
4803 );
4804 assert_eq!(
4805 state.count_tcp_frontends_raw(),
4806 2,
4807 "a rejected mismatched removal must not change either sibling"
4808 );
4809
4810 state
4813 .dispatch(&RequestType::RemoveTcpFrontend(frontend_a).into())
4814 .expect("could not remove sni-a tcp frontend");
4815 assert_eq!(state.count_tcp_frontends_raw(), 1);
4816 let remaining = state
4817 .tcp_fronts
4818 .get("cluster_sni_b")
4819 .expect("cluster_sni_b bucket must survive");
4820 assert_eq!(remaining.len(), 1);
4821 assert_eq!(remaining[0].sni, Some("other.example.com".to_string()));
4822 }
4823
4824 fn add_tcp_test_cluster(state: &mut ConfigState, cluster_id: &str) {
4835 state
4836 .dispatch(
4837 &RequestType::AddCluster(Cluster {
4838 cluster_id: cluster_id.to_string(),
4839 sticky_session: false,
4840 https_redirect: false,
4841 ..Default::default()
4842 })
4843 .into(),
4844 )
4845 .expect("could not add cluster");
4846 }
4847
4848 #[test]
4853 fn add_tcp_frontend_rejects_cross_cluster_duplicate_identity() {
4854 let address = SocketAddress::new_v4(127, 0, 0, 1, 9001);
4855 let mut state = ConfigState::new();
4856 add_tcp_test_cluster(&mut state, "cluster_x");
4857 add_tcp_test_cluster(&mut state, "cluster_y");
4858
4859 state
4860 .dispatch(
4861 &RequestType::AddTcpFrontend(RequestTcpFrontend {
4862 cluster_id: "cluster_x".to_string(),
4863 address,
4864 tags: BTreeMap::new(),
4865 sni: Some("example.com".to_string()),
4866 alpn: vec!["h2".to_string()],
4867 })
4868 .into(),
4869 )
4870 .expect("could not add cluster_x tcp frontend");
4871
4872 let hash_before = state.hash_state();
4873 let count_before = state.count_tcp_frontends_raw();
4874
4875 let result = state.dispatch(
4876 &RequestType::AddTcpFrontend(RequestTcpFrontend {
4877 cluster_id: "cluster_y".to_string(),
4878 address,
4879 tags: BTreeMap::new(),
4880 sni: Some("example.com".to_string()),
4881 alpn: vec!["h2".to_string()],
4882 })
4883 .into(),
4884 );
4885 assert!(
4886 result.is_err(),
4887 "the same (address, sni, alpn) under a different cluster_id must be rejected, \
4888 got: {result:?}"
4889 );
4890 assert_eq!(
4891 state.hash_state(),
4892 hash_before,
4893 "a rejected cross-cluster duplicate must not change the state hash"
4894 );
4895 assert_eq!(
4896 state.count_tcp_frontends_raw(),
4897 count_before,
4898 "a rejected cross-cluster duplicate must not change the frontend count"
4899 );
4900 assert_eq!(state.tcp_fronts.get("cluster_x").map(Vec::len), Some(1));
4901 assert_eq!(
4902 state.tcp_fronts.get("cluster_y").map(Vec::len).unwrap_or(0),
4903 0
4904 );
4905 }
4906
4907 #[test]
4910 fn add_tcp_frontend_rejects_reordered_alpn_duplicate() {
4911 let address = SocketAddress::new_v4(127, 0, 0, 1, 9002);
4912 let mut state = ConfigState::new();
4913 add_tcp_test_cluster(&mut state, "cluster_reorder");
4914
4915 state
4916 .dispatch(
4917 &RequestType::AddTcpFrontend(RequestTcpFrontend {
4918 cluster_id: "cluster_reorder".to_string(),
4919 address,
4920 tags: BTreeMap::new(),
4921 sni: Some("example.com".to_string()),
4922 alpn: vec!["h2".to_string(), "http/1.1".to_string()],
4923 })
4924 .into(),
4925 )
4926 .expect("could not add first tcp frontend");
4927
4928 let hash_before = state.hash_state();
4929 let count_before = state.count_tcp_frontends_raw();
4930
4931 let result = state.dispatch(
4932 &RequestType::AddTcpFrontend(RequestTcpFrontend {
4933 cluster_id: "cluster_reorder".to_string(),
4934 address,
4935 tags: BTreeMap::new(),
4936 sni: Some("example.com".to_string()),
4937 alpn: vec!["http/1.1".to_string(), "h2".to_string()],
4938 })
4939 .into(),
4940 );
4941 assert!(
4942 result.is_err(),
4943 "the same ALPN set in a different order must be rejected as a duplicate \
4944 identity, got: {result:?}"
4945 );
4946 assert_eq!(state.hash_state(), hash_before);
4947 assert_eq!(state.count_tcp_frontends_raw(), count_before);
4948 assert_eq!(
4949 state.tcp_fronts.get("cluster_reorder").map(Vec::len),
4950 Some(1)
4951 );
4952 }
4953
4954 #[test]
4958 fn remove_tcp_frontend_matches_reordered_alpn() {
4959 let address = SocketAddress::new_v4(127, 0, 0, 1, 9003);
4960 let mut state = ConfigState::new();
4961 add_tcp_test_cluster(&mut state, "cluster_remove_reorder");
4962
4963 state
4964 .dispatch(
4965 &RequestType::AddTcpFrontend(RequestTcpFrontend {
4966 cluster_id: "cluster_remove_reorder".to_string(),
4967 address,
4968 tags: BTreeMap::new(),
4969 sni: Some("example.com".to_string()),
4970 alpn: vec!["h2".to_string(), "http/1.1".to_string()],
4971 })
4972 .into(),
4973 )
4974 .expect("could not add tcp frontend");
4975
4976 let result = state.dispatch(
4977 &RequestType::RemoveTcpFrontend(RequestTcpFrontend {
4978 cluster_id: "cluster_remove_reorder".to_string(),
4979 address,
4980 tags: BTreeMap::new(),
4981 sni: Some("example.com".to_string()),
4982 alpn: vec!["http/1.1".to_string(), "h2".to_string()],
4983 })
4984 .into(),
4985 );
4986 assert!(
4987 result.is_ok(),
4988 "removing with the same ALPN set in a different order must succeed, got: {result:?}"
4989 );
4990 assert_eq!(state.count_tcp_frontends_raw(), 0);
4991 }
4992
4993 #[test]
4997 fn add_tcp_frontend_rejects_alpn_overlap_across_clusters_same_sni() {
4998 let address = SocketAddress::new_v4(127, 0, 0, 1, 9004);
4999 let mut state = ConfigState::new();
5000 add_tcp_test_cluster(&mut state, "cluster_overlap_a");
5001 add_tcp_test_cluster(&mut state, "cluster_overlap_b");
5002
5003 state
5004 .dispatch(
5005 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5006 cluster_id: "cluster_overlap_a".to_string(),
5007 address,
5008 tags: BTreeMap::new(),
5009 sni: Some("example.com".to_string()),
5010 alpn: vec!["h2".to_string()],
5011 })
5012 .into(),
5013 )
5014 .expect("could not add cluster_overlap_a tcp frontend");
5015
5016 let hash_before = state.hash_state();
5017 let count_before = state.count_tcp_frontends_raw();
5018
5019 let result = state.dispatch(
5020 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5021 cluster_id: "cluster_overlap_b".to_string(),
5022 address,
5023 tags: BTreeMap::new(),
5024 sni: Some("example.com".to_string()),
5025 alpn: vec!["h2".to_string(), "http/1.1".to_string()],
5026 })
5027 .into(),
5028 );
5029 assert!(
5030 result.is_err(),
5031 "an overlapping ALPN set on the same (address, sni) under a different cluster_id \
5032 must be rejected, got: {result:?}"
5033 );
5034 assert_eq!(state.hash_state(), hash_before);
5035 assert_eq!(state.count_tcp_frontends_raw(), count_before);
5036 assert_eq!(
5037 state
5038 .tcp_fronts
5039 .get("cluster_overlap_b")
5040 .map(Vec::len)
5041 .unwrap_or(0),
5042 0
5043 );
5044 }
5045
5046 #[test]
5049 fn add_tcp_frontend_rejects_no_sni_sni_mixing_both_directions() {
5050 let address_a = SocketAddress::new_v4(127, 0, 0, 1, 9005);
5051 let address_b = SocketAddress::new_v4(127, 0, 0, 1, 9006);
5052 let mut state = ConfigState::new();
5053 add_tcp_test_cluster(&mut state, "cluster_mix_1");
5054 add_tcp_test_cluster(&mut state, "cluster_mix_2");
5055 add_tcp_test_cluster(&mut state, "cluster_mix_3");
5056 add_tcp_test_cluster(&mut state, "cluster_mix_4");
5057
5058 state
5061 .dispatch(
5062 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5063 cluster_id: "cluster_mix_1".to_string(),
5064 address: address_a,
5065 tags: BTreeMap::new(),
5066 sni: None,
5067 alpn: vec![],
5068 })
5069 .into(),
5070 )
5071 .expect("could not add no-sni tcp frontend");
5072 let hash_before_a = state.hash_state();
5073 let count_before_a = state.count_tcp_frontends_raw();
5074 let result_a = state.dispatch(
5075 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5076 cluster_id: "cluster_mix_2".to_string(),
5077 address: address_a,
5078 tags: BTreeMap::new(),
5079 sni: Some("example.com".to_string()),
5080 alpn: vec![],
5081 })
5082 .into(),
5083 );
5084 assert!(
5085 result_a.is_err(),
5086 "an SNI-scoped frontend must not be added to a listener with an existing no-SNI \
5087 front, got: {result_a:?}"
5088 );
5089 assert_eq!(state.hash_state(), hash_before_a);
5090 assert_eq!(state.count_tcp_frontends_raw(), count_before_a);
5091
5092 state
5095 .dispatch(
5096 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5097 cluster_id: "cluster_mix_3".to_string(),
5098 address: address_b,
5099 tags: BTreeMap::new(),
5100 sni: Some("example.com".to_string()),
5101 alpn: vec![],
5102 })
5103 .into(),
5104 )
5105 .expect("could not add sni-scoped tcp frontend");
5106 let hash_before_b = state.hash_state();
5107 let count_before_b = state.count_tcp_frontends_raw();
5108 let result_b = state.dispatch(
5109 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5110 cluster_id: "cluster_mix_4".to_string(),
5111 address: address_b,
5112 tags: BTreeMap::new(),
5113 sni: None,
5114 alpn: vec![],
5115 })
5116 .into(),
5117 );
5118 assert!(
5119 result_b.is_err(),
5120 "a no-SNI frontend must not be added to a listener with existing SNI-scoped \
5121 fronts, got: {result_b:?}"
5122 );
5123 assert_eq!(state.hash_state(), hash_before_b);
5124 assert_eq!(state.count_tcp_frontends_raw(), count_before_b);
5125 }
5126
5127 #[test]
5132 fn add_tcp_frontend_rejects_malformed_sni_shape() {
5133 let address = SocketAddress::new_v4(127, 0, 0, 1, 9007);
5134 let mut state = ConfigState::new();
5135 add_tcp_test_cluster(&mut state, "cluster_malformed");
5136
5137 let hash_before = state.hash_state();
5138 let count_before = state.count_tcp_frontends_raw();
5139
5140 let result = state.dispatch(
5141 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5142 cluster_id: "cluster_malformed".to_string(),
5143 address,
5144 tags: BTreeMap::new(),
5145 sni: Some(".example.com".to_string()),
5147 alpn: vec![],
5148 })
5149 .into(),
5150 );
5151 assert!(
5152 result.is_err(),
5153 "a malformed SNI pattern must be rejected by the master, not merely by \
5154 config-load, got: {result:?}"
5155 );
5156 assert_eq!(state.hash_state(), hash_before);
5157 assert_eq!(state.count_tcp_frontends_raw(), count_before);
5158 assert_eq!(
5159 state
5160 .tcp_fronts
5161 .get("cluster_malformed")
5162 .map(Vec::len)
5163 .unwrap_or(0),
5164 0
5165 );
5166 }
5167
5168 #[test]
5174 fn add_tcp_frontend_rejects_alpn_without_sni() {
5175 let address = SocketAddress::new_v4(127, 0, 0, 1, 9009);
5176 let mut state = ConfigState::new();
5177 add_tcp_test_cluster(&mut state, "cluster_alpn_no_sni");
5178
5179 let hash_before = state.hash_state();
5180 let count_before = state.count_tcp_frontends_raw();
5181
5182 let result = state.dispatch(
5183 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5184 cluster_id: "cluster_alpn_no_sni".to_string(),
5185 address,
5186 tags: BTreeMap::new(),
5187 sni: None,
5188 alpn: vec!["h2".to_string()],
5189 })
5190 .into(),
5191 );
5192 assert!(
5193 result.is_err(),
5194 "a non-empty alpn without sni must be rejected by the master, got: {result:?}"
5195 );
5196 assert_eq!(
5197 state.hash_state(),
5198 hash_before,
5199 "a rejected alpn-without-sni add must not change the state hash"
5200 );
5201 assert_eq!(
5202 state.count_tcp_frontends_raw(),
5203 count_before,
5204 "a rejected alpn-without-sni add must not change the frontend count"
5205 );
5206 assert_eq!(
5207 state
5208 .tcp_fronts
5209 .get("cluster_alpn_no_sni")
5210 .map(Vec::len)
5211 .unwrap_or(0),
5212 0
5213 );
5214 }
5215
5216 #[test]
5222 fn add_tcp_frontend_accepts_exact_and_wildcard_catch_all_siblings() {
5223 let address = SocketAddress::new_v4(127, 0, 0, 1, 9008);
5224 let mut state = ConfigState::new();
5225 add_tcp_test_cluster(&mut state, "cluster_exact_catchall");
5226 add_tcp_test_cluster(&mut state, "cluster_wildcard_catchall");
5227
5228 state
5229 .dispatch(
5230 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5231 cluster_id: "cluster_exact_catchall".to_string(),
5232 address,
5233 tags: BTreeMap::new(),
5234 sni: Some("example.com".to_string()),
5235 alpn: vec![],
5236 })
5237 .into(),
5238 )
5239 .expect("could not add exact catch-all tcp frontend");
5240
5241 let result = state.dispatch(
5242 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5243 cluster_id: "cluster_wildcard_catchall".to_string(),
5244 address,
5245 tags: BTreeMap::new(),
5246 sni: Some("*.example.com".to_string()),
5247 alpn: vec![],
5248 })
5249 .into(),
5250 );
5251 assert!(
5252 result.is_ok(),
5253 "an exact catch-all and a wildcard catch-all on the same address must both be \
5254 accepted, got: {result:?}"
5255 );
5256 assert_eq!(state.count_tcp_frontends_raw(), 2);
5257 }
5258
5259 #[test]
5263 fn list_frontends_includes_udp() {
5264 let tcp_addr = SocketAddress::new_v4(0, 0, 0, 0, 6379);
5265 let udp_addr = SocketAddress::new_v4(0, 0, 0, 0, 5353);
5266
5267 let mut state = ConfigState::default();
5268 state
5269 .dispatch(
5270 &RequestType::AddTcpFrontend(RequestTcpFrontend {
5271 cluster_id: "tcp_cluster".to_string(),
5272 address: tcp_addr,
5273 ..Default::default()
5274 })
5275 .into(),
5276 )
5277 .expect("could not add tcp frontend");
5278 state
5279 .dispatch(
5280 &RequestType::AddUdpFrontend(RequestUdpFrontend {
5281 cluster_id: "udp_cluster".to_string(),
5282 address: udp_addr,
5283 ..Default::default()
5284 })
5285 .into(),
5286 )
5287 .expect("could not add udp frontend");
5288
5289 let all = state.list_frontends(FrontendFilters::default());
5291 assert_eq!(all.tcp_frontends.len(), 1, "tcp frontend must be listed");
5292 assert_eq!(
5293 all.udp_frontends.len(),
5294 1,
5295 "udp frontend must be listed under the default all-pass path"
5296 );
5297 assert_eq!(all.udp_frontends[0].cluster_id, "udp_cluster");
5298 assert_eq!(all.udp_frontends[0].address, udp_addr);
5299
5300 let tcp_only = state.list_frontends(FrontendFilters {
5302 tcp: true,
5303 ..Default::default()
5304 });
5305 assert_eq!(tcp_only.tcp_frontends.len(), 1);
5306 assert_eq!(
5307 tcp_only.udp_frontends.len(),
5308 1,
5309 "udp frontends ride the tcp filter"
5310 );
5311
5312 let http_only = state.list_frontends(FrontendFilters {
5314 http: true,
5315 ..Default::default()
5316 });
5317 assert!(http_only.tcp_frontends.is_empty());
5318 assert!(http_only.udp_frontends.is_empty());
5319
5320 let domain_filtered = state.list_frontends(FrontendFilters {
5322 domain: Some("example.com".to_string()),
5323 ..Default::default()
5324 });
5325 assert!(domain_filtered.tcp_frontends.is_empty());
5326 assert!(domain_filtered.udp_frontends.is_empty());
5327 }
5328
5329 #[test]
5334 fn update_https_listener_happy_path_h2_knobs() {
5335 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5336 let mut state = ConfigState::new();
5337 state
5338 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5339 .unwrap();
5340
5341 let patch = UpdateHttpsListenerConfig {
5342 address: addr,
5343 h2_max_rst_stream_per_window: Some(50),
5344 h2_max_ping_per_window: Some(20),
5345 ..Default::default()
5346 };
5347 state
5348 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5349 .expect("update must succeed");
5350
5351 let listener = state
5352 .https_listeners
5353 .get(&SocketAddr::from(addr))
5354 .expect("listener must be present");
5355 assert_eq!(listener.h2_max_rst_stream_per_window, Some(50));
5356 assert_eq!(listener.h2_max_ping_per_window, Some(20));
5357 assert_eq!(listener.front_timeout, 60);
5359 assert_eq!(listener.h2_max_settings_per_window, None);
5360 }
5361
5362 #[test]
5365 fn update_https_listener_not_found() {
5366 let mut state = ConfigState::new();
5367 let patch = UpdateHttpsListenerConfig {
5368 address: SocketAddress::new_v4(1, 2, 3, 4, 9999),
5369 h2_max_rst_stream_per_window: Some(50),
5370 ..Default::default()
5371 };
5372 let err = state
5373 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5374 .unwrap_err();
5375 assert!(
5376 matches!(
5377 err,
5378 StateError::NotFound {
5379 kind: ObjectKind::HttpsListener,
5380 ..
5381 }
5382 ),
5383 "expected NotFound, got: {err}"
5384 );
5385 }
5386
5387 #[test]
5390 fn update_https_listener_noop_patch() {
5391 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5392 let mut state = ConfigState::new();
5393 let original = make_https_listener(addr);
5394 state
5395 .dispatch(&RequestType::AddHttpsListener(original.clone()).into())
5396 .unwrap();
5397
5398 let patch = UpdateHttpsListenerConfig {
5399 address: addr,
5400 ..Default::default()
5401 };
5402 state
5403 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5404 .expect("no-op patch must succeed");
5405
5406 let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5407 assert_eq!(listener.front_timeout, original.front_timeout);
5408 assert_eq!(
5409 listener.h2_max_rst_stream_per_window,
5410 original.h2_max_rst_stream_per_window
5411 );
5412 }
5413
5414 #[test]
5416 fn update_https_listener_invalid_value_flood_knob_zero() {
5417 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5418 let mut state = ConfigState::new();
5419 state
5420 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5421 .unwrap();
5422
5423 let patch = UpdateHttpsListenerConfig {
5424 address: addr,
5425 h2_max_rst_stream_per_window: Some(0),
5426 ..Default::default()
5427 };
5428 let err = state
5429 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5430 .unwrap_err();
5431 assert!(
5432 matches!(
5433 err,
5434 StateError::InvalidValue {
5435 field: "h2_max_rst_stream_per_window",
5436 ..
5437 }
5438 ),
5439 "expected InvalidValue for flood knob 0, got: {err}"
5440 );
5441 }
5442
5443 #[test]
5449 fn add_cluster_invalid_health_check_uri_rejected() {
5450 use crate::proto::command::HealthCheckConfig;
5451
5452 let mut state = ConfigState::new();
5453 let err = state
5454 .dispatch(
5455 &RequestType::AddCluster(Cluster {
5456 cluster_id: String::from("evil_cluster"),
5457 health_check: Some(HealthCheckConfig {
5458 uri: String::from("/foo\r\nGET /admin"),
5459 interval: 5_000,
5460 timeout: 1_000,
5461 healthy_threshold: 2,
5462 unhealthy_threshold: 2,
5463 ..Default::default()
5464 }),
5465 ..Default::default()
5466 })
5467 .into(),
5468 )
5469 .unwrap_err();
5470
5471 assert!(
5472 matches!(
5473 err,
5474 StateError::InvalidValue {
5475 field: "health_check",
5476 ..
5477 }
5478 ),
5479 "expected InvalidValue for CRLF-bearing health-check URI, got: {err}"
5480 );
5481 assert!(
5482 !state.clusters.contains_key("evil_cluster"),
5483 "cluster must not be inserted when health_check fails validation",
5484 );
5485 }
5486
5487 #[test]
5489 fn update_https_listener_alpn_unknown_value_rejected() {
5490 use crate::proto::command::AlpnProtocols;
5491
5492 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5493 let mut state = ConfigState::new();
5494 state
5495 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5496 .unwrap();
5497
5498 let patch = UpdateHttpsListenerConfig {
5499 address: addr,
5500 alpn_protocols: Some(AlpnProtocols {
5501 values: vec!["h3".to_owned()],
5502 }),
5503 ..Default::default()
5504 };
5505 let err = state
5506 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5507 .unwrap_err();
5508 assert!(
5509 matches!(
5510 err,
5511 StateError::InvalidValue {
5512 field: "alpn_protocols",
5513 ..
5514 }
5515 ),
5516 "expected InvalidValue for unknown ALPN, got: {err}"
5517 );
5518 }
5519
5520 #[test]
5522 fn update_https_listener_alpn_empty_reset_accepted() {
5523 use crate::proto::command::AlpnProtocols;
5524
5525 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5526 let mut state = ConfigState::new();
5527 let mut listener = make_https_listener(addr);
5528 listener.alpn_protocols = vec!["h2".to_owned()];
5529 state
5530 .dispatch(&RequestType::AddHttpsListener(listener).into())
5531 .unwrap();
5532
5533 let patch = UpdateHttpsListenerConfig {
5534 address: addr,
5535 alpn_protocols: Some(AlpnProtocols { values: vec![] }),
5536 ..Default::default()
5537 };
5538 state
5539 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5540 .expect("empty ALPN reset must succeed");
5541
5542 let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5543 assert!(
5544 listener.alpn_protocols.is_empty(),
5545 "ALPN must have been reset to empty"
5546 );
5547 }
5548
5549 #[test]
5551 fn update_https_listener_alpn_valid_values_accepted() {
5552 use crate::proto::command::AlpnProtocols;
5553
5554 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5555 let mut state = ConfigState::new();
5556 state
5557 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5558 .unwrap();
5559
5560 let patch = UpdateHttpsListenerConfig {
5561 address: addr,
5562 alpn_protocols: Some(AlpnProtocols {
5563 values: vec!["h2".to_owned(), "http/1.1".to_owned()],
5564 }),
5565 ..Default::default()
5566 };
5567 state
5568 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5569 .expect("valid ALPN must be accepted");
5570
5571 let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5572 assert_eq!(listener.alpn_protocols, vec!["h2", "http/1.1"]);
5573 }
5574
5575 #[test]
5578 fn update_https_listener_alpn_absent_preserves_existing() {
5579 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5580 let mut state = ConfigState::new();
5581 let mut listener = make_https_listener(addr);
5582 listener.alpn_protocols = vec!["h2".to_owned()];
5583 state
5584 .dispatch(&RequestType::AddHttpsListener(listener).into())
5585 .unwrap();
5586
5587 let patch = UpdateHttpsListenerConfig {
5589 address: addr,
5590 front_timeout: Some(10),
5591 ..Default::default()
5592 };
5593 state
5594 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5595 .unwrap();
5596
5597 let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5598 assert_eq!(
5599 listener.alpn_protocols,
5600 vec!["h2"],
5601 "ALPN must be preserved when not patched"
5602 );
5603 }
5604
5605 #[test]
5607 fn update_https_listener_sozu_id_header_empty_rejected() {
5608 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5609 let mut state = ConfigState::new();
5610 state
5611 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5612 .unwrap();
5613
5614 let patch = UpdateHttpsListenerConfig {
5615 address: addr,
5616 sozu_id_header: Some(String::new()),
5617 ..Default::default()
5618 };
5619 let err = state
5620 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5621 .unwrap_err();
5622 assert!(
5623 matches!(
5624 err,
5625 StateError::InvalidValue {
5626 field: "sozu_id_header",
5627 ..
5628 }
5629 ),
5630 "expected InvalidValue for empty header name, got: {err}"
5631 );
5632 }
5633
5634 #[test]
5636 fn update_https_listener_sozu_id_header_colon_rejected() {
5637 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5638 let mut state = ConfigState::new();
5639 state
5640 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5641 .unwrap();
5642
5643 let patch = UpdateHttpsListenerConfig {
5644 address: addr,
5645 sozu_id_header: Some("bad: value".to_owned()),
5646 ..Default::default()
5647 };
5648 let err = state
5649 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5650 .unwrap_err();
5651 assert!(
5652 matches!(
5653 err,
5654 StateError::InvalidValue {
5655 field: "sozu_id_header",
5656 ..
5657 }
5658 ),
5659 "expected InvalidValue for header name with colon, got: {err}"
5660 );
5661 }
5662
5663 #[test]
5665 fn update_https_listener_sozu_id_header_valid_accepted() {
5666 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5667 let mut state = ConfigState::new();
5668 state
5669 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5670 .unwrap();
5671
5672 let patch = UpdateHttpsListenerConfig {
5673 address: addr,
5674 sozu_id_header: Some("X-Edge-Id".to_owned()),
5675 ..Default::default()
5676 };
5677 state
5678 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5679 .expect("valid header name must be accepted");
5680
5681 let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5682 assert_eq!(listener.sozu_id_header.as_deref(), Some("X-Edge-Id"));
5683 }
5684
5685 #[test]
5687 fn update_https_listener_graceful_shutdown_zero_allowed() {
5688 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5689 let mut state = ConfigState::new();
5690 state
5691 .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5692 .unwrap();
5693
5694 let patch = UpdateHttpsListenerConfig {
5695 address: addr,
5696 h2_graceful_shutdown_deadline_seconds: Some(0),
5697 ..Default::default()
5698 };
5699 state
5700 .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5701 .expect("graceful_shutdown_deadline=0 must be allowed");
5702
5703 let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5704 assert_eq!(listener.h2_graceful_shutdown_deadline_seconds, Some(0));
5705 }
5706
5707 #[test]
5711 fn update_http_listener_happy_path() {
5712 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
5713 let mut state = ConfigState::new();
5714 state
5715 .dispatch(&RequestType::AddHttpListener(make_http_listener(addr)).into())
5716 .unwrap();
5717
5718 let patch = UpdateHttpListenerConfig {
5719 address: addr,
5720 front_timeout: Some(15),
5721 h2_max_rst_stream_per_window: Some(25),
5722 ..Default::default()
5723 };
5724 state
5725 .dispatch(&RequestType::UpdateHttpListener(patch).into())
5726 .expect("HTTP update must succeed");
5727
5728 let listener = state.http_listeners.get(&SocketAddr::from(addr)).unwrap();
5729 assert_eq!(listener.front_timeout, 15);
5730 assert_eq!(listener.h2_max_rst_stream_per_window, Some(25));
5731 assert_eq!(listener.back_timeout, 30);
5733 }
5734
5735 #[test]
5737 fn update_http_listener_flood_knob_zero_rejected() {
5738 let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
5739 let mut state = ConfigState::new();
5740 state
5741 .dispatch(&RequestType::AddHttpListener(make_http_listener(addr)).into())
5742 .unwrap();
5743
5744 let patch = UpdateHttpListenerConfig {
5745 address: addr,
5746 h2_max_window_update_stream0_per_window: Some(0),
5747 ..Default::default()
5748 };
5749 let err = state
5750 .dispatch(&RequestType::UpdateHttpListener(patch).into())
5751 .unwrap_err();
5752 assert!(
5753 matches!(
5754 err,
5755 StateError::InvalidValue {
5756 field: "h2_max_window_update_stream0_per_window",
5757 ..
5758 }
5759 ),
5760 "expected InvalidValue, got: {err}"
5761 );
5762 }
5763
5764 #[test]
5768 fn update_tcp_listener_happy_path() {
5769 let addr = SocketAddress::new_v4(0, 0, 0, 0, 9000);
5770 let mut state = ConfigState::new();
5771 state
5772 .dispatch(&RequestType::AddTcpListener(make_tcp_listener(addr)).into())
5773 .unwrap();
5774
5775 let patch = UpdateTcpListenerConfig {
5776 address: addr,
5777 front_timeout: Some(5),
5778 ..Default::default()
5779 };
5780 state
5781 .dispatch(&RequestType::UpdateTcpListener(patch).into())
5782 .expect("TCP update must succeed");
5783
5784 let listener = state.tcp_listeners.get(&SocketAddr::from(addr)).unwrap();
5785 assert_eq!(listener.front_timeout, 5);
5786 assert_eq!(listener.back_timeout, 30); }
5788
5789 #[test]
5791 fn update_tcp_listener_not_found() {
5792 let mut state = ConfigState::new();
5793 let patch = UpdateTcpListenerConfig {
5794 address: SocketAddress::new_v4(9, 9, 9, 9, 9999),
5795 front_timeout: Some(5),
5796 ..Default::default()
5797 };
5798 let err = state
5799 .dispatch(&RequestType::UpdateTcpListener(patch).into())
5800 .unwrap_err();
5801 assert!(
5802 matches!(
5803 err,
5804 StateError::NotFound {
5805 kind: ObjectKind::TcpListener,
5806 ..
5807 }
5808 ),
5809 "expected NotFound, got: {err}"
5810 );
5811 }
5812
5813 #[test]
5819 fn dispatch_passes_through_set_metric_detail() {
5820 use crate::proto::command::{MetricDetail, SetMetricDetail};
5821 let mut state = ConfigState::new();
5822 let req: Request = RequestType::SetMetricDetail(SetMetricDetail {
5823 client_id: "test:1".to_owned(),
5824 detail: Some(MetricDetail::DetailBackend as i32),
5825 ttl_seconds: Some(60),
5826 clear: Some(false),
5827 reason: Some("regression-guard".to_owned()),
5828 peer_pid: None,
5829 peer_session_ulid: None,
5830 })
5831 .into();
5832 state
5833 .dispatch(&req)
5834 .expect("SetMetricDetail must traverse dispatch without UndispatchableRequest");
5835 }
5836}