rust volume: mirror Go volume server logic to gate the admin RPCs (#10748)

rust volume: gate the remaining admin RPCs behind check_grpc_admin_auth

The Go volume server gates 29 destructive VolumeServer RPCs on the
-whiteList admin check; the Rust port only gated 14. Add the gate to the
other 15 -- batch_delete, read_all_needles, fetch_and_write_needle, the
EC-shard generate/rebuild/copy/unmount/to-volume RPCs, both tier-move RPCs,
volume_copy, volume_tail_receiver, set_state, scrub_ec_volume and
volume_needle_status -- so a configured whitelist restricts them the same
way it already does on the Go side.

check_grpc_admin_auth also required peer info before checking whether any
control was configured, unlike Go's `if vs.guard == nil { return nil }`.
Short-circuit when no whitelist and no signing key are set, so in-process
callers keep working with security inactive and only the gate ordering
changes for configured servers.

tests/admin_auth_coverage.rs mirrors the Go coverage test: every handler
must either gate or be listed as intentionally open with a reason, so the
two implementations can't silently drift apart again.

seaweed-volume/src/security.rs

@@ -192,6 +192,13 @@ impl Guard {
         self.is_write_active = !is_empty_whitelist || !self.signing_key.is_empty();
     }
 
+    /// Whether any write-side control is configured: a non-empty whitelist or a
+    /// signing key. Mirrors Go's non-nil `guard` -- when this is false, every
+    /// caller is allowed and admin RPCs need not carry peer info.
+    pub fn is_write_active(&self) -> bool {
+        self.is_write_active
+    }
+
     /// Check if a remote IP is in the whitelist.
     /// Returns true if write security is inactive (no whitelist and no signing key),
     /// if the whitelist is empty, or if the IP matches.

seaweed-volume/src/server/grpc_server.rs

@@ -155,6 +155,13 @@ impl VolumeGrpcService {
     /// SocketAddr; if it is somehow None we deny, matching "if we don't
     /// know who the caller is, refuse."
     fn check_grpc_admin_auth<T>(&self, request: &Request<T>) -> Result<(), Status> {
+        // Mirror Go's `if vs.guard == nil { return nil }`: with no whitelist and
+        // no signing key, write security is inactive and every caller is allowed.
+        // Real gRPC connections always carry peer info, so requiring it below only
+        // affects in-process callers (tests, upgrades) once a control is enabled.
+        if !self.state.guard.read().unwrap().is_write_active() {
+            return Ok(());
+        }
         let remote = match request.remote_addr() {
             Some(addr) => addr,
             None => {
@@ -293,6 +300,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::BatchDeleteRequest>,
     ) -> Result<Response<volume_server_pb::BatchDeleteResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         self.state.check_maintenance()?;
         let req = request.into_inner();
         let mut results = Vec::new();
@@ -1135,6 +1143,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::SetStateRequest>,
     ) -> Result<Response<volume_server_pb::SetStateResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         let req = request.into_inner();
 
         if let Some(new_state) = &req.state {
@@ -1187,6 +1196,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeCopyRequest>,
     ) -> Result<Response<Self::VolumeCopyStream>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         self.state.check_maintenance()?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
@@ -1988,6 +1998,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::ReadAllNeedlesRequest>,
     ) -> Result<Response<Self::ReadAllNeedlesStream>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         let req = request.into_inner();
         let state = self.state.clone();
 
@@ -2188,6 +2199,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeTailReceiverRequest>,
     ) -> Result<Response<volume_server_pb::VolumeTailReceiverResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
 
@@ -2313,6 +2325,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeEcShardsGenerateRequest>,
     ) -> Result<Response<volume_server_pb::VolumeEcShardsGenerateResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         self.state.check_maintenance()?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
@@ -2413,6 +2426,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeEcShardsRebuildRequest>,
     ) -> Result<Response<volume_server_pb::VolumeEcShardsRebuildResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         self.state.check_maintenance()?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
@@ -2613,6 +2627,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeEcShardsCopyRequest>,
     ) -> Result<Response<volume_server_pb::VolumeEcShardsCopyResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         self.state.check_maintenance()?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
