1 files changed, 109 insertions, 30 deletions

diff --git a/src/api/store.rs b/src/api/store.rs
index f6cf6fd..1ab3951 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -18,6 +18,7 @@ use sha2::{Digest, Sha256};
 use tokio::sync::broadcast;
 use tracing::{debug, error, instrument, warn};
 
+use super::access::{Access, Visible, prune, prune_shallow};
 use super::auth::AuthenticatedUser;
 use super::{ApiError, store_status};
 use crate::document::meta;
@@ -100,15 +101,17 @@ pub(crate) struct DeleteQuery {
 /// `GET /store`: the whole store (`docs/openapi.yaml`).
 pub(crate) async fn get_store_root(
     State(state): State<AppState>,
+    Extension(access): Extension<Access>,
     headers: HeaderMap,
     Query(query): Query<ReadQuery>,
 ) -> Result<Response, ApiError> {
-    read(&state, Vec::new(), &headers, &query).await
+    read(&state, Vec::new(), &headers, &query, access).await
 }
 
 /// `GET /store/{key}`.
 pub(crate) async fn get_store_entry(
     State(state): State<AppState>,
+    Extension(access): Extension<Access>,
     Path(key): Path<String>,
     headers: HeaderMap,
     Query(query): Query<ReadQuery>,
@@ -121,21 +124,35 @@ pub(crate) async fn get_store_entry(
     {
         return Ok(redirect);
     }
-    read(&state, path, &headers, &query).await
+    read(&state, path, &headers, &query, access).await
 }
 
-/// A read of `path` as `query` asks (the empty path: the whole store). The root is read
-/// while not ready too (`render_read`'s `unready_live`), and is never missing: an empty
-/// store is an empty map.
+/// `403` where the user may read nothing of `path` (`Access::visible`).
+fn readable(access: &Access, path: &[String]) -> Result<(), ApiError> {
+    if access.visible(path) == Visible::Nothing {
+        return Err(ApiError(
+            StatusCode::FORBIDDEN,
+            Some("no capability to read this (docs/access.md)".to_owned()),
+        ));
+    }
+    Ok(())
+}
+
+/// A read of `path` as `query` asks (the empty path: the whole store), as much of it as
+/// `access` lets its user read (`read_shown`). The root is read while not ready too
+/// (`render_read`'s `unready_live`), and is never missing: an empty store is an empty
+/// map.
 async fn read(
     state: &AppState,
     path: Vec<String>,
     headers: &HeaderMap,
     query: &ReadQuery,
+    access: Access,
 ) -> Result<Response, ApiError> {
+    readable(&access, &path)?;
     let root = path.is_empty();
     if query.shallow {
-        return shallow(state, path, root, query).await;
+        return shallow(state, path, root, query, access).await;
     }
     let accept = headers.get(ACCEPT).cloned();
     let want_conflicts = query.conflicts;
@@ -150,7 +167,8 @@ async fn read(
                     doc,
                     &path,
                     accept.as_ref(),
-                    want_conflicts
+                    want_conflicts,
+                    &access
                 ))
             },
             move |result| match result {
@@ -194,12 +212,15 @@ impl Read {
     }
 }
 
-/// What `path` holds, read to show as `accept` asks; `406` if it can't be.
+/// What `path` holds, read to show as `accept` asks; `406` if it can't be. Of a place
+/// `access` lets its user read only partly, what it may (`access::prune`), as JSON;
+/// `403` for a native form, or if nothing of it is.
 fn read_shown(
     doc: &impl crate::document::ReadDoc,
     path: &[String],
     accept: Option<&HeaderValue>,
     conflicts: bool,
+    access: &Access,
 ) -> Result<Option<Read>, ApiError> {
     let failed = |e| store_status(e, "Failed to read store path");
     let Some(held) = store::read_held(doc, path).map_err(failed)? else {
@@ -224,6 +245,25 @@ fn read_shown(
             None,
         ),
     };
+    let (json, conflicts) = match (access.visible(path), shown) {
+        (Visible::All, _) => (json, conflicts),
+        (_, Shown::Native(_)) => return Err(StatusCode::FORBIDDEN.into()),
+        (_, shown) => {
+            let typed = matches!(shown, Shown::Json(form) if form.typed);
+            let json = match json {
+                Some(json) => Some(
+                    prune(json, &mut path.to_vec(), access, typed)
+                        .ok_or(ApiError(StatusCode::FORBIDDEN, None))?,
+                ),
+                None => None,
+            };
+            let conflicts = conflicts.map(|mut conflicts: Vec<store::ConflictReport>| {
+                conflicts.retain(|report| access.visible(&report.path) != Visible::Nothing);
+                conflicts
+            });
+            (json, conflicts)
+        }
+    };
     Ok(Some(Read {
         held,
         shown,
@@ -239,6 +279,7 @@ async fn shallow(
     path: Vec<String>,
     unready_live: bool,
     query: &ReadQuery,
+    access: Access,
 ) -> Result<Response, ApiError> {
     if query.conflicts {
         return Err(StatusCode::BAD_REQUEST.into());
@@ -248,7 +289,12 @@ async fn shallow(
         .render_read(
             unready_live,
             query.heads,
-            move |doc| read_at!(doc, at.as_deref(), |doc| store::read_shallow(doc, &path)),
+            move |doc| {
+                read_at!(doc, at.as_deref(), |doc| store::read_shallow(doc, &path)
+                    .map(
+                        |value| value.map(|value| prune_shallow(value, &path, &access))
+                    ))
+            },
             move |result| {
                 let value = result
                     .map_err(|e| store_status(e, "Failed to read store path"))?
@@ -319,6 +365,7 @@ const MAX_RETYPED_LISTED: usize = 16;
 pub(crate) async fn put_store_entry(
     State(state): State<AppState>,
     Extension(user): Extension<AuthenticatedUser>,
+    Extension(access): Extension<Access>,
     Path(key): Path<String>,
     Query(query): Query<PutQuery>,
     RawQuery(raw_query): RawQuery,
@@ -329,7 +376,7 @@ pub(crate) async fn put_store_entry(
         return Err(StatusCode::SERVICE_UNAVAILABLE.into());
     }
 
-    let path = parse_write_path(&state, &user, &key, "PUT")?;
+    let path = parse_write_path(&state, &user, &access, &key, "PUT")?;
     if let Some(redirect) = to_positive(&state, &path, false, None, raw_query.as_deref()).await? {
         return Ok(redirect);
     }
@@ -342,7 +389,8 @@ pub(crate) async fn put_store_entry(
         .mutate_doc_for(Some(user.0.clone()), move |doc| {
             let retyped = store::write_val_path(doc, &path, &body.value, writing)
                 .map_err(|e| store_status(e, "Failed to write store value"))?;
-            let read = read_shown(&*doc, &path, accept.as_ref(), false)?;
+            // 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))
         })
         .await?;
@@ -383,25 +431,33 @@ fn parse_key_path(key: &str) -> Result<Vec<String>, StatusCode> {
     Ok(segments)
 }
 
-/// A path for a write. Protocol data (`_continuum`: remake records, members, forget
-/// records, epoch metadata) is what every node acts on, so a write there acts on the
-/// whole swarm: only users listed in `--protocol-writers` may, and each such write is
+/// A path for a write, if `access` lets the user write there (`403` if not). In a swarm
+/// with a user registry, by the user's capabilities (`access.md`); in one without, the
+/// swarm's data by anyone, protocol data (`_continuum`: remake records, members, forget
+/// records, the users) only by users listed in `--protocol-writers`. Protocol data is
+/// what every node acts on, so a write there acts on the whole swarm, and each is
 /// logged. Epoch metadata (and `_continuum` as a whole, which contains it) is never
 /// writable: the protocol reads it at the snapshot change, so an edit only misleads
-/// whoever reads it later. `403` otherwise.
+/// whoever reads it later; nor is the swarm ID.
 fn parse_write_path(
     state: &AppState,
     user: &AuthenticatedUser,
+    access: &Access,
     key: &str,
     method: &str,
 ) -> Result<Vec<String>, StatusCode> {
     let path = parse_key_path(key)?;
-    if !meta::is_protocol(&path) {
+    let protocol = meta::is_protocol(&path);
+    // `_continuum` itself, epoch metadata and the swarm ID: nobody.
+    let fixed = protocol && (path.len() < 2 || path[1] == meta::EPOCHS || path[1] == meta::SWARM);
+    let allowed = match access {
+        Access::Everything => !protocol || state.protocol_writers.contains(&user.0),
+        Access::Caps(caps) => caps.allows(crate::access::Right::Write, &path),
+    };
+    if !protocol && allowed {
         return Ok(path);
     }
-    // `_continuum` itself, epoch metadata and the swarm ID: nobody.
-    let fixed = path.len() < 2 || path[1] == meta::EPOCHS || path[1] == meta::SWARM;
-    if fixed || !state.protocol_writers.contains(&user.0) {
+    if fixed || !allowed {
         warn!(user = %user.0, method, path = %key, "Refused a write to protocol data");
         return Err(StatusCode::FORBIDDEN);
     }
@@ -636,6 +692,7 @@ fn body_value(headers: &HeaderMap, body: &Bytes) -> Result<WriteBody, ApiError>
 pub(crate) async fn post_store_entry(
     State(state): State<AppState>,
     Extension(user): Extension<AuthenticatedUser>,
+    Extension(access): Extension<Access>,
     Path(key): Path<String>,
     Query(query): Query<PostQuery>,
     RawQuery(raw_query): RawQuery,
@@ -645,7 +702,7 @@ pub(crate) async fn post_store_entry(
     if !state.is_ready().await {
         return Err(StatusCode::SERVICE_UNAVAILABLE.into());
     }
-    let path = parse_write_path(&state, &user, &key, "POST")?;
+    let path = parse_write_path(&state, &user, &access, &key, "POST")?;
     if query.increment.is_some() && query.cast.is_some() {
         return Err(ApiError(
             StatusCode::BAD_REQUEST,
@@ -697,7 +754,7 @@ pub(crate) async fn post_store_entry(
             .mutate_doc_for(Some(user.0.clone()), move |doc| {
                 let retyped = store::cast_path(doc, &path, to, timestamps)
                     .map_err(|e| store_status(e, "Failed to cast a store value"))?;
-                let read = read_shown(&*doc, &path, accept.as_ref(), false)?;
+                let read = read_shown(&*doc, &path, accept.as_ref(), false, &Access::Everything)?;
                 Ok::<_, ApiError>((retyped, read))
             })
             .await?;
@@ -729,11 +786,12 @@ pub(crate) async fn post_store_entry(
 pub(crate) async fn delete_store_entry(
     State(state): State<AppState>,
     Extension(user): Extension<AuthenticatedUser>,
+    Extension(access): Extension<Access>,
     Path(key): Path<String>,
     Query(query): Query<DeleteQuery>,
     RawQuery(raw_query): RawQuery,
 ) -> Result<Response, ApiError> {
-    let path = parse_write_path(&state, &user, &key, "DELETE")?;
+    let path = parse_write_path(&state, &user, &access, &key, "DELETE")?;
 
     tracing::Span::current().record("path", path.join("/").as_str());
 
@@ -749,10 +807,15 @@ pub(crate) async fn delete_store_entry(
         return Ok(redirect);
     }
 
-    let prune = query.prune;
+    // Pruning stops at the highest map the user may write: none above it goes.
+    let prune = query.prune.then(|| {
+        (1..=path.len())
+            .find(|depth| access.allows(crate::access::Right::Write, &path[..*depth]))
+            .unwrap_or(path.len())
+    });
     let ((), heads) = state
         .mutate_doc_for(Some(user.0.clone()), move |doc| {
-            store::delete_path(doc, &path, prune)
+            store::delete_path_pruning(doc, &path, prune)
         })
         .await
         .map_err(|e| store_status(e, "Failed to delete store value"))?;
@@ -832,22 +895,33 @@ async fn render_traced(
 
 /// `GET /stream`: follows the whole store, as `GET /store` reads it (events have an
 /// empty `key`).
-pub(crate) async fn stream_store_root(State(state): State<AppState>) -> Result<Response, ApiError> {
-    stream(state, String::new(), Vec::new()).await
+pub(crate) async fn stream_store_root(
+    State(state): State<AppState>,
+    Extension(access): Extension<Access>,
+) -> Result<Response, ApiError> {
+    stream(state, String::new(), Vec::new(), access).await
 }
 
 /// `GET /stream/{key}`: follows one value.
 pub(crate) async fn stream_store_entry(
     State(state): State<AppState>,
+    Extension(access): Extension<Access>,
     Path(key): Path<String>,
 ) -> Result<Response, ApiError> {
     let path = parse_key_path(&key)?;
-    stream(state, key, path).await
+    stream(state, key, path, access).await
 }
 
 /// Newline-delimited JSON events for the value at `path` (`key` as the client wrote it):
-/// the current value, then one whenever the resolved value changes.
-async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Response, ApiError> {
+/// the current value, then one whenever the resolved value changes; as much of it as
+/// `access` lets its user read, as the stream opened (`access::prune`).
+async fn stream(
+    state: AppState,
+    key: String,
+    path: Vec<String>,
+    access: Access,
+) -> Result<Response, ApiError> {
+    readable(&access, &path)?;
     if !state.is_ready().await {
         return Err(StatusCode::SERVICE_UNAVAILABLE.into());
     }
@@ -855,8 +929,13 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo
     // Subscribe first, so nothing between the initial read and the loop is missed.
     let mut changes = state.change_events.subscribe();
 
+    let shown = move |(value, conflicted): (Option<JsonValue>, bool), path: &[String]| {
+        let value = value.and_then(|value| prune(value, &mut path.to_vec(), &access, false));
+        (value, conflicted)
+    };
     let initial = render_traced(&state, path.clone())
         .await
+        .map(|read| shown(read, &path))
         .map_err(|e| store_status(e, "Failed to read initial stream value"))?;
 
     let stream_state = state.clone();
@@ -908,7 +987,7 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo
             }
 
             let (next_value, next_conflicted) = match render_traced(&stream_state, path.clone()).await {
-                Ok(result) => result,
+                Ok(result) => shown(result, &path),
                 Err(error) => {
                     error!(error = %format!("{error:#}"), key = %stream_key, "Failed to read updated stream value");
                     yield Err(IoError::other("Failed to read updated store value"));