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.

1 files changed, 119 insertions, 3 deletions

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