mirror of
https://github.com/simstudioai/sim.git
synced 2026-10-25 09:34:44 +08:00
fix(security): meter and throttle the deployed-chat TTS relay (#6212)
* fix(security): meter and throttle the deployed-chat TTS relay POST /api/proxy/tts/stream treated "a live public chat exists" as authorization to spend the platform ElevenLabs key. A public chat id is handed to every visitor, so any anonymous caller could synthesize speech with no length cap, no rate limit and no usage accounting. Bring the relay in line with its STT sibling (/api/speech/token): - Resolve the chat's workspace and bill synthesized characters to that payer via a new `voice-output` usage source, so spend is attributable and counts against the plan's usage limit (402 once exceeded). - Throttle per IP before any database work, and per chat afterwards, to bound both one caller hammering many chats and many callers hammering one chat. - Cap `text` at 2000 characters and allowlist `voiceId`/`modelId`, so the caller can no longer choose an unbounded charge, a premium or cloned voice, or the billing model. - Drop `Access-Control-Allow-Origin: *`, which let any third-party page read the audio; deployed chat and the Office embed are same-origin. * fix(security): correct TTS metering, pricing and body cap Follow-up review of the previous commit found five defects in it: - Usage rows collided. `usage_log.event_key` is unique and inserts are conflict-do-nothing, and the key is derived from the entry's stable fields. With no explicit sourceReference, two synthesis calls of equal character count in the same workspace produced the same key, so every repeat length went unbilled — defeating the metering this change is for. Each call now carries a unique sourceReference. - Priced at $0.10 per 1k characters, twice the published ElevenLabs Flash/Turbo rate of $0.05, which would have overcharged customers 2x. - No body cap, so an anonymous caller could make the route buffer up to the shared 50 MB default before validation. Now 16 KB, as the STT sibling does. - Threshold settlement ran per sentence: several queries and a possible Stripe call on a realtime path. The workflow execution that produced the text already settles the payer. - The per-IP bucket was described as preventing database amplification. getClientIp trusts the leftmost X-Forwarded-For, so an attacker rotates past it; the comment now says the per-chat bucket is load-bearing. * refactor(chat): share the deployed-chat auth gate across voice routes Review of the previous commits surfaced duplication and one more gap: - The TTS and STT routes had grown near-identical copies of the chat auth + payer lookup. Extracted to resolveDeployedChatCaller, so the gate and the payer resolve together and cannot drift per route — that duplication is how the unmetered TTS path shipped in the first place. - Neither copy filtered chat.archivedAt, so an archived chat could still authorize spend against its former owner's workspace. The shared lookup now filters it, fixing both routes at once. Note: not covered by a test — the db chain mock does not evaluate WHERE clauses, so an assertion here could not fail. - Replaced the route's hand-rolled 429 builder with the existing enforceIpRateLimit helper, and added enforceChatRateLimit alongside the per-user/IP/workspace helpers. Gains the standard Retry-After and X-RateLimit-Reset headers plus throttle logging. - Dropped a test that asserted a module the route no longer imports was never called: it could not fail. - Narrowed the contract: unexported the single-use allowlists and dropped .passthrough() now that the body is a closed shape. * fix(security): fail closed when voice-output usage cannot be recorded Review round 1 findings: - A ledger write failure previously logged and streamed the audio anyway, leaving the spend unrecorded and the payer's usage understated. The caller is anonymous, so serving audio we could not charge for is the unmetered spend this route exists to prevent — it now returns 500. - Use generateId() from @sim/utils/id rather than crypto.randomUUID, per the AGENTS.md ID rule. generateId returns a full UUID v4, so the per-call uniqueness the usage_log event_key depends on is unchanged. * fix(chat): split long TTS text so the relay cap cannot drop audio The client sentence-splits on Western `.!?` only, so text that never matches — CJK punctuation, or a list with no terminal punctuation — accumulates and is flushed as one block at the end of the stream. Against the new 2000-character relay cap that block is rejected and the whole message plays no audio, a regression introduced by adding the cap. Split to cap-sized pieces at the single point that enqueues synthesis, so both the per-sentence path and the end-of-stream flush are covered. Prefers a whitespace or CJK punctuation boundary, falling back to a hard cut when a block has none. The server cap stays as the enforcement point. * fix(security): release the vendor stream when metering rejects the request The fail-closed branch returned 500 with the ElevenLabs response body still open, so synthesis and download kept consuming vendor and runtime resources for a caller that was already rejected. Cancel it before returning, and assert the cancellation in the test.
This commit is contained in:
Waleed
@@ -3,6 +3,7 @@
|
||||
import { type RefObject, useCallback, useEffect, useMemo, useRef, useState } from 'react'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { generateId } from '@sim/utils/id'
|
||||
import { DEFAULT_TTS_VOICE_ID } from '@/lib/api/contracts/media/tts-stream'
|
||||
import { noop } from '@/lib/core/utils/request'
|
||||
import {
|
||||
AGENT_STREAM_PROTOCOL_HEADER,
|
||||
@@ -49,7 +50,7 @@ interface ChatRequestPayload {
|
||||
}
|
||||
|
||||
const DEFAULT_VOICE_SETTINGS = {
|
||||
voiceId: 'cgSgspJ2msm6clMCkdW9', // Default ElevenLabs voice (Jessica) — Flash v2.5-optimized
|
||||
voiceId: DEFAULT_TTS_VOICE_ID,
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { MAX_TTS_TEXT_LENGTH } from '@/lib/api/contracts/media/tts-stream'
|
||||
import { splitForSynthesis } from '@/app/(interfaces)/chat/hooks/use-audio-streaming'
|
||||
|
||||
describe('splitForSynthesis', () => {
|
||||
it('leaves text within the relay cap untouched', () => {
|
||||
expect(splitForSynthesis('Short answer.')).toEqual(['Short answer.'])
|
||||
})
|
||||
|
||||
/**
|
||||
* The caller only sentence-splits on Western `.!?`, so CJK punctuation never
|
||||
* matches and the whole answer arrives as one block. Before splitting, the
|
||||
* relay rejected it and the message played no audio at all.
|
||||
*/
|
||||
it('splits CJK text that never matches the Western sentence split', () => {
|
||||
const text = '这是一个很长的回答。'.repeat(400)
|
||||
expect(text.length).toBeGreaterThan(MAX_TTS_TEXT_LENGTH)
|
||||
|
||||
const chunks = splitForSynthesis(text)
|
||||
|
||||
expect(chunks.length).toBeGreaterThan(1)
|
||||
for (const chunk of chunks) {
|
||||
expect(chunk.length).toBeLessThanOrEqual(MAX_TTS_TEXT_LENGTH)
|
||||
}
|
||||
})
|
||||
|
||||
it('splits a long list that has no terminal punctuation', () => {
|
||||
const text = Array.from({ length: 300 }, (_, i) => `- item number ${i}`).join('\n')
|
||||
expect(text.length).toBeGreaterThan(MAX_TTS_TEXT_LENGTH)
|
||||
|
||||
const chunks = splitForSynthesis(text)
|
||||
|
||||
for (const chunk of chunks) {
|
||||
expect(chunk.length).toBeLessThanOrEqual(MAX_TTS_TEXT_LENGTH)
|
||||
}
|
||||
})
|
||||
|
||||
it('preserves the spoken content across chunks', () => {
|
||||
const text = Array.from({ length: 500 }, (_, i) => `word${i}`).join(' ')
|
||||
|
||||
const chunks = splitForSynthesis(text)
|
||||
|
||||
expect(chunks.join(' ').replace(/\s+/g, ' ')).toBe(text)
|
||||
})
|
||||
|
||||
it('still caps text with no break opportunity at all', () => {
|
||||
const chunks = splitForSynthesis('a'.repeat(MAX_TTS_TEXT_LENGTH * 2 + 5))
|
||||
|
||||
expect(chunks.length).toBe(3)
|
||||
for (const chunk of chunks) {
|
||||
expect(chunk.length).toBeLessThanOrEqual(MAX_TTS_TEXT_LENGTH)
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -2,9 +2,47 @@
|
||||
|
||||
import { type RefObject, useCallback, useRef, useState } from 'react'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { DEFAULT_TTS_MODEL_ID, MAX_TTS_TEXT_LENGTH } from '@/lib/api/contracts/media/tts-stream'
|
||||
|
||||
const logger = createLogger('UseAudioStreaming')
|
||||
|
||||
/** Prefer breaking on a boundary this far into the chunk before splitting mid-word. */
|
||||
const MIN_SPLIT_RATIO = 0.6
|
||||
|
||||
/**
|
||||
* Splits text into pieces the TTS relay will accept.
|
||||
*
|
||||
* The caller sentence-splits on Western `.!?` only, so text that never matches
|
||||
* — CJK punctuation, or a list with no terminal punctuation — reaches this hook
|
||||
* as one accumulated block that can exceed the relay's per-request cap. Without
|
||||
* splitting, the relay rejects it and the whole message plays no audio.
|
||||
*/
|
||||
export function splitForSynthesis(text: string, max: number = MAX_TTS_TEXT_LENGTH): string[] {
|
||||
if (text.length <= max) return [text]
|
||||
|
||||
const chunks: string[] = []
|
||||
let rest = text
|
||||
|
||||
while (rest.length > max) {
|
||||
const window = rest.slice(0, max)
|
||||
const boundary = Math.max(
|
||||
window.lastIndexOf(' '),
|
||||
window.lastIndexOf('\n'),
|
||||
window.lastIndexOf('。'),
|
||||
window.lastIndexOf('，'),
|
||||
window.lastIndexOf('、')
|
||||
)
|
||||
const cut = boundary >= max * MIN_SPLIT_RATIO ? boundary + 1 : max
|
||||
const piece = rest.slice(0, cut).trim()
|
||||
if (piece) chunks.push(piece)
|
||||
rest = rest.slice(cut)
|
||||
}
|
||||
|
||||
const tail = rest.trim()
|
||||
if (tail) chunks.push(tail)
|
||||
return chunks
|
||||
}
|
||||
|
||||
declare global {
|
||||
interface Window {
|
||||
webkitAudioContext?: typeof AudioContext
|
||||
@@ -79,7 +117,7 @@ export function useAudioStreaming(sharedAudioContextRef?: RefObject<AudioContext
|
||||
const { text, options } = item
|
||||
const {
|
||||
voiceId,
|
||||
modelId = 'eleven_flash_v2_5',
|
||||
modelId = DEFAULT_TTS_MODEL_ID,
|
||||
chatId,
|
||||
onAudioStart,
|
||||
onAudioEnd,
|
||||
@@ -156,7 +194,9 @@ export function useAudioStreaming(sharedAudioContextRef?: RefObject<AudioContext
|
||||
abortControllerRef.current = new AbortController()
|
||||
}
|
||||
|
||||
audioQueueRef.current.push({ text, options })
|
||||
for (const piece of splitForSynthesis(text)) {
|
||||
audioQueueRef.current.push({ text: piece, options })
|
||||
}
|
||||
processAudioQueue()
|
||||
},
|
||||
[processAudioQueue]
|
||||
|
||||
@@ -0,0 +1,296 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
import {
|
||||
createMockRequest,
|
||||
queueTableRows,
|
||||
resetDbChainMock,
|
||||
resetEnvMock,
|
||||
schemaMock,
|
||||
setEnv,
|
||||
} from '@sim/testing'
|
||||
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const {
|
||||
mockRecordUsage,
|
||||
mockCheckActorUsageLimits,
|
||||
mockResolveSystemBillingAttribution,
|
||||
mockCheckAttributedUsageLimits,
|
||||
mockToBillingContext,
|
||||
mockEnforceIpRateLimit,
|
||||
mockEnforceChatRateLimit,
|
||||
} = vi.hoisted(() => ({
|
||||
mockRecordUsage: vi.fn(),
|
||||
mockCheckActorUsageLimits: vi.fn(),
|
||||
mockResolveSystemBillingAttribution: vi.fn(),
|
||||
mockCheckAttributedUsageLimits: vi.fn(),
|
||||
mockToBillingContext: vi.fn(),
|
||||
mockEnforceIpRateLimit: vi.fn(),
|
||||
mockEnforceChatRateLimit: vi.fn(),
|
||||
}))
|
||||
|
||||
const SYSTEM_BILLING_ATTRIBUTION = {
|
||||
actorUserId: 'payer-1',
|
||||
workspaceId: 'ws-1',
|
||||
organizationId: 'org-1',
|
||||
billedAccountUserId: 'payer-1',
|
||||
billingEntity: { type: 'organization' as const, id: 'org-1' },
|
||||
billingPeriod: {
|
||||
start: '2026-07-01T00:00:00.000Z',
|
||||
end: '2026-08-01T00:00:00.000Z',
|
||||
},
|
||||
payerSubscription: null,
|
||||
}
|
||||
|
||||
vi.mock('@/lib/billing/core/usage-log', () => ({ recordUsage: mockRecordUsage }))
|
||||
|
||||
vi.mock('@/lib/billing/core/billing-attribution', () => ({
|
||||
resolveSystemBillingAttribution: mockResolveSystemBillingAttribution,
|
||||
checkAttributedUsageLimits: mockCheckAttributedUsageLimits,
|
||||
toBillingContext: mockToBillingContext,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/billing/calculations/usage-monitor', () => ({
|
||||
checkActorUsageLimits: mockCheckActorUsageLimits,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/core/rate-limiter/route-helpers', () => ({
|
||||
enforceIpRateLimit: mockEnforceIpRateLimit,
|
||||
enforceChatRateLimit: mockEnforceChatRateLimit,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/core/security/deployment', () => ({ validateAuthToken: vi.fn(() => false) }))
|
||||
|
||||
import { DEFAULT_TTS_VOICE_ID, MAX_TTS_TEXT_LENGTH } from '@/lib/api/contracts/media/tts-stream'
|
||||
import { POST } from '@/app/api/proxy/tts/stream/route'
|
||||
|
||||
const publicChatRow = {
|
||||
id: 'chat-1',
|
||||
userId: 'owner-1',
|
||||
isActive: true,
|
||||
authType: 'public',
|
||||
password: null,
|
||||
workspaceId: 'ws-1',
|
||||
}
|
||||
|
||||
function validBody(overrides: Record<string, unknown> = {}) {
|
||||
return {
|
||||
text: 'Hello from the deployed chat.',
|
||||
voiceId: DEFAULT_TTS_VOICE_ID,
|
||||
chatId: 'chat-1',
|
||||
...overrides,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Minimal ElevenLabs stub returning a readable audio body. Returns the `cancel`
|
||||
* spy so tests can assert the vendor stream is released when we reject.
|
||||
*/
|
||||
function mockElevenLabsAudio() {
|
||||
const stream = new ReadableStream({
|
||||
start(controller) {
|
||||
controller.enqueue(new Uint8Array([0x49, 0x44, 0x33]))
|
||||
controller.close()
|
||||
},
|
||||
})
|
||||
const cancel = vi.fn(() => stream.cancel())
|
||||
global.fetch = vi.fn().mockResolvedValue({
|
||||
ok: true,
|
||||
status: 200,
|
||||
statusText: 'OK',
|
||||
body: { getReader: () => stream.getReader(), cancel },
|
||||
// double-cast-allowed: minimal fetch stub for the ElevenLabs stream call
|
||||
}) as unknown as typeof fetch
|
||||
return cancel
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
resetDbChainMock()
|
||||
setEnv({ ELEVENLABS_API_KEY: 'test-key' })
|
||||
mockEnforceIpRateLimit.mockResolvedValue(null)
|
||||
mockEnforceChatRateLimit.mockResolvedValue(null)
|
||||
mockRecordUsage.mockResolvedValue(undefined)
|
||||
mockCheckActorUsageLimits.mockResolvedValue({ isExceeded: false })
|
||||
mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: false })
|
||||
mockResolveSystemBillingAttribution.mockResolvedValue(SYSTEM_BILLING_ATTRIBUTION)
|
||||
mockToBillingContext.mockImplementation(
|
||||
(attribution: { billingEntity: { type: 'organization' | 'user'; id: string } }) => ({
|
||||
billingEntity: attribution.billingEntity,
|
||||
billingPeriod: {
|
||||
start: new Date('2026-07-01T00:00:00.000Z'),
|
||||
end: new Date('2026-08-01T00:00:00.000Z'),
|
||||
},
|
||||
})
|
||||
)
|
||||
mockElevenLabsAudio()
|
||||
})
|
||||
|
||||
afterAll(() => {
|
||||
resetDbChainMock()
|
||||
resetEnvMock()
|
||||
})
|
||||
|
||||
describe('POST /api/proxy/tts/stream — spend controls', () => {
|
||||
it('caps text length so one request cannot bill unbounded characters', async () => {
|
||||
const res = await POST(
|
||||
createMockRequest('POST', validBody({ text: 'a'.repeat(MAX_TTS_TEXT_LENGTH + 1) }))
|
||||
)
|
||||
|
||||
expect(res.status).toBe(400)
|
||||
expect(global.fetch).not.toHaveBeenCalled()
|
||||
expect(mockRecordUsage).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('rejects an oversized body before buffering it', async () => {
|
||||
const res = await POST(createMockRequest('POST', validBody({ padding: 'x'.repeat(32 * 1024) })))
|
||||
|
||||
expect(res.status).toBe(413)
|
||||
expect(global.fetch).not.toHaveBeenCalled()
|
||||
expect(mockRecordUsage).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('rejects a voice outside the allowlist so the caller cannot pick a premium voice', async () => {
|
||||
const res = await POST(
|
||||
createMockRequest('POST', validBody({ voiceId: '21m00Tcm4TlvDq8ikWAM' }))
|
||||
)
|
||||
|
||||
expect(res.status).toBe(400)
|
||||
expect(global.fetch).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('rejects a model outside the allowlist so the caller cannot pick the billing model', async () => {
|
||||
const res = await POST(
|
||||
createMockRequest('POST', validBody({ modelId: 'eleven_multilingual_v2' }))
|
||||
)
|
||||
|
||||
expect(res.status).toBe(400)
|
||||
expect(global.fetch).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('throttles per IP before any chat lookup so a flood cannot be amplified into queries', async () => {
|
||||
mockEnforceIpRateLimit.mockResolvedValue(new Response('Rate limit exceeded', { status: 429 }))
|
||||
|
||||
const res = await POST(createMockRequest('POST', validBody()))
|
||||
|
||||
expect(res.status).toBe(429)
|
||||
expect(mockEnforceIpRateLimit).toHaveBeenCalledWith('tts-stream', expect.anything(), {
|
||||
maxTokens: 60,
|
||||
refillRate: 30,
|
||||
refillIntervalMs: 60_000,
|
||||
})
|
||||
expect(mockEnforceChatRateLimit).not.toHaveBeenCalled()
|
||||
expect(global.fetch).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('throttles per chat so many callers cannot drain one public chat', async () => {
|
||||
queueTableRows(schemaMock.chat, [publicChatRow])
|
||||
mockEnforceChatRateLimit.mockResolvedValue(new Response('Rate limit exceeded', { status: 429 }))
|
||||
|
||||
const res = await POST(createMockRequest('POST', validBody()))
|
||||
|
||||
expect(res.status).toBe(429)
|
||||
expect(mockEnforceChatRateLimit).toHaveBeenCalledWith('tts-stream', 'chat-1', {
|
||||
maxTokens: 120,
|
||||
refillRate: 60,
|
||||
refillIntervalMs: 60_000,
|
||||
})
|
||||
expect(global.fetch).not.toHaveBeenCalled()
|
||||
expect(mockRecordUsage).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
describe('POST /api/proxy/tts/stream — attribution', () => {
|
||||
it('meters synthesized characters against the chat workspace payer', async () => {
|
||||
queueTableRows(schemaMock.chat, [publicChatRow])
|
||||
const text = 'a'.repeat(1000)
|
||||
|
||||
const res = await POST(createMockRequest('POST', validBody({ text })))
|
||||
|
||||
expect(res.status).toBe(200)
|
||||
expect(mockResolveSystemBillingAttribution).toHaveBeenCalledWith('ws-1')
|
||||
expect(mockCheckAttributedUsageLimits).toHaveBeenCalledWith(SYSTEM_BILLING_ATTRIBUTION)
|
||||
expect(mockRecordUsage).toHaveBeenCalledTimes(1)
|
||||
expect(mockRecordUsage.mock.calls[0][0]).toMatchObject({
|
||||
userId: 'payer-1',
|
||||
workspaceId: 'ws-1',
|
||||
billingEntity: { type: 'organization', id: 'org-1' },
|
||||
})
|
||||
expect(mockRecordUsage.mock.calls[0][0].entries[0]).toMatchObject({
|
||||
category: 'fixed',
|
||||
source: 'voice-output',
|
||||
cost: 0.05,
|
||||
})
|
||||
})
|
||||
|
||||
it('gives each call a unique source reference so equal-length calls are not deduplicated', async () => {
|
||||
const text = 'Same length text.'
|
||||
|
||||
queueTableRows(schemaMock.chat, [publicChatRow])
|
||||
await POST(createMockRequest('POST', validBody({ text })))
|
||||
queueTableRows(schemaMock.chat, [publicChatRow])
|
||||
await POST(createMockRequest('POST', validBody({ text })))
|
||||
|
||||
expect(mockRecordUsage).toHaveBeenCalledTimes(2)
|
||||
const first = mockRecordUsage.mock.calls[0][0].entries[0].sourceReference
|
||||
const second = mockRecordUsage.mock.calls[1][0].entries[0].sourceReference
|
||||
expect(first).toBeDefined()
|
||||
expect(first).not.toBe(second)
|
||||
})
|
||||
|
||||
it('falls back to the chat owner when the workflow has no workspace', async () => {
|
||||
queueTableRows(schemaMock.chat, [{ ...publicChatRow, workspaceId: null }])
|
||||
|
||||
const res = await POST(createMockRequest('POST', validBody()))
|
||||
|
||||
expect(res.status).toBe(200)
|
||||
expect(mockResolveSystemBillingAttribution).not.toHaveBeenCalled()
|
||||
expect(mockCheckActorUsageLimits).toHaveBeenCalledWith('owner-1')
|
||||
expect(mockRecordUsage.mock.calls[0][0]).toMatchObject({ userId: 'owner-1' })
|
||||
})
|
||||
|
||||
it('refuses to spend when the payer is over its usage limit', async () => {
|
||||
queueTableRows(schemaMock.chat, [publicChatRow])
|
||||
mockCheckAttributedUsageLimits.mockResolvedValue({
|
||||
isExceeded: true,
|
||||
message: 'Usage limit exceeded.',
|
||||
})
|
||||
|
||||
const res = await POST(createMockRequest('POST', validBody()))
|
||||
|
||||
expect(res.status).toBe(402)
|
||||
expect(global.fetch).not.toHaveBeenCalled()
|
||||
expect(mockRecordUsage).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('refuses to stream audio it could not record a charge for, releasing the vendor stream', async () => {
|
||||
const cancel = mockElevenLabsAudio()
|
||||
queueTableRows(schemaMock.chat, [publicChatRow])
|
||||
mockRecordUsage.mockRejectedValue(new Error('ledger unavailable'))
|
||||
|
||||
const res = await POST(createMockRequest('POST', validBody()))
|
||||
|
||||
expect(res.status).toBe(500)
|
||||
expect(res.headers.get('Content-Type')).not.toBe('audio/mpeg')
|
||||
expect(cancel).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('rejects an unknown chat without touching the platform key', async () => {
|
||||
queueTableRows(schemaMock.chat, [])
|
||||
|
||||
const res = await POST(createMockRequest('POST', validBody()))
|
||||
|
||||
expect(res.status).toBe(401)
|
||||
expect(global.fetch).not.toHaveBeenCalled()
|
||||
expect(mockRecordUsage).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not expose the audio stream to arbitrary origins', async () => {
|
||||
queueTableRows(schemaMock.chat, [publicChatRow])
|
||||
|
||||
const res = await POST(createMockRequest('POST', validBody()))
|
||||
|
||||
expect(res.status).toBe(200)
|
||||
expect(res.headers.get('Access-Control-Allow-Origin')).toBeNull()
|
||||
})
|
||||
})
|
||||
@@ -1,69 +1,76 @@
|
||||
import { db } from '@sim/db'
|
||||
import { chat } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { generateId } from '@sim/utils/id'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { ttsStreamContract } from '@/lib/api/contracts/media/tts-stream'
|
||||
import { parseRequest } from '@/lib/api/server'
|
||||
import { checkActorUsageLimits } from '@/lib/billing/calculations/usage-monitor'
|
||||
import {
|
||||
checkAttributedUsageLimits,
|
||||
resolveSystemBillingAttribution,
|
||||
toBillingContext,
|
||||
} from '@/lib/billing/core/billing-attribution'
|
||||
import { recordUsage } from '@/lib/billing/core/usage-log'
|
||||
import { resolveDeployedChatCaller } from '@/lib/chat/deployed-chat-caller'
|
||||
import { env } from '@/lib/core/config/env'
|
||||
import { validateAuthToken } from '@/lib/core/security/deployment'
|
||||
import { validateAlphanumericId } from '@/lib/core/security/input-validation'
|
||||
import { getCostMultiplier } from '@/lib/core/config/env-flags'
|
||||
import { enforceChatRateLimit, enforceIpRateLimit } from '@/lib/core/rate-limiter/route-helpers'
|
||||
import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
|
||||
|
||||
const logger = createLogger('ProxyTTSStreamAPI')
|
||||
|
||||
/**
|
||||
* Validates chat-based authentication for deployed chat voice mode
|
||||
* Checks if the user has a valid chat auth cookie for the given chatId
|
||||
* Filters naive floods only: `getClientIp` trusts the leftmost
|
||||
* `X-Forwarded-For` value, which the caller controls, so a deliberate attacker
|
||||
* rotates past this bucket. See {@link TTS_CHAT_RATE_LIMIT}.
|
||||
*
|
||||
* Deployed chat synthesizes sentence by sentence, so a real conversation issues
|
||||
* several requests per answer — hence the generous burst.
|
||||
*/
|
||||
async function validateChatAuth(request: NextRequest, chatId: string): Promise<boolean> {
|
||||
try {
|
||||
const chatResult = await db
|
||||
.select({
|
||||
id: chat.id,
|
||||
isActive: chat.isActive,
|
||||
authType: chat.authType,
|
||||
password: chat.password,
|
||||
})
|
||||
.from(chat)
|
||||
.where(eq(chat.id, chatId))
|
||||
.limit(1)
|
||||
const TTS_IP_RATE_LIMIT = {
|
||||
maxTokens: 60,
|
||||
refillRate: 30,
|
||||
refillIntervalMs: 60 * 1000,
|
||||
} as const
|
||||
|
||||
if (chatResult.length === 0 || !chatResult[0].isActive) {
|
||||
logger.warn('Chat not found or inactive for TTS auth:', chatId)
|
||||
return false
|
||||
}
|
||||
/**
|
||||
* The load-bearing spend control. Public chats hand their id to every visitor,
|
||||
* so the id alone cannot gate use of the platform ElevenLabs key. This bucket
|
||||
* is keyed on server-held state, bounding total spend per chat regardless of
|
||||
* how many source addresses the traffic claims to come from.
|
||||
*/
|
||||
const TTS_CHAT_RATE_LIMIT = {
|
||||
maxTokens: 120,
|
||||
refillRate: 60,
|
||||
refillIntervalMs: 60 * 1000,
|
||||
} as const
|
||||
|
||||
const chatData = chatResult[0]
|
||||
/**
|
||||
* Published ElevenLabs API rate for Flash/Turbo text-to-speech, in USD per
|
||||
* 1,000 characters. This is the vendor cost; `getCostMultiplier()` applies the
|
||||
* platform markup, matching how other metered sources are priced.
|
||||
*/
|
||||
const TTS_COST_PER_1K_CHARS = 0.05
|
||||
|
||||
if (chatData.authType === 'public') {
|
||||
return true
|
||||
}
|
||||
|
||||
const cookieName = `chat_auth_${chatId}`
|
||||
const authCookie = request.cookies.get(cookieName)
|
||||
|
||||
if (
|
||||
authCookie &&
|
||||
validateAuthToken(authCookie.value, chatId, chatData.authType, chatData.password)
|
||||
) {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
} catch (error) {
|
||||
logger.error('Error validating chat auth for TTS:', error)
|
||||
return false
|
||||
}
|
||||
}
|
||||
/**
|
||||
* The body carries at most `MAX_TTS_TEXT_LENGTH` characters of text plus two
|
||||
* short ids, so a tight cap keeps an anonymous caller from making the route
|
||||
* buffer a large payload before validation runs. Without it the shared default
|
||||
* is 50 MB.
|
||||
*/
|
||||
const MAX_TTS_BODY_BYTES = 16 * 1024
|
||||
|
||||
export const POST = withRouteHandler(async (request: NextRequest) => {
|
||||
try {
|
||||
// Throttle per IP before any database work so a flood cannot be amplified into chat lookups.
|
||||
const ipLimited = await enforceIpRateLimit('tts-stream', request, TTS_IP_RATE_LIMIT)
|
||||
if (ipLimited) return ipLimited
|
||||
|
||||
const parsed = await parseRequest(
|
||||
ttsStreamContract,
|
||||
request,
|
||||
{},
|
||||
{
|
||||
maxBodyBytes: MAX_TTS_BODY_BYTES,
|
||||
invalidJsonResponse: () => new NextResponse('Invalid request body', { status: 400 }),
|
||||
validationErrorResponse: (error) => {
|
||||
if (error.issues.some((issue) => issue.path[0] === 'chatId')) {
|
||||
@@ -77,16 +84,27 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
|
||||
|
||||
const { text, voiceId, modelId, chatId } = parsed.data.body
|
||||
|
||||
const isChatAuthed = await validateChatAuth(request, chatId)
|
||||
if (!isChatAuthed) {
|
||||
const caller = await resolveDeployedChatCaller(request, chatId)
|
||||
if (!caller.authorized) {
|
||||
logger.warn('Chat authentication failed for TTS, chatId:', chatId)
|
||||
return new Response('Unauthorized', { status: 401 })
|
||||
}
|
||||
|
||||
const voiceIdValidation = validateAlphanumericId(voiceId, 'voiceId', 255)
|
||||
if (!voiceIdValidation.isValid) {
|
||||
logger.error(`Invalid voice ID: ${voiceIdValidation.error}`)
|
||||
return new Response(voiceIdValidation.error, { status: 400 })
|
||||
const chatLimited = await enforceChatRateLimit('tts-stream', chatId, TTS_CHAT_RATE_LIMIT)
|
||||
if (chatLimited) return chatLimited
|
||||
|
||||
// Anonymous deployed chats have no human request actor, so the workspace payer is charged.
|
||||
const workspaceId = caller.workspaceId ?? undefined
|
||||
const billingAttribution = workspaceId
|
||||
? await resolveSystemBillingAttribution(workspaceId)
|
||||
: undefined
|
||||
const actorUserId = billingAttribution?.actorUserId ?? caller.ownerId
|
||||
|
||||
const usageCheck = billingAttribution
|
||||
? await checkAttributedUsageLimits(billingAttribution)
|
||||
: await checkActorUsageLimits(actorUserId)
|
||||
if (usageCheck.isExceeded) {
|
||||
return new Response(usageCheck.message || 'Usage limit exceeded.', { status: 402 })
|
||||
}
|
||||
|
||||
const apiKey = env.ELEVENLABS_API_KEY
|
||||
@@ -130,6 +148,50 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
|
||||
return new Response('No audio stream received', { status: 422 })
|
||||
}
|
||||
|
||||
/**
|
||||
* Meter once ElevenLabs has accepted the request — the characters are billed
|
||||
* to us at that point regardless of whether the client drains the stream.
|
||||
*
|
||||
* `sourceReference` must be unique per call. `usage_log.event_key` is
|
||||
* unique and inserts conflict-do-nothing, and the key is derived from the
|
||||
* entry's stable fields — without this, two synthesis calls of equal length
|
||||
* in the same workspace would collide and the second would silently go
|
||||
* unbilled. Each call is a separate charge from ElevenLabs, so each needs
|
||||
* its own row rather than being deduplicated. `generateId` rather than
|
||||
* `generateRequestId`, whose fallback truncates to 8 characters.
|
||||
*
|
||||
* A ledger failure fails the request rather than streaming anyway: the
|
||||
* caller is anonymous, so serving audio we could not charge for is exactly
|
||||
* the unmetered spend this route exists to prevent. The vendor call is
|
||||
* already paid for at this point, but the caller gains nothing from it, so
|
||||
* there is no incentive to farm ledger outages.
|
||||
*
|
||||
* No threshold settlement here: it runs per metered event elsewhere and is
|
||||
* far too heavy for a per-sentence realtime path. The workflow execution
|
||||
* that produced this text already settles the payer.
|
||||
*/
|
||||
try {
|
||||
await recordUsage({
|
||||
userId: actorUserId,
|
||||
workspaceId,
|
||||
...(billingAttribution ? toBillingContext(billingAttribution) : {}),
|
||||
entries: [
|
||||
{
|
||||
category: 'fixed',
|
||||
source: 'voice-output',
|
||||
description: `Voice output (${text.length} characters)`,
|
||||
cost: (text.length / 1000) * TTS_COST_PER_1K_CHARS * getCostMultiplier(),
|
||||
sourceReference: `voice-output:${chatId}:${generateId()}`,
|
||||
},
|
||||
],
|
||||
})
|
||||
} catch (err) {
|
||||
logger.error('Failed to record voice output usage, refusing to stream:', err)
|
||||
// Release the open vendor stream; nothing will drain it once we reject.
|
||||
await response.body?.cancel().catch(() => {})
|
||||
return new Response('Unable to record usage for this request', { status: 500 })
|
||||
}
|
||||
|
||||
const { readable, writable } = new TransformStream({
|
||||
transform(chunk, controller) {
|
||||
controller.enqueue(chunk)
|
||||
@@ -166,7 +228,6 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
|
||||
Pragma: 'no-cache',
|
||||
Expires: '0',
|
||||
'X-Content-Type-Options': 'nosniff',
|
||||
'Access-Control-Allow-Origin': '*',
|
||||
Connection: 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
'X-Stream-Type': 'real-time',
|
||||
|
||||
@@ -1,9 +1,6 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { db } from '@sim/db'
|
||||
import { chat, workflow } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { getErrorMessage } from '@sim/utils/errors'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { speechTokenBodySchema } from '@/lib/api/contracts/media/speech'
|
||||
import { parseOptionalJsonBody } from '@/lib/api/server'
|
||||
@@ -18,10 +15,10 @@ import {
|
||||
} from '@/lib/billing/core/billing-attribution'
|
||||
import { recordUsage } from '@/lib/billing/core/usage-log'
|
||||
import { checkAndBillPayerOverageThreshold } from '@/lib/billing/threshold-billing'
|
||||
import { resolveDeployedChatCaller } from '@/lib/chat/deployed-chat-caller'
|
||||
import { env } from '@/lib/core/config/env'
|
||||
import { getCostMultiplier, isBillingEnabled } from '@/lib/core/config/env-flags'
|
||||
import { RateLimiter } from '@/lib/core/rate-limiter'
|
||||
import { validateAuthToken } from '@/lib/core/security/deployment'
|
||||
import { getClientIp } from '@/lib/core/utils/request'
|
||||
import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
|
||||
import { verifyWorkspaceMembership } from '@/app/api/workflows/utils'
|
||||
@@ -55,51 +52,6 @@ function hashVoiceToken(token: string): string {
|
||||
|
||||
const rateLimiter = new RateLimiter()
|
||||
|
||||
async function validateChatAuth(
|
||||
request: NextRequest,
|
||||
chatId: string
|
||||
): Promise<{ valid: boolean; ownerId?: string; workspaceId?: string | null }> {
|
||||
try {
|
||||
const chatResult = await db
|
||||
.select({
|
||||
id: chat.id,
|
||||
userId: chat.userId,
|
||||
isActive: chat.isActive,
|
||||
authType: chat.authType,
|
||||
password: chat.password,
|
||||
workspaceId: workflow.workspaceId,
|
||||
})
|
||||
.from(chat)
|
||||
.leftJoin(workflow, eq(workflow.id, chat.workflowId))
|
||||
.where(eq(chat.id, chatId))
|
||||
.limit(1)
|
||||
|
||||
if (chatResult.length === 0 || !chatResult[0].isActive) {
|
||||
return { valid: false }
|
||||
}
|
||||
|
||||
const chatData = chatResult[0]
|
||||
|
||||
if (chatData.authType === 'public') {
|
||||
return { valid: true, ownerId: chatData.userId, workspaceId: chatData.workspaceId }
|
||||
}
|
||||
|
||||
const cookieName = `chat_auth_${chatId}`
|
||||
const authCookie = request.cookies.get(cookieName)
|
||||
if (
|
||||
authCookie &&
|
||||
validateAuthToken(authCookie.value, chatId, chatData.authType, chatData.password)
|
||||
) {
|
||||
return { valid: true, ownerId: chatData.userId, workspaceId: chatData.workspaceId }
|
||||
}
|
||||
|
||||
return { valid: false }
|
||||
} catch (error) {
|
||||
logger.error('Error validating chat auth for STT:', error)
|
||||
return { valid: false }
|
||||
}
|
||||
}
|
||||
|
||||
export const POST = withRouteHandler(async (request: NextRequest) => {
|
||||
try {
|
||||
const parsedBody = await parseOptionalJsonBody(request, MAX_SPEECH_TOKEN_BODY_BYTES)
|
||||
@@ -113,8 +65,8 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
|
||||
let billingAttribution: BillingAttributionSnapshot | undefined
|
||||
|
||||
if (chatId) {
|
||||
const chatAuth = await validateChatAuth(request, chatId)
|
||||
if (!chatAuth.valid) {
|
||||
const chatAuth = await resolveDeployedChatCaller(request, chatId)
|
||||
if (!chatAuth.authorized) {
|
||||
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
/**
|
||||
|
||||
@@ -17,4 +17,5 @@ export const USAGE_LOG_SOURCE_LABELS: Record<UsageLogSource, string> = {
|
||||
'knowledge-base': 'Knowledge Base',
|
||||
'voice-input': 'Voice input',
|
||||
enrichment: 'Enrichment',
|
||||
'voice-output': 'Voice output',
|
||||
}
|
||||
|
||||
@@ -1,14 +1,33 @@
|
||||
import { z } from 'zod'
|
||||
import { defineRouteContract } from '@/lib/api/contracts/types'
|
||||
|
||||
export const ttsStreamBodySchema = z
|
||||
.object({
|
||||
text: z.string().min(1),
|
||||
voiceId: z.string().min(1),
|
||||
modelId: z.string().optional().default('eleven_flash_v2_5'),
|
||||
chatId: z.string().min(1),
|
||||
})
|
||||
.passthrough()
|
||||
/** Default ElevenLabs voice (Jessica) — Flash v2.5-optimized. */
|
||||
export const DEFAULT_TTS_VOICE_ID = 'cgSgspJ2msm6clMCkdW9'
|
||||
|
||||
export const DEFAULT_TTS_MODEL_ID = 'eleven_flash_v2_5'
|
||||
|
||||
/**
|
||||
* Voices and models the deployed-chat relay may spend the platform ElevenLabs
|
||||
* key on. Anonymous visitors of a public chat can reach the relay, so these are
|
||||
* allowlists rather than free-form strings — otherwise the caller chooses the
|
||||
* (possibly premium or cloned) voice and the billing model.
|
||||
*/
|
||||
const ALLOWED_TTS_VOICE_IDS = [DEFAULT_TTS_VOICE_ID] as const
|
||||
const ALLOWED_TTS_MODEL_IDS = [DEFAULT_TTS_MODEL_ID] as const
|
||||
|
||||
/**
|
||||
* ElevenLabs bills per character, so an uncapped `text` is an uncapped charge.
|
||||
* Chat TTS synthesizes the streamed answer sentence by sentence, keeping real
|
||||
* requests far below this ceiling.
|
||||
*/
|
||||
export const MAX_TTS_TEXT_LENGTH = 2000
|
||||
|
||||
export const ttsStreamBodySchema = z.object({
|
||||
text: z.string().min(1).max(MAX_TTS_TEXT_LENGTH),
|
||||
voiceId: z.enum(ALLOWED_TTS_VOICE_IDS),
|
||||
modelId: z.enum(ALLOWED_TTS_MODEL_IDS).optional().default(DEFAULT_TTS_MODEL_ID),
|
||||
chatId: z.string().min(1),
|
||||
})
|
||||
|
||||
export const ttsStreamContract = defineRouteContract({
|
||||
method: 'POST',
|
||||
|
||||
@@ -278,6 +278,7 @@ export const usageLogSourceSchema = z.enum([
|
||||
'knowledge-base',
|
||||
'voice-input',
|
||||
'enrichment',
|
||||
'voice-output',
|
||||
])
|
||||
|
||||
export const usageLogPeriodSchema = z.enum(['1d', '7d', '30d', 'all', 'custom'])
|
||||
|
||||
@@ -31,6 +31,7 @@ export type UsageLogSource =
|
||||
| 'knowledge-base'
|
||||
| 'voice-input'
|
||||
| 'enrichment'
|
||||
| 'voice-output'
|
||||
|
||||
/**
|
||||
* usage_log sources that make up the "copilot" cost breakdown shown in billing
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
import { db } from '@sim/db'
|
||||
import { chat, workflow } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { and, eq, isNull } from 'drizzle-orm'
|
||||
import type { NextRequest } from 'next/server'
|
||||
import { validateAuthToken } from '@/lib/core/security/deployment'
|
||||
|
||||
const logger = createLogger('DeployedChatCaller')
|
||||
|
||||
export type DeployedChatCaller =
|
||||
| { authorized: false }
|
||||
| { authorized: true; ownerId: string; workspaceId: string | null }
|
||||
|
||||
/**
|
||||
* Resolves whether a caller may act on a deployed chat, and who pays for it.
|
||||
*
|
||||
* Anonymous visitors of a public chat are authorized, so callers that spend a
|
||||
* platform credential must meter against the returned payer rather than treat
|
||||
* authorization as permission to spend freely.
|
||||
*
|
||||
* Shared by every deployed-chat surface that bills a vendor call, so the gate
|
||||
* and the payer are resolved together and cannot drift apart per route.
|
||||
*/
|
||||
export async function resolveDeployedChatCaller(
|
||||
request: NextRequest,
|
||||
chatId: string
|
||||
): Promise<DeployedChatCaller> {
|
||||
try {
|
||||
const rows = await db
|
||||
.select({
|
||||
userId: chat.userId,
|
||||
isActive: chat.isActive,
|
||||
authType: chat.authType,
|
||||
password: chat.password,
|
||||
workspaceId: workflow.workspaceId,
|
||||
})
|
||||
.from(chat)
|
||||
.leftJoin(workflow, eq(workflow.id, chat.workflowId))
|
||||
.where(and(eq(chat.id, chatId), isNull(chat.archivedAt)))
|
||||
.limit(1)
|
||||
|
||||
if (rows.length === 0 || !rows[0].isActive) {
|
||||
logger.warn('Chat not found, archived or inactive', { chatId })
|
||||
return { authorized: false }
|
||||
}
|
||||
|
||||
const chatData = rows[0]
|
||||
const authorized = {
|
||||
authorized: true,
|
||||
ownerId: chatData.userId,
|
||||
workspaceId: chatData.workspaceId,
|
||||
} as const
|
||||
|
||||
if (chatData.authType === 'public') {
|
||||
return authorized
|
||||
}
|
||||
|
||||
const authCookie = request.cookies.get(`chat_auth_${chatId}`)
|
||||
if (
|
||||
authCookie &&
|
||||
validateAuthToken(authCookie.value, chatId, chatData.authType, chatData.password)
|
||||
) {
|
||||
return authorized
|
||||
}
|
||||
|
||||
return { authorized: false }
|
||||
} catch (error) {
|
||||
logger.error('Error resolving deployed chat caller', { chatId, error })
|
||||
return { authorized: false }
|
||||
}
|
||||
}
|
||||
@@ -89,6 +89,23 @@ export async function enforceWorkspaceRateLimit(
|
||||
return buildRateLimitResponse(resetAt)
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a per-deployed-chat token bucket. Use for endpoints an anonymous public
|
||||
* chat visitor can reach: the chat id is server-held state, so this bounds spend
|
||||
* per chat regardless of how many source addresses the traffic claims.
|
||||
*/
|
||||
export async function enforceChatRateLimit(
|
||||
bucketName: string,
|
||||
chatId: string,
|
||||
config: TokenBucketConfig = DEFAULT_USER_ROUTE_LIMIT
|
||||
): Promise<NextResponse | null> {
|
||||
const key = `route:${bucketName}:chat:${chatId}`
|
||||
const { allowed, resetAt } = await rateLimiter.checkRateLimitDirect(key, config)
|
||||
if (allowed) return null
|
||||
logger.warn('Chat rate limit exceeded', { bucket: bucketName, chatId })
|
||||
return buildRateLimitResponse(resetAt)
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a per-user limit when a userId is present, else fall back to per-IP.
|
||||
* Use for routes whose auth path may legitimately resolve without a userId
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
ALTER TYPE "public"."usage_log_source" ADD VALUE 'voice-output';
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1961,6 +1961,13 @@
|
||||
"when": 1785640502989,
|
||||
"tag": "0280_great_riptide",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 281,
|
||||
"version": "7",
|
||||
"when": 1785776917545,
|
||||
"tag": "0281_fixed_madame_web",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -3374,6 +3374,7 @@ export const usageLogSourceEnum = pgEnum('usage_log_source', [
|
||||
'knowledge-base',
|
||||
'voice-input',
|
||||
'enrichment',
|
||||
'voice-output',
|
||||
])
|
||||
|
||||
export const usageLog = pgTable(
|
||||
|
||||
Reference in New Issue
Block a user