Merge pull request 'Vouch for the account that runs mesh-cli, and ask the controller its line (hq ADR 0272)' (#69) from feat/272-the-engine-vouches-for-mesh-cli into main
This commit was merged in pull request #69.
This commit is contained in:
@@ -151,6 +151,19 @@ carries. The claim is not. The foundation is four container services, and:
|
||||
So `foundation.lock` here is a real bundle with a placeholder's content. Saying that plainly
|
||||
beats shipping a host that claims a foundation it has never raised.
|
||||
|
||||
## mesh-cli's socket (novox/hq ADR 0272)
|
||||
|
||||
The engine serves `/run/nox-mesh-host/cli.sock` for `mesh-cli`, the operator's command (`internal/meshcli`). It
|
||||
reads the asking account from the kernel (`SO_PEERCRED`) and asks the controller on this node's own subject; the
|
||||
controller decides what the line may do, and the engine runs nothing it receives. One line per connection, each
|
||||
connection's deadline set the moment it is accepted, two lines at once per account and eight for the machine,
|
||||
"busy" beyond that.
|
||||
|
||||
**A process outside the machine's own namespaces is refused**, whatever its user id: the peer's mount and user
|
||||
namespace must be the engine's. So `mesh-cli` run inside a flatpak, a toolbox or distrobox container, or any other
|
||||
container — even one running as the operator's own id — is refused with the reason; run it from the machine's own
|
||||
shell.
|
||||
|
||||
## A capability is detected, never assumed
|
||||
|
||||
The reason this is the first thing built rather than a detail of it.
|
||||
|
||||
@@ -1218,6 +1218,9 @@ func runLink(ctx context.Context, opts options) error {
|
||||
}
|
||||
// And the core builds this host placed are judged, whichever host placed them (to-be 45 §8).
|
||||
go watchWhatThisHostPlaced(aside, mine.Node, queue, say)
|
||||
// And mesh-cli is answered on this machine: the account from the kernel, the line asked of the controller
|
||||
// (novox/hq ADR 0272).
|
||||
go serveMeshCLI(aside, queue, say)
|
||||
|
||||
held := link.HoldRoused(aside, membership, queue, say, opts.timeout, rousedBySignal(ctx))
|
||||
// The act in hand finishes and is kept before this process exits: an apply that stood aside with
|
||||
@@ -1364,6 +1367,8 @@ func (w *adoptionWatch) changed(r link.Report) bool {
|
||||
// be a control surface on every machine, reachable by anything that can reach the machine, in
|
||||
// exchange for saving twenty seconds — and the whole security argument rests on there not being
|
||||
// one. A signal is delivered by the service manager to a process it already supervises.
|
||||
// (mesh-cli's socket is no such surface: it has no network address and nothing sent to it acts on
|
||||
// the machine — it is asked of the controller, novox/hq ADR 0272 §3.)
|
||||
//
|
||||
// SIGHUP, because that is the signal a long-running program conventionally reads as *look again*,
|
||||
// and nothing here is being reloaded from a file that a different signal would suit better.
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/cli"
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
"github.com/novox/mesh-host/internal/meshcli"
|
||||
)
|
||||
|
||||
// serveMeshCLI answers mesh-cli on this machine for as long as this engine serves (novox/hq ADR 0272): the account
|
||||
// from the kernel, the line asked of the controller over the link open at the time. A socket that cannot be served
|
||||
// is said once, and this engine goes on applying; mesh-cli then says the engine does not answer.
|
||||
func serveMeshCLI(ctx context.Context, queue *link.Queue, say func(string)) {
|
||||
if err := meshcli.Serve(ctx, cli.Socket, forwardOver(queue), say); err != nil {
|
||||
say("mesh-cli is not answered on this machine: " + err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
// forwardOver asks the controller over the queue's link, and follows the line to what came of it.
|
||||
func forwardOver(queue *link.Queue) meshcli.Forward {
|
||||
return func(ctx context.Context, asked meshcli.Asked) cli.Answer {
|
||||
return meshcli.AskTheController(ctx, queue.AskController, asked)
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ module github.com/novox/mesh-host
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261009103748-a5a040c9049b
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
golang.org/x/crypto v0.57.0
|
||||
golang.org/x/sys v0.48.0
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261009103748-a5a040c9049b h1:HmasScQ02QeFlAFDnmRCOEserbtDJJtutK+acBmGlNs=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261009103748-a5a040c9049b/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
|
||||
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
|
||||
|
||||
@@ -67,6 +67,11 @@ func (b OverNATS) Health(ctx context.Context, node string, body []byte) error {
|
||||
return b.Conn.FlushWithContext(flush)
|
||||
}
|
||||
|
||||
// CLISubject is where this node's engine asks the controller a line `mesh-cli` was given on this machine
|
||||
// (novox/hq ADR 0272 §3): its own node's, inside the grant every engine already has (`mesh.control.<node>.>`), so the
|
||||
// bus server is the authority on which node asked. Core NATS, never the CONTROL stream, which names its own subjects.
|
||||
func CLISubject(node string) string { return "mesh.control." + node + ".cli" }
|
||||
|
||||
// ToolSubject is a module's tool on one machine's node tools — the subject a node-engine asks a declared
|
||||
// tool check on (novox/hq ADR 0240, to-be 48 §3), granted it for its own machine's instance and no other.
|
||||
func ToolSubject(module, tool, node string) string {
|
||||
|
||||
@@ -230,7 +230,8 @@ func (l *natsLink) Ask(ctx context.Context, subject string, body []byte) ([]byte
|
||||
case err == nil:
|
||||
return reply, nil
|
||||
case errors.Is(err, nats.ErrNoResponders):
|
||||
return nil, fmt.Errorf("nothing on this machine answers %s", subject)
|
||||
// The server says nobody subscribes the subject: delivered to no one.
|
||||
return nil, fmt.Errorf("%w: nothing on this machine answers %s", ErrNotAsked, subject)
|
||||
}
|
||||
if refused := l.refusal(subject, before); refused != "" {
|
||||
return nil, fmt.Errorf("%s: %v%s", subject, err, refused)
|
||||
|
||||
@@ -544,6 +544,27 @@ func (q *Queue) AskTool(ctx context.Context, module, tool string) (bool, string,
|
||||
return ReadToolAnswer(reply)
|
||||
}
|
||||
|
||||
// AskController asks the controller a line mesh-cli was given on this machine, on this node's own subject, over
|
||||
// the link open now (novox/hq ADR 0272 §3), and returns its answer as it came. No link open is said, not waited for.
|
||||
func (q *Queue) AskController(ctx context.Context, body []byte) ([]byte, error) {
|
||||
q.init()
|
||||
q.mu.Lock()
|
||||
bus := q.bus
|
||||
q.mu.Unlock()
|
||||
if bus == nil {
|
||||
return nil, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked)
|
||||
}
|
||||
asker, ok := bus.(ToolBus)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("%w: the link to the bus open now cannot ask the controller", ErrNotAsked)
|
||||
}
|
||||
return asker.Ask(ctx, CLISubject(q.Membership.Node), body)
|
||||
}
|
||||
|
||||
// ErrNotAsked is a request the link knows it never delivered: no link open, or nobody listening on its subject. Any
|
||||
// other failure of a request may come after it was delivered, so what came of it is not known (novox/hq ADR 0272).
|
||||
var ErrNotAsked = errors.New("not asked")
|
||||
|
||||
// ReadToolAnswer reads a tool's reply envelope — `{result, error}`, as the node tools answer every call —
|
||||
// as a health answer: an error is not healthy, and so is a result that does not say `healthy: true`.
|
||||
func ReadToolAnswer(reply []byte) (bool, string, error) {
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
package meshcli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/cli"
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
)
|
||||
|
||||
// The controller answers a line as it answers every call (novox/hq ADR 0272, review): in an envelope, `result` or
|
||||
// `error`, and within its window for an answer — so a line still running is answered `running` with its call, and
|
||||
// followed by the same account on the same node until it ends. Its side holds the same field names (mesh-controller
|
||||
// internal/link, the mesh-cli field names' test).
|
||||
|
||||
// ErrNotAsked is the link's: a request it knows it never delivered.
|
||||
var ErrNotAsked = link.ErrNotAsked
|
||||
|
||||
// AskFunc asks the controller one request on this node's subject and returns its reply.
|
||||
type AskFunc func(ctx context.Context, body []byte) ([]byte, error)
|
||||
|
||||
// followEvery is how often a line still running is asked about again.
|
||||
var followEvery = 2 * time.Second
|
||||
|
||||
// envelope is a call's answer: its result, or its error.
|
||||
type envelope struct {
|
||||
Result json.RawMessage `json:"result"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
|
||||
// runningResult is what a call still running answers in its result.
|
||||
type runningResult struct {
|
||||
Running bool `json:"running"`
|
||||
Call string `json:"call"`
|
||||
}
|
||||
|
||||
// AskTheController asks the line and follows it to its answer, within ctx. What it says is one of three things,
|
||||
// never confused: what the line printed; that nothing ran, and why (the request was never taken); or that what came
|
||||
// of it is not known — it may have run — naming the call to look at when one is known.
|
||||
func AskTheController(ctx context.Context, ask AskFunc, asked Asked) cli.Answer {
|
||||
body, _ := json.Marshal(asked)
|
||||
reply, err := ask(ctx, body)
|
||||
if err != nil {
|
||||
// **Only a request the link knows it never asked ran nothing**: no link open, nobody listening. Any other
|
||||
// failure may come after the request was delivered, and the line may have run (review of ADR 0272).
|
||||
if errors.Is(err, ErrNotAsked) {
|
||||
return cli.Refusal("the controller was not reached, so nothing ran: " + err.Error())
|
||||
}
|
||||
return cli.NotKnown("", "no answer from the controller ("+err.Error()+")")
|
||||
}
|
||||
call := ""
|
||||
for {
|
||||
var env envelope
|
||||
if err := json.Unmarshal(reply, &env); err != nil {
|
||||
return cli.NotKnown(call, "the controller's answer could not be read")
|
||||
}
|
||||
if env.Error != "" {
|
||||
return cli.NotKnown(call, "the controller answered "+env.Error)
|
||||
}
|
||||
var r runningResult
|
||||
_ = json.Unmarshal(env.Result, &r)
|
||||
if !r.Running {
|
||||
var a cli.Answer
|
||||
if err := json.Unmarshal(env.Result, &a); err != nil {
|
||||
return cli.NotKnown(call, "the controller's answer could not be read")
|
||||
}
|
||||
return a
|
||||
}
|
||||
if r.Call != "" {
|
||||
call = r.Call
|
||||
}
|
||||
if call == "" {
|
||||
return cli.NotKnown("", "the controller says the line is running and names no call to follow")
|
||||
}
|
||||
// Asked again until it ends, through a link that may drop and come back: a failed follow is tried
|
||||
// again, and only ctx ending stops it.
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return cli.NotKnown(call, fmt.Sprintf("no answer within the time mesh-cli waits (%s)", cli.AskWithin))
|
||||
case <-time.After(followEvery):
|
||||
}
|
||||
body, _ := json.Marshal(Asked{Follow: call, Account: asked.Account, UID: asked.UID})
|
||||
if reply, err = ask(ctx, body); err == nil {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
package meshcli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// controllerSays is a controller answering each request in turn from a script, and recording what it was asked.
|
||||
type controllerSays struct {
|
||||
mu sync.Mutex
|
||||
script []func(Asked) ([]byte, error)
|
||||
asked []Asked
|
||||
}
|
||||
|
||||
func (c *controllerSays) ask(_ context.Context, body []byte) ([]byte, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
var a Asked
|
||||
_ = json.Unmarshal(body, &a)
|
||||
c.asked = append(c.asked, a)
|
||||
if len(c.script) == 0 {
|
||||
return nil, context.DeadlineExceeded
|
||||
}
|
||||
next := c.script[0]
|
||||
if len(c.script) > 1 {
|
||||
c.script = c.script[1:]
|
||||
}
|
||||
return next(a)
|
||||
}
|
||||
|
||||
func result(v any) func(Asked) ([]byte, error) {
|
||||
return func(Asked) ([]byte, error) { return json.Marshal(map[string]any{"result": v}) }
|
||||
}
|
||||
|
||||
func fastFollow(t *testing.T) {
|
||||
was := followEvery
|
||||
followEvery = time.Millisecond
|
||||
t.Cleanup(func() { followEvery = was })
|
||||
}
|
||||
|
||||
// A line still running when the controller first answers is followed, by its call, to what it printed.
|
||||
func TestALineStillRunningIsFollowedToItsAnswer(t *testing.T) {
|
||||
fastFollow(t)
|
||||
running := result(map[string]any{"running": true, "call": "call-9"})
|
||||
c := &controllerSays{script: []func(Asked) ([]byte, error){running, running, running,
|
||||
result(map[string]any{"stdout": []byte("pushed\n"), "exit": 0, "terminal": true})}}
|
||||
a := AskTheController(context.Background(), c.ask, Asked{Line: []string{"push", "laptop"}, Account: "op", UID: 1000})
|
||||
if string(a.Stdout) != "pushed\n" || !a.Terminal || a.Refused != "" || a.Unknown != "" {
|
||||
t.Fatalf("followed to %+v", a)
|
||||
}
|
||||
for _, f := range c.asked[1:] {
|
||||
if f.Follow != "call-9" || f.Account != "op" || len(f.Line) != 0 {
|
||||
t.Fatalf("a follow asked %+v", f)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// **No answer in time is never "nothing ran"** (review of ADR 0272): the line was asked and may have run, and the
|
||||
// call to look at is named when it is known.
|
||||
func TestNoAnswerInTimeSaysTheLineMayHaveRun(t *testing.T) {
|
||||
fastFollow(t)
|
||||
c := &controllerSays{}
|
||||
a := AskTheController(context.Background(), c.ask, Asked{Line: []string{"push"}, Account: "op"})
|
||||
if a.Refused != "" || !strings.Contains(a.Unknown, "may have run") || a.Exit == 0 {
|
||||
t.Fatalf("a request with no answer reads %+v", a)
|
||||
}
|
||||
c = &controllerSays{script: []func(Asked) ([]byte, error){result(map[string]any{"running": true, "call": "call-3"}),
|
||||
func(Asked) ([]byte, error) { return nil, context.DeadlineExceeded }}}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
||||
defer cancel()
|
||||
a = AskTheController(ctx, c.ask, Asked{Line: []string{"push"}, Account: "op"})
|
||||
if a.Refused != "" || !strings.Contains(a.Unknown, "call-3") || !strings.Contains(a.Unknown, "may have run") {
|
||||
t.Fatalf("a running line never answered reads %+v", a)
|
||||
}
|
||||
}
|
||||
|
||||
// A request the bus never carried to a controller ran nothing, and says so.
|
||||
func TestARequestNobodyTookSaysNothingRan(t *testing.T) {
|
||||
c := &controllerSays{script: []func(Asked) ([]byte, error){func(Asked) ([]byte, error) {
|
||||
return nil, fmt.Errorf("%w: nothing on this machine answers mesh.control.laptop.cli", ErrNotAsked)
|
||||
}}}
|
||||
a := AskTheController(context.Background(), c.ask, Asked{Line: []string{"status"}, Account: "op"})
|
||||
if !strings.Contains(a.Refused, "nothing ran") || a.Unknown != "" {
|
||||
t.Fatalf("a request nobody took reads %+v", a)
|
||||
}
|
||||
c = &controllerSays{script: []func(Asked) ([]byte, error){func(Asked) ([]byte, error) {
|
||||
return nil, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked)
|
||||
}}}
|
||||
if a := AskTheController(context.Background(), c.ask, Asked{Line: []string{"status"}}); !strings.Contains(a.Refused, "nothing ran") {
|
||||
t.Fatalf("no link reads %+v", a)
|
||||
}
|
||||
}
|
||||
|
||||
// **Any error once the request may have been delivered is not known, never "nothing ran"** (review of ADR 0272):
|
||||
// only a request the link says it never asked — no link open, nobody listening — ran nothing.
|
||||
func TestOnlyARequestNeverAskedSaysNothingRan(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
err error
|
||||
nothing bool
|
||||
}{
|
||||
{fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked), true},
|
||||
{fmt.Errorf("%w: nothing on this machine answers mesh.control.laptop.cli", ErrNotAsked), true},
|
||||
{errors.New("nats: connection closed"), false},
|
||||
{errors.New("mesh.control.laptop.cli: permissions violation"), false},
|
||||
{context.DeadlineExceeded, false},
|
||||
} {
|
||||
err := c.err
|
||||
ctl := &controllerSays{script: []func(Asked) ([]byte, error){func(Asked) ([]byte, error) { return nil, err }}}
|
||||
a := AskTheController(context.Background(), ctl.ask, Asked{Line: []string{"push"}, Account: "op"})
|
||||
if c.nothing && (!strings.Contains(a.Refused, "nothing ran") || a.Unknown != "") {
|
||||
t.Errorf("%v reads %+v; it was never asked", err, a)
|
||||
}
|
||||
if !c.nothing && (a.Refused != "" || !strings.Contains(a.Unknown, "may have run")) {
|
||||
t.Errorf("%v reads %+v; it may have been asked", err, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package meshcli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/cli"
|
||||
)
|
||||
|
||||
// **A connection that says nothing is bounded from the moment it is accepted** (review of ADR 0272): its deadline is
|
||||
// set before any lookup, so it is answered and closed within readWithin, holds no slot after, and asks nothing.
|
||||
func TestAConnectionThatSaysNothingIsAnsweredWithinItsDeadline(t *testing.T) {
|
||||
was := readWithin
|
||||
readWithin = 200 * time.Millisecond
|
||||
t.Cleanup(func() { readWithin = was })
|
||||
asked := false
|
||||
socket := serving(t, func(context.Context, Asked) cli.Answer {
|
||||
asked = true
|
||||
return cli.Answer{}
|
||||
})
|
||||
conn, err := net.Dial("unix", socket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
|
||||
a, err := cli.ReadAnswer(conn)
|
||||
if err != nil {
|
||||
t.Fatalf("a silent connection was not answered within its deadline: %v", err)
|
||||
}
|
||||
if asked || a.Refused == "" || !strings.Contains(a.Refused, "nothing was asked") {
|
||||
t.Fatalf("a silent connection was asked (%v) or answered %+v", asked, a)
|
||||
}
|
||||
// Its slot was given back: the account's next line is not busy.
|
||||
b, err := cli.Ask(context.Background(), socket, []string{"status"})
|
||||
if err != nil || strings.Contains(b.Refused, "busy") || !asked {
|
||||
t.Fatalf("the next line after a silent one reads %+v (%v), asked %v", b, err, asked)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,208 @@
|
||||
// Package meshcli is the node-engine's side of `mesh-cli` (novox/hq ADR 0272): a socket on this machine that any
|
||||
// account may connect to, the account read from the kernel, and the line asked of the controller.
|
||||
//
|
||||
// **The node-engine vouches; it does not decide.** It adds what only it can know — which account on this machine
|
||||
// is asking, from the kernel's peer credentials, never from anything the program says — and sends the request out
|
||||
// over its own link, on its own node's subject. The controller judges whether the line runs as its terminal, as an
|
||||
// ordinary call, or not at all (ADR 0272 §4). Nothing received here is run or applied on the machine.
|
||||
//
|
||||
// **No listening control surface** (novox/hq ADR 0004, ADR 0272 §3): the socket has no network address, so
|
||||
// nothing on another machine reaches it, and nothing sent to it acts on this one.
|
||||
package meshcli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"os/user"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/cli"
|
||||
)
|
||||
|
||||
// Asked is what the node-engine sends the controller on its own node's subject: the line, and the account the
|
||||
// kernel says is asking. The controller's side holds the same field names (a test on each side).
|
||||
type Asked struct {
|
||||
Line []string `json:"line,omitempty"`
|
||||
// Account is the login of the user id the kernel gave, from this machine's user database.
|
||||
Account string `json:"account"`
|
||||
// UID is the user id itself, so a renamed or duplicated login is still the same id.
|
||||
UID uint32 `json:"uid"`
|
||||
// Follow is a call a line runs as, asked about again by the account that asked it (AskTheController).
|
||||
Follow string `json:"follow,omitempty"`
|
||||
}
|
||||
|
||||
// Forward asks the controller and follows the line to what came of it (AskTheController): what it printed, that
|
||||
// nothing ran and why, or that it is not known.
|
||||
type Forward func(ctx context.Context, asked Asked) cli.Answer
|
||||
|
||||
// readWithin bounds how long a connection may take to say its line: a program that connects and says nothing
|
||||
// holds nothing for long. A variable so a test need not wait.
|
||||
var readWithin = 30 * time.Second
|
||||
|
||||
// atOnce bounds how many lines are asked at the same time, and perAccount how many of them one account's: a person
|
||||
// types one at a time, so more is a program on the machine. One over either bound is answered busy at once — the
|
||||
// engine never stops accepting for it, so one account cannot hold the socket from every other (review of ADR 0272).
|
||||
var (
|
||||
atOnce = 8
|
||||
perAccount = 2
|
||||
)
|
||||
|
||||
// slots counts the lines being asked, in all and per user id.
|
||||
type slots struct {
|
||||
mu sync.Mutex
|
||||
all int
|
||||
byUID map[uint32]int
|
||||
}
|
||||
|
||||
func (s *slots) take(uid uint32) bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.all >= atOnce || s.byUID[uid] >= perAccount {
|
||||
return false
|
||||
}
|
||||
s.all++
|
||||
s.byUID[uid]++
|
||||
return true
|
||||
}
|
||||
|
||||
func (s *slots) give(uid uint32) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.all--
|
||||
if s.byUID[uid]--; s.byUID[uid] <= 0 {
|
||||
delete(s.byUID, uid)
|
||||
}
|
||||
}
|
||||
|
||||
// peerOf reads the connecting process's credentials from the kernel: its user id and process id. A variable so a
|
||||
// test can say what the kernel would; it is the only source of the account.
|
||||
var peerOf = peerCredentials
|
||||
|
||||
// sameNamespaces refuses a peer whose mount or user namespace is not the engine's. A variable so a test can say.
|
||||
var sameNamespaces = namespacesOf
|
||||
|
||||
// lookupAccount names a user id from this machine's user database.
|
||||
var lookupAccount = func(uid uint32) (string, error) {
|
||||
u, err := user.LookupId(strconv.FormatUint(uint64(uid), 10))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return u.Username, nil
|
||||
}
|
||||
|
||||
// Serve listens on socket until ctx ends, and answers each connection: one request, one answer.
|
||||
//
|
||||
// The directory is made 0755 and the socket 0666: every account on the machine may ask, because the account is
|
||||
// what is vouched for and the controller decides what it may do. A file at the socket's path that is not a socket
|
||||
// is never removed: the engine says so and serves nothing.
|
||||
func Serve(ctx context.Context, socket string, forward Forward, say func(string)) error {
|
||||
dir := filepath.Dir(socket)
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return fmt.Errorf("mesh-cli's socket directory: %w", err)
|
||||
}
|
||||
// The directory is the engine's: a link, or one another account owns, could put the socket where that
|
||||
// account replaces it with its own. Said, and nothing served.
|
||||
if problem := notOwnDirectory(dir); problem != "" {
|
||||
return fmt.Errorf("mesh-cli is not served on this machine: %s", problem)
|
||||
}
|
||||
if err := os.Chmod(dir, 0o755); err != nil {
|
||||
return fmt.Errorf("mesh-cli's socket directory: %w", err)
|
||||
}
|
||||
if info, err := os.Lstat(socket); err == nil {
|
||||
if info.Mode()&os.ModeSocket == 0 {
|
||||
return fmt.Errorf("%s is there and is not a socket, and the node-engine did not put it there; "+
|
||||
"mesh-cli is not served on this machine until it is moved", socket)
|
||||
}
|
||||
// A socket left by the engine before this one: nothing listens on it any more.
|
||||
if err := os.Remove(socket); err != nil {
|
||||
return fmt.Errorf("the socket a previous engine left at %s: %w", socket, err)
|
||||
}
|
||||
}
|
||||
l, err := net.ListenUnix("unix", &net.UnixAddr{Name: socket, Net: "unix"})
|
||||
if err != nil {
|
||||
return fmt.Errorf("listening for mesh-cli at %s: %w", socket, err)
|
||||
}
|
||||
if err := os.Chmod(socket, 0o666); err != nil {
|
||||
l.Close()
|
||||
return fmt.Errorf("opening mesh-cli's socket to every account: %w", err)
|
||||
}
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
l.Close()
|
||||
}()
|
||||
held := &slots{byUID: map[uint32]int{}}
|
||||
var wg sync.WaitGroup
|
||||
defer wg.Wait()
|
||||
for {
|
||||
conn, err := l.AcceptUnix()
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
var ne net.Error
|
||||
if errors.As(err, &ne) && ne.Timeout() {
|
||||
continue
|
||||
}
|
||||
return fmt.Errorf("mesh-cli's socket: %w", err)
|
||||
}
|
||||
// Bounded from the first byte: nothing below may wait on a connection for longer than this before the
|
||||
// line is read, whatever the peer does (review of ADR 0272).
|
||||
_ = conn.SetDeadline(time.Now().Add(readWithin))
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
answer(ctx, conn, held, forward, say)
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// answer serves one connection.
|
||||
func answer(ctx context.Context, conn *net.UnixConn, held *slots, forward Forward, say func(string)) {
|
||||
defer conn.Close()
|
||||
reply := func(a cli.Answer) {
|
||||
_ = conn.SetWriteDeadline(time.Now().Add(readWithin))
|
||||
if err := cli.WriteAnswer(conn, a); err != nil && say != nil {
|
||||
say(fmt.Sprintf("mesh-cli: the answer could not be written back: %v", err))
|
||||
}
|
||||
}
|
||||
uid, pid, err := peerOf(conn)
|
||||
if err != nil {
|
||||
// Never sent as anyone: an account the kernel did not give is no account at all.
|
||||
reply(cli.Refusal("the node-engine could not read which account is asking, so nothing was asked: " + err.Error()))
|
||||
return
|
||||
}
|
||||
// The user id is this machine's account only in this machine's namespaces: a container running with the
|
||||
// operator's id shares the number and nothing else (review of ADR 0272).
|
||||
if err := sameNamespaces(pid); err != nil {
|
||||
reply(cli.Refusal("the asking process does not run in this machine's own namespaces, so it is not this " +
|
||||
"machine's account and nothing was asked: " + err.Error()))
|
||||
return
|
||||
}
|
||||
if !held.take(uid) {
|
||||
reply(cli.Refusal(fmt.Sprintf("busy: user id %d has %d lines being asked already, or the machine %d; ask "+
|
||||
"again when one has ended. Nothing was asked", uid, perAccount, atOnce)))
|
||||
return
|
||||
}
|
||||
defer held.give(uid)
|
||||
account, err := lookupAccount(uid)
|
||||
if err != nil || account == "" {
|
||||
reply(cli.Refusal(fmt.Sprintf("user id %d is not in this machine's user database, so nothing was asked", uid)))
|
||||
return
|
||||
}
|
||||
req, err := cli.ReadRequest(conn)
|
||||
if err != nil {
|
||||
reply(cli.Refusal("the request could not be read, so nothing was asked: " + err.Error()))
|
||||
return
|
||||
}
|
||||
// From here the wait is the controller's: as long as mesh-cli waits, and no longer.
|
||||
_ = conn.SetDeadline(time.Now().Add(cli.AskWithin))
|
||||
asking, cancel := context.WithTimeout(ctx, cli.AskWithin-30*time.Second)
|
||||
defer cancel()
|
||||
reply(forward(asking, Asked{Line: req.Line, Account: account, UID: uid}))
|
||||
}
|
||||
@@ -0,0 +1,218 @@
|
||||
package meshcli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net"
|
||||
"os"
|
||||
"os/user"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/cli"
|
||||
)
|
||||
|
||||
// serving starts Serve on a socket of the test's own, and stops it when the test ends.
|
||||
func serving(t *testing.T, forward Forward) string {
|
||||
t.Helper()
|
||||
socket := filepath.Join(t.TempDir(), "run", "cli.sock")
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- Serve(ctx, socket, forward, func(string) {}) }()
|
||||
t.Cleanup(func() {
|
||||
cancel()
|
||||
if err := <-done; err != nil {
|
||||
t.Errorf("Serve ended with %v", err)
|
||||
}
|
||||
})
|
||||
for i := 0; i < 200; i++ {
|
||||
if _, err := os.Stat(socket); err == nil {
|
||||
return socket
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
t.Fatal("the socket never appeared")
|
||||
return ""
|
||||
}
|
||||
|
||||
// The account sent is the one the kernel gives for the connecting process, and the answer comes back as it was.
|
||||
func TestTheAccountAskingIsTheKernelsAndTheAnswerComesBack(t *testing.T) {
|
||||
var got Asked
|
||||
socket := serving(t, func(_ context.Context, asked Asked) cli.Answer {
|
||||
got = asked
|
||||
return cli.Answer{Stdout: []byte("done\n"), Exit: 0, Terminal: true, Why: "the terminal"}
|
||||
})
|
||||
a, err := cli.Ask(context.Background(), socket, []string{"settings", "show", "zsh"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(a.Stdout) != "done\n" || !a.Terminal || a.Refused != "" {
|
||||
t.Fatalf("the answer came back as %+v", a)
|
||||
}
|
||||
me, err := user.Current()
|
||||
if err != nil {
|
||||
t.Skipf("this machine cannot name the test's account: %v", err)
|
||||
}
|
||||
if int(got.UID) != os.Getuid() || got.Account != me.Username {
|
||||
t.Fatalf("asked as %d %q, and the kernel says %d %q", got.UID, got.Account, os.Getuid(), me.Username)
|
||||
}
|
||||
if strings.Join(got.Line, " ") != "settings show zsh" {
|
||||
t.Fatalf("the line was sent as %q", got.Line)
|
||||
}
|
||||
if info, err := os.Stat(socket); err != nil || info.Mode().Perm() != 0o666 {
|
||||
t.Fatalf("the socket is not open to every account: %v %v", info.Mode(), err)
|
||||
}
|
||||
}
|
||||
|
||||
// Credentials the kernel did not give are no account: refused, and nothing is asked.
|
||||
func TestAConnectionWhoseAccountCannotBeReadIsNeverAskedAsAnyone(t *testing.T) {
|
||||
before := peerOf
|
||||
peerOf = func(*net.UnixConn) (uint32, int32, error) { return 0, 0, errors.New("no credentials") }
|
||||
t.Cleanup(func() { peerOf = before })
|
||||
asked := false
|
||||
socket := serving(t, func(context.Context, Asked) cli.Answer {
|
||||
asked = true
|
||||
return cli.Answer{}
|
||||
})
|
||||
a, err := cli.Ask(context.Background(), socket, []string{"status"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if asked || a.Refused == "" || a.Exit == 0 {
|
||||
t.Fatalf("a connection with no account was asked (%v) or not refused: %+v", asked, a)
|
||||
}
|
||||
}
|
||||
|
||||
// A request over the bound is refused and never asked.
|
||||
func TestARequestOverTheBoundIsRefused(t *testing.T) {
|
||||
asked := false
|
||||
socket := serving(t, func(context.Context, Asked) cli.Answer {
|
||||
asked = true
|
||||
return cli.Answer{}
|
||||
})
|
||||
a, err := cli.Ask(context.Background(), socket, []string{strings.Repeat("x", cli.MaxRequest)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if asked || a.Refused == "" {
|
||||
t.Fatalf("a request over the bound was asked (%v): %+v", asked, a)
|
||||
}
|
||||
}
|
||||
|
||||
// A file at the socket's path that is not a socket is never removed.
|
||||
func TestAFileThatIsNotASocketIsKept(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
socket := filepath.Join(dir, "cli.sock")
|
||||
if err := os.WriteFile(socket, []byte("mine"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err := Serve(context.Background(), socket, nil, nil)
|
||||
if err == nil || !strings.Contains(err.Error(), "not a socket") {
|
||||
t.Fatalf("a file at the socket's path was not refused: %v", err)
|
||||
}
|
||||
if body, _ := os.ReadFile(socket); string(body) != "mine" {
|
||||
t.Fatal("the file at the socket's path was touched")
|
||||
}
|
||||
}
|
||||
|
||||
// The request the controller reads holds these field names; the controller's side holds the same list
|
||||
// (mesh-controller, the mesh-cli request's test).
|
||||
func TestTheRequestToTheControllerKeepsItsFieldNames(t *testing.T) {
|
||||
body, err := json.Marshal(Asked{Line: []string{"status"}, Account: "a", UID: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := keysOf(t, body); got != "account line uid" {
|
||||
t.Fatalf("the request's fields are %q", got)
|
||||
}
|
||||
body, _ = json.Marshal(Asked{Follow: "call-1", Account: "a", UID: 1})
|
||||
if got := keysOf(t, body); got != "account follow uid" {
|
||||
t.Fatalf("a follow's fields are %q", got)
|
||||
}
|
||||
// What the controller answers as a line's result, and as a running line's: the same names its side holds.
|
||||
body, err = json.Marshal(cli.Answer{Stdout: []byte("o"), Stderr: []byte("e"), Exit: 1, Terminal: true, Why: "w",
|
||||
Refused: "r", Cut: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := keysOf(t, body); got != "cut exit refused stderr stdout terminal why" {
|
||||
t.Fatalf("the answer's fields are %q", got)
|
||||
}
|
||||
body, _ = json.Marshal(runningResult{Running: true, Call: "call-1"})
|
||||
if got := keysOf(t, body); got != "call running" {
|
||||
t.Fatalf("a running answer's fields are %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func keysOf(t *testing.T, body []byte) string {
|
||||
t.Helper()
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(body, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var keys []string
|
||||
for k := range m {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
return strings.Join(keys, " ")
|
||||
}
|
||||
|
||||
// **One account cannot hold the socket** (review of ADR 0272): its lines beyond its own bound are answered busy at
|
||||
// once, nothing is asked, and the engine goes on accepting.
|
||||
func TestOneAccountsLinesBeyondItsBoundAreAnsweredBusy(t *testing.T) {
|
||||
was := perAccount
|
||||
perAccount = 1
|
||||
t.Cleanup(func() { perAccount = was })
|
||||
release := make(chan struct{})
|
||||
started := make(chan struct{}, 2)
|
||||
socket := serving(t, func(context.Context, Asked) cli.Answer {
|
||||
started <- struct{}{}
|
||||
<-release
|
||||
return cli.Answer{}
|
||||
})
|
||||
// Released before the engine is stopped: cleanups run last first, and the engine waits for its lines.
|
||||
t.Cleanup(func() { close(release) })
|
||||
go func() { _, _ = cli.Ask(context.Background(), socket, []string{"push"}) }()
|
||||
<-started
|
||||
done := make(chan cli.Answer, 1)
|
||||
go func() {
|
||||
a, _ := cli.Ask(context.Background(), socket, []string{"status"})
|
||||
done <- a
|
||||
}()
|
||||
select {
|
||||
case a := <-done:
|
||||
if !strings.Contains(a.Refused, "busy") {
|
||||
t.Fatalf("a second line from the same account reads %+v", a)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("a second line from the same account waited instead of being answered busy")
|
||||
}
|
||||
if len(started) != 0 {
|
||||
t.Fatal("the busy line was asked")
|
||||
}
|
||||
}
|
||||
|
||||
// **A process in another namespace is not this machine's account** (review of ADR 0272): a container running with
|
||||
// the operator's user id shares the id and not the machine; refused, and nothing is asked.
|
||||
func TestAPeerInAnotherNamespaceIsRefused(t *testing.T) {
|
||||
before := sameNamespaces
|
||||
sameNamespaces = func(pid int32) error { return errors.New("its mount namespace is not the engine's") }
|
||||
t.Cleanup(func() { sameNamespaces = before })
|
||||
asked := false
|
||||
socket := serving(t, func(context.Context, Asked) cli.Answer {
|
||||
asked = true
|
||||
return cli.Answer{}
|
||||
})
|
||||
a, err := cli.Ask(context.Background(), socket, []string{"status"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if asked || !strings.Contains(a.Refused, "namespace") {
|
||||
t.Fatalf("a peer in another namespace was asked (%v): %+v", asked, a)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
//go:build linux
|
||||
|
||||
package meshcli
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
// peerCredentials is the user id of the process at the other end, as the kernel recorded it when it connected
|
||||
// (SO_PEERCRED). Nothing the process says is read.
|
||||
func peerCredentials(conn *net.UnixConn) (uint32, int32, error) {
|
||||
raw, err := conn.SyscallConn()
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
var cred *syscall.Ucred
|
||||
var inner error
|
||||
if err := raw.Control(func(fd uintptr) {
|
||||
cred, inner = syscall.GetsockoptUcred(int(fd), syscall.SOL_SOCKET, syscall.SO_PEERCRED)
|
||||
}); err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
if inner != nil {
|
||||
return 0, 0, inner
|
||||
}
|
||||
if cred == nil {
|
||||
return 0, 0, errors.New("the kernel gave no credentials")
|
||||
}
|
||||
return cred.Uid, cred.Pid, nil
|
||||
}
|
||||
|
||||
// namespacesOf refuses a process whose mount or user namespace is not this one's, read from the kernel's own
|
||||
// names for them. One that cannot be read is refused, never assumed the same.
|
||||
func namespacesOf(pid int32) error {
|
||||
for _, ns := range []string{"mnt", "user"} {
|
||||
mine, err := os.Readlink("/proc/self/ns/" + ns)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the engine's own %s namespace cannot be read: %w", ns, err)
|
||||
}
|
||||
theirs, err := os.Readlink(fmt.Sprintf("/proc/%d/ns/%s", pid, ns))
|
||||
if err != nil {
|
||||
return fmt.Errorf("process %d's %s namespace cannot be read: %w", pid, ns, err)
|
||||
}
|
||||
if theirs != mine {
|
||||
return fmt.Errorf("process %d is in %s namespace %s, the engine in %s", pid, ns, theirs, mine)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// notOwnDirectory says what is wrong with the socket's directory, or nothing: it must be a directory, not a
|
||||
// link, owned by the account the engine runs as.
|
||||
func notOwnDirectory(dir string) string {
|
||||
info, err := os.Lstat(dir)
|
||||
if err != nil {
|
||||
return err.Error()
|
||||
}
|
||||
if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
|
||||
return dir + " is not a directory the engine made"
|
||||
}
|
||||
if st, ok := info.Sys().(*syscall.Stat_t); ok && int(st.Uid) != os.Geteuid() {
|
||||
return fmt.Sprintf("%s belongs to user id %d, not to the engine's account", dir, st.Uid)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
//go:build !linux
|
||||
|
||||
package meshcli
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net"
|
||||
)
|
||||
|
||||
// peerCredentials is unknown where the kernel's way of saying it is not built here: refused, never guessed.
|
||||
func peerCredentials(*net.UnixConn) (uint32, int32, error) {
|
||||
return 0, 0, errors.New("this engine reads a peer's account on Linux only")
|
||||
}
|
||||
|
||||
// namespacesOf is never satisfied where peers cannot be read.
|
||||
func namespacesOf(int32) error { return errors.New("not read on this system") }
|
||||
|
||||
// notOwnDirectory is never satisfied where peers cannot be read.
|
||||
func notOwnDirectory(string) string { return "not served on this system" }
|
||||
Reference in New Issue
Block a user