feat(replication): v12 storage-bound audit — design-complete

src/node.rs

@@ -150,6 +150,7 @@ impl NodeBuilder {
                     Arc::clone(&p2p_arc),
                     storage_arc,
                     payment_verifier_arc,
+                    Arc::clone(&identity),
                     &self.config.root_dir,
                     fresh_rx,
                     shutdown.clone(),

src/replication/neighbor_sync.rs

@@ -182,18 +182,31 @@ pub async fn sync_with_peer(
     config: &ReplicationConfig,
     is_bootstrapping: bool,
 ) -> Option<NeighborSyncResponse> {
-    sync_with_peer_with_outcome(peer, p2p_node, storage, paid_list, config, is_bootstrapping)
-        .await
-        .map(|outcome| outcome.response)
+    sync_with_peer_with_outcome(
+        peer,
+        p2p_node,
+        storage,
+        paid_list,
+        config,
+        is_bootstrapping,
+        None,
+    )
+    .await
+    .map(|outcome| outcome.response)
 }
 
+/// `commitment`: sender's current commitment to piggyback on the request.
+/// `None` if the responder hasn't rotated one yet (e.g. fresh boot,
+/// empty storage) — receiver falls back to legacy path.
+#[allow(clippy::too_many_arguments)]
 pub(crate) async fn sync_with_peer_with_outcome(
     peer: &PeerId,
     p2p_node: &Arc<P2PNode>,
     storage: &Arc<LmdbStorage>,
     paid_list: &Arc<PaidList>,
     config: &ReplicationConfig,
     is_bootstrapping: bool,
+    commitment: Option<crate::replication::commitment::StorageCommitment>,
 ) -> Option<NeighborSyncOutcome> {
     // Build peer-targeted hint sets (Rule 7).
     let sent_replica_hints = build_replica_hints_for_peer_with_close_groups(
@@ -215,6 +228,7 @@ pub(crate) async fn sync_with_peer_with_outcome(
         replica_hints,
         paid_hints,
         bootstrapping: is_bootstrapping,
+        commitment,
     };
     let request_id = rand::thread_rng().gen::<u64>();
     let msg = ReplicationMessage {
@@ -335,11 +349,13 @@ pub async fn handle_sync_request(
         paid_list,
         config,
         is_bootstrapping,
+        None,
     )
     .await;
     (response, sender_in_rt)
 }
 
+#[allow(clippy::too_many_arguments)]
 pub(crate) async fn handle_sync_request_with_proofs(
     sender: &PeerId,
     _request: &NeighborSyncRequest,
@@ -348,6 +364,7 @@ pub(crate) async fn handle_sync_request_with_proofs(
     paid_list: &Arc<PaidList>,
     config: &ReplicationConfig,
     is_bootstrapping: bool,
+    my_commitment: Option<crate::replication::commitment::StorageCommitment>,
 ) -> (NeighborSyncResponse, Vec<SentReplicaHint>, bool) {
     let sender_in_rt = p2p_node.dht_manager().is_in_routing_table(sender).await;
 
@@ -376,6 +393,7 @@ pub(crate) async fn handle_sync_request_with_proofs(
         paid_hints,
         bootstrapping: is_bootstrapping,
         rejected_keys: Vec::new(),
+        commitment: my_commitment,
     };
 
     // Rule 4-6: accept inbound hints only if sender is in LocalRT.
@@ -977,6 +995,7 @@ mod tests {
             paid_hints: outbound_paid_hints.clone(),
             bootstrapping: false,
             rejected_keys: Vec::new(),
+            commitment: None,
         };
 
         // Inbound hints from the sender (would be in the request).

src/replication/protocol.rs

@@ -177,6 +177,14 @@ pub struct NeighborSyncRequest {
     pub paid_hints: Vec<XorName>,
     /// Whether sender is currently bootstrapping.
     pub bootstrapping: bool,
+    /// Sender's signed storage commitment (optional, see
+    /// [`crate::replication::commitment`]). `None` from old peers; from
+    /// new peers this carries the Merkle-root commitment over the
+    /// sender's claimed keys. Receivers that recognize it store it as
+    /// the per-peer "last known commitment" used to pin commitment-bound
+    /// audits.
+    #[serde(default)]
+    pub commitment: Option<crate::replication::commitment::StorageCommitment>,
 }
 
 /// Neighbor sync response carrying own hint sets.
@@ -190,6 +198,10 @@ pub struct NeighborSyncResponse {
     pub bootstrapping: bool,
     /// Keys that receiver rejected (optional feedback to sender).
     pub rejected_keys: Vec<XorName>,
+    /// Receiver's signed storage commitment (optional, see
+    /// [`NeighborSyncRequest::commitment`]).
+    #[serde(default)]
+    pub commitment: Option<crate::replication::commitment::StorageCommitment>,
 }
 
 // ---------------------------------------------------------------------------
@@ -286,6 +298,20 @@ pub struct AuditChallenge {
     pub challenged_peer_id: [u8; 32],
     /// Ordered list of keys to prove storage of.
     pub keys: Vec<XorName>,
+    /// Auditor's pin to the commitment it expects the responder to use.
+    ///
+    /// `Some(h)`: a commitment-bound audit (v12 design). The responder
+    /// must reply with `AuditResponse::CommitmentBound` whose
+    /// commitment hashes via
+    /// [`crate::replication::commitment::commitment_hash`] to exactly
+    /// `h`. Any other commitment, or a plain `Digests` reply, is an
+    /// audit failure.
+    ///
+    /// `None`: legacy plain-digest audit (today's behaviour). Allows
+    /// challenging peers from whom we haven't yet received a commitment
+    /// without breaking the existing audit flow during rollout.
+    #[serde(default)]
+    pub expected_commitment_hash: Option<[u8; 32]>,
 }
 
 /// Response to audit challenge.
@@ -316,6 +342,25 @@ pub enum AuditResponse {
         /// Human-readable rejection reason.
         reason: String,
     },
+    /// Commitment-bound proof of storage (v12 storage-bound audit).
+    ///
+    /// Returned when the challenge carried an
+    /// [`AuditChallenge::expected_commitment_hash`]. Carries the
+    /// responder's signed commitment plus per-key Merkle inclusion
+    /// proofs. The auditor verifies that:
+    ///   1. `commitment_hash(commitment) == challenge.expected_commitment_hash`
+    ///   2. The commitment's signature is valid.
+    ///   3. For each per-key entry: the Merkle path verifies the leaf
+    ///      against the commitment root AND the digest matches the
+    ///      auditor's local copy of the bytes.
+    CommitmentBound {
+        /// The challenge this response answers.
+        challenge_id: u64,
+        /// The signed commitment whose root the proofs are against.
+        commitment: crate::replication::commitment::StorageCommitment,
+        /// Per-key Merkle inclusion proofs, in challenge order.
+        per_key: Vec<crate::replication::commitment::CommitmentBoundResult>,
+    },
 }
 
 // ---------------------------------------------------------------------------
@@ -490,6 +535,141 @@ mod tests {
 
     // === Neighbor Sync roundtrips ===
 
+    // -- backwards compat across the wire-type extension --------------------
+
+    /// Backwards-compat: an old peer that has the v0 layout of
+    /// `NeighborSyncRequest` (no `commitment` field) can still decode a
+    /// message encoded by a new peer that emits `commitment: None`. This
+    /// is the realistic mixed-version case during rollout: new peers
+    /// gossip with the field; old peers must not crash.
+    ///
+    /// The check works because postcard's [`from_bytes`] is lenient on
+    /// trailing bytes — the old decoder reads what it knows about and
+    /// stops, the new fields are silently ignored. This test pins that
+    /// invariant so any future codec/library swap that breaks it is
+    /// caught immediately.
+    #[test]
+    fn old_decoder_tolerates_new_neighbor_sync_request() {
+        use serde::Deserialize;
+        #[derive(Deserialize)]
+        struct OldNeighborSyncRequest {
+            #[allow(dead_code)]
+            pub replica_hints: Vec<XorName>,
+            #[allow(dead_code)]
+            pub paid_hints: Vec<XorName>,
+            #[allow(dead_code)]
+            pub bootstrapping: bool,
+        }
+
+        let new_req = NeighborSyncRequest {
+            replica_hints: vec![[0x01; 32], [0x02; 32]],
+            paid_hints: vec![[0x03; 32]],
+            bootstrapping: true,
+            commitment: None,
+        };
+        let encoded = postcard::to_stdvec(&new_req).expect("encode");
+        let old_decoded: OldNeighborSyncRequest =
+            postcard::from_bytes(&encoded).expect("old decoder accepts");
+        // Field-by-field check would fail if old peer misaligned on the
+        // length prefix — passing decode is the structural check.
+        assert_eq!(old_decoded.replica_hints.len(), 2);
+        assert_eq!(old_decoded.paid_hints.len(), 1);
+        assert!(old_decoded.bootstrapping);
+    }
+
+    /// Same property for `NeighborSyncResponse`.
+    #[test]
+    fn old_decoder_tolerates_new_neighbor_sync_response() {
+        use serde::Deserialize;
+        #[derive(Deserialize)]
+        struct OldNeighborSyncResponse {
+            #[allow(dead_code)]
+            pub replica_hints: Vec<XorName>,
+            #[allow(dead_code)]
+            pub paid_hints: Vec<XorName>,
+            #[allow(dead_code)]
+            pub bootstrapping: bool,
+            #[allow(dead_code)]
+            pub rejected_keys: Vec<XorName>,
+        }
+
+        let new_resp = NeighborSyncResponse {
+            replica_hints: vec![[0x04; 32]],
+            paid_hints: vec![],
+            bootstrapping: false,
+            rejected_keys: vec![[0x05; 32]],
+            commitment: None,
+        };
+        let encoded = postcard::to_stdvec(&new_resp).expect("encode");
+        let old_decoded: OldNeighborSyncResponse =
+            postcard::from_bytes(&encoded).expect("old decoder accepts");
+        assert_eq!(old_decoded.replica_hints.len(), 1);
+        assert_eq!(old_decoded.rejected_keys.len(), 1);
+    }
+
+    /// `AuditChallenge` extension: old peer (no `expected_commitment_hash`
+    /// field) decodes a new-peer message OK.
+    #[test]
+    fn old_decoder_tolerates_new_audit_challenge() {
+        use serde::Deserialize;
+        #[derive(Deserialize)]
+        struct OldAuditChallenge {
+            #[allow(dead_code)]
+            pub challenge_id: u64,
+            #[allow(dead_code)]
+            pub nonce: [u8; 32],
+            #[allow(dead_code)]
+            pub challenged_peer_id: [u8; 32],
+            #[allow(dead_code)]
+            pub keys: Vec<XorName>,
+        }
+
+        let new_ch = AuditChallenge {
+            challenge_id: 7,
+            nonce: [0xAA; 32],
+            challenged_peer_id: [0xBB; 32],
+            keys: vec![[0x01; 32], [0x02; 32]],
+            expected_commitment_hash: None,
+        };
+        let encoded = postcard::to_stdvec(&new_ch).expect("encode");
+        let old_decoded: OldAuditChallenge =
+            postcard::from_bytes(&encoded).expect("old decoder accepts");
+        assert_eq!(old_decoded.challenge_id, 7);
+        assert_eq!(old_decoded.keys.len(), 2);
+    }
+
+    /// Roundtrip: a new peer can decode its own message including the
+    /// commitment field. Catches accidental serde annotation breakage
+    /// (e.g. forgetting `#[serde(default)]` on the new field).
+    #[test]
+    fn new_peer_roundtrips_with_commitment_some() {
+        use crate::replication::commitment::{sign_commitment, StorageCommitment};
+        use saorsa_pqc::api::sig::ml_dsa_65;
+
+        let (pk, sk) = ml_dsa_65().generate_keypair().expect("keygen");
+        let root = [0x7Fu8; 32];
+        let sender = [0xCCu8; 32];
+        let pk_bytes = pk.to_bytes();
+        let sig = sign_commitment(&sk, &root, 3, &sender, &pk_bytes).expect("sign");
+        let commitment = StorageCommitment {
+            root,
+            key_count: 3,
+            sender_peer_id: sender,
+            sender_public_key: pk_bytes,
+            signature: sig,
+        };
+
+        let req = NeighborSyncRequest {
+            replica_hints: vec![[0x01; 32]],
+            paid_hints: vec![],
+            bootstrapping: false,
+            commitment: Some(commitment.clone()),
+        };
+        let encoded = postcard::to_stdvec(&req).expect("encode");
+        let decoded: NeighborSyncRequest = postcard::from_bytes(&encoded).expect("new decoder");
+        assert_eq!(decoded.commitment, Some(commitment));
+    }
+
     #[test]
     fn neighbor_sync_request_roundtrip() {
         let msg = ReplicationMessage {
@@ -498,6 +678,7 @@ mod tests {
                 replica_hints: vec![[0x01; 32], [0x02; 32]],
                 paid_hints: vec![[0x03; 32]],
                 bootstrapping: true,
+                commitment: None,
             }),
         };
         let encoded = msg.encode().expect("encode should succeed");
@@ -522,6 +703,7 @@ mod tests {
                 paid_hints: vec![],
                 bootstrapping: false,
                 rejected_keys: vec![[0x05; 32], [0x06; 32]],
+                commitment: None,
             }),
         };
         let encoded = msg.encode().expect("encode should succeed");
@@ -697,6 +879,7 @@ mod tests {
                 nonce: [0xAB; 32],
                 challenged_peer_id: [0xCD; 32],
                 keys: vec![[0x01; 32], [0x02; 32]],
+                expected_commitment_hash: None,
             }),
         };
         let encoded = msg.encode().expect("encode should succeed");

src/replication/pruning.rs

@@ -710,6 +710,11 @@ fn encode_prune_audit_challenge(
         nonce,
         challenged_peer_id: *peer.as_bytes(),
         keys: vec![key],
+        // Prune-audit challenges keep legacy plain-digest semantics
+        // (caller does its own per-key digest comparison). Commitment-
+        // bound prune audits are out of scope for phase 2; revisit in
+        // phase 3 if we choose to extend coverage there.
+        expected_commitment_hash: None,
     };
     let msg = ReplicationMessage {
         request_id: challenge_id,