EE-D3: the real bus holds the tool runner's hearing of every event

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

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")
+}

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.