Add the identity command and serve it on an opt-in HTTP info API

Build slackhq/nebula#1821 ("who is this vpn address?" over a local
HTTP+JSON socket) on the command registry, so the lookup is one command
on every surface and the socket is one more local API.

identity
- `identity [-address <vpn addr>[:port]]`: no address reports this node;
  with one, this node or the peer behind an established tunnel
- Certificate fields only: name, vpn addresses, networks, unsafe
  networks, groups, fingerprint, issuer, validity, cert version; a peer's
  fingerprint is the one nebula computed when it verified the cert
- Port accepted and stripped, 4in6 unmapped, so a RemoteAddr passes
  straight through
- No tunnel is a commands.NotFound; an empty -address is refused
- Lighthouse TXT lookup shares the owner and peer lookups, keeps answering
  with the initiating certificate, and no longer panics without a default cert
- Served on ssh and nebula ctl too

Info API
- Off until `local_api.info.enabled` with `socket` and optional quoted
  octal `socket_mode` (default "0600"); nothing derived
- Serves commands declaring Command.Identity plus help; others 403 naming
  their surfaces. TestInfoAPISurface pins the set to identity
- GET /v1 is the schema; GET /v1/<cmd>?<flag>=<value> runs one; body is
  the -json rendering via commands.Render; commands.PresentationFormat
  reads json/pretty from the query the way a line is read
- 400 commands.ParamError, 403 commands.WithheldError (what Resolve now
  returns for a withheld name, so help and dispatch type-check it too),
  404 NotFound or unknown, 500 Run error, refusal with no reason, or
  recovered panic with no detail; no redirects; every miss is JSON
- listen takes a mode: bind under umask 0077 then chmod; "0660" plus a
  searchable dir opens it to a group
- The chmod, at bind and on a mode-only reload, goes through chmodSocket,
  which never follows a link: a user who can write the socket's directory
  could otherwise swap the node for a link and have a root daemon put
  0660 or 0666 on any file. Only a socket we own is chmoded. On linux the
  node is held open O_PATH|O_NOFOLLOW, checked by fstat, and chmoded
  through /proc/self/fd, so the check and the chmod land on one inode;
  fchmodat2 is the fallback without /proc. On darwin and the BSDs it is
  an Lstat check and fchmodat(AT_SYMLINK_NOFOLLOW). clearStale Lstats
  too, so a dangling link is refused rather than followed by bind on
  darwin
- One endpoint type owns bind under mutex, bind-before-release move,
  context watch and stop/unlink; each server supplies what it serves
- Reloadable through the same reconcile shape as the control socket:
  mode applied in place, unparsable section keeps what runs, control
  socket path refused by name
- Wire returns Endpoints{Control, Info} with one Stop
- Windows unsupported for now (a pipe has no mode); warned once

Tests: identity lookups and shapes; API filter, status codes, panic
recovery, socket mode, rebind, cancellation; config parsing incl. bare
0660; Wire transitions; e2e of two nebulas with the info API resolving
the peer by address and port, through shared test.HTTPDo/HTTPGet helpers.
Goldens gain identity.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Matt RichardsonandClaude Fable 5.1 committed 2026-10-21 12:26:09 -04:00
1 parent 92c67a52b2
commit 613c7e3061
38 files changed
+2937 -332

No files matched your search

