From 5f90e7d535b9866641f93b5516fef7b905d27055 Mon Sep 17 00:00:00 2001 From: Hampus Date: Thu, 3 Sep 2026 04:47:02 +0200 Subject: [PATCH] fix(api): downgrade oversized video instead of ending the call (#2419) --- .../infrastructure/DisabledLiveKitService.ts | 30 ++++ .../src/api/infrastructure/ILiveKitService.ts | 26 +++ .../src/api/infrastructure/LiveKitService.ts | 107 ++++++++++++ .../infrastructure/LiveKitWebhookService.ts | 33 +++- .../tests/LiveKitService.test.ts | 159 +++++++++++++++++- 5 files changed, 345 insertions(+), 10 deletions(-) diff --git a/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts b/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts index d3b6d8738..923f2f9e3 100644 --- a/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts +++ b/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts @@ -1,5 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-or-later +import type {TrackSource} from 'livekit-server-sdk'; import type {ChannelID, GuildID, UserID} from '../BrandedTypes'; import type {VoiceRegionMetadata, VoiceServerRecord} from '../voice/VoiceModel'; import type {ILiveKitService, ListActiveRoomsResult, ListParticipantsResult} from './ILiveKitService'; @@ -51,6 +52,27 @@ interface UpdateParticipantPermissionsParams { deaf?: boolean; } +interface MuteParticipantTrackParams { + userId: UserID; + guildId?: GuildID; + channelId: ChannelID; + connectionId: string; + regionId: string; + serverId: string; + trackSid: string; + muted: boolean; +} + +interface RevokeParticipantPublishSourceParams { + userId: UserID; + guildId?: GuildID; + channelId: ChannelID; + connectionId: string; + regionId: string; + serverId: string; + source: TrackSource; +} + export class DisabledLiveKitService implements ILiveKitService { async createToken(_params: CreateTokenParams): Promise<{ token: string; @@ -65,6 +87,14 @@ export class DisabledLiveKitService implements ILiveKitService { async disconnectParticipant(_params: DisconnectParticipantParams): Promise {} + async muteParticipantTrack(_params: MuteParticipantTrackParams): Promise { + return false; + } + + async revokeParticipantPublishSource(_params: RevokeParticipantPublishSourceParams): Promise { + return false; + } + async listParticipants(_params: { guildId?: GuildID; channelId: ChannelID; diff --git a/fluxer_api/src/api/infrastructure/ILiveKitService.ts b/fluxer_api/src/api/infrastructure/ILiveKitService.ts index 9353f3c07..637c58e12 100644 --- a/fluxer_api/src/api/infrastructure/ILiveKitService.ts +++ b/fluxer_api/src/api/infrastructure/ILiveKitService.ts @@ -1,5 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-or-later +import type {TrackSource} from 'livekit-server-sdk'; import type {ChannelID, GuildID, UserID} from '../BrandedTypes'; import type {VoiceRegionMetadata, VoiceServerRecord} from '../voice/VoiceModel'; @@ -54,6 +55,27 @@ interface DisconnectParticipantParams { serverId: string; } +interface MuteParticipantTrackParams { + userId: UserID; + guildId?: GuildID; + channelId: ChannelID; + connectionId: string; + regionId: string; + serverId: string; + trackSid: string; + muted: boolean; +} + +interface RevokeParticipantPublishSourceParams { + userId: UserID; + guildId?: GuildID; + channelId: ChannelID; + connectionId: string; + regionId: string; + serverId: string; + source: TrackSource; +} + interface ListParticipantsParams { guildId?: GuildID; channelId: ChannelID; @@ -109,6 +131,10 @@ export abstract class ILiveKitService { abstract disconnectParticipant(params: DisconnectParticipantParams): Promise; + abstract muteParticipantTrack(params: MuteParticipantTrackParams): Promise; + + abstract revokeParticipantPublishSource(params: RevokeParticipantPublishSourceParams): Promise; + abstract listParticipants(params: ListParticipantsParams): Promise; abstract listActiveRooms(): Promise; diff --git a/fluxer_api/src/api/infrastructure/LiveKitService.ts b/fluxer_api/src/api/infrastructure/LiveKitService.ts index 597915364..f3f814b8e 100644 --- a/fluxer_api/src/api/infrastructure/LiveKitService.ts +++ b/fluxer_api/src/api/infrastructure/LiveKitService.ts @@ -60,6 +60,27 @@ interface UpdateParticipantPermissionsParams { deaf?: boolean; } +interface MuteParticipantTrackParams { + userId: UserID; + guildId?: GuildID; + channelId: ChannelID; + connectionId: string; + regionId: string; + serverId: string; + trackSid: string; + muted: boolean; +} + +interface RevokeParticipantPublishSourceParams { + userId: UserID; + guildId?: GuildID; + channelId: ChannelID; + connectionId: string; + regionId: string; + serverId: string; + source: TrackSource; +} + interface ServerClientConfig { endpoint: string; apiKey: string; @@ -76,6 +97,26 @@ interface LiveKitPublishPermissions { export const VOICE_TOKEN_TTL_SECONDS = 60 * 10; +const ALL_PUBLISH_SOURCES: ReadonlyArray = [ + TrackSource.MICROPHONE, + TrackSource.CAMERA, + TrackSource.SCREEN_SHARE, + TrackSource.SCREEN_SHARE_AUDIO, +]; + +interface LiveKitPublishGrant { + canPublish: boolean; + canPublishSources: Array; +} + +export function computeRevokedPublishGrant(current: LiveKitPublishGrant, source: TrackSource): LiveKitPublishGrant { + const allowed = current.canPublishSources.length > 0 ? current.canPublishSources : ALL_PUBLISH_SOURCES; + const revoked = + source === TrackSource.SCREEN_SHARE ? [TrackSource.SCREEN_SHARE, TrackSource.SCREEN_SHARE_AUDIO] : [source]; + const canPublishSources = allowed.filter((allowedSource) => !revoked.includes(allowedSource)); + return {canPublish: current.canPublish && canPublishSources.length > 0, canPublishSources}; +} + export function computeLiveKitPublishSources(permissions: LiveKitPublishPermissions): Array { const sources: Array = []; if (permissions.canSpeak) { @@ -300,6 +341,72 @@ export class LiveKitService extends ILiveKitService { } } + async muteParticipantTrack(params: MuteParticipantTrackParams): Promise { + const {userId, guildId, channelId, connectionId, regionId, serverId, trackSid, muted} = params; + const roomName = this.getRoomName(guildId, channelId); + const participantIdentity = this.getParticipantIdentity(userId, connectionId); + const server = this.tryResolveServerClient(regionId, serverId); + if (server === null) { + Logger.debug( + {regionId, serverId, participantIdentity, roomName, trackSid}, + 'LiveKit track mute skipped, pinned server no longer exists in topology', + ); + return false; + } + try { + await server.roomServiceClient.mutePublishedTrack(roomName, participantIdentity, trackSid, muted); + return true; + } catch (error) { + if (LiveKitService.isHttp404(error)) { + Logger.debug({participantIdentity, roomName, trackSid}, 'LiveKit track no longer published, nothing to mute'); + return false; + } + Logger.error({error, participantIdentity, roomName, trackSid, muted}, 'Error muting LiveKit published track'); + return false; + } + } + + async revokeParticipantPublishSource(params: RevokeParticipantPublishSourceParams): Promise { + const {userId, guildId, channelId, connectionId, regionId, serverId, source} = params; + const roomName = this.getRoomName(guildId, channelId); + const participantIdentity = this.getParticipantIdentity(userId, connectionId); + const server = this.tryResolveServerClient(regionId, serverId); + if (server === null) { + Logger.debug( + {regionId, serverId, participantIdentity, roomName, source}, + 'LiveKit publish source revoke skipped, pinned server no longer exists in topology', + ); + return false; + } + try { + const participants = await server.roomServiceClient.listParticipants(roomName); + const participant = participants.find((p) => p.identity === participantIdentity); + if (!participant?.permission) { + Logger.debug( + {participantIdentity, roomName, source}, + 'LiveKit participant no longer in room, nothing to revoke', + ); + return false; + } + const grant = computeRevokedPublishGrant(participant.permission, source); + await server.roomServiceClient.updateParticipant(roomName, participantIdentity, undefined, { + ...participant.permission, + ...grant, + }); + return true; + } catch (error) { + if (LiveKitService.isHttp404(error)) { + Logger.debug( + {participantIdentity, roomName, source}, + 'LiveKit participant no longer in room, nothing to revoke', + ); + return false; + } + Logger.error({error, participantIdentity, roomName, source}, 'Error revoking LiveKit publish source'); + return false; + } + } + async listParticipants(params: { guildId?: GuildID; channelId: ChannelID; diff --git a/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts b/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts index ec97e4a52..33147a37c 100644 --- a/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts +++ b/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts @@ -533,11 +533,18 @@ export class LiveKitWebhookService { height: track.height, trackSource: trackSourceLabel, }, - 'User without higher video quality entitlement attempted to publish video exceeding free tier limits - disconnecting', + 'User without higher video quality entitlement published video exceeding free tier limits - muting track and revoking source', ); + if (!track.sid) { + Logger.warn( + {userId: userId.toString(), roomName: room.name, trackSource: trackSourceLabel}, + 'Track published without a sid, cannot enforce free tier video limits', + ); + return; + } const roomContext = parseRoomName(room.name); if (!roomContext) { - Logger.warn({roomName: room.name}, 'Unknown room name format, cannot disconnect'); + Logger.warn({roomName: room.name}, 'Unknown room name format, cannot enforce free tier video limits'); return; } let regionId: string | undefined; @@ -567,7 +574,7 @@ export class LiveKitWebhookService { if (!regionId || !serverId) { Logger.warn( {participantId: participant.identity, roomName: room.name, apiKey}, - 'Missing region or server info, cannot disconnect', + 'Missing region or server info, cannot enforce free tier video limits', ); return; } @@ -585,21 +592,26 @@ export class LiveKitWebhookService { height: track.height, trackSource: trackSourceLabel, }, - 'Disconnecting user without higher video quality entitlement for exceeding video quality limits', + 'Muting oversized track and revoking its publish source for user without higher video quality entitlement', ); - await this.liveKitService.disconnectParticipant({ + const muted = await this.liveKitService.muteParticipantTrack({ userId, guildId, channelId: roomContext.channelId, connectionId, regionId, serverId, + trackSid: track.sid, + muted: true, }); - await this.gatewayService.disconnectVoiceUserIfInChannel({ + const revoked = await this.liveKitService.revokeParticipantPublishSource({ + userId, guildId, channelId: roomContext.channelId, - userId, connectionId, + regionId, + serverId, + source: track.source, }); Logger.info( { @@ -611,8 +623,13 @@ export class LiveKitWebhookService { width: track.width, height: track.height, trackSource: trackSourceLabel, + trackSid: track.sid, + muted, + revoked, }, - 'Disconnected user without higher video quality entitlement for exceeding video quality limits', + muted || revoked + ? 'Enforced free tier video limits on user without higher video quality entitlement' + : 'Failed to enforce free tier video limits on user without higher video quality entitlement', ); } catch (error) { Logger.error({error}, 'Error processing track_published event'); diff --git a/fluxer_api/src/api/infrastructure/tests/LiveKitService.test.ts b/fluxer_api/src/api/infrastructure/tests/LiveKitService.test.ts index 0c1ba01e5..2e8902f6a 100644 --- a/fluxer_api/src/api/infrastructure/tests/LiveKitService.test.ts +++ b/fluxer_api/src/api/infrastructure/tests/LiveKitService.test.ts @@ -1,8 +1,19 @@ // SPDX-License-Identifier: AGPL-3.0-or-later +import type {WebhookEvent} from 'livekit-server-sdk'; import {AccessToken, TrackSource} from 'livekit-server-sdk'; -import {describe, expect, it} from 'vitest'; -import {computeLiveKitPublishSources, VOICE_TOKEN_TTL_SECONDS} from '../LiveKitService'; +import {describe, expect, it, vi} from 'vitest'; +import {createUserID} from '../../BrandedTypes'; +import {getConfig} from '../../Config'; +import type {LimitConfigService} from '../../limits/LimitConfigService'; +import type {User} from '../../models/User'; +import type {IUserRepository} from '../../user/IUserRepository'; +import type {VoiceTopology} from '../../voice/VoiceTopology'; +import type {IGatewayService} from '../IGatewayService'; +import type {ILiveKitService} from '../ILiveKitService'; +import type {IVoiceRoomStore} from '../IVoiceRoomStore'; +import {computeLiveKitPublishSources, computeRevokedPublishGrant, VOICE_TOKEN_TTL_SECONDS} from '../LiveKitService'; +import {LiveKitWebhookService} from '../LiveKitWebhookService'; function decodeJwtPayload(token: string): Record { const [, payload] = token.split('.'); @@ -44,6 +55,45 @@ describe('LiveKitService publish permissions', () => { canPublishSources: ['microphone', 'screen_share', 'screen_share_audio'], }); }); + it('revokes only the offending source and keeps the rest of the grant', () => { + expect( + computeRevokedPublishGrant( + { + canPublish: true, + canPublishSources: [ + TrackSource.MICROPHONE, + TrackSource.CAMERA, + TrackSource.SCREEN_SHARE, + TrackSource.SCREEN_SHARE_AUDIO, + ], + }, + TrackSource.CAMERA, + ), + ).toEqual({ + canPublish: true, + canPublishSources: [TrackSource.MICROPHONE, TrackSource.SCREEN_SHARE, TrackSource.SCREEN_SHARE_AUDIO], + }); + }); + it('revokes screen share audio together with screen share video', () => { + expect( + computeRevokedPublishGrant( + { + canPublish: true, + canPublishSources: [TrackSource.MICROPHONE, TrackSource.SCREEN_SHARE, TrackSource.SCREEN_SHARE_AUDIO], + }, + TrackSource.SCREEN_SHARE, + ), + ).toEqual({canPublish: true, canPublishSources: [TrackSource.MICROPHONE]}); + }); + it('treats an empty source list as every source and never leaves it empty', () => { + expect(computeRevokedPublishGrant({canPublish: true, canPublishSources: []}, TrackSource.CAMERA)).toEqual({ + canPublish: true, + canPublishSources: [TrackSource.MICROPHONE, TrackSource.SCREEN_SHARE, TrackSource.SCREEN_SHARE_AUDIO], + }); + expect( + computeRevokedPublishGrant({canPublish: true, canPublishSources: [TrackSource.CAMERA]}, TrackSource.CAMERA), + ).toEqual({canPublish: false, canPublishSources: []}); + }); it('bounds voice token lifetime to the configured TTL', async () => { const token = new AccessToken('test-key', 'test-secret', { identity: 'user_1_conn', @@ -57,3 +107,108 @@ describe('LiveKitService publish permissions', () => { expect(exp - nowSeconds).toBeGreaterThan(0); }); }); + +function createFreeUser(): User { + return { + id: createUserID(1n), + isBot: false, + premiumType: null, + premiumUntil: null, + premiumGiftExtensionEndsAt: null, + premiumWillCancel: false, + premiumGraceEndsAt: null, + flags: 0n, + premiumFlags: 0, + traits: new Set(), + } as unknown as User; +} + +function createTrackPublishedEvent(width: number, height: number): WebhookEvent { + return { + event: 'track_published', + room: {name: 'guild_2_channel_3'}, + participant: {identity: 'user_1_conn'}, + track: {type: 1, source: TrackSource.CAMERA, sid: 'TR_oversized', width, height}, + } as unknown as WebhookEvent; +} + +function createWebhookHarness() { + const muteParticipantTrack = vi.fn().mockResolvedValue(true); + const revokeParticipantPublishSource = vi.fn().mockResolvedValue(true); + const disconnectParticipant = vi.fn().mockResolvedValue(undefined); + const disconnectVoiceUserIfInChannel = vi.fn().mockResolvedValue(undefined); + const service = new LiveKitWebhookService( + { + getPinnedRoomServer: vi.fn().mockResolvedValue({regionId: 'region-1', serverId: 'region-1-server-1'}), + } as unknown as IVoiceRoomStore, + {disconnectVoiceUserIfInChannel} as unknown as IGatewayService, + {findUnique: vi.fn().mockResolvedValue(createFreeUser())} as unknown as IUserRepository, + {muteParticipantTrack, revokeParticipantPublishSource, disconnectParticipant} as unknown as ILiveKitService, + { + getAllRegions: () => [], + getServersForRegion: () => [], + registerSubscriber: () => {}, + } as unknown as VoiceTopology, + {getConfigSnapshot: () => null} as unknown as LimitConfigService, + ); + return { + service, + muteParticipantTrack, + revokeParticipantPublishSource, + disconnectParticipant, + disconnectVoiceUserIfInChannel, + }; +} + +describe('LiveKit free tier video resolution enforcement', () => { + it('mutes the oversized track and revokes its source instead of ending the call', async () => { + const { + service, + muteParticipantTrack, + revokeParticipantPublishSource, + disconnectParticipant, + disconnectVoiceUserIfInChannel, + } = createWebhookHarness(); + + await service.handleTrackPublished(createTrackPublishedEvent(1920, 1080), 'api-key'); + + expect(muteParticipantTrack).toHaveBeenCalledTimes(1); + expect(muteParticipantTrack).toHaveBeenCalledWith( + expect.objectContaining({trackSid: 'TR_oversized', muted: true, regionId: 'region-1'}), + ); + expect(revokeParticipantPublishSource).toHaveBeenCalledTimes(1); + expect(revokeParticipantPublishSource).toHaveBeenCalledWith( + expect.objectContaining({source: TrackSource.CAMERA, connectionId: 'conn', regionId: 'region-1'}), + ); + expect(disconnectParticipant).not.toHaveBeenCalled(); + expect(disconnectVoiceUserIfInChannel).not.toHaveBeenCalled(); + }); + + it('leaves tracks within the free tier limits alone', async () => { + const {service, muteParticipantTrack, revokeParticipantPublishSource, disconnectParticipant} = + createWebhookHarness(); + + await service.handleTrackPublished(createTrackPublishedEvent(1280, 720), 'api-key'); + + expect(muteParticipantTrack).not.toHaveBeenCalled(); + expect(revokeParticipantPublishSource).not.toHaveBeenCalled(); + expect(disconnectParticipant).not.toHaveBeenCalled(); + }); + + it('does not enforce resolution limits on self-hosted instances', async () => { + const {service, muteParticipantTrack, revokeParticipantPublishSource, disconnectParticipant} = + createWebhookHarness(); + const config = getConfig(); + const originalSelfHosted = config.instance.selfHosted; + config.instance.selfHosted = true; + try { + await service.handleTrackPublished(createTrackPublishedEvent(3840, 2160), 'api-key'); + } finally { + config.instance.selfHosted = originalSelfHosted; + } + + expect(muteParticipantTrack).not.toHaveBeenCalled(); + expect(revokeParticipantPublishSource).not.toHaveBeenCalled(); + expect(disconnectParticipant).not.toHaveBeenCalled(); + }); +});