Filter plugin messages forwarded to links

Drop events forwarded to a slave or instance that does not have the
plugin the event is defined in.  Reject requests for the same reason.

docs/socket.md

@@ -129,7 +129,7 @@ Slave handshake for establishing a new connection session.
 - id - integer - ID of the slave.
 - plugins - Object<string, string> -
     Object mapping plugin names to plugin version for plugins that are
-    available on the slave.  Currently not used by the master server.
+    available on the slave.
 
 ### `register_control`
 

docs/writing-plugins.md

@@ -311,10 +311,10 @@ It will also be forwarded by slaves to a specific instance.
 The following properties are recognized by the Event constructor:
 
 **type**:
-    The message type sent over the wire.  This can be any string, but
-    it must be unique across all plugins, and it's recommended that it is
-    of format `plugin_name:message_name`.  The suffix `_event` will be
-    appended to the type.
+    The message type sent over the wire.  This must start with the name
+    of the plugin followed by colon and and be unique for the plugin.
+    The type of the message sent over the socket will have the suffix
+    `_event` appended to it.
 
 **links**:
     An array of strings describing which links this event can be sent
@@ -391,11 +391,11 @@ instance specified by `instance_id`.
 The following properties are recognized by the Request constructor:
 
 **type**:
-    The message type sent over the wire.  This can be any string, but
-    it must be unique across all plugins, and it's recommended that it is
-    of format `plugin_name:message_name`.  The suffix `_request` will be
-    appended to the type for the request message sent, and the suffix
-    `_response` will be appended to the type for the response.
+    The message type sent over the wire.  This must start with the name
+    of the plugin followed by colon and and be unique for the plugin.
+    The type of the message sent over the socket will have the suffix
+    `_request` appended to it for the request and `_response` appended
+    to it for the response.
 
 **links**:
     An array of strings describing which links this request can be sent

lib/link/connectors.js

@@ -155,13 +155,6 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 		this._reconnectDelay = reconnectDelay;
 		this._timeout = 15 * 60;
 		this._startedReconnect = null;
