From 0184e2a5b895a96abe3bf0b022762d62136501dc Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 27 Oct 2026 09:27:54 +0100 Subject: [PATCH] Watch the have bucket only where the node's grant allows, so a node composing itself is not refused by the bus (issue 1126) The fold's watch of every have key ('>') was refused (permissions violation on CONSUMER.CREATE KV_node-engine_have), and every self-composing agent stopped after the seed. The composer now watches '*.node' (records) and '.>' (its own haves), as buscompose grants a node; the shadow watches '.>' too. The selfcompose tests check every watch and get against PermissionsFor(KindNode), and the agent's integration tests run the agent as the node under that grant. --- cmd/mesh-host/nodeagent_integration_test.go | 48 ++++++-- internal/selfcompose/selfcompose.go | 19 ++-- internal/selfcompose/selfcompose_test.go | 115 +++++++++++++++++--- internal/selfcompose/shadow.go | 2 +- 4 files changed, 144 insertions(+), 40 deletions(-) diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 06bc6c8..ad46d3c 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -14,6 +14,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" @@ -29,9 +30,14 @@ import ( const itNode = "it1" type itBed struct { - t *testing.T - conn *nats.Conn - js nats.JetStreamContext + t *testing.T + conn *nats.Conn + js nats.JetStreamContext + // nodeConn and nodeJS are the node's own, under its real grant (buscompose.PermissionsFor, KindNode): the agent runs + // on them, so a watch or a read the grant does not allow fails here as it fails on the bus (issue 1126, 2026-10-27 + // 09:21: a watch of every have key stopped every node agent). conn is the test's, unrestricted, for setting up. + nodeConn *nats.Conn + nodeJS nats.JetStreamContext wants nats.KeyValue composed nats.KeyValue haves nats.KeyValue @@ -44,9 +50,22 @@ type itBed struct { mu sync.Mutex } -func aServer(t *testing.T) *nats.Conn { +// aServer is an embedded bus with JetStream and two users: the test's, unrestricted, and the node's, granted exactly what +// buscompose.PermissionsFor gives a node (KindNode); it answers a connection of each. +func aServer(t *testing.T) (admin, node *nats.Conn) { t.Helper() - srv, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + perms, err := buscompose.PermissionsFor(buscompose.Principal{Kind: buscompose.KindNode, Node: itNode}) + if err != nil { + t.Fatal(err) + } + granted := &server.Permissions{Publish: &server.SubjectPermission{Allow: perms.Publish, Deny: perms.PublishDeny}, + Subscribe: &server.SubjectPermission{Allow: perms.Subscribe}} + if perms.AllowResponses { + granted.Response = &server.ResponsePermission{MaxMsgs: 1, Expires: buscompose.ResponseTTL} + } + nodeUser := buscompose.Principal{Kind: buscompose.KindNode, Node: itNode}.Username() + srv, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true, + Users: []*server.User{{Username: "test", Password: "test"}, {Username: nodeUser, Password: "node", Permissions: granted}}}) if err != nil { t.Fatal(err) } @@ -55,12 +74,15 @@ func aServer(t *testing.T) *nats.Conn { t.Fatal("the embedded server did not start") } t.Cleanup(srv.Shutdown) - conn, err := nats.Connect(srv.ClientURL()) - if err != nil { + if admin, err = nats.Connect(srv.ClientURL(), nats.UserInfo("test", "test")); err != nil { t.Fatal(err) } - t.Cleanup(conn.Close) - return conn + t.Cleanup(admin.Close) + if node, err = nats.Connect(srv.ClientURL(), nats.UserInfo(nodeUser, "node"), nats.CustomInboxPrefix("_INBOX."+nodeUser)); err != nil { + t.Fatal(err) + } + t.Cleanup(node.Close) + return admin, node } func newItBed(t *testing.T) *itBed { @@ -75,11 +97,15 @@ func newItBed(t *testing.T) *itBed { builtFor = "arch" t.Cleanup(func() { builtFor = was }) - b := &itBed{t: t, conn: aServer(t), dir: t.TempDir()} + b := &itBed{t: t, dir: t.TempDir()} + b.conn, b.nodeConn = aServer(t) var err error if b.js, err = b.conn.JetStream(); err != nil { t.Fatal(err) } + if b.nodeJS, err = b.nodeConn.JetStream(); err != nil { + t.Fatal(err) + } composedBucket, _ := delivery.ComposedBucketFor(itNode) for _, c := range []struct { name string @@ -114,7 +140,7 @@ func (b *itBed) start(trustOperator bool) (stop func()) { } b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second, Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }} - b.session.Attach(context.Background(), link.OverNATS{Conn: b.conn, JS: b.js, Node: itNode}) + b.session.Attach(context.Background(), link.OverNATS{Conn: b.nodeConn, JS: b.nodeJS, Node: itNode}) ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) go func() { diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 24f190f..e060e39 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -117,8 +117,11 @@ func (c *Composer) inputs() []input { return []input{ {delivery.AssignerWantsBucket, "want." + node + ".>", (*Composer).takeWant, nil}, {delivery.PlacementBucket, ">", (*Composer).takePlacement, nil}, - // Every node's record, and this node's haves: what it runs inside its own composition. - {delivery.HaveBucket, ">", nil, (*Composer).takeHave}, + // Every node's record, and this node's own haves (what it runs inside its own composition): two watches, each + // within the node's grant (buscompose: readGrants of the have bucket at `*.node`, NodeKeysGrants at `.>`). + // A watch of every key is refused by the bus (issue 1126, 2026-10-27 09:21: the fold's `>` stopped every agent). + {delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil, (*Composer).takeRecordChanged}, + {delivery.HaveBucket, node + ".>", nil, (*Composer).takeModuleHave}, {delivery.ModulesBucket, ">", (*Composer).takeModule, nil}, {ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, } @@ -310,14 +313,10 @@ func (c *Composer) takeRecord(u agent.Update) { c.records[node] = *h.Record } -// takeHave takes a have key: a node's record (always composed again), or one of this node's modules' haves, which -// changes what is composed only when whether it runs inside the node's own composition changes. -func (c *Composer) takeHave(u agent.Update) bool { - if isNodeKey(u.Key) { - c.takeRecord(u) - return true - } - return c.takeModuleHave(u) +// takeRecordChanged takes a node's record, which is always composed again. +func (c *Composer) takeRecordChanged(u agent.Update) bool { + c.takeRecord(u) + return true } // takeModuleHave takes one of this node's modules' haves, and says whether it changed which it runs inside its own diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 3175f53..c8b3896 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/compose" "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -18,41 +19,95 @@ import ( const lemurs = `{"module":"lemurs","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"lemurs.service","state":"running","boot":"enabled"}]}` -// fakeBus is the buckets the composer watches, each a channel a test writes updates to, and the keys Get reads. +// fakeBus is the buckets the composer watches, each watch a channel that gets the updates a test writes to its bucket +// whose key its pattern matches, as the bus's own watches do, and the keys Get reads. It keeps every watch and get +// made, so a test checks each against the node's grant. type fakeBus struct { - mu sync.Mutex - chans map[string]chan agent.Update - kv map[string][]byte // bucket/key - said []string + mu sync.Mutex + watches []patternWatch + kv map[string][]byte // bucket/key + said []string + reads [][2]string // bucket, pattern or key: every watch and get } func newFakeBus() *fakeBus { - return &fakeBus{chans: map[string]chan agent.Update{}, kv: map[string][]byte{}} -} - -func (b *fakeBus) ch(bucket string) chan agent.Update { - b.mu.Lock() - defer b.mu.Unlock() - if b.chans[bucket] == nil { - b.chans[bucket] = make(chan agent.Update, 16) - } - return b.chans[bucket] + return &fakeBus{kv: map[string][]byte{}} } func (b *fakeBus) env() Env { return Env{Node: "n1", - Watch: func(_ context.Context, bucket, _ string) (<-chan agent.Update, error) { return b.ch(bucket), nil }, + Watch: func(_ context.Context, bucket, pattern string) (<-chan agent.Update, error) { + b.mu.Lock() + defer b.mu.Unlock() + w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)} + b.watches = append(b.watches, w) + b.reads = append(b.reads, [2]string{bucket, pattern}) + return w.ch, nil + }, Get: func(_ context.Context, bucket, key string) ([]byte, error) { b.mu.Lock() defer b.mu.Unlock() + b.reads = append(b.reads, [2]string{bucket, key}) return b.kv[bucket+"/"+key], nil }, Say: func(s string) { b.mu.Lock(); b.said = append(b.said, s); b.mu.Unlock() }} } +// send gives u to every watch of bucket whose pattern matches its key; Caught to every watch of bucket. +func (b *fakeBus) send(bucket string, u agent.Update) { + b.mu.Lock() + ws := append([]patternWatch(nil), b.watches...) + b.mu.Unlock() + for _, w := range ws { + if w.bucket == bucket && (u.Caught || matches(w.pattern, u.Key)) { + w.ch <- u + } + } +} + func (b *fakeBus) put(t *testing.T, bucket, key string, value []byte) { t.Helper() - b.ch(bucket) <- agent.Update{Key: key, Value: value} + b.send(bucket, agent.Update{Key: key, Value: value}) +} + +// grantAllows says whether the node's grant (buscompose.PermissionsFor, KindNode) lets it watch or read pattern of +// bucket: the consumer a watch or a get makes (`$JS.API.CONSUMER.CREATE.KV_..$KV..`) +// and the stream's information, matched as the bus matches a permission against a subject. +func grantAllows(t *testing.T, node, bucket, pattern string) bool { + t.Helper() + perms, err := buscompose.PermissionsFor(buscompose.Principal{Kind: buscompose.KindNode, Node: node}) + if err != nil { + t.Fatal(err) + } + allowed := func(subject string) bool { + for _, deny := range perms.PublishDeny { + if subjectMatches(deny, subject) { + return false + } + } + for _, p := range perms.Publish { + if subjectMatches(p, subject) { + return true + } + } + return false + } + stream := "KV_" + bucket + return allowed("$JS.API.STREAM.INFO."+stream) && allowed("$JS.API.CONSUMER.CREATE."+stream+".c1.$KV."+bucket+"."+pattern) +} + +// subjectMatches is the bus's match of a permission against a subject: `*` any one token, `>` one or more to the end. +func subjectMatches(permission, subject string) bool { + p, s := strings.Split(permission, "."), strings.Split(subject, ".") + for i, tok := range p { + if tok == ">" { + return len(s) > i + } + if i >= len(s) || (tok != "*" && tok != s[i]) { + return false + } + } + return len(p) == len(s) } func next(t *testing.T, out <-chan agent.Update) agent.Update { @@ -135,7 +190,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want) b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant) for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} { - b.ch(bucket) <- agent.Update{Caught: true} + b.send(bucket, agent.Update{Caught: true}) } got := map[string]bool{} var own delivery.Composed @@ -163,6 +218,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { if !strings.Contains(strings.Join(ownText, " "), `"id":"oldie.service"`) { t.Errorf("the node's own composition, composed before anything was applied, lacks oldie: %v", ownText) } + // Issue 1126, 2026-10-27 09:21: every watch and get the composer makes is one the node's grant allows; a watch of + // every have key (`>`) was refused by the bus and stopped every node agent. + b.mu.Lock() + for _, r := range b.reads { + if !grantAllows(t, "n1", r[0], r[1]) { + t.Errorf("the composer reads %s of %s, which the node's grant does not allow", r[1], r[0]) + } + } + b.mu.Unlock() lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp) nodeKey, _ := delivery.ComposedNodeKey("n1") if !got[lemursKey] || !got[nodeKey] || len(got) != 2 { @@ -424,6 +488,21 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 4 }) + // Every watch and get the shadow makes is one the node's grant allows (issue 1126, 2026-10-27 09:21). + sb.mu.Lock() + for _, w := range sb.watches { + if !grantAllows(t, "n1", w.bucket, w.pattern) { + t.Errorf("the shadow watches %s of %s, which the node's grant does not allow", w.pattern, w.bucket) + } + } + sb.mu.Unlock() + readOnly.mu.Lock() + for _, r := range readOnly.reads { + if !grantAllows(t, "n1", r[0], r[1]) { + t.Errorf("the shadow reads %s of %s, which the node's grant does not allow", r[1], r[0]) + } + } + readOnly.mu.Unlock() for group, want := range map[string]func(delivery.CandidateAnswer) bool{ "same": func(a delivery.CandidateAnswer) bool { return a.Verdict == delivery.CandidateSame && a.Head == "h-same" && a.Revision == 3 && len(a.Same) == 2 diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index e03cc51..f84cf88 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -95,7 +95,7 @@ func (s *Shadow) Run(ctx context.Context) error { {delivery.ModulesBucket, ">", (*Composer).takeModule, nil}, {ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, // What this node runs moved: an apply ends in its module's have. - {delivery.HaveBucket, node + ".*", nil, nil}, + {delivery.HaveBucket, node + ".>", nil, nil}, } type tagged struct { in int