claw-store/src/serve.rs

@@ -716,7 +716,7 @@ pub fn build_app_with_v2(
         .ok();
     let mut v2_router: Router<()> = Router::new();
     if let Some(v2s) = v2_state {
-        v2_router = v2_router.merge(crate::serve_v2::routes().with_state(v2s));
+        v2_router = v2_router.merge(crate::serve_v2::build(v2s));
     }
     if let Some(dir) = v2_static_dir {
         v2_router = v2_router.nest_service(

claw-store/src/serve_v2.rs

@@ -14,17 +14,20 @@
 
 use axum::{
     extract::{Path, State},
-    routing::get,
+    http::StatusCode,
+    routing::{get, post},
     Json, Router,
 };
 use serde::{Deserialize, Serialize};
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
+use tokio::task::JoinSet;
 
 use crate::cluster::rpc::{
-    call_dashboard_status, call_dashboard_storage, CacheSummary, DashboardStatusReply,
-    DashboardStorageReply, FilesystemUsage, HotTierUsage, MountStatus, TimerStatus,
+    call_delete_tag, call_put_tag, call_dashboard_status, call_dashboard_storage, CacheSummary,
+    DashboardStatusReply, DashboardStorageReply, FilesystemUsage, HotTierUsage, MountStatus,
+    TimerStatus,
 };
 use crate::cluster::transport::{NodeIdentity, QuicClient};
 use crate::config::{Config, PeerEntry};
@@ -45,6 +48,12 @@ pub struct V2State {
     /// (gossip port, not RPC). We assume RPC = gossip + 1 in that
     /// case unless the operator sets an explicit `rpc_addr`.
     pub default_rpc_port_offset: u16,
+    /// Bearer token required on mutating endpoints (POST/DELETE).
+    /// GET stays open — the dashboard loads without credentials.
+    /// 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).
+    pub api_token: Option<String>,
 }
 
 impl V2State {
@@ -66,6 +75,7 @@ impl V2State {
             peers: cluster.peers.clone(),
             client: std::sync::Arc::new(client),
             default_rpc_port_offset: 1,
+            api_token: cfg.api_token.clone(),
         })
     }
 
@@ -469,13 +479,214 @@ impl V2State {
             peers: self.peers.clone(),
             client: self.client.clone(),
             default_rpc_port_offset: self.default_rpc_port_offset,
+            api_token: self.api_token.clone(),
         }
     }
 }
 
