starred/proxy
1
0
Fork 0
mirror of https://github.com/stalwartlabs/proxy.git synced 2026-11-03 08:13:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

v1.0.4

Browse Source
This commit is contained in:
Maurus Decimus
2026-09-10 21:05:45 +01:00
parent 3a02690854
commit 91ffeb1482
13 changed files with 1756 additions and 492 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/http/proxy.rs
+104 -13
View File
@@ -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<tokio::sync::watch::Sender<bool>>,
guard: Option<ConnGuard>,
pinned: Option<String>,
}
impl Account {
fn new(cancel: tokio::sync::watch::Sender<bool>) -> 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<ConnGuard> {
self.guard
}
}
enum Disposition {
Close,
Hijack {
backend: BoxedStream,
to_backend: Vec<u8>,
to_client: Vec<u8>,
guard: Option<ConnGuard>,
},
}
@@ -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<bool>,
) -> Result<Disposition> {
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<Vec<u8>>,
dest_id: String,
identifier: Option<String>,
}
async fn prepare(
ctx: &Ctx,
router: &HttpRouter,
@@ -899,7 +984,7 @@ async fn prepare(
head: &mut Head,
body_cap: usize,
body_idle: Duration,
) -> Result<Option<(Option<Vec<u8>>, String)>> {
) -> Result<Option<Prepared>> {
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 {
src/outbound.rs
+2 -2
View File
@@ -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<tokio::sync::watch::Receiver<bool>>) {
pub(crate) async fn wait_kick(cancel: Option<tokio::sync::watch::Receiver<bool>>) {
match cancel {
Some(mut rx) => {
while rx.changed().await.is_ok() {
src/proto/common.rs
+81 -16
View File
@@ -33,6 +33,22 @@ pub struct Handoff {
pub residual: Vec<u8>,
}
pub struct Resume {
pub stream: BoxedStream,
pub buf: Vec<u8>,
}
pub fn take_rest(it: &mut std::slice::Iter<'_, u8>) -> Vec<u8> {
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<Credential> {
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<bool>,
) -> ConnGuard {
let rx = cancel.subscribe();
self.insert(identifier, cancel, rx)
}
fn insert(
&self,
identifier: &str,
cancel: tokio::sync::watch::Sender<bool>,
rx: tokio::sync::watch::Receiver<bool>,
) -> 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<str> = 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<bool>,
) -> ConnGuard {
self.conns
.register_cancel(&self.router.normalize(identifier), cancel)
}
pub fn routing_identifier(&self, identifier: Option<String>) -> Option<String> {
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<F, Fut>(
mut stream: BoxedStream,
peer: SocketAddr,
identifier: Option<&str>,
client_residual: &[u8],
client_residual: Vec<u8>,
started: Instant,
label: &'static str,
wire: AuthWire,
@@ -818,7 +872,7 @@ pub async fn bridge_authenticated<F, Fut>(
unavailable: &[u8],
max: usize,
open: F,
) -> Result<()>
) -> Result<Option<Resume>>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<BackendLeg>>,
@@ -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<()> {
src/proto/imap.rs
+164 -98
View File
@@ -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<u8>,
auth_failures: &mut u32,
) -> Result<Decision> {
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::<Command>::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<Vec<u8>> = 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<u8>,
},
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<u8>, Vec<u8>) = 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<Step> {
let (user_b64, residual): (Vec<u8>, Vec<u8>) = if let Some(ir) = tokens.get(1) {
(ir.as_bytes().to_vec(), it.as_slice().to_vec())
(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<u8>,
) -> Result<Step> {
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<Step> {
async fn reject(
&mut self,
stream: &mut BoxedStream,
tag: &str,
message: &str,
residual: Option<Vec<u8>>,
) -> Result<Step> {
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,
})
}
}
src/proto/managesieve.rs
+175 -95
View File
@@ -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<Option<Handoff>> {
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<u8>,
auth_failures: &mut u32,
) -> Result<Option<Handoff>> {
let mut is_tls = stream.is_tls();
let starttls_listener = listener.tls == TlsMode::Starttls;
let mut receiver =
Receiver::<Command>::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<Vec<u8>> = 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<u8>,
},
Continue,
Resume {
residual: Vec<u8>,
},
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<u8>, Vec<u8>) = 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<AuthStep> {
let (user_b64, residual): (Vec<u8>, Vec<u8>) = if let Some(ir) = tokens.get(1) {
(ir.as_bytes().to_vec(), it.as_slice().to_vec())
(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<Vec<u8>>,
) -> Result<AuthStep> {
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<u8>) -> Result<(Vec<u8>, Vec<u8>)> {
src/proto/pop3.rs
+85 -71
View File
@@ -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<u8>,
auth_failures: &mut u32,
) -> Result<Option<Handoff>> {
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<u8> = Vec::with_capacity(1024);
let mut buf: Vec<u8> = seed;
let mut pending_user: Option<Vec<u8>> = 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..])),
src/proto/smtp.rs
+116 -80
View File
@@ -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<u8>,
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<Option<SubmissionHandoff>> {
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<u8> = 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(
&params,
&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<AuthStep> {
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)
src/tests/http_proxy.rs
+361
View File
@@ -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();
}
src/tests/imap.rs
+172
View File
@@ -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<crate::proto::common::Ctx>,
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;
}
src/tests/pop3.rs
+363
View File
@@ -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<std::sync::atomic::AtomicUsize>,
) -> (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<tokio::sync::Mutex<Vec<u8>>>,
) -> (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;
}