mesh-controller/internal/broker/composed_live_test.go

//go:build integration

package broker

import (
	"bytes"
	"context"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"golang.org/x/crypto/bcrypt"
)

// 1945 (the review of A1, H1): on a bus running the user list this controller composes, the operator's account (a
// person who may invoke every tool) walks release 6's hand path end to end: it writes a node's want, that node's
// composition and its digest, and stores a file over 64 KiB in the node's object bucket and replaces it; a person with
// fewer tools may do none of it.
func TestNatsTheOperatorWritesTheHandPathOnTheComposedUserList(t *testing.T) {
	hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
	if err != nil {
		t.Fatal(err)
	}
	users, err := Users(Records{Nodes: []string{"ace"},
		People: map[string][]string{"jochen": {"*"}, "guest": {"mesh-controller.status"}}})
	if err != nil {
		t.Fatal(err)
	}
	for i := range users {
		users[i].PasswordHash = string(hash)
	}
	accounts, err := ComposeAccounts(users)
	if err != nil {
		t.Fatal(err)
	}
	conf := filepath.Join(t.TempDir(), "bus.conf")
	if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+`"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil {
		t.Fatal(err)
	}
	opts, err := server.ProcessConfigFile(conf)
	if err != nil {
		t.Fatalf("the composed accounts do not parse: %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(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(s.Shutdown)
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	connect := func(user string) jetstream.JetStream {
		t.Helper()
		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user))
		if err != nil {
			t.Fatalf("%s cannot connect: %v", user, err)
		}
		t.Cleanup(nc.Close)
		js, err := jetstream.New(nc)
		if err != nil {
			t.Fatal(err)
		}
		return js
	}

	// The buckets as the controller and the assigner make them.
	cjs := connect(ControllerName)
	if err := EnsureDeclarationFiles(ctx, cjs, "ace"); err != nil {
		t.Fatal(err)
	}
	if err := EnsureComposedDigests(ctx, cjs); err != nil {
		t.Fatal(err)
	}
	if _, err := cjs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "composed-ace", History: 5}); err != nil {
		t.Fatal(err)
	}
	// The assigner's bucket, made here by the controller's account, which may create any stream.
	if _, err := cjs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-delivery-assigner_wants", History: 20}); err != nil {
		t.Fatal(err)
	}

	short := func(c context.Context) (context.Context, context.CancelFunc) {
		return context.WithTimeout(c, 5*time.Second)
	}
	for _, who := range []struct {
		user string
		may  bool
	}{{"person.jochen", true}, {"person.guest", false}} {
		js := connect(who.user)
		try := func(what string, act func(context.Context) error) {
			t.Helper()
			c, stop := short(ctx)
			defer stop()
			err := act(c)
			if (err == nil) != who.may {
				t.Errorf("%s: %s: %v (may: %v)", who.user, what, err, who.may)
			}
		}
		try("write ace's want", func(c context.Context) error {
			kv, err := js.KeyValue(c, "mesh-delivery-assigner_wants")
			if err != nil {
				return err
			}
			_, err = kv.Put(c, "want.ace.mesh-controller", []byte(`{"v":1}`))
			return err
		})
		try("write ace's composition", func(c context.Context) error {
			kv, err := js.KeyValue(c, "composed-ace")
			if err != nil {
				return err
			}
			_, err = kv.Put(c, "ace.mesh-controller.aaaa", []byte(`{"v":1}`))
			return err
		})
		try("write its digest", func(c context.Context) error {
			kv, err := js.KeyValue(c, ControllerComposedBucket)
			if err != nil {
				return err
			}
			_, err = kv.Put(c, "ace.mesh-controller.aaaa", []byte(`{"v":1}`))
			return err
		})
		try("store and replace a file over 64 KiB", func(c context.Context) error {
			store, err := js.ObjectStore(c, "declaration-files-ace")
			if err != nil {
				return err
			}
			body := bytes.Repeat([]byte("x"), 200<<10)
			if _, err := store.PutBytes(c, "f00d", body); err != nil {
				return err
			}
			_, err = store.PutBytes(c, "f00d", append(body, 'y'))
			return err
		})
	}
}