From 3d7230d952aa4a71a1aeb64fee9841d16b70ed91 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 7 Nov 2026 22:35:21 +0100 Subject: 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. --- docs/access.md | 11 ++-- docs/api.md | 4 +- src/access/policy.rs | 28 ++++++++--- src/access/screen.rs | 122 +++++++++++++++++++++++++++++++++++++++++++-- src/document/replica.rs | 15 +++++- src/document/signatures.rs | 48 ++++++++++++++++++ src/state/host.rs | 17 ++++++- src/state/mod.rs | 5 +- src/swarm/signatures.rs | 41 +++++++++++++-- src/swarm/sync.rs | 25 ++++++++-- 10 files changed, 290 insertions(+), 26 deletions(-) diff --git a/docs/access.md b/docs/access.md index df8ad53..ba9caf1 100644 --- a/docs/access.md +++ b/docs/access.md @@ -192,6 +192,12 @@ revocation since refuses nothing it allowed then, and every node judges alike. * 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. +With `"signatures": "required"`, a change also needs a signature by the node it names +([`api.md`](api.md), Signed changes): every node then signs every change it makes, and +one no signature vouches for yet waits, with whatever builds on it, until one does +(they come with the changes, or within a second on their own). A member can then no +longer write as another member; the user a change names is still its node's word. + 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 @@ -199,9 +205,8 @@ 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. +* **Prove the user.** A change is its node's (with signatures required, proven) and its + claimed user's: a member could claim a user its operator doesn't have. * **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 diff --git a/docs/api.md b/docs/api.md index c2b9cb4..5d94fac 100644 --- a/docs/api.md +++ b/docs/api.md @@ -261,7 +261,9 @@ A change left unsigned, such as a metric written many times a second, is vouched by the node's next signed one: with `--sign-every 1s` (the default), the node signs its latest change at most a second after it if nothing signed it before, so every change is vouched for within a second, at one signature a second at most; `--sign-every all` signs every -change as it is made, and `--sign-every none` signs only those above. +change as it is made, and `--sign-every none` signs only those above. A swarm whose policy requires +signatures ([`access.md`](access.md), Changes judged on every node) has every change +signed as it is made, whatever `--sign-every` says. Signatures travel with sync messages: a node passes on, for each node's writes, the latest signature it checked, and checks one it receives once it holds the change (a signature costs about 30 µs to check, and one per node and run vouches for the rest). 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 { 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, waiting: &mut Waiting, + mut signatures: Option<&mut Signatures>, log: &mut crate::document::PatchLog, ) -> Result { 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> = HashMap::new(); let mut refused: HashSet = HashSet::new(); + let mut unsigned: HashSet = HashSet::new(); let mut accepted: Vec = 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 = 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, + signatures: Option<&mut Signatures>, log: &mut crate::document::PatchLog, ) -> Result { 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 @@ -229,6 +229,19 @@ impl Replica { pub(crate) fn logged( &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( + &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); @@ -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]) { + 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 { -- cgit v1.3.1