1use std::{
13 cell::RefCell,
14 collections::{BTreeMap, HashMap, hash_map::Entry},
15 io::ErrorKind,
16 net::{Shutdown, SocketAddr as StdSocketAddr},
17 os::unix::io::AsRawFd,
18 rc::{Rc, Weak},
19 str::{from_utf8, from_utf8_unchecked},
20 sync::Arc,
21 time::{Duration, Instant},
22};
23
24use mio::{
25 Interest, Registry, Token,
26 net::{TcpListener as MioTcpListener, TcpStream as MioTcpStream},
27 unix::SourceFd,
28};
29use rustls::{
30 CipherSuite, ProtocolVersion, ServerConfig as RustlsServerConfig, ServerConnection,
31 SupportedCipherSuite, crypto::CryptoProvider,
32};
33use rusty_ulid::Ulid;
34use sozu_command::{
35 certificate::Fingerprint,
36 config::{DEFAULT_ALPN_PROTOCOLS, DEFAULT_CIPHER_LIST},
37 proto::command::{
38 AddCertificate, CertificateSummary, CertificatesByAddress, Cluster, HttpsListenerConfig,
39 ListOfCertificatesByAddress, ListenerType, RemoveCertificate, RemoveListener,
40 ReplaceCertificate, RequestHttpFrontend, ResponseContent, TlsVersion,
41 UpdateHttpsListenerConfig, WorkerRequest, WorkerResponse, request::RequestType,
42 response_content::ContentType,
43 },
44 ready::Ready,
45 response::HttpFrontend,
46 state::{
47 ClusterId, validate_alpn_protocols, validate_h2_flood_knobs_https, validate_sozu_id_header,
48 },
49};
50
51use crate::metrics::names;
52use crate::{
53 AcceptError, CachedTags, FrontendFromRequestError, L7ListenerHandler, L7Proxy, ListenerError,
54 ListenerHandler, Protocol, ProxyConfiguration, ProxyError, ProxySession, SessionIsToBeClosed,
55 SessionMetrics, SessionResult, StateMachineBuilder, StateResult,
56 backends::BackendMap,
57 crypto::{cipher_suite_by_name, default_provider, kx_group_by_name},
58 pool::Pool,
59 protocol::{
60 Pipe, SessionState,
61 http::answers::HttpAnswers,
62 http::parser::{Method, hostname_and_port},
63 mux::{self, Mux, MuxTls},
64 proxy_protocol::expect::ExpectProxyProtocol,
65 rustls::TlsHandshake,
66 },
67 router::{RouteResult, Router},
68 server::{ListenToken, SessionManager},
69 socket::{FrontRustls, server_bind},
70 timer::TimeoutContainer,
71 tls::MutexCertificateResolver,
72 util::UnwrapLog,
73};
74
75StateMachineBuilder! {
76 enum HttpsStateMachine impl SessionState {
83 Expect(ExpectProxyProtocol<MioTcpStream>, ServerConnection),
84 Handshake(TlsHandshake),
85 Mux(MuxTls),
86 WebSocket(Pipe<FrontRustls, HttpsListener>),
87 }
88}
89
90enum AlpnProtocol {
91 H2,
92 Http11,
93}
94
95#[cfg(debug_assertions)]
100fn https_stage_rank(marker: StateMarker) -> u8 {
101 match marker {
102 StateMarker::Expect => 0,
103 StateMarker::Handshake => 1,
104 StateMarker::Mux => 2,
105 StateMarker::WebSocket => 3,
106 }
107}
108
109macro_rules! log_module_context {
116 () => {{
117 let (open, reset, _, _, _) = sozu_command::logging::ansi_palette();
118 format!("{open}HTTPS{reset}\t >>>", open = open, reset = reset)
119 }};
120}
121
122macro_rules! log_context {
128 ($self:expr) => {{
129 let (open, reset, grey, gray, white) = sozu_command::logging::ansi_palette();
130 format!(
131 "{open}HTTPS{reset}\t{grey}Session{reset}({gray}frontend{reset}={white}{frontend}{reset}, {gray}peer{reset}={white}{peer}{reset})\t >>>",
132 open = open,
133 reset = reset,
134 grey = grey,
135 gray = gray,
136 white = white,
137 frontend = $self.frontend_token.0,
138 peer = $self.peer_address.map(|a| a.to_string()).unwrap_or_else(|| "<none>".to_string()),
139 )
140 }};
141}
142
143fn successful_tls_handshake_summary(sni: Option<&str>, alpn: Option<&str>) -> String {
144 format!(
145 "successful TLS handshake (sni_bytes={:?}, alpn_bytes={:?})",
146 sni.map(str::len),
147 alpn.map(str::len),
148 )
149}
150
151pub struct HttpsSession {
152 configured_backend_timeout: Duration,
153 configured_connect_timeout: Duration,
154 configured_frontend_timeout: Duration,
155 frontend_token: Token,
156 has_been_closed: bool,
157 last_event: Instant,
158 listener: Rc<RefCell<HttpsListener>>,
159 metrics: SessionMetrics,
160 peer_address: Option<StdSocketAddr>,
161 pool: Weak<RefCell<Pool>>,
162 proxy: Rc<RefCell<HttpsProxy>>,
163 public_address: StdSocketAddr,
164 state: HttpsStateMachine,
165}
166
167impl HttpsSession {
168 #[allow(clippy::too_many_arguments)]
169 pub fn new(
170 configured_backend_timeout: Duration,
171 configured_connect_timeout: Duration,
172 configured_frontend_timeout: Duration,
173 configured_request_timeout: Duration,
174 expect_proxy: bool,
175 listener: Rc<RefCell<HttpsListener>>,
176 pool: Weak<RefCell<Pool>>,
177 proxy: Rc<RefCell<HttpsProxy>>,
178 public_address: StdSocketAddr,
179 rustls_details: ServerConnection,
180 sock: MioTcpStream,
181 token: Token,
182 wait_time: Duration,
183 ) -> HttpsSession {
184 debug_assert!(
189 !configured_request_timeout.is_zero(),
190 "HTTPS session request timeout must be non-zero (would arm an immediate deadline)"
191 );
192 debug_assert!(
193 !configured_frontend_timeout.is_zero() && !configured_backend_timeout.is_zero(),
194 "HTTPS session front/back timeouts must be non-zero"
195 );
196
197 let peer_address = if expect_proxy {
198 None
200 } else {
201 sock.peer_addr().ok()
202 };
203
204 let request_id = Ulid::generate();
205 let container_frontend_timeout = TimeoutContainer::new(configured_request_timeout, token);
206
207 let state = if expect_proxy {
208 trace!("{} starting in expect proxy state", log_module_context!());
209 gauge_add!(names::protocol::PROXY_EXPECT, 1);
210 HttpsStateMachine::Expect(
211 ExpectProxyProtocol::new(container_frontend_timeout, sock, token, request_id),
212 rustls_details,
213 )
214 } else {
215 gauge_add!(names::protocol::TLS_HANDSHAKE, 1);
216 HttpsStateMachine::Handshake(TlsHandshake::new(
217 container_frontend_timeout,
218 rustls_details,
219 sock,
220 token,
221 request_id,
222 peer_address,
223 ))
224 };
225
226 debug_assert_eq!(
231 matches!(state, HttpsStateMachine::Expect(..)),
232 expect_proxy,
233 "fresh HTTPS session must start in Expect iff expect_proxy is set"
234 );
235 debug_assert!(
236 expect_proxy || matches!(state, HttpsStateMachine::Handshake(_)),
237 "non-expect-proxy HTTPS session must start in the TLS Handshake state"
238 );
239 debug_assert!(
240 !expect_proxy || peer_address.is_none(),
241 "expect-proxy peer address is only known after the PROXY header is parsed"
242 );
243
244 let metrics = SessionMetrics::new(Some(wait_time));
245 HttpsSession {
246 configured_backend_timeout,
247 configured_connect_timeout,
248 configured_frontend_timeout,
249 frontend_token: token,
250 has_been_closed: false,
251 last_event: Instant::now(),
252 listener,
253 metrics,
254 peer_address,
255 pool,
256 proxy,
257 public_address,
258 state,
259 }
260 }
261
262 pub fn upgrade(&mut self) -> SessionIsToBeClosed {
263 debug!("{} upgrade", log_context!(self));
264 #[cfg(debug_assertions)]
270 let from_marker = self.state.marker();
271 let new_state = match self.state.take() {
272 HttpsStateMachine::Expect(expect, ssl) => self.upgrade_expect(expect, ssl),
273 HttpsStateMachine::Handshake(handshake) => self.upgrade_handshake(handshake),
274 HttpsStateMachine::Mux(mux) => self.upgrade_mux(mux),
275 HttpsStateMachine::WebSocket(wss) => self.upgrade_websocket(wss),
276 HttpsStateMachine::FailedUpgrade(_) => {
277 error!(
281 "{} upgrade called on FailedUpgrade state; closing session",
282 log_context!(self)
283 );
284 None
285 }
286 };
287
288 match new_state {
289 Some(state) => {
290 #[cfg(debug_assertions)]
300 debug_assert!(
301 https_stage_rank(state.marker()) > https_stage_rank(from_marker)
302 || matches!(from_marker, StateMarker::WebSocket),
303 "HTTPS upgrade must advance the lifecycle (from {:?} to {:?})",
304 from_marker,
305 state.marker()
306 );
307 debug_assert!(
308 !state.failed(),
309 "a successful HTTPS upgrade must not yield a FailedUpgrade state"
310 );
311 self.state = state;
312 false
313 }
314 None => {
316 debug_assert!(
320 self.state.failed(),
321 "a refused HTTPS upgrade must leave the state in FailedUpgrade"
322 );
323 #[cfg(debug_assertions)]
325 debug_assert!(
326 https_stage_rank(self.state.marker()) == https_stage_rank(from_marker),
327 "FailedUpgrade must retain the origin stage marker for gauge restoration"
328 );
329 true
330 }
331 }
332 }
333
334 fn upgrade_expect(
335 &mut self,
336 mut expect: ExpectProxyProtocol<MioTcpStream>,
337 ssl: ServerConnection,
338 ) -> Option<HttpsStateMachine> {
339 if let Some(ref addresses) = expect.addresses
340 && let (Some(public_address), Some(session_address)) =
341 (addresses.destination(), addresses.source())
342 {
343 self.public_address = public_address;
344 self.peer_address = Some(session_address);
345
346 let ExpectProxyProtocol {
347 container_frontend_timeout,
348 frontend,
349 frontend_readiness: readiness,
350 request_id,
351 ..
352 } = expect;
353
354 let mut handshake = TlsHandshake::new(
355 container_frontend_timeout,
356 ssl,
357 frontend,
358 self.frontend_token,
359 request_id,
360 self.peer_address,
361 );
362 handshake.frontend_readiness = readiness;
365 handshake.frontend_readiness.event.insert(Ready::READABLE);
366
367 debug_assert_eq!(
371 self.peer_address,
372 Some(session_address),
373 "expect upgrade must adopt the PROXY-advertised source as the peer address"
374 );
375 debug_assert!(
376 handshake.frontend_readiness.event.is_readable(),
377 "handshake handed off from expect must be armed for READABLE"
378 );
379
380 gauge_add!(names::protocol::PROXY_EXPECT, -1);
381 gauge_add!(names::protocol::TLS_HANDSHAKE, 1);
382 return Some(HttpsStateMachine::Handshake(handshake));
383 }
384
385 if !expect.container_frontend_timeout.cancel() {
387 error!(
388 "{} failed to cancel request timeout on expect upgrade phase for 'expect proxy protocol with AF_UNSPEC address'",
389 log_context!(self)
390 );
391 }
392
393 None
394 }
395
396 fn upgrade_handshake(&mut self, handshake: TlsHandshake) -> Option<HttpsStateMachine> {
397 let sni_owned: Option<String> = handshake
409 .session
410 .server_name()
411 .map(|s| s.to_ascii_lowercase())
412 .map(|mut s| {
413 if s.ends_with('.') {
414 s.pop();
415 }
416 s
417 });
418 let alpn = handshake.session.alpn_protocol();
419 let alpn = alpn.and_then(|alpn| from_utf8(alpn).ok());
420 debug!(
421 "{} {}",
422 log_context!(self),
423 successful_tls_handshake_summary(sni_owned.as_deref(), alpn)
424 );
425
426 let disable_http11 = self.listener.borrow().is_http11_disabled();
431 let (alpn, alpn_label): (AlpnProtocol, Option<&'static str>) = match alpn {
436 Some("http/1.1") => {
437 if disable_http11 {
438 incr!(names::https::ALPN_REJECTED_HTTP11_DISABLED);
439 warn!(
440 "{} rejecting TLS connection: listener is H2-only but client negotiated http/1.1",
441 log_context!(self)
442 );
443 return None;
444 }
445 (AlpnProtocol::Http11, Some("http/1.1"))
446 }
447 Some("h2") => (AlpnProtocol::H2, Some("h2")),
448 Some(other) => {
449 incr!(names::https::ALPN_REJECTED_UNSUPPORTED);
456 error!(
457 "{} unsupported ALPN protocol: alpn_bytes={}",
458 log_context!(self),
459 other.len()
460 );
461 return None;
462 }
463 None => {
467 if disable_http11 {
468 incr!(names::https::ALPN_REJECTED_HTTP11_DISABLED);
469 warn!(
470 "{} rejecting TLS connection: listener is H2-only but client did not negotiate ALPN",
471 log_context!(self)
472 );
473 return None;
474 }
475 (AlpnProtocol::Http11, None)
476 }
477 };
478
479 debug_assert!(
485 !disable_http11 || matches!(alpn, AlpnProtocol::H2),
486 "H2-only listener must not dispatch an HTTP/1.1 session past ALPN"
487 );
488 debug_assert!(
489 match (&alpn, alpn_label) {
490 (AlpnProtocol::H2, Some(l)) => l == "h2",
491 (AlpnProtocol::Http11, Some(l)) => l == "http/1.1",
492 (AlpnProtocol::Http11, None) => true,
494 (AlpnProtocol::H2, None) => false,
495 },
496 "negotiated ALPN protocol and its wire label must agree"
497 );
498
499 let tls_version_label = handshake
504 .session
505 .protocol_version()
506 .and_then(rustls_version_label);
507 let tls_cipher_label = handshake
508 .session
509 .negotiated_cipher_suite()
510 .and_then(rustls_ciphersuite_label);
511 if let Some(version) = handshake.session.protocol_version() {
512 incr!(rustls_version_str(version));
513 };
514 if let Some(cipher) = handshake.session.negotiated_cipher_suite() {
515 incr!(rustls_ciphersuite_str(cipher));
516 };
517
518 gauge_add!(names::protocol::TLS_HANDSHAKE, -1);
519
520 let session_ulid = rusty_ulid::Ulid::generate();
521 let front_stream = FrontRustls {
522 stream: handshake.stream,
523 session: handshake.session,
524 peer_disconnected: false,
525 peer_reset: false,
526 session_ulid,
527 };
528 let router = mux::Router::new(
529 self.configured_backend_timeout,
530 self.configured_connect_timeout,
531 );
532 let mut context = mux::Context::new(
533 session_ulid,
534 self.pool.clone(),
535 self.listener.clone(),
536 self.peer_address,
537 self.public_address,
538 );
539 let tls_cert_names: Option<Arc<Vec<String>>> = match sni_owned.as_deref() {
594 Some(sni) => self
595 .listener
596 .borrow()
597 .resolver()
598 .names_for_sni(sni.as_bytes())
599 .and_then(|names| {
600 let mut snapshot: Vec<String> = names
601 .into_iter()
602 .map(|mut name| {
603 name.make_ascii_lowercase();
604 if name.ends_with('.') {
605 name.pop();
606 }
607 name
608 })
609 .collect();
610 snapshot.sort();
611 snapshot.dedup();
612 if snapshot.is_empty() {
613 None
614 } else {
615 Some(Arc::new(snapshot))
616 }
617 }),
618 None => None,
619 };
620 debug_assert!(
628 tls_cert_names.is_none() || sni_owned.is_some(),
629 "cert-name snapshot must not exist without an SNI to key it"
630 );
631 debug_assert!(
632 tls_cert_names
633 .as_ref()
634 .is_none_or(|names| { !names.is_empty() && names.windows(2).all(|w| w[0] < w[1]) }),
635 "cert-name snapshot must be non-empty and strictly sorted (sorted + deduped)"
636 );
637
638 context.tls_server_name = sni_owned;
642 context.tls_cert_names = tls_cert_names;
643 context.tls_version = tls_version_label;
647 context.tls_cipher = tls_cipher_label;
648 context.tls_alpn = alpn_label;
649 let mut frontend = match alpn {
650 AlpnProtocol::Http11 => {
651 incr!(names::http::ALPN_HTTP11);
652 context.create_stream(handshake.request_id, 1 << 16)?;
653 mux::Connection::new_h1_server(
654 session_ulid,
655 front_stream,
656 handshake.container_frontend_timeout,
657 )
658 }
659 AlpnProtocol::H2 => {
660 incr!(names::http::ALPN_H2);
661 let flood_config = self.listener.borrow().get_h2_flood_config();
662 let connection_config = self.listener.borrow().get_h2_connection_config();
663 let stream_idle_timeout = self.listener.borrow().get_h2_stream_idle_timeout();
664 let graceful_shutdown_deadline =
665 self.listener.borrow().get_h2_graceful_shutdown_deadline();
666 mux::Connection::new_h2_server(
667 session_ulid,
668 front_stream,
669 self.pool.clone(),
670 handshake.container_frontend_timeout,
671 flood_config,
672 connection_config,
673 stream_idle_timeout,
674 graceful_shutdown_deadline,
675 )?
676 }
677 };
678 frontend
687 .readiness_mut()
688 .event
689 .insert(Ready::READABLE | Ready::WRITABLE);
690
691 debug_assert!(
695 frontend.readiness_mut().event.is_readable()
696 && frontend.readiness_mut().event.is_writable(),
697 "post-handshake mux frontend must be armed for READABLE and WRITABLE"
698 );
699 debug_assert_eq!(
703 context.session_ulid, session_ulid,
704 "mux context and connection must share the handshake-derived session ulid"
705 );
706
707 gauge_add!(names::protocol::HTTPS, 1);
708 Some(HttpsStateMachine::Mux(Mux {
709 configured_frontend_timeout: self.configured_frontend_timeout,
710 frontend_token: self.frontend_token,
711 frontend,
712 context,
713 router,
714 session_ulid,
715 }))
716 }
717
718 fn upgrade_mux(&self, mut mux: MuxTls) -> Option<HttpsStateMachine> {
719 debug!("{} mux switching to wss", log_context!(self));
720 let Some(stream) = mux.context.streams.pop() else {
721 error!(
722 "{} upgrade_mux: no stream attached to the TLS mux session, closing",
723 log_context!(self)
724 );
725 return None;
726 };
727 let (frontend_readiness, frontend_socket, mut container_frontend_timeout) =
731 match mux.frontend {
732 mux::Connection::H1(mux::ConnectionH1 {
733 readiness,
734 socket,
735 timeout_container,
736 ..
737 }) => (readiness, socket, timeout_container),
738 mux::Connection::H2(_) => {
739 error!(
740 "{} only h1<->h1 connections can upgrade to websocket",
741 log_context!(self)
742 );
743 return None;
744 }
745 };
746
747 let mux::StreamState::Linked(back_token) = stream.state else {
748 error!(
749 "{} upgrading stream should be linked to a backend",
750 log_context!(self)
751 );
752 return None;
753 };
754 let Some(backend) = mux.router.backends.remove(&back_token) else {
755 error!(
756 "{} upgrade_mux: backend for token {:?} is missing (already disconnected?), closing",
757 log_context!(self),
758 back_token
759 );
760 return None;
761 };
762 let (cluster_id, backend, backend_readiness, backend_socket, mut container_backend_timeout) =
763 match backend {
764 mux::Connection::H1(mux::ConnectionH1 {
765 position:
766 mux::Position::Client(cluster_id, backend, mux::BackendStatus::Connected),
767 readiness,
768 socket,
769 timeout_container,
770 ..
771 }) => (cluster_id, backend, readiness, socket, timeout_container),
772 mux::Connection::H1(_) => {
773 error!(
774 "{} the backend disconnected just after upgrade, abort",
775 log_context!(self)
776 );
777 return None;
778 }
779 mux::Connection::H2(_) => {
780 error!(
781 "{} only h1<->h1 connections can upgrade to websocket",
782 log_context!(self)
783 );
784 return None;
785 }
786 };
787
788 let ws_context = stream.context.websocket_context();
789
790 container_frontend_timeout.reset();
791 container_backend_timeout.reset();
792
793 let backend_id = backend.borrow().backend_id.clone();
794 let backend_socket = backend_socket.stream;
797 let mut pipe = Pipe::new(
798 stream.back.storage.buffer,
799 Some(backend_id),
800 Some(backend_socket),
801 Some(backend),
802 Some(container_backend_timeout),
803 Some(container_frontend_timeout),
804 Some(cluster_id),
805 stream.front.storage.buffer,
806 self.frontend_token,
807 frontend_socket,
808 self.listener.clone(),
809 Protocol::HTTPS,
810 stream.context.session_id,
811 stream.context.id,
812 stream.context.session_address,
813 ws_context,
814 );
815
816 pipe.restore_readiness_events(frontend_readiness.event, backend_readiness.event);
817 pipe.set_back_token(back_token);
818 debug_assert!(
823 pipe.back_token().contains(&back_token),
824 "WSS pipe back token must be the connected backend token carried from the mux"
825 );
826 pipe.set_tls_metadata(
832 stream.context.tls_version,
833 stream.context.tls_cipher,
834 stream.context.tls_server_name.clone(),
835 stream.context.tls_alpn,
836 );
837
838 debug_assert_eq!(
844 pipe.frontend_readiness.event, frontend_readiness.event,
845 "WSS pipe must inherit the mux frontend readiness event verbatim"
846 );
847 debug_assert_eq!(
848 pipe.backend_readiness.event, backend_readiness.event,
849 "WSS pipe must inherit the mux backend readiness event verbatim"
850 );
851
852 gauge_add!(names::protocol::HTTPS, -1);
855 gauge_add!(names::protocol::WSS, 1);
856 gauge_add!(names::websocket::ACTIVE_REQUESTS, 1);
857 Some(HttpsStateMachine::WebSocket(pipe))
858 }
859
860 fn upgrade_websocket(
861 &self,
862 wss: Pipe<FrontRustls, HttpsListener>,
863 ) -> Option<HttpsStateMachine> {
864 error!(
866 "{} upgrade called on WSS, this should not happen",
867 log_context!(self)
868 );
869 Some(HttpsStateMachine::WebSocket(wss))
870 }
871
872 #[cfg(debug_assertions)]
878 fn check_invariants(&self) {
879 debug_assert!(
883 !self.configured_frontend_timeout.is_zero()
884 && !self.configured_backend_timeout.is_zero()
885 && !self.configured_connect_timeout.is_zero(),
886 "HTTPS session timeouts must stay non-zero for the session lifetime"
887 );
888 if self.state.failed() {
893 debug_assert!(
894 matches!(
895 self.state.marker(),
896 StateMarker::Expect
897 | StateMarker::Handshake
898 | StateMarker::Mux
899 | StateMarker::WebSocket
900 ),
901 "FailedUpgrade must retain a valid origin-stage marker"
902 );
903 } else {
904 debug_assert!(
905 !self.state.failed(),
906 "a non-failed session must not also report FailedUpgrade"
907 );
908 }
909 debug_assert!(
915 !matches!(self.state.marker(), StateMarker::Expect)
916 || self.peer_address.is_none()
917 || self.state.failed(),
918 "a live Expect-stage session has no resolved peer address yet"
919 );
920 }
921}
922
923impl ProxySession for HttpsSession {
924 fn close(&mut self) {
925 if self.has_been_closed {
926 return;
927 }
928 debug_assert!(
933 !self.has_been_closed,
934 "close() body must run only on a not-yet-closed session"
935 );
936
937 trace!("{} closing HTTPS session", log_context!(self));
938 self.metrics.service_stop();
939
940 match self.state.marker() {
942 StateMarker::Expect => gauge_add!(names::protocol::PROXY_EXPECT, -1),
943 StateMarker::Handshake => gauge_add!(names::protocol::TLS_HANDSHAKE, -1),
944 StateMarker::Mux => gauge_add!(names::protocol::HTTPS, -1),
945 StateMarker::WebSocket => {
946 gauge_add!(names::protocol::WSS, -1);
947 gauge_add!(names::websocket::ACTIVE_REQUESTS, -1);
948 }
949 }
950
951 if self.state.failed() {
952 match self.state.marker() {
953 StateMarker::Expect => incr!(names::https::UPGRADE_EXPECT_FAILED),
954 StateMarker::Handshake => incr!(names::https::UPGRADE_HANDSHAKE_FAILED),
955 StateMarker::Mux => incr!(names::https::UPGRADE_MUX_FAILED),
956 StateMarker::WebSocket => incr!(names::https::UPGRADE_WSS_FAILED),
957 }
958 self.state.close(self.proxy.clone(), &mut self.metrics);
962 self.proxy.borrow().remove_session(self.frontend_token);
963 self.has_been_closed = true;
964 return;
965 }
966
967 self.state.cancel_timeouts();
968 self.state.close(self.proxy.clone(), &mut self.metrics);
971
972 let front_socket = self.state.front_socket();
979 if let Err(e) = front_socket.shutdown(Shutdown::Write) {
980 if e.kind() != ErrorKind::NotConnected {
982 error!(
983 "{} error shutting down front socket({:?}): {:?}",
984 log_context!(self),
985 front_socket,
986 e
987 );
988 }
989 }
990
991 let proxy = self.proxy.borrow();
993 let fd = front_socket.as_raw_fd();
994 if let Err(e) = proxy.registry.deregister(&mut SourceFd(&fd)) {
995 error!(
996 "{} error deregistering front socket({:?}) while closing HTTPS session: {:?}",
997 log_context!(self),
998 fd,
999 e
1000 );
1001 }
1002 proxy.remove_session(self.frontend_token);
1003
1004 self.has_been_closed = true;
1005 debug_assert!(
1009 self.has_been_closed,
1010 "close() must leave the session marked closed"
1011 );
1012 }
1013
1014 fn timeout(&mut self, token: Token) -> SessionIsToBeClosed {
1015 let session_result = self.state.timeout(token, &mut self.metrics);
1016 if session_result == StateResult::CloseSession {
1017 debug!(
1018 "{} HTTPS timeout requested close: token={:?}, marker={:?}",
1019 log_context!(self),
1020 token,
1021 self.state.marker()
1022 );
1023 }
1024 session_result == StateResult::CloseSession
1025 }
1026
1027 fn protocol(&self) -> Protocol {
1028 Protocol::HTTPS
1029 }
1030
1031 fn update_readiness(&mut self, token: Token, events: Ready) {
1032 trace!(
1033 "{} token {:?} got event {}",
1034 log_context!(self),
1035 token,
1036 super::ready_to_string(events)
1037 );
1038 self.last_event = Instant::now();
1039 self.metrics.wait_start();
1040 self.state.update_readiness(token, events);
1041 }
1042
1043 fn ready(&mut self, session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
1044 self.metrics.service_start();
1045
1046 let session_result =
1047 self.state
1048 .ready(session.clone(), self.proxy.clone(), &mut self.metrics);
1049
1050 let to_be_closed = match session_result {
1051 SessionResult::Close => true,
1052 SessionResult::Continue => false,
1053 SessionResult::Upgrade => match self.upgrade() {
1054 false => self.ready(session),
1055 true => true,
1056 },
1057 };
1058 if to_be_closed {
1059 debug!(
1060 "{} HTTPS ready requested close: marker={:?}",
1061 log_context!(self),
1062 self.state.marker()
1063 );
1064 }
1065
1066 #[cfg(debug_assertions)]
1070 self.check_invariants();
1071
1072 self.metrics.service_stop();
1073 to_be_closed
1074 }
1075
1076 fn shutting_down(&mut self) -> SessionIsToBeClosed {
1077 self.state.shutting_down()
1078 }
1079
1080 fn last_event(&self) -> Instant {
1081 self.last_event
1082 }
1083
1084 fn print_session(&self) {
1085 self.state.print_state("HTTPS");
1086 error!("{} Metrics: {:?}", log_context!(self), self.metrics);
1087 }
1088
1089 fn frontend_token(&self) -> Token {
1090 self.frontend_token
1091 }
1092}
1093
1094pub type HostName = String;
1095pub type PathBegin = String;
1096
1097pub struct HttpsListener {
1098 active: bool,
1099 address: StdSocketAddr,
1100 answers: Rc<RefCell<HttpAnswers>>,
1101 config: HttpsListenerConfig,
1102 fronts: Router,
1103 listener: Option<MioTcpListener>,
1104 resolver: Arc<MutexCertificateResolver>,
1105 rustls_details: Arc<RustlsServerConfig>,
1106 tags: BTreeMap<String, CachedTags>,
1107 token: Token,
1108}
1109
1110impl ListenerHandler for HttpsListener {
1111 fn get_addr(&self) -> &StdSocketAddr {
1112 &self.address
1113 }
1114
1115 fn get_tags(&self, key: &str) -> Option<&CachedTags> {
1116 self.tags.get(key)
1117 }
1118
1119 fn set_tags(&mut self, key: String, tags: Option<BTreeMap<String, String>>) {
1120 match tags {
1121 Some(tags) => self.tags.insert(key, CachedTags::new(tags)),
1122 None => self.tags.remove(&key),
1123 };
1124 }
1125
1126 fn protocol(&self) -> Protocol {
1127 Protocol::HTTPS
1128 }
1129
1130 fn public_address(&self) -> StdSocketAddr {
1131 self.config
1132 .public_address
1133 .map(|addr| addr.into())
1134 .unwrap_or(self.address)
1135 }
1136}
1137
1138impl L7ListenerHandler for HttpsListener {
1139 fn get_sticky_name(&self) -> &str {
1140 &self.config.sticky_name
1141 }
1142
1143 fn get_sozu_id_header(&self) -> &str {
1144 self.config
1145 .sozu_id_header
1146 .as_deref()
1147 .filter(|s| !s.is_empty())
1148 .unwrap_or("Sozu-Id")
1149 }
1150
1151 fn get_connect_timeout(&self) -> u32 {
1152 self.config.connect_timeout
1153 }
1154
1155 fn frontend_from_request(
1156 &self,
1157 host: &str,
1158 uri: &str,
1159 method: &Method,
1160 ) -> Result<RouteResult, FrontendFromRequestError> {
1161 let start = Instant::now();
1162 let (remaining_input, (hostname, _)) = match hostname_and_port(host.as_bytes()) {
1163 Ok(tuple) => tuple,
1164 Err(parse_error) => {
1165 return Err(FrontendFromRequestError::HostParse {
1167 host: host.to_owned(),
1168 error: parse_error.to_string(),
1169 });
1170 }
1171 };
1172
1173 if remaining_input != &b""[..] {
1174 return Err(FrontendFromRequestError::InvalidCharsAfterHost(
1175 host.to_owned(),
1176 ));
1177 }
1178
1179 let host = unsafe { from_utf8_unchecked(hostname) };
1188
1189 let route = self.fronts.lookup(host, uri, method).map_err(|e| {
1190 incr!(names::http::FAILED_BACKEND_MATCHING);
1191 FrontendFromRequestError::NoClusterFound(e)
1192 })?;
1193
1194 let now = Instant::now();
1195
1196 if let Some(cluster) = route.cluster_id.as_deref() {
1197 time!(
1198 names::event_loop::FRONTEND_MATCHING_TIME,
1199 cluster,
1200 (now - start).as_millis()
1201 );
1202 }
1203
1204 Ok(route)
1205 }
1206
1207 fn get_answers(&self) -> &Rc<RefCell<HttpAnswers>> {
1208 &self.answers
1209 }
1210
1211 fn get_h2_flood_config(&self) -> crate::protocol::mux::H2FloodConfig {
1212 let defaults = crate::protocol::mux::H2FloodConfig::default();
1213 crate::protocol::mux::H2FloodConfig {
1214 max_rst_stream_per_window: self
1215 .config
1216 .h2_max_rst_stream_per_window
1217 .unwrap_or(defaults.max_rst_stream_per_window),
1218 max_ping_per_window: self
1219 .config
1220 .h2_max_ping_per_window
1221 .unwrap_or(defaults.max_ping_per_window),
1222 max_settings_per_window: self
1223 .config
1224 .h2_max_settings_per_window
1225 .unwrap_or(defaults.max_settings_per_window),
1226 max_empty_data_per_window: self
1227 .config
1228 .h2_max_empty_data_per_window
1229 .unwrap_or(defaults.max_empty_data_per_window),
1230 max_window_update_stream0_per_window: self
1231 .config
1232 .h2_max_window_update_stream0_per_window
1233 .unwrap_or(defaults.max_window_update_stream0_per_window),
1234 max_continuation_frames: self
1235 .config
1236 .h2_max_continuation_frames
1237 .unwrap_or(defaults.max_continuation_frames),
1238 max_glitch_count: self
1239 .config
1240 .h2_max_glitch_count
1241 .unwrap_or(defaults.max_glitch_count),
1242 max_rst_stream_lifetime: self
1243 .config
1244 .h2_max_rst_stream_lifetime
1245 .unwrap_or(defaults.max_rst_stream_lifetime),
1246 max_rst_stream_abusive_lifetime: self
1247 .config
1248 .h2_max_rst_stream_abusive_lifetime
1249 .unwrap_or(defaults.max_rst_stream_abusive_lifetime),
1250 max_rst_stream_emitted_lifetime: self
1251 .config
1252 .h2_max_rst_stream_emitted_lifetime
1253 .unwrap_or(defaults.max_rst_stream_emitted_lifetime),
1254 max_header_list_size: self
1255 .config
1256 .h2_max_header_list_size
1257 .unwrap_or(defaults.max_header_list_size),
1258 max_header_table_size: self
1259 .config
1260 .h2_max_header_table_size
1261 .unwrap_or(defaults.max_header_table_size),
1262 max_header_fields: self
1263 .config
1264 .h2_max_header_fields
1265 .unwrap_or(defaults.max_header_fields),
1266 }
1267 }
1268
1269 fn get_h2_connection_config(&self) -> crate::protocol::mux::H2ConnectionConfig {
1270 crate::protocol::mux::H2ConnectionConfig::from_optional(
1271 self.config.h2_initial_connection_window,
1272 self.config.h2_max_concurrent_streams,
1273 self.config.h2_stream_shrink_ratio,
1274 )
1275 }
1276
1277 fn get_strict_sni_binding(&self) -> bool {
1278 self.config.strict_sni_binding.unwrap_or(true)
1293 }
1294
1295 fn get_elide_x_real_ip(&self) -> bool {
1296 self.config.elide_x_real_ip.unwrap_or(false)
1297 }
1298
1299 fn get_send_x_real_ip(&self) -> bool {
1300 self.config.send_x_real_ip.unwrap_or(false)
1301 }
1302
1303 fn get_h2_stream_idle_timeout(&self) -> std::time::Duration {
1304 let seconds = self
1311 .config
1312 .h2_stream_idle_timeout_seconds
1313 .map(|s| u64::from(s.max(1)))
1314 .unwrap_or_else(|| u64::from(self.config.back_timeout).max(30));
1315 std::time::Duration::from_secs(seconds)
1316 }
1317
1318 fn get_h2_graceful_shutdown_deadline(&self) -> Option<std::time::Duration> {
1319 match self.config.h2_graceful_shutdown_deadline_seconds {
1320 None => Some(std::time::Duration::from_secs(5)),
1321 Some(0) => None,
1322 Some(s) => Some(std::time::Duration::from_secs(u64::from(s))),
1323 }
1324 }
1325}
1326
1327impl HttpsListener {
1328 pub fn is_http11_disabled(&self) -> bool {
1334 self.config.disable_http11.unwrap_or(false)
1335 }
1336
1337 pub fn resolver(&self) -> &Arc<MutexCertificateResolver> {
1343 &self.resolver
1344 }
1345
1346 pub fn try_new(
1347 config: HttpsListenerConfig,
1348 token: Token,
1349 ) -> Result<HttpsListener, ListenerError> {
1350 let resolver = Arc::new(MutexCertificateResolver::default());
1351
1352 let server_config = Arc::new(Self::create_rustls_context(&config, resolver.to_owned())?);
1353
1354 let answers = Self::build_answers(&config)?;
1355
1356 Ok(HttpsListener {
1357 listener: None,
1358 address: config.address.into(),
1359 resolver,
1360 rustls_details: server_config,
1361 active: false,
1362 fronts: Router::new(),
1363 answers: Rc::new(RefCell::new(answers)),
1364 config,
1365 token,
1366 tags: BTreeMap::new(),
1367 })
1368 }
1369
1370 fn build_answers(config: &HttpsListenerConfig) -> Result<HttpAnswers, ListenerError> {
1377 let mut answers_map = config.answers.clone();
1378 if let Some(ref legacy) = config.http_answers {
1379 crate::protocol::http::answers::merge_legacy_into_map(&mut answers_map, legacy);
1380 }
1381 HttpAnswers::new(&answers_map)
1382 .map_err(|(name, error)| ListenerError::TemplateParse(name, error))
1383 }
1384
1385 pub fn validate_config(config: &HttpsListenerConfig) -> Result<(), ListenerError> {
1396 Self::create_rustls_context(config, Arc::new(MutexCertificateResolver::default()))?;
1401 Self::build_answers(config)?;
1402 Ok(())
1403 }
1404
1405 pub fn activate(
1406 &mut self,
1407 registry: &Registry,
1408 tcp_listener: Option<MioTcpListener>,
1409 ) -> Result<Token, ListenerError> {
1410 if self.active {
1411 return Ok(self.token);
1412 }
1413 let address: StdSocketAddr = self.config.address.into();
1414
1415 let mut listener = match tcp_listener {
1416 Some(tcp_listener) => tcp_listener,
1417 None => {
1418 server_bind(address).map_err(|server_bind_error| ListenerError::Activation {
1419 address,
1420 error: server_bind_error.to_string(),
1421 })?
1422 }
1423 };
1424
1425 registry
1426 .register(&mut listener, self.token, Interest::READABLE)
1427 .map_err(ListenerError::SocketRegistration)?;
1428
1429 self.listener = Some(listener);
1430 self.active = true;
1431 debug_assert!(
1437 self.active && self.listener.is_some(),
1438 "an activated HTTPS listener must hold a bound socket and be flagged active"
1439 );
1440 Ok(self.token)
1441 }
1442
1443 pub fn create_rustls_context(
1444 config: &HttpsListenerConfig,
1445 resolver: Arc<MutexCertificateResolver>,
1446 ) -> Result<RustlsServerConfig, ListenerError> {
1447 let cipher_names = if config.cipher_list.is_empty() {
1448 DEFAULT_CIPHER_LIST.to_vec()
1449 } else {
1450 config
1451 .cipher_list
1452 .iter()
1453 .map(|s| s.as_str())
1454 .collect::<Vec<_>>()
1455 };
1456
1457 let ciphers = cipher_names
1458 .into_iter()
1459 .filter_map(|cipher| {
1460 cipher_suite_by_name(cipher).or_else(|| {
1461 error!(
1462 "{} unknown or unsupported cipher: {:?}",
1463 log_module_context!(),
1464 cipher
1465 );
1466 None
1467 })
1468 })
1469 .collect::<Vec<_>>();
1470
1471 let versions = config
1472 .versions
1473 .iter()
1474 .filter_map(|version| match TlsVersion::try_from(*version) {
1475 Ok(TlsVersion::TlsV12) => Some(&rustls::version::TLS12),
1476 Ok(TlsVersion::TlsV13) => Some(&rustls::version::TLS13),
1477 Ok(other_version) => {
1478 error!(
1479 "{} unsupported TLS version {:?}",
1480 log_module_context!(),
1481 other_version
1482 );
1483 None
1484 }
1485 Err(_) => {
1486 error!("{} unsupported TLS version", log_module_context!());
1487 None
1488 }
1489 })
1490 .collect::<Vec<_>>();
1491
1492 let kx_groups = if config.groups_list.is_empty() {
1493 default_provider().kx_groups
1494 } else {
1495 config
1496 .groups_list
1497 .iter()
1498 .filter_map(|group| match kx_group_by_name(group) {
1499 Some(kx) => Some(kx),
1500 None => {
1501 debug!("key exchange group {:?} not supported by the compiled crypto provider, skipping", group);
1502 None
1503 }
1504 })
1505 .collect::<Vec<_>>()
1506 };
1507
1508 let provider = CryptoProvider {
1509 cipher_suites: ciphers,
1510 kx_groups,
1511 ..default_provider()
1512 };
1513
1514 let mut server_config = RustlsServerConfig::builder_with_provider(provider.into())
1515 .with_protocol_versions(&versions[..])
1516 .map_err(|err| ListenerError::BuildRustls(err.to_string()))?
1517 .with_no_client_auth()
1518 .with_cert_resolver(resolver);
1519 server_config.send_tls13_tickets = config.send_tls13_tickets as usize;
1520
1521 server_config.alpn_protocols = if config.alpn_protocols.is_empty() {
1522 DEFAULT_ALPN_PROTOCOLS
1523 .iter()
1524 .map(|p| p.as_bytes().to_vec())
1525 .collect()
1526 } else {
1527 config
1528 .alpn_protocols
1529 .iter()
1530 .map(|p| p.as_bytes().to_vec())
1531 .collect()
1532 };
1533
1534 Ok(server_config)
1535 }
1536
1537 pub fn update_config(
1545 &mut self,
1546 patch: &UpdateHttpsListenerConfig,
1547 ) -> Result<(), ListenerError> {
1548 validate_h2_flood_knobs_https(patch)?;
1553 if let Some(ref alpn) = patch.alpn_protocols {
1554 validate_alpn_protocols(&alpn.values)?;
1555 }
1556 if let Some(ref hdr) = patch.sozu_id_header {
1557 validate_sozu_id_header(hdr)?;
1558 }
1559
1560 if let Some(v) = patch.public_address {
1562 self.config.public_address = Some(v);
1563 }
1564 if let Some(v) = patch.expect_proxy {
1565 self.config.expect_proxy = v;
1566 }
1567 if let Some(ref v) = patch.sticky_name {
1568 self.config.sticky_name = v.to_owned();
1569 }
1570 if let Some(v) = patch.front_timeout {
1571 self.config.front_timeout = v;
1572 }
1573 if let Some(v) = patch.back_timeout {
1574 self.config.back_timeout = v;
1575 }
1576 if let Some(v) = patch.connect_timeout {
1577 self.config.connect_timeout = v;
1578 }
1579 if let Some(v) = patch.request_timeout {
1580 self.config.request_timeout = v;
1581 }
1582 if let Some(v) = patch.strict_sni_binding {
1583 self.config.strict_sni_binding = Some(v);
1584 }
1585 if let Some(v) = patch.disable_http11 {
1586 self.config.disable_http11 = Some(v);
1587 }
1588 if let Some(ref v) = patch.sozu_id_header {
1589 self.config.sozu_id_header = Some(v.to_owned());
1590 }
1591 if let Some(v) = patch.elide_x_real_ip {
1592 self.config.elide_x_real_ip = Some(v);
1593 }
1594 if let Some(v) = patch.send_x_real_ip {
1595 self.config.send_x_real_ip = Some(v);
1596 }
1597
1598 if let Some(v) = patch.h2_max_rst_stream_per_window {
1600 self.config.h2_max_rst_stream_per_window = Some(v);
1601 }
1602 if let Some(v) = patch.h2_max_ping_per_window {
1603 self.config.h2_max_ping_per_window = Some(v);
1604 }
1605 if let Some(v) = patch.h2_max_settings_per_window {
1606 self.config.h2_max_settings_per_window = Some(v);
1607 }
1608 if let Some(v) = patch.h2_max_empty_data_per_window {
1609 self.config.h2_max_empty_data_per_window = Some(v);
1610 }
1611 if let Some(v) = patch.h2_max_continuation_frames {
1612 self.config.h2_max_continuation_frames = Some(v);
1613 }
1614 if let Some(v) = patch.h2_max_glitch_count {
1615 self.config.h2_max_glitch_count = Some(v);
1616 }
1617 if let Some(v) = patch.h2_initial_connection_window {
1618 self.config.h2_initial_connection_window = Some(v);
1619 }
1620 if let Some(v) = patch.h2_max_concurrent_streams {
1621 self.config.h2_max_concurrent_streams = Some(v);
1622 }
1623 if let Some(v) = patch.h2_stream_shrink_ratio {
1624 self.config.h2_stream_shrink_ratio = Some(v);
1625 }
1626 if let Some(v) = patch.h2_max_rst_stream_lifetime {
1627 self.config.h2_max_rst_stream_lifetime = Some(v);
1628 }
1629 if let Some(v) = patch.h2_max_rst_stream_abusive_lifetime {
1630 self.config.h2_max_rst_stream_abusive_lifetime = Some(v);
1631 }
1632 if let Some(v) = patch.h2_max_rst_stream_emitted_lifetime {
1633 self.config.h2_max_rst_stream_emitted_lifetime = Some(v);
1634 }
1635 if let Some(v) = patch.h2_max_header_list_size {
1636 self.config.h2_max_header_list_size = Some(v);
1637 }
1638 if let Some(v) = patch.h2_max_header_table_size {
1639 self.config.h2_max_header_table_size = Some(v);
1640 }
1641 if let Some(v) = patch.h2_max_header_fields {
1642 self.config.h2_max_header_fields = Some(v);
1643 }
1644 if let Some(v) = patch.h2_stream_idle_timeout_seconds {
1645 self.config.h2_stream_idle_timeout_seconds = Some(v);
1646 }
1647 if let Some(v) = patch.h2_graceful_shutdown_deadline_seconds {
1648 self.config.h2_graceful_shutdown_deadline_seconds = Some(v);
1649 }
1650 if let Some(v) = patch.h2_max_window_update_stream0_per_window {
1651 self.config.h2_max_window_update_stream0_per_window = Some(v);
1652 }
1653
1654 if let Some(ref alpn_wrapper) = patch.alpn_protocols {
1664 let mut candidate = self.config.clone();
1665 candidate.alpn_protocols = alpn_wrapper.values.clone();
1666 let new_rustls = Arc::new(Self::create_rustls_context(
1667 &candidate,
1668 self.resolver.clone(),
1669 )?);
1670 self.config.alpn_protocols = alpn_wrapper.values.clone();
1672 self.rustls_details = new_rustls;
1673 debug_assert_eq!(
1678 self.config.alpn_protocols, alpn_wrapper.values,
1679 "committed ALPN config must match the patch values exactly"
1680 );
1681 }
1682
1683 let answers_changed = patch.http_answers.is_some() || !patch.answers.is_empty();
1688 if answers_changed {
1689 if let Some(ref new_answers) = patch.http_answers {
1690 crate::sozu_command::state::merge_custom_http_answers(
1691 &mut self.config.http_answers,
1692 new_answers,
1693 );
1694 }
1695 for (code, body) in &patch.answers {
1696 if !body.is_empty() {
1697 self.config.answers.insert(code.clone(), body.clone());
1698 }
1699 }
1700
1701 let mut answers_map = self.config.answers.clone();
1702 if let Some(ref legacy) = self.config.http_answers {
1703 crate::protocol::http::answers::merge_legacy_into_map(&mut answers_map, legacy);
1704 }
1705 let mut rebuilt = HttpAnswers::new(&answers_map)
1706 .map_err(|(name, error)| ListenerError::TemplateParse(name, error))?;
1707 let preserved = std::mem::take(&mut self.answers.borrow_mut().cluster_answers);
1708 rebuilt.cluster_answers = preserved;
1709 *self.answers.borrow_mut() = rebuilt;
1710 }
1711
1712 if let Some(new_hsts) = patch.hsts {
1731 if new_hsts.enabled.is_none() {
1732 return Err(ListenerError::HstsEnabledRequired);
1733 }
1734 self.config.hsts = Some(new_hsts);
1735 let refreshed = self
1736 .fronts
1737 .refresh_inheriting_hsts(self.config.hsts.as_ref());
1738 for _ in 0..refreshed {
1739 crate::incr!(names::http::HSTS_FRONTEND_REFRESHED);
1740 }
1741 info!(
1742 "{} HTTPS listener {:?} HSTS default patched; refreshed {} inheriting \
1743 frontend(s). Explicit per-frontend overrides untouched.",
1744 log_module_context!(),
1745 self.config.address,
1746 refreshed,
1747 );
1748 crate::incr!(names::http::HSTS_LISTENER_DEFAULT_PATCHED);
1749 }
1750
1751 Ok(())
1752 }
1753
1754 pub fn add_https_front(&mut self, tls_front: HttpFrontend) -> Result<(), ListenerError> {
1755 self.add_https_front_with_hsts_origin(tls_front, crate::router::HstsOrigin::Explicit)
1756 }
1757
1758 pub fn add_https_front_with_hsts_origin(
1765 &mut self,
1766 tls_front: HttpFrontend,
1767 hsts_origin: crate::router::HstsOrigin,
1768 ) -> Result<(), ListenerError> {
1769 self.fronts
1770 .add_http_front_with_hsts_origin(&tls_front, hsts_origin)
1771 .map_err(ListenerError::AddFrontend)
1772 }
1773
1774 pub fn remove_https_front(&mut self, tls_front: HttpFrontend) -> Result<(), ListenerError> {
1775 debug!(
1776 "{} removing tls_front {:?}",
1777 log_module_context!(),
1778 tls_front
1779 );
1780 self.fronts
1781 .remove_http_front(&tls_front)
1782 .map_err(ListenerError::RemoveFrontend)
1783 }
1784
1785 fn accept(&mut self) -> Result<MioTcpStream, AcceptError> {
1786 if let Some(ref sock) = self.listener {
1787 sock.accept()
1788 .map_err(|e| match e.kind() {
1789 ErrorKind::WouldBlock => AcceptError::WouldBlock,
1790 _ => {
1791 error!("{} accept() IO error: {:?}", log_module_context!(), e);
1792 AcceptError::IoError
1793 }
1794 })
1795 .map(|(sock, _)| sock)
1796 } else {
1797 error!(
1798 "{} cannot accept connections, no listening socket available",
1799 log_module_context!()
1800 );
1801 Err(AcceptError::IoError)
1802 }
1803 }
1804}
1805
1806pub struct HttpsProxy {
1807 listeners: HashMap<Token, Rc<RefCell<HttpsListener>>>,
1808 clusters: HashMap<ClusterId, Cluster>,
1809 backends: Rc<RefCell<BackendMap>>,
1810 pool: Rc<RefCell<Pool>>,
1811 registry: Registry,
1812 sessions: Rc<RefCell<SessionManager>>,
1813}
1814
1815impl HttpsProxy {
1816 pub fn new(
1817 registry: Registry,
1818 sessions: Rc<RefCell<SessionManager>>,
1819 pool: Rc<RefCell<Pool>>,
1820 backends: Rc<RefCell<BackendMap>>,
1821 ) -> HttpsProxy {
1822 HttpsProxy {
1823 listeners: HashMap::new(),
1824 clusters: HashMap::new(),
1825 backends,
1826 pool,
1827 registry,
1828 sessions,
1829 }
1830 }
1831
1832 pub fn add_listener(
1833 &mut self,
1834 config: HttpsListenerConfig,
1835 token: Token,
1836 ) -> Result<Token, ProxyError> {
1837 match self.listeners.entry(token) {
1838 Entry::Vacant(entry) => {
1839 let https_listener =
1840 HttpsListener::try_new(config, token).map_err(ProxyError::AddListener)?;
1841 entry.insert(Rc::new(RefCell::new(https_listener)));
1842 Ok(token)
1843 }
1844 _ => Err(ProxyError::ListenerAlreadyPresent),
1845 }
1846 }
1847
1848 pub fn remove_listener(
1849 &mut self,
1850 remove: RemoveListener,
1851 ) -> Result<Option<ResponseContent>, ProxyError> {
1852 let len = self.listeners.len();
1853
1854 let remove_address = remove.address.into();
1855 self.listeners
1856 .retain(|_, listener| listener.borrow().address != remove_address);
1857
1858 if !self.listeners.len() < len {
1859 info!(
1860 "{} no HTTPS listener to remove at address {}",
1861 log_module_context!(),
1862 remove_address
1863 )
1864 }
1865 Ok(None)
1866 }
1867
1868 pub fn soft_stop(&mut self) -> Result<(), ProxyError> {
1869 let listeners: HashMap<_, _> = self.listeners.drain().collect();
1870 let mut socket_errors = vec![];
1871 for l in listeners.values() {
1872 if let Some(mut sock) = l.borrow_mut().listener.take() {
1873 debug!("{} deregistering socket {:?}", log_module_context!(), sock);
1874 if let Err(e) = self.registry.deregister(&mut sock) {
1875 let error = format!("socket {sock:?}: {e:?}");
1876 socket_errors.push(error);
1877 }
1878 }
1879 }
1880
1881 if !socket_errors.is_empty() {
1882 return Err(ProxyError::SoftStop {
1883 proxy_protocol: "HTTPS".to_string(),
1884 error: format!("Error deregistering listen sockets: {socket_errors:?}"),
1885 });
1886 }
1887
1888 Ok(())
1889 }
1890
1891 pub fn hard_stop(&mut self) -> Result<(), ProxyError> {
1892 let mut listeners: HashMap<_, _> = self.listeners.drain().collect();
1893 let mut socket_errors = vec![];
1894 for (_, l) in listeners.drain() {
1895 if let Some(mut sock) = l.borrow_mut().listener.take() {
1896 debug!("{} deregistering socket {:?}", log_module_context!(), sock);
1897 if let Err(e) = self.registry.deregister(&mut sock) {
1898 let error = format!("socket {sock:?}: {e:?}");
1899 socket_errors.push(error);
1900 }
1901 }
1902 }
1903
1904 if !socket_errors.is_empty() {
1905 return Err(ProxyError::HardStop {
1906 proxy_protocol: "HTTPS".to_string(),
1907 error: format!("Error deregistering listen sockets: {socket_errors:?}"),
1908 });
1909 }
1910
1911 Ok(())
1912 }
1913
1914 pub fn query_all_certificates(&mut self) -> Result<Option<ResponseContent>, ProxyError> {
1915 let certificates: Vec<CertificatesByAddress> = self
1916 .listeners
1917 .values()
1918 .map(|listener| {
1919 let owned = listener.borrow();
1920 let resolver = unwrap_msg!(owned.resolver.0.lock());
1921 let certificate_summaries = resolver
1922 .domains
1923 .to_hashmap()
1924 .drain()
1925 .map(|(k, fingerprint)| CertificateSummary {
1926 domain: String::from_utf8(k).unwrap(),
1927 fingerprint: fingerprint.to_string(),
1928 })
1929 .collect();
1930
1931 CertificatesByAddress {
1932 address: owned.address.into(),
1933 certificate_summaries,
1934 }
1935 })
1936 .collect();
1937
1938 let listeners_count = certificates.len();
1939 let certificates_count = certificates
1940 .iter()
1941 .map(|entry| entry.certificate_summaries.len())
1942 .fold(0usize, |total, count| total.saturating_add(count));
1943
1944 info!(
1945 "{} got Certificates::All query, listeners_count={} certificates_count={}",
1946 log_module_context!(),
1947 listeners_count,
1948 certificates_count,
1949 );
1950
1951 Ok(Some(
1952 ContentType::CertificatesByAddress(ListOfCertificatesByAddress { certificates }).into(),
1953 ))
1954 }
1955
1956 pub fn query_certificate_for_domain(
1957 &mut self,
1958 domain: String,
1959 ) -> Result<Option<ResponseContent>, ProxyError> {
1960 let certificates: Vec<CertificatesByAddress> = self
1961 .listeners
1962 .values()
1963 .map(|listener| {
1964 let owned = listener.borrow();
1965 let resolver = unwrap_msg!(owned.resolver.0.lock());
1966 let mut certificate_summaries = vec![];
1967
1968 if let Some((k, fingerprint)) = resolver.domain_lookup(domain.as_bytes(), true) {
1969 certificate_summaries.push(CertificateSummary {
1970 domain: String::from_utf8(k.to_vec()).unwrap(),
1971 fingerprint: fingerprint.to_string(),
1972 });
1973 }
1974 CertificatesByAddress {
1975 address: owned.address.into(),
1976 certificate_summaries,
1977 }
1978 })
1979 .collect();
1980
1981 let listeners_count = certificates.len();
1982 let certificates_count = certificates
1983 .iter()
1984 .map(|entry| entry.certificate_summaries.len())
1985 .fold(0usize, |total, count| total.saturating_add(count));
1986
1987 info!(
1988 "{} got Certificates::Domain query, domain_bytes={} listeners_count={} certificates_count={}",
1989 log_module_context!(),
1990 domain.len(),
1991 listeners_count,
1992 certificates_count,
1993 );
1994
1995 Ok(Some(
1996 ContentType::CertificatesByAddress(ListOfCertificatesByAddress { certificates }).into(),
1997 ))
1998 }
1999
2000 pub fn activate_listener(
2001 &mut self,
2002 addr: &StdSocketAddr,
2003 tcp_listener: Option<MioTcpListener>,
2004 ) -> Result<Token, ProxyError> {
2005 let listener = self
2006 .listeners
2007 .values()
2008 .find(|listener| listener.borrow().address == *addr)
2009 .ok_or(ProxyError::NoListenerFound(addr.to_owned()))?;
2010
2011 listener
2012 .borrow_mut()
2013 .activate(&self.registry, tcp_listener)
2014 .map_err(|listener_error| ProxyError::ListenerActivation {
2015 address: *addr,
2016 listener_error,
2017 })
2018 }
2019
2020 pub fn give_back_listeners(&mut self) -> Vec<(StdSocketAddr, MioTcpListener)> {
2021 self.listeners
2022 .values()
2023 .filter_map(|listener| {
2024 let mut owned = listener.borrow_mut();
2025 if let Some(listener) = owned.listener.take() {
2026 owned.active = false;
2029 return Some((owned.address, listener));
2030 }
2031
2032 None
2033 })
2034 .collect()
2035 }
2036
2037 pub fn give_back_listener(
2038 &mut self,
2039 address: StdSocketAddr,
2040 ) -> Result<(Token, MioTcpListener), ProxyError> {
2041 let listener = self
2042 .listeners
2043 .values()
2044 .find(|listener| listener.borrow().address == address)
2045 .ok_or(ProxyError::NoListenerFound(address))?;
2046
2047 let mut owned = listener.borrow_mut();
2048
2049 let taken_listener = owned
2050 .listener
2051 .take()
2052 .ok_or(ProxyError::UnactivatedListener)?;
2053
2054 owned.active = false;
2057
2058 Ok((owned.token, taken_listener))
2059 }
2060
2061 pub fn update_listener(&mut self, patch: UpdateHttpsListenerConfig) -> Result<(), ProxyError> {
2063 let address: std::net::SocketAddr = patch.address.into();
2064 let listener = self
2065 .listeners
2066 .values()
2067 .find(|l| l.borrow().address == address)
2068 .ok_or(ProxyError::NoListenerFound(address))?;
2069 listener
2070 .borrow_mut()
2071 .update_config(&patch)
2072 .map_err(|listener_error| ProxyError::ListenerActivation {
2073 address,
2074 listener_error,
2075 })
2076 }
2077
2078 pub fn add_cluster(
2079 &mut self,
2080 mut cluster: Cluster,
2081 ) -> Result<Option<ResponseContent>, ProxyError> {
2082 let mut cluster_overrides = cluster.answers.clone();
2083 if let Some(answer_503) = cluster.answer_503.take() {
2084 cluster_overrides
2085 .entry("503".to_owned())
2086 .or_insert(answer_503);
2087 }
2088 if !cluster_overrides.is_empty() {
2089 for listener in self.listeners.values() {
2090 listener
2091 .borrow()
2092 .answers
2093 .borrow_mut()
2094 .add_cluster_answers(&cluster.cluster_id, &cluster_overrides)
2095 .map_err(|(status, error)| {
2096 ProxyError::AddCluster(ListenerError::TemplateParse(status, error))
2097 })?;
2098 }
2099 }
2100 self.clusters.insert(cluster.cluster_id.clone(), cluster);
2101 Ok(None)
2102 }
2103
2104 pub fn remove_cluster(
2105 &mut self,
2106 cluster_id: &str,
2107 ) -> Result<Option<ResponseContent>, ProxyError> {
2108 self.clusters.remove(cluster_id);
2109 for listener in self.listeners.values() {
2110 listener
2111 .borrow()
2112 .answers
2113 .borrow_mut()
2114 .remove_cluster_answers(cluster_id);
2115 }
2116
2117 Ok(None)
2118 }
2119
2120 pub fn add_https_frontend(
2121 &mut self,
2122 front: RequestHttpFrontend,
2123 ) -> Result<Option<ResponseContent>, ProxyError> {
2124 let mut front = front.clone().to_frontend().map_err(|request_error| {
2125 ProxyError::WrongInputFrontend {
2126 front: Box::new(front),
2127 error: request_error.to_string(),
2128 }
2129 })?;
2130
2131 let mut listener = self
2132 .listeners
2133 .values()
2134 .find(|l| l.borrow().address == front.address)
2135 .ok_or(ProxyError::NoListenerFound(front.address))?
2136 .borrow_mut();
2137
2138 let hsts_origin = if front.hsts.is_none() && listener.config.hsts.is_some() {
2151 front.hsts = listener.config.hsts;
2152 crate::router::HstsOrigin::InheritedFromListenerDefault
2153 } else {
2154 crate::router::HstsOrigin::Explicit
2155 };
2156
2157 listener.set_tags(front.hostname.to_owned(), front.tags.to_owned());
2158 listener
2159 .add_https_front_with_hsts_origin(front, hsts_origin)
2160 .map_err(ProxyError::AddFrontend)?;
2161 Ok(None)
2162 }
2163
2164 pub fn remove_https_frontend(
2165 &mut self,
2166 front: RequestHttpFrontend,
2167 ) -> Result<Option<ResponseContent>, ProxyError> {
2168 let front = front.clone().to_frontend().map_err(|request_error| {
2169 ProxyError::WrongInputFrontend {
2170 front: Box::new(front),
2171 error: request_error.to_string(),
2172 }
2173 })?;
2174
2175 let mut listener = self
2176 .listeners
2177 .values()
2178 .find(|l| l.borrow().address == front.address)
2179 .ok_or(ProxyError::NoListenerFound(front.address))?
2180 .borrow_mut();
2181
2182 let hostname = front.hostname.to_owned();
2183
2184 listener
2185 .remove_https_front(front)
2186 .map_err(ProxyError::RemoveFrontend)?;
2187
2188 if !listener.fronts.has_hostname(&hostname) {
2189 listener.set_tags(hostname, None);
2190 }
2191 Ok(None)
2192 }
2193
2194 pub fn add_certificate(
2195 &mut self,
2196 add_certificate: AddCertificate,
2197 ) -> Result<Option<ResponseContent>, ProxyError> {
2198 let address = add_certificate.address.into();
2199
2200 let listener = self
2201 .listeners
2202 .values()
2203 .find(|l| l.borrow().address == address)
2204 .ok_or(ProxyError::NoListenerFound(address))?
2205 .borrow_mut();
2206
2207 let mut resolver = listener
2208 .resolver
2209 .0
2210 .lock()
2211 .map_err(|e| ProxyError::Lock(e.to_string()))?;
2212
2213 resolver
2214 .add_certificate(&add_certificate)
2215 .map_err(ProxyError::AddCertificate)?;
2216
2217 Ok(None)
2218 }
2219
2220 pub fn remove_certificate(
2222 &mut self,
2223 remove_certificate: RemoveCertificate,
2224 ) -> Result<Option<ResponseContent>, ProxyError> {
2225 let address = remove_certificate.address.into();
2226
2227 let fingerprint = Fingerprint(
2228 hex::decode(&remove_certificate.fingerprint)
2229 .map_err(ProxyError::WrongCertificateFingerprint)?,
2230 );
2231
2232 let listener = self
2233 .listeners
2234 .values()
2235 .find(|l| l.borrow().address == address)
2236 .ok_or(ProxyError::NoListenerFound(address))?
2237 .borrow_mut();
2238
2239 let mut resolver = listener
2240 .resolver
2241 .0
2242 .lock()
2243 .map_err(|e| ProxyError::Lock(e.to_string()))?;
2244
2245 resolver
2246 .remove_certificate(&fingerprint)
2247 .map_err(ProxyError::RemoveCertificate)?;
2248
2249 Ok(None)
2250 }
2251
2252 pub fn replace_certificate(
2254 &mut self,
2255 replace_certificate: ReplaceCertificate,
2256 ) -> Result<Option<ResponseContent>, ProxyError> {
2257 let address = replace_certificate.address.into();
2258
2259 let listener = self
2260 .listeners
2261 .values()
2262 .find(|l| l.borrow().address == address)
2263 .ok_or(ProxyError::NoListenerFound(address))?
2264 .borrow_mut();
2265
2266 let mut resolver = listener
2267 .resolver
2268 .0
2269 .lock()
2270 .map_err(|e| ProxyError::Lock(e.to_string()))?;
2271
2272 resolver
2273 .replace_certificate(&replace_certificate)
2274 .map_err(ProxyError::ReplaceCertificate)?;
2275
2276 Ok(None)
2277 }
2278}
2279
2280impl ProxyConfiguration for HttpsProxy {
2281 fn accept(&mut self, token: ListenToken) -> Result<MioTcpStream, AcceptError> {
2282 match self.listeners.get(&Token(token.0)) {
2283 Some(listener) => listener.borrow_mut().accept(),
2284 None => Err(AcceptError::IoError),
2285 }
2286 }
2287
2288 fn create_session(
2289 &mut self,
2290 mut frontend_sock: MioTcpStream,
2291 token: ListenToken,
2292 wait_time: Duration,
2293 proxy: Rc<RefCell<Self>>,
2294 ) -> Result<(), AcceptError> {
2295 let listener = self
2296 .listeners
2297 .get(&Token(token.0))
2298 .ok_or(AcceptError::IoError)?;
2299 if let Err(e) = frontend_sock.set_nodelay(true) {
2300 error!(
2301 "{} error setting nodelay on front socket({:?}): {:?}",
2302 log_module_context!(),
2303 frontend_sock,
2304 e
2305 );
2306 }
2307
2308 let owned = listener.borrow();
2309 let rustls_details = ServerConnection::new(owned.rustls_details.clone()).map_err(|e| {
2310 error!(
2311 "{} failed to create server session: {:?}",
2312 log_module_context!(),
2313 e
2314 );
2315 AcceptError::IoError
2316 })?;
2317
2318 let mut session_manager = self.sessions.borrow_mut();
2319 let entry = session_manager.slab.vacant_entry();
2320 let session_token = Token(entry.key());
2321 debug_assert_eq!(
2326 session_token.0,
2327 entry.key(),
2328 "HTTPS session token must equal its slab key"
2329 );
2330
2331 self.registry
2332 .register(
2333 &mut frontend_sock,
2334 session_token,
2335 Interest::READABLE | Interest::WRITABLE,
2336 )
2337 .map_err(|register_error| {
2338 error!(
2339 "{} error registering front socket({:?}): {:?}",
2340 log_module_context!(),
2341 frontend_sock,
2342 register_error
2343 );
2344 AcceptError::RegisterError
2345 })?;
2346
2347 let public_address: StdSocketAddr = match owned.config.public_address {
2348 Some(pub_addr) => pub_addr.into(),
2349 None => owned.config.address.into(),
2350 };
2351
2352 let session = Rc::new(RefCell::new(HttpsSession::new(
2353 Duration::from_secs(owned.config.back_timeout as u64),
2354 Duration::from_secs(owned.config.connect_timeout as u64),
2355 Duration::from_secs(owned.config.front_timeout as u64),
2356 Duration::from_secs(owned.config.request_timeout as u64),
2357 owned.config.expect_proxy,
2358 listener.clone(),
2359 Rc::downgrade(&self.pool),
2360 proxy,
2361 public_address,
2362 rustls_details,
2363 frontend_sock,
2364 session_token,
2365 wait_time,
2366 )));
2367 debug_assert_eq!(
2371 session.borrow().frontend_token(),
2372 session_token,
2373 "stored HTTPS session must report the slab token as its frontend token"
2374 );
2375 entry.insert(session);
2376
2377 Ok(())
2378 }
2379
2380 fn notify(&mut self, request: WorkerRequest) -> WorkerResponse {
2381 let request_id = request.id.clone();
2382
2383 let request_type = match request.content.request_type {
2384 Some(t) => t,
2385 None => return WorkerResponse::error(request_id, "Empty request"),
2386 };
2387
2388 let content_result = match request_type {
2389 RequestType::AddCluster(cluster) => {
2390 debug!(
2391 "{} {} add cluster {:?}",
2392 log_module_context!(),
2393 request_id,
2394 cluster
2395 );
2396 self.add_cluster(cluster)
2397 }
2398 RequestType::RemoveCluster(cluster_id) => {
2399 debug!(
2400 "{} {} remove cluster {:?}",
2401 log_module_context!(),
2402 request_id,
2403 cluster_id
2404 );
2405 self.remove_cluster(&cluster_id)
2406 }
2407 RequestType::AddHttpsFrontend(front) => {
2408 debug!(
2409 "{} {} add https front {:?}",
2410 log_module_context!(),
2411 request_id,
2412 front
2413 );
2414 self.add_https_frontend(front)
2415 }
2416 RequestType::RemoveHttpsFrontend(front) => {
2417 debug!(
2418 "{} {} remove https front {:?}",
2419 log_module_context!(),
2420 request_id,
2421 front
2422 );
2423 self.remove_https_frontend(front)
2424 }
2425 RequestType::AddCertificate(add_certificate) => {
2426 debug!(
2427 "{} {} add certificate: {:?}",
2428 log_module_context!(),
2429 request_id,
2430 add_certificate
2431 );
2432 self.add_certificate(add_certificate)
2433 }
2434 RequestType::RemoveCertificate(remove_certificate) => {
2435 debug!(
2436 "{} {} remove certificate: {:?}",
2437 log_module_context!(),
2438 request_id,
2439 remove_certificate
2440 );
2441 self.remove_certificate(remove_certificate)
2442 }
2443 RequestType::ReplaceCertificate(replace_certificate) => {
2444 debug!(
2445 "{} {} replace certificate: {:?}",
2446 log_module_context!(),
2447 request_id,
2448 replace_certificate
2449 );
2450 self.replace_certificate(replace_certificate)
2451 }
2452 RequestType::RemoveListener(remove) => {
2453 debug!(
2454 "{} removing HTTPS listener at address {:?}",
2455 log_module_context!(),
2456 remove.address
2457 );
2458 self.remove_listener(remove)
2459 }
2460 RequestType::SoftStop(_) => {
2461 debug!(
2462 "{} {} processing soft shutdown",
2463 log_module_context!(),
2464 request_id
2465 );
2466 match self.soft_stop() {
2467 Ok(_) => {
2468 info!(
2469 "{} {} soft stop successful",
2470 log_module_context!(),
2471 request_id
2472 );
2473 return WorkerResponse::processing(request.id);
2474 }
2475 Err(e) => Err(e),
2476 }
2477 }
2478 RequestType::HardStop(_) => {
2479 debug!(
2480 "{} {} processing hard shutdown",
2481 log_module_context!(),
2482 request_id
2483 );
2484 match self.hard_stop() {
2485 Ok(_) => {
2486 debug!(
2487 "{} {} hard stop successful",
2488 log_module_context!(),
2489 request_id
2490 );
2491 return WorkerResponse::processing(request.id);
2492 }
2493 Err(e) => Err(e),
2494 }
2495 }
2496 RequestType::Status(_) => {
2497 debug!("{} {} status", log_module_context!(), request_id);
2498 Ok(None)
2499 }
2500 RequestType::QueryCertificatesFromWorkers(filters) => {
2501 if let Some(domain) = filters.domain {
2502 debug!(
2503 "{} {} query certificate for domain_bytes={}",
2504 log_module_context!(),
2505 request_id,
2506 domain.len(),
2507 );
2508 self.query_certificate_for_domain(domain)
2509 } else {
2510 debug!(
2511 "{} {} query all certificates",
2512 log_module_context!(),
2513 request_id
2514 );
2515 self.query_all_certificates()
2516 }
2517 }
2518 other_request => {
2519 debug!(
2520 "{} {} unsupported message for HTTPS proxy, ignoring {:?}",
2521 log_module_context!(),
2522 request.id,
2523 other_request
2524 );
2525 Err(ProxyError::UnsupportedMessage)
2526 }
2527 };
2528
2529 match content_result {
2530 Ok(content) => {
2531 debug!("{} {} successful", log_module_context!(), request_id);
2532 match content {
2533 Some(content) => WorkerResponse::ok_with_content(request_id, content),
2534 None => WorkerResponse::ok(request_id),
2535 }
2536 }
2537 Err(proxy_error) => {
2538 debug!(
2539 "{} {} unsuccessful: {}",
2540 log_module_context!(),
2541 request_id,
2542 proxy_error
2543 );
2544 WorkerResponse::error(request_id, proxy_error)
2545 }
2546 }
2547 }
2548}
2549impl L7Proxy for HttpsProxy {
2550 fn kind(&self) -> ListenerType {
2551 ListenerType::Https
2552 }
2553
2554 fn register_socket(
2555 &self,
2556 socket: &mut MioTcpStream,
2557 token: Token,
2558 interest: Interest,
2559 ) -> Result<(), std::io::Error> {
2560 self.registry.register(socket, token, interest)
2561 }
2562
2563 fn deregister_socket(&self, tcp_stream: &mut MioTcpStream) -> Result<(), std::io::Error> {
2564 self.registry.deregister(tcp_stream)
2565 }
2566
2567 fn add_session(&self, session: Rc<RefCell<dyn ProxySession>>) -> Token {
2568 let mut session_manager = self.sessions.borrow_mut();
2569 let entry = session_manager.slab.vacant_entry();
2570 let token = Token(entry.key());
2571 let _entry = entry.insert(session);
2572 token
2573 }
2574
2575 fn remove_session(&self, token: Token) -> bool {
2576 let mut sessions = self.sessions.borrow_mut();
2577 sessions.untrack_all_cluster_ip(token);
2580 sessions.slab.try_remove(token.0).is_some()
2581 }
2582
2583 fn backends(&self) -> Rc<RefCell<BackendMap>> {
2584 self.backends.clone()
2585 }
2586
2587 fn clusters(&self) -> &HashMap<ClusterId, Cluster> {
2588 &self.clusters
2589 }
2590
2591 fn sessions(&self) -> Rc<RefCell<SessionManager>> {
2592 self.sessions.clone()
2593 }
2594}
2595
2596fn rustls_version_str(version: ProtocolVersion) -> &'static str {
2598 match version {
2599 ProtocolVersion::SSLv2 => "tls.version.SSLv2",
2600 ProtocolVersion::SSLv3 => "tls.version.SSLv3",
2601 ProtocolVersion::TLSv1_0 => "tls.version.TLSv1_0",
2602 ProtocolVersion::TLSv1_1 => "tls.version.TLSv1_1",
2603 ProtocolVersion::TLSv1_2 => "tls.version.TLSv1_2",
2604 ProtocolVersion::TLSv1_3 => "tls.version.TLSv1_3",
2605 ProtocolVersion::DTLSv1_0 => "tls.version.DTLSv1_0",
2606 ProtocolVersion::DTLSv1_2 => "tls.version.DTLSv1_2",
2607 ProtocolVersion::DTLSv1_3 => "tls.version.DTLSv1_3",
2608 ProtocolVersion::Unknown(_) => "tls.version.Unknown",
2609 _ => "tls.version.unimplemented",
2610 }
2611}
2612
2613pub(crate) fn rustls_version_label(version: ProtocolVersion) -> Option<&'static str> {
2620 match version {
2621 ProtocolVersion::SSLv2 => Some("SSLv2"),
2622 ProtocolVersion::SSLv3 => Some("SSLv3"),
2623 ProtocolVersion::TLSv1_0 => Some("TLSv1.0"),
2624 ProtocolVersion::TLSv1_1 => Some("TLSv1.1"),
2625 ProtocolVersion::TLSv1_2 => Some("TLSv1.2"),
2626 ProtocolVersion::TLSv1_3 => Some("TLSv1.3"),
2627 ProtocolVersion::DTLSv1_0 => Some("DTLSv1.0"),
2628 ProtocolVersion::DTLSv1_2 => Some("DTLSv1.2"),
2629 ProtocolVersion::DTLSv1_3 => Some("DTLSv1.3"),
2630 _ => None,
2631 }
2632}
2633
2634fn rustls_ciphersuite_str(cipher: SupportedCipherSuite) -> &'static str {
2636 match cipher.suite() {
2637 CipherSuite::TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256 => {
2638 "tls.cipher.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256"
2639 }
2640 CipherSuite::TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256 => {
2641 "tls.cipher.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256"
2642 }
2643 CipherSuite::TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 => {
2644 "tls.cipher.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256"
2645 }
2646 CipherSuite::TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 => {
2647 "tls.cipher.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"
2648 }
2649 CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256 => {
2650 "tls.cipher.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256"
2651 }
2652 CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384 => {
2653 "tls.cipher.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384"
2654 }
2655 CipherSuite::TLS13_CHACHA20_POLY1305_SHA256 => "tls.cipher.TLS13_CHACHA20_POLY1305_SHA256",
2656 CipherSuite::TLS13_AES_256_GCM_SHA384 => "tls.cipher.TLS13_AES_256_GCM_SHA384",
2657 CipherSuite::TLS13_AES_128_GCM_SHA256 => "tls.cipher.TLS13_AES_128_GCM_SHA256",
2658 _ => "tls.cipher.Unsupported",
2659 }
2660}
2661
2662pub(crate) fn rustls_ciphersuite_label(cipher: SupportedCipherSuite) -> Option<&'static str> {
2669 match cipher.suite() {
2670 CipherSuite::TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256 => {
2671 Some("TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256")
2672 }
2673 CipherSuite::TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256 => {
2674 Some("TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256")
2675 }
2676 CipherSuite::TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 => {
2677 Some("TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256")
2678 }
2679 CipherSuite::TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 => {
2680 Some("TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384")
2681 }
2682 CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256 => {
2683 Some("TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256")
2684 }
2685 CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384 => {
2686 Some("TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384")
2687 }
2688 CipherSuite::TLS13_CHACHA20_POLY1305_SHA256 => Some("TLS13_CHACHA20_POLY1305_SHA256"),
2689 CipherSuite::TLS13_AES_256_GCM_SHA384 => Some("TLS13_AES_256_GCM_SHA384"),
2690 CipherSuite::TLS13_AES_128_GCM_SHA256 => Some("TLS13_AES_128_GCM_SHA256"),
2691 _ => None,
2692 }
2693}
2694
2695pub mod testing {
2696 use crate::testing::*;
2697
2698 pub fn start_https_worker(
2700 config: HttpsListenerConfig,
2701 channel: ProxyChannel,
2702 max_buffers: usize,
2703 buffer_size: usize,
2704 ) -> anyhow::Result<()> {
2705 let address = config.address.into();
2706
2707 let ServerParts {
2708 event_loop,
2709 registry,
2710 sessions,
2711 pool,
2712 backends,
2713 client_scm_socket: _,
2714 server_scm_socket,
2715 server_config,
2716 } = prebuild_server(max_buffers, buffer_size, true)?;
2717
2718 let token = {
2719 let mut sessions = sessions.borrow_mut();
2720 let entry = sessions.slab.vacant_entry();
2721 let key = entry.key();
2722 let _ = entry.insert(Rc::new(RefCell::new(ListenSession {
2723 protocol: Protocol::HTTPSListen,
2724 })));
2725 Token(key)
2726 };
2727
2728 let mut proxy = HttpsProxy::new(registry, sessions.clone(), pool.clone(), backends.clone());
2729 proxy
2730 .add_listener(config, token)
2731 .with_context(|| "Failed at creating adding the listener")?;
2732 proxy
2733 .activate_listener(&address, None)
2734 .with_context(|| "Failed at creating activating the listener")?;
2735
2736 let mut server = Server::new(
2737 event_loop,
2738 channel,
2739 server_scm_socket,
2740 sessions,
2741 pool,
2742 backends,
2743 None,
2744 Some(proxy),
2745 None,
2746 server_config,
2747 None,
2748 false,
2749 )
2750 .with_context(|| "Failed at creating server")?;
2751
2752 debug!("{} starting event loop", log_module_context!());
2753 server.run();
2754 debug!("{} ending event loop", log_module_context!());
2755 Ok(())
2756 }
2757}
2758
2759#[cfg(test)]
2760mod tests {
2761 use std::sync::Arc;
2762
2763 use sozu_command::{
2764 config::ListenerBuilder,
2765 proto::command::{CertificateAndKey, SocketAddress},
2766 };
2767
2768 use super::*;
2769 use crate::router::{MethodRule, PathRule, Route, Router, pattern_trie::TrieNode};
2770
2771 #[test]
2772 fn successful_tls_handshake_log_bounds_sni_and_alpn() {
2773 const SNI_SECRET: &str = "HTTPS_HANDSHAKE_SNI_SECRET_SENTINEL";
2774 const ALPN_SECRET: &str = "HTTPS_HANDSHAKE_ALPN_SECRET_SENTINEL";
2775
2776 let sni = format!("{SNI_SECRET}{}", "x".repeat(4096));
2777 let alpn = format!("{ALPN_SECRET}{}", "x".repeat(4096));
2778 let sni_len = sni.len();
2779 let alpn_len = alpn.len();
2780 let output = successful_tls_handshake_summary(Some(&sni), Some(&alpn));
2781
2782 for secret in [SNI_SECRET, ALPN_SECRET] {
2783 assert!(
2784 !output.contains(secret),
2785 "successful TLS handshake log leaked {secret}: {output}"
2786 );
2787 }
2788 for metadata in [
2789 format!("sni_bytes=Some({sni_len})"),
2790 format!("alpn_bytes=Some({alpn_len})"),
2791 ] {
2792 assert!(
2793 output.contains(&metadata),
2794 "successful TLS handshake log omitted {metadata}: {output}"
2795 );
2796 }
2797 assert!(
2798 output.len() <= 512,
2799 "successful TLS handshake log is not bounded: {} bytes",
2800 output.len()
2801 );
2802 }
2803
2804 fn proxy_with_certificate_domain(domain: String) -> HttpsProxy {
2805 let crate::testing::ServerParts {
2806 registry,
2807 sessions,
2808 pool,
2809 backends,
2810 ..
2811 } = crate::testing::prebuild_server(16, 16_384, false)
2812 .expect("test HTTPS proxy dependencies must initialize");
2813 let address = SocketAddress::new_v4(127, 0, 0, 1, crate::testing::provide_port());
2814 let mut config = ListenerBuilder::new_https(address)
2815 .to_tls(None)
2816 .expect("test HTTPS listener config must build");
2817 config.cipher_list = vec!["TLS13_AES_256_GCM_SHA384".to_owned()];
2818 config.groups_list = vec!["P-256".to_owned()];
2819 let mut proxy = HttpsProxy::new(registry, sessions, pool, backends);
2820 proxy
2821 .add_listener(config, Token(10))
2822 .expect("test HTTPS listener must be added");
2823 proxy
2824 .add_certificate(AddCertificate {
2825 address,
2826 certificate: CertificateAndKey {
2827 certificate: include_str!("../assets/certificate.pem").to_owned(),
2828 key: include_str!("../assets/key.pem").to_owned(),
2829 certificate_chain: Vec::new(),
2830 versions: Vec::new(),
2831 names: vec![domain],
2832 },
2833 expired_at: None,
2834 })
2835 .expect("test certificate must be added");
2836 proxy
2837 }
2838
2839 #[test]
2840 fn query_all_certificates_log_redacts_response_domains() {
2841 const DOMAIN_SECRET: &str = "QUERY_ALL_CERTIFICATE_DOMAIN_SECRET_SENTINEL";
2842
2843 let domain = format!("{DOMAIN_SECRET}{}", "x".repeat(2048));
2847 let output = crate::capture_test_logs(move || {
2848 let mut proxy = proxy_with_certificate_domain(domain);
2849 proxy
2850 .query_all_certificates()
2851 .expect("certificate query must succeed");
2852 });
2853
2854 assert!(
2855 !output.contains(DOMAIN_SECRET),
2856 "Certificates::All log leaked response domain {DOMAIN_SECRET}"
2857 );
2858 for metadata in ["listeners_count=1", "certificates_count=1"] {
2859 assert!(
2860 output.contains(metadata),
2861 "Certificates::All log omitted bounded metadata {metadata}: {output}"
2862 );
2863 }
2864 assert!(
2865 output.len() <= 1024,
2866 "Certificates::All log capture is not bounded: {} bytes",
2867 output.len()
2868 );
2869 }
2870
2871 #[test]
2872 fn query_certificate_for_domain_log_redacts_request_and_response_domain() {
2873 const DOMAIN_SECRET: &str = "QUERY_ONE_CERTIFICATE_DOMAIN_SECRET_SENTINEL";
2874
2875 let domain = format!("{DOMAIN_SECRET}{}", "x".repeat(2048));
2878 let domain_len = domain.len();
2879 let output = crate::capture_test_logs_at_level("debug", move || {
2880 let mut proxy = proxy_with_certificate_domain(domain.clone());
2881 let response = proxy.notify(WorkerRequest {
2882 id: "query-domain-redaction-test".to_owned(),
2883 content: RequestType::QueryCertificatesFromWorkers(
2884 sozu_command::proto::command::QueryCertificatesFilters {
2885 domain: Some(domain),
2886 fingerprint: None,
2887 },
2888 )
2889 .into(),
2890 });
2891 assert_eq!(
2892 response.status,
2893 sozu_command::proto::command::ResponseStatus::Ok as i32,
2894 "domain certificate query must succeed: {}",
2895 response.message
2896 );
2897 });
2898
2899 assert!(
2900 !output.contains(DOMAIN_SECRET),
2901 "Certificates::Domain log leaked request or response domain {DOMAIN_SECRET}"
2902 );
2903 for metadata in [
2904 format!("domain_bytes={domain_len}"),
2905 "listeners_count=1".to_owned(),
2906 "certificates_count=1".to_owned(),
2907 ] {
2908 assert!(
2909 output.contains(&metadata),
2910 "Certificates::Domain log omitted bounded metadata {metadata}: {output}"
2911 );
2912 }
2913 assert!(
2914 output.len() <= 1024,
2915 "Certificates::Domain log capture is not bounded: {} bytes",
2916 output.len()
2917 );
2918 }
2919
2920 #[test]
2938 fn frontend_from_request_test() {
2939 let cluster_id1 = "cluster_1".to_owned();
2940 let cluster_id2 = "cluster_2".to_owned();
2941 let cluster_id3 = "cluster_3".to_owned();
2942 let uri1 = "/".to_owned();
2943 let uri2 = "/yolo".to_owned();
2944 let uri3 = "/yolo/swag".to_owned();
2945
2946 let mut fronts = Router::new();
2947 assert!(fronts.add_tree_rule(
2948 "lolcatho.st".as_bytes(),
2949 &PathRule::Prefix(uri1),
2950 &MethodRule::new(None),
2951 &Route::ClusterId(cluster_id1.clone())
2952 ));
2953 assert!(fronts.add_tree_rule(
2954 "lolcatho.st".as_bytes(),
2955 &PathRule::Prefix(uri2),
2956 &MethodRule::new(None),
2957 &Route::ClusterId(cluster_id2)
2958 ));
2959 assert!(fronts.add_tree_rule(
2960 "lolcatho.st".as_bytes(),
2961 &PathRule::Prefix(uri3),
2962 &MethodRule::new(None),
2963 &Route::ClusterId(cluster_id3)
2964 ));
2965 assert!(fronts.add_tree_rule(
2966 "other.domain".as_bytes(),
2967 &PathRule::Prefix("test".to_string()),
2968 &MethodRule::new(None),
2969 &Route::ClusterId(cluster_id1)
2970 ));
2971
2972 let address = SocketAddress::new_v4(127, 0, 0, 1, 1032);
2973 let resolver = Arc::new(MutexCertificateResolver::default());
2974
2975 let crypto_provider = Arc::new(default_provider());
2976
2977 let server_config = RustlsServerConfig::builder_with_provider(crypto_provider)
2978 .with_protocol_versions(&[&rustls::version::TLS12, &rustls::version::TLS13])
2979 .expect("could not create rustls config server")
2980 .with_no_client_auth()
2981 .with_cert_resolver(resolver.clone());
2982
2983 let rustls_details = Arc::new(server_config);
2984
2985 let default_config = ListenerBuilder::new_https(address)
2986 .to_tls(None)
2987 .expect("Could not create default HTTPS listener config");
2988
2989 println!("it doesn't even matter");
2990
2991 let listener = HttpsListener {
2992 listener: None,
2993 address: address.into(),
2994 fronts,
2995 rustls_details,
2996 resolver,
2997 answers: Rc::new(RefCell::new(
2998 HttpAnswers::new(&std::collections::BTreeMap::new()).unwrap(),
2999 )),
3000 config: default_config,
3001 token: Token(0),
3002 active: true,
3003 tags: BTreeMap::new(),
3004 };
3005
3006 println!("TEST {}", line!());
3007 let frontend1 = listener.frontend_from_request("lolcatho.st", "/", &Method::Get);
3008 assert_eq!(
3009 frontend1
3010 .expect("should find a frontend")
3011 .cluster_id
3012 .as_deref(),
3013 Some("cluster_1")
3014 );
3015 println!("TEST {}", line!());
3016 let frontend2 = listener.frontend_from_request("lolcatho.st", "/test", &Method::Get);
3017 assert_eq!(
3018 frontend2
3019 .expect("should find a frontend")
3020 .cluster_id
3021 .as_deref(),
3022 Some("cluster_1")
3023 );
3024 println!("TEST {}", line!());
3025 let frontend3 = listener.frontend_from_request("lolcatho.st", "/yolo/test", &Method::Get);
3026 assert_eq!(
3027 frontend3
3028 .expect("should find a frontend")
3029 .cluster_id
3030 .as_deref(),
3031 Some("cluster_2")
3032 );
3033 println!("TEST {}", line!());
3034 let frontend4 = listener.frontend_from_request("lolcatho.st", "/yolo/swag", &Method::Get);
3035 assert_eq!(
3036 frontend4
3037 .expect("should find a frontend")
3038 .cluster_id
3039 .as_deref(),
3040 Some("cluster_3")
3041 );
3042 println!("TEST {}", line!());
3043 let frontend5 = listener.frontend_from_request("domain", "/", &Method::Get);
3044 assert!(frontend5.is_err());
3045 }
3047
3048 #[test]
3049 fn wildcard_certificate_names() {
3050 let mut trie = TrieNode::root();
3051
3052 trie.domain_insert("*.services.clever-cloud.com".as_bytes().to_vec(), 1u8);
3053 trie.domain_insert("*.clever-cloud.com".as_bytes().to_vec(), 2u8);
3054 trie.domain_insert("services.clever-cloud.com".as_bytes().to_vec(), 0u8);
3055 trie.domain_insert(
3056 "abprefix.services.clever-cloud.com".as_bytes().to_vec(),
3057 3u8,
3058 );
3059 trie.domain_insert(
3060 "cdprefix.services.clever-cloud.com".as_bytes().to_vec(),
3061 4u8,
3062 );
3063
3064 let res = trie.domain_lookup(b"test.services.clever-cloud.com", true);
3065 println!("query result: {res:?}");
3066
3067 assert_eq!(
3068 trie.domain_lookup(b"pgstudio.services.clever-cloud.com", true),
3069 Some(&("*.services.clever-cloud.com".as_bytes().to_vec(), 1u8))
3070 );
3071 assert_eq!(
3072 trie.domain_lookup(b"test-prefix.services.clever-cloud.com", true),
3073 Some(&("*.services.clever-cloud.com".as_bytes().to_vec(), 1u8))
3074 );
3075 }
3076
3077 #[test]
3078 fn wildcard_with_subdomains() {
3079 let mut trie = TrieNode::root();
3080
3081 trie.domain_insert("*.test.example.com".as_bytes().to_vec(), 1u8);
3082 trie.domain_insert("hello.sub.test.example.com".as_bytes().to_vec(), 2u8);
3083
3084 let res = trie.domain_lookup(b"sub.test.example.com", true);
3085 println!("query result: {res:?}");
3086
3087 assert_eq!(
3088 trie.domain_lookup(b"sub.test.example.com", true),
3089 Some(&("*.test.example.com".as_bytes().to_vec(), 1u8))
3090 );
3091 assert_eq!(
3092 trie.domain_lookup(b"hello.sub.test.example.com", true),
3093 Some(&("hello.sub.test.example.com".as_bytes().to_vec(), 2u8))
3094 );
3095
3096 let mut trie = TrieNode::root();
3098
3099 trie.domain_insert("hello.sub.test.example.com".as_bytes().to_vec(), 2u8);
3100 trie.domain_insert("*.test.example.com".as_bytes().to_vec(), 1u8);
3101
3102 let res = trie.domain_lookup(b"sub.test.example.com", true);
3103 println!("query result: {res:?}");
3104
3105 assert_eq!(
3106 trie.domain_lookup(b"sub.test.example.com", true),
3107 Some(&("*.test.example.com".as_bytes().to_vec(), 1u8))
3108 );
3109 assert_eq!(
3110 trie.domain_lookup(b"hello.sub.test.example.com", true),
3111 Some(&("hello.sub.test.example.com".as_bytes().to_vec(), 2u8))
3112 );
3113 }
3114
3115 #[test]
3116 fn h2_stream_idle_timeout_inherits_back_timeout() {
3117 use std::time::Duration;
3118
3119 let address = SocketAddress::new_v4(127, 0, 0, 1, 1041);
3120 let build = |back_timeout: u32, explicit: Option<u32>| -> HttpsListener {
3121 let mut cfg = ListenerBuilder::new_https(address)
3122 .to_tls(None)
3123 .expect("default HTTPS listener config");
3124 cfg.back_timeout = back_timeout;
3125 cfg.h2_stream_idle_timeout_seconds = explicit;
3126 HttpsListener::try_new(cfg, Token(0)).expect("build listener")
3127 };
3128
3129 assert_eq!(
3131 build(180, None).get_h2_stream_idle_timeout(),
3132 Duration::from_secs(180)
3133 );
3134
3135 assert_eq!(
3137 build(5, None).get_h2_stream_idle_timeout(),
3138 Duration::from_secs(30)
3139 );
3140
3141 assert_eq!(
3143 build(180, Some(10)).get_h2_stream_idle_timeout(),
3144 Duration::from_secs(10)
3145 );
3146 assert_eq!(
3147 build(5, Some(600)).get_h2_stream_idle_timeout(),
3148 Duration::from_secs(600)
3149 );
3150
3151 assert_eq!(
3153 build(180, Some(0)).get_h2_stream_idle_timeout(),
3154 Duration::from_secs(1)
3155 );
3156 }
3157}