1629 lines
55 KiB
Rust
1629 lines
55 KiB
Rust
//! Per-connection SMTP handler.
|
|
//!
|
|
//! Manages the read/write loop for a single SMTP connection.
|
|
//! Dispatches parsed commands, handles DATA mode, and manages
|
|
//! authentication flow.
|
|
|
|
use crate::command::{parse_command, AuthMechanism, ParseError, SmtpCommand};
|
|
use crate::config::SmtpServerConfig;
|
|
use crate::data::{DataAccumulator, DataAction};
|
|
use crate::rate_limiter::RateLimiter;
|
|
use crate::response::{build_capabilities, SmtpResponse};
|
|
use crate::scram::{ScramCredentials, ScramServer};
|
|
use crate::session::{AuthState, SmtpSession};
|
|
use crate::validation;
|
|
|
|
use base64::Engine;
|
|
use base64::engine::general_purpose::STANDARD as BASE64;
|
|
use hickory_resolver::TokioResolver;
|
|
use mailer_security::MessageAuthenticator;
|
|
use serde::{Deserialize, Serialize};
|
|
use std::net::IpAddr;
|
|
use std::sync::Arc;
|
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
|
use tokio::net::TcpStream;
|
|
use tokio::sync::{mpsc, oneshot};
|
|
use tokio::time::{timeout, Duration};
|
|
use tokio_rustls::server::TlsStream;
|
|
use tracing::{debug, info, warn};
|
|
|
|
/// Events emitted by a connection handler to the server.
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
pub enum ConnectionEvent {
|
|
/// A complete email has been received and needs processing.
|
|
EmailReceived {
|
|
correlation_id: String,
|
|
session_id: String,
|
|
mail_from: String,
|
|
rcpt_to: Vec<String>,
|
|
/// Base64-encoded raw message for inline, or file path for large messages.
|
|
data: EmailData,
|
|
remote_addr: String,
|
|
client_hostname: Option<String>,
|
|
secure: bool,
|
|
authenticated_user: Option<String>,
|
|
/// In-process security results (DKIM, SPF, DMARC, content scan).
|
|
security_results: Option<serde_json::Value>,
|
|
},
|
|
/// An authentication request that needs TS validation.
|
|
AuthRequest {
|
|
correlation_id: String,
|
|
session_id: String,
|
|
username: String,
|
|
password: String,
|
|
remote_addr: String,
|
|
},
|
|
/// A SCRAM credential request — Rust needs stored credentials from TS.
|
|
ScramCredentialRequest {
|
|
correlation_id: String,
|
|
session_id: String,
|
|
username: String,
|
|
remote_addr: String,
|
|
},
|
|
}
|
|
|
|
/// How email data is transported from Rust to TS.
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
#[serde(tag = "type")]
|
|
pub enum EmailData {
|
|
/// Inline base64-encoded data (for messages <= 256KB).
|
|
#[serde(rename = "inline")]
|
|
Inline { base64: String },
|
|
/// Path to a temp file containing the raw message (for large messages).
|
|
#[serde(rename = "file")]
|
|
File { path: String },
|
|
}
|
|
|
|
/// Result of TS processing an email.
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct EmailProcessingResult {
|
|
pub accepted: bool,
|
|
pub smtp_code: Option<u16>,
|
|
pub smtp_message: Option<String>,
|
|
}
|
|
|
|
/// Result of TS processing an auth request.
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct AuthResult {
|
|
pub success: bool,
|
|
pub message: Option<String>,
|
|
}
|
|
|
|
/// Result of TS returning SCRAM credentials for a user.
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct ScramCredentialResult {
|
|
pub found: bool,
|
|
pub salt: Option<Vec<u8>>,
|
|
pub iterations: Option<u32>,
|
|
pub stored_key: Option<Vec<u8>>,
|
|
pub server_key: Option<Vec<u8>>,
|
|
}
|
|
|
|
/// Abstraction over plain and TLS streams.
|
|
pub enum SmtpStream {
|
|
Plain(BufReader<TcpStream>),
|
|
Tls(BufReader<TlsStream<TcpStream>>),
|
|
}
|
|
|
|
impl SmtpStream {
|
|
/// Read a line from the stream (up to max_line_length bytes).
|
|
pub async fn read_line(&mut self, buf: &mut String, max_len: usize) -> std::io::Result<usize> {
|
|
match self {
|
|
SmtpStream::Plain(reader) => {
|
|
let result = reader.read_line(buf).await?;
|
|
if buf.len() > max_len {
|
|
buf.truncate(max_len);
|
|
}
|
|
Ok(result)
|
|
}
|
|
SmtpStream::Tls(reader) => {
|
|
let result = reader.read_line(buf).await?;
|
|
if buf.len() > max_len {
|
|
buf.truncate(max_len);
|
|
}
|
|
Ok(result)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Read bytes from the stream into a buffer.
|
|
pub async fn read_chunk(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
|
use tokio::io::AsyncReadExt;
|
|
match self {
|
|
SmtpStream::Plain(reader) => reader.read(buf).await,
|
|
SmtpStream::Tls(reader) => reader.read(buf).await,
|
|
}
|
|
}
|
|
|
|
/// Write bytes to the stream.
|
|
pub async fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()> {
|
|
match self {
|
|
SmtpStream::Plain(reader) => reader.get_mut().write_all(buf).await,
|
|
SmtpStream::Tls(reader) => reader.get_mut().write_all(buf).await,
|
|
}
|
|
}
|
|
|
|
/// Flush the write buffer.
|
|
pub async fn flush(&mut self) -> std::io::Result<()> {
|
|
match self {
|
|
SmtpStream::Plain(reader) => reader.get_mut().flush().await,
|
|
SmtpStream::Tls(reader) => reader.get_mut().flush().await,
|
|
}
|
|
}
|
|
|
|
/// Check if the internal buffer has unread data (pipelined commands).
|
|
pub fn has_buffered_data(&self) -> bool {
|
|
match self {
|
|
SmtpStream::Plain(reader) => !reader.buffer().is_empty(),
|
|
SmtpStream::Tls(reader) => !reader.buffer().is_empty(),
|
|
}
|
|
}
|
|
|
|
/// Unwrap to get the raw TcpStream for STARTTLS upgrade.
|
|
/// Only works on Plain streams.
|
|
pub fn into_tcp_stream(self) -> Option<TcpStream> {
|
|
match self {
|
|
SmtpStream::Plain(reader) => Some(reader.into_inner()),
|
|
SmtpStream::Tls(_) => None,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Handle a single SMTP connection.
|
|
///
|
|
/// This is the main entry point spawned for each incoming connection.
|
|
pub async fn handle_connection(
|
|
mut stream: SmtpStream,
|
|
config: Arc<SmtpServerConfig>,
|
|
rate_limiter: Arc<RateLimiter>,
|
|
event_tx: mpsc::Sender<ConnectionEvent>,
|
|
callback_register: Arc<dyn CallbackRegistry + Send + Sync>,
|
|
tls_acceptor: Option<Arc<tokio_rustls::TlsAcceptor>>,
|
|
remote_addr: String,
|
|
is_secure: bool,
|
|
authenticator: Arc<MessageAuthenticator>,
|
|
resolver: Arc<TokioResolver>,
|
|
) {
|
|
let mut session = SmtpSession::new(remote_addr.clone(), is_secure);
|
|
|
|
// Check IP rate limit
|
|
if !rate_limiter.check_connection(&remote_addr) {
|
|
let resp = SmtpResponse::service_unavailable(
|
|
&config.hostname,
|
|
"Too many connections from your IP",
|
|
);
|
|
let _ = stream.write_all(&resp.to_bytes()).await;
|
|
let _ = stream.flush().await;
|
|
info!(
|
|
session_id = %session.id,
|
|
remote_addr = %remote_addr,
|
|
"Connection rejected: rate limit exceeded"
|
|
);
|
|
return;
|
|
}
|
|
|
|
// Send greeting
|
|
let greeting = SmtpResponse::greeting(&config.hostname);
|
|
if stream.write_all(&greeting.to_bytes()).await.is_err() {
|
|
return;
|
|
}
|
|
if stream.flush().await.is_err() {
|
|
return;
|
|
}
|
|
|
|
let socket_timeout = Duration::from_secs(config.socket_timeout_secs);
|
|
|
|
loop {
|
|
let mut line = String::new();
|
|
let read_result = timeout(socket_timeout, stream.read_line(&mut line, 4096)).await;
|
|
|
|
match read_result {
|
|
Err(_) => {
|
|
// Timeout
|
|
let resp = SmtpResponse::service_unavailable(
|
|
&config.hostname,
|
|
"Connection timed out",
|
|
);
|
|
let _ = stream.write_all(&resp.to_bytes()).await;
|
|
let _ = stream.flush().await;
|
|
debug!(session_id = %session.id, "Connection timed out");
|
|
break;
|
|
}
|
|
Ok(Err(e)) => {
|
|
debug!(session_id = %session.id, error = %e, "Read error");
|
|
break;
|
|
}
|
|
Ok(Ok(0)) => {
|
|
debug!(session_id = %session.id, "Client disconnected");
|
|
break;
|
|
}
|
|
Ok(Ok(_)) => {
|
|
// Process the first command
|
|
let response = process_line(
|
|
&line,
|
|
&mut session,
|
|
&mut stream,
|
|
&config,
|
|
&rate_limiter,
|
|
&event_tx,
|
|
callback_register.as_ref(),
|
|
&tls_acceptor,
|
|
&authenticator,
|
|
&resolver,
|
|
)
|
|
.await;
|
|
|
|
// Check for pipelined commands in the buffer.
|
|
// Collect pipelinable responses into a batch for single write.
|
|
let mut response_batch: Vec<u8> = Vec::new();
|
|
let mut should_break = false;
|
|
let mut starttls_signal = false;
|
|
|
|
match response {
|
|
LineResult::Response(resp) => {
|
|
response_batch.extend_from_slice(&resp.to_bytes());
|
|
}
|
|
LineResult::Quit(resp) => {
|
|
let _ = stream.write_all(&resp.to_bytes()).await;
|
|
let _ = stream.flush().await;
|
|
should_break = true;
|
|
}
|
|
LineResult::StartTlsSignal => {
|
|
starttls_signal = true;
|
|
}
|
|
LineResult::NoResponse => {}
|
|
LineResult::Disconnect => {
|
|
should_break = true;
|
|
}
|
|
}
|
|
|
|
if should_break {
|
|
break;
|
|
}
|
|
|
|
// Process additional pipelined commands from the buffer
|
|
if !starttls_signal {
|
|
while stream.has_buffered_data() {
|
|
let mut next_line = String::new();
|
|
match stream.read_line(&mut next_line, 4096).await {
|
|
Ok(0) | Err(_) => break,
|
|
Ok(_) => {
|
|
let next_response = process_line(
|
|
&next_line,
|
|
&mut session,
|
|
&mut stream,
|
|
&config,
|
|
&rate_limiter,
|
|
&event_tx,
|
|
callback_register.as_ref(),
|
|
&tls_acceptor,
|
|
&authenticator,
|
|
&resolver,
|
|
)
|
|
.await;
|
|
|
|
match next_response {
|
|
LineResult::Response(resp) => {
|
|
response_batch.extend_from_slice(&resp.to_bytes());
|
|
}
|
|
LineResult::Quit(resp) => {
|
|
response_batch.extend_from_slice(&resp.to_bytes());
|
|
should_break = true;
|
|
break;
|
|
}
|
|
LineResult::StartTlsSignal | LineResult::Disconnect => {
|
|
// Non-pipelinable: flush batch and handle
|
|
starttls_signal = matches!(next_response, LineResult::StartTlsSignal);
|
|
should_break = matches!(next_response, LineResult::Disconnect);
|
|
break;
|
|
}
|
|
LineResult::NoResponse => {}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Flush the accumulated response batch in one write
|
|
if !response_batch.is_empty() {
|
|
if stream.write_all(&response_batch).await.is_err() {
|
|
break;
|
|
}
|
|
if stream.flush().await.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
|
|
if should_break {
|
|
break;
|
|
}
|
|
|
|
if starttls_signal {
|
|
// Send 220 Ready response
|
|
let resp = SmtpResponse::new(220, "Ready to start TLS");
|
|
if stream.write_all(&resp.to_bytes()).await.is_err() {
|
|
break;
|
|
}
|
|
if stream.flush().await.is_err() {
|
|
break;
|
|
}
|
|
// Extract TCP stream and upgrade
|
|
if let Some(tcp_stream) = stream.into_tcp_stream() {
|
|
if let Some(acceptor) = &tls_acceptor {
|
|
match acceptor.accept(tcp_stream).await {
|
|
Ok(tls_stream) => {
|
|
stream = SmtpStream::Tls(BufReader::new(tls_stream));
|
|
session.secure = true;
|
|
session.state = crate::state::SmtpState::Connected;
|
|
session.client_hostname = None;
|
|
session.esmtp = false;
|
|
session.auth_state = AuthState::None;
|
|
session.envelope = Default::default();
|
|
debug!(session_id = %session.id, "TLS upgrade successful");
|
|
}
|
|
Err(e) => {
|
|
warn!(session_id = %session.id, error = %e, "TLS handshake failed");
|
|
break;
|
|
}
|
|
}
|
|
} else {
|
|
break;
|
|
}
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
info!(
|
|
session_id = %session.id,
|
|
remote_addr = %remote_addr,
|
|
messages = session.message_count,
|
|
"Connection closed"
|
|
);
|
|
}
|
|
|
|
/// Result of processing a single input line.
|
|
enum LineResult {
|
|
/// Send this response to the client.
|
|
Response(SmtpResponse),
|
|
/// Send this response then close the connection.
|
|
Quit(SmtpResponse),
|
|
/// Signal that STARTTLS should be performed (main loop sends 220 and upgrades).
|
|
StartTlsSignal,
|
|
/// No response needed (handled internally).
|
|
NoResponse,
|
|
/// Disconnect immediately.
|
|
Disconnect,
|
|
}
|
|
|
|
/// Trait for registering and resolving correlation-ID callbacks.
|
|
pub trait CallbackRegistry: Send + Sync {
|
|
/// Register a callback for email processing and return a receiver.
|
|
fn register_email_callback(
|
|
&self,
|
|
correlation_id: &str,
|
|
) -> oneshot::Receiver<EmailProcessingResult>;
|
|
|
|
/// Register a callback for auth and return a receiver.
|
|
fn register_auth_callback(
|
|
&self,
|
|
correlation_id: &str,
|
|
) -> oneshot::Receiver<AuthResult>;
|
|
|
|
/// Register a callback for SCRAM credential lookup and return a receiver.
|
|
fn register_scram_callback(
|
|
&self,
|
|
correlation_id: &str,
|
|
) -> oneshot::Receiver<ScramCredentialResult>;
|
|
}
|
|
|
|
/// Process a single input line from the client.
|
|
async fn process_line(
|
|
line: &str,
|
|
session: &mut SmtpSession,
|
|
stream: &mut SmtpStream,
|
|
config: &SmtpServerConfig,
|
|
rate_limiter: &RateLimiter,
|
|
event_tx: &mpsc::Sender<ConnectionEvent>,
|
|
callback_registry: &dyn CallbackRegistry,
|
|
tls_acceptor: &Option<Arc<tokio_rustls::TlsAcceptor>>,
|
|
authenticator: &Arc<MessageAuthenticator>,
|
|
resolver: &Arc<TokioResolver>,
|
|
) -> LineResult {
|
|
// Handle AUTH intermediate states (waiting for username/password)
|
|
match &session.auth_state {
|
|
AuthState::WaitingForUsername => {
|
|
return handle_auth_username(line.trim(), session);
|
|
}
|
|
AuthState::WaitingForPassword { .. } => {
|
|
return handle_auth_password(
|
|
line.trim(),
|
|
session,
|
|
config,
|
|
rate_limiter,
|
|
event_tx,
|
|
callback_registry,
|
|
)
|
|
.await;
|
|
}
|
|
_ => {}
|
|
}
|
|
|
|
// Parse SMTP command
|
|
let cmd = match parse_command(line) {
|
|
Ok(cmd) => cmd,
|
|
Err(ParseError::Empty) => return LineResult::NoResponse,
|
|
Err(_) => {
|
|
if session.record_invalid_command() {
|
|
return LineResult::Quit(SmtpResponse::service_unavailable(
|
|
&config.hostname,
|
|
"Too many invalid commands",
|
|
));
|
|
}
|
|
return LineResult::Response(SmtpResponse::syntax_error());
|
|
}
|
|
};
|
|
|
|
match cmd {
|
|
SmtpCommand::Ehlo(hostname) => handle_ehlo(hostname, true, session, config),
|
|
SmtpCommand::Helo(hostname) => handle_ehlo(hostname, false, session, config),
|
|
|
|
SmtpCommand::MailFrom { address, params } => {
|
|
handle_mail_from(address, params, session, config, rate_limiter)
|
|
}
|
|
|
|
SmtpCommand::RcptTo { address, params } => {
|
|
handle_rcpt_to(address, params, session, config)
|
|
}
|
|
|
|
SmtpCommand::Data => {
|
|
handle_data(session, stream, config, event_tx, callback_registry, authenticator, resolver).await
|
|
}
|
|
|
|
SmtpCommand::Rset => {
|
|
session.reset_transaction();
|
|
LineResult::Response(SmtpResponse::ok("OK"))
|
|
}
|
|
|
|
SmtpCommand::Noop => LineResult::Response(SmtpResponse::ok("OK")),
|
|
|
|
SmtpCommand::Quit => {
|
|
LineResult::Quit(SmtpResponse::closing(&config.hostname))
|
|
}
|
|
|
|
SmtpCommand::StartTls => {
|
|
handle_starttls(session, config, tls_acceptor)
|
|
}
|
|
|
|
SmtpCommand::Auth {
|
|
mechanism,
|
|
initial_response,
|
|
} => {
|
|
if matches!(mechanism, AuthMechanism::ScramSha256) {
|
|
handle_auth_scram(
|
|
initial_response,
|
|
session,
|
|
stream,
|
|
config,
|
|
rate_limiter,
|
|
event_tx,
|
|
callback_registry,
|
|
)
|
|
.await
|
|
} else {
|
|
handle_auth(
|
|
mechanism,
|
|
initial_response,
|
|
session,
|
|
config,
|
|
rate_limiter,
|
|
event_tx,
|
|
callback_registry,
|
|
)
|
|
.await
|
|
}
|
|
}
|
|
|
|
SmtpCommand::Help(_) => {
|
|
LineResult::Response(SmtpResponse::new(
|
|
214,
|
|
"EHLO HELO MAIL RCPT DATA RSET NOOP QUIT STARTTLS AUTH HELP VRFY",
|
|
))
|
|
}
|
|
|
|
SmtpCommand::Vrfy(_) => {
|
|
LineResult::Response(SmtpResponse::new(252, "Cannot VRFY user"))
|
|
}
|
|
|
|
SmtpCommand::Expn(_) => {
|
|
LineResult::Response(SmtpResponse::not_implemented())
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Handle EHLO/HELO command.
|
|
fn handle_ehlo(
|
|
hostname: String,
|
|
esmtp: bool,
|
|
session: &mut SmtpSession,
|
|
config: &SmtpServerConfig,
|
|
) -> LineResult {
|
|
if !validation::is_valid_ehlo_hostname(&hostname) {
|
|
return LineResult::Response(SmtpResponse::param_error(
|
|
"Invalid hostname",
|
|
));
|
|
}
|
|
|
|
session.reset_for_ehlo(hostname, esmtp);
|
|
|
|
if esmtp {
|
|
let caps = build_capabilities(
|
|
config.max_message_size,
|
|
config.has_tls(),
|
|
session.secure,
|
|
config.auth_enabled,
|
|
);
|
|
LineResult::Response(SmtpResponse::ehlo_response(&config.hostname, &caps))
|
|
} else {
|
|
LineResult::Response(SmtpResponse::ok(format!(
|
|
"{} Hello",
|
|
config.hostname
|
|
)))
|
|
}
|
|
}
|
|
|
|
/// Handle MAIL FROM command.
|
|
fn handle_mail_from(
|
|
address: String,
|
|
params: std::collections::HashMap<String, Option<String>>,
|
|
session: &mut SmtpSession,
|
|
config: &SmtpServerConfig,
|
|
rate_limiter: &RateLimiter,
|
|
) -> LineResult {
|
|
if !session.state.can_mail_from() {
|
|
return LineResult::Response(SmtpResponse::bad_sequence(
|
|
"Send EHLO/HELO first",
|
|
));
|
|
}
|
|
|
|
if !validation::is_valid_smtp_address(&address) {
|
|
return LineResult::Response(SmtpResponse::param_error(
|
|
"Invalid sender address",
|
|
));
|
|
}
|
|
|
|
// Check SIZE param
|
|
if let Some(Some(size_str)) = params.get("SIZE") {
|
|
match validation::validate_size_param(size_str, config.max_message_size) {
|
|
Ok(_) => {}
|
|
Err(msg) => return LineResult::Response(SmtpResponse::new(552, msg)),
|
|
}
|
|
}
|
|
|
|
// Rate limit check for sender
|
|
if !address.is_empty() && !rate_limiter.check_message(&address) {
|
|
return LineResult::Response(SmtpResponse::temp_failure(
|
|
"Too many messages from this sender, try again later",
|
|
));
|
|
}
|
|
|
|
session.envelope.mail_from = address;
|
|
session.envelope.declared_size = params
|
|
.get("SIZE")
|
|
.and_then(|v| v.as_ref())
|
|
.and_then(|s| s.parse().ok());
|
|
session.envelope.body_type = params
|
|
.get("BODY")
|
|
.and_then(|v| v.clone());
|
|
|
|
match session.state.transition_mail_from() {
|
|
Ok(new_state) => {
|
|
session.state = new_state;
|
|
LineResult::Response(SmtpResponse::ok("OK"))
|
|
}
|
|
Err(_) => LineResult::Response(SmtpResponse::bad_sequence(
|
|
"Bad sequence of commands",
|
|
)),
|
|
}
|
|
}
|
|
|
|
/// Handle RCPT TO command.
|
|
fn handle_rcpt_to(
|
|
address: String,
|
|
_params: std::collections::HashMap<String, Option<String>>,
|
|
session: &mut SmtpSession,
|
|
config: &SmtpServerConfig,
|
|
) -> LineResult {
|
|
if !session.state.can_rcpt_to() {
|
|
return LineResult::Response(SmtpResponse::bad_sequence(
|
|
"Send MAIL FROM first",
|
|
));
|
|
}
|
|
|
|
if !validation::is_valid_smtp_address(&address) || address.is_empty() {
|
|
return LineResult::Response(SmtpResponse::param_error(
|
|
"Invalid recipient address",
|
|
));
|
|
}
|
|
|
|
if session.envelope.rcpt_to.len() >= config.max_recipients as usize {
|
|
return LineResult::Response(SmtpResponse::new(
|
|
452,
|
|
"Too many recipients",
|
|
));
|
|
}
|
|
|
|
session.envelope.rcpt_to.push(address);
|
|
|
|
match session.state.transition_rcpt_to() {
|
|
Ok(new_state) => {
|
|
session.state = new_state;
|
|
LineResult::Response(SmtpResponse::ok("OK"))
|
|
}
|
|
Err(_) => LineResult::Response(SmtpResponse::bad_sequence(
|
|
"Bad sequence of commands",
|
|
)),
|
|
}
|
|
}
|
|
|
|
/// Handle DATA command: switch to data mode, accumulate, then emit event.
|
|
async fn handle_data(
|
|
session: &mut SmtpSession,
|
|
stream: &mut SmtpStream,
|
|
config: &SmtpServerConfig,
|
|
event_tx: &mpsc::Sender<ConnectionEvent>,
|
|
callback_registry: &dyn CallbackRegistry,
|
|
authenticator: &Arc<MessageAuthenticator>,
|
|
resolver: &Arc<TokioResolver>,
|
|
) -> LineResult {
|
|
if !session.state.can_data() {
|
|
return LineResult::Response(SmtpResponse::bad_sequence(
|
|
"Send RCPT TO first",
|
|
));
|
|
}
|
|
|
|
// Transition to DATA state
|
|
session.state = match session.state.transition_data() {
|
|
Ok(s) => s,
|
|
Err(_) => {
|
|
return LineResult::Response(SmtpResponse::bad_sequence(
|
|
"Bad sequence of commands",
|
|
));
|
|
}
|
|
};
|
|
|
|
// Send 354
|
|
let start_resp = SmtpResponse::start_data();
|
|
if stream.write_all(&start_resp.to_bytes()).await.is_err() {
|
|
return LineResult::Disconnect;
|
|
}
|
|
if stream.flush().await.is_err() {
|
|
return LineResult::Disconnect;
|
|
}
|
|
|
|
// Accumulate data
|
|
let mut accumulator = DataAccumulator::new(config.max_message_size);
|
|
let data_timeout = Duration::from_secs(config.data_timeout_secs);
|
|
let mut buf = [0u8; 8192];
|
|
|
|
loop {
|
|
let read_result = timeout(data_timeout, stream.read_chunk(&mut buf)).await;
|
|
match read_result {
|
|
Err(_) => {
|
|
// Data timeout
|
|
return LineResult::Quit(SmtpResponse::service_unavailable(
|
|
&config.hostname,
|
|
"Data timeout",
|
|
));
|
|
}
|
|
Ok(Err(_)) => return LineResult::Disconnect,
|
|
Ok(Ok(0)) => return LineResult::Disconnect,
|
|
Ok(Ok(n)) => {
|
|
match accumulator.process_chunk(&buf[..n]) {
|
|
DataAction::Continue => continue,
|
|
DataAction::SizeExceeded => {
|
|
// Must still read until end-of-data to stay in sync
|
|
session.state = crate::state::SmtpState::Greeted;
|
|
session.envelope = Default::default();
|
|
return LineResult::Response(SmtpResponse::size_exceeded(
|
|
config.max_message_size,
|
|
));
|
|
}
|
|
DataAction::Complete => break,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Data complete — prepare for delivery
|
|
let raw_message = accumulator.into_message().unwrap_or_default();
|
|
let correlation_id = uuid::Uuid::new_v4().to_string();
|
|
|
|
// --- In-process security pipeline (30s timeout) ---
|
|
let security_results = run_security_pipeline(
|
|
&raw_message,
|
|
&session.remote_addr,
|
|
session.client_hostname.as_deref().unwrap_or("unknown"),
|
|
&config.hostname,
|
|
&session.envelope.mail_from,
|
|
authenticator,
|
|
resolver,
|
|
)
|
|
.await;
|
|
|
|
// Determine transport: inline base64 or temp file
|
|
let email_data = if raw_message.len() <= 256 * 1024 {
|
|
EmailData::Inline {
|
|
base64: BASE64.encode(&raw_message),
|
|
}
|
|
} else {
|
|
// Write to temp file
|
|
let tmp_path = format!("/tmp/mailer-smtp-{}.eml", &correlation_id);
|
|
match tokio::fs::write(&tmp_path, &raw_message).await {
|
|
Ok(_) => EmailData::File { path: tmp_path },
|
|
Err(e) => {
|
|
warn!(error = %e, "Failed to write temp file for large email");
|
|
// Fall back to inline
|
|
EmailData::Inline {
|
|
base64: BASE64.encode(&raw_message),
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
// Register callback before sending event
|
|
let rx = callback_registry.register_email_callback(&correlation_id);
|
|
|
|
// Send event to TS
|
|
let event = ConnectionEvent::EmailReceived {
|
|
correlation_id: correlation_id.clone(),
|
|
session_id: session.id.clone(),
|
|
mail_from: session.envelope.mail_from.clone(),
|
|
rcpt_to: session.envelope.rcpt_to.clone(),
|
|
data: email_data,
|
|
remote_addr: session.remote_addr.clone(),
|
|
client_hostname: session.client_hostname.clone(),
|
|
secure: session.secure,
|
|
authenticated_user: session.authenticated_user().map(|s| s.to_string()),
|
|
security_results
|
|
};
|
|
|
|
if event_tx.send(event).await.is_err() {
|
|
warn!("Failed to send emailReceived event");
|
|
return LineResult::Response(SmtpResponse::local_error(
|
|
"Internal processing error",
|
|
));
|
|
}
|
|
|
|
// Wait for TS response with timeout
|
|
let processing_timeout = Duration::from_secs(config.processing_timeout_secs);
|
|
let result = match timeout(processing_timeout, rx).await {
|
|
Ok(Ok(result)) => result,
|
|
Ok(Err(_)) => {
|
|
warn!(correlation_id = %correlation_id, "Callback channel dropped");
|
|
EmailProcessingResult {
|
|
accepted: false,
|
|
smtp_code: Some(451),
|
|
smtp_message: Some("Processing error".into()),
|
|
}
|
|
}
|
|
Err(_) => {
|
|
warn!(correlation_id = %correlation_id, "Processing timeout");
|
|
EmailProcessingResult {
|
|
accepted: false,
|
|
smtp_code: Some(451),
|
|
smtp_message: Some("Processing timeout".into()),
|
|
}
|
|
}
|
|
};
|
|
|
|
// Reset transaction state
|
|
session.envelope = Default::default();
|
|
let _ = session.state.transition_finished();
|
|
session.state = crate::state::SmtpState::Finished;
|
|
session.record_message();
|
|
|
|
if result.accepted {
|
|
LineResult::Response(SmtpResponse::ok(
|
|
result.smtp_message.unwrap_or_else(|| "Message accepted".into()),
|
|
))
|
|
} else {
|
|
let code = result.smtp_code.unwrap_or(550);
|
|
let msg = result
|
|
.smtp_message
|
|
.unwrap_or_else(|| "Message rejected".into());
|
|
LineResult::Response(SmtpResponse::new(code, msg))
|
|
}
|
|
}
|
|
|
|
/// Handle STARTTLS command.
|
|
///
|
|
/// Returns `StartTlsSignal` to indicate the main loop should send 220 and
|
|
/// perform the TLS upgrade. The main loop handles the stream swap.
|
|
fn handle_starttls(
|
|
session: &SmtpSession,
|
|
config: &SmtpServerConfig,
|
|
tls_acceptor: &Option<Arc<tokio_rustls::TlsAcceptor>>,
|
|
) -> LineResult {
|
|
if session.secure {
|
|
return LineResult::Response(SmtpResponse::bad_sequence(
|
|
"Already using TLS",
|
|
));
|
|
}
|
|
|
|
if !session.state.can_starttls() {
|
|
return LineResult::Response(SmtpResponse::bad_sequence(
|
|
"STARTTLS not allowed in current state",
|
|
));
|
|
}
|
|
|
|
if tls_acceptor.is_none() || !config.has_tls() {
|
|
return LineResult::Response(SmtpResponse::new(
|
|
454,
|
|
"TLS not available",
|
|
));
|
|
}
|
|
|
|
// Signal the main loop to perform TLS upgrade.
|
|
// The main loop will: send 220, extract TCP stream, do TLS handshake.
|
|
LineResult::StartTlsSignal
|
|
}
|
|
|
|
/// Handle AUTH command.
|
|
async fn handle_auth(
|
|
mechanism: AuthMechanism,
|
|
initial_response: Option<String>,
|
|
session: &mut SmtpSession,
|
|
config: &SmtpServerConfig,
|
|
rate_limiter: &RateLimiter,
|
|
event_tx: &mpsc::Sender<ConnectionEvent>,
|
|
callback_registry: &dyn CallbackRegistry,
|
|
) -> LineResult {
|
|
if !config.auth_enabled {
|
|
return LineResult::Response(SmtpResponse::not_implemented());
|
|
}
|
|
|
|
if session.is_authenticated() {
|
|
return LineResult::Response(SmtpResponse::bad_sequence(
|
|
"Already authenticated",
|
|
));
|
|
}
|
|
|
|
if !session.state.can_auth() {
|
|
return LineResult::Response(SmtpResponse::bad_sequence(
|
|
"Send EHLO first",
|
|
));
|
|
}
|
|
|
|
match mechanism {
|
|
AuthMechanism::Plain => {
|
|
if let Some(response) = initial_response {
|
|
// Decode and validate immediately
|
|
return process_auth_plain(
|
|
&response,
|
|
session,
|
|
config,
|
|
rate_limiter,
|
|
event_tx,
|
|
callback_registry,
|
|
)
|
|
.await;
|
|
}
|
|
// No initial response — send challenge for credentials
|
|
session.auth_state = AuthState::WaitingForUsername;
|
|
// For PLAIN, we use an empty challenge
|
|
LineResult::Response(SmtpResponse::auth_challenge(""))
|
|
}
|
|
AuthMechanism::Login => {
|
|
if let Some(response) = initial_response {
|
|
// The initial response is the username (base64)
|
|
match BASE64.decode(response.as_bytes()) {
|
|
Ok(decoded) => {
|
|
let username = String::from_utf8_lossy(&decoded).to_string();
|
|
session.auth_state = AuthState::WaitingForPassword { username };
|
|
// Send password prompt (base64 of "Password:")
|
|
LineResult::Response(SmtpResponse::auth_challenge(
|
|
&BASE64.encode(b"Password:"),
|
|
))
|
|
}
|
|
Err(_) => LineResult::Response(SmtpResponse::param_error(
|
|
"Invalid base64 encoding",
|
|
)),
|
|
}
|
|
} else {
|
|
session.auth_state = AuthState::WaitingForUsername;
|
|
// Send username prompt (base64 of "Username:")
|
|
LineResult::Response(SmtpResponse::auth_challenge(
|
|
&BASE64.encode(b"Username:"),
|
|
))
|
|
}
|
|
}
|
|
AuthMechanism::ScramSha256 => {
|
|
// SCRAM is handled separately in process_line; this should not be reached.
|
|
LineResult::Response(SmtpResponse::not_implemented())
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Handle AUTH SCRAM-SHA-256 — full exchange in a single async function.
|
|
///
|
|
/// SCRAM is a multi-step challenge-response protocol:
|
|
/// 1. Client sends client-first-message (in initial_response or after 334)
|
|
/// 2. Server requests SCRAM credentials from TS
|
|
/// 3. Server sends server-first-message (334 challenge)
|
|
/// 4. Client sends client-final-message (proof)
|
|
/// 5. Server verifies proof and responds with 235 or 535
|
|
async fn handle_auth_scram(
|
|
initial_response: Option<String>,
|
|
session: &mut SmtpSession,
|
|
stream: &mut SmtpStream,
|
|
config: &SmtpServerConfig,
|
|
rate_limiter: &RateLimiter,
|
|
event_tx: &mpsc::Sender<ConnectionEvent>,
|
|
callback_registry: &dyn CallbackRegistry,
|
|
) -> LineResult {
|
|
if !config.auth_enabled {
|
|
return LineResult::Response(SmtpResponse::not_implemented());
|
|
}
|
|
|
|
if session.is_authenticated() {
|
|
return LineResult::Response(SmtpResponse::bad_sequence("Already authenticated"));
|
|
}
|
|
|
|
if !session.state.can_auth() {
|
|
return LineResult::Response(SmtpResponse::bad_sequence("Send EHLO first"));
|
|
}
|
|
|
|
// Step 1: Get client-first-message
|
|
let client_first_b64 = match initial_response {
|
|
Some(s) if !s.is_empty() => s,
|
|
_ => {
|
|
// No initial response — send empty 334 challenge
|
|
let resp = SmtpResponse::auth_challenge("");
|
|
if stream.write_all(&resp.to_bytes()).await.is_err() {
|
|
return LineResult::Disconnect;
|
|
}
|
|
if stream.flush().await.is_err() {
|
|
return LineResult::Disconnect;
|
|
}
|
|
// Read client-first-message
|
|
let mut line = String::new();
|
|
let socket_timeout = Duration::from_secs(config.socket_timeout_secs);
|
|
match timeout(socket_timeout, stream.read_line(&mut line, 4096)).await {
|
|
Err(_) | Ok(Err(_)) | Ok(Ok(0)) => return LineResult::Disconnect,
|
|
Ok(Ok(_)) => {}
|
|
}
|
|
let trimmed = line.trim().to_string();
|
|
if trimmed == "*" {
|
|
return LineResult::Response(SmtpResponse::new(501, "Authentication cancelled"));
|
|
}
|
|
trimmed
|
|
}
|
|
};
|
|
|
|
// Decode base64 client-first-message
|
|
let client_first_bytes = match BASE64.decode(client_first_b64.as_bytes()) {
|
|
Ok(b) => b,
|
|
Err(_) => {
|
|
return LineResult::Response(SmtpResponse::param_error("Invalid base64 encoding"));
|
|
}
|
|
};
|
|
let client_first = match String::from_utf8(client_first_bytes) {
|
|
Ok(s) => s,
|
|
Err(_) => {
|
|
return LineResult::Response(SmtpResponse::param_error("Invalid UTF-8 in SCRAM message"));
|
|
}
|
|
};
|
|
|
|
// Parse client-first-message
|
|
let mut scram = match ScramServer::from_client_first(&client_first) {
|
|
Ok(s) => s,
|
|
Err(e) => {
|
|
debug!(error = %e, "SCRAM client-first-message parse error");
|
|
return LineResult::Response(SmtpResponse::param_error(
|
|
"Invalid SCRAM client-first-message",
|
|
));
|
|
}
|
|
};
|
|
|
|
// Step 2: Request SCRAM credentials from TS
|
|
let correlation_id = uuid::Uuid::new_v4().to_string();
|
|
let rx = callback_registry.register_scram_callback(&correlation_id);
|
|
|
|
let event = ConnectionEvent::ScramCredentialRequest {
|
|
correlation_id: correlation_id.clone(),
|
|
session_id: session.id.clone(),
|
|
username: scram.username.clone(),
|
|
remote_addr: session.remote_addr.clone(),
|
|
};
|
|
|
|
if event_tx.send(event).await.is_err() {
|
|
return LineResult::Response(SmtpResponse::local_error("Internal processing error"));
|
|
}
|
|
|
|
// Wait for credentials from TS
|
|
let cred_timeout = Duration::from_secs(5);
|
|
let cred_result = match timeout(cred_timeout, rx).await {
|
|
Ok(Ok(result)) => result,
|
|
Ok(Err(_)) => {
|
|
warn!(correlation_id = %correlation_id, "SCRAM credential callback dropped");
|
|
return LineResult::Response(SmtpResponse::local_error("Internal processing error"));
|
|
}
|
|
Err(_) => {
|
|
warn!(correlation_id = %correlation_id, "SCRAM credential request timed out");
|
|
return LineResult::Response(SmtpResponse::local_error("Internal processing error"));
|
|
}
|
|
};
|
|
|
|
if !cred_result.found {
|
|
// User not found — fail auth (don't reveal that user doesn't exist)
|
|
session.auth_state = AuthState::None;
|
|
let exceeded = session.record_auth_failure(config.max_auth_failures);
|
|
if exceeded {
|
|
return LineResult::Quit(SmtpResponse::service_unavailable(
|
|
&config.hostname,
|
|
"Too many authentication failures",
|
|
));
|
|
}
|
|
return LineResult::Response(SmtpResponse::auth_failed());
|
|
}
|
|
|
|
let creds = ScramCredentials {
|
|
salt: cred_result.salt.unwrap_or_default(),
|
|
iterations: cred_result.iterations.unwrap_or(4096),
|
|
stored_key: cred_result.stored_key.unwrap_or_default(),
|
|
server_key: cred_result.server_key.unwrap_or_default(),
|
|
};
|
|
|
|
// Step 3: Generate and send server-first-message
|
|
let server_first = scram.server_first_message(creds);
|
|
let server_first_b64 = BASE64.encode(server_first.as_bytes());
|
|
|
|
let challenge = SmtpResponse::auth_challenge(&server_first_b64);
|
|
if stream.write_all(&challenge.to_bytes()).await.is_err() {
|
|
return LineResult::Disconnect;
|
|
}
|
|
if stream.flush().await.is_err() {
|
|
return LineResult::Disconnect;
|
|
}
|
|
|
|
// Step 4: Read client-final-message
|
|
let mut client_final_line = String::new();
|
|
let socket_timeout = Duration::from_secs(config.socket_timeout_secs);
|
|
match timeout(socket_timeout, stream.read_line(&mut client_final_line, 4096)).await {
|
|
Err(_) | Ok(Err(_)) | Ok(Ok(0)) => return LineResult::Disconnect,
|
|
Ok(Ok(_)) => {}
|
|
}
|
|
|
|
let client_final_b64 = client_final_line.trim();
|
|
|
|
// Cancel if *
|
|
if client_final_b64 == "*" {
|
|
session.auth_state = AuthState::None;
|
|
return LineResult::Response(SmtpResponse::new(501, "Authentication cancelled"));
|
|
}
|
|
|
|
// Decode base64 client-final-message
|
|
let client_final_bytes = match BASE64.decode(client_final_b64.as_bytes()) {
|
|
Ok(b) => b,
|
|
Err(_) => {
|
|
session.auth_state = AuthState::None;
|
|
return LineResult::Response(SmtpResponse::param_error("Invalid base64 encoding"));
|
|
}
|
|
};
|
|
let client_final = match String::from_utf8(client_final_bytes) {
|
|
Ok(s) => s,
|
|
Err(_) => {
|
|
session.auth_state = AuthState::None;
|
|
return LineResult::Response(SmtpResponse::param_error("Invalid UTF-8 in SCRAM message"));
|
|
}
|
|
};
|
|
|
|
// Step 5: Verify proof
|
|
match scram.process_client_final(&client_final) {
|
|
Ok(server_final) => {
|
|
let server_final_b64 = BASE64.encode(server_final.as_bytes());
|
|
session.auth_state = AuthState::Authenticated {
|
|
username: scram.username.clone(),
|
|
};
|
|
LineResult::Response(SmtpResponse::new(
|
|
235,
|
|
format!("2.7.0 Authentication successful {}", server_final_b64),
|
|
))
|
|
}
|
|
Err(e) => {
|
|
debug!(error = %e, "SCRAM proof verification failed");
|
|
session.auth_state = AuthState::None;
|
|
let exceeded = session.record_auth_failure(config.max_auth_failures);
|
|
if exceeded {
|
|
if !rate_limiter.check_auth_failure(&session.remote_addr) {
|
|
return LineResult::Quit(SmtpResponse::service_unavailable(
|
|
&config.hostname,
|
|
"Too many authentication failures",
|
|
));
|
|
}
|
|
return LineResult::Quit(SmtpResponse::service_unavailable(
|
|
&config.hostname,
|
|
"Too many authentication failures",
|
|
));
|
|
}
|
|
LineResult::Response(SmtpResponse::auth_failed())
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Handle username input during LOGIN auth flow.
|
|
fn handle_auth_username(line: &str, session: &mut SmtpSession) -> LineResult {
|
|
// Cancel auth if client sends "*"
|
|
if line == "*" {
|
|
session.auth_state = AuthState::None;
|
|
return LineResult::Response(SmtpResponse::new(501, "Authentication cancelled"));
|
|
}
|
|
|
|
match BASE64.decode(line.as_bytes()) {
|
|
Ok(decoded) => {
|
|
let username = String::from_utf8_lossy(&decoded).to_string();
|
|
session.auth_state = AuthState::WaitingForPassword { username };
|
|
LineResult::Response(SmtpResponse::auth_challenge(
|
|
&BASE64.encode(b"Password:"),
|
|
))
|
|
}
|
|
Err(_) => {
|
|
session.auth_state = AuthState::None;
|
|
LineResult::Response(SmtpResponse::param_error(
|
|
"Invalid base64 encoding",
|
|
))
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Handle password input during LOGIN auth flow.
|
|
async fn handle_auth_password(
|
|
line: &str,
|
|
session: &mut SmtpSession,
|
|
config: &SmtpServerConfig,
|
|
rate_limiter: &RateLimiter,
|
|
event_tx: &mpsc::Sender<ConnectionEvent>,
|
|
callback_registry: &dyn CallbackRegistry,
|
|
) -> LineResult {
|
|
// Cancel auth if client sends "*"
|
|
if line == "*" {
|
|
session.auth_state = AuthState::None;
|
|
return LineResult::Response(SmtpResponse::new(501, "Authentication cancelled"));
|
|
}
|
|
|
|
let username = match &session.auth_state {
|
|
AuthState::WaitingForPassword { username } => username.clone(),
|
|
_ => {
|
|
session.auth_state = AuthState::None;
|
|
return LineResult::Response(SmtpResponse::bad_sequence("Unexpected auth state"));
|
|
}
|
|
};
|
|
|
|
let password = match BASE64.decode(line.as_bytes()) {
|
|
Ok(decoded) => String::from_utf8_lossy(&decoded).to_string(),
|
|
Err(_) => {
|
|
session.auth_state = AuthState::None;
|
|
return LineResult::Response(SmtpResponse::param_error(
|
|
"Invalid base64 encoding",
|
|
));
|
|
}
|
|
};
|
|
|
|
validate_credentials(
|
|
&username,
|
|
&password,
|
|
session,
|
|
config,
|
|
rate_limiter,
|
|
event_tx,
|
|
callback_registry,
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Process AUTH PLAIN credentials (base64-encoded "\0username\0password").
|
|
async fn process_auth_plain(
|
|
base64_data: &str,
|
|
session: &mut SmtpSession,
|
|
config: &SmtpServerConfig,
|
|
rate_limiter: &RateLimiter,
|
|
event_tx: &mpsc::Sender<ConnectionEvent>,
|
|
callback_registry: &dyn CallbackRegistry,
|
|
) -> LineResult {
|
|
let decoded = match BASE64.decode(base64_data.as_bytes()) {
|
|
Ok(d) => d,
|
|
Err(_) => {
|
|
return LineResult::Response(SmtpResponse::param_error(
|
|
"Invalid base64 encoding",
|
|
));
|
|
}
|
|
};
|
|
|
|
// PLAIN format: \0username\0password
|
|
let parts: Vec<&[u8]> = decoded.splitn(3, |&b| b == 0).collect();
|
|
if parts.len() < 3 {
|
|
return LineResult::Response(SmtpResponse::param_error(
|
|
"Invalid PLAIN auth format",
|
|
));
|
|
}
|
|
|
|
let username = String::from_utf8_lossy(parts[1]).to_string();
|
|
let password = String::from_utf8_lossy(parts[2]).to_string();
|
|
|
|
validate_credentials(
|
|
&username,
|
|
&password,
|
|
session,
|
|
config,
|
|
rate_limiter,
|
|
event_tx,
|
|
callback_registry,
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Validate credentials by sending authRequest to TS and waiting for response.
|
|
async fn validate_credentials(
|
|
username: &str,
|
|
password: &str,
|
|
session: &mut SmtpSession,
|
|
config: &SmtpServerConfig,
|
|
rate_limiter: &RateLimiter,
|
|
event_tx: &mpsc::Sender<ConnectionEvent>,
|
|
callback_registry: &dyn CallbackRegistry,
|
|
) -> LineResult {
|
|
let correlation_id = uuid::Uuid::new_v4().to_string();
|
|
|
|
// Register callback before sending event
|
|
let rx = callback_registry.register_auth_callback(&correlation_id);
|
|
|
|
let event = ConnectionEvent::AuthRequest {
|
|
correlation_id: correlation_id.clone(),
|
|
session_id: session.id.clone(),
|
|
username: username.to_string(),
|
|
password: password.to_string(),
|
|
remote_addr: session.remote_addr.clone(),
|
|
};
|
|
|
|
if event_tx.send(event).await.is_err() {
|
|
session.auth_state = AuthState::None;
|
|
return LineResult::Response(SmtpResponse::local_error(
|
|
"Internal processing error",
|
|
));
|
|
}
|
|
|
|
// Wait for TS response
|
|
let auth_timeout = Duration::from_secs(5);
|
|
let result = match timeout(auth_timeout, rx).await {
|
|
Ok(Ok(result)) => result,
|
|
Ok(Err(_)) => AuthResult {
|
|
success: false,
|
|
message: Some("Auth processing error".into()),
|
|
},
|
|
Err(_) => AuthResult {
|
|
success: false,
|
|
message: Some("Auth timeout".into()),
|
|
},
|
|
};
|
|
|
|
if result.success {
|
|
session.auth_state = AuthState::Authenticated {
|
|
username: username.to_string(),
|
|
};
|
|
LineResult::Response(SmtpResponse::auth_success())
|
|
} else {
|
|
session.auth_state = AuthState::None;
|
|
let exceeded = session.record_auth_failure(config.max_auth_failures);
|
|
if exceeded {
|
|
if !rate_limiter.check_auth_failure(&session.remote_addr) {
|
|
return LineResult::Quit(SmtpResponse::service_unavailable(
|
|
&config.hostname,
|
|
"Too many authentication failures",
|
|
));
|
|
}
|
|
return LineResult::Quit(SmtpResponse::service_unavailable(
|
|
&config.hostname,
|
|
"Too many authentication failures",
|
|
));
|
|
}
|
|
LineResult::Response(SmtpResponse::auth_failed())
|
|
}
|
|
}
|
|
|
|
/// Extract MIME parts from a raw email message for content scanning.
|
|
///
|
|
/// Returns `(subject, text_body, html_body, attachment_filenames)`.
|
|
fn extract_mime_parts(raw_message: &[u8]) -> (Option<String>, Option<String>, Option<String>, Vec<String>) {
|
|
let parsed = match mailparse::parse_mail(raw_message) {
|
|
Ok(p) => p,
|
|
Err(e) => {
|
|
debug!(error = %e, "Failed to parse MIME for content scanning");
|
|
return (None, None, None, Vec::new());
|
|
}
|
|
};
|
|
|
|
// Extract Subject header
|
|
let subject = parsed
|
|
.headers
|
|
.iter()
|
|
.find(|h| h.get_key().eq_ignore_ascii_case("subject"))
|
|
.map(|h| h.get_value());
|
|
|
|
let mut text_body: Option<String> = None;
|
|
let mut html_body: Option<String> = None;
|
|
let mut attachments: Vec<String> = Vec::new();
|
|
|
|
// Walk the MIME tree
|
|
fn walk_parts(
|
|
part: &mailparse::ParsedMail<'_>,
|
|
text_body: &mut Option<String>,
|
|
html_body: &mut Option<String>,
|
|
attachments: &mut Vec<String>,
|
|
) {
|
|
let content_type = part.ctype.mimetype.to_lowercase();
|
|
let disposition = part.get_content_disposition();
|
|
|
|
// Check if this is an attachment
|
|
if disposition.disposition == mailparse::DispositionType::Attachment {
|
|
if let Some(filename) = disposition.params.get("filename") {
|
|
attachments.push(filename.clone());
|
|
}
|
|
} else if content_type == "text/plain" && text_body.is_none() {
|
|
if let Ok(body) = part.get_body() {
|
|
*text_body = Some(body);
|
|
}
|
|
} else if content_type == "text/html" && html_body.is_none() {
|
|
if let Ok(body) = part.get_body() {
|
|
*html_body = Some(body);
|
|
}
|
|
}
|
|
|
|
// Recurse into subparts
|
|
for sub in &part.subparts {
|
|
walk_parts(sub, text_body, html_body, attachments);
|
|
}
|
|
}
|
|
|
|
walk_parts(&parsed, &mut text_body, &mut html_body, &mut attachments);
|
|
|
|
(subject, text_body, html_body, attachments)
|
|
}
|
|
|
|
/// Run the full security pipeline: DKIM/SPF/DMARC + content scan + IP reputation.
|
|
///
|
|
/// Returns `Some(json_value)` on success or `None` if the pipeline fails or times out.
|
|
async fn run_security_pipeline(
|
|
raw_message: &[u8],
|
|
remote_addr: &str,
|
|
helo_domain: &str,
|
|
hostname: &str,
|
|
mail_from: &str,
|
|
authenticator: &Arc<MessageAuthenticator>,
|
|
resolver: &Arc<TokioResolver>,
|
|
) -> Option<serde_json::Value> {
|
|
let security_timeout = Duration::from_secs(30);
|
|
|
|
match timeout(security_timeout, run_security_pipeline_inner(
|
|
raw_message, remote_addr, helo_domain, hostname, mail_from, authenticator, resolver,
|
|
)).await {
|
|
Ok(Ok(value)) => {
|
|
debug!("In-process security pipeline completed");
|
|
Some(value)
|
|
}
|
|
Ok(Err(e)) => {
|
|
warn!(error = %e, "Security pipeline error — emitting event without results");
|
|
None
|
|
}
|
|
Err(_) => {
|
|
warn!("Security pipeline timed out (30s) — emitting event without results");
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Inner implementation of the security pipeline (no timeout wrapper).
|
|
async fn run_security_pipeline_inner(
|
|
raw_message: &[u8],
|
|
remote_addr: &str,
|
|
helo_domain: &str,
|
|
hostname: &str,
|
|
mail_from: &str,
|
|
authenticator: &Arc<MessageAuthenticator>,
|
|
resolver: &Arc<TokioResolver>,
|
|
) -> std::result::Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>> {
|
|
// Parse the remote IP address
|
|
let ip: IpAddr = remote_addr.parse().unwrap_or(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST));
|
|
|
|
// Run DKIM/SPF/DMARC and IP reputation concurrently
|
|
let (email_security, reputation) = tokio::join!(
|
|
mailer_security::verify_email_security(
|
|
raw_message, ip, helo_domain, hostname, mail_from, authenticator,
|
|
),
|
|
mailer_security::check_reputation(
|
|
ip, mailer_security::DEFAULT_DNSBL_SERVERS, resolver,
|
|
),
|
|
);
|
|
|
|
// Extract MIME parts for content scanning (synchronous)
|
|
let (subject, text_body, html_body, attachment_names) = extract_mime_parts(raw_message);
|
|
|
|
// Run content scan (synchronous)
|
|
let content_scan = mailer_security::content_scanner::scan_content(
|
|
subject.as_deref(),
|
|
text_body.as_deref(),
|
|
html_body.as_deref(),
|
|
&attachment_names,
|
|
);
|
|
|
|
// Build the combined results JSON
|
|
let mut results = serde_json::Map::new();
|
|
|
|
// DKIM/SPF/DMARC
|
|
match email_security {
|
|
Ok(sec) => {
|
|
results.insert("dkim".into(), serde_json::to_value(&sec.dkim)?);
|
|
results.insert("spf".into(), serde_json::to_value(&sec.spf)?);
|
|
results.insert("dmarc".into(), serde_json::to_value(&sec.dmarc)?);
|
|
}
|
|
Err(e) => {
|
|
warn!(error = %e, "Email security verification failed");
|
|
results.insert("dkim".into(), serde_json::Value::Array(vec![]));
|
|
results.insert("spf".into(), serde_json::Value::Null);
|
|
results.insert("dmarc".into(), serde_json::Value::Null);
|
|
}
|
|
}
|
|
|
|
// Content scan
|
|
results.insert("contentScan".into(), serde_json::to_value(&content_scan)?);
|
|
|
|
// IP reputation
|
|
match reputation {
|
|
Ok(rep) => {
|
|
results.insert("ipReputation".into(), serde_json::to_value(&rep)?);
|
|
}
|
|
Err(e) => {
|
|
warn!(error = %e, "IP reputation check failed");
|
|
results.insert("ipReputation".into(), serde_json::Value::Null);
|
|
}
|
|
}
|
|
|
|
Ok(serde_json::Value::Object(results))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn test_email_data_serialization() {
|
|
let data = EmailData::Inline {
|
|
base64: "dGVzdA==".into(),
|
|
};
|
|
let json = serde_json::to_string(&data).unwrap();
|
|
assert!(json.contains("inline"));
|
|
|
|
let data = EmailData::File {
|
|
path: "/tmp/test.eml".into(),
|
|
};
|
|
let json = serde_json::to_string(&data).unwrap();
|
|
assert!(json.contains("file"));
|
|
}
|
|
|
|
#[test]
|
|
fn test_processing_result_serialization() {
|
|
let result = EmailProcessingResult {
|
|
accepted: true,
|
|
smtp_code: Some(250),
|
|
smtp_message: Some("OK".into()),
|
|
};
|
|
let json = serde_json::to_string(&result).unwrap();
|
|
assert!(json.contains("accepted"));
|
|
}
|
|
|
|
#[test]
|
|
fn test_extract_mime_parts_simple() {
|
|
let raw = b"From: sender@example.com\r\n\
|
|
To: rcpt@example.com\r\n\
|
|
Subject: Test Subject\r\n\
|
|
Content-Type: text/plain\r\n\
|
|
\r\n\
|
|
Hello, this is a test body.\r\n";
|
|
|
|
let (subject, text, html, attachments) = extract_mime_parts(raw);
|
|
assert_eq!(subject.as_deref(), Some("Test Subject"));
|
|
assert!(text.is_some());
|
|
assert!(text.unwrap().contains("Hello, this is a test body."));
|
|
assert!(html.is_none());
|
|
assert!(attachments.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn test_extract_mime_parts_multipart() {
|
|
let raw = b"From: sender@example.com\r\n\
|
|
To: rcpt@example.com\r\n\
|
|
Subject: Multipart Test\r\n\
|
|
Content-Type: multipart/mixed; boundary=\"boundary123\"\r\n\
|
|
\r\n\
|
|
--boundary123\r\n\
|
|
Content-Type: text/plain\r\n\
|
|
\r\n\
|
|
Plain text body\r\n\
|
|
--boundary123\r\n\
|
|
Content-Type: text/html\r\n\
|
|
\r\n\
|
|
<html><body>HTML body</body></html>\r\n\
|
|
--boundary123\r\n\
|
|
Content-Type: application/octet-stream\r\n\
|
|
Content-Disposition: attachment; filename=\"report.pdf\"\r\n\
|
|
\r\n\
|
|
binary data here\r\n\
|
|
--boundary123--\r\n";
|
|
|
|
let (subject, text, html, attachments) = extract_mime_parts(raw);
|
|
assert_eq!(subject.as_deref(), Some("Multipart Test"));
|
|
assert!(text.is_some());
|
|
assert!(text.unwrap().contains("Plain text body"));
|
|
assert!(html.is_some());
|
|
assert!(html.unwrap().contains("HTML body"));
|
|
assert_eq!(attachments.len(), 1);
|
|
assert_eq!(attachments[0], "report.pdf");
|
|
}
|
|
|
|
#[test]
|
|
fn test_extract_mime_parts_no_subject() {
|
|
let raw = b"From: sender@example.com\r\n\
|
|
To: rcpt@example.com\r\n\
|
|
Content-Type: text/plain\r\n\
|
|
\r\n\
|
|
Body without subject\r\n";
|
|
|
|
let (subject, text, _html, _attachments) = extract_mime_parts(raw);
|
|
assert!(subject.is_none());
|
|
assert!(text.is_some());
|
|
}
|
|
|
|
#[test]
|
|
fn test_extract_mime_parts_invalid() {
|
|
let raw = b"this is not a valid email";
|
|
let (subject, text, html, attachments) = extract_mime_parts(raw);
|
|
// Should not panic, may or may not parse partially
|
|
// The key property is that it doesn't crash
|
|
let _ = (subject, text, html, attachments);
|
|
}
|
|
|
|
#[test]
|
|
fn test_extract_mime_parts_multiple_attachments() {
|
|
let raw = b"From: sender@example.com\r\n\
|
|
To: rcpt@example.com\r\n\
|
|
Subject: Attachments\r\n\
|
|
Content-Type: multipart/mixed; boundary=\"bound\"\r\n\
|
|
\r\n\
|
|
--bound\r\n\
|
|
Content-Type: text/plain\r\n\
|
|
\r\n\
|
|
See attached\r\n\
|
|
--bound\r\n\
|
|
Content-Type: application/pdf\r\n\
|
|
Content-Disposition: attachment; filename=\"doc1.pdf\"\r\n\
|
|
\r\n\
|
|
pdf data\r\n\
|
|
--bound\r\n\
|
|
Content-Type: application/vnd.ms-excel\r\n\
|
|
Content-Disposition: attachment; filename=\"data.xlsx\"\r\n\
|
|
\r\n\
|
|
excel data\r\n\
|
|
--bound--\r\n";
|
|
|
|
let (subject, text, _html, attachments) = extract_mime_parts(raw);
|
|
assert_eq!(subject.as_deref(), Some("Attachments"));
|
|
assert!(text.is_some());
|
|
assert_eq!(attachments.len(), 2);
|
|
assert!(attachments.contains(&"doc1.pdf".to_string()));
|
|
assert!(attachments.contains(&"data.xlsx".to_string()));
|
|
}
|
|
}
|