Merge pull request #576 from Hornwitser/fix-plugin-broadcast

Fix plugin broadcasts not being filtered
This commit is contained in:
Cooldude2606 authored and GitHub committed 2024-01-24 02:42:30 +00:00
commit a1109b9354
9 files changed
+476 -21

No files matched your search

+1
View File
@@ -17,6 +17,7 @@
- Fixed assertion error on closing or invalidating a link with messages forwarded from a virtual link.
- Fixed host hanging if stopped while an instance is in the process of creating a save during startup.
- Fixed `instance.auto_start` option.
- Fixed host log spammed with errors when plugins broadcast messages to instances with the plugin disabled.
## Version 2.0.0-alpha.14
+2
View File
@@ -1,7 +1,9 @@
import { bootstrap } from "./controller";
export { default as Controller } from "./src/Controller";
export { default as ControllerUser } from "./src/ControllerUser";
export { default as ControllerRouter } from "./src/ControllerRouter";
export { default as ControlConnection } from "./src/ControlConnection";
export { default as HostConnection } from "./src/HostConnection";
export { default as InstanceInfo } from "./src/InstanceInfo";
export { default as BaseControllerPlugin } from "./src/BaseControllerPlugin";
export { default as UserManager } from "./src/UserManager";
+1 -2
View File
@@ -6,7 +6,6 @@ import type WsServerConnector from "./WsServerConnector";
import * as lib from "@clusterio/lib";
const { logger } = lib;
import ControllerRouter from "./ControllerRouter";
import * as routes from "./routes";
@@ -24,7 +23,7 @@ export default class BaseConnection extends lib.Link {
public _controller: Controller
) {
super(connector);
this.router = new ControllerRouter(this._controller);
this.router = _controller.router;
for (let [Request, handler] of this._controller._registeredRequests) { this.handle(Request, handler); }
for (let [Request, handler] of this._controller._fallbackedRequests) { this.fallbackRequest(Request, handler); }
for (let [Event, handler] of this._controller._registeredEvents) { this.handle(Event, handler); }
+6
View File
@@ -26,6 +26,7 @@ import WsServer from "./WsServer";
import HostConnection from "./HostConnection";
import HostInfo from "./HostInfo";
import BaseControllerPlugin from "./BaseControllerPlugin";
import ControllerRouter from "./ControllerRouter";
const endpointDurationSummary = new Summary(
"clusterio_controller_http_endpoint_duration_seconds",
@@ -74,6 +75,7 @@ export default class Controller {
/** WebSocket server */
wsServer: WsServer;
router = new ControllerRouter(this);
trustedProxies: BlockList;
debugEvents: events.EventEmitter = new events.EventEmitter();
private _events: events.EventEmitter = new events.EventEmitter();
@@ -1374,7 +1376,11 @@ export default class Controller {
}
} else if (dst.id === lib.Address.instance || dst.id === lib.Address.host) {
const plugin = event.constructor.plugin;
for (let hostConnection of this.wsServer.hostConnections.values()) {
if (plugin && !hostConnection.plugins.has(plugin)) {
continue;
}
hostConnection.sendEvent(event, dst);
}
+9 -7
View File
@@ -7,8 +7,6 @@ import ControlConnection from "./ControlConnection";
const { logger } = lib;
export default class ControllerRouter {
pendingRequests: Map<any, any> = new Map();
constructor(
public controller: Controller
) {}
@@ -16,12 +14,14 @@ export default class ControllerRouter {
/**
* @param origin - Source link of the message.
* @param message - Link the message originated from.
* @param entry - message entry if message is a request or an event.
* @param hasFallback - true if fallback handling is available.
* @returns true if the message was handled
*/
forwardMessage(
origin: ControlConnection | HostConnection,
message: lib.MessageRoutable,
entry: lib.RequestEntry | lib.EventEntry | undefined,
hasFallback: boolean
): boolean {
if (!["request", "response", "responseError", "event"].includes(message.type)) {
@@ -32,7 +32,7 @@ export default class ControllerRouter {
let nextHop: ControlConnection | HostConnection | undefined;
let msg: string = "";
if (dst.type === lib.Address.broadcast) {
this.broadcastMessage(origin, message);
this.broadcastMessage(origin, message, entry as lib.EventEntry);
return true;
} else if (dst.type === lib.Address.host) {
nextHop = this.controller.wsServer.hostConnections.get(dst.id);
@@ -96,13 +96,17 @@ export default class ControllerRouter {
broadcastMessage(
origin: lib.Link,
message: lib.MessageRoutable,
entry: lib.EventEntry,
) {
let dst = message.dst;
if (message.type !== "event") {
this.warnUnrouted(message, `Unexpected broadcast of ${message.type}`);
} else if (dst.id === lib.Address.host || dst.id === lib.Address.instance) {
return;
}
const plugin = entry.Event.plugin;
if (dst.id === lib.Address.host || dst.id === lib.Address.instance) {
for (let hostConnection of this.controller.wsServer.hostConnections.values()) {
if (hostConnection !== origin) {
if (hostConnection !== origin && (!plugin || hostConnection.plugins.has(plugin))) {
hostConnection.connector.send(message);
}
}
@@ -126,5 +130,3 @@ export default class ControllerRouter {
logger.warn(msg ? `${baseMsg}: ${msg}.` : `${baseMsg}.`);
}
}
module.exports = ControllerRouter;
+3
View File
@@ -1,6 +1,9 @@
import { bootstrap } from "./host";
export { default as BaseHostPlugin } from "./src/BaseHostPlugin";
export { default as BaseInstancePlugin } from "./src/BaseInstancePlugin";
export { default as Host, HostRouter } from "./src/Host";
export { default as Instance } from "./src/Instance";
export { default as InstanceConnection } from "./src/InstanceConnection";
if (module === require.main) {
bootstrap();
+25 -10
View File
@@ -82,7 +82,7 @@ const instanceStartingMessages = new Set([
lib.InstanceExportDataRequest.name,
]);
class HostRouter {
export class HostRouter {
constructor(
public host: Host
) { }
@@ -93,16 +93,22 @@ class HostRouter {
* @param origin -
* Link the message originated from.
* @param message - Message to process.
* @param entry - message entry if message is a request or an event.
* @param hasFallback - is fallback available
* @returns true if the message was handled, false if fallback
* is requested
*/
forwardMessage(origin: lib.Link, message: lib.MessageRoutable, hasFallback: boolean) {
forwardMessage(
origin: lib.Link,
message: lib.MessageRoutable,
entry: lib.RequestEntry | lib.EventEntry | undefined,
hasFallback: boolean,
) {
let dst = message.dst;
let nextHop: lib.Link | undefined;
let msg: string | undefined;
if (dst.type === lib.Address.broadcast) {
this.broadcastMessage(origin, message);
this.broadcastMessage(origin, message, entry as lib.EventEntry);
return true;
} else if (
dst.type === lib.Address.controller
@@ -153,27 +159,36 @@ class HostRouter {
return true;
}
broadcastMessage(origin: lib.Link, message: lib.MessageRoutable) {
broadcastMessage(
origin: lib.Link,
message: lib.MessageRoutable,
entry: lib.EventEntry,
) {
let dst = message.dst;
if (message.type !== "event") {
this.warnUnrouted(message, `Unexpected broadcast of ${message.type}`);
} else if (dst.id === lib.Address.host || dst.id === lib.Address.instance) {
return;
}
const plugin = entry.Event.plugin;
if (dst.id === lib.Address.instance) {
for (let instanceConnection of this.host.instanceConnections.values()) {
if (instanceConnection !== origin) {
if (instanceConnection !== origin && (!plugin || instanceConnection.plugins.has(plugin))) {
instanceConnection.connector.send(message);
}
}
if (this.host !== origin) {
if (this.host !== origin && (!plugin || this.host.serverPlugins.has(plugin))) {
this.host.connector.send(message);
}
} else if (dst.id === lib.Address.control) {
if (this.host !== origin) {
this.host.connector.send(message);
if (!plugin || this.host.serverPlugins.has(plugin)) {
this.host.connector.send(message);
}
} else {
logger.warn(`Received control broadcast of ${(message as lib.MessageEvent).name} from master.`);
logger.warn(`Received control broadcast of ${(message as lib.MessageEvent).name} from controller.`);
}
} else {
this.warnUnrouted(message, `Unexpected broacdast target ${dst.id}`);
this.warnUnrouted(message, `Unexpected broadcast target ${dst.id}`);
}
}
+7 -2
View File
@@ -72,7 +72,12 @@ export type RequestHandler<Req, Res> = (request: Req, src: libData.Address, dst:
export type EventHandler<T> = (event: T, src: libData.Address, dst: libData.Address) => Promise<void>;
interface Router {
forwardMessage(origin: Link, message: MessageRoutable, fallback: boolean): boolean,
forwardMessage(
origin: Link,
message: MessageRoutable,
entry: RequestEntry | EventEntry | undefined,
fallback: boolean,
): boolean,
}
// Some definitions for the terminology used here:
@@ -343,7 +348,7 @@ export class Link {
let fallback =
message.type === "request" ? this._requestFallbacks.get((entry as RequestEntry).Request) : undefined
;
if (this.router.forwardMessage(this, message, Boolean(fallback))) {
if (this.router.forwardMessage(this, message, entry, Boolean(fallback))) {
return;
}
if (!fallback) {
+422
View File
@@ -0,0 +1,422 @@
"use strict";
const assert = require("assert").strict;
const lib = require("@clusterio/lib");
const { Host, Instance, InstanceConnection } = require("@clusterio/host");
const { ControlConnection, Controller, HostConnection, InstanceInfo } = require("@clusterio/controller");
const addr = lib.Address.fromShorthand;
class Control extends lib.Link { }
/**
* @param {Controller} controller
* @param {number} hostId
*/
function connectHost(controller, hostId, plugins) {
const [controllerSide, hostSide] = lib.VirtualConnector.makePair(addr("controller"), addr({ hostId }));
const registerData = new lib.RegisterHostData("", "0.0.0", hostId, plugins);
const hostConnection = new HostConnection(registerData, controllerSide, controller, "host-a.test");
controller.wsServer.hostConnections.set(hostId, hostConnection);
const hostConfig = new lib.HostConfig("host");
hostConfig.set("host.id", hostId);
return new Host(hostSide, "invalid", hostConfig, undefined, []);
}
/**
* @param {Controller} controller
* @param {Host} host
* @param {number} instanceId
*/
function connectInstance(controller, host, instanceId) {
const hostAddress = addr({ hostId: host.config.get("host.id")});
const instanceAddress = addr({ instanceId });
const [instanceSide, hostSide] = lib.VirtualConnector.makePair(instanceAddress, hostAddress);
const instanceConfig = new lib.InstanceConfig("host");
instanceConfig.set("instance.id", instanceId);
instanceConfig.set("instance.assigned_host", host.config.get("host.id"));
host.instanceInfos.set(instanceId, { path: "invalid", config: instanceConfig });
controller.instances.set(instanceId, new InstanceInfo(instanceConfig, "stopped"));
const instance = new Instance(host, instanceSide, "invalid", "invalid", instanceConfig);
const instanceConnection = new InstanceConnection(hostSide, host, instance);
host.instanceConnections.set(instanceId, instanceConnection);
return instance;
}
/**
* @param {Controller} controller
* @param {number} hostId
*/
function connectControl(controller, controlId) {
const [controllerSide, controlSide] = lib.VirtualConnector.makePair(addr("controller"), addr({ controlId }));
const registerData = new lib.RegisterControlData("", "0.0.0");
let user = controller.userManager.users.get("test");
if (!user) {
user = controller.userManager.createUser("test");
}
const controlConnection = new ControlConnection(registerData, controllerSide, controller, user, controlId);
controller.wsServer.controlConnections.set(controlId, controlConnection);
return new Control(controlSide);
}
class TestRequest {
static type = "request";
static src = ["controller", "host", "instance", "control"];
static dst = ["controller", "host", "instance", "control"];
static permission = null;
}
class TestEvent {
static type = "event";
static src = ["controller", "host", "instance", "control"];
static dst = ["controller", "host", "instance", "control"];
static permission = null;
}
class PluginEvent {
static type = "event";
static src = ["controller", "host", "instance", "control"];
static dst = ["controller", "host", "instance", "control"];
static permission = null;
static plugin = "test";
}
lib.Link.register(TestRequest);
lib.Link.register(TestEvent);
lib.Link.register(PluginEvent);
describe("Integration of link routing", function() {
/** @type {Controller} */
let controller;
/** @type {Host} */
let hostA;
/** @type {Host} */
let hostB;
/** @type {Instance} */
let instanceA1;
/** @type {Instance} */
let instanceA2;
/** @type {Instance} */
let instanceB1;
/** @type {Control} */
let controlA;
/** @type {Control} */
let controlB;
beforeEach(function() {
controller = new Controller(
{},
[],
"invalid",
new lib.ControllerConfig("controller")
);
hostA = connectHost(controller, 10, []);
hostB = connectHost(controller, 20, []);
instanceA1 = connectInstance(controller, hostA, 11);
instanceA2 = connectInstance(controller, hostA, 12);
instanceB1 = connectInstance(controller, hostB, 21);
controlA = connectControl(controller, 100);
controlB = connectControl(controller, 101);
});
/** @returns {Controller | Host | Instance} */
function get(name) {
return {
controller,
hostA,
hostB,
instanceA1,
instanceA2,
instanceB1,
controlA,
controlB,
}[name];
}
// Combinations to try. Equivalent paths have been removed.
const parties = [
["controller", ["hostA", "instanceA1", "controlA"]],
["hostA", ["controller", "hostB", "instanceA1", "instanceB1", "controlA"]],
["instanceA1", ["controller", "hostA", "hostB", "instanceA2", "instanceB1", "controlA"]],
["controlA", ["controller", "hostA", "instanceA1", "controlB"]],
];
function handle(dst, Message, handler) {
if (dst instanceof Controller) {
for (const host of dst.wsServer.hostConnections.values()) {
host.handle(Message, handler);
}
for (const control of dst.wsServer.controlConnections.values()) {
control.handle(Message, handler);
}
} else if (dst instanceof Host) {
for (const instance of dst.instanceConnections.values()) {
instance.handle(Message, handler);
}
dst.handle(Message, handler);
} else {
dst.handle(Message, handler);
}
}
/**
* @param {Controller | Host | Instance} link
* @param {Address} dst
*/
function send(link, dst, message) {
// This logic exists becasue .sendTo doesn't forward all kinds of
// messages to the correct location from all places. Idealy this
// would not be neccessary and .sendTo would do the right thing.
if (link instanceof Host) {
const instance = link.instanceConnections.get(dst.id);
if (instance) {
return instance.sendTo(dst, message);
}
}
return link.sendTo(dst, message);
}
describe("request routing", function() {
for (const [srcName, dstNames] of parties) {
it(`should route from ${srcName}`, async function() {
for (const dstName of dstNames) {
let called;
const dst = get(dstName);
handle(dst, TestRequest, async () => { called = true; });
const dstAddr = dst instanceof Controller ? addr("controller") : dst.connector.src;
await send(get(srcName), dstAddr, new TestRequest());
assert(called, `${dstName} handler was not called`);
}
});
}
});
describe("event routing", function() {
for (const [srcName, dstNames] of parties) {
it(`should route from ${srcName}`, async function() {
for (const dstName of dstNames) {
let called;
const dst = get(dstName);
handle(dst, TestEvent, async () => { called = true; });
const dstAddr = dst instanceof Controller ? addr("controller") : dst.connector.src;
send(get(srcName), dstAddr, new TestEvent());
assert(called, `${dstName} handler was not called`);
}
});
}
});
/**
* @param {Controller | Host | Instance | Control} link
* @param {"allHosts" | "allInstances" | "allControls"} dst
*/
function bcast(link, dst, message) {
// This logic exists becasue .sendTo doesn't forward all kinds of
// messages to the correct location from all places. Idealy this
// would not be neccessary and .sendTo would do the right thing.
if (link instanceof Host) {
const plugin = message.constructor.plugin;
if (dst === "allInstances") {
for (const instance of link.instanceConnections.values()) {
if (plugin && !instance.plugins.has(plugin)) {
continue;
}
instance.sendTo(dst, message);
}
}
if (plugin && !link.serverPlugins.has(plugin)) {
return;
}
}
link.sendTo(dst, message);
}
function bhandle(srcName, dst, Event, canReach = (dstB) => true) {
const asserts = [];
const dsts = {
allHosts: ["hostA", "hostB"],
allInstances: ["instanceA1", "instanceA2", "instanceB1"],
allControls: ["controlA", "controlB"],
};
for (const dstName of dsts[dst]) {
if (srcName === dstName) { continue; }
// TODO remove when broadcasting from instance to hosts is fixed, see #575
if (srcName === "instanceA1" && dstName === "hostB") { continue; }
let called;
handle(get(dstName), Event, async () => { called = true; });
const should = canReach(dstName);
asserts.push(() => assert(
!should ^ called,
`${dstName} handler was ${should ? "not " : ""}called from ${srcName}`,
));
}
return asserts;
}
describe("event broadcast", function() {
for (const srcName of ["controller", "hostA", "instanceA1", "controlA"]) {
it(`should broadcast from ${srcName}`, async function() {
for (const dstName of ["allHosts", "allInstances", "allControls"]) {
const dstAsserts = bhandle(srcName, dstName, TestEvent);
bcast(get(srcName), dstName, new TestEvent());
for (const dstAssert of dstAsserts) {
dstAssert();
}
}
});
}
});
const graph = {
controller: ["hostA", "hostB", "controlA", "controlB"],
hostA: ["controller", "instanceA1", "instanceA2"],
hostB: ["controller", "instanceB1"],
instanceA1: ["hostA"],
instanceA2: ["hostA"],
instanceB1: ["hostB"],
controlA: ["controller"],
controlB: ["controller"],
};
// Returns true if the path from src to dst goes through via
function goesVia(src, dst, via) {
// Places we've been to
const checked = new Set();
// Places left to check
const paths = new Set([src]);
while (paths.size) {
const [candidate] = paths;
paths.delete(candidate);
if (candidate === via) {
continue;
}
if (checked.has(candidate)) {
continue;
}
if (candidate === dst) {
return false;
}
checked.add(candidate);
for (const newCandidate of graph[candidate]) {
paths.add(newCandidate);
}
}
return true;
}
assert(goesVia("instanceA1", "instanceB1", "controller"));
assert(!goesVia("instanceA1", "instanceA2", "controller"));
assert(goesVia("instanceA1", "hostB", "hostA"));
assert(!goesVia("instanceA1", "hostB", "controlA"));
assert(goesVia("controlA", "controlB", "controller"));
assert(goesVia("controlA", "hostA", "controller"));
function addPluginExceptFor(exName) {
const ex = get(exName);
const exAddr = ex instanceof Controller ? addr("controller") : ex.connector.src;
for (const name of [
"controller", "hostA", "hostB", "instanceA1", "instanceA2", "instanceB1", "controlA", "controlB",
]) {
const party = get(name);
if (party instanceof Controller) {
for (const host of party.wsServer.hostConnections.values()) {
if (!host.connector.dst.equals(exAddr)) {
host.plugins.set("test", "0.0.0");
}
}
for (const control of party.wsServer.controlConnections.values()) {
if (!control.connector.dst.equals(exAddr)) {
// No way to set plugins
}
}
} else if (party instanceof Host) {
for (const instance of party.instanceConnections.values()) {
if (!instance.connector.dst.equals(exAddr)) {
instance.plugins.set("test", "0.0.0");
}
}
if (!party.connector.dst.equals(exAddr)) {
party.serverPlugins.set("test", "0.0.0");
}
} else if (party instanceof Instance) {
// Currently does not handle plugins
} else if (party instanceof Control) {
// Currently does not handle plugins
}
}
}
// When plugin events are broadcast the party sending the event should
// filter it out if the next hop it is forwarded to does not have plugin
// the event was defined in loaded.
describe("event broadcast with controller missing plugin", function() {
for (const srcName of ["hostA", "instanceA1", "controlA"]) {
it(`should broadcast from ${srcName}`, async function() {
if (srcName === "controlA") {
this.skip(); // Not implemented
}
addPluginExceptFor("controller");
for (const dstName of ["allHosts", "allInstances", "allControls"]) {
const dstAsserts = bhandle(
srcName, dstName, PluginEvent, dstB => !goesVia(srcName, dstB, "controller"),
);
bcast(get(srcName), dstName, new PluginEvent());
for (const dstAssert of dstAsserts) {
dstAssert();
}
}
});
}
});
describe("event broadcast with host missing plugin", function() {
for (const srcName of ["controller", "hostA", "instanceA1", "controlA"]) {
it(`should broadcast from ${srcName}`, async function() {
addPluginExceptFor("hostB");
for (const dstName of ["allHosts", "allInstances", "allControls"]) {
const dstAsserts = bhandle(
srcName, dstName, PluginEvent, dstB => !goesVia(srcName, dstB, "hostB"),
);
bcast(get(srcName), dstName, new PluginEvent());
for (const dstAssert of dstAsserts) {
dstAssert();
}
}
});
}
});
describe("event broadcast with instance missing plugin", function() {
for (const srcName of ["controller", "hostA", "instanceA1", "controlA"]) {
it(`should broadcast from ${srcName}`, async function() {
addPluginExceptFor("instanceA2");
for (const dstName of ["allHosts", "allInstances", "allControls"]) {
const dstAsserts = bhandle(
srcName, dstName, PluginEvent, dstB => !goesVia(srcName, dstB, "instanceA2"),
);
bcast(get(srcName), dstName, new PluginEvent());
for (const dstAssert of dstAsserts) {
dstAssert();
}
}
});
}
});
describe("event broadcast with control missing plugin", function() {
for (const srcName of ["controller", "hostA", "instanceA1", "controlA"]) {
it(`should broadcast from ${srcName}`, async function() {
this.skip(); // Currently not implemented.
addPluginExceptFor("controlB");
for (const dstName of ["allHosts", "allInstances", "allControls"]) {
const dstAsserts = bhandle(
srcName, dstName, PluginEvent, dstB => !goesVia(srcName, dstB, "controlB"),
);
bcast(get(srcName), dstName, new PluginEvent());
for (const dstAssert of dstAsserts) {
dstAssert();
}
}
});
}
});
});