+// ── write-through fan-out (Phase 9 F1) ──────────────────────────
+//
+// The aggregator is the single choke-point for cross-fleet writes,
+// so external callers (clawmates, gitea runners, ops tooling) speak
+// one URL over ordinary HTTPS instead of minting a fleet-CA leaf
+// cert per app. The aggregator holds a leaf cert and dials every
+// peer over the existing mTLS QUIC transport.
+
+fn decode_blob_id(hex: &str) -> Option<[u8; 32]> {
+    if hex.len() != 64 {
+        return None;
+    }
+    let mut out = [0u8; 32];
+    for i in 0..32 {
+        let byte = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16).ok()?;
+        out[i] = byte;
+    }
+    Some(out)
+}
+
+#[derive(Deserialize)]
+pub struct PutTagBody {
+    /// 64-char hex-encoded BlobId (blake3 hash) to bind the tag to.
+    pub blob_id: String,
+}
+
+#[derive(Serialize)]
+pub struct PeerResult {
+    pub peer: String,
+    pub ok: bool,
+    pub error: Option<String>,
+}
+
+#[derive(Serialize)]
+pub struct FanoutReply {
+    pub tag: String,
+    pub peers: Vec<PeerResult>,
+    /// Aggregate: `true` iff every peer succeeded.
+    pub all_ok: bool,
+}
+
+/// `POST /api/v2/tags/:name` — bind a tag to a blob_id on **every**
+/// peer concurrently. Callers get a per-peer result so partial
+/// failures are diagnosable (single peer down doesn't hide the
+/// success of the others).
+async fn handle_put_tag(
+    Path(name): Path<String>,
+    State(s): State<Arc<V2State>>,
+    Json(body): Json<PutTagBody>,
+) -> Result<Json<FanoutReply>, (StatusCode, String)> {
+    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 {
+        tag: name,
+        peers: results,
+        all_ok,
+    }))
+}
+
+/// `DELETE /api/v2/tags/:name` — remove a tag on every peer.
+/// Per-peer `ok=true` covers both "deleted" and "was already
+/// absent" (the underlying RPC returns `Ok(false)` for the latter;
+/// we still surface success — reversing a nonexistent tag is a
+/// no-op, not an error).
+async fn handle_delete_tag(
+    Path(name): Path<String>,
+    State(s): State<Arc<V2State>>,
+) -> Result<Json<FanoutReply>, (StatusCode, String)> {
+    if name.is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "tag name cannot be empty".to_string()));
+    }
+    let results = fanout_delete_tag(&s, &name).await;
+    let all_ok = results.iter().all(|r| r.ok);
+    Ok(Json(FanoutReply {
+        tag: name,
+        peers: results,
+        all_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 {
+        let peer = peer.clone();
+        let name = name.to_string();
+        let value = *value;
+        let s = s.clone_shallow();
+        set.spawn(async move {
+            let peer_name = peer.name.clone();
+            let res = async {
+                let conn = s.dial(&peer).await?;
+                let r = call_put_tag(&conn, &name, &value).await;
+                conn.close(quinn::VarInt::from_u32(0), b"done");
+                r
+            }
+            .await;
+            match res {
+                Ok(()) => PeerResult { peer: peer_name, ok: true, error: None },
+                Err(e) => PeerResult { peer: peer_name, ok: false, error: Some(e.to_string()) },
+            }
+        });
+    }
+    let mut out = Vec::new();
+    while let Some(joined) = set.join_next().await {
+        match joined {
+            Ok(r) => out.push(r),
+            Err(e) => out.push(PeerResult {
+                peer: "<join-error>".into(),
+                ok: false,
+                error: Some(e.to_string()),
+            }),
+        }
+    }
+    // Deterministic ordering makes the response diff-friendly and
+    // lets integration tests assert on shape without sorting.
+    out.sort_by(|a, b| a.peer.cmp(&b.peer));
+    out
+}
+
+async fn fanout_delete_tag(s: &V2State, name: &str) -> Vec<PeerResult> {
+    let mut set = JoinSet::new();
+    for peer in &s.peers {
+        let peer = peer.clone();
+        let name = name.to_string();
+        let s = s.clone_shallow();
+        set.spawn(async move {
+            let peer_name = peer.name.clone();
+            let res = async {
+                let conn = s.dial(&peer).await?;
+                // Discard the bool — "was already gone" is not an error
+                // at the fan-out layer (reversing an absent tag = no-op).
+                let _ = call_delete_tag(&conn, &name).await?;
+                conn.close(quinn::VarInt::from_u32(0), b"done");
+                Ok::<(), anyhow::Error>(())
+            }
+            .await;
+            match res {
+                Ok(()) => PeerResult { peer: peer_name, ok: true, error: None },
+                Err(e) => PeerResult { peer: peer_name, ok: false, error: Some(e.to_string()) },
+            }
+        });
+    }
+    let mut out = Vec::new();
+    while let Some(joined) = set.join_next().await {
+        match joined {
+            Ok(r) => out.push(r),
+            Err(e) => out.push(PeerResult {
+                peer: "<join-error>".into(),
+                ok: false,
+                error: Some(e.to_string()),
+            }),
+        }
+    }
+    out.sort_by(|a, b| a.peer.cmp(&b.peer));
+    out
+}
+
+// ── 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.
+async fn v2_auth(
+    State(s): State<Arc<V2State>>,
+    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();
+            }
+        }
+    }
+    next.run(request).await
+}
+
 // ── route registration ──────────────────────────────────────────
 
-pub fn routes() -> Router<Arc<V2State>> {
+/// Assemble the aggregator router with state and middleware baked in.
+/// The auth middleware needs the concrete `Arc<V2State>` at layer time
+/// (so it can read `api_token`), which is why this returns a fully-
+/// stated `Router<()>` instead of a state-generic router.
+pub fn build(state: Arc<V2State>) -> Router {
     Router::new()
         .route("/api/v2/fleet", get(handle_fleet))
         .route("/api/v2/node/:name/status", get(handle_node_status))
@@ -485,4 +696,10 @@ pub fn routes() -> Router<Arc<V2State>> {
         .route("/api/v2/storage/snapshots", get(handle_snapshots))
         .route("/api/v2/storage/ref-tracking", get(handle_ref_tracking))
         .route("/api/v2/projects", get(handle_projects))
+        .route(
+            "/api/v2/tags/:name",
+            post(handle_put_tag).delete(handle_delete_tag),
+        )
+        .route_layer(axum::middleware::from_fn_with_state(state.clone(), v2_auth))
+        .with_state(state)
 }