From 3f02bd7ff6b8781d4b92315543f97d8a0c2b0040 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 21:50:49 +0100 Subject: [PATCH 1/2] =?UTF-8?q?The=20bus=20node=20writes=20the=20bus's=20u?= =?UTF-8?q?ser=20list=20alone=20from=20the=20newest=20composition=20of=20t?= =?UTF-8?q?he=20bus=20module,=20with=20no=20want,=20and=20says=20it=20in?= =?UTF-8?q?=20the=20bus=20module's=20have=20(ADR=200547=20=C2=A78,=20issue?= =?UTF-8?q?=201945)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The list reached the bus only through a want of the bus module, which nobody writes (ADR 0539 §15): on 2026-10-25 it stood from 11:23 to 20:50 and was put back by hand. The node agent now follows its composition bucket for the newest signed composition carrying the user list, writes that file alone through the apply's file step when it differs (mode 0600, read back, recorded as the module's), moves nothing that reads it (the bus module's entrypoint reloads the server in place), never goes back to an older list, looks again after every apply of the bus module, and says source, revision, wanted and written digests, users and any error in the bus module's have (mesh-sdk UserListWritten). Read with mesh-delivery-view.node_haves (module nats); the bus's own nats_user_can says in_force once it reloaded. --- cmd/mesh-host/nodeagent.go | 12 ++ cmd/mesh-host/userlist.go | 82 +++++++++ go.mod | 2 +- go.sum | 4 + internal/apply/alone.go | 67 +++++++ internal/apply/alone_test.go | 90 ++++++++++ internal/apply/filled_test.go | 58 +++---- internal/link/have.go | 24 +++ internal/link/have_test.go | 12 ++ internal/userlist/userlist.go | 270 +++++++++++++++++++++++++++++ internal/userlist/userlist_test.go | 222 ++++++++++++++++++++++++ 11 files changed, 813 insertions(+), 30 deletions(-) create mode 100644 cmd/mesh-host/userlist.go create mode 100644 internal/apply/alone.go create mode 100644 internal/apply/alone_test.go create mode 100644 internal/userlist/userlist.go create mode 100644 internal/userlist/userlist_test.go diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 63bb2f1..2a7f037 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -391,6 +391,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess } var ag *agent.Agent kept := keptOnce(opts, signer) + // The bus's user list, written alone from the newest composition of the bus module (ADR 0547 §8): on every + // node, and a node whose compositions carry no list writes nothing. + users := userListFollower(opts, session, composedBucket, trustedKeys(opts, signer), say) + go func() { + if err := users.Run(linked); err != nil && linked.Err() == nil { + say("the bus's user list is not followed on this link: " + err.Error()) + } + }() env := agent.Env{ Node: node, Wants: bucketWatch{session, delivery.AssignerWantsBucket}, @@ -418,6 +426,10 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess actx, done := applyContext(session.WithLink(ctx)) res := applyComposition(actx, opts, c, others, sched, say) done() + // An apply of the bus module may write an older list than the newest composition holds: looked at again. + if module, _ := users.State(); module != "" && module == c.Module { + users.Again() + } afterApply() sayFacts(ctx, ag) return res diff --git a/cmd/mesh-host/userlist.go b/cmd/mesh-host/userlist.go new file mode 100644 index 0000000..370d1ea --- /dev/null +++ b/cmd/mesh-host/userlist.go @@ -0,0 +1,82 @@ +package main + +import ( + "context" + "crypto/ed25519" + "encoding/json" + "errors" + "fmt" + "os" + + "git.novox.be/novox/mesh-sdk/go/declaration" + "github.com/novox/mesh-host/internal/apply" + "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/store" + "github.com/novox/mesh-host/internal/system" + "github.com/novox/mesh-host/internal/userlist" +) + +// The bridge of novox/hq ADR 0547 §8, wired: the bus node's engine writes the bus's user list alone from the newest +// composition of the bus module, under the apply's two locks as a filled file is written (filling.go), and says what +// it wrote in the bus module's have. + +// userListFollower is the follower of the bus's user list for one link: the node's composition bucket, the keys the +// agent trusts, the write alone, and the have. +func userListFollower(opts options, session *link.Session, composedBucket string, trusted func() []ed25519.PublicKey, + say link.Announce) *userlist.Follower { + return userlist.New(userlist.Env{ + Node: session.Membership.Node, + Composed: bucketWatch{session, composedBucket}, + Trusted: trusted, + Write: func(ctx context.Context, module string, raw json.RawMessage) (string, bool, error) { + return writeUserListAlone(session.WithLink(ctx), opts, raw, say) + }, + Report: func(ctx context.Context, module string, w userlist.Written) { + session.NoteUserList(ctx, module, w) + }, + Say: announceOr(say), + }) +} + +// writeUserListAlone writes one user-list resource alone (apply.WriteAlone): its content had by digest when it is +// given so, the node's state read and saved under the apply's locks, and the digest of the file on the machine after. +func writeUserListAlone(ctx context.Context, opts options, raw []byte, say link.Announce) (string, bool, error) { + r, err := declaration.ParseResource(raw) + if err != nil { + return "", false, fmt.Errorf("the user list is not a resource: %w", err) + } + if _, ok := r.(*declaration.File); !ok { + return "", false, fmt.Errorf("the user list is a %s, not a file", r.Kind()) + } + d := &declaration.Declaration{Version: declaration.Version, Resources: []declaration.Resource{r}} + if err := apply.ResolveFileReferences(ctx, d, link.FilesFrom(ctx)); err != nil { + return "", false, fmt.Errorf("the user list could not be had by its digest: %w", err) + } + file := d.Resources[0].(*declaration.File) + + applying.Lock() + defer applying.Unlock() + unlock, err := store.Lock(opts.state, nil) + if err != nil { + return "", false, err + } + defer unlock() + known, err := store.Load(opts.state) + if err != nil { + return "", false, err + } + built, err := system.For(builtFor) + if err != nil { + return "", false, err + } + out, updated, writeErr := apply.WriteAlone(ctx, built, file, known, apply.ExecRunner, announceOr(say)) + // Saved whichever way it went, as an apply's: what was written is on the machine either way. + if err := store.Save(opts.state, updated); err != nil { + writeErr = errors.Join(writeErr, errors.New("written, and the node's state could not be saved: "+err.Error())) + } + digest := "" + if body, err := os.ReadFile(file.Path); err == nil { + digest = userlist.Digest(body) + } + return digest, writeErr == nil && out.Action != "unchanged", writeErr +} diff --git a/go.mod b/go.mod index f4302b0..c100297 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204730-0be0ffd6159e github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 diff --git a/go.sum b/go.sum index c03eccc..f843500 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,9 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f h1:lUEyNPPF3w4Q0LIgcUUD6ofkKVhCF0n7dVjMoixQmXo= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204213-f63fa86c7456 h1:He2o9vSKjgKbWOCOgOJY0WlaDh1n8RcGhvGY/dwXsBQ= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204213-f63fa86c7456/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204730-0be0ffd6159e h1:k/1NjjW5HuuuHW2ldyWTvjp+1u3fN0RO18Q57Iup6Dk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204730-0be0ffd6159e/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/internal/apply/alone.go b/internal/apply/alone.go new file mode 100644 index 0000000..c4275dc --- /dev/null +++ b/internal/apply/alone.go @@ -0,0 +1,67 @@ +package apply + +import ( + "context" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/declaration" + "github.com/novox/mesh-host/internal/store" + "github.com/novox/mesh-host/internal/system" +) + +// WriteAlone writes one file a module's composition declares, through the apply's own file step, and records it as an +// apply would; nothing else of the module is applied, and nothing that reads the file is moved (novox/hq ADR 0547 §8). +// +// **For a file whose reader takes the change itself.** The bus's user list is the one it is for: the bus module's +// entrypoint watches the file and signals the server to reload in place, so the list is in force without a want of the +// bus, which nobody writes (ADR 0539 §15), and without the container being touched. A file that a service or a container +// must be moved to read is not written alone: it goes through the module's apply, which moves what reads it. +// +// The file is written only when its content differs from what is there (the file step's own rule), read back, and +// recorded under the module whose resource it is, so the module's next apply finds it as its own and writes nothing. +// The caller holds the apply's lock. log is said the path, the action and the digest, never the content. +func WriteAlone(ctx context.Context, sys system.System, file *declaration.File, known store.State, run Runner, + log func(string)) (Outcome, store.State, error) { + if log == nil { + log = func(string) {} + } + if file.Into != "" || file.CreateOnce || file.Sealed != "" || file.VaultSealed != nil || len(file.Secrets) > 0 { + return Outcome{}, known, &Error{Resource: file.ID, Err: fmt.Errorf("%s is not a whole file of plain content, "+ + "and only such a file is written alone; nothing was written", file.Path)} + } + if file.IsReference() { + return Outcome{}, known, &Error{Resource: file.ID, Err: fmt.Errorf("%s is given by digest %s and was not "+ + "resolved to its content; it is not written empty", file.Path, file.Digest)} + } + if held, isHeld := known.HeldAt(file.ID); isHeld { + return Outcome{}, known, &Error{Resource: file.ID, Err: fmt.Errorf("%s is held as found for a module not "+ + "yet taken (%s), so nothing was written", file.Path, held.Why)} + } + was, _ := known.Find(file.ID) + if was.ID == "" { + // Only over a file the mesh wrote: a file this node holds no record of is the module's apply's to take, with + // the original kept, never this write's. A record of the path under another id (the declaration before the + // switch named it so) is the one written to, so no second record of the file is left for an apply to remove. + rec, ok := known.At(string(declaration.TypeFile), file.Path) + if !ok { + return Outcome{}, known, &Error{Resource: file.ID, Err: fmt.Errorf("this node holds no record of %s; it "+ + "is written by the module's apply first, never alone", file.Path)} + } + renamed := *file + renamed.ID = rec.ID + file, was = &renamed, rec + } + in := inputs{declares: map[string]string{file.Identity(): declaredDigest(file)}, known: &known, + groups: groupsWanted([]declaration.Resource{file}), handed: Handed{}, dirsBefore: map[string]bool{}, + files: map[string]bool{file.Path: true}} + outcome, err := applyOne(ctx, sys, file, run, map[string]bool{}, in, was, nil, nil) + if err != nil { + log(fmt.Sprintf(" failed %s (%s), alone: %v", file.ID, file.Path, err)) + return outcome, known, &Error{Resource: file.ID, Err: err} + } + recordKeeping(&known, file, outcome, was) + if moved(outcome) { + log(fmt.Sprintf(" %s %s (%s), alone: %s", outcome.Action, file.ID, file.Path, outcome.wrote)) + } + return outcome, known, nil +} diff --git a/internal/apply/alone_test.go b/internal/apply/alone_test.go new file mode 100644 index 0000000..a2a3815 --- /dev/null +++ b/internal/apply/alone_test.go @@ -0,0 +1,90 @@ +package apply + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/declaration" + "github.com/novox/mesh-host/internal/store" +) + +// Defends novox/hq ADR 0547 §8: the bus node writes the bus's user list alone, with no want, through the apply's own +// file step: only when it differs, mode 0600, read back and recorded as the module's, and nothing else moves. +func TestAFileWrittenAloneMovesNothingElse(t *testing.T) { + const was, now = "user a\n", "user a\nuser b\n" + for _, c := range []struct { + name, fields, before string + recorded bool + recordedAs string // the id the record has, when not the composition's + action, refused string + after string + }{ + {name: "no record of it", fields: `"content":%q,"mode":"0600"`, before: was, refused: "holds no record", after: was}, + {name: "changed", fields: `"content":%q,"mode":"0600"`, before: was, recorded: true, action: "updated", after: now}, + {name: "the same", fields: `"content":%q,"mode":"0600"`, before: now, recorded: true, action: "unchanged", after: now}, + {name: "recorded under the id from before the switch", fields: `"content":%q,"mode":"0600"`, before: was, + recorded: true, recordedAs: "bus-users", action: "updated", after: now}, + {name: "absent with a record", fields: `"content":%q,"mode":"0600"`, recorded: true, action: "created", after: now}, + {name: "written into", fields: `"content":%q,"into":"block","owner-at-runtime":"managed"`, before: was, + recorded: true, refused: "only such a file", after: was}, + {name: "sealed", fields: `"sealed":"c2VhbGVk"`, before: was, recorded: true, + refused: "only such a file", after: was}, + } { + t.Run(c.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "accounts.conf") + if c.before != "" { + if err := os.WriteFile(path, []byte(c.before), 0o600); err != nil { + t.Fatal(err) + } + } + fields := strings.Replace(c.fields, "%q", fmt.Sprintf("%q", now), 1) + d := parseTrusted(t, fmt.Sprintf(`{"declaration":1,"resources":[{"id":"nats.bus-users","module":"nats", + "type":"file","path":%q,`+fields+`}]}`, path)) + file := d.Resources[0].(*declaration.File) + known := store.State{} + id := file.ID + if c.recordedAs != "" { + id = c.recordedAs + } + if c.recorded { + known.Record(store.Applied{ID: id, Type: string(declaration.TypeFile), Module: "nats", Target: path, + Origin: store.OriginDeclared, Wrote: digestOf(c.before)}) + } + calls := 0 + counting := func(context.Context, string, ...string) (string, error) { calls++; return "", nil } + out, after, err := WriteAlone(context.Background(), archHost(t), file, known, counting, nil) + if c.refused != "" { + if err == nil || !strings.Contains(err.Error(), c.refused) { + t.Fatalf("not refused with %q: %v", c.refused, err) + } + } else if err != nil { + t.Fatal(err) + } else if out.Action != c.action { + t.Errorf("action %q, want %q", out.Action, c.action) + } + if got := readText(t, path); got != c.after { + t.Errorf("file holds %q, want %q", got, c.after) + } + if calls != 0 { + t.Errorf("%d commands were run: a file written alone moves nothing", calls) + } + if c.refused != "" { + return + } + if m := modeAt(t, path); m != 0o600 { + t.Errorf("mode %o", m) + } + rec, ok := after.Find(id) + if !ok || rec.Wrote != digestOf(now) || rec.Target != path { + t.Errorf("record %+v (found %v), want it to say %s was written", rec, ok, digestOf(now)) + } + if n := len(after.Resources); n != 1 { + t.Errorf("%d records of one file", n) + } + }) + } +} diff --git a/internal/apply/filled_test.go b/internal/apply/filled_test.go index 3d5fbd3..ebaecad 100644 --- a/internal/apply/filled_test.go +++ b/internal/apply/filled_test.go @@ -52,6 +52,24 @@ func TestAWriteOfAPathNotFilledIsRefused(t *testing.T) { if _, err := os.Stat(other); err == nil { t.Errorf("%s was written", other) } + t.Run("a filled path two resources declare", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "hosts") + if err := os.WriteFile(path, []byte(workstationHosts), 0o644); err != nil { + t.Fatal(err) + } + d := parseTrusted(t, fmt.Sprintf(`{"declaration":1,"filled_files":[%q],"resources":[ + {"id":"a.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.1 ace\n","owner-at-runtime":"managed"}, + {"id":"b.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.2 g14\n","owner-at-runtime":"managed"} + ]}`, path, path, path)) + _, state, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("6.6.6.6 evil\n"), + noServices, nil, nil, nil) + if err == nil || !strings.Contains(err.Error(), "2 resources declare") { + t.Fatalf("a path two resources declare was not refused: %v", err) + } + if got := readText(t, path); got != workstationHosts || len(state.Resources) != 0 { + t.Errorf("something was written or recorded: %q %+v", got, state.Resources) + } + }) } // Each ownership is written by the write verb exactly as an apply of the declared file with the same content: the @@ -112,6 +130,17 @@ func TestAFilledFileIsWrittenAsItsOwnershipSays(t *testing.T) { } }) } + t.Run("a filled file that carried a secret is not world-readable", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "token.env") + d := filledDecl(t, path, `"content":"TOKEN=${secret:token}\n","secrets":{"token":"c2VhbGVk"},"owner-at-runtime":"owned"`, "") + if _, _, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("TOKEN=abc\n"), + noServices, nil, nil, nil); err != nil { + t.Fatal(err) + } + if m := modeAt(t, path); m != 0o600 { + t.Errorf("a filled file that carried a secret is mode %o", m) + } + }) } func contentOf(t *testing.T, fields string) string { @@ -136,17 +165,6 @@ func modeAt(t *testing.T, path string) os.FileMode { // A file that carried a secret keeps the mode a secret is written with: the client filled the value, and the file // is no more readable for it. -func TestAFilledFileThatCarriedASecretIsNotWorldReadable(t *testing.T) { - path := filepath.Join(t.TempDir(), "token.env") - d := filledDecl(t, path, `"content":"TOKEN=${secret:token}\n","secrets":{"token":"c2VhbGVk"},"owner-at-runtime":"owned"`, "") - if _, _, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("TOKEN=abc\n"), - noServices, nil, nil, nil); err != nil { - t.Fatal(err) - } - if m := modeAt(t, path); m != 0o600 { - t.Errorf("a filled file that carried a secret is mode %o", m) - } -} func TestAWriteUnderReloadOnReloadsItsProgramOnce(t *testing.T) { path := filepath.Join(t.TempDir(), "daemon.conf") @@ -251,21 +269,3 @@ func count(lines []string, prefix string) int { // Two resources declaring one filled path — two modules' blocks in one file — are refused: the request carries the // content only, so it cannot say whose block it is, and nothing is written. -func TestAFilledPathTwoResourcesDeclareIsRefused(t *testing.T) { - path := filepath.Join(t.TempDir(), "hosts") - if err := os.WriteFile(path, []byte(workstationHosts), 0o644); err != nil { - t.Fatal(err) - } - d := parseTrusted(t, fmt.Sprintf(`{"declaration":1,"filled_files":[%q],"resources":[ - {"id":"a.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.1 ace\n","owner-at-runtime":"managed"}, - {"id":"b.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.2 g14\n","owner-at-runtime":"managed"} - ]}`, path, path, path)) - _, state, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("6.6.6.6 evil\n"), - noServices, nil, nil, nil) - if err == nil || !strings.Contains(err.Error(), "2 resources declare") { - t.Fatalf("a path two resources declare was not refused: %v", err) - } - if got := readText(t, path); got != workstationHosts || len(state.Resources) != 0 { - t.Errorf("something was written or recorded: %q %+v", got, state.Resources) - } -} diff --git a/internal/link/have.go b/internal/link/have.go index d47fdb9..e7d0df2 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -47,6 +47,9 @@ type haveState struct { // said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted). written map[string]string removed map[string]string + // userList is what the bus node's engine last wrote of the bus's user list, by the module whose have says it (novox/hq + // ADR 0547 §8); kept apart from the module's fact, which every apply replaces. + userList map[string]delivery.UserListWritten // unjudged is each module whose composition moved (an apply began, or the fact names another build) and that no look // has judged since it stopped moving: its health is not yet judged, never the last look's of what ran before. unjudged map[string]bool @@ -308,6 +311,9 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { h.Phase = delivery.PhaseFailed } h.Health = s.healthOf(module, f) + if w, ok := s.userList[module]; ok { + h.UserList = &w + } key, err := delivery.HaveKey(node, module) if err != nil { q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err)) @@ -466,6 +472,24 @@ func (q *Session) NoteModule(ctx context.Context, f ModuleFact) { q.writeHaves(ctx, bus) } +// NoteUserList takes what the bus node's engine wrote of the bus's user list, and writes it in the bus module's have +// where it differs from the bucket's (novox/hq ADR 0547 §8). A module this engine has said no have of yet carries it +// from its first have on. +func (q *Session) NoteUserList(ctx context.Context, module string, w delivery.UserListWritten) { + q.init() + s := &q.haves + s.mu.Lock() + if s.userList == nil { + s.userList = map[string]delivery.UserListWritten{} + } + s.userList[module] = w + s.mu.Unlock() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + q.writeHaves(ctx, bus) +} + // NoteRemoved takes that a module's own resources were removed, and writes its have with the phase removed once (issue // 1369): the key is never deleted. func (q *Session) NoteRemoved(ctx context.Context, module, build string) { diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 3a87c64..d5fdb47 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -484,6 +484,18 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth( t.Fatalf("a failing check after the start window: %+v", got) } + // The bus's user list written alone rides on the module's have, and an apply after it keeps it (ADR 0547 §8). + w := delivery.UserListWritten{Source: "n1.web." + fp, Revision: 4, Wanted: dig, Digest: dig, Users: 3, + At: time.Date(2026, 10, 25, 21, 0, 0, 0, time.UTC)} + q.NoteUserList(context.Background(), "web", w) + if got := l.have(t, "n1.web"); got.UserList == nil || got.UserList.Revision != 4 || got.FP != fp { + t.Fatalf("the user list is not in the have: %+v", got) + } + q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig}) + if got := l.have(t, "n1.web"); got.UserList == nil || got.UserList.Users != 3 { + t.Fatalf("an apply lost the user list: %+v", got) + } + // A module held for a failed backup says so and what it keeps running; removed is written once, over its key. q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig, Phase: string(delivery.PhaseHeld), Detail: "the repository is locked"}) if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseHeld || got.Detail == "" { diff --git a/internal/userlist/userlist.go b/internal/userlist/userlist.go new file mode 100644 index 0000000..91c505a --- /dev/null +++ b/internal/userlist/userlist.go @@ -0,0 +1,270 @@ +// Package userlist is the bridge of novox/hq ADR 0547 §8: the node agent of the node holding mesh-broker applies the +// bus's user list from the newest composition of the bus module the controller writes for its node, alone and with no +// want, so a new account is in force on the bus without anybody's hand. +// +// **Why it exists.** The list reached the bus only through a want of the bus module, and ADR 0539 §15 bars the assigner +// from writing one: on 2026-10-25 it stood from 11:23 to 20:50 and every new account was refused until the operator put +// it back by hand (issue 1945). The bus module's entrypoint already reloads the server in place when the file changes; +// what was missing is a writer of the file that does not wait on a want. +// +// **It watches, it never polls.** It follows the node's composition bucket. Each composition is verified as the agent +// verifies it (signed by the controller's or the operator's key) before anything of it is read. Of those that carry the +// bus's user list (a file resource `bus-users`), the newest by revision is the source; its list is written alone +// (apply.WriteAlone) when it differs from the file on the machine, and what was written is reported in the bus +// module's have. It looks again after every apply of the bus module, since an apply of an older composition writes an +// older list, and when asked (Again). +// +// It reads no want and writes none, moves no fingerprint, and touches nothing of the bus module but its user list. +package userlist + +import ( + "context" + "crypto/ed25519" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "regexp" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/novox/mesh-host/internal/agent" +) + +// ResourceID is the id the composer gives the bus's user list (mesh-sdk catalogue.BusUsersID); in a composition it may +// carry its module before it, `.bus-users`. +const ResourceID = "bus-users" + +// Written is what the follower last did: the source it took the list from, the list's digest there, the file's digest +// on the machine after the write, how many users the list holds, when, and why a write failed. +type Written = delivery.UserListWritten + +// Env is everything the follower needs of the node. +type Env struct { + Node string + Composed agent.Watcher + // Trusted is the keys a composition may be signed by, as the agent trusts them. + Trusted func() []ed25519.PublicKey + // Write writes the user list's file resource of a module's composition alone, when it differs from the file on + // the machine, and answers the digest of the file there after it (`sha256:`) and whether it changed. + Write func(ctx context.Context, module string, resource json.RawMessage) (digest string, changed bool, err error) + // Report says what was written, in the module's have. + Report func(ctx context.Context, module string, w Written) + Now func() time.Time + Say func(string) +} + +type source struct { + key string + module string + rev uint64 + resource json.RawMessage + wanted string + users int +} + +// Follower follows the node's compositions for the bus's user list. +type Follower struct { + env Env + kick chan struct{} + mu sync.Mutex + srcs map[string]source // by composition key + // cur is the source last taken, and written whether its write went through. + cur source + written bool + state Written + mod string + said map[string]bool +} + +// New is a follower for a node. +func New(env Env) *Follower { + if env.Now == nil { + env.Now = time.Now + } + if env.Say == nil { + env.Say = func(string) {} + } + return &Follower{env: env, kick: make(chan struct{}, 1), srcs: map[string]source{}, said: map[string]bool{}} +} + +// Again asks the follower to look at the file again, though no composition changed: after an apply of the bus module, +// and when a person asks. It never blocks. +func (f *Follower) Again() { + select { + case f.kick <- struct{}{}: + default: + } +} + +// State is the module whose list the follower writes ("" when no composition of this node carries one) and what it +// last did. +func (f *Follower) State() (string, Written) { + f.mu.Lock() + defer f.mu.Unlock() + return f.mod, f.state +} + +// Run follows the compositions until ctx ends. +func (f *Follower) Run(ctx context.Context) error { + updates, err := f.env.Composed.Watch(ctx, f.env.Node+".>") + if err != nil { + return fmt.Errorf("the user list cannot be followed: %w", err) + } + caught := false + for { + again := false + select { + case <-ctx.Done(): + return nil + case <-f.kick: + again = true + case u, ok := <-updates: + if !ok { + return fmt.Errorf("the watch of the compositions ended") + } + if u.Caught { + caught = true + } else { + f.take(u) + } + } + if caught { + f.settle(ctx, again) + } + } +} + +// take keeps a composition key's change when it carries the bus's user list, and forgets a key deleted. +func (f *Follower) take(u agent.Update) { + node, module, _, isNode, err := delivery.ParseComposedKey(u.Key) + if err != nil || node != f.env.Node || isNode { + return + } + f.mu.Lock() + defer f.mu.Unlock() + if u.Deleted { + delete(f.srcs, u.Key) + return + } + c, err := delivery.DecodeComposed(u.Value) + if err != nil { + delete(f.srcs, u.Key) + return + } + var trusted []ed25519.PublicKey + if f.env.Trusted != nil { + trusted = f.env.Trusted() + } + if _, err := delivery.VerifyComposed(c, trusted); err != nil { + // The agent says it is refused; a refused composition is never a source of the list. + delete(f.srcs, u.Key) + return + } + raw, content, digest, ok := userList(c.Resources) + if !ok { + delete(f.srcs, u.Key) + return + } + wanted := digest + if wanted == "" { + wanted = Digest([]byte(content)) + } + f.srcs[u.Key] = source{key: u.Key, module: module, rev: u.Revision, resource: raw, wanted: wanted, + users: Users([]byte(content))} +} + +// settle writes the newest source's list when it is newer than the one written, or the one written again when asked. +// A source never goes back: a key deleted, or a newer one refused, leaves the list written as it is, since an older +// list would take away accounts a newer one gave. +func (f *Follower) settle(ctx context.Context, again bool) { + f.mu.Lock() + newest := f.cur + for _, s := range f.srcs { + if s.rev > newest.rev { + newest = s + } + } + if newest.key == "" || (newest.rev == f.cur.rev && f.written && !again) { + f.mu.Unlock() + return + } + f.cur = newest + at := fmt.Sprintf("%s@%d", newest.key, newest.rev) + f.mu.Unlock() + + digest, changed, err := f.env.Write(ctx, newest.module, newest.resource) + w := Written{Source: newest.key, Revision: newest.rev, Wanted: newest.wanted, Digest: digest, Users: newest.users, + At: f.env.Now().UTC()} + if err != nil { + w.Error = err.Error() + } + f.mu.Lock() + prev := f.state + f.mod, f.state = newest.module, w + f.written = err == nil + f.mu.Unlock() + if err != nil { + f.sayOnce(fmt.Sprintf("the bus's user list of %s could not be written: %v", at, err)) + } else if changed { + f.env.Say(fmt.Sprintf("the bus's user list from %s is written alone (%d users, %s); the bus reloads it itself", + at, w.Users, short(digest))) + } + // Reported when what it says changed: a look again that found the file as it was says nothing new. + if f.env.Report != nil && (changed || prev.Source != w.Source || prev.Revision != w.Revision || + prev.Digest != w.Digest || prev.Error != w.Error) { + f.env.Report(ctx, newest.module, w) + } +} + +func (f *Follower) sayOnce(line string) { + f.mu.Lock() + said := f.said[line] + f.said[line] = true + f.mu.Unlock() + if !said { + f.env.Say(line) + } +} + +// userList is the bus's user list among a composition's resources: the resource, its content, and its digest when it +// is given by digest rather than content. +func userList(resources []json.RawMessage) (json.RawMessage, string, string, bool) { + for _, raw := range resources { + var head struct { + ID string `json:"id"` + Type string `json:"type"` + Content string `json:"content"` + Digest string `json:"digest"` + } + if json.Unmarshal(raw, &head) != nil || head.Type != "file" { + continue + } + if head.ID == ResourceID || strings.HasSuffix(head.ID, "."+ResourceID) { + return raw, head.Content, head.Digest, true + } + } + return nil, "", "", false +} + +// Digest is content's digest as the have says it, `sha256:`. +func Digest(content []byte) string { + sum := sha256.Sum256(content) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// userEntry is one user of the list as the composer writes it (and the bus module's own tools read it). +var userEntry = regexp.MustCompile(`\{\s*user:\s*"([^"]+)"`) + +// Users is how many users a list holds. +func Users(content []byte) int { return len(userEntry.FindAllIndex(content, -1)) } + +func short(d string) string { + d = strings.TrimPrefix(d, "sha256:") + if len(d) > 12 { + d = d[:12] + } + return d +} diff --git a/internal/userlist/userlist_test.go b/internal/userlist/userlist_test.go new file mode 100644 index 0000000..ad2c155 --- /dev/null +++ b/internal/userlist/userlist_test.go @@ -0,0 +1,222 @@ +package userlist + +import ( + "context" + "crypto/ed25519" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/novox/mesh-host/internal/agent" +) + +const node = "novox" + +func key() (ed25519.PublicKey, ed25519.PrivateKey) { + pub, k, err := ed25519.GenerateKey(nil) + if err != nil { + panic(err) + } + return pub, k +} + +var controllerPub, controllerKey = key() +var _, strangerKey = key() + +func fpOf(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) } + +// feed is a watch the test writes to: the keys there before the watch, Caught, then each put. +type feed struct { + ch chan agent.Update + rev uint64 + first []agent.Update +} + +func (f *feed) Watch(context.Context, string) (<-chan agent.Update, error) { + for _, u := range f.first { + f.ch <- u + } + f.ch <- agent.Update{Caught: true} + return f.ch, nil +} + +func list(users ...string) string { + var b strings.Builder + b.WriteString("authorization {\n users = [\n") + for _, u := range users { + b.WriteString(` {user: "` + u + `", password: "$2a$11$x"}` + "\n") + } + b.WriteString(" ]\n}\n") + return b.String() +} + +// comp is a composition of module at fp holding the user list content (none when empty), signed by signer. +func comp(t *testing.T, module, fp, content string, signer ed25519.PrivateKey) (string, []byte) { + t.Helper() + server, _ := json.Marshal(map[string]any{"id": module + ".server-conf", "type": "file", "module": module, + "path": "/var/lib/nats-module/conf/nats.conf", "content": "include accounts.conf\n"}) + rs := []json.RawMessage{server} + if content != "" { + users, _ := json.Marshal(map[string]any{"id": module + ".bus-users", "type": "file", "module": module, + "path": "/var/lib/nats-module/conf/accounts.conf", "content": content, "mode": "0600"}) + rs = append(rs, users) + } + c := delivery.Composed{Node: node, Module: module, FP: fp, Resources: rs} + k, err := delivery.ComposedKey(node, module, fp) + if err != nil { + t.Fatal(err) + } + signed, err := delivery.SignComposed(c, signer) + if err != nil { + t.Fatal(err) + } + raw, err := delivery.EncodeComposed(signed) + if err != nil { + t.Fatal(err) + } + return k, raw +} + +type put struct { + module, fp, content string + stranger, deleted bool +} + +// Defends novox/hq ADR 0547 §8: the bus node writes the newest signed list alone, only when it changed, and reports it. +func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) { + one, two, three := list("a"), list("a", "b"), list("a", "b", "c") + for _, c := range []struct { + name string + before, after []put + again bool + failWrite bool + wantWrites []string // contents written, in order + wantUsers int + wantErr string + }{ + {name: "the list there is written once", before: []put{{"nats", fpOf("a"), two, false, false}}, + wantWrites: []string{two}, wantUsers: 2}, + {name: "a newer revision is written, the older fingerprint is not", + before: []put{{"nats", fpOf("a"), one, false, false}, {"nats", fpOf("b"), two, false, false}}, + after: []put{{"nats", fpOf("a"), three, false, false}}, + wantWrites: []string{two, three}, wantUsers: 3}, + {name: "a composition signed by a stranger is no source", + before: []put{{"nats", fpOf("a"), one, false, false}}, + after: []put{{"nats", fpOf("b"), three, true, false}}, + wantWrites: []string{one}, wantUsers: 1}, + {name: "a composition without the list is ignored", before: []put{{"web", fpOf("a"), "", false, false}}}, + {name: "a deleted key is no source", + before: []put{{"nats", fpOf("a"), one, false, false}, {"nats", fpOf("b"), two, false, false}}, + after: []put{{"nats", fpOf("b"), "", false, true}, {"nats", fpOf("a"), three, false, false}}, + wantWrites: []string{two, three}, wantUsers: 3}, + {name: "asked again, the same source is written again", before: []put{{"nats", fpOf("a"), two, false, false}}, + again: true, wantWrites: []string{two, two}, wantUsers: 2}, + {name: "a failed write is said and kept", before: []put{{"nats", fpOf("a"), two, false, false}}, failWrite: true, + wantWrites: []string{two}, wantUsers: 2, wantErr: "disk full"}, + } { + t.Run(c.name, func(t *testing.T) { + f := &feed{ch: make(chan agent.Update, 64)} + var mu sync.Mutex + var writes []string + var reports []Written + reported := make(chan struct{}, 64) + fl := New(Env{Node: node, Composed: f, + Trusted: func() []ed25519.PublicKey { return []ed25519.PublicKey{controllerPub} }, + Write: func(_ context.Context, module string, r json.RawMessage) (string, bool, error) { + var res struct{ Content string } + _ = json.Unmarshal(r, &res) + mu.Lock() + writes = append(writes, res.Content) + mu.Unlock() + if c.failWrite { + return "", false, errors.New("disk full") + } + return Digest([]byte(res.Content)), true, nil + }, + Report: func(_ context.Context, module string, w Written) { + mu.Lock() + reports = append(reports, w) + mu.Unlock() + reported <- struct{}{} + }, + Now: func() time.Time { return time.Date(2026, 10, 25, 21, 50, 0, 0, time.UTC) }, + }) + update := func(p put) agent.Update { + f.rev++ + signer := controllerKey + if p.stranger { + signer = strangerKey + } + k, raw := comp(t, p.module, p.fp, p.content, signer) + if p.deleted { + return agent.Update{Key: k, Revision: f.rev, Deleted: true} + } + return agent.Update{Key: k, Value: raw, Revision: f.rev} + } + for _, p := range c.before { + f.first = append(f.first, update(p)) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- fl.Run(ctx) }() + for _, p := range c.after { + f.ch <- update(p) + } + if c.again { + // After the first write is reported, so the look again is not folded into it. + select { + case <-reported: + case <-time.After(5 * time.Second): + } + fl.Again() + } + deadline := time.After(5 * time.Second) + for { + mu.Lock() + n := len(writes) + mu.Unlock() + if n >= len(c.wantWrites) { + break + } + select { + case <-deadline: + t.Fatalf("writes %d, want %d", n, len(c.wantWrites)) + case <-time.After(5 * time.Millisecond): + } + } + time.Sleep(50 * time.Millisecond) // nothing more is written + cancel() + <-done + mu.Lock() + defer mu.Unlock() + if strings.Join(writes, "|") != strings.Join(c.wantWrites, "|") { + t.Fatalf("written %q, want %q", writes, c.wantWrites) + } + module, st := fl.State() + if len(c.wantWrites) == 0 { + if module != "" || len(reports) != 0 { + t.Fatalf("a node with no list wrote or said something: %q %+v", module, reports) + } + return + } + if module != "nats" || st.Users != c.wantUsers || st.Wanted != Digest([]byte(c.wantWrites[len(c.wantWrites)-1])) { + t.Errorf("state %q %+v", module, st) + } + if c.wantErr != "" { + if !strings.Contains(st.Error, c.wantErr) || st.Digest != "" { + t.Errorf("a failed write is not kept: %+v", st) + } + } else if st.Digest != st.Wanted || st.Error != "" { + t.Errorf("written %+v", st) + } + if len(reports) == 0 || reports[len(reports)-1] != st { + t.Errorf("reports %+v, last state %+v", reports, st) + } + }) + } +} From 5e71ab6ae6127f385aaf0903723ac80fdf0f2edf Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 22:00:05 +0100 Subject: [PATCH 2/2] =?UTF-8?q?Count=20the=20bus's=20users=20from=20the=20?= =?UTF-8?q?list=20a=20digest=20names,=20take=20only=20a=20composition=20of?= =?UTF-8?q?=20this=20node=20and=20module,=20and=20refuse=20a=20list=20that?= =?UTF-8?q?=20would=20lock=20the=20mesh=20out=20(review=20of=20the=20bridg?= =?UTF-8?q?e,=20ADR=200547=20=C2=A78)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B1: the controller sends the user list by digest (over 64 KiB), so its users are counted from the list as resolved, not from the composition's inline content, which is empty then. R1: a composition signed for another node or module than its key names is no source. R2: a list without the controller's own user, or one that drops more than half the users of the file on the machine, is not written; the have says why and the file on the machine is kept. --- cmd/mesh-host/userlist.go | 62 ++++++++++++--- internal/userlist/userlist.go | 87 +++++++++++++++++--- internal/userlist/userlist_test.go | 123 ++++++++++++++++++++++------- 3 files changed, 223 insertions(+), 49 deletions(-) diff --git a/cmd/mesh-host/userlist.go b/cmd/mesh-host/userlist.go index 370d1ea..605e1a2 100644 --- a/cmd/mesh-host/userlist.go +++ b/cmd/mesh-host/userlist.go @@ -28,8 +28,12 @@ func userListFollower(opts options, session *link.Session, composedBucket string Node: session.Membership.Node, Composed: bucketWatch{session, composedBucket}, Trusted: trusted, - Write: func(ctx context.Context, module string, raw json.RawMessage) (string, bool, error) { - return writeUserListAlone(session.WithLink(ctx), opts, raw, say) + Resolve: func(ctx context.Context, raw json.RawMessage) ([]byte, error) { + return resolveUserList(session.WithLink(ctx), raw) + }, + OnDisk: userListOnDisk, + Write: func(ctx context.Context, module string, raw json.RawMessage, body []byte) (string, bool, error) { + return writeUserListAlone(session.WithLink(ctx), opts, raw, body, say) }, Report: func(ctx context.Context, module string, w userlist.Written) { session.NoteUserList(ctx, module, w) @@ -38,21 +42,55 @@ func userListFollower(opts options, session *link.Session, composedBucket string }) } -// writeUserListAlone writes one user-list resource alone (apply.WriteAlone): its content had by digest when it is -// given so, the node's state read and saved under the apply's locks, and the digest of the file on the machine after. -func writeUserListAlone(ctx context.Context, opts options, raw []byte, say link.Announce) (string, bool, error) { +// userListFile is a user-list resource as the file it declares. +func userListFile(raw []byte) (*declaration.File, error) { r, err := declaration.ParseResource(raw) if err != nil { - return "", false, fmt.Errorf("the user list is not a resource: %w", err) + return nil, fmt.Errorf("the user list is not a resource: %w", err) } - if _, ok := r.(*declaration.File); !ok { - return "", false, fmt.Errorf("the user list is a %s, not a file", r.Kind()) + f, ok := r.(*declaration.File) + if !ok { + return nil, fmt.Errorf("the user list is a %s, not a file", r.Kind()) } - d := &declaration.Declaration{Version: declaration.Version, Resources: []declaration.Resource{r}} + return f, nil +} + +// resolveUserList is the list a user-list resource holds: its content, or the object its digest names in this node's +// file bucket, fetched and verified (apply.ResolveFileReferences, ADR 0486 §1). +func resolveUserList(ctx context.Context, raw []byte) ([]byte, error) { + f, err := userListFile(raw) + if err != nil { + return nil, err + } + d := &declaration.Declaration{Version: declaration.Version, Resources: []declaration.Resource{f}} if err := apply.ResolveFileReferences(ctx, d, link.FilesFrom(ctx)); err != nil { - return "", false, fmt.Errorf("the user list could not be had by its digest: %w", err) + return nil, fmt.Errorf("the user list could not be had by its digest: %w", err) } - file := d.Resources[0].(*declaration.File) + return []byte(d.Resources[0].(*declaration.File).Content), nil +} + +// userListOnDisk is what the user list's file holds on this machine now; nil when there is none. +func userListOnDisk(raw json.RawMessage) ([]byte, error) { + f, err := userListFile(raw) + if err != nil { + return nil, err + } + body, err := os.ReadFile(f.Path) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + return body, err +} + +// writeUserListAlone writes one user-list resource alone (apply.WriteAlone) with body as its content, the node's state +// read and saved under the apply's locks, and answers the digest of the file on the machine after. +func writeUserListAlone(ctx context.Context, opts options, raw []byte, body []byte, say link.Announce) (string, bool, error) { + declared, err := userListFile(raw) + if err != nil { + return "", false, err + } + file := *declared + file.Content, file.Digest, file.Size = string(body), "", 0 applying.Lock() defer applying.Unlock() @@ -69,7 +107,7 @@ func writeUserListAlone(ctx context.Context, opts options, raw []byte, say link. if err != nil { return "", false, err } - out, updated, writeErr := apply.WriteAlone(ctx, built, file, known, apply.ExecRunner, announceOr(say)) + out, updated, writeErr := apply.WriteAlone(ctx, built, &file, known, apply.ExecRunner, announceOr(say)) // Saved whichever way it went, as an apply's: what was written is on the machine either way. if err := store.Save(opts.state, updated); err != nil { writeErr = errors.Join(writeErr, errors.New("written, and the node's state could not be saved: "+err.Error())) diff --git a/internal/userlist/userlist.go b/internal/userlist/userlist.go index 91c505a..7992a6e 100644 --- a/internal/userlist/userlist.go +++ b/internal/userlist/userlist.go @@ -25,6 +25,7 @@ import ( "encoding/json" "fmt" "regexp" + "slices" "strings" "sync" "time" @@ -47,9 +48,15 @@ type Env struct { Composed agent.Watcher // Trusted is the keys a composition may be signed by, as the agent trusts them. Trusted func() []ed25519.PublicKey - // Write writes the user list's file resource of a module's composition alone, when it differs from the file on - // the machine, and answers the digest of the file there after it (`sha256:`) and whether it changed. - Write func(ctx context.Context, module string, resource json.RawMessage) (digest string, changed bool, err error) + // Resolve is the list a user-list resource holds: its content, or the object its digest names, fetched and + // verified (the controller sends a file over 64 KiB by digest, ADR 0486 §1, and the live list is over it). + Resolve func(ctx context.Context, resource json.RawMessage) ([]byte, error) + // OnDisk is what the resource's file holds on the machine now; nil when there is none. + OnDisk func(resource json.RawMessage) ([]byte, error) + // Write writes the user list's file resource of a module's composition alone, with body as its content, when it + // differs from the file on the machine, and answers the digest of the file there after it (`sha256:`) and + // whether it changed. + Write func(ctx context.Context, module string, resource json.RawMessage, body []byte) (digest string, changed bool, err error) // Report says what was written, in the module's have. Report func(ctx context.Context, module string, w Written) Now func() time.Time @@ -62,7 +69,6 @@ type source struct { rev uint64 resource json.RawMessage wanted string - users int } // Follower follows the node's compositions for the bus's user list. @@ -163,6 +169,13 @@ func (f *Follower) take(u agent.Update) { delete(f.srcs, u.Key) return } + if c.Node != f.env.Node || c.Module != module { + // Signed, and for another node or module than its key says: a re-put of another composition is no source. + delete(f.srcs, u.Key) + f.sayOnceLocked(fmt.Sprintf("the composition under %s is %s's of %s, and is no source of the bus's user list", + u.Key, c.Node, c.Module)) + return + } raw, content, digest, ok := userList(c.Resources) if !ok { delete(f.srcs, u.Key) @@ -172,8 +185,7 @@ func (f *Follower) take(u agent.Update) { if wanted == "" { wanted = Digest([]byte(content)) } - f.srcs[u.Key] = source{key: u.Key, module: module, rev: u.Revision, resource: raw, wanted: wanted, - users: Users([]byte(content))} + f.srcs[u.Key] = source{key: u.Key, module: module, rev: u.Revision, resource: raw, wanted: wanted} } // settle writes the newest source's list when it is newer than the one written, or the one written again when asked. @@ -195,9 +207,22 @@ func (f *Follower) settle(ctx context.Context, again bool) { at := fmt.Sprintf("%s@%d", newest.key, newest.rev) f.mu.Unlock() - digest, changed, err := f.env.Write(ctx, newest.module, newest.resource) - w := Written{Source: newest.key, Revision: newest.rev, Wanted: newest.wanted, Digest: digest, Users: newest.users, - At: f.env.Now().UTC()} + w := Written{Source: newest.key, Revision: newest.rev, Wanted: newest.wanted, At: f.env.Now().UTC()} + var changed bool + body, err := f.env.Resolve(ctx, newest.resource) + if err == nil { + w.Users = Users(body) + err = f.guard(newest.resource, body) + if err != nil { + // Not written: what is on the machine stays, and the have says why, loudly. + if disk, derr := f.env.OnDisk(newest.resource); derr == nil && disk != nil { + w.Digest = Digest(disk) + } + } + } + if err == nil { + w.Digest, changed, err = f.env.Write(ctx, newest.module, newest.resource, body) + } if err != nil { w.Error = err.Error() } @@ -210,7 +235,7 @@ func (f *Follower) settle(ctx context.Context, again bool) { f.sayOnce(fmt.Sprintf("the bus's user list of %s could not be written: %v", at, err)) } else if changed { f.env.Say(fmt.Sprintf("the bus's user list from %s is written alone (%d users, %s); the bus reloads it itself", - at, w.Users, short(digest))) + at, w.Users, short(w.Digest))) } // Reported when what it says changed: a look again that found the file as it was says nothing new. if f.env.Report != nil && (changed || prev.Source != w.Source || prev.Revision != w.Revision || @@ -219,6 +244,39 @@ func (f *Follower) settle(ctx context.Context, again bool) { } } +// ControllerUser is the controller's own user in the list: a list without it would lock the controller out of the bus. +const ControllerUser = "controller" + +// guard refuses a list that would lock the mesh out of its own bus: one without the controller's own user, or one that +// drops more than half the users the file on the machine holds. A list that shrinks that far is a composer that read +// half the mesh, not half the mesh gone; it waits for a person, and the bus keeps the list it has. +func (f *Follower) guard(resource json.RawMessage, body []byte) error { + names := Names(body) + if !slices.Contains(names, ControllerUser) { + return fmt.Errorf("refused: the list holds no user %q, and written it would lock the controller out of the "+ + "bus; the list on the machine is kept", ControllerUser) + } + if f.env.OnDisk == nil { + return nil + } + disk, err := f.env.OnDisk(resource) + if err != nil || disk == nil { + return nil + } + if had := Users(disk); had > 0 && len(names)*2 < had { + return fmt.Errorf("refused: the list holds %d users and the one on the machine %d; a list that drops more "+ + "than half is not written, and the list on the machine is kept", len(names), had) + } + return nil +} + +func (f *Follower) sayOnceLocked(line string) { + if !f.said[line] { + f.said[line] = true + f.env.Say(line) + } +} + func (f *Follower) sayOnce(line string) { f.mu.Lock() said := f.said[line] @@ -261,6 +319,15 @@ var userEntry = regexp.MustCompile(`\{\s*user:\s*"([^"]+)"`) // Users is how many users a list holds. func Users(content []byte) int { return len(userEntry.FindAllIndex(content, -1)) } +// Names is the users a list holds, as it names them. +func Names(content []byte) []string { + var out []string + for _, m := range userEntry.FindAllSubmatch(content, -1) { + out = append(out, string(m[1])) + } + return out +} + func short(d string) string { d = strings.TrimPrefix(d, "sha256:") if len(d) > 12 { diff --git a/internal/userlist/userlist_test.go b/internal/userlist/userlist_test.go index ad2c155..d4d52c1 100644 --- a/internal/userlist/userlist_test.go +++ b/internal/userlist/userlist_test.go @@ -54,18 +54,34 @@ func list(users ...string) string { return b.String() } -// comp is a composition of module at fp holding the user list content (none when empty), signed by signer. -func comp(t *testing.T, module, fp, content string, signer ed25519.PrivateKey) (string, []byte) { +// comp is a composition of module at fp holding the user list content (none when empty), signed by signer; by digest +// when byDigest, as the controller sends a file over 64 KiB; for another node or module than its key when asked. +func comp(t *testing.T, module, fp, content string, signer ed25519.PrivateKey, p put) (string, []byte) { t.Helper() - server, _ := json.Marshal(map[string]any{"id": module + ".server-conf", "type": "file", "module": module, + owner := module + if p.asModule != "" { + owner = p.asModule + } + server, _ := json.Marshal(map[string]any{"id": owner + ".server-conf", "type": "file", "module": owner, "path": "/var/lib/nats-module/conf/nats.conf", "content": "include accounts.conf\n"}) rs := []json.RawMessage{server} if content != "" { - users, _ := json.Marshal(map[string]any{"id": module + ".bus-users", "type": "file", "module": module, - "path": "/var/lib/nats-module/conf/accounts.conf", "content": content, "mode": "0600"}) + r := map[string]any{"id": owner + ".bus-users", "type": "file", "module": owner, + "path": "/var/lib/nats-module/conf/accounts.conf", "content": content, "mode": "0600"} + if p.byDigest { + delete(r, "content") + r["digest"], r["size"] = Digest([]byte(content)), len(content) + } + users, _ := json.Marshal(r) rs = append(rs, users) } c := delivery.Composed{Node: node, Module: module, FP: fp, Resources: rs} + if p.asNode != "" { + c.Node = p.asNode + } + if p.asModule != "" { + c.Module = p.asModule + } k, err := delivery.ComposedKey(node, module, fp) if err != nil { t.Fatal(err) @@ -84,58 +100,95 @@ func comp(t *testing.T, module, fp, content string, signer ed25519.PrivateKey) ( type put struct { module, fp, content string stranger, deleted bool + byDigest bool + asNode, asModule string } // Defends novox/hq ADR 0547 §8: the bus node writes the newest signed list alone, only when it changed, and reports it. func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) { - one, two, three := list("a"), list("a", "b"), list("a", "b", "c") + one, two, three := list("controller", "a"), list("controller", "a", "b"), list("controller", "a", "b", "c") + five := list("controller", "a", "b", "c", "d") for _, c := range []struct { name string before, after []put again bool failWrite bool + onDisk string wantWrites []string // contents written, in order wantUsers int wantErr string + wantRefused bool }{ - {name: "the list there is written once", before: []put{{"nats", fpOf("a"), two, false, false}}, - wantWrites: []string{two}, wantUsers: 2}, + {name: "the list there is written once", before: []put{{module: "nats", fp: fpOf("a"), content: two}}, + wantWrites: []string{two}, wantUsers: 3}, {name: "a newer revision is written, the older fingerprint is not", - before: []put{{"nats", fpOf("a"), one, false, false}, {"nats", fpOf("b"), two, false, false}}, - after: []put{{"nats", fpOf("a"), three, false, false}}, - wantWrites: []string{two, three}, wantUsers: 3}, + before: []put{{module: "nats", fp: fpOf("a"), content: one}, {module: "nats", fp: fpOf("b"), content: two}}, + after: []put{{module: "nats", fp: fpOf("a"), content: three}}, + wantWrites: []string{two, three}, wantUsers: 4}, {name: "a composition signed by a stranger is no source", - before: []put{{"nats", fpOf("a"), one, false, false}}, - after: []put{{"nats", fpOf("b"), three, true, false}}, - wantWrites: []string{one}, wantUsers: 1}, - {name: "a composition without the list is ignored", before: []put{{"web", fpOf("a"), "", false, false}}}, + before: []put{{module: "nats", fp: fpOf("a"), content: one}}, + after: []put{{module: "nats", fp: fpOf("b"), content: three, stranger: true}}, + wantWrites: []string{one}, wantUsers: 2}, + {name: "a composition without the list is ignored", before: []put{{module: "web", fp: fpOf("a"), content: ""}}}, {name: "a deleted key is no source", - before: []put{{"nats", fpOf("a"), one, false, false}, {"nats", fpOf("b"), two, false, false}}, - after: []put{{"nats", fpOf("b"), "", false, true}, {"nats", fpOf("a"), three, false, false}}, - wantWrites: []string{two, three}, wantUsers: 3}, - {name: "asked again, the same source is written again", before: []put{{"nats", fpOf("a"), two, false, false}}, - again: true, wantWrites: []string{two, two}, wantUsers: 2}, - {name: "a failed write is said and kept", before: []put{{"nats", fpOf("a"), two, false, false}}, failWrite: true, - wantWrites: []string{two}, wantUsers: 2, wantErr: "disk full"}, + before: []put{{module: "nats", fp: fpOf("a"), content: one}, {module: "nats", fp: fpOf("b"), content: two}}, + after: []put{{module: "nats", fp: fpOf("b"), content: "", deleted: true}, {module: "nats", fp: fpOf("a"), content: three}}, + wantWrites: []string{two, three}, wantUsers: 4}, + {name: "asked again, the same source is written again", before: []put{{module: "nats", fp: fpOf("a"), content: two}}, + again: true, wantWrites: []string{two, two}, wantUsers: 3}, + {name: "a list sent by digest counts the users of what it names", + before: []put{{module: "nats", fp: fpOf("a"), content: three, byDigest: true}}, + wantWrites: []string{three}, wantUsers: 4}, + {name: "another node's composition under this node's key is no source", + before: []put{{module: "nats", fp: fpOf("a"), content: two, asNode: "ace"}}}, + {name: "another module's composition under the bus module's key is no source", + before: []put{{module: "nats", fp: fpOf("a"), content: two, asModule: "web"}}}, + {name: "a list without the controller is refused and the file kept", + before: []put{{module: "nats", fp: fpOf("a"), content: list("a", "b")}}, + onDisk: two, wantUsers: 2, wantErr: "holds no user \"controller\"", wantRefused: true}, + {name: "a list that drops more than half is refused and the file kept", + before: []put{{module: "nats", fp: fpOf("a"), content: one}}, + onDisk: five, wantUsers: 2, wantErr: "drops more than half", wantRefused: true}, + {name: "a list that drops half is written", before: []put{{module: "nats", fp: fpOf("a"), content: two}}, + onDisk: list("controller", "a", "b", "c", "d", "e"), wantWrites: []string{two}, wantUsers: 3}, + {name: "a failed write is said and kept", before: []put{{module: "nats", fp: fpOf("a"), content: two}}, failWrite: true, + wantWrites: []string{two}, wantUsers: 3, wantErr: "disk full"}, } { t.Run(c.name, func(t *testing.T) { f := &feed{ch: make(chan agent.Update, 64)} + objects := map[string]string{} var mu sync.Mutex var writes []string var reports []Written reported := make(chan struct{}, 64) fl := New(Env{Node: node, Composed: f, Trusted: func() []ed25519.PublicKey { return []ed25519.PublicKey{controllerPub} }, - Write: func(_ context.Context, module string, r json.RawMessage) (string, bool, error) { - var res struct{ Content string } + Resolve: func(_ context.Context, r json.RawMessage) ([]byte, error) { + var res struct{ Content, Digest string } _ = json.Unmarshal(r, &res) + if res.Digest != "" { + body, ok := objects[res.Digest] + if !ok { + return nil, errors.New("no object " + res.Digest) + } + return []byte(body), nil + } + return []byte(res.Content), nil + }, + OnDisk: func(json.RawMessage) ([]byte, error) { + if c.onDisk == "" { + return nil, nil + } + return []byte(c.onDisk), nil + }, + Write: func(_ context.Context, module string, r json.RawMessage, body []byte) (string, bool, error) { mu.Lock() - writes = append(writes, res.Content) + writes = append(writes, string(body)) mu.Unlock() if c.failWrite { return "", false, errors.New("disk full") } - return Digest([]byte(res.Content)), true, nil + return Digest(body), true, nil }, Report: func(_ context.Context, module string, w Written) { mu.Lock() @@ -151,7 +204,10 @@ func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) { if p.stranger { signer = strangerKey } - k, raw := comp(t, p.module, p.fp, p.content, signer) + k, raw := comp(t, p.module, p.fp, p.content, signer, p) + if p.byDigest { + objects[Digest([]byte(p.content))] = p.content + } if p.deleted { return agent.Update{Key: k, Revision: f.rev, Deleted: true} } @@ -189,6 +245,13 @@ func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) { case <-time.After(5 * time.Millisecond): } } + if c.wantRefused { + select { + case <-reported: + case <-time.After(5 * time.Second): + t.Fatal("a refused list was not reported") + } + } time.Sleep(50 * time.Millisecond) // nothing more is written cancel() <-done @@ -198,6 +261,12 @@ func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) { t.Fatalf("written %q, want %q", writes, c.wantWrites) } module, st := fl.State() + if c.wantRefused { + if !strings.Contains(st.Error, c.wantErr) || st.Digest != Digest([]byte(c.onDisk)) || st.Users != c.wantUsers { + t.Fatalf("refused list's state %+v", st) + } + return + } if len(c.wantWrites) == 0 { if module != "" || len(reports) != 0 { t.Fatalf("a node with no list wrote or said something: %q %+v", module, reports)