diff --git a/crates/kv-crypto/tests/vectors/envelope.json b/crates/kv-crypto/tests/vectors/envelope.json index 909da69..13f7f36 100644 --- a/crates/kv-crypto/tests/vectors/envelope.json +++ b/crates/kv-crypto/tests/vectors/envelope.json @@ -7,10 +7,10 @@ "vectors": [ { "aad": "6b657374726f76652f76312f6974656d000101010101014101810101010101010100020202020202420282020202020202020003030303030343038303030303030303", - "ciphertext": "0ae824b61b2253f40015d22f52ee7412e5be95cf17185bd8490e0e77200d50b5f629c6dfd2e2bccc1d86a06c6453150a96e1c8fc81f1a82b4830f511e831aa66438d35ed9c7d6d11d904b3d53641f05175066f63c0d985ea91738efcd7afb6a205b4a8471fc827a11c51cb598e429227ef2c3020ccdd749f254538ca5429e95f70b7258d87a6b52018976695a15b6e77e6c47a49609fa691f602dbeb4a15ea73c87909c7267d7bf84de31762480feab0b96553c2d4046b5ed00c899e30175fbd7333e9da01329d89a3158a715a85a83d81b8a0105e3c2cdd0ffb2f52e1971d0c7ac1c57504fea274b883fd4b65dae757d758cb3ecd7828e7edd44b35cbba452261039e436beeb3aeba178d379a54abf9", + "ciphertext": "12e824b61b2253f40015d22f52ee7412e5be95cf17185bd8490e0e77200d50b5f629c6dfd2e2bccc1d86a06c6453150a96e1c8fc81f1a82b4830f511e831aa66438d35ed9c7d6d11d904b3d53641f05175066f63c0d985ea91738efcd7afb6a205b4a8471fc827a11c51cb598e429227be0e4345bdff4eae584538ca5429e95f70b7258d87a6b52018976695a15b6e77e6c47a49609fa691f602dbeb4a15ea73c87909c7267d7bf84de31762480feab0b96553c2d4046b5ed00c899e30175fbd7333e9da01329d89a3158a715a85a83d81b8a0105e3c2cdd0ffb2f52e1971d0c7ac1c57504fea274b883fd4b65dae757d758cb3ecd7828e7edd44b35cbba4522a45f88134388c878b77659fb5a7ace37", "envelope": { "alg": "xc20p", - "ct": "CugkthsiU_QAFdIvUu50EuW-lc8XGFvYSQ4OdyANULX2Kcbf0uK8zB2GoGxkUxUKluHI_IHxqCtIMPUR6DGqZkONNe2cfW0R2QSz1TZB8FF1Bm9jwNmF6pFzjvzXr7aiBbSoRx_IJ6EcUctZjkKSJ-8sMCDM3XSfJUU4ylQp6V9wtyWNh6a1IBiXZpWhW2535sR6SWCfppH2AtvrShXqc8h5CccmfXv4TeMXYkgP6rC5ZVPC1ARrXtAMiZ4wF1-9czPp2gEynYmjFYpxWoWoPYG4oBBePCzdD_svUuGXHQx6wcV1BP6idLiD_Utl2udX11jLPs14KOft1Es1y7pFImEDnkNr7rOuuheNN5pUq_k", + "ct": "EugkthsiU_QAFdIvUu50EuW-lc8XGFvYSQ4OdyANULX2Kcbf0uK8zB2GoGxkUxUKluHI_IHxqCtIMPUR6DGqZkONNe2cfW0R2QSz1TZB8FF1Bm9jwNmF6pFzjvzXr7aiBbSoRx_IJ6EcUctZjkKSJ74OQ0W9_06uWEU4ylQp6V9wtyWNh6a1IBiXZpWhW2535sR6SWCfppH2AtvrShXqc8h5CccmfXv4TeMXYkgP6rC5ZVPC1ARrXtAMiZ4wF1-9czPp2gEynYmjFYpxWoWoPYG4oBBePCzdD_svUuGXHQx6wcV1BP6idLiD_Utl2udX11jLPs14KOft1Es1y7pFIqRfiBNDiMh4t3ZZ-1p6zjc", "kid": "04040404-0404-4404-8404-040404040404", "n": "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEB", "v": 1 @@ -19,7 +19,7 @@ "kid": "04040404-0404-4404-8404-040404040404", "name": "item payload (padded)", "nonce": "010101010101010101010101010101010101010101010101", - "plaintext": "6d0000007b226964223a2230333033303330332d303330332d343330332d383330332d303330333033303330333033222c226b696e64223a226c6f67696e222c227469746c65223a224578616d706c65222c22637265617465645f6174223a312c226d6f6469666965645f6174223a317d0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000" + "plaintext": "750000007b226964223a2230333033303330332d303330332d343330332d383330332d303330333033303330333033222c226b696e64223a226c6f67696e222c227469746c65223a224578616d706c65222c22637265617465645f6174223a312c226d6f6469666965645f6174223a312c22736571223a317d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000" }, { "aad": "6b657374726f76652f76312f6974656d2d6b6579000101010101014101810101010101010100020202020202420282020202020202020003030303030343038303030303030303", diff --git a/crates/kv-crypto/vectors/gen.rs b/crates/kv-crypto/vectors/gen.rs index 4d0e780..10ebee9 100644 --- a/crates/kv-crypto/vectors/gen.rs +++ b/crates/kv-crypto/vectors/gen.rs @@ -88,7 +88,7 @@ fn envelope_vectors() -> Value { let vault = id(0x02); let item = id(0x03); let key = SymKey::from_bytes(id(0x04), [0x42; 32]); - let item_json = br#"{"id":"03030303-0303-4303-8303-030303030303","kind":"login","title":"Example","created_at":1,"modified_at":1}"#; + let item_json = br#"{"id":"03030303-0303-4303-8303-030303030303","kind":"login","title":"Example","created_at":1,"modified_at":1,"seq":1}"#; type Case = (&'static str, ObjectType, Option, Option, Vec, [u8; 24]); let cases: Vec = vec![ ("item payload (padded)", ObjectType::Item, Some(vault), Some(item), pad(item_json).to_vec(), [0x01; 24]), diff --git a/crates/kv-ffi/src/lib.rs b/crates/kv-ffi/src/lib.rs index 6b10a9d..846a4e9 100644 --- a/crates/kv-ffi/src/lib.rs +++ b/crates/kv-ffi/src/lib.rs @@ -563,6 +563,8 @@ impl VaultCore { model.favorite = item.favorite; model.tags = item.tags; model.modified_at = now.max(previous.as_ref().map_or(0, |p| p.modified_at + 1)); + (model.seq, model.restores) = + kv_sync::engine::next_local_version(self.store.as_ref(), acct.account_id, vault, &u.vault_key, id)?; let json = Zeroizing::new(serde_json::to_vec(&model).map_err(internal)?); let sealed = seal_item(acct.account_id, vault, id, &u.vault_key, &json)?; diff --git a/crates/kv-ffi/src/tests.rs b/crates/kv-ffi/src/tests.rs index e3914ce..e01ff31 100644 --- a/crates/kv-ffi/src/tests.rs +++ b/crates/kv-ffi/src/tests.rs @@ -281,3 +281,44 @@ async fn pending_approval_and_biometric_session_expiry() { connect(&c, &server); assert!(matches!(c.register_account("x".into()).await, Err(CoreError::NotAuthenticated))); } + +#[tokio::test] +async fn save_item_bumps_seq_and_replays_are_refused() { + let server = TestServer::start().await; + let a = VaultCore::new_in_memory(); + let sk = a.create_account(PW.into()).unwrap(); + connect(&a, &server); + a.register_account(SETUP_TOKEN.into()).await.unwrap(); + let id = a.save_item(login_item("Bank", "pw1")).unwrap(); + a.sync_now().await.unwrap(); + let vid = *server.state().vaults.keys().next().unwrap(); + let iid = Uuid::parse_str(&id).unwrap(); + let old = server.state().vaults[&vid].items[&iid].clone(); + + let seq_of = |core: &VaultCore| { + let (acct, u) = core.unlocked_snapshot().unwrap(); + let li = core.store.get(acct.primary_vault, iid).unwrap().unwrap(); + decrypt(&acct, &u.vault_key, &li.record).unwrap().unwrap().seq + }; + assert_eq!(seq_of(&a), 1); + for pw in ["pw2", "pw3"] { + let mut it = a.get_item(id.clone()).unwrap(); + it.password = Some(pw.into()); + a.save_item(it).unwrap(); + } + assert_eq!(seq_of(&a), 3); + a.sync_now().await.unwrap(); + + let b = VaultCore::new_in_memory(); + connect(&b, &server); + b.login(PW.into(), sk, None).await.unwrap(); + b.sync_now().await.unwrap(); + assert_eq!(seq_of(&b), 3); + + server.state().write_item(vid, old); + match b.sync_now().await { + Err(CoreError::IntegrityAlarm { reason, .. }) => assert!(reason.contains("older version"), "{reason}"), + other => panic!("expected integrity alarm, got {other:?}"), + } + assert_eq!(b.get_item(id).unwrap().password.as_deref(), Some("pw3")); +} diff --git a/crates/kv-model/src/item.rs b/crates/kv-model/src/item.rs index 02c3142..5adedfc 100644 --- a/crates/kv-model/src/item.rs +++ b/crates/kv-model/src/item.rs @@ -52,6 +52,14 @@ pub struct Item { pub android_apps: Vec, pub created_at: i64, pub modified_at: i64, + /// Per-item edit counter, bound inside the ciphertext: every new version has a larger `seq` than the + /// versions it derives from (docs/02 "Item plaintext", docs/03 "Per-item replay protection"). + #[serde(default)] + pub seq: u64, + /// Server revision of the tombstone this version was written over (edit beats delete). Only set on that + /// one version. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub restores: Option, } impl Item { @@ -71,6 +79,8 @@ impl Item { android_apps: vec![], created_at: now, modified_at: now, + seq: 1, + restores: None, } } } diff --git a/crates/kv-sync/src/engine.rs b/crates/kv-sync/src/engine.rs index bced099..616543c 100644 --- a/crates/kv-sync/src/engine.rs +++ b/crates/kv-sync/src/engine.rs @@ -1,7 +1,8 @@ //! The client sync algorithm (docs/03 "Sync algorithm", "Conflict resolution", "Integrity"). //! //! Per vault: pull `/changes` from the cursor, verify the signed manifest, push pending edits with -//! `If-Match` (on 412: pull, merge, retry), pull again, then publish a new manifest. +//! `If-Match` (on 412: pull, merge, retry), pull again, then publish a new manifest. Every pulled record is +//! checked against the local base for per-item replay ([`crate::replay`]). use std::time::{SystemTime, UNIX_EPOCH}; @@ -18,6 +19,7 @@ use crate::{ client::ApiClient, manifest::{self, ManifestBody, ManifestError}, merge::merge3, + replay::{self, Base}, SyncError, }; @@ -104,6 +106,49 @@ fn history_entry(r: &ItemRecord) -> Option { }) } +/// The decrypted form of a local base, for [`replay`]. Owns the plaintext a [`Base::Live`] borrows. +enum OpenedBase { + None, + Tombstone(u64), + Live(u64, Value), +} + +impl OpenedBase { + fn open(account_id: Uuid, vault_id: Uuid, vk: &SymKey, base: Option<&ItemRecord>) -> Result { + Ok(match base { + None => Self::None, + Some(b) if b.deleted => Self::Tombstone(b.revision), + Some(b) => match open_record(account_id, vault_id, vk, b)? { + Some(v) => Self::Live(b.revision, v), + None => Self::None, + }, + }) + } + + fn as_base(&self) -> Base<'_> { + match self { + Self::None => Base::None, + Self::Tombstone(revision) => Base::Tombstone { revision: *revision }, + Self::Live(revision, item) => Base::Live { revision: *revision, item }, + } + } +} + +/// The `seq` and `restores` a new local version of `item_id` must carry (docs/03 "Per-item replay +/// protection"). Call it before sealing a local edit and stamp the result into the plaintext. +pub fn next_local_version( + store: &dyn Store, + account_id: Uuid, + vault_id: Uuid, + vk: &SymKey, + item_id: Uuid, +) -> Result<(u64, Option), SyncError> { + let Some(local) = store.get(vault_id, item_id)? else { return Ok((1, None)) }; + let working = open_record(account_id, vault_id, vk, &local.record)?; + let base = OpenedBase::open(account_id, vault_id, vk, local.base.as_ref())?; + Ok(replay::next_version(working.as_ref(), base.as_base())) +} + /// The server's view of a vault: the base (last confirmed) version of every item. fn server_view(store: &dyn Store, vault: Uuid) -> Result, SyncError> { Ok(store.list(vault)?.into_iter().filter_map(|i| i.base).collect()) @@ -248,17 +293,42 @@ impl SyncContext<'_> { report: &mut SyncReport, ) -> Result<(), SyncError> { let Some(local) = self.store.get(vault, remote.item_id)? else { + self.check_replay(vault, vk, None, remote)?; return Ok(self.store.put_remote(vault, remote)?); }; if remote.revision <= local.base_revision() { return Ok(()); // our own write coming back, or a replay of something older } + self.check_replay(vault, vk, local.base.as_ref(), remote)?; if local.state == SyncState::Clean { return Ok(self.store.put_remote(vault, remote)?); } self.resolve_conflict(vault, vk, local, remote, report) } + /// Integrity alarm unless `remote` authenticates and may follow `base` (docs/03 "Per-item replay + /// protection"). Tombstones carry no ciphertext and pass. + fn check_replay( + &self, + vault: Uuid, + vk: &SymKey, + base: Option<&ItemRecord>, + remote: &ItemRecord, + ) -> Result<(), SyncError> { + if remote.deleted { + return Ok(()); + } + let id = remote.item_id; + let incoming = match open_record(self.account_id(), vault, vk, remote) { + Ok(Some(v)) => v, + Ok(None) => return Err(self.alarm(vault, format!("item {id}: live record without ciphertext"))), + Err(_) => return Err(self.alarm(vault, format!("item {id}: envelope does not authenticate"))), + }; + let base = OpenedBase::open(self.account_id(), vault, vk, base)?; + replay::check_successor(base.as_base(), remote.revision, &incoming) + .map_err(|e| self.alarm(vault, format!("item {id}: {e}"))) + } + /// `local` is pending and `remote` is a newer server version of the same item. fn resolve_conflict( &self, @@ -282,8 +352,15 @@ impl SyncContext<'_> { report.restored.push(id); } (false, true) => { - // Our edit beats their delete: keep ours and push it on top of the tombstone. - self.store.put(vault, &rebased(local.record))?; + // Our edit beats their delete: keep ours and push it on top of the tombstone, marked with + // the tombstone's revision so devices that never saw the delete accept a lower `seq`. + let acct = self.account_id(); + let mut mine = + open_record(acct, vault, vk, &local.record)?.ok_or(SyncError::Protocol("empty item".into()))?; + let seq = replay::seq_of(&mine); + replay::stamp(&mut mine, seq, Some(remote.revision)); + let sealed = seal_record(acct, vault, vk, id, &mine, self.device_id)?; + self.store.put(vault, &rebased(sealed))?; report.restored.push(id); } (false, false) => { @@ -295,7 +372,7 @@ impl SyncContext<'_> { let mine = open_record(acct, vault, vk, &local.record)?.ok_or(SyncError::Protocol("empty item".into()))?; let theirs = open_record(acct, vault, vk, remote)?.ok_or(SyncError::Protocol("empty item".into()))?; - let out = merge3(base.as_ref(), &mine, &theirs); + let mut out = merge3(base.as_ref(), &mine, &theirs); if out.local_lost { if let Some(h) = history_entry(&local.record) { self.store.push_history(vault, id, &h)?; @@ -306,9 +383,11 @@ impl SyncContext<'_> { self.store.push_history(vault, id, &h)?; } } - if out.merged == theirs { + if replay::same_content(&out.merged, &theirs) { self.store.put_remote(vault, remote)?; } else { + let seq = replay::seq_of(&mine).max(replay::seq_of(&theirs)) + 1; + replay::stamp(&mut out.merged, seq, None); let sealed = seal_record(acct, vault, vk, id, &out.merged, self.device_id)?; self.store.put(vault, &rebased(sealed))?; } @@ -335,8 +414,11 @@ impl SyncContext<'_> { if body.head_revision > row.cursor { return Err(SyncError::ManifestAhead); } - if body.head_revision == row.cursor && manifest::merkle_root_of(&server_view(self.store, vault)?) != body.items - { + // `/changes` only returns the latest revision of each item, so the state at `head_revision` can be + // rebuilt only when nothing we hold is newer. Then it equals our state (anything in + // (head_revision, cursor] would have been pulled), and the root must match. + let view = server_view(self.store, vault)?; + if view.iter().all(|r| r.revision <= body.head_revision) && manifest::merkle_root_of(&view) != body.items { return Err(integrity(ManifestError::MerkleMismatch)); } if body.version > row.manifest_version { diff --git a/crates/kv-sync/src/lib.rs b/crates/kv-sync/src/lib.rs index 08b4618..4137a99 100644 --- a/crates/kv-sync/src/lib.rs +++ b/crates/kv-sync/src/lib.rs @@ -8,12 +8,14 @@ //! a callback into platform code ([`DeviceSigner`]). //! * [`engine`]: pull, then push with `If-Match`, then three-way field merge on 412 ([`merge`]), then manifest verify //! and publish ([`manifest`]). +//! * [`replay`]: per-item replay check on pull (the `seq` / `restores` fields of the item plaintext). pub mod client; pub mod engine; pub mod manifest; pub mod merge; pub mod opaque; +pub mod replay; pub mod signing; #[cfg(feature = "test-server")] pub mod test_server; diff --git a/crates/kv-sync/src/merge.rs b/crates/kv-sync/src/merge.rs index acafb61..f3fccf2 100644 --- a/crates/kv-sync/src/merge.rs +++ b/crates/kv-sync/src/merge.rs @@ -1,11 +1,14 @@ //! Three-way, field-level merge of decrypted items (docs/03 "Conflict resolution"). //! //! Items are merged as JSON objects so every top-level field of `kv_model::Item` (and any field a -//! newer client adds) is a merge unit. `id` and `created_at` come from the base, `modified_at` is the -//! maximum of both sides. +//! newer client adds) is a merge unit. `id` and `created_at` come from the base, `modified_at` and +//! `seq` are the maximum of both sides, and `restores` is dropped. The caller stamps the next `seq` if it +//! writes the result as a new version ([`crate::replay`]). use serde_json::{Map, Value}; +use crate::replay::{seq_of, RESTORES, SEQ}; + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Side { Local, @@ -22,7 +25,7 @@ pub struct MergeOutcome { pub conflicting_fields: Vec, } -const FIXED: [&str; 3] = ["id", "created_at", "modified_at"]; +const FIXED: [&str; 5] = ["id", "created_at", "modified_at", SEQ, RESTORES]; fn modified_at(v: &Value) -> i64 { v.get("modified_at").and_then(Value::as_i64).unwrap_or(0) @@ -85,6 +88,7 @@ pub fn merge3(base: Option<&Value>, local: &Value, remote: &Value) -> MergeOutco out.insert("created_at".into(), c); } out.insert("modified_at".into(), Value::from(modified_at(local).max(modified_at(remote)))); + out.insert(SEQ.into(), Value::from(seq_of(local).max(seq_of(remote)))); let lost = !conflicting.is_empty(); MergeOutcome { diff --git a/crates/kv-sync/src/replay.rs b/crates/kv-sync/src/replay.rs new file mode 100644 index 0000000..7a1f92b --- /dev/null +++ b/crates/kv-sync/src/replay.rs @@ -0,0 +1,129 @@ +//! Per-item replay protection (docs/03 "Per-item replay protection"). +//! +//! The AAD binds an envelope to its item but not to a server revision, so a server could re-serve an +//! old, authentic envelope under a new revision. Every item plaintext therefore carries `seq`, an edit +//! counter that grows along every honest chain of versions, and `restores`, the revision of the +//! tombstone a version was written over (tombstones carry no `seq`). Pulled records are checked +//! against the local base, the last version the server confirmed to this device. + +use serde_json::Value; + +/// The plaintext field holding the per-item edit counter. Absent means 0. +pub const SEQ: &str = "seq"; +/// The plaintext field holding the revision of the tombstone a version was written over. +pub const RESTORES: &str = "restores"; + +pub fn seq_of(item: &Value) -> u64 { + item.get(SEQ).and_then(Value::as_u64).unwrap_or(0) +} + +pub fn restores_of(item: &Value) -> Option { + item.get(RESTORES).and_then(Value::as_u64) +} + +/// Sets `seq` and `restores` (removing `restores` when `None`). +pub fn stamp(item: &mut Value, seq: u64, restores: Option) { + let Some(obj) = item.as_object_mut() else { return }; + obj.insert(SEQ.into(), Value::from(seq)); + match restores { + Some(t) => obj.insert(RESTORES.into(), Value::from(t)), + None => obj.remove(RESTORES), + }; +} + +/// Whether two item plaintexts have the same content, ignoring `seq` and `restores`. +pub fn same_content(a: &Value, b: &Value) -> bool { + let strip = |v: &Value| { + let mut v = v.clone(); + if let Some(o) = v.as_object_mut() { + o.remove(SEQ); + o.remove(RESTORES); + } + v + }; + strip(a) == strip(b) +} + +/// The local base of an item. +#[derive(Debug, Clone, Copy)] +pub enum Base<'a> { + /// This device holds no server version of the item. + None, + Tombstone { + revision: u64, + }, + Live { + revision: u64, + item: &'a Value, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("server re-served an older version of the item (seq {incoming} after seq {held})")] +pub struct Replay { + pub held: u64, + pub incoming: u64, +} + +/// Checks that `incoming` (a live record at server `revision`) may follow `base`. +pub fn check_successor(base: Base<'_>, revision: u64, incoming: &Value) -> Result<(), Replay> { + let Base::Live { revision: base_revision, item: held } = base else { return Ok(()) }; + let (held_seq, seq) = (seq_of(held), seq_of(incoming)); + if seq > held_seq { + return Ok(()); + } + // A delete this device never saw came between its base and `incoming`. + if restores_of(incoming).is_some_and(|t| base_revision < t && t < revision) { + return Ok(()); + } + // The same content again is not a rollback. + if seq == held_seq && incoming == held { + return Ok(()); + } + Err(Replay { held: held_seq, incoming: seq }) +} + +/// `seq` and `restores` for a new local version of an item whose working copy is `local` (if live). +pub fn next_version(local: Option<&Value>, base: Base<'_>) -> (u64, Option) { + let held = match base { + Base::Live { item, .. } => seq_of(item), + _ => 0, + }; + let seq = local.map_or(0, seq_of).max(held) + 1; + let restores = match base { + Base::Tombstone { revision } => Some(revision), + _ => None, + }; + (seq, restores) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn stamp_and_same_content() { + let mut v = json!({"title": "a", "restores": 4}); + stamp(&mut v, 7, None); + assert_eq!(v, json!({"title": "a", "seq": 7})); + stamp(&mut v, 8, Some(3)); + assert_eq!(v, json!({"title": "a", "seq": 8, "restores": 3})); + assert!(same_content(&v, &json!({"title": "a"}))); + assert!(!same_content(&v, &json!({"title": "b", "seq": 8}))); + } + + #[test] + fn next_version_rules() { + let base = json!({"seq": 5}); + let local = json!({"seq": 6}); + assert_eq!(next_version(None, Base::None), (1, None)); + assert_eq!(next_version(Some(&base), Base::Live { revision: 3, item: &base }), (6, None)); + assert_eq!(next_version(Some(&local), Base::Live { revision: 3, item: &base }), (7, None)); + // A pending delete being re-saved continues from the base. + assert_eq!(next_version(None, Base::Live { revision: 3, item: &base }), (6, None)); + assert_eq!(next_version(None, Base::Tombstone { revision: 9 }), (1, Some(9))); + assert_eq!(next_version(Some(&local), Base::Tombstone { revision: 9 }), (7, Some(9))); + } +} diff --git a/crates/kv-sync/tests/replay_vectors.rs b/crates/kv-sync/tests/replay_vectors.rs new file mode 100644 index 0000000..78e2257 --- /dev/null +++ b/crates/kv-sync/tests/replay_vectors.rs @@ -0,0 +1,47 @@ +//! Checks `kv_sync::replay` against the shared decision table (kestrove-spec/vectors/item_replay.json). + +use kv_sync::replay::{check_successor, next_version, Base}; +use serde_json::Value; + +fn vectors() -> Value { + serde_json::from_str(include_str!("vectors/item_replay.json")).unwrap() +} + +fn base(v: &Value) -> Base<'_> { + if v.is_null() { + return Base::None; + } + let revision = v["revision"].as_u64().unwrap(); + if v["tombstone"] == Value::Bool(true) { + Base::Tombstone { revision } + } else { + Base::Live { revision, item: &v["item"] } + } +} + +#[test] +fn check_table() { + for case in vectors()["check"].as_array().unwrap() { + let got = check_successor(base(&case["base"]), case["revision"].as_u64().unwrap(), &case["incoming"]); + let want = case["result"].as_str().unwrap(); + assert_eq!(if got.is_ok() { "accept" } else { "alarm" }, want, "{}", case["name"]); + } +} + +#[test] +fn next_version_table() { + for case in vectors()["next_version"].as_array().unwrap() { + let local = Some(&case["local"]).filter(|v| !v.is_null()); + let (seq, restores) = next_version(local, base(&case["base"])); + assert_eq!(Some(seq), case["seq"].as_u64(), "{}", case["name"]); + assert_eq!(restores, case["restores"].as_u64(), "{}", case["name"]); + } +} + +#[test] +fn spec_copy_is_identical() { + let spec = concat!(env!("CARGO_MANIFEST_DIR"), "/../../../kestrove-spec/vectors/item_replay.json"); + if let Ok(s) = std::fs::read_to_string(spec) { + assert_eq!(s, include_str!("vectors/item_replay.json"), "kestrove-spec copy differs"); + } +} diff --git a/crates/kv-sync/tests/sync.rs b/crates/kv-sync/tests/sync.rs index 2126e56..72f7f2a 100644 --- a/crates/kv-sync/tests/sync.rs +++ b/crates/kv-sync/tests/sync.rs @@ -10,7 +10,8 @@ use kv_crypto::{ use kv_proto::{DeviceEnroll, ItemRecord, Platform}; use kv_store::{MemoryStore, Store, SyncState, VaultRow}; use kv_sync::{ - engine::{open_record, seal_record}, + engine::{next_local_version, open_record, seal_record}, + replay, signing::SoftwareSigner, test_server::{TestServer, SETUP_TOKEN}, ApiClient, DeviceSigner, LoginOutcome, ServerConfig, SyncContext, SyncError, SyncReport, @@ -102,7 +103,15 @@ impl Device { self.keyset.open_vault_key(vid, &self.store.vault(vid).unwrap().unwrap().sealed_key).unwrap() } - fn save(&self, vid: Uuid, id: Uuid, item: Value) { + /// Saves a local edit the way a client does: with the next `seq` (and `restores`) stamped in. + fn save(&self, vid: Uuid, id: Uuid, mut item: Value) { + let (seq, restores) = next_local_version(&self.store, self.keyset.account_id, vid, &self.vk(vid), id).unwrap(); + replay::stamp(&mut item, seq, restores); + self.save_raw(vid, id, item); + } + + /// Saves exactly `item`, e.g. a legacy item without `seq`. + fn save_raw(&self, vid: Uuid, id: Uuid, item: Value) { let rec = seal_record(self.keyset.account_id, vid, &self.vk(vid), id, &item, self.signer.device_id()).unwrap(); self.store.put_local(vid, &rec).unwrap(); } @@ -522,3 +531,204 @@ mod tls { assert!(matches!(client(&p, port, vec![pin]).server_info().await, Err(SyncError::Network(_)))); } } + +// ---------------------------------------------------------------- per-item replay (docs/03) + +fn assert_replay_alarm(r: Result, vid: Uuid) { + let err = r.unwrap_err(); + assert!( + matches!(err, SyncError::Integrity { vault_id, ref reason } if vault_id == vid && reason.contains("older version")), + "{err}" + ); +} + +/// The server re-serves an old, authentic envelope of an item under a new, higher revision. +#[tokio::test] +async fn replayed_old_envelope_is_an_integrity_alarm() { + let server = TestServer::start().await; + let (a, vid) = Device::first(&server).await; + let id = Uuid::new_v4(); + a.save(vid, id, item(id, "Bank", "pw1", 1)); + a.sync().await.unwrap(); + let old = server.state().vaults[&vid].items[&id].clone(); + for (i, pw) in ["pw2", "pw3"].iter().enumerate() { + a.save(vid, id, item(id, "Bank", pw, 2 + i as i64)); + a.sync().await.unwrap(); + } + let b = Device::second(&server, &a).await; + b.sync().await.unwrap(); + assert_eq!(b.read(vid, id).unwrap()["seq"], 3); + let c = Device::second(&server, &a).await; + c.sync().await.unwrap(); + let manifest_version = server.state().vaults[&vid].manifest.as_ref().unwrap().version; + + let replayed_at = server.state().write_item(vid, old); + // A clean device and a device with a pending edit of the item both refuse it. + c.save(vid, id, item(id, "Bank", "pw-c", 9)); + assert_replay_alarm(b.sync().await, vid); + assert_replay_alarm(c.sync().await, vid); + assert_replay_alarm(a.sync().await, vid); + for d in [&a, &b] { + assert_eq!(d.read(vid, id).unwrap()["password"], "pw3", "not rolled back"); + assert!(d.store.vault(vid).unwrap().unwrap().integrity_alarm); + } + // Nobody pushed or signed over the rolled-back state. + let st = server.state(); + assert_eq!(st.vaults[&vid].items[&id].revision, replayed_at); + assert_eq!(st.vaults[&vid].manifest.as_ref().unwrap().version, manifest_version); +} + +#[tokio::test] +async fn replay_of_the_current_version_is_harmless() { + let server = TestServer::start().await; + let (a, vid) = Device::first(&server).await; + let id = Uuid::new_v4(); + a.save(vid, id, item(id, "Bank", "pw1", 1)); + a.sync().await.unwrap(); + let b = Device::second(&server, &a).await; + b.sync().await.unwrap(); + let current = server.state().vaults[&vid].items[&id].clone(); + server.state().write_item(vid, current); + a.sync().await.unwrap(); + b.sync().await.unwrap(); + assert_eq!(b.read(vid, id).unwrap()["password"], "pw1"); +} + +#[tokio::test] +async fn tampered_envelope_is_an_integrity_alarm() { + let server = TestServer::start().await; + let (a, vid) = Device::first(&server).await; + let id = Uuid::new_v4(); + a.save(vid, id, item(id, "Bank", "pw1", 1)); + a.sync().await.unwrap(); + let b = Device::second(&server, &a).await; + let mut rec = server.state().vaults[&vid].items[&id].clone(); + rec.payload.as_mut().unwrap().ct[0] ^= 1; + server.state().write_item(vid, rec); + let err = b.sync().await.unwrap_err(); + assert!(matches!(err, SyncError::Integrity { ref reason, .. } if reason.contains("authenticate")), "{err}"); +} + +/// Concurrent edits from the same base carry the same `seq`; that is a merge, not an alarm, on the two +/// writers and on a third device that sees the versions one by one or only the last. +#[tokio::test] +async fn concurrent_equal_seq_edits_are_not_an_alarm() { + let server = TestServer::start().await; + let (a, vid) = Device::first(&server).await; + let id = Uuid::new_v4(); + a.save(vid, id, item(id, "Mail", "p0", 1)); + a.sync().await.unwrap(); + let b = Device::second(&server, &a).await; + let c = Device::second(&server, &a).await; + let d = Device::second(&server, &a).await; + for x in [&b, &c, &d] { + x.sync().await.unwrap(); + } + + a.save(vid, id, item(id, "Mail", "pA", 30)); + b.save(vid, id, json!({"id": id, "kind": "login", "title": "Mail", "password": "p0", "notes": "nB", "created_at": 1, "modified_at": 20})); + assert_eq!(a.read(vid, id).unwrap()["seq"], 2); + assert_eq!(b.read(vid, id).unwrap()["seq"], 2); + a.sync().await.unwrap(); + c.sync().await.unwrap(); // sees A's seq 2 + assert_eq!(b.sync().await.unwrap().merged, 1); + let merged = b.read(vid, id).unwrap(); + assert_eq!(merged["seq"], 3); + assert!(merged.get("restores").is_none()); + for x in [&a, &c, &d] { + x.sync().await.unwrap(); // d jumps from seq 1 straight to 3 + assert_eq!(x.read(vid, id).unwrap(), merged); + } + // Merging into exactly the remote content adopts it without writing a new version. + a.save(vid, id, item(id, "Mail", "same", 40)); + c.save(vid, id, item(id, "Mail", "same", 40)); + a.sync().await.unwrap(); + let head = server.state().vaults[&vid].head; + c.sync().await.unwrap(); + assert_eq!(server.state().vaults[&vid].head, head, "c adopted a's version"); + assert!(c.store.pending(vid).unwrap().is_empty()); +} + +/// An edit that beats a delete may carry a lower `seq` than versions other devices saw before the delete. +/// `restores` lets a device that never saw the tombstone accept it, and only directly after that delete. +#[tokio::test] +async fn edit_beats_delete_is_not_an_alarm_but_its_replay_is() { + let server = TestServer::start().await; + let (a, vid) = Device::first(&server).await; + let id = Uuid::new_v4(); + a.save(vid, id, item(id, "X", "x1", 1)); + a.sync().await.unwrap(); + let b = Device::second(&server, &a).await; + b.sync().await.unwrap(); // b holds seq 1 + for (i, pw) in ["x2", "x3", "x4"].iter().enumerate() { + a.save(vid, id, item(id, "X", pw, 2 + i as i64)); + a.sync().await.unwrap(); + } + let c = Device::second(&server, &a).await; + c.sync().await.unwrap(); // c holds seq 4 and will not see the tombstone + a.delete(vid, id); + a.sync().await.unwrap(); + let tombstone_rev = server.state().vaults[&vid].items[&id].revision; + let d = Device::second(&server, &a).await; + d.sync().await.unwrap(); // d holds the tombstone + + b.save(vid, id, item(id, "X", "x-b", 9)); + let r = b.sync().await.unwrap(); + assert_eq!(r.restored, vec![id]); + let restored = b.read(vid, id).unwrap(); + assert_eq!((restored["seq"].as_u64(), restored["restores"].as_u64()), (Some(2), Some(tombstone_rev))); + let restore_envelope = server.state().vaults[&vid].items[&id].clone(); + + for x in [&a, &c, &d] { + x.sync().await.unwrap(); + assert_eq!(x.read(vid, id).unwrap()["password"], "x-b"); + } + // The next edit clears `restores` and continues from the restored `seq`. + c.save(vid, id, item(id, "X", "x-c", 10)); + assert_eq!(c.read(vid, id).unwrap()["seq"], 3); + assert!(c.read(vid, id).unwrap().get("restores").is_none()); + c.sync().await.unwrap(); + a.sync().await.unwrap(); + + // Re-serving the restore later is a rollback: the receiver's base is past the tombstone now. + server.state().write_item(vid, restore_envelope); + assert_replay_alarm(a.sync().await, vid); +} + +/// Items written before `seq` existed read as seq 0; the first edit writes seq 1. +#[tokio::test] +async fn legacy_items_without_seq() { + let server = TestServer::start().await; + let (a, vid) = Device::first(&server).await; + let id = Uuid::new_v4(); + a.save_raw(vid, id, item(id, "Old", "p", 1)); + a.sync().await.unwrap(); + let b = Device::second(&server, &a).await; + b.sync().await.unwrap(); + assert!(b.read(vid, id).unwrap().get("seq").is_none()); + b.save(vid, id, item(id, "Old", "p2", 2)); + assert_eq!(b.read(vid, id).unwrap()["seq"], 1); + b.sync().await.unwrap(); + a.sync().await.unwrap(); + assert_eq!(a.read(vid, id).unwrap()["password"], "p2"); +} + +/// The Merkle root is also checked when the cursor is past the manifest's head but nothing held is newer. +#[tokio::test] +async fn hidden_item_is_caught_when_cursor_is_past_manifest_head() { + let server = TestServer::start().await; + let (a, vid) = Device::first(&server).await; + let (x, y) = (Uuid::new_v4(), Uuid::new_v4()); + a.save(vid, x, item(x, "X", "x", 1)); + a.save(vid, y, item(y, "Y", "y", 1)); + a.sync().await.unwrap(); + // Another device writes y without publishing a manifest yet, and the server hides y. + let mut rec = server.state().vaults[&vid].items[&y].clone(); + rec.updated_by = Uuid::nil(); + server.state().write_item(vid, rec); + server.state().hide_items = vec![y]; + + let b = Device::second(&server, &a).await; + let err = b.sync().await.unwrap_err(); + assert!(matches!(err, SyncError::Integrity { ref reason, .. } if reason.contains("Merkle")), "{err}"); +} diff --git a/crates/kv-sync/tests/vectors/item_replay.json b/crates/kv-sync/tests/vectors/item_replay.json new file mode 100644 index 0000000..837f4f5 --- /dev/null +++ b/crates/kv-sync/tests/vectors/item_replay.json @@ -0,0 +1,27 @@ +{ + "description": "Per-item replay protection (docs/03 'Per-item replay protection'). Item plaintexts are shown as JSON; only seq, restores and content equality matter. 'base' is the last server-confirmed version this device holds: null (none), {tombstone, revision} or {revision, item}.", + "check": [ + { "name": "no base: first sight of the item", "base": null, "revision": 7, "incoming": {"title": "a", "seq": 1}, "result": "accept" }, + { "name": "tombstone base: anything may follow a delete", "base": {"tombstone": true, "revision": 5}, "revision": 7, "incoming": {"title": "a", "seq": 1}, "result": "accept" }, + { "name": "higher seq", "base": {"revision": 5, "item": {"title": "a", "seq": 3}}, "revision": 7, "incoming": {"title": "b", "seq": 4}, "result": "accept" }, + { "name": "seq jumps by more than one", "base": {"revision": 5, "item": {"title": "a", "seq": 3}}, "revision": 7, "incoming": {"title": "b", "seq": 9}, "result": "accept" }, + { "name": "older seq: replay of an old envelope", "base": {"revision": 5, "item": {"title": "a", "seq": 3}}, "revision": 7, "incoming": {"title": "old", "seq": 2}, "result": "alarm" }, + { "name": "same seq, different content", "base": {"revision": 5, "item": {"title": "a", "seq": 3}}, "revision": 7, "incoming": {"title": "b", "seq": 3}, "result": "alarm" }, + { "name": "same seq, identical content", "base": {"revision": 5, "item": {"title": "a", "seq": 3}}, "revision": 7, "incoming": {"title": "a", "seq": 3}, "result": "accept" }, + { "name": "legacy items: absent seq is 0", "base": {"revision": 5, "item": {"title": "a"}}, "revision": 7, "incoming": {"title": "b", "seq": 0}, "result": "alarm" }, + { "name": "legacy base, first seq-aware edit", "base": {"revision": 5, "item": {"title": "a"}}, "revision": 7, "incoming": {"title": "b", "seq": 1}, "result": "accept" }, + { "name": "restore over a tombstone this device did not see", "base": {"revision": 5, "item": {"title": "a", "seq": 6}}, "revision": 9, "incoming": {"title": "r", "seq": 2, "restores": 8}, "result": "accept" }, + { "name": "restore claim older than the base", "base": {"revision": 8, "item": {"title": "a", "seq": 6}}, "revision": 9, "incoming": {"title": "r", "seq": 2, "restores": 7}, "result": "alarm" }, + { "name": "restore claim equal to the base revision", "base": {"revision": 8, "item": {"title": "a", "seq": 6}}, "revision": 9, "incoming": {"title": "r", "seq": 2, "restores": 8}, "result": "alarm" }, + { "name": "restore claim not below the record's own revision", "base": {"revision": 5, "item": {"title": "a", "seq": 6}}, "revision": 9, "incoming": {"title": "r", "seq": 2, "restores": 9}, "result": "alarm" } + ], + "next_version": [ + { "name": "new item", "local": null, "base": null, "seq": 1, "restores": null }, + { "name": "edit of a clean item", "local": {"seq": 3}, "base": {"revision": 5, "item": {"seq": 3}}, "seq": 4, "restores": null }, + { "name": "second edit before syncing", "local": {"seq": 4}, "base": {"revision": 5, "item": {"seq": 3}}, "seq": 5, "restores": null }, + { "name": "re-save after a pending local delete", "local": null, "base": {"revision": 5, "item": {"seq": 3}}, "seq": 4, "restores": null }, + { "name": "legacy item", "local": {}, "base": {"revision": 5, "item": {}}, "seq": 1, "restores": null }, + { "name": "save over a tombstone", "local": null, "base": {"tombstone": true, "revision": 8}, "seq": 1, "restores": 8 }, + { "name": "edit after a restore clears restores", "local": {"seq": 2, "restores": 8}, "base": {"revision": 9, "item": {"seq": 2, "restores": 8}}, "seq": 3, "restores": null } + ] +}