mesh-controller/internal/broker/nats_test.go

package broker

import (
	"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.order.placed")
	hasNot(t, perms.Publish, "mesh.mod.billing.>")
	hasNot(t, perms.Publish, "mesh.mod.shipping.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.send")
	hasNot(t, perms.Publish, "mesh.seat.telegram-sender.delivered")
	hasNot(t, perms.Subscribe, "mesh.seat.telegram-sender.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.send")
	has(t, perms.Publish, "mesh.seat.telegram-sender.delivered")
	hasNot(t, perms.Publish, "mesh.seat.telegram-sender.send")
}

// 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.
func TestOnlySomethingThatServesMayAnswer(t *testing.T) {
	serving, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "billing",
		Serves: []string{"status"}, PasswordHash: "x"})
	if !serving.AllowResponses {
		t.Fatal("a module serving a tool cannot answer the caller's inbox")
	}
	consumer, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "audit",
		Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
	if consumer.AllowResponses {
		t.Fatal("a pure consumer was granted the right to answer, which nothing asked it to do")
	}
}

// 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, _ := PermissionsFor(Principal{Kind: KindEnrolment, PasswordHash: "x"})
	if len(perms.Publish) != 1 || perms.Publish[0] != "mesh.control.enrol" {
		t.Fatalf("enrolment may publish %v", perms.Publish)
	}
	if len(perms.Subscribe) != 0 {
		t.Fatalf("enrolment may subscribe %v, and should hear nothing", perms.Subscribe)
	}
}

// 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")
	}
}