1use std::{
2 cell::RefCell,
3 collections::{hash_map::Entry, BTreeMap, HashMap},
4 io::ErrorKind,
5 net::{Shutdown, SocketAddr as StdSocketAddr},
6 os::unix::io::AsRawFd,
7 rc::{Rc, Weak},
8 str::{from_utf8, from_utf8_unchecked},
9 sync::Arc,
10 time::{Duration, Instant},
11};
12
13use mio::{
14 net::{TcpListener as MioTcpListener, TcpStream as MioTcpStream},
15 unix::SourceFd,
16 Interest, Registry, Token,
17};
18use rustls::{
19 crypto::{
20 ring::{
21 self,
22 cipher_suite::{
23 TLS13_AES_128_GCM_SHA256, TLS13_AES_256_GCM_SHA384, TLS13_CHACHA20_POLY1305_SHA256,
24 TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256, TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
25 TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,
26 TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256, TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
27 TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256,
28 },
29 },
30 CryptoProvider,
31 },
32 CipherSuite, ProtocolVersion, ServerConfig as RustlsServerConfig, ServerConnection,
33 SupportedCipherSuite,
34};
35use rusty_ulid::Ulid;
36
37use sozu_command::{
38 certificate::Fingerprint,
39 config::DEFAULT_CIPHER_SUITES,
40 proto::command::{
41 request::RequestType, response_content::ContentType, AddCertificate, CertificateSummary,
42 CertificatesByAddress, Cluster, HttpsListenerConfig, ListOfCertificatesByAddress,
43 ListenerType, RemoveCertificate, RemoveListener, ReplaceCertificate, RequestHttpFrontend,
44 ResponseContent, TlsVersion, WorkerRequest, WorkerResponse,
45 },
46 ready::Ready,
47 response::HttpFrontend,
48 state::ClusterId,
49};
50
51use crate::{
52 backends::BackendMap,
53 pool::Pool,
54 protocol::{
55 h2::Http2,
56 http::{
57 answers::HttpAnswers,
58 parser::{hostname_and_port, Method},
59 ResponseStream,
60 },
61 proxy_protocol::expect::ExpectProxyProtocol,
62 rustls::TlsHandshake,
63 Http, Pipe, SessionState,
64 },
65 router::{Route, Router},
66 server::{ListenToken, SessionManager},
67 socket::{server_bind, FrontRustls},
68 timer::TimeoutContainer,
69 tls::MutexCertificateResolver,
70 util::UnwrapLog,
71 AcceptError, CachedTags, FrontendFromRequestError, L7ListenerHandler, L7Proxy, ListenerError,
72 ListenerHandler, Protocol, ProxyConfiguration, ProxyError, ProxySession, SessionIsToBeClosed,
73 SessionMetrics, SessionResult, StateMachineBuilder, StateResult,
74};
75
76const SERVER_PROTOS: &[&str] = &["http/1.1"];
78
79StateMachineBuilder! {
80 enum HttpsStateMachine impl SessionState {
87 Expect(ExpectProxyProtocol<MioTcpStream>, ServerConnection),
88 Handshake(TlsHandshake),
89 Http(Http<FrontRustls, HttpsListener>),
90 WebSocket(Pipe<FrontRustls, HttpsListener>),
91 Http2(Http2<FrontRustls>) -> todo!("H2"),
92 }
93}
94
95pub enum AlpnProtocols {
96 H2,
97 Http11,
98}
99
100pub struct HttpsSession {
101 answers: Rc<RefCell<HttpAnswers>>,
102 configured_backend_timeout: Duration,
103 configured_connect_timeout: Duration,
104 configured_frontend_timeout: Duration,
105 frontend_token: Token,
106 has_been_closed: bool,
107 last_event: Instant,
108 listener: Rc<RefCell<HttpsListener>>,
109 metrics: SessionMetrics,
110 peer_address: Option<StdSocketAddr>,
111 pool: Weak<RefCell<Pool>>,
112 proxy: Rc<RefCell<HttpsProxy>>,
113 public_address: StdSocketAddr,
114 state: HttpsStateMachine,
115 sticky_name: String,
116}
117
118impl HttpsSession {
119 #[allow(clippy::too_many_arguments)]
120 pub fn new(
121 answers: Rc<RefCell<HttpAnswers>>,
122 configured_backend_timeout: Duration,
123 configured_connect_timeout: Duration,
124 configured_frontend_timeout: Duration,
125 configured_request_timeout: Duration,
126 expect_proxy: bool,
127 listener: Rc<RefCell<HttpsListener>>,
128 pool: Weak<RefCell<Pool>>,
129 proxy: Rc<RefCell<HttpsProxy>>,
130 public_address: StdSocketAddr,
131 rustls_details: ServerConnection,
132 sock: MioTcpStream,
133 sticky_name: String,
134 token: Token,
135 wait_time: Duration,
136 ) -> HttpsSession {
137 let peer_address = if expect_proxy {
138 None
140 } else {
141 sock.peer_addr().ok()
142 };
143
144 let request_id = Ulid::generate();
145 let container_frontend_timeout = TimeoutContainer::new(configured_request_timeout, token);
146
147 let state = if expect_proxy {
148 trace!("starting in expect proxy state");
149 gauge_add!("protocol.proxy.expect", 1);
150 HttpsStateMachine::Expect(
151 ExpectProxyProtocol::new(container_frontend_timeout, sock, token, request_id),
152 rustls_details,
153 )
154 } else {
155 gauge_add!("protocol.tls.handshake", 1);
156 HttpsStateMachine::Handshake(TlsHandshake::new(
157 container_frontend_timeout,
158 rustls_details,
159 sock,
160 token,
161 request_id,
162 peer_address,
163 ))
164 };
165
166 let metrics = SessionMetrics::new(Some(wait_time));
167 HttpsSession {
168 answers,
169 configured_backend_timeout,
170 configured_connect_timeout,
171 configured_frontend_timeout,
172 frontend_token: token,
173 has_been_closed: false,
174 last_event: Instant::now(),
175 listener,
176 metrics,
177 peer_address,
178 pool,
179 proxy,
180 public_address,
181 state,
182 sticky_name,
183 }
184 }
185
186 pub fn upgrade(&mut self) -> SessionIsToBeClosed {
187 debug!("HTTP::upgrade");
188 let new_state = match self.state.take() {
189 HttpsStateMachine::Expect(expect, ssl) => self.upgrade_expect(expect, ssl),
190 HttpsStateMachine::Handshake(handshake) => self.upgrade_handshake(handshake),
191 HttpsStateMachine::Http(http) => self.upgrade_http(http),
192 HttpsStateMachine::Http2(_) => self.upgrade_http2(),
193 HttpsStateMachine::WebSocket(wss) => self.upgrade_websocket(wss),
194 HttpsStateMachine::FailedUpgrade(_) => unreachable!(),
195 };
196
197 match new_state {
198 Some(state) => {
199 self.state = state;
200 false
201 }
202 None => true,
204 }
205 }
206
207 fn upgrade_expect(
208 &mut self,
209 mut expect: ExpectProxyProtocol<MioTcpStream>,
210 ssl: ServerConnection,
211 ) -> Option<HttpsStateMachine> {
212 if let Some(ref addresses) = expect.addresses {
213 if let (Some(public_address), Some(session_address)) =
214 (addresses.destination(), addresses.source())
215 {
216 self.public_address = public_address;
217 self.peer_address = Some(session_address);
218
219 let ExpectProxyProtocol {
220 container_frontend_timeout,
221 frontend,
222 frontend_readiness: readiness,
223 request_id,
224 ..
225 } = expect;
226
227 let mut handshake = TlsHandshake::new(
228 container_frontend_timeout,
229 ssl,
230 frontend,
231 self.frontend_token,
232 request_id,
233 self.peer_address,
234 );
235 handshake.frontend_readiness.event = readiness.event;
236 handshake.frontend_readiness.event.insert(Ready::READABLE);
239
240 gauge_add!("protocol.proxy.expect", -1);
241 gauge_add!("protocol.tls.handshake", 1);
242 return Some(HttpsStateMachine::Handshake(handshake));
243 }
244 }
245
246 if !expect.container_frontend_timeout.cancel() {
248 error!("failed to cancel request timeout on expect upgrade phase for 'expect proxy protocol with AF_UNSPEC address'");
249 }
250
251 None
252 }
253
254 fn upgrade_handshake(&mut self, handshake: TlsHandshake) -> Option<HttpsStateMachine> {
255 let sni = handshake.session.server_name();
262 let alpn = handshake.session.alpn_protocol();
263 let alpn = alpn.and_then(|alpn| from_utf8(alpn).ok());
264 debug!(
265 "Successful TLS Handshake with, received: {:?} {:?}",
266 sni, alpn
267 );
268
269 let alpn = match alpn {
270 Some("http/1.1") => AlpnProtocols::Http11,
271 Some("h2") => AlpnProtocols::H2,
272 Some(other) => {
273 error!("Unsupported ALPN protocol: {}", other);
274 return None;
275 }
276 None => AlpnProtocols::Http11,
278 };
279
280 if let Some(version) = handshake.session.protocol_version() {
281 incr!(rustls_version_str(version));
282 };
283 if let Some(cipher) = handshake.session.negotiated_cipher_suite() {
284 incr!(rustls_ciphersuite_str(cipher));
285 };
286
287 let front_stream = FrontRustls {
288 stream: handshake.stream,
289 session: handshake.session,
290 };
291
292 gauge_add!("protocol.tls.handshake", -1);
293 match alpn {
294 AlpnProtocols::Http11 => {
295 let mut http = Http::new(
296 self.answers.clone(),
297 self.configured_backend_timeout,
298 self.configured_connect_timeout,
299 self.configured_frontend_timeout,
300 handshake.container_frontend_timeout,
301 front_stream,
302 self.frontend_token,
303 self.listener.clone(),
304 self.pool.clone(),
305 Protocol::HTTPS,
306 self.public_address,
307 handshake.request_id,
308 self.peer_address,
309 self.sticky_name.clone(),
310 )
311 .ok()?;
312
313 http.frontend_readiness.event = handshake.frontend_readiness.event;
314
315 gauge_add!("protocol.https", 1);
316 Some(HttpsStateMachine::Http(http))
317 }
318 AlpnProtocols::H2 => {
319 let mut http = Http2::new(
320 front_stream,
321 self.frontend_token,
322 self.pool.clone(),
323 Some(self.public_address),
324 None,
325 self.sticky_name.clone(),
326 );
327
328 http.frontend.readiness.event = handshake.frontend_readiness.event;
329
330 gauge_add!("protocol.http2", 1);
331 Some(HttpsStateMachine::Http2(http))
332 }
333 }
334 }
335
336 fn upgrade_http(&self, http: Http<FrontRustls, HttpsListener>) -> Option<HttpsStateMachine> {
337 debug!("https switching to wss");
338 let front_token = self.frontend_token;
339 let back_token = match http.backend_token {
340 Some(back_token) => back_token,
341 None => {
342 warn!(
343 "Could not upgrade https request on cluster '{:?}' ({:?}) using backend '{:?}' into secure websocket for request '{}'",
344 http.context.cluster_id, self.frontend_token, http.context.backend_id, http.context.id
345 );
346 return None;
347 }
348 };
349
350 let ws_context = http.websocket_context();
351 let mut container_frontend_timeout = http.container_frontend_timeout;
352 let mut container_backend_timeout = http.container_backend_timeout;
353 container_frontend_timeout.reset();
354 container_backend_timeout.reset();
355
356 let backend_buffer = if let ResponseStream::BackendAnswer(kawa) = http.response_stream {
357 kawa.storage.buffer
358 } else {
359 return None;
360 };
361
362 let mut pipe = Pipe::new(
363 backend_buffer,
364 http.context.backend_id,
365 http.backend_socket,
366 http.backend,
367 Some(container_backend_timeout),
368 Some(container_frontend_timeout),
369 http.context.cluster_id,
370 http.request_stream.storage.buffer,
371 front_token,
372 http.frontend_socket,
373 self.listener.clone(),
374 Protocol::HTTP,
375 http.context.id,
376 http.context.session_address,
377 ws_context,
378 );
379
380 pipe.frontend_readiness.event = http.frontend_readiness.event;
381 pipe.backend_readiness.event = http.backend_readiness.event;
382 pipe.set_back_token(back_token);
383
384 gauge_add!("protocol.https", -1);
385 gauge_add!("protocol.wss", 1);
386 gauge_add!("http.active_requests", -1);
387 gauge_add!("websocket.active_requests", 1);
388 Some(HttpsStateMachine::WebSocket(pipe))
389 }
390
391 fn upgrade_http2(&self) -> Option<HttpsStateMachine> {
392 todo!()
393 }
394
395 fn upgrade_websocket(
396 &self,
397 wss: Pipe<FrontRustls, HttpsListener>,
398 ) -> Option<HttpsStateMachine> {
399 error!("Upgrade called on WSS, this should not happen");
401 Some(HttpsStateMachine::WebSocket(wss))
402 }
403}
404
405impl ProxySession for HttpsSession {
406 fn close(&mut self) {
407 if self.has_been_closed {
408 return;
409 }
410
411 trace!("Closing HTTPS session");
412 self.metrics.service_stop();
413
414 match self.state.marker() {
416 StateMarker::Expect => gauge_add!("protocol.proxy.expect", -1),
417 StateMarker::Handshake => gauge_add!("protocol.tls.handshake", -1),
418 StateMarker::Http => gauge_add!("protocol.https", -1),
419 StateMarker::WebSocket => {
420 gauge_add!("protocol.wss", -1);
421 gauge_add!("websocket.active_requests", -1);
422 }
423 StateMarker::Http2 => gauge_add!("protocol.http2", -1),
424 }
425
426 if self.state.failed() {
427 match self.state.marker() {
428 StateMarker::Expect => incr!("https.upgrade.expect.failed"),
429 StateMarker::Handshake => incr!("https.upgrade.handshake.failed"),
430 StateMarker::Http => incr!("https.upgrade.http.failed"),
431 StateMarker::WebSocket => incr!("https.upgrade.wss.failed"),
432 StateMarker::Http2 => incr!("https.upgrade.http2.failed"),
433 }
434 return;
435 }
436
437 self.state.cancel_timeouts();
438 self.state.close(self.proxy.clone(), &mut self.metrics);
441
442 let front_socket = self.state.front_socket();
443 if let Err(e) = front_socket.shutdown(Shutdown::Both) {
444 if e.kind() != ErrorKind::NotConnected {
446 error!(
447 "error shutting down front socket({:?}): {:?}",
448 front_socket, e
449 );
450 }
451 }
452
453 let proxy = self.proxy.borrow();
455 let fd = front_socket.as_raw_fd();
456 if let Err(e) = proxy.registry.deregister(&mut SourceFd(&fd)) {
457 error!(
458 "error deregistering front socket({:?}) while closing HTTPS session: {:?}",
459 fd, e
460 );
461 }
462 proxy.remove_session(self.frontend_token);
463
464 self.has_been_closed = true;
465 }
466
467 fn timeout(&mut self, token: Token) -> SessionIsToBeClosed {
468 let session_result = self.state.timeout(token, &mut self.metrics);
469 session_result == StateResult::CloseSession
470 }
471
472 fn protocol(&self) -> Protocol {
473 Protocol::HTTPS
474 }
475
476 fn update_readiness(&mut self, token: Token, events: Ready) {
477 trace!(
478 "token {:?} got event {}",
479 token,
480 super::ready_to_string(events)
481 );
482 self.last_event = Instant::now();
483 self.metrics.wait_start();
484 self.state.update_readiness(token, events);
485 }
486
487 fn ready(&mut self, session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
488 self.metrics.service_start();
489
490 let session_result =
491 self.state
492 .ready(session.clone(), self.proxy.clone(), &mut self.metrics);
493
494 let to_be_closed = match session_result {
495 SessionResult::Close => true,
496 SessionResult::Continue => false,
497 SessionResult::Upgrade => match self.upgrade() {
498 false => self.ready(session),
499 true => true,
500 },
501 };
502
503 self.metrics.service_stop();
504 to_be_closed
505 }
506
507 fn shutting_down(&mut self) -> SessionIsToBeClosed {
508 self.state.shutting_down()
509 }
510
511 fn last_event(&self) -> Instant {
512 self.last_event
513 }
514
515 fn print_session(&self) {
516 self.state.print_state("HTTPS");
517 error!("Metrics: {:?}", self.metrics);
518 }
519
520 fn frontend_token(&self) -> Token {
521 self.frontend_token
522 }
523}
524
525pub type HostName = String;
526pub type PathBegin = String;
527
528pub struct HttpsListener {
529 active: bool,
530 address: StdSocketAddr,
531 answers: Rc<RefCell<HttpAnswers>>,
532 config: HttpsListenerConfig,
533 fronts: Router,
534 listener: Option<MioTcpListener>,
535 resolver: Arc<MutexCertificateResolver>,
536 rustls_details: Arc<RustlsServerConfig>,
537 tags: BTreeMap<String, CachedTags>,
538 token: Token,
539}
540
541impl ListenerHandler for HttpsListener {
542 fn get_addr(&self) -> &StdSocketAddr {
543 &self.address
544 }
545
546 fn get_tags(&self, key: &str) -> Option<&CachedTags> {
547 self.tags.get(key)
548 }
549
550 fn set_tags(&mut self, key: String, tags: Option<BTreeMap<String, String>>) {
551 match tags {
552 Some(tags) => self.tags.insert(key, CachedTags::new(tags)),
553 None => self.tags.remove(&key),
554 };
555 }
556}
557
558impl L7ListenerHandler for HttpsListener {
559 fn get_sticky_name(&self) -> &str {
560 &self.config.sticky_name
561 }
562
563 fn get_connect_timeout(&self) -> u32 {
564 self.config.connect_timeout
565 }
566
567 fn frontend_from_request(
568 &self,
569 host: &str,
570 uri: &str,
571 method: &Method,
572 ) -> Result<Route, FrontendFromRequestError> {
573 let start = Instant::now();
574 let (remaining_input, (hostname, _)) = match hostname_and_port(host.as_bytes()) {
575 Ok(tuple) => tuple,
576 Err(parse_error) => {
577 return Err(FrontendFromRequestError::HostParse {
579 host: host.to_owned(),
580 error: parse_error.to_string(),
581 });
582 }
583 };
584
585 if remaining_input != &b""[..] {
586 return Err(FrontendFromRequestError::InvalidCharsAfterHost(
587 host.to_owned(),
588 ));
589 }
590
591 let host = unsafe { from_utf8_unchecked(hostname) };
595
596 let route = self.fronts.lookup(host, uri, method).map_err(|e| {
597 incr!("http.failed_backend_matching");
598 FrontendFromRequestError::NoClusterFound(e)
599 })?;
600
601 let now = Instant::now();
602
603 if let Route::ClusterId(cluster) = &route {
604 time!("frontend_matching_time", cluster, (now - start).as_millis());
605 }
606
607 Ok(route)
608 }
609}
610
611impl HttpsListener {
612 pub fn try_new(
613 config: HttpsListenerConfig,
614 token: Token,
615 ) -> Result<HttpsListener, ListenerError> {
616 let resolver = Arc::new(MutexCertificateResolver::default());
617
618 let server_config = Arc::new(Self::create_rustls_context(&config, resolver.to_owned())?);
619
620 Ok(HttpsListener {
621 listener: None,
622 address: config.address.into(),
623 resolver,
624 rustls_details: server_config,
625 active: false,
626 fronts: Router::new(),
627 answers: Rc::new(RefCell::new(
628 HttpAnswers::new(&config.http_answers)
629 .map_err(|(status, error)| ListenerError::TemplateParse(status, error))?,
630 )),
631 config,
632 token,
633 tags: BTreeMap::new(),
634 })
635 }
636
637 pub fn activate(
638 &mut self,
639 registry: &Registry,
640 tcp_listener: Option<MioTcpListener>,
641 ) -> Result<Token, ListenerError> {
642 if self.active {
643 return Ok(self.token);
644 }
645 let address: StdSocketAddr = self.config.address.into();
646
647 let mut listener = match tcp_listener {
648 Some(tcp_listener) => tcp_listener,
649 None => {
650 server_bind(address).map_err(|server_bind_error| ListenerError::Activation {
651 address,
652 error: server_bind_error.to_string(),
653 })?
654 }
655 };
656
657 registry
658 .register(&mut listener, self.token, Interest::READABLE)
659 .map_err(ListenerError::SocketRegistration)?;
660
661 self.listener = Some(listener);
662 self.active = true;
663 Ok(self.token)
664 }
665
666 pub fn create_rustls_context(
667 config: &HttpsListenerConfig,
668 resolver: Arc<MutexCertificateResolver>,
669 ) -> Result<RustlsServerConfig, ListenerError> {
670 let cipher_names = if config.cipher_list.is_empty() {
671 DEFAULT_CIPHER_SUITES.to_vec()
672 } else {
673 config
674 .cipher_list
675 .iter()
676 .map(|s| s.as_str())
677 .collect::<Vec<_>>()
678 };
679
680 #[rustfmt::skip]
681 let ciphers = cipher_names
682 .into_iter()
683 .filter_map(|cipher| match cipher {
684 "TLS13_CHACHA20_POLY1305_SHA256" => Some(TLS13_CHACHA20_POLY1305_SHA256),
685 "TLS13_AES_256_GCM_SHA384" => Some(TLS13_AES_256_GCM_SHA384),
686 "TLS13_AES_128_GCM_SHA256" => Some(TLS13_AES_128_GCM_SHA256),
687 "TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256" => Some(TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256),
688 "TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256" => Some(TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256),
689 "TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384" => Some(TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384),
690 "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256" => Some(TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256),
691 "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" => Some(TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384),
692 "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256" => Some(TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256),
693 other_cipher => {
694 error!("unknown cipher: {:?}", other_cipher);
695 None
696 }
697 })
698 .collect::<Vec<_>>();
699
700 let versions = config
701 .versions
702 .iter()
703 .filter_map(|version| match TlsVersion::try_from(*version) {
704 Ok(TlsVersion::TlsV12) => Some(&rustls::version::TLS12),
705 Ok(TlsVersion::TlsV13) => Some(&rustls::version::TLS13),
706 Ok(other_version) => {
707 error!("unsupported TLS version {:?}", other_version);
708 None
709 }
710 Err(_) => {
711 error!("unsupported TLS version");
712 None
713 }
714 })
715 .collect::<Vec<_>>();
716
717 let provider = CryptoProvider {
718 cipher_suites: ciphers,
719 ..ring::default_provider()
720 };
721
722 let mut server_config = RustlsServerConfig::builder_with_provider(provider.into())
723 .with_protocol_versions(&versions[..])
724 .map_err(|err| ListenerError::BuildRustls(err.to_string()))?
725 .with_no_client_auth()
726 .with_cert_resolver(resolver);
727 server_config.send_tls13_tickets = config.send_tls13_tickets as usize;
728
729 let mut protocols = SERVER_PROTOS
730 .iter()
731 .map(|proto| proto.as_bytes().to_vec())
732 .collect::<Vec<_>>();
733 server_config.alpn_protocols.append(&mut protocols);
734
735 Ok(server_config)
736 }
737
738 pub fn add_https_front(&mut self, tls_front: HttpFrontend) -> Result<(), ListenerError> {
739 self.fronts
740 .add_http_front(&tls_front)
741 .map_err(ListenerError::AddFrontend)
742 }
743
744 pub fn remove_https_front(&mut self, tls_front: HttpFrontend) -> Result<(), ListenerError> {
745 debug!("removing tls_front {:?}", tls_front);
746 self.fronts
747 .remove_http_front(&tls_front)
748 .map_err(ListenerError::RemoveFrontend)
749 }
750
751 fn accept(&mut self) -> Result<MioTcpStream, AcceptError> {
752 if let Some(ref sock) = self.listener {
753 sock.accept()
754 .map_err(|e| match e.kind() {
755 ErrorKind::WouldBlock => AcceptError::WouldBlock,
756 _ => {
757 error!("accept() IO error: {:?}", e);
758 AcceptError::IoError
759 }
760 })
761 .map(|(sock, _)| sock)
762 } else {
763 error!("cannot accept connections, no listening socket available");
764 Err(AcceptError::IoError)
765 }
766 }
767}
768
769pub struct HttpsProxy {
770 listeners: HashMap<Token, Rc<RefCell<HttpsListener>>>,
771 clusters: HashMap<ClusterId, Cluster>,
772 backends: Rc<RefCell<BackendMap>>,
773 pool: Rc<RefCell<Pool>>,
774 registry: Registry,
775 sessions: Rc<RefCell<SessionManager>>,
776}
777
778impl HttpsProxy {
779 pub fn new(
780 registry: Registry,
781 sessions: Rc<RefCell<SessionManager>>,
782 pool: Rc<RefCell<Pool>>,
783 backends: Rc<RefCell<BackendMap>>,
784 ) -> HttpsProxy {
785 HttpsProxy {
786 listeners: HashMap::new(),
787 clusters: HashMap::new(),
788 backends,
789 pool,
790 registry,
791 sessions,
792 }
793 }
794
795 pub fn add_listener(
796 &mut self,
797 config: HttpsListenerConfig,
798 token: Token,
799 ) -> Result<Token, ProxyError> {
800 match self.listeners.entry(token) {
801 Entry::Vacant(entry) => {
802 let https_listener =
803 HttpsListener::try_new(config, token).map_err(ProxyError::AddListener)?;
804 entry.insert(Rc::new(RefCell::new(https_listener)));
805 Ok(token)
806 }
807 _ => Err(ProxyError::ListenerAlreadyPresent),
808 }
809 }
810
811 pub fn remove_listener(
812 &mut self,
813 remove: RemoveListener,
814 ) -> Result<Option<ResponseContent>, ProxyError> {
815 let len = self.listeners.len();
816
817 let remove_address = remove.address.into();
818 self.listeners
819 .retain(|_, listener| listener.borrow().address != remove_address);
820
821 if !self.listeners.len() < len {
822 info!("no HTTPS listener to remove at address {}", remove_address)
823 }
824 Ok(None)
825 }
826
827 pub fn soft_stop(&mut self) -> Result<(), ProxyError> {
828 let listeners: HashMap<_, _> = self.listeners.drain().collect();
829 let mut socket_errors = vec![];
830 for (_, l) in listeners.iter() {
831 if let Some(mut sock) = l.borrow_mut().listener.take() {
832 debug!("Deregistering socket {:?}", sock);
833 if let Err(e) = self.registry.deregister(&mut sock) {
834 let error = format!("socket {sock:?}: {e:?}");
835 socket_errors.push(error);
836 }
837 }
838 }
839
840 if !socket_errors.is_empty() {
841 return Err(ProxyError::SoftStop {
842 proxy_protocol: "HTTPS".to_string(),
843 error: format!("Error deregistering listen sockets: {:?}", socket_errors),
844 });
845 }
846
847 Ok(())
848 }
849
850 pub fn hard_stop(&mut self) -> Result<(), ProxyError> {
851 let mut listeners: HashMap<_, _> = self.listeners.drain().collect();
852 let mut socket_errors = vec![];
853 for (_, l) in listeners.drain() {
854 if let Some(mut sock) = l.borrow_mut().listener.take() {
855 debug!("Deregistering socket {:?}", sock);
856 if let Err(e) = self.registry.deregister(&mut sock) {
857 let error = format!("socket {sock:?}: {e:?}");
858 socket_errors.push(error);
859 }
860 }
861 }
862
863 if !socket_errors.is_empty() {
864 return Err(ProxyError::HardStop {
865 proxy_protocol: "HTTPS".to_string(),
866 error: format!("Error deregistering listen sockets: {:?}", socket_errors),
867 });
868 }
869
870 Ok(())
871 }
872
873 pub fn query_all_certificates(&mut self) -> Result<Option<ResponseContent>, ProxyError> {
874 let certificates = self
875 .listeners
876 .values()
877 .map(|listener| {
878 let owned = listener.borrow();
879 let resolver = unwrap_msg!(owned.resolver.0.lock());
880 let certificate_summaries = resolver
881 .domains
882 .to_hashmap()
883 .drain()
884 .map(|(k, fingerprint)| CertificateSummary {
885 domain: String::from_utf8(k).unwrap(),
886 fingerprint: fingerprint.to_string(),
887 })
888 .collect();
889
890 CertificatesByAddress {
891 address: owned.address.into(),
892 certificate_summaries,
893 }
894 })
895 .collect();
896
897 info!(
898 "got Certificates::All query, answering with {:?}",
899 certificates
900 );
901
902 Ok(Some(
903 ContentType::CertificatesByAddress(ListOfCertificatesByAddress { certificates }).into(),
904 ))
905 }
906
907 pub fn query_certificate_for_domain(
908 &mut self,
909 domain: String,
910 ) -> Result<Option<ResponseContent>, ProxyError> {
911 let certificates = self
912 .listeners
913 .values()
914 .map(|listener| {
915 let owned = listener.borrow();
916 let resolver = unwrap_msg!(owned.resolver.0.lock());
917 let mut certificate_summaries = vec![];
918
919 if let Some((k, fingerprint)) = resolver.domain_lookup(domain.as_bytes(), true) {
920 certificate_summaries.push(CertificateSummary {
921 domain: String::from_utf8(k.to_vec()).unwrap(),
922 fingerprint: fingerprint.to_string(),
923 });
924 }
925 CertificatesByAddress {
926 address: owned.address.into(),
927 certificate_summaries,
928 }
929 })
930 .collect();
931
932 info!(
933 "got Certificates::Domain({}) query, answering with {:?}",
934 domain, certificates
935 );
936
937 Ok(Some(
938 ContentType::CertificatesByAddress(ListOfCertificatesByAddress { certificates }).into(),
939 ))
940 }
941
942 pub fn activate_listener(
943 &mut self,
944 addr: &StdSocketAddr,
945 tcp_listener: Option<MioTcpListener>,
946 ) -> Result<Token, ProxyError> {
947 let listener = self
948 .listeners
949 .values()
950 .find(|listener| listener.borrow().address == *addr)
951 .ok_or(ProxyError::NoListenerFound(addr.to_owned()))?;
952
953 listener
954 .borrow_mut()
955 .activate(&self.registry, tcp_listener)
956 .map_err(|listener_error| ProxyError::ListenerActivation {
957 address: *addr,
958 listener_error,
959 })
960 }
961
962 pub fn give_back_listeners(&mut self) -> Vec<(StdSocketAddr, MioTcpListener)> {
963 self.listeners
964 .values()
965 .filter_map(|listener| {
966 let mut owned = listener.borrow_mut();
967 if let Some(listener) = owned.listener.take() {
968 return Some((owned.address, listener));
969 }
970
971 None
972 })
973 .collect()
974 }
975
976 pub fn give_back_listener(
977 &mut self,
978 address: StdSocketAddr,
979 ) -> Result<(Token, MioTcpListener), ProxyError> {
980 let listener = self
981 .listeners
982 .values()
983 .find(|listener| listener.borrow().address == address)
984 .ok_or(ProxyError::NoListenerFound(address))?;
985
986 let mut owned = listener.borrow_mut();
987
988 let taken_listener = owned
989 .listener
990 .take()
991 .ok_or(ProxyError::UnactivatedListener)?;
992
993 Ok((owned.token, taken_listener))
994 }
995
996 pub fn add_cluster(
997 &mut self,
998 mut cluster: Cluster,
999 ) -> Result<Option<ResponseContent>, ProxyError> {
1000 if let Some(answer_503) = cluster.answer_503.take() {
1001 for listener in self.listeners.values() {
1002 listener
1003 .borrow()
1004 .answers
1005 .borrow_mut()
1006 .add_custom_answer(&cluster.cluster_id, answer_503.clone())
1007 .map_err(|(status, error)| {
1008 ProxyError::AddCluster(ListenerError::TemplateParse(status, error))
1009 })?;
1010 }
1011 }
1012 self.clusters.insert(cluster.cluster_id.clone(), cluster);
1013 Ok(None)
1014 }
1015
1016 pub fn remove_cluster(
1017 &mut self,
1018 cluster_id: &str,
1019 ) -> Result<Option<ResponseContent>, ProxyError> {
1020 self.clusters.remove(cluster_id);
1021 for listener in self.listeners.values() {
1022 listener
1023 .borrow()
1024 .answers
1025 .borrow_mut()
1026 .remove_custom_answer(cluster_id);
1027 }
1028
1029 Ok(None)
1030 }
1031
1032 pub fn add_https_frontend(
1033 &mut self,
1034 front: RequestHttpFrontend,
1035 ) -> Result<Option<ResponseContent>, ProxyError> {
1036 let front = front.clone().to_frontend().map_err(|request_error| {
1037 ProxyError::WrongInputFrontend {
1038 front,
1039 error: request_error.to_string(),
1040 }
1041 })?;
1042
1043 let mut listener = self
1044 .listeners
1045 .values()
1046 .find(|l| l.borrow().address == front.address)
1047 .ok_or(ProxyError::NoListenerFound(front.address))?
1048 .borrow_mut();
1049
1050 listener.set_tags(front.hostname.to_owned(), front.tags.to_owned());
1051 listener
1052 .add_https_front(front)
1053 .map_err(ProxyError::AddFrontend)?;
1054 Ok(None)
1055 }
1056
1057 pub fn remove_https_frontend(
1058 &mut self,
1059 front: RequestHttpFrontend,
1060 ) -> Result<Option<ResponseContent>, ProxyError> {
1061 let front = front.clone().to_frontend().map_err(|request_error| {
1062 ProxyError::WrongInputFrontend {
1063 front,
1064 error: request_error.to_string(),
1065 }
1066 })?;
1067
1068 let mut listener = self
1069 .listeners
1070 .values()
1071 .find(|l| l.borrow().address == front.address)
1072 .ok_or(ProxyError::NoListenerFound(front.address))?
1073 .borrow_mut();
1074
1075 listener.set_tags(front.hostname.to_owned(), None);
1076 listener
1077 .remove_https_front(front)
1078 .map_err(ProxyError::RemoveFrontend)?;
1079 Ok(None)
1080 }
1081
1082 pub fn add_certificate(
1083 &mut self,
1084 add_certificate: AddCertificate,
1085 ) -> Result<Option<ResponseContent>, ProxyError> {
1086 let address = add_certificate.address.into();
1087
1088 let listener = self
1089 .listeners
1090 .values()
1091 .find(|l| l.borrow().address == address)
1092 .ok_or(ProxyError::NoListenerFound(address))?
1093 .borrow_mut();
1094
1095 let mut resolver = listener
1096 .resolver
1097 .0
1098 .lock()
1099 .map_err(|e| ProxyError::Lock(e.to_string()))?;
1100
1101 resolver
1102 .add_certificate(&add_certificate)
1103 .map_err(ProxyError::AddCertificate)?;
1104
1105 Ok(None)
1106 }
1107
1108 pub fn remove_certificate(
1110 &mut self,
1111 remove_certificate: RemoveCertificate,
1112 ) -> Result<Option<ResponseContent>, ProxyError> {
1113 let address = remove_certificate.address.into();
1114
1115 let fingerprint = Fingerprint(
1116 hex::decode(&remove_certificate.fingerprint)
1117 .map_err(ProxyError::WrongCertificateFingerprint)?,
1118 );
1119
1120 let listener = self
1121 .listeners
1122 .values()
1123 .find(|l| l.borrow().address == address)
1124 .ok_or(ProxyError::NoListenerFound(address))?
1125 .borrow_mut();
1126
1127 let mut resolver = listener
1128 .resolver
1129 .0
1130 .lock()
1131 .map_err(|e| ProxyError::Lock(e.to_string()))?;
1132
1133 resolver
1134 .remove_certificate(&fingerprint)
1135 .map_err(ProxyError::RemoveCertificate)?;
1136
1137 Ok(None)
1138 }
1139
1140 pub fn replace_certificate(
1142 &mut self,
1143 replace_certificate: ReplaceCertificate,
1144 ) -> Result<Option<ResponseContent>, ProxyError> {
1145 let address = replace_certificate.address.into();
1146
1147 let listener = self
1148 .listeners
1149 .values()
1150 .find(|l| l.borrow().address == address)
1151 .ok_or(ProxyError::NoListenerFound(address))?
1152 .borrow_mut();
1153
1154 let mut resolver = listener
1155 .resolver
1156 .0
1157 .lock()
1158 .map_err(|e| ProxyError::Lock(e.to_string()))?;
1159
1160 resolver
1161 .replace_certificate(&replace_certificate)
1162 .map_err(ProxyError::ReplaceCertificate)?;
1163
1164 Ok(None)
1165 }
1166}
1167
1168impl ProxyConfiguration for HttpsProxy {
1169 fn accept(&mut self, token: ListenToken) -> Result<MioTcpStream, AcceptError> {
1170 match self.listeners.get(&Token(token.0)) {
1171 Some(listener) => listener.borrow_mut().accept(),
1172 None => Err(AcceptError::IoError),
1173 }
1174 }
1175
1176 fn create_session(
1177 &mut self,
1178 mut frontend_sock: MioTcpStream,
1179 token: ListenToken,
1180 wait_time: Duration,
1181 proxy: Rc<RefCell<Self>>,
1182 ) -> Result<(), AcceptError> {
1183 let listener = self
1184 .listeners
1185 .get(&Token(token.0))
1186 .ok_or(AcceptError::IoError)?;
1187 if let Err(e) = frontend_sock.set_nodelay(true) {
1188 error!(
1189 "error setting nodelay on front socket({:?}): {:?}",
1190 frontend_sock, e
1191 );
1192 }
1193
1194 let owned = listener.borrow();
1195 let rustls_details = ServerConnection::new(owned.rustls_details.clone()).map_err(|e| {
1196 error!("failed to create server session: {:?}", e);
1197 AcceptError::IoError
1198 })?;
1199
1200 let mut session_manager = self.sessions.borrow_mut();
1201 let entry = session_manager.slab.vacant_entry();
1202 let session_token = Token(entry.key());
1203
1204 self.registry
1205 .register(
1206 &mut frontend_sock,
1207 session_token,
1208 Interest::READABLE | Interest::WRITABLE,
1209 )
1210 .map_err(|register_error| {
1211 error!(
1212 "error registering front socket({:?}): {:?}",
1213 frontend_sock, register_error
1214 );
1215 AcceptError::RegisterError
1216 })?;
1217
1218 let public_address: StdSocketAddr = match owned.config.public_address {
1219 Some(pub_addr) => pub_addr.into(),
1220 None => owned.config.address.into(),
1221 };
1222
1223 let session = Rc::new(RefCell::new(HttpsSession::new(
1224 owned.answers.clone(),
1225 Duration::from_secs(owned.config.back_timeout as u64),
1226 Duration::from_secs(owned.config.connect_timeout as u64),
1227 Duration::from_secs(owned.config.front_timeout as u64),
1228 Duration::from_secs(owned.config.request_timeout as u64),
1229 owned.config.expect_proxy,
1230 listener.clone(),
1231 Rc::downgrade(&self.pool),
1232 proxy,
1233 public_address,
1234 rustls_details,
1235 frontend_sock,
1236 owned.config.sticky_name.clone(),
1237 session_token,
1238 wait_time,
1239 )));
1240 entry.insert(session);
1241
1242 Ok(())
1243 }
1244
1245 fn notify(&mut self, request: WorkerRequest) -> WorkerResponse {
1246 let request_id = request.id.clone();
1247
1248 let request_type = match request.content.request_type {
1249 Some(t) => t,
1250 None => return WorkerResponse::error(request_id, "Empty request"),
1251 };
1252
1253 let content_result = match request_type {
1254 RequestType::AddCluster(cluster) => {
1255 debug!("{} add cluster {:?}", request_id, cluster);
1256 self.add_cluster(cluster)
1257 }
1258 RequestType::RemoveCluster(cluster_id) => {
1259 debug!("{} remove cluster {:?}", request_id, cluster_id);
1260 self.remove_cluster(&cluster_id)
1261 }
1262 RequestType::AddHttpsFrontend(front) => {
1263 debug!("{} add https front {:?}", request_id, front);
1264 self.add_https_frontend(front)
1265 }
1266 RequestType::RemoveHttpsFrontend(front) => {
1267 debug!("{} remove https front {:?}", request_id, front);
1268 self.remove_https_frontend(front)
1269 }
1270 RequestType::AddCertificate(add_certificate) => {
1271 debug!("{} add certificate: {:?}", request_id, add_certificate);
1272 self.add_certificate(add_certificate)
1273 }
1274 RequestType::RemoveCertificate(remove_certificate) => {
1275 debug!(
1276 "{} remove certificate: {:?}",
1277 request_id, remove_certificate
1278 );
1279 self.remove_certificate(remove_certificate)
1280 }
1281 RequestType::ReplaceCertificate(replace_certificate) => {
1282 debug!(
1283 "{} replace certificate: {:?}",
1284 request_id, replace_certificate
1285 );
1286 self.replace_certificate(replace_certificate)
1287 }
1288 RequestType::RemoveListener(remove) => {
1289 debug!("removing HTTPS listener at address {:?}", remove.address);
1290 self.remove_listener(remove)
1291 }
1292 RequestType::SoftStop(_) => {
1293 debug!("{} processing soft shutdown", request_id);
1294 match self.soft_stop() {
1295 Ok(_) => {
1296 info!("{} soft stop successful", request_id);
1297 return WorkerResponse::processing(request.id);
1298 }
1299 Err(e) => Err(e),
1300 }
1301 }
1302 RequestType::HardStop(_) => {
1303 debug!("{} processing hard shutdown", request_id);
1304 match self.hard_stop() {
1305 Ok(_) => {
1306 debug!("{} hard stop successful", request_id);
1307 return WorkerResponse::processing(request.id);
1308 }
1309 Err(e) => Err(e),
1310 }
1311 }
1312 RequestType::Status(_) => {
1313 debug!("{} status", request_id);
1314 Ok(None)
1315 }
1316 RequestType::QueryCertificatesFromWorkers(filters) => {
1317 if let Some(domain) = filters.domain {
1318 debug!("{} query certificate for domain {}", request_id, domain);
1319 self.query_certificate_for_domain(domain)
1320 } else {
1321 debug!("{} query all certificates", request_id);
1322 self.query_all_certificates()
1323 }
1324 }
1325 other_request => {
1326 debug!(
1327 "{} unsupported message for HTTPS proxy, ignoring {:?}",
1328 request.id, other_request
1329 );
1330 Err(ProxyError::UnsupportedMessage)
1331 }
1332 };
1333
1334 match content_result {
1335 Ok(content) => {
1336 debug!("{} successful", request_id);
1337 match content {
1338 Some(content) => WorkerResponse::ok_with_content(request_id, content),
1339 None => WorkerResponse::ok(request_id),
1340 }
1341 }
1342 Err(proxy_error) => {
1343 debug!("{} unsuccessful: {}", request_id, proxy_error);
1344 WorkerResponse::error(request_id, proxy_error)
1345 }
1346 }
1347 }
1348}
1349impl L7Proxy for HttpsProxy {
1350 fn kind(&self) -> ListenerType {
1351 ListenerType::Https
1352 }
1353
1354 fn register_socket(
1355 &self,
1356 socket: &mut MioTcpStream,
1357 token: Token,
1358 interest: Interest,
1359 ) -> Result<(), std::io::Error> {
1360 self.registry.register(socket, token, interest)
1361 }
1362
1363 fn deregister_socket(&self, tcp_stream: &mut MioTcpStream) -> Result<(), std::io::Error> {
1364 self.registry.deregister(tcp_stream)
1365 }
1366
1367 fn add_session(&self, session: Rc<RefCell<dyn ProxySession>>) -> Token {
1368 let mut session_manager = self.sessions.borrow_mut();
1369 let entry = session_manager.slab.vacant_entry();
1370 let token = Token(entry.key());
1371 let _entry = entry.insert(session);
1372 token
1373 }
1374
1375 fn remove_session(&self, token: Token) -> bool {
1376 self.sessions
1377 .borrow_mut()
1378 .slab
1379 .try_remove(token.0)
1380 .is_some()
1381 }
1382
1383 fn backends(&self) -> Rc<RefCell<BackendMap>> {
1384 self.backends.clone()
1385 }
1386
1387 fn clusters(&self) -> &HashMap<ClusterId, Cluster> {
1388 &self.clusters
1389 }
1390}
1391
1392fn rustls_version_str(version: ProtocolVersion) -> &'static str {
1394 match version {
1395 ProtocolVersion::SSLv2 => "tls.version.SSLv2",
1396 ProtocolVersion::SSLv3 => "tls.version.SSLv3",
1397 ProtocolVersion::TLSv1_0 => "tls.version.TLSv1_0",
1398 ProtocolVersion::TLSv1_1 => "tls.version.TLSv1_1",
1399 ProtocolVersion::TLSv1_2 => "tls.version.TLSv1_2",
1400 ProtocolVersion::TLSv1_3 => "tls.version.TLSv1_3",
1401 ProtocolVersion::DTLSv1_0 => "tls.version.DTLSv1_0",
1402 ProtocolVersion::DTLSv1_2 => "tls.version.DTLSv1_2",
1403 ProtocolVersion::DTLSv1_3 => "tls.version.DTLSv1_3",
1404 ProtocolVersion::Unknown(_) => "tls.version.Unknown",
1405 _ => "tls.version.unimplemented",
1406 }
1407}
1408
1409fn rustls_ciphersuite_str(cipher: SupportedCipherSuite) -> &'static str {
1411 match cipher.suite() {
1412 CipherSuite::TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256 => {
1413 "tls.cipher.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256"
1414 }
1415 CipherSuite::TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256 => {
1416 "tls.cipher.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256"
1417 }
1418 CipherSuite::TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 => {
1419 "tls.cipher.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256"
1420 }
1421 CipherSuite::TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 => {
1422 "tls.cipher.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"
1423 }
1424 CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256 => {
1425 "tls.cipher.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256"
1426 }
1427 CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384 => {
1428 "tls.cipher.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384"
1429 }
1430 CipherSuite::TLS13_CHACHA20_POLY1305_SHA256 => "tls.cipher.TLS13_CHACHA20_POLY1305_SHA256",
1431 CipherSuite::TLS13_AES_256_GCM_SHA384 => "tls.cipher.TLS13_AES_256_GCM_SHA384",
1432 CipherSuite::TLS13_AES_128_GCM_SHA256 => "tls.cipher.TLS13_AES_128_GCM_SHA256",
1433 _ => "tls.cipher.Unsupported",
1434 }
1435}
1436
1437pub mod testing {
1438 use crate::testing::*;
1439
1440 pub fn start_https_worker(
1442 config: HttpsListenerConfig,
1443 channel: ProxyChannel,
1444 max_buffers: usize,
1445 buffer_size: usize,
1446 ) -> anyhow::Result<()> {
1447 let address = config.address.into();
1448
1449 let ServerParts {
1450 event_loop,
1451 registry,
1452 sessions,
1453 pool,
1454 backends,
1455 client_scm_socket: _,
1456 server_scm_socket,
1457 server_config,
1458 } = prebuild_server(max_buffers, buffer_size, true)?;
1459
1460 let token = {
1461 let mut sessions = sessions.borrow_mut();
1462 let entry = sessions.slab.vacant_entry();
1463 let key = entry.key();
1464 let _ = entry.insert(Rc::new(RefCell::new(ListenSession {
1465 protocol: Protocol::HTTPSListen,
1466 })));
1467 Token(key)
1468 };
1469
1470 let mut proxy = HttpsProxy::new(registry, sessions.clone(), pool.clone(), backends.clone());
1471 proxy
1472 .add_listener(config, token)
1473 .with_context(|| "Failed at creating adding the listener")?;
1474 proxy
1475 .activate_listener(&address, None)
1476 .with_context(|| "Failed at creating activating the listener")?;
1477
1478 let mut server = Server::new(
1479 event_loop,
1480 channel,
1481 server_scm_socket,
1482 sessions,
1483 pool,
1484 backends,
1485 None,
1486 Some(proxy),
1487 None,
1488 server_config,
1489 None,
1490 false,
1491 )
1492 .with_context(|| "Failed at creating server")?;
1493
1494 debug!("starting event loop");
1495 server.run();
1496 debug!("ending event loop");
1497 Ok(())
1498 }
1499}
1500
1501#[cfg(test)]
1502mod tests {
1503 use super::*;
1504
1505 use std::sync::Arc;
1506
1507 use sozu_command::{
1508 config::ListenerBuilder,
1509 proto::command::{CustomHttpAnswers, SocketAddress},
1510 };
1511
1512 use crate::router::{pattern_trie::TrieNode, MethodRule, PathRule, Route, Router};
1513
1514 #[test]
1532 fn frontend_from_request_test() {
1533 let cluster_id1 = "cluster_1".to_owned();
1534 let cluster_id2 = "cluster_2".to_owned();
1535 let cluster_id3 = "cluster_3".to_owned();
1536 let uri1 = "/".to_owned();
1537 let uri2 = "/yolo".to_owned();
1538 let uri3 = "/yolo/swag".to_owned();
1539
1540 let mut fronts = Router::new();
1541 assert!(fronts.add_tree_rule(
1542 "lolcatho.st".as_bytes(),
1543 &PathRule::Prefix(uri1),
1544 &MethodRule::new(None),
1545 &Route::ClusterId(cluster_id1.clone())
1546 ));
1547 assert!(fronts.add_tree_rule(
1548 "lolcatho.st".as_bytes(),
1549 &PathRule::Prefix(uri2),
1550 &MethodRule::new(None),
1551 &Route::ClusterId(cluster_id2)
1552 ));
1553 assert!(fronts.add_tree_rule(
1554 "lolcatho.st".as_bytes(),
1555 &PathRule::Prefix(uri3),
1556 &MethodRule::new(None),
1557 &Route::ClusterId(cluster_id3)
1558 ));
1559 assert!(fronts.add_tree_rule(
1560 "other.domain".as_bytes(),
1561 &PathRule::Prefix("test".to_string()),
1562 &MethodRule::new(None),
1563 &Route::ClusterId(cluster_id1)
1564 ));
1565
1566 let address = SocketAddress::new_v4(127, 0, 0, 1, 1032);
1567 let resolver = Arc::new(MutexCertificateResolver::default());
1568
1569 let crypto_provider = Arc::new(ring::default_provider());
1570
1571 let server_config = RustlsServerConfig::builder_with_provider(crypto_provider)
1572 .with_protocol_versions(&[&rustls::version::TLS12, &rustls::version::TLS13])
1573 .expect("could not create rustls config server")
1574 .with_no_client_auth()
1575 .with_cert_resolver(resolver.clone());
1576
1577 let rustls_details = Arc::new(server_config);
1578
1579 let default_config = ListenerBuilder::new_https(address.clone())
1580 .to_tls(None)
1581 .expect("Could not create default HTTPS listener config");
1582
1583 println!("it doesn't even matter");
1584
1585 let listener = HttpsListener {
1586 listener: None,
1587 address: address.into(),
1588 fronts,
1589 rustls_details,
1590 resolver,
1591 answers: Rc::new(RefCell::new(
1592 HttpAnswers::new(&Some(CustomHttpAnswers::default())).unwrap(),
1593 )),
1594 config: default_config,
1595 token: Token(0),
1596 active: true,
1597 tags: BTreeMap::new(),
1598 };
1599
1600 println!("TEST {}", line!());
1601 let frontend1 = listener.frontend_from_request("lolcatho.st", "/", &Method::Get);
1602 assert_eq!(
1603 frontend1.expect("should find a frontend"),
1604 Route::ClusterId("cluster_1".to_string())
1605 );
1606 println!("TEST {}", line!());
1607 let frontend2 = listener.frontend_from_request("lolcatho.st", "/test", &Method::Get);
1608 assert_eq!(
1609 frontend2.expect("should find a frontend"),
1610 Route::ClusterId("cluster_1".to_string())
1611 );
1612 println!("TEST {}", line!());
1613 let frontend3 = listener.frontend_from_request("lolcatho.st", "/yolo/test", &Method::Get);
1614 assert_eq!(
1615 frontend3.expect("should find a frontend"),
1616 Route::ClusterId("cluster_2".to_string())
1617 );
1618 println!("TEST {}", line!());
1619 let frontend4 = listener.frontend_from_request("lolcatho.st", "/yolo/swag", &Method::Get);
1620 assert_eq!(
1621 frontend4.expect("should find a frontend"),
1622 Route::ClusterId("cluster_3".to_string())
1623 );
1624 println!("TEST {}", line!());
1625 let frontend5 = listener.frontend_from_request("domain", "/", &Method::Get);
1626 assert!(frontend5.is_err());
1627 }
1629
1630 #[test]
1631 fn wildcard_certificate_names() {
1632 let mut trie = TrieNode::root();
1633
1634 trie.domain_insert("*.services.clever-cloud.com".as_bytes().to_vec(), 1u8);
1635 trie.domain_insert("*.clever-cloud.com".as_bytes().to_vec(), 2u8);
1636 trie.domain_insert("services.clever-cloud.com".as_bytes().to_vec(), 0u8);
1637 trie.domain_insert(
1638 "abprefix.services.clever-cloud.com".as_bytes().to_vec(),
1639 3u8,
1640 );
1641 trie.domain_insert(
1642 "cdprefix.services.clever-cloud.com".as_bytes().to_vec(),
1643 4u8,
1644 );
1645
1646 let res = trie.domain_lookup(b"test.services.clever-cloud.com", true);
1647 println!("query result: {res:?}");
1648
1649 assert_eq!(
1650 trie.domain_lookup(b"pgstudio.services.clever-cloud.com", true),
1651 Some(&("*.services.clever-cloud.com".as_bytes().to_vec(), 1u8))
1652 );
1653 assert_eq!(
1654 trie.domain_lookup(b"test-prefix.services.clever-cloud.com", true),
1655 Some(&("*.services.clever-cloud.com".as_bytes().to_vec(), 1u8))
1656 );
1657 }
1658
1659 #[test]
1660 fn wildcard_with_subdomains() {
1661 let mut trie = TrieNode::root();
1662
1663 trie.domain_insert("*.test.example.com".as_bytes().to_vec(), 1u8);
1664 trie.domain_insert("hello.sub.test.example.com".as_bytes().to_vec(), 2u8);
1665
1666 let res = trie.domain_lookup(b"sub.test.example.com", true);
1667 println!("query result: {res:?}");
1668
1669 assert_eq!(
1670 trie.domain_lookup(b"sub.test.example.com", true),
1671 Some(&("*.test.example.com".as_bytes().to_vec(), 1u8))
1672 );
1673 assert_eq!(
1674 trie.domain_lookup(b"hello.sub.test.example.com", true),
1675 Some(&("hello.sub.test.example.com".as_bytes().to_vec(), 2u8))
1676 );
1677
1678 let mut trie = TrieNode::root();
1680
1681 trie.domain_insert("hello.sub.test.example.com".as_bytes().to_vec(), 2u8);
1682 trie.domain_insert("*.test.example.com".as_bytes().to_vec(), 1u8);
1683
1684 let res = trie.domain_lookup(b"sub.test.example.com", true);
1685 println!("query result: {res:?}");
1686
1687 assert_eq!(
1688 trie.domain_lookup(b"sub.test.example.com", true),
1689 Some(&("*.test.example.com".as_bytes().to_vec(), 1u8))
1690 );
1691 assert_eq!(
1692 trie.domain_lookup(b"hello.sub.test.example.com", true),
1693 Some(&("hello.sub.test.example.com".as_bytes().to_vec(), 2u8))
1694 );
1695 }
1696}