From 94addb6315ab43c1f0506c1571ad52103bb33784 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Mon, 30 Jan 2017 18:58:37 +0000 Subject: [PATCH 1/6] Rewrite the device key query logic Only permit one query per user at a time. --- spec/integ/matrix-client-methods.spec.js | 6 +- src/crypto/DeviceList.js | 215 ++++++++++------------- src/crypto/index.js | 4 +- 3 files changed, 102 insertions(+), 123 deletions(-) diff --git a/spec/integ/matrix-client-methods.spec.js b/spec/integ/matrix-client-methods.spec.js index 4acec1aa9..b87d255d2 100644 --- a/spec/integ/matrix-client-methods.spec.js +++ b/spec/integ/matrix-client-methods.spec.js @@ -347,7 +347,11 @@ describe("MatrixClient", function() { */ httpBackend.when("POST", "/keys/query").check(function(req) { - expect(req.data).toEqual({device_keys: {boris: {}, chaz: {}}}); + expect(req.data).toEqual({device_keys: { + '@alice:localhost': {}, + 'boris': {}, + 'chaz': {}, + }}); }).respond(200, { device_keys: { boris: borisKeys, diff --git a/src/crypto/DeviceList.js b/src/crypto/DeviceList.js index 175286b34..df493b124 100644 --- a/src/crypto/DeviceList.js +++ b/src/crypto/DeviceList.js @@ -25,7 +25,6 @@ import q from 'q'; import DeviceInfo from './deviceinfo'; import olmlib from './olmlib'; -import utils from '../utils'; /** * @alias module:crypto/DeviceList @@ -36,10 +35,11 @@ export default class DeviceList { this._sessionStore = sessionStore; this._olmDevice = olmDevice; + // users with outdated device lists // userId -> true this._pendingUsersWithNewDevices = {}; - // userId -> [promise, ...] + // userId -> promise this._keyDownloadsInProgressByUser = {}; } @@ -53,45 +53,30 @@ export default class DeviceList { * module:crypto/deviceinfo|DeviceInfo}. */ downloadKeys(userIds, forceDownload) { - const self = this; - // promises we need to wait for while the download happens const promises = []; - // list of userids we need to download keys for - let downloadUsers = []; - - function perUserCatch(u) { - return function(e) { - console.warn('Error downloading keys for user ' + u + ':', e); - }; - } - - if (forceDownload) { - downloadUsers = userIds; - } else { - for (let i = 0; i < userIds.length; ++i) { - const u = userIds[i]; - - const inprogress = this._keyDownloadsInProgressByUser[u]; - if (inprogress) { - // wait for the download to complete - promises.push(q.any(inprogress).catch(perUserCatch(u))); - } else if (!this.getStoredDevicesForUser(u)) { - downloadUsers.push(u); + let needsFlush = false; + userIds.forEach((u) => { + if (this._keyDownloadsInProgressByUser[u]) { + // just wait for the existing download to complete + promises.push(this._keyDownloadsInProgressByUser[u]); + } else { + if (forceDownload || !this.getStoredDevicesForUser(u)) { + this.invalidateUserDeviceList(u); + } + if (this._pendingUsersWithNewDevices[u]) { + needsFlush = true; } } + }); + + if (needsFlush) { + promises.push(this.flushNewDeviceRequests(true)); } - if (downloadUsers.length > 0) { - const r = this._doKeyDownloadForUsers(downloadUsers); - downloadUsers.map(function(u) { - promises.push(r[u].catch(perUserCatch(u))); - }); - } - - return q.all(promises).then(function() { - return self._getDevicesFromStore(userIds); + return q.all(promises).then(() => { + return this._getDevicesFromStore(userIds); }); } @@ -210,134 +195,124 @@ export default class DeviceList { * @param {String} userId */ invalidateUserDeviceList(userId) { + if (typeof userId !== 'string') { + throw new Error('userId must be a string; was '+userId); + } this._pendingUsersWithNewDevices[userId] = true; } /** - * Start device queries for any users who sent us an m.new_device recently + * Start device queries for any users with outdated device lists + * + * We tolerate multiple concurrent device queries, but only one query per + * user. + * + * If any users already have downloads in progress, they are ignored - + * they will be refreshed when the current download completes anyway. + * + * The returned promise resolves immediately if there are no users with + * outdated device lists, or if all users with outdated device lists already + * have a query in progress. + * + * Otherwise, a new query request is made, and the promise resolves or + * once that query completes. If the query fails, the promise will reject + * if rejectOnFailure was truthy, otherwise it will still resolve. + * + * @param {Boolean?} rejectOnFailure true to make the returned promise + * reject if the device list query fails. + * + * @return {Promise} */ - flushNewDeviceRequests() { - const users = Object.keys(this._pendingUsersWithNewDevices); + flushNewDeviceRequests(rejectOnFailure) { + const users = Object.keys(this._pendingUsersWithNewDevices).filter( + (u) => !this._keyDownloadsInProgressByUser[u], + ); if (users.length === 0) { - return; + return q(); } - const r = this._doKeyDownloadForUsers(users); + let prom = this._doKeyDownloadForUsers(users).then(() => { + users.forEach((u) => { + delete this._keyDownloadsInProgressByUser[u]; + }); - // we've kicked off requests to these users: remove their - // pending flag for now. - this._pendingUsersWithNewDevices = {}; + // flush out any more requests that were blocked up while that + // was going on, but let the initial promise complete now. + // + this.flushNewDeviceRequests().done(); + }, (e) => { + console.error( + 'Error updating device key cache for ' + users + ":", e, + ); - users.map((u) => { - r[u] = r[u].catch((e) => { - console.error( - 'Error updating device keys for user ' + u + ':', e, - ); - - // reinstate the pending flags on any users which failed; this will - // mean that we will do another download in the future, but won't - // tight-loop. - // + // reinstate the pending flags on any users which failed; this will + // mean that we will do another download in the future, but won't + // tight-loop. + // + users.forEach((u) => { + delete this._keyDownloadsInProgressByUser[u]; this._pendingUsersWithNewDevices[u] = true; }); + + // TODO: schedule a retry. + throw e; }); - q.all(Object.values(r)).done(); + users.forEach((u) => { + delete this._pendingUsersWithNewDevices[u]; + this._keyDownloadsInProgressByUser[u] = prom; + }); + + if (!rejectOnFailure) { + // normally we just want to swallow the exception - we've already + // logged it futher up. + prom = prom.catch((e) => {}); + } + return prom; } /** * @param {string[]} downloadUsers list of userIds * - * @return {Object} a map from userId to a promise for a result for that user + * @return {Promise} */ _doKeyDownloadForUsers(downloadUsers) { - const self = this; - console.log('Starting key download for ' + downloadUsers); - const deferMap = {}; - const promiseMap = {}; - - downloadUsers.map(function(u) { - const deferred = q.defer(); - const promise = deferred.promise.finally(function() { - const inProgress = self._keyDownloadsInProgressByUser[u]; - utils.removeElement(inProgress, function(e) { - return e === promise; - }); - if (inProgress.length === 0) { - // no more downloads for this user; remove the element - delete self._keyDownloadsInProgressByUser[u]; - } - }); - - if (!self._keyDownloadsInProgressByUser[u]) { - self._keyDownloadsInProgressByUser[u] = []; - } - self._keyDownloadsInProgressByUser[u].push(promise); - - deferMap[u] = deferred; - promiseMap[u] = promise; - }); - - this._baseApis.downloadKeysForUsers( + return this._baseApis.downloadKeysForUsers( downloadUsers, - ).done(function(res) { + ).then((res) => { const dk = res.device_keys || {}; - for (let i = 0; i < downloadUsers.length; ++i) { - const userId = downloadUsers[i]; - var deviceId; - + for (const userId of downloadUsers) { console.log('got keys for ' + userId + ':', dk[userId]); - if (!dk[userId]) { - // no result for this user - const err = 'Unknown'; - // TODO: do something with res.failures - deferMap[userId].reject(err); - continue; - } - // map from deviceid -> deviceinfo for this user const userStore = {}; - const devs = self._sessionStore.getEndToEndDevicesForUser(userId); + const devs = this._sessionStore.getEndToEndDevicesForUser(userId); if (devs) { - for (deviceId in devs) { - if (devs.hasOwnProperty(deviceId)) { - const d = DeviceInfo.fromStorage(devs[deviceId], deviceId); - userStore[deviceId] = d; - } - } + Object.keys(devs).forEach((deviceId) => { + const d = DeviceInfo.fromStorage(devs[deviceId], deviceId); + userStore[deviceId] = d; + }); } _updateStoredDeviceKeysForUser( - self._olmDevice, userId, userStore, dk[userId], - ); + this._olmDevice, userId, userStore, dk[userId] || {}, + ); // update the session store const storage = {}; - for (deviceId in userStore) { - if (!userStore.hasOwnProperty(deviceId)) { - continue; - } - + Object.keys(userStore).forEach((deviceId) => { storage[deviceId] = userStore[deviceId].toStorage(); - } - self._sessionStore.storeEndToEndDevicesForUser( + }); + + this._sessionStore.storeEndToEndDevicesForUser( userId, storage, - ); - - deferMap[userId].resolve(); + ); } - }, function(err) { - downloadUsers.map(function(u) { - deferMap[u].reject(err); - }); }); - - return promiseMap; } } diff --git a/src/crypto/index.js b/src/crypto/index.js index 527fd9f80..2abd6a1a3 100644 --- a/src/crypto/index.js +++ b/src/crypto/index.js @@ -708,7 +708,7 @@ Crypto.prototype._onInitialSyncCompleted = function(rooms) { this._initialSyncCompleted = true; // catch up on any m.new_device events which arrived during the initial sync. - this._deviceList.flushNewDeviceRequests(); + this._deviceList.flushNewDeviceRequests().done(); if (this._sessionStore.getDeviceAnnounced()) { return; @@ -845,7 +845,7 @@ Crypto.prototype._onNewDeviceEvent = function(event) { // we delay handling these until the intialsync has completed, so that we // can do all of them together. if (this._initialSyncCompleted) { - this._deviceList.flushNewDeviceRequests(); + this._deviceList.flushNewDeviceRequests().done(); } }; From c3440c506cf98b4b29d158c9d8fa6da0e93d41b0 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Fri, 3 Feb 2017 00:10:13 +0000 Subject: [PATCH 2/6] Address review comments Update some comments, and s/flushNewDeviceRequests/refreshOutdatedDeviceLists/. --- src/crypto/DeviceList.js | 28 ++++++++++++++++++---------- src/crypto/index.js | 4 ++-- 2 files changed, 20 insertions(+), 12 deletions(-) diff --git a/src/crypto/DeviceList.js b/src/crypto/DeviceList.js index df493b124..f90e117e1 100644 --- a/src/crypto/DeviceList.js +++ b/src/crypto/DeviceList.js @@ -56,7 +56,7 @@ export default class DeviceList { // promises we need to wait for while the download happens const promises = []; - let needsFlush = false; + let needsRefresh = false; userIds.forEach((u) => { if (this._keyDownloadsInProgressByUser[u]) { // just wait for the existing download to complete @@ -66,13 +66,13 @@ export default class DeviceList { this.invalidateUserDeviceList(u); } if (this._pendingUsersWithNewDevices[u]) { - needsFlush = true; + needsRefresh = true; } } }); - if (needsFlush) { - promises.push(this.flushNewDeviceRequests(true)); + if (needsRefresh) { + promises.push(this.refreshOutdatedDeviceLists(true)); } return q.all(promises).then(() => { @@ -190,11 +190,18 @@ export default class DeviceList { * Mark the cached device list for the given user outdated. * * This doesn't set off an update, so that several users can be batched - * together. Call flushDeviceListRequests() for that. + * together. Call refreshOutdatedDeviceLists() for that. * * @param {String} userId */ invalidateUserDeviceList(userId) { + // sanity-check the userId. This is mostly paranoia, but if synapse + // can't parse the userId we give it as an mxid, it 500s the whole + // request and we can never update the device lists again (because + // the broken userId is always 'invalid' and always included in any + // refresh request). + // By checking it is at least a string, we can eliminate a class of + // silly errors. if (typeof userId !== 'string') { throw new Error('userId must be a string; was '+userId); } @@ -207,14 +214,15 @@ export default class DeviceList { * We tolerate multiple concurrent device queries, but only one query per * user. * - * If any users already have downloads in progress, they are ignored - - * they will be refreshed when the current download completes anyway. + * If any users already have downloads in progress, they are ignored - they + * will be refreshed when the current download completes anyway, so + * each user with outdated device lists will be updated eventually. * * The returned promise resolves immediately if there are no users with * outdated device lists, or if all users with outdated device lists already * have a query in progress. * - * Otherwise, a new query request is made, and the promise resolves or + * Otherwise, a new query request is made, and the promise resolves * once that query completes. If the query fails, the promise will reject * if rejectOnFailure was truthy, otherwise it will still resolve. * @@ -223,7 +231,7 @@ export default class DeviceList { * * @return {Promise} */ - flushNewDeviceRequests(rejectOnFailure) { + refreshOutdatedDeviceLists(rejectOnFailure) { const users = Object.keys(this._pendingUsersWithNewDevices).filter( (u) => !this._keyDownloadsInProgressByUser[u], ); @@ -240,7 +248,7 @@ export default class DeviceList { // flush out any more requests that were blocked up while that // was going on, but let the initial promise complete now. // - this.flushNewDeviceRequests().done(); + this.refreshOutdatedDeviceLists().done(); }, (e) => { console.error( 'Error updating device key cache for ' + users + ":", e, diff --git a/src/crypto/index.js b/src/crypto/index.js index 2abd6a1a3..f007ae401 100644 --- a/src/crypto/index.js +++ b/src/crypto/index.js @@ -708,7 +708,7 @@ Crypto.prototype._onInitialSyncCompleted = function(rooms) { this._initialSyncCompleted = true; // catch up on any m.new_device events which arrived during the initial sync. - this._deviceList.flushNewDeviceRequests().done(); + this._deviceList.refreshOutdatedDeviceLists().done(); if (this._sessionStore.getDeviceAnnounced()) { return; @@ -845,7 +845,7 @@ Crypto.prototype._onNewDeviceEvent = function(event) { // we delay handling these until the intialsync has completed, so that we // can do all of them together. if (this._initialSyncCompleted) { - this._deviceList.flushNewDeviceRequests().done(); + this._deviceList.refreshOutdatedDeviceLists().done(); } }; From 89ef4aa6e7a3eccd8edfcffb46cd5240a0368756 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Wed, 1 Feb 2017 14:01:41 +0000 Subject: [PATCH 3/6] Handle device change notifications from /sync When we get a notification from /sync that a user has updated their device list, mark the list outdated, and then fire off a device query. --- src/base-apis.js | 31 +++++++++++++++++++++++-------- src/client.js | 23 +++++++++++++++++++++-- src/crypto/DeviceList.js | 9 ++++++++- src/crypto/index.js | 34 +++++++++++++++++++++------------- src/sync.js | 18 ++++++++++++++++-- 5 files changed, 89 insertions(+), 26 deletions(-) diff --git a/src/base-apis.js b/src/base-apis.js index d928c406e..a6465c701 100644 --- a/src/base-apis.js +++ b/src/base-apis.js @@ -1018,20 +1018,35 @@ MatrixBaseApis.prototype.uploadKeysRequest = function(content, opts, callback) { * * @param {string[]} userIds list of users to get keys for * - * @param {module:client.callback=} callback + * @param {Object=} opts + * + * @param {string=} opts.token sync token to pass in the query request, to help + * the HS give the most recent results * * @return {module:client.Promise} Resolves: result object. Rejects: with * an error response ({@link module:http-api.MatrixError}). */ -MatrixBaseApis.prototype.downloadKeysForUsers = function(userIds, callback) { - const downloadQuery = {}; - - for (let i = 0; i < userIds.length; ++i) { - downloadQuery[userIds[i]] = {}; +MatrixBaseApis.prototype.downloadKeysForUsers = function(userIds, opts) { + if (utils.isFunction(opts)) { + // opts used to be 'callback'. + throw new Error( + 'downloadKeysForUsers no longer accepts a callback parameter', + ); } - const content = {device_keys: downloadQuery}; + opts = opts || {}; + + const content = { + device_keys: {}, + }; + if ('token' in opts) { + content.token = opts.token; + } + userIds.forEach((u) => { + content.device_keys[u] = {}; + }); + return this._http.authedRequestWithPrefix( - callback, "POST", "/keys/query", undefined, content, + undefined, "POST", "/keys/query", undefined, content, httpApi.PREFIX_UNSTABLE, ); }; diff --git a/src/client.js b/src/client.js index a1ba21209..def7e5cd9 100644 --- a/src/client.js +++ b/src/client.js @@ -2638,8 +2638,6 @@ MatrixClient.prototype.startClient = function(opts) { }; } - this._clientOpts = opts; - if (this._crypto) { this._crypto.uploadKeys(5).done(); const tenMinutes = 1000 * 60 * 10; @@ -2657,6 +2655,13 @@ MatrixClient.prototype.startClient = function(opts) { console.error("Still have sync object whilst not running: stopping old one"); this._syncApi.stop(); } + + // shallow-copy the opts dict before modifying and storing it + opts = Object.assign({}, opts); + + opts.crypto = this._crypto; + this._clientOpts = opts; + this._syncApi = new SyncApi(this, opts); this._syncApi.sync(); }; @@ -3040,12 +3045,26 @@ module.exports.CRYPTO_ENABLED = CRYPTO_ENABLED; * * * @event module:client~MatrixClient#"sync" + * * @param {string} state An enum representing the syncing state. One of "PREPARED", * "SYNCING", "ERROR", "STOPPED". + * * @param {?string} prevState An enum representing the previous syncing state. * One of "PREPARED", "SYNCING", "ERROR", "STOPPED" or null. + * * @param {?Object} data Data about this transition. + * * @param {MatrixError} data.err The matrix error if state=ERROR. + * + * @param {String} data.oldSyncToken The 'since' token passed to /sync. + * null for the first successful sync since this client was + * started. Only present if state=PREPARED or + * state=SYNCING. + * + * @param {String} data.nextSyncToken The 'next_batch' result from /sync, which + * will become the 'since' token for the next call to /sync. Only present if + * state=PREPARED or state=SYNCING. + * * @example * matrixClient.on("sync", function(state, prevState, data) { * switch (state) { diff --git a/src/crypto/DeviceList.js b/src/crypto/DeviceList.js index 770938b17..342ff029a 100644 --- a/src/crypto/DeviceList.js +++ b/src/crypto/DeviceList.js @@ -41,6 +41,8 @@ export default class DeviceList { // userId -> promise this._keyDownloadsInProgressByUser = {}; + + this.lastKnownSyncToken = null; } /** @@ -288,8 +290,13 @@ export default class DeviceList { _doKeyDownloadForUsers(downloadUsers) { console.log('Starting key download for ' + downloadUsers); + const token = this.lastKnownSyncToken; + const opts = {}; + if (token) { + opts.token = token; + } return this._baseApis.downloadKeysForUsers( - downloadUsers, + downloadUsers, opts, ).then((res) => { const dk = res.device_keys || {}; diff --git a/src/crypto/index.js b/src/crypto/index.js index 6a326c328..3f3fc131a 100644 --- a/src/crypto/index.js +++ b/src/crypto/index.js @@ -55,8 +55,6 @@ function Crypto(baseApis, eventEmitter, sessionStore, userId, deviceId) { this._userId = userId; this._deviceId = deviceId; - this._initialSyncCompleted = false; - this._olmDevice = new OlmDevice(sessionStore); this._deviceList = new DeviceList(baseApis, sessionStore, this._olmDevice); @@ -115,6 +113,9 @@ function _registerEventHandlers(crypto, eventEmitter) { const rooms = eventEmitter.getRooms(); crypto._onInitialSyncCompleted(rooms); } + if (syncState === "SYNCING") { + crypto._onSyncCompleted(data); + } } catch (e) { console.error("Error handling sync", e); } @@ -689,6 +690,18 @@ Crypto.prototype.decryptEvent = function(event) { alg.decryptEvent(event); }; +/** + * Handle the notification from /sync that a user has updated their device list. + * + * @param {String} userId + */ +Crypto.prototype.userDeviceListChanged = function(userId) { + this._deviceList.invalidateUserDeviceList(userId); + + // don't flush the outdated device list yet - we do it once we finish + // processing the sync. +}; + /** * handle an m.room.encryption event * @@ -716,11 +729,6 @@ Crypto.prototype._onCryptoEvent = function(event) { * @param {module:models/room[]} rooms list of rooms the client knows about */ Crypto.prototype._onInitialSyncCompleted = function(rooms) { - this._initialSyncCompleted = true; - - // catch up on any m.new_device events which arrived during the initial sync. - this._deviceList.refreshOutdatedDeviceLists().done(); - if (this._sessionStore.getDeviceAnnounced()) { return; } @@ -779,6 +787,12 @@ Crypto.prototype._onInitialSyncCompleted = function(rooms) { }); }; +Crypto.prototype._onSyncCompleted = function(syncData) { + // catch up on any new devices we got told about during the sync. + this._deviceList.lastKnownSyncToken = syncData.nextSyncToken; + this._deviceList.refreshOutdatedDeviceLists().done(); +}; + /** * Handle a key event * @@ -852,12 +866,6 @@ Crypto.prototype._onNewDeviceEvent = function(event) { } this._deviceList.invalidateUserDeviceList(userId); - - // we delay handling these until the intialsync has completed, so that we - // can do all of them together. - if (this._initialSyncCompleted) { - this._deviceList.refreshOutdatedDeviceLists().done(); - } }; diff --git a/src/sync.js b/src/sync.js index 90ba408d3..facf5d6ed 100644 --- a/src/sync.js +++ b/src/sync.js @@ -58,6 +58,7 @@ function debuglog() { * @constructor * @param {MatrixClient} client The matrix client instance to use. * @param {Object} opts Config options + * @param {module:crypto=} opts.crypto Crypto manager */ function SyncApi(client, opts) { this.client = client; @@ -529,13 +530,18 @@ SyncApi.prototype._sync = function(syncOptions) { } // emit synced events + const syncEventData = { + oldSyncToken: syncToken, + nextSyncToken: data.next_batch, + }; + if (!syncOptions.hasSyncedBefore) { - self._updateSyncState("PREPARED"); + self._updateSyncState("PREPARED", syncEventData); syncOptions.hasSyncedBefore = true; } // keep emitting SYNCING -> SYNCING for clients who want to do bulk updates - self._updateSyncState("SYNCING"); + self._updateSyncState("SYNCING", syncEventData); self._sync(syncOptions); }, function(err) { @@ -584,6 +590,7 @@ SyncApi.prototype._processSyncResponse = function(syncToken, data) { // next_batch: $token, // presence: { events: [] }, // account_data: { events: [] }, + // device_lists: { changed: ["@user:server", ... ]}, // to_device: { events: [] }, // rooms: { // invite: { @@ -859,6 +866,13 @@ SyncApi.prototype._processSyncResponse = function(syncToken, data) { client.getNotifTimelineSet().addLiveEvent(event); }); } + + // Handle device list updates + if (this.opts.crypto && data.device_lists && data.device_lists.changed) { + data.device_lists.changed.forEach((u) => { + this.opts.crypto.userDeviceListChanged(u); + }); + } }; /** From 9975786bac1210a6a56e6f072fc7a81fb6bc2892 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Wed, 1 Feb 2017 19:21:43 +0000 Subject: [PATCH 4/6] Store the token corresponding to the last device update in localstorage ... so that we can, in future, use it when restarting the client. --- src/crypto/DeviceList.js | 4 ++++ src/store/session/webstorage.js | 24 ++++++++++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/src/crypto/DeviceList.js b/src/crypto/DeviceList.js index 342ff029a..6aa89c05c 100644 --- a/src/crypto/DeviceList.js +++ b/src/crypto/DeviceList.js @@ -326,6 +326,10 @@ export default class DeviceList { this._sessionStore.storeEndToEndDevicesForUser( userId, storage, ); + + if (token) { + this._sessionStore.storeEndToEndDeviceSyncToken(token); + } } }); } diff --git a/src/store/session/webstorage.js b/src/store/session/webstorage.js index d8aa828c3..5e9bffe55 100644 --- a/src/store/session/webstorage.js +++ b/src/store/session/webstorage.js @@ -99,6 +99,27 @@ WebStorageSessionStore.prototype = { return getJsonItem(this.store, keyEndToEndDevicesForUser(userId)); }, + /** + * Store the sync token corresponding to the device list. + * + * This is used when starting the client, to get a list of the users who + * have changed their device list since the list time we were running. + * + * @param {String?} token + */ + storeEndToEndDeviceSyncToken: function(token) { + setJsonItem(this.store, KEY_END_TO_END_DEVICE_SYNC_TOKEN, token); + }, + + /** + * Get the sync token corresponding to the device list. + * + * @return {String?} token + */ + getEndToEndDeviceSyncToken: function() { + return getJsonItem(this.store, KEY_END_TO_END_DEVICE_SYNC_TOKEN); + }, + /** * Store a session between the logged-in user and another device * @param {string} deviceKey The public key of the other device. @@ -180,6 +201,7 @@ WebStorageSessionStore.prototype = { const KEY_END_TO_END_ACCOUNT = E2E_PREFIX + "account"; const KEY_END_TO_END_ANNOUNCED = E2E_PREFIX + "announced"; +const KEY_END_TO_END_DEVICE_SYNC_TOKEN = E2E_PREFIX + "device_sync_token"; function keyEndToEndDevicesForUser(userId) { return E2E_PREFIX + "devices/" + userId; @@ -199,6 +221,8 @@ function keyEndToEndRoom(roomId) { function getJsonItem(store, key) { try { + // if the key is absent, store.getItem() returns null, and + // JSON.parse(null) === null, so this returns null. return JSON.parse(store.getItem(key)); } catch (e) { debuglog("Failed to get key %s: %s", key, e); From 732a764ec66c4f962f5aecd50b6805ec7c940c5c Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Wed, 1 Feb 2017 19:53:51 +0000 Subject: [PATCH 5/6] Refactor crypto initialsync handling Pass a store into the Crypto object so that it doesn't need to make assumptions about the EventEmitter, and use the new metadata on sync events to distinguish between initialsyncs and normal syncs --- src/client.js | 1 + src/crypto/index.js | 47 +++++++++++++++++++++++++++++---------------- 2 files changed, 31 insertions(+), 17 deletions(-) diff --git a/src/client.js b/src/client.js index def7e5cd9..f58f751d9 100644 --- a/src/client.js +++ b/src/client.js @@ -158,6 +158,7 @@ function MatrixClient(opts) { this, this, opts.sessionStore, userId, this.deviceId, + this.store, ); this.olmVersion = Crypto.getOlmVersion(); diff --git a/src/crypto/index.js b/src/crypto/index.js index 3f3fc131a..49ccd47f1 100644 --- a/src/crypto/index.js +++ b/src/crypto/index.js @@ -48,12 +48,16 @@ const DeviceList = require('./DeviceList').default; * @param {string} userId The user ID for the local user * * @param {string} deviceId The identifier for this device. + * + * @param {Object} clientStore the MatrixClient data store. */ -function Crypto(baseApis, eventEmitter, sessionStore, userId, deviceId) { +function Crypto(baseApis, eventEmitter, sessionStore, userId, deviceId, + clientStore) { this._baseApis = baseApis; this._sessionStore = sessionStore; this._userId = userId; this._deviceId = deviceId; + this._clientStore = clientStore; this._olmDevice = new OlmDevice(sessionStore); this._deviceList = new DeviceList(baseApis, sessionStore, this._olmDevice); @@ -107,12 +111,6 @@ function Crypto(baseApis, eventEmitter, sessionStore, userId, deviceId) { function _registerEventHandlers(crypto, eventEmitter) { eventEmitter.on("sync", function(syncState, oldState, data) { try { - if (syncState == "PREPARED") { - // XXX ugh. we're assuming the eventEmitter is a MatrixClient. - // how can we avoid doing so? - const rooms = eventEmitter.getRooms(); - crypto._onInitialSyncCompleted(rooms); - } if (syncState === "SYNCING") { crypto._onSyncCompleted(data); } @@ -721,14 +719,34 @@ Crypto.prototype._onCryptoEvent = function(event) { }; /** - * handle the completion of the initial sync. + * handle the completion of a /sync * - * Announces the new device. + * This is called after the processing of each successful /sync response. + * It is an opportunity to do a batch process on the information received. + * + * @param {Object} syncData the data from the 'MatrixClient.sync' event + */ +Crypto.prototype._onSyncCompleted = function(syncData) { + this._deviceList.lastKnownSyncToken = syncData.nextSyncToken; + + if (!syncData.oldSyncToken) { + // an initialsync. + this._sendNewDeviceEvents(); + } + + // catch up on any new devices we got told about during the sync. + this._deviceList.refreshOutdatedDeviceLists().done(); +}; + +/** + * Send m.new_device messages to any devices we share a room with. + * + * (TODO: we can get rid of this once a suitable number of homeservers and + * clients support the more reliable device list update stream mechanism) * * @private - * @param {module:models/room[]} rooms list of rooms the client knows about */ -Crypto.prototype._onInitialSyncCompleted = function(rooms) { +Crypto.prototype._sendNewDeviceEvents = function() { if (this._sessionStore.getDeviceAnnounced()) { return; } @@ -736,6 +754,7 @@ Crypto.prototype._onInitialSyncCompleted = function(rooms) { // we need to tell all the devices in all the rooms we are members of that // we have arrived. // build a list of rooms for each user. + const rooms = this._clientStore.getRooms(); const roomsByUser = {}; for (let i = 0; i < rooms.length; i++) { const room = rooms[i]; @@ -787,12 +806,6 @@ Crypto.prototype._onInitialSyncCompleted = function(rooms) { }); }; -Crypto.prototype._onSyncCompleted = function(syncData) { - // catch up on any new devices we got told about during the sync. - this._deviceList.lastKnownSyncToken = syncData.nextSyncToken; - this._deviceList.refreshOutdatedDeviceLists().done(); -}; - /** * Handle a key event * From 8d502743a55752ad3b791c55791e72e405b22ed3 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Wed, 1 Feb 2017 19:58:45 +0000 Subject: [PATCH 6/6] Refresh device list on startup On initialsync, call the /keys/changes api to see which users have updated their devices. (On failure, invalidate all of them). --- src/base-apis.js | 22 +++++++++ src/crypto/index.js | 117 +++++++++++++++++++++++++++++++++++++------- 2 files changed, 121 insertions(+), 18 deletions(-) diff --git a/src/base-apis.js b/src/base-apis.js index a6465c701..95b250a94 100644 --- a/src/base-apis.js +++ b/src/base-apis.js @@ -1082,6 +1082,28 @@ MatrixBaseApis.prototype.claimOneTimeKeys = function(devices, key_algorithm) { ); }; +/** + * Ask the server for a list of users who have changed their device lists + * between a pair of sync tokens + * + * @param {string} oldToken + * @param {string} newToken + * + * @return {module:client.Promise} Resolves: result object. Rejects: with + * an error response ({@link module:http-api.MatrixError}). + */ +MatrixBaseApis.prototype.getKeyChanges = function(oldToken, newToken) { + const qps = { + from: oldToken, + to: newToken, + }; + + return this._http.authedRequestWithPrefix( + undefined, "GET", "/keys/changes", qps, undefined, + httpApi.PREFIX_UNSTABLE, + ); +}; + // Identity Server Operations // ========================== diff --git a/src/crypto/index.js b/src/crypto/index.js index 49ccd47f1..ee07b58fd 100644 --- a/src/crypto/index.js +++ b/src/crypto/index.js @@ -732,6 +732,22 @@ Crypto.prototype._onSyncCompleted = function(syncData) { if (!syncData.oldSyncToken) { // an initialsync. this._sendNewDeviceEvents(); + + // if we have a deviceSyncToken, we can tell the deviceList to + // invalidate devices which have changed since then. + const oldSyncToken = this._sessionStore.getEndToEndDeviceSyncToken(); + if (oldSyncToken) { + this._invalidateDeviceListsSince(oldSyncToken).catch((e) => { + // if that failed, we fall back to invalidating everyone. + console.warn("Error fetching changed device list", e); + this._invalidateDeviceListForAllActiveUsers(); + return this._deviceList.flushNewDeviceRequests(); + }).done(); + } else { + // otherwise, we have to invalidate all devices for all users we + // share a room with. + this._invalidateDeviceListForAllActiveUsers(); + } } // catch up on any new devices we got told about during the sync. @@ -754,25 +770,8 @@ Crypto.prototype._sendNewDeviceEvents = function() { // we need to tell all the devices in all the rooms we are members of that // we have arrived. // build a list of rooms for each user. - const rooms = this._clientStore.getRooms(); const roomsByUser = {}; - for (let i = 0; i < rooms.length; i++) { - const room = rooms[i]; - - // check for rooms with encryption enabled - const alg = this._roomEncryptors[room.roomId]; - if (!alg) { - continue; - } - - // ignore any rooms which we have left - const me = room.getMember(this._userId); - if (!me || ( - me.membership !== "join" && me.membership !== "invite" - )) { - continue; - } - + for (const room of this._getE2eRooms()) { const members = room.getJoinedMembers(); for (let j = 0; j < members.length; j++) { const m = members[j]; @@ -806,6 +805,88 @@ Crypto.prototype._sendNewDeviceEvents = function() { }); }; +/** + * Ask the server which users have new devices since a given token, + * invalidate them, and start an update query. + * + * @param {String} oldSyncToken + * + * @returns {Promise} resolves once the query is complete. Rejects if the + * keyChange query fails. + */ +Crypto.prototype._invalidateDeviceListsSince = function(oldSyncToken) { + return this._baseApis.getKeyChanges( + oldSyncToken, this.lastKnownSyncToken, + ).then((r) => { + if (!r.changed || !Array.isArray(r.changed)) { + return; + } + + // only invalidate users we share an e2e room with - we don't + // care about users in non-e2e rooms. + const filteredUserIds = this._getE2eRoomMembers(); + r.changed.forEach((u) => { + if (u in filteredUserIds) { + this._deviceList.invalidateUserDeviceList(u); + } + }); + return this._deviceList.flushNewDeviceRequests(); + }); +}; + +/** + * Invalidate any stored device list for any users we share an e2e room with + * + * @private + */ +Crypto.prototype._invalidateDeviceListForAllActiveUsers = function() { + Object.keys(this._getE2eRoomMembers()).forEach((m) => { + this._deviceList.invalidateUserDeviceList(m); + }); +}; + +/** + * get the users we share an e2e-enabled room with + * + * @returns {Object} userid->userid map (should be a Set but argh ES6) + */ +Crypto.prototype._getE2eRoomMembers = function() { + const userIds = Object.create(null); + + const rooms = this._getE2eRooms(); + for (const r of rooms) { + const members = r.getJoinedMembers(); + members.forEach((m) => { userIds[m.userId] = m.userId; }); + } + + return userIds; +}; + +/** + * Get a list of the e2e-enabled rooms we are members of + * + * @returns {module:models.Room[]} + */ +Crypto.prototype._getE2eRooms = function() { + return this._clientStore.getRooms().filter((room) => { + // check for rooms with encryption enabled + const alg = this._roomEncryptors[room.roomId]; + if (!alg) { + return false; + } + + // ignore any rooms which we have left + const me = room.getMember(this._userId); + if (!me || ( + me.membership !== "join" && me.membership !== "invite" + )) { + return false; + } + + return true; + }); +}; + /** * Handle a key event *