diff --git a/modules/nats/cmd/mesh-nats-snapshot/connect.go b/modules/nats/cmd/mesh-nats-snapshot/connect.go index 97aad5b7..246a34fb 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/connect.go +++ b/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..>` 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") } diff --git a/modules/nats/cmd/nats-tools/bus.go b/modules/nats/cmd/nats-tools/bus.go new file mode 100644 index 00000000..847a7cd8 --- /dev/null +++ b/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) } diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 4de769dd..4dcac6d8 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/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) } diff --git a/modules/nats/cmd/nats-tools/manifest_test.go b/modules/nats/cmd/nats-tools/manifest_test.go index b9d699d0..a9ade07e 100644 --- a/modules/nats/cmd/nats-tools/manifest_test.go +++ b/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) diff --git a/modules/nats/cmd/nats-tools/removal.go b/modules/nats/cmd/nats-tools/removal.go new file mode 100644 index 00000000..633afccd --- /dev/null +++ b/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 `.`) 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 `.`) 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 (`.`) 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)}, + } +} diff --git a/modules/nats/cmd/nats-tools/removal_test.go b/modules/nats/cmd/nats-tools/removal_test.go new file mode 100644 index 00000000..2ed7799d --- /dev/null +++ b/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) + } +} diff --git a/modules/nats/cmd/nats-tools/tools_test.go b/modules/nats/cmd/nats-tools/tools_test.go index bd0e7d0f..3e989242 100644 --- a/modules/nats/cmd/nats-tools/tools_test.go +++ b/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) + } + } +} diff --git a/modules/nats/internal/busconn/busconn.go b/modules/nats/internal/busconn/busconn.go new file mode 100644 index 00000000..09ada3c3 --- /dev/null +++ b/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..>` 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 + }, + } +} diff --git a/modules/nats/module.json b/modules/nats/module.json index 106aac92..45f83a65 100644 --- a/modules/nats/module.json +++ b/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" } }, {