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.

15 files changed, 374 insertions, 11 deletions

diff --git a/docs/access.md b/docs/access.md
index 961fc72..df8ad53 100644
--- a/docs/access.md
+++ b/docs/access.md
@@ -164,3 +164,50 @@ the last user who may manage users: each node can only judge what it holds, and
 such writes on two nodes at once would leave nobody once they merge anyway. Then,
 while nobody may manage users, the users in a node's `--protocol-writers` may do
 everything there again, to set it right.
+
+## Changes judged on every node
+
+What a user may do is checked where it writes, by the node it writes through. A swarm
+can also have every node judge every change it receives before applying it, so a node
+whose operator lets anyone write (or that claims another's users) gets nothing past
+the others: the policy, at `_continuum/access/policy`, written by a user who may write
+protocol data.
+
+```sh
+curl -X PUT localhost:3000/store/_continuum/access/policy \
+  -H 'Content-Type: application/json' -d '{"version": 1, "signatures": "optional"}'
+```
+
+A change is judged by what it touches, by the node that committed it, and by the user
+it names, as the document stood when it was written (at its dependencies): a
+revocation since refuses nothing it allowed then, and every node judges alike.
+
+* A member's entry (`_continuum/members/<id>`) and its votes are written by that member
+  only; removing an entry, also by a user who may write there.
+* Epochs and the swarm ID (`_continuum/epochs`, `_continuum/swarm`) by an epoch's first
+  change only (the genesis change, a remake's snapshot).
+* A remake's decision by the coordinator its proposal names; the rest of a record by
+  any member.
+* A user's tokens revoked by that user too (logging out).
+* Everything else, protocol data or not, by a user the registry lets write there.
+* A node that isn't a member may only register itself.
+
+In a swarm without a registry, whatever a user would need, any member may. A refused
+change stays out, and so does whatever builds on it; the node logs it (`Refused a
+change`) and lists the last thousand (`GET /admin/refused`). A change whose
+dependencies haven't come yet waits for them.
+
+What it doesn't do yet:
+
+* **Prove who wrote it.** A change is its claimed node's and user's: a member could
+  claim another member, or a user its node doesn't have. Changes are signed
+  ([`api.md`](api.md), Signed changes), but the policy doesn't require it yet.
+* **Judge a joining node's first sync.** An empty document has no policy: what it gets
+  from the node it joins through is taken as it is, as that node judged it.
+* **Agree at once.** The policy applies as each node gets it: a change received between
+  the policy's arrival on one node and on another may be refused by one and kept by
+  the other.
+
+A node that doesn't know the policy (another version, or a setting it lacks) takes no
+writes through the API (`409`) and receives without judging, until it is upgraded.
+
diff --git a/docs/api.md b/docs/api.md
index 89da7ba..c2b9cb4 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -238,7 +238,7 @@ are each node's own (`--htpasswd`, or its proxy's): `alice` on two nodes is one
 only where their operators make it so, which nothing checks.
 `GET /admin/changes` lists them,
 newest first, with the paths each touched and whether a signature by its node vouches
-for it (`verified`, Signed changes); `GET /admin/changes/settings` those that
+for it (`verified`, Signed changes), and `GET /admin/refused` those this node refused under the swarm's policy ([`access.md`](access.md), Changes judged on every node); `GET /admin/changes/settings` those that
 touched what is at `settings`.
 Every node holds every change, so user names are known to the whole swarm, and so is
 who wrote what, until a remake drops the history ([`remakes.md`](remakes.md)).
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index d067f4f..742a013 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -626,6 +626,40 @@ paths:
         '503':
           description: Node is not ready.
 
+  /admin/refused:
+    get:
+      tags: [Admin]
+      summary: Changes refused under the swarm's policy
+      description: |
+        The changes this node refused under the swarm's policy
+        (`_continuum/access/policy`, `access.md`), newest first, the last thousand at
+        most since it started or switched epochs: each with its hash, when its writer
+        made it (its clock), the node and user it names, what it touched, and why.
+      responses:
+        '200':
+          description: The refused changes.
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  type: object
+                  required: [hash, at, paths, why]
+                  properties:
+                    hash:
+                      type: string
+                    at:
+                      type: integer
+                    node:
+                      type: string
+                    user:
+                      type: string
+                    paths:
+                      type: array
+                      items:
+                        type: string
+                    why:
+                      type: string
   /admin/changes/{key}:
     get:
       tags: [Admin]
diff --git a/docs/trust.md b/docs/trust.md
index 70369e2..5444b8b 100644
--- a/docs/trust.md
+++ b/docs/trust.md
@@ -57,7 +57,10 @@ Protect the configuration like the swarm key.
 ## What members are trusted with
 
 Everything in the document. A member syncs changes to and from its peers, and every
-node acts on what the document says:
+node acts on what the document says. Unless the swarm has a policy
+([`access.md`](access.md), Changes judged on every node): then every node judges every
+change it receives by rules by path, and those below are refused where a change breaks
+them, as far as the node and user a change claims are who wrote it:
 
 * **User data**: any member can write or delete any path.
 * **Who wrote what** (`GET /admin/changes`): each change names its node and API user
diff --git a/src/access/mod.rs b/src/access/mod.rs
index 1109fa6..ea2e65c 100644
--- a/src/access/mod.rs
+++ b/src/access/mod.rs
@@ -13,12 +13,9 @@ use serde::{Deserialize, Serialize};
 
 use crate::document::meta;
 
+pub(crate) mod policy;
 pub(crate) mod registry;
-// Judging changes on receipt is what uses it; until then, the tests.
-#[cfg_attr(not(test), allow(dead_code))]
 pub(crate) mod rules;
-// Judging changes on receipt is what uses it; until then, the tests.
-#[cfg_attr(not(test), allow(dead_code))]
 pub(crate) mod screen;
 pub(crate) mod token;
 
diff --git a/src/access/policy.rs b/src/access/policy.rs
new file mode 100644
index 0000000..7ddef0e
--- /dev/null
+++ b/src/access/policy.rs
@@ -0,0 +1,93 @@
+//! The swarm's write protection policy (`_continuum/access/policy`): whether every node
+//! judges received changes by rules by path before applying them (`access::screen`,
+//! `access::rules`). Absent, nobody does: what exists keeps working as it did. Written
+//! 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
+//! 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.
+
+use serde_json::Value as JsonValue;
+
+use crate::document::{ReadDoc, meta, store};
+
+/// The policy versions this node knows.
+pub(crate) const VERSION: u64 = 1;
+
+/// Where the policy is.
+pub(crate) fn path() -> Vec<String> {
+    meta::path(&[meta::ACCESS, "policy"])
+}
+
+/// What the policy says.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum Policy {
+    /// None: changes aren't judged.
+    Open,
+    /// Rules by path enforced.
+    Enforced,
+    /// One this node doesn't know, and why.
+    Unknown(String),
+}
+
+impl Policy {
+    pub(crate) fn enforced(&self) -> bool {
+        matches!(self, Policy::Enforced)
+    }
+}
+
+/// The policy `doc` holds.
+pub(crate) fn of(doc: &impl ReadDoc) -> Policy {
+    let value = match store::read_path(doc, &path()) {
+        Ok(Some(value)) => value,
+        Ok(None) => return Policy::Open,
+        Err(error) => return Policy::Unknown(format!("it doesn't read: {error}")),
+    };
+    let version = value.get("version").and_then(JsonValue::as_u64);
+    if version != Some(VERSION) {
+        return Policy::Unknown(format!(
+            "version {}, this node knows {VERSION}",
+            value
+                .get("version")
+                .map_or("none".to_owned(), JsonValue::to_string)
+        ));
+    }
+    match value.get("signatures").and_then(JsonValue::as_str) {
+        None | Some("optional") => Policy::Enforced,
+        Some(other) => Policy::Unknown(format!("signatures {other:?}")),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::document::Document;
+    use crate::document::testing::change;
+    use serde_json::json;
+
+    fn with(policy: JsonValue) -> Policy {
+        let mut doc = Document::new();
+        change(&mut doc, |doc| {
+            store::write_path(doc, &path(), &policy).unwrap();
+        });
+        of(&doc)
+    }
+
+    #[test]
+    fn what_the_policy_says() {
+        assert_eq!(of(&Document::new()), Policy::Open);
+        assert_eq!(with(json!({"version": 1})), Policy::Enforced);
+        assert_eq!(
+            with(json!({"version": 1, "signatures": "optional"})),
+            Policy::Enforced
+        );
+        assert!(matches!(with(json!({"version": 2})), Policy::Unknown(_)));
+        assert!(matches!(
+            with(json!({"version": 1, "signatures": "sometimes"})),
+            Policy::Unknown(_)
+        ));
+        assert!(matches!(with(json!("on")), Policy::Unknown(_)));
+    }
+}
diff --git a/src/access/screen.rs b/src/access/screen.rs
index dd9ffcd..66b7fc8 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -36,6 +36,52 @@ pub(crate) struct Refused {
     pub(crate) why: String,
 }
 
+/// What screening keeps between batches, for one document (`AppState::screening`): the
+/// changes waiting for their dependencies, and those refused, for operators
+/// (`GET /admin/refused`).
+#[derive(Debug, Default)]
+pub(crate) struct Screening {
+    /// The document's (`Replica::epoch_seq`): another's start empty (`of`).
+    epoch_seq: u32,
+    pub(crate) waiting: Waiting,
+    refused: std::collections::VecDeque<Refused>,
+}
+
+impl Screening {
+    /// Itself for the document `epoch_seq`: emptied if it was another's.
+    pub(crate) fn of(&mut self, epoch_seq: u32) -> &mut Self {
+        if self.epoch_seq != epoch_seq {
+            *self = Screening {
+                epoch_seq,
+                ..Screening::default()
+            };
+        }
+        self
+    }
+
+    /// Keeps `refused`, logged, the oldest going past `MAX_REFUSED`.
+    pub(crate) fn record(&mut self, refused: Vec<Refused>) {
+        for refused in refused {
+            tracing::warn!(
+                change = %refused.change.hash(),
+                committer = refused.committer.as_deref().unwrap_or("none"),
+                user = refused.user.as_deref().unwrap_or("none"),
+                why = %refused.why,
+                "Refused a change (_continuum/access/policy)"
+            );
+            self.refused.push_back(refused);
+            if self.refused.len() > MAX_REFUSED {
+                self.refused.pop_front();
+            }
+        }
+    }
+
+    /// Those refused, oldest first.
+    pub(crate) fn refused(&self) -> impl DoubleEndedIterator<Item = &Refused> {
+        self.refused.iter()
+    }
+}
+
 /// Received changes waiting for their dependencies.
 #[derive(Debug, Default)]
 pub(crate) struct Waiting(Vec<Change>);
@@ -179,6 +225,35 @@ pub(crate) fn apply(
     Ok(screened)
 }
 
+/// [`apply`] of the changes in `bytes`, as sync messages and downloads carry them, with
+/// 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.
+pub(crate) fn apply_bytes(
+    doc: &mut Document,
+    epoch: Epoch,
+    epoch_seq: u32,
+    bytes: &[u8],
+    screening: &std::sync::Mutex<Screening>,
+    log: &mut crate::document::PatchLog,
+) -> Result<usize> {
+    use crate::locks::MutexExt;
+    if bytes.is_empty() {
+        return Ok(0);
+    }
+    if let Some(misplaced) = crate::document::changes::misplaced_author(bytes) {
+        anyhow::bail!("Refused changes: {misplaced}");
+    }
+    let changes = crate::document::changes::changes_in(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)?;
+    let applied = screened.applied;
+    screening.record(screened.refused);
+    Ok(applied)
+}
+
 /// Whether a change touching `paths` changes what rules read: the members, the user
 /// registry, remake records, the protocol data as a whole.
 fn relevant(paths: &BTreeSet<Path>) -> bool {
diff --git a/src/api/changes.rs b/src/api/changes.rs
index c3a847d..13c8dd3 100644
--- a/src/api/changes.rs
+++ b/src/api/changes.rs
@@ -139,3 +139,43 @@ fn changes(
     }
     ChangesReport { changes, more }
 }
+/// A change the policy refused (`access::screen`): its hash, when its writer made it
+/// (by its clock), who made it and for whom, as it says, what it touched, and why.
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct RefusedReport {
+    hash: String,
+    at: i64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    node: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    user: Option<String>,
+    paths: Vec<String>,
+    why: String,
+}
+
+/// `GET /admin/refused`: the changes this node refused under the swarm's policy
+/// (`_continuum/access/policy`), newest first, the last thousand at most, since it
+/// started or switched epochs.
+pub(crate) async fn get_refused(State(state): State<AppState>) -> Json<Vec<RefusedReport>> {
+    use crate::locks::MutexExt;
+    let screening = state.screening.lock_or_recover();
+    Json(
+        screening
+            .refused()
+            .rev()
+            .map(|refused| RefusedReport {
+                hash: refused.change.hash().to_string(),
+                at: refused.change.timestamp(),
+                node: refused.committer.clone(),
+                user: refused.user.clone(),
+                paths: refused
+                    .paths
+                    .iter()
+                    .map(|path| format!("/{}", path.join("/")))
+                    .collect(),
+                why: refused.why.clone(),
+            })
+            .collect(),
+    )
+}
diff --git a/src/api/mod.rs b/src/api/mod.rs
index 38868e2..a12e219 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -62,6 +62,7 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) ->
         .route("/admin/stats", get(stats::get_stats))
         .route("/admin/document", get(document::get_document))
         .route("/admin/changes", get(changes::get_changes))
