From aaffc19232b8f29fee9cf2389b3e4da62d95f4c0 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 09:56:01 +0100 Subject: [PATCH] =?UTF-8?q?The=20node=20seat=20node-engine=20served=20on?= =?UTF-8?q?=20the=20engine's=20link:=20its=20verbs=20with=20their=20proven?= =?UTF-8?q?=20caller,=20its=20acts=20kept=20in=20its=20own=20state,=20a=20?= =?UTF-8?q?hand-over=20or=20access=20act=20asked=20of=20the=20operator=20a?= =?UTF-8?q?s=20node-engine-=20and=20performed=20only=20on=20the=20wa?= =?UTF-8?q?rrant=20(issue=20774,=20ADR=200563=20=C2=A71-=C2=A73)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/accounts/setuid_cache.go | 11 + internal/link/access_act_test.go | 101 +++++ internal/link/engineseat.go | 260 +++++++++++ internal/link/handacts.go | 665 ++++++++++++++++++++++++++++ internal/link/handover_test.go | 44 ++ internal/link/hearing_nats.go | 26 +- internal/link/run.go | 15 + internal/link/session.go | 50 +++ internal/link/setuid_search_test.go | 30 ++ 9 files changed, 1195 insertions(+), 7 deletions(-) create mode 100644 internal/link/engineseat.go create mode 100644 internal/link/handacts.go diff --git a/internal/accounts/setuid_cache.go b/internal/accounts/setuid_cache.go index 777bbf5..eadd8a0 100644 --- a/internal/accounts/setuid_cache.go +++ b/internal/accounts/setuid_cache.go @@ -240,6 +240,17 @@ func (c *SetuidCache) SearchAgain() { c.forgetProgress() } +// Running says whether a search walks now (novox/hq ADR 0563 §2: the node seat starts no second while one does). +// Nil c runs none. +func (c *SetuidCache) Running() bool { + if c == nil { + return false + } + c.mu.Lock() + defer c.mu.Unlock() + return c.running +} + // SearchNow throws away the last complete search, kept or in memory, and any walk under way or kept, and has // the next look start a full search at once, past AgainAfter and any back-off (novox/hq issue 361): the // operator changed by hand what the last one found, which no apply says. Until the new one completes the diff --git a/internal/link/access_act_test.go b/internal/link/access_act_test.go index 0bd38f6..ee5c7e8 100644 --- a/internal/link/access_act_test.go +++ b/internal/link/access_act_test.go @@ -3,9 +3,14 @@ package link import ( "crypto/ed25519" "encoding/json" + "path/filepath" "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 @@ -68,6 +73,102 @@ 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) + } + 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 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 := &FileState{Path: filepath.Join(t.TempDir(), HandActsFile)} + 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, diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go new file mode 100644 index 0000000..7b902bc --- /dev/null +++ b/internal/link/engineseat.go @@ -0,0 +1,260 @@ +package link + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "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..`, and on the same +// address naming its caller, `mesh.seat.node-engine.call...`, 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, `.`; 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) +} diff --git a/internal/link/handacts.go b/internal/link/handacts.go new file mode 100644 index 0000000..5a53f69 --- /dev/null +++ b/internal/link/handacts.go @@ -0,0 +1,665 @@ +package link + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "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 +// state beside its node's state (HandActsFile), 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 + +// keptHandActs is how many acts the engine keeps, newest last. +const keptHandActs = 200 + +// 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 + +// HandActsFile is where the engine keeps its acts and its asks, beside its state. +const HandActsFile = "hand-acts.json" + +// 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 (FileState beside the node's state). + State *FileState + // 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": + 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 -------------------------------------------------------------------------------------------------- + +const actsKey = "acts" + +func (h *HandActs) read() ([]HandAct, uint64, error) { + e, err := h.State.Get(actsKey) + if err != nil || e == nil { + return nil, 0, err + } + var out []HandAct + if err := json.Unmarshal(e.Value, &out); err != nil { + return nil, 0, fmt.Errorf("the engine's record of its acts cannot be read: %w", err) + } + return out, e.Revision, nil +} + +// change reads the record, lets edit change it, and writes it back by compare-and-set, trying again on a race. +func (h *HandActs) change(edit func([]HandAct) ([]HandAct, error)) error { + for range 5 { + acts, rev, err := h.read() + if err != nil { + return err + } + acts, err = edit(acts) + if err != nil { + return err + } + if len(acts) > keptHandActs { + acts = acts[len(acts)-keptHandActs:] + } + if rev == 0 { + _, err = h.State.Create(actsKey, acts) + } else { + _, err = h.State.Update(actsKey, acts, rev) + } + if !errors.Is(err, stdio.ErrChanged) { + return err + } + } + return errors.New("the engine's record of its acts kept changing under its write") +} + +func (h *HandActs) record(a HandAct) error { + return h.change(func(acts []HandAct) ([]HandAct, error) { return append(acts, a), nil }) +} + +// end marks the open act of an ask ended, once: false when it was not open. +func (h *HandActs) end(ask string, set func(*HandAct)) (HandAct, bool, error) { + var ended HandAct + found := false + err := h.change(func(acts []HandAct) ([]HandAct, error) { + for i := range acts { + if acts[i].Ask == ask && acts[i].State == HandAsked { + set(&acts[i]) + acts[i].Ended = h.now().UTC() + ended, found = acts[i], true + } + } + return acts, nil + }) + return ended, found, err +} + +// 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 + } + 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.record(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") + } + acts, _, err := h.read() + if err != nil { + return nil, err + } + for i := len(acts) - 1; i >= 0; i-- { + a := acts[i] + if a.Act == ActSetuidSearch && a.State == HandDone && now.Sub(a.At) < SearchEvery { + return nil, fmt.Errorf("a search was started at %s by %s, and one is started in any hour: the next may start "+ + "at %s", a.At.Format(time.RFC3339), a.Caller, a.At.Add(SearchEvery).Format(time.RFC3339)) + } + } + 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() + } + if rerr := h.record(a); rerr != nil { + return nil, rerr + } + if err != nil { + return nil, err + } + return map[string]any{"node": h.Node, "said": said}, nil +} + +// ---- 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 { + acts, _, err := h.read() + if err != nil { + return true // unread is never a reason to ask again + } + 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 + err := h.change(func(acts []HandAct) ([]HandAct, error) { + for _, o := range acts { + if o.Act == act && o.Subject == j.Subject && o.State == HandAsked { + return nil, fmt.Errorf("%s of %s is asked already: the ask %s waits for the operator's answer", act, + j.Subject, o.Ask) + } + } + return append(acts, a), nil + }) + if err != nil { + 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) + } + 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 { + acts, _, err := h.read() + if err != nil { + return err + } + var a *HandAct + for i := range acts { + if acts[i].Ask == w.Ask { + a = &acts[i] + } + } + 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) + } + } + } +} + +// ---- the engine's own state --------------------------------------------------------------------------------------- + +// FileState is a store of keys with revisions in one file beside the node's state, written whole and renamed into +// place: the engine's own state, which asker.Client keeps its asks in and HandActs its acts (compare-and-set on a +// revision, as a bucket's). +type FileState struct { + Path string + mu sync.Mutex +} + +type fileEntry struct { + Value json.RawMessage `json:"value"` + Revision uint64 `json:"revision"` +} + +func (f *FileState) load() (map[string]fileEntry, error) { + raw, err := os.ReadFile(f.Path) + if errors.Is(err, os.ErrNotExist) { + return map[string]fileEntry{}, nil + } + if err != nil { + return nil, fmt.Errorf("the engine's own state %s cannot be read: %w", f.Path, err) + } + out := map[string]fileEntry{} + if err := json.Unmarshal(raw, &out); err != nil { + return nil, fmt.Errorf("the engine's own state %s is not a state: %w", f.Path, err) + } + return out, nil +} + +func (f *FileState) save(all map[string]fileEntry) error { + raw, err := json.Marshal(all) + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(f.Path), 0o700); err != nil { + return err + } + tmp := f.Path + ".tmp" + if err := os.WriteFile(tmp, raw, 0o600); err != nil { + return err + } + return os.Rename(tmp, f.Path) +} + +// Get is a key and its revision; nil when it is not kept. +func (f *FileState) Get(key string) (*stdio.StateEntry, error) { + f.mu.Lock() + defer f.mu.Unlock() + all, err := f.load() + if err != nil { + return nil, err + } + e, ok := all[key] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: key, Value: e.Value, Revision: e.Revision}, nil +} + +// Create writes a key that is not kept, or says stdio.ErrChanged. +func (f *FileState) Create(key string, value any) (uint64, error) { return f.put(key, value, 0, true) } + +// Update writes a key at the revision read, or says stdio.ErrChanged. +func (f *FileState) Update(key string, value any, revision uint64) (uint64, error) { + return f.put(key, value, revision, false) +} + +func (f *FileState) put(key string, value any, revision uint64, create bool) (uint64, error) { + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + f.mu.Lock() + defer f.mu.Unlock() + all, err := f.load() + if err != nil { + return 0, err + } + was, kept := all[key] + if (create && kept) || (!create && (!kept || was.Revision != revision)) { + return 0, stdio.ErrChanged + } + var top uint64 + for _, e := range all { + if e.Revision > top { + top = e.Revision + } + } + all[key] = fileEntry{Value: raw, Revision: top + 1} + if err := f.save(all); err != nil { + return 0, err + } + return top + 1, nil +} diff --git a/internal/link/handover_test.go b/internal/link/handover_test.go index 3442a33..3734e48 100644 --- a/internal/link/handover_test.go +++ b/internal/link/handover_test.go @@ -199,6 +199,30 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) { } 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: &FileState{Path: t.TempDir() + "/" + HandActsFile}, + 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 @@ -237,4 +261,24 @@ 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 §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) + } + } + }) } diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index da68fa2..cad3373 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -9,6 +9,7 @@ import ( "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 @@ -48,7 +49,10 @@ 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 - node string + // 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 // 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 @@ -258,6 +262,12 @@ 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() } @@ -301,10 +311,12 @@ 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) + _ Bus = (*natsLink)(nil) + _ HealthBus = (*natsLink)(nil) + _ HaveBus = (*natsLink)(nil) + _ ToolBus = (*natsLink)(nil) + _ Asker = (*natsLink)(nil) + _ Asked = (*natsLink)(nil) + _ EngineSeats = (*natsLink)(nil) + _ OperatorAsks = (*natsLink)(nil) ) diff --git a/internal/link/run.go b/internal/link/run.go index 3d17258..e66a2fa 100644 --- a/internal/link/run.go +++ b/internal/link/run.go @@ -7,6 +7,8 @@ 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. @@ -200,6 +202,19 @@ 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) diff --git a/internal/link/session.go b/internal/link/session.go index 31f8a5a..2c42dc5 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -8,6 +8,7 @@ import ( "sync" "time" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/health" ) @@ -47,6 +48,9 @@ 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 + // 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 @@ -337,3 +341,49 @@ 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 +} diff --git a/internal/link/setuid_search_test.go b/internal/link/setuid_search_test.go index 5e90229..b9c7c72 100644 --- a/internal/link/setuid_search_test.go +++ b/internal/link/setuid_search_test.go @@ -57,4 +57,34 @@ 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: &FileState{Path: t.TempDir() + "/" + HandActsFile}, 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) + } + }) }