From 1ed46dff9c30868f9f84a70e224f2299e496fcec Mon Sep 17 00:00:00 2001 From: Danielv123 Date: Fri, 30 Oct 2026 21:49:53 +0100 Subject: [PATCH] Fix issues found in review of host pairing - Saves of a datastore run one at a time, and changes made while saving are saved by the next save. Overlapping saves from approving and revoking hosts could write an older snapshot after a newer one. - Map subscribers fetch everything again when a new session starts, as the controller only sends values that exist. A host revoked while a browser was disconnected stayed in its list. - The host token check accepts the same pre-rename tokens as the WebSocket, so upgrading a host with such a token does not make it ask to join again. --- packages/controller/src/Controller.ts | 24 ++++++++++++++++++ packages/controller/src/WsServer.ts | 17 +------------ packages/controller/src/routes.ts | 12 +-------- packages/lib/src/datastore.ts | 19 ++++++++++++--- packages/lib/src/subscriptions.ts | 10 ++++++++ test/controller/routes.js | 6 +++++ test/lib/datastore.js | 35 +++++++++++++++++++++++++++ test/lib/subscriptions.test.js | 28 +++++++++++++++++++++ 8 files changed, 121 insertions(+), 30 deletions(-) diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 5908c62e..b2341cc2 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -1025,6 +1025,30 @@ export default class Controller { return jwt.sign({ aud: "host", host: hostId }, this.authSecret); } + /** + * Check a host token + * + * @param token - Host token to check. + * @returns ID of the host the token was issued to. + * @throws if the token is invalid, its host is not registered or it was issued before tokenValidAfter. + */ + verifyHostToken(token: string): number { + // migrate: allow pre-rename tokens issued to hosts before alpha-14 + const payload = jwt.verify(token, this.authSecret, { audience: ["host", "slave"] }); + if (typeof payload === "string") { + throw new Error("unexpected JsonWebToken type"); + } + const hostId = payload.host !== undefined ? payload.host : payload.slave; + const host = this.hosts.get(hostId); + if (!host) { + throw new Error("host is not registered, start the host again to request to join the cluster"); + } + if (!payload.iat || payload.iat < host.tokenValidAfter) { + throw new Error("invalid token"); + } + return hostId; + } + hostsUpdated(hosts: HostRecord[]) { let updates = hosts.map(host => host.toHostDetails()); this.subscriptions.broadcast(new lib.HostUpdatesEvent(updates)); diff --git a/packages/controller/src/WsServer.ts b/packages/controller/src/WsServer.ts index fea731a6..2d88b91c 100644 --- a/packages/controller/src/WsServer.ts +++ b/packages/controller/src/WsServer.ts @@ -252,24 +252,9 @@ ${err.stack}` } try { - // migrate: allow pre-rename tokens issued to hosts before alpha-14 - const tokenPayload = jwt.verify(data.token, this.controller.authSecret, { audience: ["host", "slave"] }); - if (typeof tokenPayload === "string") { - throw new Error("unexpected JsonWebToken type"); - } - - // migrate: allow pre-rename tokens issued to hosts before alpha-14 - if ((tokenPayload.host !== undefined ? tokenPayload.host : tokenPayload.slave) !== data.id) { + if (this.controller.verifyHostToken(data.token) !== data.id) { throw new Error("missmatched host id"); } - - let host = this.controller.hosts.get(data.id); - if (!host) { - throw new Error("host is not registered, start the host again to request to join the cluster"); - } - if (!tokenPayload.iat || tokenPayload.iat < host.tokenValidAfter) { - throw new Error("invalid token"); - } } catch (err: any) { logger.verbose(`WsServer | authentication failed for ${this.remoteAddr(req)}`); wsRejectedConnectionsCounter.inc(); diff --git a/packages/controller/src/routes.ts b/packages/controller/src/routes.ts index b9ce7803..fa967fe2 100644 --- a/packages/controller/src/routes.ts +++ b/packages/controller/src/routes.ts @@ -189,17 +189,7 @@ function validateHostToken(req: Request, res: Response, next: any) { } try { - const tokenPayload = jwt.verify(token, req.app.locals.controller.authSecret, { audience: "host" }); - if (typeof tokenPayload === "string") { - throw new Error("unexpected JsonWebToken type"); - } - let host = req.app.locals.controller.hosts.get(tokenPayload.host); - if (!host) { - throw new Error("invalid host"); - } - if (!tokenPayload.iat || tokenPayload.iat < (host.tokenValidAfter ?? 0)) { - throw new Error("invalid token"); - } + req.app.locals.controller.verifyHostToken(token); } catch (err) { res.sendStatus(401); return; diff --git a/packages/lib/src/datastore.ts b/packages/lib/src/datastore.ts index 212a7bbb..59be8d96 100644 --- a/packages/lib/src/datastore.ts +++ b/packages/lib/src/datastore.ts @@ -166,6 +166,7 @@ export abstract class Datastore< "update": [ updates: U[] ], }> { private dirty = false; + private saving: Promise = Promise.resolve(); constructor( private provider: DatastoreProvider = new MemoryDatastoreProvider(), @@ -181,10 +182,22 @@ export abstract class Datastore< // Save the datastore to the provider async save() { - if (this.dirty) { - await this.provider.save(this.data); + // Saves run one at a time so an older snapshot can't be written after a newer one. + const save = this.saving.then(async () => { + if (!this.dirty) { + return; + } + // Cleared before writing so changes made while saving are saved by the next save. this.dirty = false; - } + try { + await this.provider.save(this.data); + } catch (err) { + this.dirty = true; + throw err; + } + }); + this.saving = save.catch(() => {}); + await save; } // Load the datastore from the provider diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index cc66781e..9bb97d99 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -725,6 +725,16 @@ export class MapSubscriber< /** Current value of the subscribed resource */ values = new Map(); + override handleConnectionEvent(event: "connect" | "drop" | "resume" | "close") { + if (event === "connect") { + // A new session is only sent values that exist, not ones removed while disconnected, + // so everything is fetched again. + this.values.clear(); + this.lastUpdatedMs = -1; + } + super.handleConnectionEvent(event); + } + protected override getLastUpdatedTimeMs(event: E) { let maxTime = this.lastUpdatedMs; for (const update of event.updates) { diff --git a/test/controller/routes.js b/test/controller/routes.js index 32f427e9..b0b88cd2 100644 --- a/test/controller/routes.js +++ b/test/controller/routes.js @@ -221,6 +221,7 @@ describe("controller/src/routes", function() { controller.hosts = new SubscribableDatastore(); controller.hostPairing = new HostPairingManager(controller); controller.generateHostToken = Controller.prototype.generateHostToken; + controller.verifyHostToken = Controller.prototype.verifyHostToken; controller.wsServer.remoteAddr = req => req.socket.remoteAddress; adminToken = controller.users.signUserToken(controller.users.getByName("test")); playerToken = controller.users.signUserToken(controller.users.getByName("player")); @@ -478,6 +479,11 @@ describe("controller/src/routes", function() { } assert(controller.hosts.has(hostId)); }); + it("should respond with 200 for a host token from before the slave rename", async function() { + const { hostId } = await controller.hostPairing.createHost("Host A"); + const token = jwt.sign({ aud: "slave", slave: hostId }, controller.authSecret); + await assertResponse(await check({ "Authorization": `Bearer ${token}` }), 200, { valid: true }); + }); it("should respond with 401 without a valid host token", async function() { await controller.hostPairing.createHost("Host A"); assert.equal((await check({})).status, 401); diff --git a/test/lib/datastore.js b/test/lib/datastore.js index 4c829e76..5959bf0b 100644 --- a/test/lib/datastore.js +++ b/test/lib/datastore.js @@ -346,6 +346,41 @@ describe("lib/datastore", function() { await datastore.save(); assert.equal(datastoreProvider.methodCalls.length, 2); }); + it("should write overlapping saves in order with the latest data", async function() { + const written = []; + let release; + datastoreProvider.save = async data => { + const keys = [...data.keys()]; + if (!written.length) { + await new Promise(resolve => { release = resolve; }); + } + written.push(keys); + }; + datastore = new MockDatastore(datastoreProvider, new Map([["a", 1]])); + datastore.callTouch(); + const first = datastore.save(); + await new Promise(resolve => setImmediate(resolve)); // First write in progress + datastore.getData().set("b", 2); + datastore.callTouch(); + const second = datastore.save(); + release(); + await Promise.all([first, second]); + assert.deepEqual(written, [["a"], ["a", "b"]]); + }); + it("should save again after a failed save", async function() { + let fail = true; + datastoreProvider.save = async () => { + if (fail) { + fail = false; + throw new Error("disk full"); + } + datastoreProvider.methodCalls.push(["save"]); + }; + datastore.callTouch(); + await assert.rejects(datastore.save(), /disk full/); + await datastore.save(); + assert.deepEqual(datastoreProvider.methodCalls.at(-1), ["save"]); + }); }); describe("load", function() { diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js index 461ca0ee..c71923f8 100644 --- a/test/lib/subscriptions.test.js +++ b/test/lib/subscriptions.test.js @@ -1584,6 +1584,34 @@ describe("lib/subscriptions", function() { assert.equal(snap2.has(2), true); }); + it("should fetch everything again when a new session starts", async function() { + subscriber.subscribe(() => {}); + await events.once(mockControl.connector, "send"); + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false }, + ])); + const [before] = subscriber.getSnapshot(); + mockControl.connector.sentMessages.length = 0; + + mockControl.connector.emit("connect"); + assert.equal(subscriber.values.size, 0); + assert.notStrictEqual(subscriber.getSnapshot()[0], before); + await new Promise(resolve => setImmediate(resolve)); + const requests = mockControl.connector.sentMessages.filter(m => m.name === "SubscriptionRequest"); + assert.equal(requests.length, 1); + assert.equal(requests[0].data.lastRequestTimeMs, -1); + }); + + it("should keep values when a session is resumed", async function() { + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false }, + ])); + subscriber.handleConnectionEvent("drop"); + subscriber.handleConnectionEvent("resume"); + assert.equal(subscriber.values.size, 1); + assert.equal(subscriber.lastUpdatedMs, 10); + }); + it("should handle multiple updates correctly", async function() { await subscriber._handleEvent(new MapEvent([ { id: 1, updatedAtMs: 10, isDeleted: false },