aperx/factorio-clusterio
1
0 Fork 0
mirror of https://github.com/clusterio/clusterio.git synced 2026-11-15 15:24:52 +09:00
Code Issues Packages Projects Releases Wiki Activity

Refactor link system to JSON serialisable class pattern

Browse Source 
Reverse the direction of the API for sending messages from

    messageDefinition.send(link, { content });

to

    link.send(new MessageClass(content));

and convert all message definitions to use the JSON serialisable class
pattern and explicit registration of message handlers, along with a
end-to-end based routing of messages instead of link to link.

The experience I gained from the existing way messages were defined and
sent over links was that it was difficult to understand for people
unfamiliar with the system.  The automatic registration of message
handlers ment that code was being called as if by magic from nowhere
with no easy way to discover it.  The manual forwarding and rules for
what can be sent where and how obscured its working.

The use of pure JSON as both input and output from message handlers made
it tedious to work with.  Data had to be manually converted between the
program representation and the over the wire format, which made handler
code unplesant to write.  Particularly because the over the wire format
used snake_case while the program code uses camelCase for variables.
The idea behind the decision to do it this way was to make it more clear
which data came from the wire and which from the program, and while this
worked it didn't make message handlers any nicer to write.

The direction of the API and magic registration of message handlers also
meant that a future migration to TypeScript would be troublesome.  While
it may be possible to implement, the degree of TypeScript magic required
would go starkly against the principle of keep it simple stupid.

By refactoring to using the JSON serialisable class pattern most of
these issues go away.  There is no longer a need to distinguish the on
wire format of messages with the program representation because the only
place that deals with the wire format of messages are the toJSON and
fromJSON methods on the classes themselves.  Message handlers will work
with proper instances of classes instead of raw data, something that
makes them easier to write, and trivial to type with TypeScript.

End-to-end routing also simplifies sending messages that need to do
mulitple hops to reach the destination.  And removes the need for having
all messages be specified in plugin's info module, as they can be routed
without having to be parsed completely.

