mesh-controller/internal/broker/nats_test.go

package broker

import (
	"slices"
	"strings"
	"testing"
)

func has(t *testing.T, subjects []string, want string) {
	t.Helper()
	for _, s := range subjects {
		if s == want {
			return
		}
	}
	t.Fatalf("expected %q among %v", want, subjects)
}

func hasNot(t *testing.T, subjects []string, unwanted string) {
	t.Helper()
	for _, s := range subjects {
		if s == unwanted {
			t.Fatalf("did not expect %q among %v", unwanted, subjects)
		}
	}
}

// A module's authority comes from its declaration and nothing else (novox/hq ADR 0043).
func TestAModulePublishesOnlyWhatItEmits(t *testing.T) {
	p := Principal{Kind: KindModule, Node: "one", Module: "billing",
		Emits: []string{"order.placed"}, PasswordHash: "x"}
	perms, err := PermissionsFor(p)
	if err != nil {
		t.Fatal(err)
	}
	has(t, perms.Publish, "mesh.mod.billing.event.order.placed")
	hasNot(t, perms.Publish, "mesh.mod.billing.>")
	hasNot(t, perms.Publish, "mesh.mod.shipping.event.order.placed")
}

// The gap AMQP left open — an emitter granted the events exchange whole — is closed by per-subject
// permissions. A module cannot publish under another module's name.
func TestAModuleCannotPublishUnderAnothersName(t *testing.T) {
	perms, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "billing",
		Emits: []string{"order.placed"}, PasswordHash: "x"})
	for _, p := range perms.Publish {
		if strings.HasPrefix(p, "mesh.mod.") && !strings.HasPrefix(p, "mesh.mod.billing.") {
			t.Fatalf("billing may publish %q, which is not its own namespace", p)
		}
	}
}

// A caller of a seat may publish what the seat accepts, and nothing else of it: not its outbound
// events, and not a subscription to its inbound queue (design 29 §2).
func TestUsingASeatIsPublishOnlyAndInboundOnly(t *testing.T) {
	seat := Seat{Name: "telegram-sender", Accepts: []string{"send"}, Emits: []string{"delivered", "failed"}}
	perms, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "shop",
		Uses: []Seat{seat}, PasswordHash: "x"})
	has(t, perms.Publish, "mesh.seat.telegram-sender.accept.send")
	hasNot(t, perms.Publish, "mesh.seat.telegram-sender.event.delivered")
	hasNot(t, perms.Subscribe, "mesh.seat.telegram-sender.accept.send")
}

// The holder is the mirror image: it consumes what the seat accepts and publishes what it emits.
func TestHoldingASeatIsTheMirrorOfUsingIt(t *testing.T) {
	seat := Seat{Name: "telegram-sender", Accepts: []string{"send"}, Emits: []string{"delivered", "failed"}}
	perms, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "telegram",
		Holds: []Seat{seat}, PasswordHash: "x"})
	has(t, perms.Subscribe, "mesh.seat.telegram-sender.accept.send")
	has(t, perms.Publish, "mesh.seat.telegram-sender.event.delivered")
	hasNot(t, perms.Publish, "mesh.seat.telegram-sender.accept.send")
}

// A build machine may say everything about a build as it happens (novox/hq ADR 0157): that it
// started, and every line under the build's own id — the seat's `log.*` becomes a publish over
// one token, so a reader follows one build by subject and the holder can name no other subject.
func TestTheBuildMachineMaySayWhatItDoesUnderTheBuildsId(t *testing.T) {
	seat := Seat{Name: "mesh-build-machine", Accepts: []string{"build"}, Emits: []string{"started", "built", "log.*"}}
	perms, _ := PermissionsFor(Principal{Kind: KindModule, Node: "anchor", Module: "builder",
		Holds: []Seat{seat}, PasswordHash: "x"})
	has(t, perms.Publish, "mesh.seat.mesh-build-machine.event.started")
	has(t, perms.Publish, "mesh.seat.mesh-build-machine.event.log.*")
	hasNot(t, perms.Publish, "mesh.seat.mesh-build-machine.event.>")
}

