feat(client): hand over a decrypted payload before it is decoded

src/client.rs

@@ -63,6 +63,27 @@ use portable_atomic::{AtomicI64, AtomicU64};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
 
+/// Lease that keeps decrypted-payload events enabled for one consumer.
+///
+/// Dropping the final lease disables forwarding. The lease holds only a weak
+/// client reference, so it cannot keep the client alive.
+#[must_use = "dropping the lease immediately releases decrypted-payload forwarding"]
+pub struct DecryptedPayloadLease {
+    client: std::sync::Weak<Client>,
+}
+
+impl Drop for DecryptedPayloadLease {
+    fn drop(&mut self) {
+        let Some(client) = self.client.upgrade() else {
+            return;
+        };
+        let previous = client
+            .decrypted_payload_forwarding
+            .fetch_sub(1, Ordering::Relaxed);
+        debug_assert!(previous > 0, "decrypted-payload forwarding lease underflow");
+    }
+}
+
 /// Lease that keeps raw decoded stanza events enabled for one consumer.
 ///
 /// Dropping the final lease disables forwarding. The lease holds only a weak
@@ -1421,6 +1442,10 @@ pub struct Client {
     /// Number of consumers currently requesting `Event::RawNode` forwarding.
     raw_node_forwarding: AtomicUsize,
 
+    /// Number of consumers currently requesting `Event::DecryptedPayload`
+    /// forwarding.
+    decrypted_payload_forwarding: AtomicUsize,
+
     /// Stanza interceptors, behind the same copy-on-write snapshot the event
     /// bus uses: reading one costs a refcount bump, so the read loop allocates
     /// nothing per stanza. Registering is the rare side, and pays the copy.

src/client/accessors.rs

@@ -61,6 +61,33 @@ impl Client {
         self.raw_node_forwarding.load(Ordering::Relaxed) != 0
     }
 
+    /// Acquire decrypted-payload forwarding for one consumer.
+    ///
+    /// [`Event::DecryptedPayload`] stays enabled until every acquired lease is
+    /// dropped. While none is held nothing is emitted and nothing is cloned:
+    /// the path costs one relaxed atomic load.
+    ///
+    /// [`Event::DecryptedPayload`]: wacore::types::events::Event::DecryptedPayload
+    pub fn acquire_decrypted_payload_forwarding(self: &Arc<Self>) -> DecryptedPayloadLease {
+        let incremented = self
+            .decrypted_payload_forwarding
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |count| {
+                count.checked_add(1)
+            })
+            .is_ok();
+        assert!(
+            incremented,
+            "decrypted-payload forwarding lease counter overflow"
+        );
+        DecryptedPayloadLease {
+            client: Arc::downgrade(self),
+        }
+    }
+
+    pub(crate) fn decrypted_payload_forwarding_enabled(&self) -> bool {
+        self.decrypted_payload_forwarding.load(Ordering::Relaxed) != 0
+    }
+
     /// Register an interceptor that sees each decoded stanza before the
     /// built-in pipeline, and may take it.
     ///

src/client/lifecycle.rs

