Merge remote-tracking branch 'origin/develop' into rav/element-r/crypto_backend

Author: richvdh

.prettierignore

@@ -11,6 +11,9 @@ lib-cov
 out
 /dist
 /lib
+/examples/browser/lib
+/examples/crypto-browser/lib
+/examples/voip/lib
 
 # version file and tarball created by `npm pack` / `yarn pack`
 /git-revision.txt

spec/integ/megolm-integ.spec.ts

@@ -204,6 +204,117 @@ function getSyncResponse(roomMembers: string[]): ISyncResponse {
     };
 }
 
+/**
+ * Establish an Olm Session with the test user
+ *
+ * Waits for the test user to upload their keys, then sends a /sync response with a to-device message which will
+ * establish an Olm session.
+ *
+ * @param testClient: a TestClient for the user under test, which we expect to upload account keys, and to make a
+ *    /sync request which we will respond to.
+ * @param peerOlmAccount: an OlmAccount which will be used to initiate the Olm session.
+ */
+async function establishOlmSession(testClient: TestClient, peerOlmAccount: Olm.Account): Promise<Olm.Session> {
+    const peerE2EKeys = JSON.parse(peerOlmAccount.identity_keys());
+    const p2pSession = await createOlmSession(peerOlmAccount, testClient);
+    const olmEvent = encryptOlmEvent({
+        senderKey: peerE2EKeys.curve25519,
+        recipient: testClient,
+        p2pSession: p2pSession,
+    });
+    testClient.httpBackend.when("GET", "/sync").respond(200, {
+        next_batch: 1,
+        to_device: { events: [olmEvent] },
+    });
+    await testClient.flushSync();
+    return p2pSession;
+}
+
+/**
+ * Expect that the client shares keys with the given recipient
+ *
+ * Waits for an HTTP request to send the encrypted m.room_key to-device message; decrypts it and uses it
+ * to establish an Olm InboundGroupSession.
+ *
+ * @param senderMockHttpBackend - MockHttpBackend for the sender
+ *
+ * @param recipientUserID - the user id of the expected recipient
+ *
+ * @param recipientOlmAccount - Olm.Account for the recipient
+ *
+ * @param recipientOlmSession - an Olm.Session for the recipient, which must already have exchanged pre-key
+ *    messages with the sender. Alternatively, null, in which case we will expect a pre-key message.
+ *
+ * @returns the established inbound group session
+ */
+async function expectSendRoomKey(
+    senderMockHttpBackend: MockHttpBackend,
+    recipientUserID: string,
+    recipientOlmAccount: Olm.Account,
+    recipientOlmSession: Olm.Session | null = null,
+): Promise<Olm.InboundGroupSession> {
+    const Olm = global.Olm;
+    const testRecipientKey = JSON.parse(recipientOlmAccount.identity_keys())["curve25519"];
+
+    let inboundGroupSession: Olm.InboundGroupSession;
+
+    senderMockHttpBackend.when("PUT", "/sendToDevice/m.room.encrypted/").respond(200, (_path, content: any) => {
+        const m = content.messages[recipientUserID].DEVICE_ID;
+        const ct = m.ciphertext[testRecipientKey];
+
+        if (!recipientOlmSession) {
+            expect(ct.type).toEqual(0); // pre-key message
+            recipientOlmSession = new Olm.Session();
+            recipientOlmSession.create_inbound(recipientOlmAccount, ct.body);
+        } else {
+            expect(ct.type).toEqual(1); // regular message
+        }
+
+        const decrypted = JSON.parse(recipientOlmSession.decrypt(ct.type, ct.body));
+        expect(decrypted.type).toEqual("m.room_key");
+        inboundGroupSession = new Olm.InboundGroupSession();
+        inboundGroupSession.create(decrypted.content.session_key);
+        return {};
+    });
+
+    expect(await senderMockHttpBackend.flush("/sendToDevice/m.room.encrypted/", 1, 1000)).toEqual(1);
+    return inboundGroupSession!;
+}
+
+/**
+ * Expect that the client sends an encrypted event
+ *
+ * Waits for an HTTP request to send an encrypted message in the test room.
+ *
+ * @param senderMockHttpBackend - MockHttpBackend for the sender
+ *
+ * @param inboundGroupSessionPromise - a promise for an Olm InboundGroupSession, which will
+ *    be used to decrypt the event. We will wait for this to resolve once the HTTP request has been processed.
+ *
+ * @returns The content of the successfully-decrypted event
+ */
+async function expectSendMegolmMessage(
+    senderMockHttpBackend: MockHttpBackend,
+    inboundGroupSessionPromise: Promise<Olm.InboundGroupSession>,
+): Promise<Partial<IEvent>> {
+    let encryptedMessageContent: IContent | null = null;
+    senderMockHttpBackend.when("PUT", "/send/m.room.encrypted/").respond(200, function (_path, content: IContent) {
+        encryptedMessageContent = content;
+        return {
+            event_id: "$event_id",
+        };
+    });
+
+    expect(await senderMockHttpBackend.flush("/send/m.room.encrypted/", 1, 1000)).toEqual(1);
+
+    // In some of the tests, the room key is sent *after* the actual event, so we may need to wait for it now.
+    const inboundGroupSession = await inboundGroupSessionPromise;
+
+    const r: any = inboundGroupSession.decrypt(encryptedMessageContent!.ciphertext);
+    logger.log("Decrypted received megolm message", r);
+    return JSON.parse(r.plaintext);
+}
+
 describe("megolm", () => {
     if (!global.Olm) {
         logger.warn("not running megolm tests: Olm not present");
@@ -491,9 +602,16 @@ describe("megolm", () => {
         const room = aliceTestClient.client.getRoom(ROOM_ID)!;
         const pendingMsg = room.getPendingEvents()[0];
 
+        const inboundGroupSessionPromise = expectSendRoomKey(
+            aliceTestClient.httpBackend,
+            "@bob:xyz",
+            testOlmAccount,
+            p2pSession,
+        );
+
         await Promise.all([
             aliceTestClient.client.resendEvent(pendingMsg, room),
-            expectSendKeyAndMessage(aliceTestClient.httpBackend, "@bob:xyz", testOlmAccount, p2pSession),
+            expectSendMegolmMessage(aliceTestClient.httpBackend, inboundGroupSessionPromise),
         ]);
     });
 

spec/unit/crypto/algorithms/megolm.spec.ts

@@ -490,6 +490,26 @@ describe("MegolmDecryption", function () {
 
                 expect(mockBaseApis.queueToDevice).not.toHaveBeenCalled();
             });
+
+            it("shouldn't wedge the setup promise if sharing a room key fails", async () => {
+                // @ts-ignore - private field access
+                const initialSetupPromise = await megolmEncryption.setupPromise;
+                expect(initialSetupPromise).toBe(null);
+
+                // @ts-ignore - private field access
+                megolmEncryption.prepareSession = () => {
+                    throw new Error("Can't prepare session");
+                };
+
+                await expect(() =>
+                    // @ts-ignore - private field access
+                    megolmEncryption.ensureOutboundSession(mockRoom, {}, {}, true),
+                ).rejects.toThrow();
+
+                // @ts-ignore - private field access
+                const finalSetupPromise = await megolmEncryption.setupPromise;
+                expect(finalSetupPromise).toBe(null);
+            });
         });
     });
 

