fix(fof): key-burn replay rejection + bounded sweep lock-hold

Two security/operational findings from the deeper pre-deploy audit:

1. Key-burn replay attack (security)
   - Before: replace_fof_slot did a blind UPDATE on receive. An
     attacker replaying an older signed FoFKeyBurn diff (Monday's)
     after a newer one (Friday's) would revert the slot.
   - Fix: new fof_key_burns table tracks (post_id, slot_index) ->
     max(burned_at_ms) seen. replace_fof_slot now refuses to apply
     a burn whose timestamp is <= the stored max. Atomic transaction
     ensures the gating-update + monotonic-record stay consistent.
   - apply_fof_key_burn_locally signature gains burned_at_ms; the
     three callers (connection.rs receive, node.rs author API, the
     fof.rs roundtrip test) all updated.

2. Unbounded sweep lock-hold (operational)
   - Before: sweep_unreadable_on_new_v_x walked up to 4096 queued
     posts per call, all under one storage lock. ~2s lock-held worst
     case. Attack: spammy bio posts trigger repeated sweeps.
   - Fix: MAX_SWEEP_PER_CALL = 256. Remaining entries processed on
     subsequent V_x arrivals. Bounds lock-hold to ~250ms worst case.

One new test (158 total):
- fof_key_burn_replay_rejected: applies Monday burn, then Friday
  burn, then replays Monday — asserts stored state stays at Friday's
  pub_x.

Bio-post replay was also evaluated: vouch grants are HPKE-sealed
(unforgeable) and stored INSERT OR IGNORE on (holder, owner, epoch)
so replay is a no-op. No fix needed.

Revocation + access-grant were also evaluated as idempotent by their
storage layers (INSERT OR IGNORE / dedup-on-pub_x). Safe.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/core/src/connection.rs

@@ -6302,6 +6302,7 @@ impl ConnectionManager {
                         }
                         let _ = crate::fof::apply_fof_key_burn_locally(
                             &storage, post_id, *slot_index, new_pub_x, new_wrap_slot,
+                            *burned_at_ms,
                         );
                     }
                     BlobHeaderDiffOp::ThreadSplit { new_post_id } => {

crates/core/src/fof.rs

@@ -278,18 +278,21 @@ pub fn find_unlock_for_post(
     Ok(None)
 }
 
-/// FoF Layer 5: when a persona acquires a new V_x, walk ALL
+/// Maximum posts processed per `sweep_unreadable_on_new_v_x` call.
-/// unreadable posts for the persona and re-attempt unlock. The new
+/// Bounds lock-hold time. Unprocessed entries remain queued and will
-/// V_x can unlock posts authored by anyone — the author may hold the
+/// be tried on the next V_x arrival or via a periodic background
-/// V_x's owner as one of their vouches, so the post's wrap_slots
+/// sweep (future Layer 5+ work).
-/// include a slot under the new V_x — even if the post's author is
+const MAX_SWEEP_PER_CALL: usize = 256;
-/// someone else entirely. Successful unlocks populate
+
+/// FoF Layer 5: when a persona acquires a new V_x, walk the
+/// unreadable-posts queue and re-attempt unlock. The new V_x can
+/// unlock posts authored by anyone (the author may hold the V_x's
+/// owner as one of their vouches). Successful unlocks populate
 /// `vouch_unlock_cache` + clear the queue entry as a side effect.
 ///
-/// Called by the receive path immediately after a new V_x lands in
+/// Bounded to `MAX_SWEEP_PER_CALL` posts per invocation to cap
-/// the persona's keyring. `_v_x_owner` is informational (no narrowing
+/// lock-hold time and prevent spam-grant-triggered DoS. Remaining
-/// happens here because the sweep can't predict which authors will
+/// entries are processed on subsequent V_x arrivals.
-/// have included this V_x in their gating).
 pub fn sweep_unreadable_on_new_v_x(
     storage: &Storage,
     holder_persona_id: &NodeId,
@@ -297,7 +300,7 @@ pub fn sweep_unreadable_on_new_v_x(
 ) -> Result<usize> {
     let post_ids = storage.list_all_unreadable_posts(holder_persona_id)?;
     let mut unlocked = 0usize;
-    for post_id in post_ids {
+    for post_id in post_ids.into_iter().take(MAX_SWEEP_PER_CALL) {
         let Some((post, _vis)) = storage.get_post_with_visibility(&post_id)? else {
             let _ = storage.clear_unreadable_post(holder_persona_id, &post_id);
             continue;
@@ -850,14 +853,18 @@ pub fn verify_fof_key_burn(
 /// + pub_x at the indicated slot in the stored post's fof_gating. The
 /// old key's holders can no longer decrypt this post via this slot
 /// (locally-cached plaintext on already-read devices is out of scope).
+///
+/// `burned_at_ms` enforces monotonic ordering: older signed key-burn
+/// diffs cannot revert newer state.
 pub fn apply_fof_key_burn_locally(
     storage: &Storage,
     post_id: &[u8; 32],
     slot_index: u32,
     new_pub_x: &[u8; 32],
     new_wrap_slot: &crate::types::WrapSlot,
+    burned_at_ms: u64,
 ) -> Result<bool> {
-    storage.replace_fof_slot(post_id, slot_index, new_pub_x, new_wrap_slot)
+    storage.replace_fof_slot(post_id, slot_index, new_pub_x, new_wrap_slot, burned_at_ms)
 }
 
 /// Apply a verified access-grant to local storage. Appends the new
@@ -1420,7 +1427,7 @@ mod tests {
         ));
 
         let applied = apply_fof_key_burn_locally(
-            &s, &post_id, alice_slot_idx as u32, &new_pub_x, &new_wrap_slot,
+            &s, &post_id, alice_slot_idx as u32, &new_pub_x, &new_wrap_slot, burned_at,
         ).unwrap();
         assert!(applied);
 
@@ -1550,6 +1557,88 @@ mod tests {
         assert!(!queued.contains(&overflow_pid));
     }
 
+    /// Key-burn replay rejection: applying an older signed burn after
+    /// a newer one must not revert state.
+    #[test]
+    fn fof_key_burn_replay_rejected() {
+        use crate::types::PostingIdentity;
+        use ed25519_dalek::SigningKey;
+
+        let s = temp_storage();
+        let (alice_id, alice_seed) = make_persona(220);
+        s.upsert_posting_identity(&PostingIdentity {
+            node_id: alice_id, secret_seed: alice_seed,
+            display_name: "Alice".into(), created_at: 1000,
+        }).unwrap();
+        let mut v_me_alice = [0u8; 32];
+        rand::rng().fill_bytes(&mut v_me_alice);
+        s.insert_own_vouch_key(&alice_id, 1, &v_me_alice, 1000).unwrap();
+
+        let built = build_fof_comment_gating(&s, &alice_id).unwrap().expect("built");
+        let post_id = [0xEE; 32];
+        let post = crate::types::Post {
+            author: alice_id, content: String::new(), attachments: vec![],
+            timestamp_ms: 3000, fof_gating: Some(built.gating.clone()),
+            supersedes_post_id: None,
+        };
+        s.store_post_with_intent(
+            &post_id, &post,
+            &crate::types::PostVisibility::Public,
+            &crate::types::VisibilityIntent::Public,
+        ).unwrap();
+
+        // Find Alice's slot.
+        let alice_slot_idx = (0..built.gating.wrap_slots.len()).find(|&i| {
+            crate::crypto::open_wrap_slot(
+                &v_me_alice, &built.slot_binder_nonce,
+                &built.gating.wrap_slots[i].read_ciphertext,
+                &built.gating.wrap_slots[i].sign_ciphertext,
+            ).is_some()
+        }).expect("alice slot exists") as u32;
+
+        // First burn (Monday): switch to V_me_new1.
+        let mut v_me_new1 = [0u8; 32]; rand::rng().fill_bytes(&mut v_me_new1);
+        let mut seed1 = [0u8; 32]; rand::rng().fill_bytes(&mut seed1);
+        let pub_x1 = SigningKey::from_bytes(&seed1).verifying_key().to_bytes();
+        let sealed1 = crate::crypto::seal_wrap_slot(
+            &v_me_new1, &built.slot_binder_nonce, &built.cek, &seed1,
+        ).unwrap();
+        let wrap1 = crate::types::WrapSlot {
+            prefilter_tag: sealed1.prefilter_tag,
+            read_ciphertext: sealed1.read_ciphertext,
+            sign_ciphertext: sealed1.sign_ciphertext,
+        };
+        let monday = 100_000;
+        apply_fof_key_burn_locally(&s, &post_id, alice_slot_idx, &pub_x1, &wrap1, monday).unwrap();
+
+        // Second burn (Friday, later timestamp): switch to V_me_new2.
+        let mut v_me_new2 = [0u8; 32]; rand::rng().fill_bytes(&mut v_me_new2);
+        let mut seed2 = [0u8; 32]; rand::rng().fill_bytes(&mut seed2);
+        let pub_x2 = SigningKey::from_bytes(&seed2).verifying_key().to_bytes();
+        let sealed2 = crate::crypto::seal_wrap_slot(
+            &v_me_new2, &built.slot_binder_nonce, &built.cek, &seed2,
+        ).unwrap();
+        let wrap2 = crate::types::WrapSlot {
+            prefilter_tag: sealed2.prefilter_tag,
+            read_ciphertext: sealed2.read_ciphertext,
+            sign_ciphertext: sealed2.sign_ciphertext,
+        };
+        let friday = 200_000;
+        apply_fof_key_burn_locally(&s, &post_id, alice_slot_idx, &pub_x2, &wrap2, friday).unwrap();
+
+        // Replay of Monday's burn (older timestamp) — must be rejected.
+        let applied = apply_fof_key_burn_locally(
+            &s, &post_id, alice_slot_idx, &pub_x1, &wrap1, monday,
+        ).unwrap();
+        assert!(!applied, "older burn must be rejected as replay");
+
+        // Stored state must still reflect Friday's burn.
+        let stored = s.get_post(&post_id).unwrap().unwrap();
+        let g = stored.fof_gating.as_ref().unwrap();
+        assert_eq!(g.pub_post_set[alice_slot_idx as usize], pub_x2,
+            "Friday's pub_x preserved despite replayed Monday burn");
+    }
+
     #[test]
     fn body_bucket_rule_boundaries() {
         // Sub-1KB.

crates/core/src/node.rs

@@ -5274,6 +5274,7 @@ impl Node {
             let storage = self.storage.get().await;
             let _ = crate::fof::apply_fof_key_burn_locally(
                 &*storage, &post_id, slot_index, &new_pub_x, &new_wrap_slot,
+                burned_at_ms,
             );
         }
 

crates/core/src/storage.rs

@@ -542,6 +542,18 @@ impl Storage {
                 cek                 BLOB NOT NULL,
                 slot_binder_nonce   BLOB NOT NULL,
                 PRIMARY KEY (author_persona_id, post_id)
+            );
+            -- FoF Layer 4 hardening: per-slot key-burn monotonic
+            -- timestamp. Receivers refuse to apply key-burns with an
+            -- older timestamp than the most recent already applied at
+            -- that slot — prevents replay of older signed diffs from
+            -- reverting newer state.
+            CREATE TABLE IF NOT EXISTS fof_key_burns (
+                post_id        BLOB NOT NULL,
+                slot_index     INTEGER NOT NULL,
+                burned_at_ms   INTEGER NOT NULL,
+                new_pub_x      BLOB NOT NULL,
+                PRIMARY KEY (post_id, slot_index)
             );",
         )?;
         Ok(())
@@ -5426,13 +5438,19 @@ impl Storage {
 
     /// FoF Layer 4: replace a wrap_slot + pub_x at `slot_index` in a
     /// stored post's fof_gating. Local-only mutation; PostId
-    /// (in the `id` column) is unaffected. Returns true on success.
+    /// (in the `id` column) is unaffected. Monotonic guard: refuses
+    /// to apply if a more-recent key-burn at this slot has already
+    /// been recorded — prevents an attacker from replaying an older
+    /// signed key-burn to revert a more recent one.
+    /// Returns true on success, false on bounds error / not-found /
+    /// stale-timestamp rejection.
     pub fn replace_fof_slot(
         &self,
         post_id: &PostId,
         slot_index: u32,
         new_pub_x: &[u8; 32],
         new_wrap_slot: &crate::types::WrapSlot,
+        burned_at_ms: u64,
     ) -> anyhow::Result<bool> {
         let Some(mut post) = self.get_post(post_id)? else { return Ok(false); };
         let Some(mut gating) = post.fof_gating.take() else { return Ok(false); };
@@ -5440,13 +5458,46 @@ impl Storage {
         if idx >= gating.pub_post_set.len() || idx >= gating.wrap_slots.len() {
             return Ok(false);
         }
+        // Monotonic guard: reject burns older than the most recent we've
+        // applied at this slot. Without this, an attacker replaying an
+        // older signed key-burn diff could revert a newer one.
+        let last_burn: Option<i64> = self.conn.query_row(
+            "SELECT MAX(burned_at_ms) FROM fof_key_burns
+             WHERE post_id = ?1 AND slot_index = ?2",
+            params![post_id.as_slice(), slot_index as i64],
+            |row| row.get(0),
+        ).ok();
+        if let Some(Some(last_ts)) = last_burn.map(|v| if v == 0 { None } else { Some(v) }) {
+            if (burned_at_ms as i64) <= last_ts {
+                return Ok(false);
+            }
+        }
         gating.pub_post_set[idx] = *new_pub_x;
         gating.wrap_slots[idx] = new_wrap_slot.clone();
         let gating_json = serde_json::to_string(&gating)?;
-        self.conn.execute(
+        let tx = self.conn.unchecked_transaction()?;
+        tx.execute(
             "UPDATE posts SET fof_gating_json = ?1 WHERE id = ?2",
             params![gating_json, post_id.as_slice()],
         )?;
+        // Record the burn for monotonicity on the next attempt.
+        tx.execute(
+            "INSERT INTO fof_key_burns
+                (post_id, slot_index, burned_at_ms, new_pub_x)
+             VALUES (?1, ?2, ?3, ?4)
+             ON CONFLICT(post_id, slot_index) DO UPDATE SET
+                burned_at_ms = MAX(burned_at_ms, excluded.burned_at_ms),
+                new_pub_x = CASE
+                    WHEN excluded.burned_at_ms > burned_at_ms
+                    THEN excluded.new_pub_x ELSE new_pub_x END",
+            params![
+                post_id.as_slice(),
+                slot_index as i64,
+                burned_at_ms as i64,
+                new_pub_x.as_slice(),
+            ],
+        )?;
+        tx.commit()?;
         Ok(true)
     }