diff --git a/packages/cli/src/connectors/mongodb/connector.ts b/packages/cli/src/connectors/mongodb/connector.ts index 12350835..d6d185c8 100644 --- a/packages/cli/src/connectors/mongodb/connector.ts +++ b/packages/cli/src/connectors/mongodb/connector.ts @@ -222,6 +222,15 @@ function unionDocumentKeys(documents: readonly KtxMongoDocument[]): string[] { return keys; } +// MongoDB aggregation write stages ($out/$merge) persist results; mongo_query is read-only. +function assertReadOnlyPipeline(pipeline: Record[]): void { + for (const stage of pipeline) { + if ('$out' in stage || '$merge' in stage) { + throw new Error('mongo_query pipelines must be read-only; $out and $merge stages are not allowed.'); + } + } +} + export class KtxMongoDbScanConnector implements KtxScanConnector { readonly id: string; readonly driver = 'mongodb' as const; @@ -398,6 +407,7 @@ export class KtxMongoDbScanConnector implements KtxScanConnector { async executeQuery(input: KtxMongoQueryInput, _ctx: KtxScanContext): Promise { this.assertConnection(input.connectionId); + assertReadOnlyPipeline(input.pipeline); const database = input.database ?? this.databases[0]!; const documents = await this.clientForQuery().aggregate(database, input.collection, input.pipeline, { limit: input.limit, diff --git a/packages/cli/src/context/mcp/local-project-ports.ts b/packages/cli/src/context/mcp/local-project-ports.ts index 71aa0456..c366a658 100644 --- a/packages/cli/src/context/mcp/local-project-ports.ts +++ b/packages/cli/src/context/mcp/local-project-ports.ts @@ -75,11 +75,10 @@ async function executeMongoQuery( `Connection "${connectionId}" driver "${connector.driver}" is not a MongoDB connection; mongo_query serves mongodb connections only.`, ); } - const result = await (connector as unknown as KtxMongoDbScanConnector).executeQuery( + return (connector as unknown as KtxMongoDbScanConnector).executeQuery( { connectionId, collection: input.collection, database: input.database, pipeline: input.pipeline, limit: input.limit }, { runId: 'mcp-mongo-query' }, ); - return { headers: result.headers, rows: result.rows, rowCount: result.rowCount }; } finally { await connector?.cleanup?.(); } diff --git a/packages/cli/src/context/mcp/types.ts b/packages/cli/src/context/mcp/types.ts index 36a86115..4ef12272 100644 --- a/packages/cli/src/context/mcp/types.ts +++ b/packages/cli/src/context/mcp/types.ts @@ -5,6 +5,7 @@ import type { KtxEntityDetailsInput, KtxEntityDetailsResponse } from '../scan/en import type { KtxDiscoverDataInput, KtxDiscoverDataResponse } from '../../context/search/discover.js'; import type { KtxDictionarySearchInput, KtxDictionarySearchResponse } from '../../context/sl/dictionary-search.js'; import type { SemanticLayerQueryInput } from '../../context/sl/types.js'; +import type { KtxMongoQueryResult } from '../../connectors/mongodb/connector.js'; import type { WikiSearchLaneSummary, WikiSearchMatchReason } from '../../context/wiki/types.js'; interface KtxMcpTextContent { @@ -180,11 +181,7 @@ export interface KtxSqlExecutionMcpPort { ): Promise; } -export interface KtxMongoQueryResponse { - headers: string[]; - rows: unknown[][]; - rowCount: number; -} +export type KtxMongoQueryResponse = KtxMongoQueryResult; /** @internal */ export interface KtxMongoQueryMcpPort { diff --git a/packages/cli/test/connectors/mongodb/connector.test.ts b/packages/cli/test/connectors/mongodb/connector.test.ts index 3fd25039..7089ee21 100644 --- a/packages/cli/test/connectors/mongodb/connector.test.ts +++ b/packages/cli/test/connectors/mongodb/connector.test.ts @@ -275,4 +275,24 @@ describe('KtxMongoDbScanConnector.executeQuery', () => { ), ).rejects.toThrow(/cannot serve connection other/); }); + + it('rejects a pipeline containing a $out write stage', async () => { + const { factory } = fakeClientFactory(); + await expect( + connector(baseConnection, factory).executeQuery( + { connectionId: 'mongo-prod', collection: 'users', pipeline: [{ $out: 'exfiltrated' }], limit: 10 }, + { runId: 't' }, + ), + ).rejects.toThrow(/must be read-only/); + }); + + it('rejects a pipeline containing a $merge write stage', async () => { + const { factory } = fakeClientFactory(); + await expect( + connector(baseConnection, factory).executeQuery( + { connectionId: 'mongo-prod', collection: 'users', pipeline: [{ $merge: { into: 'x' } }], limit: 10 }, + { runId: 't' }, + ), + ).rejects.toThrow(/must be read-only/); + }); });