The bus node writes the bus's user list alone from the newest composition of the bus module, with no want, and says it in the bus module's have (ADR 0547 §8, issue 1945)
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.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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 == "" {
|
||||
|
||||
@@ -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, `<module>.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:<hex>`) 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:<hex>`.
|
||||
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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user