+        .route("/admin/refused", get(changes::get_refused))
         .route("/admin/changes/{*key}", get(changes::get_changes_at))
         .route("/admin/config", get(config::get_config))
         .route("/auth/login", axum::routing::post(login::login))
@@ -311,6 +312,7 @@ pub(crate) fn store_status(error: StoreError, context: &str) -> ApiError {
         | StoreError::NotCastable { .. } => StatusCode::CONFLICT,
         StoreError::NotFound | StoreError::UnknownHeads => StatusCode::NOT_FOUND,
         StoreError::Rejoining => StatusCode::SERVICE_UNAVAILABLE,
+        StoreError::UnknownPolicy(_) => StatusCode::CONFLICT,
         other => {
             error!(error = %format!("{other:#}"), "{context}");
             return StatusCode::INTERNAL_SERVER_ERROR.into();
diff --git a/src/document/store.rs b/src/document/store.rs
index 31c4824..778307f 100644
--- a/src/document/store.rs
+++ b/src/document/store.rs
@@ -175,6 +175,12 @@ pub(crate) enum StoreError {
     /// (`Replica::writable`).
     #[error("the node is rejoining a newer epoch; writes are refused until it is done")]
     Rejoining,
+    /// The swarm's write protection policy is one this node doesn't know
+    /// (`access::policy`): no writes through the API until it is upgraded.
+    #[error(
+        "the swarm's policy (_continuum/access/policy) is one this node doesn't know ({0}); upgrade it to write"
+    )]
+    UnknownPolicy(String),
     #[error(transparent)]
     Doc(#[from] automerge::AutomergeError),
     #[error(transparent)]
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 8cb740b..f124ee8 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -22,6 +22,16 @@ use crate::transport::SyncPayload;
 /// A sync message from a peer, decoded (`decode`), to apply (`SyncSession::receive`).
 pub(crate) struct Received(sync::Message);
 
+impl Received {
+    /// The changes it carries, as their chunks, taken out of it: applied otherwise
+    /// (`access::screen`), what is left of it says what the peer has and needs.
+    pub(crate) fn take_changes(&mut self) -> Vec<u8> {
+        let bytes = self.0.changes.iter().flatten().copied().collect();
+        self.0.changes = sync::ChunkList::empty();
+        bytes
+    }
+}
+
 /// The changes none of the others depend on, sorted as heads are (`get_heads`).
 fn tips<'a>(
     changes: impl Iterator<
diff --git a/src/node.rs b/src/node.rs
index 3d3d85d..736723b 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -162,6 +162,7 @@ impl Node {
             first_users: Arc::new(config.first_users),
             node_key: config.node_key,
             sign_every: config.sign_every,
+            screening: Default::default(),
             token_lifetime: config.token_lifetime,
             session_lifetime: config.session_lifetime,
             cors_origins: Arc::new(config.cors_origins),
diff --git a/src/state/host.rs b/src/state/host.rs
index a48a6fb..e401509 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -403,9 +403,24 @@ impl Host for AppState {
             tokio::task::spawn_blocking(move || {
                 let result = {
                     let mut replica = state.doc.blocking_write();
-                    let result = replica
-                        .apply_changes([change])
-                        .map_err(|error| error.to_string());
+                    let result = if crate::access::policy::of(&replica.doc).enforced() {
+                        // Judged first (`access::screen`).
+                        let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
+                        crate::access::screen::apply_bytes(
+                            &mut replica.doc,
+                            epoch,
+                            epoch_seq,
+                            change.raw_bytes(),
+                            &state.screening,
+                            &mut crate::document::PatchLog::inactive(),
+                        )
+                        .map(|_| ())
+                        .map_err(|error| format!("{error:#}"))
+                    } else {
+                        replica
+                            .apply_changes([change])
+                            .map_err(|error| error.to_string())
+                    };
                     replica.refresh_heads();
                     result
                 }; // lock released before notifying
diff --git a/src/state/mod.rs b/src/state/mod.rs
index ff19124..049c9e6 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -150,6 +150,9 @@ pub(crate) struct AppState {
     /// Which of its changes this node signs besides those a write's `Sign` says
     /// (`--sign-every`).
     pub(crate) sign_every: SignEvery,
+    /// Received changes waiting for their dependencies, and those refused, while the
+    /// policy has them judged (`access::screen`). Locked after the document.
+    pub(crate) screening: Arc<std::sync::Mutex<crate::access::screen::Screening>>,
     pub(crate) token_lifetime: Duration,
     pub(crate) session_lifetime: Duration,
     /// Origins whose pages may call this node's open endpoints (`api::cors`).
@@ -490,6 +493,14 @@ impl AppState {
         tokio::task::spawn_blocking(move || {
             let result = {
                 let mut replica = state.doc.blocking_write();
+                // A write through the API under a policy this node doesn't know: what
+                // the others would refuse it can't tell (`access::policy`).
+                if user.is_some()
+                    && let crate::access::policy::Policy::Unknown(why) =
+                        crate::access::policy::of(&replica.doc)
+                {
+                    return Err(E::from(StoreError::UnknownPolicy(why)));
+                }
                 // Checked here, under the lock: the caller's readiness check (if any)
                 // was before a rejoin could swap in a fresh document.
                 if !replica.writable {
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`).
         {