+45
View File
@@ -0,0 +1,45 @@
//go:build darwin || freebsd || netbsd || openbsd
package localapi
import (
"os"
"golang.org/x/sys/unix"
)
// applyAccess sets the owning group and the mode of the socket node at path, and of
// nothing a link at path names. The node is looked at without following links and must
// be a socket of ours; the chown and the chmod are fchownat and fchmodat with
// AT_SYMLINK_NOFOLLOW, which these kernels honour, so a symlink put at path in the one
// syscall between the look and either call gets its own group or mode changed, which
// means nothing, and not its target's. A socket cannot be opened, and there is no O_PATH
// here to hold the inode across the calls the way the linux side does, so a hard link
// to a file of somebody's choosing swapped in during that one syscall would be changed;
// that needs the file on the socket's filesystem and the right to link to it, which
// darwin and the BSDs grant to any user by default.
func applyAccess(path string, acc Access) error {
fi, err := os.Lstat(path)
if err != nil {
return err
}
if err := checkOwnSocket(path, fi); err != nil {
return err
}
// With no group configured the socket is handed back to the daemon's own group, so a
// reload that drops the key takes the access away; see the linux side.
gid, err := gidFor(acc)
if err != nil {
return err
}
if err := unix.Fchownat(unix.AT_FDCWD, path, -1, gid, unix.AT_SYMLINK_NOFOLLOW); err != nil {
return &os.PathError{Op: "chown", Path: path, Err: err}
}
if err := unix.Fchmodat(unix.AT_FDCWD, path, uint32(acc.Mode.Perm()), unix.AT_SYMLINK_NOFOLLOW); err != nil {
return &os.PathError{Op: "chmod", Path: path, Err: err}
}
return nil
}
+83
View File
@@ -0,0 +1,83 @@
//go:build linux
package localapi
import (
"fmt"
"os"
"strconv"
"golang.org/x/sys/unix"
)
// applyAccess sets the owning group and the mode of the socket node at path, and of
// nothing else. The node is opened O_PATH, which takes a reference to the inode without
// opening the socket itself (a socket cannot be opened, and O_PATH does not try), and
// O_NOFOLLOW, so a symlink at path is a refusal rather than a target. What the
// descriptor holds is checked to be a socket of ours, and the chown and the chmod are
// then made through /proc/self/fd, which names that inode whatever is at path by then:
// a node swapped between the look and the change, for a link or for a hard link to a
// file of somebody's choosing, is not what gets the group or the mode. This is how
// glibc implements fchmodat(AT_SYMLINK_NOFOLLOW) too.
//
// Without /proc the change falls back to fchownat and fchmodat2, which do not follow a
// link but are by path again, so a swap in the one syscall between the check and each
// would be changed. Older than kernel 6.6 with no /proc, the mode is not set and the
// socket is not served.
func applyAccess(path string, acc Access) error {
// The look is for the message: a link or a plain file is named as what it is, which
// the O_PATH open's ELOOP or the fstat below would say less well.
fi, err := os.Lstat(path)
if err != nil {
return err
}
if err := checkOwnSocket(path, fi); err != nil {
return err
}
fd, err := unix.Open(path, unix.O_PATH|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)
if err != nil {
return &os.PathError{Op: "open", Path: path, Err: err}
}
defer unix.Close(fd)
// What the descriptor holds is what the changes land on; the Lstat above looked at
// the path, which may have changed since.
var st unix.Stat_t
if err := unix.Fstat(fd, &st); err != nil {
return &os.PathError{Op: "fstat", Path: path, Err: err}
}
if st.Mode&unix.S_IFMT != unix.S_IFSOCK || st.Uid != currentUID() {
return fmt.Errorf("%s was replaced while its access was being set, refusing to change it", path)
}
// The /proc name of the inode the descriptor holds. ENOENT from it means no /proc.
held := "/proc/self/fd/" + strconv.Itoa(fd)
// The group first: a chown by root clears the setgid bit, which a socket never has,
// but the mode is pinned afterwards either way. With no group configured the socket
// is handed back to the daemon's own, so taking the key out of the config on a reload
// takes the access away as surely as a restart would.
gid, err := gidFor(acc)
if err != nil {
return err
}
err = unix.Chown(held, -1, gid)
if err == unix.ENOENT {
err = unix.Fchownat(unix.AT_FDCWD, path, -1, gid, unix.AT_SYMLINK_NOFOLLOW)
}
if err != nil {
return &os.PathError{Op: "chown", Path: path, Err: err}
}
perm := uint32(acc.Mode.Perm())
err = unix.Chmod(held, perm)
if err == unix.ENOENT {
err = unix.Fchmodat(unix.AT_FDCWD, path, perm, unix.AT_SYMLINK_NOFOLLOW)
}
if err != nil {
return &os.PathError{Op: "chmod", Path: path, Err: err}
}
return nil
}
+207
View File
@@ -0,0 +1,207 @@
package localapi
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
"net"
"sync"
)
// endpoint is the lifecycle the two local APIs share: one socket at a path, bound and
// given up under a mutex, moved by binding the new path before the old one is released,
// and stopped when the context that brought it up ends. What is served on the socket,
// and by what, is the embedding server's: it hands rebindLocked a serve function and
// gets back the thing to close.
type endpoint struct {
l *slog.Logger
// name is what the endpoint is called in its own log lines and errors, so the two
// read apart.
name string
mu sync.Mutex
// closer stops the serving when closed: the listener itself, or the server that owns
// it. It is nil while the endpoint is stopped, which is what stopped means here.
closer io.Closer
path string
access Access
unwatch func() bool
// id identifies the node listen created at path, so a reload can tell whether the
// socket is still there and Stop whether the node it would unlink is still ours.
id socketID
}
// socketID identifies a socket's node on the filesystem, by device and inode on unix.
// The zero value is no identity: a platform with no node to identify, or one that could
// not be read, and removeSocket then unlinks without looking.
type socketID struct {
dev, ino uint64
}
// Path returns the socket path currently being served, or "" when stopped.
func (e *endpoint) Path() string {
e.mu.Lock()
defer e.mu.Unlock()
return e.path
}
// rebindLocked brings the endpoint up at path with acc, moves it there, or, when it is
// already there, applies a changed access. It is the one place a socket is installed and
// the one way an endpoint comes up, so Wire's first attempt and its reload callback are
// the same call, and an endpoint already at 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. serve
// is handed the new listener once it is certain to be kept and returns what to close to
// stop serving it; what becomes of the old socket's connections is that closer's.
//
// The caller holds e.mu for the whole operation, bind included. Releasing it around
// listen would let a Stop land in between, and this would then install a fresh socket on
// an endpoint that had just shut down, leaving a live socket on disk until the process
// exits. listen does not take e.mu, so holding it costs the other methods one bind's
// wait, at most the stale socket probe, once per reload.
func (e *endpoint) rebindLocked(ctx context.Context, path string, acc Access, serve func(net.Listener) io.Closer) error {
reopening := false
if e.closer != nil && e.path == path {
if cur, ok := identifySocket(path); ok && cur == e.id {
if e.access == acc {
return nil
}
err := applyAccess(path, acc)
if err == nil {
e.access = acc
e.l.Info("Changed the "+e.name+"'s socket 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)
}
// The platform fixes access when the endpoint is created, so it is created
// again. The name cannot be held twice, so the old one goes first, and a bind
// that then fails leaves the endpoint down rather than on the old access the
// operator asked to change; that is reported below.
reopening = true
old := e.closer
e.forgetLocked()
_ = old.Close()
} else {
// The node at the path is not the one this endpoint bound: something outside
// nebula removed it, logind clearing /run/user/<uid> on logout or a unit
// cleaning a RuntimeDirectory= it shares, or replaced it. The listener is still
// open and nothing can reach it, so it is given up now and the path is bound
// afresh below; listen's stale check decides whether whatever is there now may
// be cleared. Given up first, rather than after a successful bind as a move
// does, so that a bind that fails leaves the endpoint reported as down, which
// it is, instead of serving a path nothing can reach.
e.l.Warn("The "+e.name+"'s socket is no longer at its path, binding it again", "path", path)
old := e.closer
e.forgetLocked()
_ = old.Close()
}
}
ln, err := listen(path, acc)
if err != nil {
if reopening {
return fmt.Errorf("the %s was closed to change its access and could not be created again: %w", e.name, err)
}
return err
}
id, _ := identifySocket(path)
// A reload that arrives during shutdown must not bring the endpoint back. Stop is
// not terminal (a Rebind after Stop is allowed, and is how an endpoint whose first
// bind failed comes up), so whether it is running is no guide; the context is what
// shutdown cancels, and it is checked here, while nothing else can change the state.
if err := ctx.Err(); err != nil {
release(ln, path, id)
return fmt.Errorf("not starting the %s, nebula is shutting down: %w", e.name, err)
}
old, oldPath, oldID := e.closer, e.path, e.id
e.closer = serve(ln)
e.path = path
e.access = acc
e.id = id
e.watchLocked(ctx)
// Stop serving the old socket. When it was on this same path its node is already
// gone, and the node there now is the new socket's, so only the closer goes.
if old != nil {
if oldPath == path {
_ = old.Close()
} else {
release(old, oldPath, oldID)
}
}
return nil
}
// watchLocked stops the endpoint when ctx is cancelled. One watcher covers a run of the
// endpoint, 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 endpoint must not stack a second
// watcher on the same ctx. Stop detaches the watcher, so an endpoint 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 e.mu.
func (e *endpoint) watchLocked(ctx context.Context) {
if e.unwatch != nil {
return
}
e.unwatch = context.AfterFunc(ctx, e.Stop)
}
// Stop closes the endpoint and removes its socket. Work in flight is the closer's
// business: a listener leaves the connections it already accepted to finish, a server
// cuts them off.
//
// The whole of it runs under e.mu, closer 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
// socket it is replacing.
func (e *endpoint) Stop() {
e.mu.Lock()
defer e.mu.Unlock()
if e.closer == nil {
return
}
release(e.closer, e.path, e.id)
e.forgetLocked()
}
// forgetLocked drops the state of a run once its socket has been given up, the watcher
// with it: one watcher covers one run, and a bind that follows starts a new one on the
// ctx it was given. The watcher is set whenever closer is; when the Stop that got here
// is the watcher's own, stopping one that has already fired is a no-op.
func (e *endpoint) forgetLocked() {
e.closer, e.path, e.access, e.id = nil, "", Access{}, socketID{}
if e.unwatch != nil {
e.unwatch()
e.unwatch = nil
}
}
// release gives up a socket's closer and the node at its path. The node goes first,
// while the socket is still open: for as long as it is, the node is this endpoint'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. id is the
// node listen created, and the only node removeSocket will remove: one that replaced it,
// another daemon's socket bound after ours was taken from under us, is left alone.
func release(c io.Closer, path string, id socketID) {
removeSocket(path, id)
_ = c.Close()
}
+439
View File
@@ -0,0 +1,439 @@
package localapi
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"maps"
"net"
"net/http"
"net/url"
"os"
"path"
"runtime/debug"
"slices"
"strconv"
"time"
"github.com/slackhq/nebula/commands"
"github.com/slackhq/nebula/config"
"golang.org/x/net/netutil"
)
// The info API's config keys. It is off until InfoEnabledKey is true and InfoSocketKey
// names a socket; nothing is derived, because the programs that will dial it have to be
// told where it is anyway.
const (
InfoEnabledKey = "local_api.info.enabled"
InfoSocketKey = "local_api.info.socket"
InfoModeKey = "local_api.info.socket_mode"
)
const (
// defaultInfoMode is the socket's mode when the config names none: the daemon's
// user alone, so that enabling the API does not open it to anyone until the operator
// says who.
defaultInfoMode os.FileMode = 0600
// infoMaxConns bounds the connections held open at once. Beyond it the kernel
// queues, which is what an HTTP client expects of a busy server; the control socket
// refuses instead because its clients are people at a prompt.
infoMaxConns = 64
infoWriteTimeout = 30 * time.Second
infoIdleTimeout = 60 * time.Second
infoMaxHeaderBytes = 64 * 1024
)
// The read deadlines are variables so a test can shorten them. infoReadTimeout bounds
// the whole request, not only its headers: a request is a GET with a query string and
// the handlers never read a body, but net/http drains an unread body after the handler
// returns, and without a deadline on that read a client sending a body a byte at a time
// would hold its connection slot for as long as it liked, and sixty-four of them would
// hold every slot. A body is refused as well, see refuseBody, but net/http drains it
// before it closes, so this deadline is what ends a client that announced a body and
// never sent it.
var (
infoReadHeaderTimeout = 10 * time.Second
infoReadTimeout = 15 * time.Second
)
// errInfoUnsupported is returned by Rebind on a platform without a socket mode to open.
var errInfoUnsupported = errors.New("the info API is not available on this platform")
// servedOnInfo decides what the info API will run: the commands that report identity
// and nothing else, which the commands declare about themselves. Filter adds help to
// the result, so a client can ask what is served; help on a filtered registry describes
// only that registry.
func servedOnInfo(c *commands.Command) bool {
return c.Identity
}
// notServedOnInfo explains a refusal, naming the surfaces that will run the command.
const notServedOnInfo = "the info API serves identity lookups only; the ssh debug server and nebula ctl serve the rest"
// InfoServer serves the identity commands over HTTP, on a socket an operator may open to
// other users on the host.
//
// It is the second local API, and the opposite of the control socket in who it is for.
// The control socket is for the user who runs nebula, serves nearly everything and needs
// no configuration. This one is for every other program on the machine, a web server
// that wants to know who is on the other end of a connection arriving over the overlay,
// say, and it is off until an operator names a socket and a mode for it. What it serves
// is decided by the commands themselves, Command.Identity: the identity a certificate
// asserts and nothing operational, so opening it to a group of services, or to everyone
// on the host, gives away nothing the overlay did not already tell them.
//
// HTTP, so that no client library is needed. GET /v1/<command>?<flag>=<value> runs a
// command with its options as the query string and answers with the command's JSON, the
// same value the control socket and -json produce. GET /v1 describes what is served. A
// refusal is a 4xx with {"error": reason}: 404 when the command said the thing asked
// about does not exist (commands.NotFound), 400 for a parameter or an argument the
// command would not take, 403 for a command this surface withholds.
type InfoServer struct {
endpoint
// reg is the identity commands this surface serves, filtered from the daemon's
// registry when the server was built.
reg *commands.Registry
}
// NewInfoServer returns an InfoServer that serves the identity commands in reg. reg is
// complete when it arrives: nebula registers every command before it wires its local
// APIs, and nothing registers one later, so the filter is taken once here rather than
// kept up to date against the source.
func NewInfoServer(l *slog.Logger, reg *commands.Registry) *InfoServer {
return &InfoServer{
endpoint: endpoint{l: l, name: "local info API"},
reg: reg.Filter(servedOnInfo, notServedOnInfo),
}
}
// Rebind brings the server up at path with mode, moves it there, or, when it is already
// there, applies a changed mode; endpoint.rebindLocked says what that promises. It is
// the one way the server comes up, so Wire's reload callback and its first attempt are
// the same call.
func (s *InfoServer) Rebind(ctx context.Context, path string, acc Access) error {
s.mu.Lock()
defer s.mu.Unlock()
if !infoSupported {
return errInfoUnsupported
}
return s.rebindLocked(ctx, path, acc, func(ln net.Listener) io.Closer {
srv := &http.Server{
Handler: s.handler(),
ReadHeaderTimeout: infoReadHeaderTimeout,
ReadTimeout: infoReadTimeout,
WriteTimeout: infoWriteTimeout,
IdleTimeout: infoIdleTimeout,
MaxHeaderBytes: infoMaxHeaderBytes,
ErrorLog: slog.NewLogLogger(s.l.Handler(), slog.LevelDebug),
}
s.l.Info("Local info API listening", "path", path, "access", acc.String())
go s.serve(srv, ln)
// Closing the server closes the listener Serve was handed, and a Serve that has
// not started yet closes it on its way out, so the node is never left bound by a
// listener nothing is accepting from. Requests in flight are cut off rather than
// waited for: a reload must not hang on a slow client, and the answer is a lookup
// the client can simply repeat.
return srv
})
}
// serve runs the HTTP server on ln until it is closed. Only an error other than the
// close is worth a line: Stop and Rebind are the two things that close it.
func (s *InfoServer) serve(srv *http.Server, ln net.Listener) {
err := srv.Serve(netutil.LimitListener(ln, infoMaxConns))
if err != nil && !errors.Is(err, http.ErrServerClosed) && !errors.Is(err, net.ErrClosed) {
s.l.Error("Local info API stopped serving", "error", err)
}
}
// handler builds the routes. Every answer is JSON, the misses included, so a client
// never has to parse net/http's text for a path or method it did not expect. That
// includes a path ServeMux would clean and redirect, /v1/../v1/x or //v1/x, which is
// refused outright: nothing legitimate sends one, and a redirect is one more thing a
// client would have to follow.
func (s *InfoServer) handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("GET /v1", s.handleDescribe)
mux.HandleFunc("GET /v1/{command}", s.handleCommand)
mux.HandleFunc("/v1", methodNotAllowed)
mux.HandleFunc("/v1/{command}", methodNotAllowed)
mux.HandleFunc("/", notFoundRoute)
return s.recovering(refuseBody(mux))
}
// refuseBody answers a request that carries a body with 400, before any handler sees
// it, and has the connection closed after the reply rather than kept for another
// request. Nothing here reads a body; one is a mistake, and the client is told what to
// do instead. net/http still drains what the body it announced it would send, up to the
// read deadline, so the deadline is what bounds a client that never sends it.
func refuseBody(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.ContentLength != 0 || len(r.TransferEncoding) > 0 {
w.Header().Set("Connection", "close")
writeJSONError(w, http.StatusBadRequest, "the info API takes no request body; put the command's parameters in the query string")
return
}
next.ServeHTTP(w, r)
})
}
// recovering turns a panic in a command into a 500 and a log line rather than a dropped
// connection, which is all net/http would offer, and keeps the daemon up, which is what
// matters: this surface is reachable by whoever the operator opened it to. The handlers
// produce their whole answer before writing any of it, so a panic in Run always finds
// the response unwritten.
func (s *InfoServer) recovering(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
defer func() {
if rec := recover(); rec != nil {
s.l.Error("Recovered from panic serving a local info API request",
"panic", rec,
"path", r.URL.Path,
"query", r.URL.RawQuery,
"stack", string(debug.Stack()),
)
writeJSONError(w, http.StatusInternalServerError, "command failed")
}
}()
if path.Clean(r.URL.Path) != r.URL.Path {
notFoundRoute(w, r)
return
}
next.ServeHTTP(w, r)
})
}
// handleDescribe answers GET /v1 with the schema of every command served here, the same
// document the control socket's describe returns for its own surface.
func (s *InfoServer) handleDescribe(w http.ResponseWriter, _ *http.Request) {
body, err := json.Marshal(s.reg.Schema())
if err != nil {
writeJSONError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, body)
}
// handleCommand answers GET /v1/<command>: the query string carries the command's
// parameters, and the response body is the command's JSON. A request body is never
// read. It is the control socket's serve with HTTP's vocabulary for the outcomes.
func (s *InfoServer) handleCommand(w http.ResponseWriter, r *http.Request) {
name := commands.Name(r.PathValue("command"))
reg := s.reg
cmd, err := reg.Resolve(name)
if err != nil {
// A command this surface withholds is a different answer from one that does not
// exist, and Resolve says which; only the status is picked here.
status := http.StatusNotFound
var withheld *commands.WithheldError
if errors.As(err, &withheld) {
status = http.StatusForbidden
}
writeJSONError(w, status, err.Error())
return
}
// Parsed strictly: the lenient parser on the request drops a pair it cannot decode
// and keeps the rest, and a request that lost its address would be answered as one
// that never had it, which for print-cert is this node's own certificate.
q, err := url.ParseQuery(r.URL.RawQuery)
if err != nil {
writeJSONError(w, http.StatusBadRequest, "malformed query string: "+err.Error())
return
}
params, format, err := queryParams(q, reg.SchemaFor(cmd))
if err != nil {
writeJSONError(w, http.StatusBadRequest, err.Error())
return
}
res, err := commands.Invoke(commands.Request{Command: cmd, Params: params})
if err != nil {
var pe *commands.ParamError
if errors.As(err, &pe) {
writeJSONError(w, http.StatusBadRequest, err.Error())
return
}
// Run failed. The sentence is for the operator's log; the caller, who may be
// anyone the operator admitted, is told that it failed and no more.
s.l.Warn("Local info API command failed", "command", name, "error", err)
writeJSONError(w, http.StatusInternalServerError, "command failed")
return
}
// A command that acts does so before this surface's one reply, whose body is the
// outcome; see Command.Then.
if cmd.Then != nil && res.OK {
res = cmd.Then()
}
// A refusal is answered before an empty Result is: the zero Result is a refusal with
// nothing to say, which exits 1 over ssh and must not read as success here. One with
// no reason is the command's fault, as a Run error is, and is answered the same way.
if !res.OK {
if res.Reason == "" {
s.l.Warn("Local info API command refused without a reason", "command", name)
writeJSONError(w, http.StatusInternalServerError, "command failed")
return
}
status := http.StatusBadRequest
if res.NotFound {
status = http.StatusNotFound
}
writeJSONError(w, status, res.Reason)
return
}
if res.Empty() {
w.WriteHeader(http.StatusNoContent)
return
}
// The same rendering function every surface uses, so the value here is byte for
// byte what -json prints at an ssh prompt.
buf := &bytes.Buffer{}
if err := commands.Render(cmd, res, format, buf); err != nil {
s.l.Warn("Local info API could not render a result", "command", name, "error", err)
writeJSONError(w, http.StatusInternalServerError, "command failed")
return
}
writeJSON(w, http.StatusOK, buf.Bytes())
}
// queryParams reads a query string as a command's parameters. Each key is a flag of the
// command, by name, in the form the control socket takes them; one given twice is
// refused rather than silently picking either. The presentation flags are read as a
// command line reads them, and set apart: the answer is JSON regardless, and pretty
// indents it.
func queryParams(q url.Values, schema commands.CommandSchema) (map[string]string, commands.Format, error) {
var params, presentation map[string]string
for _, k := range slices.Sorted(maps.Keys(q)) {
vs := q[k]
if len(vs) > 1 {
return nil, commands.FormatUnset, fmt.Errorf("parameter %q was given more than once", k)
}
into := &params
if k == commands.FlagJSON || k == commands.FlagPretty {
into = &presentation
}
if *into == nil {
*into = map[string]string{}
}
v := vs[0]
// ?raw is ?raw=true, as -raw is -raw=true on a command line.
if v == "" && isBoolParam(schema, k) {
v = "true"
}
(*into)[k] = v
}
format, err := commands.PresentationFormat(presentation)
if err != nil {
return nil, commands.FormatUnset, err
}
if format == commands.FormatUnset {
format = commands.FormatJSON
}
return params, format, nil
}
// isBoolParam reports whether name is one of the command's boolean flags.
func isBoolParam(schema commands.CommandSchema, name string) bool {
for _, f := range schema.Flags {
if f.Name == name {
return f.Bool
}
}
return false
}
func methodNotAllowed(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Allow", http.MethodGet)
writeJSONError(w, http.StatusMethodNotAllowed, fmt.Sprintf("%s is not allowed; the info API is read only, use GET", r.Method))
}
func notFoundRoute(w http.ResponseWriter, r *http.Request) {
writeJSONError(w, http.StatusNotFound, fmt.Sprintf("no such path %s; GET /v1 lists the commands and GET /v1/<command> runs one", r.URL.Path))
}
func writeJSON(w http.ResponseWriter, status int, body []byte) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set("X-Content-Type-Options", "nosniff")
w.Header().Set("Content-Length", strconv.Itoa(len(body)))
w.WriteHeader(status)
_, _ = w.Write(body)
}
func writeJSONError(w http.ResponseWriter, status int, msg string) {
body, _ := json.Marshal(map[string]string{"error": msg})
writeJSON(w, status, append(body, '\n'))
}
// infoConfig is the info API section of a config, read and checked.
type infoConfig struct {
Enabled bool
Path string
Access Access
}
// infoConfigFrom reads the info API section. A disabled section needs nothing else to be
// right; an enabled one needs a socket path and, if it names a mode, one that parses.
func infoConfigFrom(c *config.C) (infoConfig, error) {
cfg := infoConfig{Enabled: c.GetBool(InfoEnabledKey, false)}
if !cfg.Enabled {
return cfg, nil
}
p := c.GetString(InfoSocketKey, "")
if p == "" {
return cfg, fmt.Errorf("%s is true but %s is not set; the info API needs a socket path", InfoEnabledKey, InfoSocketKey)
}
if err := checkSocketPath(InfoSocketKey, p); err != nil {
return cfg, err
}
cfg.Path = p
mode, err := parseSocketMode(c.Get(InfoModeKey))
if err != nil {
return cfg, fmt.Errorf("%s: %w", InfoModeKey, err)
}
cfg.Access = Access{Mode: mode}
return cfg, nil
}
// parseSocketMode reads an octal mode written as a string. A bare 0660 in yaml is an
// integer, and which integer depends on the parser (octal 432 under YAML 1.1, decimal
// 660 under 1.2), so a number is refused with the fix rather than quietly applied as
// whichever mode its digits happen to spell in octal.
func parseSocketMode(raw any) (os.FileMode, error) {
switch v := raw.(type) {
case nil:
return defaultInfoMode, nil
case string:
mode, err := strconv.ParseUint(v, 8, 32)
if err != nil || mode&^0777 != 0 {
return 0, fmt.Errorf("%q is not an octal file mode such as \"0660\"", v)
}
return os.FileMode(mode), nil
default:
return 0, fmt.Errorf("must be a quoted octal string such as \"0660\", not the number %v; yaml reads a bare mode as a decimal or octal integer depending on the parser", v)
}
}
+513
View File
@@ -0,0 +1,513 @@
//go:build !windows
package localapi
import (
"bufio"
"context"
"encoding/json"
"errors"
"flag"
"fmt"
"io"
"net"
"net/http"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/slackhq/nebula/commands"
"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"
)
type whoamiFlags struct {
Name string
Shout bool
}
// infoRegistry holds one identity command, whoami, whose -name selects an outcome, and
// one command the info API must withhold.
func infoRegistry() *commands.Registry {
reg := commands.New()
reg.Register(&commands.Command{
Name: "whoami",
ShortDescription: "reports who a name belongs to",
Identity: true,
Flags: func() (*flag.FlagSet, any) {
fl := flag.NewFlagSet("", flag.ContinueOnError)
s := &whoamiFlags{}
fl.StringVar(&s.Name, "name", "", "the name to look up")
fl.BoolVar(&s.Shout, "shout", false, "answers in capitals")
return fl, s
},
Run: func(fs any, a []string) (commands.Result, error) {
if f := fs.(*whoamiFlags); f.Shout {
return commands.Value(map[string]any{"name": strings.ToUpper(f.Name)}), nil
}
switch name := fs.(*whoamiFlags).Name; name {
case "":
return commands.Value(map[string]any{"name": "me", "groups": []string{}}), nil
case "nobody":
return commands.NotFound("No active tunnel for vpn addr: %s", name), nil
case "bad":
return commands.Failed("The provided vpn addr could not be parsed: %s", name), nil
case "boom":
panic("this command is broken")
case "err":
return commands.Result{}, errors.New("an internal detail")
case "silent":
return commands.Result{}, nil
case "done":
return commands.Nothing(), nil
default:
return commands.Value(map[string]any{"name": name}), nil
}
},
})
reg.Register(echoCommand())
return reg
}
func serveInfo(t *testing.T, reg *commands.Registry, mode os.FileMode) (*InfoServer, string) {
t.Helper()
path := testSocket(t)
s := NewInfoServer(test.NewLogger(), reg)
require.NoError(t, s.Rebind(context.Background(), path, Access{Mode: mode}))
t.Cleanup(s.Stop)
return s, path
}
// httpDo sends one request over the socket at path and returns the status, body and
// headers. The host in the URL is a placeholder: the dialer ignores it.
func TestInfoServesIdentityCommands(t *testing.T) {
_, path := serveInfo(t, infoRegistry(), 0600)
status, body, hdr := test.HTTPDo(t, path, http.MethodGet, "/v1/whoami?name=alice")
assert.Equal(t, http.StatusOK, status)
assert.JSONEq(t, `{"name":"alice"}`, body)
assert.Equal(t, "application/json", hdr.Get("Content-Type"))
status, body = test.HTTPGet(t, path, "/v1/whoami")
assert.Equal(t, http.StatusOK, status)
assert.JSONEq(t, `{"name":"me","groups":[]}`, body, "a parameter left out is the command's default")
}
// TestInfoWithholdsTheRest checks the filter at the wire: a command that exists but is
// not an identity command is refused with the reason, and a name nothing is registered
// under is unknown, so a caller can tell a typo from a closed door.
func TestInfoWithholdsTheRest(t *testing.T) {
_, path := serveInfo(t, infoRegistry(), 0600)
status, body := test.HTTPGet(t, path, "/v1/echo?x=1")
assert.Equal(t, http.StatusForbidden, status)
assert.Contains(t, body, "echo is not available on this surface")
assert.Contains(t, body, "identity lookups only")
status, body = test.HTTPGet(t, path, "/v1/nope")
assert.Equal(t, http.StatusNotFound, status)
assert.Contains(t, body, `unknown command \"nope\"`)
for _, target := range []string{"/", "/v2/whoami", "/v1/", "/v1/whoami/extra", "//v1/whoami", "/v1/x/../whoami", "/v1/../v1/whoami"} {
status, body = test.HTTPGet(t, path, target)
assert.Equal(t, http.StatusNotFound, status, target)
assert.Contains(t, body, `"error"`, "%s: misses are JSON too", target)
}
}
func TestInfoStatusCodes(t *testing.T) {
_, path := serveInfo(t, infoRegistry(), 0600)
for _, tc := range []struct {
target string
status int
body string
}{
{"/v1/whoami?name=nobody", http.StatusNotFound, `{"error":"No active tunnel for vpn addr: nobody"}`},
{"/v1/whoami?name=bad", http.StatusBadRequest, `{"error":"The provided vpn addr could not be parsed: bad"}`},
{"/v1/whoami?nope=1", http.StatusBadRequest, `{"error":"whoami has no parameter \"nope\""}`},
{"/v1/whoami?name=a&name=b", http.StatusBadRequest, `{"error":"parameter \"name\" was given more than once"}`},
{"/v1/whoami?pretty=maybe", http.StatusBadRequest, `{"error":"pretty: expected a boolean, got \"maybe\""}`},
// A pair that does not decode is a malformed request, not a request without
// the pair: the lenient parser would drop it and keep the rest.
{"/v1/whoami?name=%zz", http.StatusBadRequest, `{"error":"malformed query string: invalid URL escape \"%zz\""}`},
{"/v1/whoami?name=a;shout", http.StatusBadRequest, `{"error":"malformed query string: invalid semicolon separator in query"}`},
// A boolean parameter with no value is set, as -shout is on a command line.
{"/v1/whoami?name=alice&shout", http.StatusOK, `{"name":"ALICE"}`},
{"/v1/whoami?name=alice&shout=false", http.StatusOK, `{"name":"alice"}`},
// Run failed: the caller hears that it did and no more; the detail is for the log.
{"/v1/whoami?name=err", http.StatusInternalServerError, `{"error":"command failed"}`},
// A refusal with no reason, the zero Result, fails the same way rather than
// passing as a success with nothing to say.
{"/v1/whoami?name=silent", http.StatusInternalServerError, `{"error":"command failed"}`},
} {
status, body := test.HTTPGet(t, path, tc.target)
assert.Equal(t, tc.status, status, tc.target)
assert.JSONEq(t, tc.body, body, tc.target)
}
// A success with nothing to say is the one empty reply.
status, body := test.HTTPGet(t, path, "/v1/whoami?name=done")
assert.Equal(t, http.StatusNoContent, status)
assert.Empty(t, body)
status, body, hdr := test.HTTPDo(t, path, http.MethodPost, "/v1/whoami?name=alice")
assert.Equal(t, http.StatusMethodNotAllowed, status)
assert.Equal(t, http.MethodGet, hdr.Get("Allow"))
assert.Contains(t, body, "read only")
status, _, _ = test.HTTPDo(t, path, http.MethodDelete, "/v1")
assert.Equal(t, http.StatusMethodNotAllowed, status)
}
// TestInfoPanicDoesNotKillTheDaemon: a broken command costs a request, not the overlay.
// This surface may be open to anyone on the host, so that holds against them too.
func TestInfoPanicDoesNotKillTheDaemon(t *testing.T) {
_, path := serveInfo(t, infoRegistry(), 0600)
status, body := test.HTTPGet(t, path, "/v1/whoami?name=boom")
assert.Equal(t, http.StatusInternalServerError, status)
assert.JSONEq(t, `{"error":"command failed"}`, body)
status, body = test.HTTPGet(t, path, "/v1/whoami?name=alice")
assert.Equal(t, http.StatusOK, status)
assert.JSONEq(t, `{"name":"alice"}`, body)
}
func TestInfoDescribe(t *testing.T) {
_, path := serveInfo(t, infoRegistry(), 0600)
status, body := test.HTTPGet(t, path, "/v1")
require.Equal(t, http.StatusOK, status)
var schema []commands.CommandSchema
require.NoError(t, json.Unmarshal([]byte(body), &schema))
var names []commands.Name
flags := map[commands.Name][]string{}
for _, c := range schema {
names = append(names, c.Name)
for _, f := range c.Flags {
flags[c.Name] = append(flags[c.Name], f.Name)
}
}
assert.ElementsMatch(t, []commands.Name{"help", "whoami"}, names, "only what is served is described")
assert.Equal(t, []string{"name", "shout", commands.FlagJSON, commands.FlagPretty}, flags["whoami"])
}
// TestInfoPresentationFlags: the query string takes -json and -pretty as a command line
// would. The answer is JSON either way; pretty indents it.
func TestInfoPresentationFlags(t *testing.T) {
_, path := serveInfo(t, infoRegistry(), 0600)
_, compact := test.HTTPGet(t, path, "/v1/whoami?name=alice&json=true")
//nolint:testifylint // the exact bytes, newline included, are what is under test
assert.Equal(t, "{\"name\":\"alice\"}\n", compact)
_, pretty := test.HTTPGet(t, path, "/v1/whoami?name=alice&pretty")
//nolint:testifylint // the layout is what is under test
assert.Equal(t, "{\n \"name\": \"alice\"\n}\n", pretty)
_, off := test.HTTPGet(t, path, "/v1/whoami?name=alice&pretty=false")
assert.Equal(t, compact, off)
}
// TestInfoSocketMode: the mode is the operator's, pinned regardless of the umask, and a
// changed mode on a reload is applied in place rather than by rebinding.
func TestInfoSocketMode(t *testing.T) {
old := unix.Umask(0077)
defer unix.Umask(old)
s, path := serveInfo(t, infoRegistry(), 0660)
fi, err := os.Stat(path)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0660), fi.Mode().Perm())
require.NoError(t, s.Rebind(context.Background(), path, Access{Mode: 0640}))
fi, err = os.Stat(path)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0640), fi.Mode().Perm())
assert.Equal(t, path, s.Path())
status, _ := test.HTTPGet(t, path, "/v1/whoami")
assert.Equal(t, http.StatusOK, status, "the same listener is still serving")
}
// TestInfoModeChangeRefusesASwappedSocket: whoever can write the directory can replace
// the node with a link to a file of their choosing; the daemon must not set the
// operator's mode on that file. The node is seen to be gone, the link is refused by the
// stale check, the file is untouched, and the endpoint reports itself down, since
// nothing of ours is reachable at the path.
func TestInfoModeChangeRefusesASwappedSocket(t *testing.T) {
s, path := serveInfo(t, infoRegistry(), 0600)
target := filepath.Join(filepath.Dir(path), "target")
require.NoError(t, os.WriteFile(target, nil, 0600))
require.NoError(t, os.Remove(path))
require.NoError(t, os.Symlink(target, path))
err := s.Rebind(context.Background(), path, Access{Mode: 0666})
require.ErrorContains(t, err, "is not a socket")
fi, err := os.Stat(target)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0600), fi.Mode().Perm(), "the file behind the link keeps its mode")
assert.Empty(t, s.Path(), "nothing of ours answers at the path, so nothing is served")
}
func TestInfoRebindMovesAndStopRemoves(t *testing.T) {
s, first := serveInfo(t, infoRegistry(), 0600)
second := testSocket(t)
require.NoError(t, s.Rebind(context.Background(), second, Access{Mode: 0600}))
assert.Equal(t, second, s.Path())
_, err := os.Stat(first)
assert.True(t, os.IsNotExist(err), "the old socket is removed")
status, body := test.HTTPGet(t, second, "/v1/whoami?name=alice")
assert.Equal(t, http.StatusOK, status)
assert.JSONEq(t, `{"name":"alice"}`, body)
s.Stop()
assert.Empty(t, s.Path())
_, err = os.Stat(second)
assert.True(t, os.IsNotExist(err), "Stop removes the socket")
}
func TestInfoRebindFailureKeepsTheOldSocket(t *testing.T) {
s, first := serveInfo(t, infoRegistry(), 0600)
err := s.Rebind(context.Background(), filepath.Join(privateBase(t), "missing", "n", "s.sock"), Access{Mode: 0600})
require.Error(t, err)
assert.Equal(t, first, s.Path())
status, _ := test.HTTPGet(t, first, "/v1/whoami")
assert.Equal(t, http.StatusOK, status)
}
func TestInfoContextCancellationStops(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
path := testSocket(t)
s := NewInfoServer(test.NewLogger(), infoRegistry())
require.NoError(t, s.Rebind(ctx, path, Access{Mode: 0600}))
t.Cleanup(s.Stop)
cancel()
require.Eventually(t, func() bool { return s.Path() == "" }, time.Second, 10*time.Millisecond)
_, err := os.Stat(path)
assert.True(t, os.IsNotExist(err))
// And a reload that lands during shutdown must not bring it back.
err = s.Rebind(ctx, path, Access{Mode: 0600})
require.Error(t, err)
assert.ErrorContains(t, err, "shutting down")
}
func TestInfoConfig(t *testing.T) {
load := func(t *testing.T, yaml string) (infoConfig, error) {
t.Helper()
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(yaml))
return infoConfigFrom(c)
}
cfg, err := load(t, "local_api: {}\n")
require.NoError(t, err)
assert.False(t, cfg.Enabled)
cfg, err = load(t, "local_api:\n info:\n socket: /run/nebula/info.sock\n")
require.NoError(t, err)
assert.False(t, cfg.Enabled, "a socket alone does not enable it")
_, err = load(t, "local_api:\n info:\n enabled: true\n")
require.Error(t, err)
require.ErrorContains(t, err, "needs a socket path")
cfg, err = load(t, "local_api:\n info:\n enabled: true\n socket: /run/nebula/info.sock\n")
require.NoError(t, err)
assert.Equal(t, infoConfig{Enabled: true, Path: "/run/nebula/info.sock", Access: Access{Mode: 0600}}, cfg)
cfg, err = load(t, "local_api:\n info:\n enabled: true\n socket: /run/nebula/info.sock\n socket_mode: \"0660\"\n")
require.NoError(t, err)
assert.Equal(t, infoConfig{Enabled: true, Path: "/run/nebula/info.sock", Access: Access{Mode: 0660}}, cfg)
_, err = load(t, "local_api:\n info:\n enabled: true\n socket: run/info.sock\n")
require.Error(t, err, "a relative socket path is refused rather than resolved against anything")
require.ErrorContains(t, err, "must be an absolute path")
require.ErrorContains(t, err, InfoSocketKey)
_, err = load(t, "local_api:\n info:\n enabled: true\n socket: /run/nebula/info.sock\n socket_mode: 0660\n")
require.Error(t, err, "a bare number is whichever integer the yaml parser made of it")
require.ErrorContains(t, err, "quoted octal string")
for _, bad := range []string{`"abc"`, `"1777"`, `"08"`} {
_, err = load(t, "local_api:\n info:\n enabled: true\n socket: /run/nebula/info.sock\n socket_mode: "+bad+"\n")
require.Error(t, err, bad)
require.ErrorContains(t, err, "not an octal file mode", bad)
}
_, err = load(t, fmt.Sprintf("local_api:\n info:\n enabled: true\n socket: /%s\n", strings.Repeat("x", sunPathMax)))
require.Error(t, err)
assert.ErrorContains(t, err, "too long")
}
// TestWireInfoFollowsConfig: off by default, up when enabled, down when disabled, and
// never on the control socket's path.
func TestWireInfoFollowsConfig(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
ctlPath := testSocket(t)
infoPath := testSocket(t)
sectionWithMode := func(enabled bool, socket, mode string) string {
return fmt.Sprintf("local_api:\n control:\n socket: %s\n info:\n enabled: %t\n socket: %s\n socket_mode: \"%s\"\n", ctlPath, enabled, socket, mode)
}
section := func(enabled bool, socket string) string {
return sectionWithMode(enabled, socket, "0600")
}
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(section(false, infoPath)))
var logs lockedBuffer
eps := Wire(ctx, test.NewLoggerWithOutput(&logs), c, infoRegistry())
t.Cleanup(eps.Stop)
assert.Equal(t, ctlPath, eps.Control.Path())
assert.Empty(t, eps.Info.Path(), "off until enabled")
require.NoError(t, c.ReloadConfigString(section(true, infoPath)).Err)
assert.Equal(t, infoPath, eps.Info.Path())
status, body := test.HTTPGet(t, infoPath, "/v1/whoami?name=alice")
assert.Equal(t, http.StatusOK, status)
assert.JSONEq(t, `{"name":"alice"}`, body)
// A reload that changes nothing leaves it where it is.
require.NoError(t, c.ReloadConfigString(section(true, infoPath)).Err)
assert.Equal(t, infoPath, eps.Info.Path())
// A reload that changes only the mode applies it in place.
require.NoError(t, c.ReloadConfigString(sectionWithMode(true, infoPath, "0660")).Err)
assert.Equal(t, infoPath, eps.Info.Path())
fi, err := os.Stat(infoPath)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0660), fi.Mode().Perm())
// A reload that moves the socket moves it, and removes the old one.
moved := testSocket(t)
require.NoError(t, c.ReloadConfigString(section(true, moved)).Err)
assert.Equal(t, moved, eps.Info.Path())
_, err = os.Stat(infoPath)
assert.True(t, os.IsNotExist(err))
status, body = test.HTTPGet(t, moved, "/v1/whoami?name=bob")
assert.Equal(t, http.StatusOK, status)
assert.JSONEq(t, `{"name":"bob"}`, body)
require.NoError(t, c.ReloadConfigString(section(true, infoPath)).Err)
require.Equal(t, infoPath, eps.Info.Path())
// The control socket's path is refused, and the running endpoint is kept, which is
// what the log says.
require.NoError(t, c.ReloadConfigString(section(true, ctlPath)).Err)
assert.Equal(t, infoPath, eps.Info.Path())
assert.Equal(t, ctlPath, eps.Control.Path())
assert.Contains(t, logs.String(), "keeping the current endpoint")
assert.NotContains(t, logs.String(), "Local info API is not available")
require.NoError(t, c.ReloadConfigString(section(false, infoPath)).Err)
assert.Empty(t, eps.Info.Path())
_, err = os.Stat(infoPath)
assert.True(t, os.IsNotExist(err))
// One Stop takes both down.
require.NoError(t, c.ReloadConfigString(section(true, infoPath)).Err)
eps.Stop()
assert.Empty(t, eps.Info.Path())
assert.Empty(t, eps.Control.Path())
}
// TestWireInfoKeepsRunningOnABadReload: a typo in the section must not cut off the
// programs relying on the endpoint; the operator is told and the endpoint stays.
func TestWireInfoKeepsRunningOnABadReload(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
infoPath := testSocket(t)
good := fmt.Sprintf("local_api:\n info:\n enabled: true\n socket: %s\n", infoPath)
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(good))
srv := wireInfo(ctx, test.NewLogger(), c, infoRegistry())
t.Cleanup(srv.Stop)
require.Equal(t, infoPath, srv.Path())
require.NoError(t, c.ReloadConfigString(good+" socket_mode: 0660\n").Err)
assert.Equal(t, infoPath, srv.Path())
status, _ := test.HTTPGet(t, infoPath, "/v1/whoami")
assert.Equal(t, http.StatusOK, status)
}
// rawInfoRequest opens the socket and writes req as it is, so a test can send what an
// HTTP client would not: a body that never arrives. It returns the connection.
func rawInfoRequest(t *testing.T, path, req string) net.Conn {
t.Helper()
conn, err := net.DialTimeout("unix", path, time.Second)
require.NoError(t, err)
t.Cleanup(func() { _ = conn.Close() })
_, err = conn.Write([]byte(req))
require.NoError(t, err)
return conn
}
// TestInfoRefusesARequestBody: a request that announces a body is answered 400 at once,
// before the body arrives, and the connection is closed once the read deadline has ended
// the wait for it, so a client that never sends the body holds its slot for the deadline
// and no longer.
func TestInfoRefusesARequestBody(t *testing.T) {
oldHeader, oldRead := infoReadHeaderTimeout, infoReadTimeout
infoReadHeaderTimeout, infoReadTimeout = 200*time.Millisecond, 300*time.Millisecond
t.Cleanup(func() { infoReadHeaderTimeout, infoReadTimeout = oldHeader, oldRead })
_, path := serveInfo(t, infoRegistry(), 0600)
start := time.Now()
conn := rawInfoRequest(t, path, "GET /v1/whoami HTTP/1.1\r\nHost: nebula\r\nContent-Length: 10\r\n\r\n")
require.NoError(t, conn.SetReadDeadline(time.Now().Add(2*time.Second)))
res, err := http.ReadResponse(bufio.NewReader(conn), nil)
require.NoError(t, err)
body, _ := io.ReadAll(res.Body)
res.Body.Close()
assert.Equal(t, http.StatusBadRequest, res.StatusCode)
assert.Contains(t, string(body), "takes no request body")
assert.True(t, res.Close, "the connection is not kept for another request")
assert.Less(t, time.Since(start), 200*time.Millisecond, "answered before the body, which never comes")
_, err = conn.Read(make([]byte, 1))
require.ErrorIs(t, err, io.EOF, "the server closed the connection")
assert.Less(t, time.Since(start), time.Second, "and did so at the read deadline, not later")
}
// TestInfoDropsASlowClient: a client that sends its headers and then nothing loses its
// connection at the read timeout rather than holding a slot for as long as it likes.
func TestInfoDropsASlowClient(t *testing.T) {
oldHeader, oldRead := infoReadHeaderTimeout, infoReadTimeout
infoReadHeaderTimeout, infoReadTimeout = 200*time.Millisecond, 300*time.Millisecond
t.Cleanup(func() { infoReadHeaderTimeout, infoReadTimeout = oldHeader, oldRead })
_, path := serveInfo(t, infoRegistry(), 0600)
// Half a request line, then silence: the header deadline ends it.
conn := rawInfoRequest(t, path, "GET /v1/whoami HTTP/1.1\r\nHost: neb")
require.NoError(t, conn.SetReadDeadline(time.Now().Add(2*time.Second)))
start := time.Now()
_, err := conn.Read(make([]byte, 1))
require.Error(t, err, "the server closed the connection")
assert.Less(t, time.Since(start), time.Second)
// And the server still answers the next client.
status, _ := test.HTTPGet(t, path, "/v1/whoami")
assert.Equal(t, http.StatusOK, status)
}
+23
View File
@@ -0,0 +1,23 @@
//go:build windows
package localapi
import (
"context"
"testing"
"github.com/slackhq/nebula/commands"
"github.com/slackhq/nebula/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestInfoIsNotServedOnWindows: a pipe has no mode to open to other accounts, so Rebind
// says so and leaves nothing listening, whatever the config asked for.
func TestInfoIsNotServedOnWindows(t *testing.T) {
s := NewInfoServer(test.NewLogger(), commands.New())
err := s.Rebind(context.Background(), pipePrefix+"nebula-info-test", Access{Mode: 0600})
require.ErrorIs(t, err, errInfoUnsupported)
assert.Empty(t, s.Path())
s.Stop()
}
+81 -26
View File
@@ -43,21 +43,25 @@ func withPeerFD(conn net.Conn, f func(fd int) error) error {
// in for root, or for a stranger, without owning a second account on the host.
var currentUID = func() uint32 { return uint32(os.Geteuid()) }
// infoSupported reports that the info API can be served here: a unix socket takes the
// mode an operator configures, which is the whole of its access control.
const infoSupported = true
// 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.
// looser than and which are set after bind: connect(2) needs write permission on the
// socket file on every unix we build for. The control socket is 0600 and adds the peer
// check the server applies on accept, unless a group shares it, in which case it is
// 0660 with that group and the kernel's check is the whole of it; the info socket is
// whatever the operator configured, since opening it to others is its purpose.
//
// 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.
// 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.
@@ -98,7 +102,8 @@ func listen(path string, acc Access) (net.Listener, error) {
// 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.
// the node is never looser than 0600, however briefly, whatever mode it ends up at;
// see umaskMu.
umaskMu.Lock()
old = syscall.Umask(0077)
ln, err := net.Listen(network, path)
@@ -108,6 +113,10 @@ func listen(path string, acc Access) (net.Listener, error) {
return nil, err
}
// What was just bound, so that a failure below gives up that node and not whatever
// has been put at the path since.
id, _ := identifySocket(path)
// 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
@@ -119,26 +128,47 @@ func listen(path string, acc Access) (net.Listener, error) {
// 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)
release(ln, path, id)
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
// checkOwnSocket refuses what fi found at path unless it is a socket owned by the user
// this process runs as. fi comes from a look that does not follow links, so a symlink is
// seen as one, and the mode is only ever applied to a socket of ours. checkSocketDir
// vouched for the directory at bind, but that was then: a directory loosened since, or a
// reload that moves the socket, is a chance for whoever can write it to replace the node
// with a link, or a hard link to a file of their choosing, and have a root daemon set the
// configured mode, 0660 or 0666 for an info socket, on that file. A link, another type
// of file, or somebody else's socket is reported and left as it is. applyAccess, per
// platform, is what makes the look and the change land on the same node.
func checkOwnSocket(path string, fi os.FileInfo) error {
if !isSocket(fi) {
return fmt.Errorf("%s is not a socket but %s, refusing to change its mode", path, describeNode(fi))
}
if err := os.Lchown(path, -1, gid); err != nil {
return err
if uid, ok := owner(fi); ok && uid != currentUID() {
return fmt.Errorf("%s is owned by %s, not by us, refusing to change its mode", path, describeUID(uid))
}
return nil
}
// describeNode names what kind of file fi describes, for a message about something found
// where a socket was expected.
func describeNode(fi os.FileInfo) string {
switch m := fi.Mode(); {
case m&os.ModeSymlink != 0:
return "a symlink"
case m.IsDir():
return "a directory"
case m.IsRegular():
return "a regular file"
default:
return "a " + m.Type().String() + " file"
}
return os.Chmod(path, acc.Mode.Perm())
}
// checkSocketDir refuses to serve from a directory that somebody other than the user
@@ -220,8 +250,12 @@ func describeUID(uid uint32) string {
// 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.
//
// The look is an Lstat: a symlink at path is refused as not a socket, whether or not it
// resolves. Stat would report a dangling one as nothing there, and bind(2) on darwin
// and the BSDs follows a dangling link and creates the socket node at its target.
func clearStale(path string) error {
fi, err := os.Stat(path)
fi, err := os.Lstat(path)
if err != nil {
if os.IsNotExist(err) {
return nil
@@ -253,10 +287,31 @@ func clearStale(path string) error {
// removeSocket unlinks the socket node. Its directory stays, whoever made it; see
// listen.
func removeSocket(path string) {
//
// Only the node id names is removed. One that replaced it is another daemon's, bound
// after ours was taken from under us, and is left where it is. A zero id, a node that
// could not be identified when it was bound, is removed.
func removeSocket(path string, id socketID) {
if cur, ok := identifySocket(path); ok && id != (socketID{}) && cur != id {
return
}
_ = os.Remove(path)
}
// identifySocket returns the identity of the node at path, looked at without following
// links, and false when there is nothing there or it cannot be read.
func identifySocket(path string) (socketID, bool) {
fi, err := os.Lstat(path)
if err != nil {
return socketID{}, false
}
st, ok := fi.Sys().(*syscall.Stat_t)
if !ok {
return socketID{}, false
}
return socketID{dev: uint64(st.Dev), ino: uint64(st.Ino)}, true
}
func dial(path string, timeout time.Duration) (net.Conn, error) {
return net.DialTimeout(network, path, timeout)
}
+106 -2
View File
@@ -288,6 +288,52 @@ func TestListenerDoesNotUnlinkOnClose(t *testing.T) {
assert.NoError(t, err, "closing the old listener left the newer daemon's socket in place")
}
// TestRebindRecreatesASocketRemovedFromUnderIt: logind clearing /run/user/<uid> on
// logout, or a unit cleaning a RuntimeDirectory= it shares, takes the node away while
// the listener stays open and unreachable. The next reload, which Rebinds onto the same
// path, has to notice and bind the socket again rather than return early because the
// path has not changed.
func TestRebindRecreatesASocketRemovedFromUnderIt(t *testing.T) {
path := testSocket(t)
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
t.Cleanup(s.Stop)
require.NoError(t, os.Remove(path))
_, err := (&Client{Path: path}).Describe()
require.Error(t, err, "nothing can reach the daemon with its node gone")
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
assert.Equal(t, path, s.Path())
_, err = (&Client{Path: path}).Describe()
require.NoError(t, err, "the plain reload brought the socket back")
// And a Rebind with the socket in place is still a no-op.
before, ok := identifySocket(path)
require.True(t, ok)
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
after, _ := identifySocket(path)
assert.Equal(t, before, after, "the node was left alone")
}
// TestStopLeavesAStrangersSocketAlone: once our node has been taken from under us and
// another daemon has bound the path, the node there is that daemon's, and our Stop must
// not unlink it by path.
func TestStopLeavesAStrangersSocketAlone(t *testing.T) {
path := testSocket(t)
s := NewServer(test.NewLogger(), basicRegistry())
require.NoError(t, s.Rebind(context.Background(), path, ownerOnly()))
require.NoError(t, os.Remove(path))
other, err := net.Listen(network, path)
require.NoError(t, err)
t.Cleanup(func() { other.Close() })
s.Stop()
_, err = os.Stat(path)
assert.NoError(t, err, "the newer daemon's socket is still 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.
@@ -322,7 +368,7 @@ func TestListenRefusesADirectoryOwnedBySomebodyElse(t *testing.T) {
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"))
release(ln, filepath.Join(dir, "s.sock"), socketID{})
}
// TestListenThroughASymlinkedDirectory: a link where the directory should be is judged
@@ -344,7 +390,7 @@ func TestListenThroughASymlinkedDirectory(t *testing.T) {
ln, err := listen(path, ownerOnly())
require.NoError(t, err, "our link to our directory")
release(ln, path)
release(ln, path, socketID{})
assert.NoFileExists(t, filepath.Join(real, "s.sock"))
asUID(t, me+1)
@@ -356,3 +402,61 @@ func TestListenThroughASymlinkedDirectory(t *testing.T) {
_, err = listen(path, ownerOnly())
require.ErrorContains(t, err, "writable by everyone")
}
// TestListenRefusesASymlinkAtThePath: a link where the socket should be is somebody
// else's doing, and it is refused before bind rather than followed. bind(2) on darwin
// follows a dangling link and creates the socket at its target, and the chmod after it
// would follow a link to anything, so neither the target nor the link is touched.
func TestListenRefusesASymlinkAtThePath(t *testing.T) {
path := testSocket(t)
require.NoError(t, os.Mkdir(filepath.Dir(path), 0700))
target := filepath.Join(filepath.Dir(path), "target")
require.NoError(t, os.Symlink(target, path))
_, err := listen(path, Access{Mode: 0666})
require.ErrorContains(t, err, "is not a socket")
_, statErr := os.Lstat(target)
assert.True(t, os.IsNotExist(statErr), "a dangling link's target is not created")
require.NoError(t, os.WriteFile(target, nil, 0600))
_, err = listen(path, Access{Mode: 0666})
require.ErrorContains(t, err, "is not a socket")
fi, err := os.Stat(target)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0600), fi.Mode().Perm(), "the file the link names keeps its mode")
assert.True(t, fi.Mode().IsRegular())
}
// TestChmodSocketRefusesWhatIsNotOurSocket: the mode is set through a look that does not
// follow links, so a link, a plain file, or a socket of another user's at the path is
// reported instead of chmoded. The last is seen as another user by changing who we are.
func TestChmodSocketRefusesWhatIsNotOurSocket(t *testing.T) {
dir := filepath.Join(privateBase(t), "n")
require.NoError(t, os.Mkdir(dir, 0700))
file := filepath.Join(dir, "file")
require.NoError(t, os.WriteFile(file, nil, 0600))
link := filepath.Join(dir, "link")
require.NoError(t, os.Symlink(file, link))
require.ErrorContains(t, applyAccess(link, Access{Mode: 0666}), "a symlink")
require.ErrorContains(t, applyAccess(file, Access{Mode: 0666}), "a regular file")
fi, err := os.Stat(file)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0600), fi.Mode().Perm())
sock := filepath.Join(dir, "s.sock")
ln, err := net.Listen(network, sock)
require.NoError(t, err)
defer ln.Close()
require.NoError(t, applyAccess(sock, Access{Mode: 0660}))
fi, err = os.Stat(sock)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0660), fi.Mode().Perm())
asUID(t, currentUID()+1)
require.ErrorContains(t, applyAccess(sock, Access{Mode: 0666}), "not by us")
fi, err = os.Stat(sock)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0660), fi.Mode().Perm())
}
+14 -4
View File
@@ -33,6 +33,12 @@ const network = "pipe"
// exists so a Server is built the same way on every platform.
var currentUID = func() uint32 { return 0 }
// infoSupported is false here. A pipe has no mode to open to other accounts, and the
// security descriptor listen builds admits the daemon's account alone, which is the
// opposite of what the info API is for. Serving it on windows means designing which
// accounts a pipe admits, and that has not been done.
const infoSupported = false
// 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
@@ -142,14 +148,18 @@ func sddlFor(acc Access) (string, error) {
const pipeClientAccess uint32 = windows.FILE_READ_DATA | windows.FILE_WRITE_DATA |
windows.FILE_READ_ATTRIBUTES | windows.READ_CONTROL | windows.SYNCHRONIZE
// 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, id socketID) {}
// identifySocket has no node to identify: a pipe is its handle, and it cannot be
// removed from under the daemon, so an endpoint on its path is always still there.
func identifySocket(path string) (socketID, bool) { return socketID{}, true }
// 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
+11 -7
View File
@@ -1,11 +1,15 @@
// 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.
// Package localapi serves the nebula command registry over local IPC endpoints: the
// control socket, so the user who ran nebula can inspect and control it from a CLI on
// the same machine, and the info API, so other programs on the machine can ask who is
// behind a vpn address.
//
// 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.
// The control socket's 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. It serves every command in the registry. The info API is opt in, HTTP, and as
// open as the operator makes its socket; it serves the identity commands alone, decided
// here, on the serving side; see InfoServer for what that is and why it is safe.
//
// # Protocol
//
+17 -162
View File
@@ -6,12 +6,12 @@ import (
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"math"
"net"
"runtime/debug"
"strconv"
"sync"
"sync/atomic"
"time"
@@ -39,29 +39,19 @@ const (
// Server serves a command registry over a local socket to the user who ran nebula.
type Server struct {
l *slog.Logger
endpoint
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.
// the peer check below 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
}
@@ -69,7 +59,7 @@ type Server struct {
// registered on it after the server is built.
func NewServer(l *slog.Logger, reg *commands.Registry) *Server {
return &Server{
l: l,
endpoint: endpoint{l: l, name: "local control socket"},
src: reg,
maxConns: defaultMaxConns,
slowAfter: defaultSlowAfter,
@@ -80,23 +70,9 @@ func NewServer(l *slog.Logger, reg *commands.Registry) *Server {
}
// 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.
// Stop or ctx cancellation; endpoint.rebindLocked says what that promises. 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()
@@ -106,7 +82,14 @@ func (s *Server) Rebind(ctx context.Context, path string, acc Access) error {
// 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)
err := s.rebindLocked(ctx, path, acc, func(ln net.Listener) io.Closer {
s.l.Info("Local control socket listening", "path", path, "access", acc.String())
// Closing the listener, when it is given up, ends the accept loop on the
// closed-listener error; in flight commands keep their connections, which stay
// usable until they finish.
go s.accept(ln, s.semaphore())
return ln
})
if err != nil {
// Whatever was serving still is, with the access it had.
s.trustSocket.Store(prev)
@@ -114,137 +97,9 @@ func (s *Server) Rebind(ctx context.Context, path string, acc Access) error {
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
// in a fresh one across a Stop and Rebind 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 {
+32
View File
@@ -274,6 +274,38 @@ func TestStopDetachesTheWatcher(t *testing.T) {
require.Eventually(t, func() bool { return s.Path() == "" }, time.Second, 10*time.Millisecond, "the second run's context stops it")
}
// TestGivingUpTheSocketDetachesTheWatcher: when the node is gone from under the endpoint
// and the bind that follows fails, the run is over and its watcher went with it, so a
// later Rebind is watched on the ctx it was given, as after a Stop.
func TestGivingUpTheSocketDetachesTheWatcher(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()))
// The node goes and a directory takes its place, so the bind that follows fails.
require.NoError(t, os.Remove(path))
require.NoError(t, os.Mkdir(path, 0700))
err := s.Rebind(ctxA, path, ownerOnly())
require.ErrorContains(t, err, "is not a socket")
assert.Empty(t, s.Path(), "the socket was given up before the bind failed")
require.NoError(t, os.Remove(path))
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")
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())
+137 -21
View File
@@ -9,23 +9,69 @@ import (
"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.
// Endpoints is every local API Wire brought up, with the one Stop the daemon's exit
// path calls.
type Endpoints struct {
Control *Server
Info *InfoServer
}
// Stop closes both endpoints and removes their sockets.
func (e *Endpoints) Stop() {
e.Info.Stop()
e.Control.Stop()
}
// Wire brings up the local APIs for a running nebula and keeps them following the
// config across reloads: the control socket unless the config turns it off, and the
// info API when the config asks for it.
//
// Every entry point that runs nebula as a daemon calls this, so the endpoint behaves the
// Every entry point that runs nebula as a daemon calls this, so the endpoints behave 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.
// does not live in nebula.Main: the control socket 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 {
// Call it after Control.Start, so the listeners only exist once the interface is up.
func Wire(ctx context.Context, l *slog.Logger, c *config.C, reg *commands.Registry) *Endpoints {
l = l.With("subsystem", "localapi")
return &Endpoints{
Control: wireControl(ctx, l, c, reg),
Info: wireInfo(ctx, l, c, reg),
}
}
// sharedSocketPath reports the path the config gives to both the control socket and the
// info API, when it gives them one path. The two speak different protocols, so neither
// binds it: each side says so and leaves it, and the operator fixes the config. The
// configured paths are compared, not the ones being served, so the case is caught on
// whichever side reconciles first, when the other side's bind has failed, and when a
// reload moves one socket onto the other's path.
func sharedSocketPath(c *config.C) (string, bool) {
if !ControlEnabled(c) {
return "", false
}
ctl, err := SocketPathFromConfig(c)
if err != nil {
return "", false
}
info, err := infoConfigFrom(c)
if err != nil || !info.Enabled || info.Path != ctl {
return "", false
}
return ctl, true
}
// wireControl brings up the control socket and keeps it pointed at the configured path.
func wireControl(ctx context.Context, l *slog.Logger, c *config.C, reg *commands.Registry) *Server {
srv := NewServer(l, reg)
// Registered before the first attempt, so that a path or bind that is bad at startup
@@ -55,10 +101,6 @@ func Wire(ctx context.Context, l *slog.Logger, c *config.C, reg *commands.Regist
// 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) {
@@ -81,20 +123,34 @@ func followControlConfig(ctx context.Context, l *slog.Logger, c *config.C, srv *
return
}
if p, shared := sharedSocketPath(c); shared {
l.Warn("Local control socket is configured on the info API's socket path and will not bind it",
"path", p, "current", srv.Path(), "fix", "set "+InfoSocketKey+" to another path")
return
}
old := srv.Path()
err = srv.Rebind(ctx, want, acc)
reportRebind(l, old, want, err, srv.Path())
reportRebind(l, "local control socket", old, want, err, srv.Path())
}
onReload(l, c, "local control socket", reconcile)
return reconcile
}
// onReload registers fn to run on every reload. 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 onReload(l *slog.Logger, c *config.C, what string, fn func(*config.C)) {
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)
l.Error("Recovered from panic reloading the "+what, "panic", r)
}
}()
reconcile(c)
fn(c)
})
return reconcile
}
// warnUnusable logs a config the endpoint cannot follow: with nothing serving, the
@@ -108,7 +164,7 @@ func warnUnusable(l *slog.Logger, what, current string, err error) {
}
// 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
// nothing for an endpoint 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
@@ -117,7 +173,7 @@ func warnUnusable(l *slog.Logger, what, current string, err error) {
// 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) {
func reportRebind(l *slog.Logger, what, old, want string, err error, now string) {
level := l.Warn
if errors.Is(err, errEndpointHeld) {
level = l.Error
@@ -126,15 +182,75 @@ func reportRebind(l *slog.Logger, old, want string, err error, now string) {
switch {
case err == nil:
if old != "" && old != want {
l.Info("Moved the local control socket", "from", old, "to", want)
l.Info("Moved the "+what, "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)
level("Could not start the "+what+", it 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)
level("Could not bind the "+what+" at the requested path", "error", err, "requested", want, "serving", now)
}
}
// wireInfo brings up the info API when the config asks for one and follows the config
// across reloads: enabling it, disabling it, moving its socket and changing its mode.
//
// Every reload is reconciled, as the control socket's is; see followControlConfig. A
// section that does not parse keeps whatever is running: the operator is told, and the
// programs relying on the endpoint are not cut off by a typo.
func wireInfo(ctx context.Context, l *slog.Logger, c *config.C, reg *commands.Registry) *InfoServer {
srv := NewInfoServer(l, reg)
// Said once per process rather than on every reload: it is a platform limit, not
// something the next reload could fix.
saidUnsupported := false
reconcile := func(c *config.C) {
cfg, err := infoConfigFrom(c)
if err != nil {
warnUnusable(l, "local info API", srv.Path(), err)
return
}
if !cfg.Enabled {
if srv.Path() != "" {
srv.Stop()
l.Info("Stopped the local info API, it is disabled in the reloaded config")
}
return
}
if !infoSupported {
if !saidUnsupported {
l.Warn("Local info API is enabled in the config but is not served on this platform yet", "key", InfoEnabledKey)
saidUnsupported = true
}
return
}
// The two endpoints speak different protocols; one socket cannot be both. Caught
// here, from the configured paths, because listen's own answer, that the socket is
// in use by another nebula, would send the operator looking for a second process.
if p, shared := sharedSocketPath(c); shared {
if cur := srv.Path(); cur != "" {
l.Warn("Ignoring the local info API's new socket path, it is also the control socket's; keeping the current endpoint",
"path", p, "current", cur, "fix", "set "+InfoSocketKey+" to another path")
} else {
l.Warn("Local info API is not available, its socket path is also the control socket's", "path", p, "fix", "set "+InfoSocketKey+" to another path")
}
return
}
old := srv.Path()
err = srv.Rebind(ctx, cfg.Path, cfg.Access)
reportRebind(l, "local info API", old, cfg.Path, err, srv.Path())
}
// Registered before the first attempt, as the control socket's is; see wireControl.
onReload(l, c, "local info API", reconcile)
reconcile(c)
return srv
}
// 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
+45 -6
View File
@@ -26,7 +26,7 @@ func TestWireRecoversFromBadPathAtStartup(t *testing.T) {
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())
srv := Wire(ctx, test.NewLogger(), c, basicRegistry()).Control
t.Cleanup(srv.Stop)
assert.Empty(t, srv.Path(), "an over-long path must not produce a socket")
@@ -85,7 +85,7 @@ func TestWireRetriesAFailedMoveOnPlainReload(t *testing.T) {
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(section(first)))
srv := Wire(ctx, test.NewLogger(), c, basicRegistry())
srv := Wire(ctx, test.NewLogger(), c, basicRegistry()).Control
t.Cleanup(srv.Stop)
require.Equal(t, first, srv.Path())
@@ -122,7 +122,7 @@ func TestWireRecoversOnPlainReloadAfterBadBind(t *testing.T) {
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(cfg))
srv := Wire(ctx, test.NewLogger(), c, basicRegistry())
srv := Wire(ctx, test.NewLogger(), c, basicRegistry()).Control
t.Cleanup(srv.Stop)
assert.Empty(t, srv.Path(), "listen creates one level only, so a missing runtime dir fails the bind")
@@ -144,7 +144,8 @@ func TestWireRecoversOnPlainReloadAfterBadBind(t *testing.T) {
}
// TestWireControlCanBeDisabled: local_api.control.enabled false serves no socket, and
// the key follows reloads both ways.
// the key follows reloads both ways. With the control socket off, its path is free for
// the info API to use.
func TestWireControlCanBeDisabled(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
@@ -156,8 +157,9 @@ func TestWireControlCanBeDisabled(t *testing.T) {
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(section(false)))
srv := Wire(ctx, test.NewLogger(), c, basicRegistry())
t.Cleanup(srv.Stop)
eps := Wire(ctx, test.NewLogger(), c, basicRegistry())
t.Cleanup(eps.Stop)
srv := eps.Control
assert.Empty(t, srv.Path())
_, err := os.Lstat(path)
assert.True(t, os.IsNotExist(err), "no socket is bound")
@@ -172,4 +174,41 @@ func TestWireControlCanBeDisabled(t *testing.T) {
assert.Empty(t, srv.Path())
_, err = os.Lstat(path)
assert.True(t, os.IsNotExist(err), "the socket is removed")
require.NoError(t, c.ReloadConfigString(section(false)+fmt.Sprintf(" info:\n enabled: true\n socket: %s\n", path)).Err)
assert.Equal(t, path, eps.Info.Path(), "a disabled control socket does not hold its path")
}
// TestWireRefusesOnePathForBothAPIs: a config that gives the control socket and the info
// API one path is refused by both sides, from the configured paths, so it is caught
// however the binds would have ordered themselves. Moving one off the path on a reload
// brings both up.
func TestWireRefusesOnePathForBothAPIs(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
section := func(ctl, info string) string {
return fmt.Sprintf("local_api:\n control:\n socket: %s\n info:\n enabled: true\n socket: %s\n", ctl, info)
}
shared := testSocket(t)
c := config.NewC(test.NewLogger())
require.NoError(t, c.LoadString(section(shared, shared)))
eps := Wire(ctx, test.NewLogger(), c, basicRegistry())
t.Cleanup(eps.Stop)
assert.Empty(t, eps.Control.Path(), "the control socket does not take the info API's path")
assert.Empty(t, eps.Info.Path(), "the info API does not take the control socket's path")
_, err := os.Stat(shared)
assert.True(t, os.IsNotExist(err), "neither side bound the shared path")
// A reload that moves the control socket onto the info API's path is refused the same
// way, and the control socket keeps the one it has.
other := testSocket(t)
require.NoError(t, c.ReloadConfigString(section(other, shared)).Err)
require.Equal(t, other, eps.Control.Path())
require.Equal(t, shared, eps.Info.Path())
require.NoError(t, c.ReloadConfigString(section(shared, shared)).Err)
assert.Equal(t, other, eps.Control.Path(), "a move onto the info API's path keeps the current socket")
assert.Equal(t, shared, eps.Info.Path())
}