mesh-self-check: D8 reads placement's addresses and endpoints and asks each heard holder's ban list, so the probe moves off the controller with its kind, key and words (issue 1952, ADR 0564 §6)

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@<node>, all at once. The finding is the controller's: machine.<node>.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.
This commit is contained in:
2026-10-29 23:44:35 +01:00
parent bb18541469
commit 1ba6c0de27
7 changed files with 296 additions and 25 deletions
@@ -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 {
@@ -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},
@@ -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
}
@@ -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) {
+1 -1
View File
@@ -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
)
+2 -2
View File
@@ -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=
+7 -1
View File
@@ -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",