From 344ba725541cc7cb8725c9982e7d4607eb6767a9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Mon, 6 Apr 2026 19:14:32 -0300 Subject: [PATCH] fix: validate phash from server ack to detect stale device lists MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Register an ack waiter before sending group/status messages, then validate the server's phash response in a background task. On mismatch (participant list changed during send), invalidate sender key device cache and group info cache so the next send uses fresh device lists. Non-blocking: send returns immediately after writing to socket. The phash validation runs in a detached background task with 10s timeout. Matches whatsmeow's approach (send.go:448-464) of cache invalidation on phash mismatch. WA Web does a full resend to missing devices, but cache invalidation is sufficient — the next message reaches everyone. --- src/client.rs | 14 +++++++++ src/send.rs | 79 +++++++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 87 insertions(+), 6 deletions(-) diff --git a/src/client.rs b/src/client.rs index 7e6f3abd8..0596ffcf0 100644 --- a/src/client.rs +++ b/src/client.rs @@ -3428,6 +3428,20 @@ impl Client { self.send_raw_bytes(plaintext_buf).await } + /// Register a oneshot waiter for a server ack by message ID. + /// Returns the receiver — caller sends the node separately and awaits this in background. + pub(crate) async fn register_ack_waiter( + &self, + message_id: &str, + ) -> futures::channel::oneshot::Receiver { + let (tx, rx) = futures::channel::oneshot::channel(); + self.response_waiters + .lock() + .await + .insert(message_id.to_string(), tx); + rx + } + pub(crate) async fn update_push_name_and_notify(self: &Arc, new_name: String) { let device_snapshot = self.persistence_manager.get_device_snapshot().await; let old_name = device_snapshot.push_name.clone(); diff --git a/src/send.rs b/src/send.rs index 8a4c0d278..ab46afe80 100644 --- a/src/send.rs +++ b/src/send.rs @@ -465,17 +465,29 @@ impl Client { .ensure_status_participants(prepared.node, &group_info) .await?; + let our_phash = stanza + .attrs() + .optional_string("phash") + .map(|s| s.into_owned()); + let ack_rx = if our_phash.is_some() { + Some(self.register_ack_waiter(&request_id).await) + } else { + None + }; + self.send_node(stanza).await?; + if let Some(rx) = ack_rx { + self.spawn_phash_validation(rx, our_phash.unwrap(), to.clone(), false); + } + self.update_sender_key_devices(&to_str, &prepared.skdm_devices) .await; - // Invalidate device registry for users whose devices returned 406 for user in &prepared.stale_device_users { self.invalidate_device_cache(user).await; } - // Flush cached Signal state to DB after encryption if let Err(e) = self.flush_signal_cache().await { log::error!("Failed to flush signal cache after send_status_message: {e:?}"); } @@ -604,6 +616,47 @@ impl Client { self.sender_key_device_cache.invalidate(group_jid).await; } + /// Spawn a background task to validate phash from server ack. + /// On mismatch, invalidates sender key device cache and group info cache. + fn spawn_phash_validation( + &self, + rx: futures::channel::oneshot::Receiver, + our_phash: String, + jid: Jid, + invalidate_group_cache: bool, + ) { + let Some(client) = self.self_weak.get().and_then(|w| w.upgrade()) else { + return; + }; + self.runtime + .spawn(Box::pin(async move { + let ack = match tokio::time::timeout( + std::time::Duration::from_secs(10), + rx, + ) + .await + { + Ok(Ok(node)) => node, + _ => return, + }; + if let Some(server) = ack.attrs().optional_string("phash") + && *server != our_phash + { + log::warn!( + "Phash mismatch for {jid}: ours={our_phash}, server={server}. Invalidating caches." + ); + client + .sender_key_device_cache + .invalidate(&jid.to_string()) + .await; + if invalidate_group_cache { + client.get_group_cache().await.invalidate(&jid).await; + } + } + })) + .detach(); + } + /// Ensure the status stanza has a node listing all recipient /// user JIDs. WhatsApp Web's `participantList` uses bare USER JIDs (not /// device JIDs) — `` — to tell the server @@ -1059,15 +1112,29 @@ impl Client { .await? }; + let our_phash = stanza_to_send + .attrs() + .optional_string("phash") + .map(|s| s.into_owned()); + let ack_rx = if our_phash.is_some() { + let msg_id = stanza_to_send.attrs().optional_string("id"); + Some( + self.register_ack_waiter(msg_id.as_deref().unwrap_or_default()) + .await, + ) + } else { + None + }; + self.send_node(stanza_to_send).await?; - // Update SKDM recipient cache AFTER server ACK (matches WhatsApp Web behavior). - // WA Web only calls markHasSenderKey() after the server confirms receipt. + if let Some(rx) = ack_rx { + self.spawn_phash_validation(rx, our_phash.unwrap(), tc_issue_target.clone(), true); + } + if let Some(update) = skdm_update { self.update_sender_key_devices(&update.to_str, &update.devices) .await; - // Invalidate device registry for users whose devices returned 406 - // so the next send re-fetches from server (without stale devices) for user in &update.stale_users { self.invalidate_device_cache(user).await; }