AUTH LOGIN support on IMAP, POP3, ManageSieve and SMTP submission (fixes #3)

CHANGELOG.md

@@ -2,6 +2,16 @@
 
 All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org/).
 
+## [1.0.2] - 2026-07-31
+
+### Added
+- `AUTH LOGIN` support on IMAP, POP3, ManageSieve and SMTP submission (#3).
+
+### Changed
+- `LOGIN` is now part of the default `capabilities.<protocol>.sasl` set.
+
+### Fixed
+
 ## [1.0.1] - 2026-07-30
 
 ### Added

Cargo.lock

@@ -2110,7 +2110,7 @@ dependencies = [
 
 [[package]]
 name = "proxy"
-version = "1.0.1"
+version = "1.0.2"
 dependencies = [
  "ahash",
  "arc-swap",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "proxy"
-version = "1.0.1"
+version = "1.0.2"
 edition = "2024"
 license = "AGPL-3.0-only OR LicenseRef-SEL"
 description = "Multi-protocol e-mail proxy"

src/admin.rs

@@ -191,8 +191,13 @@ async fn handle(
         match request.method.as_str() {
             "GET" => write_response(&mut stream, 200, "OK", "ok\n").await?,
             _ => {
-                write_response(&mut stream, 405, "Method Not Allowed", "method not allowed\n")
-                    .await?
+                write_response(
+                    &mut stream,
+                    405,
+                    "Method Not Allowed",
+                    "method not allowed\n",
+                )
+                .await?
             }
         }
         return Ok(());
@@ -653,8 +658,8 @@ mod tests {
     fn parse_trims_trailing_slash() {
         let req = parse_request(b"GET /stats/ HTTP/1.1\r\nHost: x\r\n").unwrap();
         assert_eq!(req.path, "/stats");
-        let req = parse_request(b"GET /cache/invalidate/?identifier=a HTTP/1.1\r\nHost: x\r\n")
-            .unwrap();
+        let req =
+            parse_request(b"GET /cache/invalidate/?identifier=a HTTP/1.1\r\nHost: x\r\n").unwrap();
         assert_eq!(req.path, "/cache/invalidate");
         assert_eq!(req.query_param("identifier").as_deref(), Some("a"));
         let req = parse_request(b"GET / HTTP/1.1\r\nHost: x\r\n").unwrap();

src/config/mod.rs

@@ -496,7 +496,8 @@ const DEFAULT_IMAP_ADVERTISE: &[&str] = &[
     "UTF8=ACCEPT",
     "JMAPACCESS",
 ];
-const DEFAULT_SASL: &[&str] = &["PLAIN", "OAUTHBEARER", "XOAUTH2"];
+const DEFAULT_SASL: &[&str] = &["PLAIN", "LOGIN", "OAUTHBEARER", "XOAUTH2"];
+pub const SUPPORTED_SASL: &[&str] = &["PLAIN", "LOGIN", "OAUTHBEARER", "XOAUTH2"];
 const DEFAULT_IMAP_BANNER: &str = "Stalwart IMAP4rev2 at your service.";
 
 fn default_imap_advertise() -> Vec<String> {
@@ -512,6 +513,30 @@ fn default_imap_banner() -> String {
     DEFAULT_IMAP_BANNER.to_string()
 }
 
+fn canonicalize_sasl_list(list: &mut [String], section: &str) -> Result<()> {
+    let mut seen: Vec<&'static str> = Vec::with_capacity(list.len());
+    for entry in list.iter_mut() {
+        let Some(canonical) = SUPPORTED_SASL
+            .iter()
+            .find(|m| m.eq_ignore_ascii_case(entry))
+            .copied()
+        else {
+            return Err(ProxyError::config(format!(
+                "capabilities.{section}.sasl lists unsupported mechanism {entry:?}; supported mechanisms are {}",
+                SUPPORTED_SASL.join(", ")
+            )));
+        };
+        if seen.contains(&canonical) {
+            return Err(ProxyError::config(format!(
+                "capabilities.{section}.sasl lists mechanism {canonical} more than once"
+            )));
+        }
+        seen.push(canonical);
+        *entry = canonical.to_string();
+    }
+    Ok(())
+}
+
 #[derive(Debug, Deserialize)]
 pub struct ManageSieveCapabilities {
     #[serde(default = "default_sieve_ext")]
@@ -796,12 +821,21 @@ fn default_lockout_duration() -> Duration {
 
 impl Config {
     pub fn parse_and_validate(raw: &str) -> Result<Self> {
-        let config: Config =
+        let mut config: Config =
             toml::from_str(raw).map_err(|e| ProxyError::config(format!("TOML parse: {e}")))?;
+        config.canonicalize_sasl()?;
         config.validate()?;
         Ok(config)
     }
 
+    fn canonicalize_sasl(&mut self) -> Result<()> {
+        let caps = &mut self.capabilities;
+        canonicalize_sasl_list(&mut caps.imap.sasl, "imap")?;
+        canonicalize_sasl_list(&mut caps.managesieve.sasl, "managesieve")?;
+        canonicalize_sasl_list(&mut caps.submission.sasl, "submission")?;
+        canonicalize_sasl_list(&mut caps.pop3.sasl, "pop3")
+    }
+
     fn destination(&self, id: &str) -> Result<&DestinationConfig> {
         self.destination
             .get(id)

src/proto/common.rs

@@ -94,14 +94,39 @@ pub fn build_greetings(config: &Config) -> Greetings {
     }
 }
 
+pub const LOGIN_USERNAME_CHALLENGE: &str = "VXNlcm5hbWU6";
+pub const LOGIN_PASSWORD_CHALLENGE: &str = "UGFzc3dvcmQ6";
+
+pub fn is_cleartext_mechanism(mech: &str) -> bool {
+    mech.eq_ignore_ascii_case("PLAIN") || mech.eq_ignore_ascii_case("LOGIN")
+}
+
 pub fn advertised_sasl(sasl: &[String], plain_ok: bool) -> Vec<&str> {
     sasl.iter()
-        .filter(|m| !m.eq_ignore_ascii_case("LOGIN"))
-        .filter(|m| plain_ok || !m.eq_ignore_ascii_case("PLAIN"))
+        .filter(|m| plain_ok || !is_cleartext_mechanism(m))
         .map(|s| s.as_str())
         .collect()
 }
 
+pub fn sasl_offered(sasl: &[String], mech: &str) -> bool {
+    sasl.iter().any(|m| m.eq_ignore_ascii_case(mech))
+}
+
+pub fn sasl_cancelled(response: &[u8]) -> bool {
+    response.trim_ascii() == b"*"
+}
+
+pub fn login_challenge(username_b64: &[u8], password_b64: &[u8]) -> Option<Vec<u8>> {
+    let username = crate::sasl::decode_sasl_ir(username_b64)?;
+    let password = crate::sasl::decode_sasl_ir(password_b64)?;
+    let mut raw = Vec::with_capacity(username.len() + password.len() + 2);
+    raw.push(0);
+    raw.extend_from_slice(&username);
+    raw.push(0);
+    raw.extend_from_slice(&password);
+    Some(raw)
+}
+
 pub fn advertises_token(haystack: &[u8], token: &str) -> bool {
     haystack
         .split(|b| *b == b'\n' || *b == b'\r')
@@ -904,6 +929,61 @@ mod tests {
         assert!(advertises_token(pop3, "XCLIENT"));
     }
 
+    fn sasl_list(mechs: &[&str]) -> Vec<String> {
+        mechs.iter().map(|s| s.to_string()).collect()
+    }
+
+    #[test]
+    fn advertised_sasl_keeps_login_when_cleartext_is_allowed() {
+        let sasl = sasl_list(&["PLAIN", "LOGIN", "OAUTHBEARER", "XOAUTH2"]);
+        assert_eq!(
+            advertised_sasl(&sasl, true),
+            vec!["PLAIN", "LOGIN", "OAUTHBEARER", "XOAUTH2"]
+        );
+    }
+
+    #[test]
+    fn advertised_sasl_drops_login_with_plain_when_cleartext_is_refused() {
+        let sasl = sasl_list(&["PLAIN", "LOGIN", "OAUTHBEARER", "XOAUTH2"]);
+        assert_eq!(
+            advertised_sasl(&sasl, false),
+            vec!["OAUTHBEARER", "XOAUTH2"]
+        );
+    }
+
+    #[test]
+    fn advertised_sasl_omits_login_when_not_configured() {
+        let sasl = sasl_list(&["PLAIN", "OAUTHBEARER"]);
+        assert_eq!(advertised_sasl(&sasl, true), vec!["PLAIN", "OAUTHBEARER"]);
+        assert!(!sasl_offered(&sasl, "LOGIN"));
+        assert!(sasl_offered(&sasl, "plain"));
+    }
+
+    #[test]
+    fn login_challenge_builds_a_plain_authentication_frame() {
+        use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
+        let user = BASE64.encode("user@example.com");
+        let pass = BASE64.encode("secret");
+        assert_eq!(
+            login_challenge(user.as_bytes(), pass.as_bytes()).unwrap(),
+            b"\0user@example.com\0secret".to_vec()
+        );
+    }
+
+    #[test]
+    fn login_challenge_rejects_invalid_base64() {
+        assert!(login_challenge(b"!!!!", b"c2VjcmV0").is_none());
+        assert!(login_challenge(b"dXNlcg==", b"!!!!").is_none());
+    }
+
+    #[test]
+    fn sasl_cancellation_is_a_bare_asterisk() {
+        assert!(sasl_cancelled(b"*"));
+        assert!(sasl_cancelled(b"*\r"));
+        assert!(!sasl_cancelled(b"KiA="));
+        assert!(!sasl_cancelled(b""));
+    }
+
     #[test]
     fn smtp_auth_line_classification() {
         assert!(matches!(

src/proto/imap.rs

@@ -325,19 +325,15 @@ impl Auth<'_> {
 
         match mech {
             Mechanism::Plain | Mechanism::OAuthBearer | Mechanism::XOauth2 => {}
-            Mechanism::Login => {
-                return self
-                    .reject(stream, tag, "AUTH LOGIN is not supported.")
-                    .await;
-            }
-            Mechanism::Other(_) => {
+            Mechanism::Login if common::sasl_offered(&self.caps.sasl, "LOGIN") => {}
+            _ => {
                 return self
                     .reject(stream, tag, "Unsupported authentication mechanism.")
                     .await;
             }
         }
 
-        if matches!(mech, Mechanism::Plain)
+        if matches!(mech, Mechanism::Plain | Mechanism::Login)
             && !plain_auth_allowed(self.caps.allow_plain_auth_without_tls, self.is_tls)
         {
             write(
@@ -349,6 +345,10 @@ impl Auth<'_> {
             return Ok(Step::Continue);
         }
 
+        if matches!(mech, Mechanism::Login) {
+            return self.authenticate_login(stream, tokens, tag, it).await;
+        }
+
         let (raw_b64, residual): (Vec<u8>, Vec<u8>) = if let Some(ir) = tokens.get(1) {
             (ir.as_bytes().to_vec(), it.as_slice().to_vec())
         } else {
@@ -366,6 +366,48 @@ impl Auth<'_> {
         self.finish(stream, mech, tag, raw, residual).await
     }
 
+    async fn authenticate_login(
+        &mut self,
+        stream: &mut BoxedStream,
+        tokens: &[Token],
+        tag: &str,
+        it: &mut std::slice::Iter<'_, u8>,
+    ) -> Result<Step> {
+        let (user_b64, residual): (Vec<u8>, Vec<u8>) = if let Some(ir) = tokens.get(1) {
+            (ir.as_bytes().to_vec(), it.as_slice().to_vec())
+        } else {
+            write(
+                &mut *stream,
+                format!("+ {}\r\n", common::LOGIN_USERNAME_CHALLENGE).as_bytes(),
+            )
+            .await?;
+            let seed = it.as_slice().to_vec();
+            read_crlf_line(stream, seed, MAX_PREAUTH).await?
+        };
+        if common::sasl_cancelled(&user_b64) {
+            return self.reject(stream, tag, "Authentication cancelled.").await;
+        }
+
+        write(
+            &mut *stream,
+            format!("+ {}\r\n", common::LOGIN_PASSWORD_CHALLENGE).as_bytes(),
+        )
+        .await?;
+        let (pass_b64, residual) = read_crlf_line(stream, residual, MAX_PREAUTH).await?;
+        if common::sasl_cancelled(&pass_b64) {
+            return self.reject(stream, tag, "Authentication cancelled.").await;
+        }
+
+        let Some(raw) = common::login_challenge(&user_b64, &pass_b64) else {
+            return self
+                .reject(stream, tag, "Invalid base64 in SASL response.")
+                .await;
+        };
+
+        self.finish(stream, Mechanism::Plain, tag, raw, residual)
+            .await
+    }
+
     async fn login(
         &mut self,
         stream: &mut BoxedStream,

src/proto/managesieve.rs

@@ -173,6 +173,7 @@ async fn preauth(
                                 ctx,
                                 &mut stream,
                                 plain_ok,
+                                common::sasl_offered(&caps.sasl, "LOGIN"),
                                 &req.tokens,
                                 &mut it,
                                 &mut auth_failures,
@@ -239,10 +240,12 @@ enum AuthStep {
     Close,
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn handle_authenticate(
     ctx: &Ctx,
     stream: &mut BoxedStream,
     plain_ok: bool,
+    login_offered: bool,
     tokens: &[crate::imap_receiver::Token],
     it: &mut std::slice::Iter<'_, u8>,
     auth_failures: &mut u32,
@@ -255,9 +258,10 @@ async fn handle_authenticate(
     let mech = Mechanism::parse(mech_tok.as_bytes());
     match mech {
         Mechanism::Plain | Mechanism::OAuthBearer | Mechanism::XOauth2 => {}
+        Mechanism::Login if login_offered => {}
         _ => return fail(stream, "Unsupported mechanism", auth_failures, max_attempts).await,
     }
-    if matches!(mech, Mechanism::Plain) && !plain_ok {
+    if matches!(mech, Mechanism::Plain | Mechanism::Login) && !plain_ok {
         write(
             &mut *stream,
             b"NO \"Cleartext authentication requires TLS\"\r\n",
@@ -266,6 +270,10 @@ async fn handle_authenticate(
         return Ok(AuthStep::Continue);
     }
 
+    if matches!(mech, Mechanism::Login) {
+        return authenticate_login(ctx, stream, tokens, it, auth_failures, max_attempts).await;
+    }
+
     let (raw_b64, residual): (Vec<u8>, Vec<u8>) = if let Some(ir) = tokens.get(1) {
         (ir.as_bytes().to_vec(), it.as_slice().to_vec())
     } else {
@@ -290,6 +298,67 @@ async fn handle_authenticate(
     })
 }
 
+async fn authenticate_login(
+    ctx: &Ctx,
+    stream: &mut BoxedStream,
+    tokens: &[crate::imap_receiver::Token],
+    it: &mut std::slice::Iter<'_, u8>,
+    auth_failures: &mut u32,
+    max_attempts: u32,
+) -> Result<AuthStep> {
+    let (user_b64, residual): (Vec<u8>, Vec<u8>) = if let Some(ir) = tokens.get(1) {
+        (ir.as_bytes().to_vec(), it.as_slice().to_vec())
+    } else {
+        write_challenge(stream, common::LOGIN_USERNAME_CHALLENGE).await?;
+        let seed = it.as_slice().to_vec();
+        read_sasl_response(stream, seed).await?
+    };
+    if common::sasl_cancelled(&user_b64) {
+        return fail(
+            stream,
+            "Authentication cancelled",
+            auth_failures,
+            max_attempts,
+        )
+        .await;
+    }
+
+    write_challenge(stream, common::LOGIN_PASSWORD_CHALLENGE).await?;
+    let (pass_b64, residual) = read_sasl_response(stream, residual).await?;
+    if common::sasl_cancelled(&pass_b64) {
+        return fail(
+            stream,
+            "Authentication cancelled",
+            auth_failures,
+            max_attempts,
+        )
+        .await;
+    }
+
+    let Some(raw) = common::login_challenge(&user_b64, &pass_b64) else {
+        return fail(stream, "Invalid base64", auth_failures, max_attempts).await;
+    };
+    let Some(cred) = credential_from_sasl(&Mechanism::Plain, &raw) else {
+        return fail(stream, "Malformed credentials", auth_failures, max_attempts).await;
+    };
+    let identifier = identifier_from_credential(&cred, &ctx.config.oauth.jwt_username_claim);
+
+    Ok(AuthStep::Done {
+        mech: Mechanism::Plain,
+        raw_challenge: raw,
+        identifier,
+        residual,
+    })
+}
+
+async fn write_challenge(stream: &mut BoxedStream, challenge: &str) -> Result<()> {
+    write(
+        &mut *stream,
+        format!("{{{}}}\r\n{challenge}\r\n", challenge.len()).as_bytes(),
+    )
+    .await
+}
+
 async fn fail(
     stream: &mut BoxedStream,
     message: &str,
@@ -325,7 +394,15 @@ async fn read_sasl_response(stream: &mut BoxedStream, seed: Vec<u8>) -> Result<(
         }
         Ok((data, rest))
     } else {
-        Ok((line, residual))
+        Ok((unquote(line), residual))
+    }
+}
+
+fn unquote(line: Vec<u8>) -> Vec<u8> {
+    if line.len() >= 2 && line.first() == Some(&b'"') && line.last() == Some(&b'"') {
+        line[1..line.len() - 1].to_vec()
+    } else {
+        line
     }
 }
 

src/proto/pop3.rs

@@ -220,6 +220,7 @@ async fn preauth(
                 let mech = Mechanism::parse(mech_bytes);
                 match &mech {
                     Mechanism::Plain | Mechanism::OAuthBearer | Mechanism::XOauth2 => {}
+                    Mechanism::Login if common::sasl_offered(&caps.sasl, "LOGIN") => {}
                     _ => {
                         write(&mut stream, b"-ERR unsupported mechanism\r\n").await?;
                         buf.drain(..consumed);
@@ -229,7 +230,7 @@ async fn preauth(
                         continue;
                     }
                 }
-                if matches!(mech, Mechanism::Plain) && !plain_ok {
+                if matches!(mech, Mechanism::Plain | Mechanism::Login) && !plain_ok {
                     write(
                         &mut stream,
                         b"-ERR cleartext authentication requires TLS\r\n",
@@ -239,6 +240,68 @@ async fn preauth(
                     continue;
                 }
 
+                if matches!(mech, Mechanism::Login) {
+                    buf.drain(..consumed);
+                    let seed = std::mem::take(&mut buf);
+                    let (user_b64, rest) = if ir.is_empty() {
+                        write(
+                            &mut stream,
+                            format!("+ {}\r\n", common::LOGIN_USERNAME_CHALLENGE).as_bytes(),
+                        )
+                        .await?;
+                        common::read_crlf_line(&mut stream, seed, MAX_PREAUTH).await?
+                    } else {
+                        (ir.to_vec(), seed)
+                    };
+                    if common::sasl_cancelled(&user_b64) {
+                        buf = rest;
+                        write(&mut stream, b"-ERR authentication cancelled\r\n").await?;
+                        if bump(&mut auth_failures, listener.max_auth_attempts) {
+                            return Ok(None);
+                        }
+                        continue;
+                    }
+                    write(
+                        &mut stream,
+                        format!("+ {}\r\n", common::LOGIN_PASSWORD_CHALLENGE).as_bytes(),
+                    )
+                    .await?;
+                    let (pass_b64, rest) =
+                        common::read_crlf_line(&mut stream, rest, MAX_PREAUTH).await?;
+                    buf = rest;
+                    if common::sasl_cancelled(&pass_b64) {
+                        write(&mut stream, b"-ERR authentication cancelled\r\n").await?;
+                        if bump(&mut auth_failures, listener.max_auth_attempts) {
+                            return Ok(None);
+                        }
+                        continue;
+                    }
+                    let Some(raw) = common::login_challenge(&user_b64, &pass_b64) else {
+                        write(&mut stream, b"-ERR invalid base64\r\n").await?;
+                        if bump(&mut auth_failures, listener.max_auth_attempts) {
+                            return Ok(None);
+                        }
+                        continue;
+                    };
+                    let Some(cred) = credential_from_sasl(&Mechanism::Plain, &raw) else {
+                        write(&mut stream, b"-ERR malformed credentials\r\n").await?;
+                        if bump(&mut auth_failures, listener.max_auth_attempts) {
+                            return Ok(None);
+                        }
+                        continue;
+                    };
+                    let identifier =
+                        identifier_from_credential(&cred, &ctx.config.oauth.jwt_username_claim);
+                    let residual = std::mem::take(&mut buf);
+                    return Ok(Some(Handoff {
+                        stream,
+                        mech: Mechanism::Plain,
+                        raw_challenge: raw,
+                        identifier,
+                        residual,
+                    }));
+                }
+
                 let raw_b64: Vec<u8> = if !ir.is_empty() {
                     let ir = ir.to_vec();
                     buf.drain(..consumed);

src/proto/smtp.rs

@@ -7,7 +7,9 @@
 use std::net::SocketAddr;
 
 use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
-use smtp_proto::{AUTH_OAUTHBEARER, AUTH_PLAIN, AUTH_XOAUTH2, Error as SmtpError, Request};
+use smtp_proto::{
+    AUTH_LOGIN, AUTH_OAUTHBEARER, AUTH_PLAIN, AUTH_XOAUTH2, Error as SmtpError, Request,
+};
 use tokio::time::timeout;
 
 use crate::config::{Forwarding, ListenerConfig, Protocol, SubmissionCapabilities, TlsMode};
@@ -181,6 +183,7 @@ async fn preauth(
                         let params = AuthParams {
                             ctx,
                             plain_ok,
+                            login_offered: common::sasl_offered(&caps.sasl, "LOGIN"),
                             ehlo_host: &ehlo_host,
                             max_attempts: listener.max_auth_attempts,
                         };
@@ -260,6 +263,7 @@ enum AuthStep {
 struct AuthParams<'a> {
     ctx: &'a Ctx,
     plain_ok: bool,
+    login_offered: bool,
     ehlo_host: &'a str,
     max_attempts: u32,
 }
@@ -275,7 +279,11 @@ async fn handle_auth(
 ) -> Result<AuthStep> {
     let max_attempts = params.max_attempts;
 
-    let Some(mech) = mech_from_flag(mechanism) else {
+    let mech = match mech_from_flag(mechanism) {
+        Some(Mechanism::Login) if !params.login_offered => None,
+        other => other,
+    };
+    let Some(mech) = mech else {
         return fail(
             stream,
             buf,
@@ -286,7 +294,7 @@ async fn handle_auth(
         )
         .await;
     };
-    if matches!(mech, Mechanism::Plain) && !params.plain_ok {
+    if matches!(mech, Mechanism::Plain | Mechanism::Login) && !params.plain_ok {
         write(
             &mut *stream,
             b"538 5.7.11 Encryption required for requested authentication mechanism\r\n",
@@ -296,6 +304,18 @@ async fn handle_auth(
         return Ok(AuthStep::Continue);
     }
 
+    if matches!(mech, Mechanism::Login) {
+        return handle_auth_login(
+            params,
+            stream,
+            buf,
+            consumed,
+            initial_response,
+            auth_failures,
+        )
+        .await;
+    }
+
     let raw_b64: Vec<u8> = if !initial_response.is_empty() {
         let ir = initial_response.as_bytes().to_vec();
         buf.drain(..consumed);
@@ -343,6 +363,96 @@ async fn handle_auth(
     })
 }
 
+async fn handle_auth_login(
+    params: &AuthParams<'_>,
+    stream: &mut BoxedStream,
+    buf: &mut Vec<u8>,
+    consumed: usize,
+    initial_response: &str,
+    auth_failures: &mut u32,
+) -> Result<AuthStep> {
+    let max_attempts = params.max_attempts;
+
+    let user_b64: Vec<u8> = if !initial_response.is_empty() {
+        let ir = initial_response.as_bytes().to_vec();
+        buf.drain(..consumed);
+        ir
+    } else {
+        write(
+            &mut *stream,
+            format!("334 {}\r\n", common::LOGIN_USERNAME_CHALLENGE).as_bytes(),
+        )
+        .await?;
+        buf.drain(..consumed);
+        let seed = std::mem::take(buf);
+        let (line, rest) = common::read_crlf_line(stream, seed, MAX_PREAUTH).await?;
+        *buf = rest;
+        line
+    };
+    if common::sasl_cancelled(&user_b64) {
+        return cancelled(stream, auth_failures, max_attempts).await;
+    }
+
+    write(
+        &mut *stream,
+        format!("334 {}\r\n", common::LOGIN_PASSWORD_CHALLENGE).as_bytes(),
+    )
+    .await?;
+    let seed = std::mem::take(buf);
+    let (pass_b64, rest) = common::read_crlf_line(stream, seed, MAX_PREAUTH).await?;
+    *buf = rest;
+    if common::sasl_cancelled(&pass_b64) {
+        return cancelled(stream, auth_failures, max_attempts).await;
+    }
+
+    let Some(raw) = common::login_challenge(&user_b64, &pass_b64) else {
+        return fail(
+            stream,
+            buf,
+            0,
+            "5.7.8 Invalid base64",
+            auth_failures,
+            max_attempts,
+        )
+        .await;
+    };
+    let Some(cred) = credential_from_sasl(&Mechanism::Plain, &raw) else {
+        return fail(
+            stream,
+            buf,
+            0,
+            "5.7.8 Malformed credentials",
+            auth_failures,
+            max_attempts,
+        )
+        .await;
+    };
+    let identifier = identifier_from_credential(&cred, &params.ctx.config.oauth.jwt_username_claim);
+    let residual = std::mem::take(buf);
+
+    Ok(AuthStep::Done {
+        mech: Mechanism::Plain,
+        raw_challenge: raw,
+        identifier,
+        residual,
+        ehlo_host: params.ehlo_host.to_string(),
+    })
+}
+
+async fn cancelled(
+    stream: &mut BoxedStream,
+    auth_failures: &mut u32,
+    max_attempts: u32,
+) -> Result<AuthStep> {
+    write(&mut *stream, b"501 5.7.0 Authentication cancelled\r\n").await?;
+    *auth_failures += 1;
+    if *auth_failures >= max_attempts {
+        Ok(AuthStep::Close)
+    } else {
+        Ok(AuthStep::Continue)
+    }
+}
+
 async fn fail(
     stream: &mut BoxedStream,
     buf: &mut Vec<u8>,
@@ -370,6 +480,8 @@ fn mech_from_flag(flag: u64) -> Option<Mechanism> {
         Some(Mechanism::OAuthBearer)
     } else if flag & AUTH_XOAUTH2 != 0 {
         Some(Mechanism::XOauth2)
+    } else if flag & AUTH_LOGIN != 0 {
+        Some(Mechanism::Login)
     } else {
         None
     }

src/tests/config_validation.rs

@@ -364,3 +364,140 @@ allow_plain_auth_without_tls = true
     );
     assert!(Config::parse_and_validate(&toml).is_ok());
 }
+
+fn submission_config_with_sasl(mapping: &str, sasl: &str) -> String {
+    format!(
+        r#"
+{}
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+proxy_protocol = false
+allow_plaintext_auth = true
+[destination.legacy.protocol.submission]
+port = 587
+tls = "plain"
+
+[listener.submission]
+protocol = "submission"
+bind = ["127.0.0.1:587"]
+tls = "plain"
+
+[capabilities.submission]
+allow_plain_auth_without_tls = true
+sasl = {sasl}
+"#,
+        mapping_block(mapping)
+    )
+}
+
+#[test]
+fn default_sasl_set_includes_login() {
+    let m = mapping_file("");
+    let toml = format!(
+        r#"
+{}
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+proxy_protocol = false
+allow_plaintext_auth = true
+[destination.legacy.protocol.submission]
+port = 587
+tls = "plain"
+
+[listener.submission]
+protocol = "submission"
+bind = ["127.0.0.1:587"]
+tls = "plain"
+"#,
+        mapping_block(&m.path().display().to_string())
+    );
+    let config = Config::parse_and_validate(&toml).unwrap();
+    for sasl in [
+        &config.capabilities.submission.sasl,
+        &config.capabilities.imap.sasl,
+        &config.capabilities.pop3.sasl,
+        &config.capabilities.managesieve.sasl,
+    ] {
+        assert!(
+            sasl.iter().any(|m| m == "LOGIN"),
+            "LOGIN missing from default sasl set {sasl:?}"
+        );
+    }
+}
+
+#[test]
+fn accepts_login_in_sasl_and_canonicalizes_case() {
+    let m = mapping_file("");
+    let toml = submission_config_with_sasl(
+        &m.path().display().to_string(),
+        r#"["plain", "Login", "oauthbearer"]"#,
+    );
+    let config = Config::parse_and_validate(&toml).unwrap();
+    assert_eq!(
+        config.capabilities.submission.sasl,
+        vec!["PLAIN", "LOGIN", "OAUTHBEARER"]
+    );
+}
+
+#[test]
+fn rejects_unsupported_sasl_mechanism() {
+    let m = mapping_file("");
+    let toml =
+        submission_config_with_sasl(&m.path().display().to_string(), r#"["PLAIN", "CRAM-MD5"]"#);
+    let err = Config::parse_and_validate(&toml).unwrap_err().to_string();
+    assert!(
+        err.contains("CRAM-MD5") && err.contains("capabilities.submission.sasl"),
+        "unexpected error: {err}"
+    );
+}
+
+#[test]
+fn rejects_duplicate_sasl_mechanism() {
+    let m = mapping_file("");
+    let toml =
+        submission_config_with_sasl(&m.path().display().to_string(), r#"["PLAIN", "plain"]"#);
+    let err = Config::parse_and_validate(&toml).unwrap_err().to_string();
+    assert!(err.contains("more than once"), "unexpected error: {err}");
+}
+
+#[test]
+fn rejects_unsupported_sasl_mechanism_in_every_section() {
+    let m = mapping_file("");
+    for section in ["imap", "pop3", "managesieve"] {
+        let toml = format!(
+            r#"
+{}
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+proxy_protocol = false
+allow_plaintext_auth = true
+[destination.legacy.protocol.imap]
+port = 143
+tls = "plain"
+
+[listener.imap]
+protocol = "imap"
+bind = ["127.0.0.1:143"]
+tls = "plain"
+
+[capabilities.{section}]
+sasl = ["NTLM"]
+"#,
+            mapping_block(&m.path().display().to_string())
+        );
+        let err = Config::parse_and_validate(&toml).unwrap_err().to_string();
+        assert!(
+            err.contains(&format!("capabilities.{section}.sasl")),
+            "unexpected error for {section}: {err}"
+        );
+    }
+}

src/tests/containers/clients.rs

@@ -161,6 +161,68 @@ pub async fn submission_send(
     }
 }
 
+async fn submission_auth_login<S: AsyncRead + AsyncWrite + Unpin>(
+    stream: &mut S,
+    login: &str,
+    password: &str,
+) -> String {
+    stream.write_all(b"EHLO proxy.test\r\n").await.unwrap();
+    let ehlo = read_until(stream, "250 ").await;
+    assert!(
+        ehlo.to_uppercase().contains("LOGIN"),
+        "submission must advertise AUTH LOGIN: {ehlo:?}"
+    );
+
+    stream.write_all(b"AUTH LOGIN\r\n").await.unwrap();
+    let user_prompt = read_until(stream, "334 ").await;
+    assert!(
+        user_prompt.contains("VXNlcm5hbWU6"),
+        "expected a username challenge: {user_prompt:?}"
+    );
+    stream
+        .write_all(format!("{}\r\n", B64.encode(login)).as_bytes())
+        .await
+        .unwrap();
+
+    let pass_prompt = read_until(stream, "334 ").await;
+    assert!(
+        pass_prompt.contains("UGFzc3dvcmQ6"),
+        "expected a password challenge: {pass_prompt:?}"
+    );
+    stream
+        .write_all(format!("{}\r\n", B64.encode(password)).as_bytes())
+        .await
+        .unwrap();
+    read_reply(stream).await
+}
+
+pub async fn submission_send_auth_login(
+    host: &str,
+    port: u16,
+    tls: Tls,
+    login: &str,
+    password: &str,
+) -> String {
+    match tls {
+        Tls::Implicit => {
+            let tcp = TcpStream::connect((host, port)).await.unwrap();
+            let mut s = tls_connect_client(tcp).await;
+            let _ = read_reply(&mut s).await;
+            submission_auth_login(&mut s, login, password).await
+        }
+        Tls::Starttls => {
+            let mut tcp = TcpStream::connect((host, port)).await.unwrap();
+            let _ = read_reply(&mut tcp).await;
+            tcp.write_all(b"EHLO proxy.test\r\n").await.unwrap();
+            let _ = read_until(&mut tcp, "250 ").await;
+            tcp.write_all(b"STARTTLS\r\n").await.unwrap();
+            let _ = read_reply(&mut tcp).await;
+            let mut s = tls_connect_client(tcp).await;
+            submission_auth_login(&mut s, login, password).await
+        }
+    }
+}
+
 pub async fn smtp_passthrough_send(host: &str, port: u16, from: &str, to: &str) -> String {
     let mut s = TcpStream::connect((host, port)).await.unwrap();
     let _ = read_reply(&mut s).await;

src/tests/containers/scenario1.rs

@@ -240,3 +240,48 @@ tls = "plain"
         "passthrough RCPT to a hosted Stalwart user should be accepted: {reply:?}"
     );
 }
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn scenario1_submission_auth_login_reaches_both_backends() {
+    let legacy = Legacy::start().await;
+    let stalwart =
+        Stalwart::start_proxied(Edition::New, "new.test", &[("carol", "CarolPass#2026")]).await;
+
+    let legacy_user = ("alice@legacy.test", "AlicePass#2026");
+    let stalwart_user = ("carol@new.test", "CarolPass#2026");
+    let map = mapping_file(&format!(
+        "{}\tlegacy\n{}\tstalwartnew\n",
+        legacy_user.0, stalwart_user.0
+    ));
+
+    let mode = Tls::Implicit;
+    let listen = pick_free_port();
+    let toml = build_config(
+        "submission",
+        mode,
+        listen,
+        legacy_endpoint(&legacy, "submission", mode),
+        stalwart_endpoint(&stalwart, "submission", mode),
+        map.path().to_str().unwrap(),
+    );
+    let proxy = Proxy::launch(&toml).await;
+    let port = proxy.port("l");
+
+    for (who, (login, password)) in [("legacy", legacy_user), ("stalwart", stalwart_user)] {
+        let reply =
+            clients::submission_send_auth_login("127.0.0.1", port, mode, login, password).await;
+        assert!(
+            reply.contains("235"),
+            "AUTH LOGIN through the proxy to the {who} backend failed: {reply:?}"
+        );
+    }
+
+    let wrong =
+        clients::submission_send_auth_login("127.0.0.1", port, mode, legacy_user.0, "WrongPass#0")
+            .await;
+    assert!(
+        wrong.contains("535") || wrong.contains("454"),
+        "a bad password over AUTH LOGIN must still be rejected by the backend: {wrong:?}"
+    );
+}

src/tests/imap.rs

@@ -446,3 +446,209 @@ tls = "starttls"
     );
     handler.abort();
 }
+
+#[tokio::test]
+async fn imap_authenticate_login_replays_as_plain() {
+    let (port, backend) = dummy_backend(|sock| backend_expect_auth(sock, "user@example.com")).await;
+    let mapping = mapping_file("");
+    let toml = config(port, &mapping.path().display().to_string());
+    let ctx = make_ctx(&toml).await;
+    let listener = ctx.config.listener["imap"].clone();
+
+    let (mut client, server) = tcp_pair().await;
+    let ctx2 = ctx.clone();
+    let handler = tokio::spawn(async move {
+        let _ = crate::proto::imap::handle(
+            &ctx2,
+            &listener,
+            Box::new(server),
+            "127.0.0.1:40060".parse().unwrap(),
+            "127.0.0.1:143".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let greeting = read_line(&mut client).await;
+    assert!(
+        greeting.contains("AUTH=LOGIN"),
+        "capability must advertise AUTH=LOGIN, got {greeting}"
+    );
+
+    send(&mut client, b"a1 AUTHENTICATE LOGIN\r\n").await;
+    let user_prompt = read_line(&mut client).await;
+    assert_eq!(user_prompt.trim_end(), "+ VXNlcm5hbWU6", "{user_prompt}");
+    send(
+        &mut client,
+        format!("{}\r\n", BASE64.encode("user@example.com")).as_bytes(),
+    )
+    .await;
+
+    let pass_prompt = read_line(&mut client).await;
+    assert_eq!(pass_prompt.trim_end(), "+ UGFzc3dvcmQ6", "{pass_prompt}");
+    send(
+        &mut client,
+        format!("{}\r\n", BASE64.encode("secret")).as_bytes(),
+    )
+    .await;
+
+    send(&mut client, b"a2 NOOP\r\n").await;
+    let resp = read_to_eof(&mut client).await;
+    let resp = String::from_utf8_lossy(&resp).into_owned();
+    assert!(resp.contains("a1 OK"), "expected a1 OK, got {resp}");
+    assert!(resp.contains("a2 OK"), "expected bridged a2 OK, got {resp}");
+
+    let _ = backend.await;
+    handler.abort();
+}
+
+#[tokio::test]
+async fn imap_authenticate_login_accepts_username_as_initial_response() {
+    let (port, backend) =
+        dummy_backend(|sock| backend_expect_auth(sock, "alice@example.com")).await;
+    let mapping = mapping_file("");
+    let toml = config(port, &mapping.path().display().to_string());
+    let ctx = make_ctx(&toml).await;
+    let listener = ctx.config.listener["imap"].clone();
+
+    let (mut client, server) = tcp_pair().await;
+    let ctx2 = ctx.clone();
+    let handler = tokio::spawn(async move {
+        let _ = crate::proto::imap::handle(
+            &ctx2,
+            &listener,
+            Box::new(server),
+            "127.0.0.1:40061".parse().unwrap(),
+            "127.0.0.1:143".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let _greeting = read_line(&mut client).await;
+    send(
+        &mut client,
+        format!(
+            "a1 AUTHENTICATE LOGIN {}\r\n",
+            BASE64.encode("alice@example.com")
+        )
+        .as_bytes(),
+    )
+    .await;
+    let pass_prompt = read_line(&mut client).await;
+    assert_eq!(pass_prompt.trim_end(), "+ UGFzc3dvcmQ6", "{pass_prompt}");
+    send(
+        &mut client,
+        format!("{}\r\n", BASE64.encode("secret")).as_bytes(),
+    )
+    .await;
+    send(&mut client, b"a2 NOOP\r\n").await;
+
+    let resp = read_until(&mut client, b"a1 OK").await;
+    assert!(find(&resp, b"a1 OK").is_some());
+    let _ = backend.await;
+    handler.abort();
+}
+
+#[tokio::test]
+async fn imap_authenticate_login_cancelled_by_client() {
+    let mapping = mapping_file("");
+    let toml = config(1, &mapping.path().display().to_string());
+    let ctx = make_ctx(&toml).await;
+    let listener = ctx.config.listener["imap"].clone();
+
+    let (mut client, server) = tcp_pair().await;
+    let ctx2 = ctx.clone();
+    let handler = tokio::spawn(async move {
+        let _ = crate::proto::imap::handle(
+            &ctx2,
+            &listener,
+            Box::new(server),
+            "127.0.0.1:40062".parse().unwrap(),
+            "127.0.0.1:143".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let _greeting = read_line(&mut client).await;
+    send(&mut client, b"a1 AUTHENTICATE LOGIN\r\n").await;
+    let _user_prompt = read_line(&mut client).await;
+    send(&mut client, b"*\r\n").await;
+
+    let resp = read_until(&mut client, b"a1 NO").await;
+    assert!(
+        find(&resp, b"a1 NO").is_some(),
+        "expected tagged NO, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+
+    send(&mut client, b"a2 CAPABILITY\r\n").await;
+    let resp = read_until(&mut client, b"a2 OK").await;
+    assert!(
+        find(&resp, b"a2 OK").is_some(),
+        "session must survive a cancelled AUTHENTICATE, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    handler.abort();
+}
+
+#[tokio::test]
+async fn imap_authenticate_login_refused_without_tls_when_plain_disabled() {
+    let mapping = mapping_file("");
+    let toml = format!(
+        r#"
+[mapping]
+source = "file"
+normalize = "lowercase"
+[mapping.file]
+path = "{}"
+
+[capabilities.imap]
+allow_plain_auth_without_tls = false
+
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+tls_server_name = "backend.example.com"
+[destination.legacy.protocol.imap]
+port = 1
+tls = "implicit"
+
+[listener.imap]
+protocol = "imap"
+bind = ["127.0.0.1:0"]
+tls = "starttls"
+"#,
+        mapping.path().display()
+    );
+    let ctx = make_ctx(&toml).await;
+    let listener = ctx.config.listener["imap"].clone();
+
+    let (mut client, server) = tcp_pair().await;
+    let ctx2 = ctx.clone();
+    let handler = tokio::spawn(async move {
+        let _ = crate::proto::imap::handle(
+            &ctx2,
+            &listener,
+            Box::new(server),
+            "127.0.0.1:40063".parse().unwrap(),
+            "127.0.0.1:143".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let greeting = read_line(&mut client).await;
+    assert!(
+        !greeting.contains("AUTH=LOGIN"),
+        "AUTH=LOGIN must not be advertised without TLS, got {greeting}"
+    );
+
+    send(&mut client, b"a1 AUTHENTICATE LOGIN\r\n").await;
+    let resp = read_until(&mut client, b"a1 NO").await;
+    assert!(
+        find(&resp, b"PRIVACYREQUIRED").is_some(),
+        "expected PRIVACYREQUIRED, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    handler.abort();
+}

src/tests/managesieve.rs

@@ -306,3 +306,133 @@ tls = "plain"
     );
     handler.abort();
 }
+
+#[tokio::test]
+async fn managesieve_authenticate_login_replays_as_plain() {
+    let (port, backend) = dummy_backend(|sock| backend_expect_auth(sock, "user@example.com")).await;
+    let mapping = mapping_file("");
+    let toml = config(port, &mapping.path().display().to_string());
+    let ctx = make_ctx(&toml).await;
+    let listener = ctx.config.listener["managesieve"].clone();
+
+    let (mut client, server) = tcp_pair().await;
+    let ctx2 = ctx.clone();
+    let handler = tokio::spawn(async move {
+        let _ = crate::proto::managesieve::handle(
+            &ctx2,
+            &listener,
+            Box::new(server),
+            "127.0.0.1:40014".parse().unwrap(),
+            "127.0.0.1:4190".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let greeting = read_until(&mut client, b"OK ").await;
+    let greeting = String::from_utf8_lossy(&greeting).into_owned();
+    assert!(
+        greeting.contains("LOGIN"),
+        "greeting must advertise SASL LOGIN, got {greeting}"
+    );
+
+    send(&mut client, b"AUTHENTICATE \"LOGIN\"\r\n").await;
+    let user_prompt = read_until(&mut client, b"VXNlcm5hbWU6\r\n").await;
+    assert!(
+        find(&user_prompt, b"{12}\r\nVXNlcm5hbWU6").is_some(),
+        "expected a username challenge literal, got {:?}",
+        String::from_utf8_lossy(&user_prompt)
+    );
+
+    send(
+        &mut client,
+        format!("\"{}\"\r\n", BASE64.encode("user@example.com")).as_bytes(),
+    )
+    .await;
+    let pass_prompt = read_until(&mut client, b"UGFzc3dvcmQ6\r\n").await;
+    assert!(
+        find(&pass_prompt, b"{12}\r\nUGFzc3dvcmQ6").is_some(),
+        "expected a password challenge literal, got {:?}",
+        String::from_utf8_lossy(&pass_prompt)
+    );
+
+    send(
+        &mut client,
+        format!("\"{}\"\r\n", BASE64.encode("secret")).as_bytes(),
+    )
+    .await;
+
+    let resp = read_to_eof(&mut client).await;
+    assert!(
+        find(&resp, b"OK \"authenticated\"").is_some(),
+        "expected OK authenticated, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    let _ = backend.await;
+    handler.abort();
+}
+
+#[tokio::test]
+async fn managesieve_authenticate_login_rejected_when_not_configured() {
+    let mapping = mapping_file("");
+    let toml = format!(
+        r#"
+[mapping]
+source = "file"
+normalize = "lowercase"
+[mapping.file]
+path = "{}"
+
+[capabilities.managesieve]
+allow_plain_auth_without_tls = true
+sasl = ["PLAIN", "OAUTHBEARER"]
+
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+proxy_protocol = false
+allow_plaintext_auth = true
+[destination.legacy.protocol.managesieve]
+port = 1
+tls = "plain"
+
+[listener.managesieve]
+protocol = "managesieve"
+bind = ["127.0.0.1:0"]
+tls = "plain"
+"#,
+        mapping.path().display()
+    );
+    let ctx = make_ctx(&toml).await;
+    let listener = ctx.config.listener["managesieve"].clone();
+
+    let (mut client, server) = tcp_pair().await;
+    let ctx2 = ctx.clone();
+    let handler = tokio::spawn(async move {
+        let _ = crate::proto::managesieve::handle(
+            &ctx2,
+            &listener,
+            Box::new(server),
+            "127.0.0.1:40015".parse().unwrap(),
+            "127.0.0.1:4190".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let greeting = read_until(&mut client, b"OK ").await;
+    let greeting = String::from_utf8_lossy(&greeting).into_owned();
+    assert!(
+        !greeting.to_uppercase().contains("LOGIN"),
+        "LOGIN must not be advertised when unconfigured, got {greeting}"
+    );
+
+    send(&mut client, b"AUTHENTICATE \"LOGIN\"\r\n").await;
+    let resp = read_until(&mut client, b"\r\n").await;
+    assert!(
+        find(&resp, b"NO \"Unsupported mechanism\"").is_some(),
+        "expected an unsupported-mechanism NO, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    handler.abort();
+}

src/tests/pop3.rs

@@ -396,3 +396,216 @@ tls = "plain"
     );
     handler.abort();
 }
+
+#[tokio::test]
+async fn pop3_auth_login_replays_as_plain() {
+    let (port, backend) = dummy_backend(|sock| backend_expect_auth(sock, "user@example.com")).await;
+    let mapping = mapping_file("");
+    let toml = config(port, &mapping.path().display().to_string());
+    let ctx = make_ctx(&toml).await;
+    let mut client = spawn_pop3(&ctx).await;
+
+    let _greeting = read_line(&mut client).await;
+
+    send(&mut client, b"CAPA\r\n").await;
+    let capa = read_until(&mut client, b".\r\n").await;
+    let capa = String::from_utf8_lossy(&capa).into_owned();
+    assert!(
+        capa.contains("LOGIN"),
+        "CAPA must advertise SASL LOGIN, got {capa}"
+    );
+
+    send(&mut client, b"AUTH LOGIN\r\n").await;
+    let user_prompt = read_until(&mut client, b"\r\n").await;
+    assert_eq!(
+        String::from_utf8_lossy(&user_prompt).trim_end(),
+        "+ VXNlcm5hbWU6"
+    );
+    send(
+        &mut client,
+        format!("{}\r\n", BASE64.encode("user@example.com")).as_bytes(),
+    )
+    .await;
+
+    let pass_prompt = read_until(&mut client, b"\r\n").await;
+    assert_eq!(
+        String::from_utf8_lossy(&pass_prompt).trim_end(),
+        "+ UGFzc3dvcmQ6"
+    );
+    send(
+        &mut client,
+        format!("{}\r\nSTAT\r\n", BASE64.encode("secret")).as_bytes(),
+    )
+    .await;
+
+    let resp = read_to_eof(&mut client).await;
+    assert!(
+        find(&resp, b"+OK logged in").is_some(),
+        "expected +OK logged in, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    assert!(
+        find(&resp, b"+OK 0 0").is_some(),
+        "expected bridged STAT reply, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    let _ = backend.await;
+}
+
+#[tokio::test]
+async fn pop3_auth_login_accepts_username_as_initial_response() {
+    let (port, backend) =
+        dummy_backend(|sock| backend_expect_auth(sock, "alice@example.com")).await;
+    let mapping = mapping_file("");
+    let toml = config(port, &mapping.path().display().to_string());
+    let ctx = make_ctx(&toml).await;
+    let mut client = spawn_pop3(&ctx).await;
+
+    let _greeting = read_line(&mut client).await;
+    send(
+        &mut client,
+        format!("AUTH LOGIN {}\r\n", BASE64.encode("alice@example.com")).as_bytes(),
+    )
+    .await;
+    let pass_prompt = read_until(&mut client, b"\r\n").await;
+    assert_eq!(
+        String::from_utf8_lossy(&pass_prompt).trim_end(),
+        "+ UGFzc3dvcmQ6"
+    );
+    send(
+        &mut client,
+        format!("{}\r\nSTAT\r\n", BASE64.encode("secret")).as_bytes(),
+    )
+    .await;
+
+    let resp = read_to_eof(&mut client).await;
+    assert!(
+        find(&resp, b"+OK logged in").is_some(),
+        "expected +OK logged in, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    let _ = backend.await;
+}
+
+#[tokio::test]
+async fn pop3_auth_login_cancelled_by_client() {
+    let mapping = mapping_file("");
+    let toml = config(1, &mapping.path().display().to_string());
+    let ctx = make_ctx(&toml).await;
+    let mut client = spawn_pop3(&ctx).await;
+
+    let _greeting = read_line(&mut client).await;
+    send(&mut client, b"AUTH LOGIN\r\n").await;
+    let _user_prompt = read_until(&mut client, b"\r\n").await;
+    send(&mut client, b"*\r\nQUIT\r\n").await;
+
+    let resp = read_to_eof(&mut client).await;
+    assert!(
+        find(&resp, b"-ERR authentication cancelled").is_some(),
+        "expected a cancel error, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    assert!(
+        find(&resp, b"+OK Bye").is_some(),
+        "session must survive a cancelled AUTH, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+}
+
+#[tokio::test]
+async fn pop3_auth_login_refused_without_tls_when_plain_disabled() {
+    let mapping = mapping_file("");
+    let toml = format!(
+        r#"
+[mapping]
+source = "file"
+normalize = "lowercase"
+[mapping.file]
+path = "{}"
+
+[capabilities.pop3]
+allow_plain_auth_without_tls = false
+
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+tls_server_name = "backend.example.com"
+[destination.legacy.protocol.pop3]
+port = 1
+tls = "implicit"
+
+[listener.pop3]
+protocol = "pop3"
+bind = ["127.0.0.1:0"]
+tls = "plain"
+"#,
+        mapping.path().display()
+    );
+    let ctx = make_ctx(&toml).await;
+    let mut client = spawn_pop3(&ctx).await;
+
+    let _greeting = read_line(&mut client).await;
+    send(&mut client, b"CAPA\r\n").await;
+    let capa = read_until(&mut client, b".\r\n").await;
+    let capa = String::from_utf8_lossy(&capa).into_owned();
+    assert!(
+        !capa.to_uppercase().contains("LOGIN"),
+        "CAPA must omit LOGIN without TLS, got {capa}"
+    );
+
+    send(&mut client, b"AUTH LOGIN\r\n").await;
+    let resp = read_until(&mut client, b"-ERR").await;
+    assert!(
+        find(&resp, b"cleartext authentication requires TLS").is_some(),
+        "expected a cleartext refusal, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+}
+
+#[tokio::test]
+async fn pop3_auth_login_rejected_when_not_configured() {
+    let mapping = mapping_file("");
+    let toml = format!(
+        r#"
+[mapping]
+source = "file"
+normalize = "lowercase"
+[mapping.file]
+path = "{}"
+
+[capabilities.pop3]
+allow_plain_auth_without_tls = true
+sasl = ["PLAIN", "OAUTHBEARER"]
+
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+proxy_protocol = false
+allow_plaintext_auth = true
+[destination.legacy.protocol.pop3]
+port = 1
+tls = "plain"
+
+[listener.pop3]
+protocol = "pop3"
+bind = ["127.0.0.1:0"]
+tls = "plain"
+"#,
+        mapping.path().display()
+    );
+    let ctx = make_ctx(&toml).await;
+    let mut client = spawn_pop3(&ctx).await;
+
+    let _greeting = read_line(&mut client).await;
+    send(&mut client, b"AUTH LOGIN\r\n").await;
+    let resp = read_until(&mut client, b"-ERR").await;
+    assert!(
+        find(&resp, b"unsupported mechanism").is_some(),
+        "expected an unsupported-mechanism error, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+}

src/tests/smtp.rs

@@ -315,3 +315,224 @@ async fn smtp_passthrough_bridges_raw() {
     let _ = backend.await;
     handler.abort();
 }
+
+fn submission_config_with_sasl(port: u16, mapping_path: &str, plain: bool, sasl: &str) -> String {
+    format!(
+        r#"
+[mapping]
+source = "file"
+normalize = "lowercase"
+[mapping.file]
+path = "{mapping_path}"
+
+[capabilities.submission]
+allow_plain_auth_without_tls = {plain}
+sasl = {sasl}
+
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+proxy_protocol = false
+allow_plaintext_auth = true
+[destination.legacy.protocol.submission]
+port = {port}
+tls = "plain"
+
+[listener.submission]
+protocol = "submission"
+bind = ["127.0.0.1:0"]
+tls = "plain"
+"#
+    )
+}
+
+async fn spawn_submission(
+    ctx: &std::sync::Arc<crate::proto::common::Ctx>,
+    peer: &str,
+) -> (tokio::net::TcpStream, tokio::task::JoinHandle<()>) {
+    let listener = ctx.config.listener["submission"].clone();
+    let (client, server) = tcp_pair().await;
+    let ctx2 = ctx.clone();
+    let peer = peer.to_string();
+    let handler = tokio::spawn(async move {
+        let _ = crate::proto::smtp::handle_submission(
+            &ctx2,
+            &listener,
+            Box::new(server),
+            peer.parse().unwrap(),
+            "127.0.0.1:587".parse().unwrap(),
+        )
+        .await;
+    });
+    (client, handler)
+}
+
+#[tokio::test]
+async fn submission_advertises_and_completes_auth_login() {
+    let (port, backend) = dummy_backend(|sock| submission_backend(sock, "user@example.com")).await;
+    let mapping = mapping_file("");
+    let toml = submission_config(port, &mapping.path().display().to_string(), true);
+    let ctx = make_ctx(&toml).await;
+
+    let (mut client, handler) = spawn_submission(&ctx, "127.0.0.1:40030").await;
+
+    let greeting = read_line(&mut client).await;
+    assert!(greeting.starts_with("220"), "greeting: {greeting}");
+
+    send(&mut client, b"EHLO test.local\r\n").await;
+    let ehlo = read_until(&mut client, b"250 ").await;
+    let ehlo = String::from_utf8_lossy(&ehlo).into_owned();
+    assert!(
+        ehlo.contains("AUTH ") && ehlo.to_uppercase().contains("LOGIN"),
+        "EHLO must advertise AUTH LOGIN, got {ehlo}"
+    );
+
+    send(&mut client, b"AUTH LOGIN\r\n").await;
+    let user_prompt = read_line(&mut client).await;
+    assert_eq!(user_prompt.trim_end(), "334 VXNlcm5hbWU6", "{user_prompt}");
+
+    send(
+        &mut client,
+        format!("{}\r\n", BASE64.encode("user@example.com")).as_bytes(),
+    )
+    .await;
+    let pass_prompt = read_line(&mut client).await;
+    assert_eq!(pass_prompt.trim_end(), "334 UGFzc3dvcmQ6", "{pass_prompt}");
+
+    send(
+        &mut client,
+        format!("{}\r\n", BASE64.encode("secret")).as_bytes(),
+    )
+    .await;
+
+    let resp = read_to_eof(&mut client).await;
+    assert!(
+        find(&resp, b"235").is_some(),
+        "expected 235, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+
+    let _ = backend.await;
+    handler.abort();
+}
+
+#[tokio::test]
+async fn submission_auth_login_accepts_username_as_initial_response() {
+    let (port, backend) = dummy_backend(|sock| submission_backend(sock, "user@example.com")).await;
+    let mapping = mapping_file("");
+    let toml = submission_config(port, &mapping.path().display().to_string(), true);
+    let ctx = make_ctx(&toml).await;
+
+    let (mut client, handler) = spawn_submission(&ctx, "127.0.0.1:40031").await;
+    let _greeting = read_line(&mut client).await;
+    send(&mut client, b"EHLO test.local\r\n").await;
+    let _ehlo = read_until(&mut client, b"250 ").await;
+
+    send(
+        &mut client,
+        format!("AUTH LOGIN {}\r\n", BASE64.encode("user@example.com")).as_bytes(),
+    )
+    .await;
+    let pass_prompt = read_line(&mut client).await;
+    assert_eq!(pass_prompt.trim_end(), "334 UGFzc3dvcmQ6", "{pass_prompt}");
+
+    send(
+        &mut client,
+        format!("{}\r\n", BASE64.encode("secret")).as_bytes(),
+    )
+    .await;
+
+    let resp = read_to_eof(&mut client).await;
+    assert!(
+        find(&resp, b"235").is_some(),
+        "expected 235, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+
+    let _ = backend.await;
+    handler.abort();
+}
+
+#[tokio::test]
+async fn submission_auth_login_cancelled_by_client() {
+    let mapping = mapping_file("");
+    let toml = submission_config(1, &mapping.path().display().to_string(), true);
+    let ctx = make_ctx(&toml).await;
+
+    let (mut client, handler) = spawn_submission(&ctx, "127.0.0.1:40032").await;
+    let _greeting = read_line(&mut client).await;
+    send(&mut client, b"EHLO test.local\r\n").await;
+    let _ehlo = read_until(&mut client, b"250 ").await;
+
+    send(&mut client, b"AUTH LOGIN\r\n").await;
+    let _user_prompt = read_line(&mut client).await;
+    send(&mut client, b"*\r\n").await;
+
+    let resp = read_until(&mut client, b"\r\n").await;
+    let resp = String::from_utf8_lossy(&resp).into_owned();
+    assert!(resp.starts_with("501"), "expected 501 cancel, got {resp}");
+
+    send(&mut client, b"NOOP\r\n").await;
+    let noop = read_line(&mut client).await;
+    assert!(
+        noop.starts_with("250"),
+        "session must survive a cancelled AUTH, got {noop}"
+    );
+    handler.abort();
+}
+
+#[tokio::test]
+async fn submission_auth_login_refused_when_cleartext_is_not_allowed() {
+    let mapping = mapping_file("");
+    let toml = submission_config(1, &mapping.path().display().to_string(), false);
+    let ctx = make_ctx(&toml).await;
+
+    let (mut client, handler) = spawn_submission(&ctx, "127.0.0.1:40033").await;
+    let _greeting = read_line(&mut client).await;
+    send(&mut client, b"EHLO test.local\r\n").await;
+    let ehlo = read_until(&mut client, b"250 ").await;
+    let ehlo = String::from_utf8_lossy(&ehlo).into_owned();
+    assert!(
+        !ehlo.to_uppercase().contains("LOGIN"),
+        "LOGIN must not be advertised without TLS, got {ehlo}"
+    );
+
+    send(&mut client, b"AUTH LOGIN\r\n").await;
+    let resp = read_until(&mut client, b"\r\n").await;
+    let resp = String::from_utf8_lossy(&resp).into_owned();
+    assert!(resp.starts_with("538"), "expected 538, got {resp}");
+    handler.abort();
+}
+
+#[tokio::test]
+async fn submission_auth_login_rejected_when_not_configured() {
+    let mapping = mapping_file("");
+    let toml = submission_config_with_sasl(
+        1,
+        &mapping.path().display().to_string(),
+        true,
+        r#"["PLAIN", "OAUTHBEARER", "XOAUTH2"]"#,
+    );
+    let ctx = make_ctx(&toml).await;
+
+    let (mut client, handler) = spawn_submission(&ctx, "127.0.0.1:40034").await;
+    let _greeting = read_line(&mut client).await;
+    send(&mut client, b"EHLO test.local\r\n").await;
+    let ehlo = read_until(&mut client, b"250 ").await;
+    let ehlo = String::from_utf8_lossy(&ehlo).into_owned();
+    assert!(
+        !ehlo.to_uppercase().contains("LOGIN"),
+        "LOGIN must not be advertised when unconfigured, got {ehlo}"
+    );
+
+    send(&mut client, b"AUTH LOGIN\r\n").await;
+    let resp = read_until(&mut client, b"\r\n").await;
+    let resp = String::from_utf8_lossy(&resp).into_owned();
+    assert!(
+        resp.starts_with("535") && resp.contains("Unsupported"),
+        "expected 535 unsupported, got {resp}"
+    );
+    handler.abort();
+}