access: signatures required

"signatures": "required": a change counts only once a signature by its
committer vouches for it, and waits until then; every node then signs
every change it makes.

Auth plan M6.

8 files changed, 279 insertions, 22 deletions

diff --git a/src/access/policy.rs b/src/access/policy.rs
index 7ddef0e..f8ccfc2 100644
--- a/src/access/policy.rs
+++ b/src/access/policy.rs
@@ -4,7 +4,9 @@
 //! by a user who may write protocol data, as any of it.
 //!
 //! `{"version": 1, "signatures": "optional"}`: rules enforced; a change is its claimed
-//! committer's whether or not a signature proves it (`document::signatures`). A policy
+//! committer's whether or not a signature proves it (`document::signatures`).
+//! `"required"`: a change is applied once a signature by its committer vouches for it,
+//! and waits until then; every node signs every change it makes. A policy
 //! this node doesn't know (another version, or a setting it lacks) stops it from taking
 //! writes through the API, and it receives without judging: it can't judge as the
 //! others do, so it writes nothing they might refuse, until it is upgraded.
@@ -26,15 +28,21 @@ pub(crate) fn path() -> Vec<String> {
 pub(crate) enum Policy {
     /// None: changes aren't judged.
     Open,
-    /// Rules by path enforced.
-    Enforced,
+    /// Rules by path enforced; with signatures `required`, only on changes a signature
+    /// proves.
+    Enforced { required: bool },
     /// One this node doesn't know, and why.
     Unknown(String),
 }
 
 impl Policy {
     pub(crate) fn enforced(&self) -> bool {
-        matches!(self, Policy::Enforced)
+        matches!(self, Policy::Enforced { .. })
+    }
+
+    /// Whether changes need a signature (`signatures: "required"`).
+    pub(crate) fn required(&self) -> bool {
+        matches!(self, Policy::Enforced { required: true })
     }
 }
 
@@ -55,7 +63,8 @@ pub(crate) fn of(doc: &impl ReadDoc) -> Policy {
         ));
     }
     match value.get("signatures").and_then(JsonValue::as_str) {
-        None | Some("optional") => Policy::Enforced,
+        None | Some("optional") => Policy::Enforced { required: false },
+        Some("required") => Policy::Enforced { required: true },
         Some(other) => Policy::Unknown(format!("signatures {other:?}")),
     }
 }
