From f84b8c8bdb5749aad4d44735a1be09fc1cceadfc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Wed, 1 Apr 2026 12:53:28 -0300 Subject: [PATCH 1/5] feat: add Signal protocol feature API and consolidate internal helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expose low-level Signal protocol operations (encrypt, decrypt, session management, participant node creation) via `client.signal().*` following the established feature pattern. Add `Event::RawNode` for raw stanza observation gated by an atomic flag. Consolidate duplicated patterns into shared helpers: - `Client::signal_adapter()` / `signal_adapter_from()` — replaces 12 inline `SignalProtocolStoreAdapter::new()` calls - `Client::session_lock_for()` — replaces 6 inline session lock patterns - `Client::get_noise_socket()` — replaces 3 inline noise socket patterns - `SignalProtocolStoreAdapter::as_signal_stores()` — replaces 6 inline `SignalStores` struct constructions - `Client::send_raw_bytes()` — `send_node` now delegates to it --- src/client.rs | 105 +++++++++++---- src/client/sessions.rs | 13 +- src/features/mod.rs | 3 + src/features/signal.rs | 254 ++++++++++++++++++++++++++++++++++++ src/lib.rs | 4 +- src/message.rs | 32 ++--- src/retry.rs | 20 +-- src/send.rs | 93 +++---------- src/store/signal_adapter.rs | 18 +++ wacore/src/send.rs | 12 +- wacore/src/types/events.rs | 7 + 11 files changed, 402 insertions(+), 159 deletions(-) create mode 100644 src/features/signal.rs diff --git a/src/client.rs b/src/client.rs index 64ff13ebd..7ec9d91f4 100644 --- a/src/client.rs +++ b/src/client.rs @@ -450,6 +450,10 @@ pub struct Client { /// Weak self-reference for spawning background tasks from `&self` methods. /// Initialized after `Arc::new(this)` in the constructor. pub(crate) self_weak: std::sync::OnceLock>, + + /// When true, emit `Event::RawNode` for every decoded stanza before router dispatch. + /// Default false — only enable when external consumers need raw protocol access. + raw_node_forwarding: AtomicBool, } impl Client { @@ -704,6 +708,7 @@ impl Client { skip_history_sync: AtomicBool::new(false), cache_config, self_weak: std::sync::OnceLock::new(), + raw_node_forwarding: AtomicBool::new(false), }; let arc = Arc::new(this); @@ -799,6 +804,68 @@ impl Client { self.core.event_bus.add_handler(handler); } + /// Enable or disable raw node forwarding. + /// When enabled, `Event::RawNode` is emitted for every decoded stanza before + /// the stanza router dispatches it. Only enable when external consumers need + /// raw protocol access (e.g. voice call stanzas). + pub fn set_raw_node_forwarding(&self, enabled: bool) { + self.raw_node_forwarding.store(enabled, Ordering::Relaxed); + } + + /// Build a [`SignalProtocolStoreAdapter`] from the current device state and signal cache. + pub(crate) async fn signal_adapter( + &self, + ) -> crate::store::signal_adapter::SignalProtocolStoreAdapter { + let device_store = self.persistence_manager.get_device_arc().await; + self.signal_adapter_from(device_store) + } + + /// Build a [`SignalProtocolStoreAdapter`] from a pre-fetched device arc. + pub(crate) fn signal_adapter_from( + &self, + device_store: Arc>, + ) -> crate::store::signal_adapter::SignalProtocolStoreAdapter { + crate::store::signal_adapter::SignalProtocolStoreAdapter::new( + device_store, + self.signal_cache.clone(), + ) + } + + /// Get the per-address session mutex from the lock cache. + pub(crate) async fn session_lock_for( + &self, + signal_addr_str: &str, + ) -> Arc> { + self.session_locks + .get_with_by_ref(signal_addr_str, async { + Arc::new(async_lock::Mutex::new(())) + }) + .await + } + + /// Get the active noise socket, or error if not connected. + pub(crate) async fn get_noise_socket( + &self, + ) -> Result, ClientError> { + self.noise_socket + .lock() + .await + .clone() + .ok_or(ClientError::NotConnected) + } + + /// Send pre-marshaled plaintext bytes through the noise socket. + pub async fn send_raw_bytes(&self, plaintext: Vec) -> Result<(), ClientError> { + let noise_socket = self.get_noise_socket().await?; + let encrypted_buf = Vec::with_capacity(plaintext.len() + 32); + noise_socket + .encrypt_and_send(plaintext, encrypted_buf) + .await?; + self.last_data_sent_ms + .store(wacore::time::now_millis().max(0) as u64, Ordering::Relaxed); + Ok(()) + } + /// Register a chatstate handler which will be invoked when a `` stanza is received. /// /// The handler receives a `ChatStateEvent` with the parsed chat state information. @@ -1349,10 +1416,9 @@ impl Client { self: &Arc, encrypted_frame: &bytes::Bytes, ) -> Option { - let noise_socket_arc = { self.noise_socket.lock().await.clone() }; - let noise_socket = match noise_socket_arc { - Some(s) => s, - None => { + let noise_socket = match self.get_noise_socket().await { + Ok(s) => s, + Err(_) => { log::error!("Cannot process frame: not connected (no noise socket)"); return None; } @@ -1519,6 +1585,14 @@ impl Client { } } + // Emit raw node for external observers (e.g. voice call handlers) + // Zero-cost when disabled: just an atomic load + if self.raw_node_forwarding.load(Ordering::Relaxed) { + self.core + .event_bus + .dispatch(&Event::RawNode(Arc::clone(&node))); + } + // Dispatch to appropriate handler using the router // Clone Arc (cheap - just reference count) not the Node itself if !self @@ -3329,39 +3403,18 @@ impl Client { } pub async fn send_node(&self, node: Node) -> Result<(), ClientError> { - let noise_socket_arc = { self.noise_socket.lock().await.clone() }; - let noise_socket = match noise_socket_arc { - Some(socket) => socket, - None => return Err(ClientError::NotConnected), - }; - debug!(target: "Client/Send", "{}", DisplayableNode(&node)); if self.sent_node_waiter_count.load(Ordering::Acquire) > 0 { self.resolve_sent_node_waiters(&Arc::new(node.clone())); } let mut plaintext_buf = Vec::with_capacity(1024); - if let Err(e) = wacore_binary::marshal::marshal_to(&node, &mut plaintext_buf) { error!("Failed to marshal node: {e:?}"); return Err(SocketError::Crypto("Marshal error".to_string()).into()); } - // Size based on plaintext + encryption overhead (16 byte tag + 3 byte frame header) - let encrypted_buf = Vec::with_capacity(plaintext_buf.len() + 32); - - if let Err(e) = noise_socket - .encrypt_and_send(plaintext_buf, encrypted_buf) - .await - { - return Err(e.into()); - } - - // WA Web: callStanza → deadSocketTimer.onOrBefore(deadSocketTime, socketId) - self.last_data_sent_ms - .store(wacore::time::now_millis().max(0) as u64, Ordering::Relaxed); - - Ok(()) + self.send_raw_bytes(plaintext_buf).await } pub(crate) async fn update_push_name_and_notify(self: &Arc, new_name: String) { diff --git a/src/client/sessions.rs b/src/client/sessions.rs index 14cf382ac..49815a4d8 100644 --- a/src/client/sessions.rs +++ b/src/client/sessions.rs @@ -198,11 +198,7 @@ impl Client { .fetch_pre_keys(jids, Some(wacore::iq::prekeys::PreKeyFetchReason::Identity)) .await?; - let device_store = self.persistence_manager.get_device_arc().await; - let mut adapter = crate::store::signal_adapter::SignalProtocolStoreAdapter::new( - device_store, - self.signal_cache.clone(), - ); + let mut adapter = self.signal_adapter().await; let mut success_count = 0; let mut missing_count = 0; @@ -213,12 +209,7 @@ impl Client { let signal_addr = jid.to_protocol_address(); // Acquire per-sender session lock to prevent race with concurrent message decryption. - let session_mutex = self - .session_locks - .get_with_by_ref(signal_addr.as_str(), async { - std::sync::Arc::new(async_lock::Mutex::new(())) - }) - .await; + let session_mutex = self.session_lock_for(signal_addr.as_str()).await; let _session_guard = session_mutex.lock().await; match process_prekey_bundle( diff --git a/src/features/mod.rs b/src/features/mod.rs index e2c673122..b42ae8278 100644 --- a/src/features/mod.rs +++ b/src/features/mod.rs @@ -10,6 +10,7 @@ pub(crate) mod newsletter; mod polls; mod presence; mod profile; +mod signal; pub(crate) mod status; mod tctoken; @@ -49,4 +50,6 @@ pub use profile::{Profile, SetProfilePictureResponse}; pub use status::{Status, StatusPrivacySetting, StatusSendOptions}; +pub use signal::Signal; + pub use tctoken::TcToken; diff --git a/src/features/signal.rs b/src/features/signal.rs new file mode 100644 index 000000000..de659e063 --- /dev/null +++ b/src/features/signal.rs @@ -0,0 +1,254 @@ +//! Low-level Signal protocol and raw transport APIs. +//! +//! Encryption, decryption, session management, and participant node creation. + +use anyhow::{Result, anyhow}; +use prost::Message as ProtoMessage; +use wacore::libsignal::protocol::{ + CiphertextMessage, PreKeySignalMessage, SignalMessage, UsePQRatchet, message_decrypt, + message_encrypt, +}; +use wacore::libsignal::store::sender_key_name::SenderKeyName; +use wacore::messages::MessageUtils; +use wacore::types::jid::JidExt; +use wacore_binary::jid::Jid; +use wacore_binary::node::Node; + +use crate::client::Client; + +/// Feature handle for Signal protocol operations. +pub struct Signal<'a> { + client: &'a Client, +} + +impl<'a> Signal<'a> { + pub(crate) fn new(client: &'a Client) -> Self { + Self { client } + } + + /// Encrypt plaintext for a single recipient using the Signal protocol. + /// + /// Returns `("msg" | "pkmsg", ciphertext_bytes)`. The caller is + /// responsible for padding if needed; this method encrypts raw bytes. + pub async fn encrypt_message( + &self, + jid: &Jid, + plaintext: &[u8], + ) -> Result<(&'static str, Vec)> { + let signal_addr = jid.to_protocol_address(); + let signal_addr_str = jid.to_protocol_address_string(); + + let lock = self.client.session_lock_for(&signal_addr_str).await; + let _guard = lock.lock().await; + let mut adapter = self.client.signal_adapter().await; + + let encrypted = message_encrypt( + plaintext, + &signal_addr, + &mut adapter.session_store, + &mut adapter.identity_store, + ) + .await?; + + match encrypted { + CiphertextMessage::PreKeySignalMessage(msg) => Ok(("pkmsg", msg.serialized().to_vec())), + CiphertextMessage::SignalMessage(msg) => Ok(("msg", msg.serialized().to_vec())), + _ => Err(anyhow!("unexpected ciphertext variant")), + } + } + + /// Decrypt a Signal protocol message from a sender. + /// + /// `msg_type` must be `"msg"` or `"pkmsg"`. Returns unpadded plaintext. + pub async fn decrypt_message( + &self, + jid: &Jid, + msg_type: &str, + ciphertext: &[u8], + ) -> Result> { + let signal_addr = jid.to_protocol_address(); + let signal_addr_str = jid.to_protocol_address_string(); + + let parsed = match msg_type { + "pkmsg" => { + CiphertextMessage::PreKeySignalMessage(PreKeySignalMessage::try_from(ciphertext)?) + } + "msg" => CiphertextMessage::SignalMessage(SignalMessage::try_from(ciphertext)?), + other => return Err(anyhow!("invalid msg_type: {other}")), + }; + + let lock = self.client.session_lock_for(&signal_addr_str).await; + let _guard = lock.lock().await; + let mut adapter = self.client.signal_adapter().await; + let mut rng = rand::make_rng::(); + + let padded = message_decrypt( + &parsed, + &signal_addr, + &mut adapter.session_store, + &mut adapter.identity_store, + &mut adapter.pre_key_store, + &adapter.signed_pre_key_store, + &mut rng, + UsePQRatchet::No, + ) + .await?; + + let unpadded = MessageUtils::unpad_message_ref(&padded, 2)?; + Ok(unpadded.to_vec()) + } + + /// Encrypt plaintext for a group using sender keys. + /// + /// Returns `(skdm_bytes, ciphertext_bytes)`. + /// + /// **Warning:** This regenerates the SKDM on every call. Callers must + /// distribute the SKDM to participants who don't yet hold the sender key. + /// For repeated sends to the same group, prefer using + /// `create_participant_nodes` which handles distribution tracking. + pub async fn encrypt_group_message( + &self, + group_jid: &Jid, + plaintext: &[u8], + ) -> Result<(Vec, Vec)> { + let own_jid = self + .client + .persistence_manager + .get_device_snapshot() + .await + .pn + .clone() + .ok_or_else(|| anyhow!("not logged in"))?; + + let mut adapter = self.client.signal_adapter().await; + let mut rng = rand::make_rng::(); + + let skdm_bytes = wacore::send::create_sender_key_distribution_message_for_group( + &mut adapter.sender_key_store, + group_jid, + &own_jid, + ) + .await?; + + let ciphertext = wacore::send::encrypt_group_message( + &mut adapter.sender_key_store, + group_jid, + &own_jid, + plaintext, + &mut rng, + ) + .await?; + + Ok((skdm_bytes, ciphertext.serialized().to_vec())) + } + + /// Decrypt a group (sender-key) message. + pub async fn decrypt_group_message( + &self, + group_jid: &Jid, + sender_jid: &Jid, + ciphertext: &[u8], + ) -> Result> { + let sender_key_name = SenderKeyName::new( + group_jid.to_string(), + sender_jid.to_protocol_address().to_string(), + ); + + let mut adapter = self.client.signal_adapter().await; + + let padded = wacore::libsignal::protocol::group_decrypt( + ciphertext, + &mut adapter.sender_key_store, + &sender_key_name, + ) + .await?; + + let unpadded = MessageUtils::unpad_message_ref(&padded, 2)?; + Ok(unpadded.to_vec()) + } + + /// Check whether a Signal session exists for `jid`. + pub async fn validate_session(&self, jid: &Jid) -> Result { + let signal_addr = jid.to_protocol_address(); + let device_store = self.client.persistence_manager.get_device_arc().await; + let device_guard = device_store.read().await; + self.client + .signal_cache + .has_session(&signal_addr, &*device_guard.backend) + .await + .map_err(|e| anyhow!("session check failed: {e}")) + } + + /// Delete Signal sessions for the given JIDs (cache + persistent store). + pub async fn delete_sessions(&self, jids: &[Jid]) -> Result<()> { + let device_store = self.client.persistence_manager.get_device_arc().await; + let device_guard = device_store.read().await; + + for jid in jids { + let addr = jid.to_protocol_address(); + let signal_addr_str = jid.to_protocol_address_string(); + + // Acquire per-address session lock to prevent races with encrypt/decrypt + let lock = self.client.session_lock_for(&signal_addr_str).await; + let _guard = lock.lock().await; + + self.client.signal_cache.delete_session(&addr).await; + device_guard + .backend + .delete_session(addr.as_str()) + .await + .map_err(|e| anyhow!("failed to delete session for {jid}: {e}"))?; + } + Ok(()) + } + + /// Create encrypted participant `` nodes for the given recipient JIDs. + /// + /// Resolves devices, ensures Signal sessions, encrypts the message for + /// each device, and returns the resulting XML nodes. + /// + /// Returns `(nodes, should_include_device_identity)`. + pub async fn create_participant_nodes( + &self, + recipient_jids: &[Jid], + message: &waproto::whatsapp::Message, + ) -> Result<(Vec, bool)> { + let device_jids = self.client.get_user_devices(recipient_jids).await?; + self.client.ensure_e2e_sessions(&device_jids).await?; + + let plaintext = MessageUtils::pad_message_v2(message.encode_to_vec()); + let mut adapter = self.client.signal_adapter().await; + let mediatype = wacore::send::media_type_from_message(message); + let hide_decrypt_fail = wacore::send::should_hide_decrypt_fail(message); + + let mut stores = adapter.as_signal_stores(); + let result = wacore::send::encrypt_for_devices( + &mut stores, + self.client, + &device_jids, + &plaintext, + hide_decrypt_fail, + mediatype, + ) + .await?; + + Ok((result.participant_nodes, result.includes_prekey_message)) + } + + /// Ensure E2E sessions exist for the given JIDs. + pub async fn assert_sessions(&self, jids: &[Jid]) -> Result<()> { + self.client.ensure_e2e_sessions(jids).await + } + + /// Get all known device JIDs for the given user JIDs via usync. + pub async fn get_user_devices(&self, jids: &[Jid]) -> Result> { + self.client.get_user_devices(jids).await + } +} + +impl Client { + /// Access low-level Signal protocol operations. + pub fn signal(&self) -> Signal<'_> { + Signal::new(self) + } +} diff --git a/src/lib.rs b/src/lib.rs index ec875644e..6f754336e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -67,8 +67,8 @@ pub use features::{ MembershipRequest, Mex, MexError, MexErrorExtensions, MexRequest, MexResponse, Newsletter, NewsletterMessage, NewsletterMessageType, NewsletterMetadata, NewsletterReactionCount, NewsletterRole, NewsletterState, NewsletterVerification, ParticipantChangeResponse, Presence, - PresenceError, PresenceStatus, Profile, ProfilePicture, SetProfilePictureResponse, Status, - StatusPrivacySetting, StatusSendOptions, SyncActionMessageRange, TcToken, + PresenceError, PresenceStatus, Profile, ProfilePicture, SetProfilePictureResponse, Signal, + Status, StatusPrivacySetting, StatusSendOptions, SyncActionMessageRange, TcToken, UnlinkSubgroupsResult, UserInfo, group_type, message_key, message_range, }; diff --git a/src/message.rs b/src/message.rs index 3f291ab5c..816e7e95e 100644 --- a/src/message.rs +++ b/src/message.rs @@ -1,5 +1,4 @@ use crate::client::Client; -use crate::store::signal_adapter::SignalProtocolStoreAdapter; use crate::types::events::Event; use crate::types::message::MessageInfo; use log::{debug, warn}; @@ -683,18 +682,10 @@ impl Client { // the SignalProtocolStoreAdapter's per-session locks (prevents ratchet counter races). let signal_address = sender_encryption_jid.to_protocol_address(); - let session_mutex = self - .session_locks - .get_with_by_ref(signal_address.as_str(), async { - std::sync::Arc::new(async_lock::Mutex::new(())) - }) - .await; + let session_mutex = self.session_lock_for(signal_address.as_str()).await; let _session_guard = session_mutex.lock().await; - let mut adapter = SignalProtocolStoreAdapter::new( - self.persistence_manager.get_device_arc().await, - self.signal_cache.clone(), - ); + let mut adapter = self.signal_adapter().await; let mut rng = rand::make_rng::(); let mut any_success = false; let mut any_duplicate = false; @@ -942,11 +933,13 @@ impl Client { let sender_jid = info.source.sender.clone(); if !sender_jid.is_bot() && !sender_jid.is_status_broadcast() { let client = self.clone(); - tokio::spawn(async move { - client - .reissue_tc_token_after_identity_change(&sender_jid) - .await; - }); + self.runtime + .spawn(Box::pin(async move { + client + .reissue_tc_token_after_identity_change(&sender_jid) + .await; + })) + .detach(); } continue; @@ -1054,10 +1047,9 @@ impl Client { if enc_nodes.is_empty() { return Ok(()); } - let device_arc = self.persistence_manager.get_device_arc().await; // Use the signal cache adapter for group decryption so sender keys are read/written // through the cache, keeping it consistent with SKDM processing. - let mut adapter = SignalProtocolStoreAdapter::new(device_arc, self.signal_cache.clone()); + let mut adapter = self.signal_adapter().await; for enc_node in enc_nodes { let ciphertext: &[u8] = match &enc_node.content { @@ -1399,15 +1391,13 @@ impl Client { }, }; - let device_arc = self.persistence_manager.get_device_arc().await; - let sender_address = sender_jid.to_protocol_address(); let sender_key_name = SenderKeyName::new(group_jid.to_string(), sender_address.to_string()); // Route through the signal cache adapter so the sender key is immediately visible // in the cache for subsequent group_decrypt calls within the same message batch. - let mut adapter = SignalProtocolStoreAdapter::new(device_arc, self.signal_cache.clone()); + let mut adapter = self.signal_adapter().await; if let Err(e) = process_sender_key_distribution_message( &sender_key_name, diff --git a/src/retry.rs b/src/retry.rs index 23dc5f7b5..98a9b6833 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -470,11 +470,7 @@ impl Client { .map(|g| g.addressing_mode) .unwrap_or_default(); - let device_store_arc = self.persistence_manager.get_device_arc().await; - let mut store_adapter = crate::store::signal_adapter::SignalProtocolStoreAdapter::new( - device_store_arc, - self.signal_cache.clone(), - ); + let mut store_adapter = self.signal_adapter().await; let stanza = wacore::send::prepare_group_retry_stanza( &mut store_adapter.session_store, @@ -634,20 +630,10 @@ impl Client { // Acquire per-sender session lock to prevent race with concurrent message decryption. // This matches the session_locks pattern used in process_session_enc_batch. - let session_mutex = self - .session_locks - .get_with_by_ref(signal_address.as_str(), async { - std::sync::Arc::new(async_lock::Mutex::new(())) - }) - .await; + let session_mutex = self.session_lock_for(signal_address.as_str()).await; let _session_guard = session_mutex.lock().await; - let device_store = self.persistence_manager.get_device_arc().await; - - let mut adapter = crate::store::signal_adapter::SignalProtocolStoreAdapter::new( - device_store, - self.signal_cache.clone(), - ); + let mut adapter = self.signal_adapter().await; process_prekey_bundle( &signal_address, diff --git a/src/send.rs b/src/send.rs index e8b4281e7..f0056ddee 100644 --- a/src/send.rs +++ b/src/send.rs @@ -1,5 +1,4 @@ use crate::client::Client; -use crate::store::signal_adapter::SignalProtocolStoreAdapter; use crate::types::message::EditAttribute; use anyhow::anyhow; use log::debug; @@ -373,15 +372,8 @@ impl Client { !key_exists }; - let mut store_adapter = - SignalProtocolStoreAdapter::new(device_store_arc.clone(), self.signal_cache.clone()); - let mut stores = wacore::send::SignalStores { - session_store: &mut store_adapter.session_store, - identity_store: &mut store_adapter.identity_store, - prekey_store: &mut store_adapter.pre_key_store, - signed_prekey_store: &store_adapter.signed_pre_key_store, - sender_key_store: &mut store_adapter.sender_key_store, - }; + let mut store_adapter = self.signal_adapter_from(device_store_arc.clone()); + let mut stores = store_adapter.as_signal_stores(); // Determine which devices need SKDM using the unified per-device map let skdm_target_devices: Option> = if force_skdm { @@ -443,17 +435,9 @@ impl Client { } self.sender_key_device_cache.invalidate(&to_str).await; - let mut store_adapter_retry = SignalProtocolStoreAdapter::new( - device_store_arc.clone(), - self.signal_cache.clone(), - ); - let mut stores_retry = wacore::send::SignalStores { - session_store: &mut store_adapter_retry.session_store, - identity_store: &mut store_adapter_retry.identity_store, - prekey_store: &mut store_adapter_retry.pre_key_store, - signed_prekey_store: &store_adapter_retry.signed_pre_key_store, - sender_key_store: &mut store_adapter_retry.sender_key_store, - }; + let mut store_adapter_retry = + self.signal_adapter_from(device_store_arc.clone()); + let mut stores_retry = store_adapter_retry.as_signal_stores(); wacore::send::prepare_group_stanza( &mut stores_retry, @@ -816,17 +800,10 @@ impl Client { let encryption_jid = self.resolve_encryption_jid(&to).await; let signal_addr_str = encryption_jid.to_protocol_address_string(); - let session_mutex = self - .session_locks - .get_with_by_ref(&signal_addr_str, async { - std::sync::Arc::new(async_lock::Mutex::new(())) - }) - .await; + let session_mutex = self.session_lock_for(&signal_addr_str).await; let _session_guard = session_mutex.lock().await; - let device_store_arc = self.persistence_manager.get_device_arc().await; - let mut store_adapter = - SignalProtocolStoreAdapter::new(device_store_arc, self.signal_cache.clone()); + let mut store_adapter = self.signal_adapter().await; wacore::send::prepare_peer_stanza( &mut store_adapter.session_store, @@ -892,18 +869,9 @@ impl Client { force_key_distribution || !key_exists }; - let mut store_adapter = SignalProtocolStoreAdapter::new( - device_store_arc.clone(), - self.signal_cache.clone(), - ); + let mut store_adapter = self.signal_adapter_from(device_store_arc.clone()); - let mut stores = wacore::send::SignalStores { - session_store: &mut store_adapter.session_store, - identity_store: &mut store_adapter.identity_store, - prekey_store: &mut store_adapter.pre_key_store, - signed_prekey_store: &store_adapter.signed_pre_key_store, - sender_key_store: &mut store_adapter.sender_key_store, - }; + let mut stores = store_adapter.as_signal_stores(); // Determine which devices need SKDM distribution using the unified // per-device sender key map (matches WA Web's participant.senderKey Map). @@ -954,17 +922,9 @@ impl Client { } self.sender_key_device_cache.invalidate(&to_str).await; - let mut store_adapter_retry = SignalProtocolStoreAdapter::new( - device_store_arc.clone(), - self.signal_cache.clone(), - ); - let mut stores_retry = wacore::send::SignalStores { - session_store: &mut store_adapter_retry.session_store, - identity_store: &mut store_adapter_retry.identity_store, - prekey_store: &mut store_adapter_retry.pre_key_store, - signed_prekey_store: &store_adapter_retry.signed_pre_key_store, - sender_key_store: &mut store_adapter_retry.sender_key_store, - }; + let mut store_adapter_retry = + self.signal_adapter_from(device_store_arc.clone()); + let mut stores_retry = store_adapter_retry.as_signal_stores(); let retry_prepared = wacore::send::prepare_group_stanza( &mut stores_retry, @@ -1068,30 +1028,16 @@ impl Client { let mut _session_mutexes = Vec::with_capacity(lock_keys.len()); for key in &lock_keys { - _session_mutexes.push( - self.session_locks - .get_with_by_ref(key, async { - std::sync::Arc::new(async_lock::Mutex::new(())) - }) - .await, - ); + _session_mutexes.push(self.session_lock_for(key).await); } let mut _session_guards = Vec::with_capacity(_session_mutexes.len()); for mutex in &_session_mutexes { _session_guards.push(mutex.lock().await); } - let device_store_arc = self.persistence_manager.get_device_arc().await; - let mut store_adapter = - SignalProtocolStoreAdapter::new(device_store_arc, self.signal_cache.clone()); - - let mut stores = wacore::send::SignalStores { - session_store: &mut store_adapter.session_store, - identity_store: &mut store_adapter.identity_store, - prekey_store: &mut store_adapter.pre_key_store, - signed_prekey_store: &store_adapter.signed_pre_key_store, - sender_key_store: &mut store_adapter.sender_key_store, - }; + let mut store_adapter = self.signal_adapter().await; + + let mut stores = store_adapter.as_signal_stores(); wacore::send::prepare_dm_stanza( &mut stores, @@ -1379,12 +1325,7 @@ impl Client { // Dedup via session_locks — bare JID won't collide with protocol addresses ("user:device") let bare = sender.to_non_ad().to_string(); - let mutex = self - .session_locks - .get_with_by_ref(bare.as_str(), async { - std::sync::Arc::new(async_lock::Mutex::new(())) - }) - .await; + let mutex = self.session_lock_for(&bare).await; let Some(_guard) = mutex.try_lock() else { return; }; diff --git a/src/store/signal_adapter.rs b/src/store/signal_adapter.rs index ca4724c52..004bf877e 100644 --- a/src/store/signal_adapter.rs +++ b/src/store/signal_adapter.rs @@ -53,6 +53,24 @@ impl SignalProtocolStoreAdapter { sender_key_store: SenderKeyAdapter(shared), } } + + pub fn as_signal_stores( + &mut self, + ) -> wacore::send::SignalStores< + '_, + SessionAdapter, + IdentityAdapter, + PreKeyAdapter, + SignedPreKeyAdapter, + > { + wacore::send::SignalStores { + session_store: &mut self.session_store, + identity_store: &mut self.identity_store, + prekey_store: &mut self.pre_key_store, + signed_prekey_store: &self.signed_pre_key_store, + sender_key_store: &mut self.sender_key_store, + } + } } #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] diff --git a/wacore/src/send.rs b/wacore/src/send.rs index 962e731ec..798bc6ccd 100644 --- a/wacore/src/send.rs +++ b/wacore/src/send.rs @@ -317,15 +317,15 @@ pub(crate) fn is_device_unregistered_error(err: &anyhow::Error) -> bool { crate::request::ServerErrorCode::from_anyhow(err).is_some_and(|e| e.code == 406) } -struct EncryptResult { - participant_nodes: Vec, - includes_prekey_message: bool, - encrypted_devices: Vec, +pub struct EncryptResult { + pub participant_nodes: Vec, + pub includes_prekey_message: bool, + pub encrypted_devices: Vec, /// True if any device returned 406 (unregistered) during prekey fetch. - had_unregistered_device: bool, + pub had_unregistered_device: bool, } -async fn encrypt_for_devices<'a, S, I, P, SP>( +pub async fn encrypt_for_devices<'a, S, I, P, SP>( stores: &mut SignalStores<'a, S, I, P, SP>, resolver: &dyn SendContextResolver, devices: &[Jid], diff --git a/wacore/src/types/events.rs b/wacore/src/types/events.rs index 4d222b938..91d91db03 100644 --- a/wacore/src/types/events.rs +++ b/wacore/src/types/events.rs @@ -449,6 +449,13 @@ pub enum Event { /// Newsletter live update (reaction counts changed, message updates, etc.). NewsletterLiveUpdate(NewsletterLiveUpdate), + + /// Raw decoded stanza, emitted before router dispatch. + /// Allows external consumers to observe protocol-level nodes (e.g. for voice calls). + /// Gated by `Client::set_raw_node_forwarding(true)` to avoid overhead when unused. + /// Serialized manually in the bridge (not via serde). + #[serde(skip)] + RawNode(Arc), } /// A newsletter live update notification, typically containing updated From e0c7703c469e77256d97d83a5742fad239bd7a45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Wed, 1 Apr 2026 13:08:43 -0300 Subject: [PATCH 2/5] =?UTF-8?q?fix:=20address=20review=20=E2=80=94=20lock?= =?UTF-8?q?=20ordering,=20missing=20session=20locks,=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - delete_sessions: move device read lock inside loop body, after session lock acquisition, to match encrypt/decrypt lock ordering and prevent AB/BA deadlock - create_participant_nodes: acquire per-device session locks before calling encrypt_for_devices (matches DM send path) - encrypt_group_message: document concurrent usage restriction - encrypt_for_devices: document session lock requirement for callers --- src/features/signal.rs | 22 ++++++++++++++++++---- wacore/src/send.rs | 4 ++++ 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/src/features/signal.rs b/src/features/signal.rs index de659e063..5fd15ad6a 100644 --- a/src/features/signal.rs +++ b/src/features/signal.rs @@ -106,6 +106,9 @@ impl<'a> Signal<'a> { /// distribute the SKDM to participants who don't yet hold the sender key. /// For repeated sends to the same group, prefer using /// `create_participant_nodes` which handles distribution tracking. + /// + /// Not safe to call concurrently with `decrypt_group_message` for the + /// same group — sender key state is not internally locked. pub async fn encrypt_group_message( &self, group_jid: &Jid, @@ -181,18 +184,18 @@ impl<'a> Signal<'a> { /// Delete Signal sessions for the given JIDs (cache + persistent store). pub async fn delete_sessions(&self, jids: &[Jid]) -> Result<()> { - let device_store = self.client.persistence_manager.get_device_arc().await; - let device_guard = device_store.read().await; - for jid in jids { let addr = jid.to_protocol_address(); let signal_addr_str = jid.to_protocol_address_string(); - // Acquire per-address session lock to prevent races with encrypt/decrypt + // Session lock first, then device lock — matches encrypt/decrypt ordering let lock = self.client.session_lock_for(&signal_addr_str).await; let _guard = lock.lock().await; self.client.signal_cache.delete_session(&addr).await; + + let device_store = self.client.persistence_manager.get_device_arc().await; + let device_guard = device_store.read().await; device_guard .backend .delete_session(addr.as_str()) @@ -216,6 +219,17 @@ impl<'a> Signal<'a> { let device_jids = self.client.get_user_devices(recipient_jids).await?; self.client.ensure_e2e_sessions(&device_jids).await?; + // Acquire per-device session locks before encrypting (matches DM send path) + let lock_keys = self.client.build_session_lock_keys(&device_jids).await; + let mut session_mutexes = Vec::with_capacity(lock_keys.len()); + for key in &lock_keys { + session_mutexes.push(self.client.session_lock_for(key).await); + } + let mut _session_guards = Vec::with_capacity(session_mutexes.len()); + for mutex in &session_mutexes { + _session_guards.push(mutex.lock().await); + } + let plaintext = MessageUtils::pad_message_v2(message.encode_to_vec()); let mut adapter = self.client.signal_adapter().await; let mediatype = wacore::send::media_type_from_message(message); diff --git a/wacore/src/send.rs b/wacore/src/send.rs index 798bc6ccd..e90865bf0 100644 --- a/wacore/src/send.rs +++ b/wacore/src/send.rs @@ -325,6 +325,10 @@ pub struct EncryptResult { pub had_unregistered_device: bool, } +/// Encrypt padded plaintext for each device JID, producing participant `` nodes. +/// +/// Callers must hold per-device session locks before calling this function — +/// concurrent ratchet mutations will corrupt Signal session state. pub async fn encrypt_for_devices<'a, S, I, P, SP>( stores: &mut SignalStores<'a, S, I, P, SP>, resolver: &dyn SendContextResolver, From 19a5b6145bd21d9127d02225495c82a7d81170dd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Wed, 1 Apr 2026 13:15:55 -0300 Subject: [PATCH 3/5] =?UTF-8?q?fix:=20address=20second=20review=20round=20?= =?UTF-8?q?=E2=80=94=20flush,=20unpad,=20group=20JID,=20dispatch=20order?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RawNode dispatch moved before xmlstreamend/IQ-response early returns so all decoded stanzas reach external observers - Add flush_signal_cache() after all Signal-mutating operations in the feature API (encrypt_message, decrypt_message, encrypt_group_message, decrypt_group_message, create_participant_nodes) — the signal cache uses deferred writes and needs explicit flushing - decrypt_message/decrypt_group_message now return raw padded bytes instead of hardcoding v2 unpadding — callers use MessageUtils::unpad_message_ref with the stanza's v attribute - encrypt_group_message uses get_own_jid_for_group() instead of hardcoded .pn, respecting LID-addressing groups - Add #[non_exhaustive] to Event enum for forward compatibility --- src/client.rs | 16 ++++++++-------- src/features/signal.rs | 39 +++++++++++++++++++++++--------------- wacore/src/types/events.rs | 1 + 3 files changed, 33 insertions(+), 23 deletions(-) diff --git a/src/client.rs b/src/client.rs index 7ec9d91f4..a4deeaeac 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1561,6 +1561,14 @@ impl Client { // Prepare deferred ACK cancellation flag (sent after dispatch unless cancelled) let mut cancelled = false; + // Emit raw node before any early returns so all decoded stanzas + // (including IQ responses and xmlstreamend) reach external observers + if self.raw_node_forwarding.load(Ordering::Relaxed) { + self.core + .event_bus + .dispatch(&Event::RawNode(Arc::clone(&node))); + } + if node.tag.as_ref() == "xmlstreamend" { if self.expected_disconnect.load(Ordering::Relaxed) { debug!("Received , expected disconnect."); @@ -1585,14 +1593,6 @@ impl Client { } } - // Emit raw node for external observers (e.g. voice call handlers) - // Zero-cost when disabled: just an atomic load - if self.raw_node_forwarding.load(Ordering::Relaxed) { - self.core - .event_bus - .dispatch(&Event::RawNode(Arc::clone(&node))); - } - // Dispatch to appropriate handler using the router // Clone Arc (cheap - just reference count) not the Node itself if !self diff --git a/src/features/signal.rs b/src/features/signal.rs index 5fd15ad6a..c2ebd7724 100644 --- a/src/features/signal.rs +++ b/src/features/signal.rs @@ -50,6 +50,9 @@ impl<'a> Signal<'a> { ) .await?; + drop(_guard); + self.client.flush_signal_cache().await?; + match encrypted { CiphertextMessage::PreKeySignalMessage(msg) => Ok(("pkmsg", msg.serialized().to_vec())), CiphertextMessage::SignalMessage(msg) => Ok(("msg", msg.serialized().to_vec())), @@ -59,7 +62,9 @@ impl<'a> Signal<'a> { /// Decrypt a Signal protocol message from a sender. /// - /// `msg_type` must be `"msg"` or `"pkmsg"`. Returns unpadded plaintext. + /// `msg_type` must be `"msg"` or `"pkmsg"`. Returns raw padded plaintext. + /// Use [`MessageUtils::unpad_message_ref`] with the stanza's `v` attribute + /// if WhatsApp message unpadding is needed. pub async fn decrypt_message( &self, jid: &Jid, @@ -82,7 +87,7 @@ impl<'a> Signal<'a> { let mut adapter = self.client.signal_adapter().await; let mut rng = rand::make_rng::(); - let padded = message_decrypt( + let plaintext = message_decrypt( &parsed, &signal_addr, &mut adapter.session_store, @@ -94,8 +99,10 @@ impl<'a> Signal<'a> { ) .await?; - let unpadded = MessageUtils::unpad_message_ref(&padded, 2)?; - Ok(unpadded.to_vec()) + drop(_guard); + self.client.flush_signal_cache().await?; + + Ok(plaintext.to_vec()) } /// Encrypt plaintext for a group using sender keys. @@ -114,14 +121,7 @@ impl<'a> Signal<'a> { group_jid: &Jid, plaintext: &[u8], ) -> Result<(Vec, Vec)> { - let own_jid = self - .client - .persistence_manager - .get_device_snapshot() - .await - .pn - .clone() - .ok_or_else(|| anyhow!("not logged in"))?; + let own_jid = self.client.get_own_jid_for_group(group_jid).await?; let mut adapter = self.client.signal_adapter().await; let mut rng = rand::make_rng::(); @@ -142,10 +142,15 @@ impl<'a> Signal<'a> { ) .await?; + self.client.flush_signal_cache().await?; + Ok((skdm_bytes, ciphertext.serialized().to_vec())) } /// Decrypt a group (sender-key) message. + /// + /// Returns raw padded plaintext. Use [`MessageUtils::unpad_message_ref`] + /// with the stanza's `v` attribute if WhatsApp message unpadding is needed. pub async fn decrypt_group_message( &self, group_jid: &Jid, @@ -159,15 +164,16 @@ impl<'a> Signal<'a> { let mut adapter = self.client.signal_adapter().await; - let padded = wacore::libsignal::protocol::group_decrypt( + let plaintext = wacore::libsignal::protocol::group_decrypt( ciphertext, &mut adapter.sender_key_store, &sender_key_name, ) .await?; - let unpadded = MessageUtils::unpad_message_ref(&padded, 2)?; - Ok(unpadded.to_vec()) + self.client.flush_signal_cache().await?; + + Ok(plaintext.to_vec()) } /// Check whether a Signal session exists for `jid`. @@ -246,6 +252,9 @@ impl<'a> Signal<'a> { ) .await?; + drop(_session_guards); + self.client.flush_signal_cache().await?; + Ok((result.participant_nodes, result.includes_prekey_message)) } diff --git a/wacore/src/types/events.rs b/wacore/src/types/events.rs index 91d91db03..16cb65dd6 100644 --- a/wacore/src/types/events.rs +++ b/wacore/src/types/events.rs @@ -380,6 +380,7 @@ pub struct DisappearingModeChanged { } #[derive(Debug, Clone, Serialize)] +#[non_exhaustive] pub enum Event { Connected(Connected), Disconnected(Disconnected), From b403569b88603fae0ec794ef8677528b2bbe6179 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Wed, 1 Apr 2026 13:25:09 -0300 Subject: [PATCH 4/5] =?UTF-8?q?fix:=20PN=E2=86=92LID=20resolution,=20lock?= =?UTF-8?q?=20ordering,=20allocation=20reduction=20in=20Signal=20API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - encrypt_message/decrypt_message: resolve PN→LID via resolve_encryption_jid before Signal session lookup, matching the internal send/receive paths that do LID-first resolution - encrypt_message/decrypt_message: derive lock key from ProtocolAddress display cache instead of computing to_signal_address_string() twice - decrypt_group_message: use to_protocol_address_string() directly for SenderKeyName, avoiding intermediate ProtocolAddress allocation - decrypt_group_message: add concurrency caveat matching encrypt side - delete_sessions: sort lock keys to prevent AB/BA deadlocks with create_participant_nodes; hoist get_device_arc() outside loop --- src/features/signal.rs | 56 +++++++++++++++++++++++++++++++----------- 1 file changed, 41 insertions(+), 15 deletions(-) diff --git a/src/features/signal.rs b/src/features/signal.rs index c2ebd7724..ac6d2e227 100644 --- a/src/features/signal.rs +++ b/src/features/signal.rs @@ -30,13 +30,19 @@ impl<'a> Signal<'a> { /// /// Returns `("msg" | "pkmsg", ciphertext_bytes)`. The caller is /// responsible for padding if needed; this method encrypts raw bytes. + /// + /// PN JIDs are resolved to LID when a LID session exists, matching + /// the internal send path. pub async fn encrypt_message( &self, jid: &Jid, plaintext: &[u8], ) -> Result<(&'static str, Vec)> { - let signal_addr = jid.to_protocol_address(); - let signal_addr_str = jid.to_protocol_address_string(); + // Resolve PN→LID to use the correct Signal session (matches send path) + let encryption_jid = self.client.resolve_encryption_jid(jid).await; + let signal_addr = encryption_jid.to_protocol_address(); + // Reuse the pre-cached display string instead of computing it separately + let signal_addr_str = signal_addr.to_string(); let lock = self.client.session_lock_for(&signal_addr_str).await; let _guard = lock.lock().await; @@ -65,15 +71,15 @@ impl<'a> Signal<'a> { /// `msg_type` must be `"msg"` or `"pkmsg"`. Returns raw padded plaintext. /// Use [`MessageUtils::unpad_message_ref`] with the stanza's `v` attribute /// if WhatsApp message unpadding is needed. + /// + /// PN JIDs are resolved to LID when a LID session exists, matching + /// the internal receive path. pub async fn decrypt_message( &self, jid: &Jid, msg_type: &str, ciphertext: &[u8], ) -> Result> { - let signal_addr = jid.to_protocol_address(); - let signal_addr_str = jid.to_protocol_address_string(); - let parsed = match msg_type { "pkmsg" => { CiphertextMessage::PreKeySignalMessage(PreKeySignalMessage::try_from(ciphertext)?) @@ -82,6 +88,10 @@ impl<'a> Signal<'a> { other => return Err(anyhow!("invalid msg_type: {other}")), }; + let encryption_jid = self.client.resolve_encryption_jid(jid).await; + let signal_addr = encryption_jid.to_protocol_address(); + let signal_addr_str = signal_addr.to_string(); + let lock = self.client.session_lock_for(&signal_addr_str).await; let _guard = lock.lock().await; let mut adapter = self.client.signal_adapter().await; @@ -151,6 +161,9 @@ impl<'a> Signal<'a> { /// /// Returns raw padded plaintext. Use [`MessageUtils::unpad_message_ref`] /// with the stanza's `v` attribute if WhatsApp message unpadding is needed. + /// + /// Not safe to call concurrently with `encrypt_group_message` for the + /// same group — sender key state is not internally locked. pub async fn decrypt_group_message( &self, group_jid: &Jid, @@ -159,7 +172,7 @@ impl<'a> Signal<'a> { ) -> Result> { let sender_key_name = SenderKeyName::new( group_jid.to_string(), - sender_jid.to_protocol_address().to_string(), + sender_jid.to_protocol_address_string(), ); let mut adapter = self.client.signal_adapter().await; @@ -190,18 +203,31 @@ impl<'a> Signal<'a> { /// Delete Signal sessions for the given JIDs (cache + persistent store). pub async fn delete_sessions(&self, jids: &[Jid]) -> Result<()> { - for jid in jids { - let addr = jid.to_protocol_address(); - let signal_addr_str = jid.to_protocol_address_string(); + if jids.is_empty() { + return Ok(()); + } - // Session lock first, then device lock — matches encrypt/decrypt ordering - let lock = self.client.session_lock_for(&signal_addr_str).await; - let _guard = lock.lock().await; + // Sort lock keys to match the global ordering used by create_participant_nodes + // and the DM send path, preventing AB/BA deadlocks. + let mut keyed: Vec<(String, &Jid)> = jids + .iter() + .map(|jid| (jid.to_protocol_address_string(), jid)) + .collect(); + keyed.sort_unstable_by(|(a, _), (b, _)| a.cmp(b)); + keyed.dedup_by(|(a, _), (b, _)| a == b); + + let mut guards = Vec::with_capacity(keyed.len()); + for (key, _) in &keyed { + let lock = self.client.session_lock_for(key).await; + guards.push(lock.lock_arc().await); + } - self.client.signal_cache.delete_session(&addr).await; + let device_store = self.client.persistence_manager.get_device_arc().await; + let device_guard = device_store.read().await; - let device_store = self.client.persistence_manager.get_device_arc().await; - let device_guard = device_store.read().await; + for (_, jid) in &keyed { + let addr = jid.to_protocol_address(); + self.client.signal_cache.delete_session(&addr).await; device_guard .backend .delete_session(addr.as_str()) From ef3530140deceb631ee58339a397efcda3027342 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Wed, 1 Apr 2026 13:43:33 -0300 Subject: [PATCH 5/5] fix: pr review things --- src/client.rs | 7 ++ src/features/mod.rs | 1 + src/features/signal.rs | 135 ++++++++++++++++++++----------------- wacore/src/types/events.rs | 3 +- 4 files changed, 84 insertions(+), 62 deletions(-) diff --git a/src/client.rs b/src/client.rs index a4deeaeac..daa5004a1 100644 --- a/src/client.rs +++ b/src/client.rs @@ -855,6 +855,13 @@ impl Client { } /// Send pre-marshaled plaintext bytes through the noise socket. + /// + /// The bytes must be a valid WABinary-marshaled stanza (as produced by + /// `wacore_binary::marshal::marshal_to`). Sending malformed data will + /// cause the server to close the connection. + /// + /// This bypasses node logging and `sent_node_waiter` resolution — use + /// [`send_node`](Client::send_node) for normal stanza sending. pub async fn send_raw_bytes(&self, plaintext: Vec) -> Result<(), ClientError> { let noise_socket = self.get_noise_socket().await?; let encrypted_buf = Vec::with_capacity(plaintext.len() + 32); diff --git a/src/features/mod.rs b/src/features/mod.rs index b42ae8278..d064aa1d1 100644 --- a/src/features/mod.rs +++ b/src/features/mod.rs @@ -51,5 +51,6 @@ pub use profile::{Profile, SetProfilePictureResponse}; pub use status::{Status, StatusPrivacySetting, StatusSendOptions}; pub use signal::Signal; +pub use wacore::message_processing::EncType; pub use tctoken::TcToken; diff --git a/src/features/signal.rs b/src/features/signal.rs index ac6d2e227..5cf5d48db 100644 --- a/src/features/signal.rs +++ b/src/features/signal.rs @@ -9,6 +9,7 @@ use wacore::libsignal::protocol::{ message_encrypt, }; use wacore::libsignal::store::sender_key_name::SenderKeyName; +use wacore::message_processing::EncType; use wacore::messages::MessageUtils; use wacore::types::jid::JidExt; use wacore_binary::jid::Jid; @@ -28,20 +29,15 @@ impl<'a> Signal<'a> { /// Encrypt plaintext for a single recipient using the Signal protocol. /// - /// Returns `("msg" | "pkmsg", ciphertext_bytes)`. The caller is - /// responsible for padding if needed; this method encrypts raw bytes. + /// Returns `(EncType, ciphertext_bytes)`. The caller is responsible + /// for padding if needed; this method encrypts raw bytes. /// /// PN JIDs are resolved to LID when a LID session exists, matching /// the internal send path. - pub async fn encrypt_message( - &self, - jid: &Jid, - plaintext: &[u8], - ) -> Result<(&'static str, Vec)> { + pub async fn encrypt_message(&self, jid: &Jid, plaintext: &[u8]) -> Result<(EncType, Vec)> { // Resolve PN→LID to use the correct Signal session (matches send path) let encryption_jid = self.client.resolve_encryption_jid(jid).await; let signal_addr = encryption_jid.to_protocol_address(); - // Reuse the pre-cached display string instead of computing it separately let signal_addr_str = signal_addr.to_string(); let lock = self.client.session_lock_for(&signal_addr_str).await; @@ -60,32 +56,39 @@ impl<'a> Signal<'a> { self.client.flush_signal_cache().await?; match encrypted { - CiphertextMessage::PreKeySignalMessage(msg) => Ok(("pkmsg", msg.serialized().to_vec())), - CiphertextMessage::SignalMessage(msg) => Ok(("msg", msg.serialized().to_vec())), + CiphertextMessage::PreKeySignalMessage(msg) => { + Ok((EncType::PreKeyMessage, msg.serialized().to_vec())) + } + CiphertextMessage::SignalMessage(msg) => { + Ok((EncType::Message, msg.serialized().to_vec())) + } _ => Err(anyhow!("unexpected ciphertext variant")), } } /// Decrypt a Signal protocol message from a sender. /// - /// `msg_type` must be `"msg"` or `"pkmsg"`. Returns raw padded plaintext. - /// Use [`MessageUtils::unpad_message_ref`] with the stanza's `v` attribute - /// if WhatsApp message unpadding is needed. + /// Returns raw padded plaintext. Use [`MessageUtils::unpad_message_ref`] + /// with the stanza's `v` attribute if WhatsApp message unpadding is needed. /// /// PN JIDs are resolved to LID when a LID session exists, matching /// the internal receive path. pub async fn decrypt_message( &self, jid: &Jid, - msg_type: &str, + enc_type: EncType, ciphertext: &[u8], ) -> Result> { - let parsed = match msg_type { - "pkmsg" => { + let parsed = match enc_type { + EncType::PreKeyMessage => { CiphertextMessage::PreKeySignalMessage(PreKeySignalMessage::try_from(ciphertext)?) } - "msg" => CiphertextMessage::SignalMessage(SignalMessage::try_from(ciphertext)?), - other => return Err(anyhow!("invalid msg_type: {other}")), + EncType::Message => { + CiphertextMessage::SignalMessage(SignalMessage::try_from(ciphertext)?) + } + EncType::SenderKey => { + return Err(anyhow!("use decrypt_group_message for sender-key messages")); + } }; let encryption_jid = self.client.resolve_encryption_jid(jid).await; @@ -117,12 +120,11 @@ impl<'a> Signal<'a> { /// Encrypt plaintext for a group using sender keys. /// - /// Returns `(skdm_bytes, ciphertext_bytes)`. - /// - /// **Warning:** This regenerates the SKDM on every call. Callers must - /// distribute the SKDM to participants who don't yet hold the sender key. - /// For repeated sends to the same group, prefer using - /// `create_participant_nodes` which handles distribution tracking. + /// Returns `(Option, ciphertext_bytes)`. The SKDM is `Some` + /// only when a new sender key was created (first encrypt for this group + /// or after key rotation). Callers must distribute the SKDM to all group + /// participants when present. This matches WA Web which only creates + /// SKDM on first group encrypt or after sender key rotation. /// /// Not safe to call concurrently with `decrypt_group_message` for the /// same group — sender key state is not internally locked. @@ -130,18 +132,37 @@ impl<'a> Signal<'a> { &self, group_jid: &Jid, plaintext: &[u8], - ) -> Result<(Vec, Vec)> { + ) -> Result<(Option>, Vec)> { let own_jid = self.client.get_own_jid_for_group(group_jid).await?; + let sender_addr = own_jid.to_protocol_address(); + let sender_key_name = SenderKeyName::new(group_jid.to_string(), sender_addr.to_string()); + + // Only create SKDM when no sender key exists (matches WA Web behavior) + let device_store = self.client.persistence_manager.get_device_arc().await; + let device_guard = device_store.read().await; + let key_exists = self + .client + .signal_cache + .get_sender_key(&sender_key_name, &*device_guard.backend) + .await? + .is_some(); + drop(device_guard); let mut adapter = self.client.signal_adapter().await; let mut rng = rand::make_rng::(); - let skdm_bytes = wacore::send::create_sender_key_distribution_message_for_group( - &mut adapter.sender_key_store, - group_jid, - &own_jid, - ) - .await?; + let skdm_bytes = if !key_exists { + Some( + wacore::send::create_sender_key_distribution_message_for_group( + &mut adapter.sender_key_store, + group_jid, + &own_jid, + ) + .await?, + ) + } else { + None + }; let ciphertext = wacore::send::encrypt_group_message( &mut adapter.sender_key_store, @@ -190,8 +211,12 @@ impl<'a> Signal<'a> { } /// Check whether a Signal session exists for `jid`. + /// + /// PN JIDs are resolved to LID when a LID mapping exists, matching + /// the encrypt/decrypt paths. pub async fn validate_session(&self, jid: &Jid) -> Result { - let signal_addr = jid.to_protocol_address(); + let resolved = self.client.resolve_encryption_jid(jid).await; + let signal_addr = resolved.to_protocol_address(); let device_store = self.client.persistence_manager.get_device_arc().await; let device_guard = device_store.read().await; self.client @@ -201,39 +226,29 @@ impl<'a> Signal<'a> { .map_err(|e| anyhow!("session check failed: {e}")) } - /// Delete Signal sessions for the given JIDs (cache + persistent store). + /// Delete Signal sessions and identity keys for the given JIDs. + /// + /// Matches WA Web's `deleteRemoteSession` which removes both session + /// and identity as a paired operation. Changes are flushed to the + /// persistent backend before returning. + /// + /// PN JIDs are resolved to LID when a LID mapping exists, matching + /// the encrypt/decrypt paths. pub async fn delete_sessions(&self, jids: &[Jid]) -> Result<()> { - if jids.is_empty() { - return Ok(()); - } + for jid in jids { + let resolved = self.client.resolve_encryption_jid(jid).await; + let addr = resolved.to_protocol_address(); + let addr_str = addr.to_string(); - // Sort lock keys to match the global ordering used by create_participant_nodes - // and the DM send path, preventing AB/BA deadlocks. - let mut keyed: Vec<(String, &Jid)> = jids - .iter() - .map(|jid| (jid.to_protocol_address_string(), jid)) - .collect(); - keyed.sort_unstable_by(|(a, _), (b, _)| a.cmp(b)); - keyed.dedup_by(|(a, _), (b, _)| a == b); - - let mut guards = Vec::with_capacity(keyed.len()); - for (key, _) in &keyed { - let lock = self.client.session_lock_for(key).await; - guards.push(lock.lock_arc().await); - } + let lock = self.client.session_lock_for(&addr_str).await; + let _guard = lock.lock().await; - let device_store = self.client.persistence_manager.get_device_arc().await; - let device_guard = device_store.read().await; - - for (_, jid) in &keyed { - let addr = jid.to_protocol_address(); + // WA Web removes session + identity together (deleteRemoteSession) self.client.signal_cache.delete_session(&addr).await; - device_guard - .backend - .delete_session(addr.as_str()) - .await - .map_err(|e| anyhow!("failed to delete session for {jid}: {e}"))?; + self.client.signal_cache.delete_identity(&addr).await; } + + self.client.flush_signal_cache().await?; Ok(()) } diff --git a/wacore/src/types/events.rs b/wacore/src/types/events.rs index 16cb65dd6..70d912f9b 100644 --- a/wacore/src/types/events.rs +++ b/wacore/src/types/events.rs @@ -452,9 +452,8 @@ pub enum Event { NewsletterLiveUpdate(NewsletterLiveUpdate), /// Raw decoded stanza, emitted before router dispatch. - /// Allows external consumers to observe protocol-level nodes (e.g. for voice calls). + /// Library extension — no WA Web equivalent (WA Web has no raw stanza observer). /// Gated by `Client::set_raw_node_forwarding(true)` to avoid overhead when unused. - /// Serialized manually in the bridge (not via serde). #[serde(skip)] RawNode(Arc), }