-
-		/**
-		 * Mapping of plugin name to plugin version loaded on the master
-		 * server.  Is `null` before handshake has taken place.
-		 * @member {?Object<string, string>}
-		 */
-		this.plugins = null;
 	}
 
 	_check(expectedStates) {
@@ -363,7 +356,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 		let { seq, type, data } = message;
 		if (type == 'hello') {
 			console.log(`SOCKET | received hello from master version ${data.version}`);
-			this.plugins = data.plugins;
+			this.emit("hello", data);
 			if (this._sessionToken) {
 				console.log("SOCKET | Attempting resume");
 				this.sendHandshake("resume", {

lib/link/messages.js

@@ -3,15 +3,36 @@
 const schema = require("lib/schema");
 const errors = require("lib/errors");
 
+/**
+ * Represents a message that can be sent over the link
+ *
+ * @memberof module:lib/link
+ */
+class Message {
+	/**
+	 * Name of the plugin this message belongs to, or null.
+	 * @returns {?string}
+	 **/
+	get plugin() {
+		if (this._plugin === undefined) {
+			let index = this.type.indexOf(":");
+			this._plugin = index === -1 ? null : this.type.slice(0, index);
+		}
+
+		return this._plugin;
+	}
+}
+
 /**
  * Represents a request sent over the link
  *
  * A request-response message that can be innitated by either party of the
  * link.
  *
+ * @extends module:lib/link.Message
  * @memberof module:lib/link
  */
-class Request {
+class Request extends Message {
 
 	/**
 	 * Creates a request
@@ -37,6 +58,7 @@ class Request {
 	constructor({
 		type, links, forwardTo = null, requestProperties = {}, responseProperties = {}
 	}) {
+		super();
 		this.type = type;
 		this.links = links;
 		this.forwardTo = forwardTo;
@@ -362,9 +384,10 @@ messages.getMetrics = new Request({
  * A one way message without any response or recipt confirmation that
  * can be innitiated by either party of the link.
  *
+ * @extends module:lib/link.Message
  * @memberof module:lib/link
  */
-class Event {
+class Event extends Message {
 
 	/**
 	 * Creates an event
@@ -390,6 +413,7 @@ class Event {
 	 *     properties that are valid in the data payload of this event.
 	 */
 	constructor({ type, links, forwardTo = null, broadcastTo = null, eventProperties = {} }) {
+		super();
 		this.type = type;
 		this.links = links;
 		this.forwardTo = forwardTo;
@@ -509,6 +533,16 @@ messages.debugWsMessage = new Event({
 	},
 });
 
+messages.instanceInitialized = new Event({
+	type: "instance_initialized",
+	links: ["instance-slave"],
+	eventProperties: {
+		"plugins": {
+			type: "object",
+			additionalProperties: { type: "string" },
+		}
+	},
+});
 messages.instanceStopped = new Event({
 	type: 'instance_stopped',
 	links: ['instance-slave'],
@@ -583,6 +617,7 @@ function attachAllMessages(link) {
 
 
 module.exports = {
+	Message,
 	Request,
 	Event,
 

lib/plugin.js

@@ -242,7 +242,10 @@ class BaseMasterPlugin {
 	 */
 	broadcastEventToSlaves(event, data={}) {
 		for (let slaveConnection of this.master.slaveConnections.values()) {
-			if (!slaveConnection.connector.closing) {
+			if (
+				!slaveConnection.connector.closing
+				&& (!event.plugin || slaveConnection.plugins.has(event.plugin))
+			) {
 				event.send(slaveConnection, data);
 			}
 		}
@@ -300,6 +303,10 @@ async function loadPluginInfos(baseDir) {
 function attachPluginMessages(link, pluginInfo, plugin) {
 	let messageDefinitions = pluginInfo.messages || [];
 	for (let [name, messageFormat] of Object.entries(messageDefinitions)) {
+		if (messageFormat.plugin !== pluginInfo.name) {
+			throw new Error(`Type of ${name} message must start with "${pluginInfo.name}:"`);
+		}
+
 		let handler = name + messageFormat.constructor.name + 'Handler';
 		if (plugin === null || !plugin[handler]) {
 			messageFormat.attach(link);

master.js

@@ -377,6 +377,9 @@ class BaseConnection extends link.Link {
 		if (!connection) {
 			throw new errors.RequestError("Slave containing instance is not connected");
 		}
+		if (request.plugin && !connection.plugins.has(request.plugin)) {
+			throw new errors.RequestError(`Slave containing instance does not have ${request.plugin} plugin`)
+		}
 
 		return await request.send(connection, message.data);
 	}
@@ -390,15 +393,17 @@ class BaseConnection extends link.Link {
 
 		let connection = slaveConnections.get(slaveId);
 		if (!connection || connection.closing) { return; }
+		if (event.plugin && !connection.plugins.has(event.plugin)) { return; }
 
 		event.send(connection, message.data);
 	}
 
 	async broadcastEventToInstance(message, event) {
 		for (let slaveConnection of slaveConnections.values()) {
-			if (slaveConnection === this || slaveConnection.connector.closing) {
-				continue; // Do not broadcast back to the source
-			}
+			// Do not broadcast back to the source
+			if (slaveConnection === this) { continue; }
+			if (slaveConnection.connector.closing) { continue; }
+			if (event.plugin && !slaveConnection.plugins.has(event.plugin)) { continue; }
 
 			event.send(slaveConnection, message.data);
 		}
@@ -600,14 +605,14 @@ class SlaveConnection extends BaseConnection {
 		this._id = registerData.id;
 		this._name = registerData.name;
 		this._version = registerData.version;
-		this._plugins = registerData.plugins;
+		this.plugins = new Map(Object.entries(registerData.plugins));
 
 		db.slaves.set(this._id, {
 			agent: this._agent,
 			id: this._id,
 			name: this._name,
 			version: this._version,
-			plugins: this._plugins,
+			plugins: registerData.plugins,
 		});
 
 		this.connector.on("close", () => {

slave.js

@@ -141,14 +141,14 @@ class Instance extends link.Link{
 		console.log(`Clusterio | Loaded plugin ${pluginInfo.name} in ${Date.now() - pluginLoadStarted}ms`);
 	}
 
-	async init(pluginInfos, masterPlugins, slave) {
+	async init(pluginInfos, slave) {
 		await this.server.init();
 
 		// load plugins
 		for (let pluginInfo of pluginInfos) {
 			if (
 				!pluginInfo.instanceEntrypoint
-				|| !Object.hasOwnProperty.call(masterPlugins, pluginInfo.name)
+				|| !slave.serverPlugins.has(pluginInfo.name)
 				|| !this.config.group(pluginInfo.name).get("enabled")
 			) {
 				continue;
@@ -161,6 +161,12 @@ class Instance extends link.Link{
 				throw err;
 			}
 		}
+
+		let plugins = {}
+		for (let [name, plugin] of this.plugins) {
+			plugins[name] = plugin.info.version;
+		}
+		link.messages.instanceInitialized.send(this, { plugins });
 	}
 
 	/**
@@ -507,6 +513,7 @@ class InstanceConnection extends link.Link {
 		super('slave', 'instance', connector);
 		this.slave = slave;
 		this.instanceId = instanceId;
+		this.plugins = new Map();
 		link.attachAllMessages(this);
 
 		for (let pluginInfo of slave.pluginInfos) {
@@ -526,6 +533,7 @@ class InstanceConnection extends link.Link {
 			await this.slave.forwardEventToMaster(message, event);
 			return;
 		}
+		if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { return; }
 
 		event.send(instanceConnection, message.data);
 	}
@@ -536,14 +544,18 @@ class InstanceConnection extends link.Link {
 
 	async broadcastEventToInstance(message, event) {
 		for (let instanceConnection of this.slave.instanceConnections.values()) {
-			if (instanceConnection === this) {
-				continue; // Do not broadcast back to the source
-			}
+			// Do not broadcast back to the source
+			if (instanceConnection === this) { continue; }
+			if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { continue; }
 
 			event.send(instanceConnection, message.data);
 		}
 	}
 
+	async instanceInitializedEventHandler(message) {
+		this.plugins = new Map(Object.entries(message.data.plugins));
+	}
+
 	async instanceStoppedEventHandler(message) {
 		this.slave.instanceConnections.delete(this.instanceId);
 	}
@@ -596,8 +608,14 @@ class Slave extends link.Link {
 		this.instanceConnections = new Map();
 		this.instanceInfos = new Map();
 
+		this.connector.on("hello", data => {
+			this.serverVersion = data.version;
+			this.serverPlugins = new Map(Object.entries(data.plugins));
+		});
+
 		this._disconnecting = false;
 		this._shuttingDown = false;
+
 		this.connector.on("close", () => {
 			if (this._shuttingDown) {
 				return;
@@ -663,6 +681,10 @@ class Slave extends link.Link {
 			throw new errors.RequestError(`Instance ID ${instanceId} is not running`);
 		}
 
+		if (request.plugin && !instanceConnection.plugins.has(request.plugin)) {
+			throw new errors.RequestError(`Instance ID ${instanceId} does not have ${request.plugin} plugin loaded`)
+		}
+
 		return await request.send(instanceConnection, message.data);
 	}
 
@@ -672,12 +694,15 @@ class Slave extends link.Link {
 
 		let instanceConnection = this.instanceConnections.get(instanceId);
 		if (!instanceConnection) { return; }
+		if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { return; }
 
 		event.send(instanceConnection, message.data);
 	}
 
 	async broadcastEventToInstance(message, event) {
 		for (let instanceConnection of this.instanceConnections.values()) {
+			if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { continue; }
+
 			event.send(instanceConnection, message.data);
 		}
 	}
@@ -736,7 +761,7 @@ class Slave extends link.Link {
 		let instance = new Instance(
 			connectionClient, instanceInfo.path, this.config.get("slave.factorio_directory"), instanceInfo.config
 		);
-		await instance.init(this.pluginInfos, this.connector.plugins, this);
+		await instance.init(this.pluginInfos, this);
 
 		// XXX: race condition on multiple simultanious calls
 		this.instanceConnections.set(instanceId, instanceConnection);

test/lib/plugin.js

@@ -96,19 +96,25 @@ describe("lib/plugin", function() {
 
 	describe("attachPluginMessages()", function() {
 		let mockLink = new link.Link('source', 'target', new mock.MockConnector());
-		let mockEvent = new link.Event({ type: 'test', links: ['target-source'] });
+		let mockEvent = new link.Event({ type: "test:test", links: ["target-source"] });
 		it("should accept pluginInfo without messages", function() {
 			plugin.attachPluginMessages(mockLink, {}, null);
 		});
 		it("should attach handler for the given message", function() {
 			let mockEventEventHandler = function() {};
-			plugin.attachPluginMessages(mockLink, { messages: { mockEvent }}, { mockEventEventHandler });
-			assert(mockLink._handlers.get('test_event'), "handler was not registered");
+			plugin.attachPluginMessages(mockLink, { name: "test", messages: { mockEvent }}, { mockEventEventHandler });
+			assert(mockLink._handlers.get("test:test_event"), "handler was not registered");
 		});
 		it("should throw if missing handler for the given message", function() {
 			assert.throws(
-				() => plugin.attachPluginMessages(mockLink, { messages: { mockEvent }}, {}),
-				new Error("Missing handler for test_event on source-target link")
+				() => plugin.attachPluginMessages(mockLink, { name: "test", messages: { mockEvent }}, {}),
+				new Error("Missing handler for test:test_event on source-target link")
+			);
+		});
+		it("should throw if message starts with the wrong prefix", function() {
+			assert.throws(
+				() => plugin.attachPluginMessages(mockLink, { name: "foo", messages: { mockEvent }}, {}),
+				new Error('Type of mockEvent message must start with "foo:"')
 			);
 		});
 	});