From 1ba6c0de270bdb372dec7c9583414a912ccbc7dd Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 23:44:35 +0100 Subject: [PATCH] =?UTF-8?q?mesh-self-check:=20D8=20reads=20placement's=20a?= =?UTF-8?q?ddresses=20and=20endpoints=20and=20asks=20each=20heard=20holder?= =?UTF-8?q?'s=20ban=20list,=20so=20the=20probe=20moves=20off=20the=20contr?= =?UTF-8?q?oller=20with=20its=20kind,=20key=20and=20words=20(issue=201952,?= =?UTF-8?q?=20ADR=200564=20=C2=A76)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mesh's own addresses come from placement (each node's private address and its endpoint); who holds node-intrusion-prevention, from seat-holders; who is heard, from the watchman's presence. Each heard holder is asked seat:node-intrusion-prevention.banned@, all at once. The finding is the controller's: machine..bans-the-mesh, urgent, the same summary and words. A heard holder that does not answer makes the probe fail to run, never pass. Rows in TestTheBusProbes; the manifest reads and invokes what the code does. --- .../cmd/mesh-self-check/manifest_test.go | 8 +- .../cmd/mesh-self-check/probes.go | 10 +- .../cmd/mesh-self-check/probes_mesh.go | 219 ++++++++++++++++++ .../cmd/mesh-self-check/probes_test.go | 70 ++++-- modules/mesh-self-check/go.mod | 2 +- modules/mesh-self-check/go.sum | 4 +- modules/mesh-self-check/module.json | 8 +- 7 files changed, 296 insertions(+), 25 deletions(-) create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go diff --git a/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go b/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go index fb245a78..e288898c 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go @@ -51,11 +51,11 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { if len(m.Emits) != 0 { t.Errorf("emits %v", m.Emits) } - if !slices.Equal(m.Invokes, []string{StreamsTool, ClosedTool, SignalsVerb}) { - t.Errorf("invokes %v, want what the probes ask", m.Invokes) + if !slices.Equal(m.Invokes, Invokes) { + t.Errorf("invokes %v, want what the probes ask, %v", m.Invokes, Invokes) } - if len(m.Reads) != 0 { - t.Errorf("reads %v: the self-check reads its facts through the bus module's tools", m.Reads) + if !slices.Equal(m.Reads, Reads) { + t.Errorf("reads %v, want what the probes read, %v", m.Reads, Reads) } states := map[string]int{} for _, s := range m.State { diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes.go b/modules/mesh-self-check/cmd/mesh-self-check/probes.go index 44be64ce..382633b6 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes.go @@ -30,6 +30,13 @@ const ( reconnectMax = 12 ) +// Invokes are the tools and seat verbs the probes ask, and Reads the states they read: the manifest's `invokes` and +// `reads`, which its test holds to these. +var ( + Invokes = []string{StreamsTool, ClosedTool, SignalsVerb, BannedVerb} + Reads = []string{PlacementRead, SeatHoldersRead, PresenceRead} +) + // Registry is the self-check's registry, in to-be 45's order. **The registry is the design's live form**: a probe // that moves here is a row added, and the registry test holds it to ADR 0564's table. func Registry() []Probe { @@ -49,8 +56,7 @@ func Registry() []Probe { {ID: "D7", Asserts: "every stream the mesh defines exists with its definition", From: "issue 208", Kind: "stream-wrong", Run: probeStreams}, {ID: "D8", Asserts: "no address the mesh owns — a machine's private address or its endpoint — is in a ban list", - From: "issue 238", Kind: "own-address-banned", - Deferred: "the machines' private addresses and endpoints: the overlay is " + storeOnly}, + From: "issue 238", Kind: "own-address-banned", Run: probeBans}, {ID: "D11", Asserts: "no provider holds a consumer retired more than thirty days without a person deciding its " + "cleanup", From: "ADR 0230", Kind: "cleanup-waiting", Deferred: "the retirement records of providers' consumers: " + storeOnly}, diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go new file mode 100644 index 00000000..a25fab0a --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go @@ -0,0 +1,219 @@ +package main + +// The probes of the mesh's machines (novox/hq ADR 0564 §6, issue 1952): each reads its facts where they live on the +// bus — placement (the seat mesh-placement's state), who holds which seat (its seat-holders), whether a node is heard +// from (the watchman's presence) — and asks each node's seat holder what only that node knows. Its assertion, kind, +// key and words are the controller's (mesh-controller probes.go), so the condition the keeper raises from it is the one +// the controller raised. + +import ( + "context" + "encoding/json" + "fmt" + "net" + "regexp" + "slices" + "sort" + "strings" + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The states the machine probes read (declared under `reads`), and the verb D8 asks each node. +const ( + PlacementRead = delivery.AssignerModule + "." + "placement" + SeatHoldersRead = delivery.SeatHoldersRead + PresenceRead = delivery.WatchmanModule + ".presence" + BannedSeat = "node-intrusion-prevention" + BannedVerb = "seat:" + BannedSeat + ".banned" + ScopeMachine = health.ScopeMachine +) + +// placements is every node's placement, by node. +func (c *Checker) placements() (map[string]delivery.Placement, error) { + keys, err := c.Store.Keys(PlacementRead) + if err != nil { + return nil, fmt.Errorf("the placement cannot be read: %w", err) + } + out := map[string]delivery.Placement{} + for _, k := range keys { + raw, ok, err := c.Store.Get(PlacementRead, k) + if err != nil { + return nil, fmt.Errorf("the placement of %s cannot be read: %w", k, err) + } + if !ok { + continue + } + var p delivery.Placement + if err := json.Unmarshal(raw, &p); err != nil { + return nil, fmt.Errorf("the placement of %s is not readable: %w", k, err) + } + if p.Node == "" { + p.Node = k + } + out[p.Node] = p + } + return out, nil +} + +// heard is each node the watchman says is online. +func (c *Checker) heard() (map[string]bool, error) { + keys, err := c.Store.Keys(PresenceRead) + if err != nil { + return nil, fmt.Errorf("the watchman's presence cannot be read: %w", err) + } + out := map[string]bool{} + for _, k := range keys { + node, _, _, err := delivery.ParsePresenceKey(k) + if err != nil || node == "" { + continue + } + raw, ok, err := c.Store.Get(PresenceRead, k) + if err != nil { + return nil, fmt.Errorf("the presence of %s cannot be read: %w", node, err) + } + var p delivery.Presence + if ok && json.Unmarshal(raw, &p) == nil && p.Online { + out[node] = true + } + } + return out, nil +} + +// holdersOf is each node on which a module holds the node seat seat, sorted. +func (c *Checker) holdersOf(seat string) ([]string, error) { + keys, err := c.Store.Keys(SeatHoldersRead) + if err != nil { + return nil, fmt.Errorf("who holds which seat cannot be read: %w", err) + } + var out []string + for _, k := range keys { + if !strings.HasPrefix(k, seat+".") { + continue + } + raw, ok, err := c.Store.Get(SeatHoldersRead, k) + if err != nil { + return nil, fmt.Errorf("the holders of %s cannot be read: %w", k, err) + } + var sh delivery.SeatHolders + if !ok || json.Unmarshal(raw, &sh) != nil || sh.Seat != seat || sh.Node == "" || len(sh.Holders) == 0 { + continue + } + if !slices.Contains(out, sh.Node) { + out = append(out, sh.Node) + } + } + sort.Strings(out) + return out, nil +} + +// --- D8 ---------------------------------------------------------------------------------------------------------- + +// ipLike finds the addresses in a ban list, whatever shape its holder answers in. +var ipLike = regexp.MustCompile(`\b(?:\d{1,3}\.){3}\d{1,3}\b`) + +// lookupHost resolves an endpoint's name; replaced in tests. +var lookupHost = func(ctx context.Context, host string) ([]string, error) { + return net.DefaultResolver.LookupHost(ctx, host) +} + +// probeBans is D8: no address the mesh owns is in any machine's ban list. The mesh's own addresses are every machine's +// private address and the address its endpoint names, from placement; the ban lists are asked of each machine's holder +// of node-intrusion-prevention that is heard from. +func probeBans(ctx context.Context, c *Checker) ([]Observation, error) { + places, err := c.placements() + if err != nil { + return nil, err + } + // address → whose, in machine names: what the summary says. The address, and the endpoint's name it was found + // by, are the evidence (ADR 0234 §6). + owned, via := map[string]string{}, map[string]string{} + nodes := make([]string, 0, len(places)) + for n := range places { + nodes = append(nodes, n) + } + sort.Strings(nodes) + for _, n := range nodes { + d := places[n].Decided + if a, _, _ := strings.Cut(d.Address, "/"); a != "" { + owned[a] = n + "'s private address" + } + if host, _, err := net.SplitHostPort(d.Endpoint); err == nil && host != "" { + if ip := net.ParseIP(host); ip != nil { + owned[ip.String()] = n + "'s endpoint" + } else if addrs, err := lookupHost(ctx, host); err == nil { + for _, a := range addrs { + owned[a], via[a] = n+"'s endpoint", host + } + } + } + } + heard, err := c.heard() + if err != nil { + return nil, err + } + all, err := c.holdersOf(BannedSeat) + if err != nil { + return nil, err + } + var holders []string + for _, n := range all { + if heard[n] { + holders = append(holders, n) + } + } + // Every machine asked at once: one after another, four holders that each wait their bound outlast the probe's + // thirty seconds, and the probe says nothing about any of them. + answers := make([]json.RawMessage, len(holders)) + errs := make([]error, len(holders)) + var wg sync.WaitGroup + for i, node := range holders { + wg.Add(1) + go func(i int, node string) { + defer wg.Done() + answers[i], errs[i] = c.Ask(BannedVerb+"@"+node, map[string]any{}) + }(i, node) + } + wg.Wait() + var out []Observation + var unasked []string + for i, node := range holders { + if errs[i] != nil { + unasked = append(unasked, errs[i].Error()) + continue + } + var banned, whose []string + for _, ip := range ipLike.FindAllString(string(answers[i]), -1) { + who, ours := owned[ip] + said := ip + " (" + who + ")" + if h := via[ip]; h != "" { + said = ip + " (" + who + ", " + h + ")" + } + if !ours || slices.Contains(banned, said) { + continue + } + banned = append(banned, said) + if !slices.Contains(whose, who) { + whose = append(whose, who) + } + } + if len(banned) > 0 { + out = append(out, Observation{Kind: "own-address-banned", Token: "bans-the-mesh", Severity: Urgent, + Subject: Subject{Scope: ScopeMachine, ID: node, Machine: node}, + Summary: fmt.Sprintf("%s's ban list holds %d of the mesh's own address(es) — %s: the mesh is locked out "+ + "of itself there (ADR 0186)", node, len(banned), strings.Join(whose, ", ")), + Said: strings.Join(banned, ", "), + Headline: node + " has banned the mesh", + Needs: fmt.Sprintf("lift the ban on %s, or wait for it to expire.", node), + Explanation: fmt.Sprintf("The intrusion protection on %s banned one of the mesh's own addresses, so the "+ + "mesh is locked out of it.", node), + Resolved: node + " no longer bans the mesh"}) + } + } + if len(unasked) > 0 { + return nil, fmt.Errorf("a ban list could not be read: %s", strings.Join(unasked, "; ")) + } + return sorted(out), nil +} diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go index 825416e0..77c3309f 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go @@ -9,6 +9,7 @@ import ( "testing" "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/delivery" ) // bus is the bus module's answers as a probe asks them: every mesh stream and consumer as defined, unless changed. @@ -18,6 +19,8 @@ type bus struct { closed string signals error asksFail error + // banned is each node's answer to node-intrusion-prevention.banned; a node absent does not answer. + banned map[string]string } func (b bus) rows() map[string]*streamRow { @@ -85,6 +88,11 @@ func (b bus) ask(key string, body any) (json.RawMessage, error) { } return json.RawMessage(`{"signals":[]}`), nil } + if node, ok := strings.CutPrefix(key, BannedVerb+"@"); ok { + if a, ok := b.banned[node]; ok { + return json.RawMessage(a), nil + } + } return nil, errors.New("nats: no responders available for request") } @@ -104,38 +112,52 @@ func TestTheBusProbes(t *testing.T) { bus bus want []string err string + mesh *memStore // the bus states the machine probes read; nil reads none }{ - {"D6 all as defined", probeConsumers, bus{}, nil, ""}, + {"D6 all as defined", probeConsumers, bus{}, nil, "", nil}, {"D6 a consumer gone", probeConsumers, bus{drop: map[string]bool{"DEAD_LETTER_NOTICES.controller": true}}, - []string{"bus.DEAD_LETTER_NOTICES.controller.missing"}, ""}, + []string{"bus.DEAD_LETTER_NOTICES.controller.missing"}, "", nil}, {"D6 a stream gone takes its consumers", probeConsumers, bus{drop: map[string]bool{"CONTROL": true}}, - []string{"bus.CONTROL.controller.missing"}, ""}, + []string{"bus.CONTROL.controller.missing"}, "", nil}, {"D6 redefined and far behind", probeConsumers, bus{change: func(rows map[string]*streamRow) { for i, c := range rows["EVENTS"].Consumers { if c.Name == "controller" { rows["EVENTS"].Consumers[i].MaxDeliver, rows["EVENTS"].Consumers[i].Pending = 9, 5000 } } - }}, []string{"bus.EVENTS.controller.behind", "bus.EVENTS.controller.redefined"}, ""}, - {"D6 the bus module silent", probeConsumers, bus{asksFail: errors.New("no responders")}, nil, "did not answer"}, - {"D7 all as defined", probeStreams, bus{}, nil, ""}, + }}, []string{"bus.EVENTS.controller.behind", "bus.EVENTS.controller.redefined"}, "", nil}, + {"D6 the bus module silent", probeConsumers, bus{asksFail: errors.New("no responders")}, nil, "did not answer", nil}, + {"D7 all as defined", probeStreams, bus{}, nil, "", nil}, {"D7 missing and redefined", probeStreams, bus{drop: map[string]bool{"NODES": true}, change: func(rows map[string]*streamRow) { rows["EVENTS"].Retention = "workqueue" - }}, []string{"bus.EVENTS.stream-redefined", "bus.NODES.stream-missing"}, ""}, - {"D15 quiet", probeReconnects, bus{}, nil, ""}, + }}, []string{"bus.EVENTS.stream-redefined", "bus.NODES.stream-missing"}, "", nil}, + {"D15 quiet", probeReconnects, bus{}, nil, "", nil}, {"D15 a loop", probeReconnects, bus{closed: `{"hours":1,"reaches":true,"users":[ {"user":"g14.node-tools","closed":30,"dropped":20,"names":[{"name":"node-tools","dropped":20,"reasons":{"Stale Connection":20}}]}, {"user":"ace.records","closed":5,"dropped":3},{"user":"(unauthenticated)","closed":90,"dropped":90}]}`}, - []string{"bus.g14.node-tools.bus-reconnects"}, ""}, - {"DW answers", probeWatchdogs, bus{}, nil, ""}, - {"DW silent", probeWatchdogs, bus{signals: errors.New("no responders")}, []string{"core.watchdogs.silent"}, ""}, - {"H-bus healthy", probeBusHealth, bus{}, nil, ""}, + []string{"bus.g14.node-tools.bus-reconnects"}, "", nil}, + {"DW answers", probeWatchdogs, bus{}, nil, "", nil}, + {"DW silent", probeWatchdogs, bus{signals: errors.New("no responders")}, []string{"core.watchdogs.silent"}, "", nil}, + {"H-bus healthy", probeBusHealth, bus{}, nil, "", nil}, {"H-bus a stream and a consumer missing", probeBusHealth, - bus{drop: map[string]bool{"NODES": true, "EVENTS.controller-merges": true}}, []string{"bus.mesh.unhealthy"}, ""}, - {"H-bus no round trip", probeBusHealth, bus{asksFail: errors.New("timeout")}, nil, "no request crossed"}, + bus{drop: map[string]bool{"NODES": true, "EVENTS.controller-merges": true}}, []string{"bus.mesh.unhealthy"}, "", nil}, + {"H-bus no round trip", probeBusHealth, bus{asksFail: errors.New("timeout")}, nil, "no request crossed", nil}, + // D8 (issue 238): placement's addresses and endpoints against each heard holder's ban list. + {"D8 nothing of the mesh banned", probeBans, bus{banned: map[string]string{"ace": `{"banned":["198.51.100.7"]}`, + "novox": `{"banned":[]}`}}, nil, "", machines()}, + {"D8 a private address and an endpoint banned", probeBans, bus{banned: map[string]string{ + "ace": `{"jails":{"sshd":["10.10.0.1","198.51.100.7"]}}`, "novox": `{"banned":["203.0.113.5"]}`}}, + []string{"machine.ace.bans-the-mesh", "machine.novox.bans-the-mesh"}, "", machines()}, + {"D8 a machine not heard from is not asked", probeBans, bus{banned: map[string]string{"novox": `{"banned":[]}`, + "ace": `{"banned":[]}`}}, nil, "", machines()}, + {"D8 a heard holder that does not answer is no pass", probeBans, bus{banned: map[string]string{}}, nil, + "a ban list could not be read", machines()}, } for _, tc := range cases { - c := &Checker{Ask: tc.bus.ask} + c := &Checker{Ask: tc.bus.ask, Store: tc.mesh} + if tc.mesh == nil { + c.Store = newStore() + } found, err := tc.probe(context.Background(), c) if tc.err != "" { if err == nil || !strings.Contains(err.Error(), tc.err) { @@ -153,6 +175,24 @@ func TestTheBusProbes(t *testing.T) { } } +// machines is a mesh of three: novox the hub with its endpoint, ace and g14 on the private network; ace and novox +// hold node-intrusion-prevention and are heard from, g14 holds it and is not. +func machines() *memStore { + s := newStore() + for _, p := range []struct { + node, address, endpoint string + online bool + }{{"novox", "10.10.0.1/16", "203.0.113.5:51820", true}, {"ace", "10.10.0.2/16", "", true}, {"g14", "10.10.0.3/16", "", false}} { + _ = s.Put(PlacementRead, p.node, delivery.Placement{V: 1, Node: p.node, + Decided: delivery.NodeDecided{Address: p.address, Endpoint: p.endpoint, Hub: p.endpoint != ""}}) + _ = s.Put(PresenceRead, p.node, delivery.Presence{V: 1, Online: p.online}) + key, _ := delivery.SeatHoldersKey(BannedSeat, p.node) + _ = s.Put(SeatHoldersRead, key, delivery.SeatHolders{V: 1, Seat: BannedSeat, Scope: "node", Node: p.node, + Holders: []delivery.SeatHolder{{Module: "fail2ban", Node: p.node}}}) + } + return s +} + // The registry holds exactly the probes ADR 0564's table moves here, each with a kind, and a deferred one names what // it lacks; the probes that end with the controller or moved to the watchman are not in it (issue 1952 SC4). func TestTheRegistryIsADR0564sTable(t *testing.T) { diff --git a/modules/mesh-self-check/go.mod b/modules/mesh-self-check/go.mod index 273063e8..c4aa9ab5 100644 --- a/modules/mesh-self-check/go.mod +++ b/modules/mesh-self-check/go.mod @@ -3,7 +3,7 @@ module mesh-self-check go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223859-85f5fd0b2266 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-self-check/go.sum b/modules/mesh-self-check/go.sum index 9ca8e0a6..ad48fccc 100644 --- a/modules/mesh-self-check/go.sum +++ b/modules/mesh-self-check/go.sum @@ -1,5 +1,5 @@ -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= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223859-85f5fd0b2266 h1:32lx0DCHm+7Ig6t8vXbfwPbk4YgbHQt1sRzmRlezFWw= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223859-85f5fd0b2266/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-self-check/module.json b/modules/mesh-self-check/module.json index 829af53e..c2c810d2 100644 --- a/modules/mesh-self-check/module.json +++ b/modules/mesh-self-check/module.json @@ -15,7 +15,13 @@ "invokes": [ "nats.nats_streams", "nats.nats_closed_connections", - "seat:mesh-watchdogs.signals" + "seat:mesh-watchdogs.signals", + "seat:node-intrusion-prevention.banned" + ], + "reads": [ + "mesh-delivery-assigner.placement", + "mesh-delivery-assigner.seat-holders", + "mesh-delivery-watchman.presence" ], "state": [ "findings",