rules: a carried remake record taken from the coordinator its snapshot names

Review finding 7, second part: a record carried into a new epoch whole
(proposal and decision in one change) was taken from any member, for
any remake ID. A member could write a forged record, decision included,
for a fresh ID in a single change.

The snapshot's epoch metadata now names the remake's coordinator, as
the quorum built it. The carried-record exception holds only in a
later epoch, only for the remake that made it, and only from the
coordinator its snapshot names, where there is no record yet. Wire
version 43: the snapshot, and so the epoch ID, changes.

The review suggested carrying the record inside the snapshot instead.
The snapshot can't hold the decision, which names the snapshot's own
hash. So the snapshot vouches for who may carry the record instead.

6 files changed, 86 insertions, 24 deletions

diff --git a/src/access/rules.rs b/src/access/rules.rs
index 2746cb3..cf2740a 100644
--- a/src/access/rules.rs
+++ b/src/access/rules.rs
@@ -194,8 +194,9 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result
             // A node forgotten while alive cancels its own record (`swarm::peer::forget`).
             (Some(meta::FORGOTTEN), Some(owner)) => owner == node || user_may(path),
             (Some(meta::REMAKES), Some(id)) => match path.get(3).map(String::as_str) {
-                // Its coordinator's; or a record carried into a new epoch whole, by the
-                // node that coordinated it (`remake::switch`): its proposal written with it.
+                // Its coordinator's; or the record of the remake that made the epoch,
+                // carried into it whole (its proposal written with it) by the coordinator
+                // its snapshot names (`remake::switch`), where there is none.
                 Some("decision") => match coordinator(&at, id) {
                     Some(coordinator) => coordinator == node,
                     None => {
@@ -203,10 +204,11 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result
                             .paths
                             .iter()
                             .any(|other| other == &meta::path(&[meta::REMAKES, id, "proposal"]))
-                            || change
-                                .paths
-                                .iter()
-                                .any(|other| other == &meta::path(&[meta::REMAKES, id]))
+                            && store::read_path(&at, &meta::path(&[meta::REMAKES, id]))
+                                .ok()
+                                .flatten()
+                                .is_none()
+                            && made_the_epoch(&at, epoch, id, node)
                     }
                 },
                 // The record, made where there is none, and its proposal, written where
@@ -410,6 +412,26 @@ fn shown(path: &Path) -> String {
     format!("/{}", path.join("/"))
 }
 
+/// Whether remake `id`, coordinated by `node`, made `epoch`: as its snapshot's metadata
+/// says (`remake::snapshot`), at `at`. Never genesis.
+fn made_the_epoch(at: &At<'_>, epoch: Epoch, id: &str, node: &str) -> bool {
+    if epoch.generation == 0 {
+        return false;
+    }
+    let generation = epoch.generation.to_string();
+    let Some(metadata) = store::read_path(at, &meta::path(&[meta::EPOCHS, &generation]))
+        .ok()
+        .flatten()
+    else {
+        return false;
+    };
+    metadata.get("remakeId").and_then(serde_json::Value::as_str) == Some(id)
+        && metadata
+            .get("coordinator")
+            .and_then(serde_json::Value::as_str)
+            == Some(node)
+}
+
 /// The coordinator remake `id`'s proposal names, at `at`.
 fn coordinator(at: &At<'_>, id: &str) -> Option<String> {
     let proposal = store::read_path(at, &meta::path(&[meta::REMAKES, id, "proposal"]))
@@ -544,24 +566,55 @@ mod tests {
         assert!(refused(judged(&doc, "b", None, &["_continuum/members"])));
         assert!(refused(judged(&doc, "b", None, &["_continuum/remakes/r1"])));
         assert!(judged(&doc, "b", None, &["_continuum/votes"]).is_ok());
-        // A record carried into a new epoch whole: its proposal with its decision.
-        assert!(
-            judged(
-                &doc,
-                "b",
-                None,
-                &[
-                    "_continuum/remakes/r2",
-                    "_continuum/remakes/r2/proposal",
-                    "_continuum/remakes/r2/decision"
-                ]
-            )
-            .is_ok()
+        // A record carried into a new epoch whole, its proposal with its decision: only
+        // the record of the remake that made the epoch, by its coordinator, where there
+        // is none.
+        let whole = [
+            "_continuum/remakes/r2",
+            "_continuum/remakes/r2/proposal",
+            "_continuum/remakes/r2/decision",
+        ];
+        assert!(refused(judged(&doc, "b", None, &whole)));
+        let mut later = doc.fork();
+        write(
+            &mut later,
+            "_continuum/epochs/1",
+            json!({"generation": 1, "remakeId": "r2", "coordinator": "b"}),
         );
-        assert!(refused(judged(
-            &doc,
+        let in_epoch_one = |doc: &Document, committer: &str, touched: &[&str]| {
+            let deps = doc.get_heads();
+            let touched = paths(touched);
+            judge(
+                doc,
+                Epoch {
+                    generation: 1,
+                    id: [7; 32],
+                },
+                &Change {
+                    hash: ChangeHash([9; 32]),
+                    deps: &deps,
+                    seen: &deps,
+                    committer: Some(committer),
+                    user: None,
+                    paths: &touched,
+                    ops: &[],
+                },
+            )
+        };
+        assert!(in_epoch_one(&later, "b", &whole).is_ok());
+        assert!(refused(in_epoch_one(&later, "a", &whole)));
+        assert!(refused(in_epoch_one(
+            &later,
+            "b",
+            &[
+                "_continuum/remakes/r3",
+                "_continuum/remakes/r3/proposal",
+                "_continuum/remakes/r3/decision"
+            ]
+        )));
+        assert!(refused(in_epoch_one(
+            &later,
             "b",
-            None,
             &["_continuum/remakes/r2/decision"]
         )));
         // The rest: a user's who may write there.
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 9830c33..4df5c90 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -718,6 +718,7 @@ mod tests {
             &mut old,
             &SnapshotParams {
                 remake_id: "c-1",
+                coordinator: "a",
                 previous: Epoch::GENESIS,
                 base_heads: &base,
                 policy,
diff --git a/src/remake/records.rs b/src/remake/records.rs
index 21b383c..21cf0ae 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -306,6 +306,7 @@ impl Record {
     ) -> Option<SnapshotRequest> {
         Some(SnapshotRequest {
             remake_id: id.to_owned(),
+            coordinator: self.proposal.coordinator.clone(),
             previous: self.proposal.from_epoch.to_epoch()?,
             base_heads: base,
             policy: serde_json::to_value(&self.proposal.policy).ok()?,
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index 220989e..7938b53 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -36,6 +36,8 @@ use crate::document::store::{self, StoreError};
 pub(crate) struct SnapshotParams<'a> {
     /// ID of the committed remake; seeds the snapshot actor.
     pub(crate) remake_id: &'a str,
+    /// The node coordinating it: the one that carries its record into the new epoch.
+    pub(crate) coordinator: &'a str,
     /// The epoch being compacted. The snapshot starts generation `previous + 1`.
     pub(crate) previous: Epoch,
     /// Heads of the base. Every node must hold all of these changes.
@@ -84,6 +86,7 @@ pub(crate) enum SnapshotError {
 #[derive(Clone, Debug)]
 pub(crate) struct SnapshotRequest {
     pub(crate) remake_id: String,
+    pub(crate) coordinator: String,
     pub(crate) previous: Epoch,
     pub(crate) base_heads: Vec<ChangeHash>,
     pub(crate) policy: JsonValue,
@@ -97,6 +100,7 @@ impl SnapshotRequest {
     pub(crate) fn params(&self) -> SnapshotParams<'_> {
         SnapshotParams {
             remake_id: &self.remake_id,
+            coordinator: &self.coordinator,
             previous: self.previous,
             base_heads: &self.base_heads,
             policy: &self.policy,
@@ -311,6 +315,7 @@ pub(crate) fn build(
     let metadata = canonical(json!({
         "generation": generation,
         "remakeId": params.remake_id,
+        "coordinator": params.coordinator,
         "previous": {
             "generation": params.previous.generation,
             "id": params.previous.id_hex(),
@@ -391,6 +396,7 @@ mod tests {
     fn params<'a>(heads: &'a [ChangeHash], policy: &'a JsonValue) -> SnapshotParams<'a> {
         SnapshotParams {
             remake_id: "c-1",
+            coordinator: "a",
             previous: Epoch::GENESIS,
             base_heads: heads,
             policy,
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 34c492f..bd40d72 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -200,7 +200,8 @@ async fn switch<H: Host>(
     // Only the coordinator carries the record into the new epoch: one writer, so the
     // nodes don't all write the same path concurrently right after switching (a merge
     // conflict, plus one change per node to sync). The epoch metadata in the snapshot
-    // keeps the essentials on every node anyway.
+    // keeps the essentials on every node anyway, and names the coordinator: the rules
+    // take the record from it only (`access::rules`).
     let carry_record = record.proposal.coordinator == host.node_id();
     let record_path = records::record_path(id, "");
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 7ff7835..e7bec62 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 42;
+pub(crate) const VERSION: u8 = 43;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {