diff --git a/.env.example b/.env.example index adc737f30b1..0eefbf9f596 100644 --- a/.env.example +++ b/.env.example @@ -426,8 +426,18 @@ NODE_MAX_OLD_SPACE_SIZE=6144 # RUM_TRACE_PROPAGATION_TARGETS=https://api.example.com # Privacy defaults: replay, console capture, and full network body capture stay off. +# Opt in separately with RUM_SESSION_REPLAY_ENABLED=true (requires RUM_ENABLED=true). +# The client masks +# all text and input values and stops recording outside its authenticated layout. +# URLs, element attributes and assets may still contain sensitive information; +# validate with synthetic data before enabling recording for real users. # Console/network capture may collect sensitive browser logs, prompts, responses, or payloads. -# RUM_DISABLE_REPLAY=true +# RUM_SESSION_REPLAY_ENABLED=false +# Deprecated fallback: RUM_DISABLE_REPLAY=false opts in only when the positive flag is unset. +# YAML telemetry.rum.enabled / telemetry.rum.sessionReplay.enabled take precedence +# over environment variables. Explicit false always wins over lower-precedence opt-ins. +# Proxy mode strips replay records even when /v1/logs remains open for other log sources. +# Direct public-token ingestion bypasses this server and must enforce its own collector policy. # RUM_CONSOLE_CAPTURE=false # RUM_ADVANCED_NETWORK_CAPTURE=false # RUM_SAMPLE_RATE=1 diff --git a/api/server/routes/__tests__/config.rum.spec.js b/api/server/routes/__tests__/config.rum.spec.js index 3cebde9ae20..20bc8b73caa 100644 --- a/api/server/routes/__tests__/config.rum.spec.js +++ b/api/server/routes/__tests__/config.rum.spec.js @@ -66,6 +66,7 @@ afterEach(() => { delete process.env.RUM_TRACE_PROPAGATION_TARGETS; delete process.env.RUM_CONSOLE_CAPTURE; delete process.env.RUM_DISABLE_REPLAY; + delete process.env.RUM_SESSION_REPLAY_ENABLED; delete process.env.RUM_ADVANCED_NETWORK_CAPTURE; delete process.env.RUM_SAMPLE_RATE; delete process.env.RUM_ENVIRONMENT; @@ -73,6 +74,49 @@ afterEach(() => { }); describe('GET /api/config RUM config', () => { + it.each([null, mockUser])('serves separate YAML RUM/replay opt-ins for %p', async (user) => { + mockGetAppConfig.mockResolvedValue({ + ...baseAppConfig, + config: { + telemetry: { + rum: { enabled: true, sessionReplay: { enabled: true } }, + }, + }, + }); + process.env.RUM_ENABLED = 'false'; + process.env.RUM_AUTH_MODE = 'proxy'; + process.env.RUM_PROXY_TARGET_URL = 'http://otel-collector:4318'; + const response = await request(createApp(user)).get('/api/config'); + expect(response.body.rum).toMatchObject({ + enabled: true, + sessionReplay: { enabled: true }, + disableReplay: false, + }); + }); + + it('lets YAML disable replay while RUM and client logs stay enabled', async () => { + mockGetAppConfig.mockResolvedValue({ + ...baseAppConfig, + config: { + telemetry: { + rum: { sessionReplay: { enabled: false } }, + }, + }, + }); + process.env.RUM_ENABLED = 'true'; + process.env.RUM_SESSION_REPLAY_ENABLED = 'true'; + process.env.RUM_CLIENT_LOGS = 'true'; + process.env.RUM_AUTH_MODE = 'proxy'; + process.env.RUM_PROXY_TARGET_URL = 'http://otel-collector:4318'; + const response = await request(createApp(mockUser)).get('/api/config'); + expect(response.body.rum).toMatchObject({ + enabled: true, + sessionReplay: { enabled: false }, + disableReplay: true, + clientLogs: true, + }); + }); + it('includes public-token RUM config when enabled with valid env', async () => { mockGetAppConfig.mockResolvedValue(baseAppConfig); process.env.RUM_ENABLED = 'true'; @@ -96,6 +140,7 @@ describe('GET /api/config RUM config', () => { tracePropagationTargets: ['https://app.example.com', 'https://api.openai.com'], consoleCapture: false, disableReplay: true, + sessionReplay: { enabled: false }, advancedNetworkCapture: false, sampleRate: 0.25, environment: 'test', @@ -137,6 +182,7 @@ describe('GET /api/config RUM config', () => { authMode: 'proxy', consoleCapture: false, disableReplay: true, + sessionReplay: { enabled: false }, advancedNetworkCapture: false, clientLogs: false, }); diff --git a/api/server/routes/__tests__/rum.spec.js b/api/server/routes/__tests__/rum.spec.js index 318f62e1135..039c95000d1 100644 --- a/api/server/routes/__tests__/rum.spec.js +++ b/api/server/routes/__tests__/rum.spec.js @@ -20,7 +20,10 @@ jest.mock('~/server/middleware', () => ({ requireRumProxyAuth: (...args) => mockRequireRumProxyAuth(...args), })); +jest.mock('~/server/services/Config/app', () => ({ getAppConfig: jest.fn() })); + jest.mock('@librechat/api', () => ({ + loadRumConfig: () => (_req, _res, next) => next(), getRumProxyBodyLimit: jest.fn(() => '3mb'), limiterCache: (...args) => mockLimiterCache(...args), createRumProxyLimiter: (...args) => mockCreateRumProxyLimiter(...args), diff --git a/api/server/routes/config.js b/api/server/routes/config.js index fa59dde9535..77c63825703 100644 --- a/api/server/routes/config.js +++ b/api/server/routes/config.js @@ -229,7 +229,6 @@ router.get('/', async function (req, res) { try { const preLoginPayload = buildPreLoginPayload(); const publicSharePayload = buildPublicSharePayload(); - const rum = getRumConfig(); if (!req.user) { const tenantId = getTenantId(); @@ -240,7 +239,7 @@ router.get('/', async function (req, res) { ...preLoginPayload, socialLogins: baseConfig?.registration?.socialLogins ?? defaultSocialLogins, turnstile: baseConfig?.turnstileConfig, - ...(rum ? { rum } : {}), + rum: getRumConfig(baseConfig?.config?.telemetry), }; const interfaceConfig = baseConfig?.interfaceConfig; @@ -349,7 +348,7 @@ router.get('/', async function (req, res) { ...resolveCodeWorkspaceInheritanceCapability(process.env.CODE_ENVIRONMENT_DECISION_VERSION), ...(codeEnvironmentTransitionVersion != null ? { codeEnvironmentTransitionVersion } : {}), ...(cloudFront ? { cloudFront } : {}), - ...(rum ? { rum } : {}), + rum: getRumConfig(appConfig?.config?.telemetry), fileUploadSseEnabled: isEnabled(process.env.FILE_UPLOAD_SSE_ENABLED), endpointsDropParamsMap: endpointsDropParamsMap, }; diff --git a/api/server/routes/rum.js b/api/server/routes/rum.js index fd6ed4ca401..ebe198add11 100644 --- a/api/server/routes/rum.js +++ b/api/server/routes/rum.js @@ -7,10 +7,13 @@ const { requireRumLogsEnabled, handleJsonParseError, createRumProxyLimiter, + loadRumConfig, } = require('@librechat/api'); const { requireRumProxyAuth } = require('~/server/middleware'); +const { getAppConfig } = require('~/server/services/Config/app'); const router = express.Router(); +router.use(loadRumConfig(getAppConfig)); const rawOtlpBody = express.raw({ limit: getRumProxyBodyLimit(), type: ['application/x-protobuf', 'application/octet-stream'], diff --git a/api/server/services/Config/rum.js b/api/server/services/Config/rum.js index 4d8eca2de92..9b12da80214 100644 --- a/api/server/services/Config/rum.js +++ b/api/server/services/Config/rum.js @@ -1,157 +1 @@ -const { - isEnabled, - isRumProxyEnabled, - getRumProxyClientUrl, - isRumClientLogsEnabled, -} = require('@librechat/api'); -const { logger } = require('@librechat/data-schemas'); - -const DEFAULT_RUM_SERVICE_NAME = 'librechat-web'; - -function parseBooleanEnv(value, defaultValue = false) { - if (value == null || value === '') { - return defaultValue; - } - - return isEnabled(value); -} - -function parseNumberEnv(value) { - if (value == null || value === '') { - return undefined; - } - - const parsed = Number(value); - return Number.isFinite(parsed) ? parsed : undefined; -} - -function parseCsvEnv(value) { - if (!value) { - return []; - } - - return value - .split(',') - .map((item) => item.trim()) - .filter(Boolean); -} - -function parseUrl(value) { - try { - return new URL(value); - } catch { - return undefined; - } -} - -function isLocalhost(url) { - return url.hostname === 'localhost' || url.hostname === '127.0.0.1' || url.hostname === '[::1]'; -} - -function isSafeRumUrl(url) { - if (url.username || url.password) { - return false; - } - - if (url.protocol === 'https:') { - return true; - } - - return url.protocol === 'http:' && isLocalhost(url); -} - -function isSafeTraceTarget(target) { - if (target.includes('*')) { - return false; - } - - const url = parseUrl(target); - if (!url || url.protocol !== 'https:') { - return false; - } - - return true; -} - -function getRumConfig() { - if (!parseBooleanEnv(process.env.RUM_ENABLED)) { - return undefined; - } - - const provider = process.env.RUM_PROVIDER || 'hyperdx'; - if (provider !== 'hyperdx') { - logger.warn(`[config] Unsupported RUM provider "${provider}", disabling RUM`); - return undefined; - } - - const authMode = process.env.RUM_AUTH_MODE || 'publicToken'; - if (authMode !== 'publicToken' && authMode !== 'proxy') { - logger.warn(`[config] Unsupported RUM auth mode "${authMode}", disabling RUM`); - return undefined; - } - - let rumUrl; - if (authMode === 'proxy') { - rumUrl = getRumProxyClientUrl(); - - if (!isRumProxyEnabled()) { - logger.warn('[config] RUM proxy mode requires RUM_PROXY_TARGET_URL, disabling RUM'); - return undefined; - } - } else { - rumUrl = process.env.RUM_URL; - const parsedUrl = rumUrl ? parseUrl(rumUrl) : undefined; - - if (!parsedUrl || !isSafeRumUrl(parsedUrl)) { - logger.warn('[config] Invalid RUM_URL, disabling RUM'); - return undefined; - } - - if (!process.env.RUM_PUBLIC_TOKEN) { - logger.warn('[config] RUM publicToken mode requires RUM_PUBLIC_TOKEN, disabling RUM'); - return undefined; - } - - rumUrl = parsedUrl.href.replace(/\/$/, ''); - } - - const rawTracePropagationTargets = parseCsvEnv(process.env.RUM_TRACE_PROPAGATION_TARGETS); - const tracePropagationTargets = rawTracePropagationTargets.filter(isSafeTraceTarget); - if (rawTracePropagationTargets.length !== tracePropagationTargets.length) { - logger.info('[config] Ignored unsafe RUM trace propagation targets'); - } - - const configuredSampleRate = parseNumberEnv(process.env.RUM_SAMPLE_RATE); - const sampleRate = - configuredSampleRate != null && configuredSampleRate >= 0 && configuredSampleRate <= 1 - ? configuredSampleRate - : undefined; - const consoleCapture = parseBooleanEnv(process.env.RUM_CONSOLE_CAPTURE); - const advancedNetworkCapture = parseBooleanEnv(process.env.RUM_ADVANCED_NETWORK_CAPTURE); - - if (consoleCapture) { - logger.warn('[config] RUM console capture is enabled and may collect sensitive browser logs'); - } - - if (advancedNetworkCapture) { - logger.warn('[config] RUM advanced network capture is enabled and may collect payload data'); - } - - return { - provider: 'hyperdx', - enabled: true, - url: rumUrl, - serviceName: process.env.RUM_SERVICE_NAME || DEFAULT_RUM_SERVICE_NAME, - authMode, - ...(authMode === 'publicToken' ? { publicToken: process.env.RUM_PUBLIC_TOKEN } : {}), - ...(tracePropagationTargets.length > 0 ? { tracePropagationTargets } : {}), - consoleCapture, - disableReplay: parseBooleanEnv(process.env.RUM_DISABLE_REPLAY, true), - advancedNetworkCapture, - ...(sampleRate != null ? { sampleRate } : {}), - ...(process.env.RUM_ENVIRONMENT ? { environment: process.env.RUM_ENVIRONMENT } : {}), - clientLogs: isRumClientLogsEnabled(), - }; -} - -module.exports = { getRumConfig }; +module.exports = { getRumConfig: require('@librechat/api').getRumConfig }; diff --git a/client/src/lib/rum/useRum.spec.tsx b/client/src/lib/rum/useRum.spec.tsx index d05966952a3..544eb3ff2a2 100644 --- a/client/src/lib/rum/useRum.spec.tsx +++ b/client/src/lib/rum/useRum.spec.tsx @@ -3,6 +3,8 @@ import useRum from './useRum'; const mockGetSessionId = jest.fn(); const mockInit = jest.fn(); +const mockStopSessionRecorder = jest.fn(); +const mockResumeSessionRecorder = jest.fn(); const mockAddAction = jest.fn(); const mockSetGlobalAttributes = jest.fn(); const mockUseGetStartupConfig = jest.fn(); @@ -14,6 +16,8 @@ jest.mock('@hyperdx/browser', () => ({ default: { addAction: (...args: unknown[]) => mockAddAction(...args), init: (...args: unknown[]) => mockInit(...args), + stopSessionRecorder: () => mockStopSessionRecorder(), + resumeSessionRecorder: () => mockResumeSessionRecorder(), getSessionId: () => mockGetSessionId(), setGlobalAttributes: (...args: unknown[]) => mockSetGlobalAttributes(...args), }, @@ -92,6 +96,8 @@ describe('useRum', () => { apiKey: 'public-token', consoleCapture: false, disableReplay: true, + maskAllInputs: true, + maskAllText: true, service: 'librechat-web', tracePropagationTargets: ['https://librechat.example.com'], url: 'https://rum.example.com', @@ -113,6 +119,7 @@ describe('useRum', () => { expect.objectContaining({ init: expect.any(Function) }), expect.any(Function), ); + expect(mockResumeSessionRecorder).not.toHaveBeenCalled(); }); it('does not initialize RUM for unsupported auth modes', async () => { @@ -136,6 +143,121 @@ describe('useRum', () => { expect(discardEarlyRumQueue).toHaveBeenCalled(); }); + it('captures masked replays only when opted in and stops recording on unmount', async () => { + mockUseGetStartupConfig.mockReturnValue({ + isFetched: true, + data: { + rum: { + provider: 'hyperdx', + enabled: true, + url: '/api/rum', + serviceName: 'librechat-web', + authMode: 'proxy', + sessionReplay: { enabled: true }, + disableReplay: true, + }, + }, + }); + + const { unmount } = renderHook(() => useRum()); + await waitFor(() => expect(mockResumeSessionRecorder).toHaveBeenCalled()); + expect(mockInit).toHaveBeenCalledWith( + expect.objectContaining({ disableReplay: false, maskAllInputs: true, maskAllText: true }), + ); + unmount(); + expect(mockStopSessionRecorder).toHaveBeenCalledTimes(1); + }); + + it('honors an explicit replay-off policy over a legacy replay opt-in', async () => { + mockUseGetStartupConfig.mockReturnValue({ + isFetched: true, + data: { + rum: { + provider: 'hyperdx', + enabled: true, + url: '/api/rum', + serviceName: 'librechat-web', + authMode: 'proxy', + sessionReplay: { enabled: false }, + disableReplay: false, + }, + }, + }); + renderHook(() => useRum()); + await waitFor(() => expect(mockInit).toHaveBeenCalled()); + expect(mockInit).toHaveBeenCalledWith(expect.objectContaining({ disableReplay: true })); + expect(mockResumeSessionRecorder).not.toHaveBeenCalled(); + }); + + it('stops recording when proxy authentication is lost', async () => { + mockUseGetStartupConfig.mockReturnValue({ + isFetched: true, + data: { + rum: { + provider: 'hyperdx', + enabled: true, + url: '/api/rum', + serviceName: 'librechat-web', + authMode: 'proxy', + disableReplay: false, + }, + }, + }); + + const { rerender } = renderHook(() => useRum()); + await waitFor(() => expect(mockResumeSessionRecorder).toHaveBeenCalled()); + mockUseAuthContext.mockReturnValue({ token: undefined, user: undefined }); + rerender(); + expect(mockStopSessionRecorder).toHaveBeenCalledTimes(1); + }); + + it('stops an active recorder when replay is disabled in startup config', async () => { + let rum = { + provider: 'hyperdx', + enabled: true, + url: '/api/rum', + serviceName: 'librechat-web', + authMode: 'proxy', + disableReplay: false, + }; + mockUseGetStartupConfig.mockImplementation(() => ({ isFetched: true, data: { rum } })); + const { rerender } = renderHook(() => useRum()); + await waitFor(() => expect(mockResumeSessionRecorder).toHaveBeenCalledTimes(1)); + + rum = { ...rum, disableReplay: true }; + rerender(); + await waitFor(() => expect(mockInit).toHaveBeenCalledTimes(2)); + expect(mockStopSessionRecorder).toHaveBeenCalledTimes(1); + expect(mockResumeSessionRecorder).toHaveBeenCalledTimes(1); + expect(mockInit).toHaveBeenLastCalledWith(expect.objectContaining({ disableReplay: true })); + }); + + it('does not re-sample general RUM when replay policy changes', async () => { + const random = jest.spyOn(Math, 'random').mockReturnValueOnce(0.1).mockReturnValue(0.9); + try { + let rum = { + provider: 'hyperdx', + enabled: true, + url: '/api/rum', + serviceName: 'librechat-web', + authMode: 'proxy', + sessionReplay: { enabled: true }, + sampleRate: 0.5, + }; + mockUseGetStartupConfig.mockImplementation(() => ({ isFetched: true, data: { rum } })); + const { rerender } = renderHook(() => useRum()); + await waitFor(() => expect(mockInit).toHaveBeenCalledTimes(1)); + rum = { ...rum, sessionReplay: { enabled: false } }; + rerender(); + await waitFor(() => expect(mockInit).toHaveBeenCalledTimes(2)); + expect(random).toHaveBeenCalledTimes(1); + expect(mockStopSessionRecorder).toHaveBeenCalledTimes(1); + expect(mockInit).toHaveBeenLastCalledWith(expect.objectContaining({ disableReplay: true })); + } finally { + random.mockRestore(); + } + }); + it('discards the early RUM queue when sampling excludes the page', async () => { mockUseGetStartupConfig.mockReturnValue({ isFetched: true, @@ -262,6 +384,8 @@ describe('useRum', () => { apiKey: 'librechat-rum-proxy', consoleCapture: false, disableReplay: true, + maskAllInputs: true, + maskAllText: true, service: 'librechat-web', tracePropagationTargets: undefined, url: '/api/rum', diff --git a/client/src/lib/rum/useRum.ts b/client/src/lib/rum/useRum.ts index 0482dbf82af..2568dacb808 100644 --- a/client/src/lib/rum/useRum.ts +++ b/client/src/lib/rum/useRum.ts @@ -26,12 +26,16 @@ type HyperDXBrowser = HyperDXActionClient & { apiKey: string; consoleCapture: boolean; disableReplay: boolean; + maskAllInputs: boolean; + maskAllText: boolean; service: string; tracePropagationTargets?: string[]; url: string; }) => void; setGlobalAttributes: (attributes: Record) => void; getSessionId?: () => string | undefined; + stopSessionRecorder: () => void; + resumeSessionRecorder: () => void; }; function shouldInitializeRum(config: TRumConfig | undefined, token: string | undefined): boolean { @@ -169,6 +173,7 @@ export default function useRum(): void { const sampledInRef = useRef(true); const hyperDxRef = useRef(undefined); const rumConfig = startupConfig?.rum; + const replayEnabled = rumConfig?.sessionReplay?.enabled ?? rumConfig?.disableReplay === false; const shouldBufferRoutes = !startupConfigFetched || !!rumConfig; const route = useMemo(() => normalizeRumPath(location.pathname), [location.pathname]); const routeRef = useRef(route); @@ -186,6 +191,7 @@ export default function useRum(): void { useEffect( () => () => { stopClientLogs(); + hyperDxRef.current?.stopSessionRecorder(); hyperDxRef.current = undefined; initializedKeyRef.current = undefined; }, @@ -197,12 +203,14 @@ export default function useRum(): void { if (startupConfigFetched) { discardEarlyRumQueue(); stopClientLogs(); + hyperDxRef.current?.stopSessionRecorder(); } return; } if (!shouldInitializeRum(rumConfig, token)) { stopClientLogs(); + hyperDxRef.current?.stopSessionRecorder(); if (rumConfig?.authMode === 'proxy') { rumProxyToken = undefined; hyperDxRef.current = undefined; @@ -222,23 +230,32 @@ export default function useRum(): void { } const identity = config.authMode === 'proxy' ? JSON.stringify([user?.tenantId, user?.id]) : ''; - const initKey = [config.url, config.serviceName, config.authMode, apiKey, identity].join(':'); + const samplingKey = [config.url, config.serviceName, config.authMode, apiKey, identity].join( + ':', + ); + const initKey = [samplingKey, replayEnabled].join(':'); const getSessionId = () => hyperDxRef.current?.getSessionId?.(); if (initializedKeyRef.current === initKey) { if (hyperDxRef.current) { restoreRumEmitter(hyperDxRef.current); + if (replayEnabled) { + hyperDxRef.current.resumeSessionRecorder(); + } } syncClientLogs(config, getSessionId); return; } - if (sampledInitKeyRef.current !== initKey) { - /* Retire the previous account's queue and correlation before accepting new records. */ - stopClientLogs(); - hyperDxRef.current = undefined; - initializedKeyRef.current = undefined; - sampledInitKeyRef.current = initKey; + /* Retire the previous recorder and correlation before accepting new records. */ + stopClientLogs(); + hyperDxRef.current?.stopSessionRecorder(); + hyperDxRef.current = undefined; + initializedKeyRef.current = undefined; + + // Changing replay policy must not re-roll the general RUM sampling decision. + if (sampledInitKeyRef.current !== samplingKey) { + sampledInitKeyRef.current = samplingKey; sampledInRef.current = typeof config.sampleRate === 'number' ? Math.random() < config.sampleRate : true; } @@ -263,7 +280,10 @@ export default function useRum(): void { advancedNetworkCapture: config.advancedNetworkCapture ?? false, apiKey, consoleCapture: config.consoleCapture ?? false, - disableReplay: config.disableReplay ?? true, + disableReplay: !replayEnabled, + // Mask chat text and typed values before replay events leave the browser. + maskAllInputs: true, + maskAllText: true, service: config.serviceName, tracePropagationTargets: config.tracePropagationTargets, url: config.url, @@ -272,6 +292,9 @@ export default function useRum(): void { hyperDxRef.current = HyperDX; initializedKeyRef.current = initKey; HyperDX.setGlobalAttributes(buildGlobalAttributes(user, config, routeRef.current)); + if (replayEnabled) { + HyperDX.resumeSessionRecorder(); + } startRumDiagnostics(HyperDX, () => routeRef.current); }) .catch(() => undefined); @@ -279,7 +302,7 @@ export default function useRum(): void { return () => { cancelled = true; }; - }, [rumConfig, startupConfigFetched, token, user]); + }, [rumConfig, replayEnabled, startupConfigFetched, token, user]); useEffect(() => { hyperDxRef.current?.setGlobalAttributes( diff --git a/client/src/routes/__tests__/RouteErrorBoundary.spec.tsx b/client/src/routes/__tests__/RouteErrorBoundary.spec.tsx index 24f3caa1e62..5346c9d91d8 100644 --- a/client/src/routes/__tests__/RouteErrorBoundary.spec.tsx +++ b/client/src/routes/__tests__/RouteErrorBoundary.spec.tsx @@ -32,7 +32,13 @@ jest.mock('~/hooks/AuthContext', () => ({ })); jest.mock('@hyperdx/browser', () => ({ __esModule: true, - default: { init: jest.fn(), addAction: jest.fn(), setGlobalAttributes: jest.fn() }, + default: { + init: jest.fn(), + addAction: jest.fn(), + setGlobalAttributes: jest.fn(), + stopSessionRecorder: jest.fn(), + resumeSessionRecorder: jest.fn(), + }, })); type ErrorFactory = () => unknown; diff --git a/librechat.example.yaml b/librechat.example.yaml index f00bd479fa6..8b41ab6b0ec 100644 --- a/librechat.example.yaml +++ b/librechat.example.yaml @@ -7,6 +7,15 @@ version: 1.3.17 # Cache settings: Set to true to enable caching cache: true +# Browser telemetry and session recording are independent opt-ins. Collector URL, +# authentication and service settings remain in .env. YAML flags override .env; +# omitting them preserves the corresponding environment settings. +# telemetry: +# rum: +# enabled: true +# sessionReplay: +# enabled: false # Explicitly opt in to masked DOM/input recordings. + # Chat Projects: the ceilings a single project may reach. Every field is optional and # defaults to the value shown, which is the behavior when the block is absent. projects: diff --git a/packages/api/src/app/metrics.ts b/packages/api/src/app/metrics.ts index 5bdb33d3a48..1016bf2bb81 100644 --- a/packages/api/src/app/metrics.ts +++ b/packages/api/src/app/metrics.ts @@ -168,6 +168,7 @@ export type GenerationStreamAttachmentOutcome = export type RumProxyEndpoint = 'traces' | 'logs' | 'unknown'; export type RumProxyResult = | 'success' + | 'replay_disabled' | 'auth_drop' | 'auth_error' | 'bad_request' diff --git a/packages/api/src/index.ts b/packages/api/src/index.ts index b3d9bd8dbf5..b0caed52679 100644 --- a/packages/api/src/index.ts +++ b/packages/api/src/index.ts @@ -134,6 +134,8 @@ export * from './stream'; export { memoryDiagnostics } from './utils/memory'; /* RUM */ export * from './rum/proxy'; +export * from './rum/policy'; +export * from './rum/config'; export * from './rum/limiter'; /* OpenAPI */ export { createOpenApiRouter } from './openapi/router'; diff --git a/packages/api/src/rum/config.ts b/packages/api/src/rum/config.ts new file mode 100644 index 00000000000..444c4a09537 --- /dev/null +++ b/packages/api/src/rum/config.ts @@ -0,0 +1,106 @@ +import { logger } from '@librechat/data-schemas'; +import type { TRumConfig, TTelemetryConfig } from 'librechat-data-provider'; +import { getRumProxyClientUrl, isRumProxyEnabled, isRumClientLogsEnabled } from './proxy'; +import { resolveRumPolicy } from './policy'; +import { isEnabled } from '~/utils'; + +function parseNumber(value: string | undefined): number | undefined { + if (!value?.trim()) return undefined; + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : undefined; +} + +function parseUrl(value: string | undefined): URL | undefined { + try { + return value ? new URL(value) : undefined; + } catch { + return undefined; + } +} + +function isSafeRumUrl(url: URL): boolean { + if (url.username || url.password) return false; + return ( + url.protocol === 'https:' || + (url.protocol === 'http:' && ['localhost', '127.0.0.1', '[::1]'].includes(url.hostname)) + ); +} + +function isSafeTraceTarget(target: string): boolean { + return !target.includes('*') && parseUrl(target)?.protocol === 'https:'; +} + +/** Publish only browser-safe settings; collector credentials stay server-side. */ +export function getRumConfig(telemetry?: TTelemetryConfig): TRumConfig | undefined { + const env = process.env; + const policy = resolveRumPolicy(telemetry, env); + if (!policy.enabled) return undefined; + if (env.RUM_PROVIDER && env.RUM_PROVIDER !== 'hyperdx') { + logger.warn(`[config] Unsupported RUM provider "${env.RUM_PROVIDER}", disabling RUM`); + return undefined; + } + const authMode = env.RUM_AUTH_MODE || 'publicToken'; + if (authMode !== 'publicToken' && authMode !== 'proxy') { + logger.warn(`[config] Unsupported RUM auth mode "${authMode}", disabling RUM`); + return undefined; + } + + let rumUrl: string; + if (authMode === 'proxy') { + if (!isRumProxyEnabled(telemetry)) { + logger.warn('[config] RUM proxy mode requires RUM_PROXY_TARGET_URL, disabling RUM'); + return undefined; + } + rumUrl = getRumProxyClientUrl(); + } else { + const url = parseUrl(env.RUM_URL); + if (!url || !isSafeRumUrl(url)) { + logger.warn('[config] Invalid RUM_URL, disabling RUM'); + return undefined; + } + if (!env.RUM_PUBLIC_TOKEN) { + logger.warn('[config] RUM publicToken mode requires RUM_PUBLIC_TOKEN, disabling RUM'); + return undefined; + } + rumUrl = url.href.replace(/\/$/, ''); + } + + const rawTargets = (env.RUM_TRACE_PROPAGATION_TARGETS ?? '') + .split(',') + .map((target) => target.trim()) + .filter(Boolean); + const tracePropagationTargets = rawTargets.filter(isSafeTraceTarget); + if (rawTargets.length !== tracePropagationTargets.length) { + logger.info('[config] Ignored unsafe RUM trace propagation targets'); + } + const configuredSampleRate = parseNumber(env.RUM_SAMPLE_RATE); + const sampleRate = + configuredSampleRate != null && configuredSampleRate >= 0 && configuredSampleRate <= 1 + ? configuredSampleRate + : undefined; + const consoleCapture = isEnabled(env.RUM_CONSOLE_CAPTURE); + const advancedNetworkCapture = isEnabled(env.RUM_ADVANCED_NETWORK_CAPTURE); + if (consoleCapture) { + logger.warn('[config] RUM console capture is enabled and may collect sensitive browser logs'); + } + if (advancedNetworkCapture) { + logger.warn('[config] RUM advanced network capture is enabled and may collect payload data'); + } + + return { + provider: 'hyperdx', + enabled: true, + url: rumUrl, + serviceName: env.RUM_SERVICE_NAME || 'librechat-web', + authMode, + ...(authMode === 'publicToken' ? { publicToken: env.RUM_PUBLIC_TOKEN } : {}), + ...(tracePropagationTargets.length > 0 ? { tracePropagationTargets } : {}), + consoleCapture, + sessionReplay: policy.sessionReplay, + disableReplay: !policy.sessionReplay.enabled, + advancedNetworkCapture, + ...(sampleRate != null ? { sampleRate } : {}), + ...(env.RUM_ENVIRONMENT ? { environment: env.RUM_ENVIRONMENT } : {}), + clientLogs: isRumClientLogsEnabled(telemetry), + }; +} diff --git a/packages/api/src/rum/policy.spec.ts b/packages/api/src/rum/policy.spec.ts new file mode 100644 index 00000000000..e5077228dc8 --- /dev/null +++ b/packages/api/src/rum/policy.spec.ts @@ -0,0 +1,105 @@ +import { telemetrySchema } from 'librechat-data-provider'; +import type { RumRequest } from './policy'; +import { resolveRumPolicy, loadRumConfig } from './policy'; + +describe('RUM replay policy', () => { + it('keeps RUM and replay off without configuration', () => { + expect(resolveRumPolicy(undefined, {})).toEqual({ + enabled: false, + sessionReplay: { enabled: false }, + }); + }); + + it.each([ + [{ RUM_ENABLED: 'true' }, true, false], + [{ RUM_ENABLED: 'true', RUM_SESSION_REPLAY_ENABLED: 'true' }, true, true], + [{ RUM_ENABLED: 'false', RUM_SESSION_REPLAY_ENABLED: 'true' }, false, false], + [{ RUM_ENABLED: 'true', RUM_DISABLE_REPLAY: 'false' }, true, true], + [ + { RUM_ENABLED: 'true', RUM_DISABLE_REPLAY: 'false', RUM_SESSION_REPLAY_ENABLED: 'false' }, + true, + false, + ], + [ + { RUM_ENABLED: 'true', RUM_DISABLE_REPLAY: 'true', RUM_SESSION_REPLAY_ENABLED: 'true' }, + true, + true, + ], + [ + { RUM_ENABLED: 'true', RUM_DISABLE_REPLAY: 'false', RUM_SESSION_REPLAY_ENABLED: 'invalid' }, + true, + false, + ], + ])('resolves environment policy %p', (env, enabled, replay) => { + expect(resolveRumPolicy(undefined, env)).toEqual({ + enabled, + sessionReplay: { enabled: replay }, + }); + }); + + it('lets YAML explicitly disable replay despite both environment opt-ins', () => { + expect( + resolveRumPolicy( + { rum: { enabled: true, sessionReplay: { enabled: false } } }, + { + RUM_ENABLED: 'true', + RUM_SESSION_REPLAY_ENABLED: 'true', + RUM_DISABLE_REPLAY: 'false', + }, + ), + ).toEqual({ enabled: true, sessionReplay: { enabled: false } }); + }); + + it('treats a raw empty YAML replay block as off before schema defaults are materialized', () => { + expect( + resolveRumPolicy( + { rum: { sessionReplay: {} } }, + { + RUM_ENABLED: 'true', + RUM_SESSION_REPLAY_ENABLED: 'true', + }, + ), + ).toEqual({ enabled: true, sessionReplay: { enabled: false } }); + }); + + it('lets YAML enable RUM and replay and keeps the RUM master switch authoritative', () => { + const replay = { enabled: true }; + expect(resolveRumPolicy({ rum: { enabled: true, sessionReplay: replay } }, {})).toEqual({ + enabled: true, + sessionReplay: replay, + }); + expect( + resolveRumPolicy( + { rum: { enabled: false, sessionReplay: replay } }, + { + RUM_ENABLED: 'true', + RUM_SESSION_REPLAY_ENABLED: 'true', + }, + ), + ).toEqual({ enabled: false, sessionReplay: { enabled: false } }); + }); + + it('validates YAML flags without injecting defaults over existing environment configuration', () => { + expect(telemetrySchema.parse({})).toEqual({}); + expect(telemetrySchema.parse({ rum: { sessionReplay: {} } })).toEqual({ + rum: { sessionReplay: { enabled: false } }, + }); + expect(telemetrySchema.safeParse({ rum: { sessionReplay: { enabled: 'true' } } }).success).toBe( + false, + ); + }); + + it('loads only the cached deployment config and fails closed if loading fails', async () => { + const getAppConfig = jest.fn().mockResolvedValue({ config: { telemetry: {} } }); + const middleware = loadRumConfig(getAppConfig); + const req = {} as RumRequest; + const next = jest.fn(); + await middleware(req, {} as Parameters[1], next); + expect(getAppConfig).toHaveBeenCalledWith({ baseOnly: true }); + expect(req.config?.config?.telemetry).toEqual({}); + const error = new Error('configuration unavailable'); + getAppConfig.mockRejectedValueOnce(error); + await middleware(req, {} as Parameters[1], next); + expect(next).toHaveBeenLastCalledWith(error); + }); +}); diff --git a/packages/api/src/rum/policy.ts b/packages/api/src/rum/policy.ts new file mode 100644 index 00000000000..59dba57b51b --- /dev/null +++ b/packages/api/src/rum/policy.ts @@ -0,0 +1,37 @@ +import type { TTelemetryConfig } from 'librechat-data-provider'; +import type { AppConfig } from '@librechat/data-schemas'; +import type { Request, RequestHandler } from 'express'; +import { isEnabled } from '~/utils'; + +export type RumRequest = Request & { + config?: { config?: { telemetry?: TTelemetryConfig } }; +}; + +export function resolveRumPolicy( + telemetry: TTelemetryConfig | undefined, + env: NodeJS.ProcessEnv, +): { enabled: boolean; sessionReplay: { enabled: boolean } } { + const enabled = telemetry?.rum?.enabled ?? isEnabled(env.RUM_ENABLED); + const replayConfig = telemetry?.rum?.sessionReplay; + const configuredReplay = replayConfig == null ? undefined : (replayConfig.enabled ?? false); + const positiveReplay = env.RUM_SESSION_REPLAY_ENABLED?.trim(); + const legacyReplay = env.RUM_DISABLE_REPLAY?.trim(); + const replayEnabled = + configuredReplay ?? + (positiveReplay ? isEnabled(positiveReplay) : !!legacyReplay && !isEnabled(legacyReplay)); + return { enabled, sessionReplay: { enabled: enabled && replayEnabled } }; +} + +/** Deployment policy comes from the cached base config, never a browser payload or user override. */ +export function loadRumConfig( + getAppConfig: (options: { baseOnly: true }) => Promise, +): RequestHandler { + return async (req: RumRequest, _res, next) => { + try { + req.config = await getAppConfig({ baseOnly: true }); + next(); + } catch (error) { + next(error); + } + }; +} diff --git a/packages/api/src/rum/proxy.spec.ts b/packages/api/src/rum/proxy.spec.ts index c9bdd2c0673..6fcf8612d5a 100644 --- a/packages/api/src/rum/proxy.spec.ts +++ b/packages/api/src/rum/proxy.spec.ts @@ -255,7 +255,7 @@ describe('RUM proxy configuration', () => { const res = makeResponse(); await proxyRumRequest( - { path: '/v1/logs', body: Buffer.from('payload'), headers: PROTOBUF_HEADERS } as never, + { path: '/v1/logs', body: Buffer.from([10, 0]), headers: PROTOBUF_HEADERS } as never, res as never, ); @@ -391,6 +391,8 @@ describe('RUM proxy upstream HTTP contract', () => { beforeEach(async () => { process.env = { ...originalEnv }; + process.env.RUM_ENABLED = 'true'; + process.env.RUM_SESSION_REPLAY_ENABLED = 'true'; jest.mocked(recordRumProxyRequest).mockClear(); collector = createServer(); await new Promise((resolve) => collector.listen(0, '127.0.0.1', resolve)); @@ -417,6 +419,85 @@ describe('RUM proxy upstream HTTP contract', () => { return app; } + it('drops stale replay batches without collector acknowledgement while ordinary logs stay enabled', async () => { + process.env.RUM_SESSION_REPLAY_ENABLED = 'false'; + process.env.RUM_CLIENT_LOGS = 'true'; + const collectorRequest = jest.fn(); + collector.on('request', collectorRequest); + const response = await request(createProxy()) + .post('/v1/logs') + .send({ + resourceLogs: [ + { + scopeLogs: [ + { + logRecords: [ + { + attributes: [{ key: 'rr-web.event', value: { intValue: '2' } }], + body: { stringValue: 'private-recording' }, + }, + ], + }, + ], + }, + ], + }); + expect(response.status).toBe(204); + expect(response.headers[RUM_COLLECTOR_ACK_HEADER]).toBeUndefined(); + expect(collectorRequest).not.toHaveBeenCalled(); + expect(recordRumProxyRequest).toHaveBeenCalledWith('logs', 'replay_disabled'); + }); + + it('strips replay records from a mixed batch before contacting the collector', async () => { + process.env.RUM_SESSION_REPLAY_ENABLED = 'false'; + process.env.RUM_CONSOLE_CAPTURE = 'true'; + let received = ''; + collector.on('request', (req, res) => { + req.on('data', (chunk: Buffer) => { + received += chunk.toString(); + }); + req.on('end', () => { + res.writeHead(202); + res.end(); + }); + }); + const response = await request(createProxy()) + .post('/v1/logs') + .send({ + resourceLogs: [ + { + scopeLogs: [ + { + logRecords: [ + { body: { stringValue: 'ordinary-error' } }, + { + attributes: [{ key: 'rr-web.event', value: { intValue: '2' } }], + body: { stringValue: 'private-recording' }, + }, + ], + }, + ], + }, + ], + }); + expect(response.status).toBe(202); + expect(received).toContain('ordinary-error'); + expect(received).not.toContain('private-recording'); + expect(response.headers[RUM_COLLECTOR_ACK_HEADER]).toBe('true'); + }); + + it('rejects malformed binary logs while replay is off without contacting the collector', async () => { + process.env.RUM_SESSION_REPLAY_ENABLED = 'false'; + const collectorRequest = jest.fn(); + collector.on('request', collectorRequest); + const response = await request(createProxy()) + .post('/v1/logs') + .set('Content-Type', 'application/x-protobuf') + .send(Buffer.from([10, 255])); + expect(response.status).toBe(400); + expect(collectorRequest).not.toHaveBeenCalled(); + }); + it.each(['traces', 'logs'])( 'sends only server credentials and intact %s payloads', async (signal) => { diff --git a/packages/api/src/rum/proxy.ts b/packages/api/src/rum/proxy.ts index 00197e10b92..3d5050d783e 100644 --- a/packages/api/src/rum/proxy.ts +++ b/packages/api/src/rum/proxy.ts @@ -2,8 +2,12 @@ import { validateHeaderValue } from 'node:http'; import { logger } from '@librechat/data-schemas'; import { RUM_COLLECTOR_ACK_HEADER } from 'librechat-data-provider'; import type { Request, Response, RequestHandler } from 'express'; +import type { TTelemetryConfig } from 'librechat-data-provider'; import type { RumProxyEndpoint, RumProxyResult } from '~/app/metrics'; +import type { RumRequest } from './policy'; import { recordRumProxyRequest } from '~/app/metrics'; +import { resolveRumPolicy } from './policy'; +import { filterReplayLogs } from './replay'; import { isEnabled } from '~/utils'; const DEFAULT_PROXY_PATH = '/api/rum'; @@ -62,9 +66,9 @@ export function getRumProxyTargetBaseUrl(): URL | undefined { return url; } -export function isRumProxyEnabled(): boolean { +export function isRumProxyEnabled(telemetry?: TTelemetryConfig): boolean { return ( - isEnabled(process.env.RUM_ENABLED) && + resolveRumPolicy(telemetry, process.env).enabled && process.env.RUM_AUTH_MODE === 'proxy' && getRumProxyTargetBaseUrl() != null ); @@ -76,8 +80,8 @@ export function isRumProxyEnabled(): boolean { * its own; proxy mode only, so the logs reach the collector through session auth and never with * a browser-held ingestion key. */ -export function isRumClientLogsEnabled(): boolean { - return isRumProxyEnabled() && isEnabled(process.env.RUM_CLIENT_LOGS); +export function isRumClientLogsEnabled(telemetry?: TTelemetryConfig): boolean { + return isRumProxyEnabled(telemetry) && isEnabled(process.env.RUM_CLIENT_LOGS); } /** @@ -85,12 +89,11 @@ export function isRumClientLogsEnabled(): boolean { * session replay produce browser logs, so with all three off the logs route refuses even tabs * still holding an older startup config, which makes turning `RUM_CLIENT_LOGS` off authoritative. */ -export function isRumLogsEndpointEnabled(): boolean { - if (!isRumProxyEnabled()) { +export function isRumLogsEndpointEnabled(telemetry?: TTelemetryConfig): boolean { + if (!isRumProxyEnabled(telemetry)) { return false; } - const replaySetting = process.env.RUM_DISABLE_REPLAY?.trim(); - const replayEnabled = !!replaySetting && !isEnabled(replaySetting); + const replayEnabled = resolveRumPolicy(telemetry, process.env).sessionReplay.enabled; return ( isEnabled(process.env.RUM_CLIENT_LOGS) || isEnabled(process.env.RUM_CONSOLE_CAPTURE) || @@ -99,16 +102,16 @@ export function isRumLogsEndpointEnabled(): boolean { } /** HTTP gates stay with the proxy policy rather than the CJS route wiring. */ -export const requireRumProxyEnabled: RequestHandler = (_req, res, next) => { - if (!isRumProxyEnabled()) { +export const requireRumProxyEnabled: RequestHandler = (req: RumRequest, res, next) => { + if (!isRumProxyEnabled(req.config?.config?.telemetry)) { res.status(404).json({ message: 'RUM proxy is not configured' }); return; } next(); }; -export const requireRumLogsEnabled: RequestHandler = (_req, res, next) => { - if (!isRumLogsEndpointEnabled()) { +export const requireRumLogsEnabled: RequestHandler = (req: RumRequest, res, next) => { + if (!isRumLogsEndpointEnabled(req.config?.config?.telemetry)) { res.status(404).json({ message: 'RUM logs are not enabled' }); return; } @@ -215,7 +218,7 @@ function getProxyHeaders( } export async function proxyRumRequest( - req: Request, + req: RumRequest, res: Response, upstreamAuthorization?: string, ): Promise { @@ -234,13 +237,31 @@ export async function proxyRumRequest( return; } - const body = getRequestBody(req); + let body = getRequestBody(req); if (!body) { recordRumProxyRequest(endpoint, 'bad_request'); res.status(400).json({ message: 'RUM payload is required' }); return; } + if ( + endpoint === 'logs' && + !resolveRumPolicy(req.config?.config?.telemetry, process.env).sessionReplay.enabled + ) { + const filtered = filterReplayLogs(body, getHeader(req, 'content-type') ?? ''); + if (!filtered.ok) { + recordRumProxyRequest(endpoint, 'bad_request'); + res.status(400).json({ message: 'Invalid RUM logs payload' }); + return; + } + if (filtered.body == null) { + recordRumProxyRequest(endpoint, 'replay_disabled'); + res.status(204).end(); + return; + } + body = filtered.body; + } + const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), getRumProxyTimeoutMs()); try { diff --git a/packages/api/src/rum/replay.spec.ts b/packages/api/src/rum/replay.spec.ts new file mode 100644 index 00000000000..5d0f9998ed3 --- /dev/null +++ b/packages/api/src/rum/replay.spec.ts @@ -0,0 +1,102 @@ +import { parse, Writer } from 'protobufjs'; +import { filterReplayLogs } from './replay'; + +const requestType = parse(`syntax = "proto3"; + message AnyValue { string stringValue = 1; } + message KeyValue { string key = 1; AnyValue value = 2; } + message Scope { string name = 1; } + message LogRecord { string severityText = 3; AnyValue body = 5; repeated KeyValue attributes = 6; } + message ScopeLogs { Scope scope = 1; repeated LogRecord logRecords = 2; string schemaUrl = 3; } + message ResourceLogs { repeated ScopeLogs scopeLogs = 2; string schemaUrl = 3; } + message ExportLogsServiceRequest { repeated ResourceLogs resourceLogs = 1; } +`).root.lookupType('ExportLogsServiceRequest'); + +const ordinary = { + severityText: 'ERROR', + body: { stringValue: 'ordinary-client-error' }, + attributes: [{ key: 'route', value: { stringValue: '/c/new' } }], +}; +const replay = { + body: { stringValue: 'private-replay-snapshot' }, + attributes: [{ key: 'rr-web.event', value: { stringValue: '2' } }], +}; +const batch = (records = [ordinary, replay], name = 'browser') => ({ + resourceLogs: [{ schemaUrl: 'schema', scopeLogs: [{ scope: { name }, logRecords: records }] }], +}); + +describe.each(['json', 'protobuf', 'octet-stream'] as const)( + 'replay-off %s filtering', + (encoding) => { + const contentType = + encoding === 'json' + ? 'application/json' + : `application/${encoding === 'protobuf' ? 'x-protobuf' : encoding}`; + const encode = (value: ReturnType) => + encoding === 'json' + ? JSON.stringify(value) + : Buffer.from(requestType.encode(requestType.create(value)).finish()); + const decode = (value: Buffer | string) => + encoding === 'json' + ? JSON.parse(value.toString()) + : requestType.toObject(requestType.decode(Buffer.from(value))); + + it('drops replay-only batches without forwarding their content', () => { + expect(filterReplayLogs(encode(batch([replay])), contentType)).toEqual({ + ok: true, + body: undefined, + }); + }); + + it('preserves ordinary records and envelope metadata in mixed batches', () => { + const result = filterReplayLogs(encode(batch()), contentType); + expect(result.ok).toBe(true); + if (!result.ok || !result.body) throw new Error('Expected ordinary log batch'); + expect(decode(result.body)).toEqual(batch([ordinary])); + expect(result.body.toString()).not.toContain('private-replay-snapshot'); + }); + + it('preserves replay-free payloads byte-for-byte', () => { + const body = encode(batch([ordinary])); + expect(filterReplayLogs(body, contentType)).toEqual({ ok: true, body }); + }); + + it('drops recorder scopes even when a record has no rr-web attributes', () => { + expect( + filterReplayLogs( + encode(batch([ordinary], '@hyperdx/otel-web-session-recorder')), + contentType, + ), + ).toEqual({ ok: true, body: undefined }); + }); + }, +); + +it.each(['not json', '{"resourceLogs":[{"scope_logs":[]}]}', '{"resource_logs":[]}'])( + 'rejects malformed or noncanonical JSON envelopes: %s', + (body) => { + expect(filterReplayLogs(body, 'application/json')).toEqual({ ok: false }); + }, +); + +it.each([Buffer.from([0]), Buffer.from([10, 255]), Buffer.from([8, 1])])( + 'rejects malformed protobuf rather than forwarding opaque logs', + (body) => { + expect(filterReplayLogs(body, 'application/x-protobuf')).toEqual({ ok: false }); + }, +); + +it('preserves unknown protobuf fields and untouched resources', () => { + const original = Buffer.from(requestType.encode(requestType.create(batch([ordinary]))).finish()); + const extra = Buffer.from(Writer.create().uint32(120).uint32(7).finish()); + const body = Buffer.concat([original, extra]); + expect(filterReplayLogs(body, 'application/x-protobuf')).toEqual({ ok: true, body }); +}); + +it('preserves unknown protobuf fields when removing a replay from a mixed batch', () => { + const original = Buffer.from(requestType.encode(requestType.create(batch())).finish()); + const extra = Buffer.from(Writer.create().uint32(120).uint32(7).finish()); + const result = filterReplayLogs(Buffer.concat([original, extra]), 'application/x-protobuf'); + if (!result.ok || !result.body) throw new Error('Expected ordinary log batch'); + const expected = Buffer.from(requestType.encode(requestType.create(batch([ordinary]))).finish()); + expect(result.body).toEqual(Buffer.concat([expected, extra])); +}); diff --git a/packages/api/src/rum/replay.ts b/packages/api/src/rum/replay.ts new file mode 100644 index 00000000000..ef1aa188742 --- /dev/null +++ b/packages/api/src/rum/replay.ts @@ -0,0 +1,163 @@ +import { z } from 'zod'; +import { Reader, Writer } from 'protobufjs/minimal'; + +const attributeSchema = z.object({ key: z.string() }).passthrough(); +const logsSchema = z + .object({ + resourceLogs: z + .array( + z + .object({ + resource: z.unknown().optional(), + schemaUrl: z.string().optional(), + scopeLogs: z + .array( + z + .object({ + schemaUrl: z.string().optional(), + scope: z.object({ name: z.string().optional() }).passthrough().optional(), + logRecords: z + .array( + z + .object({ + attributes: z.array(attributeSchema).optional(), + }) + .passthrough(), + ) + .optional(), + }) + .strict(), + ) + .optional(), + }) + .strict(), + ) + .optional(), + }) + .strict(); + +const isReplayScope = (name: string | undefined): boolean => + name === '@hyperdx/otel-web-session-recorder'; + +type FilteredMessage = { bytes: Uint8Array; records: number; removed: number }; + +/** Preserve unknown protobuf fields byte-for-byte; inspect only OTLP's nested log-record fields. */ +function filterMessage( + bytes: Uint8Array, + field: number, + filter: (value: Uint8Array) => FilteredMessage, +): FilteredMessage { + const reader = Reader.create(bytes); + const parts: Uint8Array[] = []; + let records = 0; + let removed = 0; + while (reader.pos < reader.len) { + const start = reader.pos; + const tag = reader.uint32(); + if (tag >>> 3 === 0) throw new Error('Invalid protobuf field'); + if (tag >>> 3 === field) { + if ((tag & 7) !== 2) throw new Error('Invalid protobuf wire type'); + const result = filter(reader.bytes()); + records += result.records; + removed += result.removed; + if (result.removed === 0) { + parts.push(bytes.subarray(start, reader.pos)); + } else if (result.records > 0) { + parts.push(Writer.create().uint32(tag).bytes(result.bytes).finish()); + } + } else { + reader.skipType(tag & 7); + parts.push(bytes.subarray(start, reader.pos)); + } + } + return { bytes: Buffer.concat(parts), records, removed }; +} + +function stringField(bytes: Uint8Array, field: number): string | undefined { + const reader = Reader.create(bytes); + let result: string | undefined; + while (reader.pos < reader.len) { + const tag = reader.uint32(); + if (tag >>> 3 === 0) throw new Error('Invalid protobuf field'); + if (tag >>> 3 === field) { + if ((tag & 7) !== 2) throw new Error('Invalid protobuf wire type'); + result = reader.string(); + } else { + reader.skipType(tag & 7); + } + } + return result; +} + +function filterRecord(bytes: Uint8Array): FilteredMessage { + let replay = false; + // OTLP LogRecord.attributes = field 6; KeyValue.key = field 1. + filterMessage(bytes, 6, (attribute) => { + replay ||= stringField(attribute, 1)?.startsWith('rr-web.') ?? false; + return { bytes: attribute, records: 1, removed: 0 }; + }); + return { bytes, records: replay ? 0 : 1, removed: replay ? 1 : 0 }; +} + +function filterScope(bytes: Uint8Array): FilteredMessage { + const reader = Reader.create(bytes); + let replay = false; + while (reader.pos < reader.len) { + const tag = reader.uint32(); + if (tag >>> 3 === 0) throw new Error('Invalid protobuf field'); + if (tag >>> 3 === 1) { + if ((tag & 7) !== 2) throw new Error('Invalid protobuf wire type'); + const scope = reader.bytes(); + replay ||= isReplayScope(stringField(scope, 1)); + } else { + reader.skipType(tag & 7); + } + } + // ScopeLogs.logRecords = field 2. + return filterMessage(bytes, 2, (record) => + replay ? { bytes: record, records: 0, removed: 1 } : filterRecord(record), + ); +} + +/** Fail closed on malformed logs. Mixed batches retain ordinary records and their resource metadata. */ +export function filterReplayLogs( + body: Buffer | string, + contentType: string, +): { ok: true; body?: Buffer | string } | { ok: false } { + try { + if (contentType.split(';', 1)[0].trim().toLowerCase() !== 'application/json') { + const bytes = typeof body === 'string' ? Buffer.from(body) : body; + // ExportLogsServiceRequest.resourceLogs = 1; ResourceLogs.scopeLogs = 2. + const result = filterMessage(bytes, 1, (resource) => filterMessage(resource, 2, filterScope)); + if (result.removed === 0) return { ok: true, body }; + return { + ok: true, + body: result.records > 0 ? Buffer.from(result.bytes) : undefined, + }; + } + + const parsed = logsSchema.safeParse(JSON.parse(body.toString())); + if (!parsed.success) return { ok: false }; + let removed = 0; + let records = 0; + for (const resource of parsed.data.resourceLogs ?? []) { + for (const scope of resource.scopeLogs ?? []) { + scope.logRecords = scope.logRecords?.filter((record) => { + const replay = + isReplayScope(scope.scope?.name) || + record.attributes?.some((attribute) => attribute.key.startsWith('rr-web.')); + if (replay) { + removed++; + return false; + } + records++; + return true; + }); + } + } + if (removed === 0) return { ok: true, body }; + return { ok: true, body: records > 0 ? JSON.stringify(parsed.data) : undefined }; + } catch { + return { ok: false }; + } +} diff --git a/packages/data-provider/src/config.spec.ts b/packages/data-provider/src/config.spec.ts index 3340ca64712..dee3a3ed83a 100644 --- a/packages/data-provider/src/config.spec.ts +++ b/packages/data-provider/src/config.spec.ts @@ -1,4 +1,5 @@ import type { TEndpointsConfig } from './types'; + import { allowedAddressesSchema, agentsEndpointSchema, @@ -6,6 +7,7 @@ import { DEFAULT_MAX_RETAINED_TOOL_COUNT_CHARS, bedrockModels, configSchema, + telemetrySchema, chatProjectsConfigSchema, DEFAULT_STEER_ARM_CONFIRMATION_TIMEOUT_MS, codeEnvironmentUserConfigSchema, @@ -28,6 +30,27 @@ import { resolveEndpointType, webSearchSchema, } from './config'; + +describe('independent browser replay opt-in', () => { + it('preserves absent configuration for existing environment-based RUM deployments', () => { + expect(configSchema.parse({ version: '1.3.17' }).telemetry).toBeUndefined(); + expect(telemetrySchema.parse({ rum: { enabled: true } })).toEqual({ rum: { enabled: true } }); + }); + + it('defaults an explicitly configured replay block to off', () => { + expect(telemetrySchema.parse({ rum: { enabled: true, sessionReplay: {} } })).toEqual({ + rum: { enabled: true, sessionReplay: { enabled: false } }, + }); + }); + + it.each([ + { rum: { sessionReplay: { enabled: 'true' } } }, + { rum: { sessionReplay: { enable: false } } }, + { rum: { enable: true } }, + ])('rejects invalid or misspelled policy %p', (telemetry) => { + expect(telemetrySchema.safeParse(telemetry).success).toBe(false); + }); +}); import { EModelEndpoint, isDocumentSupportedProvider } from './schemas'; import { getEndpointFileConfig, mergeFileConfig } from './file-config'; import { DEFAULT_MCP_APP_CSP_LIMITS } from './mcp/csp'; diff --git a/packages/data-provider/src/config.ts b/packages/data-provider/src/config.ts index 403b94976c8..bc1e5d9edee 100644 --- a/packages/data-provider/src/config.ts +++ b/packages/data-provider/src/config.ts @@ -3197,6 +3197,23 @@ export type TTurnstileConfig = z.infer; /** Distinguishes collector acceptance from the proxy's silent 204 authentication drop. */ export const RUM_COLLECTOR_ACK_HEADER = 'x-librechat-rum-accepted'; +export const telemetrySchema = z + .object({ + rum: z + .object({ + enabled: z.boolean().optional(), + sessionReplay: z + .object({ enabled: z.boolean().default(false) }) + .strict() + .optional(), + }) + .strict() + .optional(), + }) + .strict(); + +export type TTelemetryConfig = DeepPartial>; + export type TRumConfig = { provider: 'hyperdx'; enabled: boolean; @@ -3207,6 +3224,8 @@ export type TRumConfig = { tracePropagationTargets?: string[]; consoleCapture?: boolean; disableReplay?: boolean; + /** Positive replay policy; disableReplay is retained for older browser bundles. */ + sessionReplay?: { enabled: boolean }; advancedNetworkCapture?: boolean; sampleRate?: number; environment?: string; @@ -4058,6 +4077,7 @@ export type TToolCallPreviewsConfig = z.infer