nats: tools that remove one consumer and one node's keys from a bucket, with why and an audit event (issue 1952 N1-N3)

modules/nats/cmd/mesh-nats-snapshot/connect.go

@@ -1,105 +1,16 @@
 package main
 
 import (
-	"crypto/sha256"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/hex"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"io"
-	"os"
-	"strings"
-	"time"
+	"nats-tools/internal/busconn"
 
 	"github.com/nats-io/nats.go"
 )
 
-// Credential is what the mesh seals to a module so it can reach the bus (mesh-controller
-// cmd/mesh-controller issueWith): where, the server's certificate by fingerprint, who, and the
-// password. The address is not used here — this runs beside the server, and the server is reached on
-// loopback — but the fingerprint is: loopback or not, the certificate is the mesh's own.
-type Credential struct {
-	URL         string `json:"url"`
-	Fingerprint string `json:"fingerprint,omitempty"`
-	User        string `json:"user"`
-	Password    string `json:"password"`
-}
+// Credential is what the mesh seals to a module so it can reach the bus; see busconn.
+type Credential = busconn.Credential
 
-func readCredential(path string) (Credential, error) {
-	var raw []byte
-	var err error
-	if path == "-" {
-		raw, err = io.ReadAll(io.LimitReader(os.Stdin, 1<<20))
-	} else {
-		raw, err = os.ReadFile(path)
-	}
-	if err != nil {
-		return Credential{}, fmt.Errorf("cannot read the credential: %w", err)
-	}
-	var c Credential
-	if err := json.Unmarshal(raw, &c); err != nil {
-		return Credential{}, errors.New("the credential is not the JSON the mesh seals to a module (url, fingerprint, user, password)")
-	}
-	if c.User == "" || c.Password == "" {
-		return Credential{}, errors.New("the credential names no user or password, so the bus would refuse it")
-	}
-	return c, nil
-}
+func readCredential(path string) (Credential, error) { return busconn.Read(path) }
 
-// connect dials the bus as the credential's user. A `tls://` server is held to the credential's
-// fingerprint and nothing else — the mesh's bus presents a certificate of its own, in no trust store,
-// so a name check could only fail or be skipped. A plain `nats://` server is accepted only on
-// loopback, and is what a restore's own private server is.
 func connect(server string, c Credential) (*nats.Conn, error) {
-	opts := []nats.Option{
-		nats.UserInfo(c.User, c.Password),
-		nats.Name("mesh-nats-snapshot"),
-		// Answers come to the user's own inbox and no other: the bus grants every user its own
-		// `_INBOX.<user>.>` and nothing wider (novox/hq design 25 §4).
-		nats.CustomInboxPrefix("_INBOX." + c.User),
-		nats.Timeout(10 * time.Second),
-		// One attempt: a snapshot that cannot reach the bus fails the night, loudly, rather than
-		// waiting for it.
-		nats.NoReconnect(),
-	}
-	switch {
-	case strings.HasPrefix(server, "tls://"):
-		if c.Fingerprint == "" {
-			return nil, errors.New("the credential carries no fingerprint, so the bus's certificate cannot be checked")
-		}
-		opts = append(opts, nats.Secure(pinnedTo(c.Fingerprint)))
-	case strings.HasPrefix(server, "nats://127.0.0.1:"), strings.HasPrefix(server, "nats://localhost:"):
-		if c.Fingerprint != "" {
-			// The mesh's bus requires TLS; a plain connection to it would be refused anyway.
-			opts = append(opts, nats.Secure(pinnedTo(c.Fingerprint)))
-		}
-	default:
-		return nil, fmt.Errorf("%s: the bus is reached over tls:// (or plain nats:// on loopback, for a restore's own server)", server)
-	}
-	nc, err := nats.Connect(server, opts...)
-	if err != nil {
-		return nil, fmt.Errorf("cannot reach the bus at %s as %s: %w", server, c.User, err)
-	}
-	return nc, nil
-}
-
-// pinnedTo accepts exactly the certificate with this SHA-256 — the same pin every machine holds
-// (mesh-controller internal/broker PinnedToFingerprint).
-func pinnedTo(want string) *tls.Config {
-	return &tls.Config{
-		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
-		MinVersion:         tls.VersionTLS12,
-		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
-			if len(rawCerts) == 0 {
-				return errors.New("the bus presented no certificate")
-			}
-			sum := sha256.Sum256(rawCerts[0])
-			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
-				return fmt.Errorf("the bus presented a certificate this mesh does not know (%s)", got)
-			}
-			return nil
-		},
-	}
+	return busconn.Connect(server, c, "mesh-nats-snapshot")
 }