The explicit registration of message handlers is something that I
thought would be tedious boilerplate code that could be automated away.
But experience now shows it to be the opposite.  Allthough dull
repetition, the purpose it serves is to link together the code in a way
that humans can easily find.  With explicit registration searching the
name of the message handler in the editor will show the registration that
ties it together with the type of message it handles and the kind of
registration.  With implicit registration searching the name of the
message handler turns up only the handler with no clue as to what calls
it.
This commit is contained in:
Hornwitser committed 2023-04-23 13:19:46 +02:00
1 parent 9a967a5779
commit bcd94709d2
106 files changed
+8173 -4928
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/controller/src/BaseConnection.js
+18 -56
View File
@@ -1,8 +1,10 @@
"use strict";
const libData = require("@clusterio/lib/data");
const libErrors = require("@clusterio/lib/errors");
const libLink = require("@clusterio/lib/link");
const { logger } = require("@clusterio/lib/logging");
const libPlugin = require("@clusterio/lib/plugin");
const ControllerRouter = require("./ControllerRouter");
/**
@@ -12,68 +14,28 @@ const libPlugin = require("@clusterio/lib/plugin");
* @alias module:controller/src/BaseConnection
*/
class BaseConnection extends libLink.Link {
constructor(target, connector, controller) {
constructor(connector, controller) {
/** @member {module:controller/src/WsServerConnector} module:controller/src/BaseConnection#connector */
super("controller", target, connector);
super(connector);
this.router = new ControllerRouter(controller);
this._controller = controller;
this._disconnecting = false;
libLink.attachAllMessages(this);
for (let controllerPlugin of this._controller.plugins.values()) {
libPlugin.attachPluginMessages(this, controllerPlugin);
}
}
for (let [Request, handler] of controller._registeredRequests) { this.register(Request, handler); }
for (let [Request, handler] of controller._fallbackedRequests) { this.fallbackRequest(Request, handler); }
for (let [Event, handler] of controller._registeredEvents) { this.register(Event, handler); }
for (let [Event, handler] of controller._snoopedEvents) { this.snoopEvent(Event, handler); }
async forwardRequestToInstance(message, request) {
return await this._controller.forwardRequestToInstance(request, message.data);
}
async forwardEventToInstance(message, event) {
let instance = this._controller.instances.get(message.data.instance_id);
if (!instance) { return; }
let hostId = instance.config.get("instance.assigned_host");
if (hostId === null) { return; }
let connection = this._controller.wsServer.hostConnections.get(hostId);
if (!connection || connection.closing) { return; }
if (event.plugin && !connection.plugins.has(event.plugin)) { return; }
event.send(connection, message.data);
}
async broadcastEventToHosts(message, event) {
for (let hostConnection of this._controller.wsServer.hostConnections.values()) {
// Do not broadcast back to the source
if (hostConnection === this) { continue; }
if (hostConnection.connector.closing) { continue; }
if (event.plugin && !hostConnection.plugins.has(event.plugin)) { continue; }
event.send(hostConnection, message.data);
}
}
async broadcastEventToInstance(message, event) {
await this.broadcastEventToHosts(message, event);
}
async prepareDisconnectRequestHandler(message, request) {
await libPlugin.invokeHook(this._controller.plugins, "onPrepareHostDisconnect", this);
this._disconnecting = true;
this.connector.setClosing();
return await super.prepareDisconnectRequestHandler(message, request);
this.register(libData.ModPackGetRequest, this.handleModPackGetRequest.bind(this));
this.register(libData.ModPackGetDefaultRequest, this.handleModPackGetDefaultRequest.bind(this));
}
async disconnect(code, reason) {
this.connector.setClosing();
try {
await libLink.messages.prepareDisconnect.send(this);
await this.connector.disconnect();
} catch (err) {
if (!(err instanceof libErrors.SessionLost)) {
logger.error(`"Unexpected error preparing disconnect:\n${err.stack}`);
}
}
await this.connector.close(code, reason);
}
/**
@@ -82,19 +44,19 @@ class BaseConnection extends libLink.Link {
* @type {boolean}
*/
get connected() {
return !this._disconnecting && this.connector.connected;
return this.connector.connected;
}
async getModPackRequestHandler(message) {
let { id } = message.data;
async handleModPackGetRequest(request) {
let { id } = request;
let modPack = this._controller.modPacks.get(id);
if (!modPack) {
throw new libErrors.RequestError(`Mod pack with ID ${id} does not exist`);
}
return { mod_pack: modPack.toJSON() };
return modPack;
}
async getDefaultModPackRequestHandler(message) {
async handleModPackGetDefaultRequest() {
let id = this._controller.config.get("controller.default_mod_pack_id");
if (id === null) {
throw new libErrors.RequestError("Default mod pack not set on controller");
@@ -103,7 +65,7 @@ class BaseConnection extends libLink.Link {
if (!modPack) {
throw new libErrors.RequestError(`Default mod pack configured (${id}) does not exist`);
}
return { mod_pack: modPack.toJSON() };
return modPack;
}
}
packages/controller/src/ControlConnection.js
+331 -264
View File
@@ -30,11 +30,13 @@ const queryLogTime = new libPrometheus.Summary(
* @alias module:controller/src/ControlConnection
*/
class ControlConnection extends BaseConnection {
constructor(registerData, connector, controller, user) {
super("control", connector, controller);
constructor(registerData, connector, controller, user, id) {
super(connector, controller);
this._agent = registerData.agent;
this._version = registerData.version;
this.id = id;
this.dst = new libData.Address(libData.Address.instance, id);
/**
* The user making this connection.
@@ -44,27 +46,27 @@ class ControlConnection extends BaseConnection {
this.hostSubscriptions = {
all: false,
host_ids: [],
hostIds: [],
};
this.instanceSubscriptions = {
all: false,
instance_ids: [],
instanceIds: [],
};
this.saveListSubscriptions = {
all: false,
instance_ids: [],
instanceIds: [],
};
this.modPackSubscriptions = {
all: false,
mod_pack_ids: [],
modPackIds: [],
};
this.modSubscriptions = {
all: false,
mod_names: [],
modNames: [],
};
this.userSubscriptions = {
@@ -76,8 +78,8 @@ class ControlConnection extends BaseConnection {
this.logSubscriptions = {
all: false,
controller: false,
host_ids: [],
instance_ids: [],
hostIds: [],
instanceIds: [],
};
this.ws_dumper = null;
@@ -101,183 +103,245 @@ class ControlConnection extends BaseConnection {
}
});
}
this.register(libData.ControllerConfigGetRequest, this.handleControllerConfigGetRequest.bind(this));
this.register(libData.ControllerConfigSetFieldRequest, this.handleControllerConfigSetFieldRequest.bind(this));
this.register(libData.ControllerConfigSetPropRequest, this.handleControllerConfigSetPropRequest.bind(this));
this.register(libData.HostListRequest, this.handleHostListRequest.bind(this));
this.register(libData.HostSetSubscriptionsRequest, this.handleHostSetSubscriptionsRequest.bind(this));
this.register(libData.HostGenerateTokenRequest, this.handleHostGenerateTokenRequest.bind(this));
this.register(libData.HostConfigCreateRequest, this.handleHostConfigCreateRequest.bind(this));
this.register(libData.InstanceDetailsGetRequest, this.handleInstanceDetailsGetRequest.bind(this));
this.register(libData.InstanceDetailsListRequest, this.handleInstanceDetailsListRequest.bind(this));
this.register(
libData.InstanceDetailsSetSubscriptionsRequest, this.handleInstanceDetailsSetSubscriptionsRequest.bind(this)
);
this.register(libData.InstanceCreateRequest, this.handleInstanceCreateRequest.bind(this));
this.register(libData.InstanceDeleteRequest, this.handleInstanceDeleteRequest.bind(this));
this.register(libData.InstanceConfigGetRequest, this.handleInstanceConfigGetRequest.bind(this));
this.register(libData.InstanceConfigSetFieldRequest, this.handleInstanceConfigSetFieldRequest.bind(this));
this.register(libData.InstanceConfigSetPropRequest, this.handleInstanceConfigSetPropRequest.bind(this));
this.register(libData.InstanceAssignRequest, this.handleInstanceAssignRequest.bind(this));
this.register(
libData.InstanceSetSaveListSubscriptionsRequest,
this.handleInstanceSetSaveListSubscriptionsRequest.bind(this)
);
this.register(
libData.InstanceRenameSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller)
);
this.register(libData.InstanceCopySaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller));
this.register(
libData.InstanceDeleteSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller)
);
this.register(libData.InstanceDownloadSaveRequest, this.handleInstanceDownloadSaveRequest.bind(this));
this.register(libData.InstanceTransferSaveRequest, this.handleInstanceTransferSaveRequest.bind(this));
this.register(libData.ModPackListRequest, this.handleModPackListRequest.bind(this));
this.register(libData.ModPackSetSubscriptionsRequest, this.handleModPackSetSubscriptionsRequest.bind(this));
this.register(libData.ModPackCreateRequest, this.handleModPackCreateRequest.bind(this));
this.register(libData.ModPackUpdateRequest, this.handleModPackUpdateRequest.bind(this));
this.register(libData.ModPackDeleteRequest, this.handleModPackDeleteRequest.bind(this));
this.register(libData.ModGetRequest, this.handleModGetRequest.bind(this));
this.register(libData.ModListRequest, this.handleModListRequest.bind(this));
this.register(libData.ModSearchRequest, this.handleModSearchRequest.bind(this));
this.register(libData.ModSetSubscriptionsRequest, this.handleModSetSubscriptionsRequest.bind(this));
this.register(libData.ModDownloadRequest, this.handleModDownloadRequest.bind(this));
this.register(libData.ModDeleteRequest, this.handleModDeleteRequest.bind(this));
this.register(libData.LogSetSubscriptionsRequest, this.handleLogSetSubscriptionsRequest.bind(this));
this.register(libData.LogQueryRequest, this.handleLogQueryRequest.bind(this));
this.register(libData.PermissionListRequest, this.handlePermissionListRequest.bind(this));
this.register(libData.RoleListRequest, this.handleRoleListRequest.bind(this));
this.register(libData.RoleCreateRequest, this.handleRoleCreateRequest.bind(this));
this.register(libData.RoleUpdateRequest, this.handleRoleUpdateRequest.bind(this));
this.register(
libData.RoleGrantDefaultPermissionsRequest, this.handleRoleGrantDefaultPermissionsRequest.bind(this)
);
this.register(libData.RoleDeleteRequest, this.handleRoleDeleteRequest.bind(this));
this.register(libData.UserGetRequest, this.handleUserGetRequest.bind(this));
this.register(libData.UserListRequest, this.handleUserListRequest.bind(this));
this.register(libData.UserSetSubscriptionsRequest, this.handleUserSetSubscriptionsRequest.bind(this));
this.register(libData.UserCreateRequest, this.handleUserCreateRequest.bind(this));
this.register(libData.UserRevokeTokenRequest, this.handleUserRevokeTokenRequest.bind(this));
this.register(libData.UserUpdateRolesRequest, this.handleUserUpdateRolesRequest.bind(this));
this.register(libData.UserSetAdminRequest, this.handleUserSetAdminRequest.bind(this));
this.register(libData.UserSetBannedRequest, this.handleUserSetBannedRequest.bind(this));
this.register(libData.UserSetWhitelistedRequest, this.handleUserSetWhitelistedRequest.bind(this));
this.register(libData.UserDeleteRequest, this.handleUserDeleteRequest.bind(this));
this.register(libData.DebugDumpWsRequest, this.handleDebugDumpWsRequest.bind(this));
}
async getControllerConfigRequestHandler() {
return { serialized_config: this._controller.config.serialize("control") };
async handleControllerConfigGetRequest() {
return new libData.RawConfig(this._controller.config.serialize("control"));
}
async setControllerConfigFieldRequestHandler(message) {
this._controller.config.set(message.data.field, message.data.value, "control");
async handleControllerConfigSetFieldRequest(request) {
this._controller.config.set(request.field, request.value, "control");
}
async setControllerConfigPropRequestHandler(message) {
let { field, prop, value } = message.data;
async handleControllerConfigSetPropRequest(request) {
let { field, prop, value } = request;
this._controller.config.setProp(field, prop, value, "control");
}
async listHostsRequestHandler(message) {
async handleHostListRequest() {
let list = [];
for (let host of this._controller.hosts.values()) {
list.push({
agent: host.agent,
version: host.version,
id: host.id,
name: host.name,
public_address: host.public_address || null,
connected: this._controller.wsServer.hostConnections.has(host.id),
});
list.push(new libData.HostDetails(
host.agent,
host.version,
host.name,
host.id,
this._controller.wsServer.hostConnections.has(host.id),
host.public_address,
));
}
return { list };
return list;
}
async setHostSubscriptionsRequestHandler(message) {
this.hostSubscriptions = message.data;
async handleHostSetSubscriptionsRequest(request) {
this.hostSubscriptions = { ...request };
}
hostUpdated(host, update) {
if (
this.hostSubscriptions.all
|| this.hostSubscriptions.host_ids.includes(host.id)
|| this.hostSubscriptions.hostIds.includes(host.id)
) {
libLink.messages.hostUpdate.send(this, update);
this.send(new libData.HostUpdateEvent(update));
}
}
async generateHostTokenRequestHandler(message) {
let hostId = message.data.host_id;
async handleHostGenerateTokenRequest(message) {
let hostId = message.host_id;
if (hostId === null) {
hostId = Math.random() * 2**31 | 0;
}
return { token: this._controller.generateHostToken(hostId) };
return this._controller.generateHostToken(hostId);
}
async createHostConfigRequestHandler(message) {
async handleHostConfigCreateRequest(request) {
let hostConfig = new libConfig.HostConfig("control");
await hostConfig.init();
hostConfig.set("host.controller_url", this._controller.getControllerUrl());
if (message.data.id !== null) {
hostConfig.set("host.id", message.data.id);
if (request.id !== null) {
hostConfig.set("host.id", request.id);
}
if (message.data.name !== null) {
hostConfig.set("host.name", message.data.name);
if (request.name !== null) {
hostConfig.set("host.name", request.name);
}
if (message.data.generate_token) {
if (request.generateToken) {
this.user.checkPermission("core.host.generate_token");
hostConfig.set("host.controller_token", this._controller.generateHostToken(hostConfig.get("host.id")));
}
return { serialized_config: hostConfig.serialize() };
return new libData.RawConfig(hostConfig.serialize());
}
async getInstanceRequestHandler(message) {
let id = message.data.id;
let instance = this._controller.getRequestInstance(id);
async handleInstanceDetailsGetRequest(request) {
let instance = this._controller.getRequestInstance(request.instanceId);
return {
id,
name: instance.config.get("instance.name"),
assigned_host: instance.config.get("instance.assigned_host"),
game_port: instance.game_port || null,
status: instance.status,
};
return new libData.InstanceDetails(
instance.config.get("instance.name"),
instance.id,
instance.config.get("instance.assigned_host"),
instance.game_port || null,
instance.status,
);
}
async listInstancesRequestHandler(message) {
async handleInstanceDetailsListRequest() {
let list = [];
for (let instance of this._controller.instances.values()) {
list.push({
id: instance.id,
name: instance.config.get("instance.name"),
assigned_host: instance.config.get("instance.assigned_host"),
game_port: instance.game_port || null,
status: instance.status,
});
list.push(new libData.InstanceDetails(
instance.config.get("instance.name"),
instance.id,
instance.config.get("instance.assigned_host"),
instance.game_port || null,
instance.status,
));
}
return { list };
return list;
}
async setInstanceSubscriptionsRequestHandler(message) {
this.instanceSubscriptions = message.data;
async handleInstanceDetailsSetSubscriptionsRequest(request) {
this.instanceSubscriptions = { ...request };
}
instanceUpdated(instance) {
if (
this.instanceSubscriptions.all
|| this.instanceSubscriptions.instance_ids.includes(instance.id)
|| this.instanceSubscriptions.instanceIds.includes(instance.id)
) {
libLink.messages.instanceUpdate.send(this, {
id: instance.id,
name: instance.config.get("instance.name"),
assigned_host: instance.config.get("instance.assigned_host"),
game_port: instance.game_port || null,
status: instance.status,
});
this.send(new libData.InstanceDetailsUpdateEvent(
new libData.InstanceDetails(
instance.config.get("instance.name"),
instance.id,
instance.config.get("instance.assigned_host"),
instance.game_port || null,
instance.status,
)
));
}
}
// XXX should probably add a hook for host reuqests?
async createInstanceRequestHandler(message) {
async handleInstanceCreateRequest(request) {
let instanceConfig = new libConfig.InstanceConfig("controller");
await instanceConfig.load(message.data.serialized_config);
await instanceConfig.load(request.config);
await this._controller.instanceCreate(instanceConfig);
}
async deleteInstanceRequestHandler(message, request) {
await this._controller.instanceDelete(message.data.instance_id);
async handleInstanceDeleteRequest(request) {
await this._controller.instanceDelete(request.instanceId);
}
async getInstanceConfigRequestHandler(message) {
let instance = this._controller.getRequestInstance(message.data.instance_id);
return {
serialized_config: instance.config.serialize("control"),
};
async handleInstanceConfigGetRequest(request) {
let instance = this._controller.getRequestInstance(request.instanceId);
return new libData.InstanceConfigGetRequest.Response(instance.config.serialize("control"));
}
async setInstanceConfigFieldRequestHandler(message) {
let instance = this._controller.getRequestInstance(message.data.instance_id);
if (message.data.field === "instance.assigned_host") {
async handleInstanceConfigSetFieldRequest(request) {
let instance = this._controller.getRequestInstance(request.instanceId);
if (request.field === "instance.assigned_host") {
throw new libErrors.RequestError("instance.assigned_host must be set through the assign-host interface");
}
if (message.data.field === "instance.id") {
if (request.field === "instance.id") {
// XXX is this worth implementing? It's race condition galore.
throw new libErrors.RequestError("Setting instance.id is not supported");
}
instance.config.set(message.data.field, message.data.value, "control");
instance.config.set(request.field, request.value, "control");
await this._controller.instanceConfigUpdated(instance);
}
async setInstanceConfigPropRequestHandler(message) {
let instance = this._controller.getRequestInstance(message.data.instance_id);
let { field, prop, value } = message.data;
async handleInstanceConfigSetPropRequest(request) {
let instance = this._controller.getRequestInstance(request.instanceId);
let { field, prop, value } = request;
instance.config.setProp(field, prop, value, "control");
await this._controller.instanceConfigUpdated(instance);
}
async assignInstanceCommandRequestHandler(message, request) {
let { host_id, instance_id } = message.data;
await this._controller.instanceAssign(instance_id, host_id);
async handleInstanceAssignRequest(request) {
await this._controller.instanceAssign(request.instanceId, request.hostId);
}
async setSaveListSubscriptionsRequestHandler(message) {
this.saveListSubscriptions = message.data;
async handleInstanceSetSaveListSubscriptionsRequest(request) {
this.saveListSubscriptions = { ...request };
}
saveListUpdate(data) {
saveListUpdate(instanceId, saves) {
if (
this.saveListSubscriptions.all
|| this.saveListSubscriptions.instance_ids.includes(data.instance_id)
|| this.saveListSubscriptions.instanceIds.includes(instanceId)
) {
libLink.messages.saveListUpdate.send(this, data);
this.send(new libData.InstanceSaveListUpdateEvent(instanceId, saves));
}
}
async downloadSaveRequestHandler(message) {
let { instance_id, save } = message.data;
async handleInstanceDownloadSaveRequest(request) {
let { instanceId, name } = request;
let stream = await routes.createProxyStream(this._controller.app);
stream.filename = save;
stream.filename = name;
let ready = new Promise((resolve, reject) => {
stream.events.on("source", resolve);
@@ -287,29 +351,29 @@ class ControlConnection extends BaseConnection {
});
ready.catch(() => {});
await this._controller.forwardRequestToInstance(libLink.messages.pushSave, {
instance_id,
stream_id: stream.id,
save,
});
await this._controller.sendToHostByInstanceId(new libData.InstancePushSaveRequest(
instanceId,
stream.id,
name,
));
await ready;
return { stream_id: stream.id };
return stream.id;
}
async transferSaveRequestHandler(message, request) {
let { source_save, target_save, copy, instance_id, target_instance_id } = message.data;
if (copy) {
async handleInstanceTransferSaveRequest(request) {
// XXX if control sends the request directly to a host these checks are bypassed
if (request.copy) {
this.user.checkPermission("core.instance.save.copy");
} else if (source_save !== target_save) {
} else if (request.sourceName !== request.targetName) {
this.user.checkPermission("core.instance.save.rename");
}
if (instance_id === target_instance_id) {
if (request.sourceInstanceId === request.targetInstanceId) {
throw new libErrors.RequestError("Source and target instance may not be the same");
}
let sourceInstance = this._controller.getRequestInstance(instance_id);
let targetInstance = this._controller.getRequestInstance(target_instance_id);
let sourceInstance = this._controller.getRequestInstance(request.sourceInstanceId);
let targetInstance = this._controller.getRequestInstance(request.targetInstanceId);
let sourceHostId = sourceInstance.config.get("instance.assigned_host");
let targetHostId = targetInstance.config.get("instance.assigned_host");
if (sourceHostId === null) {
@@ -321,7 +385,7 @@ class ControlConnection extends BaseConnection {
// Let host handle request if source and target is on the same host.
if (sourceHostId === targetHostId) {
return await this.forwardRequestToInstance(message, request);
return await this._controller.sendTo(request, { hostId: sourceHostId });
}
// Check connectivity
@@ -350,42 +414,40 @@ class ControlConnection extends BaseConnection {
// Establish push from source host to stream, this is done first to
// ensure the file size is known prior to the target host pull.
await Promise.all([
this._controller.forwardRequestToInstance(libLink.messages.pushSave, {
instance_id,
stream_id: stream.id,
save: source_save,
}),
this._controller.sendTo(
new libData.InstancePushSaveRequest(request.sourceInstanceId, stream.id, request.sourceName),
{ hostId: sourceHostId }
),
events.once(stream.events, "source"),
]);
// Establish pull from target host to stream and wait for completion.
let result = await this._controller.forwardRequestToInstance(libLink.messages.pullSave, {
instance_id: target_instance_id,
stream_id: stream.id,
filename: target_save,
});
let storedName = await this._controller.sendTo(
new libData.InstancePullSaveRequest(request.targetInstanceId, stream.id, request.targetName),
{ hostId: targetHostId },
);
// Delete source save if this is not a copy
if (!copy) {
await this._controller.forwardRequestToInstance(libLink.messages.deleteSave, {
instance_id,
save: source_save,
});
if (!request.copy) {
await this._controller.sendTo(
new libData.InstanceDeleteSaveRequest(request.sourceInstanceId, request.sourceName),
{ hostId: sourceHostId }
);
}
return { save: result.save };
return storedName;
}
async listModPacksRequestHandler(message) {
return { list: [...this._controller.modPacks.values()].map(pack => pack.toJSON()) };
async handleModPackListRequest() {
return [...this._controller.modPacks.values()];
}
async setModPackSubscriptionsRequestHandler(message) {
this.modPackSubscriptions = message.data;
async handleModPackSetSubscriptionsRequest(request) {
this.modPackSubscriptions = { ...request };
}
async createModPackRequestHandler(message) {
let modPack = libData.ModPack.fromJSON(message.data.mod_pack);
async handleModPackCreateRequest(request) {
let modPack = request.modPack;
if (this._controller.modPacks.has(modPack.id)) {
throw new libErrors.RequestError(`Mod pack with ID ${modPack.id} already exist`);
}
@@ -393,8 +455,8 @@ class ControlConnection extends BaseConnection {
this._controller.modPackUpdated(modPack);
}
async updateModPackRequestHandler(message) {
let modPack = libData.ModPack.fromJSON(message.data.mod_pack);
async handleModPackUpdateRequest(request) {
let modPack = request.modPack;
if (!this._controller.modPacks.has(modPack.id)) {
throw new libErrors.RequestError(`Mod pack with ID ${modPack.id} does not exist`);
}
@@ -402,8 +464,8 @@ class ControlConnection extends BaseConnection {
this._controller.modPackUpdated(modPack);
}
async deleteModPackRequestHandler(message) {
let { id } = message.data;
async handleModPackDeleteRequest(request) {
let { id } = request;
let modPack = this._controller.modPacks.get(id);
if (!modPack) {
throw new libErrors.RequestError(`Mod pack with ID ${id} does not exist`);
@@ -416,26 +478,24 @@ class ControlConnection extends BaseConnection {
modPackUpdated(modPack) {
if (
this.modPackSubscriptions.all
|| this.modPackSubscriptions.mod_pack_ids.includes(modPack.id)
|| this.modPackSubscriptions.modPackIds.includes(modPack.id)
) {
libLink.messages.modPackUpdate.send(this, { mod_pack: modPack.toJSON() });
this.send(new libData.ModPackUpdateEvent(modPack));
}
}
async getModRequestHandler(message) {
let { name, version } = message.data;
async handleModGetRequest(request) {
let { name, version } = request;
let filename = `${name}_${version}.zip`;
let mod = this._controller.mods.get(filename);
if (!mod) {
throw new libErrors.RequestError(`Mod ${filename} does not exist`);
}
return {
mod: mod.toJSON(),
};
return mod;
}
async listModsRequestHandler(message) {
return { list: [...this._controller.mods.values()].map(mod => mod.toJSON()) };
async handleModListRequest() {
return [...this._controller.mods.values()];
}
static termsMatchesMod(terms, mod) {
@@ -456,8 +516,8 @@ class ControlConnection extends BaseConnection {
return true;
}
async searchModsRequestHandler(message) {
let query = libHelpers.parseSearchString(message.data.query, {
async handleModSearchRequest(request) {
let query = libHelpers.parseSearchString(request.query, {
name: "word",
// version
title: "word",
@@ -471,7 +531,7 @@ class ControlConnection extends BaseConnection {
// size
sha1: "word",
});
let factorioVersion = message.data.factorio_version;
let factorioVersion = request.factorioVersion;
let results = new Map();
for (let mod of this._controller.mods.values()) {
@@ -497,7 +557,7 @@ class ControlConnection extends BaseConnection {
}
let resultList = [...results.values()];
const sort = message.data.sort;
const sort = request.sort;
if (sort) {
const sorters = {
name: (a, b) => strcmp(a.versions[0].name, b.versions[0].name),
@@ -508,30 +568,30 @@ class ControlConnection extends BaseConnection {
throw new libErrors.RequestError(`Invalid value for sort: ${sort}`);
}
resultList.sort(sorters[sort]);
let order = message.data.sort_order;
let order = request.sortOrder;
if (order === "desc") {
resultList.reverse();
}
}
const page = message.data.page;
const pageSize = message.data.page_size || 10;
const page = request.page;
const pageSize = request.pageSize || 10;
resultList = resultList.slice((page - 1) * pageSize, page * pageSize);
return {
query_issues: query.issues,
page_count: Math.ceil(results.size / pageSize),
result_count: results.size,
queryIssues: query.issues,
pageCount: Math.ceil(results.size / pageSize),
resultCount: results.size,
results: resultList,
};
}
async setModSubscriptionsRequestHandler(message) {
this.modSubscriptions = message.data;
async handleModSetSubscriptionsRequest(request) {
this.modSubscriptions = { ...request };
}
async downloadModRequestHandler(message) {
let { name, version } = message.data;
async handleModDownloadRequest(request) {
let { name, version } = request;
let filename = `${name}_${version}.zip`;
let mod = this._controller.mods.get(filename);
if (!mod) {
@@ -545,30 +605,30 @@ class ControlConnection extends BaseConnection {
stream.mime = "application/zip";
stream.size = mod.size;
return { stream_id: stream.id };
return stream.id;
}
async deleteModRequestHandler(message) {
await this._controller.deleteMod(message.data.name, message.data.version);
async handleModDeleteRequest(request) {
await this._controller.deleteMod(request.name, request.version);
}
async setLogSubscriptionsRequestHandler(message) {
this.logSubscriptions = message.data;
async handleLogSetSubscriptionsRequest(request) {
this.logSubscriptions = { ...request };
this.updateLogSubscriptions();
}
modUpdated(mod) {
if (
this.modSubscriptions.all
|| this.modSubscriptions.mod_names.includes(mod.name)
|| this.modSubscriptions.modNames.includes(mod.name)
) {
libLink.messages.modUpdate.send(this, { mod: mod.toJSON() });
this.send(new libData.ModUpdateEvent(mod));
}
}
updateLogSubscriptions() {
let { all, controller, host_ids, instance_ids } = this.logSubscriptions;
if (all || controller || host_ids.length || instance_ids.length) {
let { all, controller, hostIds, instanceIds } = this.logSubscriptions;
if (all || controller || hostIds.length || instanceIds.length) {
if (!this.logTransport) {
this.logTransport = new libLoggingUtils.LinkTransport({ link: this });
this._controller.clusterLogger.add(this.logTransport);
@@ -581,57 +641,57 @@ class ControlConnection extends BaseConnection {
}
}
async queryLogRequestHandler(message) {
async handleLogQueryRequest(request) {
let observeDuration = queryLogTime.startTimer();
let { all, controller, host_ids, instance_ids } = message.data;
let { all, controller, hostIds, instanceIds } = request;
let log;
if (!all && controller && !host_ids.length && !instance_ids.length) {
log = await this._controller.queryControllerLog(message.data);
if (!all && controller && !hostIds.length && !instanceIds.length) {
log = await this._controller.queryControllerLog(request);
} else {
log = await this._controller.queryClusterLog(message.data);
log = await this._controller.queryClusterLog(request);
}
observeDuration();
return { log };
}
async listPermissionsRequestHandler(message) {
async handlePermissionListRequest() {
let list = [];
for (let permission of libUsers.permissions.values()) {
list.push({
name: permission.name,
title: permission.title,
description: permission.description,
});
list.push(new libData.RawPermission(
permission.name,
permission.title,
permission.description,
));
}
return { list };
return list;
}
async listRolesRequestHandler(message) {
async handleRoleListRequest() {
let list = [];
for (let role of this._controller.userManager.roles.values()) {
list.push({
id: role.id,
name: role.name,
description: role.description,
permissions: [...role.permissions],
});
list.push(new libData.RawRole(
role.id,
role.name,
role.description,
[...role.permissions],
));
}
return { list };
return list;
}
async createRoleRequestHandler(message) {
async handleRoleCreateRequest(request) {
let lastId = Math.max.apply(null, [...this._controller.userManager.roles.keys()]);
// Start at 5 to leave space for future default roles
let id = Math.max(5, lastId+1);
this._controller.userManager.roles.set(id, new libUsers.Role({ id, ...message.data }));
return { id };
this._controller.userManager.roles.set(id, new libUsers.Role({ id, ...request }));
return id;
}
async updateRoleRequestHandler(message) {
let { id, name, description, permissions } = message.data;
async handleRoleUpdateRequest(request) {
let { id, name, description, permissions } = request;
let role = this._controller.userManager.roles.get(id);
if (!role) {
throw new libErrors.RequestError(`Role with ID ${id} does not exist`);
@@ -643,18 +703,18 @@ class ControlConnection extends BaseConnection {
this._controller.rolePermissionsUpdated(role);
}
async grantDefaultRolePermissionsRequestHandler(message) {
let role = this._controller.userManager.roles.get(message.data.id);
async handleRoleGrantDefaultPermissionsRequest(request) {
let role = this._controller.userManager.roles.get(request.id);
if (!role) {
throw new libErrors.RequestError(`Role with ID ${message.data.id} does not exist`);
throw new libErrors.RequestError(`Role with ID ${request.id} does not exist`);
}
role.grantDefaultPermissions();
this._controller.rolePermissionsUpdated(role);
}
async deleteRoleRequestHandler(message) {
let id = message.data.id;
async handleRoleDeleteRequest(request) {
let id = request.id;
let role = this._controller.userManager.roles.get(id);
if (!role) {
throw new libErrors.RequestError(`Role with ID ${id} does not exist`);
@@ -667,44 +727,48 @@ class ControlConnection extends BaseConnection {
}
}
async getUserRequestHandler(message) {
let name = message.data.name;
async handleUserGetRequest(request) {
let name = request.name;
let user = this._controller.userManager.users.get(name);
if (!user) {
throw new libErrors.RequestError(`User ${name} does not exist`);
}
return {
name: user.name,
roles: [...user.roles].map(role => role.id),
is_admin: user.isAdmin,
is_banned: user.isBanned,
is_whitelisted: user.isWhitelisted,
ban_reason: user.banReason,
instances: [...user.instances],
player_stats: user.playerStats,
instance_stats: [...user.instanceStats],
};
return new libData.RawUser(
user.name,
[...user.roles].map(role => role.id),
[...user.instances],
user.isAdmin,
user.isBanned,
user.isWhitelisted,
user.banReason,
undefined,
user.playerStats,
[...user.instanceStats],
);
}
async listUsersRequestHandler(message) {
async handleUserListRequest() {
let list = [];
for (let user of this._controller.userManager.users.values()) {
list.push({
name: user.name,
roles: [...user.roles].map(role => role.id),
is_admin: user.isAdmin,
is_banned: user.isBanned,
is_whitelisted: user.isWhitelisted,
instances: [...user.instances],
player_stats: user.playerStats,
});
list.push(new libData.RawUser(
user.name,
[...user.roles].map(role => role.id),
[...user.instances],
user.isAdmin,
user.isBanned,
user.isWhitelisted,
undefined,
undefined,
user.playerStats,
undefined,
));
}
return { list };
return list;
}
async setUserSubscriptionsRequestHandler(message) {
this.userSubscriptions = message.data;
async handleUserSetSubscriptionsRequest(request) {
this.userSubscriptions = { ...request };
}
userUpdated(user) {
@@ -712,37 +776,39 @@ class ControlConnection extends BaseConnection {
this.userSubscriptions.all
|| this.userSubscriptions.names.includes(user.name)
) {
libLink.messages.userUpdate.send(this, {
name: user.name,
roles: [...user.roles].map(role => role.id),
is_admin: user.isAdmin,
is_banned: user.isBanned,
is_whitelisted: user.isWhitelisted,
ban_reason: user.banReason,
instances: [...user.instances],
player_stats: user.playerStats,
instance_stats: [...user.instanceStats],
is_deleted: user.isDeleted,
});
this.send(new libData.UserUpdateEvent(
new libData.RawUser(
user.name,
[...user.roles].map(role => role.id),
[...user.instances],
user.isAdmin,
user.isBanned,
user.isWhitelisted,
user.banReason,
user.isDeleted,
user.playerStats,
[...user.instanceStats],
)
));
}
}
async createUserRequestHandler(message) {
let user = this._controller.userManager.createUser(message.data.name);
async handleUserCreateRequest(request) {
let user = this._controller.userManager.createUser(request.name);
this._controller.userUpdated(user);
}
async revokeUserTokenRequestHandler(message) {
let user = this._controller.userManager.users.get(message.data.name);
async handleUserRevokeTokenRequest(request) {
let user = this._controller.userManager.users.get(request.name);
if (!user) {
throw new libErrors.RequestError(`User '${message.data.name}' does not exist`);
throw new libErrors.RequestError(`User '${request.name}' does not exist`);
}
if (user.name !== this.user.name) {
this.user.checkPermission("core.user.revoke_other_token");
}
user.invalidateToken();
for (let controlConnection of this._controller.wsServer.controlConnections) {
for (let controlConnection of this._controller.wsServer.controlConnections.values()) {
if (controlConnection.user.name === user.name) {
controlConnection.connector.terminate();
}
@@ -750,14 +816,14 @@ class ControlConnection extends BaseConnection {
this._controller.userUpdated(user);
}
async updateUserRolesRequestHandler(message) {
let user = this._controller.userManager.users.get(message.data.name);
async handleUserUpdateRolesRequest(request) {
let user = this._controller.userManager.users.get(request.name);
if (!user) {
throw new libErrors.RequestError(`User '${message.data.name}' does not exist`);
throw new libErrors.RequestError(`User '${request.name}' does not exist`);
}
let resolvedRoles = new Set();
for (let roleId of message.data.roles) {
for (let roleId of request.roles) {
let role = this._controller.userManager.roles.get(roleId);
if (!role) {
throw new libErrors.RequestError(`Role with ID ${roleId} does not exist`);
@@ -771,8 +837,8 @@ class ControlConnection extends BaseConnection {
this._controller.userUpdated(user);
}
async setUserAdminRequestHandler(message) {
let { name, create, admin } = message.data;
async handleUserSetAdminRequest(request) {
let { name, create, admin } = request;
let user = this._controller.userManager.users.get(name);
if (!user) {
if (create) {
@@ -785,11 +851,11 @@ class ControlConnection extends BaseConnection {
user.isAdmin = admin;
this._controller.userUpdated(user);
this.broadcastEventToHosts({ data: { name, admin }}, libLink.messages.adminlistUpdate);
this._controller.sendTo(new libData.InstanceAdminlistUpdateEvent(name, admin), "allInstances");
}
async setUserBannedRequestHandler(message) {
let { name, create, banned, reason } = message.data;
async handleUserSetBannedRequest(request) {
let { name, create, banned, reason } = request;
let user = this._controller.userManager.users.get(name);
if (!user) {
if (create) {
@@ -803,11 +869,11 @@ class ControlConnection extends BaseConnection {
user.isBanned = banned;
user.banReason = reason;
this._controller.userUpdated(user);
this.broadcastEventToHosts({ data: { name, banned, reason }}, libLink.messages.banlistUpdate);
this._controller.sendTo(new libData.InstanceBanlistUpdateEvent(name, banned, reason), "allInstances");
}
async setUserWhitelistedRequestHandler(message) {
let { name, create, whitelisted } = message.data;
async handleUserSetWhitelistedRequest(request) {
let { name, create, whitelisted } = request;
let user = this._controller.userManager.users.get(name);
if (!user) {
if (create) {
@@ -820,34 +886,35 @@ class ControlConnection extends BaseConnection {
user.isWhitelisted = whitelisted;
this._controller.userUpdated(user);
this.broadcastEventToHosts({ data: { name, whitelisted }}, libLink.messages.whitelistUpdate);
this._controller.sendTo(new libData.InstanceWhitelistUpdateEvent(name, whitelisted), "allInstances");
}
async deleteUserRequestHandler(message) {
let user = this._controller.userManager.users.get(message.data.name);
async handleUserDeleteRequest(request) {
let name = request.name;
let user = this._controller.userManager.users.get(name);
if (!user) {
throw new libErrors.RequestError(`User '${message.data.name}' does not exist`);
throw new libErrors.RequestError(`User '${name}' does not exist`);
}
user.isDeleted = true;
this._controller.userManager.users.delete(message.data.name);
this._controller.userManager.users.delete(name);
this._controller.userUpdated(user);
if (user.is_admin) {
this.broadcastEventToHosts({ data: { name, admin: false }}, libLink.messages.adminlistUpdate);
this._controller.sendTo(new libData.InstanceAdminlistUpdateEvent(name, false), "allInstances");
}
if (user.is_whitelisted) {
this.broadcastEventToHosts({ data: { name, whitelisted: false }}, libLink.messages.whitelistUpdate);
this._controller.sendTo(new libData.InstanceWhitelistUpdateEvent(name, false), "allInstances");
}
if (user.is_banned) {
this.broadcastEventToHosts({ data: { name, banned: false, reason: "" }}, libLink.messages.banlistUpdate);
this._controller.sendTo(new libData.InstanceBanlistUpdateEvent(name, false, ""), "allInstances");
}
}
async debugDumpWsRequestHandler(message) {
async handleDebugDumpWsRequest(request) {
this.ws_dumper = data => {
if (this.connector.connected) {
libLink.messages.debugWsMessage.send(this, data);
this.send(new libData.DebugWsMessageEvent(data.direction, data.content));
}
};
this.connector._socket.clusterio_ignore_dump = true;
packages/controller/src/Controller.js
+211 -57
View File
@@ -13,7 +13,6 @@ const libConfig = require("@clusterio/lib/config");
const libData = require("@clusterio/lib/data");
const libErrors = require("@clusterio/lib/errors");
const libFileOps = require("@clusterio/lib/file_ops");
const libLink = require("@clusterio/lib/link");
const libLoggingUtils = require("@clusterio/lib/logging_utils");
const libPlugin = require("@clusterio/lib/plugin");
const libPluginLoader = require("@clusterio/lib/plugin_loader");
@@ -114,6 +113,11 @@ class Controller {
this._state = "new";
this._shouldStop = false;
this._fallbackedRequests = new Map();
this._registeredRequests = new Map();
this._registeredEvents = new Map();
this._snoopedEvents = new Map();
this.devMiddleware = null;
}
@@ -559,16 +563,16 @@ class Controller {
}
hostUpdated(host) {
let update = {
agent: host.agent,
version: host.version,
id: host.id,
name: host.name,
public_address: host.public_address || null,
connected: this.wsServer.hostConnections.has(host.id),
};
let update = new libData.HostDetails(
host.agent,
host.version,
host.name,
host.id,
this.wsServer.hostConnections.has(host.id),
host.public_address === null ? undefined : host.public_address,
);
for (let controlConnection of this.wsServer.controlConnections) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
}
@@ -682,17 +686,16 @@ class Controller {
if (currentAssignedHost !== null && hostId !== currentAssignedHost) {
let oldHostConnection = this.wsServer.hostConnections.get(currentAssignedHost);
if (oldHostConnection && !oldHostConnection.connector.closing) {
await libLink.messages.unassignInstance.send(oldHostConnection, { instance_id: instanceId });
await oldHostConnection.send(new libData.InstanceUnassignInternalRequest(instanceId));
}
}
// Assign to target
instance.config.set("instance.assigned_host", hostId);
if (hostId !== null) {
await libLink.messages.assignInstance.send(newHostConnection, {
instance_id: instanceId,
serialized_config: instance.config.serialize("host"),
});
await newHostConnection.send(
new libData.InstanceAssignInternalRequest(instanceId, instance.config.serialize("host"))
);
} else {
instance.status = "unassigned";
this.instanceUpdated(instance);
@@ -709,8 +712,9 @@ class Controller {
*/
async instanceDelete(instanceId) {
let instance = this.getRequestInstance(instanceId);
if (instance.config.get("instance.assigned_host") !== null) {
await this.forwardRequestToInstance(libLink.messages.deleteInstance, { instance_id: instanceId });
let hostId = instance.config.get("instance.assigned_host");
if (hostId !== null) {
await this.sendTo(new libData.InstanceDeleteInternalRequest(instanceId), { hostId });
}
this.instances.delete(instanceId);
@@ -734,10 +738,9 @@ class Controller {
if (hostId !== null) {
let connection = this.wsServer.hostConnections.get(hostId);
if (connection) {
await libLink.messages.assignInstance.send(connection, {
instance_id: instance.id,
serialized_config: instance.config.serialize("host"),
});
await connection.send(
new libData.InstanceAssignInternalRequest(instance.id, instance.config.serialize("host"))
);
}
}
}
@@ -755,7 +758,7 @@ class Controller {
}
instanceUpdated(instance) {
for (let controlConnection of this.wsServer.controlConnections) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
}
@@ -764,18 +767,18 @@ class Controller {
}
}
saveListUpdate(data) {
for (let controlConnection of this.wsServer.controlConnections) {
saveListUpdate(instanceId, saves) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
}
controlConnection.saveListUpdate(data);
controlConnection.saveListUpdate(instanceId, saves);
}
}
modPackUpdated(modPack) {
for (let controlConnection of this.wsServer.controlConnections) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
}
@@ -786,7 +789,7 @@ class Controller {
}
modUpdated(mod) {
for (let controlConnection of this.wsServer.controlConnections) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
}
@@ -798,7 +801,7 @@ class Controller {
}
userUpdated(user) {
for (let controlConnection of this.wsServer.controlConnections) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
}
@@ -813,15 +816,15 @@ class Controller {
* @param {module:lib/users.User} user - User permisions updated for.
*/
userPermissionsUpdated(user) {
for (let controlConnection of this.wsServer.controlConnections) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.user === user) {
libLink.messages.accountUpdate.send(controlConnection, {
"roles": [...user.roles].map(r => ({
controlConnection.send(
new libData.AccountUpdateEvent([...user.roles].map(r => ({
name: r.name,
id: r.id,
permissions: [...r.permissions],
})),
});
})))
);
}
}
}
@@ -832,15 +835,17 @@ class Controller {
* @param {module:lib/users.Role} role - Role permisions updated for.
*/
rolePermissionsUpdated(role) {
for (let controlConnection of this.wsServer.controlConnections) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.user.roles.has(role)) {
libLink.messages.accountUpdate.send(controlConnection, {
"roles": [...controlConnection.user.roles].map(r => ({
name: r.name,
id: r.id,
permissions: [...r.permissions],
})),
});
controlConnection.send(
new libData.AccountUpdateEvent(
[...controlConnection.user.roles].map(r => ({
name: r.name,
id: r.id,
permissions: [...r.permissions],
}))
)
);
}
}
}
@@ -965,35 +970,184 @@ class Controller {
};
}
register(Class, handler) {
if (Class.type === "request") {
this.registerRequest(Class, handler);
} else if (Class.type === "event") {
this.registerEvent(Class, handler);
} else {
throw new Error(`Class ${Class.name} has unrecognized type ${Class.type}`);
}
}
registerRequest(Request, handler) {
if (this._registeredRequests.has(Request)) {
throw new Error(`Request ${Request.name} is already registered`);
}
this._registeredRequests.set(Request, handler);
}
fallbackRequest(Request, handler) {
if (this._fallbackedRequests.has(Request)) {
throw new Error(`Request ${Request.name} is already fallbacked`);
}
this._fallbackedRequests.set(Request, handler);
}
registerEvent(Event, handler) {
if (this._registeredEvents.has(Event)) {
throw new Error(`Event ${Event.name} is already registered`);
}
this._registeredEvents.set(Event, handler);
}
snoopEvent(Event, handler) {
if (this._snoopedEvents.has(Event)) {
throw new Error(`Event ${Event.name} is already snooped`);
}
this._snoopedEvents.set(Event, handler);
}
/**
* Forward the given request to the host contaning the instance
* Send request or event
*
* Finds the host which the instance with the given instance ID is
* currently assigned to and forwards the request to that instance.
* The request data must have an instance_id parameter.
* Routes the given request or event to the given destination. The
* destination argument supports address shorthands, see {@link
* module:lib/data.Address.fromShorthand}
*
* @param {module:lib/link.Request} request - The request to send.
* @param {object} data - Data to pass with the request.
* @param {number} data.instance_id -
* ID of instance to send this request towards
* @returns {Promise<object>} data returned from the request.
* @param {*} requestOrEvent - The request or event to send.
* @param {*|module:lib/data.Address} address - Where to send it.
* @returns {Promise<*>|undefined}
* Promise that resolves to the response if a request was sent or
* undefined if it was an event.
*/
async forwardRequestToInstance(request, data) {
let instance = this.getRequestInstance(data.instance_id);
sendTo(requestOrEvent, address) {
let dst = libData.Address.fromShorthand(address);
if (requestOrEvent.constructor.type === "request") {
return this.sendRequest(requestOrEvent, dst);
}
if (requestOrEvent.constructor.type === "event") {
return this.sendEvent(requestOrEvent, dst);
}
throw Error(`Unknown type ${requestOrEvent.constructor.type}.`);
}
async sendRequest(request, dst) {
let connection;
if (dst.type === libData.Address.controller) {
throw new Error("controller as dst is not supported.");
} else if (dst.type === libData.Address.control) {
connection = this.wsServer.controlConnections.get(dst.id);
if (!connection) {
throw new libErrors.RequestError("Target control connection does not exist.");
}
} else if (dst.type === libData.Address.instance) {
let instance = this.getRequestInstance(dst.id);
let hostId = instance.config.get("instance.assigned_host");
if (hostId === null) {
throw new libErrors.RequestError("Instance is not assigned to a host");
}
connection = this.wsServer.hostConnections.get(hostId);
if (!connection) {
throw new libErrors.RequestError("Host containing instance is not connected");
}
} else if (dst.type === libData.Address.host) {
connection = this.wsServer.hostConnections.get(dst.id);
if (!connection) {
throw new libErrors.RequestError("Host is not connected");
}
} else {
throw new Error(`Unknown address type ${dst.type}`);
}
return connection.sendRequest(request, dst);
}
sendEvent(event, dst) {
let connection;
if (dst.type === libData.Address.controller) {
throw new Error("controller as dst is not supported.");
} else if (dst.type === libData.Address.control) {
connection = this.wsServer.controlConnections.get(dst.id);
} else if (dst.type === libData.Address.instance) {
let instance = this.instances.get(dst.id);
if (!instance) {
return;
}
let hostId = instance.config.get("instance.assigned_host");
if (hostId === null) {
return;
}
connection = this.wsServer.hostConnections.get(hostId);
} else if (dst.type === libData.Address.host) {
connection = this.wsServer.hostConnections.get(dst.id);
} else if (dst.type === libData.Address.broadcast) {
if (dst.id === libData.Address.control) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
controlConnection.sendEvent(event, dst);
}
} else if (dst.id === libData.Address.instance || dst.id === libData.Address.host) {
for (let hostConnection of this.wsServer.hostConnections.values()) {
hostConnection.sendEvent(event, dst);
}
} else {
throw new Error(`Unexpected broacdast target ${dst.id}`);
}
return;
} else {
throw new Error(`Unknown address type ${dst.type}`);
}
if (connection) {
connection.sendEvent(event, dst);
}
}
sendToHostByInstanceId(requestOrEvent) {
if (requestOrEvent.constructor.type === "request") {
return this.sendRequestToHostByInstanceId(requestOrEvent);
}
if (requestOrEvent.constructor.type === "event") {
return this.sendEventToHostByInstanceId(requestOrEvent);
}
throw Error(`Unknown type ${requestOrEvent.constructor.type}.`);
}
async sendRequestToHostByInstanceId(request) {
let instance = this.getRequestInstance(request.instanceId);
let hostId = instance.config.get("instance.assigned_host");
if (hostId === null) {
throw new libErrors.RequestError("Instance is not assigned to a host");
}
let connection = this.wsServer.hostConnections.get(hostId);
if (!connection) {
throw new libErrors.RequestError("Host containing instance is not connected");
}
if (request.plugin && !connection.plugins.has(request.plugin)) {
throw new libErrors.RequestError(`Host containing instance does not have ${request.plugin} plugin`);
}
return await connection.send(request);
}
return await request.send(connection, data);
sendEventToHostByInstanceId(event) {
let instance = this.getRequestInstance(event.instanceId);
let hostId = instance.config.get("instance.assigned_host");
if (hostId === null) {
return;
}
let connection = this.wsServer.hostConnections.get(hostId);
if (!connection) {
return;
}
connection.send(event);
}
}
packages/controller/src/ControllerRouter.js
+123
View File
@@ -0,0 +1,123 @@
"use strict";
const libData = require("@clusterio/lib/data");
class ControllerRouter {
/** @type {module:controller/src/Controller} */
controller;
pendingRequests = new Map();
constructor(controller) {
this.controller = controller;
}
addHostConnection(connection) {
this.hostConnections.set(connection.id, connection);
}
addControlConnection(connection) {
this.controlConnections.set(connection.id, connection);
}
/**
* @param {module:lib/link.Link} origin - Source link of the message.
* @param {module:lib/data.Message} message
* Link the message originated from.
* @param {boolean} hasFallback - true if fallback handling is available.
* @returns {boolean} true if the message was handled
*/
forwardMessage(origin, message, hasFallback) {
if (!["request", "response", "responseError", "event"].includes(message.type)) {
throw new Error(`Message type ${message.type} can't be forwarded`);
}
let dst = message.dst;
let nextHop;
let msg;
if (dst.type === libData.Address.broadcast) {
this.broadcastMessage(origin, message);
return true;
} else if (dst.type === libData.Address.host) {
nextHop = this.controller.wsServer.hostConnections.get(dst.id);
if (!nextHop) {
msg = `Host ${dst.id} is offline`;
}
} else if (dst.type === libData.Address.instance) {
let instance = this.controller.instances.get(dst.id);
if (!instance) {
msg = `Instance ${dst.id} does not exist`;
} else {
let assignedHost = instance.config.get("instance.assigned_host");
if (assignedHost === null) {
msg = `Instance ${dst.id} is not assigned a host`;
} else {
nextHop = this.controller.wsServer.hostConnections.get(assignedHost);
if (!nextHop) {
msg = `Assigned host for instance ${dst.id} is offline`;
}
}
}
} else if (dst.type === libData.Address.control) {
nextHop = this.controller.wsServer.controlConnections.get(dst.id);
if (!nextHop) {
msg = "Control connection does not exist";
}
} else if (dst.type === libData.Address.controller) {
msg = `Unexpected message forwarded to ${dst}`;
}
if (nextHop === origin) {
msg = `Message would return back to sender ${origin.dst}.`;
nextHop = undefined;
}
if (message.type === "request") {
if (!nextHop) {
if (hasFallback) {
return false;
}
origin.connector.sendResponseError(
new libData.ResponseError(msg || "Unroutable destination"), message.src
);
return true;
}
}
if (nextHop) {
nextHop.connector.send(message);
} else {
this.warnUnrouted(message, msg);
}
return true;
}
broadcastMessage(origin, message) {
let dst = message.dst;
if (message.type !== "event") {
this.warnUnrouted(message, `Unexpected broadcast of ${message.type}`);
} else if (dst.id === libData.Address.host || dst.id === libData.Address.instance) {
for (let hostConnection of this.controller.wsServer.hostConnections.values()) {
if (hostConnection !== origin) {
hostConnection.connector.send(message);
}
}
} else if (dst.id === libData.Address.control) {
for (let controlConnection of this.controller.wsServer.controlConnections.values()) {
if (controlConnection !== origin) {
controlConnection.connector.send(message);
}
}
} else {
this.warnUnrouted(message, `Unexpected broacdast target ${dst.id}`);
}
}
warnUnrouted(message, msg) {
let dst = message.dst;
let baseMsg = `No destination for ${message.constructor.name} routed from ${message.src} to ${dst}`;
logger.warn(msg ? `${baseMsg}: ${msg}.` : `${baseMsg}.`);
}
}
module.exports = ControllerRouter;
packages/controller/src/HostConnection.js
+55 -41
View File
@@ -1,6 +1,6 @@
"use strict";
const libData = require("@clusterio/lib/data");
const libConfig = require("@clusterio/lib/config");
const libLink = require("@clusterio/lib/link");
const { logger } = require("@clusterio/lib/logging");
const libPlugin = require("@clusterio/lib/plugin");
const PlayerStats = require("@clusterio/lib/PlayerStats");
@@ -17,7 +17,7 @@ const InstanceInfo = require("./InstanceInfo");
*/
class HostConnection extends BaseConnection {
constructor(registerData, connector, controller) {
super("host", connector, controller);
super(connector, controller);
this._agent = registerData.agent;
this._id = registerData.id;
@@ -31,7 +31,7 @@ class HostConnection extends BaseConnection {
id: this._id,
name: this._name,
version: this._version,
public_address: registerData.public_address,
public_address: registerData.publicAddress,
plugins: registerData.plugins,
});
@@ -57,6 +57,12 @@ class HostConnection extends BaseConnection {
libPlugin.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev);
}
});
this.register(libData.InstanceStatusChangedEvent, this.handleInstanceStatusChangedEvent.bind(this));
this.register(libData.InstancesUpdateRequest, this.handleInstancesUpdateRequest.bind(this));
this.register(libData.InstanceSaveListUpdateEvent, this.handleInstanceSaveListUpdateEvent.bind(this));
this.register(libData.LogMessageEvent, this.handleLogMessageEvent.bind(this));
this.register(libData.InstancePlayerUpdateEvent, this.handleInstancePlayerUpdateEvent.bind(this));
}
_checkPluginVersions() {
@@ -83,6 +89,11 @@ class HostConnection extends BaseConnection {
}
}
async prepareDisconnect() {
await libPlugin.invokeHook(this._controller.plugins, "onPrepareHostDisconnect", this);
return await super.prepareDisconnect();
}
/**
* ID of the host this connection is connected to
*
@@ -92,15 +103,15 @@ class HostConnection extends BaseConnection {
return this._id;
}
async instanceStatusChangedEventHandler(message, event) {
let instance = this._controller.instances.get(message.data.instance_id);
async handleInstanceStatusChangedEvent(request) {
let instance = this._controller.instances.get(request.instanceId);
// It's possible to get updates from an instance that does not exist
// or is not assigned to the host it originated from if it was
// reassigned or deleted while the connection to the host it was
// originally on was down at the time.
if (!instance || instance.config.get("instance.assigned_host") !== this._id) {
logger.warn(`Got bogus update for instance id ${message.data.instance_id}`);
logger.warn(`Got bogus update for instance id ${request.instanceId}`);
return;
}
@@ -109,41 +120,40 @@ class HostConnection extends BaseConnection {
// changed. Hosts also send status updates on assignInstance which
// for hacky reason is also used to push config changes and
// restablish status after a connection loss.
if (instance.status === message.data.status) {
if (instance.status === request.status) {
return;
}
let prev = instance.status;
instance.status = message.data.status;
instance.game_port = message.data.game_port;
instance.status = request.status;
instance.game_port = request.gamePort;
logger.verbose(`Instance ${instance.config.get("instance.name")} State: ${instance.status}`);
this._controller.instanceUpdated(instance);
await libPlugin.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev);
}
async updateInstancesRequestHandler(message) {
async handleInstancesUpdateRequest(request) {
// Push updated instance configs
for (let instance of this._controller.instances.values()) {
if (instance.config.get("instance.assigned_host") === this._id) {
await libLink.messages.assignInstance.send(this, {
instance_id: instance.id,
serialized_config: instance.config.serialize("host"),
});
await this.send(
new libData.InstanceAssignInternalRequest(instance.id, instance.config.serialize("host"))
);
}
}
// Assign instances the host has but controller does not
for (let instanceData of message.data.instances) {
for (let instanceData of request.instances) {
let instanceConfig = new libConfig.InstanceConfig("controller");
await instanceConfig.load(instanceData.serialized_config, "host");
await instanceConfig.load(instanceData.config, "host");
let controllerInstance = this._controller.instances.get(instanceConfig.get("instance.id"));
if (controllerInstance) {
// Check if this instance is assigned somewhere else.
if (controllerInstance.config.get("instance.assigned_host") !== this._id) {
await libLink.messages.unassignInstance.send(this, {
instance_id: controllerInstance.id,
});
await this.send(
new libData.InstanceUnassignInternalRequest(instanceConfig.get("instance.id"))
);
continue;
}
@@ -152,7 +162,7 @@ class HostConnection extends BaseConnection {
let prev = controllerInstance.status;
controllerInstance.status = instanceData.status;
logger.verbose(`Instance ${instanceConfig.get("instance.name")} State: ${instanceData.status}`);
this._controller.instanceUpdated(instance);
this._controller.instanceUpdated(controllerInstance);
await libPlugin.invokeHook(
this._controller.plugins, "onInstanceStatusChanged", controllerInstance, prev
);
@@ -164,10 +174,11 @@ class HostConnection extends BaseConnection {
let newInstance = new InstanceInfo({ config: instanceConfig, status: instanceData.status });
this._controller.instances.set(instanceConfig.get("instance.id"), newInstance);
this._controller.addInstanceHooks(newInstance);
await libLink.messages.assignInstance.send(this, {
instance_id: instanceConfig.get("instance.id"),
serialized_config: instanceConfig.serialize("host"),
});
await this.send(
new libData.InstanceAssignInternalRequest(
instanceConfig.get("instance.id"), instanceConfig.serialize("host")
)
);
await libPlugin.invokeHook(this._controller.plugins, "onInstanceStatusChanged", newInstance, null);
}
@@ -188,40 +199,43 @@ class HostConnection extends BaseConnection {
}
}
libLink.messages.syncUserLists.send(this, { adminlist, banlist, whitelist });
await this.send(new libData.SyncUserListsEvent(adminlist, banlist, whitelist));
}
async saveListUpdateEventHandler(message) {
this._controller.saveListUpdate(message.data);
async handleInstanceSaveListUpdateEvent(event) {
this._controller.saveListUpdate(event.instanceId, event.saves);
}
async logMessageEventHandler(message) {
async handleLogMessageEvent(event) {
this._controller.clusterLogger.log({
...message.data.info,
...event.info,
host_id: this._id,
host_name: this._name,
});
}
async playerEventEventHandler(message) {
let { instance_id, name, type, stats } = message.data;
let user = this._controller.userManager.users.get(name);
async handleInstancePlayerUpdateEvent(event, src) {
let instanceId = src.id;
let user = this._controller.userManager.users.get(event.name);
if (!user) {
user = this._controller.userManager.createUser(name);
user = this._controller.userManager.createUser(event.name);
}
if (type === "join") {
user.notifyJoin(instance_id);
} else if (type === "leave") {
user.notifyLeave(instance_id);
if (event.type === "join") {
user.notifyJoin(instanceId);
} else if (event.type === "leave") {
user.notifyLeave(instanceId);
}
user.instanceStats.set(instance_id, new PlayerStats(stats));
user.instanceStats.set(instanceId, new PlayerStats(event.stats));
user.recalculatePlayerStats();
this._controller.userUpdated(user);
delete message.data.stats;
let instance = this._controller.instances.get(instance_id);
await libPlugin.invokeHook(this._controller.plugins, "onPlayerEvent", instance, message.data);
let instance = this._controller.instances.get(instanceId);
await libPlugin.invokeHook(this._controller.plugins, "onPlayerEvent", instance, {
type: event.type,
name: event.name,
reason: event.reason,
});
}
}
packages/controller/src/WsServer.js
+150 -126
View File
@@ -2,10 +2,10 @@
const jwt = require("jsonwebtoken");
const WebSocket = require("ws");
const libData = require("@clusterio/lib/data");
const libErrors = require("@clusterio/lib/errors");
const { logger } = require("@clusterio/lib/logging");
const libPrometheus = require("@clusterio/lib/prometheus");
const libSchema = require("@clusterio/lib/schema");
const ControlConnection = require("./ControlConnection");
const packageVersion = require("../package").version;
@@ -40,7 +40,7 @@ class WsServer {
this.stopAcceptingNewSessions = false;
/** @type {Array<module:controller/src/ControlConnection>} */
this.controlConnections = [];
this.controlConnections = new Map();
/** @type {Map<number, module:controller/src/HostConnection>} */
this.hostConnections = new Map();
/** @type {Map<number, module:controller/src/WsServerConnector>} */
@@ -51,6 +51,7 @@ class WsServer {
// Unique string for the session token audience
this.sessionAud = `session-${Date.now()}`;
this.nextSessionId = 1;
this.nextControlId = 1;
/** @type {module:ws.Server} */
this.wss = new WebSocket.Server({
@@ -64,7 +65,7 @@ class WsServer {
this.stopAcceptingNewSessions = true;
let disconnectTasks = [];
for (let controlConnection of this.controlConnections) {
for (let controlConnection of this.controlConnections.values()) {
disconnectTasks.push(controlConnection.disconnect(1001, "Server Quit"));
}
@@ -122,11 +123,7 @@ class WsServer {
loadedPlugins[name] = plugin.info.version;
}
socket.send(JSON.stringify({ seq: null, type: "hello", data: {
version: packageVersion,
plugins: loadedPlugins,
}}));
socket.send(JSON.stringify(new libData.MessageHello(new libData.HelloData(packageVersion, loadedPlugins))));
this.attachHandler(socket, req);
}
@@ -159,49 +156,31 @@ ${err.stack}`
});
}
async handleHandshake(message, socket, req) {
async handleHandshake(rawMessage, socket, req) {
let message;
try {
message = JSON.parse(message);
const json = JSON.parse(rawMessage);
libData.Message.validate(json);
message = libData.Message.fromJSON(json);
} catch (err) {
logger.verbose(`WsServer | closing ${req.socket.remoteAddress} after receiving invalid JSON`);
logger.verbose(`WsServer | closing ${req.socket.remoteAddress} after receiving invalid message`);
wsRejectedConnectionsCounter.inc();
socket.close(1002, "Invalid JSON");
socket.close(1002, `Invalid message: ${err.message}`);
return;
}
if (!libSchema.clientHandshake(message)) {
if (message.type === "resume") {
this.resumeSession(message.data, socket, req);
return;
}
if (message.type !== "registerHost" && message.type !== "registerControl") {
logger.verbose(`WsServer | closing ${req.socket.remoteAddress} after receiving invalid handshake`);
wsRejectedConnectionsCounter.inc();
socket.close(1002, "Bad handshake");
return;
}
let { type, data } = message;
if (type === "resume") {
let connector;
try {
let payload = jwt.verify(
data.session_token,
Buffer.from(this.controller.config.get("controller.auth_secret"), "base64"),
{ audience: this.sessionAud }
);
connector = this.activeConnectors.get(payload.sid);
if (!connector) {
throw new Error();
}
} catch (err) {
socket.send(JSON.stringify({ seq: null, type: "invalidate", data: {}}));
this.attachHandler(socket, req);
return;
}
connector.continue(socket, data.last_seq);
return;
}
if (this.stopAcceptingNewSessions) {
logger.verbose(`WsServer | closing ${req.socket.remoteAddress}, server is shutting down`);
wsRejectedConnectionsCounter.inc();
@@ -209,46 +188,20 @@ ${err.stack}`
return;
}
// Validate token
let user;
try {
if (type === "register_host") {
let tokenPayload = jwt.verify(
data.token,
Buffer.from(this.controller.config.get("controller.auth_secret"), "base64"),
// migrate: allow pre-rename tokens issued to hosts before alpha-14
{ audience: ["host", "slave"] }
);
// migrate: allow pre-rename tokens issued to hosts before alpha-14
if ((tokenPayload.host !== undefined ? tokenPayload.host : tokenPayload.slave) !== data.id) {
throw new Error("missmatched host id");
}
} else if (type === "register_control") {
let tokenPayload = jwt.verify(
data.token,
Buffer.from(this.controller.config.get("controller.auth_secret"), "base64"),
{ audience: "user" }
);
user = this.controller.userManager.users.get(tokenPayload.user);
if (!user) {
throw new Error("invalid user");
}
if (tokenPayload.iat < user.tokenValidAfter) {
throw new Error("invalid token");
}
user.checkPermission("core.control.connect");
}
} catch (err) {
logger.verbose(`WsServer | authentication failed for ${req.socket.remoteAddress}`);
wsRejectedConnectionsCounter.inc();
socket.close(4003, `Authentication failed: ${err.message}`);
if (message.type === "registerHost") {
await this.registerHost(message.data, socket, req);
return;
}
if (message.type === "registerControl") {
await this.registerControl(message.data, socket, req);
return;
}
throw new Error("This should be unreachable");
}
createSession(dst) {
let sessionId = this.nextSessionId;
this.nextSessionId += 1;
let sessionToken = jwt.sign(
@@ -257,66 +210,137 @@ ${err.stack}`
);
let sessionTimeout = this.controller.config.get("controller.session_timeout");
let heartbeatInterval = this.controller.config.get("controller.heartbeat_interval");
let connector = new WsServerConnector(sessionId, sessionTimeout, heartbeatInterval);
let connector = new WsServerConnector(dst, sessionId, sessionTimeout, heartbeatInterval);
this.activeConnectors.set(sessionId, connector);
connector.on("close", () => {
this.activeConnectors.delete(sessionId);
});
let additionalReadyData = {};
if (type === "register_host") {
let connection = this.hostConnections.get(data.id);
if (connection) {
logger.verbose(`WsServer | disconnecting existing connection for host ${data.id}`);
await connection.disconnect(1008, "Registered from another connection");
}
return [connector, sessionToken];
}
logger.info(
`WsServer | registered host ${data.name} (${data.id}) using agent ${data.agent} ${data.version}`
async registerHost(data, socket, req) {
try {
let tokenPayload = jwt.verify(
data.token,
Buffer.from(this.controller.config.get("controller.auth_secret"), "base64"),
// migrate: allow pre-rename tokens issued to hosts before alpha-14
{ audience: ["host", "slave"] }
);
if (data.agent === "Clusterio Host" && data.version !== packageVersion) {
logger.warn(
`Host ${data.name} (${data.id}) connected using version ${data.version} which does not match the ` +
`version of the controller is currently running (${packageVersion}). It may not work as expected.`
);
// migrate: allow pre-rename tokens issued to hosts before alpha-14
if ((tokenPayload.host !== undefined ? tokenPayload.host : tokenPayload.slave) !== data.id) {
throw new Error("missmatched host id");
}
connection = new HostConnection(data, connector, this.controller);
connector.on("close", () => {
if (this.hostConnections.get(data.id) === connection) {
this.hostConnections.delete(data.id);
this.controller.hostUpdated(this.controller.hosts.get(data.id));
} else {
logger.warn("Unlisted HostConnection closed");
}
});
this.hostConnections.set(data.id, connection);
this.controller.hostUpdated(this.controller.hosts.get(data.id));
} else if (type === "register_control") {
logger.verbose(`WsServer | registered control from ${req.socket.remoteAddress}`);
let connection = new ControlConnection(data, connector, this.controller, user);
connector.on("close", () => {
let index = this.controlConnections.indexOf(connection);
if (index !== -1) {
this.controlConnections.splice(index, 1);
} else {
logger.warn("Unlisted ControlConnection closed");
}
});
this.controlConnections.push(connection);
additionalReadyData["account"] = {
name: user.name,
roles: [...user.roles].map(r => ({
name: r.name,
id: r.id,
permissions: [...r.permissions],
})),
};
} catch (err) {
logger.verbose(`WsServer | authentication failed for ${req.socket.remoteAddress}`);
wsRejectedConnectionsCounter.inc();
socket.close(4003, `Authentication failed: ${err.message}`);
return;
}
connector.ready(socket, sessionToken, additionalReadyData);
let [connector, sessionToken] = this.createSession(new libData.Address(libData.Address.host, data.id));
let connection = this.hostConnections.get(data.id);
if (connection) {
logger.verbose(`WsServer | disconnecting existing connection for host ${data.id}`);
await connection.disconnect(1008, "Registered from another connection");
}
logger.info(
`WsServer | registered host ${data.name} (${data.id}) using agent ${data.agent} ${data.version}`
);
if (data.agent === "Clusterio Host" && data.version !== packageVersion) {
logger.warn(
`Host ${data.name} (${data.id}) connected using version ${data.version} which does not match the ` +
`version of the controller is currently running (${packageVersion}). It may not work as expected.`
);
}
connection = new HostConnection(data, connector, this.controller);
connector.on("close", () => {
if (this.hostConnections.get(data.id) === connection) {
this.hostConnections.delete(data.id);
this.controller.hostUpdated(this.controller.hosts.get(data.id));
} else {
logger.warn("Unlisted HostConnection closed");
}
});
this.hostConnections.set(data.id, connection);
this.controller.hostUpdated(this.controller.hosts.get(data.id));
let src = new libData.Address(libData.Address.host, data.id);
connector.ready(socket, src, sessionToken);
}
async registerControl(data, socket, req) {
let user;
try {
let tokenPayload = jwt.verify(
data.token,
Buffer.from(this.controller.config.get("controller.auth_secret"), "base64"),
{ audience: "user" }
);
user = this.controller.userManager.users.get(tokenPayload.user);
if (!user) {
throw new Error("invalid user");
}
if (tokenPayload.iat < user.tokenValidAfter) {
throw new Error("invalid token");
}
user.checkPermission("core.control.connect");
} catch (err) {
logger.verbose(`WsServer | authentication failed for ${req.socket.remoteAddress}`);
wsRejectedConnectionsCounter.inc();
socket.close(4003, `Authentication failed: ${err.message}`);
return;
}
let id = this.nextControlId;
this.nextControlId += 1;
let [connector, sessionToken] = this.createSession(new libData.Address(libData.Address.control, id));
logger.verbose(`WsServer | registered control ${id} from ${req.socket.remoteAddress}`);
let connection = new ControlConnection(data, connector, this.controller, user, id);
connector.on("close", () => {
this.controlConnections.delete(id);
});
this.controlConnections.set(id, connection);
let account = new libData.AccountDetails(
user.name,
[...user.roles].map(r => ({
name: r.name,
id: r.id,
permissions: [...r.permissions],
}))
);
let src = new libData.Address(libData.Address.control, id);
connector.ready(socket, src, sessionToken, account);
}
resumeSession(data, socket, req) {
let connector;
try {
let payload = jwt.verify(
data.sessionToken,
Buffer.from(this.controller.config.get("controller.auth_secret"), "base64"),
{ audience: this.sessionAud }
);
connector = this.activeConnectors.get(payload.sid);
if (!connector) {
throw new Error();
}
} catch (err) {
socket.send(JSON.stringify(new libData.MessageInvalidate()));
this.attachHandler(socket, req);
return;
}
connector.continue(socket, data.lastSeq);
}
}
packages/controller/src/WsServerConnector.js
+29 -38
View File
@@ -1,6 +1,7 @@
"use strict";
const events = require("events");
const libData = require("@clusterio/lib/data");
const libLink = require("@clusterio/lib/link");
const { logger } = require("@clusterio/lib/logging");
@@ -12,9 +13,10 @@ const { logger } = require("@clusterio/lib/logging");
* @alias module:controller/src/WsServerConnector
*/
class WsServerConnector extends libLink.WebSocketBaseConnector {
constructor(sessionId, sessionTimeout, heartbeatInterval) {
super(sessionTimeout);
constructor(dst, sessionId, sessionTimeout, heartbeatInterval) {
super(new libData.Address(libData.Address.controller, 0), dst);
this._sessionTimeout = sessionTimeout;
this._socket = null;
this._sessionId = sessionId;
this._heartbeatInterval = heartbeatInterval;
@@ -38,27 +40,27 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
* Sends the ready message over the socket to initiate the session.
*
* @param {Object} socket - WebSocket connection to client.
* @param {module:lib/data.Address} src - Source address for this link.
* @param {string} sessionToken -
* the session token to send to the client.
* @param {Object=} additionalData -
* extra properties to send along the ready message.
* @param {module:lib/data.AccountDetails=} account -
* account data to provide to control connection
*/
ready(socket, sessionToken, additionalData) {
ready(socket, src, sessionToken, account) {
this._socket = socket;
this._socket.send(JSON.stringify({
seq: null,
type: "ready",
data: {
session_token: sessionToken,
session_timeout: this._sessionTimeout,
heartbeat_interval: this._heartbeatInterval,
...additionalData,
},
}));
this._sendInternal(new libData.MessageReady(
new libData.ReadyData(
src,
sessionToken,
this._sessionTimeout,
this._heartbeatInterval,
account,
),
));
this._state = "connected";
this._attachSocketHandlers();
this.emit("connect");
this.emit("connect", { src: this.src });
}
/**
@@ -68,7 +70,7 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
* socket given from the message sequence given.
*
* @param {module:net.Socket} socket - New socket to continue on.
* @param {number} lastSeq - The last message the client received.
* @param {number=} lastSeq - The last message the client received.
*/
continue(socket, lastSeq) {
@@ -87,15 +89,9 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
this._timeoutId = null;
}
this._socket.send(JSON.stringify({
seq: null,
type: "continue",
data: {
last_seq: this._lastReceivedSeq,
session_timeout: this._sessionTimeout,
heartbeat_interval: this._heartbeatInterval,
},
}));
this._sendInternal(new libData.MessageContinue(
new libData.ContinueData(this._sessionTimeout, this._heartbeatInterval, this._lastReceivedSeq)
));
this._state = "connected";
this._attachSocketHandlers();
@@ -118,6 +114,7 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
this._socket.on("close", (code, reason) => {
logger.verbose(`Connector | Close (code: ${code}, reason: ${reason})`);
this.stopHeartbeat();
this._socket = null;
if (this._closing) {
this._reset();
@@ -148,18 +145,12 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
// Handle messages
this._socket.on("message", data => {
let message = JSON.parse(data);
if (["connected", "closing"].includes(this._state)) {
if (message.seq !== null) {
this._lastReceivedSeq = message.seq;
}
if (message.type === "heartbeat") {
this._processHeartbeat(message);
} else {
this.emit("message", message);
}
let message = this._parseMessage(data);
if (!message) {
return;
}
if (this._state === "connected") {
this._processMessage(message);
} else {
throw new Error(`Received message in unexpected state ${this._state}`);
packages/controller/src/routes.js
+11 -9
View File
@@ -49,12 +49,12 @@ async function getMetrics(req, res, next) {
let requests = [];
let timeout = req.app.locals.controller.config.get("controller.metrics_timeout") * 1000;
for (let hostConnection of req.app.locals.controller.wsServer.hostConnections.values()) {
for (let [hostId, hostConnection] of req.app.locals.controller.wsServer.hostConnections) {
if (!hostConnection.connected) {
continue;
}
requests.push(libHelpers.timeout(
libLink.messages.getMetrics.send(hostConnection).catch(err => {
hostConnection.send(new libData.HostMetricsRequest()).catch(err => {
if (!(err instanceof libErrors.SessionLost)) {
logger.error(`Unexpected error gathering metrics from host:\n${err.stack}`);
}
@@ -361,15 +361,17 @@ async function uploadSave(req, res) {
});
try {
let result = await Promise.race([
req.app.locals.controller.forwardRequestToInstance(libLink.messages.pullSave, {
instance_id: instanceId,
stream_id: proxyStream.id,
filename,
}),
let storedName = await Promise.race([
req.app.locals.controller.sendToHostByInstanceId(
new libData.InstancePullSaveRequest(
instanceId,
proxyStream.id,
filename,
)
),
timeout,
]);
saves.push(result.save);
saves.push(storedName);
} catch (err) {
proxyStream.events.emit("close");