Merge pull request #1055 from bbassie/fix/subscription-limits

Limit the number of subscription filters per subscriber

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.

packages/controller/src/Controller.ts

@@ -83,7 +83,7 @@ export default class Controller {
 	private _events = new events.EventEmitter<ControllerEvents>();
 
 	/** 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));

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<ControllerConfigFields> {
 			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",

packages/lib/src/subscriptions.ts

@@ -255,6 +255,14 @@ type EventData = {
 export class SubscriptionController {
 	_events = new Map<string, EventData>();
 
+	/**
+	 * @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;

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,
+					);
+				});
 			});
 		});
 	});