jschoubben
a092ee9ff9
The bus node can only take over a list that exists: each pass composes the accounts block with the vault's hashes, seals it to the key the bus node says in its have, puts it in the part's own object bucket by digest and writes the key with digest, revision, user count, users left out and which list to follow. take-over is refused, naming why, while no list is sealed to the bus node's current key or the loaded list does not grant the bus node its read. Review of fix/1945-takeover-blockers: no responders on the router's record is an error, not 'no record'; an answer whose record cannot be reached is not closed; an ask that failed to publish is taken back at the router; an error acting on an answer is logged. Pins mesh-sdk 175a5567.
1459 lines
60 KiB
Go
1459 lines
60 KiB
Go
package manifest
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io/fs"
|
|
"os"
|
|
"path/filepath"
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/asker"
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/seal"
|
|
|
|
"mesh-grants/internal/grants"
|
|
)
|
|
|
|
// fakeBus is the bus as the part sees it: other parts' states to read (refusing those named), the bus module's read
|
|
// tools, the part's own states and its events. It records every write, call and event, so a row can prove nothing was
|
|
// published or asserted.
|
|
type fakeBus struct {
|
|
mu sync.Mutex
|
|
states map[string]map[string][]byte
|
|
refused map[string]bool
|
|
loaded []grants.User
|
|
streams []map[string]any
|
|
own map[string]map[string][]byte
|
|
revs map[string]uint64
|
|
refuseTo map[string]bool
|
|
calls []string
|
|
events []string
|
|
writes []string
|
|
// The assignments stream: each subject's last membership and sequence, every publish, and whether it is refused.
|
|
stream map[string][]byte
|
|
seqs map[string]uint64
|
|
seq uint64
|
|
publishes []string
|
|
refusePub bool
|
|
// hold, when set, keeps every nats_users call waiting until it is closed: a pass waiting on the bus.
|
|
hold chan struct{}
|
|
// creates are the streams the part asked to create, refused when refuseCreate.
|
|
creates []string
|
|
refuseCreate bool
|
|
// rootFree answers the controller's root-free verb: nil is no answerer at all (no responders).
|
|
rootFree func(machines []string) (any, error)
|
|
}
|
|
|
|
func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
subject := "mesh.assignment." + node + "." + module
|
|
v, ok := b.stream[subject]
|
|
return v, b.seqs[subject], ok, nil
|
|
}
|
|
|
|
func (b *fakeBus) PublishMembership(node, module string, body []byte, after uint64) error {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
subject := "mesh.assignment." + node + "." + module
|
|
if b.refusePub {
|
|
return errors.New("Permissions Violation for Publish to " + subject)
|
|
}
|
|
if b.seqs[subject] != after {
|
|
return fmt.Errorf("wrong last sequence for %s: %d", subject, b.seqs[subject])
|
|
}
|
|
if b.stream == nil {
|
|
b.stream, b.seqs = map[string][]byte{}, map[string]uint64{}
|
|
}
|
|
b.seq++
|
|
b.stream[subject], b.seqs[subject] = body, b.seq
|
|
b.publishes = append(b.publishes, subject)
|
|
return nil
|
|
}
|
|
|
|
// CreateDeclared is the bus's create of a stream: refused when refuseCreate, held when the bus holds it, and otherwise
|
|
// held from now on, as nats_streams then reads it.
|
|
func (b *fakeBus) CreateDeclared(cfg jetstream.StreamConfig) error {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.creates = append(b.creates, cfg.Name)
|
|
if b.refuseCreate {
|
|
return errors.New("Permissions Violation for Publish to $JS.API.STREAM.CREATE." + cfg.Name)
|
|
}
|
|
for _, s := range b.streams {
|
|
if s["name"] == cfg.Name {
|
|
return grants.ErrHeld
|
|
}
|
|
}
|
|
b.streams = append(b.streams, map[string]any{"name": cfg.Name, "subjects": cfg.Subjects, "retention": "limits",
|
|
"max_age_hours": int(cfg.MaxAge.Hours()), "max_per_subject": int(cfg.MaxMsgsPerSubject)})
|
|
return nil
|
|
}
|
|
|
|
func newFakeBus() *fakeBus {
|
|
return &fakeBus{states: map[string]map[string][]byte{}, refused: map[string]bool{}, own: map[string]map[string][]byte{},
|
|
revs: map[string]uint64{}, refuseTo: map[string]bool{}}
|
|
}
|
|
|
|
func (b *fakeBus) put(state, key string, v any) {
|
|
raw, err := json.Marshal(v)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
if b.states[state] == nil {
|
|
b.states[state] = map[string][]byte{}
|
|
}
|
|
b.states[state][key] = raw
|
|
}
|
|
|
|
func (b *fakeBus) Keys(state string) ([]string, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if b.refused[state] {
|
|
return nil, errors.New("Permissions Violation for Publish to $JS.API.STREAM.INFO.KV_" + state)
|
|
}
|
|
var out []string
|
|
for k := range b.states[state] {
|
|
out = append(out, k)
|
|
}
|
|
slices.Sort(out)
|
|
return out, nil
|
|
}
|
|
|
|
func (b *fakeBus) Get(state, key string) ([]byte, bool, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if b.refused[state] {
|
|
return nil, false, errors.New("refused")
|
|
}
|
|
v, ok := b.states[state][key]
|
|
return v, ok, nil
|
|
}
|
|
|
|
func (b *fakeBus) Call(key string, body any) (json.RawMessage, error) {
|
|
b.mu.Lock()
|
|
hold := b.hold
|
|
b.mu.Unlock()
|
|
if hold != nil && key == grants.CallUsers {
|
|
<-hold
|
|
}
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.calls = append(b.calls, key)
|
|
args, _ := body.(map[string]any)
|
|
switch key {
|
|
case grants.CallUsers:
|
|
part, _ := args["user"].(string)
|
|
if part == "" {
|
|
var names []map[string]any
|
|
for _, u := range b.loaded {
|
|
names = append(names, map[string]any{"user": u.User, "publish_grants": len(u.Publish)})
|
|
}
|
|
return json.Marshal(map[string]any{"users": names})
|
|
}
|
|
var out []grants.User
|
|
for _, u := range b.loaded {
|
|
if strings.Contains(u.User, part) {
|
|
out = append(out, u)
|
|
}
|
|
}
|
|
return json.Marshal(map[string]any{"users": out})
|
|
case grants.CallStreams:
|
|
return json.Marshal(map[string]any{"detail": b.streams})
|
|
case grants.CallRootFree:
|
|
if b.rootFree == nil {
|
|
return nil, errors.New("nats: no responders available for request")
|
|
}
|
|
out, err := b.rootFree(args["machines"].([]string))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return json.Marshal(out)
|
|
}
|
|
return nil, fmt.Errorf("no tool %s", key)
|
|
}
|
|
|
|
func (b *fakeBus) ownGet(state, key string) ([]byte, uint64, bool, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
v, ok := b.own[state][key]
|
|
return v, b.revs[state+"/"+key], ok, nil
|
|
}
|
|
|
|
type ownStore struct{ b *fakeBus }
|
|
|
|
func (s ownStore) Get(state, key string) ([]byte, uint64, bool, error) { return s.b.ownGet(state, key) }
|
|
func (s ownStore) Put(state, key string, value []byte) (uint64, error) {
|
|
b := s.b
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if b.refuseTo[state] {
|
|
return 0, errors.New("Permissions Violation for Publish to $KV.mesh-grants_" + state)
|
|
}
|
|
b.writes = append(b.writes, state)
|
|
if b.own[state] == nil {
|
|
b.own[state] = map[string][]byte{}
|
|
}
|
|
b.own[state][key] = value
|
|
b.revs[state+"/"+key]++
|
|
return b.revs[state+"/"+key], nil
|
|
}
|
|
|
|
func (b *fakeBus) Emit(event string, _ any) error {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.events = append(b.events, event)
|
|
return nil
|
|
}
|
|
|
|
func (b *fakeBus) count(key string) int {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
n := 0
|
|
for _, c := range b.calls {
|
|
if c == key {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
|
|
// manifestOf is a fixture manifest of one module, in the indexer's two buckets.
|
|
func (b *fakeBus) manifestOf(module, raw string) {
|
|
digest := delivery.ManifestDigest([]byte(raw))
|
|
b.put(grants.ReadManifests, digest, delivery.Manifest{V: delivery.Version, Module: module, JSON: raw})
|
|
b.put(grants.ReadModules, module, delivery.Module{V: delivery.Version, FP2: "src2:" + strings.Repeat("ab", 32), Manifest: digest, Repository: "novox/x", Commit: "c"})
|
|
}
|
|
|
|
func (b *fakeBus) assign(node, module string) {
|
|
b.put(grants.ReadAssignments, node+"."+module, map[string]any{"node": node, "module": module, "assigned": "2026-10-25T20:00:00Z"})
|
|
}
|
|
|
|
const (
|
|
natsManifest = `{"module":"nats","version":"1","claims":[{"name":"mesh-broker"}],"own-secrets":{"broker":"/var/lib/nats/broker"}}`
|
|
xManifest = `{"module":"mod-x","version":"1","own-secrets":{"broker":"/var/lib/x/broker"},"state":[{"name":"kept","history":3}],` +
|
|
`"streams":[{"name":"things","events":["thing.added"],"max-age-days":2,"max-mib":5}],"logs":[{"name":"history"}]}`
|
|
yManifest = `{"module":"mod-y","version":"1","own-secrets":{"broker":"/var/lib/y/broker"},"reads":["mod-x.kept"]}`
|
|
)
|
|
|
|
// fixture is a mesh of two machines: a runs the bus and mod-x, b runs mod-y; the bus node's agent loaded a list.
|
|
func fixture() *fakeBus {
|
|
b := newFakeBus()
|
|
b.manifestOf("nats", natsManifest)
|
|
b.manifestOf("mod-x", xManifest)
|
|
b.manifestOf("mod-y", yManifest)
|
|
b.assign("a", "nats")
|
|
b.assign("a", "mod-x")
|
|
b.assign("b", "mod-y")
|
|
b.put(grants.ReadAssignments, "b.gone", map[string]any{"node": "b", "module": "gone", "assigned": "2026-10-01T00:00:00Z", "ended": "2026-10-02T00:00:00Z"})
|
|
b.put(grants.ReadHave, "a.node", map[string]any{"v": 1})
|
|
b.put(grants.ReadHave, "b.node", map[string]any{"v": 1})
|
|
b.put(grants.ReadHave, "a.nats", delivery.Have{V: delivery.Version, Phase: delivery.PhaseRunning, Health: delivery.HealthOK,
|
|
UserList: &delivery.UserListWritten{Source: "a.nats.src2", Revision: 7, Wanted: "sha256:aa", Digest: "sha256:aa", Users: 6}})
|
|
b.put(grants.ReadControllerGrant, "mod-x.0123456789ab0123456789ab0123456789ab0123456789ab0123456789ab0123", delivery.ControllerGrant{V: delivery.Version, Accounts: []string{"a.mod-x"}})
|
|
return b
|
|
}
|
|
|
|
func partOn(b *fakeBus) *grants.Part {
|
|
return &grants.Part{Read: b, Call: b, Store: ownStore{b}, Emit: b, Publish: b, Now: func() time.Time { return time.Date(2026, 10, 25, 22, 0, 0, 0, time.UTC) }}
|
|
}
|
|
|
|
// shadowOf is the list the part composes from b's records now.
|
|
func shadowOf(t *testing.T, b *fakeBus) []grants.User {
|
|
t.Helper()
|
|
recs, _ := grants.Records(grants.ReadInputs(b, &grants.Manifests{}))
|
|
users, err := grants.ComposeUsers(recs)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return users
|
|
}
|
|
|
|
func names(us []grants.User) []string {
|
|
var out []string
|
|
for _, u := range us {
|
|
out = append(out, u.User)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func gapNamed(gs []grants.Gap, input string) bool {
|
|
return slices.ContainsFunc(gs, func(g grants.Gap) bool { return strings.Contains(g.Input, input) })
|
|
}
|
|
|
|
// TestGrants is the grants part's one table of checks (ADR 0547 clauses 5 to 8 in shadow, issue 1945): its manifest,
|
|
// the shadow list composed from bus values, the compare, each tool, that nothing is published or asserted, and that a
|
|
// refused read or write never stops it.
|
|
func TestGrants(t *testing.T) {
|
|
t.Run("manifest", func(t *testing.T) {
|
|
raw, err := os.ReadFile("module.json")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
m, err := catalogue.ParseManifest(raw)
|
|
if err != nil {
|
|
t.Fatalf("mesh-sdk's manifest reader refuses the manifest: %v", err)
|
|
}
|
|
if m.Module != "mesh-grants" || len(m.Claims) != 0 {
|
|
t.Errorf("module %q claims %v: the part claims no seat until mesh-sdk's seat list names mesh-grants", m.Module, m.Claims)
|
|
}
|
|
for _, row := range []struct {
|
|
what string
|
|
got, want []string
|
|
}{
|
|
{"reads", m.Reads, grants.Reads},
|
|
{"invokes", m.Invokes, grants.Invokes},
|
|
{"consumes", m.Consumes, grants.Consumes},
|
|
{"emits", m.Emits, grants.Emits},
|
|
{"tools", m.Tools, grants.Tools},
|
|
} {
|
|
if !slices.Equal(slices.Sorted(slices.Values(row.got)), slices.Sorted(slices.Values(row.want))) {
|
|
t.Errorf("%s = %v, the program's are %v", row.what, row.got, row.want)
|
|
}
|
|
}
|
|
var states []string
|
|
for _, s := range m.State {
|
|
states = append(states, s.Name)
|
|
}
|
|
if !slices.Equal(slices.Sorted(slices.Values(states)), grants.States) {
|
|
t.Errorf("state = %v, want %v", states, grants.States)
|
|
}
|
|
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || m.SecretsOwner != "mesh-grants" {
|
|
t.Errorf("own-secrets %v owner %q: a credential of its own, broker", m.OwnSecrets, m.SecretsOwner)
|
|
}
|
|
})
|
|
|
|
t.Run("no timer, no update or delete, a publish only of memberships, a create only of declared streams", func(t *testing.T) {
|
|
forbidden := []string{"time.Tick", "time.NewTicker", "time.Sleep", "time.After", "time.NewTimer", "time.AfterFunc",
|
|
"PublishMsg", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue", "DeleteStream", "PurgeStream",
|
|
"DeleteKeyValue", "CreateObjectStore", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("}
|
|
// The one publish: a membership, on its own subject, in publish.go; the one create: a declared stream the bus lacks,
|
|
// in create.go (issue 1790).
|
|
publishOnly := filepath.Join("cmd", "mesh-grants", "publish.go")
|
|
createOnly := filepath.Join("cmd", "mesh-grants", "create.go")
|
|
// And the operator's asks and conditions, on the seat operator-channel's subjects, in asking.go (ADR 0547 clause 8).
|
|
asksOnly := filepath.Join("cmd", "mesh-grants", "asking.go")
|
|
sealedOnly := filepath.Join("cmd", "mesh-grants", "sealed.go")
|
|
files := 0
|
|
for _, root := range []string{"cmd", "internal"} {
|
|
err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
|
|
if err != nil || d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
|
|
return err
|
|
}
|
|
files++
|
|
src, err := os.ReadFile(path)
|
|
if path != publishOnly && path != asksOnly && (strings.Contains(string(src), ".Publish(") || strings.Contains(string(src), "mesh.assignment")) {
|
|
t.Errorf("%s publishes: only publish.go publishes memberships, and asking.go the operator's asks", path)
|
|
}
|
|
if path == asksOnly && (strings.Count(string(src), ".Publish(") != 2 || strings.Contains(string(src), "mesh.assignment")) {
|
|
t.Errorf("asking.go publishes other than an ask, a cancel and a condition, through js.Publish twice")
|
|
}
|
|
if path == publishOnly && strings.Count(string(src), ".Publish(") != 1 {
|
|
t.Errorf("publish.go publishes other than one membership")
|
|
}
|
|
if n := strings.Count(string(src), "CreateStream"); (path == createOnly && n != 1) || (path != createOnly && n != 0) {
|
|
t.Errorf("%s holds CreateStream %d times: only create.go creates, once", path, n)
|
|
}
|
|
// The one object bucket: the part's own, of the user list sealed to the bus node, in sealed.go (ADR 0547 clause 7).
|
|
if path == sealedOnly {
|
|
if strings.Count(string(src), "CreateObjectStore(") != 1 || !strings.Contains(string(src), "delivery.GrantsUserListObjects") {
|
|
t.Errorf("sealed.go creates other than the part's one object bucket")
|
|
}
|
|
src = []byte(strings.ReplaceAll(string(src), "CreateObjectStore", ""))
|
|
}
|
|
for _, f := range forbidden {
|
|
if strings.Contains(string(src), f) {
|
|
t.Errorf("%s holds %s: the part holds no timer and creates or asserts nothing", path, f)
|
|
}
|
|
}
|
|
return err
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if files < 8 {
|
|
t.Fatalf("read %d source files, want the program's", files)
|
|
}
|
|
})
|
|
|
|
t.Run("shadow list from bus values", func(t *testing.T) {
|
|
for _, row := range []struct {
|
|
name string
|
|
bend func(*fakeBus)
|
|
users []string
|
|
gap string
|
|
}{
|
|
{"every record on the bus", func(*fakeBus) {},
|
|
[]string{"a.mod-x", "a.nats", "b.mod-y", "controller", "node.a", "node.b"}, "enrolment tokens"},
|
|
{"a manifest missing is a gap and its user is left out", func(b *fakeBus) { delete(b.states[grants.ReadModules], "mod-y") },
|
|
[]string{"a.mod-x", "a.nats", "controller", "node.a", "node.b"}, "manifest of mod-y"},
|
|
{"a want's built manifest is composed, not the trunk's", func(b *fakeBus) {
|
|
built := strings.Replace(xManifest, `"history":3}]`, `"history":3},{"name":"newer"}]`, 1)
|
|
digest := delivery.ManifestDigest([]byte(built))
|
|
b.put(grants.ReadManifests, digest, delivery.Manifest{V: delivery.Version, Module: "mod-x", JSON: built})
|
|
fp := "src2:" + strings.Repeat("cd", 32)
|
|
b.put(grants.ReadWants, "want.a.mod-x", delivery.Want{V: delivery.Version, FP: fp, Composed: "sha256:" + strings.Repeat("ef", 32)})
|
|
key, _ := delivery.BuilderBuildKey("mod-x", fp)
|
|
b.put(grants.ReadBuilds, key, map[string]any{"v": delivery.Version, "state": "done", "manifest": digest})
|
|
}, []string{"a.mod-x", "a.nats", "b.mod-y", "controller", "node.a", "node.b"}, "enrolment tokens"},
|
|
{"placement refused is a gap; the machines come from the haves", func(b *fakeBus) { b.refused[grants.ReadAssignments] = true },
|
|
[]string{"controller", "node.a", "node.b"}, grants.ReadAssignments},
|
|
} {
|
|
t.Run(row.name, func(t *testing.T) {
|
|
b := fixture()
|
|
row.bend(b)
|
|
in := grants.ReadInputs(b, &grants.Manifests{})
|
|
recs, left := grants.Records(in)
|
|
users, err := grants.ComposeUsers(recs)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := names(users); !slices.Equal(got, row.users) {
|
|
t.Errorf("users = %v, want %v", got, row.users)
|
|
}
|
|
if !gapNamed(append(in.Gaps, left...), row.gap) {
|
|
t.Errorf("gaps %v name no %q", in.Gaps, row.gap)
|
|
}
|
|
if strings.HasPrefix(row.name, "a want's") {
|
|
x := users[slices.IndexFunc(users, func(u grants.User) bool { return u.User == "a.mod-x" })]
|
|
if !slices.Contains(x.Publish, "$KV.mod-x_newer.>") {
|
|
t.Errorf("a.mod-x is not granted the state its wanted build declares: %v", x.Publish)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
// A module whose only runtime-loaded part is a helper bundle (route-proxy's route-provider) says `loads` in
|
|
// the repository manifest's build.artifacts; the indexer's manifest has no derived Bundles (issue 1945).
|
|
for _, row := range []struct {
|
|
name, build string
|
|
carried bool
|
|
}{
|
|
{"a bundle that says loads is carried by node-tools", `,"build":{"artifacts":[{"name":"helper","kind":"bundle","language":"go","system":"arch","from":"cmd/h","binary":"h","loads":["h"]}]}`, true},
|
|
{"a module with no loading bundle is left to its own account", ``, false},
|
|
} {
|
|
t.Run(row.name, func(t *testing.T) {
|
|
b := fixture()
|
|
b.manifestOf("mod-z", `{"module":"mod-z","version":"1","own-secrets":{"broker":"/var/lib/z/broker"}`+row.build+`}`)
|
|
b.assign("a", "mod-z")
|
|
b.manifestOf("node-tools", `{"module":"node-tools","version":"1","own-secrets":{"broker":"/var/lib/nt/broker"}}`)
|
|
b.assign("a", "node-tools")
|
|
users := shadowOf(t, b)
|
|
tools := users[slices.IndexFunc(users, func(u grants.User) bool { return u.User == "a.node-tools" })]
|
|
if got := slices.Contains(tools.Subscribe, "mesh.mod.mod-z.tool.>"); got != row.carried {
|
|
t.Errorf("a.node-tools subscribes to mod-z's tools = %v, want %v: %v", got, row.carried, tools.Subscribe)
|
|
}
|
|
})
|
|
}
|
|
b := fixture()
|
|
users := shadowOf(t, b)
|
|
y := users[slices.IndexFunc(users, func(u grants.User) bool { return u.User == "b.mod-y" })]
|
|
if !slices.Contains(y.Publish, "$JS.API.DIRECT.GET.KV_mod-x_kept.>") {
|
|
t.Errorf("b.mod-y reads mod-x.kept and is not granted its direct get: %v", y.Publish)
|
|
}
|
|
})
|
|
|
|
t.Run("root-free: the controller's word fills the composition, failing closed (issue 1945)", func(t *testing.T) {
|
|
const router = `{"module":"messenger","version":"1","runs-as":"messenger","resources":[{"id":"account","type":"user","name":"messenger","shell":"/usr/bin/nologin","home":"/var/lib/messenger"}],"own-secrets":{"broker":"/var/lib/m/broker"},"secrets-owner":"messenger",` +
|
|
`"seats":[{"name":"operator-channel","scope":"mesh","serves":["open"]},{"name":"channel","scope":"mesh","kinded":true,"accepts":["show"]}],` +
|
|
`"claims":[{"name":"operator-channel","scope":"mesh","serves":["open"]}]}`
|
|
const telegram = `{"module":"telegram","version":"1","runs-as":"telegram","resources":[{"id":"account","type":"user","name":"telegram","shell":"/usr/bin/nologin","home":"/var/lib/telegram"}],"own-secrets":{"broker":"/var/lib/t/broker"},"secrets-owner":"telegram",` +
|
|
`"claims":[{"name":"channel","scope":"mesh","kind":"telegram","capabilities":["deliver","verified-sender"]}]}`
|
|
answer := func(free ...string) func([]string) (any, error) {
|
|
return func(machines []string) (any, error) {
|
|
var vs []map[string]any
|
|
for _, m := range machines {
|
|
ok := slices.Contains(free, m)
|
|
why := "judged free"
|
|
if !ok {
|
|
why = "its node-engine judged nothing"
|
|
}
|
|
vs = append(vs, map[string]any{"machine": m, "free": ok, "why": why})
|
|
}
|
|
return map[string]any{"machines": vs}, nil
|
|
}
|
|
}
|
|
for _, row := range []struct {
|
|
name string
|
|
routerOn []string
|
|
telegramOn string
|
|
ask func([]string) (any, error)
|
|
keeps bool
|
|
note string
|
|
}{
|
|
{"router and telegram on one free machine", []string{"a"}, "a", answer("a"), true, ""},
|
|
{"router and telegram on two free machines", []string{"a"}, "b", answer("a", "b"), true, ""},
|
|
{"the holder's machine is not free", []string{"a"}, "b", answer("a"), false, "b: not free, its node-engine judged nothing"},
|
|
{"the router's machine is not free", []string{"a"}, "b", answer("b"), false, "a: not free"},
|
|
{"an error is not free", []string{"a"}, "a", func([]string) (any, error) { return nil, errors.New("boom") }, false, "boom"},
|
|
{"no answerer is not free", []string{"a"}, "a", nil, false, "no responders"},
|
|
{"a machine left out of the answer is not free", []string{"a"}, "a",
|
|
func([]string) (any, error) { return map[string]any{"machines": []any{}}, nil }, false, "named no verdict"},
|
|
{"the router on two machines frees nothing", []string{"a", "b"}, "a", answer("a", "b"), false, ""},
|
|
} {
|
|
t.Run(row.name, func(t *testing.T) {
|
|
b := fixture()
|
|
b.manifestOf("messenger", router)
|
|
b.manifestOf("telegram", telegram)
|
|
for _, node := range row.routerOn {
|
|
b.assign(node, "messenger")
|
|
}
|
|
b.assign(row.telegramOn, "telegram")
|
|
b.rootFree = row.ask
|
|
res := partOn(b).Compose()
|
|
recs, _ := grants.Records(grants.ReadInputs(b, &grants.Manifests{}))
|
|
grants.WithRootFree(&recs, b)
|
|
kept := false
|
|
for _, k := range buscompose.PlacementsOf(recs, recs.Interchangeable).Kinds {
|
|
kept = kept || slices.Contains(k.Capabilities, "verified-sender")
|
|
}
|
|
if kept != row.keeps {
|
|
t.Errorf("verified-sender kept = %v, want %v (said %v)", kept, row.keeps, res.RootNotFree)
|
|
}
|
|
if !strings.Contains(strings.Join(res.RootNotFree, " | "), row.note) || (row.note == "" && row.keeps && len(res.RootNotFree) != 0) {
|
|
t.Errorf("the pass says %v, want it to name %q", res.RootNotFree, row.note)
|
|
}
|
|
})
|
|
}
|
|
// A root-not-free condition raised or cleared wakes a pass; another kind does not (never a timer).
|
|
wb := fixture()
|
|
var handle func(string, []byte)
|
|
wf := &grants.Follower{Part: partOn(wb), Watch: &fakeWatcher{opened: make(chan string, 16)}, Logf: func(string, ...any) {},
|
|
Hear: func(h func(string, []byte)) error { handle = h; return nil }}
|
|
wctx, wcancel := context.WithCancel(context.Background())
|
|
wdone := make(chan struct{})
|
|
go func() { wf.Run(wctx); close(wdone) }()
|
|
waitFor(t, func() bool { return wb.count(grants.CallUsers) > 0 && handle != nil })
|
|
passes := func() int { return wb.count(grants.CallUsers) }
|
|
before := passes()
|
|
handle(grants.EventConditionRaised, []byte(`{"kind":"disk-full","key":"machine.a.disk-full"}`))
|
|
time.Sleep(100 * time.Millisecond)
|
|
if passes() != before {
|
|
t.Errorf("a condition of another kind woke a pass")
|
|
}
|
|
handle(grants.EventConditionCleared, []byte(`{"kind":"root-not-free","key":"machine.a.root-not-free"}`))
|
|
waitFor(t, func() bool { return passes() > before })
|
|
wcancel()
|
|
<-wdone
|
|
if !grants.IsRootNotFree([]byte(`{"kind":"root-not-free"}`)) || !grants.IsRootNotFree([]byte(`{"key":"machine.novox.root-not-free"}`)) ||
|
|
grants.IsRootNotFree([]byte(`{"kind":"disk-full","key":"machine.novox.disk-full"}`)) {
|
|
t.Errorf("only a root-not-free condition wakes a pass")
|
|
}
|
|
})
|
|
|
|
t.Run("a predefined-key fallback is read with no list; every refused manifest listed", func(t *testing.T) {
|
|
// mod-p's value falls back to a predefined key, as claude-code's managed_settings does, and no list is on the bus
|
|
// (issue 1945); mod-z's manifest carries a key mesh-sdk's reader refuses, and mod-z is assigned nowhere.
|
|
const pManifest = `{"module":"mod-p","version":"1","own-secrets":{"broker":"/var/lib/p/broker"},` +
|
|
`"config":{"size":{"type":"number","description":"The size text shows in, in points.","predefined":"font-size","default":12,"runtime":true}}}`
|
|
const zManifest = `{"module":"mod-z","version":"1","no-such-key":true}`
|
|
b := fixture()
|
|
b.manifestOf("mod-p", pManifest)
|
|
b.manifestOf("mod-z", zManifest)
|
|
b.assign("b", "mod-p")
|
|
// mesh-wireguard is provided by the mesh, not read from a repository: no manifest in the indexer, none refused,
|
|
// no gap, and its manifest is the one mesh-sdk provides (issue 1945).
|
|
b.assign("b", "mesh-wireguard")
|
|
in := grants.ReadInputs(b, &grants.Manifests{})
|
|
// An indexer row of the same name never wins: the provided manifest is used and the row is said as a gap.
|
|
bw := fixture()
|
|
bw.manifestOf("mesh-wireguard", `{"module":"mesh-wireguard","version":"9","own-secrets":{"broker":"/x"}}`)
|
|
bw.assign("b", "mesh-wireguard")
|
|
if iw := grants.ReadInputs(bw, &grants.Manifests{}); !strings.HasPrefix(iw.Digests["mesh-wireguard"], "provided:") || !gapNamed(iw.Gaps, "manifest of mesh-wireguard") || len(iw.Refused) != 0 {
|
|
t.Errorf("an indexer manifest of mesh-wireguard won or went unsaid: %v %v", iw.Digests["mesh-wireguard"], iw.Gaps)
|
|
}
|
|
if _, ok := in.Manifests["mesh-wireguard"]; !ok || gapNamed(in.Gaps, "manifest of mesh-wireguard") || !strings.HasPrefix(in.Digests["mesh-wireguard"], "provided:") {
|
|
t.Errorf("mesh-wireguard's manifest is not taken from mesh-sdk: gaps %v", in.Gaps)
|
|
}
|
|
if len(in.Refused) != 1 || in.Refused[0].Input != "manifest of mod-z" || !gapNamed(in.Gaps, "manifest of mod-z") {
|
|
t.Errorf("refused = %v: only mod-z's, and named in the gaps", in.Refused)
|
|
}
|
|
recs, _ := grants.Records(in)
|
|
users, err := grants.ComposeUsers(recs)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !slices.Contains(names(users), "b.mod-p") {
|
|
t.Errorf("b.mod-p is not composed, so its module counts as placed nowhere: %v", names(users))
|
|
}
|
|
// The pass, the tool listing held memberships, and the log each say the refusal.
|
|
part := partOn(b)
|
|
res := part.Pass()
|
|
if len(res.Refused) != 1 || !strings.HasPrefix(res.Refused[0], "manifest of mod-z: ") {
|
|
t.Errorf("the pass says refused %v", res.Refused)
|
|
}
|
|
held, err := part.Held()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got, _ := held.(map[string]any)["refused"].([]string); len(got) != 1 {
|
|
t.Errorf("grants_membership lists refused %v", got)
|
|
}
|
|
var logged []string
|
|
var lmu sync.Mutex
|
|
f := &grants.Follower{Part: partOn(b), Watch: &fakeWatcher{opened: make(chan string, 16)}, Logf: func(format string, args ...any) {
|
|
lmu.Lock()
|
|
logged = append(logged, fmt.Sprintf(format, args...))
|
|
lmu.Unlock()
|
|
}}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { f.Run(ctx); close(done) }()
|
|
waitFor(t, func() bool {
|
|
lmu.Lock()
|
|
defer lmu.Unlock()
|
|
return strings.Contains(strings.Join(logged, "\n"), "refused manifest of mod-z: ")
|
|
})
|
|
cancel()
|
|
<-done
|
|
})
|
|
|
|
t.Run("credentials", func(t *testing.T) {
|
|
shadow := []grants.User{{User: "a.held"}, {User: "a.new"}, {User: "a.bare"}}
|
|
loaded := []grants.User{{User: "a.held"}, {User: "a.bare"}, {User: "enrol.x"}, {User: "person.op"}}
|
|
hashes := map[string]grants.HeldHash{"a.held": {Hash: "$2a$10$h1", Generation: 2}, "a.new": {Hash: "$2a$10$h2", Generation: 1},
|
|
"person.op": {Hash: "$2a$10$h3", Adopted: true}}
|
|
c := grants.CredentialsOf(shadow, loaded, hashes)
|
|
for _, row := range []struct {
|
|
user, would string
|
|
held, adopted bool
|
|
}{
|
|
{"a.bare", grants.WouldLeaveOut, false, false},
|
|
{"a.held", grants.WouldSame, true, false},
|
|
{"a.new", grants.WouldAdd, true, false},
|
|
{"enrol.x", grants.WouldLoseLogin, false, false},
|
|
{"person.op", grants.WouldNotComposed, true, true},
|
|
} {
|
|
i := slices.IndexFunc(c.Users, func(u grants.Credential) bool { return u.User == row.user })
|
|
if i < 0 || c.Users[i].Would != row.would || c.Users[i].HashHeld != row.held || c.Users[i].Adopted != row.adopted {
|
|
t.Errorf("%s: %+v, want %s held %v", row.user, c.Users, row.would, row.held)
|
|
}
|
|
}
|
|
if c.HashesHeld != 3 || !slices.Equal(c.LeftOut, []string{"a.bare"}) || !slices.Equal(c.LosesLogin, []string{"enrol.x"}) ||
|
|
!slices.Equal(c.Adds, []string{"a.new"}) {
|
|
t.Errorf("credentials = %+v", c)
|
|
}
|
|
// Composed from the vault's state: a hash changes the list's state (written on the event), and no hash is ever
|
|
// in what the part writes or says.
|
|
b := fixture()
|
|
b.loaded = shadowOf(t, b)
|
|
p := partOn(b)
|
|
before := p.Compose()
|
|
b.put(grants.ReadAccounts, "a.mod-x", map[string]any{"user": "a.mod-x", "hash": "$2a$10$secrethashbytes", "generation": 1})
|
|
after := p.Compose()
|
|
cr := after.UserList.Credentials
|
|
if !after.Written || after.Revision != before.Revision+1 || cr.HashesHeld != 1 || slices.Contains(cr.LeftOut, "a.mod-x") ||
|
|
!slices.Contains(cr.LeftOut, "b.mod-y") {
|
|
t.Errorf("a hash heard = %+v", after.UserList.Credentials)
|
|
}
|
|
why := strings.Join(p.Decide(), "; ")
|
|
if !strings.Contains(why, "no hash for") || strings.Contains(why, "a.mod-x,") {
|
|
t.Errorf("take-over is refused for %q, want the users with no hash named", why)
|
|
}
|
|
for state, keys := range b.own {
|
|
for k, v := range keys {
|
|
if strings.Contains(string(v), "secrethashbytes") {
|
|
t.Errorf("the hash is written in %s %s", state, k)
|
|
}
|
|
}
|
|
}
|
|
})
|
|
|
|
t.Run("compare", func(t *testing.T) {
|
|
b := fixture()
|
|
shadow := shadowOf(t, b)
|
|
in := grants.ReadInputs(b, &grants.Manifests{})
|
|
changed := slices.Clone(shadow)
|
|
x := slices.IndexFunc(changed, func(u grants.User) bool { return u.User == "a.mod-x" })
|
|
changed[x].Publish = append(slices.Clone(changed[x].Publish[1:]), "extra.subject")
|
|
for _, row := range []struct {
|
|
name string
|
|
loaded []grants.User
|
|
added, dropped, gone []string
|
|
differ []string
|
|
equal bool
|
|
}{
|
|
{"equal", shadow, nil, nil, nil, nil, true},
|
|
{"one user the bus lacks", shadow[1:], []string{shadow[0].User}, nil, nil, nil, false},
|
|
{"one loaded user still placed", append(slices.Clone(shadow), grants.User{User: "a.mod-x.worker"}), nil, []string{"a.mod-x.worker"}, nil, nil, false},
|
|
{"one loaded user placement no longer names", append(slices.Clone(shadow), grants.User{User: "b.gone"}), nil, []string{"b.gone"}, []string{"b.gone"}, nil, true},
|
|
{"grants differ", changed, nil, nil, nil, []string{"a.mod-x"}, false},
|
|
} {
|
|
t.Run(row.name, func(t *testing.T) {
|
|
c := grants.Compare(shadow, row.loaded, grants.Placed(in))
|
|
var differ []string
|
|
for _, d := range c.Differ {
|
|
differ = append(differ, d.User)
|
|
}
|
|
if !slices.Equal(c.Added, nonNil(row.added)) || !slices.Equal(c.Dropped, nonNil(row.dropped)) ||
|
|
!slices.Equal(c.Gone, row.gone) || !slices.Equal(differ, row.differ) || c.Equal() != row.equal {
|
|
t.Errorf("compare = %+v", c)
|
|
}
|
|
if row.name == "grants differ" && (len(c.Differ[0].PublishAdded) != 1 || c.Differ[0].PublishDropped[0] != "extra.subject") {
|
|
t.Errorf("the grant difference is not named: %+v", c.Differ[0])
|
|
}
|
|
})
|
|
}
|
|
})
|
|
|
|
t.Run("compare excuses a retired verb and a multiply claimed mesh seat's added subscriptions on its machines, nothing else", func(t *testing.T) {
|
|
const retired, store, storeInfo = "mesh.seat.mesh-delivery.tool.mark-ready", "mesh.seat.mesh-store.tool.databases", "$SRV.INFO.mesh-store.>"
|
|
two := map[string][]string{"mesh-store": {"a", "n"}}
|
|
for _, row := range []struct {
|
|
name string
|
|
user string
|
|
subShadow, subBus, pubShadow []string
|
|
multi map[string][]string
|
|
differs bool
|
|
excused int
|
|
}{
|
|
{"a retired verb the shadow dropped", "a.node-tools", []string{"x"}, []string{"x", retired}, nil, nil, false, 1},
|
|
{"a verb of a live seat the shadow lost", "a.node-tools", []string{"x"}, []string{"x", store}, nil, two, true, 0},
|
|
{"a seat claimed on two machines, added to a user on one", "a.node-tools", []string{"x", store, storeInfo}, []string{"x"}, nil, two, false, 2},
|
|
{"the same added to a user on a machine that does not claim it", "q.node-tools", []string{"x", store}, []string{"x"}, nil, two, true, 0},
|
|
{"a publish of the seat added is not excused", "a.node-tools", []string{"x"}, []string{"x"}, []string{store}, two, true, 0},
|
|
{"the same added for a seat claimed once", "a.node-tools", []string{"x", store}, []string{"x"}, nil, nil, true, 0},
|
|
} {
|
|
c := grants.Compare([]grants.User{{User: row.user, Subscribe: row.subShadow, Publish: row.pubShadow}}, []grants.User{{User: row.user, Subscribe: row.subBus}}, nil).
|
|
Excusing(grants.ExcusedBy(row.multi))
|
|
if c.Equal() == row.differs || len(c.Excused) != row.excused {
|
|
t.Errorf("%s: equal %v, excused %v", row.name, c.Equal(), c.Excused)
|
|
}
|
|
}
|
|
})
|
|
|
|
t.Run("tools", func(t *testing.T) {
|
|
b := fixture()
|
|
b.loaded = shadowOf(t, b)
|
|
b.streams = []map[string]any{{"name": "KV_mod-x_kept", "subjects": []string{"$KV.mod-x_kept.>"}, "retention": "limits", "max_per_subject": 3}}
|
|
p := partOn(b)
|
|
// The bus node says its key in its have of the bus module; the process puts the sealed list (ADR 0547 clause 7).
|
|
busKey, err := seal.GenerateKeyPair()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
withKey := func(w delivery.UserListWritten) {
|
|
w.Key = busKey.PublicKey
|
|
b.put(grants.ReadHave, "a.nats", delivery.Have{V: delivery.Version, Phase: delivery.PhaseRunning, Health: delivery.HealthOK, UserList: &w})
|
|
}
|
|
loadedBefore := delivery.UserListWritten{Source: "a.nats.src2", Revision: 7, Wanted: "sha256:aa", Digest: "sha256:aa", Users: 6}
|
|
withKey(loadedBefore)
|
|
objs := &fakeObjects{}
|
|
p.Objects = objs
|
|
open := func(g delivery.GrantsUserList) string {
|
|
t.Helper()
|
|
list, err := delivery.OpenUserList(objs.objects[g.Object], g.Digest, func(bx seal.Box) ([]byte, error) {
|
|
out, err := seal.Open(bx, busKey.PrivateKey)
|
|
return []byte(out), err
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("the sealed list does not open with the bus node's key: %v", err)
|
|
}
|
|
return string(list)
|
|
}
|
|
|
|
got, _ := p.UserList()
|
|
if !strings.Contains(fmt.Sprint(got), "nothing composed yet") {
|
|
t.Errorf("user-list before any composition = %v", got)
|
|
}
|
|
first := p.Compose()
|
|
if !first.Written || first.Revision != 1 || !first.UserList.Compare.Equal() || first.UserList.Users != 6 {
|
|
t.Fatalf("first composition = %+v", first)
|
|
}
|
|
// Written in shadow, the bus node told to follow the controller's list; every user left out while the vault holds
|
|
// no hash; a tool's part (no object bucket) writes none.
|
|
if g, _, had := p.Sealed(); !had || g.Follow != delivery.FollowController || g.Revision != 1 || g.Node != "a" ||
|
|
g.Users != 0 || len(g.LeftOut) != 6 || len(objs.objects) != 1 || !strings.Contains(open(g), "accounts {") {
|
|
t.Errorf("the sealed list in shadow = %+v, %d objects", g, len(objs.objects))
|
|
}
|
|
if _, _, had := partOn(fixture()).Sealed(); had {
|
|
t.Error("a part read before any pass holds a sealed list")
|
|
}
|
|
if tp := partOn(fixture()); tp.Compose().Sealed == nil {
|
|
t.Error("a tool's pass does not say the sealed list")
|
|
} else if _, _, had := tp.Sealed(); had {
|
|
t.Error("a tool's part wrote the sealed list")
|
|
}
|
|
if len(first.Published) != 3 || len(b.publishes) != 3 {
|
|
t.Errorf("the first composition published %v, want the three assignments' memberships", first.Published)
|
|
}
|
|
// The second pass says each membership unchanged (published at the first); the third writes nothing at all.
|
|
if again := p.Compose(); again.Written || again.Revision != 1 || len(again.Published) != 0 || again.AssertionsWritten != 0 {
|
|
t.Errorf("a composition with no input changed wrote or published: %+v", again)
|
|
}
|
|
if third := p.Compose(); third.Written || third.MembershipsWritten != 0 || len(third.Published) != 0 || len(b.publishes) != 3 {
|
|
t.Errorf("a membership equal to the one last published was published again, or a state written: %+v", third)
|
|
}
|
|
// mod-x placed on b too: b.mod-x is added and its first membership published; a.mod-x's would lose the module's
|
|
// plain subject (mod-x now runs on two machines), a loss, so it stays a shadow difference.
|
|
b.assign("b", "mod-x")
|
|
moved := p.Compose()
|
|
if !moved.Written || moved.Revision != 2 || !slices.Equal(moved.Published, []string{"b.mod-x"}) || len(moved.NotPublished) != 1 || !strings.HasPrefix(moved.NotPublished[0], "a.mod-x: "+grants.ShadowDifference) || moved.UserList.Compare.Equal() ||
|
|
!slices.Equal(moved.UserList.Compare.Added, []string{"b.mod-x"}) {
|
|
t.Errorf("a composition after placement moved = %+v", moved)
|
|
}
|
|
got, _ = p.UserList()
|
|
ul, _ := got.(map[string]any)
|
|
if st, ok := ul["written"].(grants.UserListState); !ok || st.Loaded == nil || st.Loaded.Digest != "sha256:aa" || st.BusNode != "a" ||
|
|
ul["revision"] != uint64(2) || !gapNamed(st.Gaps, "people") {
|
|
t.Errorf("user-list = %+v", got)
|
|
}
|
|
|
|
a, err := p.AssertOne("mod-x")
|
|
am, _ := a.(map[string]any)
|
|
as, _ := am["assertion"].(grants.Assertion)
|
|
if err != nil || as.Asserted || as.Says != grants.ShadowAsserts || len(as.Items) != 3 || am["asked"] == nil {
|
|
t.Fatalf("assert = %+v, %v", a, err)
|
|
}
|
|
for _, it := range as.Items {
|
|
if it.Stream == "KV_mod-x_kept" && len(it.Differences) != 0 {
|
|
t.Errorf("a bucket the bus holds as composed differs: %v", it.Differences)
|
|
}
|
|
if it.Stream == "THINGS" && !slices.Contains(it.Differences, "the bus holds no stream THINGS") {
|
|
t.Errorf("a stream the bus lacks is not named: %+v", it)
|
|
}
|
|
}
|
|
|
|
ms, _ := p.MembershipOf("a", "mod-x")
|
|
if mo, _ := ms.(map[string]any); !strings.Contains(fmt.Sprint(mo["published"]), "last_pass") || mo["composed"] == nil {
|
|
t.Errorf("membership = %v", ms)
|
|
}
|
|
gr, _ := p.Grants("mod-x")
|
|
if g, _ := gr.(map[string]any); g["writes"] != grants.WritesNoGrant || g["count"] != 1 {
|
|
t.Errorf("grants = %v", gr)
|
|
}
|
|
|
|
b.loaded = append(b.loaded, grants.User{User: "b.gone"})
|
|
fa := &fakeAsker{wheres: map[string]*asks.Warrant{}}
|
|
var forgot []string
|
|
n := 0
|
|
p.Nonce, p.Forget = func() string { n++; return fmt.Sprintf("%04d", n) }, func(id string) error { forgot = append(forgot, id); return nil }
|
|
var said []asks.Condition
|
|
p.Asker, p.Say = fa, func(c asks.Condition) error { said = append(said, c); return nil }
|
|
warrant := func(id string) asks.Warrant { return asks.Warrant{Ask: id, Outcome: asks.OutcomeChosen} }
|
|
last := func() grants.Act {
|
|
raw, _, _, _ := b.ownGet(grants.StateTakeover, grants.KeySwitch)
|
|
var s grants.Switch
|
|
_ = json.Unmarshal(raw, &s)
|
|
if s.Open != "" && s.Acts[len(s.Acts)-1].Said != "asked" {
|
|
t.Errorf("an ask is open and its act is not asked: %+v", s)
|
|
}
|
|
return s.Acts[len(s.Acts)-1]
|
|
}
|
|
asked := func(act string) string {
|
|
t.Helper()
|
|
var res struct{ Asked string }
|
|
var err error
|
|
var out any
|
|
if act == "take-over" {
|
|
out, err = p.TakeOver("shanks.node-tools", "a row")
|
|
} else {
|
|
out, err = p.HandBack("shanks.node-tools", "a row")
|
|
}
|
|
if err != nil {
|
|
t.Fatalf("%s: %v", act, err)
|
|
}
|
|
raw, _ := json.Marshal(out)
|
|
_ = json.Unmarshal(raw, &res)
|
|
return res.Asked
|
|
}
|
|
yes, no := asks.Option{ID: "yes"}, asks.Option{ID: "no"}
|
|
for _, row := range []struct {
|
|
name, act string
|
|
option asks.Option
|
|
setup func()
|
|
said, never []string // the refusal's words and the users it must not name
|
|
state string
|
|
outcome string
|
|
}{
|
|
{"a yes while a user differs is refused and said, naming it", "take-over", yes, func() {}, []string{"users differ", "b.mod-x"}, []string{"b.gone"}, grants.Shadow, "refused"},
|
|
{"a no leaves the switch", "take-over", no, func() {}, nil, nil, grants.Shadow, "declined"},
|
|
{"a yes with the lists agreeing and every hash held records the switch", "take-over", yes, func() {
|
|
b.states[grants.ReadAssignments] = nil
|
|
b.assign("a", "nats")
|
|
b.assign("a", "mod-x")
|
|
b.assign("b", "mod-y")
|
|
b.loaded = append(shadowOf(t, b), grants.User{User: "b.gone"})
|
|
for _, u := range shadowOf(t, b) {
|
|
b.put(grants.ReadAccounts, u.User, map[string]any{"user": u.User, "hash": "$2a$10$x", "generation": 1})
|
|
}
|
|
}, nil, nil, grants.TookOver, "recorded"},
|
|
{"a yes to hand back records the switch back", "hand-back", yes, func() {}, nil, nil, grants.Shadow, "recorded"},
|
|
} {
|
|
t.Run(row.name, func(t *testing.T) {
|
|
row.setup()
|
|
before := len(fa.asks)
|
|
id := asked(row.act)
|
|
if _, err := p.TakeOver("shanks.node-tools", "again"); err == nil || !strings.Contains(err.Error(), "already asked") {
|
|
t.Errorf("a second ask while one is open = %v", err)
|
|
}
|
|
p.Compose()
|
|
p.Compose()
|
|
if len(fa.asks) != before+1 || fa.asks[before].ID != id {
|
|
t.Fatalf("the operator was asked %d times, want once: %+v", len(fa.asks)-before, fa.asks)
|
|
}
|
|
ask := fa.asks[before]
|
|
if err := ask.Check(time.Date(2026, 10, 25, 22, 0, 0, 0, time.UTC)); err != nil {
|
|
t.Errorf("the router would refuse the ask: %v", err)
|
|
}
|
|
if yesOpt, ok := ask.Option("yes"); !ok || yesOpt.Level != asks.Approve || !strings.HasPrefix(yesOpt.Binds, "sha256:") {
|
|
t.Errorf("the yes does not approve and bind its act: %+v", yesOpt)
|
|
}
|
|
said = nil
|
|
if err := p.Warranted(row.option, warrant(id)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
l := last()
|
|
if l.Said != row.outcome || l.Ask != id || p.SwitchState() != row.state {
|
|
t.Errorf("the act = %+v, the switch %s; want %s and %s", l, p.SwitchState(), row.outcome, row.state)
|
|
}
|
|
for _, w := range row.said {
|
|
if !strings.Contains(l.Refused, w) || (row.outcome == "refused" && (len(said) != 1 || !strings.Contains(said[0].Explanation, w))) {
|
|
t.Errorf("the refusal %q (said %v) does not say %q", l.Refused, said, w)
|
|
}
|
|
}
|
|
for _, w := range row.never {
|
|
if strings.Contains(l.Refused, w) {
|
|
t.Errorf("the refusal names %q", w)
|
|
}
|
|
}
|
|
// A recorded switch says no refusal and clears the one a refused yes left raised.
|
|
if row.outcome == "recorded" && row.act == "take-over" && (len(said) != 1 || said[0].State != asks.StateCleared || said[0].Key != "mesh-grants.take-over-refused") {
|
|
t.Errorf("a recorded switch did not clear the refusal: %v", said)
|
|
}
|
|
if row.outcome == "recorded" && !strings.Contains(fmt.Sprint(p.TakeOverState()), "bus_node_loaded_grants_list:not yet") {
|
|
t.Errorf("the state claims the bus node follows with no loaded list: %v", fmt.Sprint(p.TakeOverState()))
|
|
}
|
|
// The key the bus node follows turns with the switch at once, and the state says switched only once the bus
|
|
// node loaded the grants part's written list.
|
|
g, _, _ := p.Sealed()
|
|
if want := map[string]string{grants.TookOver: delivery.FollowGrants, grants.Shadow: delivery.FollowController}[p.SwitchState()]; g.Follow != want {
|
|
t.Errorf("the key says follow %s with the switch %s", g.Follow, p.SwitchState())
|
|
}
|
|
effect := func() string { st, _ := p.TakeOverState(); return fmt.Sprint(st.(map[string]any)["effect"]) }
|
|
if row.outcome == "recorded" && row.act == "take-over" {
|
|
if e := effect(); strings.HasPrefix(e, "switched") || !strings.Contains(e, "not in effect") {
|
|
t.Errorf("effect before the bus node loaded = %s", e)
|
|
}
|
|
if users := strings.Count(open(g), "{ user: "); users != g.Users || g.Users == 0 || len(g.LeftOut) != 0 {
|
|
t.Errorf("the sealed list holds %d users, the key says %d, left out %v", users, g.Users, g.LeftOut)
|
|
}
|
|
withKey(delivery.UserListWritten{Source: delivery.GrantsUserListSource, Revision: g.Revision, Wanted: g.Digest, Digest: g.Digest, Users: g.Users, Follows: delivery.FollowGrants})
|
|
if e := effect(); !strings.HasPrefix(e, "switched") {
|
|
t.Errorf("effect once the bus node loaded the written list = %s", e)
|
|
}
|
|
withKey(loadedBefore)
|
|
}
|
|
})
|
|
}
|
|
t.Run("hand-back in shadow is refused at once, with nothing asked", func(t *testing.T) {
|
|
n := len(fa.asks)
|
|
if _, err := p.HandBack("shanks.node-tools", "a row"); err == nil || !strings.Contains(err.Error(), "nothing to hand back") {
|
|
t.Errorf("hand-back in shadow = %v", err)
|
|
}
|
|
p.Compose()
|
|
if len(fa.asks) != n {
|
|
t.Errorf("a refused hand-back asked the operator")
|
|
}
|
|
})
|
|
t.Run("an ask that can stay open is closed every way and can be asked again", func(t *testing.T) {
|
|
// A warrant the process missed is caught up from the router's record, whatever its outcome.
|
|
id := asked("take-over")
|
|
p.Compose()
|
|
fa.wheres[id] = &asks.Warrant{Ask: id, Outcome: asks.OutcomeChosen, Option: "no"}
|
|
p.Compose()
|
|
if l := last(); l.Ask != id || l.Said != "declined" {
|
|
t.Errorf("a missed warrant was not caught up: %+v", l)
|
|
}
|
|
// The router ended it with no choice, and the process was not told.
|
|
id = asked("take-over")
|
|
p.Compose()
|
|
fa.wheres[id] = &asks.Warrant{Ask: id, Outcome: asks.OutcomeCancelled}
|
|
p.Compose()
|
|
if l := last(); l.Ask != id || !strings.Contains(l.Said, "cancelled") {
|
|
t.Errorf("an ask the router ended was not closed: %+v", l)
|
|
}
|
|
// Said closes on any outcome for the open id, and none for another id.
|
|
id = asked("take-over")
|
|
p.Said(asker.Refused, asks.Warrant{Ask: "other", Outcome: asks.OutcomeRefused}, nil)
|
|
if _, err := p.TakeOver("x", "y"); err == nil {
|
|
t.Errorf("a warrant for another ask closed the open one")
|
|
}
|
|
p.Said(asker.Refused, asks.Warrant{Ask: id, Outcome: asks.OutcomeRefused}, nil)
|
|
// A warrant for an ask that is not the open one switches nothing.
|
|
if err := p.Warranted(yes, warrant("take-over-other")); err == nil {
|
|
t.Errorf("a warrant for an ask that is not open was acted on")
|
|
}
|
|
// Past its expiry and a margin the part closes it itself; the clock moves, not a timer.
|
|
id = asked("take-over")
|
|
now := p.Now
|
|
p.Now = func() time.Time { return now().Add(25 * time.Hour) }
|
|
p.Compose()
|
|
p.Now = now
|
|
if l := last(); l.Ask != id || l.Said != string(asks.OutcomeExpired) {
|
|
t.Errorf("an expired ask was not closed: %+v", l)
|
|
}
|
|
// An ask that could not be published is closed as refused, its half-made record taken away, and is not retried.
|
|
fa.fail = true
|
|
id = asked("take-over")
|
|
p.Compose()
|
|
p.Compose()
|
|
if l := last(); l.Ask != id || l.Said != "refused" || !slices.Contains(forgot, id) || !slices.Contains(fa.cancels, id) {
|
|
t.Errorf("a failed ask stays open, or is not taken back at the router: %+v, forgot %v, cancelled %v", l, forgot, fa.cancels)
|
|
}
|
|
fa.fail = false
|
|
// An answer the part's own record cannot take yet (its state out of reach) leaves the ask open, closed by
|
|
// nothing, and is taken at a later pass.
|
|
id = asked("take-over")
|
|
p.Compose()
|
|
fa.wheres[id] = &asks.Warrant{Ask: id, Outcome: asks.OutcomeChosen, Option: "no"}
|
|
fa.takeErr = errors.New("nats: no responders available for request")
|
|
p.Compose()
|
|
if l := last(); l.Ask != id || l.Said != "asked" {
|
|
t.Errorf("an answer not taken closed the ask: %+v", l)
|
|
}
|
|
fa.takeErr = nil
|
|
p.Compose()
|
|
if l := last(); l.Ask != id || l.Said != "declined" {
|
|
t.Errorf("the answer was not taken once its record could be reached: %+v", l)
|
|
}
|
|
asked("take-over")
|
|
})
|
|
|
|
t.Run("only memberships published, nothing asserted", func(t *testing.T) {
|
|
for _, s := range b.publishes {
|
|
if !strings.HasPrefix(s, "mesh.assignment.") {
|
|
t.Errorf("published on %s", s)
|
|
}
|
|
}
|
|
for _, w := range b.writes {
|
|
if !slices.Contains(grants.States, w) {
|
|
t.Errorf("wrote %s, which is not the part's own state", w)
|
|
}
|
|
}
|
|
for _, c := range b.calls {
|
|
if c != grants.CallUsers && c != grants.CallStreams {
|
|
t.Errorf("called %s: the part calls only the bus module's read tools", c)
|
|
}
|
|
}
|
|
for _, e := range b.events {
|
|
if !slices.Contains(grants.Emits, e) {
|
|
t.Errorf("said %s", e)
|
|
}
|
|
}
|
|
})
|
|
})
|
|
|
|
t.Run("memberships: a first or an additive one published, a loss never", func(t *testing.T) {
|
|
const key = "a.mod-x"
|
|
composed := func(b *fakeBus) map[string]any {
|
|
recs, _ := grants.Records(grants.ReadInputs(b, &grants.Manifests{}))
|
|
raw, _ := json.Marshal(grants.Memberships(recs)[key].Membership)
|
|
var m map[string]any
|
|
_ = json.Unmarshal(raw, &m)
|
|
return m
|
|
}
|
|
for _, row := range []struct {
|
|
name string
|
|
last func(map[string]any) map[string]any
|
|
published bool
|
|
names string
|
|
}{
|
|
{"a first membership is published", nil, true, ""},
|
|
{"an additive change is published, the private network carried", func(m map[string]any) map[string]any {
|
|
delete(m, "state")
|
|
m["mesh"] = []any{"10.0.0.1"}
|
|
return m
|
|
}, true, ""},
|
|
{"a change that drops verified-sender is not published, and named", func(m map[string]any) map[string]any {
|
|
m["seat-traffic"] = map[string]any{"kinds": []any{map[string]any{"seat": "channel", "kind": "telegram", "capabilities": []any{"verified-sender"}}}}
|
|
return m
|
|
}, false, "verified-sender"},
|
|
{"a change that drops any other field is not published", func(m map[string]any) map[string]any {
|
|
m["serves"] = append(m["serves"].([]any), map[string]any{"subject": "mesh.mod.mod-x.tool.gone"})
|
|
return m
|
|
}, false, "tool.gone"},
|
|
// A verb the seat catalogue retired on purpose (mark-ready, mesh-sdk d506f21, ADR 0421) is dropped and named; a
|
|
// verb of a seat the catalogue does not define is still a loss (issue 1945).
|
|
{"a change that drops only a retired seat verb is published, the verb named", func(m map[string]any) map[string]any {
|
|
m["seats"] = []any{map[string]any{"seat": "mesh-delivery", "verb": "mark-ready", "subject": "mesh.seat.mesh-delivery.tool.mark-ready"}}
|
|
return m
|
|
}, true, "mesh-delivery mark-ready"},
|
|
{"a retired verb and a module seat's verb dropped: not published, the module seat's named", func(m map[string]any) map[string]any {
|
|
m["seats"] = []any{map[string]any{"seat": "mesh-delivery", "verb": "mark-ready", "subject": "mesh.seat.mesh-delivery.tool.mark-ready"},
|
|
map[string]any{"seat": "telegram-sender", "verb": "send", "subject": "mesh.seat.telegram-sender.tool.send"}}
|
|
return m
|
|
}, false, "telegram-sender"},
|
|
} {
|
|
t.Run(row.name, func(t *testing.T) {
|
|
b := fixture()
|
|
b.loaded = shadowOf(t, b)
|
|
if row.last != nil {
|
|
raw, _ := json.Marshal(row.last(composed(b)))
|
|
b.stream = map[string][]byte{"mesh.assignment." + key: raw}
|
|
b.seqs = map[string]uint64{"mesh.assignment." + key: 4}
|
|
}
|
|
res := partOn(b).Compose()
|
|
if got := slices.Contains(res.Published, key); got != row.published {
|
|
t.Fatalf("published %v, not published %v", res.Published, res.NotPublished)
|
|
}
|
|
if row.published {
|
|
if row.names == "" && row.last != nil && !strings.Contains(string(b.stream["mesh.assignment."+key]), "10.0.0.1") {
|
|
t.Errorf("the private network of the last membership is not carried: %s", b.stream["mesh.assignment."+key])
|
|
}
|
|
if raw, _, _, _ := b.ownGet(grants.StateMemberships, key); !strings.Contains(string(raw), row.names) {
|
|
t.Errorf("the pass does not name %q: %s", row.names, raw)
|
|
}
|
|
return
|
|
}
|
|
said := strings.Join(res.NotPublished, "\n")
|
|
if !strings.Contains(said, grants.ShadowDifference) || !strings.Contains(said, row.names) {
|
|
t.Errorf("the loss is not named: %s", said)
|
|
}
|
|
ms, _ := partOn(b).MembershipOf("a", "mod-x")
|
|
if !strings.Contains(fmt.Sprint(ms), row.names) {
|
|
t.Errorf("grants_membership does not name what would be lost: %v", ms)
|
|
}
|
|
})
|
|
}
|
|
t.Run("a refused publish: no exit, said, published at the next event", func(t *testing.T) {
|
|
b := fixture()
|
|
b.loaded = shadowOf(t, b)
|
|
b.refusePub = true
|
|
p := partOn(b)
|
|
refused := p.Compose()
|
|
if len(refused.Published) != 0 || len(refused.NotPublished) != 3 || !strings.Contains(refused.NotPublished[0], "Permissions Violation") {
|
|
t.Errorf("a refused publish = %+v: said, and nothing published", refused)
|
|
}
|
|
b.refusePub = false
|
|
if next := p.Compose(); !slices.Equal(next.Published, []string{"a.mod-x", "a.nats", "b.mod-y"}) {
|
|
t.Errorf("the next event published %v", next.Published)
|
|
}
|
|
})
|
|
t.Run("a tool answers while a pass waits on the bus", func(t *testing.T) {
|
|
b := fixture()
|
|
b.loaded = shadowOf(t, b)
|
|
b.hold = make(chan struct{})
|
|
p := partOn(b)
|
|
done := make(chan struct{})
|
|
go func() { p.Compose(); close(done) }()
|
|
answered := make(chan struct{})
|
|
go func() {
|
|
_, _ = p.UserList()
|
|
_, _ = p.MembershipOf("a", "mod-x")
|
|
_, _ = p.Grants("")
|
|
_ = p.Last()
|
|
close(answered)
|
|
}()
|
|
select {
|
|
case <-answered:
|
|
case <-time.After(10 * time.Second):
|
|
t.Error("a tool waited on the pass's round trip")
|
|
}
|
|
close(b.hold)
|
|
<-done
|
|
})
|
|
})
|
|
|
|
t.Run("assertions: what the bus lacks is created once, what it holds is never touched", func(t *testing.T) {
|
|
b := fixture()
|
|
b.loaded = shadowOf(t, b)
|
|
// The bus holds mod-x's bucket with another history, and neither its log nor its stream.
|
|
b.streams = []map[string]any{{"name": "KV_mod-x_kept", "subjects": []string{"$KV.mod-x_kept.>"}, "retention": "limits", "max_per_subject": 2}}
|
|
p := partOn(b)
|
|
p.Create = b
|
|
assertion := func() grants.Assertion {
|
|
raw, _, _, _ := b.ownGet(grants.StateAssertions, "mod-x")
|
|
var a grants.Assertion
|
|
_ = json.Unmarshal(raw, &a)
|
|
return a
|
|
}
|
|
first := p.Compose()
|
|
if !slices.Equal(b.creates, []string{"LOG_mod-x_history", "THINGS"}) || len(first.Asserted) != 1 ||
|
|
!strings.HasPrefix(first.Asserted[0], "mod-x at ") || !slices.Contains(b.events, grants.EventAsserted) {
|
|
t.Fatalf("the first pass created %v, said %v, events %v", b.creates, first.Asserted, b.events)
|
|
}
|
|
a := assertion()
|
|
if !a.Asserted || a.Says != grants.CreatesMissing {
|
|
t.Errorf("the assertion kept = %+v", a)
|
|
}
|
|
for _, it := range a.Items {
|
|
switch it.Stream {
|
|
case "KV_mod-x_kept":
|
|
if it.Created != "" || !slices.Contains(it.Differences, "history 2 held, 3 composed") {
|
|
t.Errorf("a bucket the bus holds was touched, or its difference not said: %+v", it)
|
|
}
|
|
default:
|
|
if it.Created != grants.ItemCreated {
|
|
t.Errorf("%s: %q, want created", it.Stream, it.Created)
|
|
}
|
|
}
|
|
}
|
|
if again := p.Compose(); len(again.Asserted) != 0 || len(b.creates) != 2 {
|
|
t.Errorf("a pass with everything held created again: %v, %v", again.Asserted, b.creates)
|
|
}
|
|
// THINGS deleted since: not made again at the same manifest, said; a hand assert (the tool, asking the process)
|
|
// makes it again.
|
|
b.mu.Lock()
|
|
b.streams = slices.DeleteFunc(b.streams, func(s map[string]any) bool { return s["name"] == "THINGS" })
|
|
b.mu.Unlock()
|
|
gone := p.Compose()
|
|
if len(b.creates) != 2 || !strings.Contains(strings.Join(gone.NotAsserted, " "), "THINGS: created at this manifest and gone since") {
|
|
t.Errorf("a stream gone since was made again by itself, or not said: %v, %v", b.creates, gone.NotAsserted)
|
|
}
|
|
tool := partOn(b)
|
|
if got, err := tool.AssertOne("mod-x"); err != nil || fmt.Sprint(got) == "" || len(b.creates) != 2 {
|
|
t.Fatalf("the tool = %v, %v, creates %v: it asks, and creates nothing itself", got, err, b.creates)
|
|
}
|
|
if hand := p.Compose(); len(hand.Asserted) != 1 || b.creates[len(b.creates)-1] != "THINGS" {
|
|
t.Errorf("the hand assert did not create THINGS again: %v, %v", hand.Asserted, b.creates)
|
|
}
|
|
if after := p.Compose(); len(after.Asserted) != 0 {
|
|
t.Errorf("the hand ask was acted on twice: %v", after.Asserted)
|
|
}
|
|
for _, c := range b.creates {
|
|
if c == "KV_mod-x_kept" {
|
|
t.Error("the bucket the bus holds was asked to be created")
|
|
}
|
|
}
|
|
})
|
|
|
|
t.Run("assertions: a refused create is said, never stops it, and is tried at the next event", func(t *testing.T) {
|
|
b := fixture()
|
|
b.loaded = shadowOf(t, b)
|
|
b.refuseCreate = true
|
|
p := partOn(b)
|
|
p.Create = b
|
|
refused := p.Compose()
|
|
if len(refused.Asserted) != 0 || !strings.Contains(strings.Join(refused.NotAsserted, " "), "Permissions Violation") ||
|
|
slices.Contains(b.events, grants.EventAsserted) {
|
|
t.Fatalf("a refused create = %+v", refused)
|
|
}
|
|
b.refuseCreate = false
|
|
if next := p.Compose(); len(next.Asserted) != 1 || !strings.Contains(next.Asserted[0], "KV_mod-x_kept") {
|
|
t.Errorf("the next event created %v", next.Asserted)
|
|
}
|
|
// What the bus holds unread: nothing is created.
|
|
c := fixture()
|
|
c.loaded = shadowOf(t, c)
|
|
q := partOn(c)
|
|
q.Create = c
|
|
q.Call = refusingStreams{c}
|
|
if r := q.Compose(); len(c.creates) != 0 || len(r.Asserted) != 0 {
|
|
t.Errorf("created %v with what the bus holds unread", c.creates)
|
|
}
|
|
})
|
|
|
|
t.Run("memberships held back: every one is said and listed", func(t *testing.T) {
|
|
b := fixture()
|
|
b.loaded = shadowOf(t, b)
|
|
recs, _ := grants.Records(grants.ReadInputs(b, &grants.Manifests{}))
|
|
b.stream, b.seqs = map[string][]byte{}, map[string]uint64{}
|
|
for i, key := range []string{"a.mod-x", "b.mod-y"} {
|
|
raw, _ := json.Marshal(grants.Memberships(recs)[key].Membership)
|
|
var m map[string]any
|
|
_ = json.Unmarshal(raw, &m)
|
|
m["serves"] = append(m["serves"].([]any), map[string]any{"subject": "mesh.mod.lost.tool.gone"})
|
|
raw, _ = json.Marshal(m)
|
|
subject := "mesh.assignment." + key
|
|
b.stream[subject], b.seqs[subject] = raw, uint64(i+1)
|
|
}
|
|
var logged []string
|
|
var lmu sync.Mutex
|
|
w := &fakeWatcher{opened: make(chan string, 16)}
|
|
f := &grants.Follower{Part: partOn(b), Watch: w, Logf: func(format string, args ...any) {
|
|
lmu.Lock()
|
|
logged = append(logged, fmt.Sprintf(format, args...))
|
|
lmu.Unlock()
|
|
}}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { f.Run(ctx); close(done) }()
|
|
waitFor(t, func() bool {
|
|
lmu.Lock()
|
|
defer lmu.Unlock()
|
|
all := strings.Join(logged, "\n")
|
|
return strings.Contains(all, "held a.mod-x: ") && strings.Contains(all, "held b.mod-y: ")
|
|
})
|
|
cancel()
|
|
<-done
|
|
held, err := partOn(b).MembershipOf("", "")
|
|
h, _ := held.(map[string]any)
|
|
if err != nil || h["count"] != 2 || !strings.Contains(fmt.Sprint(h["held"]), "tool.gone") {
|
|
t.Errorf("grants_membership with neither node nor module = %v, %v", held, err)
|
|
}
|
|
if _, err := partOn(b).MembershipOf("a", ""); err == nil {
|
|
t.Error("a node without a module was answered")
|
|
}
|
|
})
|
|
|
|
t.Run("a refused read or write never stops it", func(t *testing.T) {
|
|
b := fixture()
|
|
b.loaded = shadowOf(t, b)
|
|
b.refuseTo[grants.StateMemberships] = true
|
|
res := partOn(b).Compose()
|
|
if !res.Written || len(res.Errors) == 0 || !strings.Contains(strings.Join(res.Errors, " "), "memberships") {
|
|
t.Errorf("a refused write of memberships = %+v: the user list is written all the same, the refusal said", res)
|
|
}
|
|
w := &fakeWatcher{refuse: map[string]int{grants.ReadModules: 1}, opened: make(chan string, 16)}
|
|
var logged []string
|
|
var lmu sync.Mutex
|
|
f := &grants.Follower{Part: partOn(b), Watch: w, Logf: func(format string, args ...any) {
|
|
lmu.Lock()
|
|
logged = append(logged, fmt.Sprintf(format, args...))
|
|
lmu.Unlock()
|
|
}}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { f.Run(ctx); close(done) }()
|
|
waitOpened(t, w, grants.ReadAssignments)
|
|
if r := f.Refused(); r[grants.ReadModules] == "" {
|
|
waitFor(t, func() bool { return f.Refused()[grants.ReadModules] != "" })
|
|
}
|
|
w.change(grants.ReadAssignments, "a.mod-x")
|
|
waitOpened(t, w, grants.ReadModules)
|
|
waitFor(t, func() bool { return len(f.Refused()) == 0 })
|
|
select {
|
|
case <-done:
|
|
t.Fatal("the follower stopped")
|
|
default:
|
|
}
|
|
cancel()
|
|
<-done
|
|
lmu.Lock()
|
|
defer lmu.Unlock()
|
|
if !strings.Contains(strings.Join(logged, "\n"), "cannot be watched yet") {
|
|
t.Errorf("the refusal was not said: %v", logged)
|
|
}
|
|
})
|
|
}
|
|
|
|
type fakeAsker struct {
|
|
asks []asks.Ask
|
|
wheres map[string]*asks.Warrant // the router's warrants, by ask
|
|
fail bool
|
|
cancels []string
|
|
// takeErr makes Take fail as the asker does when the part's own record cannot be reached: no handling, an error.
|
|
takeErr error
|
|
}
|
|
|
|
func (f *fakeAsker) Cancel(id string) error { f.cancels = append(f.cancels, id); return nil }
|
|
|
|
// fakeObjects is the part's object bucket of sealed lists.
|
|
type fakeObjects struct{ objects map[string][]byte }
|
|
|
|
func (o *fakeObjects) PutObject(name string, data []byte) error {
|
|
if o.objects == nil {
|
|
o.objects = map[string][]byte{}
|
|
}
|
|
o.objects[name] = data
|
|
return nil
|
|
}
|
|
|
|
func (o *fakeObjects) DeleteObject(name string) error { delete(o.objects, name); return nil }
|
|
|
|
func (f *fakeAsker) Ask(a asks.Ask) error {
|
|
if f.fail {
|
|
return errors.New("the bus refused the publish")
|
|
}
|
|
f.asks = append(f.asks, a)
|
|
return nil
|
|
}
|
|
|
|
func (f *fakeAsker) Where(id string) (string, *asks.Warrant, error) { return "open", f.wheres[id], nil }
|
|
|
|
// Take hands the warrant to act as the SDK's asker does for one that chose, and records any other.
|
|
func (f *fakeAsker) Take(w asks.Warrant, act func(asks.Option, asks.Warrant) error) (asker.Handled, error) {
|
|
if f.takeErr != nil {
|
|
return "", f.takeErr
|
|
}
|
|
if w.Outcome != asks.OutcomeChosen {
|
|
return asker.Recorded, nil
|
|
}
|
|
return asker.Acted, act(asks.Option{ID: w.Option}, w)
|
|
}
|
|
|
|
func nonNil(s []string) []string {
|
|
if s == nil {
|
|
return []string{}
|
|
}
|
|
return s
|
|
}
|
|
|
|
// fakeWatcher refuses a state's first watches as told, and hands changes to the watches it opened.
|
|
// refusingStreams is the bus with nats_streams refused: what the bus holds cannot be read.
|
|
type refusingStreams struct{ *fakeBus }
|
|
|
|
func (r refusingStreams) Call(key string, body any) (json.RawMessage, error) {
|
|
if key == grants.CallStreams {
|
|
return nil, errors.New("refused")
|
|
}
|
|
return r.fakeBus.Call(key, body)
|
|
}
|
|
|
|
type fakeWatcher struct {
|
|
mu sync.Mutex
|
|
refuse map[string]int
|
|
changes map[string]func(string, []byte)
|
|
opened chan string
|
|
}
|
|
|
|
func (w *fakeWatcher) Watch(state string, changed func(string, []byte), _ func(error)) error {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
if w.refuse[state] > 0 {
|
|
w.refuse[state]--
|
|
return errors.New("Permissions Violation for Publish to $JS.API.STREAM.INFO.KV_" + state)
|
|
}
|
|
if w.changes == nil {
|
|
w.changes = map[string]func(string, []byte){}
|
|
}
|
|
w.changes[state] = changed
|
|
w.opened <- state
|
|
return nil
|
|
}
|
|
|
|
func (w *fakeWatcher) change(state, key string) {
|
|
w.mu.Lock()
|
|
h := w.changes[state]
|
|
w.mu.Unlock()
|
|
h(key, []byte(`{}`))
|
|
}
|
|
|
|
func waitOpened(t *testing.T, w *fakeWatcher, state string) {
|
|
t.Helper()
|
|
deadline := time.After(10 * time.Second)
|
|
for {
|
|
select {
|
|
case s := <-w.opened:
|
|
if s == state {
|
|
return
|
|
}
|
|
case <-deadline:
|
|
t.Fatalf("%s was never watched", state)
|
|
}
|
|
}
|
|
}
|
|
|
|
func waitFor(t *testing.T, ok func() bool) {
|
|
t.Helper()
|
|
for end := time.Now().Add(10 * time.Second); time.Now().Before(end); time.Sleep(10 * time.Millisecond) {
|
|
if ok() {
|
|
return
|
|
}
|
|
}
|
|
t.Fatal("never came")
|
|
}
|