mesh-tools/node-tools/internal/runtime/ledger_credentials_test.go

package runtime

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

	"git.novox.be/novox/mesh-sdk/go/buscalls"
	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/membership"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-tools/node-tools/internal/bus"
	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)

// 1966/AC22: given the secret ledgers holding m's credential, when m applies, then the append goes on m's connection
// (the runner's own account may not publish on m's ledger, as the grants part grants it); given it missing, apply is
// refused naming it; a change of the file is taken without a restart.
func TestALedgerIsWrittenOnlyOnItsModulesLedgerConnection(t *testing.T) {
	ledgerSubjects := []string{buscompose.LedgerHistorySubject("m", "records") + ".>", buscompose.LedgerSnapshotSubject("m", "records") + ".>"}
	users := []*server.User{
		{Username: "node-tools", Password: "pw", Permissions: &server.Permissions{
			Publish: &server.SubjectPermission{Allow: []string{">"}, Deny: ledgerSubjects}}},
		// As the grants part composes a ledger account (mesh-grants buscompose ledgerPermissions): publish on its
		// module's ledger histories and snapshots, subscribe its own inbox, nothing else.
		{Username: "ledger.anchor.m", Password: "pw", Permissions: &server.Permissions{
			Publish:   &server.SubjectPermission{Allow: ledgerSubjects},
			Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX.ledger.anchor.m.>"}}}},
		{Username: "admin", Password: "pw"},
	}
	s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
		NoLog: true, NoSigs: true, Users: users})
	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)
	url := s.ClientURL()

	admin, err := nats.Connect(url, nats.UserInfo("admin", "pw"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(admin.Close)
	js, _ := jetstream.New(admin)
	ctx := context.Background()
	for _, cfg := range []jetstream.StreamConfig{
		{Name: "LEDGER_m_records", Subjects: []string{"mesh.ledger.m.records.history.>"}},
		{Name: "SNAPSHOT_m_records", Subjects: []string{"mesh.ledger.m.records.snapshot.>"}, MaxMsgsPerSubject: 1},
	} {
		if _, err := js.CreateStream(ctx, cfg); err != nil {
			t.Fatal(err)
		}
	}
	for _, b := range []string{"m_records", buscompose.GrantsMembershipsBucket} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
			t.Fatal(err)
		}
	}
	mem := mt.MembershipOf("m", "anchor", false, nil)
	mem.Bindings = []membership.BusBinding{{Local: "records", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger,
		History: "LEDGER_m_records", Subject: "mesh.ledger.m.records.history", State: "m_records", Snapshot: "mesh.ledger.m.records.snapshot",
		Types: []catalogue.LedgerType{{Name: "decision", Operations: []catalogue.LedgerOperation{
			{Name: "title", Effect: catalogue.EffectSetFields, Fields: []string{"title"}}}}}}}
	raw, _ := json.Marshal(mem)
	var sdk membership.Membership
	_ = json.Unmarshal(raw, &sdk)
	grants, _ := js.KeyValue(ctx, buscompose.GrantsMembershipsBucket)
	if _, err := grants.Put(ctx, buscompose.MembershipKey("anchor", "m"), membership.EncodeGrantsMembership(sdk)); err != nil {
		t.Fatal(err)
	}

	conn, err := bus.Connect(bus.Credential{URL: url, Module: "node-tools", Node: "anchor", User: "node-tools", Password: "pw"})
	if err != nil {
		t.Fatal(err)
	}
	conn.Logf = func(string, ...any) {}
	t.Cleanup(conn.Close)
	conn.Follow("m")
	mt.Until(t, func() error {
		if conn.Membership("m") == nil {
			return os.ErrNotExist
		}
		return nil
	})

	// Not there yet, as on a machine whose vault has not made it: no ledger credentials, and the runner serves on.
	file := filepath.Join(t.TempDir(), "ledgers")
	creds := FollowLedgerCredentials(file, bus.Dial, t.Logf)
	t.Cleanup(creds.Stop)
	all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}}
	WithLedgerWriters(creds.Writer)(all)
	m := all.forModule("m").(*moduleBus)
	apply := func(instance string) error {
		_, err := m.BusState(buscalls.LedgerApply, json.RawMessage(`{"ledger":"records","instance":"`+instance+`","operation":"title","body":{"title":"T"}}`))
		return err
	}

	// Missing: refused, naming it and why.
	if err := apply("decision/1"); err == nil || !strings.Contains(err.Error(), "m's ledger credential is missing from the tool runner's own secret ledgers (the file "+file+" is not there yet)") {
		t.Fatalf("an apply with no credential: %v", err)
	}
	// Written: taken without a restart, on m's connection (the runner's own account is denied m's ledger subjects, so an
	// append that lands went on m's), whose account stamps the operation.
	cred, _ := json.Marshal(map[string]bus.Credential{"m": {URL: url, Node: "anchor", Module: "m", User: "ledger.anchor.m", Password: "pw"}})
	if err := os.WriteFile(file, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	mt.Until(t, func() error { return apply("decision/1") })
	page, err := m.BusState(buscalls.LedgerHistory, json.RawMessage(`{"ledger":"records","instance":"decision/1"}`))
	if err != nil || !strings.Contains(string(page), `"author":{"account":"ledger.anchor.m"}`) {
		t.Fatalf("the history holds %s, %v", page, err)
	}
	// Gone again: refused again, the change taken without a restart.
	if err := os.WriteFile(file, []byte(`{}`), 0o600); err != nil {
		t.Fatal(err)
	}
	mt.Until(t, func() error {
		if err := apply("decision/2"); err == nil || !strings.Contains(err.Error(), "credential is missing") {
			return os.ErrExist
		}
		return nil
	})
}