From 54527fb55bbe7b43a014f571a4407bac5bd81497 Mon Sep 17 00:00:00 2001 From: otsmr Date: Fri, 11 Sep 2026 23:02:43 +0200 Subject: [PATCH] fix signal sessions after recovery --- rust/src/api/messages/incoming/messages.rs | 54 ++++++++++++- rust/src/backup/backup_archive.rs | 90 +++++++++++++++++++++- rust/src/database/app/tables/receipt.rs | 8 +- rust/tests/api/media.rs | 6 +- rust/tests/api/session_recovery.rs | 64 +++++++++++++++ 5 files changed, 214 insertions(+), 8 deletions(-) diff --git a/rust/src/api/messages/incoming/messages.rs b/rust/src/api/messages/incoming/messages.rs index cd3662ff..b7259316 100644 --- a/rust/src/api/messages/incoming/messages.rs +++ b/rust/src/api/messages/incoming/messages.rs @@ -782,7 +782,7 @@ pub(crate) async fn queue_decryption_error( .encode_to_vec(); NewReceipt::new(receipt_id, from_user_id, &response) .contact_will_send_receipt(false) - .insert_or_replace(tr) + .insert_if_absent(tr) .await?; Ok(()) } @@ -920,4 +920,56 @@ mod tests { let _held = FlushGuard::claim(&first).expect("nothing else holds the claim"); assert!(FlushGuard::claim(&second).is_some()); } + + #[tokio::test] + async fn a_duplicate_decryption_failure_cannot_downgrade_a_queued_session_reset() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("app.sqlite"); + let database = crate::database::app::AppDatabase::new(path.to_str().unwrap(), None, false) + .await + .unwrap(); + database.run_migrations().await.unwrap(); + sqlx::query("INSERT INTO contacts(user_id, username) VALUES (7, 'alice')") + .execute(&database.pool) + .await + .unwrap(); + + let mut transaction = database.pool.begin().await.unwrap(); + queue_decryption_error( + &mut transaction, + 7, + "failed-receipt", + proto::plaintext_content::decryption_error_message::Type::SessionResetRequired as i32, + ) + .await + .unwrap(); + // A redelivery reaches the reset limiter while the first response is + // still queued and falls back to Unknown. It must leave the stronger + // response untouched. + queue_decryption_error( + &mut transaction, + 7, + "failed-receipt", + proto::plaintext_content::decryption_error_message::Type::Unknown as i32, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let bytes = sqlx::query_scalar::<_, Vec>( + "SELECT message FROM receipts WHERE receipt_id = 'failed-receipt'", + ) + .fetch_one(&database.pool) + .await + .unwrap(); + let message = proto::Message::decode(bytes.as_slice()).unwrap(); + let error = message + .plaintext_content + .and_then(|content| content.decryption_error_message) + .expect("the queued response is a decryption error"); + assert_eq!( + error.r#type, + proto::plaintext_content::decryption_error_message::Type::SessionResetRequired as i32 + ); + } } diff --git a/rust/src/backup/backup_archive.rs b/rust/src/backup/backup_archive.rs index 7c79342f..77daf896 100644 --- a/rust/src/backup/backup_archive.rs +++ b/rust/src/backup/backup_archive.rs @@ -321,6 +321,26 @@ impl BackupArchive { ) .await?; rust_database.run_migrations().await?; + + // A backup is a snapshot of a Double Ratchet at an earlier point in + // time. Once either peer has advanced past that snapshot, restoring + // its session record cannot make it current again and can make both + // sides keep encrypting against incompatible states. Start with no + // peer sessions instead. The regular V2 send path establishes them + // lazily from a current prekey bundle, while an old inbound ciphertext + // is answered by the existing SESSION_RESET_REQUIRED recovery flow. + // + // Reset throttles describe the discarded sessions, so retaining them + // could suppress the first legitimate repair after recovery. + let mut signal_transaction = rust_database.pool.begin().await?; + sqlx::query!("DELETE FROM signal_sessions") + .execute(&mut *signal_transaction) + .await?; + sqlx::query!("DELETE FROM signal_session_resets") + .execute(&mut *signal_transaction) + .await?; + signal_transaction.commit().await?; + ctx.replace_rust_database(rust_database, &key_manager) .await?; @@ -331,10 +351,10 @@ impl BackupArchive { // marks makes the next `on_connected` publish a signed prekey and a // fresh batch of PQC prekeys that this database actually holds. // - // Sessions restored alongside them are left as they are: they are only - // broken for peers who ratcheted past this archive, and the first - // message that fails to decrypt resets that peer's session on its own. - // See `signal::reset`. + // Peer sessions were discarded above. They are rebuilt lazily from + // current bundles, avoiding use of ratchet state rewound by the + // archive. See `signal::reset` for recovery of messages already in + // flight when the backup was restored. if let Err(error) = crate::user_config::UserConfig::update(ctx, |config| { config.signal_last_signed_pre_key_updated = None; config.signal_last_pqc_pre_keys_uploaded = None; @@ -457,6 +477,37 @@ mod tests { .await .unwrap(); } + { + let rust_db = ctx.rust_db.read().await.clone(); + sqlx::query!( + r#" + INSERT INTO signal_identities(name, identity_key, timestamp) + VALUES('restored-peer', x'040506', 1) + "# + ) + .execute(&rust_db.pool) + .await + .unwrap(); + sqlx::query!( + r#" + INSERT INTO signal_sessions(name, device_id, record_bytes) + VALUES('restored-peer', 1, x'010203') + "# + ) + .execute(&rust_db.pool) + .await + .unwrap(); + sqlx::query!( + r#" + INSERT INTO signal_session_resets( + name, device_id, last_reset_at, window_started_at, resets_in_window + ) VALUES('restored-peer', 1, 1, 1, 1) + "# + ) + .execute(&rust_db.pool) + .await + .unwrap(); + } // 2. Create backup let backup_path = BackupArchive::create_backup(&ctx).await.unwrap(); @@ -510,6 +561,37 @@ mod tests { .await .unwrap(); assert_eq!(username, "original contact"); + + let rust_db = ctx.rust_db.read().await.clone(); + let session_count = + sqlx::query_scalar!(r#"SELECT COUNT(*) AS "count!: i64" FROM signal_sessions"#) + .fetch_one(&rust_db.pool) + .await + .unwrap(); + assert_eq!(session_count, 0, "restored peer sessions must be discarded"); + + let reset_count = sqlx::query_scalar!( + r#"SELECT COUNT(*) AS "count!: i64" FROM signal_session_resets"# + ) + .fetch_one(&rust_db.pool) + .await + .unwrap(); + assert_eq!( + reset_count, 0, + "reset throttles for discarded sessions must be cleared" + ); + + let identity_key = sqlx::query_scalar!( + "SELECT identity_key FROM signal_identities WHERE name = 'restored-peer'" + ) + .fetch_one(&rust_db.pool) + .await + .unwrap(); + assert_eq!( + identity_key, + vec![4, 5, 6], + "a contact's trusted identity key must survive archive recovery" + ); } } diff --git a/rust/src/database/app/tables/receipt.rs b/rust/src/database/app/tables/receipt.rs index b40de960..e3f984c4 100644 --- a/rust/src/database/app/tables/receipt.rs +++ b/rust/src/database/app/tables/receipt.rs @@ -248,11 +248,15 @@ impl<'a> NewReceipt<'a> { Ok(()) } - pub async fn insert_or_replace(&self, transaction: &mut Transaction<'_, Sqlite>) -> Result<()> { + /// Queues a response only while no response for this inbound receipt is in + /// flight. A redelivery can race the sender task, but it must not replace + /// the response that task is about to send. + pub async fn insert_if_absent(&self, transaction: &mut Transaction<'_, Sqlite>) -> Result<()> { sqlx::query!( r#" - INSERT OR REPLACE INTO receipts(receipt_id, contact_id, message, contact_will_sends_receipt, wake_receiver) + INSERT INTO receipts(receipt_id, contact_id, message, contact_will_sends_receipt, wake_receiver) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(receipt_id) DO NOTHING "#, self.receipt_id, self.contact_id, diff --git a/rust/tests/api/media.rs b/rust/tests/api/media.rs index 36a2029d..7ccbfbd2 100644 --- a/rust/tests/api/media.rs +++ b/rust/tests/api/media.rs @@ -78,7 +78,11 @@ async fn test_media_lifecycle_actions_and_reupload() -> anyhow::Result<()> { .join("mediafiles/tmp") .join(format!("{local_media_id}.webp")); std::fs::create_dir_all(plaintext_path.parent().expect("tmp directory"))?; - std::fs::write(&plaintext_path, b"plaintext media bytes")?; + let source = image::RgbImage::from_pixel(8, 6, image::Rgb([12, 34, 56])); + let plaintext = webp::Encoder::from_rgb(source.as_raw(), source.width(), source.height()) + .encode_simple(false, 80.0) + .map_err(|error| anyhow::anyhow!("could not encode media fixture: {error:?}"))?; + std::fs::write(&plaintext_path, plaintext.as_ref())?; let media_content = proto::EncryptedContent { group_id: Some(group_id.clone()), diff --git a/rust/tests/api/session_recovery.rs b/rust/tests/api/session_recovery.rs index c7a13a54..2e300f39 100644 --- a/rust/tests/api/session_recovery.rs +++ b/rust/tests/api/session_recovery.rs @@ -71,6 +71,70 @@ async fn test_signal_session_auto_recovery_on_missing_session() -> anyhow::Resul Ok(()) } +/// Archive recovery deliberately discards the receiver's restored sessions. +/// A sender may still have its pre-recovery half and send one stale ciphertext; +/// the receiver must request a reset and receive the re-encrypted message over +/// the fresh prekey session without user intervention. +#[tokio::test] +async fn test_signal_session_recovers_after_receiver_sessions_are_purged() -> anyhow::Result<()> { + init_tracing(); + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + ContactService::new(&tester_a.context) + .request_by_username(tester_b.username.clone(), true) + .await?; + tester_b + .wait_for_contact_state(tester_a.user_id, false, true) + .await?; + ContactService::new(&tester_b.context) + .accept_request(tester_a.user_id, true) + .await?; + tester_a + .wait_for_contact_state(tester_b.user_id, true, false) + .await?; + + let group_id = Group::direct_chat_id(tester_a.user_id, tester_b.user_id); + let initial_id = MessageService::new(&tester_a.context) + .insert_and_send_text(group_id.clone(), "Before session purge".into(), None, None) + .await?; + tester_b + .wait_for_text_message(&initial_id, tester_a.user_id, "Before session purge") + .await?; + + // This is the Signal-state result of restoring an archive: B retains its + // identity and prekeys, but no peer session survives. + let rust_db_b = tester_b.context.rust_db.read().await.clone(); + sqlx::query!("DELETE FROM signal_sessions") + .execute(&rust_db_b.pool) + .await?; + sqlx::query!("DELETE FROM signal_session_resets") + .execute(&rust_db_b.pool) + .await?; + + // A still encrypts with the old half. B reports that it cannot use that + // session, A rebuilds from B's current bundle, and the queued plaintext is + // re-encrypted and delivered over a PreKey message. + let recovered_id = MessageService::new(&tester_a.context) + .insert_and_send_text(group_id.clone(), "After session purge".into(), None, None) + .await?; + tester_b + .wait_for_text_message(&recovered_id, tester_a.user_id, "After session purge") + .await?; + + // The newly established session works in the restored user's direction as + // well, proving that the repair converged rather than only delivering the + // retried message. + let reply_id = MessageService::new(&tester_b.context) + .insert_and_send_text(group_id, "Reply after session purge".into(), None, None) + .await?; + tester_a + .wait_for_text_message(&reply_id, tester_b.user_id, "Reply after session purge") + .await?; + + Ok(()) +} + /// A restored backup rewinds one side's ratchet behind the peer's, so nothing /// the peer sends afterwards decrypts and resending it never helps. The /// receiver has to retire its session, the sender has to build a new one from