From 0b2d74742abbb17bad9be35444238688f60bf73e Mon Sep 17 00:00:00 2001 From: "helpcode.ai" <340500827+helpcodeai@users.noreply.github.com> Date: Wed, 21 Oct 2026 13:58:22 +0200 Subject: [PATCH 1/4] Cascade kg_value_seen on connector delete; skip gone connectors kg_value_seen.connector_id had no foreign key, so deleted connectors left their value hashes behind and every correlate pass tried to create kg_nodes for them (21.8k FK errors a day on cloud). The migration deletes the orphans and adds the foreign key (ON DELETE CASCADE, NOT VALID + VALIDATE) with an index on connector_id. The ingest now only pairs rows of connectors that existed at the start of the run, caches a refused node per pass, and drops value rows of a connector deleted mid-run instead of failing the batch. --- .../migration.sql | 37 ++++++++++++++ packages/backend/prisma/schema.prisma | 6 +++ .../kg-observational.service.spec.ts | 38 ++++++++++++++- .../kg-observational.service.ts | 48 +++++++++++++------ 4 files changed, 112 insertions(+), 17 deletions(-) create mode 100644 packages/backend/prisma/migrations/20261021100000_kg_value_seen_connector_fk/migration.sql diff --git a/packages/backend/prisma/migrations/20261021100000_kg_value_seen_connector_fk/migration.sql b/packages/backend/prisma/migrations/20261021100000_kg_value_seen_connector_fk/migration.sql new file mode 100644 index 000000000..5d4f8d0ce --- /dev/null +++ b/packages/backend/prisma/migrations/20261021100000_kg_value_seen_connector_fk/migration.sql @@ -0,0 +1,37 @@ +-- ============================================================================= +-- Migration: kg_value_seen rows belong to their connector +-- ============================================================================= +-- kg_value_seen.connector_id had no foreign key, so deleting a connector left +-- its value hashes behind (102 deleted connectors, 57.5k rows on cloud by +-- 21 Oct 2026). The observational ingest kept pairing fresh values with them +-- and tried to create kg_nodes for the deleted connector, which Postgres +-- refused on kg_nodes_connector_id_fkey some 21.8k times a day. +-- +-- Delete the orphans, then add the foreign key with ON DELETE CASCADE plus an +-- index on connector_id, so the cascade does not scan the whole table every +-- time a connector is deleted. +-- +-- Locking: the migration runs as one transaction. On cloud (2.7 M rows, +-- 1.5 GB) the orphan scan takes under a second and the index build about +-- four. CREATE INDEX blocks writes to kg_value_seen only (the KG ingest runs +-- in the background and retries on the next call). The foreign key is added +-- NOT VALID, which is instant, so "connectors" is locked against writes only +-- for the short second delete and VALIDATE that follow. The second delete +-- removes rows of a connector deleted between the first one and the moment +-- the constraint started to apply. +-- ============================================================================= + +DELETE FROM "kg_value_seen" v +WHERE NOT EXISTS (SELECT 1 FROM "connectors" c WHERE c."id" = v."connector_id"); + +CREATE INDEX "kg_value_seen_connector_id_idx" ON "kg_value_seen"("connector_id"); + +ALTER TABLE "kg_value_seen" + ADD CONSTRAINT "kg_value_seen_connector_id_fkey" + FOREIGN KEY ("connector_id") REFERENCES "connectors"("id") + ON DELETE CASCADE ON UPDATE CASCADE NOT VALID; + +DELETE FROM "kg_value_seen" v +WHERE NOT EXISTS (SELECT 1 FROM "connectors" c WHERE c."id" = v."connector_id"); + +ALTER TABLE "kg_value_seen" VALIDATE CONSTRAINT "kg_value_seen_connector_id_fkey"; diff --git a/packages/backend/prisma/schema.prisma b/packages/backend/prisma/schema.prisma index 86024c8c2..4d3e499b5 100644 --- a/packages/backend/prisma/schema.prisma +++ b/packages/backend/prisma/schema.prisma @@ -468,6 +468,7 @@ model Connector { kgNodes KgNode[] kgState KgConnectorState? kgSkills KgSkillSuggestion[] + kgValuesSeen KgValueSeen[] @@index([organizationId]) @@map("connectors") @@ -1460,6 +1461,10 @@ model KgValueSeen { organizationId String @map("organization_id") organization Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade) connectorId String @map("connector_id") + // Deleting a connector deletes its values. Without the relation they stayed + // behind and every correlate pass tried to create nodes for the deleted + // connector (21.8k foreign-key errors a day on cloud by 21 Oct 2026). + connector Connector @relation(fields: [connectorId], references: [id], onDelete: Cascade) valueHash String @map("value_hash") entity String field String @@ -1472,6 +1477,7 @@ model KgValueSeen { @@unique([organizationId, connectorId, valueHash, entity, field, direction], map: "kg_value_seen_occurrence_key") @@index([organizationId, valueHash]) @@index([seenAt]) + @@index([connectorId]) @@map("kg_value_seen") } diff --git a/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts b/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts index 357d19cba..b00860f42 100644 --- a/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts +++ b/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts @@ -226,7 +226,7 @@ describe('KgObservationalService.ingestOrganization', () => { expect(kinds).toEqual(['produces_consumes', 'same_identity']); }); - it('skips values left by a deleted connector instead of aborting the run', async () => { + it('never pairs values left by a connector that no longer exists', async () => { const { svc, prisma } = make({ pages: [[row('r1', 'c2', 1, 'erp_get_order')]], payloads: { r1: { input: {}, output: { order: { customer_id: 'CUST-12345' } } } }, @@ -235,18 +235,52 @@ describe('KgObservationalService.ingestOrganization', () => { { valueHash: 'h1', connectorId: 'gone', entity: 'customer', field: 'id', direction: 'input' }, ], }); + + await expect(svc.ingestOrganization(ORG)).resolves.toBeDefined(); + // No node is even attempted for it: each attempt was a failed insert, + // 21.8k a day in the Postgres log on cloud. + expect(prisma.kgNode.upsert).not.toHaveBeenCalledWith( + expect.objectContaining({ create: expect.objectContaining({ connectorId: 'gone' }) }), + ); + expect(prisma.kgEdge.create).not.toHaveBeenCalled(); + expect(prisma.kgConnectorState.upsert).toHaveBeenCalled(); + }); + + it('skips a connector deleted during the run, asking once instead of once per value', async () => { + const occ = (hash: string) => [ + { valueHash: hash, connectorId: 'c2', entity: 'order', field: 'customer_id', direction: 'output' }, + { valueHash: hash, connectorId: 'c1', entity: 'customer', field: 'id', direction: 'input' }, + ]; + const { svc, prisma } = make({ + pages: [[row('r1', 'c2', 1, 'erp_get_order')]], + payloads: { r1: { input: {}, output: { order: { customer_id: 'CUST-12345' } } } }, + valueSeen: ['h1', 'h2', 'h3'].flatMap(occ), + }); + // c1 was listed at the start of the run and deleted since. prisma.kgNode.upsert.mockImplementation(async (args: any) => { - if (args.create.connectorId === 'gone') { + if (args.create.connectorId === 'c1') { throw Object.assign(new Error('Foreign key constraint violated: kg_nodes_connector_id_fkey'), { code: 'P2003' }); } return { id: `node-${args.create.connectorId}-${args.create.entity}` }; }); await expect(svc.ingestOrganization(ORG)).resolves.toBeDefined(); + const forC1 = prisma.kgNode.upsert.mock.calls.filter((c: any[]) => c[0].create.connectorId === 'c1'); + expect(forC1).toHaveLength(1); expect(prisma.kgEdge.create).not.toHaveBeenCalled(); expect(prisma.kgConnectorState.upsert).toHaveBeenCalled(); }); + it('drops value rows of a connector that is gone when it writes them', async () => { + const { svc, prisma } = make({ + pages: [[row('r1', 'c2', 1, 'erp_get_order')]], + payloads: { r1: { input: {}, output: { order: { customer_id: 'CUST-12345' } } } }, + }); + await svc.ingestOrganization(ORG); + const [sql] = prisma.$executeRaw.mock.calls[0]; + expect(sql.join('?')).toMatch(/WHERE EXISTS \(SELECT 1 FROM connectors c WHERE c\.id = r\.connector_id\)/); + }); + it('writes each edge once per pass, however many values link the same nodes', async () => { const occ = (hash: string) => [ { valueHash: hash, connectorId: 'c2', entity: 'order', field: 'customer_id', direction: 'output' }, diff --git a/packages/backend/src/knowledge-graph/kg-observational.service.ts b/packages/backend/src/knowledge-graph/kg-observational.service.ts index f1a6887b9..b3132cbb1 100644 --- a/packages/backend/src/knowledge-graph/kg-observational.service.ts +++ b/packages/backend/src/knowledge-graph/kg-observational.service.ts @@ -128,6 +128,10 @@ export class KgObservationalService { const slugByConnector = new Map( connectors.map((c) => [c.id, deriveSlug(c.tools.map((t) => t.name))]), ); + // The organization's connectors as of this run. correlate() only pairs + // values of these: a value row of a connector deleted since can never get + // a node (kg_nodes has a foreign key), so trying cost a failed insert. + const liveConnectors = new Set(connectors.map((c) => c.id)); const states = await this.prisma.kgConnectorState.findMany({ where: { organizationId }, select: { connectorId: true, lastObservedAt: true }, @@ -279,7 +283,11 @@ export class KgObservationalService { } const hashes = [...newHashes]; for (let i = 0; i < hashes.length; i += WRITE_CHUNK) { - edges += await this.correlate(organizationId, hashes.slice(i, i + WRITE_CHUNK)); + edges += await this.correlate( + organizationId, + hashes.slice(i, i + WRITE_CHUNK), + liveConnectors, + ); await yieldToEventLoop(); } @@ -291,7 +299,7 @@ export class KgObservationalService { } // while pages // Apply references edges mined from response shapes. - const refCache = new Map(); + const refCache = new Map(); for (const b of refBumps.values()) { const src = await this.nodeId(refCache, organizationId, b.connectorId, b.from); const tgt = await this.nodeId(refCache, organizationId, b.connectorId, b.to); @@ -313,7 +321,7 @@ export class KgObservationalService { // Intent co-occurrence: entities whose tools satisfied the SAME captured // user request are related. A weak, suggested signal (awaits confirmation) // — this is how the chat history that led to calls extends the graph. - const coCache = new Map(); + const coCache = new Map(); for (const members of intentGroups.values()) { const list = [...members]; if (list.length < 2) continue; @@ -422,7 +430,9 @@ export class KgObservationalService { /** * One statement with seven array parameters, however many rows. Prisma's * createMany compiles a parameter per column per row, synchronously: 1,000 - * rows held the production event loop for 330-630 ms per call. + * rows held the production event loop for 330-630 ms per call. Rows of a + * connector deleted while the run was going are dropped rather than + * failing the whole batch on the foreign key. */ private async insertValueRows(rows: ValueRow[]): Promise { if (rows.length === 0) return; @@ -430,7 +440,7 @@ export class KgObservationalService { await this.prisma.$executeRaw` INSERT INTO kg_value_seen (id, organization_id, connector_id, value_hash, entity, field, direction) - SELECT * FROM unnest( + SELECT r.* FROM unnest( ${rows.map(() => randomUUID())}::text[], ${col('organizationId')}::text[], ${col('connectorId')}::text[], @@ -438,7 +448,8 @@ export class KgObservationalService { ${col('entity')}::text[], ${col('field')}::text[], ${col('direction')}::text[] - ) + ) AS r(id, organization_id, connector_id, value_hash, entity, field, direction) + WHERE EXISTS (SELECT 1 FROM connectors c WHERE c.id = r.connector_id) ON CONFLICT (organization_id, connector_id, value_hash, entity, field, direction) DO NOTHING`; } @@ -459,6 +470,7 @@ export class KgObservationalService { private async correlate( organizationId: string, hashes: string[], + liveConnectors: ReadonlySet, ): Promise { const rows = await this.prisma.kgValueSeen.findMany({ where: { organizationId, valueHash: { in: hashes } }, @@ -479,6 +491,9 @@ export class KgObservationalService { const byHash = new Map(); const seen = new Set(); for (const r of rows) { + // Deleting a connector deletes its rows (ON DELETE CASCADE); this + // covers a connector deleted during the run. + if (!liveConnectors.has(r.connectorId)) continue; const key = `${r.valueHash}|${r.connectorId}|${r.entity}|${r.direction}`; if (seen.has(key)) continue; seen.add(key); @@ -487,7 +502,7 @@ export class KgObservationalService { byHash.set(r.valueHash, list); } - const nodeCache = new Map(); // `${connectorId}::${entity}` -> nodeId + const nodeCache = new Map(); // `${connectorId}::${entity}` -> nodeId // Many values link the same two nodes, so one pass used to write the same // edge again for every one of them: a read and an update per value, most // of the database work behind an MCP call. The bumps are counted here and @@ -569,14 +584,15 @@ export class KgObservationalService { } private async nodeId( - cache: Map, + cache: Map, organizationId: string, connectorId: string, entity: string, ): Promise { const key = `${connectorId}::${entity}`; - const cached = cache.get(key); - if (cached) return cached; + // A cached null is a connector that is gone: ask once per pass, not once + // per value that names it. + if (cache.has(key)) return cache.get(key) ?? null; let node: { id: string }; try { node = await this.prisma.kgNode.upsert({ @@ -593,11 +609,13 @@ export class KgObservationalService { select: { id: true }, }); } catch (e: any) { - // kg_value_seen keeps the values of a connector after it is deleted (no - // foreign key there), so a match can name a connector that is gone. A - // node for it would violate kg_nodes_connector_id_fkey and abort the - // whole run for the organization, on every call: skip it instead. - if (e?.code === 'P2003') return null; + // A connector deleted while the run was going: a node for it would + // violate kg_nodes_connector_id_fkey and abort the whole run for the + // organization. Skip it instead. + if (e?.code === 'P2003') { + cache.set(key, null); + return null; + } throw e; } cache.set(key, node.id); From 063105712703dcd8902f523a4b3df99372ab4401 Mon Sep 17 00:00:00 2001 From: "helpcode.ai" <340500827+helpcodeai@users.noreply.github.com> Date: Wed, 21 Oct 2026 13:58:02 +0200 Subject: [PATCH 2/4] Log requests without headers and the SDK JSON-mode warning once Request logs carried every request and response header (about 1.2 KB per /mcp request) and the MCP SDK printed its responseMode 'json' warning for every per-request handler (61k lines a day), so the backend docker log only reached back about 5.7 hours. Requests now log id, method, path without query, MCP method, user agent and client address; responses only the status. The JSON-mode warning is logged once per process; other console warnings are untouched. --- docs/operations/observability.md | 5 +- .../src/common/log-serializers.spec.ts | 79 +++++++++++++++++++ .../backend/src/common/log-serializers.ts | 65 +++++++++++++++ packages/backend/src/common/logger.module.ts | 12 ++- .../src/mcp-server/mcp-endpoint.controller.ts | 6 +- .../src/mcp-server/mcp-handler.util.spec.ts | 49 ++++++++++++ .../src/mcp-server/mcp-handler.util.ts | 51 ++++++++++++ 7 files changed, 260 insertions(+), 7 deletions(-) create mode 100644 packages/backend/src/common/log-serializers.spec.ts create mode 100644 packages/backend/src/common/log-serializers.ts create mode 100644 packages/backend/src/mcp-server/mcp-handler.util.spec.ts create mode 100644 packages/backend/src/mcp-server/mcp-handler.util.ts diff --git a/docs/operations/observability.md b/docs/operations/observability.md index 306895116..1a3c82eaa 100644 --- a/docs/operations/observability.md +++ b/docs/operations/observability.md @@ -8,9 +8,10 @@ Every HTTP request emits one JSON line at completion (or earlier on error), tagg - `req.id` — UUID, present on the response as `X-Request-Id`. Quote it in bug reports. - `userId` / `orgId` / `authMethod` — propagated from the authenticated request. -- `req.method`, `req.url`, `res.statusCode`, `responseTime`. +- `req.method`, `req.url` (path only, no query string), `res.statusCode`, `responseTime`. +- `req.mcpMethod` (the `Mcp-Method` header, on MCP requests), `req.userAgent`, `req.ip` (first `X-Forwarded-For` hop, else the socket address). -Sensitive fields are stripped at the pino layer: +Request and response headers are not logged (`packages/backend/src/common/log-serializers.ts` keeps only the fields above), so no credential header can reach a line. As a second line of defence, pino also strips: - headers: `authorization`, `cookie`, `x-api-key`, `set-cookie`. - bodies: any `password`, `token`, `refreshToken`, `accessToken`, `apiKey`, `secret` field. diff --git a/packages/backend/src/common/log-serializers.spec.ts b/packages/backend/src/common/log-serializers.spec.ts new file mode 100644 index 000000000..0f9636e26 --- /dev/null +++ b/packages/backend/src/common/log-serializers.spec.ts @@ -0,0 +1,79 @@ +import pino from 'pino'; +import { Writable } from 'stream'; +import { serializeRequest, serializeResponse } from './log-serializers'; + +/** What an `/mcp` request looked like in the cloud log on 21 Oct 2026, plus credentials. */ +const mcpRequest = () => ({ + id: 'req-1', + method: 'POST', + url: '/mcp?token=one-time-secret', + query: { token: 'one-time-secret' }, + params: { path: ['mcp'] }, + headers: { + host: 'cloud.anythingmcp.com', + 'user-agent': 'Claude-User', + accept: 'application/json, text/event-stream', + 'content-type': 'application/json', + 'mcp-method': 'tools/list', + 'mcp-protocol-version': '2026-07-28', + via: '1.1 Caddy', + 'x-forwarded-for': '160.79.106.167, 10.0.0.1', + authorization: 'Bearer eyJhbGciOi.secret', + cookie: 'amcp_session=secret', + 'x-api-key': 'mcp_secret', + 'x-custom-token': 'secret', + } as Record, + remoteAddress: '::ffff:172.18.0.6', + remotePort: 32852, +}); + +describe('log serializers', () => { + it('keeps what a request is looked up by, and nothing else', () => { + expect(serializeRequest(mcpRequest())).toEqual({ + id: 'req-1', + method: 'POST', + url: '/mcp', + mcpMethod: 'tools/list', + userAgent: 'Claude-User', + ip: '160.79.106.167', + }); + }); + + it('never writes a credential, whatever header or query parameter carries it', () => { + expect(JSON.stringify(serializeRequest(mcpRequest()))).not.toMatch(/secret/); + }); + + it('falls back to the socket address and leaves out absent headers', () => { + expect( + serializeRequest({ id: 'r', method: 'GET', url: '/api/health', headers: {}, remoteAddress: '127.0.0.1' }), + ).toEqual({ id: 'r', method: 'GET', url: '/api/health', ip: '127.0.0.1' }); + }); + + it('caps a header value a client made huge', () => { + const req = mcpRequest(); + req.headers['user-agent'] = 'x'.repeat(5_000); + expect(serializeRequest(req).userAgent).toHaveLength(200); + }); + + it('keeps only the status of a response', () => { + const res = { statusCode: 200, headers: { 'set-cookie': 'secret' } }; + expect(serializeResponse(res)).toEqual({ statusCode: 200 }); + }); + + it('makes a request line a fraction of its former size through pino', () => { + const lines: string[] = []; + const sink = new Writable({ + write(chunk, _enc, cb) { + lines.push(String(chunk)); + cb(); + }, + }); + const withDefaults = pino({}, sink); + const withOurs = pino({ serializers: { req: serializeRequest } }, sink); + withDefaults.info({ req: mcpRequest() }, 'request completed'); + withOurs.info({ req: mcpRequest() }, 'request completed'); + expect(lines[0]).toMatch(/"headers"/); + expect(lines[1]).not.toMatch(/secret|"headers"/); + expect(lines[1].length).toBeLessThan(lines[0].length / 2); + }); +}); diff --git a/packages/backend/src/common/log-serializers.ts b/packages/backend/src/common/log-serializers.ts new file mode 100644 index 000000000..f838e3fed --- /dev/null +++ b/packages/backend/src/common/log-serializers.ts @@ -0,0 +1,65 @@ +/** + * What a request and its response contribute to a log line. + * + * pino-http's default serializers write every request and response header, + * and the request object is bound to every line logged while the request + * runs. On cloud that was most of the backend log: about 1.2 KB of headers + * per `/mcp` request, so the docker log (5 x 50 MB) only reached back about + * 5.7 hours on 21 Oct 2026. The fields kept here are the ones the logs are + * read for: which request (id), what (method, path, MCP method), from whom + * (client, address; user and workspace come from customProps) and how it + * ended (status; the duration is added by pino-http as responseTime). + * + * Only named headers are read, so a credential can never end up in a line + * whatever header it arrives in. The query string is dropped from the path: + * OAuth callbacks and one-time links carry codes and tokens there. + */ + +/** The request as pino's wrapped serializer hands it over (`raw` is the IncomingMessage). */ +interface SerializedRequestInput { + id?: unknown; + method?: string; + url?: string; + headers?: Record; + remoteAddress?: string; +} + +export interface LoggedRequest { + id?: unknown; + method?: string; + url?: string; + mcpMethod?: string; + userAgent?: string; + ip?: string; +} + +/** Longest header value kept; a client can send anything. */ +const MAX_HEADER_CHARS = 200; + +function header( + headers: SerializedRequestInput['headers'], + name: string, +): string | undefined { + const value = headers?.[name]; + const first = Array.isArray(value) ? value[0] : value; + if (typeof first !== 'string' || !first) return undefined; + return first.length > MAX_HEADER_CHARS ? first.slice(0, MAX_HEADER_CHARS) : first; +} + +export function serializeRequest(req: SerializedRequestInput): LoggedRequest { + const url = typeof req?.url === 'string' ? req.url.split('?')[0] : undefined; + // The first X-Forwarded-For hop is the client as the reverse proxy saw it. + const forwarded = header(req?.headers, 'x-forwarded-for')?.split(',')[0]?.trim(); + const out: LoggedRequest = { id: req?.id, method: req?.method, url }; + const mcpMethod = header(req?.headers, 'mcp-method'); + if (mcpMethod) out.mcpMethod = mcpMethod; + const userAgent = header(req?.headers, 'user-agent'); + if (userAgent) out.userAgent = userAgent; + const ip = forwarded || req?.remoteAddress; + if (ip) out.ip = ip; + return out; +} + +export function serializeResponse(res: { statusCode?: number }): { statusCode?: number } { + return { statusCode: res?.statusCode }; +} diff --git a/packages/backend/src/common/logger.module.ts b/packages/backend/src/common/logger.module.ts index 7b5b7fbee..42e560f15 100644 --- a/packages/backend/src/common/logger.module.ts +++ b/packages/backend/src/common/logger.module.ts @@ -2,6 +2,7 @@ import { Module } from '@nestjs/common'; import { LoggerModule as PinoLoggerModule } from 'nestjs-pino'; import { randomUUID } from 'crypto'; import type { IncomingMessage, ServerResponse } from 'http'; +import { serializeRequest, serializeResponse } from './log-serializers'; /** * Structured logging — Pino over the default NestJS console logger. @@ -12,9 +13,10 @@ import type { IncomingMessage, ServerResponse } from 'http'; * - Per-request correlation id: read from X-Request-Id, generated when * missing, attached to every log line and echoed back on the response so * a client can quote it when reporting bugs. - * - Authorization, cookies and X-API-Key headers are redacted. - * - The MCP endpoint is logged at trace level: it is a high-traffic JSON-RPC - * surface and we don't need a line per request unless debugging. + * - Requests and responses are logged without their headers (see + * log-serializers.ts): method, path, MCP method, client, address and + * status only, so no credential header can reach a line. The redaction + * below stays as a second line of defence. */ @Module({ imports: [ @@ -35,6 +37,10 @@ import type { IncomingMessage, ServerResponse } from 'http'; res.setHeader('X-Request-Id', id); return id; }, + serializers: { + req: serializeRequest, + res: serializeResponse, + }, customProps: (req: IncomingMessage) => ({ // Surface authenticated identity (set by passport / mcp-auth guards) userId: (req as any).user?.sub, diff --git a/packages/backend/src/mcp-server/mcp-endpoint.controller.ts b/packages/backend/src/mcp-server/mcp-endpoint.controller.ts index dd74ce34f..0ee244e5b 100644 --- a/packages/backend/src/mcp-server/mcp-endpoint.controller.ts +++ b/packages/backend/src/mcp-server/mcp-endpoint.controller.ts @@ -16,7 +16,7 @@ import { SkipThrottle, Throttle } from '@nestjs/throttler'; import { Request, Response } from 'express'; import { randomUUID } from 'crypto'; import { z } from 'zod'; -import { McpServer, createMcpHandler, isLegacyRequest } from '@modelcontextprotocol/server'; +import { McpServer, isLegacyRequest } from '@modelcontextprotocol/server'; import { NodeStreamableHTTPServerTransport as StreamableHTTPServerTransport, toNodeHandler, @@ -25,6 +25,7 @@ import { import { McpCombinedAuthGuard } from '../auth/mcp-combined-auth.guard'; import { McpPrincipalRateLimitGuard } from '../auth/mcp-principal-rate-limit.guard'; import { mcpHttpTransport } from './mcp-strategy'; +import { createQuietMcpHandler } from './mcp-handler.util'; import { toolVisibilityRole } from './mcp-server.service'; import { McpServersService } from '../mcp-servers/mcp-servers.service'; import { McpSessionManager } from '../mcp-servers/mcp-session.manager'; @@ -1296,7 +1297,8 @@ export class McpEndpointController { return; } - const handler = createMcpHandler(() => buildServer(), { + // Built per request; its JSON-mode warning is logged once per process. + const handler = createQuietMcpHandler(() => buildServer(), { legacy: 'stateless', // Mapped from the old `enableJsonResponse` boolean, deliberately NOT to // 'auto'. 'auto' answers with JSON whenever it can and upgrades to SSE diff --git a/packages/backend/src/mcp-server/mcp-handler.util.spec.ts b/packages/backend/src/mcp-server/mcp-handler.util.spec.ts new file mode 100644 index 000000000..5075c490a --- /dev/null +++ b/packages/backend/src/mcp-server/mcp-handler.util.spec.ts @@ -0,0 +1,49 @@ +import { Logger } from '@nestjs/common'; +import { McpServer } from '@modelcontextprotocol/server'; +import { + createQuietMcpHandler, + JSON_MODE_WARNING_PREFIX, + resetJsonModeWarningForTests, +} from './mcp-handler.util'; + +/** + * The SDK printed its JSON-mode warning once per handler, and the endpoint + * builds a handler per request: 61k lines a day in the cloud log. + */ +describe('createQuietMcpHandler', () => { + const factory = () => new McpServer({ name: 't', version: '1' }); + let warn: jest.SpyInstance; + let nestWarn: jest.SpyInstance; + + beforeEach(() => { + resetJsonModeWarningForTests(); + warn = jest.spyOn(console, 'warn').mockImplementation(() => undefined); + nestWarn = jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + }); + + afterEach(() => jest.restoreAllMocks()); + + it('logs the JSON-mode warning once per process, however many handlers are built', async () => { + for (let i = 0; i < 5; i++) { + await createQuietMcpHandler(factory, { legacy: 'stateless', responseMode: 'json' }).close(); + } + expect(warn).not.toHaveBeenCalledWith(expect.stringContaining(JSON_MODE_WARNING_PREFIX)); + expect(nestWarn).toHaveBeenCalledTimes(1); + expect(nestWarn).toHaveBeenCalledWith(expect.stringContaining(JSON_MODE_WARNING_PREFIX)); + }); + + it('puts console.warn back, and lets every other warning through', async () => { + const before = console.warn; + await createQuietMcpHandler(factory, { responseMode: 'json' }).close(); + expect(console.warn).toBe(before); + console.warn('something else'); + expect(warn).toHaveBeenCalledWith('something else'); + }); + + it('leaves SSE handlers untouched', async () => { + const before = console.warn; + await createQuietMcpHandler(factory, { responseMode: 'sse' }).close(); + expect(console.warn).toBe(before); + expect(nestWarn).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/backend/src/mcp-server/mcp-handler.util.ts b/packages/backend/src/mcp-server/mcp-handler.util.ts new file mode 100644 index 000000000..c88460d04 --- /dev/null +++ b/packages/backend/src/mcp-server/mcp-handler.util.ts @@ -0,0 +1,51 @@ +import { Logger } from '@nestjs/common'; +import { createMcpHandler } from '@modelcontextprotocol/server'; + +/** Start of the warning `createMcpHandler` prints for `responseMode: 'json'`. */ +export const JSON_MODE_WARNING_PREFIX = "responseMode: 'json' drops mid-call notifications"; + +const logger = new Logger('McpHandler'); +let jsonModeWarningLogged = false; + +/** + * `createMcpHandler`, with its `responseMode: 'json'` warning logged once per + * process instead of once per handler. + * + * The SDK prints the warning with console.warn every time a handler is built + * in JSON mode, and the endpoints build one per request: 61k plain-text lines + * a day on cloud, a third of what pushed the docker log past its cap within + * six hours. JSON mode is chosen on purpose there (MCP_STREAMABLE_JSON_RESPONSE, + * for clients that cannot read SSE), so the warning says nothing new after the + * first time. + * + * Only that one message is held back, and only while the handler is built: + * `createMcpHandler` is synchronous, so console.warn is restored before any + * other code runs, and any other warning printed meanwhile goes through. + */ +export function createQuietMcpHandler( + ...args: Parameters +): ReturnType { + if (args[1]?.responseMode !== 'json') return createMcpHandler(...args); + const original = console.warn; + console.warn = (...warnArgs: unknown[]) => { + const first = warnArgs[0]; + if (typeof first === 'string' && first.startsWith(JSON_MODE_WARNING_PREFIX)) { + if (!jsonModeWarningLogged) { + jsonModeWarningLogged = true; + logger.warn(`${first} (logged once per process)`); + } + return; + } + original.apply(console, warnArgs); + }; + try { + return createMcpHandler(...args); + } finally { + console.warn = original; + } +} + +/** Test hook: forget that the warning was logged. */ +export function resetJsonModeWarningForTests(): void { + jsonModeWarningLogged = false; +} From 84e6da9145180fea11f91204016ce0141dd5e07a Mon Sep 17 00:00:00 2001 From: "helpcode.ai" <340500827+helpcodeai@users.noreply.github.com> Date: Wed, 21 Oct 2026 13:58:02 +0200 Subject: [PATCH 3/4] Tell unverified users' MCP clients how to verify their email The guard refused /mcp for unverified cloud users with a plain 403, which MCP clients show as a broken connection: 13 users who had connected Claude were refused 438 times in under six hours with no explanation. The refusal now goes through a filter that answers with a minimal MCP server: the notice (masked address, sign-in link for a new code) as instructions and as the only tool, and as the result of any tool call. Nothing of the workspace is read and no email is sent, so the verification email rate is unchanged. --- .../src/auth/email-verified.guard.spec.ts | 16 +- .../backend/src/auth/email-verified.guard.ts | 21 ++- .../src/auth/mcp-email-verification.spec.ts | 161 ++++++++++++++++ .../src/auth/mcp-email-verification.ts | 177 ++++++++++++++++++ packages/backend/src/main.ts | 6 +- 5 files changed, 368 insertions(+), 13 deletions(-) create mode 100644 packages/backend/src/auth/mcp-email-verification.spec.ts create mode 100644 packages/backend/src/auth/mcp-email-verification.ts diff --git a/packages/backend/src/auth/email-verified.guard.spec.ts b/packages/backend/src/auth/email-verified.guard.spec.ts index 59a93c3ca..c6e9f763c 100644 --- a/packages/backend/src/auth/email-verified.guard.spec.ts +++ b/packages/backend/src/auth/email-verified.guard.spec.ts @@ -1,5 +1,6 @@ import { ForbiddenException } from '@nestjs/common'; import { EmailVerifiedGuard } from './email-verified.guard'; +import { McpEmailVerificationRequiredException } from './mcp-email-verification'; /** * Global guards run before the route's AuthGuard('jwt'), so req.user is unset @@ -8,7 +9,9 @@ import { EmailVerifiedGuard } from './email-verified.guard'; describe('EmailVerifiedGuard', () => { function makeGuard(opts: { cloud?: boolean; verified?: boolean }) { const prisma = { - user: { findUnique: jest.fn(async () => ({ emailVerified: opts.verified ?? false })) }, + user: { + findUnique: jest.fn(async () => ({ emailVerified: opts.verified ?? false, email: 'jane.doe@example.com' })), + }, }; const auth = { verifyToken: jest.fn((t: string) => { @@ -35,8 +38,17 @@ describe('EmailVerifiedGuard', () => { it('tells an AI client on /mcp what the person has to do', async () => { const { guard } = makeGuard({ verified: false }); const refusal = guard.canActivate(ctx({ path: '/mcp/srv1', headers: { authorization: 'Bearer good' } })); + await expect(refusal).rejects.toBeInstanceOf(McpEmailVerificationRequiredException); await expect(refusal).rejects.toBeInstanceOf(ForbiddenException); - await expect(refusal).rejects.toThrow(/confirm the email address.*sign in at https:\/\//); + await expect(refusal).rejects.toThrow(/not been verified.*ja\*\*\*@example\.com.*Sign in at https:\/\/\S+\/login/); + await expect(refusal).rejects.not.toThrow(/jane\.doe/); + }); + + it('keeps the plain refusal off /mcp', async () => { + const { guard } = makeGuard({ verified: false }); + const refusal = guard.canActivate(ctx({ path: '/api/connectors', headers: { authorization: 'Bearer good' } })); + await expect(refusal).rejects.not.toBeInstanceOf(McpEmailVerificationRequiredException); + await expect(refusal).rejects.toThrow('Email verification required'); }); it('lets a verified user through', async () => { diff --git a/packages/backend/src/auth/email-verified.guard.ts b/packages/backend/src/auth/email-verified.guard.ts index 316e9d605..774cf7dfc 100644 --- a/packages/backend/src/auth/email-verified.guard.ts +++ b/packages/backend/src/auth/email-verified.guard.ts @@ -3,6 +3,10 @@ import { ConfigService } from '@nestjs/config'; import { PrismaService } from '../common/prisma.service'; import { AuthService } from './auth.service'; import { resolveUserIdFromTokenPayload } from './resolve-user-id.util'; +import { + emailVerificationNotice, + McpEmailVerificationRequiredException, +} from './mcp-email-verification'; /** * In cloud mode, users must verify their email before they can access any @@ -80,21 +84,20 @@ export class EmailVerifiedGuard implements CanActivate { const dbUser = await this.prisma.user.findUnique({ where: { id: userId }, - select: { emailVerified: true }, + select: { emailVerified: true, email: true }, }); // Unknown user: not ours to judge (see the class comment). if (!dbUser) return true; if (!dbUser.emailVerified) { if (path.startsWith('/mcp')) { this.logger.warn(`Refused ${path}: user ${userId} has not verified their email`); - // An AI client shows this text to the person, who otherwise only sees - // the connection fail: on 9-10 Oct 2026 seven people who had connected - // Claude retried up to 122 times without knowing what was missing. - const dashboard = (this.configService.get('FRONTEND_URL') || 'https://cloud.anythingmcp.com').replace(/\/+$/, ''); - throw new ForbiddenException( - 'Email verification required: confirm the email address of your AnythingMCP account ' + - `with the link we sent when you signed up, or sign in at ${dashboard} to get a new ` + - 'link. Then try again; nothing else needs to change.', + // The person otherwise only sees the connection fail: on 9-10 Oct + // 2026 seven people who had connected Claude retried up to 122 times + // without knowing what was missing. McpEmailVerificationFilter turns + // this into an MCP answer the client shows and the model can read. + const dashboard = this.configService.get('FRONTEND_URL') || 'https://cloud.anythingmcp.com'; + throw new McpEmailVerificationRequiredException( + emailVerificationNotice(dashboard, dbUser.email), ); } throw new ForbiddenException('Email verification required'); diff --git a/packages/backend/src/auth/mcp-email-verification.spec.ts b/packages/backend/src/auth/mcp-email-verification.spec.ts new file mode 100644 index 000000000..e8be556f0 --- /dev/null +++ b/packages/backend/src/auth/mcp-email-verification.spec.ts @@ -0,0 +1,161 @@ +import { ForbiddenException, Logger } from '@nestjs/common'; +import { Client, StreamableHTTPClientTransport } from '@modelcontextprotocol/client'; +import express from 'express'; +import type { AddressInfo } from 'net'; +import type { Server } from 'http'; +import { + emailVerificationNotice, + maskEmail, + McpEmailVerificationFilter, + McpEmailVerificationRequiredException, + VERIFY_EMAIL_TOOL, +} from './mcp-email-verification'; + +describe('maskEmail', () => { + it('keeps enough to recognise the address and no more', () => { + expect(maskEmail('jane.doe@example.com')).toBe('ja***@example.com'); + expect(maskEmail('jo@example.com')).toBe('j***@example.com'); + expect(maskEmail('j@example.com')).toBe('j***@example.com'); + }); + + it('gives nothing for what is not an address', () => { + expect(maskEmail(undefined)).toBeUndefined(); + expect(maskEmail('no-at-sign')).toBeUndefined(); + expect(maskEmail('@example.com')).toBeUndefined(); + }); +}); + +describe('emailVerificationNotice', () => { + it('says what is missing, where the email went, and how to get a new code', () => { + const text = emailVerificationNotice('https://cloud.anythingmcp.com/', 'jane.doe@example.com'); + expect(text).toMatch(/email address has not been verified/); + expect(text).toContain('ja***@example.com'); + expect(text).toContain('https://cloud.anythingmcp.com/login'); + expect(text).not.toContain('jane.doe'); + }); +}); + +/** + * The guard's refusal, as an MCP client receives it. A plain 403 reads as a + * broken connection; this answers the protocol so the notice reaches the + * person and the model, with nothing of the workspace in it. + */ +describe('McpEmailVerificationFilter', () => { + const NOTICE = emailVerificationNotice('https://cloud.example.com', 'jane.doe@example.com'); + let server: Server; + let base: string; + const savedJson = process.env.MCP_STREAMABLE_JSON_RESPONSE; + + beforeAll(async () => { + jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + const filter = new McpEmailVerificationFilter(); + const app = express(); + app.use(express.json()); + // Stands in for Nest: the guard throws, the filter gets the exception. + app.all('/mcp', (req, res) => { + const host = { switchToHttp: () => ({ getRequest: () => req, getResponse: () => res }) } as any; + filter.catch(new McpEmailVerificationRequiredException(NOTICE), host); + }); + await new Promise((resolve) => { + server = app.listen(0, '127.0.0.1', () => resolve()); + }); + base = `http://127.0.0.1:${(server.address() as AddressInfo).port}/mcp`; + }); + + afterAll(async () => { + process.env.MCP_STREAMABLE_JSON_RESPONSE = savedJson; + jest.restoreAllMocks(); + await new Promise((resolve) => server.close(resolve)); + }); + + const rpc = (method: string, params: Record = {}, id = 1) => + fetch(base, { + method: 'POST', + headers: { + 'content-type': 'application/json', + accept: 'application/json, text/event-stream', + 'mcp-protocol-version': '2025-06-18', + }, + body: JSON.stringify({ jsonrpc: '2.0', id, method, params }), + }); + + /** The JSON-RPC message of a JSON or SSE-framed answer. */ + const message = async (res: Response) => { + const text = await res.text(); + if ((res.headers.get('content-type') ?? '').includes('text/event-stream')) { + const data = text.split('\n').find((l) => l.startsWith('data: ')); + return JSON.parse(data!.slice(6)); + } + return JSON.parse(text); + }; + + const initialize = () => + rpc('initialize', { + protocolVersion: '2025-06-18', + capabilities: {}, + clientInfo: { name: 'test', version: '1' }, + }); + + for (const json of [true, false]) { + describe(json ? 'with JSON responses' : 'with SSE responses', () => { + beforeAll(() => { + process.env.MCP_STREAMABLE_JSON_RESPONSE = json ? 'true' : 'false'; + }); + + it('lets the client connect and carries the notice in the instructions', async () => { + const res = await initialize(); + expect(res.status).toBe(200); + const msg = await message(res); + expect(msg.result.instructions).toBe(NOTICE); + expect(msg.result.serverInfo.name).toBe('AnythingMCP'); + }); + + it('lists only the verification tool', async () => { + const msg = await message(await rpc('tools/list')); + expect(msg.result.tools.map((t: { name: string }) => t.name)).toEqual([VERIFY_EMAIL_TOOL]); + }); + + it('answers any tool call with the notice as an error the model reads', async () => { + const msg = await message(await rpc('tools/call', { name: 'crm_list_customers', arguments: {} })); + const text = JSON.stringify(msg); + expect(text).toContain('ja***@example.com'); + expect(text).not.toContain('jane.doe'); + const call = await message(await rpc('tools/call', { name: VERIFY_EMAIL_TOOL, arguments: {} })); + expect(call.result.isError).toBe(true); + expect(call.result.content[0].text).toBe(NOTICE); + }); + }); + } + + it('works for the current SDK client, whatever protocol revision it negotiates', async () => { + delete process.env.MCP_STREAMABLE_JSON_RESPONSE; + const client = new Client({ name: 'spec', version: '1.0.0' }); + await client.connect(new StreamableHTTPClientTransport(new URL(base))); + try { + expect(client.getInstructions()).toBe(NOTICE); + expect((await client.listTools()).tools.map((t) => t.name)).toEqual([VERIFY_EMAIL_TOOL]); + const result: any = await client.callTool({ name: 'crm_list_customers', arguments: {} }); + expect(result.isError).toBe(true); + expect(result.content[0].text).toBe(NOTICE); + } finally { + await client.close(); + } + }); + + it('refuses a GET and a listen stream with the notice as a JSON-RPC error', async () => { + const get = await fetch(base, { headers: { accept: 'text/event-stream' } }); + expect(get.status).toBe(403); + expect((await get.json()).error.message).toBe(NOTICE); + + const listen = await rpc('subscriptions/listen'); + expect(listen.status).toBe(403); + expect((await listen.json()).error).toEqual({ code: -32001, message: NOTICE }); + }); + + it('is still a 403 with the notice wherever the filter is not in play', () => { + const err = new McpEmailVerificationRequiredException(NOTICE); + expect(err).toBeInstanceOf(ForbiddenException); + expect(err.getStatus()).toBe(403); + expect(err.message).toBe(NOTICE); + }); +}); diff --git a/packages/backend/src/auth/mcp-email-verification.ts b/packages/backend/src/auth/mcp-email-verification.ts new file mode 100644 index 000000000..f100de5b2 --- /dev/null +++ b/packages/backend/src/auth/mcp-email-verification.ts @@ -0,0 +1,177 @@ +import { ArgumentsHost, Catch, ExceptionFilter, ForbiddenException, Logger } from '@nestjs/common'; +import type { Request, Response } from 'express'; +import { McpServer, isLegacyRequest } from '@modelcontextprotocol/server'; +import { + NodeStreamableHTTPServerTransport, + toNodeHandler, + toWebRequest, +} from '@modelcontextprotocol/node'; +import { APP_VERSION } from '../common/app-version'; +import { createQuietMcpHandler } from '../mcp-server/mcp-handler.util'; + +/** + * What an AI client gets from /mcp while its user has not verified their + * email address (cloud only, see EmailVerifiedGuard). + * + * WHY. The guard answered a plain HTTP 403. MCP clients treat that as a + * failed connection and show neither the person nor the model why: on + * 21 Oct 2026, 13 people who had connected Claude without verifying were + * refused 438 times in under six hours, one of them 80 times, with nothing + * telling them what was missing. + * + * The request is still refused: the guard throws before any controller runs, + * and nothing of the workspace is read. This filter only shapes the refusal + * as MCP: a server with no tools of the user's, whose instructions and only + * tool carry the notice, so the client connects and the model can tell the + * person what to do. Nothing here sends an email: a new code comes from + * signing in, as before, so the rate of verification emails is unchanged. + */ + +/** Name of the one tool served to an unverified user. */ +export const VERIFY_EMAIL_TOOL = 'anythingmcp_verify_email'; + +/** `jo***@example.com`: enough for the person to recognise, no more. */ +export function maskEmail(email: string | null | undefined): string | undefined { + if (typeof email !== 'string') return undefined; + const at = email.lastIndexOf('@'); + if (at < 1 || at === email.length - 1) return undefined; + const local = email.slice(0, at); + const keep = local.length > 2 ? 2 : 1; + return `${local.slice(0, keep)}***${email.slice(at)}`; +} + +/** The text shown to the person and the model. */ +export function emailVerificationNotice(dashboardUrl: string, email?: string | null): string { + const base = dashboardUrl.replace(/\/+$/, ''); + const masked = maskEmail(email); + return ( + 'Your AnythingMCP account cannot be used yet: its email address has not been verified, ' + + 'so no tools are available. ' + + `We sent a verification email${masked ? ` to ${masked}` : ''} when you signed up: open the link ` + + 'in it (or enter its code), then try again. ' + + `No email, or the code expired? Sign in at ${base}/login and a new code is sent; check the spam ` + + 'folder too. Nothing needs to change in your AI client.' + ); +} + +/** Thrown by EmailVerifiedGuard on /mcp; a 403 with the notice wherever this filter does not apply. */ +export class McpEmailVerificationRequiredException extends ForbiddenException { + constructor(readonly notice: string) { + super(notice); + } +} + +/** A tool name worth answering under (what MCP clients send; anything else gets "not found"). */ +const TOOL_NAME = /^[A-Za-z0-9_.-]{1,128}$/; + +/** + * The server an unverified user is answered from: the notice, and nothing of + * theirs. `calledTool` is the tool the request calls, if any: it answers with + * the notice too, so a client that still knows a tool from earlier gets the + * explanation rather than "tool not found". It is never listed. + */ +export function emailVerificationServer(notice: string, calledTool?: string): McpServer { + const server = new McpServer( + { name: 'AnythingMCP', version: APP_VERSION }, + { capabilities: { tools: {} }, instructions: notice }, + ); + const answer = async () => ({ content: [{ type: 'text' as const, text: notice }], isError: true }); + server.registerTool( + VERIFY_EMAIL_TOOL, + { + description: + 'The AnythingMCP account behind this connection has not verified its email address, so its ' + + 'tools are unavailable. Call this to get the steps, and pass them on to the user.', + annotations: { readOnlyHint: true, openWorldHint: false }, + }, + answer, + ); + if (calledTool && calledTool !== VERIFY_EMAIL_TOOL && TOOL_NAME.test(calledTool)) { + server.registerTool(calledTool, { description: notice }, answer); + } + return server; +} + +/** The tool a `tools/call` request names. */ +function calledToolOf(body: unknown): string | undefined { + const b = body as { method?: unknown; params?: { name?: unknown } }; + return b?.method === 'tools/call' && typeof b.params?.name === 'string' ? b.params.name : undefined; +} + +/** A single JSON-RPC message other than a subscriptions/listen stream. */ +function isServableRequest(body: unknown): boolean { + if (!body || typeof body !== 'object' || Array.isArray(body)) return false; + const method = (body as { method?: unknown }).method; + // A listen stream would stay open for minutes and has nothing to deliver. + return typeof method === 'string' && method !== 'subscriptions/listen'; +} + +@Catch(McpEmailVerificationRequiredException) +export class McpEmailVerificationFilter implements ExceptionFilter { + private readonly logger = new Logger(McpEmailVerificationFilter.name); + + catch(exception: McpEmailVerificationRequiredException, host: ArgumentsHost): void { + const ctx = host.switchToHttp(); + const req = ctx.getRequest(); + const res = ctx.getResponse(); + if (res.headersSent) return; + + const body = (req as { body?: unknown }).body; + if (req.method === 'POST' && isServableRequest(body)) { + void this.serve(req, res, body, exception.notice).catch((err) => { + this.logger.warn(`Could not serve the email verification notice: ${err?.message ?? err}`); + this.refuse(res, exception.notice); + }); + return; + } + this.refuse(res, exception.notice); + } + + private refuse(res: Response, notice: string): void { + if (res.headersSent) return; + res.status(403).json({ jsonrpc: '2.0', error: { code: -32001, message: notice }, id: null }); + } + + private async isLegacy(req: Request, body: unknown): Promise { + try { + return await isLegacyRequest(await toWebRequest(req, body), body); + } catch { + return false; + } + } + + /** + * Same framing as the MCP endpoint (McpEndpointController.serveStateless): + * the handler for 2026-07-28 traffic, and single-shot JSON for 2025-era + * requests when JSON responses are enabled, which the handler's own + * fallback cannot produce. + */ + private async serve(req: Request, res: Response, body: unknown, notice: string): Promise { + const json = process.env.MCP_STREAMABLE_JSON_RESPONSE === 'true'; + const build = () => emailVerificationServer(notice, calledToolOf(body)); + if (json && (await this.isLegacy(req, body))) { + const server = build(); + const transport = new NodeStreamableHTTPServerTransport({ + sessionIdGenerator: undefined, + enableJsonResponse: true, + }); + try { + await server.connect(transport); + await transport.handleRequest(req, res, body); + } finally { + await transport.close().catch(() => undefined); + await server.close().catch(() => undefined); + } + return; + } + const handler = createQuietMcpHandler(build, { + legacy: 'stateless', + responseMode: json ? 'json' : 'sse', + }); + try { + await toNodeHandler(handler)(req, res, body); + } finally { + await handler.close().catch(() => undefined); + } + } +} diff --git a/packages/backend/src/main.ts b/packages/backend/src/main.ts index 7e7051c6c..f6be80277 100644 --- a/packages/backend/src/main.ts +++ b/packages/backend/src/main.ts @@ -29,6 +29,7 @@ import helmet from 'helmet'; import { AppModule } from './app.module'; import { mcpStrategy } from './mcp-server/mcp-strategy'; import { McpAuthExceptionFilter } from './auth/mcp-auth-exception.filter'; +import { McpEmailVerificationFilter } from './auth/mcp-email-verification'; import { validateRequiredSecretsAtStartup } from './common/secrets.util'; import { startMetricsServer, stopMetricsServer } from './common/metrics'; @@ -104,8 +105,9 @@ async function bootstrap() { }), ); - // Add WWW-Authenticate header to MCP 401 responses for OAuth discovery - app.useGlobalFilters(new McpAuthExceptionFilter()); + // Add WWW-Authenticate header to MCP 401 responses for OAuth discovery, + // and answer an unverified cloud user's MCP client with what to do. + app.useGlobalFilters(new McpAuthExceptionFilter(), new McpEmailVerificationFilter()); // Global validation app.useGlobalPipes( From 2bb4d4c2d4b2ed54b712b307bbf1b62b8ecef360 Mon Sep 17 00:00:00 2001 From: "helpcode.ai" <340500827+helpcodeai@users.noreply.github.com> Date: Wed, 21 Oct 2026 13:58:02 +0200 Subject: [PATCH 4/4] Move expired tokens from expiring to needs_reauth and email once A token marked expiring whose token_expires_at had passed stayed a warning when the connector was not called again, and its owner was never told (a meta-ads token expired on 20 Oct at 17:00 was still expiring a day later). An hourly sweep now marks such connectors needs_reauth (token_expired) against the same credential, which sends the single owner email, and the bearer renewal service does the same when it sees an expired token. --- .../bearer-token-renewal.service.spec.ts | 24 ++++++ .../engines/bearer-token-renewal.service.ts | 21 ++++- .../connector-auth-status.service.spec.ts | 62 +++++++++++++++ .../engines/connector-auth-status.service.ts | 77 ++++++++++++++++++- 4 files changed, 179 insertions(+), 5 deletions(-) diff --git a/packages/backend/src/connectors/engines/bearer-token-renewal.service.spec.ts b/packages/backend/src/connectors/engines/bearer-token-renewal.service.spec.ts index 3161471d0..24bd053cf 100644 --- a/packages/backend/src/connectors/engines/bearer-token-renewal.service.spec.ts +++ b/packages/backend/src/connectors/engines/bearer-token-renewal.service.spec.ts @@ -189,6 +189,30 @@ describe('BearerTokenRenewalService', () => { expect(mockedAxios.get).toHaveBeenCalledTimes(1); }); + it('once a short-lived token has expired, marks the connector as needing a new token instead of warning', async () => { + const expiresAt = Date.now() + 3_600_000; + graph({ '/debug_token': () => ({ data: { expires_at: Math.floor(expiresAt / 1000) } }) }); + expect(await service.getToken(withoutSecret(), 'conn-lapsed')).toBe(SHORT); + expect(authStatus.mark).toHaveBeenLastCalledWith( + 'conn-lapsed', + expect.objectContaining({ status: 'expiring', code: 'meta_expiring' }), + ); + + // Two hours later, nobody has pasted a new token. + const now = Date.now(); + jest.spyOn(Date, 'now').mockReturnValue(now + 2 * 3_600_000); + // The call still goes out: Meta's answer has the last word. + expect(await service.getToken(withoutSecret(), 'conn-lapsed')).toBe(SHORT); + expect(authStatus.mark).toHaveBeenLastCalledWith('conn-lapsed', { + status: 'needs_reauth', + code: 'token_expired', + message: expect.stringMatching(/expired on .* UTC and could not be renewed.*Paste a new token/), + ref: credentialRef(SHORT), + }); + await service.getToken(withoutSecret(), 'conn-lapsed'); + expect(authStatus.mark).toHaveBeenCalledTimes(2); + }); + it('records a refused exchange as a warning and keeps using the pasted token', async () => { graph({ '/debug_token': () => graphError(190, undefined, 400), diff --git a/packages/backend/src/connectors/engines/bearer-token-renewal.service.ts b/packages/backend/src/connectors/engines/bearer-token-renewal.service.ts index 5d11fb5fd..1abcaa1b1 100644 --- a/packages/backend/src/connectors/engines/bearer-token-renewal.service.ts +++ b/packages/backend/src/connectors/engines/bearer-token-renewal.service.ts @@ -6,7 +6,7 @@ import { decrypt, encrypt } from '../../common/crypto/encryption.util'; import { getRequiredSecret } from '../../common/secrets.util'; import { connectorPageUrl } from '../../common/url.util'; import { getAdapter } from '../../adapters/catalog'; -import { ConnectorAuthStatusService, credentialRef } from './connector-auth-status.service'; +import { ConnectorAuthStatusService, credentialRef, expiredTokenHint } from './connector-auth-status.service'; /** * Renewal of pasted bearer tokens that expire, for providers that let a @@ -411,7 +411,12 @@ export class BearerTokenRenewalService { await this.authStatus?.clear(connectorId, { onlyCodes: WARNING_CODES }); } - /** A token that will stop working soon and cannot be renewed here: say so. */ + /** + * A token that will stop working soon and cannot be renewed here: say so. + * One that already has is no longer a warning: the connector needs a new + * token, and its owner gets the single email (an expired token cannot be + * exchanged). The call still goes out; Meta's answer has the last word. + */ private async warnIfExpiring( connectorId: string, st: TokenState, @@ -419,6 +424,18 @@ export class BearerTokenRenewalService { failure?: string, ): Promise { if (typeof st.expiresAt !== 'number') return; + if (st.expiresAt <= Date.now()) { + const message = expiredTokenHint(new Date(st.expiresAt), connectorId); + if (this.warned.get(connectorId) === message) return; + this.warned.set(connectorId, message); + await this.authStatus?.mark(connectorId, { + status: 'needs_reauth', + code: 'token_expired', + message, + ref: st.sourceRef, + }); + return; + } const within = (spec.renewWithinDays ?? DEFAULT_RENEW_WITHIN_DAYS) * DAY_MS; if (st.expiresAt - Date.now() > within) return; if (!failure && this.shouldRenew(st, spec)) return; // will be renewed on this call diff --git a/packages/backend/src/connectors/engines/connector-auth-status.service.spec.ts b/packages/backend/src/connectors/engines/connector-auth-status.service.spec.ts index cb7ea63f3..1cbcea640 100644 --- a/packages/backend/src/connectors/engines/connector-auth-status.service.spec.ts +++ b/packages/backend/src/connectors/engines/connector-auth-status.service.spec.ts @@ -21,6 +21,11 @@ function fakePrisma(row: Record) { connector: { findUnique: jest.fn(async ({ where }: any) => (where.id === row.id ? { ...row } : null)), update: jest.fn(async ({ data }: any) => Object.assign(row, data)), + findMany: jest.fn(async ({ where }: any) => + row.authStatus === where.authStatus && row.tokenExpiresAt && row.tokenExpiresAt <= where.tokenExpiresAt.lte + ? [{ id: row.id, authStatusRef: row.authStatusRef ?? null, tokenExpiresAt: row.tokenExpiresAt }] + : [], + ), updateMany: jest.fn(async ({ where, data }: any) => { if (!matches(row, where)) return { count: 0 }; Object.assign(row, data); @@ -122,6 +127,63 @@ describe('ConnectorAuthStatusService', () => { expect(issue.ref).toMatch(/^[0-9a-f]{16}$/); }); + describe('expireLapsedTokens', () => { + const ref = credentialRef('EAAshortlived'); + const expiresAt = new Date('2026-10-20T17:00:00Z'); + const warn = { status: 'expiring' as const, code: 'meta_expiring', message: 'This is a short-lived token.', ref }; + + it('turns a warning whose token has expired into needs_reauth and emails the owner once', async () => { + await service.mark('conn-1', warn); + row.tokenExpiresAt = expiresAt; + await flush(); + expect(email.sendConnectorAuthEmail).not.toHaveBeenCalled(); + + const later = new Date('2026-10-21T10:00:00Z'); + expect(await service.expireLapsedTokens(later)).toBe(1); + await flush(); + expect(row).toMatchObject({ authStatus: 'needs_reauth', authStatusCode: 'token_expired', authStatusRef: ref }); + expect(row.authStatusMessage).toMatch(/expired on 2026-10-20 17:00 UTC and could not be renewed/); + expect(row.authNotifiedAt).toBeInstanceOf(Date); + expect(email.sendConnectorAuthEmail).toHaveBeenCalledTimes(1); + + // The next sweep finds nothing left to do. + expect(await service.expireLapsedTokens(later)).toBe(0); + await flush(); + expect(email.sendConnectorAuthEmail).toHaveBeenCalledTimes(1); + }); + + it('leaves a warning alone until its token has expired', async () => { + await service.mark('conn-1', warn); + row.tokenExpiresAt = expiresAt; + expect(await service.expireLapsedTokens(new Date('2026-10-20T16:59:00Z'))).toBe(0); + expect(row.authStatus).toBe('expiring'); + }); + + it('only looks at expiring warnings, never at other connectors with an expiry', async () => { + await service.expireLapsedTokens(new Date()); + expect(prisma.connector.findMany).toHaveBeenCalledWith( + expect.objectContaining({ where: { authStatus: 'expiring', tokenExpiresAt: { lte: expect.any(Date) } } }), + ); + }); + }); + + it('runs the expiry sweep on a timer that does not keep the process alive', () => { + jest.useFakeTimers(); + try { + const spy = jest.spyOn(service, 'expireLapsedTokens').mockResolvedValue(0); + service.onModuleInit(); + jest.advanceTimersByTime(2 * 60 * 1000); + expect(spy).toHaveBeenCalledTimes(1); + jest.advanceTimersByTime(60 * 60 * 1000); + expect(spy).toHaveBeenCalledTimes(2); + service.onModuleDestroy(); + jest.advanceTimersByTime(3 * 60 * 60 * 1000); + expect(spy).toHaveBeenCalledTimes(2); + } finally { + jest.useRealTimers(); + } + }); + it('works without an email service (self-hosted without SMTP)', async () => { const quiet = new ConnectorAuthStatusService(prisma as any); await quiet.mark('conn-1', issue); diff --git a/packages/backend/src/connectors/engines/connector-auth-status.service.ts b/packages/backend/src/connectors/engines/connector-auth-status.service.ts index ba3ff35e1..27c044782 100644 --- a/packages/backend/src/connectors/engines/connector-auth-status.service.ts +++ b/packages/backend/src/connectors/engines/connector-auth-status.service.ts @@ -1,6 +1,7 @@ -import { Injectable, Logger, Optional } from '@nestjs/common'; +import { Injectable, Logger, OnModuleDestroy, OnModuleInit, Optional } from '@nestjs/common'; import { pbkdf2Sync } from 'crypto'; import { PrismaService } from '../../common/prisma.service'; +import { connectorPageUrl } from '../../common/url.util'; import { EmailService } from '../../settings/email.service'; /** @@ -18,7 +19,9 @@ import { EmailService } from '../../settings/email.service'; * Statuses: * needs_reauth blocking: calls fail at once with `message` until the * credential changes (`ref` is a hash of the one refused) - * expiring a warning: the token works, but stops at tokenExpiresAt + * expiring a warning: the token works, but stops at tokenExpiresAt; + * once that time has passed it becomes needs_reauth + * (token_expired), on the next call or the hourly sweep * * Nothing here is ever a secret: `ref` is a truncated SHA-256. */ @@ -52,20 +55,88 @@ export function credentialRef(value: unknown): string | undefined { return ref; } +/** How often connectors whose token expired while still marked `expiring` are looked for. */ +const EXPIRY_SWEEP_MS = 60 * 60 * 1000; +/** First sweep after boot: late enough to stay out of the startup work. */ +const EXPIRY_SWEEP_FIRST_MS = 2 * 60 * 1000; +/** Connectors handled per sweep; the rest wait for the next one. */ +const EXPIRY_SWEEP_BATCH = 200; + +/** The status message of a token past its expiry that nothing could renew. */ +export function expiredTokenHint(expiresAt: Date, connectorId: string): string { + const when = expiresAt.toISOString().replace('T', ' ').slice(0, 16) + ' UTC'; + const page = connectorPageUrl(connectorId); + return ( + `The access token expired on ${when} and could not be renewed; retrying will not help. ` + + `Paste a new token into the connector's variables in AnythingMCP${page ? ` (${page})` : ''}. ` + + 'For Meta, a System User token with expiry Never does not expire, and a user token is renewed ' + + 'automatically when the app id and secret are set.' + ); +} + /** The credential a status is recorded against: the refresh token, else the pasted token. */ export function authCredentialRef(authConfig: Record | undefined): string | undefined { return credentialRef(authConfig?.refreshToken) ?? credentialRef(authConfig?.token); } @Injectable() -export class ConnectorAuthStatusService { +export class ConnectorAuthStatusService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(ConnectorAuthStatusService.name); + private sweepTimer?: ReturnType; + private firstSweep?: ReturnType; constructor( private readonly prisma: PrismaService, @Optional() private readonly email?: EmailService, ) {} + onModuleInit(): void { + const sweep = () => void this.expireLapsedTokens(); + this.firstSweep = setTimeout(sweep, EXPIRY_SWEEP_FIRST_MS); + this.firstSweep.unref?.(); + this.sweepTimer = setInterval(sweep, EXPIRY_SWEEP_MS); + this.sweepTimer.unref?.(); + } + + onModuleDestroy(): void { + if (this.firstSweep) clearTimeout(this.firstSweep); + if (this.sweepTimer) clearInterval(this.sweepTimer); + } + + /** + * An `expiring` warning whose token has expired is no longer a warning. + * The next call notices it too (Meta refuses the token, or the renewal + * service sees the expiry), but a connector nobody calls stayed `expiring` + * for good and its owner was never told: a meta-ads token that expired on + * 20 Oct 2026 at 17:00 was still a warning, with no email, a day later. + * Each one becomes needs_reauth against the same credential, which sends + * the single owner email like any other refusal. Returns how many moved. + */ + async expireLapsedTokens(now: Date = new Date()): Promise { + let rows: Array<{ id: string; authStatusRef: string | null; tokenExpiresAt: Date | null }>; + try { + rows = await this.prisma.connector.findMany({ + where: { authStatus: 'expiring', tokenExpiresAt: { lte: now } }, + select: { id: true, authStatusRef: true, tokenExpiresAt: true }, + orderBy: { tokenExpiresAt: 'asc' }, + take: EXPIRY_SWEEP_BATCH, + }); + } catch (err: any) { + this.logger.warn(`Could not look for expired connector tokens: ${err?.message ?? err}`); + return 0; + } + for (const row of rows) { + if (!row.tokenExpiresAt) continue; + await this.mark(row.id, { + status: 'needs_reauth', + code: 'token_expired', + message: expiredTokenHint(row.tokenExpiresAt, row.id), + ref: row.authStatusRef ?? undefined, + }); + } + return rows.length; + } + /** * Record an issue. A repeat of the stored one (same status, code and * credential) writes nothing, so a burst of failing calls costs one read