src/crypto/algorithms/megolm.ts

@@ -243,6 +243,23 @@ export class MegolmEncryption extends EncryptionAlgorithm {
      * @param singleOlmCreationPhase - Only perform one round of olm
      *     session creation
      *
+     * This method updates the setupPromise field of the class by chaining a new
+     * call on top of the existing promise, and then catching and discarding any
+     * errors that might happen while setting up the outbound group session. This
+     * is done to ensure that `setupPromise` always resolves to `null` or the
+     * `OutboundSessionInfo`.
+     *
+     * Using `>>=` to represent the promise chaining operation, it does the
+     * following:
+     *
+     * ```
+     * setupPromise = previousSetupPromise >>= setup >>= discardErrors
+     * ```
+     *
+     * The initial value for the `setupPromise` is a promise that resolves to
+     * `null`. The forceDiscardSession() resets setupPromise to this initial
+     * promise.
+     *
      * @returns Promise which resolves to the
      *    OutboundSessionInfo when setup is complete.
      */
@@ -255,36 +272,31 @@ export class MegolmEncryption extends EncryptionAlgorithm {
         // takes the previous OutboundSessionInfo, and considers whether to create
         // a new one. Also shares the key with any (new) devices in the room.
         //
-        // Returns the successful session whether keyshare succeeds or not.
-        //
         // returns a promise which resolves once the keyshare is successful.
         const setup = async (oldSession: OutboundSessionInfo | null): Promise<OutboundSessionInfo> => {
             const sharedHistory = isRoomSharedHistory(room);
-
             const session = await this.prepareSession(devicesInRoom, sharedHistory, oldSession);
 
-            try {
-                await this.shareSession(devicesInRoom, sharedHistory, singleOlmCreationPhase, blocked, session);
-            } catch (e) {
-                logger.error(`Failed to ensure outbound session in ${this.roomId}`, e);
-            }
+            await this.shareSession(devicesInRoom, sharedHistory, singleOlmCreationPhase, blocked, session);
 
             return session;
         };
 
         // first wait for the previous share to complete
-        const prom = this.setupPromise.then(setup);
+        const fallible = this.setupPromise.then(setup);
 
-        // Ensure any failures are logged for debugging
-        prom.catch((e) => {
+        // Ensure any failures are logged for debugging and make sure that the
+        // promise chain remains unbroken
+        //
+        // setupPromise resolves to `null` or the `OutboundSessionInfo` whether
+        // or not the share succeeds
+        this.setupPromise = fallible.catch((e) => {
             logger.error(`Failed to setup outbound session in ${this.roomId}`, e);
+            return null;
         });
 
-        // setupPromise resolves to `session` whether or not the share succeeds
-        this.setupPromise = prom;
-
         // but we return a promise which only resolves if the share was successful.
-        return prom;
+        return fallible;
     }
 
     private async prepareSession(

src/models/room.ts

@@ -2079,8 +2079,13 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
             room: this,
             client: this.client,
             pendingEventOrdering: this.opts.pendingEventOrdering,
+            receipts: this.cachedThreadReadReceipts.get(threadId) ?? [],
         });
 
+        // All read receipts should now come down from sync, we do not need to keep
+        // a reference to the cached receipts anymore.
+        this.cachedThreadReadReceipts.delete(threadId);
+
         // This is necessary to be able to jump to events in threads:
         // If we jump to an event in a thread where neither the event, nor the root,
         // nor any thread event are loaded yet, we'll load the event as well as the thread root, create the thread,
@@ -2110,17 +2115,6 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
         if (this.threadsReady) {
             this.updateThreadRootEvents(thread, toStartOfTimeline, false);
         }
-
-        // Pulling all the cached thread read receipts we've discovered when we
-        // did an initial sync, and applying them to the thread now that it exists
-        // on the client side
-        if (this.cachedThreadReadReceipts.has(threadId)) {
-            for (const { event, synthetic } of this.cachedThreadReadReceipts.get(threadId)!) {
-                this.addReceipt(event, synthetic);
-            }
-            this.cachedThreadReadReceipts.delete(threadId);
-        }
-
         this.emit(ThreadEvent.New, thread, toStartOfTimeline);
 
         return thread;

src/models/thread.ts

@@ -20,14 +20,14 @@ import { MatrixClient, PendingEventOrdering } from "../client";
 import { TypedReEmitter } from "../ReEmitter";
 import { RelationType } from "../@types/event";
 import { EventStatus, IThreadBundledRelationship, MatrixEvent, MatrixEventEvent } from "./event";
-import { EventTimeline } from "./event-timeline";
+import { Direction, EventTimeline } from "./event-timeline";
 import { EventTimelineSet, EventTimelineSetHandlerMap } from "./event-timeline-set";
 import { NotificationCountType, Room, RoomEvent } from "./room";
 import { RoomState } from "./room-state";
 import { ServerControlledNamespacedValue } from "../NamespacedValue";
 import { logger } from "../logger";
 import { ReadReceipt } from "./read-receipt";
-import { ReceiptType } from "../@types/read_receipts";
+import { Receipt, ReceiptContent, ReceiptType } from "../@types/read_receipts";
 
 export enum ThreadEvent {
     New = "Thread.new",
@@ -50,6 +50,7 @@ interface IThreadOpts {
     room: Room;
     client: MatrixClient;
     pendingEventOrdering?: PendingEventOrdering;
+    receipts?: { event: MatrixEvent; synthetic: boolean }[];
 }
 
 export enum FeatureSupport {
@@ -127,6 +128,8 @@ export class Thread extends ReadReceipt<EmittedEvents, EventHandlerMap> {
         this.room.on(RoomEvent.LocalEchoUpdated, this.onEcho);
         this.timelineSet.on(RoomEvent.Timeline, this.onTimelineEvent);
 
+        this.processReceipts(opts.receipts);
+
         // even if this thread is thought to be originating from this client, we initialise it as we may be in a
         // gappy sync and a thread around this event may already exist.
         this.updateThreadMetadata();
@@ -284,6 +287,26 @@ export class Thread extends ReadReceipt<EmittedEvents, EventHandlerMap> {
         this.timeline = this.events;
     }
 
+    /**
+     * Processes the receipts that were caught during initial sync
+     * When clients become aware of a thread, they try to retrieve those read receipts
+     * and apply them to the current thread
+     * @param receipts - A collection of the receipts cached from initial sync
+     */
+    private processReceipts(receipts: { event: MatrixEvent; synthetic: boolean }[] = []): void {
+        for (const { event, synthetic } of receipts) {
+            const content = event.getContent<ReceiptContent>();
+            Object.keys(content).forEach((eventId: string) => {
+                Object.keys(content[eventId]).forEach((receiptType: ReceiptType | string) => {
+                    Object.keys(content[eventId][receiptType]).forEach((userId: string) => {
+                        const receipt = content[eventId][receiptType][userId] as Receipt;
+                        this.addReceiptToStructure(eventId, receiptType as ReceiptType, userId, receipt, synthetic);
+                    });
+                });
+            });
+        }
+    }
+
     private getRootEventBundledRelationship(rootEvent = this.rootEvent): IThreadBundledRelationship | undefined {
         return rootEvent?.getServerAggregatedRelation<IThreadBundledRelationship>(THREAD_RELATION_TYPE.name);
     }
@@ -335,7 +358,15 @@ export class Thread extends ReadReceipt<EmittedEvents, EventHandlerMap> {
             try {
                 // if the thread has regular events, this will just load the last reply.
                 // if the thread is newly created, this will load the root event.
-                await this.client.paginateEventTimeline(this.liveTimeline, { backwards: true, limit: 1 });
+                if (this.replyCount === 0 && this.rootEvent) {
+                    this.timelineSet.addEventsToTimeline([this.rootEvent], true, this.liveTimeline, null);
+                    this.liveTimeline.setPaginationToken(null, Direction.Backward);
+                } else {
+                    await this.client.paginateEventTimeline(this.liveTimeline, {
+                        backwards: true,
+                        limit: Math.max(1, this.length),
+                    });
+                }
                 // just to make sure that, if we've created a timeline window for this thread before the thread itself
                 // existed (e.g. when creating a new thread), we'll make sure the panel is force refreshed correctly.
                 this.emit(RoomEvent.TimelineReset, this.room, this.timelineSet, true);