From 3a800aff7a491a7c4516a0c19546d69ea6e9d8c1 Mon Sep 17 00:00:00 2001 From: mdecimus Date: Sat, 6 Jan 2024 20:02:31 +0100 Subject: [PATCH] HAProxy protocol support (closes #36) --- CHANGELOG.md | 2 + Cargo.lock | 29 ++- README.md | 1 + crates/cli/Cargo.toml | 2 +- crates/imap/Cargo.toml | 2 +- crates/imap/src/core/client.rs | 26 ++- crates/imap/src/core/mailbox.rs | 9 +- crates/imap/src/core/message.rs | 3 +- crates/imap/src/core/mod.rs | 68 ++++-- crates/imap/src/core/session.rs | 187 +++++++-------- crates/imap/src/core/writer.rs | 33 +-- crates/imap/src/op/acl.rs | 7 +- crates/imap/src/op/append.rs | 6 +- crates/imap/src/op/authenticate.rs | 4 +- crates/imap/src/op/capability.rs | 4 +- crates/imap/src/op/close.rs | 4 +- crates/imap/src/op/copy_move.rs | 6 +- crates/imap/src/op/create.rs | 6 +- crates/imap/src/op/delete.rs | 6 +- crates/imap/src/op/enable.rs | 4 +- crates/imap/src/op/expunge.rs | 6 +- crates/imap/src/op/fetch.rs | 6 +- crates/imap/src/op/idle.rs | 7 +- crates/imap/src/op/list.rs | 6 +- crates/imap/src/op/login.rs | 4 +- crates/imap/src/op/logout.rs | 4 +- crates/imap/src/op/namespace.rs | 4 +- crates/imap/src/op/noop.rs | 4 +- crates/imap/src/op/rename.rs | 6 +- crates/imap/src/op/search.rs | 7 +- crates/imap/src/op/select.rs | 4 +- crates/imap/src/op/status.rs | 6 +- crates/imap/src/op/store.rs | 6 +- crates/imap/src/op/subscribe.rs | 6 +- crates/imap/src/op/thread.rs | 6 +- crates/install/Cargo.toml | 2 +- crates/jmap/Cargo.toml | 2 +- crates/jmap/src/api/http.rs | 68 ++---- crates/jmap/src/sieve/ingest.rs | 3 +- crates/jmap/src/submission/set.rs | 7 +- crates/main/Cargo.toml | 2 +- crates/managesieve/Cargo.toml | 2 +- crates/managesieve/src/core/client.rs | 5 +- crates/managesieve/src/core/mod.rs | 22 +- crates/managesieve/src/core/session.rs | 89 +++---- crates/managesieve/src/op/authenticate.rs | 6 +- crates/managesieve/src/op/capability.rs | 6 +- crates/nlp/Cargo.toml | 2 +- crates/smtp/Cargo.toml | 2 +- crates/smtp/src/config/condition.rs | 54 +---- crates/smtp/src/config/mod.rs | 8 +- crates/smtp/src/core/if_block.rs | 56 +---- crates/smtp/src/core/management.rs | 48 ++-- crates/smtp/src/core/mod.rs | 60 +---- crates/smtp/src/inbound/data.rs | 19 +- crates/smtp/src/inbound/ehlo.rs | 6 +- crates/smtp/src/inbound/mail.rs | 6 +- crates/smtp/src/inbound/milter/message.rs | 9 +- crates/smtp/src/inbound/mod.rs | 66 ------ crates/smtp/src/inbound/rcpt.rs | 6 +- crates/smtp/src/inbound/session.rs | 25 +- crates/smtp/src/inbound/spawn.rs | 100 ++++---- crates/smtp/src/scripts/exec.rs | 13 +- crates/utils/Cargo.toml | 3 +- crates/utils/src/config/ipmask.rs | 131 +++++++++++ crates/utils/src/config/listener.rs | 12 +- crates/utils/src/config/mod.rs | 4 +- crates/utils/src/listener/listen.rs | 232 ++++++++++++------- crates/utils/src/listener/mod.rs | 67 +++++- crates/utils/src/listener/stream.rs | 160 +++++++++++++ crates/utils/src/listener/tls.rs | 4 +- resources/config.zip | Bin 171392 -> 171446 bytes resources/config/common/server.toml | 1 + resources/config/jmap/listener.toml | 3 + tests/resources/imap/000.imap | 28 ++- tests/resources/imap/002.imap | 4 +- tests/resources/imap/004.imap | 56 ++++- tests/resources/imap/006.imap | 8 +- tests/resources/imap/007.imap | 52 +++-- tests/resources/imap/009.imap | 4 +- tests/resources/scripts/create_test_users.sh | 4 +- tests/src/jmap/mod.rs | 1 + tests/src/jmap/push_subscription.rs | 20 +- tests/src/smtp/config.rs | 11 +- tests/src/smtp/inbound/limits.rs | 6 +- tests/src/smtp/outbound/mod.rs | 1 + tests/src/smtp/session.rs | 36 +-- 87 files changed, 1130 insertions(+), 903 deletions(-) create mode 100644 crates/utils/src/config/ipmask.rs create mode 100644 crates/utils/src/listener/stream.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 6c83363c..c88824a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ All notable changes to this project will be documented in this file. This projec ## Added - ACME support for automatic TLS certificate generation and renewal. +- TLS certificate hot-reloading. +- HAProxy protocol support. ### Changed diff --git a/Cargo.lock b/Cargo.lock index 08b043bd..9871da99 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2551,7 +2551,7 @@ checksum = "029d73f573d8e8d63e6d5020011d3255b28c3ba85d6cf870a07184ed23de9284" [[package]] name = "imap" -version = "0.5.1" +version = "0.5.2" dependencies = [ "ahash 0.8.7", "dashmap", @@ -2728,7 +2728,7 @@ dependencies = [ [[package]] name = "jmap" -version = "0.5.1" +version = "0.5.2" dependencies = [ "aes", "aes-gcm", @@ -3138,7 +3138,7 @@ dependencies = [ [[package]] name = "mail-server" -version = "0.5.1" +version = "0.5.2" dependencies = [ "directory", "imap", @@ -3155,7 +3155,7 @@ dependencies = [ [[package]] name = "managesieve" -version = "0.5.1" +version = "0.5.2" dependencies = [ "ahash 0.8.7", "bincode", @@ -3422,7 +3422,7 @@ dependencies = [ [[package]] name = "nlp" -version = "0.5.1" +version = "0.5.2" dependencies = [ "ahash 0.8.7", "bincode", @@ -4177,6 +4177,16 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "proxy-header" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e220ac9305411757d06712209b7c2d1d35c3a1a577301e87855f6219585ecb" +dependencies = [ + "pin-project-lite", + "tokio", +] + [[package]] name = "ptr_meta" version = "0.1.4" @@ -5375,7 +5385,7 @@ checksum = "4dccd0940a2dcdf68d092b8cbab7dc0ad8fa938bf95787e1b916b0e3d0e8e970" [[package]] name = "smtp" -version = "0.5.1" +version = "0.5.2" dependencies = [ "ahash 0.8.7", "bincode", @@ -5497,7 +5507,7 @@ dependencies = [ [[package]] name = "stalwart-cli" -version = "0.5.1" +version = "0.5.2" dependencies = [ "clap", "console", @@ -5521,7 +5531,7 @@ dependencies = [ [[package]] name = "stalwart-install" -version = "0.5.1" +version = "0.5.2" dependencies = [ "base64 0.21.5", "clap", @@ -6404,7 +6414,7 @@ checksum = "711b9620af191e0cdc7468a8d14e709c3dcdb115b36f838e601583af800a370a" [[package]] name = "utils" -version = "0.5.1" +version = "0.5.2" dependencies = [ "ahash 0.8.7", "arc-swap", @@ -6421,6 +6431,7 @@ dependencies = [ "parking_lot", "pem", "privdrop", + "proxy-header", "rand", "rcgen", "reqwest", diff --git a/README.md b/README.md index 57be7a33..838b3d2a 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,7 @@ Key features: - Integration with **OpenTelemetry** to enable monitoring, tracing, and performance analysis. - **Secure and robust**: - Encryption at rest with **S/MIME** or **OpenPGP**. + - Automatic TLS certificate provisioning with [ACME](https://datatracker.ietf.org/doc/html/rfc8555). - OAuth 2.0 [authorization code](https://www.rfc-editor.org/rfc/rfc8628) and [device authorization](https://www.rfc-editor.org/rfc/rfc8628) flows. - Access Control Lists (ACLs). - Rate limiting. diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index 12963d47..581ad8ad 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -5,7 +5,7 @@ authors = ["Stalwart Labs Ltd. "] license = "AGPL-3.0-only" repository = "https://github.com/stalwartlabs/cli" homepage = "https://github.com/stalwartlabs/cli" -version = "0.5.1" +version = "0.5.2" edition = "2021" readme = "README.md" resolver = "2" diff --git a/crates/imap/Cargo.toml b/crates/imap/Cargo.toml index f13cdb9e..93c81abe 100644 --- a/crates/imap/Cargo.toml +++ b/crates/imap/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "imap" -version = "0.5.1" +version = "0.5.2" edition = "2021" resolver = "2" diff --git a/crates/imap/src/core/client.rs b/crates/imap/src/core/client.rs index 9fbeb4ef..b5ff97e0 100644 --- a/crates/imap/src/core/client.rs +++ b/crates/imap/src/core/client.rs @@ -29,12 +29,14 @@ use imap_proto::{ }; use jmap::auth::rate_limit::AuthenticatedLimiter; use parking_lot::Mutex; -use tokio::io::AsyncRead; -use utils::listener::limiter::{ConcurrencyLimiter, RateLimiter}; +use utils::listener::{ + limiter::{ConcurrencyLimiter, RateLimiter}, + SessionStream, +}; use super::{SelectedMailbox, Session, SessionData, State, IMAP}; -impl Session { +impl Session { pub async fn ingest(&mut self, bytes: &[u8]) -> crate::Result { /*for line in String::from_utf8_lossy(bytes).split("\r\n") { let c = println!("{}", line); @@ -221,7 +223,7 @@ pub fn group_requests( grouped_requests } -impl Session { +impl Session { fn is_allowed(&self, request: Request) -> Result, StatusResponse> { let state = &self.state; // Rate limit request @@ -243,7 +245,11 @@ impl Session { Command::Capability | Command::Noop | Command::Logout | Command::Id => Ok(request), Command::StartTls => { if !self.is_tls { - Ok(request) + if self.instance.acceptor.is_tls() { + Ok(request) + } else { + Err(StatusResponse::no("TLS is not available.").with_tag(request.tag)) + } } else { Err(StatusResponse::no("Already in TLS mode.").with_tag(request.tag)) } @@ -330,7 +336,7 @@ impl Session { } } -impl State { +impl State { pub fn auth_failures(&self) -> u32 { match self { State::NotAuthenticated { auth_failures, .. } => *auth_failures, @@ -338,7 +344,7 @@ impl State { } } - pub fn session_data(&self) -> Arc { + pub fn session_data(&self) -> Arc> { match self { State::Authenticated { data } => data.clone(), State::Selected { data, .. } => data.clone(), @@ -346,14 +352,14 @@ impl State { } } - pub fn mailbox_state(&self) -> (Arc, Arc) { + pub fn mailbox_state(&self) -> (Arc>, Arc) { match self { State::Selected { data, mailbox, .. } => (data.clone(), mailbox.clone()), _ => unreachable!(), } } - pub fn session_mailbox_state(&self) -> (Arc, Option>) { + pub fn session_mailbox_state(&self) -> (Arc>, Option>) { match self { State::Authenticated { data } => (data.clone(), None), State::Selected { data, mailbox, .. } => (data.clone(), mailbox.clone().into()), @@ -361,7 +367,7 @@ impl State { } } - pub fn select_data(&self) -> (Arc, Arc) { + pub fn select_data(&self) -> (Arc>, Arc) { match self { State::Selected { data, mailbox } => (data.clone(), mailbox.clone()), _ => unreachable!(), diff --git a/crates/imap/src/core/mailbox.rs b/crates/imap/src/core/mailbox.rs index 730d6359..0b18166e 100644 --- a/crates/imap/src/core/mailbox.rs +++ b/crates/imap/src/core/mailbox.rs @@ -13,19 +13,18 @@ use jmap_proto::{ }; use parking_lot::Mutex; use store::query::log::{Change, Query}; -use tokio::io::AsyncRead; -use utils::listener::limiter::InFlight; +use utils::listener::{limiter::InFlight, SessionStream}; use super::{Account, Mailbox, MailboxId, MailboxSync, Session, SessionData}; -impl SessionData { - pub async fn new( +impl SessionData { + pub async fn new( session: &Session, access_token: &AccessToken, in_flight: InFlight, ) -> crate::Result { let mut session = SessionData { - writer: session.writer.clone(), + stream_tx: session.stream_tx.clone(), jmap: session.jmap.clone(), imap: session.imap.clone(), account_id: access_token.primary_id(), diff --git a/crates/imap/src/core/message.rs b/crates/imap/src/core/message.rs index 1aed102c..ecaae59e 100644 --- a/crates/imap/src/core/message.rs +++ b/crates/imap/src/core/message.rs @@ -37,6 +37,7 @@ use store::{ roaring::RoaringBitmap, write::{assert::HashedValue, BatchBuilder, F_VALUE}, }; +use utils::listener::SessionStream; use crate::core::ImapId; @@ -44,7 +45,7 @@ use super::{Mailbox, MailboxId, MailboxState, NextMailboxState, SelectedMailbox, pub(crate) const MAX_RETRIES: usize = 10; -impl SessionData { +impl SessionData { pub async fn fetch_messages(&self, mailbox: &MailboxId) -> crate::op::Result { // Obtain message ids let message_ids = self diff --git a/crates/imap/src/core/mod.rs b/crates/imap/src/core/mod.rs index 41706642..dea1f3d5 100644 --- a/crates/imap/src/core/mod.rs +++ b/crates/imap/src/core/mod.rs @@ -41,22 +41,20 @@ use jmap::{ }, JMAP, }; -use parking_lot::Mutex; use store::roaring::RoaringBitmap; use tokio::{ - io::{AsyncRead, ReadHalf}, - sync::{mpsc, watch}, + io::{ReadHalf, WriteHalf}, + sync::watch, }; use utils::{ config::Rate, - listener::{limiter::InFlight, ServerInstance}, + listener::{limiter::InFlight, ServerInstance, SessionStream}, }; pub mod client; pub mod mailbox; pub mod message; pub mod session; -pub mod writer; #[derive(Clone)] pub struct ImapSessionManager { @@ -84,35 +82,35 @@ pub struct IMAP { pub greeting_plain: Vec, pub greeting_tls: Vec, - pub rate_limiter: DashMap>>, + pub rate_limiter: DashMap>>, pub rate_requests: Rate, pub rate_concurrent: u64, } -pub struct Session { +pub struct Session { pub jmap: Arc, pub imap: Arc, pub instance: Arc, pub receiver: Receiver, pub version: ProtocolVersion, - pub state: State, + pub state: State, pub is_tls: bool, pub is_condstore: bool, pub is_qresync: bool, - pub writer: mpsc::Sender, pub stream_rx: ReadHalf, + pub stream_tx: Arc>>, pub in_flight: InFlight, pub remote_addr: RemoteAddress, pub span: tracing::Span, } -pub struct SessionData { +pub struct SessionData { pub account_id: u32, pub jmap: Arc, pub imap: Arc, pub span: tracing::Span, pub mailboxes: parking_lot::Mutex>, - pub writer: mpsc::Sender, + pub stream_tx: Arc>>, pub state: AtomicU32, pub in_flight: InFlight, } @@ -196,20 +194,44 @@ pub struct ImapId { pub seqnum: u32, } -pub enum State { +pub enum State { NotAuthenticated { auth_failures: u32, }, Authenticated { - data: Arc, + data: Arc>, }, Selected { - data: Arc, + data: Arc>, mailbox: Arc, }, } -impl SessionData { +impl State { + pub fn try_replace_stream_tx( + self, + new_stream: Arc>>, + ) -> Option> { + match self { + State::NotAuthenticated { auth_failures } => { + State::NotAuthenticated { auth_failures }.into() + } + State::Authenticated { data } => { + Arc::try_unwrap(data).ok().map(|data| State::Authenticated { + data: Arc::new(data.replace_stream_tx(new_stream)), + }) + } + State::Selected { data, mailbox } => { + Arc::try_unwrap(data).ok().map(|data| State::Selected { + data: Arc::new(data.replace_stream_tx(new_stream)), + mailbox, + }) + } + } + } +} + +impl SessionData { pub async fn get_access_token(&self) -> crate::op::Result> { self.jmap .get_cached_access_token(self.account_id) @@ -219,4 +241,20 @@ impl SessionData { .with_code(ResponseCode::ContactAdmin) }) } + + pub fn replace_stream_tx( + self, + new_stream: Arc>>, + ) -> SessionData { + SessionData { + account_id: self.account_id, + jmap: self.jmap, + imap: self.imap, + span: self.span, + mailboxes: self.mailboxes, + stream_tx: new_stream, + state: self.state, + in_flight: self.in_flight, + } + } } diff --git a/crates/imap/src/core/session.rs b/crates/imap/src/core/session.rs index 7c8930c4..765a9591 100644 --- a/crates/imap/src/core/session.rs +++ b/crates/imap/src/core/session.rs @@ -21,40 +21,41 @@ * for more details. */ +use std::{borrow::Cow, sync::Arc}; + use imap_proto::{protocol::ProtocolVersion, receiver::Receiver}; use jmap::auth::rate_limit::RemoteAddress; -use tokio::{ - io::{AsyncRead, AsyncReadExt, AsyncWriteExt}, - net::TcpStream, - sync::oneshot, -}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio_rustls::server::TlsStream; -use utils::listener::{SessionData, SessionManager}; +use utils::listener::{stream::NullIo, SessionManager, SessionStream}; -use super::{writer, ImapSessionManager, Session, State}; +use super::{ImapSessionManager, Session, State}; impl SessionManager for ImapSessionManager { - fn spawn(&self, session: SessionData) { - let manager = self.clone(); - - tokio::spawn(async move { - if session.instance.is_tls_implicit { - if let Ok(session) = Session::>::new(session, manager).await { - session.handle_conn().await; + #[allow(clippy::manual_async_fn)] + fn handle( + self, + session: utils::listener::SessionData, + ) -> impl std::future::Future + Send { + async move { + if let Ok(mut session) = Session::new(session, self).await { + if session.handle_conn().await && session.instance.acceptor.is_tls() { + if let Ok(mut session) = session.into_tls().await { + session.handle_conn().await; + } } - } else if let Ok(session) = Session::::new(session, manager).await { - session.handle_conn().await; } - }); + } } - fn shutdown(&self) { - // No-op + #[allow(clippy::manual_async_fn)] + fn shutdown(&self) -> impl std::future::Future + Send { + async {} } } -impl Session { - pub async fn handle_conn_(&mut self) -> bool { +impl Session { + pub async fn handle_conn(&mut self) -> bool { let mut buf = vec![0; 8192]; let mut shutdown_rx = self.instance.shutdown_rx.clone(); @@ -106,15 +107,18 @@ impl Session { false } -} -impl Session { pub async fn new( - mut session: SessionData, + mut session: utils::listener::SessionData, manager: ImapSessionManager, - ) -> Result, ()> { - // Write plain text greeting - if let Err(err) = session.stream.write_all(&manager.imap.greeting_plain).await { + ) -> Result, ()> { + // Write greeting + let (is_tls, greeting) = if session.stream.is_tls() { + (true, &manager.imap.greeting_tls) + } else { + (false, &manager.imap.greeting_plain) + }; + if let Err(err) = session.stream.write_all(greeting).await { tracing::debug!(parent: &session.span, event = "error", reason = %err, "Failed to write greeting."); return Err(()); } @@ -127,8 +131,7 @@ impl Session { receiver: Receiver::with_max_request_size(manager.imap.max_request_size), version: ProtocolVersion::Rev1, state: State::NotAuthenticated { auth_failures: 0 }, - writer: writer::spawn_writer(writer::Event::Stream(stream_tx), session.span.clone()), - is_tls: false, + is_tls, is_condstore: false, is_qresync: false, imap: manager.imap, @@ -138,38 +141,37 @@ impl Session { in_flight: session.in_flight, remote_addr: RemoteAddress::IpAddress(session.remote_ip), stream_rx, + stream_tx: Arc::new(tokio::sync::Mutex::new(stream_tx)), }) } - pub async fn handle_conn(mut self) { - if self.handle_conn_().await && self.instance.acceptor.is_tls() { - if let Ok(session) = self.into_tls().await { - session.handle_conn().await; - } - } - } - - pub async fn into_tls(self) -> Result>, ()> { - // Recover WriteHalf from writer - let (tx, rx) = oneshot::channel(); - if let Err(err) = self.writer.send(writer::Event::Upgrade(tx)).await { - tracing::debug!("Failed to write to channel: {}", err); + pub async fn into_tls(self) -> Result>, ()> { + // Drop references to write half from state + let state = if let Some(state) = + self.state + .try_replace_stream_tx(Arc::new(tokio::sync::Mutex::new( + tokio::io::split(NullIo::default()).1, + ))) { + state + } else { + tracing::debug!("Failed to obtain write half state."); return Err(()); - } - let stream = if let Ok(stream_tx) = rx.await { + }; + + // Take ownership of WriteHalf and unsplit it from ReadHalf + let stream = if let Ok(stream_tx) = + Arc::try_unwrap(self.stream_tx).map(|mutex| mutex.into_inner()) + { self.stream_rx.unsplit(stream_tx) } else { - tracing::debug!("Failed to read from channel"); + tracing::debug!("Failed to take ownership of write half."); return Err(()); }; // Upgrade to TLS let (stream_rx, stream_tx) = tokio::io::split(self.instance.tls_accept(stream, &self.span).await?); - if let Err(err) = self.writer.send(writer::Event::StreamTls(stream_tx)).await { - tracing::debug!("Failed to send stream: {}", err); - return Err(()); - } + let stream_tx = Arc::new(tokio::sync::Mutex::new(stream_tx)); Ok(Session { jmap: self.jmap, @@ -177,60 +179,63 @@ impl Session { instance: self.instance, receiver: self.receiver, version: self.version, - state: self.state, + state: state.try_replace_stream_tx(stream_tx.clone()).unwrap(), is_tls: true, is_condstore: self.is_condstore, is_qresync: self.is_qresync, - writer: self.writer, span: self.span, in_flight: self.in_flight, remote_addr: self.remote_addr, stream_rx, + stream_tx, }) } } -impl Session> { - pub async fn new( - session: utils::listener::SessionData, - manager: ImapSessionManager, - ) -> Result>, ()> { - // Upgrade to TLS - let mut stream = session - .instance - .tls_accept(session.stream, &session.span) - .await?; +impl Session { + pub async fn write_bytes(&self, bytes: impl Into>) -> crate::OpResult { + let bytes = bytes.into(); + /*for line in String::from_utf8_lossy(bytes.as_ref()).split("\r\n") { + let c = println!("{}", line); + }*/ + tracing::trace!( + parent: &self.span, + event = "write", + data = std::str::from_utf8(bytes.as_ref()).unwrap_or_default(), + size = bytes.len() + ); - // Write TLS greeting - let span = session.span; - if let Err(err) = stream.write_all(&manager.imap.greeting_tls).await { - tracing::debug!(parent: &span, event = "error", reason = %err, "Failed to write greeting."); - return Err(()); + let mut stream = self.stream_tx.lock().await; + if let Err(err) = stream.write_all(bytes.as_ref()).await { + tracing::trace!(parent: &self.span, "Failed to write to stream: {}", err); + Err(()) + } else { + let _ = stream.flush().await; + Ok(()) + } + } +} + +impl super::SessionData { + pub async fn write_bytes(&self, bytes: impl Into>) -> bool { + let bytes = bytes.into(); + /*for line in String::from_utf8_lossy(bytes.as_ref()).split("\r\n") { + let c = println!("{}", line); + }*/ + tracing::trace!( + parent: &self.span, + event = "write", + data = std::str::from_utf8(bytes.as_ref()).unwrap_or_default(), + size = bytes.len() + ); + + let mut stream = self.stream_tx.lock().await; + if let Err(err) = stream.write_all(bytes.as_ref()).await { + tracing::trace!(parent: &self.span, "Failed to write to stream: {}", err); + false + } else { + let _ = stream.flush().await; + true } - let _ = stream.flush().await; - - // Spit stream into read and write halves - let (stream_rx, stream_tx) = tokio::io::split(stream); - - Ok(Session { - receiver: Receiver::with_max_request_size(manager.imap.max_request_size), - version: ProtocolVersion::Rev1, - state: State::NotAuthenticated { auth_failures: 0 }, - writer: writer::spawn_writer(writer::Event::StreamTls(stream_tx), span.clone()), - is_tls: true, - is_condstore: false, - is_qresync: false, - imap: manager.imap, - jmap: manager.jmap, - instance: session.instance, - span, - in_flight: session.in_flight, - remote_addr: RemoteAddress::IpAddress(session.remote_ip), - stream_rx, - }) - } - - pub async fn handle_conn(mut self) { - self.handle_conn_().await; } } diff --git a/crates/imap/src/core/writer.rs b/crates/imap/src/core/writer.rs index 68ac1404..07e85f7b 100644 --- a/crates/imap/src/core/writer.rs +++ b/crates/imap/src/core/writer.rs @@ -30,6 +30,7 @@ use tokio::{ }; use tokio_rustls::server::TlsStream; use tracing::debug; +use utils::listener::SessionStream; use super::{Session, SessionData}; @@ -127,35 +128,3 @@ pub fn spawn_writer(mut stream: Event, span: tracing::Span) -> mpsc::Sender Session { - pub async fn write_bytes(&self, bytes: impl Into>) -> crate::OpResult { - let bytes = bytes.into(); - /*for line in String::from_utf8_lossy(bytes.as_ref()).split("\r\n") { - let c = println!("{}", line); - }*/ - - if let Err(err) = self.writer.send(Event::Bytes(bytes)).await { - debug!("Failed to send bytes: {}", err); - Err(()) - } else { - Ok(()) - } - } -} - -impl SessionData { - pub async fn write_bytes(&self, bytes: impl Into>) -> bool { - let bytes = bytes.into(); - /*for line in String::from_utf8_lossy(bytes.as_ref()).split("\r\n") { - let c = println!("{}", line); - }*/ - - if let Err(err) = self.writer.send(Event::Bytes(bytes)).await { - debug!("Failed to send bytes: {}", err); - false - } else { - true - } - } -} diff --git a/crates/imap/src/op/acl.rs b/crates/imap/src/op/acl.rs index 0ad6806e..db6d101e 100644 --- a/crates/imap/src/op/acl.rs +++ b/crates/imap/src/op/acl.rs @@ -49,12 +49,11 @@ use jmap_proto::{ }, }; use store::write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder}; -use tokio::io::AsyncRead; -use utils::map::bitmap::Bitmap; +use utils::{listener::SessionStream, map::bitmap::Bitmap}; use crate::core::{MailboxId, Session, SessionData}; -impl Session { +impl Session { pub async fn handle_get_acl(&mut self, request: Request) -> crate::OpResult { match request.parse_acl(self.version) { Ok(arguments) => { @@ -447,7 +446,7 @@ impl Session { } } -impl SessionData { +impl SessionData { async fn get_acl_mailbox( &self, arguments: &Arguments, diff --git a/crates/imap/src/op/append.rs b/crates/imap/src/op/append.rs index 618f647e..8bf9ff42 100644 --- a/crates/imap/src/op/append.rs +++ b/crates/imap/src/op/append.rs @@ -32,13 +32,13 @@ use imap_proto::{ use jmap::email::ingest::IngestEmail; use jmap_proto::types::{acl::Acl, keyword::Keyword, state::StateChange, type_state::DataType}; use mail_parser::MessageParser; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{MailboxId, SelectedMailbox, Session, SessionData}; use super::ToModSeq; -impl Session { +impl Session { pub async fn handle_append(&mut self, request: Request) -> crate::OpResult { match request.parse_append(self.version) { Ok(arguments) => { @@ -87,7 +87,7 @@ impl Session { } } -impl SessionData { +impl SessionData { async fn append_messages( &self, arguments: Arguments, diff --git a/crates/imap/src/op/authenticate.rs b/crates/imap/src/op/authenticate.rs index 2f6df14c..fef64abf 100644 --- a/crates/imap/src/op/authenticate.rs +++ b/crates/imap/src/op/authenticate.rs @@ -30,11 +30,11 @@ use imap_proto::{ }; use mail_parser::decoders::base64::base64_decode; use mail_send::Credentials; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{Session, SessionData, State}; -impl Session { +impl Session { pub async fn handle_authenticate(&mut self, request: Request) -> crate::OpResult { match request.parse_authenticate() { Ok(mut args) => match args.mechanism { diff --git a/crates/imap/src/op/capability.rs b/crates/imap/src/op/capability.rs index 2c69fb67..8a1813c1 100644 --- a/crates/imap/src/op/capability.rs +++ b/crates/imap/src/op/capability.rs @@ -30,11 +30,11 @@ use imap_proto::{ Command, StatusResponse, }; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::Session; -impl Session { +impl Session { pub async fn handle_capability(&mut self, request: Request) -> crate::OpResult { self.write_bytes( StatusResponse::completed(Command::Capability) diff --git a/crates/imap/src/op/close.rs b/crates/imap/src/op/close.rs index 23236f78..4b36553a 100644 --- a/crates/imap/src/op/close.rs +++ b/crates/imap/src/op/close.rs @@ -23,11 +23,11 @@ use imap_proto::{receiver::Request, Command, StatusResponse}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{Session, State}; -impl Session { +impl Session { pub async fn handle_close(&mut self, request: Request) -> crate::OpResult { let (data, mailbox) = self.state.select_data(); if mailbox.is_select { diff --git a/crates/imap/src/op/copy_move.rs b/crates/imap/src/op/copy_move.rs index 2130f3ba..31c9bf03 100644 --- a/crates/imap/src/op/copy_move.rs +++ b/crates/imap/src/op/copy_move.rs @@ -37,11 +37,11 @@ use jmap_proto::{ }, }; use store::write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, F_VALUE}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{MailboxId, SelectedMailbox, Session, SessionData}; -impl Session { +impl Session { pub async fn handle_copy_move( &mut self, request: Request, @@ -115,7 +115,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn copy_move( &self, arguments: Arguments, diff --git a/crates/imap/src/op/create.rs b/crates/imap/src/op/create.rs index 4506906b..b09dc38e 100644 --- a/crates/imap/src/op/create.rs +++ b/crates/imap/src/op/create.rs @@ -35,11 +35,11 @@ use jmap_proto::{ }, }; use store::{query::Filter, roaring::RoaringBitmap, write::BatchBuilder}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{Account, Mailbox, Session, SessionData}; -impl Session { +impl Session { pub async fn handle_create(&mut self, requests: Vec>) -> crate::OpResult { let mut arguments = Vec::with_capacity(requests.len()); @@ -65,7 +65,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn create_folder(&self, arguments: Arguments) -> StatusResponse { // Refresh mailboxes if let Err(err) = self.synchronize_mailboxes(false).await { diff --git a/crates/imap/src/op/delete.rs b/crates/imap/src/op/delete.rs index b8b44c4a..71f24d73 100644 --- a/crates/imap/src/op/delete.rs +++ b/crates/imap/src/op/delete.rs @@ -24,11 +24,11 @@ use imap_proto::{protocol::delete::Arguments, receiver::Request, Command, StatusResponse}; use jmap_proto::types::{state::StateChange, type_state::DataType}; use store::write::log::ChangeLogBuilder; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{Session, SessionData}; -impl Session { +impl Session { pub async fn handle_delete(&mut self, requests: Vec>) -> crate::OpResult { let mut arguments = Vec::with_capacity(requests.len()); @@ -54,7 +54,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn delete_folder(&self, arguments: Arguments) -> StatusResponse { // Refresh mailboxes if let Err(err) = self.synchronize_mailboxes(false).await { diff --git a/crates/imap/src/op/enable.rs b/crates/imap/src/op/enable.rs index 19593471..959a7b72 100644 --- a/crates/imap/src/op/enable.rs +++ b/crates/imap/src/op/enable.rs @@ -27,11 +27,11 @@ use imap_proto::{ Command, StatusResponse, }; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::Session; -impl Session { +impl Session { pub async fn handle_enable(&mut self, request: Request) -> crate::OpResult { match request.parse_enable() { Ok(arguments) => { diff --git a/crates/imap/src/op/expunge.rs b/crates/imap/src/op/expunge.rs index 5d30e4f2..fc832bc9 100644 --- a/crates/imap/src/op/expunge.rs +++ b/crates/imap/src/op/expunge.rs @@ -39,13 +39,13 @@ use jmap_proto::{ }, }; use store::write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, F_VALUE}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{ImapId, SavedSearch, SelectedMailbox, Session, SessionData}; use super::ToModSeq; -impl Session { +impl Session { pub async fn handle_expunge( &mut self, request: Request, @@ -130,7 +130,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn expunge( &self, mailbox: Arc, diff --git a/crates/imap/src/op/fetch.rs b/crates/imap/src/op/fetch.rs index eee3f085..c0fac6a6 100644 --- a/crates/imap/src/op/fetch.rs +++ b/crates/imap/src/op/fetch.rs @@ -50,13 +50,13 @@ use store::{ query::log::{Change, Query}, write::{assert::HashedValue, BatchBuilder, F_BITMAP, F_VALUE}, }; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{SelectedMailbox, Session, SessionData}; use super::FromModSeq; -impl Session { +impl Session { pub async fn handle_fetch( &mut self, request: Request, @@ -99,7 +99,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn fetch( &self, mut arguments: Arguments, diff --git a/crates/imap/src/op/idle.rs b/crates/imap/src/op/idle.rs index 6277627e..ff29d1ad 100644 --- a/crates/imap/src/op/idle.rs +++ b/crates/imap/src/op/idle.rs @@ -37,12 +37,13 @@ use imap_proto::{ use jmap_proto::types::{collection::Collection, type_state::DataType}; use store::query::log::Query; -use tokio::io::{AsyncRead, AsyncReadExt}; +use tokio::io::AsyncReadExt; +use utils::listener::SessionStream; use utils::map::bitmap::Bitmap; use crate::core::{SelectedMailbox, Session, SessionData, State}; -impl Session { +impl Session { pub async fn handle_idle(&mut self, request: Request) -> crate::OpResult { let (data, mailbox, types) = match &self.state { State::Authenticated { data, .. } => { @@ -140,7 +141,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn write_changes( &self, mailbox: &Option>, diff --git a/crates/imap/src/op/list.rs b/crates/imap/src/op/list.rs index cdcb0198..958d1380 100644 --- a/crates/imap/src/op/list.rs +++ b/crates/imap/src/op/list.rs @@ -32,11 +32,11 @@ use imap_proto::{ Command, StatusResponse, }; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{Session, SessionData}; -impl Session { +impl Session { pub async fn handle_list(&mut self, request: Request) -> crate::OpResult { let command = request.command; let is_lsub = command == Command::Lsub; @@ -79,7 +79,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn list(&self, arguments: Arguments, is_lsub: bool, version: ProtocolVersion) { let (tag, reference_name, mut patterns, selection_options, return_options) = match arguments { diff --git a/crates/imap/src/op/login.rs b/crates/imap/src/op/login.rs index 8008978c..150c68bc 100644 --- a/crates/imap/src/op/login.rs +++ b/crates/imap/src/op/login.rs @@ -24,11 +24,11 @@ use imap_proto::{receiver::Request, Command}; use mail_send::Credentials; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::Session; -impl Session { +impl Session { pub async fn handle_login(&mut self, request: Request) -> crate::OpResult { match request.parse_login() { Ok(args) => { diff --git a/crates/imap/src/op/logout.rs b/crates/imap/src/op/logout.rs index c66bbf2a..e6e5b22e 100644 --- a/crates/imap/src/op/logout.rs +++ b/crates/imap/src/op/logout.rs @@ -23,11 +23,11 @@ use imap_proto::{receiver::Request, Command, StatusResponse}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::Session; -impl Session { +impl Session { pub async fn handle_logout(&mut self, request: Request) -> crate::OpResult { let mut response = StatusResponse::bye( concat!( diff --git a/crates/imap/src/op/namespace.rs b/crates/imap/src/op/namespace.rs index 4e865ff3..8e611956 100644 --- a/crates/imap/src/op/namespace.rs +++ b/crates/imap/src/op/namespace.rs @@ -27,11 +27,11 @@ use imap_proto::{ Command, StatusResponse, }; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::Session; -impl Session { +impl Session { pub async fn handle_namespace(&mut self, request: Request) -> crate::OpResult { self.write_bytes( StatusResponse::completed(Command::Namespace) diff --git a/crates/imap/src/op/noop.rs b/crates/imap/src/op/noop.rs index 794f7112..0a6386dd 100644 --- a/crates/imap/src/op/noop.rs +++ b/crates/imap/src/op/noop.rs @@ -23,11 +23,11 @@ use imap_proto::{receiver::Request, Command, StatusResponse}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{Session, State}; -impl Session { +impl Session { pub async fn handle_noop(&mut self, request: Request) -> crate::OpResult { match &self.state { State::Authenticated { data, .. } => { diff --git a/crates/imap/src/op/rename.rs b/crates/imap/src/op/rename.rs index 17ff5b00..0995cd2d 100644 --- a/crates/imap/src/op/rename.rs +++ b/crates/imap/src/op/rename.rs @@ -36,11 +36,11 @@ use jmap_proto::{ }, }; use store::write::{assert::HashedValue, BatchBuilder}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{Session, SessionData}; -impl Session { +impl Session { pub async fn handle_rename(&mut self, request: Request) -> crate::OpResult { match request.parse_rename(self.version) { Ok(arguments) => { @@ -56,7 +56,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn rename_folder(&self, arguments: Arguments) -> StatusResponse { // Refresh mailboxes if let Err(err) = self.synchronize_mailboxes(false).await { diff --git a/crates/imap/src/op/search.rs b/crates/imap/src/op/search.rs index e2948812..39b55274 100644 --- a/crates/imap/src/op/search.rs +++ b/crates/imap/src/op/search.rs @@ -41,13 +41,14 @@ use store::{ roaring::RoaringBitmap, write::now, }; -use tokio::{io::AsyncRead, sync::watch}; +use tokio::sync::watch; +use utils::listener::SessionStream; use crate::core::{ImapId, MailboxState, SavedSearch, SelectedMailbox, Session, SessionData}; use super::{FromModSeq, ToModSeq}; -impl Session { +impl Session { pub async fn handle_search( &mut self, request: Request, @@ -114,7 +115,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn search( &self, arguments: Arguments, diff --git a/crates/imap/src/op/select.rs b/crates/imap/src/op/select.rs index e3375a8b..435dc436 100644 --- a/crates/imap/src/op/select.rs +++ b/crates/imap/src/op/select.rs @@ -35,13 +35,13 @@ use imap_proto::{ }; use jmap_proto::types::id::Id; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{SavedSearch, SelectedMailbox, Session, State}; use super::ToModSeq; -impl Session { +impl Session { pub async fn handle_select(&mut self, request: Request) -> crate::OpResult { let is_select = request.command == Command::Select; let command = request.command; diff --git a/crates/imap/src/op/status.rs b/crates/imap/src/op/status.rs index 822e8e96..942cf65b 100644 --- a/crates/imap/src/op/status.rs +++ b/crates/imap/src/op/status.rs @@ -37,13 +37,13 @@ use store::{ roaring::RoaringBitmap, write::key::DeserializeBigEndian, IndexKeyPrefix, IterateParams, }; use store::{Deserialize, U32_LEN}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{Mailbox, Session, SessionData}; use super::ToModSeq; -impl Session { +impl Session { pub async fn handle_status(&mut self, request: Request) -> crate::OpResult { match request.parse_status(self.version) { Ok(arguments) => { @@ -82,7 +82,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn status( &self, mailbox_name: String, diff --git a/crates/imap/src/op/store.rs b/crates/imap/src/op/store.rs index a7569a14..09578041 100644 --- a/crates/imap/src/op/store.rs +++ b/crates/imap/src/op/store.rs @@ -45,13 +45,13 @@ use store::{ query::log::{Change, Query}, write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, F_VALUE}, }; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{message::MAX_RETRIES, SelectedMailbox, Session, SessionData}; use super::FromModSeq; -impl Session { +impl Session { pub async fn handle_store( &mut self, request: Request, @@ -76,7 +76,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn store( &self, arguments: Arguments, diff --git a/crates/imap/src/op/subscribe.rs b/crates/imap/src/op/subscribe.rs index a7a44be5..113d2181 100644 --- a/crates/imap/src/op/subscribe.rs +++ b/crates/imap/src/op/subscribe.rs @@ -32,11 +32,11 @@ use jmap_proto::{ }, }; use store::write::{assert::HashedValue, BatchBuilder}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{Session, SessionData}; -impl Session { +impl Session { pub async fn handle_subscribe( &mut self, request: Request, @@ -60,7 +60,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn subscribe_folder( &self, tag: String, diff --git a/crates/imap/src/op/thread.rs b/crates/imap/src/op/thread.rs index 6dbf026e..ea7961b6 100644 --- a/crates/imap/src/op/thread.rs +++ b/crates/imap/src/op/thread.rs @@ -35,11 +35,11 @@ use imap_proto::{ use jmap_proto::types::{collection::Collection, property::Property}; use store::{write::ValueClass, ValueKey}; -use tokio::io::AsyncRead; +use utils::listener::SessionStream; use crate::core::{SelectedMailbox, Session, SessionData}; -impl Session { +impl Session { pub async fn handle_thread( &mut self, request: Request, @@ -67,7 +67,7 @@ impl Session { } } -impl SessionData { +impl SessionData { pub async fn thread( &self, arguments: Arguments, diff --git a/crates/install/Cargo.toml b/crates/install/Cargo.toml index f7fb38e8..3de253bb 100644 --- a/crates/install/Cargo.toml +++ b/crates/install/Cargo.toml @@ -5,7 +5,7 @@ authors = ["Stalwart Labs Ltd. "] license = "AGPL-3.0-only" repository = "https://github.com/stalwartlabs/mail-server" homepage = "https://github.com/stalwartlabs/mail-server" -version = "0.5.1" +version = "0.5.2" edition = "2021" readme = "README.md" resolver = "2" diff --git a/crates/jmap/Cargo.toml b/crates/jmap/Cargo.toml index b6b71f94..f545dd5a 100644 --- a/crates/jmap/Cargo.toml +++ b/crates/jmap/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "jmap" -version = "0.5.1" +version = "0.5.2" edition = "2021" resolver = "2" diff --git a/crates/jmap/src/api/http.rs b/crates/jmap/src/api/http.rs index 69517ca5..a4789290 100644 --- a/crates/jmap/src/api/http.rs +++ b/crates/jmap/src/api/http.rs @@ -38,11 +38,8 @@ use jmap_proto::{ response::Response, types::{blob::BlobId, id::Id}, }; -use tokio::{ - io::{AsyncRead, AsyncWrite}, - net::TcpStream, -}; -use utils::listener::{ServerInstance, SessionData, SessionManager, TcpAcceptorResult}; + +use utils::listener::{ServerInstance, SessionData, SessionManager, SessionStream}; use crate::{ auth::{oauth::OAuthMetadata, AccessToken}, @@ -291,61 +288,22 @@ pub async fn parse_jmap_request( } impl SessionManager for JmapSessionManager { - fn spawn(&self, mut session: SessionData) { - let jmap = self.inner.clone(); - - tokio::spawn(async move { - match session.instance.acceptor.accept(session.stream).await { - TcpAcceptorResult::Tls(accept) => { - let span = session.span; - match accept.await { - Ok(stream) => { - handle_request( - jmap, - SessionData { - stream, - local_ip: session.local_ip, - remote_ip: session.remote_ip, - remote_port: session.remote_port, - span, - in_flight: session.in_flight, - instance: session.instance, - }, - ) - .await; - } - Err(err) => { - tracing::debug!( - parent: &span, - context = "tls", - event = "error", - "Failed to accept TLS connection: {}", - err - ); - } - } - } - TcpAcceptorResult::Plain(stream) => { - session.stream = stream; - handle_request(jmap, session).await; - } - TcpAcceptorResult::Close => (), - } - }); + fn handle( + self, + session: SessionData, + ) -> impl std::future::Future + Send { + handle_request(self.inner, session) } - fn shutdown(&self) { - let jmap = self.inner.clone(); - tokio::spawn(async move { - let _ = jmap.state_tx.send(state::Event::Stop).await; - }); + #[allow(clippy::manual_async_fn)] + fn shutdown(&self) -> impl std::future::Future + Send { + async { + let _ = self.inner.state_tx.send(state::Event::Stop).await; + } } } -async fn handle_request( - jmap: Arc, - session: SessionData, -) { +async fn handle_request(jmap: Arc, session: SessionData) { let span = session.span; let _in_flight = session.in_flight; diff --git a/crates/jmap/src/sieve/ingest.rs b/crates/jmap/src/sieve/ingest.rs index f2176295..03b84178 100644 --- a/crates/jmap/src/sieve/ingest.rs +++ b/crates/jmap/src/sieve/ingest.rs @@ -27,11 +27,12 @@ use directory::QueryBy; use jmap_proto::types::{collection::Collection, id::Id, keyword::Keyword, property::Property}; use mail_parser::MessageParser; use sieve::{Envelope, Event, Input, Mailbox, Recipient}; -use smtp::core::{NullIo, Session, SessionAddress}; +use smtp::core::{Session, SessionAddress}; use store::{ ahash::AHashSet, write::{now, BatchBuilder, F_VALUE}, }; +use utils::listener::stream::NullIo; use crate::{ email::ingest::{IngestEmail, IngestedEmail}, diff --git a/crates/jmap/src/submission/set.rs b/crates/jmap/src/submission/set.rs index ce9e63f9..1a27b51d 100644 --- a/crates/jmap/src/submission/set.rs +++ b/crates/jmap/src/submission/set.rs @@ -49,13 +49,16 @@ use jmap_proto::{ }; use mail_parser::{HeaderName, HeaderValue}; use smtp::{ - core::{management::QueueRequest, NullIo, Session, SessionData, State}, + core::{management::QueueRequest, Session, SessionData, State}, queue, }; use smtp_proto::{request::parser::Rfc5321Parser, MailFrom, RcptTo}; use store::write::{assert::HashedValue, log::ChangeLogBuilder, now, BatchBuilder}; use tokio::sync::oneshot; -use utils::{listener::ServerInstance, map::vec_map::VecMap}; +use utils::{ + listener::{stream::NullIo, ServerInstance}, + map::vec_map::VecMap, +}; use crate::{email::metadata::MessageMetadata, identity::set::sanitize_email, Bincode, JMAP}; diff --git a/crates/main/Cargo.toml b/crates/main/Cargo.toml index 90976ec4..ed369e3d 100644 --- a/crates/main/Cargo.toml +++ b/crates/main/Cargo.toml @@ -7,7 +7,7 @@ homepage = "https://stalw.art" keywords = ["imap", "jmap", "smtp", "email", "mail", "server"] categories = ["email"] license = "AGPL-3.0-only" -version = "0.5.1" +version = "0.5.2" edition = "2021" resolver = "2" diff --git a/crates/managesieve/Cargo.toml b/crates/managesieve/Cargo.toml index ba51fff4..cbdaf1ca 100644 --- a/crates/managesieve/Cargo.toml +++ b/crates/managesieve/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "managesieve" -version = "0.5.1" +version = "0.5.2" edition = "2021" resolver = "2" diff --git a/crates/managesieve/src/core/client.rs b/crates/managesieve/src/core/client.rs index 90745eaf..9207c557 100644 --- a/crates/managesieve/src/core/client.rs +++ b/crates/managesieve/src/core/client.rs @@ -26,10 +26,11 @@ use imap_proto::receiver::{self, Request}; use jmap_proto::types::{collection::Collection, property::Property}; use store::query::Filter; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use utils::listener::SessionStream; -use super::{Command, IsTls, ResponseCode, ResponseType, Session, State, StatusResponse}; +use super::{Command, ResponseCode, ResponseType, Session, State, StatusResponse}; -impl Session { +impl Session { pub async fn ingest(&mut self, bytes: &[u8]) -> Result { /*let tmp = "dd"; for line in String::from_utf8_lossy(bytes).split("\r\n") { diff --git a/crates/managesieve/src/core/mod.rs b/crates/managesieve/src/core/mod.rs index 803e2af1..d3dff2ae 100644 --- a/crates/managesieve/src/core/mod.rs +++ b/crates/managesieve/src/core/mod.rs @@ -32,11 +32,7 @@ use jmap::{ auth::{rate_limit::RemoteAddress, AccessToken}, JMAP, }; -use tokio::{ - io::{AsyncRead, AsyncWrite}, - net::TcpStream, -}; -use tokio_rustls::server::TlsStream; +use tokio::io::{AsyncRead, AsyncWrite}; use utils::listener::{limiter::InFlight, ServerInstance}; pub struct Session { @@ -101,22 +97,6 @@ pub enum Command { Unauthenticate, } -pub trait IsTls { - fn is_tls(&self) -> bool; -} - -impl IsTls for TcpStream { - fn is_tls(&self) -> bool { - false - } -} - -impl IsTls for TlsStream { - fn is_tls(&self) -> bool { - true - } -} - impl CommandParser for Command { fn parse(value: &[u8], _is_uid: bool) -> Option { match value { diff --git a/crates/managesieve/src/core/session.rs b/crates/managesieve/src/core/session.rs index 715aeefd..6d4242a8 100644 --- a/crates/managesieve/src/core/session.rs +++ b/crates/managesieve/src/core/session.rs @@ -23,61 +23,56 @@ use imap_proto::receiver::{self, Receiver}; use jmap::auth::rate_limit::RemoteAddress; -use tokio::{ - io::{AsyncRead, AsyncWrite}, - net::TcpStream, -}; use tokio_rustls::server::TlsStream; -use utils::listener::SessionManager; +use utils::listener::{SessionManager, SessionStream}; use crate::SERVER_GREETING; -use super::{IsTls, ManageSieveSessionManager, Session, State}; +use super::{ManageSieveSessionManager, Session, State}; impl SessionManager for ManageSieveSessionManager { - fn spawn(&self, session: utils::listener::SessionData) { - // Create session - let mut session = Session { - jmap: self.jmap.clone(), - imap: self.imap.clone(), - instance: session.instance, - state: State::NotAuthenticated { auth_failures: 0 }, - span: session.span, - stream: session.stream, - in_flight: session.in_flight, - remote_addr: RemoteAddress::IpAddress(session.remote_ip), - receiver: Receiver::with_max_request_size(self.imap.max_request_size) - .with_start_state(receiver::State::Command { is_uid: false }), - }; + #[allow(clippy::manual_async_fn)] + fn handle( + self, + session: utils::listener::SessionData, + ) -> impl std::future::Future + Send { + async move { + // Create session + let mut session = Session { + receiver: Receiver::with_max_request_size(self.imap.max_request_size) + .with_start_state(receiver::State::Command { is_uid: false }), + jmap: self.jmap, + imap: self.imap, + instance: session.instance, + state: State::NotAuthenticated { auth_failures: 0 }, + span: session.span, + stream: session.stream, + in_flight: session.in_flight, + remote_addr: RemoteAddress::IpAddress(session.remote_ip), + }; - tokio::spawn(async move { - if session.instance.is_tls_implicit { - if let Ok(mut session) = session.into_tls().await { - if session - .write(&session.handle_capability(SERVER_GREETING).await.unwrap()) - .await - .is_ok() - { - session.handle_conn().await; - } - } - } else if session + if session .write(&session.handle_capability(SERVER_GREETING).await.unwrap()) .await .is_ok() + && session.handle_conn().await + && session.instance.acceptor.is_tls() { - session.handle_conn().await; + if let Ok(mut session) = session.into_tls().await { + session.handle_conn().await; + } } - }); + } } - fn shutdown(&self) { - // No-op + #[allow(clippy::manual_async_fn)] + fn shutdown(&self) -> impl std::future::Future + Send { + async {} } } -impl Session { - pub async fn handle_conn_(&mut self) -> bool { +impl Session { + pub async fn handle_conn(&mut self) -> bool { let mut buf = vec![0; 8192]; let mut shutdown_rx = self.instance.shutdown_rx.clone(); @@ -145,10 +140,8 @@ impl Session { false } -} -impl Session { - pub async fn into_tls(self) -> Result>, ()> { + pub async fn into_tls(self) -> Result>, ()> { let span = self.span; Ok(Session { stream: self.instance.tls_accept(self.stream, &span).await?, @@ -162,18 +155,4 @@ impl Session { remote_addr: self.remote_addr, }) } - - pub async fn handle_conn(mut self) { - if self.handle_conn_().await && self.instance.acceptor.is_tls() { - if let Ok(session) = self.into_tls().await { - session.handle_conn().await; - } - } - } -} - -impl Session> { - pub async fn handle_conn(mut self) { - self.handle_conn_().await; - } } diff --git a/crates/managesieve/src/op/authenticate.rs b/crates/managesieve/src/op/authenticate.rs index 83a812e1..fc99a191 100644 --- a/crates/managesieve/src/op/authenticate.rs +++ b/crates/managesieve/src/op/authenticate.rs @@ -30,11 +30,11 @@ use imap_proto::{ }; use mail_parser::decoders::base64::base64_decode; use mail_send::Credentials; -use tokio::io::{AsyncRead, AsyncWrite}; +use utils::listener::SessionStream; -use crate::core::{Command, IsTls, Session, State, StatusResponse}; +use crate::core::{Command, Session, State, StatusResponse}; -impl Session { +impl Session { pub async fn handle_authenticate(&mut self, request: Request) -> crate::op::OpResult { if request.tokens.is_empty() { return Err(StatusResponse::no("Authentication mechanism missing.")); diff --git a/crates/managesieve/src/op/capability.rs b/crates/managesieve/src/op/capability.rs index b9af880d..27013c68 100644 --- a/crates/managesieve/src/op/capability.rs +++ b/crates/managesieve/src/op/capability.rs @@ -22,11 +22,11 @@ */ use jmap::api::session::Capabilities; -use tokio::io::{AsyncRead, AsyncWrite}; +use utils::listener::SessionStream; -use crate::core::{IsTls, Session, StatusResponse}; +use crate::core::{Session, StatusResponse}; -impl Session { +impl Session { pub async fn handle_capability(&self, message: &'static str) -> super::OpResult { let mut response = Vec::with_capacity(128); response.extend_from_slice(b"\"IMPLEMENTATION\" \"Stalwart ManageSieve v"); diff --git a/crates/nlp/Cargo.toml b/crates/nlp/Cargo.toml index 8f72759f..c1250a8e 100644 --- a/crates/nlp/Cargo.toml +++ b/crates/nlp/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nlp" -version = "0.5.1" +version = "0.5.2" edition = "2021" resolver = "2" diff --git a/crates/smtp/Cargo.toml b/crates/smtp/Cargo.toml index f807291d..668c7817 100644 --- a/crates/smtp/Cargo.toml +++ b/crates/smtp/Cargo.toml @@ -7,7 +7,7 @@ homepage = "https://stalw.art/smtp" keywords = ["smtp", "email", "mail", "server"] categories = ["email"] license = "AGPL-3.0-only" -version = "0.5.1" +version = "0.5.2" edition = "2021" resolver = "2" diff --git a/crates/smtp/src/config/condition.rs b/crates/smtp/src/config/condition.rs index 9708d984..44b608ce 100644 --- a/crates/smtp/src/config/condition.rs +++ b/crates/smtp/src/config/condition.rs @@ -21,15 +21,13 @@ * for more details. */ -use std::net::IpAddr; - use regex::Regex; use crate::config::StringMatch; -use super::{Condition, ConditionMatch, Conditions, ConfigContext, EnvelopeKey, IpAddrMask}; +use super::{Condition, ConditionMatch, Conditions, ConfigContext, EnvelopeKey}; use utils::config::{ - utils::{AsKey, ParseKey, ParseValue}, + utils::{AsKey, ParseKey}, Config, }; @@ -313,51 +311,3 @@ impl ConfigCondition for Config { Ok(conditions) } } - -impl ParseValue for IpAddrMask { - fn parse_value(key: impl AsKey, value: &str) -> super::Result { - if let Some((addr, mask)) = value.rsplit_once('/') { - if let (Ok(addr), Ok(mask)) = - (addr.trim().parse::(), mask.trim().parse::()) - { - match addr { - IpAddr::V4(addr) if (8..=32).contains(&mask) => { - return Ok(IpAddrMask::V4 { - addr, - mask: u32::MAX << (32 - mask), - }) - } - IpAddr::V6(addr) if (8..=128).contains(&mask) => { - return Ok(IpAddrMask::V6 { - addr, - mask: u128::MAX << (128 - mask), - }) - } - _ => (), - } - } - } else { - match value.trim().parse::() { - Ok(IpAddr::V4(addr)) => { - return Ok(IpAddrMask::V4 { - addr, - mask: u32::MAX, - }) - } - Ok(IpAddr::V6(addr)) => { - return Ok(IpAddrMask::V6 { - addr, - mask: u128::MAX, - }) - } - _ => (), - } - } - - Err(format!( - "Invalid IP address {:?} for property {:?}.", - value, - key.as_key() - )) - } -} diff --git a/crates/smtp/src/config/mod.rs b/crates/smtp/src/config/mod.rs index 9fa7e8ad..a8057ca4 100644 --- a/crates/smtp/src/config/mod.rs +++ b/crates/smtp/src/config/mod.rs @@ -51,7 +51,7 @@ use regex::Regex; use sieve::Sieve; use smtp_proto::MtPriority; use store::Stores; -use utils::config::{DynValue, Rate, Server, ServerProtocol}; +use utils::config::{ipmask::IpAddrMask, DynValue, Rate, Server, ServerProtocol}; use crate::{core::Lookup, inbound::milter}; @@ -193,12 +193,6 @@ pub const THROTTLE_REMOTE_IP: u16 = 1 << 7; pub const THROTTLE_LOCAL_IP: u16 = 1 << 8; pub const THROTTLE_HELO_DOMAIN: u16 = 1 << 9; -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum IpAddrMask { - V4 { addr: Ipv4Addr, mask: u32 }, - V6 { addr: Ipv6Addr, mask: u128 }, -} - pub struct Connect { pub script: IfBlock>>, } diff --git a/crates/smtp/src/core/if_block.rs b/crates/smtp/src/core/if_block.rs index 9d6e4b89..2141b20b 100644 --- a/crates/smtp/src/core/if_block.rs +++ b/crates/smtp/src/core/if_block.rs @@ -21,13 +21,12 @@ * for more details. */ -use std::{borrow::Cow, net::IpAddr, sync::Arc}; +use std::{borrow::Cow, sync::Arc}; use utils::config::{DynValue, KeyLookup}; use crate::config::{ - Condition, ConditionMatch, Conditions, EnvelopeKey, IfBlock, IpAddrMask, MaybeDynValue, - StringMatch, + Condition, ConditionMatch, Conditions, EnvelopeKey, IfBlock, MaybeDynValue, StringMatch, }; pub struct Captures<'x, T> { @@ -198,57 +197,6 @@ impl Conditions { } } -impl IpAddrMask { - pub fn matches(&self, remote: &IpAddr) -> bool { - match self { - IpAddrMask::V4 { addr, mask } => match *mask { - u32::MAX => match remote { - IpAddr::V4(remote) => addr == remote, - IpAddr::V6(remote) => { - if let Some(remote) = remote.to_ipv4_mapped() { - addr == &remote - } else { - false - } - } - }, - 0 => { - matches!(remote, IpAddr::V4(_)) - } - _ => { - u32::from_be_bytes(match remote { - IpAddr::V4(ip) => ip.octets(), - IpAddr::V6(ip) => { - if let Some(ip) = ip.to_ipv4() { - ip.octets() - } else { - return false; - } - } - }) & mask - == u32::from_be_bytes(addr.octets()) & mask - } - }, - IpAddrMask::V6 { addr, mask } => match *mask { - u128::MAX => match remote { - IpAddr::V6(remote) => remote == addr, - IpAddr::V4(remote) => &remote.to_ipv6_mapped() == addr, - }, - 0 => { - matches!(remote, IpAddr::V6(_)) - } - _ => { - u128::from_be_bytes(match remote { - IpAddr::V6(ip) => ip.octets(), - IpAddr::V4(ip) => ip.to_ipv6_mapped().octets(), - }) & mask - == u128::from_be_bytes(addr.octets()) & mask - } - }, - } - } -} - impl<'x> Captures<'x, DynValue> { pub fn into_value(self, keys: &'x impl KeyLookup) -> Cow<'x, str> { self.value.apply(self.captures, keys) diff --git a/crates/smtp/src/core/management.rs b/crates/smtp/src/core/management.rs index c2da73aa..f57c0b47 100644 --- a/crates/smtp/src/core/management.rs +++ b/crates/smtp/src/core/management.rs @@ -36,12 +36,9 @@ use hyper_util::rt::TokioIo; use mail_parser::{decoders::base64::base64_decode, DateTime}; use mail_send::Credentials; use serde::{Deserialize, Deserializer, Serialize, Serializer}; -use tokio::{ - io::{AsyncRead, AsyncWrite}, - sync::oneshot, -}; +use tokio::sync::oneshot; -use utils::listener::{limiter::InFlight, SessionManager, TcpAcceptorResult}; +use utils::listener::{limiter::InFlight, SessionData, SessionManager, SessionStream}; use crate::{ queue::{self, instant_to_timestamp, InstantFromTimestamp, QueueId, Status}, @@ -157,39 +154,26 @@ pub struct Report { } impl SessionManager for SmtpAdminSessionManager { - fn spawn(&self, session: utils::listener::SessionData) { - let core = self.inner.clone(); - tokio::spawn(async move { - match session.instance.acceptor.accept(session.stream).await { - TcpAcceptorResult::Tls(accept) => match accept.await { - Ok(stream) => { - handle_request(stream, core, session.remote_ip, session.in_flight).await; - } - Err(err) => { - tracing::debug!( - context = "tls", - event = "error", - remote.ip = session.remote_ip.to_string(), - "Failed to accept TLS management connection: {}", - err - ); - } - }, - TcpAcceptorResult::Plain(stream) => { - handle_request(stream, core, session.remote_ip, session.in_flight).await; - } - TcpAcceptorResult::Close => (), - } - }); + fn handle( + self, + session: SessionData, + ) -> impl std::future::Future + Send { + handle_request( + session.stream, + self.inner, + session.remote_ip, + session.in_flight, + ) } - fn shutdown(&self) { - // No-op + #[allow(clippy::manual_async_fn)] + fn shutdown(&self) -> impl std::future::Future + Send { + async {} } } async fn handle_request( - stream: impl AsyncRead + AsyncWrite + Unpin + 'static, + stream: impl SessionStream, core: Arc, remote_addr: IpAddr, _in_flight: InFlight, diff --git a/crates/smtp/src/core/mod.rs b/crates/smtp/src/core/mod.rs index 62e14923..04003099 100644 --- a/crates/smtp/src/core/mod.rs +++ b/crates/smtp/src/core/mod.rs @@ -49,7 +49,7 @@ use tokio_rustls::TlsConnector; use tracing::Span; use utils::{ ipc::DeliveryEvent, - listener::{limiter::InFlight, ServerInstance, TcpAcceptor}, + listener::{limiter::InFlight, stream::NullIo, ServerInstance, TcpAcceptor}, }; use crate::{ @@ -417,62 +417,6 @@ impl PartialOrd for SessionAddress { } } -#[cfg(feature = "local_delivery")] -#[derive(Default)] -pub struct NullIo { - pub tx_buf: Vec, -} - -#[cfg(feature = "local_delivery")] -impl AsyncWrite for NullIo { - fn poll_write( - mut self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, - buf: &[u8], - ) -> std::task::Poll> { - self.tx_buf.extend_from_slice(buf); - std::task::Poll::Ready(Ok(buf.len())) - } - - fn poll_flush( - self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - std::task::Poll::Ready(Ok(())) - } - - fn poll_shutdown( - self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - std::task::Poll::Ready(Ok(())) - } -} - -#[cfg(feature = "local_delivery")] -impl AsyncRead for NullIo { - fn poll_read( - self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, - _buf: &mut tokio::io::ReadBuf<'_>, - ) -> std::task::Poll> { - unreachable!() - } -} - -#[cfg(feature = "local_delivery")] -impl crate::inbound::IsTls for NullIo { - fn is_tls(&self) -> bool { - true - } - - fn write_tls_header(&self, _headers: &mut Vec) {} - - fn tls_version_and_cipher(&self) -> (&'static str, &'static str) { - ("", "") - } -} - #[cfg(feature = "local_delivery")] lazy_static::lazy_static! { static ref SIEVE: Arc = Arc::new(utils::listener::ServerInstance { @@ -482,9 +426,9 @@ static ref SIEVE: Arc = Arc::new(utils::listener::ServerInstance hostname: "localhost".to_string(), data: "localhost".to_string(), acceptor: TcpAcceptor::Plain, - is_tls_implicit: true, limiter: utils::listener::limiter::ConcurrencyLimiter::new(0), shutdown_rx: tokio::sync::watch::channel(false).1, + proxy_networks: vec![] }); } diff --git a/crates/smtp/src/inbound/data.rs b/crates/smtp/src/inbound/data.rs index 3013378c..49b7aba2 100644 --- a/crates/smtp/src/inbound/data.rs +++ b/crates/smtp/src/inbound/data.rs @@ -38,10 +38,8 @@ use sieve::runtime::Variable; use smtp_proto::{ MAIL_BY_RETURN, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS, }; -use tokio::{ - io::{AsyncRead, AsyncWrite, AsyncWriteExt}, - process::Command, -}; +use tokio::{io::AsyncWriteExt, process::Command}; +use utils::listener::SessionStream; use crate::{ core::{Session, SessionAddress, State}, @@ -50,9 +48,9 @@ use crate::{ scripts::{ScriptModification, ScriptResult}, }; -use super::{AuthResult, IsTls}; +use super::AuthResult; -impl Session { +impl Session { pub async fn queue_message(&mut self) -> Cow<'static, [u8]> { // Authenticate message let raw_message = Arc::new(std::mem::take(&mut self.data.message)); @@ -759,7 +757,14 @@ impl Session { headers.extend_from_slice(b" ["); headers.extend_from_slice(self.data.remote_ip.to_string().as_bytes()); headers.extend_from_slice(b"])\r\n\t"); - self.stream.write_tls_header(headers); + if self.stream.is_tls() { + let (version, cipher) = self.stream.tls_version_and_cipher(); + headers.extend_from_slice(b"(using "); + headers.extend_from_slice(version.as_bytes()); + headers.extend_from_slice(b" with cipher "); + headers.extend_from_slice(cipher.as_bytes()); + headers.extend_from_slice(b")\r\n\t"); + } headers.extend_from_slice(b"by "); headers.extend_from_slice(self.instance.hostname.as_bytes()); headers.extend_from_slice(b" (Stalwart SMTP) with "); diff --git a/crates/smtp/src/inbound/ehlo.rs b/crates/smtp/src/inbound/ehlo.rs index e329a92e..11a81622 100644 --- a/crates/smtp/src/inbound/ehlo.rs +++ b/crates/smtp/src/inbound/ehlo.rs @@ -26,11 +26,9 @@ use std::time::SystemTime; use crate::{core::Session, scripts::ScriptResult}; use mail_auth::spf::verify::HasLabels; use smtp_proto::*; -use tokio::io::{AsyncRead, AsyncWrite}; +use utils::listener::SessionStream; -use super::IsTls; - -impl Session { +impl Session { pub async fn handle_ehlo(&mut self, domain: String) -> Result<(), ()> { // Set EHLO domain diff --git a/crates/smtp/src/inbound/mail.rs b/crates/smtp/src/inbound/mail.rs index b678ddab..a3ee1ac5 100644 --- a/crates/smtp/src/inbound/mail.rs +++ b/crates/smtp/src/inbound/mail.rs @@ -25,7 +25,7 @@ use std::time::SystemTime; use mail_auth::{IprevOutput, IprevResult, SpfOutput, SpfResult}; use smtp_proto::{MailFrom, MAIL_BY_NOTIFY, MAIL_BY_RETURN, MAIL_REQUIRETLS}; -use tokio::io::{AsyncRead, AsyncWrite}; +use utils::listener::SessionStream; use crate::{ core::{Session, SessionAddress}, @@ -33,9 +33,7 @@ use crate::{ scripts::{ScriptModification, ScriptResult}, }; -use super::IsTls; - -impl Session { +impl Session { pub async fn handle_mail_from(&mut self, from: MailFrom) -> Result<(), ()> { if self.data.helo_domain.is_empty() && (self.params.ehlo_require diff --git a/crates/smtp/src/inbound/milter/message.rs b/crates/smtp/src/inbound/milter/message.rs index 2fef32ae..3f9bc982 100644 --- a/crates/smtp/src/inbound/milter/message.rs +++ b/crates/smtp/src/inbound/milter/message.rs @@ -26,11 +26,12 @@ use std::borrow::Cow; use mail_auth::AuthenticatedMessage; use smtp_proto::request::parser::Rfc5321Parser; use tokio::io::{AsyncRead, AsyncWrite}; +use utils::listener::SessionStream; use crate::{ config::Milter, core::{Session, SessionAddress, SessionData}, - inbound::{milter::MilterClient, IsTls}, + inbound::milter::MilterClient, queue::DomainPart, DAEMON_NAME, }; @@ -42,7 +43,7 @@ enum Rejection { Error(Error), } -impl Session { +impl Session { pub async fn run_milters( &self, message: &AuthenticatedMessage<'_>, @@ -186,8 +187,8 @@ impl Session { .helo( &self.data.helo_domain, Macros::new() - .with_cipher(tls_ciper) - .with_tls_version(tls_version), + .with_cipher(tls_ciper.as_ref()) + .with_tls_version(tls_version.as_ref()), ) .await? .assert_continue()?; diff --git a/crates/smtp/src/inbound/mod.rs b/crates/smtp/src/inbound/mod.rs index 61cc99a0..9e2bd9b5 100644 --- a/crates/smtp/src/inbound/mod.rs +++ b/crates/smtp/src/inbound/mod.rs @@ -25,8 +25,6 @@ use mail_auth::{ arc::ArcSet, dkim::Signature, dmarc::Policy, ArcOutput, AuthenticatedMessage, AuthenticationResults, DkimResult, DmarcResult, IprevResult, SpfResult, }; -use tokio::net::TcpStream; -use tokio_rustls::server::TlsStream; use crate::config::{ArcSealer, DkimSigner}; @@ -40,70 +38,6 @@ pub mod session; pub mod spawn; pub mod vrfy; -pub trait IsTls { - fn is_tls(&self) -> bool; - fn write_tls_header(&self, headers: &mut Vec); - fn tls_version_and_cipher(&self) -> (&'static str, &'static str); -} - -impl IsTls for TcpStream { - fn is_tls(&self) -> bool { - false - } - - fn write_tls_header(&self, _headers: &mut Vec) {} - - fn tls_version_and_cipher(&self) -> (&'static str, &'static str) { - ("", "") - } -} - -impl IsTls for TlsStream { - fn is_tls(&self) -> bool { - true - } - - fn tls_version_and_cipher(&self) -> (&'static str, &'static str) { - let (_, conn) = self.get_ref(); - - ( - match conn - .protocol_version() - .unwrap_or(rustls::ProtocolVersion::Unknown(0)) - { - rustls::ProtocolVersion::SSLv2 => "SSLv2", - rustls::ProtocolVersion::SSLv3 => "SSLv3", - rustls::ProtocolVersion::TLSv1_0 => "TLSv1.0", - rustls::ProtocolVersion::TLSv1_1 => "TLSv1.1", - rustls::ProtocolVersion::TLSv1_2 => "TLSv1.2", - rustls::ProtocolVersion::TLSv1_3 => "TLSv1.3", - rustls::ProtocolVersion::DTLSv1_0 => "DTLSv1.0", - rustls::ProtocolVersion::DTLSv1_2 => "DTLSv1.2", - rustls::ProtocolVersion::DTLSv1_3 => "DTLSv1.3", - _ => "unknown", - }, - match conn.negotiated_cipher_suite() { - Some(rustls::SupportedCipherSuite::Tls13(cs)) => { - cs.common.suite.as_str().unwrap_or("unknown") - } - Some(rustls::SupportedCipherSuite::Tls12(cs)) => { - cs.common.suite.as_str().unwrap_or("unknown") - } - None => "unknown", - }, - ) - } - - fn write_tls_header(&self, headers: &mut Vec) { - let (version, cipher) = self.tls_version_and_cipher(); - headers.extend_from_slice(b"(using "); - headers.extend_from_slice(version.as_bytes()); - headers.extend_from_slice(b" with cipher "); - headers.extend_from_slice(cipher.as_bytes()); - headers.extend_from_slice(b")\r\n\t"); - } -} - impl ArcSealer { pub fn seal<'x>( &self, diff --git a/crates/smtp/src/inbound/rcpt.rs b/crates/smtp/src/inbound/rcpt.rs index f4d0446d..87d809bf 100644 --- a/crates/smtp/src/inbound/rcpt.rs +++ b/crates/smtp/src/inbound/rcpt.rs @@ -24,7 +24,7 @@ use smtp_proto::{ RcptTo, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS, }; -use tokio::io::{AsyncRead, AsyncWrite}; +use utils::listener::SessionStream; use crate::{ core::{Session, SessionAddress}, @@ -32,9 +32,7 @@ use crate::{ scripts::{ScriptModification, ScriptResult}, }; -use super::IsTls; - -impl Session { +impl Session { pub async fn handle_rcpt_to(&mut self, to: RcptTo) -> Result<(), ()> { #[cfg(feature = "test_mode")] if self.instance.id.ends_with("-debug") { diff --git a/crates/smtp/src/inbound/session.rs b/crates/smtp/src/inbound/session.rs index 6c690007..f4e2af93 100644 --- a/crates/smtp/src/inbound/session.rs +++ b/crates/smtp/src/inbound/session.rs @@ -31,16 +31,19 @@ use smtp_proto::{ *, }; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -use utils::config::{KeyLookup, ServerProtocol}; +use utils::{ + config::{KeyLookup, ServerProtocol}, + listener::SessionStream, +}; use crate::{ config::EnvelopeKey, core::{Session, State}, }; -use super::{auth::SaslToken, IsTls}; +use super::auth::SaslToken; -impl Session { +impl Session { pub async fn ingest(&mut self, bytes: &[u8]) -> Result { let mut iter = bytes.iter(); let mut state = std::mem::replace(&mut self.state, State::None); @@ -137,13 +140,17 @@ impl Session { } Request::StartTls => { if !self.stream.is_tls() { - self.write(b"220 2.0.0 Ready to start TLS.\r\n").await?; - #[cfg(any(test, feature = "test_mode"))] - if self.data.helo_domain.contains("badtls") { - return Err(()); + if self.instance.acceptor.is_tls() { + self.write(b"220 2.0.0 Ready to start TLS.\r\n").await?; + #[cfg(any(test, feature = "test_mode"))] + if self.data.helo_domain.contains("badtls") { + return Err(()); + } + self.state = State::default(); + return Ok(false); + } else { + self.write(b"502 5.7.0 TLS not available.\r\n").await?; } - self.state = State::default(); - return Ok(false); } else { self.write(b"504 5.7.4 Already in TLS mode.\r\n").await?; } diff --git a/crates/smtp/src/inbound/spawn.rs b/crates/smtp/src/inbound/spawn.rs index 2f01519f..7e564304 100644 --- a/crates/smtp/src/inbound/spawn.rs +++ b/crates/smtp/src/inbound/spawn.rs @@ -23,12 +23,8 @@ use std::time::Instant; -use tokio::{ - io::{AsyncRead, AsyncWrite}, - net::TcpStream, -}; use tokio_rustls::server::TlsStream; -use utils::listener::SessionManager; +use utils::listener::{SessionManager, SessionStream}; use crate::{ core::{Session, SessionData, SessionParameters, SmtpSessionManager, State}, @@ -36,13 +32,14 @@ use crate::{ scripts::ScriptResult, }; -use super::IsTls; - impl SessionManager for SmtpSessionManager { - fn spawn(&self, session: utils::listener::SessionData) { + fn handle( + self, + session: utils::listener::SessionData, + ) -> impl std::future::Future + Send { // Create session let mut session = Session { - core: self.inner.clone(), + core: self.inner, instance: session.instance, state: State::default(), span: session.span, @@ -52,65 +49,36 @@ impl SessionManager for SmtpSessionManager { params: SessionParameters::default(), }; - tokio::spawn(async move { - // Enforce throttle - if session.is_allowed().await { - if session.instance.is_tls_implicit { - if let Ok(mut session) = session.into_tls().await { - if session.init_conn().await { - session.handle_conn().await; - } - } - } else if session.init_conn().await { + // Enforce throttle + async { + if session.is_allowed().await + && session.init_conn().await + && session.handle_conn().await + && session.instance.acceptor.is_tls() + { + if let Ok(mut session) = session.into_tls().await { session.handle_conn().await; } } - }); + } } - fn shutdown(&self) { - // We spawn to avoid using async_trait - let core = self.inner.clone(); - tokio::spawn(async move { - let _ = core.queue.tx.send(queue::Event::Stop).await; - let _ = core.report.tx.send(reporting::Event::Stop).await; + #[allow(clippy::manual_async_fn)] + fn shutdown(&self) -> impl std::future::Future + Send { + async { + let _ = self.inner.queue.tx.send(queue::Event::Stop).await; + let _ = self.inner.report.tx.send(reporting::Event::Stop).await; #[cfg(feature = "local_delivery")] - let _ = core.delivery_tx.send(utils::ipc::DeliveryEvent::Stop).await; - }); - } -} - -impl Session { - pub async fn into_tls(self) -> Result>, ()> { - let span = self.span; - Ok(Session { - stream: self.instance.tls_accept(self.stream, &span).await?, - state: self.state, - data: self.data, - instance: self.instance, - core: self.core, - in_flight: self.in_flight, - params: self.params, - span, - }) - } - - pub async fn handle_conn(mut self) { - if self.handle_conn_().await && self.instance.acceptor.is_tls() { - if let Ok(session) = self.into_tls().await { - session.handle_conn().await; - } + let _ = self + .inner + .delivery_tx + .send(utils::ipc::DeliveryEvent::Stop) + .await; } } } -impl Session> { - pub async fn handle_conn(mut self) { - self.handle_conn_().await; - } -} - -impl Session { +impl Session { pub async fn init_conn(&mut self) -> bool { self.eval_session_params().await; @@ -138,7 +106,7 @@ impl Session { true } - pub async fn handle_conn_(&mut self) -> bool { + pub async fn handle_conn(&mut self) -> bool { let mut buf = vec![0; 8192]; let mut shutdown_rx = self.instance.shutdown_rx.clone(); @@ -229,4 +197,18 @@ impl Session { false } + + pub async fn into_tls(self) -> Result>, ()> { + let span = self.span; + Ok(Session { + stream: self.instance.tls_accept(self.stream, &span).await?, + state: self.state, + data: self.data, + instance: self.instance, + core: self.core, + in_flight: self.in_flight, + params: self.params, + span, + }) + } } diff --git a/crates/smtp/src/scripts/exec.rs b/crates/smtp/src/scripts/exec.rs index f0965f52..521620b5 100644 --- a/crates/smtp/src/scripts/exec.rs +++ b/crates/smtp/src/scripts/exec.rs @@ -26,19 +26,14 @@ use std::{sync::Arc, time::SystemTime}; use mail_auth::common::resolver::ToReverseName; use sieve::{runtime::Variable, Envelope, Sieve}; use smtp_proto::*; -use tokio::{ - io::{AsyncRead, AsyncWrite}, - runtime::Handle, -}; +use tokio::runtime::Handle; +use utils::listener::SessionStream; -use crate::{ - core::Session, - inbound::{AuthResult, IsTls}, -}; +use crate::{core::Session, inbound::AuthResult}; use super::{ScriptParameters, ScriptResult}; -impl Session { +impl Session { pub fn build_script_parameters(&self, stage: &'static str) -> ScriptParameters { let (tls_version, tls_cipher) = self.stream.tls_version_and_cipher(); let mut params = ScriptParameters::new() diff --git a/crates/utils/Cargo.toml b/crates/utils/Cargo.toml index 5b76b8eb..b83fee19 100644 --- a/crates/utils/Cargo.toml +++ b/crates/utils/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "utils" -version = "0.5.1" +version = "0.5.2" edition = "2021" resolver = "2" @@ -37,6 +37,7 @@ pem = "3.0" parking_lot = "0.12" arc-swap = "1.6.0" futures = "0.3" +proxy-header = { version = "0.1.0", features = ["tokio"] } [target.'cfg(unix)'.dependencies] privdrop = "0.5.3" diff --git a/crates/utils/src/config/ipmask.rs b/crates/utils/src/config/ipmask.rs new file mode 100644 index 00000000..5f6a50b2 --- /dev/null +++ b/crates/utils/src/config/ipmask.rs @@ -0,0 +1,131 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; + +use super::utils::{AsKey, ParseValue}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum IpAddrMask { + V4 { addr: Ipv4Addr, mask: u32 }, + V6 { addr: Ipv6Addr, mask: u128 }, +} + +impl IpAddrMask { + pub fn matches(&self, remote: &IpAddr) -> bool { + match self { + IpAddrMask::V4 { addr, mask } => match *mask { + u32::MAX => match remote { + IpAddr::V4(remote) => addr == remote, + IpAddr::V6(remote) => { + if let Some(remote) = remote.to_ipv4_mapped() { + addr == &remote + } else { + false + } + } + }, + 0 => { + matches!(remote, IpAddr::V4(_)) + } + _ => { + u32::from_be_bytes(match remote { + IpAddr::V4(ip) => ip.octets(), + IpAddr::V6(ip) => { + if let Some(ip) = ip.to_ipv4() { + ip.octets() + } else { + return false; + } + } + }) & mask + == u32::from_be_bytes(addr.octets()) & mask + } + }, + IpAddrMask::V6 { addr, mask } => match *mask { + u128::MAX => match remote { + IpAddr::V6(remote) => remote == addr, + IpAddr::V4(remote) => &remote.to_ipv6_mapped() == addr, + }, + 0 => { + matches!(remote, IpAddr::V6(_)) + } + _ => { + u128::from_be_bytes(match remote { + IpAddr::V6(ip) => ip.octets(), + IpAddr::V4(ip) => ip.to_ipv6_mapped().octets(), + }) & mask + == u128::from_be_bytes(addr.octets()) & mask + } + }, + } + } +} + +impl ParseValue for IpAddrMask { + fn parse_value(key: impl AsKey, value: &str) -> super::Result { + if let Some((addr, mask)) = value.rsplit_once('/') { + if let (Ok(addr), Ok(mask)) = + (addr.trim().parse::(), mask.trim().parse::()) + { + match addr { + IpAddr::V4(addr) if (8..=32).contains(&mask) => { + return Ok(IpAddrMask::V4 { + addr, + mask: u32::MAX << (32 - mask), + }) + } + IpAddr::V6(addr) if (8..=128).contains(&mask) => { + return Ok(IpAddrMask::V6 { + addr, + mask: u128::MAX << (128 - mask), + }) + } + _ => (), + } + } + } else { + match value.trim().parse::() { + Ok(IpAddr::V4(addr)) => { + return Ok(IpAddrMask::V4 { + addr, + mask: u32::MAX, + }) + } + Ok(IpAddr::V6(addr)) => { + return Ok(IpAddrMask::V6 { + addr, + mask: u128::MAX, + }) + } + _ => (), + } + } + + Err(format!( + "Invalid IP address {:?} for property {:?}.", + value, + key.as_key() + )) + } +} diff --git a/crates/utils/src/config/listener.rs b/crates/utils/src/config/listener.rs index 735657e9..f2bcb90c 100644 --- a/crates/utils/src/config/listener.rs +++ b/crates/utils/src/config/listener.rs @@ -146,7 +146,7 @@ impl Config { "send-buffer-size" => socket.set_send_buffer_size(value.parse_key(key)?), "recv-buffer-size" => socket.set_recv_buffer_size(value.parse_key(key)?), "tos" => socket.set_tos(value.parse_key(key)?), - _ => unreachable!(), + _ => continue, } .map_err(|err| { format!("Failed to set socket option '{option}' for listener '{id}': {err}") @@ -328,6 +328,15 @@ impl Config { let protocol = self.property_require(("server.listener", id, "protocol"))?; + // Parse proxy networks + let mut proxy_networks = Vec::new(); + for (key, protocol) in self.values_or_default( + ("server.listener", id, "proxy-trusted-networks"), + "server.proxy-trusted-networks", + ) { + proxy_networks.push(protocol.parse_key(key)?); + } + Ok(Server { id: id.to_string(), internal_id: 0, @@ -364,6 +373,7 @@ impl Config { listeners, acceptor, tls_implicit, + proxy_networks, }) } } diff --git a/crates/utils/src/config/mod.rs b/crates/utils/src/config/mod.rs index 0edf72b9..a0c71805 100644 --- a/crates/utils/src/config/mod.rs +++ b/crates/utils/src/config/mod.rs @@ -23,6 +23,7 @@ pub mod cron; pub mod dynvalue; +pub mod ipmask; pub mod listener; pub mod parser; pub mod tls; @@ -47,7 +48,7 @@ use crate::{ UnwrapFailure, }; -use self::utils::ParseValue; +use self::{ipmask::IpAddrMask, utils::ParseValue}; #[derive(Debug, Default, Clone, PartialEq, Eq)] pub struct Config { @@ -62,6 +63,7 @@ pub struct Server { pub data: String, pub protocol: ServerProtocol, pub listeners: Vec, + pub proxy_networks: Vec, pub acceptor: TcpAcceptor, pub tls_implicit: bool, pub max_connections: u64, diff --git a/crates/utils/src/listener/listen.rs b/crates/utils/src/listener/listen.rs index 3812176a..6651d782 100644 --- a/crates/utils/src/listener/listen.rs +++ b/crates/utils/src/listener/listen.rs @@ -21,8 +21,13 @@ * for more details. */ -use std::{net::IpAddr, sync::Arc}; +use std::{ + net::{IpAddr, SocketAddr}, + sync::Arc, + time::Duration, +}; +use proxy_header::io::ProxiedStream; use rustls::crypto::ring::cipher_suite::TLS13_AES_128_GCM_SHA256; use tokio::{ net::{TcpListener, TcpStream}, @@ -39,7 +44,9 @@ use crate::{ UnwrapFailure, }; -use super::{limiter::ConcurrencyLimiter, ServerInstance, SessionManager, TcpAcceptorResult}; +use super::{ + limiter::ConcurrencyLimiter, ServerInstance, SessionManager, SessionStream, TcpAcceptorResult, +}; impl Server { pub fn spawn(self, manager: impl SessionManager, shutdown_rx: watch::Receiver) { @@ -55,10 +62,12 @@ impl Server { protocol: self.protocol, hostname: self.hostname, acceptor: self.acceptor, - is_tls_implicit: self.tls_implicit, + proxy_networks: self.proxy_networks, limiter: ConcurrencyLimiter::new(self.max_connections), shutdown_rx, }); + let is_tls = self.tls_implicit; + let has_proxies = !instance.proxy_networks.is_empty(); // Spawn listeners for listener in self.listeners { @@ -67,15 +76,17 @@ impl Server { protocol = ?instance.protocol, bind.ip = listener.addr.ip().to_string(), bind.port = listener.addr.port(), - tls = instance.is_tls_implicit, + tls = is_tls, "Starting listener" ); let local_ip = listener.addr.ip(); // Obtain TCP options - let nodelay = listener.nodelay; - let ttl = listener.ttl; - let linger = listener.linger; + let opts = SocketOpts { + nodelay: listener.nodelay, + ttl: listener.ttl, + linger: listener.linger, + }; // Bind socket let listener = listener.listen(); @@ -90,79 +101,42 @@ impl Server { stream = listener.accept() => { match stream { Ok((stream, remote_addr)) => { - // Convert mapped IPv6 addresses to IPv4 - let remote_ip = match remote_addr.ip() { - IpAddr::V6(ip) => { - ip.to_ipv4_mapped() - .map(IpAddr::V4) - .unwrap_or(IpAddr::V6(ip)) - } - remote_ip => remote_ip, - }; - let remote_port = remote_addr.port(); + if has_proxies && instance.proxy_networks.iter().any(|network| network.matches(&remote_addr.ip())) { + let instance = instance.clone(); + let manager = manager.clone(); - // Enforce concurrency - if let Some(in_flight) = instance.limiter.is_allowed() { - let span = tracing::info_span!( - "session", - instance = instance.id, - protocol = ?instance.protocol, - remote.ip = remote_ip.to_string(), - remote.port = remote_port, - ); + // Set socket options + opts.apply(&stream); - // Set TCP options - if let Err(err) = stream.set_nodelay(nodelay) { - tracing::warn!( - context = "tcp", - event = "error", - instance = instance.id, - protocol = ?instance.protocol, - "Failed to set no-delay: {}", err); - } - if let Some(ttl) = ttl { - if let Err(err) = stream.set_ttl(ttl) { - tracing::warn!( - context = "tcp", - event = "error", - instance = instance.id, - protocol = ?instance.protocol, - "Failed to set TTL: {}", err); + tokio::spawn(async move { + match ProxiedStream::create_from_tokio(stream, Default::default()).await { + Ok(stream) =>{ + let remote_addr = stream.proxy_header() + .proxied_address() + .map(|addr| addr.source) + .unwrap_or(remote_addr); + if let Some(session) = instance.build_session(stream, local_ip, remote_addr) { + // Spawn session + manager.spawn(session, is_tls); + } + } + Err(err) => { + tracing::trace!(context = "io", + event = "error", + instance = instance.id, + protocol = ?instance.protocol, + reason = %err, + "Failed to accept proxied TCP connection"); + } } - } - if linger.is_some() { - if let Err(err) = stream.set_linger(linger) { - tracing::warn!( - context = "tcp", - event = "error", - instance = instance.id, - protocol = ?instance.protocol, - "Failed to set linger: {}", err); - } - } - - // Spawn connection - manager.spawn(SessionData { - stream, - local_ip, - remote_ip, - remote_port, - span, - in_flight, - instance: instance.clone(), }); - } else { - tracing::info!( - context = "throttle", - event = "too-many-requests", - instance = instance.id, - protocol = ?instance.protocol, - remote.ip = remote_ip.to_string(), - remote.port = remote_port, - max_concurrent = instance.limiter.max_concurrent, - "Too many concurrent connections." - ); - }; + } else if let Some(session) = instance.build_session(stream, local_ip, remote_addr) { + // Set socket options + opts.apply(&session.stream); + + // Spawn session + manager.spawn(session, is_tls); + } } Err(err) => { tracing::trace!(context = "io", @@ -189,6 +163,106 @@ impl Server { } } +trait BuildSession { + fn build_session( + &self, + stream: T, + local_ip: IpAddr, + remote_addr: SocketAddr, + ) -> Option>; +} + +impl BuildSession for Arc { + fn build_session( + &self, + stream: T, + local_ip: IpAddr, + remote_addr: SocketAddr, + ) -> Option> { + // Convert mapped IPv6 addresses to IPv4 + let remote_ip = match remote_addr.ip() { + IpAddr::V6(ip) => ip + .to_ipv4_mapped() + .map(IpAddr::V4) + .unwrap_or(IpAddr::V6(ip)), + remote_ip => remote_ip, + }; + let remote_port = remote_addr.port(); + + // Enforce concurrency + if let Some(in_flight) = self.limiter.is_allowed() { + SessionData { + stream, + in_flight, + span: tracing::info_span!( + "session", + instance = self.id, + protocol = ?self.protocol, + remote.ip = remote_ip.to_string(), + remote.port = remote_port, + ), + local_ip, + remote_ip, + remote_port, + instance: self.clone(), + } + .into() + } else { + tracing::info!( + context = "throttle", + event = "too-many-requests", + instance = self.id, + protocol = ?self.protocol, + remote.ip = remote_ip.to_string(), + remote.port = remote_port, + max_concurrent = self.limiter.max_concurrent, + "Too many concurrent connections." + ); + None + } + } +} + +pub struct SocketOpts { + pub nodelay: bool, + pub ttl: Option, + pub linger: Option, +} + +impl SocketOpts { + pub fn apply(&self, stream: &TcpStream) { + // Set TCP options + if let Err(err) = stream.set_nodelay(self.nodelay) { + tracing::warn!( + context = "tcp", + event = "error", + "Failed to set no-delay: {}", + err + ); + } + if let Some(ttl) = self.ttl { + if let Err(err) = stream.set_ttl(ttl) { + tracing::warn!( + context = "tcp", + event = "error", + "Failed to set TTL: {}", + err + ); + } + } + if self.linger.is_some() { + if let Err(err) = stream.set_linger(self.linger) { + tracing::warn!( + context = "tcp", + event = "error", + "Failed to set linger: {}", + err + ); + } + } + } +} + impl Servers { pub fn bind(&self, config: &Config) { // Bind as root @@ -242,11 +316,11 @@ impl Listener { } impl ServerInstance { - pub async fn tls_accept( + pub async fn tls_accept( &self, - stream: TcpStream, + stream: T, span: &Span, - ) -> Result, ()> { + ) -> Result, ()> { match self.acceptor.accept(stream).await { TcpAcceptorResult::Tls(accept) => match accept.await { Ok(stream) => { diff --git a/crates/utils/src/listener/mod.rs b/crates/utils/src/listener/mod.rs index f2c2d900..18ee2b02 100644 --- a/crates/utils/src/listener/mod.rs +++ b/crates/utils/src/listener/mod.rs @@ -21,14 +21,16 @@ * for more details. */ -use std::{net::IpAddr, sync::Arc}; +use std::{borrow::Cow, net::IpAddr, sync::Arc}; -use crate::{acme::AcmeManager, config::ServerProtocol}; +use crate::{ + acme::AcmeManager, + config::{ipmask::IpAddrMask, ServerProtocol}, +}; use rustls::ServerConfig; use std::fmt::Debug; use tokio::{ io::{AsyncRead, AsyncWrite}, - net::TcpStream, sync::watch, }; use tokio_rustls::{Accept, TlsAcceptor}; @@ -37,6 +39,7 @@ use self::limiter::{ConcurrencyLimiter, InFlight}; pub mod limiter; pub mod listen; +pub mod stream; pub mod tls; pub struct ServerInstance { @@ -46,8 +49,8 @@ pub struct ServerInstance { pub hostname: String, pub data: String, pub acceptor: TcpAcceptor, - pub is_tls_implicit: bool, pub limiter: ConcurrencyLimiter, + pub proxy_networks: Vec, pub shutdown_rx: watch::Receiver, } @@ -73,7 +76,7 @@ where Close, } -pub struct SessionData { +pub struct SessionData { pub stream: T, pub local_ip: IpAddr, pub remote_ip: IpAddr, @@ -83,9 +86,59 @@ pub struct SessionData { pub instance: Arc, } +pub trait SessionStream: AsyncRead + AsyncWrite + Unpin + 'static + Sync + Send { + fn is_tls(&self) -> bool; + fn tls_version_and_cipher(&self) -> (Cow<'static, str>, Cow<'static, str>); +} + pub trait SessionManager: Sync + Send + 'static + Clone { - fn spawn(&self, session: SessionData); - fn shutdown(&self); + fn spawn(&self, mut session: SessionData, is_tls: bool) { + let manager = self.clone(); + + tokio::spawn(async move { + if is_tls { + match session.instance.acceptor.accept(session.stream).await { + TcpAcceptorResult::Tls(accept) => match accept.await { + Ok(stream) => { + let session = SessionData { + stream, + local_ip: session.local_ip, + remote_ip: session.remote_ip, + remote_port: session.remote_port, + span: session.span, + in_flight: session.in_flight, + instance: session.instance, + }; + manager.handle(session).await; + } + Err(err) => { + tracing::debug!( + context = "tls", + event = "error", + remote.ip = session.remote_ip.to_string(), + "Failed to accept TLS connection: {}", + err + ); + } + }, + TcpAcceptorResult::Plain(stream) => { + session.stream = stream; + manager.handle(session).await; + } + TcpAcceptorResult::Close => (), + } + } else { + manager.handle(session).await; + } + }); + } + + fn handle( + self, + session: SessionData, + ) -> impl std::future::Future + Send; + + fn shutdown(&self) -> impl std::future::Future + Send; } impl Debug for TcpAcceptor { diff --git a/crates/utils/src/listener/stream.rs b/crates/utils/src/listener/stream.rs new file mode 100644 index 00000000..d01e7174 --- /dev/null +++ b/crates/utils/src/listener/stream.rs @@ -0,0 +1,160 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of the Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::borrow::Cow; + +use proxy_header::io::ProxiedStream; +use tokio::{ + io::{AsyncRead, AsyncWrite}, + net::TcpStream, +}; +use tokio_rustls::server::TlsStream; + +use super::SessionStream; + +impl SessionStream for TcpStream { + fn is_tls(&self) -> bool { + false + } + + fn tls_version_and_cipher(&self) -> (Cow<'static, str>, Cow<'static, str>) { + (Cow::Borrowed(""), Cow::Borrowed("")) + } +} + +impl SessionStream for TlsStream { + fn is_tls(&self) -> bool { + true + } + + fn tls_version_and_cipher(&self) -> (Cow<'static, str>, Cow<'static, str>) { + let (_, conn) = self.get_ref(); + + ( + match conn + .protocol_version() + .unwrap_or(rustls::ProtocolVersion::Unknown(0)) + { + rustls::ProtocolVersion::SSLv2 => "SSLv2", + rustls::ProtocolVersion::SSLv3 => "SSLv3", + rustls::ProtocolVersion::TLSv1_0 => "TLSv1.0", + rustls::ProtocolVersion::TLSv1_1 => "TLSv1.1", + rustls::ProtocolVersion::TLSv1_2 => "TLSv1.2", + rustls::ProtocolVersion::TLSv1_3 => "TLSv1.3", + rustls::ProtocolVersion::DTLSv1_0 => "DTLSv1.0", + rustls::ProtocolVersion::DTLSv1_2 => "DTLSv1.2", + rustls::ProtocolVersion::DTLSv1_3 => "DTLSv1.3", + _ => "unknown", + } + .into(), + match conn.negotiated_cipher_suite() { + Some(rustls::SupportedCipherSuite::Tls13(cs)) => { + cs.common.suite.as_str().unwrap_or("unknown") + } + Some(rustls::SupportedCipherSuite::Tls12(cs)) => { + cs.common.suite.as_str().unwrap_or("unknown") + } + None => "unknown", + } + .into(), + ) + } +} + +impl SessionStream for ProxiedStream { + fn is_tls(&self) -> bool { + self.proxy_header() + .ssl() + .map_or(false, |ssl| ssl.client_ssl()) + } + + fn tls_version_and_cipher(&self) -> (Cow<'static, str>, Cow<'static, str>) { + self.proxy_header() + .ssl() + .map(|ssl| { + ( + ssl.version().unwrap_or("unknown").to_string().into(), + ssl.cipher().unwrap_or("unknown").to_string().into(), + ) + }) + .unwrap_or((Cow::Borrowed("unknown"), Cow::Borrowed("unknown"))) + } +} + +#[derive(Default)] +pub struct NullIo { + pub tx_buf: Vec, +} + +impl AsyncWrite for NullIo { + fn poll_write( + mut self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + self.tx_buf.extend_from_slice(buf); + std::task::Poll::Ready(Ok(buf.len())) + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } +} + +impl AsyncRead for NullIo { + fn poll_read( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + _buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + unreachable!() + } +} + +impl SessionStream for NullIo { + fn is_tls(&self) -> bool { + true + } + + fn tls_version_and_cipher( + &self, + ) -> ( + std::borrow::Cow<'static, str>, + std::borrow::Cow<'static, str>, + ) { + ( + std::borrow::Cow::Borrowed(""), + std::borrow::Cow::Borrowed(""), + ) + } +} diff --git a/crates/utils/src/listener/tls.rs b/crates/utils/src/listener/tls.rs index 1e12f188..91376231 100644 --- a/crates/utils/src/listener/tls.rs +++ b/crates/utils/src/listener/tls.rs @@ -42,7 +42,7 @@ use tokio_rustls::{Accept, LazyConfigAcceptor, TlsAcceptor}; use crate::{acme::resolver::IsTlsAlpnChallenge, config::tls::build_certified_key}; -use super::{TcpAcceptor, TcpAcceptorResult}; +use super::{SessionStream, TcpAcceptor, TcpAcceptorResult}; pub static TLS13_VERSION: &[&SupportedProtocolVersion] = &[&TLS13]; pub static TLS12_VERSION: &[&SupportedProtocolVersion] = &[&TLS12]; @@ -93,7 +93,7 @@ impl ResolvesServerCert for CertificateResolver { impl TcpAcceptor { pub async fn accept(&self, stream: IO) -> TcpAcceptorResult where - IO: AsyncRead + AsyncWrite + Unpin, + IO: SessionStream, { match self { TcpAcceptor::Tls(acceptor) => TcpAcceptorResult::Tls(acceptor.accept(stream)), diff --git a/resources/config.zip b/resources/config.zip index 1e4047ad1a9d20b4e6858d42581e991785ac0cd1..a53e77eac2a98c01e6046e9f878188793f3e06cf 100644 GIT binary patch delta 1580 zcmZ8hX;4#F7=7=1C>A9o0VD|ol~ELJ6>$ia8k$fQ(yA;D8X`M_?1Gk&YHJKz*byRp zBuEhzL|V}z)hbejR*{OhwJNAHj<$8Jff103*xva3(fZ@gopbKF-+u3oHj86z7Dqg+ zn30dGj(S@CMRTU_aaJBW^^-zOQxp$$?!1}mq<#7r(@M{nOwne;E=$? zDR+kFZ5sS8(va=_=yLly^OI@9><_#-UA^6$X@86wzrU81Uw?h?i}g~eB2T@9+h-W{ z-R-hrOY$Pu-0Q)krMcA2(SZHA?aw=+eENJU{hDjTpBQh=)N(GlJzf{6xZQQHN&V$v z+0fHoJ7LCvrnS{(Rf*NXQ{5M$B5UId1*>im}Gizp=w}<-`Spi9_CJY$nV~xVUBctAO_VsRxcVa?ZK)DwhpgIGI^L z|GQ<5ETcu0qP@b$y&-Yw0 z7GixhUgdt#W~8w{RaksG%J}tr>8+2;Yi+4lkd?&OCh_UxNqkrvWZ6*Rteh7+JIz1} z=Y^Lyz$|rRnLg$qS?`v$M>@IFl{@k7`1WPK5BsM+^#@LhQ(k6D-6w|v4)0g^HqXDO z_E0)Fcn&p;Wf*x;a!o|r_@C$Za{RqTUYE|!8-S8=T zn!|}+;j!3j7W3l{#|MVEDf-Sgsp@|f{HQT6D{Yz6a$_Xb^YkLG$q&{?F&}1Ri34BO zyCzVBr8i`Vx1d!f zsAvNwWI`A{hTWMEPH#n*EC{6~sLg^9+5>+#ZKqM5O(d^yPd2OtUl;7i0U`2pz#H^$ z!Nwf$r_bY=9N0(?pldE9!CCnmovu%AjlblA4DJ@N7iVc97|!lufnE!1h~)t-#Dcg5 z2elx=O?j}Ix%!OOb;t#HHxGiCqed3^=My)HSeFke#0|HAM7$JJ3m^e<8RQj^lGHFT z1Mht$>6qSWY}2DJKV?<6t>yS`$X*LiN^@3M@_0$lyT;%e)R@3+#kl72pVpu(!f6 z;Gs<=@hrulN;0@vk$~z*64N;o8H?r!>zL?h!0>3&PYr^M;l%)2#(*QmWNT>bp7a%1RQva=HKlWF_42Jv*DBnh`iuB|)bj1NS#}PZL^~CDQc)+@N zu*Gi(qTBGPp7b7)P-Xxn*e0Wa?96a7;O%6vM{YF{AqG^F=kYT(S3@Ly2)VmqJ##i4 o@Y{4U#cv+s@!jMh)@J|~Wxx!UvabOVeFY^ouyo?Xtw6f{FVdDo*8l(j delta 1591 zcmY*ZdrVVj6ucKzXl_3MmqO{_IDD5Y??hMO& zPPM!#)U3uL&-p%5Xn65@YD$E&d1-7>R)6nF@>_C==}D-lGN5r=y7A@UnHAq%ENtoA zo7dN+`LldNL{DPOlB;Lm?ec{qs$1uyC%Np{;qEZph{@SeeP>^7>+GOIuUj9_EOpH{ zUY=VmIh${~uCS#dkIZ2ztp%in%iyeH3Wyr6imTkx#2Z*`vS8@|Tyb8cnk z0={eL^fpCse{);@X6J*P)Y9Q&PlP?kzv$EJS~Xq0wy;QvS^D+$7u9Ypc|R!|npe*) zzp&gQNw=%deP)b!TnnB>&VWBecgsTKyAKVVb0j%@wmnb0v~%&7!#dYp=H9AMqt7qR z^VcvhNs-yB)a=D}n7u%y^tiVvY!%OP@sWY@BE+X3L)e4|L#U-63O7;~r?iM_j*7w` zY!IR2AuGP}O1Ihm=xBU-nEjP5!%=sv`-iC>@@0aJ^-Za1+t(MQMyM90=#yDSY4mpX6WR-OE7NZ|RoQhEP=>SE<~b(Wf0 z*;?@diEDCak`oDSR>zNB?izx*#4YTy%%Vy&~J*;JW@s1u+*ie)e zKo+^_!&rr4G4|I(2wo|GK$7IgSS$Q|kyi)`HVxwoA)VFX;X+u){)+ziy?zFNf_&0i9goqX`qf`8mLFgpc-H@J}!n>A_;#ZS*OYQm|&ntE{EV@ z11L$0Y)tQtq7qn6u0@U^-dJ4%3gTGAAwrV}J}-ep_FG(33cHEJHfGAlYo(AzdRoRI ziZb|!47ZO%TFPK0X&+!m6Qn7*qsJgCKv_9`rFw7-=41aLE5zCncJPBWx!!mM> zg%N*0)EH?w(_vJS;&me>PK)oDU^6+w1FTAbK>UXX0z7V_aoYKumzxU+@sWknr6>X( zF024w@^AvDe19TLdZWB01TTRPojr!3+sQZzP6#=hdlisMVyAFYvKf-uEqKHXTZy|j zV0#?+qpXtVX%z$B5W|ddk0peF=UYL=A)|=~AZvl;Y%Hc&Ae#iw;bQk#Ac?G%bJB+g zD4hY(TxGgwn(6riv}Px=RkY^%`JBd71<6FZkn?&~MZI>%jV2~XT{WemFMd?#kMkCT zCq5j3RwRr%&l=$>1aqx)j>;3c2&b)dDu$CdDcA;yBwoo$O-gV>n~g@0q;Zuzv(cob z>6}4U4Q1j|1}6orrQG+_P;==z4w_d>x56Jy^hegK0S~Jo0EcU7UGca8VGW7Z0O~cA z)DEP6LmL2RYiVbF>ZsJF1su?V&&VFlKzkisH^kF*upsEI>oIzK$M^Ca3hfXK^p@Pq zr_UwZA&{-Y96Q~)V!Ubx6>GqddfMjht$;z<;Ek2_pn$$?yj~Bh==g^>&`nfhX#-us z%AJ6}?}Q*+vY#d)J4okg?;gObdnl7z4$?v_c;X=4*c_bL2pYBlvl}7K<) { - let push = self.inner.clone(); - - tokio::spawn(async move { + #[allow(clippy::manual_async_fn)] + fn handle( + self, + session: SessionData, + ) -> impl std::future::Future + 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 + Send { + async {} + } } async fn expect_push(event_rx: &mut mpsc::Receiver) -> PushMessage { diff --git a/tests/src/smtp/config.rs b/tests/src/smtp/config.rs index c2bdff41..aa2875f3 100644 --- a/tests/src/smtp/config.rs +++ b/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![], }, ]; diff --git a/tests/src/smtp/inbound/limits.rs b/tests/src/smtp/inbound/limits.rs index 763cd609..c8475fdc 100644 --- a/tests/src/smtp/inbound/limits.rs +++ b/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:\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:\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:\r\n"); - session.handle_conn_().await; + session.handle_conn().await; session.response().assert_code("221 2.0.0"); } diff --git a/tests/src/smtp/outbound/mod.rs b/tests/src/smtp/outbound/mod.rs index 60e43755..10962ffa 100644 --- a/tests/src/smtp/outbound/mod.rs +++ b/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 diff --git a/tests/src/smtp/session.rs b/tests/src/smtp/session.rs index 7fd61023..970d31d4 100644 --- a/tests/src/smtp/session.rs +++ b/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) {} - - 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> { + None + } +} + impl TestConfig for ServerInstance { fn test() -> Self { Self::test_with_shutdown(watch::channel(false).1)