screen: signatures required as the policy a change came under says

Review 3, R1: what a node received was vouched for only while its own
document's policy required signatures. A joiner's first sync, and
every node's message bringing `required`, went in by the rules alone:
a malicious first peer could plant changes under other members' actors
at their next seqs, their real ones then failing as duplicates for
good.

The signatures a message carries are always offered now, and the
screen always has them: a change waits for one where the policy at the
heads it is judged at requires it (rules::Reads::required, read once
per heads), whatever the receiving document holds. Changes written
under a policy not requiring signatures wait for none, as on every
other node; the check for forged changes runs always. Changes waiting
for a signature are judged again whenever signatures come.

The test of signatures required now takes the history into an empty
document; a new one takes changes written under a policy not requiring
them without any.

2 files changed, 32 insertions, 41 deletions

diff --git a/src/swarm/signatures.rs b/src/swarm/signatures.rs
index 9ea404e..111b3da 100644
--- a/src/swarm/signatures.rs
+++ b/src/swarm/signatures.rs
@@ -75,35 +75,30 @@ pub(crate) async fn received(state: &AppState, message: SignaturesMessage) {
         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(outcome) => outcome.applied,
-                    Err(error) => {
-                        tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures");
-                        0
-                    }
-                };
+        let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
+        let replica = &mut *replica;
+        replica.signatures.offer(&replica.doc, signatures);
+        // Changes waiting for them judged again (`access::screen`): those the policy they
+        // were written under requires signed.
+        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();
+        match applied {
+            Ok(outcome) => outcome.applied,
+            Err(error) => {
+                tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures");
+                0
             }
-            0
         }
     })
     .await
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 75c20f1..e76f5f9 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -503,7 +503,7 @@ pub(crate) async fn handle_sync(
     let epoch_label = epoch;
     // Signatures of changes (`document::signatures`): a malformed list, or more than any
     // sender sends, counts for nothing.
-    let signatures = crate::document::signatures::unpack(&envelope.signatures)
+    let mut signatures = crate::document::signatures::unpack(&envelope.signatures)
         .filter(|signatures| signatures.len() <= crate::document::signatures::MAX_PER_MESSAGE)
         .unwrap_or_default();
     let from_ready = envelope.ready;
@@ -643,21 +643,18 @@ pub(crate) async fn handle_sync(
                     let result = (|| {
                         // Judged first, each as the policy it was written under has it
                         // (`access::screen`), whether this document has one yet or not;
-                        // with signatures required, by those it sent too.
-                        let policy = crate::access::policy::of(&*doc);
+                        // and vouched for by the signatures it sent too, where that policy
+                        // requires them.
                         {
                             let bytes = message.take_changes();
-                            let required = policy.required();
-                            if required {
-                                known.offer(doc, signatures.clone());
-                            }
+                            known.offer(doc, std::mem::take(&mut signatures));
                             let outcome = crate::access::screen::apply_bytes(
                                 doc,
                                 epoch,
                                 epoch_seq,
                                 &bytes,
                                 &screening,
-                                required.then_some(&mut *known),
+                                Some(&mut *known),
                                 log,
                             )?;
                             refused_again = outcome.refused_again;
@@ -708,10 +705,10 @@ pub(crate) async fn handle_sync(
             stats.asked_known_elsewhere(asked);
             peer_state.asked_known += (asked.0 + asked.1) as u64;
             replica.refresh_heads();
-            // The signatures it sent, and those waiting for what it brought.
-            if !signatures.is_empty() || replica.heads() != heads_before.as_slice() {
+            // The signatures waiting for what it brought (those it sent, offered before).
+            if replica.heads() != heads_before.as_slice() {
                 let replica = &mut *replica;
-                replica.signatures.offer(&replica.doc, signatures);
+                replica.signatures.offer(&replica.doc, Vec::new());
             }
             let heads_after = replica.heads().to_vec();
             if delivered {
@@ -955,7 +952,6 @@ async fn download_and_apply_delta(
             bail!("the epoch changed during the download");
         }
         let heads_before = replica.heads().to_vec();
-        let policy = crate::access::policy::of(&replica.doc);
         {
             // Judged first (`access::screen`), whether this document has a policy yet or
             // not.
@@ -967,7 +963,7 @@ async fn download_and_apply_delta(
                 epoch_seq,
                 &bytes,
                 &screening,
-                policy.required().then_some(&mut replica.signatures),
+                Some(&mut replica.signatures),
                 &mut crate::document::PatchLog::inactive(),
             )?;
         }