Process m.room.encryption events before emitting RoomMember events (#2914)

vector-im/element-web#23819 is an intermittent failure to correctly initiate a user verification process. The
root cause is as follows:

* In matrix-react-sdk, ensureDMExists tries to create an encrypted DM room, and assumes it is ready for use
  (including sending encrypted events) as soon as it receives a RoomStateEvent.NewMember notification
  indicating that the other user has been invited or joined. 

* However, in sync.ts, we process the membership events in a /sync response (including emitting
  RoomStateEvent.NewMember notifications), which is long before we process any m.room.encryption event.
    
* The upshot is that we can end up trying to send an encrypted event in the new room before processing
  the m.room.encryption event, which causes the crypto layer to blow up with an error of "Room was 
  previously configured to use encryption, but is no longer".

Strictly speaking, ensureDMExists probably ought to be listening for ClientEvent.Room as well as RoomStateEvent.NewMember; but that doesn't help us, because ClientEvent.Room is also emitted
before we process the crypto event.

So, we need to process the crypto event before we start emitting these other events; but a corollary of that 
is that we need to do so before we store the new room in the client's store. That makes things tricky, because
currently the crypto layer expects the room to have been stored in the client first.

So... we have to rearrange everything to pass the newly-created Room object into the crypto layer, rather than
just the room id, so that it doesn't need to rely on getting the Room from the client's store.

spec/integ/megolm-integ.spec.ts

@@ -21,16 +21,19 @@ import * as testUtils from "../test-utils/test-utils";
 import { TestClient } from "../TestClient";
 import { logger } from "../../src/logger";
 import {
-    IContent,
-    IEvent,
     IClaimOTKsResult,
-    IJoinedRoom,
+    IContent,
-    ISyncResponse,
     IDownloadKeyResult,
+    IEvent,
+    IJoinedRoom,
+    IndexedDBCryptoStore,
+    ISyncResponse,
+    IUploadKeysRequest,
     MatrixEvent,
     MatrixEventEvent,
-    IndexedDBCryptoStore,
     Room,
+    RoomMember,
+    RoomStateEvent,
 } from "../../src/matrix";
 import { IDeviceKeys } from "../../src/crypto/dehydration";
 import { DeviceInfo } from "../../src/crypto/deviceinfo";
@@ -327,7 +330,9 @@ describe("megolm", () => {
         const room = aliceTestClient.client.getRoom(ROOM_ID)!;
         const event = room.getLiveTimeline().getEvents()[0];
         expect(event.isEncrypted()).toBe(true);
-        const decryptedEvent = await testUtils.awaitDecryption(event);
+
+        // it probably won't be decrypted yet, because it takes a while to process the olm keys
+        const decryptedEvent = await testUtils.awaitDecryption(event, { waitOnDecryptionFailure: true });
         expect(decryptedEvent.getContent().body).toEqual('42');
     });
 
@@ -873,7 +878,12 @@ describe("megolm", () => {
 
         const room = aliceTestClient.client.getRoom(ROOM_ID)!;
         await room.decryptCriticalEvents();
-        expect(room.getLiveTimeline().getEvents()[0].getContent().body).toEqual('42');
+
+        // it probably won't be decrypted yet, because it takes a while to process the olm keys
+        const decryptedEvent = await testUtils.awaitDecryption(
+            room.getLiveTimeline().getEvents()[0], { waitOnDecryptionFailure: true },
+        );
+        expect(decryptedEvent.getContent().body).toEqual('42');
 
         const exported = await aliceTestClient.client.exportRoomKeys();
 
@@ -1012,7 +1022,9 @@ describe("megolm", () => {
         const room = aliceTestClient.client.getRoom(ROOM_ID)!;
         const event = room.getLiveTimeline().getEvents()[0];
         expect(event.isEncrypted()).toBe(true);
-        const decryptedEvent = await testUtils.awaitDecryption(event);
+
+        // it probably won't be decrypted yet, because it takes a while to process the olm keys
+        const decryptedEvent = await testUtils.awaitDecryption(event, { waitOnDecryptionFailure: true });
         expect(decryptedEvent.getRoomId()).toEqual(ROOM_ID);
         expect(decryptedEvent.getContent()).toEqual({});
         expect(decryptedEvent.getClearContent()).toBeUndefined();
@@ -1364,4 +1376,90 @@ describe("megolm", () => {
 
         await beccaTestClient.stop();
     });
+
+    it("allows sending an encrypted event as soon as room state arrives", async () => {
+        /* Empirically, clients expect to be able to send encrypted events as soon as the
+         * RoomStateEvent.NewMember notification is emitted, so test that works correctly.
+         */
+        const testRoomId = "!testRoom:id";
+        await aliceTestClient.start();
+
+        aliceTestClient.httpBackend.when("POST", "/keys/query")
+            .respond(200, function(_path, content: IUploadKeysRequest) {
+                return { device_keys: {} };
+            });
+
+        /* Alice makes the /createRoom call */
+        aliceTestClient.httpBackend.when("POST", "/createRoom")
+            .respond(200, { room_id: testRoomId });
+        await Promise.all([
+            aliceTestClient.client.createRoom({
+                initial_state: [{
+                    type: 'm.room.encryption',
+                    state_key: '',
+                    content: { algorithm: 'm.megolm.v1.aes-sha2' },
+                }],
+            }),
+            aliceTestClient.httpBackend.flushAllExpected(),
+        ]);
+
+        /* The sync arrives in two parts; first the m.room.create... */
+        aliceTestClient.httpBackend.when("GET", "/sync").respond(200, {
+            rooms: { join: {
+                [testRoomId]: {
+                    timeline: { events: [
+                        {
+                            type: 'm.room.create',
+                            state_key: '',
+                            event_id: "$create",
+                        },
+                        {
+                            type: 'm.room.member',
+                            state_key: aliceTestClient.getUserId(),
+                            content: { membership: "join" },
+                            event_id: "$alijoin",
+                        },
+                    ] },
+                },
+            } },
+        });
+        await aliceTestClient.flushSync();
+
+        // ... and then the e2e event and an invite ...
+        aliceTestClient.httpBackend.when("GET", "/sync").respond(200, {
+            rooms: { join: {
+                [testRoomId]: {
+                    timeline: { events: [
+                        {
+                            type: 'm.room.encryption',
+                            state_key: '',
+                            content: { algorithm: 'm.megolm.v1.aes-sha2' },
+                            event_id: "$e2e",
+                        },
+                        {
+                            type: 'm.room.member',
+                            state_key: "@other:user",
+                            content: { membership: "invite" },
+                            event_id: "$otherinvite",
+                        },
+                    ] },
+                },
+            } },
+        });
+
+        // as soon as the roomMember arrives, try to send a message
+        aliceTestClient.client.on(RoomStateEvent.NewMember, (_e, _s, member: RoomMember) => {
+            if (member.userId == "@other:user") {
+                aliceTestClient.client.sendMessage(testRoomId, { msgtype: "m.text", body: "Hello, World" });
+            }
+        });
+
+        // flush the sync and wait for the /send/ request.
+        aliceTestClient.httpBackend.when("PUT", "/send/m.room.encrypted/")
+            .respond(200, (_path, _content) => ({ event_id: "asdfgh" }));
+        await Promise.all([
+            aliceTestClient.flushSync(),
+            aliceTestClient.httpBackend.flush("/send/m.room.encrypted/", 1),
+        ]);
+    });
 });

spec/test-utils/test-utils.ts

@@ -362,22 +362,28 @@ export class MockStorageApi {
  * @param {MatrixEvent} event
  * @returns {Promise} promise which resolves (to `event`) when the event has been decrypted
  */
-export async function awaitDecryption(event: MatrixEvent): Promise<MatrixEvent> {
+export async function awaitDecryption(
+    event: MatrixEvent, { waitOnDecryptionFailure = false } = {},
+): Promise<MatrixEvent> {
     // An event is not always decrypted ahead of time
     // getClearContent is a good signal to know whether an event has been decrypted
     // already
     if (event.getClearContent() !== null) {
-        return event;
+        if (waitOnDecryptionFailure && event.isDecryptionFailure()) {
+            logger.log(`${Date.now()} event ${event.getId()} got decryption error; waiting`);
+        } else {
+            return event;
+        }
     } else {
-        logger.log(`${Date.now()} event ${event.getId()} is being decrypted; waiting`);
+        logger.log(`${Date.now()} event ${event.getId()} is not yet decrypted; waiting`);
-
-        return new Promise((resolve) => {
-            event.once(MatrixEventEvent.Decrypted, (ev) => {
-                logger.log(`${Date.now()} event ${event.getId()} now decrypted`);
-                resolve(ev);
-            });
-        });
     }
+
+    return new Promise((resolve) => {
+        event.once(MatrixEventEvent.Decrypted, (ev) => {
+            logger.log(`${Date.now()} event ${event.getId()} now decrypted`);
+            resolve(ev);
+        });
+    });
 }
 
 export const emitPromise = (e: EventEmitter, k: string): Promise<any> => new Promise(r => e.once(k, r));

spec/unit/crypto.spec.ts

@@ -19,6 +19,7 @@ import { MemoryStore } from "../../src";
 import { RoomKeyRequestState } from '../../src/crypto/OutgoingRoomKeyRequestManager';
 import { RoomMember } from '../../src/models/room-member';
 import { IStore } from '../../src/store';
+import { IRoomEncryption, RoomList } from "../../src/crypto/RoomList";
 
 const Olm = global.Olm;
 
@@ -1140,4 +1141,48 @@ describe("Crypto", function() {
             await client!.client.crypto!.start();
         });
     });
+
+    describe("setRoomEncryption", () => {
+        let mockClient: MatrixClient;
+        let mockRoomList: RoomList;
+        let clientStore: IStore;
+        let crypto: Crypto;
+
+        beforeEach(async function() {
+            mockClient = {} as MatrixClient;
+            const mockStorage = new MockStorageApi() as unknown as Storage;
+            clientStore = new MemoryStore({ localStorage: mockStorage }) as unknown as IStore;
+            const cryptoStore = new MemoryCryptoStore();
+
+            mockRoomList = {
+                getRoomEncryption: jest.fn().mockReturnValue(null),
+                setRoomEncryption: jest.fn().mockResolvedValue(undefined),
+            } as unknown as RoomList;
+
+            crypto = new Crypto(
+                mockClient,
+                "@alice:home.server",
+                "FLIBBLE",
+                clientStore,
+                cryptoStore,
+                mockRoomList,
+                [],
+            );
+        });
+
+        it("should set the algorithm if called for a known room", async () => {
+            const room = new Room("!room:id", mockClient, "@my.user:id");
+            await clientStore.storeRoom(room);
+            await crypto.setRoomEncryption("!room:id", { algorithm: "m.megolm.v1.aes-sha2" } as IRoomEncryption);
+            expect(mockRoomList!.setRoomEncryption).toHaveBeenCalledTimes(1);
+            expect(jest.mocked(mockRoomList!.setRoomEncryption).mock.calls[0][0]).toEqual("!room:id");
+        });
+
+        it("should raise if called for an unknown room", async () => {
+            await expect(async () => {
+                await crypto.setRoomEncryption("!room:id", { algorithm: "m.megolm.v1.aes-sha2" } as IRoomEncryption);
+            }).rejects.toThrow(/unknown room/);
+            expect(mockRoomList!.setRoomEncryption).not.toHaveBeenCalled();
+        });
+    });
 });

src/crypto/index.ts

@@ -86,6 +86,7 @@ import { ISyncStateData } from "../sync";
 import { CryptoStore } from "./store/base";
 import { IVerificationChannel } from "./verification/request/Channel";
 import { TypedEventEmitter } from "../models/typed-event-emitter";
+import { IContent } from "../models/event";
 
 const DeviceVerification = DeviceInfo.DeviceVerification;
 
@@ -2552,12 +2553,45 @@ export class Crypto extends TypedEventEmitter<CryptoEvent, CryptoEventHandlerMap
      * @param {boolean=} inhibitDeviceQuery true to suppress device list query for
      *   users in the room (for now). In case lazy loading is enabled,
      *   the device query is always inhibited as the members are not tracked.
+     *
+     * @deprecated It is normally incorrect to call this method directly. Encryption
+     *   is enabled by receiving an `m.room.encryption` event (which we may have sent
+     *   previously).
      */
     public async setRoomEncryption(
         roomId: string,
         config: IRoomEncryption,
         inhibitDeviceQuery?: boolean,
     ): Promise<void> {
+        const room = this.clientStore.getRoom(roomId);
+        if (!room) {
+            throw new Error(`Unable to enable encryption tracking devices in unknown room ${roomId}`);
+        }
+        await this.setRoomEncryptionImpl(room, config);
+        if (!this.lazyLoadMembers && !inhibitDeviceQuery) {
+            this.deviceList.refreshOutdatedDeviceLists();
+        }
+    }
+
+    /**
+     * Set up encryption for a room.
+     *
+     * This is called when an <tt>m.room.encryption</tt> event is received. It saves a flag
+     * for the room in the cryptoStore (if it wasn't already set), sets up an "encryptor" for
+     * the room, and enables device-list tracking for the room.
+     *
+     * It does <em>not</em> initiate a device list query for the room. That is normally
+     * done once we finish processing the sync, in onSyncCompleted.
+     *
+     * @param room The room to enable encryption in.
+     * @param config The encryption config for the room.
+     */
+    private async setRoomEncryptionImpl(
+        room: Room,
+        config: IRoomEncryption,
+    ): Promise<void> {
+        const roomId = room.roomId;
+
         // ignore crypto events with no algorithm defined
         // This will happen if a crypto event is redacted before we fetch the room state
         // It would otherwise just throw later as an unknown algorithm would, but we may
@@ -2625,14 +2659,7 @@ export class Crypto extends TypedEventEmitter<CryptoEvent, CryptoEventHandlerMap
             logger.log("Enabling encryption in " + roomId + "; " +
                 "starting to track device lists for all users therein");
 
-            await this.trackRoomDevices(roomId);
+            await this.trackRoomDevicesImpl(room);
-            // TODO: this flag is only not used from MatrixClient::setRoomEncryption
-            // which is never used (inside Element at least)
-            // but didn't want to remove it as it technically would
-            // be a breaking change.
-            if (!inhibitDeviceQuery) {
-                this.deviceList.refreshOutdatedDeviceLists();
-            }
         } else {
             logger.log("Enabling encryption in " + roomId);
         }
@@ -2645,15 +2672,30 @@ export class Crypto extends TypedEventEmitter<CryptoEvent, CryptoEventHandlerMap
      * @returns {Promise} when all devices for the room have been fetched and marked to track
      */
     public trackRoomDevices(roomId: string): Promise<void> {
+        const room = this.clientStore.getRoom(roomId);
+        if (!room) {
+            throw new Error(`Unable to start tracking devices in unknown room ${roomId}`);
+        }
+        return this.trackRoomDevicesImpl(room);
+    }
+
+    /**
+     * Make sure we are tracking the device lists for all users in this room.
+     *
+     * This is normally called when we are about to send an encrypted event, to make sure
+     * we have all the devices in the room; but it is also called when processing an
+     * m.room.encryption state event (if lazy-loading is disabled), or when members are
+     * loaded (if lazy-loading is enabled), to prepare the device list.
+     *
+     * @param room Room to enable device-list tracking in
+     */
+    private trackRoomDevicesImpl(room: Room): Promise<void> {
+        const roomId = room.roomId;
         const trackMembers = async (): Promise<void> => {
             // not an encrypted room
             if (!this.roomEncryptors.has(roomId)) {
                 return;
             }
-            const room = this.clientStore.getRoom(roomId);
-            if (!room) {
-                throw new Error(`Unable to start tracking devices in unknown room ${roomId}`);
-            }
             logger.log(`Starting to track devices for room ${roomId} ...`);
             const members = await room.getEncryptionTargetMembers();
             members.forEach((m) => {
@@ -2815,17 +2857,14 @@ export class Crypto extends TypedEventEmitter<CryptoEvent, CryptoEventHandlerMap
             // MatrixClient has already checked that this room should be encrypted,
             // so this is an unexpected situation.
             throw new Error(
-                "Room was previously configured to use encryption, but is " +
+                "Room " + roomId + " was previously configured to use encryption, but is " +
                 "no longer. Perhaps the homeserver is hiding the " +
                 "configuration event.",
             );
         }
 
-        if (!this.roomDeviceTrackingState[roomId]) {
-            this.trackRoomDevices(roomId);
-        }
         // wait for all the room devices to be loaded
-        await this.roomDeviceTrackingState[roomId];
+        await this.trackRoomDevicesImpl(room);
 
         let content = event.getContent();
         // If event has an m.relates_to then we need
@@ -2844,7 +2883,7 @@ export class Crypto extends TypedEventEmitter<CryptoEvent, CryptoEventHandlerMap
             delete content['io.element.performance_metrics'];
         }
 
-        const encryptedContent = await alg.encryptMessage(room, event.getType(), content);
+        const encryptedContent = await alg.encryptMessage(room, event.getType(), content) as IContent;
 
         if (mRelatesTo) {
             encryptedContent['m.relates_to'] = mRelatesTo;
@@ -2972,19 +3011,12 @@ export class Crypto extends TypedEventEmitter<CryptoEvent, CryptoEventHandlerMap
     /**
      * handle an m.room.encryption event
      *
-     * @param {module:models/event.MatrixEvent} event encryption event
+     * @param room in which the event was received
+     * @param event encryption event to be processed
      */
-    public async onCryptoEvent(event: MatrixEvent): Promise<void> {
+    public async onCryptoEvent(room: Room, event: MatrixEvent): Promise<void> {
-        const roomId = event.getRoomId()!;
         const content = event.getContent<IRoomEncryption>();
-
+        await this.setRoomEncryptionImpl(room, content);
-        try {
-            // inhibit the device list refresh for now - it will happen once we've
-            // finished processing the sync, in onSyncCompleted.
-            await this.setRoomEncryption(roomId, content, true);
-        } catch (e) {
-            logger.error(`Error configuring encryption in room ${roomId}`, e);
-        }
     }
 
     /**

src/sliding-sync-sdk.ts

@@ -703,7 +703,7 @@ export class SlidingSyncSdk {
         const processRoomEvent = async (e: MatrixEvent): Promise<void> => {
             client.emit(ClientEvent.Event, e);
             if (e.isState() && e.getType() == EventType.RoomEncryption && this.opts.crypto) {
-                await this.opts.crypto.onCryptoEvent(e);
+                await this.opts.crypto.onCryptoEvent(room, e);
             }
         };
 

src/sync.ts

@@ -1362,6 +1362,18 @@ export class SyncApi {
                 }
             }
 
+            // process any crypto events *before* emitting the RoomStateEvent events. This
+            // avoids a race condition if the application tries to send a message after the
+            // state event is processed, but before crypto is enabled, which then causes the
+            // crypto layer to complain.
+            if (this.opts.crypto) {
+                for (const e of stateEvents.concat(events)) {
+                    if (e.isState() && e.getType() === EventType.RoomEncryption && e.getStateKey() === "") {
+                        await this.opts.crypto.onCryptoEvent(room, e);
+                    }
+                }
+            }
+
             try {
                 await this.injectRoomEvents(room, stateEvents, events, syncEventData.fromCache);
             } catch (e) {
@@ -1389,21 +1401,11 @@ export class SyncApi {
 
             this.processEventsForNotifs(room, events);
 
-            const processRoomEvent = async (e): Promise<void> => {
+            const emitEvent = (e: MatrixEvent): boolean => client.emit(ClientEvent.Event, e);
-                client.emit(ClientEvent.Event, e);
+            stateEvents.forEach(emitEvent);
-                if (e.isState() && e.getType() == "m.room.encryption" && this.opts.crypto) {
+            events.forEach(emitEvent);
-                    await this.opts.crypto.onCryptoEvent(e);
+            ephemeralEvents.forEach(emitEvent);
-                }
+            accountDataEvents.forEach(emitEvent);
-            };
-
-            await utils.promiseMapSeries(stateEvents, processRoomEvent);
-            await utils.promiseMapSeries(events, processRoomEvent);
-            ephemeralEvents.forEach(function(e) {
-                client.emit(ClientEvent.Event, e);
-            });
-            accountDataEvents.forEach(function(e) {
-                client.emit(ClientEvent.Event, e);
-            });
 
             // Decrypt only the last message in all rooms to make sure we can generate a preview
             // And decrypt all events after the recorded read receipt to ensure an accurate