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) + } + }) + } +}