@@ -2995,6 +3010,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeEcShardsUnmountRequest>,
     ) -> Result<Response<volume_server_pb::VolumeEcShardsUnmountResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
 
@@ -3158,6 +3174,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeEcShardsToVolumeRequest>,
     ) -> Result<Response<volume_server_pb::VolumeEcShardsToVolumeResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         self.state.check_maintenance()?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
@@ -3424,6 +3441,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeTierMoveDatToRemoteRequest>,
     ) -> Result<Response<Self::VolumeTierMoveDatToRemoteStream>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         self.state.check_maintenance()?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
@@ -3590,6 +3608,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeTierMoveDatFromRemoteRequest>,
     ) -> Result<Response<Self::VolumeTierMoveDatFromRemoteStream>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         // Note: Go does NOT check maintenance mode for TierMoveDatFromRemote
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
@@ -3850,6 +3869,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::FetchAndWriteNeedleRequest>,
     ) -> Result<Response<volume_server_pb::FetchAndWriteNeedleResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         self.state.check_maintenance()?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
@@ -4105,6 +4125,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::ScrubEcVolumeRequest>,
     ) -> Result<Response<volume_server_pb::ScrubEcVolumeResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         let req = request.into_inner();
 
         // Validate mode
@@ -4462,6 +4483,7 @@ impl VolumeServer for VolumeGrpcService {
         &self,
         request: Request<volume_server_pb::VolumeNeedleStatusRequest>,
     ) -> Result<Response<volume_server_pb::VolumeNeedleStatusResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
         let req = request.into_inner();
         let vid = VolumeId(req.volume_id);
         let needle_id = NeedleId(req.needle_id);

seaweed-volume/tests/admin_auth_coverage.rs

