Every engine's node-seat subscriptions were refused at 12:10 by a user list composed without them. nats asks a refused subscription again only on a new connection, and a user-list reload makes none, so the seat answered nobody all afternoon and the messenger held every Telegram ask. The endpoints also lacked the kind and scope discovery reads, so the console said nothing in the mesh is called node-engine. Quick fix: restart the engines. Thorough fix (chosen, the second time a stale user list left an engine deaf): a refused subscription of any kind is said and the link is opened again after a pause (1 min, doubling to 30 min), so every subscription the engine makes is asked again; the seat's endpoints carry kind, seat, scope, module, description and schema like every tool runner's.
385 lines
17 KiB
Go
385 lines
17 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"crypto/ed25519"
|
|
"encoding/json"
|
|
"errors"
|
|
"sort"
|
|
"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
|
|
// (mesh-controller internal/link, TestTheHandOverAskAndAnswerKeepTheirFieldNames).
|
|
func TestTheHandOverAskAndAnswerKeepTheirFieldNames(t *testing.T) {
|
|
body, _ := json.Marshal(HandOverAsk{Node: "laptop", Path: "/srv/notes", By: "jo", Expires: time.Now(), Nonce: "n"})
|
|
if got := keysOf(t, body); got != "by expires node nonce path" {
|
|
t.Fatalf("the ask's fields are %q", got)
|
|
}
|
|
body, _ = json.Marshal(SignedHandOver{Ask: []byte("{}"), Signature: []byte("s")})
|
|
if got := keysOf(t, body); got != "ask signature" {
|
|
t.Fatalf("the signed ask's fields are %q", got)
|
|
}
|
|
body, _ = json.Marshal(HandOverAnswer{Said: "s", Refused: "r"})
|
|
if got := keysOf(t, body); got != "refused said" {
|
|
t.Fatalf("the answer's fields are %q", got)
|
|
}
|
|
if AskHandOverSubject("laptop") != "mesh.node.laptop.ask.hand-over" {
|
|
t.Fatalf("the subject is %q", AskHandOverSubject("laptop"))
|
|
}
|
|
if HandOverContext != "novox-mesh hand-over v1\n" {
|
|
t.Fatalf("the signing context is %q; the controller holds the same words", HandOverContext)
|
|
}
|
|
}
|
|
|
|
func keysOf(t *testing.T, body []byte) string {
|
|
t.Helper()
|
|
var m map[string]any
|
|
if err := json.Unmarshal(body, &m); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var keys []string
|
|
for k := range m {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Strings(keys)
|
|
return strings.Join(keys, " ")
|
|
}
|
|
|
|
const aNonce = "00112233445566778899aabbccddeeff"
|
|
|
|
// signedAsk is an ask as the controller makes it, signed with key over the context and its bytes.
|
|
func signedAsk(t *testing.T, key ed25519.PrivateKey, ask HandOverAsk) []byte {
|
|
t.Helper()
|
|
raw, err := json.Marshal(ask)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
body, err := json.Marshal(SignedHandOver{Ask: raw, Signature: ed25519.Sign(key, append([]byte(HandOverContext), raw...))})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return body
|
|
}
|
|
|
|
// anAsk is a good ask at now.
|
|
func anAsk(now time.Time) HandOverAsk {
|
|
return HandOverAsk{Node: "n1", Path: "/srv/notes", By: " jo through mesh-cli on anchor ", Expires: now.Add(HandOverGood),
|
|
Nonce: aNonce}
|
|
}
|
|
|
|
// **An ask is believed only signed by the mesh, for this node, unexpired, once** (review of issue 356): a
|
|
// responder can reach the subject through a reply, so the subject proves nothing. Each refusal below hands
|
|
// the handler nothing; the good ask reaches it with its path and who asked, and the same ask again is refused.
|
|
func TestAHandOverAskIsBelievedOnlySignedForThisNodeUnexpiredAndOnce(t *testing.T) {
|
|
m, key := aMember(t)
|
|
_, stranger, _ := ed25519.GenerateKey(nil)
|
|
now := time.Date(2026, 10, 9, 18, 0, 0, 0, time.UTC)
|
|
var given []string
|
|
handle := func(path, by string) (string, error) {
|
|
given = append(given, path+" by "+by)
|
|
if path == "/refused" {
|
|
return "", errors.New("not a directory this machine uses as found; nothing was handed over")
|
|
}
|
|
return path + " is handed to the mesh", nil
|
|
}
|
|
check := HandOverCheck{Signer: m.Signer, Node: m.Node, Now: func() time.Time { return now }, Nonces: &Nonces{}}
|
|
with := func(change func(*HandOverAsk)) HandOverAsk {
|
|
a := anAsk(now)
|
|
change(&a)
|
|
return a
|
|
}
|
|
good, _ := json.Marshal(anAsk(now))
|
|
unsignedBody, _ := json.Marshal(SignedHandOver{Ask: good})
|
|
// Signed as a declaration is, over the bytes alone: not a hand-over's signature.
|
|
asDeclaration, _ := json.Marshal(SignedHandOver{Ask: good, Signature: ed25519.Sign(key, good)})
|
|
refusals := map[string][]byte{
|
|
"too large": signedAsk(t, key, with(func(a *HandOverAsk) { a.Path = "/" + strings.Repeat("a", MaxHandOverAsk) })),
|
|
"not a signed hand-over": []byte(`not json`),
|
|
"unsigned": unsignedBody,
|
|
"signed by another key": signedAsk(t, stranger, anAsk(now)),
|
|
"signed as a declaration": asDeclaration,
|
|
"the bare ask": good,
|
|
"for another node": signedAsk(t, key, with(func(a *HandOverAsk) { a.Node = "n2" })),
|
|
"expired": signedAsk(t, key, with(func(a *HandOverAsk) { a.Expires = now })),
|
|
"no expiry": signedAsk(t, key, with(func(a *HandOverAsk) { a.Expires = time.Time{} })),
|
|
"good for too long": signedAsk(t, key, with(func(a *HandOverAsk) { a.Expires = now.Add(HandOverGood + HandOverSkew + time.Second) })),
|
|
"no nonce": signedAsk(t, key, with(func(a *HandOverAsk) { a.Nonce = "" })),
|
|
"a short nonce": signedAsk(t, key, with(func(a *HandOverAsk) { a.Nonce = "0011" })),
|
|
"nobody": signedAsk(t, key, with(func(a *HandOverAsk) { a.By = " " })),
|
|
"two lines": signedAsk(t, key, with(func(a *HandOverAsk) { a.By = "jo\nroot" })),
|
|
"a name too long": signedAsk(t, key, with(func(a *HandOverAsk) { a.By = strings.Repeat("x", MaxHandOverBy+1) })),
|
|
"a NUL in the name": signedAsk(t, key, with(func(a *HandOverAsk) { a.By = "jo\x00" })),
|
|
"no directory": signedAsk(t, key, with(func(a *HandOverAsk) { a.Path = "" })),
|
|
}
|
|
for name, body := range refusals {
|
|
a := AnswerHandOver(check, handle, body)
|
|
if a.Refused == "" || a.Said != "" || !strings.Contains(a.Refused, "nothing was handed over") {
|
|
t.Fatalf("%s was answered %+v; want a refusal saying nothing was handed over", name, a)
|
|
}
|
|
}
|
|
if len(given) != 0 {
|
|
t.Fatalf("the handler was given %v although every ask was refused", given)
|
|
}
|
|
if a := AnswerHandOver(HandOverCheck{Node: "n1", Nonces: &Nonces{}}, handle, signedAsk(t, key, anAsk(time.Now()))); a.Refused == "" {
|
|
t.Fatal("an engine with no key believed an ask")
|
|
}
|
|
if a := AnswerHandOver(check, nil, signedAsk(t, key, anAsk(now))); a.Refused == "" {
|
|
t.Fatal("an engine with no handler took a hand-over")
|
|
}
|
|
|
|
body := signedAsk(t, key, anAsk(now))
|
|
a := AnswerHandOver(check, handle, body)
|
|
if a.Said != "/srv/notes is handed to the mesh" || a.Refused != "" {
|
|
t.Fatalf("the good ask was answered %+v", a)
|
|
}
|
|
if again := AnswerHandOver(check, handle, body); !strings.Contains(again.Refused, "taken already") {
|
|
t.Fatalf("the same ask again was answered %+v", again)
|
|
}
|
|
refused := signedAsk(t, key, with(func(a *HandOverAsk) { a.Path = "/refused"; a.Nonce = strings.Repeat("ab", 16) }))
|
|
if a := AnswerHandOver(check, handle, refused); !strings.HasPrefix(a.Refused, "not a directory") {
|
|
t.Fatalf("the handler's refusal was answered %+v", a)
|
|
}
|
|
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)
|
|
if len(check.Nonces.seen) != 1 {
|
|
t.Fatalf("expired nonces are kept: %v", check.Nonces.seen)
|
|
}
|
|
}
|
|
|
|
// **A request with no reply is nobody's question**: nothing is judged, recorded or said. One with a reply is
|
|
// answered with the judgement.
|
|
func TestAHandOverIsAnsweredOnlyWhenItCarriesAReply(t *testing.T) {
|
|
m, key := aMember(t)
|
|
now := time.Now()
|
|
called := 0
|
|
handle := func(string, string) (string, error) { called++; return "handed", nil }
|
|
check := HandOverCheck{Signer: m.Signer, Node: m.Node, Nonces: &Nonces{}}
|
|
if body, answer := HandOverReply(check, handle, "", signedAsk(t, key, anAsk(now))); answer || body != nil || called != 0 {
|
|
t.Fatalf("a request with no reply was answered %q %v, handled %d", body, answer, called)
|
|
}
|
|
body, answer := HandOverReply(check, handle, "_INBOX.x", signedAsk(t, key, anAsk(now)))
|
|
var said HandOverAnswer
|
|
if !answer || json.Unmarshal(body, &said) != nil || said.Said != "handed" || called != 1 {
|
|
t.Fatalf("answered %q %v, handled %d", body, answer, called)
|
|
}
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
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)}
|
|
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)
|
|
go func() { served <- serve(ctx, l, m, q, nil, time.Second) }()
|
|
select {
|
|
case h := <-l.given:
|
|
if said, err := h("/srv/notes", "jo"); err != nil || said != "the queue's: /srv/notes by jo" {
|
|
t.Fatalf("the link was given another handler: %q %v", said, err)
|
|
}
|
|
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
|
|
// not heard; a message with no reply is answered by nobody.
|
|
func TestNatsAHandOverIsAskedOfTheNode(t *testing.T) {
|
|
conn, js := aBus(t)
|
|
m, key := aMember(t)
|
|
l := &natsLink{conn: conn, js: js, node: m.Node, signer: m.Signer}
|
|
handled := make(chan string, 4)
|
|
l.HearHandOvers(func(path, by string) (string, error) { handled <- path; return "handed " + path, nil })
|
|
defer l.Close()
|
|
if err := conn.Flush(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ask := anAsk(time.Now())
|
|
ask.Nonce = strings.Repeat("cd", 16)
|
|
reply, err := conn.Request(AskHandOverSubject(m.Node), signedAsk(t, key, ask), 5*time.Second)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var said HandOverAnswer
|
|
if err := json.Unmarshal(reply.Data, &said); err != nil || said.Said != "handed /srv/notes" {
|
|
t.Fatalf("answered %s, %v", reply.Data, err)
|
|
}
|
|
ask.Nonce = strings.Repeat("ef", 16)
|
|
if err := conn.Publish(AskHandOverSubject(m.Node), signedAsk(t, key, ask)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := conn.Request(AskHandOverSubject("another"), signedAsk(t, key, ask), 300*time.Millisecond); err == nil {
|
|
t.Fatal("another node's ask was answered here")
|
|
}
|
|
if err := conn.Flush(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
time.Sleep(100 * time.Millisecond)
|
|
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")
|
|
}
|
|
// Announced in the shape the console reads a seat's verb in (sweep 2026-10-30): without kind and scope it listed
|
|
// none of it, and <node>/node-engine.root-free answered that nothing in the mesh is called node-engine.
|
|
info, err := conn.Request("$SRV.INFO."+EngineSeat, nil, 5*time.Second)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var said struct {
|
|
Endpoints []struct {
|
|
Subject string `json:"subject"`
|
|
Metadata map[string]string `json:"metadata"`
|
|
} `json:"endpoints"`
|
|
}
|
|
if err := json.Unmarshal(info.Data, &said); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
announced := 0
|
|
for _, e := range said.Endpoints {
|
|
md := e.Metadata
|
|
if md["kind"] != "seat" || md["scope"] != "node" || md["seat"] != EngineSeat || md["module"] != "mesh-host" ||
|
|
md["node"] != m.Node || md["description"] == "" || !strings.HasPrefix(md["schema"], "{") {
|
|
t.Errorf("the endpoint on %s announces %v", e.Subject, md)
|
|
}
|
|
if e.Subject == EngineToolSubject("root-free", m.Node) {
|
|
announced++
|
|
}
|
|
}
|
|
if announced != 1 {
|
|
t.Errorf("root-free is announced on its tool subject %d times", announced)
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
})
|
|
}
|