From f75cd0239d4b63b5cb65f3334d5ef3e3ea16b6c1 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 15 Oct 2026 14:11:37 +0200 Subject: [PATCH] Hear a restricted stream only from a module's own account, never the tool runner's (issue 727) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tool runner's bus account carried the heard grants, consumer and bind word of every module it runs, and every agent session on that machine reaches the bus as that account, so any session could pull a granted module's consumer on MAIL. Per ADR 0376 §1 the controller now composes a restricted stream's consumer, grants and bind word only for a module that declares runs-as; registration refuses a consumes of a restricted stream without runs-as (§2), and a grant that composes nothing is said in doctor (D-heard) and in the controller's log on every raise and send. --- cmd/mesh-controller/busobjects.go | 16 ++- cmd/mesh-controller/doctor.go | 5 + cmd/mesh-controller/heard_withheld.go | 66 ++++++++++++ cmd/mesh-controller/heard_withheld_test.go | 43 ++++++++ cmd/mesh-controller/plain_words.go | 2 + internal/broker/derived.go | 2 +- internal/broker/modstreams.go | 59 ++++++++++- internal/broker/modstreams_test.go | 115 +++++++++++++++++++-- internal/broker/nats.go | 7 +- internal/broker/users.go | 4 +- internal/catalogue/seats_declared.go | 2 + internal/catalogue/streams.go | 35 +++++++ internal/catalogue/streams_test.go | 38 +++++++ internal/inventory/busrecords.go | 2 +- internal/inventory/streams_test.go | 2 +- 15 files changed, 377 insertions(+), 21 deletions(-) create mode 100644 cmd/mesh-controller/heard_withheld.go create mode 100644 cmd/mesh-controller/heard_withheld_test.go diff --git a/cmd/mesh-controller/busobjects.go b/cmd/mesh-controller/busobjects.go index 0700d79b..f56bdc47 100644 --- a/cmd/mesh-controller/busobjects.go +++ b/cmd/mesh-controller/busobjects.go @@ -49,10 +49,12 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra // composed from, so a module the mesh grants a consumer's subjects has that consumer waiting. // Done on every raise, not only when a credential is issued: every module moved onto this bus // by the rollout was issued on the old one, and came up with nothing to bind to (2026-09-28). - consumers, err := moduleConsumers(ctx, inv) + consumers, records, err := moduleConsumersAndRecords(ctx, inv) if err != nil { return nil, err } + // And a grant of a restricted stream that composes nothing, said on every raise and send (novox/hq ADR 0376 §2). + sayWithheld(records, " ") // And the work queues of seats that name their caller or their kind, with each holder's worker // (novox/hq ADR 0259 §3): an ask queues until the router takes it, a channel's work until that kind // takes it. @@ -165,15 +167,21 @@ func unassertedObservation(err error) conditions.Observation { // moduleConsumers is every module's durable consumer, from the records the user list is composed from. func moduleConsumers(ctx context.Context, inv *inventory.Inventory) ([]broker.ModuleConsumer, error) { + consumers, _, err := moduleConsumersAndRecords(ctx, inv) + return consumers, err +} + +// moduleConsumersAndRecords is moduleConsumers with the records they were composed from. +func moduleConsumersAndRecords(ctx context.Context, inv *inventory.Inventory) ([]broker.ModuleConsumer, broker.Records, error) { records, err := inv.BusRecords(ctx) if err != nil { - return nil, err + return nil, broker.Records{}, err } users, err := broker.Users(records) if err != nil { - return nil, err + return nil, broker.Records{}, err } - return broker.ConsumersOf(users), nil + return broker.ConsumersOf(users), records, nil } // seatTrafficObjects is the work queues and workers of seats that name their caller or their kind, from diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index a79e5675..93d9799e 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -146,6 +146,11 @@ var probeRegistry = []probe{ {ID: probeFillID, Asserts: "every module's bucket and log holds less than three quarters of its cap; one " + "said filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501", Kind: kindBucketOrLogFilling, Raises: []string{kindBucketOrLogUnread}, Phase: 1, run: probeFill}, + // A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2): the module is carried by the + // machine's tool runner, which never hears a restricted stream, so it hears nothing it was granted. + {ID: probeWithheldID, Asserts: "no grant of a restricted stream names a module that composes nothing for it: every " + + "module granted one runs as an account of its own", From: "ADR 0376 §2", Kind: kindGrantComposesNothing, + Phase: 1, run: probeWithheldHearing}, {ID: "DW", Asserts: "the watchdogs of the signals table ran within three of their intervals", From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Phase: 1, run: probeWatchdogs}, // The core's health definitions (novox/hq to-be 45 §8, ADR 0236): what a core component's new build is diff --git a/cmd/mesh-controller/heard_withheld.go b/cmd/mesh-controller/heard_withheld.go new file mode 100644 index 00000000..31c86821 --- /dev/null +++ b/cmd/mesh-controller/heard_withheld.go @@ -0,0 +1,66 @@ +package main + +import ( + "context" + "fmt" + "strings" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" +) + +// A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2). +// +// A restricted stream is heard only by a module that runs as an account of its own (`runs-as`, ADR 0259): the +// tool runner's bus account is the one every agent session on its machine reaches the bus as, so the controller +// composes no consumer, grant or bind word for a module it carries (broker.ComposedHears). Where a grant still +// names such a module, the module hears nothing it was granted, and that is said — in `doctor` and in the +// controller's log on every raise and send — naming the module, the stream and the remedy. Never silent. + +const ( + // probeWithheldID is the self-check's probe of it, and kindGrantComposesNothing the condition it raises. + probeWithheldID = "D-heard" + kindGrantComposesNothing = "grant-composes-nothing" +) + +// probeWithheldHearing says every granted hearing of a restricted stream composed for nothing. +func probeWithheldHearing(ctx context.Context, d *doctor) ([]conditions.Observation, error) { + records, err := d.open.inventory.BusRecords(ctx) + if err != nil { + return nil, err + } + return withheldObservations(broker.WithheldOf(records)), nil +} + +// withheldObservations is each withheld hearing as a condition of its module on its machine. +func withheldObservations(withheld []broker.Withheld) []conditions.Observation { + var out []conditions.Observation + for _, w := range withheld { + out = append(out, conditions.Observation{Scope: conditions.ScopeModule, + ID: w.Module + "." + strings.ToLower(w.Stream) + "." + w.Node, + Kind: kindGrantComposesNothing, Machine: w.Node, Severity: conditions.Warning, + Summary: w.Said(), Said: fmt.Sprintf("%s on %s granted %s; composed nothing", w.Module, w.Node, w.Stream)}) + } + return sortedFound(out) +} + +// sayWithheld puts every withheld hearing in the controller's log, once per raise or send that composes it. +func sayWithheld(records broker.Records, indent string) { + for _, w := range broker.WithheldOf(records) { + fmt.Printf("%sA GRANT COMPOSES NOTHING: %s\n", indent, w.Said()) + } +} + +// withheldWords are the plain words of a withheld hearing: the module, the machine and what to do. +func withheldWords(o conditions.Observation) words { + module, m := "a module", machineOr(o, "a machine") + if o.Scope == conditions.ScopeModule && idPart(o, 0) != "" { + module = idPart(o, 0) + } + return words{Headline: clip(conditions.Capital(fmt.Sprintf("%s on %s hears nothing it was granted", module, m)), 60), + Needs: fmt.Sprintf("give %s an account of its own in its manifest; the details say which stream.", module), + Explanation: fmt.Sprintf("You granted %s part of a stream that only modules running under an account of "+ + "their own may hear. It runs inside the shared tool runner on %s, which every assistant session there "+ + "also uses, so the mesh gives it none of that stream.", module, m), + Resolved: fmt.Sprintf("%s on %s hears what it was granted, or is no longer granted it", module, m)} +} diff --git a/cmd/mesh-controller/heard_withheld_test.go b/cmd/mesh-controller/heard_withheld_test.go new file mode 100644 index 00000000..76fbf102 --- /dev/null +++ b/cmd/mesh-controller/heard_withheld_test.go @@ -0,0 +1,43 @@ +package main + +import ( + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" +) + +// A grant naming a module the tool runner carries composes nothing, and doctor says so in plain words, naming the +// module, the stream and the remedy (novox/hq ADR 0376 §2). +func TestAGrantThatComposesNothingIsSaid(t *testing.T) { + records := broker.Records{Nodes: []string{"ace"}, Assigned: map[string][]broker.Declared{"ace": { + {Module: broker.RuntimeModule}, + {Module: "reader", Hears: []broker.Heard{{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, + Filters: []string{"mesh.stream.mesh-mail.event.mail.mesh.>"}}}}, + {Module: "mailer", RunsAs: "mailer", Hears: []broker.Heard{{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, + Filters: []string{"mesh.stream.mesh-mail.event.mail.mesh.>"}}}}, + }}} + found := withheldObservations(broker.WithheldOf(records)) + if len(found) != 1 { + t.Fatalf("said %+v", found) + } + o := found[0] + if o.Kind != kindGrantComposesNothing || o.Machine != "ace" || !strings.Contains(o.Summary, "reader on ace") || + !strings.Contains(o.Summary, "MAIL") || !strings.Contains(o.Summary, "runs-as") { + t.Fatalf("%+v", o) + } + w := plainWordings[kindGrantComposesNothing](o) + if why, ok := conditions.PlainWords(w, "ace"); !ok { + t.Fatalf("not plain: %s (%+v)", why, w) + } + registered := false + for _, p := range probeRegistry { + if p.ID == probeWithheldID && p.Kind == kindGrantComposesNothing && p.run != nil { + registered = true + } + } + if !registered { + t.Fatal("doctor does not run the probe of a grant that composes nothing") + } +} diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index da0511e5..a1049696 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -117,6 +117,8 @@ var plainWordings = map[string]func(conditions.Observation) words{ "reaches it. It keeps running what it has.", m), Resolved: m + " can get new instructions again"} }), + // A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2). + kindGrantComposesNothing: worded(withheldWords), kindAgentCanBecomeRoot: worded(func(o conditions.Observation) words { m := machineOr(o, "a machine") return words{Headline: "Sessions on " + m + " could become root", diff --git a/internal/broker/derived.go b/internal/broker/derived.go index 1a770ecf..cd370df1 100644 --- a/internal/broker/derived.go +++ b/internal/broker/derived.go @@ -210,7 +210,7 @@ func ConsumersOf(users []Principal) []ModuleConsumer { for _, d := range p.Carries { add(Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits, Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, - Streams: d.Streams, Hears: d.Hears}) + Streams: d.Streams, Hears: ComposedHears(d)}) } } return out diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index e6e3a315..f909c671 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -98,6 +98,62 @@ type Heard struct { Stream string Emitter string Filters []string + // Restricted says the stream is restricted by its provider's trusted config value (ADR 0363 §4): then it + // is heard only by a module on an account of its own (ADR 0376 §1, ComposedHears). + Restricted bool +} + +// ComposedHears is what of a module's hearing the controller composes (novox/hq ADR 0376 §1): every stream it +// hears, but a restricted one only for a module that runs in a runtime of its own on a bus account of its own +// (`runs-as`, ADR 0259). **The tool runner never hears a restricted stream**: a module it carries reaches the bus +// through the tool runner's account, which every agent session on that machine reaches the bus as too, so a +// consumer, its grants or its bind word composed there would hand the stream to any session. A module that +// declares no `runs-as` and keeps its own container credential gets none either: only `runs-as` is an account +// the mesh keeps from agents. +func ComposedHears(d Declared) []Heard { + if d.RunsAs != "" { + return d.Hears + } + var out []Heard + for _, h := range d.Hears { + if !h.Restricted { + out = append(out, h) + } + } + return out +} + +// Withheld is a restricted stream a grant gives a module that the controller composes nothing for, because the +// module has no account of its own (novox/hq ADR 0376 §2): said in `doctor` and in the controller's log, never +// silent. +type Withheld struct { + Node, Module, Stream, Emitter string +} + +// Said is a withheld hearing in the operator's words: the module, the stream and the remedy. +func (w Withheld) Said() string { + return fmt.Sprintf("%s on %s is granted part of %s's restricted stream %s, and hears none of it: it is carried "+ + "by the machine's tool runner, whose bus account every agent session there shares, so nothing is composed "+ + "for it. Declare runs-as in %s's manifest, so it runs as an account of its own (novox/hq ADR 0376)", + w.Module, w.Node, w.Emitter, w.Stream, w.Module) +} + +// WithheldOf is every granted hearing of a restricted stream composed for nothing, by machine and module. +func WithheldOf(r Records) []Withheld { + var out []Withheld + for _, node := range sortedCopy(r.Nodes) { + for _, d := range r.Assigned[node] { + if d.RunsAs != "" { + continue + } + for _, h := range d.Hears { + if h.Restricted && len(h.Filters) > 0 { + out = append(out, Withheld{Node: node, Module: d.Module, Stream: h.Stream, Emitter: h.Emitter}) + } + } + } + } + return out } // heardConsumer is the durable consumer a module hears a module's stream through: named as its consumer @@ -177,7 +233,8 @@ func streamsIssuedFor(d Declared) []StreamIssued { // HeardIssuedFor is every module stream an assignment hears, as its membership names it. func HeardIssuedFor(node string, d Declared) []HeardIssued { var out []HeardIssued - for _, h := range d.Hears { + // A restricted stream's bind word only for a module of its own account (novox/hq ADR 0376 §1). + for _, h := range ComposedHears(d) { if len(h.Filters) == 0 { continue } diff --git a/internal/broker/modstreams_test.go b/internal/broker/modstreams_test.go index ffa04bd2..9445a64d 100644 --- a/internal/broker/modstreams_test.go +++ b/internal/broker/modstreams_test.go @@ -14,9 +14,14 @@ var mailStream = ModuleStream{Module: "mesh-mail", Name: "MAIL", MaxAgeDays: 7, MaxMiB: 1024, Restricted: true} // mailHeard is what a reader granted the account `mesh` hears of it. -var mailHeard = Heard{Stream: "MAIL", Emitter: "mesh-mail", Filters: []string{ +var mailHeard = Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{ "mesh.stream.mesh-mail.event.mail.mesh.message.added", "mesh.stream.mesh-mail.event.mail.mesh.message.changed"}} +// newsStream is a module's stream without a grant: heard by whoever asks (ADR 0363 §5), the tool runner included. +var newsStream = ModuleStream{Module: "news", Name: "NEWS", Events: []string{"story.**"}, MaxAgeDays: 7, MaxMiB: 16} + +var newsHeard = Heard{Stream: "NEWS", Emitter: "news", Filters: []string{"mesh.stream.news.event.story.>"}} + // A module's stream is its own: subjects EVENTS does not claim, its age and size, kept by limits (ADR 0363 §1, §2). func TestAModuleStreamTakesSubjectsEventsDoesNot(t *testing.T) { s := mailStream.Stream() @@ -91,12 +96,14 @@ func TestTheProviderPublishesItsStreamAndOnlyAGrantedConsumerHearsIt(t *testing. } } -// A granted consumer has a durable consumer on the stream, with its granted subjects as filters, and an EVENTS -// consumer that hears what it gives up on delivered again (ADR 0363 §5, §7); the runtime carrying it may read both. +// A granted consumer of its own account has a durable consumer on the stream, with its granted subjects as +// filters, and an EVENTS consumer that hears what it gives up on delivered again (ADR 0363 §5, §7); a module of its +// own account is no module the runtime carries, so the runtime reads neither (ADR 0376 §1). func TestAGrantedConsumerHasAConsumerOnTheStream(t *testing.T) { records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{ "novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}}, - "ace": {{Module: RuntimeModule}, {Module: "reader", Hears: []Heard{mailHeard}, HearsAStream: true}, {Module: "stranger", HearsAStream: true}}, + "ace": {{Module: RuntimeModule}, {Module: "reader", RunsAs: "reader", Hears: []Heard{mailHeard}, HearsAStream: true}, + {Module: "stranger", HearsAStream: true}}, }} users, err := Users(records) if err != nil { @@ -118,22 +125,112 @@ func TestAGrantedConsumerHasAConsumerOnTheStream(t *testing.T) { t.Fatalf("on EVENTS: %+v", onEvents) } for _, u := range users { - if u.Kind != KindNodeTools || u.Node != "ace" { + if u.Node != "ace" || (u.Kind != KindNodeTools && u.Module != "reader") { continue } perms, err := PermissionsFor(u) if err != nil { t.Fatal(err) } - if !MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader") || !MayPublish(perms, "$JS.ACK.MAIL.ace_reader.1") { - t.Fatal("the runtime carrying the reader may not read its consumer on MAIL") + reads := MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader") && MayPublish(perms, "$JS.ACK.MAIL.ace_reader.1") + if u.Kind == KindNodeTools && reads { + t.Fatal("the runtime may read the consumer on MAIL of a module of its own account") + } + if u.Kind == KindModule && !reads { + t.Fatal("the reader's own account may not read its consumer on MAIL") } if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_stranger") { - t.Fatal("the runtime may read a consumer no module it carries is granted") + t.Fatal("a consumer no module is granted may be read") } } } +// **The tool runner never hears a restricted stream** (ADR 0376 §1, §2): a module it carries, granted part of +// MAIL, gets no consumer on MAIL, the tool runner's account no grant to ask about, pull or acknowledge one there, +// and its membership no word to bind one; a module of its own account granted the same part gets all three, on +// its own account. The grant that composes nothing is said. A stream without a grant is heard through the tool +// runner as before. +func TestTheToolRunnerNeverHearsARestrictedStream(t *testing.T) { + records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{ + "novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}, {Module: "news", Streams: []ModuleStream{newsStream}}}, + "ace": {{Module: RuntimeModule}, + {Module: "carried", Hears: []Heard{mailHeard, newsHeard}, HearsAStream: true}, + {Module: "mailer", RunsAs: "mailer", Hears: []Heard{mailHeard}, HearsAStream: true}}, + }} + users, err := Users(records) + if err != nil { + t.Fatal(err) + } + var onMail, onNews []string + for _, c := range ConsumersOf(users) { + switch c.Consumer.Stream { + case "MAIL": + onMail = append(onMail, c.Consumer.Name) + case "NEWS": + onNews = append(onNews, c.Consumer.Name) + } + } + if !slices.Equal(onMail, []string{"ace_mailer"}) { + t.Fatalf("consumers on MAIL: %v, want only the mailer's", onMail) + } + if !slices.Equal(onNews, []string{"ace_carried"}) { + t.Fatalf("consumers on NEWS: %v, want the carried module's as before", onNews) + } + for _, u := range users { + if u.Node != "ace" { + continue + } + perms, err := PermissionsFor(u) + if err != nil { + t.Fatal(err) + } + switch { + case u.Kind == KindNodeTools: + for _, s := range perms.Publish { + if strings.Contains(s, ".MAIL.") { + t.Errorf("the tool runner's account may publish %s", s) + } + } + for _, s := range perms.Subscribe { + if strings.HasPrefix(s, "mesh.stream.mesh-mail.") { + t.Errorf("the tool runner's account may subscribe %s", s) + } + } + if !MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.NEWS.ace_carried") { + t.Error("the tool runner may no longer read the carried module's consumer on a stream without a grant") + } + case u.Kind == KindModule && u.Module == "carried": + if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_carried") { + t.Error("a module without an account of its own may pull a consumer on MAIL") + } + case u.Kind == KindModule && u.Module == "mailer": + for _, s := range []string{"$JS.API.CONSUMER.INFO.MAIL.ace_mailer", "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_mailer", + "$JS.ACK.MAIL.ace_mailer.1"} { + if !MayPublish(perms, s) { + t.Errorf("the mailer's own account may not publish %s", s) + } + } + } + } + where := Placements{Nodes: map[string][]string{}} + carried := MembershipFor("ace", records.Assigned["ace"][1], where) + if !reflect.DeepEqual(carried.Hears, []HeardIssued{{Stream: "NEWS", Consumer: "ace_carried"}}) { + t.Fatalf("the carried module is told to bind %+v", carried.Hears) + } + mailer := MembershipFor("ace", records.Assigned["ace"][2], where) + if !reflect.DeepEqual(mailer.Hears, []HeardIssued{{Stream: "MAIL", Consumer: "ace_mailer"}}) { + t.Fatalf("the mailer is told to bind %+v", mailer.Hears) + } + withheld := WithheldOf(records) + if len(withheld) != 1 || withheld[0] != (Withheld{Node: "ace", Module: "carried", Stream: "MAIL", Emitter: "mesh-mail"}) { + t.Fatalf("withheld %+v", withheld) + } + if said := withheld[0].Said(); !strings.Contains(said, "carried on ace") || !strings.Contains(said, "MAIL") || + !strings.Contains(said, "runs-as") { + t.Fatalf("said %q", said) + } +} + // The membership says where a module's stream events go and which consumers on module streams to bind // (ADR 0363 §2, §8). func TestTheMembershipNamesTheStreamAndWhatIsHeard(t *testing.T) { @@ -144,7 +241,7 @@ func TestTheMembershipNamesTheStreamAndWhatIsHeard(t *testing.T) { !slices.Equal(provider.Streams[0].Events, mailStream.Events) { t.Fatalf("the provider's membership: %+v", provider.Streams) } - reader := MembershipFor("ace", Declared{Module: "reader", Hears: []Heard{mailHeard}}, where) + reader := MembershipFor("ace", Declared{Module: "reader", RunsAs: "reader", Hears: []Heard{mailHeard}}, where) if !reflect.DeepEqual(reader.Hears, []HeardIssued{{Stream: "MAIL", Consumer: "ace_reader"}}) { t.Fatalf("the reader's membership: %+v", reader.Hears) } diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 6554a0b0..b1f0b831 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -857,9 +857,10 @@ func PermissionsFor(p Principal) (Permissions, error) { for _, d := range p.Carries { own := Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits, Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, - Streams: d.Streams, Hears: d.Hears} - // And the module streams it hears, each through its own consumer there (novox/hq ADR 0363 §8). - pub = append(pub, heardGrants(own, d.Hears)...) + Streams: d.Streams, Hears: ComposedHears(d)} + // And the module streams it hears, each through its own consumer there (novox/hq ADR 0363 §8) — never + // a restricted one, which only a module on an account of its own hears (ADR 0376 §1). + pub = append(pub, heardGrants(own, own.Hears)...) if _, consumes := ConsumerFor(own); !consumes { continue } diff --git a/internal/broker/users.go b/internal/broker/users.go index 7ec4beb3..9ff3f208 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -139,13 +139,15 @@ func Users(r Records) ([]Principal, error) { Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes, State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus, PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads, - Logs: logNames(d.Logs), Streams: d.Streams, Hears: d.Hears, + Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), }) } if runtimeHere { var carried []Declared for _, d := range r.Assigned[node] { if d.RunsAs == "" { + // Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1). + d.Hears = ComposedHears(d) carried = append(carried, d) } } diff --git a/internal/catalogue/seats_declared.go b/internal/catalogue/seats_declared.go index e6e51cb0..77c64810 100644 --- a/internal/catalogue/seats_declared.go +++ b/internal/catalogue/seats_declared.go @@ -247,6 +247,8 @@ func CatalogueProblems(shelf Shelf) []string { var problems []string // A stream name has one owner (novox/hq ADR 0363). problems = append(problems, StreamNameProblems(shelf)...) + // A module that hears a restricted stream runs as an account of its own (novox/hq ADR 0376 §2). + problems = append(problems, RestrictedHearingProblems(shelf)...) // Who declares what, and who declared it first. declaredBy := map[string]string{} diff --git a/internal/catalogue/streams.go b/internal/catalogue/streams.go index 0fd19126..105d3d50 100644 --- a/internal/catalogue/streams.go +++ b/internal/catalogue/streams.go @@ -311,6 +311,41 @@ func StreamNameProblems(shelf Shelf) []string { return problems } +// RestrictedHearingProblems refuses a module that would hear a restricted stream without an account of its own +// (novox/hq ADR 0376 §2): a `consumes` naming a module literally and asking for events a restricted stream of +// that module carries, by a manifest that declares no `runs-as`. A module the machine's tool runner carries +// reaches the bus as the tool runner's account, which every agent session there shares, so the controller would +// compose nothing for it. Judged only where the provider is on the shelf, as a consumer may be installed before +// its emitter; a wildcard emitter never reaches a module's stream (ADR 0363 §5) and is not judged. +func RestrictedHearingProblems(shelf Shelf) []string { + var problems []string + for _, module := range shelfOrder(shelf) { + m := shelf[module] + if m.RunsAs != "" { + continue + } + for _, c := range m.Consumes { + emitter, rest, named := strings.Cut(c, ".") + provider, known := shelf[emitter] + if !named || emitter == OneName || emitter == TheRest || !known { + continue + } + hearing, _ := provider.StreamsHearing(rest) + for _, s := range hearing { + if s.Grant == nil { + continue + } + problems = append(problems, fmt.Sprintf( + "%s consumes %s, which %s's restricted stream %s carries, so %s must run as an account of its "+ + "own (runs-as): the machine's runtime runs as the operator's account, which every agent "+ + "runs as, and never hears a restricted stream (novox/hq ADR 0376)", + module, c, emitter, s.BusName(), module)) + } + } + } + return problems +} + // GrantedKeys reads a restricting config value (novox/hq ADR 0363 §4): each key whose list names the // module wherever it runs, or the module on this machine as `.`. A value of another shape // grants nothing; a key that is not one usable name grants nothing. diff --git a/internal/catalogue/streams_test.go b/internal/catalogue/streams_test.go index 5ea8c4e7..a5971e3c 100644 --- a/internal/catalogue/streams_test.go +++ b/internal/catalogue/streams_test.go @@ -204,3 +204,41 @@ func TestNarrowingTwoPatterns(t *testing.T) { t.Fatalf("uncovered: %v", got) } } + +// **A module that hears a restricted stream runs as an account of its own** (ADR 0376 §2): registration refuses a +// `consumes` naming a module and asking for events its restricted stream carries, unless the manifest declares +// `runs-as`; a stream without a grant, an event the provider says on EVENTS, and a wildcard emitter stay as they +// were, and a consumer whose provider is not on the shelf is judged once it is. +func TestHearingARestrictedStreamNeedsAnAccountOfItsOwn(t *testing.T) { + provider, err := ParseManifest([]byte(mailManifest(t, mailStream, `"emits":["account.added"],`))) + if err != nil { + t.Fatal(err) + } + feed := Manifest{Module: "feed", Streams: []StreamDeclaration{{Name: "items", Events: []string{"item.**"}}}} + carried := Manifest{Module: "reader", Consumes: []string{"mesh-mail.mail.*.message.added"}} + got := RestrictedHearingProblems(Shelf{"mesh-mail": provider, "reader": carried}) + if len(got) != 1 || !strings.Contains(got[0], "reader consumes mesh-mail.mail.*.message.added") || + !strings.Contains(got[0], "MAIL") || !strings.Contains(got[0], "runs-as") || !strings.Contains(got[0], "ADR 0376") { + t.Fatalf("a carried module hearing MAIL was not refused as it should be: %v", got) + } + if all := strings.Join(CatalogueProblems(Shelf{"mesh-mail": provider, "reader": carried}), "; "); !strings.Contains(all, got[0]) { + t.Fatalf("the catalogue's check does not refuse it: %s", all) + } + own := carried + own.RunsAs = "reader" + if got := RestrictedHearingProblems(Shelf{"mesh-mail": provider, "reader": own}); len(got) != 0 { + t.Fatalf("a module of its own account was refused: %v", got) + } + for _, m := range []Manifest{ + {Module: "events", Consumes: []string{"mesh-mail.account.added"}}, + {Module: "items", Consumes: []string{"feed.item.**"}}, + {Module: "audit", Consumes: []string{"**", "*.mail.**"}}, + } { + if got := RestrictedHearingProblems(Shelf{"mesh-mail": provider, "feed": feed, m.Module: m}); len(got) != 0 { + t.Errorf("%s was refused: %v", m.Module, got) + } + } + if got := RestrictedHearingProblems(Shelf{"reader": carried}); len(got) != 0 { + t.Fatalf("judged without its provider: %v", got) + } +} diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index e8cf32f5..bb6f9491 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -496,7 +496,7 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node }) var out []broker.Heard for _, p := range parts { - h := broker.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter} + h := broker.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter, Restricted: p.Stream.Grant != nil} for _, pattern := range p.Patterns { h.Filters = append(h.Filters, broker.ModuleStreamSubject(p.Emitter, broker.WirePattern(pattern))) } diff --git a/internal/inventory/streams_test.go b/internal/inventory/streams_test.go index 9efef62f..dbddf979 100644 --- a/internal/inventory/streams_test.go +++ b/internal/inventory/streams_test.go @@ -32,7 +32,7 @@ func TestWhatAModuleHearsIsComposedOntoTheStreamsSubjects(t *testing.T) { reader := catalogue.Manifest{Module: "reader"} hears, touches, onEvents := hearsOf(reader, map[string]catalogue.Manifest{"mesh-mail": provider}, "ace", grants, []string{"mesh-mail.mail.**", "billing.paid"}) - if !touches || len(hears) != 1 || hears[0].Stream != "MAIL" || + if !touches || len(hears) != 1 || hears[0].Stream != "MAIL" || !hears[0].Restricted || !slices.Equal(hears[0].Filters, []string{"mesh.stream.mesh-mail.event.mail.mesh.message.added"}) { t.Fatalf("heard %+v", hears) }