Revert L3 (mesh-host, merge 0a4c35f): its node-engine removed every held module on novox, the bus server among them, at 08:54 CET on 2026-10-29 (issue 774)
L3 lands again with two fixes: a held module is never removed, and the grants cover L3's new subjects.
This commit is contained in:
@@ -6,10 +6,6 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asker"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
)
|
||||
|
||||
// The take act's ask (novox/hq ADR 0325 §8): a hand-over's fields and the act's own in one signed object, under a
|
||||
@@ -72,112 +68,6 @@ func TestTheTakeActRunsOnlyOnAWarrant(t *testing.T) {
|
||||
if a.Said != "started" || len(given) != 1 || given[0].Warrant != "w-1" || given[0].Path != "/srv/media/movies" {
|
||||
t.Fatalf("a warranted ask did not reach the act whole: %+v %+v", a, given)
|
||||
}
|
||||
// 774 (ADR 0563 §3): a hand-over asked through the node seat is the engine's own ask, performed only on the
|
||||
// operator's warrant to it: never on the call, not on a warrant for another digest, once on a warrant heard twice,
|
||||
// and not when the disk changed since it was asked.
|
||||
t.Run("the node seat's hand-over acts only on the operator's warrant", func(t *testing.T) {
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
warrant func(asked asks.Ask, w asks.Warrant) asks.Warrant
|
||||
diskAfter string
|
||||
twice bool
|
||||
done int
|
||||
state string
|
||||
}{
|
||||
{name: "no warrant", done: 0, state: HandAsked},
|
||||
{name: "the operator's yes", warrant: yes, done: 1, state: HandDone},
|
||||
{name: "the yes heard twice", warrant: yes, twice: true, done: 1, state: HandDone},
|
||||
{name: "a warrant for another digest", warrant: func(a asks.Ask, w asks.Warrant) asks.Warrant {
|
||||
w = yes(a, w)
|
||||
w.AskDigest = "sha256:another"
|
||||
return w
|
||||
}, done: 0, state: HandAsked},
|
||||
{name: "a no", warrant: func(a asks.Ask, w asks.Warrant) asks.Warrant {
|
||||
w = yes(a, w)
|
||||
w.Option, w.Label, w.Level = "no", "No, leave it", asks.Acknowledge
|
||||
return w
|
||||
}, done: 0, state: HandRefused},
|
||||
{name: "the disk changed", warrant: yes, diskAfter: "0:0 700 9:2", done: 0, state: HandRefused},
|
||||
{name: "expired", warrant: func(a asks.Ask, w asks.Warrant) asks.Warrant {
|
||||
return asks.Warrant{Ask: a.ID, Asker: w.Asker, Outcome: asks.OutcomeExpired, At: w.At}
|
||||
}, done: 0, state: HandRefused},
|
||||
} {
|
||||
t.Run(row.name, func(t *testing.T) {
|
||||
h, asked, done := anEngine(t)
|
||||
disk := "1000:1000 755 9:1"
|
||||
h.JudgeHandOver = func(path string) (Judged, error) {
|
||||
return Judged{Subject: path, Path: path, Sets: "owner jo, group jo, mode 0750", Disk: disk,
|
||||
Words: "A directory used as found."}, nil
|
||||
}
|
||||
if _, err := h.Call(EngineCall{Verb: ActHandOver, Caller: "shanks.node-tools",
|
||||
Args: map[string]any{"path": "/srv/x", "why": "a test"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*asked) != 1 || *done != 0 {
|
||||
t.Fatalf("the call asked %d and performed %d", len(*asked), *done)
|
||||
}
|
||||
// The act is one key, and the open ask one key made by a create (ADR 0563 §3).
|
||||
if acts, _ := h.State.Keys("act."); len(acts) != 1 || acts[0] != "act."+(*asked)[0].ID {
|
||||
t.Fatalf("the act is kept as %v", acts)
|
||||
}
|
||||
if open, _ := h.State.Keys("open."); len(open) != 1 || !strings.HasPrefix(open[0], "open.hand-over.") {
|
||||
t.Fatalf("the open ask is kept as %v", open)
|
||||
}
|
||||
if _, err := h.Call(EngineCall{Verb: ActHandOver, Caller: "x", Args: map[string]any{"path": "/srv/x"}}); err == nil {
|
||||
t.Fatal("a second ask of the same path was asked while one is open")
|
||||
}
|
||||
if row.diskAfter != "" {
|
||||
disk = row.diskAfter
|
||||
}
|
||||
if row.warrant != nil {
|
||||
w := row.warrant((*asked)[0], asks.Warrant{Asker: asks.EngineAsker("n1"), At: time.Now()})
|
||||
h.Heard(w)
|
||||
if row.twice {
|
||||
h.Heard(w)
|
||||
}
|
||||
}
|
||||
got, _ := h.Acts(0)
|
||||
acts := got.(map[string]any)["acts"].([]HandAct)
|
||||
if *done != row.done || len(acts) != 1 || acts[0].State != row.state {
|
||||
t.Fatalf("performed %d (want %d), acts %+v (want %s)", *done, row.done, acts, row.state)
|
||||
}
|
||||
if open, _ := h.State.Keys("open."); (len(open) == 1) != (row.state == HandAsked) {
|
||||
t.Fatalf("an ask %s leaves its open key as %v", row.state, open)
|
||||
}
|
||||
if row.state == HandDone && (acts[0].By != "operator" || acts[0].Via == "" || acts[0].Ask == "") {
|
||||
t.Fatalf("the act does not record the warrant: %+v", acts[0])
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// yes is the operator's yes to an ask, as the router writes it.
|
||||
func yes(a asks.Ask, w asks.Warrant) asks.Warrant {
|
||||
w.Ask, w.Outcome, w.Option, w.Label, w.Level = a.ID, asks.OutcomeChosen, "yes", "Yes", asks.Approve
|
||||
w.By = &asks.Person{Who: asks.Operator, Kind: "telegram", Verified: "user id verified"}
|
||||
w.AskDigest, w.Proofs = a.Digest(), []string{"P1"}
|
||||
return w
|
||||
}
|
||||
|
||||
// anEngine is a node seat with its own state in a temporary folder, a router that keeps what it was asked, and a
|
||||
// hand-over that counts what it performed.
|
||||
func anEngine(t *testing.T) (*HandActs, *[]asks.Ask, *int) {
|
||||
t.Helper()
|
||||
var asked []asks.Ask
|
||||
done := 0
|
||||
state := &MemState{}
|
||||
h := &HandActs{Node: "n1", State: state,
|
||||
HandOver: func(path, by string) (string, error) { done++; return path + " handed over by " + by, nil }}
|
||||
h.Asker = &asker.Client{Module: asks.EngineAsker("n1"), State: state,
|
||||
Publish: func(subject string, body any, id string) error {
|
||||
if a, ok := body.(asks.Ask); ok && subject == asks.AskSubject(asks.EngineAsker("n1")) {
|
||||
asked = append(asked, a)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
Record: func(string) (*stdio.StateEntry, error) { return nil, nil }}
|
||||
return h, &asked, &done
|
||||
}
|
||||
|
||||
// The controller names an access by its declared resource id exactly, which may hold a `/`: it reaches the act,
|
||||
|
||||
@@ -1,382 +0,0 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/micro"
|
||||
)
|
||||
|
||||
// The node seat node-engine, served on the engine's own link (novox/hq ADR 0563 §1).
|
||||
//
|
||||
// Each verb is answered on this node's subject alone, `mesh.seat.node-engine.tool.<verb>.<node>`, and on the same
|
||||
// address naming its caller, `mesh.seat.node-engine.call.<verb>.<node>.<caller>`, whose last token the bus lets each user
|
||||
// publish with its own name alone (novox/hq issue 365): the caller is that token, and a call on the subject naming none is
|
||||
// from an unproven caller, recorded so. The body is the call's arguments; the answer is {result, error, node}, as every
|
||||
// tool's (mesh-tools node-tools). One service, `node-engine`, answers discovery with each verb's seat, tool and node.
|
||||
|
||||
// EngineSeat is the seat.
|
||||
const EngineSeat = catalogue.EngineSeat
|
||||
|
||||
// EngineToolSubject is where a verb of this node's seat is asked naming no caller.
|
||||
func EngineToolSubject(verb, node string) string {
|
||||
return "mesh.seat." + EngineSeat + ".tool." + verb + "." + node
|
||||
}
|
||||
|
||||
// EngineCallPattern is where a verb of this node's seat is asked naming its caller, any caller.
|
||||
func EngineCallPattern(verb, node string) string {
|
||||
return "mesh.seat." + EngineSeat + ".call." + verb + "." + node + ".*"
|
||||
}
|
||||
|
||||
// EngineEventSubject is where the seat says one of its events about this node.
|
||||
func EngineEventSubject(event, node string) string {
|
||||
return "mesh.seat." + EngineSeat + ".event." + event + "." + node
|
||||
}
|
||||
|
||||
// EngineCall is one call of the seat: its verb, its caller as the bus proved it, and its arguments.
|
||||
type EngineCall struct {
|
||||
Verb string
|
||||
Caller string
|
||||
Args map[string]any
|
||||
}
|
||||
|
||||
// EngineFunc answers one call: its result, or why it is refused.
|
||||
type EngineFunc func(EngineCall) (any, error)
|
||||
|
||||
// CallerOfSubject is the caller a call's subject names for this node's verb: the last token of a caller-named subject,
|
||||
// each `~` a dot, or stdio.Unproven on the subject that names none or on any other.
|
||||
func CallerOfSubject(subject, verb, node string) string {
|
||||
token, ok := strings.CutPrefix(subject, "mesh.seat."+EngineSeat+".call."+verb+"."+node+".")
|
||||
if !ok || token == "" || strings.ContainsAny(token, ".*> ") {
|
||||
return stdio.Unproven
|
||||
}
|
||||
return strings.ReplaceAll(token, "~", ".")
|
||||
}
|
||||
|
||||
// MaxEngineCall bounds a call's arguments: a path, an access and a few words.
|
||||
const MaxEngineCall = 16 << 10
|
||||
|
||||
// AnswerEngineCall is the envelope one call is answered with: the verb's result, or its refusal in words.
|
||||
func AnswerEngineCall(handle EngineFunc, verb, node, subject string, body []byte) []byte {
|
||||
envelope := map[string]any{"node": node}
|
||||
say := func() []byte {
|
||||
raw, err := json.Marshal(envelope)
|
||||
if err != nil {
|
||||
raw, _ = json.Marshal(map[string]any{"node": node, "error": "the answer could not be written: " + err.Error()})
|
||||
}
|
||||
return raw
|
||||
}
|
||||
args := map[string]any{}
|
||||
switch {
|
||||
case handle == nil:
|
||||
envelope["error"] = "this engine serves no verb of " + EngineSeat
|
||||
return say()
|
||||
case len(body) > MaxEngineCall:
|
||||
envelope["error"] = fmt.Sprintf("the call is %d bytes, more than a call of %s takes", len(body), EngineSeat)
|
||||
return say()
|
||||
case len(strings.TrimSpace(string(body))) > 0:
|
||||
if err := json.Unmarshal(body, &args); err != nil {
|
||||
envelope["error"] = "the call's arguments are not an object: " + err.Error()
|
||||
return say()
|
||||
}
|
||||
}
|
||||
result, err := handle(EngineCall{Verb: verb, Caller: CallerOfSubject(subject, verb, node), Args: args})
|
||||
if err != nil {
|
||||
envelope["error"] = err.Error()
|
||||
} else {
|
||||
envelope["result"] = result
|
||||
}
|
||||
return say()
|
||||
}
|
||||
|
||||
// EngineSeats is a link that serves this node's seat. Optional, as Asked is.
|
||||
type EngineSeats interface {
|
||||
ServeEngineSeat(EngineFunc)
|
||||
}
|
||||
|
||||
// ServeEngineSeat answers every verb of the seat on this node's subjects, as the service node-engine, which says each
|
||||
// verb's seat, tool and node to discovery (novox/hq ADR 0197). A bus whose user list is older than the grant refuses the
|
||||
// subscriptions, said by the client; a caller then hears nobody answer.
|
||||
func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
|
||||
if handle == nil || l.conn == nil {
|
||||
return
|
||||
}
|
||||
svc, err := micro.AddService(l.conn, micro.Config{Name: EngineSeat, Version: "1.0.0",
|
||||
Description: "the node seat node-engine of " + l.node + " (novox/hq ADR 0563)", QueueGroupDisabled: true})
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, v := range catalogue.EngineVerbs() {
|
||||
verb := v.Name
|
||||
answer := micro.HandlerFunc(func(r micro.Request) {
|
||||
_ = r.Respond(AnswerEngineCall(handle, verb, l.node, r.Subject(), r.Data()))
|
||||
})
|
||||
meta := map[string]string{"seat": EngineSeat, "tool": verb, "node": l.node}
|
||||
_ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)),
|
||||
micro.WithEndpointMetadata(meta))
|
||||
_ = svc.AddEndpoint(verb+"-called", answer, micro.WithEndpointSubject(EngineCallPattern(verb, l.node)),
|
||||
micro.WithEndpointMetadata(meta))
|
||||
}
|
||||
l.engine = svc
|
||||
}
|
||||
|
||||
// SayEngine says one of the seat's events about this node, on the events stream.
|
||||
func (l *natsLink) SayEngine(event string, body any) error {
|
||||
raw, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
_, err = l.js.Publish(EngineEventSubject(event, l.node), raw, nats.Context(ctx))
|
||||
return err
|
||||
}
|
||||
|
||||
// The engine asks the operator under its own asker name (asks.EngineAsker) on the router's seat (ADR 0259 §3, as ADR
|
||||
// 0563 §3 changes it): its ask and cancel on the seat's work stream, its warrants heard on its own decided subject, and
|
||||
// its asks' state read from the router's records under its own name.
|
||||
|
||||
// RouterRecords is the router's bucket of asks, which each asker reads under its own name: the state `asks` of the
|
||||
// operator-channel's holder, the messenger (mesh-catalog modules/messenger).
|
||||
const RouterRecords = "messenger_asks"
|
||||
|
||||
// OperatorAsks is a link that asks the operator as this node's engine. Optional.
|
||||
type OperatorAsks interface {
|
||||
PublishAsk(subject string, body any, id string) error
|
||||
RouterRecord(key string) (*stdio.StateEntry, error)
|
||||
HearWarrants(asker string, heard func(asks.Warrant)) error
|
||||
}
|
||||
|
||||
// PublishAsk publishes an ask or a cancel on the router's seat, once per id within the stream's duplicate window.
|
||||
func (l *natsLink) PublishAsk(subject string, body any, id string) error {
|
||||
raw, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
_, err = l.js.Publish(subject, raw, nats.MsgId(id), nats.Context(ctx))
|
||||
return err
|
||||
}
|
||||
|
||||
// RouterRecord reads one key of the router's records, `<asker>.<id>`; nil when it holds none.
|
||||
func (l *natsLink) RouterRecord(key string) (*stdio.StateEntry, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
msg, err := l.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+RouterRecords+".$KV."+RouterRecords+"."+key, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if status := msg.Header.Get("Status"); status == "404" {
|
||||
return nil, nil
|
||||
} else if status != "" {
|
||||
return nil, fmt.Errorf("the router's record %s could not be read: %s %s", key, status, msg.Header.Get("Description"))
|
||||
}
|
||||
return &stdio.StateEntry{Key: key, Value: msg.Data}, nil
|
||||
}
|
||||
|
||||
// HearWarrants hears the warrants for asker on its decided subject, each handed to heard as it arrives.
|
||||
func (l *natsLink) HearWarrants(asker string, heard func(asks.Warrant)) error {
|
||||
sub, err := l.conn.Subscribe(asks.DecidedSubject(asker), func(msg *nats.Msg) {
|
||||
var w asks.Warrant
|
||||
if json.Unmarshal(msg.Data, &w) == nil {
|
||||
heard(w)
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
l.warrants = sub
|
||||
return nil
|
||||
}
|
||||
|
||||
// errNoAsking is a link that cannot ask the operator.
|
||||
var errNoAsking = errors.New("the link open now cannot ask the operator")
|
||||
|
||||
// discoveryQuiet is how long the engine waits after the last answer to the bus's discovery, and discoveryMost the
|
||||
// longest it waits in all.
|
||||
const (
|
||||
discoveryQuiet = 300 * time.Millisecond
|
||||
discoveryMost = 2 * time.Second
|
||||
)
|
||||
|
||||
// HeardServed asks the bus's discovery (novox/hq ADR 0197) whether a seat's verb is answered on this machine: an
|
||||
// endpoint naming the seat, the verb as its tool and this node.
|
||||
func (l *natsLink) HeardServed(seat, verb string) (bool, error) {
|
||||
inbox := l.conn.NewRespInbox()
|
||||
sub, err := l.conn.SubscribeSync(inbox)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
defer func() { _ = sub.Unsubscribe() }()
|
||||
if err := l.conn.PublishRequest("$SRV.INFO", inbox, nil); err != nil {
|
||||
return false, fmt.Errorf("asking the bus who serves what: %w", err)
|
||||
}
|
||||
heard := false
|
||||
deadline := time.Now().Add(discoveryMost)
|
||||
for time.Now().Before(deadline) {
|
||||
msg, err := sub.NextMsg(discoveryQuiet)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
var info micro.Info
|
||||
if json.Unmarshal(msg.Data, &info) != nil {
|
||||
continue
|
||||
}
|
||||
for _, e := range info.Endpoints {
|
||||
node := e.Metadata["node"]
|
||||
if node == "" {
|
||||
node = info.ID
|
||||
}
|
||||
if e.Metadata["seat"] == seat && e.Metadata["tool"] == verb && node == l.node {
|
||||
heard = true
|
||||
}
|
||||
}
|
||||
}
|
||||
return heard, nil
|
||||
}
|
||||
|
||||
// HeardServed is the link open now asking the bus's discovery whether a seat's verb is answered on this machine.
|
||||
func (q *Session) HeardServed(seat, verb string) (bool, error) {
|
||||
q.init()
|
||||
q.mu.Lock()
|
||||
bus := q.bus
|
||||
q.mu.Unlock()
|
||||
asker, ok := bus.(interface {
|
||||
HeardServed(string, string) (bool, error)
|
||||
})
|
||||
if bus == nil || !ok {
|
||||
return false, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked)
|
||||
}
|
||||
return asker.HeardServed(seat, verb)
|
||||
}
|
||||
|
||||
// ---- the engine's acts in the bus's own storage (ADR 0563 §2, §3) -------------------------------------------------
|
||||
|
||||
// actsStore is this node's keys of ActsBucket over the link open now: bound, never created (the grants part creates
|
||||
// it, as the node-engine's declared state).
|
||||
func (l *natsLink) actsStore() (nats.KeyValue, error) {
|
||||
kv, err := l.js.KeyValue(ActsBucket)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the node-engine's state %s cannot be bound (it is created from the node-engine's "+
|
||||
"manifest; asked again once it is): %w", ActsBucket, err)
|
||||
}
|
||||
return kv, nil
|
||||
}
|
||||
|
||||
// BusState is the engine's ActStore: its own node's keys of ActsBucket, `<node>.<key>`, over whichever link the session
|
||||
// has open; with none open every read and write is refused, and nothing is kept anywhere else.
|
||||
type BusState struct {
|
||||
Session *Session
|
||||
Node string
|
||||
}
|
||||
|
||||
func (b BusState) kv() (nats.KeyValue, error) {
|
||||
b.Session.init()
|
||||
b.Session.mu.Lock()
|
||||
bus := b.Session.bus
|
||||
b.Session.mu.Unlock()
|
||||
l, ok := bus.(interface{ actsStore() (nats.KeyValue, error) })
|
||||
if bus == nil || !ok {
|
||||
return nil, fmt.Errorf("%w: this machine has no link to the bus open now, where its acts are kept", ErrNotAsked)
|
||||
}
|
||||
return l.actsStore()
|
||||
}
|
||||
|
||||
func (b BusState) key(k string) string { return b.Node + "." + k }
|
||||
|
||||
// changed is a write that lost its compare-and-set, as stdio.ErrChanged.
|
||||
func changed(err error) error {
|
||||
var api *nats.APIError
|
||||
if errors.Is(err, nats.ErrKeyExists) || (errors.As(err, &api) && api.ErrorCode == nats.JSErrCodeStreamWrongLastSequence) {
|
||||
return stdio.ErrChanged
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (b BusState) Get(key string) (*stdio.StateEntry, error) {
|
||||
kv, err := b.kv()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
e, err := kv.Get(b.key(key))
|
||||
if errors.Is(err, nats.ErrKeyNotFound) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &stdio.StateEntry{Key: key, Value: e.Value(), Revision: e.Revision()}, nil
|
||||
}
|
||||
|
||||
func (b BusState) Create(key string, value any) (uint64, error) {
|
||||
kv, err := b.kv()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
raw, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Create(b.key(key), raw)
|
||||
return rev, changed(err)
|
||||
}
|
||||
|
||||
func (b BusState) Update(key string, value any, revision uint64) (uint64, error) {
|
||||
kv, err := b.kv()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
raw, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Update(b.key(key), raw, revision)
|
||||
return rev, changed(err)
|
||||
}
|
||||
|
||||
func (b BusState) Delete(key string) error {
|
||||
kv, err := b.kv()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return kv.Delete(b.key(key))
|
||||
}
|
||||
|
||||
// Keys are this node's keys under prefix, without the node: a watch of `<node>.>` alone, the one filter the node's
|
||||
// grant names, its values left out.
|
||||
func (b BusState) Keys(prefix string) ([]string, error) {
|
||||
kv, err := b.kv()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
w, err := kv.Watch(b.Node+".>", nats.MetaOnly(), nats.IgnoreDeletes())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = w.Stop() }()
|
||||
var out []string
|
||||
for {
|
||||
select {
|
||||
case e := <-w.Updates():
|
||||
if e == nil {
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
}
|
||||
if k := strings.TrimPrefix(e.Key(), b.Node+"."); strings.HasPrefix(k, prefix) {
|
||||
out = append(out, k)
|
||||
}
|
||||
case <-time.After(10 * time.Second):
|
||||
return nil, errors.New("the node-engine's acts could not be listed within 10s")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,769 +0,0 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asker"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
)
|
||||
|
||||
// The node seat node-engine's acts (novox/hq ADR 0563).
|
||||
//
|
||||
// A hand-over and an access act are the engine's own asks, performed only on the warrant (§3): a call performs nothing.
|
||||
// The engine judges every value as it does for the controller's signed ask, then refuses in words or asks the operator
|
||||
// under its own asker name (asks.EngineAsker) at the level approve, binding the digest of the verb, the node, every
|
||||
// argument and the owner, group and mode it will set. A warrant is claimed once by compare-and-set in the engine's own
|
||||
// state (asker.Client.Take), held to the ask it answers, and to the act bound and the disk as judged, judged again now;
|
||||
// only then is the act performed and recorded with the warrant. A decline, an expiry or a changed disk says refused and
|
||||
// changes nothing. One ask is open per path or access.
|
||||
//
|
||||
// The read-only verbs need no warrant (§2): judge-directories changes nothing; setuid-search is bounded here, one at a
|
||||
// time and at most one started in an hour. Every act, with who asked as the runtime proved it, is kept in the engine's own
|
||||
// declared state acts (ActsBucket, created by the grants part), under its own node's keys alone, read by the verb acts.
|
||||
|
||||
// The acts of the seat, as its verbs name them.
|
||||
const (
|
||||
ActHandOver = "hand-over"
|
||||
ActAccess = "access-act"
|
||||
ActJudgeDirectories = "judge-directories"
|
||||
ActSetuidSearch = "setuid-search"
|
||||
)
|
||||
|
||||
// The states of an act.
|
||||
const (
|
||||
HandAsked = "asked"
|
||||
HandDone = "done"
|
||||
HandRefused = "refused"
|
||||
HandAnswered = "answered"
|
||||
)
|
||||
|
||||
// SearchEvery is how often a setuid search may be started on a node (ADR 0563 §2): one in any hour.
|
||||
const SearchEvery = time.Hour
|
||||
|
||||
// askMargin is how long past an ask's expiry the router is given to say it ended, before the engine closes it itself.
|
||||
const askMargin = 5 * time.Minute
|
||||
|
||||
// ActsBucket is the node-engine's declared state acts (its manifest's state, ADR 0563 §2, §3), the bucket the grants
|
||||
// part creates (mesh-sdk buscompose EngineActsBucket): keys `<node>.act.<id>`, `<node>.open.<act>.<digest>` and
|
||||
// `<node>.ask.<id>`, each written by this node's engine alone.
|
||||
const ActsBucket = "mesh-host_acts"
|
||||
|
||||
// Judged is what the engine judged of one act before it asks: what it is about, what it will set (the owner, group and
|
||||
// mode, in words), and what the disk holds now (Disk), compared again when the warrant arrives.
|
||||
type Judged struct {
|
||||
// Subject is the path of a hand-over, or the access of an access act.
|
||||
Subject string
|
||||
// Path is the directory the act changes, as the engine states it.
|
||||
Path string
|
||||
// Sets is the owner, group and mode the act will set, in words; bound in the ask's digest.
|
||||
Sets string
|
||||
// Disk is what the engine read on disk when it judged: owner, group, mode and the file it is.
|
||||
Disk string
|
||||
// Words is what the operator is shown of it, without a path (ADR 0234 §6): the path is in the whole words alone.
|
||||
Words string
|
||||
}
|
||||
|
||||
// HandAct is one act asked through the seat, as acts reads it.
|
||||
type HandAct struct {
|
||||
ID string `json:"id"`
|
||||
Act string `json:"act"`
|
||||
Subject string `json:"subject,omitempty"`
|
||||
Caller string `json:"caller"`
|
||||
Why string `json:"why,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
State string `json:"state"`
|
||||
Said string `json:"said,omitempty"`
|
||||
Ended time.Time `json:"ended,omitempty"`
|
||||
// Ask is the operator's ask of a hand-over or an access act; Sets and Disk what was judged when it was asked.
|
||||
Ask string `json:"ask,omitempty"`
|
||||
Sets string `json:"sets,omitempty"`
|
||||
Disk string `json:"disk,omitempty"`
|
||||
Path string `json:"path,omitempty"`
|
||||
// Expires is when the ask ends unanswered.
|
||||
Expires time.Time `json:"expires,omitempty"`
|
||||
// By, Via, Proofs are the operator's warrant, recorded with the act (ADR 0563 §3).
|
||||
By string `json:"by,omitempty"`
|
||||
Via string `json:"via,omitempty"`
|
||||
Proofs []string `json:"proofs,omitempty"`
|
||||
// Auto says the engine asked it itself, on finding a library's root holding data outside its group (§4).
|
||||
Auto bool `json:"auto,omitempty"`
|
||||
}
|
||||
|
||||
// RootVerdict is the engine's answer to root-free (ADR 0563 §6), in the controller's words.
|
||||
type RootVerdict struct {
|
||||
Machine string `json:"machine"`
|
||||
Free bool `json:"free"`
|
||||
Why string `json:"why"`
|
||||
Judged time.Time `json:"judged"`
|
||||
}
|
||||
|
||||
// HandActs is the engine's side of the seat. Every func is the engine's own judgement or act; nil refuses.
|
||||
type HandActs struct {
|
||||
Node string
|
||||
// State is where the acts and the asks are kept: this node's keys of ActsBucket (BusState), each key one act.
|
||||
State ActStore
|
||||
// Asker asks the operator as asks.EngineAsker(Node), with State as its store.
|
||||
Asker *asker.Client
|
||||
// JudgeHandOver judges a hand-over's path, refusing in words; HandOver records it on the warrant.
|
||||
JudgeHandOver func(path string) (Judged, error)
|
||||
HandOver HandOverFunc
|
||||
// JudgeAccess judges an access act on an access, refusing in words; AccessAct runs it on the warrant.
|
||||
JudgeAccess func(act, access string) (Judged, error)
|
||||
AccessAct AccessActFunc
|
||||
// JudgeDirectories, SetuidSearch and Searching are the read-only verbs' and whether a search runs now.
|
||||
JudgeDirectories JudgeDirectoriesFunc
|
||||
SetuidSearch SetuidSearchFunc
|
||||
Searching func() bool
|
||||
// RootFree is this machine's root-free verdict now.
|
||||
RootFree func() RootVerdict
|
||||
// Say says one of the seat's events, asked, done or refused; nil says nothing.
|
||||
Say func(event string, body any)
|
||||
Now func() time.Time
|
||||
// Nonce keeps two asks of one second apart; nil is random.
|
||||
Nonce func() string
|
||||
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func (h *HandActs) now() time.Time {
|
||||
if h.Now != nil {
|
||||
return h.Now()
|
||||
}
|
||||
return time.Now()
|
||||
}
|
||||
|
||||
func (h *HandActs) nonce() string {
|
||||
if h.Nonce != nil {
|
||||
return h.Nonce()
|
||||
}
|
||||
var b [2]byte
|
||||
_, _ = rand.Read(b[:])
|
||||
return hex.EncodeToString(b[:])
|
||||
}
|
||||
|
||||
func (h *HandActs) say(event string, a HandAct) {
|
||||
if h.Say != nil {
|
||||
h.Say(event, map[string]any{"act": a.Act, "subject": a.Subject, "caller": a.Caller, "ask": a.Ask,
|
||||
"state": a.State, "said": a.Said, "node": h.Node})
|
||||
}
|
||||
}
|
||||
|
||||
// Call is the seat's one entry: a verb, the caller as the runtime proved it, and its arguments.
|
||||
func (h *HandActs) Call(c EngineCall) (any, error) {
|
||||
arg := func(k string) string { s, _ := c.Args[k].(string); return strings.TrimSpace(s) }
|
||||
switch c.Verb {
|
||||
case ActHandOver:
|
||||
return h.AskHandOver(c.Caller, arg("path"), arg("why"), false)
|
||||
case ActAccess:
|
||||
return h.AskAccess(c.Caller, arg("act"), arg("access"), arg("why"), false)
|
||||
case ActJudgeDirectories:
|
||||
return h.Judge(c.Caller)
|
||||
case ActSetuidSearch:
|
||||
return h.Search(c.Caller, arg("why"))
|
||||
case "root-free":
|
||||
if h.RootFree == nil {
|
||||
return nil, errors.New("this engine judges no machine root-free")
|
||||
}
|
||||
return map[string]any{"machines": []RootVerdict{h.RootFree()}}, nil
|
||||
case "acts":
|
||||
if id := arg("id"); id != "" {
|
||||
return h.Act(id)
|
||||
}
|
||||
limit, _ := strconv.Atoi(arg("limit"))
|
||||
if f, ok := c.Args["limit"].(float64); ok {
|
||||
limit = int(f)
|
||||
}
|
||||
return h.Acts(limit)
|
||||
}
|
||||
return nil, fmt.Errorf("the node seat %s serves no verb %q", EngineSeat, c.Verb)
|
||||
}
|
||||
|
||||
// ---- the record --------------------------------------------------------------------------------------------------
|
||||
|
||||
// ActStore is the engine's own state, this node's keys alone (BusState prefixes them with the node): each act one key,
|
||||
// written by compare-and-set on its revision, an open ask one key made by a create.
|
||||
type ActStore interface {
|
||||
Get(key string) (*stdio.StateEntry, error)
|
||||
Create(key string, value any) (uint64, error)
|
||||
Update(key string, value any, revision uint64) (uint64, error)
|
||||
Delete(key string) error
|
||||
// Keys are the keys under prefix.
|
||||
Keys(prefix string) ([]string, error)
|
||||
}
|
||||
|
||||
func actKey(id string) string { return "act." + id }
|
||||
|
||||
// openKey is the one key an open ask of act on subject holds: made by a create, so a second ask of it is refused.
|
||||
func openKey(act, subject string) string {
|
||||
sum := sha256.Sum256([]byte(subject))
|
||||
return "open." + act + "." + hex.EncodeToString(sum[:12])
|
||||
}
|
||||
|
||||
// read is every act kept, oldest first.
|
||||
func (h *HandActs) read() ([]HandAct, error) {
|
||||
keys, err := h.State.Keys("act.")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the engine's acts cannot be listed: %w", err)
|
||||
}
|
||||
var out []HandAct
|
||||
for _, k := range keys {
|
||||
a, _, err := h.readOne(k)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if a != nil {
|
||||
out = append(out, *a)
|
||||
}
|
||||
}
|
||||
sort.SliceStable(out, func(i, j int) bool { return out[i].At.Before(out[j].At) })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (h *HandActs) readOne(key string) (*HandAct, uint64, error) {
|
||||
e, err := h.State.Get(key)
|
||||
if err != nil || e == nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
var a HandAct
|
||||
if err := json.Unmarshal(e.Value, &a); err != nil {
|
||||
return nil, 0, fmt.Errorf("the engine's act %s cannot be read: %w", key, err)
|
||||
}
|
||||
return &a, e.Revision, nil
|
||||
}
|
||||
|
||||
// judgedKey keeps the node's judgements and refused searches, the last keptJudged of them: any caller may ask them, so
|
||||
// they roll in one key and never add one (ADR 0563 §2: a wrong caller gains nothing, and fills no bucket).
|
||||
const (
|
||||
judgedKey = "judged"
|
||||
keptJudged = 50
|
||||
)
|
||||
|
||||
// roll adds a to the rolling key by compare-and-set, the oldest falling out.
|
||||
func (h *HandActs) roll(a HandAct) error {
|
||||
for range 5 {
|
||||
e, err := h.State.Get(judgedKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var kept []HandAct
|
||||
if e != nil {
|
||||
if err := json.Unmarshal(e.Value, &kept); err != nil {
|
||||
kept = nil // an unreadable roll is started again, never a reason to refuse
|
||||
}
|
||||
}
|
||||
kept = append(kept, a)
|
||||
if len(kept) > keptJudged {
|
||||
kept = kept[len(kept)-keptJudged:]
|
||||
}
|
||||
if e == nil {
|
||||
_, err = h.State.Create(judgedKey, kept)
|
||||
} else {
|
||||
_, err = h.State.Update(judgedKey, kept, e.Revision)
|
||||
}
|
||||
if !errors.Is(err, stdio.ErrChanged) {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return errors.New("the engine's judgements kept changing under its write")
|
||||
}
|
||||
|
||||
func (h *HandActs) rolled() ([]HandAct, error) {
|
||||
e, err := h.State.Get(judgedKey)
|
||||
if err != nil || e == nil {
|
||||
return nil, err
|
||||
}
|
||||
var kept []HandAct
|
||||
if err := json.Unmarshal(e.Value, &kept); err != nil {
|
||||
return nil, fmt.Errorf("the engine's judgements cannot be read: %w", err)
|
||||
}
|
||||
return kept, nil
|
||||
}
|
||||
|
||||
func (h *HandActs) record(a HandAct) error {
|
||||
_, err := h.State.Create(actKey(a.ID), a)
|
||||
return err
|
||||
}
|
||||
|
||||
// end marks the open act of an ask ended, once, by compare-and-set on its revision, and frees its open key: false when
|
||||
// it was not open.
|
||||
func (h *HandActs) end(ask string, set func(*HandAct)) (HandAct, bool, error) {
|
||||
for range 5 {
|
||||
a, rev, err := h.readOne(actKey(ask))
|
||||
if err != nil || a == nil || a.State != HandAsked {
|
||||
return HandAct{}, false, err
|
||||
}
|
||||
set(a)
|
||||
a.Ended = h.now().UTC()
|
||||
if _, err := h.State.Update(actKey(ask), *a, rev); errors.Is(err, stdio.ErrChanged) {
|
||||
continue
|
||||
} else if err != nil {
|
||||
return HandAct{}, false, err
|
||||
}
|
||||
_ = h.State.Delete(openKey(a.Act, a.Subject))
|
||||
return *a, true, nil
|
||||
}
|
||||
return HandAct{}, false, errors.New("the engine's act " + ask + " kept changing under its write")
|
||||
}
|
||||
|
||||
// Act is the verb acts with an id: that one act.
|
||||
func (h *HandActs) Act(id string) (any, error) {
|
||||
a, _, err := h.readOne(actKey(id))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if a == nil {
|
||||
return nil, fmt.Errorf("this engine keeps no act %q", id)
|
||||
}
|
||||
return map[string]any{"node": h.Node, "acts": []HandAct{*a}}, nil
|
||||
}
|
||||
|
||||
// Acts is the verb acts: every act kept, newest first, at most limit when given.
|
||||
func (h *HandActs) Acts(limit int) (any, error) {
|
||||
acts, err := h.read()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
judged, err := h.rolled()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
acts = append(acts, judged...)
|
||||
sort.SliceStable(acts, func(i, j int) bool { return acts[i].At.Before(acts[j].At) })
|
||||
out := make([]HandAct, 0, len(acts))
|
||||
for i := len(acts) - 1; i >= 0; i-- {
|
||||
out = append(out, acts[i])
|
||||
}
|
||||
if limit > 0 && len(out) > limit {
|
||||
out = out[:limit]
|
||||
}
|
||||
return map[string]any{"node": h.Node, "acts": out}, nil
|
||||
}
|
||||
|
||||
// ---- the read-only verbs (§2) -------------------------------------------------------------------------------------
|
||||
|
||||
// Judge is judge-directories: the engine's judgement, read-only, recorded with its caller.
|
||||
func (h *HandActs) Judge(caller string) (any, error) {
|
||||
if h.JudgeDirectories == nil {
|
||||
return nil, errors.New("this engine judges no directory")
|
||||
}
|
||||
said, err := h.JudgeDirectories(caller)
|
||||
a := HandAct{ID: h.id(ActJudgeDirectories), Act: ActJudgeDirectories, Caller: caller, At: h.now().UTC(), State: HandAnswered}
|
||||
if err != nil {
|
||||
a.State, a.Said = HandRefused, err.Error()
|
||||
}
|
||||
if rerr := h.roll(a); rerr != nil && err == nil {
|
||||
return nil, rerr
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"node": h.Node, "said": said}, nil
|
||||
}
|
||||
|
||||
// Search is setuid-search: refused while one runs and when one was started within the hour, naming when the next may
|
||||
// start; otherwise the last search is thrown away and a full one starts, recorded with its caller.
|
||||
func (h *HandActs) Search(caller, why string) (any, error) {
|
||||
if h.SetuidSearch == nil {
|
||||
return nil, errors.New("this engine keeps no search for setuid programs")
|
||||
}
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
now := h.now().UTC()
|
||||
if h.Searching != nil && h.Searching() {
|
||||
return nil, errors.New("a search for setuid programs runs on this machine now; no second is started while it does")
|
||||
}
|
||||
// The last start is a key of its own, claimed by compare-and-set before the search starts: no number of other acts,
|
||||
// nor the record of acts, moves it (ADR 0563 §2).
|
||||
if err := h.claimSearch(now, caller); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
said, err := h.SetuidSearch(caller)
|
||||
a := HandAct{ID: h.id(ActSetuidSearch), Act: ActSetuidSearch, Caller: caller, Why: why, At: now, State: HandDone, Said: said}
|
||||
if err != nil {
|
||||
a.State, a.Said = HandRefused, err.Error()
|
||||
}
|
||||
// A search started is bounded to one an hour and kept as an act of its own; a refused one, which any caller may
|
||||
// cause, rolls with the judgements.
|
||||
record := h.record
|
||||
if err != nil {
|
||||
record = h.roll
|
||||
}
|
||||
if rerr := record(a); rerr != nil {
|
||||
return nil, rerr
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"node": h.Node, "said": said}, nil
|
||||
}
|
||||
|
||||
// searchKey keeps the last search's start on this node, and libKey each library's own take ask: keys of their own, which
|
||||
// no other act's key moves, kept for the bucket's age (ninety days, the node-engine's manifest).
|
||||
const searchKey = "search.last"
|
||||
|
||||
func libKey(access string) string {
|
||||
sum := sha256.Sum256([]byte(access))
|
||||
return "lib." + hex.EncodeToString(sum[:12]) + ".asked"
|
||||
}
|
||||
|
||||
// started is the last search's start as kept, and who asked it.
|
||||
type started struct {
|
||||
At time.Time `json:"at"`
|
||||
By string `json:"by"`
|
||||
}
|
||||
|
||||
// claimSearch refuses a search within SearchEvery of the last start, naming when the next may start, and otherwise
|
||||
// claims this one by compare-and-set: a second claim at the same moment loses.
|
||||
func (h *HandActs) claimSearch(now time.Time, caller string) error {
|
||||
e, err := h.State.Get(searchKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var rev uint64
|
||||
if e != nil {
|
||||
var last started
|
||||
if err := json.Unmarshal(e.Value, &last); err != nil {
|
||||
return fmt.Errorf("the last search's start cannot be read, so none is started: %w", err)
|
||||
}
|
||||
if now.Sub(last.At) < SearchEvery {
|
||||
return fmt.Errorf("a search was started at %s by %s, and one is started in any hour: the next may start "+
|
||||
"at %s", last.At.Format(time.RFC3339), last.By, last.At.Add(SearchEvery).Format(time.RFC3339))
|
||||
}
|
||||
rev = e.Revision
|
||||
}
|
||||
claim := started{At: now, By: caller}
|
||||
if rev == 0 {
|
||||
_, err = h.State.Create(searchKey, claim)
|
||||
} else {
|
||||
_, err = h.State.Update(searchKey, claim, rev)
|
||||
}
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
return errors.New("another search was started at this moment; no second is started")
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// ---- the asks (§3, §4) --------------------------------------------------------------------------------------------
|
||||
|
||||
func (h *HandActs) id(act string) string {
|
||||
return act + "-" + h.now().UTC().Format("20060102t150405") + "-" + h.nonce()
|
||||
}
|
||||
|
||||
// bound is the act an ask's yes binds: the verb, the node, every argument and what it will set (ADR 0563 §3).
|
||||
func (h *HandActs) bound(a HandAct) asks.Act {
|
||||
act := asks.Act{"verb": a.Act, "node": h.Node, "arg.subject": a.Subject, "arg.why": a.Why, "sets": a.Sets}
|
||||
if a.Path != "" {
|
||||
act["path"] = a.Path
|
||||
}
|
||||
return act
|
||||
}
|
||||
|
||||
// AskHandOver is hand-over: judged, refused in words or asked of the operator. auto is the engine asking by itself.
|
||||
func (h *HandActs) AskHandOver(caller, path, why string, auto bool) (any, error) {
|
||||
if h.JudgeHandOver == nil || h.HandOver == nil {
|
||||
return nil, errors.New("this engine takes no hand-over")
|
||||
}
|
||||
j, err := h.JudgeHandOver(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return h.ask(ActHandOver, caller, why, auto, j)
|
||||
}
|
||||
|
||||
// AskAccess is access-act: the act take on an access, judged on the library's home, refused or asked of the operator.
|
||||
func (h *HandActs) AskAccess(caller, act, access, why string, auto bool) (any, error) {
|
||||
if h.JudgeAccess == nil || h.AccessAct == nil {
|
||||
return nil, errors.New("this engine runs no act on an accessed folder")
|
||||
}
|
||||
if act != ActTake {
|
||||
return nil, fmt.Errorf("the act asked is %q, and the one act on an access is %q", act, ActTake)
|
||||
}
|
||||
j, err := h.JudgeAccess(act, access)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return h.ask(ActAccess, caller, why, auto, j)
|
||||
}
|
||||
|
||||
// AskedFor says whether an act on subject was ever asked by the engine itself: a library's take is asked by itself once
|
||||
// (ADR 0563 §4), and again only by hand.
|
||||
func (h *HandActs) AskedFor(act, subject string) bool {
|
||||
if act == ActAccess {
|
||||
e, err := h.State.Get(libKey(subject))
|
||||
return err != nil || e != nil // unread is never a reason to ask again
|
||||
}
|
||||
acts, err := h.read()
|
||||
if err != nil {
|
||||
return true
|
||||
}
|
||||
for _, a := range acts {
|
||||
if a.Act == act && a.Subject == subject && (a.Auto || a.State == HandAsked) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (h *HandActs) ask(act, caller, why string, auto bool, j Judged) (any, error) {
|
||||
if h.Asker == nil {
|
||||
return nil, errors.New("this engine reaches no operator to ask: nothing was asked and nothing changed")
|
||||
}
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
now := h.now().UTC()
|
||||
a := HandAct{ID: h.id(act), Act: act, Subject: j.Subject, Path: j.Path, Caller: caller, Why: why, At: now,
|
||||
State: HandAsked, Sets: j.Sets, Disk: j.Disk, Expires: now.Add(asks.ApproveLasts - time.Minute), Auto: auto}
|
||||
a.Ask = a.ID
|
||||
// One ask open per path or access: its open key made by a create, which a second ask cannot make.
|
||||
if _, err := h.State.Create(openKey(act, j.Subject), map[string]string{"ask": a.Ask}); errors.Is(err, stdio.ErrChanged) {
|
||||
held := "another"
|
||||
if e, _ := h.State.Get(openKey(act, j.Subject)); e != nil {
|
||||
var o struct {
|
||||
Ask string `json:"ask"`
|
||||
}
|
||||
if json.Unmarshal(e.Value, &o) == nil && o.Ask != "" {
|
||||
held = o.Ask
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("%s of %s is asked already: the ask %s waits for the operator's answer", act, j.Subject, held)
|
||||
} else if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := h.record(a); err != nil {
|
||||
_ = h.State.Delete(openKey(act, j.Subject))
|
||||
return nil, err
|
||||
}
|
||||
bind, err := asks.ActDigest(h.bound(a))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
headline, does, stays := "Hand a directory on "+h.Node+" to the mesh?", "The mesh gives it "+j.Sets+".",
|
||||
"The directory stays as it was found."
|
||||
if act == ActAccess {
|
||||
headline, does, stays = "Keep the library "+strings.TrimPrefix(j.Subject, "library:")+" on "+h.Node+"?",
|
||||
"The library's group, group write and setgid are set on everything under its root.",
|
||||
"The library waits and nothing under its root changes."
|
||||
}
|
||||
if len([]rune(headline)) > asks.HeadlineLength {
|
||||
headline = string([]rune(headline)[:asks.HeadlineLength])
|
||||
}
|
||||
asked := asks.Ask{ID: a.Ask, Headline: headline, Who: asks.Operator,
|
||||
Explanation: j.Words + " Your yes sets " + j.Sets + "; nothing changes until you answer.",
|
||||
Whole: j.Words + " (" + j.Path + ") Your yes sets " + j.Sets + "; nothing changes until you answer.",
|
||||
Details: "Asked by " + caller + " on " + h.Node + orWhy(why),
|
||||
Options: []asks.Option{{ID: "yes", Label: "Yes", Does: does, Level: asks.Approve, Binds: bind},
|
||||
{ID: "no", Label: "No, leave it", Does: stays, Level: asks.Acknowledge}},
|
||||
Expires: a.Expires, OnExpiry: stays, About: EngineSeat + "-" + h.Node + "-" + act}
|
||||
if err := h.Asker.Ask(asked); err != nil {
|
||||
_, _, _ = h.end(a.Ask, func(x *HandAct) { x.State, x.Said = HandRefused, "the operator could not be asked: "+err.Error() })
|
||||
return nil, fmt.Errorf("the operator could not be asked, and nothing changed: %w", err)
|
||||
}
|
||||
// Once asked, a library's own take ask is kept's own take ask: once, in a key of its own (ADR 0563 §4).
|
||||
if auto && act == ActAccess {
|
||||
_, _ = h.State.Create(libKey(j.Subject), map[string]any{"ask": a.Ask, "at": now})
|
||||
}
|
||||
h.say(HandAsked, a)
|
||||
return map[string]any{"asked": a.Ask, "node": h.Node, "says": "the operator is asked on their channel; only " +
|
||||
"their yes changes anything. acts shows where it stands."}, nil
|
||||
}
|
||||
|
||||
func orWhy(why string) string {
|
||||
if why == "" {
|
||||
return ""
|
||||
}
|
||||
return ": " + why
|
||||
}
|
||||
|
||||
// Warranted is the act on the operator's answer, handed once by asker.Client.Take after it held the warrant to the
|
||||
// ask: a no is recorded refused; a yes is held to the act the option bound and the disk as judged, judged again now,
|
||||
// and only then performed and recorded done with the warrant.
|
||||
func (h *HandActs) Warranted(o asks.Option, w asks.Warrant) error {
|
||||
a, _, err := h.readOne(actKey(w.Ask))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if a == nil || a.State != HandAsked {
|
||||
return fmt.Errorf("the warrant answers the ask %s, which is not open here: nothing is done", w.Ask)
|
||||
}
|
||||
refuse := func(why string) error {
|
||||
ended, ok, err := h.end(w.Ask, func(x *HandAct) { x.State, x.Said = HandRefused, why; warrantOn(x, w) })
|
||||
if err == nil && ok {
|
||||
h.say(HandRefused, ended)
|
||||
}
|
||||
return err
|
||||
}
|
||||
if o.ID != "yes" {
|
||||
return refuse("the operator said no")
|
||||
}
|
||||
var j Judged
|
||||
switch a.Act {
|
||||
case ActHandOver:
|
||||
j, err = h.JudgeHandOver(a.Subject)
|
||||
case ActAccess:
|
||||
j, err = h.JudgeAccess(ActTake, a.Subject)
|
||||
default:
|
||||
err = fmt.Errorf("the act %s is not performed on a warrant", a.Act)
|
||||
}
|
||||
if err != nil {
|
||||
return refuse("judged again on the operator's yes, it is refused: " + err.Error())
|
||||
}
|
||||
now := *a
|
||||
now.Sets, now.Path = j.Sets, j.Path
|
||||
if err := o.Performs(h.bound(now)); err != nil {
|
||||
return refuse("what the yes bound is not what would be done now: " + err.Error())
|
||||
}
|
||||
if j.Disk != a.Disk {
|
||||
return refuse("the disk changed since the operator was asked (" + a.Disk + ", now " + j.Disk + "): ask again")
|
||||
}
|
||||
// Recorded with the way it came (ADR 0563 §5): the operator's yes to this engine's ask, not a terminal's line.
|
||||
by := operatorOf(w) + ", to " + EngineSeat + "'s ask " + w.Ask
|
||||
var said string
|
||||
switch a.Act {
|
||||
case ActHandOver:
|
||||
said, err = h.HandOver(a.Subject, by)
|
||||
case ActAccess:
|
||||
said, err = h.AccessAct(AccessAct{Act: ActTake, Access: a.Subject, Path: j.Path, Warrant: w.Ask, By: by})
|
||||
}
|
||||
if err != nil {
|
||||
return refuse(err.Error())
|
||||
}
|
||||
ended, ok, err := h.end(w.Ask, func(x *HandAct) { x.State, x.Said = HandDone, said; warrantOn(x, w) })
|
||||
if err == nil && ok {
|
||||
h.say(HandDone, ended)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func warrantOn(a *HandAct, w asks.Warrant) {
|
||||
a.Proofs = w.Proofs
|
||||
if w.By != nil {
|
||||
a.By, a.Via = w.By.Who, w.By.Kind
|
||||
if w.By.Verified != "" {
|
||||
a.Via += " (" + w.By.Verified + ")"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func operatorOf(w asks.Warrant) string {
|
||||
if w.By == nil {
|
||||
return "the operator"
|
||||
}
|
||||
return "the " + w.By.Who + " via " + w.By.Kind
|
||||
}
|
||||
|
||||
// Heard is a warrant heard on the engine's decided subject: taken once (asker.Client.Take), and an ask that ended with
|
||||
// no choice recorded refused.
|
||||
func (h *HandActs) Heard(w asks.Warrant) {
|
||||
if h.Asker == nil {
|
||||
return
|
||||
}
|
||||
// A warrant the ask kept here does not hold to (another digest, another option, too late) ends nothing: it is no
|
||||
// answer to the ask, which stays open for the operator's.
|
||||
if handled, _ := h.Asker.Take(w, h.Warranted); handled == asker.Recorded {
|
||||
why := "the ask ended " + string(w.Outcome)
|
||||
if w.Words != "" {
|
||||
why += ": " + w.Words
|
||||
}
|
||||
if ended, ok, _ := h.end(w.Ask, func(x *HandAct) { x.State, x.Said = HandRefused, why; warrantOn(x, w) }); ok {
|
||||
h.say(HandRefused, ended)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Reconcile catches up what the engine missed while it was away: a warrant the router holds for an open ask is taken,
|
||||
// and an ask past its expiry and a margin is closed refused, expired.
|
||||
func (h *HandActs) Reconcile() {
|
||||
if h.Asker == nil {
|
||||
return
|
||||
}
|
||||
acts, err := h.read()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, a := range acts {
|
||||
if a.State != HandAsked || a.Ask == "" {
|
||||
continue
|
||||
}
|
||||
if _, w, err := h.Asker.Where(a.Ask); err == nil && w != nil {
|
||||
h.Heard(*w)
|
||||
continue
|
||||
}
|
||||
if !a.Expires.IsZero() && h.now().After(a.Expires.Add(askMargin)) {
|
||||
if ended, ok, _ := h.end(a.Ask, func(x *HandAct) { x.State, x.Said = HandRefused, "the ask expired unanswered" }); ok {
|
||||
h.say(HandRefused, ended)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MemState is an ActStore in memory: an engine's acts while no bus is reached in a test.
|
||||
type MemState struct {
|
||||
mu sync.Mutex
|
||||
keys map[string]stdio.StateEntry
|
||||
top uint64
|
||||
}
|
||||
|
||||
func (m *MemState) Get(key string) (*stdio.StateEntry, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if e, ok := m.keys[key]; ok {
|
||||
return &e, nil
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *MemState) Create(key string, value any) (uint64, error) { return m.put(key, value, 0, true) }
|
||||
|
||||
func (m *MemState) Update(key string, value any, revision uint64) (uint64, error) {
|
||||
return m.put(key, value, revision, false)
|
||||
}
|
||||
|
||||
func (m *MemState) put(key string, value any, revision uint64, create bool) (uint64, error) {
|
||||
raw, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
was, kept := m.keys[key]
|
||||
if (create && kept) || (!create && (!kept || was.Revision != revision)) {
|
||||
return 0, stdio.ErrChanged
|
||||
}
|
||||
if m.keys == nil {
|
||||
m.keys = map[string]stdio.StateEntry{}
|
||||
}
|
||||
m.top++
|
||||
m.keys[key] = stdio.StateEntry{Key: key, Value: raw, Revision: m.top}
|
||||
return m.top, nil
|
||||
}
|
||||
|
||||
func (m *MemState) Delete(key string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
delete(m.keys, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MemState) Keys(prefix string) ([]string, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
var out []string
|
||||
for k := range m.keys {
|
||||
if strings.HasPrefix(k, prefix) {
|
||||
out = append(out, k)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
}
|
||||
@@ -76,13 +76,9 @@ const MaxHandOverBy = 200
|
||||
// HandOverCheck is what an ask is held to: the mesh's key, this node's name, the clock, and the nonces taken.
|
||||
type HandOverCheck struct {
|
||||
Signer ed25519.PublicKey
|
||||
// SignerNow, when set, is the controller's key as the engine trusts it at this ask, read per ask in place of Signer:
|
||||
// nil once the controller's lease is handed away, which refuses every ask signed by its key (novox/hq ADR 0563 §7,
|
||||
// ADR 0550 §2).
|
||||
SignerNow func() ed25519.PublicKey
|
||||
Node string
|
||||
Now func() time.Time
|
||||
Nonces *Nonces
|
||||
Node string
|
||||
Now func() time.Time
|
||||
Nonces *Nonces
|
||||
}
|
||||
|
||||
// Nonces are the nonces of asks taken, each kept until its ask expires: an ask is taken once. Kept in memory —
|
||||
@@ -161,14 +157,8 @@ func believedBytesWithin(check HandOverCheck, context string, withPath bool, bod
|
||||
if err := json.Unmarshal(body, &signed); err != nil {
|
||||
return HandOverAsk{}, nil, "the ask is not a signed ask: " + err.Error()
|
||||
}
|
||||
signer := check.Signer
|
||||
if check.SignerNow != nil {
|
||||
if signer = check.SignerNow(); signer == nil {
|
||||
return HandOverAsk{}, nil, "the controller's lease is handed away, and no ask signed by its key is taken"
|
||||
}
|
||||
}
|
||||
if len(signer) != ed25519.PublicKeySize ||
|
||||
!ed25519.Verify(signer, append([]byte(context), signed.Ask...), signed.Signature) {
|
||||
if len(check.Signer) != ed25519.PublicKeySize ||
|
||||
!ed25519.Verify(check.Signer, append([]byte(context), signed.Ask...), signed.Signature) {
|
||||
return HandOverAsk{}, nil, "the ask is not signed by the mesh"
|
||||
}
|
||||
var ask HandOverAsk
|
||||
|
||||
@@ -9,9 +9,6 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// The hand-over's ask and answer hold these field names; the controller's side holds the same list
|
||||
@@ -148,21 +145,6 @@ func TestAHandOverAskIsBelievedOnlySignedForThisNodeUnexpiredAndOnce(t *testing.
|
||||
if strings.Join(given, "; ") != "/srv/notes by jo through mesh-cli on anchor; /refused by jo through mesh-cli on anchor" {
|
||||
t.Fatalf("the handler was given %v", given)
|
||||
}
|
||||
// ADR 0563 §7, ADR 0550 §2: after the lease write the engine answers no ask signed by the controller's key. The key
|
||||
// is read per ask from what the engine trusts now; nil (the lease handed away) refuses, and nothing is handed over.
|
||||
trusted := m.Signer
|
||||
following := HandOverCheck{Signer: m.Signer, SignerNow: func() ed25519.PublicKey { return trusted }, Node: m.Node,
|
||||
Now: func() time.Time { return now }, Nonces: &Nonces{}}
|
||||
before := len(given)
|
||||
trusted = nil
|
||||
late := signedAsk(t, key, with(func(a *HandOverAsk) { a.Expires = now.Add(time.Minute); a.Nonce = strings.Repeat("cd", 16) }))
|
||||
if a := AnswerHandOver(following, handle, late); !strings.Contains(a.Refused, "lease is handed away") || len(given) != before {
|
||||
t.Fatalf("after the lease write an ask signed by the controller was answered %+v", a)
|
||||
}
|
||||
trusted = m.Signer
|
||||
if a := AnswerHandOver(following, handle, late); a.Refused != "" || len(given) != before+1 {
|
||||
t.Fatalf("while the controller holds the lease its ask was answered %+v", a)
|
||||
}
|
||||
// A nonce is forgotten once its ask has expired, so the set does not grow for ever.
|
||||
now = now.Add(HandOverGood + time.Second)
|
||||
check.Nonces.take("ff", now.Add(time.Second), now)
|
||||
@@ -192,23 +174,17 @@ func TestAHandOverIsAnsweredOnlyWhenItCarriesAReply(t *testing.T) {
|
||||
// handingLink is a quiet link that hears hand-overs, keeping the handler it was given.
|
||||
type handingLink struct {
|
||||
*quietLink
|
||||
given chan HandOverFunc
|
||||
signer chan func() ed25519.PublicKey
|
||||
given chan HandOverFunc
|
||||
}
|
||||
|
||||
func (l handingLink) HearHandOvers(h HandOverFunc) { l.given <- h }
|
||||
|
||||
// FollowSigner keeps what the link is told to verify the controller's asks with.
|
||||
func (l handingLink) FollowSigner(f func() ed25519.PublicKey) { l.signer <- f }
|
||||
|
||||
// **The link is given the queue's hand-over** when it is served: what the controller asks reaches the engine's
|
||||
// judgement, and nothing else.
|
||||
func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
|
||||
m, _ := aMember(t)
|
||||
l := handingLink{quietLink: newQuietLink(), given: make(chan HandOverFunc, 1), signer: make(chan func() ed25519.PublicKey, 1)}
|
||||
l := handingLink{quietLink: newQuietLink(), given: make(chan HandOverFunc, 1)}
|
||||
q := &Session{Membership: m, Timeout: time.Second}
|
||||
// The link verifies the controller's asks with the key the engine trusts now (ADR 0563 §7).
|
||||
q.ControllerKey = func() ed25519.PublicKey { return nil }
|
||||
q.HandOver = func(path, by string) (string, error) { return "the queue's: " + path + " by " + by, nil }
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
served := make(chan error, 1)
|
||||
@@ -221,40 +197,8 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the link was never given the queue's hand-over")
|
||||
}
|
||||
select {
|
||||
case f := <-l.signer:
|
||||
if f() != nil {
|
||||
t.Fatal("the link was given another key than the queue's")
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the link was never given the key the engine trusts now")
|
||||
}
|
||||
stop()
|
||||
<-served
|
||||
// 774 (ADR 0563 §1): a call of the node seat reaches its verb with its caller as the subject proves it, unproven on
|
||||
// the subject that names none; a verb the seat does not serve is refused, and the answer is a tool's envelope.
|
||||
t.Run("a call of the node seat reaches its verb with its proven caller", func(t *testing.T) {
|
||||
h := &HandActs{Node: "n1", State: &MemState{},
|
||||
RootFree: func() RootVerdict { return RootVerdict{Machine: "n1", Free: true, Why: "judged"} },
|
||||
JudgeDirectories: func(by string) (string, error) { return "judged for " + by, nil }}
|
||||
for _, row := range []struct{ verb, subject, want string }{
|
||||
{"judge-directories", "mesh.seat.node-engine.call.judge-directories.n1.shanks~node-tools", `"said":"judged for shanks.node-tools"`},
|
||||
{"judge-directories", EngineToolSubject("judge-directories", "n1"), `"said":"judged for unproven"`},
|
||||
{"judge-directories", "mesh.seat.node-engine.call.judge-directories.n2.shanks~node-tools", `"said":"judged for unproven"`},
|
||||
{"root-free", "mesh.seat.node-engine.call.root-free.n1.ace~messenger", `"free":true`},
|
||||
{"unserved", "mesh.seat.node-engine.call.unserved.n1.x", `"error":"the node seat node-engine serves no verb \"unserved\""`},
|
||||
} {
|
||||
if got := string(AnswerEngineCall(h.Call, row.verb, "n1", row.subject, []byte(`{}`))); !strings.Contains(got, row.want) ||
|
||||
!strings.Contains(got, `"node":"n1"`) {
|
||||
t.Errorf("%s on %s: %s", row.verb, row.subject, got)
|
||||
}
|
||||
}
|
||||
acts, _ := h.Acts(0)
|
||||
if got := acts.(map[string]any)["acts"].([]HandAct); len(got) != 3 || got[0].Caller != "unproven" ||
|
||||
got[2].Caller != "shanks.node-tools" || got[2].Act != ActJudgeDirectories {
|
||||
t.Errorf("the judgements are recorded as %+v", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// **Against a real server**: a signed request on the node's own subject is answered; one on another node's is
|
||||
@@ -293,63 +237,4 @@ func TestNatsAHandOverIsAskedOfTheNode(t *testing.T) {
|
||||
if len(handled) != 1 {
|
||||
t.Fatalf("handled %d asks; want only the one with a reply", len(handled))
|
||||
}
|
||||
// 774 (ADR 0563 §2, §3): the engine's acts are its node's keys of its declared state on the bus, each written by
|
||||
// compare-and-set: a second create and a stale revision lose, and another node's keys are not this node's.
|
||||
t.Run("the acts are kept in the bus's own storage under this node's keys", func(t *testing.T) {
|
||||
_ = js.DeleteKeyValue(ActsBucket)
|
||||
if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: ActsBucket}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
q := &Session{Membership: m, Timeout: time.Second}
|
||||
q.bus = l
|
||||
mine, theirs := BusState{Session: q, Node: m.Node}, BusState{Session: q, Node: "another"}
|
||||
rev, err := mine.Create("act.a-1", map[string]string{"state": "asked"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := mine.Create("act.a-1", map[string]string{}); !errors.Is(err, stdio.ErrChanged) {
|
||||
t.Errorf("a second create: %v", err)
|
||||
}
|
||||
if _, err := mine.Update("act.a-1", map[string]string{"state": "done"}, rev+7); !errors.Is(err, stdio.ErrChanged) {
|
||||
t.Errorf("an update at a stale revision: %v", err)
|
||||
}
|
||||
if _, err := mine.Update("act.a-1", map[string]string{"state": "done"}, rev); err != nil {
|
||||
t.Errorf("an update at its revision: %v", err)
|
||||
}
|
||||
_, _ = theirs.Create("act.b-1", map[string]string{})
|
||||
if keys, err := mine.Keys("act."); err != nil || len(keys) != 1 || keys[0] != "act.a-1" {
|
||||
t.Errorf("this node's acts are %v, %v", keys, err)
|
||||
}
|
||||
if e, _ := js.KeyValue(ActsBucket); e != nil {
|
||||
if got, err := e.Get(m.Node + ".act.a-1"); err != nil || !strings.Contains(string(got.Value()), "done") {
|
||||
t.Errorf("the bucket holds %v, %v", got, err)
|
||||
}
|
||||
}
|
||||
if err := mine.Delete("act.a-1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if e, err := mine.Get("act.a-1"); err != nil || e != nil {
|
||||
t.Errorf("a deleted act reads %v, %v", e, err)
|
||||
}
|
||||
})
|
||||
// 774 (ADR 0563 §1): the node seat is served on this node's subjects alone, naming its caller, and said to discovery.
|
||||
t.Run("the node seat is served on this node and discovered", func(t *testing.T) {
|
||||
l.ServeEngineSeat(func(c EngineCall) (any, error) { return c.Verb + " for " + c.Caller, nil })
|
||||
reply, err := conn.Request("mesh.seat.node-engine.call.acts."+m.Node+".shanks~node-tools", []byte(`{}`), 5*time.Second)
|
||||
if err != nil || !strings.Contains(string(reply.Data), `"result":"acts for shanks.node-tools"`) {
|
||||
t.Fatalf("the call answered %v, %v", reply, err)
|
||||
}
|
||||
if _, err := conn.Request(EngineToolSubject("acts", "another"), []byte(`{}`), 300*time.Millisecond); err == nil {
|
||||
t.Fatal("another node's verb was answered here")
|
||||
}
|
||||
for seat, want := range map[string]bool{EngineSeat: true, "node-login-shell": false} {
|
||||
verb := "acts"
|
||||
if !want {
|
||||
verb = "execute"
|
||||
}
|
||||
if heard, err := l.HeardServed(seat, verb); err != nil || heard != want {
|
||||
t.Errorf("discovery hears %s.%s here: %v, %v", seat, verb, heard, err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -274,8 +274,7 @@ func TestTheReportIsSentFirstAndTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
|
||||
// 1126 (ADR 0555 §3): with its keys, the node key carries the node's record: the profile's present capabilities and
|
||||
// networks, the outward links, the health's contract and the public keys; a report that changes none of it writes
|
||||
// nothing, a changed fact writes the key once, and a report without a profile or health leaves them standing.
|
||||
keys := RecordKeys{TunnelKey: "tk", SealingKey: "sk", ServingKey: "pk", Certificate: "cert",
|
||||
Trusts: []string{"controller key sha256:aa", "vault key sha256:bb"}} // and what it trusts (L3 s6, ADR 0550 §2)
|
||||
keys := RecordKeys{TunnelKey: "tk", SealingKey: "sk", ServingKey: "pk", Certificate: "cert"}
|
||||
q.Keys = func() RecordKeys { return keys }
|
||||
rep.Profile = map[string]any{"capabilities": []any{map[string]any{"name": "systemd", "present": true},
|
||||
map[string]any{"name": "docker", "present": false}, map[string]any{"name": "apt", "present": true}},
|
||||
@@ -287,7 +286,7 @@ func TestTheReportIsSentFirstAndTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
|
||||
t.Fatalf("the node key holds no record: %+v, %v", got, err)
|
||||
}
|
||||
wantRecord := delivery.NodeRecord{V: delivery.Version, Node: "n1", TunnelKey: "tk", SealingKey: "sk", ServingKey: "pk",
|
||||
Certificate: "cert", Trusts: keys.Trusts, Capabilities: []string{"apt", "systemd"}, OutwardLinks: []string{"eth0"}, NetworksSaid: true,
|
||||
Certificate: "cert", Capabilities: []string{"apt", "systemd"}, OutwardLinks: []string{"eth0"}, NetworksSaid: true,
|
||||
Networks: []catalogue.AttachedNetwork{{Link: "eth0", Network: "192.0.2.0/24"}}, Contract: NodeKeyContract}
|
||||
if !delivery.SameNodeRecord(*got.Record, wantRecord) {
|
||||
t.Errorf("record = %+v, want %+v", *got.Record, wantRecord)
|
||||
|
||||
@@ -6,11 +6,9 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/micro"
|
||||
)
|
||||
|
||||
// The host's link on the bus: one connection, and what is asked of this node over it. The node hears no declaration
|
||||
@@ -37,14 +35,11 @@ func natsURL(address string) string {
|
||||
|
||||
// natsLink is this node's connection as a JetStream subscription.
|
||||
type natsLink struct {
|
||||
// mu guards signerNow (FollowSigner).
|
||||
mu sync.Mutex
|
||||
signerNow func() ed25519.PublicKey
|
||||
conn *nats.Conn
|
||||
js nats.JetStreamContext
|
||||
asking *nats.Subscription
|
||||
asked chan struct{}
|
||||
handing *nats.Subscription
|
||||
conn *nats.Conn
|
||||
js nats.JetStreamContext
|
||||
asking *nats.Subscription
|
||||
asked chan struct{}
|
||||
handing *nats.Subscription
|
||||
// searching hears the controller's terminal asking for a fresh setuid search (novox/hq issue 361).
|
||||
searching *nats.Subscription
|
||||
// judging hears the controller asking for a read-only judgement of this node's directories (issue 358).
|
||||
@@ -53,10 +48,7 @@ type natsLink struct {
|
||||
acting *nats.Subscription
|
||||
// moving hears the controller asking to move a value sealed to this machine into the vault (issue 676).
|
||||
moving *nats.Subscription
|
||||
// engine serves the node seat node-engine, and warrants hears the operator's answers to its asks (ADR 0563).
|
||||
engine micro.Service
|
||||
warrants *nats.Subscription
|
||||
node string
|
||||
node string
|
||||
// signer is the mesh's key, which a hand-over ask is verified with as a declaration is (issue 356).
|
||||
signer ed25519.PublicKey
|
||||
lost chan error
|
||||
@@ -149,21 +141,6 @@ func (l *natsLink) hearAsks() {
|
||||
}
|
||||
}
|
||||
|
||||
// FollowSigner has every signed ask heard on this link verified with the key the engine trusts at that ask, the
|
||||
// controller's while it holds its lease and none from its hand-over (novox/hq ADR 0563 §7, ADR 0550 §2).
|
||||
func (l *natsLink) FollowSigner(f func() ed25519.PublicKey) {
|
||||
l.mu.Lock()
|
||||
l.signerNow = f
|
||||
l.mu.Unlock()
|
||||
}
|
||||
|
||||
// check is how a signed ask is believed on this link: the static membership signer until FollowSigner names the trust.
|
||||
func (l *natsLink) check() HandOverCheck {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
return HandOverCheck{Signer: l.signer, SignerNow: l.signerNow, Node: l.node}
|
||||
}
|
||||
|
||||
// HearHandOvers listens for the controller asking this node to hand a directory used as found to the mesh
|
||||
// (novox/hq issue 356): a request on this machine's own subject, answered once on the reply it carried. The
|
||||
// ask is verified with the mesh's key before anything is read out of it (AnswerHandOver), because a responder
|
||||
@@ -174,7 +151,7 @@ func (l *natsLink) HearHandOvers(handle HandOverFunc) {
|
||||
if handle == nil {
|
||||
return
|
||||
}
|
||||
check := l.check()
|
||||
check := HandOverCheck{Signer: l.signer, Node: l.node}
|
||||
handing, err := l.conn.Subscribe(AskHandOverSubject(l.node), func(msg *nats.Msg) {
|
||||
if body, answer := HandOverReply(check, handle, msg.Reply, msg.Data); answer {
|
||||
_ = msg.Respond(body)
|
||||
@@ -193,7 +170,7 @@ func (l *natsLink) HearSetuidSearches(handle SetuidSearchFunc) {
|
||||
if handle == nil {
|
||||
return
|
||||
}
|
||||
check := l.check()
|
||||
check := HandOverCheck{Signer: l.signer, Node: l.node}
|
||||
searching, err := l.conn.Subscribe(AskSetuidSearchSubject(l.node), func(msg *nats.Msg) {
|
||||
if body, answer := SetuidSearchReply(check, handle, msg.Reply, msg.Data); answer {
|
||||
_ = msg.Respond(body)
|
||||
@@ -212,7 +189,7 @@ func (l *natsLink) HearDirectoryJudgements(handle JudgeDirectoriesFunc) {
|
||||
if handle == nil {
|
||||
return
|
||||
}
|
||||
check := l.check()
|
||||
check := HandOverCheck{Signer: l.signer, Node: l.node}
|
||||
judging, err := l.conn.Subscribe(AskJudgeDirectoriesSubject(l.node), func(msg *nats.Msg) {
|
||||
if body, answer := JudgeDirectoriesReply(check, handle, msg.Reply, msg.Data); answer {
|
||||
_ = msg.Respond(body)
|
||||
@@ -230,7 +207,7 @@ func (l *natsLink) HearAccessActs(handle AccessActFunc) {
|
||||
if handle == nil {
|
||||
return
|
||||
}
|
||||
check := l.check()
|
||||
check := HandOverCheck{Signer: l.signer, Node: l.node}
|
||||
acting, err := l.conn.Subscribe(AskAccessActSubject(l.node), func(msg *nats.Msg) {
|
||||
if body, answer := AccessActReply(check, handle, msg.Reply, msg.Data); answer {
|
||||
_ = msg.Respond(body)
|
||||
@@ -248,7 +225,7 @@ func (l *natsLink) HearMoveSecrets(handle MoveSecretFunc) {
|
||||
if handle == nil {
|
||||
return
|
||||
}
|
||||
check := l.check()
|
||||
check := HandOverCheck{Signer: l.signer, Node: l.node}
|
||||
moving, err := l.conn.Subscribe(AskMoveSecretSubject(l.node), func(msg *nats.Msg) {
|
||||
if body, answer := MoveSecretReply(check, handle, msg.Reply, msg.Data); answer {
|
||||
_ = msg.Respond(body)
|
||||
@@ -281,12 +258,6 @@ func (l *natsLink) Close() {
|
||||
if l.judging != nil {
|
||||
_ = l.judging.Unsubscribe()
|
||||
}
|
||||
if l.engine != nil {
|
||||
_ = l.engine.Stop()
|
||||
}
|
||||
if l.warrants != nil {
|
||||
_ = l.warrants.Unsubscribe()
|
||||
}
|
||||
if l.conn != nil {
|
||||
l.conn.Close()
|
||||
}
|
||||
@@ -330,12 +301,10 @@ func (l *natsLink) Ask(ctx context.Context, subject string, body []byte) ([]byte
|
||||
// What the queue asks of a live link beside Bus, each by a type assertion that fails quietly: said here
|
||||
// so a link that drops one does not build.
|
||||
var (
|
||||
_ Bus = (*natsLink)(nil)
|
||||
_ HealthBus = (*natsLink)(nil)
|
||||
_ HaveBus = (*natsLink)(nil)
|
||||
_ ToolBus = (*natsLink)(nil)
|
||||
_ Asker = (*natsLink)(nil)
|
||||
_ Asked = (*natsLink)(nil)
|
||||
_ EngineSeats = (*natsLink)(nil)
|
||||
_ OperatorAsks = (*natsLink)(nil)
|
||||
_ Bus = (*natsLink)(nil)
|
||||
_ HealthBus = (*natsLink)(nil)
|
||||
_ HaveBus = (*natsLink)(nil)
|
||||
_ ToolBus = (*natsLink)(nil)
|
||||
_ Asker = (*natsLink)(nil)
|
||||
_ Asked = (*natsLink)(nil)
|
||||
)
|
||||
|
||||
@@ -23,8 +23,6 @@ type RecordKeys struct {
|
||||
Certificate string
|
||||
// Authority is the mesh's own certificate as this machine last applied it (public).
|
||||
Authority string
|
||||
// Trusts is what this machine trusts a mesh signature from now (delivery.Trusts, ADR 0550 §2).
|
||||
Trusts []string
|
||||
}
|
||||
|
||||
// recordOf is the node's record from one facts report and its keys; nil for a report that carries no profile (one that
|
||||
@@ -47,7 +45,7 @@ func recordOf(node string, r Report, keys RecordKeys) *delivery.NodeRecord {
|
||||
return nil
|
||||
}
|
||||
rec := &delivery.NodeRecord{V: delivery.Version, Node: node, TunnelKey: keys.TunnelKey, SealingKey: keys.SealingKey,
|
||||
ServingKey: keys.ServingKey, Certificate: keys.Certificate, Authority: keys.Authority, Trusts: keys.Trusts}
|
||||
ServingKey: keys.ServingKey, Certificate: keys.Certificate, Authority: keys.Authority}
|
||||
for _, c := range p.Capabilities {
|
||||
if c.Present && c.Name != "" {
|
||||
rec.Capabilities = append(rec.Capabilities, c.Name)
|
||||
|
||||
@@ -7,8 +7,6 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
meshasks "git.novox.be/novox/mesh-sdk/go/asks"
|
||||
)
|
||||
|
||||
// Membership is what a node needs to reach its mesh again, held by the caller.
|
||||
@@ -187,12 +185,6 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann
|
||||
}
|
||||
// And the controller asking this node to hand a directory used as found to the mesh (novox/hq issue
|
||||
// 356), answered on the link the question came over.
|
||||
// Every signed ask is verified with the key trusted at that ask, none after the lease write (ADR 0563 §7).
|
||||
if f, ok := link.(interface {
|
||||
FollowSigner(func() ed25519.PublicKey)
|
||||
}); ok && queue.ControllerKey != nil {
|
||||
f.FollowSigner(queue.ControllerKey)
|
||||
}
|
||||
if h, ok := link.(HandOvers); ok {
|
||||
h.HearHandOvers(queue.HandOver)
|
||||
}
|
||||
@@ -208,19 +200,6 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann
|
||||
if s, ok := link.(AccessActs); ok {
|
||||
s.HearAccessActs(queue.AccessAct)
|
||||
}
|
||||
// And its node seat, node-engine, with the operator's answers to its own asks (novox/hq ADR 0563): served on this
|
||||
// link, and what was answered while no link was open caught up from the router's records.
|
||||
if queue.Acts != nil {
|
||||
if s, ok := link.(EngineSeats); ok {
|
||||
s.ServeEngineSeat(queue.Acts.Call)
|
||||
}
|
||||
if a, ok := link.(OperatorAsks); ok {
|
||||
if err := a.HearWarrants(meshasks.EngineAsker(m.Node), queue.Acts.Heard); err != nil {
|
||||
say("cannot hear the operator's answers to this engine's asks: " + err.Error())
|
||||
}
|
||||
go queue.Acts.Reconcile()
|
||||
}
|
||||
}
|
||||
// And asking to move a value sealed to this machine into the vault (design 57's V3, issue 676).
|
||||
if s, ok := link.(MoveSecrets); ok {
|
||||
s.HearMoveSecrets(queue.MoveSecret)
|
||||
|
||||
@@ -2,14 +2,12 @@ package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/ed25519"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
@@ -49,12 +47,6 @@ type Session struct {
|
||||
// MoveSecret moves one value sealed to this machine into the vault, asked by the controller during the move of
|
||||
// the secrets (design 57's V3, issue 676). Nil answers every such ask with a refusal.
|
||||
MoveSecret MoveSecretFunc
|
||||
// ControllerKey is the controller's key as the engine trusts it now: nil once its lease is handed away (novox/hq
|
||||
// ADR 0563 §7). Each link verifies every signed ask with it, read per ask; nil ControllerKey keeps the membership's.
|
||||
ControllerKey func() ed25519.PublicKey
|
||||
// Acts is the node seat node-engine this engine holds (novox/hq ADR 0563): served on every link, and its asks to the
|
||||
// operator asked and heard over it. Nil serves no seat.
|
||||
Acts *HandActs
|
||||
// Keys reads the node's public keys and its certificate for its record (ADR 0555 §3), at each facts report; nil writes
|
||||
// no record.
|
||||
Keys func() RecordKeys
|
||||
@@ -345,49 +337,3 @@ func (q *Session) AskSeat(ctx context.Context, seat, verb string, args any, with
|
||||
}
|
||||
return env.Result, nil
|
||||
}
|
||||
|
||||
// PublishAsk publishes an ask or a cancel to the operator's router over the link open now (novox/hq ADR 0563 §3).
|
||||
func (q *Session) PublishAsk(subject string, body any, id string) error {
|
||||
asking, err := q.operatorAsks()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return asking.PublishAsk(subject, body, id)
|
||||
}
|
||||
|
||||
// RouterRecord reads one key of the router's records over the link open now.
|
||||
func (q *Session) RouterRecord(key string) (*stdio.StateEntry, error) {
|
||||
asking, err := q.operatorAsks()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return asking.RouterRecord(key)
|
||||
}
|
||||
|
||||
// SayEngine says one of the node seat's events over the link open now; with none open it is not said.
|
||||
func (q *Session) SayEngine(event string, body any) {
|
||||
q.init()
|
||||
q.mu.Lock()
|
||||
bus := q.bus
|
||||
q.mu.Unlock()
|
||||
if sayer, ok := bus.(interface{ SayEngine(string, any) error }); ok {
|
||||
if err := sayer.SayEngine(event, body); err != nil {
|
||||
q.Say("could not say the node seat's " + event + ": " + err.Error())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (q *Session) operatorAsks() (OperatorAsks, 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)
|
||||
}
|
||||
asking, ok := bus.(OperatorAsks)
|
||||
if !ok {
|
||||
return nil, errNoAsking
|
||||
}
|
||||
return asking, nil
|
||||
}
|
||||
|
||||
@@ -57,82 +57,4 @@ func TestASetuidSearchAskIsBelievedOnlySignedForThisNodeAndOnce(t *testing.T) {
|
||||
if a := AnswerSetuidSearch(check, nil, body); a.Refused == "" {
|
||||
t.Fatalf("no handler: %+v", a)
|
||||
}
|
||||
// 774 (ADR 0563 §2): the node seat's setuid-search starts none while one runs, at most one in any hour, naming when the
|
||||
// next may start, and records who asked.
|
||||
t.Run("the node seat's search is bounded", func(t *testing.T) {
|
||||
at, running, started := now, false, 0
|
||||
h := &HandActs{Node: "n1", State: &MemState{}, Now: func() time.Time { return at },
|
||||
Searching: func() bool { return running },
|
||||
SetuidSearch: func(by string) (string, error) { started++; return "a new search starts", nil }}
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
after time.Duration
|
||||
running bool
|
||||
refused string
|
||||
started int
|
||||
}{
|
||||
{name: "one runs", running: true, refused: "runs on this machine now", started: 0},
|
||||
{name: "the first", started: 1},
|
||||
{name: "within the hour", after: 59 * time.Minute, refused: "the next may start at 2026-10-10T03:00:00Z", started: 1},
|
||||
{name: "after the hour", after: time.Hour + time.Second, started: 2},
|
||||
} {
|
||||
at, running = now.Add(row.after), row.running
|
||||
_, err := h.Call(EngineCall{Verb: ActSetuidSearch, Caller: "ace.node-tools"})
|
||||
if (row.refused == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), row.refused)) || started != row.started {
|
||||
t.Errorf("%s: %v, %d started", row.name, err, started)
|
||||
}
|
||||
}
|
||||
acts, _ := h.Acts(1)
|
||||
if got := acts.(map[string]any)["acts"].([]HandAct); len(got) != 1 || got[0].Caller != "ace.node-tools" || got[0].State != HandDone {
|
||||
t.Errorf("the last search is recorded as %+v", got)
|
||||
}
|
||||
// The bound is kept in a key of its own (rev-l3 round 1): two hundred judgements before a search, or the acts
|
||||
// gone from the record, never let a second search start within the hour.
|
||||
h.JudgeDirectories = func(string) (string, error) { return "judged", nil }
|
||||
for range 200 {
|
||||
if _, err := h.Call(EngineCall{Verb: ActJudgeDirectories, Caller: "ace.node-tools"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
at = now.Add(time.Hour + 2*time.Minute)
|
||||
if _, err := h.Call(EngineCall{Verb: ActSetuidSearch, Caller: "ace.node-tools"}); err == nil || !strings.Contains(err.Error(), "the next may start") {
|
||||
t.Errorf("a search after two hundred judgements: %v", err)
|
||||
}
|
||||
keys, _ := h.State.Keys("act.")
|
||||
for _, k := range keys {
|
||||
_ = h.State.Delete(k)
|
||||
}
|
||||
if _, err := h.Call(EngineCall{Verb: ActSetuidSearch, Caller: "ace.node-tools"}); err == nil || !strings.Contains(err.Error(), "the next may start") || started != 2 {
|
||||
t.Errorf("a search with the acts gone: %v, %d started", err, started)
|
||||
}
|
||||
// rev-l3 round 2 (ADR 0563 §2, a wrong caller gains nothing): judgements and refused searches, which any caller
|
||||
// may ask, keep one rolling key of the node's, so ten thousand of them write a bounded number of bytes and add no
|
||||
// key to the bucket every node shares.
|
||||
keysNow := func() (int, int) {
|
||||
all, _ := h.State.Keys("")
|
||||
size := 0
|
||||
for _, k := range all {
|
||||
if e, _ := h.State.Get(k); e != nil {
|
||||
size += len(e.Value)
|
||||
}
|
||||
}
|
||||
return len(all), size
|
||||
}
|
||||
_, _ = h.Call(EngineCall{Verb: ActJudgeDirectories, Caller: "unproven"})
|
||||
keysBefore, _ := keysNow()
|
||||
for i := range 10000 {
|
||||
verb := ActJudgeDirectories
|
||||
if i%10 == 0 {
|
||||
verb = ActSetuidSearch
|
||||
}
|
||||
_, _ = h.Call(EngineCall{Verb: verb, Caller: "unproven"})
|
||||
}
|
||||
if n, size := keysNow(); n != keysBefore || size > 64<<10 {
|
||||
t.Errorf("ten thousand judgements and refused searches left %d keys (was %d) of %d bytes", n, keysBefore, size)
|
||||
}
|
||||
listed, _ := h.Acts(0)
|
||||
if got := listed.(map[string]any)["acts"].([]HandAct); len(got) == 0 || got[0].Caller != "unproven" {
|
||||
t.Errorf("acts does not show the latest judgement first: %d", len(got))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user