From 690b79d2b744a74f79506aaa26599ba45b264915 Mon Sep 17 00:00:00 2001 From: bbassie <17990055+bbassie@users.noreply.github.com> Date: Thu, 8 Oct 2026 11:10:03 +0000 Subject: [PATCH] Limit the number of subscription filters per subscriber A subscriber could keep sending subscribe requests with new filters and the controller would add every one of them to the subscriber's filter set without bound, growing memory until it ran out. Add controller.max_subscription_filters (default 10000) and reject subscribe and replace requests that would leave a subscriber holding more filters than that for one event. Co-Authored-By: Claude Opus 5.5 --- docs/configuration.md | 9 +++++ packages/controller/src/Controller.ts | 2 +- packages/lib/src/config/definitions.ts | 8 ++++ packages/lib/src/subscriptions.ts | 28 ++++++++++++++ test/lib/subscriptions.test.js | 53 ++++++++++++++++++++++++-- 5 files changed, 95 insertions(+), 5 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 818f0e48..31c13c7c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -174,6 +174,15 @@ Must be greater than 0. Defaults to 8. +### controller.max_subscription_filters + +Maximum number of filters one connection can hold for a single subscribed event. +Subscription requests that would go over this limit are rejected. +Must be an integer greater than 0. + +Defaults to 10000. + + ### controller.system_metrics_interval Interval in seconds the controller gathers CPU, memory and disk usage from itself and every connected host for display in the web interface. diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 75a0e084..da0b12f0 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -83,7 +83,7 @@ export default class Controller { private _events = new events.EventEmitter(); /** Event subscription controller */ - subscriptions = new lib.SubscriptionController(); + subscriptions = new lib.SubscriptionController(() => this.config.get("controller.max_subscription_filters")); /** User manager for the cluster, also contains access to the user records */ users: UserManager; /** Instance manager for the cluster, also contains access to the user records */ diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index 27042c44..a8558b73 100644 --- a/packages/lib/src/config/definitions.ts +++ b/packages/lib/src/config/definitions.ts @@ -26,6 +26,7 @@ export interface ControllerConfigFields { "controller.heartbeat_interval": number; "controller.session_timeout": number; "controller.metrics_timeout": number; + "controller.max_subscription_filters": number; "controller.system_metrics_interval": number; "controller.proxy_stream_timeout": number; "controller.factorio_username": string | null; @@ -208,6 +209,13 @@ export class ControllerConfig extends classes.Config { initialValue: 8, validator: validators.greaterThanZero, }, + "controller.max_subscription_filters": { + title: "Max Subscription Filters", + description: "Maximum number of filters a single connection can hold for one subscribed event.", + type: "number", + initialValue: 10000, + validator: validators.all(validators.integer, validators.greaterThanZero), + }, "controller.system_metrics_interval": { title: "System Metrics Interval", description: "Interval in seconds to collect and update system metrics for the Web UI", diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 298bf5d7..0f865827 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -255,6 +255,14 @@ type EventData = { export class SubscriptionController { _events = new Map(); + /** + * @param maxFilters - + * Optional function returning the max number of filters a subscriber can hold for one event. + */ + constructor( + private _maxFilters: () => number = () => Infinity, + ) { } + /** * Allow clients to subscribe to an event by telling the subscription controller to accept them * Has an optional subscription update handler which is called when a client subscribes @@ -329,6 +337,24 @@ export class SubscriptionController { } } + private _checkFilterCount(filters: SubscriptionFilters, existing?: SubscriptionFilters) { + if (existing?.isAll()) { + return; // union into all is a noop + } + let count = filters.size; + if (existing) { + count = existing.size; + for (const filter of filters) { + if (!existing.has(filter)) { + count += 1; + } + } + } + if (count > this._maxFilters()) { + throw new RequestError(`Subscription exceeds the limit of ${this._maxFilters()} filters`); + } + } + /** * Handle incoming subscription requests on a link * @param link - Link message was received on @@ -359,6 +385,7 @@ export class SubscriptionController { break; case "subscribe": + this._checkFilterCount(request.filters, subscriber?.filters); if (!subscriber) { eventData.subscriptions.set(index, { link, address, filters: request.filters }); } else { @@ -367,6 +394,7 @@ export class SubscriptionController { break; case "replace": + this._checkFilterCount(request.filters); if (request.filters.isEmpty()) { if (!subscriber) { return false; diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js index 780d58b8..11ec9283 100644 --- a/test/lib/subscriptions.test.js +++ b/test/lib/subscriptions.test.js @@ -391,7 +391,7 @@ describe("lib/subscriptions", function() { return [connectorData.id, new lib.Link(new MockConnector(connectorData.src, connectorData.dst))]; })), }; - subscriptions = new lib.SubscriptionController(mockController); + subscriptions = new lib.SubscriptionController(); }); describe(".handle()", function() { @@ -628,7 +628,7 @@ describe("lib/subscriptions", function() { } it("should strip the request id from broadcast events", async function() { - const ownSubscriptions = new lib.SubscriptionController(mockController); + const ownSubscriptions = new lib.SubscriptionController(); ownSubscriptions.handle(RegisteredEvent); await ownSubscriptions.handleRequest( getLink(connectorData.id), @@ -646,7 +646,7 @@ describe("lib/subscriptions", function() { }); it("should strip the request id from the replayed event", async function() { - const ownSubscriptions = new lib.SubscriptionController(mockController); + const ownSubscriptions = new lib.SubscriptionController(); ownSubscriptions.handle(RegisteredEvent, async () => new RegisteredEvent()); await ownSubscriptions.handleRequest( getLink(connectorData.id), @@ -661,7 +661,7 @@ describe("lib/subscriptions", function() { }); it("should strip the request id when replacing filters creates the subscription", async function() { - const ownSubscriptions = new lib.SubscriptionController(mockController); + const ownSubscriptions = new lib.SubscriptionController(); ownSubscriptions.handle(RegisteredEvent); await ownSubscriptions.handleRequest( getLink(connectorData.id), @@ -732,6 +732,51 @@ describe("lib/subscriptions", function() { }); }); + describe("filter limit", function() { + const connectorData = connectorSetupDate[0]; + function request(action, filters) { + return new lib.SubscriptionRequest( + RegisteredEvent.name, action, 0, lib.SubscriptionFilters.fromShorthand(filters) + ); + } + beforeEach(function() { + subscriptions = new lib.SubscriptionController(() => 2); + subscriptions.handle(RegisteredEvent); + }); + + it("should reject a subscribe that goes over the limit", async function() { + const link = getLink(0); + await subscriptions.handleRequest(link, request("subscribe", ["a", "b"]), connectorData.dst); + await assert.rejects( + subscriptions.handleRequest(link, request("subscribe", ["c"]), connectorData.dst), + lib.RequestError, + ); + subscriptions.broadcast(new RegisteredEvent(), "c"); + await new Promise(r => setImmediate(r)); + assertNoEvent(0); + }); + + it("should not count filters already subscribed to", async function() { + const link = getLink(0); + await subscriptions.handleRequest(link, request("subscribe", ["a", "b"]), connectorData.dst); + await subscriptions.handleRequest(link, request("subscribe", ["a"]), connectorData.dst); + }); + + it("should reject a replace that goes over the limit", async function() { + await assert.rejects( + subscriptions.handleRequest(getLink(0), request("replace", ["a", "b", "c"]), connectorData.dst), + lib.RequestError, + ); + }); + + it("should allow subscribing to all", async function() { + const link = getLink(0); + await subscriptions.handleRequest(link, request("subscribe", ["a", "b"]), connectorData.dst); + await subscriptions.handleRequest(link, request("subscribe"), connectorData.dst); + await subscriptions.handleRequest(link, request("subscribe", ["c", "d", "e"]), connectorData.dst); + }); + }); + describe("unsubscribe", function() { it("should remove a subscription", async function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe");