feat: cstoken (NCT) privacy token fallback + proto update

src/client.rs

@@ -91,6 +91,11 @@ struct NodeWaiter {
     tx: futures::channel::oneshot::Sender<Arc<Node>>,
 }
 
+struct SentNodeWaiter {
+    filter: NodeFilter,
+    tx: futures::channel::oneshot::Sender<Arc<Node>>,
+}
+
 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<Vec<NodeWaiter>>,
     node_waiter_count: AtomicUsize,
+    /// Waiters for raw outgoing nodes before encryption.
+    sent_node_waiters: std::sync::Mutex<Vec<SentNodeWaiter>>,
+    sent_node_waiter_count: AtomicUsize,
 
     pub(crate) unique_id: String,
     pub(crate) id_counter: Arc<AtomicU64>,
@@ -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(),
@@ -2619,10 +2629,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 +2991,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 `<tctoken>` or
+    /// `<cstoken>` was attached.
+    pub fn wait_for_sent_node(
+        &self,
+        filter: NodeFilter,
+    ) -> futures::channel::oneshot::Receiver<Arc<Node>> {
+        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<Node>) {
@@ -2977,6 +3034,26 @@ impl Client {
         }
     }
 
+    fn resolve_sent_node_waiters(&self, node: &Arc<Node>) {
+        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;
+            }
+        }
+    }
+
     pub(crate) fn update_server_time_offset(&self, node: &wacore_binary::node::Node) {
         self.unified_session.update_server_time_offset(node);
     }
@@ -3128,6 +3205,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);
 

src/history_sync.rs

@@ -255,6 +255,17 @@ 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!("Stored NCT salt from history sync ({} bytes)", salt.len());
+                    self.persistence_manager
+                        .process_command(wacore::store::commands::DeviceCommand::SetNctSalt(Some(
+                            salt,
+                        )))
+                        .await;
+                }
             }
             Some(Err(e)) => {
                 log::error!("Failed to process HistorySync data: {:?}", e);

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,8 @@ impl Client {
             participants: Vec<Jid>,
         }
         let mut skdm_update: Option<SkdmUpdate> = None;
+        let mut should_issue_tc_token_after_send = false;
+        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
@@ -1007,9 +1010,13 @@ impl Client {
             // Include tctoken in 1:1 messages (matches WhatsApp Web behavior).
             // 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)
+            should_issue_tc_token_after_send = !to.is_group()
+                && !to.is_newsletter()
+                && self
+                    .maybe_include_tc_token(&to, &mut extra_stanza_nodes)
                     .await;
+            if should_issue_tc_token_after_send {
+                debug!(target: "Client/TcToken", "Scheduled tc token issuance after send for {}", to);
             }
 
             // Acquire lock only for encryption
@@ -1049,6 +1056,10 @@ impl Client {
 
         self.send_node(stanza_to_send).await?;
 
+        if should_issue_tc_token_after_send {
+            self.issue_tc_token_after_send(&tc_issue_target).await;
+        }
+
         // Update SKDM recipient cache AFTER server ACK (matches WhatsApp Web behavior).
         // WA Web only calls markHasSenderKey() after the server confirms receipt.
         if let Some(update) = skdm_update {
@@ -1069,13 +1080,17 @@ impl Client {
         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.
+    ///
+    /// 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)
     ///
-    /// If a valid (non-expired) token exists, adds a `<tctoken>` 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<Node>) {
+    /// Returns whether we should issue a new tc token for this chat after send.
+    async fn maybe_include_tc_token(&self, to: &Jid, extra_nodes: &mut Vec<Node>) -> bool {
         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;
@@ -1091,37 +1106,39 @@ impl Client {
                 .as_ref()
                 .is_some_and(|lid| lid.is_same_user_as(to));
         if is_self {
-            return;
+            return false;
         }
 
-        // 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
+                // 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 {
@@ -1134,40 +1151,60 @@ impl Client {
                 }
             }
             _ => {
-                // 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
+    }
+
+    async fn issue_tc_token_after_send(&self, to: &Jid) {
+        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;
+        };
+
+        self.store_issued_tc_tokens(&response.tokens).await;
+    }
+
+    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 {
+            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}");
+            }
+        }
     }
 
     /// Look up a valid (non-expired) tctoken for a JID. Returns the raw token bytes if found.