aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/access/mod.rs5
-rw-r--r--src/access/policy.rs93
-rw-r--r--src/access/screen.rs75
-rw-r--r--src/api/changes.rs40
-rw-r--r--src/api/mod.rs2
-rw-r--r--src/document/store.rs6
-rw-r--r--src/document/sync.rs10
-rw-r--r--src/node.rs1
-rw-r--r--src/state/host.rs21
-rw-r--r--src/state/mod.rs11
-rw-r--r--src/swarm/sync.rs33
11 files changed, 288 insertions, 9 deletions
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`).
{