From c7d04771171b01364c44382e650bf67e5f40a46d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 25 Jul 2026 23:53:29 -0700 Subject: [PATCH] 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. --- .../volume_grpc_admin_auth_coverage_test.go | 162 ++++++++++++++++++ weed/server/volume_grpc_copy.go | 3 + weed/server/volume_grpc_erasure_coding.go | 15 ++ weed/server/volume_grpc_state.go | 3 + weed/server/volume_grpc_tail.go | 3 + weed/server/volume_grpc_tier_download.go | 3 + weed/server/volume_grpc_tier_download_test.go | 9 +- weed/server/volume_grpc_tier_test.go | 8 + weed/server/volume_grpc_tier_upload.go | 3 + 9 files changed, 208 insertions(+), 1 deletion(-) create mode 100644 weed/server/volume_grpc_admin_auth_coverage_test.go diff --git a/weed/server/volume_grpc_admin_auth_coverage_test.go b/weed/server/volume_grpc_admin_auth_coverage_test.go new file mode 100644 index 000000000..179a6a960 --- /dev/null +++ b/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" +} diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 47119b59b..0347e37db 100644 --- a/weed/server/volume_grpc_copy.go +++ b/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 } diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 4b4c9d46f..66499dec0 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/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 } diff --git a/weed/server/volume_grpc_state.go b/weed/server/volume_grpc_state.go index 52bc823d4..06bd38dbe 100644 --- a/weed/server/volume_grpc_state.go +++ b/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(), diff --git a/weed/server/volume_grpc_tail.go b/weed/server/volume_grpc_tail.go index 08aae06af..552fea2b7 100644 --- a/weed/server/volume_grpc_tail.go +++ b/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{} diff --git a/weed/server/volume_grpc_tier_download.go b/weed/server/volume_grpc_tier_download.go index 81ef87505..5860f859a 100644 --- a/weed/server/volume_grpc_tier_download.go +++ b/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)) diff --git a/weed/server/volume_grpc_tier_download_test.go b/weed/server/volume_grpc_tier_download_test.go index f346b5257..378ca13f1 100644 --- a/weed/server/volume_grpc_tier_download_test.go +++ b/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() diff --git a/weed/server/volume_grpc_tier_test.go b/weed/server/volume_grpc_tier_test.go index 1329b0bf4..f042948de 100644 --- a/weed/server/volume_grpc_tier_test.go +++ b/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 } diff --git a/weed/server/volume_grpc_tier_upload.go b/weed/server/volume_grpc_tier_upload.go index dee062b58..c6355c8c1 100644 --- a/weed/server/volume_grpc_tier_upload.go +++ b/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 }