diff --git a/Cargo.lock b/Cargo.lock index 54bded3f..42dd6ede 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3401,6 +3401,7 @@ dependencies = [ "jmap", "jmap_proto", "managesieve", + "pop3", "smtp", "store", "tokio", @@ -4270,6 +4271,24 @@ dependencies = [ "universal-hash", ] +[[package]] +name = "pop3" +version = "0.8.0" +dependencies = [ + "common", + "imap", + "jmap", + "jmap_proto", + "mail-parser", + "mail-send", + "rustls 0.22.4", + "store", + "tokio", + "tokio-rustls 0.25.0", + "tracing", + "utils", +] + [[package]] name = "portable-atomic" version = "1.6.0" @@ -6124,6 +6143,7 @@ dependencies = [ "managesieve", "nlp", "num_cpus", + "pop3", "rayon", "reqwest 0.12.4", "rustls 0.22.4", diff --git a/Cargo.toml b/Cargo.toml index d04b4b70..e1208b2d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "crates/imap-proto", "crates/smtp", "crates/managesieve", + "crates/pop3", "crates/nlp", "crates/store", "crates/directory", diff --git a/crates/common/src/config/server/listener.rs b/crates/common/src/config/server/listener.rs index f5349721..04756b18 100644 --- a/crates/common/src/config/server/listener.rs +++ b/crates/common/src/config/server/listener.rs @@ -332,6 +332,8 @@ impl ParseValue for ServerProtocol { Ok(Self::Http) } else if value.eq_ignore_ascii_case("managesieve") { Ok(Self::ManageSieve) + } else if value.eq_ignore_ascii_case("pop3") { + Ok(Self::Pop3) } else { Err(format!("Invalid server protocol type {:?}.", value,)) } diff --git a/crates/common/src/config/server/mod.rs b/crates/common/src/config/server/mod.rs index 4950cb8e..e8e4acf3 100644 --- a/crates/common/src/config/server/mod.rs +++ b/crates/common/src/config/server/mod.rs @@ -42,6 +42,7 @@ pub enum ServerProtocol { Smtp, Lmtp, Imap, + Pop3, Http, ManageSieve, } @@ -53,6 +54,7 @@ impl ServerProtocol { ServerProtocol::Lmtp => "lmtp", ServerProtocol::Imap => "imap", ServerProtocol::Http => "http", + ServerProtocol::Pop3 => "pop3", ServerProtocol::ManageSieve => "managesieve", } } diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index a63e3548..46950e98 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -319,7 +319,7 @@ impl Tracers { | Tracer::Otel { level, .. }) = tracer; let filter = match EnvFilter::builder().parse(format!( - "smtp={level},imap={level},jmap={level},store={level},common={level},utils={level},directory={level}" + "smtp={level},imap={level},jmap={level},pop3={level},store={level},common={level},utils={level},directory={level}" )) { Ok(filter) => { filter diff --git a/crates/common/src/manager/boot.rs b/crates/common/src/manager/boot.rs index 1692c14f..3e2839b0 100644 --- a/crates/common/src/manager/boot.rs +++ b/crates/common/src/manager/boot.rs @@ -425,6 +425,15 @@ bind = "[::]:993" protocol = "imap" tls.implicit = true +[server.listener.pop3] +bind = "[::]:110" +protocol = "pop3" + +[server.listener.pop3s] +bind = "[::]:995" +protocol = "pop3" +tls.implicit = true + [server.listener.sieve] bind = "[::]:4190" protocol = "managesieve" @@ -491,6 +500,15 @@ bind = "[::]:993" protocol = "imap" tls.implicit = true +[server.listener.pop3] +bind = "[::]:110" +protocol = "pop3" + +[server.listener.pop3s] +bind = "[::]:995" +protocol = "pop3" +tls.implicit = true + [server.listener.sieve] bind = "[::]:4190" protocol = "managesieve" diff --git a/crates/imap/src/core/session.rs b/crates/imap/src/core/session.rs index 91bcac12..783dce9e 100644 --- a/crates/imap/src/core/session.rs +++ b/crates/imap/src/core/session.rs @@ -21,7 +21,7 @@ * for more details. */ -use std::{borrow::Cow, sync::Arc}; +use std::sync::Arc; use common::listener::{stream::NullIo, SessionData, SessionManager, SessionStream}; use imap_proto::{protocol::ProtocolVersion, receiver::Receiver}; @@ -194,20 +194,20 @@ impl Session { } impl Session { - pub async fn write_bytes(&self, bytes: impl Into>) -> crate::OpResult { - let bytes = bytes.into(); + pub async fn write_bytes(&self, bytes: impl AsRef<[u8]>) -> crate::OpResult { + let bytes = bytes.as_ref(); /*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(), + data = std::str::from_utf8(bytes).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 { + if let Err(err) = stream.write_all(bytes).await { tracing::trace!(parent: &self.span, "Failed to write to stream: {}", err); Err(()) } else { @@ -218,15 +218,15 @@ impl Session { } impl super::SessionData { - pub async fn write_bytes(&self, bytes: impl Into>) -> bool { - let bytes = bytes.into(); + pub async fn write_bytes(&self, bytes: impl AsRef<[u8]>) -> bool { + let bytes = bytes.as_ref(); /*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(), + data = std::str::from_utf8(bytes).unwrap_or_default(), size = bytes.len() ); diff --git a/crates/main/Cargo.toml b/crates/main/Cargo.toml index b9cb48f0..9693299f 100644 --- a/crates/main/Cargo.toml +++ b/crates/main/Cargo.toml @@ -21,6 +21,7 @@ jmap = { path = "../jmap" } jmap_proto = { path = "../jmap-proto" } smtp = { path = "../smtp", features = ["local_delivery"] } imap = { path = "../imap" } +pop3 = { path = "../pop3" } managesieve = { path = "../managesieve" } common = { path = "../common" } directory = { path = "../directory" } diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index 9b796a29..31e93a61 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -31,6 +31,7 @@ use jmap::{ JMAP, }; use managesieve::core::ManageSieveSessionManager; +use pop3::Pop3SessionManager; use smtp::core::{SmtpSessionManager, SMTP}; use tokio::sync::mpsc; use utils::wait_for_shutdown; @@ -83,6 +84,12 @@ async fn main() -> std::io::Result<()> { acceptor, shutdown_rx, ), + ServerProtocol::Pop3 => server.spawn( + Pop3SessionManager::new(imap.clone()), + core.clone(), + acceptor, + shutdown_rx, + ), ServerProtocol::ManageSieve => server.spawn( ManageSieveSessionManager::new(imap.clone()), core.clone(), diff --git a/crates/managesieve/src/core/client.rs b/crates/managesieve/src/core/client.rs index e0a50144..ee4e6fe2 100644 --- a/crates/managesieve/src/core/client.rs +++ b/crates/managesieve/src/core/client.rs @@ -89,7 +89,7 @@ impl Session { self.write(&response).await?; } Err(err) => { - let disconnect = err.rtype == ResponseType::Bye; + let disconnect = matches!(err.rtype, ResponseType::Bye | ResponseType::Ok); self.write(&err.into_bytes()).await?; if disconnect { return Err(()); diff --git a/crates/managesieve/src/op/capability.rs b/crates/managesieve/src/op/capability.rs index 7495ac2b..f8d4a54c 100644 --- a/crates/managesieve/src/op/capability.rs +++ b/crates/managesieve/src/op/capability.rs @@ -37,7 +37,7 @@ impl Session { if self.stream.is_tls() || self.jmap.core.imap.allow_plain_auth { response.extend_from_slice(b"\"SASL\" \"PLAIN OAUTHBEARER\"\r\n"); } else { - response.extend_from_slice(b"\"SASL\" \"\"\r\n"); + response.extend_from_slice(b"\"SASL\" \"OAUTHBEARER\"\r\n"); }; if let Some(sieve) = self.jmap diff --git a/crates/pop3/Cargo.toml b/crates/pop3/Cargo.toml new file mode 100644 index 00000000..8f3dc764 --- /dev/null +++ b/crates/pop3/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "pop3" +version = "0.8.0" +edition = "2021" +resolver = "2" + +[dependencies] +store = { path = "../store" } +common = { path = "../common" } +jmap = { path = "../jmap" } +imap = { path = "../imap" } +utils = { path = "../utils" } +jmap_proto = { path = "../jmap-proto" } +mail-parser = { version = "0.9", features = ["full_encoding", "ludicrous_mode"] } +mail-send = { version = "0.4", default-features = false, features = ["cram-md5"] } +tracing = "0.1" +rustls = "0.22" +tokio = { version = "1.23", features = ["full"] } +tokio-rustls = { version = "0.25.0"} + +[features] +test_mode = [] diff --git a/crates/pop3/src/client.rs b/crates/pop3/src/client.rs new file mode 100644 index 00000000..cdc35abd --- /dev/null +++ b/crates/pop3/src/client.rs @@ -0,0 +1,241 @@ +/* + * Copyright (c) 2020-2022, 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 common::listener::SessionStream; +use mail_send::Credentials; + +use crate::{ + protocol::{request::Error, response::Response, Command, Mechanism}, + Session, State, +}; + +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") { + println!("<- {:?}", &line[..std::cmp::min(line.len(), 100)]); + }*/ + + let mut bytes = bytes.iter(); + let mut requests = Vec::with_capacity(2); + + loop { + match self.receiver.parse(&mut bytes) { + Ok(request) => { + // Group delete requests when possible + match (request, requests.last_mut()) { + (Command::Dele { msg }, Some(Ok(Command::DeleMany { msgs }))) => { + msgs.push(msg); + } + (Command::Dele { msg }, Some(Ok(Command::Dele { msg: other_msg }))) => { + let request = Ok(Command::DeleMany { + msgs: vec![*other_msg, msg], + }); + requests.pop(); + requests.push(request); + } + (request, _) => { + requests.push(Ok(request)); + } + } + } + Err(Error::NeedsMoreData) => { + break; + } + Err(Error::Parse(err)) => { + requests.push(Err(err)); + } + } + } + + for request in requests { + match request { + Ok(command) => match self.validate_request(command).await { + Ok(command) => match command { + Command::User { name } => { + if let State::NotAuthenticated { username, .. } = &mut self.state { + let response = format!("{name} is a valid mailbox"); + *username = Some(name); + self.write_ok(response).await?; + } else { + unreachable!(); + } + } + Command::Pass { string } => { + let username = + if let State::NotAuthenticated { username, .. } = &mut self.state { + username.take().unwrap() + } else { + unreachable!() + }; + self.handle_auth(Credentials::Plain { + username, + secret: string, + }) + .await?; + } + Command::Quit => { + self.handle_quit().await?; + } + Command::Stat => self.handle_stat().await?, + Command::List { msg } => { + self.handle_list(msg).await?; + } + Command::Retr { msg } => { + self.handle_fetch(msg, None).await?; + } + Command::Dele { msg } => self.handle_dele(vec![msg]).await?, + Command::DeleMany { msgs } => self.handle_dele(msgs).await?, + Command::Top { msg, n } => { + self.handle_fetch(msg, n.into()).await?; + } + Command::Uidl { msg } => self.handle_uidl(msg).await?, + Command::Noop => { + self.write_ok("NOOP").await?; + } + Command::Rset => { + self.handle_rset().await?; + } + Command::Capa => { + let mechanisms = + if self.stream.is_tls() || self.jmap.core.imap.allow_plain_auth { + vec![Mechanism::Plain, Mechanism::OAuthBearer] + } else { + vec![Mechanism::OAuthBearer] + }; + + self.write_bytes( + Response::Capability:: { + mechanisms, + stls: !self.stream.is_tls(), + } + .serialize(), + ) + .await?; + } + Command::Stls => { + self.write_ok("Begin TLS negotiation now").await?; + return Ok(false); + } + Command::Utf8 => { + self.write_ok("UTF8 enabled").await?; + } + Command::Auth { mechanism, params } => { + self.handle_sasl(mechanism, params).await?; + } + Command::Apop { .. } => { + self.write_err("APOP not supported.").await?; + } + }, + Err(err) => { + self.write_err(err).await?; + } + }, + Err(err) => { + self.write_err(err).await?; + } + } + } + + Ok(true) + } + + async fn validate_request( + &self, + command: Command, + ) -> Result, &'static str> { + match &command { + Command::Capa | Command::Quit | Command::Noop => Ok(command), + Command::Auth { + mechanism: Mechanism::Plain, + .. + } + | Command::User { .. } + | Command::Pass { .. } + | Command::Apop { .. } => { + if let State::NotAuthenticated { username, .. } = &self.state { + if self.stream.is_tls() || self.jmap.core.imap.allow_plain_auth { + if !matches!(command, Command::Pass { .. }) || username.is_some() { + Ok(command) + } else { + Err("Username was not provided.") + } + } else { + Err("Cannot authenticate over plain-text.") + } + } else { + Err("Already authenticated.") + } + } + Command::Auth { .. } => { + if let State::NotAuthenticated { .. } = &self.state { + Ok(command) + } else { + Err("Already authenticated.") + } + } + Command::Stls => { + if !self.stream.is_tls() { + Ok(command) + } else { + Err("Already in TLS mode.") + } + } + + Command::List { .. } + | Command::Retr { .. } + | Command::Dele { .. } + | Command::DeleMany { .. } + | Command::Top { .. } + | Command::Uidl { .. } + | Command::Utf8 + | Command::Stat + | Command::Rset => { + if let State::Authenticated { mailbox, .. } = &self.state { + if let Some(rate) = &self.jmap.core.imap.rate_requests { + match self + .jmap + .core + .storage + .lookup + .is_rate_allowed( + format!("ireq:{}", mailbox.account_id).as_bytes(), + rate, + true, + ) + .await + { + Ok(None) => Ok(command), + Ok(Some(_)) => Err("Too many requests"), + Err(_) => Err("Internal server error"), + } + } else { + Ok(command) + } + } else { + Err("Not authenticated.") + } + } + } + } +} diff --git a/crates/pop3/src/lib.rs b/crates/pop3/src/lib.rs new file mode 100644 index 00000000..00809b19 --- /dev/null +++ b/crates/pop3/src/lib.rs @@ -0,0 +1,88 @@ +/* + * Copyright (c) 2020-2022, 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, sync::Arc}; + +use common::listener::{limiter::InFlight, ServerInstance, SessionStream}; +use imap::core::{ImapInstance, Inner}; +use jmap::JMAP; +use mailbox::Mailbox; +use protocol::request::Parser; + +pub mod client; +pub mod mailbox; +pub mod op; +pub mod protocol; +pub mod session; + +static SERVER_GREETING: &str = "+OK Stalwart POP3 at your service.\r\n"; + +#[derive(Clone)] +pub struct Pop3SessionManager { + pub pop3: ImapInstance, +} + +impl Pop3SessionManager { + pub fn new(pop3: ImapInstance) -> Self { + Self { pop3 } + } +} + +pub struct Session { + pub jmap: JMAP, + pub imap: Arc, + pub instance: Arc, + pub receiver: Parser, + pub state: State, + pub stream: T, + pub in_flight: InFlight, + pub remote_addr: IpAddr, + pub span: tracing::Span, +} + +pub enum State { + NotAuthenticated { + auth_failures: u32, + username: Option, + }, + Authenticated { + mailbox: Mailbox, + in_flight: Option, + }, +} + +impl State { + pub fn mailbox(&self) -> &Mailbox { + match self { + State::Authenticated { mailbox, .. } => mailbox, + _ => unreachable!(), + } + } + + pub fn mailbox_mut(&mut self) -> &mut Mailbox { + match self { + State::Authenticated { mailbox, .. } => mailbox, + _ => unreachable!(), + } + } +} diff --git a/crates/pop3/src/mailbox.rs b/crates/pop3/src/mailbox.rs new file mode 100644 index 00000000..a640334a --- /dev/null +++ b/crates/pop3/src/mailbox.rs @@ -0,0 +1,184 @@ +/* + * Copyright (c) 2020-2022, 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::collections::BTreeMap; + +use common::listener::SessionStream; +use jmap::mailbox::{UidMailbox, INBOX_ID}; +use jmap_proto::{ + error::method::MethodError, + object::Object, + types::{collection::Collection, property::Property, value::Value}, +}; +use store::{ + ahash::AHashMap, write::key::DeserializeBigEndian, IndexKey, IterateParams, Serialize, U32_LEN, +}; + +use crate::Session; + +#[derive(Default)] +pub struct Mailbox { + pub messages: Vec, + pub account_id: u32, + pub uid_validity: u32, + pub total: u32, + pub size: u32, +} + +pub struct Message { + pub id: u32, + pub uid: u32, + pub size: u32, + pub deleted: bool, +} + +impl Session { + pub async fn fetch_mailbox(&self, account_id: u32) -> Result { + // Obtain message ids + let message_ids = self + .jmap + .get_tag( + account_id, + Collection::Email, + Property::MailboxIds, + INBOX_ID, + ) + .await? + .unwrap_or_default(); + + if message_ids.is_empty() { + return Ok(Mailbox::default()); + } + + let mut message_map = BTreeMap::new(); + let mut message_sizes = AHashMap::new(); + + // Obtain UID validity + self.jmap.mailbox_get_or_create(account_id).await?; + let uid_validity = self + .jmap + .get_property::>( + account_id, + Collection::Mailbox, + INBOX_ID, + &Property::Value, + ) + .await? + .and_then(|obj| obj.get(&Property::Cid).as_uint()) + .ok_or_else(|| { + tracing::debug!(event = "error", + context = "store", + account_id = account_id, + collection = ?Collection::Mailbox, + mailbox_id = INBOX_ID, + "Failed to obtain uid validity"); + MethodError::ServerPartialFail + }) + .map(|v| v as u32)?; + + // Obtain message sizes + self.jmap + .core + .storage + .data + .iterate( + IterateParams::new( + IndexKey { + account_id, + collection: Collection::Email.into(), + document_id: message_ids.min().unwrap(), + field: Property::Size.into(), + key: 0u32.serialize(), + }, + IndexKey { + account_id, + collection: Collection::Email.into(), + document_id: message_ids.max().unwrap(), + field: Property::Size.into(), + key: u32::MAX.serialize(), + }, + ) + .no_values(), + |key, _| { + let document_id = key.deserialize_be_u32(key.len() - U32_LEN)?; + if message_ids.contains(document_id) { + message_sizes.insert( + document_id, + key.deserialize_be_u32(key.len() - (U32_LEN * 2))?, + ); + } + + Ok(true) + }, + ) + .await + .map_err(|err| { + tracing::error!(context = "fetch_mailbox", + reason = ?err, + "Failed to iterate message sizes"); + + MethodError::ServerPartialFail + })?; + + // Sort by UID + for (message_id, uid_mailbox) in self + .jmap + .get_properties::, _, _>( + account_id, + Collection::Email, + &message_ids, + Property::MailboxIds, + ) + .await? + .into_iter() + { + // Make sure the message is still in Inbox + if let Some(item) = uid_mailbox.iter().find(|item| item.mailbox_id == INBOX_ID) { + debug_assert!(item.uid != 0, "UID is zero for message {item:?}"); + message_map.insert(item.uid, message_id); + } + } + + // Create mailbox + let mut mailbox = Mailbox { + messages: Vec::with_capacity(message_map.len()), + uid_validity, + account_id, + ..Default::default() + }; + for (uid, id) in message_map { + if let Some(size) = message_sizes.get(&id) { + mailbox.messages.push(Message { + id, + uid, + size: *size, + deleted: false, + }); + mailbox.total += 1; + mailbox.size += *size; + } + } + + Ok(mailbox) + } +} diff --git a/crates/pop3/src/op/authenticate.rs b/crates/pop3/src/op/authenticate.rs new file mode 100644 index 00000000..252d5f2e --- /dev/null +++ b/crates/pop3/src/op/authenticate.rs @@ -0,0 +1,214 @@ +/* + * Copyright (c) 2020-2022, 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 common::{ + listener::{limiter::ConcurrencyLimiter, SessionStream}, + AuthResult, +}; +use imap::op::authenticate::{decode_challenge_oauth, decode_challenge_plain}; +use jmap::auth::rate_limit::ConcurrencyLimiters; +use mail_parser::decoders::base64::base64_decode; +use mail_send::Credentials; +use std::sync::Arc; + +use crate::{ + protocol::{request, Command, Mechanism}, + Session, State, +}; + +impl Session { + pub async fn handle_sasl( + &mut self, + mechanism: Mechanism, + mut params: Vec, + ) -> Result<(), ()> { + match mechanism { + Mechanism::Plain | Mechanism::OAuthBearer => { + if !params.is_empty() { + let result = base64_decode(params.pop().unwrap().as_bytes()) + .ok_or("Failed to decode challenge.") + .and_then(|challenge| { + if mechanism == Mechanism::Plain { + decode_challenge_plain(&challenge) + } else { + decode_challenge_oauth(&challenge) + } + }); + + match result { + Ok(credentials) => self.handle_auth(credentials).await, + Err(err) => self.write_err(err).await, + } + } else { + // TODO: This hack is temporary until the SASL library is developed + self.receiver.state = request::State::Argument { + request: Command::Auth { + mechanism: mechanism.as_str().as_bytes().to_vec(), + params: vec![], + }, + num: 1, + last_is_space: true, + }; + + self.write_bytes("+\r\n").await + } + } + _ => { + self.write_err("Authentication mechanism not supported.") + .await + } + } + } + + pub async fn handle_auth(&mut self, credentials: Credentials) -> Result<(), ()> { + // Throttle authentication requests + if self + .jmap + .is_auth_allowed_soft(&self.remote_addr) + .await + .is_err() + { + tracing::debug!(parent: &self.span, + event = "disconnect", + "Too many authentication attempts, disconnecting.", + ); + + self.write_err("Too many authentication requests from this IP address.") + .await?; + return Err(()); + } + + // Authenticate + let access_token = match credentials { + Credentials::Plain { username, secret } | Credentials::XOauth2 { username, secret } => { + match self + .jmap + .authenticate_plain(&username, &secret, self.remote_addr) + .await + { + AuthResult::Success(token) => Some(token), + AuthResult::Failure => None, + AuthResult::Banned => { + self.write_err("Too many authentication requests from this IP address.") + .await?; + return Err(()); + } + } + } + Credentials::OAuthBearer { token } => { + match self + .jmap + .validate_access_token("access_token", &token) + .await + { + Ok((account_id, _, _)) => self.jmap.get_access_token(account_id).await, + Err(err) => { + tracing::debug!( + parent: &self.span, + context = "authenticate", + err = err, + "Failed to validate access token." + ); + None + } + } + } + }; + + if let Some(access_token) = access_token { + // Enforce concurrency limits + let in_flight = match self + .get_concurrency_limiter(access_token.primary_id()) + .map(|limiter| limiter.concurrent_requests.is_allowed()) + { + Some(Some(limiter)) => Some(limiter), + None => None, + Some(None) => { + tracing::debug!(parent: &self.span, + event = "disconnect", + "Too many concurrent connection.", + ); + self.write_err("Too many concurrent connections.").await?; + return Err(()); + } + }; + + // Cache access token + let access_token = Arc::new(access_token); + self.jmap.cache_access_token(access_token.clone()); + + // Fetch mailbox + match self.fetch_mailbox(access_token.primary_id()).await { + Ok(mailbox) => { + // Create session + self.state = State::Authenticated { in_flight, mailbox }; + + self.write_ok("Authentication successful").await + } + Err(_) => { + self.write_err("Temporary server failure").await?; + Err(()) + } + } + } else { + match &self.state { + State::NotAuthenticated { + auth_failures, + username, + } if *auth_failures < self.jmap.core.imap.max_auth_failures => { + self.state = State::NotAuthenticated { + auth_failures: auth_failures + 1, + username: username.clone(), + }; + self.write_err("Authentication failed").await + } + _ => { + tracing::debug!( + parent: &self.span, + event = "disconnect", + "Too many authentication failures, disconnecting.", + ); + self.write_err("Too many authentication failures").await?; + Err(()) + } + } + } + } + + pub fn get_concurrency_limiter(&self, account_id: u32) -> Option> { + let rate = self.jmap.core.imap.rate_concurrent?; + self.imap + .rate_limiter + .get(&account_id) + .map(|limiter| limiter.clone()) + .unwrap_or_else(|| { + let limiter = Arc::new(ConcurrencyLimiters { + concurrent_requests: ConcurrencyLimiter::new(rate), + concurrent_uploads: ConcurrencyLimiter::new(rate), + }); + self.imap.rate_limiter.insert(account_id, limiter.clone()); + limiter + }) + .into() + } +} diff --git a/crates/pop3/src/op/delete.rs b/crates/pop3/src/op/delete.rs new file mode 100644 index 00000000..bbde0416 --- /dev/null +++ b/crates/pop3/src/op/delete.rs @@ -0,0 +1,116 @@ +/* + * Copyright (c) 2020-2022, 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 common::listener::SessionStream; +use jmap_proto::types::{state::StateChange, type_state::DataType}; +use store::roaring::RoaringBitmap; + +use crate::{Session, State}; + +impl Session { + pub async fn handle_dele(&mut self, msgs: Vec) -> Result<(), ()> { + let mailbox = self.state.mailbox_mut(); + let mut response = Vec::new(); + + for msg in msgs { + if let Some(message) = mailbox.messages.get_mut(msg.saturating_sub(1) as usize) { + if !message.deleted { + response.extend_from_slice(format!("+OK message {msg} deleted\r\n").as_bytes()); + message.deleted = true; + } else { + response.extend_from_slice( + format!("-ERR message {msg} already deleted\r\n").as_bytes(), + ); + } + } else { + response.extend_from_slice("-ERR no such message\r\n".as_bytes()); + } + } + + self.write_bytes(response).await + } + + pub async fn handle_rset(&mut self) -> Result<(), ()> { + let mut count = 0; + let mailbox = self.state.mailbox_mut(); + for message in &mut mailbox.messages { + if message.deleted { + count += 1; + message.deleted = false; + } + } + self.write_ok(format!("{count} messages undeleted")).await + } + + pub async fn handle_quit(&mut self) -> Result<(), ()> { + if let State::Authenticated { mailbox, .. } = &self.state { + let mut deleted = RoaringBitmap::new(); + for message in &mailbox.messages { + if message.deleted { + deleted.insert(message.id); + } + } + + if !deleted.is_empty() { + let num_deleted = deleted.len(); + match self + .jmap + .emails_tombstone(mailbox.account_id, deleted) + .await + { + Ok((changes, not_deleted)) => { + if !changes.is_empty() { + if let Ok(change_id) = + self.jmap.commit_changes(mailbox.account_id, changes).await + { + self.jmap + .broadcast_state_change( + StateChange::new(mailbox.account_id) + .with_change(DataType::Email, change_id) + .with_change(DataType::Mailbox, change_id) + .with_change(DataType::Thread, change_id), + ) + .await; + } + } + if not_deleted.is_empty() { + self.write_ok(format!( + "Stalwart POP3 bids you farewell ({num_deleted} messages deleted)." + )) + .await?; + } else { + self.write_err("Some messages could not be deleted").await?; + } + } + Err(_) => { + self.write_err("Failed to delete messages").await?; + } + } + } + } else { + self.write_ok("Stalwart POP3 bids you farewell.").await?; + } + + Err(()) + } +} diff --git a/crates/pop3/src/op/fetch.rs b/crates/pop3/src/op/fetch.rs new file mode 100644 index 00000000..e06672f1 --- /dev/null +++ b/crates/pop3/src/op/fetch.rs @@ -0,0 +1,78 @@ +/* + * Copyright (c) 2020-2022, 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 common::listener::SessionStream; +use jmap::email::metadata::MessageMetadata; +use jmap_proto::types::{collection::Collection, property::Property}; +use store::write::Bincode; + +use crate::{protocol::response::Response, Session}; + +impl Session { + pub async fn handle_fetch(&mut self, msg: u32, lines: Option) -> Result<(), ()> { + let mailbox = self.state.mailbox(); + if let Some(message) = mailbox.messages.get(msg.saturating_sub(1) as usize) { + match self + .jmap + .get_property::>( + mailbox.account_id, + Collection::Email, + message.id, + &Property::BodyStructure, + ) + .await + { + Ok(Some(metadata)) => { + match self + .jmap + .get_blob(&metadata.inner.blob_hash, 0..usize::MAX) + .await + { + Ok(Some(bytes)) => { + self.write_bytes( + Response::Message:: { + bytes, + lines: lines.unwrap_or(0), + } + .serialize(), + ) + .await + } + _ => { + self.write_err( + "Failed to fetch message. Perhaps another session deleted it?", + ) + .await + } + } + } + _ => { + self.write_err("Failed to fetch message. Perhaps another session deleted it?") + .await + } + } + } else { + self.write_err("No such message").await + } + } +} diff --git a/crates/pop3/src/op/list.rs b/crates/pop3/src/op/list.rs new file mode 100644 index 00000000..e75adeaf --- /dev/null +++ b/crates/pop3/src/op/list.rs @@ -0,0 +1,75 @@ +/* + * Copyright (c) 2020-2022, 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 common::listener::SessionStream; + +use crate::{protocol::response::Response, Session}; + +impl Session { + pub async fn handle_list(&mut self, msg: Option) -> Result<(), ()> { + let mailbox = self.state.mailbox(); + if let Some(msg) = msg { + if let Some(message) = mailbox.messages.get(msg.saturating_sub(1) as usize) { + self.write_ok(format!("{} {}", msg, message.size)).await + } else { + self.write_err("No such message").await + } + } else { + self.write_bytes( + Response::List(mailbox.messages.iter().map(|m| m.size).collect::>()) + .serialize(), + ) + .await + } + } + + pub async fn handle_uidl(&mut self, msg: Option) -> Result<(), ()> { + let mailbox = self.state.mailbox(); + if let Some(msg) = msg { + if let Some(message) = mailbox.messages.get(msg.saturating_sub(1) as usize) { + self.write_ok(format!("{} {}{}", msg, mailbox.uid_validity, message.uid)) + .await + } else { + self.write_err("No such message").await + } + } else { + self.write_bytes( + Response::List( + mailbox + .messages + .iter() + .map(|m| format!("{}{}", mailbox.uid_validity, m.uid)) + .collect::>(), + ) + .serialize(), + ) + .await + } + } + + pub async fn handle_stat(&mut self) -> Result<(), ()> { + let mailbox = self.state.mailbox(); + self.write_ok(format!("{} {}", mailbox.total, mailbox.size)) + .await + } +} diff --git a/crates/pop3/src/op/mod.rs b/crates/pop3/src/op/mod.rs new file mode 100644 index 00000000..3399ab2a --- /dev/null +++ b/crates/pop3/src/op/mod.rs @@ -0,0 +1,27 @@ +/* + * Copyright (c) 2020-2022, 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. +*/ + +pub mod authenticate; +pub mod delete; +pub mod fetch; +pub mod list; diff --git a/crates/pop3/src/protocol/mod.rs b/crates/pop3/src/protocol/mod.rs new file mode 100644 index 00000000..7f6c07e8 --- /dev/null +++ b/crates/pop3/src/protocol/mod.rs @@ -0,0 +1,91 @@ +/* + * Copyright (c) 2020-2022, 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. +*/ + +pub mod request; +pub mod response; + +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub enum Command { + // Authorization state + User { + name: T, + }, + Pass { + string: T, + }, + Apop { + name: T, + digest: T, + }, + Quit, + + // Transaction state + Stat, + List { + msg: Option, + }, + Retr { + msg: u32, + }, + Dele { + msg: u32, + }, + DeleMany { + msgs: Vec, + }, + #[default] + Noop, + Rset, + Top { + msg: u32, + n: u32, + }, + Uidl { + msg: Option, + }, + + // Extensions + Capa, + Stls, + Utf8, + Auth { + mechanism: M, + params: Vec, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Mechanism { + Plain, + CramMd5, + DigestMd5, + ScramSha1, + ScramSha256, + Apop, + Ntlm, + Gssapi, + Anonymous, + External, + OAuthBearer, + XOauth2, +} diff --git a/crates/pop3/src/protocol/request.rs b/crates/pop3/src/protocol/request.rs new file mode 100644 index 00000000..0b70f0ce --- /dev/null +++ b/crates/pop3/src/protocol/request.rs @@ -0,0 +1,472 @@ +/* + * Copyright (c) 2020-2022, 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::borrow::Cow; + +use super::{Command, Mechanism}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Error { + NeedsMoreData, + Parse(Cow<'static, str>), +} + +#[derive(Default, Debug)] +pub enum State { + #[default] + Init, + Command { + buf: [u8; 4], + len: usize, + }, + Argument { + request: Command, Vec>, + num: usize, + last_is_space: bool, + }, + Error { + reason: Cow<'static, str>, + }, +} + +#[derive(Default)] +pub struct Parser { + pub state: State, +} + +const MAX_ARG_LEN: usize = 256; + +impl Parser { + pub fn parse( + &mut self, + bytes: &mut std::slice::Iter<'_, u8>, + ) -> Result, Error> { + for &byte in bytes { + match &mut self.state { + State::Init => match byte { + b' ' | b'\t' | b'\r' | b'\n' => {} + b'a'..=b'z' => { + self.state = State::Command { + buf: [byte, 0, 0, 0], + len: 1, + }; + } + b'A'..=b'Z' => { + self.state = State::Command { + buf: [byte | 0x20, 0, 0, 0], + len: 1, + }; + } + _ => { + self.state = State::Error { + reason: "Invalid command".into(), + }; + } + }, + State::Command { buf, len } => match byte { + b'a'..=b'z' | b'8' if *len < 4 => { + buf[*len] = byte; + *len += 1; + } + b'A'..=b'Z' if *len < 4 => { + buf[*len] = byte | 0x20; + *len += 1; + } + b' ' | b'\t' if *len == 4 || *len == 3 => match Command::parse(buf) { + Ok(request) => { + self.state = State::Argument { + request, + num: 0, + last_is_space: true, + }; + } + Err(err) => { + self.state = State::Error { reason: err }; + } + }, + b'\r' => {} + b'\n' if *len == 4 || *len == 3 => match Command::parse(buf) { + Ok(request) => { + self.state = State::Init; + return request.finalize(0); + } + Err(err) => { + self.state = State::Init; + return Err(Error::Parse(err)); + } + }, + _ => { + self.state = State::Error { + reason: "Invalid command".into(), + }; + } + }, + State::Argument { + request, + num, + last_is_space, + } => match byte { + b' ' | b'\t' => { + *last_is_space = true; + } + b'\r' => {} + b'\n' => { + let request = std::mem::take(request).finalize(*num); + self.state = State::Init; + return request; + } + _ => { + if *last_is_space { + *num += 1; + } + + match request.update_argument(*num, byte) { + Ok(_) => { + *last_is_space = false; + } + Err(err) => { + self.state = State::Error { reason: err }; + } + } + } + }, + State::Error { reason } => { + if byte == b'\n' { + let reason = std::mem::take(reason); + self.state = State::Init; + return Err(Error::Parse(reason)); + } + } + } + } + + Err(Error::NeedsMoreData) + } +} + +impl Command, Vec> { + pub fn parse(bytes: &[u8; 4]) -> Result> { + match (bytes[0], bytes[1], bytes[2], bytes[3]) { + (b'u', b's', b'e', b'r') => Ok(Self::User { name: Vec::new() }), + (b'u', b'i', b'd', b'l') => Ok(Self::Uidl { msg: None }), + (b'u', b't', b'f', b'8') => Ok(Self::Utf8), + (b'p', b'a', b's', b's') => Ok(Self::Pass { string: Vec::new() }), + (b'a', b'p', b'o', b'p') => Ok(Self::Apop { + name: Vec::new(), + digest: Vec::new(), + }), + (b'a', b'u', b't', b'h') => Ok(Self::Auth { + mechanism: Vec::new(), + params: Vec::new(), + }), + (b'q', b'u', b'i', b't') => Ok(Self::Quit), + (b'l', b'i', b's', b't') => Ok(Self::List { msg: None }), + (b'r', b'e', b't', b'r') => Ok(Self::Retr { msg: 0 }), + (b'r', b's', b'e', b't') => Ok(Self::Rset), + (b'd', b'e', b'l', b'e') => Ok(Self::Dele { msg: 0 }), + (b'n', b'o', b'o', b'p') => Ok(Self::Noop), + (b't', b'o', b'p', 0) => Ok(Self::Top { msg: 0, n: 0 }), + (b'c', b'a', b'p', b'a') => Ok(Self::Capa), + (b's', b't', b'l', b's') => Ok(Self::Stls), + (b's', b't', b'a', b't') => Ok(Self::Stat), + _ => Err("Invalid command".into()), + } + } + + pub fn update_argument(&mut self, arg_num: usize, byte: u8) -> Result<(), Cow<'static, str>> { + match self { + Command::User { name } if arg_num == 1 && name.len() < MAX_ARG_LEN => { + name.push(byte); + Ok(()) + } + Command::Pass { string } if arg_num == 1 && string.len() < MAX_ARG_LEN => { + string.push(byte); + Ok(()) + } + Command::Apop { name, digest } + if arg_num <= 2 && name.len() < MAX_ARG_LEN && digest.len() < MAX_ARG_LEN => + { + if arg_num == 1 { + name.push(byte); + } else { + digest.push(byte); + } + Ok(()) + } + Command::List { msg } if arg_num == 1 => add_digit(msg.get_or_insert(0), byte), + Command::Retr { msg } if arg_num == 1 => add_digit(msg, byte), + Command::Dele { msg } if arg_num == 1 => add_digit(msg, byte), + Command::Top { msg, n } if arg_num <= 2 => { + if arg_num == 1 { + add_digit(msg, byte) + } else { + add_digit(n, byte) + } + } + Command::Uidl { msg } if arg_num == 1 => add_digit(msg.get_or_insert(0), byte), + Command::Auth { mechanism, params } + if arg_num <= 4 + && mechanism.len() < 64 + && params.iter().map(|p| p.len()).sum::() < (MAX_ARG_LEN * 4) => + { + if arg_num == 1 { + mechanism.push(byte); + } else { + if params.len() < arg_num - 1 { + params.push(Vec::new()); + } + params.last_mut().unwrap().push(byte); + } + Ok(()) + } + _ => Err("Too many arguments".into()), + } + } + + pub fn finalize(self, num_args: usize) -> Result, Error> { + match self { + Command::User { name } if num_args == 1 => { + into_string(name).map(|name| Command::User { name }) + } + Command::Pass { string } if num_args == 1 => { + into_string(string).map(|string| Command::Pass { string }) + } + Command::Apop { name, digest } if num_args == 2 => { + let name = into_string(name)?; + let digest = into_string(digest)?; + Ok(Command::Apop { name, digest }) + } + Command::Quit => Ok(Command::Quit), + Command::Stat => Ok(Command::Stat), + Command::List { msg } => Ok(Command::List { msg }), + Command::Retr { msg } if num_args == 1 => Ok(Command::Retr { msg }), + Command::Dele { msg } if num_args == 1 => Ok(Command::Dele { msg }), + Command::Noop => Ok(Command::Noop), + Command::Rset => Ok(Command::Rset), + Command::Top { msg, n } if num_args == 2 => Ok(Command::Top { msg, n }), + Command::Uidl { msg } => Ok(Command::Uidl { msg }), + Command::Capa => Ok(Command::Capa), + Command::Stls => Ok(Command::Stls), + Command::Utf8 => Ok(Command::Utf8), + Command::Auth { mechanism, params } if num_args >= 1 => { + let mechanism = Mechanism::parse(&mechanism)?; + let params = params + .into_iter() + .map(into_string) + .collect::>()?; + + Ok(Command::Auth { mechanism, params }) + } + _ => Err(Error::Parse("Missing arguments".into())), + } + } +} + +#[inline(always)] +fn into_string(bytes: Vec) -> Result { + String::from_utf8(bytes).map_err(|_| Error::Parse("Invalid UTF-8".into())) +} + +#[inline(always)] +fn add_digit(num: &mut u32, byte: u8) -> Result<(), Cow<'static, str>> { + if byte.is_ascii_digit() { + *num = num + .checked_mul(10) + .and_then(|n| n.checked_add((byte - b'0') as u32)) + .ok_or("Numeric argument out of range")?; + Ok(()) + } else { + Err("Invalid digit".into()) + } +} + +impl Mechanism { + pub fn parse(value: &[u8]) -> Result { + if value.eq_ignore_ascii_case(b"PLAIN") { + Ok(Self::Plain) + } else if value.eq_ignore_ascii_case(b"CRAM-MD5") { + Ok(Self::CramMd5) + } else if value.eq_ignore_ascii_case(b"DIGEST-MD5") { + Ok(Self::DigestMd5) + } else if value.eq_ignore_ascii_case(b"SCRAM-SHA-1") { + Ok(Self::ScramSha1) + } else if value.eq_ignore_ascii_case(b"SCRAM-SHA-256") { + Ok(Self::ScramSha256) + } else if value.eq_ignore_ascii_case(b"APOP") { + Ok(Self::Apop) + } else if value.eq_ignore_ascii_case(b"NTLM") { + Ok(Self::Ntlm) + } else if value.eq_ignore_ascii_case(b"GSSAPI") { + Ok(Self::Gssapi) + } else if value.eq_ignore_ascii_case(b"ANONYMOUS") { + Ok(Self::Anonymous) + } else if value.eq_ignore_ascii_case(b"EXTERNAL") { + Ok(Self::External) + } else if value.eq_ignore_ascii_case(b"OAUTHBEARER") { + Ok(Self::OAuthBearer) + } else if value.eq_ignore_ascii_case(b"XOAUTH2") { + Ok(Self::XOauth2) + } else { + Err(Error::Parse( + format!( + "Unsupported mechanism '{}'.", + String::from_utf8_lossy(value) + ) + .into(), + )) + } + } +} + +#[cfg(test)] +mod tests { + use crate::protocol::{request::Error, Command, Mechanism}; + + use super::Parser; + + #[test] + fn parse_command() { + let mut parser = Parser::default(); + let mut chunked = String::new(); + let mut chunked_expected = Vec::new(); + + for (cmd, request) in [ + ("QuiT", Command::Quit), + (" \r\n NOOP ", Command::Noop), + ("STAT ", Command::Stat), + ("LIST ", Command::List { msg: None }), + (" list 100 ", Command::List { msg: 100.into() }), + ("retr 55", Command::Retr { msg: 55 }), + ("DELE 99", Command::Dele { msg: 99 }), + (" rset ", Command::Rset), + ("top 8000 1234", Command::Top { msg: 8000, n: 1234 }), + ("uidl", Command::Uidl { msg: None }), + ("uidl 000099999", Command::Uidl { msg: 99999.into() }), + ( + "USER test", + Command::User { + name: "test".to_string(), + }, + ), + ( + "PASS secret", + Command::Pass { + string: "secret".to_string(), + }, + ), + ( + "APOP mrose c4c9334bac560ecc979e58001b3e22fb", + Command::Apop { + name: "mrose".to_string(), + digest: "c4c9334bac560ecc979e58001b3e22fb".to_string(), + }, + ), + ("utf8", Command::Utf8), + ("capa", Command::Capa), + ( + "AUTH GSSAPI", + Command::Auth { + mechanism: Mechanism::Gssapi, + params: vec![], + }, + ), + ( + "AUTH PLAIN dGVzdAB0ZXN0AHRlc3Q=", + Command::Auth { + mechanism: Mechanism::Plain, + params: vec!["dGVzdAB0ZXN0AHRlc3Q=".to_string()], + }, + ), + ] { + assert_eq!( + parser.parse(&mut cmd.as_bytes().iter()), + Err(Error::NeedsMoreData) + ); + assert_eq!( + parser.parse(&mut b"\r\n".iter()), + Ok(request.clone()), + "{:?}", + cmd + ); + chunked.push_str(cmd); + chunked.push_str("\r\n"); + chunked_expected.push(request); + } + + for chunk_size in [1, 2, 4, 8, 16, 32, 64, 128, 256, 512] { + let mut parser = Parser::default(); + let mut requests = Vec::new(); + + for chunk in chunked.as_bytes().chunks(chunk_size) { + let mut chunk = chunk.iter(); + loop { + match parser.parse(&mut chunk) { + Ok(request) => { + requests.push(request); + } + Err(Error::NeedsMoreData) => break, + Err(err) => { + panic!("Unexpected error on chunk size {chunk_size}: {err:?}"); + } + } + } + } + + assert_eq!(requests, chunked_expected, "Chunk size: {}", chunk_size); + } + + for cmd in [ + "user", + "pass", + "user a b", + "pass c d", + "apop", + "apop a", + "apop a b c", + "quit 1", + "stat 1", + "list 1 2", + "retr", + "retr 1 2", + "dele", + "dele 1 2", + "noop 1", + "rset 1", + "top", + "top 1 2 3", + "uidl 1 2 3", + "capa 1", + "stls 1", + "utf8 1", + "auth", + "auth unknown", + ] { + assert_eq!( + parser.parse(&mut cmd.as_bytes().iter()), + Err(Error::NeedsMoreData) + ); + let result = parser.parse(&mut b"\r\n".iter()); + assert!(result.is_err(), "{:?}", result); + } + } +} diff --git a/crates/pop3/src/protocol/response.rs b/crates/pop3/src/protocol/response.rs new file mode 100644 index 00000000..7b806c74 --- /dev/null +++ b/crates/pop3/src/protocol/response.rs @@ -0,0 +1,220 @@ +/* + * Copyright (c) 2020-2022, 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::{borrow::Cow, fmt::Display}; + +use super::Mechanism; + +pub enum Response { + Ok(Cow<'static, str>), + Err(Cow<'static, str>), + List(Vec), + Message { + bytes: Vec, + lines: u32, + }, + Capability { + mechanisms: Vec, + stls: bool, + }, +} + +impl Response { + pub fn serialize(&self) -> Vec { + match self { + Response::Ok(message) => { + let mut buf = Vec::with_capacity(message.len() + 6); + buf.extend_from_slice(b"+OK "); + buf.extend_from_slice(message.as_bytes()); + buf.extend_from_slice(b"\r\n"); + buf + } + Response::Err(message) => { + let mut buf = Vec::with_capacity(message.len() + 6); + buf.extend_from_slice(b"-ERR "); + buf.extend_from_slice(message.as_bytes()); + buf.extend_from_slice(b"\r\n"); + buf + } + Response::List(octets) => { + let mut buf = Vec::with_capacity(octets.len() * 8 + 10); + buf.extend_from_slice(format!("+OK {} messages\r\n", octets.len()).as_bytes()); + for (num, octet) in octets.iter().enumerate() { + buf.extend_from_slice((num + 1).to_string().as_bytes()); + buf.extend_from_slice(b" "); + buf.extend_from_slice(octet.to_string().as_bytes()); + buf.extend_from_slice(b"\r\n"); + } + buf.extend_from_slice(b".\r\n"); + buf + } + Response::Message { bytes, lines } => { + let mut buf = Vec::with_capacity(bytes.len() + 10); + buf.extend_from_slice(b"+OK "); + buf.extend_from_slice(bytes.len().to_string().as_bytes()); + buf.extend_from_slice(b" octets\r\n"); + + let mut line_count = 0; + let mut last_byte = 0; + + // Transparency procedure + for &byte in bytes { + // POP3 requires that lines end with CRLF, do this check to ensure that + if byte == b'\n' && last_byte != b'\r' { + buf.push(b'\r'); + } + + if byte == b'.' && last_byte == b'\n' { + buf.push(b'.'); + } + buf.push(byte); + last_byte = byte; + + if *lines > 0 && byte == b'\n' { + line_count += 1; + if line_count == *lines { + break; + } + } + } + + if last_byte != b'\n' { + buf.extend_from_slice(b"\r\n"); + } + + buf.extend_from_slice(b".\r\n"); + buf + } + Response::Capability { mechanisms, stls } => { + let mut buf = Vec::with_capacity(256); + buf.extend_from_slice(b"+OK Capability list follows\r\n"); + if !mechanisms.is_empty() { + if mechanisms.contains(&Mechanism::Plain) { + buf.extend_from_slice(b"USER\r\n"); + } + buf.extend_from_slice(b"SASL"); + for mechanism in mechanisms { + buf.extend_from_slice(b" "); + buf.extend_from_slice(mechanism.as_str().as_bytes()); + } + buf.extend_from_slice(b"\r\n"); + } + + if *stls { + buf.extend_from_slice(b"STLS\r\n"); + } + + for capa in [ + "TOP", + "RESP-CODES", + "PIPELINING", + "EXPIRE NEVER", + "UIDL", + "UTF8", + "IMPLEMENTATION Stalwart Mail Server", + ] { + buf.extend_from_slice(capa.as_bytes()); + buf.extend_from_slice(b"\r\n"); + } + + buf.extend_from_slice(b".\r\n"); + buf + } + } + } +} + +impl Mechanism { + pub fn as_str(&self) -> &'static str { + match self { + Mechanism::Plain => "PLAIN", + Mechanism::CramMd5 => "CRAM-MD5", + Mechanism::DigestMd5 => "DIGEST-MD5", + Mechanism::ScramSha1 => "SCRAM-SHA-1", + Mechanism::ScramSha256 => "SCRAM-SHA-256", + Mechanism::Apop => "APOP", + Mechanism::Ntlm => "NTLM", + Mechanism::Gssapi => "GSSAPI", + Mechanism::Anonymous => "ANONYMOUS", + Mechanism::External => "EXTERNAL", + Mechanism::OAuthBearer => "OAUTHBEARER", + Mechanism::XOauth2 => "XOAUTH2", + } + } +} + +#[cfg(test)] +mod tests { + + use crate::protocol::Mechanism; + + use super::Response; + + #[test] + fn serialize_response() { + for (cmd, expected) in [ + ( + Response::Ok("message 1 deleted".into()), + "+OK message 1 deleted\r\n", + ), + ( + Response::Err("permission denied".into()), + "-ERR permission denied\r\n", + ), + ( + Response::List(vec![100, 200, 300]), + "+OK 3 messages\r\n1 100\r\n2 200\r\n3 300\r\n.\r\n", + ), + ( + Response::Capability { + mechanisms: vec![Mechanism::Plain, Mechanism::CramMd5], + stls: true, + }, + concat!( + "+OK Capability list follows\r\n", + "USER\r\n", + "SASL PLAIN CRAM-MD5\r\n", + "STLS\r\n", + "TOP\r\n", + "RESP-CODES\r\n", + "PIPELINING\r\n", + "EXPIRE NEVER\r\n", + "UIDL\r\n", + "UTF8\r\n", + "IMPLEMENTATION Stalwart Mail Server\r\n.\r\n" + ), + ), + ( + Response::Message { + bytes: "Subject: test\r\n\r\n.\r\ntest.\r\n.test\r\na" + .as_bytes() + .to_vec(), + lines: 0, + }, + "+OK 35 octets\r\nSubject: test\r\n\r\n..\r\ntest.\r\n..test\r\na\r\n.\r\n", + ), + ] { + assert_eq!(expected, String::from_utf8(cmd.serialize()).unwrap()); + } + } +} diff --git a/crates/pop3/src/session.rs b/crates/pop3/src/session.rs new file mode 100644 index 00000000..b79216aa --- /dev/null +++ b/crates/pop3/src/session.rs @@ -0,0 +1,179 @@ +/* + * Copyright (c) 2020-2022, 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::borrow::Cow; + +use common::listener::{SessionData, SessionManager, SessionStream}; +use jmap::JMAP; +use tokio_rustls::server::TlsStream; + +use crate::{ + protocol::{request::Parser, response::Response}, + Pop3SessionManager, Session, State, SERVER_GREETING, +}; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +impl SessionManager for Pop3SessionManager { + #[allow(clippy::manual_async_fn)] + fn handle( + self, + session: SessionData, + ) -> impl std::future::Future + Send { + async move { + let mut session = Session { + jmap: JMAP::from(self.pop3.jmap_instance), + imap: self.pop3.imap_inner, + instance: session.instance, + receiver: Parser::default(), + state: State::NotAuthenticated { + auth_failures: 0, + username: None, + }, + stream: session.stream, + in_flight: session.in_flight, + remote_addr: session.remote_ip, + span: session.span, + }; + + if session + .write_bytes(SERVER_GREETING.as_bytes()) + .await + .is_ok() + && session.handle_conn().await + && session.instance.acceptor.is_tls() + { + if let Ok(mut session) = session.into_tls().await { + session.handle_conn().await; + } + } + } + } + + #[allow(clippy::manual_async_fn)] + fn shutdown(&self) -> impl std::future::Future + Send { + async {} + } +} + +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(); + + loop { + tokio::select! { + result = tokio::time::timeout( + if !matches!(self.state, State::NotAuthenticated {..}) { + self.jmap.core.imap.timeout_auth + } else { + self.jmap.core.imap.timeout_unauth + }, + self.stream.read(&mut buf)) => { + match result { + Ok(Ok(bytes_read)) => { + if bytes_read > 0 { + match self.ingest(&buf[..bytes_read]).await { + Ok(true) => (), + Ok(false) => { + return true; + } + Err(_) => { + tracing::debug!(parent: &self.span, event = "disconnect", "Disconnecting client."); + break; + } + } + } else { + tracing::debug!(parent: &self.span, event = "close", "POP3 connection closed by client."); + break; + } + }, + Ok(Err(err)) => { + tracing::debug!(parent: &self.span, event = "error", reason = %err, "POP3 connection error."); + break; + }, + Err(_) => { + self.write_bytes(&b"-ERR Connection timed out.\r\n"[..]).await.ok(); + tracing::debug!(parent: &self.span, "POP3 connection timed out."); + break; + } + } + }, + _ = shutdown_rx.changed() => { + self.write_bytes(&b"* BYE Server shutting down.\r\n"[..]).await.ok(); + tracing::debug!(parent: &self.span, event = "shutdown", "POP3 server shutting down."); + break; + } + }; + } + + false + } + + pub async fn into_tls(self) -> Result>, ()> { + Ok(Session { + stream: self.instance.tls_accept(self.stream, &self.span).await?, + jmap: self.jmap, + imap: self.imap, + instance: self.instance, + receiver: self.receiver, + state: self.state, + span: self.span, + in_flight: self.in_flight, + remote_addr: self.remote_addr, + }) + } +} + +impl Session { + pub async fn write_bytes(&mut self, bytes: impl AsRef<[u8]>) -> Result<(), ()> { + let bytes = bytes.as_ref(); + /*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).unwrap_or_default(), + size = bytes.len() + ); + + if let Err(err) = self.stream.write_all(bytes.as_ref()).await { + tracing::trace!(parent: &self.span, "Failed to write to stream: {}", err); + Err(()) + } else { + let _ = self.stream.flush().await; + Ok(()) + } + } + + pub async fn write_ok(&mut self, message: impl Into>) -> Result<(), ()> { + self.write_bytes(Response::Ok::(message.into()).serialize()) + .await + } + + pub async fn write_err(&mut self, message: impl Into>) -> Result<(), ()> { + self.write_bytes(Response::Err::(message.into()).serialize()) + .await + } +} diff --git a/resources/config/config.toml b/resources/config/config.toml index dd680a65..f043265e 100644 --- a/resources/config/config.toml +++ b/resources/config/config.toml @@ -24,6 +24,15 @@ bind = ["[::]:993"] protocol = "imap" tls.implicit = true +[server.listener.pop3] +bind = "[::]:110" +protocol = "pop3" + +[server.listener.pop3s] +bind = "[::]:995" +protocol = "pop3" +tls.implicit = true + [server.listener."sieve"] bind = ["[::]:4190"] protocol = "managesieve" diff --git a/resources/webadmin.zip b/resources/webadmin.zip deleted file mode 100644 index 1b9db902..00000000 Binary files a/resources/webadmin.zip and /dev/null differ diff --git a/tests/Cargo.toml b/tests/Cargo.toml index f29f48a8..dbe7350c 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -24,6 +24,7 @@ jmap = { path = "../crates/jmap", features = ["test_mode"] } jmap_proto = { path = "../crates/jmap-proto" } imap = { path = "../crates/imap", features = ["test_mode"] } imap_proto = { path = "../crates/imap-proto" } +pop3 = { path = "../crates/pop3", features = ["test_mode"] } smtp = { path = "../crates/smtp", features = ["test_mode", "local_delivery"] } common = { path = "../crates/common", features = ["test_mode"] } managesieve = { path = "../crates/managesieve", features = ["test_mode"] } diff --git a/tests/src/imap/mod.rs b/tests/src/imap/mod.rs index 61894fc1..35504677 100644 --- a/tests/src/imap/mod.rs +++ b/tests/src/imap/mod.rs @@ -31,6 +31,7 @@ pub mod fetch; pub mod idle; pub mod mailbox; pub mod managesieve; +pub mod pop; pub mod search; pub mod store; pub mod thread; @@ -53,6 +54,7 @@ use directory::backend::internal::manage::ManageDirectory; use imap::core::{ImapSessionManager, Inner, IMAP}; use imap_proto::ResponseType; use jmap::{api::JmapSessionManager, services::IPC_CHANNEL_BUFFER, JMAP}; +use pop3::Pop3SessionManager; use smtp::core::{SmtpSessionManager, SMTP}; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines, ReadHalf, WriteHalf}, @@ -84,6 +86,12 @@ protocol = "managesieve" max-connections = 81920 tls.implicit = true +[server.listener.pop3] +bind = ["127.0.0.1:4110"] +protocol = "pop3" +max-connections = 81920 +tls.implicit = true + [server.listener.lmtp-debug] bind = ['127.0.0.1:11201'] greeting = 'Test LMTP instance' @@ -327,6 +335,12 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest { acceptor, shutdown_rx, ), + ServerProtocol::Pop3 => server.spawn( + Pop3SessionManager::new(imap.clone()), + shared_core.clone(), + acceptor, + shutdown_rx, + ), ServerProtocol::ManageSieve => server.spawn( ManageSieveSessionManager::new(imap.clone()), shared_core.clone(), @@ -358,6 +372,9 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest { lookup .create_test_user_with_email("foobar@example.com", "secret", "Bill Foobar") .await; + lookup + .create_test_user_with_email("popper@example.com", "secret", "Karl Popper") + .await; lookup .create_test_group_with_email("support@example.com", "Support Group") .await; @@ -388,7 +405,7 @@ pub async fn imap_tests() { .with_env_filter( tracing_subscriber::EnvFilter::builder() .parse( - format!("smtp={level},imap={level},jmap={level},store={level},utils={level},directory={level}"), + format!("smtp={level},imap={level},jmap={level},store={level},utils={level},common={level},pop3={level},directory={level}"), ) .unwrap(), ) @@ -453,6 +470,9 @@ pub async fn imap_tests() { // Run ManageSieve tests managesieve::test().await; + // Run POP3 tests + pop::test().await; + // Print elapsed time let elapsed = start_time.elapsed(); println!( diff --git a/tests/src/imap/pop.rs b/tests/src/imap/pop.rs new file mode 100644 index 00000000..e8cd9fa2 --- /dev/null +++ b/tests/src/imap/pop.rs @@ -0,0 +1,293 @@ +/* + * Copyright (c) 2020-2022, 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 mail_send::smtp::tls::build_tls_connector; +use rustls_pki_types::ServerName; +use std::time::Duration; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines, ReadHalf, WriteHalf}, + net::TcpStream, +}; +use tokio_rustls::client::TlsStream; + +use crate::{jmap::delivery::SmtpConnection, smtp::session::VerifyResponse}; + +pub async fn test() { + println!("Running POP3 tests..."); + + // Send 3 test emails + for i in 0..3 { + let mut lmtp = SmtpConnection::connect_port(11201).await; + lmtp.ingest( + "bill@example.com", + &["popper@example.com"], + &format!( + concat!( + "From: bill@example.com\r\n", + "To: popper@example.com\r\n", + "Subject: TPS Report {}\r\n", + "X-Spam-Status: No\r\n", + "\r\n", + "I'm going to need those TPS {} reports ASAP.\r\n", + "..\r\n", + "So, if you could do that, that'd be great." + ), + i, i + ), + ) + .await; + } + + // Connect to POP3 + let mut pop3 = Pop3Connection::connect().await; + pop3.assert_read(ResponseType::Ok).await; + + // Capabilities + pop3.send("CAPA").await; + pop3.assert_read(ResponseType::Multiline) + .await + .assert_contains("SASL PLAIN") + .assert_contains("IMPLEMENTATION"); + + // Noop + pop3.send("NOOP").await; + pop3.assert_read(ResponseType::Ok).await; + + // Authenticate user/pass + pop3.send("PASS secret").await; + pop3.assert_read(ResponseType::Err).await; + pop3.send("USER popper@example.com").await; + pop3.assert_read(ResponseType::Ok).await; + pop3.send("PASS wrong_secret").await; + pop3.assert_read(ResponseType::Err).await; + pop3.send("USER popper@example.com").await; + pop3.assert_read(ResponseType::Ok).await; + pop3.send("PASS secret").await; + pop3.assert_read(ResponseType::Ok).await; + pop3.send("QUIT").await; + + // Authenticate using AUTH PLAIN + let mut pop3 = Pop3Connection::connect().await; + pop3.assert_read(ResponseType::Ok).await; + pop3.send("AUTH PLAIN AHBvcHBlckBleGFtcGxlLmNvbQBzZWNyZXQ=") + .await; + pop3.assert_read(ResponseType::Ok).await; + + // STAT + pop3.send("STAT").await; + pop3.assert_read(ResponseType::Ok) + .await + .assert_contains("+OK 3 546"); + + // UTF8 + pop3.send("UTF8").await; + pop3.assert_read(ResponseType::Ok).await; + + // LIST + pop3.send("LIST").await; + pop3.assert_read(ResponseType::Multiline) + .await + .assert_contains("+OK 3 messages") + .assert_contains("1 182") + .assert_contains("2 182") + .assert_contains("3 182"); + pop3.send("LIST 2").await; + pop3.assert_read(ResponseType::Ok) + .await + .assert_contains("+OK 2 182"); + + // UIDL + pop3.send("UIDL").await; + pop3.assert_read(ResponseType::Multiline) + .await + .assert_contains("+OK 3 messages") + .assert_contains("1 ") + .assert_contains("2 ") + .assert_contains("3 "); + pop3.send("UIDL 2").await; + pop3.assert_read(ResponseType::Ok) + .await + .assert_contains("+OK 2 "); + + // RETR + pop3.send("RETR 1").await; + pop3.assert_read(ResponseType::Multiline) + .await + .assert_contains("+OK 182 octets") + .assert_contains("I'm going to need those TPS 0 reports ASAP.") + .assert_contains("So, if you could do that, that'd be great."); + pop3.send("RETR 3").await; + pop3.assert_read(ResponseType::Multiline) + .await + .assert_contains("+OK 182 octets") + .assert_contains("I'm going to need those TPS 2 reports ASAP.") + .assert_contains("So, if you could do that, that'd be great."); + pop3.send("RETR 4").await; + pop3.assert_read(ResponseType::Err).await; + + // TOP + pop3.send("TOP 1 4").await; + pop3.assert_read(ResponseType::Multiline) + .await + .assert_contains("+OK 182 octets") + .assert_contains("Subject: TPS Report 0") + .assert_not_contains("I'm going to need those TPS 0 reports ASAP."); + pop3.send("TOP 3 4").await; + pop3.assert_read(ResponseType::Multiline) + .await + .assert_contains("+OK 182 octets") + .assert_contains("Subject: TPS Report 2") + .assert_not_contains("I'm going to need those TPS 2 reports ASAP."); + + // DELE + RSET + QUIT (should not delete messages) + pop3.send("DELE 1").await; + pop3.assert_read(ResponseType::Ok).await; + pop3.send("DELE 4").await; + pop3.assert_read(ResponseType::Err).await; + pop3.send("RSET").await; + pop3.assert_read(ResponseType::Ok).await; + pop3.send("QUIT").await; + let mut pop3 = Pop3Connection::connect_and_login().await; + pop3.send("STAT").await; + pop3.assert_read(ResponseType::Ok) + .await + .assert_contains("+OK 3 546"); + + // DELE + QUIT (should delete messages) + pop3.send("DELE 2").await; + pop3.assert_read(ResponseType::Ok).await; + pop3.send("QUIT").await; + let mut pop3 = Pop3Connection::connect_and_login().await; + pop3.send("STAT").await; + pop3.assert_read(ResponseType::Ok) + .await + .assert_contains("+OK 2 364"); + pop3.send("TOP 1 4").await; + pop3.assert_read(ResponseType::Multiline) + .await + .assert_contains("TPS Report 0"); + pop3.send("TOP 2 4").await; + pop3.assert_read(ResponseType::Multiline) + .await + .assert_contains("TPS Report 2"); + + // DELE using pipelining + pop3.send("DELE 1\r\nDELE 2").await; + pop3.assert_read(ResponseType::Ok).await; + pop3.assert_read(ResponseType::Ok).await; + pop3.send("QUIT").await; + let mut pop3 = Pop3Connection::connect_and_login().await; + pop3.send("STAT").await; + pop3.assert_read(ResponseType::Ok) + .await + .assert_contains("+OK 0 0"); + pop3.send("QUIT").await; +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ResponseType { + Ok, + Multiline, + Err, +} + +pub struct Pop3Connection { + reader: Lines>>>, + writer: WriteHalf>, +} + +impl Pop3Connection { + pub async fn connect() -> Self { + let (reader, writer) = tokio::io::split( + build_tls_connector(true) + .connect( + ServerName::try_from("pop3.example.org").unwrap().to_owned(), + TcpStream::connect("127.0.0.1:4110").await.unwrap(), + ) + .await + .unwrap(), + ); + Pop3Connection { + reader: BufReader::new(reader).lines(), + writer, + } + } + + pub async fn connect_and_login() -> Self { + let mut pop3 = Self::connect().await; + pop3.assert_read(ResponseType::Ok).await; + pop3.send("AUTH PLAIN AHBvcHBlckBleGFtcGxlLmNvbQBzZWNyZXQ=") + .await; + pop3.assert_read(ResponseType::Ok).await; + pop3 + } + + pub async fn assert_read(&mut self, rt: ResponseType) -> Vec { + let lines = self.read(matches!(rt, ResponseType::Multiline)).await; + if lines.last().unwrap().starts_with(match rt { + ResponseType::Ok => "+OK", + ResponseType::Multiline => ".", + ResponseType::Err => "-ERR", + }) { + lines + } else { + panic!("Expected {:?} from server but got: {:?}", rt, lines); + } + } + + pub async fn read(&mut self, is_multiline: bool) -> Vec { + let mut lines = Vec::new(); + loop { + match tokio::time::timeout(Duration::from_millis(1500), self.reader.next_line()).await { + Ok(Ok(Some(line))) => { + let is_done = (!is_multiline && line.starts_with("+OK")) + || (is_multiline && line == ".") + || line.starts_with("-ERR"); + //let c = println!("<- {:?}", line); + lines.push(line); + if is_done { + return lines; + } + } + Ok(Ok(None)) => { + panic!("Invalid response: {:?}.", lines); + } + Ok(Err(err)) => { + panic!("Connection broken: {} ({:?})", err, lines); + } + Err(_) => panic!("Timeout while waiting for server response: {:?}", lines), + } + } + } + + pub async fn send(&mut self, text: &str) { + //let c = println!("-> {:?}", text); + self.writer.write_all(text.as_bytes()).await.unwrap(); + self.writer.write_all(b"\r\n").await.unwrap(); + } + + pub async fn send_raw(&mut self, text: &str) { + //let c = println!("-> {:?}", text); + self.writer.write_all(text.as_bytes()).await.unwrap(); + } +} diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 865e730f..952ad976 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -42,6 +42,7 @@ use jmap::{ use jmap_client::client::{Client, Credentials}; use jmap_proto::{error::request::RequestError, types::id::Id}; use managesieve::core::ManageSieveSessionManager; +use pop3::Pop3SessionManager; use reqwest::header; use serde::{de::DeserializeOwned, Deserialize, Serialize}; use smtp::core::{SmtpSessionManager, SMTP}; @@ -518,6 +519,12 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest { acceptor, shutdown_rx, ), + ServerProtocol::Pop3 => server.spawn( + Pop3SessionManager::new(imap.clone()), + shared_core.clone(), + acceptor, + shutdown_rx, + ), ServerProtocol::ManageSieve => server.spawn( ManageSieveSessionManager::new(imap.clone()), shared_core.clone(), diff --git a/tests/src/smtp/outbound/mod.rs b/tests/src/smtp/outbound/mod.rs index 577345f2..602ee5bc 100644 --- a/tests/src/smtp/outbound/mod.rs +++ b/tests/src/smtp/outbound/mod.rs @@ -174,7 +174,7 @@ impl TestServer { acceptor, shutdown_rx, ), - ServerProtocol::Imap | ServerProtocol::ManageSieve => { + ServerProtocol::Imap | ServerProtocol::Pop3 | ServerProtocol::ManageSieve => { unreachable!() } };