From e336ee5cfad56d8f3b7ea1ebf48d7e70cd500470 Mon Sep 17 00:00:00 2001 From: Robin Date: Mon, 12 Oct 2026 10:33:00 +0200 Subject: [PATCH] Get SFU tokens and delegate delayed leaves through the MSC4195 CS API src/livekit/auth now fronts two flows. For a transport with a LiveKit SFU `url`, tokens come from our own homeserver's MSC4195 get_token endpoint and delayed leaves are delegated through delegate_delayed_leave, both via the js-sdk's unstable client methods. Transports with only a livekit_service_url keep using the OpenID flow against the JWT service. A transport advertising both shapes lets a subscriber fall back to the OpenID flow when the homeserver answers get_token with a 404 (Synapse reports M_UNRECOGNIZED for unknown routes rather than M_NOT_FOUND), so a client can still join calls hosted on servers that have not deployed MSC4195 yet. Publishers never fall back: a homeserver that advertises a `url` but does not implement the endpoints is misconfigured and the admin should hear about it, so they get NoMatrix2AuthorizationService. Delegation uses the same flow as the token so that the participant identity the service computes matches. The local member only probes a transport's delegate_delayed_leave endpoint when it has a service URL. The developer settings tab validates a custom service URL through the same facade. --- src/livekit/auth/csApi.ts | 124 +++++++++ src/livekit/auth/index.test.ts | 252 ++++++++++++++++++ src/livekit/auth/index.ts | 201 +++++++++++++- src/settings/DeveloperSettingsTab.test.tsx | 6 +- src/settings/DeveloperSettingsTab.tsx | 6 +- .../CallViewModel/CallViewModelTestUtils.ts | 2 + .../localMember/LocalMember.test.ts | 35 ++- .../CallViewModel/localMember/LocalMember.ts | 22 +- .../remoteMembers/Connection.test.ts | 94 ++++--- .../CallViewModel/remoteMembers/Connection.ts | 10 +- .../remoteMembers/ConnectionFactory.ts | 7 +- .../remoteMembers/ECConnectionFactory.test.ts | 9 +- .../remoteMembers/integration.test.ts | 14 +- src/utils/test-fixtures.ts | 31 ++- 14 files changed, 709 insertions(+), 104 deletions(-) create mode 100644 src/livekit/auth/csApi.ts create mode 100644 src/livekit/auth/index.test.ts diff --git a/src/livekit/auth/csApi.ts b/src/livekit/auth/csApi.ts new file mode 100644 index 000000000..63b9d0cb9 --- /dev/null +++ b/src/livekit/auth/csApi.ts @@ -0,0 +1,124 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { type MatrixClient } from "matrix-js-sdk"; +import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; + +import { extractFullConfigFromToken, type SFUConfig } from "./types.ts"; +import { doNetworkOperationWithRetry } from "../../utils/matrix.ts"; + +export type ClientGetTokenParts = Pick< + MatrixClient, + "_unstable_getLivekitToken" +>; + +export interface GetSFUConfigParams { + /** + * The Matrix client. + */ + client: ClientGetTokenParts; + /** + * Data identifying the local user's session membership. + */ + membership: CallMembershipIdentityParts; + /** + * The WebSocket URL of the SFU for which we wish to get an access token. + */ + url: string; + /** + * The name of the homeserver to which the SFU belongs. + */ + serverName: string; + /** + * The ID of the Matrix room in which the session takes place. + */ + roomId: string; + /** + * The ID of the MatrixRTC slot in which the session takes place. + */ + slotId: string; +} + +/** + * Gets an {@link SFUConfig} appropriate for connecting to a given SFU, using + * the MSC4195 LiveKit client-server API endpoints. + */ +export async function getSFUConfig({ + client, + membership, + url, + serverName, + roomId, + slotId, +}: GetSFUConfigParams): Promise { + const res = await doNetworkOperationWithRetry(async () => + client._unstable_getLivekitToken({ + url, + server_name: serverName, + room_id: roomId, + slot_id: slotId, + member_id: membership.memberId, + }), + ); + return extractFullConfigFromToken({ url, jwt: res.jwt }); +} + +export type ClientDelegationParts = Pick< + MatrixClient, + "_unstable_delegateDelayedLeave" | "baseUrl" +>; + +export interface DelegateDelayedLeaveParams { + /** + * The Matrix client. + */ + client: ClientDelegationParts; + /** + * Data identifying the local user's session membership. + */ + membership: CallMembershipIdentityParts; + /** + * The WebSocket URL of the SFU to which we are publishing and wish to + * delegate the delayed leave event. + */ + url: string; + /** + * The ID of the Matrix room in which the session takes place. + */ + roomId: string; + /** + * The ID of the MatrixRTC slot in which the session takes place. + */ + slotId: string; + /** + * The delay ID of the leave event to be delegated. + */ + delayId: string; +} + +/** + * Delegates a delayed leave event to a given SFU, so that the event will + * be sent on our behalf whenever we disconnect from the SFU. + */ +export async function delegateDelayedLeave({ + client, + membership, + url, + roomId, + slotId, + delayId, +}: DelegateDelayedLeaveParams): Promise { + await doNetworkOperationWithRetry(async () => + client._unstable_delegateDelayedLeave({ + url, + room_id: roomId, + slot_id: slotId, + member_id: membership.memberId, + delay_id: delayId, + }), + ); +} diff --git a/src/livekit/auth/index.test.ts b/src/livekit/auth/index.test.ts new file mode 100644 index 000000000..015e0b036 --- /dev/null +++ b/src/livekit/auth/index.test.ts @@ -0,0 +1,252 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { afterEach, expect, type Mocked, vi, describe, it } from "vitest"; +import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; +import { logger } from "matrix-js-sdk/lib/logger"; + +import { + aliceDeviceId, + aliceId, + aliceUserId, + createJWTToken, +} from "../../utils/test-fixtures"; +import type * as csApi from "./csApi"; +import * as openID from "./openID"; +import { delegateDelayedLeave, getSFUConfig } from "."; +import { mocked } from "storybook/test"; +import { ConnectionError, MatrixError } from "matrix-js-sdk"; +import { NoMatrix2AuthorizationService } from "../../utils/errors"; +import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; + +vi.mock("./openID", () => ({ getSFUConfig: vi.fn() })); +vi.useFakeTimers(); + +const csApiJWTToken = createJWTToken(aliceId, "via CS API"); +const openIDJWTToken = createJWTToken(aliceId, "via OpenID"); + +mocked(openID).getSFUConfig.mockResolvedValue({ + url: "mock-openID-url", + jwt: openIDJWTToken, + livekitAlias: "via OpenID", + livekitIdentity: aliceId, +}); + +const client: Mocked< + csApi.ClientGetTokenParts & + csApi.ClientDelegationParts & + openID.ClientOpenIDParts +> = { + baseUrl: "https://matrix.example.org", + _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: csApiJWTToken }), + _unstable_delegateDelayedLeave: vi.fn().mockResolvedValue({}), + getOpenIdToken: vi + .fn() + .mockResolvedValue({ access_token: "mock-openID-token" }), +}; + +const membership: CallMembershipIdentityParts = { + userId: aliceUserId, + deviceId: aliceDeviceId, + memberId: "123", +}; + +const roomId = "!room:example.org"; +const slotId = "m.call#ROOM"; + +const unsupportedError = new MatrixError({ errcode: "M_UNSUPPORTED" }, 404); + +const urlTransport: UnstableLivekitTransport = { + type: "livekit", + url: "wss://livekit.example.org", +}; + +const serviceUrlTransport: UnstableLivekitTransport = { + type: "livekit", + livekit_service_url: "https://livekit-jwt.example.org", +}; + +const urlAndServiceUrlTransport: UnstableLivekitTransport = { + ...urlTransport, + ...serviceUrlTransport, +}; + +afterEach(() => vi.clearAllMocks()); + +describe("getSFUConfig", () => { + const defaultParams = { + client, + membership, + serverName: "example.org", + role: "subscriber" as const, + roomId, + slotId, + logger, + }; + + it("uses CS API with url transport", async () => { + expect( + await getSFUConfig({ ...defaultParams, transport: urlTransport }), + ).toEqual( + expect.objectContaining({ + jwt: csApiJWTToken, + livekitAlias: "via CS API", + }), + ); + expect(client._unstable_getLivekitToken).toHaveBeenCalled(); + expect(openID.getSFUConfig).not.toHaveBeenCalled(); + }); + + it("does not fall back to OpenID with url transport", async () => { + client._unstable_getLivekitToken.mockRejectedValueOnce(unsupportedError); + await expect( + getSFUConfig({ ...defaultParams, transport: urlTransport }), + ).rejects.toThrow(expect.any(NoMatrix2AuthorizationService)); + expect(client._unstable_getLivekitToken).toHaveBeenCalled(); + expect(openID.getSFUConfig).not.toHaveBeenCalled(); + }); + + it("uses OpenID with livekit_service_url transport", async () => { + expect( + await getSFUConfig({ ...defaultParams, transport: serviceUrlTransport }), + ).toEqual( + expect.objectContaining({ + jwt: openIDJWTToken, + livekitAlias: "via OpenID", + }), + ); + expect(client._unstable_getLivekitToken).not.toHaveBeenCalled(); + expect(openID.getSFUConfig).toHaveBeenCalled(); + }); + + it("uses CS API with url + livekit_service_url transport", async () => { + expect( + await getSFUConfig({ + ...defaultParams, + transport: urlAndServiceUrlTransport, + }), + ).toEqual( + expect.objectContaining({ + jwt: csApiJWTToken, + livekitAlias: "via CS API", + }), + ); + expect(client._unstable_getLivekitToken).toHaveBeenCalled(); + expect(openID.getSFUConfig).not.toHaveBeenCalled(); + }); + + it("falls back to OpenID with url + livekit_service_url transport", async () => { + client._unstable_getLivekitToken.mockRejectedValueOnce(unsupportedError); + expect( + await getSFUConfig({ + ...defaultParams, + transport: urlAndServiceUrlTransport, + }), + ).toEqual( + expect.objectContaining({ + jwt: openIDJWTToken, + livekitAlias: "via OpenID", + }), + ); + expect(client._unstable_getLivekitToken).toHaveBeenCalled(); + expect(openID.getSFUConfig).toHaveBeenCalled(); + }); + + it("does not fall back to OpenID when publishing", async () => { + client._unstable_getLivekitToken.mockRejectedValueOnce(unsupportedError); + await expect( + getSFUConfig({ + ...defaultParams, + transport: urlAndServiceUrlTransport, + role: "publisher", + }), + ).rejects.toThrow(expect.any(NoMatrix2AuthorizationService)); + expect(client._unstable_getLivekitToken).toHaveBeenCalled(); + expect(openID.getSFUConfig).not.toHaveBeenCalled(); + }); + + it("does not fall back to OpenID on connection error", async () => { + const client_ = { + ...client, + _unstable_getLivekitToken: vi + .fn() + .mockRejectedValue(new ConnectionError("No connection")), + }; + await Promise.all([ + expect( + getSFUConfig({ + ...defaultParams, + client: client_, + transport: urlAndServiceUrlTransport, + }), + ).rejects.toThrow(expect.any(ConnectionError)), + vi.runAllTimersAsync(), // Allow request to be retried + ]); + expect(client_._unstable_getLivekitToken).toHaveBeenCalled(); + expect(openID.getSFUConfig).not.toHaveBeenCalled(); + }); +}); + +describe("delegateDelayedLeave", () => { + const defaultParams = { + client, + membership, + roomId, + slotId, + delayId: "delay123", + logger, + }; + + it("uses CS API with url transport", async () => { + await delegateDelayedLeave({ ...defaultParams, transport: urlTransport }); + expect(client._unstable_delegateDelayedLeave).toHaveBeenCalled(); + expect(openID.getSFUConfig).not.toHaveBeenCalled(); + }); + + it("does not fall back to OpenID with url transport", async () => { + client._unstable_delegateDelayedLeave.mockRejectedValueOnce( + unsupportedError, + ); + await expect( + delegateDelayedLeave({ ...defaultParams, transport: urlTransport }), + ).rejects.toThrow(); + expect(client._unstable_delegateDelayedLeave).toHaveBeenCalled(); + expect(openID.getSFUConfig).not.toHaveBeenCalled(); + }); + + it("uses OpenID with livekit_service_url transport", async () => { + await delegateDelayedLeave({ + ...defaultParams, + transport: serviceUrlTransport, + }); + expect(client._unstable_delegateDelayedLeave).not.toHaveBeenCalled(); + expect(openID.getSFUConfig).toHaveBeenCalled(); + }); + + it("uses CS API with url + livekit_service_url transport", async () => { + await delegateDelayedLeave({ + ...defaultParams, + transport: urlAndServiceUrlTransport, + }); + expect(client._unstable_delegateDelayedLeave).toHaveBeenCalled(); + expect(openID.getSFUConfig).not.toHaveBeenCalled(); + }); + + it("does not fall back to OpenID with url + livekit_service_url transport", async () => { + client._unstable_delegateDelayedLeave.mockRejectedValueOnce( + unsupportedError, + ); + await expect( + delegateDelayedLeave({ + ...defaultParams, + transport: urlAndServiceUrlTransport, + }), + ).rejects.toThrow(); + expect(client._unstable_delegateDelayedLeave).toHaveBeenCalled(); + expect(openID.getSFUConfig).not.toHaveBeenCalled(); + }); +}); diff --git a/src/livekit/auth/index.ts b/src/livekit/auth/index.ts index 6c11768f7..c535e72f5 100644 --- a/src/livekit/auth/index.ts +++ b/src/livekit/auth/index.ts @@ -5,5 +5,204 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial Please see LICENSE in the repository root for full details. */ -export * from "./openID"; +import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; +import { type Logger } from "matrix-js-sdk/lib/logger"; + +import { MatrixError } from "matrix-js-sdk"; +import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; +import * as openID from "./openID"; +import * as csApi from "./csApi"; +import { type SFUConfig } from "./types"; +import { NoMatrix2AuthorizationService } from "../../utils/errors"; + +export type { ClientOpenIDParts } from "./openID"; +export type { ClientGetTokenParts, ClientDelegationParts } from "./csApi"; export type * from "./types"; + +/** + * Checks whether an error means "this homeserver does not implement the + * (unstable) MSC4195 endpoints". + * + * MSC4195 documents `M_NOT_FOUND` for this, but a homeserver without support + * answers with whatever it uses for unknown endpoints (Synapse: 404 with + * `M_UNRECOGNIZED`), so we accept any 404 as "unsupported". + * @param e The error to check. + */ +function isEndpointUnsupported(e: unknown): boolean { + return ( + e instanceof MatrixError && + (e.httpStatus === 404 || + e.errcode === "M_NOT_FOUND" || + e.errcode === "M_UNRECOGNIZED") + ); +} + +export interface GetSFUConfigParams { + /** + * The Matrix client. + */ + client: csApi.ClientGetTokenParts & openID.ClientOpenIDParts; + /** + * Data identifying the local user's session membership. + */ + membership: CallMembershipIdentityParts; + /** + * The transport for which we wish to get an access token. + */ + transport: UnstableLivekitTransport; + /** + * The name of the homeserver to which the {@link transport} belongs. + */ + serverName: string; + /** + * The ID of the Matrix room in which the session takes place. + */ + roomId: string; + /** + * The ID of the MatrixRTC slot in which the session takes place. + */ + slotId: string; + /** + * Whether we want to publish or only subscribe on the {@link transport}. + */ + role: "publisher" | "subscriber"; + logger: Logger; +} + +/** + * Gets an {@link SFUConfig} appropriate for connecting to a given transport. + */ +export async function getSFUConfig({ + client, + membership, + transport, + serverName, + roomId, + slotId, + role, + logger, +}: GetSFUConfigParams): Promise { + if ("url" in transport) { + try { + return await csApi.getSFUConfig({ + client, + membership, + url: transport.url, + serverName, + roomId, + slotId, + }); + } catch (e) { + if (isEndpointUnsupported(e)) { + // Publishers should never fall back. If a homeserver advertises a `url` + // but doesn't support MSC4195 itself, that's a misconfiguration which + // the admin should know about so they can prepare for the future + // deprecation of `livekit_service_url`. + const mayFallBack = + role === "subscriber" && "livekit_service_url" in transport; + if (mayFallBack) { + logger.warn( + `Homeserver does not support the MSC4195 get_token endpoint, falling back to the OpenID flow with service ${transport.livekit_service_url}`, + ); + // Fall through + } else { + logger.error( + `Homeserver does not support the MSC4195 get_token endpoint, cannot get token for transport ${transport.url}`, + ); + throw new NoMatrix2AuthorizationService(e as Error); + } + } else { + throw e; + } + } + } + + return await openID.getSFUConfig({ + client, + membership, + serviceUrl: transport.livekit_service_url, + roomId, + slotId, + role, + logger, + }); +} + +export interface DelegateDelayedLeaveParams { + /** + * The Matrix client. + */ + client: csApi.ClientDelegationParts & openID.ClientOpenIDParts; + /** + * Data identifying the local user's session membership. + */ + membership: CallMembershipIdentityParts; + /** + * The transport to which we are publishing and wish to delegate the delayed + * leave event. + */ + transport: UnstableLivekitTransport; + /** + * The ID of the Matrix room in which the session takes place. + */ + roomId: string; + /** + * The ID of the MatrixRTC slot in which the session takes place. + */ + slotId: string; + /** + * The delay ID of the leave event to be delegated. + */ + delayId: string; + logger: Logger; +} + +/** + * Delegates a delayed leave event to a given transport, so that the event will + * be sent on our behalf whenever we disconnect from the transport's SFU. + */ +export async function delegateDelayedLeave({ + client, + membership, + transport, + roomId, + slotId, + delayId, + logger, +}: DelegateDelayedLeaveParams): Promise { + // It's important that we use the same flow (CS API or OpenID) as whatever we + // would've used to get a JWT token for this transport, since the participant + // identity computed by the service needs to match. + if ("url" in transport) { + try { + await csApi.delegateDelayedLeave({ + client, + membership, + url: transport.url, + roomId, + slotId, + delayId, + }); + } catch (e) { + if (isEndpointUnsupported(e)) + logger.error( + "Server misconfigured: transport contains a 'url' field but server does not support the MSC4195 delegate_delayed_leave endpoint", + ); + throw e; + } + } else { + // This will technically cause the service to issue a new JWT token, but + // it's safe to discard. We're only interested in triggering delegation. + await openID.getSFUConfig({ + client, + membership, + serviceUrl: transport.livekit_service_url, + roomId, + slotId, + role: "publisher", + delayEndpointBaseUrl: client.baseUrl, + delayId, + logger, + }); + } +} diff --git a/src/settings/DeveloperSettingsTab.test.tsx b/src/settings/DeveloperSettingsTab.test.tsx index bb0808761..034249ced 100644 --- a/src/settings/DeveloperSettingsTab.test.tsx +++ b/src/settings/DeveloperSettingsTab.test.tsx @@ -133,6 +133,7 @@ describe("DeveloperSettingsTab", () => { getCrypto: () => ({ getVersion: (): string => "x" }), getUserId: () => "@u:hs", getDeviceId: () => "DEVICE", + getDomain: () => "hs", } as unknown as MatrixClient; it("will not update custom livekit url without roomId", async () => { const user = userEvent.setup(); @@ -228,7 +229,10 @@ describe("DeveloperSettingsTab", () => { await user.click(saveButton); expect(getSFUConfig).toHaveBeenCalledWith( expect.objectContaining({ - serviceUrl: "wss://example.livekit.valid", + transport: { + type: "livekit", + livekit_service_url: "wss://example.livekit.valid", + }, roomId: "#testRoom", }), ); diff --git a/src/settings/DeveloperSettingsTab.tsx b/src/settings/DeveloperSettingsTab.tsx index c8cc2a524..d7a9164c5 100644 --- a/src/settings/DeveloperSettingsTab.tsx +++ b/src/settings/DeveloperSettingsTab.tsx @@ -521,7 +521,11 @@ export const DeveloperSettingsTab: FC = ({ await getSFUConfig({ client, membership: { userId, deviceId, memberId: "" }, - serviceUrl: customLivekitUrlTextBuffer, + transport: { + type: "livekit", + livekit_service_url: customLivekitUrlTextBuffer, + }, + serverName: client.getDomain()!, roomId, slotId: "org.example.fake-slot#room", role: "publisher", diff --git a/src/state/CallViewModel/CallViewModelTestUtils.ts b/src/state/CallViewModel/CallViewModelTestUtils.ts index 809dc75a5..93c190ace 100644 --- a/src/state/CallViewModel/CallViewModelTestUtils.ts +++ b/src/state/CallViewModel/CallViewModelTestUtils.ts @@ -122,6 +122,8 @@ export function withCallViewModel(mode: MatrixRTCMode) { }; const room = mockMatrixRoom({ client: new (class extends EventEmitter { + public readonly baseUrl = "https://matrix.example.com"; + public getUserId(): string | undefined { return localRtcMember.userId; } diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index f1c95d1cb..2d721f7d6 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/src/state/CallViewModel/localMember/LocalMember.test.ts @@ -116,6 +116,7 @@ const delegatedTimings: ResolvedDelayedLeaveTimings = { }; const mockedClient = { + baseUrl: "https://matrix.example.org", getDomain: vi.fn().mockReturnValue("example.org"), getDeviceId: vi.fn().mockReturnValue("AAAA"), getOpenIdToken: vi.fn().mockResolvedValue({ @@ -124,6 +125,7 @@ const mockedClient = { matrix_server_name: "localhost", expires_in: 10000, }), + _unstable_delegateDelayedLeave: vi.fn().mockResolvedValue({}), }; describe("enterRTCSession", () => { @@ -239,6 +241,7 @@ describe("LocalMembership", () => { ownMembershipIdentity: ownMemberMock, client: mockedClient, delayId$: constant(null), + matrixRTCMode: MATRIX_RTC_MODE, }; beforeEach(() => { @@ -304,9 +307,11 @@ describe("LocalMembership", () => { expect(internalLogger).toHaveBeenCalledWith(expextedLog); }); - const mockTransport = { - livekit_service_url: "a", - } as UnstableLivekitTransport; + const mockTransport: UnstableLivekitTransport = { + type: "livekit", + url: "https://sfu.example.org", + livekit_service_url: "https://jwt.example.org", + }; const mockTransportLocator: livekitAuth.TransportLocator = { transport: mockTransport, @@ -336,10 +341,10 @@ describe("LocalMembership", () => { livekitRoom: mockLivekitRoom({}), } as unknown as Connection; - const authCallSpy = vi - .spyOn(livekitAuth, "getSFUConfig") - .mockImplementation(() => mockedClient.getOpenIdToken()); - afterEach(() => authCallSpy.mockClear()); + const delegationSpy = vi + .spyOn(livekitAuth, "delegateDelayedLeave") + .mockResolvedValue(); + afterEach(() => delegationSpy.mockClear()); it.each([ ["no", null, timings], @@ -348,7 +353,11 @@ describe("LocalMembership", () => { "https://matrix.example.org/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave", delegatedTimings, ], - ["transport", "/a/delegate_delayed_leave", delegatedTimings], + [ + "transport", + "https://jwt.example.org/delegate_delayed_leave", + delegatedTimings, + ], ])( "joins session with %s delegation support", async (_serviceName, delegationUrl, delayedLeaveTimings) => { @@ -379,22 +388,20 @@ describe("LocalMembership", () => { delayedLeaveTimings, ); - expect(authCallSpy).not.toHaveBeenCalled(); + expect(delegationSpy).not.toHaveBeenCalled(); delayId$.next("leave1"); await flushPromises(); if (delegationUrl === null) { - expect(authCallSpy).not.toHaveBeenCalled(); + expect(delegationSpy).not.toHaveBeenCalled(); } else { // Delegation is supported in this test case, so go on to check that // LocalMember actually performs delegation const expectDelegation = (delayId: string) => - expect(authCallSpy).toHaveBeenLastCalledWith({ + expect(delegationSpy).toHaveBeenLastCalledWith({ client: mockedClient, membership: ownMemberMock, - serviceUrl: "a", + transport: mockTransport, roomId: "!test-room-id:example.org", - role: "publisher", - delayEndpointBaseUrl: "https://matrix.example.org", delayId, logger: expect.anything(), }); diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index 14f853403..244c728dc 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -79,8 +79,9 @@ import { } from "../remoteMembers/Connection.ts"; import { type HomeserverConnected } from "./HomeserverConnected.ts"; import { + type ClientDelegationParts, + delegateDelayedLeave, type ClientOpenIDParts, - getSFUConfig, type TransportLocator, } from "../../../livekit/auth"; @@ -152,7 +153,7 @@ interface Props { roomId: string; ownMembershipIdentity: CallMembershipIdentityParts; localTransport$: Observable; - client: ClientOpenIDParts; + client: ClientDelegationParts & ClientOpenIDParts; matrixRTCSession: Pick< MatrixRTCSession, "slotId" | "updateCallIntent" | "leave" @@ -720,22 +721,19 @@ export const createLocalMembership$ = ({ async ([joinParams, delayId]) => { if (joinParams?.delegationSupported && delayId !== null) { try { - if (!("livekit_service_url" in joinParams.transport)) - throw new Error("Transport has no livekit_service_url"); - // This will technically cause the service to issue a new JWT token, - // but it's safe to discard. We're only interested in triggering - // delegation. - await getSFUConfig({ + logger.info( + `Delegating delayed leave to ${JSON.stringify(joinParams.transport)}…`, + ); + await delegateDelayedLeave({ client, membership: ownMembershipIdentity, - serviceUrl: joinParams.transport.livekit_service_url, - roomId, + transport: joinParams.transport, slotId: matrixRTCSession.slotId, - role: "publisher", - delayEndpointBaseUrl: baseUrl, + roomId, delayId, logger, }); + logger.info("Delayed leave successfully delegated"); } catch (e) { // TODO: Surface this to the user as a service interruption? logger.error(`Failed to delegate leave`, e); diff --git a/src/state/CallViewModel/remoteMembers/Connection.test.ts b/src/state/CallViewModel/remoteMembers/Connection.test.ts index f0fc6bfd5..bdab8e7af 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.test.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.test.ts @@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details. import { afterEach, + beforeEach, describe, expect, it, @@ -27,9 +28,12 @@ import { } from "livekit-client"; import fetchMock from "fetch-mock"; import EventEmitter from "events"; -import { type IOpenIDToken } from "matrix-js-sdk"; +import { MatrixError } from "matrix-js-sdk"; import { logger, type Logger } from "matrix-js-sdk/lib/logger"; -import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; +import { + type LivekitGetTokenResponse, + type UnstableLivekitTransport, +} from "matrix-js-sdk/lib/matrixrtc"; import { Connection, @@ -40,11 +44,14 @@ import { ObservableScope } from "../../ObservableScope.ts"; import { FailToGetOpenIdToken } from "../../../utils/errors.ts"; import { testJWTToken } from "../../../utils/test-fixtures.ts"; import { mockRemoteParticipant, ownMemberMock } from "../../../utils/test.ts"; -import { type ClientOpenIDParts } from "../../../livekit/auth"; +import { + type ClientGetTokenParts, + type ClientOpenIDParts, +} from "../../../livekit/auth"; let testScope: ObservableScope; -let client: MockedObject; +let client: MockedObject; let fakeLivekitRoom: MockedObject; @@ -54,21 +61,21 @@ let fakeLocalParticipant: MockedObject; const ROOM_ID = "!roomID:example.org"; const SLOT_ID = "m.call#ROOM"; -const livekitServiceUrl = "https://matrix-rtc.example.org/livekit/jwt"; -const livekitFocus: UnstableLivekitTransport = { - livekit_service_url: livekitServiceUrl, +const transport: UnstableLivekitTransport = { type: "livekit", + url: "wss://matrix-rtc.m.localhost/livekit/sfu", }; function setupTest(): void { testScope = new ObservableScope(); - client = vi.mocked({ + client = vi.mocked({ getOpenIdToken: vi.fn().mockResolvedValue({ access_token: "rYsmGUEwNjKgJYyeNUkZseJN", token_type: "Bearer", matrix_server_name: "example.org", expires_in: 3600, }), + _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }), }); localParticipantEventEmiter = new EventEmitter(); @@ -117,23 +124,13 @@ function setupRemoteConnection(): Connection { roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: livekitFocus, + transport, serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, livekitRoomFactory: () => fakeLivekitRoom, }; - fetchMock.post(`${livekitServiceUrl}/get_token`, () => { - return { - status: 200, - body: { - url: "wss://matrix-rtc.m.localhost/livekit/sfu", - jwt: testJWTToken, - }, - }; - }); - fakeLivekitRoom.connect.mockImplementation(async (): Promise => { const changeEv = RoomEvent.ConnectionStateChanged; @@ -155,6 +152,19 @@ afterEach(() => { }); describe("Start connection states", () => { + beforeEach(() => { + fetchMock.post( + `https://matrix-rtc.example.org/livekit/jwt/get_token`, + () => { + return { + // Return a non-retryable error, if not, the retry logic will + // wait and fail the test with a timeout. + status: 404, + }; + }, + ); + }); + it("start in initialized state", () => { setupTest(); @@ -163,7 +173,7 @@ describe("Start connection states", () => { roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: livekitFocus, + transport, serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, @@ -174,7 +184,7 @@ describe("Start connection states", () => { expect(connection.state$.getValue()).toEqual({ state: "initialized" }); }); - it("fail to getOpenId token then error state", async () => { + it("fail to get SFU token then error state", async () => { setupTest(); vi.useFakeTimers(); @@ -183,7 +193,7 @@ describe("Start connection states", () => { roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: livekitFocus, + transport, serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, @@ -198,10 +208,10 @@ describe("Start connection states", () => { }); onTestFinished(() => s.unsubscribe()); - const deferred = Promise.withResolvers(); + const deferred = Promise.withResolvers(); - client.getOpenIdToken.mockImplementation( - async (): Promise => { + client._unstable_getLivekitToken.mockImplementation( + async (): Promise => { return await deferred.promise; }, ); @@ -221,7 +231,7 @@ describe("Start connection states", () => { capturedState = capturedStates.pop(); if (capturedState?.state === "error") { expect(capturedState.error.message).toEqual("Something went wrong"); - expect(connection.transport).toEqual(livekitFocus); + expect(connection.transport).toEqual(transport); } else { expect.fail( "Expected FailedToStart state but got " + JSON.stringify(capturedState), @@ -238,7 +248,7 @@ describe("Start connection states", () => { roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: livekitFocus, + transport, serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, @@ -254,16 +264,15 @@ describe("Start connection states", () => { onTestFinished(() => s.unsubscribe()); const deferredSFU = Promise.withResolvers(); - // mock the /get_token call - fetchMock.post(`${livekitServiceUrl}/get_token`, async () => { + client._unstable_getLivekitToken.mockImplementation(async () => { await deferredSFU.promise; - return { - status: 500, - body: { + throw new MatrixError( + { errcode: "M_LOOKUP_FAILED", error: "Failed to look up user info from homeserver", }, - }; + 500, + ); }); connection.start().catch(() => { @@ -286,7 +295,7 @@ describe("Start connection states", () => { expect(capturedState.error.cause.message).toContain( "Failed to look up user info from homeserver", ); - expect(connection.transport).toEqual(livekitFocus); + expect(connection.transport).toEqual(transport); } else { expect.fail( "Expected FailedToStart state but got " + JSON.stringify(capturedState), @@ -303,7 +312,7 @@ describe("Start connection states", () => { roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: livekitFocus, + transport, serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, @@ -319,17 +328,6 @@ describe("Start connection states", () => { onTestFinished(() => s.unsubscribe()); const deferredSFU = Promise.withResolvers(); - // mock the /get_token call - fetchMock.post(`${livekitServiceUrl}/get_token`, () => { - return { - status: 200, - body: { - url: "wss://matrix-rtc.m.localhost/livekit/sfu", - jwt: testJWTToken, - }, - }; - }); - fakeLivekitRoom.connect.mockImplementation(async () => { await deferredSFU.promise; throw new Error("Failed to connect to livekit"); @@ -356,7 +354,7 @@ describe("Start connection states", () => { expect(capturedState.error.cause.message).toContain( "Failed to connect to livekit", ); - expect(connection.transport).toEqual(livekitFocus); + expect(connection.transport).toEqual(transport); } else { expect.fail( "Expected FailedToStart state but got " + JSON.stringify(capturedState), @@ -564,7 +562,7 @@ describe("remote track logging", () => { roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: livekitFocus, + transport, serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, diff --git a/src/state/CallViewModel/remoteMembers/Connection.ts b/src/state/CallViewModel/remoteMembers/Connection.ts index c0fe4fd9a..7e94b39f1 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.ts @@ -39,6 +39,7 @@ import { UnknownCallError, } from "../../../utils/errors.ts"; import { + type ClientGetTokenParts, getSFUConfig, type ClientOpenIDParts, type SFUConfig, @@ -54,7 +55,7 @@ export interface ConnectionOpts { /** The name of the homeserver to which the {@link transport} belongs. */ serverName: string; /** The Matrix client to use for OpenID and SFU config requests. */ - client: ClientOpenIDParts; + client: ClientGetTokenParts & ClientOpenIDParts; /** The ID of the Matrix room in which the session takes place. */ roomId: string; /** The ID of the MatrixRTC slot in which the session takes place. */ @@ -134,7 +135,7 @@ export class Connection { protected stopped = false; // TODO: can we just keep the ConnectionOpts object instead of spreading? - private readonly client: ClientOpenIDParts; + private readonly client: ClientGetTokenParts & ClientOpenIDParts; private readonly roomId: string; private readonly slotId: string; private readonly logger: Logger; @@ -385,13 +386,12 @@ export class Connection { } protected async getSFUConfig(): Promise { - if (!("livekit_service_url" in this.transport)) - throw new Error("Transport has no livekit_service_url"); return await getSFUConfig({ client: this.client, membership: this.ownMembershipIdentity, role: this.role, - serviceUrl: this.transport.livekit_service_url, + transport: this.transport, + serverName: this.serverName, roomId: this.roomId, slotId: this.slotId, logger: this.logger, diff --git a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts b/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts index 86cc01b51..e5240f096 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts @@ -35,7 +35,10 @@ import { noiseSuppressionSetting, autoGainControlSetting, } from "../../../settings/settings.ts"; -import { type ClientOpenIDParts } from "../../../livekit/auth"; +import { + type ClientGetTokenParts, + type ClientOpenIDParts, +} from "../../../livekit/auth"; // TODO evaluate if this should be done like the Publisher Factory export interface ConnectionFactory { @@ -65,7 +68,7 @@ export class ECConnectionFactory implements ConnectionFactory { * @param livekitRoomFactory - Optional factory function (for testing) to create LivekitRoom instances. If not provided, a default factory is used. */ public constructor( - private client: ClientOpenIDParts, + private client: ClientGetTokenParts & ClientOpenIDParts, private readonly roomId: string, private readonly slotId: string, private devices: MediaDevices, diff --git a/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts b/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts index 5d9abd0fe..be6c149f4 100644 --- a/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts +++ b/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts @@ -14,7 +14,10 @@ import EventEmitter from "events"; import { ObservableScope } from "../../ObservableScope.ts"; import { ECConnectionFactory } from "./ConnectionFactory.ts"; -import { type ClientOpenIDParts } from "../../../livekit/auth"; +import { + type ClientGetTokenParts, + type ClientOpenIDParts, +} from "../../../livekit/auth"; import { exampleTransport, mockMediaDevices, @@ -32,6 +35,7 @@ import { cameraBitrate, cameraCodec, } from "../../../settings/settings.ts"; +import { testJWTToken } from "../../../utils/test-fixtures.ts"; // At the top of your test file, after imports vi.mock("livekit-client", async (importOriginal) => { @@ -51,12 +55,13 @@ vi.mock("livekit-client", async (importOriginal) => { }); let testScope: ObservableScope; -let mockClient: ClientOpenIDParts; +let mockClient: ClientGetTokenParts & ClientOpenIDParts; beforeEach(() => { testScope = new ObservableScope(); mockClient = { getOpenIdToken: vi.fn().mockReturnValue(""), + _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }), }; }); diff --git a/src/state/CallViewModel/remoteMembers/integration.test.ts b/src/state/CallViewModel/remoteMembers/integration.test.ts index 39737ae6a..bf90e9eb4 100644 --- a/src/state/CallViewModel/remoteMembers/integration.test.ts +++ b/src/state/CallViewModel/remoteMembers/integration.test.ts @@ -18,7 +18,10 @@ import { trackEpoch, } from "../../ObservableScope.ts"; import { ECConnectionFactory } from "./ConnectionFactory.ts"; -import { type ClientOpenIDParts } from "../../../livekit/auth"; +import { + type ClientGetTokenParts, + type ClientOpenIDParts, +} from "../../../livekit/auth"; import { exampleTransportLocator, mockMediaDevices, @@ -40,7 +43,7 @@ import { deepCompare } from "matrix-js-sdk/lib/utils"; let testScope: ObservableScope; let ecConnectionFactory: ECConnectionFactory; -let mockClient: ClientOpenIDParts; +let mockClient: ClientGetTokenParts & ClientOpenIDParts; let lkRoomFactory: () => LivekitRoom; const createdMockLivekitRooms: Map = new Map(); @@ -49,6 +52,7 @@ beforeEach(() => { testScope = new ObservableScope(); mockClient = { getOpenIdToken: vi.fn().mockReturnValue(""), + _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }), }; lkRoomFactory = vi.fn().mockImplementation(() => { @@ -82,7 +86,7 @@ beforeEach(() => { ); //TODO a bit annoying to have to do a http mock? - const sfuConfigResponse = (url: string): { status: number; body: object } => { + fetchMock.post(`path:/sfu/get`, (url) => { const domain = new URL(url).hostname; // Extract the domain from the URL return { status: 200, @@ -91,9 +95,7 @@ beforeEach(() => { jwt: testJWTToken, }, }; - }; - fetchMock.post(`path:/sfu/get`, sfuConfigResponse); - fetchMock.post(`path:/get_token`, sfuConfigResponse); + }); }); afterEach(() => { diff --git a/src/utils/test-fixtures.ts b/src/utils/test-fixtures.ts index dcdb9f9c9..dac0c6972 100644 --- a/src/utils/test-fixtures.ts +++ b/src/utils/test-fixtures.ts @@ -64,16 +64,23 @@ export const daveRTL = mockMatrixRoomMember(daveRTLRtcMember, { rawDisplayName: "\u202eevaD", }); -export const testJWTToken = [ - {}, // header - { - // payload - sub: "@me:example.org:ABCDEF", - video: { - room: "!example_room_id", +export function createJWTToken(sub: string, room: string): string { + return [ + {}, // header + { + // payload + sub, + video: { + room, + }, }, - }, - {}, // signature -] - .map((d) => global.btoa(JSON.stringify(d))) - .join("."); + {}, // signature + ] + .map((d) => global.btoa(JSON.stringify(d))) + .join("."); +} + +export const testJWTToken = createJWTToken( + "@me:example.org:ABCDEF", + "!example_room_id", +);