From f913d85e66cf161cda29c662e5cbb4589030824e Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:37:08 +0100 Subject: [PATCH] EE-D3: the real bus holds the tool runner's hearing of every event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The grants part's real-bus test gains a case: the tool runner's composed account hears a module's event, a module stream's and a restricted stream's whose grant names no tool runner, still after that grant changes and the accounts are composed again; it publishes none, hears no accept or proof, and gets no consumer or acknowledgement on the restricted stream. Builds: design 34 every-event §9, §27 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-grants/composed_integration_test.go | 220 ++++++++++++++++++ .../cmd/mesh-grants/integration_test.go | 1 + 2 files changed, 221 insertions(+) diff --git a/modules/mesh-grants/cmd/mesh-grants/composed_integration_test.go b/modules/mesh-grants/cmd/mesh-grants/composed_integration_test.go index 188a48e3..fb8fe2e7 100644 --- a/modules/mesh-grants/cmd/mesh-grants/composed_integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/composed_integration_test.go @@ -11,10 +11,12 @@ import ( "os" "path/filepath" "slices" + "strings" "testing" "time" "git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate" + sdkbc "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/seal" @@ -403,3 +405,221 @@ func manifestIn(t *testing.T, module string) catalogue.Manifest { } return m } + +// composedBus is a server loading the accounts composed from records, its conf file rewritten and reloaded by recompose. +type composedBus struct { + t *testing.T + s *server.Server + conf string +} + +func startComposedBus(t *testing.T, records buscompose.Records) *composedBus { + dir := t.TempDir() + b := &composedBus{t: t, conf: filepath.Join(dir, "server.conf")} + b.write(records, dir) + opts, err := server.ProcessConfigFile(b.conf) + if err != nil { + t.Fatalf("the composed list is not the server's: %v", err) + } + opts.NoLog, opts.NoSigs = true, true + if b.s, err = server.NewServer(opts); err != nil { + t.Fatal(err) + } + go b.s.Start() + if !b.s.ReadyForConnections(30 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(b.s.Shutdown) + return b +} + +func (b *composedBus) write(records buscompose.Records, dir string) { + users, err := buscompose.Users(records) + if err != nil { + b.t.Fatal(err) + } + for i := range users { + users[i].PasswordHash = "pw-" + users[i].Username() + } + accounts, err := buscompose.ComposeAccounts(users) + if err != nil { + b.t.Fatal(err) + } + if err := os.WriteFile(b.conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: \""+dir+"\" }\n"+accounts), 0o600); err != nil { + b.t.Fatal(err) + } +} + +// recompose writes the accounts composed from records again and has the server load them. +func (b *composedBus) recompose(records buscompose.Records) { + b.write(records, filepath.Dir(b.conf)) + if err := b.s.Reload(); err != nil { + b.t.Fatalf("the server does not load the composed list again: %v", err) + } +} + +// dial connects as a composed user; the server's refusals of its subscriptions and publishes arrive on refusals. +func (b *composedBus) dial(user string) (*nats.Conn, chan string) { + refusals := make(chan string, 16) + nc, err := nats.Connect(b.s.ClientURL(), nats.UserInfo(user, "pw-"+user), nats.CustomInboxPrefix("_INBOX."+user), + nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err.Error() })) + if err != nil { + b.t.Fatalf("%s: %v", user, err) + } + b.t.Cleanup(nc.Close) + return nc, refusals +} + +// refusedWithin waits, within a bound, for the server's refusal naming subject. +func refusedWithin(refusals chan string, subject string) bool { + bound := time.NewTimer(5 * time.Second) + defer bound.Stop() + for { + select { + case e := <-refusals: + if strings.Contains(e, "Permissions Violation") && strings.Contains(e, subject) { + return true + } + case <-bound.C: + return false + } + } +} + +// theToolRunnerHearsEveryEventOnTheBus, run by TestTheGrantsPartWhole, defends design 34 every-event §9 and §27 on a +// real server (ADR 0670 §1, §3, ADR 0671 §1, §2): the tool runner's account, as composed, hears a module's event, a +// module stream's and a restricted stream's whose grant value names no tool runner, still after that value changes and +// the accounts are composed again; it publishes none of them, hears no accept or proof, and gets no durable consumer +// and no acknowledgement on the restricted stream; another module hears no event it does not consume, and a module +// consuming the restricted stream with no entry in its grant value gets no consumer and no read. +func theToolRunnerHearsEveryEventOnTheBus(t *testing.T) { + const ledger = "mesh.stream.shop.event.booked" + records := func(auditorEntry []string) buscompose.Records { + return buscompose.Records{Nodes: []string{"a"}, Assigned: map[string][]buscompose.Declared{"a": { + {Module: buscompose.GrantsModule}, + {Module: "node-tools", ToolRunner: true}, + {Module: "mesh-delivery-view", Emits: []string{"delivered"}}, + {Module: "plex"}, + {Module: "shop", Streams: []buscompose.ModuleStream{ + {Module: "shop", Name: "SHOP", Events: []string{"opened"}, MaxAgeDays: 1, MaxMiB: 1}, + {Module: "shop", Name: "LEDGER", Events: []string{"booked"}, MaxAgeDays: 1, MaxMiB: 1, Restricted: true}}}, + {Module: "auditor", RunsAs: "auditor", Consumes: []string{"shop.booked"}, + Hears: []buscompose.Heard{{Stream: "LEDGER", Emitter: "shop", Filters: auditorEntry, Restricted: true}}}, + }}} + } + bus := startComposedBus(t, records(nil)) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + refusedIn := func() context.Context { + c, stop := context.WithTimeout(ctx, 2*time.Second) + t.Cleanup(stop) + return c + } + grantsConn, _ := bus.dial("a." + buscompose.GrantsModule) + grants, err := jetstream.New(grantsConn) + if err != nil { + t.Fatal(err) + } + for _, st := range []jetstream.StreamConfig{{Name: "SHOP", Subjects: []string{"mesh.stream.shop.event.opened"}}, + {Name: "LEDGER", Subjects: []string{ledger}}} { + if _, err := grants.CreateStream(ctx, st); err != nil { + t.Fatalf("the grants part creates the declared stream %s: %v", st.Name, err) + } + } + runner, runnerRefused := bus.dial("a.node-tools") + heard := make(chan *nats.Msg, 16) + for _, s := range sdkbc.EveryEvent() { + if _, err := runner.ChanSubscribe(s, heard); err != nil { + t.Fatal(err) + } + } + if err := runner.Flush(); err != nil { + t.Fatal(err) + } + hears := func(subject, from string) { + t.Helper() + select { + case m := <-heard: + if m.Subject != subject { + t.Errorf("the tool runner heard %s, want %s published by %s", m.Subject, subject, from) + } + case <-time.After(5 * time.Second): + t.Errorf("the tool runner heard nothing, want %s published by %s", subject, from) + } + } + view, _ := bus.dial("a.mesh-delivery-view") + shop, _ := bus.dial("a.shop") + for _, e := range []struct { + conn *nats.Conn + subject, from string + }{ + {view, "mesh.mod.mesh-delivery-view.event.delivered", "the view"}, + {shop, "mesh.stream.shop.event.opened", "a module's stream"}, + {shop, ledger, "a restricted stream whose grant names no tool runner"}} { + if err := e.conn.Publish(e.subject, []byte(`{}`)); err != nil { + t.Fatal(err) + } + if err := e.conn.Flush(); err != nil { + t.Fatal(err) + } + hears(e.subject, e.from) + } + // It publishes no event, and hears no accept nor proof (they are never events). + if err := runner.Publish("mesh.mod.mesh-delivery-view.event.delivered", []byte(`{}`)); err != nil { + t.Fatal(err) + } + if !refusedWithin(runnerRefused, "mesh.mod.mesh-delivery-view.event.delivered") { + t.Error("the tool runner published the view's event") + } + for _, s := range []string{"mesh.seat.channel.accept.show.k", "mesh.seat.intake.proof.code.k"} { + if _, err := runner.SubscribeSync(s); err != nil { + t.Fatal(err) + } + if !refusedWithin(runnerRefused, s) { + t.Errorf("the tool runner subscribed %s", s) + } + } + // No durable consumer, no acknowledgement on the restricted stream. + runnerJS, err := jetstream.New(runner) + if err != nil { + t.Fatal(err) + } + if _, err := runnerJS.CreateConsumer(refusedIn(), "LEDGER", jetstream.ConsumerConfig{Durable: "node-tools"}); err == nil { + t.Error("the tool runner made a durable consumer on the restricted stream") + } + if err := runner.Publish("$JS.ACK.LEDGER.node-tools.1.1.1.1.0", nil); err != nil { + t.Fatal(err) + } + if !refusedWithin(runnerRefused, "$JS.ACK.LEDGER") { + t.Error("the tool runner acknowledged on the restricted stream") + } + // Another module hears no event it does not consume. + plex, plexRefused := bus.dial("a.plex") + if _, err := plex.SubscribeSync("mesh.mod.mesh-delivery-view.event.delivered"); err != nil { + t.Fatal(err) + } + if !refusedWithin(plexRefused, "mesh.mod.mesh-delivery-view.event.delivered") { + t.Error("plex subscribed an event it does not consume") + } + // A consumer of the restricted stream with no entry in its grant value gets no consumer and no read. + auditor, _ := bus.dial("a.auditor") + auditorJS, err := jetstream.New(auditor) + if err != nil { + t.Fatal(err) + } + if _, err := auditorJS.CreateConsumer(refusedIn(), "LEDGER", jetstream.ConsumerConfig{Durable: "auditor"}); err == nil { + t.Error("a consumer with no entry made a durable consumer on the restricted stream") + } + if _, err := auditorJS.Stream(refusedIn(), "LEDGER"); err == nil { + t.Error("a consumer with no entry reads the restricted stream") + } + // The grant value changes and the accounts are composed again: the tool runner still hears the restricted stream. + bus.recompose(records([]string{ledger})) + if err := shop.Publish(ledger, []byte(`{}`)); err != nil { + t.Fatal(err) + } + if err := shop.Flush(); err != nil { + t.Fatal(err) + } + hears(ledger, "the restricted stream after its grant value changed") +} diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index 2d4e9f82..4b5fc605 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -627,6 +627,7 @@ func TestTheGrantsPartWhole(t *testing.T) { // And what the part composes holds on a server that loads it (one integration test per module, the test budget). t.Run("the composed grants hold on a server that loads them", theBusNodeFollowsTheSealedListUnderTheComposedGrants) + t.Run("the tool runner hears every event on a server that loads its grants", theToolRunnerHearsEveryEventOnTheBus) } // lockedLog is the part's log, read by the test while the part writes it.