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.

2 files changed, 104 insertions, 19 deletions

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))
 }