mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-15 02:28:31 +02:00
* 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.
This commit is contained in:
1 parent
e2f96687ff
commit
135af25b55
4 files changed
+103
-17
No files matched your search
@@ -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)
|
||||
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user