fix(eventstream): add TopicPolicy interface for subscription authorization (PILOT-251)

service.go

@@ -44,20 +44,49 @@ const (
 // bound, growing the subscriber slice linearly.
 const maxSubsPerTopic = 1000
 
+// TopicPolicy is the authorization gate for topic subscription.
+// Implementations check whether a peer (identified by its 48-bit
+// virtual address) may subscribe to a given topic.
+//
+// The defaultAllowPolicy permits all subscriptions (backward
+// compatible). Daemon-level callers that need topic-level access
+// control provide a restricted implementation via SetTopicPolicy.
+type TopicPolicy interface {
+	// AllowSubscribe returns true if the peer at remoteAddr is
+	// permitted to subscribe to the named topic.
+	AllowSubscribe(remoteAddr coreapi.Addr, topic string) bool
+}
+
+// defaultAllowPolicy permits all subscriptions. Used when no
+// TopicPolicy has been set.
+type defaultAllowPolicy struct{}
+
+func (defaultAllowPolicy) AllowSubscribe(_ coreapi.Addr, _ string) bool { return true }
+
 // Service is the L11 plugin adapter. cmd/daemon/main.go (L12) and
 // cmd/pilotctl _daemon-run construct it via NewService and register
 // via daemon.RegisterPlugin.
 type Service struct {
-	listener coreapi.Listener
-	deps     coreapi.Deps
-	cancel   context.CancelFunc
-	done     chan struct{}
-	broker   *broker
+	listener    coreapi.Listener
+	deps        coreapi.Deps
+	cancel      context.CancelFunc
+	done        chan struct{}
+	broker      *broker
+	topicPolicy TopicPolicy
 }
 
 // NewService returns a Service ready for daemon.RegisterPlugin.
 func NewService() *Service {
-	return &Service{}
+	return &Service{topicPolicy: defaultAllowPolicy{}}
+}
+
+// SetTopicPolicy installs a TopicPolicy for subscription authorization.
+// Must be called before Start. A nil value is ignored (policy stays at
+// the previous setting, which defaults to allow-all).
+func (s *Service) SetTopicPolicy(p TopicPolicy) {
+	if p != nil {
+		s.topicPolicy = p
+	}
 }
 
 func (s *Service) Name() string { return "eventstream" }
@@ -73,7 +102,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, s.topicPolicy)
 
 	runCtx, cancel := context.WithCancel(ctx)
 	s.cancel = cancel
@@ -149,17 +178,19 @@ func (s *subscriber) remote() string {
 // Service; one instance per daemon. Tracks subscribers by topic plus
 // per-publisher rate buckets.
 type broker struct {
-	mu     sync.RWMutex
-	subs   map[string][]*subscriber
-	rateMu sync.Mutex
-	rate   map[*subscriber]*publishBucket
-	events coreapi.EventBus // for pubsub.* observability events
+	mu          sync.RWMutex
+	subs        map[string][]*subscriber
+	rateMu      sync.Mutex
+	rate        map[*subscriber]*publishBucket
+	events      coreapi.EventBus // for pubsub.* observability events
+	topicPolicy TopicPolicy
 }
 
-func newBroker(events coreapi.EventBus) *broker {
+func newBroker(events coreapi.EventBus, policy TopicPolicy) *broker {
 	return &broker{
-		subs:   make(map[string][]*subscriber),
-		events: events,
+		subs:        make(map[string][]*subscriber),
+		events:      events,
+		topicPolicy: policy,
 	}
 }
 
@@ -235,6 +266,23 @@ func (b *broker) handleConn(sub *subscriber) {
 	}
 	slog.Info("eventstream broker: subscribe received", "remote", sub.remote(), "topic", subEvt.Topic)
 	reqTopic := subEvt.Topic
+
+	// Topic-level authorization gate (SECURITY_PLAN AI-016).
+	// The topic policy defaults to allow-all for backward compatibility;
+	// daemon-level callers can install a restricted policy via
+	// Service.SetTopicPolicy.
+	if !b.topicPolicy.AllowSubscribe(sub.conn.RemoteAddr(), reqTopic) {
+		slog.Warn("eventstream broker: subscribe denied by topic policy",
+			"remote", sub.remote(),
+			"topic", reqTopic)
+		if b.events != nil {
+			b.events.Publish("pubsub.subscribe_denied", map[string]any{
+				"topic": reqTopic, "remote": sub.remote(),
+			})
+		}
+		return
+	}
+
 	if !b.addSub(reqTopic, sub) {
 		slog.Warn("eventstream broker: subscriber rejected, topic at cap",
 			"remote", sub.remote(),

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, defaultAllowPolicy{})
 
 	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, defaultAllowPolicy{})
 
 	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, defaultAllowPolicy{})
 
 	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, defaultAllowPolicy{})
 
 	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, defaultAllowPolicy{})
 
 	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, defaultAllowPolicy{})
 
 	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, defaultAllowPolicy{})
 	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, defaultAllowPolicy{})
 	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, defaultAllowPolicy{})
 	// 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, defaultAllowPolicy{})
 	sub := stubSubscriber()
 
 	// Seed a bucket with a lastRefill far in the past so the elapsed
