Merge pull request 'Phase 5 (1/4): keep a facts snapshot for merge checks, and say when it goes stale (hq ADR 0237, S14)' (#95) from feat/facts-snapshot into main
mesh/delivery held for a person: merged without a passing check: only a person decides that it goes on

This commit was merged in pull request #95.
This commit is contained in:
2026-10-06 19:19:19 +00:00
13 changed files with 1921 additions and 3 deletions
+612
View File
@@ -0,0 +1,612 @@
package main
import (
"context"
"crypto/sha256"
"encoding/json"
"errors"
"flag"
"fmt"
"os"
"slices"
"sort"
"strings"
"sync"
"time"
"github.com/novox/mesh-host/validate"
"github.com/novox/mesh-controller/internal/artifacts"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/catalogue"
snapshot "github.com/novox/mesh-controller/internal/facts"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// The facts snapshot (novox/hq to-be 45 §9, ADR 0227 rule 9): what a merge check needs to judge a change
// against the mesh that runs, written by the controller to the artifact store, where the build seat reads
// it. internal/facts says what it holds and what it never holds; this composes it from the store and
// keeps it current.
// factsEvery is how often the snapshot is composed. It is kept when it moved — a machine, an
// assignment, a seat, a setting, a build — or once a day when nothing did, so its age says the
// controller is still writing it (S14).
var factsEvery = 10 * time.Minute
// factsDaily is how old a snapshot of an unchanged mesh may grow before it is written again.
const factsDaily = 24 * time.Hour
// factsStaleAfter is S14's bound: a snapshot older than this is one no check should be fed.
const factsStaleAfter = 48 * time.Hour
// factsExport is what this controller knows of the snapshot it keeps: when the newest was taken, its
// content, and the last attempt's error.
type factsExport struct {
mu sync.Mutex
taken time.Time
content string
digest string
err error
began time.Time
}
// exportedFacts is this process's export, read by S14.
var exportedFacts = &factsExport{}
func (e *factsExport) last() (taken time.Time, digest string, began time.Time, err error) {
e.mu.Lock()
defer e.mu.Unlock()
return e.taken, e.digest, e.began, e.err
}
func (e *factsExport) kept(f snapshot.Facts, content, digest string) {
e.mu.Lock()
defer e.mu.Unlock()
e.taken, e.content, e.digest, e.err = f.Taken, content, digest, nil
}
func (e *factsExport) failed(err error) {
e.mu.Lock()
defer e.mu.Unlock()
e.err = err
}
// exportingFacts keeps the snapshot current for as long as this controller holds the lease: ctx ends
// when it stops acting.
func exportingFacts(ctx context.Context, open *stores, busVersion func() string) {
exportedFacts.mu.Lock()
exportedFacts.began = time.Now()
exportedFacts.mu.Unlock()
// What the store holds already, so a restarted controller neither writes an unchanged snapshot again
// nor reads its age as zero.
if address, err := factsStore(ctx, open); err == nil {
if body, digest, err := (artifacts.Store{Address: address}).GetTagged(ctx, snapshot.Repository, snapshot.Tag); err == nil {
if f, err := snapshot.Decode(body); err == nil {
content, _ := f.Content()
exportedFacts.kept(f, content, digest)
}
}
}
failing := ""
first := time.NewTimer(time.Minute)
defer first.Stop()
tick := time.NewTicker(factsEvery)
defer tick.Stop()
for {
select {
case <-ctx.Done():
return
case <-first.C:
case <-tick.C:
}
wrote, err := exportFacts(ctx, open, busVersion(), false)
why := ""
if err != nil {
why = err.Error()
exportedFacts.failed(err)
}
if why != failing {
if why != "" {
fmt.Printf("the facts snapshot cannot be kept: %s\n", why)
} else {
fmt.Println("the facts snapshot is kept again")
}
failing = why
}
if wrote != "" {
fmt.Printf("the facts snapshot moved and is kept as %s\n", short(strings.TrimPrefix(wrote, "sha256:")))
}
}
}
// exportFacts composes the snapshot and keeps it when it moved, or when the one kept is a day old, or
// when told to. Answers the digest it kept, empty when it kept nothing.
func exportFacts(ctx context.Context, open *stores, busVersion string, force bool) (string, error) {
f, err := gatherFacts(ctx, open, busVersion)
if err != nil {
return "", err
}
content, err := f.Content()
if err != nil {
return "", err
}
exportedFacts.mu.Lock()
unchanged := content == exportedFacts.content && time.Since(exportedFacts.taken) < factsDaily
exportedFacts.mu.Unlock()
if unchanged && !force {
return "", nil
}
body, err := f.Encode()
if err != nil {
return "", err
}
address, err := factsStore(ctx, open)
if err != nil {
return "", err
}
digest, err := (artifacts.Store{Address: address}).PutTagged(ctx, snapshot.Repository, snapshot.Tag, snapshot.MediaType, body)
if err != nil && digest == "" {
return "", err
}
exportedFacts.kept(f, content, digest)
return digest, err
}
// factsStore is the artifact store as this controller reaches it.
func factsStore(ctx context.Context, open *stores) (string, error) {
shelf, err := open.inventory.Catalogue(ctx)
if err != nil {
return "", err
}
address, err := artifactStoreAddress(ctx, open.inventory, shelf, "")
if err != nil {
return "", err
}
if address == "" {
return "", errors.New("the artifact store is not on the private network, so there is nowhere to keep the facts")
}
return address, nil
}
// gatherFacts composes one snapshot from the store: read only, nothing made, nothing sent.
func gatherFacts(ctx context.Context, open *stores, busVersion string) (snapshot.Facts, error) {
inv := open.inventory
f := snapshot.Facts{Format: snapshot.Format, Taken: time.Now().UTC(), Controller: snapshot.Build{Version: version}}
f.Versions.Bus = busVersion
storeVersion, err := inv.ServerVersion(ctx)
if err != nil {
return snapshot.Facts{}, fmt.Errorf("the store will not say its version: %w", err)
}
f.Versions.Store = storeVersion
nodes, err := inv.Nodes(ctx)
if err != nil {
return snapshot.Facts{}, err
}
overlays, err := inv.Overlays(ctx)
if err != nil {
return snapshot.Facts{}, err
}
place := map[string]inventory.Overlay{}
for _, o := range overlays {
place[o.Name] = o
}
entries, err := inv.Catalogued(ctx)
if err != nil {
return snapshot.Facts{}, err
}
shelf := map[string]catalogue.Manifest{}
for _, e := range entries {
shelf[e.Manifest.Module] = e.Manifest
}
current, err := inv.CurrentBuilds(ctx)
if err != nil {
return snapshot.Facts{}, err
}
read, err := inv.ReadRepositories(ctx)
if err != nil {
return snapshot.Facts{}, err
}
edges, err := inv.Dependencies(ctx)
if err != nil {
return snapshot.Facts{}, err
}
holdings, err := inv.Holdings(ctx)
if err != nil {
return snapshot.Facts{}, err
}
// **Every name first**, so text read afterwards — a setting naming a machine, a problem naming a
// site — has it replaced wherever it appears.
scrub := snapshot.NewScrubber()
domains := map[string]string{}
for _, n := range nodes {
scrub.Machine(n.Name)
if n.Account != "" && n.Account != "root" {
scrub.Account(n.Account)
}
d, err := inv.PublicDomainOf(ctx, n.Name)
if err != nil {
return snapshot.Facts{}, err
}
domains[n.Name] = scrub.Domain(d)
}
for _, o := range overlays {
scrub.Site(o.Site)
}
if c, ok := current["mesh-controller"]; ok {
f.Controller.Commit = c.Commit
}
gens, gensErr := generators(ctx, open)
hostShelf := shelf[hostModule]
meshWide := map[string]bool{}
engines := map[string]bool{}
for _, n := range nodes {
m := snapshot.Machine{Name: scrub.Machine(n.Name), Length: len(n.Name), Adopted: n.Adopted,
AccountHome: scrub.Text(n.AccountHome), NodeEngine: n.HostVersion, PublicDomain: domains[n.Name]}
switch n.Account {
case "", "root":
m.Account = n.Account
default:
m.Account = scrub.Account(n.Account)
}
if n.HostVersion != "" {
engines[n.HostVersion] = true
}
m.System = systemOf(hostShelf, n.HostVersion)
m.Libc = libcOf(m.System)
reported, err := inv.DescribedOf(ctx, n.Name)
if err != nil {
return snapshot.Facts{}, err
}
m.Architecture, m.Kernel = reported.Architecture, reported.Kernel
capabilities, err := inv.Profile(ctx, n.Name)
if err != nil {
return snapshot.Facts{}, err
}
for _, c := range capabilities {
kept := snapshot.Capability{Name: c.Name, Present: c.Present}
// The detail only where it is a version: everything else a detector says — a ruleset, a
// device, a path — is the machine's own business and no check reads it.
if c.Present && (c.Name == "container-runtime" || c.Name == "package-manager") {
kept.Detail = scrub.Text(c.Detail)
}
m.Capabilities = append(m.Capabilities, kept)
}
if o, ok := place[n.Name]; ok {
m.Site, m.Hub, m.Public, m.OnNetwork = scrub.Site(o.Site), o.Hub, o.Endpoint != "", o.Address != ""
}
assigned, err := inv.Assigned(ctx, n.Name)
if err != nil {
return snapshot.Facts{}, err
}
m.Assigned = assigned
sent, known, err := inv.SentBuilds(ctx, n.Name)
if err != nil {
return snapshot.Facts{}, err
}
if known && slices.Contains(assigned, broker.RuntimeModule) {
m.NodeTools = sent[broker.RuntimeModule]
}
pins, err := inv.PinsFor(ctx, n.Name)
if err != nil {
return snapshot.Facts{}, err
}
for provision, c := range pins {
m.Pins = append(m.Pins, snapshot.Pin{Provision: provision, Machine: scrub.Machine(c.Node), Module: c.Module})
}
for _, module := range assigned {
held, err := inv.SecretsOf(ctx, n.Name, module)
if err != nil {
return snapshot.Facts{}, err
}
for _, h := range held {
if h.Origin == inventory.OriginAccepted {
m.Accepted = append(m.Accepted, snapshot.Accepted{Module: module, Name: h.Name,
Provider: scrub.Machine(h.Provider), Local: h.Local})
}
}
layers, err := inv.SettingsFor(ctx, n.Name, module)
if err != nil {
return snapshot.Facts{}, err
}
for _, layer := range layers {
if layer.From == catalogue.MeshWideLayer {
if !meshWide[module] {
meshWide[module] = true
f.Settings = append(f.Settings, snapshot.Settings{Module: module, Values: scrub.Values(layer.Values)})
}
continue
}
m.Settings = append(m.Settings, snapshot.Settings{Module: module, Values: scrub.Values(layer.Values)})
}
}
m.Declaration = declarationFacts(ctx, open, n.Name, gens, gensErr, scrub)
if ctx.Err() != nil {
return snapshot.Facts{}, ctx.Err()
}
f.Machines = append(f.Machines, m)
}
for e := range engines {
f.Versions.NodeEngines = append(f.Versions.NodeEngines, e)
}
// Seats and their holders, and from them the roles a machine is named by.
roles := map[string][]string{}
seats := map[string]*snapshot.Seat{}
for _, h := range holdings {
key := h.Claim + "\x00" + h.Scope
s, ok := seats[key]
if !ok {
s = &snapshot.Seat{Name: h.Claim, Scope: h.Scope}
seats[key] = s
}
s.Holders = append(s.Holders, snapshot.Holder{Machine: scrub.Machine(h.Node), Module: h.Module})
if h.Scope == catalogue.ScopeMesh {
role := "holds " + h.Claim
if h.Claim == catalogue.ControllerSeatName {
role = "the control node"
}
roles[h.Node] = append(roles[h.Node], role)
}
}
for _, s := range seats {
f.Seats = append(f.Seats, *s)
}
for i := range f.Machines {
for _, n := range nodes {
if scrub.Machine(n.Name) != f.Machines[i].Name {
continue
}
f.Machines[i].Roles = roles[n.Name]
if f.Machines[i].Hub {
f.Machines[i].Roles = append(f.Machines[i].Roles, "the hub")
}
}
}
// Every module, as the mesh holds it, and where it is built from.
newest := map[string]inventory.Source{}
count := map[string]int{}
for _, e := range entries {
raw, err := json.Marshal(e.Manifest)
if err != nil {
return snapshot.Facts{}, err
}
mod := snapshot.Module{Name: e.Manifest.Module, Repository: e.Source.Repository, Path: e.Source.Path,
Commit: e.Source.BuiltFrom, Provided: e.Provided, RollOut: current[e.Manifest.Module].RollOut,
Manifest: raw}
for _, r := range read[e.Manifest.Module] {
mod.Reads = append(mod.Reads, r.Repository)
}
f.Modules = append(f.Modules, mod)
if e.Provided || e.Source.Repository == "" {
continue
}
count[e.Source.Repository]++
if was, ok := newest[e.Source.Repository]; !ok || e.Source.Seen.After(was.Seen) {
newest[e.Source.Repository] = e.Source
}
}
for repository, s := range newest {
commit := s.Head
if commit == "" {
commit = s.BuiltFrom
}
f.Sources = append(f.Sources, snapshot.Source{Repository: repository, Commit: commit, Modules: count[repository]})
}
for _, e := range edges {
f.Edges = append(f.Edges, snapshot.Edge{From: e.From, To: e.To, Kind: e.Kind})
}
f.Sorted()
return f, nil
}
// declarationFacts is how one machine's declaration composes now, as the next push would compose it and
// without making anything (D1's composition), and whether the node-engine's validator takes it.
func declarationFacts(ctx context.Context, open *stores, node string, gens map[string]catalogue.Generator,
gensErr error, scrub *snapshot.Scrubber) snapshot.Declaration {
var d snapshot.Declaration
if gensErr != nil {
d.Problems = []string{scrub.Text("the private network cannot be computed: " + oneLine(gensErr.Error()))}
return d
}
declared, problems, err := composedAndValidated(ctx, open, node, gens, Foreseeing)
if err != nil {
d.Problems = []string{scrub.Text(oneLine(err.Error()))}
return d
}
for _, p := range problems {
d.Problems = append(d.Problems, scrub.Text(p))
}
d.Composes = len(problems) == 0
if body, err := declared.Body(); err == nil {
d.Digest = fmt.Sprintf("sha256:%x", sha256.Sum256(body))
}
d.Resources = resourceNames(declared.Resources)
for module, why := range declared.leftOutWhy {
if d.LeftOut == nil {
d.LeftOut = map[string]string{}
}
d.LeftOut[module] = scrub.Text(why)
}
for _, o := range declared.withheld {
d.Withheld = append(d.Withheld, scrub.Text(o.String()))
}
for _, u := range declared.unbound {
d.Unbound = append(d.Unbound, scrub.Text(u.String()))
}
sort.Strings(d.Withheld)
sort.Strings(d.Unbound)
return d
}
// composedAndValidated composes one machine's declaration — as a push would (Allocating) or as the next
// push will without making anything (Foreseeing) — with the order it was last sent, and runs the
// node-engine's own validator over the body. An error is that it did not compose; problems are what the
// validator refuses.
func composedAndValidated(ctx context.Context, open *stores, node string, gens map[string]catalogue.Generator,
choosing Choosing) (sendable, []string, error) {
plan, settings, err := planFor(ctx, open, node)
if err != nil {
return sendable{}, nil, err
}
declared, err := declarationWith(ctx, open, node, plan, settings, gens, choosing)
if err != nil {
return sendable{}, nil, err
}
record, err := open.inventory.NodeByName(ctx, node)
if err != nil {
return sendable{}, nil, err
}
if declared.Sequence, err = open.inventory.Sequence(ctx, record.ID); err != nil {
return sendable{}, nil, err
}
if declared.Epoch, err = open.inventory.SentEpoch(ctx, record.ID); err != nil {
return sendable{}, nil, err
}
body, err := declared.Body()
if err != nil {
return sendable{}, nil, err
}
return declared, validate.Declaration(body), nil
}
// resourceNames are a declaration's resources as `type:id`, sorted.
func resourceNames(resources []map[string]any) []string {
out := make([]string, 0, len(resources))
for _, r := range resources {
kind, _ := r["type"].(string)
id, _ := r["id"].(string)
out = append(out, kind+":"+id)
}
sort.Strings(out)
return out
}
// systemOf is the system a node-engine of that version was built for, read from the build the mesh
// holds: the artifact a resource delivered into `versions/<version>` came from is named for its system
// (`host-arch`). Empty when the version is not a delivered one — an engine placed by hand.
func systemOf(host catalogue.Manifest, version string) string {
if version == "" {
return ""
}
for _, r := range host.Resources {
path, _ := r["path"].(string)
if !strings.HasSuffix(path, "/versions/"+version) {
continue
}
source, _ := r["source"].(string)
for _, part := range strings.Split(source, "/") {
if system, ok := strings.CutPrefix(part, "host-"); ok && system != "" {
return system
}
}
}
return ""
}
// libcOf is the C library of a system the node-engine is built for.
func libcOf(system string) string {
switch system {
case "arch":
return "glibc"
case "alpine":
return "musl"
case "android":
return "bionic"
}
return ""
}
// factsCommand is `facts`: what the controller keeps, and keeping it now.
//
// facts the snapshot the artifact store holds: when, which, how many machines
// facts show the same, whole, as JSON
// facts export compose and keep one now
// facts compose compose one and print it, keeping nothing
func factsCommand(ctx context.Context, args []string) error {
set := flag.NewFlagSet("facts", flag.ContinueOnError)
rest, err := parseAround(set, args)
if err != nil {
return err
}
what := ""
if len(rest) > 0 {
what = rest[0]
}
open, err := openStores(ctx)
if err != nil {
return err
}
defer open.Close()
switch what {
case "", "show":
address, err := factsStore(ctx, open)
if err != nil {
return err
}
body, digest, err := (artifacts.Store{Address: address}).GetTagged(ctx, snapshot.Repository, snapshot.Tag)
if err != nil {
return err
}
if what == "show" {
_, err := os.Stdout.Write(body)
return err
}
f, err := snapshot.Decode(body)
if err != nil {
return err
}
fmt.Printf("the facts snapshot kept as %s:%s is %s, taken %s (%s ago) by controller %s\n",
snapshot.Repository, snapshot.Tag, short(strings.TrimPrefix(digest, "sha256:")),
f.Taken.Format(time.RFC3339), ago(time.Since(f.Taken)), orNone(f.Controller.Commit))
fmt.Printf(" %d machine(s), %d module(s), %d seat(s); the longest machine name is %d characters\n",
len(f.Machines), len(f.Modules), len(f.Seats), f.Longest())
fmt.Printf(" the bus runs %s, the store %s\n", orNone(f.Versions.Bus), orNone(f.Versions.Store))
for _, m := range f.Machines {
state := "composes"
if !m.Declaration.Composes {
state = "does NOT compose: " + strings.Join(m.Declaration.Problems, "; ")
}
fmt.Printf(" %-12s %s; %d module(s); %s\n", m.Name, m.Described(), len(m.Assigned), state)
}
return nil
case "export", "compose":
busVersion := ""
if server, err := connectLink(ctx, nil, nil, nil); err == nil {
busVersion = busVersionOf(server)
server.Close()
}
if what == "compose" {
f, err := gatherFacts(ctx, open, busVersion)
if err != nil {
return err
}
body, err := f.Encode()
if err != nil {
return err
}
_, err = os.Stdout.Write(append(body, '\n'))
return err
}
digest, err := exportFacts(ctx, open, busVersion, true)
if err != nil {
return err
}
fmt.Printf("the facts snapshot is kept as %s:%s, %s\n", snapshot.Repository, snapshot.Tag, digest)
return nil
}
return fmt.Errorf("facts [show|export|compose], not %q", what)
}
// busVersionOf is the bus server's release, as it told this connection.
func busVersionOf(server *link.Server) string {
if bus, ok := server.Bus().(link.OverNATS); ok && bus.Conn != nil {
return bus.Conn.ConnectedServerVersion()
}
return ""
}
+137
View File
@@ -0,0 +1,137 @@
package main
import (
"regexp"
"strings"
"testing"
"github.com/novox/mesh-controller/internal/catalogue"
snapshot "github.com/novox/mesh-controller/internal/facts"
)
// The facts snapshot (novox/hq to-be 45 §9): composed from the store, every machine under a pseudonym
// of its name's length, and nothing of the installation in it — no secret, no address, no name.
// aMeshWithSecrets is aMesh with a provider and its consumers, a value given by hand, settings carrying
// a password, an address and a machine's name, and a push's worth of credentials made.
func aMeshWithSecrets(t *testing.T) (*stores, []string) {
t.Helper()
open := aMesh(t)
ctx := t.Context()
register(t, open, catalogue.Manifest{Module: "objects", Version: "1",
Provides: []catalogue.Offer{{Name: "s3-bucket", Scope: catalogue.ScopeMesh,
Identity: &catalogue.OfferIdentity{Max: 20, In: "an S3 access key"}}},
Receives: map[string]string{"s3-bucket": "/var/lib/mesh/objects/mesh.json"}})
register(t, open, catalogue.Manifest{Module: "files", Version: "1", Requires: []string{"s3-bucket"},
Resources: []map[string]any{{"id": "config", "type": "file", "path": "/etc/files/config.json",
"mode": "0600", "content": "{}", "merge": "json"}}})
for _, a := range [][2]string{{"anchor", "objects"}, {"laptop", "files"}} {
if _, err := assign(ctx, open, a[0], a[1]); err != nil {
t.Fatalf("assign %s %s: %v", a[0], a[1], err)
}
}
secrets := []string{"Hunter2-Is-Not-A-Password-9f8e7d", "0123456789abcdefABCDEF0123456789zz"}
if err := open.inventory.SetSettings(ctx, "", "files", map[string]any{
"admin_password": secrets[0], "upstream": "10.77.0.9", "hub": "anchor"}); err != nil {
t.Fatal(err)
}
if err := open.inventory.SetSettings(ctx, "laptop", "files", map[string]any{
"note": "reach me at 192.168.1.135, token " + secrets[1]}); err != nil {
t.Fatal(err)
}
// A push's worth of composition, which makes the pair credential the consumer is sent.
gens, err := generators(ctx, open)
if err != nil {
t.Fatal(err)
}
for _, node := range []string{"laptop", "anchor"} {
if _, _, err := composedAndValidated(ctx, open, node, gens, Allocating); err != nil {
t.Fatalf("%s does not compose: %v", node, err)
}
}
issued, err := open.inventory.SecretsFrom(ctx, "anchor")
if err != nil || len(issued) == 0 {
t.Fatalf("no credential was made for the consumer: %v", err)
}
for _, s := range issued {
// Sealed, never kept plain (ADR 0004): the sealed blobs are what the store holds, and none may leave.
secrets = append(secrets, s.ForConsumer, s.ForProvider)
}
return open, secrets
}
func TestTheFactsCarryNoSecretNoAddressAndNoName(t *testing.T) {
open, secrets := aMeshWithSecrets(t)
f, err := gatherFacts(t.Context(), open, "2.11.17")
if err != nil {
t.Fatal(err)
}
body, err := f.Encode()
if err != nil {
t.Fatal(err)
}
text := string(body)
for _, s := range secrets {
if s != "" && strings.Contains(text, s) {
t.Errorf("a secret is in the snapshot: %q", s)
}
}
for _, leaked := range []string{"anchor", "laptop", "10.77.0.", "192.168.1.135", ".example:51820"} {
if strings.Contains(text, leaked) {
t.Errorf("%q is in the snapshot", leaked)
}
}
// Every address it carries is a documentation address.
for _, a := range regexp.MustCompile(`\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b`).FindAllString(text, -1) {
if !strings.HasPrefix(a, "192.0.2.") && !strings.HasPrefix(a, "198.51.100.") && !strings.HasPrefix(a, "203.0.113.") {
t.Errorf("%s is an address outside the documentation ranges", a)
}
}
// What a check needs is there: every machine, its length, its modules, its declaration composing.
if len(f.Machines) != 2 || f.Longest() != len("laptop") {
t.Fatalf("machines %+v, longest %d", f.Machines, f.Longest())
}
anchor := snapshot.Pseudonym("machine", "anchor")
m, ok := f.Machine(anchor)
if !ok || !m.Hub || !strings.Contains(m.Described(), "the hub") || len(m.Name) != len("anchor") {
t.Fatalf("the anchor reads as %+v", m)
}
if !m.Declaration.Composes || m.Declaration.Digest == "" || len(m.Declaration.Resources) == 0 {
t.Errorf("the anchor's declaration reads as %+v", m.Declaration)
}
laptop, _ := f.Machine(snapshot.Pseudonym("machine", "laptop"))
if strings.Join(laptop.Assigned, ",") != "files,mesh-wireguard" && !strings.Contains(strings.Join(laptop.Assigned, ","), "files") {
t.Errorf("the laptop's assignments read as %v", laptop.Assigned)
}
var meshWide map[string]any
for _, s := range f.Settings {
if s.Module == "files" {
meshWide = s.Values
}
}
if meshWide["admin_password"] != snapshot.Withheld || meshWide["hub"] != anchor {
t.Errorf("the mesh-wide settings read as %v", meshWide)
}
if f.Versions.Bus != "2.11.17" || f.Versions.Store == "" {
t.Errorf("versions read as %+v", f.Versions)
}
var objects bool
for _, mod := range f.Modules {
objects = objects || mod.Name == "objects" && len(mod.Manifest) > 0
}
if !objects {
t.Error("the modules the mesh holds are not in the snapshot")
}
// And an unchanged mesh is the same content a moment later.
again, err := gatherFacts(t.Context(), open, "2.11.17")
if err != nil {
t.Fatal(err)
}
a, _ := f.Content()
b, _ := again.Content()
if a != b {
t.Error("two snapshots of an unchanged mesh differ, so it would be written again every ten minutes")
}
}
+3
View File
@@ -114,6 +114,9 @@ func run() error {
return brokerCommand(ctx, args[1:])
case "serve":
return serve(ctx)
// The facts snapshot a merge check is fed (novox/hq to-be 45 §9).
case "facts":
return factsCommand(ctx, args[1:])
case "upgrade":
return upgradeCommand(ctx, args[1:])
// The bus as a planned step (novox/hq to-be 45 §8, ADR 0236).
+3
View File
@@ -212,6 +212,9 @@ func serve(ctx context.Context) (err error) {
givenEvents = bus
// Composed now and kept current, before the verb that answers from it is served.
go statusFrom.keep(ctx)
// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
// controller holds the lease, read by the build seat from the artifact store.
go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
// Every call carries the lease's epoch, and its record is written only under the lease (novox/hq
// to-be 45 §6).
link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) })
+30 -3
View File
@@ -183,9 +183,11 @@ var signalsTable = []signalRow{
newest: func(f *signalFacts) time.Time {
return newestOf(f.staleRefusals, func(w link.WriterRefusals) time.Time { return w.Last })
}},
{Row: "S14", Signal: "facts snapshot exported", Emitter: "controller", Trigger: "daily",
Bound: "2 days", Kind: "facts-stale", Severity: conditions.Warning, Phase: 5,
Deferred: "the facts snapshot is built in Phase 5 (to-be 45 §9): nothing exports one yet"},
{Row: "S14", Signal: "facts snapshot exported", Emitter: "controller", Trigger: "when it moved, and daily",
Bound: "2 days: a snapshot older than that, or none kept since this controller began two days ago",
Kind: "facts-stale", Severity: conditions.Warning, Phase: 5,
needs: func(*signalFacts) error { return nil }, watch: watchFacts,
newest: func(f *signalFacts) time.Time { return f.facts.taken }},
{Row: "S15", Signal: "a hand act with a cause already recorded", Emitter: "hand-act log",
Trigger: "each act", Bound: "the second within 14 days; clears when fewer than two remain within 14 days",
Kind: "healer-wanted", Severity: conditions.Warning, Phase: 3,
@@ -195,6 +197,31 @@ var signalsTable = []signalRow{
}},
}
// watchFacts is S14: the snapshot a merge check is fed is older than its bound, or none was kept since
// this controller began that long ago (novox/hq to-be 45 §9). A check fed a stale snapshot judges a change
// against a mesh that no longer is, which is the fault the snapshot exists to end.
func watchFacts(f *signalFacts) []conditions.Observation {
since := f.facts.taken
if since.IsZero() {
since = f.facts.began
}
if since.IsZero() || f.now.Sub(since) <= factsStaleAfter {
return nil
}
said := "none kept since this controller began " + ago(f.now.Sub(since)) + " ago"
if !f.facts.taken.IsZero() {
said = "the newest kept was taken " + ago(f.now.Sub(f.facts.taken)) + " ago"
}
if f.facts.err != nil {
said += "; the last attempt: " + oneLine(f.facts.err.Error())
}
return []conditions.Observation{{Scope: conditions.ScopeCore, ID: "facts", Kind: "facts-stale", Token: "stale",
Severity: conditions.Warning,
Summary: fmt.Sprintf("the facts snapshot merge checks are fed is stale (bound %s): a change is judged against a "+
"mesh that no longer is", ago(factsStaleAfter)),
Said: said}}
}
// newestOf is the newest time among things.
func newestOf[T any](list []T, at func(T) time.Time) time.Time {
var newest time.Time
+6
View File
@@ -37,6 +37,7 @@ func calm(now time.Time) *signalFacts {
selfCheck: selfCheckFacts{last: now.Add(-time.Minute), every: 5 * time.Minute},
lostConsumers: map[string]bool{}, epochs: map[int64]inventory.Epoch{},
lease: leaseFacts{held: true, epoch: 57, renewed: now.Add(-2 * time.Second)},
facts: factsFacts{taken: now.Add(-time.Hour), began: now.Add(-time.Hour)},
}
}
@@ -125,6 +126,11 @@ var suppressions = map[string]suppression{
inside: func(f *signalFacts) { f.staleRefusals = refusedBy(f.now, 41, 5) },
past: func(f *signalFacts) { f.staleRefusals = refusedBy(f.now, 41, 6) },
},
// The snapshot a merge check is fed, older than two days; or none kept by a controller two days up.
"S14": {
inside: func(f *signalFacts) { f.facts.taken = f.now.Add(-47 * time.Hour) },
past: func(f *signalFacts) { f.facts.taken = f.now.Add(-49 * time.Hour) },
},
// Twice by hand within a fortnight is a healer wanted; once, or the first of two a day too old, is not.
"S15": {
inside: func(f *signalFacts) {
+11
View File
@@ -89,6 +89,16 @@ type signalFacts struct {
// lease is this controller's standing to the lease, and the epochs that ended lately (S12).
lease leaseFacts
leaseErr error
// facts is the snapshot this controller keeps for merge checks (S14).
facts factsFacts
}
// factsFacts is when the newest facts snapshot was taken, when this controller began keeping it, and
// the last attempt's error.
type factsFacts struct {
taken, began time.Time
err error
}
type leaseFacts struct {
@@ -313,6 +323,7 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
f.advisories = link.Advisories.Since(now.Add(-advisoryQuiet))
f.lostConsumers, f.advisoriesErr = w.lostConsumers(ctx, f.advisories)
f.handActs, f.handActsErr = w.gatherHandActs(ctx, now)
f.facts.taken, _, f.facts.began, f.facts.err = exportedFacts.last()
return f
}
+179
View File
@@ -0,0 +1,179 @@
package artifacts
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"strings"
)
// A document the mesh keeps under a name, read by whoever asks for that name (novox/hq to-be 45 §9).
//
// **The one thing the mesh names by tag.** Everything a machine runs is pinned by digest (ADR 0189), and
// holders are put untagged so the collector's own rule keeps them. The facts snapshot is the opposite
// case: what a reader wants is *the newest*, never a particular one, and a tag is the store's own word
// for that. So it is put as the smallest OCI manifest naming one layer, under a tag; putting the next
// moves the tag, and **the manifest it replaced is deleted by its digest**: the store's nightly collector
// keeps every manifest, tagged or not, and marks what each names — so a replaced snapshot left in place
// would be kept for ever, one more every day. Deleted, its layer is named by nothing and the next
// collection takes it: the store keeps the newest, and nothing grows.
// MaxTagged is the largest document put or read this way: a snapshot of a large mesh is a few
// megabytes, and a reader is never made to swallow an answer of any size.
const MaxTagged = 64 << 20
// ErrNoTag is the answer when the store holds nothing under the tag: never put, or collected.
var ErrNoTag = errors.New("the artifact store holds nothing under that name")
// PutTagged puts body as the only layer of a manifest in repository and points tag at it. Answers the
// layer's digest — what a reader quotes as "the snapshot I read".
func (s Store) PutTagged(ctx context.Context, repository, tag, mediaType string, body []byte) (string, error) {
if s.Address == "" {
return "", fmt.Errorf("this mesh has no artifact store on its network to keep %s:%s in", repository, tag)
}
if len(body) > MaxTagged {
return "", fmt.Errorf("%s:%s is %d bytes, over the %d the store is given", repository, tag, len(body), MaxTagged)
}
sum := sha256.Sum256(body)
digest := "sha256:" + hex.EncodeToString(sum[:])
// What the tag names now, so it can be let go of once the new one stands. Asked before the put:
// after it, the tag names the new one.
replaced, err := s.manifestDigest(ctx, repository, tag)
if err != nil {
return "", err
}
if err := s.putBlob(ctx, repository, digest, body); err != nil {
return "", err
}
if err := s.putBlob(ctx, repository, emptyDigest, emptyConfig); err != nil {
return "", err
}
manifest, err := json.Marshal(holderManifest{
SchemaVersion: 2,
MediaType: mediaManifest,
Config: descriptor{MediaType: mediaEmpty, Digest: emptyDigest, Size: int64(len(emptyConfig))},
Layers: []descriptor{{MediaType: mediaType, Digest: digest, Size: int64(len(body))}},
})
if err != nil {
return "", err
}
request, err := http.NewRequestWithContext(ctx, http.MethodPut, s.url(repository, "manifests", tag),
bytes.NewReader(manifest))
if err != nil {
return "", err
}
request.Header.Set("Content-Type", mediaManifest)
response, err := s.client().Do(request)
if err != nil {
return "", fmt.Errorf("cannot reach the artifact store at %s: %w", s.Address, err)
}
defer response.Body.Close()
if response.StatusCode != http.StatusCreated {
said, _ := io.ReadAll(io.LimitReader(response.Body, 4096))
return "", fmt.Errorf("the artifact store refused %s:%s: %s %s", repository, tag, response.Status,
strings.TrimSpace(string(said)))
}
mSum := sha256.Sum256(manifest)
if put := "sha256:" + hex.EncodeToString(mSum[:]); replaced != "" && replaced != put {
// The new one stands; the old one is let go of. A refusal here leaves one more snapshot in the
// store, which is said and is not a failure of the put: the tag already names the new one.
if err := s.remove(ctx, s.url(repository, "manifests", replaced), repository+"/manifests/"+replaced); err != nil &&
err != Gone {
return digest, fmt.Errorf("%s:%s now names the new document, and the one it replaced could not be "+
"let go of: %w", repository, tag, err)
}
}
return digest, nil
}
// manifestDigest is the digest of the manifest tag names in repository; empty when it names none.
func (s Store) manifestDigest(ctx context.Context, repository, tag string) (string, error) {
request, err := http.NewRequestWithContext(ctx, http.MethodHead, s.url(repository, "manifests", tag), nil)
if err != nil {
return "", err
}
for _, media := range manifestAccept {
request.Header.Add("Accept", media)
}
response, err := s.client().Do(request)
if err != nil {
return "", fmt.Errorf("cannot reach the artifact store at %s: %w", s.Address, err)
}
defer response.Body.Close()
switch response.StatusCode {
case http.StatusOK:
return response.Header.Get("Docker-Content-Digest"), nil
case http.StatusNotFound:
return "", nil
default:
return "", fmt.Errorf("the artifact store answered %s for %s:%s", response.Status, repository, tag)
}
}
// GetTagged reads the one layer of the manifest tag names in repository, and its digest. ErrNoTag when
// the store holds nothing under it.
func (s Store) GetTagged(ctx context.Context, repository, tag string) ([]byte, string, error) {
if s.Address == "" {
return nil, "", fmt.Errorf("this mesh has no artifact store on its network to read %s:%s from", repository, tag)
}
request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url(repository, "manifests", tag), nil)
if err != nil {
return nil, "", err
}
for _, media := range manifestAccept {
request.Header.Add("Accept", media)
}
response, err := s.client().Do(request)
if err != nil {
return nil, "", fmt.Errorf("cannot reach the artifact store at %s: %w", s.Address, err)
}
defer response.Body.Close()
switch response.StatusCode {
case http.StatusOK:
case http.StatusNotFound:
return nil, "", fmt.Errorf("%w: %s:%s", ErrNoTag, repository, tag)
default:
return nil, "", fmt.Errorf("the artifact store answered %s for %s:%s", response.Status, repository, tag)
}
var m holderManifest
if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&m); err != nil {
return nil, "", fmt.Errorf("%s:%s is not a manifest the mesh wrote: %w", repository, tag, err)
}
if len(m.Layers) != 1 {
return nil, "", fmt.Errorf("%s:%s names %d layers; the mesh writes one", repository, tag, len(m.Layers))
}
layer := m.Layers[0]
if layer.Size > MaxTagged {
return nil, "", fmt.Errorf("%s:%s is %d bytes, over the %d a reader takes", repository, tag, layer.Size, MaxTagged)
}
blob, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url(repository, "blobs", layer.Digest), nil)
if err != nil {
return nil, "", err
}
got, err := s.client().Do(blob)
if err != nil {
return nil, "", fmt.Errorf("cannot reach the artifact store at %s: %w", s.Address, err)
}
defer got.Body.Close()
if got.StatusCode != http.StatusOK {
return nil, "", fmt.Errorf("the artifact store names %s for %s:%s and answered %s for it",
layer.Digest, repository, tag, got.Status)
}
body, err := io.ReadAll(io.LimitReader(got.Body, MaxTagged+1))
if err != nil {
return nil, "", err
}
// **Read back against its own digest**: a reader is told which snapshot it read, and a truncated or
// substituted body must not pass as that one.
sum := sha256.Sum256(body)
if "sha256:"+hex.EncodeToString(sum[:]) != layer.Digest {
return nil, "", fmt.Errorf("%s:%s read back as something other than %s", repository, tag, layer.Digest)
}
return body, layer.Digest, nil
}
+88
View File
@@ -0,0 +1,88 @@
package artifacts
import (
"context"
"errors"
"fmt"
"os"
"os/exec"
"strings"
"testing"
"time"
)
// The facts snapshot is put under a tag and read back by it (novox/hq to-be 45 §9).
//
// Against the very registry the mesh's store runs, because what is asserted is the registry's answer:
// that a manifest put by tag is read by tag, that the layer comes back whole and checked, and that the
// snapshot a newer one replaced is the collector's to take while the newest is kept. Raised as the
// collector test above says, with MESH_TEST_REGISTRY and MESH_TEST_REGISTRY_CONTAINER.
func TestLiveADocumentPutUnderATagIsReadBackAndOnlyTheNewestIsKept(t *testing.T) {
address := os.Getenv("MESH_TEST_REGISTRY")
container := os.Getenv("MESH_TEST_REGISTRY_CONTAINER")
if address == "" || container == "" {
t.Skip("no MESH_TEST_REGISTRY / MESH_TEST_REGISTRY_CONTAINER; see the collector test's comment for the registry to raise")
}
ctx := context.Background()
store := Store{Address: address}
repository := fmt.Sprintf("facts-live-%d", time.Now().UnixNano())
if _, _, err := store.GetTagged(ctx, repository, "latest"); !errors.Is(err, ErrNoTag) {
t.Fatalf("nothing was put and the read said %v, not that nothing is there", err)
}
first := []byte(`{"facts":1,"taken":"first"}`)
firstDigest, err := store.PutTagged(ctx, repository, "latest", "application/vnd.novox.mesh.facts.v1+json", first)
if err != nil {
t.Fatal(err)
}
second := []byte(`{"facts":1,"taken":"second"}`)
secondDigest, err := store.PutTagged(ctx, repository, "latest", "application/vnd.novox.mesh.facts.v1+json", second)
if err != nil {
t.Fatal(err)
}
got, digest, err := store.GetTagged(ctx, repository, "latest")
if err != nil {
t.Fatal(err)
}
if string(got) != string(second) || digest != secondDigest {
t.Fatalf("read back %q (%s), not the newest put %q (%s)", got, digest, second, secondDigest)
}
// The collector, as the store's nightly step runs it — with no `--delete-untagged`: the newest kept,
// the replaced one taken because its manifest was let go of.
out, err := exec.Command("docker", "exec", container, "registry", "garbage-collect",
"/etc/docker/registry/config.yml").CombinedOutput()
if err != nil {
t.Fatalf("the collector did not run: %v\n%s", err, out)
}
if got, _, err := store.GetTagged(ctx, repository, "latest"); err != nil || string(got) != string(second) {
t.Fatalf("after collection the newest reads %q, %v", got, err)
}
// On the store's disk, not as the running server answers: it caches blob descriptors in memory.
onDisk := func(digest string) bool {
hex := strings.TrimPrefix(digest, "sha256:")
path := "/var/lib/registry/docker/registry/v2/blobs/sha256/" + hex[:2] + "/" + hex + "/data"
return exec.Command("docker", "exec", container, "test", "-f", path).Run() == nil
}
if onDisk(firstDigest) {
t.Errorf("the replaced snapshot %s is still in the store after collection; nothing would ever take it", firstDigest)
}
if !onDisk(secondDigest) {
t.Errorf("the collector took the newest snapshot %s", secondDigest)
}
if !strings.HasPrefix(secondDigest, "sha256:") {
t.Errorf("the digest said %q", secondDigest)
}
}
// A store with no address is said, not dialled.
func TestADocumentWithNowhereToGoIsRefusedByName(t *testing.T) {
if _, err := (Store{}).PutTagged(context.Background(), "facts", "latest", "x", []byte("{}")); err == nil ||
!strings.Contains(err.Error(), "no artifact store") {
t.Errorf("put with no store said %v", err)
}
if _, _, err := (Store{}).GetTagged(context.Background(), "facts", "latest"); err == nil ||
!strings.Contains(err.Error(), "no artifact store") {
t.Errorf("read with no store said %v", err)
}
}
+334
View File
@@ -0,0 +1,334 @@
// Package facts is the mesh's facts snapshot: what a check needs to judge a change against the mesh
// that runs, and nothing it could leak (novox/hq to-be 45 §9, ADR 0227 rule 9).
//
// **Why it exists.** Every check the mesh had was right about the world it was given, and none was
// given the mesh's world: a module passed every test and refused the anchor's whole declaration because
// a real machine's name made its identity 23 characters (issue 263); a manifest passed the catalogue
// check and was refused by the node-engine (issue 236); a resolver answer that glibc forgave was final
// to musl (issue 262). The controller holds those facts. It writes them here, the build seat reads them,
// and a merge check composes every machine of the snapshot with the change applied.
//
// **What it holds, and what it never holds.** Every machine — under a stable pseudonym of the same length
// as its name, because the length is what a name limit meets — with its roles, system, C library,
// architecture, builds, what it reported it can do, what is assigned there, its pins and settings;
// every seat and its holders; every module the mesh holds, as its manifest; the sources the mesh built
// them from; the versions of the bus, the store and the node-engine it runs; and how each machine's
// declaration composes today. **No secret and no address**: a setting whose key or value reads as a
// secret is withheld, an address is replaced by one from a documentation range, and a machine's name, a
// site, an account and a public domain are replaced wherever they appear. So a snapshot can be copied
// into a test, a replay or a pull request without carrying anything of the installation it came from.
package facts
import (
"crypto/sha256"
"encoding/json"
"fmt"
"sort"
"time"
)
// Format is the snapshot's version. A reader refuses one newer than it knows: a field it cannot read
// is a fact it would judge without.
const Format = 1
// Repository and Tag are where the controller keeps the newest snapshot in the artifact store, and
// MediaType what it is kept as.
const (
Repository = "facts"
Tag = "latest"
MediaType = "application/vnd.novox.mesh.facts.v1+json"
)
// Facts is one snapshot.
type Facts struct {
Format int `json:"facts"`
// Taken is when the controller composed it.
Taken time.Time `json:"taken"`
// Controller is the controller build that composed it — the one the mesh runs, which is the one a
// change to the catalogue must be readable by (version skew).
Controller Build `json:"controller"`
// Versions are what the mesh runs of the things its tests stand in for.
Versions Versions `json:"versions"`
// Sources are the repositories the mesh builds modules from, each with the newest commit it built.
Sources []Source `json:"sources"`
// Machines, in name order of their pseudonyms.
Machines []Machine `json:"machines"`
// Seats are every seat held, with its holders.
Seats []Seat `json:"seats"`
// Modules are every module the mesh holds, as it holds them.
Modules []Module `json:"modules"`
// Settings are the mesh-wide layer of every module's settings, scrubbed.
Settings []Settings `json:"settings,omitempty"`
// Edges are the build dependencies between modules, as the mesh recorded them — what a merge's
// rebuild width is computed from.
Edges []Edge `json:"edges,omitempty"`
}
// Build names one build of a core component.
type Build struct {
Version string `json:"version,omitempty"`
Commit string `json:"commit,omitempty"`
}
// Versions are what the mesh runs.
type Versions struct {
// Bus is the bus server's release, as the server tells a client that connects.
Bus string `json:"bus,omitempty"`
// Store is the store's server version, as the store answers it.
Store string `json:"store,omitempty"`
// NodeEngines are the node-engine builds the machines report, each once.
NodeEngines []string `json:"node-engines,omitempty"`
}
// Source is a repository the mesh builds from, and the newest commit it built a module of.
type Source struct {
Repository string `json:"repository"`
Commit string `json:"commit,omitempty"`
Modules int `json:"modules"`
}
// Machine is one machine, under its pseudonym.
type Machine struct {
Name string `json:"name"`
// Length is the length of its real name, which the pseudonym keeps.
Length int `json:"length"`
// Roles are what it is to the mesh, in words: the control node, the hub, the bus's machine, a
// holder of a mesh seat. What a check names when it fails one.
Roles []string `json:"roles,omitempty"`
// System is the node-engine's system (arch, alpine, …), Libc its C library, as far as the mesh knows.
System string `json:"system,omitempty"`
Libc string `json:"libc,omitempty"`
Architecture string `json:"architecture,omitempty"`
Kernel string `json:"kernel,omitempty"`
// NodeEngine and NodeTools are the builds it runs.
NodeEngine string `json:"node-engine,omitempty"`
NodeTools string `json:"node-tools,omitempty"`
// Capabilities are what it reported it can do; the detail kept only where it is a version.
Capabilities []Capability `json:"capabilities,omitempty"`
// Site, Hub, Public: where it is on the private network — its site (a pseudonym), whether it is the
// hub, whether others can dial it. No address.
Site string `json:"site,omitempty"`
Hub bool `json:"hub,omitempty"`
Public bool `json:"public,omitempty"`
// OnNetwork is whether it has a place on the private network at all.
OnNetwork bool `json:"on-network,omitempty"`
// Adopted is whether the mesh adopted it rather than converged it.
Adopted bool `json:"adopted,omitempty"`
// Account is the operator's login there (a pseudonym of the same length), and AccountHome where its
// home is when that is not the derived one.
Account string `json:"account,omitempty"`
AccountHome string `json:"account-home,omitempty"`
// PublicDomain is the domain it answers for, its labels replaced.
PublicDomain string `json:"public-domain,omitempty"`
// Assigned is every module assigned there.
Assigned []string `json:"assigned,omitempty"`
// Pins are where it was told a provision comes from.
Pins []Pin `json:"pins,omitempty"`
// Settings are its own layer of each module's settings, scrubbed.
Settings []Settings `json:"settings,omitempty"`
// Accepted are the secrets a person gave the mesh for modules here, by name only: the mesh cannot
// make them, so a check composing this machine gives each a stand-in instead of refusing it.
Accepted []Accepted `json:"accepted,omitempty"`
// Declaration is how its declaration composes today, as the controller that took this saw it.
Declaration Declaration `json:"declaration"`
}
// Capability is one thing a machine reported it can or cannot do.
type Capability struct {
Name string `json:"name"`
Present bool `json:"present"`
Detail string `json:"detail,omitempty"`
}
// Pin is one provision a machine was told where to take from.
type Pin struct {
Provision string `json:"provision"`
Machine string `json:"machine"`
Module string `json:"module,omitempty"`
}
// Accepted is one secret a person gave: the module's own when Provider is empty, else the credential it
// takes from that machine's provider under Local.
type Accepted struct {
Module string `json:"module"`
Name string `json:"name"`
Provider string `json:"provider,omitempty"`
Local string `json:"local,omitempty"`
}
// Settings is one layer of one module's settings.
type Settings struct {
Module string `json:"module"`
Values map[string]any `json:"values"`
}
// Declaration is how one machine's declaration composed when the snapshot was taken.
type Declaration struct {
// Composes is whether it composed and the node-engine's validator took it.
Composes bool `json:"composes"`
// Digest is its body's digest — composed as the next push would, so it moves with what the mesh
// would send, and is not the digest of what was last sent.
Digest string `json:"digest,omitempty"`
// Resources are what it declares, as `type:id`, sorted.
Resources []string `json:"resources,omitempty"`
// Problems are why it does not compose or validate, scrubbed.
Problems []string `json:"problems,omitempty"`
// LeftOut are the modules assigned there and left out of it, each with why.
LeftOut map[string]string `json:"left-out,omitempty"`
// Withheld are the consumers its grants leave out because an identity overflows the provision's
// bound (ADR 0225), and Unbound the credentials on record for consumers bound elsewhere (issue 274),
// each said in words.
Withheld []string `json:"withheld,omitempty"`
Unbound []string `json:"unbound,omitempty"`
}
// Seat is one seat and the machines holding it.
type Seat struct {
Name string `json:"name"`
Scope string `json:"scope"`
Holders []Holder `json:"holders"`
}
// Holder is one module on one machine holding a seat.
type Holder struct {
Machine string `json:"machine"`
Module string `json:"module"`
}
// Module is one module the mesh holds.
type Module struct {
Name string `json:"name"`
// Repository and Path are where it is built from; empty for one that came with the controller.
Repository string `json:"repository,omitempty"`
Path string `json:"path,omitempty"`
// Commit is the commit the manifest the mesh holds was read at.
Commit string `json:"commit,omitempty"`
// Provided is a module that came with the controller rather than from a repository.
Provided bool `json:"provided,omitempty"`
// RollOut is its upgrade policy: rolled out when built, or recorded.
RollOut bool `json:"roll-out,omitempty"`
// Reads are the other repositories its build read source from.
Reads []string `json:"reads,omitempty"`
// Manifest is the module as the mesh holds it: artifacts resolved to the builds it runs.
Manifest json.RawMessage `json:"manifest"`
}
// Edge is one build dependency: From is built standing on To.
type Edge struct {
From string `json:"from"`
To string `json:"to"`
Kind string `json:"kind,omitempty"`
}
// Sorted puts every list in a fixed order, so two snapshots of one mesh are the same bytes apart from
// when they were taken.
func (f *Facts) Sorted() {
sort.Slice(f.Machines, func(i, j int) bool { return f.Machines[i].Name < f.Machines[j].Name })
for i := range f.Machines {
m := &f.Machines[i]
sort.Strings(m.Roles)
sort.Strings(m.Assigned)
sort.Slice(m.Capabilities, func(a, b int) bool { return m.Capabilities[a].Name < m.Capabilities[b].Name })
sort.Slice(m.Pins, func(a, b int) bool { return m.Pins[a].Provision < m.Pins[b].Provision })
sort.Slice(m.Settings, func(a, b int) bool { return m.Settings[a].Module < m.Settings[b].Module })
sort.Slice(m.Accepted, func(a, b int) bool {
x, y := m.Accepted[a], m.Accepted[b]
return x.Module+"\x00"+x.Name+"\x00"+x.Provider+"\x00"+x.Local < y.Module+"\x00"+y.Name+"\x00"+y.Provider+"\x00"+y.Local
})
sort.Strings(m.Declaration.Resources)
}
sort.Slice(f.Seats, func(i, j int) bool {
if f.Seats[i].Name != f.Seats[j].Name {
return f.Seats[i].Name < f.Seats[j].Name
}
return f.Seats[i].Scope < f.Seats[j].Scope
})
for i := range f.Seats {
h := f.Seats[i].Holders
sort.Slice(h, func(a, b int) bool {
if h[a].Machine != h[b].Machine {
return h[a].Machine < h[b].Machine
}
return h[a].Module < h[b].Module
})
}
sort.Slice(f.Modules, func(i, j int) bool { return f.Modules[i].Name < f.Modules[j].Name })
sort.Slice(f.Sources, func(i, j int) bool { return f.Sources[i].Repository < f.Sources[j].Repository })
sort.Slice(f.Settings, func(i, j int) bool { return f.Settings[i].Module < f.Settings[j].Module })
sort.Slice(f.Edges, func(i, j int) bool {
if f.Edges[i].From != f.Edges[j].From {
return f.Edges[i].From < f.Edges[j].From
}
return f.Edges[i].To < f.Edges[j].To
})
sort.Strings(f.Versions.NodeEngines)
}
// Encode is the snapshot as it is kept: sorted, indented, so a person can read the one the build seat
// read and a diff of two says what moved.
func (f Facts) Encode() ([]byte, error) {
f.Sorted()
return json.MarshalIndent(f, "", " ")
}
// Content is the digest of everything but when it was taken: two snapshots of an unchanged mesh have
// the same content, which is how the controller tells a change from another day.
func (f Facts) Content() (string, error) {
f.Taken = time.Time{}
body, err := f.Encode()
if err != nil {
return "", err
}
sum := sha256.Sum256(body)
return fmt.Sprintf("sha256:%x", sum), nil
}
// Decode reads a snapshot, refusing one newer than this reader knows and one that is not a snapshot.
func Decode(body []byte) (Facts, error) {
var f Facts
if err := json.Unmarshal(body, &f); err != nil {
return Facts{}, fmt.Errorf("not a facts snapshot: %w", err)
}
switch {
case f.Format == 0:
return Facts{}, fmt.Errorf("not a facts snapshot: it says no format")
case f.Format > Format:
return Facts{}, fmt.Errorf("a facts snapshot of format %d, and this reads format %d: a newer controller "+
"took it, and what it says beyond %d would be judged without", f.Format, Format, Format)
}
return f, nil
}
// Longest is the length of the longest machine name in the snapshot — what a consumer's identity is
// judged on (novox/hq ADR 0225).
func (f Facts) Longest() int {
longest := 0
for _, m := range f.Machines {
if m.Length > longest {
longest = m.Length
}
}
return longest
}
// Machine is the machine of that pseudonym.
func (f Facts) Machine(name string) (Machine, bool) {
for _, m := range f.Machines {
if m.Name == name {
return m, true
}
}
return Machine{}, false
}
// Described is how a check names a machine: its roles, then its pseudonym.
func (m Machine) Described() string {
if len(m.Roles) == 0 {
return "a machine (" + m.Name + ")"
}
out := m.Roles[0]
for _, r := range m.Roles[1:] {
out += ", " + r
}
return out + " (" + m.Name + ")"
}
+159
View File
@@ -0,0 +1,159 @@
package facts
import (
"fmt"
"net"
"regexp"
"strings"
"testing"
"time"
)
// A pseudonym keeps what a limit meets — the length, and letters where letters were — and nothing else.
func TestAPseudonymKeepsTheLengthAndShapeAndIsStable(t *testing.T) {
for _, name := range []string{"ace", "g14", "novox", "shanks", "home-server", "a"} {
p := Pseudonym("machine", name)
if len(p) != len(name) {
t.Errorf("%s became %s: %d characters for %d", name, p, len(p), len(name))
}
if p == name {
t.Errorf("%s was kept as itself", name)
}
if p != Pseudonym("machine", name) {
t.Errorf("%s is not stable", name)
}
for i := range name {
isLetter := func(c byte) bool { return c >= 'a' && c <= 'z' }
isDigit := func(c byte) bool { return c >= '0' && c <= '9' }
if isLetter(name[i]) != isLetter(p[i]) || isDigit(name[i]) != isDigit(p[i]) {
t.Errorf("%s became %s: the shape moved at %d", name, p, i)
}
}
}
if Pseudonym("machine", "ace") == Pseudonym("site", "ace") {
t.Error("a site and a machine of one name share a pseudonym, so a snapshot says they are one thing")
}
}
// Nothing of the installation survives the scrubber: names, domains, accounts, addresses, mail, secrets.
func TestTheScrubberLeavesNothingOfTheInstallation(t *testing.T) {
s := NewScrubber()
machine := s.Machine("homeserver")
s.Account("jochens")
domain := s.Domain("zurag.be")
if len(domain) != len("zurag.be") || !strings.HasSuffix(domain, ".be") || domain == "zurag.be" {
t.Errorf("the domain became %q", domain)
}
in := map[string]any{
"hub": "homeserver",
"listen": "10.42.0.7:51820",
"upstream": []any{"192.168.1.135", "fd00::1"},
"site": "https://grafana.zurag.be/login",
"admin": "jschoubben@gmail.com",
"home": "/home/jochens/.ssh",
"api_key": "sk-live-abcdef",
"nested": map[string]any{"password": "hunter2", "port": float64(5432)},
"opaque": "a8F3kQ9zL2mX7vB4nC6dE1rT5yU0iO8pA3sD",
"dsn": "postgres://app:s3cr3tpass@db.internal:5432/app",
"pem": "-----BEGIN PRIVATE KEY-----\nMIIB",
"plain": "a-long-plain-module-directory-name",
"digest": "sha256:44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a",
"version": "2.11.17",
"homeserver": true,
}
out := s.Values(in)
flat := flatten(out)
for _, leaked := range []string{"homeserver\"", "10.42.0.7", "192.168.1.135", "fd00::1", "zurag", "jschoubben",
"gmail", "jochens", "sk-live", "hunter2", "a8F3kQ9zL2mX7vB4nC6dE1rT5yU0iO8pA3sD", "s3cr3tpass", "MIIB"} {
if strings.Contains(flat, leaked) {
t.Errorf("%q survived the scrubber:\n%s", leaked, flat)
}
}
if out["hub"] != machine {
t.Errorf("a machine named in a setting became %v, not its pseudonym %s", out["hub"], machine)
}
for _, kept := range []string{"a-long-plain-module-directory-name", "2.11.17", "sha256:44136fa3", "5432"} {
if !strings.Contains(flat, kept) {
t.Errorf("%q was scrubbed, and it is not the installation's:\n%s", kept, flat)
}
}
// Every address left is a documentation address.
for _, a := range regexp.MustCompile(`[0-9a-f:.]{7,}`).FindAllString(flat, -1) {
ip := net.ParseIP(strings.Trim(a, ".:"))
if ip == nil {
continue
}
doc := false
for _, cidr := range []string{"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24", "2001:db8::/32"} {
_, n, _ := net.ParseCIDR(cidr)
doc = doc || n.Contains(ip)
}
if !doc {
t.Errorf("%s is not a documentation address", a)
}
}
// The same address is always the same stand-in.
if s.Text("10.42.0.7") != s.Text("at 10.42.0.7")[3:] {
t.Error("one address became two stand-ins")
}
}
func flatten(v any) string {
var b strings.Builder
var walk func(any)
walk = func(v any) {
switch t := v.(type) {
case map[string]any:
for k, e := range t {
b.WriteString(k + "\"=")
walk(e)
b.WriteString("\n")
}
case []any:
for _, e := range t {
walk(e)
b.WriteString(",")
}
case string:
b.WriteString(t + "\"")
default:
b.WriteString(fmt.Sprint(t))
}
}
walk(v)
return b.String()
}
// Two snapshots of one mesh, taken apart, are one content.
func TestASnapshotOfAnUnchangedMeshIsTheSameContentAnotherDay(t *testing.T) {
f := Facts{Format: Format, Taken: time.Now(), Machines: []Machine{{Name: "b"}, {Name: "a"}}}
g := f
g.Taken = f.Taken.Add(24 * time.Hour)
g.Machines = []Machine{{Name: "a"}, {Name: "b"}}
a, err := f.Content()
if err != nil {
t.Fatal(err)
}
b, _ := g.Content()
if a != b {
t.Error("the same mesh a day later reads as a change")
}
g.Machines = append(g.Machines, Machine{Name: "c"})
if c, _ := g.Content(); c == a {
t.Error("a machine added reads as no change")
}
}
// A reader refuses a snapshot it cannot read whole.
func TestANewerSnapshotIsRefusedNotHalfRead(t *testing.T) {
if _, err := Decode([]byte(`{"facts": 99}`)); err == nil || !strings.Contains(err.Error(), "newer controller") {
t.Errorf("a newer format read as %v", err)
}
if _, err := Decode([]byte(`{"machines": []}`)); err == nil {
t.Error("a document with no format read as a snapshot")
}
f, err := Decode([]byte(`{"facts": 1, "machines": [{"name": "abc", "length": 3}, {"name": "defgh", "length": 5}]}`))
if err != nil || f.Longest() != 5 {
t.Errorf("read %+v, %v", f, err)
}
}
+317
View File
@@ -0,0 +1,317 @@
package facts
import (
"crypto/sha256"
"fmt"
"net"
"regexp"
"sort"
"strings"
)
// Withheld is what a value that reads as a secret becomes. A check composing with it gets a stand-in of
// the right kind (a string), never the value.
const Withheld = "withheld"
// Pseudonym is the stable stand-in for a name: the same length, letters for letters and digits for
// digits, anything else kept where it was. Stable, so two snapshots of one mesh name a machine alike and
// a check's verdict can be compared across them; derived from the name alone, so it needs no key to be
// kept anywhere.
//
// It hides nothing from somebody who can guess the names: that is not its purpose. Its purpose is that a
// snapshot — and every fixture, replay or pull-request comment made from one — carries no name of the
// installation it came from, while every length a limit meets stays the length it was.
func Pseudonym(kind, name string) string {
sum := sha256.Sum256([]byte("novox-mesh-facts\x00" + kind + "\x00" + name))
out := []byte(name)
for i, c := range out {
b := sum[i%len(sum)] ^ byte(i/len(sum))
switch {
case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z':
out[i] = 'a' + b%26
case c >= '0' && c <= '9':
out[i] = '0' + b%10
}
}
// A name must still read as one: a machine's begins with a letter.
if len(out) > 0 && (out[0] < 'a' || out[0] > 'z') && name[0] >= 'a' && name[0] <= 'z' {
out[0] = 'a' + sum[0]%26
}
return string(out)
}
// Scrubber replaces what a snapshot must not carry, wherever it appears in text.
type Scrubber struct {
// replace is every real word and what it becomes, longest first, so a domain is replaced before a
// label inside it.
replace [][2]string
seen map[string]bool
}
// NewScrubber knows the installation's names: machines, sites, accounts, public domains.
func NewScrubber() *Scrubber { return &Scrubber{seen: map[string]bool{}} }
// Machine registers a machine's name and answers its pseudonym.
func (s *Scrubber) Machine(name string) string { return s.word("machine", name) }
// Site registers a site's name and answers its pseudonym.
func (s *Scrubber) Site(name string) string { return s.word("site", name) }
// Account registers an account's name and answers its pseudonym.
func (s *Scrubber) Account(name string) string { return s.word("account", name) }
// Domain registers a public domain and answers its stand-in: each label but the last replaced, so the
// shape and every length stay.
func (s *Scrubber) Domain(domain string) string {
if domain == "" {
return ""
}
labels := strings.Split(domain, ".")
for i := range labels {
if i == len(labels)-1 && len(labels) > 1 {
break
}
labels[i] = Pseudonym("domain", labels[i])
}
out := strings.Join(labels, ".")
s.add(domain, out)
return out
}
func (s *Scrubber) word(kind, name string) string {
if name == "" {
return ""
}
out := Pseudonym(kind, name)
s.add(name, out)
return out
}
func (s *Scrubber) add(from, to string) {
if from == "" || s.seen[from] {
return
}
s.seen[from] = true
s.replace = append(s.replace, [2]string{from, to})
sort.SliceStable(s.replace, func(i, j int) bool { return len(s.replace[i][0]) > len(s.replace[j][0]) })
}
// wordBoundary is what may stand beside a name for it to be that name and not part of another word.
func wordBoundary(c byte) bool {
return !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '_')
}
// Text is s with every known name replaced, every address put in a documentation range, every address
// of mail replaced, and every run that reads as a secret withheld.
func (s *Scrubber) Text(text string) string {
if text == "" {
return text
}
text = secretRuns(text)
text = emails.ReplaceAllStringFunc(text, func(mail string) string {
if strings.HasPrefix(mail, Withheld+"@") {
return mail // a URL's withheld credentials, not an address of mail
}
return "someone@example.org"
})
text = addresses(text)
for _, r := range s.replace {
text = replaceWord(text, r[0], r[1])
}
return text
}
// replaceWord replaces from where it stands as a word of its own.
func replaceWord(text, from, to string) string {
var b strings.Builder
for {
i := strings.Index(text, from)
if i < 0 {
b.WriteString(text)
return b.String()
}
end := i + len(from)
if (i == 0 || wordBoundary(text[i-1])) && (end == len(text) || wordBoundary(text[end])) {
b.WriteString(text[:i])
b.WriteString(to)
} else {
b.WriteString(text[:end])
}
text = text[end:]
}
}
// Values is a settings layer with every key that names a secret withheld, and every string scrubbed.
func (s *Scrubber) Values(values map[string]any) map[string]any {
out := make(map[string]any, len(values))
for k, v := range values {
// A key may itself be a name — a map of machines to something.
key := s.Text(k)
if secretKey.MatchString(k) {
out[key] = withheldLike(v)
continue
}
out[key] = s.value(v)
}
return out
}
func (s *Scrubber) value(v any) any {
switch t := v.(type) {
case string:
return s.Text(t)
case map[string]any:
return s.Values(t)
case []any:
out := make([]any, len(t))
for i, e := range t {
out[i] = s.value(e)
}
return out
default:
return v
}
}
// withheldLike is a stand-in of the same kind: a string for a string, a list for a list, so a check
// composing a setting still finds the shape it expects.
func withheldLike(v any) any {
switch t := v.(type) {
case nil:
return nil
case bool, float64, int, int64:
return t
case []any:
out := make([]any, len(t))
for i, e := range t {
out[i] = withheldLike(e)
}
return out
case map[string]any:
out := map[string]any{}
for k, e := range t {
out[k] = withheldLike(e)
}
return out
default:
return Withheld
}
}
// secretKey is a setting's key that names a secret.
var secretKey = regexp.MustCompile(`(?i)(secret|passw|token|api[-_]?key|private[-_]?key|credential|bearer|cookie|salt|signing)`)
// emails are addresses of mail: a person's name, or an installation's domain, in either half.
var emails = regexp.MustCompile(`[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}`)
// secretRun is a run of characters a key, a token or a hash is made of, long enough to be one.
var secretRun = regexp.MustCompile(`[A-Za-z0-9+/=_\-]{24,}`)
// userinfo is the credentials part of a URL.
var userinfo = regexp.MustCompile(`(://)[^/@\s:]+:[^/@\s]+@`)
// secretRuns withholds a URL's credentials and every run that reads as a key: long, and mixing letters
// with digits or symbols. A long plain word (a path, a module's name) is left alone.
func secretRuns(text string) string {
if strings.Contains(text, "-----BEGIN") {
return Withheld
}
text = userinfo.ReplaceAllString(text, "${1}"+Withheld+"@")
var b strings.Builder
last := 0
for _, at := range secretRun.FindAllStringIndex(text, -1) {
run := text[at[0]:at[1]]
b.WriteString(text[last:at[0]])
last = at[1]
// A digest names an artifact, not a secret.
if strings.HasSuffix(text[:at[0]], "sha256:") {
b.WriteString(run)
continue
}
b.WriteString(judgeRun(run))
}
b.WriteString(text[last:])
return b.String()
}
// judgeRun is a run withheld when it reads as a key: long, and mixing letters with digits or symbols.
func judgeRun(run string) string {
{
var lower, upper, digit, symbol bool
for _, c := range run {
switch {
case c >= 'a' && c <= 'z':
lower = true
case c >= 'A' && c <= 'Z':
upper = true
case c >= '0' && c <= '9':
digit = true
default:
symbol = true
}
}
classes := 0
for _, b := range []bool{lower, upper, digit, symbol} {
if b {
classes++
}
}
// A sha256 digest names an artifact, not a secret, and a dashed word is a name.
if strings.HasPrefix(run, "sha256") || (!digit && !upper) {
return run
}
if classes >= 3 || (digit && (lower || upper) && len(run) >= 32) {
return Withheld
}
return run
}
}
// addresses puts every IP address in text into a documentation range (RFC 5737, RFC 3849): the same
// address always becomes the same stand-in, so two settings naming one machine still name one.
func addresses(text string) string {
var b strings.Builder
i := 0
for i < len(text) {
if !addressChar(text[i]) {
b.WriteByte(text[i])
i++
continue
}
j := i
for j < len(text) && addressChar(text[j]) {
j++
}
b.WriteString(standIn(text[i:j]))
i = j
}
return b.String()
}
func addressChar(c byte) bool {
return c >= '0' && c <= '9' || c >= 'a' && c <= 'f' || c >= 'A' && c <= 'F' || c == '.' || c == ':'
}
// standIn is the documentation address for a run that is an address, or the run itself.
func standIn(run string) string {
trimmed := strings.TrimRight(run, ".:")
tail := run[len(trimmed):]
ip := net.ParseIP(trimmed)
switch {
case ip == nil:
// A port after an IPv4 address reads as part of the run: try without it.
if host, port, ok := strings.Cut(trimmed, ":"); ok && strings.Count(trimmed, ":") == 1 &&
net.ParseIP(host) != nil && strings.Contains(host, ".") {
return standIn(host) + ":" + port + tail
}
return run
case ip.To4() != nil && strings.Contains(trimmed, "."):
sum := sha256.Sum256([]byte("v4\x00" + trimmed))
ranges := []string{"192.0.2", "198.51.100", "203.0.113"}
return fmt.Sprintf("%s.%d", ranges[sum[0]%3], 1+sum[1]%254) + tail
case strings.Count(trimmed, ":") >= 2:
sum := sha256.Sum256([]byte("v6\x00" + trimmed))
return fmt.Sprintf("2001:db8::%x:%x", uint16(sum[0])<<8|uint16(sum[1]), uint16(sum[2])<<8|uint16(sum[3])) + tail
}
return run
}
+42
View File
@@ -0,0 +1,42 @@
package inventory
import (
"context"
"encoding/json"
"errors"
"fmt"
"github.com/jackc/pgx/v5"
)
// Described is what a machine said about itself beyond its capabilities: the architecture and kernel
// its node-engine runs on (novox/hq to-be 45 §9, the facts snapshot).
type Described struct {
Architecture string `json:"architecture"`
Kernel string `json:"kernel"`
}
// DescribedOf is the architecture and kernel a machine last reported; empty for one that never has.
func (i *Inventory) DescribedOf(ctx context.Context, nodeName string) (Described, error) {
var raw []byte
err := i.store.Pool().QueryRow(ctx, `select profile from node where name = $1`, nodeName).Scan(&raw)
if errors.Is(err, pgx.ErrNoRows) {
return Described{}, fmt.Errorf("%w: %s", ErrNoSuchNode, nodeName)
}
if err != nil || len(raw) == 0 {
return Described{}, err
}
var r Described
if err := json.Unmarshal(raw, &r); err != nil {
return Described{}, err
}
return r, nil
}
// ServerVersion is the store's own word for the release it runs — the version a test suite standing in
// for it must run (novox/hq ADR 0227 rule 9).
func (i *Inventory) ServerVersion(ctx context.Context) (string, error) {
var v string
err := i.store.Pool().QueryRow(ctx, `show server_version`).Scan(&v)
return v, err
}