From 226e9123b0390fdfd624d1fca290b331b8ad806b Mon Sep 17 00:00:00 2001 From: Assis Date: Tue, 10 Nov 2026 01:25:17 -0300 Subject: [PATCH 1/8] fix: keep cold ADV lookups outside receive admission Singleflight cold primary lookups with bounded query admission and cancellation-aware generation fencing. Keep owned persistence under the primary lock until storage finishes, and revalidate trust before decryption. Preserve client layout and account namespace boundaries. Add eight regressions for offline head-of-line blocking, shared success/failure, timeout, cancellation, trust rotation, persistence ownership and memory accounting. Simplify equivalent boolean expressions for stable Clippy 1.95 without relaxing assertions. Remote scoped fmt, Clippy and receive/prekey/retry/usync/queue/layout tests passed. --- src/inbound_queue.rs | 8 +- src/message/receive.rs | 226 +++++++++++++++----- src/message/tests.rs | 12 +- src/prekeys.rs | 255 ++++++++++++++++++++++- src/prekeys/adv_bootstrap_tests.rs | 324 +++++++++++++++++++++++++++++ src/send/mod.rs | 6 +- wacore/src/send/group.rs | 4 +- 7 files changed, 755 insertions(+), 80 deletions(-) create mode 100644 src/prekeys/adv_bootstrap_tests.rs diff --git a/src/inbound_queue.rs b/src/inbound_queue.rs index e3c9b8dce..fbec6def0 100644 --- a/src/inbound_queue.rs +++ b/src/inbound_queue.rs @@ -127,6 +127,7 @@ pub(crate) struct QueueBudgets { pub chat: Arc, pub ack: Arc, pub work: Arc, + pub adv_bootstrap: std::sync::OnceLock>, per_chat: Arc, pub draining: std::sync::atomic::AtomicBool, } @@ -137,6 +138,7 @@ impl Default for QueueBudgets { chat: Arc::new(ByteBudget::new(GLOBAL_CHAT_BYTES)), ack: Arc::new(ByteBudget::new(ACK_BYTES)), work: Arc::default(), + adv_bootstrap: std::sync::OnceLock::new(), per_chat: Arc::default(), draining: false.into(), } @@ -162,7 +164,11 @@ impl QueueBudgets { // Controller, global counters and work scope are counted once. + size_of::() + 2 * (size_of::() + 2 * size_of::()) - + size_of::() + 2 * size_of::(); + + size_of::() + 2 * size_of::() + + self + .adv_bootstrap + .get() + .map_or(0, |registry| registry.retained_bytes()); ( wacore::stats::CollectionStats::new(entries.len() as u64, registry_bytes as u64), GLOBAL_CHAT_BYTES.saturating_sub(self.chat.available()) as u64, diff --git a/src/message/receive.rs b/src/message/receive.rs index d2721b7c8..e93b0a9d1 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -211,6 +211,35 @@ impl Client { .await } + async fn preflight_incoming_companion_identity( + &self, + sender: &Jid, + certificate: Option<&[u8]>, + payloads: &[EncPayload], + ) -> Option { + if sender.device == 0 { + return None; + } + for payload in payloads { + if payload.enc_type != EncType::PreKeyMessage { + continue; + } + let Ok(message) = PreKeySignalMessage::try_from(payload.ciphertext.clone()) else { + continue; + }; + let identity = *message.identity_key().public_key().public_key_bytes(); + drop(message); + if self + .validate_incoming_companion_identity(sender, certificate, &identity) + .await + .is_err() + { + return Some(payload.enc_index); + } + } + None + } + async fn validate_incoming_companion_identity_inner( &self, sender: &Jid, @@ -684,7 +713,7 @@ impl Client { }) } - /// Phase 2: acquire permit, decrypt payloads, flush. No node borrows. + /// Phase 2: preflight account trust, acquire permit, decrypt, flush. #[cfg_attr( feature = "tracing", tracing::instrument(name = "wa.recv.process", level = "debug", skip_all) @@ -706,6 +735,22 @@ impl Client { ack_recipient, } = msg; + // Cold ADV may wait for an IQ. Keep that wait outside the drain permit + // so one unavailable account cannot stop every chat. This preflight + // never advances a ratchet; the account fence rechecks trust under the + // primary lock before decrypting. Stale lanes skip even the preflight. + let invalid_companion_enc = + if self.connection_generation.load(Ordering::Acquire) == lane_generation { + self.preflight_incoming_companion_identity( + &sender_encryption_jid, + device_identity.as_deref(), + &session_payloads, + ) + .await + } else { + None + }; + // Acquire the global processing permit (1 during offline sync, N after). // The helper re-acquires across a 1→N semaphore swap (offline→online): // without that, a task waiting on the old 1-permit semaphore would be @@ -744,63 +789,43 @@ impl Client { return; } - if sender_encryption_jid.device != 0 { - for payload in &session_payloads { - if payload.enc_type != EncType::PreKeyMessage { - continue; - } - let Ok(message) = PreKeySignalMessage::try_from(payload.ciphertext.clone()) else { - continue; + if let Some(invalid_companion_enc) = invalid_companion_enc { + // WA Web answers this with a retry receipt (reason + // AdvFailure). A bare return leaves the stanza unacked + // and the server redelivers it every reconnect, then + // closes the stream with ``. + log::warn!( + "Companion prekey message with an unverifiable ADV identity; requesting retry" + ); + for abandoned in session_payloads + .iter() + .chain(&group_payloads) + .chain(&bot_payloads) + { + let reason = if abandoned.enc_index == invalid_companion_enc { + EncDecryptFailureReason::UntrustedIdentity + } else { + EncDecryptFailureReason::NotAttempted }; - let identity = *message.identity_key().public_key().public_key_bytes(); - drop(message); - if self - .validate_incoming_companion_identity( - &sender_encryption_jid, - device_identity.as_deref(), - &identity, - ) - .await - .is_err() - { - // WA Web answers this with a retry receipt (reason - // AdvFailure). A bare return leaves the stanza unacked - // and the server redelivers it every reconnect, then - // closes the stream with ``. - log::warn!( - "Companion prekey message with an unverifiable ADV identity; requesting retry" - ); - for abandoned in session_payloads - .iter() - .chain(&group_payloads) - .chain(&bot_payloads) - { - let reason = if abandoned.enc_index == payload.enc_index { - EncDecryptFailureReason::UntrustedIdentity - } else { - EncDecryptFailureReason::NotAttempted - }; - self.report_enc_decrypt_failure( - &info, - abandoned.enc_index, - abandoned.enc_type.as_wire_str(), - reason, - ); - } - // Otherwise a sender-echoed count is lost after a restart or - // eviction and the retry cap restarts at 1. - self.preseed_sender_retry_count(&info, max_sender_retry_count) - .await; - self.handle_decrypt_failure( - &info, - RetryReason::AdvFailure, - decrypt_fail_mode, - ack_recipient.as_ref(), - ) - .await; - return; - } + self.report_enc_decrypt_failure( + &info, + abandoned.enc_index, + abandoned.enc_type.as_wire_str(), + reason, + ); } + // Otherwise a sender-echoed count is lost after a restart or + // eviction and the retry cap restarts at 1. + self.preseed_sender_retry_count(&info, max_sender_retry_count) + .await; + self.handle_decrypt_failure( + &info, + RetryReason::AdvFailure, + decrypt_fail_mode, + ack_recipient.as_ref(), + ) + .await; + return; } let stanza_origin = raw_pending::StanzaOrigin::new(); @@ -3197,6 +3222,95 @@ mod adv_tests { ) } + #[tokio::test] + async fn cold_adv_preflight_leaves_drain_and_warm_companion_unblocked() { + let (client, transport) = crate::test_utils::create_iq_test_client().await; + client.enter_drain_mode_for_tests(); + let cold = Jid::pn_device("12025550126", 7); + let warm = cold.with_device(9); + let (payloads, certificate, _) = companion_payload_fixture(); + let (warm_payloads, warm_certificate, warm_identity) = companion_payload_fixture(); + client + .signal_cache + .put_identity(&warm.to_protocol_address(), &warm_identity) + .await; + let generation = client.connection_generation.load(Ordering::Acquire); + let incoming = { + let client = client.clone(); + let cold = cold.clone(); + tokio::spawn(async move { + client + .process_classified_message( + ClassifiedMessage { + device_identity: Some(certificate.into()), + info: Arc::new(MessageInfo { + id: "COLD_ADV_PREFLIGHT".into(), + source: crate::types::message::MessageSource { + chat: cold.to_non_ad(), + sender: cold.clone(), + ..Default::default() + }, + ..Default::default() + }), + sender_encryption_jid: cold, + session_payloads: payloads, + group_payloads: vec![], + bot_payloads: vec![], + max_sender_retry_count: 0, + decrypt_fail_mode: crate::types::events::DecryptFailMode::Show, + ack_recipient: None, + }, + generation, + ) + .await; + }) + }; + // Leave the real IQ unanswered. A warm message in another chat may + // now own the drain permit, even if its companion shares this account. + let _request = crate::test_utils::decode_sent_iq(&transport, 0).await; + let permit = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.acquire_message_processing_permit(), + ) + .await + .expect("cold ADV must not own the offline processing permit"); + assert!( + client + .validate_incoming_companion_identity( + &warm, + Some(&warm_certificate), + &warm_identity, + ) + .await + .is_ok() + ); + let fence = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.lock_incoming_companion_account(&warm, Some(&warm_certificate), &warm_payloads), + ) + .await + .expect("cold ADV must not own the warm companion's primary lock") + .unwrap(); + assert!( + client + .cached_adv_account_identity(&cold) + .await + .unwrap() + .is_none() + ); + assert_eq!( + transport.sent_count(), + 1, + "no ACK before trust/decrypt persistence" + ); + drop(fence); + drop(permit); + incoming.abort(); + assert!(incoming.await.unwrap_err().is_cancelled()); + assert!(client.response_waiters_guard().is_empty()); + client.signal_shutdown_sync(); + } + #[tokio::test] async fn queued_companion_receive_rechecks_trust_after_recovery_publication() { use wacore::libsignal::protocol::KeyPair; diff --git a/src/message/tests.rs b/src/message/tests.rs index c78d58b08..52c979847 100644 --- a/src/message/tests.rs +++ b/src/message/tests.rs @@ -6551,9 +6551,9 @@ async fn explicit_retry_sends_only_the_retry_receipt() { continue; }; if receipt.tag.as_ref() != "receipt" - || !receipt + || receipt .get_attr("id") - .is_some_and(|value| value.as_str() == "EXPLICIT-RETRY") + .is_none_or(|value| value.as_str() != "EXPLICIT-RETRY") { continue; } @@ -7467,16 +7467,14 @@ async fn decrypt_peer_message_after( let buf = decode_frame(index, frame)?; let node = wacore_binary::marshal::unmarshal_packed_ref(&buf).ok()?; if node.tag.as_ref() != "message" - || !node - .get_attr("to") - .is_some_and(|value| value.as_str() == to) + || node.get_attr("to").is_none_or(|value| value.as_str() != to) { return None; } let enc = node.get_optional_child_by_tag(&["enc"])?; - if !enc + if enc .get_attr("type") - .is_some_and(|value| value.as_str() == "msg") + .is_none_or(|value| value.as_str() != "msg") { return None; } diff --git a/src/prekeys.rs b/src/prekeys.rs index bc0b9d960..dda1c8771 100644 --- a/src/prekeys.rs +++ b/src/prekeys.rs @@ -7,8 +7,10 @@ use crate::client::Client; use anyhow; use anyhow::Context as _; +use futures::FutureExt as _; use log; +use std::sync::Arc; use std::sync::atomic::Ordering; use wacore::iq::prekeys::{ DigestKeyBundleSpec, PreKeyCountSpec, PreKeyFetchReason, PreKeyFetchSpec, PreKeyUploadSpec, @@ -26,6 +28,129 @@ pub(crate) const DEFAULT_WANTED_PRE_KEY_COUNT: usize = 812; const MIN_PRE_KEY_COUNT: usize = 5; +/// Separate from Signal locks: a slow cold lookup must not block a warm +/// companion that owns the sole offline-processing permit. Only active flights +/// remain registered; a lease releases its joiners even when cancelled. +pub(crate) struct AdvBootstrapRegistry { + flights: std::sync::Mutex>>, + queries: async_lock::Semaphore, +} + +impl Default for AdvBootstrapRegistry { + fn default() -> Self { + Self { + flights: Default::default(), + queries: async_lock::Semaphore::new(8), + } + } +} + +type AdvBootstrapResult = Result<[u8; 32], Arc>; + +struct AdvBootstrapFlight { + result: std::sync::Mutex>, + released: async_channel::Receiver, +} + +struct AdvBootstrapLease { + registry: Arc, + key: (u64, Jid), + flight: Arc, + _release: async_channel::Sender, +} + +impl Drop for AdvBootstrapLease { + fn drop(&mut self) { + let mut flights = self + .registry + .flights + .lock() + .unwrap_or_else(|p| p.into_inner()); + if flights + .get(&self.key) + .is_some_and(|flight| Arc::ptr_eq(flight, &self.flight)) + { + flights.remove(&self.key); + } + } +} + +impl AdvBootstrapRegistry { + /// Lower bound for owned heap: registry, retained map capacity, keys and + /// registered flights. Runtime/channel waiter allocations and the opaque + /// allocation behind an anyhow error are deliberately not estimated. + pub(crate) fn retained_bytes(&self) -> usize { + use wacore::stats::HeapSize; + let flights = self.flights.lock().unwrap_or_else(|p| p.into_inner()); + size_of::() + + 2 * size_of::() + + wacore::stats::hash_table_bytes( + flights.capacity(), + size_of::<((u64, Jid), Arc)>(), + ) + + flights + .keys() + .map(|(_, primary)| { + // The map and leader lease each own the account key. + 2 * primary.heap_bytes() + + size_of::() + + 2 * size_of::() + }) + .sum::() + } + + fn claim( + self: &Arc, + generation: u64, + primary: &Jid, + ) -> Result> { + let key = (generation, primary.clone()); + let mut flights = self.flights.lock().unwrap_or_else(|p| p.into_inner()); + if let Some(flight) = flights.get(&key) { + return Err(flight.clone()); + } + let (release, released) = async_channel::bounded(1); + let flight = Arc::new(AdvBootstrapFlight { + result: Default::default(), + released, + }); + flights.insert(key.clone(), flight.clone()); + Ok(AdvBootstrapLease { + registry: self.clone(), + key, + flight, + _release: release, + }) + } +} + +#[derive(Debug)] +struct SharedAdvBootstrapError(Arc); + +impl std::fmt::Display for SharedAdvBootstrapError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + std::fmt::Display::fmt(self.0.as_ref(), f) + } +} + +impl std::error::Error for SharedAdvBootstrapError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.0.as_ref().as_ref()) + } +} + +fn ensure_adv_bootstrap_scope( + generation: &portable_atomic::AtomicU64, + expected: u64, + shutdown: &wacore::runtime::ShutdownSignal, +) -> anyhow::Result<()> { + anyhow::ensure!( + !shutdown.is_fired() && generation.load(Ordering::Acquire) == expected, + "connection changed during ADV bootstrap" + ); + Ok(()) +} + /// Whether `upload_pre_keys` should upload, given the `force` flag and the server's /// reported pre-key count. The prekey-low path forces, matching WA Web's /// `handlePreKeyLow` which uploads unconditionally, so `force` bypasses the count guard. @@ -159,6 +284,12 @@ fn authenticated_primary_identity( } impl Client { + fn adv_bootstrap(&self) -> &Arc { + self.inbound_queues() + .adv_bootstrap + .get_or_init(Arc::default) + } + /// Read a fresh, signed primary key without publishing a trust change. pub(crate) async fn fetch_adv_account_identity(&self, jid: &Jid) -> anyhow::Result<[u8; 32]> { let primary = jid.with_device(0); @@ -320,23 +451,121 @@ impl Client { /// session does. Device-list blobs never rotate an already stored identity. pub(crate) async fn ensure_adv_account_identity(&self, jid: &Jid) -> anyhow::Result<[u8; 32]> { use wacore::types::jid::JidExt; + let generation = self.connection_generation.load(Ordering::Acquire); + let shutdown = self.connection_shutdown_signal(); let primary = jid.with_device(0); + let bootstrap = async { + ensure_adv_bootstrap_scope(&self.connection_generation, generation, &shutdown)?; + { + // An existing account operation may be publishing the anchor. + // Read behind it, then release the lock before cold admission. + let address = primary.to_protocol_address(); + let lock = self.session_lock_for(address.as_str()).await; + let _guard = lock.lock().await; + ensure_adv_bootstrap_scope(&self.connection_generation, generation, &shutdown)?; + let cached = self.cached_adv_account_identity(&primary).await?; + ensure_adv_bootstrap_scope(&self.connection_generation, generation, &shutdown)?; + if let Some(identity) = cached { + return Ok(identity); + } + } + match self.adv_bootstrap().claim(generation, &primary) { + Ok(lease) => { + let result = self + .bootstrap_adv_account_identity(&primary, generation, &shutdown) + .await + .map_err(Arc::new); + ensure_adv_bootstrap_scope(&self.connection_generation, generation, &shutdown)?; + *lease + .flight + .result + .lock() + .unwrap_or_else(|p| p.into_inner()) = Some(result.clone()); + result.map_err(|error| anyhow::Error::new(SharedAdvBootstrapError(error))) + } + Err(flight) => { + let _ = flight.released.recv().await; + ensure_adv_bootstrap_scope(&self.connection_generation, generation, &shutdown)?; + let result = flight + .result + .lock() + .unwrap_or_else(|p| p.into_inner()) + .clone(); + result + .ok_or_else(|| anyhow::anyhow!("ADV bootstrap cancelled"))? + .map_err(|error| anyhow::Error::new(SharedAdvBootstrapError(error))) + } + } + }; + futures::select_biased! { + _ = wacore::runtime::wait_for_shutdown(&shutdown).fuse() => { + anyhow::bail!("connection closed during ADV bootstrap") + }, + result = Box::pin(bootstrap).fuse() => result, + } + } + + async fn bootstrap_adv_account_identity( + &self, + primary: &Jid, + generation: u64, + shutdown: &wacore::runtime::ShutdownSignal, + ) -> anyhow::Result<[u8; 32]> { + use wacore::types::jid::JidExt; + let query = self.adv_bootstrap().queries.acquire().await; + ensure_adv_bootstrap_scope(&self.connection_generation, generation, shutdown)?; + // A previous flight may have published while this caller waited for + // admission. No Signal lock is held across either admission or the IQ. + let cached = self.cached_adv_account_identity(primary).await?; + ensure_adv_bootstrap_scope(&self.connection_generation, generation, shutdown)?; + if let Some(identity) = cached { + return Ok(identity); + } + let key = self.query_adv_account_identity(primary, None).await?; + drop(query); let address = primary.to_protocol_address(); let lock = self.session_lock_for(address.as_str()).await; - let _guard = lock.lock().await; - if let Some(identity) = self.cached_adv_account_identity(jid).await? { + let guard = lock.lock_arc().await; + ensure_adv_bootstrap_scope(&self.connection_generation, generation, shutdown)?; + // Recovery or a primary message may have established a different + // anchor during the query. Validate the caller against current trust; + // a delayed cold response must never replace it. + let cached = self.cached_adv_account_identity(primary).await?; + ensure_adv_bootstrap_scope(&self.connection_generation, generation, shutdown)?; + if let Some(identity) = cached { return Ok(identity); } - let key = self.query_adv_account_identity(&primary, None).await?; + let work = self + .inbound_queues() + .work + .try_track() + .ok_or_else(|| anyhow::anyhow!("ADV persistence admission closed"))?; let backend = self.persistence_manager.backend(); - if !self - .signal_cache - .persist_identity_if_absent_or_equal(&address, &key, &*backend) - .await? - { - anyhow::bail!("primary identity changed during ADV bootstrap"); - } - Ok(key) + let cache = self.signal_cache.clone(); + let current_generation = self.connection_generation.clone(); + let shutdown = shutdown.clone(); + let (send, receive) = futures::channel::oneshot::channel(); + // Once storage is admitted, its writer owns the primary fence and the + // drain tracking until the backend finishes. Cancelling an IQ caller + // must not let recovery overtake a host write that continues detached. + self.runtime.spawn_detached(Box::pin(async move { + let _guard = guard; + let _work = work; + let result = async { + ensure_adv_bootstrap_scope(¤t_generation, generation, &shutdown)?; + let accepted = cache + .persist_identity_if_absent_or_equal(&address, &key, &*backend) + .await?; + ensure_adv_bootstrap_scope(¤t_generation, generation, &shutdown)?; + anyhow::ensure!(accepted, "primary identity changed during ADV bootstrap"); + Ok(key) + } + .await; + let _ = send.send(result); + })); + receive + .await + .map_err(|_| anyhow::anyhow!("ADV persistence worker stopped"))? } /// Load, for each companion JID, its account (device 0) identity key from the @@ -1130,6 +1359,10 @@ impl Client { } } +#[cfg(test)] +#[path = "prekeys/adv_bootstrap_tests.rs"] +mod adv_bootstrap_tests; + #[cfg(test)] mod tests { use super::{ diff --git a/src/prekeys/adv_bootstrap_tests.rs b/src/prekeys/adv_bootstrap_tests.rs new file mode 100644 index 000000000..5d7b59eb4 --- /dev/null +++ b/src/prekeys/adv_bootstrap_tests.rs @@ -0,0 +1,324 @@ +use super::*; +use crate::test_utils::{ + create_iq_test_client, decode_sent_iq, node_to_owned_ref, poll_until, wait_for_lock_waiter, +}; +use std::time::Duration; +use wacore::iq::prekeys::PreKeyBundleUserNode; +use wacore::libsignal::protocol::{IdentityKey, PreKeyBundle}; +use wacore::protocol::ProtocolNode; +use wacore::types::jid::JidExt; +use wacore_binary::builder::NodeBuilder; + +#[test] +fn adv_bootstrap_memory_reports_retained_capacity_after_cancelled_flights() { + let budgets = crate::inbound_queue::QueueBudgets::default(); + let baseline = budgets.memory_stats().0; + assert!(budgets.adv_bootstrap.get().is_none()); + let registry = budgets.adv_bootstrap.get_or_init(Arc::default); + let initialized = budgets.memory_stats().0; + assert!(initialized.bytes > baseline.bytes); + let mut first_peak = None; + for _ in 0..16 { + let leases: Vec<_> = (0..8) + .map(|index| { + registry + .claim(0, &Jid::pn(format!("12025551{index:03}"))) + .ok() + .expect("independent account flight") + }) + .collect(); + let active = budgets.memory_stats().0; + assert!(active.bytes > initialized.bytes); + assert_eq!(active.entries, baseline.entries); + let first_peak = *first_peak.get_or_insert(active.bytes); + drop(leases); + assert!(registry.flights.lock().unwrap().is_empty()); + let retained = budgets.memory_stats().0; + assert_eq!(retained.entries, baseline.entries); + assert!( + retained.bytes > initialized.bytes, + "empty map retains its buffer" + ); + assert!( + retained.bytes < active.bytes, + "cancelled flights release their allocations" + ); + assert!( + retained.bytes <= first_peak, + "repeated bursts must not accumulate retained bytes" + ); + } +} + +fn request( + client: &Arc, + companion: Jid, +) -> tokio::task::JoinHandle> { + let client = client.clone(); + tokio::spawn(async move { client.ensure_adv_account_identity(&companion).await }) +} + +async fn wait_for_joiner(client: &Client) { + poll_until("another caller to join the ADV flight", || { + client + .adv_bootstrap() + .flights + .lock() + .unwrap() + .values() + .any(|flight| Arc::strong_count(flight) > 2) + }) + .await; +} + +async fn answer(client: &Arc, id: &str, primary: &Jid, account: &KeyPair, valid: bool) { + let mut rng = rand::make_rng::(); + let signed = KeyPair::generate(&mut rng); + let mut signature = account + .private_key + .calculate_signature(&signed.public_key.serialize(), &mut rng) + .unwrap() + .to_vec(); + if !valid { + signature[0] ^= 1; + } + let bundle = PreKeyBundle::new( + 1, + 0u32.into(), + None, + 1u32.into(), + signed.public_key, + signature, + IdentityKey::new(account.public_key), + ) + .unwrap(); + let user = PreKeyBundleUserNode::from_bundle(primary.clone(), &bundle, None) + .unwrap() + .into_node(); + // Exercise the actual IQ correlator, not direct waiter completion. + client + .process_node(node_to_owned_ref( + &NodeBuilder::new("iq") + .attr("id", id) + .attr("type", "result") + .children([NodeBuilder::new("list").children([user]).build()]) + .build(), + )) + .await; +} + +#[tokio::test] +async fn adv_bootstrap_shares_success_and_invalid_signature_without_refetching() { + for valid in [true, false] { + let (client, transport) = create_iq_test_client().await; + let primary = Jid::pn("12025550170"); + let leader = request(&client, primary.with_device(7)); + let iq = decode_sent_iq(&transport, 0).await; + let id = iq.get().attrs().optional_string("id").unwrap().into_owned(); + let joined = request(&client, primary.with_device(9)); + wait_for_joiner(&client).await; + let account = KeyPair::generate(&mut rand::make_rng::()); + answer(&client, &id, &primary, &account, valid).await; + let first = leader.await.unwrap(); + let second = joined.await.unwrap(); + assert_eq!(first.is_ok(), valid); + assert_eq!(second.is_ok(), valid); + if valid { + assert_eq!(first.unwrap(), second.unwrap()); + } + assert_eq!(transport.sent_count(), 1); + let persisted = client + .persistence_manager + .backend() + .load_identity(primary.to_protocol_address().as_str()) + .await + .unwrap(); + assert_eq!(persisted.is_some(), valid); + assert!(client.adv_bootstrap().flights.lock().unwrap().is_empty()); + assert!(client.response_waiters_guard().is_empty()); + client.signal_shutdown_sync(); + } +} + +#[tokio::test(start_paused = true)] +async fn adv_bootstrap_shares_timeout_and_releases_all_admission() { + let (client, transport) = create_iq_test_client().await; + let primary = Jid::pn("12025550171"); + let leader = request(&client, primary.with_device(7)); + let _iq = decode_sent_iq(&transport, 0).await; + let joined = request(&client, primary.with_device(9)); + wait_for_joiner(&client).await; + tokio::time::advance(Duration::from_secs(76)).await; + assert!(leader.await.unwrap().is_err()); + assert!(joined.await.unwrap().is_err()); + assert_eq!(transport.sent_count(), 1); + assert!(client.adv_bootstrap().flights.lock().unwrap().is_empty()); + assert!(client.response_waiters_guard().is_empty()); + assert!( + client + .cached_adv_account_identity(&primary) + .await + .unwrap() + .is_none() + ); + client.signal_shutdown_sync(); +} + +#[tokio::test] +async fn adv_bootstrap_caps_queries_and_cancellation_releases_slots() { + let (client, transport) = create_iq_test_client().await; + let mut tasks = Vec::new(); + for index in 0..10 { + tasks.push(request( + &client, + Jid::pn_device(format!("12025550{index:03}"), 7), + )); + } + poll_until("all cold accounts to enter admission", || { + client.adv_bootstrap().flights.lock().unwrap().len() == 10 + }) + .await; + poll_until("eight ADV queries to reach the transport", || { + transport.sent_count() == 8 + }) + .await; + assert_eq!(client.response_waiters_guard().len(), 8); + for task in &tasks { + task.abort(); + } + for task in tasks { + assert!(task.await.unwrap_err().is_cancelled()); + } + assert!(client.adv_bootstrap().flights.lock().unwrap().is_empty()); + assert!(client.response_waiters_guard().is_empty()); + let permits: Vec<_> = (0..8) + .map(|_| client.adv_bootstrap().queries.try_acquire().unwrap()) + .collect(); + assert!(client.adv_bootstrap().queries.try_acquire().is_none()); + drop(permits); + client.signal_shutdown_sync(); +} + +#[tokio::test] +async fn adv_bootstrap_cancelled_leader_releases_joiner_and_allows_a_new_flight() { + let (client, transport) = create_iq_test_client().await; + let primary = Jid::pn("12025550172"); + let leader = request(&client, primary.with_device(7)); + let _first = decode_sent_iq(&transport, 0).await; + let joined = request(&client, primary.with_device(9)); + wait_for_joiner(&client).await; + leader.abort(); + assert!(leader.await.unwrap_err().is_cancelled()); + assert!(joined.await.unwrap().is_err()); + assert!(client.adv_bootstrap().flights.lock().unwrap().is_empty()); + let retry = request(&client, primary.with_device(7)); + let iq = decode_sent_iq(&transport, 1).await; + let id = iq.get().attrs().optional_string("id").unwrap().into_owned(); + let account = KeyPair::generate(&mut rand::make_rng::()); + answer(&client, &id, &primary, &account, true).await; + assert!(retry.await.unwrap().is_ok()); + client.signal_shutdown_sync(); +} + +#[tokio::test] +async fn adv_bootstrap_rechecks_current_trust_and_connection_before_publication() { + for (retire_connection, fire_shutdown) in [(false, false), (true, false), (false, true)] { + let (client, transport) = create_iq_test_client().await; + let primary = Jid::pn("12025550173"); + let address = primary.to_protocol_address(); + let lock = client.session_lock_for(address.as_str()).await; + let bootstrap = request(&client, primary.with_device(7)); + let iq = decode_sent_iq(&transport, 0).await; + let id = iq.get().attrs().optional_string("id").unwrap().into_owned(); + let guard = lock.lock().await; + let baseline = Arc::strong_count(&lock); + let stale = KeyPair::generate(&mut rand::make_rng::()); + answer(&client, &id, &primary, &stale, true).await; + wait_for_lock_waiter(&lock, baseline).await; + let current = KeyPair::generate(&mut rand::make_rng::()); + if fire_shutdown { + client.notify_connection_shutdown(); + } else if retire_connection { + client.connection_generation.fetch_add(1, Ordering::AcqRel); + } else { + client + .signal_cache + .persist_identity_if_absent_or_equal( + &address, + current.public_key.public_key_bytes(), + &*client.persistence_manager.backend(), + ) + .await + .unwrap(); + } + drop(guard); + let result = bootstrap.await.unwrap(); + let persisted = client + .persistence_manager + .backend() + .load_identity(address.as_str()) + .await + .unwrap(); + if retire_connection || fire_shutdown { + assert!(result.is_err()); + assert!(persisted.is_none()); + } else { + assert_eq!(result.unwrap(), *current.public_key.public_key_bytes()); + assert_eq!(persisted, Some(*current.public_key.public_key_bytes())); + } + assert!(client.adv_bootstrap().flights.lock().unwrap().is_empty()); + client.signal_shutdown_sync(); + } +} + +#[tokio::test] +async fn adv_bootstrap_owned_writer_keeps_primary_fence_after_caller_cancellation() { + use crate::store::release::tests::ProbeBackend; + use crate::test_utils::create_iq_test_client_with_backend; + + for gate_method in ["put_identity", "after:put_identity"] { + let (mut probe, _) = ProbeBackend::new(); + let (release, gate) = async_channel::unbounded(); + let (entered, arrived) = async_channel::unbounded(); + probe.gate = Some(gate); + probe.entered = Some(entered); + probe.gate_method = gate_method; + let (client, transport) = create_iq_test_client_with_backend(Arc::new(probe)).await; + let primary = Jid::pn("12025550174"); + let address = primary.to_protocol_address(); + let lock = client.session_lock_for(address.as_str()).await; + let bootstrap = request(&client, primary.with_device(7)); + let iq = decode_sent_iq(&transport, 0).await; + let id = iq.get().attrs().optional_string("id").unwrap().into_owned(); + let account = KeyPair::generate(&mut rand::make_rng::()); + answer(&client, &id, &primary, &account, true).await; + tokio::time::timeout(Duration::from_secs(5), arrived.recv()) + .await + .unwrap() + .unwrap(); + bootstrap.abort(); + assert!(bootstrap.await.unwrap_err().is_cancelled()); + assert!(client.adv_bootstrap().flights.lock().unwrap().is_empty()); + assert!( + lock.try_lock().is_none(), + "the actual writer still owns the account fence" + ); + assert_eq!(client.inbound_queues().work.pending(), 1); + release.send(()).await.unwrap(); + poll_until("the owned ADV writer to finish", || { + client.inbound_queues().work.pending() == 0 + }) + .await; + assert!(lock.try_lock().is_some()); + assert_eq!( + client + .persistence_manager + .backend() + .load_identity(address.as_str()) + .await + .unwrap(), + Some(*account.public_key.public_key_bytes()) + ); + client.signal_shutdown_sync(); + } +} diff --git a/src/send/mod.rs b/src/send/mod.rs index f9201511a..92c27dff7 100644 --- a/src/send/mod.rs +++ b/src/send/mod.rs @@ -2122,10 +2122,10 @@ impl Client { wacore::send::retain_skdm_distribution_targets(&mut devices, own_sending_jid); } else if let Some(cached_map) = cached_map { devices.retain(|device| { - !device.is_hosted() - && !(device.user == own_sending_jid.user + !(device.is_hosted() + || (device.user == own_sending_jid.user && device.device == own_sending_jid.device) - && !cached_map.device_and_primary_warm(&device.user, device.device) + || cached_map.device_and_primary_warm(&device.user, device.device)) }); } log::debug!( diff --git a/wacore/src/send/group.rs b/wacore/src/send/group.rs index 01233f762..0f69aeb1c 100644 --- a/wacore/src/send/group.rs +++ b/wacore/src/send/group.rs @@ -8,8 +8,8 @@ use anyhow::Context as _; /// by both normal and targeted sends. pub fn retain_skdm_distribution_targets(devices: &mut Vec, own_sending_jid: &Jid) { devices.retain(|device| { - !(device.user == own_sending_jid.user && device.device == own_sending_jid.device) - && !device.is_hosted() + !((device.user == own_sending_jid.user && device.device == own_sending_jid.device) + || device.is_hosted()) }); } From 28122a45c0cd4cbe9f22874c386c4161b19b8f18 Mon Sep 17 00:00:00 2001 From: Assis Date: Tue, 10 Nov 2026 03:13:15 -0300 Subject: [PATCH 2/8] fix: resolve hosted ADV account anchors Resolve ADV trust through the ordinary account family without merging hosted Signal device identities. Preserve paired-self protection and account-before-device lock ordering across send, receive and recovery. Linux Rust 1.95 validation: fmt, scoped Clippy -D warnings, 1,120 focused tests passed with two existing ignored cases. Client layout is unchanged. --- agent_docs/adv_account_identity.md | 46 ++ src/client/adapters.rs | 2 +- src/client/lid_pn.rs | 4 +- src/message/hosted_adv_tests.rs | 505 ++++++++++++++++++ src/message/receive.rs | 4 + src/prekeys.rs | 10 +- src/prekeys/adv_bootstrap_tests.rs | 172 ++++++ src/retry.rs | 5 +- src/send/mod.rs | 2 +- src/test_utils.rs | 2 +- src/usync.rs | 76 ++- .../src/emit/group_create_conformance.rs | 11 +- wacore/src/prekeys.rs | 79 ++- .../src/stanza/groups/create_conformance.rs | 19 +- wacore/src/types/jid.rs | 55 +- 15 files changed, 958 insertions(+), 34 deletions(-) create mode 100644 agent_docs/adv_account_identity.md create mode 100644 src/message/hosted_adv_tests.rs diff --git a/agent_docs/adv_account_identity.md b/agent_docs/adv_account_identity.md new file mode 100644 index 000000000..647b1f5bd --- /dev/null +++ b/agent_docs/adv_account_identity.md @@ -0,0 +1,46 @@ +# ADV account anchors and hosted devices + +The ADV account anchor and the Signal device address are different identities. +For a `user:99@hosted` device the anchor is `user@s.whatsapp.net`; for +`user:99@hosted.lid` it is `user@lid`. The device session/identity retains its +original hosted namespace. `adv_account_jid` is restricted to selecting the +anchor; do not replace general JID normalization or Signal address rendering. + +Static reference: whatspec commit +`ae3880925bacd23863cc30ec480c9f82bfd869f1`, schema `4.5.0`, WA +`2.3000.1052559453`, 676 bundles, set hash +`d6ab7ae601eda281cfac570a68de1d787adde8f4d778f0bb994416f661a21abf`. +The `bundle-store` asset `bundles-2.3000.1052559453-.tar.xz` has +SHA256 `2958d99a3d84141fa8479f1f991e06246e68c063ca8f618a8d5d361287387f62`. +The archive and all 676 bundle hashes/sizes were checked against the pinned +lock. Complete modules were extracted using an AST, without executing them. + +Relevant original byte offsets (start inclusive, end exclusive): + +| Bundle | Module | Offsets | +| --- | --- | --- | +| `0RwvWNH8_qZ.js` | `WAWebAdvSignatureApi` | 767477–775940 | +| `0RwvWNH8_qZ.js` | `WAWebHandleAdvDeviceNotificationUtils` | 731532–733850 | +| `331d6f8883e8cd73ff267508b8e5889920b0219c7fd88325ee56ee22eefdcb00.js` | `WAWebWidFactory` | 545429–549200 | +| same | `WAWebWid` | 535329–545312 | +| same | `WAWebSignalAddress` | 3107699–3108968 | + +`validateADVwithIdentityKey` reads the original device identity separately +from the anchor selected with `asUserWidOrThrow`. That helper maps Hosted to +the ordinary PN family and HostedLid to LID. SignalAddress preserves hosted +device addressing; Wid rejects a hosted primary with device 0. This is static +Web evidence, not a claim about the server's response to malformed IQs. + +Account fences must select the same anchor as verification. Multi-lock +acquisition groups by account family, then user/device, retaining the original +server as a tie-breaker. Plain lexical server ordering would acquire a hosted +LID device before its ordinary LID anchor, opposite to inbound processing. +The comparator is only for Signal locks; phash and device-list ordering retain +their existing rules. + +The paired own-account fallback remains trusted only for an exact account +family/user match. Embedded certificate keys cannot replace an existing +anchor, and both signatures still use the original device's hosted semantics. +Cold bootstrap retains bounded singleflight, generation cancellation and owned +durable publication. A hosted USync list follows its existing separately +authenticated policy and cannot replace our paired primary identity. diff --git a/src/client/adapters.rs b/src/client/adapters.rs index 7e760c38a..657207773 100644 --- a/src/client/adapters.rs +++ b/src/client/adapters.rs @@ -68,7 +68,7 @@ impl Client { if jid.device == 0 { return None; } - let primary = jid.with_device(0).to_protocol_address(); + let primary = wacore::types::jid::adv_account_jid(jid).to_protocol_address(); Some( self.session_lock_for(primary.as_str()) .await diff --git a/src/client/lid_pn.rs b/src/client/lid_pn.rs index 0079d05ce..6a14b9801 100644 --- a/src/client/lid_pn.rs +++ b/src/client/lid_pn.rs @@ -1063,8 +1063,8 @@ impl Client { // must not publish new trust while this iteration retains old // session material. Use the send path's order for the whole set. let mut keys = vec![ - from.with_device(0), - to.with_device(0), + wacore::types::jid::adv_account_jid(from), + wacore::types::jid::adv_account_jid(to), pn_jid.clone(), lid_jid.clone(), ]; diff --git a/src/message/hosted_adv_tests.rs b/src/message/hosted_adv_tests.rs new file mode 100644 index 000000000..9c5a53d96 --- /dev/null +++ b/src/message/hosted_adv_tests.rs @@ -0,0 +1,505 @@ +use super::*; +use std::time::Duration; +use wacore::libsignal::protocol::{IdentityKey, KeyPair}; +use wacore::store::commands::DeviceCommand; +use wacore::types::jid::adv_account_jid; + +fn fixture( + account: &KeyPair, + hosted: bool, + correct_prefix: bool, +) -> (Vec, Vec, [u8; 32]) { + let device = KeyPair::generate(&mut rand::make_rng::()); + let identity = IdentityKey::new(device.public_key); + let inner = SignalMessage::new( + 3, + &[0; 32], + device.public_key, + 0, + 0, + &[0; 16], + &identity, + &identity, + ) + .unwrap(); + let message = + PreKeySignalMessage::new(3, 1234, None, 1.into(), device.public_key, identity, inner) + .unwrap(); + let mut details = waproto::whatsapp::ADVDeviceIdentity::default(); + details.key_index = Some(7); + details.device_type = Some(if hosted { + waproto::whatsapp::ADVEncryptionType::HOSTED + } else { + waproto::whatsapp::ADVEncryptionType::E2EE + }); + let certificate = waproto::codec::adv_signed_device_identity_to_vec( + &wacore::adv::test_util::signed_identity( + account, + &device, + &waproto::codec::adv_device_identity_to_vec(&details), + if hosted { &[6, 5] } else { &[6, 0] }, + Some(if hosted == correct_prefix { + &[6, 6] + } else { + &[6, 1] + }), + true, + ), + ); + ( + vec![EncPayload { + ciphertext: message.serialize().into(), + enc_type: EncType::PreKeyMessage, + padding_version: 2, + enc_index: 0, + state: None, + session_type: None, + }], + certificate, + *device.public_key.public_key_bytes(), + ) +} + +async fn pair(client: &Client, own: &Jid) { + let command = if own.server == wacore_binary::Server::Lid { + DeviceCommand::SetLid(Some(own.with_device(1))) + } else { + DeviceCommand::SetId(Some(own.with_device(1))) + }; + client.persistence_manager.process_command(command).await; + let mut account = waproto::whatsapp::ADVSignedDeviceIdentity::default(); + account.account_signature_key = Some( + crate::test_utils::adv_test_account() + .public_key + .public_key_bytes() + .to_vec(), + ); + client + .persistence_manager + .process_command(DeviceCommand::SetAccount(Some(account))) + .await; +} + +#[tokio::test] +async fn hosted_own_adv_uses_paired_anchor_without_iq_or_signal_namespace_changes() { + for own in [Jid::pn("12025550181"), Jid::lid("100000000000181")] { + let (client, transport) = crate::test_utils::create_iq_test_client().await; + pair(&client, &own).await; + for hosted in [false, true] { + let sender = own.with_device_hosting(if hosted { 99 } else { 7 }, hosted); + let original = sender.to_protocol_address(); + let (payloads, certificate, identity) = + fixture(crate::test_utils::adv_test_account(), hosted, true); + assert_eq!( + client + .preflight_incoming_companion_identity(&sender, Some(&certificate), &payloads) + .await, + None + ); + let fence = client + .lock_incoming_companion_account(&sender, Some(&certificate), &payloads) + .await + .unwrap(); + assert_eq!(sender.to_protocol_address(), original); + let anchor_lock = client + .session_lock_for(own.to_protocol_address().as_str()) + .await; + assert!(anchor_lock.try_lock().is_none()); + drop(fence); + assert_eq!( + transport.sent_count(), + 0, + "paired trust requires neither IQ nor ACK" + ); + let backend = client.persistence_manager.backend(); + assert!( + backend + .load_identity(sender.with_device(0).to_protocol_address().as_str()) + .await + .unwrap() + .is_none() + ); + assert!( + client + .signal_cache + .get_identity(&original, &*backend) + .await + .unwrap() + .is_none() + ); + // A known device remains keyed by its original Signal namespace. + client.signal_cache.put_identity(&original, &identity).await; + assert!( + client + .validate_incoming_companion_identity(&sender, None, &identity) + .await + .is_ok() + ); + if hosted { + assert!( + client + .signal_cache + .get_identity(&own.with_device(99).to_protocol_address(), &*backend) + .await + .unwrap() + .is_none() + ); + } + } + client.signal_shutdown_sync(); + } +} + +#[tokio::test] +async fn hosted_and_ordinary_adv_reject_wrong_keys_signatures_and_prefixes() { + for own in [Jid::pn("12025550182"), Jid::lid("100000000000182")] { + let (client, transport) = crate::test_utils::create_iq_test_client().await; + pair(&client, &own).await; + for hosted in [false, true] { + let sender = own.with_device_hosting(if hosted { 99 } else { 7 }, hosted); + let foreign = KeyPair::generate(&mut rand::make_rng::()); + let (_, certificate, identity) = fixture(&foreign, hosted, true); + assert!( + client + .validate_incoming_companion_identity(&sender, Some(&certificate), &identity) + .await + .is_err() + ); + let (_, certificate, identity) = + fixture(crate::test_utils::adv_test_account(), hosted, false); + assert!( + client + .validate_incoming_companion_identity(&sender, Some(&certificate), &identity) + .await + .is_err() + ); + let (_, certificate, identity) = + fixture(crate::test_utils::adv_test_account(), hosted, true); + for account_signature in [false, true] { + let mut corrupted = + waproto::codec::adv_signed_device_identity_decode(&certificate).unwrap(); + let signature = if account_signature { + &mut corrupted.account_signature + } else { + &mut corrupted.device_signature + }; + signature.as_mut().unwrap()[0] ^= 1; + assert!( + client + .validate_incoming_companion_identity( + &sender, + Some(&waproto::codec::adv_signed_device_identity_to_vec( + &corrupted + )), + &identity + ) + .await + .is_err() + ); + } + assert!( + client + .signal_cache + .get_identity( + &sender.to_protocol_address(), + &*client.persistence_manager.backend() + ) + .await + .unwrap() + .is_none() + ); + } + assert_eq!( + transport.sent_count(), + 0, + "rejected certificates cannot send ACKs or fetch replacement trust" + ); + client.signal_shutdown_sync(); + } +} + +#[tokio::test] +async fn hosted_adv_does_not_borrow_paired_trust_from_the_other_addressing_family() { + for own in [Jid::pn("12025550183"), Jid::lid("12025550183")] { + let (client, transport) = crate::test_utils::create_iq_test_client().await; + pair(&client, &own).await; + let other = if own.server == wacore_binary::Server::Pn { + Jid::lid(own.user.clone()) + } else { + Jid::pn(own.user.clone()) + }; + let sender = other.with_device_hosting(99, true); + assert!( + client + .cached_adv_account_identity(&sender) + .await + .unwrap() + .is_none() + ); + let (_, certificate, identity) = fixture(crate::test_utils::adv_test_account(), true, true); + let operation = { + let client = client.clone(); + tokio::spawn(async move { + client + .validate_incoming_companion_identity(&sender, Some(&certificate), &identity) + .await + }) + }; + let iq = crate::test_utils::decode_sent_iq(&transport, 0).await; + let node = iq.get(); + let requested = node + .get_optional_child("key") + .unwrap() + .get_optional_child("user") + .unwrap() + .attrs() + .optional_jid("jid") + .unwrap(); + assert_eq!( + requested, other, + "the untrusted family must establish its own ordinary anchor" + ); + let id = node.attrs().optional_string("id").unwrap(); + let response = wacore_binary::builder::NodeBuilder::new("iq") + .attr("type", "result") + .children([wacore_binary::builder::NodeBuilder::new("list").build()]) + .build(); + crate::test_utils::answer_iq(&client, &id, &response).await; + assert!(operation.await.unwrap().is_err()); + assert_eq!( + transport.sent_count(), + 1, + "no ACK for an unverified message" + ); + client.signal_shutdown_sync(); + } +} + +#[tokio::test] +async fn hosted_receive_rechecks_the_ordinary_anchor_after_rotation() { + for own in [Jid::pn("12025550184"), Jid::lid("100000000000184")] { + let (client, transport) = crate::test_utils::create_iq_test_client().await; + pair(&client, &own).await; + let sender = own.with_device_hosting(99, true); + let (payloads, certificate, identity) = + fixture(crate::test_utils::adv_test_account(), true, true); + client + .validate_incoming_companion_identity(&sender, Some(&certificate), &identity) + .await + .unwrap(); + let lock = client + .session_lock_for(adv_account_jid(&sender).to_protocol_address().as_str()) + .await; + let recovery = lock.lock_arc().await; + let baseline = Arc::strong_count(&lock); + let receive = { + let client = client.clone(); + let sender = sender.clone(); + tokio::spawn(async move { + client + .lock_incoming_companion_account(&sender, Some(&certificate), &payloads) + .await + .map(|_| ()) + }) + }; + crate::test_utils::wait_for_lock_waiter(&lock, baseline).await; + let replacement = KeyPair::generate(&mut rand::make_rng::()); + client + .signal_cache + .put_identity( + &own.to_protocol_address(), + replacement.public_key.public_key_bytes(), + ) + .await; + drop(recovery); + assert!( + tokio::time::timeout(Duration::from_secs(5), receive) + .await + .unwrap() + .unwrap() + .is_err() + ); + assert_eq!( + client.cached_adv_account_identity(&sender).await.unwrap(), + Some(*replacement.public_key.public_key_bytes()) + ); + assert!( + client + .signal_cache + .get_identity( + &sender.to_protocol_address(), + &*client.persistence_manager.backend() + ) + .await + .unwrap() + .is_none() + ); + assert_eq!(transport.sent_count(), 0); + client.signal_shutdown_sync(); + } +} + +#[tokio::test] +async fn hosted_lid_participant_send_and_receive_use_one_account_before_device_lock_order() { + use crate::store::release::tests::ProbeBackend; + use wacore::proto_helpers::MessageBuilderExt; + use wacore::store::traits::{DeviceInfo, DeviceListRecord}; + let peer = Jid::lid("100000000000187"); + let hosted = peer.with_device_hosting(99, true); + let own = Jid::pn("12025550187"); + let (mut probe, _) = ProbeBackend::new(); + // Persist a real Signal session before constructing the observed client, + // leaving its cache cold so the send crosses the backend admission gate. + let (scratch, _) = + crate::test_utils::create_iq_test_client_with_backend(probe.backend.clone()).await; + crate::test_utils::seed_peer_session(&scratch, &hosted).await; + crate::test_utils::seed_adv_primary(&scratch, &hosted).await; + scratch + .signal_cache + .flush(probe.backend.as_ref()) + .await + .unwrap(); + probe + .backend + .save(&scratch.persistence_manager.get_device_snapshot()) + .await + .unwrap(); + scratch.signal_shutdown_sync(); + + let (release, gate) = async_channel::unbounded(); + let (entered, arrived) = async_channel::unbounded(); + probe.gate = Some(gate); + probe.entered = Some(entered); + probe.gate_method = "after:get_session"; + let (client, transport) = + crate::test_utils::create_iq_test_client_with_backend(Arc::new(probe)).await; + client + .persistence_manager + .process_command(DeviceCommand::SetId(Some(own.clone()))) + .await; + for (user, devices) in [ + (own.user.as_str(), vec![DeviceInfo::new(0, None)]), + ( + peer.user.as_str(), + vec![DeviceInfo::new(99, Some(7)).with_hosting(true)], + ), + ] { + client + .update_device_list(DeviceListRecord { + user: user.into(), + devices: devices.into_boxed_slice(), + timestamp: wacore::time::now_secs(), + phash: None, + raw_id: None, + }) + .await + .unwrap(); + } + let (payloads, certificate, identity) = + fixture(crate::test_utils::adv_test_account(), true, true); + client + .validate_incoming_companion_identity(&hosted, Some(&certificate), &identity) + .await + .unwrap(); + let fence = client + .lock_incoming_companion_account(&hosted, Some(&certificate), &payloads) + .await + .unwrap(); + let anchor = client + .session_lock_for(peer.to_protocol_address().as_str()) + .await; + let device = client + .session_lock_for(hosted.to_protocol_address().as_str()) + .await; + let baseline = Arc::strong_count(&anchor); + let mut send = { + let client = client.clone(); + let peer = peer.clone(); + let hosted = hosted.clone(); + tokio::spawn(async move { + // Normal DM fanout deliberately excludes hosted devices. This + // public pairwise builder retains them and uses the production + // send lock ordering, real Signal encryption and pre-wire commit. + let message = waproto::whatsapp::Message::text("synthetic hosted lock ordering"); + let (nodes, _) = client + .signal() + .create_participant_nodes(std::slice::from_ref(&peer), &message) + .await?; + assert_eq!(nodes.len(), 1); + assert_eq!( + nodes[0].as_node_ref().attrs().optional_jid("jid"), + Some(hosted) + ); + client + .send_node( + wacore_binary::builder::NodeBuilder::new("message") + .attr("to", peer) + .attr("id", "SYNTHETIC_HOSTED_PARTICIPANT_SEND") + .attr("type", "text") + .children([wacore_binary::builder::NodeBuilder::new("participants") + .children(nodes) + .build()]) + .build(), + ) + .await?; + Ok::<(), anyhow::Error>(()) + }) + }; + tokio::select! { + result = &mut send => panic!("participant send finished before its backend gate: {result:?}"), + arrived = tokio::time::timeout(Duration::from_secs(5), arrived.recv()) => { + arrived.expect("participant send reaches the cold session read").unwrap(); + }, + } + release.send(()).await.unwrap(); + crate::test_utils::wait_for_lock_waiter(&anchor, baseline).await; + assert!( + device.try_lock().is_some(), + "send must wait on the ordinary LID account before taking the hosted device" + ); + let info = Arc::new(MessageInfo { + id: "SYNTHETIC_HOSTED_LOCK_ORDER".into(), + source: crate::types::message::MessageSource { + chat: peer.clone(), + sender: hosted.clone(), + ..Default::default() + }, + ..Default::default() + }); + // Real receive decryption owns the acquired account fence. The fixture's + // absent signed prekey fails safely; it must still release both locks. + tokio::time::timeout( + Duration::from_secs(5), + client.clone().decrypt_session_enc_batch_with_origin( + payloads, + &info, + &hosted, + crate::types::events::DecryptFailMode::Show, + None, + (raw_pending::StanzaOrigin::new(), fence), + ), + ) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), send) + .await + .unwrap() + .unwrap() + .unwrap(); + assert!( + transport.sent_count() > 0, + "production send reaches the Noise transport" + ); + assert!(anchor.try_lock().is_some()); + assert!(device.try_lock().is_some()); + assert!( + client + .persistence_manager + .backend() + .load_identity(hosted.with_device(0).to_protocol_address().as_str()) + .await + .unwrap() + .is_none() + ); + client.signal_shutdown_sync(); +} diff --git a/src/message/receive.rs b/src/message/receive.rs index e93b0a9d1..2bec724ad 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -3174,6 +3174,10 @@ mod tests { } } +#[cfg(test)] +#[path = "hosted_adv_tests.rs"] +mod hosted_adv_tests; + #[cfg(test)] mod adv_tests { use super::*; diff --git a/src/prekeys.rs b/src/prekeys.rs index dda1c8771..39ce697c2 100644 --- a/src/prekeys.rs +++ b/src/prekeys.rs @@ -292,7 +292,7 @@ impl Client { /// Read a fresh, signed primary key without publishing a trust change. pub(crate) async fn fetch_adv_account_identity(&self, jid: &Jid) -> anyhow::Result<[u8; 32]> { - let primary = jid.with_device(0); + let primary = wacore::types::jid::adv_account_jid(jid); self.query_adv_account_identity(&primary, Some(PreKeyFetchReason::Identity)) .await } @@ -339,7 +339,7 @@ impl Client { let mut requested = jids.to_vec(); let mut bootstrap_primaries = Vec::new(); for jid in &cold_companions { - let primary = jid.with_device(0); + let primary = wacore::types::jid::adv_account_jid(jid); if !bootstrap_primaries.contains(&primary) { bootstrap_primaries.push(primary.clone()); } @@ -365,7 +365,7 @@ impl Client { // primary trust anchor. An omitted primary must not leave a session // authenticated only by a sender-selected account key. for companion in &cold_companions { - let primary = companion.with_device(0); + let primary = wacore::types::jid::adv_account_jid(companion); if let Some(bundle) = outcome.bundles.get(&primary) { authenticated_primary_identity(bundle)?; } else { @@ -417,7 +417,7 @@ impl Client { jid: &Jid, ) -> anyhow::Result> { use wacore::types::jid::JidExt; - let primary = jid.with_device(0); + let primary = wacore::types::jid::adv_account_jid(jid); let backend = self.persistence_manager.backend(); if let Some(identity) = self .signal_cache @@ -453,7 +453,7 @@ impl Client { use wacore::types::jid::JidExt; let generation = self.connection_generation.load(Ordering::Acquire); let shutdown = self.connection_shutdown_signal(); - let primary = jid.with_device(0); + let primary = wacore::types::jid::adv_account_jid(jid); let bootstrap = async { ensure_adv_bootstrap_scope(&self.connection_generation, generation, &shutdown)?; { diff --git a/src/prekeys/adv_bootstrap_tests.rs b/src/prekeys/adv_bootstrap_tests.rs index 5d7b59eb4..399fba82c 100644 --- a/src/prekeys/adv_bootstrap_tests.rs +++ b/src/prekeys/adv_bootstrap_tests.rs @@ -58,6 +58,178 @@ fn request( tokio::spawn(async move { client.ensure_adv_account_identity(&companion).await }) } +#[tokio::test] +async fn hosted_and_ordinary_companions_share_one_canonical_cold_bootstrap() { + for primary in [Jid::pn("12025550185"), Jid::lid("100000000000185")] { + let (client, transport) = create_iq_test_client().await; + let hosted = primary.with_device_hosting(99, true); + let leader = request(&client, hosted.clone()); + let iq = decode_sent_iq(&transport, 0).await; + let node = iq.get(); + let requested = node + .get_optional_child("key") + .unwrap() + .get_optional_child("user") + .unwrap() + .attrs() + .optional_jid("jid") + .unwrap(); + assert_eq!(requested, primary); + let id = node.attrs().optional_string("id").unwrap().into_owned(); + let joined = request(&client, primary.with_device(7)); + wait_for_joiner(&client).await; + let account = KeyPair::generate(&mut rand::make_rng::()); + answer(&client, &id, &primary, &account, true).await; + assert_eq!( + leader.await.unwrap().unwrap(), + *account.public_key.public_key_bytes() + ); + assert_eq!( + joined.await.unwrap().unwrap(), + *account.public_key.public_key_bytes() + ); + assert_eq!(transport.sent_count(), 1); + let backend = client.persistence_manager.backend(); + assert_eq!( + backend + .load_identity(primary.to_protocol_address().as_str()) + .await + .unwrap(), + Some(*account.public_key.public_key_bytes()) + ); + assert!( + backend + .load_identity(hosted.with_device(0).to_protocol_address().as_str()) + .await + .unwrap() + .is_none() + ); + assert!(client.adv_bootstrap().flights.lock().unwrap().is_empty()); + client.signal_shutdown_sync(); + } +} + +#[tokio::test] +async fn cold_hosted_prekey_fetch_authenticates_and_persists_only_the_ordinary_anchor() { + for primary in [Jid::pn("12025550189"), Jid::lid("100000000000189")] { + let (client, transport) = create_iq_test_client().await; + let hosted = primary.with_device_hosting(99, true); + let ordinary = primary.with_device(7); + let requested = vec![hosted.clone(), ordinary.clone()]; + let fetch = { + let client = client.clone(); + let requested = requested.clone(); + tokio::spawn(async move { client.fetch_pre_keys(&requested, None).await }) + }; + let iq = decode_sent_iq(&transport, 0).await; + let node = iq.get(); + let users: Vec<_> = node + .get_optional_child("key") + .unwrap() + .children() + .unwrap() + .iter() + .map(|user| user.attrs().optional_jid("jid").unwrap()) + .collect(); + assert_eq!( + users, + vec![hosted.clone(), ordinary.clone(), primary.clone()] + ); + let id = node.attrs().optional_string("id").unwrap().into_owned(); + let account = KeyPair::generate(&mut rand::make_rng::()); + let make_bundle = |key: &KeyPair| { + let mut rng = rand::make_rng::(); + let signed = KeyPair::generate(&mut rng); + PreKeyBundle::new( + 1, + 0u32.into(), + None, + 1u32.into(), + signed.public_key, + key.private_key + .calculate_signature(&signed.public_key.serialize(), &mut rng) + .unwrap() + .to_vec(), + IdentityKey::new(key.public_key), + ) + .unwrap() + }; + let mut replies = Vec::new(); + for jid in &requested { + let device = KeyPair::generate(&mut rand::make_rng::()); + let certificate = waproto::codec::adv_signed_device_identity_to_vec( + &wacore::adv::test_util::signed_identity( + &account, + &device, + b"\x18\x07", + &[6, 0], + Some(if jid == &hosted { &[6, 6] } else { &[6, 1] }), + false, + ), + ); + replies.push( + PreKeyBundleUserNode::from_bundle( + jid.clone(), + &make_bundle(&device), + Some(certificate), + ) + .unwrap() + .into_node(), + ); + } + replies.push( + PreKeyBundleUserNode::from_bundle(primary.clone(), &make_bundle(&account), None) + .unwrap() + .into_node(), + ); + client + .process_node(node_to_owned_ref( + &NodeBuilder::new("iq") + .attr("id", id) + .attr("type", "result") + .children([NodeBuilder::new("list").children(replies).build()]) + .build(), + )) + .await; + let result = fetch.await.unwrap().unwrap(); + assert_eq!( + result.bundles.len(), + 2, + "bootstrap primary is not an unsolicited result" + ); + for jid in requested { + assert_eq!( + result + .bundles + .get(&jid) + .unwrap() + .account_identity() + .unwrap() + .public_key() + .public_key_bytes(), + account.public_key.public_key_bytes() + ); + } + let backend = client.persistence_manager.backend(); + assert_eq!( + backend + .load_identity(primary.to_protocol_address().as_str()) + .await + .unwrap(), + Some(*account.public_key.public_key_bytes()) + ); + assert!( + backend + .load_identity(hosted.with_device(0).to_protocol_address().as_str()) + .await + .unwrap() + .is_none() + ); + assert_eq!(transport.sent_count(), 1); + client.signal_shutdown_sync(); + } +} + async fn wait_for_joiner(client: &Client) { poll_until("another caller to join the ADV flight", || { client diff --git a/src/retry.rs b/src/retry.rs index 783f37a7d..30c8596c9 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -4968,7 +4968,10 @@ mod tests { .await; client .signal_cache - .put_identity(&jid.with_device(0).to_protocol_address(), account_key) + .put_identity( + &wacore::types::jid::adv_account_jid(&jid).to_protocol_address(), + account_key, + ) .await; for valid in [false, true] { let dev_prefix: &[u8; 2] = if hosted_device == valid { diff --git a/src/send/mod.rs b/src/send/mod.rs index 92c27dff7..4f6ae2b2b 100644 --- a/src/send/mod.rs +++ b/src/send/mod.rs @@ -370,7 +370,7 @@ fn sort_session_lock_keys(keys: &mut Vec) { let primaries: Vec<_> = keys .iter() .filter(|jid| jid.device != 0) - .map(|jid| jid.with_device(0)) + .map(wacore::types::jid::adv_account_jid) .collect(); keys.extend(primaries); keys.sort_unstable_by(wacore::types::jid::cmp_for_lock_order); diff --git a/src/test_utils.rs b/src/test_utils.rs index d96cc7d5b..7ad68d86b 100644 --- a/src/test_utils.rs +++ b/src/test_utils.rs @@ -105,7 +105,7 @@ pub(crate) async fn seed_adv_primary(client: &Client, user: &Jid) { client .signal_cache .put_identity( - &user.with_device(0).to_protocol_address(), + &wacore::types::jid::adv_account_jid(user).to_protocol_address(), adv_test_account().public_key.public_key_bytes(), ) .await; diff --git a/src/usync.rs b/src/usync.rs index 5f6d43148..43539fc65 100644 --- a/src/usync.rs +++ b/src/usync.rs @@ -390,7 +390,7 @@ impl Client { .pn .iter() .chain(snapshot.lid.iter()) - .any(|own| own.is_same_chat_as(user)) + .any(|own| own.is_same_chat_as(&wacore::types::jid::adv_account_jid(user))) { return Err(error); } @@ -685,7 +685,7 @@ impl Client { let mut addresses: Vec<_> = recovered .previous_identities .iter() - .map(|(jid, _)| jid.with_device(0)) + .map(|(jid, _)| wacore::types::jid::adv_account_jid(jid)) .collect(); // Match PN/LID migration before taking mapping/registry locks. addresses.sort_unstable_by(wacore::types::jid::cmp_for_lock_order); @@ -1022,11 +1022,9 @@ impl Client { let identity_update = if let Some(identity) = verified_identity { let previous_identity = self.cached_adv_account_identity(&user_list.user).await?; let snapshot = self.persistence_manager.get_device_snapshot(); - let is_self = snapshot - .pn - .iter() - .chain(snapshot.lid.iter()) - .any(|own| own.is_same_chat_as(&user_list.user)); + let is_self = snapshot.pn.iter().chain(snapshot.lid.iter()).any(|own| { + own.is_same_chat_as(&wacore::types::jid::adv_account_jid(&user_list.user)) + }); if is_self && previous_identity != Some(identity) && mode == DeviceListApplyMode::GroupRefresh @@ -1040,7 +1038,7 @@ impl Client { let mut identities = Vec::with_capacity(2); use wacore::types::jid::JidExt as _; identities.push(( - user_list.user.with_device(0).to_protocol_address(), + wacore::types::jid::adv_account_jid(&user_list.user).to_protocol_address(), previous_identity, )); if !is_self { @@ -2140,6 +2138,68 @@ mod tests { client.signal_shutdown_sync(); } + /// The hosted wire family must not hide ownership of our account anchor. + #[tokio::test] + async fn hosted_addressed_usync_cannot_replace_our_paired_account_anchor() { + use wacore::store::commands::DeviceCommand; + use wacore::usync::{UserDeviceList, UsyncDevice}; + for own in [Jid::pn("12025550188"), Jid::lid("100000000000188")] { + let client = create_test_client().await; + let hosted = own.with_device_hosting(99, true); + let response = authenticated_response_fixture( + &client, + DeviceListResponse { + device_lists: vec![UserDeviceList { + user: hosted.to_non_ad(), + devices: vec![UsyncDevice::new(99, Some(1)).with_hosting(true)], + phash: None, + key_index_bytes: Some(signed_key_index_bytes(vec![1], 1)), + }], + lid_mappings: vec![], + }, + ) + .await; + client + .signal_cache + .delete_identity(&own.to_protocol_address()) + .await; + let paired = wacore::libsignal::protocol::KeyPair::generate(&mut rand::make_rng::< + rand::rngs::StdRng, + >()); + let mut account = waproto::whatsapp::ADVSignedDeviceIdentity::default(); + account.account_signature_key = Some(paired.public_key.public_key_bytes().to_vec()); + let command = if own.server == wacore_binary::Server::Lid { + DeviceCommand::SetLid(Some(own.with_device(1))) + } else { + DeviceCommand::SetId(Some(own.with_device(1))) + }; + client.persistence_manager.process_command(command).await; + client + .persistence_manager + .process_command(DeviceCommand::SetAccount(Some(account))) + .await; + let error = client + .process_device_list_response(&response, Freshness::Refresh) + .await + .unwrap_err(); + assert!(error.to_string().contains("paired primary identity")); + assert_eq!( + client.cached_adv_account_identity(&hosted).await.unwrap(), + Some(*paired.public_key.public_key_bytes()) + ); + assert!( + client + .persistence_manager + .backend() + .load_identity(hosted.with_device(0).to_protocol_address().as_str()) + .await + .unwrap() + .is_none() + ); + client.signal_shutdown_sync(); + } + } + /// A later invalid user must not commit an earlier hosted identity update. #[tokio::test] async fn rejected_usync_batch_preserves_hosted_trust_and_sessions() { diff --git a/tools/whatspec-codegen/src/emit/group_create_conformance.rs b/tools/whatspec-codegen/src/emit/group_create_conformance.rs index ed14953be..f53a17f75 100644 --- a/tools/whatspec-codegen/src/emit/group_create_conformance.rs +++ b/tools/whatspec-codegen/src/emit/group_create_conformance.rs @@ -92,7 +92,16 @@ fn optional_bool(value: &Value) -> Result { fn compare(out: &mut String, actual: &str, expected: &str, label: &str) { match expected { "true" => out.push_str(&format!("assert!({actual},{});\n", rust_str(label))), - "false" => out.push_str(&format!("assert!(!({actual}),{});\n", rust_str(label))), + "false" => { + if let Some(option) = actual.strip_suffix(".is_some()") { + out.push_str(&format!( + "assert!({option}.is_none(),{});\n", + rust_str(label) + )); + } else { + out.push_str(&format!("assert!(!({actual}),{});\n", rust_str(label))); + } + } _ => out.push_str(&format!( "assert_eq!({actual},{expected},{});\n", rust_str(label) diff --git a/wacore/src/prekeys.rs b/wacore/src/prekeys.rs index 628b8bd97..3b6bf51f0 100644 --- a/wacore/src/prekeys.rs +++ b/wacore/src/prekeys.rs @@ -291,7 +291,7 @@ impl PreKeyUtils { } let account_identity = account_identities .get(&jid) - .or_else(|| primary_identities.get(&jid.with_device(0))); + .or_else(|| primary_identities.get(&crate::types::jid::adv_account_jid(&jid))); let bundle = match Self::node_to_pre_key_bundle_ref(&jid, user_node_ref, account_identity) { Ok(b) => b, @@ -779,6 +779,83 @@ mod tests { .bundles } + #[test] + fn hosted_prekey_bootstrap_uses_the_ordinary_primary_in_the_same_response() { + let mut rng = rand::make_rng::(); + let account = KeyPair::generate(&mut rng); + let device = KeyPair::generate(&mut rng); + for primary in [Jid::pn("12025550186"), Jid::lid("100000000000186")] { + for hosted in [false, true] { + let companion = primary.with_device_hosting(if hosted { 99 } else { 7 }, hosted); + let make_bundle = |key: &KeyPair| { + let mut rng = rand::make_rng::(); + let signed = KeyPair::generate(&mut rng); + PreKeyBundle::new( + 1, + 0u32.into(), + None, + 1u32.into(), + signed.public_key, + key.private_key + .calculate_signature(&signed.public_key.serialize(), &mut rng) + .unwrap() + .to_vec(), + IdentityKey::new(key.public_key), + ) + .unwrap() + }; + let certificate = waproto::codec::adv_signed_device_identity_to_vec( + &crate::adv::test_util::signed_identity( + &account, + &device, + b"\x18\x07", + &[6, 0], + Some(if hosted { &[6, 6] } else { &[6, 1] }), + false, + ), + ); + let device_node = PreKeyBundleUserNode::from_bundle( + companion.clone(), + &make_bundle(&device), + Some(certificate), + ) + .unwrap() + .into_node(); + let primary_node = PreKeyBundleUserNode::from_bundle( + primary.clone(), + &make_bundle(&account), + None, + ) + .unwrap() + .into_node(); + let response = NodeBuilder::new("iq") + .children([NodeBuilder::new("list") + .children([device_node, primary_node]) + .build()]) + .build(); + let outcome = + PreKeyUtils::parse_prekeys_response(&response.as_node_ref(), &HashMap::new()) + .unwrap(); + let bundle = outcome + .bundles + .get(&companion) + .expect("companion certified by ordinary primary"); + assert_eq!( + bundle + .account_identity() + .unwrap() + .public_key() + .public_key_bytes(), + account.public_key.public_key_bytes() + ); + assert!(outcome.bundles.contains_key(&primary)); + if hosted { + assert!(!outcome.bundles.contains_key(&companion.with_device(0))); + } + } + } + } + #[test] fn prekey_fetch_validates_mixed_and_hosted_adv_chains() { use crate::adv::test_util; diff --git a/wacore/src/stanza/groups/create_conformance.rs b/wacore/src/stanza/groups/create_conformance.rs index 1d31b8f7d..cab88a37d 100644 --- a/wacore/src/stanza/groups/create_conformance.rs +++ b/wacore/src/stanza/groups/create_conformance.rs @@ -228,7 +228,7 @@ fn isolated_official_create_helpers_match_rust_projection() { "default-and-unconditional-participant-username.memberShareGroupHistoryMode" ); assert!( - !(group.ephemerality.is_some()), + group.ephemerality.is_none(), "default-and-unconditional-participant-username" ); assert_eq!( @@ -1125,10 +1125,7 @@ fn isolated_official_create_helpers_match_rust_projection() { None, "parent-mode-flags-and-reason.memberShareGroupHistoryMode" ); - assert!( - !(group.ephemerality.is_some()), - "parent-mode-flags-and-reason" - ); + assert!(group.ephemerality.is_none(), "parent-mode-flags-and-reason"); assert_eq!( group.ephemerality.as_ref().map(|v| v.duration), None, @@ -1374,7 +1371,7 @@ fn isolated_official_create_helpers_match_rust_projection() { "time-cast-wraps-before-normalization.memberShareGroupHistoryMode" ); assert!( - !(group.ephemerality.is_some()), + group.ephemerality.is_none(), "time-cast-wraps-before-normalization" ); assert_eq!( @@ -1702,7 +1699,7 @@ fn isolated_official_create_helpers_match_rust_projection() { "history-share-mode-admin_share.memberShareGroupHistoryMode" ); assert!( - !(group.ephemerality.is_some()), + group.ephemerality.is_none(), "history-share-mode-admin_share" ); assert_eq!( @@ -1953,7 +1950,7 @@ fn isolated_official_create_helpers_match_rust_projection() { "history-share-mode-all_member_share.memberShareGroupHistoryMode" ); assert!( - !(group.ephemerality.is_some()), + group.ephemerality.is_none(), "history-share-mode-all_member_share" ); assert_eq!( @@ -2183,7 +2180,7 @@ fn isolated_official_create_helpers_match_rust_projection() { "history-share-mode-future_mode.memberShareGroupHistoryMode" ); assert!( - !(group.ephemerality.is_some()), + group.ephemerality.is_none(), "history-share-mode-future_mode" ); assert_eq!( @@ -2436,7 +2433,7 @@ fn isolated_official_create_helpers_match_rust_projection() { "history-is-read-without-feature-gate.memberShareGroupHistoryMode" ); assert!( - !(group.ephemerality.is_some()), + group.ephemerality.is_none(), "history-is-read-without-feature-gate" ); assert_eq!( @@ -2748,7 +2745,7 @@ fn isolated_official_create_helpers_match_rust_projection() { "disabled-creator-country-and-username-success.memberShareGroupHistoryMode" ); assert!( - !(group.ephemerality.is_some()), + group.ephemerality.is_none(), "disabled-creator-country-and-username-success" ); assert_eq!( diff --git a/wacore/src/types/jid.rs b/wacore/src/types/jid.rs index ffae078a7..49dd8b827 100644 --- a/wacore/src/types/jid.rs +++ b/wacore/src/types/jid.rs @@ -117,12 +117,35 @@ pub fn write_protocol_address_to(jid: &Jid, buf: &mut W push_protocol_address_to(jid, buf); } +/// The account that authenticates a companion's ADV certificate. +/// +/// WA Web's `asUserWidOrThrow` maps hosted devices back to their PN/LID +/// account. This is deliberately separate from Signal device addressing: +/// a hosted device's identity/session must keep its original namespace. +pub fn adv_account_jid(jid: &Jid) -> Jid { + jid.with_device_hosting(0, false) +} + +fn account_server(server: Server) -> Server { + match server { + Server::Hosted => Server::Pn, + Server::HostedLid => Server::Lid, + server => server, + } +} + /// Consistent ordering for deadlock-free multi-lock acquisition. +/// +/// Group by account family before ordering devices, so the ordinary account +/// anchor precedes its hosted companions too (`hosted.lid` sorts before `lid` +/// lexically). Keep the actual server as a tie-breaker: different Signal +/// namespaces must never compare equal and disappear during deduplication. pub fn cmp_for_lock_order(a: &Jid, b: &Jid) -> std::cmp::Ordering { - mapped_server(a.server.as_str()) - .cmp(mapped_server(b.server.as_str())) + mapped_server(account_server(a.server).as_str()) + .cmp(mapped_server(account_server(b.server).as_str())) .then_with(|| a.user.cmp(&b.user)) .then_with(|| a.device.cmp(&b.device)) + .then_with(|| mapped_server(a.server.as_str()).cmp(mapped_server(b.server.as_str()))) } /// Sort and deduplicate by user identity (user + server). @@ -244,6 +267,34 @@ mod tests { use super::*; use std::str::FromStr; + #[test] + fn adv_accounts_share_anchors_without_merging_signal_namespaces() { + for user in [Jid::pn("12025550180"), Jid::lid("100000000000180")] { + let ordinary = user.with_device(99); + let hosted = user.with_device_hosting(99, true); + assert_eq!(adv_account_jid(&ordinary), user); + assert_eq!(adv_account_jid(&hosted), user); + assert_ne!(ordinary.to_protocol_address(), hosted.to_protocol_address()); + let mut keys = vec![hosted.clone(), ordinary.clone(), user.clone()]; + keys.sort_unstable_by(cmp_for_lock_order); + keys.dedup_by(|a, b| cmp_for_lock_order(a, b).is_eq()); + assert_eq!(keys.len(), 3, "each Signal namespace keeps its own lock"); + assert_eq!(keys[0], user, "account anchor precedes every companion"); + for a in &keys { + for b in &keys { + assert_eq!( + cmp_for_lock_order(a, b).is_eq(), + a.to_protocol_address() == b.to_protocol_address() + ); + } + } + } + assert_eq!( + adv_account_jid(&Jid::group("12025550180-1")), + Jid::group("12025550180-1") + ); + } + #[test] fn test_signal_address_string_lid() { let jid = Jid::from_str("123456789@lid").unwrap(); From f8cbb37f8516ac7d01e90ca275746252a1543ad7 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 10 Nov 2026 10:48:32 -0300 Subject: [PATCH 3/8] fix: retain the preflight helper contract for hosted fixtures --- src/message/receive.rs | 34 +++++++++++++++++++++++----------- 1 file changed, 23 insertions(+), 11 deletions(-) diff --git a/src/message/receive.rs b/src/message/receive.rs index b0a0c3ce4..8eb36ec51 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -216,6 +216,16 @@ impl Client { sender: &Jid, certificate: Option<&[u8]>, payloads: &[EncPayload], + ) -> Option { + self.preflight_incoming_companion_identity_inner(sender, certificate, payloads, false) + .await + } + + async fn preflight_incoming_companion_identity_inner( + &self, + sender: &Jid, + certificate: Option<&[u8]>, + payloads: &[EncPayload], account_locked: bool, ) -> Option { if sender.device == 0 { @@ -754,7 +764,6 @@ impl Client { &sender_encryption_jid, device_identity.as_deref(), &session_payloads, - false, ) .await } else { @@ -804,7 +813,7 @@ impl Client { // cached trust only so no network wait enters processing admission. let invalid_companion_enc = if invalid_companion_enc.is_some() { let _account = self.lock_companion_account(&sender_encryption_jid).await; - self.preflight_incoming_companion_identity( + self.preflight_incoming_companion_identity_inner( &sender_encryption_jid, device_identity.as_deref(), &session_payloads, @@ -3360,12 +3369,7 @@ mod adv_tests { let permit = client.acquire_message_processing_permit().await; assert_eq!( client - .preflight_incoming_companion_identity( - &sender, - Some(&certificate), - &payloads, - false, - ) + .preflight_incoming_companion_identity(&sender, Some(&certificate), &payloads,) .await, Some(0) ); @@ -3373,9 +3377,17 @@ mod adv_tests { drop(permit); let _permit = client.acquire_message_processing_permit().await; let _account = client.lock_companion_account(&sender).await; - assert_eq!(client.preflight_incoming_companion_identity( - &sender, Some(&certificate), &payloads, true, - ).await, None); + assert_eq!( + client + .preflight_incoming_companion_identity_inner( + &sender, + Some(&certificate), + &payloads, + true, + ) + .await, + None + ); client.signal_shutdown_sync(); } From 4d0ca77c97807b2010a9d5866cdfd13b67596c89 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 10 Nov 2026 11:00:03 -0300 Subject: [PATCH 4/8] test: add a connected retention future layout diagnostic --- src/client/node_io.rs | 50 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/src/client/node_io.rs b/src/client/node_io.rs index 93bd159aa..15a732ad1 100644 --- a/src/client/node_io.rs +++ b/src/client/node_io.rs @@ -2820,6 +2820,56 @@ impl Client { #[cfg(test)] mod tests { + /// Compare this diagnostic on the exact base and head with identical flags. + /// It constructs futures without polling them or opening a network session. + #[tokio::test] + #[ignore = "layout diagnostic: run on base and head with --ignored --nocapture"] + async fn audit_connected_retention_future_layouts() { + use std::mem::{size_of, size_of_val}; + let client = crate::test_utils::create_test_client().await; + let shutdown = client.connection_shutdown_signal(); + let node = crate::test_utils::node_to_owned_ref( + &wacore_binary::builder::NodeBuilder::new("iq").build(), + ); + println!( + "QueueBudgets: {} bytes", + size_of::() + ); + println!( + "read_messages_loop: {} bytes", + size_of_val(&client.read_messages_loop()) + ); + println!( + "read_messages_loop_inner: {} bytes", + size_of_val(&client.read_messages_loop_inner()) + ); + println!( + "process_node_scoped: {} bytes", + size_of_val(&client.process_node_scoped(node.clone(), &shutdown, false)) + ); + println!( + "dispatch_node_scoped: {} bytes", + size_of_val(&client.dispatch_node_scoped(node, &shutdown, Some(StanzaTag::Iq), 0)) + ); + let reader_client = client.clone(); + let reader = async move { + let connection = reader_client.connect().await?; + connection.read_until_disconnected().await; + Ok::<(), crate::client::ConnectError>(()) + }; + println!( + "connected fixture reader task: {} bytes", + size_of_val(&reader) + ); + drop(reader); + let primary = wacore_binary::Jid::pn("12025550189"); + println!( + "ensure_adv_account_identity: {} bytes", + size_of_val(&client.ensure_adv_account_identity(&primary)) + ); + client.signal_shutdown_sync(); + } + #[cfg(feature = "mobile-core")] use crate::test_utils::primary_startup_client; From 151a7dde04f8be0eb3633863c37724e5fdcfea98 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Tue, 10 Nov 2026 12:52:10 -0300 Subject: [PATCH 5/8] fix: fence hosted USync refreshes on canonical account anchors --- src/client/node_io.rs | 8 ++--- src/prekeys.rs | 5 ++- src/usync.rs | 52 ++++++++++++++++++++++++++-- src/usync/ordinary_recovery_tests.rs | 12 ++++--- 4 files changed, 65 insertions(+), 12 deletions(-) diff --git a/src/client/node_io.rs b/src/client/node_io.rs index 15a732ad1..7cad6ee47 100644 --- a/src/client/node_io.rs +++ b/src/client/node_io.rs @@ -2828,9 +2828,7 @@ mod tests { use std::mem::{size_of, size_of_val}; let client = crate::test_utils::create_test_client().await; let shutdown = client.connection_shutdown_signal(); - let node = crate::test_utils::node_to_owned_ref( - &wacore_binary::builder::NodeBuilder::new("iq").build(), - ); + let node = crate::test_utils::node_to_owned_ref(&NodeBuilder::new("iq").build()); println!( "QueueBudgets: {} bytes", size_of::() @@ -2855,14 +2853,14 @@ mod tests { let reader = async move { let connection = reader_client.connect().await?; connection.read_until_disconnected().await; - Ok::<(), crate::client::ConnectError>(()) + Ok::<(), ConnectError>(()) }; println!( "connected fixture reader task: {} bytes", size_of_val(&reader) ); drop(reader); - let primary = wacore_binary::Jid::pn("12025550189"); + let primary = Jid::pn("12025550189"); println!( "ensure_adv_account_identity: {} bytes", size_of_val(&client.ensure_adv_account_identity(&primary)) diff --git a/src/prekeys.rs b/src/prekeys.rs index ab400f581..076928c3d 100644 --- a/src/prekeys.rs +++ b/src/prekeys.rs @@ -575,7 +575,10 @@ impl Client { .persist_adv_bootstrap_identity_inner(&primary_owned, &key, epoch) .await?; ensure_adv_bootstrap_scope(¤t_generation, generation, &shutdown)?; - anyhow::ensure!(accepted, "primary identity changed during ADV bootstrap"); + anyhow::ensure!( + accepted, + crate::usync::DeviceListIdentityConflict::Bootstrap + ); Ok(key) } .await; diff --git a/src/usync.rs b/src/usync.rs index f4acf4c7c..82c4523e9 100644 --- a/src/usync.rs +++ b/src/usync.rs @@ -1803,7 +1803,10 @@ impl Client { )> { for _ in 0..DEVICE_REFRESH_MAX_ATTEMPTS { let mut aliases = Vec::with_capacity(users.len()); - let mut addresses: Vec<_> = users.iter().map(|user| user.with_device(0)).collect(); + let mut addresses: Vec<_> = users + .iter() + .map(wacore::types::jid::adv_account_jid) + .collect(); for user in users { let alias = self.device_list_identity_alias(user).await; anyhow::ensure!( @@ -1814,7 +1817,7 @@ impl Client { )), "account alias changed during ADV recovery" ); - addresses.extend(alias.iter().map(|alias| alias.with_device(0))); + addresses.extend(alias.iter().map(wacore::types::jid::adv_account_jid)); aliases.push(alias); } addresses.sort_unstable_by(wacore::types::jid::cmp_for_lock_order); @@ -3383,6 +3386,51 @@ mod tests { client.signal_shutdown_sync(); } + #[tokio::test] + async fn hosted_refresh_waits_for_the_ordinary_account_fence() { + use std::sync::Arc; + use wacore::usync::{UserDeviceList, UsyncDevice}; + for primary in [Jid::pn("12025550190"), Jid::lid("100000000000190")] { + let client = create_test_client().await; + let hosted = primary.with_device_hosting(99, true).to_non_ad(); + let response = authenticated_response_fixture( + &client, + DeviceListResponse { + device_lists: vec![UserDeviceList { + user: hosted.clone(), + devices: vec![UsyncDevice::new(99, Some(1)).with_hosting(true)], + phash: None, + key_index_bytes: Some(signed_key_index_bytes(vec![1], 1)), + }], + lid_mappings: vec![], + }, + ) + .await; + let lock = client + .session_lock_for(primary.to_protocol_address().as_str()) + .await; + let guard = lock.lock().await; + let baseline = Arc::strong_count(&lock); + let worker = client.clone(); + let refresh = tokio::spawn(async move { + worker + .process_device_list_response(&response, Freshness::Refresh) + .await + }); + crate::test_utils::wait_for_lock_waiter(&lock, baseline).await; + assert!(!refresh.is_finished()); + assert!(client.load_device_record_for_jid(&primary).await.is_none()); + drop(guard); + tokio::time::timeout(std::time::Duration::from_secs(5), refresh) + .await + .unwrap() + .unwrap() + .unwrap(); + assert!(client.load_device_record_for_jid(&primary).await.is_some()); + client.signal_shutdown_sync(); + } + } + /// The hosted wire family must not hide ownership of our account anchor. #[tokio::test] async fn hosted_addressed_usync_cannot_replace_our_paired_account_anchor() { diff --git a/src/usync/ordinary_recovery_tests.rs b/src/usync/ordinary_recovery_tests.rs index 10b0f529e..64bba6d90 100644 --- a/src/usync/ordinary_recovery_tests.rs +++ b/src/usync/ordinary_recovery_tests.rs @@ -4190,12 +4190,16 @@ async fn adv_review_wire_hosted_subjects_cannot_bypass_the_account_epoch_graph() ); assert_eq!(client.cached_adv_account_identity(&lid).await.unwrap(), old); } + let hosted = Jid::new(lid.user.as_str(), wacore_binary::Server::HostedLid); + assert_eq!( + client.cached_adv_account_identity(&hosted).await.unwrap(), + old + ); assert!( client - .cached_adv_account_identity(&Jid::new( - lid.user.as_str(), - wacore_binary::Server::HostedLid - )) + .persistence_manager + .backend() + .load_identity(hosted.to_protocol_address().as_str()) .await .unwrap() .is_none(), From 2e4068e3b975c36522259077c7e2ba9f99e8caf1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Tue, 10 Nov 2026 14:03:28 -0300 Subject: [PATCH 6/8] fix: preserve typed conflicts across shared ADV bootstrap --- src/prekeys.rs | 17 +++++-- src/prekeys/adv_bootstrap_tests.rs | 66 +++++++++++++++++++++++- src/usync/ordinary_recovery_tests.rs | 75 ++++++++++++++++++++++++++++ 3 files changed, 151 insertions(+), 7 deletions(-) diff --git a/src/prekeys.rs b/src/prekeys.rs index 076928c3d..beac23d33 100644 --- a/src/prekeys.rs +++ b/src/prekeys.rs @@ -127,6 +127,16 @@ impl AdvBootstrapRegistry { #[derive(Debug)] struct SharedAdvBootstrapError(Arc); +fn shared_adv_bootstrap_error(error: Arc) -> anyhow::Error { + use crate::usync::DeviceListIdentityConflict; + match error.downcast_ref::() { + Some(DeviceListIdentityConflict::Alias) => DeviceListIdentityConflict::Alias.into(), + Some(DeviceListIdentityConflict::Primary) => DeviceListIdentityConflict::Primary.into(), + Some(DeviceListIdentityConflict::Bootstrap) => DeviceListIdentityConflict::Bootstrap.into(), + None => anyhow::Error::new(SharedAdvBootstrapError(error)), + } +} + impl std::fmt::Display for SharedAdvBootstrapError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { std::fmt::Display::fmt(self.0.as_ref(), f) @@ -485,8 +495,7 @@ impl Client { .result .lock() .unwrap_or_else(|p| p.into_inner()) = Some(result.clone()); - return result - .map_err(|error| anyhow::Error::new(SharedAdvBootstrapError(error))); + return result.map_err(shared_adv_bootstrap_error); } Err(flight) => { let _ = flight.released.recv().await; @@ -501,9 +510,7 @@ impl Client { .unwrap_or_else(|p| p.into_inner()) .clone(); if let Some(result) = result { - return result.map_err(|error| { - anyhow::Error::new(SharedAdvBootstrapError(error)) - }); + return result.map_err(shared_adv_bootstrap_error); } // Cancellation belongs to the leader, not its valid joiners. } diff --git a/src/prekeys/adv_bootstrap_tests.rs b/src/prekeys/adv_bootstrap_tests.rs index 6999017aa..503113262 100644 --- a/src/prekeys/adv_bootstrap_tests.rs +++ b/src/prekeys/adv_bootstrap_tests.rs @@ -22,7 +22,7 @@ fn adv_bootstrap_memory_reports_retained_capacity_after_cancelled_flights() { let leases: Vec<_> = (0..8) .map(|index| { registry - .claim(0, &Jid::pn(format!("12025551{index:03}"))) + .claim(0, &Jid::pn(format!("1202555{:04}", 100 + index))) .ok() .expect("independent account flight") }) @@ -243,6 +243,68 @@ async fn wait_for_joiner(client: &Client) { .await; } +#[test] +fn shared_adv_bootstrap_preserves_typed_conflicts_and_other_error_sources() { + use crate::usync::DeviceListIdentityConflict as Conflict; + for conflict in [Conflict::Alias, Conflict::Primary, Conflict::Bootstrap] { + let message = conflict.to_string(); + let shared = Arc::new(anyhow::Error::new(conflict).context("bootstrap transaction")); + for _ in 0..2 { + let error = shared_adv_bootstrap_error(shared.clone()); + assert!(error.is::()); + assert_eq!( + error.downcast_ref::().unwrap().to_string(), + message + ); + } + } + let shared = Arc::new(anyhow::Error::new(std::io::Error::other( + "storage unavailable", + ))); + let error = shared_adv_bootstrap_error(shared); + assert!(!error.is::()); + assert!(error.chain().any(|source| source.is::())); +} + +#[tokio::test] +async fn adv_bootstrap_alias_conflict_remains_typed_for_leader_and_joiner() { + let (client, transport) = create_iq_test_client().await; + let primary = Jid::pn("12025550191"); + let lid = Jid::lid("100000000000191"); + let leader = request(&client, primary.with_device(7)); + let iq = decode_sent_iq(&transport, 0).await; + let id = iq.get().attrs().optional_string("id").unwrap().into_owned(); + let joiner = request(&client, primary.with_device(9)); + wait_for_joiner(&client).await; + client + .lid_pn_cache + .add(&crate::lid_pn_cache::LidPnEntry::new( + lid.user.to_string(), + primary.user.to_string(), + crate::lid_pn_cache::LearningSource::Usync, + )) + .await; + client.lid_pn_cache.evict_reverse_for_tests(&lid.user).await; + let account = KeyPair::generate(&mut rand::make_rng::()); + answer(&client, &id, &primary, &account, true).await; + for caller in [leader, joiner] { + let error = caller.await.unwrap().unwrap_err(); + assert!(matches!( + error.downcast_ref::(), + Some(crate::usync::DeviceListIdentityConflict::Alias) + )); + } + assert_eq!(transport.sent_count(), 1); + assert!( + client + .cached_adv_account_identity(&primary) + .await + .unwrap() + .is_none() + ); + client.signal_shutdown_sync(); +} + async fn answer(client: &Arc, id: &str, primary: &Jid, account: &KeyPair, valid: bool) { let mut rng = rand::make_rng::(); let signed = KeyPair::generate(&mut rng); @@ -343,7 +405,7 @@ async fn adv_bootstrap_caps_queries_and_cancellation_releases_slots() { for index in 0..10 { tasks.push(request( &client, - Jid::pn_device(format!("12025550{index:03}"), 7), + Jid::pn_device(format!("1202555{:04}", 110 + index), 7), )); } poll_until("all cold accounts to enter admission", || { diff --git a/src/usync/ordinary_recovery_tests.rs b/src/usync/ordinary_recovery_tests.rs index 64bba6d90..1191f43fb 100644 --- a/src/usync/ordinary_recovery_tests.rs +++ b/src/usync/ordinary_recovery_tests.rs @@ -1813,6 +1813,81 @@ async fn group_identity_conflicts_retain_only_the_affected_contact() { } } +/// A cold contact's alias conflict must not prevent valid peers from publishing. +#[tokio::test] +async fn group_refresh_publishes_peers_after_a_cold_bootstrap_alias_conflict() { + let (client, transport) = create_iq_test_client().await; + let cold = Jid::pn("12025550195"); + let good = Jid::pn("12025550196"); + let lid = Jid::lid("100000000000195"); + crate::test_utils::seed_adv_primary(&client, &good).await; + for user in [&cold, &good] { + client + .update_device_list(DeviceListRecord { + user: user.user.as_str().into(), + devices: [DeviceInfo::new(0, None), DeviceInfo::new(3, Some(1))].into(), + raw_id: Some(1), + timestamp: 1, + phash: None, + }) + .await + .unwrap(); + } + let mut response = migration_response(&cold); + response + .device_lists + .extend(migration_response(&good).device_lists); + let users = [cold.clone(), good.clone()]; + let worker = client.clone(); + let operation = tokio::spawn(async move { + worker + .apply_device_list_response( + &response, + DeviceListApplyMode::GroupRefresh, + None, + Some(&users), + ) + .await + }); + let iq = decode_sent_iq(&transport, 0).await; + let id = iq.get().attrs().optional_string("id").unwrap().into_owned(); + client + .lid_pn_cache + .add(&crate::lid_pn_cache::LidPnEntry::new( + lid.user.to_string(), + cold.user.to_string(), + crate::lid_pn_cache::LearningSource::Usync, + )) + .await; + client.lid_pn_cache.evict_reverse_for_tests(&lid.user).await; + answer_iq( + &client, + &id, + &primary_response(&cold, crate::test_utils::adv_test_account(), true), + ) + .await; + let devices = tokio::time::timeout(Duration::from_secs(5), operation) + .await + .unwrap() + .unwrap() + .unwrap() + .unwrap(); + assert!(devices.contains(&good.with_device(7)), "{devices:?}"); + assert!(!devices.contains(&cold.with_device(7))); + assert!(client.load_device_record_for_jid(&good).await.is_some()); + assert!( + client + .load_device_record_for_jid(&cold) + .await + .unwrap() + .devices + .iter() + .any(|device| device.device_id() == 3) + ); + assert_eq!(transport.sent_count(), 1); + client.signal_shutdown_sync(); +} + /// A store fault must not masquerade as a per-contact trust rejection. #[tokio::test] async fn group_identity_storage_failure_is_not_partial_success() { From b61a405483bb098dc1fd943b5da721fc954437a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Tue, 10 Nov 2026 14:45:35 -0300 Subject: [PATCH 7/8] docs: describe ADV bootstrap contracts and regression scenarios --- src/inbound_queue.rs | 1 + src/message/hosted_adv_tests.rs | 7 +++++++ src/message/receive.rs | 3 +++ src/prekeys.rs | 7 +++++++ src/prekeys/adv_bootstrap_tests.rs | 14 ++++++++++++++ src/test_utils.rs | 1 + src/usync.rs | 2 ++ wacore/src/prekeys.rs | 1 + wacore/src/types/jid.rs | 2 ++ 9 files changed, 38 insertions(+) diff --git a/src/inbound_queue.rs b/src/inbound_queue.rs index fbec6def0..006c39bcb 100644 --- a/src/inbound_queue.rs +++ b/src/inbound_queue.rs @@ -145,6 +145,7 @@ impl Default for QueueBudgets { } } impl QueueBudgets { + /// Count queue controllers, live chat budgets and retained ADV flights, including empty map capacity. pub(crate) fn memory_stats(&self) -> (wacore::stats::CollectionStats, u64, u64) { use wacore::stats::HeapSize; let entries = self.per_chat.lock().unwrap_or_else(|p| p.into_inner()); diff --git a/src/message/hosted_adv_tests.rs b/src/message/hosted_adv_tests.rs index 9c5a53d96..437de4b48 100644 --- a/src/message/hosted_adv_tests.rs +++ b/src/message/hosted_adv_tests.rs @@ -4,6 +4,7 @@ use wacore::libsignal::protocol::{IdentityKey, KeyPair}; use wacore::store::commands::DeviceCommand; use wacore::types::jid::adv_account_jid; +/// Build signed companion prekey material with independently selectable hosting and signature prefixes. fn fixture( account: &KeyPair, hosted: bool, @@ -60,6 +61,7 @@ fn fixture( ) } +/// Publish a paired own-account anchor through device commands for PN or LID tests. async fn pair(client: &Client, own: &Jid) { let command = if own.server == wacore_binary::Server::Lid { DeviceCommand::SetLid(Some(own.with_device(1))) @@ -80,6 +82,7 @@ async fn pair(client: &Client, own: &Jid) { .await; } +/// Paired hosted devices reuse ordinary account trust while retaining their hosted Signal addresses. #[tokio::test] async fn hosted_own_adv_uses_paired_anchor_without_iq_or_signal_namespace_changes() { for own in [Jid::pn("12025550181"), Jid::lid("100000000000181")] { @@ -150,6 +153,7 @@ async fn hosted_own_adv_uses_paired_anchor_without_iq_or_signal_namespace_change } } +/// Reject mismatched anchors and signature domains for both ordinary and hosted companions. #[tokio::test] async fn hosted_and_ordinary_adv_reject_wrong_keys_signatures_and_prefixes() { for own in [Jid::pn("12025550182"), Jid::lid("100000000000182")] { @@ -218,6 +222,7 @@ async fn hosted_and_ordinary_adv_reject_wrong_keys_signatures_and_prefixes() { } } +/// Matching user text must not let PN and LID devices borrow each other's paired trust. #[tokio::test] async fn hosted_adv_does_not_borrow_paired_trust_from_the_other_addressing_family() { for own in [Jid::pn("12025550183"), Jid::lid("12025550183")] { @@ -275,6 +280,7 @@ async fn hosted_adv_does_not_borrow_paired_trust_from_the_other_addressing_famil } } +/// Queued hosted receives must validate against a rotated ordinary anchor after acquiring its fence. #[tokio::test] async fn hosted_receive_rechecks_the_ordinary_anchor_after_rotation() { for own in [Jid::pn("12025550184"), Jid::lid("100000000000184")] { @@ -339,6 +345,7 @@ async fn hosted_receive_rechecks_the_ordinary_anchor_after_rotation() { } } +/// Concurrent hosted LID send/receive paths must agree on account-before-device lock ordering. #[tokio::test] async fn hosted_lid_participant_send_and_receive_use_one_account_before_device_lock_order() { use crate::store::release::tests::ProbeBackend; diff --git a/src/message/receive.rs b/src/message/receive.rs index 8eb36ec51..1e4ecda6e 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -211,6 +211,7 @@ impl Client { .await } + /// Find the first unverifiable companion prekey before acquiring a processing permit. async fn preflight_incoming_companion_identity( &self, sender: &Jid, @@ -221,6 +222,7 @@ impl Client { .await } + /// Validate prekey certificates without advancing sessions; account-locked callers use cached trust only. async fn preflight_incoming_companion_identity_inner( &self, sender: &Jid, @@ -3261,6 +3263,7 @@ mod adv_tests { ) } + /// An unanswered cold IQ must leave offline admission available to a warm companion. #[tokio::test] async fn cold_adv_preflight_leaves_drain_and_warm_companion_unblocked() { let (client, transport) = crate::test_utils::create_iq_test_client().await; diff --git a/src/prekeys.rs b/src/prekeys.rs index beac23d33..8842f2586 100644 --- a/src/prekeys.rs +++ b/src/prekeys.rs @@ -60,6 +60,7 @@ struct AdvBootstrapLease { } impl Drop for AdvBootstrapLease { + /// Remove only this lease's flight so cancellation wakes joiners without erasing a replacement. fn drop(&mut self) { let mut flights = self .registry @@ -99,6 +100,7 @@ impl AdvBootstrapRegistry { .sum::() } + /// Join an account/generation flight, or become its leader until the returned lease drops. fn claim( self: &Arc, generation: u64, @@ -127,6 +129,7 @@ impl AdvBootstrapRegistry { #[derive(Debug)] struct SharedAdvBootstrapError(Arc); +/// Keep trust conflicts downcastable for per-contact group rejection; share other error sources. fn shared_adv_bootstrap_error(error: Arc) -> anyhow::Error { use crate::usync::DeviceListIdentityConflict; match error.downcast_ref::() { @@ -149,6 +152,7 @@ impl std::error::Error for SharedAdvBootstrapError { } } +/// Reject work after shutdown or a connection-generation change, including delayed IQ replies. fn ensure_adv_bootstrap_scope( generation: &portable_atomic::AtomicU64, expected: u64, @@ -294,6 +298,7 @@ fn authenticated_primary_identity( } impl Client { + /// Allocate flight coordination only when an account needs cold ADV discovery. fn adv_bootstrap(&self) -> &Arc { self.inbound_queues() .adv_bootstrap @@ -525,6 +530,8 @@ impl Client { } } + /// Bound cold IQ concurrency, then recheck trust under the account fence before owned persistence. + /// Cancelling the caller cannot release the admitted writer's fence or storage ownership. async fn bootstrap_adv_account_identity( &self, primary: &Jid, diff --git a/src/prekeys/adv_bootstrap_tests.rs b/src/prekeys/adv_bootstrap_tests.rs index 503113262..17be0164f 100644 --- a/src/prekeys/adv_bootstrap_tests.rs +++ b/src/prekeys/adv_bootstrap_tests.rs @@ -9,6 +9,7 @@ use wacore::protocol::ProtocolNode; use wacore::types::jid::JidExt; use wacore_binary::builder::NodeBuilder; +/// Repeated cancelled bursts release flight allocations without accumulating retained map capacity. #[test] fn adv_bootstrap_memory_reports_retained_capacity_after_cancelled_flights() { let budgets = crate::inbound_queue::QueueBudgets::default(); @@ -50,6 +51,7 @@ fn adv_bootstrap_memory_reports_retained_capacity_after_cancelled_flights() { } } +/// Spawn a caller whose cancellation can be controlled independently of the shared flight. fn request( client: &Arc, companion: Jid, @@ -58,6 +60,7 @@ fn request( tokio::spawn(async move { client.ensure_adv_account_identity(&companion).await }) } +/// Hosted and ordinary devices share one IQ and persist only their ordinary account anchor. #[tokio::test] async fn hosted_and_ordinary_companions_share_one_canonical_cold_bootstrap() { for primary in [Jid::pn("12025550185"), Jid::lid("100000000000185")] { @@ -109,6 +112,7 @@ async fn hosted_and_ordinary_companions_share_one_canonical_cold_bootstrap() { } } +/// A mixed prekey response authenticates hosted devices against its ordinary primary bundle. #[tokio::test] async fn cold_hosted_prekey_fetch_authenticates_and_persists_only_the_ordinary_anchor() { for primary in [Jid::pn("12025550189"), Jid::lid("100000000000189")] { @@ -230,6 +234,7 @@ async fn cold_hosted_prekey_fetch_authenticates_and_persists_only_the_ordinary_a } } +/// Wait for a second owner of the registered flight before exercising cancellation or shared results. async fn wait_for_joiner(client: &Client) { poll_until("another caller to join the ADV flight", || { client @@ -243,6 +248,7 @@ async fn wait_for_joiner(client: &Client) { .await; } +/// Trust conflicts stay downcastable through sharing; storage errors retain their original source chain. #[test] fn shared_adv_bootstrap_preserves_typed_conflicts_and_other_error_sources() { use crate::usync::DeviceListIdentityConflict as Conflict; @@ -266,6 +272,7 @@ fn shared_adv_bootstrap_preserves_typed_conflicts_and_other_error_sources() { assert!(error.chain().any(|source| source.is::())); } +/// An alias losing reciprocity during an IQ must produce the same typed rejection for both callers. #[tokio::test] async fn adv_bootstrap_alias_conflict_remains_typed_for_leader_and_joiner() { let (client, transport) = create_iq_test_client().await; @@ -305,6 +312,7 @@ async fn adv_bootstrap_alias_conflict_remains_typed_for_leader_and_joiner() { client.signal_shutdown_sync(); } +/// Deliver a signed primary bundle through the IQ correlator, optionally corrupting its signature. async fn answer(client: &Arc, id: &str, primary: &Jid, account: &KeyPair, valid: bool) { let mut rng = rand::make_rng::(); let signed = KeyPair::generate(&mut rng); @@ -341,6 +349,7 @@ async fn answer(client: &Arc, id: &str, primary: &Jid, account: &KeyPair .await; } +/// Concurrent callers share both authenticated success and signature failure without a second IQ. #[tokio::test] async fn adv_bootstrap_shares_success_and_invalid_signature_without_refetching() { for valid in [true, false] { @@ -374,6 +383,7 @@ async fn adv_bootstrap_shares_success_and_invalid_signature_without_refetching() } } +/// A shared IQ timeout must release the flight and every query admission slot. #[tokio::test(start_paused = true)] async fn adv_bootstrap_shares_timeout_and_releases_all_admission() { let (client, transport) = create_iq_test_client().await; @@ -398,6 +408,7 @@ async fn adv_bootstrap_shares_timeout_and_releases_all_admission() { client.signal_shutdown_sync(); } +/// At most eight cold IQs may run; cancelling callers releases slots for queued accounts. #[tokio::test] async fn adv_bootstrap_caps_queries_and_cancellation_releases_slots() { let (client, transport) = create_iq_test_client().await; @@ -433,6 +444,7 @@ async fn adv_bootstrap_caps_queries_and_cancellation_releases_slots() { client.signal_shutdown_sync(); } +/// A surviving joiner reclaims a result-less flight instead of inheriting its leader's cancellation. #[tokio::test] async fn adv_bootstrap_cancelled_leader_allows_valid_joiner_to_take_over() { let (client, transport) = create_iq_test_client().await; @@ -456,6 +468,7 @@ async fn adv_bootstrap_cancelled_leader_allows_valid_joiner_to_take_over() { client.signal_shutdown_sync(); } +/// Delayed IQ results cannot overwrite newer trust or publish into a retired connection. #[tokio::test] async fn adv_bootstrap_rechecks_current_trust_and_connection_before_publication() { for (retire_connection, fire_shutdown) in [(false, false), (true, false), (false, true)] { @@ -507,6 +520,7 @@ async fn adv_bootstrap_rechecks_current_trust_and_connection_before_publication( } } +/// Caller cancellation cannot release the account fence or drain ticket before backend completion. #[tokio::test] async fn adv_bootstrap_owned_writer_keeps_primary_fence_after_caller_cancellation() { use crate::store::release::tests::ProbeBackend; diff --git a/src/test_utils.rs b/src/test_utils.rs index 039828611..3f6bb76e4 100644 --- a/src/test_utils.rs +++ b/src/test_utils.rs @@ -99,6 +99,7 @@ pub(crate) fn adv_test_account() -> &'static wacore::libsignal::protocol::KeyPai }) } +/// Seed the shared test account key at the canonical ADV anchor, including for hosted device JIDs. #[cfg(test)] pub(crate) async fn seed_adv_primary(client: &Client, user: &Jid) { use wacore::types::jid::JidExt; diff --git a/src/usync.rs b/src/usync.rs index 82c4523e9..ce4efa7fc 100644 --- a/src/usync.rs +++ b/src/usync.rs @@ -1908,6 +1908,7 @@ impl Client { .map_err(|_| anyhow::anyhow!("bootstrap identity worker stopped"))? } + /// Persist a first-contact anchor only while reciprocal aliases and the captured identity epoch remain valid. pub(crate) async fn persist_adv_bootstrap_identity_inner( &self, primary: &Jid, @@ -3386,6 +3387,7 @@ mod tests { client.signal_shutdown_sync(); } + /// Hosted-addressed refreshes must wait on the same ordinary account lock as sends and receives. #[tokio::test] async fn hosted_refresh_waits_for_the_ordinary_account_fence() { use std::sync::Arc; diff --git a/wacore/src/prekeys.rs b/wacore/src/prekeys.rs index 3b6bf51f0..ba921667a 100644 --- a/wacore/src/prekeys.rs +++ b/wacore/src/prekeys.rs @@ -779,6 +779,7 @@ mod tests { .bundles } + /// Hosted companion bundles require an authenticated ordinary primary from the same response. #[test] fn hosted_prekey_bootstrap_uses_the_ordinary_primary_in_the_same_response() { let mut rng = rand::make_rng::(); diff --git a/wacore/src/types/jid.rs b/wacore/src/types/jid.rs index 49dd8b827..c82924e22 100644 --- a/wacore/src/types/jid.rs +++ b/wacore/src/types/jid.rs @@ -126,6 +126,7 @@ pub fn adv_account_jid(jid: &Jid) -> Jid { jid.with_device_hosting(0, false) } +/// Group hosted namespaces with their account family solely for lock ordering. fn account_server(server: Server) -> Server { match server { Server::Hosted => Server::Pn, @@ -267,6 +268,7 @@ mod tests { use super::*; use std::str::FromStr; + /// ADV anchors collapse hosting, but Signal namespaces remain distinct and account locks sort first. #[test] fn adv_accounts_share_anchors_without_merging_signal_namespaces() { for user in [Jid::pn("12025550180"), Jid::lid("100000000000180")] { From 476c73be01e6f4348b9aa8fd1d51fee7ab6f8793 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Tue, 10 Nov 2026 16:38:29 -0300 Subject: [PATCH 8/8] fix: keep optional queue task state off cold connected clients --- src/client/accessors.rs | 2 +- src/client/node_io.rs | 8 ++-- src/inbound_queue.rs | 75 ++++++++++++++++++++++++++---- src/message/durability.rs | 4 +- src/prekeys.rs | 4 +- src/prekeys/adv_bootstrap_tests.rs | 3 +- 6 files changed, 76 insertions(+), 20 deletions(-) diff --git a/src/client/accessors.rs b/src/client/accessors.rs index 7921924b3..a6260794e 100644 --- a/src/client/accessors.rs +++ b/src/client/accessors.rs @@ -645,7 +645,7 @@ impl Client { transport_ack_queue: self .inbound_queues .get() - .and_then(|q| q.ack_queue.get()) + .and_then(|q| q.ack_sender()) .map_or(0, |tx| tx.len()), delivery_receipt_queue: self.delivery_receipt_queue.get().map_or(0, |tx| tx.len()), response_waiters, diff --git a/src/client/node_io.rs b/src/client/node_io.rs index 7cad6ee47..4d6e486bc 100644 --- a/src/client/node_io.rs +++ b/src/client/node_io.rs @@ -1064,7 +1064,7 @@ impl Client { drop(node); let tx = self .inbound_queues() - .ack_queue + .ack_queue() .get_or_init(|| self.start_transport_ack_worker()); let mut lease = crate::inbound_queue::reserve(&self.inbound_queues().ack, bytes.capacity()); for _ in 0..crate::inbound_queue::SATURATION_RETRY_TURNS { @@ -3372,7 +3372,7 @@ mod tests { use wacore_binary::builder::NodeBuilder; let (client, _transport) = create_iq_test_client().await; let (tx, rx) = async_channel::bounded(1); - assert!(client.inbound_queues().ack_queue.set(tx).is_ok()); + assert!(client.inbound_queues().ack_queue().set(tx).is_ok()); let receipt = || { node_to_owned_ref( &NodeBuilder::new("receipt") @@ -3463,7 +3463,7 @@ mod tests { use wacore_binary::builder::NodeBuilder; let (client, _transport) = create_iq_test_client().await; let (tx, rx) = async_channel::bounded(1); - assert!(client.inbound_queues().ack_queue.set(tx).is_ok()); + assert!(client.inbound_queues().ack_queue().set(tx).is_ok()); let old = client.connection_shutdown_signal(); client.reset_connection_shutdown(); let receipt = node_to_owned_ref( @@ -3520,7 +3520,7 @@ mod tests { use wacore_binary::builder::NodeBuilder; let (client, _transport) = create_iq_test_client().await; let (tx, rx) = async_channel::bounded(1); - assert!(client.inbound_queues().ack_queue.set(tx).is_ok()); + assert!(client.inbound_queues().ack_queue().set(tx).is_ok()); let receipt = || { node_to_owned_ref( &NodeBuilder::new("receipt") diff --git a/src/inbound_queue.rs b/src/inbound_queue.rs index 006c39bcb..5499c433b 100644 --- a/src/inbound_queue.rs +++ b/src/inbound_queue.rs @@ -122,29 +122,53 @@ pub(crate) async fn yield_for_capacity(runtime: &dyn wacore::runtime::Runtime) { } } +/// ACK delivery and cold ADV discovery are optional on a connected client. +/// Share their lazy attachment so unused services do not enlarge its controller. +#[derive(Default)] +struct QueueTasks { + ack_queue: std::sync::OnceLock>>, + adv_bootstrap: std::sync::OnceLock>, +} + pub(crate) struct QueueBudgets { - pub ack_queue: std::sync::OnceLock>>, + tasks: std::sync::OnceLock>, pub chat: Arc, pub ack: Arc, pub work: Arc, - pub adv_bootstrap: std::sync::OnceLock>, per_chat: Arc, pub draining: std::sync::atomic::AtomicBool, } impl Default for QueueBudgets { fn default() -> Self { Self { - ack_queue: std::sync::OnceLock::new(), + tasks: std::sync::OnceLock::new(), chat: Arc::new(ByteBudget::new(GLOBAL_CHAT_BYTES)), ack: Arc::new(ByteBudget::new(ACK_BYTES)), work: Arc::default(), - adv_bootstrap: std::sync::OnceLock::new(), per_chat: Arc::default(), draining: false.into(), } } } impl QueueBudgets { + /// Attach the ACK queue only when a stanza actually needs deferred delivery. + pub(crate) fn ack_queue(&self) -> &std::sync::OnceLock>> { + &self.tasks.get_or_init(Box::default).ack_queue + } + + /// Inspect an existing ACK worker without allocating optional task state. + pub(crate) fn ack_sender(&self) -> Option<&async_channel::Sender>> { + self.tasks.get().and_then(|tasks| tasks.ack_queue.get()) + } + + /// Attach account-flight coordination only when ADV discovery needs it. + pub(crate) fn adv_bootstrap(&self) -> &Arc { + self.tasks + .get_or_init(Box::default) + .adv_bootstrap + .get_or_init(Arc::default) + } + /// Count queue controllers, live chat budgets and retained ADV flights, including empty map capacity. pub(crate) fn memory_stats(&self) -> (wacore::stats::CollectionStats, u64, u64) { use wacore::stats::HeapSize; @@ -166,10 +190,13 @@ impl QueueBudgets { + size_of::() + 2 * (size_of::() + 2 * size_of::()) + size_of::() + 2 * size_of::() - + self - .adv_bootstrap - .get() - .map_or(0, |registry| registry.retained_bytes()); + + self.tasks.get().map_or(0, |tasks| { + size_of::() + + tasks + .adv_bootstrap + .get() + .map_or(0, |registry| registry.retained_bytes()) + }); ( wacore::stats::CollectionStats::new(entries.len() as u64, registry_bytes as u64), GLOBAL_CHAT_BYTES.saturating_sub(self.chat.available()) as u64, @@ -357,6 +384,38 @@ impl crate::client::Client { #[cfg(test)] mod tests { + /// Optional workers must not add inline words or allocate during stats reads. + #[test] + fn cold_queue_controller_keeps_its_pre_adv_footprint() { + // The pre-ADV controller had one ACK slot, four Arc pointers and a + // draining flag padded to pointer alignment. Keep that representation. + let legacy = size_of::>>>() + + 5 * size_of::(); + assert_eq!(size_of::(), legacy); + let budgets = QueueBudgets::default(); + let before = budgets.memory_stats().0.bytes; + assert!(budgets.ack_sender().is_none()); + assert_eq!(budgets.memory_stats().0.bytes, before); + assert!(budgets.tasks.get().is_none()); + } + + /// Count the shared sidecar once, then count ADV ownership when it is used. + #[test] + fn optional_queue_task_allocations_remain_visible() { + let budgets = QueueBudgets::default(); + let cold = budgets.memory_stats().0.bytes; + let (tx, _rx) = async_channel::bounded(1); + assert!(budgets.ack_queue().set(tx).is_ok()); + let ack = budgets.memory_stats().0.bytes; + assert_eq!(ack - cold, size_of::() as u64); + let registry = budgets.adv_bootstrap(); + assert_eq!( + budgets.memory_stats().0.bytes - ack, + registry.retained_bytes() as u64 + ); + assert!(budgets.ack_sender().is_some()); + } + use super::*; use crate::client::{ChatLane, QueuedChatMessage}; use async_lock::Mutex; diff --git a/src/message/durability.rs b/src/message/durability.rs index 86d317deb..42488bc6d 100644 --- a/src/message/durability.rs +++ b/src/message/durability.rs @@ -648,7 +648,7 @@ mod tests { client .inbound_queues .get() - .and_then(|q| q.ack_queue.get()) + .and_then(|q| q.ack_sender()) .is_none(), "a successful subset hook must not acknowledge the retained group" ); @@ -666,7 +666,7 @@ mod tests { || client .inbound_queues .get() - .and_then(|q| q.ack_queue.get()) + .and_then(|q| q.ack_sender()) .is_some(), "only the complete replay may schedule the stanza acknowledgement" ); diff --git a/src/prekeys.rs b/src/prekeys.rs index 8842f2586..fef869eb5 100644 --- a/src/prekeys.rs +++ b/src/prekeys.rs @@ -300,9 +300,7 @@ fn authenticated_primary_identity( impl Client { /// Allocate flight coordination only when an account needs cold ADV discovery. fn adv_bootstrap(&self) -> &Arc { - self.inbound_queues() - .adv_bootstrap - .get_or_init(Arc::default) + self.inbound_queues().adv_bootstrap() } /// Read a fresh, signed primary key without publishing a trust change. diff --git a/src/prekeys/adv_bootstrap_tests.rs b/src/prekeys/adv_bootstrap_tests.rs index 17be0164f..12c1af6fc 100644 --- a/src/prekeys/adv_bootstrap_tests.rs +++ b/src/prekeys/adv_bootstrap_tests.rs @@ -14,8 +14,7 @@ use wacore_binary::builder::NodeBuilder; fn adv_bootstrap_memory_reports_retained_capacity_after_cancelled_flights() { let budgets = crate::inbound_queue::QueueBudgets::default(); let baseline = budgets.memory_stats().0; - assert!(budgets.adv_bootstrap.get().is_none()); - let registry = budgets.adv_bootstrap.get_or_init(Arc::default); + let registry = budgets.adv_bootstrap(); let initialized = budgets.memory_stats().0; assert!(initialized.bytes > baseline.bytes); let mut first_peak = None;