From 135af25b55776687e13c844dee4494d74edb9d24 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 27 Apr 2026 23:10:36 -0700 Subject: [PATCH] fix(grpc): require host match before routing dials to local Unix socket (#9254) (#9257) * fix(grpc): require host match before routing dials to local Unix socket (#9254) resolveLocalGrpcSocket keyed the Unix-socket hijack on port alone, so a remote peer reusing a local gRPC port (e.g. a standalone `weed volume` defaulting port.grpc=17334 against a `weed server` whose in-process volume socket is also on 17334) had its inbound RPCs silently rerouted to the local socket. In a cross-DC replication=100 cluster this surfaced as persistent volume-grow failure: both AllocateVolume RPCs landed on the local volume server, the second returned "Volume Id N already exists!", the grow rolled back, and S3 PUTs to the new collection returned 500. Track a per-port set of host strings that count as "this machine" and require the dial host to be in that set before redirecting. Loopback aliases (localhost, 127.0.0.1, ::1, "") are always included so same-process dials via loopback still take the socket fast path. * test(grpc): cover empty-host bare-port dial in local-socket regression test The empty alias is registered explicitly so SplitHostPort outputs like ":17334" (which can occur when a caller dials a bare port) take the local-socket fast path. Add a case so that path is exercised. --- weed/command/mini.go | 10 +++--- weed/command/server.go | 2 +- weed/pb/grpc_client_server.go | 56 ++++++++++++++++++++++++------ weed/pb/grpc_client_server_test.go | 52 +++++++++++++++++++++++++++ 4 files changed, 103 insertions(+), 17 deletions(-) diff --git a/weed/command/mini.go b/weed/command/mini.go index 60989ff45..daf1c1ac0 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -952,13 +952,13 @@ func runMini(cmd *Command, args []string) bool { // Register Unix socket paths for gRPC services so local inter-service // communication goes through Unix sockets instead of TCP. - pb.RegisterLocalGrpcSocket(*miniMasterOptions.portGrpc, fmt.Sprintf("/tmp/seaweedfs-master-grpc-%d.sock", *miniMasterOptions.portGrpc)) - pb.RegisterLocalGrpcSocket(*miniOptions.v.portGrpc, fmt.Sprintf("/tmp/seaweedfs-volume-grpc-%d.sock", *miniOptions.v.portGrpc)) - pb.RegisterLocalGrpcSocket(*miniFilerOptions.portGrpc, fmt.Sprintf("/tmp/seaweedfs-filer-grpc-%d.sock", *miniFilerOptions.portGrpc)) + pb.RegisterLocalGrpcSocket(*miniIp, *miniMasterOptions.portGrpc, fmt.Sprintf("/tmp/seaweedfs-master-grpc-%d.sock", *miniMasterOptions.portGrpc)) + pb.RegisterLocalGrpcSocket(*miniIp, *miniOptions.v.portGrpc, fmt.Sprintf("/tmp/seaweedfs-volume-grpc-%d.sock", *miniOptions.v.portGrpc)) + pb.RegisterLocalGrpcSocket(*miniIp, *miniFilerOptions.portGrpc, fmt.Sprintf("/tmp/seaweedfs-filer-grpc-%d.sock", *miniFilerOptions.portGrpc)) if *miniS3Options.portGrpc > 0 { - pb.RegisterLocalGrpcSocket(*miniS3Options.portGrpc, fmt.Sprintf("/tmp/seaweedfs-s3-grpc-%d.sock", *miniS3Options.portGrpc)) + pb.RegisterLocalGrpcSocket(*miniIp, *miniS3Options.portGrpc, fmt.Sprintf("/tmp/seaweedfs-s3-grpc-%d.sock", *miniS3Options.portGrpc)) } - pb.RegisterLocalGrpcSocket(*miniAdminOptions.grpcPort, fmt.Sprintf("/tmp/seaweedfs-admin-grpc-%d.sock", *miniAdminOptions.grpcPort)) + pb.RegisterLocalGrpcSocket(*miniIp, *miniAdminOptions.grpcPort, fmt.Sprintf("/tmp/seaweedfs-admin-grpc-%d.sock", *miniAdminOptions.grpcPort)) go stats_collect.StartMetricsServer(*miniMetricsHttpIp, *miniMetricsHttpPort) diff --git a/weed/command/server.go b/weed/command/server.go index c6e42d9a4..227748006 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -350,7 +350,7 @@ func runServer(cmd *Command, args []string) bool { if *svc.portGrpc == 0 { *svc.portGrpc = 10000 + *svc.port } - pb.RegisterLocalGrpcSocket(*svc.portGrpc, fmt.Sprintf("/tmp/seaweedfs-%s-grpc-%d.sock", svc.name, *svc.portGrpc)) + pb.RegisterLocalGrpcSocket(*serverIp, *svc.portGrpc, fmt.Sprintf("/tmp/seaweedfs-%s-grpc-%d.sock", svc.name, *svc.portGrpc)) } } diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index e49eb59b1..e26d6278b 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -45,10 +45,23 @@ var ( grpcClients = make(map[string]*versionedGrpcClient) grpcClientsLock sync.Mutex - // localGrpcSockets maps gRPC port numbers to Unix socket paths. - // When registered (by mini mode), gRPC clients connect via Unix socket - // instead of TCP for local services. - localGrpcSockets = make(map[int]string) + // localGrpcSockets maps gRPC port numbers to Unix socket paths for + // gRPC services running in this process. Used by both the server side + // (to start serving on the socket) and the client side (to dial it + // instead of TCP for same-host RPCs). + localGrpcSockets = make(map[int]string) + // localGrpcHosts maps gRPC port → set of host strings that count as + // "this machine" for that port. Only dials whose target host is in this + // set are routed through the Unix socket; dials to other hosts that + // happen to use the same port number go over TCP as normal. + // + // Without this host check the hijack was keyed purely on port, so a + // remote server reusing one of our local gRPC ports (e.g. a standalone + // `weed volume -port=7334` defaulting `port.grpc=17334` against a + // `weed server` whose in-process volume socket is also on 17334) had + // its inbound RPCs silently rerouted to our local Unix socket — see + // issue #9254. + localGrpcHosts = make(map[int]map[string]struct{}) localGrpcSocketsLock sync.RWMutex ) @@ -63,12 +76,26 @@ func init() { http.DefaultTransport.(*http.Transport).MaxIdleConns = 1024 } -// RegisterLocalGrpcSocket registers a Unix socket path for a gRPC port. -// When a gRPC client dials an address on this port, it uses the Unix socket. -func RegisterLocalGrpcSocket(grpcPort int, socketPath string) { +// RegisterLocalGrpcSocket registers a Unix socket path for a gRPC service +// running on host:grpcPort in this process. After this is called, any +// GrpcDial to host:grpcPort (or to a loopback alias of host on the same +// port) is routed through the Unix socket. Dials to any other host on the +// same port still go over TCP. +func RegisterLocalGrpcSocket(host string, grpcPort int, socketPath string) { localGrpcSocketsLock.Lock() defer localGrpcSocketsLock.Unlock() localGrpcSockets[grpcPort] = socketPath + hosts, ok := localGrpcHosts[grpcPort] + if !ok { + hosts = make(map[string]struct{}) + localGrpcHosts[grpcPort] = hosts + } + // The advertised host plus the well-known loopback aliases all refer + // to "this machine"; the empty entry covers SplitHostPort outputs like + // ":17334". + for _, h := range []string{host, "", "localhost", "127.0.0.1", "::1"} { + hosts[h] = struct{}{} + } } // GetLocalGrpcSocket returns the Unix socket path for a gRPC port, or empty if not registered. @@ -78,10 +105,12 @@ func GetLocalGrpcSocket(grpcPort int) string { return localGrpcSockets[grpcPort] } -// resolveLocalGrpcSocket extracts the port from a gRPC address and returns -// the registered Unix socket path, if any. +// resolveLocalGrpcSocket returns the Unix socket path to use for an outgoing +// gRPC dial, or empty if the dial should go over TCP. It matches both the +// host and the port: a remote peer that happens to reuse a local service's +// gRPC port must not be rerouted to the local socket (issue #9254). func resolveLocalGrpcSocket(address string) string { - _, portStr, err := net.SplitHostPort(address) + host, portStr, err := net.SplitHostPort(address) if err != nil { return "" } @@ -89,7 +118,12 @@ func resolveLocalGrpcSocket(address string) string { if err != nil { return "" } - return GetLocalGrpcSocket(port) + localGrpcSocketsLock.RLock() + defer localGrpcSocketsLock.RUnlock() + if _, ok := localGrpcHosts[port][host]; !ok { + return "" + } + return localGrpcSockets[port] } // ServeGrpcOnLocalSocket starts serving a gRPC server on a Unix socket diff --git a/weed/pb/grpc_client_server_test.go b/weed/pb/grpc_client_server_test.go index 5d3f4aa16..210578d68 100644 --- a/weed/pb/grpc_client_server_test.go +++ b/weed/pb/grpc_client_server_test.go @@ -66,3 +66,55 @@ func TestIsClientSideMarshalError_RequiresGrpcStatus(t *testing.T) { t.Fatalf("plain error must not match the marshal-error carve-out") } } + +// TestResolveLocalGrpcSocket_RemotePortCollision is a regression test for +// issue #9254. A `weed server` process registers a Unix socket for its +// in-process volume server on host A. A standalone `weed volume` on host B +// happens to use the same gRPC port. Dials from the master to host B must +// continue out over TCP — they must NOT be hijacked into host A's local +// socket on the basis of port match alone. +func TestResolveLocalGrpcSocket_RemotePortCollision(t *testing.T) { + // Snapshot and restore global state so the test does not leak into others. + localGrpcSocketsLock.Lock() + prevSockets := localGrpcSockets + prevHosts := localGrpcHosts + localGrpcSockets = make(map[int]string) + localGrpcHosts = make(map[int]map[string]struct{}) + localGrpcSocketsLock.Unlock() + t.Cleanup(func() { + localGrpcSocketsLock.Lock() + localGrpcSockets = prevSockets + localGrpcHosts = prevHosts + localGrpcSocketsLock.Unlock() + }) + + const localHost = "10.0.0.2" + const remoteHost = "10.0.0.3" + const collidingPort = 17334 + const socketPath = "/tmp/seaweedfs-volume-grpc-17334.sock" + + RegisterLocalGrpcSocket(localHost, collidingPort, socketPath) + + cases := []struct { + name string + address string + want string + }{ + {"local advertised host routes to socket", localHost + ":17334", socketPath}, + {"loopback v4 routes to socket", "127.0.0.1:17334", socketPath}, + {"localhost routes to socket", "localhost:17334", socketPath}, + {"loopback v6 routes to socket", "[::1]:17334", socketPath}, + {"empty host (bare port) routes to socket", ":17334", socketPath}, + {"remote host with same port stays on TCP", remoteHost + ":17334", ""}, + {"unrelated host with same port stays on TCP", "192.168.1.5:17334", ""}, + {"unregistered port stays on TCP", localHost + ":17335", ""}, + {"malformed address stays on TCP", "not-a-host-port", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := resolveLocalGrpcSocket(tc.address); got != tc.want { + t.Fatalf("resolveLocalGrpcSocket(%q) = %q, want %q", tc.address, got, tc.want) + } + }) + } +}