// Without an ack permission a durable consumer never really consumes: every message it receives is
// redelivered forever, refused by the permission list it already has (design 25 §4).
func TestAModuleMayAckItsOwnDeliveriesAndNoOthers(t *testing.T) {
	perms, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "billing",
		Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
	has(t, perms.Publish, "$JS.ACK.EVENTS.one_billing.>")
	hasNot(t, perms.Publish, "$JS.ACK.>")
	hasNot(t, perms.Publish, "$JS.ACK.EVENTS.one_shop.>")
}

// With one account, inbox privacy is the permission list or it is nothing.
func TestAnInboxIsScopedToItsOwner(t *testing.T) {
	perms, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "billing", PasswordHash: "x"})
	has(t, perms.Subscribe, "_INBOX.one.billing.>")
	hasNot(t, perms.Subscribe, "_INBOX.>")
	hasNot(t, perms.Subscribe, "_INBOX.one.shop.>")
}

// A responder answers on the caller's inbox, which it has no permission for. allow_responses is
// what makes a scoped inbox workable at all — the authority is bounded by having been asked. A
// module is asked on its own namespace and may answer; a node and a person are never asked.
func TestOnlyWhatCanBeAskedMayAnswer(t *testing.T) {
	module, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "audit",
		Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
	if !module.AllowResponses {
		t.Fatal("a module cannot answer a tool call on its own namespace")
	}
	node, _ := PermissionsFor(Principal{Kind: KindNode, Node: "one", PasswordHash: "x"})
	if node.AllowResponses {
		t.Fatal("a node was granted the right to answer, and nothing asks a node anything")
	}
}

// A module serves every tool under its own name, and no other module's.
func TestAModuleServesItsOwnNamespaceAndNoOthers(t *testing.T) {
	p, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "gitea", PasswordHash: "x"})
	if !slices.Contains(p.Subscribe, "mesh.mod.gitea.tool.>") {
		t.Fatalf("a module may not serve its own tools: %v", p.Subscribe)
	}
	for _, s := range p.Subscribe {
		if strings.HasPrefix(s, "mesh.mod.") && !strings.HasPrefix(s, "mesh.mod.gitea.") {
			t.Fatalf("a module may subscribe another's namespace: %s", s)
		}
	}
}

// A host reaches its own node's control traffic and its own declaration, and nothing of any
// other node's.
func TestAHostIsConfinedToItsOwnNode(t *testing.T) {
	perms, _ := PermissionsFor(Principal{Kind: KindNode, Node: "one", PasswordHash: "x"})
	has(t, perms.Publish, "mesh.control.one.>")
	has(t, perms.Subscribe, "mesh.node.one.declare")
	hasNot(t, perms.Subscribe, "mesh.node.two.declare")
	hasNot(t, perms.Subscribe, "mesh.node.>")
}

// A leaked enrolment token is useless for anything but enrolling (design 25 §6).
func TestTheEnrolmentUserCanOnlyEnrol(t *testing.T) {
	perms, err := PermissionsFor(Principal{Kind: KindEnrolment, Node: "anchor", PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	if len(perms.Publish) != 1 || perms.Publish[0] != "mesh.control.enrol" {
		t.Fatalf("enrolment may publish %v", perms.Publish)
	}
	// Its own inbox and nothing else. **Nothing else** is the point: no declaration, no event, and
	// no other machine's answer — and the inbox itself is needed, because a node that cannot
	// subscribe one waits out its timeout against a mesh that answered.
	if len(perms.Subscribe) != 1 || perms.Subscribe[0] != "_INBOX.enrol.anchor.>" {
		t.Fatalf("enrolment may subscribe %v, which is not its own inbox alone", perms.Subscribe)
	}
}

// An enrolment user that names no node is refused: its inbox would be an empty subject token, and
// one that every nameless enrolment user shared — which is one machine reading the credentials
// sealed to another.
func TestAnEnrolmentUserWithoutANodeIsRefused(t *testing.T) {
	if _, err := PermissionsFor(Principal{Kind: KindEnrolment, PasswordHash: "x"}); err == nil {
		t.Fatal("an enrolment user with no node was composed, so its inbox is shared")
	}
}

// A name that would widen a permission is refused rather than quietly stretching one.
func TestANameThatWouldWidenAPermissionIsRefused(t *testing.T) {
	for _, bad := range []string{"bill.ing", "billing.>", "*", "bil>ling"} {
		if _, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: bad, PasswordHash: "x"}); err == nil {
			t.Fatalf("%q was accepted as part of a subject", bad)
		}
	}
}

