Robin
13d131c2e9
To correctly implement the legacy "oldest membership" mode, we need the code to be more nuanced about the local transport. Specifically, it needs to allow for the transport we advertise in our membership to be different from the transport that we connect to and publish media on. Otherwise, if these two are yoked together, members will resend their memberships whenever an SFU hop occurs, which an attacker could use to cause an amplified wave of state changes.
462 lines
16 KiB
TypeScript
462 lines
16 KiB
TypeScript
/*
|
|
Copyright 2025 Element Creations Ltd.
|
|
|
|
SPDX-License-IdFentifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
|
Please see LICENSE in the repository root for full details.
|
|
*/
|
|
|
|
import {
|
|
type CallMembership,
|
|
isLivekitTransportConfig,
|
|
type Transport,
|
|
type LivekitTransportConfig,
|
|
} from "matrix-js-sdk/lib/matrixrtc";
|
|
import { MatrixError, type MatrixClient } from "matrix-js-sdk";
|
|
import {
|
|
distinctUntilChanged,
|
|
first,
|
|
from,
|
|
map,
|
|
merge,
|
|
of,
|
|
startWith,
|
|
switchMap,
|
|
tap,
|
|
} from "rxjs";
|
|
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
|
|
import { AutoDiscovery } from "matrix-js-sdk/lib/autodiscovery";
|
|
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
|
|
|
import { type Behavior } from "../../Behavior.ts";
|
|
import { type Epoch, type ObservableScope } from "../../ObservableScope.ts";
|
|
import { Config } from "../../../config/Config.ts";
|
|
import {
|
|
FailToGetOpenIdToken,
|
|
MatrixRTCTransportMissingError,
|
|
NoMatrix2AuthorizationService,
|
|
} from "../../../utils/errors.ts";
|
|
import {
|
|
getSFUConfigWithOpenID,
|
|
type SFUConfig,
|
|
type OpenIDClientParts,
|
|
} from "../../../livekit/openIDSFU.ts";
|
|
import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts";
|
|
import { customLivekitUrl } from "../../../settings/settings.ts";
|
|
|
|
const logger = rootLogger.getChild("[LocalTransport]");
|
|
|
|
/*
|
|
* It figures out “which LiveKit focus URL/alias the local user should use,”
|
|
* optionally aligning with the oldest member, and ensures the SFU path is primed
|
|
* before advertising that choice.
|
|
*/
|
|
interface Props {
|
|
scope: ObservableScope;
|
|
ownMembershipIdentity: CallMembershipIdentityParts;
|
|
memberships$: Behavior<Epoch<CallMembership[]>>;
|
|
client: Pick<
|
|
MatrixClient,
|
|
"getDomain" | "baseUrl" | "_unstable_getRTCTransports"
|
|
> &
|
|
OpenIDClientParts;
|
|
// Used by the jwt service to create the livekit room and compute the livekit alias.
|
|
roomId: string;
|
|
useOldestMember: boolean;
|
|
forceJwtEndpoint: JwtEndpointVersion;
|
|
delayId$: Behavior<string | null>;
|
|
}
|
|
|
|
export enum JwtEndpointVersion {
|
|
Legacy = "legacy",
|
|
Matrix_2_0 = "matrix_2_0",
|
|
}
|
|
|
|
// TODO livekit_alias-cleanup
|
|
// 1. We need to move away from transports map to connections!!!
|
|
//
|
|
// 2. We need to stop sending livekit_alias all together
|
|
//
|
|
//
|
|
// 1.
|
|
// Transports are just the jwt service adress but do not contain the information which room on this transport to use.
|
|
// That requires slot and roomId.
|
|
//
|
|
// We need one connection per room on the transport.
|
|
//
|
|
// We need an object that contains:
|
|
// transport
|
|
// roomId
|
|
// slotId
|
|
//
|
|
// To map to the connections. Prosposal: `ConnectionIdentifier`
|
|
//
|
|
// 2.
|
|
// We need to make sure we do not sent livekit_alias in sticky events and that we drop all code for sending state events!
|
|
export interface LocalTransportWithSFUConfig {
|
|
transport: LivekitTransportConfig;
|
|
sfuConfig: SFUConfig;
|
|
}
|
|
|
|
export function isLocalTransportWithSFUConfig(
|
|
obj: LivekitTransportConfig | LocalTransportWithSFUConfig,
|
|
): obj is LocalTransportWithSFUConfig {
|
|
return "transport" in obj && "sfuConfig" in obj;
|
|
}
|
|
|
|
interface LocalTransport {
|
|
/**
|
|
* The transport to be advertised in our MatrixRTC membership. `null` when not
|
|
* yet fetched/validated.
|
|
*/
|
|
advertised$: Behavior<LivekitTransportConfig | null>;
|
|
/**
|
|
* The transport to connect to and publish media on. `null` when not yet known
|
|
* or available.
|
|
*/
|
|
active$: Behavior<LocalTransportWithSFUConfig | null>;
|
|
}
|
|
|
|
/**
|
|
* Connects to the JWT service and determines the transports that the local member should use.
|
|
*
|
|
* @prop useOldestMember Whether to use the same transport as the oldest member.
|
|
* This will only update once the first oldest member appears. Will not recompute if the oldest member leaves.
|
|
* @prop useOldJwtEndpoint Whether to set forceOldJwtEndpoint on the returned transport and to use the old JWT endpoint.
|
|
* This is used when the connection manager needs to know if it has to use the legacy endpoint which implies a string concatenated rtcBackendIdentity.
|
|
* (which is expected for non sticky event based rtc member events)
|
|
* @returns The transport to advertise in the local MatrixRTC membership, along with the transport to actively publish media to.
|
|
* @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
|
|
*/
|
|
export const createLocalTransport$ = ({
|
|
scope,
|
|
memberships$,
|
|
ownMembershipIdentity,
|
|
client,
|
|
roomId,
|
|
useOldestMember,
|
|
forceJwtEndpoint,
|
|
delayId$,
|
|
}: Props): LocalTransport => {
|
|
/**
|
|
* The LiveKit transport in use by the oldest RTC membership. `null` when the
|
|
* oldest member has no such transport.
|
|
*/
|
|
const oldestMemberTransport$ = scope.behavior<LivekitTransportConfig | null>(
|
|
memberships$.pipe(
|
|
map((memberships) => {
|
|
const oldestMember = memberships.value[0];
|
|
if (oldestMember === undefined) {
|
|
logger.info("Oldest member: not found");
|
|
return null;
|
|
}
|
|
const transport = oldestMember.getTransport(oldestMember);
|
|
if (transport === undefined) {
|
|
logger.warn(
|
|
`Oldest member: ${oldestMember.userId}|${oldestMember.deviceId}|${oldestMember.memberId} has no transport`,
|
|
);
|
|
return null;
|
|
}
|
|
if (!isLivekitTransportConfig(transport)) {
|
|
logger.warn(
|
|
`Oldest member: ${oldestMember.userId}|${oldestMember.deviceId}|${oldestMember.memberId} has invalid transport`,
|
|
);
|
|
return null;
|
|
}
|
|
logger.info(
|
|
"Oldest member: ${oldestMember.userId}|${oldestMember.deviceId}|${oldestMember.memberId} has valid transport",
|
|
);
|
|
return transport;
|
|
}),
|
|
distinctUntilChanged(areLivekitTransportsEqual),
|
|
),
|
|
);
|
|
|
|
/**
|
|
* The transport that we would personally prefer to publish on (if not for the
|
|
* transport preferences of others, perhaps). `null` until fetched and
|
|
* validated.
|
|
*
|
|
* @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
|
|
*/
|
|
const preferredTransport$ =
|
|
scope.behavior<LocalTransportWithSFUConfig | null>(
|
|
// preferredTransport$ (used for multi sfu) needs to know if we are using the old or new
|
|
// jwt endpoint (`get_token` vs `sfu/get`) based on that the jwt endpoint will compute the rtcBackendIdentity
|
|
// differently. (sha(`${userId}|${deviceId}|${memberId}`) vs `${userId}|${deviceId}|${memberId}`)
|
|
// When using sticky events (we need to use the new endpoint).
|
|
customLivekitUrl.value$.pipe(
|
|
switchMap((customUrl) =>
|
|
startWith<LocalTransportWithSFUConfig | null>(null)(
|
|
// Fetch the SFU config, and repeat this asynchronously for every
|
|
// change in delay ID.
|
|
delayId$.pipe(
|
|
switchMap(async (delayId) => {
|
|
logger.info(
|
|
"Creating preferred transport based on: ",
|
|
"customUrl: ",
|
|
customUrl,
|
|
"delayId: ",
|
|
delayId,
|
|
"forceJwtEndpoint: ",
|
|
forceJwtEndpoint,
|
|
);
|
|
return makeTransport(
|
|
client,
|
|
ownMembershipIdentity,
|
|
roomId,
|
|
customUrl,
|
|
forceJwtEndpoint,
|
|
delayId ?? undefined,
|
|
);
|
|
}),
|
|
// We deliberately hide any changes to the SFU config because we
|
|
// do not actually want the app to reconnect whenever the JWT
|
|
// token changes due to us delegating a new delayed event. The
|
|
// initial SFU config for the transport is all the app needs.
|
|
distinctUntilChanged((prev, next) =>
|
|
areLivekitTransportsEqual(prev.transport, next.transport),
|
|
),
|
|
),
|
|
),
|
|
),
|
|
),
|
|
);
|
|
|
|
if (useOldestMember) {
|
|
// --- Oldest member mode ---
|
|
return {
|
|
// Never update the transport that we advertise in our membership. Just
|
|
// take the first valid oldest member or preferred transport that we learn
|
|
// about, and stick with that. This avoids unnecessary SFU hops and room
|
|
// state changes.
|
|
advertised$: scope.behavior(
|
|
merge(
|
|
oldestMemberTransport$,
|
|
preferredTransport$.pipe(map((t) => t?.transport ?? null)),
|
|
).pipe(
|
|
first((t) => t !== null),
|
|
tap((t) =>
|
|
logger.info(`Advertise transport: ${t.livekit_service_url}`),
|
|
),
|
|
),
|
|
null,
|
|
),
|
|
// Publish on the transport used by the oldest member.
|
|
active$: scope.behavior(
|
|
oldestMemberTransport$.pipe(
|
|
switchMap((transport) => {
|
|
// Oldest member not available (or invalid SFU config).
|
|
if (transport === null) return of(null);
|
|
// Oldest member available: fetch the SFU config.
|
|
const fetchOldestMemberTransport =
|
|
async (): Promise<LocalTransportWithSFUConfig> => ({
|
|
transport,
|
|
sfuConfig: await getSFUConfigWithOpenID(
|
|
client,
|
|
ownMembershipIdentity,
|
|
transport.livekit_service_url,
|
|
roomId,
|
|
{ forceJwtEndpoint: JwtEndpointVersion.Legacy },
|
|
logger,
|
|
),
|
|
});
|
|
return from(fetchOldestMemberTransport()).pipe(startWith(null));
|
|
}),
|
|
tap((t) =>
|
|
logger.info(
|
|
`Publish on transport: ${t?.transport.livekit_service_url}`,
|
|
),
|
|
),
|
|
),
|
|
),
|
|
};
|
|
}
|
|
|
|
// --- Multi-SFU mode ---
|
|
// Always publish on and advertise the preferred transport.
|
|
return {
|
|
advertised$: scope.behavior(
|
|
preferredTransport$.pipe(
|
|
map((t) => t?.transport ?? null),
|
|
distinctUntilChanged(areLivekitTransportsEqual),
|
|
),
|
|
),
|
|
active$: preferredTransport$,
|
|
};
|
|
};
|
|
|
|
const FOCI_WK_KEY = "org.matrix.msc4143.rtc_foci";
|
|
|
|
/**
|
|
* Determine the correct Transport for the current session, including
|
|
* validating auth against the service to ensure it's correct.
|
|
* Prefers in order:
|
|
*
|
|
|
|
* 1. The `urlFromDevSettings` value. If this cannot be validated, the function will throw.
|
|
* 2. The transports returned via the homeserver.
|
|
* 3. The transports returned via .well-known.
|
|
* 4. The transport configured in Element Call's config.
|
|
*
|
|
* @param client The authenticated Matrix client for the current user
|
|
* @param membership The membership identity of the user.
|
|
* @param roomId The ID of the room to be connected to.
|
|
* @param urlFromDevSettings Override URL provided by the user's local config.
|
|
* @param forceJwtEndpoint Whether to force a specific JWT endpoint
|
|
* - `Legacy` / `Matrix_2_0`
|
|
* - `get_token` / `sfu/get`
|
|
* - not hashing / hashing the backendIdentity
|
|
* @param delayId the delay id passed to the jwt service.
|
|
*
|
|
* @returns A fully validated transport config.
|
|
* @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
|
|
*/
|
|
async function makeTransport(
|
|
client: Pick<
|
|
MatrixClient,
|
|
"getDomain" | "baseUrl" | "_unstable_getRTCTransports"
|
|
> &
|
|
OpenIDClientParts,
|
|
membership: CallMembershipIdentityParts,
|
|
roomId: string,
|
|
urlFromDevSettings: string | null,
|
|
forceJwtEndpoint: JwtEndpointVersion,
|
|
delayId?: string,
|
|
): Promise<LocalTransportWithSFUConfig> {
|
|
logger.trace("Searching for a preferred transport");
|
|
|
|
async function doOpenIdAndJWTFromUrl(
|
|
url: string,
|
|
): Promise<LocalTransportWithSFUConfig> {
|
|
const sfuConfig = await getSFUConfigWithOpenID(
|
|
client,
|
|
membership,
|
|
url,
|
|
roomId,
|
|
{
|
|
forceJwtEndpoint: forceJwtEndpoint,
|
|
delayEndpointBaseUrl: client.baseUrl,
|
|
delayId,
|
|
},
|
|
logger,
|
|
);
|
|
return {
|
|
transport: {
|
|
type: "livekit",
|
|
livekit_service_url: url,
|
|
},
|
|
sfuConfig,
|
|
};
|
|
}
|
|
// We will call `getSFUConfigWithOpenID` once per transport here as it's our
|
|
// only mechanism of valiation. 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.
|
|
//
|
|
// If the OpenID request were to fail then it's acceptable for us to fail
|
|
// this function early, as we assume the homeserver has got some problems.
|
|
|
|
// DEVTOOL: Highest priority: Load from devtool setting
|
|
if (urlFromDevSettings !== null) {
|
|
// Validate that the SFU is up. Otherwise, we want to fail on this
|
|
// as we don't permit other SFUs.
|
|
// This will call the jwt/sfu/get endpoint to pre create the livekit room.
|
|
logger.info("Using LiveKit transport from dev tools: ", urlFromDevSettings);
|
|
return await doOpenIdAndJWTFromUrl(urlFromDevSettings);
|
|
}
|
|
|
|
async function getFirstUsableTransport(
|
|
transports: Transport[],
|
|
): Promise<LocalTransportWithSFUConfig | null> {
|
|
for (const potentialTransport of transports) {
|
|
if (isLivekitTransportConfig(potentialTransport)) {
|
|
try {
|
|
// This will call the jwt/sfu/get endpoint to pre create the livekit room.
|
|
return await doOpenIdAndJWTFromUrl(
|
|
potentialTransport.livekit_service_url,
|
|
);
|
|
} catch (ex) {
|
|
// Explictly throw these
|
|
if (ex instanceof FailToGetOpenIdToken) {
|
|
throw ex;
|
|
}
|
|
if (ex instanceof NoMatrix2AuthorizationService) {
|
|
throw ex;
|
|
}
|
|
logger.debug(
|
|
`Could not use SFU service "${potentialTransport.livekit_service_url}" as SFU`,
|
|
ex,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
|
|
// MSC4143: Attempt to fetch transports from backend.
|
|
if ("_unstable_getRTCTransports" in client) {
|
|
try {
|
|
const transportList = await client._unstable_getRTCTransports();
|
|
const selectedTransport = await getFirstUsableTransport(transportList);
|
|
if (selectedTransport) {
|
|
logger.info(
|
|
"Using backend-configured (client.getRTCTransports) SFU",
|
|
selectedTransport,
|
|
);
|
|
return selectedTransport;
|
|
}
|
|
} catch (ex) {
|
|
if (ex instanceof MatrixError && ex.httpStatus === 404) {
|
|
// Expected, this is an unstable endpoint and it's not required.
|
|
// There will be expected 404 errors in the console. When we check if synapse supports the endpoint.
|
|
logger.debug(
|
|
"Matrix homeserver does not provide any RTC transports via `/rtc/transports` (will retry with well-known.)",
|
|
);
|
|
} else if (ex instanceof FailToGetOpenIdToken) {
|
|
throw ex;
|
|
} else {
|
|
// We got an error that wasn't just missing support for the feature, so log it loudly.
|
|
logger.error(
|
|
"Unexpected error fetching RTC transports from backend",
|
|
ex,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Legacy MSC4143 (to be removed) WELL_KNOWN: Prioritize the .well-known/matrix/client, if available.
|
|
const domain = client.getDomain();
|
|
if (domain) {
|
|
// we use AutoDiscovery instead of relying on the MatrixClient having already
|
|
// been fully configured and started
|
|
const wellKnownFoci = (await AutoDiscovery.getRawClientConfig(domain))?.[
|
|
FOCI_WK_KEY
|
|
];
|
|
const selectedTransport = Array.isArray(wellKnownFoci)
|
|
? await getFirstUsableTransport(wellKnownFoci)
|
|
: null;
|
|
if (selectedTransport) {
|
|
logger.info("Using .well-known SFU", selectedTransport);
|
|
return selectedTransport;
|
|
}
|
|
}
|
|
|
|
// CONFIG: Least prioritized; Load from config file
|
|
const urlFromConf = Config.get().livekit?.livekit_service_url;
|
|
if (urlFromConf) {
|
|
try {
|
|
// This will call the jwt/sfu/get endpoint to pre create the livekit room.
|
|
logger.info("Using config SFU", urlFromConf);
|
|
return await doOpenIdAndJWTFromUrl(urlFromConf);
|
|
} catch (ex) {
|
|
if (ex instanceof FailToGetOpenIdToken) {
|
|
throw ex;
|
|
}
|
|
logger.error("Failed to validate config SFU", ex);
|
|
}
|
|
}
|
|
|
|
// If we do not have returned a transport by now we throw an error
|
|
throw new MatrixRTCTransportMissingError(domain ?? "");
|
|
}
|