Remove the mesh-cli forward to the controller
nox now calls the seats' verbs through the console itself (mesh-catalog #593), and the controller the engine forwarded its lines to ended (issue 1126), so the socket, its forward and the subject it asked on go. The kernel's word on a peer moves into the filled-file socket, its one remaining reader. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -171,26 +171,6 @@ 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.
|
||||
|
||||
**Only a login session can be the terminal.** The engine reads the asking process's cgroup and sends its login
|
||||
session's scope (`session-<id>.scope` under the account's own `user-<uid>.slice`, which logind's own record in
|
||||
`/run/systemd/sessions/<id>` must say is this account's, of class `user`), or nothing, read the moment the
|
||||
connection is accepted; the controller makes
|
||||
nothing else the terminal. A service running as the operator — the tool runner, a unit of the account's user
|
||||
manager — is at most an ordinary call.
|
||||
|
||||
**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.
|
||||
|
||||
@@ -100,7 +100,7 @@ func versionAt(self, stamped string) string {
|
||||
const usage = `nox-mesh-host — the node-engine
|
||||
|
||||
A daemon: its launcher runs it, and it applies what the mesh declares for this machine. The
|
||||
operator's command is nox (mesh-cli), on any node, from their own account (novox/hq ADR 0272);
|
||||
operator's command is nox (mesh-cli), on any node, from their own account, through the seats' verbs;
|
||||
the subcommands here are for installing a machine, joining it, and rescuing one no mesh can
|
||||
reach, run as root at the machine.
|
||||
|
||||
@@ -1064,9 +1064,6 @@ func runLink(ctx context.Context, opts options) error {
|
||||
}, BusyAskEvery)
|
||||
// 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)
|
||||
// And a filled file is written on the provisioner client's request, one at a time with the apply (novox/hq ADR
|
||||
// 0349 §4).
|
||||
go serveFilling(aside, opts, mine.Membership.Signer, queue, mine.Node, say)
|
||||
@@ -1087,8 +1084,6 @@ func runLink(ctx context.Context, opts options) error {
|
||||
// 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.
|
||||
|
||||
@@ -1,25 +0,0 @@
|
||||
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.Session, 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.Session) meshcli.Forward {
|
||||
return func(ctx context.Context, asked meshcli.Asked) cli.Answer {
|
||||
return meshcli.AskTheController(ctx, queue.AskController, asked)
|
||||
}
|
||||
}
|
||||
@@ -35,10 +35,9 @@ import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"github.com/novox/mesh-host/internal/meshcli"
|
||||
)
|
||||
|
||||
// Socket is where the node-engine accepts a write of a filled file, beside mesh-cli's socket.
|
||||
// Socket is where the node-engine accepts a write of a filled file.
|
||||
const Socket = "/run/nox-mesh-host/fill.sock"
|
||||
|
||||
// MaxRequest bounds one request, its content encoded in it: a configuration file, never an archive.
|
||||
@@ -293,11 +292,11 @@ func UIDOf(account string) (uint32, error) {
|
||||
return uint32(id), nil
|
||||
}
|
||||
|
||||
// peerOf and sameNamespaces are the kernel's word on the asking process, as mesh-cli's socket reads it. Variables so
|
||||
// peerOf and sameNamespaces are the kernel's word on the asking process (peer_linux.go). Variables so
|
||||
// a test can say.
|
||||
var (
|
||||
peerOf = meshcli.PeerCredentials
|
||||
sameNamespaces = meshcli.SameNamespaces
|
||||
peerOf = peerCredentials
|
||||
sameNamespaces = namespacesOf
|
||||
)
|
||||
|
||||
// within bounds how long an accepted connection may take to say its request, and refusedWithin how long a busy one
|
||||
@@ -352,7 +351,7 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return fmt.Errorf("the filled files' socket directory: %w", err)
|
||||
}
|
||||
if problem := meshcli.NotOwnDirectory(dir); problem != "" {
|
||||
if problem := notOwnDirectory(dir); problem != "" {
|
||||
return fmt.Errorf("filled files are not written on request on this machine: %s", problem)
|
||||
}
|
||||
if err := os.Chmod(dir, 0o755); err != nil {
|
||||
|
||||
@@ -532,7 +532,7 @@ func TestTheSocketGoesOnAfterAnAcceptFails(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The socket's directory is given the mode the socket needs, as mesh-cli's is.
|
||||
// The socket's directory is given the mode the socket needs.
|
||||
func TestTheSocketsDirectoryIsOpenedToEveryAccount(t *testing.T) {
|
||||
dir, err := os.MkdirTemp("", "fill")
|
||||
if err != nil {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//go:build linux
|
||||
|
||||
package meshcli
|
||||
package filling
|
||||
|
||||
import (
|
||||
"errors"
|
||||
@@ -1,6 +1,6 @@
|
||||
//go:build !linux
|
||||
|
||||
package meshcli
|
||||
package filling
|
||||
|
||||
import (
|
||||
"errors"
|
||||
@@ -46,11 +46,6 @@ 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 {
|
||||
|
||||
@@ -290,23 +290,6 @@ func (q *Session) AskTool(ctx context.Context, module, tool string) (health.Answ
|
||||
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 *Session) 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")
|
||||
|
||||
@@ -1,92 +0,0 @@
|
||||
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, Session: asked.Session})
|
||||
if reply, err = ask(ctx, body); err == nil {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,123 +0,0 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,42 +0,0 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -1,339 +0,0 @@
|
||||
// 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 (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"os/user"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"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"`
|
||||
// Session is the login session the asking process runs in — its systemd scope, `session-<id>.scope` under the
|
||||
// account's own slice — as the kernel's cgroup says it; empty for anything else: a system service, a user unit,
|
||||
// a process another account's session started. Only a login session is a person at a terminal (novox/hq ADR
|
||||
// 0272): the controller makes nothing else the terminal.
|
||||
Session string `json:"session,omitempty"`
|
||||
// Stdin is what mesh-cli's standard input held, at most cli.MaxStdin: carried to the controller with the line
|
||||
// and nowhere else. The engine says it in no journal, keeps it in no state, and never sends it again: a follow
|
||||
// carries none. The controller gives it to a line that runs as its terminal alone.
|
||||
Stdin []byte `json:"stdin,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
|
||||
|
||||
// refusedWithin bounds how long the engine waits for the request of a line it refuses before it is asked: a client
|
||||
// that has connected writes its request within microseconds, so a refused connection holds nothing for long. Never
|
||||
// longer than readWithin. A variable so a test can say.
|
||||
var refusedWithin = 1 * time.Second
|
||||
|
||||
// refusalsWaiting bounds how many refusals wait for their request at the same time, on the whole engine; drainCap
|
||||
// bounds how much of a refused request is read, and it is read into a small buffer and dropped. The socket is open
|
||||
// to every account, so a refusal must cost the engine little however many connections ask (review of ADR 0272).
|
||||
// Variables so a test can say.
|
||||
var (
|
||||
refusalsWaiting = make(chan struct{}, 32)
|
||||
drainCap = 4 << 10
|
||||
)
|
||||
|
||||
// drain reads and drops what a refused client wrote, up to the end of its request line, the end of the stream, the
|
||||
// connection's deadline or drainCap bytes, whichever comes first. Nothing is kept.
|
||||
func drain(r io.Reader) {
|
||||
var buf [512]byte
|
||||
for read := 0; read < drainCap; {
|
||||
n, err := r.Read(buf[:min(len(buf), drainCap-read)])
|
||||
read += n
|
||||
if bytes.IndexByte(buf[:n], '\n') >= 0 || err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
|
||||
// sessionOf reads the login session a process runs in, from the kernel. A variable so a test can say.
|
||||
var sessionOf = func(pid int32, uid uint32) string {
|
||||
raw, err := os.ReadFile(fmt.Sprintf("/proc/%d/cgroup", pid))
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return userSessionIn(string(raw), uid)
|
||||
}
|
||||
|
||||
// sessionsDir is where logind keeps a record of each session it holds. A variable so a test can say.
|
||||
var sessionsDir = "/run/systemd/sessions"
|
||||
|
||||
// userSessionIn is the login session a cgroup names, when logind's own record of it says it is a person's: this
|
||||
// account's (UID), of class `user` — not a greeter's or the user manager's — and of this very scope. The cgroup
|
||||
// alone names a scope; only the record says whose session it is. No record, or one that does not say so, is no
|
||||
// session.
|
||||
func userSessionIn(cgroup string, uid uint32) string {
|
||||
scope := loginSessionIn(cgroup, uid)
|
||||
id, ok := strings.CutPrefix(strings.TrimSuffix(scope, ".scope"), "session-")
|
||||
if !ok || id == "" || strings.ContainsAny(id, "/.") {
|
||||
return ""
|
||||
}
|
||||
raw, err := os.ReadFile(filepath.Join(sessionsDir, id))
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
// **The record is systemd's private format**, not an interface it promises: `KEY=value` lines logind writes for
|
||||
// itself. Only UID, CLASS and SCOPE are read, each at its first occurrence — logind writes each once, and a later
|
||||
// line saying otherwise is not the record's word. A systemd that changes the format, renames a key or stops
|
||||
// writing the file fails closed here: no session is read, so nobody is the terminal until this reads it again.
|
||||
record := map[string]string{}
|
||||
for _, line := range strings.Split(string(raw), "\n") {
|
||||
k, v, ok := strings.Cut(line, "=")
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if _, said := record[k]; !said {
|
||||
record[k] = v
|
||||
}
|
||||
}
|
||||
if record["UID"] != strconv.FormatUint(uint64(uid), 10) || record["CLASS"] != "user" || record["SCOPE"] != scope {
|
||||
return ""
|
||||
}
|
||||
return scope
|
||||
}
|
||||
|
||||
// loginSessionIn is the login session scope a process's cgroup names, for its own account: the unified hierarchy's
|
||||
// line, `0::/user.slice/user-<uid>.slice/session-<id>.scope`, and nothing below or beside it. Anything else is no
|
||||
// login session — a service under system.slice, a unit under the account's user manager (user@<uid>.service), a
|
||||
// scope of another account's.
|
||||
func loginSessionIn(cgroup string, uid uint32) string {
|
||||
for _, line := range strings.Split(cgroup, "\n") {
|
||||
path, ok := strings.CutPrefix(line, "0::")
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
prefix := fmt.Sprintf("/user.slice/user-%d.slice/", uid)
|
||||
scope, ok := strings.CutPrefix(path, prefix)
|
||||
if !ok || strings.Contains(scope, "/") || !strings.HasPrefix(scope, "session-") || !strings.HasSuffix(scope, ".scope") {
|
||||
return ""
|
||||
}
|
||||
return scope
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// 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).
|
||||
readBy := time.Now().Add(readWithin)
|
||||
_ = conn.SetDeadline(readBy)
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
answer(ctx, conn, readBy, held, forward, say)
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// answer serves one connection; readBy is the deadline set when it was accepted, for reading its line.
|
||||
func answer(ctx context.Context, conn *net.UnixConn, readBy time.Time, 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))
|
||||
}
|
||||
}
|
||||
// A refusal is written only after the request it answers is read (novox/hq issue 434). The client writes its
|
||||
// request first and reads the answer after; an engine that answers and closes before that write has landed
|
||||
// makes the write fail with a broken pipe, and the client never reads the refusal waiting for it — under load a
|
||||
// line beyond its account's bound read as an error instead of busy. The wait ends at refusedWithin, or earlier
|
||||
// at the deadline set when the connection was accepted; what was written is drained and never asked. Only
|
||||
// refusalsWaiting refusals wait at once: over that the refusal is written at once, as before, and only then can a
|
||||
// client still find its write broken.
|
||||
refuse := func(why string) {
|
||||
select {
|
||||
case refusalsWaiting <- struct{}{}:
|
||||
by := time.Now().Add(refusedWithin)
|
||||
if readBy.Before(by) {
|
||||
by = readBy
|
||||
}
|
||||
_ = conn.SetReadDeadline(by)
|
||||
drain(conn)
|
||||
<-refusalsWaiting
|
||||
default:
|
||||
}
|
||||
reply(cli.Refusal(why))
|
||||
}
|
||||
uid, pid, err := peerOf(conn)
|
||||
if err != nil {
|
||||
// Never sent as anyone: an account the kernel did not give is no account at all.
|
||||
refuse("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 {
|
||||
refuse("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
|
||||
}
|
||||
// And the login session, now: judged as the peer was when it connected, before it can exec or exit while its
|
||||
// line is read (review of ADR 0272).
|
||||
session := sessionOf(pid, uid)
|
||||
// Looked up before a slot is taken, so a refused account never holds one while it is refused.
|
||||
account, err := lookupAccount(uid)
|
||||
if err != nil || account == "" {
|
||||
refuse(fmt.Sprintf("user id %d is not in this machine's user database, so nothing was asked", uid))
|
||||
return
|
||||
}
|
||||
if !held.take(uid) {
|
||||
refuse(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)
|
||||
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, Session: session, Stdin: req.Stdin}))
|
||||
}
|
||||
@@ -1,534 +0,0 @@
|
||||
package meshcli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"os/user"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"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)
|
||||
}
|
||||
body, _ = json.Marshal(Asked{Line: []string{"status"}, Account: "a", UID: 1, Session: "session-3.scope"})
|
||||
if got := keysOf(t, body); got != "account line session uid" {
|
||||
t.Fatalf("the request's fields are %q", got)
|
||||
}
|
||||
body, _ = json.Marshal(Asked{Line: []string{"status"}, Account: "a", UID: 1})
|
||||
if got := keysOf(t, body); got != "account line uid" {
|
||||
t.Fatalf("the request's fields are %q", got)
|
||||
}
|
||||
body, _ = json.Marshal(Asked{Line: []string{"secret"}, Account: "a", UID: 1, Stdin: []byte("x")})
|
||||
if got := keysOf(t, body); got != "account line stdin uid" {
|
||||
t.Fatalf("a request with standard input has the fields %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, " ")
|
||||
}
|
||||
|
||||
// holdALine asks one line and returns once the engine is asking it, so it holds its account's slot. A first line
|
||||
// that ends or never arrives fails the test at once, with what came of it, instead of hanging the suite (novox/hq
|
||||
// issue 434).
|
||||
func holdALine(t *testing.T, socket string, started <-chan struct{}) {
|
||||
t.Helper()
|
||||
first := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := cli.Ask(context.Background(), socket, []string{"push"})
|
||||
first <- err
|
||||
}()
|
||||
select {
|
||||
case <-started:
|
||||
case err := <-first:
|
||||
t.Fatalf("the first line ended before it was asked: %v", err)
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the first line was never asked")
|
||||
}
|
||||
}
|
||||
|
||||
// **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) })
|
||||
holdALine(t, socket, started)
|
||||
type asked struct {
|
||||
a cli.Answer
|
||||
err error
|
||||
}
|
||||
done := make(chan asked, 1)
|
||||
go func() {
|
||||
a, err := cli.Ask(context.Background(), socket, []string{"status"})
|
||||
done <- asked{a, err}
|
||||
}()
|
||||
select {
|
||||
case got := <-done:
|
||||
// An error is a busy line the client never read (novox/hq issue 434), not an answer.
|
||||
if got.err != nil || !strings.Contains(got.a.Refused, "busy") {
|
||||
t.Fatalf("a second line from the same account reads %+v (%v)", got.a, got.err)
|
||||
}
|
||||
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 refusal waits for the request it answers** (novox/hq issue 434): the client writes its line first and reads
|
||||
// the answer after. An engine that answered busy and closed before that write landed broke the client's write, and
|
||||
// the refusal was never read — under the whole suite's load the busy line read as an error. Here the client
|
||||
// connects and holds its line back: no answer may come before the line, and the line, once written, is answered
|
||||
// busy. Ordered by the client's own write, so it holds however slow the machine is.
|
||||
func TestARefusalWaitsForTheRequestItAnswers(t *testing.T) {
|
||||
was, wasWithin := perAccount, refusedWithin
|
||||
perAccount, refusedWithin = 1, time.Minute
|
||||
t.Cleanup(func() { perAccount, refusedWithin = was, wasWithin })
|
||||
release := make(chan struct{})
|
||||
started := make(chan struct{}, 2)
|
||||
socket := serving(t, func(context.Context, Asked) cli.Answer {
|
||||
started <- struct{}{}
|
||||
<-release
|
||||
return cli.Answer{}
|
||||
})
|
||||
t.Cleanup(func() { close(release) })
|
||||
holdALine(t, socket, started)
|
||||
conn, err := net.Dial("unix", socket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
// Nothing is written yet: an engine that answers now answers before the request. The wait is far inside
|
||||
// refusedWithin, so an engine that waits is still waiting when it ends.
|
||||
_ = conn.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
|
||||
var b [1]byte
|
||||
if n, err := conn.Read(b[:]); n > 0 || !errors.Is(err, os.ErrDeadlineExceeded) {
|
||||
t.Fatalf("the engine answered before the request was written (%d bytes, %v)", n, err)
|
||||
}
|
||||
_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
|
||||
if err := cli.WriteRequest(conn, cli.Request{Line: []string{"status"}}); err != nil {
|
||||
t.Fatalf("the request could not be written: %v", err)
|
||||
}
|
||||
a, err := cli.ReadAnswer(conn)
|
||||
if err != nil || !strings.Contains(a.Refused, "busy") {
|
||||
t.Fatalf("a line written after the engine chose busy reads %+v (%v)", a, err)
|
||||
}
|
||||
if len(started) != 0 {
|
||||
t.Fatal("the busy line was asked")
|
||||
}
|
||||
}
|
||||
|
||||
// **Refusals that wait are bounded** (novox/hq issue 434, review of ADR 0272): with refusalsWaiting already full, a
|
||||
// further refused line is answered at once, without waiting for its request.
|
||||
func TestARefusalOverTheWaitingBoundIsAnsweredAtOnce(t *testing.T) {
|
||||
was, wasWithin, wasWaiting := perAccount, refusedWithin, refusalsWaiting
|
||||
perAccount, refusedWithin = 1, time.Minute
|
||||
refusalsWaiting = make(chan struct{}, 1)
|
||||
refusalsWaiting <- struct{}{}
|
||||
t.Cleanup(func() { perAccount, refusedWithin, refusalsWaiting = was, wasWithin, wasWaiting })
|
||||
release := make(chan struct{})
|
||||
started := make(chan struct{}, 2)
|
||||
socket := serving(t, func(context.Context, Asked) cli.Answer {
|
||||
started <- struct{}{}
|
||||
<-release
|
||||
return cli.Answer{}
|
||||
})
|
||||
t.Cleanup(func() { close(release) })
|
||||
holdALine(t, socket, started)
|
||||
conn, err := net.Dial("unix", socket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
// Nothing is written: a refusal that waited would wait refusedWithin, a minute, for it.
|
||||
_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
|
||||
a, err := cli.ReadAnswer(conn)
|
||||
if err != nil || !strings.Contains(a.Refused, "busy") {
|
||||
t.Fatalf("a refusal over the waiting bound reads %+v (%v)", a, err)
|
||||
}
|
||||
}
|
||||
|
||||
// countingReader is an endless request with no line end, counting what was read of it.
|
||||
type countingReader struct{ read int }
|
||||
|
||||
func (c *countingReader) Read(p []byte) (int, error) {
|
||||
for i := range p {
|
||||
p[i] = 'x'
|
||||
}
|
||||
c.read += len(p)
|
||||
return len(p), nil
|
||||
}
|
||||
|
||||
// **A refused request is drained, not kept** (novox/hq issue 434): what a refused client writes is read up to
|
||||
// drainCap bytes and no further, however much it sends.
|
||||
func TestARefusedRequestIsDrainedNoFurtherThanItsCap(t *testing.T) {
|
||||
r := &countingReader{}
|
||||
drain(r)
|
||||
if r.read != drainCap {
|
||||
t.Fatalf("drained %d bytes of an endless request; the cap is %d", r.read, drainCap)
|
||||
}
|
||||
after := &countingReader{}
|
||||
drain(io.MultiReader(strings.NewReader("{\"line\":[\"status\"]}\n"), after))
|
||||
if after.read != 0 {
|
||||
t.Fatalf("draining read %d bytes past the end of the request line", after.read)
|
||||
}
|
||||
}
|
||||
|
||||
// **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)
|
||||
}
|
||||
}
|
||||
|
||||
// **The session a line comes from is the kernel's word, and only a login session counts** (review of ADR 0272): a
|
||||
// service running as the operator — the tool runner, a user unit — is not a person at a terminal. The engine reads
|
||||
// the peer's cgroup and sends the login session's scope, or nothing; the controller makes no other terminal.
|
||||
func TestTheLoginSessionAskingIsReadFromTheKernel(t *testing.T) {
|
||||
for _, c := range []struct{ cgroup, session string }{
|
||||
{"0::/user.slice/user-1000.slice/session-3.scope\n", "session-3.scope"},
|
||||
{"0::/user.slice/user-1000.slice/session-c1.scope\n", "session-c1.scope"},
|
||||
{"0::/user.slice/user-1000.slice/user@1000.service/app.slice/tmux.scope\n", ""},
|
||||
{"0::/system.slice/node-tools.service\n", ""},
|
||||
{"0::/user.slice/user-1001.slice/session-3.scope\n", ""}, // another account's session
|
||||
{"12:pids:/user.slice\n0::/user.slice/user-1000.slice/session-9.scope/sub\n", ""},
|
||||
// Below a session scope, in a scope of its own: only the check that the scope is the account slice's direct
|
||||
// child refuses it.
|
||||
{"0::/user.slice/user-1000.slice/session-9.scope/x.scope\n", ""},
|
||||
{"", ""},
|
||||
} {
|
||||
if got := loginSessionIn(c.cgroup, 1000); got != c.session {
|
||||
t.Errorf("%q read as %q, want %q", c.cgroup, got, c.session)
|
||||
}
|
||||
}
|
||||
// And it is sent: the test process's own, whatever it is, as the kernel says it.
|
||||
var got Asked
|
||||
socket := serving(t, func(_ context.Context, asked Asked) cli.Answer {
|
||||
got = asked
|
||||
return cli.Answer{}
|
||||
})
|
||||
if _, err := cli.Ask(context.Background(), socket, []string{"status"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, _ := os.ReadFile("/proc/self/cgroup")
|
||||
if want := userSessionIn(string(raw), uint32(os.Getuid())); got.Session != want {
|
||||
t.Fatalf("sent session %q, the kernel says %q", got.Session, want)
|
||||
}
|
||||
}
|
||||
|
||||
// **A login session is a person's: systemd's own record says so** (review of ADR 0272): the scope the cgroup names
|
||||
// must be a session logind keeps for this very account, of class `user` — not a greeter's, not the user manager's,
|
||||
// not one whose record names another account — and read at accept, with the namespaces, before anything else.
|
||||
func TestALoginSessionIsAUsersByItsOwnRecord(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
was := sessionsDir
|
||||
sessionsDir = dir
|
||||
t.Cleanup(func() { sessionsDir = was })
|
||||
write := func(id, body string) {
|
||||
if err := os.WriteFile(filepath.Join(dir, id), []byte(body), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
write("9", "UID=1000\nUSER=op\nCLASS=user\nSCOPE=session-9.scope\nACTIVE=1\n")
|
||||
write("3", "UID=1000\nCLASS=greeter\nSCOPE=session-3.scope\n")
|
||||
write("5", "UID=1000\nCLASS=user\nSCOPE=session-6.scope\n")
|
||||
// The account the record names is not the slice's: only the record's UID check refuses it.
|
||||
write("8", "UID=1001\nCLASS=user\nSCOPE=session-8.scope\n")
|
||||
// A name with a dot, held as a record that would pass: only the check of the id's characters refuses it.
|
||||
write("a.b", "UID=1000\nCLASS=user\nSCOPE=session-a.b.scope\n")
|
||||
// A key said twice: the first is the record's word (systemd writes each once); a later one is not read.
|
||||
write("11", "UID=1001\nCLASS=user\nSCOPE=session-11.scope\nUID=1000\n")
|
||||
write("12", "UID=1000\nCLASS=user\nSCOPE=session-12.scope\nCLASS=greeter\nUID=1001\n")
|
||||
for _, c := range []struct {
|
||||
cgroup string
|
||||
uid uint32
|
||||
session string
|
||||
}{
|
||||
{"0::/user.slice/user-1000.slice/session-9.scope\n", 1000, "session-9.scope"},
|
||||
{"0::/user.slice/user-1000.slice/x.scope\n", 1000, ""},
|
||||
{"0::/user.slice/user-1000.slice/session-9.scope\n", 1001, ""}, // asked as another uid than the slice's
|
||||
{"0::/user.slice/user-1000.slice/session-8.scope\n", 1000, ""}, // the record names uid 1001
|
||||
{"0::/user.slice/user-1000.slice/session-3.scope\n", 1000, ""}, // a greeter's
|
||||
{"0::/user.slice/user-1000.slice/session-5.scope\n", 1000, ""}, // a record naming another scope
|
||||
{"0::/user.slice/user-1000.slice/session-7.scope\n", 1000, ""}, // no record at all
|
||||
{"0::/user.slice/user-1000.slice/session-a.b.scope\n", 1000, ""},
|
||||
{"0::/user.slice/user-1000.slice/session-...scope\n", 1000, ""},
|
||||
{"0::/user.slice/user-1000.slice/session-11.scope\n", 1000, ""}, // first UID is 1001
|
||||
{"0::/user.slice/user-1000.slice/session-12.scope\n", 1000, "session-12.scope"}, // first UID and CLASS stand
|
||||
} {
|
||||
if got := userSessionIn(c.cgroup, c.uid); got != c.session {
|
||||
t.Errorf("%q as uid %d read as %q, want %q", c.cgroup, c.uid, got, c.session)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// **The session is read at accept, with the namespaces, before the line** (review of ADR 0272): the peer is judged as
|
||||
// it was when it connected. The client sends its line only once the session has been read; were the session read
|
||||
// after the line — in the forward, say — the line would never be sent and the test fails.
|
||||
func TestTheSessionIsReadBeforeTheLine(t *testing.T) {
|
||||
var mu sync.Mutex
|
||||
var order []string
|
||||
note := func(s string) { mu.Lock(); order = append(order, s); mu.Unlock() }
|
||||
read := make(chan struct{}, 1)
|
||||
beforeNS, beforeSession := sameNamespaces, sessionOf
|
||||
sameNamespaces = func(int32) error { note("namespaces"); return nil }
|
||||
sessionOf = func(int32, uint32) string { note("session"); read <- struct{}{}; return "session-1.scope" }
|
||||
t.Cleanup(func() { sameNamespaces, sessionOf = beforeNS, beforeSession })
|
||||
var got Asked
|
||||
socket := serving(t, func(_ context.Context, asked Asked) cli.Answer {
|
||||
note("forward")
|
||||
got = asked
|
||||
return cli.Answer{}
|
||||
})
|
||||
conn, err := net.Dial("unix", socket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
select {
|
||||
case <-read:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("the session was not read before the line was sent")
|
||||
}
|
||||
note("line sent")
|
||||
if err := cli.WriteRequest(conn, cli.Request{Line: []string{"status"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
|
||||
if _, err := cli.ReadAnswer(conn); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if strings.Join(order, ",") != "namespaces,session,line sent,forward" || got.Session != "session-1.scope" {
|
||||
t.Fatalf("read in the order %v, sent %q", order, got.Session)
|
||||
}
|
||||
}
|
||||
|
||||
// mesh-cli's standard input reaches the controller with its line, whole, and only with the line: a follow of the
|
||||
// same call carries none, and the engine says it nowhere (novox/hq ADR 0259 §10, ADR 0272).
|
||||
func TestStandardInputGoesToTheControllerWithTheLineAndNowhereElse(t *testing.T) {
|
||||
secret := []byte("123456789:the-bot-token-given-at-the-terminal\n")
|
||||
var said []string
|
||||
var mu sync.Mutex
|
||||
var asked [][]byte
|
||||
ask := func(_ context.Context, body []byte) ([]byte, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
asked = append(asked, append([]byte(nil), body...))
|
||||
if len(asked) == 1 {
|
||||
return []byte(`{"result":{"running":true,"call":"call-7"}}`), nil
|
||||
}
|
||||
return []byte(`{"result":{"exit":0,"terminal":true}}`), nil
|
||||
}
|
||||
was := followEvery
|
||||
followEvery = time.Millisecond
|
||||
t.Cleanup(func() { followEvery = was })
|
||||
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, func(ctx context.Context, a Asked) cli.Answer { return AskTheController(ctx, ask, a) },
|
||||
func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() })
|
||||
}()
|
||||
defer func() { cancel(); <-done }()
|
||||
for i := 0; i < 200; i++ {
|
||||
if _, err := os.Stat(socket); err == nil {
|
||||
break
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
a, err := cli.AskWith(context.Background(), socket, cli.Request{Line: []string{"secret", "accept", "anchor",
|
||||
"telegram", "telegram-token", "--from", "-"}, Stdin: secret})
|
||||
if err != nil || a.Exit != 0 {
|
||||
t.Fatalf("%v %+v", err, a)
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if len(asked) != 2 {
|
||||
t.Fatalf("asked %d times, want the line and one follow", len(asked))
|
||||
}
|
||||
var first, follow Asked
|
||||
_ = json.Unmarshal(asked[0], &first)
|
||||
_ = json.Unmarshal(asked[1], &follow)
|
||||
if string(first.Stdin) != string(secret) {
|
||||
t.Fatalf("the controller was not given the standard input whole: %q", first.Stdin)
|
||||
}
|
||||
if len(follow.Stdin) != 0 || strings.Contains(string(asked[1]), "stdin") {
|
||||
t.Fatalf("a follow carried the standard input again: %s", asked[1])
|
||||
}
|
||||
for _, s := range said {
|
||||
if strings.Contains(s, "bot-token") {
|
||||
t.Fatalf("the engine said the standard input: %s", s)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,19 +0,0 @@
|
||||
package meshcli
|
||||
|
||||
import "net"
|
||||
|
||||
// The kernel's word on who is at the other end of a socket, shared with the node-engine's other local socket, the
|
||||
// write of a filled file (internal/filling, novox/hq ADR 0349 §4): one reading of it, so the two never judge a peer
|
||||
// differently.
|
||||
|
||||
// PeerCredentials is the user id and process id of the process at the other end, as the kernel recorded them when
|
||||
// it connected (SO_PEERCRED). Nothing the process says is read.
|
||||
func PeerCredentials(conn *net.UnixConn) (uint32, int32, error) { return peerCredentials(conn) }
|
||||
|
||||
// SameNamespaces refuses a process whose mount or user namespace is not the engine's: a user id is this machine's
|
||||
// account only in this machine's namespaces.
|
||||
func SameNamespaces(pid int32) error { return namespacesOf(pid) }
|
||||
|
||||
// NotOwnDirectory says what is wrong with a 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 { return notOwnDirectory(dir) }
|
||||
Reference in New Issue
Block a user