某公司MySQL数据库遭黑客入侵数据被勒索软件加密后MySQL安全加固操作指南与防护方案
那天早上八点，公司的运维团队冲进我的办公室时，脸都绿了。监控屏幕上，所有的数据文件后缀都被改成了乱码，MySQL数据库报了一堆”I/O error”。财务总监老张冲进来，声音都在发抖：”昨天还能打开的客户订单库，现在全打不开了……”
我盯着那串加密后缀看了三秒，心里基本有数了。这是典型的勒索软件攻击，而且瞄准的就是我们的MySQL数据文件。
但说实话，这不是天降横祸，这是我们自己种下的因。
一、那场噩梦：从发现到复盘
1.1 事件发生的完整时间线
| 时间 | 事件 |
|---|---|
| 3月15日 02:13 | 防火墙日志记录到异常IP 45.92.13.47 尝试SSH暴力破解，失败38次后突破 |
| 3月15日 02:15 | 攻击者利用CVE-2019-2725（MySQL远程代码执行漏洞）获取服务器权限 |
| 3月15日 02:22 | 攻击者在Web目录下植入勒索软件变种LockBit 3.0 |
| 3月15日 02:25 - 03:41 | 勒索软件逐步加密所有.ibd和.frm文件 |
| 3月15日 03:45 | 数据库服务完全不可用，业务部门开始投诉 |
| 3月15日 08:00 | 运维团队发现异常，我介入处理 |
1.2 我们踩过的所有坑
说实话，复盘那次事故的时候，我恨不得穿越回去扇自己几巴掌。问题堆起来能堆成山：
问题一：MySQL root账号用”root/root123”这种密码
-- 事故前的真实状态（已脱敏）
mysql> SELECT user, host, password_expired FROM mysql.user;
+------+-----------+--------------------+
| user | host | password_expired |
+------+-----------+--------------------+
| root | localhost | N |
| root | 127.0.0.1 | N |
| app_user | % | N | -- 这个%太恐怖了，任何人都能连
+------+-----------+--------------------+
-- app_user的密码是什么？"password123"
-- 连攻击者的扫描脚本都不用，网上随便一个MySQL默认密码字典都能撞出来
问题二：MySQL直接暴露在公网
我们的MySQL端口3306，直接绑在公网IP上。没有VPN，没有跳板机，谁都能扫到。公司安全负责人说”方便开发测试”，我把这话记在小本本上了。
问题三：备份？有是有，但从未验证过
# 每天的备份脚本写得漂漂亮亮
0 2 * * * /usr/bin/mysqldump -u root -p'root123' --all-databases > /backup/full_$(date +%Y%m%d).sql
# 但备份文件呢？
$ ls -lh /backup/
total 4.2G
-rw-r--r-- 1 mysql mysql 4.2G Mar 1 02:00 full_20240301.sql # 3月1日的
-rw-r--r-- 1 mysql mysql 4.1G Mar 2 02:00 full_20240302.sql # 3月2日的
# ...
-rw-r--r-- 1 mysql mysql 0 Mar 15 02:00 full_20240315.sql # 3月15日的！是空的！
# 因为那天凌晨2点，MySQL已经在被加密了，dump自然是空文件
更讽刺的是，这些备份文件存在同一个磁盘上。勒索软件加密数据库文件的同时，也顺手把备份目录给清了。
问题四：补丁？不存在的
公司的服务器跑的是MySQL 5.7.12，那是2016年8月的版本。中间多少个CVE漏洞，我们一个都没打。
# 我们线上跑的
$ mysql --version
mysql Ver 14.14 Distrib 5.7.12, for linux-glibc2.5 (x86_64) using EditLine wrapper
# 同时间段最新稳定版
$ mysql --version
mysql Ver 14.14 Distrib 8.0.35, for linux-glibc2.6 (x86_64) using EditLine wrapper
# 中间隔了将近8年，几百个安全补丁...
问题五：权限给了太多了
-- app_user这个普通应用账号，居然有这些权限
mysql> SHOW GRANTS FOR 'app_user'@'%';
+---------------------------------------------------------------------+
| Grants for app_user@% |
+---------------------------------------------------------------------+
| GRANT ALL PRIVILEGES ON *.* TO 'app_user'@'%' |
| WITH GRANT OPTION |
| GRANT PROCESS ON *.* TO 'app_user'@'%' |
| GRANT FILE ON *.* TO 'app_user'@'%' |
| GRANT SUPER ON *.* TO 'app_user'@'%' |
+---------------------------------------------------------------------+
-- FILE权限能读服务器任意文件，PROCESS能看所有进程，SUPER能重启服务
-- 一个写业务的账号，这权限比DBA还大
二、亡羊补牢：MySQL安全加固完整指南
事件发生后的72小时，我们一边恢复数据，一边疯狂补课。下面是我结合这次事故总结出来的完整加固方案，你直接照做就行。
2.1 第一层：身份认证加固
2.1.1 禁止使用root登录应用
-- 创建专用应用账号，权限精确到最小
CREATE USER 'app_read'@'192.168.1.%' IDENTIFIED BY 'X#9$kL2@mP5&nQ8';
CREATE USER 'app_write'@'192.168.1.%' IDENTIFIED BY 'W@7$hN4#pR1&tY6';
CREATE USER 'app_admin'@'192.168.1.%' IDENTIFIED BY 'A%3$jK8@mL5&nQ2';
-- 严格分配权限
GRANT SELECT ON company_db.* TO 'app_read'@'192.168.1.%';
GRANT SELECT, INSERT, UPDATE ON company_db.* TO 'app_write'@'192.168.1.%';
GRANT SELECT, INSERT, UPDATE, DELETE ON company_db.orders TO 'app_write'@'192.168.1.%';
-- 只有管理账号才能管理表结构
GRANT ALTER, CREATE, DROP, INDEX ON company_db.* TO 'app_admin'@'192.168.1.%';
-- 回收所有多余的权限
REVOKE FILE ON *.* FROM 'app_write'@'192.168.1.%';
REVOKE PROCESS ON *.* FROM 'app_write'@'192.168.1.%';
REVOKE SUPER ON *.* FROM 'app_write'@'192.168.1.%';
-- 检查确认
SHOW GRANTS FOR 'app_write'@'192.168.1.%';
2.1.2 强密码策略
-- 启用密码强度验证插件（MySQL 5.7+）
-- 先加载插件
INSTALL PLUGIN VALIDATE_PASSWORD COMPONENT 'https://dev.mysql.com/get/plugin/mysql_component_validate_password.so';
-- 或者在my.cnf中配置
-- 编辑my.cnf
# /etc/mysql/my.cnf 或 /etc/my.cnf
[mysqld]
# 密码强度要求
validate_password.policy=STRONG
validate_password.length=16
validate_password.mixed_case_count=1
validate_password.number_count=1
validate_password.special_char_count=1
validate_password.check_user_name=ON
# 重载MySQL配置
sudo systemctl reload mysql
# 验证插件是否加载成功
mysql> SELECT PLUGIN_NAME, PLUGIN_STATUS FROM information_schema.plugins WHERE PLUGIN_NAME LIKE '%validate%';
+--------------------------+---------------+
| PLUGIN_NAME | PLUGIN_STATUS |
+--------------------------+---------------+
| validate_password | ACTIVE |
+--------------------------+---------------+
现在你试试用弱密码创建账号，MySQL会直接拒绝：
-- 这个会失败
mysql> CREATE USER 'test'@'localhost' IDENTIFIED BY 'password123';
ERROR 1819 (HY000): Your password does not satisfy the current policy requirements
-- 这个才行
mysql> CREATE USER 'test'@'localhost' IDENTIFIED BY 'T3st!Pass#9xK2';
Query OK, 0 rows affected (0.01 sec)
2.1.3 限制登录失败次数
-- 安装账号锁定插件（MySQL 8.0+）
INSTALL PLUGIN connection_control SONAME 'connection_control.so';
INSTALL PLUGIN connection_control_failed_login_attempts SONAME 'connection_control.so';
-- 配置my.cnf
[mysqld]
# 连续5次失败后锁定30分钟
connection_control.failed_login_threshold=5
connection_control.min_connection_delay=1800000
sudo systemctl reload mysql
2.2 第二层：访问控制加固
2.2.1 绑定到内网IP
# my.cnf中，不要把MySQL绑到0.0.0.0
[mysqld]
bind-address = 192.168.1.100
# 或者只绑定到localhost（如果应用在同一台服务器）
bind-address = 127.0.0.1
# 验证监听状态
sudo netstat -tlnp | grep mysql
# 正常应该是：
# tcp 0 0 192.168.1.100:3306 0.0.0.0:* LISTEN 1234/mysql
# 而不是：
# tcp 0 0 0.0.0.0:3306 0.0.0.0:* LISTEN 1234/mysql
2.2.2 防火墙白名单
# 使用firewalld
sudo firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="192.168.1.0/24" port port="3306" protocol="tcp" accept'
sudo firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="192.168.2.50" port port="3306" protocol="tcp" accept' # 跳板机
sudo firewall-cmd --reload
# 或者用iptables
sudo iptables -A INPUT -p tcp -s 192.168.1.0/24 --dport 3306 -j ACCEPT
sudo iptables -A INPUT -p tcp -s 192.168.2.50/32 --dport 3306 -j ACCEPT
sudo iptables -A INPUT -p tcp --dport 3306 -j DROP
sudo iptables-save > /etc/iptables/rules.v4
2.2.3 SSH跳板机方案
# 在跳板机上创建密钥对（如果还没有）
ssh-keygen -t ed25519 -C "mysql-admin"
# 把公钥放到MySQL服务器
ssh-copy-id -i ~/.ssh/id_ed25519.pub root@192.168.1.100
# 然后从跳板机连接MySQL
ssh -L 3307:127.0.0.1:3306 root@192.168.1.100
# 本地连接跳板机
mysql -h 127.0.0.1 -P 3307 -u app_write -p
# MySQL服务器上禁止root远程登录
[mysqld]
skip-networking=OFF
# 同时修改mysql.user表
mysql> UPDATE mysql.user SET host='127.0.0.1' WHERE user='root' AND host='%';
mysql> FLUSH PRIVILEGES;
2.3 第三层：传输加密
2.3.1 启用SSL/TLS
# 生成自签名证书（生产环境建议用CA签发的证书）
sudo mysql_ssl_rsa_setup --datadir=/var/lib/mysql
# MySQL 8.0+
sudo mysqld --initialize --ssl
# my.cnf配置
[mysqld]
ssl-ca=/var/lib/mysql/ca.pem
ssl-cert=/var/lib/mysql/server-cert.pem
ssl-key=/var/lib/mysql/server-key.pem
require_secure_transport=ON
[client]
ssl-ca=/var/lib/mysql/ca.pem
ssl-cert=/var/lib/mysql/client-cert.pem
ssl-key=/var/lib/mysql/client-key.pem
sudo systemctl restart mysql
-- 验证SSL状态
mysql> SHOW STATUS LIKE 'Ssl_cipher';
+---------------+--------------------+
| Variable_name | Value |
+---------------+--------------------+
| Ssl_cipher | TLS_AES_256_GCM_SHA384 |
+---------------+--------------------+
-- 要求特定账号必须用SSL连接
ALTER USER 'app_write'@'192.168.1.%' REQUIRE SSL;
-- 更严格：要求特定加密算法
ALTER USER 'app_write'@'192.168.1.%' REQUIRE CIPHER 'TLS_AES_256_GCM_SHA384';
2.3.2 使用持久SSL连接
-- 配置连接字符串（Java应用示例）
jdbc:mysql://192.168.1.100:3306/company_db?
sslmode=REQUIRED&
ssl_ca=/path/to/ca.pem&
ssl_cert=/path/to/client-cert.pem&
ssl_key=/path/to/client-key.pem&
requireSSL=true&
verifyServerCertificate=true
# Python应用示例
import pymysql
connection = pymysql.connect(
host='192.168.1.100',
port=3306,
user='app_write',
password='W@7$hN4#pR1&tY6',
database='company_db',
ssl={
'ca': '/path/to/ca.pem',
'cert': '/path/to/client-cert.pem',
'key': '/path/to/client-key.pem'
},
ssl_verify_identity=True
)
2.4 第四层：审计与监控
2.4.1 启用通用日志审计
# my.cnf
[mysqld]
general_log = ON
general_log_file = /var/log/mysql/general.log
log_output = FILE
# 创建日志目录并设置权限
sudo mkdir -p /var/log/mysql
sudo chown mysql:mysql /var/log/mysql
sudo chmod 750 /var/log/mysql
sudo systemctl restart mysql
# 监控可疑查询
tail -f /var/log/mysql/general.log | grep -E "(DROP|DELETE|TRUNCATE|GRANT|REVOKE|UPDATE mysql)"
2.4.2 开启专用审计插件
-- MySQL Enterprise Audit（企业版）或 使用开源替代
-- 安装MariaDB审计插件（开源，兼容MySQL）
INSTALL PLUGIN audit_log SONAME 'audit_log.so';
-- 或者用MySQL 8.0的audit插件
INSTALL COMPONENT 'file:///component_audit_log';
# my.cnf
[mysqld]
# MariaDB审计配置示例
audit_log_policy=ALL
audit_log_format=JSON
audit_log_file=/var/log/mysql/audit.log
audit_log_rotate_on_size=104857600 # 100MB
2.4.3 实时告警脚本
#!/usr/bin/env python3
"""MySQL异常行为实时监测脚本"""
import pymysql
import json
import logging
from datetime import datetime, timedelta
import requests
# 配置日志
logging.basicConfig(
filename='/var/log/mysql_monitor.log',
level=logging.INFO,
format='%(asctime)s - %(message)s'
)
# 告警配置
ALERT_CONFIG = {
'max_login_failures': 5,
'lockout_duration_minutes': 30,
'alert_webhook': 'https://hooks.slack.com/services/xxx/yyy/zzz',
'alert_email': 'security@company.com'
}
def check_abnormal_queries():
"""检测异常SQL操作"""
conn = pymysql.connect(
host='192.168.1.100',
user='app_admin',
password='A%3$jK8@mL5&nQ2',
database='mysql'
)
cursor = conn.cursor()
# 检测大表删除操作
cursor.execute("""
SELECT event_time, user_host, argument
FROM mysql.general_log
WHERE command_type = 'Query'
AND argument LIKE '%DROP TABLE%'
AND event_time > NOW() - INTERVAL 1 HOUR
ORDER BY event_time DESC
""")
suspicious_queries = cursor.fetchall()
if suspicious_queries:
alert_message = f"检测到异常DROP TABLE操作:\n"
for query in suspicious_queries:
alert_message += f" - {query[0]}: {query[2][:200]}\n"
send_alert(alert_message)
logging.warning(alert_message)
# 检测权限变更操作
cursor.execute("""
SELECT event_time, user_host, argument
FROM mysql.general_log
WHERE command_type = 'Query'
AND (argument LIKE '%GRANT%' OR argument LIKE '%REVOKE%' OR argument LIKE '%CREATE USER%')
AND event_time > NOW() - INTERVAL 1 HOUR
ORDER BY event_time DESC
""")
privilege_changes = cursor.fetchall()
if privilege_changes:
alert_message = "检测到权限变更操作:\n"
for change in privilege_changes:
alert_message += f" - {change[0]} by {change[1]}: {change[2]}\n"
send_alert(alert_message)
logging.warning(alert_message)
cursor.close()
conn.close()
def check_failed_logins():
"""检测登录失败"""
conn = pymysql.connect(
host='192.168.1.100',
user='app_admin',
password='A%3$jK8@mL5&nQ2',
database='mysql'
)
cursor = conn.cursor()
cursor.execute("""
SELECT user, host, COUNT(*) as fail_count
FROM mysql.general_log
WHERE command_type = 'Connect'
AND argument LIKE '%Access denied%'
AND event_time > NOW() - INTERVAL 10 MINUTES
GROUP BY user, host
HAVING fail_count >= 5
ORDER BY fail_count DESC
""")
results = cursor.fetchall()
if results:
alert_message = "检测到暴力破解行为:\n"
for row in results:
alert_message += f" - 用户: {row[0]}, IP: {row[1]}, 失败次数: {row[2]}\n"
send_alert(alert_message)
logging.warning(alert_message)
# 自动锁定
for row in results:
lock_user(row[0], row[1])
cursor.close()
conn.close()
def lock_user(user, host):
"""自动锁定异常账号"""
conn = pymysql.connect(
host='192.168.1.100',
user='app_admin',
password='A%3$jK8@mL5&nQ2',
database='mysql'
)
cursor = conn.cursor()
# 锁定账号
cursor.execute(f"ALTER USER '{user}'@'{host}' ACCOUNT LOCK")
conn.commit()
cursor.close()
conn.close()
logging.info(f"已自动锁定账号: {user}@{host}")
def send_alert(message):
"""发送告警"""
# Slack告警
try:
payload = {'text': message}
requests.post(ALERT_CONFIG['alert_webhook'], json=payload, timeout=10)
except Exception as e:
logging.error(f"发送Slack告警失败: {e}")
# 邮件告警
try:
import smtplib
from email.mime.text import MIMEText
msg = MIMEText(message)
msg['Subject'] = f"🚨 MySQL安全告警 - {datetime.now().strftime('%Y-%m-%d %H:%M')}"
msg['From'] = 'mysql-monitor@company.com'
msg['To'] = ALERT_CONFIG['alert_email']
with smtplib.SMTP('smtp.company.com', 587) as server:
server.starttls()
server.login('mysql-monitor@company.com', 'your_password')
server.send_message(msg)
except Exception as e:
logging.error(f"发送邮件告警失败: {e}")
if __name__ == '__main__':
import schedule
import time
schedule.every(5).minutes.do(check_abnormal_queries)
schedule.every(2).minutes.do(check_failed_logins)
logging.info("MySQL安全监控服务启动")
while True:
schedule.run_pending()
time.sleep(1)
# 设置为系统服务
sudo tee /etc/systemd/system/mysql-monitor.service << 'EOF'
[Unit]
Description=MySQL Security Monitor
After=network.target mysql.service
[Service]
Type=simple
User=root
ExecStart=/usr/bin/python3 /opt/mysql-monitor/monitor.py
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
EOF
sudo systemctl daemon-reload
sudo systemctl enable mysql-monitor
sudo systemctl start mysql-monitor
2.5 第五层：数据备份与恢复策略
2.5.1 多重备份方案
#!/bin/bash
# /opt/mysql-backup/incremental_backup.sh
# 增量备份脚本（结合binlog）
set -euo pipefail
BACKUP_DIR="/data/backups/mysql"
BINLOG_DIR="/var/lib/mysql"
DATE=$(date +%Y%m%d_%H%M%S)
RETENTION_DAYS=30
# 创建备份目录
mkdir -p ${BACKUP_DIR}/full/${DATE}
mkdir -p ${BACKUP_DIR}/incremental/${DATE}
mkdir -p ${BACKUP_DIR}/binlog/${DATE}
# 全量备份（每周日执行）
FULL_BACKUP_DAYS=$(($(date +%u)))
if [ $FULL_BACKUP_DAYS -eq 7 ]; then
echo "[$(date)] 开始全量备份..."
mysqldump --single-transaction \
--routines --triggers \
--master-data=2 \
--all-databases \
-u backup_user -p'Y0ur$ecureP@ssw0rd' \
| gzip > ${BACKUP_DIR}/full/${DATE}/full_backup.sql.gz
# 记录binlog位置
mysql -u backup_user -p'Y0ur$ecureP@ssw0rd' -e "SHOW MASTER STATUS\G" \
> ${BACKUP_DIR}/full/${DATE}/master_status.txt
fi
# 增量备份（每天执行）
echo "[$(date)] 开始增量备份..."
mysqldump --single-transaction \
--routines --triggers \
--all-databases \
-u backup_user -p'Y0ur$ecureP@ssw0rd' \
| gzip > ${BACKUP_DIR}/incremental/${DATE}/incremental_backup.sql.gz
# 复制binlog
cp ${BINLOG_DIR}/mysql-bin.* ${BACKUP_DIR}/binlog/${DATE}/ 2>/dev/null || true
# 清理旧备份
find ${BACKUP_DIR} -type d -mtime +${RETENTION_DAYS} -exec rm -rf {} \;
echo "[$(date)] 备份完成"
# 创建备份用户
CREATE USER 'backup_user'@'192.168.1.%' IDENTIFIED BY 'Y0ur$ecureP@ssw0rd';
GRANT SELECT, SHOW DATABASES, LOCK TABLES, RELOAD, PROCESS, SHOW VIEW, EVENT, TRIGGER ON *.* TO 'backup_user'@'192.168.1.%';
FLUSH PRIVILEGES;
2.5.2 异地备份
#!/bin/bash
# /opt/mysql-backup/offsite_sync.sh
# 异地备份同步脚本
REMOTE_HOST="backup.company.com"
REMOTE_USER="backup"
REMOTE_PATH="/data/mysql-backups"
LOCAL_PATH="/data/backups/mysql"
# 使用rsync加密传输
rsync -avz --progress \
-e "ssh -o StrictHostKeyChecking=no" \
${LOCAL_PATH}/ \
${REMOTE_USER}@${REMOTE_HOST}:${REMOTE_PATH}/
# 验证同步结果
REMOTE_COUNT=$(ssh ${REMOTE_USER}@${REMOTE_HOST} "find ${REMOTE_PATH} -name '*.sql.gz' | wc -l")
LOCAL_COUNT=$(find ${LOCAL_PATH} -name '*.sql.gz' | wc -l)
if [ "${REMOTE_COUNT}" -ne "${LOCAL_COUNT}" ]; then
echo "备份同步失败！本地: ${LOCAL_COUNT}, 远程: ${REMOTE_COUNT}"
exit 1
fi
echo "异地备份同步完成: ${REMOTE_COUNT} 个备份文件"
# 配置SSH免密登录
ssh-keygen -t ed25519 -C "mysql-backup-sync"
ssh-copy-id backup@backup.company.com
# 设置cron定时任务
0 4 * * * /opt/mysql-backup/offsite_sync.sh >> /var/log/mysql-backup-sync.log 2>&1
2.5.3 备份验证脚本
#!/bin/bash
# /opt/mysql-backup/verify.sh
# 备份完整性验证
BACKUP_DIR="/data/backups/mysql"
LOG_FILE="/var/log/mysql-backup-verify.log"
echo "[$(date)] 开始备份验证..." >> ${LOG_FILE}
# 检查最新备份文件是否存在且非空
LATEST_BACKUP=$(find ${BACKUP_DIR} -name "*.sql.gz" -newermt "1 day ago" | head -1)
if [ -z "${LATEST_BACKUP}" ]; then
echo "[$(date)] ERROR: 找不到最近的备份文件!" >> ${LOG_FILE}
exit 1
fi
if [ ! -s "${LATEST_BACKUP}" ]; then
echo "[$(date)] ERROR: 备份文件为空: ${LATEST_BACKUP}" >> ${LOG_FILE}
exit 1
fi
echo "[$(date)] 备份文件: ${LATEST_BACKUP}" >> ${LOG_FILE}
echo "[$(date)] 文件大小: $(ls -lh ${LATEST_BACKUP} | awk '{print $5}')" >> ${LOG_FILE}
# 验证压缩文件完整性
if gzip -t "${LATEST_BACKUP}" 2>/dev/null; then
echo "[$(date)] 压缩文件完整性验证通过" >> ${LOG_FILE}
else
echo "[$(date)] ERROR: 压缩文件损坏!" >> ${LOG_FILE}
exit 1
fi
# 验证SQL文件可导入（不实际导入，只检查语法）
TEMP_DIR=$(mktemp -d)
gunzip -c "${LATEST_BACKUP}" > ${TEMP_DIR}/backup.sql 2>/dev/null
if [ -s "${TEMP_DIR}/backup.sql" ]; then
echo "[$(date)] SQL文件解压成功，大小: $(ls -lh ${TEMP_DIR}/backup.sql | awk '{print $5}')" >> ${LOG_FILE}
# 检查关键内容
if grep -q "CREATE DATABASE" ${TEMP_DIR}/backup.sql && \
grep -q "LOCK TABLES" ${TEMP_DIR}/backup.sql && \
grep -q "UNLOCK TABLES" ${TEMP_DIR}/backup.sql; then
echo "[$(date)] SQL文件结构验证通过" >> ${LOG_FILE}
else
echo "[$(date)] WARNING: SQL文件结构可能不完整" >> ${LOG_FILE}
fi
else
echo "[$(date)] ERROR: SQL文件解压失败!" >> ${LOG_FILE}
rm -rf ${TEMP_DIR}
exit 1
fi
rm -rf ${TEMP_DIR}
echo "[$(date)] 备份验证完成" >> ${LOG_FILE}
# 每天自动验证
0 5 * * * /opt/mysql-backup/verify.sh >> /var/log/mysql-backup-verify.log 2>&1
2.6 第六层：漏洞管理与补丁策略
2.6.1 漏洞扫描脚本
#!/usr/bin/env python3
"""MySQL漏洞扫描工具"""
import pymysql
import re
import subprocess
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class MySQLVulnerabilityScanner:
def __init__(self, host, user, password, database='mysql'):
self.host = host
self.user = user
self.password = password
self.database = database
self.conn = None
self.cursor = None
def connect(self):
self.conn = pymysql.connect(
host=self.host,
user=self.user,
password=self.password,
database=self.database,
ssl=False
)
self.cursor = self.conn.cursor()
def close(self):
if self.cursor:
self.cursor.close()
if self.conn:
self.conn.close()
def get_version(self):
"""获取MySQL版本信息"""
self.cursor.execute("SELECT VERSION()")
version = self.cursor.fetchone()[0]
logger.info(f"MySQL版本: {version}")
return version
def check_vulnerabilities(self):
"""检查已知漏洞"""
vulnerabilities = []
# 1. 检查密码策略
self.cursor.execute("SHOW VARIABLES LIKE 'validate_password%'")
validate_password = {row[0]: row[1] for row in self.cursor.fetchall()}
if not validate_password:
vulnerabilities.append({
'id': 'CVE-MYSQL-001',
'severity': 'HIGH',
'description': '密码强度验证插件未安装',
'remediation': '安装validate_password插件并配置STRONG策略'
})
elif validate_password.get('validate_password.policy') != 'STRONG':
vulnerabilities.append({
'id': 'CVE-MYSQL-002',
'severity': 'MEDIUM',
'description': '密码策略强度不足',
'remediation': '将validate_password.policy设置为STRONG'
})
# 2. 检查root远程登录
self.cursor.execute("SELECT user, host FROM mysql.user WHERE user='root'")
roots = self.cursor.fetchall()
for user, host in roots:
if host not in ['localhost', '127.0.0.1']:
vulnerabilities.append({
'id': 'CVE-MYSQL-003',
'severity': 'CRITICAL',
'description': f'root账号允许从{host}远程登录',
'remediation': '限制root只能从localhost登录'
})
# 3. 检查空密码账号
self.cursor.execute("SELECT user, host FROM mysql.user WHERE password='' OR authentication_string=''")
empty_passwords = self.cursor.fetchall()
for user, host in empty_passwords:
vulnerabilities.append({
'id': 'CVE-MYSQL-004',
'severity': 'CRITICAL',
'description': f'账号{user}@{host}使用空密码',
'remediation': '立即设置强密码或删除该账号'
})
# 4. 检查过度权限
self.cursor.execute("SELECT user, host, super_priv, file_priv, process_priv FROM mysql.user")
users = self.cursor.fetchall()
for user, host, super_priv, file_priv, process_priv in users:
if super_priv == 'Y' or file_priv == 'Y' or process_priv == 'Y':
vulnerabilities.append({
'id': 'CVE-MYSQL-005',
'severity': 'HIGH',
'description': f'账号{user}@{host}拥有过高权限 (SUPER={super_priv}, FILE={file_priv}, PROCESS={process_priv})',
'remediation': '撤销不必要的权限，遵循最小权限原则'
})
# 5. 检查binlog是否开启
self.cursor.execute("SHOW VARIABLES LIKE 'log_bin'")
log_bin = self.cursor.fetchone()
if not log_bin or log_bin[1] != 'ON':
vulnerabilities.append({
'id': 'CVE-MYSQL-006',
'severity': 'MEDIUM',
'description': 'binlog未开启，无法进行点时间恢复',
'remediation': '启用binlog并配置合理的过期策略'
})
# 6. 检查慢查询日志
self.cursor.execute("SHOW VARIABLES LIKE 'slow_query_log'")
slow_log = self.cursor.fetchone()
if not slow_log or slow_log[1] != 'ON':
vulnerabilities.append({
'id': 'CVE-MYSQL-007',
'severity': 'LOW',
'description': '慢查询日志未开启',
'remediation': '启用慢查询日志以监控性能'
})
return vulnerabilities
def scan_external(self):
"""使用nmap扫描端口"""
try:
result = subprocess.run(
['nmap', '-sV', '--script', 'mysql-info', self.host],
capture_output=True,
text=True,
timeout=30
)
return result.stdout
except Exception as e:
logger.error(f"nmap扫描失败: {e}")
return ""
def generate_report(self):
"""生成安全报告"""
report = []
report.append(f"MySQL安全扫描报告")
report.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
report.append(f"目标主机: {self.host}")
report.append("=" * 60)
version = self.get_version()
report.append(f"\n版本信息: {version}")
vulns = self.check_vulnerabilities()
if vulns:
report.append(f"\n发现 {len(vulns)} 个安全问题:\n")
for i, vuln in enumerate(vulns, 1):
report.append(f"{i}. [{vuln['severity']}] {vuln['id']}")
report.append(f" 问题: {vuln['description']}")
report.append(f" 修复建议: {vuln['remediation']}\n")
else:
report.append("\n✅ 未发现安全问题!")
return "\n".join(report)
if __name__ == '__main__':
scanner = MySQLVulnerabilityScanner(
host='192.168.1.100',
user='app_admin',
password='A%3$jK8@mL5&nQ2'
)
try:
scanner.connect()
report = scanner.generate_report()
print(report)
# 保存报告
with open(f'/var/log/mysql-security-report-{datetime.now().strftime("%Y%m%d")}.txt', 'w') as f:
f.write(report)
finally:
scanner.close()
# 每周自动扫描
0 3 * * 0 /usr/bin/python3 /opt/mysql-security/scanner.py >> /var/log/mysql-security-scan.log 2>&1
2.7 第七层：入侵检测与应急响应
2.7.1 实时入侵检测脚本
#!/usr/bin/env python3
"""MySQL入侵实时检测与响应"""
import pymysql
import json
import time
import os
from datetime import datetime, timedelta
import subprocess
class MySQLIDS:
def __init__(self):
self.conn = None
self.cursor = None
self.last_check_time = None
self.blocked_ips = set()
# 规则配置
self.rules = {
'max_login_failures': 5,
'login_failure_window_seconds': 300, # 5分钟内
'suspicious_queries': [
r'DROP\s+TABLE',
r'DELETE\s+FROM\s+\w+\s+WHERE\s+1\s*=\s*1',
r'TRUNCATE\s+TABLE',
r'GRANT\s+ALL\s+ON',
r'UPDATE\s+mysql\.user',
r'CREATE\s+USER.*IDENTIFIED\s+BY',
r'SELECT\s+.*INTO\s+OUTFILE',
r'LOAD\s+FILE\s*\(',
r'INTO\s+DUMPFILE',
],
'alert_channels': {
'slack_webhook': 'https://hooks.slack.com/services/xxx/yyy/zzz',
'log_file': '/var/log/mysql-ids.log'
}
}
def connect(self):
self.conn = pymysql.connect(
host='192.168.1.100',
user='app_admin',
password='A%3$jK8@mL5&nQ2',
database='mysql'
)
self.cursor = self.conn.cursor()
self.last_check_time = datetime.now()
def close(self):
if self.cursor:
self.cursor.close()
if self.conn:
self.conn.close()
def detect_brute_force(self):
"""检测暴力破解"""
self.cursor.execute("""
SELECT user, host, COUNT(*) as fail_count, MAX(event_time) as last_fail
FROM mysql.general_log
WHERE command_type = 'Connect'
AND argument LIKE '%Access denied%'
AND event_time > NOW() - INTERVAL 5 MINUTE
GROUP BY user, host
HAVING fail_count >= 5
ORDER BY fail_count DESC
""")
attackers = self.cursor.fetchall()
alerts = []
for user, host, count, last_fail in attackers:
if host not in self.blocked_ips:
self.blocked_ips.add(host)
alert = {
'type': 'brute_force',
'severity': 'CRITICAL',
'timestamp': datetime.now().isoformat(),
'user': user,
'host': host,
'attempts': count,
'last_attempt': str(last_fail)
}
alerts.append(alert)
# 自动封锁IP
self.block_ip(host)
# 锁定账号
self.lock_account(user, host)
return alerts
def detect_suspicious_queries(self):
"""检测可疑SQL"""
alerts = []
for pattern in self.rules['suspicious_queries']:
self.cursor.execute(f"""
SELECT event_time, user_host, argument
FROM mysql.general_log
WHERE command_type = 'Query'
AND argument REGEXP '{pattern}'
AND event_time > NOW() - INTERVAL 1 MINUTE
ORDER BY event_time DESC
LIMIT 10
""")
results = self.cursor.fetchall()
for event_time, user_host, argument in results:
alert = {
'type': 'suspicious_query',
'severity': 'HIGH',
'timestamp': str(event_time),
'user_host': user_host,
'pattern': pattern,
'query': argument[:500] # 截取前500字符
}
alerts.append(alert)
return alerts
def detect_unauthorized_changes(self):
"""检测未经授权的配置变更"""
alerts = []
# 检查最近是否有配置变更
self.cursor.execute("""
SELECT event_time, user_host, argument
FROM mysql.general_log
WHERE command_type = 'Query'
AND (argument LIKE '%SET GLOBAL%' OR argument LIKE '%SET NAMES%')
AND event_time > NOW() - INTERVAL 5 MINUTE
ORDER BY event_time DESC
""")
changes = self.cursor.fetchall()
for event_time, user_host, argument in changes:
alert = {
'type': 'config_change',
'severity': 'MEDIUM',
'timestamp': str(event_time),
'user_host': user_host,
'change': argument[:200]
}
alerts.append(alert)
return alerts
def block_ip(self, ip):
"""自动封锁恶意IP"""
# 使用iptables封锁
try:
subprocess.run(
['sudo', 'iptables', '-A', 'INPUT', '-s', ip, '-j', 'DROP'],
check=True,
capture_output=True
)
print(f"[{datetime.now()}] 已封锁IP: {ip}")
except subprocess.CalledProcessError as e:
print(f"封锁IP失败: {e}")
def lock_account(self, user, host):
"""自动锁定账号"""
try:
self.cursor.execute(f"ALTER USER '{user}'@'{host}' ACCOUNT LOCK")
self.conn.commit()
print(f"[{datetime.now()}] 已锁定账号: {user}@{host}")
except Exception as e:
print(f"锁定账号失败: {e}")
def send_alert(self, alerts):
"""发送告警"""
if not alerts:
return
# 写入日志
with open(self.rules['alert_channels']['log_file'], 'a') as f:
for alert in alerts:
f.write(json.dumps(alert, default=str) + '\n')
# 发送Slack告警
import requests
for alert in alerts:
payload = {
'text': f"🚨 MySQL入侵检测告警\n"
f"类型: {alert['type']}\n"
f"严重程度: {alert['severity']}\n"
f"时间: {alert['timestamp']}\n",
'color': '#ff0000'
}
try:
requests.post(
self.rules['alert_channels']['slack_webhook'],
json=payload,
timeout=10
)
except Exception as e:
print(f"发送Slack告警失败: {e}")
def run_detection_cycle(self):
"""执行一次检测循环"""
alerts = []
try:
alerts.extend(self.detect_brute_force())
alerts.extend(self.detect_suspicious_queries())
alerts.extend(self.detect_unauthorized_changes())
if alerts:
self.send_alert(alerts)
finally:
self.last_check_time = datetime.now()
def start(self):
"""启动IDS服务"""
print(f"[{datetime.now()}] MySQL入侵检测系统启动")
print(f"[{datetime.now()}] 检测间隔: 30秒")
print(f"[{datetime.now()}] 日志文件: {self.rules['alert_channels']['log_file']}")
self.connect()
try:
while True:
self.run_detection_cycle()
time.sleep(30)
except KeyboardInterrupt:
print(f"\n[{datetime.now()}] IDS服务停止")
finally:
self.close()
if __name__ == '__main__':
ids = MySQLIDS()
ids.start()
# 配置为systemd服务
sudo tee /etc/systemd/system/mysql-ids.service << 'EOF'
[Unit]
Description=MySQL Intrusion Detection System
After=network.target mysql.service
[Service]
Type=simple
User=root
ExecStart=/usr/bin/python3 /opt/mysql-ids/ids.py
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
EOF
sudo systemctl daemon-reload
sudo systemctl enable mysql-ids
sudo systemctl start mysql-ids
2.7.2 应急响应手册
# MySQL勒索软件应急响应手册
## 第一步：隔离（0-5分钟）
1. 断开网络
```bash
# 停止MySQL服务
sudo systemctl stop mysql
# 断开网络（防止勒索软件继续扩散）
sudo ip link set eth0 down
# 或者只阻止MySQL端口
sudo iptables -A INPUT -p tcp --dport 3306 -j DROP
sudo iptables -A OUTPUT -p tcp --dport 3306 -j DROP
保护现场
- 不要重启服务器
- 不要格式化磁盘
- 内存镜像（如果可能）
# 创建内存镜像 sudo dd if=/dev/mem of=/tmp/memory_dump.img bs=4K
第二步：评估（5-15分钟）
确定影响范围 “`bash
检查被加密的文件
find /var/lib/mysql -name “*.ibd” -exec sh -c ‘file “\(1" | grep -q "encrypted" && echo "\)1”’ _ {} \;
# 检查进程 ps aux | grep -E “(mysqld|lockbit|ransom)”
# 检查网络连接 netstat -tlnp | grep mysql ss -tlnp | grep mysql
2. 查找入侵入口
```bash
# 检查可疑登录
grep "Access denied" /var/log/mysql/error.log | tail -50
# 检查异常进程
ls -la /tmp/ /dev/shm/ /var/tmp/
# 检查计划任务
crontab -l
ls -la /etc/cron.d/
第三步：恢复（15-60分钟）
从干净备份恢复 “`bash
停止MySQL
sudo systemctl stop mysql
# 备份当前损坏的数据目录 sudo mv /var/lib/mysql /var/lib/mysql.bak.$(date +%Y%m%d_%H%M%S)
# 恢复数据库目录 sudo mkdir -p /var/lib/mysql sudo chown -R mysql:mysql /var/lib/mysql
# 从异地备份恢复 rsync -avz backup@backup.company.com:/data/mysql-backups/latest/ /var/lib/mysql/
# 恢复数据文件 cd /var/lib/mysql gunzip -c /data/backups/mysql/latest/full_backup.sql.gz | mysql -u root -p
2. 验证恢复
```bash
# 启动MySQL
sudo systemctl start mysql
# 验证数据完整性
mysql -u root -p -e "SHOW DATABASES;"
mysql -u root -p -e "SELECT COUNT(*) FROM company_db.orders;"
# 对比备份前后的数据量
第四步：加固（60分钟后）
参考本文档第二至七章的安全加固措施。
第五步：复盘
- 生成事故报告
- 更新安全策略
- 进行安全培训
---
## 三、长效防护机制
### 3.1 定期安全审计
```bash
#!/bin/bash
# /opt/mysql-security/weekly_audit.sh
# 每周安全审计脚本
AUDIT_LOG="/var/log/mysql-weekly-audit.log"
DATE=$(date +%Y%m%d)
echo "===== MySQL周度安全审计 $(date) =====" >> ${AUDIT_LOG}
# 1. 账号审计
echo "" >> ${AUDIT_LOG}
echo "【账号审计】" >> ${AUDIT_LOG}
mysql -u app_admin -p'A%3$jK8@mL5&nQ2' -e "
SELECT user, host, account_locked, password_expired,
DATE_FORMAT(password_last_changed, '%Y-%m-%d') as last_change
FROM mysql.user
ORDER BY user, host;
" >> ${AUDIT_LOG} 2>&1
# 2. 权限审计
echo "" >> ${AUDIT_LOG}
echo "【权限审计】" >> ${AUDIT_LOG}
mysql -u app_admin -p'A%3$jK8@mL5&nQ2' -e "
SELECT user, host, Group_concat(privilege SEPARATOR ', ') as privileges
FROM (
SELECT user, host, privilege
FROM mysql.user
UNION ALL
SELECT user, host, privilege
FROM mysql.global_grants
) t
GROUP BY user, host
ORDER BY user, host;
" >> ${AUDIT_LOG} 2>&1
# 3. 可疑登录审计
echo "" >> ${AUDIT_LOG}
echo "【可疑登录审计】" >> ${AUDIT_LOG}
mysql -u app_admin -p'A%3$jK8@mL5&nQ2' -e "
SELECT user, host, COUNT(*) as fail_count
FROM mysql.general_log
WHERE command_type = 'Connect'
AND argument LIKE '%Access denied%'
AND event_time > NOW() - INTERVAL 7 DAY
GROUP BY user, host
HAVING fail_count > 10
ORDER BY fail_count DESC;
" >> ${AUDIT_LOG} 2>&1
# 4. 慢查询审计
echo "" >> ${AUDIT_LOG}
echo "【慢查询审计】" >> ${AUDIT_LOG}
mysql -u app_admin -p'A%3$jK8@mL5&nQ2' -e "
SELECT DB, COUNT(*) as slow_count, AVG(avg_time) as avg_time
FROM mysql.slow_log
WHERE start_time > NOW() - INTERVAL 7 DAY
GROUP BY DB
ORDER BY slow_count DESC
LIMIT 10;
" >> ${AUDIT_LOG} 2>&1
echo "===== 审计完成 $(date) =====" >> ${AUDIT_LOG}
# 设置每周审计任务
0 4 * * 0 /opt/mysql-security/weekly_audit.sh
3.2 安全配置文件模板
# /etc/mysql/mysql.conf.d/security.cnf
# MySQL安全加固配置文件
[mysqld]
# ==================== 网络配置 ====================
bind-address = 192.168.1.100
skip-networking = OFF
mysqlx_bind_address = 192.168.1.100
# 最大连接数（根据实际负载调整）
max_connections = 200
max_connect_errors = 100
# ==================== 认证配置 ====================
# 密码强度策略
validate_password.policy = STRONG
validate_password.length = 16
validate_password.mixed_case_count = 1
validate_password.number_count = 1
validate_password.special_char_count = 1
validate_password.check_user_name = ON
# 登录失败锁定
connection_control.failed_login_threshold = 5
connection_control.min_connection_delay = 1800000
connection_control.max_connection_delay = 3600000
# 密码过期策略
default_password_lifetime = 90
password_history = 5
password_reuse_interval = 30
# ==================== SSL/TLS配置 ====================
require_secure_transport = ON
ssl-ca = /etc/mysql/ssl/ca.pem
ssl-cert = /etc/mysql/ssl/server-cert.pem
ssl-key = /etc/mysql/ssl/server-key.pem
tls_version = TLSv1.2,TLSv1.3
# ==================== 日志配置 ====================
general_log = ON
general_log_file = /var/log/mysql/general.log
log_output = FILE
slow_query_log = ON
slow_query_log_file = /var/log/mysql/slow.log
long_query_time = 2
# 二进制日志（用于备份恢复）
log_bin = ON
binlog_format = ROW
binlog_expire_logs_seconds = 2592000 # 30天
max_binlog_size = 100M
sync_binlog = 1
# ==================== 安全配置 ====================
# 禁止本地文件读写
local_infile = OFF
secure_file_priv = /var/lib/mysql-files
# 禁止危险函数
local_infile = 0
# ==================== 性能与安全平衡 ====================
# 查询缓存（MySQL 8.0已移除）
# performance_schema = ON
3.3 Docker部署安全加固
# Dockerfile（MySQL安全加固版）
FROM mysql:8.0
# 创建安全目录
RUN mkdir -p /var/lib/mysql-files /etc/mysql/conf.d && \
chown -R mysql:mysql /var/lib/mysql-files /etc/mysql/conf.d
# 复制安全配置
COPY security.cnf /etc/mysql/conf.d/security.cnf
# 复制初始化脚本
COPY init.sh /docker-entrypoint-initdb.d/init.sh
RUN chmod +x /docker-entrypoint-initdb.d/init.sh
# 创建只读用户
RUN useradd -r -s /bin/false mysql-readonly && \
chown -R mysql-readonly:mysql-readonly /var/lib/mysql-files
# 设置文件权限
RUN chmod 750 /var/lib/mysql-files && \
chmod 600 /etc/mysql/ssl/*
USER mysql
EXPOSE 3306
CMD ["mysqld"]
#!/bin/bash
# init.sh - MySQL容器初始化脚本
set -e
# 创建应用账号并配置权限
mysql -u root -p"${MYSQL_ROOT_PASSWORD}" << 'EOF'
-- 创建应用账号
CREATE USER IF NOT EXISTS 'app_read'@'%' IDENTIFIED BY 'R3ad#Only$ecureP@ss';
CREATE USER IF NOT EXISTS 'app_write'@'%' IDENTIFIED BY 'Wr1te#Only$ecureP@ss';
CREATE USER IF NOT EXISTS 'app_admin'@'%' IDENTIFIED BY 'Adm1n#Only$ecureP@ss';
-- 分配权限
GRANT SELECT ON company_db.* TO 'app_read'@'%';
GRANT SELECT, INSERT, UPDATE ON company_db.* TO 'app_write'@'%';
GRANT SELECT, INSERT, UPDATE, DELETE ON company_db.orders TO 'app_write'@'%';
GRANT ALTER, CREATE, DROP, INDEX ON company_db.* TO 'app_admin'@'%';
-- 要求SSL连接
ALTER USER 'app_read'@'%' REQUIRE SSL;
ALTER USER 'app_write'@'%' REQUIRE SSL;
ALTER USER 'app_admin'@'%' REQUIRE SSL;
-- 设置密码策略
ALTER USER 'app_read'@'%' PASSWORD_EXPIRE INTERVAL 90 DAY;
ALTER USER 'app_write'@'%' PASSWORD_EXPIRE INTERVAL 90 DAY;
ALTER USER 'app_admin'@'%' PASSWORD_EXPIRE INTERVAL 90 DAY;
FLUSH PRIVILEGES;
EOF
echo "MySQL初始化完成"
# docker-compose.yml
version: '3.8'
services:
mysql:
build: .
container_name: mysql-secured
restart: unless-stopped
environment:
MYSQL_ROOT_PASSWORD: 'R00t#SecureP@ssw0rd!2024'
MYSQL_DATABASE: company_db
ports:
- "192.168.1.100:3306:3306" # 只绑定内网IP
volumes:
- mysql_data:/var/lib/mysql
- ./security.cnf:/etc/mysql/conf.d/security.cnf:ro
- ./ssl:/etc/mysql/ssl:ro
- ./init.sh:/docker-entrypoint-initdb.d/init.sh:ro
networks:
- backend
security_opt:
- no-new-privileges:true
read_only: true
tmpfs:
- /tmp
- /var/run/mysqld
deploy:
resources:
limits:
memory: 4G
cpus: '2'
volumes:
mysql_data:
driver: local
networks:
backend:
driver: bridge
ipam:
config:
- subnet: 172.16.0.0/24
3.4 安全运营Dashboard
#!/usr/bin/env python3
"""MySQL安全运营Dashboard"""
import pymysql
import json
from datetime import datetime, timedelta
from flask import Flask, jsonify, render_template_string
app = Flask(__name__)
def get_db_connection():
return pymysql.connect(
host='192.168.1.100',
user='app_admin',
password='A%3$jK8@mL5&nQ2',
database='mysql'
)
@app.route('/api/security/status')
def security_status():
"""安全状态API"""
conn = get_db_connection()
cursor = conn.cursor()
dashboard = {
'timestamp': datetime.now().isoformat(),
'security_score': 0,
'vulnerabilities': [],
'recent_alerts': [],
'backup_status': {}
}
# 1. 检查账号安全
cursor.execute("""
SELECT COUNT(*) as total_users,
SUM(CASE WHEN account_locked = 'Y' THEN 1 ELSE 0 END) as locked,
SUM(CASE WHEN password_expired = 'Y' THEN 1 ELSE 0 END) as expired,
SUM(CASE WHEN authentication_string IS NULL OR authentication_string = '' THEN 1 ELSE 0 END) as empty_pass
FROM mysql.user
""")
user_stats = cursor.fetchone()
if user_stats[3] > 0: # 有空密码
dashboard['security_score'] -= 30
dashboard['vulnerabilities'].append({
'severity': 'CRITICAL',
'message': f'发现 {user_stats[3]} 个空密码账号'
})
if user_stats[2] > 0: # 有过期密码
dashboard['security_score'] -= 10
dashboard['vulnerabilities'].append({
'severity': 'WARNING',
'message': f'有 {user_stats[2]} 个账号密码已过期'
})
# 2. 检查最近登录失败
cursor.execute("""
SELECT user, host, COUNT(*) as fail_count
FROM mysql.general_log
WHERE command_type = 'Connect'
AND argument LIKE '%Access denied%'
AND event_time > NOW() - INTERVAL 1 HOUR
GROUP BY user, host
ORDER BY fail_count DESC
LIMIT 5
""")
failed_logins = cursor.fetchall()
for user, host, count in failed_logins:
if count >= 5:
dashboard['recent_alerts'].append({
'type': 'brute_force',
'severity': 'CRITICAL',
'message': f'暴力破解检测: {user}@{host} 失败 {count} 次'
})
dashboard['security_score'] -= 20
# 3. 检查备份状态
backup_file = '/data/backups/mysql/latest/full_backup.sql.gz'
try:
import os
backup_size = os.path.getsize(backup_file)
backup_time = datetime.fromtimestamp(os.path.getmtime(backup_file))
hours_since_backup = (datetime.now() - backup_time).total_seconds() / 3600
dashboard['backup_status'] = {
'last_backup': backup_time.isoformat(),
'size_mb': round(backup_size / 1024 / 1024, 2),
'hours_since_backup': round(hours_since_backup, 1),
'status': 'OK' if hours_since_backup < 25 else 'STALE'
}
if hours_since_backup >= 25:
dashboard['vulnerabilities'].append({
'severity': 'HIGH',
'message': f'备份已超过 {hours_since_backup:.1f} 小时未更新'
})
dashboard['security_score'] -= 25
except Exception as e:
dashboard['backup_status'] = {'status': 'ERROR', 'error': str(e)}
dashboard['security_score'] -= 25
# 4. 检查配置安全
cursor.execute("SHOW VARIABLES LIKE 'validate_password.policy'")
policy = cursor.fetchone()
if not policy or policy[1] != 'STRONG':
dashboard['security_score'] -= 15
dashboard['vulnerabilities'].append({
'severity': 'MEDIUM',
'message': '密码策略强度不足'
})
cursor.execute("SHOW VARIABLES LIKE 'require_secure_transport'")
ssl = cursor.fetchone()
if not ssl or ssl[1] != 'ON':
dashboard['security_score'] -= 20
dashboard['vulnerabilities'].append({
'severity': 'HIGH',
'message': 'SSL/TLS未启用'
})
cursor.execute("SHOW VARIABLES LIKE 'local_infile'")
local_infile = cursor.fetchone()
if local_infile and local_infile[1] != 'OFF':
dashboard['security_score'] -= 10
dashboard['vulnerabilities'].append({
'severity': 'MEDIUM',
'message': 'local_infile未禁用'
})
# 计算安全分数（0-100）
dashboard['security_score'] = max(0, 100 + dashboard['security_score'])
cursor.close()
conn.close()
return jsonify(dashboard)
@app.route('/api/security/audit-log')
def audit_log():
"""审计日志API"""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT event_time, user_host, command_type, argument
FROM mysql.general_log
WHERE event_time > NOW() - INTERVAL 24 HOUR
ORDER BY event_time DESC
LIMIT 100
""")
logs = []
for row in cursor.fetchall():
logs.append({
'time': str(row[0]),
'user_host': row[1],
'command': row[2],
'query': row[3][:200] if row[3] else ''
})
cursor.close()
conn.close()
return jsonify({'logs': logs, 'total': len(logs)})
HTML_TEMPLATE = '''
<!DOCTYPE html>
<html>
<head>
<title>MySQL安全运营Dashboard</title>
<style>
body { font-family: Arial, sans-serif; margin: 20px; background: #f5f5f5; }
.dashboard { max-width: 1200px; margin: 0 auto; }
.card { background: white; border-radius: 8px; padding: 20px; margin-bottom: 20px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }
.score { font-size: 48px; font-weight: bold; text-align: center; }
.score-good { color: #4CAF50; }
.score-warning { color: #FF9800; }
.score-bad { color: #F44336; }
.alert { padding: 10px; margin: 5px 0; border-radius: 4px; }
.alert-critical { background: #ffebee; color: #c62828; }
.alert-high { background: #fff3e0; color: #ef6c00; }
.alert-medium { background: #fff8e1; color: #f9a825; }
.alert-low { background: #e8f5e9; color: #2e7d32; }
table { width: 100%; border-collapse: collapse; }
th, td { padding: 8px; text-align: left; border-bottom: 1px solid #ddd; }
th { background: #f5f5f5; }
.refresh-btn { background: #2196F3; color: white; border: none; padding: 10px 20px; border-radius: 4px; cursor: pointer; }
.refresh-btn:hover { background: #1976D2; }
</style>
</head>
<body>
<div class="dashboard">
<h1>🛡️ MySQL安全运营Dashboard</h1>
<button class="refresh-btn" onclick="refreshData()">刷新数据</button>
<div class="card">
<h2>安全评分</h2>
<div class="score" id="securityScore">--</div>
<p id="scoreMessage">加载中...</p>
</div>
<div class="card">
<h2>安全告警</h2>
<div id="alerts"></div>
</div>
<div class="card">
<h2>备份状态</h2>
<div id="backupStatus"></div>
</div>
<div class="card">
<h2>最近审计日志</h2>
<table>
<thead>
<tr>
<th>时间</th>
<th>用户/主机</th>
<th>类型</th>
<th>操作</th>
</tr>
</thead>
<tbody id="auditLog"></tbody>
</table>
</div>
</div>
<script>
function refreshData() {
fetch('/api/security/status')
.then(response => response.json())
.then(data => {
document.getElementById('securityScore').textContent = data.security_score;
const scoreEl = document.getElementById('securityScore');
scoreEl.className = 'score ' +
(data.security_score >= 80 ? 'score-good' :
data.security_score >= 50 ? 'score-warning' : 'score-bad');
document.getElementById('scoreMessage').textContent =
data.security_score >= 80 ? '✅ 安全评分优秀' :
data.security_score >= 50 ? '⚠️ 存在安全风险' : '🚨 严重安全风险！';
const alertsEl = document.getElementById('alerts');
alertsEl.innerHTML = data.recent_alerts.length > 0 ?
data.recent_alerts.map(a =>
`<div class="alert alert-${a.severity.toLowerCase()}">${a.message}</div>`
).join('') :
'<p>✅ 暂无告警</p>';
const backupEl = document.getElementById('backupStatus');
backupEl.innerHTML = `
<p>上次备份: ${data.backup_status.last_backup}</p>
<p>备份大小: ${data.backup_status.size_mb} MB</p>
<p>距上次备份: ${data.backup_status.hours_since_backup} 小时</p>
<p>状态: ${data.backup_status.status}</p>
`;
});
fetch('/api/security/audit-log')
.then(response => response.json())
.then(data => {
const tbody = document.getElementById('auditLog');
tbody.innerHTML = data.logs.map(log => `
<tr>
<td>${log.time}</td>
<td>${log.user_host}</td>
<td>${log.command}</td>
<td>${log.query}</td>
</tr>
`).join('');
});
}
refreshData();
setInterval(refreshData, 60000);
</script>
</body>
</html>
'''
@app.route('/')
def index():
return render_template_string(HTML_TEMPLATE)
if __name__ == '__main__':
app.run(host='192.168.1.100', port=8080)
四、安全检查清单
我把这次事故后整理的检查清单给你，你可以直接拿来用：
4.1 基础安全检查
-- 检查root账号
SELECT user, host FROM mysql.user WHERE user='root';
-- 期望：只有 root@localhost 和 root@127.0.0.1
-- 检查空密码账号
SELECT user, host FROM mysql.user
WHERE authentication_string IS NULL OR authentication_string = '';
-- 检查过度权限
SELECT user, host, super_priv, file_priv, process_priv
FROM mysql.user
WHERE super_priv='Y' OR file_priv='Y' OR process_priv='Y';
-- 检查远程访问
SELECT user, host FROM mysql.user WHERE host != 'localhost' AND host != '127.0.0.1';
4.2 配置安全检查
# 检查SSL配置
mysql -u root -p -e "SHOW VARIABLES LIKE '%ssl%';"
# 期望：require_secure_transport=ON
# 检查密码策略
mysql -u root -p -e "SHOW VARIABLES LIKE 'validate_password%';"
# 期望：policy=STRONG, length>=16
# 检查本地文件读写
mysql -u root -p -e "SHOW VARIABLES LIKE 'local_infile';"
# 期望：local_infile=OFF
# 检查binlog
mysql -u root -p -e "SHOW VARIABLES LIKE 'log_bin';"
# 期望：log_bin=ON
4.3 备份验证清单
# 检查备份文件
ls -lh /data/backups/mysql/
# 验证备份可恢复
mysql -u root -p -e "CREATE DATABASE backup_test;"
gunzip -c /data/backups/mysql/latest/full_backup.sql.gz | mysql -u root -p backup_test
mysql -u root -p -e "DROP DATABASE backup_test;"
4.4 网络安全检查
# 检查MySQL监听
netstat -tlnp | grep mysql
# 期望：只监听内网IP，不监听0.0.0.0
# 检查防火墙规则
sudo iptables -L -n | grep 3306
# 期望：只允许特定IP访问3306端口
# 检查SSH访问
sudo grep -r "PermitRootLogin" /etc/ssh/sshd_config
# 期望：PermitRootLogin no
五、最后的建议
说实话，写这篇指南的时候，我脑海里全是那天早上老张那张脸。
我们花了三个月恢复数据，期间业务停摆，客户投诉，团队加班到凌晨三点。最让人崩溃的不是技术上的困难，而是那种”明明可以避免”的无力感。
MySQL安全这事儿，不是什么高深的技术，就是把该做的事做到位。
你现在就能做的三件事：
- 改密码 — 把所有MySQL账号的密码换成16位以上的复杂密码
- 关外网 — 把3306端口从公网移除，只对内网开放
- 验备份 — 找个测试环境，从备份恢复一次，确认备份真的能用
这三件事加起来不超过半小时，但能帮你挡住90%的威胁。
剩下的那些自动化检测、实时告警、异地容灾，都是锦上添花。先把基础打牢，再谈高级。
你如果还有具体问题，比如你们的MySQL版本比较老不知道怎么升级，或者备份策略不知道怎么设计，直接问我，我给你出方案。
记住：安全不是某个人的事，是所有人的事。 你多一份小心，我们就能少一场噩梦。