refactor: extract RoomKeyTransport class for key distribution

spec/unit/matrixrtc/MatrixRTCSession.spec.ts

@@ -584,7 +584,7 @@ describe("MatrixRTCSession", () => {
                     });
 
                     sess!.joinRoomSession([mockFocus], mockFocus, { manageMediaKeys: true });
-                    jest.advanceTimersByTime(10000);
+                    await jest.runAllTimersAsync();
 
                     await eventSentPromise;
 

src/matrixrtc/EncryptionManager.ts

@@ -1,14 +1,12 @@
-import { type MatrixClient } from "../client.ts";
 import { logger as rootLogger } from "../logger.ts";
 import { type MatrixEvent } from "../models/event.ts";
-import { type Room } from "../models/room.ts";
 import { type EncryptionConfig } from "./MatrixRTCSession.ts";
 import { secureRandomBase64Url } from "../randomstring.ts";
-import { type EncryptionKeysEventContent } from "./types.ts";
 import { decodeBase64, encodeUnpaddedBase64 } from "../base64.ts";
-import { type MatrixError, safeGetRetryAfterMs } from "../http-api/errors.ts";
+import { safeGetRetryAfterMs } from "../http-api/errors.ts";
 import { type CallMembership } from "./CallMembership.ts";
-import { EventType } from "../@types/event.ts";
+import { type IKeyTransport } from "./IKeyTransport.ts";
+
 const logger = rootLogger.getChild("MatrixRTCSession");
 
 /**
@@ -40,8 +38,11 @@ export type Statistics = {
  */
 export interface IEncryptionManager {
     join(joinConfig: EncryptionConfig | undefined): void;
+
     leave(): void;
+
     onMembershipsUpdate(oldMemberships: CallMembership[]): void;
+
     /**
      * Process `m.call.encryption_keys` events to track the encryption keys for call participants.
      * This should be called each time the relevant event is received from a room timeline.
@@ -50,7 +51,9 @@ export interface IEncryptionManager {
      * @param event the event to process
      */
     onCallEncryptionEventReceived(event: MatrixEvent): void;
+
     getEncryptionKeys(): Map<string, Array<{ key: Uint8Array; timestamp: number }>>;
+
     statistics: Statistics;
 }
 
