Add mesh-grants in shadow: compose the bus's user list, memberships and assertions from the bus and compare, writing only its own state (ADR 0547 clauses 5-8, issue 1945)
The grants part must compose with the controller's own composer before it can take over the list; in shadow it proves the two agree, names every input the bus does not carry yet, and takes no effect. It claims no seat (mesh-sdk's seat list does not name mesh-grants for the running composer), serves module tools, and keeps running when a read or write is refused.
This commit is contained in:
@@ -0,0 +1,220 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// requestWithin bounds one request to the bus: a read the bus refuses is answered by no reply, so the bound is how a
|
||||
// refusal is told from a slow answer. It is no clock of the part's work: nothing is ever started by it.
|
||||
const requestWithin = 10 * time.Second
|
||||
|
||||
// bucketOf is the bucket a read names: `<module>.<state>` is `<module>_<state>`.
|
||||
func bucketOf(state string) string {
|
||||
at := strings.LastIndex(state, ".")
|
||||
if at <= 0 {
|
||||
return state
|
||||
}
|
||||
return state[:at] + "_" + state[at+1:]
|
||||
}
|
||||
|
||||
// busReader reads other parts' state on the bus, and keeps every value of a state it watches, so a composition reads
|
||||
// those from memory. A bucket it cannot open (refused, or not there yet) is an error of the call, never the end.
|
||||
type busReader struct {
|
||||
ctx context.Context
|
||||
js jetstream.JetStream
|
||||
|
||||
mu sync.Mutex
|
||||
kvs map[string]jetstream.KeyValue
|
||||
world map[string]map[string][]byte
|
||||
settled map[string]bool
|
||||
}
|
||||
|
||||
func newBusReader(ctx context.Context, js jetstream.JetStream) *busReader {
|
||||
return &busReader{ctx: ctx, js: js, kvs: map[string]jetstream.KeyValue{}, world: map[string]map[string][]byte{},
|
||||
settled: map[string]bool{}}
|
||||
}
|
||||
|
||||
func (b *busReader) kv(state string) (jetstream.KeyValue, error) {
|
||||
b.mu.Lock()
|
||||
kv, ok := b.kvs[state]
|
||||
b.mu.Unlock()
|
||||
if ok {
|
||||
return kv, nil
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(b.ctx, requestWithin)
|
||||
defer cancel()
|
||||
kv, err := b.js.KeyValue(ctx, bucketOf(state))
|
||||
if err != nil {
|
||||
if errors.Is(err, context.DeadlineExceeded) {
|
||||
err = fmt.Errorf("no answer in %s: a read the bus refuses is answered by none (%w)", requestWithin, err)
|
||||
}
|
||||
return nil, fmt.Errorf("the bucket %s cannot be opened: %w", bucketOf(state), err)
|
||||
}
|
||||
b.mu.Lock()
|
||||
b.kvs[state] = kv
|
||||
b.mu.Unlock()
|
||||
return kv, nil
|
||||
}
|
||||
|
||||
func (b *busReader) Keys(state string) ([]string, error) {
|
||||
b.mu.Lock()
|
||||
if b.settled[state] {
|
||||
out := make([]string, 0, len(b.world[state]))
|
||||
for k := range b.world[state] {
|
||||
out = append(out, k)
|
||||
}
|
||||
b.mu.Unlock()
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
}
|
||||
b.mu.Unlock()
|
||||
kv, err := b.kv(state)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(b.ctx, requestWithin)
|
||||
defer cancel()
|
||||
keys, err := kv.Keys(ctx)
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return []string{}, nil
|
||||
}
|
||||
return keys, err
|
||||
}
|
||||
|
||||
func (b *busReader) Get(state, key string) ([]byte, bool, error) {
|
||||
b.mu.Lock()
|
||||
if b.settled[state] {
|
||||
v, ok := b.world[state][key]
|
||||
b.mu.Unlock()
|
||||
return v, ok, nil
|
||||
}
|
||||
b.mu.Unlock()
|
||||
kv, err := b.kv(state)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(b.ctx, requestWithin)
|
||||
defer cancel()
|
||||
e, err := kv.Get(ctx, key)
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return nil, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
return e.Value(), true, nil
|
||||
}
|
||||
|
||||
// Watch keeps every value of a state from now on and hands each change after what stands to changed; it returns once
|
||||
// what stands is held. The watch ending is handed to ended, and the state is read from the bus again until watched.
|
||||
func (b *busReader) Watch(state string, changed func(string, []byte), ended func(error)) error {
|
||||
kv, err := b.kv(state)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
w, err := kv.WatchAll(b.ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the bucket %s cannot be watched: %w", bucketOf(state), err)
|
||||
}
|
||||
put := func(e jetstream.KeyValueEntry) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.world[state] == nil {
|
||||
b.world[state] = map[string][]byte{}
|
||||
}
|
||||
if e.Operation() == jetstream.KeyValuePut {
|
||||
b.world[state][e.Key()] = e.Value()
|
||||
} else {
|
||||
delete(b.world[state], e.Key())
|
||||
}
|
||||
}
|
||||
lost := func(err error) {
|
||||
w.Stop()
|
||||
b.mu.Lock()
|
||||
b.settled[state] = false
|
||||
delete(b.world, state)
|
||||
delete(b.kvs, state)
|
||||
b.mu.Unlock()
|
||||
ended(err)
|
||||
}
|
||||
settling, stop := context.WithTimeout(b.ctx, requestWithin)
|
||||
defer stop()
|
||||
settle:
|
||||
for {
|
||||
select {
|
||||
case <-settling.Done():
|
||||
w.Stop()
|
||||
if b.ctx.Err() != nil {
|
||||
return b.ctx.Err()
|
||||
}
|
||||
return fmt.Errorf("the watch of %s handed nothing in %s: a watch the bus refuses is answered by none", bucketOf(state), requestWithin)
|
||||
case e, ok := <-w.Updates():
|
||||
if !ok {
|
||||
w.Stop()
|
||||
return fmt.Errorf("the watch of %s ended before it held what stands", bucketOf(state))
|
||||
}
|
||||
if e == nil {
|
||||
break settle
|
||||
}
|
||||
put(e)
|
||||
}
|
||||
}
|
||||
b.mu.Lock()
|
||||
b.settled[state] = true
|
||||
b.mu.Unlock()
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-b.ctx.Done():
|
||||
w.Stop()
|
||||
return
|
||||
case e, ok := <-w.Updates():
|
||||
if !ok {
|
||||
lost(errors.New("the server ended it"))
|
||||
return
|
||||
}
|
||||
if e != nil {
|
||||
put(e)
|
||||
var value []byte
|
||||
if e.Operation() == jetstream.KeyValuePut {
|
||||
value = e.Value()
|
||||
}
|
||||
changed(e.Key(), value)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// ownStore is the part's own state on its process connection: a state it did not announce, or one the bus refuses,
|
||||
// fails the call, named.
|
||||
type ownStore struct{ c *process.Conn }
|
||||
|
||||
func (s ownStore) Get(state, key string) ([]byte, uint64, bool, error) {
|
||||
e, err := s.c.State(state).Get(key)
|
||||
if err != nil || e == nil {
|
||||
return nil, 0, false, err
|
||||
}
|
||||
return e.Value, e.Revision, true, nil
|
||||
}
|
||||
|
||||
func (s ownStore) Put(state, key string, value []byte) (uint64, error) {
|
||||
return s.c.State(state).Put(key, json.RawMessage(value))
|
||||
}
|
||||
|
||||
// emitting is the process connection as the part's emitter: its events, with no options.
|
||||
type emitting struct{ c *process.Conn }
|
||||
|
||||
func (e emitting) Emit(event string, body any) error { return e.c.Emit(event, body) }
|
||||
@@ -0,0 +1,111 @@
|
||||
// Command mesh-grants is the grants part (novox/hq ADR 0547 clauses 5 to 8, issue 1945), in shadow for the first
|
||||
// milestone: a process on its own bus credential that watches placement, the indexer's manifests, mesh-config's
|
||||
// resolved values and the node agents' haves, and on every change composes the bus's user list, each assignment's
|
||||
// membership and each manifest's buckets, logs and streams with the bus composer the controller uses (mesh-sdk
|
||||
// buscompose), compares them with what the bus loaded and holds (through the bus module's own read tools), and writes
|
||||
// only its own state. It publishes no membership, asserts nothing, writes no grant and no user list the bus node reads.
|
||||
// It holds no timer, and never exits for a refused read or write: a refusal is said and tried again at the next event.
|
||||
//
|
||||
// **No seat in this milestone.** ADR 0547 clause 5 gives the part the seat `mesh-grants`; a `mesh-*` seat must be in
|
||||
// mesh-sdk's own seat list, which the running controller's composer does not know, so the part serves module tools
|
||||
// (grants_*) and claims no seat until take-over.
|
||||
//
|
||||
// Started by the tool runner, with no credential of its own, it serves those tools (a process never serves, ADR 0436).
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"syscall"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-grants/internal/grants"
|
||||
)
|
||||
|
||||
func main() {
|
||||
if os.Getenv("MESH_BROKER_FILE") == "" {
|
||||
if err := serveTools(); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "mesh-grants:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
return
|
||||
}
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
if err := run(ctx, os.Getenv, os.Stderr); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "mesh-grants:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
// keptStates is what the part keeps, each readable (ADR 0495 §2-§3): the view and a session read them.
|
||||
func keptStates() []announce.State {
|
||||
var out []announce.State
|
||||
for _, s := range grants.States {
|
||||
out = append(out, announce.State{Name: s, Readable: true})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// run is the process: it returns when ctx ends, or when it has no credential or cannot dial the bus at all. A bucket or
|
||||
// read the bus refuses never ends it.
|
||||
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
|
||||
logger := log.New(logOut, "mesh-grants: ", log.LstdFlags)
|
||||
path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
|
||||
if path == "" {
|
||||
return errors.New("MESH_BROKER_FILE is not set, so there is no credential to read the bus with")
|
||||
}
|
||||
account, err := process.ReadAccount(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
f := &grants.Follower{Logf: logger.Printf}
|
||||
conn, err := process.Connect(account, nats.Name(account.Node+"."+account.Module), nats.RetryOnFailedConnect(true),
|
||||
nats.ReconnectHandler(func(*nats.Conn) { f.Wake() }))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close()
|
||||
if err := conn.Keeps(keptStates()...); err != nil {
|
||||
logger.Printf("what the part keeps could not be said on the bus, so its readers are not granted until it is: %v", err)
|
||||
}
|
||||
// The reads of other parts' state, on a connection of the same account: a read the bus refuses is said and composed
|
||||
// past (busReader), never the end of the process.
|
||||
opts := []nats.Option{
|
||||
nats.Name(account.Node + "." + account.Module + ".reads"),
|
||||
nats.UserInfo(account.User, account.Password),
|
||||
nats.CustomInboxPrefix(account.InboxPrefix()),
|
||||
nats.MaxReconnects(-1),
|
||||
nats.RetryOnFailedConnect(true),
|
||||
nats.ReconnectHandler(func(*nats.Conn) { f.Wake() }),
|
||||
}
|
||||
if strings.TrimSpace(account.Fingerprint) != "" {
|
||||
opts = append(opts, nats.Secure(process.PinnedToFingerprint(account.Fingerprint)))
|
||||
}
|
||||
nc, err := nats.Connect(account.URL, opts...)
|
||||
if err != nil {
|
||||
return fmt.Errorf("connecting to the bus: %w", err)
|
||||
}
|
||||
defer nc.Close()
|
||||
js, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
bus := newBusReader(ctx, js)
|
||||
f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn}}
|
||||
f.Watch = bus
|
||||
logger.Printf("started on the account %s, in shadow: composing and comparing, writing only its own state", account.User)
|
||||
f.Run(ctx)
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"mesh-grants/internal/grants"
|
||||
)
|
||||
|
||||
// The part's tools, served through the tool runner from its bundle (ADR 0436: a process never serves). They read and
|
||||
// write through the runtime as the module: its own states, the reads its manifest names, the bus module's read tools.
|
||||
|
||||
// runtimeReader reads another part's state through the runtime, `<module>.<state>`.
|
||||
type runtimeReader struct{}
|
||||
|
||||
func (runtimeReader) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
|
||||
|
||||
func (runtimeReader) Get(state, key string) ([]byte, bool, error) {
|
||||
e, err := stdio.State(state).Get(key)
|
||||
if err != nil || e == nil {
|
||||
return nil, false, err
|
||||
}
|
||||
return e.Value, true, nil
|
||||
}
|
||||
|
||||
// runtimeStore is the part's own state through the runtime.
|
||||
type runtimeStore struct{}
|
||||
|
||||
func (runtimeStore) Get(state, key string) ([]byte, uint64, bool, error) {
|
||||
e, err := stdio.State(state).Get(key)
|
||||
if err != nil || e == nil {
|
||||
return nil, 0, false, err
|
||||
}
|
||||
return e.Value, e.Revision, true, nil
|
||||
}
|
||||
|
||||
func (runtimeStore) Put(state, key string, value []byte) (uint64, error) {
|
||||
return stdio.State(state).Put(key, json.RawMessage(value))
|
||||
}
|
||||
|
||||
// runtimeBus calls tools and says events through the runtime.
|
||||
type runtimeBus struct{}
|
||||
|
||||
func (runtimeBus) Call(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }
|
||||
func (runtimeBus) Emit(event string, body any) error { return stdio.Emit(event, body) }
|
||||
|
||||
func schema(properties map[string]string, required ...string) map[string]any {
|
||||
props := map[string]any{}
|
||||
for name, description := range properties {
|
||||
props[name] = map[string]any{"type": "string", "description": description}
|
||||
}
|
||||
out := map[string]any{"type": "object", "properties": props}
|
||||
if len(required) > 0 {
|
||||
out["required"] = required
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// toolsOf are the part's tools over a part (grants.Tools names them).
|
||||
func toolsOf(p *grants.Part) []stdio.Tool {
|
||||
return []stdio.Tool{
|
||||
{Name: grants.ToolGrants, Description: "What the grants part holds of grants (novox/hq ADR 0547 clause 6): in this milestone it " +
|
||||
"writes none (grants.<module>.<fp> from the third milestone, clause 5), and shows beside it the grants the controller " +
|
||||
"wrote (mesh-controller_grants), all or one module's with the accounts each names. Read-only.",
|
||||
Input: schema(map[string]string{"module": "one module"}),
|
||||
Run: func(a map[string]any) (any, error) { return p.Grants(grants.Answer(a, "module")) }},
|
||||
{Name: grants.ToolMembership, Description: "One module's membership on one node (ADR 0547 clause 6): as composed now from the " +
|
||||
"bus's records with the bus composer, as last kept in the part's state, and as published (not readable from here: the bus " +
|
||||
"grants each module its own alone), with what the composition cannot read. Read-only.",
|
||||
Input: schema(map[string]string{"node": "the machine", "module": "the module"}, "node", "module"),
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
return p.MembershipOf(grants.Answer(a, "node"), grants.Answer(a, "module"))
|
||||
}},
|
||||
{Name: grants.ToolUserList, Description: "The bus's user list in shadow (ADR 0547 clauses 6 and 8): the list the grants part " +
|
||||
"composed (digest, revision, users), what the bus node says it loaded (source, revision, digest, users, error), and how " +
|
||||
"they differ: users added, users dropped (and of those, gone from placement), each user's grant differences, the " +
|
||||
"assignments left out, and every input the bus does not carry, named. Read-only.",
|
||||
Run: func(map[string]any) (any, error) { return p.UserList() }},
|
||||
{Name: grants.ToolComposeNow, Description: "Compose again now (ADR 0547 clause 6), as the part does on every change it hears: the " +
|
||||
"user list, the memberships and the assertions, compared with what the bus loaded and holds; writes its own state only " +
|
||||
"where a value changed (one revision when an input changed, none when not) and says user-list-changed and " +
|
||||
"membership-changed for each. In shadow it publishes, asserts and grants nothing.",
|
||||
Run: func(map[string]any) (any, error) { return p.Compose(), nil }},
|
||||
{Name: grants.ToolAssert, Description: "One module's buckets, logs and streams (ADR 0547 clause 6), composed from its manifest at " +
|
||||
"the indexer's trunk with their limits, beside what the bus holds (through the bus module's nats_streams). In shadow it " +
|
||||
"asserts nothing, and says so.",
|
||||
Input: schema(map[string]string{"module": "the module"}, "module"),
|
||||
Run: func(a map[string]any) (any, error) { return p.AssertOne(grants.Answer(a, "module")) }},
|
||||
{Name: grants.ToolTakeOver, Description: "The operator's switch to the grants part's list (ADR 0547 clause 8), refused without the " +
|
||||
"operator's credential (a person's account), refused while any user of the shadow list differs from the loaded list (but " +
|
||||
"those placement no longer names), naming them; in this milestone always refused, since the enrolment users and the " +
|
||||
"persons cannot be composed. Every ask is recorded in the part's state takeover.",
|
||||
Input: schema(map[string]string{"why": "why, kept with the act"}, "why"),
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
return nil, p.TakeOver(stdio.CallerOf(a), grants.Answer(a, "why"))
|
||||
}},
|
||||
{Name: grants.ToolHandBack, Description: "The operator's switch back to the controller's list (ADR 0547 clause 8), refused without " +
|
||||
"the operator's credential; in shadow there is nothing to hand back, and it says so. Every ask is recorded.",
|
||||
Input: schema(map[string]string{"why": "why, kept with the act"}, "why"),
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
return nil, p.HandBack(stdio.CallerOf(a), grants.Answer(a, "why"))
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
// serveTools serves the tools to the runtime until it stops.
|
||||
func serveTools() error {
|
||||
p := &grants.Part{Read: runtimeReader{}, Call: runtimeBus{}, Store: runtimeStore{}, Emit: runtimeBus{}}
|
||||
return stdio.Serve("", toolsOf(p))
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
module mesh-grants
|
||||
|
||||
go 1.26.0
|
||||
|
||||
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025205458-b5f8604d037a
|
||||
|
||||
require (
|
||||
github.com/klauspost/compress v1.20.0 // indirect
|
||||
github.com/nats-io/nats.go v1.54.0 // indirect
|
||||
github.com/nats-io/nkeys v0.4.16 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
golang.org/x/crypto v0.57.0 // indirect
|
||||
golang.org/x/sys v0.48.0 // indirect
|
||||
)
|
||||
@@ -0,0 +1,26 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025205458-b5f8604d037a h1:yqCMZJSVRZ7VoBiZ9iIcYhQKn0BDhOJIRwZ430auLA0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025205458-b5f8604d037a/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=
|
||||
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
|
||||
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
|
||||
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
|
||||
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
|
||||
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
|
||||
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
|
||||
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
|
||||
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
|
||||
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
|
||||
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
|
||||
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
|
||||
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
|
||||
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
@@ -0,0 +1,587 @@
|
||||
package manifest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"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
|
||||
}
|
||||
|
||||
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()
|
||||
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})
|
||||
}
|
||||
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, 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},
|
||||
{"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 publish, no create", func(t *testing.T) {
|
||||
forbidden := []string{"time.Tick", "time.NewTicker", "time.Sleep", "time.After", "time.NewTimer", "time.AfterFunc",
|
||||
".Publish(", "PublishMsg", "CreateStream", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue",
|
||||
"CreateObjectStore", "mesh.assignment", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("}
|
||||
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)
|
||||
for _, f := range forbidden {
|
||||
if strings.Contains(string(src), f) {
|
||||
t.Errorf("%s holds %s: in shadow the part holds no timer and publishes or creates 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"},
|
||||
{"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)
|
||||
}
|
||||
})
|
||||
}
|
||||
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("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("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)
|
||||
|
||||
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)
|
||||
}
|
||||
if again := p.Compose(); again.Written || again.Revision != 1 || again.MembershipsWritten != 0 || again.AssertionsWritten != 0 {
|
||||
t.Errorf("a composition with no input changed wrote: %+v", again)
|
||||
}
|
||||
// mod-x placed on b too: b.mod-x is added, and a.mod-x's membership moves with b's, since mod-x now runs on two
|
||||
// machines and answers only on its machines' subjects.
|
||||
b.assign("b", "mod-x")
|
||||
moved := p.Compose()
|
||||
if !moved.Written || moved.Revision != 2 || moved.MembershipsWritten != 2 || 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")
|
||||
as, _ := a.(grants.Assertion)
|
||||
if err != nil || as.Asserted || as.Says != grants.ShadowAsserts || len(as.Items) != 3 {
|
||||
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); mo["published"] != grants.PublishedUnread || mo["composed"] == nil || mo["kept"] == 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"})
|
||||
for _, row := range []struct {
|
||||
name, act, caller string
|
||||
says, never []string
|
||||
}{
|
||||
{"take-over without the operator's credential", "take-over", "shanks.node-tools", []string{"operator's alone"}, nil},
|
||||
{"hand-back without the operator's credential", "hand-back", "unproven", []string{"operator's alone"}, nil},
|
||||
{"take-over while a user differs names it", "take-over", "person.operator", []string{"users differ", "b.mod-x", grants.NotComposable}, []string{"b.gone"}},
|
||||
{"hand-back in shadow", "hand-back", "person.operator", []string{"nothing to hand back"}, nil},
|
||||
} {
|
||||
t.Run(row.name, func(t *testing.T) {
|
||||
var err error
|
||||
if row.act == "take-over" {
|
||||
err = p.TakeOver(row.caller, "a row")
|
||||
} else {
|
||||
err = p.HandBack(row.caller, "a row")
|
||||
}
|
||||
if err == nil {
|
||||
t.Fatalf("%s by %s was not refused", row.act, row.caller)
|
||||
}
|
||||
for _, s := range row.says {
|
||||
if !strings.Contains(err.Error(), s) {
|
||||
t.Errorf("refusal %q does not say %q", err, s)
|
||||
}
|
||||
}
|
||||
for _, s := range row.never {
|
||||
if strings.Contains(err.Error(), s) {
|
||||
t.Errorf("refusal %q names %q", err, s)
|
||||
}
|
||||
}
|
||||
raw, _, _, _ := b.ownGet(grants.StateTakeover, grants.KeySwitch)
|
||||
var s grants.Switch
|
||||
_ = json.Unmarshal(raw, &s)
|
||||
last := s.Acts[len(s.Acts)-1]
|
||||
if s.State != grants.Shadow || last.Act != row.act || last.By != row.caller || last.Refused == "" {
|
||||
t.Errorf("the act is not recorded: %+v", s)
|
||||
}
|
||||
})
|
||||
}
|
||||
t.Run("take-over with only a user placement no longer names is refused for want of records alone", func(t *testing.T) {
|
||||
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"})
|
||||
err := p.TakeOver("person.operator", "lists agree")
|
||||
if err == nil || strings.Contains(err.Error(), "differ") || !strings.Contains(err.Error(), grants.NotComposable) {
|
||||
t.Errorf("take-over = %v, want refused for the enrolment users and persons alone", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("nothing published or asserted", func(t *testing.T) {
|
||||
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("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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
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.
|
||||
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")
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// Item is one bucket, log or stream a manifest declares, as composed, beside what the bus holds of it.
|
||||
type Item struct {
|
||||
Kind string `json:"kind"`
|
||||
Stream string `json:"stream"`
|
||||
// Composed is what an assertion would ask for; Held what the bus holds, absent when it holds no such stream.
|
||||
Composed Limits `json:"composed"`
|
||||
Held *Limits `json:"held,omitempty"`
|
||||
// Differences says how they differ, in words; empty when they agree in what can be read from here.
|
||||
Differences []string `json:"differences,omitempty"`
|
||||
}
|
||||
|
||||
// Limits are the facts of a stream an assertion sets and the bus module's nats_streams reads back. Its size cap is not
|
||||
// read back by that tool, so it is composed and not compared.
|
||||
type Limits struct {
|
||||
Subjects []string `json:"subjects"`
|
||||
Retention string `json:"retention"`
|
||||
MaxAgeHours int `json:"max_age_hours,omitempty"`
|
||||
MaxPerSubject int `json:"max_per_subject,omitempty"`
|
||||
MaxBytes int64 `json:"max_bytes,omitempty"`
|
||||
}
|
||||
|
||||
// Assertion is one module's declared buckets, logs and streams, composed and compared. Asserted is always false in
|
||||
// shadow: nothing is created or changed.
|
||||
type Assertion struct {
|
||||
Module string `json:"module"`
|
||||
Manifest string `json:"manifest"`
|
||||
Items []Item `json:"items"`
|
||||
Asserted bool `json:"asserted"`
|
||||
Says string `json:"says"`
|
||||
}
|
||||
|
||||
// ShadowAsserts is what every assertion says in this milestone.
|
||||
const ShadowAsserts = "in shadow: composed and compared, nothing asserted (ADR 0547 clause 5 asserts from take-over)"
|
||||
|
||||
// Composed is a module's buckets, logs and streams as an assertion would ask for them. Pure.
|
||||
func Composed(m catalogue.Manifest) []Item {
|
||||
var out []Item
|
||||
for _, b := range bucketsOf(m) {
|
||||
per := b.History
|
||||
if per < 1 {
|
||||
per = 1
|
||||
}
|
||||
out = append(out, Item{Kind: "bucket", Stream: "KV_" + b.Bucket(), Composed: Limits{
|
||||
Subjects: []string{"$KV." + b.Bucket() + ".>"}, Retention: string(buscompose.RetentionLimits),
|
||||
MaxAgeHours: b.TTLSeconds / 3600, MaxPerSubject: per}})
|
||||
}
|
||||
for _, l := range logsOf(m) {
|
||||
c := l.Config()
|
||||
out = append(out, Item{Kind: "log", Stream: l.Stream(), Composed: Limits{Subjects: set(c.Subjects),
|
||||
Retention: string(buscompose.RetentionLimits), MaxAgeHours: int(c.MaxAge.Hours()), MaxBytes: c.MaxBytes}})
|
||||
}
|
||||
for _, s := range streamsOf(m) {
|
||||
st := s.Stream()
|
||||
out = append(out, Item{Kind: "stream", Stream: st.Name, Composed: Limits{Subjects: set(st.Subjects),
|
||||
Retention: string(st.Retention), MaxAgeHours: st.MaxAge / 3600, MaxBytes: st.MaxBytes}})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// HeldStreams reads every stream the bus holds through its module's nats_streams, by name.
|
||||
func HeldStreams(c Caller) (map[string]Limits, error) {
|
||||
raw, err := c.Call(CallStreams, map[string]any{})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", CallStreams, err)
|
||||
}
|
||||
var got struct {
|
||||
Detail []struct {
|
||||
Name string `json:"name"`
|
||||
Subjects []string `json:"subjects"`
|
||||
Retention string `json:"retention"`
|
||||
MaxAgeHours int `json:"max_age_hours"`
|
||||
MaxPerSubject int `json:"max_per_subject"`
|
||||
} `json:"detail"`
|
||||
}
|
||||
if err := json.Unmarshal(unwrap(raw), &got); err != nil {
|
||||
return nil, fmt.Errorf("%s answered no streams: %w", CallStreams, err)
|
||||
}
|
||||
out := map[string]Limits{}
|
||||
for _, s := range got.Detail {
|
||||
out[s.Name] = Limits{Subjects: set(s.Subjects), Retention: s.Retention, MaxAgeHours: s.MaxAgeHours, MaxPerSubject: s.MaxPerSubject}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Assert composes a module's items and compares each with what the bus holds; held nil is not read. It asserts nothing.
|
||||
func Assert(module, digest string, m catalogue.Manifest, held map[string]Limits, heldErr error) Assertion {
|
||||
a := Assertion{Module: module, Manifest: digest, Items: Composed(m), Says: ShadowAsserts}
|
||||
for i := range a.Items {
|
||||
it := &a.Items[i]
|
||||
if heldErr != nil {
|
||||
it.Differences = []string{"what the bus holds was not read: " + heldErr.Error()}
|
||||
continue
|
||||
}
|
||||
h, ok := held[it.Stream]
|
||||
if !ok {
|
||||
it.Differences = []string{"the bus holds no stream " + it.Stream}
|
||||
continue
|
||||
}
|
||||
it.Held = &h
|
||||
if !slices.Equal(h.Subjects, it.Composed.Subjects) {
|
||||
it.Differences = append(it.Differences, fmt.Sprintf("subjects %v held, %v composed", h.Subjects, it.Composed.Subjects))
|
||||
}
|
||||
if h.Retention != it.Composed.Retention {
|
||||
it.Differences = append(it.Differences, fmt.Sprintf("retention %s held, %s composed", h.Retention, it.Composed.Retention))
|
||||
}
|
||||
if h.MaxAgeHours != it.Composed.MaxAgeHours {
|
||||
it.Differences = append(it.Differences, fmt.Sprintf("age %d h held, %d h composed", h.MaxAgeHours, it.Composed.MaxAgeHours))
|
||||
}
|
||||
if it.Kind == "bucket" && max(h.MaxPerSubject, 1) != it.Composed.MaxPerSubject {
|
||||
it.Differences = append(it.Differences, fmt.Sprintf("history %d held, %d composed", h.MaxPerSubject, it.Composed.MaxPerSubject))
|
||||
}
|
||||
}
|
||||
sort.SliceStable(a.Items, func(i, j int) bool { return a.Items[i].Stream < a.Items[j].Stream })
|
||||
return a
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// Watcher watches another part's state: changed hears each change after what stands, ended hears the watch end. A watch
|
||||
// refused (a read not granted yet, a bucket not there) is an error, and is tried again at the next wake.
|
||||
type Watcher interface {
|
||||
Watch(state string, changed func(key string, value []byte), ended func(error)) error
|
||||
}
|
||||
|
||||
// Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved
|
||||
// config values, and the node agents' haves (a machine's first have, and the bus node's loaded list).
|
||||
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave}
|
||||
|
||||
// Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot
|
||||
// open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect).
|
||||
type Follower struct {
|
||||
Part *Part
|
||||
Watch Watcher
|
||||
Logf func(string, ...any)
|
||||
|
||||
wake chan struct{}
|
||||
once sync.Once
|
||||
mu sync.Mutex
|
||||
watched map[string]bool
|
||||
refused map[string]string
|
||||
have map[string]string
|
||||
}
|
||||
|
||||
func (f *Follower) init() {
|
||||
f.once.Do(func() {
|
||||
f.wake = make(chan struct{}, 1)
|
||||
f.watched, f.refused, f.have = map[string]bool{}, map[string]string{}, map[string]string{}
|
||||
})
|
||||
}
|
||||
|
||||
// Wake asks for one more composition: after a change, and after a reconnect.
|
||||
func (f *Follower) Wake() {
|
||||
f.init()
|
||||
select {
|
||||
case f.wake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// Refused is every watch refused at the last try, with why.
|
||||
func (f *Follower) Refused() map[string]string {
|
||||
f.init()
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
out := map[string]string{}
|
||||
for k, v := range f.refused {
|
||||
out[k] = v
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (f *Follower) logf(format string, args ...any) {
|
||||
if f.Logf != nil {
|
||||
f.Logf(format, args...)
|
||||
}
|
||||
}
|
||||
|
||||
// heard says whether a change composes again: every change but a have's, and of the haves a machine's first and the
|
||||
// bus module's loaded list moving.
|
||||
func (f *Follower) heard(state, key string, value []byte) bool {
|
||||
if state != ReadHave {
|
||||
return true
|
||||
}
|
||||
node, module, isNode, err := delivery.ParseHaveKey(key)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if isNode {
|
||||
if _, seen := f.have[node]; seen {
|
||||
return false
|
||||
}
|
||||
f.have[node] = ""
|
||||
return true
|
||||
}
|
||||
if module != BusModule {
|
||||
return false
|
||||
}
|
||||
var h struct {
|
||||
UserList json.RawMessage `json:"user-list"`
|
||||
}
|
||||
_ = json.Unmarshal(value, &h)
|
||||
was, seen := f.have[key]
|
||||
f.have[key] = string(h.UserList)
|
||||
return !seen || was != string(h.UserList)
|
||||
}
|
||||
|
||||
// open opens every watch not open yet; one refused is said and left for the next wake.
|
||||
func (f *Follower) open() {
|
||||
for _, state := range Followed {
|
||||
f.mu.Lock()
|
||||
open := f.watched[state]
|
||||
f.mu.Unlock()
|
||||
if open {
|
||||
continue
|
||||
}
|
||||
state := state
|
||||
err := f.Watch.Watch(state, func(key string, value []byte) {
|
||||
if f.heard(state, key, value) {
|
||||
f.Wake()
|
||||
}
|
||||
}, func(err error) {
|
||||
f.mu.Lock()
|
||||
f.watched[state] = false
|
||||
f.mu.Unlock()
|
||||
f.logf("the watch of %s ended (%v): opened again at the next change", state, err)
|
||||
f.Wake()
|
||||
})
|
||||
f.mu.Lock()
|
||||
if err != nil {
|
||||
f.refused[state] = err.Error()
|
||||
f.logf("%s cannot be watched yet (%v): composed without it, and tried again at the next change or reconnect", state, err)
|
||||
} else {
|
||||
f.watched[state] = true
|
||||
delete(f.refused, state)
|
||||
}
|
||||
f.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// Run composes at start and after every change until ctx ends. It returns only when ctx ends.
|
||||
func (f *Follower) Run(ctx context.Context) {
|
||||
f.init()
|
||||
f.Wake()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-f.wake:
|
||||
f.open()
|
||||
res := f.Part.Compose()
|
||||
for _, e := range res.Errors {
|
||||
f.logf("%s", e)
|
||||
}
|
||||
if res.Written {
|
||||
f.logf("the shadow list moved to revision %d: %d users, %d differing from the loaded list",
|
||||
res.Revision, res.UserList.Users, len(res.UserList.Compare.Differing()))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,279 @@
|
||||
// Package grants is the grants part in shadow (novox/hq ADR 0547 clauses 5 to 8, issue 1945): it composes the bus's user
|
||||
// list, each module's membership and each manifest's buckets, logs and streams from what the bus holds, with the bus
|
||||
// composer the controller uses (mesh-sdk buscompose), and compares them with what the bus loaded and holds. It writes only
|
||||
// its own state and takes no effect: it publishes no membership, asserts nothing, writes no grant and no user list the
|
||||
// bus node reads, until the operator's take-over, which this milestone always refuses.
|
||||
package grants
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The states of other parts the grants part reads, as its manifest's reads name them (`<module>.<state>`).
|
||||
const (
|
||||
ReadAssignments = "mesh-controller.assignments"
|
||||
ReadControllerGrant = "mesh-controller.grants"
|
||||
ReadModules = "mesh-delivery-indexer.modules"
|
||||
ReadManifests = "mesh-delivery-indexer.manifests"
|
||||
ReadHave = "node-engine.have"
|
||||
ReadResolved = "mesh-config.resolved"
|
||||
)
|
||||
|
||||
// Reads is every state the part reads, in the manifest's order.
|
||||
var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave}
|
||||
|
||||
// The bus module's tools the part calls, as its manifest's invokes name them: read-only, each.
|
||||
const (
|
||||
CallUsers = "nats.nats_users"
|
||||
CallUserCan = "nats.nats_user_can"
|
||||
CallStreams = "nats.nats_streams"
|
||||
)
|
||||
|
||||
// Invokes is every tool the part calls.
|
||||
var Invokes = []string{CallUsers, CallUserCan, CallStreams}
|
||||
|
||||
// Reader reads another part's state, named `<module>.<state>`. A refusal (a read not granted yet, a bucket not there) is
|
||||
// an error the part names as a gap and composes past: it never stops for one.
|
||||
type Reader interface {
|
||||
Keys(state string) ([]string, error)
|
||||
Get(state, key string) ([]byte, bool, error)
|
||||
}
|
||||
|
||||
// Caller calls another module's tool, `<module>.<tool>`, and answers its result.
|
||||
type Caller interface {
|
||||
Call(key string, body any) (json.RawMessage, error)
|
||||
}
|
||||
|
||||
// Gap is an input the shadow cannot read from the bus, named with why: never guessed.
|
||||
type Gap struct {
|
||||
Input string `json:"input"`
|
||||
Why string `json:"why"`
|
||||
}
|
||||
|
||||
// Standing gaps: what the controller composes from its own store, which no bus value carries in this milestone.
|
||||
var standingGaps = []Gap{
|
||||
{Input: "enrolment tokens", Why: "the live enrolment tokens are in the controller's store alone: no enrolment user is composed"},
|
||||
{Input: "people", Why: "the people and the tools they may invoke are in the controller's store alone: no person user is composed"},
|
||||
{Input: "seat holdings", Why: "who holds each seat held once for the mesh is in the controller's store alone: a mesh seat claimed on two machines is composed as held on both"},
|
||||
{Input: "announced states", Why: "the controller does not announce its states-announced readable, so every read is composed from `reads`, as when announcements cannot be read"},
|
||||
{Input: "credentials", Why: "the shadow holds no hash, so a user it composes that the loaded list lacks may be one left out for want of a credential"},
|
||||
}
|
||||
|
||||
// assignment is a key of the controller's assignments as composing needs it (mesh-controller internal/chain Assignment):
|
||||
// a value that names no time assigned (a module pulled in), has ended, or waits in a state, is no assignment.
|
||||
type assignment struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
Assigned time.Time `json:"assigned,omitzero"`
|
||||
Ended time.Time `json:"ended,omitzero"`
|
||||
State string `json:"state,omitempty"`
|
||||
}
|
||||
|
||||
func (a assignment) stands() bool { return !a.Assigned.IsZero() && a.Ended.IsZero() && a.State == "" }
|
||||
|
||||
// resolved is a key of mesh-config's resolved values, `<node>.<module>.<key>`.
|
||||
type resolved struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
Key string `json:"key"`
|
||||
Value json.RawMessage `json:"value,omitempty"`
|
||||
}
|
||||
|
||||
// Inputs is what the shadow composes from, read from the bus.
|
||||
type Inputs struct {
|
||||
// Nodes is every machine seen: in a node agent's have or a standing assignment.
|
||||
Nodes []string
|
||||
// Assigned is each machine's standing assignments, sorted.
|
||||
Assigned map[string][]string
|
||||
// Manifests is each module's manifest at the indexer's trunk, by module.
|
||||
Manifests map[string]catalogue.Manifest
|
||||
// Digests is each module's manifest digest, by module.
|
||||
Digests map[string]string
|
||||
// Values is each resolved config value, by `<node>/<module>/<key>`.
|
||||
Values map[string]any
|
||||
// BusNode is the machine running the bus's module, and Loaded what its node agent says it loaded.
|
||||
BusNode string
|
||||
Loaded *delivery.UserListWritten
|
||||
Gaps []Gap
|
||||
}
|
||||
|
||||
// BusModule is the bus's module: it holds mesh-broker.
|
||||
const BusModule = "nats"
|
||||
|
||||
// Manifests caches manifests by digest: a digest names one manifest for ever.
|
||||
type Manifests struct {
|
||||
mu sync.Mutex
|
||||
parsed map[string]catalogue.Manifest
|
||||
}
|
||||
|
||||
func (c *Manifests) get(r Reader, digest string) (catalogue.Manifest, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if m, ok := c.parsed[digest]; ok {
|
||||
return m, nil
|
||||
}
|
||||
raw, found, err := r.Get(ReadManifests, digest)
|
||||
if err != nil {
|
||||
return catalogue.Manifest{}, err
|
||||
}
|
||||
if !found {
|
||||
return catalogue.Manifest{}, fmt.Errorf("the indexer holds no manifest %s", digest)
|
||||
}
|
||||
v, err := delivery.DecodeManifest(raw)
|
||||
if err != nil {
|
||||
return catalogue.Manifest{}, err
|
||||
}
|
||||
m, err := catalogue.ParseManifest([]byte(v.JSON))
|
||||
if err != nil {
|
||||
return catalogue.Manifest{}, fmt.Errorf("the manifest %s is refused by mesh-sdk's reader: %w", digest, err)
|
||||
}
|
||||
if c.parsed == nil {
|
||||
c.parsed = map[string]catalogue.Manifest{}
|
||||
}
|
||||
c.parsed[digest] = m
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// each reads every current value of a state; a refusal is a gap and an empty map.
|
||||
func each(r Reader, state string, gaps *[]Gap) map[string][]byte {
|
||||
keys, err := r.Keys(state)
|
||||
if err != nil {
|
||||
*gaps = append(*gaps, Gap{Input: state, Why: "refused or absent: " + err.Error()})
|
||||
return map[string][]byte{}
|
||||
}
|
||||
out := map[string][]byte{}
|
||||
for _, k := range keys {
|
||||
v, found, err := r.Get(state, k)
|
||||
if err != nil {
|
||||
*gaps = append(*gaps, Gap{Input: state + " " + k, Why: err.Error()})
|
||||
continue
|
||||
}
|
||||
if found {
|
||||
out[k] = v
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ReadInputs reads what the shadow composes from. It never fails: what it cannot read is a gap.
|
||||
func ReadInputs(r Reader, cache *Manifests) Inputs {
|
||||
in := Inputs{Assigned: map[string][]string{}, Manifests: map[string]catalogue.Manifest{}, Digests: map[string]string{},
|
||||
Values: map[string]any{}, Gaps: append([]Gap(nil), standingGaps...)}
|
||||
nodes := map[string]bool{}
|
||||
for _, raw := range each(r, ReadAssignments, &in.Gaps) {
|
||||
var a assignment
|
||||
if json.Unmarshal(raw, &a) != nil || !a.stands() || a.Node == "" || a.Module == "" {
|
||||
continue
|
||||
}
|
||||
nodes[a.Node] = true
|
||||
in.Assigned[a.Node] = append(in.Assigned[a.Node], a.Module)
|
||||
if a.Module == BusModule {
|
||||
in.BusNode = a.Node
|
||||
}
|
||||
}
|
||||
haveKeys, err := r.Keys(ReadHave)
|
||||
if err != nil {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: ReadHave, Why: "refused or absent: " + err.Error()})
|
||||
}
|
||||
for _, k := range haveKeys {
|
||||
if node, _, isNode, err := delivery.ParseHaveKey(k); err == nil && isNode {
|
||||
nodes[node] = true
|
||||
}
|
||||
}
|
||||
in.Gaps = append(in.Gaps, Gap{Input: "machines", Why: "read from the node agents' haves and the assignments: a machine with neither is not seen"})
|
||||
for n := range nodes {
|
||||
in.Nodes = append(in.Nodes, n)
|
||||
}
|
||||
sort.Strings(in.Nodes)
|
||||
if in.BusNode != "" {
|
||||
if raw, found, err := r.Get(ReadHave, in.BusNode+"."+BusModule); err != nil {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: "the loaded list's have", Why: err.Error()})
|
||||
} else if found {
|
||||
if h, err := delivery.DecodeHave(raw); err == nil {
|
||||
in.Loaded = h.UserList
|
||||
}
|
||||
}
|
||||
}
|
||||
wanted := map[string]bool{}
|
||||
for node := range in.Assigned {
|
||||
sort.Strings(in.Assigned[node])
|
||||
for _, m := range in.Assigned[node] {
|
||||
wanted[m] = true
|
||||
}
|
||||
}
|
||||
modules := make([]string, 0, len(wanted))
|
||||
for m := range wanted {
|
||||
modules = append(modules, m)
|
||||
}
|
||||
sort.Strings(modules)
|
||||
// The seats every module declares, across the whole catalogue: read every module the indexer holds, not only the
|
||||
// assigned ones, since a seat is declared by one module and held by another.
|
||||
indexed, err := r.Keys(ReadModules)
|
||||
if err != nil {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: ReadModules, Why: "refused or absent: " + err.Error()})
|
||||
}
|
||||
for _, module := range indexed {
|
||||
raw, found, err := r.Get(ReadModules, module)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
v, err := delivery.DecodeModule(raw)
|
||||
if err == nil && v.Manifest == "" {
|
||||
err = fmt.Errorf("the indexer's value names no manifest digest")
|
||||
}
|
||||
if err != nil {
|
||||
if wanted[module] {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: "manifest of " + module, Why: err.Error()})
|
||||
}
|
||||
continue
|
||||
}
|
||||
m, err := cache.get(r, v.Manifest)
|
||||
if err != nil {
|
||||
if wanted[module] {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: "manifest of " + module, Why: err.Error()})
|
||||
}
|
||||
continue
|
||||
}
|
||||
in.Manifests[module] = m
|
||||
in.Digests[module] = v.Manifest
|
||||
}
|
||||
for _, module := range modules {
|
||||
if _, ok := in.Manifests[module]; !ok && !gapNamed(in.Gaps, "manifest of "+module) {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: "manifest of " + module,
|
||||
Why: "assigned, and the indexer holds no manifest of it at the trunk (a module registered from outside the indexed repositories): composed with no authority"})
|
||||
}
|
||||
}
|
||||
for _, raw := range each(r, ReadResolved, &in.Gaps) {
|
||||
var v resolved
|
||||
if json.Unmarshal(raw, &v) != nil || v.Node == "" || v.Module == "" || v.Key == "" || len(v.Value) == 0 {
|
||||
continue
|
||||
}
|
||||
var value any
|
||||
if json.Unmarshal(v.Value, &value) == nil {
|
||||
in.Values[v.Node+"/"+v.Module+"/"+v.Key] = value
|
||||
}
|
||||
}
|
||||
sortGaps(in.Gaps)
|
||||
return in
|
||||
}
|
||||
|
||||
func gapNamed(gaps []Gap, input string) bool {
|
||||
for _, g := range gaps {
|
||||
if g.Input == input {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func sortGaps(g []Gap) {
|
||||
sort.SliceStable(g, func(i, j int) bool { return strings.Compare(g[i].Input, g[j].Input) < 0 })
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
)
|
||||
|
||||
// PublishedUnread is why the membership as published is not compared: the bus grants each module the membership of its
|
||||
// own assignment alone, so no other part reads another's from the assignments stream.
|
||||
const PublishedUnread = "not readable from here: the bus grants each module its own membership alone (and the controller has published none since release 6 removed the push, issue 1945 note 44)"
|
||||
|
||||
// MembershipGaps are what a membership carries that the shadow cannot compose from the bus.
|
||||
var MembershipGaps = []Gap{
|
||||
{Input: "membership: private network and received contributions", Why: "kept in each node's composition, which only that node reads: composed without them"},
|
||||
{Input: "membership: root-free machines", Why: "judged by the controller on the bus at its composition, not kept: a verified sender is composed as from no root-free machine"},
|
||||
}
|
||||
|
||||
// Membership is one module's membership on one node, as composed now, and its digest.
|
||||
type Membership struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
Digest string `json:"digest"`
|
||||
Membership membership.Membership `json:"membership"`
|
||||
Published string `json:"published"`
|
||||
}
|
||||
|
||||
// Memberships composes every assignment's membership from the records (buscompose.MembershipFor, as the controller's
|
||||
// parked composition pass, mesh-controller 003eee8a, issued them), by `<node>.<module>`. Pure.
|
||||
func Memberships(r buscompose.Records) map[string]Membership {
|
||||
where := buscompose.PlacementsOf(r, r.Interchangeable)
|
||||
out := map[string]Membership{}
|
||||
for node, declared := range r.Assigned {
|
||||
for _, d := range declared {
|
||||
m := buscompose.MembershipFor(node, d, where)
|
||||
raw, _ := json.Marshal(m)
|
||||
sum := sha256.Sum256(raw)
|
||||
out[node+"."+d.Module] = Membership{Node: node, Module: d.Module, Digest: "sha256:" + hex.EncodeToString(sum[:]),
|
||||
Membership: m, Published: PublishedUnread}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,316 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The part's own states, as its manifest declares them, and the keys of the two that hold one value.
|
||||
const (
|
||||
StateUserList = "user-list"
|
||||
StateMemberships = "memberships"
|
||||
StateAssertions = "assertions"
|
||||
StateTakeover = "takeover"
|
||||
KeyList = "list"
|
||||
KeySwitch = "switch"
|
||||
)
|
||||
|
||||
// States is every state the part keeps.
|
||||
var States = []string{StateAssertions, StateMemberships, StateTakeover, StateUserList}
|
||||
|
||||
// The events the part says.
|
||||
const (
|
||||
EventUserList = "user-list-changed"
|
||||
EventMembership = "membership-changed"
|
||||
)
|
||||
|
||||
// Emits is every event the part says.
|
||||
var Emits = []string{EventMembership, EventUserList}
|
||||
|
||||
// Shadow is the switch's one state in this milestone: the bus node follows the controller's list.
|
||||
const Shadow = "shadow"
|
||||
|
||||
// Store is the part's own state, by local name: the only thing the part writes.
|
||||
type Store interface {
|
||||
Get(state, key string) ([]byte, uint64, bool, error)
|
||||
Put(state, key string, value []byte) (uint64, error)
|
||||
}
|
||||
|
||||
// Emitter says one of the part's events.
|
||||
type Emitter interface {
|
||||
Emit(event string, body any) error
|
||||
}
|
||||
|
||||
// UserListState is the value of `user-list` in shadow: the list the grants part composed (users and grants, no hashes),
|
||||
// what the bus loaded, and how they differ. Inputs is the digest of all it says but At: a value is written again only
|
||||
// when it changes.
|
||||
type UserListState struct {
|
||||
V int `json:"v"`
|
||||
Mode string `json:"mode"`
|
||||
Inputs string `json:"inputs"`
|
||||
Digest string `json:"digest"`
|
||||
Users int `json:"users"`
|
||||
// LeftOut are the assignments the shadow list leaves out, each with why.
|
||||
LeftOut []Gap `json:"left_out"`
|
||||
Compare Comparison `json:"compare"`
|
||||
Loaded *delivery.UserListWritten `json:"loaded,omitempty"`
|
||||
BusNode string `json:"bus_node,omitempty"`
|
||||
Gaps []Gap `json:"gaps"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
func (s UserListState) inputs() string {
|
||||
s.At, s.Inputs = time.Time{}, ""
|
||||
raw, _ := json.Marshal(s)
|
||||
sum := sha256.Sum256(raw)
|
||||
return "sha256:" + hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// Result is what one composition did.
|
||||
type Result struct {
|
||||
UserList UserListState `json:"user_list"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Written bool `json:"written"`
|
||||
Memberships int `json:"memberships"`
|
||||
MembershipsWritten int `json:"memberships_written"`
|
||||
Assertions int `json:"assertions"`
|
||||
AssertionsWritten int `json:"assertions_written"`
|
||||
Errors []string `json:"errors,omitempty"`
|
||||
PublishedOrAsserted string `json:"published_or_asserted"`
|
||||
}
|
||||
|
||||
// NothingTakesEffect is what every composition says it did on the bus beyond its own state.
|
||||
const NothingTakesEffect = "nothing: in shadow the grants part writes only its own state (no membership, assertion, grant or user list the bus node reads)"
|
||||
|
||||
// Part is the grants part: it reads, calls and writes through these, and nothing else.
|
||||
type Part struct {
|
||||
Read Reader
|
||||
Call Caller
|
||||
Store Store
|
||||
Emit Emitter
|
||||
Now func() time.Time
|
||||
|
||||
cache Manifests
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func (p *Part) now() time.Time {
|
||||
if p.Now == nil {
|
||||
return time.Now().UTC()
|
||||
}
|
||||
return p.Now().UTC()
|
||||
}
|
||||
|
||||
// Compose composes the user list, the memberships and the assertions from what the bus holds now, compares them, and
|
||||
// writes each of its own keys whose value changed, saying each change. It never stops at a refusal: a read refused is
|
||||
// a gap, a write refused is said in Errors, and the next event composes again.
|
||||
func (p *Part) Compose() Result {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
res := Result{PublishedOrAsserted: NothingTakesEffect}
|
||||
in := ReadInputs(p.Read, &p.cache)
|
||||
recs, left := Records(in)
|
||||
st := UserListState{V: 1, Mode: Shadow, LeftOut: left, Loaded: in.Loaded, BusNode: in.BusNode,
|
||||
Gaps: append(append([]Gap(nil), in.Gaps...), MembershipGaps...), At: p.now()}
|
||||
if st.LeftOut == nil {
|
||||
st.LeftOut = []Gap{}
|
||||
}
|
||||
users, err := ComposeUsers(recs)
|
||||
if err != nil {
|
||||
res.Errors = append(res.Errors, "the shadow list could not be composed: "+err.Error())
|
||||
}
|
||||
st.Digest, st.Users = Digest(users), len(users)
|
||||
loaded, err := LoadedUsers(p.Call)
|
||||
st.Compare = Compare(users, loaded, Placed(in))
|
||||
if err != nil {
|
||||
st.Compare.Unread = err.Error()
|
||||
}
|
||||
st.Inputs = st.inputs()
|
||||
res.UserList = st
|
||||
if raw, rev, found, err := p.Store.Get(StateUserList, KeyList); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("its own %s cannot be read: %v", StateUserList, err))
|
||||
} else {
|
||||
var was UserListState
|
||||
if found && json.Unmarshal(raw, &was) == nil && was.Inputs == st.Inputs {
|
||||
res.Revision = rev
|
||||
} else if data, err := json.Marshal(st); err == nil {
|
||||
if rev, err := p.Store.Put(StateUserList, KeyList, data); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("its own %s cannot be written: %v", StateUserList, err))
|
||||
} else {
|
||||
res.Revision, res.Written = rev, true
|
||||
p.say(&res, EventUserList, map[string]any{"digest": st.Digest, "revision": rev, "users": st.Users,
|
||||
"differing": len(st.Compare.Differing()), "mode": Shadow})
|
||||
}
|
||||
}
|
||||
}
|
||||
ms := Memberships(recs)
|
||||
res.Memberships = len(ms)
|
||||
for _, key := range sortedKeys(ms) {
|
||||
m := ms[key]
|
||||
data, _ := json.Marshal(m)
|
||||
if p.changed(&res, StateMemberships, key, data) {
|
||||
res.MembershipsWritten++
|
||||
p.say(&res, EventMembership, map[string]any{"node": m.Node, "module": m.Module, "digest": m.Digest, "mode": Shadow})
|
||||
}
|
||||
}
|
||||
held, heldErr := HeldStreams(p.Call)
|
||||
for _, module := range sortedKeys(in.Manifests) {
|
||||
if !assigned(in, module) {
|
||||
continue
|
||||
}
|
||||
a := Assert(module, in.Digests[module], in.Manifests[module], held, heldErr)
|
||||
res.Assertions++
|
||||
data, _ := json.Marshal(a)
|
||||
if p.changed(&res, StateAssertions, module, data) {
|
||||
res.AssertionsWritten++
|
||||
}
|
||||
}
|
||||
if _, _, found, err := p.Store.Get(StateTakeover, KeySwitch); err == nil && !found {
|
||||
data, _ := json.Marshal(Switch{V: 1, State: Shadow, Acts: []Act{}})
|
||||
if _, err := p.Store.Put(StateTakeover, KeySwitch, data); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("its own %s cannot be written: %v", StateTakeover, err))
|
||||
}
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
func assigned(in Inputs, module string) bool {
|
||||
for _, ms := range in.Assigned {
|
||||
for _, m := range ms {
|
||||
if m == module {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// changed writes key when its value differs from the one held, and says whether it wrote.
|
||||
func (p *Part) changed(res *Result, state, key string, data []byte) bool {
|
||||
raw, _, found, err := p.Store.Get(state, key)
|
||||
if err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be read: %v", state, key, err))
|
||||
return false
|
||||
}
|
||||
if found && bytes.Equal(raw, data) {
|
||||
return false
|
||||
}
|
||||
if _, err := p.Store.Put(state, key, data); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be written: %v", state, key, err))
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (p *Part) say(res *Result, event string, body any) {
|
||||
if p.Emit == nil {
|
||||
return
|
||||
}
|
||||
if err := p.Emit.Emit(event, body); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("the event %s could not be said: %v", event, err))
|
||||
}
|
||||
}
|
||||
|
||||
func sortedKeys[V any](m map[string]V) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// Switch is the value of `takeover`: which list the bus node follows, and every act on the switch.
|
||||
type Switch struct {
|
||||
V int `json:"v"`
|
||||
State string `json:"state"`
|
||||
Acts []Act `json:"acts"`
|
||||
}
|
||||
|
||||
// Act is one take-over or hand-back asked, done or refused.
|
||||
type Act struct {
|
||||
Act string `json:"act"`
|
||||
By string `json:"by"`
|
||||
Why string `json:"why,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
Refused string `json:"refused,omitempty"`
|
||||
}
|
||||
|
||||
// keptActs is how many acts the switch keeps, newest last.
|
||||
const keptActs = 50
|
||||
|
||||
// Operator says whether a caller is the operator's credential: a person's own bus account (`person.<name>`), which no
|
||||
// session of an agent holds (an agent calls as its machine's tool runner).
|
||||
func Operator(caller string) bool {
|
||||
name, ok := strings.CutPrefix(caller, "person.")
|
||||
return ok && name != ""
|
||||
}
|
||||
|
||||
// NotComposable is why take-over is refused in this milestone, whatever the lists say.
|
||||
const NotComposable = "the enrolment users and the persons cannot be composed: they live only in the controller's store, and take-over waits until they have records on the bus (ADR 0547 clause 9, version 26)"
|
||||
|
||||
// record keeps an act on the switch; a refused write is said in the answer.
|
||||
func (p *Part) record(a Act) string {
|
||||
raw, _, found, err := p.Store.Get(StateTakeover, KeySwitch)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("; the act is not recorded: %v", err)
|
||||
}
|
||||
s := Switch{V: 1, State: Shadow}
|
||||
if found {
|
||||
_ = json.Unmarshal(raw, &s)
|
||||
}
|
||||
s.Acts = append(s.Acts, a)
|
||||
if len(s.Acts) > keptActs {
|
||||
s.Acts = s.Acts[len(s.Acts)-keptActs:]
|
||||
}
|
||||
data, _ := json.Marshal(s)
|
||||
if _, err := p.Store.Put(StateTakeover, KeySwitch, data); err != nil {
|
||||
return fmt.Sprintf("; the act is not recorded: %v", err)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// TakeOver is the operator's switch to the grants part's list. It is refused without the operator's credential, while
|
||||
// any user of the shadow list differs from the loaded list (but those both agree are gone), and in this milestone
|
||||
// always, since the enrolment users and the persons cannot be composed. Every ask is recorded.
|
||||
func (p *Part) TakeOver(caller, why string) error {
|
||||
a := Act{Act: "take-over", By: caller, Why: why, At: p.now()}
|
||||
if !Operator(caller) {
|
||||
a.Refused = fmt.Sprintf("take-over is the operator's alone: %s is not a person's account", caller)
|
||||
return fmt.Errorf("%s%s", a.Refused, p.record(a))
|
||||
}
|
||||
res := p.Compose()
|
||||
var reasons []string
|
||||
if c := res.UserList.Compare; !c.Equal() {
|
||||
if c.Unread != "" {
|
||||
reasons = append(reasons, "the loaded list could not be read: "+c.Unread)
|
||||
}
|
||||
if d := c.Differing(); len(d) > 0 {
|
||||
reasons = append(reasons, fmt.Sprintf("%d users differ between the shadow list and the loaded list: %s", len(d), strings.Join(d, ", ")))
|
||||
}
|
||||
}
|
||||
reasons = append(reasons, NotComposable)
|
||||
a.Refused = "take-over refused: " + strings.Join(reasons, "; ")
|
||||
return fmt.Errorf("%s%s", a.Refused, p.record(a))
|
||||
}
|
||||
|
||||
// HandBack is the operator's switch back to the controller's list. Refused without the operator's credential, and
|
||||
// while the part is in shadow, since the bus node follows the controller's list already. Every ask is recorded.
|
||||
func (p *Part) HandBack(caller, why string) error {
|
||||
a := Act{Act: "hand-back", By: caller, Why: why, At: p.now()}
|
||||
if !Operator(caller) {
|
||||
a.Refused = fmt.Sprintf("hand-back is the operator's alone: %s is not a person's account", caller)
|
||||
return fmt.Errorf("%s%s", a.Refused, p.record(a))
|
||||
}
|
||||
a.Refused = "nothing to hand back: the grants part is in shadow and never took over, so the bus node follows the controller's list already"
|
||||
return fmt.Errorf("%s%s", a.Refused, p.record(a))
|
||||
}
|
||||
@@ -0,0 +1,266 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// Records is what the bus composer composes from, rebuilt from the bus (mesh-controller internal/inventory busrecords.go,
|
||||
// BusRecords, read the same way from Inputs instead of the controller's store). Pure. An assigned module whose manifest
|
||||
// the indexer does not hold is left out and named, where the controller refuses the whole composition: the shadow
|
||||
// composes what it can and says what it could not.
|
||||
func Records(in Inputs) (buscompose.Records, []Gap) {
|
||||
var gaps []Gap
|
||||
seats := map[string]catalogue.SeatDeclaration{}
|
||||
declarers := map[string]string{}
|
||||
for _, m := range in.Manifests {
|
||||
for _, s := range m.DefinesSeats {
|
||||
seats[s.Name] = s
|
||||
declarers[s.Name] = m.Module
|
||||
}
|
||||
}
|
||||
for _, own := range catalogue.SeatsWithAProtocol() {
|
||||
seats[own.Name] = catalogue.SeatDeclaration{Name: own.Name, Scope: own.Scope, Accepts: own.Accepts, Emits: own.Emits, Serves: own.Serves}
|
||||
}
|
||||
out := buscompose.Records{Assigned: map[string][]buscompose.Declared{}, People: map[string][]string{},
|
||||
Interchangeable: map[string]bool{}}
|
||||
values := func(provider, setting string) []any {
|
||||
var vs []any
|
||||
for _, node := range in.Nodes {
|
||||
if v, ok := in.Values[node+"/"+provider+"/"+setting]; ok && slices.Contains(in.Assigned[node], provider) {
|
||||
vs = append(vs, v)
|
||||
}
|
||||
}
|
||||
return vs
|
||||
}
|
||||
for _, node := range in.Nodes {
|
||||
out.Nodes = append(out.Nodes, node)
|
||||
var here []catalogue.Manifest
|
||||
for _, module := range in.Assigned[node] {
|
||||
if m, ok := in.Manifests[module]; ok {
|
||||
here = append(here, m)
|
||||
}
|
||||
}
|
||||
for _, module := range in.Assigned[node] {
|
||||
m, known := in.Manifests[module]
|
||||
if !known {
|
||||
gaps = append(gaps, Gap{Input: module + " on " + node, Why: "left out of the shadow list: no manifest to derive what it may say from"})
|
||||
continue
|
||||
}
|
||||
d := declaredFor(m, seats, declarers)
|
||||
for _, sr := range catalogue.ReadsGranted(m, here, node) {
|
||||
d.KeyedReads = append(d.KeyedReads, buscompose.KeyedRead{Read: sr.Read, Key: sr.Key})
|
||||
}
|
||||
d.Hears, d.HearsAStream, d.Consumes = hearsOf(m, in.Manifests, node, values, d.Consumes)
|
||||
out.Assigned[node] = append(out.Assigned[node], d)
|
||||
if m.Instances == catalogue.InstancesInterchangeable {
|
||||
out.Interchangeable[m.Module] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
return out, gaps
|
||||
}
|
||||
|
||||
// declaredFor is mesh-controller's declaredFor (internal/inventory/busrecords.go), copied: one module's manifest as the
|
||||
// composer needs it.
|
||||
func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaration, declarers map[string]string) buscompose.Declared {
|
||||
var fromModules []string
|
||||
var watches []buscompose.Seat
|
||||
for _, c := range m.Consumes {
|
||||
emitter, event, named := strings.Cut(c, ".")
|
||||
if named {
|
||||
if s, isASeat := seats[emitter]; isASeat && s.Kinded {
|
||||
ev := strings.TrimSuffix(event, ".*")
|
||||
if catalogue.SeatSays(s.Emits, ev) {
|
||||
watches = append(watches, buscompose.Seat{Name: s.Name, Scope: s.Scope, Emits: []string{ev},
|
||||
Kinded: true, Proofs: s.Proofs, DeclaredBy: declarers[s.Name]})
|
||||
continue
|
||||
}
|
||||
}
|
||||
if s, isASeat := seats[emitter]; isASeat && catalogue.SeatSays(s.Emits, event) {
|
||||
watches = append(watches, buscompose.Seat{Name: s.Name, Emits: []string{event}})
|
||||
continue
|
||||
}
|
||||
}
|
||||
fromModules = append(fromModules, c)
|
||||
}
|
||||
d := buscompose.Declared{
|
||||
Module: m.Module, Emits: m.EmitsAll(), Consumes: fromModules, Watches: watches, Serves: m.Tools,
|
||||
Invokes: m.Invokes, Waits: waitsOf(m, seats), State: bucketsOf(m), Reads: m.Reads, Logs: logsOf(m),
|
||||
Streams: streamsOf(m), Queues: queueOwnersOf(m), Processes: processesOf(m), Checks: catalogue.HealthChecks(m),
|
||||
RunsAs: m.RunsAs, Issued: issuedTools(m),
|
||||
}
|
||||
for _, b := range m.Bundles {
|
||||
if len(b.Loads) > 0 {
|
||||
d.Loads = true
|
||||
}
|
||||
}
|
||||
if _, reads := m.OwnSecrets["broker"]; !reads {
|
||||
d.NoAccount = true
|
||||
}
|
||||
if _, reads := m.OwnSecrets[buscompose.ProvisionerSecret]; reads && m.ClaimsSeat(buscompose.NodeProvisionerSeat) {
|
||||
d.ProvisionsNode = true
|
||||
}
|
||||
d.SnapshotsTheBus = m.ClaimsSeat(catalogue.BrokerSeat)
|
||||
for _, c := range m.Claims {
|
||||
if s, hasAProtocol := seats[c.Name]; hasAProtocol {
|
||||
held := asSeat(s, declarers[s.Name])
|
||||
held.Kind = c.Kind
|
||||
held.Capabilities = c.Capabilities
|
||||
d.Holds = append(d.Holds, held)
|
||||
}
|
||||
}
|
||||
for _, name := range m.Uses {
|
||||
if s, declaredSomewhere := seats[name]; declaredSomewhere {
|
||||
d.Uses = append(d.Uses, asSeat(s, declarers[s.Name]))
|
||||
}
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
func waitsOf(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaration) []buscompose.Wait {
|
||||
var out []buscompose.Wait
|
||||
for _, w := range m.Waits {
|
||||
emitter, event, _ := strings.Cut(w, ".")
|
||||
first, _, _ := strings.Cut(event, ".")
|
||||
wait := buscompose.Wait{Name: w, Emitter: emitter, Event: event}
|
||||
if s, isASeat := seats[emitter]; isASeat {
|
||||
switch {
|
||||
case s.Kinded && catalogue.SeatSays(s.Emits, first):
|
||||
wait.Seat, wait.Last = true, "kind"
|
||||
case s.NamedByCaller(first) && catalogue.SeatSays(s.Emits, first):
|
||||
wait.Seat, wait.Last = true, "asker"
|
||||
case catalogue.SeatSays(s.Emits, event):
|
||||
wait.Seat = true
|
||||
}
|
||||
}
|
||||
out = append(out, wait)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func issuedTools(m catalogue.Manifest) []string {
|
||||
var out []string
|
||||
for _, o := range m.Provides {
|
||||
if tool, issued := m.IssuedCredentialOf(o.Name); issued && !slices.Contains(out, tool) {
|
||||
out = append(out, tool)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func bucketsOf(m catalogue.Manifest) []buscompose.Bucket {
|
||||
var out []buscompose.Bucket
|
||||
for _, s := range m.State {
|
||||
out = append(out, buscompose.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds,
|
||||
PerMachine: s.PerMachine, KeyTTL: s.KeyTTL})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func logsOf(m catalogue.Manifest) []buscompose.Log {
|
||||
var out []buscompose.Log
|
||||
for _, l := range m.Logs {
|
||||
out = append(out, buscompose.Log{Module: m.Module, Name: l.Name, MaxMiB: l.Cap()})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func asSeat(s catalogue.SeatDeclaration, declarer string) buscompose.Seat {
|
||||
seat := buscompose.Seat{Name: s.Name, Scope: s.Scope, Accepts: s.Accepts, Emits: s.Emits,
|
||||
Serves: catalogue.VerbNames(s.Serves), Kinded: s.Kinded, ByCaller: s.ByCaller, Proofs: s.Proofs, DeclaredBy: declarer}
|
||||
if own, ok := catalogue.SeatNamed(s.Name); ok && own.PulledByKind {
|
||||
seat.PulledByKind = true
|
||||
}
|
||||
for _, r := range s.Records {
|
||||
if declarer != "" {
|
||||
seat.Records = append(seat.Records, buscompose.BucketName(declarer, r))
|
||||
}
|
||||
}
|
||||
return seat
|
||||
}
|
||||
|
||||
func streamsOf(m catalogue.Manifest) []buscompose.ModuleStream {
|
||||
var out []buscompose.ModuleStream
|
||||
for _, s := range m.Streams {
|
||||
events := make([]string, 0, len(s.Events))
|
||||
for _, e := range s.Events {
|
||||
events = append(events, buscompose.WirePattern(e))
|
||||
}
|
||||
out = append(out, buscompose.ModuleStream{Module: m.Module, Name: s.BusName(), Events: events,
|
||||
MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil,
|
||||
WorkQueue: s.Retention == catalogue.StreamRetentionWorkQueue, Duplicates: s.Duplicates})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func queueOwnersOf(m catalogue.Manifest) []buscompose.QueueOwner {
|
||||
buses := m.ProcessBuses()
|
||||
if len(buses) == 0 {
|
||||
return nil
|
||||
}
|
||||
byName := map[string]catalogue.StreamDeclaration{}
|
||||
for _, s := range m.Streams {
|
||||
byName[s.Name] = s
|
||||
}
|
||||
var out []buscompose.QueueOwner
|
||||
for _, r := range m.Resources {
|
||||
id := fmt.Sprint(r["id"])
|
||||
b, declared := buses[id]
|
||||
if !declared {
|
||||
continue
|
||||
}
|
||||
for _, q := range b.Queues {
|
||||
s, known := byName[q]
|
||||
if !known || s.Retention != catalogue.StreamRetentionWorkQueue {
|
||||
continue
|
||||
}
|
||||
out = append(out, buscompose.QueueOwner{Process: id, Stream: s.BusName(), MaxDeliver: s.MaxDeliveries})
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func processesOf(m catalogue.Manifest) []buscompose.ProcessPrincipal {
|
||||
buses := m.ProcessBuses()
|
||||
if len(buses) == 0 {
|
||||
return nil
|
||||
}
|
||||
queues := queueOwnersOf(m)
|
||||
var out []buscompose.ProcessPrincipal
|
||||
for _, r := range m.Resources {
|
||||
id := fmt.Sprint(r["id"])
|
||||
b, declared := buses[id]
|
||||
if !declared {
|
||||
continue
|
||||
}
|
||||
pr := buscompose.ProcessPrincipal{Process: id, Emits: b.Emits, State: b.State, Asks: b.Asks, Invokes: b.Invokes}
|
||||
for _, q := range queues {
|
||||
if q.Process == id {
|
||||
pr.Queues = append(pr.Queues, q)
|
||||
}
|
||||
}
|
||||
out = append(out, pr)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node string, values func(provider, setting string) []any,
|
||||
consumes []string) ([]buscompose.Heard, bool, []string) {
|
||||
reading := m
|
||||
reading.Consumes = consumes
|
||||
parts, touches, onEvents := catalogue.Hearing(reading, declared, node, values)
|
||||
var out []buscompose.Heard
|
||||
for _, p := range parts {
|
||||
h := buscompose.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter, Restricted: p.Stream.Grant != nil}
|
||||
for _, pattern := range p.Patterns {
|
||||
h.Filters = append(h.Filters, buscompose.ModuleStreamSubject(p.Emitter, buscompose.WirePattern(pattern)))
|
||||
}
|
||||
out = append(out, h)
|
||||
}
|
||||
return out, touches, onEvents
|
||||
}
|
||||
@@ -0,0 +1,150 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The part's tools, as its manifest lists them. Module tools in this milestone: a `mesh-*` seat must be in mesh-sdk's own
|
||||
// seat list, which the running controller's composer does not know, so the part claims no seat until take-over.
|
||||
const (
|
||||
ToolGrants = "grants_grants"
|
||||
ToolMembership = "grants_membership"
|
||||
ToolUserList = "grants_user_list"
|
||||
ToolComposeNow = "grants_compose_now"
|
||||
ToolAssert = "grants_assert"
|
||||
ToolTakeOver = "grants_take_over"
|
||||
ToolHandBack = "grants_hand_back"
|
||||
)
|
||||
|
||||
// Tools is every tool, in the manifest's order.
|
||||
var Tools = []string{ToolGrants, ToolMembership, ToolUserList, ToolComposeNow, ToolAssert, ToolTakeOver, ToolHandBack}
|
||||
|
||||
// WritesNoGrant is what grants says of the part's own grants in this milestone.
|
||||
const WritesNoGrant = "none: the grants part writes no grants.<module>.<fp> before ADR 0547's third milestone (clause 5); the controller's are beside"
|
||||
|
||||
// Grants answers the tool grants: what the part writes of grants (none), and the controller's, one module's or all.
|
||||
func (p *Part) Grants(module string) (any, error) {
|
||||
keys, err := p.Read.Keys(ReadControllerGrant)
|
||||
if err != nil {
|
||||
return map[string]any{"writes": WritesNoGrant, "controller_unread": err.Error()}, nil
|
||||
}
|
||||
type grant struct {
|
||||
Module string `json:"module"`
|
||||
Fingerprint string `json:"fingerprint"`
|
||||
Accounts []string `json:"accounts,omitempty"`
|
||||
}
|
||||
out := []grant{}
|
||||
for _, k := range keys {
|
||||
m, fp, err := delivery.ParseControllerGrantKey(k)
|
||||
if err != nil || (module != "" && m != module) {
|
||||
continue
|
||||
}
|
||||
g := grant{Module: m, Fingerprint: fp}
|
||||
if module != "" {
|
||||
if raw, found, err := p.Read.Get(ReadControllerGrant, k); err == nil && found {
|
||||
if v, err := delivery.DecodeControllerGrant(raw); err == nil {
|
||||
g.Accounts = v.Accounts
|
||||
}
|
||||
}
|
||||
}
|
||||
out = append(out, g)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Module+out[i].Fingerprint < out[j].Module+out[j].Fingerprint })
|
||||
return map[string]any{"writes": WritesNoGrant, "controller": out, "count": len(out)}, nil
|
||||
}
|
||||
|
||||
// MembershipOf answers the tool membership: one module on one node as composed now, as last kept, and as published.
|
||||
func (p *Part) MembershipOf(node, module string) (any, error) {
|
||||
if node == "" || module == "" {
|
||||
return nil, errors.New("membership names a node and a module")
|
||||
}
|
||||
in := ReadInputs(p.Read, &p.cache)
|
||||
recs, left := Records(in)
|
||||
m, ok := Memberships(recs)[node+"."+module]
|
||||
out := map[string]any{"node": node, "module": module, "published": PublishedUnread, "gaps": MembershipGaps}
|
||||
if ok {
|
||||
out["composed"] = m
|
||||
} else {
|
||||
why := fmt.Sprintf("%s is not assigned to %s on the bus", module, node)
|
||||
for _, g := range left {
|
||||
if g.Input == module+" on "+node {
|
||||
why = g.Why
|
||||
}
|
||||
}
|
||||
out["composed"] = why
|
||||
}
|
||||
if raw, rev, found, err := p.Store.Get(StateMemberships, node+"."+module); err != nil {
|
||||
out["kept"] = "not readable: " + err.Error()
|
||||
} else if found {
|
||||
out["kept"] = map[string]any{"revision": rev, "value": json.RawMessage(raw)}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// UserList answers the tool user-list: the shadow list as last written, beside what the bus node says it loaded now.
|
||||
func (p *Part) UserList() (any, error) {
|
||||
out := map[string]any{"mode": Shadow}
|
||||
raw, rev, found, err := p.Store.Get(StateUserList, KeyList)
|
||||
switch {
|
||||
case err != nil:
|
||||
out["written"] = "not readable: " + err.Error()
|
||||
case !found:
|
||||
out["written"] = "nothing composed yet: grants_compose_now composes it"
|
||||
default:
|
||||
var st UserListState
|
||||
if err := json.Unmarshal(raw, &st); err != nil {
|
||||
return nil, fmt.Errorf("its own user list cannot be read: %w", err)
|
||||
}
|
||||
out["revision"], out["written"] = rev, st
|
||||
if st.BusNode != "" {
|
||||
if raw, found, err := p.Read.Get(ReadHave, st.BusNode+"."+BusModule); err == nil && found {
|
||||
if h, err := delivery.DecodeHave(raw); err == nil {
|
||||
out["loaded_now"] = h.UserList
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// AssertOne answers the tool assert: one module's buckets, logs and streams as composed from its manifest at the
|
||||
// indexer's trunk, beside what the bus holds. It asserts nothing.
|
||||
func (p *Part) AssertOne(module string) (any, error) {
|
||||
if module == "" {
|
||||
return nil, errors.New("assert names a module")
|
||||
}
|
||||
raw, found, err := p.Read.Get(ReadModules, module)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the indexer's %s cannot be read: %w", module, err)
|
||||
}
|
||||
if !found {
|
||||
return nil, fmt.Errorf("the indexer holds no module %s", module)
|
||||
}
|
||||
v, err := delivery.DecodeModule(raw)
|
||||
if err != nil || v.Manifest == "" {
|
||||
return nil, fmt.Errorf("the indexer names no manifest of %s", module)
|
||||
}
|
||||
m, err := p.cache.get(p.Read, v.Manifest)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
held, heldErr := HeldStreams(p.Call)
|
||||
return Assert(module, v.Manifest, m, held, heldErr), nil
|
||||
}
|
||||
|
||||
// Answer is a tool's arguments read as text.
|
||||
func Answer(args map[string]any, name string) string {
|
||||
switch v := args[name].(type) {
|
||||
case string:
|
||||
return strings.TrimSpace(v)
|
||||
case bool:
|
||||
return fmt.Sprint(v)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
)
|
||||
|
||||
// User is one bus user and its grants, as the bus module's nats_users answers them: never a password hash.
|
||||
type User struct {
|
||||
User string `json:"user"`
|
||||
Publish []string `json:"publish"`
|
||||
Subscribe []string `json:"subscribe"`
|
||||
AllowResponses bool `json:"allow_responses"`
|
||||
}
|
||||
|
||||
// ComposeUsers is the shadow list: every user the composer derives from the records, with its grants. Pure.
|
||||
func ComposeUsers(r buscompose.Records) ([]User, error) {
|
||||
principals, err := buscompose.Users(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]User, 0, len(principals))
|
||||
for _, p := range principals {
|
||||
perms, err := buscompose.PermissionsFor(p)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", p.Username(), err)
|
||||
}
|
||||
out = append(out, User{User: p.Username(), Publish: set(perms.Publish), Subscribe: set(perms.Subscribe),
|
||||
AllowResponses: perms.AllowResponses})
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].User < out[j].User })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func set(in []string) []string {
|
||||
out := slices.Clone(in)
|
||||
sort.Strings(out)
|
||||
out = slices.Compact(out)
|
||||
if out == nil {
|
||||
out = []string{}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Digest is a list's digest, `sha256:<hex>`, over its users and grants in order.
|
||||
func Digest(users []User) string {
|
||||
raw, _ := json.Marshal(users)
|
||||
sum := sha256.Sum256(raw)
|
||||
return "sha256:" + hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// LoadedUsers is the list the bus loaded, as its module's nats_users reads it from the file: every user by name, then
|
||||
// each one's grants in full.
|
||||
func LoadedUsers(c Caller) ([]User, error) {
|
||||
raw, err := c.Call(CallUsers, map[string]any{})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", CallUsers, err)
|
||||
}
|
||||
var names struct {
|
||||
Users []struct {
|
||||
User string `json:"user"`
|
||||
} `json:"users"`
|
||||
}
|
||||
if err := json.Unmarshal(unwrap(raw), &names); err != nil {
|
||||
return nil, fmt.Errorf("%s answered no list of users: %w", CallUsers, err)
|
||||
}
|
||||
full := map[string]User{}
|
||||
ask := func(part string) error {
|
||||
raw, err := c.Call(CallUsers, map[string]any{"user": part})
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s %s: %w", CallUsers, part, err)
|
||||
}
|
||||
var got struct {
|
||||
Users []User `json:"users"`
|
||||
}
|
||||
if err := json.Unmarshal(unwrap(raw), &got); err != nil {
|
||||
return fmt.Errorf("%s %s answered no grants: %w", CallUsers, part, err)
|
||||
}
|
||||
for _, u := range got.Users {
|
||||
u.Publish, u.Subscribe = set(u.Publish), set(u.Subscribe)
|
||||
full[u.User] = u
|
||||
}
|
||||
return nil
|
||||
}
|
||||
// Every name with a dot in one call; each name without one (the controller) alone.
|
||||
if err := ask("."); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]User, 0, len(names.Users))
|
||||
for _, n := range names.Users {
|
||||
if _, ok := full[n.User]; !ok {
|
||||
if err := ask(n.User); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
u, ok := full[n.User]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("%s names %s and gives no grants for it", CallUsers, n.User)
|
||||
}
|
||||
out = append(out, u)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].User < out[j].User })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// unwrap takes a tool's answer out of an MCP content wrapper, when a runtime hands one.
|
||||
func unwrap(raw json.RawMessage) []byte {
|
||||
var wrapped struct {
|
||||
Content []struct {
|
||||
Text string `json:"text"`
|
||||
} `json:"content"`
|
||||
}
|
||||
if json.Unmarshal(raw, &wrapped) == nil && len(wrapped.Content) == 1 && wrapped.Content[0].Text != "" {
|
||||
return []byte(wrapped.Content[0].Text)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
// UserDiff is how one user's grants differ: what the shadow list grants that the loaded list does not, and back.
|
||||
type UserDiff struct {
|
||||
User string `json:"user"`
|
||||
PublishAdded []string `json:"publish_added,omitempty"`
|
||||
PublishDropped []string `json:"publish_dropped,omitempty"`
|
||||
SubscribeAdded []string `json:"subscribe_added,omitempty"`
|
||||
SubscribeDropped []string `json:"subscribe_dropped,omitempty"`
|
||||
Responses string `json:"responses,omitempty"`
|
||||
}
|
||||
|
||||
// Comparison is the shadow list against the loaded list.
|
||||
type Comparison struct {
|
||||
Shadow int `json:"shadow"`
|
||||
Loaded int `json:"loaded"`
|
||||
// Added are users in the shadow list the loaded list lacks (one left out for want of a credential among them);
|
||||
// Dropped are loaded users the shadow list lacks; Gone are those of Dropped whose machine or module placement no
|
||||
// longer names, which both lists agree are gone.
|
||||
Added []string `json:"added"`
|
||||
Dropped []string `json:"dropped"`
|
||||
Gone []string `json:"gone,omitempty"`
|
||||
Differ []UserDiff `json:"differ"`
|
||||
Same int `json:"same"`
|
||||
// Unread says the loaded list could not be read, and why: nothing is compared then.
|
||||
Unread string `json:"unread,omitempty"`
|
||||
}
|
||||
|
||||
// Equal says the two lists agree in every user, but those both agree are gone.
|
||||
func (c Comparison) Equal() bool {
|
||||
return c.Unread == "" && len(c.Added) == 0 && len(c.Differ) == 0 && len(c.Dropped) == len(c.Gone)
|
||||
}
|
||||
|
||||
// Differing names every user that keeps the lists apart: added, dropped and not gone, or granted otherwise.
|
||||
func (c Comparison) Differing() []string {
|
||||
var out []string
|
||||
out = append(out, c.Added...)
|
||||
for _, d := range c.Dropped {
|
||||
if !slices.Contains(c.Gone, d) {
|
||||
out = append(out, d)
|
||||
}
|
||||
}
|
||||
for _, d := range c.Differ {
|
||||
out = append(out, d.User)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// Compare compares the shadow list with the loaded one. placed says whether placement still names a loaded user's
|
||||
// machine and module; nil names every one. Pure.
|
||||
func Compare(shadow, loaded []User, placed func(user string) bool) Comparison {
|
||||
c := Comparison{Shadow: len(shadow), Loaded: len(loaded), Added: []string{}, Dropped: []string{}, Differ: []UserDiff{}}
|
||||
have := map[string]User{}
|
||||
for _, u := range loaded {
|
||||
have[u.User] = u
|
||||
}
|
||||
want := map[string]bool{}
|
||||
for _, s := range shadow {
|
||||
want[s.User] = true
|
||||
l, ok := have[s.User]
|
||||
if !ok {
|
||||
c.Added = append(c.Added, s.User)
|
||||
continue
|
||||
}
|
||||
d := UserDiff{User: s.User, PublishAdded: minus(s.Publish, l.Publish), PublishDropped: minus(l.Publish, s.Publish),
|
||||
SubscribeAdded: minus(s.Subscribe, l.Subscribe), SubscribeDropped: minus(l.Subscribe, s.Subscribe)}
|
||||
if s.AllowResponses != l.AllowResponses {
|
||||
d.Responses = fmt.Sprintf("shadow %v, loaded %v", s.AllowResponses, l.AllowResponses)
|
||||
}
|
||||
if len(d.PublishAdded)+len(d.PublishDropped)+len(d.SubscribeAdded)+len(d.SubscribeDropped) == 0 && d.Responses == "" {
|
||||
c.Same++
|
||||
continue
|
||||
}
|
||||
c.Differ = append(c.Differ, d)
|
||||
}
|
||||
for _, l := range loaded {
|
||||
if !want[l.User] {
|
||||
c.Dropped = append(c.Dropped, l.User)
|
||||
if placed != nil && !placed(l.User) {
|
||||
c.Gone = append(c.Gone, l.User)
|
||||
}
|
||||
}
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func minus(a, b []string) []string {
|
||||
var out []string
|
||||
for _, x := range a {
|
||||
if !slices.Contains(b, x) {
|
||||
out = append(out, x)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Placed says whether placement still names a user's machine and module: a machine's users (`node.<n>`,
|
||||
// `provisioner.<n>`) while the machine is seen, a module's or process's (`<n>.<module>[.<process>]`) while the module is
|
||||
// assigned there. The controller, enrolment users and persons are never judged gone from here.
|
||||
func Placed(in Inputs) func(user string) bool {
|
||||
return func(user string) bool {
|
||||
parts := strings.Split(user, ".")
|
||||
switch {
|
||||
case len(parts) < 2:
|
||||
return true
|
||||
case parts[0] == "node" || parts[0] == "provisioner":
|
||||
return slices.Contains(in.Nodes, parts[1])
|
||||
case parts[0] == "person" || parts[0] == "enrol":
|
||||
return true
|
||||
}
|
||||
return slices.Contains(in.Nodes, parts[0]) && slices.Contains(in.Assigned[parts[0]], parts[1])
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
{
|
||||
"module": "mesh-grants",
|
||||
"version": "1",
|
||||
"slug": "grants",
|
||||
"reads": [
|
||||
"mesh-config.resolved",
|
||||
"mesh-controller.assignments",
|
||||
"mesh-controller.grants",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"node-engine.have"
|
||||
],
|
||||
"invokes": [
|
||||
"nats.nats_streams",
|
||||
"nats.nats_user_can",
|
||||
"nats.nats_users"
|
||||
],
|
||||
"emits": [
|
||||
"membership-changed",
|
||||
"user-list-changed"
|
||||
],
|
||||
"tools": [
|
||||
"grants_grants",
|
||||
"grants_membership",
|
||||
"grants_user_list",
|
||||
"grants_compose_now",
|
||||
"grants_assert",
|
||||
"grants_take_over",
|
||||
"grants_hand_back"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "assertions", "history": 5 },
|
||||
{ "name": "memberships", "history": 5 },
|
||||
{ "name": "takeover", "history": 20 },
|
||||
{ "name": "user-list", "history": 20 }
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
},
|
||||
"secrets-owner": "mesh-grants",
|
||||
"resources": [
|
||||
{
|
||||
"id": "account",
|
||||
"type": "user",
|
||||
"name": "mesh-grants",
|
||||
"shell": "/usr/bin/nologin",
|
||||
"home": "/var/lib/mesh-grants"
|
||||
},
|
||||
{
|
||||
"id": "state",
|
||||
"type": "directory",
|
||||
"mode": "0700",
|
||||
"owner": "mesh-grants",
|
||||
"place": ".",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "grants",
|
||||
"type": "process",
|
||||
"name": "mesh-grants",
|
||||
"artifact": "program",
|
||||
"run": [
|
||||
"./mesh-grants"
|
||||
],
|
||||
"env": {
|
||||
"MESH_BROKER_FILE": "${dir:state}/broker"
|
||||
},
|
||||
"user": "mesh-grants",
|
||||
"health": {
|
||||
"kind": "unit"
|
||||
}
|
||||
}
|
||||
],
|
||||
"build": {
|
||||
"artifacts": [
|
||||
{
|
||||
"name": "program",
|
||||
"kind": "bundle",
|
||||
"language": "go",
|
||||
"system": "arch",
|
||||
"from": "cmd/mesh-grants",
|
||||
"binary": "mesh-grants",
|
||||
"loads": [
|
||||
"mesh-grants"
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user