diff --git a/apps/desktop/package.json b/apps/desktop/package.json index c7d63d30..69f6fc39 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -1,6 +1,6 @@ { "name": "@pairux/desktop", - "version": "0.9.75", + "version": "0.9.76", "private": true, "description": "PairUX Desktop - Screen sharing with remote control", "author": "PairUX Team ", diff --git a/apps/desktop/src/main/input/injector.ts b/apps/desktop/src/main/input/injector.ts index 45c0d8fc..d78e8442 100644 --- a/apps/desktop/src/main/input/injector.ts +++ b/apps/desktop/src/main/input/injector.ts @@ -14,6 +14,8 @@ import { resolveCaptureBoundsForSource } from '../capture/captureDisplay'; export type InputInjectionDiagnostics = InputDiagnostics; let injector: RemoteInputInjector | null = null; +const REJECTION_LOG_INTERVAL_MS = 5_000; +let lastRejectionLogAt = 0; function getInjector(): RemoteInputInjector { const selection = getInputBackendSelection(); @@ -29,7 +31,13 @@ function getInjector(): RemoteInputInjector { // barrier while leaving corner UI (Start button, menu bar) clickable. // Other platforms get the guest's exact coordinates. edgeMarginPx: selection.platform === 'linux' ? 1 : 0, + // This is a second, process-side guard after renderer coalescing. It also + // protects IPC callers that bypass the normal host hook. + maxEventsPerSecond: 120, onRejected: (reason, event, detail) => { + const now = Date.now(); + if (now - lastRejectionLogAt < REJECTION_LOG_INTERVAL_MS) return; + lastRejectionLogAt = now; console.warn('[InputInjector] Rejected input event', { reason, detail, @@ -44,6 +52,7 @@ function getInjector(): RemoteInputInjector { /** Test seam: drop the singleton so the next call re-selects a backend. */ export function resetInputInjector(): void { injector = null; + lastRejectionLogAt = 0; backendPrimary = null; captureSourceId = null; } diff --git a/apps/desktop/src/main/ipc/input.test.ts b/apps/desktop/src/main/ipc/input.test.ts index 33a1cb57..1eb70b34 100644 --- a/apps/desktop/src/main/ipc/input.test.ts +++ b/apps/desktop/src/main/ipc/input.test.ts @@ -207,6 +207,16 @@ describe('IPC Input Handlers', () => { expect(injectInput).toHaveBeenCalledTimes(2); expect(result).toEqual({ success: true, count: 2 }); }); + + it('caps oversized batches before they reach the main process injector', async () => { + const handler = mockIpcMainHandlers.get('input:injectBatch')!; + const event: InputEvent = { type: 'mouse', action: 'move', x: 0.5, y: 0.5 }; + + const result = await handler({}, { events: Array.from({ length: 100 }, () => event) }); + + expect(injectInput).toHaveBeenCalledTimes(64); + expect(result).toEqual({ success: true, count: 64 }); + }); }); describe('input:emergencyStop handler', () => { diff --git a/apps/desktop/src/main/ipc/input.ts b/apps/desktop/src/main/ipc/input.ts index 6e6a2fa1..1d78b7f6 100644 --- a/apps/desktop/src/main/ipc/input.ts +++ b/apps/desktop/src/main/ipc/input.ts @@ -155,12 +155,15 @@ export function registerInputHandlers(): void { return { success: true }; }); - // Batch inject multiple events (for better performance) + // Batch inject multiple events (for better performance). Keep this bounded: + // the renderer is not a trust boundary and a huge array would otherwise + // monopolize the main process before the injector's per-event limiter runs. ipcMain.handle('input:injectBatch', async (_event, args: { events: InputEvent[] }) => { - for (const event of args.events) { + const events = Array.isArray(args.events) ? args.events.slice(0, 64) : []; + for (const event of events) { await injectInput(event); } - return { success: true, count: args.events.length }; + return { success: true, count: events.length }; }); // Emergency stop - release all keys/buttons and disable injection diff --git a/apps/desktop/src/renderer/hooks/useInputInjection.test.ts b/apps/desktop/src/renderer/hooks/useInputInjection.test.ts index 24c87575..742040d1 100644 --- a/apps/desktop/src/renderer/hooks/useInputInjection.test.ts +++ b/apps/desktop/src/renderer/hooks/useInputInjection.test.ts @@ -242,7 +242,7 @@ describe('useInputInjection', () => { expect(mockElectronAPI.invoke).toHaveBeenCalledWith('input:inject', { event }); }); - it('should batch mouse move events', async () => { + it('should coalesce mouse moves to the most recent position each frame', async () => { const { result } = renderHook(() => useInputInjection({ enabled: true })); await act(async () => { @@ -266,10 +266,93 @@ describe('useInputInjection', () => { }); expect(mockElectronAPI.invoke).toHaveBeenCalledWith('input:injectBatch', { - events: expect.arrayContaining([moveEvent1, moveEvent2]), + events: [moveEvent2], }); }); + it('should coalesce wheel events without losing their total distance', async () => { + const { result } = renderHook(() => useInputInjection({ enabled: true })); + + await act(async () => { + await vi.runAllTimersAsync(); + }); + + await act(async () => { + await result.current.injectEvent({ + type: 'mouse', + action: 'scroll', + deltaX: 2, + deltaY: 3, + deltaMode: 0, + x: 0.1, + y: 0.1, + }); + await result.current.injectEvent({ + type: 'mouse', + action: 'scroll', + deltaX: 4, + deltaY: 5, + deltaMode: 0, + x: 0.2, + y: 0.2, + }); + await vi.advanceTimersByTimeAsync(20); + }); + + expect(mockElectronAPI.invoke).toHaveBeenCalledWith('input:injectBatch', { + events: [ + { + type: 'mouse', + action: 'scroll', + deltaX: 6, + deltaY: 8, + deltaMode: 0, + x: 0.2, + y: 0.2, + }, + ], + }); + }); + + it('should rate-limit click floods but always release an accepted button', async () => { + const { result } = renderHook(() => useInputInjection({ enabled: true })); + + await act(async () => { + await vi.runAllTimersAsync(); + }); + + const downEvent: InputEvent = { + type: 'mouse', + action: 'down', + button: 'left', + x: 0.5, + y: 0.5, + }; + const upEvent: InputEvent = { ...downEvent, action: 'up' }; + + await act(async () => { + await result.current.injectEvent(downEvent); + await Promise.all( + Array.from({ length: 40 }, () => + result.current.injectEvent({ + type: 'mouse', + action: 'click', + button: 'left', + x: 0.5, + y: 0.5, + }) + ) + ); + await result.current.injectEvent(upEvent); + }); + + const injected = mockElectronAPI.invoke.mock.calls.filter( + ([channel]) => channel === 'input:inject' + ); + expect(injected.length).toBeLessThan(20); + expect(injected.at(-1)).toEqual(['input:inject', { event: upEvent }]); + }); + it('should not inject when disabled', async () => { const { result } = renderHook(() => useInputInjection({ enabled: false })); diff --git a/apps/desktop/src/renderer/hooks/useInputInjection.ts b/apps/desktop/src/renderer/hooks/useInputInjection.ts index 590b78d0..b7cab44c 100644 --- a/apps/desktop/src/renderer/hooks/useInputInjection.ts +++ b/apps/desktop/src/renderer/hooks/useInputInjection.ts @@ -7,6 +7,14 @@ import { useEffect, useCallback, useRef, useState } from 'react'; import type { InputEvent } from '@pairux/shared-types'; import type { InputInjectionDiagnostics } from '../../preload/api'; +// Network-originated input must not be able to grow the host's IPC queue +// without bound. Continuous gestures are sampled once per frame, while button +// and keyboard events are rate-limited below. +const INPUT_FRAME_MS = 16; +const MAX_DISCRETE_EVENTS_PER_SECOND = 30; +const DISCRETE_EVENT_BURST = 15; +const MAX_QUEUED_INJECTIONS = 32; + interface UseInputInjectionOptions { /** * Optional declarative enablement. Omit it when the host needs to await @@ -52,13 +60,20 @@ export function useInputInjection({ const isEnabledRef = useRef(false); const [isInitialized, setIsInitialized] = useState(false); const [diagnostics, setDiagnostics] = useState(null); - const pendingEvents = useRef([]); + const pendingMove = useRef(null); + const pendingScroll = useRef | null>( + null + ); const flushTimeout = useRef | null>(null); // IPC handlers may run concurrently. Keep every OS injection in the order // it arrived, especially move -> down -> up. Without this queue, a button // down that is waiting for a pending move batch can be overtaken by its up, // leaving a mouse button held on the host desktop. const injectionQueue = useRef>(Promise.resolve()); + const queuedInjectionCount = useRef(0); + const discreteRateLimit = useRef({ tokens: DISCRETE_EVENT_BURST, updatedAt: Date.now() }); + const pressedMouseButtons = useRef(new Set()); + const pressedKeys = useRef(new Set()); const activate = useCallback(async (): Promise => { try { @@ -91,7 +106,19 @@ export function useInputInjection({ }, []); const enqueueInjection = useCallback( - (operation: () => Promise, errorMessage: string): Promise => { + ( + operation: () => Promise, + errorMessage: string, + isRequiredRelease = false + ): Promise => { + // A release for an accepted press gets a bounded exception: it + // prevents stuck input while the pressed-key/button sets limit how many + // such exceptions a sender can create. + if (queuedInjectionCount.current >= MAX_QUEUED_INJECTIONS && !isRequiredRelease) { + return Promise.resolve(); + } + + queuedInjectionCount.current += 1; const queued = injectionQueue.current.then(async () => { try { await operation(); @@ -102,12 +129,46 @@ export function useInputInjection({ // The operation catches its own error, so later input is never blocked // behind a rejected promise. - injectionQueue.current = queued; - return queued; + injectionQueue.current = queued.finally(() => { + queuedInjectionCount.current -= 1; + }); + return injectionQueue.current; }, [] ); + const consumeDiscreteEventToken = useCallback((event: InputEvent): boolean => { + const isMouseRelease = + event.type === 'mouse' && + event.action === 'up' && + pressedMouseButtons.current.delete(event.button); + const isKeyRelease = + event.type === 'keyboard' && event.action === 'up' && pressedKeys.current.delete(event.code); + + // A release for an accepted press must always get through. Otherwise rate + // limiting could leave the host with a key or button stuck down. + if (isMouseRelease || isKeyRelease) return true; + + const now = Date.now(); + const elapsedSeconds = Math.max(0, now - discreteRateLimit.current.updatedAt) / 1000; + discreteRateLimit.current.tokens = Math.min( + DISCRETE_EVENT_BURST, + discreteRateLimit.current.tokens + elapsedSeconds * MAX_DISCRETE_EVENTS_PER_SECOND + ); + discreteRateLimit.current.updatedAt = now; + + if (discreteRateLimit.current.tokens < 1) return false; + discreteRateLimit.current.tokens -= 1; + + if (event.type === 'mouse' && event.action === 'down') { + pressedMouseButtons.current.add(event.button); + } else if (event.type === 'keyboard' && event.action === 'down') { + pressedKeys.current.add(event.code); + } + + return true; + }, []); + // Initialize input injection system on mount useEffect(() => { const init = async () => { @@ -178,10 +239,12 @@ export function useInputInjection({ // Flush pending events in batch const flushEvents = useCallback(async () => { - if (pendingEvents.current.length === 0) return; - - const events = [...pendingEvents.current]; - pendingEvents.current = []; + const events = [pendingMove.current, pendingScroll.current].filter( + (event): event is InputEvent => event !== null + ); + pendingMove.current = null; + pendingScroll.current = null; + if (events.length === 0) return; await enqueueInjection( () => window.electronAPI.invoke('input:injectBatch', { events }), @@ -194,16 +257,43 @@ export function useInputInjection({ async (event: InputEvent) => { if (!isEnabledRef.current) return; - // For mouse moves, batch them up + // Intermediate mouse positions are not useful to the host. Keep only + // the latest one and sample it once per frame. if (event.type === 'mouse' && event.action === 'move') { - pendingEvents.current.push(event); + pendingMove.current = event; - // Flush after a short delay to batch moves + // Flush after a short delay to sample at most once per frame. flushTimeout.current ??= setTimeout(() => { flushTimeout.current = null; void flushEvents(); - }, 16); // ~60fps + }, INPUT_FRAME_MS); // ~60fps + } else if (event.type === 'mouse' && event.action === 'scroll') { + // Trackpads emit scrolls at a very high rate. Preserve the cumulative + // distance while turning the whole frame into one OS injection. + const existing = pendingScroll.current; + pendingScroll.current = + existing && existing.deltaMode === event.deltaMode + ? { + ...event, + deltaX: existing.deltaX + event.deltaX, + deltaY: existing.deltaY + event.deltaY, + } + : event; + + flushTimeout.current ??= setTimeout(() => { + flushTimeout.current = null; + void flushEvents(); + }, INPUT_FRAME_MS); } else { + const isRequiredRelease = + (event.type === 'mouse' && + event.action === 'up' && + pressedMouseButtons.current.has(event.button)) || + (event.type === 'keyboard' && + event.action === 'up' && + pressedKeys.current.has(event.code)); + if (!consumeDiscreteEventToken(event)) return; + // For clicks and keyboard, inject immediately // But first flush any pending moves if (flushTimeout.current) { @@ -217,13 +307,14 @@ export function useInputInjection({ const moveFlush = flushEvents(); const injection = enqueueInjection( () => window.electronAPI.invoke('input:inject', { event }), - '[useInputInjection] Failed to inject:' + '[useInputInjection] Failed to inject:', + isRequiredRelease ); await moveFlush; await injection; } }, - [flushEvents, enqueueInjection] + [flushEvents, enqueueInjection, consumeDiscreteEventToken] ); // Inject multiple events in batch @@ -254,10 +345,17 @@ export function useInputInjection({ // Cleanup on unmount useEffect(() => { + const mouseButtons = pressedMouseButtons.current; + const keys = pressedKeys.current; + return () => { if (flushTimeout.current) { clearTimeout(flushTimeout.current); } + pendingMove.current = null; + pendingScroll.current = null; + mouseButtons.clear(); + keys.clear(); // Disable injection when component unmounts if (isEnabled) { isEnabledRef.current = false; diff --git a/apps/desktop/src/renderer/hooks/useWebRTCHostAPI.ts b/apps/desktop/src/renderer/hooks/useWebRTCHostAPI.ts index 5e98aa14..7f8cce98 100644 --- a/apps/desktop/src/renderer/hooks/useWebRTCHostAPI.ts +++ b/apps/desktop/src/renderer/hooks/useWebRTCHostAPI.ts @@ -43,6 +43,8 @@ const BITRATE_PRESETS: Record = { // Stats collection and reporting interval const STATS_INTERVAL = 30000; // 30 seconds +const REJECTED_INPUT_LOG_INTERVAL_MS = 5_000; +const MAX_CONTROL_MESSAGE_BYTES = 16 * 1024; // Default ICE servers (STUN only — overridden with TURN from the SSE connected event) const DEFAULT_ICE_SERVERS: RTCIceServer[] = [ @@ -165,6 +167,7 @@ export function useWebRTCHostAPI({ // synchronous authority for this transport. const controllingViewerRef = useRef(null); const lastInputSequenceRef = useRef(new Map()); + const lastRejectedInputLogAtRef = useRef(new Map()); const getPreferredHostAudioTrack = useCallback( (streamOverride?: MediaStream | null): MediaStreamTrack | null => { @@ -353,6 +356,7 @@ export function useWebRTCHostAPI({ // Handle data channel messages const handleDataChannelMessage = useCallback((viewerId: string, event: MessageEvent) => { + if (typeof event.data !== 'string' || event.data.length > MAX_CONTROL_MESSAGE_BYTES) return; try { const message = JSON.parse(event.data) as ControlMessage | InputMessage; @@ -392,12 +396,19 @@ export function useWebRTCHostAPI({ message.sequence < 0 || (lastSequence !== undefined && message.sequence <= lastSequence) ) { - console.warn('[WebRTCHost] Dropping unauthorized or stale input', { - viewerId, - controller: controllingViewerRef.current, - sequence: message.sequence, - lastSequence: lastSequence ?? null, - }); + // Logging every rejected packet can itself make the renderer + // unresponsive when a peer floods this data channel. + const now = Date.now(); + const lastLoggedAt = lastRejectedInputLogAtRef.current.get(viewerId) ?? 0; + if (now - lastLoggedAt >= REJECTED_INPUT_LOG_INTERVAL_MS) { + lastRejectedInputLogAtRef.current.set(viewerId, now); + console.warn('[WebRTCHost] Dropping unauthorized or stale input', { + viewerId, + controller: controllingViewerRef.current, + sequence: message.sequence, + lastSequence: lastSequence ?? null, + }); + } return; } lastInputSequenceRef.current.set(viewerId, message.sequence); @@ -650,6 +661,8 @@ export function useWebRTCHostAPI({ viewer.peerConnection.close(); viewersRef.current.delete(viewerId); pendingCandidatesRef.current.delete(viewerId); + lastInputSequenceRef.current.delete(viewerId); + lastRejectedInputLogAtRef.current.delete(viewerId); setViewers(new Map(viewersRef.current)); onViewerLeft?.(viewerId); } diff --git a/apps/desktop/src/renderer/hooks/useWebRTCHostSFUAPI.ts b/apps/desktop/src/renderer/hooks/useWebRTCHostSFUAPI.ts index ca0d621a..3ae0e047 100644 --- a/apps/desktop/src/renderer/hooks/useWebRTCHostSFUAPI.ts +++ b/apps/desktop/src/renderer/hooks/useWebRTCHostSFUAPI.ts @@ -37,6 +37,8 @@ const LIVEKIT_URL = process.env.NEXT_PUBLIC_LIVEKIT_URL ?? ''; const encoder = new TextEncoder(); const decoder = new TextDecoder(); +const REJECTED_INPUT_LOG_INTERVAL_MS = 5_000; +const MAX_CONTROL_MESSAGE_BYTES = 16 * 1024; export interface ViewerConnection { id: string; @@ -139,6 +141,7 @@ export function useWebRTCHostSFUAPI({ // the handoff window. const controllingViewerRef = useRef(null); const lastInputSequenceRef = useRef(new Map()); + const lastRejectedInputLogAtRef = useRef(new Map()); // Send data to a specific participant or all const sendData = useCallback((message: unknown, targetIdentity?: string, reliable = true) => { @@ -156,6 +159,7 @@ export function useWebRTCHostSFUAPI({ // Handle data messages from viewers const handleDataReceived = useCallback((payload: Uint8Array, participant?: RemoteParticipant) => { if (!participant) return; + if (payload.byteLength > MAX_CONTROL_MESSAGE_BYTES) return; const viewerId = participant.identity; try { @@ -198,12 +202,20 @@ export function useWebRTCHostSFUAPI({ message.sequence < 0 || (lastSequence !== undefined && message.sequence <= lastSequence) ) { - console.warn('[WebRTCHostSFU] Dropping unauthorized or stale input', { - viewerId, - controller: controllingViewerRef.current, - sequence: message.sequence, - lastSequence: lastSequence ?? null, - }); + // Do not let a hostile sender freeze the renderer through its + // own console output. Keep enough telemetry to diagnose a bad + // client without logging thousands of rejected packets. + const now = Date.now(); + const lastLoggedAt = lastRejectedInputLogAtRef.current.get(viewerId) ?? 0; + if (now - lastLoggedAt >= REJECTED_INPUT_LOG_INTERVAL_MS) { + lastRejectedInputLogAtRef.current.set(viewerId, now); + console.warn('[WebRTCHostSFU] Dropping unauthorized or stale input', { + viewerId, + controller: controllingViewerRef.current, + sequence: message.sequence, + lastSequence: lastSequence ?? null, + }); + } return; } lastInputSequenceRef.current.set(viewerId, message.sequence); @@ -301,6 +313,8 @@ export function useWebRTCHostSFUAPI({ viewer?.amplifiedAudio?.dispose(); viewersRef.current.delete(identity); + lastInputSequenceRef.current.delete(identity); + lastRejectedInputLogAtRef.current.delete(identity); setViewers(new Map(viewersRef.current)); setControllingViewer((prev) => (prev === identity ? null : prev)); onViewerLeftRef.current?.(identity); diff --git a/apps/desktop/src/renderer/hooks/useWebRTCViewerSFUAPI.ts b/apps/desktop/src/renderer/hooks/useWebRTCViewerSFUAPI.ts index dd7f3e5a..25acafcc 100644 --- a/apps/desktop/src/renderer/hooks/useWebRTCViewerSFUAPI.ts +++ b/apps/desktop/src/renderer/hooks/useWebRTCViewerSFUAPI.ts @@ -223,7 +223,11 @@ export function useWebRTCViewerSFUAPI({ sequence: inputSequenceRef.current++, event, }; - sendData(message); + // Pointer motion and trackpad scroll are superseded by the next sample. + // Send them as datagrams so congestion drops stale movement instead of + // queueing it behind a flood of reliable data packets. + const isContinuous = event.type === 'mouse' && (event.action === 'move' || event.action === 'scroll'); + sendData(message, !isContinuous); }, [controlState, dataChannelReady, sendData] ); diff --git a/apps/installer/package.json b/apps/installer/package.json index 3c453dc9..8632f58a 100644 --- a/apps/installer/package.json +++ b/apps/installer/package.json @@ -1,6 +1,6 @@ { "name": "@pairux/installer", - "version": "0.9.75", + "version": "0.9.76", "private": true, "description": "PairUX Desktop App Installer Service", "type": "module", diff --git a/apps/livekit/package.json b/apps/livekit/package.json index 28b14a80..4f96ce9a 100644 --- a/apps/livekit/package.json +++ b/apps/livekit/package.json @@ -1,6 +1,6 @@ { "name": "@pairux/livekit", - "version": "0.9.75", + "version": "0.9.76", "private": true, "description": "PairUX LiveKit SFU Server", "scripts": { diff --git a/apps/turn/package.json b/apps/turn/package.json index 73328c75..036adcc2 100644 --- a/apps/turn/package.json +++ b/apps/turn/package.json @@ -1,6 +1,6 @@ { "name": "@pairux/turn", - "version": "0.9.75", + "version": "0.9.76", "private": true, "description": "PairUX TURN/STUN Server (coturn)", "scripts": { diff --git a/apps/web/package.json b/apps/web/package.json index 221c99af..80ae1547 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -1,6 +1,6 @@ { "name": "@pairux/web", - "version": "0.9.75", + "version": "0.9.76", "private": true, "type": "module", "scripts": { diff --git a/apps/web/src/app/api/chat/send/route.ts b/apps/web/src/app/api/chat/send/route.ts index d1c08647..8300a85a 100644 --- a/apps/web/src/app/api/chat/send/route.ts +++ b/apps/web/src/app/api/chat/send/route.ts @@ -1,28 +1,9 @@ import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; import { sendChatMessageSchema } from '@/lib/validations'; import { successResponse, errorResponse, handleApiError } from '@/lib/api'; +import { FixedWindowRateLimiter } from '@/lib/rate-limit'; -// Simple in-memory rate limiter (for MVP - use Redis in production) -const rateLimitMap = new Map(); -const RATE_LIMIT = 10; // messages per minute -const RATE_WINDOW = 60 * 1000; // 1 minute in milliseconds - -function checkRateLimit(key: string): boolean { - const now = Date.now(); - const entry = rateLimitMap.get(key); - - if (!entry || now > entry.resetTime) { - rateLimitMap.set(key, { count: 1, resetTime: now + RATE_WINDOW }); - return true; - } - - if (entry.count >= RATE_LIMIT) { - return false; - } - - entry.count++; - return true; -} +const messagesBySender = new FixedWindowRateLimiter(10, 60_000); // POST /api/chat/send - Send a chat message export async function POST(request: Request) { @@ -37,7 +18,7 @@ export async function POST(request: Request) { // Rate limit by user ID or participant ID const rateLimitKey = user?.id ?? participantId ?? 'anonymous'; - if (!checkRateLimit(rateLimitKey)) { + if (!messagesBySender.check(rateLimitKey).success) { return errorResponse('Rate limit exceeded. Please wait before sending more messages.', 429); } diff --git a/apps/web/src/app/api/livekit/token/route.ts b/apps/web/src/app/api/livekit/token/route.ts index c2ac4f01..0a6036c6 100644 --- a/apps/web/src/app/api/livekit/token/route.ts +++ b/apps/web/src/app/api/livekit/token/route.ts @@ -6,6 +6,13 @@ import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; import { serviceClient } from '@/lib/supabase/service'; import { successResponse, errorResponse, handleApiError } from '@/lib/api'; import { getIceServers } from '@/lib/ice-servers'; +import { FixedWindowRateLimiter, getClientIp } from '@/lib/rate-limit'; + +// A token is valid for 24 hours, so legitimate clients never need to mint +// them rapidly. These limits protect the database and LiveKit from connection +// churn while still leaving generous headroom for reconnects. +const tokenRequestsByIp = new FixedWindowRateLimiter(60, 60_000); +const tokenRequestsByParticipant = new FixedWindowRateLimiter(20, 60_000); const tokenRequestSchema = z.object({ sessionId: z.string().uuid('Invalid session ID'), @@ -16,9 +23,22 @@ const tokenRequestSchema = z.object({ export async function POST(request: Request) { try { + const ipLimit = tokenRequestsByIp.check(getClientIp(request)); + if (!ipLimit.success) { + return errorResponse(`Too many token requests. Try again in ${String(ipLimit.retryAfterSeconds)} seconds.`, 429); + } + const body: unknown = await request.json().catch(() => ({})); const { sessionId, participantName, participantId, isHost } = tokenRequestSchema.parse(body); + const participantLimit = tokenRequestsByParticipant.check(`${sessionId}:${participantId}`); + if (!participantLimit.success) { + return errorResponse( + `Too many connection attempts. Try again in ${String(participantLimit.retryAfterSeconds)} seconds.`, + 429 + ); + } + const apiKey = process.env.LIVEKIT_API_KEY; const apiSecret = process.env.LIVEKIT_API_SECRET; diff --git a/apps/web/src/app/api/sessions/[sessionId]/signal/route.ts b/apps/web/src/app/api/sessions/[sessionId]/signal/route.ts index db9c0ad8..6d4d6e06 100644 --- a/apps/web/src/app/api/sessions/[sessionId]/signal/route.ts +++ b/apps/web/src/app/api/sessions/[sessionId]/signal/route.ts @@ -1,5 +1,6 @@ import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; import { successResponse, errorResponse, handleApiError } from '@/lib/api'; +import { FixedWindowRateLimiter, getClientIp } from '@/lib/rate-limit'; import { z } from 'zod'; // Type for session (until Supabase types are regenerated) @@ -26,27 +27,11 @@ const signalSchema = z.object({ timestamp: z.number(), }); -// Simple in-memory rate limiter -const rateLimitMap = new Map(); -const RATE_LIMIT = 100; // signals per minute (higher than chat - ICE candidates can be frequent) -const RATE_WINDOW = 60 * 1000; - -function checkRateLimit(key: string): boolean { - const now = Date.now(); - const entry = rateLimitMap.get(key); - - if (!entry || now > entry.resetTime) { - rateLimitMap.set(key, { count: 1, resetTime: now + RATE_WINDOW }); - return true; - } - - if (entry.count >= RATE_LIMIT) { - return false; - } - - entry.count++; - return true; -} +// ICE candidates can legitimately arrive in bursts. Keep a generous +// participant allowance while adding an IP ceiling to stop identity rotation +// from bypassing the limiter. +const signalsBySender = new FixedWindowRateLimiter(120, 60_000); +const signalsByIp = new FixedWindowRateLimiter(300, 60_000); // POST /api/sessions/[sessionId]/signal - Send a signaling message export async function POST( @@ -61,9 +46,9 @@ export async function POST( const supabase = await createClient(); const { user } = await getAuthenticatedUser(supabase); - // Rate limit by sender ID - const rateLimitKey = signal.senderId; - if (!checkRateLimit(rateLimitKey)) { + const senderLimit = signalsBySender.check(`${sessionId}:${signal.senderId}`); + const ipLimit = signalsByIp.check(getClientIp(request)); + if (!senderLimit.success || !ipLimit.success) { return errorResponse('Rate limit exceeded', 429); } diff --git a/apps/web/src/app/api/sessions/[sessionId]/stats/route.ts b/apps/web/src/app/api/sessions/[sessionId]/stats/route.ts index b4dfce64..f284d0b8 100644 --- a/apps/web/src/app/api/sessions/[sessionId]/stats/route.ts +++ b/apps/web/src/app/api/sessions/[sessionId]/stats/route.ts @@ -1,5 +1,6 @@ import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; import { successResponse, errorResponse, handleApiError } from '@/lib/api'; +import { FixedWindowRateLimiter } from '@/lib/rate-limit'; import { z } from 'zod'; // Type for session (until Supabase types are regenerated) @@ -62,21 +63,9 @@ const statsReportSchema = z.object({ reportInterval: z.number().default(30000), }); -// Rate limit: 1 report per 10 seconds per participant -const rateLimitMap = new Map(); -const RATE_LIMIT_MS = 10000; - -function checkRateLimit(key: string): boolean { - const now = Date.now(); - const lastReport = rateLimitMap.get(key); - - if (lastReport && now - lastReport < RATE_LIMIT_MS) { - return false; - } - - rateLimitMap.set(key, now); - return true; -} +// Rate limit: 1 report per 10 seconds per participant. The shared limiter is +// capped so generated participant IDs cannot turn this into a memory leak. +const reportsByParticipant = new FixedWindowRateLimiter(1, 10_000); // POST /api/sessions/[sessionId]/stats - Report usage statistics export async function POST( @@ -93,7 +82,7 @@ export async function POST( // Rate limit by participant ID const rateLimitKey = `${sessionId}:${stats.participantId}`; - if (!checkRateLimit(rateLimitKey)) { + if (!reportsByParticipant.check(rateLimitKey).success) { return errorResponse('Rate limit exceeded - report less frequently', 429); } diff --git a/apps/web/src/app/api/sessions/join/[joinCode]/route.ts b/apps/web/src/app/api/sessions/join/[joinCode]/route.ts index 17c4feb9..130c18c3 100644 --- a/apps/web/src/app/api/sessions/join/[joinCode]/route.ts +++ b/apps/web/src/app/api/sessions/join/[joinCode]/route.ts @@ -2,15 +2,26 @@ import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; import { serviceClient } from '@/lib/supabase/service'; import { guestJoinSchema } from '@/lib/validations'; import { successResponse, errorResponse, handleApiError } from '@/lib/api'; +import { FixedWindowRateLimiter, getClientIp } from '@/lib/rate-limit'; + +// Joining performs database work and can trigger a host notification. Bound it +// by source IP and room code so a shared public link cannot be used to fan out +// an unbounded burst of work. +const joinLookupsByIp = new FixedWindowRateLimiter(120, 60_000); +const joinAttemptsByIpAndCode = new FixedWindowRateLimiter(12, 60_000); interface RouteParams { params: Promise<{ joinCode: string }>; } // GET /api/sessions/join/[joinCode] - Lookup session by join code -export async function GET(_request: Request, { params }: RouteParams) { +export async function GET(request: Request, { params }: RouteParams) { try { const { joinCode } = await params; + const lookupLimit = joinLookupsByIp.check(getClientIp(request)); + if (!lookupLimit.success) { + return errorResponse(`Too many lookup requests. Try again in ${String(lookupLimit.retryAfterSeconds)} seconds.`, 429); + } const supabase = await createClient(); // Lookup session by join code @@ -93,6 +104,10 @@ export async function GET(_request: Request, { params }: RouteParams) { export async function POST(request: Request, { params }: RouteParams) { try { const { joinCode } = await params; + const joinLimit = joinAttemptsByIpAndCode.check(`${getClientIp(request)}:${joinCode.toUpperCase()}`); + if (!joinLimit.success) { + return errorResponse(`Too many join attempts. Try again in ${String(joinLimit.retryAfterSeconds)} seconds.`, 429); + } const body = (await request.json().catch(() => ({}))) as { displayName?: string }; const supabase = await createClient(); diff --git a/apps/web/src/hooks/useWebRTCSFU.ts b/apps/web/src/hooks/useWebRTCSFU.ts index f726f83d..22c15e83 100644 --- a/apps/web/src/hooks/useWebRTCSFU.ts +++ b/apps/web/src/hooks/useWebRTCSFU.ts @@ -192,7 +192,10 @@ export function useWebRTCSFU({ sequence: inputSequenceRef.current++, event, }; - sendData(message); + // Stale motion is never useful; unreliable delivery prevents a lagging + // client from building an ever-growing reliable data backlog. + const isContinuous = event.type === 'mouse' && (event.action === 'move' || event.action === 'scroll'); + sendData(message, !isContinuous); }, [controlState, dataChannelReady, sendData] ); diff --git a/apps/web/src/lib/rate-limit.test.ts b/apps/web/src/lib/rate-limit.test.ts new file mode 100644 index 00000000..1887febe --- /dev/null +++ b/apps/web/src/lib/rate-limit.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, it } from 'vitest'; +import { FixedWindowRateLimiter, getClientIp } from './rate-limit'; + +describe('FixedWindowRateLimiter', () => { + it('enforces a per-key limit until the window expires', () => { + const limiter = new FixedWindowRateLimiter(2, 1_000); + + expect(limiter.check('viewer', 0).success).toBe(true); + expect(limiter.check('viewer', 1).success).toBe(true); + expect(limiter.check('viewer', 2)).toEqual({ success: false, retryAfterSeconds: 1 }); + expect(limiter.check('viewer', 1_000).success).toBe(true); + }); + + it('uses the first forwarded address as the client identity', () => { + expect( + getClientIp(new Request('https://pairux.com', { headers: { 'x-forwarded-for': '1.2.3.4, 5.6.7.8' } })) + ).toBe('1.2.3.4'); + }); +}); diff --git a/apps/web/src/lib/rate-limit.ts b/apps/web/src/lib/rate-limit.ts new file mode 100644 index 00000000..4ebe392e --- /dev/null +++ b/apps/web/src/lib/rate-limit.ts @@ -0,0 +1,63 @@ +/** + * Small, bounded, process-local rate limiter for abuse-sensitive API routes. + * + * It is deliberately a first line of defence: production deployments should + * also enforce equivalent limits at the CDN/WAF. Keeping this limiter bounded + * means an attacker cannot exhaust server memory just by inventing new keys. + */ +export interface RateLimitResult { + success: boolean; + retryAfterSeconds: number; +} + +interface RateLimitEntry { + count: number; + resetAt: number; +} + +export class FixedWindowRateLimiter { + private readonly entries = new Map(); + + constructor( + private readonly limit: number, + private readonly windowMs: number, + private readonly maxEntries = 10_000 + ) {} + + check(key: string, now = Date.now()): RateLimitResult { + const existing = this.entries.get(key); + + if (!existing || now >= existing.resetAt) { + this.prune(now); + this.entries.set(key, { count: 1, resetAt: now + this.windowMs }); + return { success: true, retryAfterSeconds: Math.ceil(this.windowMs / 1000) }; + } + + const retryAfterSeconds = Math.max(1, Math.ceil((existing.resetAt - now) / 1000)); + if (existing.count >= this.limit) return { success: false, retryAfterSeconds }; + + existing.count += 1; + return { success: true, retryAfterSeconds }; + } + + reset(): void { + this.entries.clear(); + } + + private prune(now: number): void { + for (const [key, entry] of this.entries) { + if (entry.resetAt <= now || this.entries.size >= this.maxEntries) this.entries.delete(key); + if (this.entries.size < this.maxEntries) break; + } + } +} + +/** Use proxy-provided client IP headers, with a conservative shared fallback. */ +export function getClientIp(request: Request): string { + const realIp = request.headers.get('x-real-ip'); + if (realIp) return realIp; + + const forwardedFor = request.headers.get('x-forwarded-for'); + return forwardedFor?.split(',', 1)[0]?.trim() || 'unknown'; +} + diff --git a/package.json b/package.json index 885ef331..1140d94a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "pairux", - "version": "0.9.75", + "version": "0.9.76", "private": true, "description": "Collaborative desktop screen sharing with remote control", "type": "module", diff --git a/packages/shared-types/package.json b/packages/shared-types/package.json index 7719646b..565cf7aa 100644 --- a/packages/shared-types/package.json +++ b/packages/shared-types/package.json @@ -1,6 +1,6 @@ { "name": "@pairux/shared-types", - "version": "0.9.75", + "version": "0.9.76", "private": true, "type": "module", "main": "./dist/index.js",