feat(GOV-012): 实现完整 SHOW CREATE 结构导出

This commit is contained in:
2026-08-18 02:02:41 +08:00
parent 7e9d9065bb
commit dc8ad7d4b6
2 changed files with 586 additions and 0 deletions
+246
View File
@@ -1,5 +1,8 @@
import { createHash } from 'node:crypto';
import { createRequire } from 'node:module'; import { createRequire } from 'node:module';
import mysql from 'mysql2/promise';
const require = createRequire(import.meta.url); const require = createRequire(import.meta.url);
const mysql2Version = require('mysql2/package.json').version; const mysql2Version = require('mysql2/package.json').version;
const DEVELOPMENT_DATABASE = 'a_share_analysis_dev'; const DEVELOPMENT_DATABASE = 'a_share_analysis_dev';
@@ -30,6 +33,21 @@ WHERE EVENT_SCHEMA = ?
ORDER BY EVENT_NAME`, 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({ export const MYSQL2_EXPORTER_METADATA = Object.freeze({
name: 'mysql2-show-create', name: 'mysql2-show-create',
version: '1.0.0', version: '1.0.0',
@@ -256,3 +274,231 @@ export function orderViewNames(catalog) {
} }
return ordered; return ordered;
} }
function sha256(content) {
return createHash('sha256').update(content, 'utf8').digest('hex');
}
/** 只屏蔽 SQL 字符串字面量,保留标识符参与 DEFINER 和凭据扫描。 */
function maskStringLiterals(sql) {
return String(sql).replace(/'(?:''|\\.|[^'\\])*'/gs, "''");
}
/**
* 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',
)
.replace(/\sAUTO_INCREMENT=\d+(?=\s(?:DEFAULT\s+)?CHARSET=)/i, '')
.trim()
.replace(/;+\s*$/, '');
const inspectable = maskStringLiterals(normalized);
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)
) {
throw new SchemaExportError('UNSAFE_DEFINITION', 'Schema 对象定义包含禁止内容');
}
return normalized;
}
export async function readMysql57ServerMetadata(
connection,
database,
limits = DEFAULT_EXPORT_LIMITS,
) {
const [rows] = await connection.query({
sql: SERVER_METADATA_QUERY,
values: [database],
timeout: limits.queryTimeoutMs,
});
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)}`;
const [rows] = await connection.query({ sql, timeout: limits.queryTimeoutMs });
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;
}
@@ -7,9 +7,14 @@ import {
SchemaExportError, SchemaExportError,
assertMysql57ServerMetadata, assertMysql57ServerMetadata,
buildConnectionOptions, buildConnectionOptions,
exportMysql57Schema,
normalizeCreateDefinition,
orderViewNames, orderViewNames,
quoteIdentifier, quoteIdentifier,
readDefinitions,
readMysql57ServerMetadata,
readSchemaCatalog, readSchemaCatalog,
renderSchemaSql,
} from './mysql2-schema-exporter.mjs'; } from './mysql2-schema-exporter.mjs';
const config = Object.freeze({ const config = Object.freeze({
@@ -226,3 +231,338 @@ test('目录拒绝未知表类型和超过对象上限', async () => {
(error) => error instanceof SchemaExportError && error.code === 'OBJECT_LIMIT_EXCEEDED', (error) => error instanceof SchemaExportError && error.code === 'OBJECT_LIMIT_EXCEEDED',
); );
}); });
const completeCatalog = Object.freeze({
tables: ['market_snapshot'],
views: ['market_dashboard', 'market_summary'],
viewDependencies: [
{
view: 'market_dashboard',
referencedSchema: 'a_share_analysis_dev',
referencedObject: 'market_summary',
},
{
view: 'market_summary',
referencedSchema: 'a_share_analysis_dev',
referencedObject: 'market_snapshot',
},
],
routines: [
{ name: 'calculate_breadth', type: 'FUNCTION' },
{ name: 'refresh_market_snapshot', type: 'PROCEDURE' },
],
triggers: [
{
name: 'snapshot_before_insert',
table: 'market_snapshot',
timing: 'BEFORE',
event: 'INSERT',
order: 1,
},
],
events: ['purge_market_snapshot'],
});
const createRows = Object.freeze({
'SHOW CREATE TABLE `a_share_analysis_dev`.`market_snapshot`': {
'Create Table':
'CREATE TABLE `market_snapshot` (`id` bigint NOT NULL AUTO_INCREMENT) ENGINE=InnoDB AUTO_INCREMENT=12 DEFAULT CHARSET=utf8mb4',
},
'SHOW CREATE VIEW `a_share_analysis_dev`.`market_summary`': {
'Create View':
'CREATE ALGORITHM=UNDEFINED DEFINER=`schema_user`@`192.168.1.10` SQL SECURITY DEFINER VIEW `market_summary` AS select `market_snapshot`.`id` AS `id` from `market_snapshot`',
},
'SHOW CREATE VIEW `a_share_analysis_dev`.`market_dashboard`': {
'Create View':
'CREATE ALGORITHM=UNDEFINED DEFINER=`schema_user`@`192.168.1.10` SQL SECURITY DEFINER VIEW `market_dashboard` AS select `market_summary`.`id` AS `id` from `market_summary`',
},
'SHOW CREATE FUNCTION `a_share_analysis_dev`.`calculate_breadth`': {
'Create Function':
"CREATE DEFINER=`schema_user`@`192.168.1.10` FUNCTION `calculate_breadth`() RETURNS int RETURN 1",
},
'SHOW CREATE PROCEDURE `a_share_analysis_dev`.`refresh_market_snapshot`': {
'Create Procedure':
'CREATE DEFINER=`schema_user`@`192.168.1.10` PROCEDURE `refresh_market_snapshot`()\nBEGIN\n UPDATE `market_snapshot` SET `id` = `id`;\nEND',
},
'SHOW CREATE TRIGGER `a_share_analysis_dev`.`snapshot_before_insert`': {
'SQL Original Statement':
'CREATE DEFINER=`schema_user`@`192.168.1.10` TRIGGER `snapshot_before_insert` BEFORE INSERT ON `market_snapshot` FOR EACH ROW SET NEW.id = NEW.id',
},
'SHOW CREATE EVENT `a_share_analysis_dev`.`purge_market_snapshot`': {
'Create Event':
'CREATE DEFINER=`schema_user`@`192.168.1.10` EVENT `purge_market_snapshot` ON SCHEDULE EVERY 1 DAY DO DELETE FROM `market_snapshot` WHERE `id` < 0',
},
});
function catalogRowsFor(sql) {
if (sql.includes('information_schema.TABLES')) {
return [
{ TABLE_NAME: 'market_snapshot', TABLE_TYPE: 'BASE TABLE' },
{ TABLE_NAME: 'market_dashboard', TABLE_TYPE: 'VIEW' },
{ TABLE_NAME: 'market_summary', TABLE_TYPE: 'VIEW' },
];
}
if (sql.includes('information_schema.VIEW_TABLE_USAGE')) {
return [
{
VIEW_NAME: 'market_dashboard',
REFERENCED_SCHEMA_NAME: 'a_share_analysis_dev',
REFERENCED_TABLE_NAME: 'market_summary',
},
{
VIEW_NAME: 'market_summary',
REFERENCED_SCHEMA_NAME: 'a_share_analysis_dev',
REFERENCED_TABLE_NAME: 'market_snapshot',
},
];
}
if (sql.includes('information_schema.ROUTINES')) {
return [
{ ROUTINE_NAME: 'calculate_breadth', ROUTINE_TYPE: 'FUNCTION' },
{ ROUTINE_NAME: 'refresh_market_snapshot', ROUTINE_TYPE: 'PROCEDURE' },
];
}
if (sql.includes('information_schema.TRIGGERS')) {
return [
{
TRIGGER_NAME: 'snapshot_before_insert',
EVENT_OBJECT_TABLE: 'market_snapshot',
ACTION_TIMING: 'BEFORE',
EVENT_MANIPULATION: 'INSERT',
ACTION_ORDER: 1,
},
];
}
if (sql.includes('information_schema.EVENTS')) {
return [{ EVENT_NAME: 'purge_market_snapshot' }];
}
return undefined;
}
function createExportConnection({ mutateCatalog, mutateDefinition, queryError } = {}) {
let catalogRound = 0;
let definitionRound = 0;
let tableCatalogCalls = 0;
let tableDefinitionCalls = 0;
const connection = {
endCalls: 0,
async query(options) {
if (queryError) throw queryError;
if (options.sql.includes('information_schema.SCHEMATA')) {
return [
[
{
SERVER_VERSION: '5.7.42-log',
DATABASE_CHARSET: 'utf8mb4',
DATABASE_COLLATION: 'utf8mb4_general_ci',
},
],
[],
];
}
const catalogRows = catalogRowsFor(options.sql);
if (catalogRows) {
if (options.sql.includes('information_schema.TABLES')) {
tableCatalogCalls += 1;
catalogRound = tableCatalogCalls > 1 ? 1 : 0;
}
const rows = mutateCatalog?.(options.sql, catalogRows, catalogRound) ?? catalogRows;
return [rows, []];
}
const row = createRows[options.sql];
assert.ok(row, `未识别的 SHOW CREATE:${options.sql}`);
if (options.sql.includes('SHOW CREATE TABLE')) {
tableDefinitionCalls += 1;
definitionRound = tableDefinitionCalls > 1 ? 1 : 0;
}
const changed = mutateDefinition?.(options.sql, row, definitionRound);
return [changed === undefined ? [row] : changed, []];
},
async end() {
connection.endCalls += 1;
},
};
return connection;
}
test('读取并校验固定开发库服务端元数据', async () => {
const connection = createExportConnection();
assert.deepEqual(
await readMysql57ServerMetadata(connection, 'a_share_analysis_dev', DEFAULT_EXPORT_LIMITS),
{
serverVersion: '5.7.42-log',
databaseCharset: 'utf8mb4',
databaseCollation: 'utf8mb4_general_ci',
},
);
});
test('SHOW CREATE 覆盖全部对象并按可恢复顺序渲染', async () => {
const connection = createExportConnection();
const definitions = await readDefinitions(connection, completeCatalog, DEFAULT_EXPORT_LIMITS);
const sql = renderSchemaSql(completeCatalog, definitions);
const order = [
'CREATE TABLE `market_snapshot`',
'FUNCTION `calculate_breadth`',
'PROCEDURE `refresh_market_snapshot`',
'VIEW `market_summary`',
'VIEW `market_dashboard`',
'TRIGGER `snapshot_before_insert`',
'EVENT `purge_market_snapshot`',
].map((fragment) => sql.indexOf(fragment));
assert.ok(order.every((index) => index >= 0));
assert.deepEqual(order, [...order].sort((left, right) => left - right));
assert.match(sql, /DELIMITER ;;\nCREATE PROCEDURE[\s\S]+END ;;\nDELIMITER ;/);
assert.doesNotMatch(sql, /DEFINER\s*=/i);
assert.doesNotMatch(sql, /AUTO_INCREMENT=12/);
assert.match(sql, /`id` bigint NOT NULL AUTO_INCREMENT/);
assert.match(sql, /UPDATE `market_snapshot`/);
});
test('定义规范化只移除账号绑定和当前自增值', () => {
const routine = normalizeCreateDefinition(
"CREATE DEFINER=`user`@`host` PROCEDURE `sample`() SELECT 'DEFINER=`kept`@`text` AUTO_INCREMENT=99'",
);
const table = normalizeCreateDefinition(
'CREATE TABLE `sample` (`id` bigint AUTO_INCREMENT) ENGINE=InnoDB AUTO_INCREMENT=99 DEFAULT CHARSET=utf8mb4;',
);
assert.equal(
routine,
"CREATE PROCEDURE `sample`() SELECT 'DEFINER=`kept`@`text` AUTO_INCREMENT=99'",
);
assert.equal(
table,
'CREATE TABLE `sample` (`id` bigint AUTO_INCREMENT) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4',
);
});
test('完整导出双重读取一致并始终关闭连接', async () => {
const connection = createExportConnection();
let receivedOptions;
const result = await exportMysql57Schema({
config,
createConnection: async (options) => {
receivedOptions = options;
return connection;
},
now: () => 100,
});
assert.equal(receivedOptions.multipleStatements, false);
assert.equal(connection.endCalls, 1);
assert.deepEqual(result.objectCounts, {
tables: 1,
views: 2,
triggers: 1,
routines: 2,
events: 1,
});
assert.equal(result.durationMs, 0);
assert.equal(result.metadata.driverName, 'mysql2');
});
test('导出拒绝目录变化和定义变化', async () => {
const catalogConnection = createExportConnection({
mutateCatalog(sql, rows, round) {
if (round === 1 && sql.includes('information_schema.TABLES')) {
return [...rows, { TABLE_NAME: 'new_table', TABLE_TYPE: 'BASE TABLE' }];
}
return rows;
},
});
await assert.rejects(
() =>
exportMysql57Schema({
config,
createConnection: async () => catalogConnection,
now: () => 100,
}),
(error) => error.code === 'CATALOG_CHANGED_DURING_EXPORT',
);
assert.equal(catalogConnection.endCalls, 1);
const definitionConnection = createExportConnection({
mutateDefinition(sql, row, round) {
if (round === 1 && sql.includes('SHOW CREATE TABLE')) {
return [{ ...row, 'Create Table': `${row['Create Table']} COMMENT='changed'` }];
}
return undefined;
},
});
await assert.rejects(
() =>
exportMysql57Schema({
config,
createConnection: async () => definitionConnection,
now: () => 100,
}),
(error) => error.code === 'DEFINITION_CHANGED_DURING_EXPORT',
);
assert.equal(definitionConnection.endCalls, 1);
});
test('SHOW CREATE 拒绝空结果、错误形状和过大定义', async () => {
for (const [changed, expectedCode] of [
[[], 'SHOW_CREATE_EMPTY'],
[[{ unexpected: 'value' }], 'SHOW_CREATE_SHAPE_INVALID'],
]) {
const connection = createExportConnection({
mutateDefinition(sql) {
return sql.includes('SHOW CREATE TABLE') ? changed : undefined;
},
});
await assert.rejects(
() => readDefinitions(connection, completeCatalog, DEFAULT_EXPORT_LIMITS),
(error) => error.code === expectedCode,
);
}
const oversized = createExportConnection({
mutateDefinition(sql, row) {
return sql.includes('SHOW CREATE TABLE')
? [{ ...row, 'Create Table': 'x'.repeat(100) }]
: undefined;
},
});
await assert.rejects(
() =>
readDefinitions(oversized, completeCatalog, {
...DEFAULT_EXPORT_LIMITS,
maxDefinitionBytes: 32,
}),
(error) => error.code === 'DEFINITION_LIMIT_EXCEEDED',
);
});
test('查询失败和总超时只返回脱敏错误并关闭连接', async () => {
const failed = createExportConnection({
queryError: new Error('connect 127.0.0.1 password=not-for-logs'),
});
await assert.rejects(
() =>
exportMysql57Schema({
config,
createConnection: async () => failed,
now: () => 100,
}),
(error) =>
error.code === 'DATABASE_QUERY_FAILED' &&
!error.message.includes('127.0.0.1') &&
!error.message.includes('not-for-logs'),
);
assert.equal(failed.endCalls, 1);
const timedOut = createExportConnection();
let tick = 0;
await assert.rejects(
() =>
exportMysql57Schema({
config,
createConnection: async () => timedOut,
now: () => (tick++ === 0 ? 0 : 30_001),
}),
(error) => error.code === 'EXPORT_TIMEOUT',
);
assert.equal(timedOut.endCalls, 1);
});