diff --git a/internal/catalogue/declaration.go b/internal/catalogue/declaration.go index 73ca1288..a2d75976 100644 --- a/internal/catalogue/declaration.go +++ b/internal/catalogue/declaration.go @@ -1184,6 +1184,10 @@ func (r Resolution) compose(with Rendering, owner map[string]string, delete(copied, SecretsInEnvironment) delete(copied, NamesOnPurpose) delete(copied, TrustedField) + // A process's bus stays controller-side (novox/hq ADR 0436 §1): never composed onto the + // node-engine's process resource, whose wire declaration is mesh-host's own, vendored, with a + // strict decoder that refuses an unknown key. + delete(copied, ProcessBusField) // **An operator's value, from the assignment** (novox/hq ADR 0112, ADR 0155): what a // definition may not carry because it is true of one installation only. Filled from // the same layers a mergeable file takes, and refused when no layer set it. diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index 3d983484..bcbb4c69 100644 --- a/internal/catalogue/manifest.go +++ b/internal/catalogue/manifest.go @@ -1990,6 +1990,7 @@ func parseManifest(raw []byte, predefined PredefinedList) (Manifest, error) { // facts about the catalogue and are checked at registration (CatalogueProblems). problems = append(problems, declaredSeatProblems(m)...) problems = append(problems, RunsAsProblems(m)...) + problems = append(problems, ProcessBusProblems(m)...) problems = append(problems, ReadsHolderProblems(m)...) problems = append(problems, FilesHolderProblems(m)...) if m.Computed != "" && len(m.Resources) > 0 { @@ -2855,12 +2856,7 @@ func replacesProblems(m Manifest) []string { func invokeProblems(m Manifest) []string { var problems []string for _, t := range m.Invokes { - if t == "*" { - continue - } - t = strings.TrimPrefix(t, "seat:") - module, tool, named := strings.Cut(t, ".") - if !named || !name.MatchString(module) || !toolName.MatchString(tool) { + if !invokeNameValid(t) { problems = append(problems, fmt.Sprintf( "%s invokes %q, which does not name a tool: a module invokes ., or "+ "* for every tool on the mesh (novox/hq ADR 0152)", m.Module, t)) diff --git a/internal/catalogue/process_bus.go b/internal/catalogue/process_bus.go new file mode 100644 index 00000000..445fc68a --- /dev/null +++ b/internal/catalogue/process_bus.go @@ -0,0 +1,189 @@ +package catalogue + +import ( + "bytes" + "encoding/json" + "fmt" + "strings" +) + +// ProcessBusField is the key a process resource declares its bus on. Controller-side only: never composed +// onto the node-engine's process resource (novox/hq ADR 0436 §1) — the wire declaration is mesh-host's own, +// vendored, with a strict decoder that refuses an unknown key. +const ProcessBusField = "bus" + +// ProcessBus is what one process of a module does on the bus (novox/hq ADR 0436 §1, §2, §3, §5): its own bus +// account, granted exactly this and nothing of the module's other grants or the tool runner's — never a seat +// to hold or serve, never a tool call or seat verb answered, never a restricted stream heard, never a +// warrant published (§3, §5). +type ProcessBus struct { + // Emits are the events this process emits, named locally, as the module's own Emits names them + // (EmitsAll): a process emits only what its module may emit as a whole. + Emits []string `json:"emits,omitempty"` + // Queues are the module's own work-queue streams (ADR 0436 §6) this process consumes. + Queues []string `json:"queues,omitempty"` + // State is the module's own state, by local name, this process reads and writes. + State []string `json:"state,omitempty"` + // Asks says whether this process reads the router's records of this module's own asks. + Asks bool `json:"asks,omitempty"` + // Invokes are the tools and seat verbs this process calls, each . or a role's + // seat:. (as Manifest.Invokes), or * for every tool on the mesh. + Invokes []string `json:"invokes,omitempty"` +} + +// UnmarshalJSON reads a process's bus strictly: an unknown key — serves, a seat to hold or claim, or +// anything else that would answer a tool call or a seat's verb — is refused here, by name (novox/hq ADR +// 0436 §3). +func (b *ProcessBus) UnmarshalJSON(raw []byte) error { + type plain ProcessBus + var p plain + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.DisallowUnknownFields() + if err := dec.Decode(&p); err != nil { + return fmt.Errorf("a process's bus is {emits, queues, state, asks, invokes}: %w", typedUnknown(err)) + } + *b = ProcessBus(p) + return nil +} + +// processBusOf is the bus a process resource declares, and whether it declared one at all. +func processBusOf(r map[string]any) (bus ProcessBus, present bool, err error) { + raw, present := r["bus"] + if !present { + return ProcessBus{}, false, nil + } + body, err := json.Marshal(raw) + if err != nil { + return ProcessBus{}, true, err + } + if err := json.Unmarshal(body, &bus); err != nil { + return ProcessBus{}, true, err + } + return bus, true, nil +} + +// ProcessBuses is every process of this module that declares bus, by its id (the manifest's own resource +// id, novox/hq ADR 0436, settled by the lead) — what the grant composition reads to grant a process +// account exactly what it emits, consumes and calls (novox/hq ADR 0436 §2). +func (m Manifest) ProcessBuses() map[string]ProcessBus { + var out map[string]ProcessBus + for _, r := range m.Resources { + if fmt.Sprint(r["type"]) != "process" { + continue + } + b, present, err := processBusOf(r) + if !present || err != nil { + continue + } + if out == nil { + out = map[string]ProcessBus{} + } + out[fmt.Sprint(r["id"])] = b + } + return out +} + +// ProcessBusProblems is what is wrong with a manifest's processes' bus declarations (novox/hq ADR 0436 §1, +// §3): the account rule is modelled on RunsAsProblems (seats_declared.go:522-550), and an invoked tool or +// seat verb's shape on invokeProblems. +func ProcessBusProblems(m Manifest) []string { + var problems []string + streams := map[string]bool{} + for _, s := range m.Streams { + streams[s.Name] = true + } + state := map[string]bool{} + for _, s := range m.State { + state[s.Name] = true + } + emits := map[string]bool{} + for _, e := range m.EmitsAll() { + emits[e] = true + } + for _, r := range m.Resources { + if fmt.Sprint(r["type"]) != "process" { + continue + } + id := fmt.Sprint(r["id"]) + if _, present := r["bus"]; !present { + continue + } + b, _, err := processBusOf(r) + if err != nil { + problems = append(problems, fmt.Sprintf("%s's process %s declares bus: %v", m.Module, id, err)) + continue + } + user, _ := r["user"].(string) + problems = append(problems, processAccountProblems(m, id, user)...) + for _, e := range b.Emits { + if !emits[e] { + problems = append(problems, fmt.Sprintf( + "%s's process %s's bus emits %q, which %s does not declare as its own (novox/hq ADR 0436 §1)", + m.Module, id, e, m.Module)) + } + } + for _, q := range b.Queues { + if !streams[q] { + problems = append(problems, fmt.Sprintf( + "%s's process %s's bus queues %q, which %s does not declare as its own stream (novox/hq "+ + "ADR 0436 §1)", m.Module, id, q, m.Module)) + } + } + for _, s := range b.State { + if !state[s] { + problems = append(problems, fmt.Sprintf( + "%s's process %s's bus reads and writes state %q, which %s does not declare as its own "+ + "(novox/hq ADR 0436 §1)", m.Module, id, s, m.Module)) + } + } + for _, t := range b.Invokes { + if !invokeNameValid(t) { + problems = append(problems, fmt.Sprintf( + "%s's process %s's bus invokes %q, which does not name a tool: a process invokes "+ + "., or a role's seat:. (novox/hq ADR 0436 §1)", m.Module, id, t)) + } + } + } + return problems +} + +// processAccountProblems is what is wrong with one process's bus account (novox/hq ADR 0436 §1, §2): an +// account the module makes — a `user` resource of that name — never root and never the operator account, +// which a manifest cannot say safely (a node with no account has nothing to resolve it to, novox/hq ADR +// 0436, settled by the lead). +func processAccountProblems(m Manifest, processID, user string) []string { + switch { + case user == "": + return []string{fmt.Sprintf( + "%s's process %s declares bus and names no account; a process account is a user resource the "+ + "module makes, never root and never the operator account (novox/hq ADR 0436 §1, §2)", + m.Module, processID)} + case user == "root": + return []string{fmt.Sprintf( + "%s's process %s runs as root; a process account is the module's own, never root (novox/hq ADR "+ + "0436 §1, §2)", m.Module, processID)} + case user == "${machine:account}": + return []string{fmt.Sprintf( + "%s's process %s runs as the operator account; a process account is the module's own, never the "+ + "operator's (novox/hq ADR 0436 §1, §2)", m.Module, processID)} + } + for _, r := range m.Resources { + if fmt.Sprint(r["type"]) == "user" && fmt.Sprint(r["name"]) == user { + return nil + } + } + return []string{fmt.Sprintf( + "%s's process %s runs as %s, which it does not make: a user resource named %s (novox/hq ADR 0436 §1, "+ + "§2)", m.Module, processID, user, user)} +} + +// invokeNameValid says a module or process's invoked entry names a tool: ., a role's +// seat:., or * for every tool on the mesh (novox/hq ADR 0152, ADR 0436 §1). +func invokeNameValid(t string) bool { + if t == "*" { + return true + } + t = strings.TrimPrefix(t, "seat:") + first, second, named := strings.Cut(t, ".") + return named && name.MatchString(first) && toolName.MatchString(second) +} diff --git a/internal/catalogue/process_bus_test.go b/internal/catalogue/process_bus_test.go new file mode 100644 index 00000000..22ac13dc --- /dev/null +++ b/internal/catalogue/process_bus_test.go @@ -0,0 +1,188 @@ +package catalogue + +import ( + "strings" + "testing" +) + +// A process resource may declare `bus` (novox/hq ADR 0436 §1): what it emits, the module's own work-queue +// streams it consumes, the module's own state it reads and writes, whether it reads the router's records of +// this module's own asks, and the tools and seat verbs it calls. The fixture is mesh-mail's future manifest +// shape (ADR 0436 §7, §8), copied from mesh-catalog/modules/mesh-mail/module.json and extended as settled by +// the lead: process id "send" is the manifest's own `id` (internal/catalogue/runtime.go:72-74 is the model for +// a process id); the account file uses the controller's real JSON tags (modules.go:1041-1056 issueWith). +func mailSenderProcessManifest(t *testing.T, bus string) string { + t.Helper() + process := `{"id":"send","type":"process","run":["./mesh-mail","send"],"user":"mesh-mail-send"` + if bus != "" { + process += `,"bus":` + bus + } + process += `}` + return `{"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":64,"retention":"work-queue","max-deliveries":5,"duplicates":24}],` + + `"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` + process + `]}` +} + +const mailSenderBus = `{"emits":["mail.send.sent","mail.send.failed"],"queues":["mail-sends"],` + + `"state":["sends"],"asks":true,"invokes":["seat:operator-channel.tell"]}` + +// 1179/AC1: a well-formed bus on a process whose account the module makes is accepted, and the catalogue +// gives it whole to the grant composition (ProcessBuses, which 1180 reads). +func TestAProcessDeclaresItsBusAndIsAccepted(t *testing.T) { + m, err := ParseManifest([]byte(mailSenderProcessManifest(t, mailSenderBus))) + if err != nil { + t.Fatalf("a well-formed process bus was refused: %v", err) + } + buses := m.ProcessBuses() + b, ok := buses["send"] + if !ok { + t.Fatalf("the process's bus was not given to the grant composition: %+v", buses) + } + if !strSliceEqual(b.Emits, []string{"mail.send.sent", "mail.send.failed"}) || + !strSliceEqual(b.Queues, []string{"mail-sends"}) || !strSliceEqual(b.State, []string{"sends"}) || + !b.Asks || !strSliceEqual(b.Invokes, []string{"seat:operator-channel.tell"}) { + t.Fatalf("the bus read as %+v", b) + } +} + +func strSliceEqual(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// 1179/AC2: a bus on a process with no user, user root, or the operator account is refused naming the +// process and the account rule. +func TestAProcessBusAccountMustBeTheModulesOwn(t *testing.T) { + for name, c := range map[string]struct{ user, says string }{ + "no user": {"", "send"}, + "user root": {"root", "root"}, + "the operator account": {"${machine:account}", "operator"}, + } { + raw := `{"module":"mesh-mail","version":"1","emits":["mail.send.sent"],"state":["sends"],` + + `"streams":[{"name":"mail-sends","events":["mail.send.approved"],"retention":"work-queue",` + + `"max-deliveries":5}],"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` + + `{"id":"send","type":"process","run":["./mesh-mail","send"]` + if c.user != "" { + raw += `,"user":"` + c.user + `"` + } + raw += `,"bus":{"emits":["mail.send.sent"]}}]}` + _, err := ParseManifest([]byte(raw)) + if err == nil || !strings.Contains(err.Error(), "send") || !strings.Contains(err.Error(), c.says) { + t.Errorf("%s: %v; want a refusal naming the process and %q", name, err, c.says) + } + } +} + +// 1179/AC3: a bus naming a stream, state or event the module does not declare as its own is refused naming +// it. +func TestAProcessBusNamesOnlyTheModulesOwn(t *testing.T) { + for name, c := range map[string]struct{ bus, says string }{ + "another module's event": {`{"emits":["billing.paid"]}`, "billing.paid"}, + "another module's queue": {`{"queues":["other-queue"]}`, "other-queue"}, + "another module's state": {`{"state":["other-state"]}`, "other-state"}, + } { + _, err := ParseManifest([]byte(mailSenderProcessManifest(t, c.bus))) + if err == nil || !strings.Contains(err.Error(), c.says) { + t.Errorf("%s: %v; want a refusal naming %q", name, err, c.says) + } + } +} + +// 1179/AC4: a bus declaring serves, holds, claims or a seat is refused naming the key; a bus with invokes of +// a tool or a seat verb is accepted. +func TestAProcessBusNeverServesOrHoldsASeat(t *testing.T) { + for name, c := range map[string]struct{ bus, says string }{ + "serves": {`{"emits":["mail.send.sent"],"serves":["send"]}`, "serves"}, + "holds": {`{"emits":["mail.send.sent"],"holds":["mail-sender"]}`, "holds"}, + "claims": {`{"emits":["mail.send.sent"],"claims":["mail-sender"]}`, "claims"}, + "seat": {`{"emits":["mail.send.sent"],"seat":"mail-sender"}`, "seat"}, + } { + _, err := ParseManifest([]byte(mailSenderProcessManifest(t, c.bus))) + if err == nil || !strings.Contains(err.Error(), c.says) { + t.Errorf("%s: %v; want a refusal naming %q", name, err, c.says) + } + } + accepted := `{"emits":["mail.send.sent","mail.send.failed"],"invokes":["seat:operator-channel.tell","mail.status"]}` + if _, err := ParseManifest([]byte(mailSenderProcessManifest(t, accepted))); err != nil { + t.Fatalf("a bus invoking a tool and a seat verb was refused: %v", err) + } +} + +// 1179/AC5: a process declaring bus is composed with no bus key on the wire, and is otherwise byte for byte +// what main composes for the same process without bus. +func TestAProcessesBusNeverReachesTheWire(t *testing.T) { + withBus, err := ParseManifest([]byte(mailSenderProcessManifest(t, mailSenderBus))) + if err != nil { + t.Fatal(err) + } + withoutBus, err := ParseManifest([]byte(mailSenderProcessManifest(t, ""))) + if err != nil { + t.Fatal(err) + } + with := Rendering{Needed: map[string]map[string]string{}} + outWith, err := Resolution{Node: "novox", Modules: []Manifest{withBus}}.Declaration(with) + if err != nil { + t.Fatalf("composing with bus: %v", err) + } + outWithout, err := Resolution{Node: "novox", Modules: []Manifest{withoutBus}}.Declaration(with) + if err != nil { + t.Fatalf("composing without bus: %v", err) + } + var withProcess, withoutProcess map[string]any + for _, r := range outWith { + if r["id"] == "mesh-mail.send" { + withProcess = r + } + } + for _, r := range outWithout { + if r["id"] == "mesh-mail.send" { + withoutProcess = r + } + } + if withProcess == nil || withoutProcess == nil { + t.Fatalf("the process was not composed at all: with=%v without=%v", withProcess, withoutProcess) + } + if _, has := withProcess["bus"]; has { + t.Fatalf("the process resource carried a bus key onto the wire: %v", withProcess) + } + delete(withProcess, "bus") + if !mapsEqual(withProcess, withoutProcess) { + t.Fatalf("with bus declared, the process composed differently: %+v\nwant: %+v", withProcess, withoutProcess) + } +} + +func mapsEqual(a, b map[string]any) bool { + if len(a) != len(b) { + return false + } + for k, v := range a { + bv, ok := b[k] + if !ok { + return false + } + if av, ok := v.([]any); ok { + bbv, ok := bv.([]any) + if !ok || len(av) != len(bbv) { + return false + } + for i := range av { + if av[i] != bbv[i] { + return false + } + } + continue + } + if v != bv { + return false + } + } + return true +}