diff --git a/src/message.rs b/src/message.rs index aa75cf004..e29c87346 100644 --- a/src/message.rs +++ b/src/message.rs @@ -5,6 +5,7 @@ use log::{debug, warn}; use std::sync::Arc; use wacore::libsignal::crypto::DecryptionError; use wacore::libsignal::protocol::SenderKeyDistributionMessage; +use wacore::libsignal::protocol::SenderKeyStore; use wacore::libsignal::protocol::group_decrypt; use wacore::libsignal::protocol::process_sender_key_distribution_message; use wacore::libsignal::protocol::{ diff --git a/src/message/receive.rs b/src/message/receive.rs index ed52cd1a8..70421f904 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -1208,6 +1208,15 @@ impl Client { let sender_address = sender_for_sk.to_protocol_address(); let sender_key_name = make_sender_key_name(&info.source.chat, &sender_address); + // Two workers for the same (group, sender) can coexist after a chat-lane + // eviction; without a lock they race the ratchet advance and drop a chain + // step, leaving later skmsg undecryptable until the sender rotates an SKDM. + // The 1:1 path holds the analogous session_lock_for around its decrypt. + let chain_lock = adapter + .sender_key_store + .sender_key_lock(&sender_key_name) + .await; + for payload in payloads { let ciphertext = &payload.ciphertext[..]; let padding_version = payload.padding_version; @@ -1219,8 +1228,10 @@ impl Client { info.source.sender.observe() ); - let decrypt_result = - group_decrypt(ciphertext, &mut adapter.sender_key_store, &sender_key_name).await; + let decrypt_result = { + let _chain_guard = chain_lock.lock().await; + group_decrypt(ciphertext, &mut adapter.sender_key_store, &sender_key_name).await + }; match decrypt_result { Ok(padded_plaintext) => { diff --git a/src/message/tests.rs b/src/message/tests.rs index 0c620655e..0c3f42e60 100644 --- a/src/message/tests.rs +++ b/src/message/tests.rs @@ -6628,6 +6628,120 @@ async fn bad_session_plaintext_skips_skmsg_sibling_after_nack() { ); } +/// Happy path: a group skmsg with an established sender key decrypts through +/// process_group_enc_batch — now under the sender-key chain lock — and surfaces +/// its content. The added lock must not block the normal decrypt. +#[tokio::test] +async fn group_skmsg_decrypts_under_sender_key_lock() { + use wacore::messages::MessageUtils; + use wacore::types::events::ChannelEventHandler; + + let (client, _transport) = capturing_client("group_skmsg_lock_happy").await; + let (handler, rx) = ChannelEventHandler::new(); + client.core.event_bus.add_handler(handler); + + let (bundle, bob_jid) = bobs_prekey_bundle(&client).await; + let bob_addr = bob_jid.to_protocol_address(); + let mut alice = AlicePeer::new("146824178450540@lid").await; + alice.install_bob_session(&bob_addr, &bundle).await; + + let group: Jid = "120363408782575460@g.us".parse().expect("group"); + let skdm = alice.create_group_skdm(&group).await; + let skdm_plaintext = MessageUtils::encode_and_pad(&wa::Message { + sender_key_distribution_message: buffa::MessageField::some(skdm), + ..Default::default() + }); + let skdm_ct = alice.encrypt(&bob_addr, &skdm_plaintext).await; + + let content = MessageUtils::encode_and_pad(&wa::Message { + conversation: Some("hello group".to_string()), + ..Default::default() + }); + let skmsg = alice.encrypt_group_message(&group, &content).await; + + let id = "GROUP_SKMSG_LOCK_HAPPY"; + let info = group_message_info(id, &group, &alice.jid, false); + process_group_classified_with_sessions( + &client, + info, + &alice.jid, + vec![enc_payload_from_ciphertext(&skdm_ct)], + vec![skmsg_payload_from_bytes(skmsg)], + ) + .await; + + let mut texts = Vec::new(); + for _ in 0..80 { + texts = message_texts_for_id(&rx, id); + if !texts.is_empty() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + assert_eq!( + texts, + vec!["hello group".to_string()], + "group skmsg content must surface after decrypt under the chain lock" + ); +} + +/// Bad path: a group skmsg whose sender key was never distributed hits +/// NoSenderKeyState inside process_group_enc_batch (still under the lock) and +/// takes the retry path — one undecryptable event, no user content. +#[tokio::test] +async fn group_skmsg_without_sender_key_takes_retry_path() { + use wacore::messages::MessageUtils; + use wacore::types::events::ChannelEventHandler; + + let (client, _transport) = capturing_client("group_skmsg_lock_bad").await; + let (handler, rx) = ChannelEventHandler::new(); + client.core.event_bus.add_handler(handler); + let recorder = Arc::new(EventRecorder::default()); + client.register_handler(recorder.clone()); + + let mut alice = AlicePeer::new("146824178450541@lid").await; + let group: Jid = "120363408782575461@g.us".parse().expect("group"); + + // Init alice's own sender key so she can encrypt, but never deliver the SKDM: + // Bob has no sender key for this (group, sender). + let _ = alice.create_group_skdm(&group).await; + let content = MessageUtils::encode_and_pad(&wa::Message { + conversation: Some("no key".to_string()), + ..Default::default() + }); + let skmsg = alice.encrypt_group_message(&group, &content).await; + + let id = "GROUP_SKMSG_LOCK_BAD"; + let info = group_message_info(id, &group, &alice.jid, false); + process_group_classified_with_payloads( + &client, + info, + &alice.jid, + vec![], + vec![skmsg_payload_from_bytes(skmsg)], + vec![], + ) + .await; + + let mut undecryptable = 0; + for _ in 0..80 { + undecryptable = recorder.undecryptable().len(); + if undecryptable > 0 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + assert_eq!( + undecryptable, 1, + "a missing sender key must surface exactly one undecryptable event" + ); + assert_eq!( + message_texts_for_id(&rx, id), + Vec::::new(), + "no user content is dispatched without a sender key" + ); +} + #[tokio::test] async fn skdm_only_session_with_msmsg_waits_for_bot_payload_response() { use wacore::messages::MessageUtils;