From 91ffeb1482432ce321a13da7808a247c1610a832 Mon Sep 17 00:00:00 2001 From: Maurus Decimus <11444311+mdecimus@users.noreply.github.com> Date: Thu, 10 Sep 2026 21:05:45 +0100 Subject: [PATCH] v1.0.4 --- CHANGELOG.md | 27 ++- Cargo.lock | 221 ++++++++++++------------ Cargo.toml | 2 +- src/http/proxy.rs | 117 +++++++++++-- src/outbound.rs | 4 +- src/proto/common.rs | 97 +++++++++-- src/proto/imap.rs | 262 +++++++++++++++++----------- src/proto/managesieve.rs | 270 +++++++++++++++++++---------- src/proto/pop3.rs | 156 +++++++++-------- src/proto/smtp.rs | 196 ++++++++++++--------- src/tests/http_proxy.rs | 361 ++++++++++++++++++++++++++++++++++++++ src/tests/imap.rs | 172 +++++++++++++++++++ src/tests/pop3.rs | 363 +++++++++++++++++++++++++++++++++++++++ 13 files changed, 1756 insertions(+), 492 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6228e26..4eee501 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,23 @@ All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org/). +## [1.0.4] - 2026-09-10 + +### Added + +### Changed +- HTTP and JMAP connections are pinned to the account that authenticated them. +- HTTP and JMAP routing identifiers are split on `routing.master_user_separators`. +- A backend authentication failure now counts towards `listener.max_auth_attempts`. + +### Fixed +- A backend authentication failure no longer leaves the client bridged to the backend unparsed. +- `POST /connections/kick` now covers HTTP and JMAP sessions, including upgraded WebSocket bridges. +- `POST /connections/delay` now covers HTTP and JMAP logins. +- A rejected SASL response is no longer re-parsed as a command, which charged two authentication failures for one client mistake and could drop a command pipelined behind it. +- An empty routing identifier is no longer treated as an account, so it is neither registered nor used to pin an HTTP connection. +- A backend leg rejected at authentication is now closed with the protocol's logout command instead of having its socket dropped. + ## [1.0.3] - 2026-08-30 ### Added @@ -24,22 +41,18 @@ All notable changes to this project will be documented in this file. This projec ## [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. +- `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). +- 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 -- Initial release of the migration proxy: pre-auth routing and socket bridging - for SMTP, submission, LMTP, IMAP, POP3, ManageSieve and HTTP/HTTPS. +- Initial release of the migration proxy: pre-auth routing and socket bridging for SMTP, submission, LMTP, IMAP, POP3, ManageSieve and HTTP/HTTPS. ### Changed diff --git a/Cargo.lock b/Cargo.lock index 6c92895..0eea623 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -167,7 +167,7 @@ checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" dependencies = [ "proc-macro2", "quote", - "syn 3.0.4", + "syn 3.0.5", ] [[package]] @@ -193,9 +193,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" [[package]] name = "aws-lc-rs" -version = "1.18.0" +version = "1.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" +checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e" dependencies = [ "aws-lc-sys", "untrusted 0.7.1", @@ -204,9 +204,9 @@ dependencies = [ [[package]] name = "aws-lc-sys" -version = "0.44.0" +version = "0.45.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f09fae7be8bb3174e05c6afdb34199e6dc0c7c04ba9fa237b1967adfbde27483" +checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27" dependencies = [ "cc", "cmake", @@ -296,9 +296,9 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" [[package]] name = "bitflags" -version = "2.13.1" +version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" dependencies = [ "serde_core", ] @@ -349,7 +349,7 @@ checksum = "dbe8358268799ebb3e4df23cb9d47f4c72bbc4f5247e2fa6a1bf7b6c0baea220" dependencies = [ "async-stream", "base64 0.22.1", - "bitflags 2.13.1", + "bitflags 2.13.2", "bollard-buildkit-proto", "bollard-stubs", "bytes", @@ -453,9 +453,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" -version = "1.4.4" +version = "1.4.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273" +checksum = "005ec2760ca554fae18df7a11195552ec576cd665632a881bc011d5bb2fd4d80" dependencies = [ "find-msvc-tools", "jobserver", @@ -598,36 +598,36 @@ checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" [[package]] name = "crossbeam-channel" -version = "0.5.16" +version = "0.5.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +checksum = "98b0cc327b5bc766e7fda9c9260cc0fa81b43a8e240440422dff70788e3f9ef1" dependencies = [ "crossbeam-utils", ] [[package]] name = "crossbeam-epoch" -version = "0.9.20" +version = "0.9.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +checksum = "dc74980687109a3b14c72fd458107bf0baa1da1a1a805e178d15501ba9b86d9d" dependencies = [ "crossbeam-utils", ] [[package]] name = "crossbeam-queue" -version = "0.3.13" +version = "0.3.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "803d13fb3b09d88be9f4dbc29062c66b19bf7170867ceb746d2a8689bf6c7a26" +checksum = "03e8bd762f7479489c70ed6c768ddca99d7296857de437a68dcb2a94365b3fae" dependencies = [ "crossbeam-utils", ] [[package]] name = "crossbeam-utils" -version = "0.8.22" +version = "0.8.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" +checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6" [[package]] name = "crypto-common" @@ -669,9 +669,9 @@ dependencies = [ [[package]] name = "darling" -version = "0.23.0" +version = "0.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" +checksum = "ed17f5901b6630b993ca003def43f2f8ef4014fc13b047b57aad617ff32bc2ec" dependencies = [ "darling_core", "darling_macro", @@ -679,26 +679,26 @@ dependencies = [ [[package]] name = "darling_core" -version = "0.23.0" +version = "0.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0" +checksum = "6837e2cf7485aaae18f86181d2f0e9a7ed297a025e220aeabf63fdebd3a2ddff" dependencies = [ "ident_case", "proc-macro2", "quote", "strsim", - "syn 2.0.119", + "syn 3.0.5", ] [[package]] name = "darling_macro" -version = "0.23.0" +version = "0.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" +checksum = "2ac7135c3ef02b2f7833bbeb1be5ba7f966dcde8a87c6b87f65a778d71a02785" dependencies = [ "darling_core", "quote", - "syn 2.0.119", + "syn 3.0.5", ] [[package]] @@ -799,7 +799,7 @@ checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" dependencies = [ "proc-macro2", "quote", - "syn 3.0.4", + "syn 3.0.5", ] [[package]] @@ -905,9 +905,9 @@ dependencies = [ [[package]] name = "find-msvc-tools" -version = "0.1.11" +version = "0.1.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" +checksum = "3e0f1c7c3a72c66fd80abe965175f7523475c0489a87d3ff9d6e8c87d87a9d2d" [[package]] name = "flume" @@ -1014,7 +1014,7 @@ checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" dependencies = [ "proc-macro2", "quote", - "syn 3.0.4", + "syn 3.0.5", ] [[package]] @@ -1120,7 +1120,7 @@ dependencies = [ "futures-core", "futures-sink", "http", - "indexmap 2.14.1", + "indexmap 2.14.2", "slab", "tokio", "tokio-util", @@ -1156,7 +1156,7 @@ version = "0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dd1246c0e5493286aeb2dde35b1f4eb9c4ce00e628641210a5e553fc001a1f26" dependencies = [ - "indexmap 2.14.1", + "indexmap 2.14.2", "proc-macro2", "quote", "syn 2.0.119", @@ -1179,9 +1179,9 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hermit-abi" -version = "0.5.2" +version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +checksum = "e17592d60ebacc7d5e169f4663c5f84f9161cc90328abcfe8456f41e4dfcb284" [[package]] name = "hex" @@ -1273,9 +1273,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hybrid-array" -version = "0.4.14" +version = "0.4.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" +checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" dependencies = [ "typenum", ] @@ -1529,9 +1529,9 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.14.1" +version = "2.14.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "07aa2048142242915a31d35844fb311e0e53fcca590c3a0a40dcf1b841fa09eb" +checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855" dependencies = [ "equivalent", "hashbrown 0.17.1", @@ -1541,9 +1541,9 @@ dependencies = [ [[package]] name = "ipnet" -version = "2.12.1" +version = "2.12.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a756c3fac73139e83f14c2d742155dd2b78d3ee56597b419a0579b7bdd6dd78" +checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" [[package]] name = "itertools" @@ -1696,9 +1696,9 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.104" +version = "0.3.105" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0e0c1080212aad755ea003d18543e8768dd432c48819efd73a7bf1e39b7a5a3a" +checksum = "ce57d20d1ea864ce2ac172ab472d409214f4fd359f0b2a2775abdf522e2af99e" dependencies = [ "cfg-if", "futures-util", @@ -1763,9 +1763,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] name = "mail-parser" -version = "0.11.8" +version = "0.11.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b3a9080c1fb8190e232df37a10aa1b3d6b08be084e537069913f025e0ce86c5" +checksum = "4ec00bda90c6e645a54506c630c2820cd6b1890cfd2b0a169b50f74b2b8c7c86" dependencies = [ "hashify", ] @@ -1837,9 +1837,9 @@ checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" [[package]] name = "mio" -version = "1.2.2" +version = "1.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" +checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" dependencies = [ "libc", "wasi", @@ -2145,9 +2145,9 @@ checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" [[package]] name = "portable-atomic-util" -version = "0.2.7" +version = "0.2.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618" +checksum = "10ab3eb7f3becc3a1cbc4f2c6f20267996cfc1a6467a873763411b136a122715" dependencies = [ "portable-atomic", ] @@ -2219,7 +2219,7 @@ dependencies = [ [[package]] name = "proxy" -version = "1.0.3" +version = "1.0.4" dependencies = [ "ahash", "arc-swap", @@ -2462,9 +2462,9 @@ dependencies = [ [[package]] name = "redis" -version = "1.6.0" +version = "1.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e37a4ca5c6ca42aa3e6df2fd32b987a65d32a4c2159a6f3fe0fd1df306a2658f" +checksum = "2acbc41a996f7652b2ddd9dfd98cc4ff602cfd742ae35382f07f608405ab50ed" dependencies = [ "arc-swap", "arcstr", @@ -2494,7 +2494,7 @@ version = "0.5.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" dependencies = [ - "bitflags 2.13.1", + "bitflags 2.13.2", ] [[package]] @@ -2514,7 +2514,7 @@ checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" dependencies = [ "proc-macro2", "quote", - "syn 3.0.4", + "syn 3.0.5", ] [[package]] @@ -2548,11 +2548,11 @@ checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" [[package]] name = "reqwest" -version = "0.13.4" +version = "0.13.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" +checksum = "16a1cfa75cc186dd73d5818e510e042e40927bccc9c236b061cea97e1eb08029" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "bytes", "futures-core", "futures-util", @@ -2629,7 +2629,7 @@ version = "1.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" dependencies = [ - "bitflags 2.13.1", + "bitflags 2.13.2", "errno", "libc", "linux-raw-sys", @@ -2638,9 +2638,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.43" +version = "0.23.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +checksum = "6725596c3f2c3a0aef021139e145d4eafe314a6623e4680ca83852b2c67ab2ba" dependencies = [ "aws-lc-rs", "once_cell", @@ -2787,7 +2787,7 @@ version = "3.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" dependencies = [ - "bitflags 2.13.1", + "bitflags 2.13.2", "core-foundation", "core-foundation-sys", "libc", @@ -2837,7 +2837,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn 3.0.4", + "syn 3.0.5", ] [[package]] @@ -2861,7 +2861,7 @@ checksum = "8d3b1629de253c70a0508c3899572da79ca359fdab27c7920ff00406df418906" dependencies = [ "proc-macro2", "quote", - "syn 3.0.4", + "syn 3.0.5", ] [[package]] @@ -2887,16 +2887,16 @@ dependencies = [ [[package]] name = "serde_with" -version = "3.22.0" +version = "3.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ee78f1fbe43ac4a0e47aadb3dbd357b69eb0d3793e948624cd03dd2750ab1c0a" +checksum = "935177bb8c0cd8ca1a4e6d1a2ac8988bea69cab4f9d3a31311e012ad27868ea4" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "bs58", "chrono", "hex", "indexmap 1.9.3", - "indexmap 2.14.1", + "indexmap 2.14.2", "jiff", "schemars 0.9.0", "schemars 1.2.2", @@ -2908,14 +2908,14 @@ dependencies = [ [[package]] name = "serde_with_macros" -version = "3.22.0" +version = "3.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8705578779c2b6bd90d84d66eb2e206b708b1a4d7b9f17641b293545bf1c7e46" +checksum = "1d607aa01a3cb0ad757d6fd216136910db3c97b102fe686585689615a02dbcdc" dependencies = [ "darling", "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.5", ] [[package]] @@ -3049,18 +3049,21 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "smallvec" -version = "1.15.2" +version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +checksum = "b9be42f50aa861c555654aa3a37f52f4b1074bacf4e48fe0ef7fa584e80f1f0f" dependencies = [ "serde", ] [[package]] name = "smtp-proto" -version = "0.2.3" +version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e365d0fff6ae5036364ab62f236fa669ce05acbd5878df32d3755d48cfc1491b" +checksum = "707104487221ff447b5b796b5049e5c09cf52ff9fc1c8abba4695b89ac4b0f37" +dependencies = [ + "memchr", +] [[package]] name = "socket2" @@ -3113,7 +3116,7 @@ dependencies = [ "futures-util", "hashbrown 0.16.1", "hashlink", - "indexmap 2.14.1", + "indexmap 2.14.2", "log", "memchr", "percent-encoding", @@ -3174,7 +3177,7 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "90b8020fe17c5f2c245bfa2505d7ef59c5604839527c740266ad2214acebea27" dependencies = [ - "bitflags 2.13.1", + "bitflags 2.13.2", "byteorder", "bytes", "crc", @@ -3202,7 +3205,7 @@ checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e" dependencies = [ "atoi", "base64 0.22.1", - "bitflags 2.13.1", + "bitflags 2.13.2", "byteorder", "crc", "dotenvy", @@ -3318,9 +3321,9 @@ dependencies = [ [[package]] name = "syn" -version = "3.0.4" +version = "3.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6275cddf4610d1775e6d1fe9469b2e77d0f39fd98fb7450901b821e0c53649f" +checksum = "12df2e0110f65b775f769bb17ef989067a1d931b2eb822bd4346631eeada89f9" dependencies = [ "proc-macro2", "quote", @@ -3414,7 +3417,7 @@ checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" dependencies = [ "proc-macro2", "quote", - "syn 3.0.4", + "syn 3.0.5", ] [[package]] @@ -3468,9 +3471,9 @@ dependencies = [ [[package]] name = "tinyvec" -version = "1.12.0" +version = "1.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb4ebadaa0af04fab11ae01eb5f9fdb5f9c5b875506e210e71c07873528baa7f" +checksum = "4cf0ded5c4e56918d8f8a339e1bb67d038d3bc6d144ac407904015ba2e4cde9b" dependencies = [ "tinyvec_macros", ] @@ -3506,14 +3509,14 @@ checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" dependencies = [ "proc-macro2", "quote", - "syn 3.0.4", + "syn 3.0.5", ] [[package]] name = "tokio-rustls" -version = "0.26.4" +version = "0.26.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +checksum = "b0c85f2c3ef0b1cd58b36682f4b17aaa995f0e5db534d85692b4903abce21f67" dependencies = [ "rustls", "tokio", @@ -3562,11 +3565,11 @@ dependencies = [ [[package]] name = "toml" -version = "1.1.4+spec-1.1.0" +version = "1.1.5+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3aace63f4bbcdfc2c965b059de67119c89c4017a70d633be6c104910f67056f5" +checksum = "12c0ba9680044b4ce98d391a62094047eada0d64860b80166c39f4a6b5640785" dependencies = [ - "indexmap 2.14.1", + "indexmap 2.14.2", "serde_core", "serde_spanned", "toml_datetime", @@ -3647,7 +3650,7 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" dependencies = [ "futures-core", "futures-util", - "indexmap 2.14.1", + "indexmap 2.14.2", "pin-project-lite", "slab", "sync_wrapper", @@ -3664,7 +3667,7 @@ version = "0.6.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" dependencies = [ - "bitflags 2.13.1", + "bitflags 2.13.2", "bytes", "futures-util", "http", @@ -3840,9 +3843,9 @@ checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" [[package]] name = "uuid" -version = "1.26.0" +version = "1.26.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812" +checksum = "2ef6dac1e96601b4fb3acccccff2139741fcb757cb9a36089bf5be91cfb285ce" dependencies = [ "getrandom 0.4.3", "js-sys", @@ -3903,9 +3906,9 @@ dependencies = [ [[package]] name = "wasm-bindgen" -version = "0.2.127" +version = "0.2.128" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b70935747edd64d89de3efa29d73789b806c15798f8e7dca4d8ac356b50ce70" +checksum = "aecb87a33d3b0c5e3b7aa46336eaf486cffafbd281b195e4c8b80d50df2351bf" dependencies = [ "cfg-if", "once_cell", @@ -3916,9 +3919,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-futures" -version = "0.4.77" +version = "0.4.78" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b7777d5cc23d0e91404e53ce2d5e8ec7acae3026b16233dba62cd3246457950" +checksum = "6ef4c5d3d2cdf5c54f4231181768f5510842e350db025faf1f7163b1030ed928" dependencies = [ "js-sys", "wasm-bindgen", @@ -3926,9 +3929,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.127" +version = "0.2.128" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77775f8f3f7217702089053b94958f8f54061a3f663417df76e19cbdcca29bc1" +checksum = "a690d511e3c1a8b3a55e33511e3c2c00c78415cd23650f32b808627f5696b9ed" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -3936,22 +3939,22 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.127" +version = "0.2.128" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e11d33f857dc2fb11b8bc75aee111aa9cbeb12cd9f25efd3d4c2a3dd4e235284" +checksum = "411e4887f0071ef2d2164a9d5fdf2d20efbef78fccd3a78b0c10a1dc5295e48a" dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.5", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-shared" -version = "0.2.127" +version = "0.2.128" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ef64dbcc55df09c7e5a46182d181c2cfa3e925f3da937ea764728b4bbb9dcbf" +checksum = "81941cd78d0c92026c33e5e01312845a4cb1e9af3407f9134b100dd03144103e" dependencies = [ "unicode-ident", ] @@ -3971,9 +3974,9 @@ dependencies = [ [[package]] name = "web-sys" -version = "0.3.104" +version = "0.3.105" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c435338968042f4f59a557f690a253676d47ce13ceb55d70100e7facf6620a30" +checksum = "9fbddc4a036f00ec4f18c83445bd3115cb306a91da554919a099d9222fe4a7f8" dependencies = [ "js-sys", "wasm-bindgen", @@ -4272,18 +4275,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.56" +version = "0.8.57" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb" +checksum = "d35102a9f36d089ccae9e4c6802bc118be4487b80aaffc0ab4e0cf5ce92d2873" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.56" +version = "0.8.57" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1" +checksum = "146c01f5ab44258da43cf276c74a2763db2ff3969c9c652c3f2de07041d0b2bc" dependencies = [ "proc-macro2", "quote", @@ -4347,7 +4350,7 @@ checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da" dependencies = [ "proc-macro2", "quote", - "syn 3.0.4", + "syn 3.0.5", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 915c020..63dffcc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "proxy" -version = "1.0.3" +version = "1.0.4" edition = "2024" license = "AGPL-3.0-only OR LicenseRef-SEL" description = "Multi-protocol e-mail proxy" diff --git a/src/http/proxy.rs b/src/http/proxy.rs index a0a8ec8..c07acf0 100644 --- a/src/http/proxy.rs +++ b/src/http/proxy.rs @@ -13,7 +13,7 @@ use tokio::time::timeout; use crate::config::{ForwardedMode, ListenerConfig, Protocol, TlsMode}; use crate::error::{ProxyError, Result}; use crate::net::BoxedStream; -use crate::proto::common::Ctx; +use crate::proto::common::{ConnGuard, Ctx}; use super::head::{BodyFraming, Head, ParseError, ParseOutcome, error_response}; use super::router::{HttpRouter, RouteOutcome}; @@ -379,12 +379,55 @@ fn response_no_body(req_method: &str, head: &Head) -> bool { matches!(head.framing, BodyFraming::None) } +struct Account { + cancel: Option>, + guard: Option, + pinned: Option, +} + +impl Account { + fn new(cancel: tokio::sync::watch::Sender) -> Self { + Account { + cancel: Some(cancel), + guard: None, + pinned: None, + } + } + + fn accepts(&self, ctx: &Ctx, identifier: Option<&str>) -> bool { + match (&self.pinned, identifier) { + (Some(pinned), Some(id)) => *pinned == ctx.router.normalize(id), + _ => true, + } + } + + fn pin(&mut self, ctx: &Ctx, identifier: Option<&str>) -> bool { + if self.pinned.is_some() { + return false; + } + let Some(id) = identifier else { + return false; + }; + let Some(cancel) = self.cancel.take() else { + return false; + }; + self.pinned = Some(ctx.router.normalize(id)); + self.guard = Some(ctx.register_conn_cancel(id, cancel)); + true + } + + fn into_guard(self) -> Option { + self.guard + } +} + enum Disposition { Close, Hijack { backend: BoxedStream, to_backend: Vec, to_client: Vec, + guard: Option, }, } @@ -400,7 +443,16 @@ pub async fn handle( local: SocketAddr, ) -> Result<()> { let bridge_idle = ctx.config.server.bridge_idle; - match serve(ctx, listener, &mut stream, peer, local).await { + let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false); + + let served = tokio::select! { + result = serve(ctx, listener, &mut stream, peer, local, cancel_tx) => result, + _ = crate::outbound::wait_kick(Some(cancel_rx)) => { + return Err(ProxyError::backend("disconnected by admin")); + } + }; + + match served { Ok(Disposition::Close) => { shutdown_client(&mut stream).await; Ok(()) @@ -409,9 +461,18 @@ pub async fn handle( backend, to_backend, to_client, + guard, }) => { - crate::outbound::bridge(stream, backend, &to_backend, &to_client, bridge_idle, None) - .await?; + let cancel = guard.as_ref().map(|g| g.cancel()); + crate::outbound::bridge( + stream, + backend, + &to_backend, + &to_client, + bridge_idle, + cancel, + ) + .await?; Ok(()) } Err(e) => Err(e), @@ -424,6 +485,7 @@ async fn serve( stream: &mut BoxedStream, peer: SocketAddr, local: SocketAddr, + cancel: tokio::sync::watch::Sender, ) -> Result { let router = ctx.http_router.as_ref(); let max_head = router.max_head_size; @@ -459,7 +521,11 @@ async fn serve( let mut first_head = first_head; sanitize_inbound_forwarded(&mut first_head, listener); - let (buffered_body, dest_id) = match prepare( + let Prepared { + body: buffered_body, + dest_id, + identifier, + } = match prepare( ctx, router, stream, @@ -477,6 +543,11 @@ async fn serve( } }; + let mut account = Account::new(cancel); + if account.pin(ctx, identifier.as_deref()) { + crate::proto::common::hold_login(ctx, identifier.as_deref()).await; + } + let (dest, ep, tls_cfg) = match ctx.endpoint(&dest_id, Protocol::Http) { Ok(t) => t, Err(_) => { @@ -552,6 +623,7 @@ async fn serve( backend, to_backend, to_client, + guard: account.into_guard(), }); } Ok(ExchangeResult::Close) => return Ok(Disposition::Close), @@ -576,7 +648,11 @@ async fn serve( let mut next = next; sanitize_inbound_forwarded(&mut next, listener); - let (next_body, next_dest) = match prepare( + let Prepared { + body: next_body, + dest_id: next_dest, + identifier: next_identifier, + } = match prepare( ctx, router, stream, @@ -594,7 +670,7 @@ async fn serve( } }; - if next_dest != dest_id { + if next_dest != dest_id || !account.accepts(ctx, next_identifier.as_deref()) { if let Some(body) = &next_body { drain_buffered(stream, &mut reader, &next, body.len(), idle.body) .await @@ -603,6 +679,9 @@ async fn serve( let _ = write_client(stream, &error_response(421, "Misdirected Request")).await; return Ok(Disposition::Close); } + if account.pin(ctx, next_identifier.as_deref()) { + crate::proto::common::hold_login(ctx, next_identifier.as_deref()).await; + } current_head = next; current_body = next_body; @@ -891,6 +970,12 @@ async fn drain_buffered( Ok(()) } +struct Prepared { + body: Option>, + dest_id: String, + identifier: Option, +} + async fn prepare( ctx: &Ctx, router: &HttpRouter, @@ -899,7 +984,7 @@ async fn prepare( head: &mut Head, body_cap: usize, body_idle: Duration, -) -> Result>, String)>> { +) -> Result> { let idx = match router.match_index(head) { Some(i) => i, None => return Ok(None), @@ -915,21 +1000,27 @@ async fn prepare( None }; - let dest_id = match router.route_at(idx, head, body.as_deref()) { - RouteOutcome::Pin { dest_id } => dest_id, + let (dest_id, identifier) = match router.route_at(idx, head, body.as_deref()) { + RouteOutcome::Pin { dest_id } => (dest_id, None), RouteOutcome::Identifier { identifier, fallback, } => { - if identifier.is_none() && fallback.is_none() { + let identifier = ctx.routing_identifier(identifier); + let dest_id = if identifier.is_none() && fallback.is_none() { ctx.resolve(None).await.to_string() } else { ctx.resolve(identifier.as_deref()).await.to_string() - } + }; + (dest_id, identifier) } }; - Ok(Some((body, dest_id))) + Ok(Some(Prepared { + body, + dest_id, + identifier, + })) } fn reason(status: u16) -> &'static str { diff --git a/src/outbound.rs b/src/outbound.rs index 0987288..bb19ead 100644 --- a/src/outbound.rs +++ b/src/outbound.rs @@ -190,7 +190,7 @@ pub async fn write_all(stream: &mut BoxedStream, bytes: &[u8]) -> Result<()> { const BRIDGE_BUF: usize = 16 * 1024; -const SHUTDOWN_GRACE: Duration = Duration::from_secs(3); +pub(crate) const SHUTDOWN_GRACE: Duration = Duration::from_secs(3); pub async fn bridge( mut client: BoxedStream, @@ -245,7 +245,7 @@ pub async fn bridge( } } -async fn wait_kick(cancel: Option>) { +pub(crate) async fn wait_kick(cancel: Option>) { match cancel { Some(mut rx) => { while rx.changed().await.is_ok() { diff --git a/src/proto/common.rs b/src/proto/common.rs index c5e5fac..a8f2337 100644 --- a/src/proto/common.rs +++ b/src/proto/common.rs @@ -33,6 +33,22 @@ pub struct Handoff { pub residual: Vec, } +pub struct Resume { + pub stream: BoxedStream, + pub buf: Vec, +} + +pub fn take_rest(it: &mut std::slice::Iter<'_, u8>) -> Vec { + let rest = it.as_slice().to_vec(); + *it = Default::default(); + rest +} + +pub fn bump(failures: &mut u32, max: u32) -> bool { + *failures += 1; + *failures >= max +} + pub fn credential_from_sasl(mech: &Mechanism, raw: &[u8]) -> Option { match mech { Mechanism::Plain => Credential::decode_sasl_challenge_plain(raw), @@ -326,9 +342,27 @@ impl ConnRegistry { } pub fn register(&self, identifier: &str) -> ConnGuard { + let (cancel, rx) = tokio::sync::watch::channel(false); + self.insert(identifier, cancel, rx) + } + + pub fn register_cancel( + &self, + identifier: &str, + cancel: tokio::sync::watch::Sender, + ) -> ConnGuard { + let rx = cancel.subscribe(); + self.insert(identifier, cancel, rx) + } + + fn insert( + &self, + identifier: &str, + cancel: tokio::sync::watch::Sender, + rx: tokio::sync::watch::Receiver, + ) -> ConnGuard { static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); let id = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - let (cancel, rx) = tokio::sync::watch::channel(false); let key: Arc = Arc::from(identifier); let shard_idx = self.shard_idx(&key); self.shards[shard_idx] @@ -513,11 +547,21 @@ impl Ctx { self.conns.register(&self.router.normalize(identifier)) } + pub fn register_conn_cancel( + &self, + identifier: &str, + cancel: tokio::sync::watch::Sender, + ) -> ConnGuard { + self.conns + .register_cancel(&self.router.normalize(identifier), cancel) + } + pub fn routing_identifier(&self, identifier: Option) -> Option { - identifier.map(|id| { - crate::route::master_account(&id, &self.config.routing.master_user_separators) - .to_string() - }) + let mut id = identifier?; + let account = + crate::route::master_account(&id, &self.config.routing.master_user_separators); + id.truncate(account.len()); + (!id.is_empty()).then_some(id) } pub fn hide_auth_errors(&self, dest_id: &str) -> bool { @@ -679,6 +723,16 @@ pub enum AuthWire { ManageSieve, } +impl AuthWire { + fn logout(&self) -> &'static [u8] { + match self { + AuthWire::Imap { .. } => b"ZZ LOGOUT\r\n", + AuthWire::Pop3 | AuthWire::Smtp => b"QUIT\r\n", + AuthWire::ManageSieve => b"LOGOUT\r\n", + } + } +} + enum AuthLine { Continuation, Passthrough, @@ -810,7 +864,7 @@ pub async fn bridge_authenticated( mut stream: BoxedStream, peer: SocketAddr, identifier: Option<&str>, - client_residual: &[u8], + client_residual: Vec, started: Instant, label: &'static str, wire: AuthWire, @@ -818,7 +872,7 @@ pub async fn bridge_authenticated( unavailable: &[u8], max: usize, open: F, -) -> Result<()> +) -> Result> where F: FnMut() -> Fut, Fut: Future>, @@ -832,7 +886,7 @@ where "rejecting login: routing identifier contains invalid characters" ); let _ = write(&mut stream, unavailable).await; - return Ok(()); + return Ok(None); } let BackendLeg { @@ -844,10 +898,11 @@ where Err(e) => { tracing::warn!(error = %e, protocol = label, destination = %dest_id, "backend unavailable"); let _ = write(&mut stream, unavailable).await; - return Ok(()); + return Ok(None); } }; + let logout = wire.logout(); let (auth_frame, pending_response) = auth.wire(accepts_initial_response); crate::outbound::write_all(&mut backend, auth_frame).await?; @@ -872,16 +927,26 @@ where } }; - let guard = if auth_ok { - identifier.map(|id| ctx.register_conn(id)) - } else { - None - }; + if !auth_ok { + tracing::debug!( + protocol = label, + destination = %dest_id, + "backend rejected authentication; resuming pre-authentication dialogue" + ); + let _ = crate::outbound::write_all(&mut backend, logout).await; + let _ = tokio::time::timeout(crate::outbound::SHUTDOWN_GRACE, backend.shutdown()).await; + return Ok(Some(Resume { + stream, + buf: client_residual, + })); + } + + let guard = identifier.map(|id| ctx.register_conn(id)); let cancel = guard.as_ref().map(|g| g.cancel()); let (to_backend, to_client) = crate::outbound::bridge( stream, backend, - client_residual, + &client_residual, &backend_residual, ctx.config.server.bridge_idle, cancel, @@ -895,7 +960,7 @@ where secs = started.elapsed().as_secs_f64(), "session closed" ); - Ok(()) + Ok(None) } pub async fn write(stream: &mut BoxedStream, bytes: &[u8]) -> Result<()> { diff --git a/src/proto/imap.rs b/src/proto/imap.rs index 6c75c96..1a43de5 100644 --- a/src/proto/imap.rs +++ b/src/proto/imap.rs @@ -39,72 +39,100 @@ enum Decision { pub async fn handle( ctx: &Ctx, listener: &ListenerConfig, - stream: BoxedStream, + mut stream: BoxedStream, peer: SocketAddr, local: SocketAddr, ) -> Result<()> { let caps = &ctx.config.capabilities.imap; + let starttls_listener = listener.tls == TlsMode::Starttls; - let decision = match timeout( - listener.preauth_timeout, - preauth(ctx, listener, caps, stream), - ) - .await - { - Ok(Ok(d)) => d, - Ok(Err(e)) => return Err(e), - Err(_) => return Ok(()), - }; - - let Decision::Authenticate { - stream, - mech, - tag, - raw_challenge, - identifier, - residual, - } = decision - else { - return Ok(()); - }; - - let identifier = ctx.routing_identifier(identifier); - let dest_id = ctx.resolve(identifier.as_deref()).await; - tracing::debug!(destination = %dest_id, "imap routing decision"); - - let started = std::time::Instant::now(); - common::hold_login(ctx, identifier.as_deref()).await; - - let client_tls = stream.is_tls(); - let fwd = forward::ForwardInfo::new(peer, local, client_tls, None, ctx.config.server.proxy_ttl); - - 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 is_tls = stream.is_tls(); + let greeting = format!( + "* OK [CAPABILITY {}] {}\r\n", + ctx.greetings.imap.get(is_tls, starttls_listener && !is_tls), + caps.banner ); - let unavailable = format!("{tag} NO [UNAVAILABLE] backend unavailable\r\n"); + write(&mut stream, greeting.as_bytes()).await?; - common::bridge_authenticated( - ctx, - &dest_id, - stream, - peer, - identifier.as_deref(), - &residual, - started, - "imap", - common::AuthWire::Imap { tag }, - &auth, - unavailable.as_bytes(), - MAX_PREAUTH, - || open_backend(ctx, &dest_id, &fwd), - ) - .await + let mut seed = Vec::new(); + let mut auth_failures = 0u32; + + loop { + let decision = match timeout( + listener.preauth_timeout, + preauth(ctx, listener, caps, stream, seed, &mut auth_failures), + ) + .await + { + Ok(Ok(d)) => d, + Ok(Err(e)) => return Err(e), + Err(_) => return Ok(()), + }; + + let Decision::Authenticate { + stream: client, + mech, + tag, + raw_challenge, + identifier, + residual, + } = decision + else { + return Ok(()); + }; + + let identifier = ctx.routing_identifier(identifier); + let dest_id = ctx.resolve(identifier.as_deref()).await; + tracing::debug!(destination = %dest_id, "imap routing decision"); + + let started = std::time::Instant::now(); + common::hold_login(ctx, identifier.as_deref()).await; + + let client_tls = client.is_tls(); + let fwd = + forward::ForwardInfo::new(peer, local, client_tls, None, ctx.config.server.proxy_ttl); + + 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"); + + let resume = common::bridge_authenticated( + ctx, + &dest_id, + client, + peer, + identifier.as_deref(), + residual, + started, + "imap", + common::AuthWire::Imap { tag }, + &auth, + unavailable.as_bytes(), + MAX_PREAUTH, + || open_backend(ctx, &dest_id, &fwd), + ) + .await?; + + let Some(common::Resume { + stream: client, + buf, + }) = resume + else { + return Ok(()); + }; + if common::bump(&mut auth_failures, listener.max_auth_attempts) { + return Ok(()); + } + stream = client; + seed = buf; + } } async fn preauth( @@ -112,6 +140,8 @@ async fn preauth( listener: &ListenerConfig, caps: &ImapCapabilities, mut stream: BoxedStream, + seed: Vec, + auth_failures: &mut u32, ) -> Result { let starttls_listener = listener.tls == TlsMode::Starttls; let mut auth = Auth { @@ -119,31 +149,29 @@ async fn preauth( caps, is_tls: stream.is_tls(), max_attempts: listener.max_auth_attempts, - auth_failures: 0, + auth_failures, }; - let greeting = format!( - "* OK [CAPABILITY {}] {}\r\n", - ctx.greetings - .imap - .get(auth.is_tls, starttls_listener && !auth.is_tls), - caps.banner - ); - write(&mut stream, greeting.as_bytes()).await?; - let mut receiver = Receiver::::new(); receiver.max_request_size = MAX_PREAUTH; let mut read = [0u8; 8192]; + let mut pending = seed; loop { - let n = stream - .read(&mut read) - .await - .map_err(|e| ProxyError::Protocol(format!("reading client: {e}").into()))?; - if n == 0 { - return Ok(Decision::Closed); - } - let mut it = read[..n].iter(); + let chunk: &[u8] = if pending.is_empty() { + let n = stream + .read(&mut read) + .await + .map_err(|e| ProxyError::Protocol(format!("reading client: {e}").into()))?; + if n == 0 { + return Ok(Decision::Closed); + } + &read[..n] + } else { + &pending + }; + let mut it = chunk.iter(); + let mut carry: Option> = None; loop { match receiver.parse(&mut it) { @@ -234,6 +262,10 @@ async fn preauth( residual, }); } + Step::Resume { residual } => { + carry = Some(residual); + break; + } Step::Continue => {} Step::Close => return Ok(Decision::Closed), } @@ -255,6 +287,10 @@ async fn preauth( residual, }); } + Step::Resume { residual } => { + carry = Some(residual); + break; + } Step::Continue => {} Step::Close => return Ok(Decision::Closed), } @@ -280,13 +316,17 @@ async fn preauth( } Err(RecvError::Parse { message }) => { write(&mut stream, format!("* BAD {message}\r\n").as_bytes()).await?; - auth.auth_failures += 1; - if auth.auth_failures >= auth.max_attempts { + if common::bump(auth.auth_failures, auth.max_attempts) { return Ok(Decision::Closed); } } } } + drop(it); + match carry { + Some(rest) => pending = rest, + None => pending.clear(), + } } } @@ -295,11 +335,14 @@ struct Auth<'a> { caps: &'a ImapCapabilities, is_tls: bool, max_attempts: u32, - auth_failures: u32, + auth_failures: &'a mut u32, } enum Step { Continue, + Resume { + residual: Vec, + }, Close, Finalize { mech: Mechanism, @@ -332,7 +375,7 @@ impl Auth<'_> { Mechanism::Login if common::sasl_offered(&self.caps.sasl, "LOGIN") => {} _ => { return self - .reject(stream, tag, "Unsupported authentication mechanism.") + .reject(stream, tag, "Unsupported authentication mechanism.", None) .await; } } @@ -354,16 +397,21 @@ impl Auth<'_> { } let (raw_b64, residual): (Vec, Vec) = if let Some(ir) = tokens.get(1) { - (ir.as_bytes().to_vec(), it.as_slice().to_vec()) + (ir.as_bytes().to_vec(), common::take_rest(it)) } else { write(&mut *stream, b"+ \r\n").await?; - let seed = it.as_slice().to_vec(); + let seed = common::take_rest(it); read_crlf_line(stream, seed, MAX_PREAUTH).await? }; let Some(raw) = decode_sasl_ir(&raw_b64) else { return self - .reject(stream, tag, "Invalid base64 in SASL response.") + .reject( + stream, + tag, + "Invalid base64 in SASL response.", + Some(residual), + ) .await; }; @@ -378,18 +426,20 @@ impl Auth<'_> { it: &mut std::slice::Iter<'_, u8>, ) -> Result { let (user_b64, residual): (Vec, Vec) = if let Some(ir) = tokens.get(1) { - (ir.as_bytes().to_vec(), it.as_slice().to_vec()) + (ir.as_bytes().to_vec(), common::take_rest(it)) } else { write( &mut *stream, format!("+ {}\r\n", common::LOGIN_USERNAME_CHALLENGE).as_bytes(), ) .await?; - let seed = it.as_slice().to_vec(); + let seed = common::take_rest(it); read_crlf_line(stream, seed, MAX_PREAUTH).await? }; if common::sasl_cancelled(&user_b64) { - return self.reject(stream, tag, "Authentication cancelled.").await; + return self + .reject(stream, tag, "Authentication cancelled.", Some(residual)) + .await; } write( @@ -399,12 +449,19 @@ impl Auth<'_> { .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; + return self + .reject(stream, tag, "Authentication cancelled.", Some(residual)) + .await; } let Some(raw) = common::login_challenge(&user_b64, &pass_b64) else { return self - .reject(stream, tag, "Invalid base64 in SASL response.") + .reject( + stream, + tag, + "Invalid base64 in SASL response.", + Some(residual), + ) .await; }; @@ -429,7 +486,7 @@ impl Auth<'_> { } let (Some(user), Some(pass)) = (tokens.first(), tokens.get(1)) else { return self - .reject(stream, tag, "LOGIN requires a username and password.") + .reject(stream, tag, "LOGIN requires a username and password.", None) .await; }; let (user, pass) = (user.as_bytes(), pass.as_bytes()); @@ -438,7 +495,7 @@ impl Auth<'_> { raw.extend_from_slice(user); raw.push(0); raw.extend_from_slice(pass); - let residual = it.as_slice().to_vec(); + let residual = common::take_rest(it); self.finish(stream, Mechanism::Plain, tag, raw, residual) .await @@ -453,7 +510,9 @@ impl Auth<'_> { residual: Vec, ) -> Result { let Some(cred) = common::credential_from_sasl(&mech, &raw) else { - return self.reject(stream, tag, "Malformed credentials.").await; + return self + .reject(stream, tag, "Malformed credentials.", Some(residual)) + .await; }; let identifier = identifier_from_credential(&cred, &self.ctx.config.oauth.jwt_username_claim); @@ -466,14 +525,21 @@ impl Auth<'_> { }) } - async fn reject(&mut self, stream: &mut BoxedStream, tag: &str, message: &str) -> Result { + async fn reject( + &mut self, + stream: &mut BoxedStream, + tag: &str, + message: &str, + residual: Option>, + ) -> Result { write(&mut *stream, format!("{tag} NO {message}\r\n").as_bytes()).await?; - self.auth_failures += 1; - if self.auth_failures >= self.max_attempts { - Ok(Step::Close) - } else { - Ok(Step::Continue) + if common::bump(self.auth_failures, self.max_attempts) { + return Ok(Step::Close); } + Ok(match residual { + Some(residual) => Step::Resume { residual }, + None => Step::Continue, + }) } } diff --git a/src/proto/managesieve.rs b/src/proto/managesieve.rs index db12380..5024f6f 100644 --- a/src/proto/managesieve.rs +++ b/src/proto/managesieve.rs @@ -16,7 +16,7 @@ use crate::imap_receiver::{Command, Error as RecvError, Mechanism, Receiver, Sta use crate::net::BoxedStream; use crate::outbound; use crate::proto::common::{ - self, Ctx, Handoff, assert_egress_encrypted, credential_from_sasl, mechanism_name, + self, Ctx, Handoff, assert_egress_encrypted, bump, credential_from_sasl, mechanism_name, plain_auth_allowed, write, }; use crate::proto::forward; @@ -28,80 +28,14 @@ const MAX_PREAUTH: usize = 64 * 1024; pub async fn handle( ctx: &Ctx, listener: &ListenerConfig, - stream: BoxedStream, + mut stream: BoxedStream, peer: SocketAddr, local: SocketAddr, ) -> Result<()> { let caps = &ctx.config.capabilities.managesieve; - - let handoff = match timeout( - listener.preauth_timeout, - preauth(ctx, listener, caps, stream), - ) - .await - { - Ok(Ok(Some(h))) => h, - Ok(Ok(None)) => return Ok(()), - Ok(Err(e)) => return Err(e), - Err(_) => return Ok(()), - }; - - let Handoff { - stream, - mech, - raw_challenge, - identifier, - residual, - } = handoff; - - let identifier = ctx.routing_identifier(identifier); - let dest_id = ctx.resolve(identifier.as_deref()).await; - let started = std::time::Instant::now(); - common::hold_login(ctx, identifier.as_deref()).await; - let fwd = forward::ForwardInfo::new( - peer, - local, - stream.is_tls(), - None, - ctx.config.server.proxy_ttl, - ); - - 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, - b"NO \"backend unavailable\"\r\n", - MAX_PREAUTH, - || open_backend(ctx, &dest_id, &fwd), - ) - .await -} - -async fn preauth( - ctx: &Ctx, - listener: &ListenerConfig, - caps: &ManageSieveCapabilities, - mut stream: BoxedStream, -) -> Result> { - let mut is_tls = stream.is_tls(); let starttls_listener = listener.tls == TlsMode::Starttls; + let is_tls = stream.is_tls(); write( &mut stream, ctx.greetings @@ -111,21 +45,115 @@ async fn preauth( ) .await?; + let mut seed = Vec::new(); + let mut auth_failures = 0u32; + + loop { + let handoff = match timeout( + listener.preauth_timeout, + preauth(ctx, listener, caps, stream, seed, &mut auth_failures), + ) + .await + { + Ok(Ok(Some(h))) => h, + Ok(Ok(None)) => return Ok(()), + Ok(Err(e)) => return Err(e), + Err(_) => return Ok(()), + }; + + let Handoff { + stream: client, + mech, + raw_challenge, + identifier, + residual, + } = handoff; + + let identifier = ctx.routing_identifier(identifier); + let dest_id = ctx.resolve(identifier.as_deref()).await; + let started = std::time::Instant::now(); + common::hold_login(ctx, identifier.as_deref()).await; + let fwd = forward::ForwardInfo::new( + peer, + local, + client.is_tls(), + None, + ctx.config.server.proxy_ttl, + ); + + let auth = common::AuthFrame::inline( + format!( + "AUTHENTICATE \"{}\" \"{}\"\r\n", + mechanism_name(&mech), + BASE64.encode(&raw_challenge) + ) + .into_bytes(), + ); + + let resume = common::bridge_authenticated( + ctx, + &dest_id, + client, + peer, + identifier.as_deref(), + residual, + started, + "managesieve", + common::AuthWire::ManageSieve, + &auth, + b"NO \"backend unavailable\"\r\n", + MAX_PREAUTH, + || open_backend(ctx, &dest_id, &fwd), + ) + .await?; + + let Some(common::Resume { + stream: client, + buf, + }) = resume + else { + return Ok(()); + }; + if bump(&mut auth_failures, listener.max_auth_attempts) { + return Ok(()); + } + stream = client; + seed = buf; + } +} + +async fn preauth( + ctx: &Ctx, + listener: &ListenerConfig, + caps: &ManageSieveCapabilities, + mut stream: BoxedStream, + seed: Vec, + auth_failures: &mut u32, +) -> Result> { + let mut is_tls = stream.is_tls(); + let starttls_listener = listener.tls == TlsMode::Starttls; + let mut receiver = Receiver::::new().with_start_state(State::Command { is_uid: false }); receiver.max_request_size = MAX_PREAUTH; - let mut auth_failures = 0u32; let mut read = [0u8; 8192]; + let mut pending = seed; loop { - let n = stream - .read(&mut read) - .await - .map_err(|e| ProxyError::Protocol(format!("reading client: {e}").into()))?; - if n == 0 { - return Ok(None); - } - let mut it = read[..n].iter(); + let chunk: &[u8] = if pending.is_empty() { + let n = stream + .read(&mut read) + .await + .map_err(|e| ProxyError::Protocol(format!("reading client: {e}").into()))?; + if n == 0 { + return Ok(None); + } + &read[..n] + } else { + &pending + }; + let mut it = chunk.iter(); + let mut carry: Option> = None; loop { match receiver.parse(&mut it) { @@ -179,7 +207,7 @@ async fn preauth( common::sasl_offered(&caps.sasl, "LOGIN"), &req.tokens, &mut it, - &mut auth_failures, + &mut *auth_failures, listener.max_auth_attempts, ) .await? @@ -198,6 +226,10 @@ async fn preauth( residual, })); } + AuthStep::Resume { residual } => { + carry = Some(residual); + break; + } AuthStep::Continue => {} AuthStep::Close => return Ok(None), } @@ -222,13 +254,17 @@ async fn preauth( format!("NO \"{}\"\r\n", sanitize(&message)).as_bytes(), ) .await?; - auth_failures += 1; - if auth_failures >= listener.max_auth_attempts { + if bump(auth_failures, listener.max_auth_attempts) { return Ok(None); } } } } + drop(it); + match carry { + Some(rest) => pending = rest, + None => pending.clear(), + } } } @@ -240,6 +276,9 @@ enum AuthStep { residual: Vec, }, Continue, + Resume { + residual: Vec, + }, Close, } @@ -262,7 +301,16 @@ async fn handle_authenticate( match mech { Mechanism::Plain | Mechanism::OAuthBearer | Mechanism::XOauth2 => {} Mechanism::Login if login_offered => {} - _ => return fail(stream, "Unsupported mechanism", auth_failures, max_attempts).await, + _ => { + return fail( + stream, + "Unsupported mechanism", + auth_failures, + max_attempts, + None, + ) + .await; + } } if matches!(mech, Mechanism::Plain | Mechanism::Login) && !plain_ok { write( @@ -278,18 +326,32 @@ async fn handle_authenticate( } let (raw_b64, residual): (Vec, Vec) = if let Some(ir) = tokens.get(1) { - (ir.as_bytes().to_vec(), it.as_slice().to_vec()) + (ir.as_bytes().to_vec(), common::take_rest(it)) } else { write(&mut *stream, b"{0}\r\n").await?; - let seed = it.as_slice().to_vec(); + let seed = common::take_rest(it); read_sasl_response(stream, seed).await? }; let Some(raw) = decode_sasl_ir(&raw_b64) else { - return fail(stream, "Invalid base64", auth_failures, max_attempts).await; + return fail( + stream, + "Invalid base64", + auth_failures, + max_attempts, + Some(residual), + ) + .await; }; let Some(cred) = credential_from_sasl(&mech, &raw) else { - return fail(stream, "Malformed credentials", auth_failures, max_attempts).await; + return fail( + stream, + "Malformed credentials", + auth_failures, + max_attempts, + Some(residual), + ) + .await; }; let identifier = identifier_from_credential(&cred, &ctx.config.oauth.jwt_username_claim); @@ -310,10 +372,10 @@ async fn authenticate_login( max_attempts: u32, ) -> Result { let (user_b64, residual): (Vec, Vec) = if let Some(ir) = tokens.get(1) { - (ir.as_bytes().to_vec(), it.as_slice().to_vec()) + (ir.as_bytes().to_vec(), common::take_rest(it)) } else { write_challenge(stream, common::LOGIN_USERNAME_CHALLENGE).await?; - let seed = it.as_slice().to_vec(); + let seed = common::take_rest(it); read_sasl_response(stream, seed).await? }; if common::sasl_cancelled(&user_b64) { @@ -322,6 +384,7 @@ async fn authenticate_login( "Authentication cancelled", auth_failures, max_attempts, + Some(residual), ) .await; } @@ -334,15 +397,30 @@ async fn authenticate_login( "Authentication cancelled", auth_failures, max_attempts, + Some(residual), ) .await; } let Some(raw) = common::login_challenge(&user_b64, &pass_b64) else { - return fail(stream, "Invalid base64", auth_failures, max_attempts).await; + return fail( + stream, + "Invalid base64", + auth_failures, + max_attempts, + Some(residual), + ) + .await; }; let Some(cred) = credential_from_sasl(&Mechanism::Plain, &raw) else { - return fail(stream, "Malformed credentials", auth_failures, max_attempts).await; + return fail( + stream, + "Malformed credentials", + auth_failures, + max_attempts, + Some(residual), + ) + .await; }; let identifier = identifier_from_credential(&cred, &ctx.config.oauth.jwt_username_claim); @@ -367,14 +445,16 @@ async fn fail( message: &str, auth_failures: &mut u32, max_attempts: u32, + residual: Option>, ) -> Result { write(&mut *stream, format!("NO \"{message}\"\r\n").as_bytes()).await?; - *auth_failures += 1; - if *auth_failures >= max_attempts { - Ok(AuthStep::Close) - } else { - Ok(AuthStep::Continue) + if bump(auth_failures, max_attempts) { + return Ok(AuthStep::Close); } + Ok(match residual { + Some(residual) => AuthStep::Resume { residual }, + None => AuthStep::Continue, + }) } async fn read_sasl_response(stream: &mut BoxedStream, seed: Vec) -> Result<(Vec, Vec)> { diff --git a/src/proto/pop3.rs b/src/proto/pop3.rs index 1f9ed20..5c925c3 100644 --- a/src/proto/pop3.rs +++ b/src/proto/pop3.rs @@ -15,7 +15,7 @@ use crate::imap_receiver::Mechanism; use crate::net::BoxedStream; use crate::outbound; use crate::proto::common::{ - self, Ctx, Handoff, assert_egress_encrypted, credential_from_sasl, mechanism_name, + self, Ctx, Handoff, assert_egress_encrypted, bump, credential_from_sasl, mechanism_name, plain_auth_allowed, write, }; use crate::proto::forward; @@ -27,69 +27,89 @@ const MAX_PREAUTH: usize = 64 * 1024; pub async fn handle( ctx: &Ctx, listener: &ListenerConfig, - stream: BoxedStream, + mut stream: BoxedStream, peer: SocketAddr, local: SocketAddr, ) -> Result<()> { let caps = &ctx.config.capabilities.pop3; - let handoff = match timeout( - listener.preauth_timeout, - preauth(ctx, listener, caps, stream), - ) - .await - { - Ok(Ok(Some(h))) => h, - Ok(Ok(None)) => return Ok(()), - Ok(Err(e)) => return Err(e), - Err(_) => return Ok(()), - }; + write(&mut stream, format!("+OK {}\r\n", caps.banner).as_bytes()).await?; - let Handoff { - stream, - mech, - raw_challenge, - identifier, - residual, - } = handoff; + let mut seed = Vec::with_capacity(1024); + let mut auth_failures = 0u32; - let identifier = ctx.routing_identifier(identifier); - let dest_id = ctx.resolve(identifier.as_deref()).await; - let started = std::time::Instant::now(); - common::hold_login(ctx, identifier.as_deref()).await; - let fwd = forward::ForwardInfo::new( - peer, - local, - stream.is_tls(), - None, - ctx.config.server.proxy_ttl, - ); - - let auth = common::AuthFrame::inline( - format!( - "AUTH {} {}\r\n", - mechanism_name(&mech), - BASE64.encode(&raw_challenge) + loop { + let handoff = match timeout( + listener.preauth_timeout, + preauth(ctx, listener, caps, stream, seed, &mut auth_failures), ) - .into_bytes(), - ); + .await + { + Ok(Ok(Some(h))) => h, + Ok(Ok(None)) => return Ok(()), + Ok(Err(e)) => return Err(e), + Err(_) => return Ok(()), + }; - common::bridge_authenticated( - ctx, - &dest_id, - stream, - peer, - identifier.as_deref(), - &residual, - started, - "pop3", - common::AuthWire::Pop3, - &auth, - b"-ERR backend unavailable\r\n", - MAX_PREAUTH, - || open_backend(ctx, &dest_id, &fwd), - ) - .await + let Handoff { + stream: client, + mech, + raw_challenge, + identifier, + residual, + } = handoff; + + let identifier = ctx.routing_identifier(identifier); + let dest_id = ctx.resolve(identifier.as_deref()).await; + let started = std::time::Instant::now(); + common::hold_login(ctx, identifier.as_deref()).await; + let fwd = forward::ForwardInfo::new( + peer, + local, + client.is_tls(), + None, + ctx.config.server.proxy_ttl, + ); + + let auth = common::AuthFrame::inline( + format!( + "AUTH {} {}\r\n", + mechanism_name(&mech), + BASE64.encode(&raw_challenge) + ) + .into_bytes(), + ); + + let resume = common::bridge_authenticated( + ctx, + &dest_id, + client, + peer, + identifier.as_deref(), + residual, + started, + "pop3", + common::AuthWire::Pop3, + &auth, + b"-ERR backend unavailable\r\n", + MAX_PREAUTH, + || open_backend(ctx, &dest_id, &fwd), + ) + .await?; + + let Some(common::Resume { + stream: client, + buf, + }) = resume + else { + return Ok(()); + }; + if bump(&mut auth_failures, listener.max_auth_attempts) { + return Ok(()); + } + stream = client; + seed = buf; + } } async fn preauth( @@ -97,15 +117,14 @@ async fn preauth( listener: &ListenerConfig, caps: &Pop3Capabilities, mut stream: BoxedStream, + seed: Vec, + auth_failures: &mut u32, ) -> Result> { let mut is_tls = stream.is_tls(); let starttls_listener = listener.tls == TlsMode::Starttls; - write(&mut stream, format!("+OK {}\r\n", caps.banner).as_bytes()).await?; - - let mut buf: Vec = Vec::with_capacity(1024); + let mut buf: Vec = seed; let mut pending_user: Option> = None; - let mut auth_failures = 0u32; loop { let Some(nl) = buf.iter().position(|&b| b == b'\n') else { @@ -227,7 +246,7 @@ async fn preauth( _ => { write(&mut stream, b"-ERR unsupported mechanism\r\n").await?; buf.drain(..consumed); - if bump(&mut auth_failures, listener.max_auth_attempts) { + if bump(auth_failures, listener.max_auth_attempts) { return Ok(None); } continue; @@ -259,7 +278,7 @@ async fn preauth( 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) { + if bump(auth_failures, listener.max_auth_attempts) { return Ok(None); } continue; @@ -274,21 +293,21 @@ async fn preauth( 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) { + if bump(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) { + if bump(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) { + if bump(auth_failures, listener.max_auth_attempts) { return Ok(None); } continue; @@ -321,14 +340,14 @@ async fn preauth( let Some(raw) = decode_sasl_ir(&raw_b64) else { write(&mut stream, b"-ERR invalid base64\r\n").await?; - if bump(&mut auth_failures, listener.max_auth_attempts) { + if bump(auth_failures, listener.max_auth_attempts) { return Ok(None); } continue; }; let Some(cred) = credential_from_sasl(&mech, &raw) else { write(&mut stream, b"-ERR malformed credentials\r\n").await?; - if bump(&mut auth_failures, listener.max_auth_attempts) { + if bump(auth_failures, listener.max_auth_attempts) { return Ok(None); } continue; @@ -352,11 +371,6 @@ async fn preauth( } } -fn bump(failures: &mut u32, max: u32) -> bool { - *failures += 1; - *failures >= max -} - fn split_verb(line: &[u8]) -> (&[u8], &[u8]) { match line.iter().position(|&b| b == b' ') { Some(pos) => (&line[..pos], trim_leading_space(&line[pos + 1..])), diff --git a/src/proto/smtp.rs b/src/proto/smtp.rs index c12b2a3..f9a5299 100644 --- a/src/proto/smtp.rs +++ b/src/proto/smtp.rs @@ -18,7 +18,7 @@ use crate::imap_receiver::Mechanism; use crate::net::BoxedStream; use crate::outbound; use crate::proto::common::{ - self, Ctx, Handoff, assert_egress_encrypted, credential_from_sasl, mechanism_name, + self, Ctx, Handoff, assert_egress_encrypted, bump, credential_from_sasl, mechanism_name, plain_auth_allowed, write, }; use crate::proto::forward; @@ -32,73 +32,116 @@ struct SubmissionHandoff { ehlo_host: String, } +struct SubmissionState { + buf: Vec, + auth_failures: u32, + ehlo_host: String, +} + +impl Default for SubmissionState { + fn default() -> Self { + SubmissionState { + buf: Vec::with_capacity(1024), + auth_failures: 0, + ehlo_host: String::new(), + } + } +} + pub async fn handle_submission( ctx: &Ctx, listener: &ListenerConfig, - stream: BoxedStream, + mut stream: BoxedStream, peer: SocketAddr, local: SocketAddr, ) -> Result<()> { let caps = &ctx.config.capabilities.submission; - let sh = match timeout( - listener.preauth_timeout, - preauth(ctx, listener, caps, stream, local), + let hostname = ctx.config.server.hostname_for(local); + write( + &mut stream, + format!("220 {hostname} ESMTP {}\r\n", caps.banner).as_bytes(), ) - .await - { - Ok(Ok(Some(h))) => h, - Ok(Ok(None)) => return Ok(()), - Ok(Err(e)) => return Err(e), - Err(_) => return Ok(()), - }; + .await?; - let SubmissionHandoff { handoff, ehlo_host } = sh; - let Handoff { - stream, - mech, - raw_challenge, - identifier, - residual, - } = handoff; + let mut state = SubmissionState::default(); - let identifier = ctx.routing_identifier(identifier); - let dest_id = ctx.resolve(identifier.as_deref()).await; - let started = std::time::Instant::now(); - common::hold_login(ctx, identifier.as_deref()).await; - let fwd = forward::ForwardInfo::new( - peer, - local, - stream.is_tls(), - Some(ehlo_host), - ctx.config.server.proxy_ttl, - ); - - let auth = common::AuthFrame::inline( - format!( - "AUTH {} {}\r\n", - mechanism_name(&mech), - BASE64.encode(&raw_challenge) + loop { + let sh = match timeout( + listener.preauth_timeout, + preauth(ctx, listener, caps, stream, &hostname, &mut state), ) - .into_bytes(), - ); + .await + { + Ok(Ok(Some(h))) => h, + Ok(Ok(None)) => return Ok(()), + Ok(Err(e)) => return Err(e), + Err(_) => return Ok(()), + }; - common::bridge_authenticated( - ctx, - &dest_id, - stream, - peer, - identifier.as_deref(), - &residual, - started, - "submission", - common::AuthWire::Smtp, - &auth, - b"421 4.3.0 backend unavailable\r\n", - MAX_PREAUTH, - || open_backend(ctx, &dest_id, &fwd), - ) - .await + let SubmissionHandoff { + handoff, + ehlo_host: announced, + } = sh; + let Handoff { + stream: client, + mech, + raw_challenge, + identifier, + residual, + } = handoff; + + let identifier = ctx.routing_identifier(identifier); + let dest_id = ctx.resolve(identifier.as_deref()).await; + let started = std::time::Instant::now(); + common::hold_login(ctx, identifier.as_deref()).await; + let fwd = forward::ForwardInfo::new( + peer, + local, + client.is_tls(), + Some(announced), + ctx.config.server.proxy_ttl, + ); + + let auth = common::AuthFrame::inline( + format!( + "AUTH {} {}\r\n", + mechanism_name(&mech), + BASE64.encode(&raw_challenge) + ) + .into_bytes(), + ); + + let resume = common::bridge_authenticated( + ctx, + &dest_id, + client, + peer, + identifier.as_deref(), + residual, + started, + "submission", + common::AuthWire::Smtp, + &auth, + b"421 4.3.0 backend unavailable\r\n", + MAX_PREAUTH, + || open_backend(ctx, &dest_id, &fwd), + ) + .await?; + + let Some(common::Resume { + stream: client, + buf, + }) = resume + else { + return Ok(()); + }; + if bump(&mut state.auth_failures, listener.max_auth_attempts) { + return Ok(()); + } + stream = client; + state.buf = buf; + } } async fn preauth( @@ -106,21 +149,17 @@ async fn preauth( listener: &ListenerConfig, caps: &SubmissionCapabilities, mut stream: BoxedStream, - local: SocketAddr, + hostname: &str, + state: &mut SubmissionState, ) -> Result> { let mut is_tls = stream.is_tls(); let starttls_listener = listener.tls == TlsMode::Starttls; - let hostname = ctx.config.server.hostname_for(local); - write( - &mut stream, - format!("220 {hostname} ESMTP {}\r\n", caps.banner).as_bytes(), - ) - .await?; - - let mut buf: Vec = Vec::with_capacity(1024); - let mut ehlo_host = String::new(); - let mut auth_failures = 0u32; + let SubmissionState { + buf, + auth_failures, + ehlo_host, + } = state; loop { let mut it = buf.iter(); @@ -133,10 +172,10 @@ async fn preauth( let plain_ok = plain_auth_allowed(caps.allow_plain_auth_without_tls, is_tls); match req { Request::Ehlo { host } | Request::Lhlo { host } => { - ehlo_host = host.to_string(); + *ehlo_host = host.to_string(); let resp = ehlo_response( - &hostname, - &ehlo_host, + hostname, + ehlo_host, caps, starttls_listener && !is_tls, plain_ok, @@ -145,7 +184,7 @@ async fn preauth( buf.drain(..consumed); } Request::Helo { host } => { - ehlo_host = host.to_string(); + *ehlo_host = host.to_string(); write(&mut stream, format!("250 {hostname}\r\n").as_bytes()).await?; buf.drain(..consumed); } @@ -187,17 +226,17 @@ async fn preauth( ctx, plain_ok, login_offered: common::sasl_offered(&caps.sasl, "LOGIN"), - ehlo_host: &ehlo_host, + ehlo_host, max_attempts: listener.max_auth_attempts, }; match handle_auth( ¶ms, &mut stream, - &mut buf, + buf, consumed, mechanism, &ir, - &mut auth_failures, + &mut *auth_failures, ) .await? { @@ -233,7 +272,7 @@ async fn preauth( if buf.len() > MAX_PREAUTH { return Err(ProxyError::preauth("smtp line too long")); } - let n = common::fill(&mut stream, &mut buf, MAX_PREAUTH + 8192).await?; + let n = common::fill(&mut stream, buf, MAX_PREAUTH + 8192).await?; if n == 0 { return Ok(None); } @@ -242,8 +281,7 @@ async fn preauth( let drop_to = if consumed == 0 { buf.len() } else { consumed }; buf.drain(..drop_to); write(&mut stream, b"500 5.5.2 Syntax error\r\n").await?; - auth_failures += 1; - if auth_failures >= listener.max_auth_attempts { + if bump(auth_failures, listener.max_auth_attempts) { return Ok(None); } } @@ -448,8 +486,7 @@ async fn cancelled( max_attempts: u32, ) -> Result { write(&mut *stream, b"501 5.7.0 Authentication cancelled\r\n").await?; - *auth_failures += 1; - if *auth_failures >= max_attempts { + if bump(auth_failures, max_attempts) { Ok(AuthStep::Close) } else { Ok(AuthStep::Continue) @@ -468,8 +505,7 @@ async fn fail( buf.drain(..drain); } write(&mut *stream, format!("535 {message}\r\n").as_bytes()).await?; - *auth_failures += 1; - if *auth_failures >= max_attempts { + if bump(auth_failures, max_attempts) { Ok(AuthStep::Close) } else { Ok(AuthStep::Continue) diff --git a/src/tests/http_proxy.rs b/src/tests/http_proxy.rs index 763f3ca..73999c9 100644 --- a/src/tests/http_proxy.rs +++ b/src/tests/http_proxy.rs @@ -985,3 +985,364 @@ async fn http_chunked_request_and_response_round_trip() { let _ = backend.await; handler.abort(); } + +fn extract_http_config(host: &str, port: u16, mapping_path: &str) -> String { + format!( + r#" +[mapping] +source = "file" +normalize = "lowercase" +[mapping.file] +path = "{mapping_path}" + +[routing] +default_destination = "web" + +[destination.web] +host = "{host}" +proxy_protocol = false +allow_plaintext_auth = true +[destination.web.protocol.http] +port = {port} +tls = "plain" + +[[http.route]] +match = "/**" +extract = {{ from = "authorization" }} + +[listener.http] +protocol = "http" +bind = ["127.0.0.1:0"] +tls = "plain" +"# + ) +} + +#[tokio::test] +async fn http_authenticated_keepalive_connection_is_kickable() { + async fn backend(mut sock: tokio::net::TcpStream) { + read_req_head(&mut sock).await; + sock.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") + .await + .unwrap(); + let mut sink = [0u8; 1024]; + loop { + match sock.read(&mut sink).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + } + + let (port, backend) = dummy_backend(backend).await; + let mapping = mapping_file("user@example.com\tweb\n"); + let toml = extract_http_config("127.0.0.1", port, &mapping.path().display().to_string()); + let ctx = make_ctx(&toml).await; + let listener = ctx.config.listener["http"].clone(); + + let (mut client, srv) = tcp_pair().await; + let ctx2 = ctx.clone(); + let handler = tokio::spawn(async move { + let _ = crate::http::proxy::handle( + &ctx2, + &listener, + Box::new(srv), + "127.0.0.1:40070".parse().unwrap(), + "127.0.0.1:80".parse().unwrap(), + ) + .await; + }); + + let creds = BASE64.encode("user@example.com:pw"); + let req = format!("GET /jmap HTTP/1.1\r\nHost: x\r\nAuthorization: Basic {creds}\r\n\r\n"); + send(&mut client, req.as_bytes()).await; + let resp = read_until(&mut client, b"ok").await; + assert!(find(&resp, b"200").is_some(), "expected 200"); + + assert_eq!( + ctx.conns.kick(&ctx.router.normalize("user@example.com")), + 1, + "an authenticated http connection must be registered for its account" + ); + + let n = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.read(&mut [0u8; 64]), + ) + .await + .expect("kick did not disconnect the client in time") + .unwrap(); + assert_eq!(n, 0, "expected EOF after kick, got {n} bytes"); + + let _ = backend.await; + handler.abort(); +} + +#[tokio::test] +async fn http_websocket_bridge_is_kickable() { + async fn backend(mut sock: tokio::net::TcpStream) { + read_req_head(&mut sock).await; + sock.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + ) + .await + .unwrap(); + let mut sink = [0u8; 1024]; + loop { + match sock.read(&mut sink).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + } + + let (port, backend) = dummy_backend(backend).await; + let mapping = mapping_file("user@example.com\tweb\n"); + let toml = extract_http_config("127.0.0.1", port, &mapping.path().display().to_string()); + let ctx = make_ctx(&toml).await; + let listener = ctx.config.listener["http"].clone(); + + let (mut client, srv) = tcp_pair().await; + let ctx2 = ctx.clone(); + let handler = tokio::spawn(async move { + let _ = crate::http::proxy::handle( + &ctx2, + &listener, + Box::new(srv), + "127.0.0.1:40071".parse().unwrap(), + "127.0.0.1:80".parse().unwrap(), + ) + .await; + }); + + let creds = BASE64.encode("user@example.com:pw"); + let req = format!( + "GET /ws HTTP/1.1\r\nHost: x\r\nAuthorization: Basic {creds}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n" + ); + send(&mut client, req.as_bytes()).await; + + let upgrade = read_until(&mut client, b"\r\n\r\n").await; + assert!(find(&upgrade, b"101").is_some(), "expected the 101 upgrade"); + + assert_eq!( + ctx.conns.kick(&ctx.router.normalize("user@example.com")), + 1, + "an upgraded websocket must stay registered for its account" + ); + + let n = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.read(&mut [0u8; 64]), + ) + .await + .expect("kick did not disconnect the websocket in time") + .unwrap(); + assert_eq!(n, 0, "expected EOF after kick, got {n} bytes"); + + let _ = backend.await; + handler.abort(); +} + +#[tokio::test] +async fn http_keepalive_rejects_a_second_account_with_421() { + async fn backend(mut sock: tokio::net::TcpStream) { + for _ in 0..2 { + read_req_head(&mut sock).await; + if sock + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") + .await + .is_err() + { + return; + } + } + let mut sink = [0u8; 1024]; + loop { + match sock.read(&mut sink).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + } + + let (port, backend) = dummy_backend(backend).await; + let mapping = mapping_file("user@example.com\tweb\nother@example.com\tweb\n"); + let toml = extract_http_config("127.0.0.1", port, &mapping.path().display().to_string()); + let ctx = make_ctx(&toml).await; + let listener = ctx.config.listener["http"].clone(); + + let (mut client, srv) = tcp_pair().await; + let ctx2 = ctx.clone(); + let handler = tokio::spawn(async move { + let _ = crate::http::proxy::handle( + &ctx2, + &listener, + Box::new(srv), + "127.0.0.1:40072".parse().unwrap(), + "127.0.0.1:80".parse().unwrap(), + ) + .await; + }); + + let first = BASE64.encode("user@example.com:pw"); + send( + &mut client, + format!("GET /jmap HTTP/1.1\r\nHost: x\r\nAuthorization: Basic {first}\r\n\r\n").as_bytes(), + ) + .await; + let resp = read_until(&mut client, b"ok").await; + assert!( + find(&resp, b"200").is_some(), + "expected 200 for the pinned account" + ); + + let second = BASE64.encode("other@example.com:pw"); + send( + &mut client, + format!("GET /jmap HTTP/1.1\r\nHost: x\r\nAuthorization: Basic {second}\r\n\r\n") + .as_bytes(), + ) + .await; + let tail = read_to_eof(&mut client).await; + assert!( + find(&tail, b"421").is_some(), + "a second account on a pinned connection must get 421, got {:?}", + String::from_utf8_lossy(&tail) + ); + + let _ = backend.await; + handler.abort(); +} + +#[tokio::test] +async fn http_empty_identifier_does_not_pin_or_register_the_connection() { + async fn backend(mut sock: tokio::net::TcpStream) { + for _ in 0..2 { + read_req_head(&mut sock).await; + if sock + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") + .await + .is_err() + { + return; + } + } + let mut sink = [0u8; 1024]; + loop { + match sock.read(&mut sink).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + } + + let (port, backend) = dummy_backend(backend).await; + let mapping = mapping_file("user@example.com\tweb\n"); + let toml = extract_http_config("127.0.0.1", port, &mapping.path().display().to_string()); + let ctx = make_ctx(&toml).await; + let listener = ctx.config.listener["http"].clone(); + + let (mut client, srv) = tcp_pair().await; + let ctx2 = ctx.clone(); + let handler = tokio::spawn(async move { + let _ = crate::http::proxy::handle( + &ctx2, + &listener, + Box::new(srv), + "127.0.0.1:40073".parse().unwrap(), + "127.0.0.1:80".parse().unwrap(), + ) + .await; + }); + + let empty = BASE64.encode(":pw"); + send( + &mut client, + format!("GET /jmap HTTP/1.1\r\nHost: x\r\nAuthorization: Basic {empty}\r\n\r\n").as_bytes(), + ) + .await; + let first = read_until(&mut client, b"ok").await; + assert!( + find(&first, b"200").is_some(), + "expected 200 for the empty login" + ); + + assert_eq!( + ctx.conns.kick(""), + 0, + "an empty identifier must not be registered as an account" + ); + + let creds = BASE64.encode("user@example.com:pw"); + send( + &mut client, + format!("GET /jmap HTTP/1.1\r\nHost: x\r\nAuthorization: Basic {creds}\r\n\r\n").as_bytes(), + ) + .await; + let second = read_until(&mut client, b"ok").await; + assert!( + find(&second, b"421").is_none(), + "an empty first login must not pin the connection, got {:?}", + String::from_utf8_lossy(&second) + ); + assert_eq!( + ctx.conns.kick(&ctx.router.normalize("user@example.com")), + 1, + "the real account must be the one registered" + ); + + let _ = backend.await; + handler.abort(); +} + +#[tokio::test] +async fn http_login_delay_holds_an_authenticated_request() { + async fn backend(mut sock: tokio::net::TcpStream) { + read_req_head(&mut sock).await; + let _ = sock + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") + .await; + } + + let (port, backend) = dummy_backend(backend).await; + let mapping = mapping_file("user@example.com\tweb\n"); + let toml = extract_http_config("127.0.0.1", port, &mapping.path().display().to_string()); + let ctx = make_ctx(&toml).await; + let listener = ctx.config.listener["http"].clone(); + + ctx.conns.set_delay( + &ctx.router.normalize("user@example.com"), + std::time::Instant::now() + std::time::Duration::from_millis(1500), + ); + + let (mut client, srv) = tcp_pair().await; + let ctx2 = ctx.clone(); + let handler = tokio::spawn(async move { + let _ = crate::http::proxy::handle( + &ctx2, + &listener, + Box::new(srv), + "127.0.0.1:40075".parse().unwrap(), + "127.0.0.1:80".parse().unwrap(), + ) + .await; + }); + + let started = std::time::Instant::now(); + let creds = BASE64.encode("user@example.com:pw"); + send( + &mut client, + format!("GET /jmap HTTP/1.1\r\nHost: x\r\nAuthorization: Basic {creds}\r\nConnection: close\r\n\r\n").as_bytes(), + ) + .await; + let resp = read_to_eof(&mut client).await; + assert!(find(&resp, b"200").is_some(), "expected 200"); + assert!( + started.elapsed() >= std::time::Duration::from_millis(1400), + "the login delay must hold an http request too, elapsed {:?}", + started.elapsed() + ); + + let _ = backend.await; + handler.abort(); +} diff --git a/src/tests/imap.rs b/src/tests/imap.rs index b9321f9..d2cf8db 100644 --- a/src/tests/imap.rs +++ b/src/tests/imap.rs @@ -801,3 +801,175 @@ async fn imap_keeps_inline_initial_response_when_backend_advertises_sasl_ir() { let _ = backend.await; handler.abort(); } + +#[tokio::test] +async fn imap_pipelined_command_after_failed_auth_is_parsed_not_bridged() { + let (port, backend) = dummy_backend(|mut sock| async move { + sock.write_all(b"* OK [CAPABILITY IMAP4rev2 SASL-IR] backend ready\r\n") + .await + .unwrap(); + let mut buf = Vec::new(); + let mut scratch = [0u8; 1024]; + while find(&buf, b"AUTHENTICATE").is_none() || find(&buf, b"\r\n").is_none() { + match sock.read(&mut scratch).await { + Ok(0) | Err(_) => return, + Ok(n) => buf.extend_from_slice(&scratch[..n]), + } + } + let _ = sock + .write_all(b"a1 NO [AUTHENTICATIONFAILED] nope\r\n") + .await; + let mut sink = [0u8; 1024]; + loop { + match sock.read(&mut sink).await { + Ok(0) | Err(_) => break, + Ok(n) => { + assert!( + find(&sink[..n], b"CAPABILITY").is_none(), + "pipelined client bytes must not reach the backend after a failed auth" + ); + } + } + } + }) + .await; + + let mapping = mapping_file("user@example.com\tlegacy\n"); + 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), + "203.0.113.9:40010".parse().unwrap(), + "127.0.0.1:143".parse().unwrap(), + ) + .await; + }); + + let _ = read_line(&mut client).await; + send( + &mut client, + b"a1 LOGIN user@example.com wrong\r\na2 CAPABILITY\r\n", + ) + .await; + + let seen = read_until(&mut client, b"a2 OK").await; + assert!( + find(&seen, b"a1 NO").is_some(), + "expected the backend rejection, got {:?}", + String::from_utf8_lossy(&seen) + ); + assert!( + find(&seen, b"a2 OK CAPABILITY completed").is_some(), + "the pipelined command must be answered by the proxy, got {:?}", + String::from_utf8_lossy(&seen) + ); + + drop(client); + let _ = backend.await; + let _ = handler.await; +} + +async fn preauth_only_ctx() -> ( + std::sync::Arc, + tokio::task::JoinHandle<()>, +) { + let (port, backend) = dummy_backend(|_sock| async move {}).await; + let mapping = mapping_file("user@example.com\tlegacy\n"); + let toml = config(port, &mapping.path().display().to_string()); + (make_ctx(&toml).await, backend) +} + +#[tokio::test] +async fn imap_rejected_sasl_response_is_not_reparsed_as_a_command() { + let (ctx, backend) = preauth_only_ctx().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), + "203.0.113.9:40011".parse().unwrap(), + "127.0.0.1:143".parse().unwrap(), + ) + .await; + }); + + let _ = read_line(&mut client).await; + send( + &mut client, + b"a1 AUTHENTICATE PLAIN\r\n@@@\r\na2 CAPABILITY\r\n", + ) + .await; + + let seen = read_until(&mut client, b"a2 OK").await; + assert!( + find(&seen, b"a1 NO").is_some(), + "expected the SASL response to be rejected, got {:?}", + String::from_utf8_lossy(&seen) + ); + assert!( + find(&seen, b"* BAD").is_none(), + "the rejected SASL response must not be re-parsed as a command, got {:?}", + String::from_utf8_lossy(&seen) + ); + assert!( + find(&seen, b"a2 OK CAPABILITY completed").is_some(), + "the pipelined command must still be answered, got {:?}", + String::from_utf8_lossy(&seen) + ); + + drop(client); + backend.abort(); + let _ = handler.await; +} + +#[tokio::test] +async fn imap_command_pipelined_behind_a_rejected_sasl_response_is_not_lost() { + let (ctx, backend) = preauth_only_ctx().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), + "203.0.113.9:40012".parse().unwrap(), + "127.0.0.1:143".parse().unwrap(), + ) + .await; + }); + + let _ = read_line(&mut client).await; + send(&mut client, b"a1 AUTHENTICATE PLAIN\r\n").await; + let challenge = read_until(&mut client, b"+ ").await; + assert!(find(&challenge, b"+ ").is_some(), "expected a continuation"); + + send(&mut client, b"@@@\r\na2 CAPABILITY\r\n").await; + let seen = read_until(&mut client, b"a2 OK").await; + assert!( + find(&seen, b"a1 NO").is_some(), + "expected the SASL response to be rejected, got {:?}", + String::from_utf8_lossy(&seen) + ); + assert!( + find(&seen, b"a2 OK CAPABILITY completed").is_some(), + "the command behind the rejected response must not be dropped, got {:?}", + String::from_utf8_lossy(&seen) + ); + + drop(client); + backend.abort(); + let _ = handler.await; +} diff --git a/src/tests/pop3.rs b/src/tests/pop3.rs index 50b86e8..d2fe17e 100644 --- a/src/tests/pop3.rs +++ b/src/tests/pop3.rs @@ -609,3 +609,366 @@ tls = "plain" String::from_utf8_lossy(&resp) ); } + +fn retry_config(backend_port: u16, mapping_path: &str, max_auth_attempts: u32) -> String { + format!( + r#" +[mapping] +source = "file" +normalize = "lowercase" +[mapping.file] +path = "{mapping_path}" + +[capabilities.pop3] +allow_plain_auth_without_tls = true + +[routing] +default_destination = "legacy" + +[destination.legacy] +host = "127.0.0.1" +forwarding = "none" +allow_plaintext_auth = true +[destination.legacy.protocol.pop3] +port = {backend_port} +tls = "plain" + +[listener.pop3] +protocol = "pop3" +bind = ["127.0.0.1:0"] +tls = "plain" +max_auth_attempts = {max_auth_attempts} +"# + ) +} + +async fn rejecting_backend( + attempts: std::sync::Arc, +) -> (u16, tokio::task::JoinHandle<()>) { + let listener = tokio::net::TcpListener::bind(unused_addr()).await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let handle = tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let _ = sock.set_nodelay(true); + tokio::spawn(async move { + if sock.write_all(b"+OK ready\r\n").await.is_err() { + return; + } + let mut buf = Vec::new(); + let mut scratch = [0u8; 1024]; + while find(&buf, b"\r\n").is_none() { + match sock.read(&mut scratch).await { + Ok(0) | Err(_) => return, + Ok(n) => buf.extend_from_slice(&scratch[..n]), + } + } + let _ = sock.write_all(b"-ERR bad password\r\n").await; + let mut sink = [0u8; 1024]; + loop { + match sock.read(&mut sink).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + }); + } + }); + (port, handle) +} + +#[tokio::test] +async fn pop3_failed_backend_auth_does_not_hand_the_client_a_raw_bridge() { + let attempts = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let (backend_port, backend) = rejecting_backend(attempts.clone()).await; + + let mapping = mapping_file("user@example.com\tlegacy\n"); + let toml = retry_config(backend_port, &mapping.path().display().to_string(), 10); + let ctx = make_ctx(&toml).await; + let listener = ctx.config.listener["pop3"].clone(); + + let (mut client, server) = tcp_pair().await; + let ctx2 = ctx.clone(); + let handler = tokio::spawn(async move { + let _ = crate::proto::pop3::handle( + &ctx2, + &listener, + Box::new(server), + "203.0.113.9:40000".parse().unwrap(), + "127.0.0.1:110".parse().unwrap(), + ) + .await; + }); + + let greeting = read_line(&mut client).await; + assert!(greeting.starts_with("+OK"), "greeting: {greeting}"); + + send(&mut client, b"USER user@example.com\r\nPASS wrong\r\n").await; + let failure = read_until(&mut client, b"-ERR").await; + assert!(find(&failure, b"-ERR").is_some(), "expected a rejection"); + + send(&mut client, b"CAPA\r\n").await; + let capa = read_until(&mut client, b"\r\n.\r\n").await; + assert!( + find(&capa, b"USER").is_some(), + "the proxy must still be parsing after a failed auth, got {:?}", + String::from_utf8_lossy(&capa) + ); + assert_eq!( + attempts.load(std::sync::atomic::Ordering::SeqCst), + 1, + "CAPA must be answered by the proxy, not forwarded to a backend" + ); + + drop(client); + backend.abort(); + let _ = handler.await; +} + +#[tokio::test] +async fn pop3_rejected_auth_closes_the_backend_leg_with_quit() { + let (backend_port, backend) = dummy_backend(|mut sock| async move { + sock.write_all(b"+OK ready\r\n").await.unwrap(); + let mut buf = Vec::new(); + let mut scratch = [0u8; 1024]; + while find(&buf, b"\r\n").is_none() { + match sock.read(&mut scratch).await { + Ok(0) | Err(_) => return, + Ok(n) => buf.extend_from_slice(&scratch[..n]), + } + } + let _ = sock.write_all(b"-ERR bad password\r\n").await; + buf.clear(); + loop { + match sock.read(&mut scratch).await { + Ok(0) | Err(_) => break, + Ok(n) => buf.extend_from_slice(&scratch[..n]), + } + } + assert!( + find(&buf, b"QUIT").is_some(), + "backend leg must be closed with QUIT, saw {:?}", + String::from_utf8_lossy(&buf) + ); + }) + .await; + + let mapping = mapping_file("user@example.com\tlegacy\n"); + let toml = retry_config(backend_port, &mapping.path().display().to_string(), 10); + let ctx = make_ctx(&toml).await; + let listener = ctx.config.listener["pop3"].clone(); + + let (mut client, server) = tcp_pair().await; + let ctx2 = ctx.clone(); + let handler = tokio::spawn(async move { + let _ = crate::proto::pop3::handle( + &ctx2, + &listener, + Box::new(server), + "203.0.113.9:40003".parse().unwrap(), + "127.0.0.1:110".parse().unwrap(), + ) + .await; + }); + + let _ = read_line(&mut client).await; + send(&mut client, b"USER user@example.com\r\nPASS wrong\r\n").await; + let _ = read_until(&mut client, b"-ERR").await; + + drop(client); + backend.await.expect("backend assertion failed"); + let _ = handler.await; +} + +#[tokio::test] +async fn pop3_backend_auth_failures_count_toward_max_auth_attempts() { + let attempts = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let (backend_port, backend) = rejecting_backend(attempts.clone()).await; + + let mapping = mapping_file("user@example.com\tlegacy\n"); + let toml = retry_config(backend_port, &mapping.path().display().to_string(), 2); + let ctx = make_ctx(&toml).await; + let listener = ctx.config.listener["pop3"].clone(); + + let (mut client, server) = tcp_pair().await; + let ctx2 = ctx.clone(); + let handler = tokio::spawn(async move { + let _ = crate::proto::pop3::handle( + &ctx2, + &listener, + Box::new(server), + "203.0.113.9:40001".parse().unwrap(), + "127.0.0.1:110".parse().unwrap(), + ) + .await; + }); + + let _ = read_line(&mut client).await; + send(&mut client, b"USER user@example.com\r\nPASS wrong\r\n").await; + let _ = read_until(&mut client, b"-ERR").await; + send(&mut client, b"USER user@example.com\r\nPASS wrong\r\n").await; + + let tail = read_to_eof(&mut client).await; + assert!( + find(&tail, b"-ERR").is_some(), + "expected the second rejection, got {:?}", + String::from_utf8_lossy(&tail) + ); + assert_eq!( + attempts.load(std::sync::atomic::Ordering::SeqCst), + 2, + "max_auth_attempts must bound the backend connections a client can force" + ); + + backend.abort(); + let _ = handler.await; +} + +#[tokio::test] +async fn pop3_retry_after_failed_auth_is_routed_and_registered_for_its_own_account() { + async fn accepting_backend( + seen: std::sync::Arc>>, + ) -> (u16, tokio::task::JoinHandle<()>) { + let listener = tokio::net::TcpListener::bind(unused_addr()).await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let handle = tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let seen = seen.clone(); + let _ = sock.set_nodelay(true); + tokio::spawn(async move { + if sock.write_all(b"+OK ready\r\n").await.is_err() { + return; + } + let mut buf = Vec::new(); + let mut scratch = [0u8; 1024]; + while find(&buf, b"\r\n").is_none() { + match sock.read(&mut scratch).await { + Ok(0) | Err(_) => return, + Ok(n) => buf.extend_from_slice(&scratch[..n]), + } + } + seen.lock().await.extend_from_slice(&buf); + let _ = sock.write_all(b"+OK logged in\r\n").await; + let mut sink = [0u8; 1024]; + loop { + match sock.read(&mut sink).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + }); + } + }); + (port, handle) + } + + let rejects = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let (legacy_port, legacy) = rejecting_backend(rejects.clone()).await; + let seen = std::sync::Arc::new(tokio::sync::Mutex::new(Vec::new())); + let (modern_port, modern) = accepting_backend(seen.clone()).await; + + let mapping = mapping_file("alice@example.com\tlegacy\nbob@example.com\tmodern\n"); + let toml = format!( + r#" +[mapping] +source = "file" +normalize = "lowercase" +[mapping.file] +path = "{}" + +[capabilities.pop3] +allow_plain_auth_without_tls = true + +[routing] +default_destination = "legacy" + +[destination.legacy] +host = "127.0.0.1" +forwarding = "none" +allow_plaintext_auth = true +[destination.legacy.protocol.pop3] +port = {legacy_port} +tls = "plain" + +[destination.modern] +host = "127.0.0.1" +forwarding = "none" +allow_plaintext_auth = true +[destination.modern.protocol.pop3] +port = {modern_port} +tls = "plain" + +[listener.pop3] +protocol = "pop3" +bind = ["127.0.0.1:0"] +tls = "plain" +max_auth_attempts = 10 +"#, + mapping.path().display() + ); + + let ctx = make_ctx(&toml).await; + let listener = ctx.config.listener["pop3"].clone(); + + let (mut client, server) = tcp_pair().await; + let ctx2 = ctx.clone(); + let handler = tokio::spawn(async move { + let _ = crate::proto::pop3::handle( + &ctx2, + &listener, + Box::new(server), + "203.0.113.9:40002".parse().unwrap(), + "127.0.0.1:110".parse().unwrap(), + ) + .await; + }); + + let _ = read_line(&mut client).await; + send(&mut client, b"USER alice@example.com\r\nPASS wrong\r\n").await; + let _ = read_until(&mut client, b"-ERR").await; + + send(&mut client, b"USER bob@example.com\r\nPASS right\r\n").await; + let ok = read_until(&mut client, b"+OK logged in").await; + assert!( + find(&ok, b"+OK logged in").is_some(), + "second account should authenticate, got {:?}", + String::from_utf8_lossy(&ok) + ); + + let replayed = seen.lock().await.clone(); + let expected = BASE64.encode("\0bob@example.com\0right"); + assert!( + find(&replayed, expected.as_bytes()).is_some(), + "the retry must be re-routed to bob's destination, backend saw {:?}", + String::from_utf8_lossy(&replayed) + ); + assert_eq!( + rejects.load(std::sync::atomic::Ordering::SeqCst), + 1, + "alice's destination must see only the first attempt" + ); + + assert_eq!( + ctx.conns.kick(&ctx.router.normalize("alice@example.com")), + 0, + "the failed account must not hold a registration" + ); + assert_eq!( + ctx.conns.kick(&ctx.router.normalize("bob@example.com")), + 1, + "the account that actually authenticated must be kickable" + ); + + let n = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.read(&mut [0u8; 64]), + ) + .await + .expect("kick did not disconnect the client in time") + .unwrap(); + assert_eq!(n, 0, "expected EOF after kick, got {n} bytes"); + + legacy.abort(); + modern.abort(); + let _ = handler.await; +}