diff --git a/src/message.rs b/src/message.rs index bdd25f43b..5797d44a8 100644 --- a/src/message.rs +++ b/src/message.rs @@ -833,27 +833,34 @@ impl Client { any_duplicate = true; } else if matches!(retry_err, SignalProtocolError::InvalidPreKeyId) { - // InvalidPreKeyId after identity change means the sender is using - // an old prekey that we no longer have. This typically happens when: - // 1. The sender reinstalled WhatsApp and cached our old prekey bundle - // 2. The prekey they're using has been consumed or rotated out - // - // Solution: Send a retry receipt with a fresh prekey so the sender - // can establish a new session and resend the message. - log::warn!( - "[msg:{}] Decryption failed for {} due to InvalidPreKeyId after identity change. \ - The sender is using an old prekey we no longer have. \ - Sending retry receipt with fresh keys.", - info.id, - address - ); - - // Send retry receipt so the sender fetches our new prekey bundle - dispatched_undecryptable = self.handle_decrypt_failure( - info, - RetryReason::InvalidKeyId, - decrypt_fail_mode, - ); + // Session may exist under PN address after identity change + if self + .try_pn_to_lid_migration_decrypt( + sender_encryption_jid, + &signal_address, + &parsed_message, + &mut adapter, + &mut rng, + &enc_type, + padding_version, + info, + ) + .await + { + any_success = true; + } else { + log::warn!( + "[msg:{}] InvalidPreKeyId after identity change for {}. \ + Sending retry receipt with fresh keys.", + info.id, + address + ); + dispatched_undecryptable = self.handle_decrypt_failure( + info, + RetryReason::InvalidKeyId, + decrypt_fail_mode, + ); + } } else { log::error!( "[msg:{}] Decryption failed even after clearing untrusted identity for {}: {:?}", @@ -951,18 +958,27 @@ impl Client { self.handle_decrypt_failure(info, reason, decrypt_fail_mode); continue; } else if matches!(e, SignalProtocolError::InvalidPreKeyId) { - // InvalidPreKeyId means the sender is using a PreKey ID that we don't have. - // This typically happens when: - // 1. We were offline for a long time - // 2. The sender established a session with us using a prekey from the server - // 3. We never received the initial session-establishing message - // 4. Now we're receiving messages with counters 3, 4, 5... referencing that prekey - // - // The sender thinks they have a valid session, but we never had it. - // We need to send a retry receipt with fresh prekeys so the sender can: - // 1. Delete their old session - // 2. Fetch our new prekeys from the retry receipt - // 3. Create a NEW session and resend with counter 0 + // InvalidPreKeyId on a PreKeyMessage can also mean the + // session exists under a PN address (legacy migration). + // Migrating lets Signal use the existing ratchet state + // instead of looking up the consumed one-time prekey. + if self + .try_pn_to_lid_migration_decrypt( + sender_encryption_jid, + &signal_address, + &parsed_message, + &mut adapter, + &mut rng, + &enc_type, + padding_version, + info, + ) + .await + { + any_success = true; + continue; + } + log::warn!( "[msg:{}] Decryption failed for {} message from {} due to InvalidPreKeyId. \ Sender is using a prekey we don't have (likely session established while offline). \ diff --git a/src/send.rs b/src/send.rs index f12359858..6da30280a 100644 --- a/src/send.rs +++ b/src/send.rs @@ -501,7 +501,7 @@ impl Client { } if let Some((rx, phash)) = ack { - self.spawn_phash_validation(rx, phash, to.clone(), false, request_id.clone()); + self.spawn_phash_validation(rx, phash, to.clone(), true, request_id.clone()); } self.update_sender_key_devices(&to_str, &prepared.skdm_devices) @@ -666,6 +666,16 @@ impl Client { log::warn!( "Phash mismatch for {jid}: ours={our_phash}, server={server}. Invalidating caches." ); + // DM phash covers both recipient + own devices + // (WA Web: syncDeviceListJob([recipient, me])) + if !jid.is_group() && !jid.is_status_broadcast() { + client.invalidate_device_cache(&jid.user).await; + if let Some(own_pn) = + &client.persistence_manager.get_device_snapshot().await.pn + { + client.invalidate_device_cache(&own_pn.user).await; + } + } client .sender_key_device_cache .invalidate(&jid.to_string()) @@ -856,6 +866,7 @@ impl Client { let mut used_cached_tc_token_key: Option = None; let tc_issue_target = to.clone(); + let mut dm_phash: Option = None; let stanza_to_send: wacore_binary::Node = if peer && !to.is_group() { // Peer messages are only valid for individual users, not groups // Resolve encryption JID and acquire lock ONLY for encryption @@ -1056,20 +1067,55 @@ impl Client { } } - // DM fanout: bare recipient (device 0) + own companion devices. - // WA Web (MsgCreateFanoutStanza.js): for CHAT fanout with a single - // primary device, encrypts directly for that device only. Own devices - // get per-device enc for multi-device self-sync. The server routes - // the bare enc to the correct recipient device. + // DM fanout: all known recipient devices + own companions. + // WAWebSendUserMsgJob reads local device table only on the send + // path; WAWebDBDeviceListFanout excludes hosted devices. let recipient_bare = self.resolve_encryption_jid(&to).await.to_non_ad(); - // Populate device registry for retry handling - let _ = self.get_user_devices(std::slice::from_ref(&to)).await; - let own_devices = self.get_user_devices(std::slice::from_ref(own_jid)).await?; + // Local registry first; network warm only on miss to avoid + // unnecessary LID-migration side effects from get_user_devices + let mut recipient_cached = self.get_devices_from_registry(&recipient_bare).await; + if recipient_cached.is_none() { + let _ = self.get_user_devices(std::slice::from_ref(&to)).await; + recipient_cached = self.get_devices_from_registry(&recipient_bare).await; + } + + let mut own_cached = self.get_devices_from_registry(own_jid).await; + if own_cached.is_none() { + let _ = self.get_user_devices(std::slice::from_ref(own_jid)).await; + own_cached = self.get_devices_from_registry(own_jid).await; + } + + // Build device list, filter hosted in-place, reuse Vecs + let mut all_dm_jids = match recipient_cached { + Some(mut devices) => { + devices.retain(|j| !j.is_hosted()); + devices + } + // No record at all — bare JID, server handles fanout + None => vec![recipient_bare], + }; + + if let Some(mut own_devices) = own_cached { + own_devices.retain(|j| !j.is_hosted()); + all_dm_jids.append(&mut own_devices); + } - let mut all_dm_jids = Vec::with_capacity(1 + own_devices.len()); - all_dm_jids.push(recipient_bare); - all_dm_jids.extend(own_devices); + // Exclude exact sender device (WA Web: isMeDevice in getFanOutList) + // so ensure_e2e_sessions never creates a self-session + let own_lid = device_snapshot.lid.as_ref(); + all_dm_jids.retain(|j| { + let is_sender = (j.is_same_user_as(own_jid) && j.device == own_jid.device) + || own_lid.is_some_and(|lid| j.is_same_user_as(lid) && j.device == lid.device); + !is_sender + }); + + // Dedup for self-DMs: recipient and own device lists overlap + // when sending to own account (WA Web uses Map keyed by toString) + { + let mut seen = std::collections::HashSet::with_capacity(all_dm_jids.len()); + all_dm_jids.retain(|j| seen.insert(j.clone())); + } self.ensure_e2e_sessions(&all_dm_jids).await?; @@ -1098,7 +1144,7 @@ impl Client { let mut stores = store_adapter.as_signal_stores(); - wacore::send::prepare_dm_stanza( + let prepared = wacore::send::prepare_dm_stanza( &mut stores, self, own_jid, @@ -1111,13 +1157,12 @@ impl Client { &extra_stanza_nodes, all_dm_jids, ) - .await? + .await?; + dm_phash = prepared.phash; + prepared.node }; - let ack = if let Some(phash) = stanza_to_send - .attrs() - .optional_string("phash") - .map(|s| s.into_owned()) + let ack = if let Some(phash) = dm_phash && let Some(msg_id) = stanza_to_send .attrs() .optional_string("id") @@ -1137,7 +1182,7 @@ impl Client { } if let Some((rx, phash, msg_id)) = ack { - self.spawn_phash_validation(rx, phash, tc_issue_target.clone(), true, msg_id); + self.spawn_phash_validation(rx, phash, tc_issue_target.clone(), false, msg_id); } if let Some(update) = skdm_update { diff --git a/wacore/src/send.rs b/wacore/src/send.rs index 65ed72550..d0d626e5d 100644 --- a/wacore/src/send.rs +++ b/wacore/src/send.rs @@ -675,6 +675,16 @@ fn partition_dm_devices( (recipient_devices, own_other_devices) } +/// Result of `prepare_dm_stanza` — carries the stanza node and the +/// locally computed phash for server ACK validation. +pub struct PreparedDmStanza { + pub node: Node, + /// Locally computed phash from the sent device set. Not sent on the + /// wire (WA Web only sends phash for groups). Used by the caller to + /// compare against the server's ACK phash for device-list drift detection. + pub phash: Option, +} + #[allow(clippy::too_many_arguments)] pub async fn prepare_dm_stanza< 'a, @@ -694,7 +704,7 @@ pub async fn prepare_dm_stanza< edit: Option, extra_stanza_nodes: &[Node], all_devices: Vec, -) -> Result { +) -> Result { let reporting_result = generate_reporting_token(message, &request_id, &to_jid, &to_jid, None); let message_for_encryption = if let Some(ref result) = reporting_result { @@ -705,21 +715,29 @@ pub async fn prepare_dm_stanza< let recipient_plaintext = MessageUtils::encode_and_pad(&message_for_encryption); + // Partition first so phash reflects the actual sent set (sender excluded) + let total_devices = all_devices.len(); + let (recipient_devices, own_other_devices) = + partition_dm_devices(all_devices, own_jid, own_lid); + + let phash = { + let mut sent = Vec::with_capacity(recipient_devices.len() + own_other_devices.len()); + sent.extend_from_slice(&recipient_devices); + sent.extend_from_slice(&own_other_devices); + MessageUtils::participant_list_hash(&sent).ok() + }; + let dsm = wa::Message { device_sent_message: Some(Box::new(DeviceSentMessage { destination_jid: Some(to_jid.to_string()), message: Some(Box::new(message_for_encryption)), - phash: Some(String::new()), + phash: None, // WA Web only sets DSM phash for groups })), ..Default::default() }; let own_devices_plaintext = MessageUtils::encode_and_pad(&dsm); - let total_devices = all_devices.len(); - let (recipient_devices, own_other_devices) = - partition_dm_devices(all_devices, own_jid, own_lid); - let mut participant_nodes = Vec::with_capacity(total_devices); let mut includes_prekey_message = false; @@ -730,6 +748,12 @@ pub async fn prepare_dm_stanza< let mediatype = media_type_from_message(message); + // NOTE: WA Web has a bare- fast path for single primary device + // (WAWebSendMsgCreateFanoutStanza). Not implemented here because + // encrypt_for_devices always wraps in nodes; + // a bare-enc mode would require refactoring the encryption layer. + // The form is accepted by the server regardless. + if !recipient_devices.is_empty() { let result = encrypt_for_devices( stores, @@ -796,7 +820,10 @@ pub async fn prepare_dm_stanza< let stanza = stanza_builder.children(message_content_nodes).build(); - Ok(stanza) + Ok(PreparedDmStanza { + node: stanza, + phash, + }) } pub async fn prepare_peer_stanza(