volume: widen the gRPC admin gate and stop it drifting (#10443)

* volume: gate the admin RPCs that only shell and workers call

checkGrpcAdminAuth covered 19 of the 48 VolumeServer RPCs, so an operator who
sets -whiteList expecting it to cover the gRPC surface gets partial coverage.

Extend it to ten that mutate state and are only ever called by the shell or a
worker: SetState, VolumeCopy, the EC generate/rebuild/copy/unmount/to-volume
pair, both tier moves, and VolumeTailReceiver. That is safe because the same
callers already reach gated RPCs today -- VolumeMarkReadonly, VacuumVolume*,
VolumeEcShardsDelete, VolumeDelete -- so a whitelist deployment already lists
those hosts. Nothing here is on a master or peer path, which is what made the
earlier fail-closed gate break multi-host clusters.

The split is by caller rather than by blast radius: the guard matches a peer IP
against the whitelist, and a whitelist holds masters, shell hosts and workers,
not every peer volume server. Gating a call one volume server makes to another
would break replication, EC and tiering, so those stay open.

Two test fakes embedded a nil grpc.ServerStream and only implemented Send;
they now implement Context, which the streaming RPCs read to authorize.

* volume: fail the build when a gRPC method skips the admin gate

The admin gate is an opt-in list in a 48-method service, which is how it
drifted down to covering 19 of them: nothing tied adding an RPC to deciding
whether it needed the gate.

Parse volume_server.proto, walk the AST of every *VolumeServer method, and
require each RPC to either call checkGrpcAdminAuth or appear in
ungatedVolumeServerRPCs with the reason it stays open. A stale entry naming an
RPC that no longer exists fails too, so the list can't quietly stop exempting
anything.

The exemptions are the cluster-internal calls -- replica sync, EC shard
distribution, vacuum reads, backup, tailing -- plus the read-only and liveness
RPCs. Closing the cluster-internal ones needs a peer identity rather than an
IP whitelist; recording them here makes that a visible decision instead of an
omission.

The AST walk also corrects the count: a line-window scan credits
VacuumVolumeCheck and VolumeServerStatus with a neighbouring function's guard.

weed/server/volume_grpc_admin_auth_coverage_test.go

@@ -0,0 +1,162 @@
+package weed_server
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sort"
+	"strings"
+	"testing"
+)
+
+// ungatedVolumeServerRPCs are the VolumeServer gRPC methods that intentionally
+// run without checkGrpcAdminAuth, each with the reason it stays open. Every
+// other RPC in volume_server.proto must call the gate.
+//
+// The split is by caller, not by how destructive the method is: the guard
+// checks the peer IP against -whiteList, and an operator's whitelist holds
+// masters, shell hosts and workers -- not every peer volume server. Gating a
+// call that one volume server makes to another therefore breaks replication, EC
+// and tiering in exactly the way the fail-closed gate did before it was
+// reverted. Those calls are listed here and need a different mechanism (a
+// cluster-peer identity) before they can be closed.
+var ungatedVolumeServerRPCs = map[string]string{
+	// Cluster-internal: issued volume server -> volume server.
+	"CopyFile":              "replica sync and EC task pull whole files from a peer",
+	"ReadNeedleBlob":        "replica sync, vacuum and EC rebuild read needles from a peer",
+	"ReadNeedleMeta":        "replica sync compares needle metadata across peers",
+	"WriteNeedleBlob":       "replica sync repairs a peer's needle",
+	"ReceiveFile":           "EC shard distribution pushes shards to a peer",
+	"ReadVolumeFileStatus":  "the copy path queries the source volume server",
+	"VolumeEcShardRead":     "a volume server reads EC shards held by a peer",
+	"VolumeEcBlobDelete":    "EC delete is fanned out to the shard holders",
+	"VolumeEcShardsInfo":    "EC verification polls shard holders",
+	"VolumeEcShardsMount":   "EC shard distribution mounts on the receiving peer",
+	"VolumeIncrementalCopy": "volume backup pulls increments from a peer",
+	"VolumeSyncStatus":      "sync compares volume state across peers",
+	"VolumeTailSender":      "the tail source streams to the receiving peer",
+	"VolumeStatus":          "replica sync and the master's vacuum loop poll volume status",
+
+	// Read-only or liveness: no state change, and gating them would break
+	// health checking and monitoring without closing a write path. They do
+	// disclose topology and usage detail, so gating them is a defensible
+	// tightening -- it just needs to be a decision rather than an oversight,
+	// which is what this list is for.
+	"Ping":               "liveness probe",
+	"GetState":           "read-only volume server state",
+	"Query":              "read-only data query",
+	"VacuumVolumeCheck":  "read-only garbage ratio; the vacuum steps that act on it are gated",
+	"VolumeServerStatus": "read-only status, the gRPC counterpart of the /status page",
+}
+
+// TestVolumeServerAdminAuthCoverage fails when a VolumeServer RPC is neither
+// gated by checkGrpcAdminAuth nor listed in ungatedVolumeServerRPCs with a
+// reason. A new RPC therefore cannot be added without someone deciding which
+// side of the boundary it sits on -- an allowlist this size drifts otherwise,
+// which is how the gate ended up covering less than half the service.
+func TestVolumeServerAdminAuthCoverage(t *testing.T) {
+	declared := rpcNamesFromProto(t)
+	if len(declared) < 40 {
+		t.Fatalf("parsed only %d RPCs from the proto, expected the full service", len(declared))
+	}
+
+	gated := gatedVolumeServerMethods(t)
+
+	for _, rpc := range declared {
+		_, exempt := ungatedVolumeServerRPCs[rpc]
+		switch {
+		case gated[rpc] && exempt:
+			t.Errorf("%s calls checkGrpcAdminAuth but is also listed as intentionally ungated; drop it from ungatedVolumeServerRPCs", rpc)
+		case !gated[rpc] && !exempt:
+			t.Errorf("%s does not call checkGrpcAdminAuth and is not listed in ungatedVolumeServerRPCs; "+
+				"gate it, or add it with the reason it must stay open", rpc)
+		}
+	}
+
+	// Keep the exemption list honest: an entry naming an RPC that no longer
+	// exists hides the fact that nothing is being exempted.
+	declaredSet := make(map[string]struct{}, len(declared))
+	for _, rpc := range declared {
+		declaredSet[rpc] = struct{}{}
+	}
+	var stale []string
+	for rpc := range ungatedVolumeServerRPCs {
+		if _, ok := declaredSet[rpc]; !ok {
+			stale = append(stale, rpc)
+		}
+	}
+	sort.Strings(stale)
+	for _, rpc := range stale {
+		t.Errorf("ungatedVolumeServerRPCs lists %q, which is not an RPC in volume_server.proto", rpc)
+	}
+}
+
+func rpcNamesFromProto(t *testing.T) []string {
+	t.Helper()
+	path := filepath.Join("..", "pb", "volume_server.proto")
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read %s: %v", path, err)
+	}
+	re := regexp.MustCompile(`(?m)^\s*rpc\s+([A-Za-z0-9_]+)\s*\(`)
+	var names []string
+	for _, m := range re.FindAllStringSubmatch(string(data), -1) {
+		names = append(names, m[1])
+	}
+	sort.Strings(names)
+	return names
+}
+
+// gatedVolumeServerMethods reports which methods on *VolumeServer call
+// checkGrpcAdminAuth anywhere in their body. It walks the AST rather than
+// scanning a fixed window of lines so a guard placed after an early
+// maintenance-mode check still counts.
+func gatedVolumeServerMethods(t *testing.T) map[string]bool {
+	t.Helper()
+	fset := token.NewFileSet()
+	pkgs, err := parser.ParseDir(fset, ".", func(fi os.FileInfo) bool {
+		return strings.HasSuffix(fi.Name(), ".go") && !strings.HasSuffix(fi.Name(), "_test.go")
+	}, 0)
+	if err != nil {
+		t.Fatalf("parse package: %v", err)
+	}
+
+	gated := make(map[string]bool)
+	for _, pkg := range pkgs {
+		for _, file := range pkg.Files {
+			for _, decl := range file.Decls {
+				fn, ok := decl.(*ast.FuncDecl)
+				if !ok || fn.Recv == nil || fn.Body == nil {
+					continue
+				}
+				if !isVolumeServerReceiver(fn.Recv) {
+					continue
+				}
+				ast.Inspect(fn.Body, func(n ast.Node) bool {
+					sel, ok := n.(*ast.SelectorExpr)
+					if ok && sel.Sel.Name == "checkGrpcAdminAuth" {
+						gated[fn.Name.Name] = true
+						return false
+					}
+					return true
+				})
+			}
+		}
+	}
+	return gated
+}
+
+func isVolumeServerReceiver(recv *ast.FieldList) bool {
+	if len(recv.List) != 1 {
+		return false
+	}
+	star, ok := recv.List[0].Type.(*ast.StarExpr)
+	if !ok {
+		return false
+	}
+	ident, ok := star.X.(*ast.Ident)
+	return ok && ident.Name == "VolumeServer"
+}

