/**
* MySQL Security - Audit and Firewall Tools
*
* Tools for security auditing, firewall monitoring, and compliance.
*/
import { z } from "zod";
import type { MySQLAdapter } from "../../MySQLAdapter.js";
import type {
ToolDefinition,
RequestContext,
} from "../../../../types/index.js";
// =============================================================================
// Zod Schemas
// ============================================================================
const AuditLogSchema = z.object({
limit: z.number().default(100).describe("Maximum number of records"),
user: z.string().optional().describe("Filter by username"),
eventType: z
.string()
.optional()
.describe('Filter by event type (e.g., "CONNECT", "QUERY")'),
startTime: z.string().optional().describe("Start time filter (ISO 8601)"),
});
const FirewallRulesSchema = z.object({
user: z.string().optional().describe("Filter by username"),
mode: z
.enum(["RECORDING", "PROTECTING", "DETECTING", "OFF"])
.optional()
.describe("Filter by mode"),
});
// =============================================================================
// Tool Creation Functions
// =============================================================================
/**
* Query audit log
*/
export function createSecurityAuditTool(adapter: MySQLAdapter): ToolDefinition {
return {
name: "mysql_security_audit",
title: "MySQL Security Audit Log",
description:
"Query the MySQL audit log (requires Enterprise Audit or compatible plugin).",
group: "security",
inputSchema: AuditLogSchema,
requiredScopes: ["admin"],
annotations: {
readOnlyHint: true,
idempotentHint: true,
},
handler: async (params: unknown, _context: RequestContext) => {
const { limit, user, eventType, startTime } =
AuditLogSchema.parse(params);
// First check if audit log table exists
try {
const checkResult = await adapter.executeQuery(`
SELECT TABLE_NAME
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'mysql'
AND TABLE_NAME = 'audit_log'
`);
if (!checkResult.rows || checkResult.rows.length === 0) {
// Try performance_schema alternative
let query = `
SELECT
EVENT_NAME as event,
OBJECT_TYPE as objectType,
OBJECT_NAME as objectName,
CURRENT_USER as user,
HOST as host,
TIMER_START as startTime
FROM performance_schema.events_statements_history
`;
const conditions: string[] = [];
const queryParams: unknown[] = [];
if (user) {
conditions.push("CURRENT_USER LIKE ?");
queryParams.push(`%${user}%`);
}
if (conditions.length > 0) {
query += " WHERE " + conditions.join(" AND ");
}
query += ` ORDER BY TIMER_START DESC LIMIT ${String(limit)}`;
const result = await adapter.executeQuery(query, queryParams);
return {
source: "performance_schema",
message: "Using performance_schema as audit log is not available",
events: result.rows ?? [],
count: result.rows?.length ?? 0,
};
}
// Query actual audit log
let query = `
SELECT *
FROM mysql.audit_log
`;
const conditions: string[] = [];
const queryParams: unknown[] = [];
if (user) {
conditions.push("user LIKE ?");
queryParams.push(`%${user}%`);
}
if (eventType) {
conditions.push("event_type = ?");
queryParams.push(eventType);
}
if (startTime) {
conditions.push("timestamp >= ?");
queryParams.push(startTime);
}
if (conditions.length > 0) {
query += " WHERE " + conditions.join(" AND ");
}
query += ` ORDER BY timestamp DESC LIMIT ${String(limit)}`;
const result = await adapter.executeQuery(query, queryParams);
return {
source: "mysql.audit_log",
events: result.rows ?? [],
count: result.rows?.length ?? 0,
};
} catch {
return {
available: false,
message:
"Audit logging is not enabled. Install MySQL Enterprise Audit or Percona Audit plugin.",
suggestion:
'Install audit plugin with: INSTALL PLUGIN audit_log SONAME "audit_log.so"',
};
}
},
};
}
/**
* Get firewall status
*/
export function createSecurityFirewallStatusTool(
adapter: MySQLAdapter,
): ToolDefinition {
return {
name: "mysql_security_firewall_status",
title: "MySQL Firewall Status",
description: "Get MySQL Enterprise Firewall plugin status.",
group: "security",
inputSchema: z.object({}),
requiredScopes: ["read"],
annotations: {
readOnlyHint: true,
idempotentHint: true,
},
handler: async (_params: unknown, _context: RequestContext) => {
try {
// Check if firewall plugin is installed
const pluginResult = await adapter.executeQuery(`
SELECT PLUGIN_NAME, PLUGIN_STATUS
FROM information_schema.PLUGINS
WHERE PLUGIN_NAME LIKE '%firewall%'
`);
if (!pluginResult.rows || pluginResult.rows.length === 0) {
return {
installed: false,
message: "MySQL Enterprise Firewall is not installed",
suggestion:
'Install with: INSTALL PLUGIN mysql_firewall SONAME "firewall.so"',
};
}
// Get firewall variables
const varsResult = await adapter.executeQuery(
"SHOW VARIABLES LIKE 'mysql_firewall%'",
);
const variables: Record<string, unknown> = Object.fromEntries(
(varsResult.rows ?? []).map((row) => {
const r = row;
const varName =
typeof r["Variable_name"] === "string" ? r["Variable_name"] : "";
return [varName, r["Value"]];
}),
);
return {
installed: true,
plugins: pluginResult.rows,
configuration: variables,
};
} catch {
return {
installed: false,
message: "Firewall plugin check failed",
suggestion: "Ensure you have privileges to view plugin information",
};
}
},
};
}
/**
* List firewall rules
*/
export function createSecurityFirewallRulesTool(
adapter: MySQLAdapter,
): ToolDefinition {
return {
name: "mysql_security_firewall_rules",
title: "MySQL Firewall Rules",
description: "List MySQL Enterprise Firewall allowlist rules.",
group: "security",
inputSchema: FirewallRulesSchema,
requiredScopes: ["admin"],
annotations: {
readOnlyHint: true,
idempotentHint: true,
},
handler: async (params: unknown, _context: RequestContext) => {
const { user, mode } = FirewallRulesSchema.parse(params);
try {
// Get firewall users
let usersQuery = `
SELECT USERHOST, MODE
FROM mysql.firewall_users
`;
const conditions: string[] = [];
const queryParams: unknown[] = [];
if (user) {
conditions.push("USERHOST LIKE ?");
queryParams.push(`%${user}%`);
}
if (mode) {
conditions.push("MODE = ?");
queryParams.push(mode);
}
if (conditions.length > 0) {
usersQuery += " WHERE " + conditions.join(" AND ");
}
const usersResult = await adapter.executeQuery(usersQuery, queryParams);
// Get firewall whitelist
let rulesQuery = `
SELECT USERHOST, RULE
FROM mysql.firewall_whitelist
`;
if (user) {
rulesQuery += " WHERE USERHOST LIKE ?";
}
const rulesResult = await adapter.executeQuery(
rulesQuery,
user ? [`%${user}%`] : [],
);
return {
users: usersResult.rows ?? [],
rules: rulesResult.rows ?? [],
userCount: usersResult.rows?.length ?? 0,
ruleCount: rulesResult.rows?.length ?? 0,
};
} catch {
return {
available: false,
message:
"Firewall tables not accessible. Ensure MySQL Enterprise Firewall is installed and you have appropriate privileges.",
};
}
},
};
}
