IMAP authentication is no longer replayed with an inline initial response to backends that do not advertise SASL-IR (fixes #4)

CHANGELOG.md

@@ -2,6 +2,25 @@
 
 All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org/).
 
+## [1.0.3] - 2026-08-30
+
+### Added
+
+### Changed
+
+### Fixed
+- IMAP authentication is no longer replayed with an inline initial response to backends that do not advertise `SASL-IR` (#4).
+
+## [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.toml

@@ -1,6 +1,6 @@
 [package]
 name = "proxy"
-version = "1.0.1"
+version = "1.0.3"
 edition = "2024"
 license = "AGPL-3.0-only OR LicenseRef-SEL"
 description = "Multi-protocol e-mail proxy"
@@ -49,8 +49,7 @@ rustls-platform-verifier = "0.7"
 [dev-dependencies]
 tempfile = "3"
 mockito = "1.7"
-testcontainers = "0.27"
-testcontainers-modules = { version = "0.15", features = ["redis", "postgres"] }
+testcontainers = "0.28"
 pwhash = "1"
 jmap-client = { version = "0.4", default-features = false, features = ["websockets", "async", "aws_lc_rs"] }
 futures-util = "0.3"

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')
@@ -598,6 +623,55 @@ pub async fn hold_login(ctx: &Ctx, identifier: Option<&str>) {
     }
 }
 