@@ -78,10 +87,15 @@ mod tests {
     #[test]
     fn what_the_policy_says() {
         assert_eq!(of(&Document::new()), Policy::Open);
-        assert_eq!(with(json!({"version": 1})), Policy::Enforced);
+        let optional = Policy::Enforced { required: false };
+        assert_eq!(with(json!({"version": 1})), optional);
         assert_eq!(
             with(json!({"version": 1, "signatures": "optional"})),
-            Policy::Enforced
+            optional
+        );
+        assert_eq!(
+            with(json!({"version": 1, "signatures": "required"})),
+            Policy::Enforced { required: true }
         );
         assert!(matches!(with(json!({"version": 2})), Policy::Unknown(_)));
         assert!(matches!(
diff --git a/src/access/screen.rs b/src/access/screen.rs
index 66b7fc8..507de1d 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -10,6 +10,9 @@
 //!
 //! A change whose dependencies are neither in the document nor among those received
 //! waits (`Waiting`) until they come, a bounded number; Automerge never sees it before.
+//! With signatures required (`access::policy`), so does one no signature by its
+//! committer vouches for yet (`Signatures::vouch_batch`), and whatever builds on it,
+//! until one comes, but an epoch's root, which no node signs.
 
 use std::collections::{BTreeSet, HashMap, HashSet};
 
@@ -19,6 +22,7 @@ use super::rules::{self, Refusal};
 use crate::document::inspect::{Inspector, Path};
 use crate::document::meta;
 use crate::document::replica::{Epoch, user_of};
+use crate::document::signatures::Signatures;
 use crate::document::{Change, ChangeHash, Document, ReadDoc};
 
 /// Changes waiting for their dependencies, kept at most.
@@ -102,11 +106,13 @@ pub(crate) struct Screened {
 /// Judges `received` (and those `waiting` before) and applies to `doc` (of `epoch`)
 /// those rules allow, what they change recorded in `log`. Those already in `doc` are
 /// left out; those whose dependencies are missing wait.
+/// With `signatures` (required), those no signature vouches for wait too.
 pub(crate) fn apply(
     doc: &mut Document,
     epoch: Epoch,
     received: Vec<Change>,
     waiting: &mut Waiting,
+    mut signatures: Option<&mut Signatures>,
     log: &mut crate::document::PatchLog,
 ) -> Result<Screened> {
     let mut screened = Screened::default();
@@ -120,8 +126,8 @@ pub(crate) fn apply(
         .collect();
     let (ready, unready) = in_order(doc, candidates);
     waiting.0 = unready;
-    waiting.0.truncate(MAX_WAITING);
     if ready.is_empty() {
+        waiting.0.truncate(MAX_WAITING);
         return Ok(screened);
     }
 
@@ -150,10 +156,14 @@ pub(crate) fn apply(
             named
         })
         .collect();
+    if let Some(signatures) = signatures.as_deref_mut() {
+        signatures.vouch_batch(&ready, &committers);
+    }
 
     // The changes of the document each accepted one, not applied yet, builds on.
     let mut frontier: HashMap<ChangeHash, Vec<ChangeHash>> = HashMap::new();
     let mut refused: HashSet<ChangeHash> = HashSet::new();
+    let mut unsigned: HashSet<ChangeHash> = HashSet::new();
     let mut accepted: Vec<Change> = Vec::new();
     for (index, change) in ready.iter().enumerate() {
         let hash = change.hash();
@@ -173,6 +183,19 @@ pub(crate) fn apply(
                 .push(refuse(format!("it builds on refused change {dep}")));
             continue;
         }
+        // No signature vouches for it yet, or for what it builds on: it waits.
+        let vouched = |signatures: &Signatures| {
+            change.deps().is_empty() || signatures.verified(change.actor_id(), change.seq())
+        };
+        if signatures
+            .as_deref()
+            .is_some_and(|signatures| !vouched(signatures))
+            || change.deps().iter().any(|dep| unsigned.contains(dep))
+        {
+            unsigned.insert(hash);
+            waiting.0.push(change.clone());
+            continue;
+        }
         let mut seen_at: BTreeSet<ChangeHash> = BTreeSet::new();
         for dep in change.deps() {
             match frontier.get(dep) {
@@ -222,6 +245,7 @@ pub(crate) fn apply(
     if !accepted.is_empty() {
         doc.apply_changes_log_patches(accepted, log)?;
     }
+    waiting.0.truncate(MAX_WAITING);
     Ok(screened)
 }
 
@@ -229,16 +253,19 @@ pub(crate) fn apply(
 /// what `screening` keeps for document `epoch_seq` (of `epoch`). Bytes that don't load
 /// as changes, or with a misplaced author (`changes::misplaced_author`), are refused
 /// whole.
+/// With `signatures` (required), those no signature vouches for wait too; and with no
+/// bytes, those waiting are judged again (signatures came since).
 pub(crate) fn apply_bytes(
     doc: &mut Document,
     epoch: Epoch,
     epoch_seq: u32,
     bytes: &[u8],
     screening: &std::sync::Mutex<Screening>,
+    signatures: Option<&mut Signatures>,
     log: &mut crate::document::PatchLog,
 ) -> Result<usize> {
     use crate::locks::MutexExt;
-    if bytes.is_empty() {
+    if bytes.is_empty() && signatures.is_none() {
         return Ok(0);
     }
     if let Some(misplaced) = crate::document::changes::misplaced_author(bytes) {
@@ -248,7 +275,10 @@ pub(crate) fn apply_bytes(
         .ok_or_else(|| anyhow::anyhow!("Refused changes that don't load"))?;
     let mut screening = screening.lock_or_recover();
     let screening = screening.of(epoch_seq);
-    let screened = apply(doc, epoch, changes, &mut screening.waiting, log)?;
+    if changes.is_empty() && screening.waiting.len() == 0 {
+        return Ok(0);
+    }
+    let screened = apply(doc, epoch, changes, &mut screening.waiting, signatures, log)?;
     let applied = screened.applied;
     screening.record(screened.refused);
     Ok(applied)
@@ -339,6 +369,7 @@ mod tests {
             Epoch::GENESIS,
             changes,
             waiting,
+            None,
             &mut crate::document::PatchLog::inactive(),
         )
         .unwrap()
@@ -450,4 +481,89 @@ mod tests {
             Some(json!(2))
         );
     }
+
+    /// With signatures required, changes no signature vouches for wait, with what
+    /// builds on them, until one comes; one signature vouches for its actor's earlier
+    /// changes too.
+    #[test]
+    fn unsigned_changes_wait_for_a_signature() {
+        let key = crate::identity::NodeKey::from_seed([3; 32]);
+        let node = key.id();
+        let mut theirs = Document::new();
+        let alice = registry::entry(&node, 1, &[(path("notes"), &[Right::Write][..])]);
+        transact(
+            &mut theirs,
+            UnixMillis(1),
+            &Writer {
+                node: &node,
+                user: None,
+            },
+            |doc| {
+                store::write_path(doc, &path("_continuum/swarm"), &json!("s"))?;
+                store::write_path(doc, &path("_continuum/users/alice"), &alice)
+            },
+        )
+        .unwrap();
+        let root = theirs.get_heads();
+        write(
+            &mut theirs,
+            &node,
+            None,
+            &format!("_continuum/members/{node}"),
+            json!({}),
+        );
+        write(&mut theirs, &node, Some("alice"), "notes/x", json!(1));
+        let last = theirs.get_heads()[0];
+
+        let mut ours = Document::new();
+        let mut waiting = Waiting::default();
+        let mut signatures = Signatures::default();
+        let required =
+            |doc: &mut Document, waiting: &mut Waiting, signatures: &mut Signatures, changes| {
+                apply(
+                    doc,
+                    Epoch::GENESIS,
+                    changes,
+                    waiting,
+                    Some(signatures),
+                    &mut crate::document::PatchLog::inactive(),
+                )
+                .unwrap()
+            };
+        // Unsigned: only the root goes in (no node signs it).
+        let screened = required(
+            &mut ours,
+            &mut waiting,
+            &mut signatures,
+            received(&theirs, &[]),
+        );
+        assert_eq!(screened.applied, 1);
+        assert_eq!(ours.get_heads(), root);
+        assert_eq!(waiting.len(), 2);
+        // A forged signature vouches for nothing.
+        let forged = crate::identity::NodeKey::from_seed([4; 32]);
+        signatures.offer(
+            &ours,
+            vec![(
+                last,
+                forged.sign(&crate::document::signatures::message(&last)),
+            )],
+        );
+        let screened = required(&mut ours, &mut waiting, &mut signatures, Vec::new());
+        assert_eq!(screened.applied, 0);
+        assert_eq!(waiting.len(), 2);
+        // The last one's: both in.
+        signatures.offer(
+            &ours,
+            vec![(last, key.sign(&crate::document::signatures::message(&last)))],
+        );
+        let screened = required(&mut ours, &mut waiting, &mut signatures, Vec::new());
+        assert_eq!(screened.applied, 2);
+        assert!(screened.refused.is_empty());
+        assert_eq!(waiting.len(), 0);
+        assert_eq!(
+            store::read_path(&ours, &path("notes/x")).unwrap(),
+            Some(json!(1))
+        );
+    }
 }
diff --git a/src/document/replica.rs b/src/document/replica.rs
index a409969..cdf789c 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -230,6 +230,19 @@ impl Replica {
         &mut self,
         f: impl FnOnce(&mut Document, &mut automerge::PatchLog) -> (T, bool),
     ) -> T {
+        self.logged_signed(|doc, _, log| f(doc, log))
+    }
+
+    /// [`logged`](Self::logged), `f` given the document's signatures too
+    /// (`access::screen`, judging received changes, checks theirs).
+    pub(crate) fn logged_signed<T>(
+        &mut self,
+        f: impl FnOnce(
+            &mut Document,
+            &mut super::signatures::Signatures,
+            &mut automerge::PatchLog,
+        ) -> (T, bool),
+    ) -> T {
         use crate::locks::MutexExt;
         let before = sorted_heads(&self.doc);
         let mut log = self.log.lock_or_recover();
@@ -238,7 +251,7 @@ impl Replica {
             // lead here.
             *log = Log::at(self.epoch_seq, before);
         }
-        let (out, whole) = f(&mut self.doc, &mut log.patches);
+        let (out, whole) = f(&mut self.doc, &mut self.signatures, &mut log.patches);
         let after = sorted_heads(&self.doc);
         if whole {
             log.to = after;
diff --git a/src/document/signatures.rs b/src/document/signatures.rs
index 47e5917..63b19a5 100644
--- a/src/document/signatures.rs
+++ b/src/document/signatures.rs
@@ -340,6 +340,54 @@ impl Signatures {
         (signatures, next)
     }
 
+    /// Received signatures waiting for their changes (`offer`) checked against
+    /// `changes`, not in the document yet, each made by its committer (`committers`, as
+    /// `access::screen` finds them): one valid vouches for its actor's changes up to it,
+    /// these among them.
+    pub(crate) fn vouch_batch(&mut self, changes: &[super::Change], committers: &[Option<String>]) {
+        for (change, committer) in changes.iter().zip(committers) {
+            let hash = change.hash();
+            let Some(signature) = self.pending.remove(&hash) else {
+                continue;
+            };
+            let valid = committer
+                .as_deref()
+                .and_then(crate::identity::public_of)
+                .is_some_and(|key| crate::identity::verify(&key, &message(&hash), &signature));
+            if !valid {
+                continue;
+            }
+            self.insert(hash, signature);
+            let actor = change.actor_id();
+            if self
+                .latest
+                .get(actor)
+                .is_none_or(|latest| latest.seq < change.seq())
+            {
+                self.stamp += 1;
+                self.latest.insert(
+                    actor.clone(),
+                    Latest {
+                        seq: change.seq(),
+                        hash,
+                        stamp: self.stamp,
+                    },
+                );
+            }
+        }
+    }
+
+    /// Signs this node's latest change in `doc` with `key` if no signature of it is kept:
+    /// after a switch or a rejoin, the changes it replayed there (`remake::delta`), made
+    /// without one, vouched for at once.
+    pub(crate) fn sign_own_latest(&mut self, doc: &Document, key: &crate::identity::NodeKey) {
+        if let Some(change) = doc.get_last_local_change()
+            && !self.by_change.contains_key(&change.hash())
+        {
+            self.sign(doc, change.hash(), key);
+        }
+    }
+
     /// Whether a signature vouches for `actor`'s change `seq`.
     pub(crate) fn verified(&self, actor: &super::ActorId, seq: u64) -> bool {
         self.latest
diff --git a/src/state/host.rs b/src/state/host.rs
index e401509..fb2e2e9 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -24,8 +24,13 @@ impl AppState {
     /// document. (Lock order doc -> peer, as in sync generation.)
     fn swap_document(&self, replica: &mut Replica, epoch: Epoch, doc: Document) -> Document {
         let old = std::mem::replace(&mut replica.doc, doc);
-        // The old document's signatures are of its changes.
+        // The old document's signatures are of its changes; this node's replayed into
+        // the new one, signed at once.
         replica.signatures = Default::default();
+        if let Some(key) = &self.node_key {
+            let replica = &mut *replica;
+            replica.signatures.sign_own_latest(&replica.doc, key);
+        }
         replica.epoch = epoch;
         replica.epoch_seq += 1;
         // Publishes the epoch with the heads (`Replica::stamps`), under the same lock,
@@ -340,6 +345,11 @@ impl Host for AppState {
                     // take the document's lock first (`mutate_doc`).
                     let (old, resumed) = state.take_stale(old);
                     let out = f(&mut replica.doc, &old);
+                    // What it replayed, vouched for at once.
+                    if let Some(key) = &state.node_key {
+                        let replica = &mut *replica;
+                        replica.signatures.sign_own_latest(&replica.doc, key);
+                    }
                     replica.writable = true;
                     replica.refresh_heads();
                     if resumed {
@@ -403,15 +413,18 @@ impl Host for AppState {
             tokio::task::spawn_blocking(move || {
                 let result = {
                     let mut replica = state.doc.blocking_write();
-                    let result = if crate::access::policy::of(&replica.doc).enforced() {
+                    let policy = crate::access::policy::of(&replica.doc);
+                    let result = if policy.enforced() {
                         // Judged first (`access::screen`).
                         let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
+                        let replica = &mut *replica;
                         crate::access::screen::apply_bytes(
                             &mut replica.doc,
                             epoch,
                             epoch_seq,
                             change.raw_bytes(),
                             &state.screening,
+                            policy.required().then_some(&mut replica.signatures),
                             &mut crate::document::PatchLog::inactive(),
                         )
                         .map(|_| ())
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 049c9e6..294b446 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -542,7 +542,10 @@ impl AppState {
                     if after != before
                         && let [hash] = after[..]
                     {
-                        if state.sign_every == SignEvery::Change || sign.wanted(&replica.doc) {
+                        if state.sign_every == SignEvery::Change
+                            || sign.wanted(&replica.doc)
+                            || crate::access::policy::of(&replica.doc).required()
+                        {
                             let replica = &mut *replica;
                             replica.signatures.sign(&replica.doc, hash, key);
                         } else {
diff --git a/src/swarm/signatures.rs b/src/swarm/signatures.rs
index c172455..309ccc2 100644
--- a/src/swarm/signatures.rs
+++ b/src/swarm/signatures.rs
@@ -68,12 +68,47 @@ pub(crate) async fn received(state: &AppState, message: SignaturesMessage) {
         return;
     };
     let doc = std::sync::Arc::clone(&state.doc);
-    let _ = tokio::task::spawn_blocking(move || {
+    let screening = std::sync::Arc::clone(&state.screening);
+    // How many waiting changes went in, if any: published as received ones are.
+    let applied = tokio::task::spawn_blocking(move || {
         let mut replica = doc.blocking_write();
-        if replica.epoch == message.epoch {
+        if replica.epoch != message.epoch {
+            return 0;
+        }
+        {
+            let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
             let replica = &mut *replica;
             replica.signatures.offer(&replica.doc, signatures);
+            // With signatures required, changes waiting for them judged again
+            // (`access::screen`).
+            if crate::access::policy::of(&replica.doc).required() {
+                let applied = replica.logged_signed(|doc, known, log| {
+                    let applied = crate::access::screen::apply_bytes(
+                        doc,
+                        epoch,
+                        epoch_seq,
+                        &[],
+                        &screening,
+                        Some(known),
+                        log,
+                    );
+                    (applied, true)
+                });
+                replica.refresh_heads();
+                return match applied {
+                    Ok(applied) => applied,
+                    Err(error) => {
+                        tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures");
+                        0
+                    }
+                };
+            }
+            0
         }
     })
-    .await;
+    .await
+    .unwrap_or(0);
+    if applied > 0 {
+        state.publish_change();
+    }
 }
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 8ca7d99..ff7fa5a 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -608,13 +608,25 @@ pub(crate) async fn handle_sync(
             let epoch_seq = replica.epoch_seq;
             let mut message = message;
             replica
-                .logged(|doc, log| {
+                .logged_signed(|doc, known, log| {
                     let result = (|| {
-                        // Judged first, as the policy has it (`access::screen`).
-                        if crate::access::policy::of(&*doc).enforced() {
+                        // Judged first, as the policy has it (`access::screen`); with
+                        // signatures required, by those it sent too.
+                        let policy = crate::access::policy::of(&*doc);
+                        if policy.enforced() {
                             let bytes = message.take_changes();
+                            let required = policy.required();
+                            if required {
+                                known.offer(doc, signatures.clone());
+                            }
                             crate::access::screen::apply_bytes(
-                                doc, epoch, epoch_seq, &bytes, &screening, log,
+                                doc,
+                                epoch,
+                                epoch_seq,
+                                &bytes,
+                                &screening,
+                                required.then_some(&mut *known),
+                                log,
                             )?;
                         }
                         sync.receive_logged(doc, message, log)
@@ -879,15 +891,18 @@ async fn download_and_apply_delta(
             bail!("the epoch changed during the download");
         }
         let heads_before = replica.heads().to_vec();
-        if crate::access::policy::of(&replica.doc).enforced() {
+        let policy = crate::access::policy::of(&replica.doc);
+        if policy.enforced() {
             // Judged first (`access::screen`).
             let epoch_seq = replica.epoch_seq;
+            let replica = &mut *replica;
             crate::access::screen::apply_bytes(
                 &mut replica.doc,
                 epoch,
                 epoch_seq,
                 &bytes,
                 &screening,
+                policy.required().then_some(&mut replica.signatures),
                 &mut crate::document::PatchLog::inactive(),
             )?;
         } else {