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/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/adapters.rs b/src/client/adapters.rs index 2473d2464..2bfcc1568 100644 --- a/src/client/adapters.rs +++ b/src/client/adapters.rs @@ -76,7 +76,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 8d32fc69d..d155966dc 100644 --- a/src/client/lid_pn.rs +++ b/src/client/lid_pn.rs @@ -1203,8 +1203,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/client/node_io.rs b/src/client/node_io.rs index 0ebbc0b6a..d81a6d081 100644 --- a/src/client/node_io.rs +++ b/src/client/node_io.rs @@ -1168,7 +1168,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 { @@ -2924,6 +2924,54 @@ 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(&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::<(), ConnectError>(()) + }; + println!( + "connected fixture reader task: {} bytes", + size_of_val(&reader) + ); + drop(reader); + let primary = 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; @@ -3428,7 +3476,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") @@ -3519,7 +3567,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( @@ -3576,7 +3624,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 e3c9b8dce..5499c433b 100644 --- a/src/inbound_queue.rs +++ b/src/inbound_queue.rs @@ -122,8 +122,16 @@ 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, @@ -133,7 +141,7 @@ pub(crate) struct QueueBudgets { 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(), @@ -143,6 +151,25 @@ impl Default for QueueBudgets { } } 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; let entries = self.per_chat.lock().unwrap_or_else(|p| p.into_inner()); @@ -162,7 +189,14 @@ 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.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, @@ -350,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/message/hosted_adv_tests.rs b/src/message/hosted_adv_tests.rs new file mode 100644 index 000000000..e6ffeb21e --- /dev/null +++ b/src/message/hosted_adv_tests.rs @@ -0,0 +1,517 @@ +use super::*; +use std::time::Duration; +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, + 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(), + ) +} + +/// 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))) + } 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; +} + +/// 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")] { + 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(); + } +} + +/// 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")] { + 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(); + } +} + +/// 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")] { + 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(); + } +} + +/// 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")] { + 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(); + } +} + +/// 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; + 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 generation = client.connection_generation.load(Ordering::Acquire); + let mut processing_permit = Some(client.acquire_message_processing_permit().await); + 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), + &mut processing_permit, + generation, + ), + ) + .await + .unwrap(); + drop(processing_permit); + 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 199b1480f..e2caa1a39 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -212,6 +212,56 @@ impl Client { .await } + /// Find the first unverifiable companion prekey before acquiring a processing permit. + async fn preflight_incoming_companion_identity( + &self, + sender: &Jid, + certificate: Option<&[u8]>, + payloads: &[EncPayload], + ) -> Option { + self.preflight_incoming_companion_identity_inner(sender, certificate, payloads, false) + .await + } + + /// Validate prekey certificates without advancing sessions; account-locked callers use cached trust only. + async fn preflight_incoming_companion_identity_inner( + &self, + sender: &Jid, + certificate: Option<&[u8]>, + payloads: &[EncPayload], + account_locked: bool, + ) -> 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); + let validation = if account_locked { + self.validate_incoming_companion_identity_inner( + sender, + certificate, + &identity, + true, + ) + .await + } else { + self.validate_incoming_companion_identity(sender, certificate, &identity) + .await + }; + if validation.is_err() { + return Some(payload.enc_index); + } + } + None + } + async fn validate_incoming_companion_identity_inner( &self, sender: &Jid, @@ -712,7 +762,8 @@ impl Client { } } - /// Phase 2: acquire permit, decrypt payloads, flush. No node borrows. + /// Phase 2: preflight account trust, acquire a processing permit, decrypt, + /// then suspend admission for any quiesced flush. No node borrows. #[cfg_attr( feature = "tracing", tracing::instrument(name = "wa.recv.process", level = "debug", skip_all) @@ -734,9 +785,25 @@ impl Client { ack_recipient, } = msg; - // 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 + // 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 one permit from the current bounded processing pool. + // The helper re-acquires across a drain→live semaphore swap: + // without that, a task waiting on the old processing pool would be // silently dropped, losing pkmsg messages carrying SKDM (sender key // distribution) — and a lost SKDM fails ALL subsequent skmsg from that // sender with "No sender key state". @@ -759,63 +826,59 @@ 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; + // A recovery may publish while the offline permit is unavailable. + // Recheck negative preflight results under the account fence, using + // 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_inner( + &sender_encryption_jid, + device_identity.as_deref(), + &session_payloads, + true, + ) + .await + } else { + None + }; + + 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(); @@ -3389,6 +3452,10 @@ mod tests { } } +#[cfg(test)] +#[path = "hosted_adv_tests.rs"] +mod hosted_adv_tests; + #[cfg(test)] mod adv_tests { use super::*; @@ -3437,6 +3504,137 @@ 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; + 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 permits, even if its companion shares this account. + let _request = crate::test_utils::decode_sent_iq(&transport, 0).await; + let permits = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.acquire_message_quiescence(), + ) + .await + .expect("cold ADV must leave every offline processing permit available"); + 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(permits); + incoming.abort(); + assert!(incoming.await.unwrap_err().is_cancelled()); + assert!(client.response_waiters_guard().is_empty()); + client.signal_shutdown_sync(); + } + + /// A negative preflight can become valid before offline admission opens. + #[tokio::test] + async fn negative_adv_preflight_uses_recovered_trust_after_admission() { + use wacore::libsignal::protocol::KeyPair; + let client = crate::test_utils::create_test_client().await; + client.enter_drain_mode_for_tests(); + let sender = Jid::pn_device("12025550129", 7); + let (payloads, certificate, _) = companion_payload_fixture(); + let stale = KeyPair::generate(&mut rand::make_rng::()); + client + .signal_cache + .put_identity( + &sender.with_device(0).to_protocol_address(), + stale.public_key.public_key_bytes(), + ) + .await; + let permits = client.acquire_message_quiescence().await; + assert_eq!( + client + .preflight_incoming_companion_identity(&sender, Some(&certificate), &payloads,) + .await, + Some(0) + ); + crate::test_utils::seed_adv_primary(&client, &sender).await; + drop(permits); + let _permit = client.acquire_message_processing_permit().await; + let _account = client.lock_companion_account(&sender).await; + assert_eq!( + client + .preflight_incoming_companion_identity_inner( + &sender, + Some(&certificate), + &payloads, + true, + ) + .await, + None + ); + 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 7f3388cbe..84f083ed8 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 0b85b857d..5a5c3d6bb 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,143 @@ 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 waiting for offline-processing admission. 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 { + /// Remove only this lease's flight so cancellation wakes joiners without erasing a replacement. + 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::() + } + + /// Join an account/generation flight, or become its leader until the returned lease drops. + 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); + +/// 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::() { + 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) + } +} + +impl std::error::Error for SharedAdvBootstrapError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.0.as_ref().as_ref()) + } +} + +/// Reject work after shutdown or a connection-generation change, including delayed IQ replies. +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,9 +298,14 @@ 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() + } + /// 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 } @@ -209,7 +353,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()); } @@ -235,7 +379,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 { @@ -285,7 +429,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 @@ -319,22 +463,142 @@ 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 primary = jid.with_device(0); - 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 generation = self.connection_generation.load(Ordering::Acquire); + let shutdown = self.connection_shutdown_signal(); + let primary = wacore::types::jid::adv_account_jid(jid); + 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); + } + } + loop { + 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()); + return result.map_err(shared_adv_bootstrap_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(); + if let Some(result) = result { + return result.map_err(shared_adv_bootstrap_error); + } + // Cancellation belongs to the leader, not its valid joiners. + } + } + } + }; + futures::select_biased! { + _ = wacore::runtime::wait_for_shutdown(&shutdown).fuse() => { + anyhow::bail!("connection closed during ADV bootstrap") + }, + result = Box::pin(bootstrap).fuse() => result, + } + } + + /// 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, + 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 epoch = self.signal_cache.identity_continuity.primary_snapshot(); - let key = self.query_adv_account_identity(&primary, None).await?; - if !self - .persist_adv_bootstrap_identity_in_epoch(&primary, &key, epoch) - .await? - { - anyhow::bail!(crate::usync::DeviceListIdentityConflict::Bootstrap); + 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_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); } - Ok(key) + let work = self + .inbound_queues() + .work + .try_track() + .ok_or_else(|| anyhow::anyhow!("ADV persistence admission closed"))?; + let client = self + .self_weak + .get() + .and_then(|weak| weak.upgrade()) + .ok_or_else(|| anyhow::anyhow!("client ownership unavailable during bootstrap"))?; + let current_generation = self.connection_generation.clone(); + let shutdown = shutdown.clone(); + let primary_owned = primary.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 = client + .persist_adv_bootstrap_identity_inner(&primary_owned, &key, epoch) + .await?; + ensure_adv_bootstrap_scope(¤t_generation, generation, &shutdown)?; + anyhow::ensure!( + accepted, + crate::usync::DeviceListIdentityConflict::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 @@ -1128,6 +1392,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..12c1af6fc --- /dev/null +++ b/src/prekeys/adv_bootstrap_tests.rs @@ -0,0 +1,610 @@ +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; + +/// 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(); + let baseline = budgets.memory_stats().0; + let registry = budgets.adv_bootstrap(); + 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!("1202555{:04}", 100 + index))) + .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" + ); + } +} + +/// Spawn a caller whose cancellation can be controlled independently of the shared flight. +fn request( + client: &Arc, + companion: Jid, +) -> tokio::task::JoinHandle> { + let client = client.clone(); + 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")] { + 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(); + } +} + +/// 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")] { + 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(); + } +} + +/// 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 + .adv_bootstrap() + .flights + .lock() + .unwrap() + .values() + .any(|flight| Arc::strong_count(flight) > 2) + }) + .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; + 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::())); +} + +/// 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; + 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(); +} + +/// 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); + 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; +} + +/// 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] { + 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(); + } +} + +/// 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; + 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(); +} + +/// 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; + let mut tasks = Vec::new(); + for index in 0..10 { + tasks.push(request( + &client, + Jid::pn_device(format!("1202555{:04}", 110 + index), 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(); +} + +/// 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; + 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()); + + 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_eq!( + joined.await.unwrap().unwrap(), + *account.public_key.public_key_bytes() + ); + assert!(client.adv_bootstrap().flights.lock().unwrap().is_empty()); + 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)] { + 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(); + } +} + +/// 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; + 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(); + } +} + +/// Dropping all callers cannot release storage while the admitted write owns it. +#[tokio::test] +async fn adv_bootstrap_writer_delays_store_release_after_client_drop() { + use crate::store::release::tests::ProbeBackend; + use crate::test_utils::create_iq_test_client_with_backend; + let (mut probe, dropped) = 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 = "put_identity"; + let (client, transport) = create_iq_test_client_with_backend(Arc::new(probe)).await; + let released = client.store_release(); + let primary = Jid::pn("12025550179"); + 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()); + client.signal_shutdown_sync(); + drop(client); + let mut wait = Box::pin(released.wait()); + assert!(futures::poll!(&mut wait).is_pending()); + assert!(!dropped.load(Ordering::SeqCst)); + release.send(()).await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), wait) + .await + .unwrap(); + assert!(dropped.load(Ordering::SeqCst)); +} 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 18095d4f2..ffd154b81 100644 --- a/src/send/mod.rs +++ b/src/send/mod.rs @@ -371,7 +371,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); @@ -2123,10 +2123,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/src/test_utils.rs b/src/test_utils.rs index ad0d78511..bcff5e460 100644 --- a/src/test_utils.rs +++ b/src/test_utils.rs @@ -99,13 +99,14 @@ 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; 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 45fda0de1..79afbc697 100644 --- a/src/usync.rs +++ b/src/usync.rs @@ -769,7 +769,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); } @@ -2221,7 +2221,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 (position, user) in users.iter().enumerate() { let alias = self.device_list_identity_alias(user).await; // A continuation has already learned its own accepted mappings. @@ -2239,7 +2242,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); @@ -2330,7 +2333,8 @@ impl Client { .map_err(|_| anyhow::anyhow!("bootstrap identity worker stopped"))? } - async fn persist_adv_bootstrap_identity_inner( + /// 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, key: &[u8; 32], @@ -2684,11 +2688,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 @@ -2702,7 +2704,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 { @@ -3841,6 +3843,114 @@ 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; + 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() { + 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/src/usync/ordinary_recovery_tests.rs b/src/usync/ordinary_recovery_tests.rs index fca865498..ed2fd74b6 100644 --- a/src/usync/ordinary_recovery_tests.rs +++ b/src/usync/ordinary_recovery_tests.rs @@ -3767,6 +3767,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() { @@ -6178,12 +6253,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(), 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..ba921667a 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,84 @@ 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::(); + 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/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()) }); } 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..c82924e22 100644 --- a/wacore/src/types/jid.rs +++ b/wacore/src/types/jid.rs @@ -117,12 +117,36 @@ 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) +} + +/// Group hosted namespaces with their account family solely for lock ordering. +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 +268,35 @@ 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")] { + 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();