The broker asserts the have bucket and grants each node its own keys of it, the controller none (issue 1136)

This commit is contained in:
2026-10-18 19:04:55 +02:00
parent 6746ab6492
commit 661e8eff2f
10 changed files with 724 additions and 19 deletions
+25 -2
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
)
@@ -189,14 +190,15 @@ func IsControllerBucket(bucket string) bool {
bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket ||
bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket ||
bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == CheckRunningBucket ||
bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket
bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket ||
bucket == delivery.HaveBucket
}
// ControllerBuckets are the controller's own buckets, in the order they are asserted.
func ControllerBuckets() []string {
return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket,
AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket, CheckRunningBucket,
DeliveryMergesBucket, BuildsBucket, RolloutsBucket}
DeliveryMergesBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket}
}
// ControllerBucketsAsserter is what raising the controller's buckets needs of a connection.
@@ -370,6 +372,12 @@ func (j *JetStream) EnsureControllerBuckets() error {
return fmt.Errorf("asserting bucket %s: %w", c.bucket, err)
}
}
// **The have bucket: what each machine has of each module** (ADR 0429 §1, issue 1136). Twenty revisions of each key and
// no age. Asserted by the controller, written by each machine's node-engine under its own node's keys and by nobody
// else: the controller's grant leaves it out.
if _, err := js.CreateOrUpdateKeyValue(ctx, HaveBucketConfig()); err != nil {
return fmt.Errorf("asserting bucket %s: %w", delivery.HaveBucket, err)
}
// **The assignments keep the marker age they had as a mirror.** A stream that allows messages an age cannot be
// made to disallow them (the server refuses: "message TTL status can not be disabled"), so the bucket that was
// the assignments mirror is asserted with the age it was made with. Nothing is purged from it, so no marker is ever
@@ -439,3 +447,18 @@ func EnsureLeaseBucket(ctx context.Context, js jetstream.JetStream) error {
}
return nil
}
// HaveBucketConfig is the options of the have bucket (novox/hq ADR 0429 §1): delivery.HaveHistory revisions of each
// key, no age, a value under a kilobyte (a have is JSON of about 500 bytes), 64 MiB on disk like the assignments.
func HaveBucketConfig() jetstream.KeyValueConfig {
return jetstream.KeyValueConfig{
Bucket: delivery.HaveBucket,
Description: "what each machine has of each module, one key `<node>.<module>` (and `<node>.node` for the machine's " +
"own facts) (novox/hq ADR 0429, design 47): written by each machine's node-engine under its own node's keys " +
"alone, never by the controller; read by the delivery view and the watchman",
History: delivery.HaveHistory,
MaxValueSize: 4 << 10,
MaxBytes: 64 << 20,
Storage: jetstream.FileStorage,
}
}
+154
View File
@@ -0,0 +1,154 @@
//go:build integration
package broker
import (
"context"
"errors"
"os"
"path/filepath"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"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"
)
// **Each node's credential writes its own have keys and no other** (novox/hq ADR 0429 §1, issue 1136, 1136/AC4): on a
// real server whose users are composed as the controller composes them, node alpha's put of alpha.zsh is taken, its
// put of beta.zsh is refused by the server, a module reading node-engine.have is refused any put and its watch sees
// alpha.zsh.
func TestNatsANodePutsOnlyItsOwnHaveKeys(t *testing.T) { // 1136/AC4
hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
if err != nil {
t.Fatal(err)
}
users, err := Users(Records{Nodes: []string{"alpha", "beta"}, Assigned: map[string][]Declared{
"alpha": {{Module: RuntimeModule}, {Module: "mesh-delivery-view", Reads: []string{"node-engine.have"}}},
"beta": {{Module: RuntimeModule}},
}})
if err != nil {
t.Fatal(err)
}
for i := range users {
users[i].PasswordHash = string(hash)
}
accounts, err := ComposeAccounts(users)
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")
}
t.Cleanup(s.Shutdown)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
connect := func(user string) (*nats.Conn, chan error) {
t.Helper()
refused := make(chan error, 8)
nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
if err != nil {
t.Fatalf("%s cannot connect: %v", user, err)
}
t.Cleanup(nc.Close)
return nc, refused
}
controller, _ := connect(ControllerName)
cjs, err := jetstream.New(controller)
if err != nil {
t.Fatal(err)
}
if _, err := cjs.CreateOrUpdateKeyValue(ctx, HaveBucketConfig()); err != nil {
t.Fatalf("the controller cannot make %s: %v", delivery.HaveBucket, err)
}
alpha, _ := connect("node.alpha")
ajs, err := jetstream.New(alpha)
if err != nil {
t.Fatal(err)
}
akv, err := ajs.KeyValue(ctx, delivery.HaveBucket)
if err != nil {
t.Fatalf("node alpha cannot bind %s: %v", delivery.HaveBucket, err)
}
if _, err := akv.Put(ctx, "alpha.zsh", []byte(`{"v":1}`)); err != nil {
t.Fatalf("node alpha cannot put alpha.zsh: %v", err)
}
if e, err := akv.Get(ctx, "alpha.zsh"); err != nil || string(e.Value()) != `{"v":1}` {
t.Errorf("node alpha reads %v (%v) of its own key", e, err)
}
reader, readerRefused := connect("alpha.mesh-delivery-view")
for _, other := range []struct {
name string
nc *nats.Conn
ch chan error
key string
}{{"node alpha putting beta.zsh", alpha, nil, "beta.zsh"},
{"the reading module putting alpha.zsh", reader, readerRefused, "alpha.zsh"}} {
ch := other.ch
if ch == nil {
// alpha's own refusals arrive on its own handler: reconnect with one that is watched.
var nc *nats.Conn
nc, ch = connect("node.alpha")
other.nc = nc
}
if err := other.nc.Publish("$KV."+delivery.HaveBucket+"."+other.key, []byte(`{"v":99}`)); err != nil {
t.Fatal(err)
}
_ = other.nc.Flush()
select {
case err := <-ch:
if !errors.Is(err, nats.ErrPermissionViolation) {
t.Errorf("%s: the server said %v, want a permissions violation", other.name, err)
}
case <-time.After(3 * time.Second):
t.Errorf("the server did not refuse %s", other.name)
}
}
// The reader watches and sees alpha.zsh, and beta.zsh was never taken.
rjs, err := jetstream.New(reader)
if err != nil {
t.Fatal(err)
}
rkv, err := rjs.KeyValue(ctx, delivery.HaveBucket)
if err != nil {
t.Fatalf("a module reading node-engine.have cannot bind it: %v", err)
}
w, err := rkv.WatchAll(ctx)
if err != nil {
t.Fatalf("the reading module cannot watch: %v", err)
}
defer w.Stop()
seen := map[string]string{}
for e := range w.Updates() {
if e == nil {
break
}
seen[e.Key()] = string(e.Value())
}
if seen["alpha.zsh"] != `{"v":1}` || len(seen) != 1 {
t.Errorf("the reading module's watch sees %v, want alpha.zsh alone", seen)
}
}
+106
View File
@@ -0,0 +1,106 @@
package broker
import (
"slices"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
)
// 1136/AC1: the controller's start asserts node-engine_have with the history delivery keeps, and the bucket is one of
// the controller's own, so raising the buckets does not name it undeclared (ADR 0429 §1).
func TestTheHaveBucketIsTheControllersAndKeepsTwentyRevisions(t *testing.T) { // 1136/AC1
cfg := HaveBucketConfig()
if cfg.Bucket != delivery.HaveBucket || cfg.History != delivery.HaveHistory || delivery.HaveHistory != 20 {
t.Fatalf("the have bucket is asserted as %q with history %d, want %q with 20", cfg.Bucket, cfg.History, delivery.HaveBucket)
}
if cfg.Storage != jetstream.FileStorage || cfg.MaxBytes <= 0 || cfg.MaxValueSize <= 0 {
t.Fatalf("the have bucket is not bounded on disk: %+v", cfg)
}
if !IsControllerBucket(delivery.HaveBucket) || !slices.Contains(ControllerBuckets(), delivery.HaveBucket) {
t.Fatalf("%s is not among the controller's buckets", delivery.HaveBucket)
}
undeclared, err := RaiseBuckets(fakeBuckets{names: []string{delivery.HaveBucket}}, nil)
if err != nil {
t.Fatal(err)
}
if len(undeclared) != 0 {
t.Fatalf("%v are raised as undeclared", undeclared)
}
}
// 1136/AC2: a node may write and read the keys of its own node of the have bucket, with what a get, create, update and
// watch of them need, and no other key of it; the controller does not write it; an unsafe node name is refused.
func TestANodeWritesAndReadsOnlyItsOwnHaveKeys(t *testing.T) { // 1136/AC2
perms, err := PermissionsFor(Principal{Kind: KindNode, Node: "one", PasswordHash: "x"})
if err != nil {
t.Fatal(err)
}
b, stream := delivery.HaveBucket, "KV_"+delivery.HaveBucket
for _, want := range []string{
"$KV." + b + ".one.>",
"$JS.API.STREAM.INFO." + stream,
"$JS.API.DIRECT.GET." + stream + ".$KV." + b + ".one.>",
"$JS.API.CONSUMER.CREATE." + stream + ".*.$KV." + b + ".one.>",
"$JS.API.CONSUMER.DELETE." + stream + ".>",
"$JS.FC." + stream + ".>",
} {
has(t, perms.Publish, want)
}
for _, p := range perms.Publish {
if !strings.Contains(p, delivery.HaveBucket) {
continue
}
if strings.Contains(p, "KV."+b+".>") || strings.Contains(p, "KV."+b+".two") || strings.HasSuffix(p, "."+b+".*") ||
strings.Contains(p, "DIRECT.GET."+stream+".>") || strings.Contains(p, "CONSUMER.CREATE."+stream+".>") {
t.Errorf("the node is granted %q, which reaches keys of other nodes", p)
}
}
if err := CheckWriters(Principal{Kind: KindNode, Node: "one"}, perms.Publish); err != nil {
t.Fatalf("the node's own have keys are refused as a second writer: %v", err)
}
controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
if err != nil {
t.Fatal(err)
}
for _, p := range controller.Publish {
if SubjectsOverlap(p, "$KV."+b+".one.zsh") {
t.Errorf("the controller may write the have keys through %q", p)
}
}
for _, bad := range []string{"on.e", "one.>", "*", "o>ne"} {
if _, err := PermissionsFor(Principal{Kind: KindNode, Node: bad, PasswordHash: "x"}); err == nil {
t.Errorf("node %q was accepted as part of a have subject", bad)
}
}
// A principal that is not a node writes no have key: the writers table refuses it.
if err := CheckWriters(Principal{Kind: KindModule, Module: "billing", Node: "one"}, []string{"$KV." + b + ".one.>"}); err == nil {
t.Error("a module granted the have keys of a node is not refused as a second writer")
}
if err := CheckWriters(Principal{Kind: KindNode, Node: "two"}, []string{"$KV." + b + ".one.>"}); err == nil {
t.Error("a node granted another node's have keys is not refused as a second writer")
}
}
// 1136/AC3: a module whose manifest reads node-engine.have, mesh-controller.assignments and mesh-controller.builds may
// watch and get every key of the three and write none.
func TestAModuleReadingHaveAssignmentsAndBuildsReadsAndWritesNone(t *testing.T) { // 1136/AC3
perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-view", PasswordHash: "x",
Reads: []string{"node-engine.have", "mesh-controller.assignments", "mesh-controller.builds"}})
if err != nil {
t.Fatal(err)
}
for _, b := range []string{delivery.HaveBucket, AssignmentsBucket, BuildsBucket} {
stream := "KV_" + b
has(t, perms.Publish, "$JS.API.STREAM.INFO."+stream)
has(t, perms.Publish, "$JS.API.DIRECT.GET."+stream+".>")
has(t, perms.Publish, "$JS.API.CONSUMER.CREATE."+stream+".>")
for _, p := range perms.Publish {
if SubjectsOverlap(p, "$KV."+b+".any.key") {
t.Errorf("the reading module may write %s through %q", b, p)
}
}
}
}
+11
View File
@@ -21,6 +21,7 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/lease"
)
@@ -712,6 +713,9 @@ func PermissionsFor(p Principal) (Permissions, error) {
// hand, which it alone writes. A put is a publish to the bucket's subject, which `$JS.API.>`
// does not cover; each bucket named, not `$KV.>`, which would let it write any module's state.
for _, bucket := range ControllerBuckets() {
if bucket == delivery.HaveBucket {
continue // each node's node-engine writes its own keys of it, the controller none (ADR 0429 §1)
}
pub = append(pub, "$KV."+bucket+".>")
}
// **And what the bus says about itself, read-only** (novox/hq to-be 45 §3, S9): a durable
@@ -800,6 +804,13 @@ func PermissionsFor(p Principal) (Permissions, error) {
// contract is lease/witness.go): it asks its own machine's node tools PING, and, where the
// machine runs the controller, reads the lease's one key — read, never written.
pub = append(pub, WitnessSubjects(p)...)
// And the keys of its own node in the have bucket, which it alone writes (novox/hq ADR 0429 §1, issue 1136):
// what the node has of each module, read by the view and the watchman. Never the controller's.
haveKeys, err := NodeKeysGrants(delivery.HaveBucket, p.Node)
if err != nil {
return Permissions{}, err
}
pub = append(pub, haveKeys...)
// And, while the vault serves it, the vault's move-in in its own name and nobody else's (issue 676): the
// values sealed to this machine, which only it opens, handed in sealed to the vault's key for the call.
if p.MovesIntoTheVault {
+32 -16
View File
@@ -65,6 +65,37 @@ func bucketOfRead(read string) (string, bool) {
return BucketName(module, name), true
}
// readGrants is what reaching a bucket's keys needs: binding to it, a direct get, an ordered consumer for listing and
// watching and the deletion of that consumer and its flow control. key is one key's subject tail, a pattern such as
// `one.>` for a prefix of keys, or empty for every key. The bucket is made with AllowDirect, so a get is the direct get
// and never the stream's message get.
func readGrants(bucket, key string) []string {
stream := "KV_" + bucket
out := []string{"$JS.API.STREAM.INFO." + stream}
if key == "" {
out = append(out,
"$JS.API.DIRECT.GET."+stream+".>",
"$JS.API.CONSUMER.CREATE."+stream+".>")
} else {
out = append(out,
"$JS.API.DIRECT.GET."+stream+".$KV."+bucket+"."+key,
"$JS.API.CONSUMER.CREATE."+stream+".*.$KV."+bucket+"."+key)
}
return append(out,
"$JS.API.CONSUMER.DELETE."+stream+".>",
"$JS.FC."+stream+".>")
}
// NodeKeysGrants is what a node's node-engine is granted of a bucket it writes by node, `<node>.>` and no other key
// (novox/hq ADR 0429 §1: each key one writer, enforced by credentials): the writes of those keys and what a get,
// create, update and watch of them need. The node name is checked as every subject token is.
func NodeKeysGrants(bucket, node string) ([]string, error) {
if !safeSubject.MatchString(node) {
return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", node)
}
return append([]string{"$KV." + bucket + "." + node + ".>"}, readGrants(bucket, node+".>")...), nil
}
// stateGrants is what a principal publishes to reach the state its modules keep and read: for every
// bucket, binding to it, reading a key directly, and an ordered consumer for listing and watching,
// created and deleted on the bucket's own stream, with its flow control answered; for a bucket an
@@ -84,22 +115,7 @@ func bucketOfRead(read string) (string, bool) {
// consumer (named at random by the client) and flow control; none of them reads or writes a value.
func stateGrants(a stateAccess) []string {
var out []string
read := func(bucket, key string) {
stream := "KV_" + bucket
out = append(out, "$JS.API.STREAM.INFO."+stream)
if key == "" {
out = append(out,
"$JS.API.DIRECT.GET."+stream+".>",
"$JS.API.CONSUMER.CREATE."+stream+".>")
} else {
out = append(out,
"$JS.API.DIRECT.GET."+stream+".$KV."+bucket+"."+key,
"$JS.API.CONSUMER.CREATE."+stream+".*.$KV."+bucket+"."+key)
}
out = append(out,
"$JS.API.CONSUMER.DELETE."+stream+".>",
"$JS.FC."+stream+".>")
}
read := func(bucket, key string) { out = append(out, readGrants(bucket, key)...) }
machineKey := ""
if safeSubject.MatchString(a.Node) {
machineKey = a.Node
+1 -1
View File
@@ -33,7 +33,7 @@ accounts {
subscribe: { allow: ["_INBOX.enrol.one.>"] }
} }
{ user: "node.one", password: "$2a$11$nnnnnnnnnnnnnnnnnnnnnn", permissions: {
publish: { allow: ["$JS.ACK.NODES.one.>", "$JS.API.CONSUMER.INFO.NODES.one", "$SRV.PING.node-tools.one", "mesh.control.one.>"] }
publish: { allow: ["$JS.ACK.NODES.one.>", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.one.>", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.CONSUMER.INFO.NODES.one", "$JS.API.DIRECT.GET.KV_node-engine_have.$KV.node-engine_have.one.>", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.FC.KV_node-engine_have.>", "$KV.node-engine_have.one.>", "$SRV.PING.node-tools.one", "mesh.control.one.>"] }
subscribe: { allow: ["_DELIVER.one", "_DELIVER.one.>", "_INBOX.node.one.>", "mesh.node.one.ask.hand-over", "mesh.node.one.ask.judge-directories", "mesh.node.one.ask.move-secret", "mesh.node.one.ask.report", "mesh.node.one.ask.setuid-search", "mesh.node.one.declare"] }
allow_responses: { max: 1, ttl: "1m" }
} }
+14
View File
@@ -4,6 +4,8 @@ import (
"fmt"
"slices"
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// The writers table (novox/hq to-be 45 §1, ADR 0227 rule 1), compiled in.
@@ -92,6 +94,13 @@ func ownProvisioner(p Principal, pattern string) bool {
return p.Kind == KindProvisioner && len(tokens) > 3 && tokens[2] == p.Node
}
// ownNodeKeys is a node writing the have keys of its own node: `$KV.<bucket>.<node>.…`, the node in the third token
// and no wildcard there (novox/hq ADR 0429 §1).
func ownNodeKeys(p Principal, pattern string) bool {
tokens := strings.Split(pattern, ".")
return p.Kind == KindNode && len(tokens) > 2 && tokens[2] == p.Node
}
// kvOf is a bucket's write subjects.
func kvOf(bucket string) []string { return []string{"$KV." + bucket + ".>"} }
@@ -205,6 +214,11 @@ var WritersTable = []WriterRow{
{State: "the rollouts of the chain", Writer: "controller (lease holder), compare-and-set on the key's revision",
KeptIn: "key-value " + RolloutsBucket + ", sixty-four revisions of each key, none removed",
Others: "read through `reads` as mesh-controller." + ChainRollouts, Subjects: kvOf(RolloutsBucket), Writes: isController},
// What each machine has of each module (novox/hq ADR 0429 §1, design 47): written by that machine's node-engine under
// its own node's keys, never by the controller; the view and the watchman read it.
{State: "a machine's have", Writer: "the node-engine of the machine, under its own keys `<node>.>`",
KeptIn: "key-value " + delivery.HaveBucket + ", twenty revisions of each key",
Others: "read through `reads` as node-engine.have", Subjects: kvOf(delivery.HaveBucket), Writes: ownNodeKeys},
{State: "the declared-config mirror", Writer: "controller (lease holder), from its store, only on a change",
KeptIn: "key-value " + DeclaredConfigBucket, Others: "read through `reads` as mesh-controller." + MirrorDeclaredConfig,
Subjects: kvOf(DeclaredConfigBucket), Writes: isController},
+2
View File
@@ -41,6 +41,8 @@ var designRows = []string{
// The chain beside the assignments, the record itself and no mirror of one (ADR 0408, design 47, issue 1089).
"the builds of the chain",
"the rollouts of the chain",
// What each machine has of each module, written by its node-engine under its own keys (ADR 0429 §1, issue 1136).
"a machine's have",
"the declared-config mirror",
"the predefined-keys mirror",
// The provisioner client's word on what it placed (novox/hq ADR 0349 §4, design 57 P6).