Phase 9 F4: namespaced tokens for multi-tenant aggregator (#104)

claw-store/src/serve_v2.rs

@@ -30,7 +30,7 @@ use crate::cluster::rpc::{
     TimerStatus,
 };
 use crate::cluster::transport::{NodeIdentity, QuicClient};
-use crate::config::{Config, PeerEntry};
+use crate::config::{Config, PeerEntry, TokenEntry};
 
 /// Aggregator runtime: one QuicClient, one peer list, one identity.
 ///
@@ -53,7 +53,16 @@ pub struct V2State {
     /// Sourced from the aggregator's own `config.toml` `api_token`.
     /// `None` disables the check (only appropriate on a trusted LAN,
     /// e.g. Tailscale-only + loopback bind).
+    ///
+    /// Treated as an **admin** token — no namespace constraint. For
+    /// multi-tenant use, prefer the per-app tokens in `token_entries`.
     pub api_token: Option<String>,
+
+    /// Per-app tokens with optional namespace scoping (Phase 9 F4).
+    /// Merged with `api_token`: `api_token` is admin (no namespace);
+    /// entries here can be admin (no `namespace`) or scoped (must
+    /// match `<namespace>:*` on tag names). First match wins.
+    pub token_entries: Vec<TokenEntry>,
 }
 
 impl V2State {
@@ -76,6 +85,11 @@ impl V2State {
             client: std::sync::Arc::new(client),
             default_rpc_port_offset: 1,
             api_token: cfg.api_token.clone(),
+            token_entries: cfg
+                .aggregator
+                .as_ref()
+                .map(|a| a.tokens.clone())
+                .unwrap_or_default(),
         })
     }
 
@@ -480,6 +494,38 @@ impl V2State {
             client: self.client.clone(),
             default_rpc_port_offset: self.default_rpc_port_offset,
             api_token: self.api_token.clone(),
+            token_entries: self.token_entries.clone(),
+        }
+    }
+}
+
+/// The identity attached to an authenticated request. Either an admin
+/// (unrestricted) or a per-app token bound to a `namespace` prefix.
+/// Handlers use this to decide whether a tag name is in-scope.
+#[derive(Debug, Clone)]
+pub enum AuthedCaller {
+    /// Any tag name allowed. Sourced from either `api_token` (legacy
+    /// admin) or a `[[aggregator.tokens]]` entry with no namespace.
+    Admin,
+    /// Restricted to tags whose name is `<namespace>:<anything>`.
+    Namespaced { namespace: String },
+    /// No auth was configured on the aggregator at all. Trusted-LAN
+    /// mode — everything allowed, no bearer required.
+    Open,
+}
+
+impl AuthedCaller {
+    /// Returns `true` iff this caller may pin/unpin/list a tag with
+    /// this name. Admin + Open allow anything; namespaced callers
+    /// require an exact `<namespace>:` prefix (with the separator,
+    /// so `workspace:42` cannot silently reach `workspace:420:*`).
+    pub fn may_touch_tag(&self, tag: &str) -> bool {
+        match self {
+            AuthedCaller::Admin | AuthedCaller::Open => true,
+            AuthedCaller::Namespaced { namespace } => {
+                let prefix = format!("{namespace}:");
+                tag.starts_with(&prefix)
+            }
         }
     }
 }
