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");