//go:build integration package main import ( "context" "fmt" "os" "path/filepath" "sort" "strings" "testing" "time" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) // The two tools that change the bus, against a server shaped like the mesh's, on the credential the module's // grant proposal gives them and nothing wider (issue 1952 N2, N3): the bus itself refuses what the grant leaves out, // so this is also the proof that the proposed grant is enough and no wider than it needs to be. func TestTheRemovalToolsChangeOnlyWhatTheyAreAskedAndTheGrantAllows(t *testing.T) { dir := t.TempDir() conf := filepath.Join(dir, "nats.conf") if err := os.WriteFile(conf, []byte(fmt.Sprintf(`listen: "127.0.0.1:-1" jetstream { store_dir: %q } accounts { MESH { jetstream: enabled, users: [ { user: "admin", password: "admin" }, { user: "nats-tools", password: "tools", permissions: { publish: { allow: [ "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller", "$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$JS.API.STREAM.INFO.KV_mesh-host_acts", "$KV.mesh-host_acts.>" ] }, subscribe: { allow: ["_INBOX.nats-tools.>"] } } } ] } } `, filepath.Join(dir, "js"))), 0o600); err != nil { t.Fatal(err) } opts, err := server.ProcessConfigFile(conf) if err != nil { t.Fatal(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 server did not start") } defer s.Shutdown() answerWithin = 2 * time.Second ctx := context.Background() admin, err := nats.Connect(s.ClientURL(), nats.UserInfo("admin", "admin")) if err != nil { t.Fatal(err) } defer admin.Close() ajs, _ := jetstream.New(admin) if _, err := ajs.CreateStream(ctx, jetstream.StreamConfig{Name: "DEAD_LETTER_NOTICES", Subjects: []string{"dl.>"}, Retention: jetstream.WorkQueuePolicy}); err != nil { t.Fatal(err) } for _, c := range []struct{ name, filter string }{{"controller", "dl.a"}, {"watchman", "dl.b"}} { if _, err := ajs.CreateConsumer(ctx, "DEAD_LETTER_NOTICES", jetstream.ConsumerConfig{Durable: c.name, FilterSubject: c.filter, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } } acts, err := ajs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-host_acts"}) if err != nil { t.Fatal(err) } other, _ := ajs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "other_state"}) for _, k := range []string{"g14.1", "g14.2", "g14.sub.3", "g14x.4", "ace.5"} { acts.Put(ctx, k, []byte("v")) } other.Put(ctx, "g14.1", []byte("v")) bus := func(context.Context) (jetstream.JetStream, func(), error) { nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("nats-tools", "tools"), nats.CustomInboxPrefix("_INBOX.nats-tools")) if err != nil { return nil, nil, err } js, err := jetstream.New(nc) return js, nc.Close, err } var said []string say := func(event string, body any) error { said = append(said, event); return nil } keys := func(kv jetstream.KeyValue) string { ks, _ := kv.Keys(ctx) sort.Strings(ks) return strings.Join(ks, ",") } // N2: the consumer goes, the stream and its other consumer stay; refused without why and for an unknown one. if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller"}); err == nil || !strings.Contains(err.Error(), "why is required") { t.Fatalf("without why: %v", err) } if _, err := ajs.Consumer(ctx, "DEAD_LETTER_NOTICES", "controller"); err != nil { t.Fatal("refused without why, but the consumer is gone") } // An unknown consumer: the granted name that does not exist is answered "not found"; an ungranted one is dropped. if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "nobody", "why": "x"}); err == nil || !strings.Contains(err.Error(), "not grant") && !strings.Contains(err.Error(), "no consumer") { t.Fatalf("an unknown consumer: %v", err) } if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "watchman", "why": "x"}); err == nil { t.Fatal("a consumer outside the grant was removed") } else if !strings.Contains(err.Error(), "grant") { t.Fatalf("a consumer outside the grant: the refusal does not say so: %v", err) } if _, err := ajs.Consumer(ctx, "DEAD_LETTER_NOTICES", "watchman"); err != nil { t.Fatalf("the grant did not hold, the watchman's consumer: %v", err) } got, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller", "why": "the lease write"}) if err != nil { t.Fatal(err) } if r := got.(ConsumerRemoved); !r.Removed || r.Was["filter_subject"] != "dl.a" || r.Why != "the lease write" { t.Fatalf("%+v", r) } if _, err := ajs.Consumer(ctx, "DEAD_LETTER_NOTICES", "controller"); err == nil { t.Fatal("the consumer is still there") } if _, err := ajs.Consumer(ctx, "DEAD_LETTER_NOTICES", "watchman"); err != nil { t.Fatalf("the other consumer went too: %v", err) } if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller", "why": "again"}); err == nil || !strings.Contains(err.Error(), "no consumer") { t.Fatalf("a consumer already gone: %v", err) } // N3: only the node's keys go; counted first; refused without why, for a whole bucket, and outside the grant. if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14"}); err == nil || !strings.Contains(err.Error(), "why is required") { t.Fatalf("without why: %v", err) } if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "why": "x"}); err == nil { t.Fatal("a removal naming neither node nor prefix was not refused") } if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "other_state", "node": "g14", "why": "x"}); err == nil { t.Fatal("a bucket outside the grant was opened") } if k := keys(acts); k != "ace.5,g14.1,g14.2,g14.sub.3,g14x.4" { t.Fatalf("keys after the refusals: %s", k) } dry, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14", "why": "look", "dry_run": true}) if err != nil || dry.(KeysRemoved).Count != 3 || dry.(KeysRemoved).Removed != 0 { t.Fatalf("dry run: %+v %v", dry, err) } if p, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "prefix": "g14.s", "why": "look", "dry_run": true}); err != nil || p.(KeysRemoved).Count != 1 { t.Fatalf("a prefix that ends inside a token: %+v %v", p, err) } if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14", "why": "x", "expect": 2.0}); err == nil { t.Fatal("a count that differs from the expected was not refused") } got, err = RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14", "why": "the acts bucket is full", "expect": 3.0}) if err != nil { t.Fatal(err) } if r := got.(KeysRemoved); r.Removed != 3 || r.Count != 3 { t.Fatalf("%+v", r) } if k := keys(acts); k != "ace.5,g14x.4" { t.Fatalf("keys left: %s", k) } if k := keys(other); k != "g14.1" { t.Fatalf("another bucket changed: %s", k) } if strings.Join(said, ",") != "consumer-removed,bucket-keys-removed" { t.Fatalf("the audit events said: %v", said) } }