The bus node follows the grants part's sealed user list, and only the grants part reads the vault's hashes (ADR 0547 clauses 7, 9; issue 1945)
The grants part writes its list as an object of its own bucket, sealed to the bus node's key, and names it in user-list.sealed; the node-engine of the node holding mesh-broker is granted that one key and an object by name, the grants part its bucket (the controller's copy carries the same grants, byte for byte, golden regenerated there). A state may name its readers (announce, NamedReaders): mesh-vault.bus-accounts is read by mesh-grants alone.
This commit is contained in:
@@ -0,0 +1,206 @@
|
||||
//go:build integration
|
||||
|
||||
package buscompose
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/seal"
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// Defends novox/hq ADR 0547 clause 7 on a real server (issue 1945): with the list this composer writes, the grants part
|
||||
// creates its object bucket and its state, puts the sealed list and names it in `user-list.sealed`; the node-engine of
|
||||
// the node holding mesh-broker watches that one key, reads the object by its name and opens it with its key, and may
|
||||
// neither list the bucket, nor read another key of the state, nor write; another node reads nothing of either.
|
||||
func TestTheBusNodeFollowsTheGrantsPartsSealedListUnderTheComposedGrants(t *testing.T) {
|
||||
records := Records{
|
||||
Nodes: []string{"novox", "ace"},
|
||||
Assigned: map[string][]Declared{
|
||||
"novox": {{Module: "nats", SnapshotsTheBus: true},
|
||||
{Module: GrantsModule, State: []Bucket{{Module: GrantsModule, Name: "user-list"}}}},
|
||||
"ace": {{Module: "plex"}},
|
||||
},
|
||||
}
|
||||
users, err := Users(records)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := range users {
|
||||
users[i].PasswordHash = "pw-" + users[i].Username()
|
||||
}
|
||||
accounts, err := ComposeAccounts(users)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dir := t.TempDir()
|
||||
conf := filepath.Join(dir, "server.conf")
|
||||
if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: \""+dir+"\" }\n"+accounts), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts, err := server.ProcessConfigFile(conf)
|
||||
if err != nil {
|
||||
t.Fatalf("the composed list is not the server's: %v", err)
|
||||
}
|
||||
opts.NoLog, opts.NoSigs = true, true
|
||||
s, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go s.Start()
|
||||
if !s.ReadyForConnections(30 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(s.Shutdown)
|
||||
dial := func(user string) *nats.Conn {
|
||||
nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw-"+user), nats.CustomInboxPrefix("_INBOX."+user))
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", user, err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
return nc
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// The grants part: its state and its object bucket, the sealed list put and named.
|
||||
grants, err := jetstream.New(dial("novox." + GrantsModule))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
kv, err := grants.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: GrantsUserListBucket})
|
||||
if err != nil {
|
||||
t.Fatalf("the grants part's state: %v", err)
|
||||
}
|
||||
objects, err := grants.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: GrantsUserListObjects})
|
||||
if err != nil {
|
||||
t.Fatalf("the grants part's object bucket: %v", err)
|
||||
}
|
||||
node, err := seal.GenerateKeyPair()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
list := []byte(accounts)
|
||||
digest := delivery.UserListDigest(list)
|
||||
object, err := delivery.SealUserList(list, node.PublicKey)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
name := delivery.UserListObjectName(digest)
|
||||
for range 2 { // put twice: a second put of a name purges its old chunks
|
||||
if _, err := objects.PutBytes(ctx, name, object); err != nil {
|
||||
t.Fatalf("the grants part put its object: %v", err)
|
||||
}
|
||||
}
|
||||
value, err := delivery.EncodeGrantsUserList(delivery.GrantsUserList{Follow: delivery.FollowGrants, Digest: digest,
|
||||
Object: name, SealedTo: delivery.SealingKeyID(node.PublicKey), Node: "novox", Revision: 1, Users: len(users)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := kv.Put(ctx, GrantsUserListKey, value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := kv.Put(ctx, "list", []byte(`{"v":1}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// The bus node's engine: the key watched, the object by name (as mesh-host's link reads a declaration's file).
|
||||
busNode := dial("node.novox")
|
||||
js, err := busNode.JetStream(nats.MaxWait(5 * time.Second))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keys, err := js.KeyValue(GrantsUserListBucket)
|
||||
if err != nil {
|
||||
t.Fatalf("the bus node binds the grants part's state: %v", err)
|
||||
}
|
||||
w, err := keys.Watch(GrantsUserListKey)
|
||||
if err != nil {
|
||||
t.Fatalf("the bus node watches the key: %v", err)
|
||||
}
|
||||
defer w.Stop()
|
||||
var got delivery.GrantsUserList
|
||||
select {
|
||||
case e := <-w.Updates():
|
||||
if e == nil {
|
||||
t.Fatal("the watch found no value")
|
||||
}
|
||||
if got, err = delivery.DecodeGrantsUserList(e.Value()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
case <-ctx.Done():
|
||||
t.Fatal("the watch said nothing")
|
||||
}
|
||||
read := func(js nats.JetStreamContext, name string) ([]byte, error) {
|
||||
meta, err := js.GetLastMsg(GrantsUserListObjectStream, "$O."+GrantsUserListObjects+".M."+
|
||||
base64.URLEncoding.EncodeToString([]byte(name)), nats.DirectGet())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var info struct {
|
||||
NUID string `json:"nuid"`
|
||||
Size int `json:"size"`
|
||||
}
|
||||
if err := json.Unmarshal(meta.Data, &info); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sub, err := js.SubscribeSync("$O."+GrantsUserListObjects+".C."+info.NUID, nats.OrderedConsumer(),
|
||||
nats.BindStream(GrantsUserListObjectStream))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer sub.Unsubscribe()
|
||||
var body []byte
|
||||
for len(body) < info.Size {
|
||||
m, err := sub.NextMsg(5 * time.Second)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
body = append(body, m.Data...)
|
||||
}
|
||||
return body, nil
|
||||
}
|
||||
body, err := read(js, got.Object)
|
||||
if err != nil {
|
||||
t.Fatalf("the bus node reads the object by its name: %v", err)
|
||||
}
|
||||
opened, err := delivery.OpenUserList(body, got.Digest, func(b seal.Box) ([]byte, error) {
|
||||
s, err := seal.Open(b, node.PrivateKey)
|
||||
return []byte(s), err
|
||||
})
|
||||
if err != nil || !bytes.Equal(opened, list) {
|
||||
t.Fatalf("the bus node opened %d bytes, %v", len(opened), err)
|
||||
}
|
||||
|
||||
// And nothing beyond: no other key, no list of the bucket, no write.
|
||||
if _, err := keys.Get("list"); err == nil {
|
||||
t.Error("the bus node read another key of the grants part's state")
|
||||
}
|
||||
if _, err := js.StreamInfo(GrantsUserListObjectStream); err == nil {
|
||||
t.Error("the bus node read the object stream's info, which lists every name")
|
||||
}
|
||||
if _, err := js.Publish("$O."+GrantsUserListObjects+".M.x", []byte("x")); err == nil {
|
||||
t.Error("the bus node wrote the grants part's bucket")
|
||||
}
|
||||
// Another node reads nothing of it.
|
||||
other, err := dial("node.ace").JetStream(nats.MaxWait(2 * time.Second))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := read(other, got.Object); err == nil {
|
||||
t.Error("another node read the object")
|
||||
}
|
||||
if _, err := other.GetLastMsg("KV_"+GrantsUserListBucket, "$KV."+GrantsUserListBucket+"."+GrantsUserListKey, nats.DirectGet()); err == nil {
|
||||
t.Errorf("another node read the key: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user