mesh-catalog/modules/nats/cmd/nats-tools/removal_test.go

//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)
	}
}