diff --git a/modules/mesh-grants/internal/buscompose/buscompose_test.go b/modules/mesh-grants/internal/buscompose/buscompose_test.go index b0be24e3..caffd5e4 100644 --- a/modules/mesh-grants/internal/buscompose/buscompose_test.go +++ b/modules/mesh-grants/internal/buscompose/buscompose_test.go @@ -1070,7 +1070,9 @@ func theBusModulesProcessKeepsBusState(t *testing.T) { r := Records{Nodes: []string{"a"}, Assigned: map[string][]Declared{"a": { {Module: "node-tools", ToolRunner: true}, {Module: "nats", SnapshotsTheBus: true, Holds: []Seat{{Name: catalogue.BrokerSeat, Serves: BrokerVerbs()}}, - Processes: []ProcessPrincipal{{Process: "keeper", Emits: []string{"created"}}, {Process: "nats-tools", Tools: true, + Consumes: []string{"mesh-delivery-indexer.manifest-accepted"}, Uses: []Seat{{Name: "operator-channel", Accepts: []string{"condition"}, ByCaller: []string{"condition"}}}, + Streams: []ModuleStream{{Module: "nats", Name: "BROKER_EVENTS", Events: []string{"created", "refused"}, MaxAgeDays: 30, MaxMiB: 256}}, + Processes: []ProcessPrincipal{{Process: "keeper", State: []string{"asked"}}, {Process: "nats-tools", Tools: true, Removes: []string{catalogue.RemovesEngineActs}}}}, {Module: "mod-x"}, {Module: "admin-ish", Invokes: []string{"*"}}, @@ -1107,13 +1109,38 @@ func theBusModulesProcessKeepsBusState(t *testing.T) { t.Errorf("the bus module's process may define %s", s) } } + // A process serves nothing (ADR 0436 §3): it hears its module's consumes on its own consumer, says its module's own + // stream's events and raises its conditions through the operator channel. for _, v := range BrokerVerbs() { - if !MaySubscribe(keeper, "mesh.seat.mesh-broker.tool."+v) { - t.Errorf("the bus module's process does not serve %s", v) + if MaySubscribe(keeper, "mesh.seat.mesh-broker.tool."+v) { + t.Errorf("the bus module's process serves %s", v) } } - if !keeper.AllowResponses { - t.Error("the bus module's process cannot answer what it serves") + if keeper.AllowResponses { + t.Error("the bus module's process answers what it was asked") + } + for _, s := range []string{"$JS.API.CONSUMER.MSG.NEXT.EVENTS.a_nats_keeper", "$JS.ACK.EVENTS.a_nats_keeper.x", + "mesh.stream.nats.event.created", "mesh.seat.operator-channel.accept.condition.nats"} { + if !MayPublish(keeper, s) { + t.Errorf("the bus module's process may not publish %s", s) + } + } + made := false + for _, mc := range AccountConsumers(users) { + if mc.Consumer.Stream == "EVENTS" && mc.Consumer.Name == "a_nats_keeper" && + slices.Equal(mc.Consumer.Filters, []string{"mesh.mod.mesh-delivery-indexer.event.manifest-accepted"}) { + made = true + } + } + if !made { + t.Errorf("the bus module's process's consumer is not made: %+v", AccountConsumers(users)) + } + // Its module's tools serve the reading verbs, and its tools bundle reads the module's own stream back. + if !MaySubscribe(perms["a.node-tools"], "mesh.seat.mesh-broker.tool.list") { + t.Error("nothing serves the seat's list") + } + if !MayPublish(perms["a.nats.nats-tools"], "$JS.API.CONSUMER.CREATE.BROKER_EVENTS.x.y") { + t.Error("the bus module's tools cannot read its events back") } for user, p := range perms { if user == "a.nats.keeper" { diff --git a/modules/mesh-grants/internal/buscompose/busstate.go b/modules/mesh-grants/internal/buscompose/busstate.go index 461bdce5..0f9197ec 100644 --- a/modules/mesh-grants/internal/buscompose/busstate.go +++ b/modules/mesh-grants/internal/buscompose/busstate.go @@ -11,9 +11,12 @@ import ( // **The bus module's process has a bus account of its own** (novox/hq ADR 0595 §4, §5; design 25 § 12.2, § 12.3; // DLV-G2): composed for the process of the module holding the seat mesh-broker (a seat, never a module's name), it // creates, updates, reads, snapshots and deletes every module's bus state under that module's names only, each stream -// by its name (a key-value bucket's and an object store's by the stream the bus keeps it in), and serves the seat's -// eight verbs. No other account holds the four acting verbs' subjects, the tool runner's included: each is denied to -// every principal whose publish would reach it, and only the bus module's process subscribes to them. +// by its name (a key-value bucket's and an object store's by the stream the bus keeps it in); it hears what its module +// consumes (the indexer's manifest-accepted) on a durable consumer of its own, says its module's own stream's events, +// and raises its conditions through the operator channel. As every process account it serves nothing (ADR 0436 §3): the +// seat's four reading verbs are served by its module's tools, and the four acting verbs by nobody on the bus, a +// person's ask reaching the process through its own state. No account holds the acting verbs' subjects, the tool +// runner's included: each is denied to every principal whose publish would reach it. // // The names are one rule (BusStateStreams), read by the bus module where it creates them: until mesh-sdk carries it // as one function (asked of the mesh-sdk member of the milestone), this is the grants part's side of it. @@ -80,6 +83,45 @@ func BusStateStreams(module string, e catalogue.BusStateEntry) []string { return slices.Compact(out) } +// keeperConsumer is the durable consumer the bus module's process hears its module's consumes on, on EVENTS: +// `__`, as a process's own consumer is named (QueueConsumer), filtered on the consumed events. +func keeperConsumer(p Principal) (Consumer, bool) { + var filters []string + for _, c := range p.Consumes { + if s, err := consumedSubject(c); err == nil && !strings.HasPrefix(s, moduleStreamPrefix) { + filters = append(filters, s) + } + } + if len(filters) == 0 { + return Consumer{}, false + } + sort.Strings(filters) + return Consumer{Name: p.Node + "_" + p.Module + "_" + p.Process, Stream: "EVENTS", Filters: slices.Compact(filters), + AckWaitSeconds: 30, MaxDeliver: 5, FromNow: true, + Why: "what the bus module's process " + p.Process + " hears of its module's consumes (novox/hq ADR 0595 §4): the " + + "indexer's accepted verdicts, from now, its first view at start holding what came before"}, true +} + +// keeperConsumerGrants are the subjects the process reads and acknowledges its consumer through. +func keeperConsumerGrants(c Consumer) []string { + return []string{"$JS.API.CONSUMER.INFO." + c.Stream + "." + c.Name, "$JS.API.CONSUMER.MSG.NEXT." + c.Stream + "." + c.Name, + "$JS.ACK." + c.Stream + "." + c.Name + ".>"} +} + +// ownStreamReadGrants are what reading a module's own streams back needs (the bus module's tool history): each stream's +// information and an ordered consumer of its own, as a log's reader has (readLogGrants). +func ownStreamReadGrants(module string, streams []ModuleStream) []string { + var out []string + for _, s := range streams { + if s.Module != module || !safeSubject.MatchString(s.Name) { + continue + } + out = append(out, "$JS.API.STREAM.INFO."+s.Name, "$JS.API.CONSUMER.CREATE."+s.Name+".>", + "$JS.API.CONSUMER.MSG.NEXT."+s.Name+".*", "$JS.API.CONSUMER.INFO."+s.Name+".*", "$JS.API.CONSUMER.DELETE."+s.Name+".*") + } + return out +} + // BusStateNames are every stream the bus module's process may define: those of every entry a manifest asks for now // (asked), and those the bus module said it created and has not dropped (bound, by module: names of streams), so an // orphan whose manifest left the trunk can still be dropped. Sorted, once each. @@ -126,3 +168,14 @@ func keepsBusState(p Principal, pattern string) bool { } return false } + +// withoutActing are a holder's served verbs of mesh-broker without the acting four. +func withoutActing(served []string) []string { + var out []string + for _, v := range served { + if !slices.Contains(BrokerActingVerbs, v) { + out = append(out, v) + } + } + return out +} diff --git a/modules/mesh-grants/internal/buscompose/derived.go b/modules/mesh-grants/internal/buscompose/derived.go index 001726ac..8085e2de 100644 --- a/modules/mesh-grants/internal/buscompose/derived.go +++ b/modules/mesh-grants/internal/buscompose/derived.go @@ -229,6 +229,12 @@ func ConsumersOf(users []Principal) []ModuleConsumer { if p.Kind == KindModule { add(p) } + // The bus module's process hears its module's consumes on a consumer of its own (busstate.go). + if p.Kind == KindProcess && p.KeepsBus { + if c, ok := keeperConsumer(p); ok { + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c}) + } + } } for _, p := range users { if p.Kind != KindNodeTools { diff --git a/modules/mesh-grants/internal/buscompose/nats.go b/modules/mesh-grants/internal/buscompose/nats.go index d31b8b17..b56873b0 100644 --- a/modules/mesh-grants/internal/buscompose/nats.go +++ b/modules/mesh-grants/internal/buscompose/nats.go @@ -110,10 +110,9 @@ type Principal struct { // Ledgers are the local names of the ledgers its module asks for in `requires`, for a KindLedger principal alone // (issue 1966 D5): its whole authority is publishing on those ledgers' histories and snapshots (LedgerPublishGrants). Ledgers []string - // ServesBroker says this is the process of the module holding mesh-broker, the bus module's own (busstate.go, ADR - // 0595 §4, §5): it serves the seat's eight verbs, and KeepsBusState are the streams it defines, every module's bus - // state by its name. - ServesBroker bool + // KeepsBus says this is the process of the module holding mesh-broker, the bus module's own (busstate.go, ADR 0595 + // §4, §5), and KeepsBusState are the streams it defines, every module's bus state by its name. + KeepsBus bool KeepsBusState []string // Process is the process resource's id, for a KindProcess principal alone (novox/hq ADR 0436 §2): // Username and inbox are ... @@ -597,6 +596,8 @@ func PermissionsFor(p Principal) (Permissions, error) { if err != nil { return Permissions{}, err } + // The bus module's tools bundle reads its module's own stream back (its tool history, design 25 § 12.3). + pub = append(pub, ownStreamReadGrants(p.Module, p.Streams)...) sort.Strings(pub) if err := CheckWriters(p, pub); err != nil { return Permissions{}, err @@ -899,10 +900,10 @@ func PermissionsFor(p Principal) (Permissions, error) { "$JS.API.CONSUMER.INFO."+consumerStream(p)+"."+consumerDurable(p), "$JS.API.CONSUMER.MSG.NEXT."+consumerStream(p)+"."+consumerDurable(p)) - // 3. Seats it holds: full participation. But mesh-broker's verbs, which its module's process serves (busstate.go). + // 3. Seats it holds: full participation. But mesh-broker's acting verbs, which nobody serves on the bus (busstate.go). for _, s := range p.Holds { if s.Name == catalogue.BrokerSeat { - s.Serves = nil + s.Serves = withoutActing(s.Serves) } if s.isNewTraffic() { // Composed by SeatTrafficOf below, worker and all; only its tools are served here. @@ -999,9 +1000,9 @@ func PermissionsFor(p Principal) (Permissions, error) { // And its own streams' events (novox/hq ADR 0363 §2). pub = append(pub, ownStreamGrants(d.Module, d.Streams)...) for _, s := range d.Holds { - // mesh-broker's verbs are its holder's process's, never the runtime's (busstate.go). + // mesh-broker's acting verbs are served by nobody on the bus (busstate.go). if s.Name == catalogue.BrokerSeat { - continue + s.Serves = withoutActing(s.Serves) } for _, t := range s.Serves { sub = append(sub, seatToolSubject(s, t, p.Node)) @@ -1196,7 +1197,7 @@ func PermissionsFor(p Principal) (Permissions, error) { // provisioner client is asked by its own machine's tool runner alone, on its own machine's subject (novox/hq // ADR 0370 §3), and answers that. A person is never asked anything, and is granted nothing here. AllowResponses: p.Kind == KindModule || p.Kind == KindNodeTools || p.Kind == KindNode || - p.Kind == KindProvisioner || (p.Kind == KindProcess && p.ServesBroker), + p.Kind == KindProvisioner, }, nil } diff --git a/modules/mesh-grants/internal/buscompose/process_grants.go b/modules/mesh-grants/internal/buscompose/process_grants.go index 4fd6b97d..26f81c37 100644 --- a/modules/mesh-grants/internal/buscompose/process_grants.go +++ b/modules/mesh-grants/internal/buscompose/process_grants.go @@ -52,11 +52,17 @@ func processGrants(p Principal) (pub, sub []string, err error) { } pub = append(pub, invoked...) - // 6. The bus module's process (busstate.go, ADR 0595 §4, §5): every module's bus state it keeps, by name, and the - // seat mesh-broker's eight verbs it serves. - if p.ServesBroker { + // 6. The bus module's process (busstate.go, ADR 0595 §4, §5): every module's bus state it keeps, by name; its own + // consumer of what its module consumes; its module's own stream's events; and its conditions through the seats its + // module uses (the operator channel's accept.condition in its module's name). + if p.KeepsBus { pub = append(pub, busStateKeeperGrants(p.KeepsBusState)...) - sub = append(sub, BrokerVerbSubjects(BrokerVerbs())...) + if c, ok := keeperConsumer(p); ok { + pub = append(pub, keeperConsumerGrants(c)...) + } + pub = append(pub, ownStreamGrants(p.Module, p.Streams)...) + tp, ts := seatTrafficGrants(SeatTrafficOf(p.Module, nil, p.Uses, nil)) + pub, sub = append(pub, tp...), append(sub, ts...) } return pub, sub, nil diff --git a/modules/mesh-grants/internal/buscompose/users.go b/modules/mesh-grants/internal/buscompose/users.go index d348334f..40ca5aa6 100644 --- a/modules/mesh-grants/internal/buscompose/users.go +++ b/modules/mesh-grants/internal/buscompose/users.go @@ -239,10 +239,15 @@ func Users(r Records) ([]Principal, error) { Emits: pr.Emits, Queues: pr.Queues, State: pr.State, Uses: d.Uses, Asks: pr.Asks, Invokes: pr.Invokes, Removes: pr.Removes, }) - // The process of the module holding mesh-broker keeps every module's bus state (busstate.go, ADR 0595 §4). + // The process of the module holding mesh-broker keeps every module's bus state (busstate.go, ADR 0595 §4), + // and its tools bundle reads its module's own stream back. if d.SnapshotsTheBus && !pr.Tools && len(pr.Removes) == 0 { - out[len(out)-1].ServesBroker = true - out[len(out)-1].KeepsBusState = BusStateNames(r.BusState, r.Bound) + k := &out[len(out)-1] + k.KeepsBus, k.KeepsBusState = true, BusStateNames(r.BusState, r.Bound) + k.Consumes, k.Streams = d.Consumes, d.Streams + } + if d.SnapshotsTheBus && len(pr.Removes) > 0 { + out[len(out)-1].Streams = d.Streams } // The ended controller's leftovers include each node's declaration files' bucket (ADR 0590 clause 6). if slices.Contains(pr.Removes, catalogue.RemovesControllerLeftovers) {