mesh-controller/internal/broker/system_exports_live_test.go

//go:build integration

package broker

import (
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

// **The system account's exports reach MESH on a reload, with no restart** (novox/hq ADR 0431 §3, issue 1147,
// 1147/AC3): the embedded pinned server starts with today's composed accounts (no `$SYS`, no import); the new file is
// written and the server reloaded; the reload succeeds; a user with the watchman's grants hears another user's connect
// and disconnect and gets an answer to `mesh.bus.connz` that lists the open connections; another module's user is
// refused a subscribe on `mesh.bus.connect`.
func TestNatsTheSystemAccountsExportsReachMeshOnAReload(t *testing.T) { // 1147/AC3
	hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
	if err != nil {
		t.Fatal(err)
	}
	users, err := Users(Records{Nodes: []string{"one"}, Assigned: map[string][]Declared{
		"one": {{Module: RuntimeModule}, {Module: WatchmanModule}, {Module: "gitea"}, {Module: "shop"}},
	}})
	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)
	}
	// Today's file: the same accounts without the system account's exports and MESH's imports.
	today := strings.Replace(strings.Replace(accounts, meshSystemImports, "", 1), systemAccount, "", 1)
	if today == accounts || strings.Contains(today, "$SYS") {
		t.Fatal("today's composed file could not be derived from the new one")
	}
	dir := t.TempDir()
	conf := filepath.Join(dir, "bus.conf")
	write := func(accounts string) {
		t.Helper()
		if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+`"`+dir+`/js"`+" }\n"+accounts), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	write(today)
	opts, err := server.ProcessConfigFile(conf)
	if err != nil {
		t.Fatalf("today's 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)

	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
	}
	// A connection open across the reload: the reload must not drop it.
	shop, _ := connect("one.shop")

	write(accounts)
	if err := s.Reload(); err != nil {
		t.Fatalf("the server does not take the system account's exports on a reload: %v", err)
	}
	if !shop.IsConnected() {
		t.Error("the reload dropped a connection that was open")
	}

	watchman, _ := connect("one." + WatchmanModule)
	connects := make(chan *nats.Msg, 16)
	disconnects := make(chan *nats.Msg, 16)
	if _, err := watchman.ChanSubscribe(BusConnectSubject, connects); err != nil {
		t.Fatal(err)
	}
	if _, err := watchman.ChanSubscribe(BusDisconnectSubject, disconnects); err != nil {
		t.Fatal(err)
	}
	if err := watchman.Flush(); err != nil {
		t.Fatal(err)
	}
	// Another user connects and disconnects: the watchman hears both, and the events name that user.
	gitea, _ := connect("one.gitea")
	// The server says each connection's own events too, the watchman's among them, and may say its connect after the
	// subscription is made (issue 1169): the event looked for is the one naming the other user, whatever comes before it.
	hears := func(ch chan *nats.Msg, what string) {
		t.Helper()
		bound := time.After(5 * time.Second)
		var others []string
		for {
			select {
			case m := <-ch:
				if strings.Contains(string(m.Data), "one.gitea") {
					return
				}
				others = append(others, string(m.Data))
			case <-bound:
				t.Errorf("the watchman did not hear another user's %s; it heard only %q", what, others)
				return
			}
		}
	}
	hears(connects, "connect")
	// The answer to the connections request lists the open connections, the shop's among them. The server names the
	// account from the subject the export clamps (MESH), but leaves the user's name out unless the request asks for it
	// with "auth": true (monitor.go ConnzOptions.Username), so the watchman asks so.
	reply, err := watchman.Request(BusConnzSubject, []byte(`{"auth":true}`), 5*time.Second)
	if err != nil {
		t.Fatalf("mesh.bus.connz is not answered: %v", err)
	}
	var answer struct {
		Data struct {
			Connections []struct {
				AuthorizedUser string `json:"authorized_user"`
			} `json:"connections"`
		} `json:"data"`
	}
	if err := json.Unmarshal(reply.Data, &answer); err != nil {
		t.Fatalf("the connections answer is not JSON: %v: %s", err, reply.Data)
	}
	var listed bool
	for _, c := range answer.Data.Connections {
		listed = listed || c.AuthorizedUser == "one.shop"
	}
	if !listed {
		t.Errorf("the connections answer does not list the shop's connection: %s", reply.Data)
	}
	gitea.Close()
	hears(disconnects, "disconnect")

	// Another module's user is refused a subscribe on the events and a publish on the request.
	other, refused := connect("one.shop")
	for _, act := range []func() error{
		func() error { _, err := other.SubscribeSync(BusConnectSubject); return err },
		func() error { _, err := other.SubscribeSync(BusDisconnectSubject); return err },
		func() error { return other.Publish(BusConnzSubject, []byte(`{}`)) },
	} {
		if err := act(); err != nil {
			t.Fatal(err)
		}
		_ = other.Flush()
		select {
		case err := <-refused:
			if !errors.Is(err, nats.ErrPermissionViolation) {
				t.Errorf("the server said %v, want a permissions violation", err)
			}
		case <-time.After(3 * time.Second):
			t.Error("the server did not refuse another module's user the bus's connection subjects")
		}
	}
}