diff --git a/apps/nsqd/options.go b/apps/nsqd/options.go index f93597013..9e311d64e 100644 --- a/apps/nsqd/options.go +++ b/apps/nsqd/options.go @@ -131,6 +131,8 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet { flagSet.String("https-address", opts.HTTPSAddress, ": to listen on for HTTPS clients") flagSet.String("http-address", opts.HTTPAddress, "address to listen on for HTTP clients (: for TCP/IP or for unix socket)") flagSet.String("tcp-address", opts.TCPAddress, "address to listen on for TCP clients (: for TCP/IP or for unix socket)") + flagSet.String("unix-socket-path", opts.UnixSocketPath, " to listen on for HTTP clients on a unix socket (in addition to --http-address)") + flagSet.String("unix-socket-permission", opts.UnixSocketPermission, "file permission of the unix socket as an octal number (e.g. 0660); empty means do not chmod") authHTTPAddresses := app.StringArray{} flagSet.Var(&authHTTPAddresses, "auth-http-address", ": or a full url to query auth server (may be given multiple times)") diff --git a/contrib/nsqd.cfg.example b/contrib/nsqd.cfg.example index 15033268e..65c420a87 100644 --- a/contrib/nsqd.cfg.example +++ b/contrib/nsqd.cfg.example @@ -13,6 +13,14 @@ http_address = "0.0.0.0:4151" ## : to listen on for HTTPS clients # https_address = "0.0.0.0:4152" +## to listen on for HTTP clients on a unix socket +## (in addition to http_address, served without TLS) +# unix_socket_path = "/var/run/nsqd.sock" + +## file permission of the unix socket as an octal number (e.g. 0660); +## empty/unset means do not chmod (umask applies) +# unix_socket_permission = "0660" + ## address that will be registered with lookupd (defaults to the OS hostname) # broadcast_address = "" diff --git a/nsqd/nsqd.go b/nsqd/nsqd.go index 523d1665b..9ea1377e4 100644 --- a/nsqd/nsqd.go +++ b/nsqd/nsqd.go @@ -12,9 +12,11 @@ import ( "net" "os" "path" + "strconv" "strings" "sync" "sync/atomic" + "syscall" "time" "github.com/nsqio/nsq/internal/clusterinfo" @@ -61,6 +63,7 @@ type NSQD struct { tcpListener net.Listener httpListener net.Listener httpsListener net.Listener + udsListener net.Listener tlsConfig *tls.Config clientTLSConfig *tls.Config @@ -165,6 +168,48 @@ func New(opts *Options) (*NSQD, error) { return nil, fmt.Errorf("listen (%s) failed - %s", opts.HTTPSAddress, err) } } + if opts.UnixSocketPath != "" { + var mode os.FileMode + if opts.UnixSocketPermission != "" { + parsed, perr := strconv.ParseUint(opts.UnixSocketPermission, 8, 32) + if perr != nil { + return nil, fmt.Errorf("invalid --unix-socket-permission %q (expected octal, e.g. 0660) - %s", opts.UnixSocketPermission, perr) + } + mode = os.FileMode(parsed) + } + // Remove a leftover socket file from a previous crashed run so net.Listen + // does not fail with "address already in use". Only remove the path if it + // is actually a unix socket — never clobber a regular file or directory. + if info, statErr := os.Stat(opts.UnixSocketPath); statErr == nil { + if info.Mode()&os.ModeSocket == 0 { + return nil, fmt.Errorf("refusing to overwrite existing non-socket file at %s", opts.UnixSocketPath) + } + if rmErr := os.Remove(opts.UnixSocketPath); rmErr != nil { + return nil, fmt.Errorf("failed to remove stale unix socket %s - %s", opts.UnixSocketPath, rmErr) + } + } + // Narrow the umask while we bind so the socket inode is created with the + // intended permissions from the start, closing the race window between + // net.Listen and the follow-up os.Chmod. + if mode != 0 { + oldMask := syscall.Umask(int(^mode & 0777)) + n.udsListener, err = net.Listen("unix", opts.UnixSocketPath) + syscall.Umask(oldMask) + } else { + n.udsListener, err = net.Listen("unix", opts.UnixSocketPath) + } + if err != nil { + return nil, fmt.Errorf("listen (%s) failed - %s", opts.UnixSocketPath, err) + } + if mode != 0 { + // Belt-and-braces: ensure the final mode matches even if the platform + // applied additional restrictions during Listen. + err = os.Chmod(opts.UnixSocketPath, mode) + if err != nil { + return nil, fmt.Errorf("failed to set unix socket permission - %s", err) + } + } + } if opts.BroadcastHTTPPort == 0 { tcpAddr, ok := n.RealHTTPAddr().(*net.TCPAddr) if ok { @@ -229,6 +274,13 @@ func (n *NSQD) RealHTTPSAddr() *net.TCPAddr { return n.httpsListener.Addr().(*net.TCPAddr) } +func (n *NSQD) RealUDSAddr() net.Addr { + if n.udsListener == nil { + return &net.UnixAddr{} + } + return n.udsListener.Addr() +} + func (n *NSQD) SetHealth(err error) { n.errValue.Store(errStore{err: err}) } @@ -281,6 +333,15 @@ func (n *NSQD) Main() error { exitFunc(http_api.Serve(n.httpsListener, httpsServer, "HTTPS", n.logf)) }) } + if n.udsListener != nil { + // The UDS HTTP listener is intentionally served as plain HTTP and + // ignores TLSRequired: access control is delegated to filesystem + // permissions on the socket path (see --unix-socket-permission). + udsHttpServer := newHTTPServer(n, false, false) + n.waitGroup.Wrap(func() { + exitFunc(http_api.Serve(n.udsListener, udsHttpServer, "HTTP", n.logf)) + }) + } n.waitGroup.Wrap(n.queueScanLoop) n.waitGroup.Wrap(n.lookupLoop) @@ -460,6 +521,10 @@ func (n *NSQD) Exit() { n.httpsListener.Close() } + if n.udsListener != nil { + n.udsListener.Close() + } + n.Lock() err := n.PersistMetadata() if err != nil { diff --git a/nsqd/options.go b/nsqd/options.go index 8997c7265..b41330b1a 100644 --- a/nsqd/options.go +++ b/nsqd/options.go @@ -22,6 +22,8 @@ type Options struct { TCPAddress string `flag:"tcp-address"` HTTPAddress string `flag:"http-address"` HTTPSAddress string `flag:"https-address"` + UnixSocketPath string `flag:"unix-socket-path"` + UnixSocketPermission string `flag:"unix-socket-permission"` BroadcastAddress string `flag:"broadcast-address"` BroadcastTCPPort int `flag:"broadcast-tcp-port"` BroadcastHTTPPort int `flag:"broadcast-http-port"` @@ -105,6 +107,7 @@ func NewOptions() *Options { TCPAddress: "0.0.0.0:4150", HTTPAddress: "0.0.0.0:4151", HTTPSAddress: "0.0.0.0:4152", + UnixSocketPath: "", BroadcastAddress: hostname, BroadcastTCPPort: 0, BroadcastHTTPPort: 0, diff --git a/nsqd/uds_http_test.go b/nsqd/uds_http_test.go new file mode 100644 index 000000000..d582a57da --- /dev/null +++ b/nsqd/uds_http_test.go @@ -0,0 +1,338 @@ +package nsqd + +import ( + "bytes" + "context" + "fmt" + "io" + "net" + "net/http" + "os" + "path/filepath" + "runtime" + "strconv" + "syscall" + "testing" + "time" + + "github.com/nsqio/nsq/internal/test" +) + +func skipIfNoUnixSockets(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("unix sockets not supported on windows") + } +} + +// startNSQDWithUDS configures an NSQD with both a TCP HTTP listener (on a random +// port) and an additional Unix-domain-socket HTTP listener. Returns the TCP HTTP +// address and the UDS path. The caller is responsible for nsqd.Exit() and +// removing opts.DataPath. +func startNSQDWithUDS(t *testing.T, opts *Options, perm string) (httpAddr net.Addr, udsPath string, nsqd *NSQD) { + t.Helper() + if opts.UnixSocketPath == "" { + opts.UnixSocketPath = filepath.Join(t.TempDir(), "nsqd.sock") + } + udsPath = opts.UnixSocketPath + opts.UnixSocketPermission = perm + _, httpAddr, nsqd = mustStartNSQD(opts) + // wait briefly for Main() to actually start serving on the UDS + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if _, err := os.Stat(udsPath); err == nil { + break + } + time.Sleep(10 * time.Millisecond) + } + return httpAddr, udsPath, nsqd +} + +func udsHTTPClient(sockPath string) *http.Client { + return &http.Client{ + Timeout: 5 * time.Second, + Transport: &http.Transport{ + DialContext: func(_ context.Context, _, _ string) (net.Conn, error) { + return net.Dial("unix", sockPath) + }, + }, + } +} + +// (1) UDS-HTTP und TCP-HTTP koexistieren und teilen sich denselben NSQD-State. +func TestUDSAndTCPHTTPCoexist(t *testing.T) { + skipIfNoUnixSockets(t) + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + httpAddr, udsPath, nsqd := startNSQDWithUDS(t, opts, "") + defer os.RemoveAll(opts.DataPath) + defer nsqd.Exit() + + topicName := "test_uds_tcp_coexist" + strconv.Itoa(int(time.Now().Unix())) + topic := nsqd.GetTopic(topicName) + + // publish via TCP HTTP + tcpURL := fmt.Sprintf("http://%s/pub?topic=%s", httpAddr, topicName) + resp, err := http.Post(tcpURL, "application/octet-stream", bytes.NewBufferString("from-tcp")) + test.Nil(t, err) + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + test.Equal(t, 200, resp.StatusCode) + test.Equal(t, "OK", string(body)) + + // publish via UDS HTTP + udsClient := udsHTTPClient(udsPath) + udsURL := fmt.Sprintf("http://unix/pub?topic=%s", topicName) + resp, err = udsClient.Post(udsURL, "application/octet-stream", bytes.NewBufferString("from-uds")) + test.Nil(t, err) + body, _ = io.ReadAll(resp.Body) + resp.Body.Close() + test.Equal(t, 200, resp.StatusCode) + test.Equal(t, "OK", string(body)) + + time.Sleep(25 * time.Millisecond) + test.Equal(t, int64(2), topic.Depth()) +} + +// (2) Standard-HTTP-Endpunkte (/ping, /info, /stats) sind über UDS erreichbar. +func TestUDSServesHTTPEndpoints(t *testing.T) { + skipIfNoUnixSockets(t) + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + _, udsPath, nsqd := startNSQDWithUDS(t, opts, "") + defer os.RemoveAll(opts.DataPath) + defer nsqd.Exit() + + client := udsHTTPClient(udsPath) + + for _, endpoint := range []string{"/ping", "/info", "/stats"} { + resp, err := client.Get("http://unix" + endpoint) + test.Nil(t, err) + _, _ = io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != 200 { + t.Fatalf("%s returned %d", endpoint, resp.StatusCode) + } + } +} + +// (3) Explizit gesetzte Permission wird angewandt. +func TestUDSPermissionApplied(t *testing.T) { + skipIfNoUnixSockets(t) + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + _, udsPath, nsqd := startNSQDWithUDS(t, opts, "0600") + defer os.RemoveAll(opts.DataPath) + defer nsqd.Exit() + + info, err := os.Stat(udsPath) + test.Nil(t, err) + test.Equal(t, os.FileMode(0600), info.Mode().Perm()) +} + +// (4) Default-Permission (leer): kein Chmod, Modus ergibt sich aus umask. +func TestUDSPermissionDefault(t *testing.T) { + skipIfNoUnixSockets(t) + + oldMask := syscall.Umask(0022) + defer syscall.Umask(oldMask) + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + _, udsPath, nsqd := startNSQDWithUDS(t, opts, "") + defer os.RemoveAll(opts.DataPath) + defer nsqd.Exit() + + info, err := os.Stat(udsPath) + test.Nil(t, err) + // Mit umask 0022 darf das gid/other-write-Bit nicht gesetzt sein. + if info.Mode().Perm()&0022 != 0 { + t.Fatalf("expected umask 0022 to clear group/other write bits, got %o", info.Mode().Perm()) + } +} + +// (5) Ungültiger Pfad → New(opts) liefert Fehler statt Panic. +func TestUDSListenFailureWhenPathInvalid(t *testing.T) { + skipIfNoUnixSockets(t) + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + opts.TCPAddress = "127.0.0.1:0" + opts.HTTPAddress = "127.0.0.1:0" + opts.HTTPSAddress = "127.0.0.1:0" + tmp, _ := os.MkdirTemp("", "nsq-test-") + defer os.RemoveAll(tmp) + opts.DataPath = tmp + opts.UnixSocketPath = "/this/dir/does/not/exist/nsqd.sock" + + n, err := New(opts) + if err == nil { + n.Exit() + t.Fatal("expected error for unreachable UDS path") + } +} + +// (6) Ungültige Permission (kein Octal) → New(opts) liefert Fehler. +func TestUDSInvalidPermissionRejected(t *testing.T) { + skipIfNoUnixSockets(t) + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + opts.TCPAddress = "127.0.0.1:0" + opts.HTTPAddress = "127.0.0.1:0" + opts.HTTPSAddress = "127.0.0.1:0" + tmp, _ := os.MkdirTemp("", "nsq-test-") + defer os.RemoveAll(tmp) + opts.DataPath = tmp + opts.UnixSocketPath = filepath.Join(t.TempDir(), "nsqd.sock") + opts.UnixSocketPermission = "garbage" + + n, err := New(opts) + if err == nil { + n.Exit() + t.Fatal("expected error for invalid permission string") + } +} + +// (7) Socket-Datei wird beim sauberen Exit entfernt. +func TestUDSCleanupOnExit(t *testing.T) { + skipIfNoUnixSockets(t) + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + _, udsPath, nsqd := startNSQDWithUDS(t, opts, "") + defer os.RemoveAll(opts.DataPath) + + _, err := os.Stat(udsPath) + test.Nil(t, err) + + nsqd.Exit() + + if _, err := os.Stat(udsPath); !os.IsNotExist(err) { + t.Fatalf("expected socket file to be removed after Exit, stat err=%v", err) + } +} + +// (8a) Existierende Datei am Socket-Pfad, die KEIN Socket ist, wird abgelehnt +// (Schutz davor, versehentlich eine reguläre Datei zu löschen). +func TestUDSExistingRegularFileRejected(t *testing.T) { + skipIfNoUnixSockets(t) + + regular := filepath.Join(t.TempDir(), "not-a-socket") + f, err := os.Create(regular) + test.Nil(t, err) + f.Close() + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + opts.TCPAddress = "127.0.0.1:0" + opts.HTTPAddress = "127.0.0.1:0" + opts.HTTPSAddress = "127.0.0.1:0" + tmp, _ := os.MkdirTemp("", "nsq-test-") + defer os.RemoveAll(tmp) + opts.DataPath = tmp + opts.UnixSocketPath = regular + + n, err := New(opts) + if err == nil { + n.Exit() + t.Fatal("expected error when path exists and is not a socket") + } + // Die reguläre Datei darf nicht gelöscht worden sein. + if _, statErr := os.Stat(regular); statErr != nil { + t.Fatalf("regular file must not be removed, stat err=%v", statErr) + } +} + +// (8b) Verwaister Socket aus einem vorigen Crash wird automatisch entfernt +// und Listen gelingt. +func TestUDSStaleSocketAutoRemoved(t *testing.T) { + skipIfNoUnixSockets(t) + + udsPath := filepath.Join(t.TempDir(), "stale.sock") + + // Simuliere verwaisten Socket: kurz binden, Listener leaken (kein Close), + // dann den File-Descriptor manuell schließen, damit nur das Inode bleibt. + stale, err := net.Listen("unix", udsPath) + test.Nil(t, err) + if ul, ok := stale.(*net.UnixListener); ok { + ul.SetUnlinkOnClose(false) + } + stale.Close() + info, err := os.Stat(udsPath) + test.Nil(t, err) + if info.Mode()&os.ModeSocket == 0 { + t.Fatalf("test setup: expected socket inode at %s", udsPath) + } + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + opts.UnixSocketPath = udsPath + _, gotPath, nsqd := startNSQDWithUDS(t, opts, "") + defer os.RemoveAll(opts.DataPath) + defer nsqd.Exit() + + test.Equal(t, udsPath, gotPath) + if _, err := os.Stat(udsPath); err != nil { + t.Fatalf("expected fresh socket at %s after auto-remove, err=%v", udsPath, err) + } +} + +// (9) TLSRequired beeinflusst den UDS-HTTP-Server nicht (lokaler Trust). +func TestUDSIgnoresTLSRequired(t *testing.T) { + skipIfNoUnixSockets(t) + + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + opts.TLSCert = "./test/certs/server.pem" + opts.TLSKey = "./test/certs/server.key" + opts.TLSRequired = TLSRequired + // vermeidet, dass der HTTPS-Listener auf dem TLS-Default-Port versucht zu binden + opts.HTTPSAddress = "127.0.0.1:0" + + _, udsPath, nsqd := startNSQDWithUDS(t, opts, "") + defer os.RemoveAll(opts.DataPath) + defer nsqd.Exit() + + topicName := "test_uds_tls_required" + strconv.Itoa(int(time.Now().Unix())) + nsqd.GetTopic(topicName) + + client := udsHTTPClient(udsPath) + resp, err := client.Post("http://unix/pub?topic="+topicName, "application/octet-stream", bytes.NewBufferString("payload")) + test.Nil(t, err) + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + test.Equal(t, 200, resp.StatusCode) + test.Equal(t, "OK", string(body)) +} + +// (10) Defaults: Feature ist opt-in. +func TestUDSOptionDefaults(t *testing.T) { + opts := NewOptions() + test.Equal(t, "", opts.UnixSocketPath) + test.Equal(t, "", opts.UnixSocketPermission) +} + +// (11) RealUDSAddr liefert die tatsächlich gebundene Adresse bzw. einen Zero-Value. +func TestRealUDSAddr(t *testing.T) { + skipIfNoUnixSockets(t) + + // ohne UDS: Zero-Value, kein Panic + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + _, _, nsqd := mustStartNSQD(opts) + defer os.RemoveAll(opts.DataPath) + addr := nsqd.RealUDSAddr() + test.Equal(t, "", addr.String()) + nsqd.Exit() + + // mit UDS: Adresse == konfigurierter Pfad + opts2 := NewOptions() + opts2.Logger = test.NewTestLogger(t) + _, udsPath, nsqd2 := startNSQDWithUDS(t, opts2, "") + defer os.RemoveAll(opts2.DataPath) + defer nsqd2.Exit() + test.Equal(t, udsPath, nsqd2.RealUDSAddr().String()) +}