diff --git a/src/client.rs b/src/client.rs index 64ff13ebd..daa5004a1 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,75 @@ 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. + /// + /// 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); + 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 +1423,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; } @@ -1495,6 +1568,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."); @@ -3329,39 +3410,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..d064aa1d1 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,7 @@ 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 new file mode 100644 index 000000000..5cf5d48db --- /dev/null +++ b/src/features/signal.rs @@ -0,0 +1,318 @@ +//! 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::message_processing::EncType; +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 `(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<(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(); + 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; + + let encrypted = message_encrypt( + plaintext, + &signal_addr, + &mut adapter.session_store, + &mut adapter.identity_store, + ) + .await?; + + drop(_guard); + self.client.flush_signal_cache().await?; + + match encrypted { + 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. + /// + /// 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, + enc_type: EncType, + ciphertext: &[u8], + ) -> Result> { + let parsed = match enc_type { + EncType::PreKeyMessage => { + CiphertextMessage::PreKeySignalMessage(PreKeySignalMessage::try_from(ciphertext)?) + } + 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; + 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; + let mut rng = rand::make_rng::(); + + let plaintext = 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?; + + drop(_guard); + self.client.flush_signal_cache().await?; + + Ok(plaintext.to_vec()) + } + + /// Encrypt plaintext for a group using sender keys. + /// + /// 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. + pub async fn encrypt_group_message( + &self, + group_jid: &Jid, + plaintext: &[u8], + ) -> 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 = 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, + group_jid, + &own_jid, + plaintext, + &mut rng, + ) + .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. + /// + /// 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, + sender_jid: &Jid, + ciphertext: &[u8], + ) -> Result> { + let sender_key_name = SenderKeyName::new( + group_jid.to_string(), + sender_jid.to_protocol_address_string(), + ); + + let mut adapter = self.client.signal_adapter().await; + + let plaintext = wacore::libsignal::protocol::group_decrypt( + ciphertext, + &mut adapter.sender_key_store, + &sender_key_name, + ) + .await?; + + self.client.flush_signal_cache().await?; + + Ok(plaintext.to_vec()) + } + + /// 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 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 + .signal_cache + .has_session(&signal_addr, &*device_guard.backend) + .await + .map_err(|e| anyhow!("session check failed: {e}")) + } + + /// 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<()> { + 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(); + + let lock = self.client.session_lock_for(&addr_str).await; + let _guard = lock.lock().await; + + // WA Web removes session + identity together (deleteRemoteSession) + self.client.signal_cache.delete_session(&addr).await; + self.client.signal_cache.delete_identity(&addr).await; + } + + self.client.flush_signal_cache().await?; + 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?; + + // 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); + 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?; + + drop(_session_guards); + self.client.flush_signal_cache().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..e90865bf0 100644 --- a/wacore/src/send.rs +++ b/wacore/src/send.rs @@ -317,15 +317,19 @@ 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>( +/// 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, devices: &[Jid], diff --git a/wacore/src/types/events.rs b/wacore/src/types/events.rs index 4d222b938..70d912f9b 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), @@ -449,6 +450,12 @@ pub enum Event { /// Newsletter live update (reaction counts changed, message updates, etc.). NewsletterLiveUpdate(NewsletterLiveUpdate), + + /// Raw decoded stanza, emitted before router dispatch. + /// 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. + #[serde(skip)] + RawNode(Arc), } /// A newsletter live update notification, typically containing updated