mirror of
https://github.com/vector-im/element-call.git
synced 2026-02-02 04:05:56 +00:00
temp refactored membership rtcidentity
This commit is contained in:
@@ -6,18 +6,13 @@ Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { BaseKeyProvider } from "livekit-client";
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
type MatrixRTCSession,
|
||||
MatrixRTCSessionEvent,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
||||
import { firstValueFrom } from "rxjs";
|
||||
|
||||
import {
|
||||
computeLivekitParticipantIdentity$,
|
||||
livekitIdentityInput,
|
||||
} from "../state/CallViewModel/remoteMembers/LivekitParticipantIdentity";
|
||||
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
|
||||
const logger = rootLogger.getChild("[MatrixKeyProvider]");
|
||||
|
||||
export class MatrixKeyProvider extends BaseKeyProvider {
|
||||
private rtcSession?: MatrixRTCSession;
|
||||
@@ -32,6 +27,10 @@ export class MatrixKeyProvider extends BaseKeyProvider {
|
||||
MatrixRTCSessionEvent.EncryptionKeyChanged,
|
||||
this.onEncryptionKeyChanged,
|
||||
);
|
||||
this.rtcSession.off(
|
||||
MatrixRTCSessionEvent.MembershipsChanged,
|
||||
this.onMembershipsChanged,
|
||||
);
|
||||
}
|
||||
|
||||
this.rtcSession = rtcSession;
|
||||
@@ -40,55 +39,86 @@ export class MatrixKeyProvider extends BaseKeyProvider {
|
||||
MatrixRTCSessionEvent.EncryptionKeyChanged,
|
||||
this.onEncryptionKeyChanged,
|
||||
);
|
||||
this.rtcSession.on(
|
||||
MatrixRTCSessionEvent.MembershipsChanged,
|
||||
this.onMembershipsChanged,
|
||||
);
|
||||
|
||||
// The new session could be aware of keys of which the old session wasn't,
|
||||
// so emit key changed events
|
||||
this.rtcSession.reemitEncryptionKeys();
|
||||
}
|
||||
|
||||
private keyCache = new Array<{
|
||||
membership: CallMembershipIdentityParts;
|
||||
encryptionKey: Uint8Array;
|
||||
encryptionKeyIndex: number;
|
||||
}>();
|
||||
|
||||
private onMembershipsChanged = (): void => {
|
||||
const duplicatedArray = this.keyCache;
|
||||
// Reset key cache first. It will get repopulated when calling `onEncryptionKeyChanged`
|
||||
this.keyCache = [];
|
||||
let next = duplicatedArray.pop();
|
||||
while (next !== undefined) {
|
||||
logger.debug(
|
||||
"[KeyCache] remove key event from the cache and try adding it again. For membership: ",
|
||||
next.membership,
|
||||
);
|
||||
this.onEncryptionKeyChanged(
|
||||
next.encryptionKey,
|
||||
next.encryptionKeyIndex,
|
||||
next.membership,
|
||||
);
|
||||
next = duplicatedArray.pop();
|
||||
}
|
||||
};
|
||||
|
||||
private onEncryptionKeyChanged = (
|
||||
encryptionKey: Uint8Array,
|
||||
encryptionKeyIndex: number,
|
||||
membership: CallMembershipIdentityParts,
|
||||
): void => {
|
||||
const unhashedIdentity = livekitIdentityInput(membership);
|
||||
|
||||
// This is the only way we can get the kind of the membership event we just received the key for.
|
||||
// best case we want to recompute this once the memberships change (you can receive the key before the participant...)
|
||||
//
|
||||
// TODO change this to `?? "rtc"` for newer versions.
|
||||
const kind =
|
||||
this.rtcSession?.memberships.find(
|
||||
(m) =>
|
||||
m.userId === membership.userId &&
|
||||
m.deviceId === membership.deviceId &&
|
||||
m.memberId === membership.memberId,
|
||||
)?.kind ?? "session";
|
||||
const membershipFull = this.rtcSession?.memberships.find(
|
||||
(m) =>
|
||||
m.userId === membership.userId &&
|
||||
m.deviceId === membership.deviceId &&
|
||||
m.memberId === membership.memberId,
|
||||
);
|
||||
if (!membershipFull) {
|
||||
logger.debug(
|
||||
"[KeyCache] Added key event to the cache because we do not have a membership for it (yet): ",
|
||||
membership,
|
||||
);
|
||||
this.keyCache.push({ membership, encryptionKey, encryptionKeyIndex });
|
||||
return;
|
||||
}
|
||||
|
||||
Promise.all([
|
||||
crypto.subtle.importKey("raw", encryptionKey, "HKDF", false, [
|
||||
crypto.subtle
|
||||
.importKey("raw", encryptionKey, "HKDF", false, [
|
||||
"deriveBits",
|
||||
"deriveKey",
|
||||
]),
|
||||
firstValueFrom(computeLivekitParticipantIdentity$(membership, kind)),
|
||||
]).then(
|
||||
([keyMaterial, livekitParticipantId]) => {
|
||||
this.onSetEncryptionKey(
|
||||
keyMaterial,
|
||||
livekitParticipantId,
|
||||
encryptionKeyIndex,
|
||||
);
|
||||
])
|
||||
.then(
|
||||
(keyMaterial) => {
|
||||
this.onSetEncryptionKey(
|
||||
keyMaterial,
|
||||
membershipFull.rtcBackendIdentity,
|
||||
encryptionKeyIndex,
|
||||
);
|
||||
|
||||
logger.debug(
|
||||
`Sent new key to livekit room=${this.rtcSession?.room.roomId} participantId=${livekitParticipantId} (before hash: ${unhashedIdentity}) encryptionKeyIndex=${encryptionKeyIndex}`,
|
||||
);
|
||||
},
|
||||
(e) => {
|
||||
logger.error(
|
||||
`Failed to create key material from buffer for livekit room=${this.rtcSession?.room.roomId} participantId before hash=${unhashedIdentity} encryptionKeyIndex=${encryptionKeyIndex}`,
|
||||
e,
|
||||
);
|
||||
},
|
||||
);
|
||||
logger.debug(
|
||||
`Sent new key to livekit room=${this.rtcSession?.room.roomId} participantId=${membershipFull.rtcBackendIdentity} (before hash: ${membershipFull.userId}) encryptionKeyIndex=${encryptionKeyIndex}`,
|
||||
);
|
||||
},
|
||||
(e) => {
|
||||
logger.error(
|
||||
`Failed to create key material from buffer for livekit room=${this.rtcSession?.room.roomId} participantId before hash=${membershipFull.userId} encryptionKeyIndex=${encryptionKeyIndex}`,
|
||||
e,
|
||||
);
|
||||
},
|
||||
);
|
||||
};
|
||||
}
|
||||
|
||||
@@ -44,7 +44,6 @@ import {
|
||||
mockRtcMembership,
|
||||
testScope,
|
||||
exampleTransport,
|
||||
mockComputeLivekitParticipantIdentity$,
|
||||
} from "../../utils/test.ts";
|
||||
import { E2eeType } from "../../e2ee/e2eeType.ts";
|
||||
import {
|
||||
@@ -88,14 +87,6 @@ vi.mock(
|
||||
}),
|
||||
);
|
||||
|
||||
vi.mock(
|
||||
import("./remoteMembers/LivekitParticipantIdentity.ts"),
|
||||
async (importOriginal) => ({
|
||||
...(await importOriginal()),
|
||||
computeLivekitParticipantIdentity$: mockComputeLivekitParticipantIdentity$,
|
||||
}),
|
||||
);
|
||||
|
||||
const yesNo = {
|
||||
y: true,
|
||||
n: false,
|
||||
|
||||
@@ -1,53 +0,0 @@
|
||||
/*
|
||||
Copyright 2025 New Vector Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { encodeUnpaddedBase64Url } from "matrix-js-sdk";
|
||||
import { sha256 } from "matrix-js-sdk/lib/digest";
|
||||
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
||||
import { from, type Observable } from "rxjs";
|
||||
|
||||
const livekitParticipantIdentityCache = new Map<string, string>();
|
||||
|
||||
/**
|
||||
* The string that is computed based on the membership and used for the computing the hash.
|
||||
* `${userId}:${deviceId}:${membershipID}`
|
||||
* as the direct imput for: await sha256(input)
|
||||
*/
|
||||
export const livekitIdentityInput = ({
|
||||
userId,
|
||||
deviceId,
|
||||
memberId,
|
||||
}: CallMembershipIdentityParts): string => `${userId}|${deviceId}|${memberId}`;
|
||||
|
||||
export function computeLivekitParticipantIdentity$(
|
||||
membership: CallMembershipIdentityParts,
|
||||
kind: "rtc" | "session",
|
||||
): Observable<string> {
|
||||
const compute = async (): Promise<string> => {
|
||||
switch (kind) {
|
||||
case "rtc": {
|
||||
const input = livekitIdentityInput(membership);
|
||||
if (livekitParticipantIdentityCache.size > 400)
|
||||
// prevent memory leaks in a stupid/simple way
|
||||
livekitParticipantIdentityCache.clear();
|
||||
// TODO use non deprecated memberId
|
||||
if (livekitParticipantIdentityCache.has(input))
|
||||
return livekitParticipantIdentityCache.get(input)!;
|
||||
else {
|
||||
const hashBuffer = await sha256(input);
|
||||
const hashedString = encodeUnpaddedBase64Url(hashBuffer);
|
||||
livekitParticipantIdentityCache.set(input, hashedString);
|
||||
return hashedString;
|
||||
}
|
||||
}
|
||||
case "session":
|
||||
default:
|
||||
return `${membership.userId}:${membership.deviceId}`;
|
||||
}
|
||||
};
|
||||
return from(compute());
|
||||
}
|
||||
@@ -10,7 +10,7 @@ import {
|
||||
type LivekitTransport,
|
||||
type CallMembership,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { combineLatest, filter, map, switchMap } from "rxjs";
|
||||
import { combineLatest, filter, map } from "rxjs";
|
||||
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
|
||||
|
||||
import { type Behavior } from "../../Behavior";
|
||||
@@ -18,7 +18,6 @@ import { type IConnectionManager } from "./ConnectionManager";
|
||||
import { Epoch, type ObservableScope } from "../../ObservableScope";
|
||||
import { type Connection } from "./Connection";
|
||||
import { generateItemsWithEpoch } from "../../../utils/observable";
|
||||
import { computeLivekitParticipantIdentity$ } from "./LivekitParticipantIdentity";
|
||||
|
||||
const logger = rootLogger.getChild("[MatrixLivekitMembers]");
|
||||
|
||||
@@ -82,45 +81,12 @@ export function createMatrixLivekitMembers$({
|
||||
membershipsWithTransport$,
|
||||
connectionManager,
|
||||
}: Props): Behavior<Epoch<RemoteMatrixLivekitMember[]>> {
|
||||
/**
|
||||
* This internal observable is used to compute the async sha256 hash of the user's identity.
|
||||
* a promise is treated like an observable. So we can switchMap on the promise from the identity computation.
|
||||
* The last update to `membershipsWithTransport$` will always be the last promise we pass to switchMap.
|
||||
* So we will eventually always end up with the latest memberships and their identities.
|
||||
*/
|
||||
const membershipsWithTransportAndLivekitIdentity$ =
|
||||
membershipsWithTransport$.pipe(
|
||||
switchMap((membershipsWithTransport) => {
|
||||
const { value, epoch } = membershipsWithTransport;
|
||||
const membershipsWithTransportAndLkIdentityPromises = value.map(
|
||||
(obj) => {
|
||||
return computeLivekitParticipantIdentity$(
|
||||
obj.membership,
|
||||
obj.membership.kind,
|
||||
);
|
||||
},
|
||||
);
|
||||
return combineLatest(
|
||||
membershipsWithTransportAndLkIdentityPromises,
|
||||
).pipe(
|
||||
map((identities) => {
|
||||
const membershipsWithTransportAndLkIdentity = value.map(
|
||||
({ transport, membership }, index) => {
|
||||
return { transport, membership, identity: identities[index] };
|
||||
},
|
||||
);
|
||||
return new Epoch(membershipsWithTransportAndLkIdentity, epoch);
|
||||
}),
|
||||
);
|
||||
}),
|
||||
);
|
||||
|
||||
/**
|
||||
* Stream of all the call members and their associated livekit data (if available).
|
||||
*/
|
||||
return scope.behavior(
|
||||
combineLatest([
|
||||
membershipsWithTransportAndLivekitIdentity$,
|
||||
membershipsWithTransport$,
|
||||
connectionManager.connectionManagerData$,
|
||||
]).pipe(
|
||||
filter((values) =>
|
||||
@@ -131,37 +97,34 @@ export function createMatrixLivekitMembers$({
|
||||
// Generator function.
|
||||
// creates an array of `{key, data}[]`
|
||||
// Each change in the keys (new key, missing key) will result in a call to the factory function.
|
||||
function* ([membershipsWithTransportAndLivekitIdentity, managerData]) {
|
||||
for (const {
|
||||
membership,
|
||||
transport,
|
||||
identity,
|
||||
} of membershipsWithTransportAndLivekitIdentity) {
|
||||
function* ([membershipsWithTransport, managerData]) {
|
||||
for (const { membership, transport } of membershipsWithTransport) {
|
||||
const participants = transport
|
||||
? managerData.getParticipantForTransport(transport)
|
||||
: [];
|
||||
const participant =
|
||||
participants.find((p) => p.identity == identity) ?? null;
|
||||
participants.find(
|
||||
(p) => p.identity == membership.rtcBackendIdentity,
|
||||
) ?? null;
|
||||
const connection = transport
|
||||
? managerData.getConnectionForTransport(transport)
|
||||
: null;
|
||||
|
||||
yield {
|
||||
keys: [identity, membership.userId, membership.deviceId],
|
||||
keys: [membership.userId, membership.deviceId],
|
||||
data: { membership, participant, connection },
|
||||
};
|
||||
}
|
||||
},
|
||||
// Each update where the key of the generator array do not change will result in updates to the `data$` observable in the factory.
|
||||
(scope, data$, identity, userId, deviceId) => {
|
||||
(scope, data$, userId, deviceId) => {
|
||||
logger.debug(
|
||||
`Generating member for livekitIdentity: ${identity}, userId:deviceId: ${userId}${deviceId}`,
|
||||
`Generating member for livekitIdentity: ${data$.value.membership.rtcBackendIdentity}, userId:deviceId: ${userId}${deviceId}`,
|
||||
);
|
||||
const { participant$, ...rest } = scope.splitBehavior(data$);
|
||||
// will only get called once per `participantId, userId` pair.
|
||||
// updates to data$ and as a result to displayName$ and mxcAvatarUrl$ are more frequent.
|
||||
return {
|
||||
identity,
|
||||
userId,
|
||||
participant: { type: "remote" as const, value$: participant$ },
|
||||
...rest,
|
||||
|
||||
@@ -22,7 +22,6 @@ import { ECConnectionFactory } from "./ConnectionFactory.ts";
|
||||
import { type OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
|
||||
import {
|
||||
mockCallMembership,
|
||||
mockComputeLivekitParticipantIdentity$,
|
||||
mockMediaDevices,
|
||||
ownMemberMock,
|
||||
withTestScheduler,
|
||||
@@ -45,11 +44,6 @@ let lkRoomFactory: () => LivekitRoom;
|
||||
|
||||
const createdMockLivekitRooms: Map<string, LivekitRoom> = new Map();
|
||||
|
||||
vi.mock(import("./LivekitParticipantIdentity.ts"), async (importOriginal) => ({
|
||||
...(await importOriginal()),
|
||||
computeLivekitParticipantIdentity$: mockComputeLivekitParticipantIdentity$,
|
||||
}));
|
||||
|
||||
beforeEach(() => {
|
||||
testScope = new ObservableScope();
|
||||
mockClient = {
|
||||
|
||||
@@ -252,7 +252,8 @@ export function mockRtcMembership(
|
||||
content: data,
|
||||
});
|
||||
|
||||
const cms = new CallMembership(event, data);
|
||||
const membershipData = CallMembership.membershipDataFromMatrixEvent(event);
|
||||
const cms = new CallMembership(event, membershipData, "xx");
|
||||
vi.mocked(cms).getTransport = vi.fn().mockReturnValue(fociPreferred[0]);
|
||||
return cms;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user