Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-09-22 13:01:21 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat: queue messages for SMTP before encryption

Browse Source 
Headers like From and Autocrypt are now added late,
right before sending the message over SMTP.
This way we advertise the latest list of transports
and use the correct From address in the encrypted part
even for messages queued while being offline.

BCC-self recipients are also added late.
For unencrypted messages we only want to send a copy
to the sending address, but we don't know the sending address
when queueing the message.
Adding bcc-self recipients when dequeuing the message
also makes it possible to send copies to updated list of relays.
This commit is contained in:
link2xt
2026-09-09 09:32:23 +00:00
committed by l
parent 81982273e8
commit 373f1840a6
21 changed files with 699 additions and 360 deletions
Download Patch File Download Diff File Expand all files Collapse all files

337
src/chat.rs
View File

@@ -11,7 +11,6 @@ use std::time::Duration;
use anyhow::{Context as _, Result, anyhow, bail, ensure};
use chrono::TimeZone;
use deltachat_contact_tools::{ContactAddress, sanitize_bidi_characters, sanitize_single_line};
use humansize::{BINARY, format_size};
use mail_builder::mime::MimePart;
use serde::{Deserialize, Serialize};
use strum_macros::EnumIter;
@@ -27,26 +26,22 @@ use crate::constants::{
use crate::contact::{self, Contact, ContactId, Origin};
use crate::context::Context;
use crate::debug_logging::maybe_set_logging_xdc;
use crate::download::{
DownloadState, PRE_MSG_ATTACHMENT_SIZE_THRESHOLD, PRE_MSG_SIZE_WARNING_THRESHOLD,
};
use crate::download::{DownloadState, PRE_MSG_ATTACHMENT_SIZE_THRESHOLD};
use crate::ensure_and_debug_assert_eq;
use crate::ephemeral::{Timer as EphemeralTimer, start_chat_ephemeral_timers};
use crate::events::EventType;
use crate::key;
use crate::key::{Fingerprint, self_fingerprint};
use crate::location;
use crate::key::{DcKey as _, Fingerprint, self_fingerprint};
use crate::log::{LogExt, warn};
use crate::logged_debug_assert;
use crate::message::{self, Message, MessageState, MsgId, Viewtype};
use crate::mimefactory;
use crate::mimefactory::{MimeFactory, RenderedEmail};
use crate::mimefactory::{MimeFactory, QueueSideEffects, QueuedMail, ToBeQueuedMail};
use crate::mimeparser::SystemMessage;
use crate::param::{Param, Params};
use crate::pgp::addresses_from_public_key;
use crate::reaction::broadcast_reactions;
use crate::receive_imf::ReceivedMsg;
use crate::smtp::{self, send_msg_to_smtp};
use crate::smtp::send_msg_to_smtp;
use crate::stock_str;
use crate::sync::{self, Sync::*, SyncData};
use crate::tools::{
@@ -336,16 +331,18 @@ impl ChatId {
Ok(chat_id)
}
async fn set_selfavatar_timestamp(self, context: &Context, timestamp: i64) -> Result<()> {
context
.sql
fn set_selfavatar_timestamp(
self,
transaction: &mut rusqlite::Transaction<'_>,
timestamp: i64,
) -> Result<()> {
transaction
.execute(
"UPDATE contacts
SET selfavatar_sent=?
WHERE id IN(SELECT contact_id FROM chats_contacts WHERE chat_id=? AND add_timestamp >= remove_timestamp)",
(timestamp, self),
)
.await?;
) ?;
Ok(())
}
@@ -2776,11 +2773,8 @@ async fn render_mime_message_and_pre_message(
context: &Context,
msg: &mut Message,
mimefactory: MimeFactory,
) -> Result<(Option<RenderedEmail>, RenderedEmail)> {
let from_addr = context.get_primary_self_addr().await?;
let public_key = key::load_self_public_key(context).await?;
let secret_key = key::load_self_secret_key(context).await?;
bcc_self: bool,
) -> Result<(Option<ToBeQueuedMail>, ToBeQueuedMail)> {
let needs_pre_message = msg.viewtype.has_file()
&& mimefactory.will_be_encrypted() // unencrypted is likely email, we don't want to spam by sending multiple messages
&& msg
@@ -2797,54 +2791,121 @@ async fn render_mime_message_and_pre_message(
let mut mimefactory_post_msg = mimefactory.clone();
mimefactory_post_msg.set_as_post_message();
let (queued_msg, side_effects) = Box::pin(mimefactory_post_msg.into_queued_mail(context))
.await
.context("Failed to render post-message")?;
let rendered_msg = mimefactory::render_queued_mail(
queued_msg,
&public_key,
&secret_key,
from_addr.clone(),
side_effects,
)?;
let (queued_msg, side_effects) =
Box::pin(mimefactory_post_msg.into_queued_mail(context, bcc_self))
.await
.context("Failed to render post-message")?;
let mut mimefactory_pre_msg = mimefactory;
mimefactory_pre_msg.set_as_pre_message_for(&rendered_msg);
mimefactory_pre_msg.set_as_pre_message_for(&queued_msg.rfc724_mid);
let (queued_pre_msg, pre_side_effects) =
Box::pin(mimefactory_pre_msg.into_queued_mail(context))
Box::pin(mimefactory_pre_msg.into_queued_mail(context, bcc_self))
.await
.context("pre-message failed to render")?;
let rendered_pre_msg = mimefactory::render_queued_mail(
queued_pre_msg,
&public_key,
&secret_key,
from_addr,
pre_side_effects,
)?;
if rendered_pre_msg.message.len() > PRE_MSG_SIZE_WARNING_THRESHOLD {
warn!(
context,
"Pre-message for message {} is larger than expected: {}.",
msg.id,
rendered_pre_msg.message.len()
);
Ok((
Some((queued_pre_msg, pre_side_effects)),
(queued_msg, side_effects),
))
} else {
let (queued_msg, side_effects) =
Box::pin(mimefactory.into_queued_mail(context, bcc_self)).await?;
Ok((None, (queued_msg, side_effects)))
}
}
/// Process side effects and store queued mail.
pub(crate) fn enqueue_mail(
transaction: &mut rusqlite::Transaction<'_>,
now: i64,
msg_id: MsgId,
queued_mail: &QueuedMail,
side_effects: Option<&QueueSideEffects>,
) -> Result<i64> {
if let Some(side_effects) = side_effects {
if let Some(last_added_location_timestamp) = side_effects.last_added_location_timestamp {
transaction.execute(
"UPDATE chats SET locations_last_sent=? WHERE id=?;",
(last_added_location_timestamp, side_effects.chat_id),
)?;
}
Ok((Some(rendered_pre_msg), rendered_msg))
} else {
let (queued_msg, side_effects) = Box::pin(mimefactory.into_queued_mail(context)).await?;
let rendered_msg = mimefactory::render_queued_mail(
queued_msg,
&public_key,
&secret_key,
from_addr,
side_effects,
)?;
if side_effects.avatar_is_attached {
side_effects
.chat_id
.set_selfavatar_timestamp(transaction, now)
.context("Failed to set selfavatar timestamp")?;
}
Ok((None, rendered_msg))
if let Some(ref sync_ids) = side_effects.sync_ids_to_delete {
transaction.execute(
&format!("DELETE FROM multi_device_sync WHERE id IN ({sync_ids})"),
(),
)?;
}
}
// Store mail into queue.
let all_recipients = queued_mail.recipients.join(" ");
let is_encrypted = queued_mail.encryption.is_encrypted();
transaction
.execute(
"
INSERT INTO smtp2 (
display_name,
rfc724_mid,
mime,
should_attach_pubkey,
should_compress,
should_sign,
msg_id,
recipients,
bcc_self,
is_encrypted,
shared_secret,
encryption_fingerprints
)
VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
)
",
(
&queued_mail.display_name,
&queued_mail.rfc724_mid,
&queued_mail.raw_message,
queued_mail.should_attach_pubkey,
queued_mail.should_compress,
queued_mail.should_sign,
msg_id,
&all_recipients,
queued_mail.bcc_self,
is_encrypted,
if let mimefactory::QueuedEncryption::Symmetric { ref shared_secret } =
queued_mail.encryption
{
shared_secret
} else {
""
},
if let mimefactory::QueuedEncryption::Asymmetric {
ref encryption_pubkeys,
} = queued_mail.encryption
{
let res: Vec<String> = encryption_pubkeys
.iter()
.map(|pubkey| pubkey.dc_fingerprint().hex())
.collect();
res.join(" ")
} else {
"".to_string()
},
),
)
.context("Failed to insert a row into smtp2 table")?;
let row_id = transaction.last_insert_rowid();
Ok(row_id)
}
/// Constructs jobs for sending a message and inserts them into the `smtp` table.
@@ -2857,6 +2918,8 @@ async fn render_mime_message_and_pre_message(
///
/// The caller has to interrupt SMTP loop or otherwise process new rows.
async fn create_send_msg_jobs(context: &Context, msg: &mut Message) -> Result<Vec<i64>> {
let now = time();
let cmd = msg.param.get_cmd();
if cmd == SystemMessage::GroupNameChanged || cmd == SystemMessage::GroupDescriptionChanged {
msg.chat_id
@@ -2888,12 +2951,14 @@ async fn create_send_msg_jobs(context: &Context, msg: &mut Message) -> Result<Ve
return Err(err);
}
};
let mut recipients = mimefactory.recipients();
let recipients = mimefactory.recipients();
debug_assert!(!recipients.iter().any(|s| s.is_empty()));
let bcc_self = context.get_config_bool(Config::BccSelf).await?;
// Default Webxdc integrations are hidden messages and must not be sent out:
if (msg.param.get_int(Param::WebxdcIntegration).is_some() && msg.hidden)
// This may happen eg. for groups with only SELF and bcc_self disabled:
|| (!context.get_config_bool(Config::BccSelf).await? && recipients.is_empty())
|| (!bcc_self && recipients.is_empty())
{
info!(
context,
@@ -2906,8 +2971,9 @@ async fn create_send_msg_jobs(context: &Context, msg: &mut Message) -> Result<Ve
return Ok(Vec::new());
}
let (rendered_pre_msg, rendered_msg) =
match render_mime_message_and_pre_message(context, msg, mimefactory).await {
let is_encrypted = mimefactory.will_be_encrypted();
let (queued_pre_msg_pair, queued_msg_pair) =
match render_mime_message_and_pre_message(context, msg, mimefactory, bcc_self).await {
Ok(res) => Ok(res),
Err(err) => {
message::set_msg_failed(context, msg, &err.to_string()).await?;
@@ -2915,29 +2981,13 @@ async fn create_send_msg_jobs(context: &Context, msg: &mut Message) -> Result<Ve
}
}?;
if let (post_msg, Some(pre_msg)) = (&rendered_msg, &rendered_pre_msg) {
info!(
context,
"Message {} sizes: pre-message: {}; post-message: {}.",
msg.id,
format_size(pre_msg.message.len(), BINARY),
format_size(post_msg.message.len(), BINARY),
);
if let Some((pre_msg, _)) = &queued_pre_msg_pair {
msg.pre_rfc724_mid = pre_msg.rfc724_mid.clone();
} else {
info!(
context,
"Message {} will be sent in one shot (no pre- and post-message). Size: {}.",
msg.id,
format_size(rendered_msg.message.len(), BINARY),
);
}
if context.get_config_bool(Config::BccSelf).await? {
smtp::add_self_recipients(context, &mut recipients, rendered_msg.is_encrypted).await?;
}
let (queued_msg, side_effects) = queued_msg_pair;
if needs_encryption && !rendered_msg.is_encrypted {
if needs_encryption && !is_encrypted {
let addr = context.get_config(Config::ConfiguredAddr).await?;
let text = stock_str::unencrypted_email(
context,
@@ -2967,95 +3017,62 @@ async fn create_send_msg_jobs(context: &Context, msg: &mut Message) -> Result<Ve
);
}
let now = time();
if let Some(last_added_location_timestamp) =
rendered_msg.side_effects.last_added_location_timestamp
{
location::set_kml_sent_timestamp(context, msg.chat_id, last_added_location_timestamp)
.await?;
if let Some(ref side_effects) = side_effects {
msg.subject.clone_from(&side_effects.subject);
}
if rendered_msg.side_effects.avatar_is_attached
|| rendered_pre_msg
.as_ref()
.is_some_and(|msg| msg.side_effects.avatar_is_attached)
{
msg.chat_id
.set_selfavatar_timestamp(context, now)
.await
.context("Failed to set selfavatar timestamp")?;
}
if rendered_msg.is_encrypted {
if is_encrypted {
msg.param.set_int(Param::GuaranteeE2ee, 1);
} else {
msg.param.remove(Param::GuaranteeE2ee);
}
msg.subject.clone_from(&rendered_msg.side_effects.subject);
// Sort the message to the bottom. Employ `msgs_index7` to compute `timestamp`.
context
.sql
.execute(
"
UPDATE msgs SET
timestamp=(
SELECT MAX(timestamp) FROM msgs INDEXED BY msgs_index7 WHERE
-- From `InFresh` to `OutDelivered` inclusive, except `OutDraft`.
state IN(10,13,16,18,20,24,26) AND
hidden IN(0,1) AND
chat_id=? AND
id<=?
),
pre_rfc724_mid=?, subject=?, param=?
WHERE id=?
",
(
msg.chat_id,
msg.id,
&msg.pre_rfc724_mid,
&msg.subject,
msg.param.to_string(),
msg.id,
),
)
.await?;
let trans_fn = |t: &mut rusqlite::Transaction| {
let mut row_ids = Vec::<i64>::new();
if let Some(sync_ids) = rendered_msg.side_effects.sync_ids_to_delete {
t.execute(
&format!("DELETE FROM multi_device_sync WHERE id IN ({sync_ids})"),
(),
)?;
}
if !recipients.is_empty() {
let mut stmt = t.prepare(
"INSERT INTO smtp (rfc724_mid, recipients, mime, msg_id)
VALUES (?1, ?2, ?3, ?4)",
)?;
let all_recipients = recipients.join(" ");
if let Some(pre_msg) = &rendered_pre_msg {
let row_id = stmt.insert((
&pre_msg.rfc724_mid,
&all_recipients,
&pre_msg.message,
.transaction(|transaction| {
// Sort the message to the bottom. Employ `msgs_index7` to compute `timestamp`.
transaction.execute(
"
UPDATE msgs SET
timestamp=(
SELECT MAX(timestamp) FROM msgs INDEXED BY msgs_index7 WHERE
-- From `InFresh` to `OutDelivered` inclusive, except `OutDraft`.
state IN(10,13,16,18,20,24,26) AND
hidden IN(0,1) AND
chat_id=? AND
id<=?
),
pre_rfc724_mid=?, subject=?, param=?
WHERE id=?
",
(
msg.chat_id,
msg.id,
))?;
row_ids.push(row_id);
&msg.pre_rfc724_mid,
&msg.subject,
msg.param.to_string(),
msg.id,
),
)?;
let mut row_ids = Vec::new();
if let Some((queued_pre_msg, pre_side_effects)) = queued_pre_msg_pair {
let row_id = enqueue_mail(
transaction,
now,
msg.id,
&queued_pre_msg,
pre_side_effects.as_ref(),
)
.context("Failed to enqueue pre-message")?;
row_ids.push(row_id)
}
let row_id = stmt.insert((
&rendered_msg.rfc724_mid,
&all_recipients,
&rendered_msg.message,
msg.id,
))?;
row_ids.push(row_id);
}
Ok(row_ids)
};
context.sql.transaction(trans_fn).await
row_ids.push(
enqueue_mail(transaction, now, msg.id, &queued_msg, side_effects.as_ref())
.context("Failed to enqueue message")?,
);
Ok(row_ids)
})
.await
}
/// Sends a text message to the given chat.

5
src/chat/chat_tests.rs
View File

@@ -1702,7 +1702,10 @@ async fn test_shall_attach_selfavatar() -> Result<()> {
add_contact_to_chat(alice, chat_id, contact_id).await?;
assert!(shall_attach_selfavatar(alice, chat_id).await?);
chat_id.set_selfavatar_timestamp(alice, time()).await?;
alice
.sql
.transaction(|transaction| chat_id.set_selfavatar_timestamp(transaction, time()))
.await?;
assert!(!shall_attach_selfavatar(alice, chat_id).await?);
alice.set_config(Config::Selfavatar, None).await?; // setting to None also forces re-sending

7
src/config.rs
View File

@@ -809,13 +809,6 @@ impl Context {
"Failed to update add_timestamp for the new sending transport",
)?;
// Clean up SMTP queue.
//
// The messages in the queue have a different
// From address so we cannot send them over
// the new SMTP transport.
transaction.execute("DELETE FROM smtp", ())?;
Ok(())
})
.await?;

3
src/download.rs
View File

@@ -23,9 +23,6 @@ pub(crate) use post_msg_metadata::PostMsgMetadata;
/// KiB).
pub(crate) const PRE_MSG_ATTACHMENT_SIZE_THRESHOLD: u64 = 140_000;
/// Max size for pre messages. A warning is emitted when this is exceeded.
pub(crate) const PRE_MSG_SIZE_WARNING_THRESHOLD: usize = 150_000;
/// Download state of the message.
#[derive(
Debug,

2
src/ephemeral/ephemeral_tests.rs
View File

@@ -676,7 +676,7 @@ async fn test_ephemeral_msg_offline() -> Result<()> {
.await?;
let mut msg = Message::new_text("hi".to_string());
assert!(chat::send_msg_sync(alice, chat.id, &mut msg).await.is_err());
let stmt = "SELECT COUNT(*) FROM smtp WHERE msg_id=?";
let stmt = "SELECT COUNT(*) FROM smtp2 WHERE msg_id=?";
assert!(alice.sql.exists(stmt, (msg.id,)).await?);
let now = time();
check_msg_will_be_deleted(alice, msg.id, &chat, now, now + i64::from(duration) + 1).await?;

10
src/keyupdate.rs
View File

@@ -40,7 +40,7 @@ use crate::contact::ContactId;
use crate::context::Context;
use crate::key::{DcKey, SignedPublicKey};
use crate::log::warn;
use crate::mimefactory::render_keyupdate_message;
use crate::mimefactory::keyupdate_message;
use crate::pgp::{pubkey_can_encrypt, relay_addrs};
use crate::smtp::insert_into_smtp;
use crate::tools::{create_outgoing_rfc724_mid, time};
@@ -145,14 +145,14 @@ async fn keyupdate_recipients(
/// Returns the deduplicated relay addresses to put into the SMTP envelope
/// for the keyupdate encrypted to a `chunk` of recipients.
fn envelope_recipients(chunk: &[KeyupdateRecipient]) -> String {
fn envelope_recipients(chunk: &[KeyupdateRecipient]) -> Vec<String> {
let mut addrs = BTreeSet::new();
for recipient in chunk {
for relay in &recipient.relays {
addrs.insert(addr_normalize(relay));
}
}
Vec::from_iter(addrs).join(" ")
Vec::from_iter(addrs)
}
/// Returns the relay list in the format stored in [`Config::KeyupdateBaseline`].
@@ -192,8 +192,8 @@ pub(crate) async fn maybe_send_keyupdate_message(context: &Context) -> Result<()
let envelope = envelope_recipients(chunk);
let rfc724_mid = create_outgoing_rfc724_mid();
let keys = chunk.iter().map(|r| r.public_key.clone()).collect();
let rendered_message = render_keyupdate_message(context, &rfc724_mid, keys).await?;
insert_into_smtp(context, &rfc724_mid, &envelope, rendered_message).await?;
let queued_msg = keyupdate_message(context, &rfc724_mid, keys, envelope).await?;
insert_into_smtp(context, &rfc724_mid, &queued_msg).await?;
}
// Record only after queueing, so failed queueing is retried by a later check.

16
src/location.rs
View File

@@ -601,22 +601,6 @@ pub fn get_message_kml(timestamp: i64, latitude: f64, longitude: f64) -> String
)
}
/// Sets the timestamp of the last time location was sent in the chat.
pub async fn set_kml_sent_timestamp(
context: &Context,
chat_id: ChatId,
timestamp: i64,
) -> Result<()> {
context
.sql
.execute(
"UPDATE chats SET locations_last_sent=? WHERE id=?;",
(timestamp, chat_id),
)
.await?;
Ok(())
}
/// Sets the location of the message.
pub async fn set_msg_location_id(context: &Context, msg_id: MsgId, location_id: u32) -> Result<()> {
context

2
src/message.rs
View File

@@ -1697,7 +1697,7 @@ pub async fn delete_msgs_ext(
if !msg.pre_rfc724_mid.is_empty() {
stmt.execute((&msg.pre_rfc724_mid,))?;
}
trans.execute("DELETE FROM smtp WHERE msg_id=?", (msg_id,))?;
trans.execute("DELETE FROM smtp2 WHERE msg_id=?", (msg_id,))?;
trans.execute(
"DELETE FROM download WHERE rfc724_mid=?",
(&msg.rfc724_mid,),

2
src/message/message_tests.rs
View File

@@ -626,7 +626,7 @@ async fn test_delete_msgs_offline() -> Result<()> {
let chat_id = alice.create_chat_id(bob).await;
let mut msg = Message::new_text("hi".to_string());
assert!(chat::send_msg_sync(alice, chat_id, &mut msg).await.is_err());
let stmt = "SELECT COUNT(*) FROM smtp WHERE msg_id=?";
let stmt = "SELECT COUNT(*) FROM smtp2 WHERE msg_id=?";
assert!(alice.sql.exists(stmt, (msg.id,)).await?);
delete_msgs(alice, &[msg.id]).await?;
assert!(!alice.sql.exists(stmt, (msg.id,)).await?);

162
src/mimefactory.rs
View File

@@ -17,7 +17,7 @@ use tokio::fs;
use crate::aheader::{Aheader, EncryptPreference};
use crate::blob::BlobObject;
use crate::chat::{self, Chat, PARAM_BROADCAST_SECRET, load_broadcast_secret};
use crate::chat::{self, Chat, ChatId, PARAM_BROADCAST_SECRET, load_broadcast_secret};
use crate::config::Config;
use crate::constants::{Chattype, DC_FROM_HANDSHAKE};
use crate::contact::{Contact, ContactId, Origin};
@@ -262,33 +262,48 @@ pub(crate) struct QueuedMail {
/// but without the From, Autocrypt and Message-ID headers.
///
/// For encrypted messages this is the OpenPGP payload.
raw_message: Vec<u8>,
pub(crate) raw_message: Vec<u8>,
/// Display name to put in the `From:` field.
///
/// Email address is not determined yet here.
display_name: String,
pub(crate) display_name: String,
/// Message-ID.
rfc724_mid: String,
pub(crate) rfc724_mid: String,
/// Whether the message is encrypted and encryption keys.
encryption: QueuedEncryption,
pub(crate) encryption: QueuedEncryption,
/// If true, Autocrypt header should be added before sending.
should_attach_pubkey: bool,
pub(crate) should_attach_pubkey: bool,
/// If true, OpenPGP compression may be used.
should_compress: bool,
pub(crate) should_compress: bool,
/// If true, encrypted message should be signed as well.
should_sign: bool,
/// If true, encrypted message should be signed.
pub(crate) should_sign: bool,
/// Recipient addresses.
pub(crate) recipients: Vec<String>,
/// Addresses the messages was already sent to.
pub(crate) sent_to: Vec<String>,
/// If true, own addresses should be added to the list of recipients.
///
/// For unencrypted messages, only the sending addresses should be added.
/// For encrypted messages, all published addresses should be added.
pub(crate) bcc_self: bool,
}
/// Side effects that should be applied at the same time
/// as the message is persisted in the queue.
#[derive(Debug, Clone, Default)]
pub struct RenderSideEffects {
pub struct QueueSideEffects {
/// ID of the chat side effects should be applied to.
pub chat_id: ChatId,
/// Largest timestamp of the location sent in `location.kml` in this message.
pub last_added_location_timestamp: Option<i64>,
@@ -307,6 +322,9 @@ pub struct RenderSideEffects {
pub subject: String,
}
/// Email message ready to be queued with the side effects that should be applied at the same time.
pub(crate) type ToBeQueuedMail = (QueuedMail, Option<QueueSideEffects>);
/// Renders [`QueuedMail`].
///
/// Adds headers:
@@ -320,7 +338,6 @@ pub(crate) fn render_queued_mail(
public_key: &SignedPublicKey,
secret_key: &SignedSecretKey,
from_addr: String,
side_effects: RenderSideEffects,
) -> Result<RenderedEmail> {
let QueuedMail {
rfc724_mid,
@@ -330,6 +347,9 @@ pub(crate) fn render_queued_mail(
should_attach_pubkey,
should_compress,
should_sign,
recipients: _,
sent_to: _,
bcc_self: _,
} = queued_mail;
let mut inner_headers: Vec<u8> = Vec::new();
@@ -524,23 +544,30 @@ pub(crate) fn render_queued_mail(
full_message.extend(message);
Ok(RenderedEmail {
message: String::from_utf8_lossy(&full_message).to_string(),
is_encrypted,
rfc724_mid,
side_effects,
})
}
/// Renders queued mail with the current sending address.
pub(crate) async fn render_queued_mail_with_context(
queued_mail: QueuedMail,
context: &Context,
) -> Result<RenderedEmail> {
let from_addr = context.get_primary_self_addr().await?;
let public_key = key::load_self_public_key(context).await?;
let secret_key = key::load_self_secret_key(context).await?;
let rendered_mail = render_queued_mail(queued_mail, &public_key, &secret_key, from_addr)?;
Ok(rendered_mail)
}
/// Result of rendering a message, ready to be submitted to a send job.
#[derive(Debug, Clone)]
pub struct RenderedEmail {
pub message: String,
pub is_encrypted: bool,
/// Message ID (Message in the sense of Email)
pub rfc724_mid: String,
pub side_effects: RenderSideEffects,
}
fn new_address_with_name(name: &str, address: String) -> Address<'static> {
@@ -1392,17 +1419,12 @@ impl MimeFactory {
/// Used for MDNs because they are fully rendered and sent in one go,
/// rather than first creating a [`QueuedMail`] and sending it later.
pub async fn render(self, context: &Context) -> Result<RenderedEmail> {
let from_addr = context.get_primary_self_addr().await?;
let public_key = key::load_self_public_key(context).await?;
let secret_key = key::load_self_secret_key(context).await?;
let (queued_mail, side_effects) = Box::pin(self.into_queued_mail(context)).await?;
let rendered_mail = render_queued_mail(
queued_mail,
&public_key,
&secret_key,
from_addr,
side_effects,
)?;
// Does not matter, we are not going to return the QueuedMail.
let bcc_self = false;
let (queued_mail, _side_effects) =
Box::pin(self.into_queued_mail(context, bcc_self)).await?;
let rendered_mail = render_queued_mail_with_context(queued_mail, context).await?;
Ok(rendered_mail)
}
@@ -1412,7 +1434,8 @@ impl MimeFactory {
pub(crate) async fn into_queued_mail(
mut self,
context: &Context,
) -> Result<(QueuedMail, RenderSideEffects)> {
bcc_self: bool,
) -> Result<ToBeQueuedMail> {
let rfc724_mid = match &self.loaded {
Loaded::Message { msg, .. } => match &self.pre_message_mode {
PreMessageMode::Pre { .. } => {
@@ -1434,9 +1457,7 @@ impl MimeFactory {
let is_encrypted = self.will_be_encrypted();
let last_added_location_timestamp;
let avatar_is_attached;
let sync_ids_to_delete;
let side_effects: Option<QueueSideEffects>;
let message: MimePart<'static> = match &self.loaded {
Loaded::Message { msg, .. } => {
@@ -1444,15 +1465,21 @@ impl MimeFactory {
let RenderedMessage {
main_part,
mut parts,
last_added_location_timestamp: tmp_last_added_location_timestamp,
avatar_is_attached: tmp_avatar_is_attached,
sync_ids_to_delete: tmp_sync_ids_to_delete,
last_added_location_timestamp,
avatar_is_attached,
sync_ids_to_delete,
} = self
.render_message(context, &mut headers, &grpimage, is_encrypted)
.await?;
last_added_location_timestamp = tmp_last_added_location_timestamp;
avatar_is_attached = tmp_avatar_is_attached;
sync_ids_to_delete = tmp_sync_ids_to_delete;
side_effects = Some(QueueSideEffects {
chat_id: msg.chat_id,
avatar_is_attached,
sync_ids_to_delete,
last_added_location_timestamp,
subject: subject_str,
});
if parts.is_empty() {
// Single part, render as regular message.
main_part
@@ -1470,9 +1497,7 @@ impl MimeFactory {
}
}
Loaded::Mdn { .. } => {
last_added_location_timestamp = None;
avatar_is_attached = false;
sync_ids_to_delete = None;
side_effects = None;
self.render_mdn()?
}
};
@@ -1482,12 +1507,6 @@ impl MimeFactory {
Loaded::Mdn { .. } => self.update_mdn_pubkey_attachment(context).await?,
};
let is_post_message = self.pre_message_mode == PreMessageMode::Post;
let side_effects = RenderSideEffects {
avatar_is_attached,
sync_ids_to_delete,
last_added_location_timestamp,
subject: subject_str,
};
let is_securejoin_message = match &self.loaded {
Loaded::Message { msg, .. } => msg.param.get_cmd() == SystemMessage::SecurejoinMessage,
@@ -1605,7 +1624,7 @@ impl MimeFactory {
let is_mdn = matches!(self.loaded, Loaded::Mdn { .. });
let should_sign = true;
let message = if self.will_be_encrypted() {
let message = if is_encrypted {
add_headers_to_encrypted_part(message, headers)
} else if is_mdn {
// Never add outer multipart/mixed wrapper to MDN
@@ -1641,6 +1660,7 @@ impl MimeFactory {
})
};
let raw_message = part_to_bytes(message);
let recipients = self.recipients();
let queued_email = QueuedMail {
raw_message,
@@ -1650,6 +1670,9 @@ impl MimeFactory {
should_attach_pubkey,
should_sign,
should_compress,
recipients,
sent_to: Vec::new(),
bcc_self,
};
Ok((queued_email, side_effects))
}
@@ -2335,9 +2358,9 @@ impl MimeFactory {
self.pre_message_mode = PreMessageMode::Post;
}
pub fn set_as_pre_message_for(&mut self, post_message: &RenderedEmail) {
pub fn set_as_pre_message_for(&mut self, rfc724_mid: &str) {
self.pre_message_mode = PreMessageMode::Pre {
post_msg_rfc724_mid: post_message.rfc724_mid.clone(),
post_msg_rfc724_mid: rfc724_mid.to_string(),
};
}
}
@@ -2515,29 +2538,15 @@ async fn non_chat_headers(
Ok(headers)
}
/// Renders `queued_mail` for SMTP with the own key pair and sending address.
async fn render_with_self_key(context: &Context, queued_mail: QueuedMail) -> Result<String> {
let public_key = key::load_self_public_key(context).await?;
let secret_key = key::load_self_secret_key(context).await?;
let from_addr = context.get_primary_self_addr().await?;
let rendered_mail = render_queued_mail(
queued_mail,
&public_key,
&secret_key,
from_addr,
RenderSideEffects::default(),
)?;
Ok(rendered_mail.message)
}
pub(crate) async fn render_symm_encrypted_securejoin_message(
pub(crate) async fn symm_encrypted_securejoin_message(
context: &Context,
step: &str,
rfc724_mid: &str,
should_attach_pubkey: bool,
auth: &str,
shared_secret: &str,
) -> Result<String> {
recipients: Vec<String>,
) -> Result<QueuedMail> {
info!(context, "Sending secure-join message {step:?}.");
let message: MimePart<'static> = MimePart::new("text/plain", "Secure-Join");
@@ -2562,9 +2571,13 @@ pub(crate) async fn render_symm_encrypted_securejoin_message(
// there are no compression side channels
// leaking information about the tokens.
should_compress: false,
recipients,
sent_to: Vec::new(),
// Never send a copy of SecureJoin message to self.
bcc_self: false,
};
render_with_self_key(context, queued_mail).await
Ok(queued_mail)
}
/// Returns the body of a keyupdate message, shaped like a receipt notification.
@@ -2593,13 +2606,14 @@ fn keyupdate_body() -> MimePart<'static> {
)
}
/// Renders a keyupdate message informing the owners of `recipient_keys`
/// Returns a keyupdate message informing the owners of `recipient_keys`
/// about the current key and relay list, see [`crate::keyupdate`].
pub(crate) async fn render_keyupdate_message(
pub(crate) async fn keyupdate_message(
context: &Context,
rfc724_mid: &str,
recipient_keys: Vec<SignedPublicKey>,
) -> Result<String> {
recipients: Vec<String>,
) -> Result<QueuedMail> {
info!(
context,
"Sending keyupdate message to {} recipients.",
@@ -2627,9 +2641,11 @@ pub(crate) async fn render_keyupdate_message(
// Disable compression to avoid side channels, message body is small anyway.
should_compress: false,
recipients,
sent_to: Vec::new(),
bcc_self: false,
};
render_with_self_key(context, queued_mail).await
Ok(queued_mail)
}
/// Renders MIME part into a vector of bytes.

4
src/mimefactory/mimefactory_tests.rs
View File

@@ -306,9 +306,9 @@ async fn test_mdn_create_encrypted() -> Result<()> {
message::markseen_msgs(&bob, vec![rcvd.id]).await?;
let mimefactory =
MimeFactory::from_mdn(&bob, rcvd.from_id, rcvd.rfc724_mid.clone(), vec![]).await?;
assert!(!mimefactory.will_be_encrypted());
let rendered_msg = mimefactory.render(&bob).await?;
assert!(!rendered_msg.is_encrypted);
assert!(!rendered_msg.message.contains("Bob Examplenet"));
assert!(!rendered_msg.message.contains("Alice Exampleorg"));
let bob_alice_contact = bob.add_or_lookup_contact(&alice).await;
@@ -319,9 +319,9 @@ async fn test_mdn_create_encrypted() -> Result<()> {
message::markseen_msgs(&bob, vec![rcvd.id]).await?;
let mimefactory = MimeFactory::from_mdn(&bob, rcvd.from_id, rcvd.rfc724_mid, vec![]).await?;
assert!(mimefactory.will_be_encrypted());
let rendered_msg = mimefactory.render(&bob).await?;
assert!(rendered_msg.is_encrypted);
assert!(!rendered_msg.message.contains("Bob Examplenet"));
assert!(!rendered_msg.message.contains("Alice Exampleorg"));

12
src/receive_imf.rs
View File

@@ -568,16 +568,12 @@ pub(crate) async fn receive_imf_inner(
//
// Note that messages with long recipient lists are sent out in chunks,
// removing already sent recipients from the job after each chunk.
// Self recipients are added at the end so removing the job
// removes the last chunk which apparently went out fine.
let self_addr = context.get_primary_self_addr().await?;
// Self recipients are sent in the end,
// so if we received a copy, the message has been sent out
// to all recipients.
context
.sql
.execute(
"DELETE FROM smtp \
WHERE rfc724_mid=?1 AND (recipients LIKE ?2 OR recipients LIKE ('% ' || ?2))",
(rfc724_mid_orig, &self_addr),
)
.execute("DELETE FROM smtp2 WHERE rfc724_mid=?", (rfc724_mid_orig,))
.await?;
if !msg_has_pending_smtp_job(context, msg_id).await? {
msg_id.set_delivered(context).await?;

35
src/receive_imf/receive_imf_tests.rs
View File

@@ -15,7 +15,9 @@ use crate::headerdef::HeaderDefMap as _;
use crate::imap::prefetch_should_download;
use crate::imex::{ImexMode, imex};
use crate::key;
use crate::mimefactory;
use crate::securejoin::get_securejoin_qr;
use crate::smtp;
use crate::test_utils;
use crate::test_utils::{
TestContext, TestContextManager, alice_keypair, get_chat_msg, mark_as_verified,
@@ -5847,15 +5849,32 @@ async fn test_mark_message_as_delivered_only_after_sent_out_fully() -> Result<()
/// This simulates the case that a message is successfully sent out,
/// but the 'OK' answer from the server doesn't arrive,
/// so that the SMTP row stays in the database.
pub(crate) async fn first_row_in_smtp_queue(alice: &TestContext) -> (MsgId, String) {
alice
pub(crate) async fn first_row_in_smtp_queue(context: &TestContext) -> (MsgId, String) {
let (rowid, msg_id) = context
.sql
.query_row_optional("SELECT msg_id, mime FROM smtp ORDER BY id", (), |row| {
let msg_id: MsgId = row.get(0)?;
let mime: String = row.get(1)?;
Ok((msg_id, mime))
})
.query_row_optional(
"SELECT id, msg_id FROM smtp2 ORDER BY id LIMIT 1",
(),
|row| {
let rowid: i64 = row.get(0)?;
let msg_id: MsgId = row.get(1)?;
Ok((rowid, msg_id))
},
)
.await
.expect("query_row_optional failed")
.expect("No SMTP row found")
.expect("No SMTP row found");
let query_only = true;
let queued_mail = context
.sql
.transaction_ext(query_only, |transaction| {
smtp::load_queued_mail(transaction, rowid)
})
.await
.unwrap();
let rendered_mail = mimefactory::render_queued_mail_with_context(queued_mail, context)
.await
.unwrap();
(msg_id, rendered_mail.message)
}

8
src/securejoin.rs
View File

@@ -554,21 +554,23 @@ pub(crate) async fn handle_securejoin_handshake(
}
let rfc724_mid = create_outgoing_rfc724_mid();
let addr = ContactAddress::new(&mime_message.from.addr)?;
let addr = mime_message.from.addr.clone();
let attach_self_pubkey = true;
let self_fp = self_fingerprint(context).await?;
let shared_secret = format!("securejoin/{self_fp}/{auth}");
let rendered_message = mimefactory::render_symm_encrypted_securejoin_message(
let recipients = vec![addr];
let queued_message = mimefactory::symm_encrypted_securejoin_message(
context,
"vc-pubkey",
&rfc724_mid,
attach_self_pubkey,
auth,
&shared_secret,
recipients,
)
.await?;
insert_into_smtp(context, &rfc724_mid, &addr, rendered_message).await?;
insert_into_smtp(context, &rfc724_mid, &queued_message).await?;
context.scheduler.interrupt_smtp().await;
Ok(HandshakeMessage::Done)

7
src/securejoin/bob.rs
View File

@@ -325,22 +325,23 @@ pub(crate) async fn send_handshake_message(
if invite.is_v3() && matches!(step, BobHandshakeMsg::Request) {
// Send a minimal symmetrically-encrypted vc-request-pubkey message
let rfc724_mid = create_outgoing_rfc724_mid();
let recipients = invite.addrs().join(" ");
let recipients = invite.addrs();
let alice_fp = invite.fingerprint().hex();
let auth = invite.authcode();
let shared_secret = format!("securejoin/{alice_fp}/{auth}");
let attach_self_pubkey = false;
let rendered_message = mimefactory::render_symm_encrypted_securejoin_message(
let queued_msg = mimefactory::symm_encrypted_securejoin_message(
context,
"vc-request-pubkey",
&rfc724_mid,
attach_self_pubkey,
auth,
&shared_secret,
recipients.to_vec(),
)
.await?;
insert_into_smtp(context, &rfc724_mid, &queued_msg).await?;
insert_into_smtp(context, &rfc724_mid, &recipients, rendered_message).await?;
context.scheduler.interrupt_smtp().await;
} else {
let mut msg = Message {

234
src/smtp.rs
View File

@@ -3,20 +3,29 @@
mod connect;
pub mod send;
use std::collections::BTreeSet;
use anyhow::{Context as _, Error, Result, bail, format_err};
use async_smtp::response::{Category, Code, Detail};
use async_smtp::{EmailAddress, SmtpTransport};
use pgp::composed::SignedPublicKey;
use rusqlite::OptionalExtension as _;
use tokio::task;
use crate::chat;
use crate::chat::{ChatId, add_info_msg_with_cmd};
use crate::config::Config;
use crate::contact::{Contact, ContactId};
use crate::context::Context;
use crate::events::EventType;
use crate::key;
use crate::key::DcKey;
use crate::log::{LogExt, warn};
use crate::message::Message;
use crate::message::{self, MsgId};
use crate::mimefactory;
use crate::mimefactory::MimeFactory;
use crate::mimefactory::QueuedMail;
use crate::net::proxy::ProxyConfig;
use crate::net::session::SessionBufStream;
use crate::scheduler::connectivity::ConnectivityStore;
@@ -328,21 +337,17 @@ pub(crate) async fn smtp_send(
}
/// Inserts a tombstone for `rfc724_mid`
/// and queues the rendered message for SMTP sending.
/// and queues the message for SMTP sending.
pub(crate) async fn insert_into_smtp(
context: &Context,
rfc724_mid: &str,
recipients: &str,
rendered_message: String,
queued_msg: &QueuedMail,
) -> Result<()> {
let now = tools::time();
let msg_id = message::insert_tombstone(context, rfc724_mid).await?;
context
.sql
.execute(
"INSERT INTO smtp (rfc724_mid, recipients, mime, msg_id)
VALUES (?1, ?2, ?3, ?4)",
(&rfc724_mid, &recipients, &rendered_message, msg_id),
)
.transaction(|transaction| chat::enqueue_mail(transaction, now, msg_id, queued_msg, None))
.await?;
Ok(())
}
@@ -370,23 +375,31 @@ pub(crate) async fn send_msg_to_smtp(
// database.
context
.sql
.execute("UPDATE smtp SET retries=retries+1 WHERE id=?", (rowid,))
.execute("UPDATE smtp2 SET retries=retries+1 WHERE id=?", (rowid,))
.await
.context("failed to update retries count")?;
let Some((body, recipients, msg_id, retries)) = context
let Some((queued_mail, msg_id, retries)) = context
.sql
.query_row_optional(
"SELECT mime, recipients, msg_id, retries FROM smtp WHERE id=?",
(rowid,),
|row| {
let mime: String = row.get(0)?;
let recipients: String = row.get(1)?;
let msg_id: MsgId = row.get(2)?;
let retries: i64 = row.get(3)?;
Ok((mime, recipients, msg_id, retries))
},
)
.transaction_ext(true, |transaction| {
let Some((msg_id, retries)) = transaction
.query_row(
"SELECT msg_id, retries FROM smtp2 WHERE id=?",
(rowid,),
|row| {
let msg_id: MsgId = row.get(0)?;
let retries: i64 = row.get(1)?;
Ok((msg_id, retries))
},
)
.optional()?
else {
return Ok(None);
};
let queued_mail = load_queued_mail(transaction, rowid)
.with_context(|| format!("Failed to load queued mail for {rowid}"))?;
Ok(Some((queued_mail, msg_id, retries)))
})
.await?
else {
return Ok(());
@@ -394,7 +407,7 @@ pub(crate) async fn send_msg_to_smtp(
if retries > 6 {
context
.sql
.execute("DELETE FROM smtp WHERE id=?", (rowid,))
.execute("DELETE FROM smtp2 WHERE id=?", (rowid,))
.await
.context("Failed to remove message with exceeded retry limit from smtp table")?;
if let Some(mut msg) = Message::load_from_db_optional(context, msg_id).await? {
@@ -403,13 +416,22 @@ pub(crate) async fn send_msg_to_smtp(
}
return Ok(());
}
info!(
context,
"Try number {retries} to send message {msg_id} (entry {rowid}) over SMTP."
);
let mut recipients = queued_mail.recipients.clone();
if queued_mail.bcc_self {
add_self_recipients(
context,
&mut recipients,
queued_mail.encryption.is_encrypted(),
)
.await
.context("Failed to add self recipients")?;
}
let mut sent_to_set: BTreeSet<String> =
BTreeSet::from_iter(queued_mail.sent_to.iter().cloned());
let recipients_list = recipients
.split(' ')
.into_iter()
.filter(|addr| !sent_to_set.contains(AsRef::<str>::as_ref(addr)))
.filter_map(
|addr| match async_smtp::EmailAddress::new(addr.to_string()) {
Ok(addr) => Some(addr),
@@ -421,6 +443,23 @@ pub(crate) async fn send_msg_to_smtp(
)
.collect::<Vec<_>>();
let public_key = key::load_self_public_key(context).await?;
let secret_key = key::load_self_secret_key(context).await?;
let from_addr = smtp
.from
.as_ref()
.context("No From address available, likely not connected")?
.to_string();
let rendered_mail =
mimefactory::render_queued_mail(queued_mail, &public_key, &secret_key, from_addr)?;
let body = rendered_mail.message;
info!(
context,
"Try number {retries} to send message {msg_id} (entry {rowid}) over SMTP."
);
let chunk_size = context.get_max_smtp_rcpt_to().await?.max(1);
let mut unsent = recipients_list.as_slice();
let status = loop {
@@ -432,14 +471,23 @@ pub(crate) async fn send_msg_to_smtp(
if !matches!(status, SendResult::Success) || rest.is_empty() {
break status;
}
let rest_str = rest
for sent_to_addr in chunk {
let sent_to_addr_string = sent_to_addr.to_string();
if !sent_to_set.insert(sent_to_addr_string) {
error!(context, "Attempted to send to {sent_to_addr} twice.");
}
}
let sent_to_str = sent_to_set
.iter()
.map(|a| a.as_ref())
.collect::<Vec<_>>()
.collect::<Vec<&str>>()
.join(" ");
context
.sql
.execute("UPDATE smtp SET recipients=? WHERE id=?", (rest_str, rowid))
.execute(
"UPDATE smtp2 SET sent_to=? WHERE id=?",
(sent_to_str, rowid),
)
.await?;
unsent = rest;
};
@@ -449,7 +497,7 @@ pub(crate) async fn send_msg_to_smtp(
SendResult::Success => {
context
.sql
.execute("DELETE FROM smtp WHERE id=?", (rowid,))
.execute("DELETE FROM smtp2 WHERE id=?", (rowid,))
.await?;
}
SendResult::Failure(ref err) => {
@@ -493,7 +541,7 @@ pub(crate) async fn send_msg_to_smtp(
}
context
.sql
.execute("DELETE FROM smtp WHERE id=?", (rowid,))
.execute("DELETE FROM smtp2 WHERE id=?", (rowid,))
.await?;
}
};
@@ -516,7 +564,7 @@ pub(crate) async fn msg_has_pending_smtp_job(
) -> Result<bool, Error> {
context
.sql
.exists("SELECT COUNT(*) FROM smtp WHERE msg_id=?", (msg_id,))
.exists("SELECT COUNT(*) FROM smtp2 WHERE msg_id=?", (msg_id,))
.await
}
@@ -549,7 +597,7 @@ pub(crate) async fn send_smtp_messages(context: &Context, connection: &mut Smtp)
let rowids = context
.sql
.query_map_vec("SELECT id FROM smtp ORDER BY id ASC", (), |row| {
.query_map_vec("SELECT id FROM smtp2 ORDER BY id ASC", (), |row| {
let rowid: i64 = row.get(0)?;
Ok(rowid)
})
@@ -754,6 +802,122 @@ pub(crate) async fn add_self_recipients(
/// Returns true if SMTP queue is empty.
pub(crate) async fn is_queue_empty(context: &Context) -> Result<bool> {
let sending_finished = !context.sql.exists("SELECT COUNT(*) FROM smtp", ()).await?;
let sending_finished = !context.sql.exists("SELECT COUNT(*) FROM smtp2", ()).await?;
Ok(sending_finished)
}
/// Loads the queued mail from `smtp2` table and the list of recipients.
pub(crate) fn load_queued_mail(
transaction: &mut rusqlite::Transaction<'_>,
row_id: i64,
) -> Result<QueuedMail> {
let (mut queued_mail, encryption_fingerprints) = transaction
.query_row_and_then(
"
SELECT display_name,
rfc724_mid,
mime,
should_attach_pubkey,
should_compress,
should_sign,
is_encrypted,
shared_secret,
encryption_fingerprints,
recipients,
sent_to,
bcc_self
FROM smtp2 WHERE id = ?
",
(row_id,),
|row| {
let display_name: String = row.get(0)?;
let rfc724_mid: String = row.get(1)?;
let raw_message: Vec<u8> = row.get(2)?;
let should_attach_pubkey: bool = row.get(3)?;
let should_compress: bool = row.get(4)?;
let should_sign: bool = row.get(5)?;
let is_encrypted: bool = row.get(6)?;
let shared_secret: String = row.get(7)?;
let encryption_fingerprints: String = row.get(8)?;
let encryption_fingerprints: Vec<String> = if encryption_fingerprints.is_empty() {
Vec::new()
} else {
encryption_fingerprints
.split(' ')
.map(|s| s.to_string())
.collect()
};
let recipients: String = row.get(9)?;
let recipients: Vec<String> = if recipients.is_empty() {
Vec::new()
} else {
recipients.split(' ').map(|s| s.to_string()).collect()
};
debug_assert!(!recipients.iter().any(|s| s.is_empty()));
let sent_to: String = row.get(10)?;
let sent_to: Vec<String> = if sent_to.is_empty() {
Vec::new()
} else {
sent_to.split(' ').map(|s| s.to_string()).collect()
};
let bcc_self: bool = row.get(11)?;
let encryption = match (
is_encrypted,
shared_secret.is_empty(),
encryption_fingerprints.is_empty(),
) {
(false, true, true) => mimefactory::QueuedEncryption::No,
(true, false, true) => {
mimefactory::QueuedEncryption::Symmetric { shared_secret }
}
(true, true, _) => mimefactory::QueuedEncryption::Asymmetric {
// Public keys are loaded below based on the encryption fingerprints.
encryption_pubkeys: Vec::new(),
},
_ => bail!("Invalid encryption in smtp2 row"),
};
Ok::<_, anyhow::Error>((
QueuedMail {
raw_message,
display_name,
rfc724_mid,
encryption,
should_attach_pubkey,
should_compress,
should_sign,
recipients,
sent_to,
bcc_self,
},
encryption_fingerprints,
))
},
)
.with_context(|| format!("Failed to select row {row_id} from smtp2 table"))?;
if let mimefactory::QueuedEncryption::Asymmetric {
ref mut encryption_pubkeys,
} = queued_mail.encryption
{
for fingerprint in encryption_fingerprints {
let public_key_bytes: Option<Vec<u8>> = transaction
.query_row(
"SELECT public_key FROM public_keys WHERE fingerprint=?",
(fingerprint,),
|row| {
let bytes: Vec<u8> = row.get(0)?;
Ok(bytes)
},
)
.optional()
.context("Failed to select public key by fingerprint")?;
if let Some(public_key_bytes) = public_key_bytes {
let public_key = SignedPublicKey::from_slice(&public_key_bytes)?;
encryption_pubkeys.push(public_key);
}
}
}
Ok(queued_mail)
}

31
src/sql/migrations.rs
View File

@@ -2642,6 +2642,37 @@ UPDATE msgs SET state=24 WHERE state=18; -- Change OutPreparing to OutFailed.
.await?;
}
inc_and_check(&mut migration_version, 166)?;
if dbversion < migration_version {
sql.execute_migration(
"
UPDATE msgs
SET state=24, -- OutFailed
error='Message sending canceled by upgrade'
WHERE state=20 AND id IN (SELECT msg_id FROM smtp); -- OutPending
DELETE FROM smtp;
CREATE TABLE smtp2 (
id INTEGER PRIMARY KEY AUTOINCREMENT,
display_name TEXT NOT NULL,
rfc724_mid TEXT NOT NULL,
mime BLOB NOT NULL,
should_attach_pubkey INTEGER NOT NULL,
should_compress INTEGER NOT NULL,
should_sign INTEGER NOT NULL,
msg_id INTEGER NOT NULL,
recipients TEXT NOT NULL,
sent_to TEXT NOT NULL DEFAULT '',
bcc_self INTEGER NOT NULL,
is_encrypted INTEGER NOT NULL,
shared_secret TEXT NOT NULL DEFAULT '',
encryption_fingerprints TEXT NOT NULL DEFAULT '',
retries INTEGER NOT NULL DEFAULT 0
) STRICT;",
migration_version,
)
.await?;
}
let new_version = sql
.get_raw_config_int(VERSION_CFG)
.await?

96
src/test_utils.rs
View File

@@ -37,10 +37,12 @@ use crate::context::Context;
use crate::events::{Event, EventEmitter, EventType, Events};
use crate::key::{self, DcKey, self_fingerprint};
use crate::message::{Message, MessageState, MsgId};
use crate::mimefactory;
use crate::mimeparser::{MimeMessage, SystemMessage};
use crate::pgp::SeipdVersion;
use crate::receive_imf::{ReceivedMsg, receive_imf};
use crate::securejoin::{get_securejoin_qr, join_securejoin};
use crate::smtp;
use crate::smtp::msg_has_pending_smtp_job;
use crate::stock_str::StockStrings;
use crate::tools::time;
@@ -602,28 +604,46 @@ impl TestContext {
pub async fn pop_sent_msg_ext(&self, rev_order: bool) -> Option<SentMessage<'_>> {
let mut query = "
SELECT id, msg_id, mime, recipients
FROM smtp
SELECT id, msg_id
FROM smtp2
ORDER BY id"
.to_string();
if rev_order {
query += " DESC";
}
let (rowid, msg_id, payload, recipients) = self
let (rowid, msg_id) = self
.ctx
.sql
.query_row_optional(&query, (), |row| {
let rowid: i64 = row.get(0)?;
let msg_id: MsgId = row.get(1)?;
let mime: String = row.get(2)?;
let recipients: String = row.get(3)?;
Ok((rowid, msg_id, mime, recipients))
Ok((rowid, msg_id))
})
.await
.expect("query_row_optional failed")?;
let query_only = true;
let mut queued_mail = self
.ctx
.sql
.transaction_ext(query_only, |transaction| {
smtp::load_queued_mail(transaction, rowid)
})
.await
.expect("Failed to load queued mail");
if queued_mail.bcc_self {
smtp::add_self_recipients(
&self.ctx,
&mut queued_mail.recipients,
queued_mail.encryption.is_encrypted(),
)
.await
.expect("Failed to add self recipients");
}
let recipients = queued_mail.recipients.join(" ");
debug_assert!(!recipients.starts_with(" "));
self.ctx
.sql
.execute("DELETE FROM smtp WHERE id=?;", (rowid,))
.execute("DELETE FROM smtp2 WHERE id=?;", (rowid,))
.await
.expect("failed to remove job");
if !msg_has_pending_smtp_job(self, msg_id)
@@ -643,6 +663,11 @@ ORDER BY id"
.expect("Failed to update timestamp_sent");
}
let rendered_mail = mimefactory::render_queued_mail_with_context(queued_mail, self)
.await
.expect("Failed to render queued mail");
let payload = rendered_mail.message;
let payload_headers = payload.split("\r\n\r\n").next().unwrap().lines();
let payload_header_names: Vec<_> = payload_headers
.map(|h| h.split(':').next().unwrap())
@@ -681,31 +706,56 @@ ORDER BY id"
}
pub async fn get_smtp_rows_for_msg<'a>(&'a self, msg_id: MsgId) -> Vec<SentMessage<'a>> {
let sent_msgs = self
let mut sent_msgs = Vec::new();
for rowid in self
.ctx
.sql
.query_map_vec(
"SELECT mime, recipients FROM smtp WHERE msg_id=?",
(msg_id,),
|row| {
let mime: String = row.get(0)?;
let recipients: String = row.get(1)?;
Ok((mime, recipients))
},
)
.query_map_vec("SELECT id FROM smtp2 WHERE msg_id=?", (msg_id,), |row| {
let rowid: i64 = row.get(0)?;
Ok(rowid)
})
.await
.unwrap()
.into_iter()
.map(|(mime, recipients)| SentMessage {
payload: mime,
{
let query_only = true;
let mut queued_mail = self
.ctx
.sql
.transaction_ext(query_only, |transaction| {
smtp::load_queued_mail(transaction, rowid)
})
.await
.expect("Failed to load queued mail");
if queued_mail.bcc_self {
smtp::add_self_recipients(
&self.ctx,
&mut queued_mail.recipients,
queued_mail.encryption.is_encrypted(),
)
.await
.expect("Failed to add self recipients");
}
let recipients = queued_mail.recipients.join(" ");
let rendered_mail = mimefactory::render_queued_mail_with_context(queued_mail, self)
.await
.expect("Failed to render queued mail");
let payload = rendered_mail.message;
debug_assert!(!recipients.starts_with(" "));
let sent_message = SentMessage {
payload,
sender_msg_id: msg_id,
sender_context: &self.ctx,
recipients,
})
.collect();
};
sent_msgs.push(sent_message)
}
self.ctx
.sql
.execute("DELETE FROM smtp WHERE msg_id=?", (msg_id,))
.execute("DELETE FROM smtp2 WHERE msg_id=?", (msg_id,))
.await
.expect("Delete smtp jobs");
if msg_id

4
src/tests/pre_messages/sending.rs
View File

@@ -346,7 +346,7 @@ async fn test_render_webxdc_status_update_object_range() -> Result<()> {
.unwrap();
t.pop_sent_msg().await;
assert_eq!(t.sql.count("SELECT COUNT(*) FROM smtp", ()).await?, 0);
assert_eq!(t.sql.count("SELECT COUNT(*) FROM smtp2", ()).await?, 0);
let long_text = String::from_utf8(vec![b'a'; 300_000])?;
assert!(long_text.len() > PRE_MSG_ATTACHMENT_SIZE_THRESHOLD.try_into().unwrap());
@@ -354,6 +354,6 @@ async fn test_render_webxdc_status_update_object_range() -> Result<()> {
.await?;
t.flush_status_updates().await?;
assert_eq!(t.sql.count("SELECT COUNT(*) FROM smtp", ()).await?, 1);
assert_eq!(t.sql.count("SELECT COUNT(*) FROM smtp2", ()).await?, 1);
Ok(())
}