modules/nats/cmd/nats-tools/bus.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"github.com/nats-io/nats.go/jetstream"
+	"nats-tools/internal/busconn"
+)
+
+// LiveBus is the bus as the module's own credential reaches it. Nothing connects until a tool that changes the
+// bus is called, so the tools that read (through the monitoring endpoint) answer as they did with no credential.
+// MESH_BROKER_FILE names the credential (the manifest's own secret `broker`); MESH_NATS_SERVER where the bus is as
+// seen from this machine (default: its own port on loopback).
+func LiveBus() Bus {
+	return func(ctx context.Context) (jetstream.JetStream, func(), error) {
+		file := env("MESH_BROKER_FILE", "")
+		if file == "" {
+			return nil, nil, errors.New("MESH_BROKER_FILE is not set, so this tool has no credential to reach the bus with (the module's manifest sets it)")
+		}
+		c, err := busconn.Read(file)
+		if err != nil {
+			return nil, nil, err
+		}
+		nc, err := busconn.Connect(env("MESH_NATS_SERVER", "tls://127.0.0.1:4222"), c, "nats-tools")
+		if err != nil {
+			return nil, nil, err
+		}
+		js, err := jetstream.New(nc)
+		if err != nil {
+			nc.Close()
+			return nil, nil, fmt.Errorf("JetStream on the bus: %w", err)
+		}
+		return js, nc.Close, nil
+	}
+}
+
+// SaidByTheRuntime is the audit line: an event of this module, published through the runtime that launched it.
+func SaidByTheRuntime(event string, body any) error { return stdio.Emit(event, body) }

modules/nats/cmd/nats-tools/main.go

@@ -1,6 +1,7 @@
 // nats-tools: the mesh bus's own tools (novox/hq design 25), a Go bundle the node's runtime launches on the
-// machine that runs the bus. Everything here reads — the server's monitoring API and the user list the
-// controller composed — and nothing changes the bus. stdout is the MCP channel; this says nothing else.
+// machine that runs the bus. Almost everything here reads — the server's monitoring API and the user list the
+// controller composed. The two that change the bus (removal.go) are the only ones that connect to it, on the
+// module's own credential, and say what they did. stdout is the MCP channel; this says nothing else.
 package main
 
 import (
@@ -151,7 +152,7 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written
 }
 
 func main() {
-	if err := stdio.Serve("", tools(LiveMonitor(), ReadUsers, UsersWritten)); err != nil {
+	if err := stdio.Serve("", append(tools(LiveMonitor(), ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime)...)); err != nil {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(1)
 	}

modules/nats/cmd/nats-tools/manifest_test.go

@@ -3,6 +3,7 @@ package main
 import (
 	"encoding/json"
 	"os"
+	"slices"
 	"strings"
 	"testing"
 )
@@ -24,11 +25,14 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 				} `json:"backup"`
 			} `json:"own"`
 		} `json:"data"`
+		Tools     []string         `json:"tools"`
+		Emits     []string         `json:"emits"`
 		Resources []map[string]any `json:"resources"`
 		Build     struct {
 			On        any `json:"on"`
 			Artifacts []struct {
 				Name, Kind, From, Language, System, Binary string
+				Env                                        map[string]string
 			} `json:"artifacts"`
 		} `json:"build"`
 	}
@@ -38,7 +42,10 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 	if m.Build.On != nil {
 		t.Error("build.on is declared: nothing is built on a base image any more")
 	}
-	artifacts := map[string]struct{ Name, Kind, From, Language, System, Binary string }{}
+	artifacts := map[string]struct {
+		Name, Kind, From, Language, System, Binary string
+		Env                                        map[string]string
+	}{}
 	for _, a := range m.Build.Artifacts {
 		artifacts[a.Name] = a
 		if a.Kind == "image" {
@@ -103,6 +110,23 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 			t.Errorf("%s is not in the container's restart-on: a new program would leave the old one running", p.resource)
 		}
 	}
