diff --git a/crates/chatd/src/adapters/tiny.rs b/crates/chatd/src/adapters/tiny.rs index a175b16c..675dd864 100644 --- a/crates/chatd/src/adapters/tiny.rs +++ b/crates/chatd/src/adapters/tiny.rs @@ -73,6 +73,11 @@ pub fn prove_cache_history_limit() -> Result<(), String> { e2ee::prove_cache_history_limit() } +/// The lab's durable receive/cache commit proof. +pub fn prove_opened_history_cache_commit() -> Result<(), String> { + e2ee::prove_opened_history_cache_commit() +} + /// The lab's reauthentication history proof, using real encrypted device/cache files. pub fn prove_reauthenticated_history() -> Result<(), String> { e2ee::prove_reauthenticated_history() diff --git a/crates/chatd/src/adapters/tiny/e2ee.rs b/crates/chatd/src/adapters/tiny/e2ee.rs index c811ccdd..c7263ebd 100644 --- a/crates/chatd/src/adapters/tiny/e2ee.rs +++ b/crates/chatd/src/adapters/tiny/e2ee.rs @@ -621,23 +621,47 @@ impl Cache { } // MLS cannot open the same incoming ciphertext again after its ratchet advances. // This is durable history: dropping new rows at a size limit loses them on restart. - if let Ok(plain) = serde_json::to_vec(&opened) { - if let Ok(blob) = e2ee::store::seal_cache_value(key, id.as_bytes(), &plain) { - // An interrupted append may leave no newline. Fence the next complete record - // from that tail; readers already skip empty lines and invalid records. - let line = format!("\n{id} {}\n", STANDARD.encode(blob)); - let written = std::fs::OpenOptions::new() - .create(true) - .append(true) - .open(&self.path) - .and_then(|mut file| file.write_all(line.as_bytes())); - if let Err(err) = written { - tracing::warn!("chatd: Tiny e2ee: the cache could not be written: {err}"); - } - } + if let Err(err) = self.write_sealed(key, id, &opened) { + tracing::warn!("chatd: Tiny e2ee: the cache could not be written: {err}"); } + // Soft path (outgoing/import/siblings): keep the in-memory row even when the append + // failed. Receive must use [`Cache::put_durable`] so a failed append cannot claim success. self.rows.insert(id.to_owned(), opened); } + + /// Strict sealed append used on authenticated receive. Serialization, sealing, write or sync + /// failure is returned and the in-memory map is left unchanged. Unlike [`Cache::put`], a + /// failed append here cannot claim success. + fn put_durable(&mut self, key: &CacheKey, id: &str, opened: Opened) -> Result<(), String> { + if !super::safe_id(id) { + return Err("message id is not safe for the sealed cache".into()); + } + self.write_sealed(key, id, &opened)? + .sync_all().map_err(|err| err.to_string())?; + // A newly created log needs its directory entry durable before the ratchet is saved. + #[cfg(unix)] + std::fs::File::open(self.path.parent().ok_or("sealed cache has no directory")?) + .and_then(|dir| dir.sync_all()).map_err(|err| err.to_string())?; + self.rows.insert(id.to_owned(), opened); + Ok(()) + } + + fn write_sealed(&self, key: &CacheKey, id: &str, opened: &Opened) -> Result { + let plain = serde_json::to_vec(opened).map_err(|err| err.to_string())?; + let blob = e2ee::store::seal_cache_value(key, id.as_bytes(), &plain) + .map_err(|err| err.to_string())?; + // An interrupted append may leave no newline. Fence the next complete record + // from that tail; readers already skip empty lines and invalid records. + let line = format!("\n{id} {}\n", STANDARD.encode(blob)); + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&self.path) + .map_err(|err| err.to_string())?; + file.write_all(line.as_bytes()) + .map_err(|err| err.to_string())?; + Ok(file) + } } /// Daemon regression: a long conversation keeps newly opened history across restarts. @@ -718,6 +742,185 @@ pub fn prove_cache_append_tail() -> Result<(), String> { Ok(()) } +/// Lab proof: a failed sealed-cache append must not consume the receive ratchet; after repair the +/// same ciphertext opens and survives restart. Daemon marker only; not native GUI proof. +pub fn prove_opened_history_cache_commit() -> Result<(), String> { + for fail_cache in [false, true] { + let fixture = tempfile::tempdir().map_err(|err| err.to_string())?; + let a_dir = fixture.path().join("sender"); + let b_dir = fixture.path().join("receiver"); + private_dir(&a_dir).map_err(|err| err.to_string())?; + private_dir(&b_dir).map_err(|err| err.to_string())?; + let alice = UserId::random().map_err(|err| err.to_string())?; + let bea = UserId::random().map_err(|err| err.to_string())?; + let mut a = Engine::create( + alice, + Box::new(FileStore::new(&a_dir)), + StateKey::generate().map_err(|err| err.to_string())?, + STATE, + ) + .map_err(|err| err.to_string())?; + let state_key = StateKey::generate().map_err(|err| err.to_string())?; + let cache_key = CacheKey::generate().map_err(|err| err.to_string())?; + let mut b = Engine::create( + bea, + Box::new(FileStore::new(&b_dir)), + state_key.clone(), + STATE, + ) + .map_err(|err| err.to_string())?; + a.verify_devices( + bea, + &[DeviceRecord { + device_id: b.device_id(), + signature_key: b.signature_key(), + cross_signature: None, + created_at: 1, + }], + ) + .map_err(|err| err.to_string())?; + b.verify_devices( + alice, + &[DeviceRecord { + device_id: a.device_id(), + signature_key: a.signature_key(), + cross_signature: None, + created_at: 1, + }], + ) + .map_err(|err| err.to_string())?; + let conv = ConversationId::random().map_err(|err| err.to_string())?; + let server = e2ee::identity::DeviceKey::generate().map_err(|err| err.to_string())?; + let public = server.public_key(); + a.create_group(conv, &public).map_err(|err| err.to_string())?; + let package = b + .key_packages(1) + .map_err(|err| err.to_string())? + .remove(0); + let crypto = openmls_rust_crypto::OpenMlsRustCrypto::default(); + use openmls::prelude::tls_codec::{DeserializeBytes as _, Serialize as _}; + use openmls::prelude::{ + ExternalProposal, GroupEpoch, GroupId, KeyPackageIn, MlsMessageOut, ProtocolVersion, + SenderExtensionIndex, + }; + use openmls_traits::OpenMlsProvider as _; + let parsed = KeyPackageIn::tls_deserialize_exact_bytes(&package) + .map_err(|err| err.to_string())? + .validate(crypto.crypto(), ProtocolVersion::Mls10) + .map_err(|err| err.to_string())?; + let proposal: MlsMessageOut = ExternalProposal::new_add::( + parsed, + GroupId::from_slice(&conv.0), + GroupEpoch::from(a.epoch(conv).map_err(|err| err.to_string())?), + &server, + SenderExtensionIndex::new(0), + ) + .map_err(|err| err.to_string())?; + let proposals = vec![proposal] + .tls_serialize_detached() + .map_err(|err| err.to_string())?; + let outcome = a + .handle_proposals(conv, &proposals) + .map_err(|err| err.to_string())?; + if !outcome.rejected.is_empty() { + return Err("add proposal was rejected".into()); + } + let bundle = a + .commit_pending(conv) + .map_err(|err| err.to_string())? + .ok_or("missing commit bundle")?; + a.confirm_commit(conv).map_err(|err| err.to_string())?; + b.handle_welcome(conv, &bundle.welcomes[0].data, Some(&public)) + .map_err(|err| err.to_string())?; + let sent = a + .encrypt( + conv, + Kind::Message, + None, + &Envelope::message("fixture history before restart"), + ) + .map_err(|err| err.to_string())?; + let id = conv.to_string(); + let cache_path = b_dir.join(format!("cache-{id}.log")); + if fail_cache { + std::fs::create_dir(&cache_path).map_err(|err| err.to_string())?; + } + let saved = Saved { + base_url: "https://prove.invalid".into(), + token: "synthetic".into(), + user_id: bea.to_string(), + email: None, + display_name: "Fixture".into(), + role: "member".into(), + plain_http_ok: false, + e2ee: None, + }; + let author = alice.to_string(); + let from = a.device_id().to_string(); + let ciphertext = b64(&sent.ciphertext); + let row = Row { + id: "m1", + author_id: &author, + device_id: &from, + kind: "message", + ciphertext: Some(&ciphertext), + epoch: sent.epoch, + generation: None, + group_id: None, + target_id: None, + client_id: None, + created_at: "2026-01-01T00:00:00Z", + }; + let before = std::fs::read(b_dir.join(STATE)).map_err(|err| err.to_string())?; + let mut device = Device::new(&saved); + device.dir = b_dir.clone(); + device.engine = Some(b); + device.cache_key = Some(cache_key.clone()); + device.state = DeviceState::Ready; + let first = device.read(&id, std::slice::from_ref(&row)); + if fail_cache { + if !matches!(first.get("m1"), Some(Read::Undecryptable)) { + return Err("failed cache commit claimed an opened row".into()); + } + if std::fs::read(b_dir.join(STATE)).map_err(|err| err.to_string())? != before { + return Err("receive ratchet advanced despite a failed cache append".into()); + } + std::fs::remove_dir(&cache_path).map_err(|err| err.to_string())?; + let repaired = device.read(&id, std::slice::from_ref(&row)); + if !matches!( + repaired.get("m1"), + Some(Read::Opened(o)) if o.text == "fixture history before restart" + ) { + return Err("ciphertext did not open after cache path repair".into()); + } + } else if !matches!( + first.get("m1"), + Some(Read::Opened(o)) if o.text == "fixture history before restart" + ) { + return Err("control receive did not open".into()); + } + drop(device); + let b = Engine::open(Box::new(FileStore::new(&b_dir)), state_key, STATE) + .map_err(|err| err.to_string())? + .ok_or("receiver state missing after restart")?; + let mut restarted = Device::new(&saved); + restarted.dir = b_dir; + restarted.engine = Some(b); + restarted.cache_key = Some(cache_key); + restarted.state = DeviceState::Ready; + let after = restarted.read(&id, std::slice::from_ref(&row)); + if !matches!( + after.get("m1"), + Some(Read::Opened(o)) if o.text == "fixture history before restart" + ) { + return Err(format!( + "history did not survive restart (fail_cache={fail_cache})" + )); + } + } + Ok(()) +} + /// Lab proof: renewing a session still opens the original device and sealed history. pub fn prove_reauthenticated_history() -> Result<(), String> { let dir = tempfile::tempdir().map_err(|err| err.to_string())?; @@ -2049,6 +2252,31 @@ impl Device { } } + /// Durable receive commit: sealed cache append must succeed before the advanced ratchet is + /// kept. Failures leave no in-memory opened row. + fn remember_durable( + &mut self, + conversation_id: &str, + id: &str, + opened: Opened, + ) -> Result<(), Error> { + let Some(key) = self.cache_key.clone() else { + return Err(Error::Storage("sealed cache key is missing".into())); + }; + let Some(cache) = self.cache(conversation_id) else { + return Err(Error::Storage("sealed cache could not be opened".into())); + }; + cache.put_durable(&key, id, opened.clone()).map_err(|err| { + tracing::warn!("chatd: Tiny e2ee: opened history could not be saved: {err}"); + Error::Storage(err) + })?; + for file in &opened.attachments { + self.files + .insert(file.id.clone(), (conversation_id.to_owned(), file.clone())); + } + Ok(()) + } + /// What a page of rows says, oldest first, the edits folded into their targets. pub(crate) fn read(&mut self, conversation_id: &str, rows: &[Row]) -> HashMap { let mut out = HashMap::new(); @@ -2136,10 +2364,20 @@ impl Device { }; let (Ok(kind), Ok(author), Ok(device)) = (Kind::parse(row.kind), UserId::parse(row.author_id), DeviceId::parse(row.device_id)) else { return fallback(); }; let claimed = Claimed { author: Some(author), device: Some(device), kind, target_id: row.target_id }; - let Some(engine) = self.engine.as_mut() else { + let Some(mut engine) = self.engine.take() else { return fallback(); }; - let mut opened = engine.decrypt(conv, &ciphertext, &claimed); + let mut opened = Self::receive_committed( + &mut engine, + self, + conversation_id, + conv, + row, + &ciphertext, + &claimed, + kind, + legacy.as_ref(), + ); // FutureEpoch: behind on handshake. NoGroup / NotAMember: accepted a repair but not // the Welcome yet. Limit one sync attempt per encryption group on this page. if matches!( @@ -2147,6 +2385,7 @@ impl Device { Err(Error::FutureEpoch { .. } | Error::NoGroup | Error::NotAMember) ) && synced.insert(conv) { + self.engine = Some(engine); let generation = row.generation.unwrap_or(0); let outer = ConversationId::parse(conversation_id).ok(); let active = outer.and_then(|outer| self.engine.as_ref()?.active_group(outer).ok()); @@ -2156,38 +2395,28 @@ impl Device { self.feed_at(conversation_id, conv, generation, &key).map(|_| ()) } else { self.sync(conversation_id) }; if let Err(why) = sync { tracing::warn!("chatd: Tiny e2ee: {conversation_id}: {why}"); } - let Some(engine) = self.engine.as_mut() else { + let Some(mut again) = self.engine.take() else { return fallback(); }; - opened = engine.decrypt(conv, &ciphertext, &claimed); + opened = Self::receive_committed( + &mut again, + self, + conversation_id, + conv, + row, + &ciphertext, + &claimed, + kind, + legacy.as_ref(), + ); + engine = again; } + self.engine = Some(engine); if matches!(opened, Err(Error::FutureEpoch { .. } | Error::NoGroup)) { self.unopened.insert(conversation_id.to_owned()); } match opened { - Ok(incoming) => { - let opened = Opened { - text: incoming.envelope.text.clone(), - generation: Some(row.generation.unwrap_or(0)), - auth: Some(Authentication::new(ConversationId::parse(conversation_id).expect("checked conversation"), conv, row.generation.unwrap_or(0), incoming.epoch, &incoming.sender, incoming.kind, incoming.target_id.as_deref())), - saved_attribution: None, - reply_to: incoming.envelope.reply_to.clone(), - sent_at: incoming.envelope.sent_at, - attachments: incoming - .envelope - .attachments - .iter() - .map(OpenedFile::of) - .collect(), - edits: incoming.target_id.clone().filter(|_| kind == Kind::Edit), - edited_by: None, - at: super::messages::unix_of(row.created_at).unwrap_or(0), - imported_from: None, - }; - if legacy.as_ref().is_some_and(|saved| saved.text != opened.text || saved.attachments != opened.attachments || saved.edited_by.is_some()) { - return fallback(); - } - self.remember(conversation_id, row.id, opened.clone()); + Ok(opened) => { if kind == Kind::Edit { Read::Edit } else { @@ -2201,6 +2430,65 @@ impl Device { } } + /// Decrypt, durably seal the authenticated opened row, then persist the advanced ratchet. + /// A refused cache commit restores the pre-read ratchet so the genuine ciphertext still opens + /// after storage is repaired. + fn receive_committed( + engine: &mut Engine, + device: &mut Device, + conversation_id: &str, + conv: ConversationId, + row: &Row<'_>, + ciphertext: &[u8], + claimed: &Claimed<'_>, + kind: Kind, + legacy: Option<&Opened>, + ) -> Result { + let mut committed = None; + engine.decrypt_committed(conv, ciphertext, claimed, |incoming| { + let opened = Opened { + text: incoming.envelope.text.clone(), + generation: Some(row.generation.unwrap_or(0)), + auth: Some(Authentication::new( + ConversationId::parse(conversation_id).expect("checked conversation"), + conv, + row.generation.unwrap_or(0), + incoming.epoch, + &incoming.sender, + incoming.kind, + incoming.target_id.as_deref(), + )), + saved_attribution: None, + reply_to: incoming.envelope.reply_to.clone(), + sent_at: incoming.envelope.sent_at, + attachments: incoming + .envelope + .attachments + .iter() + .map(OpenedFile::of) + .collect(), + edits: incoming.target_id.clone().filter(|_| kind == Kind::Edit), + edited_by: None, + at: super::messages::unix_of(row.created_at).unwrap_or(0), + imported_from: None, + }; + if !opened.auth.as_ref().is_some_and(|auth| auth.matches(conversation_id, row)) { + return Err(Error::Authentication); + } + if legacy.is_some_and(|saved| { + saved.text != opened.text + || saved.attachments != opened.attachments + || saved.edited_by.is_some() + }) { + return Err(Error::Authentication); + } + device.remember_durable(conversation_id, row.id, opened.clone())?; + committed = Some(opened); + Ok(()) + })?; + committed.ok_or_else(|| Error::Storage("receive commit produced no opened row".into())) + } + /// Encrypts a message or an edit and posts it. The plaintext is kept in the cache, under the /// id the server gives it. pub(crate) fn post( @@ -3140,6 +3428,36 @@ mod tests { assert_eq!(prove_cache_history_limit(), Ok(())); } + #[test] + fn opened_history_cache_commit_prove_matches_receive_regression() { + assert_eq!(prove_opened_history_cache_commit(), Ok(())); + } + + #[test] + fn cache_attachment_registration_follows_commit_mode() { + let (mut device, _) = posting_device("cache-attachment-commit", 0); + let fixture = tempfile::tempdir().unwrap(); + device.dir = fixture.path().to_owned(); + let file = OpenedFile { + id: "11111111-1111-4111-8111-111111111111".into(), key: b64(&[7; 32]), sha256: b64(&[9; 32]), + name: "picture.png".into(), mime: "image/png".into(), bytes: 1, + ..OpenedFile::default() + }; + let mut opened = Opened { attachments: vec![file], ..Opened::default() }; + device.remember("33333333-3333-4333-8333-333333333333", "m1", opened.clone()); + assert!(device.sealed_file("11111111-1111-4111-8111-111111111111").is_some(), + "outgoing/import history must retain immediate attachment lookup"); + opened.attachments[0].id = "22222222-2222-4222-8222-222222222222".into(); + let blocked = fixture.path().join("cache-44444444-4444-4444-8444-444444444444.log"); + std::fs::create_dir(&blocked).unwrap(); + assert!(device.remember_durable("44444444-4444-4444-8444-444444444444", "m2", opened.clone()).is_err()); + assert!(device.sealed_file("22222222-2222-4222-8222-222222222222").is_none(), + "a refused durable commit must not register its attachment"); + std::fs::remove_dir(&blocked).unwrap(); + device.remember_durable("44444444-4444-4444-8444-444444444444", "m2", opened).unwrap(); + assert!(device.sealed_file("22222222-2222-4222-8222-222222222222").is_some()); + } + #[test] fn a_recovery_key_is_128_bits_in_six_groups_and_reads_back() { let (text, bytes) = new_recovery_key().unwrap(); diff --git a/crates/chatd/src/adapters/tiny/e2ee/receive_tests.rs b/crates/chatd/src/adapters/tiny/e2ee/receive_tests.rs index ca96760c..f3dd1038 100644 --- a/crates/chatd/src/adapters/tiny/e2ee/receive_tests.rs +++ b/crates/chatd/src/adapters/tiny/e2ee/receive_tests.rs @@ -304,6 +304,148 @@ fn add( (bundle, proposals) } +/// A failed sealed-cache append must not leave only a consumed receive ratchet. After storage is +/// repaired, the same ciphertext opens and survives a cold restart with the same keys. +#[test] +fn opened_history_survives_a_failed_cache_append_and_restart() { + assert_eq!(prove_opened_history_cache_commit(), Ok(())); +} + +/// Cache durable + state save lagging: restart serves the sealed row; a mismatched claim still +/// spends no key and does not replace authenticated cache metadata. +#[test] +fn sealed_cache_survives_state_save_lag_and_rejects_mismatched_replay() { + let fixture = tempfile::tempdir().unwrap(); + let a_dir = fixture.path().join("sender"); + let b_dir = fixture.path().join("receiver"); + private_dir(&a_dir).unwrap(); + private_dir(&b_dir).unwrap(); + let alice = UserId::random().unwrap(); + let bea = UserId::random().unwrap(); + let mut a = Engine::create( + alice, + Box::new(FileStore::new(&a_dir)), + StateKey::generate().unwrap(), + STATE, + ) + .unwrap(); + let state_key = StateKey::generate().unwrap(); + let cache_key = CacheKey::generate().unwrap(); + let mut b = Engine::create( + bea, + Box::new(FileStore::new(&b_dir)), + state_key.clone(), + STATE, + ) + .unwrap(); + a.verify_devices(bea, &[record(&b, None, 1)]).unwrap(); + b.verify_devices(alice, &[record(&a, None, 1)]).unwrap(); + let conv = ConversationId::random().unwrap(); + let server = DeviceKey::generate().unwrap(); + let public = server.public_key(); + a.create_group(conv, &public).unwrap(); + let (bundle, _) = add(&mut a, conv, &server, &b.key_packages(1).unwrap().remove(0)); + b.handle_welcome(conv, &bundle.welcomes[0].data, Some(&public)) + .unwrap(); + let sent = a + .encrypt( + conv, + Kind::Message, + None, + &Envelope::message("lagged state history"), + ) + .unwrap(); + let id = conv.to_string(); + let before = std::fs::read(b_dir.join(STATE)).unwrap(); + let saved = Saved { + base_url: "https://diagnostic.invalid".into(), + token: "synthetic".into(), + user_id: bea.to_string(), + email: None, + display_name: "Fixture".into(), + role: "member".into(), + plain_http_ok: false, + e2ee: None, + }; + let author = alice.to_string(); + let from = a.device_id().to_string(); + let ciphertext = b64(&sent.ciphertext); + let row = Row { + id: "m1", + author_id: &author, + device_id: &from, + kind: "message", + ciphertext: Some(&ciphertext), + epoch: sent.epoch, + generation: None, + group_id: None, + target_id: None, + client_id: None, + created_at: "2026-01-01T00:00:00Z", + }; + let mut device = { + let mut device = Device::new(&saved); + device.dir = b_dir.clone(); + device.engine = Some(b); + device.cache_key = Some(cache_key.clone()); + device.state = DeviceState::Ready; + device + }; + let wrong_epoch = Row { epoch: sent.epoch + 1, ..row }; + let refused = device.read(&id, std::slice::from_ref(&wrong_epoch)); + assert!(matches!(refused.get("m1"), Some(Read::Undecryptable)), + "wire context must match before the receive is durably accepted"); + assert_eq!(std::fs::read(b_dir.join(STATE)).unwrap(), before, + "a mismatched row must not consume the genuine ciphertext"); + assert!(device.cache(&id).unwrap().rows.get("m1").is_none()); + let opened = device.read(&id, std::slice::from_ref(&row)); + assert!( + matches!(opened.get("m1"), Some(Read::Opened(o)) if o.text == "lagged state history") + ); + // Simulate the opposite cut: durable cache present, on-disk ratchet not yet advanced. + std::fs::write(b_dir.join(STATE), &before).unwrap(); + drop(device); + let b = Engine::open(Box::new(FileStore::new(&b_dir)), state_key.clone(), STATE) + .unwrap() + .unwrap(); + let mut restarted = { + let mut device = Device::new(&saved); + device.dir = b_dir.clone(); + device.engine = Some(b); + device.cache_key = Some(cache_key.clone()); + device.state = DeviceState::Ready; + device + }; + let after = restarted.read(&id, std::slice::from_ref(&row)); + assert!( + matches!(after.get("m1"), Some(Read::Opened(o)) if o.text == "lagged state history"), + "sealed cache must serve history when the ratchet snapshot lags" + ); + let other = UserId::random().unwrap().to_string(); + let mismatched = Row { + id: "m1", + author_id: &other, + device_id: &from, + kind: "message", + ciphertext: Some(&ciphertext), + epoch: sent.epoch, + generation: None, + group_id: None, + target_id: None, + client_id: None, + created_at: "2026-01-01T00:00:00Z", + }; + let refused = restarted.read(&id, std::slice::from_ref(&mismatched)); + assert!( + matches!(refused.get("m1"), Some(Read::Undecryptable)), + "cached auth must refuse a mismatched claim without replacing the sealed row" + ); + let again = restarted.read(&id, std::slice::from_ref(&row)); + assert!( + matches!(again.get("m1"), Some(Read::Opened(o)) if o.text == "lagged state history") + ); +} + #[test] fn mixed_generation_receive_does_not_spend_current_welcome_on_blocked_historical_rows() { let alice = UserId::parse("11111111-1111-4111-8111-111111111111").unwrap(); diff --git a/crates/chatd/src/main.rs b/crates/chatd/src/main.rs index d9a792a8..a96360aa 100644 --- a/crates/chatd/src/main.rs +++ b/crates/chatd/src/main.rs @@ -127,6 +127,15 @@ fn main() -> io::Result<()> { eprintln!("prove-cache-append-tail:0"); return Ok(()); } + // A failed sealed-cache append must not leave only a consumed receive ratchet. + if std::env::var("TINYCLIENT_PROVE").ok().as_deref() == Some("opened-history-cache-commit") { + if let Err(err) = chatd::adapters::tiny::prove_opened_history_cache_commit() { + eprintln!("opened-history-cache-commit: {err}"); + std::process::exit(1); + } + eprintln!("prove-opened-history-cache-commit:0"); + return Ok(()); + } // The lab's keep-opened-sealed driver. A build without this returns to the server below and stays up. if std::env::var("TINYCLIENT_PROVE").ok().as_deref() == Some("keep-opened-sealed") { if let Err(err) = chatd::adapters::tiny::prove_keep_opened_sealed() { diff --git a/crates/chatd/src/recorder.rs b/crates/chatd/src/recorder.rs index 2c662bba..ebea4f33 100644 --- a/crates/chatd/src/recorder.rs +++ b/crates/chatd/src/recorder.rs @@ -254,7 +254,14 @@ mod tests { std::thread::sleep(Duration::from_millis(5)); } drop(hub); - std::fs::remove_dir_all(dir).unwrap(); + // Recorded is published before the reader exits and drops its Hub/SQLite handle. + // Windows cannot delete the database until that asynchronous cleanup finishes. + let cleanup_deadline = Instant::now() + Duration::from_secs(5); + while let Err(err) = std::fs::remove_dir_all(&dir) { + assert!(cfg!(windows) && err.raw_os_error() == Some(32) + && Instant::now() < cleanup_deadline, "recorder fixture cleanup: {err}"); + std::thread::sleep(Duration::from_millis(5)); + } } fn recording(since: u64) -> Recording { diff --git a/crates/e2ee/src/messaging.rs b/crates/e2ee/src/messaging.rs index afef4e2e..3537c0b6 100644 --- a/crates/e2ee/src/messaging.rs +++ b/crates/e2ee/src/messaging.rs @@ -243,6 +243,50 @@ impl Engine { ciphertext: &[u8], claimed: &Claimed<'_>, ) -> Result { + // Existing callers keep the lenient save: a receive that also writes opened history must + // use [`Engine::decrypt_committed`] so a failed durable commit cannot leave only a consumed + // ratchet. + let (incoming, checkpoint) = self.decrypt_open(conv, ciphertext, claimed)?; + drop(checkpoint); + self.persist(false)?; + Ok(incoming) + } + + /// Like [`Engine::decrypt`], but runs `commit` while the receive checkpoint is still + /// reversible and before the advanced ratchet is written. If `commit` fails, the pre-read + /// state is restored and the receive spends no key. After a successful commit, the caller + /// must already have durable recovery data. State saving then follows [`Engine::decrypt`]: + /// a failed save leaves the accepted receive consumed in memory and reports the error on + /// [`Engine::flush`]. It must never rewind an accepted receive to match a lagging disk. + pub fn decrypt_committed( + &mut self, + conv: ConversationId, + ciphertext: &[u8], + claimed: &Claimed<'_>, + commit: F, + ) -> Result + where + F: FnOnce(&Incoming) -> Result<()>, + { + let (incoming, checkpoint) = self.decrypt_open(conv, ciphertext, claimed)?; + if let Err(error) = commit(&incoming) { + self.provider.restore(checkpoint); + self.groups.remove(&conv); + return Err(error); + } + drop(checkpoint); + self.persist(false)?; + Ok(incoming) + } + + /// Authenticate and open a ciphertext, leaving the provider checkpoint for the caller to + /// drop (accept) or restore (refuse). + fn decrypt_open( + &mut self, + conv: ConversationId, + ciphertext: &[u8], + claimed: &Claimed<'_>, + ) -> Result<(Incoming, crate::provider::Checkpoint)> { self.repair_receive_allowed(conv)?; self.usable(conv)?; self.load_group(conv)?; @@ -339,17 +383,18 @@ impl Engine { return Err(error); } }; - drop(checkpoint); let sender_trusted = trust.is_trusted(sender.user, sender.device, &sender.signature_key); - self.persist(false)?; - Ok(Incoming { - sender, - sender_trusted, - epoch, - kind: claimed.kind, - target_id: claimed.target_id.map(str::to_owned), - envelope, - }) + Ok(( + Incoming { + sender, + sender_trusted, + epoch, + kind: claimed.kind, + target_id: claimed.target_id.map(str::to_owned), + envelope, + }, + checkpoint, + )) } /// A ciphertext under this device's leaf, at the generation its next real message will use, diff --git a/crates/e2ee/tests/store.rs b/crates/e2ee/tests/store.rs index e1578e1f..f19ec96d 100644 --- a/crates/e2ee/tests/store.rs +++ b/crates/e2ee/tests/store.rs @@ -177,6 +177,105 @@ fn encrypting_and_making_key_packages_fail_when_the_state_cannot_be_written() { assert_eq!(store.saves.load(Ordering::SeqCst), before + 1); } +#[test] +fn decrypt_committed_restores_ratchet_when_commit_refuses() { + let mut world = World::new(); + let a = world.person("a"); + let b = world.person("b"); + world.sync_trust(); + world.start(a); + world.change(&[a], a, &[b], &[]); + let conv = world.conv(); + let sent = world.send(a, "must stay openable after a refused commit"); + let claimed = Claimed { + author: Some(world.clients[a].user()), + device: Some(world.clients[a].device()), + kind: Kind::Message, + target_id: None, + }; + let err = world.clients[b] + .engine + .decrypt_committed(conv, &sent, &claimed, |_| Err(Error::Storage("cache full".into()))) + .expect_err("refused commit must surface"); + assert!(matches!(err, Error::Storage(_))); + assert_eq!( + world.clients[b] + .engine + .decrypt(conv, &sent, &claimed) + .unwrap() + .envelope + .text, + "must stay openable after a refused commit" + ); +} + +#[test] +fn decrypt_committed_keeps_plaintext_when_state_save_lags() { + let store = Shared::default(); + let key = StateKey::generate().unwrap(); + let mut world = World::new(); + let a = world.person("a"); + let mut bob = + Engine::create(UserId::random().unwrap(), Box::new(store.clone()), key.clone(), "s") + .unwrap(); + world.sync_trust(); + let alice_user = world.clients[a].user(); + bob.verify_devices(alice_user, &world.directory[&alice_user].clone()) + .unwrap(); + world.start(a); + let conv = world.conv(); + let bob_record = e2ee::identity::DeviceRecord { + device_id: bob.device_id(), + signature_key: bob.signature_key(), + cross_signature: None, + created_at: 1, + }; + world.clients[a] + .engine + .verify_devices(bob.user_id(), &[bob_record]) + .unwrap(); + let package = bob.key_packages(1).unwrap().remove(0); + world.clients[a] + .engine + .handle_proposals(conv, &world.ds.proposals(&[&package], &[], 0)) + .unwrap(); + let bundle = world.clients[a] + .engine + .commit_pending(conv) + .unwrap() + .unwrap(); + world.clients[a].engine.confirm_commit(conv).unwrap(); + bob.handle_welcome(conv, &bundle.welcomes[0].data, Some(&world.ds.public())) + .unwrap(); + let sent = world.send(a, "opened while state lags"); + let claimed = Claimed { + author: Some(alice_user), + device: None, + kind: Kind::Message, + target_id: None, + }; + let before = store.load("s").unwrap().unwrap(); + store.fail.store(true, Ordering::SeqCst); + let mut committed = false; + let opened = bob + .decrypt_committed(conv, &sent, &claimed, |_| { + committed = true; + Ok(()) + }) + .expect("plaintext remains available when only the state save fails"); + assert!(committed); + assert_eq!(opened.envelope.text, "opened while state lags"); + assert_eq!(store.load("s").unwrap().unwrap(), before); + assert!(bob.flush().is_err(), "the failed state save must remain visible"); + store.fail.store(false, Ordering::SeqCst); + // An accepted receive must stay consumed in memory even while its state save lags. + assert!(matches!(bob.decrypt(conv, &sent, &claimed), Err(Error::Undecryptable)), + "state-save failure must not rewind an accepted receive"); + bob.flush().unwrap(); + let mut reopened = Engine::open(Box::new(store.clone()), key, "s").unwrap().unwrap(); + assert!(matches!(reopened.decrypt(conv, &sent, &claimed), Err(Error::Undecryptable))); +} + #[test] fn receiving_survives_a_failed_write_and_reports_it_on_flush() { let store = Shared::default(); diff --git a/e2e/scenarios/opened-history-cache-commit/README.md b/e2e/scenarios/opened-history-cache-commit/README.md new file mode 100644 index 00000000..840c4c47 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/README.md @@ -0,0 +1,78 @@ +# Opened history before receive-state persistence + +A failed sealed-cache append must not leave only a consumed receive ratchet. This +scenario drives the **normal Windows or Linux X11 app** (release-profile `tinyclient` + `chatd` ++ `mediad`) against an owned disposable Tiny backend and synthetic people. + +## Matched case + +1. Prepare a valid encrypted conversation and receiver device with no prior target + cache log. Wrong login, missing group, or failed setup is not a valid RED. +2. Block only the target sealed-cache append with an empty directory at its + otherwise absent `cache-.log` path. Parent/state store stays + writable. An owned peer delivers a unique encrypted marker. +3. Under the fault, the base must actually decrypt (so later loss is not missing + keys). The fixed build must refuse a durable open (a temporary stub is not + success). +4. Cold-stop owned processes, remove only the blocker, empty only this fixture's + SQLite message index, keep conversation/MLS/keys, and restart. +5. Shared assertion: `opened plaintext is still present after cache-fault repair + and cold restart` — fails on the base for that history-loss reason, passes on + the branch. + +## Layout + +| File | Role | +| --- | --- | +| `fixture_controller.py` | Synthetic fixture paths, empty-dir blocker, SQLite wipe | +| `session_contract.py` | Literal-space `TINYCLIENT_SESSION_TINY SERVER` env | +| `ipc.py` | Fixture IPC via Windows `dlctl` or Linux profile Unix socket (not native proof) | +| `native_windows.py` / `owned_input.ps1` | Person-facing window input/pixels | +| `native_linux.py` | Owned executable/PID/foreground guards, X11 input and captured-image OCR | +| `drive.py` / platform launchers | Orchestrates the matched case | + +`TINYCLIENT_PROVE` is refused. Diagnostic IPC may confirm decrypt/refuse; it does +not replace screenshots or window actions. Recordings follow `e2e/README.md` +(recorder before first step, playable mp4, ≥360px short side, SHA/step burn-in, +`steps.jsonl` / `proof.json`). + +The conversation pixel oracle uses the Windows OCR API or Linux Tesseract word geometry. +Linux enlarges the captured image before OCR so small glyphs retain punctuation; +the exact marker match and conversation-pane boundary remain unchanged. +Missing OCR or a missing window is setup-incomplete. Both phases +must observe the exact server-assigned message ID in a scoped messages page; +an empty page while asynchronous backfill is running is never history-loss RED. +The peer is opened through the normal Ctrl+K Jump command, with literal query text +observed before Enter and foreground-PID guards. There are no guessed coordinates. +Message pixels are checked below the fresh-conversation intro in the opened pane; +a matching sidebar preview cannot pass that assertion. +Use a single-word synthetic display name per peer (for example Ada and Bea); +preflight rejects unsupported names before any profile is changed. + +## Environment + +- `E2E_OUT`, `E2E_T0`, `E2E_APP_DIR` — lab contract +- `E2E_CACHE_COMMIT_FIXTURE` — absolute JSON describing owned profiles/sessions, + conversation id, and marker (`opened-history-cache-commit-…`) +- `E2E_DLCTL` — optional absolute `dlctl` when it is not beside the app bundle + +Linux requires an X11 session with `DISPLAY` and its authentication environment, plus `wmctrl`, +`xdotool`, `ffmpeg` and `tesseract`. A missing dependency or Wayland session is setup-incomplete. +Before each input, the adapter checks the foreground window's PID and the owned executable. +Linux fixture IPC reads the same four-byte-length-prefixed JSON frames from the profile's +`chat.sock`; an explicit `E2E_DLCTL` override retains the helper route. + +Each fixture session must include persistent `e2ee.state` and `e2ee.cache` wrapping +keys (base64, 32 bytes each), created once for a NEW fixture before its first app +launch. Missing/invalid keys refuse setup before any profile is changed. Keep +these synthetic keys private; never regenerate them to resume an existing device. + +No real host/provider names, IPs, SIDs, tokens, or old user data belong in public +scripts or recordings. Fixture bootstrap stays env-driven and private. + +## Scope + +Windows and Linux X11 native adapters are available. This does not establish aggregate +`e2e-proof` / release status across platforms. Component daemon prove modes are +not the native evidence for this scenario. No production snapshot rewind, key +reset, or replay/attribution relaxation. diff --git a/e2e/scenarios/opened-history-cache-commit/drive.linux.sh b/e2e/scenarios/opened-history-cache-commit/drive.linux.sh new file mode 100644 index 00000000..352b6cbe --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/drive.linux.sh @@ -0,0 +1,3 @@ +#!/usr/bin/env bash +set -euo pipefail +exec python3 "$(dirname "$0")/drive.py" diff --git a/e2e/scenarios/opened-history-cache-commit/drive.macos.sh b/e2e/scenarios/opened-history-cache-commit/drive.macos.sh new file mode 100644 index 00000000..f2967273 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/drive.macos.sh @@ -0,0 +1,24 @@ +#!/usr/bin/env bash +# opened-history-cache-commit: person-facing proof is Windows-native only. +# A missing-platform run must not be labeled as the sealed-cache history-loss RED. +set -euo pipefail +OUT="${E2E_OUT:?}" +T0="${E2E_T0:-0}" +mkdir -p "$OUT" +python3 - <<'PY' "$OUT" "$T0" +import json, sys, time +out, t0 = sys.argv[1], float(sys.argv[2]) +steps = out + "/steps.jsonl" +open(steps, "w").close() +line = { + "t": round(time.time() - t0, 1), + "step": "Windows native pixel driver is required; macOS is not claimed for this matched case", + "ok": False, +} +open(steps, "a", encoding="utf-8").write(json.dumps(line) + "\n") +print("[FAILED]", line["step"], flush=True) +open(out + "/setup-incomplete.json", "w", encoding="utf-8").write( + json.dumps({"error": "macos-not-claimed", "valid_red": False}, indent=2) + "\n" +) +sys.exit(1) +PY diff --git a/e2e/scenarios/opened-history-cache-commit/drive.py b/e2e/scenarios/opened-history-cache-commit/drive.py new file mode 100644 index 00000000..80b75a1d --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/drive.py @@ -0,0 +1,446 @@ +#!/usr/bin/env python3 +"""Normal-app sealed-cache append fault: base loses history; branch keeps it. + +This driver does NOT use TINYCLIENT_PROVE. Fixture setup lives in +fixture_controller.py; person-facing observation uses the native platform adapter. + +Matched case (same protocol on base and branch): + 1. Owned synthetic receiver/sender profiles from E2E_CACHE_COMMIT_FIXTURE + 2. Empty-directory blocker on the absent sealed-cache log path + 3. Owned peer sends a unique encrypted marker + 4. Observe under fault (base must decrypt; fixed must refuse durable open) + 5. Cold stop; remove blocker; wipe only this fixture's SQLite message index + 6. Restart and assert the shared step: + "opened plaintext is still present after cache-fault repair and cold restart" + +Env from the lab: E2E_OUT, E2E_T0, E2E_APP_DIR, E2E_CACHE_COMMIT_FIXTURE. +E2E_SABOTAGE=1 forces a failing observation without claiming history-loss RED. +Wrong session/login/group or missing fixture is setup-incomplete, not valid RED. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +import uuid +from pathlib import Path + +import fixture_controller as fixture +import ipc +import session_contract + +OUT = os.environ.get("E2E_OUT", "") +T0 = float(os.environ.get("E2E_T0") or time.time()) +APP = os.environ.get("E2E_APP_DIR", "") +SABOTAGE = os.environ.get("E2E_SABOTAGE") == "1" + +FAILED = False +STEPS = "" +STARTED: list[subprocess.Popen] = [] + + +def now() -> float: + return time.time() - T0 + + +def step(words: str, ok: bool, shot: str = "") -> None: + global FAILED + if not ok: + FAILED = True + payload = {"t": round(now(), 1), "step": words, "ok": bool(ok)} + if shot: + payload["shot"] = shot + line = json.dumps(payload, ensure_ascii=True) + print(("[ok] " if ok else "[FAILED] ") + words, flush=True) + if STEPS: + with open(STEPS, "a", encoding="utf-8") as handle: + handle.write(line + "\n") + + +def binaries(app_dir: str): + if sys.platform == "win32": + return ( + os.path.join(app_dir, "chatd.exe"), + os.path.join(app_dir, "tinyclient.exe"), + os.path.join(app_dir, "mediad.exe"), + ) + if sys.platform == "darwin" and app_dir.endswith(".app"): + mac = os.path.join(app_dir, "Contents", "MacOS") + return os.path.join(mac, "chatd"), os.path.join(mac, "tinyclient"), os.path.join(mac, "mediad") + return os.path.join(app_dir, "chatd"), os.path.join(app_dir, "tinyclient"), os.path.join(app_dir, "mediad") + + +def popen(args, env, cwd, out_log, err_log, window=False): + kwargs = { + "cwd": cwd, + "env": env, + "stdin": subprocess.DEVNULL, + "stdout": open(out_log, "w", encoding="utf-8"), + "stderr": open(err_log, "w", encoding="utf-8"), + } + if sys.platform == "win32": + kwargs["creationflags"] = 0 if window else 0x08000000 + proc = subprocess.Popen(args, **kwargs) + STARTED.append(proc) + return proc + + +def stop_owned() -> None: + for proc in reversed(STARTED): + if proc.poll() is not None: + continue + try: + proc.terminate() + proc.wait(timeout=8) + except Exception: + try: + proc.kill() + proc.wait(timeout=8) + except Exception: + raise fixture.FixtureError("owned process did not stop; refusing cold-restart proof") + + +def wait_connected(profile: str, user_id: str, app_dir: str, seconds: float = 60.0) -> bool: + def ready(): + try: + answer = ipc.status(profile, app_dir) + except Exception: + return False + return session_contract.connected(answer, user_id) + + return fixture.wait_until(seconds, ready) + + +def wait_target(profile: str, app_dir: str, contract: dict, message_id: str, seconds=45): + key = {"network": "tiny", "account": contract["receiver"]["user_id"], + "id": contract["conversation_id"]} + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + answer = ipc.messages(profile, app_dir, "tiny", key["account"], key["id"]) + state = ipc.target_state(answer, key, message_id, contract["marker"]) + if state != "missing": + return state, answer + time.sleep(0.5) # Messages answers the cache while queuing asynchronous backfill. + raise fixture.FixtureError("exact delivered message never appeared; no history-loss RED") + + +def wait_delivered(profile, app_dir, contract, seconds=60): + key = {"network": "tiny", "account": contract["sender"]["user_id"], + "id": contract["conversation_id"]} + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + answer = ipc.messages(profile, app_dir, "tiny", key["account"], key["id"]) + message_id = ipc.delivered_id(answer, key, contract["marker"]) + if message_id: + return message_id + time.sleep(0.5) + raise fixture.FixtureError("peer send never acquired an encrypted server message id") + + +def incomplete(reason: str, **extra) -> int: + step("owned synthetic fixture is ready (%s)" % reason, False) + payload = {"error": reason, "valid_red": False} + payload.update(extra) + fixture.write_receipt(OUT, "setup-incomplete.json", payload) + return 1 + + +def classify_case(during_state, after_state, during_pixels, after_pixels): + if during_state == "opened" and after_state == "undecryptable" and during_pixels is True and after_pixels is False: + return "base-history-loss-red" + if during_state == "undecryptable" and after_state == "opened" and during_pixels is False and after_pixels is True: + return "branch-recoverable-green" + return "setup-incomplete" + + +def main() -> int: + global STEPS + if not OUT or not APP: + print("E2E_OUT and E2E_APP_DIR are required", file=sys.stderr) + return 1 + os.makedirs(OUT, exist_ok=True) + STEPS = os.path.join(OUT, "steps.jsonl") + open(STEPS, "w", encoding="utf-8").close() + + if os.environ.get("TINYCLIENT_PROVE"): + step("TINYCLIENT_PROVE must not be set for this normal-app driver", False) + return 1 + + chatd_bin, tiny_bin, mediad_bin = binaries(APP) + if not os.path.isfile(chatd_bin) or not os.path.isfile(tiny_bin): + step("release-profile chatd and tinyclient are in E2E_APP_DIR", False) + return 1 + if not os.path.isfile(mediad_bin): + step("mediad is in E2E_APP_DIR for a coherent bundle", False) + return 1 + + if SABOTAGE: + step("opened plaintext is still present after cache-fault repair and cold restart", False) + return 1 + + try: + contract = fixture.load_contract() + except fixture.FixtureError as err: + return incomplete(str(err)) + + if sys.platform == "win32": + import native_windows as native + require_native = native.require_windows + elif sys.platform.startswith("linux"): + import native_linux as native + require_native = native.require_linux + else: + return incomplete("no native pixel adapter for this platform", platform=sys.platform) + try: + require_native() + except native.NativeError as err: + return incomplete(str(err), platform=sys.platform) + receiver = contract["receiver"] + sender = contract["sender"] + recv_profile = str(fixture.receiver_profile(contract)) + send_profile = str(fixture.sender_profile(contract)) + marker = contract["marker"] + conv = contract["conversation_id"] + + shots = Path(OUT) / "shots" + shots.mkdir(parents=True, exist_ok=True) + + recv_env = session_contract.environment(os.environ, recv_profile, receiver["session"]) + send_env = session_contract.environment(os.environ, send_profile, sender["session"]) + + window = None + try: + cache_path = fixture.cache_log_path( + fixture.receiver_profile(contract), receiver["user_id"], conv + ) + if cache_path.exists() and not cache_path.is_dir(): + return incomplete("sealed-cache path already occupied by a file") + + blocker = fixture.place_append_blocker(contract) + step("empty-directory blocker occupies the absent sealed-cache log path", blocker.is_dir()) + fixture.write_receipt(OUT, "blocker.json", {"basename": blocker.name, "is_dir": True}) + + popen( + [chatd_bin], + send_env, + send_profile, + os.path.join(OUT, "sender-chatd.out.log"), + os.path.join(OUT, "sender-chatd.err.log"), + ) + popen( + [chatd_bin], + recv_env, + recv_profile, + os.path.join(OUT, "receiver-chatd.out.log"), + os.path.join(OUT, "receiver-chatd.err.log"), + ) + send_ok = wait_connected(send_profile, sender["user_id"], APP, 90) + recv_ok = wait_connected(recv_profile, receiver["user_id"], APP, 90) + step("owned sender and receiver chatd are connected", send_ok and recv_ok) + if not (send_ok and recv_ok): + return incomplete( + "chatd not connected", + sender_connected=send_ok, + receiver_connected=recv_ok, + ) + + def e2ee_ready(profile: str, user_id: str) -> bool: + try: + answer = ipc.e2ee_status(profile, APP, user_id) + except Exception: + return False + return ipc.e2ee_device_state(answer) == "ready" + + send_e2ee = fixture.wait_until(90, lambda: e2ee_ready(send_profile, sender["user_id"])) + recv_e2ee = fixture.wait_until(90, lambda: e2ee_ready(recv_profile, receiver["user_id"])) + step("owned sender and receiver e2ee devices are ready", send_e2ee and recv_e2ee) + if not (send_e2ee and recv_e2ee): + return incomplete( + "e2ee device not ready", + sender_e2ee=send_e2ee, + receiver_e2ee=recv_e2ee, + ) + + client_id = "cache-commit-%s" % uuid.uuid4().hex[:12] + try: + sent = ipc.send_text( + send_profile, APP, "tiny", sender["user_id"], conv, marker, client_id + ) + send_ok = sent.get("op") != "error" + except Exception as err: + send_ok = False + sent = {"error": str(err)} + step("owned peer delivered the unique encrypted marker", send_ok) + fixture.write_receipt(OUT, "peer-send.json", {"ok": send_ok, "client_id": client_id, "answer": sent}) + if not send_ok: + return incomplete("peer send failed", answer=sent) + + message_id = wait_delivered(send_profile, APP, contract) + fixture.write_receipt(OUT, "delivered-message.json", {"message_id": message_id}) + during_state, during_ipc = wait_target(recv_profile, APP, contract, message_id) + during_plain = during_state == "opened" + fixture.write_receipt(OUT, "during-fault-ipc.json", { + "plain": during_plain, + "answer": during_ipc, + }) + + window = popen( + [tiny_bin], + recv_env, + recv_profile, + os.path.join(OUT, "receiver-gui.out.log"), + os.path.join(OUT, "receiver-gui.err.log"), + window=True, + ) + shown = native.wait_window(window.pid, 60) + step("receiver window is visible under the fault", shown) + if not shown: + return incomplete("receiver window never appeared") + step("open the peer through Jump under the fault", True) + peer_name = sender["session"]["display_name"] + selected = native.select_sidebar_peer(window.pid, peer_name, str(shots / "before-select-during.png")) + fixture.write_receipt(OUT, "during-navigation.json", selected) + time.sleep(2.0) + during_shot = str(shots / "during-fault.png") + native.shot(window.pid, during_shot, printwindow=False) + during_pixels = native.conversation_mentions_marker(during_shot, marker) + step("during-fault window captured for pixel review", True, "shots/during-fault.png") + if during_pixels is None: + return incomplete("pixel text oracle unavailable; no native proof claimed") + + if window.poll() is None: + window.terminate() + try: + window.wait(timeout=15) + except Exception: + window.kill() + window = None + stop_owned() + STARTED.clear() + time.sleep(1.0) + step("owned apps and daemons stopped cold", True) + + fixture.clear_append_blocker(contract) + wiped = fixture.wipe_message_index(contract) + wipe_ok = wiped["after"] == 0 and wiped["conversation_rows"] >= 1 + step("blocker removed and only this fixture's SQLite message index emptied", wipe_ok) + fixture.write_receipt(OUT, "wipe.json", wiped) + if not wipe_ok: + return incomplete("sqlite wipe failed", wipe=wiped) + + popen( + [chatd_bin], + recv_env, + recv_profile, + os.path.join(OUT, "receiver-chatd-restart.out.log"), + os.path.join(OUT, "receiver-chatd-restart.err.log"), + ) + recv_ok = wait_connected(recv_profile, receiver["user_id"], APP, 90) + step("receiver chatd reconnected after repair", recv_ok) + if not recv_ok: + return incomplete("receiver did not reconnect after repair") + after_state, after_ipc = wait_target(recv_profile, APP, contract, message_id) + after_plain = after_state == "opened" + cache_file = fixture.cache_is_file(contract) + fixture.write_receipt(OUT, "after-restart-ipc.json", { + "plain": after_plain, + "cache_is_file": cache_file, + "during_plain": during_plain, + "answer": after_ipc, + }) + + window = popen( + [tiny_bin], + recv_env, + recv_profile, + os.path.join(OUT, "receiver-gui-restart.out.log"), + os.path.join(OUT, "receiver-gui-restart.err.log"), + window=True, + ) + shown = native.wait_window(window.pid, 60) + step("receiver window is visible after restart", shown) + if not shown: + return incomplete("receiver window never appeared after restart") + step("open the peer through Jump after restart", True) + selected = native.select_sidebar_peer(window.pid, peer_name, str(shots / "before-select-after.png")) + fixture.write_receipt(OUT, "after-navigation.json", selected) + time.sleep(2.0) + after_shot = str(shots / "after-restart.png") + native.shot(window.pid, after_shot, printwindow=False) + after_pixels = native.conversation_mentions_marker(after_shot, marker) + step("after-restart window captured for pixel review", True, "shots/after-restart.png") + if after_pixels is None: + return incomplete("pixel text oracle unavailable after restart") + + # Classify the matched case. Wrong login / never decrypted / generic loss is incomplete. + case = classify_case(during_state, after_state, during_pixels, after_pixels) + fixture.write_receipt(OUT, "pixel-observations.json", { + "during": during_pixels, "after": after_pixels, "class": case, + "during_state": during_state, "after_state": after_state, "message_id": message_id, + }) + if case == "base-history-loss-red": + step("message opened while the sealed-cache append was blocked", True) + step( + "opened plaintext is still present after cache-fault repair and cold restart", + False, + ) + fixture.write_receipt(OUT, "proof-local.json", { + "class": "base-history-loss-red", + "valid_red": True, + "during_plain": True, + "after_plain": False, + "cache_is_file": cache_file, + }) + elif case == "branch-recoverable-green": + if not cache_file: + return incomplete("opened row has no durable cache file after repair") + step("failed cache commit did not claim a durable open", True) + step( + "opened plaintext is still present after cache-fault repair and cold restart", + True, + ) + fixture.write_receipt(OUT, "proof-local.json", { + "class": "branch-recoverable-green", + "valid_red": False, + "during_plain": False, + "after_plain": True, + "cache_is_file": cache_file, + }) + elif during_plain and after_plain: + return incomplete( + "plaintext remained after wipe/restart; not the sealed-cache history-loss case", + during_plain=True, + after_plain=True, + cache_is_file=cache_file, + ) + else: + return incomplete( + "message states and window pixels do not demonstrate the matched history transition", + during_state=during_state, + after_state=after_state, + during_pixels=during_pixels, + after_pixels=after_pixels, + ) + + return 1 if FAILED else 0 + except fixture.FixtureError as err: + return incomplete(str(err)) + except Exception as err: + step("driver: %s" % err, False) + fixture.write_receipt(OUT, "driver-error.json", {"error": str(err), "valid_red": False}) + return 1 + finally: + if window and window.poll() is None: + try: + window.terminate() + window.wait(timeout=10) + except Exception: + pass + stop_owned() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/e2e/scenarios/opened-history-cache-commit/drive.windows.ps1 b/e2e/scenarios/opened-history-cache-commit/drive.windows.ps1 new file mode 100644 index 00000000..e17949b2 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/drive.windows.ps1 @@ -0,0 +1,18 @@ +# opened-history-cache-commit on Windows. Lab already films the console session. +# Fixture controller + native pixel driver (no TINYCLIENT_PROVE). +$ErrorActionPreference = 'Stop' +$drive = Join-Path $PSScriptRoot 'drive.py' +if (Get-Command py -ErrorAction SilentlyContinue) { + & py -3 $drive + exit $LASTEXITCODE +} +if (Get-Command python -ErrorAction SilentlyContinue) { + & python $drive + exit $LASTEXITCODE +} +if (Get-Command python3 -ErrorAction SilentlyContinue) { + & python3 $drive + exit $LASTEXITCODE +} +Write-Host 'python is not on PATH (tried py -3, python, python3)' +exit 1 diff --git a/e2e/scenarios/opened-history-cache-commit/fixture_controller.py b/e2e/scenarios/opened-history-cache-commit/fixture_controller.py new file mode 100644 index 00000000..670b2101 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/fixture_controller.py @@ -0,0 +1,200 @@ +"""Owned synthetic fixture controller for the sealed-cache append fault. + +This module prepares and mutates disposable receiver/sender profiles. It does not +drive the person-facing window. Diagnostic IPC is allowed here; native pixel +proof belongs in native_windows.py / drive.windows.ps1. + +Fault class (cache-write-order199): an empty directory occupies the otherwise +absent sealed-cache log path `e2ee/Tiny-/text/cache-.log` +while its parent remains writable. Only that append fails. + +Environment (all paths absolute; no host/provider names or secrets in-tree): + E2E_CACHE_COMMIT_FIXTURE JSON file describing the run (required for a matched case) +""" + +from __future__ import annotations + +import json +import base64 +import binascii +import os +import sqlite3 +import time +from pathlib import Path +from typing import Any, Mapping + + +MARKER_PREFIX = "opened-history-cache-commit-" + + +class FixtureError(RuntimeError): + """Setup failed for a reason that is not the history-loss assertion.""" + + +def load_contract(path: str | None = None) -> dict[str, Any]: + raw = path or os.environ.get("E2E_CACHE_COMMIT_FIXTURE", "").strip() + if not raw: + raise FixtureError( + "E2E_CACHE_COMMIT_FIXTURE must name an owned synthetic fixture JSON; " + "missing fixture is not a valid base RED" + ) + contract_path = Path(raw) + if not contract_path.is_file(): + raise FixtureError("fixture contract file is missing: %s" % contract_path) + data = json.loads(contract_path.read_text(encoding="utf-8")) + for key in ("marker", "conversation_id", "receiver", "sender"): + if key not in data: + raise FixtureError("fixture contract missing %s" % key) + if not str(data["marker"]).startswith(MARKER_PREFIX): + raise FixtureError("fixture marker must start with %s" % MARKER_PREFIX) + # Scratch sessions bypass the keychain. Never let each restart silently create + # new wrapping keys and discard the device state this scenario is testing. + for role in ("receiver", "sender"): + person = data[role] + session = person.get("session", {}) + keys = session.get("e2ee") if isinstance(session, Mapping) else None + if not isinstance(keys, Mapping) or session.get("user_id") != person.get("user_id"): + raise FixtureError("%s needs its own stable session.e2ee wrapping keys before first launch" % role) + name = session.get("display_name") + if not isinstance(name, str) or not name or name.split() != [name]: + raise FixtureError("%s needs a single-word synthetic display_name for sidebar pixel selection" % role) + for name in ("state", "cache"): + try: + encoded = keys.get(name) + if not isinstance(encoded, str) or len(base64.b64decode(encoded, validate=True)) != 32: + raise ValueError("invalid key") + except (ValueError, binascii.Error): + raise FixtureError("%s has missing/invalid durable wrapping keys" % role) from None + return data + + +def receiver_profile(contract: Mapping[str, Any]) -> Path: + profile = Path(contract["receiver"]["profile"]) + if not profile.is_absolute(): + raise FixtureError("receiver.profile must be absolute") + return profile + + +def sender_profile(contract: Mapping[str, Any]) -> Path: + profile = Path(contract["sender"]["profile"]) + if not profile.is_absolute(): + raise FixtureError("sender.profile must be absolute") + return profile + + +def e2ee_text_dir(profile: Path, user_id: str) -> Path: + safe = "".join( + ch if ch.isalnum() or ch == "-" else "_" for ch in ("Tiny-%s" % user_id) + )[:120] + return profile / "e2ee" / safe / "text" + + +def cache_log_path(profile: Path, user_id: str, conversation_id: str) -> Path: + return e2ee_text_dir(profile, user_id) / ("cache-%s.log" % conversation_id) + + +def place_append_blocker(contract: Mapping[str, Any]) -> Path: + """Create an empty directory at the sealed-cache log path. Parent must stay writable.""" + receiver = contract["receiver"] + profile = receiver_profile(contract) + text_dir = e2ee_text_dir(profile, receiver["user_id"]) + text_dir.mkdir(parents=True, exist_ok=True) + path = cache_log_path(profile, receiver["user_id"], contract["conversation_id"]) + if path.is_file(): + raise FixtureError("cache log already exists; refusing to overwrite prior sealed history") + if path.is_dir(): + return path + if path.exists(): + raise FixtureError("cache path exists with unexpected type: %s" % path) + path.mkdir(parents=False, exist_ok=False) + if not path.is_dir(): + raise FixtureError("failed to place empty-directory cache blocker") + # Parent must remain a normal directory (writable state store lives beside the cache). + if not text_dir.is_dir(): + raise FixtureError("e2ee text parent disappeared after blocker placement") + return path + + +def clear_append_blocker(contract: Mapping[str, Any]) -> None: + path = cache_log_path( + receiver_profile(contract), + contract["receiver"]["user_id"], + contract["conversation_id"], + ) + if path.is_dir(): + path.rmdir() + elif path.exists() and not path.is_file(): + raise FixtureError("cannot clear blocker at unexpected path type: %s" % path) + + +def wipe_message_index(contract: Mapping[str, Any]) -> dict[str, Any]: + """Delete only message rows for this fixture conversation; keep conversation/MLS/keys.""" + profile = receiver_profile(contract) + db_path = profile / "chat.sqlite" + if not db_path.is_file(): + raise FixtureError("receiver chat.sqlite missing") + conv = contract["conversation_id"] + scope = ("tiny", contract["receiver"]["user_id"], conv) + uri = "file:%s?mode=rw" % db_path.as_posix() + conn = sqlite3.connect(uri, uri=True, timeout=30) + conn.isolation_level = None + try: + conn.execute("PRAGMA busy_timeout=30000") + conn.execute("BEGIN IMMEDIATE") + conv_rows = conn.execute( + "SELECT COUNT(*) FROM conversation WHERE network = ? AND account = ? AND id = ?", + scope, + ).fetchone()[0] + if conv_rows != 1: + raise FixtureError("receiver conversation missing; refusing to change message index") + before = conn.execute( + "SELECT COUNT(*) FROM message WHERE network = ? AND account = ? AND conversation = ?", scope + ).fetchone()[0] + deleted = conn.execute( + "DELETE FROM message WHERE network = ? AND account = ? AND conversation = ?", scope + ).rowcount + conn.execute("COMMIT") + after = conn.execute( + "SELECT COUNT(*) FROM message WHERE network = ? AND account = ? AND conversation = ?", scope + ).fetchone()[0] + finally: + conn.close() + if conv_rows < 1: + raise FixtureError("conversation row missing after message wipe; wrong fixture") + if after != 0: + raise FixtureError("message index wipe left rows for the target conversation") + return {"before": before, "deleted": deleted, "after": after, "conversation_rows": conv_rows} + + +def cache_is_file(contract: Mapping[str, Any]) -> bool: + path = cache_log_path( + receiver_profile(contract), + contract["receiver"]["user_id"], + contract["conversation_id"], + ) + return path.is_file() + + +def cache_blocker_present(contract: Mapping[str, Any]) -> bool: + path = cache_log_path( + receiver_profile(contract), + contract["receiver"]["user_id"], + contract["conversation_id"], + ) + return path.is_dir() + + +def write_receipt(out_dir: str, name: str, payload: Mapping[str, Any]) -> Path: + path = Path(out_dir) / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return path + + +def wait_until(seconds: float, predicate, sleep: float = 0.2) -> bool: + deadline = time.time() + seconds + while time.time() < deadline: + if predicate(): + return True + time.sleep(sleep) + return False diff --git a/e2e/scenarios/opened-history-cache-commit/ipc.py b/e2e/scenarios/opened-history-cache-commit/ipc.py new file mode 100644 index 00000000..fea99294 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/ipc.py @@ -0,0 +1,196 @@ +"""Minimal chatd IPC helper for fixture setup and diagnostic guards only.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import socket +import struct +import subprocess +import sys +import uuid +from typing import Any, Mapping + + +def dlctl_bin(app_dir: str) -> str: + override = os.environ.get("E2E_DLCTL", "").strip() + if override and os.path.isfile(override): + return override + name = "dlctl.exe" if sys.platform == "win32" else "dlctl" + path = os.path.join(app_dir, name) + if os.path.isfile(path): + return path + return "" + + +def _unix_request(profile, payload, timeout): + directory = Path(profile) + if not directory.is_absolute(): + raise RuntimeError("fixture profile must be absolute") + body = json.dumps(payload, separators=(",", ":")).encode() + if len(body) > 1024 * 1024: # protocol::FRAME_LIMIT + raise RuntimeError("fixture IPC request exceeds the protocol size limit") + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as stream: + stream.settimeout(timeout) + stream.connect(str(directory / "chat.sock")) + stream.sendall(struct.pack("!I", len(body)) + body) + + def read_exact(length): + chunks = bytearray() + while len(chunks) < length: + chunk = stream.recv(length - len(chunks)) + if not chunk: + raise RuntimeError("fixture IPC closed before the frame completed") + chunks.extend(chunk) + return bytes(chunks) + + length = struct.unpack("!I", read_exact(4))[0] + if length > 1024 * 1024: + raise RuntimeError("fixture IPC answer exceeds the diagnostic size limit") + answer = json.loads(read_exact(length)) + if not isinstance(answer, dict): + raise RuntimeError("fixture IPC answer is not an object") + return answer + + +def request(profile: str, payload: Mapping[str, Any], app_dir: str = "", timeout: int = 60) -> dict[str, Any]: + if sys.platform.startswith("linux") and not os.environ.get("E2E_DLCTL", "").strip(): + return _unix_request(profile, payload, timeout) + binary = dlctl_bin(app_dir) if app_dir else dlctl_bin("") + if not binary: + raise RuntimeError( + "dlctl is required for fixture IPC (place dlctl beside the app or set E2E_DLCTL)" + ) + body = json.dumps(payload, separators=(",", ":")) + done = subprocess.run( + [binary, profile, body], + capture_output=True, + text=True, + timeout=timeout, + ) + if done.returncode not in (0, 4): + raise RuntimeError( + "dlctl failed (%s): %s" % (done.returncode, (done.stderr or done.stdout)[-500:]) + ) + return json.loads(done.stdout or "{}") + + +def status(profile: str, app_dir: str) -> dict[str, Any]: + return request(profile, {"op": "status"}, app_dir=app_dir) + + +def e2ee_status(profile: str, app_dir: str, user_id: str) -> dict[str, Any]: + return request( + profile, + { + "op": "tiny_e2ee", + "account": {"network": "tiny", "account": user_id}, + "ask": {"op": "status"}, + }, + app_dir=app_dir, + timeout=45, + ) + + +def e2ee_device_state(answer: Mapping[str, Any]) -> str: + status = answer.get("status") if isinstance(answer.get("status"), Mapping) else answer + if not isinstance(status, Mapping): + return "" + device = status.get("device") + if isinstance(device, Mapping): + return str(device.get("state") or "") + return str(device or "") + + +def send_text( + profile: str, + app_dir: str, + network: str, + account: str, + conversation: str, + text: str, + client_id: str, +) -> dict[str, Any]: + return request( + profile, + { + "op": "send", + "conversation": {"network": network, "account": account, "id": conversation}, + "client_id": client_id, + "text": text, + }, + app_dir=app_dir, + ) + + +def messages( + profile: str, + app_dir: str, + network: str, + account: str, + conversation: str, + limit: int = 50, +) -> dict[str, Any]: + return request( + profile, + { + "op": "messages", + "conversation": {"network": network, "account": account, "id": conversation}, + "limit": int(limit), + }, + app_dir=app_dir, + ) + + +def row_send_state(row: Mapping[str, Any]) -> str: + if "state" not in row: + return "sent" # serde omits the default Sent variant. + state = row["state"] + if isinstance(state, Mapping): + value = state.get("state") + return value if value in ("sent", "pending", "failed") else "invalid" + return "invalid" + + +def text_contains_marker(answer: Mapping[str, Any], marker: str) -> bool: + if answer.get("op") != "messages": + return False + return any( + isinstance(row, Mapping) + and row.get("text") == marker + and row.get("seal") == "encrypted" + and row_send_state(row) == "sent" + for row in answer.get("messages", []) + ) + + +def target_state(answer, conversation, message_id, marker): + if answer.get("op") != "messages" or answer.get("conversation") != conversation: + raise ValueError("history reply is not a messages page for the expected account/conversation") + rows = answer.get("messages") + if not isinstance(rows, list): + raise ValueError("history reply has no messages array") + target = next((row for row in rows if isinstance(row, Mapping) and row.get("id") == message_id), None) + if target is None: + return "missing" + if text_contains_marker({"op": "messages", "messages": [target]}, marker): + return "opened" + if target.get("seal") == "undecryptable" and row_send_state(target) == "sent": + return "undecryptable" + raise ValueError("target row is neither the exact opened message nor an undecryptable row") + + +def delivered_id(answer, conversation, marker): + if answer.get("op") != "messages" or answer.get("conversation") != conversation: + raise ValueError("sender history reply is not scoped to the expected conversation") + for row in answer.get("messages", []): + if not isinstance(row, Mapping): + continue + if not text_contains_marker({"op": "messages", "messages": [row]}, marker): + continue + try: + return str(uuid.UUID(row.get("id", ""))) + except (ValueError, TypeError, AttributeError): + continue + return None diff --git a/e2e/scenarios/opened-history-cache-commit/native_linux.py b/e2e/scenarios/opened-history-cache-commit/native_linux.py new file mode 100644 index 00000000..42a33a57 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/native_linux.py @@ -0,0 +1,165 @@ +"""Owned X11 input and image-only observations for the Linux history driver.""" + +import csv +import io +import os +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +import time + + +class NativeError(RuntimeError): + pass + + +def _run(args, timeout=15): + result = subprocess.run(args, capture_output=True, text=True, timeout=timeout) + if result.returncode: + raise NativeError("native command failed: " + result.stderr[-400:]) + return result.stdout.strip() + + +def require_linux(): + if not sys.platform.startswith("linux") or not os.environ.get("DISPLAY"): + raise NativeError("Linux X11 DISPLAY is required") + if os.environ.get("XDG_SESSION_TYPE") == "wayland": + raise NativeError("this adapter requires an X11 session") + for tool in ("wmctrl", "xdotool", "ffmpeg", "tesseract"): + if not shutil.which(tool): + raise NativeError("native test dependency is missing: " + tool) + + +def _owned(pid): + app = os.environ.get("E2E_APP_DIR", "") + if not app or pid <= 0: + raise NativeError("an owned app directory and PID are required") + try: + expected = (Path(app) / "tinyclient").resolve(strict=True) + actual = (Path("/proc") / str(pid) / "exe").resolve(strict=True) + except OSError as error: + raise NativeError("owned GUI process is unavailable") from error + if actual != expected: + raise NativeError("GUI PID executable differs from the owned app") + + +def rect(pid): + _owned(pid) + candidates = [] + for line in _run(["wmctrl", "-lpG"]).splitlines(): + fields = line.split(None, 8) + if len(fields) < 8 or not fields[2].isdigit() or int(fields[2]) != pid: + continue + try: + x, y, width, height = map(int, fields[3:7]) + if width > 0 and height > 0: + candidates.append({"window_id": int(fields[0], 16), "left": x, "top": y, + "width": width, "height": height}) + except ValueError: + continue + if not candidates: + raise NativeError("owned GUI window is unavailable") + return max(candidates, key=lambda item: item["width"] * item["height"]) + + +def wait_window(pid, seconds=45): + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + try: + rect(pid) + return True + except NativeError: + time.sleep(.25) + return False + + +def _front(pid): + window = rect(pid)["window_id"] + _run(["wmctrl", "-ia", hex(window)]) + time.sleep(.1) + _focused(pid, window) + return window + + +def _focused(pid, window): + _owned(pid) + active = int(_run(["xdotool", "getactivewindow"])) + owner = int(_run(["xdotool", "getwindowpid", str(active)])) + if active != window or owner != pid: + raise NativeError("foreground window is not the owned GUI; input refused") + + +def _key(pid, window, key): + _focused(pid, window) + _run(["xdotool", "key", "--clearmodifiers", key]) + _focused(pid, window) + + +def shot(pid, path, printwindow=False): + window = _front(pid) + _run(["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "x11grab", + "-window_id", str(window), "-i", os.environ["DISPLAY"], "-frames:v", "1", str(path)]) + _owned(pid) + + +def ocr_words(path): + # Small native glyphs can lose a hyphen at 1x. Enlarge the actual captured + # image rather than relaxing the exact marker match. + with tempfile.TemporaryDirectory(prefix="tiny-history-ocr-") as folder: + enlarged = Path(folder) / "pixels.png" + _run(["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", + str(Path(path).resolve()), "-vf", "scale=iw*2:ih*2:flags=lanczos", str(enlarged)]) + output = _run(["tesseract", str(enlarged), "stdout", "--psm", "11", "tsv"], timeout=30) + words = [] + for row in csv.DictReader(io.StringIO(output), delimiter="\t", quoting=csv.QUOTE_NONE): + if row.get("level") != "5" or not row.get("text", "").strip(): + continue + try: + words.append({"text": row["text"], "x": int(row["left"]) // 2, "y": int(row["top"]) // 2, + "width": int(row["width"]) // 2, "height": int(row["height"]) // 2}) + except (ValueError, KeyError): + raise NativeError("OCR word geometry is malformed") + return words + + +def conversation_marker(words, marker): + anchors = [] + for word in words: + if word["text"].casefold() != "this" or word["x"] < 300 or word["y"] < 50: + continue + nearby = [item for item in words if item["x"] >= word["x"] + and word["y"] - 2 <= item["y"] <= word["y"] + 2 * word["height"] + 4] + phrase = "".join(item["text"] for item in nearby).casefold() + if phrase.startswith("thisisthestartofyourconversation"): + anchors.append((word, max(item["y"] + item["height"] for item in nearby))) + if not anchors: + return None + anchor, bottom = min(anchors, key=lambda item: (item[0]["x"], item[0]["y"])) + text = "".join(word["text"] for word in words + if word["x"] >= anchor["x"] and word["y"] > bottom) + return marker.casefold() in "".join(text.split()).casefold() + + +def conversation_mentions_marker(path, marker): + try: + return conversation_marker(ocr_words(path), marker) + except (NativeError, OSError, ValueError, subprocess.TimeoutExpired): + return None + + +def select_sidebar_peer(pid, name, screenshot): + window = _front(pid) + _key(pid, window, "ctrl+k") + time.sleep(.4) + _focused(pid, window) + _run(["xdotool", "type", "--clearmodifiers", "--delay", "40", "--", name]) + _focused(pid, window) + time.sleep(.5) + shot(pid, screenshot) + text = " ".join(word["text"] for word in ocr_words(screenshot)) + if name.casefold() not in text.casefold(): + raise NativeError("peer query is not visible; Enter refused") + _key(pid, window, "Return") + return {"method": "ctrl-k query enter", "name": name, "owned_window": window} diff --git a/e2e/scenarios/opened-history-cache-commit/native_windows.py b/e2e/scenarios/opened-history-cache-commit/native_windows.py new file mode 100644 index 00000000..debae5b8 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/native_windows.py @@ -0,0 +1,146 @@ +"""Person-facing Windows driver actions for opened-history-cache-commit. + +Uses owned_input.ps1 with foreground/hit-PID guards. Does not call chatd prove +modes and does not treat IPC Send as native evidence. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from pathlib import Path + + +HERE = Path(__file__).resolve().parent +OWNED_INPUT = HERE / "owned_input.ps1" + + +class NativeError(RuntimeError): + pass + + +def _ps(args: list[str], timeout: int = 60) -> str: + cmd = [ + "powershell.exe", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + str(OWNED_INPUT), + *args, + ] + done = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + if done.returncode != 0: + raise NativeError( + (done.stdout or "")[-400:] + "\n" + (done.stderr or "")[-400:] + ) + return (done.stdout or "").strip() + + +def prepare(pid: int, other_pid: int = 0) -> None: + args = ["-ExpectedPid", str(pid), "-Action", "prepare"] + if other_pid: + args += ["-OtherPid", str(other_pid)] + _ps(args) + + +def front(pid: int) -> None: + _ps(["-ExpectedPid", str(pid), "-Action", "front"]) + + +def shot(pid: int, path: str, printwindow: bool = False) -> None: + action = "printwindow" if printwindow else "capture" + _ps(["-ExpectedPid", str(pid), "-Action", action, "-Screenshot", path]) + + +def click(pid: int, x: int, y: int) -> None: + _ps(["-ExpectedPid", str(pid), "-Action", "click", "-X", str(x), "-Y", str(y)]) + + +def windows_ocr(path): + done = subprocess.run([ + "powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass", + "-File", str(HERE / "ocr.windows.ps1"), "-Path", str(Path(path).resolve()), + ], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=20) + if done.returncode != 0: + raise NativeError("Windows pixel text observation unavailable") + return json.loads(done.stdout) + + +def select_sidebar_peer(pid, name, screenshot): + """Use the app's normal Jump command, with query pixels checked before Enter.""" + prepare(pid) + _ps(["-ExpectedPid", str(pid), "-Action", "type", "-Text", "^k"]) + time.sleep(0.4) + # SendKeys treats these characters as commands unless escaped. + literal = "".join("{{}" if ch == "{" else "{}}" if ch == "}" else + "{" + ch + "}" if ch in "+^%~()[]" else ch for ch in name) + _ps(["-ExpectedPid", str(pid), "-Action", "type", "-Text", literal]) + time.sleep(0.7) + shot(pid, screenshot) + if name.casefold() not in windows_ocr(screenshot)["text"].casefold(): + raise NativeError("peer query is not visible in Jump; refused Enter") + _ps(["-ExpectedPid", str(pid), "-Action", "type", "-Text", "{ENTER}"]) + return {"method": "ctrl-k query enter", "name": name} + + +def rect(pid: int) -> dict: + raw = _ps(["-ExpectedPid", str(pid), "-Action", "rect"]) + return json.loads(raw) + + +def wait_window(pid: int, seconds: float = 45.0) -> bool: + deadline = time.time() + seconds + while time.time() < deadline: + try: + rect(pid) + return True + except NativeError: + time.sleep(0.4) + return False + + +def png_mentions_marker(path: str, marker: str) -> bool | None: + """Best-effort pixel/text gate. + + Uses Tesseract when installed, otherwise the Windows OCR API. Missing/broken OCR returns + None, which must make the run incomplete rather than certifying IPC as pixels. + """ + try: + from PIL import Image # type: ignore + import pytesseract # type: ignore + image = Image.open(path) + text = pytesseract.image_to_string(image) + except Exception: + try: + if sys.platform != "win32": + return None + text = windows_ocr(path)["text"] + except Exception: + return None + return marker in "".join(text.split()) + + +def conversation_mentions_marker(path, marker): + try: + words = windows_ocr(path)["words"] + anchors = [word for word in words if word.get("text", "").casefold() == "this" + and word.get("x", 0) >= 300 and word.get("y", 0) >= 50] + if not anchors: + return None # No fresh-conversation intro; a sidebar preview cannot certify the pane. + anchor = min(anchors, key=lambda word: word["x"]) + text = "".join(word["text"] for word in words + if word["x"] >= anchor["x"] and word["y"] > anchor["y"] + anchor["height"]) + return marker in "".join(text.split()) + except Exception: + return None + + +def require_windows() -> None: + if sys.platform != "win32": + raise NativeError("native_windows driver is Windows-only") + if not OWNED_INPUT.is_file(): + raise NativeError("owned_input.ps1 missing beside native_windows.py") diff --git a/e2e/scenarios/opened-history-cache-commit/ocr.windows.ps1 b/e2e/scenarios/opened-history-cache-commit/ocr.windows.ps1 new file mode 100644 index 00000000..0b23e5d3 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/ocr.windows.ps1 @@ -0,0 +1,36 @@ +param([Parameter(Mandatory)][string]$Path) +$ErrorActionPreference='Stop' +$ProgressPreference='SilentlyContinue' +Add-Type -AssemblyName System.Runtime.WindowsRuntime +[void][Windows.Storage.StorageFile,Windows.Storage,ContentType=WindowsRuntime] +[void][Windows.Storage.FileAccessMode,Windows.Storage,ContentType=WindowsRuntime] +[void][Windows.Storage.Streams.IRandomAccessStream,Windows.Storage.Streams,ContentType=WindowsRuntime] +[void][Windows.Graphics.Imaging.BitmapDecoder,Windows.Graphics.Imaging,ContentType=WindowsRuntime] +[void][Windows.Graphics.Imaging.SoftwareBitmap,Windows.Graphics.Imaging,ContentType=WindowsRuntime] +[void][Windows.Media.Ocr.OcrEngine,Windows.Foundation,ContentType=WindowsRuntime] +[void][Windows.Media.Ocr.OcrResult,Windows.Foundation,ContentType=WindowsRuntime] +$taskMethod=[System.WindowsRuntimeSystemExtensions].GetMethods() | Where-Object { + $_.Name -eq 'AsTask' -and $_.IsGenericMethodDefinition -and $_.GetParameters().Count -eq 1 -and $_.GetParameters()[0].ParameterType.Name -eq 'IAsyncOperation`1' +} | Select-Object -First 1 +function Await($Operation,[Type]$Type) { + $task=$taskMethod.MakeGenericMethod($Type).Invoke($null,@($Operation)) + $task.GetAwaiter().GetResult() +} +$file=Await ([Windows.Storage.StorageFile]::GetFileFromPathAsync((Resolve-Path -LiteralPath $Path).Path)) ([Windows.Storage.StorageFile]) +$stream=Await ($file.OpenAsync([Windows.Storage.FileAccessMode]::Read)) ([Windows.Storage.Streams.IRandomAccessStream]) +$bitmap=$null +try { + $decoder=Await ([Windows.Graphics.Imaging.BitmapDecoder]::CreateAsync($stream)) ([Windows.Graphics.Imaging.BitmapDecoder]) + $bitmap=Await ($decoder.GetSoftwareBitmapAsync()) ([Windows.Graphics.Imaging.SoftwareBitmap]) + $engine=[Windows.Media.Ocr.OcrEngine]::TryCreateFromUserProfileLanguages() + if($null -eq $engine) { throw 'No Windows OCR language available' } + $result=Await ($engine.RecognizeAsync($bitmap)) ([Windows.Media.Ocr.OcrResult]) + $words=@(foreach($line in $result.Lines) { foreach($word in $line.Words) { + $r=$word.BoundingRect + [pscustomobject]@{text=$word.Text;x=$r.X;y=$r.Y;width=$r.Width;height=$r.Height} + } }) + [pscustomobject]@{text=$result.Text;words=$words} | ConvertTo-Json -Depth 5 -Compress +} finally { + if($bitmap) { $bitmap.Dispose() } + $stream.Dispose() +} diff --git a/e2e/scenarios/opened-history-cache-commit/owned_input.ps1 b/e2e/scenarios/opened-history-cache-commit/owned_input.ps1 new file mode 100644 index 00000000..4b219fec --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/owned_input.ps1 @@ -0,0 +1,132 @@ +<# +.SYNOPSIS + Identity-checked Windows input/capture for this scenario only. + Foreground and hit-PID must match ExpectedPid before click/type/capture. + Adapted from the proven composer-input178 / native-history202 helpers. +#> +param( + [Parameter(Mandatory = $true)][int]$ExpectedPid, + [Parameter(Mandatory = $true)][ValidateSet( + 'prepare', 'front', 'capture', 'printwindow', 'click', 'type', 'rect' + )][string]$Action, + [int]$OtherPid = 0, + [int]$ScreenX = 0, + [int]$ScreenY = 0, + [int]$X = 0, + [int]$Y = 0, + [string]$Screenshot, + [string]$Text +) +$ErrorActionPreference = 'Stop' +Add-Type -AssemblyName System.Windows.Forms, System.Drawing +if (-not ('TinyOwnedInput204' -as [type])) { + Add-Type @' +using System; +using System.Runtime.InteropServices; +public static class TinyOwnedInput204 { + public delegate bool EnumProc(IntPtr h, IntPtr p); + [StructLayout(LayoutKind.Sequential)] public struct RECT { public int L,T,R,B; } + [StructLayout(LayoutKind.Sequential)] public struct POINT { public int X,Y; } + [DllImport("user32.dll")] public static extern bool SetProcessDPIAware(); + [DllImport("user32.dll")] public static extern bool EnumWindows(EnumProc f, IntPtr p); + [DllImport("user32.dll")] public static extern bool IsWindowVisible(IntPtr h); + [DllImport("user32.dll")] public static extern uint GetWindowThreadProcessId(IntPtr h,out uint p); + [DllImport("user32.dll")] public static extern bool GetWindowRect(IntPtr h,out RECT r); + [DllImport("user32.dll")] public static extern IntPtr GetForegroundWindow(); + [DllImport("user32.dll")] public static extern IntPtr WindowFromPoint(POINT p); + [DllImport("user32.dll")] public static extern bool ShowWindow(IntPtr h,int n); + [DllImport("user32.dll")] public static extern bool SetForegroundWindow(IntPtr h); + [DllImport("user32.dll")] public static extern bool SetCursorPos(int x,int y); + [DllImport("user32.dll")] public static extern void mouse_event(uint f,uint x,uint y,uint d,UIntPtr e); + [DllImport("user32.dll")] public static extern bool PrintWindow(IntPtr h, IntPtr hdcBlt, uint nFlags); + [DllImport("user32.dll")] public static extern bool ClientToScreen(IntPtr h,ref POINT p); + public static uint Pid(IntPtr h) { uint p; GetWindowThreadProcessId(h,out p); return p; } + public static IntPtr Find(int p) { + IntPtr result=IntPtr.Zero; long area=0; + EnumWindows((h,l)=>{ if(Pid(h)==p && IsWindowVisible(h)) { RECT r; GetWindowRect(h,out r); long a=(long)(r.R-r.L)*(r.B-r.T); if(a>area){area=a;result=h;} } return true;},IntPtr.Zero); + return result; + } +} +'@ +} +[TinyOwnedInput204]::SetProcessDPIAware() | Out-Null +$target = [TinyOwnedInput204]::Find($ExpectedPid) +if ($target -eq [IntPtr]::Zero) { throw "No visible owned window for PID $ExpectedPid" } + +if ($Action -eq 'prepare') { + if ($OtherPid -eq $ExpectedPid) { throw 'OtherPid must name a different owned process' } + if ($OtherPid -gt 0) { + $other = [TinyOwnedInput204]::Find($OtherPid) + if ($other -ne [IntPtr]::Zero) { [TinyOwnedInput204]::ShowWindow($other, 6) | Out-Null } + } + [TinyOwnedInput204]::ShowWindow($target, 3) | Out-Null + [TinyOwnedInput204]::SetForegroundWindow($target) | Out-Null + Start-Sleep -Milliseconds 200 +} + +if ($Action -in @('prepare', 'front', 'click', 'type', 'capture', 'printwindow')) { + [TinyOwnedInput204]::ShowWindow($target, 9) | Out-Null + [TinyOwnedInput204]::SetForegroundWindow($target) | Out-Null + Start-Sleep -Milliseconds 150 + $foregroundPid = [TinyOwnedInput204]::Pid([TinyOwnedInput204]::GetForegroundWindow()) + if ($foregroundPid -ne $ExpectedPid) { + throw "Foreground PID $foregroundPid differs from expected $ExpectedPid; no input sent" + } +} + +$r = New-Object TinyOwnedInput204+RECT +[TinyOwnedInput204]::GetWindowRect($target, [ref]$r) | Out-Null + +switch ($Action) { + 'rect' { + [pscustomobject]@{ + hwnd = [int64]$target; left = $r.L; top = $r.T + width = ($r.R - $r.L); height = ($r.B - $r.T) + } | ConvertTo-Json -Compress + } + 'prepare' { Write-Host "prepared pid=$ExpectedPid" } + 'front' { Write-Host "front pid=$ExpectedPid" } + 'capture' { + if (-not $Screenshot) { throw 'Screenshot path required' } + $bounds = [Drawing.Rectangle]::FromLTRB($r.L, $r.T, $r.R, $r.B) + $bmp = New-Object Drawing.Bitmap $bounds.Width, $bounds.Height + $g = [Drawing.Graphics]::FromImage($bmp) + $g.CopyFromScreen($bounds.Location, [Drawing.Point]::Empty, $bounds.Size) + $g.Dispose() + $bmp.Save($Screenshot, [Drawing.Imaging.ImageFormat]::Png) + $bmp.Dispose() + Write-Host "saved $Screenshot" + } + 'printwindow' { + if (-not $Screenshot) { throw 'Screenshot path required' } + $bmp = New-Object Drawing.Bitmap ($r.R - $r.L), ($r.B - $r.T) + $g = [Drawing.Graphics]::FromImage($bmp) + $dc = $g.GetHdc() + [TinyOwnedInput204]::PrintWindow($target, $dc, 2) | Out-Null + $g.ReleaseHdc($dc); $g.Dispose() + $bmp.Save($Screenshot, [Drawing.Imaging.ImageFormat]::Png) + $bmp.Dispose() + Write-Host "saved $Screenshot" + } + 'click' { + $screenX = if ($ScreenX) { $ScreenX } else { $r.L + $X } + $screenY = if ($ScreenY) { $ScreenY } else { $r.T + $Y } + $pt = New-Object TinyOwnedInput204+POINT + $pt.X = $screenX; $pt.Y = $screenY + $hit = [TinyOwnedInput204]::WindowFromPoint($pt) + $hitPid = if ($hit -ne [IntPtr]::Zero) { [TinyOwnedInput204]::Pid($hit) } else { 0 } + if ($hitPid -ne $ExpectedPid) { + throw "Hit PID $hitPid at $screenX,$screenY differs from expected $ExpectedPid" + } + [TinyOwnedInput204]::SetCursorPos($screenX, $screenY) | Out-Null + [TinyOwnedInput204]::mouse_event(0x0002, 0, 0, 0, [UIntPtr]::Zero) # LEFTDOWN + Start-Sleep -Milliseconds 30 + [TinyOwnedInput204]::mouse_event(0x0004, 0, 0, 0, [UIntPtr]::Zero) # LEFTUP + Write-Host "clicked $screenX,$screenY hitPid=$hitPid" + } + 'type' { + if (-not $Text) { throw 'Text required' } + [Windows.Forms.SendKeys]::SendWait($Text) + Write-Host ("typed length={0}" -f $Text.Length) + } +} diff --git a/e2e/scenarios/opened-history-cache-commit/scenario.json b/e2e/scenarios/opened-history-cache-commit/scenario.json new file mode 100644 index 00000000..03fec42d --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/scenario.json @@ -0,0 +1,16 @@ +{ + "id": "opened-history-cache-commit", + "title": "A failed sealed-cache append cannot leave only a consumed receive ratchet", + "platforms": [ + "windows", + "linux" + ], + "not_applicable": { + "macos": "Person-facing matched-case proof is Windows-native window input/pixels; macOS is not claimed here.", + "android": "Desktop sealed-cache fault is observed on the Windows app window; mobile runtime proof is not claimed.", + "ios": "Desktop sealed-cache fault is observed on the Windows app window; mobile runtime proof is not claimed." + }, + "needs": [], + "release": false, + "max_still_seconds": 90 +} diff --git a/e2e/scenarios/opened-history-cache-commit/session_contract.py b/e2e/scenarios/opened-history-cache-commit/session_contract.py new file mode 100644 index 00000000..66eeb01f --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/session_contract.py @@ -0,0 +1,45 @@ +"""Session environment helper for scratch Tiny runs. + +Keeps a literal space in TINYCLIENT_SESSION_TINY SERVER and strips other +TINYCLIENT_SESSION_* keys so a lab host's signed-in profile cannot leak in. +""" + +from __future__ import annotations + +import json +from typing import Mapping, MutableMapping + +SESSION_VARIABLE = "TINYCLIENT_SESSION_TINY SERVER" + + +def environment(parent: Mapping[str, str], profile: str, session: Mapping[str, object]) -> dict[str, str]: + result = { + key: value + for key, value in parent.items() + if not key.upper().startswith("TINYCLIENT_SESSION_") + } + result["TINYCLIENT_DIR"] = str(profile) + result["TINYCLIENT_LOG_LEVEL"] = parent.get("TINYCLIENT_LOG_LEVEL", "info") + result[SESSION_VARIABLE] = json.dumps(session, separators=(",", ":")) + return result + + +def connected(status: Mapping[str, object], user_id: str) -> bool: + if status.get("op") != "status": + return False + accounts = status.get("accounts") or [] + if not isinstance(accounts, list): + return False + for row in accounts: + if not isinstance(row, Mapping): + continue + key = row.get("key") + if key == {"network": "tiny", "account": user_id} and row.get("link") == "connected": + return True + return False + + +def apply(env: MutableMapping[str, str], profile: str, session: Mapping[str, object]) -> None: + built = environment(env, profile, session) + env.clear() + env.update(built) diff --git a/e2e/scenarios/opened-history-cache-commit/test_contract.py b/e2e/scenarios/opened-history-cache-commit/test_contract.py new file mode 100644 index 00000000..c34854fa --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/test_contract.py @@ -0,0 +1,112 @@ +"""Regression checks for fixture identity and authenticated-history diagnostics.""" + +import json +import os +import subprocess +import sys +import unittest + +import ipc +import session_contract + + +class ContractTests(unittest.TestCase): + def test_scratch_child_receives_tiny_server_session_without_other_accounts(self): + session = {"user_id": "receiver", "token": "synthetic-only"} + env = session_contract.environment( + {**os.environ, "TINYCLIENT_SESSION_DISCORD": "unrelated", + "TINYCLIENT_SESSION_TINYSERVER": "wrong"}, + "scratch-receiver", session, + ) + child = subprocess.run( + [sys.executable, "-c", + "import os,json; print(json.dumps({k:v for k,v in os.environ.items() " + "if k.startswith('TINYCLIENT_SESSION_')}))"], + env=env, text=True, capture_output=True, check=True, + ) + self.assertEqual(json.loads(child.stdout), { + "TINYCLIENT_SESSION_TINY SERVER": json.dumps(session, separators=(",", ":")), + }) + + def test_marker_must_be_opened_message_text(self): + marker = "opened-history-cache-commit-only-target" + for answer in ( + {"op": "error", "message": marker}, + {"op": "messages", "messages": [{"text": "other", "seal": "encrypted", + "reply_to": {"text": marker}}]}, + {"op": "messages", "messages": [{"text": marker, "seal": "undecryptable"}]}, + {"op": "messages", "messages": [{"text": marker, "seal": "encrypted", "state": "sending"}]}, + {"op": "messages", "messages": [{"text": marker + "-suffix", "seal": "encrypted"}]}, + ): + with self.subTest(answer=answer): + self.assertFalse(ipc.text_contains_marker(answer, marker)) + self.assertTrue(ipc.text_contains_marker({ + "op": "messages", "messages": [ + {"id": "message-one", "text": marker, "seal": "encrypted"}, + ], + }, marker)) + + def test_missing_pixel_observation_cannot_certify_history_recovery(self): + import drive + self.assertEqual(drive.classify_case("opened", "undecryptable", False, False), "setup-incomplete") + self.assertEqual(drive.classify_case("undecryptable", "opened", False, False), "setup-incomplete") + self.assertEqual(drive.classify_case("opened", "undecryptable", True, False), "base-history-loss-red") + self.assertEqual(drive.classify_case("undecryptable", "opened", False, True), "branch-recoverable-green") + self.assertEqual(drive.classify_case("missing", "missing", False, False), "setup-incomplete") + self.assertEqual(drive.classify_case("opened", "missing", True, False), "setup-incomplete") + self.assertEqual(drive.classify_case("opened", "undecryptable", True, None), "setup-incomplete") + self.assertEqual(drive.classify_case("undecryptable", "opened", None, True), "setup-incomplete") + + def test_target_observation_waits_for_exact_scoped_row_not_empty_backfill(self): + key = {"network": "tiny", "account": "receiver", "id": "conversation"} + marker = "opened-history-cache-commit-only-target" + def answer(rows, scope=key): + return {"op": "messages", "conversation": scope, "messages": rows} + for payload in ( + {"op": "error", "message": "offline"}, + answer([], {**key, "account": "other"}), + ): + with self.subTest(payload=payload), self.assertRaises(ValueError): + ipc.target_state(payload, key, "target-id", marker) + self.assertEqual(ipc.target_state(answer([]), key, "target-id", marker), "missing") + self.assertEqual(ipc.target_state(answer([ + {"id": "other-id", "text": marker, "seal": "encrypted"}, + ]), key, "target-id", marker), "missing") + self.assertEqual(ipc.target_state(answer([ + {"id": "target-id", "seal": "undecryptable"}, + ]), key, "target-id", marker), "undecryptable") + self.assertEqual(ipc.target_state(answer([ + {"id": "target-id", "text": marker, "seal": "encrypted"}, + ]), key, "target-id", marker), "opened") + + + def test_peer_delivery_requires_encrypted_sent_server_id(self): + key = {"network": "tiny", "account": "sender", "id": "conversation"} + marker = "opened-history-cache-commit-only-target" + message_id = "11111111-1111-4111-8111-111111111111" + payload = {"op": "messages", "conversation": key, "messages": [ + {"id": message_id, "text": marker, "seal": "encrypted", "state": "sending"}, + ]} + self.assertIsNone(ipc.delivered_id(payload, key, marker)) + payload["messages"][0].pop("state") + self.assertEqual(ipc.delivered_id(payload, key, marker), message_id) + payload["messages"][0]["id"] = "temporary-local-id" + self.assertIsNone(ipc.delivered_id(payload, key, marker)) + + def test_backfill_is_observed_until_exact_row_arrives(self): + from unittest.mock import patch + import drive + key = {"network": "tiny", "account": "receiver", "id": "conversation"} + marker = "opened-history-cache-commit-only-target" + initial = {"op": "messages", "conversation": key, "messages": []} + arrived = {**initial, "messages": [{"id": "target-id", "seal": "undecryptable"}]} + contract = {"receiver": {"user_id": "receiver"}, + "conversation_id": "conversation", "marker": marker} + with patch.object(ipc, "messages", side_effect=[initial, arrived]): + state, page = drive.wait_target("scratch", "bundle", contract, "target-id", seconds=2) + self.assertEqual(state, "undecryptable") + self.assertEqual(page["messages"][0]["id"], "target-id") + + +if __name__ == "__main__": + unittest.main() diff --git a/e2e/scenarios/opened-history-cache-commit/test_fixture_controller.py b/e2e/scenarios/opened-history-cache-commit/test_fixture_controller.py new file mode 100644 index 00000000..484a9550 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/test_fixture_controller.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +"""Self-test for fixture_controller without chatd or network.""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import tempfile +import unittest +from pathlib import Path + +import fixture_controller as fixture + + +class FixtureControllerTests(unittest.TestCase): + def test_wipe_is_scoped_to_receiver_and_checks_conversation_before_delete(self): + with tempfile.TemporaryDirectory() as tmp: + profile = Path(tmp) + conn = sqlite3.connect(profile / "chat.sqlite") + conn.executescript(""" + CREATE TABLE conversation (network TEXT, account TEXT, id TEXT); + CREATE TABLE message (network TEXT, account TEXT, conversation TEXT, id TEXT); + INSERT INTO conversation VALUES ('tiny', 'receiver', 'target'); + INSERT INTO conversation VALUES ('tiny', 'other', 'target'); + INSERT INTO message VALUES ('tiny', 'receiver', 'target', 'mine'); + INSERT INTO message VALUES ('tiny', 'other', 'target', 'theirs'); + INSERT INTO message VALUES ('tiny', 'receiver', 'missing', 'orphan'); + """) + conn.commit() + conn.close() + def rows(query): + db = sqlite3.connect(profile / "chat.sqlite") + try: + return db.execute(query).fetchall() + finally: + db.close() + contract = {"conversation_id": "target", "receiver": { + "user_id": "receiver", "profile": str(profile), + }} + fixture.wipe_message_index(contract) + self.assertEqual(rows("SELECT id FROM message ORDER BY id"), + [("orphan",), ("theirs",)]) + contract["conversation_id"] = "missing" + with self.assertRaises(fixture.FixtureError): + fixture.wipe_message_index(contract) + self.assertEqual(rows("SELECT COUNT(*) FROM message WHERE id='orphan'"), [(1,)]) + + def test_blocker_is_empty_directory_and_clears(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + profile = root / "receiver" + user = "11111111-1111-4111-8111-111111111111" + conv = "33333333-3333-4333-8333-333333333333" + contract = { + "marker": "opened-history-cache-commit-test", + "conversation_id": conv, + "receiver": {"user_id": user, "profile": str(profile)}, + "sender": {"user_id": "22222222-2222-4222-8222-222222222222", "profile": str(root / "sender")}, + } + path = fixture.place_append_blocker(contract) + self.assertTrue(path.is_dir()) + self.assertFalse(path.is_file()) + parent = path.parent + self.assertTrue(parent.is_dir()) + (parent / "state.bin").write_bytes(b"keep-parent-writable") + fixture.clear_append_blocker(contract) + self.assertFalse(path.exists()) + self.assertEqual((parent / "state.bin").read_bytes(), b"keep-parent-writable") + + def test_wipe_keeps_conversation_row(self): + with tempfile.TemporaryDirectory() as tmp: + profile = Path(tmp) / "receiver" + profile.mkdir() + db = profile / "chat.sqlite" + conv = "33333333-3333-4333-8333-333333333333" + conn = sqlite3.connect(db) + user = "11111111-1111-4111-8111-111111111111" + conn.execute("CREATE TABLE conversation (network TEXT, account TEXT, id TEXT)") + conn.execute("CREATE TABLE message (network TEXT, account TEXT, id TEXT, conversation TEXT, text TEXT)") + conn.execute("INSERT INTO conversation VALUES ('tiny', ?, ?)", (user, conv)) + conn.execute( + "INSERT INTO message VALUES ('tiny', ?, 'm1', ?, 'hello')", + (user, conv), + ) + conn.commit() + conn.close() + contract = { + "marker": "opened-history-cache-commit-test", + "conversation_id": conv, + "receiver": { + "user_id": "11111111-1111-4111-8111-111111111111", + "profile": str(profile), + }, + "sender": { + "user_id": "22222222-2222-4222-8222-222222222222", + "profile": str(Path(tmp) / "sender"), + }, + } + wiped = fixture.wipe_message_index(contract) + self.assertEqual(wiped["after"], 0) + self.assertEqual(wiped["conversation_rows"], 1) + + def test_missing_fixture_is_not_valid_red(self): + env = os.environ.pop("E2E_CACHE_COMMIT_FIXTURE", None) + try: + with self.assertRaises(fixture.FixtureError) as raised: + fixture.load_contract("") + self.assertIn("not a valid base RED", str(raised.exception)) + finally: + if env is not None: + os.environ["E2E_CACHE_COMMIT_FIXTURE"] = env + + +if __name__ == "__main__": + unittest.main() diff --git a/e2e/scenarios/opened-history-cache-commit/test_fixture_keys.py b/e2e/scenarios/opened-history-cache-commit/test_fixture_keys.py new file mode 100644 index 00000000..2a86421c --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/test_fixture_keys.py @@ -0,0 +1,60 @@ +"""Scratch restarts must retain the same wrapping keys from their first launch.""" + +import base64 +import json +from pathlib import Path +import tempfile +import unittest + +import fixture_controller as fixture +import ipc + + +class FixtureKeysTests(unittest.TestCase): + def test_missing_or_bad_wrapping_keys_refuse_before_any_profile_is_created(self): + with tempfile.TemporaryDirectory() as scratch: + root = Path(scratch) + state = base64.b64encode(bytes(range(32))).decode() + cache = base64.b64encode(bytes(range(32, 64))).decode() + contract = { + "marker": "opened-history-cache-commit-fern-clover", "conversation_id": "conversation", + "receiver": {"user_id": "ada", "profile": str(root / "ada"), + "session": {"user_id": "ada", "display_name": "Ada", "e2ee": {"state": state, "cache": cache}}}, + "sender": {"user_id": "bea", "profile": str(root / "bea"), + "session": {"user_id": "bea", "display_name": "Bea", "e2ee": {"state": state, "cache": cache}}}, + } + path = root / "fixture.json" + for keys in (None, {}, {"state": state}, {"state": "not-base64", "cache": cache}, + {"state": base64.b64encode(b"short").decode(), "cache": cache}): + contract["receiver"]["session"]["e2ee"] = keys + path.write_text(json.dumps(contract), encoding="utf-8") + with self.subTest(keys_present=keys is not None), self.assertRaises(fixture.FixtureError): + fixture.load_contract(str(path)) + self.assertFalse((root / "ada").exists()) + contract["receiver"]["session"]["e2ee"] = {"state": state, "cache": cache} + path.write_text(json.dumps(contract), encoding="utf-8") + before = path.read_bytes() + one = fixture.load_contract(str(path)) + two = fixture.load_contract(str(path)) + self.assertEqual(one, two) + self.assertEqual(path.read_bytes(), before) + contract["sender"]["session"]["display_name"] = "Bea Test" + path.write_text(json.dumps(contract), encoding="utf-8") + with self.assertRaisesRegex(fixture.FixtureError, "single-word"): + fixture.load_contract(str(path)) + + def test_malformed_state_never_counts_as_a_delivered_encrypted_message(self): + marker = "opened-history-cache-commit-fern-clover" + for state in (None, {}, {"other": "sent"}, {"state": "pending"}, + {"state": "failed", "reason": "refused"}): + with self.subTest(state=state): + self.assertFalse(ipc.text_contains_marker({"op": "messages", "messages": [{ + "id": "one", "text": marker, "seal": "encrypted", "state": state, + }]}, marker)) + self.assertTrue(ipc.text_contains_marker({"op": "messages", "messages": [{ + "id": "one", "text": marker, "seal": "encrypted", "state": {"state": "sent"}, + }]}, marker)) + + +if __name__ == "__main__": + unittest.main() diff --git a/e2e/scenarios/opened-history-cache-commit/test_native_linux.py b/e2e/scenarios/opened-history-cache-commit/test_native_linux.py new file mode 100644 index 00000000..7009c6f6 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/test_native_linux.py @@ -0,0 +1,41 @@ +import unittest + +import native_linux + + +def word(text, x, y): + return {"text": text, "x": x, "y": y, "width": 60, "height": 18} + + +def intro(): + return [word(text, 420 + index * 65, 160) + for index, text in enumerate("This is the start of your conversation with Bea".split())] + + +class LinuxPaneTests(unittest.TestCase): + def test_sidebar_preview_cannot_certify_the_conversation(self): + words = [word("opened-history-cache-commit-rose", 25, 70), + *intro(), word("Can't", 420, 230), word("decrypt", 490, 230)] + self.assertIs(native_linux.conversation_marker(words, "opened-history-cache-commit-rose"), False) + + def test_body_marker_survives_ocr_word_and_line_splits(self): + words = [*intro(), word("opened-history-", 420, 220), + word("cache-commit-rose", 420, 245)] + self.assertIs(native_linux.conversation_marker(words, "opened-history-cache-commit-rose"), True) + + def test_partial_body_marker_does_not_pass(self): + words = [*intro(), word("opened-history-cache-commit", 420, 220)] + self.assertIs(native_linux.conversation_marker(words, "opened-history-cache-commit-rose"), False) + + def test_missing_conversation_intro_is_setup_incomplete(self): + self.assertIsNone(native_linux.conversation_marker( + [word("opened-history-cache-commit-rose", 25, 70)], "opened-history-cache-commit-rose")) + + def test_welcome_this_computer_text_is_not_a_conversation_intro(self): + words = [word("Link", 400, 513), word("this", 427, 513), word("computer", 460, 513), + word("Your", 350, 701), word("sign-in", 390, 701), word("this", 646, 701)] + self.assertIsNone(native_linux.conversation_marker(words, "opened-history-cache-commit-rose")) + + +if __name__ == "__main__": + unittest.main() diff --git a/e2e/scenarios/opened-history-cache-commit/test_pixels.py b/e2e/scenarios/opened-history-cache-commit/test_pixels.py new file mode 100644 index 00000000..b9545777 --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/test_pixels.py @@ -0,0 +1,51 @@ +"""Windows pixel-oracle integration; no app/IPC substitutes for image text.""" + +import base64 +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +import native_windows + + +@unittest.skipUnless(sys.platform == "win32", "requires Windows OCR") +class PixelTests(unittest.TestCase): + def test_reads_actual_png_and_rejects_an_absent_marker_without_tesseract(self): + with tempfile.TemporaryDirectory() as scratch: + image = Path(scratch) / "pixels.png" + body = Path(scratch) / "body.png" + path = str(image).replace("'", "''") + body_path = str(body).replace("'", "''") + script = f""" +Add-Type -AssemblyName System.Drawing +$bmp=New-Object Drawing.Bitmap 1600,300 +$g=[Drawing.Graphics]::FromImage($bmp) +$g.Clear([Drawing.Color]::White) +$font=New-Object Drawing.Font 'Segoe UI',26 +$g.DrawString('opened-history-cache-commit-fern-clover',$font,[Drawing.Brushes]::Black,20,60) +$g.DrawString('Bea',$font,[Drawing.Brushes]::Black,20,120) +$g.DrawString('Bea',$font,[Drawing.Brushes]::Black,500,120) +$g.DrawString('This is the start of your conversation with Bea.',$font,[Drawing.Brushes]::Black,400,160) +$bmp.Save('{path}',[Drawing.Imaging.ImageFormat]::Png) +$g.DrawString('opened-history-cache-commit-fern-clover',$font,[Drawing.Brushes]::Black,400,210) +$bmp.Save('{body_path}',[Drawing.Imaging.ImageFormat]::Png) +$font.Dispose(); $g.Dispose(); $bmp.Dispose() +""" + encoded = base64.b64encode(script.encode("utf-16-le")).decode("ascii") + subprocess.run(["powershell.exe", "-NoProfile", "-EncodedCommand", encoded], + check=True, capture_output=True, timeout=15) + self.assertIs(native_windows.png_mentions_marker(str(image), + "opened-history-cache-commit-fern-clover"), True) + self.assertIs(native_windows.png_mentions_marker(str(image), + "opened-history-cache-commit-wrong-marker"), False) + self.assertIs(native_windows.conversation_mentions_marker(str(image), + "opened-history-cache-commit-fern-clover"), False) + self.assertIs(native_windows.conversation_mentions_marker(str(body), + "opened-history-cache-commit-fern-clover"), True) + + +if __name__ == "__main__": + unittest.main() diff --git a/e2e/scenarios/opened-history-cache-commit/test_unix_ipc.py b/e2e/scenarios/opened-history-cache-commit/test_unix_ipc.py new file mode 100644 index 00000000..c35dd55a --- /dev/null +++ b/e2e/scenarios/opened-history-cache-commit/test_unix_ipc.py @@ -0,0 +1,61 @@ +import json +from pathlib import Path +import socket +import struct +import sys +import tempfile +import threading +import unittest + +import ipc + + +@unittest.skipUnless(sys.platform.startswith("linux"), "Linux Unix-socket transport") +class UnixIpcTests(unittest.TestCase): + def test_request_uses_the_profile_socket_and_length_prefixed_json(self): + with tempfile.TemporaryDirectory() as folder: + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(str(Path(folder) / "chat.sock")) + server.listen(1) + server.settimeout(1) + seen = [] + response = {"op": "status", "accounts": []} + + def serve(): + try: + connection, _ = server.accept() + with connection: + connection.settimeout(2) + def read_exact(length): + data = b"" + while len(data) < length: + chunk = connection.recv(length - len(data)) + if not chunk: + raise OSError("fixture client closed a partial frame") + data += chunk + return data + size = struct.unpack("!I", read_exact(4))[0] + data = read_exact(size) + seen.append(json.loads(data)) + encoded = json.dumps(response).encode() + connection.sendall(struct.pack("!I", len(encoded)) + encoded) + except (OSError, ValueError): + pass + + peer = threading.Thread(target=serve) + peer.start() + try: + try: + answer = ipc.request(folder, {"op": "status"}, app_dir=folder, timeout=2) + except RuntimeError as error: + answer = {"op": "error", "detail": str(error)} + self.assertEqual(answer, response) + self.assertEqual(seen, [{"op": "status"}]) + finally: + server.close() + peer.join(timeout=3) + self.assertFalse(peer.is_alive()) + + +if __name__ == "__main__": + unittest.main()