mesh-cli: follow a line to its answer, bound each account, refuse other namespaces (review of hq ADR 0272)
- A request that timed out was said as "the controller was not reached, nothing ran", though the line may have run. The engine now follows a line the controller answers as running, by its call, until it ends; when no answer comes in time it says the line may have run and names the call. "Nothing ran" is said only of a request no controller took. - One account could hold the socket: accepting waited for a free slot, and a connection could stall before its deadline was set. The deadline is set at accept, before any lookup; each user id has two lines at once and the machine eight, and one over is answered busy at once. - A container running with the operator's user id shares the number, not the machine: a peer whose mount or user namespace is not the engine's is refused, and nothing is asked. SDK pinned at the go/cli branch's newest commit until go/v0.1.9 is tagged.
This commit is contained in:
@@ -2,8 +2,6 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/cli"
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
@@ -19,21 +17,9 @@ func serveMeshCLI(ctx context.Context, queue *link.Queue, say func(string)) {
|
||||
}
|
||||
}
|
||||
|
||||
// forwardOver asks the controller over the queue's link, and reads its answer as mesh-cli's.
|
||||
// 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, error) {
|
||||
body, err := json.Marshal(asked)
|
||||
if err != nil {
|
||||
return cli.Answer{}, err
|
||||
}
|
||||
reply, err := queue.AskController(ctx, body)
|
||||
if err != nil {
|
||||
return cli.Answer{}, err
|
||||
}
|
||||
var a cli.Answer
|
||||
if err := json.Unmarshal(reply, &a); err != nil {
|
||||
return cli.Answer{}, fmt.Errorf("the controller's answer is not one: %w", err)
|
||||
}
|
||||
return a, nil
|
||||
return func(ctx context.Context, asked meshcli.Asked) cli.Answer {
|
||||
return meshcli.AskTheController(ctx, queue.AskController, asked)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261009100314-bfe82315f42c
|
||||
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
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261009100314-bfe82315f42c h1:8hjQ3FYt7mGdZv+HX0Vk+n3pdaS3W5EIlwsrvSXXhRY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.8-0.20261009100314-bfe82315f42c/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
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=
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
package meshcli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/cli"
|
||||
)
|
||||
|
||||
// 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).
|
||||
|
||||
// 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 {
|
||||
if errors.Is(err, context.DeadlineExceeded) || ctx.Err() != nil {
|
||||
return cli.NotKnown("", "no answer from the controller in time")
|
||||
}
|
||||
// Never delivered to a controller: no link, no subscriber, the bus refusing the publish.
|
||||
return cli.Refusal("the controller was not reached, so nothing ran: " + 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,98 @@
|
||||
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("nothing on this machine answers mesh.control.laptop.cli")
|
||||
}}}
|
||||
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, errors.New("this machine has no link to the bus open now")
|
||||
}}}
|
||||
if a := AskTheController(context.Background(), c.ask, Asked{Line: []string{"status"}}); !strings.Contains(a.Refused, "nothing ran") {
|
||||
t.Fatalf("no link reads %+v", a)
|
||||
}
|
||||
}
|
||||
+69
-27
@@ -28,28 +28,65 @@ import (
|
||||
// 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"`
|
||||
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 is answered. An error is said to the program as a refusal: nothing ran.
|
||||
type Forward func(ctx context.Context, asked Asked) (cli.Answer, error)
|
||||
// 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.
|
||||
const readWithin = 30 * time.Second
|
||||
|
||||
// atOnce bounds how many lines are asked at the same time. A person types one at a time; more than this is a
|
||||
// program on the machine, and it waits.
|
||||
const atOnce = 8
|
||||
// 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
|
||||
)
|
||||
|
||||
// peerOf reads the connecting process's credentials from the kernel. A variable so a test can say what the kernel
|
||||
// would; it is the only source of the account.
|
||||
// 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))
|
||||
@@ -99,7 +136,7 @@ func Serve(ctx context.Context, socket string, forward Forward, say func(string)
|
||||
<-ctx.Done()
|
||||
l.Close()
|
||||
}()
|
||||
slots := make(chan struct{}, atOnce)
|
||||
held := &slots{byUID: map[uint32]int{}}
|
||||
var wg sync.WaitGroup
|
||||
defer wg.Wait()
|
||||
for {
|
||||
@@ -114,23 +151,19 @@ func Serve(ctx context.Context, socket string, forward Forward, say func(string)
|
||||
}
|
||||
return fmt.Errorf("mesh-cli's socket: %w", err)
|
||||
}
|
||||
select {
|
||||
case slots <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
conn.Close()
|
||||
return nil
|
||||
}
|
||||
// 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()
|
||||
defer func() { <-slots }()
|
||||
answer(ctx, conn, forward, say)
|
||||
answer(ctx, conn, held, forward, say)
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// answer serves one connection.
|
||||
func answer(ctx context.Context, conn *net.UnixConn, forward Forward, say func(string)) {
|
||||
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))
|
||||
@@ -138,29 +171,38 @@ func answer(ctx context.Context, conn *net.UnixConn, forward Forward, say func(s
|
||||
say(fmt.Sprintf("mesh-cli: the answer could not be written back: %v", err))
|
||||
}
|
||||
}
|
||||
uid, err := peerOf(conn)
|
||||
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
|
||||
}
|
||||
_ = conn.SetReadDeadline(time.Now().Add(readWithin))
|
||||
req, err := cli.ReadRequest(conn)
|
||||
if err != nil {
|
||||
reply(cli.Refusal("the request could not be read, so nothing was asked: " + err.Error()))
|
||||
return
|
||||
}
|
||||
asking, cancel := context.WithTimeout(ctx, cli.AskWithin-time.Minute/2)
|
||||
// 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()
|
||||
a, err := forward(asking, Asked{Line: req.Line, Account: account, UID: uid})
|
||||
if err != nil {
|
||||
reply(cli.Refusal("the controller was not reached, so nothing ran: " + err.Error()))
|
||||
return
|
||||
}
|
||||
reply(a)
|
||||
reply(forward(asking, Asked{Line: req.Line, Account: account, UID: uid}))
|
||||
}
|
||||
|
||||
@@ -42,9 +42,9 @@ func serving(t *testing.T, forward Forward) string {
|
||||
// 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, error) {
|
||||
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"}, nil
|
||||
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 {
|
||||
@@ -71,12 +71,12 @@ func TestTheAccountAskingIsTheKernelsAndTheAnswerComesBack(t *testing.T) {
|
||||
// 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, error) { return 0, errors.New("no credentials") }
|
||||
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, error) {
|
||||
socket := serving(t, func(context.Context, Asked) cli.Answer {
|
||||
asked = true
|
||||
return cli.Answer{}, nil
|
||||
return cli.Answer{}
|
||||
})
|
||||
a, err := cli.Ask(context.Background(), socket, []string{"status"})
|
||||
if err != nil {
|
||||
@@ -87,26 +87,12 @@ func TestAConnectionWhoseAccountCannotBeReadIsNeverAskedAsAnyone(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A controller out of reach is said: nothing ran, and why.
|
||||
func TestAControllerOutOfReachIsSaidAsNothingRan(t *testing.T) {
|
||||
socket := serving(t, func(context.Context, Asked) (cli.Answer, error) {
|
||||
return cli.Answer{}, errors.New("this machine has no link to the bus open now")
|
||||
})
|
||||
a, err := cli.Ask(context.Background(), socket, []string{"status"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(a.Refused, "nothing ran") || !strings.Contains(a.Refused, "no link") || a.Exit != 1 {
|
||||
t.Fatalf("an unreached controller was not said: %+v", 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, error) {
|
||||
socket := serving(t, func(context.Context, Asked) cli.Answer {
|
||||
asked = true
|
||||
return cli.Answer{}, nil
|
||||
return cli.Answer{}
|
||||
})
|
||||
a, err := cli.Ask(context.Background(), socket, []string{strings.Repeat("x", cli.MaxRequest)})
|
||||
if err != nil {
|
||||
@@ -143,6 +129,11 @@ func TestTheRequestToTheControllerKeepsItsFieldNames(t *testing.T) {
|
||||
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 {
|
||||
@@ -151,6 +142,10 @@ func TestTheRequestToTheControllerKeepsItsFieldNames(t *testing.T) {
|
||||
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 {
|
||||
@@ -166,3 +161,58 @@ func keysOf(t *testing.T, body []byte) string {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,25 +12,44 @@ import (
|
||||
|
||||
// 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, error) {
|
||||
func peerCredentials(conn *net.UnixConn) (uint32, int32, error) {
|
||||
raw, err := conn.SyscallConn()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
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, err
|
||||
return 0, 0, err
|
||||
}
|
||||
if inner != nil {
|
||||
return 0, inner
|
||||
return 0, 0, inner
|
||||
}
|
||||
if cred == nil {
|
||||
return 0, errors.New("the kernel gave no credentials")
|
||||
return 0, 0, errors.New("the kernel gave no credentials")
|
||||
}
|
||||
return cred.Uid, nil
|
||||
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
|
||||
|
||||
@@ -8,9 +8,12 @@ import (
|
||||
)
|
||||
|
||||
// peerCredentials is unknown where the kernel's way of saying it is not built here: refused, never guessed.
|
||||
func peerCredentials(*net.UnixConn) (uint32, error) {
|
||||
return 0, errors.New("this engine reads a peer's account on Linux only")
|
||||
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