@@ -532,13 +578,12 @@ pub struct FanoutReply {
 async fn handle_put_tag(
     Path(name): Path<String>,
     State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
     Json(body): Json<PutTagBody>,
 ) -> Result<Json<FanoutReply>, (StatusCode, String)> {
+    check_tag_scope(&caller, &name)?;
     let value = decode_blob_id(&body.blob_id)
         .ok_or_else(|| (StatusCode::BAD_REQUEST, "blob_id must be 64-char hex".to_string()))?;
-    if name.is_empty() {
-        return Err((StatusCode::BAD_REQUEST, "tag name cannot be empty".to_string()));
-    }
     let results = fanout_put_tag(&s, &name, &value).await;
     let all_ok = results.iter().all(|r| r.ok);
     Ok(Json(FanoutReply {
@@ -556,10 +601,9 @@ async fn handle_put_tag(
 async fn handle_delete_tag(
     Path(name): Path<String>,
     State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
 ) -> Result<Json<FanoutReply>, (StatusCode, String)> {
-    if name.is_empty() {
-        return Err((StatusCode::BAD_REQUEST, "tag name cannot be empty".to_string()));
-    }
+    check_tag_scope(&caller, &name)?;
     let results = fanout_delete_tag(&s, &name).await;
     let all_ok = results.iter().all(|r| r.ok);
     Ok(Json(FanoutReply {
@@ -569,6 +613,26 @@ async fn handle_delete_tag(
     }))
 }
 
+/// Combined name-shape + namespace check. Empty name → 400; out-of-
+/// namespace → 403 with a clear message so multi-tenant callers can
+/// tell "you can't touch that" from "your token was bogus" (401).
+fn check_tag_scope(caller: &AuthedCaller, name: &str) -> Result<(), (StatusCode, String)> {
+    if name.is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "tag name cannot be empty".into()));
+    }
+    if !caller.may_touch_tag(name) {
+        let ns = match caller {
+            AuthedCaller::Namespaced { namespace } => namespace.as_str(),
+            _ => "(none)",
+        };
+        return Err((
+            StatusCode::FORBIDDEN,
+            format!("tag \"{name}\" is outside your namespace \"{ns}:\""),
+        ));
+    }
+    Ok(())
+}
+
 async fn fanout_put_tag(s: &V2State, name: &str, value: &[u8; 32]) -> Vec<PeerResult> {
     let mut set = JoinSet::new();
     for peer in &s.peers {
@@ -648,33 +712,67 @@ async fn fanout_delete_tag(s: &V2State, name: &str) -> Vec<PeerResult> {
 
 // ── auth middleware ─────────────────────────────────────────────
 
-/// If the aggregator's config sets `api_token`, mutating methods
-/// (POST/DELETE) must present a matching `Authorization: Bearer …`
-/// header. GET is always open so the dashboard loads unmodified.
+/// Resolve the request's bearer against the aggregator's known
+/// tokens and produce an [`AuthedCaller`]. On GET/HEAD, we still
+/// resolve so filtered list handlers can scope by namespace — but
+/// missing/bad auth just yields [`AuthedCaller::Open`] to preserve
+/// the "dashboard loads without credentials" property; open callers
+/// only see unrestricted views, so nothing sensitive leaks.
+fn resolve_caller(s: &V2State, headers: &axum::http::HeaderMap) -> Option<AuthedCaller> {
+    // No auth configured anywhere → trusted-LAN mode.
+    if s.api_token.is_none() && s.token_entries.is_empty() {
+        return Some(AuthedCaller::Open);
+    }
+    let provided = headers
+        .get(axum::http::header::AUTHORIZATION)
+        .and_then(|v| v.to_str().ok())
+        .and_then(|v| v.strip_prefix("Bearer "))?;
+    if let Some(admin) = &s.api_token {
+        if provided == admin.as_str() {
+            return Some(AuthedCaller::Admin);
+        }
+    }
+    for entry in &s.token_entries {
+        if provided == entry.token.as_str() {
+            return Some(match &entry.namespace {
+                Some(ns) => AuthedCaller::Namespaced { namespace: ns.clone() },
+                None => AuthedCaller::Admin,
+            });
+        }
+    }
+    None
+}
+
+/// Middleware: on mutating methods, require a resolved caller and
+/// stash it in request extensions for handlers. GET/HEAD stays open
+/// (see `resolve_caller` for the rationale).
 async fn v2_auth(
     State(s): State<Arc<V2State>>,
-    request: axum::extract::Request,
+    mut request: axum::extract::Request,
     next: axum::middleware::Next,
 ) -> axum::response::Response {
     let method = request.method().clone();
-    let needs_auth =
-        method != axum::http::Method::GET && method != axum::http::Method::HEAD;
-    if needs_auth {
-        if let Some(expected) = &s.api_token {
-            let provided = request
-                .headers()
-                .get(axum::http::header::AUTHORIZATION)
-                .and_then(|v| v.to_str().ok())
-                .and_then(|v| v.strip_prefix("Bearer "));
-            if provided != Some(expected.as_str()) {
-                return axum::response::Response::builder()
-                    .status(StatusCode::UNAUTHORIZED)
-                    .header("content-type", "application/json")
-                    .body(axum::body::Body::from(
-                        r#"{"error":"unauthorized"}"#,
-                    ))
-                    .unwrap_or_default();
-            }
+    let is_read = method == axum::http::Method::GET || method == axum::http::Method::HEAD;
+    let caller = resolve_caller(&s, request.headers());
+    match (is_read, caller) {
+        // Reads always pass; auth is best-effort for future
+        // namespace-scoped list filtering.
+        (true, Some(c)) => {
+            request.extensions_mut().insert(c);
+        }
+        (true, None) => {
+            request.extensions_mut().insert(AuthedCaller::Open);
+        }
+        // Writes require a resolved caller.
+        (false, Some(c)) => {
+            request.extensions_mut().insert(c);
+        }
+        (false, None) => {
+            return axum::response::Response::builder()
+                .status(StatusCode::UNAUTHORIZED)
+                .header("content-type", "application/json")
+                .body(axum::body::Body::from(r#"{"error":"unauthorized"}"#))
+                .unwrap_or_default();
         }
     }
     next.run(request).await