@@ -506,6 +506,7 @@ impl Client {
             saver_handle: std::sync::OnceLock::new(),
             alloc_meter: std::sync::OnceLock::new(),
             raw_node_forwarding: AtomicUsize::new(0),
+            decrypted_payload_forwarding: AtomicUsize::new(0),
             stanza_interceptors: std::sync::RwLock::new(Arc::new(Vec::new())),
             stanza_interceptor_count: AtomicUsize::new(0),
             next_interceptor_id: AtomicU64::new(0),

src/lib.rs

@@ -126,7 +126,9 @@ pub use client::{
     StatsSnapshot, StorageResourceReport, TransportResourceReport,
 };
 pub use client::{CallError, Voip};
-pub use client::{Client, ClientBuild, ClientBuilder, ClientBuilderError, RawNodeLease};
+pub use client::{
+    Client, ClientBuild, ClientBuilder, ClientBuilderError, DecryptedPayloadLease, RawNodeLease,
+};
 #[cfg(feature = "client-lifecycle")]
 #[cfg_attr(docsrs, doc(cfg(feature = "client-lifecycle")))]
 pub use client::{ClientLifecycle, ConnectionScope, ConnectionScopeState};
@@ -234,7 +236,9 @@ pub mod version;
 /// `use whatsapp_rust::prelude::*;`.
 pub mod prelude {
     pub use crate::bot::{Bot, BotBuilder, BotHandle, EventDelivery, MessageContext};
-    pub use crate::client::{Client, ClientBuilder, ClientBuilderError, ClientError, RawNodeLease};
+    pub use crate::client::{
+        Client, ClientBuilder, ClientBuilderError, ClientError, DecryptedPayloadLease, RawNodeLease,
+    };
     #[cfg(feature = "client-lifecycle")]
     #[cfg_attr(docsrs, doc(cfg(feature = "client-lifecycle")))]
     pub use crate::client::{ClientLifecycle, ConnectionScope, ConnectionScopeState};

src/message.rs

@@ -71,28 +71,54 @@ pub(crate) struct EncPayload {
     pub ciphertext: bytes::Bytes,
     pub enc_type: EncType,
     pub padding_version: u8,
+    /// Position in the order [`message_enc_nodes_for_device`] yields, counting
+    /// from zero: direct `<enc>` children first, then this device's under
+    /// `<participants><to>`.
+    ///
+    /// Recorded during classification because nothing downstream can recover
+    /// it: payloads are split into per-kind buckets and encs that produce no
+    /// payload are skipped, so a position within a bucket is not a position in
+    /// the stanza.
+    pub enc_index: usize,
 }
 
 impl EncPayload {
-    fn from_parts(ciphertext: bytes::Bytes, enc_node: &NodeRef<'_>) -> Option<Self> {
+    fn from_parts(
+        ciphertext: bytes::Bytes,
+        enc_node: &NodeRef<'_>,
+        enc_index: usize,
+    ) -> Option<Self> {
         let enc_type = EncType::from_wire(enc_node.attrs().optional_string("type")?.as_ref())?;
         let padding_version = enc_node.attrs().optional_u64("v").unwrap_or(2) as u8;
         Some(Self {
             ciphertext,
             enc_type,
             padding_version,
+            enc_index,
         })
     }
 
     /// Zero-copy extraction from an OwnedNodeRef.
-    pub(crate) fn from_owned_node(owner: &OwnedNodeRef, enc_node: &NodeRef<'_>) -> Option<Self> {
-        Self::from_parts(owner.slice_bytes(enc_node.content_bytes()?), enc_node)
+    pub(crate) fn from_owned_node(
+        owner: &OwnedNodeRef,
+        enc_node: &NodeRef<'_>,
+        enc_index: usize,
+    ) -> Option<Self> {
+        Self::from_parts(
+            owner.slice_bytes(enc_node.content_bytes()?),
+            enc_node,
+            enc_index,
+        )
     }
 
     /// Copying extraction from a NodeRef (used in tests where there's no OwnedNodeRef).
     #[cfg(test)]
-    pub(crate) fn from_node_ref(node: &NodeRef<'_>) -> Option<Self> {
-        Self::from_parts(bytes::Bytes::copy_from_slice(node.content_bytes()?), node)
+    pub(crate) fn from_node_ref(node: &NodeRef<'_>, enc_index: usize) -> Option<Self> {
+        Self::from_parts(
+            bytes::Bytes::copy_from_slice(node.content_bytes()?),
+            node,
+            enc_index,
+        )
     }
 }
 
@@ -195,6 +221,9 @@ struct DeferredPlaintext {
     enc_type: &'static str,
     plaintext: Vec<u8>,
     padding_version: u8,
+    /// Which `<enc>` in the stanza produced this — [`EncPayload::enc_index`],
+    /// carried through because the buffer drains after the decrypt loop.
+    enc_index: usize,
 }
 
 fn should_process_skmsg_after_session(

src/message/msg_secret.rs

@@ -474,6 +474,10 @@ impl Client {
         use wacore::bot_message::{BotMessageContext, decrypt_bot_message};
         use wacore::protocol::nack::NackReason;
 
+        // Read off before the payload is consumed below.
+        let enc_index = payload.enc_index;
+        let enc_type = payload.enc_type.as_wire_str();
+
         let ms_msg = match waproto::codec::message_secret_message_decode(&payload.ciphertext) {
             Ok(m) => m,
             Err(e) => {
@@ -692,7 +696,23 @@ impl Client {
             },
         };
 
-        let msg = match waproto::codec::message_decode(plaintext.as_slice()) {
+        // Forwarded before decoding, like the Signal path: a bot payload that
+        // opens but does not decode is nacked and dropped, and the secret it
+        // was opened with is single-use, so nothing can ask for it again.
+        // `Bytes::from` takes the buffer over rather than copying it.
+        let plaintext = bytes::Bytes::from(plaintext);
+        if self.decrypted_payload_forwarding_enabled() {
+            self.core.event_bus.dispatch(Event::DecryptedPayload(
+                wacore::types::events::DecryptedPayload::builder()
+                    .info(Arc::clone(info))
+                    .enc_index(enc_index)
+                    .enc_type(enc_type)
+                    .payload(plaintext.clone())
+                    .build(),
+            ));
+        }
+
+        let msg = match waproto::codec::message_decode(&plaintext) {
             Ok(m) => m,
             Err(e) => {
                 log::warn!(

src/message/receive.rs

@@ -250,7 +250,7 @@ impl Client {
         // per enc node. `None` (the common zero-handler bot) skips the lookup.
         let custom_enc_handlers = self.custom_enc_handlers.get();
 
-        for enc_node in &all_enc_nodes {
+        for (enc_index, enc_node) in all_enc_nodes.iter().enumerate() {
             max_sender_retry_count = max_sender_retry_count.max(sender_retry_count(enc_node));
 
             // Parse decrypt-fail attribute (WA Web: e.maybeAttrString("decrypt-fail") === "hide")
@@ -307,7 +307,7 @@ impl Client {
                 continue;
             }
 
-            let payload = match EncPayload::from_owned_node(node, enc_node) {
+            let payload = match EncPayload::from_owned_node(node, enc_node, enc_index) {
                 Some(p) => p,
                 None => {
                     log::warn!("Enc node {enc_type} has no content");
@@ -653,6 +653,7 @@ impl Client {
                 ciphertext,
                 enc_type,
                 padding_version,
+                enc_index,
             } = payload;
             let enc_type_str = enc_type.as_wire_str();
             #[cfg(feature = "tracing")]
@@ -776,6 +777,7 @@ impl Client {
                         enc_type,
                         plaintext: decrypted.plaintext,
                         padding_version,
+                        enc_index,
                     });
                 }
                 Err(e) => {
@@ -886,6 +888,7 @@ impl Client {
                                     enc_type,
                                     plaintext: decrypted.plaintext,
                                     padding_version,
+                                    enc_index,
                                 });
                             }
                             Err(retry_err) => {
@@ -916,6 +919,7 @@ impl Client {
                                             &mut rng,
                                             enc_type,
                                             padding_version,
+                                            enc_index,
                                             info,
                                             &session_mutex,
                                             &mut session_guard,
@@ -999,6 +1003,7 @@ impl Client {
                                 &mut rng,
                                 enc_type,
                                 padding_version,
+                                enc_index,
                                 info,
                                 &session_mutex,
                                 &mut session_guard,
@@ -1043,6 +1048,7 @@ impl Client {
                                 &mut rng,
                                 enc_type,
                                 padding_version,
+                                enc_index,
                                 info,
                                 &session_mutex,
                                 &mut session_guard,
@@ -1096,6 +1102,7 @@ impl Client {
                                 &mut rng,
                                 enc_type,
                                 padding_version,
+                                enc_index,
                                 info,
                                 &session_mutex,
                                 &mut session_guard,
@@ -1191,10 +1198,11 @@ impl Client {
             enc_type,
             plaintext,
             padding_version,
+            enc_index,
         } in deferred
         {
             match self
-                .handle_decrypted_plaintext(enc_type, plaintext, padding_version, info)
+                .handle_decrypted_plaintext(enc_type, plaintext, padding_version, enc_index, info)
                 .await
             {
                 Ok(plaintext_outcome) => {
@@ -1251,6 +1259,7 @@ impl Client {
         for payload in payloads {
             let ciphertext = &payload.ciphertext[..];
             let padding_version = payload.padding_version;
+            let enc_index = payload.enc_index;
 
             log::debug!(
                 "Looking up sender key for group {} with sender address {} (from sender JID: {})",
@@ -1284,6 +1293,7 @@ impl Client {
                             "skmsg",
                             padded_plaintext,
                             padding_version,
+                            enc_index,
                             info,
                         )
                         .await
@@ -1439,16 +1449,30 @@ impl Client {
     #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.handle_plaintext", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id, enc_type = %enc_type), err(Debug)))]
     pub(crate) async fn handle_decrypted_plaintext(
         self: &Arc<Self>,
-        enc_type: &str,
+        enc_type: &'static str,
         padded_plaintext: Vec<u8>,
         padding_version: u8,
+        enc_index: usize,
         info: &Arc<MessageInfo>,
     ) -> Result<PlaintextHandleOutcome, anyhow::Error> {
+        let source = wacore::messages::unpad_plaintext(padded_plaintext, padding_version)?;
+
+        // Emitted before decoding, so a payload this build cannot decode still
+        // reaches a consumer that wants it. Nothing is cloned while no lease is
+        // held, and a `Bytes` clone is a refcount bump when one is.
+        if self.decrypted_payload_forwarding_enabled() {
+            self.core.event_bus.dispatch(Event::DecryptedPayload(
+                wacore::types::events::DecryptedPayload::builder()
+                    .info(Arc::clone(info))
+                    .enc_index(enc_index)
+                    .enc_type(enc_type)
+                    .payload(source.clone())
+                    .build(),
+            ));
+        }
+
         let (original_msg, history_sync_taken) =
-            wacore::messages::decode_plaintext_detached_history_sync(
-                padded_plaintext,
-                padding_version,
-            )?;
+            wacore::messages::decode_unpadded_detached_history_sync(source)?;
         log::debug!(
             "[msg:{}] Successfully decrypted message from {}: type={} [batch path]",
             info.id,
@@ -1637,6 +1661,7 @@ impl Client {
         rng: &mut rand::rngs::StdRng,
         enc_type: &'static str,
         padding_version: u8,
+        enc_index: usize,
         info: &Arc<MessageInfo>,
         session_mutex: &Arc<async_lock::Mutex<()>>,
         session_guard: &mut Option<async_lock::MutexGuardArc<()>>,
@@ -1696,6 +1721,7 @@ impl Client {
                     enc_type,
                     plaintext: decrypted.plaintext,
                     padding_version,
+                    enc_index,
                 });
                 MigrationDecryptResult::Decrypted
             }
@@ -1774,6 +1800,7 @@ mod enc_bucket_tests {
 
     fn payload(enc_type: EncType) -> EncPayload {
         EncPayload {
+            enc_index: 0,
             ciphertext: bytes::Bytes::from_static(b"ct"),
             enc_type,
             padding_version: 2,