//go:build integration package broker import ( "context" "errors" "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" ) // **A holder of the build seat reads the plans building and the server refuses its write** (novox/hq ADR 0483 clause 2, issue // 1687): on a real server whose users are composed as the controller composes them, the controller makes the bucket and puts a // plan's key; the build agent's credential lists the keys and gets one, and its put of a key is refused with a permissions // violation. It also reads the first waiting ask of a check queue by a direct get, which takes nothing from the queue. func TestNatsTheBuildSeatsHoldersReadThePlansBuildingAndTheServerRefusesTheirWrite(t *testing.T) { // 1687 hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost) if err != nil { t.Fatal(err) } seat := Seat{Name: TheBuildSeat, Scope: "node", Accepts: []string{"build"}, Emits: []string{"built", "started"}} users, err := Users(Records{Nodes: []string{"alpha"}, Assigned: map[string][]Declared{ "alpha": {{Module: RuntimeModule}, {Module: "build-agent", Holds: []Seat{seat}}}, }}) 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) cj := OnConn(controller) if err := AssertMeshStreams(cj); err != nil { t.Fatalf("the controller cannot make the streams: %v", err) } if err := cj.EnsureControllerBuckets(); err != nil { t.Fatalf("the controller cannot make its buckets: %v", err) } cjs, _ := jetstream.New(controller) ckv, err := cjs.KeyValue(ctx, PlansBuildingBucket) if err != nil { t.Fatal(err) } if _, err := ckv.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil { t.Fatalf("the controller cannot put a plan's key: %v", err) } holder, refused := connect("alpha.build-agent") hjs, err := jetstream.New(holder) if err != nil { t.Fatal(err) } kv, err := hjs.KeyValue(ctx, PlansBuildingBucket) if err != nil { t.Fatalf("a holder of the build seat cannot bind %s: %v", PlansBuildingBucket, err) } keys, err := kv.Keys(ctx) if err != nil || len(keys) != 1 || keys[0] != "plan-1" { t.Fatalf("the holder lists %v (%v), want plan-1", keys, err) } if e, err := kv.Get(ctx, "plan-1"); err != nil || e == nil { t.Fatalf("the holder cannot get the key: %v", err) } // Its write is refused by the server. if err := holder.Publish("$KV."+PlansBuildingBucket+".plan-2", []byte(`{}`)); err != nil { t.Fatal(err) } _ = holder.Flush() select { case err := <-refused: if !errors.Is(err, nats.ErrPermissionViolation) { t.Errorf("the server said %v, want a permissions violation", err) } case <-time.After(3 * time.Second): t.Error("the server did not refuse the holder's put of a plan's key") } if _, err := ckv.Get(ctx, "plan-2"); !errors.Is(err, jetstream.ErrKeyNotFound) { t.Errorf("the refused put left a key: %v", err) } // And the direct get of a check queue's first waiting ask, which is granted. reply, err := holder.RequestWithContext(ctx, "$JS.API.DIRECT.GET."+ChecksOtherStream, []byte(`{"seq":1,"next_by_subj":"`+ CheckSubjectPrefix(TheBuildSeat, false)+`>"}`)) if err != nil { t.Fatalf("the holder's direct get of %s was not answered: %v", ChecksOtherStream, err) } if reply.Header.Get("Status") != "404" { t.Errorf("an empty queue answered %q, want 404", reply.Header.Get("Status")) } }