diff --git a/internal/broker/composed.go b/internal/broker/composed.go index 8631849a..8bfb0634 100644 --- a/internal/broker/composed.go +++ b/internal/broker/composed.go @@ -128,7 +128,10 @@ func HandPathWrites(nodes []string) ([]string, error) { if err != nil { return nil, err } - out = append(out, files, "$JS.API.STREAM.INFO.OBJ_declaration-files-"+node) + stream := "OBJ_" + DeclarationFilesBucket + node + // The object store's put reads an object's metadata already there and, replacing it, purges its old chunks. + out = append(out, files, "$JS.API.STREAM.INFO."+stream, "$JS.API.DIRECT.GET."+stream+".>", + "$JS.API.STREAM.MSG.GET."+stream, "$JS.API.STREAM.PURGE."+stream) } return out, nil } diff --git a/internal/broker/composed_live_test.go b/internal/broker/composed_live_test.go new file mode 100644 index 00000000..b7de6fce --- /dev/null +++ b/internal/broker/composed_live_test.go @@ -0,0 +1,142 @@ +//go:build integration + +package broker + +import ( + "bytes" + "context" + "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" +) + +// 1945 (the review of A1, H1): on a bus running the user list this controller composes, the operator's account (a +// person who may invoke every tool) walks release 6's hand path end to end: it writes a node's want, that node's +// composition and its digest, and stores a file over 64 KiB in the node's object bucket and replaces it; a person with +// fewer tools may do none of it. +func TestNatsTheOperatorWritesTheHandPathOnTheComposedUserList(t *testing.T) { + hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost) + if err != nil { + t.Fatal(err) + } + users, err := Users(Records{Nodes: []string{"ace"}, + People: map[string][]string{"jochen": {"*"}, "guest": {"mesh-controller.status"}}}) + 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(), 60*time.Second) + defer cancel() + connect := func(user string) jetstream.JetStream { + t.Helper() + nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user)) + if err != nil { + t.Fatalf("%s cannot connect: %v", user, err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + return js + } + + // The buckets as the controller and the assigner make them. + cjs := connect(ControllerName) + if err := EnsureDeclarationFiles(ctx, cjs, "ace"); err != nil { + t.Fatal(err) + } + if err := EnsureComposedDigests(ctx, cjs); err != nil { + t.Fatal(err) + } + if _, err := cjs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "composed-ace", History: 5}); err != nil { + t.Fatal(err) + } + // The assigner's bucket, made here by the controller's account, which may create any stream. + if _, err := cjs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-delivery-assigner_wants", History: 20}); err != nil { + t.Fatal(err) + } + + short := func(c context.Context) (context.Context, context.CancelFunc) { return context.WithTimeout(c, 5*time.Second) } + for _, who := range []struct { + user string + may bool + }{{"person.jochen", true}, {"person.guest", false}} { + js := connect(who.user) + try := func(what string, act func(context.Context) error) { + t.Helper() + c, stop := short(ctx) + defer stop() + err := act(c) + if (err == nil) != who.may { + t.Errorf("%s: %s: %v (may: %v)", who.user, what, err, who.may) + } + } + try("write ace's want", func(c context.Context) error { + kv, err := js.KeyValue(c, "mesh-delivery-assigner_wants") + if err != nil { + return err + } + _, err = kv.Put(c, "want.ace.mesh-controller", []byte(`{"v":1}`)) + return err + }) + try("write ace's composition", func(c context.Context) error { + kv, err := js.KeyValue(c, "composed-ace") + if err != nil { + return err + } + _, err = kv.Put(c, "ace.mesh-controller.aaaa", []byte(`{"v":1}`)) + return err + }) + try("write its digest", func(c context.Context) error { + kv, err := js.KeyValue(c, ControllerComposedBucket) + if err != nil { + return err + } + _, err = kv.Put(c, "ace.mesh-controller.aaaa", []byte(`{"v":1}`)) + return err + }) + try("store and replace a file over 64 KiB", func(c context.Context) error { + store, err := js.ObjectStore(c, "declaration-files-ace") + if err != nil { + return err + } + body := bytes.Repeat([]byte("x"), 200<<10) + if _, err := store.PutBytes(c, "f00d", body); err != nil { + return err + } + _, err = store.PutBytes(c, "f00d", append(body, 'y')) + return err + }) + } +} diff --git a/internal/broker/composed_test.go b/internal/broker/composed_test.go index a5d5a309..19e95265 100644 --- a/internal/broker/composed_test.go +++ b/internal/broker/composed_test.go @@ -76,7 +76,9 @@ func TestTheOperatorWritesTheHandPathAheadOfTheSwitch(t *testing.T) { t.Fatal(err) } hand := []string{"$KV.mesh-delivery-assigner_wants.want.>", "$KV.composed-ace.>", "$KV.composed-novox.>", - "$KV.mesh-controller_composed.>", "$O.declaration-files-ace.>"} + "$KV.mesh-controller_composed.>", "$O.declaration-files-ace.>", + "$JS.API.DIRECT.GET.OBJ_declaration-files-ace.>", "$JS.API.STREAM.MSG.GET.OBJ_declaration-files-ace", + "$JS.API.STREAM.PURGE.OBJ_declaration-files-ace"} for _, s := range hand { if got := slices.Contains(perms.Publish, s); got != (u.Module == "jochen") { t.Errorf("%s may publish %s: %v", u.Module, s, got) diff --git a/internal/broker/log_test.go b/internal/broker/log_test.go index b1988cdd..1d8ac164 100644 --- a/internal/broker/log_test.go +++ b/internal/broker/log_test.go @@ -206,8 +206,11 @@ func TestNoPathDeletesALogOrABucket(t *testing.T) { }) for _, lit := range stringsIn(f) { if strings.Contains(lit, "$JS.API.STREAM.DELETE") || strings.Contains(lit, "$JS.API.STREAM.PURGE") { - // Only the writers table names it, as a subject no one but the controller may publish. - if !strings.HasSuffix(filepath.ToSlash(path), "internal/broker/writers.go") { + // Only the writers table names it, as a subject no one but the controller may publish; and the operator's + // release 6 hand path, whose object store purges the chunks of an object it replaces in a node's + // object bucket (novox/hq ADR 0539 clause 10): a grant, never a call of this repository. + if !strings.HasSuffix(filepath.ToSlash(path), "internal/broker/writers.go") && + !strings.HasSuffix(filepath.ToSlash(path), "internal/broker/composed.go") { found = append(found, path+": "+lit) } }