rusmes_smtp/session/
mod.rs

1//! SMTP session state machine and handler
2
3mod data;
4
5use crate::command::SmtpCommand;
6use crate::parser::parse_command_smtputf8;
7use crate::response::SmtpResponse;
8use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
9use ipnetwork::IpNetwork;
10use rusmes_auth::AuthBackend;
11use rusmes_core::{MailProcessorRouter, RateLimiter};
12use rusmes_proto::{MailAddress, Username};
13use rusmes_storage::StorageBackend;
14use std::collections::HashMap;
15use std::net::SocketAddr;
16use std::sync::Arc;
17use std::time::{Duration, Instant};
18use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
19use tokio::net::TcpStream;
20use tokio::sync::RwLock;
21
22/// SMTP session state
23#[derive(Debug, Clone, PartialEq)]
24pub enum SmtpState {
25    /// Initial state before connection
26    Initial,
27    /// Connected, waiting for HELO/EHLO
28    Connected,
29    /// Authenticated (if AUTH required)
30    Authenticated,
31    /// In mail transaction (after MAIL FROM)
32    MailTransaction,
33    /// Receiving message data (after DATA command)
34    Data,
35    /// Quit command received
36    Quit,
37}
38
39/// SMTP transaction state
40#[derive(Debug, Clone)]
41pub struct SmtpTransaction {
42    sender: Option<MailAddress>,
43    recipients: Vec<MailAddress>,
44    helo_name: Option<String>,
45    message_size: usize,
46    /// Declared message size from MAIL FROM SIZE parameter
47    declared_size: Option<usize>,
48    /// BODY parameter value (7BIT, 8BITMIME, BINARYMIME)
49    body_type: Option<String>,
50    /// SMTPUTF8 flag
51    smtputf8: bool,
52    /// BDAT state for CHUNKING extension (RFC 3030)
53    bdat_state: Option<crate::BdatState>,
54    /// Message data received via DATA command
55    message_data: Vec<u8>,
56}
57
58impl SmtpTransaction {
59    fn new() -> Self {
60        Self {
61            sender: None,
62            recipients: Vec::new(),
63            helo_name: None,
64            message_size: 0,
65            declared_size: None,
66            body_type: None,
67            smtputf8: false,
68            bdat_state: None,
69            message_data: Vec::new(),
70        }
71    }
72
73    fn reset(&mut self) {
74        self.sender = None;
75        self.recipients.clear();
76        self.message_size = 0;
77        self.declared_size = None;
78        self.body_type = None;
79        self.smtputf8 = false;
80        self.bdat_state = None;
81        self.message_data.clear();
82    }
83
84    fn is_valid(&self) -> bool {
85        self.sender.is_some() && !self.recipients.is_empty()
86    }
87}
88
89/// SMTP session configuration
90#[derive(Debug, Clone)]
91pub struct SmtpConfig {
92    pub hostname: String,
93    pub max_message_size: usize,
94    pub require_auth: bool,
95    pub enable_starttls: bool,
96    pub check_recipient_exists: bool,
97    pub reject_unknown_recipients: bool,
98    /// CIDR networks allowed to relay mail (e.g., "192.168.0.0/16")
99    pub relay_networks: Vec<String>,
100    /// Local domains that this server accepts mail for
101    pub local_domains: Vec<String>,
102    /// Total connection timeout (max session duration)
103    pub connection_timeout: Duration,
104    /// Idle timeout (time between commands)
105    pub idle_timeout: Duration,
106    /// Blocked CIDR networks — connections from these IPs are silently dropped
107    /// immediately after TCP accept (before the SMTP banner is sent).
108    pub blocked_networks: Vec<IpNetwork>,
109    /// Maximum size of an in-memory DATA buffer before spilling to a tempfile.
110    ///
111    /// Messages exceeding this threshold are written to a temporary file on disk
112    /// and delivered as [`rusmes_proto::MessageBody::Large`] to the storage
113    /// pipeline.  Defaults to 1 MiB.
114    pub data_tempfile_threshold: usize,
115    /// Directory used to write DATA spill tempfiles.
116    ///
117    /// Defaults to the OS temporary directory ([`std::env::temp_dir()`]).
118    /// Tests can override this field with an isolated directory to avoid
119    /// interference when multiple test processes run concurrently.
120    pub data_spill_dir: std::path::PathBuf,
121}
122
123impl Default for SmtpConfig {
124    fn default() -> Self {
125        Self {
126            hostname: "localhost".to_string(),
127            max_message_size: 10 * 1024 * 1024, // 10MB
128            require_auth: false,
129            enable_starttls: false,
130            check_recipient_exists: true,
131            reject_unknown_recipients: true,
132            relay_networks: vec!["127.0.0.0/8".to_string()],
133            local_domains: vec!["localhost".to_string()],
134            connection_timeout: Duration::from_secs(3600), // 1 hour
135            idle_timeout: Duration::from_secs(300),        // 5 minutes
136            blocked_networks: Vec::new(),
137            data_tempfile_threshold: 1024 * 1024, // 1 MiB
138            data_spill_dir: std::env::temp_dir(),
139        }
140    }
141}
142
143/// Cache entry for recipient validation
144#[derive(Debug, Clone)]
145struct RecipientCacheEntry {
146    exists: bool,
147    cached_at: Instant,
148}
149
150/// SCRAM-SHA-256 authentication state
151///
152/// Captures the data needed to verify the client's proof in the second SCRAM round
153/// trip (RFC 5802 §5). The credential bundle (`stored_key` + `server_key`) is fetched
154/// from the [`AuthBackend`] during the first round trip and cached here so the
155/// verification round trip does not need to re-query the backend.
156#[derive(Debug, Clone)]
157struct ScramState {
158    client_first_bare: String,
159    server_first: String,
160    nonce: String,
161    username: String,
162    /// `SHA-256(ClientKey)` — used to verify the client proof.
163    stored_key: Vec<u8>,
164    /// `HMAC-SHA-256(SaltedPassword, "Server Key")` — used to compute the server
165    /// signature returned to the client on success.
166    server_key: Vec<u8>,
167}
168
169/// SMTP session handler
170pub struct SmtpSession {
171    remote_addr: SocketAddr,
172    state: SmtpState,
173    transaction: SmtpTransaction,
174    config: SmtpConfig,
175    authenticated: bool,
176    #[allow(dead_code)]
177    username: Option<String>,
178    #[allow(dead_code)]
179    relaying_allowed: bool,
180    #[allow(dead_code)]
181    processor_router: Arc<MailProcessorRouter>,
182    auth_backend: Arc<dyn AuthBackend>,
183    rate_limiter: Arc<RateLimiter>,
184    storage_backend: Arc<dyn StorageBackend>,
185    recipient_cache: Arc<RwLock<HashMap<String, RecipientCacheEntry>>>,
186    /// CRAM-MD5 challenge for ongoing authentication
187    cram_md5_challenge: Option<String>,
188    /// SCRAM-SHA-256 authentication state
189    scram_state: Option<ScramState>,
190    /// Whether the client greeted with EHLO (as opposed to HELO).
191    ///
192    /// RFC 6531 §3 forbids using the SMTPUTF8 MAIL parameter when the session
193    /// was opened with a plain HELO greeting — SMTPUTF8 is an ESMTP extension
194    /// and requires EHLO negotiation.
195    ehlo_used: bool,
196    /// Peer certificate chain captured after a mutual-TLS handshake.
197    ///
198    /// Populated by the TLS handshake handler when the
199    /// submission server is configured with `client_auth = "optional"` or
200    /// `"required"` and the client presented a certificate.  `None` means
201    /// either mTLS is disabled or the client did not send a certificate
202    /// (allowed when `client_auth = "optional"`).
203    pub peer_certificates: Option<Vec<rustls::pki_types::CertificateDer<'static>>>,
204}
205
206/// SMTP session with stream
207pub struct SmtpSessionHandler {
208    session: SmtpSession,
209    stream: TcpStream,
210    /// Support for PIPELINING - buffer of commands to process
211    #[allow(dead_code)]
212    pipelined_commands: Vec<String>,
213    /// Connection start time
214    #[allow(dead_code)]
215    connection_started: Instant,
216    /// Last command received time
217    #[allow(dead_code)]
218    last_command: Instant,
219}
220
221impl SmtpSessionHandler {
222    /// Create a new SMTP session handler
223    #[allow(clippy::too_many_arguments)]
224    pub fn new(
225        stream: TcpStream,
226        remote_addr: SocketAddr,
227        config: SmtpConfig,
228        processor_router: Arc<MailProcessorRouter>,
229        auth_backend: Arc<dyn AuthBackend>,
230        rate_limiter: Arc<RateLimiter>,
231        storage_backend: Arc<dyn StorageBackend>,
232    ) -> Self {
233        let now = Instant::now();
234        Self {
235            session: SmtpSession {
236                remote_addr,
237                state: SmtpState::Connected,
238                transaction: SmtpTransaction::new(),
239                config,
240                authenticated: false,
241                username: None,
242                relaying_allowed: false,
243                processor_router,
244                auth_backend,
245                rate_limiter,
246                storage_backend,
247                recipient_cache: Arc::new(RwLock::new(HashMap::new())),
248                cram_md5_challenge: None,
249                scram_state: None,
250                ehlo_used: false,
251                peer_certificates: None,
252            },
253            stream,
254            pipelined_commands: Vec::new(),
255            connection_started: now,
256            last_command: now,
257        }
258    }
259
260    /// Handle the SMTP session
261    pub async fn handle(mut self) -> anyhow::Result<()> {
262        // Track this session in the active-connections gauge and the TLS counter.
263        // The guard's Drop decrements the gauge regardless of how this method exits
264        // (success, ?, panic-unwind, early break).
265        let metrics = rusmes_metrics::global_metrics();
266        let _conn_guard = metrics.connection_guard("smtp");
267        // Count every accepted TCP connection as one SMTP connection.
268        metrics.inc_smtp_connections();
269        // SMTP sessions start plaintext on the standard MSA port; STARTTLS upgrades happen
270        // mid-session. The implicit-TLS variant ("smtps" on 465) would be wrapped before
271        // SmtpSessionHandler::new — we treat it as the same code path here and label `no`
272        // up-front; a future STARTTLS branch in the handler should call
273        // `metrics.inc_tls_session(rusmes_metrics::tls_label::STARTTLS)` on successful upgrade.
274        metrics.inc_tls_session(rusmes_metrics::tls_label::NO);
275
276        let (read_half, write_half) = tokio::io::split(self.stream);
277        let mut reader = BufReader::new(read_half);
278        let mut writer = BufWriter::new(write_half);
279
280        // Send greeting
281        Self::write_response_to(
282            &mut writer,
283            SmtpResponse::service_ready(&self.session.config.hostname),
284            &self.session.remote_addr,
285        )
286        .await?;
287
288        let mut line = String::new();
289
290        loop {
291            // Check total connection timeout
292            if self.connection_started.elapsed() > self.session.config.connection_timeout {
293                tracing::info!(
294                    "Connection timeout exceeded for {}",
295                    self.session.remote_addr
296                );
297                Self::write_response_to(
298                    &mut writer,
299                    SmtpResponse::new(421, "4.4.2 Connection timeout - closing connection"),
300                    &self.session.remote_addr,
301                )
302                .await?;
303                break;
304            }
305
306            line.clear();
307
308            // Read command with idle timeout
309            let n = match tokio::time::timeout(
310                self.session.config.idle_timeout,
311                reader.read_line(&mut line),
312            )
313            .await
314            {
315                Ok(Ok(n)) => n,
316                Ok(Err(e)) => {
317                    tracing::error!("Read error from {}: {}", self.session.remote_addr, e);
318                    break;
319                }
320                Err(_) => {
321                    // Idle timeout — send RFC 5321 compliant closing response
322                    tracing::info!(
323                        peer = %self.session.remote_addr,
324                        "smtp.session idle timeout, closing"
325                    );
326                    Self::write_response_to(
327                        &mut writer,
328                        SmtpResponse::new(421, "4.4.2 Connection timed out due to inactivity"),
329                        &self.session.remote_addr,
330                    )
331                    .await?;
332                    break;
333                }
334            };
335
336            if n == 0 {
337                break; // EOF
338            }
339
340            // Update last command time
341            self.last_command = Instant::now();
342
343            let line_trimmed = line.trim();
344            tracing::debug!(
345                "SMTP command from {}: {}",
346                self.session.remote_addr,
347                line_trimmed
348            );
349
350            // Check if we're waiting for CRAM-MD5 response
351            if self.session.cram_md5_challenge.is_some() {
352                let response = match self.session.handle_cram_md5_response(line_trimmed).await {
353                    Ok(resp) => resp,
354                    Err(e) => {
355                        tracing::error!("Error handling CRAM-MD5 response: {}", e);
356                        SmtpResponse::new(535, "5.7.8 Authentication credentials invalid")
357                    }
358                };
359                Self::write_response_to(&mut writer, response, &self.session.remote_addr).await?;
360                continue;
361            }
362
363            // Check if we're waiting for SCRAM-SHA-256 response
364            if self.session.scram_state.is_some() {
365                let response = match self.session.handle_scram_client_final(line_trimmed).await {
366                    Ok(resp) => resp,
367                    Err(e) => {
368                        tracing::error!("Error handling SCRAM-SHA-256 client-final: {}", e);
369                        self.session.scram_state = None;
370                        SmtpResponse::new(535, "5.7.8 Authentication credentials invalid")
371                    }
372                };
373                Self::write_response_to(&mut writer, response, &self.session.remote_addr).await?;
374                continue;
375            }
376
377            // Parse command — use the SMTPUTF8-aware variant when the client
378            // opened the session with EHLO (RFC 6531 §3 requires EHLO).
379            let command = match parse_command_smtputf8(line_trimmed, self.session.ehlo_used) {
380                Ok(cmd) => cmd,
381                Err(e) => {
382                    tracing::warn!("Failed to parse command: {}", e);
383                    Self::write_response_to(
384                        &mut writer,
385                        SmtpResponse::syntax_error("Command not recognized"),
386                        &self.session.remote_addr,
387                    )
388                    .await?;
389                    continue;
390                }
391            };
392
393            // Handle command
394            let response = match self.session.handle_command(command.clone()).await {
395                Ok(resp) => resp,
396                Err(e) => {
397                    tracing::error!("Error handling command: {}", e);
398                    rusmes_metrics::global_metrics().inc_smtp_errors();
399                    SmtpResponse::local_error("Internal server error")
400                }
401            };
402
403            Self::write_response_to(&mut writer, response, &self.session.remote_addr).await?;
404
405            // Check if we should close the connection
406            if self.session.state == SmtpState::Quit {
407                break;
408            }
409
410            // For PIPELINING: DATA command requires special handling
411            // After DATA is accepted, we must stop processing pipelined commands
412            // and read the message data
413            if matches!(command, SmtpCommand::Data) && self.session.state == SmtpState::Data {
414                // Read message data (until .<CRLF>)
415                let remote_addr = self.session.remote_addr;
416                if let Err(e) = Self::handle_data_input(
417                    &mut self.session,
418                    &mut reader,
419                    &mut writer,
420                    &remote_addr,
421                )
422                .await
423                {
424                    tracing::error!("Error reading message data: {}", e);
425                    Self::write_response_to(
426                        &mut writer,
427                        SmtpResponse::local_error("Error reading message data"),
428                        &remote_addr,
429                    )
430                    .await?;
431                }
432            }
433        }
434
435        Ok(())
436    }
437
438    /// Handle DATA input (message body) — delegates to `data` submodule.
439    ///
440    /// Implements RFC 5321 §4.5.2 transparency (dot-stuffing removal) and
441    /// the hybrid in-memory / tempfile spill policy governed by
442    /// [`SmtpConfig::data_tempfile_threshold`].
443    pub(crate) async fn handle_data_input<R, W>(
444        session: &mut SmtpSession,
445        reader: &mut R,
446        writer: &mut W,
447        remote_addr: &SocketAddr,
448    ) -> anyhow::Result<()>
449    where
450        R: AsyncBufReadExt + Unpin,
451        W: AsyncWriteExt + Unpin,
452    {
453        data::handle_data_input(session, reader, writer, remote_addr).await
454    }
455
456    /// Write a response to a writer
457    pub(crate) async fn write_response_to<W: AsyncWriteExt + Unpin>(
458        writer: &mut W,
459        response: SmtpResponse,
460        remote_addr: &SocketAddr,
461    ) -> anyhow::Result<()> {
462        let formatted = response.format();
463        tracing::debug!("SMTP response to {}: {}", remote_addr, formatted.trim());
464        writer.write_all(formatted.as_bytes()).await?;
465        writer.flush().await?;
466        Ok(())
467    }
468}
469
470impl SmtpSession {
471    /// Handle a single SMTP command
472    async fn handle_command(&mut self, command: SmtpCommand) -> anyhow::Result<SmtpResponse> {
473        match command {
474            SmtpCommand::Helo(domain) => self.handle_helo(domain).await,
475            SmtpCommand::Ehlo(domain) => self.handle_ehlo(domain).await,
476            SmtpCommand::Mail { from, params } => self.handle_mail(from, params).await,
477            SmtpCommand::Rcpt { to, params } => self.handle_rcpt(to, params).await,
478            SmtpCommand::Data => self.handle_data().await,
479            SmtpCommand::Bdat { chunk_size, last } => self.handle_bdat(chunk_size, last).await,
480            SmtpCommand::Rset => self.handle_rset().await,
481            SmtpCommand::Noop => Ok(SmtpResponse::ok_simple()),
482            SmtpCommand::Quit => self.handle_quit().await,
483            SmtpCommand::StartTls => self.handle_starttls().await,
484            SmtpCommand::Auth {
485                mechanism,
486                initial_response,
487            } => self.handle_auth(mechanism, initial_response).await,
488            _ => Ok(SmtpResponse::not_implemented("Command not implemented")),
489        }
490    }
491
492    /// Handle HELO command
493    async fn handle_helo(&mut self, domain: String) -> anyhow::Result<SmtpResponse> {
494        if self.state != SmtpState::Connected && self.state != SmtpState::Authenticated {
495            return Ok(SmtpResponse::bad_sequence("Out of sequence"));
496        }
497
498        tracing::info!(
499            peer = %self.remote_addr,
500            client_hostname = %domain,
501            "smtp.session HELO received"
502        );
503        self.transaction.helo_name = Some(domain);
504        // HELO does not enable ESMTP extensions — SMTPUTF8 requires EHLO.
505        self.ehlo_used = false;
506        self.state = SmtpState::Authenticated;
507
508        Ok(SmtpResponse::ok(format!(
509            "{} Hello {}",
510            self.config.hostname,
511            self.remote_addr.ip()
512        )))
513    }
514
515    /// Handle EHLO command
516    async fn handle_ehlo(&mut self, domain: String) -> anyhow::Result<SmtpResponse> {
517        if self.state != SmtpState::Connected && self.state != SmtpState::Authenticated {
518            return Ok(SmtpResponse::bad_sequence("Out of sequence"));
519        }
520
521        tracing::info!(
522            peer = %self.remote_addr,
523            client_hostname = %domain,
524            "smtp.session EHLO received"
525        );
526        self.transaction.helo_name = Some(domain);
527        // EHLO enables ESMTP extensions — SMTPUTF8 is now available.
528        self.ehlo_used = true;
529        self.state = SmtpState::Authenticated;
530
531        let mut extensions = vec![
532            format!("SIZE {}", self.config.max_message_size),
533            "8BITMIME".to_string(),
534            "SMTPUTF8".to_string(),
535            "PIPELINING".to_string(),
536            "CHUNKING".to_string(), // RFC 3030 - BDAT support
537        ];
538
539        if self.config.enable_starttls {
540            extensions.push("STARTTLS".to_string());
541        }
542
543        if self.config.require_auth {
544            extensions.push("AUTH PLAIN LOGIN CRAM-MD5 SCRAM-SHA-256".to_string());
545        }
546
547        Ok(SmtpResponse::ehlo(&self.config.hostname, extensions))
548    }
549
550    /// Handle MAIL FROM command
551    async fn handle_mail(
552        &mut self,
553        from: MailAddress,
554        params: Vec<crate::command::MailParam>,
555    ) -> anyhow::Result<SmtpResponse> {
556        if self.state != SmtpState::Authenticated {
557            return Ok(SmtpResponse::bad_sequence("Must send HELO/EHLO first"));
558        }
559
560        if self.config.require_auth && !self.authenticated {
561            return Ok(SmtpResponse::bad_sequence("Authentication required"));
562        }
563
564        // Check message rate limit (IP + sender combined for tightest control)
565        let ip = self.remote_addr.ip();
566        if !self
567            .rate_limiter
568            .allow_message_ip_and_sender(ip, &from.as_string())
569            .await
570        {
571            tracing::warn!("Message rate limit exceeded for {} from {}", from, ip);
572            return Ok(SmtpResponse::mailbox_unavailable(
573                "Rate limit exceeded, please try again later",
574            ));
575        }
576
577        // Process ESMTP parameters
578        for param in params {
579            match param.keyword.to_uppercase().as_str() {
580                "SIZE" => {
581                    // RFC 1870 - SIZE extension
582                    if let Some(size_str) = param.value {
583                        match size_str.parse::<usize>() {
584                            Ok(size) => {
585                                if size > self.config.max_message_size {
586                                    return Ok(SmtpResponse::storage_exceeded(format!(
587                                        "Message size {} exceeds maximum {}",
588                                        size, self.config.max_message_size
589                                    )));
590                                }
591                                self.transaction.declared_size = Some(size);
592                            }
593                            Err(_) => {
594                                return Ok(SmtpResponse::parameter_error("Invalid SIZE parameter"));
595                            }
596                        }
597                    } else {
598                        return Ok(SmtpResponse::parameter_error(
599                            "SIZE parameter requires a value",
600                        ));
601                    }
602                }
603                "BODY" => {
604                    // RFC 6152 - 8BITMIME extension
605                    if let Some(body_value) = param.value {
606                        let body_upper = body_value.to_uppercase();
607                        match body_upper.as_str() {
608                            "7BIT" | "8BITMIME" => {
609                                self.transaction.body_type = Some(body_upper);
610                            }
611                            _ => {
612                                return Ok(SmtpResponse::parameter_not_implemented(format!(
613                                    "Unsupported BODY type: {}",
614                                    body_value
615                                )));
616                            }
617                        }
618                    } else {
619                        return Ok(SmtpResponse::parameter_error(
620                            "BODY parameter requires a value",
621                        ));
622                    }
623                }
624                "SMTPUTF8" => {
625                    // RFC 6531 §3.4 — SMTPUTF8 is an ESMTP extension; it is
626                    // only available after EHLO (not HELO). The parameter must
627                    // carry no value.
628                    if !self.ehlo_used {
629                        return Ok(SmtpResponse::parameter_error(
630                            "SMTPUTF8 requires EHLO (not HELO)",
631                        ));
632                    }
633                    if param.value.is_none() {
634                        self.transaction.smtputf8 = true;
635                    } else {
636                        return Ok(SmtpResponse::parameter_error(
637                            "SMTPUTF8 parameter must not have a value",
638                        ));
639                    }
640                }
641                _ => {
642                    // Unknown parameter - ignore per RFC 5321
643                    tracing::debug!("Unknown MAIL parameter: {}", param.keyword);
644                }
645            }
646        }
647
648        // RFC 6531 §3.4: if the reverse-path contains a non-ASCII local-part,
649        // the client MUST have declared SMTPUTF8 in this MAIL FROM command.
650        // Reject with 501 5.5.4 if the parameter was omitted.
651        if from.local_part().bytes().any(|b| b >= 0x80) && !self.transaction.smtputf8 {
652            return Ok(SmtpResponse::new(
653                501,
654                "5.5.4 Non-ASCII local-part requires SMTPUTF8 parameter (RFC 6531 §3.4)",
655            ));
656        }
657
658        tracing::info!(
659            peer = %self.remote_addr,
660            mail_from = %from,
661            "smtp.session MAIL FROM accepted"
662        );
663        self.transaction.sender = Some(from.clone());
664        self.state = SmtpState::MailTransaction;
665
666        Ok(SmtpResponse::ok(format!("Sender {} OK", from)))
667    }
668
669    /// Handle RCPT TO command
670    async fn handle_rcpt(
671        &mut self,
672        to: MailAddress,
673        params: Vec<crate::command::MailParam>,
674    ) -> anyhow::Result<SmtpResponse> {
675        if self.state != SmtpState::MailTransaction {
676            return Ok(SmtpResponse::bad_sequence("Must send MAIL FROM first"));
677        }
678
679        // Process ESMTP parameters (for future extensions like DSN)
680        for param in params {
681            // Unknown parameter - ignore per RFC 5321
682            tracing::debug!("Unknown RCPT parameter: {}", param.keyword);
683        }
684
685        // Check relay authorization
686        if !self.is_relay_allowed(&to) {
687            tracing::info!(
688                peer = %self.remote_addr,
689                rcpt_to = %to,
690                "smtp.session RCPT TO rejected: relaying denied"
691            );
692            return Ok(SmtpResponse::new(550, "5.7.1 Relaying denied"));
693        }
694
695        // Validate recipient if configured
696        if self.config.check_recipient_exists {
697            // Skip validation for relay-authorized senders
698            if !self.authenticated && !self.relaying_allowed {
699                match self.validate_recipient(&to).await {
700                    Ok(true) => {
701                        // Recipient exists, continue
702                    }
703                    Ok(false) => {
704                        if self.config.reject_unknown_recipients {
705                            tracing::warn!("Rejecting unknown recipient: {}", to);
706                            return Ok(SmtpResponse::new(
707                                550,
708                                format!("5.1.1 User unknown: {}", to),
709                            ));
710                        } else {
711                            // Accept but log
712                            tracing::info!(
713                                "Accepting unknown recipient (rejection disabled): {}",
714                                to
715                            );
716                        }
717                    }
718                    Err(e) => {
719                        tracing::error!("Error validating recipient {}: {}", to, e);
720                        // On error, fail open to avoid blocking legitimate mail
721                        tracing::warn!("Accepting recipient {} due to validation error", to);
722                    }
723                }
724            }
725        }
726
727        // Add recipient
728        tracing::info!(
729            peer = %self.remote_addr,
730            rcpt_to = %to,
731            "smtp.session RCPT TO accepted"
732        );
733        self.transaction.recipients.push(to.clone());
734
735        Ok(SmtpResponse::ok(format!("Recipient {} OK", to)))
736    }
737
738    /// Handle DATA command
739    async fn handle_data(&mut self) -> anyhow::Result<SmtpResponse> {
740        if self.state != SmtpState::MailTransaction {
741            return Ok(SmtpResponse::bad_sequence("Must send RCPT TO first"));
742        }
743
744        if !self.transaction.is_valid() {
745            return Ok(SmtpResponse::bad_sequence("Need at least one recipient"));
746        }
747
748        self.state = SmtpState::Data;
749        Ok(SmtpResponse::start_data())
750    }
751
752    /// Handle BDAT command (RFC 3030 CHUNKING)
753    ///
754    /// This method only validates the command and prepares for chunk reception.
755    /// Actual chunk data reading must be done by the caller after receiving this response.
756    async fn handle_bdat(&mut self, chunk_size: usize, last: bool) -> anyhow::Result<SmtpResponse> {
757        // BDAT can only be used after MAIL FROM and RCPT TO
758        if self.state != SmtpState::MailTransaction {
759            return Ok(SmtpResponse::bad_sequence(
760                "Must send MAIL FROM and RCPT TO first",
761            ));
762        }
763
764        if !self.transaction.is_valid() {
765            return Ok(SmtpResponse::bad_sequence(
766                "Need sender and at least one recipient",
767            ));
768        }
769
770        // Initialize BDAT state if not already present
771        if self.transaction.bdat_state.is_none() {
772            self.transaction.bdat_state = Some(crate::BdatState::new(self.config.max_message_size));
773        }
774
775        // Note: The actual chunk data reading happens outside this method
776        // The caller must read exactly chunk_size bytes and call add_chunk on bdat_state
777
778        // Check if this would exceed size limits (preliminary check)
779        // Safety: we just initialized bdat_state above if it was None
780        let bdat_state = match self.transaction.bdat_state.as_ref() {
781            Some(state) => state,
782            None => {
783                return Err(anyhow::anyhow!(
784                    "Internal error: bdat_state not initialized"
785                ))
786            }
787        };
788        if bdat_state.total_size() + chunk_size > self.config.max_message_size {
789            return Ok(SmtpResponse::storage_exceeded(format!(
790                "Message size {} exceeds maximum {}",
791                bdat_state.total_size() + chunk_size,
792                self.config.max_message_size
793            )));
794        }
795
796        // If this is the LAST chunk and message will be complete, log it
797        if last {
798            let sender_display = self
799                .transaction
800                .sender
801                .as_ref()
802                .map(|a| a.to_string())
803                .unwrap_or_else(|| "<unknown>".to_string());
804            tracing::info!(
805                "BDAT LAST chunk ({} bytes) from {} with {} recipient(s)",
806                chunk_size,
807                sender_display,
808                self.transaction.recipients.len()
809            );
810        }
811
812        // Return success - caller must now read chunk_size bytes
813        Ok(SmtpResponse::ok(format!("{} octets received", chunk_size)))
814    }
815
816    /// Handle RSET command
817    async fn handle_rset(&mut self) -> anyhow::Result<SmtpResponse> {
818        self.transaction.reset();
819        self.state = SmtpState::Authenticated;
820        Ok(SmtpResponse::ok_simple())
821    }
822
823    /// Handle QUIT command
824    async fn handle_quit(&mut self) -> anyhow::Result<SmtpResponse> {
825        tracing::info!(
826            peer = %self.remote_addr,
827            "smtp.session QUIT received"
828        );
829        self.state = SmtpState::Quit;
830        Ok(SmtpResponse::closing())
831    }
832
833    /// Handle STARTTLS command
834    async fn handle_starttls(&mut self) -> anyhow::Result<SmtpResponse> {
835        if !self.config.enable_starttls {
836            return Ok(SmtpResponse::not_implemented("STARTTLS not available"));
837        }
838
839        // Record the STARTTLS upgrade in the metrics layer. The actual TLS upgrade is
840        // performed by the server-side stream wrapper (still being wired); we count the
841        // request-and-agree event here because that's the operationally meaningful signal
842        // (a client successfully negotiated an upgrade). When the upgrade is wired in,
843        // this call should remain — it is the right semantic event for the counter.
844        rusmes_metrics::global_metrics().inc_tls_session(rusmes_metrics::tls_label::STARTTLS);
845
846        Ok(SmtpResponse::new(220, "Ready to start TLS"))
847    }
848
849    /// Handle AUTH command
850    async fn handle_auth(
851        &mut self,
852        mechanism: String,
853        initial_response: Option<String>,
854    ) -> anyhow::Result<SmtpResponse> {
855        if !self.config.require_auth {
856            return Ok(SmtpResponse::not_implemented("AUTH not available"));
857        }
858
859        match mechanism.to_uppercase().as_str() {
860            "CRAM-MD5" => self.handle_auth_cram_md5().await,
861            "SCRAM-SHA-256" => self.handle_auth_scram_sha256(initial_response).await,
862            "PLAIN" => {
863                if let Some(response) = initial_response {
864                    self.handle_auth_plain(response).await
865                } else {
866                    // Request credentials
867                    Ok(SmtpResponse::new(334, ""))
868                }
869            }
870            "LOGIN" => {
871                // LOGIN authentication requires multi-step exchange
872                // Send "334 VXNlcm5hbWU6" (Username: in base64)
873                Ok(SmtpResponse::new(334, "VXNlcm5hbWU6"))
874            }
875            _ => Ok(SmtpResponse::parameter_not_implemented(
876                "Authentication mechanism not supported",
877            )),
878        }
879    }
880
881    /// Handle PLAIN authentication
882    async fn handle_auth_plain(&mut self, response: String) -> anyhow::Result<SmtpResponse> {
883        // Parse credentials
884        let (username, password) = match crate::auth::parse_plain_auth(&response) {
885            Ok(creds) => creds,
886            Err(e) => {
887                tracing::warn!("Failed to parse PLAIN auth: {}", e);
888                rusmes_metrics::global_metrics().inc_smtp_auth_failure();
889                return Ok(SmtpResponse::new(535, "5.7.8 Authentication failed"));
890            }
891        };
892
893        // Create Username object
894        let username_obj = match rusmes_proto::Username::new(username.clone()) {
895            Ok(u) => u,
896            Err(e) => {
897                tracing::warn!("Invalid username '{}': {}", username, e);
898                rusmes_metrics::global_metrics().inc_smtp_auth_failure();
899                return Ok(SmtpResponse::new(535, "5.7.8 Authentication failed"));
900            }
901        };
902
903        // Authenticate with backend
904        match self
905            .auth_backend
906            .authenticate(&username_obj, &password)
907            .await
908        {
909            Ok(true) => {
910                self.authenticated = true;
911                self.username = Some(username.clone());
912                tracing::info!(
913                    peer = %self.remote_addr,
914                    username = %username,
915                    mechanism = "PLAIN",
916                    "smtp.session AUTH success"
917                );
918                rusmes_metrics::global_metrics().inc_smtp_auth_success();
919                Ok(SmtpResponse::new(235, "2.7.0 Authentication successful"))
920            }
921            Ok(false) => {
922                tracing::warn!(
923                    peer = %self.remote_addr,
924                    username = %username,
925                    mechanism = "PLAIN",
926                    "smtp.session AUTH failure"
927                );
928                rusmes_metrics::global_metrics().inc_smtp_auth_failure();
929                Ok(SmtpResponse::new(535, "5.7.8 Authentication failed"))
930            }
931            Err(e) => {
932                tracing::error!(
933                    peer = %self.remote_addr,
934                    username = %username,
935                    error = %e,
936                    "smtp.session AUTH backend error"
937                );
938                rusmes_metrics::global_metrics().inc_smtp_auth_failure();
939                Ok(SmtpResponse::new(535, "5.7.8 Authentication failed"))
940            }
941        }
942    }
943
944    /// Handle CRAM-MD5 authentication - send challenge
945    async fn handle_auth_cram_md5(&mut self) -> anyhow::Result<SmtpResponse> {
946        // Generate challenge
947        let challenge = crate::auth::generate_cram_md5_challenge(&self.config.hostname)?;
948
949        // Store challenge for verification
950        self.cram_md5_challenge = Some(challenge.clone());
951
952        // Encode and send challenge
953        let encoded = crate::auth::encode_challenge(&challenge);
954        Ok(SmtpResponse::new(334, encoded))
955    }
956
957    /// Handle CRAM-MD5 response
958    async fn handle_cram_md5_response(
959        &mut self,
960        response_line: &str,
961    ) -> anyhow::Result<SmtpResponse> {
962        // Get the challenge (must be set)
963        let challenge = self
964            .cram_md5_challenge
965            .take()
966            .ok_or_else(|| anyhow::anyhow!("No CRAM-MD5 challenge pending"))?;
967
968        // Check for SASL abort
969        if response_line == "*" {
970            tracing::info!("CRAM-MD5 authentication aborted by client");
971            return Ok(SmtpResponse::new(501, "5.7.0 Authentication aborted"));
972        }
973
974        // Decode response
975        let decoded = crate::auth::decode_response(response_line)?;
976
977        // Parse username and HMAC
978        let (username, client_hmac) = crate::auth::parse_cram_md5_response(&decoded)?;
979
980        // IMPORTANT: CRAM-MD5 requires plaintext passwords or password-equivalent secrets
981        // The current AuthBackend uses bcrypt, which is one-way hashing
982        // For CRAM-MD5 to work, we would need:
983        // 1. A separate plaintext password store (security risk)
984        // 2. A password-equivalent secret store
985        // 3. A different authentication backend
986        //
987        // For now, we'll try to authenticate but it will fail with bcrypt
988        // This is documented limitation - CRAM-MD5 is not compatible with secure password storage
989
990        tracing::warn!(
991            "CRAM-MD5 authentication attempted for user '{}', but cannot verify HMAC with bcrypt-hashed passwords",
992            username
993        );
994
995        // We cannot compute the expected HMAC without the plaintext password
996        // The proper solution would be to store CRAM-MD5 secrets separately
997        // or use a more modern authentication mechanism like SCRAM
998
999        // For demonstration purposes, we log the authentication attempt
1000        tracing::info!(
1001            "CRAM-MD5 authentication for user '{}' from {} - challenge: {}, client_hmac: {}",
1002            username,
1003            self.remote_addr,
1004            challenge,
1005            client_hmac
1006        );
1007
1008        // Check if user exists
1009        let username_obj = rusmes_proto::Username::new(username.to_string())
1010            .map_err(|e| anyhow::anyhow!("Invalid username: {}", e))?;
1011
1012        let user_exists = self.auth_backend.verify_identity(&username_obj).await?;
1013
1014        if !user_exists {
1015            tracing::warn!(
1016                "CRAM-MD5 authentication failed: user '{}' does not exist",
1017                username
1018            );
1019            rusmes_metrics::global_metrics().inc_smtp_auth_failure();
1020            return Ok(SmtpResponse::new(
1021                535,
1022                "5.7.8 Authentication credentials invalid",
1023            ));
1024        }
1025
1026        // Since we cannot verify HMAC with bcrypt, reject the authentication
1027        // In a real implementation with plaintext or reversible password storage,
1028        // we would:
1029        // 1. Get password from auth backend
1030        // 2. Compute expected HMAC: compute_cram_md5_hmac(password, challenge)
1031        // 3. Compare with client_hmac (constant-time comparison)
1032
1033        tracing::warn!(
1034            "CRAM-MD5 authentication rejected: mechanism requires plaintext password storage"
1035        );
1036
1037        rusmes_metrics::global_metrics().inc_smtp_auth_failure();
1038        Ok(SmtpResponse::new(
1039            535,
1040            "5.7.8 Authentication credentials invalid",
1041        ))
1042    }
1043
1044    /// Check if relay is allowed for the given recipient
1045    ///
1046    /// Returns `true` if:
1047    /// - User is authenticated, OR
1048    /// - Client IP is in relay_networks (CIDR notation), OR
1049    /// - Recipient domain is a local domain
1050    fn is_relay_allowed(&self, recipient: &MailAddress) -> bool {
1051        // Allow if authenticated
1052        if self.authenticated {
1053            tracing::debug!(
1054                "Relay allowed for {} from {}: authenticated user",
1055                recipient,
1056                self.remote_addr.ip()
1057            );
1058            return true;
1059        }
1060
1061        // Allow if client IP is in relay_networks
1062        if crate::is_ip_in_networks(self.remote_addr.ip(), &self.config.relay_networks) {
1063            tracing::debug!(
1064                "Relay allowed for {} from {}: client IP in relay_networks",
1065                recipient,
1066                self.remote_addr.ip()
1067            );
1068            return true;
1069        }
1070
1071        // Allow if recipient is local domain
1072        let recipient_domain = recipient.domain().as_str();
1073        for local_domain in &self.config.local_domains {
1074            if recipient_domain.eq_ignore_ascii_case(local_domain) {
1075                tracing::debug!(
1076                    "Relay allowed for {} from {}: local domain",
1077                    recipient,
1078                    self.remote_addr.ip()
1079                );
1080                return true;
1081            }
1082        }
1083
1084        // Deny relay
1085        tracing::warn!(
1086            "Relay denied for {} from {}: not authenticated, not in relay_networks, not local domain",
1087            recipient,
1088            self.remote_addr.ip()
1089        );
1090        false
1091    }
1092
1093    /// Validate recipient against storage backend with caching
1094    async fn validate_recipient(&self, recipient: &MailAddress) -> anyhow::Result<bool> {
1095        // Cache TTL: 5 minutes
1096        const CACHE_TTL: Duration = Duration::from_secs(300);
1097
1098        let cache_key = recipient.as_string();
1099
1100        // Check cache first
1101        {
1102            let cache = self.recipient_cache.read().await;
1103            if let Some(entry) = cache.get(&cache_key) {
1104                if entry.cached_at.elapsed() < CACHE_TTL {
1105                    tracing::debug!("Recipient validation cache hit for {}", recipient);
1106                    return Ok(entry.exists);
1107                }
1108            }
1109        }
1110
1111        // Cache miss or expired, query storage backend
1112        tracing::debug!(
1113            "Recipient validation cache miss for {}, querying storage",
1114            recipient
1115        );
1116
1117        // Extract username from mail address
1118        let username = Username::new(recipient.local_part())?;
1119
1120        // Query storage backend for mailboxes
1121        let mailbox_store = self.storage_backend.mailbox_store();
1122        let mailboxes = mailbox_store.list_mailboxes(&username).await?;
1123
1124        let exists = !mailboxes.is_empty();
1125
1126        // Update cache
1127        {
1128            let mut cache = self.recipient_cache.write().await;
1129            cache.insert(
1130                cache_key,
1131                RecipientCacheEntry {
1132                    exists,
1133                    cached_at: Instant::now(),
1134                },
1135            );
1136        }
1137
1138        Ok(exists)
1139    }
1140
1141    /// Handle SCRAM-SHA-256 authentication - initial client-first message
1142    async fn handle_auth_scram_sha256(
1143        &mut self,
1144        initial_response: Option<String>,
1145    ) -> anyhow::Result<SmtpResponse> {
1146        // SCRAM-SHA-256 requires client-first message
1147        // If not provided, send 334 continuation to request it
1148        let client_first_encoded = match initial_response {
1149            Some(resp) => resp,
1150            None => {
1151                // Send initial continuation to request client-first
1152                return Ok(SmtpResponse::new(334, ""));
1153            }
1154        };
1155
1156        // Decode client-first message
1157        let client_first_decoded = BASE64
1158            .decode(client_first_encoded.trim())
1159            .map_err(|e| anyhow::anyhow!("Failed to decode client-first: {}", e))?;
1160        let client_first_str = String::from_utf8(client_first_decoded)
1161            .map_err(|e| anyhow::anyhow!("Failed to decode UTF-8: {}", e))?;
1162
1163        // Parse client-first message
1164        let (username, client_nonce, client_first_bare) =
1165            crate::auth::parse_scram_client_first(&client_first_str)?;
1166
1167        // Generate server nonce and combine with client nonce
1168        let server_nonce = crate::auth::generate_scram_server_nonce()?;
1169        let nonce = format!("{}{}", client_nonce, server_nonce);
1170
1171        // Fetch the user's RFC 5802 SCRAM credential bundle from the auth backend.
1172        //
1173        // `Ok(None)` means the backend has no SCRAM material for this user — either
1174        // it does not support SCRAM at all (SQL/LDAP/OAuth2 default), or the user
1175        // exists but has not been enrolled in SCRAM. Per RFC 4954 §4 we respond
1176        // with 504 5.5.4 (mechanism not available); the client may fall back to
1177        // PLAIN/LOGIN which use the bcrypt password.
1178        let creds = match self.auth_backend.fetch_scram_credentials(&username).await {
1179            Ok(Some(creds)) => creds,
1180            Ok(None) => {
1181                tracing::info!(
1182                    "SCRAM-SHA-256 declined for user '{}': no SCRAM credentials stored",
1183                    username
1184                );
1185                // 504 means "mechanism not available", not an auth failure per se, but we
1186                // record it as a failure so operators can track declining SCRAM attempts.
1187                rusmes_metrics::global_metrics().inc_smtp_auth_failure();
1188                return Ok(SmtpResponse::new(
1189                    504,
1190                    "5.5.4 SCRAM-SHA-256 mechanism not available for this user",
1191                ));
1192            }
1193            Err(e) => {
1194                tracing::error!(
1195                    "SCRAM-SHA-256 credential lookup failed for user '{}': {}",
1196                    username,
1197                    e
1198                );
1199                rusmes_metrics::global_metrics().inc_smtp_auth_failure();
1200                return Ok(SmtpResponse::new(
1201                    454,
1202                    "4.7.0 Temporary authentication failure",
1203                ));
1204            }
1205        };
1206
1207        // Build server-first message from the user's actual SCRAM credentials.
1208        let server_first = format!(
1209            "r={},s={},i={}",
1210            nonce,
1211            BASE64.encode(&creds.salt),
1212            creds.iteration_count
1213        );
1214
1215        // Store state (including the credential bundle) for the verification round.
1216        self.scram_state = Some(ScramState {
1217            client_first_bare: client_first_bare.clone(),
1218            server_first: server_first.clone(),
1219            nonce: nonce.clone(),
1220            username: username.clone(),
1221            stored_key: creds.stored_key,
1222            server_key: creds.server_key,
1223        });
1224
1225        // Send server-first as base64-encoded 334 response
1226        let server_first_encoded = BASE64.encode(server_first.as_bytes());
1227        Ok(SmtpResponse::new(334, server_first_encoded))
1228    }
1229
1230    /// Handle SCRAM-SHA-256 client-final message
1231    async fn handle_scram_client_final(
1232        &mut self,
1233        client_final_line: &str,
1234    ) -> anyhow::Result<SmtpResponse> {
1235        // Get stored state
1236        let state = self
1237            .scram_state
1238            .take()
1239            .ok_or_else(|| anyhow::anyhow!("No SCRAM state"))?;
1240
1241        // Check for SASL abort
1242        if client_final_line == "*" {
1243            tracing::info!("SCRAM-SHA-256 authentication aborted by client");
1244            return Ok(SmtpResponse::new(501, "5.7.0 Authentication aborted"));
1245        }
1246
1247        // Decode client-final message
1248        let client_final_decoded = BASE64
1249            .decode(client_final_line.trim())
1250            .map_err(|e| anyhow::anyhow!("Failed to decode client-final: {}", e))?;
1251        let client_final_str = String::from_utf8(client_final_decoded)
1252            .map_err(|e| anyhow::anyhow!("Failed to decode UTF-8: {}", e))?;
1253
1254        // Parse client-final message
1255        let (_channel_binding, nonce, proof, client_final_without_proof) =
1256            crate::auth::parse_scram_client_final(&client_final_str)?;
1257
1258        // Verify nonce matches
1259        if nonce != state.nonce {
1260            tracing::warn!(
1261                "SCRAM-SHA-256 nonce mismatch for user '{}': expected '{}', got '{}'",
1262                state.username,
1263                state.nonce,
1264                nonce
1265            );
1266            rusmes_metrics::global_metrics().inc_smtp_auth_failure();
1267            return Ok(SmtpResponse::new(
1268                535,
1269                "5.7.8 Authentication credentials invalid",
1270            ));
1271        }
1272
1273        // RFC 5802 §3 AuthMessage =
1274        //   client-first-message-bare + "," + server-first-message + "," + client-final-message-without-proof
1275        let auth_message = format!(
1276            "{},{},{}",
1277            state.client_first_bare, state.server_first, client_final_without_proof
1278        );
1279
1280        // Verify the client proof against the stored key.
1281        let proof_valid =
1282            crate::auth::verify_scram_client_proof(&state.stored_key, &auth_message, &proof)?;
1283
1284        if !proof_valid {
1285            tracing::warn!(
1286                "SCRAM-SHA-256 authentication failed for user '{}': client proof did not verify",
1287                state.username
1288            );
1289            rusmes_metrics::global_metrics().inc_smtp_auth_failure();
1290            return Ok(SmtpResponse::new(
1291                535,
1292                "5.7.8 Authentication credentials invalid",
1293            ));
1294        }
1295
1296        // Compute the server signature and embed it in the success response per
1297        // RFC 4954 §6 (additional success data is base64-encoded onto the 235 line).
1298        let server_signature =
1299            crate::auth::compute_scram_server_signature(&state.server_key, &auth_message)?;
1300        let server_final = format!("v={}", server_signature);
1301        let server_final_b64 = BASE64.encode(server_final.as_bytes());
1302
1303        // Mark the session as authenticated and bind the username.
1304        self.authenticated = true;
1305        self.username = Some(state.username.clone());
1306
1307        tracing::info!(
1308            "User '{}' authenticated successfully (SCRAM-SHA-256)",
1309            state.username
1310        );
1311
1312        rusmes_metrics::global_metrics().inc_smtp_auth_success();
1313        Ok(SmtpResponse::new(
1314            235,
1315            format!("2.7.0 {} Authentication successful", server_final_b64),
1316        ))
1317    }
1318}
1319
1320#[cfg(test)]
1321mod tests;
rusmes_smtp/session/mod.rs

rusmes_smtp/session/
mod.rs
