access: keeping the swarm manageable

A write that would leave its own writer unable to manage users
(deleting, disabling or narrowing its entry, removing the registry) is
refused; --protocol-writers can set a registry right that nobody may
manage any more.

Auth plan M1.

4 files changed, 143 insertions, 21 deletions

diff --git a/docs/access.md b/docs/access.md
index 042e0d5..b9e0296 100644
--- a/docs/access.md
+++ b/docs/access.md
@@ -88,5 +88,16 @@ curl -u alice --request PUT --header 'Content-Type: application/json' \
 ```
 
 From that write on, every node serves only the users the registry has, so the first
-entry must be the writer's own, with `["_continuum"]`, or nobody may change the
-registry again; add the other users the same way.
+entry must be the writer's own, with `["_continuum"]` (a first write that isn't is
+refused, `409`); add the other users the same way.
+
+## Keeping the swarm manageable
+
+A user may not lock itself out: a write that would keep its writer from managing users
+(enabled, with `write` on `_continuum/users`) when it could, deleting, disabling or
+narrowing its own entry, is refused (`409`), and nothing of it is written; so is one
+that would remove the registry (`_continuum/users`). Others are not kept from removing
+the last user who may manage users: each node can only judge what it holds, and two
+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.
diff --git a/src/access/registry.rs b/src/access/registry.rs
index 4b4304a..948e0de 100644
--- a/src/access/registry.rs
+++ b/src/access/registry.rs
@@ -108,6 +108,14 @@ pub(crate) fn user(doc: &impl ReadDoc, name: &str) -> Option<User> {
     value.is_object().then(|| User::from_json(&value))
 }
 
+/// The users who may manage users: enabled, with `write` on `_continuum/users`.
+pub(crate) fn managers(doc: &impl ReadDoc) -> Vec<String> {
+    names(doc)
+        .into_iter()
+        .filter(|name| caps_of(doc, name).allows(Right::Write, &users_dir()))
+        .collect()
+}
+
 /// What user `name` may do: none if it isn't registered (or is disabled).
 pub(crate) fn caps_of(doc: &impl ReadDoc, name: &str) -> Caps {
     user(doc, name).map(|user| user.caps()).unwrap_or_default()
@@ -301,6 +309,24 @@ mod tests {
     }
 
     #[test]
+    fn who_may_manage_users() {
+        let mut doc = Document::new();
+        for (name, admin) in [("alice", true), ("bob", false)] {
+            let entry = entry("x", 0, &first_caps(admin));
+            change(&mut doc, |tx| {
+                store::write_path(tx, &user_path(name), &entry).unwrap()
+            });
+        }
+        assert_eq!(managers(&doc), vec!["alice"]);
+        let mut disabled = user_path("alice");
+        disabled.push("disabled".to_owned());
+        change(&mut doc, |tx| {
+            store::write_path(tx, &disabled, &json!(true)).unwrap()
+        });
+        assert!(managers(&doc).is_empty());
+    }
+
+    #[test]
     fn concurrent_grants_both_hold() {
         let mut ours = Document::new();
         let entry = entry("x", 0, &[]);
diff --git a/src/api/access.rs b/src/api/access.rs
index 8a3e630..ade6542 100644
--- a/src/api/access.rs
+++ b/src/api/access.rs
@@ -9,7 +9,7 @@ use axum::{
     response::Response,
 };
 use serde_json::Value as JsonValue;
-use tracing::debug;
+use tracing::{debug, warn};
 
 use super::auth::AuthenticatedUser;
 use super::error_response;
@@ -194,6 +194,13 @@ pub(crate) async fn authorize(
         let replica = state.doc.read().await;
         if !registry::exists(&replica.doc) {
             Access::Everything
+        } else if state.protocol_writers.contains(&user)
+            && registry::managers(&replica.doc).is_empty()
+        {
+            // Nobody may manage users any more (two removals at once, merged; or an
+            // entry damaged): this node's protocol writers may set it right.
+            warn!(user = %user, "The user registry has nobody who may manage users; a protocol writer may do everything until it has");
+            Access::Everything
         } else {
             match registry::user(&replica.doc, &user) {
                 Some(entry) => Access::Caps(entry.caps()),
diff --git a/src/api/store.rs b/src/api/store.rs
index 1ab3951..e1f9b53 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -385,10 +385,13 @@ pub(crate) async fn put_store_entry(
     let writing = store::Writing::new(body.sticky, body.timestamps, query.retype);
 
     // A value that can't be shown as asked is refused, and the write with it.
+    let writer = user.0.clone();
     let ((retyped, read), heads) = state
         .mutate_doc_for(Some(user.0.clone()), move |doc| {
+            let before = Before::of(&*doc, &writer);
             let retyped = store::write_val_path(doc, &path, &body.value, writing)
                 .map_err(|e| store_status(e, "Failed to write store value"))?;
+            registry_kept(&*doc, &path, before, &writer)?;
             // What was written, shown back to whoever wrote it.
             let read = read_shown(&*doc, &path, accept.as_ref(), false, &Access::Everything)?;
             Ok::<_, ApiError>((retyped, read))
@@ -465,6 +468,70 @@ fn parse_write_path(
     Ok(path)
 }
 
+/// The registry as a write found it (`registry_kept`): whether it existed, and whether
+/// the writing user could manage users (`registry::managers`).
+#[derive(Clone, Copy)]
+struct Before {
+    existed: bool,
+    managed: bool,
+}
+
+impl Before {
+    fn of(doc: &impl crate::document::ReadDoc, user: &str) -> Self {
+        use crate::access::registry;
+        let existed = registry::exists(doc);
+        Self {
+            existed,
+            managed: existed
+                && registry::caps_of(doc, user)
+                    .allows(crate::access::Right::Write, &registry::users_dir()),
+        }
+    }
+}
+
+/// `409` for a write to `path` by `user` that would lock that user out of managing
+/// users: one that could and no longer may (deleting, disabling or narrowing its own
+/// entry), and the write that makes the registry, if it doesn't make `user` one who may
+/// (from then on only registered users are served). And for one that would remove
+/// the registry, back to a swarm without one. Checked in the write's transaction,
+/// after it: the write is dropped whole. Others are not kept from removing the last who
+/// may manage users: no node could keep two such writes at once on two nodes from
+/// leaving nobody once they merge; then the users in `--protocol-writers` may do
+/// everything again (`access::authorize`).
+fn registry_kept(
+    doc: &impl crate::document::ReadDoc,
+    path: &[String],
+    before: Before,
+    user: &str,
+) -> Result<(), ApiError> {
+    use crate::access::registry;
+    if !store::paths_related(path, &registry::users_dir()) {
+        return Ok(());
+    }
+    let refuse = |why: &str| {
+        Err(ApiError(
+            StatusCode::CONFLICT,
+            Some(format!("{why} (docs/access.md)")),
+        ))
+    };
+    if !registry::exists(doc) {
+        return if before.existed {
+            refuse("this would remove the user registry")
+        } else {
+            Ok(())
+        };
+    }
+    let manages =
+        registry::caps_of(doc, user).allows(crate::access::Right::Write, &registry::users_dir());
+    if !before.existed && !manages {
+        return refuse("the registry's first users must include you, with write on _continuum");
+    }
+    if before.managed && !manages {
+        return refuse("this would keep you from managing users");
+    }
+    Ok(())
+}
+
 /// How a value is shown, as the request's `Accept` asks.
 #[derive(Clone, Copy, Debug)]
 enum Shown {
@@ -718,12 +785,16 @@ pub(crate) async fn post_store_entry(
         return Ok(redirect);
     }
     if let Some(by) = query.increment {
+        let writer = user.0.clone();
         let (value, heads) = state
             .mutate_doc_for(Some(user.0.clone()), move |doc| {
-                store::increment_path(doc, &path, by)
+                let before = Before::of(&*doc, &writer);
+                let value = store::increment_path(doc, &path, by)
+                    .map_err(|e| store_status(e, "Failed to increment a counter"))?;
+                registry_kept(&*doc, &path, before, &writer)?;
+                Ok::<_, ApiError>(value)
             })
-            .await
-            .map_err(|e| store_status(e, "Failed to increment a counter"))?;
+            .await?;
         return Ok(with_heads(
             {
                 let counter = ScalarValue::counter(value);
@@ -750,10 +821,13 @@ pub(crate) async fn post_store_entry(
         };
         let accept = headers.get(ACCEPT).cloned();
         // A value that can't be shown as asked is refused, and the cast with it.
+        let writer = user.0.clone();
         let ((retyped, read), heads) = state
             .mutate_doc_for(Some(user.0.clone()), move |doc| {
+                let before = Before::of(&*doc, &writer);
                 let retyped = store::cast_path(doc, &path, to, timestamps)
                     .map_err(|e| store_status(e, "Failed to cast a store value"))?;
+                registry_kept(&*doc, &path, before, &writer)?;
                 let read = read_shown(&*doc, &path, accept.as_ref(), false, &Access::Everything)?;
                 Ok::<_, ApiError>((retyped, read))
             })
@@ -763,22 +837,23 @@ pub(crate) async fn post_store_entry(
     }
     let index = path.last().and_then(|leaf| store::list_index(leaf));
     let value = body_value(&headers, &body)?;
-    let inserted = state
+    let writer = user.0.clone();
+    let ((), heads) = state
         .mutate_doc_for(Some(user.0.clone()), move |doc| {
+            let before = Before::of(&*doc, &writer);
             store::insert_val_path(doc, &path, &value.value)
+                .map_err(|e| store_status(e, "Failed to insert into a list"))?;
+            registry_kept(&*doc, &path, before, &writer)
         })
-        .await;
-    match inserted {
-        Ok(((), heads)) => Ok(with_heads(
-            (
-                StatusCode::CREATED,
-                axum::Json(serde_json::json!({ "index": index })),
-            )
-                .into_response(),
-            &heads,
-        )),
-        Err(error) => Err(store_status(error, "Failed to insert into a list")),
-    }
+        .await?;
+    Ok(with_heads(
+        (
+            StatusCode::CREATED,
+            axum::Json(serde_json::json!({ "index": index })),
+        )
+            .into_response(),
+        &heads,
+    ))
 }
 
 /// `DELETE /store/{key}`.
@@ -813,12 +888,15 @@ pub(crate) async fn delete_store_entry(
             .find(|depth| access.allows(crate::access::Right::Write, &path[..*depth]))
             .unwrap_or(path.len())
     });
+    let writer = user.0.clone();
     let ((), heads) = state
         .mutate_doc_for(Some(user.0.clone()), move |doc| {
+            let before = Before::of(&*doc, &writer);
             store::delete_path_pruning(doc, &path, prune)
+                .map_err(|e| store_status(e, "Failed to delete store value"))?;
+            registry_kept(&*doc, &path, before, &writer)
         })
-        .await
-        .map_err(|e| store_status(e, "Failed to delete store value"))?;
+        .await?;
 
     Ok(with_heads(StatusCode::NO_CONTENT.into_response(), &heads))
 }