kv-sync: refuse per-item replays (seq/restores in the item plaintext)

A malicious server could re-serve an old but valid envelope of an item under
a new, higher revision: the AAD does not bind the revision and the Merkle
root was only checked when head_revision == cursor. The client took it as
the latest version, rolled the item back and signed over it in its next
manifest.

* kv-model Item: `seq: u64` (serde default 0) and `restores: Option<u64>`.
* kv-sync replay module: pure decision rules (check_successor,
  next_version, stamp, same_content). Every pulled live record must
  authenticate and have seq > the base's seq, unless it carries
  restores = t with base_revision < t < revision (edit beat a delete this
  device never saw) or is identical to the base. Otherwise: Integrity alarm.
  Tombstones and items without a base pass.
* engine: check on every pulled record; edit-beats-delete re-seals with
  restores = tombstone revision; merges write seq = max + 1 and adopt the
  remote version when only seq/restores differ. Public
  engine::next_local_version gives the seq/restores for a local save
  (used by kv-ffi save_item and the tests).
* verify_manifest checks the Merkle root whenever no held item is newer
  than head_revision (previously only at head == cursor).
* Tests: replay alarm (clean and pending receivers, plus kv-ffi), harmless
  replay of the current version, tampered envelope, concurrent equal-seq
  edits across three devices, edit-beats-delete with a device that skipped
  the tombstone and a later replay of the restore, legacy items without seq,
  hidden item with cursor past the manifest head, and the shared
  item_replay.json decision table.
* gen-vectors: envelope vector item payload now includes seq.

FFI surface unchanged.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

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<HistoryEntry> {
     })
 }
 
+/// 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<Self, SyncError> {
+        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<u64>), 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<Vec<ItemRecord>, 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 {

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;

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<String>,
 }
 
-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 {

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<u64> {
+    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<u64>) {
+    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<u64>) {
+    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)));
+    }
+}