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 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. Issue 1126 B5 (ADR 0590
// §6): the controller's leftovers go the same way, a consumer of EVENTS and a bucket's stream among them, while EVENTS
// and its other consumers stay.
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.ASSIGNMENTS",
        "$JS.API.STREAM.DELETE.ASSIGNMENTS",
        "$JS.API.CONSUMER.INFO.EVENTS.controller-merges",
        "$JS.API.CONSUMER.DELETE.EVENTS.controller-merges",
        "$JS.API.STREAM.INFO.KV_mesh-controller_rollouts",
        "$JS.API.STREAM.DELETE.KV_mesh-controller_rollouts",
        "$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)
	}
	if _, err := ajs.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNMENTS", Subjects: []string{"mesh.assignment.*.*"}, MaxMsgsPerSubject: 1}); err != nil {
		t.Fatal(err)
	}
	for _, sub := range []string{"mesh.assignment.g14.a", "mesh.assignment.ace.b"} {
		if _, err := ajs.Publish(ctx, sub, []byte("{}")); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := ajs.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.events.>"}}); err != nil {
		t.Fatal(err)
	}
	for _, c := range []string{"controller-merges", "mesh-grants"} {
		if _, err := ajs.CreateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: c, AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
			t.Fatal(err)
		}
	}
	rollouts, err := ajs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_rollouts"})
	if err != nil {
		t.Fatal(err)
	}
	rollouts.Put(ctx, "p-1", []byte("{}"))
	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)
	}

	// Research 072 row 33: the stream ASSIGNMENTS goes whole, once; read first, and asking again is no error; a stream
	// outside the grant is not touched.
	if _, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS"}); err == nil || !strings.Contains(err.Error(), "why is required") {
		t.Fatalf("a stream without why: %v", err)
	}
	if _, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "why": "x"}); err == nil || !strings.Contains(err.Error(), "grant") {
		t.Fatalf("a stream outside the grant: %v", err)
	}
	if d, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS", "why": "look", "dry_run": true}); err != nil || d.(StreamRemoved).Messages != 2 || d.(StreamRemoved).Removed {
		t.Fatalf("a dry run: %+v %v", d, err)
	}
	if _, err := ajs.Stream(ctx, "ASSIGNMENTS"); err != nil {
		t.Fatalf("a dry run removed the stream: %v", err)
	}
	if r, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS", "why": "the lease write"}); err != nil || !r.(StreamRemoved).Removed || r.(StreamRemoved).Messages != 2 {
		t.Fatalf("the removal: %+v %v", r, err)
	}
	if _, err := ajs.Stream(ctx, "ASSIGNMENTS"); err == nil {
		t.Fatal("ASSIGNMENTS is still there")
	}
	if r, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS", "why": "again"}); err != nil || r.(StreamRemoved).Removed || r.(StreamRemoved).Present {
		t.Fatalf("a second removal: %+v %v", r, err)
	}

	// Issue 1126 B5: the controller's consumer of EVENTS goes and the stream's other consumer stays; its bucket's stream
	// goes whole; EVENTS itself is outside the grant.
	if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "EVENTS", "consumer": "mesh-grants", "why": "x"}); err == nil {
		t.Fatal("a consumer of EVENTS outside the grant was removed")
	}
	if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "EVENTS", "consumer": "controller-merges", "why": "the controller ended"}); err != nil {
		t.Fatalf("the controller's consumer of EVENTS: %v", err)
	}
	if _, err := ajs.Consumer(ctx, "EVENTS", "mesh-grants"); err != nil {
		t.Fatalf("the other consumer of EVENTS went: %v", err)
	}
	if r, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "KV_mesh-controller_rollouts", "why": "the controller ended"}); err != nil || !r.(StreamRemoved).Removed || r.(StreamRemoved).Messages != 1 {
		t.Fatalf("the controller's bucket: %+v %v", r, err)
	}
	if _, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "EVENTS", "why": "x"}); err == nil || !strings.Contains(err.Error(), "consumers") && !strings.Contains(err.Error(), "grant") {
		t.Fatalf("EVENTS: %v", err)
	}
	if _, err := ajs.Stream(ctx, "EVENTS"); err != nil {
		t.Fatalf("EVENTS went: %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,stream-removed,consumer-removed,stream-removed,bucket-keys-removed" {
		t.Fatalf("the audit events said: %v", said)
	}
}