jschoubben
98f4ff6163
The bus module on the bus node runs from what its agent holds (its own composition, or the declaration kept before the switch: every novox have read 2026-11-02 names its own composition or none, and nats names a build alone), while composed-<node> was written by the controller, which ended (issue 1126). The follower now looks at the agent's record of what runs whenever it changed, writes again only when the file it names moved, and trusts no composition signer. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
398 lines
16 KiB
Go
398 lines
16 KiB
Go
package userlist
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/seal"
|
|
"github.com/novox/mesh-host/internal/agent"
|
|
)
|
|
|
|
const node = "novox"
|
|
|
|
func fpOf(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) }
|
|
|
|
// feed is a watch the test writes to: the keys there before the watch, Caught, then each put. watching is closed once
|
|
// the watch began, so a put after it is never handed before the keys there before it.
|
|
type feed struct {
|
|
ch chan agent.Update
|
|
rev uint64
|
|
first []agent.Update
|
|
// watching is closed once the watch caught up, so a later put is not read before the keys there before it.
|
|
watching chan struct{}
|
|
once sync.Once
|
|
// err refuses the watch, as the bus refuses a watch it grants nothing of.
|
|
err error
|
|
}
|
|
|
|
func (f *feed) Watch(context.Context, string) (<-chan agent.Update, error) {
|
|
if f.err != nil {
|
|
return nil, f.err
|
|
}
|
|
for _, u := range f.first {
|
|
f.ch <- u
|
|
}
|
|
f.ch <- agent.Update{Caught: true}
|
|
f.once.Do(func() {
|
|
if f.watching != nil {
|
|
close(f.watching)
|
|
}
|
|
})
|
|
return f.ch, nil
|
|
}
|
|
|
|
func list(users ...string) string {
|
|
var b strings.Builder
|
|
b.WriteString("authorization {\n users = [\n")
|
|
for _, u := range users {
|
|
b.WriteString(` {user: "` + u + `", password: "$2a$11$x"}` + "\n")
|
|
}
|
|
b.WriteString(" ]\n}\n")
|
|
return b.String()
|
|
}
|
|
|
|
// comp is a composition of module at fp as this node runs it, holding the user list content (none when empty); by
|
|
// digest when byDigest, as a file over 64 KiB is given; for another node than this one when asked.
|
|
func comp(module, fp, content string, p put) delivery.Composed {
|
|
server, _ := json.Marshal(map[string]any{"id": module + ".server-conf", "type": "file", "module": module,
|
|
"path": "/var/lib/nats-module/conf/nats.conf", "content": "include accounts.conf\n"})
|
|
rs := []json.RawMessage{server}
|
|
if content != "" {
|
|
r := map[string]any{"id": module + ".bus-users", "type": "file", "module": module,
|
|
"path": "/var/lib/nats-module/conf/accounts.conf", "content": content, "mode": "0600"}
|
|
if p.byDigest {
|
|
delete(r, "content")
|
|
r["digest"], r["size"] = Digest([]byte(content)), len(content)
|
|
}
|
|
users, _ := json.Marshal(r)
|
|
rs = append(rs, users)
|
|
}
|
|
c := delivery.Composed{Node: node, Module: module, FP: fp, Resources: rs}
|
|
if p.asNode != "" {
|
|
c.Node = p.asNode
|
|
}
|
|
return c
|
|
}
|
|
|
|
// gput is the grants part's key as the bus node hears it at the watch's start (ADR 0547 clauses 7, 8): which list to
|
|
// follow, and the list sealed to the node's key; one sealed to another key, or one that does not open to the key's digest.
|
|
type gput struct {
|
|
follow, content string
|
|
otherKey, badDigest bool
|
|
// after puts it once the grants part's list was written, as a hand-back comes after a take-over.
|
|
after bool
|
|
// unreadable and deleted make the key a value that does not read, or a deleted key, instead of a list (774/S31).
|
|
unreadable, deleted bool
|
|
}
|
|
|
|
type put struct {
|
|
module, fp, content string
|
|
byDigest bool
|
|
asNode string
|
|
}
|
|
|
|
// Defends novox/hq ADR 0547 §8: the bus node writes the grants part's list alone, only when it changed, and reports it,
|
|
// into the file the bus module it runs declares (issue 1126, B17: what runs, never a composition bucket).
|
|
func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) {
|
|
one, two, three := list("controller", "a"), list("controller", "a", "b"), list("controller", "a", "b", "c")
|
|
five := list("controller", "a", "b", "c", "d")
|
|
for _, c := range []struct {
|
|
name string
|
|
before, after []put
|
|
failWrite bool
|
|
onDisk string
|
|
wantWrites []string // contents written, in order
|
|
wantUsers int
|
|
wantErr string
|
|
wantRefused bool
|
|
grants []gput
|
|
// grantsRefused refuses the watch of the grants part's key; kept is the key last heard, kept on the machine;
|
|
// wantUnwatched says the have then says why it is not watched.
|
|
grantsRefused bool
|
|
kept *gput
|
|
wantUnwatched bool
|
|
// wantSource and wantFollows are what the last have says it took the list from and follows.
|
|
wantSource, wantFollows string
|
|
}{
|
|
// Issue 1126, A6 (ADR 0547 clauses 7, 8, 11): the grants part's list alone, written into the file the bus module
|
|
// this node runs declares; the list a composition carries is never written, whatever becomes of the grants key.
|
|
// B17: the file's place is read from what runs, as the agent holds it, never from a composition bucket.
|
|
{name: "the grants part's list is written into the file the bus module it runs declares",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowGrants, content: three}},
|
|
wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants},
|
|
{name: "a bus module it runs by digest still names the file",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two, byDigest: true}}, grants: []gput{{follow: delivery.FollowGrants, content: three}},
|
|
wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants},
|
|
{name: "a newer composition's list is not written",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}},
|
|
after: []put{{module: "nats", fp: fpOf("b"), content: five}},
|
|
grants: []gput{{follow: delivery.FollowGrants, content: three}},
|
|
wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants},
|
|
{name: "a composition without the list is ignored", before: []put{{module: "web", fp: fpOf("a"), content: ""}},
|
|
grants: []gput{{follow: delivery.FollowGrants, content: three}}},
|
|
{name: "another node's composition declares no file",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two, asNode: "ace"}}, grants: []gput{{follow: delivery.FollowGrants, content: three}}},
|
|
{name: "an unreadable grants key with none heard before writes nothing, never the composition's list",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{unreadable: true}},
|
|
onDisk: two, wantErr: "key is not heard", wantRefused: true, wantSource: delivery.GrantsUserListSource},
|
|
{name: "a refused watch with no key kept writes nothing and says it is not watched",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grantsRefused: true, wantUnwatched: true,
|
|
onDisk: two, wantErr: "key is not heard", wantRefused: true, wantSource: delivery.GrantsUserListSource},
|
|
{name: "a refused watch keeps the key last heard",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grantsRefused: true, wantUnwatched: true,
|
|
kept: &gput{follow: delivery.FollowGrants, content: three},
|
|
wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants},
|
|
{name: "a grants key that turns unreadable keeps the loaded list",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}},
|
|
grants: []gput{{follow: delivery.FollowGrants, content: three}, {unreadable: true, after: true}},
|
|
wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants},
|
|
{name: "a grants key that is deleted keeps the loaded list",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}},
|
|
grants: []gput{{follow: delivery.FollowGrants, content: three}, {deleted: true, after: true}},
|
|
wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants},
|
|
{name: "a grants list that drops more than half is refused and the file kept",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowGrants, content: one}},
|
|
onDisk: five, wantUsers: 2, wantErr: "drops more than half", wantRefused: true, wantSource: delivery.GrantsUserListSource},
|
|
{name: "a failed write is said and kept", before: []put{{module: "nats", fp: fpOf("a"), content: two}},
|
|
grants: []gput{{follow: delivery.FollowGrants, content: three}}, failWrite: true,
|
|
wantWrites: []string{three}, wantUsers: 4, wantErr: "disk full", wantSource: delivery.GrantsUserListSource},
|
|
{name: "a grants list that does not open to its digest is refused and the file kept",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowGrants, content: three, badDigest: true}},
|
|
onDisk: two, wantErr: "could not be had", wantRefused: true, wantSource: delivery.GrantsUserListSource},
|
|
{name: "a grants list sealed to another key is refused and the file kept",
|
|
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowGrants, content: three, otherKey: true}},
|
|
onDisk: two, wantErr: "sealed to another key", wantRefused: true, wantSource: delivery.GrantsUserListSource},
|
|
} {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
var mu sync.Mutex
|
|
var running []delivery.Composed
|
|
var writes []string
|
|
var reports []Written
|
|
reported := make(chan struct{}, 64)
|
|
gf := &feed{ch: make(chan agent.Update, 64)}
|
|
if c.grantsRefused {
|
|
gf.err = errors.New("Permissions Violation for Publish to $JS.API.STREAM.INFO.KV_mesh-grants_user-list")
|
|
}
|
|
var grantsAfter []agent.Update
|
|
var keptRaw []byte
|
|
sealed := map[string][]byte{}
|
|
nodeKey, err := seal.GenerateKeyPair()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
gs := c.grants
|
|
if c.kept != nil {
|
|
gs = append(append([]gput(nil), gs...), *c.kept)
|
|
}
|
|
for i, g := range gs {
|
|
if g.unreadable || g.deleted {
|
|
u := agent.Update{Key: delivery.GrantsUserListKey, Value: []byte("not a user-list key"), Revision: uint64(i + 1), Deleted: g.deleted}
|
|
grantsAfter = append(grantsAfter, u)
|
|
continue
|
|
}
|
|
to := nodeKey.PublicKey
|
|
if g.otherKey {
|
|
other, _ := seal.GenerateKeyPair()
|
|
to = other.PublicKey
|
|
}
|
|
digest := delivery.UserListDigest([]byte(g.content))
|
|
if g.badDigest {
|
|
digest = delivery.UserListDigest([]byte("another list"))
|
|
}
|
|
object, err := delivery.SealUserList([]byte(g.content), to)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sealed[delivery.UserListObjectName(digest)] = object
|
|
raw, err := delivery.EncodeGrantsUserList(delivery.GrantsUserList{Follow: g.follow, Digest: digest,
|
|
Object: delivery.UserListObjectName(digest), SealedTo: delivery.SealingKeyID(to), Node: node, Revision: uint64(i + 1)})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
u := agent.Update{Key: delivery.GrantsUserListKey, Value: raw, Revision: uint64(i + 1)}
|
|
switch {
|
|
case c.kept != nil && i == len(gs)-1:
|
|
keptRaw = raw
|
|
case g.after:
|
|
grantsAfter = append(grantsAfter, u)
|
|
default:
|
|
gf.first = append(gf.first, u)
|
|
}
|
|
}
|
|
fl := New(Env{Node: node, Grants: gf,
|
|
Running: func() []delivery.Composed {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
return append([]delivery.Composed(nil), running...)
|
|
},
|
|
Kept: func() ([]byte, uint64) { return keptRaw, 9 },
|
|
Keep: func([]byte, uint64) {},
|
|
Backoff: func(int) time.Duration { return time.Hour },
|
|
Fetch: func(_ context.Context, name string) ([]byte, error) {
|
|
if o, ok := sealed[name]; ok {
|
|
return o, nil
|
|
}
|
|
return nil, errors.New("no object " + name)
|
|
},
|
|
Open: func(b seal.Box) ([]byte, error) {
|
|
out, err := seal.Open(b, nodeKey.PrivateKey)
|
|
return []byte(out), err
|
|
},
|
|
Key: func() string { return nodeKey.PublicKey },
|
|
OnDisk: func(json.RawMessage) ([]byte, error) {
|
|
if c.onDisk == "" {
|
|
return nil, nil
|
|
}
|
|
return []byte(c.onDisk), nil
|
|
},
|
|
Write: func(_ context.Context, module string, r json.RawMessage, body []byte) (string, bool, error) {
|
|
mu.Lock()
|
|
writes = append(writes, string(body))
|
|
mu.Unlock()
|
|
if c.failWrite {
|
|
return "", false, errors.New("disk full")
|
|
}
|
|
return Digest(body), true, nil
|
|
},
|
|
Report: func(_ context.Context, module string, w Written) {
|
|
mu.Lock()
|
|
reports = append(reports, w)
|
|
mu.Unlock()
|
|
reported <- struct{}{}
|
|
},
|
|
Now: func() time.Time { return time.Date(2026, 10, 25, 21, 50, 0, 0, time.UTC) },
|
|
})
|
|
for _, p := range c.before {
|
|
running = append(running, comp(p.module, p.fp, p.content, p))
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
done := make(chan error, 1)
|
|
go func() { done <- fl.Run(ctx) }()
|
|
// What runs changes after the first list is written: the first write waits for the grants part's key.
|
|
if len(c.after) > 0 {
|
|
for until := time.Now().Add(5 * time.Second); time.Now().Before(until); time.Sleep(5 * time.Millisecond) {
|
|
mu.Lock()
|
|
n := len(writes)
|
|
mu.Unlock()
|
|
if n > 0 {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
for _, p := range c.after {
|
|
mu.Lock()
|
|
running = []delivery.Composed{comp(p.module, p.fp, p.content, p)}
|
|
mu.Unlock()
|
|
fl.RunsChanged()
|
|
}
|
|
if len(grantsAfter) > 0 {
|
|
for until := time.Now().Add(5 * time.Second); ; time.Sleep(5 * time.Millisecond) {
|
|
if _, st := fl.State(); st.Source == delivery.GrantsUserListSource || time.Now().After(until) {
|
|
break
|
|
}
|
|
}
|
|
for _, u := range grantsAfter {
|
|
gf.ch <- u
|
|
}
|
|
}
|
|
deadline := time.After(5 * time.Second)
|
|
for {
|
|
mu.Lock()
|
|
n := len(writes)
|
|
mu.Unlock()
|
|
if n >= len(c.wantWrites) {
|
|
break
|
|
}
|
|
select {
|
|
case <-deadline:
|
|
t.Fatalf("writes %d, want %d", n, len(c.wantWrites))
|
|
case <-time.After(5 * time.Millisecond):
|
|
}
|
|
}
|
|
if c.wantRefused {
|
|
select {
|
|
case <-reported:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("a refused list was not reported")
|
|
}
|
|
}
|
|
if c.wantSource != "" {
|
|
for until := time.Now().Add(5 * time.Second); ; {
|
|
if _, st := fl.State(); st.Source == c.wantSource && (c.wantErr == "" || st.Error != "") || time.Now().After(until) {
|
|
break
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
}
|
|
time.Sleep(50 * time.Millisecond) // nothing more is written
|
|
cancel()
|
|
<-done
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if strings.Join(writes, "|") != strings.Join(c.wantWrites, "|") {
|
|
t.Fatalf("written %q, want %q", writes, c.wantWrites)
|
|
}
|
|
module, st := fl.State()
|
|
if c.wantUnwatched != (st.Unwatched != "") {
|
|
t.Fatalf("the have says unwatched %q", st.Unwatched)
|
|
}
|
|
// While the grants part's list is followed, an apply of the bus module's composition writes not its list.
|
|
if cmp := comp("nats", fpOf("a"), two, put{}); c.wantFollows != "" {
|
|
held, err := fl.Hold(cmp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, _, hasList := userList(held.Resources); hasList || len(held.Resources) < 1 {
|
|
t.Fatalf("an apply while following %s holds the composition's list: %v", c.wantFollows, hasList)
|
|
}
|
|
// A list on the machine that cannot be read stops the apply; it is never left out to be removed.
|
|
if c.wantFollows == delivery.FollowGrants {
|
|
readErr := New(Env{Node: node, OnDisk: func(json.RawMessage) ([]byte, error) {
|
|
return nil, errors.New("permission denied")
|
|
}})
|
|
readErr.grants = &delivery.GrantsUserList{Follow: delivery.FollowGrants}
|
|
if _, err := readErr.Hold(cmp); err == nil || !strings.Contains(err.Error(), "could not be read") {
|
|
t.Fatalf("an unreadable list on the machine was held as %v", err)
|
|
}
|
|
}
|
|
}
|
|
if c.wantSource != "" && (st.Source != c.wantSource || st.Key != nodeKey.PublicKey ||
|
|
(c.wantFollows != "" && st.Follows != c.wantFollows)) {
|
|
t.Fatalf("the have says source %q follows %q key %t; want %q %q", st.Source, st.Follows, st.Key == nodeKey.PublicKey, c.wantSource, c.wantFollows)
|
|
}
|
|
if c.wantRefused {
|
|
if !strings.Contains(st.Error, c.wantErr) || st.Digest != Digest([]byte(c.onDisk)) || st.Users != c.wantUsers {
|
|
t.Fatalf("refused list's state %+v", st)
|
|
}
|
|
return
|
|
}
|
|
if len(c.wantWrites) == 0 {
|
|
if module != "" || len(reports) != 0 {
|
|
t.Fatalf("a node with no list wrote or said something: %q %+v", module, reports)
|
|
}
|
|
return
|
|
}
|
|
if module != "nats" || st.Users != c.wantUsers || st.Wanted != Digest([]byte(c.wantWrites[len(c.wantWrites)-1])) {
|
|
t.Errorf("state %q %+v", module, st)
|
|
}
|
|
if c.wantErr != "" {
|
|
if !strings.Contains(st.Error, c.wantErr) || st.Digest != "" {
|
|
t.Errorf("a failed write is not kept: %+v", st)
|
|
}
|
|
} else if st.Digest != st.Wanted || st.Error != "" {
|
|
t.Errorf("written %+v", st)
|
|
}
|
|
if len(reports) == 0 || reports[len(reports)-1] != st {
|
|
t.Errorf("reports %+v, last state %+v", reports, st)
|
|
}
|
|
})
|
|
}
|
|
}
|