// The entrypoint reloads on the file's digest changing, so an unchanged mesh must compose an
// identical file — otherwise every controller restart signals a reload of the whole bus.
func TestComposingTwiceGivesTheSameBytes(t *testing.T) {
	s := Server{ClientPort: 4222, MonitoringPort: 8222, StoreDir: "/data",
		TLSCert: "/tls/tls.crt", TLSKey: "/tls/tls.key", TLSCA: "/tls/ca.crt"}
	ps := []Principal{
		{Kind: KindModule, Node: "two", Module: "shop", Emits: []string{"order.placed"}, PasswordHash: "b"},
		{Kind: KindController, PasswordHash: "c"},
		{Kind: KindModule, Node: "one", Module: "billing", Consumes: []string{"shop.order.placed"}, PasswordHash: "a"},
	}
	first, err := Compose(s, ps)
	if err != nil {
		t.Fatal(err)
	}
	shuffled := []Principal{ps[2], ps[0], ps[1]}
	second, err := Compose(s, shuffled)
	if err != nil {
		t.Fatal(err)
	}
	if first != second {
		t.Fatal("composition is order-dependent; every controller restart would reload the bus")
	}
}

// A user without a password is a user anybody is.
func TestAUserWithoutAPasswordIsRefused(t *testing.T) {
	_, err := Compose(Server{ClientPort: 4222}, []Principal{{Kind: KindController}})
	if err == nil {
		t.Fatal("composed a user with no password hash")
	}
}

