From f6a48ac5ef08e7ffe34cea2e2a8362e71f9fd09e Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 02:55:29 +0200 Subject: [PATCH] Let module issue mint and deliver a bus process's account (issue 1181) --- cmd/mesh-controller/modules.go | 85 ++++++++ cmd/mesh-controller/process_issue_test.go | 98 +++++++++ internal/broker/derived.go | 2 +- internal/broker/modstreams.go | 4 +- .../modstreams_workqueue_integration_test.go | 2 +- internal/broker/process_grants.go | 2 +- internal/broker/process_live_test.go | 203 ++++++++++++++++++ internal/inventory/busrecords.go | 6 +- internal/inventory/bususers.go | 3 + 9 files changed, 397 insertions(+), 8 deletions(-) create mode 100644 cmd/mesh-controller/process_issue_test.go create mode 100644 internal/broker/process_live_test.go diff --git a/cmd/mesh-controller/modules.go b/cmd/mesh-controller/modules.go index 754bc044..49248710 100644 --- a/cmd/mesh-controller/modules.go +++ b/cmd/mesh-controller/modules.go @@ -930,9 +930,24 @@ func mayIssue(m catalogue.Manifest) error { 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() { + 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 "+ + "is nowhere to deliver its account; a process with a bus declares \"own-secrets\": "+ + "{%q: } (novox/hq ADR 0436 §2)", m.Module, id, name, name) + } + } return nil } +// processSecretName is the own secret a process's bus account is delivered in (novox/hq ADR 0436 §2): +// broker-, never the module's own `broker`. +func processSecretName(processID string) string { return "broker-" + processID } + // 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 @@ -968,12 +983,82 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue return err } } + // Every process with a bus declaration gets an account of its own (novox/hq ADR 0436 §1, §2), minted and + // delivered whether or not the module itself speaks on the bus — its own secret is broker-, + // never the module's `broker`. + for _, pr := range inventory.ProcessesOf(m) { + if err := issueProcess(ctx, inv, m, pr, node, busAddress, known, reachable); err != nil { + return err + } + } if !runsTheProvisionerClient(m) { return nil } return issueProvisionerClient(ctx, inv, m, node, known, reachable) } +// issueProcess gives one process its account: minted under its own name .., +// delivered as the module's own secret broker- in the provisioner client's form (novox/hq ADR +// 0436 §2), handed to the vault as any covered account is, and its work-queue consumer or consumers +// created where the bus can be reached — never the module's claims, since a process holds no seat. +func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal, + node, busAddress string, known broker.Broker, reachable string) error { + user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username() + password, err := inv.MintBusPassword(ctx, inventory.BusUser{ + Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module, + }) + if err != nil { + return err + } + name := processSecretName(pr.Process) + held, err := heldProcessAccount(known, reachable, node, m.Module, user, password) + if err != nil { + return err + } + if err := inv.AcceptSecretForModule(ctx, node, m.Module, name, string(held)); err != nil { + return err + } + if err := handIssuedToVault(ctx, inv, m, node, name, string(held)); err != nil { + return err + } + if len(pr.Queues) > 0 { + if busAddress == "" { + fmt.Printf(" %s's process %s consumes its work queue; its consumer is created when the bus is "+ + "reachable (`push --behind`, then `rollout mint` again is harmless)\n", m.Module, pr.Process) + } else { + js, err := broker.Dial(busAddress) + if err != nil { + return fmt.Errorf("the credential is minted and the mesh cannot reach the bus to create "+ + "%s's process %s's work-queue consumer: %w", m.Module, pr.Process, err) + } + defer js.Close() + who := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process} + for _, q := range pr.Queues { + if err := js.EnsureConsumer(broker.QueueConsumer(who, q)); err != nil { + return err + } + } + } + } + fmt.Printf("bus user %s minted for %s's process %s, scoped to its bus declaration\n", user, m.Module, pr.Process) + fmt.Printf(" sealed to %s as %s. It arrives with the next push — `push %s` to send it\n", node, name, node) + return nil +} + +// heldProcessAccount is a process's account, in the provisioner client's form (novox/hq ADR 0436 §2): +// {url, fingerprint, node, module, user, password} — never claims, since a process holds no seat (§3). +func heldProcessAccount(known broker.Broker, reachable, node, module, user, password string) ([]byte, error) { + return json.Marshal(struct { + URL string `json:"url"` + Fingerprint string `json:"fingerprint,omitempty"` + Node string `json:"node"` + Module string `json:"module"` + User string `json:"user"` + Password string `json:"password"` + }{URL: "nats://" + reachable, Fingerprint: known.Fingerprint, Node: node, Module: module, User: user, + Password: password}) +} + // runsTheProvisionerClient says a module runs its machine's provisioner client, which reaches the bus on an // account of its own (novox/hq ADR 0349 clause 11, design 57 P3): it claims the seat and declares the own secret // that account is delivered in. The same test the bus's user list makes (inventory's declaredFor). diff --git a/cmd/mesh-controller/process_issue_test.go b/cmd/mesh-controller/process_issue_test.go new file mode 100644 index 00000000..fb37842d --- /dev/null +++ b/cmd/mesh-controller/process_issue_test.go @@ -0,0 +1,98 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/broker" + "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) — +// 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"}}, + "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"]}}]}`)) + if err != nil { + panic(err) + } + return m +} + +// 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) { + m := mailSenderWithProcess() + delete(m.OwnSecrets, "broker-send") + err := mayIssue(m) + if err == nil { + t.Fatal("a process with no broker- own secret was issued an account") + } + for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("the refusal does not say %q: %v", want, err) + } + } + if err := mayIssue(mailSenderWithProcess()); err != nil { + t.Errorf("a process declaring its own broker- secret was refused: %v", err) + } +} + +// 1181/AC1: the account held for a process is the provisioner client's form — {url, fingerprint, node, +// module, user, password} — named for the process, never the module's own account. +func TestAProcesssHeldAccountIsTheProvisionersForm(t *testing.T) { + user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username() + if user != "novox.mesh-mail.send" { + t.Fatalf("the process's username: %q", user) + } + held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail", + user, "a-password") + if err != nil { + t.Fatal(err) + } + var got map[string]any + if err := json.Unmarshal(held, &got); err != nil { + t.Fatal(err) + } + want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox", + "module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"} + for k, v := range want { + if got[k] != v { + t.Errorf("%s: %v, want %v", k, got[k], v) + } + } + if len(got) != len(want) { + t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want) + } +} + +// 1181/AC1: composed, the process's own secret belongs to its own account at 0600 (novox/hq ADR 0436 §1, §2; +// the owner mechanism is issue 1170's, proven generically there — this is mesh-mail's own shape). +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"}}, + }) + if err != nil { + t.Fatalf("composing the declaration: %v", err) + } + var found bool + for _, res := range out { + if res["type"] == "file" && res["path"] == "/var/lib/mesh-mail/broker-send" { + found = true + if got := res["owner"]; got != "mesh-mail-send" { + t.Fatalf("broker-send's owner: %v; want mesh-mail-send", got) + } + } + } + if !found { + t.Fatal("broker-send was not composed at all") + } +} diff --git a/internal/broker/derived.go b/internal/broker/derived.go index 55503638..b1aa7848 100644 --- a/internal/broker/derived.go +++ b/internal/broker/derived.go @@ -216,7 +216,7 @@ func ConsumersOf(users []Principal) []ModuleConsumer { // And the one durable consumer of each of its own work-queue streams, for the one process its // bus.queues names (novox/hq ADR 0436 §1, §2, §6). for _, q := range p.Queues { - out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: queueConsumer(p, q)}) + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: QueueConsumer(p, q)}) } } for _, p := range users { diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index cdc01984..4a79df6d 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -116,10 +116,10 @@ type QueueOwner struct { MaxDeliver int } -// queueConsumer is the durable consumer the one process a module's bus.queues names is granted on its own +// QueueConsumer is the durable consumer the one process a module's bus.queues names is granted on its own // work-queue stream (novox/hq ADR 0436 §1, §2, §6): no filter, since it is the stream's only consumer — // unlike heardConsumer, which is one of possibly several hearing a module's ordinary stream. -func queueConsumer(p Principal, q QueueOwner) Consumer { +func QueueConsumer(p Principal, q QueueOwner) Consumer { return Consumer{ Name: p.Node + "_" + p.Module + "_" + q.Process, Stream: q.Stream, diff --git a/internal/broker/modstreams_workqueue_integration_test.go b/internal/broker/modstreams_workqueue_integration_test.go index fbab47a9..5e51b68f 100644 --- a/internal/broker/modstreams_workqueue_integration_test.go +++ b/internal/broker/modstreams_workqueue_integration_test.go @@ -34,7 +34,7 @@ func TestAWorkQueueStreamOnARealBus(t *testing.T) { owner := QueueOwner{Process: "send", Stream: stream.Name, MaxDeliver: 5} sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x"} - consumer := queueConsumer(sender, owner) + consumer := QueueConsumer(sender, owner) consumer.AckWaitSeconds = 1 if err := js.EnsureConsumer(consumer); err != nil { t.Fatal(err) diff --git a/internal/broker/process_grants.go b/internal/broker/process_grants.go index d0689e46..9bbead83 100644 --- a/internal/broker/process_grants.go +++ b/internal/broker/process_grants.go @@ -19,7 +19,7 @@ func processGrants(p Principal) (pub, sub []string, err error) { // 2. The one durable consumer of each work-queue stream its bus names (novox/hq ADR 0436 §1, §2, §6): // asked about, pulled from and acknowledged, by the same name ConsumersOf composes the consumer under - // (queueConsumer) — and no other stream's. + // (QueueConsumer) — and no other stream's. for _, q := range p.Queues { consumer := p.Node + "_" + p.Module + "_" + q.Process pub = append(pub, diff --git a/internal/broker/process_live_test.go b/internal/broker/process_live_test.go new file mode 100644 index 00000000..f7ed5044 --- /dev/null +++ b/internal/broker/process_live_test.go @@ -0,0 +1,203 @@ +//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) + } + } +} diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index a78d2d8a..09fd6215 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -192,7 +192,7 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio Queues: queueOwnersOf(m), // And every process of its own that declares bus, each granted a bus account of its own (novox/hq // ADR 0436 §1, §2). - Processes: processesOf(m), + Processes: ProcessesOf(m), // And the tools its health asks (novox/hq ADR 0240): the machine's node-engine is granted them. Checks: catalogue.HealthChecks(m), // And whether it runs as an account of its own (novox/hq ADR 0259 §8). @@ -465,11 +465,11 @@ func queueOwnersOf(m catalogue.Manifest) []broker.QueueOwner { return out } -// processesOf is every process of a module that declares bus, as the broker composes its own account from +// ProcessesOf is every process of a module that declares bus, as the broker composes its own account from // (novox/hq ADR 0436 §1, §2): its own emits, its own work-queue consumers (queueOwnersOf, filtered to each // process's own), its own state, whether it reads its module's own asks, and the tools and seat verbs it // calls. -func processesOf(m catalogue.Manifest) []broker.ProcessPrincipal { +func ProcessesOf(m catalogue.Manifest) []broker.ProcessPrincipal { buses := m.ProcessBuses() if len(buses) == 0 { return nil diff --git a/internal/inventory/bususers.go b/internal/inventory/bususers.go index 6a954909..388e6aaf 100644 --- a/internal/inventory/bususers.go +++ b/internal/inventory/bususers.go @@ -68,6 +68,9 @@ const ( // BusProvisioner is a machine's provisioner client (novox/hq ADR 0349 clause 11, design 57 P3): an account of // its own, named for the machine. BusProvisioner = "provisioner" + // BusProcess is one process of a module that declares bus (novox/hq ADR 0436 §1, §2): an account of its + // own, named for the machine, the module and the process. + BusProcess = "process" ) // MintBusPassword makes a bus password and records its hash under a username, replacing whatever was