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,39 @@
 
 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
+- `server.host_down_min_clients`: number of distinct client addresses a
+  failure streak must span before client-attributable failures can mark a destination
+  down.
+- `GET /destinations` now also reports `failure_sources` and `host_level_failure`.
+
+### Changed
+
+### Fixed
+- The circuit breaker no longer marks a destination down because a backend rejects
+  the connections of a single banned client (#2).
+
 ## [1.0.0] - 2026-06-08
 
 ### Added

Cargo.toml

@@ -1,8 +1,7 @@
 [package]
 name = "proxy"
-version = "1.0.0"
+version = "1.0.3"
 edition = "2024"
-rust-version = "1.85"
 license = "AGPL-3.0-only OR LicenseRef-SEL"
 description = "Multi-protocol e-mail proxy"
 authors = ["Stalwart Labs LLC <hello@stalw.art>"]
@@ -17,22 +16,22 @@ name = "proxy"
 path = "src/main.rs"
 
 [dependencies]
-tokio = { version = "1.52", features = ["net", "io-util", "rt-multi-thread", "macros", "signal", "sync", "time"] }
+tokio = { version = "1.53", features = ["net", "io-util", "rt-multi-thread", "macros", "signal", "sync", "time"] }
 rustls = { version = "0.23", default-features = false, features = ["std", "aws_lc_rs", "tls12"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["aws_lc_rs", "tls12"] }
 rustls-pemfile = "2.2"
-rustls-pki-types = "1.14"
+rustls-pki-types = "1.15"
 aws-lc-rs = "1.17"
 proxy-header = { version = "0.1.2", features = ["tokio"] }
-smtp-proto = "0.2.1"
+smtp-proto = "0.2.2"
 mail-parser = "0.11"
-base64 = "0.22"
+base64 = "0.23"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 toml = "1"
 globset = "0.4"
-regex = "1.12"
-redis = { version = "1.2", features = ["tokio-comp", "aio", "connection-manager"] }
+regex = "1.13"
+redis = { version = "1.5", features = ["tokio-comp", "aio", "connection-manager"] }
 sqlx = { version = "0.9", default-features = false, features = ["runtime-tokio", "tls-rustls-aws-lc-rs", "any", "postgres", "mysql", "sqlite"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
@@ -50,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"

README.md

@@ -76,10 +76,10 @@ Additionally, you may purchase an [Enterprise License](https://stalw.art/enterpr
 
 ## License
 
-This project is dual-licensed under the **GNU Affero General Public License v3.0** (AGPL-3.0; as published by the Free Software Foundation) and the **Stalwart Enterprise License v1 (SELv1)**:
+This project is dual-licensed under the **GNU Affero General Public License v3.0** (AGPL-3.0; as published by the Free Software Foundation) and the **Stalwart Enterprise License v2 (SELv2)**:
 
 - The [GNU Affero General Public License v3.0](./LICENSES/AGPL-3.0-only.txt) is a free software license that ensures your freedom to use, modify, and distribute the software, with the condition that any modified versions of the software must also be distributed under the same license. 
-- The [Stalwart Enterprise License v1 (SELv1)](./LICENSES/LicenseRef-SEL.txt) is a proprietary license designed for commercial use. It offers additional features and greater flexibility for businesses that do not wish to comply with the AGPL-3.0 license requirements. 
+- The [Stalwart Enterprise License v2 (SELv2)](./LICENSES/LicenseRef-SEL.txt) is a proprietary license designed for commercial use. It offers additional features and greater flexibility for businesses that do not wish to comply with the AGPL-3.0 license requirements. 
 
 Each file in this project contains a license notice at the top, indicating the applicable license(s). The license notice follows the [REUSE guidelines](https://reuse.software/) to ensure clarity and consistency. The full text of each license is available in the [LICENSES](./LICENSES/) directory.
 

resources/config.dovecot-postfix.toml

@@ -0,0 +1,176 @@
+# ===========================================================================
+# Migration proxy: Dovecot + Postfix to Stalwart sample configuration.
+#
+# This configuration fronts a legacy mail stack (Dovecot serving IMAP, POP3 and
+# ManageSieve, Postfix handling SMTP submission) and a new Stalwart server at the
+# same time, migrating accounts from the legacy stack to Stalwart one at a time.
+# Clients keep talking to a single hostname on the usual ports; the proxy decides
+# per account which backend currently holds the mailbox.
+#
+# The legacy backend speaks the Dovecot and Postfix conventions: it expects the
+# real client address through the XCLIENT command (and the IMAP ID command), not
+# through a PROXY protocol header. The legacy destination therefore uses
+# forwarding = "xclient", while the Stalwart destination uses forwarding =
+# "proxy". For XCLIENT to be honoured, Dovecot and Postfix must be told to trust
+# the proxy; see the companion file resources/dovecot-postfix-trust.conf.
+#
+# Inbound mail on port 25 is handed to Stalwart, which performs split delivery:
+# mail for a migrated account is delivered locally, and mail for an account still
+# on the legacy stack is relayed back to Postfix.
+#
+# Replace every value marked CHANGE ME and review the rest before starting.
+# ===========================================================================
+
+[server]
+hostname = "mail.example.com"          # CHANGE ME (public hostname clients use)
+shutdown_grace = "30s"
+log_level = "info"
+
+# TLS certificate the proxy presents to clients (the public hostname's cert).
+[tls.certificate.default]
+cert = "/etc/proxy/tls/fullchain.pem"  # CHANGE ME
+key = "/etc/proxy/tls/privkey.pem"     # CHANGE ME
+default = true
+
+[tls.protocols]
+min_version = "1.2"
+
+# ---------------------------------------------------------------------------
+# Routing.
+#
+# default_destination is "legacy" so that accounts not yet migrated resolve to
+# the Dovecot/Postfix stack. As each account is migrated, an entry pointing it at
+# "stalwart" is added to the mapping file.
+#
+# smtp_passthrough_destination is "stalwart": port 25 carries no login to route
+# on, so it is handed whole to the new server, which performs split delivery and
+# relays not-yet-migrated recipients back to Postfix.
+# ---------------------------------------------------------------------------
+[routing]
+default_destination = "legacy"
+smtp_passthrough_destination = "stalwart"
+
+[mapping]
+source = "file"
+normalize = "lowercase"
+positive_ttl = "10m"
+negative_ttl = "30s"
+
+[mapping.file]
+# Whitespace-separated "identifier  destination" lines. A migrated account gets a
+# line pointing it at "stalwart"; everything else falls through to "legacy". For
+# example:
+#   alice@example.com   stalwart
+path = "/etc/proxy/mappings.tsv"       # CHANGE ME
+
+# ---------------------------------------------------------------------------
+# Legacy Dovecot + Postfix backend. Announced to the backend over XCLIENT and the
+# IMAP ID command (forwarding = "xclient"), which is what Dovecot and Postfix
+# understand. The backend leg uses STARTTLS on the classic plaintext ports.
+#
+# The legacy server presents a self-signed certificate. Pin it by fingerprint
+# with tls_pinned_cert_sha256 (recommended), or, on a trusted internal network,
+# set tls_allow_invalid_certs = true.
+# ---------------------------------------------------------------------------
+[destination.legacy]
+host = "legacy.internal.example.com"   # CHANGE ME (internal address of the legacy stack)
+tls_server_name = "legacy.internal.example.com"
+forwarding = "xclient"
+# tls_pinned_cert_sha256 = "aa:bb:...:99"   # CHANGE ME (legacy cert SHA-256), or:
+tls_allow_invalid_certs = true
+
+[destination.legacy.protocol.imap]
+port = 143
+tls = "starttls"
+[destination.legacy.protocol.pop3]
+port = 110
+tls = "starttls"
+[destination.legacy.protocol.submission]
+port = 587
+tls = "starttls"
+[destination.legacy.protocol.managesieve]
+port = 4190
+tls = "starttls"
+
+# ---------------------------------------------------------------------------
+# New Stalwart backend. Announced over the PROXY protocol (forwarding = "proxy",
+# proxy_protocol = true); Stalwart is configured to trust the proxy's address.
+# Declares an smtp endpoint because it is the pass-through destination for port 25.
+# ---------------------------------------------------------------------------
+[destination.stalwart]
+host = "new.internal.example.com"      # CHANGE ME (internal address of the new server)
+tls_server_name = "new.internal.example.com"
+forwarding = "proxy"
+proxy_protocol = true
+forwarded = true
+# tls_allow_invalid_certs = true       # if the new server uses a self-signed cert
+
+[destination.stalwart.protocol.imap]
+port = 993
+tls = "implicit"
+[destination.stalwart.protocol.pop3]
+port = 995
+tls = "implicit"
+[destination.stalwart.protocol.submission]
+port = 587
+tls = "starttls"
+[destination.stalwart.protocol.managesieve]
+port = 4190
+tls = "starttls"
+[destination.stalwart.protocol.smtp]
+port = 25
+tls = "plain"
+
+# ---------------------------------------------------------------------------
+# Client-facing listeners on the familiar ports. Port 25 is a raw pass-through to
+# Stalwart and therefore plain (a pass-through listener cannot offer STARTTLS).
+#
+# Note: Dovecot and Postfix serve no HTTP or JMAP, so no HTTP listener is defined
+# here. Once accounts are on Stalwart, JMAP and the web interface are served by
+# the new server; an HTTP listener routing to "stalwart" can be added at that
+# point if a single hostname for web access is wanted.
+# ---------------------------------------------------------------------------
+[listener.imaps]
+protocol = "imap"
+bind = ["0.0.0.0:993", "[::]:993"]
+tls = "implicit"
+[listener.imap]
+protocol = "imap"
+bind = ["0.0.0.0:143", "[::]:143"]
+tls = "starttls"
+[listener.pop3s]
+protocol = "pop3"
+bind = ["0.0.0.0:995", "[::]:995"]
+tls = "implicit"
+[listener.pop3]
+protocol = "pop3"
+bind = ["0.0.0.0:110", "[::]:110"]
+tls = "starttls"
+[listener.submission]
+protocol = "submission"
+bind = ["0.0.0.0:587", "[::]:587"]
+tls = "starttls"
+[listener.submissions]
+protocol = "submission"
+bind = ["0.0.0.0:465", "[::]:465"]
+tls = "implicit"
+[listener.managesieve]
+protocol = "managesieve"
+bind = ["0.0.0.0:4190", "[::]:4190"]
+tls = "starttls"
+[listener.smtp]
+protocol = "smtp"
+bind = ["0.0.0.0:25", "[::]:25"]
+tls = "plain"
+
+# ---------------------------------------------------------------------------
+# Management API: mapping changes, cache invalidation and connection control.
+# TLS is mandatory. Bind to loopback or a trusted management network only.
+# ---------------------------------------------------------------------------
+[admin]
+bind = "127.0.0.1:9443"
+tls = "implicit"
+bearer_token_file = "/etc/proxy/admin.token"  # CHANGE ME
+min_token_len = 32
+lockout_threshold = 5
+lockout_duration = "5m"

resources/config.stalwart-upgrade.toml

@@ -0,0 +1,252 @@
+# ===========================================================================
+# Migration proxy: Stalwart upgrade (old -> new) sample configuration.
+#
+# This configuration fronts two Stalwart deployments at the same time: an older
+# one ("old", typically v0.15 or below) that still holds most accounts, and a
+# new one ("new", the latest release) that accounts are migrated onto one by
+# one. Clients keep talking to a single hostname on the usual ports; the proxy
+# decides per account which deployment currently holds the mailbox.
+#
+# Before each account is migrated, it resolves to "old". Inbound mail on port 25
+# is always handed to "new", which performs split delivery: mail for an account
+# that already lives on "new" is delivered locally, while mail for an account
+# still on "old" is relayed there. As each account is moved its mapping is
+# flipped to "new".
+#
+# Replace every value marked CHANGE ME and review the rest before starting.
+# ===========================================================================
+
+[server]
+# Public hostname clients connect to. Advertised in protocol greetings.
+hostname = "mail.example.org"          # CHANGE ME
+shutdown_grace = "30s"
+log_level = "info"
+
+# TLS certificate the proxy presents to clients. This is the certificate for
+# the public hostname above; it terminates implicit TLS and STARTTLS.
+[tls.certificate.default]
+cert = "/etc/proxy/tls/fullchain.pem"  # CHANGE ME
+key = "/etc/proxy/tls/privkey.pem"     # CHANGE ME
+default = true
+
+[tls.protocols]
+min_version = "1.2"
+
+# ---------------------------------------------------------------------------
+# Routing.
+#
+# default_destination is where accounts with no explicit mapping resolve. While
+# the migration is in progress this must be "old", so that accounts that have
+# not been moved yet keep reaching their current mailbox.
+#
+# smtp_passthrough_destination is the fixed target for raw SMTP on port 25 and
+# LMTP on port 24. It must be "new", because the new deployment is the one that
+# performs split delivery and relays not-yet-migrated recipients back to old.
+# ---------------------------------------------------------------------------
+[routing]
+default_destination = "old"
+smtp_passthrough_destination = "new"
+
+# ---------------------------------------------------------------------------
+# Mapping store: identifier -> destination. A migrated account gets a line
+# pointing it at "new"; everything else falls through to default_destination.
+# The identifier is the login the client presents (the e-mail address, or for
+# older servers the bare account name), normalized to lowercase.
+# ---------------------------------------------------------------------------
+[mapping]
+source = "file"
+normalize = "lowercase"
+positive_ttl = "10m"
+negative_ttl = "30s"
+
+[mapping.file]
+# A whitespace-separated file of "identifier  destination" lines, for example:
+#   alice@example.org   new
+#   alice               new
+# List both the e-mail address and the bare account name when the old server
+# allowed logging in with either form. New entries take effect after a mapping
+# reload (see the management API) or a cache expiry.
+path = "/etc/proxy/mappings.tsv"       # CHANGE ME
+
+# ---------------------------------------------------------------------------
+# Destinations.
+#
+# Both backends are Stalwart and both are configured to accept the PROXY
+# protocol on their listeners, so the proxy announces the real client over a
+# PROXY v2 header (forwarding = "proxy", proxy_protocol = true) on every mail
+# leg, and appends Forwarded / X-Forwarded-For on the HTTP leg (forwarded =
+# true). The old server is the single exception: its port 25 must NOT expect a
+# PROXY header, because the new server relays to it as an ordinary SMTP client.
+# That exception is configured on the old server, not here.
+#
+# host is the internal address each deployment now listens on. tls_server_name
+# is the name presented for SNI and certificate matching on the backend leg.
+# If a backend uses a self-signed or internal certificate, either pin it with
+# tls_pinned_cert_sha256 or, for a trusted internal network only, uncomment
+# tls_allow_invalid_certs.
+# ---------------------------------------------------------------------------
+[destination.old]
+host = "old.internal.example.org"      # CHANGE ME
+tls_server_name = "old.internal.example.org"
+forwarding = "proxy"
+proxy_protocol = true
+forwarded = true
+# tls_allow_invalid_certs = true
+
+[destination.old.protocol.imap]
+port = 993
+tls = "implicit"
+[destination.old.protocol.pop3]
+port = 995
+tls = "implicit"
+[destination.old.protocol.submission]
+port = 587
+tls = "starttls"
+[destination.old.protocol.managesieve]
+port = 4190
+tls = "starttls"
+[destination.old.protocol.http]
+port = 443
+tls = "implicit"
+
+[destination.new]
+host = "new.internal.example.org"      # CHANGE ME
+tls_server_name = "new.internal.example.org"
+forwarding = "proxy"
+proxy_protocol = true
+forwarded = true
+# tls_allow_invalid_certs = true
+
+[destination.new.protocol.imap]
+port = 993
+tls = "implicit"
+[destination.new.protocol.pop3]
+port = 995
+tls = "implicit"
+[destination.new.protocol.submission]
+port = 587
+tls = "starttls"
+[destination.new.protocol.managesieve]
+port = 4190
+tls = "starttls"
+[destination.new.protocol.http]
+port = 443
+tls = "implicit"
+# Required because new is the smtp_passthrough_destination (port 25 / LMTP).
+[destination.new.protocol.smtp]
+port = 25
+tls = "plain"
+
+# ---------------------------------------------------------------------------
+# HTTP routing. Rules are evaluated top to bottom and the first match wins.
+#
+# The web interfaces and login endpoints differ between the two versions, so a
+# handful of paths that are unique to each version are pinned to the backend
+# that serves them. The OAuth token exchange is shared, so it is split by the
+# client_id present in the request body. Everything else, including JMAP,
+# WebDAV and the JMAP session resource, carries credentials and is routed by
+# the account those credentials identify, falling back to the default
+# destination when no account can be determined.
+# ---------------------------------------------------------------------------
+
+# Endpoints served only by the old deployment.
+[[http.route]]
+match = "/api/oauth"
+destination = "old"
+[[http.route]]
+match = "/authorize"
+destination = "old"
+[[http.route]]
+match = "/authorize/**"
+destination = "old"
+
+# Endpoints served only by the new deployment.
+[[http.route]]
+match = "/api/auth"
+destination = "new"
+[[http.route]]
+match = "/api/discover/**"
+destination = "new"
+[[http.route]]
+match = "/admin/**"
+destination = "new"
+[[http.route]]
+match = "/account/**"
+destination = "new"
+[[http.route]]
+match = "/login"
+destination = "new"
+[[http.route]]
+match = "/device"
+destination = "new"
+
+# OAuth token exchange is shared; route it by the web client that requested it.
+# Add matching mapping entries:  webadmin  old   and   stalwart-webui  new
+[[http.route]]
+match = "/auth/token"
+extract = { from = "body", regex = "client_id=([^&]+)" }
+fallback = "old"
+
+# Everything else: route by the authenticated account, fall back to default.
+[[http.route]]
+match = "/**"
+extract = { from = "authorization" }
+fallback = "default"
+
+# ---------------------------------------------------------------------------
+# Listeners: the public ports clients connect to. These take over the ports the
+# old server used to answer on directly.
+# ---------------------------------------------------------------------------
+[listener.imaps]
+protocol = "imap"
+bind = ["0.0.0.0:993", "[::]:993"]
+tls = "implicit"
+[listener.imap]
+protocol = "imap"
+bind = ["0.0.0.0:143", "[::]:143"]
+tls = "starttls"
+[listener.pop3s]
+protocol = "pop3"
+bind = ["0.0.0.0:995", "[::]:995"]
+tls = "implicit"
+[listener.pop3]
+protocol = "pop3"
+bind = ["0.0.0.0:110", "[::]:110"]
+tls = "starttls"
+[listener.submission]
+protocol = "submission"
+bind = ["0.0.0.0:587", "[::]:587"]
+tls = "starttls"
+[listener.submissions]
+protocol = "submission"
+bind = ["0.0.0.0:465", "[::]:465"]
+tls = "implicit"
+[listener.managesieve]
+protocol = "managesieve"
+bind = ["0.0.0.0:4190", "[::]:4190"]
+tls = "starttls"
+[listener.https]
+protocol = "http"
+bind = ["0.0.0.0:443", "[::]:443"]
+tls = "implicit"
+# Strip any Forwarded / X-Forwarded-For headers arriving from clients rather
+# than trusting them; the proxy sets its own based on the real peer.
+forwarded = "off"
+# Raw SMTP. Pass-through to new (smtp_passthrough_destination). Must be plain:
+# a pass-through listener cannot offer STARTTLS.
+[listener.smtp]
+protocol = "smtp"
+bind = ["0.0.0.0:25", "[::]:25"]
+tls = "plain"
+
+# ---------------------------------------------------------------------------
+# Management API: mapping changes, cache invalidation and connection control.
+# TLS is mandatory. Bind it to loopback or a trusted management network only.
+# ---------------------------------------------------------------------------
+[admin]
+bind = "127.0.0.1:9443"
+tls = "implicit"
+bearer_token_file = "/etc/proxy/admin.token"  # CHANGE ME
+min_token_len = 32
+lockout_threshold = 5
+lockout_duration = "5m"

resources/dovecot-postfix-trust.conf

@@ -0,0 +1,65 @@
+# ===========================================================================
+# Legacy backend trust: Dovecot + Postfix settings for the migration proxy.
+#
+# These are the changes the legacy Dovecot and Postfix stack needs so that the
+# migration proxy can front it. They do two things:
+#
+#   1. Trust the proxy to announce the real client address. The proxy forwards it
+#      over the XCLIENT command (Postfix submission and Dovecot ManageSieve/POP3)
+#      and the IMAP ID command, instead of a PROXY protocol header. Without this
+#      trust the backend records the proxy's address as the client; logins still
+#      work, but the real client address is lost.
+#
+#   2. Provide a master user, so the migration tool (Vandelay) can read any
+#      mailbox over IMAP with a single administrative credential rather than each
+#      user's password.
+#
+# Replace the address 203.0.113.10 with the proxy's actual address. Restrict the
+# trust to the proxy, not to the whole network, so that arbitrary hosts cannot
+# spoof a client address.
+# ===========================================================================
+
+# ---------------------------------------------------------------------------
+# Dovecot: /etc/dovecot/dovecot.conf (or a file under /etc/dovecot/conf.d)
+# ---------------------------------------------------------------------------
+
+# Trust the proxy to forward the real client address via XCLIENT and the IMAP ID
+# command. List the proxy's address only.
+login_trusted_networks = 203.0.113.10/32
+
+# Master user support, used by Vandelay to read any mailbox during migration. The
+# login presented is "<user>%<master-user>" (here the separator is "%"), with the
+# master user's password. For example, reading alice's mailbox uses the login
+# "alice@example.com%vmadmin" and vmadmin's password.
+auth_master_user_separator = %
+
+passdb {
+  driver = passwd-file
+  master = yes
+  args = scheme=PLAIN username_format=%u /etc/dovecot/master-users
+}
+
+# /etc/dovecot/master-users (one line; use a real password hash in production,
+# for example one produced by "doveadm pw -s SHA512-CRYPT"):
+#   vmadmin:{PLAIN}ChangeThisMasterPassword
+
+# ---------------------------------------------------------------------------
+# Postfix: apply with postconf, or add to /etc/postfix/main.cf
+# ---------------------------------------------------------------------------
+
+# Trust the proxy to announce the real client over XCLIENT on the SMTP and
+# submission services. List the proxy's address only.
+#
+#   postconf -e "smtpd_authorized_xclient_hosts=203.0.113.10/32"
+#
+# main.cf equivalent:
+# smtpd_authorized_xclient_hosts = 203.0.113.10/32
+
+# After changing either server, reload it:
+#   doveadm reload          # Dovecot
+#   postfix reload          # Postfix
+#
+# To confirm XCLIENT is being honoured, sign in through the proxy and check the
+# Dovecot login line: the "rip=" field should show the real client address rather
+# than the proxy's. If it shows the proxy's address, login_trusted_networks does
+# not include the proxy.

resources/migrate_v015_to_plan.py

@@ -0,0 +1,298 @@
+#!/usr/bin/env python3
+"""Build a Stalwart CLI migration plan from a Stalwart v0.15 server.
+
+This reads principal (account) definitions from a running Stalwart v0.15
+server through its management API and emits an NDJSON plan that the Stalwart
+CLI can apply against a new (latest) Stalwart server to recreate those
+accounts:
+
+    stalwart-cli apply --file migration-plan.ndjson
+
+Each line of the plan is a create operation in the form consumed by
+`stalwart-cli apply`:
+
+    {"@type":"create","object":"Domain","value":{"<ref>":{...}}}
+    {"@type":"create","object":"Account","value":{"<ref>":{"@type":"User",...}}}
+
+Accounts reference their domain through a plan-local "#<ref>" placeholder that
+the CLI resolves while applying. When a domain already exists on the target
+server, pass --existing-domain NAME=ID so accounts reference the real id
+instead of trying to recreate the domain.
+
+Password hashes are migrated verbatim, so users keep their existing passwords.
+App passwords and one-time-password (TOTP) secrets are not migrated.
+
+If a username is given, only that account is exported; otherwise every
+individual (optionally every group with --include-groups) is exported.
+"""
+
+import argparse
+import base64
+import json
+import ssl
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+
+PAGE_SIZE = 200
+REQUEST_TIMEOUT = 60
+
+
+class ApiError(RuntimeError):
+    pass
+
+
+class StalwartV015Client:
+    def __init__(self, url, token=None, username=None, password=None, insecure=False):
+        self.base = url.rstrip("/")
+        if token:
+            self.auth = "Bearer " + token
+        elif username is not None:
+            raw = f"{username}:{password or ''}".encode()
+            self.auth = "Basic " + base64.b64encode(raw).decode()
+        else:
+            raise ValueError("either a token or a username must be provided")
+        self.ctx = None
+        if insecure:
+            self.ctx = ssl.create_default_context()
+            self.ctx.check_hostname = False
+            self.ctx.verify_mode = ssl.CERT_NONE
+
+    def get(self, path, params=None):
+        url = self.base + path
+        if params:
+            url += "?" + urllib.parse.urlencode(params)
+        req = urllib.request.Request(url, method="GET")
+        req.add_header("Authorization", self.auth)
+        req.add_header("Accept", "application/json")
+        try:
+            with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT, context=self.ctx) as resp:
+                payload = json.loads(resp.read().decode())
+        except urllib.error.HTTPError as exc:
+            body = exc.read().decode(errors="replace")
+            raise ApiError(f"{exc.code} {exc.reason} for {path}: {body}") from exc
+        except urllib.error.URLError as exc:
+            raise ApiError(f"cannot reach {url}: {exc.reason}") from exc
+        if isinstance(payload, dict) and "data" in payload:
+            return payload["data"]
+        return payload
+
+    def list_principal_names(self, types):
+        names = []
+        for kind in types:
+            page = 1
+            while True:
+                data = self.get(
+                    "/api/principal",
+                    {"page": page, "limit": PAGE_SIZE, "types": kind},
+                )
+                items = data.get("items") or []
+                for item in items:
+                    name = principal_name(item)
+                    if name:
+                        names.append(name)
+                total = data.get("total") or 0
+                if page * PAGE_SIZE >= total or not items:
+                    break
+                page += 1
+        return names
+
+    def fetch_principal(self, name):
+        return self.get("/api/principal/" + urllib.parse.quote(name, safe=""))
+
+
+def principal_name(record):
+    name = record.get("name")
+    if isinstance(name, dict):
+        return name.get("string")
+    return name
+
+
+def scalar_list(value):
+    if value is None:
+        return []
+    if isinstance(value, list):
+        return value
+    if isinstance(value, dict):
+        if "stringList" in value:
+            return value["stringList"]
+        if "string" in value:
+            return [value["string"]]
+        return list(value.values())
+    return [value]
+
+
+def scalar(value):
+    if isinstance(value, dict):
+        for key in ("string", "integer"):
+            if key in value:
+                return value[key]
+        return None
+    return value
+
+
+def domain_ref(domain):
+    safe = "".join(c if c.isalnum() else "_" for c in domain)
+    return "domain_" + safe
+
+
+def split_address(record):
+    name = principal_name(record) or ""
+    emails = [e for e in scalar_list(record.get("emails")) if e]
+    primary = emails[0] if emails else name
+    if "@" in primary:
+        local, _, domain = primary.partition("@")
+    elif "@" in name:
+        local, _, domain = name.partition("@")
+        primary = name
+    else:
+        local, domain = name, None
+    return local, domain, primary, emails
+
+
+def build_credentials(record):
+    creds = {}
+    index = 0
+    for secret in scalar_list(record.get("secrets")):
+        if not isinstance(secret, str) or not secret:
+            continue
+        if secret.startswith("$app$"):
+            continue
+        if secret.startswith("otpauth://"):
+            continue
+        creds[str(index)] = {"@type": "Password", "secret": secret}
+        index += 1
+    return creds
+
+
+def build_aliases(primary, emails, domain_refs, existing):
+    aliases = {}
+    index = 0
+    for email in emails:
+        if email == primary or "@" not in email:
+            continue
+        local, _, domain = email.partition("@")
+        ref = existing.get(domain) or ("#" + domain_ref(domain))
+        aliases[str(index)] = {"name": local, "domainId": ref}
+        domain_refs.add(domain)
+        index += 1
+    return aliases
+
+
+def build_account(record, domain_value, existing):
+    kind = scalar(record.get("type"))
+    local, domain, primary, emails = split_address(record)
+    if not domain:
+        raise ApiError(
+            f"principal {principal_name(record)!r} has no email domain; cannot map it"
+        )
+    seen_domains = {domain}
+    domain_id = existing.get(domain) or ("#" + domain_ref(domain))
+
+    account = {
+        "@type": "Group" if kind == "group" else "User",
+        "name": local,
+        "domainId": domain_id,
+    }
+    description = scalar(record.get("description"))
+    if description:
+        account["description"] = description
+    creds = build_credentials(record)
+    if creds and kind != "group":
+        account["credentials"] = creds
+    if kind != "group":
+        account["roles"] = {"@type": "User"}
+        account["encryptionAtRest"] = {"@type": "Disabled"}
+    aliases = build_aliases(primary, emails, seen_domains, existing)
+    if aliases:
+        account["aliases"] = aliases
+
+    for dom in seen_domains:
+        if dom not in existing:
+            domain_value.setdefault(domain_ref(dom), {"name": dom})
+    return account
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Build a Stalwart CLI migration plan from a Stalwart v0.15 server.",
+    )
+    parser.add_argument("username", nargs="?", help="export only this account (default: all)")
+    parser.add_argument("--url", required=True, help="base URL of the v0.15 server")
+    parser.add_argument("--token", help="admin bearer token")
+    parser.add_argument("--user", help="admin username (HTTP basic auth)")
+    parser.add_argument("--password", help="admin password (HTTP basic auth)")
+    parser.add_argument("--insecure", action="store_true", help="skip TLS verification")
+    parser.add_argument("--include-groups", action="store_true", help="also export groups")
+    parser.add_argument(
+        "--existing-domain",
+        action="append",
+        default=[],
+        metavar="NAME=ID",
+        help="reference an existing target domain by id instead of creating it",
+    )
+    parser.add_argument(
+        "--output",
+        help="write the plan to this file (default: stdout)",
+    )
+    args = parser.parse_args()
+
+    if not args.token and args.user is None:
+        parser.error("provide --token or --user/--password")
+
+    existing = {}
+    for entry in args.existing_domain:
+        if "=" not in entry:
+            parser.error(f"--existing-domain expects NAME=ID, got {entry!r}")
+        name, _, value = entry.partition("=")
+        existing[name.strip()] = value.strip()
+
+    client = StalwartV015Client(
+        args.url,
+        token=args.token,
+        username=args.user,
+        password=args.password,
+        insecure=args.insecure,
+    )
+
+    if args.username:
+        names = [args.username]
+    else:
+        types = ["individual"]
+        if args.include_groups:
+            types.append("group")
+        names = client.list_principal_names(types)
+
+    if not names:
+        sys.stderr.write("no accounts to export\n")
+        return 1
+
+    domain_value = {}
+    account_value = {}
+    for name in names:
+        record = client.fetch_principal(name)
+        account = build_account(record, domain_value, existing)
+        account_value["acct_" + str(scalar(record.get("id")) or name)] = account
+
+    ops = []
+    if domain_value:
+        ops.append({"@type": "create", "object": "Domain", "value": domain_value})
+    ops.append({"@type": "create", "object": "Account", "value": account_value})
+
+    out = open(args.output, "w") if args.output else sys.stdout
+    try:
+        for op in ops:
+            out.write(json.dumps(op) + "\n")
+    finally:
+        if args.output:
+            out.close()
+
+    sys.stderr.write(
+        f"exported {len(account_value)} account(s), {len(domain_value)} domain(s)\n"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

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(());
@@ -237,6 +242,8 @@ async fn handle(
                     serde_json::json!({
                         "destination": id,
                         "consecutive_failures": s.consecutive_failures,
+                        "failure_sources": s.failure_sources,
+                        "host_level_failure": s.host_level_failure,
                         "down": s.down,
                     })
                 })
@@ -651,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

@@ -140,6 +140,8 @@ pub struct ServerConfig {
     pub backend_connect_retries: u32,
     #[serde(default = "default_host_down_threshold")]
     pub host_down_threshold: u32,
+    #[serde(default = "default_host_down_min_clients")]
+    pub host_down_min_clients: u32,
     #[serde(default = "default_host_down_cooldown", with = "duration")]
     pub host_down_cooldown: Duration,
     #[serde(default = "default_proxy_ttl")]
@@ -159,6 +161,7 @@ impl Default for ServerConfig {
             backend_timeout: DEFAULT_BACKEND_TIMEOUT,
             backend_connect_retries: DEFAULT_BACKEND_CONNECT_RETRIES,
             host_down_threshold: DEFAULT_HOST_DOWN_THRESHOLD,
+            host_down_min_clients: DEFAULT_HOST_DOWN_MIN_CLIENTS,
             host_down_cooldown: DEFAULT_HOST_DOWN_COOLDOWN,
             proxy_ttl: DEFAULT_PROXY_TTL,
             log_level: DEFAULT_LOG_LEVEL.to_string(),
@@ -181,6 +184,7 @@ const DEFAULT_BRIDGE_IDLE: Duration = Duration::from_secs(1800);
 const DEFAULT_BACKEND_TIMEOUT: Duration = Duration::from_secs(10);
 const DEFAULT_BACKEND_CONNECT_RETRIES: u32 = 1;
 const DEFAULT_HOST_DOWN_THRESHOLD: u32 = 5;
+const DEFAULT_HOST_DOWN_MIN_CLIENTS: u32 = 2;
 const DEFAULT_HOST_DOWN_COOLDOWN: Duration = Duration::from_secs(30);
 const DEFAULT_PROXY_TTL: u32 = 7;
 const DEFAULT_LOG_LEVEL: &str = "info";
@@ -200,6 +204,9 @@ fn default_backend_connect_retries() -> u32 {
 fn default_host_down_threshold() -> u32 {
     DEFAULT_HOST_DOWN_THRESHOLD
 }
+fn default_host_down_min_clients() -> u32 {
+    DEFAULT_HOST_DOWN_MIN_CLIENTS
+}
 fn default_host_down_cooldown() -> Duration {
     DEFAULT_HOST_DOWN_COOLDOWN
 }
@@ -489,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> {
@@ -505,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")]
@@ -789,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/error.rs

@@ -25,6 +25,9 @@ pub enum ProxyError {
     #[error("backend connection failed: {0}")]
     BackendConnect(Cow<'static, str>),
 
+    #[error("backend session failed: {0}")]
+    BackendSession(Cow<'static, str>),
+
     #[error("protocol error: {0}")]
     Protocol(Cow<'static, str>),
 
@@ -52,8 +55,15 @@ impl ProxyError {
         ProxyError::BackendConnect(msg.into())
     }
 
+    pub fn backend_session(msg: impl Into<Cow<'static, str>>) -> Self {
+        ProxyError::BackendSession(msg.into())
+    }
+
     pub fn is_retryable(&self) -> bool {
-        matches!(self, ProxyError::BackendConnect(_) | ProxyError::Io(_))
+        matches!(
+            self,
+            ProxyError::BackendConnect(_) | ProxyError::BackendSession(_) | ProxyError::Io(_)
+        )
     }
 
     pub fn protocol(msg: impl Into<Cow<'static, str>>) -> Self {

src/http/proxy.rs

@@ -502,7 +502,7 @@ async fn serve(
         let _ = write_client(stream, &error_response(502, "Bad Gateway")).await;
         return Ok(Disposition::Close);
     }
-    let mut backend = match crate::proto::common::establish(ctx, &dest_id, || {
+    let mut backend = match crate::proto::common::establish(ctx, &dest_id, peer, || {
         crate::outbound::dial(dest, ep, fwd, &tls_cfg, peer, local)
     })
     .await

src/outbound.rs

@@ -134,7 +134,7 @@ pub async fn tls_connect(
     let tls = connector
         .connect(name, stream)
         .await
-        .map_err(|e| ProxyError::backend_connect(format!("backend TLS handshake failed: {e}")))?;
+        .map_err(|e| ProxyError::backend_session(format!("backend TLS handshake failed: {e}")))?;
     Ok(Box::new(tls))
 }
 
@@ -160,9 +160,9 @@ pub async fn read_line(
         let n = stream
             .read(&mut buf)
             .await
-            .map_err(|e| ProxyError::backend_connect(format!("reading backend: {e}")))?;
+            .map_err(|e| ProxyError::backend_session(format!("reading backend: {e}")))?;
         if n == 0 {
-            return Err(ProxyError::backend_connect(
+            return Err(ProxyError::backend_session(
                 "backend closed before sending a complete line",
             ));
         }

src/proto/common.rs

@@ -6,7 +6,7 @@
 
 use std::collections::HashMap;
 use std::future::Future;
-use std::net::SocketAddr;
+use std::net::{IpAddr, SocketAddr};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -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')
@@ -141,9 +166,35 @@ pub struct DestHealth {
 #[derive(Default)]
 struct HealthState {
     consecutive_failures: u32,
+    failure_sources: AHashSet<IpAddr>,
+    host_level_failure: bool,
     down_until: Option<Instant>,
 }
 
+impl HealthState {
+    fn clear_evidence(&mut self) {
+        self.failure_sources.clear();
+        self.host_level_failure = false;
+    }
+}
+
+const MAX_TRACKED_FAILURE_SOURCES: usize = 64;
+
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub enum Attribution {
+    Host,
+    Client(IpAddr),
+}
+
+impl Attribution {
+    fn of(err: &ProxyError, peer: IpAddr) -> Self {
+        match err {
+            ProxyError::BackendConnect(_) => Attribution::Host,
+            _ => Attribution::Client(peer),
+        }
+    }
+}
+
 impl DestHealth {
     fn gate(&self) -> Result<()> {
         let mut st = self.inner.lock();
@@ -154,26 +205,39 @@ impl DestHealth {
                 ));
             }
             st.down_until = None;
+            st.clear_evidence();
         }
         Ok(())
     }
 
-    fn record(&self, ok: bool, threshold: u32, cooldown: Duration) {
+    fn record_success(&self) {
         let mut st = self.inner.lock();
-        if ok {
-            st.consecutive_failures = 0;
-            st.down_until = None;
-        } else {
-            st.consecutive_failures = st.consecutive_failures.saturating_add(1);
-            if threshold > 0 && st.consecutive_failures >= threshold {
-                st.down_until = Some(Instant::now() + cooldown);
+        st.consecutive_failures = 0;
+        st.clear_evidence();
+        st.down_until = None;
+    }
+
+    fn record_failure(&self, at: Attribution, limits: &HealthLimits) {
+        let mut st = self.inner.lock();
+        st.consecutive_failures = st.consecutive_failures.saturating_add(1);
+        match at {
+            Attribution::Host => st.host_level_failure = true,
+            Attribution::Client(ip) => {
+                if st.failure_sources.len() < MAX_TRACKED_FAILURE_SOURCES {
+                    st.failure_sources.insert(ip);
+                }
             }
         }
+        let broad = st.host_level_failure || st.failure_sources.len() as u32 >= limits.min_clients;
+        if limits.threshold > 0 && st.consecutive_failures >= limits.threshold && broad {
+            st.down_until = Some(Instant::now() + limits.cooldown);
+        }
     }
 
     pub fn reset(&self) {
         let mut st = self.inner.lock();
         st.consecutive_failures = 0;
+        st.clear_evidence();
         st.down_until = None;
     }
 
@@ -182,13 +246,33 @@ impl DestHealth {
         let down = matches!(st.down_until, Some(until) if Instant::now() < until);
         HealthSnapshot {
             consecutive_failures: st.consecutive_failures,
+            failure_sources: st.failure_sources.len() as u32,
+            host_level_failure: st.host_level_failure,
             down,
         }
     }
 }
 
+pub struct HealthLimits {
+    threshold: u32,
+    min_clients: u32,
+    cooldown: Duration,
+}
+
+impl HealthLimits {
+    fn for_server(server: &crate::config::ServerConfig) -> Self {
+        HealthLimits {
+            threshold: server.host_down_threshold,
+            min_clients: server.host_down_min_clients,
+            cooldown: server.host_down_cooldown,
+        }
+    }
+}
+
 pub struct HealthSnapshot {
     pub consecutive_failures: u32,
+    pub failure_sources: u32,
+    pub host_level_failure: bool,
     pub down: bool,
 }
 
@@ -465,7 +549,12 @@ pub fn assert_egress_encrypted(
 
 const BACKEND_RETRY_DELAY: Duration = Duration::from_millis(200);
 
-pub async fn establish<F, Fut, T>(ctx: &Ctx, dest_id: &str, mut open: F) -> Result<T>
+pub async fn establish<F, Fut, T>(
+    ctx: &Ctx,
+    dest_id: &str,
+    peer: SocketAddr,
+    mut open: F,
+) -> Result<T>
 where
     F: FnMut() -> Fut,
     Fut: Future<Output = Result<T>>,
@@ -475,6 +564,7 @@ where
     }
 
     let server = &ctx.config.server;
+    let limits = HealthLimits::for_server(server);
     let mut attempt: u32 = 0;
     loop {
         let outcome = match tokio::time::timeout(server.backend_timeout, open()).await {
@@ -487,7 +577,7 @@ where
         match outcome {
             Ok(value) => {
                 if let Some(health) = ctx.dest_health.get(dest_id) {
-                    health.record(true, server.host_down_threshold, server.host_down_cooldown);
+                    health.record_success();
                 }
                 return Ok(value);
             }
@@ -504,7 +594,15 @@ where
                     continue;
                 }
                 if let Some(health) = ctx.dest_health.get(dest_id) {
-                    health.record(false, server.host_down_threshold, server.host_down_cooldown);
+                    let at = Attribution::of(&err, peer.ip());
+                    if matches!(at, Attribution::Client(_)) {
+                        tracing::debug!(
+                            destination = %dest_id,
+                            error = %err,
+                            "backend failure attributed to the client, not counted as host evidence on its own"
+                        );
+                    }
+                    health.record_failure(at, &limits);
                 }
                 return Err(err);
             }
@@ -525,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,
@@ -617,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> {
@@ -625,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);
@@ -659,19 +808,20 @@ pub async fn bridge_authenticated<F, Fut>(
     ctx: &Ctx,
     dest_id: &str,
     mut stream: BoxedStream,
+    peer: SocketAddr,
     identifier: Option<&str>,
     client_residual: &[u8],
     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)
@@ -685,7 +835,11 @@ where
         return Ok(());
     }
 
-    let (mut backend, mut backend_residual) = match establish(ctx, dest_id, 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");
@@ -694,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);
@@ -704,6 +859,7 @@ where
             &mut backend,
             &mut backend_residual,
             wire,
+            pending_response,
             hide,
             max,
         ),
@@ -811,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";
@@ -830,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!(
@@ -854,29 +1096,189 @@ mod tests {
         ));
     }
 
+    fn limits(threshold: u32, min_clients: u32, cooldown: Duration) -> HealthLimits {
+        HealthLimits {
+            threshold,
+            min_clients,
+            cooldown,
+        }
+    }
+
+    fn client(addr: &str) -> Attribution {
+        Attribution::Client(addr.parse().unwrap())
+    }
+
     #[test]
     fn dest_health_opens_after_threshold_and_recovers() {
         let h = DestHealth::default();
-        let cooldown = Duration::from_secs(60);
+        let lim = limits(3, 2, Duration::from_secs(60));
 
-        h.record(false, 3, cooldown);
-        h.record(false, 3, cooldown);
+        h.record_failure(Attribution::Host, &lim);
+        h.record_failure(Attribution::Host, &lim);
         assert!(h.gate().is_ok());
 
-        h.record(false, 3, cooldown);
+        h.record_failure(Attribution::Host, &lim);
         assert!(h.gate().is_err());
 
-        h.record(true, 3, cooldown);
+        h.record_success();
         assert!(h.gate().is_ok());
     }
 
     #[test]
     fn dest_health_cooldown_elapses_into_a_probe() {
         let h = DestHealth::default();
-        h.record(false, 1, Duration::from_secs(0));
+        h.record_failure(Attribution::Host, &limits(1, 2, Duration::from_secs(0)));
         assert!(h.gate().is_ok());
     }
 
+    #[test]
+    fn attribution_follows_the_failure_stage() {
+        let ip: IpAddr = "203.0.113.9".parse().unwrap();
+        assert_eq!(
+            Attribution::of(&ProxyError::backend_connect("refused"), ip),
+            Attribution::Host
+        );
+        assert_eq!(
+            Attribution::of(&ProxyError::backend("closed after PROXY header"), ip),
+            Attribution::Client(ip)
+        );
+        assert_eq!(
+            Attribution::of(
+                &ProxyError::backend_session("backend closed before sending a complete line"),
+                ip
+            ),
+            Attribution::Client(ip)
+        );
+        assert_eq!(
+            Attribution::of(
+                &ProxyError::backend_session("backend TLS handshake failed"),
+                ip
+            ),
+            Attribution::Client(ip)
+        );
+        assert_eq!(
+            Attribution::of(
+                &ProxyError::Io(std::io::Error::from(std::io::ErrorKind::ConnectionReset)),
+                ip
+            ),
+            Attribution::Client(ip)
+        );
+    }
+
+    #[test]
+    fn banned_client_alone_never_opens_the_breaker() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(60));
+        for _ in 0..50 {
+            h.record_failure(client("198.51.100.7"), &lim);
+        }
+        assert!(h.gate().is_ok());
+        let s = h.snapshot();
+        assert_eq!(s.consecutive_failures, 50);
+        assert_eq!(s.failure_sources, 1);
+        assert!(!s.down);
+    }
+
+    #[test]
+    fn failures_spanning_distinct_clients_open_the_breaker() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(60));
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        assert!(h.gate().is_ok());
+
+        h.record_failure(client("203.0.113.4"), &lim);
+        assert!(h.gate().is_err());
+        assert_eq!(h.snapshot().failure_sources, 2);
+    }
+
+    #[test]
+    fn host_level_failure_opens_the_breaker_without_breadth() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(60));
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(Attribution::Host, &lim);
+        assert!(h.gate().is_err());
+        let s = h.snapshot();
+        assert!(s.host_level_failure);
+        assert!(s.down);
+    }
+
+    #[test]
+    fn cooldown_expiry_clears_stale_evidence() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(0));
+        for _ in 0..3 {
+            h.record_failure(Attribution::Host, &lim);
+        }
+        assert!(h.gate().is_ok());
+        let s = h.snapshot();
+        assert!(!s.host_level_failure);
+        assert_eq!(s.failure_sources, 0);
+
+        h.record_failure(
+            client("198.51.100.7"),
+            &limits(3, 2, Duration::from_secs(60)),
+        );
+        assert!(h.gate().is_ok());
+    }
+
+    #[test]
+    fn cooldown_expiry_still_re_opens_on_a_failed_host_probe() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(0));
+        for _ in 0..3 {
+            h.record_failure(Attribution::Host, &lim);
+        }
+        assert!(h.gate().is_ok());
+
+        h.record_failure(Attribution::Host, &limits(3, 2, Duration::from_secs(60)));
+        assert!(h.gate().is_err());
+    }
+
+    #[test]
+    fn success_clears_the_failure_streak() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(60));
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("203.0.113.4"), &lim);
+        h.record_success();
+        let s = h.snapshot();
+        assert_eq!(s.consecutive_failures, 0);
+        assert_eq!(s.failure_sources, 0);
+
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        assert!(h.gate().is_ok());
+    }
+
+    #[test]
+    fn min_clients_zero_counts_failures_alone() {
+        let h = DestHealth::default();
+        let lim = limits(3, 0, Duration::from_secs(60));
+        for _ in 0..3 {
+            h.record_failure(client("198.51.100.7"), &lim);
+        }
+        assert!(h.gate().is_err());
+    }
+
+    #[test]
+    fn tracked_failure_sources_are_bounded() {
+        let h = DestHealth::default();
+        let lim = limits(0, 2, Duration::from_secs(60));
+        for i in 0..(MAX_TRACKED_FAILURE_SOURCES + 100) {
+            let ip = IpAddr::from([10, (i / 256) as u8, (i % 256) as u8, 1]);
+            h.record_failure(Attribution::Client(ip), &lim);
+        }
+        assert_eq!(
+            h.snapshot().failure_sources as usize,
+            MAX_TRACKED_FAILURE_SOURCES
+        );
+    }
+
     #[tokio::test]
     async fn conn_registry_kick_signals_live_sessions() {
         let reg = ConnRegistry::default();
@@ -912,7 +1314,7 @@ mod tests {
     fn dest_health_threshold_zero_never_trips() {
         let h = DestHealth::default();
         for _ in 0..100 {
-            h.record(false, 0, Duration::from_secs(60));
+            h.record_failure(Attribution::Host, &limits(0, 2, Duration::from_secs(60)));
         }
         assert!(h.gate().is_ok());
     }

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");
 
@@ -89,12 +93,13 @@ pub async fn handle(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,
         "imap",
         common::AuthWire::Imap { tag },
-        auth_frame.as_bytes(),
+        &auth,
         unavailable.as_bytes(),
         MAX_PREAUTH,
         || open_backend(ctx, &dest_id, &fwd),
@@ -324,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(
@@ -348,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 {
@@ -365,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,
@@ -434,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,
@@ -447,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") {
@@ -457,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?;
     }
 
@@ -466,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,22 +66,26 @@ 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(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,
         "managesieve",
         common::AuthWire::ManageSieve,
-        auth_frame.as_bytes(),
+        &auth,
         b"NO \"backend unavailable\"\r\n",
         MAX_PREAUTH,
         || open_backend(ctx, &dest_id, &fwd),
@@ -172,6 +176,7 @@ async fn preauth(
                                 ctx,
                                 &mut stream,
                                 plain_ok,
+                                common::sasl_offered(&caps.sasl, "LOGIN"),
                                 &req.tokens,
                                 &mut it,
                                 &mut auth_failures,
@@ -238,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,
@@ -254,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",
@@ -265,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 {
@@ -289,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,
@@ -324,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
     }
 }
 
@@ -390,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,
@@ -431,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,22 +65,26 @@ 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(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,
         "pop3",
         common::AuthWire::Pop3,
-        auth_frame.as_bytes(),
+        &auth,
         b"-ERR backend unavailable\r\n",
         MAX_PREAUTH,
         || open_backend(ctx, &dest_id, &fwd),
@@ -219,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);
@@ -228,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",
@@ -238,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);
@@ -340,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,
@@ -383,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,22 +74,26 @@ 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(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         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),
@@ -180,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,
                         };
@@ -259,6 +266,7 @@ enum AuthStep {
 struct AuthParams<'a> {
     ctx: &'a Ctx,
     plain_ok: bool,
+    login_offered: bool,
     ehlo_host: &'a str,
     max_attempts: u32,
 }
@@ -274,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,
@@ -285,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",
@@ -295,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);
@@ -342,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>,
@@ -369,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
     }
@@ -410,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,
@@ -455,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> {
@@ -572,7 +688,7 @@ pub async fn handle_passthrough(
     let (dest, ep, tls_cfg) = ctx.endpoint(&dest_id, proto)?;
     let fwd = dest.forwarding_for(proto);
     forward::guard_loop(&ctx.self_binds, dest, ep, fwd, ctx.config.server.proxy_ttl)?;
-    let backend = common::establish(ctx, &dest_id, || {
+    let backend = common::establish(ctx, &dest_id, peer, || {
         outbound::dial(dest, ep, fwd, &tls_cfg, peer, local)
     })
     .await?;

src/tests/admin.rs

@@ -473,6 +473,10 @@ async fn admin_mapping_crud_and_reads() {
         body.contains("legacy") && body.contains("secondary"),
         "body: {body}"
     );
+    assert!(
+        body.contains("failure_sources") && body.contains("host_level_failure"),
+        "body: {body}"
+    );
 
     let (status, _) = admin_request(
         addr,

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/legacy.rs

@@ -7,10 +7,12 @@
 use std::time::Duration;
 
 use testcontainers::core::{IntoContainerPort, WaitFor};
-use testcontainers::runners::{AsyncBuilder, AsyncRunner};
-use testcontainers::{ContainerAsync, GenericBuildableImage, GenericImage, ImageExt};
+use testcontainers::runners::AsyncRunner;
+use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 use tokio::net::TcpStream;
 
+use super::{ensure_image, owner_labels};
+
 pub const MASTER_USER: &str = "admin";
 pub const MASTER_PASSWORD: &str = "MasterPass#2026";
 
@@ -218,15 +220,17 @@ pub struct Legacy {
 
 impl Legacy {
     pub async fn start() -> Legacy {
-        let image: GenericImage = GenericBuildableImage::new("proxy-legacymail", "test")
-            .with_dockerfile_string(DOCKERFILE.to_string())
-            .with_data(DOVECOT_CONF.as_bytes().to_vec(), "dovecot.conf")
-            .with_data(USERS.as_bytes().to_vec(), "users")
-            .with_data(MASTER_USERS.as_bytes().to_vec(), "master-users")
-            .with_data(ENTRYPOINT.as_bytes().to_vec(), "entrypoint.sh")
-            .build_image()
-            .await
-            .expect("build legacy image");
+        let image = ensure_image(
+            "proxy-legacymail",
+            DOCKERFILE,
+            &[
+                ("dovecot.conf", DOVECOT_CONF.as_bytes()),
+                ("users", USERS.as_bytes()),
+                ("master-users", MASTER_USERS.as_bytes()),
+                ("entrypoint.sh", ENTRYPOINT.as_bytes()),
+            ],
+        )
+        .await;
 
         let container = image
             .with_exposed_port(143u16.tcp())
@@ -239,6 +243,7 @@ impl Legacy {
             .with_exposed_port(25u16.tcp())
             .with_wait_for(WaitFor::message_on_stdout("LEGACY-MAIL READY"))
             .with_startup_timeout(Duration::from_secs(240))
+            .with_labels(owner_labels())
             .start()
             .await
             .expect("start legacy container");

src/tests/containers/mod.rs

@@ -5,12 +5,16 @@
  */
 
 use std::collections::HashMap;
+use std::hash::{BuildHasher, Hash, Hasher};
 use std::net::{SocketAddr, TcpListener as StdTcpListener};
-use std::sync::Arc;
-use std::time::Duration;
+use std::process::{Command, Stdio};
+use std::sync::{Arc, Once, OnceLock};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
+use testcontainers::runners::AsyncBuilder;
+use testcontainers::{GenericBuildableImage, GenericImage};
 use tokio::net::TcpStream;
-use tokio::sync::watch;
+use tokio::sync::{Mutex, watch};
 use tokio::task::JoinHandle;
 
 use crate::net::acceptor::Listener;
@@ -34,11 +38,100 @@ pub mod toolbox_matrix;
 
 pub const HOST_GATEWAY: &str = "host.docker.internal";
 
+pub const OWNER_LABEL: &str = "art.stalw.proxy.test";
+
 pub fn pick_free_port() -> u16 {
     let l = StdTcpListener::bind("127.0.0.1:0").unwrap();
     l.local_addr().unwrap().port()
 }
 
+fn run_id() -> &'static str {
+    static RUN_ID: OnceLock<String> = OnceLock::new();
+    RUN_ID.get_or_init(|| {
+        let nanos = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_nanos();
+        format!("{}-{nanos}", std::process::id())
+    })
+}
+
+pub fn owner_labels() -> [(&'static str, &'static str); 1] {
+    sweep_abandoned_containers();
+    [(OWNER_LABEL, run_id())]
+}
+
+fn docker_ids(filter: &str) -> Vec<String> {
+    Command::new("docker")
+        .args(["ps", "-aq", "--filter", filter])
+        .output()
+        .map(|out| {
+            String::from_utf8_lossy(&out.stdout)
+                .split_whitespace()
+                .map(String::from)
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
+fn sweep_abandoned_containers() {
+    static SWEPT: Once = Once::new();
+    SWEPT.call_once(|| {
+        let mine = docker_ids(&format!("label={OWNER_LABEL}={}", run_id()));
+        let stale: Vec<String> = docker_ids(&format!("label={OWNER_LABEL}"))
+            .into_iter()
+            .filter(|id| !mine.contains(id))
+            .collect();
+        if stale.is_empty() {
+            return;
+        }
+        eprintln!("removing {} abandoned test container(s)", stale.len());
+        let _ = Command::new("docker")
+            .args(["rm", "-f", "-v"])
+            .args(&stale)
+            .output();
+    });
+}
+
+fn content_tag(dockerfile: &str, data: &[(&str, &[u8])]) -> String {
+    let mut hasher = ahash::RandomState::with_seeds(0x5354, 0x414c, 0x5741, 0x5254).build_hasher();
+    dockerfile.hash(&mut hasher);
+    for (target, bytes) in data {
+        target.hash(&mut hasher);
+        bytes.hash(&mut hasher);
+    }
+    format!("{:016x}", hasher.finish())
+}
+
+async fn image_present(reference: &str) -> bool {
+    tokio::process::Command::new("docker")
+        .args(["image", "inspect", reference])
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .status()
+        .await
+        .map(|status| status.success())
+        .unwrap_or(false)
+}
+
+pub async fn ensure_image(name: &str, dockerfile: &str, data: &[(&str, &[u8])]) -> GenericImage {
+    static BUILD_LOCK: Mutex<()> = Mutex::const_new(());
+    let tag = content_tag(dockerfile, data);
+    let _guard = BUILD_LOCK.lock().await;
+    if image_present(&format!("{name}:{tag}")).await {
+        return GenericImage::new(name.to_owned(), tag);
+    }
+    let mut builder =
+        GenericBuildableImage::new(name, &tag).with_dockerfile_string(dockerfile.to_string());
+    for (target, bytes) in data {
+        builder = builder.with_data(bytes.to_vec(), *target);
+    }
+    builder
+        .build_image()
+        .await
+        .unwrap_or_else(|e| panic!("build image {name}:{tag}: {e}"))
+}
+
 pub struct Proxy {
     shutdown: watch::Sender<bool>,
     handles: Vec<JoinHandle<()>>,

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/containers/stalwart.rs

@@ -12,7 +12,7 @@ use testcontainers::runners::AsyncRunner;
 use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 
 use super::http::{self, basic_auth};
-use super::pick_free_port;
+use super::{owner_labels, pick_free_port};
 
 pub const NEW_IMAGE: &str = "stalwartlabs/stalwart";
 pub const NEW_TAG: &str = "latest";
@@ -114,6 +114,7 @@ impl Stalwart {
             .with_mapped_port(ports.sieve, 4190u16.tcp())
             .with_mapped_port(ports.smtp, 25u16.tcp())
             .with_startup_timeout(Duration::from_secs(180))
+            .with_labels(owner_labels())
             .start()
             .await
             .expect("start new stalwart");
@@ -196,6 +197,7 @@ impl Stalwart {
             .with_mapped_port(ports.sieve, 4190u16.tcp())
             .with_mapped_port(ports.smtp, 25u16.tcp())
             .with_startup_timeout(Duration::from_secs(180))
+            .with_labels(owner_labels())
             .start()
             .await
             .expect("start old stalwart");

src/tests/containers/toolbox.rs

@@ -7,11 +7,11 @@
 use std::time::Duration;
 
 use testcontainers::core::{ExecCommand, Host, WaitFor};
-use testcontainers::runners::{AsyncBuilder, AsyncRunner};
-use testcontainers::{ContainerAsync, GenericBuildableImage, GenericImage, ImageExt};
+use testcontainers::runners::AsyncRunner;
+use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 use tokio::io::AsyncReadExt;
 
-use super::HOST_GATEWAY;
+use super::{HOST_GATEWAY, ensure_image, owner_labels};
 
 const DOCKERFILE: &str = r#"FROM debian:trixie-slim
 ENV DEBIAN_FRONTEND=noninteractive
@@ -54,16 +54,13 @@ impl Exec {
 
 impl Toolbox {
     pub async fn start() -> Toolbox {
-        let image: GenericImage = GenericBuildableImage::new("proxy-toolbox", "test")
-            .with_dockerfile_string(DOCKERFILE.to_string())
-            .build_image()
-            .await
-            .expect("build toolbox image");
+        let image = ensure_image("proxy-toolbox", DOCKERFILE, &[]).await;
 
         let container = image
             .with_wait_for(WaitFor::seconds(1))
             .with_startup_timeout(Duration::from_secs(300))
             .with_host(HOST_GATEWAY, Host::HostGateway)
+            .with_labels(owner_labels())
             .start()
             .await
             .expect("start toolbox container");

src/tests/dest_health.rs

@@ -0,0 +1,274 @@
+/*
+ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ */
+
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::task::JoinHandle;
+
+use super::harness::*;
+use crate::proto::common::Ctx;
+
+fn config(port: u16, mapping_path: &str, threshold: u32, min_clients: u32) -> String {
+    format!(
+        r#"
+[server]
+host_down_threshold = {threshold}
+host_down_min_clients = {min_clients}
+host_down_cooldown = "60s"
+backend_connect_retries = 0
+
+[mapping]
+source = "file"
+normalize = "lowercase"
+[mapping.file]
+path = "{mapping_path}"
+
+[capabilities.imap]
+allow_plain_auth_without_tls = true
+
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+proxy_protocol = false
+allow_plaintext_auth = true
+[destination.legacy.protocol.imap]
+port = {port}
+tls = "plain"
+
+[listener.imap]
+protocol = "imap"
+bind = ["127.0.0.1:0"]
+tls = "plain"
+"#
+    )
+}
+
+struct Backend {
+    port: u16,
+    serving: Arc<AtomicBool>,
+    accepted: Arc<AtomicU32>,
+    handle: JoinHandle<()>,
+}
+
+async fn banning_backend() -> Backend {
+    let listener = TcpListener::bind(unused_addr()).await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let serving = Arc::new(AtomicBool::new(false));
+    let accepted = Arc::new(AtomicU32::new(0));
+    let serving_task = serving.clone();
+    let accepted_task = accepted.clone();
+    let handle = tokio::spawn(async move {
+        loop {
+            let Ok((socket, _)) = listener.accept().await else {
+                return;
+            };
+            let _ = socket.set_nodelay(true);
+            accepted_task.fetch_add(1, Ordering::Relaxed);
+            if serving_task.load(Ordering::Relaxed) {
+                tokio::spawn(serve_imap_auth(socket));
+            } else {
+                drop(socket);
+            }
+        }
+    });
+    Backend {
+        port,
+        serving,
+        accepted,
+        handle,
+    }
+}
+
+async fn serve_imap_auth(mut socket: TcpStream) {
+    let _ = socket
+        .write_all(b"* OK [CAPABILITY IMAP4rev2] backend ready\r\n")
+        .await;
+
+    let mut buf = Vec::new();
+    let mut scratch = [0u8; 1024];
+    let tag = loop {
+        let Ok(n) = socket.read(&mut scratch).await else {
+            return;
+        };
+        if n == 0 {
+            return;
+        }
+        buf.extend_from_slice(&scratch[..n]);
+        if let Some(end) = find(&buf, b"\r\n") {
+            let line = String::from_utf8_lossy(&buf[..end]).into_owned();
+            break line.split(' ').next().unwrap_or("a1").to_string();
+        }
+    };
+    let _ = socket
+        .write_all(format!("{tag} OK [CAPABILITY IMAP4rev2] authenticated\r\n").as_bytes())
+        .await;
+
+    loop {
+        if find(&buf, b"NOOP").is_some() {
+            break;
+        }
+        let Ok(n) = socket.read(&mut scratch).await else {
+            return;
+        };
+        if n == 0 {
+            return;
+        }
+        buf.extend_from_slice(&scratch[..n]);
+    }
+    let _ = socket.write_all(b"a2 OK NOOP done\r\n").await;
+}
+
+async fn imap_login(ctx: &Arc<Ctx>, peer: &str, user: &str) -> Vec<u8> {
+    let listener = ctx.config.listener["imap"].clone();
+    let peer: SocketAddr = peer.parse().unwrap();
+    let (mut client, server) = tcp_pair().await;
+    let ctx_task = ctx.clone();
+    let handler = tokio::spawn(async move {
+        let _ = crate::proto::imap::handle(
+            &ctx_task,
+            &listener,
+            Box::new(server),
+            peer,
+            "127.0.0.1:143".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let _greeting = read_line(&mut client).await;
+    send(
+        &mut client,
+        format!("a1 LOGIN {user} secret\r\na2 NOOP\r\n").as_bytes(),
+    )
+    .await;
+    let resp = read_to_eof(&mut client).await;
+    handler.abort();
+    resp
+}
+
+#[tokio::test]
+async fn one_banned_client_does_not_mark_the_destination_down() {
+    let backend = banning_backend().await;
+    let mapping = mapping_file("");
+    let toml = config(backend.port, &mapping.path().display().to_string(), 3, 2);
+    let ctx = make_ctx(&toml).await;
+
+    for i in 0..5 {
+        let resp = imap_login(&ctx, "198.51.100.7:40000", "banned@example.com").await;
+        assert!(
+            find(&resp, b"a1 NO [UNAVAILABLE]").is_some(),
+            "attempt {i} should report the backend unavailable, got {:?}",
+            String::from_utf8_lossy(&resp)
+        );
+    }
+
+    let health = ctx.dest_health.get("legacy").unwrap().snapshot();
+    assert_eq!(health.consecutive_failures, 5);
+    assert_eq!(health.failure_sources, 1);
+    assert!(!health.host_level_failure);
+    assert!(
+        !health.down,
+        "a single client must not open the circuit breaker"
+    );
+
+    backend.serving.store(true, Ordering::Relaxed);
+    let resp = imap_login(&ctx, "203.0.113.4:40000", "alice@example.com").await;
+    assert!(
+        find(&resp, b"a1 OK").is_some(),
+        "an unrelated account must still reach the backend, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    assert!(!ctx.dest_health.get("legacy").unwrap().snapshot().down);
+    assert!(backend.accepted.load(Ordering::Relaxed) >= 6);
+
+    backend.handle.abort();
+}
+
+#[tokio::test]
+async fn rejections_spanning_distinct_clients_mark_the_destination_down() {
+    let backend = banning_backend().await;
+    let mapping = mapping_file("");
+    let toml = config(backend.port, &mapping.path().display().to_string(), 3, 2);
+    let ctx = make_ctx(&toml).await;
+
+    for peer in [
+        "198.51.100.7:40000",
+        "198.51.100.7:40001",
+        "203.0.113.4:40000",
+    ] {
+        let resp = imap_login(&ctx, peer, "user@example.com").await;
+        assert!(find(&resp, b"a1 NO [UNAVAILABLE]").is_some());
+    }
+
+    let health = ctx.dest_health.get("legacy").unwrap().snapshot();
+    assert_eq!(health.failure_sources, 2);
+    assert!(
+        health.down,
+        "failures from distinct clients are evidence the host is unhealthy"
+    );
+
+    let accepted_before = backend.accepted.load(Ordering::Relaxed);
+    backend.serving.store(true, Ordering::Relaxed);
+    let resp = imap_login(&ctx, "192.0.2.9:40000", "alice@example.com").await;
+    assert!(find(&resp, b"a1 NO [UNAVAILABLE]").is_some());
+    assert_eq!(
+        backend.accepted.load(Ordering::Relaxed),
+        accepted_before,
+        "an open breaker must short-circuit before dialing the backend"
+    );
+
+    backend.handle.abort();
+}
+
+#[tokio::test]
+async fn min_clients_zero_restores_counting_by_failures_alone() {
+    let backend = banning_backend().await;
+    let mapping = mapping_file("");
+    let toml = config(backend.port, &mapping.path().display().to_string(), 3, 0);
+    let ctx = make_ctx(&toml).await;
+
+    for _ in 0..3 {
+        let resp = imap_login(&ctx, "198.51.100.7:40000", "banned@example.com").await;
+        assert!(find(&resp, b"a1 NO [UNAVAILABLE]").is_some());
+    }
+
+    let health = ctx.dest_health.get("legacy").unwrap().snapshot();
+    assert_eq!(health.failure_sources, 1);
+    assert!(
+        health.down,
+        "without the breadth requirement a single client still opens the breaker"
+    );
+
+    backend.handle.abort();
+}
+
+#[tokio::test]
+async fn unreachable_host_marks_the_destination_down_from_one_client() {
+    let closed_port = {
+        let l = TcpListener::bind(unused_addr()).await.unwrap();
+        l.local_addr().unwrap().port()
+    };
+    let mapping = mapping_file("");
+    let toml = config(closed_port, &mapping.path().display().to_string(), 3, 2);
+    let ctx = make_ctx(&toml).await;
+
+    for _ in 0..3 {
+        let resp = imap_login(&ctx, "198.51.100.7:40000", "user@example.com").await;
+        assert!(find(&resp, b"a1 NO [UNAVAILABLE]").is_some());
+    }
+
+    let health = ctx.dest_health.get("legacy").unwrap().snapshot();
+    assert!(health.host_level_failure);
+    assert!(
+        health.down,
+        "connect-stage failures are host evidence regardless of how many clients saw them"
+    );
+}

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/mod.rs

@@ -7,6 +7,7 @@
 mod admin;
 mod config_validation;
 mod containers;
+mod dest_health;
 mod forwarding;
 mod harness;
 mod http_proxy;

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