// A person reaches the mesh's tools from a workstation (design 25 §7). Their authority is a list
// of tools and nothing else.
func TestAPersonMayAskOnlyTheToolsTheyWereGiven(t *testing.T) {
	perms, err := PermissionsFor(Principal{Kind: KindPerson, Module: "jo",
		Invokes: []string{"shop.price", "telegram.status"}, PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	has(t, perms.Publish, "mesh.mod.shop.tool.price")
	has(t, perms.Publish, "mesh.mod.telegram.tool.status")
	hasNot(t, perms.Publish, "mesh.mod.shop.tool.refund")
	hasNot(t, perms.Publish, "mesh.mod.*.tool.>")
}

// An administrator gets every tool, which is a different grant and looks like one.
func TestAnAdministratorMayAskAnyTool(t *testing.T) {
	perms, _ := PermissionsFor(Principal{Kind: KindPerson, Module: "jo",
		Invokes: []string{"*"}, PasswordHash: "x"})
	has(t, perms.Publish, "mesh.mod.*.tool.>")
}

// **Nothing but tools.** A person who could publish an event would be able to claim a module
// said something; one who could publish control traffic would be a second controller.
func TestAPersonReachesNothingButTools(t *testing.T) {
	perms, _ := PermissionsFor(Principal{Kind: KindPerson, Module: "jo",
		Invokes: []string{"*"}, PasswordHash: "x"})
	for _, p := range perms.Publish {
		if !strings.Contains(p, ".tool.") {
			t.Errorf("a person may publish %q, which is not a tool call", p)
		}
	}
	for _, s := range perms.Subscribe {
		if !strings.HasPrefix(s, "_INBOX.person.") {
			t.Errorf("a person may subscribe %q; only their own inbox should be reachable", s)
		}
	}
}

// A person has no durable consumer, because nothing is delivered to a person — so no ack
// subject, and an ack permission would be authority over something that does not exist.
func TestAPersonHasNoAckSubject(t *testing.T) {
	perms, _ := PermissionsFor(Principal{Kind: KindPerson, Module: "jo",
		Invokes: []string{"*"}, PasswordHash: "x"})
	for _, p := range perms.Publish {
		if strings.HasPrefix(p, "$JS.ACK") {
			t.Errorf("a person was granted %q, and has no consumer to acknowledge", p)
		}
	}
}

// A person asks and is answered; they never answer. allow_responses would let a person reply to
// a request — which, on a bus where anyone may serve a tool, is somebody impersonating a module.
func TestAPersonMayNotAnswer(t *testing.T) {
	perms, _ := PermissionsFor(Principal{Kind: KindPerson, Module: "jo",
		Invokes: []string{"*"}, PasswordHash: "x"})
	if perms.AllowResponses {
		t.Fatal("a person may answer a request, which is impersonating a module")
	}
}

// Two people do not share an inbox, or one would read the other's answers.
func TestTwoPeopleDoNotShareAnInbox(t *testing.T) {
	a, _ := PermissionsFor(Principal{Kind: KindPerson, Module: "jo", Invokes: []string{"*"}, PasswordHash: "x"})
	b, _ := PermissionsFor(Principal{Kind: KindPerson, Module: "sam", Invokes: []string{"*"}, PasswordHash: "x"})
	if a.Subscribe[0] == b.Subscribe[0] {
		t.Fatalf("both read %s", a.Subscribe[0])
	}
}

// A malformed grant is refused rather than widened into something that happens to parse.
func TestAToolGrantThatNamesNoToolIsRefused(t *testing.T) {
	if _, err := PermissionsFor(Principal{Kind: KindPerson, Module: "jo",
		Invokes: []string{"shop"}, PasswordHash: "x"}); err == nil {
		t.Fatal("a grant naming a module but no tool was accepted")
	}
}

// The controller can answer an enrolment, and reach no other inbox.
//
// **`allow_responses` does not cover this and that is the trap.** It permits one reply to the reply
// subject of a message the user received — and a message a JetStream consumer delivers has had that
// field claimed for the consumer's own ack address, so the address the controller actually answers is
// the one the request carried in its payload, which the server does not recognise as a reply subject
// at all.
//
// Found against a real server, after a live test on an *unpermissioned* one had passed: every
// enrolment on the mesh would have timed out while the controller logged success.
func TestTheControllerCanAnswerAnEnrolmentAndReachNoOtherInbox(t *testing.T) {
	ctl, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	enrolling, err := PermissionsFor(Principal{Kind: KindEnrolment, Node: "anchor", PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}

	// Whatever the enrolling node waits on, the controller must be able to publish to.
	if len(enrolling.Subscribe) != 1 {
		t.Fatalf("an enrolling node subscribes %v, and this test knows only how to check one",
			enrolling.Subscribe)
	}
	waitsOn := enrolling.Subscribe[0]
	if !covers(ctl.Publish, waitsOn) {
		t.Fatalf("the controller may publish %v, none of which reaches %s — so every enrolment on "+
			"the mesh times out while the controller logs success", ctl.Publish, waitsOn)
	}

	// And nothing wider. A node's own inbox and a module's are not the controller's to write into:
	// that is the blanket grant design 25 §4 refuses.
	for _, other := range []string{"_INBOX.node.anchor.x", "_INBOX.one.shop.x", "_INBOX.person.ada.x"} {
		if covers(ctl.Publish, other) {
			t.Errorf("the controller can publish to %s, which is an inbox privacy the permission "+
				"list is the only thing protecting", other)
		}
	}
}

// covers says whether any granted subject pattern admits one concrete subject, with NATS's own
// wildcard meanings: `*` is one token, `>` is the rest.
func covers(granted []string, subject string) bool {
	want := strings.Split(subject, ".")
	for _, pattern := range granted {
		if admits(strings.Split(pattern, "."), want) {
			return true
		}
	}
	return false
}

func admits(pattern, subject []string) bool {
	for i, token := range pattern {
		if token == ">" {
			return i < len(subject)
		}
		if i >= len(subject) {
			return false
		}
		if token != "*" && token != subject[i] {
			return false
		}
	}
	return len(pattern) == len(subject)
}

// A module pulls its own consumer — asks about it, asks it for messages — and no other module's.
func TestAModulePullsItsOwnConsumerAndNoOthers(t *testing.T) {
	p, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "audit",
		Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
	for _, want := range []string{"$JS.API.CONSUMER.INFO.EVENTS.one_audit", "$JS.API.CONSUMER.MSG.NEXT.EVENTS.one_audit"} {
		if !slices.Contains(p.Publish, want) {
			t.Errorf("a module cannot bind its own consumer: %v lacks %s", p.Publish, want)
		}
	}
	for _, s := range p.Publish {
		if strings.Contains(s, "CONSUMER.") && !strings.HasSuffix(s, ".one_audit") {
			t.Errorf("a module may reach another consumer: %s", s)
		}
	}
	for _, s := range p.Subscribe {
		if strings.HasPrefix(s, "_DELIVER.") {
			t.Errorf("a module is granted a push delivery it never binds: %s", s)
		}
	}
}

// The runtime's authority is the union of what the modules it carries would have been granted for
// their tools (novox/hq ADR 0175): every carried module's tool namespace, every held seat's verbs
// on this node, every module's membership on this node, and a call to anything. Nothing it
// consumes, because it reacts to nothing.
func TestTheRuntimeServesTheUnionAndConsumesNothing(t *testing.T) {
	filter := Seat{Name: "node-packet-filter", Scope: "node", Serves: []string{"rules", "reload"}}
	p := Principal{Kind: KindNodeTools, Node: "anchor", Module: RuntimeModule, Carries: []Declared{
		{Module: "nftables", Holds: []Seat{filter}, Serves: []string{"firewall_rules"}},
		{Module: "zsh", Emits: []string{"shell.opened"}, Consumes: []string{"shop.order.placed"}},
		{Module: RuntimeModule},
	}}
	perms, err := PermissionsFor(p)
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{
		"mesh.mod.nftables.tool.>", "mesh.mod.zsh.tool.>", "mesh.mod." + RuntimeModule + ".tool.>",
		"mesh.seat.node-packet-filter.tool.rules.anchor", "mesh.seat.node-packet-filter.tool.reload.anchor",
		"mesh.assignment.anchor.*",
		"_INBOX.anchor." + RuntimeModule + ".>",
	} {
		if !contains(perms.Subscribe, want) {
			t.Errorf("the runtime may not subscribe %s: %v", want, perms.Subscribe)
		}
	}
	for _, want := range []string{
		"mesh.mod.*.tool.>", "mesh.seat.*.tool.>",
		"$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.anchor.*",
		"mesh.mod.zsh.event.shell.opened",
	} {
		if !contains(perms.Publish, want) {
			t.Errorf("the runtime may not publish %s: %v", want, perms.Publish)
		}
	}
	// Nothing of what a carried module consumes, and no consumer of its own to ack.
	for _, s := range perms.Subscribe {
		if strings.Contains(s, ".event.") || strings.HasPrefix(s, "_DELIVER.") {
			t.Errorf("the runtime was granted a delivery it has no consumer for: %s", s)
		}
	}
	for _, s := range perms.Publish {
		if strings.HasPrefix(s, "$JS.ACK.") || strings.Contains(s, "CONSUMER") {
			t.Errorf("the runtime was granted a consumer's subject and has no consumer: %s", s)
		}
	}
	if !perms.AllowResponses {
		t.Error("the runtime answers what it is asked, and may not reply")
	}
	if _, needed := ConsumerFor(p); needed {
		t.Error("a consumer would be made for the runtime, which consumes nothing")
	}
	// Each subject once: the file is read as the mesh's authority model.
	seen := map[string]bool{}
	for _, s := range append(append([]string{}, perms.Subscribe...), perms.Publish...) {
		if seen[s] {
			t.Errorf("%s is granted twice", s)
		}
		seen[s] = true
	}
}

func contains(list []string, want string) bool {
	for _, s := range list {
		if s == want {
			return true
		}
	}
	return false
}

// A node-scoped seat's work is shared (novox/hq ADR 0190): its holder on any machine subscribes the
// seat's one work subject, with no node in it, so holders on several machines read one queue. The
// node token belongs to a seat's tools, which are asked of one machine (design 33 §4), not to its work.
func TestANodeSeatsWorkSubjectCarriesNoNode(t *testing.T) {
	seat := Seat{Name: "node-build-agent", Scope: "node", Accepts: []string{"build"}, Serves: []string{"status"}}
	perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "anchor", Module: "build-agent", Holds: []Seat{seat}})
	if err != nil {
		t.Fatal(err)
	}
	has(t, perms.Subscribe, "mesh.seat.node-build-agent.accept.build")
	hasNot(t, perms.Subscribe, "mesh.seat.node-build-agent.accept.build.anchor")
	// And its tools still carry the machine.
	has(t, perms.Subscribe, "mesh.seat.node-build-agent.tool.status.anchor")
	// The controller asks the role, not a machine.
	controller, err := PermissionsFor(Principal{Kind: KindController})
	if err != nil {
		t.Fatal(err)
	}
	has(t, controller.Publish, "mesh.seat.node-build-agent.accept.>")
}