diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9e3b1cd..00288fe 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -131,5 +131,5 @@ jobs: node-version: '22' cache: npm - run: npm ci - - name: Live MySQL concurrent upload and memory admission - run: node --test --test-concurrency=1 tests/uploadAdmission.mysql.integration.test.js tests/userMemoryAdmission.mysql.integration.test.js + - name: Live MySQL persistence concurrency acceptance + run: node --test --test-concurrency=1 tests/uploadAdmission.mysql.integration.test.js tests/userMemoryAdmission.mysql.integration.test.js tests/dataSourceTestAtomic.mysql.integration.test.js diff --git a/config/size-baseline.json b/config/size-baseline.json index a86b57d..450bfe4 100644 --- a/config/size-baseline.json +++ b/config/size-baseline.json @@ -64,7 +64,7 @@ "src/chartIntent.js": 122, "src/chatAttachments.js": 79, "src/chatStreamPresenter.js": 123, - "src/codeExecution.js": 697, + "src/codeExecution.js": 692, "src/columnSemantics.js": 95, "src/conditionBindings.js": 64, "src/config.js": 184, @@ -82,7 +82,7 @@ "src/datasourceCrypto.js": 55, "src/datasourceLifecycle.js": 15, "src/datasourcePool.js": 24, - "src/dataSourceRepository.js": 184, + "src/dataSourceRepository.js": 180, "src/derivedMetrics.js": 391, "src/dialects/common.js": 25, "src/dialects/index.js": 27, @@ -92,7 +92,7 @@ "src/distributedLock.js": 121, "src/executionRecovery.js": 61, "src/fastQueryPath.js": 116, - "src/feedback.js": 76, + "src/feedback.js": 45, "src/fileName.js": 96, "src/fileParsing.js": 320, "src/gateRegistry.js": 447, @@ -129,16 +129,16 @@ "src/localIndicatorValidation.js": 38, "src/mcp/audit.js": 103, "src/mcp/auth.js": 54, - "src/mcp/config.js": 73, + "src/mcp/config.js": 57, "src/mcp/connectorPackage.js": 115, - "src/mcp/index.js": 93, + "src/mcp/index.js": 56, "src/mcp/modeGate.js": 46, "src/mcp/rateLimit.js": 17, "src/mcp/requestContext.js": 22, "src/mcp/server.js": 30, - "src/mcp/tools/execute_query.js": 119, + "src/mcp/tools/execute_query.js": 93, "src/mcp/tools/get_policy.js": 33, - "src/mcp/tools/get_snapshot.js": 49, + "src/mcp/tools/get_snapshot.js": 33, "src/mcp/tools/list_metrics.js": 45, "src/mcp/tools/resolve_query.js": 115, "src/mcp/tools/shared.js": 55, @@ -160,7 +160,7 @@ "src/platformDatabase.js": 61, "src/platformLogRepository.js": 215, "src/presentationFallbacks.js": 63, - "src/processArtifacts.js": 173, + "src/processArtifacts.js": 89, "src/promptSanitizer.js": 123, "src/queryContract.js": 358, "src/queryContractCompiler.js": 2285, @@ -203,7 +203,7 @@ "src/turnRouter.js": 143, "src/uploads.js": 505, "src/uploadAdmission.js": 94, - "src/userMemory.js": 642, + "src/userMemory.js": 637, "src/userMemoryDistiller.js": 499, "src/userMemoryMaintenance.js": 156, "src/userMemoryModel.js": 138, @@ -212,11 +212,20 @@ "src/userRepository.js": 141, "src/valueClarification.js": 186, "src/workflow.js": 230, + "src/workspace.js": 464, "src/workspaceTransforms.js": 189, - "src/workspace.js": 566, "src/workspaceRepository.js": 287, "src/xlsx.js": 369, "src/xlsxReader.js": 480, + "src/skillAudits.js": 119, "src/platformAsyncLogRepository.js": 168, - "src/skillAudits.js": 119 + "src/dataSourceRow.js": 25, + "src/mcp/configValues.js": 34, + "src/feedbackSubmission.js": 69, + "src/processArtifactDraft.js": 119, + "src/mcp/tools/contractExecution.js": 39, + "src/mcp/tools/snapshotAccess.js": 41, + "src/mcp/configDefaults.js": 15, + "src/mcp/runtimePolicy.js": 46, + "src/codeGateAudit.js": 22 } diff --git a/src/codeExecution.js b/src/codeExecution.js index e3e6109..95c109e 100644 --- a/src/codeExecution.js +++ b/src/codeExecution.js @@ -3,7 +3,7 @@ import fs from 'node:fs'; import path from 'node:path'; import { spawn, spawnSync } from 'node:child_process'; import { DEFAULT_PYTHON_BIN } from './config.js'; -import { incrementCounter } from './metrics.js'; +import { recordCodeGateBlock } from './codeGateAudit.js'; const BLOCKED_IMPORTS = [ 'ctypes', @@ -595,20 +595,7 @@ export class CodeExecutionService { validateCode(code); } catch (error) { // CODE-001 is a hard, auditable gate: record the block before surfacing it. - incrementCounter('gate_block_total', { - gateId: 'CODE-001', - phase: 'EXECUTE', - blocking: 'true', - }); - this.database?.addAuditLog?.({ - userId, - action: 'CODE_GATE_BLOCK', - detail: { - sessionId: sessionId ?? null, - code: error.code ?? 'CODE_DATASOURCE_ACCESS_FORBIDDEN', - message: error.message, - }, - }); + await recordCodeGateBlock(this.database, { userId, sessionId, error }); throw error; } const runId = crypto.randomUUID(); diff --git a/src/codeGateAudit.js b/src/codeGateAudit.js new file mode 100644 index 0000000..b820532 --- /dev/null +++ b/src/codeGateAudit.js @@ -0,0 +1,21 @@ +import { incrementCounter } from './metrics.js'; +import { callPlatformDatabase } from './platformDatabaseAccess.js'; + +export async function recordCodeGateBlock(database, { userId, sessionId, error }) { + incrementCounter('gate_block_total', { + gateId: 'CODE-001', phase: 'EXECUTE', blocking: 'true', + }); + const entry = { + userId, + action: 'CODE_GATE_BLOCK', + detail: { + sessionId: sessionId ?? null, + code: error.code ?? 'CODE_DATASOURCE_ACCESS_FORBIDDEN', + message: error.message, + }, + }; + if (database?.isAsyncPlatformStore || typeof database?.addAuditLogAsync === 'function') { + return await callPlatformDatabase(database, 'addAuditLogAsync', 'addAuditLog', entry); + } + return database?.addAuditLog?.(entry); +} diff --git a/src/dataSourceRepository.js b/src/dataSourceRepository.js index e2967c8..520e17c 100644 --- a/src/dataSourceRepository.js +++ b/src/dataSourceRepository.js @@ -1,3 +1,4 @@ +import { mapRow } from './dataSourceRow.js'; import { normalizeDataSource } from './dialects/index.js'; import { dataSourceTestState } from './datasourceLifecycle.js'; @@ -5,40 +6,17 @@ function nowIso() { return new Date().toISOString(); } -function parseJson(value, fallback) { - if (value === null || value === undefined || value === '') { - return fallback; - } - try { - return JSON.parse(value); - } catch { - return fallback; - } -} - function asJson(value) { return JSON.stringify(value ?? {}); } -function asBool(value) { - return value === 1 || value === true; -} - export class DataSourceRepository { constructor(db, asyncStore = db) { this.db = db; this.asyncStore = asyncStore; } - mapRow(row) { - return row ? { - ...row, - options: parseJson(row.optionsJson, {}), - lastTestOk: row.lastTestOk === null || row.lastTestOk === undefined - ? null - : asBool(row.lastTestOk), - } : null; - } + mapRow(row) { return mapRow(row); } async listDataSourcesAsync() { const rows = await this.asyncStore.query(`SELECT id, code, name, db_type AS dbType, @@ -84,17 +62,15 @@ export class DataSourceRepository { async updateDataSourceTestAsync(id, { ok, message }, source = null) { const current = source ?? await this.getDataSourceAsync(id); if (!current) return null; - const unchanged = (await this.asyncStore.query(`SELECT COUNT(*) AS count FROM data_sources - WHERE id=? AND code=? AND db_type=? AND host=? AND port=? AND database_name=? - AND username=? AND encrypted_password=? AND options_json=?`, [ - Number(id), current.code, current.dbType, current.host, Number(current.port), - current.databaseName, current.username, current.encryptedPassword ?? '', - JSON.stringify(current.options ?? {}), - ]))[0]?.count; - if (Number(unchanged) > 0) { - await this.asyncStore.execute(`UPDATE data_sources SET last_test_at=?, last_test_ok=?, - last_test_message=? WHERE id=?`, [nowIso(), ok ? 1 : 0, String(message ?? ''), Number(id)]); - } + const fields = ['code', 'db_type', 'host', 'database_name', 'username', 'encrypted_password', 'options_json']; + const match = fields.map(field => this.asyncStore.backend === 'mysql' + ? `BINARY ${field} = BINARY ?` : `${field} = ?`).join(' AND '); + await this.asyncStore.execute(`UPDATE data_sources SET last_test_at=?, last_test_ok=?, + last_test_message=? WHERE id=? AND port=? AND ${match}`, [ + nowIso(), ok ? 1 : 0, String(message ?? ''), Number(id), Number(current.port), + current.code, current.dbType, current.host, current.databaseName, current.username, + current.encryptedPassword ?? '', JSON.stringify(current.options ?? {}), + ]); return this.getDataSourceAsync(id); } diff --git a/src/dataSourceRow.js b/src/dataSourceRow.js new file mode 100644 index 0000000..b8fd8b7 --- /dev/null +++ b/src/dataSourceRow.js @@ -0,0 +1,24 @@ +function parseJson(value, fallback) { + if (value === null || value === undefined || value === '') { + return fallback; + } + try { + return JSON.parse(value); + } catch { + return fallback; + } +} + +function asBool(value) { + return value === 1 || value === true; +} + +export function mapRow(row) { + return row ? { + ...row, + options: parseJson(row.optionsJson, {}), + lastTestOk: row.lastTestOk === null || row.lastTestOk === undefined + ? null + : asBool(row.lastTestOk), + } : null; + } diff --git a/src/feedback.js b/src/feedback.js index bd308a2..8f22314 100644 --- a/src/feedback.js +++ b/src/feedback.js @@ -1,77 +1,14 @@ +import { submitFeedback, submitFeedbackAsync } from './feedbackSubmission.js'; + export class FeedbackService { constructor(database, growth = null) { this.database = database; this.growth = growth; } - submit({ - userId, - themeId, - sessionId, - messageId, - question, - correct, - comment, - corrected, - }) { - const message = messageId - ? this.database.getChatMessage(messageId, userId) - : null; - if (messageId && !message) { - throw new Error('assistant message not found'); - } - const resolvedThemeId = themeId - ?? message?.result?.theme?.id - ?? this.database.getChatSession(sessionId, userId)?.themeId; - if (!userId || !resolvedThemeId) { - throw new Error('user and theme are required'); - } - const id = this.database.saveFeedback({ - userId, - themeId: resolvedThemeId, - sessionId: sessionId ?? message?.sessionId, - messageId, - question: question || message?.result?.question || '', - correct: Boolean(correct), - comment, - corrected, - }); - if (!correct) { - this.growth?.record({ - term: question || message?.result?.question || `message-${messageId}`, - kind: 'FEEDBACK', - source: 'USER_FEEDBACK', - context: { - userId, - themeId: resolvedThemeId, - sessionId: sessionId ?? message?.sessionId, - messageId, - comment, - corrected, - }, - }); - } - return { - id, - correct: Boolean(correct), - message: '反馈已记录', - }; - } + submit(args) { return submitFeedback(this, args); } - async submitAsync(args) { - if (!this.database.isAsyncPlatformStore) return this.submit(args); - const { userId, themeId, sessionId, messageId, question, correct, comment, corrected } = args; - const message = messageId ? await this.database.getChatMessageAsync(messageId, userId) : null; - if (messageId && !message) throw new Error('assistant message not found'); - const session = !themeId && !message ? await this.database.getChatSessionAsync(sessionId, userId) : null; - const resolvedThemeId = themeId ?? message?.result?.theme?.id ?? session?.themeId; - if (!userId || !resolvedThemeId) throw new Error('user and theme are required'); - const id = await this.database.saveFeedbackAsync({ userId, themeId: resolvedThemeId, sessionId: sessionId ?? message?.sessionId, - messageId, question: question || message?.result?.question || '', correct: Boolean(correct), comment, corrected }); - if (!correct) this.growth?.record({ term: question || message?.result?.question || `message-${messageId}`, kind: 'FEEDBACK', source: 'USER_FEEDBACK', - context: { userId, themeId: resolvedThemeId, sessionId: sessionId ?? message?.sessionId, messageId, comment, corrected } }); - return { id, correct: Boolean(correct), message: '反馈已记录' }; - } + submitAsync(args) { return submitFeedbackAsync(this, args); } list({ themeId = null, correct = null, limit = 100 } = {}) { return this.database.listFeedback({ themeId, correct, limit }); diff --git a/src/feedbackSubmission.js b/src/feedbackSubmission.js new file mode 100644 index 0000000..1a49361 --- /dev/null +++ b/src/feedbackSubmission.js @@ -0,0 +1,68 @@ +export function submitFeedback(service, { + userId, + themeId, + sessionId, + messageId, + question, + correct, + comment, + corrected, + }) { + const message = messageId + ? service.database.getChatMessage(messageId, userId) + : null; + if (messageId && !message) { + throw new Error('assistant message not found'); + } + const resolvedThemeId = themeId + ?? message?.result?.theme?.id + ?? service.database.getChatSession(sessionId, userId)?.themeId; + if (!userId || !resolvedThemeId) { + throw new Error('user and theme are required'); + } + const id = service.database.saveFeedback({ + userId, + themeId: resolvedThemeId, + sessionId: sessionId ?? message?.sessionId, + messageId, + question: question || message?.result?.question || '', + correct: Boolean(correct), + comment, + corrected, + }); + if (!correct) { + service.growth?.record({ + term: question || message?.result?.question || `message-${messageId}`, + kind: 'FEEDBACK', + source: 'USER_FEEDBACK', + context: { + userId, + themeId: resolvedThemeId, + sessionId: sessionId ?? message?.sessionId, + messageId, + comment, + corrected, + }, + }); + } + return { + id, + correct: Boolean(correct), + message: '反馈已记录', + }; + } + +export async function submitFeedbackAsync(service, args) { + if (!service.database.isAsyncPlatformStore) return submitFeedback(service, args); + const { userId, themeId, sessionId, messageId, question, correct, comment, corrected } = args; + const message = messageId ? await service.database.getChatMessageAsync(messageId, userId) : null; + if (messageId && !message) throw new Error('assistant message not found'); + const session = !themeId && !message ? await service.database.getChatSessionAsync(sessionId, userId) : null; + const resolvedThemeId = themeId ?? message?.result?.theme?.id ?? session?.themeId; + if (!userId || !resolvedThemeId) throw new Error('user and theme are required'); + const id = await service.database.saveFeedbackAsync({ userId, themeId: resolvedThemeId, sessionId: sessionId ?? message?.sessionId, + messageId, question: question || message?.result?.question || '', correct: Boolean(correct), comment, corrected }); + if (!correct) await service.growth?.record({ term: question || message?.result?.question || `message-${messageId}`, kind: 'FEEDBACK', source: 'USER_FEEDBACK', + context: { userId, themeId: resolvedThemeId, sessionId: sessionId ?? message?.sessionId, messageId, comment, corrected } }); + return { id, correct: Boolean(correct), message: '反馈已记录' }; + } diff --git a/src/mcp/config.js b/src/mcp/config.js index b498546..1627cee 100644 --- a/src/mcp/config.js +++ b/src/mcp/config.js @@ -2,88 +2,55 @@ // 默认拒绝原则：配置无效时回退到只读模式（disabled 关闭 / read 最小可见）。 import { callPlatformDatabase } from '../platformDatabaseAccess.js'; -export const DEFAULT_MCP_CONFIG = { - enabled: false, - mode: 'read', - toolOverrides: {}, - tenantScope: [], - rateLimit: { rpm: 60, burst: 10 }, - audit: { enabled: true, level: 'full' }, - sessionMode: 'stateful', -}; +import { coerce } from './configValues.js'; +export { DEFAULT_MCP_CONFIG, getDefaultMcpConfig } from './configValues.js'; -const VALID_MODES = ['disabled', 'read', 'standard', 'full']; const CACHE_TTL = 5000; - -let cache = { value: null, at: 0 }; - -export function getDefaultMcpConfig() { - return JSON.parse(JSON.stringify(DEFAULT_MCP_CONFIG)); -} - -function coerce(raw) { - const cfg = getDefaultMcpConfig(); - if (!raw || typeof raw !== 'object') { - return cfg; - } - if (typeof raw.enabled === 'boolean') cfg.enabled = raw.enabled; - if (VALID_MODES.includes(raw.mode)) cfg.mode = raw.mode; - if (raw.toolOverrides && typeof raw.toolOverrides === 'object') { - cfg.toolOverrides = { ...raw.toolOverrides }; - } - if (Array.isArray(raw.tenantScope)) cfg.tenantScope = raw.tenantScope.map(String); - if (raw.rateLimit && typeof raw.rateLimit === 'object') { - const rpm = Number(raw.rateLimit.rpm); - const burst = Number(raw.rateLimit.burst); - if (Number.isFinite(rpm) && rpm > 0) cfg.rateLimit.rpm = rpm; - if (Number.isFinite(burst) && burst > 0) cfg.rateLimit.burst = burst; - } - if (raw.audit && typeof raw.audit === 'object') { - if (typeof raw.audit.enabled === 'boolean') cfg.audit.enabled = raw.audit.enabled; - if (['full', 'summary', 'off'].includes(raw.audit.level)) cfg.audit.level = raw.audit.level; - } - if (raw.sessionMode === 'stateless' || raw.sessionMode === 'stateful') { - cfg.sessionMode = raw.sessionMode; - } - return cfg; +let cache = new WeakMap(); +function entryFor(database) { + if (!cache.has(database)) cache.set(database, { value: null, at: 0 }); + return cache.get(database); } export function getMcpConfig(database) { + const entry = entryFor(database); const now = Date.now(); - if (cache.value && now - cache.at < CACHE_TTL) { - return cache.value; + if (entry.value && now - entry.at < CACHE_TTL) { + return entry.value; } const record = database.getPlatformSetting('mcp', null); const raw = record && record.value && typeof record.value === 'object' ? record.value : null; const cfg = coerce(raw); - cache = { value: cfg, at: now }; + cache.set(database, { value: cfg, at: now }); return cfg; } export function setMcpConfig(database, patch) { const next = coerce(patch && typeof patch === 'object' ? patch : {}); database.savePlatformSetting('mcp', next); - cache = { value: next, at: Date.now() }; + cache.set(database, { value: next, at: Date.now() }); return next; } export async function getMcpConfigAsync(database) { + const entry = entryFor(database); const now = Date.now(); - if (cache.value && now - cache.at < CACHE_TTL) return cache.value; + if (entry.value && now - entry.at < CACHE_TTL) return entry.value; const record = await callPlatformDatabase(database, 'getPlatformSettingAsync', 'getPlatformSetting', 'mcp', null); const cfg = coerce(record?.value); - cache = { value: cfg, at: now }; + if (cache.get(database) !== entry) return cache.get(database)?.value ?? cfg; + cache.set(database, { value: cfg, at: now }); return cfg; } export async function setMcpConfigAsync(database, patch) { const next = coerce(patch && typeof patch === 'object' ? patch : {}); await callPlatformDatabase(database, 'savePlatformSettingAsync', 'savePlatformSetting', 'mcp', next); - cache = { value: next, at: Date.now() }; + cache.set(database, { value: next, at: Date.now() }); return next; } // 仅用于测试或进程内重置，避免缓存陈旧。 export function _resetMcpConfigCache() { - cache = { value: null, at: 0 }; + cache = new WeakMap(); } diff --git a/src/mcp/configDefaults.js b/src/mcp/configDefaults.js new file mode 100644 index 0000000..62d2fa5 --- /dev/null +++ b/src/mcp/configDefaults.js @@ -0,0 +1,14 @@ +export const DEFAULT_MCP_CONFIG = { + enabled: false, + mode: 'read', + toolOverrides: {}, + tenantScope: [], + rateLimit: { rpm: 60, burst: 10 }, + audit: { enabled: true, level: 'full' }, + sessionMode: 'stateful', + allowedOrigins: [], +}; + +export function getDefaultMcpConfig() { + return JSON.parse(JSON.stringify(DEFAULT_MCP_CONFIG)); +} diff --git a/src/mcp/configValues.js b/src/mcp/configValues.js new file mode 100644 index 0000000..9f993da --- /dev/null +++ b/src/mcp/configValues.js @@ -0,0 +1,33 @@ +import { getDefaultMcpConfig } from './configDefaults.js'; +export { DEFAULT_MCP_CONFIG, getDefaultMcpConfig } from './configDefaults.js'; + +const VALID_MODES = ['disabled', 'read', 'standard', 'full']; + + +export function coerce(raw) { + const cfg = getDefaultMcpConfig(); + if (!raw || typeof raw !== 'object') { + return cfg; + } + if (typeof raw.enabled === 'boolean') cfg.enabled = raw.enabled; + if (VALID_MODES.includes(raw.mode)) cfg.mode = raw.mode; + if (raw.toolOverrides && typeof raw.toolOverrides === 'object') { + cfg.toolOverrides = { ...raw.toolOverrides }; + } + if (Array.isArray(raw.tenantScope)) cfg.tenantScope = raw.tenantScope.map(String); + if (Array.isArray(raw.allowedOrigins)) cfg.allowedOrigins = raw.allowedOrigins.map(String); + if (raw.rateLimit && typeof raw.rateLimit === 'object') { + const rpm = Number(raw.rateLimit.rpm); + const burst = Number(raw.rateLimit.burst); + if (Number.isFinite(rpm) && rpm > 0) cfg.rateLimit.rpm = rpm; + if (Number.isFinite(burst) && burst > 0) cfg.rateLimit.burst = burst; + } + if (raw.audit && typeof raw.audit === 'object') { + if (typeof raw.audit.enabled === 'boolean') cfg.audit.enabled = raw.audit.enabled; + if (['full', 'summary', 'off'].includes(raw.audit.level)) cfg.audit.level = raw.audit.level; + } + if (raw.sessionMode === 'stateless' || raw.sessionMode === 'stateful') { + cfg.sessionMode = raw.sessionMode; + } + return cfg; +} diff --git a/src/mcp/index.js b/src/mcp/index.js index 2271cf8..a5d19e2 100644 --- a/src/mcp/index.js +++ b/src/mcp/index.js @@ -1,13 +1,11 @@ +import { createMcpRuntime } from './runtimePolicy.js'; // MCP 模块入口：对外暴露 { handle, isEnabled, close }。 // 负责 API Key 鉴权、Origin 校验、速率限制、请求级上下文注入、原始 body 读取， // 并把请求委派给 transport 层。路由处理器只需 await mcp.handle(ctx)。 import { randomUUID } from 'node:crypto'; -import { createMcpSessionManager } from './transport.js'; -import { createMcpServer } from './server.js'; import { getMcpConfig, getMcpConfigAsync } from './config.js'; -import { createMcpRateLimiter } from './rateLimit.js'; -import { validateApiKey, validateApiKeyAsync, isOriginAllowed } from './auth.js'; -import { MCP_ERROR, auditToolCall } from './audit.js'; +import { validateApiKeyAsync, isOriginAllowed } from './auth.js'; +import { MCP_ERROR } from './audit.js'; function jsonRpcError(res, status, code, message) { res.writeHead(status, { 'content-type': 'application/json' }); @@ -15,43 +13,8 @@ function jsonRpcError(res, status, code, message) { } export function createMcpModule(application) { - const { database, agent, workspace, config } = application; - let rateLimiter = null; - let manager = null; - let managerMode = null; - - function ensureRateLimiter(cfg) { - if (!rateLimiter) { - rateLimiter = createMcpRateLimiter({ - rpm: cfg.rateLimit?.rpm ?? 60, - burst: cfg.rateLimit?.burst ?? 10, - }); - } - return rateLimiter; - } - - function ensureManager(cfg) { - if (!manager || managerMode !== cfg.sessionMode) { - rateLimiter = ensureRateLimiter(cfg); - manager = createMcpSessionManager({ - sessionMode: cfg.sessionMode, - buildServer: (deps) => createMcpServer({ config: cfg, deps }), - buildDeps: () => ({ - agent, - workspace, - database, - config, - rateLimiter, - auditToolCall, - requestContext: { - tenantId: 'default', keyId: null, scopes: ['mcp'], tenantScope: [], requestId: null, - }, - }), - }); - managerMode = cfg.sessionMode; - } - return manager; - } + const { database } = application; + const runtime = createMcpRuntime(application); async function handle(ctx) { const cfg = await getMcpConfigAsync(database); @@ -67,7 +30,7 @@ export function createMcpModule(application) { if (!isOriginAllowed(ctx.request.headers.origin, cfg.allowedOrigins)) { return jsonRpcError(ctx.response, 403, MCP_ERROR.FORBIDDEN, 'Origin 不被允许'); } - const limiter = ensureRateLimiter(cfg); + const limiter = runtime.ensureRateLimiter(cfg); if (!limiter.check(keyInfo.keyId)) { return jsonRpcError(ctx.response, 429, MCP_ERROR.RATE_LIMITED, '请求过于频繁，请稍后重试'); } @@ -78,7 +41,7 @@ export function createMcpModule(application) { tenantScope: cfg.tenantScope ?? [], requestId: randomUUID(), }; - const m = ensureManager(cfg); + const m = runtime.ensureManager(cfg); // 不预读 body：交由 SDK 的 StreamableHTTPServerTransport 自行读取请求流， // 避免流被消费后 SDK 解析到空 body（Parse error）。 await m.handle(ctx.request, ctx.response, requestContext); @@ -87,6 +50,6 @@ export function createMcpModule(application) { return { handle, isEnabled: () => getMcpConfig(database).enabled, - close: () => manager?.closeAll(), + close: () => runtime.close(), }; } diff --git a/src/mcp/runtimePolicy.js b/src/mcp/runtimePolicy.js new file mode 100644 index 0000000..65c939e --- /dev/null +++ b/src/mcp/runtimePolicy.js @@ -0,0 +1,45 @@ +import { createMcpSessionManager } from './transport.js'; +import { createMcpServer } from './server.js'; +import { createMcpRateLimiter } from './rateLimit.js'; +import { auditToolCall } from './audit.js'; + +export function createMcpRuntime(application) { + const { database, agent, workspace, config } = application; + let rateLimiter = null; + let limiterPolicy = null; + let manager = null; + let managerPolicy = null; + + function ensureRateLimiter(cfg) { + const rpm = cfg.rateLimit?.rpm ?? 60; + const burst = cfg.rateLimit?.burst ?? 10; + const policy = JSON.stringify([rpm, burst]); + if (!rateLimiter || limiterPolicy !== policy) { + rateLimiter = createMcpRateLimiter({ rpm, burst }); + limiterPolicy = policy; + } + return rateLimiter; + } + + function ensureManager(cfg) { + const overrides = Object.entries(cfg.toolOverrides ?? {}).sort(([left], [right]) => left.localeCompare(right)); + const policy = JSON.stringify([cfg.sessionMode, cfg.mode, overrides]); + if (!manager || managerPolicy !== policy) { + manager?.closeAll(); + ensureRateLimiter(cfg); + manager = createMcpSessionManager({ + sessionMode: cfg.sessionMode, + buildServer: deps => createMcpServer({ config: cfg, deps }), + buildDeps: () => ({ agent, workspace, database, config, rateLimiter, auditToolCall, + requestContext: { + tenantId: 'default', keyId: null, scopes: ['mcp'], tenantScope: [], requestId: null, + }, + }), + }); + managerPolicy = policy; + } + return manager; + } + + return { ensureRateLimiter, ensureManager, close: () => manager?.closeAll() }; +} diff --git a/src/mcp/tools/contractExecution.js b/src/mcp/tools/contractExecution.js new file mode 100644 index 0000000..f328b0c --- /dev/null +++ b/src/mcp/tools/contractExecution.js @@ -0,0 +1,38 @@ +export function buildDatasetRequest(contract, fields = []) { + const fieldByName = new Map(fields.map((f) => [String(f.fieldName), f])); + const fieldByBiz = new Map(fields.map((f) => [String(f.bizName ?? f.name), f])); + const resolveField = (name) => { + const direct = fieldByName.get(String(name)) ?? fieldByBiz.get(String(name)); + return direct?.fieldName ?? name; + }; + const window = (contract.timeWindows && contract.timeWindows[0]) || contract.dateInfo || null; + const timeField = fields.find((f) => f.role === 'TIME'); + const dateRange = (timeField && window) + ? { field: timeField.fieldName, startDate: window.startDate ?? null, endDate: window.endDate ?? null } + : undefined; + return { + metrics: (contract.metrics ?? []).map((m) => ({ field: resolveField(m) })), + dimensions: (contract.dimensions ?? []).map((d) => resolveField(d)), + filters: (contract.filters ?? []).map((f) => ({ + field: resolveField(f.bizName ?? f.field), + operator: f.operator, + value: f.value, + scope: f.scope ?? 'ROW', + })), + dateRange, + order: contract.order ?? [], + limit: contract.limit ?? 200, + timeGrain: contract.timeGrain ?? null, + analysisPipeline: Boolean(contract.analysisPipeline), + }; +} + +export function parseContract(record) { + if (!record) return null; + if (record.contractJson && typeof record.contractJson === 'object') return record.contractJson; + try { + return JSON.parse(record.contractJson); + } catch { + return null; + } +} diff --git a/src/mcp/tools/execute_query.js b/src/mcp/tools/execute_query.js index f8f0b51..1a736ee 100644 --- a/src/mcp/tools/execute_query.js +++ b/src/mcp/tools/execute_query.js @@ -1,49 +1,11 @@ // 工具 execute_query：凭 contractId 执行冻结的查询契约，返回列定义与结果行。 // 结果确定性由契约保证，不触发 LLM 推理；执行前复检一次门禁。 import { z } from 'zod'; +import { buildDatasetRequest, parseContract } from './contractExecution.js'; import { - wrapToolHandler, buildMeta, MCP_ERROR, fail, normalizeResult, + wrapToolHandler, buildMeta, assertTenantScope, MCP_ERROR, fail, normalizeResult, } from './shared.js'; -function buildDatasetRequest(contract, fields = []) { - const fieldByName = new Map(fields.map((f) => [String(f.fieldName), f])); - const fieldByBiz = new Map(fields.map((f) => [String(f.bizName ?? f.name), f])); - const resolveField = (name) => { - const direct = fieldByName.get(String(name)) ?? fieldByBiz.get(String(name)); - return direct?.fieldName ?? name; - }; - const window = (contract.timeWindows && contract.timeWindows[0]) || contract.dateInfo || null; - const timeField = fields.find((f) => f.role === 'TIME'); - const dateRange = (timeField && window) - ? { field: timeField.fieldName, startDate: window.startDate ?? null, endDate: window.endDate ?? null } - : undefined; - return { - metrics: (contract.metrics ?? []).map((m) => ({ field: resolveField(m) })), - dimensions: (contract.dimensions ?? []).map((d) => resolveField(d)), - filters: (contract.filters ?? []).map((f) => ({ - field: resolveField(f.bizName ?? f.field), - operator: f.operator, - value: f.value, - scope: f.scope ?? 'ROW', - })), - dateRange, - order: contract.order ?? [], - limit: contract.limit ?? 200, - timeGrain: contract.timeGrain ?? null, - analysisPipeline: Boolean(contract.analysisPipeline), - }; -} - -function parseContract(record) { - if (!record) return null; - if (record.contractJson && typeof record.contractJson === 'object') return record.contractJson; - try { - return JSON.parse(record.contractJson); - } catch { - return null; - } -} - export function registerExecuteQuery(server, ctx) { server.registerTool('execute_query', { title: '执行查询契约', @@ -59,6 +21,10 @@ export function registerExecuteQuery(server, ctx) { ? await database.getMcpQueryContractAsync(args.contractId) : database.getMcpQueryContract(args.contractId); if (!record) throw fail(MCP_ERROR.CONTRACT_NOT_FOUND, '查询契约不存在，请重新执行 resolve_query'); + if (String(record.tenantId ?? 'default') !== String(requestContext?.tenantId ?? 'default')) { + throw fail(MCP_ERROR.FORBIDDEN, '无权限执行此查询契约'); + } + assertTenantScope(ctx, record.themeId); if (record.expiresAt && new Date(record.expiresAt).getTime() < Date.now()) { throw fail(MCP_ERROR.CONTRACT_EXPIRED, '查询契约已过期，请重新编译契约'); } diff --git a/src/mcp/tools/get_snapshot.js b/src/mcp/tools/get_snapshot.js index 8ee5dae..0293c5c 100644 --- a/src/mcp/tools/get_snapshot.js +++ b/src/mcp/tools/get_snapshot.js @@ -1,28 +1,10 @@ // 工具 get_snapshot：读取工作区已落库的分析产物（只读）。 import { z } from 'zod'; +import { buildAuthorizedSnapshot } from './snapshotAccess.js'; import { - wrapToolHandler, buildMeta, MCP_ERROR, fail, normalizeResult, + wrapToolHandler, buildMeta, MCP_ERROR, fail, } from './shared.js'; -function trimArtifact(artifact) { - const out = { - id: artifact.id, - kind: artifact.kind ?? artifact.type, - title: artifact.title ?? artifact.subject ?? null, - createdAt: artifact.createdAt ?? null, - }; - const payload = artifact.payload; - if (payload && typeof payload === 'object') { - if (Array.isArray(payload.columns)) out.columns = payload.columns.slice(0, 50); - if (Array.isArray(payload.rows)) { - out.rowCount = payload.rows.length; - out.previewRows = payload.rows.slice(0, 20); - } - if (typeof payload.text === 'string') out.textExcerpt = payload.text.slice(0, 1000); - } - return out; -} - export function registerGetSnapshot(server, ctx) { server.registerTool('get_snapshot', { title: '获取工作区产物', @@ -30,7 +12,7 @@ export function registerGetSnapshot(server, ctx) { + '元数据与预览。只读。', inputSchema: { artifactId: z.string().describe('工作区产物 ID'), - userId: z.number().int().describe('用户 ID（用于鉴权隔离）'), + userId: z.number().int().positive().describe('用户 ID（用于鉴权隔离）'), }, }, wrapToolHandler(async (args) => { const { workspace, database, requestContext, auditToolCall } = ctx; @@ -44,7 +26,7 @@ export function registerGetSnapshot(server, ctx) { throw fail(MCP_ERROR.CONTRACT_NOT_FOUND, '产物不存在或无权限访问'); } if (!artifact) throw fail(MCP_ERROR.CONTRACT_NOT_FOUND, '产物不存在'); - return { content: [{ type: 'text', text: JSON.stringify(trimArtifact(artifact), null, 2) }] }; + return { content: [{ type: 'text', text: JSON.stringify(await buildAuthorizedSnapshot(ctx, artifact, args.userId), null, 2) }] }; }); })); } diff --git a/src/mcp/tools/snapshotAccess.js b/src/mcp/tools/snapshotAccess.js new file mode 100644 index 0000000..5c658ec --- /dev/null +++ b/src/mcp/tools/snapshotAccess.js @@ -0,0 +1,40 @@ +import { callPlatformDatabase } from '../../platformDatabaseAccess.js'; +import { assertTenantScope, MCP_ERROR, fail } from './shared.js'; + +export async function buildAuthorizedSnapshot(ctx, artifact, userId) { + if (!Number.isSafeInteger(userId) || userId <= 0) { + throw fail(MCP_ERROR.FORBIDDEN, '无权限访问此产物'); + } + const workspace = await callPlatformDatabase(ctx.database, 'getWorkspaceAsync', 'getWorkspace', artifact.workspaceId, userId); + if (!workspace || workspace.themeId == null) throw fail(MCP_ERROR.FORBIDDEN, '无权限访问此产物'); + assertTenantScope(ctx, workspace.themeId); + let resolved; + try { + resolved = ctx.agent.resolveContextAsync + ? await ctx.agent.resolveContextAsync(userId, workspace.themeId) + : ctx.agent.resolveContext(userId, workspace.themeId); + } catch { + throw fail(MCP_ERROR.FORBIDDEN, '无权限访问此产物'); + } + if (!resolved?.scope || resolved.scope.allowed === false) throw fail(MCP_ERROR.FORBIDDEN, '无权限访问此产物'); + return trimArtifact(artifact); +} + +function trimArtifact(artifact) { + const out = { + id: artifact.id, + kind: artifact.kind ?? artifact.type, + title: artifact.title ?? artifact.subject ?? null, + createdAt: artifact.createdAt ?? null, + }; + const payload = artifact.payload; + if (payload && typeof payload === 'object') { + if (Array.isArray(payload.columns)) out.columns = payload.columns.slice(0, 50); + if (Array.isArray(payload.rows)) { + out.rowCount = payload.rows.length; + out.previewRows = payload.rows.slice(0, 20); + } + if (typeof payload.text === 'string') out.textExcerpt = payload.text.slice(0, 1000); + } + return out; +} diff --git a/src/processArtifactDraft.js b/src/processArtifactDraft.js new file mode 100644 index 0000000..8293890 --- /dev/null +++ b/src/processArtifactDraft.js @@ -0,0 +1,118 @@ +// 工作区过程文件：把 agent 每个产出数据的环节落成用户可直接读懂的数据文件。 +// 只保留数据行和人话元信息，不保存环节参数、原始响应等技术载荷，也不承载业务口径。 + +const ROW_LIMIT = 200; +const MAX_STEPS_PER_RUN = 60; + +const KIND_LABELS = { + TABLE: '数据表', + SQL_RESULT: '查询结果', +}; + +function isRecord(value) { + return Boolean(value) && typeof value === 'object' && !Array.isArray(value); +} + +// 工具返回的数据形态可能是完整结果、抽样结果或嵌套在 data 下，这里统一识别。 +function extractTable(result) { + if (!isRecord(result)) { + return null; + } + const columns = Array.isArray(result.columns) ? result.columns : null; + const rows = Array.isArray(result.rows) + ? result.rows + : Array.isArray(result.data?.rows) ? result.data.rows : null; + const sampleRows = Array.isArray(result.sampleRows) ? result.sampleRows : null; + const source = rows ?? sampleRows; + if (!columns || !source || columns.length === 0 || source.length === 0) { + return null; + } + return { + columns, + rows: source.slice(0, ROW_LIMIT), + truncated: source.length > ROW_LIMIT, + sampled: !rows, + }; +} + +function collectRelatedArtifactIds(result) { + if (!isRecord(result)) { + return []; + } + const ids = [ + result.workspaceArtifactId, + result.codeArtifactId, + ...(Array.isArray(result.artifactIds) ? result.artifactIds : []), + ...(Array.isArray(result.inputArtifactIds) ? result.inputArtifactIds : []), + ...(Array.isArray(result.outputs) ? result.outputs.map((item) => item?.artifactId) : []), + ]; + return [...new Set(ids.filter(Boolean).map(String))]; +} + +// 把一次工具调用整理成过程文件草稿。 +// 只有产出可读数据表的环节才落盘；目录、契约、原始 JSON 等纯技术载荷直接跳过。 +function buildProcessArtifact({ + tool, + result = null, + error = null, + step = {}, + previousArtifactId = null, +}) { + const table = error ? null : extractTable(result); + if (!table) { + return null; + } + const sequence = Number(step.sequence) > 0 ? Number(step.sequence) : null; + const label = String(step.title ?? tool ?? '执行步骤'); + const dataKind = result?.sql ? 'SQL_RESULT' : 'TABLE'; + const relatedArtifactIds = collectRelatedArtifactIds(result); + const inputArtifactIds = [ + ...new Set([...(previousArtifactId ? [previousArtifactId] : []), ...relatedArtifactIds]), + ]; + const rowCount = Number(result?.rowCount ?? table.rows.length); + const durationMs = Number(step.durationMs); + const metadata = { + process: true, + dataKind, + dataKindLabel: KIND_LABELS[dataKind], + stepId: step.id ?? null, + sequence, + tool: tool ?? null, + toolTitle: label, + stageCode: step.stageCode ?? null, + skillCode: step.skillCode ?? null, + skillName: step.skillName ?? null, + status: error ? 'error' : 'success', + startedAt: step.startedAt ?? null, + durationMs: Number.isFinite(durationMs) ? durationMs : null, + summary: String(step.detail ?? ''), + rowCount, + columnCount: table.columns.length, + capturedRowCount: table.rows.length, + sampled: Boolean(table.sampled), + truncated: Boolean(table.truncated), + inputArtifactIds, + parentArtifactIds: inputArtifactIds, + outputArtifactIds: relatedArtifactIds, + }; + const payload = { + step: { + id: step.id ?? null, + sequence, + tool: tool ?? null, + title: label, + stageCode: step.stageCode ?? null, + status: metadata.status, + durationMs: metadata.durationMs, + }, + dataKind, + summary: metadata.summary, + truncated: { rows: Boolean(table.truncated) }, + data: { columns: table.columns, rows: table.rows }, + }; + const rowLabel = ` · ${rowCount} 行`; + const title = `${sequence ? `步骤 ${String(sequence).padStart(2, '0')} · ` : ''}${label}${rowLabel}`; + return { title, metadata, payload }; +} + +export { buildProcessArtifact, KIND_LABELS, MAX_STEPS_PER_RUN }; diff --git a/src/processArtifacts.js b/src/processArtifacts.js index f920d38..c625d66 100644 --- a/src/processArtifacts.js +++ b/src/processArtifacts.js @@ -1,119 +1,4 @@ -// 工作区过程文件：把 agent 每个产出数据的环节落成用户可直接读懂的数据文件。 -// 只保留数据行和人话元信息，不保存环节参数、原始响应等技术载荷，也不承载业务口径。 - -const ROW_LIMIT = 200; -const MAX_STEPS_PER_RUN = 60; - -const KIND_LABELS = { - TABLE: '数据表', - SQL_RESULT: '查询结果', -}; - -function isRecord(value) { - return Boolean(value) && typeof value === 'object' && !Array.isArray(value); -} - -// 工具返回的数据形态可能是完整结果、抽样结果或嵌套在 data 下，这里统一识别。 -function extractTable(result) { - if (!isRecord(result)) { - return null; - } - const columns = Array.isArray(result.columns) ? result.columns : null; - const rows = Array.isArray(result.rows) - ? result.rows - : Array.isArray(result.data?.rows) ? result.data.rows : null; - const sampleRows = Array.isArray(result.sampleRows) ? result.sampleRows : null; - const source = rows ?? sampleRows; - if (!columns || !source || columns.length === 0 || source.length === 0) { - return null; - } - return { - columns, - rows: source.slice(0, ROW_LIMIT), - truncated: source.length > ROW_LIMIT, - sampled: !rows, - }; -} - -function collectRelatedArtifactIds(result) { - if (!isRecord(result)) { - return []; - } - const ids = [ - result.workspaceArtifactId, - result.codeArtifactId, - ...(Array.isArray(result.artifactIds) ? result.artifactIds : []), - ...(Array.isArray(result.inputArtifactIds) ? result.inputArtifactIds : []), - ...(Array.isArray(result.outputs) ? result.outputs.map((item) => item?.artifactId) : []), - ]; - return [...new Set(ids.filter(Boolean).map(String))]; -} - -// 把一次工具调用整理成过程文件草稿。 -// 只有产出可读数据表的环节才落盘；目录、契约、原始 JSON 等纯技术载荷直接跳过。 -function buildProcessArtifact({ - tool, - result = null, - error = null, - step = {}, - previousArtifactId = null, -}) { - const table = error ? null : extractTable(result); - if (!table) { - return null; - } - const sequence = Number(step.sequence) > 0 ? Number(step.sequence) : null; - const label = String(step.title ?? tool ?? '执行步骤'); - const dataKind = result?.sql ? 'SQL_RESULT' : 'TABLE'; - const relatedArtifactIds = collectRelatedArtifactIds(result); - const inputArtifactIds = [ - ...new Set([...(previousArtifactId ? [previousArtifactId] : []), ...relatedArtifactIds]), - ]; - const rowCount = Number(result?.rowCount ?? table.rows.length); - const durationMs = Number(step.durationMs); - const metadata = { - process: true, - dataKind, - dataKindLabel: KIND_LABELS[dataKind], - stepId: step.id ?? null, - sequence, - tool: tool ?? null, - toolTitle: label, - stageCode: step.stageCode ?? null, - skillCode: step.skillCode ?? null, - skillName: step.skillName ?? null, - status: error ? 'error' : 'success', - startedAt: step.startedAt ?? null, - durationMs: Number.isFinite(durationMs) ? durationMs : null, - summary: String(step.detail ?? ''), - rowCount, - columnCount: table.columns.length, - capturedRowCount: table.rows.length, - sampled: Boolean(table.sampled), - truncated: Boolean(table.truncated), - inputArtifactIds, - parentArtifactIds: inputArtifactIds, - outputArtifactIds: relatedArtifactIds, - }; - const payload = { - step: { - id: step.id ?? null, - sequence, - tool: tool ?? null, - title: label, - stageCode: step.stageCode ?? null, - status: metadata.status, - durationMs: metadata.durationMs, - }, - dataKind, - summary: metadata.summary, - truncated: { rows: Boolean(table.truncated) }, - data: { columns: table.columns, rows: table.rows }, - }; - const rowLabel = ` · ${rowCount} 行`; - const title = `${sequence ? `步骤 ${String(sequence).padStart(2, '0')} · ` : ''}${label}${rowLabel}`; - return { title, metadata, payload }; -} +import { buildProcessArtifact, KIND_LABELS, MAX_STEPS_PER_RUN } from './processArtifactDraft.js'; // 记录器：agent 每完成一个数据环节就调用一次；没有可读数据的环节直接跳过， // 写入失败也只跳过，不影响主流程。 @@ -168,7 +53,7 @@ function createProcessArtifactRecorder({ function createProcessArtifactRecorderAsync(options) { const { service, workspaceId, userId, sessionId, emit = () => {}, artifactIds = null, maxSteps = MAX_STEPS_PER_RUN } = options; const state = { previousArtifactId: null, count: 0 }; - return async ({ tool, result = null, error = null, step = {} }) => { + const record = async ({ tool, result = null, error = null, step = {} }) => { if (!service || !workspaceId || state.count >= maxSteps) return null; try { const draft = buildProcessArtifact({ tool, result, error, step, previousArtifactId: state.previousArtifactId }); @@ -186,6 +71,12 @@ function createProcessArtifactRecorderAsync(options) { return null; } }; + let pending = Promise.resolve(); + return input => { + const result = pending.then(() => record(input)); + pending = result.catch(() => {}); + return result; + }; } export { diff --git a/tests/codeGateAudit.test.js b/tests/codeGateAudit.test.js new file mode 100644 index 0000000..45a9f57 --- /dev/null +++ b/tests/codeGateAudit.test.js @@ -0,0 +1,62 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { CodeExecutionService } from '../src/codeExecution.js'; +import { getCounter } from '../src/metrics.js'; + +for (const fails of [false, true]) { + test(`code gate awaits async audit ${fails ? 'failure' : 'persistence'} before rejecting`, async t => { + const database = new PlatformDatabase(':memory:'); + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'code-gate-audit-')); + t.after(async () => { await database.closeAsync(); fs.rmSync(directory, { recursive: true, force: true }); }); + const userId = database.getUserByUsername('admin').id; + let release, started = false; + const gate = new Promise(resolve => { release = resolve; }); + const failure = new Error('audit driver refused'); + const adapter = { isAsyncPlatformStore: true, + async addAuditLogAsync(entry) { + started = true; await gate; + if (fails) throw failure; + return database.addAuditLog(entry); + }, + addAuditLog() { throw new Error('sync fallback forbidden'); }, + }; + const labels = { gateId: 'CODE-001', phase: 'EXECUTE', blocking: 'true' }; + const count = getCounter('gate_block_total', labels); + const service = new CodeExecutionService({ database: adapter, workspace: {}, + config: { codeExecutionRoot: path.join(directory, 'runs'), pythonBin: '__must_not_start__' } }); + let completed = false; + const outcome = service.run({ userId, sessionId: 7, workspaceId: 1, code: 'import sqlite3' }) + .then(value => { completed = true; return { value }; }, error => { completed = true; return { error }; }); + await new Promise(setImmediate); + try { assert.equal(started, true); assert.equal(completed, false); } + finally { release(); await outcome; } + const { error } = await outcome; + assert.equal(getCounter('gate_block_total', labels), count + 1); + assert.equal(fs.existsSync(service.root), false); + if (fails) { + assert.equal(error, failure); + assert.equal(database.listAuditLogs().length, 0); + } else { + assert.equal(error.code, 'CODE_DATASOURCE_ACCESS_FORBIDDEN'); + const [audit] = database.listAuditLogs(); + assert.equal(audit.action, 'CODE_GATE_BLOCK'); + assert.equal(audit.userId, userId); + assert.equal(audit.detail.sessionId, 7); + assert.equal(audit.detail.code, error.code); + } + }); +} + +test('code gate retains synchronous audit compatibility', async t => { + const database = new PlatformDatabase(':memory:'); + t.after(() => database.closeAsync()); + const service = new CodeExecutionService({ database: { addAuditLog: entry => database.addAuditLog(entry) }, + workspace: {}, config: { pythonBin: '__must_not_start__' } }); + await assert.rejects(service.run({ userId: database.getUserByUsername('admin').id, code: 'import sqlite3' }), + error => error.code === 'CODE_DATASOURCE_ACCESS_FORBIDDEN'); + assert.equal(database.listAuditLogs()[0].action, 'CODE_GATE_BLOCK'); +}); diff --git a/tests/dataSourceTestAtomic.mysql.integration.test.js b/tests/dataSourceTestAtomic.mysql.integration.test.js new file mode 100644 index 0000000..eb83556 --- /dev/null +++ b/tests/dataSourceTestAtomic.mysql.integration.test.js @@ -0,0 +1,49 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { initializePlatformDatabase } from '../src/platformDatabase.js'; +import { DataSourceRepository } from '../src/dataSourceRepository.js'; + +const databaseName = process.env.YOLO_MYSQL_UPLOAD_TEST_DATABASE; +test('live MySQL rejects stale connection results including case-only credential changes', { skip: !databaseName }, async t => { + const platform = await initializePlatformDatabase({ backend: 'mysql', mysql: { + host: process.env.YOLO_MYSQL_UPLOAD_TEST_HOST ?? '127.0.0.1', + port: Number(process.env.YOLO_MYSQL_UPLOAD_TEST_PORT ?? 3306), database: databaseName, + user: process.env.YOLO_MYSQL_UPLOAD_TEST_USER ?? 'root', + password: process.env.YOLO_MYSQL_UPLOAD_TEST_PASSWORD ?? 'test', + } }); + let sourceId; + t.after(async () => { + try { if (sourceId) await platform.database.execute('DELETE FROM data_sources WHERE id=?', [sourceId]); } + finally { await platform.close(); } + }); + const database = platform.database; + const repo = database.dataSourceRepository; + const source = await repo.saveDataSourceAsync({ code: 'cas_source', name: 'CAS source', dbType: 'MYSQL', + host: 'old.example', username: 'reader', encryptedPassword: 'Cipher', options: { tag: 'Case' } }); + sourceId = source.id; + for (const [field, replacement] of [['host', 'OLD.EXAMPLE'], ['username', 'READER'], + ['encrypted_password', 'cipher'], ['options_json', '{"tag":"case"}']]) { + const original = await repo.getDataSourceAsync(source.id); + let edited = false; + const interleaved = new DataSourceRepository(null, { + backend: 'mysql', query: (...args) => database.query(...args), + async execute(sql, bindings) { + if (!edited) { + edited = true; + await database.execute(`UPDATE data_sources SET ${field}=?, last_test_at=NULL, + last_test_ok=NULL, last_test_message='' WHERE id=?`, [replacement, source.id]); + } + return database.execute(sql, bindings); + }, + }); + const result = await interleaved.updateDataSourceTestAsync(source.id, { ok: true, message: 'stale' }, original); + assert.equal(edited, true); + assert.equal(result.lastTestAt, null, field); + assert.equal(result.lastTestOk, null, field); + assert.equal(result.lastTestMessage, '', field); + const fresh = await repo.updateDataSourceTestAsync(source.id, { ok: false, message: 'fresh' }, result); + assert.equal(fresh.lastTestOk, false, field); + assert.equal(fresh.lastTestMessage, 'fresh', field); + assert.ok(fresh.lastTestAt, field); + } +}); diff --git a/tests/dataSourceTestAtomic.test.js b/tests/dataSourceTestAtomic.test.js new file mode 100644 index 0000000..8ab9e55 --- /dev/null +++ b/tests/dataSourceTestAtomic.test.js @@ -0,0 +1,38 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { DataSourceRepository } from '../src/dataSourceRepository.js'; + +for (const ok of [true, false]) { + test(`a stale ${ok ? 'successful' : 'failed'} test cannot overwrite a connection edited immediately before persistence`, async t => { + const database = new PlatformDatabase(':memory:'); + t.after(() => database.closeAsync()); + const original = database.saveDataSource({ code: 'atomic_test', name: 'Atomic test', dbType: 'MYSQL', + host: 'old.example', username: 'reader', options: { connectTimeout: 1234 } }); + let updated = false; + const repository = new DataSourceRepository(null, { + async query(sql, bindings = []) { return database.db.prepare(sql).all(...bindings); }, + async execute(sql, bindings = []) { + if (!updated) { + updated = true; + database.saveDataSource({ ...original, host: 'new.example' }, original.id); + } + const result = database.db.prepare(sql).run(...bindings); + return { affectedRows: Number(result.changes) }; + }, + }); + const result = await repository.updateDataSourceTestAsync(original.id, { ok, message: 'old connection result' }, original); + assert.equal(updated, true); + assert.equal(result.host, 'new.example'); + assert.equal(result.lastTestAt, null); + assert.equal(result.lastTestOk, null); + assert.equal(result.lastTestMessage, ''); + const current = await repository.getDataSourceAsync(original.id); + const fresh = await repository.updateDataSourceTestAsync(original.id, { ok: true, message: 'fresh connection' }, current); + assert.equal(fresh.lastTestOk, true); + assert.equal(fresh.lastTestMessage, 'fresh connection'); + assert.match(fresh.lastTestAt, /^\d{4}-/); + database.db.prepare('DELETE FROM data_sources WHERE id = ?').run(original.id); + assert.equal(await repository.updateDataSourceTestAsync(original.id, { ok }, current), null); + }); +} diff --git a/tests/feedbackAsyncCompletion.test.js b/tests/feedbackAsyncCompletion.test.js new file mode 100644 index 0000000..17ec32b --- /dev/null +++ b/tests/feedbackAsyncCompletion.test.js @@ -0,0 +1,69 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { FeedbackService } from '../src/feedback.js'; +import { GrowthService } from '../src/growth.js'; + +for (const fails of [false, true]) { + test(`negative async feedback ${fails ? 'propagates growth persistence failure' : 'waits for its knowledge gap'}`, async t => { + const database = new PlatformDatabase(':memory:'); + t.after(() => database.closeAsync()); + const failure = new Error('knowledge gap insert failed'); + let release, entered; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const adapter = { + isAsyncPlatformStore: true, + async saveFeedbackAsync(input) { return database.saveFeedback(input); }, + recordKnowledgeGapAsync(input) { + const write = (async () => { + entered(); await gate; + if (fails) throw failure; + return database.recordKnowledgeGap(input); + })(); + // Observe the injected driver's promise even on the broken implementation. + write.catch(() => {}); + return write; + }, + }; + const service = new FeedbackService(adapter, new GrowthService(adapter)); + const args = { userId: database.getUserByUsername('admin').id, themeId: 1, + question: 'Which sales measure?', correct: false, comment: 'Use signed sales' }; + let completed = false; + const pending = service.submitAsync(args); + const outcome = pending.then(value => { completed = true; return { value }; }, + error => { completed = true; return { error }; }); + await ready; + await new Promise(setImmediate); + try { assert.equal(completed, false, 'submission must remain pending until growth persistence finishes'); } + finally { release(); await outcome; } + const result = await outcome; + assert.equal(database.listFeedback().length, 1); + if (fails) { + assert.equal(result.error, failure); + assert.equal(database.listKnowledgeGaps().length, 0); + } else { + assert.equal(result.value.correct, false); + const gap = database.listKnowledgeGaps()[0]; + assert.equal(gap.kind, 'FEEDBACK'); + assert.equal(gap.source, 'USER_FEEDBACK'); + assert.equal(gap.term, args.question); + assert.equal(gap.context.comment, args.comment); + } + }); +} + +test('positive async feedback does not create a knowledge gap', async t => { + const database = new PlatformDatabase(':memory:'); + t.after(() => database.closeAsync()); + const adapter = { isAsyncPlatformStore: true, + async saveFeedbackAsync(input) { return database.saveFeedback(input); }, + recordKnowledgeGapAsync() { throw new Error('positive feedback must not enter growth persistence'); }, + }; + const service = new FeedbackService(adapter, new GrowthService(adapter)); + const result = await service.submitAsync({ userId: database.getUserByUsername('admin').id, + themeId: 1, question: 'Sales?', correct: true }); + assert.equal(result.correct, true); + assert.equal(database.listFeedback().length, 1); + assert.equal(database.listKnowledgeGaps().length, 0); +}); diff --git a/tests/fixtures/mcpHttpRuntime.js b/tests/fixtures/mcpHttpRuntime.js new file mode 100644 index 0000000..1bff3bc --- /dev/null +++ b/tests/fixtures/mcpHttpRuntime.js @@ -0,0 +1,46 @@ +import { createServer } from 'node:http'; +import { PlatformDatabase } from '../../src/database.js'; +import { createMcpModule } from '../../src/mcp/index.js'; +import { generateApiKey } from '../../src/mcp/auth.js'; +import { setMcpConfigAsync, _resetMcpConfigCache } from '../../src/mcp/config.js'; + +export async function mcpHttpRuntime(t, { asyncOnly = false, ...config } = {}) { + const sqlite = new PlatformDatabase(':memory:'); + const fail = () => { throw new Error('sync fallback forbidden'); }; + const database = asyncOnly ? { + isAsyncPlatformStore: true, + async getPlatformSettingAsync(...args) { return sqlite.getPlatformSetting(...args); }, + async savePlatformSettingAsync(...args) { return sqlite.savePlatformSetting(...args); }, + async getMcpApiKeyByHashAsync(...args) { return sqlite.getMcpApiKeyByHash(...args); }, + async addAuditLogAsync(...args) { return sqlite.addAuditLog(...args); }, + getPlatformSetting: fail, savePlatformSetting: fail, + } : sqlite; + const { rawKey } = generateApiKey(sqlite, { name: 'HTTP runtime fixture', tenantId: 'fixture' }); + _resetMcpConfigCache(); + const settings = { enabled: true, mode: 'full', sessionMode: 'stateful', + rateLimit: { rpm: 1000, burst: 100 }, ...config }; + await setMcpConfigAsync(database, settings); + const module = createMcpModule({ database, config: {}, workspace: {}, agent: { + async getIndicatorsForUser() { return { indicators: [], total: 0 }; }, + } }); + const server = createServer((request, response) => { + module.handle({ request, response }).catch(() => { + if (!response.headersSent) response.writeHead(500); + response.end(); + }); + }); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + t.after(async () => { + module.close(); server.closeAllConnections(); + await new Promise(resolve => server.close(resolve)); + await sqlite.closeAsync(); _resetMcpConfigCache(); + }); + const url = `http://127.0.0.1:${server.address().port}/mcp`; + const headers = { authorization: `Bearer ${rawKey}`, 'content-type': 'application/json', + accept: 'application/json, text/event-stream' }; + const initialize = extraHeaders => fetch(url, { method: 'POST', headers: { ...headers, ...extraHeaders }, + body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'initialize', params: { + protocolVersion: '2025-03-26', capabilities: {}, clientInfo: { name: 'fixture', version: '1' }, + } }) }); + return { sqlite, database, settings, module, url, headers, initialize }; +} diff --git a/tests/mcpConfigCache.test.js b/tests/mcpConfigCache.test.js new file mode 100644 index 0000000..79019cf --- /dev/null +++ b/tests/mcpConfigCache.test.js @@ -0,0 +1,85 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { getMcpConfig, getMcpConfigAsync, setMcpConfig, setMcpConfigAsync, + _resetMcpConfigCache } from '../src/mcp/config.js'; + +function fixture(t) { + const database = new PlatformDatabase(':memory:'); + t.after(() => database.closeAsync()); + _resetMcpConfigCache(); + t.after(_resetMcpConfigCache); + return database; +} + +for (const asyncOnly of [false, true]) { + test(`MCP config cache isolates ${asyncOnly ? 'async-only' : 'synchronous'} databases`, async t => { + const a = fixture(t); + const b = fixture(t); + const wrap = database => asyncOnly ? { + isAsyncPlatformStore: true, + async getPlatformSettingAsync(...args) { return database.getPlatformSetting(...args); }, + async savePlatformSettingAsync(...args) { return database.savePlatformSetting(...args); }, + } : database; + const first = wrap(a), second = wrap(b); + const read = asyncOnly ? getMcpConfigAsync : getMcpConfig; + const save = asyncOnly ? setMcpConfigAsync : setMcpConfig; + await save(first, { enabled: true, mode: 'full', tenantScope: ['first'] }); + assert.equal((await read(second)).enabled, false); + await save(second, { enabled: true, mode: 'read', tenantScope: ['second'] }); + assert.deepEqual((await read(first)).tenantScope, ['first']); + assert.deepEqual((await read(second)).tenantScope, ['second']); + }); +} + +test('an older async config read cannot overwrite a completed administrator save', async t => { + const database = fixture(t); + database.savePlatformSetting('mcp', { enabled: true, mode: 'full' }); + let release, started; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { started = resolve; }); + const adapter = { + isAsyncPlatformStore: true, + async getPlatformSettingAsync(...args) { + const snapshot = database.getPlatformSetting(...args); + started(); await gate; return snapshot; + }, + async savePlatformSettingAsync(...args) { return database.savePlatformSetting(...args); }, + }; + const pending = getMcpConfigAsync(adapter); + await ready; + await setMcpConfigAsync(adapter, { enabled: false, mode: 'disabled' }); + release(); + assert.equal((await pending).enabled, false); + assert.equal((await getMcpConfigAsync(adapter)).mode, 'disabled'); + assert.equal(database.getPlatformSetting('mcp').value.enabled, false); +}); + +test('config cache expires after five seconds and reset invalidates every database', t => { + const database = fixture(t); + let now = 1000; + t.mock.method(Date, 'now', () => now); + setMcpConfig(database, { enabled: true, mode: 'full' }); + database.savePlatformSetting('mcp', { enabled: false, mode: 'disabled' }); + now = 5999; + assert.equal(getMcpConfig(database).enabled, true); + now = 6000; + assert.equal(getMcpConfig(database).enabled, false); + database.savePlatformSetting('mcp', { enabled: true, mode: 'read' }); + _resetMcpConfigCache(); + assert.equal(getMcpConfig(database).enabled, true); +}); + +test('failed config saves leave the last successful cached policy intact', async t => { + const database = fixture(t); + const failure = new Error('settings write failed'); + const adapter = { + isAsyncPlatformStore: true, + async getPlatformSettingAsync(...args) { return database.getPlatformSetting(...args); }, + async savePlatformSettingAsync() { throw failure; }, + }; + database.savePlatformSetting('mcp', { enabled: false, mode: 'disabled' }); + await getMcpConfigAsync(adapter); + await assert.rejects(setMcpConfigAsync(adapter, { enabled: true, mode: 'full' }), error => error === failure); + assert.equal((await getMcpConfigAsync(adapter)).enabled, false); +}); diff --git a/tests/mcpContractAuthorization.test.js b/tests/mcpContractAuthorization.test.js new file mode 100644 index 0000000..3dd6155 --- /dev/null +++ b/tests/mcpContractAuthorization.test.js @@ -0,0 +1,55 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { createMcpServer } from '../src/mcp/server.js'; +import { auditToolCall } from '../src/mcp/audit.js'; + +for (const asyncOnly of [false, true]) for (const denied of ['tenant', 'scope']) { + test(`saved MCP contracts reject ${denied} violations on ${asyncOnly ? 'async-only' : 'SQLite'} storage`, async t => { + const database = new PlatformDatabase(':memory:'); + t.after(() => database.closeAsync()); + const userId = database.getUserByUsername('admin').id; + database.saveMcpQueryContract({ contractId: 'mcq_private', tenantId: 'tenant-a', userId, + themeId: 1, sourceType: 'INDICATOR', indicatorId: 'sales_amount', + contractJson: { metrics: ['sales_amount'], dimensions: [], dateInfo: { dateMode: 'ALL' } }, + expiresAt: '2999-01-01T00:00:00.000Z' }); + const fail = () => { throw new Error('sync fallback forbidden'); }; + let contexts = 0, queries = 0; + const deps = { database: asyncOnly ? { + async getMcpQueryContractAsync(id) { return database.getMcpQueryContract(id); }, + async addAuditLogAsync(entry) { return database.addAuditLog(entry); }, + getMcpQueryContract: fail, + } : database, auditToolCall, + requestContext: denied === 'tenant' ? { tenantId: 'tenant-b', tenantScope: [], keyId: 'key-b' } + : { tenantId: 'tenant-a', tenantScope: [2], keyId: 'key-a' }, + agent: { + async resolveContextAsync() { contexts++; return { theme: { id: 1 }, scope: { allowed: true } }; }, + queryContractCompiler: { review: () => ({ valid: true }) }, + indicatorService: { + async getIndicator() { return { id: 'sales_amount' }; }, + async queryIndicator() { queries++; return { columns: ['sales'], rows: [{ sales: 12 }] }; }, + }, + } }; + const server = createMcpServer({ config: { mode: 'full' }, deps }); + t.after(() => server.close()); + const execute = () => server._registeredTools.execute_query.handler({ contractId: 'mcq_private' }); + const crossTenant = await execute(); + assert.equal(crossTenant.isError, true); + assert.equal(JSON.parse(crossTenant.content[0].text).error.code, -32002); + assert.equal(contexts, 0); + assert.equal(queries, 0); + deps.requestContext = { tenantId: 'tenant-a', tenantScope: [2], keyId: 'key-a' }; + const narrowed = await execute(); + assert.equal(narrowed.isError, true); + assert.equal(JSON.parse(narrowed.content[0].text).error.code, -32002); + assert.equal(contexts, 0); + assert.equal(queries, 0); + deps.requestContext = { tenantId: 'tenant-a', tenantScope: ['1'], keyId: 'key-a' }; + const allowed = await execute(); + assert.equal(allowed.isError, undefined); + assert.equal(JSON.parse(allowed.content[0].text).rows[0].sales, 12); + assert.equal(contexts, 1); + assert.equal(queries, 1); + assert.deepEqual(database.listAuditLogs().map(row => row.detail.outcome), ['success', 'error', 'error']); + }); +} diff --git a/tests/mcpOriginConfig.test.js b/tests/mcpOriginConfig.test.js new file mode 100644 index 0000000..b40b72b --- /dev/null +++ b/tests/mcpOriginConfig.test.js @@ -0,0 +1,31 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { mcpHttpRuntime } from './fixtures/mcpHttpRuntime.js'; +import { getDefaultMcpConfig, getMcpConfigAsync, setMcpConfigAsync, _resetMcpConfigCache } from '../src/mcp/config.js'; + +for (const asyncOnly of [false, true]) { + test(`persisted MCP origins control HTTP admission on ${asyncOnly ? 'async-only' : 'SQLite'} storage`, async t => { + const env = await mcpHttpRuntime(t, { asyncOnly, allowedOrigins: ['https://trusted.example'] }); + const denied = await env.initialize({ origin: 'https://other.example' }); + assert.equal(denied.status, 403); + assert.equal((await denied.json()).error.code, -32002); + assert.equal((await env.initialize({})).status, 403); + const allowed = await env.initialize({ origin: 'https://trusted.example' }); + assert.equal(allowed.status, 200); + assert.match(await allowed.text(), /yolo-data-agent/); + _resetMcpConfigCache(); + assert.deepEqual((await getMcpConfigAsync(env.database)).allowedOrigins, ['https://trusted.example']); + assert.deepEqual(env.sqlite.getPlatformSetting('mcp').value.allowedOrigins, ['https://trusted.example']); + await setMcpConfigAsync(env.database, { ...env.settings, allowedOrigins: ['invalid-origin'] }); + assert.equal((await env.initialize({ origin: 'https://trusted.example' })).status, 403); + await setMcpConfigAsync(env.database, { ...env.settings, allowedOrigins: [] }); + assert.equal((await env.initialize({})).status, 200); + }); +} + +test('default MCP origin arrays are independently owned', () => { + const first = getDefaultMcpConfig(); + assert.deepEqual(first.allowedOrigins, []); + first.allowedOrigins.push('https://example.test'); + assert.deepEqual(getDefaultMcpConfig().allowedOrigins, []); +}); diff --git a/tests/mcpRuntimePolicy.test.js b/tests/mcpRuntimePolicy.test.js new file mode 100644 index 0000000..7efcc08 --- /dev/null +++ b/tests/mcpRuntimePolicy.test.js @@ -0,0 +1,56 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { Client } from '@modelcontextprotocol/sdk/client/index.js'; +import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; +import { mcpHttpRuntime } from './fixtures/mcpHttpRuntime.js'; +import { setMcpConfigAsync } from '../src/mcp/config.js'; + +async function connect(t, env) { + const client = new Client({ name: 'runtime-policy', version: '1' }); + const transport = new StreamableHTTPClientTransport(new URL(env.url), { requestInit: { headers: env.headers } }); + await client.connect(transport); + t.after(() => client.close()); + return client; +} +for (const sessionMode of ['stateful', 'stateless']) for (const change of ['mode', 'toolOverride']) { + test(`${sessionMode} MCP refreshes ${change} on the next request`, async t => { + const env = await mcpHttpRuntime(t, { sessionMode, asyncOnly: sessionMode === 'stateless' }); + const first = await connect(t, env); + assert.ok((await first.listTools()).tools.some(tool => tool.name === 'execute_query')); + const policy = { ...env.settings, ...(change === 'mode' ? { mode: 'read' } : { toolOverrides: { execute_query: false } }) }; + await setMcpConfigAsync(env.database, policy); + const current = await connect(t, env); + assert.equal((await current.listTools()).tools.some(tool => tool.name === 'execute_query'), false); + let stale; + try { stale = await first.listTools(); } catch { stale = null; } + assert.equal(stale?.tools.some(tool => tool.name === 'execute_query') ?? false, false); + await setMcpConfigAsync(env.database, env.settings); + const restored = await connect(t, env); + assert.ok((await restored.listTools()).tools.some(tool => tool.name === 'execute_query')); + }); +} + +test('MCP applies new burst limits while stable settings preserve consumed tokens', async t => { + const env = await mcpHttpRuntime(t, { asyncOnly: true, rateLimit: { rpm: 0.01, burst: 3 } }); + assert.equal((await env.initialize({})).status, 200); + await setMcpConfigAsync(env.database, { ...env.settings, rateLimit: { rpm: 0.01, burst: 1 } }); + assert.equal((await env.initialize({})).status, 200); + assert.equal((await env.initialize({})).status, 429); + await setMcpConfigAsync(env.database, { ...env.settings, rateLimit: { rpm: 0.01, burst: 2 } }); + assert.equal((await env.initialize({})).status, 200); + assert.equal((await env.initialize({})).status, 200); + assert.equal((await env.initialize({})).status, 429); +}); + +test('MCP applies updated refill rates without waiting for a restart', async t => { + let now = Date.now(); + t.mock.method(Date, 'now', () => now); + const env = await mcpHttpRuntime(t, { rateLimit: { rpm: 60, burst: 1 } }); + assert.equal((await env.initialize({})).status, 200); + assert.equal((await env.initialize({})).status, 429); + await setMcpConfigAsync(env.database, { ...env.settings, rateLimit: { rpm: 120, burst: 1 } }); + assert.equal((await env.initialize({})).status, 200); + now += 500; + assert.equal((await env.initialize({})).status, 200); + assert.equal((await env.initialize({})).status, 429); +}); diff --git a/tests/mcpSnapshotAuthorization.test.js b/tests/mcpSnapshotAuthorization.test.js new file mode 100644 index 0000000..8928e35 --- /dev/null +++ b/tests/mcpSnapshotAuthorization.test.js @@ -0,0 +1,64 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { WorkspaceService } from '../src/workspace.js'; +import { resolveAccessScope } from '../src/permissions.js'; +import { createMcpServer } from '../src/mcp/server.js'; +import { auditToolCall } from '../src/mcp/audit.js'; + +for (const asyncOnly of [false, true]) for (const denied of ['scope', 'grant']) { + test(`MCP snapshot rejects ${denied} violations using ${asyncOnly ? 'async-only' : 'SQLite'} storage`, async t => { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'snapshot-auth-')); + const database = new PlatformDatabase(path.join(directory, 'test.db')); + t.after(async () => { await database.closeAsync(); fs.rmSync(directory, { recursive: true, force: true }); }); + const service = new WorkspaceService(database); + const user = database.getUserByUsername('east_manager'); + const theme = database.listThemes().find(item => item.name === '经营总览'); + const session = database.createChatSession({ userId: user.id, themeId: theme.id }); + const parent = service.ensureForSession({ userId: user.id, themeId: theme.id, sessionId: session.id }); + const artifact = database.createWorkspaceArtifact({ workspaceId: parent.id, userId: user.id, + sessionId: session.id, artifactType: 'RESULT', title: 'Private sales', + payload: { columns: Array.from({ length: 60 }, (_, i) => `col${i}`), + rows: Array.from({ length: 25 }, (_, i) => ({ confidential: i })), text: 'x'.repeat(1200) } }); + if (denied === 'grant') database.db.prepare('UPDATE user_theme_grants SET can_query=0 WHERE user_id=? AND theme_id=?').run(user.id, theme.id); + let missingParent = false; + const getParent = (...args) => missingParent ? null : database.getWorkspace(...args); + const resolve = (userId, themeId) => ({ scope: resolveAccessScope({ user: database.getUser(userId), + theme: database.getTheme(themeId), permissionProfile: database.getPermissionProfile(userId) }) }); + const fail = () => { throw new Error('sync fallback forbidden'); }; + const deps = { requestContext: { tenantId: 'tenant-a', tenantScope: denied === 'scope' ? [-1] : [theme.id] }, + database: asyncOnly ? { + async getWorkspaceAsync(...args) { return getParent(...args); }, + async addAuditLogAsync(entry) { return database.addAuditLog(entry); }, getWorkspace: fail, + } : { getWorkspace: getParent, addAuditLog: entry => database.addAuditLog(entry) }, + workspace: asyncOnly ? { getArtifactAsync: input => service.getArtifactAsync(input), getArtifact: fail } + : { getArtifact: input => service.getArtifact(input) }, + agent: asyncOnly ? { async resolveContextAsync(...args) { return resolve(...args); }, resolveContext: fail } + : { resolveContext: resolve }, auditToolCall, + }; + const server = createMcpServer({ config: { mode: 'full' }, deps }); + t.after(() => server.close()); + const read = (userId = user.id) => server._registeredTools.get_snapshot.handler({ artifactId: artifact.id, userId }); + const forbidden = await read(); + assert.equal(forbidden.isError, true); + assert.equal(JSON.parse(forbidden.content[0].text).error.code, -32002); + assert.equal(forbidden.content[0].text.includes('confidential'), false); + deps.requestContext = { tenantId: 'tenant-a', tenantScope: [String(theme.id)] }; + database.db.prepare('UPDATE user_theme_grants SET can_query=1 WHERE user_id=? AND theme_id=?').run(user.id, theme.id); + const allowed = await read(); + assert.equal(allowed.isError, undefined); + const preview = JSON.parse(allowed.content[0].text); + assert.equal(preview.rowCount, 25); + assert.equal(preview.previewRows.length, 20); + assert.equal(preview.columns.length, 50); + assert.equal(preview.textExcerpt.length, 1000); + assert.equal((await read(database.getUserByUsername('admin').id)).isError, true); + assert.equal((await read(0)).isError, true); + missingParent = true; + assert.equal((await read()).isError, true); + assert.equal(database.listAuditLogs()[0].detail.outcome, 'error'); + }); +} diff --git a/tests/processArtifacts.concurrent.test.js b/tests/processArtifacts.concurrent.test.js new file mode 100644 index 0000000..5c83e55 --- /dev/null +++ b/tests/processArtifacts.concurrent.test.js @@ -0,0 +1,76 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { PlatformDatabase } from '../src/database.js'; +import { WorkspaceService } from '../src/workspace.js'; +import { createProcessArtifactRecorderAsync } from '../src/processArtifacts.js'; + +function fixture(t) { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'process-concurrency-')); + const database = new PlatformDatabase(path.join(directory, 'test.db')); + t.after(async () => { + await database.closeAsync(); + fs.rmSync(directory, { recursive: true, force: true }); + }); + const service = new WorkspaceService(database); + const userId = database.getUserByUsername('admin').id; + const session = database.createChatSession({ userId, themeId: 1 }); + const workspace = service.ensureForSession({ userId, themeId: 1, sessionId: session.id }); + return { database, service, options: { workspaceId: workspace.id, userId, sessionId: session.id } }; +} + +const input = sequence => ({ tool: 'query', result: { columns: [{ name: 'sales' }], rows: [{ sales: sequence }] }, + step: { sequence, title: `Step ${sequence}` } }); + +for (const maxSteps of [2, 3]) { + test(`overlapping process writes preserve ${maxSteps === 2 ? 'the step cap' : 'ordered lineage'}`, async t => { + const { service, options } = fixture(t); + let release, entered, writes = 0; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const adapter = { async createProcessArtifactAsync(args) { + if (++writes === 1) { entered(); await gate; } + return service.createProcessArtifactAsync(args); + } }; + const artifactIds = [], events = []; + const record = createProcessArtifactRecorderAsync({ ...options, service: adapter, maxSteps, + artifactIds, emit: event => events.push(event) }); + assert.equal(await record({ tool: 'catalog', result: {} }), null); + const first = record(input(1)); + await ready; + const second = record(input(2)); + const third = record(input(3)); + release(); + const results = await Promise.all([first, second, third]); + const stored = await service.listArtifactsAsync(options); + assert.equal(stored.length, maxSteps); + const created = results.filter(Boolean); + assert.equal(created.length, maxSteps); + assert.deepEqual(artifactIds, created.map(item => item.id)); + assert.deepEqual(events.map(event => event.artifactId), artifactIds); + for (let index = 1; index < created.length; index++) { + assert.deepEqual(created[index].metadata.inputArtifactIds, [created[index - 1].id]); + } + if (maxSteps === 2) assert.equal(results[2], null); + }); +} + +test('a failed process write does not poison the queue or consume its cap', async t => { + const { service, options } = fixture(t); + let attempts = 0; + const record = createProcessArtifactRecorderAsync({ ...options, maxSteps: 1, + service: { async createProcessArtifactAsync(args) { + if (++attempts === 1) throw new Error('write rejected'); + return service.createProcessArtifactAsync(args); + } }, + }); + const [failed, saved, capped] = await Promise.all([record(input(1)), record(input(2)), record(input(3))]); + assert.equal(failed, null); + assert.ok(saved.id); + assert.equal(capped, null); + assert.equal(attempts, 2); + assert.equal((await service.listArtifactsAsync(options)).length, 1); + assert.deepEqual(saved.metadata.inputArtifactIds, []); +});