diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 838269d24..03c2a7792 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -145,6 +145,37 @@ pub struct VolumeGrpcService { } impl VolumeGrpcService { + /// Verifies the gRPC caller is allowed to invoke a destructive admin + /// operation. Mirrors the Go side's checkGrpcAdminAuth: an empty + /// whitelist accepts everyone (insecure-by-default for tests and + /// upgrades), a populated whitelist accepts only matching peer IPs. + /// + /// `remote_addr()` on a real gRPC connection always yields the peer's + /// SocketAddr; if it is somehow None we deny, matching "if we don't + /// know who the caller is, refuse." + fn check_grpc_admin_auth(&self, request: &Request) -> Result<(), Status> { + let remote = match request.remote_addr() { + Some(addr) => addr, + None => { + tracing::warn!("gRPC admin auth failed: no peer info"); + return Err(Status::permission_denied("no peer info")); + } + }; + let host = remote.ip().to_string(); + let guard = self.state.guard.read().unwrap(); + if !guard.check_whitelist(&host) { + tracing::warn!( + "gRPC admin auth failed: {} is not whitelisted (remote: {})", + host, + remote, + ); + return Err(Status::permission_denied(format!( + "not authorized: {host}" + ))); + } + Ok(()) + } + async fn notify_master_volume_readonly( &self, info: &MasterVolumeInfo, @@ -459,6 +490,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; self.state.check_maintenance()?; let req = request.into_inner(); let vid = VolumeId(req.volume_id); @@ -518,6 +550,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; self.state.check_maintenance()?; let vid = VolumeId(request.into_inner().volume_id); @@ -553,6 +586,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; self.state.check_maintenance()?; let vid = VolumeId(request.into_inner().volume_id); let mut store = self.state.store.write().unwrap(); @@ -568,6 +602,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; let collection = &request.into_inner().collection; let mut store = self.state.store.write().unwrap(); store @@ -580,6 +615,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; self.state.check_maintenance()?; let req = request.into_inner(); let vid = VolumeId(req.volume_id); @@ -728,6 +764,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; let req = request.into_inner(); let vid = VolumeId(req.volume_id); @@ -744,6 +781,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; let vid = VolumeId(request.into_inner().volume_id); let mut store = self.state.store.write().unwrap(); // Go returns nil when volume is not found (idempotent unmount) @@ -757,6 +795,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; self.state.check_maintenance()?; let req = request.into_inner(); let vid = VolumeId(req.volume_id); @@ -780,6 +819,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; let req = request.into_inner(); let vid = VolumeId(req.volume_id); // Go: volume lookup (L239-241) happens before maintenance check (L166 in makeVolumeReadonly) @@ -799,6 +839,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; let req = request.into_inner(); let vid = VolumeId(req.volume_id); let info = { @@ -848,6 +889,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; self.state.check_maintenance()?; let req = request.into_inner(); let vid = VolumeId(req.volume_id); @@ -2542,6 +2584,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; self.state.check_maintenance()?; let req = request.into_inner(); let vid = VolumeId(req.volume_id); @@ -3240,8 +3283,9 @@ impl VolumeServer for VolumeGrpcService { async fn volume_server_leave( &self, - _request: Request, + request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; *self.state.is_stopping.write().unwrap() = true; self.state.is_heartbeating.store(false, Ordering::Relaxed); // Wake heartbeat loop to send deregistration. @@ -3391,6 +3435,7 @@ impl VolumeServer for VolumeGrpcService { &self, request: Request, ) -> Result, Status> { + self.check_grpc_admin_auth(&request)?; let req = request.into_inner(); // Validate mode diff --git a/weed/security/guard.go b/weed/security/guard.go index a41cb0288..ea71fc954 100644 --- a/weed/security/guard.go +++ b/weed/security/guard.go @@ -98,31 +98,35 @@ func GetActualRemoteHost(r *http.Request) string { } func (g *Guard) checkWhiteList(w http.ResponseWriter, r *http.Request) error { - if g.isEmptyWhiteList { - return nil - } - host := GetActualRemoteHost(r) - - // Check exact match first (works for both IPs and hostnames) - if _, ok := g.whiteListIp[host]; ok { + if g.IsWhiteListed(host) { return nil } + glog.V(0).Infof("Not in whitelist: %s (original RemoteAddr: %s)", host, r.RemoteAddr) + return fmt.Errorf("Not in whitelist: %s", host) +} - // Check CIDR ranges (only for valid IP addresses) +// IsWhiteListed returns true if the given host IP is allowed by the guard. +// When no whitelist is configured (security inactive), all hosts are allowed. +func (g *Guard) IsWhiteListed(host string) bool { + if !g.isWriteActive { + return true + } + if g.isEmptyWhiteList { + return true + } + if _, ok := g.whiteListIp[host]; ok { + return true + } remote := net.ParseIP(host) if remote != nil { for _, cidrnet := range g.whiteListCIDR { - // If the whitelist entry contains a "/" it - // is a CIDR range, and we should check the if cidrnet.Contains(remote) { - return nil + return true } } } - - glog.V(0).Infof("Not in whitelist: %s (original RemoteAddr: %s)", host, r.RemoteAddr) - return fmt.Errorf("Not in whitelist: %s", host) + return false } func (g *Guard) UpdateWhiteList(whiteList []string) { diff --git a/weed/security/guard_whitelist_test.go b/weed/security/guard_whitelist_test.go new file mode 100644 index 000000000..5f304d06c --- /dev/null +++ b/weed/security/guard_whitelist_test.go @@ -0,0 +1,50 @@ +package security + +import "testing" + +// TestIsWhiteListedEmptyConfigAllowsEverything pins the contract that the +// gRPC admin auth helper relies on: when no whitelist and no signing key +// are configured, IsWhiteListed accepts any host (including the empty +// string and unparseable peer addresses like the gRPC "@" passthrough +// form). Otherwise insecure default deployments would lock themselves out +// of every destructive admin RPC. +func TestIsWhiteListedEmptyConfigAllowsEverything(t *testing.T) { + g := NewGuard(nil, "", 0, "", 0) + for _, host := range []string{"", "@", "127.0.0.1", "::1", "garbage:value"} { + if !g.IsWhiteListed(host) { + t.Errorf("empty config should accept host=%q, got false", host) + } + } +} + +func TestIsWhiteListedWithListRejectsUnknown(t *testing.T) { + g := NewGuard([]string{"10.0.0.1", "192.168.1.0/24"}, "", 0, "", 0) + cases := []struct { + host string + want bool + }{ + {"10.0.0.1", true}, + {"192.168.1.42", true}, + {"192.168.2.1", false}, + {"127.0.0.1", false}, + {"", false}, + {"@", false}, + } + for _, tc := range cases { + if got := g.IsWhiteListed(tc.host); got != tc.want { + t.Errorf("IsWhiteListed(%q) = %v want %v", tc.host, got, tc.want) + } + } +} + +func TestIsWhiteListedWithSigningKeyButNoWhitelistAllowsAll(t *testing.T) { + // JWT-only mode: signing key set, whitelist empty. Without this branch + // the gRPC admin auth helper would falsely deny in-process callers when + // the operator wires up signing keys but hasn't enumerated IPs. + g := NewGuard(nil, "deadbeef", 0, "", 0) + for _, host := range []string{"", "@", "127.0.0.1"} { + if !g.IsWhiteListed(host) { + t.Errorf("signing-key + empty whitelist should accept host=%q", host) + } + } +} diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index c2286ce5b..751314eb1 100644 --- a/weed/server/volume_grpc_admin.go +++ b/weed/server/volume_grpc_admin.go @@ -3,9 +3,14 @@ package weed_server import ( "context" "fmt" + "net" "path/filepath" "time" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/peer" + "google.golang.org/grpc/status" + "github.com/seaweedfs/seaweedfs/weed/util/version" "github.com/seaweedfs/seaweedfs/weed/storage" @@ -22,10 +27,53 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/types" ) +// checkGrpcAdminAuth verifies the gRPC caller is authorized for destructive +// admin operations by checking the peer address against the guard's whitelist. +// +// IP extraction prefers a typed *net.TCPAddr where available, falling back to +// SplitHostPort on the string form, then to the raw string. The fallback +// chain matters because in-process/passthrough connections used in tests +// surface as unparseable strings like "@"; with an empty whitelist the +// allow-all branch in IsWhiteListed accepts them, with a whitelist they're +// denied as expected. +// +// Failed authorization attempts are logged so an operator running with a +// configured whitelist can spot misconfigured callers and probe attempts. +func (vs *VolumeServer) checkGrpcAdminAuth(ctx context.Context) error { + if vs.guard == nil { + return nil + } + pr, ok := peer.FromContext(ctx) + if !ok { + // Real gRPC connections always populate peer info; if we don't know + // who the caller is, deny. + glog.V(0).Infof("gRPC admin auth failed: no peer info") + return status.Error(codes.PermissionDenied, "no peer info") + } + addr := pr.Addr.String() + var host string + if tcpAddr, ok := pr.Addr.(*net.TCPAddr); ok { + host = tcpAddr.IP.String() + } else if h, _, splitErr := net.SplitHostPort(addr); splitErr == nil { + host = h + } else { + host = addr + } + if !vs.guard.IsWhiteListed(host) { + glog.V(0).Infof("gRPC admin auth failed: %s is not whitelisted (remote: %s)", host, addr) + return status.Errorf(codes.PermissionDenied, "not authorized: %s", host) + } + return nil +} + func (vs *VolumeServer) DeleteCollection(ctx context.Context, req *volume_server_pb.DeleteCollectionRequest) (*volume_server_pb.DeleteCollectionResponse, error) { resp := &volume_server_pb.DeleteCollectionResponse{} + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return resp, err + } + err := vs.store.DeleteCollection(req.Collection) if err != nil { @@ -41,6 +89,10 @@ func (vs *VolumeServer) DeleteCollection(ctx context.Context, req *volume_server func (vs *VolumeServer) AllocateVolume(ctx context.Context, req *volume_server_pb.AllocateVolumeRequest) (*volume_server_pb.AllocateVolumeResponse, error) { resp := &volume_server_pb.AllocateVolumeResponse{} + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return resp, err + } + if err := vs.CheckMaintenanceMode(); err != nil { return resp, err } @@ -72,6 +124,10 @@ func (vs *VolumeServer) VolumeMount(ctx context.Context, req *volume_server_pb.V resp := &volume_server_pb.VolumeMountResponse{} + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return resp, err + } + err := vs.store.MountVolume(needle.VolumeId(req.VolumeId)) if err != nil { @@ -88,6 +144,10 @@ func (vs *VolumeServer) VolumeUnmount(ctx context.Context, req *volume_server_pb resp := &volume_server_pb.VolumeUnmountResponse{} + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return resp, err + } + err := vs.store.UnmountVolume(needle.VolumeId(req.VolumeId)) if err != nil { @@ -103,6 +163,10 @@ func (vs *VolumeServer) VolumeUnmount(ctx context.Context, req *volume_server_pb func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb.VolumeDeleteRequest) (*volume_server_pb.VolumeDeleteResponse, error) { resp := &volume_server_pb.VolumeDeleteResponse{} + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return resp, err + } + if err := vs.CheckMaintenanceMode(); err != nil { return resp, err } @@ -122,6 +186,10 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb. func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) { resp := &volume_server_pb.VolumeConfigureResponse{} + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return resp, err + } + if err := vs.CheckMaintenanceMode(); err != nil { return resp, err } @@ -236,6 +304,10 @@ func (vs *VolumeServer) notifyMasterVolumeReadonly(ctx context.Context, v *stora func (vs *VolumeServer) VolumeMarkReadonly(ctx context.Context, req *volume_server_pb.VolumeMarkReadonlyRequest) (*volume_server_pb.VolumeMarkReadonlyResponse, error) { resp := &volume_server_pb.VolumeMarkReadonlyResponse{} + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return resp, err + } + v := vs.store.GetVolume(needle.VolumeId(req.VolumeId)) if v == nil { return resp, fmt.Errorf("volume %d not found", req.VolumeId) @@ -251,6 +323,10 @@ func (vs *VolumeServer) VolumeMarkReadonly(ctx context.Context, req *volume_serv func (vs *VolumeServer) VolumeMarkWritable(ctx context.Context, req *volume_server_pb.VolumeMarkWritableRequest) (*volume_server_pb.VolumeMarkWritableResponse, error) { resp := &volume_server_pb.VolumeMarkWritableResponse{} + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return resp, err + } + v := vs.store.GetVolume(needle.VolumeId(req.VolumeId)) if v == nil { return resp, fmt.Errorf("volume %d not found", req.VolumeId) @@ -308,6 +384,10 @@ func (vs *VolumeServer) VolumeServerLeave(ctx context.Context, req *volume_serve resp := &volume_server_pb.VolumeServerLeaveResponse{} + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return resp, err + } + vs.StopHeartbeat() return resp, nil diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index ec05a209d..92862c378 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -316,6 +316,9 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv // VolumeEcShardsDelete local delete the .ecx and some ec data slices if not needed // the shard should not be mounted before calling this. func (vs *VolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, 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_scrub.go b/weed/server/volume_grpc_scrub.go index 68acc5616..27afe2dac 100644 --- a/weed/server/volume_grpc_scrub.go +++ b/weed/server/volume_grpc_scrub.go @@ -14,6 +14,9 @@ import ( ) func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.ScrubVolumeRequest) (*volume_server_pb.ScrubVolumeResponse, error) { + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return nil, err + } vids := []needle.VolumeId{} if len(req.GetVolumeIds()) == 0 { for _, l := range vs.store.Locations { diff --git a/weed/server/volume_grpc_vacuum.go b/weed/server/volume_grpc_vacuum.go index 0b32c1d5d..9ee88d6f8 100644 --- a/weed/server/volume_grpc_vacuum.go +++ b/weed/server/volume_grpc_vacuum.go @@ -34,6 +34,9 @@ func (vs *VolumeServer) VacuumVolumeCheck(ctx context.Context, req *volume_serve } func (vs *VolumeServer) VacuumVolumeCompact(req *volume_server_pb.VacuumVolumeCompactRequest, stream volume_server_pb.VolumeServer_VacuumVolumeCompactServer) error { + if err := vs.checkGrpcAdminAuth(stream.Context()); err != nil { + return err + } if err := vs.CheckMaintenanceMode(); err != nil { return err } @@ -80,6 +83,9 @@ func (vs *VolumeServer) VacuumVolumeCompact(req *volume_server_pb.VacuumVolumeCo } func (vs *VolumeServer) VacuumVolumeCommit(ctx context.Context, req *volume_server_pb.VacuumVolumeCommitRequest) (*volume_server_pb.VacuumVolumeCommitResponse, error) { + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return nil, err + } if err := vs.CheckMaintenanceMode(); err != nil { return nil, err } @@ -106,6 +112,9 @@ func (vs *VolumeServer) VacuumVolumeCommit(ctx context.Context, req *volume_serv } func (vs *VolumeServer) VacuumVolumeCleanup(ctx context.Context, req *volume_server_pb.VacuumVolumeCleanupRequest) (*volume_server_pb.VacuumVolumeCleanupResponse, error) { + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return nil, err + } if err := vs.CheckMaintenanceMode(); err != nil { return nil, err }