From 507426fe20cf7cc7eeabf4ba8d85fbaa68b2b288 Mon Sep 17 00:00:00 2001 From: Christoph Dyllick-Brenzinger Date: Fri, 4 Sep 2026 15:46:00 +0200 Subject: [PATCH] Stop leaking connection slots and reclaim unused sessions (v1.6.3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A client that opened 14 sessions in 51 seconds and made one tool call met "Connection limit exceeded" on everything after. The 20-slot pool per API token was correct; the accounting around it was not. - A POST /mcp without a session ID acquired a slot before the transport opened. If the request never became a session — malformed body, aborted client, transport error — `onclose` never fired and the session was never registered with the idle sweeper, so the slot was unreachable until the process restarted. Released from `res.on('close')` now, and the activeSessions gauge no longer decrements for a session that never was. - A session that initialized and was then abandoned held its slot for the full 10-minute idle timeout. Sessions that have never made a call are now reclaimed after 30 s; sessions doing real work keep the old timeout. - getClientIp read the leftmost X-Forwarded-For entry. Caddy appends the real peer to whatever the client sent, so that entry was attacker controlled: a client could pick its own rate-limit bucket, evade the per-IP limit, or poison the bucket another tenant was counted in. It now reads the rightmost hop, and Caddy overwrites the header rather than appending to it. - "Connection limit exceeded" logged only an IP, which is the proxy's for everyone. It now carries a token fingerprint plus active/limit, so the line says whose pool is full and how full. The limit itself stays at 20. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Tn4FT2sVqoFz1QdpNDNAy7 --- CLAUDE.md | 9 + docker-compose.yml | 5 + package.json | 2 +- src/http/httpServer.ts | 62 +++++- src/ratelimit/connectionCounter.ts | 7 +- tests/connectionSlots.spec.ts | 312 +++++++++++++++++++++++++++++ 6 files changed, 390 insertions(+), 7 deletions(-) create mode 100644 tests/connectionSlots.spec.ts diff --git a/CLAUDE.md b/CLAUDE.md index c8dc767..3e3386a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -50,6 +50,15 @@ Copy `.env.example` to `.env` for local development. - **Selfhosted** (default): Single API token from env, one client per process. Supports multi-base via `SEATABLE_BASES`. - **Managed** (`SEATABLE_MODE=managed`): HTTP-only, each client authenticates with their own Bearer token **on every request** — the `mcp-session-id` header is a routing value, never a credential, and a request must resolve to the identity that created the session. Token validated against SeaTable (`src/auth/tokenValidator.ts`) with positive (1 min) / negative (1 min) cache. Rate limiting via `src/ratelimit/` (per-token, per-IP, global, concurrent connections). +### Connection slots and session lifetime + +The concurrent-connection limit (20, per API token, `src/ratelimit/index.ts`) is acquired on session creation and released on `DELETE`, transport close, or by the idle sweeper. Two rules keep the pool from silting up, both covered by `tests/connectionSlots.spec.ts`: + +- An initialize request that never produces a session (malformed body, aborted client) releases its slot from `res.on('close')`. Without that the slot is unreachable — the transport never opened, so `onclose` never fires, and the sweeper never sees a session that was never registered. +- A session that initialized but never made a call is reclaimed after **30 s** (`unusedSessionTimeoutMs`) instead of the ordinary 10-minute idle timeout. Reconnect-happy clients leave these behind by the dozen; each one holds a slot. + +Client IP for rate limiting comes from the **rightmost** `X-Forwarded-For` entry — the hop our own proxy appended. The leftmost entry is client-supplied and would let a caller pick its own rate-limit bucket. `docker-compose.yml` additionally has Caddy overwrite the header rather than append to it. + ### OAuth (managed mode) `src/auth/oauthProvider.ts` bridges SeaTable API tokens into an OAuth 2.0 authorization code flow. Client registrations and issued tokens are **stateless sealed envelopes** (`src/auth/tokenCipher.ts`, AES-256-GCM keyed from `SEATABLE_TOKEN_SECRET`), so no server-side store is needed and they survive restarts. The `client_id` carries the client's registered `redirect_uris`; `/authorize` rejects anything it cannot open. PKCE `S256` is mandatory and every code is bound to client + exact callback + challenge. The raw SeaTable API token is never returned — `resolveAccessToken()` unseals it server-side. diff --git a/docker-compose.yml b/docker-compose.yml index 80aba98..5ecc6af 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -23,6 +23,11 @@ services: labels: caddy: ${SEATABLE_MCP_HOSTNAME} caddy.reverse_proxy: "{{upstreams 3000}}" + # Replace X-Forwarded-For instead of appending to it. Caddy appends by + # default, which leaves a client-supplied value in front of the real peer + # — and that value is what rate limiting would be keyed on. Overwriting + # here means the header carries exactly one hop: the actual client. + caddy.reverse_proxy.header_up: "X-Forwarded-For {remote_host}" environment: - SEATABLE_SERVER_URL - SEATABLE_API_TOKEN diff --git a/package.json b/package.json index aa3fa84..0798628 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@seatable/mcp-seatable", - "version": "1.6.2", + "version": "1.6.3", "type": "module", "license": "MIT", "mcpName": "io.github.seatable/seatable", diff --git a/src/http/httpServer.ts b/src/http/httpServer.ts index 30980bf..f8228ab 100644 --- a/src/http/httpServer.ts +++ b/src/http/httpServer.ts @@ -17,6 +17,12 @@ export interface StartHttpServerOptions { port?: number /** Session idle timeout in ms (default: 10 minutes) */ sessionIdleTimeoutMs?: number + /** + * Idle timeout for a session that initialized but never made a call + * (default: 30 seconds). These are the sessions a reconnect-happy client + * leaves behind, and each one holds a connection slot while it lives. + */ + unusedSessionTimeoutMs?: number /** Interval for checking idle sessions in ms (default: 60 seconds) */ sessionCheckIntervalMs?: number } @@ -26,6 +32,8 @@ type ActiveSession = { apiToken?: string /** Digest of the SeaTable API token that created this session; every later request must resolve to the same one. */ apiTokenDigest?: Buffer + /** False until the client makes its first call. Sessions that never do are reclaimed far sooner. */ + used: boolean lastActivity: number close: () => Promise } @@ -39,6 +47,11 @@ function sessionFingerprint(sessionId: string): string { return createHash('sha256').update(sessionId).digest('hex').slice(0, 12) } +/** Same rule for API tokens: enough to correlate one tenant's lines, never the credential. */ +function tokenFingerprint(apiToken: string): string { + return createHash('sha256').update(apiToken).digest('hex').slice(0, 12) +} + const MAX_BODY_SIZE = 10 * 1024 * 1024 // 10 MB async function parseJsonBody(req: IncomingMessage): Promise { @@ -155,10 +168,21 @@ export async function startHttpServer(options: StartHttpServerOptions = {}) { const trustProxy = env.TRUST_PROXY ?? true + /** + * The rightmost X-Forwarded-For entry is the one our own reverse proxy + * appended, so it is the only one a client cannot forge. Reading the + * leftmost entry instead let a client choose its own rate-limit bucket: + * evade the per-IP limit by rotating the header, or poison the bucket + * another tenant is being counted in. + */ function getClientIp(req: IncomingMessage): string { if (trustProxy) { const forwarded = req.headers['x-forwarded-for'] - if (typeof forwarded === 'string') return forwarded.split(',')[0].trim() + const chain = Array.isArray(forwarded) ? forwarded.join(',') : forwarded + if (typeof chain === 'string') { + const hops = chain.split(',').map((hop) => hop.trim()).filter(Boolean) + if (hops.length > 0) return hops[hops.length - 1] + } } return req.socket.remoteAddress ?? 'unknown' } @@ -246,7 +270,15 @@ export async function startHttpServer(options: StartHttpServerOptions = {}) { // Connection limit (managed mode) if (rateLimiter && apiToken) { if (!rateLimiter.connections.acquire(apiToken)) { - logger.warn({ ip: getClientIp(req) }, 'Connection limit exceeded') + logger.warn( + { + ip: getClientIp(req), + token: tokenFingerprint(apiToken), + active: rateLimiter.connections.active(apiToken), + limit: rateLimiter.connections.maxConnections, + }, + 'Connection limit exceeded' + ) res.writeHead(429, { 'content-type': 'text/plain' }).end('Too many concurrent connections') return } @@ -254,15 +286,18 @@ export async function startHttpServer(options: StartHttpServerOptions = {}) { } const mcpServer = buildServer(apiToken ? { apiToken } : undefined) + let sessionEstablished = false const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), onsessioninitialized: (id) => { + sessionEstablished = true mcpServer.setSessionId(id) logger.info({ session: sessionFingerprint(id) }, 'Session initialized') sessions.set(id, { transport, apiToken, apiTokenDigest: apiToken ? digest(apiToken) : undefined, + used: false, lastActivity: Date.now(), close: cleanup, }) @@ -274,7 +309,7 @@ export async function startHttpServer(options: StartHttpServerOptions = {}) { const cleanup = async () => { if (cleaned) return cleaned = true - activeSessions.dec() + if (sessionEstablished) activeSessions.dec() if (apiToken && rateLimiter) { rateLimiter.connections.release(apiToken) activeConnections.dec() @@ -298,6 +333,17 @@ export async function startHttpServer(options: StartHttpServerOptions = {}) { void cleanup() } + /* + * An initialize request that never produces a session — a malformed + * body the SDK rejects, a client that hangs up, a transport error — + * leaves the transport unopened, so `onclose` never fires and the + * idle sweeper never sees it either. Without this the slot it took + * above was held until the process restarted. + */ + res.on('close', () => { + if (!sessionEstablished) void cleanup() + }) + await mcpServer.connect(transport) await transport.handleRequest(req, res, body) return @@ -339,6 +385,7 @@ export async function startHttpServer(options: StartHttpServerOptions = {}) { } } + session.used = true session.lastActivity = Date.now() await session.transport.handleRequest(req, res, body) return @@ -483,11 +530,16 @@ export async function startHttpServer(options: StartHttpServerOptions = {}) { // Idle session cleanup const sessionIdleTimeoutMs = options.sessionIdleTimeoutMs ?? 10 * 60 * 1000 const sessionCheckIntervalMs = options.sessionCheckIntervalMs ?? 60 * 1000 + // A session that never made a call gets a much shorter leash than one doing + // real work — it is holding a connection slot for nothing. Never longer + // than the ordinary idle timeout, whatever the caller passes. + const unusedSessionTimeoutMs = Math.min(options.unusedSessionTimeoutMs ?? 30 * 1000, sessionIdleTimeoutMs) const idleCheckInterval = setInterval(() => { const now = Date.now() for (const [sessionId, session] of sessions.entries()) { - if (now - session.lastActivity > sessionIdleTimeoutMs) { - logger.info({ session: sessionFingerprint(sessionId) }, 'Closing idle session') + const timeout = session.used ? sessionIdleTimeoutMs : unusedSessionTimeoutMs + if (now - session.lastActivity > timeout) { + logger.info({ session: sessionFingerprint(sessionId), used: session.used }, 'Closing idle session') void session.close() } } diff --git a/src/ratelimit/connectionCounter.ts b/src/ratelimit/connectionCounter.ts index c161096..9538527 100644 --- a/src/ratelimit/connectionCounter.ts +++ b/src/ratelimit/connectionCounter.ts @@ -2,7 +2,7 @@ * Tracks concurrent connections per key with a configurable limit. */ export class ConnectionCounter { - private readonly maxConnections: number + readonly maxConnections: number private readonly counts = new Map() constructor(maxConnections: number) { @@ -24,4 +24,9 @@ export class ConnectionCounter { this.counts.set(key, current - 1) } } + + /** Slots currently held for a key — for diagnosing an exhausted pool. */ + active(key: string): number { + return this.counts.get(key) ?? 0 + } } diff --git a/tests/connectionSlots.spec.ts b/tests/connectionSlots.spec.ts new file mode 100644 index 0000000..438f378 --- /dev/null +++ b/tests/connectionSlots.spec.ts @@ -0,0 +1,312 @@ +import type { AddressInfo } from 'node:net' + +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' + +vi.mock('../src/metrics/metricsServer', () => ({ + startMetricsServer: vi.fn().mockResolvedValue(undefined), +})) + +const VALID_TOKENS = new Set(['slot-token-1', 'slot-token-2', 'slot-token-3', 'slot-token-4']) + +vi.mock('../src/auth/tokenValidator', () => ({ + TokenValidator: class { + async validate(token: string): Promise { + return VALID_TOKENS.has(token) + } + cleanup(): void {} + destroy(): void {} + }, +})) + +const logCalls: { fields: Record; msg: string }[] = [] +vi.mock('../src/logger', () => { + const record = (a: unknown, b?: unknown) => { + if (typeof a === 'object' && a !== null) logCalls.push({ fields: a as Record, msg: String(b ?? '') }) + else logCalls.push({ fields: {}, msg: String(a) }) + } + return { logger: { fatal: record, error: record, warn: record, info: record, debug: record, trace: record } } +}) + +import { startHttpServer } from '../src/http/httpServer' + +/** + * The production incident these tests encode: a client opened 14 sessions in 51 + * seconds, made one tool call, and then met "Connection limit exceeded" for + * every further attempt. Two separate defects fed it — slots that a failed + * initialize never gave back, and slots that a successful-but-unused session + * held for the full 10-minute idle timeout. + */ + +const JSON_HEADERS = { + 'content-type': 'application/json', + accept: 'application/json, text/event-stream', +} + +const INIT_BODY = JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: '2025-03-26', capabilities: {}, clientInfo: { name: 'test', version: '1.0' } }, +}) + +const find = (needle: string) => logCalls.filter((c) => c.msg.includes(needle)) + +describe('Connection slot accounting', () => { + let server: ReturnType + let baseUrl: string + + beforeAll(async () => { + process.env.SEATABLE_SERVER_URL = 'http://localhost' + process.env.SEATABLE_MODE = 'managed' + process.env.SEATABLE_MOCK = 'true' + process.env.SEATABLE_TOKEN_SECRET = 'connection-slots-spec-secret-long-enough' + delete process.env.SEATABLE_API_TOKEN + + server = await startHttpServer({ port: 0 }) + baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}` + }) + + afterAll(async () => { + delete process.env.SEATABLE_MODE + delete process.env.SEATABLE_TOKEN_SECRET + if (server) await new Promise((resolve) => server.close(() => resolve())) + }) + + beforeEach(() => { + logCalls.length = 0 + }) + + /** + * A POST without a session ID that is not an initialize request acquires a + * slot, then loses it: the transport never opens, so `onclose` never fires, + * and the session never lands in the idle sweeper's map either. Every such + * request used to burn one of the 20 slots until the process restarted. + */ + it('does not leak a slot when the initialize request is rejected', async () => { + const token = 'slot-token-1' + const ip = '198.51.100.1' + + for (let i = 0; i < 21; i++) { + const res = await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + // No session ID and not an initialize call — the SDK rejects this. + body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'tools/list', params: {} }), + }) + expect(res.status).not.toBe(429) + } + + // All 21 slots must have been handed back, so an honest client still gets in. + const res = await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + body: INIT_BODY, + }) + expect(res.status).toBe(200) + expect(res.headers.get('mcp-session-id')).toBeTruthy() + }) + + it('releases the slot when a session is closed with DELETE', async () => { + const token = 'slot-token-2' + const ip = '198.51.100.2' + const ids: string[] = [] + + for (let i = 0; i < 20; i++) { + const res = await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + body: INIT_BODY, + }) + expect(res.status).toBe(200) + ids.push(res.headers.get('mcp-session-id')!) + } + + // Pool is full. + const blocked = await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + body: INIT_BODY, + }) + expect(blocked.status).toBe(429) + + await fetch(`${baseUrl}/mcp`, { + method: 'DELETE', + headers: { 'mcp-session-id': ids[0], authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + }) + + const allowed = await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + body: INIT_BODY, + }) + expect(allowed.status).toBe(200) + + for (const id of ids.slice(1)) { + await fetch(`${baseUrl}/mcp`, { + method: 'DELETE', + headers: { 'mcp-session-id': id, authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + }) + } + }) + + /** + * "Connection limit exceeded" used to log only an IP — and behind a proxy + * that IP is the same for everyone, so the line could not answer "whose + * pool is full?" or "how full?". + */ + it('logs which pool is exhausted and how full it is', async () => { + const token = 'slot-token-3' + const ip = '198.51.100.3' + const ids: string[] = [] + + for (let i = 0; i < 20; i++) { + const res = await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + body: INIT_BODY, + }) + ids.push(res.headers.get('mcp-session-id')!) + } + + await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + body: INIT_BODY, + }) + + const denied = find('Connection limit exceeded') + expect(denied.length).toBeGreaterThan(0) + const fields = denied[denied.length - 1].fields + expect(fields).toHaveProperty('token') + expect(String(fields.token)).not.toContain(token) + expect(fields).toHaveProperty('active', 20) + expect(fields).toHaveProperty('limit', 20) + + for (const id of ids) { + await fetch(`${baseUrl}/mcp`, { + method: 'DELETE', + headers: { 'mcp-session-id': id, authorization: `Bearer ${token}`, 'x-forwarded-for': ip }, + }) + } + }) +}) + +/** + * A session that initializes and is then abandoned is the common client bug. + * It should not hold a slot for the full idle timeout that working sessions get. + */ +describe('Unused session timeout', () => { + let server: ReturnType + let baseUrl: string + + beforeEach(async () => { + process.env.SEATABLE_SERVER_URL = 'http://localhost' + process.env.SEATABLE_API_TOKEN = 'test-token' + process.env.SEATABLE_MOCK = 'true' + delete process.env.SEATABLE_MODE + + server = await startHttpServer({ + port: 0, + sessionIdleTimeoutMs: 60_000, + unusedSessionTimeoutMs: 150, + sessionCheckIntervalMs: 50, + }) + baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}` + }) + + afterEach(async () => { + if (server) await new Promise((resolve) => server.close(() => resolve())) + }) + + async function init(): Promise { + const res = await fetch(`${baseUrl}/mcp`, { method: 'POST', headers: JSON_HEADERS, body: INIT_BODY }) + expect(res.status).toBe(200) + return res.headers.get('mcp-session-id')! + } + + it('reclaims a session that never made a call', async () => { + const sessionId = await init() + + await new Promise((r) => setTimeout(r, 300)) + + const res = await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, 'mcp-session-id': sessionId }, + body: JSON.stringify({ jsonrpc: '2.0', id: 2, method: 'tools/list', params: {} }), + }) + expect(res.status).toBe(404) + }) + + it('keeps a session that has done real work on the full idle timeout', async () => { + const sessionId = await init() + + const worked = await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, 'mcp-session-id': sessionId }, + body: JSON.stringify({ jsonrpc: '2.0', id: 2, method: 'tools/list', params: {} }), + }) + expect(worked.status).toBe(200) + + // Well past the unused timeout, far short of the idle timeout. + await new Promise((r) => setTimeout(r, 300)) + + const res = await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, 'mcp-session-id': sessionId }, + body: JSON.stringify({ jsonrpc: '2.0', id: 3, method: 'tools/list', params: {} }), + }) + expect(res.status).toBe(200) + }) +}) + +/** + * getClientIp took the *first* X-Forwarded-For entry. Caddy appends the real + * peer to whatever the client sent, so the first entry is attacker-controlled: + * a client can pick its own rate-limit bucket, evade the per-IP limit, and + * poison the bucket another tenant is using. + */ +describe('Client IP attribution behind the proxy', () => { + let server: ReturnType + let baseUrl: string + + beforeAll(async () => { + process.env.SEATABLE_SERVER_URL = 'http://localhost' + process.env.SEATABLE_MODE = 'managed' + process.env.SEATABLE_MOCK = 'true' + process.env.SEATABLE_TOKEN_SECRET = 'connection-slots-ip-spec-secret-long-enough' + delete process.env.SEATABLE_API_TOKEN + + server = await startHttpServer({ port: 0 }) + baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}` + }) + + afterAll(async () => { + delete process.env.SEATABLE_MODE + delete process.env.SEATABLE_TOKEN_SECRET + if (server) await new Promise((resolve) => server.close(() => resolve())) + }) + + beforeEach(() => { + logCalls.length = 0 + }) + + it('uses the entry our own proxy appended, not the one the client sent', async () => { + await fetch(`${baseUrl}/mcp`, { + method: 'POST', + headers: { ...JSON_HEADERS, 'x-forwarded-for': '10.9.9.9, 203.0.113.42' }, + body: INIT_BODY, + }) + + const warned = find('Missing Authorization header') + expect(warned.length).toBeGreaterThan(0) + expect(warned[warned.length - 1].fields.ip).toBe('203.0.113.42') + }) + + it('falls back to the socket peer when no header is present', async () => { + await fetch(`${baseUrl}/mcp`, { method: 'POST', headers: JSON_HEADERS, body: INIT_BODY }) + + const warned = find('Missing Authorization header') + expect(warned.length).toBeGreaterThan(0) + expect(String(warned[warned.length - 1].fields.ip)).toContain('127.0.0.1') + }) +})