fix(store): survive one corrupt body on bulk read paths #429
No files matched your search
@@ -875,6 +875,11 @@ impl HeadlessRunner {
|
||||
// Follow `next_page_token` to the end so the command shows the whole
|
||||
// folder, transparently exercising the keyset pagination convention.
|
||||
let mut messages = Vec::new();
|
||||
// Messages the daemon could not return because their stored bodies
|
||||
// failed authentication. Reported rather than quietly missing: a
|
||||
// listing that is one message short with no explanation is a lie of
|
||||
// omission, and this is the only place the user would ever see it.
|
||||
let mut unreadable: Vec<String> = Vec::new();
|
||||
let mut page_token = String::new();
|
||||
loop {
|
||||
match client
|
||||
@@ -888,6 +893,7 @@ impl HeadlessRunner {
|
||||
Ok(response) => {
|
||||
let page = response.into_inner();
|
||||
messages.extend(page.messages);
|
||||
unreadable.extend(page.unreadable_message_ids);
|
||||
if page.next_page_token.is_empty() {
|
||||
break;
|
||||
}
|
||||
@@ -904,10 +910,20 @@ impl HeadlessRunner {
|
||||
messages.iter().map(message_proto_to_json).collect();
|
||||
format_json(&json!({
|
||||
"folder": folder,
|
||||
"messages": messages_json
|
||||
"messages": messages_json,
|
||||
"unreadable_message_ids": unreadable
|
||||
}))
|
||||
} else {
|
||||
} else if unreadable.is_empty() {
|
||||
format!("Folder '{}': {} message(s) found", folder, messages.len())
|
||||
} else {
|
||||
format!(
|
||||
"Folder '{}': {} message(s) found; {} omitted (stored body failed \
|
||||
authentication): {}",
|
||||
folder,
|
||||
messages.len(),
|
||||
unreadable.len(),
|
||||
unreadable.join(", ")
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1450,12 +1466,26 @@ impl HeadlessRunner {
|
||||
"output_path": resp.output_path,
|
||||
"message_count": resp.message_count,
|
||||
"bytes_written": resp.bytes_written,
|
||||
"unreadable_message_ids": resp.unreadable_message_ids,
|
||||
}))
|
||||
} else {
|
||||
} else if resp.unreadable_message_ids.is_empty() {
|
||||
format!(
|
||||
"Exported {} message(s) to '{}' ({} bytes)",
|
||||
resp.message_count, resp.output_path, resp.bytes_written
|
||||
)
|
||||
} else {
|
||||
// Named, not just counted: this archive outlives the store,
|
||||
// so the user needs to be able to record exactly which mail
|
||||
// it does not contain.
|
||||
format!(
|
||||
"Exported {} message(s) to '{}' ({} bytes); {} message(s) omitted \
|
||||
(stored body failed authentication): {}",
|
||||
resp.message_count,
|
||||
resp.output_path,
|
||||
resp.bytes_written,
|
||||
resp.unreadable_message_ids.len(),
|
||||
resp.unreadable_message_ids.join(", ")
|
||||
)
|
||||
}
|
||||
}
|
||||
Err(status) => Self::render_status_error("export_mailbox", &status, json_mode),
|
||||
@@ -2018,6 +2048,7 @@ impl HeadlessRunner {
|
||||
"actions_applied_count": p.actions_applied_count,
|
||||
"last_message_id": p.last_message_id,
|
||||
"done": p.done,
|
||||
"unreadable_count": p.unreadable_count,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
@@ -2025,8 +2056,11 @@ impl HeadlessRunner {
|
||||
} else {
|
||||
match updates.last() {
|
||||
Some(last) => format!(
|
||||
"Triage complete: scanned={} matched={} actions_applied={}",
|
||||
last.scanned_count, last.matched_count, last.actions_applied_count
|
||||
"Triage complete: scanned={} matched={} actions_applied={} unreadable={}",
|
||||
last.scanned_count,
|
||||
last.matched_count,
|
||||
last.actions_applied_count,
|
||||
last.unreadable_count
|
||||
),
|
||||
None => "Triage produced no progress updates.".to_string(),
|
||||
}
|
||||
@@ -3844,6 +3878,7 @@ mod tests {
|
||||
let mut message = stub_message();
|
||||
message.folder_id = req.folder_id;
|
||||
Ok(tonic::Response::new(ListMessagesResponse {
|
||||
unreadable_message_ids: Vec::new(),
|
||||
messages: vec![message],
|
||||
next_page_token: String::new(),
|
||||
}))
|
||||
@@ -5055,6 +5090,7 @@ mod tests {
|
||||
) -> Result<tonic::Response<Self::TriageStream>, tonic::Status> {
|
||||
let updates = vec![
|
||||
Ok(TriageProgress {
|
||||
unreadable_count: 0,
|
||||
scanned_count: 2,
|
||||
matched_count: 1,
|
||||
actions_applied_count: 1,
|
||||
@@ -5062,6 +5098,7 @@ mod tests {
|
||||
done: false,
|
||||
}),
|
||||
Ok(TriageProgress {
|
||||
unreadable_count: 0,
|
||||
scanned_count: 3,
|
||||
matched_count: 2,
|
||||
actions_applied_count: 2,
|
||||
@@ -5377,6 +5414,7 @@ mod tests {
|
||||
let output_path = req.output_path.clone();
|
||||
*self.last_request.lock().unwrap_or_else(|e| e.into_inner()) = Some(req);
|
||||
Ok(tonic::Response::new(ExportResponse {
|
||||
unreadable_message_ids: Vec::new(),
|
||||
output_path,
|
||||
message_count: 7,
|
||||
bytes_written: 4096,
|
||||
|
||||
Binary file not shown.
@@ -695,6 +695,16 @@ message ListMessagesResponse {
|
||||
repeated Message messages = 1;
|
||||
// Opaque cursor for the next page, or empty when this is the last page.
|
||||
string next_page_token = 2;
|
||||
// IDs of messages that occupied this page but could not be returned: their
|
||||
// stored body ciphertext failed authentication (tampering or bit-rot).
|
||||
//
|
||||
// One unreadable row does not fail the page -- the other messages are still
|
||||
// listed -- but it is never returned with an empty or substituted body
|
||||
// either, because that is indistinguishable from a genuinely empty message.
|
||||
// It is named here instead, so a caller can say what it is not showing.
|
||||
// `GetMessage` on one of these IDs still fails, which is the honest answer
|
||||
// when the whole point of the call is that one message's content.
|
||||
repeated string unreadable_message_ids = 3;
|
||||
}
|
||||
|
||||
// GetMessageRequest.
|
||||
@@ -1220,6 +1230,15 @@ message TriageProgress {
|
||||
// True only on the final message of the stream, once the whole store has
|
||||
// been scanned.
|
||||
bool done = 5;
|
||||
// Total messages walked but not evaluated, because their stored body
|
||||
// ciphertext failed authentication.
|
||||
//
|
||||
// A count rather than a list: this accumulates over a scan of the entire
|
||||
// store and has no bound worth streaming. Each one is logged by name at
|
||||
// WARN. Rules are never evaluated against a body the store could not
|
||||
// authenticate -- a filter that acts on invented content is worse than one
|
||||
// that reports it did not run.
|
||||
uint64 unreadable_count = 6;
|
||||
}
|
||||
|
||||
// Export exposes the daemon's real mailbox export engine
|
||||
@@ -1287,6 +1306,15 @@ message ExportResponse {
|
||||
uint64 message_count = 2;
|
||||
// Total bytes actually written to `output_path`.
|
||||
uint64 bytes_written = 3;
|
||||
// IDs of messages in scope that were NOT written, because their stored body
|
||||
// ciphertext failed authentication.
|
||||
//
|
||||
// An export is the artifact that outlives the store, so the gap in it is
|
||||
// named rather than merely counted: the caller can record exactly which mail
|
||||
// its archive does not contain. Writing them with an empty or placeholder
|
||||
// body instead would put text into an archive that the store never held, and
|
||||
// no later reader could tell it apart from the real thing.
|
||||
repeated string unreadable_message_ids = 4;
|
||||
}
|
||||
|
||||
// Calendar exposes the READ path over the daemon's real, persistent calendar
|
||||
|
||||
+444
-130
@@ -323,17 +323,87 @@ pub struct SaveOutcome {
|
||||
/// as the only place both can reach.
|
||||
pub use nuncio_core::model::PlacementKey;
|
||||
|
||||
/// A message paired with the mailbox occupancy it was read through.
|
||||
/// A stored row whose body ciphertext would not authenticate on read.
|
||||
///
|
||||
/// Carries only columns the store keeps in the clear -- the message key and its
|
||||
/// receipt time. It never carries body content, because the body is precisely
|
||||
/// what failed to authenticate and is therefore not trustworthy enough to hand
|
||||
/// on, log, or index.
|
||||
///
|
||||
/// `received_at` travels with the key so an unreadable row can still hold its
|
||||
/// slot in an ordering or a keyset page. Dropping it from the sequence outright
|
||||
/// would move every later row forward and make a keyset cursor point at the
|
||||
/// wrong place.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct UnreadableBody {
|
||||
/// Store key of the row whose body would not authenticate.
|
||||
pub message_key: String,
|
||||
/// The row's receipt time, in the same units as `Email::received_at`.
|
||||
pub received_at: i64,
|
||||
}
|
||||
|
||||
/// One row of a **bulk** message read: `Ok` when the stored body authenticated,
|
||||
/// `Err` when it did not.
|
||||
///
|
||||
/// Single-item and bulk reads answer different questions, and this type is why
|
||||
/// they are allowed to answer them differently.
|
||||
///
|
||||
/// [`DatabaseEngine::get_message`] is asked for one named message. It fails
|
||||
/// closed -- returning `Err` for the whole call -- because a message handed back
|
||||
/// with an empty body is indistinguishable from a genuinely empty one, which
|
||||
/// would hide exactly the tampering the AEAD exists to detect.
|
||||
///
|
||||
/// A listing, a rescan or an export is asked for a *set*. Failing the whole set
|
||||
/// over one bit-rotted row hides the other forty-nine just as effectively as an
|
||||
/// empty body hides the one. So the bulk paths keep every row in its slot and
|
||||
/// refuse to invent a body for the ones that will not authenticate: the caller
|
||||
/// receives `Err(UnreadableBody)` for those rows and intact messages for the
|
||||
/// rest. Because the failure rides in the row rather than in the call, a caller
|
||||
/// cannot consume the page without meeting it.
|
||||
pub type MessageRead = Result<nuncio_core::model::Email, UnreadableBody>;
|
||||
|
||||
/// The identity columns every bulk-read row carries, readable body or not.
|
||||
///
|
||||
/// Lets ordering, keyset cursors and progress counters treat an unreadable row
|
||||
/// as the row it still is, without first having to know whether its body
|
||||
/// decrypted.
|
||||
pub trait BulkReadRow {
|
||||
/// Store key of this row.
|
||||
fn message_key(&self) -> &str;
|
||||
/// Receipt time of this row.
|
||||
fn received_at(&self) -> i64;
|
||||
}
|
||||
|
||||
impl BulkReadRow for MessageRead {
|
||||
fn message_key(&self) -> &str {
|
||||
match self {
|
||||
Ok(email) => &email.id,
|
||||
Err(unreadable) => &unreadable.message_key,
|
||||
}
|
||||
}
|
||||
|
||||
fn received_at(&self) -> i64 {
|
||||
match self {
|
||||
Ok(email) => email.received_at,
|
||||
Err(unreadable) => unreadable.received_at,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A bulk-read message row paired with the mailbox occupancy it was read
|
||||
/// through.
|
||||
///
|
||||
/// The two travel together out of every folder-scoped read: the caller almost
|
||||
/// always needs the per-mailbox facts (the read flag, the addressing UID) as
|
||||
/// well as the message, and re-deriving them would cost a query per row.
|
||||
/// well as the message, and re-deriving them would cost a query per row. The
|
||||
/// placement is present even for a row whose body would not authenticate --
|
||||
/// where the row sits is not encrypted, and the caller still needs it to page.
|
||||
///
|
||||
/// Deliberately not called `PlacedMessage`: `nuncio_mail::PlacedMessage` is a
|
||||
/// different type carrying a third field (the identity tier a backend derived
|
||||
/// the key from), and the sync path imports both. One name for two shapes in
|
||||
/// one call site is a trap worth a longer name to avoid.
|
||||
pub type MessageWithPlacement = (nuncio_core::model::Email, nuncio_core::model::Placement);
|
||||
pub type MessageWithPlacement = (MessageRead, nuncio_core::model::Placement);
|
||||
|
||||
/// Keyset cursor for [`DatabaseEngine::list_messages_page`]: the
|
||||
/// `(received_at, message_key, uidvalidity, uid)` of the last row of a page.
|
||||
@@ -377,6 +447,15 @@ type MessageWithPlacementRow = (
|
||||
i64,
|
||||
);
|
||||
|
||||
/// Drop a joined row's placement columns, leaving the identity-only
|
||||
/// [`MessageRow`] the body decoders take. Consumes the row so the message
|
||||
/// columns move rather than clone.
|
||||
fn narrow_placed_row(row: MessageWithPlacementRow) -> MessageRow {
|
||||
(
|
||||
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9,
|
||||
)
|
||||
}
|
||||
|
||||
/// What a [`DatabaseEngine::delete_placements`] call removed.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct DeleteOutcome {
|
||||
@@ -1235,12 +1314,31 @@ impl DatabaseEngine {
|
||||
.await
|
||||
.map_err(DatabaseError::Query)?;
|
||||
|
||||
let mut unreadable = 0usize;
|
||||
for (message_key, subject, sender, body_plain) in rows {
|
||||
let dec_plain = body_plain
|
||||
let dec_plain = match body_plain
|
||||
.map(|p| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p))
|
||||
.transpose()
|
||||
.map_err(|e| DatabaseError::Decryption(e.to_string()))?
|
||||
.unwrap_or_default();
|
||||
{
|
||||
Ok(plain) => plain.unwrap_or_default(),
|
||||
Err(e) => {
|
||||
// This runs on every open, so failing the whole backfill on
|
||||
// one bit-rotted row would stop the daemon booting at all.
|
||||
// The row is left out of the index entirely rather than
|
||||
// indexed with an empty body: an FTS entry marks a message
|
||||
// permanently indexed, so an empty one would make its body
|
||||
// permanently unsearchable with nothing left to report it.
|
||||
// Leaving it out means every subsequent open re-reports it,
|
||||
// and a repaired row indexes itself without intervention.
|
||||
unreadable += 1;
|
||||
tracing::warn!(
|
||||
message_key = %message_key,
|
||||
error = %e,
|
||||
"stored message body failed authentication; leaving it out of the search index"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO messages_fts (message_key, subject, sender, body_plain) VALUES (?, ?, ?, ?)",
|
||||
@@ -1254,6 +1352,14 @@ impl DatabaseEngine {
|
||||
.map_err(DatabaseError::Query)?;
|
||||
}
|
||||
|
||||
if unreadable > 0 {
|
||||
tracing::warn!(
|
||||
unreadable,
|
||||
"search index backfill skipped messages whose stored bodies failed \
|
||||
authentication; their bodies are not searchable until the rows are repaired"
|
||||
);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -1879,6 +1985,11 @@ impl DatabaseEngine {
|
||||
/// The placement travels with the message because the caller almost always
|
||||
/// needs the per-mailbox facts -- the read flag and the addressing UID --
|
||||
/// and re-deriving them would mean a second query per row.
|
||||
///
|
||||
/// A row whose stored body will not authenticate comes back as
|
||||
/// `Err(UnreadableBody)` in its own slot rather than failing the listing --
|
||||
/// see [`MessageRead`] for why a bulk read differs from
|
||||
/// [`Self::get_message`] here.
|
||||
pub async fn list_messages(
|
||||
&self,
|
||||
account_id: &str,
|
||||
@@ -1906,39 +2017,14 @@ impl DatabaseEngine {
|
||||
|
||||
let mut out = Vec::with_capacity(rows.len());
|
||||
for row in rows {
|
||||
let dec_plain = row
|
||||
.6
|
||||
.map(|p| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p))
|
||||
.transpose()
|
||||
.map_err(|e| DatabaseError::Decryption(e.to_string()))?;
|
||||
let dec_html = row
|
||||
.7
|
||||
.map(|h| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &h))
|
||||
.transpose()
|
||||
.map_err(|e| DatabaseError::Decryption(e.to_string()))?;
|
||||
|
||||
out.push((
|
||||
nuncio_core::model::Email {
|
||||
id: row.0,
|
||||
account_id: row.1,
|
||||
subject: row.2,
|
||||
sender: row.3,
|
||||
recipient: row.4,
|
||||
received_at: row.5,
|
||||
body_plain: dec_plain,
|
||||
body_html: dec_html,
|
||||
attachments: Vec::new(),
|
||||
message_id: row.8,
|
||||
content_hash: row.9,
|
||||
},
|
||||
nuncio_core::model::Placement {
|
||||
account_id: account_id.to_string(),
|
||||
folder_id: folder_id.to_string(),
|
||||
uid_validity: row.10,
|
||||
remote_id: row.11,
|
||||
read: row.12 != 0,
|
||||
},
|
||||
));
|
||||
let placement = nuncio_core::model::Placement {
|
||||
account_id: account_id.to_string(),
|
||||
folder_id: folder_id.to_string(),
|
||||
uid_validity: row.10.clone(),
|
||||
remote_id: row.11.clone(),
|
||||
read: row.12 != 0,
|
||||
};
|
||||
out.push((self.bulk_email_from_row(narrow_placed_row(row)), placement));
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
@@ -1962,6 +2048,11 @@ impl DatabaseEngine {
|
||||
/// `uidvalidity`/`uid` are compared as the text they are stored as, which is
|
||||
/// an arbitrary but total order -- and matching the `ORDER BY` exactly is
|
||||
/// all a keyset needs.
|
||||
///
|
||||
/// A row whose stored body will not authenticate occupies its slot in the
|
||||
/// page as `Err(UnreadableBody)` -- see [`MessageRead`]. Keeping the slot is
|
||||
/// what lets the keyset stay exact: the page is still as long as the query's
|
||||
/// window, so the cursor still names a real position in the sequence.
|
||||
pub async fn list_messages_page(
|
||||
&self,
|
||||
account_id: &str,
|
||||
@@ -2018,45 +2109,27 @@ impl DatabaseEngine {
|
||||
let has_more = rows.len() > page_size;
|
||||
let mut placed = Vec::with_capacity(rows.len().min(page_size));
|
||||
for row in rows.into_iter().take(page_size) {
|
||||
let dec_plain = row
|
||||
.6
|
||||
.map(|p| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p))
|
||||
.transpose()
|
||||
.map_err(|e| DatabaseError::Decryption(e.to_string()))?;
|
||||
let dec_html = row
|
||||
.7
|
||||
.map(|h| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &h))
|
||||
.transpose()
|
||||
.map_err(|e| DatabaseError::Decryption(e.to_string()))?;
|
||||
placed.push((
|
||||
nuncio_core::model::Email {
|
||||
id: row.0,
|
||||
account_id: row.1,
|
||||
subject: row.2,
|
||||
sender: row.3,
|
||||
recipient: row.4,
|
||||
received_at: row.5,
|
||||
body_plain: dec_plain,
|
||||
body_html: dec_html,
|
||||
attachments: Vec::new(),
|
||||
message_id: row.8,
|
||||
content_hash: row.9,
|
||||
},
|
||||
nuncio_core::model::Placement {
|
||||
account_id: account_id.to_string(),
|
||||
folder_id: folder_id.to_string(),
|
||||
uid_validity: row.10,
|
||||
remote_id: row.11,
|
||||
read: row.12 != 0,
|
||||
},
|
||||
));
|
||||
let placement = nuncio_core::model::Placement {
|
||||
account_id: account_id.to_string(),
|
||||
folder_id: folder_id.to_string(),
|
||||
uid_validity: row.10.clone(),
|
||||
remote_id: row.11.clone(),
|
||||
read: row.12 != 0,
|
||||
};
|
||||
// An unreadable row stays in the page rather than being filtered
|
||||
// out of it. The keyset is positional: dropping rows here would
|
||||
// make the page shorter than the query's own window, so `has_more`
|
||||
// and the cursor below would both describe a sequence the store
|
||||
// never returned, and rows between the last surviving row and the
|
||||
// query's real boundary would become unreachable.
|
||||
placed.push((self.bulk_email_from_row(narrow_placed_row(row)), placement));
|
||||
}
|
||||
|
||||
let next = if has_more {
|
||||
placed.last().map(|(e, p)| {
|
||||
placed.last().map(|(read, p)| {
|
||||
(
|
||||
e.received_at,
|
||||
e.id.clone(),
|
||||
read.received_at(),
|
||||
read.message_key().to_string(),
|
||||
p.uid_validity.clone(),
|
||||
p.remote_id.clone(),
|
||||
)
|
||||
@@ -2089,30 +2162,11 @@ impl DatabaseEngine {
|
||||
.await
|
||||
.map_err(DatabaseError::Query)?;
|
||||
|
||||
let dec_plain = row
|
||||
.6
|
||||
.map(|p| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &p))
|
||||
.transpose()
|
||||
.map_err(|e| DatabaseError::Decryption(e.to_string()))?;
|
||||
let dec_html = row
|
||||
.7
|
||||
.map(|h| crate::cipher::PayloadCipher::decrypt_text_at_rest(&self.storage_key, &h))
|
||||
.transpose()
|
||||
.map_err(|e| DatabaseError::Decryption(e.to_string()))?;
|
||||
|
||||
Ok(nuncio_core::model::Email {
|
||||
id: row.0,
|
||||
account_id: row.1,
|
||||
subject: row.2,
|
||||
sender: row.3,
|
||||
recipient: row.4,
|
||||
received_at: row.5,
|
||||
body_plain: dec_plain,
|
||||
body_html: dec_html,
|
||||
attachments: Vec::new(),
|
||||
message_id: row.8,
|
||||
content_hash: row.9,
|
||||
})
|
||||
// Fail-closed on purpose, and deliberately NOT the bulk decoder: this
|
||||
// call names one message, so there is no other row for the caller to
|
||||
// still receive, and a message returned with an empty body reads
|
||||
// exactly like a genuinely empty one.
|
||||
self.email_from_row(row)
|
||||
}
|
||||
|
||||
/// Every placement currently stored for one folder of one account.
|
||||
@@ -3182,11 +3236,18 @@ impl DatabaseEngine {
|
||||
///
|
||||
/// Walks message identities, not placements: a chunked pass over the store
|
||||
/// wants each message exactly once, however many folders it occupies.
|
||||
///
|
||||
/// A row whose stored body will not authenticate is returned as
|
||||
/// `Err(UnreadableBody)` in its own slot -- see [`MessageRead`]. Keeping the
|
||||
/// slot matters twice over here: a scan pages on chunk length and resumes
|
||||
/// from the last key it saw, so silently shortening a chunk would both
|
||||
/// declare the scan finished early and rewind the cursor past rows it never
|
||||
/// looked at.
|
||||
pub async fn get_message_chunk(
|
||||
&self,
|
||||
last_key: &str,
|
||||
limit: usize,
|
||||
) -> Result<Vec<nuncio_core::model::Email>, DatabaseError> {
|
||||
) -> Result<Vec<MessageRead>, DatabaseError> {
|
||||
let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new(
|
||||
"SELECT message_key, account_id, subject, sender, recipient, received_at, \
|
||||
body_plain, body_html, message_id, content_hash FROM messages ",
|
||||
@@ -3205,7 +3266,10 @@ impl DatabaseEngine {
|
||||
.await
|
||||
.map_err(DatabaseError::Query)?;
|
||||
|
||||
rows.into_iter().map(|r| self.email_from_row(r)).collect()
|
||||
Ok(rows
|
||||
.into_iter()
|
||||
.map(|r| self.bulk_email_from_row(r))
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Query messages across the WHOLE store for export purposes,
|
||||
@@ -3217,11 +3281,18 @@ impl DatabaseEngine {
|
||||
/// The folder filter is an `EXISTS` over `placements` rather than a join:
|
||||
/// a message occupying two folders must still export as one message, and a
|
||||
/// join would emit it once per matching placement.
|
||||
///
|
||||
/// A row whose stored body will not authenticate is returned as
|
||||
/// `Err(UnreadableBody)` rather than failing the export -- see
|
||||
/// [`MessageRead`]. An archive is the one artifact a caller keeps after the
|
||||
/// store is gone, so the caller is handed the key of every message it will
|
||||
/// not contain and can name the gap; what it must never do is write a
|
||||
/// message with a body the store could not authenticate.
|
||||
pub async fn list_messages_for_export(
|
||||
&self,
|
||||
account_id: Option<&str>,
|
||||
folder_id: Option<&str>,
|
||||
) -> Result<Vec<nuncio_core::model::Email>, DatabaseError> {
|
||||
) -> Result<Vec<MessageRead>, DatabaseError> {
|
||||
let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new(
|
||||
"SELECT m.message_key, m.account_id, m.subject, m.sender, m.recipient, \
|
||||
m.received_at, m.body_plain, m.body_html, m.message_id, m.content_hash \
|
||||
@@ -3257,7 +3328,35 @@ impl DatabaseEngine {
|
||||
.await
|
||||
.map_err(DatabaseError::Query)?;
|
||||
|
||||
rows.into_iter().map(|r| self.email_from_row(r)).collect()
|
||||
Ok(rows
|
||||
.into_iter()
|
||||
.map(|r| self.bulk_email_from_row(r))
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// The bulk-read counterpart to [`Self::email_from_row`]: the same
|
||||
/// decryption, but a body that will not authenticate is reported as
|
||||
/// [`UnreadableBody`] for that row alone instead of failing the whole read.
|
||||
///
|
||||
/// The failure is logged at `warn!` naming the message key, so an operator
|
||||
/// watching the daemon sees which row is at fault rather than a listing that
|
||||
/// is quietly one message short. The log deliberately carries the key and
|
||||
/// the cipher's own error only -- never body content, which is the exact
|
||||
/// material that failed to authenticate and must not be echoed anywhere.
|
||||
fn bulk_email_from_row(&self, row: MessageRow) -> MessageRead {
|
||||
let message_key = row.0.clone();
|
||||
let received_at = row.5;
|
||||
self.email_from_row(row).map_err(|e| {
|
||||
tracing::warn!(
|
||||
message_key = %message_key,
|
||||
error = %e,
|
||||
"stored message body failed authentication; returning this row as unreadable"
|
||||
);
|
||||
UnreadableBody {
|
||||
message_key,
|
||||
received_at,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Rebuild an [`nuncio_core::model::Email`] from the identity-column row
|
||||
@@ -3552,6 +3651,13 @@ fn compute_log_hash(
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// The message behind a bulk-read row that the fixture guarantees is
|
||||
/// readable. Panics on an unreadable row so a test asserting on message
|
||||
/// content can never silently pass over one that failed to authenticate.
|
||||
fn readable(read: &MessageRead) -> &nuncio_core::model::Email {
|
||||
read.as_ref().expect("this fixture row's body is readable")
|
||||
}
|
||||
use sqlx::Connection;
|
||||
|
||||
/// Proves that a transient error surfacing from the integrity probe (here, a
|
||||
@@ -3831,7 +3937,7 @@ mod tests {
|
||||
.await
|
||||
.expect("list messages succeeds");
|
||||
assert_eq!(msgs.len(), 1);
|
||||
assert_eq!(msgs[0].0.id, "msg-db-100");
|
||||
assert_eq!(readable(&msgs[0].0).id, "msg-db-100");
|
||||
assert_eq!(msgs[0].1.remote_id, "100");
|
||||
|
||||
let folders = engine
|
||||
@@ -3844,9 +3950,13 @@ mod tests {
|
||||
}
|
||||
|
||||
/// A `body_plain` column corrupted after a legitimate write (simulating tampering or
|
||||
/// bit-rot) must make `get_message` and `list_messages` return
|
||||
/// bit-rot) must make the SINGLE-ITEM `get_message` return
|
||||
/// `Err(DatabaseError::Decryption)`, never a silently-empty body -- that would defeat
|
||||
/// the AEAD's integrity guarantee exactly where it matters: detecting tampering on read.
|
||||
///
|
||||
/// The bulk paths are covered by
|
||||
/// `one_corrupt_body_does_not_destroy_a_bulk_read`, which asserts the other half of
|
||||
/// the contract: they must not fail closed over one row, and must not hide it either.
|
||||
#[tokio::test]
|
||||
async fn corrupted_body_ciphertext_fails_closed_on_read() {
|
||||
let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
|
||||
@@ -3906,20 +4016,218 @@ mod tests {
|
||||
.expect_err("get_message must fail closed on corrupted ciphertext");
|
||||
assert!(matches!(err, DatabaseError::Decryption(_)));
|
||||
|
||||
let err = engine
|
||||
// The bulk paths deliberately do NOT fail the whole call here -- but the one
|
||||
// corrupt row still never comes back carrying a body.
|
||||
let listed = engine
|
||||
.list_messages("acct-1", "INBOX", 10)
|
||||
.await
|
||||
.expect_err("list_messages must fail closed on corrupted ciphertext");
|
||||
assert!(matches!(err, DatabaseError::Decryption(_)));
|
||||
.expect("a bulk listing must survive one unreadable row");
|
||||
assert_eq!(listed.len(), 1);
|
||||
assert_eq!(
|
||||
listed[0].0.as_ref().err().map(|u| u.message_key.as_str()),
|
||||
Some("msg-db-corrupt"),
|
||||
"the corrupt row must be reported as unreadable, not returned with a body"
|
||||
);
|
||||
}
|
||||
|
||||
// The export path reads the same ciphertext column and must fail closed too --
|
||||
// an export that silently dropped an unauthenticated body would write a file
|
||||
// the user could not tell apart from a genuinely empty message.
|
||||
let err = engine
|
||||
/// Corrupt the stored `body_plain` ciphertext of one message in place, leaving it
|
||||
/// valid hex so the failure is an AEAD authentication failure rather than a decode
|
||||
/// error -- the same shape a flipped bit on disk or a tampering edit would produce.
|
||||
async fn corrupt_stored_body(engine: &DatabaseEngine, message_key: &str) {
|
||||
let (stored,): (String,) =
|
||||
sqlx::query_as("SELECT body_plain FROM messages WHERE message_key = ?")
|
||||
.bind(message_key)
|
||||
.fetch_one(&engine.pool)
|
||||
.await
|
||||
.expect("fetch stored ciphertext");
|
||||
let mut bytes = hex::decode(&stored).expect("stored ciphertext is valid hex");
|
||||
let last_idx = bytes.len() - 1;
|
||||
bytes[last_idx] ^= 0xFF;
|
||||
|
||||
sqlx::query("UPDATE messages SET body_plain = ? WHERE message_key = ?")
|
||||
.bind(hex::encode(bytes))
|
||||
.bind(message_key)
|
||||
.execute(&engine.pool)
|
||||
.await
|
||||
.expect("corrupt stored ciphertext");
|
||||
}
|
||||
|
||||
/// Split a bulk read into the keys that came back whole and the keys reported
|
||||
/// unreadable, asserting along the way that nothing arrived with a fabricated body.
|
||||
fn split_bulk_read(rows: &[MessageRead]) -> (Vec<&str>, Vec<&str>) {
|
||||
let mut good = Vec::new();
|
||||
let mut bad = Vec::new();
|
||||
for row in rows {
|
||||
match row {
|
||||
Ok(email) => {
|
||||
assert!(
|
||||
email.body_plain.as_deref().is_some_and(|b| !b.is_empty()),
|
||||
"a readable row must carry its real body, never an empty one"
|
||||
);
|
||||
good.push(email.id.as_str());
|
||||
}
|
||||
Err(unreadable) => bad.push(unreadable.message_key.as_str()),
|
||||
}
|
||||
}
|
||||
(good, bad)
|
||||
}
|
||||
|
||||
/// Store `keys` as messages in one folder, with a distinct body per position.
|
||||
async fn save_bulk_fixture(engine: &DatabaseEngine, keys: &[&str]) {
|
||||
for (idx, key) in keys.iter().enumerate() {
|
||||
let email = nuncio_core::model::Email {
|
||||
id: (*key).to_string(),
|
||||
account_id: "acct-1".to_string(),
|
||||
subject: format!("Message {idx}"),
|
||||
sender: "alice@nuncio.mx".to_string(),
|
||||
recipient: "bob@nuncio.mx".to_string(),
|
||||
received_at: 1_700_000_000 + idx as i64,
|
||||
body_plain: Some(format!("body number {idx}")),
|
||||
body_html: None,
|
||||
attachments: Vec::new(),
|
||||
message_id: None,
|
||||
content_hash: None,
|
||||
};
|
||||
let placement = nuncio_core::model::Placement {
|
||||
account_id: "acct-1".to_string(),
|
||||
folder_id: "INBOX".to_string(),
|
||||
uid_validity: "1".to_string(),
|
||||
remote_id: idx.to_string(),
|
||||
read: false,
|
||||
};
|
||||
engine
|
||||
.save_email_at(
|
||||
&email,
|
||||
nuncio_core::model::IdentitySource::Surrogate,
|
||||
&placement,
|
||||
)
|
||||
.await
|
||||
.expect("save email succeeds");
|
||||
}
|
||||
}
|
||||
|
||||
/// The counterpart to `corrupted_body_ciphertext_fails_closed_on_read`: one corrupt
|
||||
/// row must not destroy a whole page, a whole rescan, or a whole export -- and must
|
||||
/// not vanish from them either.
|
||||
///
|
||||
/// Three messages are stored and the middle one's ciphertext is corrupted. Every bulk
|
||||
/// path must return the two good messages, account for the bad one by key, and never
|
||||
/// hand back a row carrying a body it could not authenticate. The single-item read of
|
||||
/// that same message must still fail closed, which is what makes the two behaviours a
|
||||
/// deliberate distinction rather than an inconsistency.
|
||||
#[tokio::test]
|
||||
async fn one_corrupt_body_does_not_destroy_a_bulk_read() {
|
||||
let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
|
||||
save_bulk_fixture(&engine, &["msg-good-1", "msg-rotten", "msg-good-2"]).await;
|
||||
corrupt_stored_body(&engine, "msg-rotten").await;
|
||||
|
||||
let listed = engine
|
||||
.list_messages("acct-1", "INBOX", 10)
|
||||
.await
|
||||
.expect("listing survives one unreadable row");
|
||||
let rows: Vec<MessageRead> = listed.into_iter().map(|(read, _)| read).collect();
|
||||
let (good, bad) = split_bulk_read(&rows);
|
||||
assert_eq!(good.len(), 2, "both intact messages are still listed");
|
||||
assert_eq!(bad, vec!["msg-rotten"]);
|
||||
|
||||
// The keyset page must keep the unreadable row in its slot: a page one row short
|
||||
// of its own window would make the cursor name a position the store never
|
||||
// returned, and rows past it unreachable.
|
||||
let (page, next) = engine
|
||||
.list_messages_page("acct-1", "INBOX", None, 2)
|
||||
.await
|
||||
.expect("paged listing survives one unreadable row");
|
||||
assert_eq!(page.len(), 2, "the unreadable row still occupies a slot");
|
||||
assert!(next.is_some(), "a third row remains, so a cursor is owed");
|
||||
let page_rows: Vec<MessageRead> = page.into_iter().map(|(read, _)| read).collect();
|
||||
let (_, page_bad) = split_bulk_read(&page_rows);
|
||||
assert_eq!(page_bad, vec!["msg-rotten"]);
|
||||
|
||||
let chunk = engine
|
||||
.get_message_chunk("", 10)
|
||||
.await
|
||||
.expect("chunked scan survives one unreadable row");
|
||||
assert_eq!(chunk.len(), 3, "the scan still walks every row");
|
||||
let (good, bad) = split_bulk_read(&chunk);
|
||||
assert_eq!(good.len(), 2);
|
||||
assert_eq!(bad, vec!["msg-rotten"]);
|
||||
|
||||
let exported = engine
|
||||
.list_messages_for_export(None, None)
|
||||
.await
|
||||
.expect_err("export must fail closed on corrupted ciphertext");
|
||||
.expect("export survives one unreadable row");
|
||||
let (good, bad) = split_bulk_read(&exported);
|
||||
assert_eq!(
|
||||
good.len(),
|
||||
2,
|
||||
"an export must still contain every message it can authenticate"
|
||||
);
|
||||
assert_eq!(
|
||||
bad,
|
||||
vec!["msg-rotten"],
|
||||
"the message missing from the archive must be named, not silently dropped"
|
||||
);
|
||||
|
||||
// The whole point of the distinction: asking for that one message by name still
|
||||
// fails closed, because there is no other row left for the caller to receive.
|
||||
let err = engine
|
||||
.get_message("msg-rotten")
|
||||
.await
|
||||
.expect_err("the single-item read must still fail closed");
|
||||
assert!(matches!(err, DatabaseError::Decryption(_)));
|
||||
|
||||
// And the good messages are still individually readable -- the corruption is
|
||||
// isolated to its own row, not smeared across the store.
|
||||
assert_eq!(
|
||||
engine
|
||||
.get_message("msg-good-1")
|
||||
.await
|
||||
.expect("an intact message is still readable")
|
||||
.body_plain
|
||||
.as_deref(),
|
||||
Some("body number 0")
|
||||
);
|
||||
}
|
||||
|
||||
/// `migrate()` runs on every open and backfills the search index from stored
|
||||
/// ciphertext, so a single bit-rotted body must not stop the daemon booting. The
|
||||
/// affected row is left out of the index (never indexed with an empty body, which
|
||||
/// would mark it permanently indexed and permanently unsearchable), while every other
|
||||
/// message indexes and searches normally.
|
||||
#[tokio::test]
|
||||
async fn fts_backfill_skips_one_corrupt_body_instead_of_failing_the_open() {
|
||||
let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
|
||||
save_bulk_fixture(&engine, &["msg-index-good", "msg-index-rotten"]).await;
|
||||
|
||||
// Drop both FTS entries so the backfill has real work, then corrupt one body.
|
||||
sqlx::query("DELETE FROM messages_fts")
|
||||
.execute(&engine.pool)
|
||||
.await
|
||||
.expect("clear the index");
|
||||
corrupt_stored_body(&engine, "msg-index-rotten").await;
|
||||
|
||||
engine
|
||||
.migrate()
|
||||
.await
|
||||
.expect("one unreadable body must not stop the store opening");
|
||||
|
||||
let search = crate::search::SearchEngine::new(&engine);
|
||||
let hits = search
|
||||
.search_messages("number 0")
|
||||
.await
|
||||
.expect("search succeeds");
|
||||
assert_eq!(hits.len(), 1, "the intact message is indexed and findable");
|
||||
assert_eq!(hits[0].id, "msg-index-good");
|
||||
|
||||
// The corrupt row is absent from the index rather than present with an empty
|
||||
// body: it must stay a candidate for backfill so a repaired row heals itself.
|
||||
let (indexed,): (i64,) =
|
||||
sqlx::query_as("SELECT COUNT(*) FROM messages_fts WHERE message_key = ?")
|
||||
.bind("msg-index-rotten")
|
||||
.fetch_one(&engine.pool)
|
||||
.await
|
||||
.expect("count index rows");
|
||||
assert_eq!(indexed, 0);
|
||||
}
|
||||
|
||||
fn sample_calendar_event(id: &str, account_id: &str) -> nuncio_core::model::CalendarEvent {
|
||||
@@ -4758,14 +5066,14 @@ mod tests {
|
||||
|
||||
let chunk1 = engine.get_message_chunk("", 3).await.unwrap();
|
||||
assert_eq!(chunk1.len(), 3);
|
||||
assert_eq!(chunk1[0].id, "msg-001");
|
||||
assert_eq!(chunk1[2].id, "msg-003");
|
||||
assert_eq!(readable(&chunk1[0]).id, "msg-001");
|
||||
assert_eq!(readable(&chunk1[2]).id, "msg-003");
|
||||
|
||||
let last_id = &chunk1.last().unwrap().id;
|
||||
let last_id = &readable(chunk1.last().unwrap()).id;
|
||||
let chunk2 = engine.get_message_chunk(last_id, 3).await.unwrap();
|
||||
assert_eq!(chunk2.len(), 2);
|
||||
assert_eq!(chunk2[0].id, "msg-004");
|
||||
assert_eq!(chunk2[1].id, "msg-005");
|
||||
assert_eq!(readable(&chunk2[0]).id, "msg-004");
|
||||
assert_eq!(readable(&chunk2[1]).id, "msg-005");
|
||||
|
||||
let mutation = nuncio_filter::OutboxManager::create_mutation(
|
||||
"rule-1",
|
||||
@@ -5920,7 +6228,7 @@ mod tests {
|
||||
assert!(rows.len() <= 2, "page must not exceed page_size");
|
||||
seen.extend(
|
||||
rows.iter()
|
||||
.map(|(e, p)| (e.id.clone(), p.remote_id.clone())),
|
||||
.map(|(e, p)| (readable(e).id.clone(), p.remote_id.clone())),
|
||||
);
|
||||
match next {
|
||||
Some(cursor) => after = Some(cursor),
|
||||
@@ -6055,10 +6363,10 @@ mod tests {
|
||||
.expect("list");
|
||||
let found = listed
|
||||
.iter()
|
||||
.find(|(m, _)| m.id == pair.0.id)
|
||||
.find(|(m, _)| readable(m).id == pair.0.id)
|
||||
.expect("the saved message is listed");
|
||||
assert_eq!(found.0.message_id, pair.0.message_id);
|
||||
assert_eq!(found.0.content_hash, pair.0.content_hash);
|
||||
assert_eq!(readable(&found.0).message_id, pair.0.message_id);
|
||||
assert_eq!(readable(&found.0).content_hash, pair.0.content_hash);
|
||||
|
||||
let exported = engine
|
||||
.list_messages_for_export(None, None)
|
||||
@@ -6066,10 +6374,10 @@ mod tests {
|
||||
.expect("export");
|
||||
let found = exported
|
||||
.iter()
|
||||
.find(|m| m.id == pair.0.id)
|
||||
.find(|m| readable(m).id == pair.0.id)
|
||||
.expect("the saved message is exportable");
|
||||
assert_eq!(found.message_id, pair.0.message_id);
|
||||
assert_eq!(found.content_hash, pair.0.content_hash);
|
||||
assert_eq!(readable(found).message_id, pair.0.message_id);
|
||||
assert_eq!(readable(found).content_hash, pair.0.content_hash);
|
||||
|
||||
// A message with no Message-ID stays None rather than becoming "".
|
||||
let plain = synced_email("acct-1", "INBOX", "42", "8");
|
||||
@@ -6134,8 +6442,8 @@ mod tests {
|
||||
|
||||
let all = engine.list_messages_for_export(None, None).await.unwrap();
|
||||
assert_eq!(all.len(), 2);
|
||||
assert_eq!(all[0].id, "msg-1");
|
||||
assert_eq!(all[1].id, "msg-2");
|
||||
assert_eq!(readable(&all[0]).id, "msg-1");
|
||||
assert_eq!(readable(&all[1]).id, "msg-2");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -6149,7 +6457,7 @@ mod tests {
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(scoped.len(), 1);
|
||||
assert_eq!(scoped[0].id, "msg-1");
|
||||
assert_eq!(readable(&scoped[0]).id, "msg-1");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -6163,7 +6471,7 @@ mod tests {
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(scoped.len(), 1);
|
||||
assert_eq!(scoped[0].id, "msg-2");
|
||||
assert_eq!(readable(&scoped[0]).id, "msg-2");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -6178,7 +6486,7 @@ mod tests {
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(scoped.len(), 1);
|
||||
assert_eq!(scoped[0].id, "msg-1");
|
||||
assert_eq!(readable(&scoped[0]).id, "msg-1");
|
||||
}
|
||||
|
||||
/// An export whose WORM audit record cannot be persisted must never report success --
|
||||
@@ -6189,7 +6497,13 @@ mod tests {
|
||||
async fn export_messages_to_file_fails_when_worm_audit_write_fails() {
|
||||
let (engine, dir) = DatabaseEngine::connect_ephemeral().await.unwrap();
|
||||
save_export_message(&engine, "msg-1", "acct-a", "inbox").await;
|
||||
let messages = engine.list_messages_for_export(None, None).await.unwrap();
|
||||
let messages: Vec<_> = engine
|
||||
.list_messages_for_export(None, None)
|
||||
.await
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.map(|read| read.expect("fixture bodies are readable"))
|
||||
.collect();
|
||||
|
||||
engine.close().await;
|
||||
|
||||
|
||||
+159
-13
@@ -59,7 +59,7 @@ use nuncio_proto::v1::{
|
||||
UpdateAccountResponse, UpdateAvailable, UpdateRuleRequest, UpdateRuleResponse,
|
||||
ValidateRuleRequest, ValidateRuleResponse, VerifyChainRequest, VerifyChainResponse,
|
||||
};
|
||||
use nuncio_store::db::DatabaseEngine;
|
||||
use nuncio_store::db::{BulkReadRow, DatabaseEngine};
|
||||
use nuncio_store::search::SearchEngine;
|
||||
use nuncio_store::vault::SecretManager;
|
||||
use std::pin::Pin;
|
||||
@@ -977,7 +977,9 @@ fn map_attachment_from_proto(attachment: AttachmentProto) -> nuncio_core::model:
|
||||
/// `folder_id` and `read` come from the placement: both are per-mailbox facts
|
||||
/// now, and the wire message still carries exactly one of each, so it describes
|
||||
/// one occupancy of the message rather than the message as a whole.
|
||||
fn map_email_to_proto(placed: nuncio_store::db::MessageWithPlacement) -> MessageProto {
|
||||
fn map_email_to_proto(
|
||||
placed: (nuncio_core::model::Email, nuncio_core::model::Placement),
|
||||
) -> MessageProto {
|
||||
let (email, placement) = placed;
|
||||
MessageProto {
|
||||
id: email.id,
|
||||
@@ -1778,11 +1780,14 @@ impl MailGrpcService {
|
||||
// Re-impose the store's ordering over the concatenated per-account
|
||||
// pages, so the cursor this returns is a position in the merged
|
||||
// sequence rather than in whichever account happened to be last.
|
||||
// Ordered on the identity columns, which every row carries whether or
|
||||
// not its body decrypted -- so a row the store could not read still
|
||||
// sorts into its rightful place and still occupies a page slot.
|
||||
rows.sort_by(|(a_mail, a_place), (b_mail, b_place)| {
|
||||
b_mail
|
||||
.received_at
|
||||
.cmp(&a_mail.received_at)
|
||||
.then_with(|| b_mail.id.cmp(&a_mail.id))
|
||||
.received_at()
|
||||
.cmp(&a_mail.received_at())
|
||||
.then_with(|| b_mail.message_key().cmp(a_mail.message_key()))
|
||||
.then_with(|| b_place.uid_validity.cmp(&a_place.uid_validity))
|
||||
.then_with(|| b_place.remote_id.cmp(&a_place.remote_id))
|
||||
});
|
||||
@@ -1791,10 +1796,10 @@ impl MailGrpcService {
|
||||
rows.truncate(page_size);
|
||||
let next = has_more
|
||||
.then(|| {
|
||||
rows.last().map(|(email, placement)| {
|
||||
rows.last().map(|(read, placement)| {
|
||||
(
|
||||
email.received_at,
|
||||
email.id.clone(),
|
||||
read.received_at(),
|
||||
read.message_key().to_string(),
|
||||
placement.uid_validity.clone(),
|
||||
placement.remote_id.clone(),
|
||||
)
|
||||
@@ -2061,13 +2066,36 @@ impl Mail for MailGrpcService {
|
||||
])
|
||||
})
|
||||
.unwrap_or_default();
|
||||
let mut messages: Vec<MessageProto> =
|
||||
messages.into_iter().map(map_email_to_proto).collect();
|
||||
// A row whose stored body would not authenticate is reported by ID
|
||||
// instead of being listed. It is neither dropped in silence (the caller
|
||||
// would see a page one message short and have no way to know) nor
|
||||
// substituted with an empty body (indistinguishable from a message that
|
||||
// genuinely has none). The rest of the page is unaffected: one
|
||||
// bit-rotted row must not take a folder listing down with it.
|
||||
let mut unreadable_message_ids = Vec::new();
|
||||
let mut placed = Vec::with_capacity(messages.len());
|
||||
for (read, placement) in messages {
|
||||
match read {
|
||||
Ok(email) => placed.push((email, placement)),
|
||||
Err(unreadable) => unreadable_message_ids.push(unreadable.message_key),
|
||||
}
|
||||
}
|
||||
if !unreadable_message_ids.is_empty() {
|
||||
tracing::warn!(
|
||||
folder_id = %req.folder_id,
|
||||
unreadable = unreadable_message_ids.len(),
|
||||
"Mail: omitted messages from a listing page whose stored bodies failed \
|
||||
authentication"
|
||||
);
|
||||
}
|
||||
|
||||
let mut messages: Vec<MessageProto> = placed.into_iter().map(map_email_to_proto).collect();
|
||||
fill_placements(&self.db, &mut messages).await;
|
||||
|
||||
Ok(Response::new(ListMessagesResponse {
|
||||
messages,
|
||||
next_page_token,
|
||||
unreadable_message_ids,
|
||||
}))
|
||||
}
|
||||
|
||||
@@ -2971,6 +2999,7 @@ impl Filters for FiltersGrpcService {
|
||||
let mut scanned: u64 = 0;
|
||||
let mut matched: u64 = 0;
|
||||
let mut applied: u64 = 0;
|
||||
let mut unreadable: u64 = 0;
|
||||
|
||||
loop {
|
||||
let batch = match db.get_message_chunk(&last_id, chunk_size).await {
|
||||
@@ -2986,7 +3015,26 @@ impl Filters for FiltersGrpcService {
|
||||
};
|
||||
let is_last_page = batch.len() < chunk_size;
|
||||
|
||||
for email in &batch {
|
||||
for read in &batch {
|
||||
// A message whose stored body will not authenticate is
|
||||
// counted and skipped, never evaluated: rules read the
|
||||
// body, so acting on one the store could not
|
||||
// authenticate would let corrupted bytes drive real
|
||||
// side effects. Skipping it does not stop the scan --
|
||||
// one bad row must not cost the whole store its rescan.
|
||||
let email = match read {
|
||||
Ok(email) => email,
|
||||
Err(un) => {
|
||||
unreadable += 1;
|
||||
tracing::warn!(
|
||||
message_key = %un.message_key,
|
||||
"Filters: skipping a message during triage whose stored \
|
||||
body failed authentication"
|
||||
);
|
||||
scanned += 1;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
// Messages belonging to an account this daemon does not
|
||||
// own are counted as scanned but never acted on, so the
|
||||
// progress total still reflects the whole store. A
|
||||
@@ -3025,8 +3073,11 @@ impl Filters for FiltersGrpcService {
|
||||
}
|
||||
}
|
||||
|
||||
// Advanced from the last row the query returned, readable or
|
||||
// not: resuming from the last *readable* row would re-scan
|
||||
// every unreadable row after it, forever.
|
||||
if let Some(last) = batch.last() {
|
||||
last_id = last.id.clone();
|
||||
last_id = last.message_key().to_string();
|
||||
}
|
||||
|
||||
let done = is_last_page;
|
||||
@@ -3037,6 +3088,7 @@ impl Filters for FiltersGrpcService {
|
||||
actions_applied_count: applied,
|
||||
last_message_id: last_id.clone(),
|
||||
done,
|
||||
unreadable_count: unreadable,
|
||||
}))
|
||||
.await
|
||||
.is_err()
|
||||
@@ -3051,6 +3103,7 @@ impl Filters for FiltersGrpcService {
|
||||
scanned,
|
||||
matched,
|
||||
applied,
|
||||
unreadable,
|
||||
"Filters: triage rescan finished"
|
||||
);
|
||||
return;
|
||||
@@ -3115,7 +3168,7 @@ impl Export for ExportGrpcService {
|
||||
};
|
||||
tracing::info!(format = ?format, scope = scope_kind, "Export: mailbox export started");
|
||||
|
||||
let messages = match req.scope {
|
||||
let read_rows = match req.scope {
|
||||
Some(export_request::Scope::AccountId(account_id)) => {
|
||||
self.db
|
||||
.list_messages_for_export(Some(&account_id), None)
|
||||
@@ -3130,6 +3183,29 @@ impl Export for ExportGrpcService {
|
||||
}
|
||||
.map_err(|e| Status::internal(format!("failed to load messages for export: {e}")))?;
|
||||
|
||||
// A message whose stored body will not authenticate is left out of the
|
||||
// archive and named in the response. The alternative -- writing it with
|
||||
// an empty or "unavailable" body -- would put content into a file that
|
||||
// outlives this store and that no later reader could distinguish from
|
||||
// mail the user actually received. Reporting the gap keeps the archive
|
||||
// honest about what it does contain, and one such row does not cost the
|
||||
// user the export of everything else.
|
||||
let mut unreadable_message_ids = Vec::new();
|
||||
let mut messages = Vec::with_capacity(read_rows.len());
|
||||
for read in read_rows {
|
||||
match read {
|
||||
Ok(email) => messages.push(email),
|
||||
Err(unreadable) => unreadable_message_ids.push(unreadable.message_key),
|
||||
}
|
||||
}
|
||||
if !unreadable_message_ids.is_empty() {
|
||||
tracing::warn!(
|
||||
unreadable = unreadable_message_ids.len(),
|
||||
"Export: omitted messages whose stored bodies failed authentication; \
|
||||
the archive is incomplete and the omitted ids are reported to the caller"
|
||||
);
|
||||
}
|
||||
|
||||
let output_path = std::path::PathBuf::from(&req.output_path);
|
||||
let summary = self
|
||||
.db
|
||||
@@ -3147,6 +3223,7 @@ impl Export for ExportGrpcService {
|
||||
output_path: summary.output_path,
|
||||
message_count: summary.message_count as u64,
|
||||
bytes_written: summary.bytes_written,
|
||||
unreadable_message_ids,
|
||||
}))
|
||||
}
|
||||
}
|
||||
@@ -5370,6 +5447,75 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// One corrupted stored body must not take a folder listing down with it,
|
||||
/// and must not disappear from it either.
|
||||
///
|
||||
/// This is the wire-level half of the store's
|
||||
/// `one_corrupt_body_does_not_destroy_a_bulk_read`: the page still carries
|
||||
/// every message whose body authenticated, the one that did not is named in
|
||||
/// `unreadable_message_ids` rather than returned with an invented body, and
|
||||
/// `GetMessage` on that same id still fails closed -- there the caller asked
|
||||
/// for exactly that content, so an answer without it would be a lie.
|
||||
#[tokio::test]
|
||||
async fn list_messages_reports_a_corrupt_body_instead_of_failing_the_page() {
|
||||
let (db, _dir) = DatabaseEngine::connect_ephemeral()
|
||||
.await
|
||||
.expect("connect ephemeral test db");
|
||||
seed_in_account(&db, "acct-a", "m-1", "inbox", 1_700_000_001).await;
|
||||
seed_in_account(&db, "acct-a", "m-rotten", "inbox", 1_700_000_002).await;
|
||||
seed_in_account(&db, "acct-a", "m-3", "inbox", 1_700_000_003).await;
|
||||
|
||||
// Flip a byte of the stored ciphertext, leaving it valid hex, so the
|
||||
// read fails AEAD authentication exactly as bit-rot or tampering would.
|
||||
let (stored,): (String,) =
|
||||
sqlx::query_as("SELECT body_plain FROM messages WHERE message_key = ?")
|
||||
.bind("m-rotten")
|
||||
.fetch_one(db.pool())
|
||||
.await
|
||||
.expect("fetch stored ciphertext");
|
||||
let mut bytes = hex::decode(&stored).expect("stored ciphertext is valid hex");
|
||||
let last = bytes.len() - 1;
|
||||
bytes[last] ^= 0xFF;
|
||||
sqlx::query("UPDATE messages SET body_plain = ? WHERE message_key = ?")
|
||||
.bind(hex::encode(bytes))
|
||||
.bind("m-rotten")
|
||||
.execute(db.pool())
|
||||
.await
|
||||
.expect("corrupt stored ciphertext");
|
||||
|
||||
let mut client = mail_client_over(db).await;
|
||||
|
||||
let resp = client
|
||||
.list_messages(authed_bearer_request(ListMessagesRequest {
|
||||
folder_id: "inbox".to_string(),
|
||||
page_size: 10,
|
||||
page_token: String::new(),
|
||||
}))
|
||||
.await
|
||||
.expect("one unreadable row must not fail the whole listing")
|
||||
.into_inner();
|
||||
|
||||
let ids: Vec<&str> = resp.messages.iter().map(|m| m.id.as_str()).collect();
|
||||
assert_eq!(
|
||||
ids,
|
||||
vec!["m-3", "m-1"],
|
||||
"every message that authenticated must still be listed"
|
||||
);
|
||||
assert_eq!(
|
||||
resp.unreadable_message_ids,
|
||||
vec!["m-rotten".to_string()],
|
||||
"the omitted message must be named, not silently missing"
|
||||
);
|
||||
|
||||
let err = client
|
||||
.get_message(authed_bearer_request(GetMessageRequest {
|
||||
message_id: "m-rotten".to_string(),
|
||||
}))
|
||||
.await
|
||||
.expect_err("the single-item read must still fail closed");
|
||||
assert_eq!(err.code(), tonic::Code::NotFound);
|
||||
}
|
||||
|
||||
/// A configured account holding nothing in this folder must not truncate
|
||||
/// the listing.
|
||||
///
|
||||
|
||||
@@ -747,7 +747,12 @@ mod tests {
|
||||
assert_eq!(persisted.len(), 2);
|
||||
let subjects: Vec<String> = persisted
|
||||
.iter()
|
||||
.map(|(email, _placement)| email.subject.clone())
|
||||
.map(|(read, _placement)| {
|
||||
read.as_ref()
|
||||
.expect("fixture body is readable")
|
||||
.subject
|
||||
.clone()
|
||||
})
|
||||
.collect();
|
||||
assert!(subjects.contains(&"Hello".to_string()));
|
||||
assert!(subjects.contains(&"World".to_string()));
|
||||
@@ -948,7 +953,10 @@ mod tests {
|
||||
.expect("jmap message persisted");
|
||||
assert_eq!(persisted.len(), 1);
|
||||
let (email, placement) = &persisted[0];
|
||||
assert_eq!(email.subject, "Welcome to JMAP Sync");
|
||||
assert_eq!(
|
||||
email.as_ref().expect("fixture body is readable").subject,
|
||||
"Welcome to JMAP Sync"
|
||||
);
|
||||
assert_eq!(placement.remote_id, "jmap-msg-1");
|
||||
|
||||
assert_eq!(
|
||||
|
||||
Reference in new issue
Block a user