From 7d850de90dc933af32db962f0548b51b8dfa7288 Mon Sep 17 00:00:00 2001 From: otsmr Date: Sun, 6 Sep 2026 04:07:06 +0200 Subject: [PATCH] fix group issue --- .../entries/chat_webxdc.entry.dart | 1 - ...d570b008e98adf4a94a763e298b7606f8da11.json | 12 + ...20a93f258716b668c500d3535927064cfc863.json | 38 ++++ ...cbd57e4bc60e7a0abb519f1bd80348f13c37c.json | 21 ++ ...dc52fbe79d66f075bc47b0fd1c21a70fe04a5.json | 12 + ...818aa1d104eff8fcacadfa5bf6c427aa8e0a8.json | 12 + ...ebdded92ee03640c347b6017d2229b6a686e6.json | 12 + ...0a069626a920e14d669ce097c0c4720ec01a3.json | 26 +++ ...638a25806d9be376027c4ac86955a42135ef2.json | 12 - ...f566a1060e3ece94dd0de35d6fdea8648f93e.json | 21 ++ ...046ee9530cd01d899c05b605be9222dcdef98.json | 12 + ...a737a2562b8d174d04998ac729157df8230d8.json | 12 + rust/src/api/messages/incoming.rs | 180 ++++++++++++++- rust/src/api/messages/incoming/messages.rs | 114 +++++++--- rust/src/api/proto/client/transport.proto | 6 + rust/src/backup/backup_archive.rs | 18 ++ .../0017_failed_decryption_claims.sql | 10 + .../0018_drop_failed_decryption_claims.sql | 11 + rust/src/database/app/tables/groups.rs | 27 +++ .../signal/migrations/0004_session_resets.sql | 14 ++ rust/src/error.rs | 49 +++- rust/src/services/contacts.rs | 6 + rust/src/services/groups/mod.rs | 55 +++++ rust/src/services/messages.rs | 35 ++- rust/src/signal/engine.rs | 215 +++++++++++++++++- rust/src/signal/mod.rs | 1 + rust/src/signal/reset.rs | 192 ++++++++++++++++ rust/tests/api/session_recovery.rs | 187 ++++++++++++++- 28 files changed, 1256 insertions(+), 55 deletions(-) create mode 100644 rust/.sqlx/query-1c17357b35a0a3063df88521641d570b008e98adf4a94a763e298b7606f8da11.json create mode 100644 rust/.sqlx/query-3b2c67b996d35fade4d83005cbd20a93f258716b668c500d3535927064cfc863.json create mode 100644 rust/.sqlx/query-413d276a374ee54f098a809ed55cbd57e4bc60e7a0abb519f1bd80348f13c37c.json create mode 100644 rust/.sqlx/query-44772267c7aef1f2c6c453dc25adc52fbe79d66f075bc47b0fd1c21a70fe04a5.json create mode 100644 rust/.sqlx/query-58f456ebf839809283af9ad02b7818aa1d104eff8fcacadfa5bf6c427aa8e0a8.json create mode 100644 rust/.sqlx/query-7cdd5e1cb18606d1fb36fafc939ebdded92ee03640c347b6017d2229b6a686e6.json create mode 100644 rust/.sqlx/query-ca6f26d1b5a9ffd0072c890ffed0a069626a920e14d669ce097c0c4720ec01a3.json delete mode 100644 rust/.sqlx/query-d2688abdeef4590a1c762a1683a638a25806d9be376027c4ac86955a42135ef2.json create mode 100644 rust/.sqlx/query-d6f2b713b634721d0c5992f24eef566a1060e3ece94dd0de35d6fdea8648f93e.json create mode 100644 rust/.sqlx/query-e765e45ef6696e2b39570a04b53046ee9530cd01d899c05b605be9222dcdef98.json create mode 100644 rust/.sqlx/query-e79f0f07d9c5ab9080f6c14afeea737a2562b8d174d04998ac729157df8230d8.json create mode 100644 rust/src/database/app/migrations/0017_failed_decryption_claims.sql create mode 100644 rust/src/database/app/migrations/0018_drop_failed_decryption_claims.sql create mode 100644 rust/src/database/signal/migrations/0004_session_resets.sql create mode 100644 rust/src/signal/reset.rs diff --git a/lib/src/visual/views/chats/chat_messages_components/entries/chat_webxdc.entry.dart b/lib/src/visual/views/chats/chat_messages_components/entries/chat_webxdc.entry.dart index 8e812ab8..4d350bdb 100644 --- a/lib/src/visual/views/chats/chat_messages_components/entries/chat_webxdc.entry.dart +++ b/lib/src/visual/views/chats/chat_messages_components/entries/chat_webxdc.entry.dart @@ -7,7 +7,6 @@ import 'package:twonly/src/services/webxdc/webxdc.service.dart'; import 'package:twonly/src/services/webxdc/webxdc_host.dart'; import 'package:twonly/src/utils/misc.dart'; import 'package:twonly/src/visual/views/chats/chat_messages_components/entries/common.dart'; -import 'package:twonly/src/visual/views/chats/chat_messages_components/entries/friendly_message_time.comp.dart'; import 'package:twonly/src/visual/views/chats/chat_messages_components/entries/webxdc_app_icon.comp.dart'; import 'package:twonly/src/visual/views/webxdc/webxdc_app.view.dart'; diff --git a/rust/.sqlx/query-1c17357b35a0a3063df88521641d570b008e98adf4a94a763e298b7606f8da11.json b/rust/.sqlx/query-1c17357b35a0a3063df88521641d570b008e98adf4a94a763e298b7606f8da11.json new file mode 100644 index 00000000..f24c5334 --- /dev/null +++ b/rust/.sqlx/query-1c17357b35a0a3063df88521641d570b008e98adf4a94a763e298b7606f8da11.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n INSERT INTO signal_session_resets(\n name, device_id, last_reset_at, window_started_at, resets_in_window\n )\n VALUES (?, ?, ?, ?, 1)\n ON CONFLICT(name, device_id) DO UPDATE SET\n last_reset_at = excluded.last_reset_at,\n -- A window that has run out starts over with this reset as its\n -- first, so a peer that broke months ago is not still blocked.\n window_started_at = CASE\n WHEN signal_session_resets.window_started_at <= ?\n THEN excluded.last_reset_at\n ELSE signal_session_resets.window_started_at\n END,\n resets_in_window = CASE\n WHEN signal_session_resets.window_started_at <= ? THEN 1\n ELSE signal_session_resets.resets_in_window + 1\n END\n -- Every SET expression above reads the row as it was before the\n -- update, and this predicate decides whether the update happens at\n -- all: zero affected rows means the claim was refused. A first\n -- reset for a peer takes the INSERT path and is always allowed.\n WHERE signal_session_resets.last_reset_at <= ?\n AND (signal_session_resets.window_started_at <= ?\n OR signal_session_resets.resets_in_window < ?)\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 9 + }, + "nullable": [] + }, + "hash": "1c17357b35a0a3063df88521641d570b008e98adf4a94a763e298b7606f8da11" +} diff --git a/rust/.sqlx/query-3b2c67b996d35fade4d83005cbd20a93f258716b668c500d3535927064cfc863.json b/rust/.sqlx/query-3b2c67b996d35fade4d83005cbd20a93f258716b668c500d3535927064cfc863.json new file mode 100644 index 00000000..8e64428e --- /dev/null +++ b/rust/.sqlx/query-3b2c67b996d35fade4d83005cbd20a93f258716b668c500d3535927064cfc863.json @@ -0,0 +1,38 @@ +{ + "db_name": "SQLite", + "query": "SELECT contact_id, member_state FROM group_members WHERE group_id = ?", + "describe": { + "columns": [ + { + "name": "contact_id", + "ordinal": 0, + "type_info": "Integer", + "origin": { + "Table": { + "table": "group_members", + "name": "contact_id" + } + } + }, + { + "name": "member_state", + "ordinal": 1, + "type_info": "Text", + "origin": { + "Table": { + "table": "group_members", + "name": "member_state" + } + } + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false, + true + ] + }, + "hash": "3b2c67b996d35fade4d83005cbd20a93f258716b668c500d3535927064cfc863" +} diff --git a/rust/.sqlx/query-413d276a374ee54f098a809ed55cbd57e4bc60e7a0abb519f1bd80348f13c37c.json b/rust/.sqlx/query-413d276a374ee54f098a809ed55cbd57e4bc60e7a0abb519f1bd80348f13c37c.json new file mode 100644 index 00000000..5637071e --- /dev/null +++ b/rust/.sqlx/query-413d276a374ee54f098a809ed55cbd57e4bc60e7a0abb519f1bd80348f13c37c.json @@ -0,0 +1,21 @@ +{ + "db_name": "SQLite", + "query": "SELECT COUNT(*) FROM messages WHERE message_id = ?", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer", + "origin": "Expression" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "413d276a374ee54f098a809ed55cbd57e4bc60e7a0abb519f1bd80348f13c37c" +} diff --git a/rust/.sqlx/query-44772267c7aef1f2c6c453dc25adc52fbe79d66f075bc47b0fd1c21a70fe04a5.json b/rust/.sqlx/query-44772267c7aef1f2c6c453dc25adc52fbe79d66f075bc47b0fd1c21a70fe04a5.json new file mode 100644 index 00000000..a16fe36d --- /dev/null +++ b/rust/.sqlx/query-44772267c7aef1f2c6c453dc25adc52fbe79d66f075bc47b0fd1c21a70fe04a5.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n UPDATE signal_session_resets\n SET last_reset_at = last_reset_at - ?, window_started_at = window_started_at - ?\n WHERE name = ?\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 3 + }, + "nullable": [] + }, + "hash": "44772267c7aef1f2c6c453dc25adc52fbe79d66f075bc47b0fd1c21a70fe04a5" +} diff --git a/rust/.sqlx/query-58f456ebf839809283af9ad02b7818aa1d104eff8fcacadfa5bf6c427aa8e0a8.json b/rust/.sqlx/query-58f456ebf839809283af9ad02b7818aa1d104eff8fcacadfa5bf6c427aa8e0a8.json new file mode 100644 index 00000000..992721ed --- /dev/null +++ b/rust/.sqlx/query-58f456ebf839809283af9ad02b7818aa1d104eff8fcacadfa5bf6c427aa8e0a8.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM signal_sessions WHERE name = ? AND device_id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 2 + }, + "nullable": [] + }, + "hash": "58f456ebf839809283af9ad02b7818aa1d104eff8fcacadfa5bf6c427aa8e0a8" +} diff --git a/rust/.sqlx/query-7cdd5e1cb18606d1fb36fafc939ebdded92ee03640c347b6017d2229b6a686e6.json b/rust/.sqlx/query-7cdd5e1cb18606d1fb36fafc939ebdded92ee03640c347b6017d2229b6a686e6.json new file mode 100644 index 00000000..057ef9c8 --- /dev/null +++ b/rust/.sqlx/query-7cdd5e1cb18606d1fb36fafc939ebdded92ee03640c347b6017d2229b6a686e6.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM signal_session_resets WHERE name = ? AND device_id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 2 + }, + "nullable": [] + }, + "hash": "7cdd5e1cb18606d1fb36fafc939ebdded92ee03640c347b6017d2229b6a686e6" +} diff --git a/rust/.sqlx/query-ca6f26d1b5a9ffd0072c890ffed0a069626a920e14d669ce097c0c4720ec01a3.json b/rust/.sqlx/query-ca6f26d1b5a9ffd0072c890ffed0a069626a920e14d669ce097c0c4720ec01a3.json new file mode 100644 index 00000000..f31d87aa --- /dev/null +++ b/rust/.sqlx/query-ca6f26d1b5a9ffd0072c890ffed0a069626a920e14d669ce097c0c4720ec01a3.json @@ -0,0 +1,26 @@ +{ + "db_name": "SQLite", + "query": "SELECT record_bytes FROM signal_sessions WHERE name = ? AND device_id = 1", + "describe": { + "columns": [ + { + "name": "record_bytes", + "ordinal": 0, + "type_info": "Blob", + "origin": { + "Table": { + "table": "signal_sessions", + "name": "record_bytes" + } + } + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "ca6f26d1b5a9ffd0072c890ffed0a069626a920e14d669ce097c0c4720ec01a3" +} diff --git a/rust/.sqlx/query-d2688abdeef4590a1c762a1683a638a25806d9be376027c4ac86955a42135ef2.json b/rust/.sqlx/query-d2688abdeef4590a1c762a1683a638a25806d9be376027c4ac86955a42135ef2.json deleted file mode 100644 index 0a1da0c1..00000000 --- a/rust/.sqlx/query-d2688abdeef4590a1c762a1683a638a25806d9be376027c4ac86955a42135ef2.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "\n UPDATE receipts\n SET receipt_id = ?,\n mark_for_retry = CAST(strftime('%s', 'now') AS INTEGER),\n retry_count = retry_count + 1,\n ack_by_server_at = NULL\n WHERE receipt_id = ? AND contact_id = ?\n ", - "describe": { - "columns": [], - "parameters": { - "Right": 3 - }, - "nullable": [] - }, - "hash": "d2688abdeef4590a1c762a1683a638a25806d9be376027c4ac86955a42135ef2" -} diff --git a/rust/.sqlx/query-d6f2b713b634721d0c5992f24eef566a1060e3ece94dd0de35d6fdea8648f93e.json b/rust/.sqlx/query-d6f2b713b634721d0c5992f24eef566a1060e3ece94dd0de35d6fdea8648f93e.json new file mode 100644 index 00000000..683490f4 --- /dev/null +++ b/rust/.sqlx/query-d6f2b713b634721d0c5992f24eef566a1060e3ece94dd0de35d6fdea8648f93e.json @@ -0,0 +1,21 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT EXISTS(\n SELECT 1\n FROM group_members\n WHERE group_id = ? AND contact_id = ? AND member_state = 'leftGroup'\n )\n ", + "describe": { + "columns": [ + { + "name": "EXISTS(\n SELECT 1\n FROM group_members\n WHERE group_id = ? AND contact_id = ? AND member_state = 'leftGroup'\n )", + "ordinal": 0, + "type_info": "Integer", + "origin": "Expression" + } + ], + "parameters": { + "Right": 2 + }, + "nullable": [ + false + ] + }, + "hash": "d6f2b713b634721d0c5992f24eef566a1060e3ece94dd0de35d6fdea8648f93e" +} diff --git a/rust/.sqlx/query-e765e45ef6696e2b39570a04b53046ee9530cd01d899c05b605be9222dcdef98.json b/rust/.sqlx/query-e765e45ef6696e2b39570a04b53046ee9530cd01d899c05b605be9222dcdef98.json new file mode 100644 index 00000000..4a48fbc4 --- /dev/null +++ b/rust/.sqlx/query-e765e45ef6696e2b39570a04b53046ee9530cd01d899c05b605be9222dcdef98.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n UPDATE receipts\n SET receipt_id = ?,\n mark_for_retry = CAST(strftime('%s', 'now') AS INTEGER),\n retry_count = retry_count + 1,\n ack_by_server_at = NULL,\n -- A rebuild needs a server round-trip, so it cannot run inside this\n -- transaction. `release_deferred_receipts` frees the receipt once\n -- the new session stands.\n deferred_until_session = CASE\n WHEN ? THEN CAST(strftime('%s', 'now') AS INTEGER)\n ELSE deferred_until_session\n END\n WHERE receipt_id = ? AND contact_id = ?\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 4 + }, + "nullable": [] + }, + "hash": "e765e45ef6696e2b39570a04b53046ee9530cd01d899c05b605be9222dcdef98" +} diff --git a/rust/.sqlx/query-e79f0f07d9c5ab9080f6c14afeea737a2562b8d174d04998ac729157df8230d8.json b/rust/.sqlx/query-e79f0f07d9c5ab9080f6c14afeea737a2562b8d174d04998ac729157df8230d8.json new file mode 100644 index 00000000..c0884c48 --- /dev/null +++ b/rust/.sqlx/query-e79f0f07d9c5ab9080f6c14afeea737a2562b8d174d04998ac729157df8230d8.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE signal_sessions SET record_bytes = ? WHERE name = ? AND device_id = 1", + "describe": { + "columns": [], + "parameters": { + "Right": 2 + }, + "nullable": [] + }, + "hash": "e79f0f07d9c5ab9080f6c14afeea737a2562b8d174d04998ac729157df8230d8" +} diff --git a/rust/src/api/messages/incoming.rs b/rust/src/api/messages/incoming.rs index c33a1734..6a7b0b0e 100644 --- a/rust/src/api/messages/incoming.rs +++ b/rust/src/api/messages/incoming.rs @@ -20,7 +20,7 @@ use crate::api::messages::content_type_kind; use crate::api::messages::incoming::messages::{ ensure_contact_exists, handle_plaintext_content, handle_sender_delivery_receipt, process_encrypted_or_queue_error, queue_decryption_error, queue_sender_delivery_receipt, - retransmit_queued_receipts, spawn_receipt_delivery, + release_deferred_receipts, retransmit_queued_receipts, spawn_receipt_delivery, }; use crate::api::proto::client as proto; use crate::api::proto::server_to_client::NewMessage; @@ -29,9 +29,12 @@ use crate::context::Context; use crate::database::app::tables::{Contact, Group, Receipt}; use crate::error::{Result, TwonlyError}; use crate::services::contacts::ContactService; +use crate::services::groups::GroupService; +use crate::signal::reset::SessionResetLimiter; use client_to_server::response::{ok, Response}; use prost::Message as _; use proto::message::Type; +use proto::plaintext_content::decryption_error_message::Type as DecryptionErrorType; use server_to_client::v0::Kind; use sqlx::{Sqlite, Transaction}; use std::collections::HashMap; @@ -239,6 +242,60 @@ pub(crate) async fn handle_request_new_pqc_prekeys( )) } +/// Retires our session with a peer when it can no longer open what they send, +/// and picks the error type the peer is answered with. +/// +/// A session whose ratchet has diverged -- the usual cause is one side +/// restoring a backup, which rewinds its record behind the peer's -- never +/// recovers by decrypting the same ciphertext again. Retiring it locally is +/// only half the repair: the peer holds the matching half and has to open a +/// new session, which `SESSION_RESET_REQUIRED` asks them to do. Anything else +/// (a duplicate, an identity change, a storage failure) keeps the plain +/// `UNKNOWN` error, whose retry the peer already knows how to handle. +/// +/// A refused rate-limit claim also falls back to `UNKNOWN`: the session stays +/// as it is and the message fails, rather than two clients trading resets. +async fn reset_unusable_session( + ctx: &Arc, + from_user_id: i64, + error: &TwonlyError, +) -> DecryptionErrorType { + if !matches!(error, TwonlyError::SignalSessionUnusable(_)) { + return DecryptionErrorType::Unknown; + } + + let name = from_user_id.to_string(); + let rust_database = ctx.rust_db.read().await.clone(); + + match SessionResetLimiter::claim(&rust_database.pool, &name, 1).await { + Ok(false) => return DecryptionErrorType::Unknown, + Ok(true) => {} + Err(error) => { + tracing::warn!(from_user_id, "could not claim a session reset: {error}"); + return DecryptionErrorType::Unknown; + } + } + + let engine = ctx.signal_engine.lock().await; + let Some(engine) = engine.as_ref() else { + return DecryptionErrorType::Unknown; + }; + + match engine.reset_session(&name, 1, false).await { + Ok(_) => { + tracing::warn!( + from_user_id, + "signal session could not decrypt and was reset; asking the peer for a new one" + ); + DecryptionErrorType::SessionResetRequired + } + Err(error) => { + tracing::warn!(from_user_id, "could not reset the signal session: {error}"); + DecryptionErrorType::Unknown + } + } +} + /// Rebuilds a v2 Signal session for a peer that still speaks the legacy /// protocol, so the decryption error queued for the message can be answered /// with a session the peer can upgrade to. @@ -298,7 +355,10 @@ async fn upgrade_legacy_session_to_v2( kind = tracing::field::Empty ) )] -pub(crate) async fn handle_decoded_server_message( +/// Entry point for one inbound server message, whichever transport carried it. +/// Public so tests can deliver the same message twice and prove a redelivery is +/// handled rather than dropped. +pub async fn handle_decoded_server_message( ctx: &Arc, from_user_id: i64, message: proto::Message, @@ -360,7 +420,21 @@ pub(crate) async fn handle_decoded_server_message( tracing::info!("Resuming a redelivered message from its parked plaintext"); } - if !claimed && resumed_plaintext.is_none() { + // A claim with no plaintext parked on it says nothing about whether the + // message was handled: a decryption that failed leaves exactly the same + // trace as one that succeeded and committed. Dropping the redelivery on + // that ambiguity is what left a peer whose session had broken with no way + // back to the decrypt path -- and no answer at all, so they resent the same + // receipt forever. Decrypt again and let the outcome say which it was: a + // message that opens was never handled, and one the ratchet has already + // consumed comes back as a duplicate and is acknowledged below. + let retry_decryption = !claimed && resumed_plaintext.is_none() && is_encrypted_message; + + if retry_decryption { + tracing::info!("Decrypting a redelivered message that was claimed but not handled"); + } + + if !claimed && resumed_plaintext.is_none() && !retry_decryption { // Delivery receipts are terminal messages and must never themselves be // acknowledged. For regular messages Dart retries the delivery receipt // after ten days, atomically claiming the retry by moving created_at. @@ -384,6 +458,7 @@ pub(crate) async fn handle_decoded_server_message( } let mut sends_error_response = false; + let mut rebuild_session = false; match message_type { Type::SenderDeliveryReceipt => { @@ -459,12 +534,30 @@ pub(crate) async fn handle_decoded_server_message( .await? .is_some(); } + // The ratchet has already consumed this message, so it was + // handled and the peer is only missing the answer. Falling + // through without an error response queues the sender delivery + // receipt below, which is the terminal message that stops them + // resending it. + Err(TwonlyError::SignalDuplicateMessage(reason)) => { + tracing::info!( + receipt_id = message.receipt_id, + "acknowledging a message this session already handled: {reason}" + ); + } Err(error) => { tracing::warn!( receipt_id = message.receipt_id, "V2 decryption failed: {error}" ); - queue_decryption_error(&mut t, from_user_id, &message.receipt_id, 0).await?; + let error_type = reset_unusable_session(ctx, from_user_id, &error).await; + queue_decryption_error( + &mut t, + from_user_id, + &message.receipt_id, + error_type as i32, + ) + .await?; sends_error_response = true; } } @@ -474,8 +567,9 @@ pub(crate) async fn handle_decoded_server_message( let plaintext = message.plaintext_content.ok_or_else(|| { TwonlyError::Generic("plaintext client message has no content".into()) })?; - handle_plaintext_content(ctx, &mut t, from_user_id, &message.receipt_id, plaintext) - .await?; + rebuild_session = + handle_plaintext_content(ctx, &mut t, from_user_id, &message.receipt_id, plaintext) + .await?; } Type::TestNotification => {} } @@ -500,6 +594,19 @@ pub(crate) async fn handle_decoded_server_message( let ctx = ctx.clone(); tokio::spawn(async move { + // The peer reset their session, so the parked receipt has to wait for a + // session built from a fresh prekey bundle. `establish_signal_session` + // releases it, and the flush below then sends it as a prekey message + // the peer can open without any prior state. + if rebuild_session { + if let Err(error) = rebuild_session_after_peer_reset(&ctx, from_user_id).await { + tracing::warn!( + from_user_id, + "could not rebuild the signal session the peer reset: {error}" + ); + } + } + if let Err(error) = retransmit_queued_receipts(&ctx).await { tracing::warn!("failed to flush messages queued during inbound handling: {error}"); } @@ -508,6 +615,58 @@ pub(crate) async fn handle_decoded_server_message( Ok(()) } +/// Answers a peer's session reset by opening a new session with them. +/// +/// Our own record is retired first: the peer discarded the half that matches +/// it, so keeping it as the current state would only encrypt more messages +/// they cannot read. `establish_signal_session` then installs a session built +/// from their current prekey bundle and releases everything parked for it. +async fn rebuild_session_after_peer_reset(ctx: &Arc, from_user_id: i64) -> Result<()> { + let outcome = rebuild_session(ctx, from_user_id).await; + + // `handle_plaintext_content` parked the receipt for this rebuild, and + // nothing else is coming to free it. A refused claim means a rebuild is + // already in flight or just finished, and a failed one leaves the send path + // to park the receipt again for the reason it actually failed on -- both + // are better than a receipt that waits for a peer who may never write + // again. + let database = ctx.app_db.read().await.clone(); + release_deferred_receipts(&database, from_user_id).await?; + + outcome +} + +async fn rebuild_session(ctx: &Arc, from_user_id: i64) -> Result<()> { + let name = from_user_id.to_string(); + let rust_database = ctx.rust_db.read().await.clone(); + + // The same budget the receiving side spends, so a pair of clients that + // cannot agree on a session stops trading rebuilds instead of looping. + if !SessionResetLimiter::claim(&rust_database.pool, &name, 1).await? { + return Ok(()); + } + + { + let engine = ctx.signal_engine.lock().await; + let engine = engine.as_ref().ok_or(TwonlyError::SignalIdentityNotFound)?; + engine.reset_session(&name, 1, false).await?; + } + + ContactService::new(ctx) + .establish_signal_session(from_user_id, None) + .await?; + + // The repair worked, so the next unrelated failure starts from a full + // budget rather than what this one spent. + SessionResetLimiter::clear(&rust_database.pool, &name, 1).await?; + + tracing::info!( + from_user_id, + "rebuilt the signal session after a peer reset" + ); + Ok(()) +} + /// Dispatches already-decrypted client-to-client content to its concrete /// feature module. Transport decoding and Signal decryption do not belong in /// this dispatcher. @@ -675,6 +834,15 @@ async fn handle_encrypted_inner( groups::ensure_group_member(t, from_user_id, &group_id).await?; + // `ensure_group_member` accepts a member the outgoing fan-out excludes, so + // a stale `leftGroup` row makes the group one-way without a word at either + // end. Their message says the row is wrong; the group server settles it. + // Scheduled, not awaited: it is a network round trip and this transaction + // holds the app database's one connection. + if Group::has_member_left(t, &group_id, from_user_id).await? { + GroupService::spawn_stale_membership_refresh(ctx, group_id.clone(), from_user_id); + } + if content.resend_group_public_key.is_some() { return groups::handle_resend_group_public_key(t, from_user_id, &group_id).await; } diff --git a/rust/src/api/messages/incoming/messages.rs b/rust/src/api/messages/incoming/messages.rs index dfea773d..c50fd1bc 100644 --- a/rust/src/api/messages/incoming/messages.rs +++ b/rust/src/api/messages/incoming/messages.rs @@ -521,13 +521,49 @@ pub(crate) async fn send_queued_receipt(ctx: &Arc, receipt_id: &str) -> return Ok(()); }; - // The server has no account for this contact any more, so preparing the - // payload would only fetch a prekey bundle that answers `UserIdNotFound` - // on every retry. Drop the receipt instead of queueing it forever. - if row.account_deleted != 0 { - Receipt::delete(&app_db.pool, receipt_id).await?; - return Ok(()); - } + // `account_deleted` is a latch: one `UserIdNotFound` on any contact-scoped + // request sets it, and this check runs before anything that could clear it, + // so a peer who has since re-registered would stay unreachable forever -- + // every message dropped here, unsent, unacknowledged and unlogged. Ask the + // server once more before believing it, and drop only what the server + // positively denies. + let row = if row.account_deleted != 0 { + match ContactService::new(ctx) + .establish_signal_session(row.contact_id, None) + .await + { + Err(TwonlyError::PeerAccountDeleted(contact_id)) => { + tracing::warn!( + receipt_id, + contact_id, + "dropping a message for an account the server does not know" + ); + Receipt::delete(&app_db.pool, receipt_id).await?; + return Ok(()); + } + // The account exists; it has just published no bundle. An existing + // session may still encrypt for it, and if not, preparing the + // payload parks the receipt for the right reason. + Ok(()) | Err(TwonlyError::PeerHasNoPrekeyBundle(_)) => { + tracing::info!( + contact_id = row.contact_id, + "contact is registered after all; clearing the deleted-account mark" + ); + } + // Anything else -- a network failure above all -- must not be read + // as a deleted account. Leave the receipt queued for a later sweep. + Err(error) => return Err(error), + } + + // `establish_signal_session` cleared the mark and may have changed + // `signal_version`, so the payload is prepared from the current row. + match load_queued_receipt_row(&app_db.pool, receipt_id).await? { + Some(row) => row, + None => return Ok(()), + } + } else { + row + }; let receipt = match prepare_queued_receipt_from_row(ctx, receipt_id, row).await { Ok(receipt) => receipt, @@ -770,33 +806,57 @@ pub(crate) async fn handle_sender_delivery_receipt( Ok(()) } +/// Requeues the receipt a peer could not process, under a new ID. +/// +/// Returns whether the peer also retired their Signal session, in which case +/// the caller has to rebuild one before the requeued receipt can go out. The +/// receipt is parked here so the flush that follows the caller's commit cannot +/// send it through the session the peer just threw away. pub async fn handle_plaintext_content( _ctx: &Context, transaction: &mut Transaction<'_, Sqlite>, from_user_id: i64, receipt_id: &str, plaintext: proto::PlaintextContent, -) -> Result> { - if plaintext.decryption_error_message.is_some() || plaintext.retry_control_error.is_some() { - let new_receipt_id = uuid::Uuid::new_v4().to_string(); - sqlx::query!( - r#" - UPDATE receipts - SET receipt_id = ?, - mark_for_retry = CAST(strftime('%s', 'now') AS INTEGER), - retry_count = retry_count + 1, - ack_by_server_at = NULL - WHERE receipt_id = ? AND contact_id = ? - "#, - new_receipt_id, - receipt_id, - from_user_id, - ) - .execute(&mut **transaction) - .await?; - return Ok(Some(new_receipt_id)); +) -> Result { + if plaintext.decryption_error_message.is_none() && plaintext.retry_control_error.is_none() { + return Ok(false); } - Ok(None) + + // An unknown type comes from a client newer than this one. Treating it as + // a plain retry request is what every release before session resets did, + // so it stays the fallback rather than an error. + let rebuild_session = plaintext.decryption_error_message.is_some_and(|error| { + proto::plaintext_content::decryption_error_message::Type::try_from(error.r#type) + == Ok(proto::plaintext_content::decryption_error_message::Type::SessionResetRequired) + }); + + let new_receipt_id = uuid::Uuid::new_v4().to_string(); + sqlx::query!( + r#" + UPDATE receipts + SET receipt_id = ?, + mark_for_retry = CAST(strftime('%s', 'now') AS INTEGER), + retry_count = retry_count + 1, + ack_by_server_at = NULL, + -- A rebuild needs a server round-trip, so it cannot run inside this + -- transaction. `release_deferred_receipts` frees the receipt once + -- the new session stands. + deferred_until_session = CASE + WHEN ? THEN CAST(strftime('%s', 'now') AS INTEGER) + ELSE deferred_until_session + END + WHERE receipt_id = ? AND contact_id = ? + "#, + new_receipt_id, + rebuild_session, + receipt_id, + from_user_id, + ) + .execute(&mut **transaction) + .await?; + + Ok(rebuild_session) } #[cfg(test)] diff --git a/rust/src/api/proto/client/transport.proto b/rust/src/api/proto/client/transport.proto index d77400f0..95e5586f 100644 --- a/rust/src/api/proto/client/transport.proto +++ b/rust/src/api/proto/client/transport.proto @@ -25,6 +25,12 @@ message PlaintextContent { enum Type { UNKNOWN = 0; PREKEY_UNKNOWN = 1; + // The receiver's session could not open the message and will never be + // able to, so it retired the session. Resending the same ciphertext is + // pointless: the sender has to build a new session from a fresh prekey + // bundle first. Older clients ignore the type and only retry, which is + // the previous behaviour. + SESSION_RESET_REQUIRED = 2; } Type type = 1; } diff --git a/rust/src/backup/backup_archive.rs b/rust/src/backup/backup_archive.rs index 188b124a..7c79342f 100644 --- a/rust/src/backup/backup_archive.rs +++ b/rust/src/backup/backup_archive.rs @@ -324,6 +324,24 @@ impl BackupArchive { ctx.replace_rust_database(rust_database, &key_manager) .await?; + // The restored `user.json` says prekeys were published recently, but + // the ones the server hands out may have been uploaded after this + // archive was written, and their private halves are not in it. Sessions + // peers build from those bundles could never be opened. Clearing the + // 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`. + 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; + }) { + tracing::warn!("could not schedule a prekey republish after the restore: {error}"); + } + std::fs::remove_dir_all(&restore_temp_dir)?; Ok(()) diff --git a/rust/src/database/app/migrations/0017_failed_decryption_claims.sql b/rust/src/database/app/migrations/0017_failed_decryption_claims.sql new file mode 100644 index 00000000..eb07834e --- /dev/null +++ b/rust/src/database/app/migrations/0017_failed_decryption_claims.sql @@ -0,0 +1,10 @@ +-- Records that a claimed message never decrypted. +-- +-- The claim is committed before the message is handled, so a redelivery is not +-- processed twice. Without this column a claim left behind by a failed +-- decryption looks exactly like one left by a message that was handled, and the +-- redelivery is dropped -- which is how a peer whose session broke could never +-- reach the decrypt path again, and so never trigger the session reset that +-- would have repaired it. +ALTER TABLE received_receipts + ADD COLUMN decryption_failed INTEGER NOT NULL DEFAULT 0; diff --git a/rust/src/database/app/migrations/0018_drop_failed_decryption_claims.sql b/rust/src/database/app/migrations/0018_drop_failed_decryption_claims.sql new file mode 100644 index 00000000..73bc5e24 --- /dev/null +++ b/rust/src/database/app/migrations/0018_drop_failed_decryption_claims.sql @@ -0,0 +1,11 @@ +-- Drops the column added by 0017. +-- +-- It marked claims left behind by a failed decryption so their redelivery +-- would decrypt again. That distinction turned out to be both unknowable and +-- unnecessary: a claim carrying no parked plaintext says nothing about whether +-- its message was handled, and claims written before the column existed could +-- never be marked at all. Every redelivered encrypted message is decrypted +-- again instead, and the outcome answers the question -- it opens, or it comes +-- back as a duplicate the ratchet already consumed. +ALTER TABLE received_receipts + DROP COLUMN decryption_failed; diff --git a/rust/src/database/app/tables/groups.rs b/rust/src/database/app/tables/groups.rs index 4ce77ee6..b0ca0b01 100644 --- a/rust/src/database/app/tables/groups.rs +++ b/rust/src/database/app/tables/groups.rs @@ -262,6 +262,33 @@ impl Group { Ok(is_direct) } + /// Whether this member is recorded as having left the group. + /// + /// [`Self::is_member`] deliberately ignores `member_state` so a member who + /// left can still be seen to have written; the outgoing fan-out does not. + /// This is the seam between the two, for callers that need to know the + /// directions disagree. + pub async fn has_member_left( + tr: &mut Transaction<'_, Sqlite>, + group_id: &str, + contact_id: i64, + ) -> Result { + Ok(sqlx::query_scalar!( + r#" + SELECT EXISTS( + SELECT 1 + FROM group_members + WHERE group_id = ? AND contact_id = ? AND member_state = 'leftGroup' + ) + "#, + group_id, + contact_id, + ) + .fetch_one(&mut **tr) + .await? + != 0) + } + pub async fn is_member( tr: &mut Transaction<'_, Sqlite>, group_id: &str, diff --git a/rust/src/database/signal/migrations/0004_session_resets.sql b/rust/src/database/signal/migrations/0004_session_resets.sql new file mode 100644 index 00000000..dff01174 --- /dev/null +++ b/rust/src/database/signal/migrations/0004_session_resets.sql @@ -0,0 +1,14 @@ +-- Bookkeeping for Signal session resets. +-- +-- A reset asks the peer to open a fresh session. If their side cannot honour +-- that -- a mismatched identity key, a peer stuck on a state we reject -- both +-- clients would keep answering each other's reset requests forever. These rows +-- bound that: one reset per peer per cooldown, and a capped number per window. +CREATE TABLE IF NOT EXISTS signal_session_resets ( + name TEXT NOT NULL, + device_id INTEGER NOT NULL, + last_reset_at INTEGER NOT NULL, + window_started_at INTEGER NOT NULL, + resets_in_window INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (name, device_id) +); diff --git a/rust/src/error.rs b/rust/src/error.rs index de36d912..6dd55c7e 100644 --- a/rust/src/error.rs +++ b/rust/src/error.rs @@ -115,6 +115,13 @@ pub enum TwonlyError { #[error("peer {0} has published no prekey bundle")] PeerHasNoPrekeyBundle(i64), + /// The server does not know this account. Distinct from every other reason + /// a lookup can fail, because it is the only one that justifies dropping a + /// message instead of retrying it: a network error must never be mistaken + /// for a peer who no longer exists. + #[error("the server does not know peer {0}")] + PeerAccountDeleted(i64), + #[error("{0}")] IoError(#[from] std::io::Error), @@ -153,6 +160,20 @@ pub enum TwonlyError { #[error("Signal protocol error: {0}")] Signal(String), + + /// The Signal session for a peer can no longer decrypt what they send: + /// the ratchet states have diverged past repair, the session record is + /// structurally invalid, or it is gone entirely. Retrying the same + /// ciphertext against it never succeeds, so callers reset the session and + /// ask the peer to open a fresh one. See `signal::reset`. + #[error("Signal session is unusable: {0}")] + SignalSessionUnusable(String), + + /// The peer sent a message this session already ratcheted past. The + /// session is healthy; the message is simply a duplicate and must not + /// trigger a reset. + #[error("duplicated Signal message: {0}")] + SignalDuplicateMessage(String), } impl From for TwonlyError { @@ -169,6 +190,32 @@ impl From for TwonlyError { impl From for TwonlyError { fn from(error: libsignal_protocol::SignalProtocolError) -> Self { - TwonlyError::Signal(error.to_string()) + use libsignal_protocol::SignalProtocolError as Error; + + let message = error.to_string(); + match error { + // libsignal collapses every failed candidate session into a single + // `InvalidMessage`; the per-session causes ("post-quantum ratchet + // error: epoch not in valid range", "message keys not found", ...) + // only reach its log. Whatever the cause, no session in the record + // could open the message, which is exactly the reset condition. + Error::InvalidMessage(..) + | Error::SessionNotFound(..) + | Error::InvalidSessionStructure(..) + | Error::InvalidRegistrationId(..) + // A restored backup can lack the private prekey for a bundle the + // server still hands out, so inbound sessions built from it can + // never be opened either. + | Error::InvalidPreKeyId + | Error::InvalidSignedPreKeyId + | Error::InvalidKyberPreKeyId => TwonlyError::SignalSessionUnusable(message), + + Error::DuplicatedMessage(..) => TwonlyError::SignalDuplicateMessage(message), + + // An identity change is not a broken session: resetting would + // silently trust a new key. It stays a plain Signal error so the + // key-verification flow keeps owning it. + _ => TwonlyError::Signal(message), + } } } diff --git a/rust/src/services/contacts.rs b/rust/src/services/contacts.rs index a2ee1946..7d397a69 100644 --- a/rust/src/services/contacts.rs +++ b/rust/src/services/contacts.rs @@ -84,6 +84,12 @@ impl ContactService { let user = match Server::get_user_by_id(&self.ctx, user_id).await? { ServerResult::Ok(user) => user, ServerResult::ErrorCode(code) => { + // Typed, because the send path drops a message only for an + // account the server positively denies. Anything else is a + // reason to try again later. + if code == crate::api::proto::error::ErrorCode::UserIdNotFound as i32 { + return Err(TwonlyError::PeerAccountDeleted(user_id)); + } return Err(TwonlyError::Generic(format!( "Could not load prekey bundle for user {user_id}: server error {code}" ))); diff --git a/rust/src/services/groups/mod.rs b/rust/src/services/groups/mod.rs index f87d1c7d..d2e8d728 100644 --- a/rust/src/services/groups/mod.rs +++ b/rust/src/services/groups/mod.rs @@ -98,6 +98,35 @@ struct PendingCompaction { static PUBLIC_KEY_REQUESTS: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); +/// When each group was last refreshed because a member it excludes wrote to it. +/// +/// Process-local for the same reason as [`PUBLIC_KEY_REQUESTS`]: it exists to +/// keep a busy group from refetching its state on every inbound message while +/// the server still reports the member as gone, not to remember across +/// restarts. +static STALE_MEMBERSHIP_REFRESHES: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +/// Shortest gap between two refreshes prompted by the same group. +const STALE_MEMBERSHIP_REFRESH_INTERVAL: Duration = Duration::from_secs(300); + +/// Records a refresh and reports whether it should be fetched at all. +fn claim_stale_membership_refresh(group_id: &str) -> bool { + let mut refreshes = match STALE_MEMBERSHIP_REFRESHES.lock() { + Ok(refreshes) => refreshes, + Err(poisoned) => poisoned.into_inner(), + }; + let now = Instant::now(); + if refreshes + .get(group_id) + .is_some_and(|at| now.duration_since(*at) < STALE_MEMBERSHIP_REFRESH_INTERVAL) + { + return false; + } + refreshes.insert(group_id.to_owned(), now); + true +} + /// Records a request and reports whether it should be sent at all. fn claim_public_key_request(group_id: &str, contact_id: i64, force: bool) -> bool { let mut requests = match PUBLIC_KEY_REQUESTS.lock() { @@ -858,6 +887,32 @@ impl GroupService { /// lock, invert the lock order this module depends on. Running it detached /// lets the caller commit first; every call site treats the refresh as best /// effort already. + /// Reconciles a group whose local membership contradicts what the group + /// itself is doing. + /// + /// A member recorded as having left is excluded from every outgoing group + /// message, while `Group::is_member` accepts what they send: they can talk + /// to us and we can never talk to them, and nothing says so at either end. + /// A message from such a member is proof the local row is stale -- a + /// restored backup predating their join, or a membership update lost while + /// their session was broken. The group server holds the authority, so the + /// state is refetched rather than the message being taken as permission to + /// rejoin them; if the server agrees they are gone, nothing changes and the + /// interval keeps a busy group from asking again. + pub fn spawn_stale_membership_refresh(ctx: &Arc, group_id: String, contact_id: i64) { + if !claim_stale_membership_refresh(&group_id) { + return; + } + + tracing::info!( + group_id, + contact_id, + "a group member recorded as having left is still writing to the group; \ + refreshing the group state" + ); + Self::spawn_state_refresh(ctx, Some(group_id)); + } + pub fn spawn_state_refresh(ctx: &Arc, group_id: Option) { let ctx = ctx.clone(); tokio::spawn(async move { diff --git a/rust/src/services/messages.rs b/rust/src/services/messages.rs index 8bd14932..440f08b0 100644 --- a/rust/src/services/messages.rs +++ b/rust/src/services/messages.rs @@ -93,14 +93,37 @@ impl MessageService { sqlx::query!("UPDATE groups SET last_message_exchange = CAST(strftime('%s','now') AS INTEGER) WHERE group_id = ?", group_id) .execute(&mut **t).await?; } - let members = sqlx::query_scalar!( - r#"SELECT contact_id FROM group_members - WHERE group_id = ? AND (member_state IS NULL OR member_state != 'leftGroup')"#, + // Members are filtered here rather than in SQL so an excluded one can be + // named. A member the sender believes has left is absent from every + // group message, and silently: a membership row the group has moved + // past -- a restored backup holding a stale state, say -- looks exactly + // like a delivery that failed, from either end. + let rows = sqlx::query!( + r#"SELECT contact_id, member_state FROM group_members WHERE group_id = ?"#, group_id, ) .fetch_all(&mut **t) .await?; + let mut members = Vec::with_capacity(rows.len()); + for row in rows { + if row.member_state.as_deref() == Some("leftGroup") { + tracing::info!( + group_id, + contact_id = row.contact_id, + "not sending to a group member recorded as having left" + ); + continue; + } + members.push(row.contact_id); + } + + tracing::info!( + group_id, + members = members.len(), + "sending a group message to its members" + ); + let bytes = content.encode_to_vec(); let mut receipt_ids = Vec::new(); @@ -123,6 +146,12 @@ impl MessageService { .fetch_one(&mut **t) .await?; if count > 10 { + tracing::info!( + group_id, + contact_id, + open_receipts = count, + "not sending to a group member with too many open receipts" + ); continue; } } diff --git a/rust/src/signal/engine.rs b/rust/src/signal/engine.rs index 95311f08..70181c58 100644 --- a/rust/src/signal/engine.rs +++ b/rust/src/signal/engine.rs @@ -10,8 +10,8 @@ use chrono::{Duration, Utc}; use libsignal_protocol::{ message_encrypt, process_prekey_bundle, CiphertextMessageType, DeviceId, GenericSignedPreKey, IdentityKey, IdentityKeyPair, IdentityKeyStore, KyberPreKeyId, KyberPreKeyStore, PreKeyBundle, - PreKeyId, PreKeySignalMessage, PreKeyStore, ProtocolAddress, PublicKey, SignalMessage, - SignedPreKeyId, SignedPreKeyStore, Timestamp, + PreKeyId, PreKeySignalMessage, PreKeyStore, ProtocolAddress, PublicKey, SessionStore, + SignalMessage, SignedPreKeyId, SignedPreKeyStore, Timestamp, }; use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; @@ -491,6 +491,71 @@ impl RustSignalEngine { Ok(()) } + /// Retires the session with a peer so a fresh one can replace it. + /// + /// The current state is archived rather than dropped: libsignal keeps + /// trying archived states on decrypt, so messages the peer sent on chains + /// we can still follow keep arriving while the broken chain stops being + /// the one new traffic is measured against. With no current state left, + /// encrypting for the peer reports the session as missing, which + /// `encrypt_v2_with_session_recovery` already repairs from a fresh prekey + /// bundle, and an inbound prekey message installs a new session directly. + /// + /// `hard` deletes the record instead, dropping the archived states too. + /// Only a user-initiated reset should ask for that: it makes every message + /// still in flight from the peer undecryptable. + /// + /// Returns whether a session existed. + pub async fn reset_session(&self, name: &str, device_id: u32, hard: bool) -> Result { + let mut store_guard = self.store.lock().await; + let store = &mut *store_guard; + + let d_id = DeviceId::try_from(device_id) + .map_err(|_| TwonlyError::Generic(format!("Invalid device id: {}", device_id)))?; + let address = ProtocolAddress::new(name.to_owned(), d_id); + + if hard { + let device_id_value: u32 = device_id; + let deleted = sqlx::query!( + r#"DELETE FROM signal_sessions WHERE name = ? AND device_id = ?"#, + name, + device_id_value, + ) + .execute(&store.pool) + .await? + .rows_affected(); + tracing::warn!(name, device_id, "deleted the signal session"); + return Ok(deleted != 0); + } + + let Some(mut record) = store + .session_store + .load_session(&address) + .assert_send() + .await + .map_err(|e| TwonlyError::Signal(e.to_string()))? + else { + return Ok(false); + }; + + // Idempotent: a record whose current state was already archived is left + // alone by libsignal, and the archived list is capped, so repeated + // resets for the same peer cannot grow the record without bound. + record + .archive_current_state() + .map_err(|e| TwonlyError::Signal(e.to_string()))?; + + store + .session_store + .store_session(&address, &record) + .assert_send() + .await + .map_err(|e| TwonlyError::Signal(e.to_string()))?; + + tracing::warn!(name, device_id, "archived the signal session for a reset"); + Ok(true) + } + pub async fn encrypt_message( &self, name: String, @@ -574,7 +639,10 @@ impl RustSignalEngine { ) .assert_send() .await - .map_err(|e| TwonlyError::Signal(e.to_string()))? + // Classified rather than stringified: the caller has to tell a + // session that can never open this message from a duplicate or + // an identity change. See `TwonlyError::SignalSessionUnusable`. + .map_err(TwonlyError::from)? } CiphertextMessageType::PreKey => { let message = PreKeySignalMessage::try_from(ciphertext) @@ -592,7 +660,7 @@ impl RustSignalEngine { ) .assert_send() .await - .map_err(|e| TwonlyError::Signal(e.to_string()))? + .map_err(TwonlyError::from)? } _ => { return Err(TwonlyError::Signal("Invalid message type".to_string())); @@ -699,6 +767,145 @@ mod tests { assert_eq!(plaintext.to_vec(), decrypted); } + /// Encrypts on one engine and decrypts on the other. + async fn ping(from: &RustSignalEngine, to: &RustSignalEngine, text: &str) -> Result> { + let ciphertext = from + .encrypt_message(to.local_name.clone(), 1, text.as_bytes().to_vec()) + .await?; + to.decrypt_message(from.local_name.clone(), 1, ciphertext) + .await + } + + /// Overwrites a peer's session record, the way restoring a backup puts an + /// older signal database back in place. + async fn restore_session(engine: &RustSignalEngine, name: &str, record: &[u8]) { + let store = engine.store.lock().await; + sqlx::query!( + "UPDATE signal_sessions SET record_bytes = ? WHERE name = ? AND device_id = 1", + record, + name, + ) + .execute(&store.pool) + .await + .unwrap(); + } + + async fn session_record(engine: &RustSignalEngine, name: &str) -> Vec { + let store = engine.store.lock().await; + sqlx::query_scalar!( + "SELECT record_bytes FROM signal_sessions WHERE name = ? AND device_id = 1", + name, + ) + .fetch_one(&store.pool) + .await + .unwrap() + } + + /// A restored backup rewinds one side's ratchet behind the other's. The + /// session cannot be decrypted with again, the failure is reported as such, + /// and a reset plus a fresh prekey bundle brings the pair back. + #[tokio::test] + async fn a_restored_session_is_reported_unusable_and_recovers_after_a_reset() { + let (alice, _alice_dir) = create_test_engine("alice").await; + let (bob, _bob_dir) = create_test_engine("bob").await; + + alice + .process_prekey_bundle("bob".to_string(), 1, bob.generate_bundle().await.unwrap()) + .await + .unwrap(); + + assert_eq!(ping(&alice, &bob, "one").await.unwrap(), b"one"); + assert_eq!(ping(&bob, &alice, "two").await.unwrap(), b"two"); + + // The state a backup taken at this point would carry. + let backup = session_record(&bob, "alice").await; + + // Both sides keep ratcheting, so the archive falls behind a ratchet + // step whose private key it never held. + for _ in 0..3 { + ping(&alice, &bob, "before").await.unwrap(); + ping(&bob, &alice, "before").await.unwrap(); + } + + restore_session(&bob, "alice", &backup).await; + + let ciphertext = alice + .encrypt_message("bob".to_string(), 1, b"after the restore".to_vec()) + .await + .unwrap(); + let error = bob + .decrypt_message("alice".to_string(), 1, ciphertext.clone()) + .await + .expect_err("a rewound session cannot open a message from the peer's current chain"); + assert!( + matches!(error, TwonlyError::SignalSessionUnusable(_)), + "a diverged ratchet has to be classified as unusable, got: {error:?}" + ); + + // Retrying the very same ciphertext is what the old error path did, and + // it fails the same way -- which is why a reset is the only repair. + assert!(bob + .decrypt_message("alice".to_string(), 1, ciphertext) + .await + .is_err()); + + // Bob retires his half and asks Alice for a new session; Alice retires + // hers and builds one from Bob's current bundle. + assert!(bob.reset_session("alice", 1, false).await.unwrap()); + assert!(alice.reset_session("bob", 1, false).await.unwrap()); + alice + .process_prekey_bundle("bob".to_string(), 1, bob.generate_bundle().await.unwrap()) + .await + .unwrap(); + + assert_eq!(ping(&alice, &bob, "after").await.unwrap(), b"after"); + assert_eq!(ping(&bob, &alice, "back").await.unwrap(), b"back"); + } + + /// Archiving keeps the states a peer may still be sending on, so messages + /// that crossed the reset in flight are not lost with it. + #[tokio::test] + async fn a_reset_still_decrypts_messages_sent_before_it() { + let (alice, _alice_dir) = create_test_engine("alice").await; + let (bob, _bob_dir) = create_test_engine("bob").await; + + alice + .process_prekey_bundle("bob".to_string(), 1, bob.generate_bundle().await.unwrap()) + .await + .unwrap(); + + // Bob only has a record once he has received from Alice. + assert_eq!(ping(&alice, &bob, "hello").await.unwrap(), b"hello"); + + let in_flight = alice + .encrypt_message("bob".to_string(), 1, b"sent before the reset".to_vec()) + .await + .unwrap(); + + assert!(bob.reset_session("alice", 1, false).await.unwrap()); + + assert_eq!( + bob.decrypt_message("alice".to_string(), 1, in_flight) + .await + .unwrap(), + b"sent before the reset" + ); + + // A hard reset is the user-initiated one and does drop them. + let in_flight = alice + .encrypt_message("bob".to_string(), 1, b"dropped".to_vec()) + .await + .unwrap(); + assert!(bob.reset_session("alice", 1, true).await.unwrap()); + assert!(bob + .decrypt_message("alice".to_string(), 1, in_flight) + .await + .is_err()); + + // Resetting a peer we have no session with is not an error. + assert!(!bob.reset_session("carol", 1, false).await.unwrap()); + } + #[tokio::test] async fn test_twonly_api_100_messages() -> std::result::Result<(), Box> { use crate::database::signal::Database; diff --git a/rust/src/signal/mod.rs b/rust/src/signal/mod.rs index a6385ebe..5bfd71ce 100644 --- a/rust/src/signal/mod.rs +++ b/rust/src/signal/mod.rs @@ -5,4 +5,5 @@ pub mod assert_send; pub mod engine; +pub mod reset; pub mod store; diff --git a/rust/src/signal/reset.rs b/rust/src/signal/reset.rs new file mode 100644 index 00000000..9c746add --- /dev/null +++ b/rust/src/signal/reset.rs @@ -0,0 +1,192 @@ +/* + * Copyright (c) 2026, Tobias Müller git@tsmr.eu + * + */ + +//! Rate limiting for Signal session resets. +//! +//! Resetting a session is a repair both sides have to agree on: we retire our +//! record and ask the peer to open a new one. When the peer cannot honour that +//! -- their identity key no longer matches the one we trust, or their own +//! record is equally broken -- every message would trigger another reset +//! request in each direction. The counters here bound that loop: a peer gets at +//! most one reset per [`RESET_COOLDOWN`], and at most [`MAX_RESETS_PER_WINDOW`] +//! within [`RESET_WINDOW`], after which the message is dropped and the failure +//! is left to the user-visible retry path. + +use crate::error::Result; +use sqlx::SqlitePool; + +/// Shortest gap between two resets of the same session. A reset needs a server +/// round-trip and a message each way, so anything faster only stacks duplicate +/// repairs for a repair already in flight. +const RESET_COOLDOWN: i64 = 60; + +/// Window the reset budget is counted over. +const RESET_WINDOW: i64 = 60 * 60; + +/// Resets allowed per peer per [`RESET_WINDOW`]. A repair that works needs one; +/// a handful covers messages that crossed it in flight. Beyond that the session +/// is not the problem and resetting again will not make it one. +const MAX_RESETS_PER_WINDOW: i64 = 5; + +pub(crate) struct SessionResetLimiter; + +impl SessionResetLimiter { + /// Claims permission to reset the session with `name`/`device_id`. + /// + /// Returns `false` when the peer is inside the cooldown or has spent its + /// budget for the window, in which case the caller must leave the session + /// alone and let the message fail. + pub(crate) async fn claim(pool: &SqlitePool, name: &str, device_id: u32) -> Result { + let now = crate::utils::current_time().timestamp(); + let window_start = now - RESET_WINDOW; + + // One statement so two inbound messages racing on the same peer cannot + // both read a stale count and each claim the last slot. + let claimed = sqlx::query!( + r#" + INSERT INTO signal_session_resets( + name, device_id, last_reset_at, window_started_at, resets_in_window + ) + VALUES (?, ?, ?, ?, 1) + ON CONFLICT(name, device_id) DO UPDATE SET + last_reset_at = excluded.last_reset_at, + -- A window that has run out starts over with this reset as its + -- first, so a peer that broke months ago is not still blocked. + window_started_at = CASE + WHEN signal_session_resets.window_started_at <= ? + THEN excluded.last_reset_at + ELSE signal_session_resets.window_started_at + END, + resets_in_window = CASE + WHEN signal_session_resets.window_started_at <= ? THEN 1 + ELSE signal_session_resets.resets_in_window + 1 + END + -- Every SET expression above reads the row as it was before the + -- update, and this predicate decides whether the update happens at + -- all: zero affected rows means the claim was refused. A first + -- reset for a peer takes the INSERT path and is always allowed. + WHERE signal_session_resets.last_reset_at <= ? + AND (signal_session_resets.window_started_at <= ? + OR signal_session_resets.resets_in_window < ?) + "#, + name, + device_id, + now, + now, + window_start, + window_start, + // Cooldown: the update only lands once the previous reset is old + // enough for a repair to have had a chance to complete. + now - RESET_COOLDOWN, + window_start, + MAX_RESETS_PER_WINDOW, + ) + .execute(pool) + .await? + .rows_affected() + != 0; + + if !claimed { + tracing::warn!( + name, + device_id, + "signal session reset suppressed: the peer is inside the cooldown or over budget" + ); + } + + Ok(claimed) + } + + /// Forgets the reset history for a peer, so a session that works again + /// starts from a full budget. + pub(crate) async fn clear(pool: &SqlitePool, name: &str, device_id: u32) -> Result<()> { + sqlx::query!( + r#"DELETE FROM signal_session_resets WHERE name = ? AND device_id = ?"#, + name, + device_id, + ) + .execute(pool) + .await?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::database::signal::Database; + + async fn test_pool() -> SqlitePool { + let database = Database::new(&"sqlite::memory:".to_string(), None, false) + .await + .unwrap(); + database.run_migrations().await.unwrap(); + database.pool.clone() + } + + /// Moves a peer's bookkeeping back in time, standing in for a cooldown or a + /// window that has since elapsed. + async fn age_by(pool: &SqlitePool, name: &str, seconds: i64) { + sqlx::query!( + r#" + UPDATE signal_session_resets + SET last_reset_at = last_reset_at - ?, window_started_at = window_started_at - ? + WHERE name = ? + "#, + seconds, + seconds, + name, + ) + .execute(pool) + .await + .unwrap(); + } + + #[tokio::test] + async fn the_cooldown_holds_off_a_second_reset_for_the_same_peer() { + let pool = test_pool().await; + + assert!(SessionResetLimiter::claim(&pool, "1", 1).await.unwrap()); + // The repair from the first claim is still in flight. + assert!(!SessionResetLimiter::claim(&pool, "1", 1).await.unwrap()); + // A different peer has its own budget. + assert!(SessionResetLimiter::claim(&pool, "2", 1).await.unwrap()); + + age_by(&pool, "1", RESET_COOLDOWN).await; + assert!(SessionResetLimiter::claim(&pool, "1", 1).await.unwrap()); + } + + #[tokio::test] + async fn a_peer_that_cannot_be_repaired_runs_out_of_budget() { + let pool = test_pool().await; + + for reset in 0..MAX_RESETS_PER_WINDOW { + assert!( + SessionResetLimiter::claim(&pool, "1", 1).await.unwrap(), + "reset {reset} is within the budget" + ); + age_by(&pool, "1", RESET_COOLDOWN).await; + } + + // Out of budget: the cooldown has passed, but the window has not. + assert!(!SessionResetLimiter::claim(&pool, "1", 1).await.unwrap()); + + // Ageing past the window starts the budget over, so a peer that broke + // long ago is not blocked forever. + age_by(&pool, "1", RESET_WINDOW).await; + assert!(SessionResetLimiter::claim(&pool, "1", 1).await.unwrap()); + } + + #[tokio::test] + async fn a_repaired_session_starts_from_a_full_budget() { + let pool = test_pool().await; + + assert!(SessionResetLimiter::claim(&pool, "1", 1).await.unwrap()); + SessionResetLimiter::clear(&pool, "1", 1).await.unwrap(); + + // Neither the cooldown nor the spent budget survives the clear. + assert!(SessionResetLimiter::claim(&pool, "1", 1).await.unwrap()); + } +} diff --git a/rust/tests/api/session_recovery.rs b/rust/tests/api/session_recovery.rs index 47e7e8bf..c7a13a54 100644 --- a/rust/tests/api/session_recovery.rs +++ b/rust/tests/api/session_recovery.rs @@ -1,6 +1,10 @@ use super::{init_tracing, Tester}; +use prost::Message as _; +use rust_lib_twonly::api::messages::incoming::handle_decoded_server_message; +use rust_lib_twonly::api::messages::outgoing::send_c2c_message_to_contact; +use rust_lib_twonly::api::proto::client::{self as proto, encrypted_content}; use rust_lib_twonly::bridge::api::ApiConnectionState; -use rust_lib_twonly::database::app::tables::Group; +use rust_lib_twonly::database::app::tables::{Group, Receipt}; use rust_lib_twonly::services::contacts::ContactService; use rust_lib_twonly::services::messages::MessageService; @@ -66,3 +70,184 @@ async fn test_signal_session_auto_recovery_on_missing_session() -> anyhow::Resul 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 +/// a fresh prekey bundle, and the parked message has to arrive over it. +#[tokio::test] +async fn test_signal_session_reset_after_a_restored_backup() -> 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 message_id = MessageService::new(&tester_a.context) + .insert_and_send_text(group_id.clone(), "Before the backup".into(), None, None) + .await?; + tester_b + .wait_for_text_message(&message_id, tester_a.user_id, "Before the backup") + .await?; + + // The session state a backup taken at this point would hold. + let name_a = tester_a.user_id.to_string(); + let rust_db_b = tester_b.context.rust_db.read().await.clone(); + let backup = sqlx::query_scalar!( + "SELECT record_bytes FROM signal_sessions WHERE name = ? AND device_id = 1", + name_a, + ) + .fetch_one(&rust_db_b.pool) + .await?; + + // Both sides ratchet on, past the step the archive was taken at. + for round in 0..2 { + let from_b = MessageService::new(&tester_b.context) + .insert_and_send_text(group_id.clone(), format!("B {round}"), None, None) + .await?; + tester_a + .wait_for_text_message(&from_b, tester_b.user_id, &format!("B {round}")) + .await?; + let from_a = MessageService::new(&tester_a.context) + .insert_and_send_text(group_id.clone(), format!("A {round}"), None, None) + .await?; + tester_b + .wait_for_text_message(&from_a, tester_a.user_id, &format!("A {round}")) + .await?; + } + + // B restores the backup, putting its half of the session back in the past. + sqlx::query!( + "UPDATE signal_sessions SET record_bytes = ? WHERE name = ? AND device_id = 1", + backup, + name_a, + ) + .execute(&rust_db_b.pool) + .await?; + + // B cannot decrypt this and cannot be helped by a resend: it resets its + // session and asks A for a new one, A rebuilds from B's prekey bundle and + // sends the parked message over it. + let message_id = MessageService::new(&tester_a.context) + .insert_and_send_text(group_id.clone(), "After the restore".into(), None, None) + .await?; + tester_b + .wait_for_text_message(&message_id, tester_a.user_id, "After the restore") + .await?; + + // The repaired session carries traffic in both directions. + let message_id = MessageService::new(&tester_b.context) + .insert_and_send_text( + group_id.clone(), + "Reply on the new session".into(), + None, + None, + ) + .await?; + tester_a + .wait_for_text_message(&message_id, tester_b.user_id, "Reply on the new session") + .await?; + + Ok(()) +} + +/// A claim whose message was never handled -- the trace a failed decryption +/// leaves -- must not swallow the redelivery. +/// +/// The claim exists to stop a message being processed twice, but it is written +/// before the message is handled, so a decryption that failed leaves exactly +/// the same trace as one that succeeded. Dropping the redelivery on that +/// ambiguity loses the message, and answers the peer with nothing, so they +/// resend the same receipt indefinitely. +#[tokio::test] +async fn test_a_claimed_but_unhandled_message_is_delivered_on_redelivery() -> 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 sender_message_id = uuid::Uuid::new_v4().to_string(); + + // A encrypts the message without sending it, so the test controls when B + // sees it and can claim its receipt first. + let content = proto::EncryptedContent { + group_id: Some(group_id.clone()), + text_message: Some(encrypted_content::TextMessage { + sender_message_id: sender_message_id.clone(), + text: "Claimed but never handled".into(), + timestamp: chrono::Utc::now().timestamp_millis(), + quote_message_id: None, + additional_message_data: None, + }), + ..Default::default() + }; + let encoded = send_c2c_message_to_contact() + .ctx(&tester_a.context) + .contact_id(tester_b.user_id) + .encrypted_content(content.encode_to_vec()) + .only_return_encrypted_data(true) + .call() + .await? + .expect("the encrypted message is returned rather than sent"); + let message = proto::Message::decode(encoded.as_slice())?; + + // B claims the receipt and handles nothing, exactly as a decryption that + // failed leaves it. + { + let database = tester_b.context.app_db.read().await.clone(); + let mut t = database.pool.begin().await?; + assert!(Receipt::claim_received(&mut t, &message.receipt_id).await?); + t.commit().await?; + } + + // The claim must not swallow it: this content has never been handled. + handle_decoded_server_message(&tester_b.context, tester_a.user_id, message.clone()).await?; + tester_b + .wait_for_text_message( + &sender_message_id, + tester_a.user_id, + "Claimed but never handled", + ) + .await?; + + // Now it genuinely has been handled, and a further copy is the duplicate it + // looks like: the ratchet has consumed it, so it is acknowledged rather + // than delivered a second time. + handle_decoded_server_message(&tester_b.context, tester_a.user_id, message).await?; + let database = tester_b.context.app_db.read().await.clone(); + let copies = sqlx::query_scalar!( + "SELECT COUNT(*) FROM messages WHERE message_id = ?", + sender_message_id, + ) + .fetch_one(&database.pool) + .await?; + assert_eq!(copies, 1, "a redelivered message must not be stored twice"); + + Ok(()) +}