From d8e18518a7156e205f481cc3454268a7fa9cd8f2 Mon Sep 17 00:00:00 2001 From: "Matthew (Pilot Protocol)" Date: Sun, 12 Jul 2026 23:57:57 +0000 Subject: [PATCH] fix(eventstream): gate subscription on TrustChecker (PILOT-251) Add an auth gate in handleConn: when the trustedagents plugin is registered (deps.Trust != nil), reject subscription attempts from peers whose node ID is NOT on the trusted-agents allowlist. When trustedagents is NOT loaded (deps.Trust == nil), fall back to the previous behavior (allow all) for backward compatibility. Changes: - service.go: broker stores optional TrustChecker, handleConn checks sub.conn.RemoteAddr().Node against trust.IsTrusted before addSub - zz_added_coverage_test.go: +2 tests (reject untrusted, accept trusted) - zz_more_test.go, zz_resilience_test.go: update newBroker call sites Closes PILOT-251 --- service.go | 20 +++++- zz_added_coverage_test.go | 141 +++++++++++++++++++++++++++++++++++--- zz_more_test.go | 6 +- zz_resilience_test.go | 8 +-- 4 files changed, 156 insertions(+), 19 deletions(-) diff --git a/service.go b/service.go index 9a7cf22..233ae2a 100644 --- a/service.go +++ b/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", diff --git a/zz_added_coverage_test.go b/zz_added_coverage_test.go index 13b0bb9..19b81a7 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, 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). diff --git a/zz_more_test.go b/zz_more_test.go index 9ec3ffe..b919391 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, 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 diff --git a/zz_resilience_test.go b/zz_resilience_test.go index bdb1ee5..5ebf4d3 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, 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)