[Security] DistributedPubSubMediator per-topic subscriber list unbounded

[pathosDev]

Severity / Size

• Severity: MEDIUM
• Size: S
• Threat model: closed-group cluster — pubsub mediator is reachable by every local actor and accepts gossip from peers. A buggy or malicious actor can spam Subscribe to a single topic; a peer can claim "I have N remote subscribers for topic X" without bound.

Affected files

• src/cluster/pubsub/DistributedPubSubMediator.ts:33-38 — SubscriberSet.local: Map<string, ActorRef> and SubscriberSet.remoteNodes: Set<string> — no cap on either.
• src/cluster/pubsub/DistributedPubSubMediator.ts:103-122 — handleSubscribe adds unconditionally.
• src/cluster/pubsub/DistributedPubSubMediator.ts:57 — topics: Map<string, SubscriberSet> — no cap on distinct topics either.
• src/cluster/pubsub/DistributedPubSubMediator.ts (gossip handler — same file) — gossip adds remote-node entries unconditionally.

Relationship to #137 (Receptionist)

This issue and #137 are structurally identical — both are cluster-wide registries with subscribers that grow without bound. The fix shape is the same:

• Per-key/topic subscriber cap.
• Total-subscriber cap across all keys/topics.
• DeathWatch + auto-cleanup.
• New SubscribeRejected-style message for caller feedback.
• Metric for ops visibility.

Consider coordinated implementation — share helper types (BoundedSubscriberRegistry<K>) between Receptionist and PubSubMediator to avoid divergence. Track separately as a refactor candidate after both ship.

The differences vs #137:

1. PubSub has two unbounded growth axes: local subscribers per topic, AND distinct topics (no automatic GC of topics with zero local + zero remote subscribers — there is a cleanup at handleUnsubscribe but not for topics that never had subscribers cleared).
2. PubSub has a remoteNodes set per topic — a malicious gossip peer can claim it has subscribers for arbitrary new topics, ballooning the topic count.
3. Publish fan-out cost scales with subscriber count → bounded subs is also a publish-latency guarantee.

Background

DistributedPubSubMediator holds topics: Map<string, SubscriberSet>, where each set has:

• local: Map<string, ActorRef> — local actors that subscribed.
• remoteNodes: Set<string> — remote nodes claiming at least one subscriber on this topic.

The handlers:

private handleSubscribe(msg: Subscribe): void {
  const set = this.getOrCreateSet(msg.topic);
  const key = msg.ref.path.toString();
  if (!set.local.has(key)) {
    set.local.set(key, msg.ref);    // <-- no cap
    this.version++;
    changed = true;
  }
  // ...
}

getOrCreateSet(topic) creates a new entry for any unknown topic — no cap on topic count either. A loop of mediator.tell(new Subscribe('topic-' + i, ref)) for i = 1..1M creates 1M topics in the map.

The gossip handler (in the same file) receives PubSubGossipMsg which lists { topic: string, hasLocalSubscribers: boolean } per topic; for each one with hasLocalSubscribers: true, the receiver adds the gossip sender to topics.get(topic).remoteNodes. A malicious peer gossiping 100K fake topics adds 100K topic entries on every receiver.

Publish fan-out:

private handlePublish(msg: Publish): void {
  const set = this.topics.get(msg.topic);
  if (!set) return;
  // fan-out: tell every local subscriber, and one envelope per remote node
  for (const ref of set.local.values()) ref.tell(msg.body);
  for (const node of set.remoteNodes) this.cluster.transport.send(NodeAddress.parse(node), wireMsg);
}

Cost per publish: O(local subscribers + remote nodes). At 1M subscribers, every publish is a 1M-iteration loop in the hot path.

Exploit walkthrough

Step 1 — App author uses pubsub for fan-out:

// Subscriber side, per-session actor:
mediator.tell(new Subscribe('notifications', this.self));
// Publisher side:
mediator.tell(new Publish('notifications', { kind: 'broadcast', payload: '...' }));

Step 2 — Two failure paths simultaneously:

• Per-topic explosion: a worker that spawns sessions per HTTP request subscribes on preStart; if cleanup is forgotten (or the actor dies before Unsubscribe), the subscriber sticks until the mediator restarts. After 100K requests → 100K subs on notifications topic. Publish to that topic takes ~50ms (1M-iteration fan-out × cluster.transport.send batching).
• Topic explosion: another worker subscribes to per-user topics ('user-' + userId); attacker iterates userId from 1..1M. The mediator's topics map balloons to 1M entries. Gossip ticks now serialise 1M topic names (~30MB per gossip message). Cluster bandwidth saturated.

Step 3 — Cascade: mediator's mailbox fills (publishes take ms each instead of µs), supervisor sees a slow actor, restarts it → all subscribers lost → every session re-subscribes simultaneously → thundering-herd. Publish-subscribe is effectively unavailable for minutes.

Realistic worst case: broadcast/notification infrastructure unavailable for entire pubsub-using app. Combined with #137 (Receptionist), the cluster's discovery + broadcast layer is unusable.

How the 8 already-landed security fixes inform this

