import { createHash } from 'node:crypto'; import { createRequire } from 'node:module'; import mysql from 'mysql2/promise'; const require = createRequire(import.meta.url); const mysql2Version = require('mysql2/package.json').version; const DEVELOPMENT_DATABASE = 'a_share_analysis_dev'; const VALID_TRIGGER_TIMINGS = new Set(['BEFORE', 'AFTER']); const VALID_TRIGGER_EVENTS = new Set(['INSERT', 'UPDATE', 'DELETE']); const CATALOG_QUERIES = Object.freeze({ tables: `SELECT TABLE_NAME, TABLE_TYPE FROM information_schema.TABLES WHERE TABLE_SCHEMA = ? ORDER BY TABLE_NAME`, viewDependencies: `SELECT VIEW_NAME, TABLE_SCHEMA AS REFERENCED_SCHEMA_NAME, TABLE_NAME AS REFERENCED_TABLE_NAME FROM information_schema.VIEW_TABLE_USAGE WHERE VIEW_SCHEMA = ? ORDER BY VIEW_NAME, REFERENCED_SCHEMA_NAME, REFERENCED_TABLE_NAME`, routines: `SELECT ROUTINE_NAME, ROUTINE_TYPE FROM information_schema.ROUTINES WHERE ROUTINE_SCHEMA = ? ORDER BY ROUTINE_TYPE, ROUTINE_NAME`, triggers: `SELECT TRIGGER_NAME, EVENT_OBJECT_TABLE, ACTION_TIMING, EVENT_MANIPULATION, ACTION_ORDER FROM information_schema.TRIGGERS WHERE TRIGGER_SCHEMA = ? ORDER BY EVENT_OBJECT_TABLE, ACTION_TIMING, EVENT_MANIPULATION, ACTION_ORDER, TRIGGER_NAME`, events: `SELECT EVENT_NAME FROM information_schema.EVENTS WHERE EVENT_SCHEMA = ? ORDER BY EVENT_NAME`, }); const SERVER_METADATA_QUERY = `SELECT VERSION() AS SERVER_VERSION, DEFAULT_CHARACTER_SET_NAME AS DATABASE_CHARSET, DEFAULT_COLLATION_NAME AS DATABASE_COLLATION FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = ?`; const CREATE_RESULT_COLUMNS = Object.freeze({ TABLE: 'Create Table', VIEW: 'Create View', PROCEDURE: 'Create Procedure', FUNCTION: 'Create Function', TRIGGER: 'SQL Original Statement', EVENT: 'Create Event', }); export const MYSQL2_EXPORTER_METADATA = Object.freeze({ name: 'mysql2-show-create', version: '1.0.0', driverName: 'mysql2', driverVersion: mysql2Version, }); export const DEFAULT_EXPORT_LIMITS = Object.freeze({ connectTimeoutMs: 5_000, queryTimeoutMs: 5_000, totalTimeoutMs: 30_000, maxObjects: 5_000, maxDefinitionBytes: 1_048_576, }); export class SchemaExportError extends Error { constructor(code, message) { super(message); this.name = 'SchemaExportError'; this.code = code; } } /** * 配置在 Schema 治理入口已完成 development、主机白名单和固定库名校验;这里仅映射 * 连接所需字段,固定禁用多语句以防止未来元数据查询被拼接为任意 SQL。 */ export function buildConnectionOptions(config) { return { host: config.MYSQL_HOST, port: Number(config.MYSQL_PORT), user: config.MYSQL_USERNAME, password: config.MYSQL_PASSWORD, database: config.MYSQL_DATABASE, charset: 'utf8mb4', timezone: '+08:00', connectTimeout: DEFAULT_EXPORT_LIMITS.connectTimeoutMs, multipleStatements: false, enableKeepAlive: true, }; } /** * 元数据对象名仍需防御性转义,避免控制字符破坏 `SHOW CREATE` 查询边界;名称不合法时 * 不回显输入,确保异常路径不会把潜在敏感值写入日志。 */ export function quoteIdentifier(identifier) { if (typeof identifier !== 'string' || identifier.length === 0 || /[\u0000-\u001F\u007F]/.test(identifier)) { throw new SchemaExportError('INVALID_IDENTIFIER', 'Schema 标识符不能为空且不得包含控制字符'); } return `\`${identifier.replaceAll('`', '``')}\``; } /** * 导出器只接受项目数据库兼容基线,拒绝 MariaDB 和 MySQL 5.7 之外的服务器,避免相同 * `SHOW CREATE` 文本在不兼容语义下被误认为可恢复的 Schema 快照。 */ export function assertMysql57ServerMetadata(metadata) { const serverVersion = metadata?.serverVersion ?? ''; if (!/^5\.7\.\d+(?:[-+].*)?$/.test(serverVersion) || /mariadb/i.test(serverVersion)) { throw new SchemaExportError('UNSUPPORTED_SERVER_VERSION', 'MySQL 服务端必须为 5.7.x'); } if (metadata.databaseCharset !== 'utf8mb4') { throw new SchemaExportError('INVALID_DATABASE_CHARSET', '数据库字符集必须为 utf8mb4'); } if (metadata.databaseCollation !== 'utf8mb4_general_ci') { throw new SchemaExportError('INVALID_DATABASE_COLLATION', '数据库排序规则必须为 utf8mb4_general_ci'); } } function requireCatalogText(value) { if (typeof value !== 'string' || value.length === 0) { throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 对象目录字段无效'); } return value; } async function queryCatalogRows(connection, sql, database, limits) { let result; try { result = await connection.query({ sql, values: [database], timeout: limits.queryTimeoutMs, }); } catch (error) { // mysql2 错误可能包含连接 URL、用户名或密码,目录读取的公共接口一律不透传原始文本。 sanitizeDatabaseQueryError(error); } if (!Array.isArray(result) || !Array.isArray(result[0])) { throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 对象目录查询结果无效'); } return result[0]; } function assertObjectLimit(count, limits) { if (!Number.isInteger(limits.maxObjects) || limits.maxObjects < 0 || count > limits.maxObjects) { throw new SchemaExportError('OBJECT_LIMIT_EXCEEDED', 'Schema 对象数量超过安全上限'); } } /** * 目录查询只从 information_schema 读取固定列,库名始终作为参数传入。每类对象 * 在进入后续 SHOW CREATE 前先转换成稳定的小对象,并在累计数量越界时立即停止。 */ export async function readSchemaCatalog(connection, database, limits = DEFAULT_EXPORT_LIMITS) { const tableRows = await queryCatalogRows(connection, CATALOG_QUERIES.tables, database, limits); const tables = []; const views = []; for (const row of tableRows) { const name = requireCatalogText(row.TABLE_NAME); if (row.TABLE_TYPE === 'BASE TABLE') tables.push(name); else if (row.TABLE_TYPE === 'VIEW') views.push(name); else throw new SchemaExportError('UNKNOWN_TABLE_TYPE', 'Schema 目录包含未知表类型'); } tables.sort(); views.sort(); let objectCount = tables.length + views.length; assertObjectLimit(objectCount, limits); const dependencyRows = await queryCatalogRows( connection, CATALOG_QUERIES.viewDependencies, database, limits, ); const viewDependencies = dependencyRows.map((row) => ({ view: requireCatalogText(row.VIEW_NAME), referencedSchema: requireCatalogText(row.REFERENCED_SCHEMA_NAME), referencedObject: requireCatalogText(row.REFERENCED_TABLE_NAME), })); const routineRows = await queryCatalogRows(connection, CATALOG_QUERIES.routines, database, limits); const routines = routineRows.map((row) => { const type = requireCatalogText(row.ROUTINE_TYPE); if (!['FUNCTION', 'PROCEDURE'].includes(type)) { throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 目录包含未知例程类型'); } return { name: requireCatalogText(row.ROUTINE_NAME), type }; }); routines.sort((left, right) => left.type.localeCompare(right.type) || left.name.localeCompare(right.name)); objectCount += routines.length; assertObjectLimit(objectCount, limits); const triggerRows = await queryCatalogRows(connection, CATALOG_QUERIES.triggers, database, limits); const triggers = triggerRows.map((row) => { const timing = requireCatalogText(row.ACTION_TIMING); const event = requireCatalogText(row.EVENT_MANIPULATION); const rawOrder = row.ACTION_ORDER; const order = Number(rawOrder); if ( !VALID_TRIGGER_TIMINGS.has(timing) || !VALID_TRIGGER_EVENTS.has(event) || (typeof rawOrder !== 'number' && typeof rawOrder !== 'string') || (typeof rawOrder === 'string' && rawOrder.trim().length === 0) || !Number.isInteger(order) || order < 1 ) { throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 触发器执行顺序无效'); } return { name: requireCatalogText(row.TRIGGER_NAME), table: requireCatalogText(row.EVENT_OBJECT_TABLE), timing, event, order, }; }); triggers.sort( (left, right) => left.table.localeCompare(right.table) || left.timing.localeCompare(right.timing) || left.event.localeCompare(right.event) || left.order - right.order || left.name.localeCompare(right.name), ); objectCount += triggers.length; assertObjectLimit(objectCount, limits); const eventRows = await queryCatalogRows(connection, CATALOG_QUERIES.events, database, limits); const events = eventRows.map((row) => requireCatalogText(row.EVENT_NAME)).sort(); objectCount += events.length; assertObjectLimit(objectCount, limits); return Object.freeze({ tables: Object.freeze(tables), views: Object.freeze(views), viewDependencies: Object.freeze(viewDependencies.map((dependency) => Object.freeze(dependency))), routines: Object.freeze(routines.map((routine) => Object.freeze(routine))), triggers: Object.freeze(triggers.map((trigger) => Object.freeze(trigger))), events: Object.freeze(events), }); } /** * 视图恢复顺序使用稳定 Kahn 拓扑排序。跨库或未知依赖不能被默默忽略,否则恢复 * 出来的 Schema 可能引用未捕获对象;同层节点始终按名称排序以保证逐字节确定性。 */ export function orderViewNames(catalog) { const tables = new Set(catalog.tables ?? []); const views = new Set(catalog.views ?? []); const incoming = new Map([...views].map((view) => [view, 0])); const dependents = new Map([...views].map((view) => [view, []])); for (const dependency of catalog.viewDependencies ?? []) { if (dependency.referencedSchema !== DEVELOPMENT_DATABASE) { throw new SchemaExportError( 'CROSS_SCHEMA_VIEW_DEPENDENCY', 'Schema 视图包含跨库依赖', ); } if (!views.has(dependency.view)) { throw new SchemaExportError('UNKNOWN_VIEW_DEPENDENCY', 'Schema 视图依赖来源未知'); } if (tables.has(dependency.referencedObject)) continue; if (!views.has(dependency.referencedObject)) { throw new SchemaExportError('UNKNOWN_VIEW_DEPENDENCY', 'Schema 视图依赖对象未知'); } incoming.set(dependency.view, incoming.get(dependency.view) + 1); dependents.get(dependency.referencedObject).push(dependency.view); } for (const names of dependents.values()) names.sort(); const ready = [...views].filter((view) => incoming.get(view) === 0).sort(); const ordered = []; while (ready.length > 0) { const view = ready.shift(); ordered.push(view); for (const dependent of dependents.get(view)) { incoming.set(dependent, incoming.get(dependent) - 1); if (incoming.get(dependent) === 0) { ready.push(dependent); ready.sort(); } } } if (ordered.length !== views.size) { throw new SchemaExportError('VIEW_DEPENDENCY_CYCLE', 'Schema 视图依赖存在循环'); } return ordered; } function sha256(content) { return createHash('sha256').update(content, 'utf8').digest('hex'); } /** 只屏蔽 SQL 字符串字面量,保留标识符参与 DEFINER 和凭据扫描。 */ function maskStringLiterals(sql) { return String(sql).replace(/'(?:''|\\.|[^'\\])*'/gs, "''"); } function removeTableAutoIncrementOutsideLiterals(sql) { const literals = []; const masked = String(sql).replace(/'(?:''|\\.|[^'\\])*'/gs, (literal) => { const index = literals.push(literal) - 1; return `\u0001${index}\u0002`; }); const cleaned = masked.replace( /\sAUTO_INCREMENT=\d+(?=\s(?:DEFAULT\s+)?CHARSET=)/i, '', ); return cleaned.replace(/\u0001(\d+)\u0002/g, (_match, index) => literals[Number(index)]); } /** * SHOW CREATE 的格式由服务端决定,这里只移除与结构无关的账号绑定、当前表自增值、 * 行尾分号和换行差异,不重排对象体,也不删除字符串字面量中的业务文本。 */ export function normalizeCreateDefinition(definition) { const normalized = String(definition) .replace(/\r\n?/g, '\n') .replace( /^(CREATE\s+(?:ALGORITHM=\S+\s+)?)DEFINER=`(?:``|[^`])+`@`(?:`(?:``|[^`])+`|'(?:''|[^'])+'|\S+)\s+/i, '$1', ) .trim() .replace(/;+\s*$/, ''); const withoutCurrentAutoIncrement = removeTableAutoIncrementOutsideLiterals(normalized); const inspectable = maskStringLiterals(withoutCurrentAutoIncrement); if ( /\bDEFINER\s*=/i.test(inspectable) || /\b(?:password|passwd|pwd|api[_-]?key|api[_-]?token|access[_-]?token|client[_-]?secret|secret)\s*[:=]/i.test( inspectable, ) || /\b(?:mysql|mariadb):\/\/[^\s'"`]+/i.test(inspectable) || /(?:--|\/\*)\s*(?:host|hostname|server)\s*[:=]\s*[^\s*]+/i.test(inspectable) ) { throw new SchemaExportError('UNSAFE_DEFINITION', 'Schema 对象定义包含禁止内容'); } return withoutCurrentAutoIncrement; } function sanitizeDatabaseQueryError(error) { if (error instanceof SchemaExportError) throw error; throw new SchemaExportError('DATABASE_QUERY_FAILED', '开发库 Schema 查询失败'); } export async function readMysql57ServerMetadata( connection, database, limits = DEFAULT_EXPORT_LIMITS, ) { let rows; try { [rows] = await connection.query({ sql: SERVER_METADATA_QUERY, values: [database], timeout: limits.queryTimeoutMs, }); } catch (error) { sanitizeDatabaseQueryError(error); } if (!Array.isArray(rows) || rows.length !== 1) { throw new SchemaExportError('DATABASE_METADATA_INVALID', '开发库元数据查询结果无效'); } const metadata = { serverVersion: rows[0].SERVER_VERSION, databaseCharset: rows[0].DATABASE_CHARSET, databaseCollation: rows[0].DATABASE_COLLATION, }; assertMysql57ServerMetadata(metadata); return Object.freeze(metadata); } function getCreateDescriptors(catalog) { const viewOrder = orderViewNames(catalog); const viewDefinitions = new Map((catalog.views ?? []).map((name) => [name, { name, type: 'VIEW' }])); return [ ...(catalog.tables ?? []).map((name) => ({ name, type: 'TABLE' })), ...(catalog.routines ?? []).map(({ name, type }) => ({ name, type })), ...viewOrder.map((name) => viewDefinitions.get(name)), ...(catalog.triggers ?? []).map(({ name }) => ({ name, type: 'TRIGGER' })), ...(catalog.events ?? []).map((name) => ({ name, type: 'EVENT' })), ]; } async function readCreateDefinition(connection, descriptor, limits) { const sql = `SHOW CREATE ${descriptor.type} ${quoteIdentifier(DEVELOPMENT_DATABASE)}.${quoteIdentifier(descriptor.name)}`; let rows; try { [rows] = await connection.query({ sql, timeout: limits.queryTimeoutMs }); } catch (error) { sanitizeDatabaseQueryError(error); } if (!Array.isArray(rows) || rows.length === 0) { throw new SchemaExportError('SHOW_CREATE_EMPTY', 'SHOW CREATE 未返回对象定义'); } const rawDefinition = rows[0]?.[CREATE_RESULT_COLUMNS[descriptor.type]]; if (typeof rawDefinition !== 'string' || rawDefinition.length === 0) { throw new SchemaExportError('SHOW_CREATE_SHAPE_INVALID', 'SHOW CREATE 返回列无效'); } if (Buffer.byteLength(rawDefinition, 'utf8') > limits.maxDefinitionBytes) { throw new SchemaExportError('DEFINITION_LIMIT_EXCEEDED', 'Schema 对象定义超过安全上限'); } return Object.freeze({ ...descriptor, definition: normalizeCreateDefinition(rawDefinition), sourceSha256: sha256(rawDefinition.replace(/\r\n?/g, '\n')), }); } export async function readDefinitions(connection, catalog, limits = DEFAULT_EXPORT_LIMITS) { const definitions = []; for (const descriptor of getCreateDescriptors(catalog)) { definitions.push(await readCreateDefinition(connection, descriptor, limits)); } return Object.freeze(definitions); } function appendDefinition(parts, definition) { if (['TABLE', 'VIEW'].includes(definition.type)) { parts.push(`${definition.definition};`); return; } parts.push(`DELIMITER ;;\n${definition.definition} ;;\nDELIMITER ;`); } /** 固定恢复顺序和会话保护,不重新格式化对象体。 */ export function renderSchemaSql(catalog, definitions) { const definitionsByKey = new Map( definitions.map((definition) => [`${definition.type}:${definition.name}`, definition]), ); const parts = [ 'SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0;', 'SET NAMES utf8mb4;', ]; for (const descriptor of getCreateDescriptors(catalog)) { const definition = definitionsByKey.get(`${descriptor.type}:${descriptor.name}`); if (!definition) { throw new SchemaExportError('SHOW_CREATE_SHAPE_INVALID', 'Schema 对象定义集合不完整'); } appendDefinition(parts, definition); } parts.push('SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS;'); return `${parts.join('\n\n')}\n`; } function assertWithinDeadline(startedAt, now, limits) { if (now() - startedAt > limits.totalTimeoutMs) { throw new SchemaExportError('EXPORT_TIMEOUT', 'Schema 导出超过总时长限制'); } } function buildObjectCounts(catalog) { return Object.freeze({ tables: catalog.tables.length, views: catalog.views.length, triggers: catalog.triggers.length, routines: catalog.routines.length, events: catalog.events.length, }); } /** * 单连接执行两轮目录和定义读取;任何漂移都失败。底层数据库异常统一脱敏,连接在 * 成功、查询失败、校验失败和超时路径均只关闭一次。 */ export async function exportMysql57Schema({ config, createConnection = mysql.createConnection, now = Date.now, limits = DEFAULT_EXPORT_LIMITS, }) { const startedAt = now(); let connection; let result; let failure; try { connection = await createConnection(buildConnectionOptions(config)); const guardedConnection = { async query(options) { assertWithinDeadline(startedAt, now, limits); const queryResult = await connection.query(options); assertWithinDeadline(startedAt, now, limits); return queryResult; }, }; const serverMetadata = await readMysql57ServerMetadata( guardedConnection, config.MYSQL_DATABASE, limits, ); const beforeCatalog = await readSchemaCatalog( guardedConnection, config.MYSQL_DATABASE, limits, ); const beforeDefinitions = await readDefinitions(guardedConnection, beforeCatalog, limits); const afterCatalog = await readSchemaCatalog( guardedConnection, config.MYSQL_DATABASE, limits, ); if (sha256(JSON.stringify(beforeCatalog)) !== sha256(JSON.stringify(afterCatalog))) { throw new SchemaExportError( 'CATALOG_CHANGED_DURING_EXPORT', 'Schema 对象目录在导出期间发生变化', ); } const afterDefinitions = await readDefinitions(guardedConnection, afterCatalog, limits); if (sha256(JSON.stringify(beforeDefinitions)) !== sha256(JSON.stringify(afterDefinitions))) { throw new SchemaExportError( 'DEFINITION_CHANGED_DURING_EXPORT', 'Schema 对象定义在导出期间发生变化', ); } assertWithinDeadline(startedAt, now, limits); result = Object.freeze({ metadata: Object.freeze({ exporterName: MYSQL2_EXPORTER_METADATA.name, exporterVersion: MYSQL2_EXPORTER_METADATA.version, driverName: MYSQL2_EXPORTER_METADATA.driverName, driverVersion: MYSQL2_EXPORTER_METADATA.driverVersion, ...serverMetadata, }), rawSql: renderSchemaSql(beforeCatalog, beforeDefinitions), objectCounts: buildObjectCounts(beforeCatalog), durationMs: now() - startedAt, }); } catch (error) { failure = error instanceof SchemaExportError ? error : new SchemaExportError('DATABASE_QUERY_FAILED', '开发库 Schema 查询失败'); } finally { if (connection) { try { await connection.end(); } catch { if (!failure) { failure = new SchemaExportError('DATABASE_QUERY_FAILED', '开发库 Schema 连接关闭失败'); } } } } if (failure) throw failure; return result; }