filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

DKIMv2 signing

Browse Source
This commit is contained in:
Maurus Decimus
2026-07-03 16:51:57 +02:00
parent 186d9feecb
commit c5796b4e7a
20 changed files with 588 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
crates/smtp/src/queue/dsn.rs
View File

@@ -9,8 +9,9 @@ use super::{
Error, ErrorDetails, HostResponse, Message, MessageSource, QueueEnvelope, RCPT_DSN_SENT,
Recipient, Status,
};
use crate::inbound::dkim::DkimSign;
use crate::queue::spool::QueueParams;
use crate::queue::{MessageWrapper, UnexpectedResponse};
use crate::reporting::send::MtaReportSend;
use common::Server;
use mail_builder::MessageBuilder;
use mail_builder::headers::HeaderType;
@@ -42,19 +43,18 @@ impl SendDsn for Server {
.expand_and_add_recipient(message.message.return_path.as_ref(), self)
.await;
// Sign message
let signature = self
.sign_message(message, &self.core.smtp.queue.dsn.sign, &dsn)
.await;
// Queue DSN
let dkim_signers = self
.eval_signers(
&self.core.smtp.queue.dsn.sign,
&message.message,
message.span_id,
)
.await;
dsn_message
.queue(
signature.as_deref(),
&dsn,
message.span_id,
self,
MessageSource::Dsn,
QueueParams::new(&dsn, message.span_id, self, MessageSource::Dsn)
.with_dkim_signers(dkim_signers),
)
.await;
}

11
crates/smtp/src/queue/mod.rs
View File