@@ -71,9 +74,11 @@ export class EncryptionManager implements IEncryptionManager {
     private get updateEncryptionKeyThrottle(): number {
         return this.joinConfig?.updateEncryptionKeyThrottle ?? 3_000;
     }
+
     private get makeKeyDelay(): number {
         return this.joinConfig?.makeKeyDelay ?? 3_000;
     }
+
     private get useKeyDelay(): number {
         return this.joinConfig?.useKeyDelay ?? 5_000;
     }
@@ -99,9 +104,10 @@ export class EncryptionManager implements IEncryptionManager {
     private joinConfig: EncryptionConfig | undefined;
 
     public constructor(
-        private client: Pick<MatrixClient, "sendEvent" | "getDeviceId" | "getUserId" | "cancelPendingEvent">,
+        private userId: string,
-        private room: Pick<Room, "roomId">,
+        private deviceId: string,
         private getMemberships: () => CallMembership[],
+        private transport: IKeyTransport,
         private onEncryptionKeysChanged: (
             keyBin: Uint8Array<ArrayBufferLike>,
             encryptionKeyIndex: number,
@@ -112,7 +118,9 @@ export class EncryptionManager implements IEncryptionManager {
     public getEncryptionKeys(): Map<string, Array<{ key: Uint8Array; timestamp: number }>> {
         return this.encryptionKeys;
     }
+
     private joined = false;
+
     public join(joinConfig: EncryptionConfig): void {
         this.joinConfig = joinConfig;
         this.joined = true;
@@ -124,15 +132,10 @@ export class EncryptionManager implements IEncryptionManager {
     }
 
     public leave(): void {
-        const userId = this.client.getUserId();
-        const deviceId = this.client.getDeviceId();
-
-        if (!userId) throw new Error("No userId");
-        if (!deviceId) throw new Error("No deviceId");
         // clear our encryption keys as we're done with them now (we'll
         // make new keys if we rejoin). We leave keys for other participants
         // as they may still be using the same ones.
-        this.encryptionKeys.set(getParticipantId(userId, deviceId), []);
+        this.encryptionKeys.set(getParticipantId(this.userId, this.deviceId), []);
 
         if (this.makeNewKeyTimeout !== undefined) {
             clearTimeout(this.makeNewKeyTimeout);
@@ -146,9 +149,9 @@ export class EncryptionManager implements IEncryptionManager {
         this.manageMediaKeys = false;
         this.joined = false;
     }
+
     // TODO deduplicate this method. It also is in MatrixRTCSession.
-    private isMyMembership = (m: CallMembership): boolean =>
+    private isMyMembership = (m: CallMembership): boolean => m.sender === this.userId && m.deviceId === this.deviceId;
-        m.sender === this.client.getUserId() && m.deviceId === this.client.getDeviceId();
 
     public onMembershipsUpdate(oldMemberships: CallMembership[]): void {
         if (this.manageMediaKeys && this.joined) {
@@ -204,16 +207,17 @@ export class EncryptionManager implements IEncryptionManager {
      * @returns The index of the new key
      */
     private makeNewSenderKey(delayBeforeUse = false): number {
-        const userId = this.client.getUserId();
-        const deviceId = this.client.getDeviceId();
-
-        if (!userId) throw new Error("No userId");
-        if (!deviceId) throw new Error("No deviceId");
-
         const encryptionKey = secureRandomBase64Url(16);
         const encryptionKeyIndex = this.getNewEncryptionKeyIndex();
         logger.info("Generated new key at index " + encryptionKeyIndex);
-        this.setEncryptionKey(userId, deviceId, encryptionKeyIndex, encryptionKey, Date.now(), delayBeforeUse);
+        this.setEncryptionKey(
+            this.userId,
+            this.deviceId,
+            encryptionKeyIndex,
+            encryptionKey,
+            Date.now(),
+            delayBeforeUse,
+        );
         return encryptionKeyIndex;
     }
 
@@ -266,13 +270,7 @@ export class EncryptionManager implements IEncryptionManager {
 
         logger.info(`Sending encryption keys event. indexToSend=${indexToSend}`);
 
-        const userId = this.client.getUserId();
+        const myKeys = this.getKeysForParticipant(this.userId, this.deviceId);
-        const deviceId = this.client.getDeviceId();
-
-        if (!userId) throw new Error("No userId");
-        if (!deviceId) throw new Error("No deviceId");
-
-        const myKeys = this.getKeysForParticipant(userId, deviceId);
 
         if (!myKeys) {
             logger.warn("Tried to send encryption keys event but no keys found!");
@@ -288,35 +286,15 @@ export class EncryptionManager implements IEncryptionManager {
         const keyToSend = myKeys[keyIndexToSend];
 
         try {
-            const content: EncryptionKeysEventContent = {
-                keys: [
-                    {
-                        index: keyIndexToSend,
-                        key: encodeUnpaddedBase64(keyToSend),
-                    },
-                ],
-                device_id: deviceId,
-                call_id: "",
-                sent_ts: Date.now(),
-            };
-
             this.statistics.counters.roomEventEncryptionKeysSent += 1;
-
+            await this.transport.sendKey(encodeUnpaddedBase64(keyToSend), keyIndexToSend);
-            await this.client.sendEvent(this.room.roomId, EventType.CallEncryptionKeysPrefix, content);
-
             logger.debug(
-                `Embedded-E2EE-LOG updateEncryptionKeyEvent participantId=${userId}:${deviceId} numKeys=${myKeys.length} currentKeyIndex=${this.currentEncryptionKeyIndex} keyIndexToSend=${keyIndexToSend}`,
+                `Embedded-E2EE-LOG updateEncryptionKeyEvent participantId=${this.userId}:${this.deviceId} numKeys=${myKeys.length} currentKeyIndex=${this.currentEncryptionKeyIndex} keyIndexToSend=${keyIndexToSend}`,
                 this.encryptionKeys,
             );
         } catch (error) {
-            const matrixError = error as MatrixError;
-            if (matrixError.event) {
-                // cancel the pending event: we'll just generate a new one with our latest
-                // keys when we resend
-                this.client.cancelPendingEvent(matrixError.event);
-            }
             if (this.keysEventUpdateTimeout === undefined) {
-                const resendDelay = safeGetRetryAfterMs(matrixError, 5000);
+                const resendDelay = safeGetRetryAfterMs(error, 5000);
                 logger.warn(`Failed to send m.call.encryption_key, retrying in ${resendDelay}`, error);
                 this.keysEventUpdateTimeout = setTimeout(() => void this.sendEncryptionKeysEvent(), resendDelay);
             } else {
@@ -326,74 +304,15 @@ export class EncryptionManager implements IEncryptionManager {
     };
 
     public onCallEncryptionEventReceived = (event: MatrixEvent): void => {
-        const userId = event.getSender();
+        this.transport.receiveRoomEvent(
-        const content = event.getContent<EncryptionKeysEventContent>();
+            event,
-
+            this.statistics,
-        const deviceId = content["device_id"];
+            (userId, deviceId, encryptionKeyIndex, encryptionKeyString, timestamp) => {
-        const callId = content["call_id"];
+                this.setEncryptionKey(userId, deviceId, encryptionKeyIndex, encryptionKeyString, timestamp);
-
+            },
-        if (!userId) {
+        );
-            logger.warn(`Received m.call.encryption_keys with no userId: callId=${callId}`);
-            return;
-        }
-
-        // We currently only handle callId = "" (which is the default for room scoped calls)
-        if (callId !== "") {
-            logger.warn(
-                `Received m.call.encryption_keys with unsupported callId: userId=${userId}, deviceId=${deviceId}, callId=${callId}`,
-            );
-            return;
-        }
-
-        if (!Array.isArray(content.keys)) {
-            logger.warn(`Received m.call.encryption_keys where keys wasn't an array: callId=${callId}`);
-            return;
-        }
-
-        if (userId === this.client.getUserId() && deviceId === this.client.getDeviceId()) {
-            // We store our own sender key in the same set along with keys from others, so it's
-            // important we don't allow our own keys to be set by one of these events (apart from
-            // the fact that we don't need it anyway because we already know our own keys).
-            logger.info("Ignoring our own keys event");
-            return;
-        }
-
-        this.statistics.counters.roomEventEncryptionKeysReceived += 1;
-        const age = Date.now() - (typeof content.sent_ts === "number" ? content.sent_ts : event.getTs());
-        this.statistics.totals.roomEventEncryptionKeysReceivedTotalAge += age;
-
-        for (const key of content.keys) {
-            if (!key) {
-                logger.info("Ignoring false-y key in keys event");
-                continue;
-            }
-
-            const encryptionKey = key.key;
-            const encryptionKeyIndex = key.index;
-
-            if (
-                !encryptionKey ||
-                encryptionKeyIndex === undefined ||
-                encryptionKeyIndex === null ||
-                callId === undefined ||
-                callId === null ||
-                typeof deviceId !== "string" ||
-                typeof callId !== "string" ||
-                typeof encryptionKey !== "string" ||
-                typeof encryptionKeyIndex !== "number"
-            ) {
-                logger.warn(
-                    `Malformed call encryption_key: userId=${userId}, deviceId=${deviceId}, encryptionKeyIndex=${encryptionKeyIndex} callId=${callId}`,
-                );
-            } else {
-                logger.debug(
-                    `Embedded-E2EE-LOG onCallEncryption userId=${userId}:${deviceId} encryptionKeyIndex=${encryptionKeyIndex} age=${age}ms`,
-                    this.encryptionKeys,
-                );
-                this.setEncryptionKey(userId, deviceId, encryptionKeyIndex, encryptionKey, event.getTs());
-            }
-        }
     };
+
     private storeLastMembershipFingerprints(): void {
         this.lastMembershipFingerprints = new Set(
             this.getMemberships()
@@ -466,14 +385,14 @@ export class EncryptionManager implements IEncryptionManager {
             const useKeyTimeout = setTimeout(() => {
                 this.setNewKeyTimeouts.delete(useKeyTimeout);
                 logger.info(`Delayed-emitting key changed event for ${participantId} idx ${encryptionKeyIndex}`);
-                if (userId === this.client.getUserId() && deviceId === this.client.getDeviceId()) {
+                if (userId === this.userId && deviceId === this.deviceId) {
                     this.currentEncryptionKeyIndex = encryptionKeyIndex;
                 }
                 this.onEncryptionKeysChanged(keyBin, encryptionKeyIndex, participantId);
             }, this.useKeyDelay);
             this.setNewKeyTimeouts.add(useKeyTimeout);
         } else {
-            if (userId === this.client.getUserId() && deviceId === this.client.getDeviceId()) {
+            if (userId === this.userId && deviceId === this.deviceId) {
                 this.currentEncryptionKeyIndex = encryptionKeyIndex;
             }
             this.onEncryptionKeysChanged(keyBin, encryptionKeyIndex, participantId);
@@ -493,8 +412,10 @@ export class EncryptionManager implements IEncryptionManager {
 }
 
 const getParticipantId = (userId: string, deviceId: string): string => `${userId}:${deviceId}`;
+
 function keysEqual(a: Uint8Array | undefined, b: Uint8Array | undefined): boolean {
     if (a === b) return true;
     return !!a && !!b && a.length === b.length && a.every((x, i) => x === b[i]);
 }
+
 const getParticipantIdFromMembership = (m: CallMembership): string => getParticipantId(m.sender!, m.deviceId);

src/matrixrtc/IKeyTransport.ts

@@ -0,0 +1,49 @@
+/*
+Copyright 2025 The Matrix.org Foundation C.I.C.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import { type MatrixEvent } from "../models/event.ts";
+import { type Statistics } from "./EncryptionManager.ts";
+
+/**
+ * Generic interface for the transport used to share room keys.
+ * Keys can be shared using different transports, e.g. to-device messages or room messages.
+ */
+export interface IKeyTransport {
+    /**
+     * Sends the current user media key.
+     * @param keyBase64Encoded
+     * @param index
+     */
+    sendKey(keyBase64Encoded: string, index: number): Promise<void>;
+
+    /**
+     * Takes an incoming event from the transport and extracts the key information.
+     * @param event
+     * @param statistics
+     * @param callback
+     */
+    receiveRoomEvent(
+        event: MatrixEvent,
+        statistics: Statistics,
+        callback: (
+            userId: string,
+            deviceId: string,
+            encryptionKeyIndex: number,
+            encryptionKeyString: string,
+            timestamp: number,
+        ) => void,
+    ): void;
+}

src/matrixrtc/MatrixRTCSession.ts

@@ -30,6 +30,7 @@ import { EncryptionManager, type IEncryptionManager, type Statistics } from "./E
 import { LegacyMembershipManager } from "./LegacyMembershipManager.ts";
 import { logDurationSync } from "../utils.ts";
 import type { IMembershipManager } from "./types.ts";
+import { RoomKeyTransport } from "./RoomKeyTransport.ts";
 
 const logger = rootLogger.getChild("MatrixRTCSession");
 
@@ -306,10 +307,13 @@ export class MatrixRTCSession extends TypedEventEmitter<MatrixRTCSessionEvent, M
         // TODO: double check if this is actually needed. Should be covered by refreshRoom in MatrixRTCSessionManager
         roomState?.on(RoomStateEvent.Members, this.onRoomMemberUpdate);
         this.setExpiryTimer();
+
+        const transport = new RoomKeyTransport(this.roomSubset.roomId, this.client);
         this.encryptionManager = new EncryptionManager(
-            this.client,
+            this.client.getUserId()!,
-            this.roomSubset,
+            this.client.getDeviceId()!,
             () => this.memberships,
+            transport,
             (keyBin: Uint8Array<ArrayBufferLike>, encryptionKeyIndex: number, participantId: string) => {
                 this.emit(MatrixRTCSessionEvent.EncryptionKeyChanged, keyBin, encryptionKeyIndex, participantId);
             },

src/matrixrtc/RoomKeyTransport.ts

@@ -0,0 +1,141 @@
+/*
+Copyright 2025 The Matrix.org Foundation C.I.C.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import type { MatrixClient } from "../client.ts";
+import type { EncryptionKeysEventContent } from "./types.ts";
+import { EventType } from "../@types/event.ts";
+import { type MatrixError } from "../http-api/errors.ts";
+import { logger, type Logger } from "../logger.ts";
+import { type IKeyTransport } from "./IKeyTransport.ts";
+import { type MatrixEvent } from "../models/event.ts";
+import { type Statistics } from "./EncryptionManager.ts";
+
+export class RoomKeyTransport implements IKeyTransport {
+    private readonly prefixedLogger: Logger;
+
+    public constructor(
+        private roomId: string,
+        private client: Pick<MatrixClient, "sendEvent" | "getDeviceId" | "getUserId" | "cancelPendingEvent">,
+    ) {
+        this.prefixedLogger = logger.getChild(`[RTC: ${roomId} RoomKeyTransport]`);
+    }
+
+    public async sendKey(keyBase64Encoded: string, index: number): Promise<void> {
+        const content: EncryptionKeysEventContent = {
+            keys: [
+                {
+                    index: index,
+                    key: keyBase64Encoded,
+                },
+            ],
+            device_id: this.client.getDeviceId()!,
+            call_id: "",
+            sent_ts: Date.now(),
+        };
+
+        try {
+            await this.client.sendEvent(this.roomId, EventType.CallEncryptionKeysPrefix, content);
+        } catch (error) {
+            this.prefixedLogger.error("Failed to send call encryption keys", error);
+            const matrixError = error as MatrixError;
+            if (matrixError.event) {
+                // cancel the pending event: we'll just generate a new one with our latest
+                // keys when we resend
+                this.client.cancelPendingEvent(matrixError.event);
+            }
+            throw error;
+        }
+    }
+
+    public receiveRoomEvent(
+        event: MatrixEvent,
+        statistics: Statistics,
+        callback: (
+            userId: string,
+            deviceId: string,
+            encryptionKeyIndex: number,
+            encryptionKeyString: string,
+            timestamp: number,
+        ) => void,
+    ): void {
+        const userId = event.getSender();
+        const content = event.getContent<EncryptionKeysEventContent>();
+
+        const deviceId = content["device_id"];
+        const callId = content["call_id"];
+
+        if (!userId) {
+            logger.warn(`Received m.call.encryption_keys with no userId: callId=${callId}`);
+            return;
+        }
+
+        // We currently only handle callId = "" (which is the default for room scoped calls)
+        if (callId !== "") {
+            logger.warn(
+                `Received m.call.encryption_keys with unsupported callId: userId=${userId}, deviceId=${deviceId}, callId=${callId}`,
+            );
+            return;
+        }
+
+        if (!Array.isArray(content.keys)) {
+            logger.warn(`Received m.call.encryption_keys where keys wasn't an array: callId=${callId}`);
+            return;
+        }
+
+        if (userId === this.client.getUserId() && deviceId === this.client.getDeviceId()) {
+            // We store our own sender key in the same set along with keys from others, so it's
+            // important we don't allow our own keys to be set by one of these events (apart from
+            // the fact that we don't need it anyway because we already know our own keys).
+            logger.info("Ignoring our own keys event");
+            return;
+        }
+
+        statistics.counters.roomEventEncryptionKeysReceived += 1;
+        const age = Date.now() - (typeof content.sent_ts === "number" ? content.sent_ts : event.getTs());
+        statistics.totals.roomEventEncryptionKeysReceivedTotalAge += age;
+
+        for (const key of content.keys) {
+            if (!key) {
+                logger.info("Ignoring false-y key in keys event");
+                continue;
+            }
+
+            const encryptionKey = key.key;
+            const encryptionKeyIndex = key.index;
+
+            if (
+                !encryptionKey ||
+                encryptionKeyIndex === undefined ||
+                encryptionKeyIndex === null ||
+                callId === undefined ||
+                callId === null ||
+                typeof deviceId !== "string" ||
+                typeof callId !== "string" ||
+                typeof encryptionKey !== "string" ||
+                typeof encryptionKeyIndex !== "number"
+            ) {
+                logger.warn(
+                    `Malformed call encryption_key: userId=${userId}, deviceId=${deviceId}, encryptionKeyIndex=${encryptionKeyIndex} callId=${callId}`,
+                );
+            } else {
+                logger.debug(
+                    `Embedded-E2EE-LOG onCallEncryption userId=${userId}:${deviceId} encryptionKeyIndex=${encryptionKeyIndex} age=${age}ms`,
+                );
+                callback(userId, deviceId, encryptionKeyIndex, encryptionKey, event.getTs());
+            }
+        }
+    }
+}