inbuxa Relay: ingest with session context, and live verdicts #254
No files matched your search
@@ -9,7 +9,9 @@
|
||||
//! the settings and recipient set a relay is sent, and each node's view of
|
||||
//! each relay's link, under `E` `s` + relay id + node id.
|
||||
|
||||
use super::{DownstreamKeys, FEATURE, Json, KIND_STATUS, Relay, get_versioned, replace};
|
||||
use super::{
|
||||
DownstreamKeys, FEATURE, Json, KIND_INGESTED, KIND_STATUS, Relay, get_versioned, replace,
|
||||
};
|
||||
use inbuxa_relay_proto::{
|
||||
pin::{Pin, pins_match},
|
||||
recipient::{self, RecipientDomain, RecipientHash},
|
||||
@@ -319,6 +321,101 @@ pub(super) fn clear_statuses(batch: &mut BatchBuilder, relay_id: u32, nodes: &[N
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Messages taken ---------------------------------------------------------
|
||||
|
||||
/// The longest `Inbound` id kept: the relay's are 25 characters.
|
||||
pub const INBOUND_ID_MAX: usize = 64;
|
||||
|
||||
/// Whether `id` can be an `Inbound` id: never anything but letters and
|
||||
/// digits.
|
||||
pub fn is_inbound_id(id: &str) -> bool {
|
||||
!id.is_empty() && id.len() <= INBOUND_ID_MAX && id.bytes().all(|b| b.is_ascii_alphanumeric())
|
||||
}
|
||||
|
||||
/// A relay's message, taken at `at`.
|
||||
#[derive(Debug, Clone, SerdeSerialize, SerdeDeserialize)]
|
||||
struct Taken {
|
||||
id: String,
|
||||
at: u64,
|
||||
}
|
||||
|
||||
fn ingested_class(relay_id: u32, inbound_id: &[u8]) -> ValueClass {
|
||||
let mut key = Vec::with_capacity(6 + inbound_id.len());
|
||||
key.push(FEATURE);
|
||||
key.push(KIND_INGESTED);
|
||||
key.extend_from_slice(&relay_id.to_be_bytes());
|
||||
key.extend_from_slice(inbound_id);
|
||||
ValueClass::Any(AnyClass {
|
||||
subspace: SUBSPACE_INBUXA,
|
||||
key,
|
||||
})
|
||||
}
|
||||
|
||||
/// RL-16: whether a relay's message was taken already.
|
||||
pub async fn was_ingested(data: &Store, relay_id: u32, inbound_id: &str) -> trc::Result<bool> {
|
||||
data.get_value::<Json<Taken>>(ValueKey::from(ingested_class(
|
||||
relay_id,
|
||||
inbound_id.as_bytes(),
|
||||
)))
|
||||
.await
|
||||
.caused_by(trc::location!())
|
||||
.map(|value| value.is_some())
|
||||
}
|
||||
|
||||
/// RL-16: records a relay's message as taken, before the relay is told so.
|
||||
pub async fn record_ingested(
|
||||
data: &Store,
|
||||
relay_id: u32,
|
||||
inbound_id: &str,
|
||||
now: u64,
|
||||
) -> trc::Result<()> {
|
||||
let taken = Taken {
|
||||
id: inbound_id.to_string(),
|
||||
at: now,
|
||||
};
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch.set(
|
||||
ingested_class(relay_id, inbound_id.as_bytes()),
|
||||
Json(&taken).serialize()?,
|
||||
);
|
||||
data.write(batch.build_all())
|
||||
.await
|
||||
.caused_by(trc::location!())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Forgets the messages from a relay taken before `before`: RL-16 keeps
|
||||
/// them for `heldMaxAge` and a day. How many went.
|
||||
pub async fn forget_ingested(data: &Store, relay_id: u32, before: u64) -> trc::Result<usize> {
|
||||
let mut old = Vec::new();
|
||||
data.iterate(
|
||||
IterateParams::new(
|
||||
ValueKey::from(ingested_class(relay_id, &[])),
|
||||
ValueKey::from(ingested_class(relay_id, &[u8::MAX; INBOUND_ID_MAX + 1])),
|
||||
),
|
||||
|_, value| {
|
||||
if let Ok(Json(taken)) = Json::<Taken>::deserialize(value)
|
||||
&& taken.at < before
|
||||
{
|
||||
old.push(taken.id);
|
||||
}
|
||||
Ok(true)
|
||||
},
|
||||
)
|
||||
.await
|
||||
.caused_by(trc::location!())?;
|
||||
for chunk in old.chunks(500) {
|
||||
let mut batch = BatchBuilder::new();
|
||||
for id in chunk {
|
||||
batch.clear(ingested_class(relay_id, id.as_bytes()));
|
||||
}
|
||||
data.write(batch.build_all())
|
||||
.await
|
||||
.caused_by(trc::location!())?;
|
||||
}
|
||||
Ok(old.len())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@@ -51,6 +51,7 @@ const FEATURE: u8 = b'E';
|
||||
const KIND_KEYS: u8 = b'k';
|
||||
const KIND_RELAY: u8 = b'r';
|
||||
const KIND_STATUS: u8 = b's';
|
||||
const KIND_INGESTED: u8 = b'i';
|
||||
const CREATE_ATTEMPTS: usize = 5;
|
||||
|
||||
const MAX_NAME: usize = 100;
|
||||
@@ -569,6 +570,7 @@ pub async fn delete(data: &Store, id: u32) -> trc::Result<()> {
|
||||
data.write(batch.build_all())
|
||||
.await
|
||||
.caused_by(trc::location!())?;
|
||||
link::forget_ingested(data, id, u64::MAX).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -840,6 +840,8 @@ pub enum ExpressionVariable {
|
||||
UrlOriginal = 89,
|
||||
Value = 90,
|
||||
ValueLower = 91,
|
||||
// inbuxa: RL-23, the relay a message came through
|
||||
Relay = 92,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
|
||||
@@ -3616,6 +3618,7 @@ pub static MTA_CONNECTION_VARIABLE: &[ExpressionVariable] = &[
|
||||
ExpressionVariable::IsTls,
|
||||
ExpressionVariable::Asn,
|
||||
ExpressionVariable::Country,
|
||||
ExpressionVariable::Relay,
|
||||
];
|
||||
|
||||
pub static MTA_EHLO_VARIABLE: &[ExpressionVariable] = &[
|
||||
@@ -3629,6 +3632,7 @@ pub static MTA_EHLO_VARIABLE: &[ExpressionVariable] = &[
|
||||
ExpressionVariable::HeloDomain,
|
||||
ExpressionVariable::Asn,
|
||||
ExpressionVariable::Country,
|
||||
ExpressionVariable::Relay,
|
||||
];
|
||||
|
||||
pub static MTA_MAIL_FROM_VARIABLE: &[ExpressionVariable] = &[
|
||||
@@ -3644,6 +3648,7 @@ pub static MTA_MAIL_FROM_VARIABLE: &[ExpressionVariable] = &[
|
||||
ExpressionVariable::AuthenticatedAs,
|
||||
ExpressionVariable::Asn,
|
||||
ExpressionVariable::Country,
|
||||
ExpressionVariable::Relay,
|
||||
];
|
||||
|
||||
pub static MTA_QUEUE_HOST_VARIABLE: &[ExpressionVariable] = &[
|
||||
@@ -3720,6 +3725,7 @@ pub static MTA_RCPT_TO_VARIABLE: &[ExpressionVariable] = &[
|
||||
ExpressionVariable::HeloDomain,
|
||||
ExpressionVariable::Asn,
|
||||
ExpressionVariable::Country,
|
||||
ExpressionVariable::Relay,
|
||||
];
|
||||
|
||||
pub static MTA_RCPT_VARIABLE: &[ExpressionVariable] = &[ExpressionVariable::Rcpt];
|
||||
|
||||
@@ -3775,6 +3775,7 @@ impl EnumImpl for ExpressionVariable {
|
||||
b"url_original" => ExpressionVariable::UrlOriginal,
|
||||
b"value" => ExpressionVariable::Value,
|
||||
b"value_lower" => ExpressionVariable::ValueLower,
|
||||
b"relay" => ExpressionVariable::Relay,
|
||||
}
|
||||
.copied()
|
||||
}
|
||||
@@ -3873,6 +3874,7 @@ impl EnumImpl for ExpressionVariable {
|
||||
ExpressionVariable::UrlOriginal => "url_original",
|
||||
ExpressionVariable::Value => "value",
|
||||
ExpressionVariable::ValueLower => "value_lower",
|
||||
ExpressionVariable::Relay => "relay",
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3974,11 +3976,12 @@ impl EnumImpl for ExpressionVariable {
|
||||
89 => Some(ExpressionVariable::UrlOriginal),
|
||||
90 => Some(ExpressionVariable::Value),
|
||||
91 => Some(ExpressionVariable::ValueLower),
|
||||
92 => Some(ExpressionVariable::Relay),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
const COUNT: usize = 92;
|
||||
const COUNT: usize = 93;
|
||||
}
|
||||
|
||||
impl serde::Serialize for ExpressionVariable {
|
||||
|
||||
@@ -61,11 +61,13 @@ struct Entry {
|
||||
}
|
||||
|
||||
impl Entry {
|
||||
/// A message waiting for a verdict is pending even while leased: the
|
||||
/// downstream has to know to answer it.
|
||||
fn state(&self, now: u64) -> InboundState {
|
||||
if self.lease.as_ref().is_some_and(|l| l.until > now) {
|
||||
InboundState::Claimed
|
||||
} else if self.waiter.is_some() {
|
||||
if self.waiter.is_some() {
|
||||
InboundState::Pending
|
||||
} else if self.lease.as_ref().is_some_and(|l| l.until > now) {
|
||||
InboundState::Claimed
|
||||
} else {
|
||||
InboundState::Held
|
||||
}
|
||||
@@ -267,13 +269,15 @@ impl Spool {
|
||||
/// The JMAP object for one message.
|
||||
pub fn get(&self, id: &str, now: u64) -> Option<Inbound> {
|
||||
let e = self.entries.get(id)?;
|
||||
let state = e.state(now);
|
||||
Some(Inbound {
|
||||
id: e.meta.id.clone(),
|
||||
received_at: time::utc_date(e.meta.received_at),
|
||||
state,
|
||||
claimed_until: (state == InboundState::Claimed)
|
||||
.then(|| time::utc_date(e.lease.as_ref().map_or(0, |l| l.until))),
|
||||
state: e.state(now),
|
||||
claimed_until: e
|
||||
.lease
|
||||
.as_ref()
|
||||
.filter(|l| l.until > now)
|
||||
.map(|l| time::utc_date(l.until)),
|
||||
envelope: e.meta.envelope.clone(),
|
||||
session: e.meta.session.clone(),
|
||||
size: e.meta.size,
|
||||
|
||||
@@ -589,6 +589,10 @@ async fn relay_end_to_end() {
|
||||
let got = ws.ok("Inbound/get", json!({ "ids": [live_id] })).await;
|
||||
assert_eq!(got["list"][0]["state"], "pending");
|
||||
ws.ok("Inbound/claim", json!({})).await;
|
||||
// Claimed, it still waits for its verdict
|
||||
let got = ws.ok("Inbound/get", json!({ "ids": [live_id] })).await;
|
||||
assert_eq!(got["list"][0]["state"], "pending");
|
||||
assert!(got["list"][0]["claimedUntil"].is_string());
|
||||
let verdict = ws
|
||||
.ok(
|
||||
"Inbound/set",
|
||||
|
||||
@@ -14,10 +14,19 @@
|
||||
//! relay's settings and recipient set the same as this server's: in full
|
||||
//! when the link comes up, then as changes, every few seconds.
|
||||
//!
|
||||
//! Push is asked for `Relay` only. Asking for `Inbound` would put the relay
|
||||
//! in live mode, which waits for verdicts this node doesn't give yet.
|
||||
//! Push is asked for `Relay` and `Inbound`, which puts the relay in live
|
||||
//! mode (RL-12). Every message the relay takes is claimed, unsealed, and
|
||||
//! judged by this server's own inbound pipeline with the relay's session
|
||||
//! (RL-23): a live one gets this server's verdict before the sender is
|
||||
//! answered, a held one is delivered, or reported to a sender who passed
|
||||
//! SPF or DMARC (RL-17, RL-18). A message taken is recorded before the
|
||||
//! relay is told, so one offered again isn't delivered twice (RL-16).
|
||||
|
||||
use common::{BuildServer, Inner, Server};
|
||||
use common::{
|
||||
BuildServer, Inner, Server,
|
||||
config::server::{DEFAULT_TLS_TIMEOUT, ServerProtocol},
|
||||
network::{ServerInstance, TcpAcceptor, limiter::ConcurrencyLimiter},
|
||||
};
|
||||
use futures::{SinkExt, StreamExt};
|
||||
use http_body_util::{BodyExt, Full};
|
||||
use hyper::body::Bytes;
|
||||
@@ -28,7 +37,8 @@ use inbuxa_features::{
|
||||
self as relays, DownstreamKeys, JoinState, Relay,
|
||||
link::{
|
||||
JoinRecorded, LinkState, LocalDomain, NODE_CERT_RENEW_AFTER, NodeCertificate, NodeLink,
|
||||
RecipientPlan, issue_node_certificate, pushed_settings, record_join, set_status,
|
||||
RecipientPlan, forget_ingested, is_inbound_id, issue_node_certificate, pushed_settings,
|
||||
record_ingested, record_join, set_status, was_ingested,
|
||||
},
|
||||
},
|
||||
};
|
||||
@@ -37,8 +47,11 @@ use inbuxa_relay_proto::{
|
||||
jmap::{ClientFrame, Invocation, Request, ServerFrame, Session, method},
|
||||
pin::{Pin, pins_match, spki_pin},
|
||||
recipient::{RecipientDomain, RecipientHash},
|
||||
token,
|
||||
types::{Capability, RELAY_SINGLETON, RelaySettings, RelayStatus},
|
||||
seal, token,
|
||||
types::{
|
||||
Address, Capability, Inbound, InboundState, RELAY_SINGLETON, RelaySettings, RelayStatus,
|
||||
Reply, Verdict,
|
||||
},
|
||||
};
|
||||
use registry::schema::{
|
||||
prelude::ObjectType,
|
||||
@@ -50,16 +63,25 @@ use rustls::{
|
||||
};
|
||||
use rustls_pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime};
|
||||
use serde_json::{Value, json};
|
||||
use smtp::{
|
||||
core::Relayed,
|
||||
inbound::relayed::{RelayedMessage, RelayedOutcome, ingest_relayed},
|
||||
};
|
||||
use std::{
|
||||
collections::{BTreeMap, BTreeSet, HashMap},
|
||||
collections::{BTreeMap, BTreeSet, HashMap, HashSet},
|
||||
sync::{Arc, Mutex},
|
||||
time::Duration,
|
||||
};
|
||||
use store::{registry::RegistryQuery, write::now};
|
||||
use tokio::{net::TcpStream, sync::watch, task::JoinHandle};
|
||||
use tokio::{
|
||||
net::TcpStream,
|
||||
sync::{mpsc, watch},
|
||||
task::JoinHandle,
|
||||
};
|
||||
use tokio_rustls::{TlsConnector, client::TlsStream};
|
||||
use tokio_tungstenite::{WebSocketStream, tungstenite::Message};
|
||||
use types::id::Id;
|
||||
use utils::snowflake::SnowflakeIdGenerator;
|
||||
|
||||
/// How often the list of relays is read again.
|
||||
const SCAN: Duration = Duration::from_secs(10);
|
||||
@@ -79,16 +101,40 @@ const MAX_DELTA: usize = 2_000;
|
||||
const ACCOUNT_ID: &str = "relay";
|
||||
/// Any name: the relay is known by its key, never its name (RL-6).
|
||||
const SERVER_NAME: &str = "relay.invalid";
|
||||
/// Messages one link takes at once.
|
||||
const TAKE_MAX: usize = 8;
|
||||
/// How long a claimed message is this node's (RL-15).
|
||||
const LEASE_SECONDS: u64 = 300;
|
||||
/// How often messages taken long ago are forgotten (RL-16).
|
||||
const FORGET_EVERY: u64 = 3_600;
|
||||
|
||||
pub fn spawn_relay_links(inner: Arc<Inner>) {
|
||||
tokio::spawn(async move {
|
||||
let certs = Arc::new(NodeCertificates::default());
|
||||
// Relayed mail's sessions: port 25's, but no listener's
|
||||
let instance = Arc::new(ServerInstance {
|
||||
id: "relay".to_string(),
|
||||
protocol: ServerProtocol::Smtp,
|
||||
acceptor: TcpAcceptor::Plain,
|
||||
limiter: ConcurrencyLimiter::new(100),
|
||||
tls_timeout: DEFAULT_TLS_TIMEOUT,
|
||||
shutdown_rx: watch::channel(false).1,
|
||||
proxy_networks: vec![],
|
||||
span_id_gen: Arc::new(SnowflakeIdGenerator::new()),
|
||||
});
|
||||
let mut links: HashMap<u32, Running> = HashMap::new();
|
||||
let mut forgot_at = 0;
|
||||
loop {
|
||||
let server = inner.build_server();
|
||||
if let Err(err) = reconcile(&inner, &server, &certs, &mut links).await {
|
||||
if let Err(err) = reconcile(&inner, &server, &certs, &instance, &mut links).await {
|
||||
trc::error!(err.details("Failed to start relay links"));
|
||||
}
|
||||
if now() >= forgot_at + FORGET_EVERY {
|
||||
forgot_at = now();
|
||||
if let Err(err) = forget_taken(&server).await {
|
||||
trc::error!(err.details("Failed to forget relayed messages taken"));
|
||||
}
|
||||
}
|
||||
let woken = tokio::select! {
|
||||
_ = tokio::time::sleep(SCAN) => false,
|
||||
_ = relays::CHANGED.notified() => true,
|
||||
@@ -115,6 +161,7 @@ async fn reconcile(
|
||||
inner: &Arc<Inner>,
|
||||
server: &Server,
|
||||
certs: &Arc<NodeCertificates>,
|
||||
instance: &Arc<ServerInstance>,
|
||||
links: &mut HashMap<u32, Running>,
|
||||
) -> trc::Result<()> {
|
||||
let data = server.store();
|
||||
@@ -145,7 +192,13 @@ async fn reconcile(
|
||||
continue;
|
||||
}
|
||||
let (wake, woken) = watch::channel(0);
|
||||
let handle = tokio::spawn(run_link(inner.clone(), id, certs.clone(), woken));
|
||||
let handle = tokio::spawn(run_link(
|
||||
inner.clone(),
|
||||
id,
|
||||
certs.clone(),
|
||||
instance.clone(),
|
||||
woken,
|
||||
));
|
||||
links.insert(
|
||||
id,
|
||||
Running {
|
||||
@@ -158,6 +211,16 @@ async fn reconcile(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// RL-16: a relay's messages are remembered for its `heldMaxAge` and a day.
|
||||
async fn forget_taken(server: &Server) -> trc::Result<()> {
|
||||
let data = server.store();
|
||||
for relay in relays::all(data).await? {
|
||||
let keep = (u64::from(relay.settings.held_max_age) + 1) * 86_400;
|
||||
forget_ingested(data, relay.id, now().saturating_sub(keep)).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// This node's link certificate, made again once it is old.
|
||||
#[derive(Default)]
|
||||
struct NodeCertificates(Mutex<Option<Arc<NodeCertificate>>>);
|
||||
@@ -210,6 +273,7 @@ async fn run_link(
|
||||
inner: Arc<Inner>,
|
||||
relay_id: u32,
|
||||
certs: Arc<NodeCertificates>,
|
||||
instance: Arc<ServerInstance>,
|
||||
mut woken: watch::Receiver<u64>,
|
||||
) {
|
||||
let mut backoff = BACKOFF_MIN;
|
||||
@@ -247,7 +311,16 @@ async fn run_link(
|
||||
"The join token ran out before the relay joined. Start a new join.",
|
||||
)),
|
||||
JoinState::Joined => {
|
||||
session(&server, &relay, &keys, &cert, &mut woken, &mut status).await
|
||||
session(
|
||||
&server,
|
||||
&instance,
|
||||
&relay,
|
||||
&keys,
|
||||
&cert,
|
||||
&mut woken,
|
||||
&mut status,
|
||||
)
|
||||
.await
|
||||
}
|
||||
},
|
||||
Err(err) => Err(err),
|
||||
@@ -651,11 +724,12 @@ async fn join(data: &store::Store, relay: &Relay, cert: &NodeCertificate) -> Res
|
||||
|
||||
// ---- The linked session ---------------------------------------------------
|
||||
|
||||
/// A WebSocket to the relay, with the `Relay` changes it pushed meanwhile.
|
||||
/// A WebSocket to the relay, with the changes it pushed meanwhile.
|
||||
struct Ws {
|
||||
stream: WebSocketStream<TlsStream<TcpStream>>,
|
||||
next_id: u64,
|
||||
relay_changed: bool,
|
||||
inbound_changed: bool,
|
||||
}
|
||||
|
||||
impl Ws {
|
||||
@@ -685,6 +759,7 @@ impl Ws {
|
||||
stream,
|
||||
next_id: 0,
|
||||
relay_changed: false,
|
||||
inbound_changed: false,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -716,10 +791,11 @@ impl Ws {
|
||||
}
|
||||
|
||||
fn note(&mut self, frame: &ServerFrame) {
|
||||
if let ServerFrame::StateChange { changed, .. } = frame
|
||||
&& changed.values().any(|types| types.contains_key("Relay"))
|
||||
{
|
||||
self.relay_changed = true;
|
||||
if let ServerFrame::StateChange { changed, .. } = frame {
|
||||
for types in changed.values() {
|
||||
self.relay_changed |= types.contains_key("Relay");
|
||||
self.inbound_changed |= types.contains_key("Inbound");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -801,6 +877,7 @@ struct Sent {
|
||||
/// turned off, moved or joined again, or this node stops.
|
||||
async fn session(
|
||||
server: &Server,
|
||||
instance: &Arc<ServerInstance>,
|
||||
relay: &Relay,
|
||||
keys: &DownstreamKeys,
|
||||
cert: &NodeCertificate,
|
||||
@@ -809,11 +886,12 @@ async fn session(
|
||||
) -> Result<(), LinkError> {
|
||||
let data = server.store();
|
||||
let pin = relay.join.relay_pin;
|
||||
let dialer = Dialer::new(relay, cert, pin)?;
|
||||
let dialer = Arc::new(Dialer::new(relay, cert, pin)?);
|
||||
dialer.check_session().await?;
|
||||
let mut ws = Ws::open(&dialer).await?;
|
||||
// Inbound too: live mode (RL-12)
|
||||
ws.send(&ClientFrame::WebSocketPushEnable {
|
||||
data_types: Some(vec!["Relay".into()]),
|
||||
data_types: Some(vec!["Relay".into(), "Inbound".into()]),
|
||||
push_state: None,
|
||||
})
|
||||
.await?;
|
||||
@@ -823,6 +901,19 @@ async fn session(
|
||||
status.set(data, relay.id, LinkState::Linked, None).await;
|
||||
refresh_status(&mut ws, data, &relay, &sent, status).await?;
|
||||
|
||||
let (answers_tx, mut answers) = mpsc::unbounded_channel();
|
||||
let mut taker = Taker {
|
||||
server: server.clone(),
|
||||
instance: instance.clone(),
|
||||
dialer: dialer.clone(),
|
||||
relay_id: relay.id,
|
||||
relay_name: relay.name.clone(),
|
||||
secret: keys.sealing_secret,
|
||||
taking: HashSet::new(),
|
||||
answers: answers_tx,
|
||||
};
|
||||
taker.take(&mut ws).await?;
|
||||
|
||||
let mut tick = tokio::time::interval(SYNC);
|
||||
tick.reset();
|
||||
loop {
|
||||
@@ -833,6 +924,14 @@ async fn session(
|
||||
if ws.relay_changed {
|
||||
refresh_status(&mut ws, data, &relay, &sent, status).await?;
|
||||
}
|
||||
if ws.inbound_changed {
|
||||
taker.take(&mut ws).await?;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
Some(answer) = answers.recv() => {
|
||||
taker.answer(&mut ws, answer).await?;
|
||||
taker.take(&mut ws).await?;
|
||||
continue;
|
||||
}
|
||||
_ = tick.tick() => {}
|
||||
@@ -856,6 +955,300 @@ async fn session(
|
||||
}
|
||||
sync(server, &relay, keys, &dialer, &mut ws, &mut sent).await?;
|
||||
refresh_status(&mut ws, data, &relay, &sent, status).await?;
|
||||
taker.relay_name = relay.name.clone();
|
||||
taker.take(&mut ws).await?;
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Inbound mail -----------------------------------------------------------
|
||||
|
||||
/// Takes the relay's messages for this node: a few at a time, each judged in
|
||||
/// its own task, each answered on the link when judged.
|
||||
struct Taker {
|
||||
server: Server,
|
||||
instance: Arc<ServerInstance>,
|
||||
dialer: Arc<Dialer>,
|
||||
relay_id: u32,
|
||||
relay_name: String,
|
||||
secret: [u8; 32],
|
||||
/// Claimed and still being judged.
|
||||
taking: HashSet<String>,
|
||||
answers: mpsc::UnboundedSender<Answer>,
|
||||
}
|
||||
|
||||
/// What a judged message's answer to the relay is.
|
||||
struct Answer {
|
||||
id: String,
|
||||
action: Action,
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
enum Action {
|
||||
/// Taken: the relay deletes it, and a live sender hears 250.
|
||||
Destroy,
|
||||
/// A live message refused: the relay gives the sender this reply.
|
||||
Verdict(Verdict),
|
||||
/// Not now: the lease runs out and it is offered again (RL-16).
|
||||
Leave,
|
||||
}
|
||||
|
||||
impl Taker {
|
||||
/// Claims what the relay holds, up to [`TAKE_MAX`] at once.
|
||||
async fn take(&mut self, ws: &mut Ws) -> Result<(), LinkError> {
|
||||
ws.inbound_changed = false;
|
||||
let room = TAKE_MAX.saturating_sub(self.taking.len());
|
||||
if room == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
let claimed = ws
|
||||
.ok(
|
||||
method::INBOUND_CLAIM,
|
||||
json!({"limit": room, "leaseSeconds": LEASE_SECONDS}),
|
||||
)
|
||||
.await?;
|
||||
let ids: Vec<&str> = claimed["ids"]
|
||||
.as_array()
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.filter_map(Value::as_str)
|
||||
.collect();
|
||||
if ids.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let got = ws.ok(method::INBOUND_GET, json!({"ids": ids})).await?;
|
||||
for inbound in got["list"].as_array().into_iter().flatten() {
|
||||
let Ok(inbound) = serde_json::from_value::<Inbound>(inbound.clone()) else {
|
||||
continue;
|
||||
};
|
||||
if !is_inbound_id(&inbound.id)
|
||||
|| !is_inbound_id(&inbound.blob_id)
|
||||
|| !self.taking.insert(inbound.id.clone())
|
||||
{
|
||||
continue;
|
||||
}
|
||||
let job = Job {
|
||||
server: self.server.clone(),
|
||||
instance: self.instance.clone(),
|
||||
dialer: self.dialer.clone(),
|
||||
relay_id: self.relay_id,
|
||||
relay_name: self.relay_name.clone(),
|
||||
secret: self.secret,
|
||||
};
|
||||
let answers = self.answers.clone();
|
||||
tokio::spawn(async move {
|
||||
let id = inbound.id.clone();
|
||||
let action = job.judge(inbound).await;
|
||||
let _ = answers.send(Answer { id, action });
|
||||
});
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Tells the relay what became of one message.
|
||||
async fn answer(&mut self, ws: &mut Ws, answer: Answer) -> Result<(), LinkError> {
|
||||
self.taking.remove(&answer.id);
|
||||
match answer.action {
|
||||
Action::Destroy => {
|
||||
ws.ok(method::INBOUND_SET, json!({"destroy": [answer.id]}))
|
||||
.await?;
|
||||
}
|
||||
Action::Verdict(verdict) => {
|
||||
// Refused as `notUpdated` if it stopped waiting meanwhile:
|
||||
// then it is held, comes back when the lease runs out, and
|
||||
// is judged as held
|
||||
ws.ok(
|
||||
method::INBOUND_SET,
|
||||
json!({"update": {answer.id: {"verdict": verdict}}}),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
Action::Leave => {}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Everything one message's judging needs, apart from the link.
|
||||
struct Job {
|
||||
server: Server,
|
||||
instance: Arc<ServerInstance>,
|
||||
dialer: Arc<Dialer>,
|
||||
relay_id: u32,
|
||||
relay_name: String,
|
||||
secret: [u8; 32],
|
||||
}
|
||||
|
||||
impl Job {
|
||||
async fn judge(self, inbound: Inbound) -> Action {
|
||||
let data = self.server.store().clone();
|
||||
// RL-16: offered again after this server took it
|
||||
match was_ingested(&data, self.relay_id, &inbound.id).await {
|
||||
Ok(true) => return Action::Destroy,
|
||||
Ok(false) => {}
|
||||
Err(err) => {
|
||||
trc::error!(err.details("Failed to read the relayed messages taken"));
|
||||
return Action::Leave;
|
||||
}
|
||||
}
|
||||
let live = inbound.state == InboundState::Pending;
|
||||
|
||||
let message = match self.fetch(&inbound).await {
|
||||
Ok(message) => message,
|
||||
Err(reason) => {
|
||||
trc::error!(
|
||||
trc::EventType::Server(trc::ServerEvent::ThreadError)
|
||||
.reason(reason)
|
||||
.details("Failed to take a relayed message")
|
||||
.id(inbound.id.clone())
|
||||
);
|
||||
return Action::Leave;
|
||||
}
|
||||
};
|
||||
|
||||
let input = relayed_message(&self.relay_name, &inbound, !live, message);
|
||||
let outcome = ingest_relayed(self.server.clone(), self.instance.clone(), input).await;
|
||||
let action = action_for(&outcome, live);
|
||||
if action == Action::Destroy
|
||||
&& let Err(err) = record_ingested(&data, self.relay_id, &inbound.id, now()).await
|
||||
{
|
||||
trc::error!(err.details("Failed to record a relayed message as taken"));
|
||||
}
|
||||
action
|
||||
}
|
||||
|
||||
/// The message, downloaded and unsealed (RL-21).
|
||||
async fn fetch(&self, inbound: &Inbound) -> Result<Vec<u8>, String> {
|
||||
let (status, sealed) = self
|
||||
.dialer
|
||||
.http(
|
||||
"GET",
|
||||
&format!("/jmap/download/relay/{}/msg", inbound.blob_id),
|
||||
"application/octet-stream",
|
||||
Vec::new(),
|
||||
)
|
||||
.await
|
||||
.map_err(|err| err.message)?;
|
||||
if status != 200 {
|
||||
return Err(format!(
|
||||
"The relay answered {status} for a held message's download."
|
||||
));
|
||||
}
|
||||
seal::open(&self.secret, inbound.id.as_bytes(), &sealed)
|
||||
.map_err(|err| format!("A relayed message didn't unseal: {err}"))
|
||||
}
|
||||
}
|
||||
|
||||
/// The relay's session, as this server's pipeline takes it (RL-23).
|
||||
fn relayed_message(relay: &str, inbound: &Inbound, held: bool, message: Vec<u8>) -> RelayedMessage {
|
||||
let address = |address: &Address| {
|
||||
(
|
||||
address.email.clone(),
|
||||
address
|
||||
.parameters
|
||||
.iter()
|
||||
.flatten()
|
||||
.map(|(key, value)| (key.clone(), value.clone()))
|
||||
.collect(),
|
||||
)
|
||||
};
|
||||
let session = &inbound.session;
|
||||
RelayedMessage {
|
||||
relayed: Relayed {
|
||||
relay: relay.to_string(),
|
||||
held,
|
||||
tls: session
|
||||
.tls
|
||||
.as_ref()
|
||||
.map(|tls| (tls.version.clone(), tls.cipher.clone())),
|
||||
dmarc_pass: false,
|
||||
},
|
||||
local_ip: session.local_ip,
|
||||
remote_ip: session.remote_ip,
|
||||
remote_port: session.remote_port,
|
||||
helo: session.helo.clone(),
|
||||
mail_from: address(&inbound.envelope.mail_from),
|
||||
rcpt_to: inbound.envelope.rcpt_to.iter().map(address).collect(),
|
||||
message,
|
||||
}
|
||||
}
|
||||
|
||||
/// What the relay is told, from what this server answered.
|
||||
fn action_for(outcome: &RelayedOutcome, live: bool) -> Action {
|
||||
match outcome {
|
||||
RelayedOutcome::Accepted { .. } => Action::Destroy,
|
||||
RelayedOutcome::Refused { reply, .. } if live => Action::Verdict(verdict_for(reply)),
|
||||
// Held and refused for good: reported to the sender if it may be
|
||||
// (RL-18), and gone
|
||||
RelayedOutcome::Refused { reply, .. } if reply.starts_with('5') => Action::Destroy,
|
||||
// Held and refused for now: tried again after the lease
|
||||
RelayedOutcome::Refused { .. } => Action::Leave,
|
||||
}
|
||||
}
|
||||
|
||||
/// A live message's verdict: this server's reply, as the relay gives it.
|
||||
fn verdict_for(reply: &str) -> Verdict {
|
||||
let mut parts = reply.splitn(3, ' ');
|
||||
let code = parts.next().and_then(|code| code.parse::<u16>().ok());
|
||||
let class = code.map_or(4, |code| code / 100);
|
||||
let (enhanced, text) = match parts.next() {
|
||||
Some(esc)
|
||||
if esc.split('.').count() == 3
|
||||
&& esc.split('.').all(|p| p.parse::<u16>().is_ok())
|
||||
&& esc.starts_with(char::from(b'0' + class as u8)) =>
|
||||
{
|
||||
(
|
||||
esc.to_string(),
|
||||
parts.next().unwrap_or_default().to_string(),
|
||||
)
|
||||
}
|
||||
Some(word) => (
|
||||
format!("{class}.0.0"),
|
||||
[word, parts.next().unwrap_or_default()].join(" "),
|
||||
),
|
||||
None => (format!("{class}.0.0"), String::new()),
|
||||
};
|
||||
// Not a reply at all: its words, as a temporary failure
|
||||
let text = if code.is_none() {
|
||||
reply.to_string()
|
||||
} else {
|
||||
text
|
||||
};
|
||||
let text: String = text
|
||||
.trim()
|
||||
.chars()
|
||||
.filter(|c| !c.is_control())
|
||||
.take(400)
|
||||
.collect();
|
||||
let text = if text.is_empty() {
|
||||
"Refused".to_string()
|
||||
} else {
|
||||
text
|
||||
};
|
||||
let verdict = match (code, class) {
|
||||
(Some(code), 5) => Verdict::Reject(Reply {
|
||||
code,
|
||||
enhanced,
|
||||
text,
|
||||
}),
|
||||
(Some(code), 4) => Verdict::Tempfail(Reply {
|
||||
code,
|
||||
enhanced,
|
||||
text,
|
||||
}),
|
||||
_ => Verdict::Tempfail(Reply {
|
||||
code: 451,
|
||||
enhanced: "4.3.0".into(),
|
||||
text,
|
||||
}),
|
||||
};
|
||||
if verdict.is_valid() {
|
||||
verdict
|
||||
} else {
|
||||
Verdict::Tempfail(Reply {
|
||||
code: 451,
|
||||
enhanced: "4.3.0".into(),
|
||||
text: "Try again later".into(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1060,3 +1453,70 @@ async fn local_recipients(server: &Server) -> trc::Result<(Vec<LocalDomain>, Vec
|
||||
}
|
||||
Ok((local, addresses))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn reply(code: u16, enhanced: &str, text: &str) -> Reply {
|
||||
Reply {
|
||||
code,
|
||||
enhanced: enhanced.into(),
|
||||
text: text.into(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn verdicts() {
|
||||
assert_eq!(
|
||||
verdict_for("550 5.7.1 Message rejected due to excessive spam score."),
|
||||
Verdict::Reject(reply(
|
||||
550,
|
||||
"5.7.1",
|
||||
"Message rejected due to excessive spam score."
|
||||
))
|
||||
);
|
||||
assert_eq!(
|
||||
verdict_for("450 4.4.6 Too many Received headers."),
|
||||
Verdict::Tempfail(reply(450, "4.4.6", "Too many Received headers."))
|
||||
);
|
||||
// No enhanced code, or one of the wrong class: the class's own
|
||||
assert_eq!(
|
||||
verdict_for("554 no thanks"),
|
||||
Verdict::Reject(reply(554, "5.0.0", "no thanks"))
|
||||
);
|
||||
assert_eq!(
|
||||
verdict_for("451 5.7.1 odd"),
|
||||
Verdict::Tempfail(reply(451, "4.0.0", "5.7.1 odd"))
|
||||
);
|
||||
// Anything else is a temporary failure, never a 250
|
||||
assert_eq!(
|
||||
verdict_for("garbage"),
|
||||
Verdict::Tempfail(reply(451, "4.3.0", "garbage"))
|
||||
);
|
||||
assert!(verdict_for("550 5.1.1").is_valid());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn actions() {
|
||||
let accepted = RelayedOutcome::Accepted {
|
||||
refused: vec![("[email protected]".into(), "550 5.1.1 No".into())],
|
||||
};
|
||||
let refused = |reply: &str| RelayedOutcome::Refused {
|
||||
reply: reply.into(),
|
||||
refused: vec![],
|
||||
};
|
||||
assert_eq!(action_for(&accepted, true), Action::Destroy);
|
||||
assert_eq!(action_for(&accepted, false), Action::Destroy);
|
||||
assert!(matches!(
|
||||
action_for(&refused("550 5.7.1 Spam"), true),
|
||||
Action::Verdict(Verdict::Reject(_))
|
||||
));
|
||||
// Held: refused for good is reported and gone, for now is retried
|
||||
assert_eq!(action_for(&refused("550 5.1.1 No"), false), Action::Destroy);
|
||||
assert_eq!(
|
||||
action_for(&refused("451 4.3.0 Later"), false),
|
||||
Action::Leave
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -109,6 +109,23 @@ pub struct SessionData {
|
||||
// to, and recipients rules added, by rule name
|
||||
pub journal_marks: Vec<u32>,
|
||||
pub journal_added: Vec<(String, String)>,
|
||||
// inbuxa: RL-23: a message a relay took on port 25, judged here as if
|
||||
// this server had taken the session
|
||||
pub relayed: Option<Relayed>,
|
||||
}
|
||||
|
||||
/// inbuxa: RL-23: what a relay saw of the session it took a message in.
|
||||
/// Only the relay link service sets it, for a pinned relay.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct Relayed {
|
||||
/// The relay's name, for `relay` in expressions.
|
||||
pub relay: String,
|
||||
/// The relay already said 250 (RL-17).
|
||||
pub held: bool,
|
||||
/// TLS version and cipher between the sender and the relay.
|
||||
pub tls: Option<(String, String)>,
|
||||
/// DMARC passed, once DATA checked it (RL-18).
|
||||
pub dmarc_pass: bool,
|
||||
}
|
||||
|
||||
/// inbuxa: a DATA refusal by DLP rules: blocked, or a warning the sender
|
||||
@@ -199,6 +216,7 @@ impl SessionData {
|
||||
journal_marks: Vec::new(),
|
||||
journal_added: Vec::new(),
|
||||
access_token: None,
|
||||
relayed: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -328,6 +346,7 @@ impl SessionData {
|
||||
journal_marks: Vec::new(),
|
||||
journal_added: Vec::new(),
|
||||
access_token: None,
|
||||
relayed: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -428,6 +428,10 @@ impl<T: SessionStream> Session<T> {
|
||||
|
||||
let dmarc_result = dmarc_output.result();
|
||||
let pass = dmarc_result == DmarcResult::Pass;
|
||||
// inbuxa: RL-18, a held message's failures bounce only then
|
||||
if let Some(relayed) = self.data.relayed.as_mut() {
|
||||
relayed.dmarc_pass = pass;
|
||||
}
|
||||
let strict = dmarc.is_strict();
|
||||
let is_temp_fail = matches!(dmarc_result, DmarcResult::TempError(_));
|
||||
let rejected = strict
|
||||
@@ -604,6 +608,25 @@ impl<T: SessionStream> Session<T> {
|
||||
| rcpt_spam_flag(scores.spam_percentage(user_score));
|
||||
}
|
||||
}
|
||||
// inbuxa: RL-17: held mail was already accepted, so it is
|
||||
// never dropped or refused for spam: it goes to Junk
|
||||
SpamFilterAction::Discard | SpamFilterAction::Reject
|
||||
if self.data.relayed.as_ref().is_some_and(|r| r.held) =>
|
||||
{
|
||||
trc::event!(
|
||||
Spam(SpamEvent::Classify),
|
||||
SpanId = self.data.session_id,
|
||||
QueueId = message_id,
|
||||
Result = "junk",
|
||||
Reason = "Held by a relay, over the reject threshold: filed as Junk.",
|
||||
);
|
||||
headers.extend_from_slice(b"X-Inbuxa-Relay: held, over reject threshold\r\n");
|
||||
for recipient in self.data.rcpt_to.iter_mut() {
|
||||
recipient.flags = (recipient.flags & !RCPT_SPAM_MASK) | rcpt_spam_flag(100);
|
||||
}
|
||||
let scores = &self.server.core.spam.scores;
|
||||
spam_result = Some((scores.reject_threshold.max(scores.spam_threshold), 100));
|
||||
}
|
||||
SpamFilterAction::Discard => {
|
||||
trc::event!(
|
||||
Spam(SpamEvent::Classify),
|
||||
@@ -1184,8 +1207,13 @@ impl<T: SessionStream> Session<T> {
|
||||
headers.extend_from_slice(b")");
|
||||
}
|
||||
headers.extend_from_slice(b")\r\n\t");
|
||||
if self.stream.is_tls() {
|
||||
let (version, cipher) = self.stream.tls_version_and_cipher();
|
||||
// inbuxa: RL-23: a relayed message's TLS is the relay's with the sender
|
||||
let tls = match self.data.relayed.as_ref().and_then(|r| r.tls.as_ref()) {
|
||||
Some((version, cipher)) => Some((version.clone().into(), cipher.clone().into())),
|
||||
None if self.stream.is_tls() => Some(self.stream.tls_version_and_cipher()),
|
||||
None => None,
|
||||
};
|
||||
if let Some((version, cipher)) = tls {
|
||||
headers.extend_from_slice(b"(using ");
|
||||
headers.extend_from_slice(version.as_bytes());
|
||||
headers.extend_from_slice(b" with cipher ");
|
||||
@@ -1195,7 +1223,7 @@ impl<T: SessionStream> Session<T> {
|
||||
headers.extend_from_slice(b"by ");
|
||||
headers.extend_from_slice(self.hostname.as_bytes());
|
||||
headers.extend_from_slice(concat!(" (", types::brand!(), " SMTP) with ").as_bytes());
|
||||
headers.extend_from_slice(match (self.stream.is_tls(), !self.is_authenticated()) {
|
||||
headers.extend_from_slice(match (self.is_tls(), !self.is_authenticated()) {
|
||||
(true, true) => b"ESMTPS",
|
||||
(true, false) => b"ESMTPSA",
|
||||
(false, true) => b"ESMTP",
|
||||
|
||||
@@ -18,6 +18,7 @@ pub mod mail;
|
||||
pub mod mailflow; // inbuxa: DLP and mail flow rules
|
||||
pub mod milter;
|
||||
pub mod rcpt;
|
||||
pub mod relayed; // inbuxa: RL-23, mail an inbuxa Relay took
|
||||
pub mod session;
|
||||
pub mod spam;
|
||||
pub mod spawn;
|
||||
|
||||
@@ -0,0 +1,489 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs LLC
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! inbuxa: RL-23: a message an inbuxa Relay took on port 25, judged here as
|
||||
//! if this server had taken the SMTP session itself.
|
||||
//!
|
||||
//! The relay's session is played again against this server's own handlers:
|
||||
//! connect, EHLO, MAIL FROM, every RCPT TO, then DATA, with the original
|
||||
//! sender's address, port and HELO. SPF, reverse IP, DMARC, DNSBL, the spam
|
||||
//! filter, the AI classifier, Sieve and the directory all see the real
|
||||
//! facts, and the answer is the one this server would have given.
|
||||
//!
|
||||
//! Only the relay link service calls this, for a relay whose key it pinned.
|
||||
//! Mail arriving on this server's own port 25 is never treated as relayed.
|
||||
|
||||
use crate::{
|
||||
core::{Relayed, Session, SessionData, SessionParameters, State},
|
||||
queue::{
|
||||
Error, ErrorDetails, MessageSource, Recipient, Status, UnexpectedResponse, dsn::SendDsn,
|
||||
spool::SmtpSpool,
|
||||
},
|
||||
};
|
||||
use common::{
|
||||
Server,
|
||||
network::{ServerInstance, stream::NullIo},
|
||||
};
|
||||
use mail_auth::SpfResult;
|
||||
use smtp_proto::{
|
||||
MAIL_RET_FULL, MAIL_RET_HDRS, MailFrom, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE,
|
||||
RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS, RcptTo, Request, Response,
|
||||
request::receiver::RequestReceiver,
|
||||
};
|
||||
use std::{borrow::Cow, net::IpAddr, sync::Arc, time::Duration};
|
||||
use store::write::{BatchBuilder, BlobLink, BlobOp, now};
|
||||
use types::blob_hash::BlobHash;
|
||||
|
||||
/// An address and its SMTP parameters, as the relay passed them on.
|
||||
pub type EnvelopeAddress = (String, Vec<(String, Option<String>)>);
|
||||
|
||||
/// The session a relay took a message in, and the message.
|
||||
pub struct RelayedMessage {
|
||||
pub relayed: Relayed,
|
||||
pub local_ip: IpAddr,
|
||||
pub remote_ip: IpAddr,
|
||||
pub remote_port: u16,
|
||||
pub helo: String,
|
||||
/// The return path and its MAIL FROM parameters.
|
||||
pub mail_from: EnvelopeAddress,
|
||||
/// Each recipient and its RCPT TO parameters.
|
||||
pub rcpt_to: Vec<EnvelopeAddress>,
|
||||
pub message: Vec<u8>,
|
||||
}
|
||||
|
||||
/// What this server answered.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum RelayedOutcome {
|
||||
/// Queued for every recipient but those in `refused`, each with its
|
||||
/// reply.
|
||||
Accepted { refused: Vec<(String, String)> },
|
||||
/// Refused as a whole, with this reply. `refused` has the recipients
|
||||
/// refused one by one before it, if any.
|
||||
Refused {
|
||||
reply: String,
|
||||
refused: Vec<(String, String)>,
|
||||
},
|
||||
}
|
||||
|
||||
/// How long a held message's headers stay stored for its DSN.
|
||||
const DSN_HEADERS_FOR: u64 = 600;
|
||||
|
||||
/// Runs `input` through this server's inbound pipeline. A held message's
|
||||
/// refusals are reported to its sender only when the sender passed SPF or
|
||||
/// DMARC here (RL-18); a live one's whole refusal is the relay's to give.
|
||||
pub async fn ingest_relayed(
|
||||
server: Server,
|
||||
instance: Arc<ServerInstance>,
|
||||
input: RelayedMessage,
|
||||
) -> RelayedOutcome {
|
||||
let session_id = instance.span_id_gen.generate();
|
||||
let asn_geo_data = server.lookup_asn_country(input.remote_ip).await;
|
||||
// Port 25: the relay took it on port 25, and this server's settings
|
||||
// tell port 25 from submission by it
|
||||
let mut data = SessionData::new(
|
||||
input.local_ip,
|
||||
25,
|
||||
input.remote_ip,
|
||||
input.remote_port,
|
||||
asn_geo_data,
|
||||
session_id,
|
||||
);
|
||||
let held = input.relayed.held;
|
||||
data.relayed = Some(input.relayed);
|
||||
let mut session = Session {
|
||||
hostname: String::new(),
|
||||
state: State::None,
|
||||
instance,
|
||||
server,
|
||||
stream: NullIo::default(),
|
||||
data,
|
||||
params: SessionParameters::default(),
|
||||
};
|
||||
|
||||
let mut lines = Lines::default();
|
||||
let outcome = Box::pin(session.play(
|
||||
input.helo,
|
||||
input.mail_from,
|
||||
input.rcpt_to,
|
||||
input.message,
|
||||
&mut lines,
|
||||
))
|
||||
.await;
|
||||
if held {
|
||||
let refused = match &outcome {
|
||||
RelayedOutcome::Accepted { refused } => refused.clone(),
|
||||
RelayedOutcome::Refused { reply, refused } if reply.starts_with('5') => {
|
||||
// Every recipient that got as far as RCPT, refused at once
|
||||
let mut all = refused.clone();
|
||||
all.extend(
|
||||
session
|
||||
.data
|
||||
.rcpt_to
|
||||
.iter()
|
||||
.map(|rcpt| (rcpt.address.clone(), reply.clone())),
|
||||
);
|
||||
all
|
||||
}
|
||||
RelayedOutcome::Refused { .. } => Vec::new(),
|
||||
};
|
||||
if !refused.is_empty() {
|
||||
session.bounce_held(&refused, &lines).await;
|
||||
}
|
||||
}
|
||||
outcome
|
||||
}
|
||||
|
||||
/// The MAIL FROM and RCPT TO commands as parsed, for a DSN.
|
||||
#[derive(Default)]
|
||||
struct Lines {
|
||||
mail_from: Option<MailFrom<String>>,
|
||||
rcpt_to: Vec<RcptTo<String>>,
|
||||
}
|
||||
|
||||
impl Session<NullIo> {
|
||||
async fn play(
|
||||
&mut self,
|
||||
helo: String,
|
||||
mail_from: EnvelopeAddress,
|
||||
rcpt_to: Vec<EnvelopeAddress>,
|
||||
message: Vec<u8>,
|
||||
lines: &mut Lines,
|
||||
) -> RelayedOutcome {
|
||||
let mut refused = Vec::new();
|
||||
let refuse = |reply: String, refused: Vec<(String, String)>| RelayedOutcome::Refused {
|
||||
reply,
|
||||
refused,
|
||||
};
|
||||
|
||||
// Connect: rate limits, then the connect stage's scripts and filters
|
||||
if !self.is_allowed().await {
|
||||
return refuse(
|
||||
"421 4.7.0 Too many connections, try again later.".into(),
|
||||
refused,
|
||||
);
|
||||
}
|
||||
if !self.init_conn().await {
|
||||
let reply = self
|
||||
.has_failed()
|
||||
.unwrap_or_else(|| "421 4.3.0 Try again later.".into());
|
||||
return refuse(reply, refused);
|
||||
}
|
||||
self.stream.tx_buf.clear();
|
||||
|
||||
// EHLO
|
||||
let _ = self.handle_ehlo(Cow::Owned(helo), true).await;
|
||||
if let Some(reply) = self.has_failed() {
|
||||
return refuse(reply, refused);
|
||||
}
|
||||
|
||||
// MAIL FROM
|
||||
let line = command("MAIL FROM", &mail_from);
|
||||
let mut receiver = RequestReceiver::default();
|
||||
let Ok(Request::Mail { from }) = receiver.ingest(&mut line.iter()) else {
|
||||
return refuse("501 5.5.4 Invalid MAIL FROM.".into(), refused);
|
||||
};
|
||||
lines.mail_from = Some(from.clone().into_owned());
|
||||
let _ = self.handle_mail_from(from).await;
|
||||
if let Some(reply) = self.has_failed() {
|
||||
return refuse(reply, refused);
|
||||
}
|
||||
|
||||
// RCPT TO: the relay already checked each recipient (RL-1); this
|
||||
// server checks again, by its own directory and rules
|
||||
self.params.rcpt_errors_wait = Duration::ZERO;
|
||||
for (address, parameters) in rcpt_to {
|
||||
let line = command("RCPT TO", &(address.clone(), parameters));
|
||||
let mut receiver = RequestReceiver::default();
|
||||
let Ok(Request::Rcpt { mut to }) = receiver.ingest(&mut line.iter()) else {
|
||||
refused.push((address, "501 5.5.4 Invalid RCPT TO.".into()));
|
||||
continue;
|
||||
};
|
||||
if !self.params.rcpt_dsn {
|
||||
// The relay offers DSN to everyone; this server may not
|
||||
to.flags &= !(RCPT_NOTIFY_DELAY
|
||||
| RCPT_NOTIFY_FAILURE
|
||||
| RCPT_NOTIFY_NEVER
|
||||
| RCPT_NOTIFY_SUCCESS);
|
||||
to.orcpt = None;
|
||||
}
|
||||
lines.rcpt_to.push(to.clone().into_owned());
|
||||
let _ = self.handle_rcpt_to(to).await;
|
||||
if let Some(reply) = self.has_failed() {
|
||||
refused.push((address, reply));
|
||||
}
|
||||
}
|
||||
if self.data.rcpt_to.is_empty() {
|
||||
// The reply the sender would have been given: a temporary one
|
||||
// if any recipient had one, so it tries again
|
||||
let reply = refused
|
||||
.iter()
|
||||
.map(|(_, reply)| reply)
|
||||
.find(|reply| reply.starts_with('4'))
|
||||
.or(refused.first().map(|(_, reply)| reply))
|
||||
.cloned()
|
||||
.unwrap_or_else(|| "554 5.5.1 No valid recipients.".into());
|
||||
return refuse(reply, refused);
|
||||
}
|
||||
|
||||
// DATA
|
||||
if let Some(reply) = self.can_send_data().await {
|
||||
return refuse(reply_text(reply), refused);
|
||||
}
|
||||
if message.len() >= self.params.max_message_size {
|
||||
return refuse("552 5.3.4 Message too big for system.".into(), refused);
|
||||
}
|
||||
self.data.message = message;
|
||||
let reply = self.queue_message().await;
|
||||
let reply = reply_text(&reply);
|
||||
if reply.starts_with('2') {
|
||||
RelayedOutcome::Accepted { refused }
|
||||
} else {
|
||||
refuse(reply, refused)
|
||||
}
|
||||
}
|
||||
|
||||
/// RL-18: a DSN for recipients refused after the relay had said 250,
|
||||
/// only when the sender passed SPF or DMARC here, so a forged sender
|
||||
/// gets none and the relay is no backscatter source.
|
||||
async fn bounce_held(&mut self, refused: &[(String, String)], lines: &Lines) {
|
||||
let Some(mail_from) = self.data.mail_from.as_ref() else {
|
||||
return;
|
||||
};
|
||||
let spf_pass = self
|
||||
.data
|
||||
.spf_mail_from
|
||||
.as_ref()
|
||||
.is_some_and(|spf| spf.result() == SpfResult::Pass);
|
||||
let dmarc_pass = self.data.relayed.as_ref().is_some_and(|r| r.dmarc_pass);
|
||||
if mail_from.address_lcase.is_empty() || !(spf_pass || dmarc_pass) {
|
||||
trc::event!(
|
||||
Delivery(trc::DeliveryEvent::DoubleBounce),
|
||||
SpanId = self.data.session_id,
|
||||
From = mail_from.address_lcase.clone(),
|
||||
Reason = "Held by a relay and refused here; the sender passed neither SPF nor DMARC, so no DSN.",
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
let mut wrapper = self.server.new_message(
|
||||
mail_from.address.as_str(),
|
||||
MessageSource::Unauthenticated { dmarc_pass },
|
||||
self.data.session_id,
|
||||
);
|
||||
if let Some(from) = &lines.mail_from {
|
||||
wrapper.message.env_id = from.env_id.as_deref().map(Into::into);
|
||||
wrapper.message.flags |= from.flags & (MAIL_RET_FULL | MAIL_RET_HDRS);
|
||||
}
|
||||
wrapper.message.received_from_ip = self.data.remote_ip;
|
||||
wrapper.message.received_via_port = 25;
|
||||
for (address, reply) in refused {
|
||||
let line = lines
|
||||
.rcpt_to
|
||||
.iter()
|
||||
.find(|to| to.address.eq_ignore_ascii_case(address));
|
||||
let flags = line.map_or(0, |to| to.flags);
|
||||
let flags = if flags
|
||||
& (RCPT_NOTIFY_DELAY
|
||||
| RCPT_NOTIFY_FAILURE
|
||||
| RCPT_NOTIFY_NEVER
|
||||
| RCPT_NOTIFY_SUCCESS)
|
||||
== 0
|
||||
{
|
||||
RCPT_NOTIFY_FAILURE
|
||||
} else {
|
||||
flags
|
||||
};
|
||||
let mut recipient = Recipient::new(address).with_flags(flags);
|
||||
recipient.orcpt = line.and_then(|to| to.orcpt.as_deref()).map(Into::into);
|
||||
recipient.status = Status::PermanentFailure(ErrorDetails {
|
||||
entity: self.hostname.as_str().into(),
|
||||
details: Error::UnexpectedResponse(UnexpectedResponse {
|
||||
command: format!("RCPT TO:<{address}>").into_boxed_str(),
|
||||
response: parse_response(reply),
|
||||
}),
|
||||
});
|
||||
wrapper.message.recipients.push(recipient);
|
||||
}
|
||||
|
||||
// The DSN quotes the message's headers from the blob store
|
||||
let message = &self.data.message;
|
||||
let headers = &message[..header_end(message)];
|
||||
wrapper.message.blob_hash = BlobHash::generate(headers);
|
||||
wrapper.message.size = message.len() as u64;
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch.set(
|
||||
BlobOp::Link {
|
||||
hash: wrapper.message.blob_hash.clone(),
|
||||
to: BlobLink::Temporary {
|
||||
until: now() + DSN_HEADERS_FOR,
|
||||
},
|
||||
},
|
||||
vec![],
|
||||
);
|
||||
let stored = match self.server.store().write(batch.build_all()).await {
|
||||
Ok(_) => self
|
||||
.server
|
||||
.blob_store()
|
||||
.put_blob(
|
||||
wrapper.message.blob_hash.as_slice(),
|
||||
headers,
|
||||
self.server.core.email.compression,
|
||||
)
|
||||
.await
|
||||
.map(|_| ()),
|
||||
Err(err) => Err(err),
|
||||
};
|
||||
if let Err(err) = stored {
|
||||
trc::error!(
|
||||
err.details("Failed to store a relayed message's headers for its DSN.")
|
||||
.span_id(self.data.session_id)
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
return;
|
||||
}
|
||||
self.server.send_dsn(&mut wrapper).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// `MAIL FROM:<address> KEY=value...` or the RCPT TO equivalent. Anything
|
||||
/// that could end the line or start another parameter is dropped.
|
||||
fn command(verb: &str, (address, parameters): &EnvelopeAddress) -> Vec<u8> {
|
||||
let clean = |text: &str| -> String {
|
||||
text.chars()
|
||||
.filter(|c| !c.is_control() && !matches!(c, ' ' | '<' | '>'))
|
||||
.collect()
|
||||
};
|
||||
let mut line = format!("{verb}:<{}>", clean(address));
|
||||
for (key, value) in parameters {
|
||||
let key: String = clean(key).replace('=', "");
|
||||
if key.is_empty() {
|
||||
continue;
|
||||
}
|
||||
line.push(' ');
|
||||
line.push_str(&key);
|
||||
if let Some(value) = value {
|
||||
line.push('=');
|
||||
line.push_str(&clean(value));
|
||||
}
|
||||
}
|
||||
line.push_str("\r\n");
|
||||
line.into_bytes()
|
||||
}
|
||||
|
||||
/// A reply as text, its last line if it has several.
|
||||
fn reply_text(reply: &[u8]) -> String {
|
||||
let text = String::from_utf8_lossy(reply);
|
||||
text.trim_end()
|
||||
.rsplit("\r\n")
|
||||
.next()
|
||||
.unwrap_or_default()
|
||||
.chars()
|
||||
.take(512)
|
||||
.collect::<String>()
|
||||
.trim()
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// `550 5.1.1 text` as a response.
|
||||
fn parse_response(reply: &str) -> Response<Box<str>> {
|
||||
let mut parts = reply.splitn(3, ' ');
|
||||
let code = parts.next().and_then(|c| c.parse().ok()).unwrap_or(550);
|
||||
let (esc, message) = match parts.next() {
|
||||
Some(esc) if enhanced(esc).is_some() => (
|
||||
enhanced(esc).unwrap_or([5, 0, 0]),
|
||||
parts.next().unwrap_or_default(),
|
||||
),
|
||||
other => (
|
||||
[(code / 100) as u8, 0, 0],
|
||||
reply
|
||||
.split_once(' ')
|
||||
.map(|(_, rest)| rest)
|
||||
.filter(|_| other.is_some())
|
||||
.unwrap_or_default(),
|
||||
),
|
||||
};
|
||||
Response {
|
||||
code,
|
||||
esc,
|
||||
message: message.into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn enhanced(text: &str) -> Option<[u8; 3]> {
|
||||
let mut parts = text.split('.').map(|p| p.parse::<u8>().ok());
|
||||
let esc = [parts.next()??, parts.next()??, parts.next()??];
|
||||
parts.next().is_none().then_some(esc)
|
||||
}
|
||||
|
||||
/// Where the headers end: after the blank line, or the whole message.
|
||||
fn header_end(message: &[u8]) -> usize {
|
||||
message
|
||||
.windows(4)
|
||||
.position(|w| w == b"\r\n\r\n")
|
||||
.map_or(message.len(), |pos| pos + 4)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn builds_commands() {
|
||||
let line = command(
|
||||
"RCPT TO",
|
||||
&(
|
||||
"[email protected]".into(),
|
||||
vec![
|
||||
("NOTIFY".into(), Some("FAILURE,DELAY".into())),
|
||||
("ORCPT".into(), Some("rfc822;[email protected]".into())),
|
||||
],
|
||||
),
|
||||
);
|
||||
assert_eq!(
|
||||
line,
|
||||
b"RCPT TO:<[email protected]> NOTIFY=FAILURE,DELAY ORCPT=rfc822;[email protected]\r\n"
|
||||
);
|
||||
let mut receiver = RequestReceiver::default();
|
||||
let Ok(Request::Rcpt { to }) = receiver.ingest(&mut line.iter()) else {
|
||||
panic!("not parsed");
|
||||
};
|
||||
assert_eq!(to.address, "[email protected]");
|
||||
assert_ne!(to.flags & RCPT_NOTIFY_FAILURE, 0);
|
||||
|
||||
// Nothing can end the line or add a parameter
|
||||
let line = command(
|
||||
"MAIL FROM",
|
||||
&(
|
||||
"[email protected]> AUTH=x\r\nRCPT TO:<[email protected]".into(),
|
||||
vec![("BODY=X ".into(), Some("8BIT MIME\r\n".into()))],
|
||||
),
|
||||
);
|
||||
assert_eq!(
|
||||
line,
|
||||
b"MAIL FROM:<[email protected]=xRCPTTO:[email protected]> BODYX=8BITMIME\r\n"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reads_replies() {
|
||||
assert_eq!(
|
||||
reply_text(b"550-5.7.1 First\r\n550 5.7.1 Message rejected.\r\n"),
|
||||
"550 5.7.1 Message rejected."
|
||||
);
|
||||
let response = parse_response("550 5.1.1 Mailbox unavailable");
|
||||
assert_eq!(
|
||||
(response.code, response.esc, response.message.as_ref()),
|
||||
(550, [5, 1, 1], "Mailbox unavailable")
|
||||
);
|
||||
let response = parse_response("451 try later");
|
||||
assert_eq!(
|
||||
(response.code, response.esc, response.message.as_ref()),
|
||||
(451, [4, 0, 0], "try later")
|
||||
);
|
||||
assert_eq!(header_end(b"A: b\r\n\r\nbody"), 8);
|
||||
assert_eq!(header_end(b"A: b"), 4);
|
||||
}
|
||||
}
|
||||
@@ -29,6 +29,12 @@ use crate::core::{Session, State};
|
||||
use super::auth::{AUTH_BEARER, SaslToken};
|
||||
|
||||
impl<T: SessionStream> Session<T> {
|
||||
/// inbuxa: TLS on this session, or between the sender and the relay
|
||||
/// that took it (RL-23).
|
||||
pub fn is_tls(&self) -> bool {
|
||||
self.stream.is_tls() || self.data.relayed.as_ref().is_some_and(|r| r.tls.is_some())
|
||||
}
|
||||
|
||||
pub async fn ingest(&mut self, bytes: &[u8]) -> Result<bool, ()> {
|
||||
let mut iter = bytes.iter();
|
||||
let mut state = std::mem::replace(&mut self.state, State::None);
|
||||
@@ -623,7 +629,7 @@ impl<T: SessionStream> ResolveVariable for Session<T> {
|
||||
ExpressionVariable::RemotePort => self.data.remote_port.into(),
|
||||
ExpressionVariable::LocalIp => self.data.local_ip_str.as_str().into(),
|
||||
ExpressionVariable::LocalPort => self.data.local_port.into(),
|
||||
ExpressionVariable::IsTls => self.stream.is_tls().into(),
|
||||
ExpressionVariable::IsTls => self.is_tls().into(),
|
||||
ExpressionVariable::Priority => self.data.priority.to_compact_string().into(),
|
||||
ExpressionVariable::Protocol => self.instance.protocol.as_str().into(),
|
||||
ExpressionVariable::Asn => self
|
||||
@@ -642,6 +648,13 @@ impl<T: SessionStream> ResolveVariable for Session<T> {
|
||||
.map(|c| c.as_str())
|
||||
.unwrap_or_default()
|
||||
.into(),
|
||||
ExpressionVariable::Relay => self
|
||||
.data
|
||||
.relayed
|
||||
.as_ref()
|
||||
.map(|r| r.relay.as_str())
|
||||
.unwrap_or_default()
|
||||
.into(),
|
||||
_ => expr::Variable::default(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::core::Session;
|
||||
@@ -70,7 +72,7 @@ impl<T: SessionStream> Session<T> {
|
||||
authenticated_as: self.data.authenticated_as.as_ref().map(|a| a.name()),
|
||||
asn: self.data.asn_geo_data.asn.as_ref().map(|a| a.id),
|
||||
country: self.data.asn_geo_data.country.as_ref().map(|c| c.as_str()),
|
||||
is_tls: self.stream.is_tls(),
|
||||
is_tls: self.is_tls(),
|
||||
env_from: self
|
||||
.data
|
||||
.mail_from
|
||||
|
||||
@@ -227,6 +227,12 @@ so a reboot without the link comes back with the same settings (RL-22).
|
||||
stop someone holding the relay's key from testing guessed addresses
|
||||
offline; nothing on the relay could, since the relay has to check real
|
||||
ones.
|
||||
- A domain whose recipients an external directory (LDAP, SQL, OIDC)
|
||||
answers for can't be listed. Its recipient set holds the accounts the
|
||||
downstream has already seen from that directory, and nothing more: no
|
||||
catch-all, which would break RL-1. Anyone else there is refused at the
|
||||
relay until they first sign in (decided 2026-10-29). A live lookup over
|
||||
the link may come later.
|
||||
- Deltas use `Recipient/set` (create and destroy by hash). A full refresh
|
||||
uses the custom `Recipient/replace {blobId}`: a sorted file of 32-byte
|
||||
hashes, swapped in atomically. The relay's `Recipient` state string lets
|
||||
@@ -238,8 +244,8 @@ so a reboot without the link comes back with the same settings (RL-22).
|
||||
|---|---|
|
||||
| `id` | Stable for the message's whole life. The downstream deduplicates on it (RL-16) |
|
||||
| `receivedAt` | When the relay accepted it (or got the final dot, if pending) |
|
||||
| `state` | `pending` (live, awaiting a verdict), `held`, or `claimed` |
|
||||
| `claimedUntil` | End of the current lease, when claimed |
|
||||
| `state` | `pending` (live, awaiting a verdict, claimed or not), `held`, or `claimed` (held and leased) |
|
||||
| `claimedUntil` | End of the current lease, when leased |
|
||||
| `envelope` | RFC 8621 `Envelope`: `mailFrom` and `rcptTo`, with parameters |
|
||||
| `session` | `{remoteIp, remotePort, localIp, helo, tls: {version, cipher} or null, ptr}` |
|
||||
| `size` | Bytes |
|
||||
@@ -675,8 +681,17 @@ In progress. Done so far:
|
||||
recipient set (whole, then deltas), and reports each node's link in the
|
||||
relay's `links` (PR #253). Node link certificates last 30 days and are
|
||||
renewed after 20.
|
||||
- Not done yet: ingest, live verdicts, `inbuxa:RelayMessage`, relay events
|
||||
in the trace, and the link-down alert.
|
||||
- Ingest and live verdicts: the link service asks for `Inbound` push, so
|
||||
the relay is live while a node is linked. Each message is claimed,
|
||||
unsealed, and played through the downstream's own SMTP handlers with
|
||||
the relay's session on port 25 (RL-23), with `relay` in expressions.
|
||||
Live, the downstream's reply is the verdict; held, a refusal for good is
|
||||
reported only to a sender who passed SPF or DMARC (RL-18), one for now
|
||||
is retried after the lease, and spam over the reject threshold goes to
|
||||
Junk (RL-17). Every message taken is recorded until `heldMaxAge` and a
|
||||
day have passed (RL-16). A pending message stays `pending` when claimed.
|
||||
- Not done yet: `inbuxa:RelayMessage`, relay events in the trace, and the
|
||||
link-down alert.
|
||||
|
||||
Order:
|
||||
|
||||
|
||||
Binary file not shown.
@@ -1 +1 @@
|
||||
Hc6Z_iXa_4ma9Pch8CLDiLJ7nuLZC_hz5_jqi4Dp2x4
|
||||
B5ARrqKFTX10Ib2fIAsGacr8yxF4NyI3IwmJEPFiTP8
|
||||
+148
-3
@@ -13,7 +13,10 @@
|
||||
//!
|
||||
//! `relay_link_tests` runs a real `inbuxa-relay` (build it first with
|
||||
//! `cargo build -p inbuxa-relay`): the server joins it, sets it up, and its
|
||||
//! port 25 takes mail only for this server's real recipients.
|
||||
//! port 25 takes mail only for this server's real recipients. Live, a
|
||||
//! message is delivered before the sender hears 250, and this server's
|
||||
//! refusal is the sender's reply; held while the link is off, it is
|
||||
//! delivered once the link is back, exactly once.
|
||||
|
||||
use crate::utils::{
|
||||
account::Account,
|
||||
@@ -450,6 +453,102 @@ async fn rcpt_replies(port: u16, recipients: &[&str]) -> Vec<String> {
|
||||
replies.split_off(2)
|
||||
}
|
||||
|
||||
/// Sends one message through the relay; the reply to its final dot.
|
||||
async fn send_mail(port: u16, rcpt: &str, message: &str) -> String {
|
||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||
let stream = tokio::net::TcpStream::connect(("127.0.0.1", port))
|
||||
.await
|
||||
.unwrap();
|
||||
let (read, mut write) = stream.into_split();
|
||||
let mut read = BufReader::new(read);
|
||||
async fn reply(read: &mut BufReader<tokio::net::tcp::OwnedReadHalf>) -> String {
|
||||
let mut all = String::new();
|
||||
loop {
|
||||
let mut line = String::new();
|
||||
read.read_line(&mut line).await.unwrap();
|
||||
all.push_str(&line);
|
||||
if line.len() < 4 || line.as_bytes()[3] != b'-' {
|
||||
return all;
|
||||
}
|
||||
}
|
||||
}
|
||||
assert!(reply(&mut read).await.starts_with("220"));
|
||||
for line in [
|
||||
"EHLO sender.example".to_string(),
|
||||
"MAIL FROM:<[email protected]>".to_string(),
|
||||
format!("RCPT TO:<{rcpt}>"),
|
||||
"DATA".to_string(),
|
||||
] {
|
||||
write
|
||||
.write_all(format!("{line}\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
let answer = reply(&mut read).await;
|
||||
assert!(
|
||||
answer.starts_with('2') || answer.starts_with('3'),
|
||||
"{line}: {answer}"
|
||||
);
|
||||
}
|
||||
let body = message.replace('\n', "\r\n");
|
||||
write
|
||||
.write_all(format!("{body}\r\n.\r\n").as_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
let answer = reply(&mut read).await;
|
||||
let _ = write.write_all(b"QUIT\r\n").await;
|
||||
answer
|
||||
}
|
||||
|
||||
/// The subjects of every message in `account`'s mailboxes.
|
||||
async fn subjects(account: &Account) -> Vec<String> {
|
||||
let response = account
|
||||
.jmap_request(
|
||||
&["urn:ietf:params:jmap:core", "urn:ietf:params:jmap:mail"],
|
||||
json!([
|
||||
["Email/query", {"accountId": account.id_string()}, "q"],
|
||||
["Email/get", {
|
||||
"accountId": account.id_string(),
|
||||
"#ids": {"resultOf": "q", "name": "Email/query", "path": "/ids"},
|
||||
"properties": ["subject"]
|
||||
}, "g"]
|
||||
]),
|
||||
)
|
||||
.await;
|
||||
response
|
||||
.0
|
||||
.pointer("/methodResponses/1/1/list")
|
||||
.and_then(Value::as_array)
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.filter_map(|email| email["subject"].as_str().map(str::to_string))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Waits until `account` has `count` messages titled `subject`.
|
||||
async fn wait_for_mail(account: &Account, subject: &str, count: usize) {
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
|
||||
loop {
|
||||
let found = subjects(account).await;
|
||||
let n = found.iter().filter(|s| *s == subject).count();
|
||||
if n == count {
|
||||
return;
|
||||
}
|
||||
assert!(
|
||||
n < count && std::time::Instant::now() < deadline,
|
||||
"{count} of {subject:?}: {found:?}"
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||
}
|
||||
}
|
||||
|
||||
fn message(subject: &str, extra_headers: &str) -> String {
|
||||
format!(
|
||||
"{extra_headers}From: Someone <[email protected]>\nTo: [email protected]\n\
|
||||
Subject: {subject}\nMessage-ID: <{}@sender.example>\n\nHello from outside.",
|
||||
subject.replace(' ', "-")
|
||||
)
|
||||
}
|
||||
|
||||
pub async fn link(test: &mut TestServer) {
|
||||
println!("Running inbuxa Relay link tests...");
|
||||
let admin = test.account("[email protected]");
|
||||
@@ -459,7 +558,7 @@ pub async fn link(test: &mut TestServer) {
|
||||
"{} is missing: run cargo build -p inbuxa-relay",
|
||||
binary.display()
|
||||
);
|
||||
admin
|
||||
let jane = admin
|
||||
.create_user_account(
|
||||
"[email protected]",
|
||||
"jane-secret-4417",
|
||||
@@ -525,7 +624,7 @@ pub async fn link(test: &mut TestServer) {
|
||||
.as_array()
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.any(|link| link["state"] == "linked" && link["relay"]["mode"] == "held")
|
||||
.any(|link| link["state"] == "linked" && link["relay"]["mode"] == "live")
|
||||
})
|
||||
.await;
|
||||
assert!(relay["linkKeyPin"].as_str().is_some(), "{relay}");
|
||||
@@ -556,6 +655,27 @@ pub async fn link(test: &mut TestServer) {
|
||||
"not served: {replies:?}"
|
||||
);
|
||||
|
||||
// Live (RL-12): delivered here before the sender hears 250
|
||||
let reply = send_mail(smtp_port, "[email protected]", &message("Live one", "")).await;
|
||||
assert!(reply.starts_with("250"), "{reply}");
|
||||
wait_for_mail(&jane, "Live one", 1).await;
|
||||
|
||||
// Live: this server's refusal is the sender's reply, and the relay
|
||||
// keeps nothing
|
||||
let loop_headers =
|
||||
"Received: from a.example by b.example; Thu, 29 Oct 2026 10:00:00 +0000\n".repeat(60);
|
||||
let reply = send_mail(
|
||||
smtp_port,
|
||||
"[email protected]",
|
||||
&message("Looped one", &loop_headers),
|
||||
)
|
||||
.await;
|
||||
assert!(reply.starts_with("450 4.4.6"), "{reply}");
|
||||
wait_for(admin, &id, "the relay kept nothing", 30, |relay| {
|
||||
relay["links"][0]["relay"]["spool"]["inbound"] == 0
|
||||
})
|
||||
.await;
|
||||
|
||||
// A new account reaches the relay as a change
|
||||
admin
|
||||
.create_user_account("[email protected]", "joe-secret-5529", "Joe", &[], vec![])
|
||||
@@ -606,6 +726,31 @@ pub async fn link(test: &mut TestServer) {
|
||||
})
|
||||
.await;
|
||||
|
||||
// Held while the link is off: the relay says 250 itself, and the
|
||||
// message is delivered once the link is back, once
|
||||
let reply = send_mail(smtp_port, "[email protected]", &message("Held one", "")).await;
|
||||
assert!(reply.starts_with("250"), "{reply}");
|
||||
call(
|
||||
admin,
|
||||
"inbuxa:Relay/set",
|
||||
json!({"update": {&id: {"enabled": true}}}),
|
||||
)
|
||||
.await;
|
||||
wait_for_mail(&jane, "Held one", 1).await;
|
||||
wait_for(admin, &id, "the held message was taken", 30, |relay| {
|
||||
relay["links"][0]["state"] == "linked"
|
||||
&& relay["links"][0]["relay"]["spool"]["inbound"] == 0
|
||||
})
|
||||
.await;
|
||||
assert_eq!(
|
||||
subjects(&jane)
|
||||
.await
|
||||
.iter()
|
||||
.filter(|s| *s == "Held one" || *s == "Live one")
|
||||
.count(),
|
||||
2
|
||||
);
|
||||
|
||||
call(admin, "inbuxa:Relay/set", json!({"destroy": [&id]})).await;
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user