@@ -0,0 +1,133 @@
+//! Parity guard for the gRPC admin-auth gate.
+//!
+//! Every `VolumeServer` handler must either call `check_grpc_admin_auth` or be
+//! listed here as intentionally ungated, with the reason it stays open. Mirrors
+//! the Go side's `TestVolumeServerAdminAuthCoverage`
+//! (`weed/server/volume_grpc_admin_auth_coverage_test.go`). The gate had
+//! silently dropped off 15 handlers on the Rust side; this keeps the two
+//! implementations from drifting apart again.
+
+use std::collections::{HashMap, HashSet};
+
+/// Handlers that intentionally run without `check_grpc_admin_auth`, each with
+/// the reason. Kept in sync with `ungatedVolumeServerRPCs` on the Go side.
+///
+/// The split is by caller, not by how destructive the call is: the guard checks
+/// the peer IP against `-whiteList`, which holds masters, shell hosts and
+/// workers -- not every peer volume server. Gating a volume-server ->
+/// volume-server call therefore breaks replication, EC and tiering, so those
+/// calls stay open and need a cluster-peer identity before they can be closed.
+fn ungated_handlers() -> HashMap<&'static str, &'static str> {
+    [
+        // Cluster-internal: issued volume server -> volume server.
+        ("copy_file", "replica sync and EC task pull whole files from a peer"),
+        ("read_needle_blob", "replica sync, vacuum and EC rebuild read needles from a peer"),
+        ("read_needle_meta", "replica sync compares needle metadata across peers"),
+        ("write_needle_blob", "replica sync repairs a peer's needle"),
+        ("receive_file", "EC shard distribution pushes shards to a peer"),
+        ("read_volume_file_status", "the copy path queries the source volume server"),
+        ("volume_ec_shard_read", "a volume server reads EC shards held by a peer"),
+        ("volume_ec_blob_delete", "EC delete is fanned out to the shard holders"),
+        ("volume_ec_shards_info", "EC verification polls shard holders"),
+        ("volume_ec_shards_mount", "EC shard distribution mounts on the receiving peer"),
+        ("volume_incremental_copy", "volume backup pulls increments from a peer"),
+        ("volume_sync_status", "sync compares volume state across peers"),
+        ("volume_tail_sender", "the tail source streams to the receiving peer"),
+        ("volume_status", "replica sync and the master's vacuum loop poll volume status"),
+        // Read-only or liveness: no state change.
+        ("ping", "liveness probe"),
+        ("get_state", "read-only volume server state"),
+        ("query", "read-only data query"),
+        ("vacuum_volume_check", "read-only garbage ratio; the vacuum steps that act on it are gated"),
+        ("volume_server_status", "read-only status, the gRPC counterpart of the /status page"),
+    ]
+    .into_iter()
+    .collect()
+}
+
+/// Whether a handler body actually invokes the gate. We can't run a real AST
+/// pass like the Go side without pulling in a parser, so approximate it: drop
+/// `//` line comments (the realistic way a stray mention would sneak in) and
+/// look for the call form `check_grpc_admin_auth(`, not the bare identifier, so
+/// a comment or doc reference can't make an ungated handler pass.
+fn calls_gate(body: &str) -> bool {
+    body.lines()
+        .map(|line| line.split_once("//").map_or(line, |(code, _)| code))
+        .any(|code| code.contains("check_grpc_admin_auth("))
+}
+
+/// Enumerate the `VolumeServer` trait handlers and whether each gates on
+/// `check_grpc_admin_auth`, by scanning the source of the trait impl block.
+fn handler_gating(src: &str) -> Vec<(String, bool)> {
+    let impl_start = src
+        .find("impl VolumeServer for VolumeGrpcService {")
+        .expect("locate the VolumeServer trait impl");
+    let after = &src[impl_start..];
+    // rustfmt keeps every method body indented, so the impl's own closing brace
+    // is the first line that begins at column 0.
+    let impl_end = after
+        .match_indices("\n}")
+        .next()
+        .map(|(i, _)| i + 1)
+        .expect("locate the end of the trait impl");
+    let block = &after[..impl_end];
+
+    let marker = "\n    async fn ";
+    let starts: Vec<usize> = block.match_indices(marker).map(|(i, _)| i).collect();
+    let mut handlers = Vec::with_capacity(starts.len());
+    for (i, &start) in starts.iter().enumerate() {
+        let name_start = start + marker.len();
+        let name_end = name_start
+            + block[name_start..]
+                .find('(')
+                .expect("handler signature has an argument list");
+        let name = block[name_start..name_end].to_string();
+        let body_end = starts.get(i + 1).copied().unwrap_or(block.len());
+        let gated = calls_gate(&block[start..body_end]);
+        handlers.push((name, gated));
+    }
+    handlers
+}
+
+#[test]
+fn volume_server_admin_auth_coverage() {
+    let src = std::fs::read_to_string(concat!(
+        env!("CARGO_MANIFEST_DIR"),
+        "/src/server/grpc_server.rs"
+    ))
+    .expect("read grpc_server.rs");
+
+    let handlers = handler_gating(&src);
+    assert!(
+        handlers.len() >= 40,
+        "parsed only {} handlers from the trait impl, expected the full service",
+        handlers.len()
+    );
+
+    let ungated = ungated_handlers();
+    let mut problems = Vec::new();
+    for (name, gated) in &handlers {
+        match (gated, ungated.contains_key(name.as_str())) {
+            (true, true) => problems.push(format!(
+                "{name} calls check_grpc_admin_auth but is also listed as intentionally ungated; drop it from ungated_handlers"
+            )),
+            (false, false) => problems.push(format!(
+                "{name} does not call check_grpc_admin_auth and is not listed as intentionally ungated; gate it, or add it with the reason it must stay open"
+            )),
+            _ => {}
+        }
+    }
+
+    // Keep the exemption list honest: an entry for a handler that no longer
+    // exists hides the fact that nothing is being exempted.
+    let names: HashSet<&str> = handlers.iter().map(|(n, _)| n.as_str()).collect();
+    for name in ungated.keys() {
+        if !names.contains(name) {
+            problems.push(format!(
+                "ungated_handlers lists \"{name}\", which is not a VolumeServer handler"
+            ));
+        }
+    }
+
+    assert!(problems.is_empty(), "admin-auth coverage gaps:\n{}", problems.join("\n"));
+}