Merge remote-tracking branch 'origin/staging' into feat/connector-service-account-auth

Author: icecrasher321

apps/sim/app/api/chat/[identifier]/route.test.ts

@@ -14,6 +14,7 @@ import {
   workflowsApiUtilsMock,
   workflowsApiUtilsMockFns,
 } from '@sim/testing'
+import { NextResponse } from 'next/server'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 /**
@@ -65,10 +66,18 @@ const createMockStream = () => {
   })
 }
 
-const { mockValidateChatAuth, mockSetChatAuthCookie, mockProcessChatFiles } = vi.hoisted(() => ({
+const {
+  mockValidateChatAuth,
+  mockSetChatAuthCookie,
+  mockProcessChatFiles,
+  mockEnforceIpRateLimit,
+  mockEnforceResourceRateLimit,
+} = vi.hoisted(() => ({
   mockValidateChatAuth: vi.fn().mockResolvedValue({ authorized: true }),
   mockSetChatAuthCookie: vi.fn(),
   mockProcessChatFiles: vi.fn(),
+  mockEnforceIpRateLimit: vi.fn(),
+  mockEnforceResourceRateLimit: vi.fn(),
 }))
 
 const mockCreateErrorResponse = workflowsApiUtilsMockFns.mockCreateErrorResponse
@@ -117,6 +126,12 @@ vi.mock('@/lib/core/utils/sse', () => ({
 
 vi.mock('@/lib/core/security/encryption', () => encryptionMock)
 
+vi.mock('@/lib/core/rate-limiter', () => ({
+  enforceIpRateLimitWithIndependentBackstop: mockEnforceIpRateLimit,
+  enforceResourceRateLimit: mockEnforceResourceRateLimit,
+}))
+
+import { RATE_LIMITS } from '@/lib/core/rate-limiter/types'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { executeWorkflow } from '@/lib/workflows/executor/execute-workflow'
 import { createStreamingResponse } from '@/lib/workflows/streaming/streaming'
@@ -182,6 +197,8 @@ describe('Chat Identifier API Route', () => {
     })
 
     mockValidateChatAuth.mockResolvedValue({ authorized: true })
+    mockEnforceIpRateLimit.mockResolvedValue(null)
+    mockEnforceResourceRateLimit.mockResolvedValue(null)
     mockProcessChatFiles.mockResolvedValue([])
     mockCreateErrorResponse.mockImplementation((message: string, status: number, code?: string) => {
       return new Response(
@@ -335,6 +352,107 @@ describe('Chat Identifier API Route', () => {
       expect(mockSetChatAuthCookie).toHaveBeenCalledWith(expect.anything(), passwordDeployment)
     })
 
+    describe('execution rate limit', () => {
+      it.each([
+        ['per-IP', mockEnforceIpRateLimit],
+        ['per-deployment', mockEnforceResourceRateLimit],
+      ])("refuses on the %s bucket before the owner's budget is reserved", async (_, bucket) => {
+        bucket.mockResolvedValue(
+          NextResponse.json({ error: 'Rate limit exceeded' }, { status: 429 })
+        )
+        const req = createMockNextRequest('POST', { input: 'drain the wallet' })
+
+        const response = await POST(req, { params: Promise.resolve({ identifier: 'test-chat' }) })
+
+        expect(response.status).toBe(429)
+        expect(preprocessExecution).not.toHaveBeenCalled()
+        expect(createStreamingResponse).not.toHaveBeenCalled()
+        expect(mockProcessChatFiles).not.toHaveBeenCalled()
+      })
+
+      it('debits buckets keyed on the deployment, not the workflow', async () => {
+        const req = createMockNextRequest('POST', { input: 'hello' })
+
+        await POST(req, { params: Promise.resolve({ identifier: 'test-chat' }) })
+
+        expect(mockEnforceIpRateLimit).toHaveBeenCalledWith(
+          'chat-execute',
+          req,
+          expect.objectContaining({ refillIntervalMs: 60_000 }),
+          'chat-id'
+        )
+        expect(mockEnforceResourceRateLimit).toHaveBeenCalledWith(
+          'chat-execute',
+          'chat-id',
+          expect.objectContaining({ refillIntervalMs: 60_000 })
+        )
+      })
+
+      it('leaves the deployment bucket untouched when the IP bucket refuses', async () => {
+        mockEnforceIpRateLimit.mockResolvedValue(NextResponse.json({}, { status: 429 }))
+        const req = createMockNextRequest('POST', { input: 'flood' })
+
+        await POST(req, { params: Promise.resolve({ identifier: 'test-chat' }) })
+
+        expect(mockEnforceResourceRateLimit).not.toHaveBeenCalled()
+      })
+
+      /**
+       * The invariant the ceiling exists to hold. A chat execution debits the
+       * workspace `sync` counter the owner's API, webhook and scheduled runs
+       * share, so a ceiling at or above a plan's own rate never refuses before
+       * that shared counter is drained — the availability half of the attack.
+       * Asserted against every plan, including free, and on burst as well as
+       * sustained rate, since either one reaching the plan bucket first is the
+       * same hole.
+       */
+      it.each(Object.keys(RATE_LIMITS))(
+        'stays under the %s plan sync budget it debits',
+        async (plan) => {
+          const req = createMockNextRequest('POST', { input: 'hello' })
+
+          await POST(req, { params: Promise.resolve({ identifier: 'test-chat' }) })
+
+          const planBucket = RATE_LIMITS[plan as keyof typeof RATE_LIMITS].sync
+          const [, , config] = mockEnforceResourceRateLimit.mock.calls[0]
+          expect(config.refillRate).toBeLessThan(planBucket.refillRate)
+          expect(config.maxTokens).toBeLessThan(planBucket.maxTokens)
+        }
+      )
+
+      /** One host must not be able to take the whole deployment's allowance. */
+      it('holds the per-IP bucket under the per-deployment one', async () => {
+        const req = createMockNextRequest('POST', { input: 'hello' })
+
+        await POST(req, { params: Promise.resolve({ identifier: 'test-chat' }) })
+
+        const [, , ipConfig] = mockEnforceIpRateLimit.mock.calls[0]
+        const [, , deploymentConfig] = mockEnforceResourceRateLimit.mock.calls[0]
+        expect(ipConfig.refillRate).toBeLessThan(deploymentConfig.refillRate)
+      })
+
+      it('leaves the gate-configuration fetch unmetered', async () => {
+        const passwordDeployment = {
+          ...mockChatResult[0],
+          authType: 'password',
+          password: 'encrypted-password',
+        }
+        dbChainMockFns.select.mockImplementation(() => ({
+          from: vi.fn().mockReturnValue({
+            where: vi.fn().mockReturnValue({
+              limit: vi.fn().mockReturnValue([passwordDeployment]),
+            }),
+          }),
+        }))
+        const req = createMockNextRequest('POST', { password: 'test-password' })
+
+        await POST(req, { params: Promise.resolve({ identifier: 'password-protected-chat' }) })
+
+        expect(mockEnforceIpRateLimit).not.toHaveBeenCalled()
+        expect(mockEnforceResourceRateLimit).not.toHaveBeenCalled()
+      })
+    })
+
     it('should return 400 for requests without input', async () => {
       const req = createMockNextRequest('POST', {})
       const params = Promise.resolve({ identifier: 'test-chat' })

apps/sim/app/api/chat/[identifier]/route.ts

@@ -9,6 +9,12 @@ import { parseRequest } from '@/lib/api/server'
 import { releaseExecutionSlot } from '@/lib/billing/calculations/usage-reservation'
 import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate'
 import { env } from '@/lib/core/config/env'
+import {
+  enforceIpRateLimitWithIndependentBackstop,
+  enforceResourceRateLimit,
+  type TokenBucketConfig,
+} from '@/lib/core/rate-limiter'
+import { RATE_LIMITS } from '@/lib/core/rate-limiter/types'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
@@ -49,6 +55,56 @@ export const runtime = 'nodejs'
 
 const CHAT_MAX_REQUEST_BYTES = Number.parseInt(env.CHAT_MAX_REQUEST_BYTES, 10) || 220 * 1024 * 1024
 
+/** A sustained per-minute rate, with the 2x burst allowance the plan buckets use. */
+function executionsPerMinute(perMinute: number): TokenBucketConfig {
+  return { maxTokens: perMinute * 2, refillRate: perMinute, refillIntervalMs: 60_000 }
+}
+
+/**
+ * What one deployed chat may spend of its owner's workspace allowance.
+ *
+ * A chat execution debits the workspace `sync` counter, which is the same
+ * counter the owner's API, webhook and scheduled runs draw from. So this
+ * ceiling only does its job while it sits *below* that counter: above it, a
+ * flood empties the shared budget before this bucket ever refuses, and the
+ * billing attack becomes an availability attack on unrelated production
+ * workloads.
+ *
+ * Derived from the plan table rather than picked, because no fixed number holds
+ * that invariant — the rates differ per plan and every one is operator
+ * overridable through `RATE_LIMIT_*_SYNC`. A fraction of the smallest
+ * configured rate keeps a public chat under the shared budget on every plan and
+ * cannot drift if one of those defaults changes.
+ *
+ * The floor is deliberately shared by all plans for now. Sizing the slice to
+ * the *payer's* own plan needs the subscription, which `preprocessExecution`
+ * resolves a few lines after this runs, not here.
+ *
+ * A configured rate of `1` is the one value where this lands equal to the plan
+ * rather than under it, because no positive integer is below 1. It is inert:
+ * a workspace allowed one execution per minute has no capacity left to starve,
+ * and the two buckets then exhaust together rather than one masking the other.
+ */
+const CHAT_EXECUTION_RATE_PER_MINUTE = Math.max(
+  1,
+  Math.floor(Math.min(...Object.values(RATE_LIMITS).map((plan) => plan.sync.refillRate)) * 0.8)
+)
+
+const CHAT_EXECUTION_LIMIT = executionsPerMinute(CHAT_EXECUTION_RATE_PER_MINUTE)
+
+/**
+ * Executions one client IP may drive against a single deployed chat.
+ *
+ * Half the per-deployment rate, so a single source can never consume the whole
+ * allowance and leave the rest of the audience with none. It is above one
+ * person's chat cadence but not above a busy office behind one NAT — which
+ * costs little in practice, since traffic that heavy from one address would
+ * meet the per-deployment ceiling moments later anyway.
+ */
+const CHAT_EXECUTION_IP_LIMIT = executionsPerMinute(
+  Math.max(1, Math.floor(CHAT_EXECUTION_RATE_PER_MINUTE / 2))
+)
+
 export const POST = withRouteHandler(
   async (request: NextRequest, context: { params: Promise<{ identifier: string }> }) => {
     const { identifier } = await context.params
@@ -169,6 +225,23 @@ export const POST = withRouteHandler(
         return createErrorResponse('No input provided', 400)
       }
 
+      // Both buckets apply regardless of the chat's auth type: an email or SSO
+      // visitor is still not the payer.
+      const ipLimited = await enforceIpRateLimitWithIndependentBackstop(
+        'chat-execute',
+        request,
+        CHAT_EXECUTION_IP_LIMIT,
+        deployment.id
+      )
+      if (ipLimited) return ipLimited
+
+      const deploymentLimited = await enforceResourceRateLimit(
+        'chat-execute',
+        deployment.id,
+        CHAT_EXECUTION_LIMIT
+      )
+      if (deploymentLimited) return deploymentLimited
+
       const executionId = generateId()
 
       const loggingSession = new LoggingSession(

apps/sim/app/api/chat/validate/route.test.ts

@@ -0,0 +1,75 @@
+/**
+ * Tests for the chat identifier availability endpoint.
+ *
+ * @vitest-environment node
+ */
+import { authMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
+import { NextRequest, NextResponse } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockEnforceUserRateLimit } = vi.hoisted(() => ({
+  mockEnforceUserRateLimit: vi.fn(),
+}))
+
+vi.mock('@/lib/core/rate-limiter', () => ({
+  enforceUserRateLimit: mockEnforceUserRateLimit,
+}))
+
+import { GET } from '@/app/api/chat/validate/route'
+
+function request(identifier: string) {
+  return new NextRequest(`http://localhost:3000/api/chat/validate?identifier=${identifier}`)
+}
+
+describe('chat identifier validation route', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    authMockFns.mockGetSession.mockResolvedValue({
+      user: { id: 'user-1' },
+      session: { id: 'session-1' },
+    })
+    mockEnforceUserRateLimit.mockResolvedValue(null)
+  })
+
+  it('refuses an anonymous caller before answering', async () => {
+    authMockFns.mockGetSession.mockResolvedValue(null)
+
+    const response = await GET(request('assistant'))
+
+    expect(response.status).toBe(401)
+    expect(mockEnforceUserRateLimit).not.toHaveBeenCalled()
+  })
+
+  it('reports a taken identifier to a signed-in caller', async () => {
+    queueTableRows(schemaMock.chat, [{ id: 'chat-1' }])
+
+    const response = await GET(request('assistant'))
+
+    expect(response.status).toBe(200)
+    expect(await response.json()).toEqual({
+      available: false,
+      error: 'This identifier is already in use',
+    })
+  })
+
+  it('reports a free identifier to a signed-in caller', async () => {
+    const response = await GET(request('bot'))
+
+    expect(response.status).toBe(200)
+    expect(await response.json()).toEqual({ available: true, error: null })
+  })
+
+  it('caps how far one caller can walk a dictionary', async () => {
+    mockEnforceUserRateLimit.mockResolvedValue(NextResponse.json({}, { status: 429 }))
+
+    const response = await GET(request('support'))
+
+    expect(response.status).toBe(429)
+    expect(mockEnforceUserRateLimit).toHaveBeenCalledWith(
+      'chat-identifier-check',
+      'user-1',
+      expect.objectContaining({ maxTokens: 60, refillIntervalMs: 60_000 })
+    )
+  })
+})

apps/sim/app/api/chat/validate/route.ts

@@ -5,16 +5,39 @@ import { and, eq, isNull } from 'drizzle-orm'
 import type { NextRequest } from 'next/server'
 import { identifierValidationQuerySchema } from '@/lib/api/contracts/chats'
 import { getValidationErrorMessage } from '@/lib/api/server'
+import { getSession } from '@/lib/auth'
+import { enforceUserRateLimit, type TokenBucketConfig } from '@/lib/core/rate-limiter'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
 
 const logger = createLogger('ChatValidateAPI')
 
 /**
- * GET endpoint to validate chat identifier availability
+ * Caps how far one caller can walk a dictionary of identifiers. Sized for a
+ * debounced availability field, which sends one request per pause in typing.
+ */
+const IDENTIFIER_CHECK_RATE_LIMIT: TokenBucketConfig = {
+  maxTokens: 60,
+  refillRate: 60,
+  refillIntervalMs: 60_000,
+}
+
+/**
+ * GET endpoint to validate chat identifier availability.
+ *
+ * Chat identifiers are globally unique, so availability cannot be scoped to a
+ * workspace and there is no resource here to authorize. What the endpoint must
+ * not be is anonymous: `available: false` names a live deployment, and the chat
+ * behind it executes its owner's workflow on their budget for anyone holding
+ * the identifier, so an unmetered answer is a deployment inventory.
  */
 export const GET = withRouteHandler(async (request: NextRequest) => {
   try {
+    const session = await getSession()
+    if (!session?.user?.id) {
+      return createErrorResponse('Unauthorized', 401)
+    }
+
     const { searchParams } = new URL(request.url)
     const identifier = searchParams.get('identifier')
 
@@ -34,6 +57,13 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       return createErrorResponse(errorMessage, 400)
     }
 
+    const rateLimited = await enforceUserRateLimit(
+      'chat-identifier-check',
+      session.user.id,
+      IDENTIFIER_CHECK_RATE_LIMIT
+    )
+    if (rateLimited) return rateLimited
+
     const { identifier: validatedIdentifier } = validation.data
 
     const existingChat = await db