fix(store): survive one corrupt body on bulk read paths #429

Merged
KofTwentyTwo merged 1 commits from fix/bulk-path-corrupt-body into dev 2026-08-15 02:14:37 +00:00
6 changed files with 684 additions and 150 deletions

No files matched your search

+43 -5
View File
@@ -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
View File
@@ -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
View File
@@ -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.
///
+10 -2
View File
@@ -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!(