access: the policy enforced

_continuum/access/policy ({"version": 1}) has every node screen every
change it receives, on every path changes come in. A policy a node
doesn't know stops its writes through the API. GET /admin/refused.

Auth plan M6.

1 files changed, 31 insertions, 2 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index ed3679d..8ca7d99 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -541,6 +541,7 @@ pub(crate) async fn handle_sync(
         let peer_state_mutex = Arc::clone(&peer_state_mutex);
         let stats = Arc::clone(&state.stats);
         let throughput = Arc::clone(&state.throughput);
+        let screening = Arc::clone(&state.screening);
         let rtt = state
             .lock_rtt()
             .effective(peer, Instant::now(), state.timing.rtt_fresh_for());
@@ -604,8 +605,22 @@ pub(crate) async fn handle_sync(
             let said = message.heads().to_vec();
 
             let sync = &mut peer_state.sync;
+            let epoch_seq = replica.epoch_seq;
+            let mut message = message;
             replica
-                .logged(|doc, log| (sync.receive_logged(doc, message, log), true))
+                .logged(|doc, log| {
+                    let result = (|| {
+                        // Judged first, as the policy has it (`access::screen`).
+                        if crate::access::policy::of(&*doc).enforced() {
+                            let bytes = message.take_changes();
+                            crate::access::screen::apply_bytes(
+                                doc, epoch, epoch_seq, &bytes, &screening, log,
+                            )?;
+                        }
+                        sync.receive_logged(doc, message, log)
+                    })();
+                    (result, true)
+                })
                 .context("Automerge sync failed")?;
             peer_state.last_sync_rx = Some(Instant::now());
             if let Some(wait) = &peer_state.sync.reply {
@@ -857,13 +872,27 @@ async fn download_and_apply_delta(
     let measure_propagation = state.initial_sync_done.load(Ordering::Acquire);
     let clock = Arc::clone(&state.clock);
     let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
+    let screening = Arc::clone(&state.screening);
     tokio::task::spawn_blocking(move || -> Result<bool> {
         let mut replica = doc.blocking_write();
         if replica.epoch != epoch {
             bail!("the epoch changed during the download");
         }
         let heads_before = replica.heads().to_vec();
-        crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
+        if crate::access::policy::of(&replica.doc).enforced() {
+            // Judged first (`access::screen`).
+            let epoch_seq = replica.epoch_seq;
+            crate::access::screen::apply_bytes(
+                &mut replica.doc,
+                epoch,
+                epoch_seq,
+                &bytes,
+                &screening,
+                &mut crate::document::PatchLog::inactive(),
+            )?;
+        } else {
+            crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
+        }
         replica.refresh_heads();
         // Signatures waiting for what it brought (`Signatures::offer`).
         {