jschoubben
fe7d4504bf
The controller ended by its unassignment and its lease key expired, so the gate on a handed lease kept retirement, rotation and node-login rotation off for good (issue 1126, A3). Drop the gate, the lease watch and the manifest's read of mesh-controller.lease; vault_status no longer reports a lease. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
422 lines
12 KiB
Go
422 lines
12 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
// The vault whole against a real bus (novox/hq ADR 0561 §1 to §4, issue 774's acceptance 12): the built binary, played
|
|
// over its real stdin and stdout as the node's tool runtime plays it (tools/call with the caller in _meta; mesh/state.*
|
|
// and mesh/publish answered from a real nats-server's key-value buckets and subjects), its store in a directory of its
|
|
// own. A want written to the grants part's state is minted with no lease asked (issue 774: the vault is
|
|
// the one issuer): the hash is in bus-accounts, own answers the value sealed to the node, the value is the controller's
|
|
// shape and its hash's password, account-issued is said; an adopted user is never minted.
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
"golang.org/x/crypto/bcrypt"
|
|
"golang.org/x/crypto/nacl/box"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
|
"git.novox.be/novox/mesh-sdk/go/seal"
|
|
"git.novox.be/novox/mesh-sdk/go/testbus"
|
|
)
|
|
|
|
// wire is one line between the runtime and the bundle (the SDK's stdio message).
|
|
type wire struct {
|
|
ID json.RawMessage `json:"id,omitempty"`
|
|
Method string `json:"method,omitempty"`
|
|
Params json.RawMessage `json:"params,omitempty"`
|
|
Result json.RawMessage `json:"result,omitempty"`
|
|
Error *struct {
|
|
Code int `json:"code"`
|
|
Message string `json:"message"`
|
|
} `json:"error,omitempty"`
|
|
}
|
|
|
|
// player is the node's tool runtime as far as the vault needs it, over a real bus.
|
|
type player struct {
|
|
t *testing.T
|
|
ctx context.Context
|
|
nc *nats.Conn
|
|
js jetstream.JetStream
|
|
|
|
mu sync.Mutex
|
|
in io.Writer
|
|
next int
|
|
pending map[string]chan wire
|
|
said []string
|
|
// ask answers a mesh/ask (a call of another module's tool); nil answers none, as a call with no responder.
|
|
ask func(key string, body json.RawMessage) (any, error)
|
|
}
|
|
|
|
// bucketOf is the bucket a state names: the vault's own by its local name, another's as `<owner>.<name>`.
|
|
func bucketOf(state string) string {
|
|
if owner, name, ok := strings.Cut(state, "."); ok {
|
|
return owner + "_" + name
|
|
}
|
|
return "mesh-vault_" + state
|
|
}
|
|
|
|
func (p *player) send(m map[string]any) {
|
|
raw, _ := json.Marshal(m)
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
_, _ = p.in.Write(append(raw, '\n'))
|
|
}
|
|
|
|
func (p *player) answer(id json.RawMessage, result any, err error) {
|
|
if err != nil {
|
|
p.send(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
|
|
return
|
|
}
|
|
p.send(map[string]any{"id": id, "result": result})
|
|
}
|
|
|
|
func (p *player) kv(state string) (jetstream.KeyValue, error) {
|
|
return p.js.KeyValue(p.ctx, bucketOf(state))
|
|
}
|
|
|
|
// serve answers what the bundle asks, and hands the answers to what the test asked it.
|
|
func (p *player) serve(out io.Reader) {
|
|
scanner := bufio.NewScanner(out)
|
|
scanner.Buffer(make([]byte, 1<<20), 1<<24)
|
|
for scanner.Scan() {
|
|
var m wire
|
|
if json.Unmarshal(scanner.Bytes(), &m) != nil {
|
|
continue
|
|
}
|
|
if m.Method == "" {
|
|
var key string
|
|
_ = json.Unmarshal(m.ID, &key)
|
|
p.mu.Lock()
|
|
ch := p.pending[key]
|
|
delete(p.pending, key)
|
|
p.mu.Unlock()
|
|
if ch != nil {
|
|
ch <- m
|
|
}
|
|
continue
|
|
}
|
|
go p.handle(m)
|
|
}
|
|
}
|
|
|
|
func (p *player) handle(m wire) {
|
|
var a struct {
|
|
State string `json:"state"`
|
|
Key string `json:"key"`
|
|
Value json.RawMessage `json:"value"`
|
|
Body json.RawMessage `json:"body"`
|
|
}
|
|
_ = json.Unmarshal(m.Params, &a)
|
|
switch m.Method {
|
|
case "mesh/state.get":
|
|
kv, err := p.kv(a.State)
|
|
if err != nil {
|
|
p.answer(m.ID, nil, err)
|
|
return
|
|
}
|
|
e, err := kv.Get(p.ctx, a.Key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
p.answer(m.ID, nil, nil)
|
|
return
|
|
}
|
|
if err != nil {
|
|
p.answer(m.ID, nil, err)
|
|
return
|
|
}
|
|
p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil)
|
|
case "mesh/state.put":
|
|
kv, err := p.kv(a.State)
|
|
if err == nil {
|
|
var rev uint64
|
|
rev, err = kv.Put(p.ctx, a.Key, a.Value)
|
|
if err == nil {
|
|
p.answer(m.ID, map[string]any{"revision": rev}, nil)
|
|
return
|
|
}
|
|
}
|
|
p.answer(m.ID, nil, err)
|
|
case "mesh/state.keys":
|
|
kv, err := p.kv(a.State)
|
|
if err != nil {
|
|
p.answer(m.ID, nil, err)
|
|
return
|
|
}
|
|
keys, err := kv.Keys(p.ctx)
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
keys, err = []string{}, nil
|
|
}
|
|
p.answer(m.ID, keys, err)
|
|
case "mesh/state.delete":
|
|
kv, err := p.kv(a.State)
|
|
if err == nil {
|
|
err = kv.Delete(p.ctx, a.Key)
|
|
}
|
|
p.answer(m.ID, map[string]any{}, err)
|
|
case "mesh/state.watch":
|
|
kv, err := p.kv(a.State)
|
|
if err != nil {
|
|
p.answer(m.ID, nil, err)
|
|
return
|
|
}
|
|
key := strings.ReplaceAll(a.Key, "**", ">")
|
|
if key == "" {
|
|
key = ">"
|
|
}
|
|
w, err := kv.Watch(p.ctx, key)
|
|
if err != nil {
|
|
p.answer(m.ID, nil, err)
|
|
return
|
|
}
|
|
var watch string
|
|
_ = json.Unmarshal(m.ID, &watch)
|
|
current := true
|
|
for e := range w.Updates() {
|
|
if e == nil {
|
|
current = false
|
|
p.answer(m.ID, map[string]any{}, nil)
|
|
continue
|
|
}
|
|
op := "put"
|
|
if e.Operation() != jetstream.KeyValuePut {
|
|
op = "delete"
|
|
}
|
|
change := map[string]any{"state": a.State, "key": e.Key(), "op": op, "revision": e.Revision(), "current": current}
|
|
if op == "put" {
|
|
change["value"] = json.RawMessage(e.Value())
|
|
}
|
|
p.mu.Lock()
|
|
p.next++
|
|
id := fmt.Sprintf("w-%d", p.next)
|
|
p.mu.Unlock()
|
|
p.send(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watch, "change": change}})
|
|
}
|
|
case "mesh/publish":
|
|
var e struct {
|
|
Key string `json:"key"`
|
|
Body json.RawMessage `json:"body"`
|
|
}
|
|
_ = json.Unmarshal(m.Params, &e)
|
|
p.mu.Lock()
|
|
p.said = append(p.said, e.Key+" "+string(e.Body))
|
|
p.mu.Unlock()
|
|
p.answer(m.ID, map[string]any{}, p.nc.Publish("mesh.mod.mesh-vault.event."+e.Key, e.Body))
|
|
case "mesh/ask":
|
|
var ask struct {
|
|
Key string `json:"key"`
|
|
Body json.RawMessage `json:"body"`
|
|
}
|
|
_ = json.Unmarshal(m.Params, &ask)
|
|
if p.ask != nil {
|
|
result, err := p.ask(ask.Key, ask.Body)
|
|
p.answer(m.ID, result, err)
|
|
return
|
|
}
|
|
p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
|
|
default:
|
|
// the rest: nothing answers on this bus, as a call with no responder.
|
|
p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
|
|
}
|
|
}
|
|
|
|
// call is a tools/call of the vault as caller, answered as the tool's JSON.
|
|
func (p *player) call(name, caller string, args map[string]any) (json.RawMessage, error) {
|
|
p.mu.Lock()
|
|
p.next++
|
|
id := fmt.Sprintf("c-%d", p.next)
|
|
ch := make(chan wire, 1)
|
|
p.pending[id] = ch
|
|
p.mu.Unlock()
|
|
p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": name, "arguments": args,
|
|
"_meta": map[string]any{"mesh/caller": caller}}})
|
|
select {
|
|
case m := <-ch:
|
|
var r struct {
|
|
Content []struct {
|
|
Text string `json:"text"`
|
|
} `json:"content"`
|
|
IsError bool `json:"isError"`
|
|
}
|
|
if m.Error != nil {
|
|
return nil, errors.New(m.Error.Message)
|
|
}
|
|
if err := json.Unmarshal(m.Result, &r); err != nil || len(r.Content) != 1 {
|
|
return nil, fmt.Errorf("%s answered %s", name, m.Result)
|
|
}
|
|
if r.IsError {
|
|
return nil, errors.New(r.Content[0].Text)
|
|
}
|
|
return json.RawMessage(r.Content[0].Text), nil
|
|
case <-time.After(30 * time.Second):
|
|
return nil, fmt.Errorf("%s did not answer in 30s", name)
|
|
}
|
|
}
|
|
|
|
func (p *player) put(t *testing.T, state, key string, v any) {
|
|
t.Helper()
|
|
kv, err := p.kv(state)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
raw, _ := json.Marshal(v)
|
|
if _, err := kv.Put(p.ctx, key, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func (p *player) hashOf(t *testing.T, user string) *AccountHash {
|
|
t.Helper()
|
|
kv, err := p.kv(BusAccountsState)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
e, err := kv.Get(p.ctx, user)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var a AccountHash
|
|
if err := json.Unmarshal(e.Value(), &a); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return &a
|
|
}
|
|
|
|
func eventually(t *testing.T, what string, ok func() bool) {
|
|
t.Helper()
|
|
for end := time.Now().Add(30 * time.Second); time.Now().Before(end); time.Sleep(50 * time.Millisecond) {
|
|
if ok() {
|
|
return
|
|
}
|
|
}
|
|
t.Fatalf("never: %s", what)
|
|
}
|
|
|
|
// 774/AC12.
|
|
func TestTheVaultWholeMintsOnTheBusWhoeverHoldsTheLease(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
t.Cleanup(cancel)
|
|
dir := t.TempDir()
|
|
bin := filepath.Join(dir, "mesh-vault")
|
|
build := exec.Command("go", "build", "-o", bin, ".")
|
|
build.Stderr = os.Stderr
|
|
if err := build.Run(); err != nil {
|
|
t.Fatalf("the vault does not build: %v", err)
|
|
}
|
|
nc, err := nats.Connect(testbus.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(nc.Close)
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, b := range []string{"mesh-vault_bus-accounts", "mesh-grants_accounts",
|
|
"mesh-provisioner_plans", "node-engine_have", "mesh-delivery-assigner_placement"} {
|
|
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 2}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
// The operator key, as the mesh exports it to the vault's root.
|
|
opPub, _, err := box.GenerateKey(rand.Reader)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
root := filepath.Join(dir, "root")
|
|
_ = os.MkdirAll(root, 0o700)
|
|
export, _ := json.Marshal(map[string]string{"operator-key": base64.StdEncoding.EncodeToString(opPub[:])})
|
|
if err := os.WriteFile(filepath.Join(root, "export.json"), export, 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
cmd := exec.CommandContext(ctx, bin)
|
|
cmd.Env = append(os.Environ(), "MESH_VAULT_STORE="+filepath.Join(dir, "store"), "MESH_VAULT_STATE="+filepath.Join(dir, "state"),
|
|
"MESH_VAULT_ROOT="+root, "MESH_VAULT_LEDGER="+filepath.Join(dir, "ledger"), "MESH_RECEIVES="+filepath.Join(dir, "mesh.json"),
|
|
"MESH_NODE=novox", "MESH_MODULE=mesh-vault")
|
|
stdin, _ := cmd.StdinPipe()
|
|
stdout, _ := cmd.StdoutPipe()
|
|
cmd.Stderr = os.Stderr
|
|
if err := cmd.Start(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() })
|
|
p := &player{t: t, ctx: ctx, nc: nc, js: js, in: stdin, pending: map[string]chan wire{}}
|
|
go p.serve(stdout)
|
|
|
|
status := func() MintState {
|
|
raw, err := p.call("vault_status", "person.jochen", nil)
|
|
if err != nil {
|
|
return MintState{}
|
|
}
|
|
var s struct {
|
|
Minting MintState `json:"minting"`
|
|
}
|
|
_ = json.Unmarshal(raw, &s)
|
|
return s.Minting
|
|
}
|
|
// No lease key: the want is minted all the same (issue 1126).
|
|
x, y := aWant("mod-x"), aWant("mod-y")
|
|
p.put(t, "mesh-vault.bus-accounts", y.User, AccountHash{User: y.User, Hash: "$2a$04$adoptedadoptedadoptedadoptedadoptedadoptedadopte", Adopted: true})
|
|
p.put(t, "mesh-grants.accounts", x.User, x)
|
|
p.put(t, "mesh-grants.accounts", y.User, y)
|
|
eventually(t, "the account minted", func() bool { return p.hashOf(t, x.User) != nil })
|
|
if st := status(); !st.On || len(st.HeldBack) != 0 {
|
|
t.Errorf("vault_status says %+v", st)
|
|
}
|
|
nodePub, nodePriv := aSealingKey(t)
|
|
if _, err := p.call(Seat+".node-key", "node.ace", map[string]any{"key": nodePub}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
raw, err := p.call(Seat+".own", "node.ace", map[string]any{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var own OwnAnswer
|
|
if err := json.Unmarshal(raw, &own); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(own.Own) != 1 || own.Own[0].Module != "mod-x" || own.Own[0].Name != "broker" || own.Own[0].Origin != OriginMade {
|
|
t.Fatalf("own answers %+v", own.Own)
|
|
}
|
|
value, err := seal.Open(own.Own[0].Sealed, nodePriv)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var held struct{ Password string }
|
|
_ = json.Unmarshal([]byte(value), &held)
|
|
shape, _ := buscompose.HeldCredential(x, held.Password)
|
|
if value != string(shape) {
|
|
t.Errorf("the minted value %q is not the controller's shape %q", value, shape)
|
|
}
|
|
if h := p.hashOf(t, x.User); bcrypt.CompareHashAndPassword([]byte(h.Hash), []byte(held.Password)) != nil || h.Adopted || h.Generation != 1 {
|
|
t.Errorf("bus-accounts %s holds %+v, not the minted password's hash", x.User, h)
|
|
}
|
|
if h := p.hashOf(t, y.User); h == nil || !h.Adopted {
|
|
t.Errorf("the adopted %s was replaced: %+v", y.User, h)
|
|
}
|
|
p.mu.Lock()
|
|
said := strings.Join(p.said, "\n")
|
|
p.mu.Unlock()
|
|
if !strings.Contains(said, AccountIssuedEvent+` {"`) || !strings.Contains(said, `"user":"ace.mod-x"`) || strings.Contains(said, held.Password) {
|
|
t.Errorf("the vault said: %s", said)
|
|
}
|
|
}
|