diff --git a/agent_docs/signal_durability.md b/agent_docs/signal_durability.md index 9d9136a51..29fd20077 100644 --- a/agent_docs/signal_durability.md +++ b/agent_docs/signal_durability.md @@ -19,6 +19,25 @@ previously persisted lease and can use write-behind. A send at the bound raises it by `SENDER_CHAIN_RESERVATION_BATCH` and must wait for a successful durable flush before its ciphertext is published. +A reservation describes one sender chain, not the address. A DH ratchet +replaces the chain in place with fresh random key material and drops the old +one without archiving it, so the inherited ceiling stops describing anything +reachable: `rebase_lease_after_sender_chain_reset` lowers it back to one batch +as part of the same mutation. Lowering is sound only there — no snapshot can +pair the retired chain with the rebased ceiling — and only downward, so a +counter is never published under a ceiling that is not yet durable. Leave it +stranded and a long monologue followed by one peer reply puts the gap past +`MAX_RESERVATION_FAST_FORWARD`, where recovery can neither burn nor accept the +record. A chain that is *archived* rather than discarded keeps its claim: +`promote_fresh_state` burns the outgoing state to the ceiling before resetting +it. + +An undecodable session row is reported absent rather than surfaced as a load +error. Every path that could replace it — the peer's next pre-key message, the +retry repair — must load it first, so a propagated error strands the address +permanently. `wa_session_record_quarantined_total` counts these; steady state +is zero. + The cache takes ownership of transient record gates. A failed write, a checked out record skipped by a flush, or a tombstone whose delete failed must remain gated. Only the backend operation that persisted that address may release it. diff --git a/wacore/libsignal/src/protocol/session_cipher.rs b/wacore/libsignal/src/protocol/session_cipher.rs index d06f281cc..dcf8b9dba 100644 --- a/wacore/libsignal/src/protocol/session_cipher.rs +++ b/wacore/libsignal/src/protocol/session_cipher.rs @@ -851,7 +851,10 @@ fn create_decryption_failure_log( } enum RecordDecryptState { - Current(DecryptSnapshot), + Current { + snapshot: DecryptSnapshot, + sender_chain_reset: bool, + }, Previous { state: Box, effect: StateDecryptEffect, @@ -920,7 +923,7 @@ impl RecordDecryptTransaction<'_> { .as_ref() .expect("a live decrypt transaction owns its rollback") { - RecordDecryptState::Current(_) => self + RecordDecryptState::Current { .. } => self .record .session_state() .expect("a current decrypt transaction keeps the current state installed"), @@ -938,19 +941,35 @@ impl RecordDecryptTransaction<'_> { .take() .expect("a live decrypt transaction owns its rollback") { - RecordDecryptState::Current(_) => { + RecordDecryptState::Current { + sender_chain_reset, .. + } => { let state = self .record .session_state_mut() .expect("a current decrypt transaction keeps the current state installed"); state.clear_unacknowledged_pre_key_message(); + if sender_chain_reset { + self.record.rebase_lease_after_sender_chain_reset(); + } } RecordDecryptState::Previous { mut state, effect, .. } => { + let sender_chain_reset = effect.sender_chain_reset(); effect.commit(&mut state); state.clear_unacknowledged_pre_key_message(); - self.record.promote_state(*state); + if sender_chain_reset { + // The ratchet gave this state a chain built from a fresh + // random ephemeral: no counter on it can have been spent, + // so it takes the same path as any other fresh ratchet + // rather than having the outgoing chain's lease burned + // into it (which, past the fast-forward ceiling, would + // drop the chain outright). + self.record.promote_fresh_state(*state); + } else { + self.record.promote_state(*state); + } } } std::mem::take(&mut self.plaintext) @@ -963,7 +982,7 @@ impl Drop for RecordDecryptTransaction<'_> { return; }; match state { - RecordDecryptState::Current(snapshot) => self + RecordDecryptState::Current { snapshot, .. } => self .record .session_state_mut() .expect("a current decrypt transaction keeps the current state installed") @@ -1043,13 +1062,17 @@ fn decrypt_message_with_record<'a, R: Rng + CryptoRng>( chain_key, plaintext: result.plaintext, })), - StateDecryptEffect::Applied(snapshot) => { - Ok(RecordDecrypt::Transaction(RecordDecryptTransaction { - record, - state: Some(RecordDecryptState::Current(snapshot)), - plaintext: result.plaintext, - })) - } + StateDecryptEffect::Applied { + snapshot, + sender_chain_reset, + } => Ok(RecordDecrypt::Transaction(RecordDecryptTransaction { + record, + state: Some(RecordDecryptState::Current { + snapshot, + sender_chain_reset, + }), + plaintext: result.plaintext, + })), }; } Err(SignalProtocolError::DuplicatedMessage(chain, counter)) @@ -1269,10 +1292,26 @@ enum StateDecryptEffect { ratchet_key: PublicKey, chain_key: ChainKey, }, - Applied(DecryptSnapshot), + Applied { + snapshot: DecryptSnapshot, + /// A DH ratchet replaced this state's sender chain, so the record's + /// counter lease no longer describes the chain it is about to gate. + sender_chain_reset: bool, + }, } impl StateDecryptEffect { + /// The in-order fast path never ratchets, so only an applied decrypt can + /// have retired a sender chain. + fn sender_chain_reset(&self) -> bool { + match self { + Self::DeferredChainKey { .. } => false, + Self::Applied { + sender_chain_reset, .. + } => *sender_chain_reset, + } + } + fn commit(self, state: &mut SessionState) { match self { Self::DeferredChainKey { @@ -1283,14 +1322,14 @@ impl StateDecryptEffect { .set_receiver_chain_key(&ratchet_key, &chain_key) .expect("the deferred in-order receiver chain remains installed"); } - Self::Applied(_) => {} + Self::Applied { .. } => {} } } fn rollback(self, state: &mut SessionState) { match self { Self::DeferredChainKey { .. } => {} - Self::Applied(snapshot) => state.restore_decrypt_snapshot(snapshot), + Self::Applied { snapshot, .. } => state.restore_decrypt_snapshot(snapshot), } } } @@ -1359,9 +1398,12 @@ fn decrypt_message_with_state( counter, ); match result { - Ok(plaintext) => Ok(StateDecryptResult { + Ok((plaintext, sender_chain_reset)) => Ok(StateDecryptResult { plaintext, - effect: StateDecryptEffect::Applied(snapshot), + effect: StateDecryptEffect::Applied { + snapshot, + sender_chain_reset, + }, }), Err(e) => { state.restore_decrypt_snapshot(snapshot); @@ -1381,7 +1423,7 @@ fn decrypt_with_pending_state( their_ephemeral: &PublicKey, receiver_chain: Option, counter: u32, -) -> Result> { +) -> Result<(Vec, bool)> { if let Some(ReceiverChainState::Closed { next_index }) = receiver_chain { if counter >= next_index { return Err(SignalProtocolError::InvalidSessionStructure( @@ -1398,7 +1440,8 @@ fn decrypt_with_pending_state( original_message_type, remote_address, message_key_gen, - ); + ) + .map(|plaintext| (plaintext, false)); } let (chain_key, deferred_ratchet) = @@ -1425,11 +1468,14 @@ fn decrypt_with_pending_state( // Generating a new sender ratchet requires fresh entropy and a second DH. // Neither can affect inbound MAC verification, so perform them only after // the candidate session has authenticated the message. - if let Some(deferred_ratchet) = deferred_ratchet { + let sender_chain_reset = if let Some(deferred_ratchet) = deferred_ratchet { deferred_ratchet.apply(state, their_ephemeral, csprng)?; - } + true + } else { + false + }; - Ok(plaintext) + Ok((plaintext, sender_chain_reset)) } fn decrypt_with_message_keys( diff --git a/wacore/libsignal/src/protocol/state/session.rs b/wacore/libsignal/src/protocol/state/session.rs index f19bedb03..84d7ee092 100644 --- a/wacore/libsignal/src/protocol/state/session.rs +++ b/wacore/libsignal/src/protocol/state/session.rs @@ -858,6 +858,34 @@ impl SessionRecord { self.pending_reservation = true; } + /// Rebase the lease after a DH ratchet replaced the leased sender chain + /// in place. + /// + /// The ceiling bounds counters that a durable snapshot of the *retired* + /// chain may already have published. A ratchet derives the replacement + /// from a fresh random ephemeral and overwrites the old chain without + /// archiving it, so nothing reachable from this record can reissue those + /// counters and the inherited ceiling no longer describes anything. Left + /// in place it strands the lease arbitrarily far above the new chain's + /// index — a monologue of a few thousand sends followed by one peer reply + /// is enough to push the gap past `MAX_RESERVATION_FAST_FORWARD`, and a + /// recovery reload then refuses the record outright, permanently + /// stranding the address. + /// + /// Lowering is sound only because the swap and the rebase are a single + /// mutation of one record: no snapshot can pair the retired chain with + /// the rebased ceiling. Keeping one batch (rather than dropping to zero) + /// leaves the fresh chain's first counters lease-covered, so steady-state + /// ping-pong keeps its write-behind send path. + pub fn rebase_lease_after_sender_chain_reset(&mut self) { + // Never raises: a counter must not be published under a ceiling that + // is not yet durable. An in-chain lease is always within one batch of + // the live index, so this is a no-op outside a chain replacement. + self.reserved_sender_chain_index = self + .reserved_sender_chain_index + .min(consts::SENDER_CHAIN_RESERVATION_BATCH); + } + pub fn has_pending_reservation(&self) -> bool { self.pending_reservation } diff --git a/wacore/libsignal/tests/counter_lease.rs b/wacore/libsignal/tests/counter_lease.rs index 5fc90b78e..73c8f01ef 100644 --- a/wacore/libsignal/tests/counter_lease.rs +++ b/wacore/libsignal/tests/counter_lease.rs @@ -9,7 +9,9 @@ use async_trait::async_trait; use std::collections::HashMap; -use wacore_libsignal::protocol::consts::SENDER_CHAIN_RESERVATION_BATCH; +use wacore_libsignal::protocol::consts::{ + MAX_RESERVATION_FAST_FORWARD, SENDER_CHAIN_RESERVATION_BATCH, +}; use wacore_libsignal::protocol::{ CiphertextMessage, Direction, GenericSignedPreKey, IdentityChange, IdentityKey, IdentityKeyPair, IdentityKeyStore, KeyPair, PreKeyBundle, PreKeyId, PreKeyRecord, PreKeyStore, @@ -308,6 +310,46 @@ fn record_of(peer: &Peer, remote: &ProtocolAddress) -> SessionRecord { .clone() } +/// Live index of the record's current sender chain. +fn sender_chain_index(record: &SessionRecord) -> u32 { + record + .session_state() + .expect("current session") + .get_sender_chain_key() + .expect("sender chain") + .index() +} + +/// Send until the lease ceiling passes `target`, mimicking a bot that +/// monologues at one peer (its own primary device gets a copy of every +/// outgoing message) without that peer ever replying. +fn monologue_until_lease_exceeds(alice: &mut Peer, bob: &ProtocolAddress, target: u32) { + let mut sends = 0u32; + while record_of(alice, bob).reserved_sender_chain_index() <= target { + send(alice, bob, b"monologue"); + sends += 1; + assert!( + sends < target + 2 * SENDER_CHAIN_RESERVATION_BATCH, + "runaway" + ); + } +} + +/// The two incarnations a store hands to the record: the same one for a +/// reload inside a live cache, a different one after a restart or a lossy +/// cache reset. +const LIVE_INCARNATION: [u8; 16] = [0xA1; 16]; +const RESTART_INCARNATION: [u8; 16] = [0xB2; 16]; + +fn store_roundtrip( + record: &SessionRecord, + reload_as: &[u8; 16], +) -> Result { + let mut bytes = Vec::new(); + record.serialize_into_for_store(&mut bytes, &LIVE_INCARNATION); + SessionRecord::deserialize_for_store(&bytes, reload_as) +} + /// Simulate the store layer acknowledging a durable flush: take over the wire /// gate and return the serialized snapshot that "reached storage". fn ack_flush(peer: &mut Peer, remote: &ProtocolAddress) -> Vec { @@ -533,3 +575,138 @@ fn lease_round_trips_through_storage_and_legacy_records_load_untouched() { let reloaded = SessionRecord::deserialize(&legacy).expect("legacy deserializes"); assert_eq!(reloaded.reserved_sender_chain_index(), 0); } + +// ---- stranded lease after a DH ratchet (issue #1146) ------------------------ +// +// The lease ceiling is a record-level counter; the sender chain it bounds is +// per-ratchet-epoch and restarts at zero every time the peer replies. A long +// monologue followed by one reply used to leave the ceiling stranded thousands +// of counters above the live index, which no send ever created — and which a +// recovery reload could neither burn nor accept. + +/// The regression itself: after the ratchet the ceiling must describe the +/// chain that is actually installed, not the one that was retired. +#[test] +fn a_dh_ratchet_rebases_the_lease_onto_the_fresh_chain() { + let mut alice = Peer::new("alice"); + let mut bob = Peer::new("bob"); + establish(&mut alice, &mut bob); + + // Alice monologues past the load-time fast-forward ceiling. A bot that + // copies every outgoing message to its own primary device reaches this in + // a couple of thousand sends. + monologue_until_lease_exceeds(&mut alice, &bob.address, MAX_RESERVATION_FAST_FORWARD); + let stranded_ceiling = record_of(&alice, &bob.address).reserved_sender_chain_index(); + assert!(stranded_ceiling > MAX_RESERVATION_FAST_FORWARD); + + // Bob finally replies: Alice DH-ratchets onto a chain that starts at zero. + let ct = send(&mut bob, &alice.address, b"reply"); + receive(&mut alice, &bob.address, &ct).expect("decrypt reply"); + + let record = record_of(&alice, &bob.address); + assert_eq!(sender_chain_index(&record), 0, "fresh chain starts at 0"); + assert!( + record.reserved_sender_chain_index() <= SENDER_CHAIN_RESERVATION_BATCH, + "the retired chain's ceiling ({}) must not survive onto the fresh chain", + record.reserved_sender_chain_index() + ); +} + +/// The operator-visible symptom: a stranded ceiling turned the stored row +/// into a hard load failure for every path that touches the address — decrypt, +/// encrypt, and retry repair alike — and only after a restart or a lossy cache +/// reset, since a live reload skips the fast-forward entirely. +#[test] +fn a_ratcheted_record_still_loads_after_a_restart() { + let mut alice = Peer::new("alice"); + let mut bob = Peer::new("bob"); + establish(&mut alice, &mut bob); + monologue_until_lease_exceeds(&mut alice, &bob.address, MAX_RESERVATION_FAST_FORWARD); + + let ct = send(&mut bob, &alice.address, b"reply"); + receive(&mut alice, &bob.address, &ct).expect("decrypt reply"); + let record = record_of(&alice, &bob.address); + + store_roundtrip(&record, &LIVE_INCARNATION).expect("a live reload never fast-forwards"); + let recovered = + store_roundtrip(&record, &RESTART_INCARNATION).expect("a restart must not strand the row"); + assert!( + recovered.reserved_sender_chain_index() - sender_chain_index(&recovered) + <= SENDER_CHAIN_RESERVATION_BATCH, + "recovery must burn at most one batch" + ); +} + +/// Performance guard: rebasing must not cost the fresh chain its lease +/// coverage. Dropping the ceiling to zero instead of one batch would put a +/// synchronous durability flush in front of every reply in a ping-pong. +#[test] +fn the_rebased_lease_still_covers_the_fresh_chain_without_a_flush() { + let mut alice = Peer::new("alice"); + let mut bob = Peer::new("bob"); + establish(&mut alice, &mut bob); + monologue_until_lease_exceeds(&mut alice, &bob.address, MAX_RESERVATION_FAST_FORWARD); + ack_flush(&mut alice, &bob.address); + + let ct = send(&mut bob, &alice.address, b"reply"); + receive(&mut alice, &bob.address, &ct).expect("decrypt reply"); + + for counter in 0..SENDER_CHAIN_RESERVATION_BATCH { + let ct = send(&mut alice, &bob.address, b"post-ratchet"); + assert_eq!(wire_counter(&ct), counter); + assert!( + !record_of(&alice, &bob.address).has_pending_reservation(), + "counter {counter} of the fresh chain must ride the rebased lease" + ); + } + + // ...and the batch boundary still re-raises and re-gates as before. + let ct = send(&mut alice, &bob.address, b"boundary"); + assert_eq!(wire_counter(&ct), SENDER_CHAIN_RESERVATION_BATCH); + assert!(record_of(&alice, &bob.address).has_pending_reservation()); +} + +/// Lowering a ceiling is only safe if it can never uncover a counter that +/// already reached the wire. Publish across the ratchet and both sides of a +/// crash, and assert the resumed chain repeats nothing. +#[test] +fn a_rebased_lease_never_republishes_a_counter_across_a_crash() { + let mut alice = Peer::new("alice"); + let mut bob = Peer::new("bob"); + establish(&mut alice, &mut bob); + monologue_until_lease_exceeds(&mut alice, &bob.address, MAX_RESERVATION_FAST_FORWARD); + ack_flush(&mut alice, &bob.address); + + let ct = send(&mut bob, &alice.address, b"reply"); + receive(&mut alice, &bob.address, &ct).expect("decrypt reply"); + + // Everything below is on the post-ratchet chain, so counters are + // comparable across the crash. + let mut published = Vec::new(); + for _ in 0..3 { + published.push(wire_counter(&send( + &mut alice, + &bob.address, + b"pre-snapshot", + ))); + } + let snapshot = ack_flush(&mut alice, &bob.address); + for _ in 0..5 { + published.push(wire_counter(&send(&mut alice, &bob.address, b"unflushed"))); + } + + crash_reload(&mut alice, &bob.address, &snapshot); + + let ct = send(&mut alice, &bob.address, b"after crash"); + let resumed = wire_counter(&ct); + assert!( + !published.contains(&resumed), + "counter {resumed} was already published before the crash" + ); + assert_eq!( + resumed, SENDER_CHAIN_RESERVATION_BATCH, + "recovery burns the rebased lease, not the retired chain's ceiling" + ); + let pt = receive(&mut bob, &alice.address, &ct).expect("bob decrypts across the burned gap"); + assert_eq!(&pt[..], b"after crash"); +} diff --git a/wacore/src/store/signal_cache.rs b/wacore/src/store/signal_cache.rs index 360fec23a..d0980235e 100644 --- a/wacore/src/store/signal_cache.rs +++ b/wacore/src/store/signal_cache.rs @@ -692,6 +692,35 @@ impl SignalStoreCache { // === Sessions (object cache — serialize only during flush) === + /// Decode a stored session, quarantining a blob this build cannot read. + /// + /// Deserialization is a pure function of the bytes, so a row that fails + /// once fails identically forever — and it fails on *every* path that must + /// load the address, including the decrypt of the peer's next pre-key + /// message and the retry repair, which are precisely the paths that would + /// otherwise replace it. Propagating the error therefore strands the + /// address until an operator deletes the row by hand. Reporting it as + /// absent instead lets the ordinary no-session recovery fetch a pre-key + /// bundle and overwrite it. Nothing is lost: a record we cannot decode can + /// derive no key material, so it cannot repeat a counter either. + fn decode_stored_session( + key: &str, + bytes: &[u8], + incarnation: &StoreIncarnation, + ) -> Option { + match SessionRecord::deserialize_for_store(bytes, incarnation) { + Ok(record) => Some(record), + Err(error) => { + log::error!( + "discarding unreadable session row for addr#{:016x}: {error} — recovering with a fresh session", + wacore_binary::jid::observe_token(key) + ); + crate::telemetry::session_record_quarantined(); + None + } + } + } + /// Takes ownership of the cached session, leaving a `CheckedOut` marker. /// Callers must return the record with [`put_session`](Self::put_session) after use. pub async fn get_session( @@ -738,9 +767,11 @@ impl SignalStoreCache { CachedSessionCheckout::Busy => anyhow::bail!("session is already checked out"), CachedSessionCheckout::Missing(checkout) => checkout, }; - match backend_result { - Some(bytes) => { - let record = SessionRecord::deserialize_for_store(&bytes, &state.incarnation)?; + match backend_result + .as_deref() + .and_then(|bytes| Self::decode_stored_session(key, bytes, &state.incarnation)) + { + Some(record) => { state.cache.insert( Arc::from(key), SessionEntry::CheckedOut { @@ -808,12 +839,12 @@ impl SignalStoreCache { SessionEntry::Absent | SessionEntry::CheckedOut { .. } => Ok(None), }; } - match backend_result { - Some(bytes) => { - let record = Arc::new(SessionRecord::deserialize_for_store( - &bytes, - &state.incarnation, - )?); + match backend_result + .as_deref() + .and_then(|bytes| Self::decode_stored_session(key, bytes, &state.incarnation)) + { + Some(record) => { + let record = Arc::new(record); state .cache .insert(Arc::from(key), SessionEntry::Present(record.clone())); @@ -870,8 +901,15 @@ impl SignalStoreCache { } /// Non-destructive existence check; an empty checkout remains absent. - /// Backend misses are negative-cached; hits are not cached to skip - /// deserialization (the subsequent `get_session` will cache on demand). + /// + /// A cold probe reads and decodes the row rather than asking the backend + /// whether it exists. Row existence alone would report a quarantined + /// session as present, and this is the probe that decides whether a send + /// fetches a pre-key bundle: answering `true` for a row that + /// [`Self::checkout_session`] will then discard skips the recovery, and the + /// send fails or silently drops that recipient from the fan-out. The decode + /// is not wasted work either, since the record it produces is cached for + /// the checkout that follows. pub async fn has_session( &self, address: &ProtocolAddress, @@ -885,15 +923,21 @@ impl SignalStoreCache { } } // Backend I/O outside the lock - let exists = backend.has_session(key).await?; + let backend_result = backend.get_session(key).await?; let mut state = self.lock_sessions().await; if let Some(entry) = state.cache.get(key) { return Ok(entry.exists()); } - if !exists { - state.cache.insert(Arc::from(key), SessionEntry::Absent); - state.evict_if_needed(self.max_entries); - } + let entry = match backend_result + .as_deref() + .and_then(|bytes| Self::decode_stored_session(key, bytes, &state.incarnation)) + { + Some(record) => SessionEntry::Present(Arc::new(record)), + None => SessionEntry::Absent, + }; + let exists = entry.exists(); + state.cache.insert(Arc::from(key), entry); + state.evict_if_needed(self.max_entries); Ok(exists) } @@ -1232,7 +1276,26 @@ impl SignalStoreCache { }; let durable = match durable { Some(d) => d, - None => backend.has_session(addr.as_ref()).await?, + // Row existence is not enough: a row that does not + // decode is no session at all, and deleting the + // prekey against it is the very outcome this block + // exists to prevent -- a redelivered pkmsg would + // have neither a usable session nor the prekey to + // rebuild one. Decoded under the sessions lock we + // already hold, so the decision stays atomic + // against a decrypt storing its own session. + None => backend + .get_session(addr.as_ref()) + .await? + .as_deref() + .and_then(|bytes| { + Self::decode_stored_session( + addr.as_ref(), + bytes, + &state.incarnation, + ) + }) + .is_some(), }; if durable { deletable.push(*id); @@ -2280,6 +2343,47 @@ mod consumed_prekey_atomicity_tests { ); } + /// The same, for a row that is present but does not decode. Row existence + /// alone would call it durable and delete the prekey, leaving a redelivered + /// pkmsg with neither a usable session nor the prekey to rebuild one -- + /// which is the exact outcome the deferral rule exists to prevent. + #[tokio::test] + async fn prekey_behind_an_unreadable_session_row_survives_flush() { + use super::lease_reload_tests::leased_session; + use crate::libsignal::protocol::consts::MAX_RESERVATION_FAST_FORWARD; + + let backend = InMemoryBackend::new(); + let addr = seed(&backend).await; + + // Persist a row that only fails to decode after a restart, so the + // backend genuinely holds bytes for this address. + let writer = SignalStoreCache::with_max_entries_and_incarnation( + DEFAULT_MAX_CACHE_ENTRIES, + [0xA1; 16], + ); + let mut stranded = leased_session(); + stranded.reserve_sender_chain_counters(MAX_RESERVATION_FAST_FORWARD); + writer.put_session(&addr, stranded).await; + writer.flush(&backend).await.unwrap(); + assert!( + backend.get_session(addr.as_str()).await.unwrap().is_some(), + "the row is there; what follows is about whether it decodes" + ); + + // A different incarnation: the reload has to fast-forward, and refuses. + let restarted = SignalStoreCache::with_max_entries_and_incarnation( + DEFAULT_MAX_CACHE_ENTRIES, + [0xB2; 16], + ); + restarted.remove_prekey(PREKEY_ID, addr.as_str()).await; + restarted.flush(&backend).await.unwrap(); + + assert!( + backend.load_prekey(PREKEY_ID).await.unwrap().is_some(), + "a prekey behind a row that does not decode must survive the flush" + ); + } + /// A prekey buffered for a session that is not durable (its volatile session /// was dropped before the buffer insert landed, e.g. a disconnect clear() /// racing the consume path) must NOT be deleted: removing the durable prekey @@ -2958,7 +3062,7 @@ mod lease_reload_tests { SenderKeyName::from_parts("group@g.us", "15550001000@s.whatsapp.net:0") } - fn leased_session() -> SessionRecord { + pub(super) fn leased_session() -> SessionRecord { let mut rng = rand::make_rng::(); let local = IdentityKey::new(KeyPair::generate(&mut rng).public_key); let remote = IdentityKey::new(KeyPair::generate(&mut rng).public_key); @@ -3047,6 +3151,111 @@ mod lease_reload_tests { ); } + /// A row whose lease is stranded above its chain (issue #1146: written by + /// a build that let a DH ratchet retire the chain without rebasing the + /// ceiling) cannot be fast-forwarded on recovery. It must not become a + /// hard error on every load: that strands the address, because the very + /// paths that would replace the session — the peer's next pre-key message + /// and the retry repair — have to load it first. Report it absent so the + /// no-session recovery replaces it. + #[tokio::test] + async fn an_unreadable_session_row_is_reported_absent_so_recovery_can_replace_it() { + let backend = InMemoryBackend::new(); + let cache = SignalStoreCache::with_max_entries_and_incarnation( + DEFAULT_MAX_CACHE_ENTRIES, + [0xA1; 16], + ); + let address = ProtocolAddress::new("15550001009", 1.into()); + + let mut stranded = leased_session(); + stranded.reserve_sender_chain_counters( + crate::libsignal::protocol::consts::MAX_RESERVATION_FAST_FORWARD, + ); + assert_eq!(session_chain_index(&stranded), 0); + cache.put_session(&address, stranded).await; + cache.flush(&backend).await.expect("flush"); + + // A live reload never fast-forwards, so the row still looks fine here. + cache.clear_after_flush().await; + assert!( + cache + .get_session(&address, &backend) + .await + .expect("live reload") + .is_some() + ); + + // A restart (or lossy reset) is where recovery has to fast-forward. + let restarted = SignalStoreCache::with_max_entries_and_incarnation( + DEFAULT_MAX_CACHE_ENTRIES, + [0xB2; 16], + ); + assert!( + restarted + .get_session(&address, &backend) + .await + .expect("an unreadable row must not fail the load") + .is_none() + ); + assert!( + !restarted + .has_session(&address, &backend) + .await + .expect("has_session"), + "the quarantined address must look session-less so ensure_e2e_sessions rebuilds it" + ); + } + + /// The existence probe on a cold cache is what decides whether a send + /// fetches a pre-key bundle, and it runs before anything loads the record. + /// Asking the backend whether the row exists answers `true` for a row the + /// very next checkout will discard, so the recovery is skipped and the send + /// either fails or drops that recipient from the fan-out. + /// + /// Distinct from the test above, which reaches `has_session` only after a + /// `get_session` has already negative-cached the address: that one passes + /// against the backend-existence probe too. + #[tokio::test] + async fn a_cold_existence_probe_does_not_report_a_quarantined_row_as_present() { + let backend = InMemoryBackend::new(); + let cache = SignalStoreCache::with_max_entries_and_incarnation( + DEFAULT_MAX_CACHE_ENTRIES, + [0xA1; 16], + ); + let address = ProtocolAddress::new("15550001010", 1.into()); + + let mut stranded = leased_session(); + stranded.reserve_sender_chain_counters( + crate::libsignal::protocol::consts::MAX_RESERVATION_FAST_FORWARD, + ); + cache.put_session(&address, stranded).await; + cache.flush(&backend).await.expect("flush"); + + // Nothing has touched this address in this incarnation: the probe is + // the first thing to reach the row, exactly as it is on a real restart. + let restarted = SignalStoreCache::with_max_entries_and_incarnation( + DEFAULT_MAX_CACHE_ENTRIES, + [0xB2; 16], + ); + assert!( + !restarted + .has_session(&address, &backend) + .await + .expect("a quarantined row must not fail the probe"), + "a row the next checkout would discard must not be reported present" + ); + + // And the negative answer is cached, so the send that follows keeps + // seeing it session-less rather than re-reading the same row. + assert!( + restarted + .get_session(&address, &backend) + .await + .expect("checkout") + .is_none() + ); + } + #[tokio::test] async fn incomplete_session_flush_retains_newer_state_and_fails_closed_on_recovery() { let backend = InMemoryBackend::new(); diff --git a/wacore/src/store/signal_cache_durability_chaos.rs b/wacore/src/store/signal_cache_durability_chaos.rs index 108f93566..7e38a5cdf 100644 --- a/wacore/src/store/signal_cache_durability_chaos.rs +++ b/wacore/src/store/signal_cache_durability_chaos.rs @@ -56,6 +56,7 @@ impl SenderKeyStore for CachedSenderKeyStore<'_> { enum Action { DmSend { fail_gate: bool }, GroupSend { fail_gate: bool }, + DmRatchet, DmCancel, DeliverGroup { newest: bool }, Flush, @@ -88,7 +89,7 @@ impl SplitMix64 { } fn action(&mut self) -> Action { - match self.next() % 25 { + match self.next() % 26 { 0 => Action::DmSend { fail_gate: true }, 1..=6 => Action::DmSend { fail_gate: false }, 7 => Action::GroupSend { fail_gate: true }, @@ -106,6 +107,7 @@ impl SplitMix64 { 22 => Action::DeleteGroup, 23 => Action::CheckoutDuringFlush, 24 => Action::RecoverGroup, + 25 => Action::DmRatchet, _ => unreachable!(), } } @@ -166,6 +168,7 @@ impl ChaosHarness { Action::GroupSend { fail_gate } => { self.group_send(fail_gate).await?; } + Action::DmRatchet => self.dm_ratchet().await?, Action::DmCancel => self.dm_cancel().await?, Action::DeliverGroup { newest } => { self.deliver_group(newest).await?; @@ -229,6 +232,38 @@ impl ChaosHarness { Ok(published) } + /// Inbound peer message that carries a new ratchet key: the sender chain + /// is replaced in place with fresh random material at counter zero and the + /// retired chain is discarded, not archived. This is a decrypt-side + /// advance, so it is dirty but never wire-gated. + /// + /// The replacement chain restarts the counter the record's lease bounds, + /// so without a rebase the ceiling drifts arbitrarily far above the live + /// index and recovery can no longer fast-forward it. Every interleaving + /// with reload, crash, and clear still has to satisfy `published_dm`. + async fn dm_ratchet(&mut self) -> Result<()> { + let (record, checkout) = self + .cache + .checkout_session(&self.dm_address, &self.backend) + .await?; + let Some(mut record) = record else { + self.cache + .cancel_session_checkout(&self.dm_address, checkout); + return Ok(()); + }; + let mut chain = [0; 32]; + self.crypto_rng.fill(&mut chain); + record + .session_state_mut() + .context("DM session state missing")? + .set_sender_chain( + &KeyPair::generate(&mut self.crypto_rng), + &ChainKey::new(chain, 0), + ); + record.rebase_lease_after_sender_chain_reset(); + self.commit_dm(record, checkout, true).await + } + async fn dm_cancel(&mut self) -> Result<()> { let (record, checkout) = self .cache diff --git a/wacore/src/telemetry.rs b/wacore/src/telemetry.rs index 72527c88b..403754eae 100644 --- a/wacore/src/telemetry.rs +++ b/wacore/src/telemetry.rs @@ -57,6 +57,13 @@ mod imp { pub fn base_key_collision() { counter!("wa_base_key_collision_total").increment(1); } + /// A stored Signal session blob could not be decoded and was reported as + /// absent so the no-session recovery can replace it. Steady state is zero: + /// a non-zero rate means rows are being written in a shape this build + /// cannot read back. + pub fn session_record_quarantined() { + counter!("wa_session_record_quarantined_total").increment(1); + } /// IQ request completed, by result (`ok`/`timeout`/`error`). Emitted at the /// single request chokepoint, so it covers both raw and spec-based IQs. pub fn iq(result: &'static str) { @@ -150,6 +157,11 @@ mod imp { Unit::Count, "Base-key collisions that forced a fresh session" ); + describe_counter!( + "wa_session_record_quarantined_total", + Unit::Count, + "Undecodable session rows reported as absent for recovery" + ); describe_counter!( "wa_iq_total", Unit::Count, @@ -228,6 +240,8 @@ mod imp { #[inline] pub fn base_key_collision() {} #[inline] + pub fn session_record_quarantined() {} + #[inline] pub fn iq(_result: &'static str) {} #[inline] pub fn reconnect() {}