@@ -69,7 +69,7 @@ pub struct Message {
pub priority: i16,
pub size: u64,
pub quota_keys: Box<[QuotaKey]>,
pub metadata: Box<[Metadata]>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -91,9 +91,10 @@ pub struct MessageWrapper {
Eq,
serde::Deserialize,
)]
pub enum QuotaKey {
Size { key: Box<[u8]>, id: u64 },
Count { key: Box<[u8]>, id: u64 },
pub enum Metadata {
QueueSize { key: Box<[u8]>, id: u64 },
QueueCount { key: Box<[u8]>, id: u64 },
Headers { value: Box<[u8]>, id: u64 },
}
#[derive(
@@ -127,7 +128,7 @@ pub const FROM_REPORT: u64 = 1 << 36;
pub const FROM_AUTOGENERATED: u64 = 1 << 37;
pub const RCPT_DSN_SENT: u64 = 1 << 32;
//pub const RCPT_STATUS_CHANGED: u64 = 1 << 33;
//pub const RCPT_UNDISCLOSED: u64 = 1 << 33;
pub const RCPT_SPAM_PAYLOAD: u64 = 1 << 34;
#[derive(

48
crates/smtp/src/queue/quota.rs
View File

@@ -4,7 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use super::{QueueEnvelope, QuotaKey, Status};
use super::{Metadata, QueueEnvelope, Status};
use crate::{core::throttle::NewKey, queue::MessageWrapper};
use ahash::AHashSet;
use common::{Server, config::smtp::queue::QueueQuota, expr::functions::ResolveVariable};
@@ -18,20 +18,23 @@ use trc::QueueEvent;
use utils::DomainPart;
pub trait HasQueueQuota: Sync + Send {
fn has_quota(&self, message: &mut MessageWrapper) -> impl Future<Output = bool> + Send;
fn has_quota(
&self,
message: &mut MessageWrapper,
) -> impl Future<Output = Option<Vec<Metadata>>> + Send;
fn check_quota<'x>(
&'x self,
quota: &'x QueueQuota,
envelope: &impl ResolveVariable,
size: u64,
id: u64,
refs: &mut Vec<QuotaKey>,
refs: &mut Vec<Metadata>,
session_id: u64,
) -> impl Future<Output = bool> + Send;
}
impl HasQueueQuota for Server {
async fn has_quota(&self, message: &mut MessageWrapper) -> bool {
async fn has_quota(&self, message: &mut MessageWrapper) -> Option<Vec<Metadata>> {
let mut quota_keys = Vec::new();
if !self.core.smtp.queue.quota.sender.is_empty() {
@@ -54,7 +57,7 @@ impl HasQueueQuota for Server {
Type = "Sender"
);
return false;
return None;
}
}
}
@@ -82,7 +85,7 @@ impl HasQueueQuota for Server {
Type = "Domain"
);
return false;
return None;
}
}
}
@@ -108,14 +111,12 @@ impl HasQueueQuota for Server {
Type = "Recipient"
);
return false;
return None;
}
}
}
message.message.quota_keys = quota_keys.into_boxed_slice();
true
Some(quota_keys)
}
async fn check_quota<'x>(
@@ -124,7 +125,7 @@ impl HasQueueQuota for Server {
envelope: &impl ResolveVariable,
size: u64,
id: u64,
refs: &mut Vec<QuotaKey>,
refs: &mut Vec<Metadata>,
session_id: u64,
) -> bool {
if !quota.expr.is_empty()
@@ -147,7 +148,7 @@ impl HasQueueQuota for Server {
if used_size + size > max_size {
return false;
} else {
refs.push(QuotaKey::Size {
refs.push(Metadata::QueueSize {
key: key.as_ref().into(),
id,
});
@@ -167,7 +168,7 @@ impl HasQueueQuota for Server {
if total_messages + 1 > max_messages {
return false;
} else {
refs.push(QuotaKey::Count {
refs.push(Metadata::QueueCount {
key: key.as_ref().into(),
id,
});
@@ -180,7 +181,12 @@ impl HasQueueQuota for Server {
impl MessageWrapper {
pub fn release_quota(&mut self, batch: &mut BatchBuilder) {
if self.message.quota_keys.is_empty() {
if !self.message.metadata.iter().any(|metadata| {
matches!(
metadata,
Metadata::QueueSize { .. } | Metadata::QueueCount { .. }
)
}) {
return;
}
let mut quota_ids = Vec::with_capacity(self.message.recipients.len());
@@ -199,27 +205,27 @@ impl MessageWrapper {
}
if !quota_ids.is_empty() {
let mut quota_keys = Vec::new();
for quota_key in std::mem::take(&mut self.message.quota_keys) {
match quota_key {
QuotaKey::Count { id, key } if quota_ids.contains(&id) => {
let mut metadata = Vec::new();
for entry in std::mem::take(&mut self.message.metadata) {
match entry {
Metadata::QueueCount { id, key } if quota_ids.contains(&id) => {
batch.add(
ValueClass::Queue(QueueClass::QuotaCount(key.into_vec())),
-1,
);
}
QuotaKey::Size { id, key } if quota_ids.contains(&id) => {
Metadata::QueueSize { id, key } if quota_ids.contains(&id) => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key.into_vec())),
-(self.message.size as i64),
);
}
_ => {
quota_keys.push(quota_key);
metadata.push(entry);
}
}
}
self.message.quota_keys = quota_keys.into_boxed_slice();
self.message.metadata = metadata.into_boxed_slice();
}
}
}

141
crates/smtp/src/queue/spool.rs
View File

@@ -5,19 +5,22 @@
*/
use super::{
ArchivedMessage, ArchivedStatus, Message, MessageSource, QueueEnvelope, QueueId, QueuedMessage,
QuotaKey, Recipient, Schedule, Status,
ArchivedMessage, ArchivedStatus, Message, MessageSource, Metadata, QueueEnvelope, QueueId,
QueuedMessage, Recipient, Schedule, Status,
};
use crate::inbound::dkim::DkimSign;
use crate::queue::manager::{LockedMessage, Queue};
use crate::queue::{
FROM_AUTHENTICATED, FROM_AUTOGENERATED, FROM_DSN, FROM_REPORT, FROM_UNAUTHENTICATED,
FROM_UNAUTHENTICATED_DMARC, MessageWrapper,
};
use ahash::{AHashMap, AHashSet};
use common::config::smtp::auth::DkimSigners;
use common::config::smtp::queue::{ArchivedQueueExpiry, QueueName};
use common::ipc::QueueEvent;
use common::network::RcptResolution;
use common::{KV_LOCK_QUEUE_MESSAGE, Server};
use mail_auth::AuthenticatedMessage;
use registry::schema::prelude::{ObjectType, Property};
use registry::schema::structs::SpamTrainingSample;
use registry::types::datetime::UTCDateTime;
@@ -27,6 +30,7 @@ use std::borrow::Cow;
use std::collections::hash_map::Entry;
use std::future::Future;
use std::net::{IpAddr, Ipv4Addr};
use std::sync::Arc;
use std::time::SystemTime;
use store::write::key::DeserializeBigEndian;
use store::write::serialize::rkyv_deserialize;
@@ -100,7 +104,7 @@ impl SmtpSpool for Server {
priority: 0,
size: 0,
blob_hash: Default::default(),
quota_keys: Default::default(),
metadata: Default::default(),
received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
received_via_port: 0,
},
@@ -308,15 +312,38 @@ fn lock_id(queue_id: QueueId, queue_name: QueueName) -> [u8; 16] {
id
}
pub(crate) struct QueueParams<'x, 'y> {
pub raw_message: &'x [u8],
pub raw_headers: Option<&'x [u8]>,
pub metadata: Vec<Metadata>,
pub original_raw_message: Option<&'x [u8]>,
pub original_authenticated_message: Option<AuthenticatedMessage<'x>>,
pub dkim_signers: Option<Arc<DkimSigners>>,
pub session_id: u64,
pub server: &'y Server,
pub source: MessageSource,
}
impl MessageWrapper {
pub async fn queue(
mut self,
raw_headers: Option<&[u8]>,
raw_message: &[u8],
session_id: u64,
server: &Server,
source: MessageSource,
) -> bool {
pub(crate) async fn queue<'x, 'y>(mut self, mut params: QueueParams<'x, 'y>) -> bool {
// Add DKIM signatures
let dkim_headers = if params.dkim_signers.is_some() {
params.server.sign_message(&mut self, &mut params).await
} else {
None
};
// Fetch params
let QueueParams {
raw_message,
raw_headers,
session_id,
server,
source,
metadata,
..
} = params;
// Set flags
let (flags, event, train_spam) = match source {
MessageSource::Authenticated => (
@@ -351,8 +378,12 @@ impl MessageWrapper {
self.message.flags |= flags;
// Write blob
let message = if let Some(raw_headers) = raw_headers {
let mut message = Vec::with_capacity(raw_headers.len() + raw_message.len());
let raw_headers = raw_headers.unwrap_or_default();
let dkim_headers = dkim_headers.as_deref().unwrap_or_default();
let message = if !raw_headers.is_empty() || !dkim_headers.is_empty() {
let mut message =
Vec::with_capacity(raw_headers.len() + dkim_headers.len() + raw_message.len());
message.extend_from_slice(dkim_headers);
message.extend_from_slice(raw_headers);
message.extend_from_slice(raw_message);
Cow::Owned(message)
@@ -361,10 +392,11 @@ impl MessageWrapper {
};
self.message.blob_hash = BlobHash::generate(message.as_ref());
// Generate id
// Update size
if self.message.size == 0 {
self.message.size = message.len() as u64;
}
self.message.metadata = metadata.into_boxed_slice();
// Reserve and write blob
let mut batch = BatchBuilder::new();
@@ -434,17 +466,18 @@ impl MessageWrapper {
let mut batch = BatchBuilder::new();
// Reserve quotas
for quota_key in &self.message.quota_keys {
match quota_key {
QuotaKey::Count { key, .. } => {
for metadata in &self.message.metadata {
match metadata {
Metadata::QueueCount { key, .. } => {
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.to_vec())), 1);
}
QuotaKey::Size { key, .. } => {
Metadata::QueueSize { key, .. } => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key.to_vec())),
self.message.size as i64,
);
}
Metadata::Headers { .. } => {}
}
}
@@ -756,17 +789,18 @@ impl MessageWrapper {
}
// Release all quotas
for quota_key in self.message.quota_keys {
match quota_key {
QuotaKey::Count { key, .. } => {
for metadata in self.message.metadata {
match metadata {
Metadata::QueueCount { key, .. } => {
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.to_vec())), -1);
}
QuotaKey::Size { key, .. } => {
Metadata::QueueSize { key, .. } => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key.to_vec())),
-(self.message.size as i64),
);
}
Metadata::Headers { .. } => {}
}
}
@@ -918,17 +952,18 @@ impl MessageWrapper {
)));
}
for quota_key in self.message.quota_keys {
match quota_key {
QuotaKey::Count { key, .. } => {
for metadata in self.message.metadata {
match metadata {
Metadata::QueueCount { key, .. } => {
batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.to_vec())), -1);
}
QuotaKey::Size { key, .. } => {
Metadata::QueueSize { key, .. } => {
batch.add(
ValueClass::Queue(QueueClass::QuotaSize(key.to_vec())),
-(self.message.size as i64),
);
}
Metadata::Headers { .. } => {}
}
}
@@ -1047,3 +1082,57 @@ impl ArchivedMessage {
next_notify
}
}
impl<'x, 'y> QueueParams<'x, 'y> {
pub fn new(
raw_message: &'x [u8],
session_id: u64,
server: &'y Server,
source: MessageSource,
) -> Self {
QueueParams {
raw_message,
dkim_signers: None,
raw_headers: None,
session_id,
server,
source,
original_raw_message: None,
original_authenticated_message: None,
metadata: Vec::new(),
}
}
pub fn with_original_authenticated_message(
mut self,
authenticated_message: AuthenticatedMessage<'x>,
) -> Self {
self.original_authenticated_message = Some(authenticated_message);
self
}
pub fn with_original_raw_message(mut self, raw_message: &'x [u8]) -> Self {
self.original_raw_message = Some(raw_message);
self
}
pub fn with_dkim_signers(mut self, dkim_signers: Option<Arc<DkimSigners>>) -> Self {
self.dkim_signers = dkim_signers;
self
}
pub fn with_raw_headers(mut self, raw_headers: &'x [u8]) -> Self {
self.raw_headers = Some(raw_headers);
self
}
pub fn with_raw_headers_opt(mut self, raw_headers: Option<&'x [u8]>) -> Self {
self.raw_headers = raw_headers;
self
}
pub fn with_metadata(mut self, metadata: Vec<Metadata>) -> Self {
self.metadata = metadata;
self
}
}