feat(E2EE): Async E2EE keys exchange (#32197)

.changeset/orange-clocks-raise.md

@@ -0,0 +1,8 @@
+---
+'@rocket.chat/model-typings': minor
+'@rocket.chat/core-typings': minor
+'@rocket.chat/rest-typings': minor
+'@rocket.chat/meteor': minor
+---
+
+Async End-to-End Encrypted rooms key distribution process. Users now don't need to be online to get the keys of their subscribed encrypted rooms, the key distribution process is now async and users can recieve keys even when they are not online.

apps/meteor/app/api/server/v1/e2e.ts

@@ -1,13 +1,18 @@
 import type { IUser } from '@rocket.chat/core-typings';
+import { Subscriptions } from '@rocket.chat/models';
 import {
 	ise2eGetUsersOfRoomWithoutKeyParamsGET,
 	ise2eSetRoomKeyIDParamsPOST,
 	ise2eSetUserPublicAndPrivateKeysParamsPOST,
 	ise2eUpdateGroupKeyParamsPOST,
+	isE2EProvideUsersGroupKeyProps,
+	isE2EFetchUsersWaitingForGroupKeyProps,
 } from '@rocket.chat/rest-typings';
 import { Meteor } from 'meteor/meteor';
 
 import { handleSuggestedGroupKey } from '../../../e2e/server/functions/handleSuggestedGroupKey';
+import { provideUsersSuggestedGroupKeys } from '../../../e2e/server/functions/provideUsersSuggestedGroupKeys';
+import { settings } from '../../../settings/server';
 import { API } from '../api';
 
 API.v1.addRoute(
@@ -188,6 +193,9 @@ API.v1.addRoute(
 	{
 		authRequired: true,
 		validateParams: ise2eUpdateGroupKeyParamsPOST,
+		deprecation: {
+			version: '8.0.0',
+		},
 	},
 	{
 		async post() {
@@ -233,3 +241,46 @@ API.v1.addRoute(
 		},
 	},
 );
+
+API.v1.addRoute(
+	'e2e.fetchUsersWaitingForGroupKey',
+	{
+		authRequired: true,
+		validateParams: isE2EFetchUsersWaitingForGroupKeyProps,
+	},
+	{
+		async get() {
+			if (!settings.get('E2E_Enable')) {
+				return API.v1.success({ usersWaitingForE2EKeys: {} });
+			}
+
+			const { roomIds = [] } = this.queryParams;
+			const usersWaitingForE2EKeys = (await Subscriptions.findUsersWithPublicE2EKeyByRids(roomIds, this.userId).toArray()).reduce<
+				Record<string, { _id: string; public_key: string }[]>
+			>((acc, { rid, users }) => ({ [rid]: users, ...acc }), {});
+
+			return API.v1.success({
+				usersWaitingForE2EKeys,
+			});
+		},
+	},
+);
+
+API.v1.addRoute(
+	'e2e.provideUsersSuggestedGroupKeys',
+	{
+		authRequired: true,
+		validateParams: isE2EProvideUsersGroupKeyProps,
+	},
+	{
+		async post() {
+			if (!settings.get('E2E_Enable')) {
+				return API.v1.success();
+			}
+
+			await provideUsersSuggestedGroupKeys(this.userId, this.bodyParams.usersSuggestedGroupKeys);
+
+			return API.v1.success();
+		},
+	},
+);

apps/meteor/app/e2e/client/rocketchat.e2e.room.js

@@ -94,6 +94,10 @@ export class E2ERoom extends Emitter {
 		logError(`E2E ROOM { state: ${this.state}, rid: ${this.roomId} }`, ...msg);
 	}
 
+	hasSessionKey() {
+		return !!this.groupSessionKey;
+	}
+
 	getState() {
 		return this.state;
 	}
@@ -317,17 +321,29 @@ export class E2ERoom extends Emitter {
 	async encryptKeyForOtherParticipants() {
 		// Encrypt generated session key for every user in room and publish to subscription model.
 		try {
-			const { users } = await sdk.call('e2e.getUsersOfRoomWithoutKey', this.roomId);
-			users.forEach((user) => this.encryptForParticipant(user));
+			const users = (await sdk.call('e2e.getUsersOfRoomWithoutKey', this.roomId)).users.filter((user) => user?.e2e?.public_key);
+
+			if (!users.length) {
+				return;
+			}
+
+			const usersSuggestedGroupKeys = { [this.roomId]: [] };
+			for await (const user of users) {
+				const encryptedGroupKey = await this.encryptGroupKeyForParticipant(user.e2e.public_key);
+
+				usersSuggestedGroupKeys[this.roomId].push({ _id: user._id, key: encryptedGroupKey });
+			}
+
+			await sdk.rest.post('/v1/e2e.provideUsersSuggestedGroupKeys', { usersSuggestedGroupKeys });
 		} catch (error) {
 			return this.error('Error getting room users: ', error);
 		}
 	}
 
-	async encryptForParticipant(user) {
+	async encryptGroupKeyForParticipant(public_key) {
 		let userKey;
 		try {
-			userKey = await importRSAKey(JSON.parse(user.e2e.public_key), ['encrypt']);
+			userKey = await importRSAKey(JSON.parse(public_key), ['encrypt']);
 		} catch (error) {
 			return this.error('Error importing user key: ', error);
 		}
@@ -336,8 +352,8 @@ export class E2ERoom extends Emitter {
 		// Encrypt session key for this user with his/her public key
 		try {
 			const encryptedUserKey = await encryptRSA(userKey, toArrayBuffer(this.sessionKeyExportedString));
-			// Key has been encrypted. Publish to that user's subscription model for this room.
-			await sdk.call('e2e.updateGroupKey', this.roomId, user._id, this.keyID + Base64.encode(new Uint8Array(encryptedUserKey)));
+			const encryptedUserKeyToString = this.keyID + Base64.encode(new Uint8Array(encryptedUserKey));
+			return encryptedUserKeyToString;
 		} catch (error) {
 			return this.error('Error encrypting user key: ', error);
 		}
@@ -510,4 +526,20 @@ export class E2ERoom extends Emitter {
 		this.on('STATE_CHANGED', cb);
 		return () => this.off('STATE_CHANGED', cb);
 	}
+
+	async encryptGroupKeyForParticipantsWaitingForTheKeys(users) {
+		if (!this.isReady()) {
+			return;
+		}
+
+		const usersWithKeys = await Promise.all(
+			users.map(async (user) => {
+				const { _id, public_key } = user;
+				const key = await this.encryptGroupKeyForParticipant(public_key);
+				return { _id, key };
+			}),
+		);
+
+		return usersWithKeys;
+	}
 }

apps/meteor/app/e2e/client/rocketchat.e2e.ts

@@ -1,10 +1,11 @@
 import QueryString from 'querystring';
 import URL from 'url';
 
-import type { IE2EEMessage, IMessage, IRoom, ISubscription, IUploadWithUser, MessageAttachment } from '@rocket.chat/core-typings';
+import type { IE2EEMessage, IMessage, IRoom, ISubscription, IUser, IUploadWithUser, MessageAttachment } from '@rocket.chat/core-typings';
 import { isE2EEMessage } from '@rocket.chat/core-typings';
 import { Emitter } from '@rocket.chat/emitter';
 import EJSON from 'ejson';
+import _ from 'lodash';
 import { Meteor } from 'meteor/meteor';
 import { Tracker } from 'meteor/tracker';
 
@@ -18,6 +19,7 @@ import EnterE2EPasswordModal from '../../../client/views/e2e/EnterE2EPasswordMod
 import SaveE2EPasswordModal from '../../../client/views/e2e/SaveE2EPasswordModal';
 import { createQuoteAttachment } from '../../../lib/createQuoteAttachment';
 import { getMessageUrlRegex } from '../../../lib/getMessageUrlRegex';
+import { isTruthy } from '../../../lib/isTruthy';
 import { ChatRoom, Subscriptions, Messages } from '../../models/client';
 import { settings } from '../../settings/client';
 import { getUserAvatarURL } from '../../utils/client';
@@ -40,6 +42,7 @@ import {
 } from './helper';
 import { log, logError } from './logger';
 import { E2ERoom } from './rocketchat.e2e.room';
+
 import './events.js';
 
 let failedToDecodeKey = false;
@@ -49,6 +52,7 @@ type KeyPair = {
 	private_key: string | null;
 };
 
+const ROOM_KEY_EXCHANGE_SIZE = 10;
 const E2EEStateDependency = new Tracker.Dependency();
 
 class E2E extends Emitter {
@@ -62,12 +66,18 @@ class E2E extends Emitter {
 
 	public privateKey: CryptoKey | undefined;
 
+	private keyDistributionInterval: ReturnType<typeof setInterval> | null;
+
 	private state: E2EEState;
 
+	private observable: Meteor.LiveQueryHandle | undefined;
+
 	constructor() {
 		super();
 		this.started = false;
 		this.instancesByRoomId = {};
+		this.keyDistributionInterval = null;
+		this.observable = undefined;
 
 		this.on('E2E_STATE_CHANGED', ({ prevState, nextState }) => {
 			this.log(`${prevState} -> ${nextState}`);
@@ -81,6 +91,18 @@ class E2E extends Emitter {
 			await this.onE2EEReady();
 		});
 
+		this.on(E2EEState.DISABLED, () => {
+			this.observable?.stop();
+		});
+
+		this.on(E2EEState.NOT_STARTED, () => {
+			this.observable?.stop();
+		});
+
+		this.on(E2EEState.ERROR, () => {
+			this.observable?.stop();
+		});
+
 		this.setState(E2EEState.NOT_STARTED);
 	}
 
@@ -109,12 +131,80 @@ class E2E extends Emitter {
 
 	async onE2EEReady() {
 		this.log('startClient -> Done');
-		this.log('decryptSubscriptions');
 		this.initiateHandshake();
+		await this.handleAsyncE2EESuggestedKey();
+		this.log('decryptSubscriptions');
 		await this.decryptSubscriptions();
 		this.log('decryptSubscriptions -> Done');
 		await this.initiateDecryptingPendingMessages();
 		this.log('DecryptingPendingMessages -> Done');
+		await this.initiateKeyDistribution();
+		this.log('initiateKeyDistribution -> Done');
+		this.observeSubscriptions();
+		this.log('observing subscriptions');
+	}
+
+	observeSubscriptions() {
+		this.observable?.stop();
+
+		this.observable = Subscriptions.find().observe({
+			changed: (sub: ISubscription) => {
+				setTimeout(async () => {
+					this.log('Subscription changed', sub);
+					if (!sub.encrypted && !sub.E2EKey) {
+						this.removeInstanceByRoomId(sub.rid);
+						return;
+					}
+
+					const e2eRoom = await this.getInstanceByRoomId(sub.rid);
+					if (!e2eRoom) {
+						return;
+					}
+
+					if (sub.E2ESuggestedKey) {
+						if (await e2eRoom.importGroupKey(sub.E2ESuggestedKey)) {
+							await this.acceptSuggestedKey(sub.rid);
+							e2eRoom.keyReceived();
+						} else {
+							console.warn('Invalid E2ESuggestedKey, rejecting', sub.E2ESuggestedKey);
+							await this.rejectSuggestedKey(sub.rid);
+						}
+					}
+
+					sub.encrypted ? e2eRoom.resume() : e2eRoom.pause();
+
+					// Cover private groups and direct messages
+					if (!e2eRoom.isSupportedRoomType(sub.t)) {
+						e2eRoom.disable();
+						return;
+					}
+
+					if (sub.E2EKey && e2eRoom.isWaitingKeys()) {
+						e2eRoom.keyReceived();
+						return;
+					}
+
+					if (!e2eRoom.isReady()) {
+						return;
+					}
+
+					await e2eRoom.decryptSubscription();
+				}, 0);
+			},
+			added: (sub: ISubscription) => {
+				setTimeout(async () => {
+					this.log('Subscription added', sub);
+					if (!sub.encrypted && !sub.E2EKey) {
+						return;
+					}
+					return this.getInstanceByRoomId(sub.rid);
+				}, 0);
+			},
+			removed: (sub: ISubscription) => {
+				this.log('Subscription removed', sub);
+				this.removeInstanceByRoomId(sub.rid);
+			},
+		});
 	}
 
 	shouldAskForE2EEPassword() {
@@ -134,6 +224,32 @@ class E2E extends Emitter {
 		this.emit(nextState);
 	}
 
+	async handleAsyncE2EESuggestedKey() {
+		const subs = Subscriptions.find({ E2ESuggestedKey: { $exists: true } }).fetch();
+		await Promise.all(
+			subs
+				.filter((sub) => sub.E2ESuggestedKey && !sub.E2EKey)
+				.map(async (sub) => {
+					const e2eRoom = await e2e.getInstanceByRoomId(sub.rid);
+
+					if (!e2eRoom) {
+						return;
+					}
+
+					if (await e2eRoom.importGroupKey(sub.E2ESuggestedKey)) {
+						this.log('Imported valid E2E suggested key');
+						await e2e.acceptSuggestedKey(sub.rid);
+						e2eRoom.keyReceived();
+					} else {
+						this.error('Invalid E2ESuggestedKey, rejecting', sub.E2ESuggestedKey);
+						await e2e.rejectSuggestedKey(sub.rid);
+					}
+
+					sub.encrypted ? e2eRoom.resume() : e2eRoom.pause();
+				}),
+		);
+	}
+
 	async getInstanceByRoomId(rid: IRoom['_id']): Promise<E2ERoom | null> {
 		const room = await waitUntilFind(() => ChatRoom.findOne({ _id: rid }));
 
@@ -301,6 +417,8 @@ class E2E extends Emitter {
 		this.instancesByRoomId = {};
 		this.privateKey = undefined;
 		this.started = false;
+		this.keyDistributionInterval && clearInterval(this.keyDistributionInterval);
+		this.keyDistributionInterval = null;
 		this.setState(E2EEState.DISABLED);
 	}
 
@@ -602,6 +720,9 @@ class E2E extends Emitter {
 	}
 
 	async parseQuoteAttachment(message: IE2EEMessage): Promise<IE2EEMessage> {
+		if (!message?.msg) {
+			return message;
+		}
 		const urls = message.msg.match(getMessageUrlRegex()) || [];
 
 		await Promise.all(
@@ -647,6 +768,99 @@ class E2E extends Emitter {
 
 		return message;
 	}
+
+	async getSuggestedE2EEKeys(usersWaitingForE2EKeys: Record<IRoom['_id'], { _id: IUser['_id']; public_key: string }[]>) {
+		const roomIds = Object.keys(usersWaitingForE2EKeys);
+		return Object.fromEntries(
+			(
+				await Promise.all(
+					roomIds.map(async (room) => {
+						const e2eRoom = await this.getInstanceByRoomId(room);
+
+						if (!e2eRoom) {
+							return;
+						}
+						const usersWithKeys = await e2eRoom.encryptGroupKeyForParticipantsWaitingForTheKeys(usersWaitingForE2EKeys[room]);
+
+						if (!usersWithKeys) {
+							return;
+						}
+
+						return [room, usersWithKeys];
+					}),
+				)
+			).filter(isTruthy),
+		);
+	}
+
+	async getSample(roomIds: string[], limit = 3): Promise<string[]> {
+		if (limit === 0) {
+			return [];
+		}
+
+		const randomRoomIds = _.sampleSize(roomIds, ROOM_KEY_EXCHANGE_SIZE);
+
+		const sampleIds: string[] = [];
+		for await (const roomId of randomRoomIds) {
+			const e2eroom = await this.getInstanceByRoomId(roomId);
+			if (!e2eroom?.hasSessionKey()) {
+				continue;
+			}
+
+			sampleIds.push(roomId);
+		}
+
+		if (!sampleIds.length) {
+			return this.getSample(roomIds, limit - 1);
+		}
+
+		return sampleIds;
+	}
+
+	async initiateKeyDistribution() {
+		if (this.keyDistributionInterval) {
+			return;
+		}
+
+		const keyDistribution = async () => {
+			const roomIds = ChatRoom.find({
+				'usersWaitingForE2EKeys': { $exists: true },
+				'usersWaitingForE2EKeys.userId': { $ne: Meteor.userId() },
+			}).map((room) => room._id);
+			if (!roomIds.length) {
+				return;
+			}
+
+			// Prevent function from running and doing nothing when theres something to do
+			const sampleIds = await this.getSample(roomIds);
+
+			if (!sampleIds.length) {
+				return;
+			}
+
+			const { usersWaitingForE2EKeys = {} } = await sdk.rest.get('/v1/e2e.fetchUsersWaitingForGroupKey', { roomIds: sampleIds });
+
+			if (!Object.keys(usersWaitingForE2EKeys).length) {
+				return;
+			}
+
+			const userKeysWithRooms = await this.getSuggestedE2EEKeys(usersWaitingForE2EKeys);
+
+			if (!Object.keys(userKeysWithRooms).length) {
+				return;
+			}
+
+			try {
+				await sdk.rest.post('/v1/e2e.provideUsersSuggestedGroupKeys', { usersSuggestedGroupKeys: userKeysWithRooms });
+			} catch (error) {
+				return this.error('Error providing group key to users: ', error);
+			}
+		};
+
+		// Run first call right away, then schedule for 10s in the future
+		await keyDistribution();
+		this.keyDistributionInterval = setInterval(keyDistribution, 10000);
+	}
 }
 
 export const e2e = new E2E();

apps/meteor/app/e2e/server/functions/handleSuggestedGroupKey.ts

@@ -1,4 +1,4 @@
-import { Subscriptions } from '@rocket.chat/models';
+import { Rooms, Subscriptions } from '@rocket.chat/models';
 import { Meteor } from 'meteor/meteor';
 
 export async function handleSuggestedGroupKey(
@@ -23,6 +23,11 @@ export async function handleSuggestedGroupKey(
 
 	if (handle === 'accept') {
 		await Subscriptions.setGroupE2EKey(sub._id, suggestedKey);
+		await Rooms.removeUsersFromE2EEQueueByRoomId(sub.rid, [userId]);
+	}
+
+	if (handle === 'reject') {
+		await Rooms.addUserIdToE2EEQueueByRoomIds([sub.rid], userId);
 	}
 
 	await Subscriptions.unsetGroupE2ESuggestedKey(sub._id);

apps/meteor/app/e2e/server/functions/provideUsersSuggestedGroupKeys.ts

@@ -0,0 +1,33 @@
+import type { IRoom, IUser } from '@rocket.chat/core-typings';
+import { Rooms, Subscriptions } from '@rocket.chat/models';
+
+import { canAccessRoomIdAsync } from '../../../authorization/server/functions/canAccessRoom';
+
+export const provideUsersSuggestedGroupKeys = async (
+	userId: IUser['_id'],
+	usersSuggestedGroupKeys: Record<IRoom['_id'], { _id: IUser['_id']; key: string }[]>,
+) => {
+	const roomIds = Object.keys(usersSuggestedGroupKeys);
+
+	if (!roomIds.length) {
+		return;
+	}
+
+	// Process should try to process all rooms i have access instead of dying if one is not
+	for await (const roomId of roomIds) {
+		if (!(await canAccessRoomIdAsync(roomId, userId))) {
+			continue;
+		}
+
+		const usersWithSuggestedKeys = [];
+		for await (const user of usersSuggestedGroupKeys[roomId]) {
+			const { modifiedCount } = await Subscriptions.setGroupE2ESuggestedKey(user._id, roomId, user.key);
+			if (!modifiedCount) {
+				continue;
+			}
+			usersWithSuggestedKeys.push(user._id);
+		}
+
+		await Rooms.removeUsersFromE2EEQueueByRoomId(roomId, usersWithSuggestedKeys);
+	}
+};

apps/meteor/app/e2e/server/methods/setUserPublicAndPrivateKeys.ts

@@ -1,4 +1,4 @@
-import { Users } from '@rocket.chat/models';
+import { Rooms, Users } from '@rocket.chat/models';
 import type { ServerMethods } from '@rocket.chat/ui-contexts';
 import { Meteor } from 'meteor/meteor';
 
@@ -33,5 +33,8 @@ Meteor.methods<ServerMethods>({
 			private_key: keyPair.private_key,
 			public_key: keyPair.public_key,
 		});
+
+		const subscribedRoomIds = await Rooms.getSubscribedRoomIdsWithoutE2EKeys(userId);
+		await Rooms.addUserIdToE2EEQueueByRoomIds(subscribedRoomIds, userId);
 	},
 });

apps/meteor/app/e2e/server/methods/updateGroupKey.ts

@@ -2,6 +2,8 @@ import { Subscriptions } from '@rocket.chat/models';
 import type { ServerMethods } from '@rocket.chat/ui-contexts';
 import { Meteor } from 'meteor/meteor';
 
+import { methodDeprecationLogger } from '../../../lib/server/lib/deprecationWarningLogger';
+
 declare module '@rocket.chat/ui-contexts' {
 	// eslint-disable-next-line @typescript-eslint/naming-convention
 	interface ServerMethods {
@@ -11,6 +13,7 @@ declare module '@rocket.chat/ui-contexts' {
 
 Meteor.methods<ServerMethods>({
 	async 'e2e.updateGroupKey'(rid, uid, key) {
+		methodDeprecationLogger.method('e2e.updateGroupKey', '8.0.0');
 		const userId = Meteor.userId();
 		if (!userId) {
 			throw new Meteor.Error('error-invalid-user', 'Invalid user', { method: 'e2e.acceptSuggestedGroupKey' });
@@ -27,10 +30,7 @@ Meteor.methods<ServerMethods>({
 			}
 
 			// uid also has subscription to this room
-			const userSub = await Subscriptions.findOneByRoomIdAndUserId(rid, uid);
-			if (userSub) {
-				await Subscriptions.setGroupE2ESuggestedKey(userSub._id, key);
-			}
+			await Subscriptions.setGroupE2ESuggestedKey(uid, rid, key);
 		}
 	},
 });

apps/meteor/app/lib/server/functions/addUserToRoom.ts

@@ -9,6 +9,7 @@ import { RoomMemberActions } from '../../../../definition/IRoomTypeConfig';
 import { callbacks } from '../../../../lib/callbacks';
 import { getSubscriptionAutotranslateDefaultConfig } from '../../../../server/lib/getSubscriptionAutotranslateDefaultConfig';
 import { roomCoordinator } from '../../../../server/lib/rooms/roomCoordinator';
+import { settings } from '../../../settings/server';
 import { getDefaultSubscriptionPref } from '../../../utils/lib/getDefaultSubscriptionPref';
 import { notifyOnRoomChangedById } from '../lib/notifyListener';
 
@@ -132,5 +133,9 @@ export const addUserToRoom = async function (
 		await Team.addMember(inviter || userToBeAdded, userToBeAdded._id, room.teamId);
 	}
 
+	if (room.encrypted && settings.get('E2E_Enable') && userToBeAdded.e2e?.public_key) {
+		await Rooms.addUserIdToE2EEQueueByRoomIds([room._id], userToBeAdded._id);
+	}
+
 	return true;
 };

apps/meteor/app/lib/server/functions/removeUserFromRoom.ts

@@ -7,6 +7,7 @@ import { Meteor } from 'meteor/meteor';
 
 import { afterLeaveRoomCallback } from '../../../../lib/callbacks/afterLeaveRoomCallback';
 import { beforeLeaveRoomCallback } from '../../../../lib/callbacks/beforeLeaveRoomCallback';
+import { settings } from '../../../settings/server';
 import { notifyOnRoomChangedById } from '../lib/notifyListener';
 
 export const removeUserFromRoom = async function (
@@ -65,6 +66,10 @@ export const removeUserFromRoom = async function (
 		await Team.removeMember(room.teamId, user._id);
 	}
 
+	if (room.encrypted && settings.get('E2E_Enable')) {
+		await Rooms.removeUsersFromE2EEQueueByRoomId(room._id, [user._id]);
+	}
+
 	// TODO: CACHE: maybe a queue?
 	await afterLeaveRoomCallback.run(user, room);
 

apps/meteor/client/startup/e2e.ts

@@ -1,13 +1,12 @@
-import type { IMessage, ISubscription } from '@rocket.chat/core-typings';
+import type { IMessage } from '@rocket.chat/core-typings';
 import { isE2EEPinnedMessage } from '@rocket.chat/core-typings';
 import { Meteor } from 'meteor/meteor';
 import { Tracker } from 'meteor/tracker';
 
 import { E2EEState } from '../../app/e2e/client/E2EEState';
 import { e2e } from '../../app/e2e/client/rocketchat.e2e';
-import { Subscriptions, ChatRoom } from '../../app/models/client';
+import { ChatRoom } from '../../app/models/client';
 import { settings } from '../../app/settings/client';
-import { sdk } from '../../app/utils/client/lib/SDKClient';
 import { onClientBeforeSendMessage } from '../lib/onClientBeforeSendMessage';
 import { onClientMessageReceived } from '../lib/onClientMessageReceived';
 import { isLayoutEmbedded } from '../lib/utils/isLayoutEmbedded';
@@ -17,10 +16,12 @@ import { router } from '../providers/RouterProvider';
 Meteor.startup(() => {
 	Tracker.autorun(() => {
 		if (!Meteor.userId()) {
+			e2e.log('Not logged in');
 			return;
 		}
 
 		if (!window.crypto) {
+			e2e.error('No crypto support');
 			return;
 		}
 
@@ -29,98 +30,33 @@ Meteor.startup(() => {
 		const adminEmbedded = isLayoutEmbedded() && router.getLocationPathname().startsWith('/admin');
 
 		if (enabled && !adminEmbedded) {
+			e2e.log('E2E enabled starting client');
 			e2e.startClient();
 		} else {
+			e2e.log('E2E disabled');
 			e2e.setState(E2EEState.DISABLED);
 			e2e.closeAlert();
 		}
 	});
 
-	let observable: Meteor.LiveQueryHandle | null = null;
 	let offClientMessageReceived: undefined | (() => void);
 	let offClientBeforeSendMessage: undefined | (() => void);
-	let unsubNotifyUser: undefined | (() => void);
 	let listenersAttached = false;
 
 	Tracker.autorun(() => {
 		if (!e2e.isReady()) {
+			e2e.log('Not ready');
 			offClientMessageReceived?.();
-			unsubNotifyUser?.();
-			unsubNotifyUser = undefined;
-			observable?.stop();
 			offClientBeforeSendMessage?.();
 			listenersAttached = false;
 			return;
 		}
 
 		if (listenersAttached) {
+			e2e.log('Listeners already attached');
 			return;
 		}
 
-		unsubNotifyUser = sdk.stream('notify-user', [`${Meteor.userId()}/e2ekeyRequest`], async (roomId, keyId): Promise<void> => {
-			const e2eRoom = await e2e.getInstanceByRoomId(roomId);
-			if (!e2eRoom) {
-				return;
-			}
-
-			e2eRoom.provideKeyToUser(keyId);
-		}).stop;
-
-		observable = Subscriptions.find().observe({
-			changed: async (sub: ISubscription) => {
-				setTimeout(async () => {
-					if (!sub.encrypted && !sub.E2EKey) {
-						e2e.removeInstanceByRoomId(sub.rid);
-						return;
-					}
-
-					const e2eRoom = await e2e.getInstanceByRoomId(sub.rid);
-					if (!e2eRoom) {
-						return;
-					}
-
-					if (sub.E2ESuggestedKey) {
-						if (await e2eRoom.importGroupKey(sub.E2ESuggestedKey)) {
-							e2e.acceptSuggestedKey(sub.rid);
-						} else {
-							console.warn('Invalid E2ESuggestedKey, rejecting', sub.E2ESuggestedKey);
-							e2e.rejectSuggestedKey(sub.rid);
-						}
-					}
-
-					sub.encrypted ? e2eRoom.resume() : e2eRoom.pause();
-
-					// Cover private groups and direct messages
-					if (!e2eRoom.isSupportedRoomType(sub.t)) {
-						e2eRoom.disable();
-						return;
-					}
-
-					if (sub.E2EKey && e2eRoom.isWaitingKeys()) {
-						e2eRoom.keyReceived();
-						return;
-					}
-
-					if (!e2eRoom.isReady()) {
-						return;
-					}
-
-					e2eRoom.decryptSubscription();
-				}, 0);
-			},
-			added: async (sub: ISubscription) => {
-				setTimeout(async () => {
-					if (!sub.encrypted && !sub.E2EKey) {
-						return;
-					}
-					return e2e.getInstanceByRoomId(sub.rid);
-				}, 0);
-			},
-			removed: (sub: ISubscription) => {
-				e2e.removeInstanceByRoomId(sub.rid);
-			},
-		});
-
 		offClientMessageReceived = onClientMessageReceived.use(async (msg: IMessage) => {
 			const e2eRoom = await e2e.getInstanceByRoomId(msg.rid);
 			if (!e2eRoom?.shouldConvertReceivedMessages()) {
@@ -157,5 +93,6 @@ Meteor.startup(() => {
 		});
 
 		listenersAttached = true;
+		e2e.log('Listeners attached', listenersAttached);
 	});
 });

apps/meteor/client/views/teams/contextualBar/channels/hooks/useTeamsChannelList.ts

@@ -41,12 +41,13 @@ export const useTeamsChannelList = (
 			});
 
 			return {
-				items: rooms.map(({ _updatedAt, lastMessage, lm, ts, webRtcCallStartTime, ...room }) => ({
+				items: rooms.map(({ _updatedAt, lastMessage, lm, ts, webRtcCallStartTime, usersWaitingForE2EKeys, ...room }) => ({
 					...(lm && { lm: new Date(lm) }),
 					...(ts && { ts: new Date(ts) }),
 					_updatedAt: new Date(_updatedAt),
 					...(lastMessage && { lastMessage: mapMessageFromApi(lastMessage) }),
 					...(webRtcCallStartTime && { webRtcCallStartTime: new Date(webRtcCallStartTime) }),
+					...(usersWaitingForE2EKeys && usersWaitingForE2EKeys.map(({ userId, ts }) => ({ userId, ts: new Date(ts) }))),
 					...room,
 				})),
 				itemCount: total,

apps/meteor/lib/publishFields.ts

@@ -120,4 +120,5 @@ export const roomFields = {
 	callDuration: 1,
 	callTotalHoldTime: 1,
 	callWaitingTime: 1,
+	usersWaitingForE2EKeys: 1,
 } as const;

apps/meteor/server/lib/resetUserE2EKey.ts

@@ -1,5 +1,5 @@
 import { api } from '@rocket.chat/core-services';
-import { Subscriptions, Users } from '@rocket.chat/models';
+import { Subscriptions, Users, Rooms } from '@rocket.chat/models';
 import { Meteor } from 'meteor/meteor';
 
 import * as Mailer from '../../app/mailer/server/api';
@@ -71,6 +71,7 @@ export async function resetUserE2EEncriptionKey(uid: string, notifyUser: boolean
 
 	await Users.resetE2EKey(uid);
 	await Subscriptions.resetUserE2EKey(uid);
+	await Rooms.removeUserFromE2EEQueue(uid);
 
 	// Force the user to logout, so that the keys can be generated again
 	await Users.unsetLoginTokens(uid);

apps/meteor/server/methods/removeUserFromRoom.ts

@@ -8,6 +8,7 @@ import { canAccessRoomAsync, getUsersInRole } from '../../app/authorization/serv
 import { hasPermissionAsync } from '../../app/authorization/server/functions/hasPermission';
 import { hasRoleAsync } from '../../app/authorization/server/functions/hasRole';
 import { notifyOnRoomChanged } from '../../app/lib/server/lib/notifyListener';
+import { settings } from '../../app/settings/server';
 import { RoomMemberActions } from '../../definition/IRoomTypeConfig';
 import { callbacks } from '../../lib/callbacks';
 import { afterRemoveFromRoomCallback } from '../../lib/callbacks/afterRemoveFromRoomCallback';
@@ -89,7 +90,11 @@ export const removeUserFromRoomMethod = async (fromId: string, data: { rid: stri
 		await Team.removeMember(room.teamId, removedUser._id);
 	}
 
-	setImmediate(async () => {
+	if (room.encrypted && settings.get('E2E_Enable')) {
+		await Rooms.removeUsersFromE2EEQueueByRoomId(room._id, [removedUser._id]);
+	}
+
+	setImmediate(() => {
 		void afterRemoveFromRoomCallback.run({ removedUser, userWhoRemoved: fromUser }, room);
 		void notifyOnRoomChanged(room);
 	});

apps/meteor/server/models/raw/Rooms.ts

@@ -93,6 +93,16 @@ export class RoomsRaw extends BaseRaw<IRoom> implements IRoomsModel {
 				sparse: true,
 			},
 			{ key: { t: 1, ts: 1 } },
+			{
+				key: {
+					'usersWaitingForE2EKeys.userId': 1,
+				},
+				partialFilterExpression: {
+					'usersWaitingForE2EKeys.userId': {
+						$exists: true,
+					},
+				},
+			},
 		];
 	}
 
@@ -1977,4 +1987,106 @@ export class RoomsRaw extends BaseRaw<IRoom> implements IRoomsModel {
 
 		return this.updateOne(query, update);
 	}
+
+	async getSubscribedRoomIdsWithoutE2EKeys(uid: IUser['_id']): Promise<IRoom['_id'][]> {
+		return (
+			await this.col
+				.aggregate([
+					{ $match: { encrypted: true } },
+					{
+						$lookup: {
+							from: 'rocketchat_subscription',
+							localField: '_id',
+							foreignField: 'rid',
+							as: 'subs',
+						},
+					},
+					{
+						$unwind: '$subs',
+					},
+					{
+						$match: {
+							'subs.u._id': uid,
+							'subs.E2EKey': {
+								$exists: false,
+							},
+						},
+					},
+					{
+						$project: {
+							_id: 1,
+						},
+					},
+				])
+				.toArray()
+		).map(({ _id }) => _id);
+	}
+
+	addUserIdToE2EEQueueByRoomIds(roomIds: IRoom['_id'][], uid: IUser['_id']): Promise<Document | UpdateResult> {
+		const query: Filter<IRoom> = {
+			'_id': {
+				$in: roomIds,
+			},
+			'usersWaitingForE2EKeys.userId': { $ne: uid },
+			'encrypted': true,
+		};
+
+		const update: UpdateFilter<IRoom> = {
+			$push: {
+				usersWaitingForE2EKeys: {
+					$each: [
+						{
+							userId: uid,
+							ts: new Date(),
+						},
+					],
+					$slice: -50,
+				},
+			},
+		};
+
+		return this.updateMany(query, update);
+	}
+
+	async removeUsersFromE2EEQueueByRoomId(roomId: IRoom['_id'], uids: IUser['_id'][]): Promise<Document | UpdateResult> {
+		const query: Filter<IRoom> = {
+			'_id': roomId,
+			'usersWaitingForE2EKeys.userId': {
+				$in: uids,
+			},
+			'encrypted': true,
+		};
+
+		const update: UpdateFilter<IRoom> = {
+			$pull: {
+				usersWaitingForE2EKeys: { userId: { $in: uids } },
+			},
+		};
+
+		await this.updateMany(query, update);
+
+		return this.updateMany(
+			{
+				'_id': roomId,
+				'usersWaitingForE2EKeys.0': { $exists: false },
+				'encrypted': true,
+			},
+			{ $unset: { usersWaitingForE2EKeys: 1 } },
+		);
+	}
+
+	async removeUserFromE2EEQueue(uid: IUser['_id']): Promise<Document | UpdateResult> {
+		const query: Filter<IRoom> = {
+			'usersWaitingForE2EKeys.userId': uid,
+			'encrypted': true,
+		};
+
+		const update: UpdateFilter<IRoom> = {
+			$pull: {
+				usersWaitingForE2EKeys: { userId: uid },
+			},
+		};
+
+		return this.updateMany(query, update);
+	}
 }

apps/meteor/server/models/raw/Subscriptions.ts

@@ -27,6 +27,7 @@ import type {
 	UpdateFilter,
 	InsertOneResult,
 	InsertManyResult,
+	AggregationCursor,
 } from 'mongodb';
 
 import { getDefaultSubscriptionPref } from '../../../app/utils/lib/getDefaultSubscriptionPref';
@@ -538,8 +539,8 @@ export class SubscriptionsRaw extends BaseRaw<ISubscription> implements ISubscri
 		return this.findOneById(_id);
 	}
 
-	setGroupE2ESuggestedKey(_id: string, key: string): Promise<UpdateResult | Document> {
-		const query = { _id };
+	setGroupE2ESuggestedKey(uid: string, rid: string, key: string): Promise<UpdateResult> {
+		const query = { rid, 'u._id': uid };
 		const update = { $set: { E2ESuggestedKey: key } };
 		return this.updateOne(query, update);
 	}
@@ -674,6 +675,61 @@ export class SubscriptionsRaw extends BaseRaw<ISubscription> implements ISubscri
 		return this.find(query, options);
 	}
 
+	findUsersWithPublicE2EKeyByRids(
+		rids: IRoom['_id'][],
+		excludeUserId: IUser['_id'],
+		usersLimit = 50,
+	): AggregationCursor<{ rid: IRoom['_id']; users: { _id: IUser['_id']; public_key: string }[] }> {
+		return this.col.aggregate([
+			{
+				$match: {
+					'rid': {
+						$in: rids,
+					},
+					'E2EKey': {
+						$exists: false,
+					},
+					'u._id': {
+						$ne: excludeUserId,
+					},
+				},
+			},
+			{
+				$lookup: {
+					from: 'users',
+					localField: 'u._id',
+					foreignField: '_id',
+					as: 'user',
+				},
+			},
+			{
+				$unwind: '$user',
+			},
+			{
+				$match: {
+					'user.e2e.public_key': {
+						$exists: 1,
+					},
+				},
+			},
+			{
+				$group: {
+					_id: {
+						rid: '$rid',
+					},
+					users: { $push: { _id: '$user._id', public_key: '$user.e2e.public_key' } },
+				},
+			},
+			{
+				$project: {
+					rid: '$_id.rid',
+					users: { $slice: ['$users', usersLimit] },
+					_id: 0,
+				},
+			},
+		]);
+	}
+
 	updateAudioNotificationValueById(_id: string, audioNotificationValue: string): Promise<UpdateResult> {
 		const query = {
 			_id,
@@ -871,6 +927,7 @@ export class SubscriptionsRaw extends BaseRaw<ISubscription> implements ISubscri
 			{
 				$unset: {
 					E2EKey: '',
+					E2ESuggestedKey: 1,
 				},
 			},
 		);

packages/core-typings/src/IRoom.ts

@@ -95,6 +95,8 @@ export interface IRoom extends IRocketChatRecord {
 	customFields?: Record<string, any>;
 
 	channel?: { _id: string };
+
+	usersWaitingForE2EKeys?: { userId: IUser['_id']; ts: Date }[];
 }
 
 export const isRoomWithJoinCode = (room: Partial<IRoom>): room is IRoomWithJoinCode =>

packages/i18n/src/locales/en.i18n.json

@@ -1803,9 +1803,9 @@
   "E2E_password_request_text": "To access your encrypted channels and direct messages, enter your encryption password. This is not stored on the server, so you’ll need to use it on every device.",
   "E2E_password_reveal_text": "Create secure private rooms and direct messages with end-to-end encryption. This password won’t be stored on the server. You can use it on all your devices.",
   "E2E_password_save_text": "This will only be displayed once, please save it now.",
-  "E2E_Reset_Email_Content": "You've been automatically logged out. When you login again, Rocket.Chat will generate a new key and restore your access to any encrypted room that has one or more members online. Due to the nature of the E2E encryption, Rocket.Chat will not be able to restore access to any encrypted room that has no member online.",
-  "E2E_Reset_Key_Explanation": "This option will remove your current E2E key and log you out. <BR/>When you login again, Rocket.Chat will generate you a new key and restore your access to any encrypted room that has one or more members online.<BR/>Due to the nature of the E2E encryption, Rocket.Chat will not be able to restore access to any encrypted room that has no member online.",
-  "E2E_Reset_Other_Key_Warning": "Reset the current E2E key will log out the user. When the user login again, Rocket.Chat will generate a new key and restore the user access to any encrypted room that has one or more members online. Due to the nature of the E2E encryption, Rocket.Chat will not be able to restore access to any encrypted room that has no member online.",
+  "E2E_Reset_Email_Content": "You've been automatically logged out. When you log in again, a new key will be generated and access will be restored to any encrypted room with at least one member online. If no members are online, access will be restored as soon as a member comes online.",
+  "E2E_Reset_Key_Explanation": "This will remove your current E2EE key and log you out. <BR/>When you log in again, a new key will be generated and access will be restored to any encrypted room with at least one member online.<BR/>If no members are online, access will be restored as soon as a member comes online.",
+  "E2E_Reset_Other_Key_Warning": "Resetting the E2EE key will log out the user. When the user logs in again, a new key will be generated and access will be restored to any encrypted room with at least one member online. If no members are online, access will be restored as soon as a member comes online.",
   "E2E_unavailable_for_federation": "E2E is unavailable for federated rooms",
   "ECDH_Enabled": "Enable second layer encryption for data transport",
   "Edit": "Edit",

packages/model-typings/src/models/IRoomsModel.ts

@@ -278,4 +278,8 @@ export interface IRoomsModel extends IBaseModel<IRoom> {
 	removeDirectRoomContainingUsername(username: string): Promise<DeleteResult>;
 	countDiscussions(): Promise<number>;
 	setOTRForDMByRoomID(rid: string): Promise<UpdateResult>;
+	addUserIdToE2EEQueueByRoomIds(roomIds: IRoom['_id'][], uid: IUser['_id']): Promise<Document | UpdateResult>;
+	getSubscribedRoomIdsWithoutE2EKeys(uid: IUser['_id']): Promise<IRoom['_id'][]>;
+	removeUsersFromE2EEQueueByRoomId(roomId: IRoom['_id'], uids: IUser['_id'][]): Promise<Document | UpdateResult>;
+	removeUserFromE2EEQueue(uid: IUser['_id']): Promise<Document | UpdateResult>;
 }

packages/model-typings/src/models/ISubscriptionsModel.ts

@@ -9,6 +9,7 @@ import type {
 	Filter,
 	InsertOneResult,
 	InsertManyResult,
+	AggregationCursor,
 } from 'mongodb';
 
 import type { IBaseModel } from './IBaseModel';
@@ -96,7 +97,7 @@ export interface ISubscriptionsModel extends IBaseModel<ISubscription> {
 
 	setGroupE2EKey(_id: string, key: string): Promise<ISubscription | null>;
 
-	setGroupE2ESuggestedKey(_id: string, key: string): Promise<UpdateResult | Document>;
+	setGroupE2ESuggestedKey(uid: string, rid: string, key: string): Promise<UpdateResult>;
 
 	unsetGroupE2ESuggestedKey(_id: string): Promise<UpdateResult | Document>;
 
@@ -128,6 +129,11 @@ export interface ISubscriptionsModel extends IBaseModel<ISubscription> {
 	getAutoTranslateLanguagesByRoomAndNotUser(rid: string, userId: string): Promise<(string | undefined)[]>;
 
 	findByRidWithoutE2EKey(rid: string, options: FindOptions<ISubscription>): FindCursor<ISubscription>;
+	findUsersWithPublicE2EKeyByRids(
+		rids: IRoom['_id'][],
+		excludeUserId: IUser['_id'],
+		usersLimit?: number,
+	): AggregationCursor<{ rid: IRoom['_id']; users: { _id: IUser['_id']; public_key: string }[] }>;
 	findByUserId(userId: string, options?: FindOptions<ISubscription>): FindCursor<ISubscription>;
 	cachedFindByUserId(userId: string, options?: FindOptions<ISubscription>): FindCursor<ISubscription>;
 	updateAutoTranslateById(_id: string, autoTranslate: boolean): Promise<UpdateResult>;

packages/rest-typings/src/index.ts

@@ -258,6 +258,7 @@ export * from './v1/e2e/e2eGetUsersOfRoomWithoutKeyParamsGET';
 export * from './v1/e2e/e2eSetRoomKeyIDParamsPOST';
 export * from './v1/e2e/e2eSetUserPublicAndPrivateKeysParamsPOST';
 export * from './v1/e2e/e2eUpdateGroupKeyParamsPOST';
+export * from './v1/e2e';
 export * from './v1/import';
 export * from './v1/voip';
 export * from './v1/email-inbox';

packages/rest-typings/src/v1/e2e.ts

@@ -1,4 +1,4 @@
-import type { IUser } from '@rocket.chat/core-typings';
+import type { IRoom, IUser } from '@rocket.chat/core-typings';
 import Ajv from 'ajv';
 
 const ajv = new Ajv({
@@ -88,6 +88,55 @@ const E2eSetRoomKeyIdSchema = {
 
 export const isE2eSetRoomKeyIdProps = ajv.compile<E2eSetRoomKeyIdProps>(E2eSetRoomKeyIdSchema);
 
+type E2EProvideUsersGroupKeyProps = {
+	usersSuggestedGroupKeys: Record<IRoom['_id'], { _id: IUser['_id']; key: string }[]>;
+};
+
+const E2EProvideUsersGroupKeySchema = {
+	type: 'object',
+	properties: {
+		usersSuggestedGroupKeys: {
+			type: 'object',
+			additionalProperties: {
+				type: 'array',
+				items: {
+					type: 'object',
+					properties: {
+						_id: { type: 'string' },
+						key: { type: 'string' },
+					},
+					required: ['_id', 'key'],
+					additionalProperties: false,
+				},
+			},
+		},
+	},
+	required: ['usersSuggestedGroupKeys'],
+	additionalProperties: false,
+};
+
+export const isE2EProvideUsersGroupKeyProps = ajv.compile<E2EProvideUsersGroupKeyProps>(E2EProvideUsersGroupKeySchema);
+
+type E2EFetchUsersWaitingForGroupKeyProps = { roomIds: string[] };
+
+const E2EFetchUsersWaitingForGroupKeySchema = {
+	type: 'object',
+	properties: {
+		roomIds: {
+			type: 'array',
+			items: {
+				type: 'string',
+			},
+		},
+	},
+	required: ['roomIds'],
+	additionalProperties: false,
+};
+
+export const isE2EFetchUsersWaitingForGroupKeyProps = ajv.compile<E2EFetchUsersWaitingForGroupKeyProps>(
+	E2EFetchUsersWaitingForGroupKeySchema,
+);
+
 export type E2eEndpoints = {
 	'/v1/e2e.setUserPublicAndPrivateKeys': {
 		POST: (params: E2eSetUserPublicAndPrivateKeysProps) => void;
@@ -112,4 +161,12 @@ export type E2eEndpoints = {
 	'/v1/e2e.fetchMyKeys': {
 		GET: () => { public_key: string; private_key: string };
 	};
+	'/v1/e2e.fetchUsersWaitingForGroupKey': {
+		GET: (params: E2EFetchUsersWaitingForGroupKeyProps) => {
+			usersWaitingForE2EKeys: Record<IRoom['_id'], { _id: IUser['_id']; public_key: string }[]>;
+		};
+	};
+	'/v1/e2e.provideUsersSuggestedGroupKeys': {
+		POST: (params: E2EProvideUsersGroupKeyProps) => void;
+	};
 };