Add a local control socket and nebula ctl (#1975)

Add support for accessing the previously ssh-only debug commands via
`nebula ctl <command>.`

- unix socket (named pipe on windows) for communication
- sockets scoped to the user + an optional group
- in cmd/nebula and cmd/nebula-service, but not nebula.Main

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Matt RichardsonandClaude Fable 5.1 authored and GitHub committed 2026-10-21 11:16:00 -04:00
1 parent 9a5bdb508e
commit 0f767ce529
54 files changed
+7581 -925

No files matched your search

+79
View File
@@ -0,0 +1,79 @@
package localapi
import (
"errors"
"fmt"
"os"
)
// Access is who may reach a socket beyond the account nebula runs as: the node's mode,
// and the group that shares the socket when the config names one.
//
// It is the dockerd model (-G): the operator names a group, nebula hands the socket to
// it, and who is in the group is managed outside nebula. On unix the mode is the
// socket node's and the group becomes its owning group; on windows the group is an
// entry in the pipe's security descriptor and the mode means nothing to a pipe.
type Access struct {
Mode os.FileMode
Group Group
}
// Group is a group named in the config, resolved to what the platform checks. The zero
// Group is none.
type Group struct {
// Name is what the config said.
Name string
// ID is the platform's identifier for it: a gid on unix, a SID on windows.
ID string
}
const (
// ownerOnlyMode is a socket's mode when no group shares it: the account nebula runs
// as, and nobody else.
ownerOnlyMode os.FileMode = 0600
// groupSocketMode is the mode a socket takes when a group shares it and the config
// names no other: the account nebula runs as and the group, and nobody else.
groupSocketMode os.FileMode = 0660
)
// errReopenForAccess is returned by applyAccess on a platform whose access control is
// fixed when the endpoint is created, so a change means creating it again.
var errReopenForAccess = errors.New("access is fixed at listen time")
// ownerOnly is the control socket's access when no group is configured. The mode
// itself is not configurable there; what an operator can choose is a group,
// CtlGroupKey, which makes it groupSocketMode.
func ownerOnly() Access { return Access{Mode: ownerOnlyMode} }
// shared reports whether a group shares the socket.
func (a Access) shared() bool { return a.Group.ID != "" }
// String renders the access for a log line.
func (a Access) String() string {
if !a.shared() {
return modeString(a.Mode)
}
return fmt.Sprintf("%s group %s (%s)", modeString(a.Mode), a.Group.Name, a.Group.ID)
}
// groupFromConfig reads the group a key names, the zero Group when it names none. The
// name is resolved here, when the config is read, so a group that does not exist is a
// config error reported with the key rather than a bind that fails later.
func groupFromConfig(c interface{ GetString(string, string) string }, key string) (Group, error) {
name := c.GetString(key, "")
if name == "" {
return Group{}, nil
}
id, err := lookupGroupID(name)
if err != nil {
return Group{}, fmt.Errorf("%s: %w", key, err)
}
return Group{Name: name, ID: id}, nil
}
func modeString(m os.FileMode) string {
return fmt.Sprintf("%04o", m)
}
+176
View File
@@ -0,0 +1,176 @@
//go:build !windows
package localapi
import (
"context"
"os"
"os/user"
"strconv"
"syscall"
"testing"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/sys/unix"
)
// ownGroup returns a group this process may hand a socket to: its own effective group,
// by name and by gid, which is the one chown accepts from an unprivileged user.
func ownGroup(t *testing.T) (name, gid string) {
t.Helper()
gid = strconv.Itoa(os.Getegid())
g, err := user.LookupGroupId(gid)
require.NoError(t, err)
return g.Name, gid
}
func socketGroup(t *testing.T, path string) uint32 {
t.Helper()
fi, err := os.Lstat(path)
require.NoError(t, err)
st, ok := fi.Sys().(*syscall.Stat_t)
require.True(t, ok)
return uint32(st.Gid)
}
// TestLookupGroupID: a group is named as the host names it or by its gid; a gid the
// host does not list is accepted as chown would accept it, and a name it does not list
// is a mistake to report.
func TestLookupGroupID(t *testing.T) {
name, gid := ownGroup(t)
got, err := lookupGroupID(name)
require.NoError(t, err)
assert.Equal(t, gid, got)
got, err = lookupGroupID(gid)
require.NoError(t, err)
assert.Equal(t, gid, got)
got, err = lookupGroupID("4000000000")
require.NoError(t, err)
assert.Equal(t, "4000000000", got)
_, err = lookupGroupID("no-such-group-nebula-test")
require.ErrorContains(t, err, "does not exist")
}
// TestControlAccessFromConfig: no group is the owner-only socket; a group is 0660 with
// that group; a group the host does not know is a config error naming the key.
func TestControlAccessFromConfig(t *testing.T) {
name, gid := ownGroup(t)
load := func(body string) (Access, error) {
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(body))
return controlAccessFrom(c)
}
acc, err := load("local_api: {}\n")
require.NoError(t, err)
assert.Equal(t, ownerOnly(), acc)
acc, err = load("local_api:\n control:\n group: " + name + "\n")
require.NoError(t, err)
assert.Equal(t, os.FileMode(0660), acc.Mode)
assert.Equal(t, name, acc.Group.Name)
assert.Equal(t, gid, acc.Group.ID)
_, err = load("local_api:\n control:\n group: no-such-group-nebula-test\n")
require.ErrorContains(t, err, CtlGroupKey)
assert.ErrorContains(t, err, "does not exist")
}
// TestControlSocketIsHandedToTheGroup: the socket is created with the group's gid and
// 0660, whatever the umask, and a reload that takes the group away tightens the socket
// in place, without moving it.
func TestControlSocketIsHandedToTheGroup(t *testing.T) {
old := unix.Umask(0077)
defer unix.Umask(old)
name, gid := ownGroup(t)
want, err := strconv.ParseUint(gid, 10, 32)
require.NoError(t, err)
s := NewServer(test.NewLogger(), basicRegistry())
path := testSocket(t)
require.NoError(t, s.Rebind(context.Background(), path, Access{Mode: 0660, Group: Group{Name: name, ID: gid}}))
defer s.Stop()
fi, err := os.Lstat(path)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0660), fi.Mode().Perm())
assert.Equal(t, uint32(want), socketGroup(t, path))
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
fi, err = os.Lstat(path)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0600), fi.Mode().Perm())
assert.Equal(t, path, s.Path())
_, err = (&Client{Path: path}).Describe()
require.NoError(t, err, "the same listener is still serving")
}
// TestControlSocketWithAGroupTrustsTheKernel: with a group sharing the socket, the
// kernel's check on connect is the access control and the daemon's peer check stands
// aside, so a connection the peer check alone would refuse is served. Without a group
// the same connection is refused; see TestPeerCheckRefusesAnotherUser.
func TestControlSocketWithAGroupTrustsTheKernel(t *testing.T) {
if !peerCheckSupported {
t.Skip("no peer credentials on this platform")
}
name, gid := ownGroup(t)
me := currentUID()
asUID(t, me+1)
s := NewServer(test.NewLogger(), basicRegistry())
asUID(t, me)
path := testSocket(t)
require.NoError(t, s.Rebind(context.Background(), path, Access{Mode: 0660, Group: Group{Name: name, ID: gid}}))
defer s.Stop()
_, err := (&Client{Path: path}).Describe()
require.NoError(t, err, "the socket's group admitted us; the peer check must not refuse")
// Taking the group away on a reload puts the peer check back.
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
_, err = (&Client{Path: path}).Describe()
require.ErrorContains(t, err, "not permitted")
}
// TestRemovingTheGroupHandsTheSocketBack: a reload that drops the group key leaves the
// mode the operator set but returns the socket to the daemon's own group, so the old
// group's members are no longer admitted. The socket is first handed to a group this
// process belongs to other than its own, which not every host offers.
func TestRemovingTheGroupHandsTheSocketBack(t *testing.T) {
own := os.Getegid()
groups, err := os.Getgroups()
require.NoError(t, err)
other := -1
for _, g := range groups {
if g != own {
other = g
break
}
}
if other < 0 {
t.Skip("this process belongs to no group but its own")
}
gid := strconv.Itoa(other)
s := NewServer(test.NewLogger(), basicRegistry())
path := testSocket(t)
require.NoError(t, s.Rebind(context.Background(), path, Access{Mode: 0660, Group: Group{Name: gid, ID: gid}}))
defer s.Stop()
assert.Equal(t, uint32(other), socketGroup(t, path))
require.NoError(t, s.Rebind(context.Background(), path, Access{Mode: 0660}))
assert.Equal(t, uint32(own), socketGroup(t, path), "the daemon's own group again")
fi, err := os.Lstat(path)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0660), fi.Mode().Perm(), "the mode is the operator's")
}
+44
View File
@@ -0,0 +1,44 @@
//go:build windows
package localapi
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/sys/windows"
)
// TestLookupGroupIDOnWindows: an account name resolves to its SID, a SID written out
// is accepted as it is, and an account the host does not know is reported.
func TestLookupGroupIDOnWindows(t *testing.T) {
users, err := windows.CreateWellKnownSid(windows.WinBuiltinUsersSid)
require.NoError(t, err)
got, err := lookupGroupID("Users")
require.NoError(t, err)
assert.Equal(t, users.String(), got)
got, err = lookupGroupID(users.String())
require.NoError(t, err)
assert.Equal(t, users.String(), got)
_, err = lookupGroupID("no-such-account-nebula-test")
require.ErrorContains(t, err, "does not exist")
}
// TestSDDLAdmitsTheGroup: the descriptor is the owner, SYSTEM and Administrators with
// full access, plus the group with what a client needs and nothing generic, so a
// member cannot create an instance of the pipe; see pipeClientAccess.
func TestSDDLAdmitsTheGroup(t *testing.T) {
sddl, err := sddlFor(ownerOnly())
require.NoError(t, err)
assert.NotContains(t, sddl, "S-1-5-32-545")
sddl, err = sddlFor(Access{Mode: groupSocketMode, Group: Group{Name: "Users", ID: "S-1-5-32-545"}})
require.NoError(t, err)
assert.Contains(t, sddl, "(A;;GA;;;SY)(A;;GA;;;BA)(A;;0x120083;;;S-1-5-32-545)")
assert.Zero(t, pipeClientAccess&windows.FILE_APPEND_DATA, "FILE_APPEND_DATA is FILE_CREATE_PIPE_INSTANCE on a pipe")
assert.Zero(t, pipeClientAccess&(windows.GENERIC_READ|windows.GENERIC_WRITE|windows.GENERIC_ALL))
}
+163
View File
@@ -0,0 +1,163 @@
package localapi
import (
"encoding/json"
"errors"
"fmt"
"io"
"net"
"time"
"github.com/slackhq/nebula/commands"
)
// DialTimeout bounds the connect to a socket that exists but is not being served.
const DialTimeout = 5 * time.Second
// ResponseTimeout bounds the exchange once connected: sending the request and waiting
// for the daemon's answer. The daemon drops a client that is slow to send in 10s and one
// that is slow to read in 30s, so this is the other side of the same bargain, with room
// for the slowest command (a hostmap dump on a large network) before a stalled daemon is
// given up on. Client.Timeout overrides it.
const ResponseTimeout = 60 * time.Second
// MaxResponseBytes bounds what a client will read back. A response carries one command's
// output, and a hostmap dump on a large network is the realistic worst case.
const MaxResponseBytes = 64 << 20
// Client talks to a nebula listening at Path.
type Client struct {
Path string
// Timeout bounds the request and response once connected. Zero means
// ResponseTimeout.
Timeout time.Duration
}
// ReplyError is an answer the client could not read: the daemon was reached and wrote
// something back, but not a response this client understands. The daemon is not
// nebula, or not a version this client speaks to.
type ReplyError struct {
Err error
}
func (e *ReplyError) Error() string { return "the daemon's reply could not be read: " + e.Err.Error() }
func (e *ReplyError) Unwrap() error { return e.Err }
// DaemonError is a refusal the daemon answered with: the connection worked, the request
// was read, and the daemon said no. It is told apart from a failure to reach the daemon
// so that a client does not send its user after a socket path, or a sudo, when the daemon
// was right there and answered. A protocol version the daemon does not speak, a socket at
// its connection limit and a peer the daemon will not serve all arrive as one of these.
type DaemonError struct {
Msg string
}
func (e *DaemonError) Error() string { return e.Msg }
// Describe asks the daemon which commands it serves and what parameters they take, so a
// client can offer them without having been built against this daemon.
func (c *Client) Describe() ([]commands.CommandSchema, error) {
res, err := c.do(&Request{Version: ProtocolVersion, Describe: true})
if err != nil {
return nil, err
}
if res.Error != "" {
return nil, &DaemonError{Msg: res.Error}
}
schema := []commands.CommandSchema{}
if err := json.Unmarshal(res.Result, &schema); err != nil {
return nil, &ReplyError{Err: fmt.Errorf("could not parse the command schema: %w", err)}
}
return schema, nil
}
// Invoke runs one command and returns what the daemon said. params are the command's
// options by name, as the strings a command line parsed them from; a program speaking
// the protocol directly may send JSON's own types instead, see Request.Params. An empty
// format asks for JSON, which is the API's default; "text" asks the daemon to apply the
// command's own human rendering.
func (c *Client) Invoke(command string, params map[string]string, args []string, format string) (*Response, error) {
var wire map[string]any
if len(params) > 0 {
wire = make(map[string]any, len(params))
for k, v := range params {
wire[k] = v
}
}
return c.do(&Request{
Version: ProtocolVersion,
Command: command,
Params: wire,
Args: args,
Format: format,
})
}
// do sends one request and reads the one response.
//
// Nothing is checked about the socket before the connect: its mode is the kernel's to
// enforce, and a socket that is not ours to connect to is refused there with a
// permission error, which the caller turns into advice.
func (c *Client) do(req *Request) (*Response, error) {
conn, err := dial(c.Path, DialTimeout)
if err != nil {
return nil, err
}
defer conn.Close()
// Without this a daemon that accepts and then stalls, blocked inside a command or
// stopped with SIGSTOP, hangs the client for good, and a script wrapping it never
// completes. One deadline covers the write and the read: the answer is due within
// the budget, however the time is spent.
timeout := c.Timeout
if timeout <= 0 {
timeout = ResponseTimeout
}
_ = conn.SetDeadline(time.Now().Add(timeout))
// If we get a write error, save that fact but then look for a response from the server.
// Some error conditions manifest as the server accepting our connection, writing an
// error, and then closing the connection. Our write races against that close.
writeErr := error(nil)
if err := writeFrame(conn, req); err != nil {
writeErr = fmt.Errorf("could not send request: %w", err)
}
res, err := readResponse(conn)
if err != nil {
// Something answered and it was not a response: the daemon was reached, and
// that is the message even when it hung up before the request was written,
// which a daemon that is not nebula is likely to do.
var fe *frameError
if errors.As(err, &fe) {
return nil, &ReplyError{Err: err}
}
if writeErr != nil {
return nil, writeErr
}
var ne net.Error
if errors.As(err, &ne) && ne.Timeout() {
return nil, fmt.Errorf("the daemon did not answer within %s", timeout)
}
return nil, err
}
return res, nil
}
func readResponse(conn net.Conn) (*Response, error) {
res := &Response{}
if err := readFrame(conn, MaxResponseBytes, "response", res, false); err != nil {
if errors.Is(err, io.EOF) {
return nil, errors.New("the daemon closed the connection without answering")
}
return nil, err
}
return res, nil
}
+39
View File
@@ -0,0 +1,39 @@
//go:build !windows
package localapi
import (
"fmt"
"os"
"os/user"
"strconv"
)
// lookupGroupID resolves a group's name, or a numeric gid, to the gid as a string. A
// gid that the group database does not list is accepted as chown would accept it: the
// operator may be naming a group the host will learn about later, or one that only a
// directory service knows.
func lookupGroupID(name string) (string, error) {
if g, err := user.LookupGroup(name); err == nil {
return g.Gid, nil
}
if _, err := strconv.ParseUint(name, 10, 32); err == nil {
return name, nil
}
return "", fmt.Errorf("group %q does not exist on this host", name)
}
// gidFor returns the gid the socket is owned by under acc: the configured group's, or
// the daemon's own effective group when none is configured.
func gidFor(acc Access) (int, error) {
if !acc.shared() {
return os.Getegid(), nil
}
gid, err := strconv.ParseUint(acc.Group.ID, 10, 32)
if err != nil {
return 0, fmt.Errorf("group %q resolved to %q, which is not a gid", acc.Group.Name, acc.Group.ID)
}
return int(gid), nil
}
+30
View File
@@ -0,0 +1,30 @@
//go:build windows
package localapi
import (
"fmt"
"strings"
"golang.org/x/sys/windows"
)
// lookupGroupID resolves an account name, local or domain qualified, or a SID written
// out, to the SID as a string, which is how the pipe's security descriptor names it.
// Any account the host can resolve is accepted, as docker accepts any SID for its -G: a
// group is the natural thing to name, but the descriptor does not care.
func lookupGroupID(name string) (string, error) {
if strings.HasPrefix(strings.ToUpper(name), "S-1-") {
sid, err := windows.StringToSid(name)
if err != nil {
return "", fmt.Errorf("%q is not a SID: %w", name, err)
}
return sid.String(), nil
}
sid, _, _, err := windows.LookupSID("", name)
if err != nil {
return "", fmt.Errorf("account %q does not exist on this host: %w", name, err)
}
return sid.String(), nil
}
+58
View File
@@ -0,0 +1,58 @@
package localapi
// DirName is the directory under /var/run that holds nebula's sockets by default.
const DirName = "nebula"
// CtlEnabledKey is the optional config key that turns the control socket off. It is on
// unless the config says false, and reloadable: a reload that turns it off removes the
// socket, and one that turns it back on brings it up again.
const CtlEnabledKey = "local_api.control.enabled"
// ControlEnabled reports whether a config leaves the control socket on. nebula ctl reads
// it too, so a client pointed at a config that turned the socket off says so rather than
// failing to dial.
func ControlEnabled(c interface{ GetBool(string, bool) bool }) bool {
return c.GetBool(CtlEnabledKey, true)
}
// CtlSocketKey is the optional config key naming the control socket. Unset, the socket
// is DefaultSocketPath, one fixed path per platform that both the daemon and nebula ctl
// know, which is what lets the endpoint work with no configuration on either side. It
// suits a daemon running as root, which is what the tun device needs anyway. A daemon run as another user, or a second daemon on
// one host, names its own. The CLI reads the same key, so when both sides see the same
// config file they agree on the socket with nothing more said.
const CtlSocketKey = "local_api.control.socket"
// CtlGroupKey is the optional config key naming a group that shares the control socket.
// Unset, the socket admits the account nebula runs as and nobody else. Set, the socket
// is handed to the group, 0660 on unix and an entry in the pipe's security descriptor on
// windows; see Access and Server.trustSocket.
const CtlGroupKey = "local_api.control.group"
// controlAccessFrom returns the control socket's access a config implies: the group
// CtlGroupKey names, if any, and the mode that goes with having one or not.
func controlAccessFrom(c interface{ GetString(string, string) string }) (Access, error) {
group, err := groupFromConfig(c, CtlGroupKey)
if err != nil {
return Access{}, err
}
if group.ID == "" {
return ownerOnly(), nil
}
return Access{Mode: groupSocketMode, Group: group}, nil
}
// SocketPathFromConfig returns the control socket path a config implies: CtlSocketKey
// when it is set, else DefaultSocketPath. A configured path must be absolute, so that
// the daemon and nebula ctl, started from wherever the operator happens to be, name the
// same socket from one config, and one this platform can bind; see checkSocketPath.
func SocketPathFromConfig(c interface{ GetString(string, string) string }) (string, error) {
override := c.GetString(CtlSocketKey, "")
if override == "" {
return DefaultSocketPath, nil
}
if err := checkSocketPath(CtlSocketKey, override); err != nil {
return "", err
}
return override, nil
}
+52
View File
@@ -0,0 +1,52 @@
//go:build !windows
package localapi
import (
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
type fakeConfig map[string]string
func (f fakeConfig) GetString(k, d string) string {
if v, ok := f[k]; ok {
return v
}
return d
}
// TestSocketPathFromConfigUsesTheDefault is what makes the endpoint work with no
// configuration: the daemon and the CLI both land on the one fixed path.
func TestSocketPathFromConfigUsesTheDefault(t *testing.T) {
p, err := SocketPathFromConfig(fakeConfig{})
require.NoError(t, err)
assert.Equal(t, "/var/run/nebula/ctl.sock", p)
assert.Equal(t, DefaultSocketPath, p)
}
func TestSocketPathFromConfigHonoursOverride(t *testing.T) {
p, err := SocketPathFromConfig(fakeConfig{CtlSocketKey: "/tmp/custom.sock"})
require.NoError(t, err)
assert.Equal(t, "/tmp/custom.sock", p, "the path is used as given")
}
// TestSocketPathFromConfigRefusesARelativeOverride: the daemon and the CLI are started
// from different directories, so a relative path would name two sockets from one config.
// It is refused rather than resolved against anything.
func TestSocketPathFromConfigRefusesARelativeOverride(t *testing.T) {
for _, rel := range []string{"run/ctl.sock", "./ctl.sock", "ctl.sock", "../ctl.sock"} {
_, err := SocketPathFromConfig(fakeConfig{CtlSocketKey: rel})
require.Error(t, err, rel)
require.ErrorContains(t, err, "must be an absolute path", rel)
assert.ErrorContains(t, err, CtlSocketKey, rel)
}
}
func TestSocketPathFromConfigRejectsOverlongOverride(t *testing.T) {
_, err := SocketPathFromConfig(fakeConfig{CtlSocketKey: "/" + strings.Repeat("x", sunPathMax)})
assert.ErrorContains(t, err, "too long for this platform")
}
+25
View File
@@ -0,0 +1,25 @@
//go:build darwin || freebsd
package localapi
import (
"net"
"golang.org/x/sys/unix"
)
// peerCheckSupported reports that this platform can identify a connecting process.
const peerCheckSupported = true
// peerUID returns the uid of the process on the other end of conn.
func peerUID(conn net.Conn) (uid uint32, err error) {
err = withPeerFD(conn, func(fd int) error {
xu, credErr := unix.GetsockoptXucred(fd, unix.SOL_LOCAL, unix.LOCAL_PEERCRED)
if credErr != nil {
return credErr
}
uid = xu.Uid
return nil
})
return uid, err
}
+25
View File
@@ -0,0 +1,25 @@
//go:build linux
package localapi
import (
"net"
"golang.org/x/sys/unix"
)
// peerCheckSupported reports that this platform can identify a connecting process.
const peerCheckSupported = true
// peerUID returns the uid of the process on the other end of conn.
func peerUID(conn net.Conn) (uid uint32, err error) {
err = withPeerFD(conn, func(fd int) error {
cred, credErr := unix.GetsockoptUcred(fd, unix.SOL_SOCKET, unix.SO_PEERCRED)
if credErr != nil {
return credErr
}
uid = cred.Uid
return nil
})
return uid, err
}
+62
View File
@@ -0,0 +1,62 @@
//go:build netbsd
package localapi
import (
"encoding/binary"
"fmt"
"net"
"unsafe"
"golang.org/x/sys/unix"
)
const peerCheckSupported = true
// NetBSD reports the peer's identity through LOCAL_PEEREID at the SOL_LOCAL level,
// filling a `struct unpcbid { pid_t unp_pid; uid_t unp_euid; gid_t unp_egid; }`.
// golang.org/x/sys/unix has no typed wrapper for it, so the getsockopt goes out by hand
// exactly the way that package's own generated wrappers do on this platform.
const (
solLocal = 0
localPeerEID = 0x0003
unpcbidSize = 12
unpcbidEUIDOffs = 4
)
// peerUID returns the uid of the process on the other end of conn.
//
// Anything unexpected here is an error rather than a default, so a wrong constant or an
// unsupported kernel refuses connections instead of silently admitting them.
func peerUID(conn net.Conn) (uid uint32, err error) {
err = withPeerFD(conn, func(fd int) error {
var (
buf [unpcbidSize]byte
size = uint32(unpcbidSize)
)
_, _, errno := unix.Syscall6(
unix.SYS_GETSOCKOPT,
uintptr(fd),
uintptr(solLocal),
uintptr(localPeerEID),
uintptr(unsafe.Pointer(&buf[0])),
uintptr(unsafe.Pointer(&size)),
0,
)
if errno != 0 {
return errno
}
if size != unpcbidSize {
return fmt.Errorf("LOCAL_PEEREID returned %d bytes, expected %d", size, unpcbidSize)
}
// Read, not cast: the buffer is bytes, and a cast would fault on an
// architecture that wants the uint32 aligned.
uid = binary.NativeEndian.Uint32(buf[unpcbidEUIDOffs:])
return nil
})
return uid, err
}
+34
View File
@@ -0,0 +1,34 @@
//go:build openbsd
package localapi
import (
"net"
"golang.org/x/sys/unix"
)
// peerCheckSupported reports that this platform can identify a connecting process.
const peerCheckSupported = true
// peerUID returns the uid of the process on the other end of conn.
//
// OpenBSD reports the peer through SO_PEERCRED at SOL_SOCKET, filling a
// `struct sockpeercred { uid_t uid; gid_t gid; pid_t pid; }`. System calls here have to
// go through libc, so the raw getsockopt netbsd uses is not an option, and
// golang.org/x/sys/unix has no typed wrapper for this struct. It does export
// GetsockoptInt, which goes through its libc trampoline with a 4 byte buffer, and the
// kernel copies out min(caller's length, option length): the first four bytes, which are
// the uid. That is a dependence on the field order, the same one netbsd's offset carries.
func peerUID(conn net.Conn) (uid uint32, err error) {
err = withPeerFD(conn, func(fd int) error {
v, credErr := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_PEERCRED)
if credErr != nil {
return credErr
}
// A uid_t is 32 bits; GetsockoptInt read exactly that many and widened it.
uid = uint32(v)
return nil
})
return uid, err
}
+20
View File
@@ -0,0 +1,20 @@
//go:build windows
package localapi
import (
"errors"
"net"
)
// peerCheckSupported is false where the connecting process is not identified separately.
//
// On windows there is nothing to add: the pipe's security descriptor already admits
// only the account nebula runs as, which is the check, not a second opinion on it.
const peerCheckSupported = false
var errNoPeerCheck = errors.New("peer credentials are not available on this platform")
func peerUID(conn net.Conn) (uint32, error) {
return 0, errNoPeerCheck
}
+302
View File
@@ -0,0 +1,302 @@
//go:build !windows
package localapi
import (
"errors"
"fmt"
"net"
"os"
"os/user"
"path/filepath"
"strconv"
"sync"
"syscall"
"time"
)
const network = "unix"
// withPeerFD runs f on the descriptor behind conn, which must be a unix socket, and
// returns whichever error came first: not a unix socket, no raw access to it, or f's
// own. It is the scaffolding every platform's peerUID shares; only the getsockopt inside
// differs.
func withPeerFD(conn net.Conn, f func(fd int) error) error {
uc, ok := conn.(*net.UnixConn)
if !ok {
return fmt.Errorf("not a unix connection")
}
raw, err := uc.SyscallConn()
if err != nil {
return err
}
var inner error
if err := raw.Control(func(fd uintptr) { inner = f(int(fd)) }); err != nil {
return err
}
return inner
}
// currentUID is the user the owner checks are made for. A variable so a test can stand
// in for root, or for a stranger, without owning a second account on the host.
var currentUID = func() uint32 { return uint32(os.Geteuid()) }
// listen binds the socket at path with the given access.
//
// Access control is the socket's own mode and group, which the node is created no
// looser than and which are set after bind, together with the peer check the server
// applies on accept unless a group shares the socket, in which case the kernel's check
// is the whole of it: connect(2) needs write permission on the socket file on every
// unix we build for.
//
// The directory is not part of that, and is never chmoded: a directory a service
// manager made for us is served from at whatever mode it came with (systemd's
// RuntimeDirectoryMode= is 0755 by default), and one an operator pointed us into keeps
// whatever else it holds readable. The one we make ourselves is 0755, searchable by
// everyone, so that a nebula ctl run as another user reaches the socket and gets the
// kernel's answer, which for a member of the configured group is a connection and for
// anyone else is permission denied.
func listen(path string, acc Access) (net.Listener, error) {
// The kernel silently truncates an over long sun_path and bind then fails with
// "invalid argument", which tells an operator nothing. Say what is actually wrong.
if pathTooLong(path) {
return nil, fmt.Errorf("socket path is %d bytes, over the %d byte limit for a unix socket: %s", len(path), maxSocketPath, path)
}
dir := filepath.Dir(path)
// One level only. Every directory defaultPathFor offers sits directly under one that
// exists, and an operator who names a deeper path creates it; a missing parent is a
// mistake to report, not a tree to build. A directory made here stays when the
// socket goes: an empty one costs nothing, and the next start wants it again.
//
// The mode comes from the umask around the call, never from a chmod afterwards: a
// chmod by path could land on whatever somebody who can write the parent swapped in
// between the two calls, and a chmod through a descriptor on a root-owned directory
// they moved there, since root owns both. Born 0755 there is nothing to fix up.
// The umask is process wide and the window is one syscall; see umaskMu.
umaskMu.Lock()
old := syscall.Umask(0022)
err := os.Mkdir(dir, 0755)
syscall.Umask(old)
umaskMu.Unlock()
if err != nil && !errors.Is(err, os.ErrExist) {
return nil, fmt.Errorf("could not create %s: %w", dir, err)
}
// Whether it was made here or came with the environment, the directory has to be
// one nobody else could change; see checkSocketDir.
if err := checkSocketDir(dir); err != nil {
return nil, err
}
if err := clearStale(path); err != nil {
return nil, err
}
// bind creates the socket node at the process umask, and a connect(2) between that
// and a later chmod would be queued for accept. Tighten the umask around the bind so
// the node is never looser than 0600, however briefly; see umaskMu.
umaskMu.Lock()
old = syscall.Umask(0077)
ln, err := net.Listen(network, path)
syscall.Umask(old)
umaskMu.Unlock()
if err != nil {
return nil, err
}
// The node at path is given up by release, which unlinks it before the listener
// closes; see there for why the order matters. Go's listener would unlink on Close
// as well, and an unlink after the close could only be of whatever bound the path
// next, so that is turned off and release is the one place the node is removed.
if ul, ok := ln.(*net.UnixListener); ok {
ul.SetUnlinkOnClose(false)
}
// The umask only promises no looser than 0600; this pins the mode and the group
// exactly, and is the one place a failure to secure the socket is reported.
if err := applyAccess(path, acc); err != nil {
release(ln, path)
return nil, fmt.Errorf("could not secure %s: %w", path, err)
}
return ln, nil
}
// applyAccess sets the owning group and the mode of the socket node at path. With no
// group configured the socket is handed back to the daemon's own group, so taking the
// key out of the config on a reload takes the access away as surely as a restart would.
// The chown is Lchown, so a link at the path is not followed to its target.
func applyAccess(path string, acc Access) error {
gid, err := gidFor(acc)
if err != nil {
return err
}
if err := os.Lchown(path, -1, gid); err != nil {
return err
}
return os.Chmod(path, acc.Mode.Perm())
}
// checkSocketDir refuses to serve from a directory that somebody other than the user
// nebula runs as, or root, could change. The socket's mode and group say who may connect
// to it; what makes the path they connect to the daemon's is the directory, since
// whoever can write it can put a socket of their own at the path and answer the clients
// in the daemon's place. So the directory must be owned by the daemon's user or by root
// and writable by its owner alone; a sticky bit is not an exception. A runtime directory
// a service manager made for another user, or /var/run/nebula created ahead of a root
// daemon by a member of group daemon, where darwin lets that group write /var/run, are
// both refused with the owner named.
//
// A link where the directory should be is judged twice: whoever owns the link decides
// where the path leads, so the link must be ours or root's too, and the directory it
// resolves to is then held to the rule above. /var/run is such a link on linux, root's,
// to /run, and a socket named directly under it is served; a link somebody else planted
// is refused as theirs.
//
// Only this directory is checked, not the ones above it. /var/run itself is group
// writable on darwin by design, and the parents are the operator's to trust, as they are
// for every file nebula reads. The check is made at bind, so a directory loosened while
// the daemon runs is noticed at the next reload.
func checkSocketDir(dir string) error {
fi, err := os.Lstat(dir)
if err != nil {
return err
}
if fi.Mode()&os.ModeSymlink != 0 {
if uid, ok := owner(fi); ok && uid != currentUID() && uid != 0 {
return fmt.Errorf("%s is a symlink owned by %s, not by the user nebula runs as or by root, and its owner decides where the path leads; name a socket path that does not go through it", dir, describeUID(uid))
}
if fi, err = os.Stat(dir); err != nil {
return err
}
}
if !fi.IsDir() {
return fmt.Errorf("%s is not a directory", dir)
}
if uid, ok := owner(fi); ok && uid != currentUID() && uid != 0 {
return fmt.Errorf("%s is owned by %s, not by the user nebula runs as or by root, and its owner could replace the socket; make it ours or root's, or name a socket path in a directory that is", dir, describeUID(uid))
}
if perm := fi.Mode().Perm(); perm&0022 != 0 {
who := "its group"
if perm&0002 != 0 {
who = "everyone"
}
return fmt.Errorf("%s is writable by %s (mode %04o), who could replace the socket; make it writable by its owner alone, or name a socket path in a directory that is", dir, who, perm)
}
return nil
}
// owner returns the uid that owns the file fi describes.
func owner(fi os.FileInfo) (uint32, bool) {
st, ok := fi.Sys().(*syscall.Stat_t)
if !ok {
return 0, false
}
return uint32(st.Uid), true
}
// describeUID renders a uid with its account name when the host can say what it is, so
// a message about somebody else's socket tells the operator who that is.
func describeUID(uid uint32) string {
id := strconv.FormatUint(uint64(uid), 10)
if u, err := user.LookupId(id); err == nil && u.Username != "" {
return "uid " + id + " (" + u.Username + ")"
}
return "uid " + id
}
// clearStale removes a leftover socket file, but only after proving nothing is
// listening on it. An unclean restart leaves one behind every time, so this is a
// routine path; blindly unlinking would let a second daemon steal a live endpoint.
//
// Liveness is a plain connect. Only a refused connect proves there is nothing behind
// the file. Anything else, a timeout or a permission error, is a socket we cannot vouch
// for, and it is left alone with the reason in the error.
func clearStale(path string) error {
fi, err := os.Stat(path)
if err != nil {
if os.IsNotExist(err) {
return nil
}
return err
}
if !isSocket(fi) {
return fmt.Errorf("%s exists and is not a socket", path)
}
conn, err := dial(path, 250*time.Millisecond)
if err == nil {
_ = conn.Close()
return fmt.Errorf("%s is already in use by another nebula instance", path)
}
if errors.Is(err, os.ErrNotExist) {
// Gone between the stat and the connect; nothing to clear.
return nil
}
if !errors.Is(err, syscall.ECONNREFUSED) {
return fmt.Errorf("%s exists and could not be shown to be stale, refusing to remove it: %w", path, err)
}
return os.Remove(path)
}
// removeSocket unlinks the socket node. Its directory stays, whoever made it; see
// listen.
func removeSocket(path string) {
_ = os.Remove(path)
}
func dial(path string, timeout time.Duration) (net.Conn, error) {
return net.DialTimeout(network, path, timeout)
}
// sunPathMax is the smallest sun_path any platform we build for offers (104 on darwin,
// 108 on linux). Overrunning it truncates the path rather than failing cleanly, so we
// check against the tighter bound everywhere.
const sunPathMax = 104
// maxSocketPath is the longest path that actually binds. sun_path holds the trailing
// NUL as well, and Go's syscall package refuses a path that fills the whole buffer
// with a bare EINVAL, so the usable limit is one byte short of the buffer.
const maxSocketPath = sunPathMax - 1
// umaskMu serialises the two places the process umask is set around one syscall: the
// umask is process wide, and two endpoints binding at once, a reload landing during
// startup, would otherwise interleave their save and restore and leave the daemon's
// umask at one of ours. It is held for the syscall alone. Another goroutine creating a
// file in that moment gets the umask of the window, 0022 or 0077, rather than the
// process's own; nothing creates files on a hot path.
var umaskMu sync.Mutex
func pathTooLong(path string) bool { return len(path) > maxSocketPath }
// DefaultSocketPath is the control socket when the config names none. It is root's to
// create, as the tun device is root's to open; a daemon run as anyone else names a path
// of its own with CtlSocketKey.
const DefaultSocketPath = "/var/run/" + DirName + "/ctl.sock"
// checkSocketPath refuses a configured socket path that is relative, or too long to
// bind. A relative path would mean a different socket to the daemon and to nebula ctl,
// each resolving it against the directory it happened to be started from.
func checkSocketPath(key, value string) error {
if !filepath.IsAbs(value) {
return fmt.Errorf("%s must be an absolute path, got %q", key, value)
}
if pathTooLong(value) {
return fmt.Errorf("%s is %d bytes, too long for this platform: %s", key, len(value), value)
}
return nil
}
func isSocket(fi os.FileInfo) bool { return fi.Mode()&os.ModeSocket != 0 }
+358
View File
@@ -0,0 +1,358 @@
//go:build !windows
package localapi
import (
"bufio"
"context"
"net"
"os"
"path/filepath"
"strings"
"syscall"
"testing"
"time"
"github.com/slackhq/nebula/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// privateBase returns a short directory of our own, 0700, for tests to lay sockets out
// under, short enough that a path under it fits sun_path.
func privateBase(t *testing.T) string {
t.Helper()
return test.ShortTempDir(t)
}
// openDir returns a world writable directory under base, the shape of /tmp.
func openDir(t *testing.T, base string) string {
t.Helper()
open := filepath.Join(base, "open")
require.NoError(t, os.Mkdir(open, 0777))
require.NoError(t, os.Chmod(open, 0777|os.ModeSticky))
return open
}
// TestListenLeavesAServiceManagersDirectoryAlone covers the directory a service manager
// pre-creates: systemd's RuntimeDirectory is 0755 by default. The socket is 0600, which
// is what refuses other users, so the directory is served from as it came and nothing
// is chmoded that the operator did not set.
func TestListenLeavesAServiceManagersDirectoryAlone(t *testing.T) {
base := privateBase(t)
dir := filepath.Join(base, DirName)
require.NoError(t, os.Mkdir(dir, 0755))
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(context.Background(), filepath.Join(dir, "s.sock"), ownerOnly()))
t.Cleanup(s.Stop)
fi, err := os.Stat(dir)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0755), fi.Mode().Perm(), "the directory is left exactly as it was")
fi, err = os.Stat(s.Path())
require.NoError(t, err)
assert.Equal(t, os.FileMode(0600), fi.Mode().Perm(), "the socket is what is private")
_, err = (&Client{Path: s.Path()}).Describe()
require.NoError(t, err)
}
// TestListenCreatesOneLevelOnly: the socket directory is created when it is missing,
// but nothing above it is. A runtime directory that does not exist is the operator's to
// provide, and the failure says which directory could not be created.
func TestListenCreatesOneLevelOnly(t *testing.T) {
base := privateBase(t)
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(context.Background(), filepath.Join(base, DirName, "s.sock"), ownerOnly()))
s.Stop()
dir := filepath.Join(base, "missing", DirName)
err := s.Rebind(context.Background(), filepath.Join(dir, "s.sock"), ownerOnly())
require.Error(t, err)
require.ErrorContains(t, err, "could not create "+dir)
assert.NoDirExists(t, filepath.Join(base, "missing"))
}
// TestListenRestoresTheUmask: listen tightens the process umask around bind so the socket
// node is never created looser than 0600. That is process wide state, so it must come
// back exactly as it was, on success and on failure alike.
func TestListenRestoresTheUmask(t *testing.T) {
old := syscall.Umask(0022)
t.Cleanup(func() { syscall.Umask(old) })
base := privateBase(t)
ln, err := listen(filepath.Join(base, DirName, "s.sock"), ownerOnly())
require.NoError(t, err)
t.Cleanup(func() { ln.Close() })
assert.Equal(t, 0022, syscall.Umask(0022), "umask after a successful listen")
_, err = listen(filepath.Join(base, DirName, "s.sock"), ownerOnly())
require.Error(t, err, "the path is already bound")
assert.Equal(t, 0022, syscall.Umask(0022), "umask after a failed listen")
}
// TestListenServesFromASharedDirectory: an operator who points local_api.control.socket
// into a directory holding other files, /etc/nebula say, gets a socket there and keeps
// the config and certs readable by everyone who could read them before.
func TestListenServesFromASharedDirectory(t *testing.T) {
base := privateBase(t)
dir := filepath.Join(base, "etc-nebula")
require.NoError(t, os.Mkdir(dir, 0755))
cfg := filepath.Join(dir, "config.yml")
require.NoError(t, os.WriteFile(cfg, []byte("{}\n"), 0644))
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(context.Background(), filepath.Join(dir, "ctl.sock"), ownerOnly()))
t.Cleanup(s.Stop)
fi, err := os.Stat(dir)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0755), fi.Mode().Perm())
fi, err = os.Stat(cfg)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0644), fi.Mode().Perm())
}
func TestOverlongPathIsDiagnosed(t *testing.T) {
long := filepath.Join(os.TempDir(), string(make([]byte, 0)))
for len(long) <= sunPathMax {
long = filepath.Join(long, "abcdefghij")
}
s := NewServer(test.NewLogger(), basicRegistry())
err := s.Rebind(context.Background(), filepath.Join(long, "s.sock"), ownerOnly())
require.Error(t, err)
assert.ErrorContains(t, err, "byte limit for a unix socket")
}
// TestPathLimitIsExact pins the boundary: sun_path has room for the NUL as well, and Go
// refuses a path that fills the whole buffer, so the last usable length is one short.
func TestPathLimitIsExact(t *testing.T) {
assert.False(t, pathTooLong(strings.Repeat("a", maxSocketPath)))
assert.True(t, pathTooLong(strings.Repeat("a", maxSocketPath+1)))
assert.True(t, pathTooLong(strings.Repeat("a", sunPathMax)))
}
// TestNamedPathIsUsedAsGiven: a path the operator named is served from, and dialed,
// wherever it is, a directory other users can write to included; the socket's own mode
// is what keeps them out.
func TestNamedPathIsUsedAsGiven(t *testing.T) {
base := privateBase(t)
path := filepath.Join(openDir(t, base), DirName, "s.sock")
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
t.Cleanup(s.Stop)
_, err := (&Client{Path: path}).Describe()
require.NoError(t, err)
}
// TestStaleSocketIsCleared: a dead socket in a good directory is the routine leftover of
// a hard kill, and the daemon's stale check must see it as free to replace.
func TestStaleSocketIsCleared(t *testing.T) {
path := testSocket(t)
require.NoError(t, os.Mkdir(filepath.Dir(path), 0700))
ln, err := net.Listen(network, path)
require.NoError(t, err)
require.NoError(t, os.Chmod(path, 0600))
// Close the listener but keep the file, the way a hard kill does.
ln.(*net.UnixListener).SetUnlinkOnClose(false)
require.NoError(t, ln.Close())
_, statErr := os.Stat(path)
require.NoError(t, statErr, "the socket file is still there")
require.NoError(t, clearStale(path), "a stale socket is cleared, not reported as in use")
_, statErr = os.Stat(path)
assert.True(t, os.IsNotExist(statErr))
}
// TestClientSaysWhenTheReplyCannotBeRead: something answered on the socket, but not
// with a response. That is reported as a reply that could not be read, which is what
// it is, and not as a failure to reach the daemon, whether the other side read the
// request first or hung up on it: a daemon that is not nebula may do either.
func TestClientSaysWhenTheReplyCannotBeRead(t *testing.T) {
for _, readsFirst := range []bool{true, false} {
path := testSocket(t)
require.NoError(t, os.Mkdir(filepath.Dir(path), 0700))
ln, err := net.Listen(network, path)
require.NoError(t, err)
require.NoError(t, os.Chmod(path, 0600))
go func() {
c, err := ln.Accept()
if err != nil {
return
}
if readsFirst {
_, _ = bufio.NewReader(c).ReadBytes('\n')
}
_, _ = c.Write([]byte("this is not a response\n"))
c.Close()
}()
_, err = (&Client{Path: path}).Describe()
var unread *ReplyError
require.ErrorAs(t, err, &unread, "reads first: %v", readsFirst)
assert.Contains(t, err.Error(), "the daemon's reply could not be read")
ln.Close()
}
}
// TestClientGivesUpOnASilentDaemon: a daemon that accepts and then never answers must
// not hang the client for good. A script wrapping nebula ctl has no other way out.
func TestClientGivesUpOnASilentDaemon(t *testing.T) {
path := testSocket(t)
require.NoError(t, os.Mkdir(filepath.Dir(path), 0700))
ln, err := net.Listen(network, path)
require.NoError(t, err)
defer ln.Close()
require.NoError(t, os.Chmod(path, 0600))
accepted := make(chan net.Conn, 1)
go func() {
c, err := ln.Accept()
if err == nil {
accepted <- c // held open, never written to
}
}()
start := time.Now()
_, err = (&Client{Path: path, Timeout: 200 * time.Millisecond}).Describe()
require.Error(t, err)
require.ErrorContains(t, err, "did not answer within")
assert.Less(t, time.Since(start), 5*time.Second)
select {
case c := <-accepted:
_ = c.Close()
default:
}
}
// asUID makes the owner checks run as uid for the rest of the test. The socket the test
// serves is still owned by the real user, so from uid's point of view it belongs to
// somebody else: the shape of a service account's daemon, and of an impostor's.
func asUID(t *testing.T, uid uint32) {
t.Helper()
real := currentUID
currentUID = func() uint32 { return uid }
t.Cleanup(func() { currentUID = real })
}
// TestPeerCheckRefusesAnotherUser: a server built as one user and serving as another
// refuses the real user at the peer check, and the refusal reaches the client as the
// daemon's answer rather than as a failure to reach it.
func TestPeerCheckRefusesAnotherUser(t *testing.T) {
if !peerCheckSupported {
t.Skip("no peer credentials on this platform")
}
me := currentUID()
asUID(t, me+1)
s := NewServer(test.NewLogger(), basicRegistry())
asUID(t, me)
path := testSocket(t)
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
defer s.Stop()
_, err := (&Client{Path: path}).Describe()
require.Error(t, err)
require.ErrorContains(t, err, "not permitted")
var refused *DaemonError
assert.ErrorAs(t, err, &refused, "the daemon answered; this is not a reachability failure")
}
// TestListenerDoesNotUnlinkOnClose: the server unlinks its socket node itself, before
// the listener closes, and the listener must not unlink again afterwards. By the time a
// close happens the path may belong to a newer daemon that found the node stale and
// bound its own, and a second unlink would take that daemon's socket from under it.
func TestListenerDoesNotUnlinkOnClose(t *testing.T) {
path := filepath.Join(privateBase(t), DirName, "s.sock")
old, err := listen(path, ownerOnly())
require.NoError(t, err)
// A newer daemon's start, with the old node gone and the path bound afresh.
require.NoError(t, os.Remove(path))
next, err := net.Listen(network, path)
require.NoError(t, err)
t.Cleanup(func() { next.Close() })
require.NoError(t, old.Close())
_, err = os.Stat(path)
assert.NoError(t, err, "closing the old listener left the newer daemon's socket in place")
}
// TestListenRefusesADirectoryOthersCanWrite: whoever can write the socket's directory
// can put a socket of their own at the path and answer the clients, who trust the path;
// so a directory writable by its group or by everyone is refused, and nothing is bound.
func TestListenRefusesADirectoryOthersCanWrite(t *testing.T) {
for _, mode := range []os.FileMode{0775, 0757, 0777, 01777} {
dir := filepath.Join(privateBase(t), "loose")
require.NoError(t, os.Mkdir(dir, mode))
require.NoError(t, os.Chmod(dir, mode))
_, err := listen(filepath.Join(dir, "s.sock"), ownerOnly())
require.ErrorContains(t, err, "writable by", "mode %04o", mode)
assert.NoFileExists(t, filepath.Join(dir, "s.sock"))
}
}
// TestListenRefusesADirectoryOwnedBySomebodyElse: a directory that is neither ours nor
// root's is refused, since its owner could replace the socket. The test's own directory
// stands in for somebody else's by the process pretending to be another user.
func TestListenRefusesADirectoryOwnedBySomebodyElse(t *testing.T) {
me := currentUID()
if me == 0 {
t.Skip("root's directories are accepted whoever is asking")
}
dir := filepath.Join(privateBase(t), "theirs")
require.NoError(t, os.Mkdir(dir, 0755))
asUID(t, me+1)
_, err := listen(filepath.Join(dir, "s.sock"), ownerOnly())
require.ErrorContains(t, err, "is owned by")
assert.NoFileExists(t, filepath.Join(dir, "s.sock"))
asUID(t, me)
ln, err := listen(filepath.Join(dir, "s.sock"), ownerOnly())
require.NoError(t, err, "the same directory, seen as our own, serves")
release(ln, filepath.Join(dir, "s.sock"))
}
// TestListenThroughASymlinkedDirectory: a link where the directory should be is judged
// by who owns it, since they decide where the path leads, and then by the directory it
// resolves to. Our own link to a proper directory serves, as /var/run on linux is root's
// link to /run; somebody else's link is refused, and so is a link of ours to a directory
// others can write.
func TestListenThroughASymlinkedDirectory(t *testing.T) {
me := currentUID()
if me == 0 {
t.Skip("root's links are accepted whoever is asking")
}
base := privateBase(t)
real := filepath.Join(base, "real")
require.NoError(t, os.Mkdir(real, 0755))
link := filepath.Join(base, "link")
require.NoError(t, os.Symlink(real, link))
path := filepath.Join(link, "s.sock")
ln, err := listen(path, ownerOnly())
require.NoError(t, err, "our link to our directory")
release(ln, path)
assert.NoFileExists(t, filepath.Join(real, "s.sock"))
asUID(t, me+1)
_, err = listen(path, ownerOnly())
require.ErrorContains(t, err, "is a symlink owned by")
asUID(t, me)
require.NoError(t, os.Chmod(real, 0777))
_, err = listen(path, ownerOnly())
require.ErrorContains(t, err, "writable by everyone")
}
+255
View File
@@ -0,0 +1,255 @@
//go:build windows
package localapi
import (
"context"
"errors"
"fmt"
"net"
"strings"
"time"
"github.com/Microsoft/go-winio"
"golang.org/x/sys/windows"
)
// pipePrefix is where windows named pipes live. The pipe's own security descriptor
// carries the access control that the socket's mode carries on unix.
const pipePrefix = `\\.\pipe\`
// protectedPrefix is the pipe namespace that only administrators and SYSTEM may create
// names under; the kernel refuses everyone else. The default name lives there, so the
// name itself says who could have created the pipe answering it: no ordinary user can
// have put anything under it ahead of the daemon. Tailscale, WireGuard and NetBird keep
// their daemon pipes there for the same reason. nebula on windows is already an
// administrator, since creating the tun adapter requires it.
const protectedPrefix = pipePrefix + `ProtectedPrefix\Administrators\`
const network = "pipe"
// currentUID is the identity the unix owner checks are made for. There is no uid here
// and no per-connection check that would read one, so the value is never consulted; it
// exists so a Server is built the same way on every platform.
var currentUID = func() uint32 { return 0 }
// listen creates the named pipe at path with a security descriptor that admits the user
// nebula runs as, SYSTEM and Administrators, and the group acc names, if any, for
// reading and writing, which is what connecting to a pipe takes. acc's mode means
// nothing to a pipe.
//
// The first instance is created, never opened, so a name some other process already
// holds is refused rather than joined: that process would otherwise answer clients in
// nebula's place. Under protectedPrefix only an administrator could hold it; anywhere
// else any account could. Either way the refusal is reported as errEndpointHeld for the
// caller to log as the event it is, except when the refusal is the namespace's own: a
// daemon that is not an administrator cannot create under protectedPrefix at all, and
// that is told apart so the operator is sent to elevate or to name a pipe elsewhere
// rather than to look for a process that is not there.
func listen(path string, acc Access) (net.Listener, error) {
sddl, err := sddlFor(acc)
if err != nil {
return nil, fmt.Errorf("could not build a security descriptor for %s: %w", path, err)
}
ln, err := winio.ListenPipe(path, &winio.PipeConfig{
SecurityDescriptor: sddl,
InputBufferSize: int32(MaxRequestBytes),
OutputBufferSize: 64 * 1024,
})
if err != nil {
return nil, listenRefused(path, isAdministrator(), err)
}
return ln, nil
}
// listenRefused turns a failure to create the pipe at path into the error the caller
// logs. admin says whether this process may create under protectedPrefix.
//
// ERROR_ACCESS_DENIED is ambiguous: it is what the kernel says to a non-administrator
// creating under protectedPrefix, and also what an existing instance whose DACL does
// not admit us says to a second create. A process that is an administrator cannot be
// getting the first, so for it, and for every name outside the prefix, access denied
// means held.
func listenRefused(path string, admin bool, err error) error {
if errors.Is(err, windows.ERROR_ACCESS_DENIED) && isProtectedPath(path) && !admin {
return fmt.Errorf("creating %s requires an administrator, as only administrators may create pipes under %s; run nebula elevated, or set %s to a pipe name outside it: %w", path, protectedPrefix, CtlSocketKey, err)
}
if errors.Is(err, windows.ERROR_ALREADY_EXISTS) || errors.Is(err, windows.ERROR_ACCESS_DENIED) {
return fmt.Errorf("%w: %s: %w", errEndpointHeld, path, err)
}
return err
}
// isProtectedPath says whether path is a name under protectedPrefix. Pipe names are
// case insensitive, so the comparison is too.
func isProtectedPath(path string) bool {
return len(path) > len(protectedPrefix) && strings.EqualFold(path[:len(protectedPrefix)], protectedPrefix)
}
// isAdministrator says whether this process holds Administrators as an enabled group,
// which is what creating under protectedPrefix takes. SYSTEM holds it; an elevated
// administrator holds it; an administrator's unelevated token carries it deny-only,
// which the membership check correctly reads as no. The membership is checked against
// the thread's own token: CheckTokenMembership wants an impersonation token and makes
// one from the primary token when handed none. A check that cannot be made is read as
// not an administrator, which only changes the wording of a failure that has already
// happened.
func isAdministrator() bool {
admins, err := windows.CreateWellKnownSid(windows.WinBuiltinAdministratorsSid)
if err != nil {
return false
}
member, err := windows.Token(0).IsMember(admins)
return err == nil && member
}
// sddlFor grants full access to the account nebula is running as, to SYSTEM and to
// Administrators, and pipeClientAccess to the group acc names, if any; nobody else. P
// makes the DACL protected, so it does not inherit entries that would widen it. It is
// dockerd's descriptor for its pipe, with its -G group in the same place, except for
// the group's mask; see pipeClientAccess.
//
// Administrators is the counterpart of the uid 0 allowance in the unix peer check. The
// service normally runs as LocalSystem, so without it an operator in an elevated shell
// could not reach the daemon at all; and an administrator can already stop the service,
// so this grants no authority they did not have.
func sddlFor(acc Access) (string, error) {
token := windows.GetCurrentProcessToken()
user, err := token.GetTokenUser()
if err != nil {
return "", err
}
sddl := fmt.Sprintf("D:P(A;;GA;;;%s)(A;;GA;;;SY)(A;;GA;;;BA)", user.User.Sid.String())
if acc.shared() {
sddl += fmt.Sprintf("(A;;0x%x;;;%s)", pipeClientAccess, acc.Group.ID)
}
return sddl, nil
}
// pipeClientAccess is what a client needs on the pipe to talk to the daemon, and all a
// configured group is granted: read and write the data, read the pipe's attributes
// and its security descriptor, which the owner check needs, and wait on it. Not the
// generic rights. On a named pipe GENERIC_WRITE carries FILE_CREATE_PIPE_INSTANCE, the
// right to put a second instance under the same name, and a group member with that
// could sit on the name and answer other clients in the daemon's place; the client's
// owner check would not catch it, since the instance it reached would be the
// member's. A client therefore has to ask for exactly this, as dial does, and a
// program that opens the info pipe with generic read and write is refused.
const pipeClientAccess uint32 = windows.FILE_READ_DATA | windows.FILE_WRITE_DATA |
windows.FILE_READ_ATTRIBUTES | windows.READ_CONTROL | windows.SYNCHRONIZE
// applyAccess cannot change a pipe's access in place: its security descriptor is fixed
// when it is created, so a change means creating it again, which the caller does.
func applyAccess(path string, acc Access) error { return errReopenForAccess }
// removeSocket is a no-op: a named pipe disappears when its last handle closes, so
// there is nothing to unlink and no stale instance to clean up.
func removeSocket(path string) {}
// dial connects to the pipe and, before the connection carries anything, checks who
// serves it. The request names a command and its arguments, so it must not reach a
// pipe that some other account put under nebula's name. For the default name the
// namespace already promises that, since only an administrator could have created it;
// the check is what covers a CtlSocketKey override outside protectedPrefix, and costs
// nothing to make on every name.
func dial(path string, timeout time.Duration) (net.Conn, error) {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
conn, err := winio.DialPipeAccess(ctx, path, pipeClientAccess)
if err != nil {
return nil, err
}
if err := checkPipeOwner(conn, path); err != nil {
conn.Close()
return nil, err
}
return conn, nil
}
// checkPipeOwner refuses a pipe whose owner is not an account that could be running
// nebula. It is the unix owner check made from the other side: the handle winio dialed
// exposes the pipe object, and the object's owner is whoever created it.
//
// The owner is what makes this sound. An account can only create an object owned by
// itself, or by a SID it holds with the owner flag, unless it has SeRestorePrivilege,
// which ordinary users do not; so a pipe owned by SYSTEM, Administrators or a service
// account was put there by one of them, and never by a user squatting the name. winio
// dials with SECURITY_ANONYMOUS, so the impostor could not have impersonated the client
// either, but it could have read the command and answered it with whatever it liked.
func checkPipeOwner(conn net.Conn, path string) error {
h, ok := conn.(interface{ Fd() uintptr })
if !ok {
return fmt.Errorf("%s: cannot identify who serves the pipe", path)
}
sd, err := windows.GetSecurityInfo(windows.Handle(h.Fd()), windows.SE_KERNEL_OBJECT, windows.OWNER_SECURITY_INFORMATION)
if err != nil {
return fmt.Errorf("%s: could not read the pipe's owner: %w", path, err)
}
owner, _, err := sd.Owner()
if err != nil {
return fmt.Errorf("%s: could not read the pipe's owner: %w", path, err)
}
if owner == nil {
return fmt.Errorf("%s has no owner; refusing to send to it", path)
}
me, err := windows.GetCurrentProcessToken().GetTokenUser()
if err != nil {
return fmt.Errorf("%s: could not read own identity: %w", path, err)
}
if ownerAccepted(owner, me.User.Sid) {
return nil
}
return fmt.Errorf("%s is served by %s, not by nebula; run nebula ctl as that user, or pass -config or -socket to name the daemon's pipe", path, describeSID(owner))
}
// ownerAccepted says whether a pipe owned by owner may be sent a request by a client
// running as me: the client's own account, and the two accounts a daemon runs under
// that no ordinary user can create an object as, SYSTEM and Administrators. Another
// account is refused, whoever the client is; the daemon's DACL decides who may connect,
// and this decides who may answer.
func ownerAccepted(owner, me *windows.SID) bool {
return owner.Equals(me) ||
owner.IsWellKnown(windows.WinLocalSystemSid) ||
owner.IsWellKnown(windows.WinBuiltinAdministratorsSid)
}
// describeSID renders a SID with its account name when the host can say what it is, so
// a message about somebody else's pipe tells the operator who that is.
func describeSID(sid *windows.SID) string {
account, domain, _, err := sid.LookupAccount("")
if err != nil || account == "" {
return sid.String()
}
if domain == "" {
return account
}
return domain + `\` + account
}
// DefaultSocketPath is the control pipe when the config names none. It is under
// protectedPrefix, so whatever answers it was created by an administrator or SYSTEM; see
// there.
const DefaultSocketPath = protectedPrefix + "nebula-ctl"
// checkSocketPath refuses a configured name that is not a full pipe name. A pipe lives
// under pipePrefix and nowhere else, so a bare name has no meaning here, and one under
// pipePrefix is used as given. An override outside protectedPrefix is a name any
// account could have created first; the owner check in dial is what protects a client
// of one.
func checkSocketPath(key, value string) error {
if len(value) <= len(pipePrefix) || !strings.EqualFold(value[:len(pipePrefix)], pipePrefix) {
return fmt.Errorf(`%s must be a full pipe name under %s, got %q`, key, pipePrefix, value)
}
return nil
}
+186
View File
@@ -0,0 +1,186 @@
//go:build windows
package localapi
import (
"errors"
"net"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/sys/windows"
)
// TestOwnerAccepted pins which accounts may answer a nebula ctl request: the client's
// own, and the accounts a daemon runs as that an ordinary user cannot create an object
// owned by. Any other user is refused, since any user can create any pipe name.
func TestOwnerAccepted(t *testing.T) {
me, err := windows.GetCurrentProcessToken().GetTokenUser()
require.NoError(t, err)
wellKnown := func(typ windows.WELL_KNOWN_SID_TYPE) *windows.SID {
sid, err := windows.CreateWellKnownSid(typ)
require.NoError(t, err)
return sid
}
parse := func(s string) *windows.SID {
sid, err := windows.StringToSid(s)
require.NoError(t, err)
return sid
}
accepted := map[string]*windows.SID{
"self": me.User.Sid,
"SYSTEM": wellKnown(windows.WinLocalSystemSid),
"Administrators": wellKnown(windows.WinBuiltinAdministratorsSid),
}
for name, sid := range accepted {
assert.True(t, ownerAccepted(sid, me.User.Sid), name)
}
refused := map[string]*windows.SID{
"Everyone": wellKnown(windows.WinWorldSid),
"Users": wellKnown(windows.WinBuiltinUsersSid),
"another user": parse("S-1-5-21-1-2-3-1001"),
"a service's own account": parse("S-1-5-80-1-2-3-4-5"),
"LocalService": wellKnown(windows.WinLocalServiceSid),
}
for name, sid := range refused {
assert.False(t, ownerAccepted(sid, me.User.Sid), name)
}
}
// TestDefaultPathIsProtected pins the default name to the namespace only administrators
// and SYSTEM can create under, which is what lets nebula ctl trust the default name
// from its spelling alone.
func TestDefaultPathIsProtected(t *testing.T) {
p := DefaultSocketPath
assert.True(t, isProtectedPath(p), p)
assert.Equal(t, `\\.\pipe\ProtectedPrefix\Administrators\nebula-ctl`, p)
}
// TestConfiguredPipeNameMustBeFull: a configured name is used as given, and only a full
// pipe name means anything here.
func TestConfiguredPipeNameMustBeFull(t *testing.T) {
p, err := SocketPathFromConfig(fakeWinConfig{CtlSocketKey: `\\.\pipe\nebula-ctl`})
require.NoError(t, err)
assert.Equal(t, `\\.\pipe\nebula-ctl`, p)
for _, bad := range []string{"nebula-ctl", `C:\nebula\ctl.sock`, `\\.\pipe\`} {
_, err := SocketPathFromConfig(fakeWinConfig{CtlSocketKey: bad})
require.Error(t, err, bad)
assert.ErrorContains(t, err, "full pipe name", bad)
}
}
type fakeWinConfig map[string]string
func (f fakeWinConfig) GetString(k, d string) string {
if v, ok := f[k]; ok {
return v
}
return d
}
func TestIsProtectedPath(t *testing.T) {
assert.True(t, isProtectedPath(`\\.\pipe\ProtectedPrefix\Administrators\nebula-1`))
assert.True(t, isProtectedPath(`\\.\pipe\protectedprefix\administrators\nebula-1`), "pipe names are case insensitive")
assert.False(t, isProtectedPath(`\\.\pipe\ProtectedPrefix\Administrators\`), "the prefix alone names nothing")
assert.False(t, isProtectedPath(`\\.\pipe\nebula-1`))
assert.False(t, isProtectedPath(`\\.\pipe\ProtectedPrefix\Users\nebula-1`))
}
// TestListenRefused pins how a failed create is reported. Access denied under the
// protected prefix is the namespace refusing a non-administrator, and says so; the same
// error for an administrator, or for any name outside the prefix, is a name some other
// process holds, and is errEndpointHeld so the daemon logs it as an error.
func TestListenRefused(t *testing.T) {
protected := `\\.\pipe\ProtectedPrefix\Administrators\nebula-1`
plain := `\\.\pipe\nebula-1`
err := listenRefused(protected, false, windows.ERROR_ACCESS_DENIED)
assert.NotErrorIs(t, err, errEndpointHeld)
assert.ErrorIs(t, err, windows.ERROR_ACCESS_DENIED)
assert.Contains(t, err.Error(), "requires an administrator")
assert.Contains(t, err.Error(), CtlSocketKey)
assert.ErrorIs(t, listenRefused(protected, true, windows.ERROR_ACCESS_DENIED), errEndpointHeld)
assert.ErrorIs(t, listenRefused(plain, false, windows.ERROR_ACCESS_DENIED), errEndpointHeld)
assert.ErrorIs(t, listenRefused(protected, false, windows.ERROR_ALREADY_EXISTS), errEndpointHeld)
assert.ErrorIs(t, listenRefused(plain, true, windows.ERROR_ALREADY_EXISTS), errEndpointHeld)
other := errors.New("something else")
assert.Same(t, other, listenRefused(protected, false, other))
}
// TestListenProtectedName creates and dials a name under the protected prefix. It needs
// an administrator, and otherwise checks only that the refusal is the one listen
// promises a non-administrator.
func TestListenProtectedName(t *testing.T) {
path := protectedPrefix + "nebula-test-" + t.Name()
ln, err := listen(path, ownerOnly())
if !isAdministrator() {
require.Error(t, err)
assert.Contains(t, err.Error(), "requires an administrator")
t.Skip("not an administrator; the protected namespace refused the create as it should")
}
require.NoError(t, err)
t.Cleanup(func() { ln.Close() })
accepted := make(chan error, 1)
go func() {
conn, err := ln.Accept()
if err == nil {
conn.Close()
}
accepted <- err
}()
conn, err := dial(path, time.Second)
require.NoError(t, err)
conn.Close()
require.NoError(t, <-accepted)
}
// TestDialAcceptsOwnPipe dials a pipe this process created outside the protected
// prefix, as an override might name, which the owner check must pass: the pipe's owner
// is this account, or Administrators when the test is elevated.
func TestDialAcceptsOwnPipe(t *testing.T) {
path := pipePrefix + "nebula-test-" + t.Name()
ln, err := listen(path, ownerOnly())
require.NoError(t, err)
t.Cleanup(func() { ln.Close() })
accepted := make(chan error, 1)
go func() {
conn, err := ln.Accept()
if err == nil {
conn.Close()
}
accepted <- err
}()
conn, err := dial(path, time.Second)
require.NoError(t, err)
conn.Close()
require.NoError(t, <-accepted)
}
// TestListenRefusesHeldName pins that a name another listener already holds is refused
// with errEndpointHeld rather than joined, which is what keeps an impostor from being
// served alongside, or in place of, the daemon.
func TestListenRefusesHeldName(t *testing.T) {
path := pipePrefix + "nebula-test-" + t.Name()
first, err := listen(path, ownerOnly())
require.NoError(t, err)
t.Cleanup(func() { first.Close() })
var second net.Listener
second, err = listen(path, ownerOnly())
if second != nil {
second.Close()
}
require.ErrorIs(t, err, errEndpointHeld)
}
+167
View File
@@ -0,0 +1,167 @@
// Package localapi serves the nebula command registry over a local IPC endpoint so the
// user who ran nebula can inspect and control it from a CLI on the same machine.
//
// Access is the socket's own: without a group it is 0600 to the user who ran nebula,
// and the server checks the peer's credentials on every connection besides; with a
// group configured it is 0660 to that group, and the kernel's check on connect is the
// whole of it, so a member has the same control of the daemon as its user. Every
// command in the registry is served; nothing is filtered on either side.
//
// # Protocol
//
// One JSON object in, one JSON object out, both newline terminated. It is an API, not a
// command line: a request names a command and passes its options as named parameters,
// rather than shipping an argv the daemon has to parse. A client that is itself a CLI
// asks for the schema first (Request.Describe) and turns the user's flags into those
// parameters; a client that is not need never think in flags at all.
package localapi
import (
"bufio"
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
)
// ProtocolVersion is the version of the request and response envelopes. A client sending
// a version this server does not know gets a sentence back rather than a parse error,
// which matters because the daemon and the CLI are upgraded independently.
const ProtocolVersion = 1
// MaxRequestBytes bounds the request. A request names a command and its parameters, so it
// has no business being large, and an unbounded read is a way for a local process to
// consume memory the packet path needs.
const MaxRequestBytes = 64 * 1024
// Request is the single JSON object a client sends.
type Request struct {
Version int `json:"version"`
// Describe asks for the schema of every command this surface serves, in place of
// running one. Command, Params and Args are ignored.
Describe bool `json:"describe,omitempty"`
// Command names the command to run.
Command string `json:"command,omitempty"`
// Params carries the command's options by name, with JSON's own types, true rather
// than "true", 3 rather than "3", or as the strings a command line parsed them from,
// which is what nebula ctl sends. Unknown names are an error, so a typo is reported
// instead of ignored. The schema lists json and pretty on every command because the
// command lines take them; here they are refused, since Format is how a request
// chooses its rendering.
Params map[string]any `json:"params,omitempty"`
// Args carries the command's operands, the values that are not options: the vpn
// address `print-tunnel` prints, for instance. They stay positional because the
// commands do not name them.
Args []string `json:"args,omitempty"`
// Format asks for a rendering other than the default. "" and "json" return the
// command's result as a JSON value, which is what a program wants. "text" asks the
// daemon to run the command's own human rendering and return it as a string, which is
// how `nebula ctl` shows an operator the same output the ssh server would.
Format string `json:"format,omitempty"`
}
// Response is the single JSON object a server sends back.
type Response struct {
Version int `json:"version"`
// Result is what a request for JSON produced: the command's value, or the schema
// when the request was a Describe, as a JSON value, so a caller reads it directly
// instead of parsing a string that happens to contain JSON.
//
// Whitespace inside a JSON result is not preserved: an envelope normalises it, and
// in an API it carries no meaning anyway. Formatting a result for a human to read is
// the client's job.
Result json.RawMessage `json:"result,omitempty"`
// Text is what a request for text produced: the command's own human rendering, the
// bytes it wrote and nothing else. They travel as bytes, base64 in the envelope,
// rather than as a JSON string, because a string is UTF-8 and the rendering need
// not be: a name from a certificate is whatever the certificate holds, and the ssh
// server passes it through, so this does too.
Text []byte `json:"text,omitempty"`
// Error explains a failure in a sentence, and is absent on success. It is the only
// signal a caller needs: there is no separate status field, because a command that
// failed always says why and one that succeeded never does. A command line client
// turns its presence into a non-zero exit.
//
// A command that ran and refused sets Error too, alongside its Result when the
// caller asked for JSON. That is how a refusal and a request that never ran are
// told apart: the first has both, the second has Error alone.
Error string `json:"error,omitempty"`
}
// frameError is a frame that arrived and was not one: oversized, or not the JSON object
// the protocol defines. It is told from a failed read so the client can say the daemon
// answered with something it could not read, rather than that it could not be reached.
type frameError struct{ err error }
func (e *frameError) Error() string { return e.err.Error() }
func (e *frameError) Unwrap() error { return e.err }
// readFrame reads one frame, a newline terminated JSON object, from r into v. With
// writeFrame it is where the framing rule lives, for both directions: a frame is a
// single line, a line over max bytes is refused rather than buffered, and a final line
// with no newline before EOF still counts. what names the frame in errors ("request", "response"); a
// read failure is returned wrapped, so a caller can still see io.EOF or a timeout in it.
// strict refuses a field v does not declare, which is the daemon's reading of a request:
// a client that misspells one deserves to hear about it, as it does for a parameter.
// The client reads a response leniently, so a field a newer daemon adds is passed over
// rather than fatal.
func readFrame(r io.Reader, max int, what string, v any, strict bool) error {
br := bufio.NewReader(io.LimitReader(r, int64(max)+1))
line, err := br.ReadBytes('\n')
if err != nil && (len(line) == 0 || !errors.Is(err, io.EOF)) {
// EOF after a partial line is the unterminated final line the rule allows, and
// an oversized line looks the same once the LimitReader cuts it off, so both go
// on to be judged by their length. Anything else is a failed read, whatever was
// buffered first: a deadline that lands mid frame is a timeout, not a parse
// error in half a document.
return fmt.Errorf("could not read %s: %w", what, err)
}
// The limit is on the body, not the frame, so the terminator does not count
// against it. A body that exceeds max is cut off by the LimitReader before its
// newline, so a missing terminator is also what an oversized line looks like.
body := bytes.TrimSuffix(line, []byte{'\n'})
if len(body) > max {
return &frameError{fmt.Errorf("%s exceeded %d bytes", what, max)}
}
dec := json.NewDecoder(bytes.NewReader(body))
if strict {
dec.DisallowUnknownFields()
}
if err := dec.Decode(v); err != nil {
return &frameError{fmt.Errorf("could not parse %s: %w", what, err)}
}
// Strict is one object and nothing after it: a second object, or stray bytes, on
// the line would be read as part of a request that happened to decode, and is
// refused the way an unknown field is.
if strict {
if _, err := dec.Token(); err != io.EOF {
return fmt.Errorf("could not parse %s: unexpected data after the object", what)
}
}
return nil
}
// writeFrame writes v to w as one frame: the JSON object and the newline that ends it.
func writeFrame(w io.Writer, v any) error {
body, err := json.Marshal(v)
if err != nil {
return fmt.Errorf("could not encode frame: %w", err)
}
_, err = w.Write(append(body, '\n'))
return err
}
+102
View File
@@ -0,0 +1,102 @@
package localapi
import (
"bytes"
"fmt"
"io"
"net"
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestReadFrameLimitIsOnTheBody pins the framing rule at its edge: the limit bounds the
// JSON body, so the newline that terminates the frame must not count against it.
func TestReadFrameLimitIsOnTheBody(t *testing.T) {
// A JSON string whose encoded form is exactly n bytes, with or without a terminator.
body := func(n int) string {
return fmt.Sprintf(`"%s"`, strings.Repeat("a", n-2))
}
var v string
// Exactly at the limit, terminated: accepted.
v = ""
require.NoError(t, readFrame(bytes.NewReader([]byte(body(64)+"\n")), 64, "request", &v, true))
assert.Len(t, v, 62)
// Exactly at the limit, EOF instead of a newline: still accepted.
v = ""
require.NoError(t, readFrame(bytes.NewReader([]byte(body(64))), 64, "request", &v, true))
assert.Len(t, v, 62)
// One byte over: refused as oversized, not as a parse failure.
err := readFrame(bytes.NewReader([]byte(body(65)+"\n")), 64, "request", &v, true)
require.Error(t, err)
assert.Contains(t, err.Error(), "exceeded 64 bytes")
}
// stallingReader hands out its bytes and then fails with err, the shape of a peer that
// sends part of a frame and then stops, until the deadline on the connection lands.
type stallingReader struct {
data []byte
err error
}
func (r *stallingReader) Read(p []byte) (int, error) {
if len(r.data) == 0 {
return 0, r.err
}
n := copy(p, r.data)
r.data = r.data[n:]
return n, nil
}
type timeoutError struct{}
func (timeoutError) Error() string { return "i/o timeout" }
func (timeoutError) Timeout() bool { return true }
func (timeoutError) Temporary() bool { return true }
// TestReadFramePartialLineThenReadErrorIsTheError: a read that fails after part of a
// frame has arrived is a failed read, and the caller sees that error. The old behaviour
// treated the fragment as the whole frame and reported a parse failure, so a client
// that stalled mid request was told its JSON was malformed, and a daemon that stalled
// mid response hid the timeout the client's error path looks for.
func TestReadFramePartialLineThenReadErrorIsTheError(t *testing.T) {
var v map[string]any
err := readFrame(&stallingReader{data: []byte(`{"command": "hel`), err: timeoutError{}}, 1024, "request", &v, true)
require.Error(t, err)
var ne net.Error
require.ErrorAs(t, err, &ne, "the timeout is still visible through the wrapping: %v", err)
assert.True(t, ne.Timeout())
assert.NotContains(t, err.Error(), "parse", "a stalled peer is not a parse failure")
// EOF after an unterminated line is the one exception: a final line with no newline
// is still a frame.
v = nil
require.NoError(t, readFrame(&stallingReader{data: []byte(`{"command": "help"}`), err: io.EOF}, 1024, "request", &v, true))
assert.Equal(t, "help", v["command"])
// And an unterminated line that is EOF only because the limit cut it off is still
// reported as oversized.
err = readFrame(bytes.NewReader([]byte(strings.Repeat("a", 70)+"\n")), 64, "request", &v, true)
require.Error(t, err)
assert.Contains(t, err.Error(), "exceeded 64 bytes")
}
// TestReadFrameUnknownFieldIsRefusedWhenStrict: the daemon refuses a request field it
// does not know, as it refuses an unknown parameter, so a misspelling is reported rather
// than ignored; the client, reading leniently, passes over a field a newer daemon sends.
func TestReadFrameUnknownFieldIsRefusedWhenStrict(t *testing.T) {
var req Request
err := readFrame(bytes.NewReader([]byte(`{"version":1,"comand":"help"}`+"\n")), 1024, "request", &req, true)
require.ErrorContains(t, err, `unknown field "comand"`)
var res Response
require.NoError(t, readFrame(bytes.NewReader([]byte(`{"version":1,"later":true}`+"\n")), 1024, "response", &res, false))
assert.Equal(t, 1, res.Version)
}
+570
View File
@@ -0,0 +1,570 @@
package localapi
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"math"
"net"
"runtime/debug"
"strconv"
"sync"
"sync/atomic"
"time"
"github.com/slackhq/nebula/commands"
)
const (
// defaultMaxConns caps concurrent connections. Exceeding it is rejected rather than
// queued: a queue turns a load problem into a latency problem and hides it.
defaultMaxConns = 8
// readTimeout bounds how long a client may take to send its request line.
readTimeout = 10 * time.Second
// writeTimeout bounds a single write. A client that stops reading is dropped rather
// than waited on, so a paused terminal cannot pin daemon resources.
writeTimeout = 30 * time.Second
// defaultSlowAfter is how long a command may run before the daemon says so. A
// command holds a connection slot until it returns, and one that never returns
// holds it for good; after enough of those the socket answers only "busy". The
// warning is what tells an operator that from load, and names the command.
defaultSlowAfter = 30 * time.Second
)
// Server serves a command registry over a local socket to the user who ran nebula.
type Server struct {
l *slog.Logger
src *commands.Registry
maxConns int
slowAfter time.Duration
mu sync.Mutex
ln net.Listener
path string
running bool
unwatch func() bool
sem chan struct{}
euid uint32
// access is what the socket at path was bound with, so a reload that changes only
// the group or the mode is applied without moving the socket.
access Access
// trustSocket is set while a group shares the socket. The kernel admitted the peer
// through the socket's mode and group, which is the whole access control then, and
// the peer check in serve stands aside: on unix it sees the peer's uid and primary
// gid only, so it could not tell a member of the group from anyone else.
trustSocket atomic.Bool
}
// NewServer returns a Server that will serve every command in reg, including one
// registered on it after the server is built.
func NewServer(l *slog.Logger, reg *commands.Registry) *Server {
return &Server{
l: l,
src: reg,
maxConns: defaultMaxConns,
slowAfter: defaultSlowAfter,
// The same identity the client's owner check is made for, so the peer check
// cannot disagree with it, and so a test can stand in for it the same way.
euid: currentUID(),
}
}
// Rebind brings the server up at path, moves a running server there, and serves it until
// Stop or ctx cancellation. It is the one way the server comes up, so Wire's first
// attempt and its reload callback are the same call, and a server already serving path
// is left as it is: Wire registers the callback before the first attempt, so a HUP that
// lands between the two has the callback do the first bind, and the first attempt then
// finds the socket it was about to create already up.
//
// The new socket is bound before the old one is given up: a bad path in a reloaded
// config leaves the operator with the endpoint they already had rather than none.
//
// s.mu is held for the whole operation, bind included. Releasing it around listen would
// let a Stop land in between, and this would then install a fresh listener on a server
// that had just shut down, leaving a live socket on disk until the process exits. listen
// does not take s.mu, so holding it costs the other methods one bind's wait, at most the
// stale socket probe, once per reload.
//
// An error here is never fatal to the caller: nebula keeps forwarding packets without a
// control endpoint, and the caller is expected to log and carry on.
func (s *Server) Rebind(ctx context.Context, path string, acc Access) error {
s.mu.Lock()
defer s.mu.Unlock()
// Set before the access is applied, not after, so no connection the new access
// admits is checked against the old setting. Early is safe either way: a member
// arriving before a group's mode is set is refused by the kernel, and with a group
// being taken away the peer check applies a moment before the mode tightens.
prev := s.trustSocket.Swap(acc.shared())
err := s.rebindLocked(ctx, path, acc)
if err != nil {
// Whatever was serving still is, with the access it had.
s.trustSocket.Store(prev)
}
return err
}
func (s *Server) rebindLocked(ctx context.Context, path string, acc Access) error {
reopening := false
if s.running && s.path == path {
if s.access == acc {
return nil
}
// Only the access changed. On unix it is set on the socket in place; on windows
// the descriptor is fixed when the pipe is created, so the pipe is created again.
// The name cannot be held twice, so the old one goes first, and a bind that then
// fails leaves the server down rather than on the access the operator asked to
// change; that is reported.
err := applyAccess(path, acc)
if err == nil {
s.access = acc
s.l.Info("Changed the local control socket's access", "path", path, "access", acc.String())
return nil
}
if !errors.Is(err, errReopenForAccess) {
return fmt.Errorf("could not change the access of %s: %w", path, err)
}
reopening = true
release(s.ln, s.path)
s.ln, s.path, s.running, s.access = nil, "", false, Access{}
}
ln, err := listen(path, acc)
if err != nil {
if reopening {
return fmt.Errorf("the local control socket was closed to change its access and could not be created again: %w", err)
}
return err
}
// A reload that arrives during shutdown must not bring the endpoint back. Stop is
// not terminal (Rebind after Stop is allowed, and is how a server whose first bind
// failed comes up), so running is no guide; the context is what shutdown cancels,
// and it is checked here, while nothing else can change the server's state.
if err := ctx.Err(); err != nil {
release(ln, path)
return fmt.Errorf("not starting the local control socket, nebula is shutting down: %w", err)
}
oldLn, oldPath := s.ln, s.path
s.ln = ln
s.path = path
s.access = acc
s.running = true
sem := s.semaphore()
s.watchLocked(ctx)
// Stop accepting on the old listener. In flight commands keep their connections,
// which stay usable until they finish. Its accept loop exits on the closed-listener
// error, so it is gone as soon as this returns. oldPath differs from path whenever
// oldLn is set: a running server on the same path returned above.
if oldLn != nil {
release(oldLn, oldPath)
}
s.l.Info("Local control socket listening", "path", path, "access", acc.String())
go func() {
s.accept(ln, sem)
}()
return nil
}
// watchLocked stops the server when ctx is cancelled. One watcher covers a run of the
// server, from the first bind that succeeds until Stop: a daemon whose initial listen
// failed and which later Rebinds onto a good path from a reloaded config still needs its
// socket removed on shutdown, and a Rebind on a running server must not stack a second
// watcher on the same ctx. Stop detaches the watcher, so a server started again
// afterwards is watched on the ctx that started it, and the ctx of an earlier run no
// longer reaches it whenever that one ends. Callers hold s.mu.
func (s *Server) watchLocked(ctx context.Context) {
if s.unwatch != nil {
return
}
s.unwatch = context.AfterFunc(ctx, s.Stop)
}
// Path returns the socket path currently being served, or "" when stopped.
func (s *Server) Path() string {
s.mu.Lock()
defer s.mu.Unlock()
return s.path
}
// Stop closes the listener and removes the socket. In flight commands are allowed to
// finish; Stop does not wait for them.
//
// The whole of it runs under s.mu, listener and socket included, so a Rebind cannot land
// between the state change and the close and find a path that is still bound by the
// listener it is replacing.
func (s *Server) Stop() {
s.mu.Lock()
defer s.mu.Unlock()
if !s.running {
return
}
release(s.ln, s.path)
s.running = false
s.access = Access{}
s.ln = nil
s.path = ""
// Set whenever running is. When this Stop is the watcher's own, stopping a watcher
// that has already fired is a no-op.
s.unwatch()
s.unwatch = nil
}
// release gives up a listener and the socket node at its path. The node goes first,
// while the listener is still open: for as long as it is, the node is this server's,
// since a starting daemon's stale check refuses to remove a socket that answers. The
// other order opens a window between the close and the unlink in which a new daemon can
// find the node stale, clear it and bind its own at the same path, which the unlink
// then takes from under it, leaving that daemon serving a socket nothing can reach.
// listen turns off the listener's own unlink on close so this is the only unlink.
func release(ln net.Listener, path string) {
removeSocket(path)
_ = ln.Close()
}
// semaphore returns the connection limiter, creating it on first use. It is made once
// and never replaced: a handler releases into the channel it acquired from, so swapping
// in a fresh one across a Stop and Start would leave any handler still in flight
// blocked forever on a channel nothing had filled. Callers hold s.mu.
func (s *Server) semaphore() chan struct{} {
if s.sem == nil {
s.sem = make(chan struct{}, s.maxConns)
}
return s.sem
}
// Accept failures that are not the listener closing are retried after a pause that
// doubles from acceptRetryMin up to acceptRetryMax, the same shape as net/http.Serve.
const (
acceptRetryMin = 5 * time.Millisecond
acceptRetryMax = time.Second
)
// accept serves connections from ln until it is closed. sem is the limiter to acquire
// from and release into; it is passed rather than read from s so the handlers spawned
// here never touch s.sem without the lock.
//
// Only a closed listener ends the loop; Stop and Rebind are the two things that close
// it. Every other error is retried: the process being out of descriptors (EMFILE,
// ENFILE) or a client that hung up between the kernel's queue and us (ECONNABORTED)
// are passing conditions, and giving up on them would leave a socket on disk that
// nothing answers, with the reason buried in a log line from minutes or days earlier.
func (s *Server) accept(ln net.Listener, sem chan struct{}) {
var delay time.Duration
for {
conn, err := ln.Accept()
if err != nil {
if errors.Is(err, net.ErrClosed) {
return
}
if ne, ok := err.(net.Error); ok && ne.Timeout() {
continue
}
if delay == 0 {
delay = acceptRetryMin
} else {
delay = min(delay*2, acceptRetryMax)
}
s.l.Warn("Local control socket could not accept a connection, retrying", "error", err, "retry_in", delay)
time.Sleep(delay)
continue
}
delay = 0
select {
case sem <- struct{}{}:
default:
// At capacity. Say so and hang up rather than holding the connection open.
// Logged too: a slot is held for as long as a command runs, and the slow
// command warning in serve says which ones are still running.
s.l.Warn("Local control socket is at capacity, refusing a connection", "max", s.maxConns)
s.reply(conn, &Response{Version: ProtocolVersion, Error: "local control socket is busy, try again"})
_ = conn.Close()
continue
}
go func() {
defer func() { <-sem }()
s.handle(conn)
}()
}
}
// handle serves one request. It recovers from panics in command callbacks: this
// surface is present on every install and reachable by any local process running as
// the daemon's user, so a malformed argument must cost a connection rather than the
// overlay.
func (s *Server) handle(conn net.Conn) {
defer conn.Close()
var args []string
defer func() {
if r := recover(); r != nil {
s.l.Error("Recovered from panic serving a local control command",
"panic", r,
"args", args,
"stack", string(debug.Stack()),
)
s.reply(conn, &Response{Version: ProtocolVersion, Error: "command failed"})
}
}()
// Where the platform can identify the caller, check it, unless a group shares the
// socket; see trustSocket. Otherwise listen's mode or security descriptor stands.
if peerCheckSupported && !s.trustSocket.Load() {
uid, err := peerUID(conn)
if err != nil {
s.l.Warn("Refusing local control connection with unreadable credentials", "error", err)
return
}
// Access is restricted to the user who ran nebula. Root is allowed through
// because it can reach the daemon's memory regardless of what this check says.
if uid != s.euid && uid != 0 {
s.l.Warn("Refusing local control connection from another user", "peerUid", uid)
s.reply(conn, &Response{Version: ProtocolVersion, Error: "not permitted"})
return
}
}
_ = conn.SetReadDeadline(time.Now().Add(readTimeout))
req, err := readRequest(conn)
if err != nil {
s.reply(conn, &Response{Version: ProtocolVersion, Error: err.Error()})
return
}
args = req.Args
// No deadline while a command runs; the write deadline below is what bounds a client
// that stops reading.
_ = conn.SetReadDeadline(time.Time{})
_ = conn.SetWriteDeadline(time.Now().Add(writeTimeout))
s.reply(conn, s.serve(req))
}
// serve runs one request and describes the outcome.
//
// Nothing bounds a command's run; a context would have to reach every Run. What is
// bounded is the silence: a command still running at slowAfter is named in the log, with
// the slot it is holding, so a socket that has gone "busy" can be traced to the commands
// that wedged it. The timer is stopped however the command ends, a panic included, so
// the line is only ever written about a command that is still running.
func (s *Server) serve(req *Request) *Response {
slow := time.AfterFunc(s.slowAfter, func() {
s.l.Warn("Local control command is still running and holding a connection slot",
"command", req.Command,
"args", req.Args,
"after", s.slowAfter,
"max", s.maxConns,
)
})
defer slow.Stop()
res := &Response{Version: ProtocolVersion}
reg := s.src
if req.Describe {
schema, err := json.Marshal(reg.Schema())
if err != nil {
res.Error = err.Error()
return res
}
res.Result = schema
return res
}
if req.Command == "" {
res.Error = "no command given"
return res
}
params, err := stringParams(req.Params)
if err != nil {
res.Error = err.Error()
return res
}
format, err := parseFormat(req.Format)
if err != nil {
res.Error = err.Error()
return res
}
cmd, err := reg.Resolve(commands.Name(req.Command))
if err != nil {
s.l.Debug("Local control command refused", "command", req.Command, "error", err)
res.Error = err.Error()
return res
}
r, err := commands.Invoke(commands.Request{Command: cmd, Params: params, Args: req.Args})
if err != nil {
s.l.Debug("Local control command failed", "command", req.Command, "error", err)
res.Error = err.Error()
return res
}
// A command that acts does so now, before the one reply this surface gives, and the
// reply carries both halves: the line Run promised and the outcome Then produced.
// In JSON only the outcome travels, since the promise is for a person to read.
var preface []byte
if cmd.Then != nil && r.OK {
if format == commands.FormatText && !r.Empty() {
buf := &bytes.Buffer{}
if err := commands.Render(cmd, r, format, buf); err == nil {
preface = buf.Bytes()
}
}
r = cmd.Then()
}
// A command with nothing to say gets an empty, successful reply.
if r.Empty() {
res.Text = preface
return res
}
return render(res, cmd, r, format, preface)
}
// render turns a command's Result into the response, in the format the caller asked for,
// through commands.Render, so the JSON is byte for byte what -json prints over ssh.
func render(res *Response, cmd *commands.Command, r commands.Result, format commands.Format, preface []byte) *Response {
buf := &bytes.Buffer{}
if err := commands.Render(cmd, r, format, buf); err != nil {
res.Error = err.Error()
return res
}
if format != commands.FormatText {
// The caller is a program. Carry the payload as JSON rather than as a string that
// happens to contain some; the newline Render ends a line with is not the value's.
res.Result = bytes.TrimSuffix(buf.Bytes(), []byte("\n"))
// A command that ran and refused is a failure to whoever asked, in either format.
// The structured outcome still travels in Result for a program to read; Error
// carries the reason so the one rule, Error set means failed, holds here too and
// a client never has to guess from the payload.
if !r.OK {
res.Error = r.Reason
}
return res
}
// A command that ran and refused reads as an error to whoever asked for text: they
// are a person or a script, and both expect a refusal on the error stream.
if !r.OK {
res.Error = string(append(preface, buf.Bytes()...))
return res
}
// The bytes the command wrote, as they are; see Response.Text.
res.Text = append(preface, buf.Bytes()...)
return res
}
// parseFormat turns the request's format into the one the registry understands. An
// unknown value is an error rather than a silent default, so a client that asks for
// something this daemon cannot do hears about it.
func parseFormat(s string) (commands.Format, error) {
switch s {
case "", "json":
return commands.FormatJSON, nil
case "text":
return commands.FormatText, nil
default:
return commands.FormatJSON, fmt.Errorf("unknown format %q, expected \"json\" or \"text\"", s)
}
}
// reply writes one response frame.
func (s *Server) reply(conn net.Conn, res *Response) {
_ = conn.SetWriteDeadline(time.Now().Add(writeTimeout))
if err := writeFrame(conn, res); err != nil {
s.l.Debug("Could not send a local control response", "error", err)
}
}
// stringParams converts JSON's types into the strings flag.Value.Set expects, so a
// caller writes true and 3 rather than "true" and "3".
func stringParams(in map[string]any) (map[string]string, error) {
if len(in) == 0 {
return nil, nil
}
out := make(map[string]string, len(in))
for k, v := range in {
switch t := v.(type) {
case bool:
out[k] = strconv.FormatBool(t)
case string:
out[k] = t
case float64:
// encoding/json gives every number as a float64; render whole ones without
// a decimal point so an int flag accepts them.
if t != math.Trunc(t) {
out[k] = strconv.FormatFloat(t, 'f', -1, 64)
continue
}
// A whole float64 can be far outside int64, and Go's conversion of one that is
// gives an unspecified value rather than an error, so the bounds are checked
// first. MinInt64 is exactly representable and is in range. MaxInt64 is not:
// as a float64 it rounds up to 2^63, which is the first value out of range,
// hence the >= on that side.
if t < math.MinInt64 || t >= math.MaxInt64 {
return nil, fmt.Errorf("parameter %q is %s, outside the range of a 64-bit integer",
k, strconv.FormatFloat(t, 'g', -1, 64))
}
out[k] = strconv.FormatInt(int64(t), 10)
case nil:
return nil, fmt.Errorf("parameter %q has no value", k)
default:
return nil, fmt.Errorf("parameter %q is not a string, number or boolean", k)
}
}
return out, nil
}
func readRequest(conn net.Conn) (*Request, error) {
req := &Request{}
if err := readFrame(conn, MaxRequestBytes, "request", req, true); err != nil {
return nil, err
}
if req.Version != ProtocolVersion {
return nil, fmt.Errorf(
"this nebula speaks local control protocol v%d, the client sent v%d; the two binaries are different versions",
ProtocolVersion, req.Version,
)
}
return req, nil
}
+950
View File
@@ -0,0 +1,950 @@
//go:build !windows
package localapi
import (
"bytes"
"context"
"encoding/json"
"errors"
"flag"
"fmt"
"io"
"math"
"net"
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"github.com/slackhq/nebula/commands"
"github.com/slackhq/nebula/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/sys/unix"
)
// testSocket returns a short socket path whose directory does not exist yet and whose
// parent is private: one level below the test's own directory, so that listen has a
// directory of its own to create.
func testSocket(t *testing.T) string {
t.Helper()
return filepath.Join(test.ShortTempDir(t), "n", "s.sock")
}
func serve(t *testing.T, reg *commands.Registry) (*Server, string) {
t.Helper()
path := testSocket(t)
s := NewServer(test.NewLogger(), reg)
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
t.Cleanup(s.Stop)
return s, path
}
// serveAt starts a server on an exact path, for tests that care where the socket is.
func serveAt(t *testing.T, path string) (*Server, func()) {
t.Helper()
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
return s, s.Stop
}
// echoCommand reports its operands, for tests on either surface.
func echoCommand() *commands.Command {
return &commands.Command{
Name: "echo",
ShortDescription: "echoes its arguments",
Run: func(fs any, a []string) (commands.Result, error) {
return commands.Value(fmt.Sprintf("%v", a)), nil
},
}
}
func basicRegistry() *commands.Registry {
reg := commands.New()
reg.Register(echoCommand())
reg.Register(&commands.Command{
Name: "boom",
ShortDescription: "panics",
Run: func(fs any, a []string) (commands.Result, error) {
panic("this command is broken")
},
})
return reg
}
// run invokes one command and returns its output and its error text. An empty error is
// success: the protocol has no separate status field.
func run(t *testing.T, path, command string, args ...string) (string, string) {
t.Helper()
res, err := (&Client{Path: path}).Invoke(command, nil, args, "")
require.NoError(t, err)
return textOf(res), res.Error
}
// textOf is what a test reads as a command's output: the text when text was asked for,
// else the JSON result, unquoted when it is a JSON string so a test can compare words.
func textOf(res *Response) string {
if len(res.Text) > 0 {
return string(res.Text)
}
if len(res.Result) > 0 && res.Result[0] == '"' {
var s string
if err := json.Unmarshal(res.Result, &s); err == nil {
return s
}
}
return string(res.Result)
}
func TestServeAllowedCommand(t *testing.T) {
_, path := serve(t, basicRegistry())
out, errOut := run(t, path, "echo", "a", "b")
assert.Equal(t, "[a b]", out)
assert.Empty(t, errOut, "a command that worked reports no error")
}
// TestPanicInCommandDoesNotKillTheDaemon: a broken command costs a connection, not the
// overlay.
func TestPanicInCommandDoesNotKillTheDaemon(t *testing.T) {
_, path := serve(t, basicRegistry())
out, errOut := run(t, path, "boom")
assert.Empty(t, out)
assert.Contains(t, errOut, "command failed")
// The server is still serving, which is the point of the test.
out, errOut = run(t, path, "echo", "still", "here")
assert.Equal(t, "[still here]", out)
assert.Empty(t, errOut)
}
// TestSocketPermissions: the modes are pinned, not left to the umask. A 077 umask would
// make the directory 0700 and hide a root daemon's socket from an unprivileged nebula
// ctl.
func TestSocketPermissions(t *testing.T) {
for _, umask := range []int{0, 0077} {
t.Run(fmt.Sprintf("umask %04o", umask), func(t *testing.T) {
old := unix.Umask(umask)
defer unix.Umask(old)
_, path := serve(t, basicRegistry())
fi, err := os.Stat(path)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0600), fi.Mode().Perm(), "socket must not be reachable by other users")
di, err := os.Stat(filepath.Dir(path))
require.NoError(t, err)
assert.Equal(t, os.FileMode(0755), di.Mode().Perm(), "a directory we create ourselves is searchable, so ServedByRoot can see into it")
})
}
}
func TestLiveSocketIsNotStolen(t *testing.T) {
first, path := serve(t, basicRegistry())
require.NotNil(t, first)
second := NewServer(test.NewLogger(), basicRegistry())
err := second.Rebind(context.Background(), path, ownerOnly())
assert.ErrorContains(t, err, "already in use", "a live endpoint must not be unlinked by a second instance")
}
// TestLiveSocketIsNotStolenBeforeItsChmod: between an instance's bind and its chmod the
// socket is at whatever mode the umask left. A second instance starting in that window
// must still see it as live, not stale: only a refused connect proves nothing is behind
// the file.
func TestLiveSocketIsNotStolenBeforeItsChmod(t *testing.T) {
first, path := serve(t, basicRegistry())
require.NotNil(t, first)
require.NoError(t, os.Chmod(path, 0666))
second := NewServer(test.NewLogger(), basicRegistry())
err := second.Rebind(context.Background(), path, ownerOnly())
require.ErrorContains(t, err, "already in use", "a live socket at any mode is not thereby dead")
_, err = os.Stat(path)
require.NoError(t, err, "the first instance's socket is still there")
require.NoError(t, os.Chmod(path, 0600))
out, _ := run(t, path, "echo", "ok")
assert.Equal(t, "[ok]", out, "and still served")
}
func TestRebindMovesTheSocket(t *testing.T) {
s, path := serve(t, basicRegistry())
next := filepath.Join(filepath.Dir(path), "n.sock")
require.NoError(t, s.Rebind(context.Background(), next, ownerOnly()))
assert.Equal(t, next, s.Path())
out, _ := run(t, next, "echo", "moved")
assert.Equal(t, "[moved]", out)
_, err := os.Stat(path)
assert.True(t, os.IsNotExist(err), "the old socket should be unlinked")
}
func TestRebindFailureKeepsTheOldSocket(t *testing.T) {
s, path := serve(t, basicRegistry())
err := s.Rebind(context.Background(), "/nonexistent-dir-for-tests/x.sock", ownerOnly())
require.Error(t, err)
assert.Equal(t, path, s.Path(), "a bad path must leave the working endpoint alone")
out, _ := run(t, path, "echo", "ok")
assert.Equal(t, "[ok]", out)
}
// TestSemaphoreSurvivesRestart pins the rule that the connection limiter is made once. A
// handler releases into the channel it acquired from, so if Rebind replaced it after a
// Stop, a command still running across the restart would block forever on the new,
// empty channel and leak its goroutine and connection.
func TestSemaphoreSurvivesRestart(t *testing.T) {
s := NewServer(test.NewLogger(), basicRegistry())
path := testSocket(t)
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
first := s.sem
require.NotNil(t, first)
s.Stop()
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
defer s.Stop()
assert.Equal(t, first, s.sem, "Rebind after Stop must keep the same semaphore")
require.NoError(t, s.Rebind(context.Background(), testSocket(t), ownerOnly()))
assert.Equal(t, first, s.sem, "Rebind must keep the same semaphore")
}
// TestRestartIsWatchedOnItsOwnContext: the ctx watcher is installed by the first bind of
// a run and cleared by Stop, so a server stopped by one context's cancellation and
// started again under another is stopped when that second context is cancelled, not
// left serving until the process exits.
func TestRestartIsWatchedOnItsOwnContext(t *testing.T) {
s := NewServer(test.NewLogger(), basicRegistry())
path := testSocket(t)
ctxA, cancelA := context.WithCancel(context.Background())
require.NoError(t, s.Rebind(ctxA, path, ownerOnly()))
cancelA()
require.Eventually(t, func() bool { return s.Path() == "" }, time.Second, 10*time.Millisecond, "the first context must stop the server")
ctxB, cancelB := context.WithCancel(context.Background())
require.NoError(t, s.Rebind(ctxB, path, ownerOnly()))
defer s.Stop()
assert.Equal(t, path, s.Path())
cancelB()
require.Eventually(t, func() bool { return s.Path() == "" }, time.Second, 10*time.Millisecond, "the second context must stop the server too")
_, err := os.Stat(path)
assert.True(t, os.IsNotExist(err), "the socket must be removed when the second context stops the server")
}
// TestStopDetachesTheWatcher: a server stopped by Stop and started again under another
// context belongs to that context alone. The one that started the earlier run, ending
// later, must not take the new run down with it.
func TestStopDetachesTheWatcher(t *testing.T) {
s := NewServer(test.NewLogger(), basicRegistry())
path := testSocket(t)
ctxA, cancelA := context.WithCancel(context.Background())
t.Cleanup(cancelA)
require.NoError(t, s.Rebind(ctxA, path, ownerOnly()))
s.Stop()
ctxB, cancelB := context.WithCancel(context.Background())
t.Cleanup(cancelB)
require.NoError(t, s.Rebind(ctxB, path, ownerOnly()))
t.Cleanup(s.Stop)
cancelA()
time.Sleep(50 * time.Millisecond)
assert.Equal(t, path, s.Path(), "the first run's context has nothing to do with the second run")
_, err := (&Client{Path: path}).Describe()
require.NoError(t, err)
cancelB()
require.Eventually(t, func() bool { return s.Path() == "" }, time.Second, 10*time.Millisecond, "the second run's context stops it")
}
func TestVersionMismatchIsExplained(t *testing.T) {
_, path := serve(t, basicRegistry())
conn, err := net.Dial("unix", path)
require.NoError(t, err)
defer conn.Close()
body, _ := json.Marshal(Request{Version: ProtocolVersion + 1, Command: "echo"})
_, err = conn.Write(append(body, '\n'))
require.NoError(t, err)
res, err := readResponse(conn)
require.NoError(t, err)
assert.Contains(t, res.Error, "different versions")
}
func TestOversizedRequestIsRejected(t *testing.T) {
_, path := serve(t, basicRegistry())
conn, err := net.Dial("unix", path)
require.NoError(t, err)
defer conn.Close()
_, err = conn.Write(append(bytes.Repeat([]byte("a"), MaxRequestBytes+10), '\n'))
if err == nil {
res, rerr := readResponse(conn)
require.NoError(t, rerr)
assert.NotEmpty(t, res.Error)
}
}
// TestConcurrentClientsAreBounded checks that going over the connection cap is answered
// rather than queued, so a burst of clients cannot pile up against the packet path.
// TestSlowCommandIsNamedInTheLog: a command that does not return holds its slot for
// good, and enough of them leave the socket answering only "busy". Both are logged, the
// command by name once it has run past slowAfter and each refused connection as it
// happens, so a busy socket can be traced to what wedged it.
func TestSlowCommandIsNamedInTheLog(t *testing.T) {
reg := commands.New()
release := make(chan struct{})
reg.Register(&commands.Command{
Name: "block",
ShortDescription: "blocks until released",
Run: func(fs any, a []string) (commands.Result, error) {
<-release
return commands.Value("done"), nil
},
})
var logs lockedBuffer
s := NewServer(test.NewLoggerWithOutput(&logs), reg)
s.maxConns = 1
s.slowAfter = 20 * time.Millisecond
path := testSocket(t)
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
defer s.Stop()
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
_, _ = (&Client{Path: path}).Invoke("block", nil, []string{"x"}, "")
}()
require.Eventually(t, func() bool {
return strings.Contains(logs.String(), "still running")
}, 2*time.Second, 10*time.Millisecond)
assert.Contains(t, logs.String(), "command=block")
assert.Contains(t, logs.String(), "args=[x]")
res, err := (&Client{Path: path}).Invoke("block", nil, nil, "")
require.NoError(t, err)
assert.Contains(t, res.Error, "busy")
assert.Contains(t, logs.String(), "at capacity")
close(release)
wg.Wait()
}
// TestPanickedCommandIsNotReportedAsStillRunning: the slow command line is about a slot
// that is still held. A command that panicked has given its slot back, and the client
// its answer, so no line must claim otherwise after the fact.
func TestPanickedCommandIsNotReportedAsStillRunning(t *testing.T) {
reg := commands.New()
reg.Register(&commands.Command{
Name: "boom",
ShortDescription: "panics",
Run: func(fs any, a []string) (commands.Result, error) {
panic("this command is broken")
},
})
var logs lockedBuffer
s := NewServer(test.NewLoggerWithOutput(&logs), reg)
s.slowAfter = 20 * time.Millisecond
path := testSocket(t)
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
defer s.Stop()
res, err := (&Client{Path: path}).Invoke("boom", nil, nil, "")
require.NoError(t, err)
assert.Equal(t, "command failed", res.Error)
time.Sleep(100 * time.Millisecond)
assert.Contains(t, logs.String(), "Recovered from panic")
assert.NotContains(t, logs.String(), "still running")
}
// lockedBuffer is a strings.Builder the server's goroutines and the test can share.
type lockedBuffer struct {
mu sync.Mutex
b strings.Builder
}
func (l *lockedBuffer) Write(p []byte) (int, error) {
l.mu.Lock()
defer l.mu.Unlock()
return l.b.Write(p)
}
func (l *lockedBuffer) String() string {
l.mu.Lock()
defer l.mu.Unlock()
return l.b.String()
}
// flakyListener fails its first few Accepts the way a process that is out of file
// descriptors does, then behaves.
type flakyListener struct {
net.Listener
failures int
}
func (f *flakyListener) Accept() (net.Conn, error) {
if f.failures > 0 {
f.failures--
return nil, &net.OpError{Op: "accept", Net: "unix", Err: errors.New("too many open files")}
}
return f.Listener.Accept()
}
// TestAcceptSurvivesTransientErrors: an Accept that fails for a reason other than the
// listener being closed is retried, so a moment at the descriptor limit does not take
// the control socket down for the rest of the process. Closing the listener is still
// what ends the loop.
func TestAcceptSurvivesTransientErrors(t *testing.T) {
path := testSocket(t)
require.NoError(t, os.Mkdir(filepath.Dir(path), 0700))
ln, err := listen(path, ownerOnly())
require.NoError(t, err)
s := NewServer(test.NewLogger(), basicRegistry())
done := make(chan struct{})
go func() {
defer close(done)
s.accept(&flakyListener{Listener: ln, failures: 3}, make(chan struct{}, 1))
}()
out, errText := run(t, path, "echo", "still", "here")
assert.Empty(t, errText)
assert.Equal(t, "[still here]", out)
require.NoError(t, ln.Close())
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("accept loop did not exit when the listener was closed")
}
}
func TestStopRemovesTheSocket(t *testing.T) {
s, path := serve(t, basicRegistry())
s.Stop()
_, err := os.Stat(path)
assert.True(t, os.IsNotExist(err))
}
func TestContextCancellationStops(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
path := testSocket(t)
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(ctx, path, ownerOnly()))
cancel()
assert.Eventually(t, func() bool {
_, err := os.Stat(path)
return os.IsNotExist(err)
}, 2*time.Second, 20*time.Millisecond)
}
// TestNamedParametersDriveFlags is the point of the protocol: a caller passes options by
// name, with JSON's own types, and never assembles a command line.
// TestStringParamsNumbers: JSON numbers arrive as float64. Whole ones render as integers
// so int flags accept them, and a whole one outside int64 is refused rather than
// converted, since Go's float-to-int conversion of an out-of-range value is unspecified
// and would set a number the caller never sent.
func TestStringParamsNumbers(t *testing.T) {
cases := map[string]struct {
in float64
want string
wantErr bool
}{
"whole": {in: 3, want: "3"},
"negative whole": {in: -1, want: "-1"},
"fraction": {in: 1.5, want: "1.5"},
"min int64": {in: math.MinInt64, want: "-9223372036854775808"},
"largest in range": {in: 9223372036854774784, want: "9223372036854774784"},
"2^63": {in: 9223372036854775808, wantErr: true},
"1e19": {in: 1e19, wantErr: true},
"below min int64": {in: -9.3e18, wantErr: true},
}
for name, tc := range cases {
t.Run(name, func(t *testing.T) {
out, err := stringParams(map[string]any{"n": tc.in})
if tc.wantErr {
require.Error(t, err)
require.ErrorContains(t, err, `parameter "n" is `)
assert.ErrorContains(t, err, "outside the range of a 64-bit integer")
return
}
require.NoError(t, err)
assert.Equal(t, tc.want, out["n"])
})
}
}
func TestNamedParametersDriveFlags(t *testing.T) {
reg := commands.New()
reg.Register(&commands.Command{
Name: "report",
ShortDescription: "reports its flags",
Flags: func() (*flag.FlagSet, any) {
fl := flag.NewFlagSet("", flag.ContinueOnError)
s := struct {
Verbose bool
Label string
}{}
fl.BoolVar(&s.Verbose, "verbose", false, "say more")
fl.StringVar(&s.Label, "label", "none", "a label")
return fl, &s
},
Run: func(fs any, a []string) (commands.Result, error) {
s := fs.(*struct {
Verbose bool
Label string
})
return commands.Value(fmt.Sprintf("verbose=%v label=%s args=%v", s.Verbose, s.Label, a)), nil
},
})
_, path := serve(t, reg)
res, err := (&Client{Path: path}).Invoke("report",
map[string]string{"verbose": "true", "label": "here"},
[]string{"an-operand"},
"",
)
require.NoError(t, err)
assert.Empty(t, res.Error)
assert.Equal(t, "verbose=true label=here args=[an-operand]", textOf(res))
}
// TestUnknownParameterIsReported: a misspelled parameter is an error rather than a
// silently applied default, which is the difference between an API and a guess.
func TestUnknownParameterIsReported(t *testing.T) {
reg := commands.New()
reg.Register(&commands.Command{
Name: "report",
ShortDescription: "takes a flag",
Flags: func() (*flag.FlagSet, any) {
fl := flag.NewFlagSet("", flag.ContinueOnError)
fl.Bool("verbose", false, "say more")
return fl, nil
},
Run: func(fs any, a []string) (commands.Result, error) { return commands.Nothing(), nil },
})
_, path := serve(t, reg)
res, err := (&Client{Path: path}).Invoke("report", map[string]string{"verbos": "true"}, nil, "")
require.NoError(t, err)
assert.Contains(t, res.Error, `report has no parameter "verbos"`)
}
func TestParametersOnACommandThatTakesNoneAreReported(t *testing.T) {
_, path := serve(t, basicRegistry())
res, err := (&Client{Path: path}).Invoke("echo", map[string]string{"nope": "true"}, nil, "")
require.NoError(t, err)
assert.Contains(t, res.Error, "echo takes no parameters")
}
// TestJSONOutputIsAJSONValue: a command that emits JSON lands in the response as JSON,
// not as a string a caller has to parse a second time.
func TestJSONOutputIsAJSONValue(t *testing.T) {
reg := commands.New()
reg.Register(&commands.Command{
Name: "emit",
ShortDescription: "emits json",
Run: func(fs any, a []string) (commands.Result, error) {
return commands.Value(map[string]any{"hosts": []int{1, 2, 3}}), nil
},
})
_, path := serve(t, reg)
res, err := (&Client{Path: path}).Invoke("emit", nil, nil, "")
require.NoError(t, err)
var decoded struct {
Hosts []int `json:"hosts"`
}
require.NoError(t, json.Unmarshal(res.Result, &decoded))
assert.Equal(t, []int{1, 2, 3}, decoded.Hosts)
}
// TestStringValueIsAJSONString: a command whose value is a string answers a JSON request
// with a JSON string, quotes and all.
func TestStringValueIsAJSONString(t *testing.T) {
_, path := serve(t, basicRegistry())
res, err := (&Client{Path: path}).Invoke("echo", nil, []string{"hi"}, "")
require.NoError(t, err)
var s string
require.NoError(t, json.Unmarshal(res.Result, &s))
assert.Equal(t, "[hi]", s)
}
func TestDescribeListsTheCommands(t *testing.T) {
_, path := serve(t, basicRegistry())
schema, err := (&Client{Path: path}).Describe()
require.NoError(t, err)
names := make([]string, 0, len(schema))
for _, c := range schema {
names = append(names, string(c.Name))
}
assert.ElementsMatch(t, []string{"help", "echo", "boom"}, names)
}
// TestPresentationParametersAreRefused: the schema lists json and pretty on every
// command, because the command lines take them, but a request has one way to choose its
// rendering, the format field. A client that sends them as parameters is pointed there
// rather than given a second, weaker way to say the same thing.
func TestPresentationParametersAreRefused(t *testing.T) {
reg := commands.New()
reg.Register(&commands.Command{
Name: "pair",
ShortDescription: "returns a pair",
Run: func(fs any, a []string) (commands.Result, error) {
return commands.Value(map[string]int{"a": 1}), nil
},
Text: func(v any, w io.Writer) error {
_, err := fmt.Fprintln(w, "a is one")
return err
},
})
_, path := serve(t, reg)
c := &Client{Path: path}
for _, name := range []string{"json", "pretty"} {
res, err := c.Invoke("pair", map[string]string{name: "true"}, nil, "text")
require.NoError(t, err)
assert.Contains(t, res.Error, name+" is a command line flag, not a parameter")
assert.Contains(t, res.Error, "format")
assert.Empty(t, res.Result, "a refused request runs nothing")
}
// The format field is the one that counts.
res, err := c.Invoke("pair", nil, nil, "text")
require.NoError(t, err)
require.Empty(t, res.Error)
assert.Equal(t, "a is one\n", textOf(res))
res, err = c.Invoke("pair", nil, nil, "")
require.NoError(t, err)
require.Empty(t, res.Error)
assert.JSONEq(t, `{"a":1}`, string(res.Result), "the API default is JSON")
}
// TestTextRequestToATextlessCommand: a command with no human rendering still answers a
// text request when the answer is a refusal, since a refusal always has one. Only a
// successful result has nothing to say in text, and that is reported then, by the daemon.
func TestTextRequestToATextlessCommand(t *testing.T) {
reg := commands.New()
reg.Register(&commands.Command{
Name: "tunnel",
ShortDescription: "data on success, a refusal without an operand",
Run: func(fs any, a []string) (commands.Result, error) {
if len(a) == 0 {
return commands.Failed("No vpn address was provided"), nil
}
return commands.Value(map[string]int{"index": 7}), nil
},
JSONOnly: true,
})
_, path := serve(t, reg)
c := &Client{Path: path}
res, err := c.Invoke("tunnel", nil, nil, "text")
require.NoError(t, err)
assert.Equal(t, "No vpn address was provided\n", res.Error, "the refusal is rendered, not the missing text form")
res, err = c.Invoke("tunnel", nil, []string{"10.0.0.1"}, "text")
require.NoError(t, err)
assert.Contains(t, res.Error, "tunnel has no human readable output yet; pass -json or -pretty")
assert.Empty(t, res.Result)
res, err = c.Invoke("tunnel", nil, []string{"10.0.0.1"}, "json")
require.NoError(t, err)
require.Empty(t, res.Error)
assert.JSONEq(t, `{"index":7}`, string(res.Result))
// A refusal asked for as JSON carries the outcome for a program and the reason in
// Error, so a client exits non-zero on the envelope alone, never by sniffing the
// payload for an ok field some unrelated result might also have.
res, err = c.Invoke("tunnel", nil, nil, "json")
require.NoError(t, err)
assert.Equal(t, "No vpn address was provided", res.Error)
assert.JSONEq(t, `{"ok":false,"reason":"No vpn address was provided"}`, string(res.Result))
}
// TestRebindAfterFailedBindStillStopsOnCancel covers a daemon whose initial listen
// failed and which later Rebinds onto a good path from a reloaded config. The context
// watcher has to be installed by that Rebind, or the socket outlives the daemon.
func TestRebindAfterFailedBindStillStopsOnCancel(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
base := test.ShortTempDir(t)
// listen creates one level only, so a socket directory under a parent that is not
// there yet fails to bind, the shape of a runtime directory that does not exist.
s := NewServer(test.NewLogger(), basicRegistry())
require.Error(t, s.Rebind(ctx, filepath.Join(base, "missing", "n", "s.sock"), ownerOnly()))
good := filepath.Join(base, "n", "s.sock")
require.NoError(t, s.Rebind(ctx, good, ownerOnly()))
t.Cleanup(s.Stop)
_, err := (&Client{Path: good}).Describe()
require.NoError(t, err)
cancel()
assert.Eventually(t, func() bool {
_, err := os.Stat(good)
return os.IsNotExist(err)
}, 2*time.Second, 20*time.Millisecond)
}
// TestCommandRegisteredAfterStartIsServed: the server filters the registry per request,
// so a command an embedder registers after wiring the socket is served, and described,
// just as it would be over ssh.
func TestCommandRegisteredAfterStartIsServed(t *testing.T) {
reg := basicRegistry()
_, path := serve(t, reg)
reg.Register(&commands.Command{
Name: "late",
ShortDescription: "registered after the server started",
Run: func(fs any, a []string) (commands.Result, error) {
return commands.Value("late"), nil
},
})
c := &Client{Path: path}
schema, err := c.Describe()
require.NoError(t, err)
names := make([]string, 0, len(schema))
for _, cs := range schema {
names = append(names, string(cs.Name))
}
assert.Contains(t, names, "late")
res, err := c.Invoke("late", nil, nil, "")
require.NoError(t, err)
require.Empty(t, res.Error)
assert.JSONEq(t, `"late"`, string(res.Result))
}
// TestThenIsAnsweredInTheOneReply: this surface answers in one piece, so a command that
// acts does so before the reply, and the reply carries Run's line and Then's outcome in
// text, and the outcome alone in JSON.
func TestThenIsAnsweredInTheOneReply(t *testing.T) {
reg := commands.New()
acted := false
reg.Register(&commands.Command{
Name: "act",
ShortDescription: "replies, then acts",
Run: func(fs any, a []string) (commands.Result, error) {
return commands.Did("about to act"), nil
},
Then: func() commands.Result {
acted = true
return commands.Did("acted")
},
})
_, path := serve(t, reg)
res, err := (&Client{Path: path}).Invoke("act", nil, nil, "text")
require.NoError(t, err)
assert.True(t, acted)
assert.Empty(t, res.Error)
assert.Equal(t, "about to act\nacted\n", string(res.Text))
res, err = (&Client{Path: path}).Invoke("act", nil, nil, "")
require.NoError(t, err)
assert.JSONEq(t, `{"ok":true,"reason":"acted"}`, string(res.Result))
// A refusal from Then still carries Run's line ahead of it in text, as ssh prints it.
acted = false
reg.Register(&commands.Command{
Name: "fail",
ShortDescription: "replies, then fails to act",
Run: func(fs any, a []string) (commands.Result, error) {
return commands.Did("about to act"), nil
},
Then: func() commands.Result {
return commands.Failed("could not act")
},
})
res, err = (&Client{Path: path}).Invoke("fail", nil, nil, "text")
require.NoError(t, err)
assert.Equal(t, "about to act\ncould not act\n", res.Error)
assert.Empty(t, res.Text)
// A Run that refused is the outcome, and Then does not run.
acted = false
reg.Register(&commands.Command{
Name: "refuse",
ShortDescription: "refuses, so never acts",
Run: func(fs any, a []string) (commands.Result, error) {
return commands.Failed("not like that"), nil
},
Then: func() commands.Result {
acted = true
return commands.Did("acted anyway")
},
})
res, err = (&Client{Path: path}).Invoke("refuse", nil, nil, "text")
require.NoError(t, err)
assert.Equal(t, "not like that\n", res.Error)
assert.False(t, acted)
}
// TestRebindRefusesDuringShutdown: a reload that lands after the daemon's context is
// cancelled must not bring the endpoint back. Stop is not terminal, so Rebind has to
// take the answer from the context rather than from the running flag.
func TestRebindRefusesDuringShutdown(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Built by hand rather than through serve: the context watcher is installed once,
// by the first bind, and this test needs it on a context it can cancel.
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(ctx, testSocket(t), ownerOnly()))
t.Cleanup(s.Stop)
cancel()
assert.Eventually(t, func() bool { return s.Path() == "" }, 2*time.Second, 20*time.Millisecond, "the watcher stops the server")
next := testSocket(t)
err := s.Rebind(ctx, next, ownerOnly())
require.Error(t, err)
require.ErrorContains(t, err, "shutting down")
assert.Empty(t, s.Path(), "the server stays stopped")
_, statErr := os.Stat(next)
assert.True(t, os.IsNotExist(statErr), "and the refused socket is not left on disk")
}
// TestRebindAndStopSerialize hammers Rebind against Stop. Whichever runs last decides
// the final state, so after the dust settles a Stop must leave nothing listening and no
// socket file behind, on either path.
func TestRebindAndStopSerialize(t *testing.T) {
a := testSocket(t)
b := testSocket(t)
for i := 0; i < 50; i++ {
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(context.Background(), a, ownerOnly()))
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
_ = s.Rebind(context.Background(), b, ownerOnly())
}()
go func() {
defer wg.Done()
s.Stop()
}()
wg.Wait()
// The Rebind may have run after the Stop, which restarts the server by design.
// A final Stop then has to take everything down.
s.Stop()
assert.Empty(t, s.Path())
for _, p := range []string{a, b} {
_, err := os.Stat(p)
assert.True(t, os.IsNotExist(err), "iteration %d: %s left behind", i, p)
}
}
}
// TestUnknownRequestFieldIsRefused: a request with a field the protocol does not define
// is answered with the field named, the way an unknown parameter is, and not run.
func TestUnknownRequestFieldIsRefused(t *testing.T) {
_, path := serve(t, basicRegistry())
conn, err := dial(path, time.Second)
require.NoError(t, err)
defer conn.Close()
require.NoError(t, writeFrame(conn, map[string]any{"version": ProtocolVersion, "command": "echo", "arguments": []string{"x"}}))
var res Response
require.NoError(t, readFrame(conn, MaxResponseBytes, "response", &res, false))
assert.Contains(t, res.Error, `unknown field "arguments"`)
assert.Empty(t, res.Result)
}
// TestTrailingBytesOnARequestAreRefused: a request line is one object and nothing
// more; a second object or stray bytes after it are refused rather than ignored.
func TestTrailingBytesOnARequestAreRefused(t *testing.T) {
_, path := serve(t, basicRegistry())
for _, line := range []string{
`{"version":1,"command":"echo"} junk` + "\n",
`{"version":1,"command":"echo"}{"version":1,"command":"boom"}` + "\n",
} {
conn, err := dial(path, time.Second)
require.NoError(t, err)
_, err = conn.Write([]byte(line))
require.NoError(t, err)
var res Response
require.NoError(t, readFrame(conn, MaxResponseBytes, "response", &res, false))
assert.Contains(t, res.Error, "unexpected data after the object", line)
conn.Close()
}
}
// TestTextOutputKeepsTheBytes: a text rendering travels as the bytes the command wrote.
// A JSON string would have had to be UTF-8, and a byte that is not would have come out
// as U+FFFD, where the ssh server passes it through.
func TestTextOutputKeepsTheBytes(t *testing.T) {
reg := commands.New()
raw := []byte{'n', 'a', 'm', 'e', ' ', 0xff, 0xfe, '\n'}
reg.Register(&commands.Command{
Name: "raw",
ShortDescription: "writes what a certificate held",
Run: func(fs any, a []string) (commands.Result, error) { return commands.Value("raw"), nil },
Text: func(v any, w io.Writer) error {
_, err := w.Write(raw)
return err
},
})
_, path := serve(t, reg)
res, err := (&Client{Path: path}).Invoke("raw", nil, nil, "text")
require.NoError(t, err)
assert.Empty(t, res.Error)
assert.Equal(t, raw, res.Text)
assert.Empty(t, res.Result, "text is not also carried as JSON")
}
+144
View File
@@ -0,0 +1,144 @@
package localapi
import (
"context"
"errors"
"log/slog"
"github.com/slackhq/nebula/commands"
"github.com/slackhq/nebula/config"
)
// Wire brings up the local control socket for a running nebula, unless the config turns
// it off, and keeps it following the config across reloads.
//
// Every entry point that runs nebula as a daemon calls this, so the endpoint behaves the
// same whether nebula was started from a shell or by a service manager. It deliberately
// does not live in nebula.Main: the endpoint is on by default, so an embedder inside an
// app sandbox would have to know to turn off a socket it cannot use. Wiring it at the
// daemon entry points means embedders get the command registry and no listener.
//
// Every failure below is a warning, bar one. Nebula is expected to keep forwarding
// packets whether or not anyone can reach it; but a name that another process holds is
// not a misconfiguration, it is a process that will be answering nebula ctl in the
// daemon's place, and that is logged as an error.
//
// Call it after Control.Start, so the listener only exists once the interface is up.
func Wire(ctx context.Context, l *slog.Logger, c *config.C, reg *commands.Registry) *Server {
l = l.With("subsystem", "localapi")
srv := NewServer(l, reg)
// Registered before the first attempt, so that a path or bind that is bad at startup
// can be fixed with a config edit and a reload, the same as one that goes bad later,
// and so that a HUP landing in between is harmless: Rebind on a server already at
// the path is a no-op.
reconcile := followControlConfig(ctx, l, c, srv)
reconcile(c)
if !ControlEnabled(c) {
l.Info("Local control socket is disabled", "key", CtlEnabledKey)
}
return srv
}
// followControlConfig registers the reload callback that keeps the socket where
// local_api.control.socket says it is, moving it when the key changes and bringing one
// up while there is none, and returns it for the first attempt to call.
//
// The key is reloadable because the client finds the socket by reading the config: if the
// two could disagree, the file on disk would name a socket that is not there.
//
// Every reload is reconciled rather than gated on the key having changed. Rebind on a
// server already at the path is a no-op, so that costs nothing; and a bind that failed
// for a reason outside the config, a runtime directory that was not there yet, is fixed
// outside the config too, so the reload that follows the fix is a plain one. That holds
// for a move as much as for the first bind: a daemon left on its old path by a move that
// could not bind is one the client, reading the new path from the same file, cannot
// reach, and the operator's fix is the directory and a reload, not an edit to an
// unrelated line to make the key read as changed.
//
// The callback recovers from its own panics rather than relying on the config package,
// whose callback loop has no isolation of its own. A panic there would take the daemon
// down and skip every callback queued behind this one.
func followControlConfig(ctx context.Context, l *slog.Logger, c *config.C, srv *Server) func(*config.C) {
reconcile := func(c *config.C) {
if !ControlEnabled(c) {
if srv.Path() != "" {
srv.Stop()
l.Info("Stopped the local control socket, it is disabled in the reloaded config")
}
return
}
want, err := SocketPathFromConfig(c)
if err != nil {
warnUnusable(l, "local control socket", srv.Path(), err)
return
}
acc, err := controlAccessFrom(c)
if err != nil {
warnUnusable(l, "local control socket", srv.Path(), err)
return
}
old := srv.Path()
err = srv.Rebind(ctx, want, acc)
reportRebind(l, old, want, err, srv.Path())
}
c.RegisterReloadCallback(func(c *config.C, _ *config.Reload) {
defer func() {
if r := recover(); r != nil {
l.Error("Recovered from panic reloading the local control socket", "panic", r)
}
}()
reconcile(c)
})
return reconcile
}
// warnUnusable logs a config the endpoint cannot follow: with nothing serving, the
// endpoint is not available; with something serving, it is kept as it is.
func warnUnusable(l *slog.Logger, what, current string, err error) {
if current == "" {
l.Warn("The "+what+" is not available, its config is unusable", "error", err)
return
}
l.Warn("Ignoring unusable "+what+" config from the reload, keeping the current endpoint", "error", err, "current", current)
}
// reportRebind logs what a Rebind did beyond the listening line the bind itself writes:
// nothing for a server left where it was, a move with both paths, and a failure with
// what the operator is left with. old is the path served before the call, "" for none,
// want the one asked for, and now the path served after it: on unix a failed move keeps
// the old socket, and now is old; on windows a change of access closes the pipe before
// creating it again, and a failure there leaves nothing serving, so now is "".
//
// Nearly every failure is the operator's environment, a missing runtime directory or a
// bad path, and a warning; a name held by another process is an error, since nothing
// else will say that something is sitting where the daemon should be.
func reportRebind(l *slog.Logger, old, want string, err error, now string) {
level := l.Warn
if errors.Is(err, errEndpointHeld) {
level = l.Error
}
switch {
case err == nil:
if old != "" && old != want {
l.Info("Moved the local control socket", "from", old, "to", want)
}
case old == "":
level("Could not start the local control socket, control commands will not be available until a reload binds it", "error", err, "path", want)
default:
level("Could not bind the local control socket at the requested path", "error", err, "requested", want, "serving", now)
}
}
// errEndpointHeld is returned by listen when the endpoint's name is already held by
// some other process, which nebula will not serve alongside: whoever holds it would be
// answering nebula ctl. It is only raised on windows, where the default name can be
// held by another administrator's process and an override outside the protected
// namespace by any account's; on unix a path that is still being served is an earlier
// nebula, which clearStale reports as in use.
var errEndpointHeld = errors.New("the local control socket's name is held by another process")
+175
View File
@@ -0,0 +1,175 @@
//go:build !windows
package localapi
import (
"context"
"fmt"
"os"
"path/filepath"
"strings"
"testing"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestWireRecoversFromBadPathAtStartup covers a socket path that is unusable when nebula
// starts. The operator's fix is a config edit and a reload, so the reload has to be able
// to bring up a socket that never existed, not only move one that does.
func TestWireRecoversFromBadPathAtStartup(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(fmt.Sprintf("local_api:\n control:\n socket: /%s\n", strings.Repeat("x", sunPathMax))))
srv := Wire(ctx, test.NewLogger(), c, basicRegistry())
t.Cleanup(srv.Stop)
assert.Empty(t, srv.Path(), "an over-long path must not produce a socket")
good := testSocket(t)
require.NoError(t, c.ReloadConfigString(fmt.Sprintf("local_api:\n control:\n socket: %s\n", good)).Err)
assert.Equal(t, good, srv.Path())
out, errText := run(t, good, "echo", "hi")
assert.Empty(t, errText)
assert.Equal(t, "[hi]", out)
}
// TestRebindAfterReloadBoundTheSocket covers a HUP that lands inside Wire, after the
// reload callback is registered and before the first attempt runs. The callback does the
// first bind; the attempt must then find the socket it was about to create already up
// and leave it, not fail or rebind it, which would have Wire warn that commands are
// unavailable while they are.
func TestRebindAfterReloadBoundTheSocket(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
path := testSocket(t)
cfg := fmt.Sprintf("local_api:\n control:\n socket: %s\n", path)
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(cfg))
srv := NewServer(test.NewLogger(), basicRegistry())
t.Cleanup(srv.Stop)
// Wire's first half, then the HUP.
followControlConfig(ctx, test.NewLogger(), c, srv)
require.NoError(t, c.ReloadConfigString(cfg).Err)
require.Equal(t, path, srv.Path(), "the reload callback does the first bind")
// Wire's second half.
require.NoError(t, srv.Rebind(ctx, path, ownerOnly()))
assert.Equal(t, path, srv.Path())
out, errText := run(t, path, "echo", "hi")
assert.Empty(t, errText)
assert.Equal(t, "[hi]", out)
}
// TestWireRetriesAFailedMoveOnPlainReload covers a move whose bind fails for a reason
// outside the config: the key names a path under a runtime directory that is not there
// yet. The daemon keeps the old socket, but the client reads the new path from the same
// file and cannot reach it, so the operator creates the directory and reloads. The key
// has not changed since the move was asked for, and the reload still has to make it.
func TestWireRetriesAFailedMoveOnPlainReload(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
section := func(path string) string {
return fmt.Sprintf("local_api:\n control:\n socket: %s\n", path)
}
first := testSocket(t)
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(section(first)))
srv := Wire(ctx, test.NewLogger(), c, basicRegistry())
t.Cleanup(srv.Stop)
require.Equal(t, first, srv.Path())
runtimeDir := filepath.Join(privateBase(t), "run")
moved := filepath.Join(runtimeDir, DirName, "s.sock")
require.NoError(t, c.ReloadConfigString(section(moved)).Err)
assert.Equal(t, first, srv.Path(), "a move that cannot bind keeps the socket it had")
require.NoError(t, os.Mkdir(runtimeDir, 0700))
require.NoError(t, c.ReloadConfigString(section(moved)).Err)
require.False(t, c.HasChanged(CtlSocketKey), "the config is the same; the fix was on disk")
assert.Equal(t, moved, srv.Path())
_, err := os.Stat(first)
assert.True(t, os.IsNotExist(err), "the old socket is given up once the move binds")
out, errText := run(t, moved, "echo", "hi")
assert.Empty(t, errText)
assert.Equal(t, "[hi]", out)
}
// TestWireRecoversOnPlainReloadAfterBadBind covers a path that is fine but a bind that
// is not: the runtime directory is missing when nebula starts. The fix is made outside
// the config, so the config does not change; a plain reload still has to bring the
// socket up.
func TestWireRecoversOnPlainReloadAfterBadBind(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
base := privateBase(t)
runtimeDir := filepath.Join(base, "run")
path := filepath.Join(runtimeDir, DirName, "s.sock")
cfg := fmt.Sprintf("local_api:\n control:\n socket: %s\n", path)
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(cfg))
srv := Wire(ctx, test.NewLogger(), c, basicRegistry())
t.Cleanup(srv.Stop)
assert.Empty(t, srv.Path(), "listen creates one level only, so a missing runtime dir fails the bind")
require.NoError(t, os.Mkdir(runtimeDir, 0700))
require.NoError(t, c.ReloadConfigString(cfg).Err)
require.False(t, c.HasChanged(CtlSocketKey), "the config is the same; the fix was on disk")
assert.Equal(t, path, srv.Path())
out, errText := run(t, path, "echo", "hi")
assert.Empty(t, errText)
assert.Equal(t, "[hi]", out)
// And a reload with a socket up and no change in the key is still a no-op.
require.NoError(t, c.ReloadConfigString(cfg).Err)
assert.Equal(t, path, srv.Path())
out, errText = run(t, path, "echo", "still")
assert.Empty(t, errText)
assert.Equal(t, "[still]", out)
}
// TestWireControlCanBeDisabled: local_api.control.enabled false serves no socket, and
// the key follows reloads both ways.
func TestWireControlCanBeDisabled(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
path := testSocket(t)
section := func(enabled bool) string {
return fmt.Sprintf("local_api:\n control:\n enabled: %t\n socket: %s\n", enabled, path)
}
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(section(false)))
srv := Wire(ctx, test.NewLogger(), c, basicRegistry())
t.Cleanup(srv.Stop)
assert.Empty(t, srv.Path())
_, err := os.Lstat(path)
assert.True(t, os.IsNotExist(err), "no socket is bound")
require.NoError(t, c.ReloadConfigString(section(true)).Err)
assert.Equal(t, path, srv.Path())
out, errText := run(t, path, "echo", "on")
assert.Empty(t, errText)
assert.Equal(t, "[on]", out)
require.NoError(t, c.ReloadConfigString(section(false)).Err)
assert.Empty(t, srv.Path())
_, err = os.Lstat(path)
assert.True(t, os.IsNotExist(err), "the socket is removed")
}