diff options
| -rw-r--r-- | docs/api.md | 5 | ||||
| -rw-r--r-- | docs/example.toml | 7 | ||||
| -rw-r--r-- | src/config/mod.rs | 9 | ||||
| -rw-r--r-- | src/document/signatures.rs | 142 | ||||
| -rw-r--r-- | src/main.rs | 1 | ||||
| -rw-r--r-- | src/node.rs | 6 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 1 | ||||
| -rw-r--r-- | src/state/mod.rs | 33 |
8 files changed, 198 insertions, 6 deletions
diff --git a/docs/api.md b/docs/api.md index e380c95..6a328a5 100644 --- a/docs/api.md +++ b/docs/api.md @@ -257,7 +257,10 @@ every change is signed, only those that * or were written with `?signed=true`. A change left unsigned, such as a metric written many times a second, is vouched for -by the node's next signed one. +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. ## Protocol data diff --git a/docs/example.toml b/docs/example.toml index 222db52..034976f 100644 --- a/docs/example.toml +++ b/docs/example.toml @@ -73,6 +73,13 @@ htpasswd = "/etc/continuum/htpasswd" # 0: never. # stream-heartbeat = "30s" +# Which of its changes this node signs with its node key besides those touching +# protocol data, paths listed in _continuum/access/signed, or written with +# ?signed=true: "all", "none" (one left unsigned is vouched for by the node's next +# signed change), or its latest change, if unsigned, this often (docs/api.md, Signed +# changes). +# sign-every = "1s" + # How long a token this node issues (POST /auth/login, /auth/refresh) is taken, by every # node of the swarm (docs/access.md, Tokens). Refreshed before it ends, it lasts until # the login's end (session-lifetime). diff --git a/src/config/mod.rs b/src/config/mod.rs index f0e5dc5..0da7ed7 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -124,6 +124,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "HTTP API users (from htpasswd or user-header) who may write protocol data under /store/_continuum (remake records, members, forget records), e.g. to repair a stuck remake, in a swarm without a user registry. Such writes act on the whole swarm and are logged. Epoch metadata and the swarm ID are never writable. With a registry its capabilities decide instead (docs/access.md), but for a registry nobody may manage users of any more: then these may do everything on this node, to set it right.", ), option( + "sign-every", + Kind::Text, + "all|none|DURATION", + Some("1s"), + "Which of its changes this node signs with its node key besides those touching protocol data, paths listed in _continuum/access/signed, or written with ?signed=true: all of them; none (one left unsigned is vouched for by the node's next signed change); or its latest change, if unsigned, this often, so every change is vouched for within it (docs/api.md, Signed changes).", + ), + option( "token-lifetime", Kind::Duration, "DURATION", @@ -566,6 +573,8 @@ pub(crate) struct Settings { pub(crate) bootstrap_users: Vec<String>, #[serde(default)] pub(crate) cors_origins: Vec<String>, + #[serde(default)] + pub(crate) sign_every: crate::document::signatures::SignEvery, #[serde( default = "default_token_lifetime", with = "crate::units::duration_option" diff --git a/src/document/signatures.rs b/src/document/signatures.rs index 5d50a33..e92cf9c 100644 --- a/src/document/signatures.rs +++ b/src/document/signatures.rs @@ -8,7 +8,9 @@ //! it verifies up to). So not every change needs one: a change is signed when it //! touches protocol data (`_continuum`) or a path the swarm lists as signed //! (`_continuum/access/signed`, `listed`), or when its write asks for it (`Sign`); a -//! change left unsigned is vouched for by the node's next signed one. +//! change left unsigned is vouched for by the node's next signed one, which, as +//! `--sign-every` says, is its latest change, signed a while after it if nothing signed +//! it before (`SignEvery`), or every change at once. use std::collections::HashMap; @@ -74,10 +76,86 @@ pub(crate) fn listed(doc: &Document) -> Vec<Vec<String>> { .collect() } +/// Which of this node's changes it signs besides those `Sign` says (`--sign-every`). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum SignEvery { + /// Every change, as it is made (`all`, `true`). + Change, + /// No others (`none`, `false`): one left unsigned waits for the node's next signed + /// change. + Never, + /// Its latest change, if unsigned, this often: every change is vouched for within it. + Interval(std::time::Duration), +} + +impl Default for SignEvery { + fn default() -> Self { + SignEvery::Interval(std::time::Duration::from_secs(1)) + } +} + +impl std::fmt::Display for SignEvery { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SignEvery::Change => f.write_str("all"), + SignEvery::Never => f.write_str("none"), + SignEvery::Interval(every) => f.write_str(&crate::units::format_duration(*every)), + } + } +} + +impl std::str::FromStr for SignEvery { + type Err = String; + + /// `all` or `true`, `none` or `false`, or a duration (`units::parse_duration`; 0 is + /// `none`). + fn from_str(text: &str) -> Result<Self, String> { + match text.trim() { + "all" | "true" => Ok(SignEvery::Change), + "none" | "false" => Ok(SignEvery::Never), + other => match crate::units::parse_duration(other) { + Ok(every) if every.is_zero() => Ok(SignEvery::Never), + Ok(every) => Ok(SignEvery::Interval(every)), + Err(error) => Err(format!( + "{other:?} is neither all, none nor a duration ({error})" + )), + }, + } + } +} + +impl serde::Serialize for SignEvery { + fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> { + s.serialize_str(&self.to_string()) + } +} + +impl<'de> serde::Deserialize<'de> for SignEvery { + fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> { + use serde::de::Error as _; + match serde_json::Value::deserialize(d)? { + serde_json::Value::Bool(all) => Ok(if all { + SignEvery::Change + } else { + SignEvery::Never + }), + serde_json::Value::String(text) => text.parse().map_err(D::Error::custom), + serde_json::Value::Number(number) => { + number.to_string().parse().map_err(D::Error::custom) + } + other => Err(D::Error::custom(format!( + "{other} is neither all, none nor a duration" + ))), + } + } +} + /// A document's signatures, by change. #[derive(Debug, Default)] pub(crate) struct Signatures { by_change: HashMap<ChangeHash, [u8; 64]>, + /// This node's latest change, if no signature vouches for it yet (`SignEvery`). + unsigned: Option<ChangeHash>, } impl Signatures { @@ -85,9 +163,33 @@ impl Signatures { self.by_change.insert(hash, signature); } - /// Signs `hash` with `key` and keeps the signature. + /// Signs `hash`, this node's latest change, with `key` and keeps the signature: it + /// vouches for every change of the node's before it too. pub(crate) fn sign(&mut self, hash: ChangeHash, key: &crate::identity::NodeKey) { self.insert(hash, key.sign(&message(&hash))); + self.unsigned = None; + } + + /// Whether this node's latest change waits for a signature (`sign_latest`). + pub(crate) fn pending(&self) -> bool { + self.unsigned.is_some() + } + + /// `hash`, this node's latest change, is left unsigned for now. + pub(crate) fn left_unsigned(&mut self, hash: ChangeHash) { + self.unsigned = Some(hash); + } + + /// Signs this node's latest change if no signature vouches for it yet; whether it + /// did. + pub(crate) fn sign_latest(&mut self, key: &crate::identity::NodeKey) -> bool { + match self.unsigned { + Some(hash) => { + self.sign(hash, key); + true + } + None => false, + } } } @@ -150,6 +252,42 @@ mod tests { assert!(check(&named, &hash, &key.sign(&message(&hash))).is_none()); } + /// `--sign-every`: `all`, `none`, a duration; 0 is `none`. + #[test] + fn sign_every_reads() { + use std::time::Duration; + let read = |text: &str| text.parse::<SignEvery>(); + assert_eq!(read("all"), Ok(SignEvery::Change)); + assert_eq!(read("true"), Ok(SignEvery::Change)); + assert_eq!(read("none"), Ok(SignEvery::Never)); + assert_eq!(read("0"), Ok(SignEvery::Never)); + assert_eq!( + read("500ms"), + Ok(SignEvery::Interval(Duration::from_millis(500))) + ); + assert!(read("sometimes").is_err()); + let toml: SignEvery = serde_json::from_value(json!(false)).unwrap(); + assert_eq!(toml, SignEvery::Never); + assert_eq!(SignEvery::default().to_string(), "1s"); + } + + /// The latest change left unsigned is signed later, once; a signature since covers it. + #[test] + fn the_latest_change_is_signed_later() { + let key = NodeKey::from_seed([1; 32]); + let mut signatures = Signatures::default(); + assert!(!signatures.sign_latest(&key)); + let a = ChangeHash([1; 32]); + let b = ChangeHash([2; 32]); + signatures.left_unsigned(a); + assert!(signatures.sign_latest(&key)); + assert!(signatures.by_change.contains_key(&a)); + assert!(!signatures.sign_latest(&key)); + signatures.left_unsigned(a); + signatures.sign(b, &key); + assert!(!signatures.sign_latest(&key)); + } + /// Protocol data, listed paths (within or above them) and asked-for writes are /// signed; others aren't. #[test] diff --git a/src/main.rs b/src/main.rs index 074890f..4dda536 100644 --- a/src/main.rs +++ b/src/main.rs @@ -252,6 +252,7 @@ async fn main() -> Result<()> { protocol_writers: args.protocol_writers.iter().cloned().collect(), first_users, node_key: Some(Arc::clone(&node_key)), + sign_every: args.sign_every, token_lifetime: args.token_lifetime, session_lifetime: args.session_lifetime, cors_origins: args.cors_origins.clone(), diff --git a/src/node.rs b/src/node.rs index e64db6c..c1a26de 100644 --- a/src/node.rs +++ b/src/node.rs @@ -68,6 +68,8 @@ pub(crate) struct NodeConfig { /// (`document::signatures`): every node's, its node ID being its public key; none in /// the simulations without the identity layer, whose node IDs are names. pub(crate) node_key: Option<Arc<crate::identity::NodeKey>>, + /// `--sign-every`. + pub(crate) sign_every: crate::document::signatures::SignEvery, /// `--token-lifetime`, `--session-lifetime`. pub(crate) token_lifetime: Duration, pub(crate) session_lifetime: Duration, @@ -153,6 +155,7 @@ impl Node { protocol_writers: Arc::new(config.protocol_writers), first_users: Arc::new(config.first_users), node_key: config.node_key, + sign_every: config.sign_every, token_lifetime: config.token_lifetime, session_lifetime: config.session_lifetime, cors_origins: Arc::new(config.cors_origins), @@ -230,6 +233,9 @@ impl Node { tokio::spawn(peer::probe_when_reachable(state.clone(), reachable)); tokio::spawn(remake::run_worker(state.clone())); + if let crate::document::signatures::SignEvery::Interval(every) = state.sign_every { + tokio::spawn(state.clone().sign_latest_every(every)); + } trace!("Remake worker task spawned"); tokio::spawn(swarm::topology::run_worker(state.clone())); diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index 5128d11..3f0f47d 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -214,6 +214,7 @@ impl Sim { first_users: Vec::new(), // Under the identity layer, its key signs as a node's does. node_key: self.identity.then(|| Arc::new(sim_key(id))), + sign_every: Default::default(), token_lifetime: Duration::from_secs(3600), session_lifetime: Duration::from_secs(86400), cors_origins: Vec::new(), diff --git a/src/state/mod.rs b/src/state/mod.rs index ef1559d..1fc8ea8 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -38,7 +38,7 @@ use tracing::{debug, error, warn}; use crate::document::lock::DocumentLock; use crate::document::replica::{Epoch, Stamp}; -use crate::document::signatures::Sign; +use crate::document::signatures::{Sign, SignEvery}; use crate::document::store::{ChangeSet, StoreError}; use crate::document::{Document, Transaction}; use crate::locks::{MutexExt, RwLockExt}; @@ -147,6 +147,9 @@ pub(crate) struct AppState { /// Signs the tokens this node issues and the changes it commits /// (`NodeConfig::node_key`). pub(crate) node_key: Option<Arc<crate::identity::NodeKey>>, + /// Which of its changes this node signs besides those a write's `Sign` says + /// (`--sign-every`). + pub(crate) sign_every: SignEvery, pub(crate) token_lifetime: Duration, pub(crate) session_lifetime: Duration, /// Origins whose pages may call this node's open endpoints (`api::cors`). @@ -527,9 +530,12 @@ impl AppState { let after = replica.doc.get_heads(); if after != before && let [hash] = after[..] - && sign.wanted(&replica.doc) { - replica.signatures.sign(hash, key); + if state.sign_every == SignEvery::Change || sign.wanted(&replica.doc) { + replica.signatures.sign(hash, key); + } else { + replica.signatures.left_unsigned(hash); + } } } replica.refresh_heads(); @@ -545,6 +551,27 @@ impl AppState { .expect("document write task panicked") } + /// This node's latest change signed, if no signature vouches for it yet, every + /// `every` (`--sign-every`): every change it makes is vouched for within that. + pub(crate) async fn sign_latest_every(self, every: Duration) { + let Some(key) = self.node_key.clone() else { + return; + }; + let mut ticks = tokio::time::interval(every); + ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + ticks.tick().await; + let state = self.clone(); + let key = Arc::clone(&key); + let _ = tokio::task::spawn_blocking(move || { + // Looked at first under the read lock: mostly there is nothing to sign. + state.doc.blocking_read().signatures.pending() + && state.doc.blocking_write().signatures.sign_latest(&key) + }) + .await; + } + } + pub(crate) fn publish_change(&self) { self.notify_store_changed(); if let Err(error) = self.sync_events.send(SyncEvent::DocumentChanged) { |