v1.0.1 (closes #2)

CHANGELOG.md

@@ -2,6 +2,20 @@
 
 All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org/).
 
+## [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.1"
 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"] }

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

@@ -237,6 +237,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,
                     })
                 })

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
 }

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};
 
@@ -141,9 +141,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 +180,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 +221,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 +524,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 +539,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 +552,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 +569,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);
             }
@@ -659,6 +732,7 @@ 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,
@@ -685,7 +759,7 @@ where
         return Ok(());
     }
 
-    let (mut backend, mut backend_residual) = match establish(ctx, dest_id, open).await {
+    let (mut backend, mut backend_residual) = match establish(ctx, dest_id, peer, open).await {
         Ok(b) => b,
         Err(e) => {
             tracing::warn!(error = %e, protocol = label, destination = %dest_id, "backend unavailable");
@@ -854,29 +928,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 +1146,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

@@ -89,6 +89,7 @@ pub async fn handle(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,

src/proto/managesieve.rs

@@ -76,6 +76,7 @@ pub async fn handle(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,

src/proto/pop3.rs

@@ -75,6 +75,7 @@ pub async fn handle(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,

src/proto/smtp.rs

@@ -82,6 +82,7 @@ pub async fn handle_submission(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,
@@ -572,7 +573,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/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/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/mod.rs

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