• Wire-frame DoS cap — bound a growth vector at the boundary. Same shape: cap per-topic subs and total topic count.
• Hello-handshake hijack defence — verify identity. Same shape: validate gossip-claimed remote subscribers (rate-limit per sender; reject implausible counts).
• Snapshot seq integrity — strict bound on input. Same shape.
• [Security] Receptionist subscribers set is unbounded #137 (Receptionist) — once it lands, the same BoundedSubscriberRegistry<K> helper covers both.

Fix design

Track 1 — Per-topic local-subscriber cap (primary). Default: 1024. Same shape as #137.

export interface DistributedPubSubSettings {
  readonly cluster: Cluster;
  readonly gossipIntervalMs?: number;
  readonly maxSubscribersPerTopic?: number;     // default 1024
  readonly maxTopics?: number;                  // default 8192
  readonly maxRemoteNodesPerTopic?: number;     // default 256 (cluster size bound)
}

private handleSubscribe(msg: Subscribe): void {
  const set = this.getOrCreateSet(msg.topic);
  if (set === null) {
    // topic cap reached at getOrCreateSet
    msg.replyTo?.tell(new SubscribeRejected(msg.topic, 'topic-cap-reached'));
    return;
  }
  if (set.local.size >= this.maxSubscribersPerTopic) {
    this.log.warn(`[pubsub] subscriber cap (${this.maxSubscribersPerTopic}) reached for topic "${msg.topic}"`);
    msg.replyTo?.tell(new SubscribeRejected(msg.topic, 'subscriber-cap-reached'));
    return;
  }
  // ... existing add logic ...
}

private getOrCreateSet(topic: string): SubscriberSet | null {
  let s = this.topics.get(topic);
  if (s) return s;
  if (this.topics.size >= this.maxTopics) {
    this.log.warn(`[pubsub] topic cap (${this.maxTopics}) reached; rejecting new topic "${topic}"`);
    this.metrics.counter('pubsub_topic_rejected_total', { reason: 'cap-reached' }).inc();
    return null;
  }
  s = { local: new Map(), remoteNodes: new Set() };
  this.topics.set(topic, s);
  return s;
}

Track 2 — DeathWatch + auto-cleanup. Same as #137. Mediator watches each subscriber on accept; on Terminated, removes from all topics + drops empty topics.

Track 3 — Gossip-claimed-remote-subscribers validation. In handleGossip, reject gossip from a sender that claims subscribers on >maxTopicsClaimedPerSender topics (default 1024) — the legitimate case where one node has subscribers on >1024 topics is rare; an attacker generating fake topics is the adversarial case.

Track 4 — Remote-nodes-per-topic cap. maxRemoteNodesPerTopic (default = cluster size cap, e.g. 256). Reject gossip claiming a new remote node for a topic if the cap is reached. Combined with #138's cluster-member cap, this is double-bounded.

Track 5 — Empty-topic GC. When a topic has zero local subscribers AND zero remote nodes, drop it from topics. Already done in handleUnsubscribe but not on gossip path → add to gossip merge logic.

Track 6 — Metric. pubsub_subscribers_total{topic} Gauge (bucketed via #131's cap) + pubsub_topics_total Gauge + pubsub_topic_rejected_total Counter.

Track 7 — Documentation. README "Known security caveats":

- DistributedPubSub: 1024 subscribers per topic + 8192 distinct
  topics by default.  Reaching either cap returns SubscribeRejected.
  Gossip-claimed remote subscribers also capped per sender.

API surface

new DistributedPubSubSettings({
  cluster,
  maxSubscribersPerTopic: 4096,
  maxTopics: 32768,
  maxRemoteNodesPerTopic: 512,
});

// Subscriber receives one of:
//   SubscribeAck(msg)         — success
//   SubscribeRejected(topic, reason)  — capped

Backward compatibility

Behaviour change for apps with >1024 subscribers per topic OR >8192 distinct topics. Opt-out via Infinity. Document.

Test plan

1. Per-topic cap — 1024 subscribes succeed; 1025th → SubscribeRejected('subscriber-cap-reached').
2. Topic cap — 8192 distinct topics succeed; 8193rd → SubscribeRejected('topic-cap-reached').
3. Remote-nodes cap — gossip from 256 fake nodes for one topic accepted; 257th node rejected.
4. Auto-cleanup on death — subscribe; stop subscriber; verify topic is dropped if empty.
5. Gossip-claimed-topics cap — peer claims subscribers on 1025 topics → 1025th rejected, peer logged as suspicious.
6. Metric correctness — gauges match actual state.
7. Publish fan-out at cap — publish to a 1024-subscriber topic takes O(1024) work, not O(N).
8. Regression — existing pubsub tests pass.

Acceptance criteria

• DistributedPubSubSettings exposes maxSubscribersPerTopic, maxTopics, maxRemoteNodesPerTopic.
• All three caps enforced at handlers + gossip path.
• DeathWatch + auto-cleanup; empty topics dropped.
• SubscribeRejected message class exported (shareable with [Security] Receptionist subscribers set is unbounded #137).
• Metrics emitted.
• CHANGELOG + README "Known security caveats" updated.
• Test suite covers all 8 cases above.
• (Optional / separate refactor) Share BoundedSubscriberRegistry<K> helper with Receptionist.