diff --git a/docs/configuration.md b/docs/configuration.md index 818f0e48..843b7831 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -174,6 +174,16 @@ 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. +Set to 0 for no limit. +Lowering the limit does not remove filters connections already hold. + +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 4d94f3af..87739e81 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: lib.SubscriptionController; /** 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 */ @@ -273,6 +273,7 @@ export default class Controller { }); // Handle subscriptions for all internal properties + this.subscriptions = new lib.SubscriptionController(config.get("controller.max_subscription_filters")); this.subscriptions.handle(lib.SystemInfoUpdateEvent, this.handleSystemInfoSubscription.bind(this)); this.subscriptions.handle(lib.HostUpdatesEvent, this.handleHostSubscription.bind(this)); this.subscriptions.handle(lib.InstanceDetailsUpdatesEvent, this.handleInstanceDetailsSubscription.bind(this)); diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index 27042c44..423f0a99 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,14 @@ 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 connection can hold for one subscribed event, 0 for no limit.", + type: "number", + initialValue: 10000, + restartRequired: true, + validator: validators.all(validators.integer, validators.greaterThanEqualZero), + }, "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..bcad8b2a 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 - + * Max number of filters a subscriber can hold for one event, 0 for no limit. + */ + constructor( + private _maxFilters = 0, + ) { } + /** * 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 (this._maxFilters === 0 || 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..e605a0fc 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), @@ -730,6 +730,64 @@ describe("lib/subscriptions", function() { await onceConnectorSend(0); assertLastEvent(0, RegisteredEvent); }); + + it("should reject filters over the limit", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + const ownSubscriptions = new lib.SubscriptionController(2); + ownSubscriptions.handle(RegisteredEvent); + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]); + await ownSubscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "baz"); + await assert.rejects( + ownSubscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src), + lib.RequestError, + ); + + ownSubscriptions.broadcast(new RegisteredEvent(), "baz"); + await new Promise(r => setImmediate(r)); + assertNoEvent(0); + }); + + it("should not count existing filters towards the limit", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + const ownSubscriptions = new lib.SubscriptionController(2); + ownSubscriptions.handle(RegisteredEvent); + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]); + await ownSubscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo"); + await ownSubscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + }); + + it("should not limit filters when subscribed to all", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + const ownSubscriptions = new lib.SubscriptionController(2); + ownSubscriptions.handle(RegisteredEvent); + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"); + await ownSubscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest( + RegisteredEvent.name, "subscribe", 0, ["foo", "bar", "baz"] + ); + await ownSubscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + }); + + it("should not limit filters when the limit is 0", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request = new lib.SubscriptionRequest( + RegisteredEvent.name, "subscribe", 0, ["foo", "bar", "baz"] + ); + await subscriptions.handleRequest(link, request, connectorData.dst, connectorData.src); + }); }); describe("unsubscribe", function() { @@ -852,6 +910,21 @@ describe("lib/subscriptions", function() { await new Promise(r => setImmediate(r)); assertNoEvent(0); }); + + it("should reject filters over the limit", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + const ownSubscriptions = new lib.SubscriptionController(2); + ownSubscriptions.handle(RegisteredEvent); + + const request = new lib.SubscriptionRequest( + RegisteredEvent.name, "replace", 0, ["foo", "bar", "baz"] + ); + await assert.rejects( + ownSubscriptions.handleRequest(link, request, connectorData.dst, connectorData.src), + lib.RequestError, + ); + }); }); }); });