diff --git a/Cargo.lock b/Cargo.lock index 08ab1affc..3679f02dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -624,10 +624,12 @@ dependencies = [ "chrono", "dhat", "env_logger", + "futures", "log", "tokio", "uuid", "wacore", + "wacore-binary", "whatsapp-rust", "whatsapp-rust-sqlite-storage", "whatsapp-rust-tokio-transport", diff --git a/src/client.rs b/src/client.rs index 040b4df65..66a42d10b 100644 --- a/src/client.rs +++ b/src/client.rs @@ -91,6 +91,11 @@ struct NodeWaiter { tx: futures::channel::oneshot::Sender>, } +struct SentNodeWaiter { + filter: NodeFilter, + tx: futures::channel::oneshot::Sender>, +} + use async_lock::Mutex; use async_lock::RwLock; use std::time::Duration; @@ -282,6 +287,9 @@ pub struct Client { /// Guarded by `node_waiter_count` for zero-cost when no waiters are active. node_waiters: std::sync::Mutex>, node_waiter_count: AtomicUsize, + /// Waiters for raw outgoing nodes before encryption. + sent_node_waiters: std::sync::Mutex>, + sent_node_waiter_count: AtomicUsize, pub(crate) unique_id: String, pub(crate) id_counter: Arc, @@ -582,6 +590,8 @@ impl Client { response_waiters: Arc::new(Mutex::new(HashMap::new())), node_waiters: std::sync::Mutex::new(Vec::new()), node_waiter_count: AtomicUsize::new(0), + sent_node_waiters: std::sync::Mutex::new(Vec::new()), + sent_node_waiter_count: AtomicUsize::new(0), unique_id: format!("{}.{}", unique_id_bytes[0], unique_id_bytes[1]), id_counter: Arc::new(AtomicU64::new(0)), unified_session: crate::unified_session::UnifiedSessionManager::new(), @@ -1051,6 +1061,12 @@ impl Client { } async fn cleanup_connection_state(&self) { + // Note: node_waiters are intentionally NOT cleared here — they are + // cross-connection (callers may register a waiter before an action that + // completes on a subsequent connection, e.g. after 515 reconnect). + // sent_node_waiters ARE cleared because they match pre-encryption + // outgoing stanzas, which are transport-scoped. + self.clear_sent_node_waiters(); self.is_logged_in.store(false, Ordering::Relaxed); self.is_ready.store(false, Ordering::Relaxed); // Signal the keepalive loop (and any other tasks) to exit promptly. @@ -2619,10 +2635,38 @@ impl Client { ) { use wacore::types::events::Event; - if m.operation != wa::syncd_mutation::SyncdOperation::Set { + if m.index.is_empty() { return; } - if m.index.is_empty() { + + // NCT salt sync — handles both "set" (store salt) and "remove" (clear salt). + // Source: WAWebNctSaltSync, syncd collection RegularHigh, action "nct_salt_sync". + if m.index[0] == "nct_salt_sync" { + if m.operation == wa::syncd_mutation::SyncdOperation::Remove { + debug!(target: "Client/AppState", "Removing NCT salt via app state sync"); + self.persistence_manager + .process_command(DeviceCommand::SetNctSalt(None)) + .await; + } else if let Some(val) = &m.action_value + && let Some(act) = &val.nct_salt_sync_action + && let Some(salt) = &act.salt + { + if salt.is_empty() { + warn!(target: "Client/AppState", "nct_salt_sync mutation has empty salt, ignoring"); + } else { + debug!(target: "Client/AppState", "Stored NCT salt via app state sync ({} bytes)", salt.len()); + self.persistence_manager + .process_command(DeviceCommand::SetNctSalt(Some(salt.clone()))) + .await; + } + } else { + warn!(target: "Client/AppState", "nct_salt_sync mutation missing salt in action value"); + } + return; + } + + // All remaining mutations only care about Set operations + if m.operation != wa::syncd_mutation::SyncdOperation::Set { return; } @@ -2953,6 +2997,25 @@ impl Client { rx } + /// Register a waiter for an outgoing node before it is encrypted and sent. + /// + /// This is intended for tests and diagnostics that need to inspect the raw + /// stanza built by the client, such as asserting whether `` or + /// `` was attached. + pub fn wait_for_sent_node( + &self, + filter: NodeFilter, + ) -> futures::channel::oneshot::Receiver> { + let (tx, rx) = futures::channel::oneshot::channel(); + self.sent_node_waiter_count.fetch_add(1, Ordering::Release); + let mut waiters = self + .sent_node_waiters + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + waiters.push(SentNodeWaiter { filter, tx }); + rx + } + /// Check pending node waiters against an incoming node. /// Only called when `node_waiter_count > 0`. fn resolve_node_waiters(&self, node: &Arc) { @@ -2977,6 +3040,39 @@ impl Client { } } + fn resolve_sent_node_waiters(&self, node: &Arc) { + let mut waiters = self + .sent_node_waiters + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let mut i = 0; + while i < waiters.len() { + if waiters[i].tx.is_canceled() { + waiters.swap_remove(i); + self.sent_node_waiter_count.fetch_sub(1, Ordering::Release); + } else if waiters[i].filter.matches(node) { + let w = waiters.swap_remove(i); + self.sent_node_waiter_count.fetch_sub(1, Ordering::Release); + let _ = w.tx.send(Arc::clone(node)); + } else { + i += 1; + } + } + } + + fn clear_sent_node_waiters(&self) { + let mut waiters = self + .sent_node_waiters + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let count = waiters.len(); + if count > 0 { + waiters.clear(); + self.sent_node_waiter_count + .fetch_sub(count, Ordering::Release); + } + } + pub(crate) fn update_server_time_offset(&self, node: &wacore_binary::node::Node) { self.unified_session.update_server_time_offset(node); } @@ -3128,6 +3224,9 @@ impl Client { }; debug!(target: "Client/Send", "{}", DisplayableNode(&node)); + if self.sent_node_waiter_count.load(Ordering::Acquire) > 0 { + self.resolve_sent_node_waiters(&Arc::new(node.clone())); + } let mut plaintext_buf = Vec::with_capacity(1024); diff --git a/src/features/contacts.rs b/src/features/contacts.rs index 5adf48982..de145637d 100644 --- a/src/features/contacts.rs +++ b/src/features/contacts.rs @@ -25,6 +25,35 @@ impl<'a> Contacts<'a> { Self { client } } + async fn persist_lid_mappings<'b, I>(&self, entries: I) + where + I: IntoIterator)>, + { + for (jid, lid) in entries { + let Some(lid) = lid else { + continue; + }; + if !jid.is_pn() || !lid.is_lid() { + continue; + } + if let Err(err) = self + .client + .add_lid_pn_mapping( + &lid.user, + &jid.user, + crate::lid_pn_cache::LearningSource::Usync, + ) + .await + { + log::warn!( + "Failed to persist usync LID mapping {} -> {}: {err}", + jid, + lid + ); + } + } + } + pub async fn is_on_whatsapp(&self, phones: &[&str]) -> Result> { if phones.is_empty() { return Ok(Vec::new()); @@ -50,7 +79,10 @@ impl<'a> Contacts<'a> { let phone_strings: Vec = phones.iter().map(|s| s.to_string()).collect(); let spec = ContactInfoSpec::new(phone_strings, request_id); - Ok(self.client.execute(spec).await?) + let info = self.client.execute(spec).await?; + self.persist_lid_mappings(info.iter().map(|entry| (&entry.jid, entry.lid.as_ref()))) + .await; + Ok(info) } pub async fn get_profile_picture( @@ -98,7 +130,10 @@ impl<'a> Contacts<'a> { let request_id = self.client.generate_request_id(); let spec = UserInfoSpec::new(jids.to_vec(), request_id); - Ok(self.client.execute(spec).await?) + let info = self.client.execute(spec).await?; + self.persist_lid_mappings(info.values().map(|entry| (&entry.jid, entry.lid.as_ref()))) + .await; + Ok(info) } } diff --git a/src/history_sync.rs b/src/history_sync.rs index d78d8ea86..47ff8c560 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -255,6 +255,20 @@ impl Client { log::info!("Updating own push name from history sync to '{new_name}'"); self.update_push_name_and_notify(new_name).await; } + + // Store NCT salt if found. + // WA Web: storeNctSaltFromHistorySync in MsgHandlerAction.js + if let Some(salt) = sync_result.nct_salt { + log::info!( + "History sync provided NCT salt ({} bytes); applying as backfill only", + salt.len() + ); + self.persistence_manager + .process_command( + wacore::store::commands::DeviceCommand::SetNctSaltFromHistorySync(salt), + ) + .await; + } } Some(Err(e)) => { log::error!("Failed to process HistorySync data: {:?}", e); diff --git a/src/send.rs b/src/send.rs index bf52db37b..3b75690b3 100644 --- a/src/send.rs +++ b/src/send.rs @@ -2,6 +2,7 @@ use crate::client::Client; use crate::store::signal_adapter::SignalProtocolStoreAdapter; use crate::types::message::EditAttribute; use anyhow::anyhow; +use log::debug; use wacore::client::context::SendContextResolver; use wacore::libsignal::protocol::SignalProtocolError; use wacore::types::jid::JidExt; @@ -721,6 +722,9 @@ impl Client { participants: Vec, } let mut skdm_update: Option = None; + let mut should_issue_tc_token_after_send = false; + let mut used_cached_tc_token_key: Option = None; + let tc_issue_target = to.clone(); let stanza_to_send: wacore_binary::Node = if peer && !to.is_group() { // Peer messages are only valid for individual users, not groups @@ -1008,8 +1012,16 @@ impl Client { // Skip for newsletters, groups, and own JID. let mut extra_stanza_nodes = extra_stanza_nodes; if !to.is_group() && !to.is_newsletter() { - self.maybe_include_tc_token(&to, &mut extra_stanza_nodes) + let (should_issue_after_send, cached_token_key) = self + .maybe_include_tc_token(&to, &mut extra_stanza_nodes) .await; + should_issue_tc_token_after_send = should_issue_after_send; + if should_issue_after_send { + used_cached_tc_token_key = cached_token_key; + } + } + if should_issue_tc_token_after_send { + debug!(target: "Client/TcToken", "Scheduled tc token issuance after send for {}", to); } // Acquire lock only for encryption @@ -1066,19 +1078,44 @@ impl Client { log::error!("Failed to flush signal cache after send_message_impl: {e:?}"); } + // Issue new tc token after send if a bucket boundary was crossed. + // WA Web fires this concurrently (MsgJob.js: sendTcToken is not awaited), + // but our send methods take &self not &Arc so we can't spawn here. + // Placed last so it doesn't block SKDM or signal-cache flush. + // + // WA Web only updates tcTokenSenderTimestamp after a successful issuance + // (TcTokenChatAction.js), so we gate the sender_timestamp mark on success + // to allow retry on the next send if the IQ failed. + let issued_ok = if should_issue_tc_token_after_send { + self.issue_tc_token_after_send(&tc_issue_target).await + } else { + false + }; + if issued_ok && let Some(token_key) = used_cached_tc_token_key { + self.mark_tc_token_used_after_send(&token_key).await; + } + Ok(()) } - /// Look up and include a tctoken in outgoing 1:1 message stanza nodes. + /// Look up and include a privacy token in outgoing 1:1 message stanza nodes. /// - /// If a valid (non-expired) token exists, adds a `` child node. - /// If the token is missing or expired, attempts to issue new tokens via IQ. - async fn maybe_include_tc_token(&self, to: &Jid, extra_nodes: &mut Vec) { + /// Follows WA Web's fallback chain (MsgCreateFanoutStanza.js): + /// 1. tctoken — from stored trusted contact token (if valid, non-expired) + /// 2. cstoken — HMAC-SHA256(nct_salt, recipient_lid) fallback for first-contact + /// 3. No token — message sent without token (server may return 463) + /// + /// Returns whether we should issue a new tc token after send, and the cache key + /// of the attached valid tc token when that token should be marked as used. + async fn maybe_include_tc_token( + &self, + to: &Jid, + extra_nodes: &mut Vec, + ) -> (bool, Option) { use wacore::iq::tctoken::{ - IssuePrivacyTokensSpec, build_tc_token_node, is_tc_token_expired, + build_cs_token_node, build_tc_token_node, compute_cs_token, is_tc_token_expired, should_send_new_tc_token, }; - use wacore::store::traits::TcTokenEntry; // Skip for own JID — no need to send privacy token to ourselves let snapshot = self.persistence_manager.get_device_snapshot().await; @@ -1091,83 +1128,131 @@ impl Client { .as_ref() .is_some_and(|lid| lid.is_same_user_as(to)); if is_self { - return; + return (false, None); } - // Resolve the destination to a LID for token lookup - let token_jid = if to.is_lid() { - to.user.clone() + // Resolve the destination to a LID user string once — reused for + // tctoken lookup, issuance, and cstoken HMAC input. + // Returns Some(lid_user) if resolved, None if no LID mapping exists. + let resolved_lid_user = if to.is_lid() { + Some(to.user.clone()) } else { - match self.lid_pn_cache.get_current_lid(&to.user).await { - Some(lid) => lid, - None => to.user.clone(), - } + self.lid_pn_cache.get_current_lid(&to.user).await }; + let token_jid = resolved_lid_user.as_deref().unwrap_or(&to.user).to_string(); let backend = self.persistence_manager.backend(); - // Look up existing token + // Look up existing tctoken let existing = match backend.get_tc_token(&token_jid).await { Ok(entry) => entry, Err(e) => { log::warn!(target: "Client/TcToken", "Failed to get tc_token for {}: {e}", token_jid); - return; + None } }; + let should_issue_after_send = + should_send_new_tc_token(existing.as_ref().and_then(|entry| entry.sender_timestamp)); + match existing { - Some(entry) if !is_tc_token_expired(entry.token_timestamp) => { - // Valid token — include it in the stanza + Some(entry) + if !is_tc_token_expired(entry.token_timestamp) && !entry.token.is_empty() => + { + // Valid tctoken — include it in the stanza extra_nodes.push(build_tc_token_node(&entry.token)); - - // Check if we should re-issue (bucket boundary crossed). - // Update sender_timestamp to mark we've sent our token in this bucket. - if should_send_new_tc_token(entry.sender_timestamp) { - let now = wacore::time::now_secs(); - let updated_entry = TcTokenEntry { - sender_timestamp: Some(now), - ..entry - }; - if let Err(e) = backend.put_tc_token(&token_jid, &updated_entry).await { - log::warn!(target: "Client/TcToken", "Failed to update sender_timestamp: {e}"); - } - } + return (should_issue_after_send, Some(token_jid)); } _ => { - // Token missing or expired — try to issue - let to_lid = self.resolve_to_lid_jid(to).await; - match self - .execute(IssuePrivacyTokensSpec::new(std::slice::from_ref(&to_lid))) - .await + if let Some(salt) = &snapshot.nct_salt + && let Some(lid_user) = &resolved_lid_user { - Ok(response) => { - let now = wacore::time::now_secs(); - for received in &response.tokens { - let entry = TcTokenEntry { - token: received.token.clone(), - token_timestamp: received.timestamp, - sender_timestamp: Some(now), - }; - - // Store the received token - let store_jid = received.jid.user.clone(); - if let Err(e) = backend.put_tc_token(&store_jid, &entry).await { - log::warn!(target: "Client/TcToken", "Failed to store issued tc_token: {e}"); - } - - // Include in message stanza - if !received.token.is_empty() { - extra_nodes.push(build_tc_token_node(&received.token)); - } - } - } - Err(e) => { - log::debug!(target: "Client/TcToken", "Failed to issue tc_token for {}: {e}", to_lid); - // Don't fail the message send — tctoken is optional - } + // HMAC input is "user@lid" (account LID without device suffix), + // matching WA Web's accountLid.toString() + let recipient_lid = wacore_binary::jid::Jid::new(lid_user, "lid").to_string(); + let cs_token = compute_cs_token(salt, &recipient_lid); + extra_nodes.push(build_cs_token_node(&cs_token)); + log::debug!(target: "Client/CsToken", "Attached cstoken for {} (NCT fallback)", to); + } else { + log::debug!(target: "Client/CsToken", "No tctoken or NCT salt/LID available for {}", to); } } } + + (should_issue_after_send, None) + } + + /// Returns `true` if the issuance IQ succeeded. + async fn issue_tc_token_after_send(&self, to: &Jid) -> bool { + use wacore::iq::tctoken::IssuePrivacyTokensSpec; + + let to_lid = self.resolve_to_lid_jid(to).await; + let Ok(response) = self + .execute(IssuePrivacyTokensSpec::new(std::slice::from_ref(&to_lid))) + .await + else { + log::debug!(target: "Client/TcToken", "Failed to issue tc_token for {}", to_lid); + return false; + }; + + self.store_issued_tc_tokens(&response.tokens).await; + true + } + + async fn store_issued_tc_tokens(&self, tokens: &[wacore::iq::tctoken::ReceivedTcToken]) { + use wacore::store::traits::TcTokenEntry; + + if tokens.is_empty() { + return; + } + + let backend = self.persistence_manager.backend(); + let now = wacore::time::now_secs(); + for received in tokens { + if received.token.is_empty() { + log::warn!(target: "Client/TcToken", "Server returned empty tc_token for {}, skipping", received.jid); + continue; + } + + let entry = TcTokenEntry { + token: received.token.clone(), + token_timestamp: received.timestamp, + sender_timestamp: Some(now), + }; + + let store_jid = received.jid.user.clone(); + if let Err(e) = backend.put_tc_token(&store_jid, &entry).await { + log::warn!(target: "Client/TcToken", "Failed to store issued tc_token: {e}"); + } + } + } + + async fn mark_tc_token_used_after_send(&self, token_key: &str) { + use wacore::store::traits::TcTokenEntry; + + let backend = self.persistence_manager.backend(); + let existing = match backend.get_tc_token(token_key).await { + Ok(entry) => entry, + Err(e) => { + log::warn!(target: "Client/TcToken", "Failed to reload tc_token for {}: {e}", token_key); + return; + } + }; + + let Some(entry) = existing else { + return; + }; + if entry.token.is_empty() { + return; + } + + let updated_entry = TcTokenEntry { + sender_timestamp: Some(wacore::time::now_secs()), + ..entry + }; + if let Err(e) = backend.put_tc_token(token_key, &updated_entry).await { + log::warn!(target: "Client/TcToken", "Failed to update sender_timestamp for {}: {e}", token_key); + } } /// Look up a valid (non-expired) tctoken for a JID. Returns the raw token bytes if found. @@ -1187,7 +1272,11 @@ impl Client { let backend = self.persistence_manager.backend(); match backend.get_tc_token(&token_jid).await { - Ok(Some(entry)) if !is_tc_token_expired(entry.token_timestamp) => Some(entry.token), + Ok(Some(entry)) + if !entry.token.is_empty() && !is_tc_token_expired(entry.token_timestamp) => + { + Some(entry.token) + } Ok(_) => None, Err(e) => { log::warn!(target: "Client/TcToken", "Failed to get tc_token for {}: {e}", token_jid); diff --git a/storages/sqlite-storage/migrations/2026-03-23-000000_add_nct_salt/down.sql b/storages/sqlite-storage/migrations/2026-03-23-000000_add_nct_salt/down.sql new file mode 100644 index 000000000..1b745aff5 --- /dev/null +++ b/storages/sqlite-storage/migrations/2026-03-23-000000_add_nct_salt/down.sql @@ -0,0 +1,52 @@ +-- Remove nct_salt column from device table + +CREATE TABLE device_backup AS SELECT + id, lid, pn, registration_id, noise_key, identity_key, + signed_pre_key, signed_pre_key_id, signed_pre_key_signature, + adv_secret_key, account, push_name, + app_version_primary, app_version_secondary, app_version_tertiary, + app_version_last_fetched_ms, edge_routing_info, props_hash, + next_pre_key_id +FROM device; + +DROP TABLE device; + +CREATE TABLE device ( + id INTEGER NOT NULL PRIMARY KEY, + lid TEXT NOT NULL DEFAULT '', + pn TEXT NOT NULL DEFAULT '', + registration_id INTEGER NOT NULL, + noise_key BLOB NOT NULL, + identity_key BLOB NOT NULL, + signed_pre_key BLOB NOT NULL, + signed_pre_key_id INTEGER NOT NULL, + signed_pre_key_signature BLOB NOT NULL, + adv_secret_key BLOB NOT NULL, + account BLOB, + push_name TEXT NOT NULL DEFAULT '', + app_version_primary INTEGER NOT NULL DEFAULT 0, + app_version_secondary INTEGER NOT NULL DEFAULT 0, + app_version_tertiary BIGINT NOT NULL DEFAULT 0, + app_version_last_fetched_ms BIGINT NOT NULL DEFAULT 0, + edge_routing_info BLOB, + props_hash TEXT, + next_pre_key_id INTEGER NOT NULL DEFAULT 0 +); + +INSERT INTO device ( + id, lid, pn, registration_id, noise_key, identity_key, + signed_pre_key, signed_pre_key_id, signed_pre_key_signature, + adv_secret_key, account, push_name, + app_version_primary, app_version_secondary, app_version_tertiary, + app_version_last_fetched_ms, edge_routing_info, props_hash, + next_pre_key_id +) SELECT + id, lid, pn, registration_id, noise_key, identity_key, + signed_pre_key, signed_pre_key_id, signed_pre_key_signature, + adv_secret_key, account, push_name, + app_version_primary, app_version_secondary, app_version_tertiary, + app_version_last_fetched_ms, edge_routing_info, props_hash, + next_pre_key_id +FROM device_backup; + +DROP TABLE device_backup; diff --git a/storages/sqlite-storage/migrations/2026-03-23-000000_add_nct_salt/up.sql b/storages/sqlite-storage/migrations/2026-03-23-000000_add_nct_salt/up.sql new file mode 100644 index 000000000..a8181d6ee --- /dev/null +++ b/storages/sqlite-storage/migrations/2026-03-23-000000_add_nct_salt/up.sql @@ -0,0 +1,4 @@ +-- Add NCT salt column to device table. +-- Server-provisioned salt for computing cstoken (HMAC-SHA256 fallback privacy token). +-- NULL means no salt has been provisioned yet. +ALTER TABLE device ADD COLUMN nct_salt BLOB; diff --git a/storages/sqlite-storage/src/schema.rs b/storages/sqlite-storage/src/schema.rs index 9da3a5204..b98b4d474 100644 --- a/storages/sqlite-storage/src/schema.rs +++ b/storages/sqlite-storage/src/schema.rs @@ -68,6 +68,7 @@ diesel::table! { edge_routing_info -> Nullable, props_hash -> Nullable, next_pre_key_id -> Integer, + nct_salt -> Nullable, } } diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 9c93b07cd..feb7fe390 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -80,6 +80,7 @@ struct DeviceRow { edge_routing_info: Option>, props_hash: Option, next_pre_key_id: i32, + nct_salt: Option>, } #[derive(Clone)] @@ -320,6 +321,7 @@ impl SqliteStore { let edge_routing_info = device_data.edge_routing_info.clone(); let props_hash = device_data.props_hash.clone(); let next_pre_key_id = device_data.next_pre_key_id as i32; + let nct_salt = device_data.nct_salt.clone(); let new_lid = device_data .lid .as_ref() @@ -357,6 +359,7 @@ impl SqliteStore { device::edge_routing_info.eq(edge_routing_info.clone()), device::props_hash.eq(props_hash.clone()), device::next_pre_key_id.eq(next_pre_key_id), + device::nct_salt.eq(nct_salt.clone()), )) .on_conflict(device::id) .do_update() @@ -379,6 +382,7 @@ impl SqliteStore { device::edge_routing_info.eq(edge_routing_info), device::props_hash.eq(props_hash), device::next_pre_key_id.eq(next_pre_key_id), + device::nct_salt.eq(nct_salt), )) .execute(&mut conn) .map_err(|e| StoreError::Database(e.to_string()))?; @@ -441,6 +445,7 @@ impl SqliteStore { device::edge_routing_info.eq(None::>), device::props_hash.eq(None::), device::next_pre_key_id.eq(new_device.next_pre_key_id as i32), + device::nct_salt.eq(None::>), )) .execute(&mut conn) .map_err(|e| StoreError::Database(e.to_string()))?; @@ -559,6 +564,8 @@ impl SqliteStore { edge_routing_info: row.edge_routing_info, props_hash: row.props_hash, next_pre_key_id: row.next_pre_key_id as u32, + nct_salt: row.nct_salt, + nct_salt_sync_seen: false, })) } else { Ok(None) diff --git a/tests/e2e/Cargo.toml b/tests/e2e/Cargo.toml index 57386496f..bbbb59259 100644 --- a/tests/e2e/Cargo.toml +++ b/tests/e2e/Cargo.toml @@ -13,9 +13,11 @@ dhat-heap = ["dep:dhat"] [dependencies] anyhow = { workspace = true } dhat = { version = "0.3", optional = true } +futures = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] } uuid = { workspace = true, features = ["v4"] } wacore = { path = "../../wacore" } +wacore-binary = { path = "../../wacore/binary" } whatsapp-rust = { path = "../..", features = ["danger-skip-tls-verify", "debug-diagnostics"] } whatsapp-rust-sqlite-storage = { path = "../../storages/sqlite-storage" } whatsapp-rust-tokio-transport = { path = "../../transports/tokio-transport", features = [ diff --git a/tests/e2e/src/lib.rs b/tests/e2e/src/lib.rs index 1ffda52bf..a447685f9 100644 --- a/tests/e2e/src/lib.rs +++ b/tests/e2e/src/lib.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use wacore::store::traits::TcTokenEntry; use wacore::types::events::{Event, EventHandler}; +use wacore_binary::node::Node; use whatsapp_rust::Jid; use whatsapp_rust::bot::Bot; use whatsapp_rust::store::traits::Backend; @@ -25,6 +26,22 @@ pub fn mock_server_url() -> String { std::env::var("MOCK_SERVER_URL").unwrap_or_else(|_| "wss://127.0.0.1:8080/ws/chat".to_string()) } +pub fn unique_push_name(prefix: &str) -> String { + format!("{}_{}", prefix, uuid::Uuid::new_v4()) +} + +pub fn restricted_push_name(prefix: &str) -> String { + format!("restricted:{}", unique_push_name(prefix)) +} + +pub fn scenario_push_name(prefix: &str, flags: &[&str]) -> String { + assert!( + !flags.is_empty(), + "scenario_push_name requires at least one flag" + ); + format!("scenario:{}:{}", flags.join(","), unique_push_name(prefix)) +} + /// Event handler that sends events to a tokio broadcast channel for test assertions. pub struct ChannelEventHandler { tx: tokio::sync::broadcast::Sender, @@ -213,6 +230,44 @@ impl TestClient { } } + pub fn sent_message_waiter( + &self, + msg_id: &str, + ) -> futures::channel::oneshot::Receiver> { + self.client + .wait_for_sent_node(whatsapp_rust::NodeFilter::tag("message").attr("id", msg_id)) + } + + pub fn next_sent_message_waiter(&self) -> futures::channel::oneshot::Receiver> { + self.client + .wait_for_sent_node(whatsapp_rust::NodeFilter::tag("message")) + } + + pub async fn nct_salt(&self) -> Option> { + self.client + .persistence_manager() + .get_device_snapshot() + .await + .nct_salt + .clone() + } + + pub async fn wait_for_nct_salt(&self, timeout_secs: u64) -> anyhow::Result> { + let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(timeout_secs); + + loop { + if let Some(salt) = self.nct_salt().await { + return Ok(salt); + } + + if tokio::time::Instant::now() >= deadline { + return Err(anyhow::anyhow!("Timed out waiting for NCT salt")); + } + + tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; + } + } + // ── Event waiting ─────────────────────────────────────────────────────── /// Wait for an event matching the predicate, with a timeout in seconds. diff --git a/tests/e2e/tests/privacy_tokens.rs b/tests/e2e/tests/privacy_tokens.rs index ceae6628b..686428228 100644 --- a/tests/e2e/tests/privacy_tokens.rs +++ b/tests/e2e/tests/privacy_tokens.rs @@ -1,11 +1,86 @@ -use e2e_tests::{TestClient, send_and_expect_text, text_msg}; +use e2e_tests::{ + TestClient, restricted_push_name, scenario_push_name, send_and_expect_text, text_msg, +}; use log::info; +use std::sync::Arc; use wacore::iq::tctoken::tc_token_expiration_cutoff; use wacore::store::traits::TcTokenEntry; +use wacore::types::events::Event; +use wacore_binary::node::Node; use whatsapp_rust::{NodeFilter, SendOptions}; -fn unique_push_name(prefix: &str) -> String { - format!("{}_{}", prefix, uuid::Uuid::new_v4()) +fn has_child(node: &Node, tag: &str) -> bool { + node.children() + .map(|children| children.iter().any(|child| child.tag == tag)) + .unwrap_or(false) +} + +fn has_descendant(node: &Node, tag: &str) -> bool { + node.children().is_some_and(|children| { + children + .iter() + .any(|child| child.tag == tag || has_descendant(child, tag)) + }) +} + +async fn send_first_message_and_expect_463( + sender: &TestClient, + recipient: &mut TestClient, + recipient_jid: &whatsapp_rust::Jid, + text: &str, +) -> anyhow::Result> { + let msg_id = format!("E2E463{}", uuid::Uuid::new_v4().simple()); + send_message_and_expect_463_with_id(sender, recipient, recipient_jid, text, msg_id).await +} + +async fn send_message_and_expect_463_with_id( + sender: &TestClient, + recipient: &mut TestClient, + recipient_jid: &whatsapp_rust::Jid, + text: &str, + msg_id: String, +) -> anyhow::Result> { + let waiter = sender.client.wait_for_node( + NodeFilter::tag("ack") + .attr("id", msg_id.clone()) + .attr("class", "message") + .attr("from", recipient_jid.to_string()) + .attr("error", "463"), + ); + + let returned_id = sender + .client + .send_message_with_options( + recipient_jid.clone(), + text_msg(text), + SendOptions { + message_id: Some(msg_id.clone()), + ..Default::default() + }, + ) + .await?; + assert_eq!(returned_id, msg_id); + + let ack = tokio::time::timeout(tokio::time::Duration::from_secs(15), waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for 463 nack"))? + .map_err(|_| anyhow::anyhow!("463 nack waiter was canceled"))?; + assert_eq!(ack.tag, "ack"); + assert_eq!( + ack.attrs.get("error").map(|v| v.to_string()), + Some("463".to_string()) + ); + + let expected_text = text.to_string(); + recipient + .assert_no_event( + 5, + move |e| matches!(e, wacore::types::events::Event::Message(msg, _) if msg.conversation.as_deref() == Some(expected_text.as_str())), + "restricted recipient should not receive first-contact message without privacy token", + ) + .await?; + + Ok(ack) } #[tokio::test] @@ -77,7 +152,7 @@ async fn test_issue_tokens_api_delivers_notification_and_updates_index() -> anyh async fn test_reply_to_restricted_contact_uses_received_tc_token() -> anyhow::Result<()> { let _ = env_logger::builder().is_test(true).try_init(); - let restricted_name = format!("restricted:{}", unique_push_name("e2e_tctok_reply_a")); + let restricted_name = restricted_push_name("e2e_tctok_reply_a"); let mut client_a = TestClient::connect_as("e2e_tctok_reply_a", &restricted_name).await?; let mut client_b = TestClient::connect("e2e_tctok_reply_b").await?; @@ -120,53 +195,18 @@ async fn test_reply_to_restricted_contact_uses_received_tc_token() -> anyhow::Re async fn test_first_message_to_restricted_contact_receives_463_nack() -> anyhow::Result<()> { let _ = env_logger::builder().is_test(true).try_init(); - let restricted_name = format!("restricted:{}", unique_push_name("e2e_tctok_463_a")); + let restricted_name = restricted_push_name("e2e_tctok_463_a"); let mut client_a = TestClient::connect_as("e2e_tctok_463_a", &restricted_name).await?; let client_b = TestClient::connect("e2e_tctok_463_b").await?; let jid_a = client_a.jid().await; - let msg_id = format!("E2E463{}", uuid::Uuid::new_v4().simple()); - let waiter = client_b.client.wait_for_node( - NodeFilter::tag("ack") - .attr("id", msg_id.clone()) - .attr("class", "message") - .attr("from", jid_a.to_string()) - .attr("error", "463"), - ); - - let returned_id = client_b - .client - .send_message_with_options( - jid_a.clone(), - text_msg("first contact to restricted account"), - SendOptions { - message_id: Some(msg_id.clone()), - ..Default::default() - }, - ) - .await?; - assert_eq!( - returned_id, msg_id, - "send should preserve caller-provided message ID" - ); - - let ack = tokio::time::timeout(tokio::time::Duration::from_secs(15), waiter) - .await - .map_err(|_| anyhow::anyhow!("Timed out waiting for 463 nack"))? - .map_err(|_| anyhow::anyhow!("463 nack waiter was canceled"))?; - assert_eq!(ack.tag, "ack"); - assert_eq!( - ack.attrs.get("error").map(|v| v.to_string()), - Some("463".to_string()) - ); - - client_a - .assert_no_event( - 5, - |e| matches!(e, wacore::types::events::Event::Message(msg, _) if msg.conversation.as_deref() == Some("first contact to restricted account")), - "restricted recipient should not receive first-contact message without tcToken", - ) - .await?; + send_first_message_and_expect_463( + &client_b, + &mut client_a, + &jid_a, + "first contact to restricted account", + ) + .await?; client_a.disconnect().await; client_b.disconnect().await; @@ -177,8 +217,8 @@ async fn test_first_message_to_restricted_contact_receives_463_nack() -> anyhow: async fn test_tc_token_notification_reaches_all_connected_devices() -> anyhow::Result<()> { let _ = env_logger::builder().is_test(true).try_init(); - let restricted_name = format!("restricted:{}", unique_push_name("e2e_tctok_multi_a")); - let shared_b_name = unique_push_name("e2e_tctok_multi_b"); + let restricted_name = restricted_push_name("e2e_tctok_multi_a"); + let shared_b_name = e2e_tests::unique_push_name("e2e_tctok_multi_b"); let client_a = TestClient::connect_as("e2e_tctok_multi_a", &restricted_name).await?; let mut client_b1 = TestClient::connect_as("e2e_tctok_multi_b1", &shared_b_name).await?; @@ -221,7 +261,7 @@ async fn test_tc_token_notification_reaches_all_connected_devices() -> anyhow::R async fn test_tc_token_survives_reconnect() -> anyhow::Result<()> { let _ = env_logger::builder().is_test(true).try_init(); - let restricted_name = format!("restricted:{}", unique_push_name("e2e_tctok_recon_a")); + let restricted_name = restricted_push_name("e2e_tctok_recon_a"); let mut client_a = TestClient::connect_as("e2e_tctok_recon_a", &restricted_name).await?; let mut client_b = TestClient::connect("e2e_tctok_recon_b").await?; @@ -310,3 +350,746 @@ async fn test_prune_expired_tc_tokens_removes_only_stale_entries() -> anyhow::Re client.disconnect().await; Ok(()) } + +#[tokio::test] +async fn test_only_nct_send_ab_without_salt_still_receives_463() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = restricted_push_name("e2e_cstok_send_only_a"); + let sender_name = scenario_push_name( + "e2e_cstok_send_only_b", + &[ + "nct_send_ab=1", + "nct_history_delivery=0", + "nct_syncd_delivery=0", + ], + ); + let mut client_a = TestClient::connect_as("e2e_cstok_send_only_a", &restricted_name).await?; + let client_b = TestClient::connect_as("e2e_cstok_send_only_b", &sender_name).await?; + + assert!( + client_b.nct_salt().await.is_none(), + "send AB alone should not create NCT salt" + ); + + let key_a = client_a.tc_token_key().await?; + assert!( + client_b.client.tc_token().get(&key_a).await?.is_none(), + "sender should not have a tc token for recipient before first contact" + ); + + let jid_a_lid = client_a + .client + .get_lid() + .await + .expect("restricted recipient should have a LID"); + let msg_id = format!("E2ECSNEG1{}", uuid::Uuid::new_v4().simple()); + let sent_msg_id = msg_id.clone(); + let sent_waiter = client_b.sent_message_waiter(&sent_msg_id); + send_message_and_expect_463_with_id( + &client_b, + &mut client_a, + &jid_a_lid, + "send-ab-only first contact", + msg_id, + ) + .await?; + let sent = tokio::time::timeout(tokio::time::Duration::from_secs(10), sent_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for sent message node"))? + .map_err(|_| anyhow::anyhow!("sent message waiter was canceled"))?; + assert_eq!( + sent.attrs.get("to").map(|v| v.to_string()), + Some(jid_a_lid.to_string()) + ); + assert!(!has_child(&sent, "tctoken")); + assert!(!has_child(&sent, "cstoken")); + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_send_and_syncd_ab_without_delivery_still_receives_463() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = restricted_push_name("e2e_cstok_ab_only_a"); + let sender_name = scenario_push_name( + "e2e_cstok_ab_only_b", + &[ + "nct_send_ab=1", + "nct_syncd_ab=1", + "nct_history_delivery=0", + "nct_syncd_delivery=0", + ], + ); + let mut client_a = TestClient::connect_as("e2e_cstok_ab_only_a", &restricted_name).await?; + let client_b = TestClient::connect_as("e2e_cstok_ab_only_b", &sender_name).await?; + + assert!( + client_b.nct_salt().await.is_none(), + "AB props without delivery should still leave NCT salt unset" + ); + + let key_a = client_a.tc_token_key().await?; + assert!( + client_b.client.tc_token().get(&key_a).await?.is_none(), + "sender should not have a tc token for recipient before first contact" + ); + + let jid_a_lid = client_a + .client + .get_lid() + .await + .expect("restricted recipient should have a LID"); + let msg_id = format!("E2ECSNEG2{}", uuid::Uuid::new_v4().simple()); + let sent_msg_id = msg_id.clone(); + let sent_waiter = client_b.sent_message_waiter(&sent_msg_id); + send_message_and_expect_463_with_id( + &client_b, + &mut client_a, + &jid_a_lid, + "send-and-syncd-ab first contact", + msg_id, + ) + .await?; + let sent = tokio::time::timeout(tokio::time::Duration::from_secs(10), sent_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for sent message node"))? + .map_err(|_| anyhow::anyhow!("sent message waiter was canceled"))?; + assert_eq!( + sent.attrs.get("to").map(|v| v.to_string()), + Some(jid_a_lid.to_string()) + ); + assert!(!has_child(&sent, "tctoken")); + assert!(!has_child(&sent, "cstoken")); + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_history_sync_nct_salt_enables_cstoken_first_contact() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = restricted_push_name("e2e_cstok_hist_a"); + let sender_name = scenario_push_name( + "e2e_cstok_hist_b", + &[ + "nct_send_ab=1", + "nct_history_ab=1", + "nct_history_delivery=1", + "nct_syncd_delivery=0", + ], + ); + let mut client_a = TestClient::connect_as("e2e_cstok_hist_a", &restricted_name).await?; + let client_b = TestClient::connect_as("e2e_cstok_hist_b", &sender_name).await?; + + let nct_salt = client_b.wait_for_nct_salt(10).await?; + assert!( + !nct_salt.is_empty(), + "history sync should provision NCT salt" + ); + + let key_a = client_a.tc_token_key().await?; + assert!( + client_b.client.tc_token().get(&key_a).await?.is_none(), + "sender should not have a tc token for recipient before cstoken fallback" + ); + + let jid_a_lid = client_a + .client + .get_lid() + .await + .expect("restricted recipient should have a LID"); + let sent_waiter = client_b.next_sent_message_waiter(); + client_b + .client + .send_message_with_options( + jid_a_lid, + text_msg("history-sync cstoken first contact"), + SendOptions { + message_id: Some(format!("E2ECSHIST{}", uuid::Uuid::new_v4().simple())), + ..Default::default() + }, + ) + .await?; + let sent = tokio::time::timeout(tokio::time::Duration::from_secs(10), sent_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for next sent message node"))? + .map_err(|_| anyhow::anyhow!("sent message waiter was canceled"))?; + assert!(has_child(&sent, "cstoken")); + assert!(!has_child(&sent, "tctoken")); + client_a + .wait_for_text("history-sync cstoken first contact", 30) + .await?; + + assert!( + client_b.client.tc_token().get(&key_a).await?.is_none(), + "cstoken fallback should not materialize a tc token entry for the recipient" + ); + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_cstoken_only_first_contact_succeeds_when_tctoken_disabled() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = scenario_push_name( + "e2e_cstok_only_a", + &["restricted", "tc_enabled=0", "cs_enabled=1"], + ); + let sender_name = scenario_push_name( + "e2e_cstok_only_b", + &[ + "nct_send_ab=1", + "nct_history_ab=1", + "nct_history_delivery=1", + "nct_syncd_delivery=0", + ], + ); + let mut client_a = TestClient::connect_as("e2e_cstok_only_a", &restricted_name).await?; + let client_b = TestClient::connect_as("e2e_cstok_only_b", &sender_name).await?; + + let nct_salt = client_b.wait_for_nct_salt(10).await?; + assert!( + !nct_salt.is_empty(), + "history sync should provision NCT salt" + ); + + let key_a = client_a.tc_token_key().await?; + assert!( + client_b.client.tc_token().get(&key_a).await?.is_none(), + "sender should not have a tc token for recipient before first contact" + ); + + let jid_a_lid = client_a + .client + .get_lid() + .await + .expect("restricted recipient should have a LID"); + let sent_waiter = client_b.next_sent_message_waiter(); + client_b + .client + .send_message_with_options( + jid_a_lid, + text_msg("cstoken-only first contact"), + SendOptions { + message_id: Some(format!("E2ECSONLY{}", uuid::Uuid::new_v4().simple())), + ..Default::default() + }, + ) + .await?; + let sent = tokio::time::timeout(tokio::time::Duration::from_secs(10), sent_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for next sent message node"))? + .map_err(|_| anyhow::anyhow!("sent message waiter was canceled"))?; + assert!(has_child(&sent, "cstoken")); + assert!(!has_child(&sent, "tctoken")); + client_a + .wait_for_text("cstoken-only first contact", 30) + .await?; + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_syncd_nct_salt_enables_cstoken_first_contact() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = restricted_push_name("e2e_cstok_syncd_a"); + let sender_name = scenario_push_name( + "e2e_cstok_syncd_b", + &[ + "nct_send_ab=1", + "nct_syncd_ab=1", + "nct_history_delivery=0", + "nct_syncd_delivery=1", + ], + ); + let mut client_a = TestClient::connect_as("e2e_cstok_syncd_a", &restricted_name).await?; + let client_b = TestClient::connect_as("e2e_cstok_syncd_b", &sender_name).await?; + + let nct_salt = client_b.wait_for_nct_salt(10).await?; + assert!( + !nct_salt.is_empty(), + "app state sync should provision NCT salt" + ); + + let key_a = client_a.tc_token_key().await?; + assert!( + client_b.client.tc_token().get(&key_a).await?.is_none(), + "sender should not have a tc token for recipient before cstoken fallback" + ); + + let jid_a_lid = client_a + .client + .get_lid() + .await + .expect("restricted recipient should have a LID"); + let sent_waiter = client_b.next_sent_message_waiter(); + client_b + .client + .send_message_with_options( + jid_a_lid, + text_msg("syncd cstoken first contact"), + SendOptions { + message_id: Some(format!("E2ECSSYN{}", uuid::Uuid::new_v4().simple())), + ..Default::default() + }, + ) + .await?; + let sent = tokio::time::timeout(tokio::time::Duration::from_secs(10), sent_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for next sent message node"))? + .map_err(|_| anyhow::anyhow!("sent message waiter was canceled"))?; + assert!(has_child(&sent, "cstoken")); + assert!(!has_child(&sent, "tctoken")); + client_a + .wait_for_text("syncd cstoken first contact", 30) + .await?; + + assert!( + client_b.client.tc_token().get(&key_a).await?.is_none(), + "cstoken fallback should not materialize a tc token entry for the recipient" + ); + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_clearing_nct_salt_locally_makes_first_contact_fail_again() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = scenario_push_name( + "e2e_cstok_remove_a", + &["restricted", "tc_enabled=0", "cs_enabled=1"], + ); + let sender_name = scenario_push_name( + "e2e_cstok_remove_b", + &[ + "nct_send_ab=1", + "nct_syncd_ab=1", + "nct_history_delivery=0", + "nct_syncd_delivery=1", + ], + ); + let mut client_a = TestClient::connect_as("e2e_cstok_remove_a", &restricted_name).await?; + let client_b = TestClient::connect_as("e2e_cstok_remove_b", &sender_name).await?; + + let initial_salt = client_b.wait_for_nct_salt(10).await?; + assert!(!initial_salt.is_empty(), "syncd should provision NCT salt"); + + let jid_a_lid = client_a + .client + .get_lid() + .await + .expect("restricted recipient should have a LID"); + client_b + .client + .send_message(jid_a_lid, text_msg("remove-salt initial success")) + .await?; + client_a + .wait_for_text("remove-salt initial success", 30) + .await?; + + client_b + .client + .persistence_manager() + .process_command(whatsapp_rust::store::commands::DeviceCommand::SetNctSalt( + None, + )) + .await; + assert!( + client_b.nct_salt().await.is_none(), + "NCT salt should be cleared locally" + ); + + let restricted_name_c = scenario_push_name( + "e2e_cstok_remove_c", + &["restricted", "tc_enabled=0", "cs_enabled=1"], + ); + let mut client_c = TestClient::connect_as("e2e_cstok_remove_c", &restricted_name_c).await?; + let jid_c_lid = client_c + .client + .get_lid() + .await + .expect("restricted recipient should have a LID"); + send_first_message_and_expect_463( + &client_b, + &mut client_c, + &jid_c_lid, + "remove-salt first contact should fail", + ) + .await?; + + client_a.disconnect().await; + client_b.disconnect().await; + client_c.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_tctoken_only_reply_succeeds_when_cstoken_disabled() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = scenario_push_name( + "e2e_tctok_only_a", + &["restricted", "tc_enabled=1", "cs_enabled=0"], + ); + let mut client_a = TestClient::connect_as("e2e_tctok_only_a", &restricted_name).await?; + let mut client_b = TestClient::connect("e2e_tctok_only_b").await?; + + let jid_a = client_a.jid().await; + let jid_b = client_b.jid().await; + + send_and_expect_text( + &client_a.client, + &mut client_b, + &jid_b, + "seed tctoken-only reply path", + 30, + ) + .await?; + + let key_a = client_a.tc_token_key().await?; + let initial_entry = client_b.wait_for_tc_token(&key_a, 10).await?; + assert!( + !initial_entry.token.is_empty(), + "sender should have a valid tc token before reply" + ); + + let sent_waiter = client_b.next_sent_message_waiter(); + client_b + .client + .send_message_with_options( + jid_a.clone(), + text_msg("tctoken-only reply"), + SendOptions { + message_id: Some(format!("E2ETCONLY{}", uuid::Uuid::new_v4().simple())), + ..Default::default() + }, + ) + .await?; + let sent = tokio::time::timeout(tokio::time::Duration::from_secs(10), sent_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for next sent message node"))? + .map_err(|_| anyhow::anyhow!("sent message waiter was canceled"))?; + assert!(has_child(&sent, "tctoken")); + assert!(!has_child(&sent, "cstoken")); + client_a.wait_for_text("tctoken-only reply", 30).await?; + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_nct_salt_survives_reconnect_and_still_allows_first_contact() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = restricted_push_name("e2e_cstok_recon_a"); + let sender_name = scenario_push_name( + "e2e_cstok_recon_b", + &[ + "nct_send_ab=1", + "nct_syncd_ab=1", + "nct_history_delivery=0", + "nct_syncd_delivery=1", + ], + ); + let mut client_a = TestClient::connect_as("e2e_cstok_recon_a", &restricted_name).await?; + let mut client_b = TestClient::connect_as("e2e_cstok_recon_b", &sender_name).await?; + + let before_reconnect = client_b.wait_for_nct_salt(10).await?; + client_b.reconnect_and_wait().await?; + client_b + .client + .wait_for_startup_sync(tokio::time::Duration::from_secs(15)) + .await?; + + let after_reconnect = client_b.wait_for_nct_salt(5).await?; + assert_eq!( + after_reconnect, before_reconnect, + "NCT salt should survive reconnect" + ); + + let key_a = client_a.tc_token_key().await?; + assert!( + client_b.client.tc_token().get(&key_a).await?.is_none(), + "sender should still have no tc token for recipient before first contact" + ); + + let jid_a_lid = client_a + .client + .get_lid() + .await + .expect("restricted recipient should have a LID"); + let sent_waiter = client_b.next_sent_message_waiter(); + client_b + .client + .send_message_with_options( + jid_a_lid, + text_msg("reconnect cstoken first contact"), + SendOptions { + message_id: Some(format!("E2ECSRECON{}", uuid::Uuid::new_v4().simple())), + ..Default::default() + }, + ) + .await?; + let sent = tokio::time::timeout(tokio::time::Duration::from_secs(10), sent_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for next sent message node"))? + .map_err(|_| anyhow::anyhow!("sent message waiter was canceled"))?; + assert!(has_child(&sent, "cstoken")); + assert!(!has_child(&sent, "tctoken")); + client_a + .wait_for_text("reconnect cstoken first contact", 30) + .await?; + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_pn_target_first_contact_uses_cstoken_after_lid_resolution() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = scenario_push_name( + "e2e_cstok_pn_a", + &["restricted", "tc_enabled=0", "cs_enabled=1"], + ); + let sender_name = scenario_push_name( + "e2e_cstok_pn_b", + &[ + "nct_send_ab=1", + "nct_history_ab=1", + "nct_history_delivery=1", + "nct_syncd_delivery=0", + ], + ); + let mut client_a = TestClient::connect_as("e2e_cstok_pn_a", &restricted_name).await?; + let client_b = TestClient::connect_as("e2e_cstok_pn_b", &sender_name).await?; + + let nct_salt = client_b.wait_for_nct_salt(10).await?; + assert!( + !nct_salt.is_empty(), + "history sync should provision NCT salt" + ); + + let jid_a_pn = client_a.jid().await; + let key_a = client_a.tc_token_key().await?; + assert!( + client_b.client.tc_token().get(&key_a).await?.is_none(), + "sender should not have a tc token for recipient before first contact" + ); + + let user_info = client_b + .client + .contacts() + .get_user_info(std::slice::from_ref(&jid_a_pn)) + .await?; + let resolved = user_info + .get(&jid_a_pn) + .expect("PN-target usync info should exist"); + assert!( + resolved.lid.is_some(), + "PN-target usync info should carry the recipient account LID" + ); + + let msg_id = format!("E2ECSPN{}", uuid::Uuid::new_v4().simple()); + let sent_waiter = client_b.sent_message_waiter(&msg_id); + client_b + .client + .send_message_with_options( + jid_a_pn.clone(), + text_msg("pn-target cstoken first contact"), + SendOptions { + message_id: Some(msg_id), + ..Default::default() + }, + ) + .await?; + let sent = tokio::time::timeout(tokio::time::Duration::from_secs(10), sent_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for PN-target sent message node"))? + .map_err(|_| anyhow::anyhow!("PN-target sent message waiter was canceled"))?; + assert_eq!( + sent.attrs.get("to").map(|v| v.to_string()), + Some(jid_a_pn.to_string()) + ); + assert!(has_child(&sent, "cstoken")); + assert!(!has_child(&sent, "tctoken")); + + client_a + .wait_for_text("pn-target cstoken first contact", 30) + .await?; + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_restricted_profile_picture_requires_tctoken() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = restricted_push_name("e2e_ppic_priv_a"); + let client_a = TestClient::connect_as("e2e_ppic_priv_a", &restricted_name).await?; + let mut client_b = TestClient::connect("e2e_ppic_priv_b").await?; + + client_a + .client + .profile() + .set_profile_picture(vec![0xFF, 0xD8, 0xFF, 0xE0, 0x00, 0x10]) + .await?; + + let jid_a = client_a.jid().await; + let jid_b = client_b.jid().await; + + let denied_waiter = client_b.client.wait_for_sent_node( + NodeFilter::tag("iq") + .attr("type", "get") + .attr("xmlns", "w:profile:picture") + .attr("target", jid_a.to_string()), + ); + let denied = client_b + .client + .contacts() + .get_profile_picture(&jid_a, false) + .await?; + let denied_node = tokio::time::timeout(tokio::time::Duration::from_secs(10), denied_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for denied profile picture IQ"))? + .map_err(|_| anyhow::anyhow!("denied profile picture waiter was canceled"))?; + assert!(!has_descendant(&denied_node, "tctoken")); + assert!(!has_descendant(&denied_node, "cstoken")); + assert!( + denied.is_none(), + "restricted profile picture should be hidden without a tc token" + ); + + send_and_expect_text( + &client_a.client, + &mut client_b, + &jid_b, + "seed profile picture tc token", + 30, + ) + .await?; + let key_a = client_a.tc_token_key().await?; + client_b.wait_for_tc_token(&key_a, 10).await?; + + let allowed_waiter = client_b.client.wait_for_sent_node( + NodeFilter::tag("iq") + .attr("type", "get") + .attr("xmlns", "w:profile:picture") + .attr("target", jid_a.to_string()), + ); + let allowed = client_b + .client + .contacts() + .get_profile_picture(&jid_a, false) + .await?; + let allowed_node = tokio::time::timeout(tokio::time::Duration::from_secs(10), allowed_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for allowed profile picture IQ"))? + .map_err(|_| anyhow::anyhow!("allowed profile picture waiter was canceled"))?; + assert!(has_descendant(&allowed_node, "tctoken")); + assert!(!has_descendant(&allowed_node, "cstoken")); + assert!( + allowed.is_some(), + "restricted profile picture should be visible once tc token exists" + ); + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} + +#[tokio::test] +async fn test_restricted_presence_subscribe_requires_tctoken() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let restricted_name = restricted_push_name("e2e_presence_priv_a"); + let client_a = TestClient::connect_as("e2e_presence_priv_a", &restricted_name).await?; + let mut client_b = TestClient::connect("e2e_presence_priv_b").await?; + + let jid_a = client_a.jid().await; + let jid_b = client_b.jid().await; + + let denied_waiter = client_b.client.wait_for_sent_node( + NodeFilter::tag("presence") + .attr("type", "subscribe") + .attr("to", jid_a.to_string()), + ); + client_b.client.presence().subscribe(&jid_a).await?; + let denied_node = tokio::time::timeout(tokio::time::Duration::from_secs(10), denied_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for denied presence subscribe"))? + .map_err(|_| anyhow::anyhow!("denied presence subscribe waiter was canceled"))?; + assert!(!has_child(&denied_node, "tctoken")); + assert!(!has_child(&denied_node, "cstoken")); + + client_a.client.presence().set_unavailable().await?; + client_b + .assert_no_event( + 5, + |e| matches!(e, Event::Presence(update) if update.from == jid_a && update.unavailable), + "restricted presence subscribe without tctoken should not deliver updates", + ) + .await?; + + send_and_expect_text( + &client_a.client, + &mut client_b, + &jid_b, + "seed presence tc token", + 30, + ) + .await?; + let key_a = client_a.tc_token_key().await?; + client_b.wait_for_tc_token(&key_a, 10).await?; + + let allowed_waiter = client_b.client.wait_for_sent_node( + NodeFilter::tag("presence") + .attr("type", "subscribe") + .attr("to", jid_a.to_string()), + ); + client_b.client.presence().subscribe(&jid_a).await?; + let allowed_node = tokio::time::timeout(tokio::time::Duration::from_secs(10), allowed_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for allowed presence subscribe"))? + .map_err(|_| anyhow::anyhow!("allowed presence subscribe waiter was canceled"))?; + assert!(has_child(&allowed_node, "tctoken")); + assert!(!has_child(&allowed_node, "cstoken")); + + let _ = client_b + .wait_for_event( + 5, + |e| matches!(e, Event::Presence(update) if update.from == jid_a && update.unavailable), + ) + .await?; + + client_a.client.presence().set_available().await?; + let _ = client_b + .wait_for_event( + 10, + |e| matches!(e, Event::Presence(update) if update.from == jid_a && !update.unavailable), + ) + .await?; + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} diff --git a/wacore/src/history_sync.rs b/wacore/src/history_sync.rs index 303b0d083..946527879 100644 --- a/wacore/src/history_sync.rs +++ b/wacore/src/history_sync.rs @@ -18,7 +18,10 @@ pub enum HistorySyncError { #[derive(Debug, Default)] pub struct HistorySyncResult { pub own_pushname: Option, - + /// NCT salt from HistorySync field 19 (nctSalt). + /// Delivered during initial pairing so cstoken is available immediately. + /// Source: WAWeb/History/MsgHandlerAction.js:storeNctSaltFromHistorySync + pub nct_salt: Option>, pub conversations_processed: usize, } @@ -111,6 +114,21 @@ where pos = end; } + // field 19 = nctSalt (optional bytes, length-delimited) + // Delivered during initial pairing so cstoken is available immediately. + // Source: storeNctSaltFromHistorySync in WAWeb/History/MsgHandlerAction.js + 19 if wire_type_raw == wire_type::LENGTH_DELIMITED => { + let (len, vlen) = read_varint(&buf[pos..])?; + pos += vlen; + let end = checked_end(pos, len, buf.len(), "nctSalt")?; + + let salt = buf[pos..end].to_vec(); + if !salt.is_empty() { + result.nct_salt = Some(salt); + } + pos = end; + } + _ => { pos = skip_field(wire_type_raw, &buf, pos)?; } @@ -188,3 +206,69 @@ fn skip_field(wire_type: u32, buf: &[u8], pos: usize) -> Result Vec { + let proto_bytes = hs.encode_to_vec(); + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&proto_bytes).unwrap(); + encoder.finish().unwrap() + } + + #[test] + fn test_nct_salt_extracted_from_history_sync() { + let salt = vec![0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]; + let hs = wa::HistorySync { + sync_type: wa::history_sync::HistorySyncType::InitialBootstrap as i32, + nct_salt: Some(salt.clone()), + ..Default::default() + }; + + let compressed = encode_and_compress(&hs); + let result = process_history_sync::(compressed, None, None, None).unwrap(); + + assert_eq!(result.nct_salt, Some(salt)); + } + + #[test] + fn test_nct_salt_none_when_absent() { + let hs = wa::HistorySync { + sync_type: wa::history_sync::HistorySyncType::InitialBootstrap as i32, + ..Default::default() + }; + + let compressed = encode_and_compress(&hs); + let result = process_history_sync::(compressed, None, None, None).unwrap(); + + assert!(result.nct_salt.is_none()); + } + + #[test] + fn test_nct_salt_and_pushname_coexist() { + let salt = vec![0x01, 0x02, 0x03]; + let hs = wa::HistorySync { + sync_type: wa::history_sync::HistorySyncType::InitialBootstrap as i32, + nct_salt: Some(salt.clone()), + pushnames: vec![wa::Pushname { + id: Some("0000000000".into()), + pushname: Some("TestUser".into()), + }], + ..Default::default() + }; + + let compressed = encode_and_compress(&hs); + let result = + process_history_sync::(compressed, Some("0000000000"), None, None).unwrap(); + + assert_eq!(result.nct_salt, Some(salt)); + assert_eq!(result.own_pushname.as_deref(), Some("TestUser")); + } +} diff --git a/wacore/src/iq/tctoken.rs b/wacore/src/iq/tctoken.rs index c3c3a6f4f..2bcb5eadf 100644 --- a/wacore/src/iq/tctoken.rs +++ b/wacore/src/iq/tctoken.rs @@ -249,6 +249,27 @@ pub fn parse_privacy_token_notification( Ok(tokens) } +/// Compute a cstoken (client-side token / NCT) for a recipient. +/// +/// This is the fallback token used when no tctoken exists for the recipient. +/// Matches WA Web: `genCsTokenBody` in `MsgCreateFanoutStanza.js`. +/// +/// `salt` — NCT salt from app state sync (raw bytes, not base64). +/// `recipient_lid` — The recipient's bare account LID string (e.g. `"12345@lid"`). +pub fn compute_cs_token(salt: &[u8], recipient_lid: &str) -> Vec { + use hmac::{Hmac, Mac}; + use sha2::Sha256; + + let mut mac = Hmac::::new_from_slice(salt).expect("HMAC-SHA256 accepts any key length"); + mac.update(recipient_lid.as_bytes()); + mac.finalize().into_bytes().to_vec() +} + +/// Build a `` stanza child for including in outgoing messages. +pub fn build_cs_token_node(token: &[u8]) -> Node { + NodeBuilder::new("cstoken").bytes(token.to_vec()).build() +} + /// Build a `` stanza child for including in outgoing messages. pub fn build_tc_token_node(token: &[u8]) -> Node { NodeBuilder::new("tctoken").bytes(token.to_vec()).build() @@ -491,12 +512,62 @@ mod tests { #[test] fn test_issue_privacy_tokens_spec_new_from_slice() { - let jid1: Jid = "100000000000001@lid".parse().unwrap(); - let jid2: Jid = "100000000000002@lid".parse().unwrap(); + let jid1: Jid = "alice@lid".parse().unwrap(); + let jid2: Jid = "bob@lid".parse().unwrap(); let jids = [jid1.clone(), jid2.clone()]; let spec = IssuePrivacyTokensSpec::new(&jids); assert_eq!(spec.jids.len(), 2); assert_eq!(spec.jids[0], jid1); assert_eq!(spec.jids[1], jid2); } + + #[test] + fn test_compute_cs_token_deterministic() { + let salt = b"test_salt_bytes_16"; + let lid = "alice@lid"; + let token1 = compute_cs_token(salt, lid); + let token2 = compute_cs_token(salt, lid); + assert_eq!(token1, token2); + assert_eq!(token1.len(), 32); // HMAC-SHA256 output is 32 bytes + } + + #[test] + fn test_compute_cs_token_different_lids() { + let salt = b"test_salt_bytes_16"; + let token1 = compute_cs_token(salt, "alice@lid"); + let token2 = compute_cs_token(salt, "bob@lid"); + assert_ne!(token1, token2); + } + + #[test] + fn test_compute_cs_token_different_salts() { + let lid = "alice@lid"; + let token1 = compute_cs_token(b"salt_a", lid); + let token2 = compute_cs_token(b"salt_b", lid); + assert_ne!(token1, token2); + } + + #[test] + fn test_compute_cs_token_known_answer() { + // Pre-computed HMAC-SHA256 to catch accidental algorithm changes + // (e.g., if someone swaps key/data arguments). + let salt = b"whatsapp_nct_salt_example"; + let lid = "alice@lid"; + let expected: [u8; 32] = [ + 0x7c, 0x6a, 0xfc, 0x32, 0x57, 0x85, 0xac, 0x3c, 0x4f, 0x57, 0x1e, 0x64, 0x8a, 0x3b, + 0xb8, 0x22, 0xf0, 0xe2, 0xe4, 0x94, 0x34, 0x81, 0x2e, 0xd2, 0x80, 0x9a, 0xea, 0x2e, + 0x70, 0x43, 0xb5, 0x76, + ]; + assert_eq!(compute_cs_token(salt, lid), expected); + } + + #[test] + fn test_build_cs_token_node() { + let node = build_cs_token_node(&[0xAA, 0xBB, 0xCC]); + assert_eq!(node.tag, "cstoken"); + match &node.content { + Some(NodeContent::Bytes(data)) => assert_eq!(data, &[0xAA, 0xBB, 0xCC]), + _ => panic!("Expected binary content"), + } + } } diff --git a/wacore/src/store/commands.rs b/wacore/src/store/commands.rs index e0931edcc..35e102948 100644 --- a/wacore/src/store/commands.rs +++ b/wacore/src/store/commands.rs @@ -17,6 +17,8 @@ pub enum DeviceCommand { SetPropsHash(Option), SetNextPreKeyId(u32), SetAdvSecretKey([u8; 32]), + SetNctSalt(Option>), + SetNctSaltFromHistorySync(Vec), } pub fn apply_command_to_device(device: &mut Device, command: DeviceCommand) { @@ -51,5 +53,79 @@ pub fn apply_command_to_device(device: &mut Device, command: DeviceCommand) { DeviceCommand::SetAdvSecretKey(key) => { device.adv_secret_key = key; } + DeviceCommand::SetNctSalt(salt) => { + device.nct_salt = salt; + device.nct_salt_sync_seen = true; + } + DeviceCommand::SetNctSaltFromHistorySync(salt) => { + if !salt.is_empty() && !device.nct_salt_sync_seen && device.nct_salt.is_none() { + device.nct_salt = Some(salt); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::{DeviceCommand, apply_command_to_device}; + use crate::store::Device; + + #[test] + fn test_history_sync_salt_backfills_when_no_syncd_mutation_was_seen() { + let mut device = Device::new(); + let salt = vec![1, 2, 3, 4]; + + apply_command_to_device( + &mut device, + DeviceCommand::SetNctSaltFromHistorySync(salt.clone()), + ); + + assert_eq!(device.nct_salt, Some(salt)); + assert!(!device.nct_salt_sync_seen); + } + + #[test] + fn test_history_sync_salt_does_not_resurrect_after_remove() { + let mut device = Device::new(); + + apply_command_to_device(&mut device, DeviceCommand::SetNctSalt(None)); + apply_command_to_device( + &mut device, + DeviceCommand::SetNctSaltFromHistorySync(vec![9, 9, 9]), + ); + + assert_eq!(device.nct_salt, None); + assert!(device.nct_salt_sync_seen); + } + + #[test] + fn test_history_sync_salt_does_not_overwrite_syncd_value() { + let mut device = Device::new(); + let syncd_salt = vec![7, 8, 9]; + + apply_command_to_device( + &mut device, + DeviceCommand::SetNctSalt(Some(syncd_salt.clone())), + ); + apply_command_to_device( + &mut device, + DeviceCommand::SetNctSaltFromHistorySync(vec![1, 2, 3]), + ); + + assert_eq!(device.nct_salt, Some(syncd_salt)); + assert!(device.nct_salt_sync_seen); + } + + #[test] + fn test_history_sync_empty_salt_is_ignored() { + let mut device = Device::new(); + + apply_command_to_device( + &mut device, + DeviceCommand::SetNctSaltFromHistorySync(vec![]), + ); + + assert_eq!(device.nct_salt, None); + assert!(!device.nct_salt_sync_seen); } } diff --git a/wacore/src/store/device.rs b/wacore/src/store/device.rs index d07ff1f27..3c89b1497 100644 --- a/wacore/src/store/device.rs +++ b/wacore/src/store/device.rs @@ -158,6 +158,15 @@ pub struct Device { /// Prevents prekey ID collisions when prekeys are consumed non-sequentially. #[serde(default)] pub next_pre_key_id: u32, + /// NCT salt provisioned by the server via app state sync or history sync. + /// Used to compute cstoken = HMAC-SHA256(salt, recipient_lid) as a fallback + /// when no tctoken is available for first-contact messaging. + #[serde(default)] + pub nct_salt: Option>, + /// Runtime-only marker that an authoritative nct_salt_sync mutation was seen. + /// This prevents stale history sync data from resurrecting a cleared salt. + #[serde(skip)] + pub nct_salt_sync_seen: bool, } impl Default for Device { @@ -209,6 +218,8 @@ impl Device { edge_routing_info: None, props_hash: None, next_pre_key_id: 1, + nct_salt: None, + nct_salt_sync_seen: false, } }