mesh-sdk/go/buscompose/userlist_switch_integration_test.go

//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)
	}
	// Its state is made as the grants part makes a declared state, by its stream's name (CreateKeyValue would ask the
	// account's info, which no module is granted).
	if _, err := grants.CreateStream(ctx, jetstream.StreamConfig{Name: "KV_" + GrantsUserListBucket,
		Subjects: []string{"$KV." + GrantsUserListBucket + ".>"}, MaxMsgsPerSubject: 1, AllowDirect: true,
		AllowRollup: true, DenyDelete: true, Discard: jetstream.DiscardNew}); err != nil {
		t.Fatalf("the grants part's state: %v", err)
	}
	kv, err := grants.KeyValue(ctx, GrantsUserListBucket)
	if err != nil {
		t.Fatalf("the grants part binds its 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)
	}
}