weed/server/volume_grpc_copy.go

@@ -27,6 +27,9 @@ const BufferSizeLimit = 1024 * 1024 * 2
 
 // VolumeCopy copy the .idx .dat .vif files, and mount the volume
 func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stream volume_server_pb.VolumeServer_VolumeCopyServer) error {
+	if err := vs.checkGrpcAdminAuth(stream.Context()); err != nil {
+		return err
+	}
 	if err := vs.CheckMaintenanceMode(); err != nil {
 		return err
 	}

weed/server/volume_grpc_erasure_coding.go

@@ -43,6 +43,9 @@ Steps to apply erasure coding to .dat .idx files
 
 // VolumeEcShardsGenerate generates the .ecx and .ec00 ~ .ec13 files
 func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_server_pb.VolumeEcShardsGenerateRequest) (*volume_server_pb.VolumeEcShardsGenerateResponse, error) {
+	if err := vs.checkGrpcAdminAuth(ctx); err != nil {
+		return nil, err
+	}
 	if err := vs.CheckMaintenanceMode(); err != nil {
 		return nil, err
 	}
@@ -182,6 +185,9 @@ func recordEcRebuild(result string, d time.Duration) {
 
 // VolumeEcShardsRebuild generates the any of the missing .ec00 ~ .ec13 files
 func (vs *VolumeServer) VolumeEcShardsRebuild(ctx context.Context, req *volume_server_pb.VolumeEcShardsRebuildRequest) (*volume_server_pb.VolumeEcShardsRebuildResponse, error) {
+	if err := vs.checkGrpcAdminAuth(ctx); err != nil {
+		return nil, err
+	}
 	if err := vs.CheckMaintenanceMode(); err != nil {
 		return nil, err
 	}
@@ -292,6 +298,9 @@ func (vs *VolumeServer) VolumeEcShardsRebuild(ctx context.Context, req *volume_s
 
 // VolumeEcShardsCopy copy the .ecx and some ec data slices
 func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) {
+	if err := vs.checkGrpcAdminAuth(ctx); err != nil {
+		return nil, err
+	}
 	if err := vs.CheckMaintenanceMode(); err != nil {
 		return nil, err
 	}
@@ -769,6 +778,9 @@ func (vs *VolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_ser
 }
 
 func (vs *VolumeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_server_pb.VolumeEcShardsUnmountRequest) (*volume_server_pb.VolumeEcShardsUnmountResponse, error) {
+	if err := vs.checkGrpcAdminAuth(ctx); err != nil {
+		return nil, err
+	}
 
 	glog.V(0).Infof("VolumeEcShardsUnmount: %v", req)
 
@@ -900,6 +912,9 @@ func (vs *VolumeServer) VolumeEcBlobDelete(ctx context.Context, req *volume_serv
 
 // VolumeEcShardsToVolume generates the .idx, .dat files from .ecx, .ecj and .ec01 ~ .ec14 files
 func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_server_pb.VolumeEcShardsToVolumeRequest) (*volume_server_pb.VolumeEcShardsToVolumeResponse, error) {
+	if err := vs.checkGrpcAdminAuth(ctx); err != nil {
+		return nil, err
+	}
 	if err := vs.CheckMaintenanceMode(); err != nil {
 		return nil, err
 	}

weed/server/volume_grpc_state.go

@@ -17,6 +17,9 @@ func (vs *VolumeServer) GetState(ctx context.Context, req *volume_server_pb.GetS
 
 // SetState updates state flags for volume servers.
 func (vs *VolumeServer) SetState(ctx context.Context, req *volume_server_pb.SetStateRequest) (*volume_server_pb.SetStateResponse, error) {
+	if err := vs.checkGrpcAdminAuth(ctx); err != nil {
+		return nil, err
+	}
 	err := vs.store.State.Update(req.GetState())
 	resp := &volume_server_pb.SetStateResponse{
 		State: vs.store.State.Proto(),

weed/server/volume_grpc_tail.go

@@ -81,6 +81,9 @@ func sendNeedlesSince(stream volume_server_pb.VolumeServer_VolumeTailSenderServe
 }
 
 func (vs *VolumeServer) VolumeTailReceiver(ctx context.Context, req *volume_server_pb.VolumeTailReceiverRequest) (*volume_server_pb.VolumeTailReceiverResponse, error) {
+	if err := vs.checkGrpcAdminAuth(ctx); err != nil {
+		return nil, err
+	}
 
 	resp := &volume_server_pb.VolumeTailReceiverResponse{}
 

weed/server/volume_grpc_tier_download.go

@@ -16,6 +16,9 @@ import (
 
 // VolumeTierMoveDatFromRemote copy dat file from a remote tier to local volume server
 func (vs *VolumeServer) VolumeTierMoveDatFromRemote(req *volume_server_pb.VolumeTierMoveDatFromRemoteRequest, stream volume_server_pb.VolumeServer_VolumeTierMoveDatFromRemoteServer) error {
+	if err := vs.checkGrpcAdminAuth(stream.Context()); err != nil {
+		return err
+	}
 
 	// find existing volume
 	v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))

weed/server/volume_grpc_tier_download_test.go

@@ -2,6 +2,7 @@ package weed_server
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io"
 	"os"
@@ -131,7 +132,9 @@ func (f *tierTestBackendFile) GetStat() (int64, time.Time, error) {
 	return int64(files[0].FileSize), time.Unix(int64(files[0].ModifiedTime), 0), nil
 }
 
-// fakeTierStream is a no-op server stream for the tier-download RPC.
+// fakeTierStream is a no-op server stream for the tier-download RPC. The
+// embedded grpc.ServerStream is nil, so Context is implemented here rather than
+// promoted -- the RPC reads it to authorize the caller.
 type fakeTierStream struct {
 	grpc.ServerStream
 }
@@ -140,6 +143,10 @@ func (s *fakeTierStream) Send(*volume_server_pb.VolumeTierMoveDatFromRemoteRespo
 	return nil
 }
 
+func (s *fakeTierStream) Context() context.Context {
+	return context.Background()
+}
+
 func newTierTestStore(t *testing.T, dir string) *storage.Store {
 	t.Helper()
 	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()

weed/server/volume_grpc_tier_test.go

@@ -1,6 +1,7 @@
 package weed_server
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"os"
@@ -21,6 +22,9 @@ import (
 
 const tierTimestampTestBackendName = "tier_timestamp_test.default"
 
+// discardServerStream drops everything sent to it. The embedded
+// grpc.ServerStream is nil, so Context is implemented here rather than promoted
+// -- the tier RPCs read it to authorize the caller.
 type discardServerStream[T any] struct {
 	grpc.ServerStream
 }
@@ -29,6 +33,10 @@ func (s *discardServerStream[T]) Send(*T) error {
 	return nil
 }
 
+func (s *discardServerStream[T]) Context() context.Context {
+	return context.Background()
+}
+
 type tierTimestampTestBackend struct {
 	root string
 }

weed/server/volume_grpc_tier_upload.go

@@ -12,6 +12,9 @@ import (
 
 // VolumeTierMoveDatToRemote copy dat file to a remote tier
 func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTierMoveDatToRemoteRequest, stream volume_server_pb.VolumeServer_VolumeTierMoveDatToRemoteServer) error {
+	if err := vs.checkGrpcAdminAuth(stream.Context()); err != nil {
+		return err
+	}
 	if err := vs.CheckMaintenanceMode(); err != nil {
 		return err
 	}