From 4e33e288b26997a9647568e2bc057dde6fd1fb2a Mon Sep 17 00:00:00 2001 From: "Matthew (bot)" Date: Mon, 13 Jul 2026 14:22:34 +0000 Subject: [PATCH] fix(eventstream): add TopicPolicy interface for subscription authorization (PILOT-251) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce a TopicPolicy interface that gates topic subscription in the eventstream broker's handleConn path. Before this change, any peer that completed L6 key exchange could subscribe to any topic — no access control existed (SECURITY_PLAN AI-016). Changes: - Add TopicPolicy interface with AllowSubscribe(remoteAddr, topic) - Default policy (defaultAllowPolicy) permits all subscriptions, preserving backward compatibility for existing deployments - Broker holds the policy and checks it before addSub; denied subscriptions log a warning and publish pubsub.subscribe_denied bus event - Service exports SetTopicPolicy(p) so daemon-level callers can install restricted policies when topic-level access control is required - Tests for both allowed and denied subscription paths Closes PILOT-251 --- service.go | 78 ++++++++++++++++++----- zz_added_coverage_test.go | 126 +++++++++++++++++++++++++++++++++++--- zz_more_test.go | 6 +- zz_resilience_test.go | 8 +-- 4 files changed, 186 insertions(+), 32 deletions(-) diff --git a/service.go b/service.go index 9a7cf22..6a11f8c 100644 --- a/service.go +++ b/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(), diff --git a/zz_added_coverage_test.go b/zz_added_coverage_test.go index 13b0bb9..1b12e5a 100644 --- a/zz_added_coverage_test.go +++ b/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") + } +} diff --git a/zz_more_test.go b/zz_more_test.go index 9ec3ffe..09501a1 100644 --- a/zz_more_test.go +++ b/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 diff --git a/zz_resilience_test.go b/zz_resilience_test.go index bdb1ee5..f38aa49 100644 --- a/zz_resilience_test.go +++ b/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)