diff --git a/checks/credentialform/go.mod b/checks/credentialform/go.mod new file mode 100644 index 00000000..4172035d --- /dev/null +++ b/checks/credentialform/go.mod @@ -0,0 +1,5 @@ +module checks/credentialform + +go 1.26.0 + +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a diff --git a/checks/credentialform/go.sum b/checks/credentialform/go.sum new file mode 100644 index 00000000..ea2abb0e --- /dev/null +++ b/checks/credentialform/go.sum @@ -0,0 +1,2 @@ +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a h1:elC2IjB+qbeXG+pWItLz71DenegbrdsqbcH5kyGo56Q= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= diff --git a/checks/credentialform/main.go b/checks/credentialform/main.go new file mode 100644 index 00000000..836dfda4 --- /dev/null +++ b/checks/credentialform/main.go @@ -0,0 +1,36 @@ +// Package credentialform refuses an offer that does not say the form of its credential: an own secret, an issuer, or +// `form` pair or none (the lead, 2026-10-29). It reads every modules/*/module.json through mesh-sdk's +// CredentialFormProblems. +package credentialform + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// Problems is every offer of the manifests under root/modules that says no credential form, one line each. +func Problems(root string) ([]string, error) { + files, err := filepath.Glob(filepath.Join(root, "modules", "*", "module.json")) + if err != nil { + return nil, err + } + sort.Strings(files) + var out []string + for _, f := range files { + raw, err := os.ReadFile(f) + if err != nil { + return nil, err + } + var m catalogue.Manifest + if err := json.Unmarshal(raw, &m); err != nil { + return nil, fmt.Errorf("%s: %w", f, err) + } + out = append(out, catalogue.CredentialFormProblems(m)...) + } + return out, nil +} diff --git a/checks/credentialform/main_test.go b/checks/credentialform/main_test.go new file mode 100644 index 00000000..0038a018 --- /dev/null +++ b/checks/credentialform/main_test.go @@ -0,0 +1,46 @@ +package credentialform + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func aCatalogue(t *testing.T, manifests map[string]string) string { + t.Helper() + root := t.TempDir() + for module, body := range manifests { + dir := filepath.Join(root, "modules", module) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "module.json"), []byte(body), 0o644); err != nil { + t.Fatal(err) + } + } + return root +} + +// An offer that says no form is refused and named; one that says pair, none or an own secret passes; and the real +// catalogue is clean. +func TestEveryOfferSaysTheFormOfItsCredential(t *testing.T) { + root := aCatalogue(t, map[string]string{ + "silent": `{"module":"silent","provides":[{"name":"thing","scope":"mesh"}]}`, + "says": `{"module":"says","provides":[{"name":"a","credential":{"form":"pair"}},{"name":"b","credential":{"form":"none"}},{"name":"c","credential":{"own":"jwt"}}]}`, + }) + got, err := Problems(root) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 || !strings.Contains(got[0], `silent provides "thing"`) { + t.Fatalf("refused %v, wanted the one silent offer", got) + } + real, err := Problems(filepath.Join("..", "..")) + if err != nil { + t.Fatal(err) + } + if len(real) != 0 { + t.Fatalf("the catalogue has offers with no credential form:\n%s", strings.Join(real, "\n")) + } +} diff --git a/checks/credentialform/reads-outside b/checks/credentialform/reads-outside new file mode 100644 index 00000000..8ff16290 --- /dev/null +++ b/checks/credentialform/reads-outside @@ -0,0 +1,3 @@ +# What these tests read outside their package, as paths from the repository's root (issue 1936). +# every manifest, for what each offer says of its credential's form. +modules/*/module.json diff --git a/modules/buildah/module.json b/modules/buildah/module.json index 70ddd0a4..bdfbe96b 100644 --- a/modules/buildah/module.json +++ b/modules/buildah/module.json @@ -8,7 +8,10 @@ "provides": [ { "name": "rootless-image-build", - "reach": "machine" + "reach": "machine", + "credential": { + "form": "none" + } } ], "resources": [ diff --git a/modules/distribution/module.json b/modules/distribution/module.json index eb9f7c24..5037f429 100644 --- a/modules/distribution/module.json +++ b/modules/distribution/module.json @@ -4,7 +4,10 @@ "provides": [ { "name": "artifact-store", - "scope": "mesh" + "scope": "mesh", + "credential": { + "form": "none" + } } ], "claims": [ diff --git a/modules/dnsmasq/module.json b/modules/dnsmasq/module.json index 58cc0da5..7e832067 100644 --- a/modules/dnsmasq/module.json +++ b/modules/dnsmasq/module.json @@ -9,7 +9,10 @@ { "name": "wildcard-resolution", "scope": "mesh", - "identity": false + "identity": false, + "credential": { + "form": "none" + } } ], "requires": [ diff --git a/modules/go/module.json b/modules/go/module.json index 48a6bf57..9783eda4 100644 --- a/modules/go/module.json +++ b/modules/go/module.json @@ -7,7 +7,10 @@ "provides": [ { "name": "go-toolchain", - "reach": "machine" + "reach": "machine", + "credential": { + "form": "none" + } } ], "resources": [ diff --git a/modules/influxdb/module.json b/modules/influxdb/module.json index 53389a5f..13a7905f 100644 --- a/modules/influxdb/module.json +++ b/modules/influxdb/module.json @@ -7,6 +7,9 @@ "scope": "mesh", "identity": { "in": "an InfluxDB v1 authorization" + }, + "credential": { + "form": "pair" } } ], diff --git a/modules/mailu/module.json b/modules/mailu/module.json index 93d72adb..0faccc58 100644 --- a/modules/mailu/module.json +++ b/modules/mailu/module.json @@ -665,7 +665,14 @@ "path": "${dir:mesh-state}/config.json", "mode": "0600", "template": "files/config.json", - "takes-config": ["domain", "proxy-address", "sitename", "website", "url", "container"], + "takes-config": [ + "domain", + "proxy-address", + "sitename", + "website", + "url", + "container" + ], "merge": "json" }, { @@ -758,6 +765,9 @@ "identity": { "max": 64, "in": "a mailbox's local part" + }, + "credential": { + "form": "pair" } } ], diff --git a/modules/mesh-build-go/go.mod b/modules/mesh-build-go/go.mod index 769060a0..1b0e40ed 100644 --- a/modules/mesh-build-go/go.mod +++ b/modules/mesh-build-go/go.mod @@ -3,7 +3,7 @@ module meshbuildgo go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028190821-9e5666b4d6f5 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a github.com/distribution/distribution/v3 v3.1.2 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 diff --git a/modules/mesh-build-go/go.sum b/modules/mesh-build-go/go.sum index 42f768be..a31534ed 100644 --- a/modules/mesh-build-go/go.sum +++ b/modules/mesh-build-go/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028190821-9e5666b4d6f5 h1:L7EJbCPZfJRCYHyco5Ck3ui+pWc//sp+h5nnPqSarng= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028190821-9e5666b4d6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a h1:elC2IjB+qbeXG+pWItLz71DenegbrdsqbcH5kyGo56Q= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI= github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= diff --git a/modules/mesh-build-image/go.mod b/modules/mesh-build-image/go.mod index cdfea6a9..db5eee27 100644 --- a/modules/mesh-build-image/go.mod +++ b/modules/mesh-build-image/go.mod @@ -3,7 +3,7 @@ module meshbuildimage go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028190821-9e5666b4d6f5 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-build-image/go.sum b/modules/mesh-build-image/go.sum index ce021e6d..9ca8e0a6 100644 --- a/modules/mesh-build-image/go.sum +++ b/modules/mesh-build-image/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028190821-9e5666b4d6f5 h1:L7EJbCPZfJRCYHyco5Ck3ui+pWc//sp+h5nnPqSarng= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028190821-9e5666b4d6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a h1:elC2IjB+qbeXG+pWItLz71DenegbrdsqbcH5kyGo56Q= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-build-image/mesh-build-image b/modules/mesh-build-image/mesh-build-image new file mode 100755 index 00000000..f7cf1852 Binary files /dev/null and b/modules/mesh-build-image/mesh-build-image differ diff --git a/modules/mesh-build-node/go.mod b/modules/mesh-build-node/go.mod index 3e828400..fcb82a5f 100644 --- a/modules/mesh-build-node/go.mod +++ b/modules/mesh-build-node/go.mod @@ -3,7 +3,7 @@ module meshbuildnode go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028190821-9e5666b4d6f5 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a github.com/distribution/distribution/v3 v3.1.2 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-build-node/go.sum b/modules/mesh-build-node/go.sum index cdad4f6b..1ae199f0 100644 --- a/modules/mesh-build-node/go.sum +++ b/modules/mesh-build-node/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028190821-9e5666b4d6f5 h1:L7EJbCPZfJRCYHyco5Ck3ui+pWc//sp+h5nnPqSarng= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028190821-9e5666b4d6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a h1:elC2IjB+qbeXG+pWItLz71DenegbrdsqbcH5kyGo56Q= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI= github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= diff --git a/modules/mesh-build-node/mesh-build-node b/modules/mesh-build-node/mesh-build-node new file mode 100755 index 00000000..1293e701 Binary files /dev/null and b/modules/mesh-build-node/mesh-build-node differ diff --git a/modules/mesh-watcher/cmd/mesh-watcher/main.go b/modules/mesh-watcher/cmd/mesh-watcher/main.go index 8a64c157..22914015 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/main.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/main.go @@ -15,6 +15,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/health" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } @@ -98,8 +99,10 @@ func run(w *Watcher, l *listening, node string) { time.Sleep(10 * time.Minute) } }() + follow := newTakeover() + go followTakeover(follow) handle := func(e stdio.Envelope) error { - if !isHeartbeat(e.Key) { + if !follow.heard(e.Key) { return nil } b, err := DecodeBeat(e.Body) @@ -113,6 +116,9 @@ func run(w *Watcher, l *listening, node string) { time.Sleep(500 * time.Millisecond) for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := stdio.Subscribe(ControllerSeat+".*", handle) + if err == nil { + err = stdio.Subscribe(health.SelfCheckModule+".*", handle) + } if err == nil { l.set("listening") logf("[mesh-watcher] listening for the self-check's heartbeat; watching from %s", node) diff --git a/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go b/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go index 37052121..ee7f1c9e 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go @@ -34,7 +34,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { if err := json.Unmarshal(raw, &m); err != nil { t.Fatal(err) } - if !reflect.DeepEqual(m.Consumes, []string{ControllerSeat + "." + HeartbeatEvent}) { + if !reflect.DeepEqual(m.Consumes, []string{ControllerSeat + "." + HeartbeatEvent, "mesh-self-check.checked"}) { t.Fatalf("consumes %v", m.Consumes) } if !reflect.DeepEqual(m.Invokes, []string{m.Module + ".watcher_ping", "seat:mesh-controller.seats"}) { diff --git a/modules/mesh-watcher/cmd/mesh-watcher/takeover.go b/modules/mesh-watcher/cmd/mesh-watcher/takeover.go new file mode 100644 index 00000000..540d70c3 --- /dev/null +++ b/modules/mesh-watcher/cmd/mesh-watcher/takeover.go @@ -0,0 +1,59 @@ +package main + +import ( + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// takeover is which self-check run the watcher hears: the controller's `doctor-heartbeat` until the keeper's +// take-over key names the keeper, then mesh-self-check's `checked`, whose body is the same run (novox/hq ADR 0564 +// clause 9, issue 1952 W1). An absent key follows the controller. +type takeover struct { + mu sync.Mutex + t health.Takeover +} + +func newTakeover() *takeover { + return &takeover{t: health.Takeover{V: health.TakeoverVersion, Follow: health.FollowController}} +} + +// heard says whether key is the followed self-check event. +func (f *takeover) heard(key string) bool { + f.mu.Lock() + defer f.mu.Unlock() + seat, event := f.t.SelfCheckEvent() + return key == seat+"."+event +} + +func (f *takeover) set(value []byte, found bool) error { + t, err := health.DecodeTakeover(value, found) + if err != nil { + return err + } + f.mu.Lock() + f.t = t + f.mu.Unlock() + return nil +} + +// followTakeover watches the key for the life of the process; one it cannot read leaves the event it heard. +func followTakeover(f *takeover) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := stdio.State(health.KeeperModule+"."+health.TakeoverState).Watch(health.TakeoverKey, func(c stdio.StateChange) error { + if err := f.set(c.Value, c.Op != "delete" && len(c.Value) > 0); err != nil { + logf("[mesh-watcher] the take-over key is unreadable (%v); no change in what it hears", err) + } else { + logf("[mesh-watcher] the take-over key now follows %s", f.t.Follow) + } + return nil + }) + if err == nil { + return + } + logf("[mesh-watcher] not watching the take-over key yet (%v)", err) + time.Sleep(wait) + } +} diff --git a/modules/mesh-watcher/go.mod b/modules/mesh-watcher/go.mod index 6c449fbb..168dd6c0 100644 --- a/modules/mesh-watcher/go.mod +++ b/modules/mesh-watcher/go.mod @@ -1,5 +1,5 @@ module mesh-watcher -go 1.22 +go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.7 +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a diff --git a/modules/mesh-watcher/go.sum b/modules/mesh-watcher/go.sum index b474419e..ea2abb0e 100644 --- a/modules/mesh-watcher/go.sum +++ b/modules/mesh-watcher/go.sum @@ -1,2 +1,2 @@ -git.novox.be/novox/mesh-sdk/go v0.1.7 h1:C0sTQmtTiyYH7bnqZb7PusXnqA37gKuT7Nqjn9gG47w= -git.novox.be/novox/mesh-sdk/go v0.1.7/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a h1:elC2IjB+qbeXG+pWItLz71DenegbrdsqbcH5kyGo56Q= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= diff --git a/modules/mesh-watcher/module.json b/modules/mesh-watcher/module.json index e9543a5d..26ed5097 100644 --- a/modules/mesh-watcher/module.json +++ b/modules/mesh-watcher/module.json @@ -3,7 +3,11 @@ "version": "1", "slug": "watch", "consumes": [ - "mesh-controller.doctor-heartbeat" + "mesh-controller.doctor-heartbeat", + "mesh-self-check.checked" + ], + "reads": [ + "mesh-conditions.takeover" ], "invokes": [ "mesh-watcher.watcher_ping", diff --git a/modules/messenger/cmd/messenger/condition.go b/modules/messenger/cmd/messenger/condition.go index 1378d596..7a218fad 100644 --- a/modules/messenger/cmd/messenger/condition.go +++ b/modules/messenger/cmd/messenger/condition.go @@ -122,9 +122,12 @@ func (c Condition) SilencedAt(now time.Time) bool { // eventOf is the local event name of an envelope's key: `mesh-controller.condition-raised` is // `condition-raised`. Anything not of the controller's seat is not a condition event. -func eventOf(key string) (string, bool) { +func eventOf(key string) (string, bool) { return eventOfSeat(key, ControllerSeat) } + +// eventOfSeat is eventOf for the seat the take-over key says is followed: the controller's, or the keeper's (ADR 0564 clause 9). +func eventOfSeat(key, seat string) (string, bool) { emitter, event, ok := strings.Cut(key, ".") - if !ok || emitter != ControllerSeat { + if !ok || emitter != seat { return "", false } switch event { diff --git a/modules/messenger/cmd/messenger/main.go b/modules/messenger/cmd/messenger/main.go index e6c963f8..2cc92c95 100644 --- a/modules/messenger/cmd/messenger/main.go +++ b/modules/messenger/cmd/messenger/main.go @@ -18,6 +18,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/health" ) func errorf(format string, a ...any) error { return fmt.Errorf(format, a...) } @@ -415,8 +416,9 @@ func run(h *Holder, l *listening) { // The state first: what is open now is read from the controller before any event is taken, so // what the stream replays is history (novox/hq issue 271). Not reachable yet, it is asked for // again in the loop below, and old events meanwhile are kept as state, never said. + follow := newTakeover() h.Start() - syncFromController(h) + syncFromController(h, follow) go func() { tick := time.NewTicker(5 * time.Second) defer tick.Stop() @@ -429,10 +431,11 @@ func run(h *Holder, l *listening) { } if h.WantsSync() && now.Sub(lastAsk) >= time.Minute { lastAsk = now - syncFromController(h) + syncFromController(h, follow) } } }() + go followTakeover(follow, h) handle := func(e stdio.Envelope) error { if event, ok := strings.CutPrefix(e.Key, ViewModule+"."); ok { // A module in motion on a node, settled or stuck: a walk's start or end now that plans are gone (ADR 0308 §1, ADR 0539 §17). @@ -446,7 +449,9 @@ func run(h *Holder, l *listening) { } return nil } - event, ok := eventOf(e.Key) + // Only the seat the take-over key names is heard: the other's events are dropped, so the keeper's conditions + // are never held a second time beside the controller's (ADR 0564 clause 9). + event, ok := eventOfSeat(e.Key, follow.seat()) if !ok { return nil } @@ -476,6 +481,9 @@ func run(h *Holder, l *listening) { }() for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := stdio.Subscribe(ControllerSeat+".*", handle) + if err == nil { + err = stdio.Subscribe(health.KeeperModule+".*", handle) + } if err == nil { l.set("listening") logf("[messenger] listening for the controller's condition events") @@ -625,7 +633,7 @@ func followKinds(h *Holder) { } // syncFromController reads the controller's open conditions and gives them to the holder. -func syncFromController(h *Holder) { +func syncFromController(h *Holder, follow *takeover) { asked := h.Now() type answer struct { raw json.RawMessage @@ -633,7 +641,7 @@ func syncFromController(h *Holder) { } done := make(chan answer, 1) go func() { - raw, err := stdio.Ask("seat:"+ControllerSeat+".conditions", map[string]any{}) + raw, err := stdio.Ask("seat:"+follow.seat()+".conditions", map[string]any{}) done <- answer{raw, err} }() var a answer diff --git a/modules/messenger/cmd/messenger/manifest_test.go b/modules/messenger/cmd/messenger/manifest_test.go index 9fe112a0..6e61b0de 100644 --- a/modules/messenger/cmd/messenger/manifest_test.go +++ b/modules/messenger/cmd/messenger/manifest_test.go @@ -81,13 +81,14 @@ func TestItDeclaresTheSeatsOfAnAskAndHoldsTheOperatorChannel(t *testing.T) { // } if !reflect.DeepEqual(m.Consumes, []string{ ControllerSeat + "." + EventRaised, ControllerSeat + "." + EventChanged, ControllerSeat + "." + EventCleared, + "mesh-conditions." + EventRaised, "mesh-conditions." + EventChanged, "mesh-conditions." + EventCleared, ViewModule + "." + EventMoving, ViewModule + "." + EventSettled, ViewModule + "." + EventStuck, "intake.choice", "intake.link", "intake.failed", "intake.standing"}) { t.Fatalf("consumes %v", m.Consumes) } // It performs nothing: the verbs it calls read the open conditions, whether root is free, and which issues // track which conditions. if !reflect.DeepEqual(m.Emits, []string{"refused"}) || !reflect.DeepEqual(m.Invokes, []string{"seat:" + ControllerSeat + ".conditions", - "seat:" + EngineSeat + ".root-free", TrackingVerb}) { + "seat:mesh-conditions.conditions", "seat:" + EngineSeat + ".root-free", TrackingVerb}) { t.Fatalf("emits %v invokes %v", m.Emits, m.Invokes) } var state []string diff --git a/modules/messenger/cmd/messenger/sync.go b/modules/messenger/cmd/messenger/sync.go index efeaa6ee..34d1d696 100644 --- a/modules/messenger/cmd/messenger/sync.go +++ b/modules/messenger/cmd/messenger/sync.go @@ -180,3 +180,11 @@ func (h *Holder) WantsSync() bool { h.init() return h.wantSync || h.syncedAt.IsZero() || h.Now().Sub(h.syncedAt) >= ResyncEvery } + +// AskSync has the next pass of the loop read the open conditions again (the take-over key named another seat). +func (h *Holder) AskSync() { + h.mu.Lock() + h.init() + h.wantSync = true + h.mu.Unlock() +} diff --git a/modules/messenger/cmd/messenger/takeover.go b/modules/messenger/cmd/messenger/takeover.go new file mode 100644 index 00000000..e02ae421 --- /dev/null +++ b/modules/messenger/cmd/messenger/takeover.go @@ -0,0 +1,59 @@ +package main + +import ( + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// takeover is which seat's condition events the messenger hears: the controller's until the keeper's take-over key +// says otherwise (novox/hq ADR 0564 clause 9, issue 1952 R1). An absent key follows the controller. +type takeover struct { + mu sync.Mutex + t health.Takeover +} + +func newTakeover() *takeover { + return &takeover{t: health.Takeover{V: health.TakeoverVersion, Follow: health.FollowController}} +} + +func (f *takeover) seat() string { f.mu.Lock(); defer f.mu.Unlock(); return f.t.Seat() } + +// set takes the key as a watch handed it and says whether the followed seat changed. +func (f *takeover) set(value []byte, found bool) (changed bool, err error) { + t, err := health.DecodeTakeover(value, found) + if err != nil { + return false, err + } + f.mu.Lock() + defer f.mu.Unlock() + changed = f.t.Seat() != t.Seat() + f.t = t + return changed, nil +} + +// followTakeover watches the take-over key for the life of the process, and asks the holder to read the followed +// seat's open conditions again when it changes. A key it cannot read leaves the seat it followed. +func followTakeover(f *takeover, h *Holder) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := stdio.State(health.KeeperModule+"."+health.TakeoverState).Watch(health.TakeoverKey, func(c stdio.StateChange) error { + changed, err := f.set(c.Value, c.Op != "delete" && len(c.Value) > 0) + if err != nil { + logf("[messenger] the take-over key is unreadable (%v); still hearing %s", err, f.seat()) + return nil + } + if changed { + logf("[messenger] the take-over key now names %s", f.seat()) + h.AskSync() + } + return nil + }) + if err == nil { + return + } + logf("[messenger] not watching the take-over key yet (%v); hearing %s", err, f.seat()) + time.Sleep(wait) + } +} diff --git a/modules/messenger/go.mod b/modules/messenger/go.mod index e70013a7..5eb5c65e 100644 --- a/modules/messenger/go.mod +++ b/modules/messenger/go.mod @@ -3,7 +3,7 @@ module messenger go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/messenger/go.sum b/modules/messenger/go.sum index aa13456d..9ca8e0a6 100644 --- a/modules/messenger/go.sum +++ b/modules/messenger/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9 h1:qHpN/fpTDqijfMVoepxUT6mPkfcydb4LiPKsIokv6iM= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a h1:elC2IjB+qbeXG+pWItLz71DenegbrdsqbcH5kyGo56Q= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/messenger/module.json b/modules/messenger/module.json index 26ec11a3..eec860f0 100644 --- a/modules/messenger/module.json +++ b/modules/messenger/module.json @@ -77,6 +77,9 @@ "mesh-controller.condition-raised", "mesh-controller.condition-changed", "mesh-controller.condition-cleared", + "mesh-conditions.condition-raised", + "mesh-conditions.condition-changed", + "mesh-conditions.condition-cleared", "mesh-delivery-view.moving", "mesh-delivery-view.settled", "mesh-delivery-view.stuck", @@ -90,9 +93,13 @@ ], "invokes": [ "seat:mesh-controller.conditions", + "seat:mesh-conditions.conditions", "seat:node-engine.root-free", "seat:issue-tracker.tracking" ], + "reads": [ + "mesh-conditions.takeover" + ], "state": [ "open", "sent", diff --git a/modules/minio/module.json b/modules/minio/module.json index 01a966af..9136cd7d 100644 --- a/modules/minio/module.json +++ b/modules/minio/module.json @@ -12,6 +12,9 @@ "identity": { "max": 20, "in": "an S3 access key" + }, + "credential": { + "form": "pair" } } ], diff --git a/modules/mongodb/module.json b/modules/mongodb/module.json index cbb12681..fc7d5212 100644 --- a/modules/mongodb/module.json +++ b/modules/mongodb/module.json @@ -12,6 +12,9 @@ "identity": { "max": 63, "in": "a MongoDB database name" + }, + "credential": { + "form": "pair" } } ], diff --git a/modules/mosquitto/module.json b/modules/mosquitto/module.json index 21058e0a..a7213d44 100644 --- a/modules/mosquitto/module.json +++ b/modules/mosquitto/module.json @@ -8,6 +8,9 @@ "scope": "mesh", "identity": { "in": "a Mosquitto client and topic prefix" + }, + "credential": { + "form": "pair" } } ], diff --git a/modules/mssql/module.json b/modules/mssql/module.json index ed4e4488..35cdbe6e 100644 --- a/modules/mssql/module.json +++ b/modules/mssql/module.json @@ -12,6 +12,9 @@ "identity": { "max": 128, "in": "a SQL Server login and database name" + }, + "credential": { + "form": "pair" } } ], 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..ea37bc88 --- /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. +// NATS_TOOLS_CREDENTIAL names the credential (a word of the bundle's own: the runtime strips MESH_BROKER_FILE) (the process nats-tools's own secret `broker-nats-tools`); 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("NATS_TOOLS_CREDENTIAL", "") + if file == "" { + return nil, nil, errors.New("NATS_TOOLS_CREDENTIAL 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, fmt.Errorf("the credential at %s (NATS_TOOLS_CREDENTIAL) cannot be read: %w", file, 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..2007b1ef 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["NATS_TOOLS_CREDENTIAL"] != "${dir:tools-secret}/broker-nats-tools" { + t.Errorf("the tools' credential is %q: the process nats-tools' own secret broker-nats-tools, under a word of the bundle's own (the runtime strips MESH_BROKER_FILE)", a.Env["NATS_TOOLS_CREDENTIAL"]) + } + } 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..3aff66e5 --- /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 (NATS_TOOLS_CREDENTIAL); 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/go.mod b/modules/nats/go.mod index 9f7e38c0..feb4efed 100644 --- a/modules/nats/go.mod +++ b/modules/nats/go.mod @@ -1,12 +1,12 @@ module nats-tools -go 1.25.0 +go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.7 - github.com/klauspost/compress v1.18.5 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a + github.com/klauspost/compress v1.20.0 github.com/nats-io/nats-server/v2 v2.11.17 - github.com/nats-io/nats.go v1.51.0 + github.com/nats-io/nats.go v1.54.0 ) require ( @@ -14,9 +14,9 @@ require ( github.com/google/go-tpm v0.9.8 // indirect github.com/minio/highwayhash v1.0.4 // indirect github.com/nats-io/jwt/v2 v2.8.1 // indirect - github.com/nats-io/nkeys v0.4.15 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect - golang.org/x/crypto v0.50.0 // indirect - golang.org/x/sys v0.43.0 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 // indirect golang.org/x/time v0.15.0 // indirect ) diff --git a/modules/nats/go.sum b/modules/nats/go.sum index b28aa2f0..9ca8e0a6 100644 --- a/modules/nats/go.sum +++ b/modules/nats/go.sum @@ -1,27 +1,27 @@ -git.novox.be/novox/mesh-sdk/go v0.1.7 h1:C0sTQmtTiyYH7bnqZb7PusXnqA37gKuT7Nqjn9gG47w= -git.novox.be/novox/mesh-sdk/go v0.1.7/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a h1:elC2IjB+qbeXG+pWItLz71DenegbrdsqbcH5kyGo56Q= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= -github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI= -github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= -github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4= -github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= -golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= -golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= -golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= 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..d2d02efe 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -8,7 +8,10 @@ "provides": [ { "name": "mesh-bus", - "scope": "mesh" + "scope": "mesh", + "credential": { + "form": "none" + } } ], "claims": [ @@ -19,12 +22,16 @@ ], "bus-users": "/var/lib/nats-module/conf/accounts.conf", "own-secrets": { - "broker": "${dir:mesh-state}/broker" + "broker": "${dir:mesh-state}/broker", + "broker-nats-tools": "${dir:tools-secret}/broker-nats-tools" }, "capabilities": [ "container-runtime" ], - "emits": [], + "emits": [ + "consumer-removed", + "bucket-keys-removed" + ], "consumes": [], "listens": [ { @@ -45,7 +52,9 @@ "nats_buckets", "nats_users", "nats_user_can", - "nats_snapshot_verify" + "nats_snapshot_verify", + "nats_consumer_remove", + "nats_bucket_remove" ], "guards": [ 8222 @@ -86,6 +95,13 @@ "mode": "0700", "unknown": "program" }, + { + "id": "tools-secret", + "type": "directory", + "path": "/var/lib/mesh/nats-tools", + "mode": "0711", + "unknown": "program" + }, { "id": "snapshots", "type": "directory", @@ -190,11 +206,18 @@ "loads": [ "nats-tools" ], + "bus": { + "removes": [ + "DEAD_LETTER_NOTICES.controller", + "mesh-host_acts" + ] + }, "env": { "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}", + "NATS_TOOLS_CREDENTIAL": "${dir:tools-secret}/broker-nats-tools" } }, { diff --git a/modules/nfs-server/module.json b/modules/nfs-server/module.json index f9994d16..d9210c12 100644 --- a/modules/nfs-server/module.json +++ b/modules/nfs-server/module.json @@ -13,7 +13,10 @@ { "name": "nfs-share", "scope": "mesh", - "identity": false + "identity": false, + "credential": { + "form": "none" + } } ], "data": { @@ -51,20 +54,20 @@ "seat:mesh-controller.seats", "seat:mesh-controller.data" ], - "config": { - "shares": { - "type": "string", - "description": "The folders this machine exports, as space-separated name=path; the operator's, with no default (novox/hq ADR 0263).", - "trusted": true, - "required": true - }, - "grants": { - "type": "string", - "description": "Which nodes may mount which share, and how: share=node:rw|ro[,node:rw|ro], entries separated by spaces, or none to export to no node whatever a node wants (novox/hq ADR 0263).", - "trusted": true, - "default": "none" - } + "config": { + "shares": { + "type": "string", + "description": "The folders this machine exports, as space-separated name=path; the operator's, with no default (novox/hq ADR 0263).", + "trusted": true, + "required": true }, + "grants": { + "type": "string", + "description": "Which nodes may mount which share, and how: share=node:rw|ro[,node:rw|ro], entries separated by spaces, or none to export to no node whatever a node wants (novox/hq ADR 0263).", + "trusted": true, + "default": "none" + } + }, "facts": { "machines": { "path": "/etc/nfs-server/machines", diff --git a/modules/nodejs/module.json b/modules/nodejs/module.json index c3573444..2f4a049a 100644 --- a/modules/nodejs/module.json +++ b/modules/nodejs/module.json @@ -7,7 +7,10 @@ "provides": [ { "name": "node-toolchain", - "reach": "machine" + "reach": "machine", + "credential": { + "form": "none" + } } ], "resources": [ diff --git a/modules/ollama/module.json b/modules/ollama/module.json index 2d5fb48c..312ef6bf 100644 --- a/modules/ollama/module.json +++ b/modules/ollama/module.json @@ -7,7 +7,10 @@ "provides": [ { "name": "model-access", - "scope": "node" + "scope": "node", + "credential": { + "form": "none" + } } ], "listens": [ diff --git a/modules/redis/module.json b/modules/redis/module.json index 58e11064..4cc6c4e5 100644 --- a/modules/redis/module.json +++ b/modules/redis/module.json @@ -7,6 +7,9 @@ "scope": "mesh", "identity": { "in": "a Redis ACL user and key prefix" + }, + "credential": { + "form": "pair" } } ], diff --git a/modules/showcase/module.json b/modules/showcase/module.json index e657e2ab..6b0db672 100644 --- a/modules/showcase/module.json +++ b/modules/showcase/module.json @@ -9,7 +9,10 @@ "provides": [ { "name": "greeting", - "scope": "mesh" + "scope": "mesh", + "credential": { + "form": "none" + } } ], "serves": { diff --git a/modules/step-ca/module.json b/modules/step-ca/module.json index b11d2b99..3f44ead7 100644 --- a/modules/step-ca/module.json +++ b/modules/step-ca/module.json @@ -7,7 +7,10 @@ "provides": [ { "name": "internal-acme-ca", - "scope": "mesh" + "scope": "mesh", + "credential": { + "form": "none" + } } ], "serves": { diff --git a/modules/systemd-resolved/module.json b/modules/systemd-resolved/module.json index b850afcf..24b3bc85 100644 --- a/modules/systemd-resolved/module.json +++ b/modules/systemd-resolved/module.json @@ -8,7 +8,10 @@ "provides": [ { "name": "split-dns", - "reach": "machine" + "reach": "machine", + "credential": { + "form": "none" + } } ], "requires": [ diff --git a/modules/umami/module.json b/modules/umami/module.json index 22ead419..e3b9aa0d 100644 --- a/modules/umami/module.json +++ b/modules/umami/module.json @@ -32,7 +32,10 @@ "provides": [ { "name": "analytics", - "scope": "mesh" + "scope": "mesh", + "credential": { + "form": "none" + } } ], "serves": { diff --git a/modules/xorg/module.json b/modules/xorg/module.json index 7b0b6b1e..1685dcb4 100644 --- a/modules/xorg/module.json +++ b/modules/xorg/module.json @@ -9,7 +9,10 @@ "provides": [ { "name": "x11-display", - "reach": "machine" + "reach": "machine", + "credential": { + "form": "none" + } } ], "claims": [