+	// The tools that change the bus (issue 1952 N1-N3): listed, their audit events declared, and the credential
+	// they dial with the module's own secret `broker`.
+	for _, want := range []string{"nats_consumer_remove", "nats_bucket_remove"} {
+		if !slices.Contains(m.Tools, want) {
+			t.Errorf("tools do not list %s", want)
+		}
+	}
+	for _, want := range []string{"consumer-removed", "bucket-keys-removed"} {
+		if !slices.Contains(m.Emits, want) {
+			t.Errorf("emits do not list %s: the removal tools say it as the audit line", want)
+		}
+	}
+	for _, a := range m.Build.Artifacts {
+		if a.Name == "tools" && a.Env["MESH_BROKER_FILE"] != "${dir:mesh-state}/broker" {
+			t.Errorf("the tools' credential is %q: the module's own secret broker", a.Env["MESH_BROKER_FILE"])
+		}
+	}
 	args, _ := server["args"].([]any)
 	if len(args) != 2 || args[0] != "sh" || args[1] != "/opt/mesh-entrypoint/entrypoint.sh" {
 		t.Errorf("args are %v: the entrypoint run through sh, so no execute bit is relied on", args)

modules/nats/cmd/nats-tools/removal.go

@@ -0,0 +1,246 @@
+package main
+
+// The two tools of this module that change the bus (novox/hq ADR 0564 clause 5, issue 1952 notes 6, 7 and 9):
+// remove one named consumer, and remove one node's keys under a prefix of a key-value bucket. Everything else
+// here reads.
+//
+// **Narrow by what the tool needs, and narrower by what the bus grants.** The tools reach the bus on the
+// credential the module's manifest names (MESH_BROKER_FILE); the bus refuses whatever that credential is not
+// granted, so a wrong stream, consumer or bucket is refused by the server and said here, not trusted to the
+// arguments. Each tool refuses without `why`, reads what it would remove first (and says it, so a person can ask
+// with dry_run and then for real), removes, reads that it is gone, and says what it did as an event of the module
+// (the audit trail: who asked is in the call the controller audits, why and what are in the event).
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"regexp"
+	"sort"
+	"strings"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Bus opens JetStream on the bus as the module's credential for one call; close ends the connection.
+type Bus func(ctx context.Context) (js jetstream.JetStream, close func(), err error)
+
+// Said keeps the audit line: an event of this module.
+type Said func(event string, body any) error
+
+const (
+	consumerRemoved   = "consumer-removed"
+	bucketKeysRemoved = "bucket-keys-removed"
+	// maxKeysShown bounds the keys an answer lists; the count is always whole.
+	maxKeysShown = 200
+)
+
+var (
+	// A stream, consumer or bucket is one subject token: no dot, no wildcard, no space.
+	name = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
+	// A key prefix is subject tokens and their dots; no wildcard, so it names keys and not a pattern.
+	keyPrefix = regexp.MustCompile(`^[A-Za-z0-9_./=-]+$`)
+)
+
+func why(a map[string]any) (string, error) {
+	w := strings.TrimSpace(arg(a, "why"))
+	if w == "" {
+		return "", errors.New("why is required: say why this is removed, for the audit trail (novox/hq ADR 0564 clause 5); nothing was removed")
+	}
+	return w, nil
+}
+
+// answerWithin is how long the bus is given to answer one request. **The bus does not refuse a request its
+// credential is not granted: it drops it** (a permissions violation is told to the connection's error handler,
+// never answered), so an unanswered request is how a refusal shows here.
+var answerWithin = 10 * time.Second
+
+// unanswered says what a request with no answer most likely is.
+func unanswered(err error, what string) error {
+	if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) {
+		return fmt.Errorf("%s: the bus did not answer within %s; it drops, without an answer, what the module's credential is not granted, "+
+			"so this is most likely outside its grant (or the bus is not reachable): %w", what, answerWithin, err)
+	}
+	return fmt.Errorf("%s: %w", what, err)
+}
+
+func step(ctx context.Context) (context.Context, context.CancelFunc) {
+	return context.WithTimeout(ctx, answerWithin)
+}
+
+// ConsumerRemoved is what nats_consumer_remove answers and says.
+type ConsumerRemoved struct {
+	Stream   string         `json:"stream"`
+	Consumer string         `json:"consumer"`
+	Why      string         `json:"why"`
+	Was      map[string]any `json:"was"`
+	Removed  bool           `json:"removed"`
+	At       time.Time      `json:"at"`
+}
+
+// RemoveConsumer removes one named consumer of one stream, after reading it, and says so.
+func RemoveConsumer(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
+	w, err := why(a)
+	if err != nil {
+		return nil, err
+	}
+	stream, consumer := arg(a, "stream"), arg(a, "consumer")
+	if !name.MatchString(stream) || !name.MatchString(consumer) {
+		return nil, errors.New("stream and consumer are required, each one name (letters, digits, - and _): nothing was removed")
+	}
+	js, closeBus, err := bus(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer closeBus()
+	sctx, cancel := step(ctx)
+	c, err := js.Consumer(sctx, stream, consumer)
+	cancel()
+	if err != nil {
+		if errors.Is(err, jetstream.ErrConsumerNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) {
+			return nil, fmt.Errorf("%s has no consumer %s (%v): nothing was removed", stream, consumer, err)
+		}
+		return nil, unanswered(err, fmt.Sprintf("the consumer %s of %s could not be read, so it was not removed", consumer, stream))
+	}
+	info := c.CachedInfo()
+	was := map[string]any{
+		"durable": info.Config.Durable, "filter_subject": info.Config.FilterSubject, "filter_subjects": info.Config.FilterSubjects,
+		"deliver_subject": info.Config.DeliverSubject, "pending": info.NumPending, "ack_pending": info.NumAckPending,
+		"redelivered": info.NumRedelivered, "created": info.Created,
+	}
+	sctx, cancel = step(ctx)
+	err = js.DeleteConsumer(sctx, stream, consumer)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("consumer %s of %s was not removed", consumer, stream))
+	}
+	sctx, cancel = step(ctx)
+	_, err = js.Consumer(sctx, stream, consumer)
+	cancel()
+	if err == nil {
+		return nil, fmt.Errorf("consumer %s of %s is still there after the removal was accepted", consumer, stream)
+	}
+	out := ConsumerRemoved{Stream: stream, Consumer: consumer, Why: w, Was: was, Removed: true, At: time.Now().UTC()}
+	if err := said(consumerRemoved, out); err != nil {
+		return nil, fmt.Errorf("consumer %s of %s WAS removed (%s) but the audit event could not be said: %w", consumer, stream, w, err)
+	}
+	return out, nil
+}
+
+// KeysRemoved is what nats_bucket_remove answers and says.
+type KeysRemoved struct {
+	Bucket  string    `json:"bucket"`
+	Prefix  string    `json:"prefix"`
+	Why     string    `json:"why"`
+	DryRun  bool      `json:"dry_run"`
+	Count   int       `json:"count"`
+	Keys    []string  `json:"keys"`
+	Cut     bool      `json:"keys_cut,omitempty"`
+	Removed int       `json:"removed"`
+	At      time.Time `json:"at"`
+}
+
+// RemoveKeys removes the keys of a key-value bucket under a prefix: counted first, and only those.
+func RemoveKeys(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
+	w, err := why(a)
+	if err != nil {
+		return nil, err
+	}
+	bucket, node, prefix := arg(a, "bucket"), arg(a, "node"), arg(a, "prefix")
+	if !name.MatchString(bucket) {
+		return nil, errors.New("bucket is required, a bucket's name as nats_buckets lists it: nothing was removed")
+	}
+	if (node == "") == (prefix == "") {
+		return nil, errors.New("give exactly one of node (that node's keys: those starting `<node>.`) or prefix (keys starting so): " +
+			"a removal of a whole bucket is not this tool's; nothing was removed")
+	}
+	if node != "" {
+		if !name.MatchString(node) {
+			return nil, errors.New("node is one machine's name: nothing was removed")
+		}
+		prefix = node + "."
+	}
+	if !keyPrefix.MatchString(prefix) {
+		return nil, errors.New("prefix is key text without wildcards or spaces: nothing was removed")
+	}
+	dry, _ := a["dry_run"].(bool)
+	js, closeBus, err := bus(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer closeBus()
+	sctx, cancel := step(ctx)
+	kv, err := js.KeyValue(sctx, bucket)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("bucket %s could not be opened, nothing was removed", bucket))
+	}
+	sctx, cancel = step(ctx)
+	stream, err := js.Stream(sctx, "KV_"+bucket)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("bucket %s has no stream to read its keys from", bucket))
+	}
+	// The server narrows to the prefix's tokens when the prefix ends on a token; any other prefix is cut here.
+	filter := "$KV." + bucket + ".>"
+	if strings.HasSuffix(prefix, ".") {
+		filter = "$KV." + bucket + "." + prefix + ">"
+	}
+	sctx, cancel = step(ctx)
+	info, err := stream.Info(sctx, jetstream.WithSubjectFilter(filter))
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("the keys of %s could not be read, so none were removed", bucket))
+	}
+	var keys []string
+	for subject := range info.State.Subjects {
+		if key := strings.TrimPrefix(subject, "$KV."+bucket+"."); strings.HasPrefix(key, prefix) {
+			keys = append(keys, key)
+		}
+	}
+	sort.Strings(keys)
+	out := KeysRemoved{Bucket: bucket, Prefix: prefix, Why: w, DryRun: dry, Count: len(keys), Keys: keys, At: time.Now().UTC()}
+	if len(keys) > maxKeysShown {
+		out.Keys, out.Cut = keys[:maxKeysShown], true
+	}
+	if want, ok := a["expect"].(float64); ok && int(want) != len(keys) {
+		return nil, fmt.Errorf("%d keys of %s start with %q, not the %d expected: nothing was removed", len(keys), bucket, prefix, int(want))
+	}
+	if dry || len(keys) == 0 {
+		return out, nil
+	}
+	for _, k := range keys {
+		sctx, cancel := step(ctx)
+		err := kv.Purge(sctx, k)
+		cancel()
+		if err != nil {
+			return nil, unanswered(err, fmt.Sprintf("removing %q from %s failed after %d of %d keys, the rest were left", k, bucket, out.Removed, len(keys)))
+		}
+		out.Removed++
+	}
+	if err := said(bucketKeysRemoved, out); err != nil {
+		return nil, fmt.Errorf("%d keys of %s under %q WERE removed (%s) but the audit event could not be said: %w", out.Removed, bucket, prefix, w, err)
+	}
+	return out, nil
+}
+
+// removalTools are the module's two tools that change the bus; every other tool here reads.
+func removalTools(bus Bus, said Said) []stdio.Tool {
+	run := func(f func(context.Context, Bus, Said, map[string]any) (any, error)) func(map[string]any) (any, error) {
+		return func(a map[string]any) (any, error) { return f(context.Background(), bus, said, a) }
+	}
+	return []stdio.Tool{
+		{Name: "nats_consumer_remove",
+			Description: "Removes (deletes) one named JetStream consumer of one stream, e.g. the consumer `controller` of DEAD_LETTER_NOTICES at the lease write (novox/hq ADR 0564 clause 5). Needs `why`, kept in the audit trail; refuses an unknown stream or consumer; reads the consumer first and answers what it was (filters, pending, unacknowledged); the stream and its other consumers are not touched. The bus refuses any consumer the module's credential is not granted.",
+			Input:       map[string]any{"stream": str("the stream, e.g. DEAD_LETTER_NOTICES"), "consumer": str("the consumer's name, e.g. controller"), "why": str("why it is removed, for the audit trail (required)")},
+			Run:         run(RemoveConsumer)},
+		{Name: "nats_bucket_remove",
+			Description: "Removes (purges) keys from a key-value bucket: one node's keys (`node`, the keys starting `<node>.`) or the keys under a `prefix`, e.g. a full acts bucket of the node-engine. Needs `why`, kept in the audit trail; never the whole bucket. Counts the keys first and lists them (`dry_run` only counts; `expect` refuses when the count differs); removes only those keys, and their history. See the buckets and their keys with nats_buckets. The bus refuses any bucket the module's credential is not granted.",
+			Input: map[string]any{"bucket": str("the bucket as nats_buckets lists it, e.g. mesh-host_acts"), "node": str("a node whose keys (`<node>.`) go"), "prefix": str("key prefix instead of node"),
+				"why": str("why they are removed, for the audit trail (required)"), "dry_run": map[string]any{"type": "boolean", "description": "count and list the keys, remove none"}, "expect": num("the count of keys expected; refuses when it differs")},
+			Run: run(RemoveKeys)},
+	}
+}

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