@@ -709,3 +709,109 @@ func TestServer_Publish_TopicLiteralStar(t *testing.T) {
 		// expected — no second delivery.
 	}
 }
+
+// restrictPolicy denies subscription to any topic except "public.allowed".
+type restrictPolicy struct{}
+
+func (restrictPolicy) AllowSubscribe(_ coreapi.Addr, topic string) bool {
+	return topic == "public.allowed"
+}
+
+// TestBroker_HandleConn_TopicPolicyDeny verifies that a restrictive
+// TopicPolicy rejects subscription and publishes
+// pubsub.subscribe_denied instead of adding the subscriber.
+func TestBroker_HandleConn_TopicPolicyDeny(t *testing.T) {
+	t.Parallel()
+	bus := &stubEventBus{}
+	b := newBroker(bus, restrictPolicy{})
+
+	subStream, clientEnd := newPipeStreamPair()
+	sub := newSubscriber(subStream)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		b.handleConn(sub)
+	}()
+
+	// Subscribe to a topic that the policy forbids.
+	if err := WriteEvent(clientEnd, &Event{Topic: "forbidden"}); err != nil {
+		t.Fatalf("WriteEvent subscribe: %v", err)
+	}
+
+	// The broker must NOT add the subscriber to any topic.
+	time.Sleep(100 * time.Millisecond)
+	b.mu.RLock()
+	n := len(b.subs["forbidden"])
+	b.mu.RUnlock()
+	if n != 0 {
+		t.Errorf("forbidden topic has %d subs, want 0", n)
+	}
+
+	// The broker must have emitted pubsub.subscribe_denied.
+	if !bus.seen("pubsub.subscribe_denied") {
+		t.Errorf("expected pubsub.subscribe_denied event, got %v", bus.topics)
+	}
+
+	_ = clientEnd.Close()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handleConn did not return after client close")
+	}
+}
+
+// TestBroker_HandleConn_TopicPolicyAllow verifies that the policy
+// allows subscription when AllowSubscribe returns true.
+func TestBroker_HandleConn_TopicPolicyAllow(t *testing.T) {
+	t.Parallel()
+	bus := &stubEventBus{}
+	b := newBroker(bus, restrictPolicy{})
+
+	subStream, clientEnd := newPipeStreamPair()
+	sub := newSubscriber(subStream)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		b.handleConn(sub)
+	}()
+
+	// Subscribe to a topic the policy allows.
+	if err := WriteEvent(clientEnd, &Event{Topic: "public.allowed"}); err != nil {
+		t.Fatalf("WriteEvent subscribe: %v", err)
+	}
+
+	// The broker must add the subscriber to the allowed topic.
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		b.mu.RLock()
+		n := len(b.subs["public.allowed"])
+		b.mu.RUnlock()
+		if n == 1 {
+			break
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	b.mu.RLock()
+	n := len(b.subs["public.allowed"])
+	b.mu.RUnlock()
+	if n != 1 {
+		t.Errorf("public.allowed topic has %d subs, want 1", n)
+	}
+
+	// Must have published pubsub.subscribed, not subscribe_denied.
+	if !bus.seen("pubsub.subscribed") {
+		t.Errorf("expected pubsub.subscribed event, got %v", bus.topics)
+	}
+	if bus.seen("pubsub.subscribe_denied") {
+		t.Errorf("unexpected pubsub.subscribe_denied for allowed topic")
+	}
+
+	_ = clientEnd.Close()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handleConn did not return after client close")
+	}
+}

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, defaultAllowPolicy{})
 	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, defaultAllowPolicy{})
 	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, defaultAllowPolicy{})
 	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, defaultAllowPolicy{})
 	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, defaultAllowPolicy{})
 	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, defaultAllowPolicy{})
 	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, defaultAllowPolicy{})
 	healthy := stubSubscriber()
 	flaky := stubSubscriber()
 	b.addSub("shared", healthy)