Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,7 @@ whatsapp-rust-ureq-http-client = { path = "./http_clients/ureq-client", version
[dev-dependencies]
aes = { workspace = true }
cbc = { version = "0.2", features = ["alloc", "block-padding"] }
flate2 = { workspace = true }
hkdf = { workspace = true }
hmac = { workspace = true }
sha2 = { workspace = true }
Expand Down
155 changes: 151 additions & 4 deletions src/history_sync.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
use crate::types::events::{Event, LazyHistorySync};
use std::sync::Arc;
use wacore::history_sync::{TcTokenCandidate, process_history_sync};
use wacore::history_sync::{HistoryMsgSecretRecord, TcTokenCandidate, process_history_sync};
use wacore::store::traits::TcTokenEntry;
use wacore_binary::{Jid, JidExt as _};
use waproto::whatsapp::message::HistorySyncNotification;

use crate::client::Client;
Expand Down Expand Up @@ -145,6 +146,7 @@ impl Client {
};

let has_listeners = self.core.event_bus.has_handlers();
let retain_history_blob = has_listeners;

// Small blobs (PushName, Recent): decode inline to avoid spawn_blocking overhead.
// Large blobs: use blocking thread to avoid stalling the async runtime.
Expand All @@ -153,7 +155,7 @@ impl Client {
Some(process_history_sync(
compressed_data,
own_user.as_deref(),
has_listeners,
retain_history_blob,
compressed_size_hint,
))
} else {
Expand All @@ -162,7 +164,7 @@ impl Client {
let result = process_history_sync(
compressed_data,
own_user.as_deref(),
has_listeners,
retain_history_blob,
compressed_size_hint,
);
let _ = result_tx.send(result);
Expand Down Expand Up @@ -215,7 +217,9 @@ impl Client {
self.store_tc_token_candidate(candidate).await;
}

// Dispatch a single event with the full decompressed blob
self.store_history_sync_msg_secrets(sync_result.msg_secret_records)
.await;

if let Some(decompressed) = sync_result.decompressed_bytes {
let lazy_hs = LazyHistorySync::new(
decompressed,
Expand All @@ -240,6 +244,31 @@ impl Client {
}
}

async fn store_history_sync_msg_secrets(&self, records: Vec<HistoryMsgSecretRecord>) -> usize {
let device_snapshot = self.persistence_manager.get_device_snapshot().await;
let own_pn = device_snapshot.pn.as_ref().map(|j| j.to_non_ad());
let own_lid = device_snapshot.lid.as_ref().map(|j| j.to_non_ad());

let mut stored = 0usize;
for record in records {
let Ok(chat) = record.chat_id.parse::<Jid>() else {
continue;
};
let senders =
history_msg_secret_senders(&chat, &record, own_pn.as_ref(), own_lid.as_ref());
for sender in senders {
if self
.persist_msg_secret_bytes(&chat, &sender, &record.msg_id, &record.secret)
.await
{
stored += 1;
}
}
}

stored
}

/// Ask the phone to re-upload a history-sync blob whose download failed,
/// by sending a `<receipt type="server-error" category="peer">` with the
/// blob's `media_key`.
Expand Down Expand Up @@ -323,3 +352,121 @@ impl Client {
}
}
}

fn history_msg_secret_senders(
chat: &Jid,
record: &HistoryMsgSecretRecord,
own_pn: Option<&Jid>,
own_lid: Option<&Jid>,
) -> Vec<Jid> {
let mut senders = Vec::with_capacity(2);

if record.from_me {
if let Some(lid) = own_lid {
push_unique_sender(&mut senders, lid.to_non_ad());
}
if let Some(pn) = own_pn {
push_unique_sender(&mut senders, pn.to_non_ad());
}
return senders;
}

if chat.is_pn() || chat.is_lid() || chat.is_bot() {
senders.push(chat.to_non_ad());
return senders;
}

if let Some(raw_sender) = record
.key_participant
.as_deref()
.or(record.web_msg_participant.as_deref())
&& let Ok(sender) = raw_sender.parse::<Jid>()
{
senders.push(sender.to_non_ad());
}

senders
}

fn push_unique_sender(senders: &mut Vec<Jid>, sender: Jid) {
if !senders.contains(&sender) {
senders.push(sender);
}
}

#[cfg(test)]
mod tests {
use super::*;
use flate2::{Compression, write::ZlibEncoder};
use prost::Message as ProtoMessage;
use std::io::Write;
use std::sync::atomic::Ordering;
use waproto::whatsapp as wa;

fn compress_history_sync(history_sync: &wa::HistorySync) -> Vec<u8> {
let raw = history_sync.encode_to_vec();
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(&raw).expect("zlib write");
encoder.finish().expect("zlib finish")
}

#[tokio::test]
async fn process_history_sync_task_stores_message_secrets_without_handlers() {
let client = crate::test_utils::create_test_client_with_name("history_msg_secret").await;
client
.persistence_manager
.process_command(wacore::store::commands::DeviceCommand::SetId(Some(
"5511000000001:0@s.whatsapp.net".parse().unwrap(),
)))
.await;
client.is_running.store(true, Ordering::Relaxed);

let chat = "5511777776666@s.whatsapp.net";
let parent_id = "HIST_PARENT";
let secret = vec![0x44u8; 32];
let history_sync = wa::HistorySync {
sync_type: wa::history_sync::HistorySyncType::InitialBootstrap as i32,
conversations: vec![wa::Conversation {
id: chat.to_string(),
messages: vec![wa::HistorySyncMsg {
message: Some(wa::WebMessageInfo {
key: wa::MessageKey {
remote_jid: Some(chat.to_string()),
from_me: Some(false),
id: Some(parent_id.to_string()),
participant: None,
},
message: Some(wa::Message {
conversation: Some("historical".to_string()),
..Default::default()
}),
message_secret: Some(secret.clone()),
..Default::default()
}),
msg_order_id: Some(1),
}],
..Default::default()
}],
..Default::default()
};
let compressed = compress_history_sync(&history_sync);
let notification = HistorySyncNotification {
file_length: Some(compressed.len() as u64),
sync_type: Some(wa::message::HistorySyncType::InitialBootstrap as i32),
initial_hist_bootstrap_inline_payload: Some(compressed),
..Default::default()
};

client
.process_history_sync_task("HIST_SYNC_SECRET".to_string(), notification)
.await;

let got = client
.persistence_manager
.backend()
.get_msg_secret(chat, chat, parent_id)
.await
.unwrap();
assert_eq!(got, Some(secret));
}
}
Loading
Loading