diff --git a/cmd/mesh-controller/modules.go b/cmd/mesh-controller/modules.go index 49248710..030f2af0 100644 --- a/cmd/mesh-controller/modules.go +++ b/cmd/mesh-controller/modules.go @@ -926,14 +926,19 @@ func brokerReachableAt(ctx context.Context, inv *inventory.Inventory, known brok // A module that runs its machine's provisioner client and declares no broker is issued the client's account // alone (issue 917): the client speaks on the bus only as itself. func mayIssue(m catalogue.Manifest) error { - if _, reads := m.OwnSecrets["broker"]; !reads && !runsTheProvisionerClient(m) { + processes := m.ProcessBuses() + // A module whose own tools run in the machine's tool runner — mesh-mail's real shape — declares no + // `broker` of its own at all (novox/hq ADR 0436 §2): its process's account is a different account, + // delivered as its own secret below, so a process with a bus is reason enough to pass this check on + // its own, before the module's own `broker` is asked for. + if _, reads := m.OwnSecrets["broker"]; !reads && !runsTheProvisionerClient(m) && len(processes) == 0 { return fmt.Errorf("%s declares no own secret named broker, so there is nowhere to deliver "+ "an account; a module that speaks on the bus declares \"own-secrets\": {\"broker\": }", m.Module) } // And every process with a bus declaration (novox/hq ADR 0436 §2): its own account is delivered as the // module's own secret broker-, never the module's `broker`, so a process with a bus and no // such secret has nowhere to read an account either. - for id := range m.ProcessBuses() { + for id := range processes { name := processSecretName(id) if _, reads := m.OwnSecrets[name]; !reads { return fmt.Errorf("%s's process %s declares bus, and declares no own secret named %s, so there "+ @@ -948,6 +953,18 @@ func mayIssue(m catalogue.Manifest) error { // broker-, never the module's own `broker`. func processSecretName(processID string) string { return "broker-" + processID } +// processPrincipalFor is the one process of m's bus declarations named processID, as `rollout mint` finds it +// again from a KindProcess principal's bare id (novox/hq ADR 0436 §1, §2): false when the manifest no longer +// declares it. +func processPrincipalFor(m catalogue.Manifest, processID string) (broker.ProcessPrincipal, bool) { + for _, pr := range inventory.ProcessesOf(m) { + if pr.Process == processID { + return pr, true + } + } + return broker.ProcessPrincipal{}, false +} + // issueOnTheNewBus gives an assigned module its credential on the bus being built. // // **Three things differ from a management call, and each is the point of the move.** The credential diff --git a/cmd/mesh-controller/process_issue_integration_test.go b/cmd/mesh-controller/process_issue_integration_test.go new file mode 100644 index 00000000..bd17f106 --- /dev/null +++ b/cmd/mesh-controller/process_issue_integration_test.go @@ -0,0 +1,104 @@ +//go:build integration + +package main + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/broker" +) + +// aTestBrokerCertificate puts a real self-signed certificate on disk, so broker.FromEnvironment can read a +// fingerprint from it without a live bus — the same shape internal/broker's own writeCertificate makes. +func aTestBrokerCertificate(t *testing.T) string { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + template := x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "mesh-broker"}, + NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour)} + der, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "tls.crt") + if err := os.WriteFile(path, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o644); err != nil { + t.Fatal(err) + } + return path +} + +// review round 1, findings 3 and 4 (1181/AC1, AC2): module issue mints and delivers a process's own +// account against the store — no bus to dial, busAddress "" — for mesh-mail's real manifest shape (review +// finding 1: no own secret `broker` at all, its tools served by the tool runner), and issuing it again is +// the path every bus account takes (`rollout mint`'s new KindProcess case, and a re-run of `module issue`): +// it rotates the password, and the old hash no longer matches. +func TestModuleIssueMintsAndDeliversAProcesssAccount(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + theVaultIsInstalled(t, open) + + m := mailSenderWithProcess() + m.Moved = true // so handIssuedToVault's accountIsCovered is true, and the hand-off is exercised whole + register(t, open, m) + + vault := newStandInVault(t) + vaultAskerForIssue = func(context.Context) vaultAsker { return vault.ask } + t.Cleanup(func() { vaultAskerForIssue = func(context.Context) vaultAsker { return newStandInVault(nil).ask } }) + + t.Setenv(broker.CertificateVar, aTestBrokerCertificate(t)) + t.Setenv(broker.AddressVar, "nats://novox.example:4222") + + if err := issueOnTheNewBus(ctx, inv, m, "laptop", ""); err != nil { + t.Fatal(err) + } + user := broker.Principal{Kind: broker.KindProcess, Node: "laptop", Module: "mesh-mail", Process: "send"}.Username() + if user != "laptop.mesh-mail.send" { + t.Fatalf("the process's username: %q", user) + } + hash1, held, err := inv.BusUserHash(ctx, user) + if err != nil || !held || hash1 == "" { + t.Fatalf("the process's account was not minted: held %v, %v", held, err) + } + + // 1181/AC1: accepted with the provisioner client's JSON, and handed to the vault whole. + taken := vault.taken["accept"] + for _, want := range []string{`"url":"nats://novox.example:4222"`, `"node":"laptop"`, `"module":"mesh-mail"`, + `"user":"laptop.mesh-mail.send"`, `"password"`} { + if !strings.Contains(taken, want) { + t.Errorf("the vault was handed %q, without %s", taken, want) + } + } + if !strings.Contains(vault.said(), `"name":"broker-send"`) { + t.Errorf("the vault was not told the secret's name broker-send: %s", vault.said()) + } + // The file's own owner (mesh-mail-send, review finding 7-adjacent) is 1170's mechanism, proven generically + // there and for this fixture in TestAProcessesOwnSecretIsOwnedByItsOwnAccount. + + // 1181/AC2: issued again — the path `rollout mint`'s new KindProcess case and a re-run of `module issue` + // both take — rotates the password; the old hash no longer matches what is recorded. + if err := issueOnTheNewBus(ctx, inv, m, "laptop", ""); err != nil { + t.Fatal(err) + } + hash2, held, err := inv.BusUserHash(ctx, user) + if err != nil || !held { + t.Fatalf("the process's account is gone after a re-issue: held %v, %v", held, err) + } + if hash1 == hash2 { + t.Fatal("issuing the process's account again did not rotate its password") + } +} diff --git a/cmd/mesh-controller/process_issue_test.go b/cmd/mesh-controller/process_issue_test.go index fb37842d..bf4d02b3 100644 --- a/cmd/mesh-controller/process_issue_test.go +++ b/cmd/mesh-controller/process_issue_test.go @@ -9,13 +9,13 @@ import ( "github.com/novox/mesh-controller/internal/catalogue" ) -// mailSenderWithProcess is mesh-mail's future manifest shape (novox/hq ADR 0436 §7, §8): the process "send" -// declares bus, and its own secret broker-send is where module issue delivers its account (ADR 0436 §2) — +// mailSenderWithProcess is mesh-mail's real manifest shape (novox/hq ADR 0436 §7, §8, the lead's fixture): +// no own secret named `broker` at all — its tools run in the machine's tool runner — and the process "send" +// declares bus, whose own secret broker-send is where module issue delivers its account (ADR 0436 §2), // owned by mesh-mail-send, the user resource the process declares, at mode 0600 as every own secret is. func mailSenderWithProcess() catalogue.Manifest { m, err := catalogue.ParseManifest([]byte(`{"module":"mesh-mail","version":"1","emits":["mail.send.sent"], - "own-secrets":{"broker":{"path":"/var/lib/mesh-mail/broker"}, - "broker-send":{"path":"/var/lib/mesh-mail/broker-send","owner":"mesh-mail-send"}}, + "own-secrets":{"broker-send":{"path":"/var/lib/mesh-mail/broker-send","owner":"mesh-mail-send"}}, "resources":[{"id":"account","type":"user","name":"mesh-mail-send"}, {"id":"send","type":"process","run":["./mesh-mail","send"],"user":"mesh-mail-send", "bus":{"emits":["mail.send.sent"]}}]}`)) @@ -25,6 +25,14 @@ func mailSenderWithProcess() catalogue.Manifest { return m } +// review round 1, finding 1 (1181): a module declaring no `broker` of its own at all — mesh-mail's real +// shape, tools served by the tool runner — still passes mayIssue on its process's bus declaration alone. +func TestAModuleWithNoBrokerAndOnlyAProcessMayBeIssued(t *testing.T) { + if err := mayIssue(mailSenderWithProcess()); err != nil { + t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err) + } +} + // 1181/AC1: a process with a bus declaration and no own secret broker- cannot be issued — the // same refusal mayIssue already gives a module with no `broker` (novox/hq ADR 0436 §2). func TestAProcessWithNoBrokerSecretCannotBeIssued(t *testing.T) { @@ -78,7 +86,7 @@ func TestAProcessesOwnSecretIsOwnedByItsOwnAccount(t *testing.T) { m := mailSenderWithProcess() r := catalogue.Resolution{Node: "novox", Modules: []catalogue.Manifest{m}} out, err := r.Declaration(catalogue.Rendering{ - Needed: map[string]map[string]string{"mesh-mail": {"broker": "sealed-broker", "broker-send": "sealed-account"}}, + Needed: map[string]map[string]string{"mesh-mail": {"broker-send": "sealed-account"}}, }) if err != nil { t.Fatalf("composing the declaration: %v", err) diff --git a/cmd/mesh-controller/process_live_test.go b/cmd/mesh-controller/process_live_test.go new file mode 100644 index 00000000..dfff373a --- /dev/null +++ b/cmd/mesh-controller/process_live_test.go @@ -0,0 +1,280 @@ +//go:build integration + +package main + +import ( + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// mailSenderLiveManifest is mesh-mail's future manifest shape whole (novox/hq ADR 0436 §7, §8): the +// process "send" declares every clause of bus (emits, its one work-queue stream, its state, asks, and its +// invoked seat verb), so the account module issue delivers for it can be put through every grant at once. +func mailSenderLiveManifest(t *testing.T) catalogue.Manifest { + t.Helper() + m, err := catalogue.ParseManifest([]byte(`{"module":"mesh-mail","version":"1", + "emits":["mail.send.sent","mail.send.failed"],"state":["sends"], + "streams":[{"name":"mail-sends","events":["mail.send.approved"],"max-age-days":1,"max-mib":8, + "retention":"work-queue","max-deliveries":5}], + "own-secrets":{"broker-send":{"path":"/var/lib/mesh-mail/broker-send","owner":"mesh-mail-send"}}, + "resources":[{"id":"account","type":"user","name":"mesh-mail-send"}, + {"id":"send","type":"process","run":["./mesh-mail","send"],"user":"mesh-mail-send", + "bus":{"emits":["mail.send.sent","mail.send.failed"],"queues":["mail-sends"],"state":["sends"], + "asks":true,"invokes":["seat:operator-channel.tell"]}}]}`)) + if err != nil { + t.Fatal(err) + } + return m +} + +// operatorChannelWithTell is operator-channel as messenger declares it (broker.operatorChannel, novox/hq +// ADR 0259 §3), with the one verb this release adds: `tell` (novox/hq ADR 0436 §1's own fixture), so the +// process's invoked seat verb has a holder to answer it. +func operatorChannelWithTell() broker.Seat { + return broker.Seat{Name: "operator-channel", Scope: "mesh", Accepts: []string{"ask", "cancel"}, + Emits: []string{"decided"}, Serves: []string{"open", "history", "notify", "tell"}, + ByCaller: []string{"ask", "cancel", "decided"}, Records: []string{"messenger_asks"}} +} + +// review round 1, finding 5 (1181/AC3), the release's acceptance test, on a real bus and the store: the +// account module issue delivers for mesh-mail's process "send" — composed from the manifest the real way +// (inventory.ProcessesOf -> broker.Users -> broker.ComposeAccounts), minted with the password module issue +// mints (aMesh) — does exactly what its bus declares and nothing else. +func TestAProcessAccountOnARealBusAndStore(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + + m := mailSenderLiveManifest(t) + operatorChannel := operatorChannelWithTell() + records := broker.Records{Nodes: []string{"laptop"}, Assigned: map[string][]broker.Declared{"laptop": { + {Module: "messenger", Holds: []broker.Seat{operatorChannel}}, + {Module: "mesh-mail", NoAccount: true, Uses: []broker.Seat{operatorChannel}, + Processes: inventory.ProcessesOf(m)}, + }}} + users, err := broker.Users(records) + if err != nil { + t.Fatal(err) + } + + // The three principals this test drives, each minted the password module issue mints — the same + // inv.MintBusPassword call issueProcess makes for the process account. + var admin, messenger, sender broker.Principal + for _, p := range users { + switch { + case p.Kind == broker.KindController: + admin = p + case p.Module == "messenger": + messenger = p + case p.Kind == broker.KindProcess: + sender = p + } + } + if admin.Kind == "" || messenger.Kind == "" || sender.Kind == "" { + t.Fatalf("not every principal was composed: admin=%+v messenger=%+v sender=%+v", admin, messenger, sender) + } + if sender.Username() != "laptop.mesh-mail.send" { + t.Fatalf("the process's username: %q", sender.Username()) + } + + passwords := map[string]string{} + hashes := map[string]string{} + for _, p := range []struct { + principal broker.Principal + kind string + }{{admin, inventory.BusController}, {messenger, inventory.BusModule}, {sender, inventory.BusProcess}} { + pw, err := inv.MintBusPassword(ctx, inventory.BusUser{Username: p.principal.Username(), Kind: p.kind, + Node: p.principal.Node, Module: p.principal.Module}) + if err != nil { + t.Fatal(err) + } + passwords[p.principal.Username()] = pw + hash, held, err := inv.BusUserHash(ctx, p.principal.Username()) + if err != nil || !held { + t.Fatalf("the minted password of %s was not recorded: %v, %v", p.principal.Username(), held, err) + } + hashes[p.principal.Username()] = hash + } + filled, missing := broker.WithPasswords([]broker.Principal{admin, messenger, sender}, hashes) + if len(missing) > 0 { + t.Fatalf("no password for %v", missing) + } + + accounts, err := broker.ComposeAccounts(filled) + if err != nil { + t.Fatal(err) + } + conf := filepath.Join(t.TempDir(), "bus.conf") + if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+ + `"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil { + t.Fatal(err) + } + opts, err := server.ProcessConfigFile(conf) + if err != nil { + t.Fatalf("the composed accounts do not parse: %v", err) + } + opts.NoLog, opts.NoSigs = true, true + s, err := server.NewServer(opts) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + defer s.Shutdown() + + // Every connection names its own inbox prefix: the grant is `_INBOX..>`, and the client + // library's own default random inbox matches nothing granted, so replies are refused without it. + adminConn, err := broker.Dial(s.ClientURL(), nats.UserInfo(admin.Username(), passwords[admin.Username()]), + nats.CustomInboxPrefix("_INBOX."+admin.Username())) + if err != nil { + t.Fatal(err) + } + defer adminConn.Close() + mailSends := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"}, + MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1} + if err := adminConn.EnsureStream(mailSends.Stream()); err != nil { + t.Fatal(err) + } + if err := adminConn.EnsureConsumer(broker.QueueConsumer(sender, sender.Queues[0])); err != nil { + t.Fatal(err) + } + if err := adminConn.EnsureBucket(broker.Bucket{Module: "mesh-mail", Name: "sends"}); err != nil { + t.Fatal(err) + } + js, err := jetstream.New(adminConn.Conn()) + if err != nil { + t.Fatal(err) + } + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "messenger_asks"}); err != nil { + t.Fatal(err) + } + + // messenger, the router, answers exactly its granted verbs — never a wildcard the grant does not give it. + holder, err := nats.Connect(s.ClientURL(), nats.UserInfo(messenger.Username(), passwords[messenger.Username()]), + nats.CustomInboxPrefix("_INBOX."+messenger.Username())) + if err != nil { + t.Fatal(err) + } + defer holder.Close() + heard := make(chan string, 1) + for _, verb := range []string{"open", "history", "notify", "tell"} { + if _, err := holder.Subscribe("mesh.seat.operator-channel.tool."+verb, func(m *nats.Msg) { + heard <- m.Subject + _ = m.Respond([]byte(`{"result":{}}`)) + }); err != nil { + t.Fatal(err) + } + } + _ = holder.Flush() + + refusals := make(chan error, 8) + sender2, err := nats.Connect(s.ClientURL(), nats.UserInfo(sender.Username(), passwords[sender.Username()]), + nats.CustomInboxPrefix("_INBOX."+sender.Username()), + nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err })) + if err != nil { + t.Fatal(err) + } + defer sender2.Close() + + // 1. Its event. + if err := sender2.Publish("mesh.mod.mesh-mail.event.mail.send.sent", []byte("{}")); err != nil { + t.Fatal(err) + } + + // 2. Its state: writes and reads its own bucket. + senderKV, err := jetstream.New(sender2) + if err != nil { + t.Fatal(err) + } + bucket, err := senderKV.KeyValue(ctx, "mesh-mail_sends") + if err != nil { + t.Fatal(err) + } + if _, err := bucket.Put(ctx, "key", []byte("value")); err != nil { + t.Fatalf("the process may not write its own state: %v", err) + } + if got, err := bucket.Get(ctx, "key"); err != nil || string(got.Value()) != "value" { + t.Fatalf("the process may not read its own state back: %v, %v", got, err) + } + + // 3. Its queue: pulls and acknowledges MAIL_SENDS through its own consumer. + if _, err := js.Publish(ctx, broker.ModuleStreamSubject("mesh-mail", "mail.send.approved"), []byte("{}")); err != nil { + t.Fatal(err) + } + reading, err := jetstream.New(sender2) + if err != nil { + t.Fatal(err) + } + consumer, err := reading.Consumer(ctx, "MAIL_SENDS", broker.QueueConsumer(sender, sender.Queues[0]).Name) + if err != nil { + t.Fatalf("the process may not bind its own queue consumer: %v", err) + } + batch, err := consumer.Fetch(1, jetstream.FetchMaxWait(3*time.Second)) + if err != nil { + t.Fatal(err) + } + got := false + for msg := range batch.Messages() { + got = true + if err := msg.Ack(); err != nil { + t.Fatalf("the process may not acknowledge its own queue: %v", err) + } + } + if !got { + t.Fatal("the process did not pull the message its own queue holds") + } + + // 4. Its invoked seat verb, in its own name. + if _, err := sender2.Request("mesh.seat.operator-channel.tool.tell", []byte(`{}`), 3*time.Second); err != nil { + t.Fatalf("the process may not call operator-channel.tell: %v", err) + } + select { + case s := <-heard: + if s != "mesh.seat.operator-channel.tool.tell" { + t.Fatalf("messenger heard %s", s) + } + case <-time.After(time.Second): + t.Fatal("messenger never heard the call") + } + + // 5. Refused, one at a time, so each refusal is counted against the attempt that caused it rather than + // any other: answering a tool call it holds no seat and serves no tool for. + if _, err := sender2.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil { + t.Fatal(err) + } + _ = sender2.Flush() + select { + case err := <-refusals: + if !errors.Is(err, nats.ErrPermissionViolation) { + t.Errorf("subscribing a tool call was not refused as a permissions violation: %v", err) + } + case <-time.After(3 * time.Second): + t.Error("subscribing a tool call the process does not serve was not refused") + } + // 6. Refused: publishing a warrant — only the holder, messenger, may say operator-channel decided. + if err := sender2.Publish("mesh.seat.operator-channel.event.decided.mesh-mail", []byte("{}")); err != nil { + t.Fatal(err) + } + _ = sender2.Flush() + select { + case err := <-refusals: + if !errors.Is(err, nats.ErrPermissionViolation) { + t.Errorf("publishing a warrant was not refused as a permissions violation: %v", err) + } + case <-time.After(3 * time.Second): + t.Error("publishing a warrant the process does not hold the seat for was not refused") + } +} diff --git a/cmd/mesh-controller/rollout.go b/cmd/mesh-controller/rollout.go index d1fb3b47..fbee3e1e 100644 --- a/cmd/mesh-controller/rollout.go +++ b/cmd/mesh-controller/rollout.go @@ -392,6 +392,32 @@ func rolloutMint(ctx context.Context, again bool) error { } modules++ + case broker.KindProcess: + // One process's own account (novox/hq ADR 0436 §1, §2), delivered as the module's own secret + // broker- — minted here exactly as `module issue` mints it, so a process issued + // once still gets a new password every time the bus's accounts are issued again. + m, inShelf := shelf[p.Module] + if !inShelf { + skipped++ + continue + } + pr, found := processPrincipalFor(m, p.Process) + if !found { + skipped++ + continue + } + name := processSecretName(p.Process) + if _, reads := m.OwnSecrets[name]; !reads { + fmt.Printf(" %s's process %s declares bus but declares no `%s` secret to receive a "+ + "credential in; skipped\n", p.Module, p.Process, name) + skipped++ + continue + } + if err := issueProcess(ctx, inv, m, pr, p.Node, "", known, busAddress); err != nil { + return err + } + modules++ + default: skipped++ } diff --git a/internal/broker/process_live_test.go b/internal/broker/process_live_test.go deleted file mode 100644 index f7ed5044..00000000 --- a/internal/broker/process_live_test.go +++ /dev/null @@ -1,203 +0,0 @@ -//go:build integration - -package broker - -import ( - "errors" - "os" - "path/filepath" - "testing" - "time" - - "github.com/nats-io/nats-server/v2/server" - "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" - "golang.org/x/crypto/bcrypt" -) - -// 1181/AC3, the release's acceptance test, on a real bus: the account module issue delivers for mesh-mail's -// process "send" does exactly what its bus declares — pull and acknowledge its one work-queue consumer, -// publish its event, read and write its state, and call operator-channel's tell in its own name — and -// nothing it does not: no tool call or seat verb answered, no warrant published. Modelled on -// callers_live_test.go's TestAServerComposedFromTheGrantsRefusesACallNamingAnother. -// -// go test -tags integration ./internal/broker/ -run TestAProcessAccountOnARealBus -func TestAProcessAccountOnARealBus(t *testing.T) { - hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost) - if err != nil { - t.Fatal(err) - } - controller := Principal{Kind: KindController, PasswordHash: string(hash)} - messenger := Principal{Kind: KindModule, Node: "novox", Module: "messenger", PasswordHash: string(hash), - Holds: []Seat{operatorChannel()}} - process := Principal{Kind: KindProcess, Node: "novox", Module: "mesh-mail", Process: "send", - PasswordHash: string(hash), Emits: []string{"mail.send.sent", "mail.send.failed"}, - Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}}, - State: []string{"sends"}, Asks: true, Uses: []Seat{operatorChannel()}, - Invokes: []string{"seat:operator-channel.tell"}, - } - accounts, err := ComposeAccounts([]Principal{controller, messenger, process}) - if err != nil { - t.Fatal(err) - } - conf := filepath.Join(t.TempDir(), "bus.conf") - if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+ - `"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil { - t.Fatal(err) - } - opts, err := server.ProcessConfigFile(conf) - if err != nil { - t.Fatalf("the composed accounts do not parse: %v", err) - } - opts.NoLog, opts.NoSigs = true, true - s, err := server.NewServer(opts) - if err != nil { - t.Fatal(err) - } - go s.Start() - if !s.ReadyForConnections(10 * time.Second) { - t.Fatal("the bus did not come up") - } - defer s.Shutdown() - - // The controller sets up the stream, the one consumer, and the process's state bucket — exactly as - // `module issue` and the controller's raise would, before anything connects as the process. - admin, err := Dial(s.ClientURL(), nats.UserInfo("controller", "pw")) - if err != nil { - t.Fatal(err) - } - defer admin.Close() - mailSends := ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"}, - MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1} - if err := admin.EnsureStream(mailSends.Stream()); err != nil { - t.Fatal(err) - } - if err := admin.EnsureConsumer(QueueConsumer(process, process.Queues[0])); err != nil { - t.Fatal(err) - } - if err := admin.EnsureBucket(Bucket{Module: "mesh-mail", Name: "sends"}); err != nil { - t.Fatal(err) - } - - // The router's own records bucket operator-channel names (novox/hq ADR 0259 §3): present so the - // process's direct-get of its own module's asks is a real subject, not merely one it is permitted. - js, err := jetstream.New(admin.Conn()) - if err != nil { - t.Fatal(err) - } - if _, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: "messenger_asks"}); err != nil { - t.Fatal(err) - } - - // messenger, the router, answers operator-channel's tell in the caller's own name. - holder, err := nats.Connect(s.ClientURL(), nats.UserInfo("novox.messenger", "pw")) - if err != nil { - t.Fatal(err) - } - defer holder.Close() - heard := make(chan string, 1) - if _, err := holder.Subscribe("mesh.seat.operator-channel.tool.>", func(m *nats.Msg) { - heard <- m.Subject - _ = m.Respond([]byte(`{"result":{}}`)) - }); err != nil { - t.Fatal(err) - } - _ = holder.Flush() - - refusals := make(chan error, 8) - sender, err := nats.Connect(s.ClientURL(), nats.UserInfo("novox.mesh-mail.send", "pw"), - nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err })) - if err != nil { - t.Fatal(err) - } - defer sender.Close() - - // 1. Its event. - if err := sender.Publish("mesh.mod.mesh-mail.event.mail.send.sent", []byte("{}")); err != nil { - t.Fatal(err) - } - - // 2. Its state: writes and reads its own bucket. - kv, err := jetstream.New(sender) - if err != nil { - t.Fatal(err) - } - bucket, err := kv.KeyValue(t.Context(), "mesh-mail_sends") - if err != nil { - t.Fatal(err) - } - if _, err := bucket.Put(t.Context(), "key", []byte("value")); err != nil { - t.Fatalf("the process may not write its own state: %v", err) - } - if got, err := bucket.Get(t.Context(), "key"); err != nil || string(got.Value()) != "value" { - t.Fatalf("the process may not read its own state back: %v, %v", got, err) - } - - // 3. Its queue: pulls and acknowledges MAIL_SENDS through its own consumer. - js2, err := jetstream.New(admin.Conn()) - if err != nil { - t.Fatal(err) - } - if _, err := js2.Publish(t.Context(), ModuleStreamSubject("mesh-mail", "mail.send.approved"), []byte("{}")); err != nil { - t.Fatal(err) - } - reading, err := jetstream.New(sender) - if err != nil { - t.Fatal(err) - } - consumer, err := reading.Consumer(t.Context(), "MAIL_SENDS", QueueConsumer(process, process.Queues[0]).Name) - if err != nil { - t.Fatalf("the process may not bind its own queue consumer: %v", err) - } - batch, err := consumer.Fetch(1, jetstream.FetchMaxWait(3*time.Second)) - if err != nil { - t.Fatal(err) - } - got := false - for msg := range batch.Messages() { - got = true - if err := msg.Ack(); err != nil { - t.Fatalf("the process may not acknowledge its own queue: %v", err) - } - } - if !got { - t.Fatal("the process did not pull the message its own queue holds") - } - - // 4. Its invoked seat verb, in its own name. - if _, err := sender.Request("mesh.seat.operator-channel.tool.tell", []byte(`{}`), 3*time.Second); err != nil { - t.Fatalf("the process may not call operator-channel.tell: %v", err) - } - select { - case s := <-heard: - if s != "mesh.seat.operator-channel.tool.tell" { - t.Fatalf("messenger heard %s", s) - } - case <-time.After(time.Second): - t.Fatal("messenger never heard the call") - } - - // 5. Refused: answering a tool call (it holds no seat, serves no tool). - if _, err := sender.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil { - t.Fatal(err) - } - _ = sender.Flush() - // 6. Refused: publishing a warrant (only the holder, messenger, may say operator-channel decided). - if err := sender.Publish("mesh.seat.operator-channel.event.decided.mesh-mail", []byte("{}")); err != nil { - t.Fatal(err) - } - _ = sender.Flush() - seen := 0 - timeout := time.After(3 * time.Second) - for seen < 2 { - select { - case err := <-refusals: - if !errors.Is(err, nats.ErrPermissionViolation) { - t.Errorf("a refusal was not a permissions violation: %v", err) - } - seen++ - case <-timeout: - t.Fatalf("saw %d of 2 expected permission violations", seen) - } - } -}