mesh-controller/internal/broker/have_bucket_live_test.go

//go:build integration

package broker

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

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"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"
)

// **Each node's credential writes its own have keys and no other** (novox/hq ADR 0429 §1, issue 1136, 1136/AC4): on a
// real server whose users are composed as the controller composes them, node alpha's put of alpha.zsh is taken, its
// put of beta.zsh is refused by the server, a module reading node-engine.have is refused any put and its watch sees
// alpha.zsh.
func TestNatsANodePutsOnlyItsOwnHaveKeys(t *testing.T) { // 1136/AC4
	hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
	if err != nil {
		t.Fatal(err)
	}
	users, err := Users(Records{Nodes: []string{"alpha", "beta"}, Assigned: map[string][]Declared{
		"alpha": {{Module: RuntimeModule}, {Module: "mesh-delivery-view", Reads: []string{"node-engine.have"}}},
		"beta":  {{Module: RuntimeModule}},
	}})
	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(), 30*time.Second)
	defer cancel()
	connect := func(user string) (*nats.Conn, chan error) {
		t.Helper()
		refused := make(chan error, 8)
		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
			nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
		if err != nil {
			t.Fatalf("%s cannot connect: %v", user, err)
		}
		t.Cleanup(nc.Close)
		return nc, refused
	}
	controller, _ := connect(ControllerName)
	cjs, err := jetstream.New(controller)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := cjs.CreateOrUpdateKeyValue(ctx, HaveBucketConfig()); err != nil {
		t.Fatalf("the controller cannot make %s: %v", delivery.HaveBucket, err)
	}

	alpha, _ := connect("node.alpha")
	ajs, err := jetstream.New(alpha)
	if err != nil {
		t.Fatal(err)
	}
	akv, err := ajs.KeyValue(ctx, delivery.HaveBucket)
	if err != nil {
		t.Fatalf("node alpha cannot bind %s: %v", delivery.HaveBucket, err)
	}
	if _, err := akv.Put(ctx, "alpha.zsh", []byte(`{"v":1}`)); err != nil {
		t.Fatalf("node alpha cannot put alpha.zsh: %v", err)
	}
	if e, err := akv.Get(ctx, "alpha.zsh"); err != nil || string(e.Value()) != `{"v":1}` {
		t.Errorf("node alpha reads %v (%v) of its own key", e, err)
	}

	reader, readerRefused := connect("alpha.mesh-delivery-view")
	for _, other := range []struct {
		name string
		nc   *nats.Conn
		ch   chan error
		key  string
	}{{"node alpha putting beta.zsh", alpha, nil, "beta.zsh"},
		{"the reading module putting alpha.zsh", reader, readerRefused, "alpha.zsh"}} {
		ch := other.ch
		if ch == nil {
			// alpha's own refusals arrive on its own handler: reconnect with one that is watched.
			var nc *nats.Conn
			nc, ch = connect("node.alpha")
			other.nc = nc
		}
		if err := other.nc.Publish("$KV."+delivery.HaveBucket+"."+other.key, []byte(`{"v":99}`)); err != nil {
			t.Fatal(err)
		}
		_ = other.nc.Flush()
		select {
		case err := <-ch:
			if !errors.Is(err, nats.ErrPermissionViolation) {
				t.Errorf("%s: the server said %v, want a permissions violation", other.name, err)
			}
		case <-time.After(3 * time.Second):
			t.Errorf("the server did not refuse %s", other.name)
		}
	}

	// The reader watches and sees alpha.zsh, and beta.zsh was never taken.
	rjs, err := jetstream.New(reader)
	if err != nil {
		t.Fatal(err)
	}
	rkv, err := rjs.KeyValue(ctx, delivery.HaveBucket)
	if err != nil {
		t.Fatalf("a module reading node-engine.have cannot bind it: %v", err)
	}
	w, err := rkv.WatchAll(ctx)
	if err != nil {
		t.Fatalf("the reading module cannot watch: %v", err)
	}
	defer w.Stop()
	seen := map[string]string{}
	for e := range w.Updates() {
		if e == nil {
			break
		}
		seen[e.Key()] = string(e.Value())
	}
	if seen["alpha.zsh"] != `{"v":1}` || len(seen) != 1 {
		t.Errorf("the reading module's watch sees %v, want alpha.zsh alone", seen)
	}
}