HAProxy protocol support (closes #36)

tests/src/jmap/mod.rs

@@ -75,6 +75,7 @@ bind = ["127.0.0.1:8899"]
 url = "https://127.0.0.1:8899"
 protocol = "jmap"
 max-connections = 81920
+tls.implicit = true
 
 [server.listener.lmtp-debug]
 bind = ['127.0.0.1:11200']

tests/src/jmap/push_subscription.rs

@@ -46,7 +46,7 @@ use jmap_client::{mailbox::Role, push_subscription::Keys};
 use jmap_proto::types::{id::Id, type_state::DataType};
 use store::ahash::AHashSet;
 
-use tokio::{net::TcpStream, sync::mpsc};
+use tokio::sync::mpsc;
 use utils::listener::SessionData;
 
 use crate::{
@@ -286,10 +286,13 @@ struct PushVerification {
 }
 
 impl utils::listener::SessionManager for SessionManager {
-    fn spawn(&self, session: SessionData<TcpStream>) {
-        let push = self.inner.clone();
-
-        tokio::spawn(async move {
+    #[allow(clippy::manual_async_fn)]
+    fn handle<T: utils::listener::SessionStream>(
+        self,
+        session: SessionData<T>,
+    ) -> impl std::future::Future<Output = ()> + Send {
+        async move {
+            let push = self.inner;
             let _ = http1::Builder::new()
                 .keep_alive(false)
                 .serve_connection(
@@ -346,10 +349,13 @@ impl utils::listener::SessionManager for SessionManager {
                     }),
                 )
                 .await;
-        });
+        }
     }
 
-    fn shutdown(&self) {}
+    #[allow(clippy::manual_async_fn)]
+    fn shutdown(&self) -> impl std::future::Future<Output = ()> + Send {
+        async {}
+    }
 }
 
 async fn expect_push(event_rx: &mut mpsc::Receiver<PushMessage>) -> PushMessage {

tests/src/smtp/config.rs

@@ -38,7 +38,9 @@ use store::{
 use tokio::net::TcpSocket;
 
 use utils::{
-    config::{Config, DynValue, KeyLookup, Listener, Rate, Server, ServerProtocol},
+    config::{
+        ipmask::IpAddrMask, Config, DynValue, KeyLookup, Listener, Rate, Server, ServerProtocol,
+    },
     listener::TcpAcceptor,
 };
 
@@ -47,8 +49,8 @@ use ahash::AHashMap;
 use smtp::{
     config::{
         condition::ConfigCondition, if_block::ConfigIf, throttle::ConfigThrottle, Condition,
-        ConditionMatch, Conditions, ConfigContext, EnvelopeKey, IfBlock, IfThen, IpAddrMask,
-        StringMatch, Throttle, THROTTLE_AUTH_AS, THROTTLE_REMOTE_IP, THROTTLE_SENDER_DOMAIN,
+        ConditionMatch, Conditions, ConfigContext, EnvelopeKey, IfBlock, IfThen, StringMatch,
+        Throttle, THROTTLE_AUTH_AS, THROTTLE_REMOTE_IP, THROTTLE_SENDER_DOMAIN,
     },
     core::Lookup,
 };
@@ -455,6 +457,7 @@ fn parse_servers() {
             acceptor: TcpAcceptor::Plain,
             tls_implicit: false,
             max_connections: 8192,
+            proxy_networks: vec![],
         },
         Server {
             id: "smtps".to_string(),
@@ -483,6 +486,7 @@ fn parse_servers() {
             acceptor: TcpAcceptor::Plain,
             tls_implicit: true,
             max_connections: 1024,
+            proxy_networks: vec![],
         },
         Server {
             id: "submission".to_string(),
@@ -501,6 +505,7 @@ fn parse_servers() {
             acceptor: TcpAcceptor::Plain,
             tls_implicit: true,
             max_connections: 8192,
+            proxy_networks: vec![],
         },
     ];
 

tests/src/smtp/inbound/limits.rs

@@ -65,7 +65,7 @@ async fn limits() {
     // Exceed transfer quota
     session.eval_session_params().await;
     session.write_rx("MAIL FROM:<this_is_a_long@command_over_10_chars.com>\r\n");
-    session.handle_conn_().await;
+    session.handle_conn().await;
     session.response().assert_code("451 4.7.28");
 
     // Loitering
@@ -74,7 +74,7 @@ async fn limits() {
     session.eval_session_params().await;
     tokio::time::sleep(Duration::from_millis(600)).await;
     session.write_rx("MAIL FROM:<this_is_a_long@command_over_10_chars.com>\r\n");
-    session.handle_conn_().await;
+    session.handle_conn().await;
     session.response().assert_code("453 4.3.2");
 
     // Timeout
@@ -82,6 +82,6 @@ async fn limits() {
     session.data.valid_until = Instant::now();
     session.eval_session_params().await;
     session.write_rx("MAIL FROM:<this_is_a_long@command_over_10_chars.com>\r\n");
-    session.handle_conn_().await;
+    session.handle_conn().await;
     session.response().assert_code("221 2.0.0");
 }

tests/src/smtp/outbound/mod.rs

@@ -56,6 +56,7 @@ tls.implicit = true
 [server.listener.management-debug]
 bind = ['127.0.0.1:9980']
 protocol = 'http'
+tls.implicit = true
 
 [server.socket]
 reuse-addr = true

tests/src/smtp/session.rs

@@ -21,20 +21,19 @@
  * for more details.
 */
 
-use std::{path::PathBuf, sync::Arc};
+use std::{borrow::Cow, path::PathBuf, sync::Arc};
 
+use rustls::{server::ResolvesServerCert, ServerConfig};
 use tokio::{
     io::{AsyncRead, AsyncWrite},
     sync::watch,
 };
 
-use smtp::{
-    core::{Session, SessionAddress, SessionData, SessionParameters, State, SMTP},
-    inbound::IsTls,
-};
+use smtp::core::{Session, SessionAddress, SessionData, SessionParameters, State, SMTP};
+use tokio_rustls::TlsAcceptor;
 use utils::{
     config::ServerProtocol,
-    listener::{limiter::ConcurrencyLimiter, ServerInstance, TcpAcceptor},
+    listener::{limiter::ConcurrencyLimiter, ServerInstance, SessionStream, TcpAcceptor},
 };
 
 use super::TestConfig;
@@ -86,15 +85,13 @@ impl AsyncWrite for DummyIo {
     }
 }
 
-impl IsTls for DummyIo {
+impl SessionStream for DummyIo {
     fn is_tls(&self) -> bool {
         self.tls
     }
 
-    fn write_tls_header(&self, _headers: &mut Vec<u8>) {}
-
-    fn tls_version_and_cipher(&self) -> (&'static str, &'static str) {
-        ("", "")
+    fn tls_version_and_cipher(&self) -> (Cow<'static, str>, Cow<'static, str>) {
+        ("".into(), "".into())
     }
 }
 
@@ -368,14 +365,27 @@ impl TestServerInstance for ServerInstance {
             hostname: "mx.example.org".to_string(),
             protocol: ServerProtocol::Smtp,
             data: "220 mx.example.org at your service.\r\n".to_string(),
-            acceptor: TcpAcceptor::Plain,
-            is_tls_implicit: false,
+            acceptor: TcpAcceptor::Tls(TlsAcceptor::from(Arc::new(
+                ServerConfig::builder()
+                    .with_no_client_auth()
+                    .with_cert_resolver(Arc::new(DummyCertResolver)),
+            ))),
             limiter: ConcurrencyLimiter::new(100),
             shutdown_rx,
+            proxy_networks: vec![],
         }
     }
 }
 
+#[derive(Debug)]
+pub struct DummyCertResolver;
+
+impl ResolvesServerCert for DummyCertResolver {
+    fn resolve(&self, _: rustls::server::ClientHello) -> Option<Arc<rustls::sign::CertifiedKey>> {
+        None
+    }
+}
+
 impl TestConfig for ServerInstance {
     fn test() -> Self {
         Self::test_with_shutdown(watch::channel(false).1)