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", +);