mesh-catalog/modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go

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, from the state's keys in either form: a node key and its module keys,
// or the form before, `placement.<node>` (delivery.PlacementsOf; research 072 row 37).
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)
	}
	values := map[string][]byte{}
	for _, k := range keys {
		raw, ok, err := c.Store.Get(PlacementRead, k)
		if err != nil {
			return nil, fmt.Errorf("the placement's key %s cannot be read: %w", k, err)
		}
		if ok {
			values[k] = raw
		}
	}
	out, err := delivery.PlacementsOf(values)
	if err != nil {
		return nil, fmt.Errorf("the placement is not readable: %w", err)
	}
	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", Ref: "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
}