+pub struct BackendLeg {
+    pub stream: BoxedStream,
+    pub residual: Vec<u8>,
+    pub accepts_initial_response: bool,
+}
+
+impl BackendLeg {
+    pub fn new(stream: BoxedStream, residual: Vec<u8>) -> Self {
+        Self {
+            stream,
+            residual,
+            accepts_initial_response: true,
+        }
+    }
+
+    pub fn with_initial_response(mut self, accepted: bool) -> Self {
+        self.accepts_initial_response = accepted;
+        self
+    }
+}
+
+pub struct AuthFrame {
+    inline: Vec<u8>,
+    deferred: Option<(Vec<u8>, Vec<u8>)>,
+}
+
+impl AuthFrame {
+    pub fn inline(frame: Vec<u8>) -> Self {
+        Self {
+            inline: frame,
+            deferred: None,
+        }
+    }
+
+    pub fn with_deferred(mut self, command: Vec<u8>, response: Vec<u8>) -> Self {
+        self.deferred = Some((command, response));
+        self
+    }
+
+    fn wire(&self, accepts_initial_response: bool) -> (&[u8], Option<&[u8]>) {
+        match &self.deferred {
+            Some((command, response)) if !accepts_initial_response => {
+                (command.as_slice(), Some(response.as_slice()))
+            }
+            _ => (self.inline.as_slice(), None),
+        }
+    }
+}
+
 pub enum AuthWire {
     Imap { tag: String },
     Pop3,
@@ -690,6 +764,7 @@ pub async fn complete_backend_auth(
     backend: &mut BoxedStream,
     residual: &mut Vec<u8>,
     wire: AuthWire,
+    mut pending_response: Option<&[u8]>,
     hide: bool,
     max: usize,
 ) -> Result<bool> {
@@ -698,7 +773,8 @@ pub async fn complete_backend_auth(
         let line = crate::outbound::read_line(backend, residual, max).await?;
         match classify_auth_line(&line, &wire) {
             AuthLine::Continuation => {
-                crate::outbound::write_all(backend, SASL_EMPTY_RESPONSE).await?;
+                let response = pending_response.take().unwrap_or(SASL_EMPTY_RESPONSE);
+                crate::outbound::write_all(backend, response).await?;
             }
             AuthLine::Passthrough => {
                 client_out.extend_from_slice(&line);
@@ -738,14 +814,14 @@ pub async fn bridge_authenticated<F, Fut>(
     started: Instant,
     label: &'static str,
     wire: AuthWire,
-    auth_frame: &[u8],
+    auth: &AuthFrame,
     unavailable: &[u8],
     max: usize,
     open: F,
 ) -> Result<()>
 where
     F: FnMut() -> Fut,
-    Fut: Future<Output = Result<(BoxedStream, Vec<u8>)>>,
+    Fut: Future<Output = Result<BackendLeg>>,
 {
     if let Some(id) = identifier
         && !crate::token::valid_routing_identifier(id)
@@ -759,7 +835,11 @@ where
         return Ok(());
     }
 
-    let (mut backend, mut backend_residual) = match establish(ctx, dest_id, peer, open).await {
+    let BackendLeg {
+        stream: mut backend,
+        residual: mut backend_residual,
+        accepts_initial_response,
+    } = match establish(ctx, dest_id, peer, open).await {
         Ok(b) => b,
         Err(e) => {
             tracing::warn!(error = %e, protocol = label, destination = %dest_id, "backend unavailable");
@@ -768,6 +848,7 @@ where
         }
     };
 
+    let (auth_frame, pending_response) = auth.wire(accepts_initial_response);
     crate::outbound::write_all(&mut backend, auth_frame).await?;
 
     let hide = ctx.hide_auth_errors(dest_id);
@@ -778,6 +859,7 @@ where
             &mut backend,
             &mut backend_residual,
             wire,
+            pending_response,
             hide,
             max,
         ),
@@ -885,6 +967,37 @@ pub async fn read_crlf_line(
 mod tests {
     use super::*;
 
+    #[test]
+    fn auth_frame_defers_initial_response_when_not_accepted() {
+        let frame = AuthFrame::inline(b"a1 AUTHENTICATE PLAIN AGFAYgBj\r\n".to_vec())
+            .with_deferred(
+                b"a1 AUTHENTICATE PLAIN\r\n".to_vec(),
+                b"AGFAYgBj\r\n".to_vec(),
+            );
+
+        assert_eq!(
+            frame.wire(true),
+            (b"a1 AUTHENTICATE PLAIN AGFAYgBj\r\n".as_slice(), None)
+        );
+        assert_eq!(
+            frame.wire(false),
+            (
+                b"a1 AUTHENTICATE PLAIN\r\n".as_slice(),
+                Some(b"AGFAYgBj\r\n".as_slice())
+            )
+        );
+    }
+
+    #[test]
+    fn auth_frame_without_deferred_form_stays_inline() {
+        let frame = AuthFrame::inline(b"AUTH PLAIN AGFAYgBj\r\n".to_vec());
+
+        assert_eq!(
+            frame.wire(false),
+            (b"AUTH PLAIN AGFAYgBj\r\n".as_slice(), None)
+        );
+    }
+
     #[test]
     fn advertises_token_handles_smtp_ehlo_continuation_lines() {
         let ehlo = b"250-mail.example\r\n250-PIPELINING\r\n250-STARTTLS\r\n250-AUTH PLAIN LOGIN\r\n250 CHUNKING\r\n";
@@ -904,6 +1017,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

@@ -78,10 +78,14 @@ pub async fn handle(
     let client_tls = stream.is_tls();
     let fwd = forward::ForwardInfo::new(peer, local, client_tls, None, ctx.config.server.proxy_ttl);
 
-    let auth_frame = format!(
-        "{tag} AUTHENTICATE {} {}\r\n",
-        mechanism_name(&mech),
-        BASE64.encode(&raw_challenge)
+    let mech_name = mechanism_name(&mech);
+    let response = BASE64.encode(&raw_challenge);
+    let auth = common::AuthFrame::inline(
+        format!("{tag} AUTHENTICATE {mech_name} {response}\r\n").into_bytes(),
+    )
+    .with_deferred(
+        format!("{tag} AUTHENTICATE {mech_name}\r\n").into_bytes(),
+        format!("{response}\r\n").into_bytes(),
     );
     let unavailable = format!("{tag} NO [UNAVAILABLE] backend unavailable\r\n");
 
@@ -95,7 +99,7 @@ pub async fn handle(
         started,
         "imap",
         common::AuthWire::Imap { tag },
-        auth_frame.as_bytes(),
+        &auth,
         unavailable.as_bytes(),
         MAX_PREAUTH,
         || open_backend(ctx, &dest_id, &fwd),
@@ -325,19 +329,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 +349,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 +370,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,
@@ -435,7 +481,7 @@ async fn open_backend(
     ctx: &Ctx,
     dest_id: &str,
     fwd: &forward::ForwardInfo,
-) -> Result<(BoxedStream, Vec<u8>)> {
+) -> Result<common::BackendLeg> {
     let common::DialedEndpoint {
         dest,
         ep,
@@ -448,6 +494,7 @@ async fn open_backend(
     let mut residual = Vec::new();
 
     let banner = outbound::read_line(&mut backend, &mut residual, MAX_PREAUTH).await?;
+    let mut sasl_ir = common::advertises_token(&banner, "SASL-IR");
 
     if ep.tls == TlsMode::Starttls {
         if banner.contains(&b'[') && !common::advertises_token(&banner, "STARTTLS") {
@@ -458,6 +505,7 @@ async fn open_backend(
         if !tagged_ok(&resp, b"P1") {
             return Err(ProxyError::backend("backend refused STARTTLS"));
         }
+        sasl_ir = common::advertises_token(&resp, "SASL-IR");
         backend = outbound::tls_connect(backend, &tls_cfg, dest).await?;
     }
 
@@ -467,7 +515,7 @@ async fn open_backend(
         forward::send_imap_id(&mut backend, &mut residual, &banner, fwd, MAX_PREAUTH).await?;
     }
 
-    Ok((backend, residual))
+    Ok(common::BackendLeg::new(backend, residual).with_initial_response(sasl_ir))
 }
 
 fn tagged_ok(line: &[u8], tag: &[u8]) -> bool {

src/proto/managesieve.rs

@@ -66,10 +66,13 @@ pub async fn handle(
         ctx.config.server.proxy_ttl,
     );
 
-    let auth_frame = format!(
-        "AUTHENTICATE \"{}\" \"{}\"\r\n",
-        mechanism_name(&mech),
-        BASE64.encode(&raw_challenge)
+    let auth = common::AuthFrame::inline(
+        format!(
+            "AUTHENTICATE \"{}\" \"{}\"\r\n",
+            mechanism_name(&mech),
+            BASE64.encode(&raw_challenge)
+        )
+        .into_bytes(),
     );
 
     common::bridge_authenticated(
@@ -82,7 +85,7 @@ pub async fn handle(
         started,
         "managesieve",
         common::AuthWire::ManageSieve,
-        auth_frame.as_bytes(),
+        &auth,
         b"NO \"backend unavailable\"\r\n",
         MAX_PREAUTH,
         || open_backend(ctx, &dest_id, &fwd),
@@ -173,6 +176,7 @@ async fn preauth(
                                 ctx,
                                 &mut stream,
                                 plain_ok,
+                                common::sasl_offered(&caps.sasl, "LOGIN"),
                                 &req.tokens,
                                 &mut it,
                                 &mut auth_failures,
@@ -239,10 +243,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 +261,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 +273,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 +301,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 +397,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
     }
 }
 
@@ -391,7 +471,7 @@ async fn open_backend(
     ctx: &Ctx,
     dest_id: &str,
     fwd: &forward::ForwardInfo,
-) -> Result<(BoxedStream, Vec<u8>)> {
+) -> Result<common::BackendLeg> {
     let common::DialedEndpoint {
         dest,
         ep,
@@ -432,7 +512,7 @@ async fn open_backend(
         .await?;
     }
 
-    Ok((backend, residual))
+    Ok(common::BackendLeg::new(backend, residual))
 }
 
 async fn read_capabilities(stream: &mut BoxedStream, residual: &mut Vec<u8>) -> Result<Vec<u8>> {

src/proto/pop3.rs

@@ -65,10 +65,13 @@ pub async fn handle(
         ctx.config.server.proxy_ttl,
     );
 
-    let auth_frame = format!(
-        "AUTH {} {}\r\n",
-        mechanism_name(&mech),
-        BASE64.encode(&raw_challenge)
+    let auth = common::AuthFrame::inline(
+        format!(
+            "AUTH {} {}\r\n",
+            mechanism_name(&mech),
+            BASE64.encode(&raw_challenge)
+        )
+        .into_bytes(),
     );
 
     common::bridge_authenticated(
@@ -81,7 +84,7 @@ pub async fn handle(
         started,
         "pop3",
         common::AuthWire::Pop3,
-        auth_frame.as_bytes(),
+        &auth,
         b"-ERR backend unavailable\r\n",
         MAX_PREAUTH,
         || open_backend(ctx, &dest_id, &fwd),
@@ -220,6 +223,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 +233,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 +243,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);
@@ -341,7 +407,7 @@ async fn open_backend(
     ctx: &Ctx,
     dest_id: &str,
     fwd: &forward::ForwardInfo,
-) -> Result<(BoxedStream, Vec<u8>)> {
+) -> Result<common::BackendLeg> {
     let common::DialedEndpoint {
         dest,
         ep,
@@ -384,7 +450,7 @@ async fn open_backend(
         .await?;
     }
 
-    Ok((backend, residual))
+    Ok(common::BackendLeg::new(backend, residual))
 }
 
 async fn backend_advertises_stls(

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};
@@ -72,10 +74,13 @@ pub async fn handle_submission(
         ctx.config.server.proxy_ttl,
     );
 
-    let auth_frame = format!(
-        "AUTH {} {}\r\n",
-        mechanism_name(&mech),
-        BASE64.encode(&raw_challenge)
+    let auth = common::AuthFrame::inline(
+        format!(
+            "AUTH {} {}\r\n",
+            mechanism_name(&mech),
+            BASE64.encode(&raw_challenge)
+        )
+        .into_bytes(),
     );
 
     common::bridge_authenticated(
@@ -88,7 +93,7 @@ pub async fn handle_submission(
         started,
         "submission",
         common::AuthWire::Smtp,
-        auth_frame.as_bytes(),
+        &auth,
         b"421 4.3.0 backend unavailable\r\n",
         MAX_PREAUTH,
         || open_backend(ctx, &dest_id, &fwd),
@@ -181,6 +186,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 +266,7 @@ enum AuthStep {
 struct AuthParams<'a> {
     ctx: &'a Ctx,
     plain_ok: bool,
+    login_offered: bool,
     ehlo_host: &'a str,
     max_attempts: u32,
 }
@@ -275,7 +282,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 +297,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 +307,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 +366,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 +483,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
     }
@@ -411,7 +526,7 @@ async fn open_backend(
     ctx: &Ctx,
     dest_id: &str,
     fwd: &forward::ForwardInfo,
-) -> Result<(BoxedStream, Vec<u8>)> {
+) -> Result<common::BackendLeg> {
     let common::DialedEndpoint {
         dest,
         ep,
@@ -456,7 +571,7 @@ async fn open_backend(
         }
     }
 
-    Ok((backend, residual))
+    Ok(common::BackendLeg::new(backend, residual))
 }
 
 fn advertised_xclient_attrs(ehlo_reply: &[u8]) -> Vec<String> {

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

@@ -57,7 +57,7 @@ async fn backend_expect_mech(
     mech: &'static str,
     want_user: &'static str,
 ) {
-    sock.write_all(b"* OK [CAPABILITY IMAP4rev2] backend ready\r\n")
+    sock.write_all(b"* OK [CAPABILITY IMAP4rev2 SASL-IR] backend ready\r\n")
         .await
         .unwrap();
 
@@ -446,3 +446,358 @@ 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();
+}
+
+const EXCHANGE_GREETING: &[u8] =
+    b"* OK [CAPABILITY IMAP4 IMAP4rev1 AUTH=PLAIN AUTH=NTLM AUTH=GSSAPI UIDPLUS MOVE ID \
+      CHILDREN IDLE NAMESPACE LITERAL+] backend ready\r\n";
+
+async fn backend_without_sasl_ir(mut sock: tokio::net::TcpStream, want_user: &'static str) {
+    sock.write_all(EXCHANGE_GREETING).await.unwrap();
+
+    let command = read_line(&mut sock).await;
+    assert_eq!(
+        command, "a1 AUTHENTICATE PLAIN",
+        "backend without SASL-IR must not receive an inline initial response"
+    );
+    sock.write_all(b"+ \r\n").await.unwrap();
+
+    let response = read_line(&mut sock).await;
+    let raw = BASE64.decode(response.trim()).unwrap();
+    assert_eq!(
+        raw,
+        format!("\0{want_user}\0secret").into_bytes(),
+        "deferred initial response should carry the credentials"
+    );
+    sock.write_all(b"a1 OK [CAPABILITY IMAP4rev1] authenticated\r\n")
+        .await
+        .unwrap();
+
+    let _ = read_until(&mut sock, b"NOOP").await;
+    sock.write_all(b"a2 OK NOOP done\r\n").await.unwrap();
+}
+
+async fn run_against_backend_without_sasl_ir(
+    ctx: std::sync::Arc<crate::proto::common::Ctx>,
+    first: &[u8],
+) -> Vec<u8> {
+    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:40010".parse().unwrap(),
+            "127.0.0.1:143".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let _greeting = read_line(&mut client).await;
+    send(&mut client, first).await;
+    let resp = read_until(&mut client, b"a1 OK").await;
+    send(&mut client, b"a2 NOOP\r\n").await;
+    let mut resp = resp;
+    resp.extend_from_slice(&read_until(&mut client, b"a2 OK").await);
+    handler.abort();
+    resp
+}
+
+#[tokio::test]
+async fn imap_defers_initial_response_when_backend_lacks_sasl_ir() {
+    let (port, backend) =
+        dummy_backend(|sock| backend_without_sasl_ir(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 ir = BASE64.encode(b"\0alice@example.com\0secret");
+    let resp = run_against_backend_without_sasl_ir(
+        ctx,
+        format!("a1 AUTHENTICATE PLAIN {ir}\r\n").as_bytes(),
+    )
+    .await;
+
+    assert!(
+        find(&resp, b"a1 OK").is_some(),
+        "expected a1 OK, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    assert!(
+        find(&resp, b"a2 OK").is_some(),
+        "expected bridged a2 OK, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    let _ = backend.await;
+}
+
+#[tokio::test]
+async fn imap_login_defers_initial_response_when_backend_lacks_sasl_ir() {
+    let (port, backend) =
+        dummy_backend(|sock| backend_without_sasl_ir(sock, "bob@example.com")).await;
+    let mapping = mapping_file("");
+    let toml = config(port, &mapping.path().display().to_string());
+    let ctx = make_ctx(&toml).await;
+
+    let resp =
+        run_against_backend_without_sasl_ir(ctx, b"a1 LOGIN bob@example.com secret\r\n").await;
+
+    assert!(
+        find(&resp, b"a1 OK").is_some(),
+        "expected a1 OK, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    let _ = backend.await;
+}
+
+#[tokio::test]
+async fn imap_keeps_inline_initial_response_when_backend_advertises_sasl_ir() {
+    let (port, backend) = dummy_backend(|mut sock: tokio::net::TcpStream| async move {
+        sock.write_all(b"* OK [CAPABILITY IMAP4rev2 SASL-IR] backend ready\r\n")
+            .await
+            .unwrap();
+        let command = read_line(&mut sock).await;
+        let ir = BASE64.encode(b"\0dave@example.com\0secret");
+        assert_eq!(command, format!("a1 AUTHENTICATE PLAIN {ir}"));
+        sock.write_all(b"a1 OK authenticated\r\n").await.unwrap();
+        let _ = read_until(&mut sock, b"NOOP").await;
+        sock.write_all(b"a2 OK NOOP done\r\n").await.unwrap();
+    })
+    .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:40011".parse().unwrap(),
+            "127.0.0.1:143".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let _greeting = read_line(&mut client).await;
+    let ir = BASE64.encode(b"\0dave@example.com\0secret");
+    send(
+        &mut client,
+        format!("a1 AUTHENTICATE PLAIN {ir}\r\n").as_bytes(),
+    )
+    .await;
+    let resp = read_until(&mut client, b"a1 OK").await;
+    assert!(find(&resp, b"a1 OK").is_some());
+    let _ = backend.await;
+    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/redis_store.rs

@@ -4,16 +4,28 @@
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
  */
 
+use testcontainers::core::{IntoContainerPort, WaitFor};
 use testcontainers::runners::AsyncRunner;
-use testcontainers_modules::redis::{REDIS_PORT, Redis};
+use testcontainers::{ContainerAsync, GenericImage};
 
 use crate::config::RedisMappingConfig;
 use crate::route::store::redis::RedisStore;
 
+const REDIS_PORT: u16 = 6379;
+
+async fn redis_container() -> ContainerAsync<GenericImage> {
+    GenericImage::new("redis", "7.4-alpine")
+        .with_exposed_port(REDIS_PORT.tcp())
+        .with_wait_for(WaitFor::message_on_stdout("Ready to accept connections"))
+        .start()
+        .await
+        .unwrap()
+}
+
 #[tokio::test]
 async fn redis_store_resolves_and_misses() {
-    let node = Redis::default().start().await.unwrap();
-    let port = node.get_host_port_ipv4(REDIS_PORT).await.unwrap();
+    let node = redis_container().await;
+    let port = node.get_host_port_ipv4(REDIS_PORT.tcp()).await.unwrap();
     let url = format!("redis://127.0.0.1:{port}");
 
     let client = redis::Client::open(url.as_str()).unwrap();

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();
+}

src/tests/sql_store.rs

@@ -4,16 +4,36 @@
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
  */
 
+use testcontainers::core::{IntoContainerPort, WaitFor};
 use testcontainers::runners::AsyncRunner;
-use testcontainers_modules::postgres::Postgres;
+use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 
 use crate::config::SqlMappingConfig;
 use crate::route::store::sql::SqlStore;
 
+const POSTGRES_PORT: u16 = 5432;
+
+async fn postgres_container() -> ContainerAsync<GenericImage> {
+    GenericImage::new("postgres", "17-alpine")
+        .with_exposed_port(POSTGRES_PORT.tcp())
+        .with_wait_for(WaitFor::message_on_stderr(
+            "database system is ready to accept connections",
+        ))
+        .with_wait_for(WaitFor::message_on_stderr(
+            "database system is ready to accept connections",
+        ))
+        .with_env_var("POSTGRES_USER", "postgres")
+        .with_env_var("POSTGRES_PASSWORD", "postgres")
+        .with_env_var("POSTGRES_DB", "postgres")
+        .start()
+        .await
+        .unwrap()
+}
+
 #[tokio::test]
 async fn sql_store_resolves_and_misses() {
-    let node = Postgres::default().start().await.unwrap();
-    let port = node.get_host_port_ipv4(5432).await.unwrap();
+    let node = postgres_container().await;
+    let port = node.get_host_port_ipv4(POSTGRES_PORT.tcp()).await.unwrap();
     let url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");
 
     sqlx::any::install_default_drivers();
@@ -49,8 +69,8 @@ async fn sql_store_resolves_and_misses() {
 
 #[tokio::test]
 async fn sql_store_upsert_and_remove() {
-    let node = Postgres::default().start().await.unwrap();
-    let port = node.get_host_port_ipv4(5432).await.unwrap();
+    let node = postgres_container().await;
+    let port = node.get_host_port_ipv4(POSTGRES_PORT.tcp()).await.unwrap();
     let url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");
 
     sqlx::any::install_default_drivers();