@@ -0,0 +1,173 @@
+//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)
+	}
+}

modules/nats/cmd/nats-tools/tools_test.go

@@ -1,12 +1,15 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
 )
 
 const connz = `{"connections":[
@@ -208,3 +211,47 @@ func TestAGrantIsInForceOnlyWhenTheServerReloadedAfterTheListWasWritten(t *testi
 		t.Error("a write time that is not one was read")
 	}
 }
+
+// **The tools that change the bus are two, each refuses before it reaches the bus, and every other tool reads**
+// (issue 1952 N1-N3). A row per refusal: the bus is never opened for them.
+func TestTheRemovalToolsRefuseBeforeTheyReachTheBus(t *testing.T) {
+	bus := func(context.Context) (jetstream.JetStream, func(), error) {
+		t.Fatal("the bus was opened for a call that must be refused")
+		return nil, nil, nil
+	}
+	said := func(string, any) error { t.Fatal("an audit event was said for a refused call"); return nil }
+	for _, c := range []struct {
+		tool string
+		args map[string]any
+		want string
+	}{
+		{"nats_consumer_remove", map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller"}, "why is required"},
+		{"nats_consumer_remove", map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller", "why": "  "}, "why is required"},
+		{"nats_consumer_remove", map[string]any{"stream": "A.B", "consumer": "c", "why": "x"}, "stream and consumer are required"},
+		{"nats_consumer_remove", map[string]any{"stream": "S", "consumer": ">", "why": "x"}, "stream and consumer are required"},
+		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "node": "g14"}, "why is required"},
+		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "why": "x"}, "exactly one of node"},
+		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "node": "g14", "prefix": "g", "why": "x"}, "exactly one of node"},
+		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "prefix": "g*", "why": "x"}, "without wildcards"},
+		{"nats_bucket_remove", map[string]any{"bucket": "", "node": "g14", "why": "x"}, "bucket is required"},
+	} {
+		var run func(map[string]any) (any, error)
+		for _, tl := range removalTools(bus, said) {
+			if tl.Name == c.tool {
+				run = tl.Run
+			}
+		}
+		if run == nil {
+			t.Fatalf("no tool %s", c.tool)
+		}
+		if _, err := run(c.args); err == nil || !strings.Contains(err.Error(), c.want) {
+			t.Errorf("%s %v: %v, want a refusal saying %q", c.tool, c.args, err, c.want)
+		}
+	}
+	// The read tools stay read tools: none of the first ten takes a `why`, and the list is these ten.
+	for _, tl := range tools(nil, nil, nil) {
+		if _, ok := tl.Input["why"]; ok || strings.Contains(tl.Name, "remove") {
+			t.Errorf("%s changes the bus; only the removal tools may", tl.Name)
+		}
+	}
+}

modules/nats/internal/busconn/busconn.go

@@ -0,0 +1,106 @@
+// Package busconn dials the bus as a credential the mesh sealed to a module: the one place the module's
+// programs learn how (the snapshot, and the tools that change the bus).
+package busconn
+
+import (
+	"crypto/sha256"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// Credential is what the mesh seals to a module so it can reach the bus (mesh-controller
+// cmd/mesh-controller issueWith): where, the server's certificate by fingerprint, who, and the
+// password. The address is not used here — this runs beside the server, and the server is reached on
+// loopback — but the fingerprint is: loopback or not, the certificate is the mesh's own.
+type Credential struct {
+	URL         string `json:"url"`
+	Fingerprint string `json:"fingerprint,omitempty"`
+	User        string `json:"user"`
+	Password    string `json:"password"`
+}
+
+func Read(path string) (Credential, error) {
+	var raw []byte
+	var err error
+	if path == "-" {
+		raw, err = io.ReadAll(io.LimitReader(os.Stdin, 1<<20))
+	} else {
+		raw, err = os.ReadFile(path)
+	}
+	if err != nil {
+		return Credential{}, fmt.Errorf("cannot read the credential: %w", err)
+	}
+	var c Credential
+	if err := json.Unmarshal(raw, &c); err != nil {
+		return Credential{}, errors.New("the credential is not the JSON the mesh seals to a module (url, fingerprint, user, password)")
+	}
+	if c.User == "" || c.Password == "" {
+		return Credential{}, errors.New("the credential names no user or password, so the bus would refuse it")
+	}
+	return c, nil
+}
+
+// Connect dials the bus as the credential's user. A `tls://` server is held to the credential's
+// fingerprint and nothing else — the mesh's bus presents a certificate of its own, in no trust store,
+// so a name check could only fail or be skipped. A plain `nats://` server is accepted only on
+// loopback, and is what a restore's own private server is.
+func Connect(server string, c Credential, name string) (*nats.Conn, error) {
+	opts := []nats.Option{
+		nats.UserInfo(c.User, c.Password),
+		nats.Name(name),
+		// Answers come to the user's own inbox and no other: the bus grants every user its own
+		// `_INBOX.<user>.>` and nothing wider (novox/hq design 25 §4).
+		nats.CustomInboxPrefix("_INBOX." + c.User),
+		nats.Timeout(10 * time.Second),
+		// One attempt: a program that cannot reach the bus fails, loudly, rather than waiting for it.
+		nats.NoReconnect(),
+	}
+	switch {
+	case strings.HasPrefix(server, "tls://"):
+		if c.Fingerprint == "" {
+			return nil, errors.New("the credential carries no fingerprint, so the bus's certificate cannot be checked")
+		}
+		opts = append(opts, nats.Secure(PinnedTo(c.Fingerprint)))
+	case strings.HasPrefix(server, "nats://127.0.0.1:"), strings.HasPrefix(server, "nats://localhost:"):
+		if c.Fingerprint != "" {
+			// The mesh's bus requires TLS; a plain connection to it would be refused anyway.
+			opts = append(opts, nats.Secure(PinnedTo(c.Fingerprint)))
+		}
+	default:
+		return nil, fmt.Errorf("%s: the bus is reached over tls:// (or plain nats:// on loopback, for a restore's own server)", server)
+	}
+	nc, err := nats.Connect(server, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("cannot reach the bus at %s as %s: %w", server, c.User, err)
+	}
+	return nc, nil
+}
+
+// PinnedTo accepts exactly the certificate with this SHA-256 — the same pin every machine holds
+// (mesh-controller internal/broker PinnedToFingerprint).
+func PinnedTo(want string) *tls.Config {
+	return &tls.Config{
+		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
+		MinVersion:         tls.VersionTLS12,
+		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
+			if len(rawCerts) == 0 {
+				return errors.New("the bus presented no certificate")
+			}
+			sum := sha256.Sum256(rawCerts[0])
+			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
+				return fmt.Errorf("the bus presented a certificate this mesh does not know (%s)", got)
+			}
+			return nil
+		},
+	}
+}

modules/nats/module.json

@@ -24,7 +24,10 @@
   "capabilities": [
     "container-runtime"
   ],
-  "emits": [],
+  "emits": [
+    "consumer-removed",
+    "bucket-keys-removed"
+  ],
   "consumes": [],
   "listens": [
     {
@@ -45,7 +48,9 @@
     "nats_buckets",
     "nats_users",
     "nats_user_can",
-    "nats_snapshot_verify"
+    "nats_snapshot_verify",
+    "nats_consumer_remove",
+    "nats_bucket_remove"
   ],
   "guards": [
     8222
@@ -194,7 +199,8 @@
           "MESH_NATS_MONITOR": "http://127.0.0.1:8222",
           "MESH_NATS_CONTAINER": "mesh-broker-nats",
           "MESH_NATS_USERS_FILE": "/etc/nats/accounts.conf",
-          "MESH_NATS_SNAPSHOTS_DIR": "${dir:snapshots}"
+          "MESH_NATS_SNAPSHOTS_DIR": "${dir:snapshots}",
+          "MESH_BROKER_FILE": "${dir:mesh-state}/broker"
         }
       },
       {