From 0e582932e74030955fb6f6701ae4e5db0712142d Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Sun, 6 Sep 2026 23:45:02 +0200 Subject: [PATCH 1/4] feat: configure gRPC UNIX socket attributes --- config.go | 46 ++++++- go.mod | 2 +- go.sum | 4 +- plugin.go | 11 +- schema.json | 3 + tests/go.mod | 2 +- tests/go.sum | 4 +- tests/unix_socket_test.go | 244 ++++++++++++++++++++++++++++++++++++++ 8 files changed, 307 insertions(+), 9 deletions(-) create mode 100644 tests/unix_socket_test.go diff --git a/config.go b/config.go index 731b6e0..60f8620 100644 --- a/config.go +++ b/config.go @@ -3,14 +3,19 @@ package grpc import ( "crypto/tls" stderr "errors" + "fmt" "math" "os" + "reflect" + "strconv" "strings" "time" "github.com/bmatcuk/doublestar/v4" "github.com/roadrunner-server/errors" + "github.com/roadrunner-server/grpc/v6/api" "github.com/roadrunner-server/pool/v2/pool" + "github.com/roadrunner-server/tcplisten" ) type ClientAuthType string @@ -24,8 +29,9 @@ const ( ) type Config struct { - Listen string `mapstructure:"listen"` - Proto []string `mapstructure:"proto"` + Listen string `mapstructure:"listen"` + UnixSocket *tcplisten.UnixSocketOptions `mapstructure:"unix_socket"` + Proto []string `mapstructure:"proto"` TLS *TLS `mapstructure:"tls"` @@ -64,6 +70,9 @@ func (c *Config) InitDefaults() error { //nolint:gocyclo,gocognit if !strings.Contains(c.Listen, ":") { return errors.E(op, errors.Errorf("malformed grpc address, provided: %s", c.Listen)) } + if err := c.UnixSocket.Validate(c.Listen); err != nil { + return errors.E(op, fmt.Errorf("grpc.unix_socket: %w", err)) + } protos := make([]string, 0, len(c.Proto)) for _, path := range c.Proto { @@ -184,3 +193,36 @@ func (c *Config) EnableTLS() bool { } return false } + +// validateUnixSocketIDs rejects values that weak decoding can convert to valid IDs. +func validateUnixSocketIDs(cfg api.Configurer) error { + const key = pluginName + ".unix_socket" + var options map[string]any + if err := cfg.UnmarshalKey(key, &options); err != nil { + return fmt.Errorf("%s: %w", key, err) + } + for _, field := range []string{"uid", "gid"} { + if options[field] == nil { + continue + } + value := reflect.ValueOf(options[field]) + valid := false + switch { + case value.CanInt(): + id := value.Int() + valid = id >= 0 && id < math.MaxUint32 + case value.CanUint(): + valid = value.Uint() < math.MaxUint32 + case value.Kind() == reflect.String: + id, err := strconv.ParseInt(value.String(), 0, strconv.IntSize) + valid = err == nil && id >= 0 && id < math.MaxUint32 + case value.CanFloat(): + id := value.Float() + valid = id >= 0 && id < math.MaxUint32 && math.Trunc(id) == id + } + if !valid { + return fmt.Errorf("%s.%s: must be an integer between 0 and 4294967294", key, field) + } + } + return nil +} diff --git a/go.mod b/go.mod index d4c1794..6842790 100644 --- a/go.mod +++ b/go.mod @@ -15,7 +15,7 @@ require ( github.com/roadrunner-server/errors v1.5.0 github.com/roadrunner-server/goridge/v4 v4.0.0-beta.3 github.com/roadrunner-server/pool/v2 v2.0.0-beta.1 - github.com/roadrunner-server/tcplisten v1.5.2 + github.com/roadrunner-server/tcplisten v1.6.0 github.com/stretchr/testify v1.12.1 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.71.0 go.opentelemetry.io/contrib/propagators/jaeger v1.46.0 diff --git a/go.sum b/go.sum index 2d2dcc3..db41348 100644 --- a/go.sum +++ b/go.sum @@ -44,8 +44,8 @@ github.com/roadrunner-server/goridge/v4 v4.0.0-beta.3 h1:+kUw00/fpqwdMWrPMYW+OZH github.com/roadrunner-server/goridge/v4 v4.0.0-beta.3/go.mod h1:1aHppV68y/VqRED/AsfNg59sft9aQOhqgr5Z5n49jbM= github.com/roadrunner-server/pool/v2 v2.0.0-beta.1 h1:jpYXFtdD6QGAdAGPgMxrNi3j1CegCRpb2y+A+3GnXFA= github.com/roadrunner-server/pool/v2 v2.0.0-beta.1/go.mod h1:Bo1wT7RtL3eyQHXBUohNhtj/yAmRt6Rq8smuBg5pWkY= -github.com/roadrunner-server/tcplisten v1.5.2 h1:nn8yXYrhRDkfQ9AAu4V075uT4fZRmOnpxkawgE+bWPA= -github.com/roadrunner-server/tcplisten v1.5.2/go.mod h1:DufGBz7Dlx2KrNe/4RukEvGMTqZKB0Uve1GztwcyyR8= +github.com/roadrunner-server/tcplisten v1.6.0 h1:xfFeA2PZTmwJdwc/InhJGq200ew/lfTDReF3oa4AyI4= +github.com/roadrunner-server/tcplisten v1.6.0/go.mod h1:M01BcmhsBiek8WfkiRQwVXwVamgZ5YV36Wa0hz937dA= github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI= github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= diff --git a/plugin.go b/plugin.go index 1453db0..52a9d1d 100644 --- a/plugin.go +++ b/plugin.go @@ -86,6 +86,15 @@ func (p *Plugin) Init(cfg api.Configurer, log api.Logger, server api.Server) err if err != nil { return errors.E(op, err) } + if cfg.Has(pluginName + ".unix_socket") { + if err = validateUnixSocketIDs(cfg); err != nil { + return errors.E(op, err) + } + // Viper can omit an empty map from the decoded section. + if p.config.UnixSocket == nil { + p.config.UnixSocket = &tcplisten.UnixSocketOptions{} + } + } err = p.config.InitDefaults() if err != nil { @@ -168,7 +177,7 @@ func (p *Plugin) Serve() chan error { return errCh } - l, err := tcplisten.CreateListener(p.config.Listen) + l, err := tcplisten.CreateListenerWithOptions(p.config.Listen, p.config.UnixSocket) if err != nil { errCh <- errors.E(op, err) return errCh diff --git a/schema.json b/schema.json index bc630d7..c682480 100644 --- a/schema.json +++ b/schema.json @@ -20,6 +20,9 @@ "tcp://127.0.0.1:${TCP_PORT}" ] }, + "unix_socket": { + "$ref": "https://raw.githubusercontent.com/roadrunner-server/tcplisten/v1.6.0/schema.json" + }, "proto": { "type": "array", "minItems": 1, diff --git a/tests/go.mod b/tests/go.mod index 1b7caae..fb4b391 100644 --- a/tests/go.mod +++ b/tests/go.mod @@ -52,7 +52,7 @@ require ( github.com/roadrunner-server/errors v1.5.0 // indirect github.com/roadrunner-server/events v1.0.1 // indirect github.com/roadrunner-server/pool/v2 v2.0.0-beta.1 // indirect - github.com/roadrunner-server/tcplisten v1.5.2 // indirect + github.com/roadrunner-server/tcplisten v1.6.0 // indirect github.com/sagikazarmark/locafero v0.12.0 // indirect github.com/shirou/gopsutil v3.21.11+incompatible // indirect github.com/spf13/afero v1.15.0 // indirect diff --git a/tests/go.sum b/tests/go.sum index 73ac0af..c42cc08 100644 --- a/tests/go.sum +++ b/tests/go.sum @@ -92,8 +92,8 @@ github.com/roadrunner-server/server/v6 v6.0.0-beta.7 h1:EiRKdWFPOYLoYy53xoLbyU88 github.com/roadrunner-server/server/v6 v6.0.0-beta.7/go.mod h1:uq0yIZgp1v80BGIHPZHKHFyYIWTVJJvofThaC8QWf7w= github.com/roadrunner-server/status/v6 v6.0.0-beta.8 h1:Ya1/vZnPgRUc4eWDTVihLbptS5z9olBUypwXSnH6olw= github.com/roadrunner-server/status/v6 v6.0.0-beta.8/go.mod h1:y/TqFnBItSHmhvnvZD+xqX5hTcIRGcQBoVrltizy9Pg= -github.com/roadrunner-server/tcplisten v1.5.2 h1:nn8yXYrhRDkfQ9AAu4V075uT4fZRmOnpxkawgE+bWPA= -github.com/roadrunner-server/tcplisten v1.5.2/go.mod h1:DufGBz7Dlx2KrNe/4RukEvGMTqZKB0Uve1GztwcyyR8= +github.com/roadrunner-server/tcplisten v1.6.0 h1:xfFeA2PZTmwJdwc/InhJGq200ew/lfTDReF3oa4AyI4= +github.com/roadrunner-server/tcplisten v1.6.0/go.mod h1:M01BcmhsBiek8WfkiRQwVXwVamgZ5YV36Wa0hz937dA= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/sagikazarmark/locafero v0.12.0 h1:/NQhBAkUb4+fH1jivKHWusDYFjMOOKU88eegjfxfHb4= diff --git a/tests/unix_socket_test.go b/tests/unix_socket_test.go new file mode 100644 index 0000000..c95cea2 --- /dev/null +++ b/tests/unix_socket_test.go @@ -0,0 +1,244 @@ +//go:build linux || darwin || freebsd + +package grpc_test + +import ( + "context" + "fmt" + "log/slog" + "net" + "os" + "path/filepath" + "syscall" + "testing" + "time" + + "tests/helpers" + mocklogger "tests/mock" + + "github.com/roadrunner-server/config/v6" + grpcPlugin "github.com/roadrunner-server/grpc/v6" + "github.com/roadrunner-server/server/v6" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + grpchealth "google.golang.org/grpc/health/grpc_health_v1" +) + +func TestUnixSocketConfig(t *testing.T) { + t.Chdir(t.TempDir()) + log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) + + for _, tc := range []struct { + name string + listen string + options string + flags []string + mode string + zeroIDs bool + wantErr string + }{ + {name: "omitted", listen: "tcp://127.0.0.1:0"}, + {name: "null", listen: "tcp://127.0.0.1:0", options: "null"}, + {name: "quoted mode", listen: "unix://grpc.sock", options: `{mode: "0000"}`, mode: "0000"}, + {name: "zero IDs", listen: "unix://grpc.sock", options: "{uid: 0, gid: 0}", zeroIDs: true}, + {name: "string overrides", listen: "unix://grpc.sock", options: `{mode: "0600"}`, flags: []string{"grpc.unix_socket.mode=0640", "grpc.unix_socket.uid=0", "grpc.unix_socket.gid=0"}, mode: "0640", zeroIDs: true}, + {name: "TCP options", listen: "tcp://127.0.0.1:0", options: `{mode: "0600"}`, wantErr: "grpc.unix_socket"}, + {name: "empty TCP options", listen: "tcp://127.0.0.1:0", options: "{}", wantErr: "grpc.unix_socket"}, + {name: "invalid mode", listen: "unix://grpc.sock", options: `{mode: "0780"}`, wantErr: "grpc.unix_socket"}, + {name: "unquoted mode", listen: "unix://grpc.sock", options: "{mode: 0600}", wantErr: "grpc.unix_socket"}, + {name: "empty address", options: "{}", wantErr: "malformed grpc address"}, + {name: "empty socket address", listen: "unix://", options: "{}", wantErr: "grpc.unix_socket"}, + } { + t.Run(tc.name, func(t *testing.T) { + cfg := unixSocketConfig(t, tc.listen, tc.options, tc.flags) + var decoded grpcPlugin.Config + require.NoError(t, cfg.UnmarshalKey("grpc", &decoded)) + + p := &grpcPlugin.Plugin{} + err := p.Init(cfg, log, nil) + if tc.wantErr != "" { + require.ErrorContains(t, err, tc.wantErr) + require.NoFileExists(t, "grpc.sock") + return + } + require.NoError(t, err) + require.NoError(t, decoded.InitDefaults()) + require.Equal(t, tc.listen, decoded.Listen) + if tc.options == "" || tc.options == "null" { + require.Nil(t, decoded.UnixSocket) + return + } + require.NotNil(t, decoded.UnixSocket) + require.Equal(t, tc.mode, decoded.UnixSocket.Mode) + if tc.zeroIDs { + require.NotNil(t, decoded.UnixSocket.UID) + require.NotNil(t, decoded.UnixSocket.GID) + require.Zero(t, *decoded.UnixSocket.UID) + require.Zero(t, *decoded.UnixSocket.GID) + } else { + require.Nil(t, decoded.UnixSocket.UID) + require.Nil(t, decoded.UnixSocket.GID) + } + }) + } +} + +func TestUnixSocketIDs(t *testing.T) { + t.Chdir(t.TempDir()) + t.Setenv("RR_TEST_UNIX_SOCKET_ID", "33") + t.Setenv("RR_TEST_UNIX_SOCKET_UNSET_ID", "") + require.NoError(t, os.Unsetenv("RR_TEST_UNIX_SOCKET_UNSET_ID")) + log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) + + for _, field := range []string{"uid", "gid"} { + for _, tc := range []struct { + name string + value string + valid bool + want int + json bool + parentError bool + }{ + {name: "false", value: "false"}, + {name: "true", value: "true"}, + {name: "fraction", value: "1.9"}, + {name: "negative fraction", value: "-0.5"}, + {name: "empty string", value: `""`}, + {name: "unset environment", value: `"${RR_TEST_UNIX_SOCKET_UNSET_ID}"`}, + {name: "negative ID", value: "-1"}, + {name: "oversized ID", value: "4294967295"}, + {name: "oversized float", value: "4294967295.0"}, + {name: "oversized unsigned ID", value: "18446744073709551615"}, + {name: "oversized string ID", value: `"4294967295"`}, + {name: "string overflow", value: `"18446744073709551616"`, parentError: true}, + {name: "NaN", value: ".nan"}, + {name: "infinity", value: ".inf"}, + {name: "map", value: "{id: 33}", parentError: true}, + {name: "slice", value: "[33]", parentError: true}, + {name: "JSON fraction", value: "1.9", json: true}, + {name: "zero", value: "0", valid: true}, + {name: "null", value: "null", valid: true}, + {name: "integer", value: "33", valid: true, want: 33}, + {name: "integer float", value: "33.0", valid: true, want: 33}, + {name: "JSON integer float", value: "33.0", valid: true, want: 33, json: true}, + {name: "environment", value: `"${RR_TEST_UNIX_SOCKET_ID}"`, valid: true, want: 33}, + {name: "octal string", value: `"041"`, valid: true, want: 33}, + {name: "hexadecimal string", value: `"0x21"`, valid: true, want: 33}, + } { + t.Run(field+"/"+tc.name, func(t *testing.T) { + var cfg *config.Plugin + if tc.json { + path := filepath.Join(t.TempDir(), ".rr.json") + contents := fmt.Sprintf(`{"version":"3","grpc":{"listen":"unix://grpc.sock","unix_socket":{"mode":"0600",%q:%s}}}`, field, tc.value) + require.NoError(t, os.WriteFile(path, []byte(contents), 0o600)) + cfg = &config.Plugin{Path: path} + require.NoError(t, cfg.Init()) + } else { + options := fmt.Sprintf(`{mode: "0600", %s: %s}`, field, tc.value) + cfg = unixSocketConfig(t, "unix://grpc.sock", options, nil) + } + p := &grpcPlugin.Plugin{} + err := p.Init(cfg, log, nil) + require.NoFileExists(t, "grpc.sock") + if !tc.valid { + prefix := "grpc.unix_socket." + if tc.parentError { + prefix = "unix_socket." + } + require.ErrorContains(t, err, prefix+field) + return + } + require.NoError(t, err) + var decoded grpcPlugin.Config + require.NoError(t, cfg.UnmarshalKey("grpc", &decoded)) + require.NotNil(t, decoded.UnixSocket) + id, other := decoded.UnixSocket.UID, decoded.UnixSocket.GID + if field == "gid" { + id, other = other, id + } + require.Nil(t, other) + if tc.value == "null" { + require.Nil(t, id) + } else { + require.NotNil(t, id) + require.Equal(t, tc.want, *id) + } + }) + } + } +} + +func TestUnixSocketServe(t *testing.T) { + for _, tc := range []struct { + name string + flags []string + mode os.FileMode + }{ + {name: "TCP without options"}, + {name: "quoted mode", mode: 0o600}, + {name: "string override", flags: []string{"grpc.unix_socket.mode=0640"}, mode: 0o640}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Chdir(t.TempDir()) + listen, target := "unix://grpc.sock", "unix:grpc.sock" + options := fmt.Sprintf(`{mode: "0600", uid: %d, gid: %d}`, os.Getuid(), os.Getgid()) + if tc.mode == 0 { + var lc net.ListenConfig + ln, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + target = ln.Addr().String() + listen, options = "tcp://"+target, "" + require.NoError(t, ln.Close()) + } + cfg := unixSocketConfig(t, listen, options, tc.flags) + log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) + rrServer := &server.Plugin{} + require.NoError(t, rrServer.Init(cfg, log)) + t.Cleanup(func() { require.NoError(t, rrServer.Stop(context.Background())) }) + p := &grpcPlugin.Plugin{} + require.NoError(t, p.Init(cfg, log, rrServer)) + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + require.NoError(t, p.Stop(ctx)) + }) + errCh := p.Serve() + select { + case err := <-errCh: + t.Fatalf("gRPC serve: %v", err) + default: + } + require.Empty(t, p.Workers()) + + if tc.mode != 0 { + info, err := os.Stat("grpc.sock") + require.NoError(t, err) + require.Equal(t, tc.mode, info.Mode().Perm()) + stat, ok := info.Sys().(*syscall.Stat_t) + require.True(t, ok) + require.EqualValues(t, os.Getuid(), stat.Uid) + require.EqualValues(t, os.Getgid(), stat.Gid) + } + + conn := helpers.Dial(t, target) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + health, err := grpchealth.NewHealthClient(conn).Check(ctx, &grpchealth.HealthCheckRequest{}, grpc.WaitForReady(true)) + require.NoError(t, err) + require.Equal(t, grpchealth.HealthCheckResponse_SERVING, health.GetStatus()) + }) + } +} + +func unixSocketConfig(t *testing.T, listen, options string, flags []string) *config.Plugin { + t.Helper() + contents := fmt.Sprintf("version: \"3\"\nserver:\n command: [unused]\ngrpc:\n listen: %q\n pool:\n debug: true\n", listen) + if options != "" { + contents += " unix_socket: " + options + "\n" + } + path := filepath.Join(t.TempDir(), ".rr.yaml") + require.NoError(t, os.WriteFile(path, []byte(contents), 0o600)) + cfg := &config.Plugin{Path: path, Flags: flags} + require.NoError(t, cfg.Init()) + return cfg +} From 5ac4531b9683328f84d627a5c03e7d6122cc41d0 Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Mon, 7 Sep 2026 11:50:33 +0200 Subject: [PATCH 2/4] refactor: simplify UNIX socket configuration --- config.go | 36 ------ config_test.go | 11 ++ plugin.go | 10 -- tests/unix_socket_test.go | 256 +++++++++++++++++--------------------- 4 files changed, 125 insertions(+), 188 deletions(-) diff --git a/config.go b/config.go index 60f8620..c41c07b 100644 --- a/config.go +++ b/config.go @@ -6,14 +6,11 @@ import ( "fmt" "math" "os" - "reflect" - "strconv" "strings" "time" "github.com/bmatcuk/doublestar/v4" "github.com/roadrunner-server/errors" - "github.com/roadrunner-server/grpc/v6/api" "github.com/roadrunner-server/pool/v2/pool" "github.com/roadrunner-server/tcplisten" ) @@ -193,36 +190,3 @@ func (c *Config) EnableTLS() bool { } return false } - -// validateUnixSocketIDs rejects values that weak decoding can convert to valid IDs. -func validateUnixSocketIDs(cfg api.Configurer) error { - const key = pluginName + ".unix_socket" - var options map[string]any - if err := cfg.UnmarshalKey(key, &options); err != nil { - return fmt.Errorf("%s: %w", key, err) - } - for _, field := range []string{"uid", "gid"} { - if options[field] == nil { - continue - } - value := reflect.ValueOf(options[field]) - valid := false - switch { - case value.CanInt(): - id := value.Int() - valid = id >= 0 && id < math.MaxUint32 - case value.CanUint(): - valid = value.Uint() < math.MaxUint32 - case value.Kind() == reflect.String: - id, err := strconv.ParseInt(value.String(), 0, strconv.IntSize) - valid = err == nil && id >= 0 && id < math.MaxUint32 - case value.CanFloat(): - id := value.Float() - valid = id >= 0 && id < math.MaxUint32 && math.Trunc(id) == id - } - if !valid { - return fmt.Errorf("%s.%s: must be an integer between 0 and 4294967294", key, field) - } - } - return nil -} diff --git a/config_test.go b/config_test.go index 9198969..8138e13 100644 --- a/config_test.go +++ b/config_test.go @@ -5,10 +5,21 @@ import ( "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) const separator = string(filepath.Separator) +func TestConfigUnixSocketDefaults(t *testing.T) { + for _, listen := range []string{"127.0.0.1:0", "tcp://127.0.0.1:0", "unix://grpc.sock"} { + t.Run(listen, func(t *testing.T) { + cfg := &Config{Listen: listen} + require.NoError(t, cfg.InitDefaults()) + require.Nil(t, cfg.UnixSocket) + }) + } +} + func TestInitDefaults(t *testing.T) { c := Config{} assert.Error(t, c.InitDefaults()) diff --git a/plugin.go b/plugin.go index 52a9d1d..15f5988 100644 --- a/plugin.go +++ b/plugin.go @@ -86,16 +86,6 @@ func (p *Plugin) Init(cfg api.Configurer, log api.Logger, server api.Server) err if err != nil { return errors.E(op, err) } - if cfg.Has(pluginName + ".unix_socket") { - if err = validateUnixSocketIDs(cfg); err != nil { - return errors.E(op, err) - } - // Viper can omit an empty map from the decoded section. - if p.config.UnixSocket == nil { - p.config.UnixSocket = &tcplisten.UnixSocketOptions{} - } - } - err = p.config.InitDefaults() if err != nil { return errors.E(op, err) diff --git a/tests/unix_socket_test.go b/tests/unix_socket_test.go index c95cea2..3a08262 100644 --- a/tests/unix_socket_test.go +++ b/tests/unix_socket_test.go @@ -9,12 +9,17 @@ import ( "net" "os" "path/filepath" + "slices" + "strconv" + "strings" + "sync" "syscall" "testing" "time" "tests/helpers" mocklogger "tests/mock" + "tests/proto/service" "github.com/roadrunner-server/config/v6" grpcPlugin "github.com/roadrunner-server/grpc/v6" @@ -25,150 +30,53 @@ import ( ) func TestUnixSocketConfig(t *testing.T) { - t.Chdir(t.TempDir()) log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) for _, tc := range []struct { name string listen string options string - flags []string - mode string - zeroIDs bool wantErr string }{ - {name: "omitted", listen: "tcp://127.0.0.1:0"}, - {name: "null", listen: "tcp://127.0.0.1:0", options: "null"}, - {name: "quoted mode", listen: "unix://grpc.sock", options: `{mode: "0000"}`, mode: "0000"}, - {name: "zero IDs", listen: "unix://grpc.sock", options: "{uid: 0, gid: 0}", zeroIDs: true}, - {name: "string overrides", listen: "unix://grpc.sock", options: `{mode: "0600"}`, flags: []string{"grpc.unix_socket.mode=0640", "grpc.unix_socket.uid=0", "grpc.unix_socket.gid=0"}, mode: "0640", zeroIDs: true}, - {name: "TCP options", listen: "tcp://127.0.0.1:0", options: `{mode: "0600"}`, wantErr: "grpc.unix_socket"}, - {name: "empty TCP options", listen: "tcp://127.0.0.1:0", options: "{}", wantErr: "grpc.unix_socket"}, - {name: "invalid mode", listen: "unix://grpc.sock", options: `{mode: "0780"}`, wantErr: "grpc.unix_socket"}, - {name: "unquoted mode", listen: "unix://grpc.sock", options: "{mode: 0600}", wantErr: "grpc.unix_socket"}, - {name: "empty address", options: "{}", wantErr: "malformed grpc address"}, - {name: "empty socket address", listen: "unix://", options: "{}", wantErr: "grpc.unix_socket"}, + {name: "TCP defaults", listen: "tcp://127.0.0.1:0"}, + {name: "UNIX defaults", listen: "unix://grpc.sock"}, + {name: "empty options", listen: "unix://grpc.sock", options: "{}"}, + {name: "TCP empty options", listen: "tcp://127.0.0.1:0", options: "{}"}, + {name: "mode only", listen: "unix://grpc.sock", options: `{mode: "0600"}`}, + {name: "explicit zero", listen: "unix://grpc.sock", options: `{mode: "0000", uid: 0, gid: 0}`}, + {name: "unset mode", listen: "unix://grpc.sock", options: "{uid: 0, gid: 0}"}, + {name: "TCP options", listen: "tcp://127.0.0.1:0", options: `{mode: "0600"}`, wantErr: "filesystem unix:// address"}, + {name: "invalid mode", listen: "unix://grpc.sock", options: `{mode: "0780"}`, wantErr: "invalid unix socket mode"}, + {name: "unquoted mode", listen: "unix://grpc.sock", options: "{mode: 0600}", wantErr: "invalid unix socket mode"}, + {name: "empty address", options: `{mode: "0600"}`, wantErr: "malformed grpc address"}, + {name: "empty socket path", listen: "unix://", options: `{mode: "0600"}`, wantErr: "filesystem unix:// address"}, + {name: "scalar options", listen: "unix://grpc.sock", options: "false", wantErr: "expected a map"}, + {name: "negative UID", listen: "unix://grpc.sock", options: "{uid: -1}", wantErr: "invalid unix socket uid"}, + {name: "negative GID", listen: "unix://grpc.sock", options: "{gid: -1}", wantErr: "invalid unix socket gid"}, + {name: "reserved UID", listen: "unix://grpc.sock", options: "{uid: 4294967295}", wantErr: "invalid unix socket uid"}, + {name: "reserved GID", listen: "unix://grpc.sock", options: "{gid: 4294967295}", wantErr: "invalid unix socket gid"}, } { t.Run(tc.name, func(t *testing.T) { - cfg := unixSocketConfig(t, tc.listen, tc.options, tc.flags) - var decoded grpcPlugin.Config - require.NoError(t, cfg.UnmarshalKey("grpc", &decoded)) - + cfg := unixSocketConfig(t, tc.listen, tc.options, nil) p := &grpcPlugin.Plugin{} err := p.Init(cfg, log, nil) if tc.wantErr != "" { require.ErrorContains(t, err, tc.wantErr) - require.NoFileExists(t, "grpc.sock") return } require.NoError(t, err) - require.NoError(t, decoded.InitDefaults()) - require.Equal(t, tc.listen, decoded.Listen) - if tc.options == "" || tc.options == "null" { - require.Nil(t, decoded.UnixSocket) - return - } - require.NotNil(t, decoded.UnixSocket) - require.Equal(t, tc.mode, decoded.UnixSocket.Mode) - if tc.zeroIDs { - require.NotNil(t, decoded.UnixSocket.UID) - require.NotNil(t, decoded.UnixSocket.GID) - require.Zero(t, *decoded.UnixSocket.UID) - require.Zero(t, *decoded.UnixSocket.GID) - } else { - require.Nil(t, decoded.UnixSocket.UID) - require.Nil(t, decoded.UnixSocket.GID) - } }) } } -func TestUnixSocketIDs(t *testing.T) { - t.Chdir(t.TempDir()) - t.Setenv("RR_TEST_UNIX_SOCKET_ID", "33") - t.Setenv("RR_TEST_UNIX_SOCKET_UNSET_ID", "") - require.NoError(t, os.Unsetenv("RR_TEST_UNIX_SOCKET_UNSET_ID")) - log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) - - for _, field := range []string{"uid", "gid"} { - for _, tc := range []struct { - name string - value string - valid bool - want int - json bool - parentError bool - }{ - {name: "false", value: "false"}, - {name: "true", value: "true"}, - {name: "fraction", value: "1.9"}, - {name: "negative fraction", value: "-0.5"}, - {name: "empty string", value: `""`}, - {name: "unset environment", value: `"${RR_TEST_UNIX_SOCKET_UNSET_ID}"`}, - {name: "negative ID", value: "-1"}, - {name: "oversized ID", value: "4294967295"}, - {name: "oversized float", value: "4294967295.0"}, - {name: "oversized unsigned ID", value: "18446744073709551615"}, - {name: "oversized string ID", value: `"4294967295"`}, - {name: "string overflow", value: `"18446744073709551616"`, parentError: true}, - {name: "NaN", value: ".nan"}, - {name: "infinity", value: ".inf"}, - {name: "map", value: "{id: 33}", parentError: true}, - {name: "slice", value: "[33]", parentError: true}, - {name: "JSON fraction", value: "1.9", json: true}, - {name: "zero", value: "0", valid: true}, - {name: "null", value: "null", valid: true}, - {name: "integer", value: "33", valid: true, want: 33}, - {name: "integer float", value: "33.0", valid: true, want: 33}, - {name: "JSON integer float", value: "33.0", valid: true, want: 33, json: true}, - {name: "environment", value: `"${RR_TEST_UNIX_SOCKET_ID}"`, valid: true, want: 33}, - {name: "octal string", value: `"041"`, valid: true, want: 33}, - {name: "hexadecimal string", value: `"0x21"`, valid: true, want: 33}, - } { - t.Run(field+"/"+tc.name, func(t *testing.T) { - var cfg *config.Plugin - if tc.json { - path := filepath.Join(t.TempDir(), ".rr.json") - contents := fmt.Sprintf(`{"version":"3","grpc":{"listen":"unix://grpc.sock","unix_socket":{"mode":"0600",%q:%s}}}`, field, tc.value) - require.NoError(t, os.WriteFile(path, []byte(contents), 0o600)) - cfg = &config.Plugin{Path: path} - require.NoError(t, cfg.Init()) - } else { - options := fmt.Sprintf(`{mode: "0600", %s: %s}`, field, tc.value) - cfg = unixSocketConfig(t, "unix://grpc.sock", options, nil) - } - p := &grpcPlugin.Plugin{} - err := p.Init(cfg, log, nil) - require.NoFileExists(t, "grpc.sock") - if !tc.valid { - prefix := "grpc.unix_socket." - if tc.parentError { - prefix = "unix_socket." - } - require.ErrorContains(t, err, prefix+field) - return - } - require.NoError(t, err) - var decoded grpcPlugin.Config - require.NoError(t, cfg.UnmarshalKey("grpc", &decoded)) - require.NotNil(t, decoded.UnixSocket) - id, other := decoded.UnixSocket.UID, decoded.UnixSocket.GID - if field == "gid" { - id, other = other, id - } - require.Nil(t, other) - if tc.value == "null" { - require.Nil(t, id) - } else { - require.NotNil(t, id) - require.Equal(t, tc.want, *id) - } - }) - } - } -} - func TestUnixSocketServe(t *testing.T) { + worker, err := filepath.Abs("php_test_files/worker-grpc.php") + require.NoError(t, err) + proto, err := filepath.Abs("proto/service/service.proto") + require.NoError(t, err) + t.Setenv("RR_TEST_SOCKET_UID", strconv.Itoa(os.Getuid())) + t.Setenv("RR_TEST_SOCKET_GID", strconv.Itoa(os.Getgid())) + for _, tc := range []struct { name string flags []string @@ -181,58 +89,122 @@ func TestUnixSocketServe(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Chdir(t.TempDir()) listen, target := "unix://grpc.sock", "unix:grpc.sock" - options := fmt.Sprintf(`{mode: "0600", uid: %d, gid: %d}`, os.Getuid(), os.Getgid()) + options := `{mode: "0600", uid: "${RR_TEST_SOCKET_UID}", gid: "${RR_TEST_SOCKET_GID}"}` if tc.mode == 0 { var lc net.ListenConfig - ln, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") - require.NoError(t, err) + ln, errL := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, errL) target = ln.Addr().String() listen, options = "tcp://"+target, "" require.NoError(t, ln.Close()) } - cfg := unixSocketConfig(t, listen, options, tc.flags) + flags := append([]string{"server.command=php " + worker, "grpc.proto=" + proto}, tc.flags...) + cfg := unixSocketConfig(t, listen, options, flags) log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) rrServer := &server.Plugin{} require.NoError(t, rrServer.Init(cfg, log)) t.Cleanup(func() { require.NoError(t, rrServer.Stop(context.Background())) }) p := &grpcPlugin.Plugin{} require.NoError(t, p.Init(cfg, log, rrServer)) - t.Cleanup(func() { + stop := sync.OnceValue(func() error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - require.NoError(t, p.Stop(ctx)) + return p.Stop(ctx) }) + t.Cleanup(func() { require.NoError(t, stop()) }) errCh := p.Serve() select { - case err := <-errCh: - t.Fatalf("gRPC serve: %v", err) + case errS := <-errCh: + t.Fatalf("gRPC serve: %v", errS) default: } - require.Empty(t, p.Workers()) + + conn := helpers.Dial(t, target) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + response, errR := service.NewEchoClient(conn).Ping(ctx, &service.Message{Msg: tc.name}, grpc.WaitForReady(true)) + require.NoError(t, errR) + require.Equal(t, strings.ToUpper(tc.name), response.GetMsg()) + health, errH := grpchealth.NewHealthClient(conn).Check(ctx, &grpchealth.HealthCheckRequest{}) + require.NoError(t, errH) + require.Equal(t, grpchealth.HealthCheckResponse_SERVING, health.GetStatus()) if tc.mode != 0 { - info, err := os.Stat("grpc.sock") - require.NoError(t, err) + info, errS := os.Stat("grpc.sock") + require.NoError(t, errS) + require.NotZero(t, info.Mode()&os.ModeSocket) require.Equal(t, tc.mode, info.Mode().Perm()) - stat, ok := info.Sys().(*syscall.Stat_t) - require.True(t, ok) + stat := info.Sys().(*syscall.Stat_t) require.EqualValues(t, os.Getuid(), stat.Uid) require.EqualValues(t, os.Getgid(), stat.Gid) } + require.NoError(t, conn.Close()) + require.NoError(t, stop()) + if tc.mode != 0 { + _, errS := os.Stat("grpc.sock") + require.ErrorIs(t, errS, os.ErrNotExist) + } + }) + } +} - conn := helpers.Dial(t, target) - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) - defer cancel() - health, err := grpchealth.NewHealthClient(conn).Check(ctx, &grpchealth.HealthCheckRequest{}, grpc.WaitForReady(true)) - require.NoError(t, err) - require.Equal(t, grpchealth.HealthCheckResponse_SERVING, health.GetStatus()) +func TestUnixSocketOwnershipError(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("Requires an unprivileged process.") + } + groups, err := os.Getgroups() + require.NoError(t, err) + otherGID := 0 + for otherGID == os.Getegid() || slices.Contains(groups, otherGID) { + otherGID++ + } + + for _, tc := range []struct { + field string + id int + }{ + {field: "uid", id: 0}, + {field: "gid", id: otherGID}, + } { + t.Run(tc.field, func(t *testing.T) { + t.Chdir(t.TempDir()) + t.Setenv("RR_TEST_SOCKET_ID", strconv.Itoa(tc.id)) + options := fmt.Sprintf(`{%s: "${RR_TEST_SOCKET_ID}"}`, tc.field) + cfg := unixSocketConfig(t, "unix://ownership.sock", options, nil) + log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) + rrServer := &server.Plugin{} + require.NoError(t, rrServer.Init(cfg, log)) + t.Cleanup(func() { require.NoError(t, rrServer.Stop(context.Background())) }) + p := &grpcPlugin.Plugin{} + require.NoError(t, p.Init(cfg, log, rrServer)) + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + require.NoError(t, p.Stop(ctx)) + }) + select { + case errS := <-p.Serve(): + require.ErrorContains(t, errS, "chown unix socket") + case <-time.After(5 * time.Second): + t.Fatal("gRPC did not report the ownership error") + } + _, errS := os.Stat("ownership.sock") + require.ErrorIs(t, errS, os.ErrNotExist) }) } } func unixSocketConfig(t *testing.T, listen, options string, flags []string) *config.Plugin { t.Helper() - contents := fmt.Sprintf("version: \"3\"\nserver:\n command: [unused]\ngrpc:\n listen: %q\n pool:\n debug: true\n", listen) + contents := fmt.Sprintf(`version: "3" +server: + command: [unused] +grpc: + listen: %q + pool: + debug: true + destroy_timeout: 5s +`, listen) if options != "" { contents += " unix_socket: " + options + "\n" } From b81e1098888e971089fe33238a7dd61cd00ed4a9 Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Mon, 7 Sep 2026 17:36:02 +0200 Subject: [PATCH 3/4] test: simplify UNIX socket coverage --- config_test.go | 11 ----------- tests/unix_socket_test.go | 40 +++++++++++++-------------------------- 2 files changed, 13 insertions(+), 38 deletions(-) diff --git a/config_test.go b/config_test.go index 8138e13..9198969 100644 --- a/config_test.go +++ b/config_test.go @@ -5,21 +5,10 @@ import ( "testing" "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) const separator = string(filepath.Separator) -func TestConfigUnixSocketDefaults(t *testing.T) { - for _, listen := range []string{"127.0.0.1:0", "tcp://127.0.0.1:0", "unix://grpc.sock"} { - t.Run(listen, func(t *testing.T) { - cfg := &Config{Listen: listen} - require.NoError(t, cfg.InitDefaults()) - require.Nil(t, cfg.UnixSocket) - }) - } -} - func TestInitDefaults(t *testing.T) { c := Config{} assert.Error(t, c.InitDefaults()) diff --git a/tests/unix_socket_test.go b/tests/unix_socket_test.go index 3a08262..6d08cb2 100644 --- a/tests/unix_socket_test.go +++ b/tests/unix_socket_test.go @@ -6,7 +6,6 @@ import ( "context" "fmt" "log/slog" - "net" "os" "path/filepath" "slices" @@ -82,24 +81,14 @@ func TestUnixSocketServe(t *testing.T) { flags []string mode os.FileMode }{ - {name: "TCP without options"}, {name: "quoted mode", mode: 0o600}, {name: "string override", flags: []string{"grpc.unix_socket.mode=0640"}, mode: 0o640}, } { t.Run(tc.name, func(t *testing.T) { t.Chdir(t.TempDir()) - listen, target := "unix://grpc.sock", "unix:grpc.sock" - options := `{mode: "0600", uid: "${RR_TEST_SOCKET_UID}", gid: "${RR_TEST_SOCKET_GID}"}` - if tc.mode == 0 { - var lc net.ListenConfig - ln, errL := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") - require.NoError(t, errL) - target = ln.Addr().String() - listen, options = "tcp://"+target, "" - require.NoError(t, ln.Close()) - } + const options = `{mode: "0600", uid: "${RR_TEST_SOCKET_UID}", gid: "${RR_TEST_SOCKET_GID}"}` flags := append([]string{"server.command=php " + worker, "grpc.proto=" + proto}, tc.flags...) - cfg := unixSocketConfig(t, listen, options, flags) + cfg := unixSocketConfig(t, "unix://grpc.sock", options, flags) log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) rrServer := &server.Plugin{} require.NoError(t, rrServer.Init(cfg, log)) @@ -119,7 +108,7 @@ func TestUnixSocketServe(t *testing.T) { default: } - conn := helpers.Dial(t, target) + conn := helpers.Dial(t, "unix:grpc.sock") ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) defer cancel() response, errR := service.NewEchoClient(conn).Ping(ctx, &service.Message{Msg: tc.name}, grpc.WaitForReady(true)) @@ -129,21 +118,17 @@ func TestUnixSocketServe(t *testing.T) { require.NoError(t, errH) require.Equal(t, grpchealth.HealthCheckResponse_SERVING, health.GetStatus()) - if tc.mode != 0 { - info, errS := os.Stat("grpc.sock") - require.NoError(t, errS) - require.NotZero(t, info.Mode()&os.ModeSocket) - require.Equal(t, tc.mode, info.Mode().Perm()) - stat := info.Sys().(*syscall.Stat_t) - require.EqualValues(t, os.Getuid(), stat.Uid) - require.EqualValues(t, os.Getgid(), stat.Gid) - } + info, errS := os.Stat("grpc.sock") + require.NoError(t, errS) + require.NotZero(t, info.Mode()&os.ModeSocket) + require.Equal(t, tc.mode, info.Mode().Perm()) + stat := info.Sys().(*syscall.Stat_t) + require.EqualValues(t, os.Getuid(), stat.Uid) + require.EqualValues(t, os.Getgid(), stat.Gid) require.NoError(t, conn.Close()) require.NoError(t, stop()) - if tc.mode != 0 { - _, errS := os.Stat("grpc.sock") - require.ErrorIs(t, errS, os.ErrNotExist) - } + _, errS = os.Stat("grpc.sock") + require.ErrorIs(t, errS, os.ErrNotExist) }) } } @@ -196,6 +181,7 @@ func TestUnixSocketOwnershipError(t *testing.T) { func unixSocketConfig(t *testing.T, listen, options string, flags []string) *config.Plugin { t.Helper() + contents := fmt.Sprintf(`version: "3" server: command: [unused] From 682eef9cfd0b8e5df2c006f8a65fdc3bcf41deb6 Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Mon, 7 Sep 2026 19:02:52 +0200 Subject: [PATCH 4/4] test: split UNIX socket checks by behavior --- tests/helpers/rr.go | 8 +- tests/unix_socket_test.go | 199 +++++++++++++------------------------- 2 files changed, 73 insertions(+), 134 deletions(-) diff --git a/tests/helpers/rr.go b/tests/helpers/rr.go index 03b48cb..516be88 100644 --- a/tests/helpers/rr.go +++ b/tests/helpers/rr.go @@ -29,6 +29,7 @@ const ( // bootCfg holds the options applied to a container before it is started. type bootCfg struct { version string + flags []string logLevel slog.Level logger loggerKind probe func(ctx context.Context) bool @@ -50,6 +51,11 @@ func WithConfigVersion(v string) Option { return func(b *bootCfg) { b.version = v } } +// WithConfigFlags applies configuration overrides. +func WithConfigFlags(flags ...string) Option { + return func(b *bootCfg) { b.flags = flags } +} + // WithLogLevel sets the endure container log level (debug by default). func WithLogLevel(l slog.Level) Option { return func(b *bootCfg) { b.logLevel = l } @@ -171,7 +177,7 @@ func newContainer(t *testing.T, cfgPath string, plugins []any, opts []Option) (* o(bc) } - cfg := &config.Plugin{Version: bc.version, Path: cfgPath} + cfg := &config.Plugin{Version: bc.version, Path: cfgPath, Flags: bc.flags} rr := &RR{} all := []any{cfg} diff --git a/tests/unix_socket_test.go b/tests/unix_socket_test.go index 6d08cb2..2e4d57b 100644 --- a/tests/unix_socket_test.go +++ b/tests/unix_socket_test.go @@ -3,32 +3,21 @@ package grpc_test import ( - "context" "fmt" "log/slog" "os" "path/filepath" - "slices" - "strconv" - "strings" - "sync" - "syscall" "testing" - "time" "tests/helpers" mocklogger "tests/mock" - "tests/proto/service" "github.com/roadrunner-server/config/v6" grpcPlugin "github.com/roadrunner-server/grpc/v6" - "github.com/roadrunner-server/server/v6" "github.com/stretchr/testify/require" - "google.golang.org/grpc" - grpchealth "google.golang.org/grpc/health/grpc_health_v1" ) -func TestUnixSocketConfig(t *testing.T) { +func TestUnixSocketConfigRejectsInvalidOptions(t *testing.T) { log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) for _, tc := range []struct { @@ -37,17 +26,9 @@ func TestUnixSocketConfig(t *testing.T) { options string wantErr string }{ - {name: "TCP defaults", listen: "tcp://127.0.0.1:0"}, - {name: "UNIX defaults", listen: "unix://grpc.sock"}, - {name: "empty options", listen: "unix://grpc.sock", options: "{}"}, - {name: "TCP empty options", listen: "tcp://127.0.0.1:0", options: "{}"}, - {name: "mode only", listen: "unix://grpc.sock", options: `{mode: "0600"}`}, - {name: "explicit zero", listen: "unix://grpc.sock", options: `{mode: "0000", uid: 0, gid: 0}`}, - {name: "unset mode", listen: "unix://grpc.sock", options: "{uid: 0, gid: 0}"}, {name: "TCP options", listen: "tcp://127.0.0.1:0", options: `{mode: "0600"}`, wantErr: "filesystem unix:// address"}, {name: "invalid mode", listen: "unix://grpc.sock", options: `{mode: "0780"}`, wantErr: "invalid unix socket mode"}, {name: "unquoted mode", listen: "unix://grpc.sock", options: "{mode: 0600}", wantErr: "invalid unix socket mode"}, - {name: "empty address", options: `{mode: "0600"}`, wantErr: "malformed grpc address"}, {name: "empty socket path", listen: "unix://", options: `{mode: "0600"}`, wantErr: "filesystem unix:// address"}, {name: "scalar options", listen: "unix://grpc.sock", options: "false", wantErr: "expected a map"}, {name: "negative UID", listen: "unix://grpc.sock", options: "{uid: -1}", wantErr: "invalid unix socket uid"}, @@ -56,147 +37,99 @@ func TestUnixSocketConfig(t *testing.T) { {name: "reserved GID", listen: "unix://grpc.sock", options: "{gid: 4294967295}", wantErr: "invalid unix socket gid"}, } { t.Run(tc.name, func(t *testing.T) { - cfg := unixSocketConfig(t, tc.listen, tc.options, nil) + cfg := &config.Plugin{Path: unixSocketConfig(t, tc.listen, tc.options)} + require.NoError(t, cfg.Init()) p := &grpcPlugin.Plugin{} - err := p.Init(cfg, log, nil) - if tc.wantErr != "" { - require.ErrorContains(t, err, tc.wantErr) - return - } - require.NoError(t, err) + require.ErrorContains(t, p.Init(cfg, log, nil), tc.wantErr) }) } } -func TestUnixSocketServe(t *testing.T) { - worker, err := filepath.Abs("php_test_files/worker-grpc.php") - require.NoError(t, err) - proto, err := filepath.Abs("proto/service/service.proto") - require.NoError(t, err) - t.Setenv("RR_TEST_SOCKET_UID", strconv.Itoa(os.Getuid())) - t.Setenv("RR_TEST_SOCKET_GID", strconv.Itoa(os.Getgid())) - +func TestUnixSocketMode(t *testing.T) { for _, tc := range []struct { - name string - flags []string - mode os.FileMode + name string + fileMode string + flags []string + wantMode os.FileMode }{ - {name: "quoted mode", mode: 0o600}, - {name: "string override", flags: []string{"grpc.unix_socket.mode=0640"}, mode: 0o640}, + {name: "quoted 0600", fileMode: "0600", wantMode: 0o600}, + {name: "quoted 0640", fileMode: "0640", wantMode: 0o640}, + {name: "mode override", fileMode: "0600", flags: []string{"grpc.unix_socket.mode=0640"}, wantMode: 0o640}, } { t.Run(tc.name, func(t *testing.T) { - t.Chdir(t.TempDir()) - const options = `{mode: "0600", uid: "${RR_TEST_SOCKET_UID}", gid: "${RR_TEST_SOCKET_GID}"}` - flags := append([]string{"server.command=php " + worker, "grpc.proto=" + proto}, tc.flags...) - cfg := unixSocketConfig(t, "unix://grpc.sock", options, flags) - log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) - rrServer := &server.Plugin{} - require.NoError(t, rrServer.Init(cfg, log)) - t.Cleanup(func() { require.NoError(t, rrServer.Stop(context.Background())) }) - p := &grpcPlugin.Plugin{} - require.NoError(t, p.Init(cfg, log, rrServer)) - stop := sync.OnceValue(func() error { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - return p.Stop(ctx) - }) - t.Cleanup(func() { require.NoError(t, stop()) }) - errCh := p.Serve() - select { - case errS := <-errCh: - t.Fatalf("gRPC serve: %v", errS) - default: - } - - conn := helpers.Dial(t, "unix:grpc.sock") - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) - defer cancel() - response, errR := service.NewEchoClient(conn).Ping(ctx, &service.Message{Msg: tc.name}, grpc.WaitForReady(true)) - require.NoError(t, errR) - require.Equal(t, strings.ToUpper(tc.name), response.GetMsg()) - health, errH := grpchealth.NewHealthClient(conn).Check(ctx, &grpchealth.HealthCheckRequest{}) - require.NoError(t, errH) - require.Equal(t, grpchealth.HealthCheckResponse_SERVING, health.GetStatus()) - - info, errS := os.Stat("grpc.sock") - require.NoError(t, errS) - require.NotZero(t, info.Mode()&os.ModeSocket) - require.Equal(t, tc.mode, info.Mode().Perm()) - stat := info.Sys().(*syscall.Stat_t) - require.EqualValues(t, os.Getuid(), stat.Uid) - require.EqualValues(t, os.Getgid(), stat.Gid) - require.NoError(t, conn.Close()) - require.NoError(t, stop()) - _, errS = os.Stat("grpc.sock") - require.ErrorIs(t, errS, os.ErrNotExist) + socket := unixSocketPath(t) + cfgPath := unixSocketConfig(t, "unix://"+socket, fmt.Sprintf(`{mode: %q}`, tc.fileMode)) + helpers.Start(t, cfgPath, grpcPlugins(), helpers.WithConfigFlags(tc.flags...)) + + info, err := os.Stat(socket) + require.NoError(t, err) + require.Equal(t, tc.wantMode, info.Mode().Perm()) }) } } -func TestUnixSocketOwnershipError(t *testing.T) { +func TestUnixSocketPing(t *testing.T) { + socket := unixSocketPath(t) + cfgPath := unixSocketConfig(t, "unix://"+socket, `{mode: "0600"}`) + helpers.Start(t, cfgPath, grpcPlugins()) + + got, err := ping(t, helpers.Dial(t, "unix://"+socket), "TOST") + + require.NoError(t, err) + require.Equal(t, "TOST", got) +} + +func TestUnixSocketStopRemovesListener(t *testing.T) { + socket := unixSocketPath(t) + cfgPath := unixSocketConfig(t, "unix://"+socket, `{mode: "0600"}`) + _, stop := helpers.Start(t, cfgPath, grpcPlugins()) + require.FileExists(t, socket) + + stop() + + require.NoFileExists(t, socket) +} + +func TestUnixSocketOwnershipErrorRemovesListener(t *testing.T) { if os.Geteuid() == 0 { t.Skip("Requires an unprivileged process.") } - groups, err := os.Getgroups() - require.NoError(t, err) - otherGID := 0 - for otherGID == os.Getegid() || slices.Contains(groups, otherGID) { - otherGID++ - } - for _, tc := range []struct { - field string - id int - }{ - {field: "uid", id: 0}, - {field: "gid", id: otherGID}, - } { - t.Run(tc.field, func(t *testing.T) { - t.Chdir(t.TempDir()) - t.Setenv("RR_TEST_SOCKET_ID", strconv.Itoa(tc.id)) - options := fmt.Sprintf(`{%s: "${RR_TEST_SOCKET_ID}"}`, tc.field) - cfg := unixSocketConfig(t, "unix://ownership.sock", options, nil) - log := mocklogger.NewLogger(slog.New(slog.DiscardHandler)) - rrServer := &server.Plugin{} - require.NoError(t, rrServer.Init(cfg, log)) - t.Cleanup(func() { require.NoError(t, rrServer.Stop(context.Background())) }) - p := &grpcPlugin.Plugin{} - require.NoError(t, p.Init(cfg, log, rrServer)) - t.Cleanup(func() { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - require.NoError(t, p.Stop(ctx)) - }) - select { - case errS := <-p.Serve(): - require.ErrorContains(t, errS, "chown unix socket") - case <-time.After(5 * time.Second): - t.Fatal("gRPC did not report the ownership error") - } - _, errS := os.Stat("ownership.sock") - require.ErrorIs(t, errS, os.ErrNotExist) - }) - } + socket := unixSocketPath(t) + cfgPath := unixSocketConfig(t, "unix://"+socket, "{uid: 0}") + err := helpers.StartExpectServeError(t, cfgPath, grpcPlugins()) + require.ErrorContains(t, err, "chown unix socket") + + _, err = os.Stat(socket) + require.ErrorIs(t, err, os.ErrNotExist) } -func unixSocketConfig(t *testing.T, listen, options string, flags []string) *config.Plugin { +func unixSocketPath(t *testing.T) string { + t.Helper() + + // Keep the socket path below the macOS length limit. + dir, err := os.MkdirTemp("", "rr-grpc-") + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, os.RemoveAll(dir)) }) + return filepath.Join(dir, "grpc.sock") +} + +func unixSocketConfig(t *testing.T, listen, options string) string { t.Helper() contents := fmt.Sprintf(`version: "3" server: - command: [unused] + command: "php php_test_files/worker-grpc.php" grpc: listen: %q + proto: + - "proto/service/service.proto" + unix_socket: %s pool: debug: true destroy_timeout: 5s -`, listen) - if options != "" { - contents += " unix_socket: " + options + "\n" - } +`, listen, options) path := filepath.Join(t.TempDir(), ".rr.yaml") require.NoError(t, os.WriteFile(path, []byte(contents), 0o600)) - cfg := &config.Plugin{Path: path, Flags: flags} - require.NoError(t, cfg.Init()) - return cfg + return path }