356 lines
16 KiB
Go
356 lines
16 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")
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
})
|
|
}
|