fix(eventstream): gate subscription on TrustChecker (PILOT-251)

service.go

@@ -73,7 +73,7 @@ func (s *Service) Start(ctx context.Context, deps coreapi.Deps) error {
 		return fmt.Errorf("eventstream: listen on port %d: %w", protocol.PortEventStream, err)
 	}
 	s.listener = ln
-	s.broker = newBroker(deps.Events)
+	s.broker = newBroker(deps.Events, deps.Trust)
 
 	runCtx, cancel := context.WithCancel(ctx)
 	s.cancel = cancel
@@ -154,12 +154,14 @@ type broker struct {
 	rateMu sync.Mutex
 	rate   map[*subscriber]*publishBucket
 	events coreapi.EventBus // for pubsub.* observability events
+	trust  coreapi.TrustChecker // optional — nil means no trust gate loaded
 }
 
-func newBroker(events coreapi.EventBus) *broker {
+func newBroker(events coreapi.EventBus, trust coreapi.TrustChecker) *broker {
 	return &broker{
 		subs:   make(map[string][]*subscriber),
 		events: events,
+		trust:  trust,
 	}
 }
 
@@ -234,6 +236,20 @@ func (b *broker) handleConn(sub *subscriber) {
 		return
 	}
 	slog.Info("eventstream broker: subscribe received", "remote", sub.remote(), "topic", subEvt.Topic)
+
+	// Auth gate (PILOT-251): when a TrustChecker is loaded (trustedagents
+	// plugin registered), reject subscription if the peer is NOT on the
+	// allowlist. When trust is nil (trustedagents not loaded), fall back
+	// to the previous behavior — allow all — for backward compatibility.
+	nodeID := sub.conn.RemoteAddr().Node
+	if b.trust != nil {
+		if _, ok := b.trust.IsTrusted(nodeID); !ok {
+			slog.Warn("eventstream broker: rejecting subscription — peer not trusted",
+				"remote", sub.remote(), "topic", subEvt.Topic, "nodeID", nodeID)
+			return
+		}
+	}
+
 	reqTopic := subEvt.Topic
 	if !b.addSub(reqTopic, sub) {
 		slog.Warn("eventstream broker: subscriber rejected, topic at cap",

zz_added_coverage_test.go

@@ -100,7 +100,7 @@ func (s *pipeStream) SetWriteDeadline(time.Time) error { return nil }
 func TestBroker_HandleConn_SubscribeEmitsEvent(t *testing.T) {
 	t.Parallel()
 	bus := &stubEventBus{}
-	b := newBroker(bus)
+	b := newBroker(bus, nil)
 
 	subStream, clientEnd := newPipeStreamPair()
 	sub := newSubscriber(subStream)
@@ -153,7 +153,7 @@ func TestBroker_HandleConn_SubscribeEmitsEvent(t *testing.T) {
 func TestBroker_HandleConn_SubscribeReadFails(t *testing.T) {
 	t.Parallel()
 	bus := &stubEventBus{}
-	b := newBroker(bus)
+	b := newBroker(bus, nil)
 
 	subStream, clientEnd := newPipeStreamPair()
 	sub := newSubscriber(subStream)
@@ -190,7 +190,7 @@ func TestBroker_HandleConn_SubscribeReadFails(t *testing.T) {
 func TestBroker_HandleConn_PublishedEventBusBranch(t *testing.T) {
 	t.Parallel()
 	bus := &stubEventBus{}
-	b := newBroker(bus)
+	b := newBroker(bus, nil)
 
 	subStream, clientEnd := newPipeStreamPair()
 	sub := newSubscriber(subStream)
@@ -248,7 +248,7 @@ func TestBroker_HandleConn_PublishedEventBusBranch(t *testing.T) {
 func TestBroker_HandleConn_RateLimitBusBranch(t *testing.T) {
 	t.Parallel()
 	bus := &stubEventBus{}
-	b := newBroker(bus)
+	b := newBroker(bus, nil)
 
 	subStream, clientEnd := newPipeStreamPair()
 	sub := newSubscriber(subStream)
@@ -315,7 +315,7 @@ func TestBroker_HandleConn_RateLimitBusBranch(t *testing.T) {
 // (lines 312-315 — wildcard subscriber receives non-wildcard event).
 func TestPublishWith_WildcardFanout(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 
 	topicSub := stubSubscriber()
 	wildSub := stubSubscriber()
@@ -348,7 +348,7 @@ func TestPublishWith_WildcardFanout(t *testing.T) {
 // should not receive its own publish.
 func TestPublishWith_WildcardSenderSkipped(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 
 	wildSender := stubSubscriber()
 	wildOther := stubSubscriber()
@@ -381,7 +381,7 @@ func TestPublishWith_WildcardSenderSkipped(t *testing.T) {
 // otherwise double-deliver).
 func TestPublishWith_TopicIsWildcardDoesNotDoubleDeliver(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	wildSub := stubSubscriber()
 	b.addSub("*", wildSub)
 
@@ -408,7 +408,7 @@ func TestPublishWith_TopicIsWildcardDoesNotDoubleDeliver(t *testing.T) {
 func TestPublishWith_BusPublishedBranch(t *testing.T) {
 	t.Parallel()
 	bus := &stubEventBus{}
-	b := newBroker(bus)
+	b := newBroker(bus, nil)
 	sub := stubSubscriber()
 	b.addSub("t", sub)
 
@@ -451,7 +451,7 @@ func TestService_StopContextCancel(t *testing.T) {
 // the per-topic subscriber cap is reached (memory-DoS protection).
 func TestBroker_AddSub_RejectsAtCap(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	// Seed the topic with exactly maxSubsPerTopic subscribers.
 	for i := 0; i < maxSubsPerTopic; i++ {
 		sub := stubSubscriber()
@@ -487,7 +487,7 @@ func TestBroker_AddSub_RejectsAtCap(t *testing.T) {
 // so it must be clamped.
 func TestTakeToken_RefillCappedAtBurst(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	sub := stubSubscriber()
 
 	// Seed a bucket with a lastRefill far in the past so the elapsed
@@ -664,6 +664,127 @@ func TestServer_Publish_WildcardSenderSkipped(t *testing.T) {
 	}
 }
 
+// --- Trust gate (PILOT-251) ------------------------------------------------
+
+// stubTrustChecker is a test double that reports a fixed set of trusted
+// node IDs.
+type stubTrustChecker struct {
+	mu      sync.Mutex
+	trusted map[uint32]string
+}
+
+func (stc *stubTrustChecker) IsTrusted(nodeID uint32) (string, bool) {
+	stc.mu.Lock()
+	defer stc.mu.Unlock()
+	name, ok := stc.trusted[nodeID]
+	return name, ok
+}
+
+func (stc *stubTrustChecker) setTrusted(nodeIDs ...uint32) {
+	stc.mu.Lock()
+	defer stc.mu.Unlock()
+	if stc.trusted == nil {
+		stc.trusted = make(map[uint32]string)
+	}
+	for _, id := range nodeIDs {
+		stc.trusted[id] = "test-peer"
+	}
+}
+
+// TestBroker_HandleConn_TrustGateRejectsUntrusted verifies that
+// handleConn rejects a subscriber whose node ID is not trusted when
+// the broker has a TrustChecker.
+func TestBroker_HandleConn_TrustGateRejectsUntrusted(t *testing.T) {
+	t.Parallel()
+
+	// Broker with a trust checker that has NO trusted peers.
+	tc := &stubTrustChecker{}
+	b := newBroker(nil, tc)
+
+	subStream, clientEnd := newPipeStreamPair()
+	sub := newSubscriber(subStream)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		b.handleConn(sub)
+	}()
+
+	// Client writes subscribe event.
+	if err := WriteEvent(clientEnd, &Event{Topic: "secret"}); err != nil {
+		t.Fatalf("WriteEvent subscribe: %v", err)
+	}
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handleConn should have returned after rejecting untrusted subscriber")
+	}
+
+	// The subscriber must NOT have been added.
+	b.mu.RLock()
+	n := len(b.subs["secret"])
+	b.mu.RUnlock()
+	if n != 0 {
+		t.Errorf("untrusted peer added to subs, got %d subscribers, want 0", n)
+	}
+
+	_ = clientEnd.Close()
+}
+
+// TestBroker_HandleConn_TrustGateAcceptsTrusted verifies that handleConn
+// accepts a subscriber whose node ID matches the trusted allowlist.
+func TestBroker_HandleConn_TrustGateAcceptsTrusted(t *testing.T) {
+	t.Parallel()
+
+	// Broker with a trust checker that trusts our peer (Node = 0xBEEF
+	// from pipeStream stub).
+	tc := &stubTrustChecker{}
+	tc.setTrusted(0xBEEF)
+	b := newBroker(nil, tc)
+
+	subStream, clientEnd := newPipeStreamPair()
+	sub := newSubscriber(subStream)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		b.handleConn(sub)
+	}()
+
+	// Client writes subscribe event.
+	if err := WriteEvent(clientEnd, &Event{Topic: "public"}); err != nil {
+		t.Fatalf("WriteEvent subscribe: %v", err)
+	}
+
+	// Wait for subscription to be registered.
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		b.mu.RLock()
+		n := len(b.subs["public"])
+		b.mu.RUnlock()
+		if n == 1 {
+			break
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	b.mu.RLock()
+	n := len(b.subs["public"])
+	b.mu.RUnlock()
+	if n != 1 {
+		t.Errorf("trusted peer NOT added to subs, got %d subscribers, want 1", n)
+	}
+
+	// Close to let handleConn exit.
+	_ = clientEnd.Close()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handleConn did not return after client close")
+	}
+}
+
 // TestServer_Publish_TopicLiteralStar covers the "evt.Topic == '*'" skip
 // branch in Server.publish (line 111): when topic literally is "*", the
 // wildcard fan-out loop is bypassed (would otherwise double-deliver).

zz_more_test.go

@@ -55,7 +55,7 @@ func TestNewSubscriber_NilWrap(t *testing.T) {
 // without going through handleConn.
 func TestBrokerAddRemoveSub_DirectCalls(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	sub1 := stubSubscriber()
 	sub2 := stubSubscriber()
 
@@ -89,7 +89,7 @@ func TestBrokerAddRemoveSub_DirectCalls(t *testing.T) {
 // publisher's bucket should be gone after the call.
 func TestBrokerForgetPublisher(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	sub := stubSubscriber()
 	// Prime the bucket.
 	if !b.takeToken(sub) {
@@ -107,7 +107,7 @@ func TestBrokerForgetPublisher(t *testing.T) {
 // TestTakeToken_BurstAndRecover exercises the rate-limit drain + refill.
 func TestTakeToken_BurstAndRecover(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	sub := stubSubscriber()
 	// Drain the entire burst budget.
 	drained := 0

zz_resilience_test.go

@@ -47,7 +47,7 @@ func stubSubscriber() *subscriber {
 // the retry succeeds — subscriber stays alive, counter never increments.
 func TestPubsubRetryAbsorbsSingleTransientFailure(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	sub := stubSubscriber()
 	b.addSub("topic-a", sub)
 
@@ -73,7 +73,7 @@ func TestPubsubRetryAbsorbsSingleTransientFailure(t *testing.T) {
 // maxConsecutivePublishFailures across publishes.
 func TestPubsubKeepsSubscriberBelowFailureThreshold(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	sub := stubSubscriber()
 	b.addSub("topic-fail", sub)
 
@@ -107,7 +107,7 @@ func TestPubsubKeepsSubscriberBelowFailureThreshold(t *testing.T) {
 // counter restarted from 0, threshold is 3).
 func TestPubsubFailureCounterResetsOnSuccess(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	sub := stubSubscriber()
 	b.addSub("topic-mix", sub)
 
@@ -151,7 +151,7 @@ func TestPubsubFailureCounterResetsOnSuccess(t *testing.T) {
 // the flaky sub drops.
 func TestPubsubOneSubscribersFailureDoesNotKillOthers(t *testing.T) {
 	t.Parallel()
-	b := newBroker(nil)
+	b := newBroker(nil, nil)
 	healthy := stubSubscriber()
 	flaky := stubSubscriber()
 	b.addSub("shared", healthy)