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)