Matrix/element-call-Github
1
0
Fork 0
mirror of https://github.com/vector-im/element-call.git synced 2026-10-16 23:45:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move SFU authentication into livekit/auth and choose the endpoint by role

Browse Source 
The OpenID flow moves to src/livekit/auth/openID.ts with the SFU config
types in a sibling file, takes an options object, and replaces the
MatrixRTC mode with the role we take on the transport plus the session's
slot ID instead of a hard-coded m.call#ROOM.

This changes behaviour: publishers on a livekit_service_url transport
always use the legacy /sfu/get endpoint, so that subscribers can predict
their user:device participant identity, and subscribers always use the
default /get_token endpoint. The MatrixRTC mode no longer influences which
endpoint is used.
This commit is contained in:
Robin
2026-10-12 10:17:39 +02:00
committed by Johannes Marbach
parent f260755f46
commit 17c3223883
21 changed files with 708 additions and 804 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/livekit/auth/index.ts
+9
View File
@@ -0,0 +1,9 @@
/*
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.
*/
export * from "./openID";
export type * from "./types";
src/livekit/auth/openID.test.ts
+200
View File
@@ -0,0 +1,200 @@
/*
Copyright 2025 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 {
beforeEach,
afterEach,
describe,
expect,
it,
type MockedObject,
vitest,
} from "vitest";
import fetchMock from "fetch-mock";
import { MatrixError } from "matrix-js-sdk";
import {
getSFUConfig,
type GetSFUConfigParams,
type ClientOpenIDParts,
} from "./openID";
import { testJWTToken } from "../../utils/test-fixtures";
import { ownMemberMock } from "../../utils/test";
import { FailToGetOpenIdToken } from "../../utils/errors";
import { logger } from "matrix-js-sdk/lib/logger";
const serviceUrl = "https://sfu.example.org";
const delayParams = {
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
};
describe("getSFUConfig", () => {
let matrixClient: MockedObject<ClientOpenIDParts>;
let params: GetSFUConfigParams;
beforeEach(() => {
fetchMock.catch(404);
matrixClient = { getOpenIdToken: vitest.fn() };
params = {
client: matrixClient,
membership: ownMemberMock,
serviceUrl: "https://sfu.example.org",
roomId: "!example_room_id",
slotId: "m.call#ROOM",
role: "publisher",
logger,
};
});
afterEach(async () => {
void (await fetchMock.flush());
vitest.clearAllMocks();
fetchMock.reset();
});
it("should handle fetching a token", async () => {
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 200,
body: { url: serviceUrl, jwt: testJWTToken },
};
});
const config = await getSFUConfig(params);
expect(config).toEqual({
jwt: testJWTToken,
url: serviceUrl,
livekitIdentity: "@me:example.org:ABCDEF",
livekitAlias: "!example_room_id",
});
});
it("should fail if the SFU errors", async () => {
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
});
try {
await getSFUConfig(params);
} catch (ex: unknown) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toBeInstanceOf(MatrixError);
const mxError = (ex as Error).cause as MatrixError;
expect(mxError.message).toEqual(
"MatrixError: [500] Failed to look up user info from homeserver",
);
return;
}
expect.fail("Expected test to throw;");
});
it("should retry without delay params if the JWT service legacy endpoint returns M_BAD_JSON 400", async () => {
let callCount = 0;
fetchMock.post(
"https://sfu.example.org/sfu/get",
(url, opts) => {
callCount++;
const body = JSON.parse(opts.body as string);
// First call: check if it has delay parts and return 400
if (callCount === 1) {
expect(body).toHaveProperty("delay_id", "mock_delay_id");
return {
status: 400,
body: { errcode: "M_BAD_JSON", error: "Unsupported parameters" },
};
}
// Second call: check if delay parts were stripped and return success
expect(body).not.toHaveProperty("delay_id");
expect(body).not.toHaveProperty("delay_timeout");
expect(body).not.toHaveProperty("delay_cs_api_url");
return {
status: 200,
body: { url: serviceUrl, jwt: testJWTToken },
};
},
{ overwriteRoutes: true },
);
// Note: Assuming getSFUConfig eventually calls getLiveKitJWT
const config = await getSFUConfig({ ...params, ...delayParams });
expect(config.jwt).toBe(testJWTToken);
expect(callCount).toBe(2);
});
it("should successfully send delay parameters to the JWT service legacy endpoint", async () => {
fetchMock.post(
"https://sfu.example.org/sfu/get",
(url, opts) => {
const body = JSON.parse(opts.body as string);
// Verify, that the request contains the expected delay parameters
if (
body.delay_id === "mock_delay_id" &&
body.delay_timeout === 3600000 &&
body.delay_cs_api_url === "https://matrix.homeserverserver.org"
) {
return {
status: 200,
body: { url: serviceUrl, jwt: testJWTToken },
};
}
return {
status: 400,
body: { error: "Missing expected delay params" },
};
},
{ overwriteRoutes: true },
);
const config = await getSFUConfig({ ...params, ...delayParams });
// Prüfe das Ergebnis
expect(config).toMatchObject({
jwt: testJWTToken,
url: serviceUrl,
});
});
it("should retry fetching the openid token", async () => {
let count = 0;
matrixClient.getOpenIdToken.mockImplementation(async () => {
count++;
if (count < 2) {
throw Error("Test failure");
}
return Promise.resolve({
token_type: "Bearer",
access_token: "foobar",
matrix_server_name: "example.org",
expires_in: 30,
});
});
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 200,
body: { url: serviceUrl, jwt: testJWTToken },
};
});
const config = await getSFUConfig({ ...params, ...delayParams });
expect(config).toEqual({
jwt: testJWTToken,
url: serviceUrl,
livekitIdentity: "@me:example.org:ABCDEF",
livekitAlias: "!example_room_id",
});
});
});
src/livekit/auth/openID.ts
+262
View File
@@ -0,0 +1,262 @@
/*
Copyright 2023, 2024 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 {
type IOpenIDToken,
type MatrixClient,
parseErrorResponse,
} from "matrix-js-sdk";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { FailToGetOpenIdToken } from "../../utils/errors";
import { doNetworkOperationWithRetry } from "../../utils/matrix";
import { Config } from "../../config/Config";
import { extractFullConfigFromToken, type SFUConfig } from "./types";
// The bits we need from MatrixClient
export type ClientOpenIDParts = Pick<MatrixClient, "getOpenIdToken">;
export interface GetSFUConfigParams {
client: ClientOpenIDParts;
/**
* Data identifying the local user's session membership.
*/
membership: CallMembershipIdentityParts;
/**
* The base URL of the LiveKit JWT service.
*/
serviceUrl: 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";
/**
* The base URL of the Matrix homeserver, for delayed leave delegation.
*/
delayEndpointBaseUrl?: string;
/**
* The delay ID of the leave event to be delegated to the JWT service.
*/
delayId?: string;
logger: Logger;
}
/**
* Gets an {@link SFUConfig} appropriate for connecting to the SFU behind a
* given service URL, using the legacy OpenID flow.
* @throws {@link FailToGetOpenIdToken}
*/
export async function getSFUConfig({
client,
membership,
serviceUrl,
roomId,
slotId,
role,
delayEndpointBaseUrl,
delayId,
logger,
}: GetSFUConfigParams): Promise<SFUConfig> {
let openIdToken: IOpenIDToken;
try {
openIdToken = await doNetworkOperationWithRetry(async () =>
client.getOpenIdToken(),
);
} catch (error) {
throw new FailToGetOpenIdToken(
error instanceof Error ? error : new Error("Unknown error"),
);
}
logger.debug("Got openID token", openIdToken);
let sfuConfig: { url: string; jwt: string } | undefined;
let endpoint: "default" | "legacy";
if (role === "publisher") {
// When publishing a legacy transport (one with a `livekit_service_url`),
// subscribers will expect our participant identity to use the legacy
// `@user_id:device_id` format. Only the legacy JWT service endpoint assigns
// identities in this format, so we must use it.
endpoint = "legacy";
} else {
// Use the default endpoint as it affords us slightly better metadata
// protection (pseudonymous participant identities).
endpoint = "default";
}
try {
logger.info(
`Trying to get JWT with ${endpoint} endpoint for focus ${serviceUrl}...`,
);
sfuConfig = await (
endpoint === "legacy" ? getLiveKitJWTLegacy : getLiveKitJWT
)(
membership,
serviceUrl,
roomId,
slotId,
openIdToken,
delayEndpointBaseUrl,
delayId,
);
logger.info(`Got JWT from call's active focus URL.`);
return extractFullConfigFromToken(sfuConfig);
} catch (e) {
logger.error(`Failed fetching jwt with ${endpoint} endpoint:`, e);
throw new FailToGetOpenIdToken(
e instanceof Error ? e : new Error(`Unknown error ${e}`),
);
}
}
/**
* Gets a JWT token appropriate for connecting to the SFU behind a given service
* URL, using the legacy `/sfu/get` endpoint, which assigns us a legacy
* (`<user_id>:<device_id>`) participant identity.
*/
async function getLiveKitJWTLegacy(
membership: CallMembershipIdentityParts,
livekitServiceURL: string,
matrixRoomId: string,
_slotId: string, // Unused - the legacy endpoint is from before slots existed.
openIDToken: IOpenIDToken,
delayEndpointBaseUrl?: string,
delayId?: string,
): Promise<{ url: string; jwt: string }> {
interface IDelayParams {
delay_id?: string;
delay_timeout?: number;
delay_cs_api_url?: string;
}
let bodyDalayParts: IDelayParams = {};
// Also check for empty string
if (delayId && delayEndpointBaseUrl) {
bodyDalayParts = {
delay_id: delayId,
delay_timeout:
Config.get().matrix_rtc_session.delegated_delayed_leave.delay_ms,
delay_cs_api_url: delayEndpointBaseUrl,
};
}
const makeRequest = async (delayParts: IDelayParams): Promise<Response> => {
return await fetch(livekitServiceURL + "/sfu/get", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({
// The legacy JWT endpoint uses only the matrix room id to calculate the livekit room alias.
// However, the livekit room alias is provided as part of the JWT payload.
room: matrixRoomId,
openid_token: openIDToken,
device_id: membership.deviceId,
...delayParts,
}),
});
};
const res = await doNetworkOperationWithRetry(async () => {
let response = await makeRequest(bodyDalayParts);
// Old service compatibility check
const oldServiceDoesNotSupportDelayParts =
response.status === 400 && Object.keys(bodyDalayParts).length > 0;
// If http status 400 with M_BAD_JSON and we sent delay parts, retry without them
if (oldServiceDoesNotSupportDelayParts) {
try {
const errorBody = await response.json();
if (errorBody.errcode === "M_BAD_JSON") {
response = await makeRequest({});
}
} catch {
// If we can't parse the error, treat as real error
}
}
return response;
});
if (!res.ok) {
throw parseErrorResponse(res, await res.text());
}
return await res.json();
}
class NotSupportedError extends Error {
public constructor(message: string) {
super(message);
this.name = "NotSupported";
}
}
/**
* Gets a JWT token appropriate for connecting to the SFU behind a given service
* URL, using the default `/get_token` endpoint, which assigns us a hashed
* (pseudonymised) participant identity.
*/
async function getLiveKitJWT(
membership: CallMembershipIdentityParts,
livekitServiceURL: string,
matrixRoomId: string,
slotId: string,
openIDToken: IOpenIDToken,
delayEndpointBaseUrl?: string,
delayId?: string,
): Promise<{ url: string; jwt: string }> {
const { userId, deviceId, memberId } = membership;
const body = {
room_id: matrixRoomId,
slot_id: slotId,
openid_token: openIDToken,
member: {
id: memberId,
claimed_user_id: userId,
claimed_device_id: deviceId,
},
};
let bodyDalayParts = {};
// Also check for empty string
if (delayId && delayEndpointBaseUrl) {
bodyDalayParts = {
delay_id: delayId,
delay_timeout:
Config.get().matrix_rtc_session.delegated_delayed_leave.delay_ms,
delay_cs_api_url: delayEndpointBaseUrl,
};
}
const res = await doNetworkOperationWithRetry(async () => {
return await fetch(livekitServiceURL + "/get_token", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ ...body, ...bodyDalayParts }),
});
});
if (!res.ok) {
const msg = "SFU Config fetch failed with status code " + res.status;
if (res.status === 404) {
throw new NotSupportedError(msg);
} else {
throw parseErrorResponse(res, await res.text());
}
}
return await res.json();
}
src/livekit/auth/types.ts
+79
View File
@@ -0,0 +1,79 @@
/*
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.
*/
/**
* Configuration and access tokens provided by the SFU on successful authentication.
*/
export interface SFUConfig {
/**
* The WebSocket URL of the LiveKit SFU. This is what we connect the LiveKit
* room to. Note that this is NOT the JWT service URL
* (`livekit_service_url`): the legacy JWT service returns this URL as part of
* its response, while the CS-Api based flow requires us to know it upfront.
*/
url: string;
jwt: string;
livekitAlias: string;
// NOTE: Currently unused.
livekitIdentity: string;
}
/**
* Decoded details from the JWT.
*/
interface SFUJWTPayload {
/**
* Expiration time for the JWT.
* Note: This value is in seconds since Unix epoch.
*/
exp: number;
/**
* Name of the instance which authored the JWT
*/
iss: string;
/**
* Time at which the JWT can start to be used.
* Note: This value is in seconds since Unix epoch.
*/
nbf: number;
/**
* Subject. The Livekit alias in this context.
*/
sub: string;
/**
* The set of permissions for the user.
*/
video: {
canPublish: boolean;
canSubscribe: boolean;
room: string;
roomJoin: boolean;
};
}
/**
* Complements the SFU websocket url and the JWT with the information encoded in
* the JWT payload itself.
* @param sfuConfig The SFU websocket url and the JWT to connect with.
* @returns The full SFU config, including the LiveKit alias and identity.
*/
export function extractFullConfigFromToken(sfuConfig: {
url: string;
jwt: string;
}): SFUConfig {
const [, payloadStr] = sfuConfig.jwt.split(".");
const payload = JSON.parse(global.atob(payloadStr)) as SFUJWTPayload;
return {
jwt: sfuConfig.jwt,
url: sfuConfig.url,
livekitAlias: payload.video.room,
// NOTE: Currently unused.
// Probably also not helpful since we now compute the backendIdentity on joining the call so we can use it for the encryption manager.
// The only reason for us to know it locally is to connect the right users with the lk world. (and to set our own keys)
livekitIdentity: payload.sub,
};
}
src/livekit/openIDSFU.test.ts
-329
View File
@@ -1,329 +0,0 @@
/*
Copyright 2025 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 {
beforeEach,
afterEach,
describe,
expect,
it,
type MockedObject,
vitest,
} from "vitest";
import fetchMock from "fetch-mock";
import { MatrixError } from "matrix-js-sdk";
import { getSFUConfigWithOpenID, type OpenIDClientParts } from "./openIDSFU";
import { testJWTToken } from "../utils/test-fixtures";
import { ownMemberMock } from "../utils/test";
import { FailToGetOpenIdToken } from "../utils/errors";
const sfuUrl = "https://sfu.example.org";
describe("getSFUConfigWithOpenID", () => {
let matrixClient: MockedObject<OpenIDClientParts>;
beforeEach(() => {
fetchMock.catch(404);
matrixClient = {
getOpenIdToken: vitest.fn(),
getDeviceId: vitest.fn(),
};
});
afterEach(() => {
vitest.clearAllMocks();
fetchMock.reset();
});
it("should handle fetching a token", async () => {
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
});
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
);
expect(config).toEqual({
jwt: testJWTToken,
url: sfuUrl,
livekitIdentity: "@me:example.org:ABCDEF",
livekitAlias: "!example_room_id",
});
void (await fetchMock.flush());
});
it("should fail if the SFU errors", async () => {
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
});
try {
await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
);
} catch (ex: unknown) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toBeInstanceOf(MatrixError);
const mxError = (ex as Error).cause as MatrixError;
expect(mxError.message).toEqual(
"MatrixError: [500] Failed to look up user info from homeserver",
);
void (await fetchMock.flush());
return;
}
expect.fail("Expected test to throw;");
});
it("should retry without delay params if the JWT service legacy endpoint returns M_BAD_JSON 400", async () => {
let callCount = 0;
fetchMock.post(
"https://sfu.example.org/sfu/get",
(url, opts) => {
callCount++;
const body = JSON.parse(opts.body as string);
// First call: check if it has delay parts and return 400
if (callCount === 1) {
expect(body).toHaveProperty("delay_id", "mock_delay_id");
return {
status: 400,
body: { errcode: "M_BAD_JSON", error: "Unsupported parameters" },
};
}
// Second call: check if delay parts were stripped and return success
expect(body).not.toHaveProperty("delay_id");
expect(body).not.toHaveProperty("delay_timeout");
expect(body).not.toHaveProperty("delay_cs_api_url");
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
},
{ overwriteRoutes: true },
);
// Note: Assuming getSFUConfigWithOpenID eventually calls getLiveKitJWT
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
},
);
expect(config.jwt).toBe(testJWTToken);
expect(callCount).toBe(2);
void (await fetchMock.flush());
});
it("should successfully send delay parameters to the JWT service legacy endpoint", async () => {
fetchMock.post(
"https://sfu.example.org/sfu/get",
(url, opts) => {
const body = JSON.parse(opts.body as string);
// Verify, that the request contains the expected delay parameters
if (
body.delay_id === "mock_delay_id" &&
body.delay_timeout === 3600000 &&
body.delay_cs_api_url === "https://homeserverserver.org/cs_api"
) {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
}
return {
status: 400,
body: { error: "Missing expected delay params" },
};
},
{ overwriteRoutes: true },
);
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://homeserverserver.org/cs_api",
delayId: "mock_delay_id",
},
);
// Prüfe das Ergebnis
expect(config).toMatchObject({
jwt: testJWTToken,
url: sfuUrl,
});
void (await fetchMock.flush());
});
it("should try legacy and then new endpoint with delay delegation", async () => {
fetchMock.post("https://sfu.example.org/get_token", () => {
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
});
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
});
try {
await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
},
);
} catch (ex) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toBeInstanceOf(MatrixError);
const mxError = (ex as Error).cause as MatrixError;
expect(mxError.message).toEqual(
"MatrixError: [500] Failed to look up user info from homeserver",
);
void (await fetchMock.flush());
}
const calls = fetchMock.calls();
expect(calls.length).toBe(2);
expect(calls[0][0]).toStrictEqual("https://sfu.example.org/get_token");
expect(calls[0][1]).toStrictEqual({
// check if it uses correct delayID!
body: '{"room_id":"!example_room_id","slot_id":"m.call#ROOM","member":{"id":"@alice:example.org:DEVICE","claimed_user_id":"@alice:example.org","claimed_device_id":"DEVICE"},"delay_id":"mock_delay_id","delay_timeout":3600000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}',
method: "POST",
headers: {
"Content-Type": "application/json",
},
});
expect(calls[1][0]).toStrictEqual("https://sfu.example.org/sfu/get");
expect(calls[1][1]).toStrictEqual({
body: '{"room":"!example_room_id","device_id":"DEVICE","delay_id":"mock_delay_id","delay_timeout":3600000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}',
headers: {
"Content-Type": "application/json",
},
method: "POST",
});
});
it("dont try legacy if endpoint with delay delegation is sucessful", async () => {
fetchMock.post("https://sfu.example.org/get_token", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
});
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 500,
body: { error: "Test failure" },
};
});
try {
await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
},
);
} catch (ex) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toEqual(
new Error("SFU Config fetch failed with status code 500"),
);
void (await fetchMock.flush());
}
const calls = fetchMock.calls();
expect(calls.length).toBe(1);
expect(calls[0][0]).toStrictEqual("https://sfu.example.org/get_token");
expect(calls[0][1]).toStrictEqual({
// check if it uses correct delayID!
body: '{"room_id":"!example_room_id","slot_id":"m.call#ROOM","member":{"id":"@alice:example.org:DEVICE","claimed_user_id":"@alice:example.org","claimed_device_id":"DEVICE"},"delay_id":"mock_delay_id","delay_timeout":3600000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}',
method: "POST",
headers: {
"Content-Type": "application/json",
},
});
});
it("should retry fetching the openid token", async () => {
let count = 0;
matrixClient.getOpenIdToken.mockImplementation(async () => {
count++;
if (count < 2) {
throw Error("Test failure");
}
return Promise.resolve({
token_type: "Bearer",
access_token: "foobar",
matrix_server_name: "example.org",
expires_in: 30,
});
});
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
});
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
);
expect(config).toEqual({
jwt: testJWTToken,
url: sfuUrl,
livekitIdentity: "@me:example.org:ABCDEF",
livekitAlias: "!example_room_id",
});
void (await fetchMock.flush());
});
});
src/livekit/openIDSFU.ts
-315
View File
@@ -1,315 +0,0 @@
/*
Copyright 2023, 2024 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 {
type IOpenIDToken,
type MatrixClient,
parseErrorResponse,
} from "matrix-js-sdk";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
FailToGetOpenIdToken,
NoMatrix2AuthorizationService,
} from "../utils/errors";
import { doNetworkOperationWithRetry } from "../utils/matrix";
import { Config } from "../config/Config";
import { MatrixRTCMode } from "../config/ConfigOptions";
/**
* Configuration and access tokens provided by the SFU on successful authentication.
*/
export interface SFUConfig {
url: string;
jwt: string;
livekitAlias: string;
// NOTE: Currently unused.
livekitIdentity: string;
}
/**
* Decoded details from the JWT.
*/
interface SFUJWTPayload {
/**
* Expiration time for the JWT.
* Note: This value is in seconds since Unix epoch.
*/
exp: number;
/**
* Name of the instance which authored the JWT
*/
iss: string;
/**
* Time at which the JWT can start to be used.
* Note: This value is in seconds since Unix epoch.
*/
nbf: number;
/**
* Subject. The Livekit alias in this context.
*/
sub: string;
/**
* The set of permissions for the user.
*/
video: {
canPublish: boolean;
canSubscribe: boolean;
room: string;
roomJoin: boolean;
};
}
// The bits we need from MatrixClient
export type OpenIDClientParts = Pick<
MatrixClient,
"getOpenIdToken" | "getDeviceId"
>;
/**
* Gets a bearer token from the homeserver and then use it to authenticate
* to the matrix RTC backend in order to get acces to the SFU.
* It has built-in retry for calls to the homeserver with a backoff policy.
* @param client The Matrix client
* @param membership Our own membership identity parts used to send to jwt service.
* @param serviceUrl The URL of the livekit SFU service
* @param roomId The room id used in the jwt request. This is NOT the livekit_alias. The jwt service will provide the alias. It maps matrix room ids <-> Livekit aliases.
* @param opts Additional options to modify which endpoint with which data will be used to acquire the jwt token.
* @param opts.matrixRTCMode Determines which version of the JWT endpoint to use, which affects whether the
* RTC backend identity is based on string concatenation (legacy) or a hash (Matrix 2.0).
* This function by default uses whatever is possible with the current jwt service installed next to the SFU.
* For remote connections this does not matter, since we will not publish there we can rely on the newest option.
* @param opts.delayEndpointBaseUrl The URL of the matrix homeserver.
* @param opts.delayId The delay id used for the jwt service to manage.
* @param logger optional logger.
* @returns Object containing the token information
* @throws FailToGetOpenIdToken
*/
export async function getSFUConfigWithOpenID(
client: OpenIDClientParts,
membership: CallMembershipIdentityParts,
serviceUrl: string,
roomId: string,
opts?: {
matrixRTCMode?: MatrixRTCMode;
delayEndpointBaseUrl?: string;
delayId?: string;
},
logger?: Logger,
): Promise<SFUConfig> {
let openIdToken: IOpenIDToken;
try {
openIdToken = await doNetworkOperationWithRetry(async () =>
client.getOpenIdToken(),
);
} catch (error) {
throw new FailToGetOpenIdToken(
error instanceof Error ? error : new Error("Unknown error"),
);
}
logger?.debug("Got openID token", openIdToken);
let sfuConfig: { url: string; jwt: string } | undefined;
const tryBothJwtEndpoints = opts?.matrixRTCMode === undefined; // This is for SFUs where we do not publish.
const forceMatrix2Jwt = opts?.matrixRTCMode === MatrixRTCMode.Matrix_2_0;
// We want to start using the new endpoint (with optional delay delegation)
// if we can use both or if we are forced to use the new one.
if (tryBothJwtEndpoints || forceMatrix2Jwt) {
try {
logger?.info(
`Trying to get JWT via default endpoint for focus ${serviceUrl}...`,
);
const sfuConfig = await getLiveKitJWT(
membership,
serviceUrl,
roomId,
openIdToken,
opts?.delayEndpointBaseUrl,
opts?.delayId,
);
return extractFullConfigFromToken(sfuConfig);
} catch (e) {
logger?.debug(`Failed fetching jwt with matrix 2.0 endpoint:`, e);
// Make this throw a hard error in case we force the matrix2.0 endpoint.
if (forceMatrix2Jwt) {
throw new NoMatrix2AuthorizationService(e as Error);
}
}
}
// DEPRECATED
// here we either have a sfuConfig or we already exited because of `if (forceMatrix2) throw ...`
// The only case we can get into this condition is, if `forceMatrix2` is `false`
try {
logger?.info(
`Trying to get JWT with legacy endpoint for focus ${serviceUrl}...`,
);
sfuConfig = await getLiveKitJWTLegacy(
membership.deviceId,
serviceUrl,
roomId,
openIdToken,
opts?.delayEndpointBaseUrl,
opts?.delayId,
);
logger?.info(`Got JWT from call's active focus URL.`);
return extractFullConfigFromToken(sfuConfig);
} catch (ex) {
throw new FailToGetOpenIdToken(
ex instanceof Error ? ex : new Error(`Unknown error ${ex}`),
);
}
}
function extractFullConfigFromToken(sfuConfig: {
url: string;
jwt: string;
}): SFUConfig {
const [, payloadStr] = sfuConfig.jwt.split(".");
const payload = JSON.parse(global.atob(payloadStr)) as SFUJWTPayload;
return {
jwt: sfuConfig.jwt,
url: sfuConfig.url,
livekitAlias: payload.video.room,
// NOTE: Currently unused.
// Probably also not helpful since we now compute the backendIdentity on joining the call so we can use it for the encryption manager.
// The only reason for us to know it locally is to connect the right users with the lk world. (and to set our own keys)
livekitIdentity: payload.sub,
};
}
async function getLiveKitJWTLegacy(
deviceId: string,
livekitServiceURL: string,
matrixRoomId: string,
openIDToken: IOpenIDToken,
delayEndpointBaseUrl?: string,
delayId?: string,
): Promise<{ url: string; jwt: string }> {
interface IDelayParams {
delay_id?: string;
delay_timeout?: number;
delay_cs_api_url?: string;
}
let bodyDalayParts: IDelayParams = {};
// Also check for empty string
if (delayId && delayEndpointBaseUrl) {
bodyDalayParts = {
delay_id: delayId,
delay_timeout:
Config.get().matrix_rtc_session.delegated_delayed_leave.delay_ms,
delay_cs_api_url: delayEndpointBaseUrl,
};
}
const makeRequest = async (delayParts: IDelayParams): Promise<Response> => {
return await fetch(livekitServiceURL + "/sfu/get", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({
// The legacy JWT endpoint uses only the matrix room id to calculate the livekit room alias.
// However, the livekit room alias is provided as part of the JWT payload.
room: matrixRoomId,
openid_token: openIDToken,
device_id: deviceId,
...delayParts,
}),
});
};
const res = await doNetworkOperationWithRetry(async () => {
let response = await makeRequest(bodyDalayParts);
// Old service compatibility check
const oldServiceDoesNotSupportDelayParts =
response.status === 400 && Object.keys(bodyDalayParts).length > 0;
// If http status 400 with M_BAD_JSON and we sent delay parts, retry without them
if (oldServiceDoesNotSupportDelayParts) {
try {
const errorBody = await response.json();
if (errorBody.errcode === "M_BAD_JSON") {
response = await makeRequest({});
}
} catch {
// If we can't parse the error, treat as real error
}
}
return response;
});
if (!res.ok) {
throw parseErrorResponse(res, await res.text());
}
return await res.json();
}
class NotSupportedError extends Error {
public constructor(message: string) {
super(message);
this.name = "NotSupported";
}
}
export async function getLiveKitJWT(
membership: CallMembershipIdentityParts,
livekitServiceURL: string,
matrixRoomId: string,
openIDToken: IOpenIDToken,
delayEndpointBaseUrl?: string,
delayId?: string,
): Promise<{ url: string; jwt: string }> {
const { userId, deviceId, memberId } = membership;
const body = {
room_id: matrixRoomId,
slot_id: "m.call#ROOM",
openid_token: openIDToken,
member: {
id: memberId,
claimed_user_id: userId,
claimed_device_id: deviceId,
},
};
let bodyDalayParts = {};
// Also check for empty string
if (delayId && delayEndpointBaseUrl) {
bodyDalayParts = {
delay_id: delayId,
delay_timeout:
Config.get().matrix_rtc_session.delegated_delayed_leave.delay_ms,
delay_cs_api_url: delayEndpointBaseUrl,
};
}
const res = await doNetworkOperationWithRetry(async () => {
return await fetch(livekitServiceURL + "/get_token", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ ...body, ...bodyDalayParts }),
});
});
if (!res.ok) {
const msg = "SFU Config fetch failed with status code " + res.status;
if (res.status === 404) {
throw new NotSupportedError(msg);
} else {
throw parseErrorResponse(res, await res.text());
}
}
return await res.json();
}
src/settings/DeveloperSettingsTab.test.tsx
+12 -12
View File
@@ -18,7 +18,6 @@ import {
} from "./DeveloperSettingsTab";
import { outOfCallDeveloperSettingsTabViewModel } from "./DeveloperSettingsTabViewModel";
import { createStaticViewModel } from "../state/ViewModel";
import { getSFUConfigWithOpenID } from "../livekit/openIDSFU";
import {
customLivekitUrl as customLivekitUrlSetting,
enableExtendedLivekitLogs as enableExtendedLivekitLogsSetting,
@@ -26,6 +25,7 @@ import {
} from "./settings";
import { MatrixRTCMode } from "../config/ConfigOptions";
import { mockConfig } from "../utils/test";
import { getSFUConfig } from "../livekit/auth";
// Mock url params hook to avoid environment-dependent snapshot churn.
vi.mock("../UrlParams", () => ({
@@ -36,8 +36,8 @@ vi.mock("../UrlParams", () => ({
}));
// IMPORTANT: mock the same specifier used by DeveloperSettingsTab
vi.mock("../livekit/openIDSFU", () => ({
getSFUConfigWithOpenID: vi.fn().mockResolvedValue({
vi.mock("../livekit/auth", () => ({
getSFUConfig: vi.fn().mockResolvedValue({
url: "mock-url",
jwt: "mock-jwt",
}),
@@ -153,7 +153,7 @@ describe("DeveloperSettingsTab", () => {
const saveButton = screen.getByRole("button", { name: "Save" });
await user.click(saveButton);
expect(getSFUConfigWithOpenID).not.toHaveBeenCalled();
expect(getSFUConfig).not.toHaveBeenCalled();
expect(customLivekitUrlSetting.getValue()).toBe(null);
});
@@ -176,7 +176,7 @@ describe("DeveloperSettingsTab", () => {
const saveButton = screen.getByRole("button", { name: "Save" });
await user.click(saveButton);
expect(getSFUConfigWithOpenID).not.toHaveBeenCalled();
expect(getSFUConfig).not.toHaveBeenCalled();
expect(customLivekitUrlSetting.getValue()).toBe(null);
});
@@ -202,7 +202,7 @@ describe("DeveloperSettingsTab", () => {
name: "Reset overwrite",
});
await user.click(cancelButton);
expect(getSFUConfigWithOpenID).not.toHaveBeenCalled();
expect(getSFUConfig).not.toHaveBeenCalled();
expect(customLivekitUrlSetting.getValue()).toBe(null);
});
@@ -226,11 +226,11 @@ describe("DeveloperSettingsTab", () => {
const saveButton = screen.getByRole("button", { name: "Save" });
await user.click(saveButton);
expect(getSFUConfigWithOpenID).toHaveBeenCalledWith(
expect.anything(),
expect.anything(),
"wss://example.livekit.valid",
"#testRoom",
expect(getSFUConfig).toHaveBeenCalledWith(
expect.objectContaining({
serviceUrl: "wss://example.livekit.valid",
roomId: "#testRoom",
}),
);
expect(customLivekitUrlSetting.getValue()).toBe(
@@ -256,7 +256,7 @@ describe("DeveloperSettingsTab", () => {
await user.type(input, "wss://example.livekit.valid");
const saveButton = screen.getByRole("button", { name: "Save" });
(getSFUConfigWithOpenID as Mock).mockImplementation(() => {
(getSFUConfig as Mock).mockImplementation(() => {
throw new Error("Invalid URL");
});
await user.click(saveButton);
src/settings/DeveloperSettingsTab.tsx
+8 -5
View File
@@ -67,9 +67,9 @@ import styles from "./DeveloperSettingsTab.module.css";
import settingsStyles from "./SettingsModal.module.css";
import { Slider } from "../Slider";
import { useUrlParams } from "../UrlParams";
import { getSFUConfigWithOpenID } from "../livekit/openIDSFU";
import { useBehavior } from "../useBehavior";
import { type ViewModel } from "../state/ViewModel.ts";
import { getSFUConfig } from "../livekit/auth/index.ts";
/**
* The state of MatrixRTC's media key rotation.
@@ -518,12 +518,15 @@ export const DeveloperSettingsTab: FC<Props> = ({
if (userId === null || deviceId === null) {
throw new Error("Invalid user or device ID");
}
await getSFUConfigWithOpenID(
await getSFUConfig({
client,
{ userId, deviceId, memberId: "" },
customLivekitUrlTextBuffer,
membership: { userId, deviceId, memberId: "" },
serviceUrl: customLivekitUrlTextBuffer,
roomId,
);
slotId: "org.example.fake-slot#room",
role: "publisher",
logger,
});
setCustomLivekitUrlUpdateError(null);
setCustomLivekitUrl(customLivekitUrlTextBuffer);
} catch {
src/state/CallViewModel/CallViewModel.ts
+2 -2
View File
@@ -570,7 +570,7 @@ export function createCallViewModel$(
ownMembershipIdentity,
client,
roomId: matrixRoom.roomId,
matrixRTCMode,
slotId: matrixRTCSession.slotId,
}),
);
@@ -579,6 +579,7 @@ export function createCallViewModel$(
new ECConnectionFactory(
client,
matrixRoom.roomId,
matrixRTCSession.slotId,
mediaDevices,
trackProcessorState$,
livekitKeyProvider,
@@ -659,7 +660,6 @@ export function createCallViewModel$(
).pipe(map(([delayId]) => delayId ?? null)),
matrixRTCSession.delayId ?? null,
),
matrixRTCMode,
logger: logger.getChild(`[${Date.now()}]`),
});
src/state/CallViewModel/CallViewModelTestUtils.ts
+1
View File
@@ -229,6 +229,7 @@ export function withCallViewModel(mode: MatrixRTCMode) {
existingSFUConfig: sfuConfig,
client: room.client,
roomId: room.roomId,
slotId: "m.call#ROOM",
livekitRoomFactory,
},
logger,
src/state/CallViewModel/localMember/LocalMember.test.ts
+13 -15
View File
@@ -66,7 +66,7 @@ import { type Publisher } from "./Publisher";
import { initializeWidget } from "../../../widget";
import { nullHostBridge } from "../../../HostBridge";
import { type LocalTransport } from "./LocalTransport";
import * as openIDSFU from "../../../livekit/openIDSFU";
import * as livekitAuth from "../../../livekit/auth";
initializeWidget();
@@ -237,7 +237,6 @@ describe("LocalMembership", () => {
ownMembershipIdentity: ownMemberMock,
client: mockedClient,
delayId$: constant(null),
matrixRTCMode: MATRIX_RTC_MODE,
};
beforeEach(() => {
@@ -306,6 +305,7 @@ describe("LocalMembership", () => {
scope,
...defaultCreateLocalMemberValues,
matrixRTCSession: {
slotId: "m.call#ROOM",
updateCallIntent: vi.fn().mockImplementation(reject),
leave: vi.fn(),
},
@@ -359,7 +359,7 @@ describe("LocalMembership", () => {
} as unknown as Connection;
const authCallSpy = vi
.spyOn(openIDSFU, "getSFUConfigWithOpenID")
.spyOn(livekitAuth, "getSFUConfig")
.mockImplementation(() => mockedClient.getOpenIdToken());
afterEach(() => authCallSpy.mockClear());
@@ -412,18 +412,16 @@ describe("LocalMembership", () => {
// Delegation is supported in this test case, so go on to check that
// LocalMember actually performs delegation
const expectDelegation = (delayId: string) =>
expect(authCallSpy).toHaveBeenLastCalledWith(
mockedClient,
ownMemberMock,
"a",
"!test-room-id:example.org",
{
matrixRTCMode: MATRIX_RTC_MODE,
delayEndpointBaseUrl: "https://matrix.example.org",
delayId,
},
expect.anything(),
);
expect(authCallSpy).toHaveBeenLastCalledWith({
client: mockedClient,
membership: ownMemberMock,
serviceUrl: "a",
roomId: "!test-room-id:example.org",
role: "publisher",
delayEndpointBaseUrl: "https://matrix.example.org",
delayId,
logger: expect.anything(),
});
expectDelegation("leave1");
delayId$.next("leave2"); // Can change delegated leaves
src/state/CallViewModel/localMember/LocalMember.ts
+14 -11
View File
@@ -16,7 +16,6 @@ import {
MediaDeviceFailure,
} from "livekit-client";
import { observeParticipantEvents } from "@livekit/components-core";
import { type MatrixClient } from "matrix-js-sdk";
import {
Status as RTCSessionStatus,
type LivekitTransport,
@@ -80,7 +79,7 @@ import {
} from "../remoteMembers/Connection.ts";
import { type HomeserverConnected } from "./HomeserverConnected.ts";
import { type LocalTransport } from "./LocalTransport.ts";
import { getSFUConfigWithOpenID } from "../../../livekit/openIDSFU.ts";
import { type ClientOpenIDParts, getSFUConfig } from "../../../livekit/auth";
export enum TransportState {
/** Not even a transport is available to the LocalMembership */
@@ -150,15 +149,17 @@ interface Props {
roomId: string;
ownMembershipIdentity: CallMembershipIdentityParts;
localTransport$: Observable<LocalTransport>;
client: Pick<MatrixClient, "getDeviceId" | "getOpenIdToken">;
matrixRTCSession: Pick<MatrixRTCSession, "updateCallIntent" | "leave">;
client: ClientOpenIDParts;
matrixRTCSession: Pick<
MatrixRTCSession,
"slotId" | "updateCallIntent" | "leave"
>;
/** Whether to hide the screen-sharing button. */
hideScreensharing: boolean;
/** The application hosting Element Call, to be kept informed of join/leave. */
hostBridge: HostBridge;
baseUrl: string;
delayId$: Behavior<string | null>;
matrixRTCMode: MatrixRTCMode;
logger: Logger;
}
@@ -206,7 +207,6 @@ export const createLocalMembership$ = ({
hostBridge,
ownMembershipIdentity,
delayId$,
matrixRTCMode,
}: Props): {
/**
* This request to start audio and video tracks.
@@ -713,14 +713,17 @@ export const createLocalMembership$ = ({
// 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 getSFUConfigWithOpenID(
await getSFUConfig({
client,
ownMembershipIdentity,
joinParams.transport.livekit_service_url,
membership: ownMembershipIdentity,
serviceUrl: joinParams.transport.livekit_service_url,
roomId,
{ matrixRTCMode, delayEndpointBaseUrl: baseUrl, delayId },
slotId: matrixRTCSession.slotId,
role: "publisher",
delayEndpointBaseUrl: baseUrl,
delayId,
logger,
);
});
} catch (e) {
// TODO: Surface this to the user as a service interruption?
logger.error(
src/state/CallViewModel/localMember/LocalTransport.test.ts
+13 -25
View File
@@ -22,13 +22,12 @@ import {
MatrixRTCTransportMissingError,
FailToGetOpenIdToken,
} from "../../../utils/errors";
import * as openIDSFU from "../../../livekit/openIDSFU";
import * as livekitAuth from "../../../livekit/auth";
import { customLivekitUrl } from "../../../settings/settings";
import { testJWTToken } from "../../../utils/test-fixtures";
import { MatrixRTCMode } from "../../../config/ConfigOptions";
describe("LocalTransport", () => {
const openIdResponse: openIDSFU.SFUConfig = {
const openIdResponse: livekitAuth.SFUConfig = {
url: "https://lk.example.org",
jwt: testJWTToken,
livekitAlias: "Akph4alDMhen",
@@ -47,10 +46,9 @@ describe("LocalTransport", () => {
getDomain: () => "example.org",
// These won't be called in this error path but satisfy the type
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
matrixRTCMode: MatrixRTCMode.Compatibility,
slotId: "m.call#ROOM",
}),
).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
});
@@ -60,7 +58,7 @@ describe("LocalTransport", () => {
mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" },
});
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
vi.spyOn(livekitAuth, "getSFUConfig").mockRejectedValue(
new FailToGetOpenIdToken(new Error("no openid")),
);
@@ -72,10 +70,9 @@ describe("LocalTransport", () => {
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
matrixRTCMode: MatrixRTCMode.Compatibility,
slotId: "m.call#ROOM",
}),
).rejects.toThrow(new FailToGetOpenIdToken(new Error("no openid")));
});
@@ -86,7 +83,7 @@ describe("LocalTransport", () => {
livekit: { livekit_service_url: "https://lk.example.org" },
});
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue({
vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue({
url: "https://lk.example.org",
jwt: "jwt",
livekitAlias: "Akph4alDMhen",
@@ -101,10 +98,9 @@ describe("LocalTransport", () => {
_unstable_getRTCTransports: async () => Promise.resolve([]),
getDomain: () => "example.org",
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
matrixRTCMode: MatrixRTCMode.Compatibility,
slotId: "m.call#ROOM",
}),
).toStrictEqual({
transport: {
@@ -132,13 +128,12 @@ describe("LocalTransport", () => {
localTransportOpts = {
ownMembershipIdentity: ownMemberMock,
roomId: "!example_room_id",
matrixRTCMode: MatrixRTCMode.Compatibility,
slotId: "m.call#ROOM",
client: {
getDomain: vi.fn().mockReturnValue("example.org"),
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
};
});
@@ -151,9 +146,7 @@ describe("LocalTransport", () => {
mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" },
});
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
openIdResponse,
);
vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(openIdResponse);
expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
transport: {
@@ -171,9 +164,7 @@ describe("LocalTransport", () => {
it("supports getting transport via user settings", async () => {
customLivekitUrl.setValue("https://lk.example.org");
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
openIdResponse,
);
vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(openIdResponse);
expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
transport: {
@@ -193,9 +184,7 @@ describe("LocalTransport", () => {
localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", livekit_service_url: "https://lk.example.org" },
]);
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
openIdResponse,
);
vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(openIdResponse);
expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
transport: {
@@ -215,7 +204,7 @@ describe("LocalTransport", () => {
localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", livekit_service_url: "https://lk.example.org" },
]);
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
vi.spyOn(livekitAuth, "getSFUConfig").mockRejectedValue(
new FailToGetOpenIdToken(new Error("Test driven error")),
);
@@ -229,14 +218,13 @@ describe("LocalTransport", () => {
getLocalTransport({
ownMembershipIdentity: ownMemberMock,
roomId: "!example_room_id",
matrixRTCMode: MatrixRTCMode.Compatibility,
slotId: "m.call#ROOM",
client: {
getDomain: () => "example.org",
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
// These won't be called in this error path but satisfy the type
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
}),
).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
src/state/CallViewModel/localMember/LocalTransport.ts
+17 -17
View File
@@ -17,21 +17,20 @@ import {
NoMatrix2AuthorizationService,
} from "../../../utils/errors.ts";
import {
getSFUConfigWithOpenID,
getSFUConfig,
type SFUConfig,
type OpenIDClientParts,
} from "../../../livekit/openIDSFU.ts";
type ClientOpenIDParts,
} from "../../../livekit/auth";
import { customLivekitUrl } from "../../../settings/settings.ts";
import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
import { type MatrixRTCMode } from "../../../config/ConfigOptions.ts";
interface Props {
ownMembershipIdentity: CallMembershipIdentityParts;
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> &
OpenIDClientParts;
ClientOpenIDParts;
// Used by the jwt service to create the livekit room and compute the livekit alias.
roomId: string;
matrixRTCMode: MatrixRTCMode;
slotId: string;
}
export interface LocalTransport {
@@ -58,7 +57,7 @@ export async function getLocalTransport({
ownMembershipIdentity,
client,
roomId,
matrixRTCMode,
slotId,
}: Props): Promise<LocalTransport> {
const logger = rootLogger.getChild("[LocalTransport]");
const discovery = new RtcTransportAutoDiscovery({
@@ -79,9 +78,9 @@ export async function getLocalTransport({
try {
return await doOpenIdAndJWTFromUrl(
transport,
matrixRTCMode,
ownMembershipIdentity,
roomId,
slotId,
client,
logger,
);
@@ -96,35 +95,36 @@ export async function getLocalTransport({
/**
* Utility to ensure the user can authenticate with the SFU.
* We will call `getSFUConfigWithOpenID` once per transport here as it's our
* We will call `getSFUConfig` once per transport here as it's our
* only mechanism of validation. This means we will also ask the
* homeserver for a OpenID token a few times. Since OpenID tokens are single
* use we don't want to risk any issues by re-using a token.
*
* @param transport The transport to authenticate with.
* @param matrixRTCMode Whether to force the JWT endpoint to be used.
* @param membership The identity of the local member.
* @param roomId The room ID to use for the JWT.
* @param slotId The slot ID to use for the JWT.
* @param client The client to use for the OpenID token.
*
* @throws FailToGetOpenIdToken, NoMatrix2AuthorizationService
*/
async function doOpenIdAndJWTFromUrl(
transport: LivekitTransport,
matrixRTCMode: MatrixRTCMode,
membership: CallMembershipIdentityParts,
roomId: string,
client: Pick<MatrixClient, "_unstable_getRTCTransports"> & OpenIDClientParts,
logger?: Logger,
slotId: string,
client: Pick<MatrixClient, "_unstable_getRTCTransports"> & ClientOpenIDParts,
logger: Logger,
): Promise<LocalTransport> {
const sfuConfig = await getSFUConfigWithOpenID(
const sfuConfig = await getSFUConfig({
client,
membership,
transport.livekit_service_url,
serviceUrl: transport.livekit_service_url,
roomId,
{ matrixRTCMode },
slotId,
role: "publisher",
logger,
);
});
return {
transport,
sfuConfig,
src/state/CallViewModel/remoteMembers/Connection.test.ts
+33 -38
View File
@@ -8,7 +8,6 @@ Please see LICENSE in the repository root for full details.
import {
afterEach,
beforeEach,
describe,
expect,
it,
@@ -38,14 +37,14 @@ import {
type ConnectionOpts,
} from "./Connection.ts";
import { ObservableScope } from "../../ObservableScope.ts";
import { type OpenIDClientParts } from "../../../livekit/openIDSFU.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";
let testScope: ObservableScope;
let client: MockedObject<OpenIDClientParts>;
let client: MockedObject<ClientOpenIDParts>;
let fakeLivekitRoom: MockedObject<LivekitRoom>;
@@ -53,6 +52,7 @@ let localParticipantEventEmiter: EventEmitter;
let fakeLocalParticipant: MockedObject<LocalParticipant>;
const ROOM_ID = "!roomID:example.org";
const SLOT_ID = "m.call#ROOM";
const livekitFocus: LivekitTransport = {
livekit_service_url: "https://matrix-rtc.example.org/livekit/jwt",
@@ -61,15 +61,14 @@ const livekitFocus: LivekitTransport = {
function setupTest(): void {
testScope = new ObservableScope();
client = vi.mocked<OpenIDClientParts>({
client = vi.mocked<ClientOpenIDParts>({
getOpenIdToken: vi.fn().mockResolvedValue({
access_token: "rYsmGUEwNjKgJYyeNUkZseJN",
token_type: "Bearer",
matrix_server_name: "example.org",
expires_in: 3600,
}),
getDeviceId: vi.fn().mockReturnValue("ABCDEF"),
} as unknown as OpenIDClientParts);
});
localParticipantEventEmiter = new EventEmitter();
@@ -113,15 +112,16 @@ function setupTest(): void {
function setupRemoteConnection(): Connection {
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
};
fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, () => {
fetchMock.post(`${livekitFocus.livekit_service_url}/get_token`, () => {
return {
status: 200,
body: {
@@ -152,25 +152,13 @@ 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();
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
@@ -186,8 +174,9 @@ describe("Start connection states", () => {
vi.useFakeTimers();
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
@@ -238,8 +227,9 @@ describe("Start connection states", () => {
vi.useFakeTimers();
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
@@ -255,17 +245,20 @@ describe("Start connection states", () => {
onTestFinished(() => s.unsubscribe());
const deferredSFU = Promise.withResolvers<void>();
// mock the /sfu/get call
fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, async () => {
await deferredSFU.promise;
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
});
// mock the /get_token call
fetchMock.post(
`${livekitFocus.livekit_service_url}/get_token`,
async () => {
await deferredSFU.promise;
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
},
);
connection.start().catch(() => {
// expected to throw
@@ -300,8 +293,9 @@ describe("Start connection states", () => {
vi.useFakeTimers();
const opts: ConnectionOpts = {
client: client,
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
@@ -317,8 +311,8 @@ describe("Start connection states", () => {
onTestFinished(() => s.unsubscribe());
const deferredSFU = Promise.withResolvers<void>();
// mock the /sfu/get call
fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, () => {
// mock the /get_token call
fetchMock.post(`${livekitFocus.livekit_service_url}/get_token`, () => {
return {
status: 200,
body: {
@@ -560,6 +554,7 @@ describe("remote track logging", () => {
{
client,
roomId: ROOM_ID,
slotId: SLOT_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
src/state/CallViewModel/remoteMembers/Connection.ts
+23 -19
View File
@@ -28,11 +28,6 @@ import { BehaviorSubject } from "rxjs";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import {
getSFUConfigWithOpenID,
type OpenIDClientParts,
type SFUConfig,
} from "../../../livekit/openIDSFU.ts";
import { type Behavior } from "../../Behavior.ts";
import { type ObservableScope } from "../../ObservableScope.ts";
import {
@@ -43,6 +38,11 @@ import {
SFURoomCreationRestrictedError,
UnknownCallError,
} from "../../../utils/errors.ts";
import {
getSFUConfig,
type ClientOpenIDParts,
type SFUConfig,
} from "../../../livekit/auth";
export interface ConnectionOpts {
/**
@@ -55,9 +55,11 @@ export interface ConnectionOpts {
/** The media transport to connect to. */
transport: LivekitTransport;
/** The Matrix client to use for OpenID and SFU config requests. */
client: OpenIDClientParts;
/** The room ID this connection is associated with. */
client: 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. */
slotId: string;
/** The observable scope to use for this connection. */
scope: ObservableScope;
@@ -123,8 +125,9 @@ export class Connection {
protected stopped = false;
// TODO: can we just keep the ConnectionOpts object instead of spreading?
private readonly client: OpenIDClientParts;
private readonly client: ClientOpenIDParts;
private readonly roomId: string;
private readonly slotId: string;
private readonly logger: Logger;
private readonly ownMembershipIdentity: CallMembershipIdentityParts;
private readonly existingSFUConfig?: SFUConfig;
@@ -139,6 +142,7 @@ export class Connection {
this.ownMembershipIdentity = opts.ownMembershipIdentity;
this.existingSFUConfig = opts.existingSFUConfig;
this.roomId = opts.roomId;
this.slotId = opts.slotId;
this.logger = logger.getChild(
`[Connection ${JSON.stringify(opts.transport)}]`,
);
@@ -292,7 +296,7 @@ export class Connection {
try {
this._state$.next({ state: "authenticating" });
// We should already have this information after creating the localTransport.
// only call getSFUConfigWithOpenID for connections where we do not have a token yet. (existingJwtTokenData === undefined)
// only call getSFUConfig for connections where we do not have a token yet. (existingJwtTokenData === undefined)
const { url, jwt, livekitAlias } =
this.existingSFUConfig ??
(await this.getSFUConfigForRemoteConnection());
@@ -308,7 +312,7 @@ export class Connection {
// If we were stopped while fetching the config, don't proceed to connect
if (this.stopped) return;
// Setup observer once we are done with getSFUConfigWithOpenID
// Setup observer once we are done with getSFUConfig
connectionStateObserver(this.livekitRoom)
.pipe(this.scope.bind())
.subscribe((livekitState) =>
@@ -380,15 +384,15 @@ export class Connection {
protected async getSFUConfigForRemoteConnection(): Promise<SFUConfig> {
// This will only be called for sfu's where we do not publish ourselves.
// For the local connection we will use the existingJwtTokenData
return await getSFUConfigWithOpenID(
this.client,
this.ownMembershipIdentity,
this.transport.livekit_service_url,
this.roomId,
// dont pass any custom opts for the subscribe only connections
{},
this.logger,
);
return await getSFUConfig({
client: this.client,
membership: this.ownMembershipIdentity,
serviceUrl: this.transport.livekit_service_url,
roomId: this.roomId,
slotId: this.slotId,
role: "subscriber",
logger: this.logger,
});
}
/**
src/state/CallViewModel/remoteMembers/ConnectionFactory.ts
+6 -6
View File
@@ -20,10 +20,6 @@ import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type ObservableScope } from "../../ObservableScope.ts";
import { Connection } from "./Connection.ts";
import type {
OpenIDClientParts,
SFUConfig,
} from "../../../livekit/openIDSFU.ts";
import type { MediaDevices } from "../../MediaDevices.ts";
import type { Behavior } from "../../Behavior.ts";
import type { ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
@@ -39,6 +35,7 @@ import {
noiseSuppressionSetting,
autoGainControlSetting,
} from "../../../settings/settings.ts";
import { type ClientOpenIDParts, type SFUConfig } from "../../../livekit/auth";
// TODO evaluate if this should be done like the Publisher Factory
export interface ConnectionFactory {
@@ -58,7 +55,8 @@ export class ECConnectionFactory implements ConnectionFactory {
* Creates a ConnectionFactory for LiveKit connections.
*
* @param client - The OpenID client parts for authentication, needed to get openID and JWT tokens.
* @param roomId - The current room ID.
* @param roomId - The ID of the Matrix room in which the session takes place.
* @param slotId - The ID of the MatrixRTC slot in which the session takes place.
* @param devices - Used for video/audio out/in capture options.
* @param processorState$ - Effects like background blur (only for publishing connection?)
* @param livekitKeyProvider - Optional key provider for end-to-end encryption.
@@ -66,8 +64,9 @@ 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: OpenIDClientParts,
private client: ClientOpenIDParts,
private readonly roomId: string,
private readonly slotId: string,
private devices: MediaDevices,
private processorState$: Behavior<ProcessorState>,
livekitKeyProvider: BaseKeyProvider | undefined,
@@ -112,6 +111,7 @@ export class ECConnectionFactory implements ConnectionFactory {
{
existingSFUConfig: sfuConfig,
roomId: this.roomId,
slotId: this.slotId,
transport,
client: this.client,
scope: scope,
src/state/CallViewModel/remoteMembers/ConnectionManager.ts
+1 -1
View File
@@ -31,7 +31,7 @@ import {
isLocalTransport,
type LocalTransport,
} from "../localMember/LocalTransport.ts";
import { type SFUConfig } from "../../../livekit/openIDSFU.ts";
import { type SFUConfig } from "../../../livekit/auth";
export class ConnectionManagerData {
private readonly store: Map<
src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts
+7 -3
View File
@@ -14,7 +14,7 @@ import EventEmitter from "events";
import { ObservableScope } from "../../ObservableScope.ts";
import { ECConnectionFactory } from "./ConnectionFactory.ts";
import type { OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
import { type ClientOpenIDParts } from "../../../livekit/auth";
import {
exampleTransport,
mockMediaDevices,
@@ -51,13 +51,12 @@ vi.mock("livekit-client", async (importOriginal) => {
});
let testScope: ObservableScope;
let mockClient: OpenIDClientParts;
let mockClient: ClientOpenIDParts;
beforeEach(() => {
testScope = new ObservableScope();
mockClient = {
getOpenIdToken: vi.fn().mockReturnValue(""),
getDeviceId: vi.fn().mockReturnValue("DEV000"),
};
});
@@ -79,6 +78,7 @@ describe("ECConnectionFactory - Audio inputs options", () => {
const ecConnectionFactory = new ECConnectionFactory(
mockClient,
"!roomid:example.org",
"m.call#ROOM",
mockMediaDevices({}),
new BehaviorSubject<ProcessorState>({
supported: true,
@@ -121,6 +121,7 @@ describe("ECConnectionFactory - ControlledAudioDevice", () => {
const ecConnectionFactory = new ECConnectionFactory(
mockClient,
"!roomid:example.org",
"m.call#ROOM",
mockMediaDevices({
audioOutput: {
available$: constant(new Map<never, never>()),
@@ -162,6 +163,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
const ecConnectionFactory = new ECConnectionFactory(
mockClient,
"!roomid:example.org",
"m.call#ROOM",
mockMediaDevices({}),
new BehaviorSubject<ProcessorState>({
supported: true,
@@ -199,6 +201,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
const ecConnectionFactory = new ECConnectionFactory(
mockClient,
"!roomid:example.org",
"m.call#ROOM",
mockMediaDevices({}),
new BehaviorSubject<ProcessorState>({
supported: true,
@@ -237,6 +240,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
const ecConnectionFactory = new ECConnectionFactory(
mockClient,
"!roomid:example.org",
"m.call#ROOM",
mockMediaDevices({}),
new BehaviorSubject<ProcessorState>({
supported: true,
src/state/CallViewModel/remoteMembers/integration.test.ts
+7 -5
View File
@@ -19,7 +19,7 @@ import {
trackEpoch,
} from "../../ObservableScope.ts";
import { ECConnectionFactory } from "./ConnectionFactory.ts";
import { type OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
import { type ClientOpenIDParts } from "../../../livekit/auth";
import {
mockMediaDevices,
mockRtcMembership,
@@ -40,7 +40,7 @@ import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts";
let testScope: ObservableScope;
let ecConnectionFactory: ECConnectionFactory;
let mockClient: OpenIDClientParts;
let mockClient: ClientOpenIDParts;
let lkRoomFactory: () => LivekitRoom;
const createdMockLivekitRooms: Map<string, LivekitRoom> = new Map();
@@ -49,7 +49,6 @@ beforeEach(() => {
testScope = new ObservableScope();
mockClient = {
getOpenIdToken: vi.fn().mockReturnValue(""),
getDeviceId: vi.fn().mockReturnValue("DEV000"),
};
lkRoomFactory = vi.fn().mockImplementation(() => {
@@ -71,6 +70,7 @@ beforeEach(() => {
ecConnectionFactory = new ECConnectionFactory(
mockClient,
"!roomid:example.org",
"m.call#ROOM",
mockMediaDevices({}),
new BehaviorSubject<ProcessorState>({
supported: true,
@@ -82,7 +82,7 @@ beforeEach(() => {
);
//TODO a bit annoying to have to do a http mock?
fetchMock.post(`path:/sfu/get`, (url) => {
const sfuConfigResponse = (url: string): { status: number; body: object } => {
const domain = new URL(url).hostname; // Extract the domain from the URL
return {
status: 200,
@@ -91,7 +91,9 @@ beforeEach(() => {
jwt: testJWTToken,
},
};
});
};
fetchMock.post(`path:/sfu/get`, sfuConfigResponse);
fetchMock.post(`path:/get_token`, sfuConfigResponse);
});
afterEach(() => {
src/utils/test.ts
+1 -1
View File
@@ -78,7 +78,7 @@ import {
type RemoteScreenShareViewModel,
} from "../state/media/RemoteScreenShareViewModel";
import { Connection } from "../state/CallViewModel/remoteMembers/Connection";
import { type SFUConfig } from "../livekit/openIDSFU";
import { type SFUConfig } from "../livekit/auth";
export function withFakeTimers(continuation: () => void): void {
vi.useFakeTimers();