MySQL数据库安全配置规范与实战要点

-- 创建只读用户（用于报表查询）
CREATE USER 'report_user'@'10.0.0.%' IDENTIFIED BY 'ComplexPass123!@#';
GRANT SELECT ON sales_db.* TO 'report_user'@'10.0.0.%';
 
-- 创建应用用户（仅限特定操作）
CREATE USER 'app_user'@'localhost' IDENTIFIED BY 'AppSecure456!@#';
GRANT SELECT, INSERT, UPDATE ON app_db.user_table TO 'app_user'@'localhost';
GRANT SELECT, INSERT ON app_db.log_table TO 'app_user'@'localhost';
 
-- 创建管理用户（按功能分离）
CREATE USER 'dba_admin'@'localhost' IDENTIFIED BY 'AdminSecure789!@#';
GRANT ALL PRIVILEGES ON *.* TO 'dba_admin'@'localhost' WITH GRANT OPTION;
 
-- 撤销危险权限
REVOKE FILE ON *.* FROM 'app_user'@'localhost';
REVOKE PROCESS ON *.* FROM 'app_user'@'localhost';

-- 查看用户权限
SHOW GRANTS FOR 'app_user'@'localhost';
 
-- 查看所有用户及其权限
SELECT user, host, authentication_string, account_locked
FROM mysql.user 
WHERE user NOT IN ('mysql.sys', 'mysql.session', 'mysql.infoschema');
 
-- 监控权限变更
SELECT user, host, db, command, time, state
FROM information_schema.processlist 
WHERE command != 'Sleep';

# /etc/mysql/mysql.conf.d/mysqld.cnf
[mysqld]
# 仅绑定到内网地址
bind-address = 10.0.0.100
# 或者仅监听本地回环
# bind-address = 127.0.0.1
 
# 设置MySQL端口（避免使用默认3306）
port = 3307
 
# 禁用DNS反向解析（提高连接速度，避免DNS劫持）
skip-name-resolve

# Ubuntu/Debian系统 - UFW配置
# 允许特定IP访问MySQL端口
sudo ufw allow from 10.0.0.0/24 to any port 3307
sudo ufw allow from 192.168.1.0/24 to any port 3307
 
# 拒绝其他所有访问
sudo ufw deny 3307
 
# CentOS/RHEL系统 - firewalld配置
# 创建新的服务规则
sudo firewall-cmd --permanent --new-service=mysql-custom
sudo firewall-cmd --permanent --service=mysql-custom --add-port=3307/tcp
 
# 允许特定网段访问
sudo firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="10.0.0.0/24" service name="mysql-custom" accept'
sudo firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="192.168.1.0/24" service name="mysql-custom" accept'
 
# 重新加载防火墙规则
sudo firewall-cmd --reload

# MySQL服务器SSL配置
[mysqld]
# 启用SSL
ssl-ca = /etc/mysql/ssl/ca.pem
ssl-cert = /etc/mysql/ssl/server-cert.pem
ssl-key = /etc/mysql/ssl/server-key.pem
 
# 强制使用SSL连接（MySQL 5.7+）
require_secure_transport = ON
 
# 设置TLS版本（禁用不安全的TLSv1.0/1.1）
tls_version = TLSv1.2,TLSv1.3

# 生成SSL证书（使用OpenSSL）
# 创建CA私钥
openssl genrsa 4096 > ca-key.pem
 
# 创建CA证书
openssl req -new -x509 -nodes -days 3650 -key ca-key.pem -out ca.pem
 
# 创建服务器私钥
openssl req -newkey rsa:4096 -days 3650 -nodes -keyout server-key.pem -out server-req.pem
 
# 移除私钥密码
openssl rsa -in server-key.pem -out server-key.pem
 
# 使用CA签发服务器证书
openssl x509 -req -in server-req.pem -days 3650 -CA ca.pem -CAkey ca-key.pem -set_serial 01 -out server-cert.pem

-- 创建加密表空间（MySQL 8.0+）
CREATE TABLESPACE encrypted_ts
ADD DATAFILE 'encrypted_ts.ibd'
ENCRYPTION='Y';
 
-- 在加密表空间中创建表
CREATE TABLE sensitive_data (
    id INT PRIMARY KEY AUTO_INCREMENT,
    user_ssn VARCHAR(255),
    credit_card VARCHAR(255),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) TABLESPACE encrypted_ts;
 
-- 对现有表启用加密
ALTER TABLE users ENCRYPTION='Y';

# Python应用层加密示例
import hashlib
import secrets
from cryptography.fernet import Fernet
 
class DataEncryption:
    def __init__(self):
        # 生成或加载加密密钥
        self.key = self.load_or_generate_key()
        self.cipher = Fernet(self.key)
    
    def load_or_generate_key(self):
        """从安全位置加载或生成加密密钥"""
        try:
            with open('/secure/path/encryption.key', 'rb') as key_file:
                return key_file.read()
        except FileNotFoundError:
            key = Fernet.generate_key()
            with open('/secure/path/encryption.key', 'wb') as key_file:
                key_file.write(key)
            return key
    
    def encrypt_sensitive_data(self, data):
        """加密敏感数据"""
        return self.cipher.encrypt(data.encode()).decode()
    
    def decrypt_sensitive_data(self, encrypted_data):
        """解密敏感数据"""
        return self.cipher.decrypt(encrypted_data.encode()).decode()
 
# 使用示例
encryption = DataEncryption()
encrypted_ssn = encryption.encrypt_sensitive_data("123-45-6789")

-- 使用SHA-256哈希存储密码（不推荐，仅作示例）
INSERT INTO users (username, password_hash) 
VALUES ('john_doe', SHA2('user_password', 256));
 
-- 更好的方式是使用专门的密码哈希函数（如bcrypt）
-- 这里展示数据库层面的验证
SELECT * FROM users 
WHERE username = 'john_doe' 
AND password_hash = SHA2('input_password', 256);

# 安装审计插件（MySQL Enterprise Edition）
[mysqld]
plugin-load-add=audit_log.so
audit-log=FORCE_PLUS_PERMANENT
 
# 审计日志配置
audit_log_policy=ALL
audit_log_format=JSON
audit_log_file=/var/log/mysql/audit.log
audit_log_rotate_on_size=100M
audit_log_rotations=10

# 启用通用查询日志（仅开发环境使用）
[mysqld]
general_log = 1
general_log_file = /var/log/mysql/general.log
log_queries_not_using_indexes = 1
 
# 慢查询日志配置
slow_query_log = 1
slow_query_log_file = /var/log/mysql/slow.log
long_query_time = 2
log_slow_admin_statements = 1

#!/bin/bash
# MySQL安全监控脚本
 
# 配置变量
MYSQL_USER="monitor_user"
MYSQL_PASS="monitor_pass"
MYSQL_HOST="localhost"
LOG_FILE="/var/log/mysql/security_monitor.log"
ALERT_EMAIL="dba@company.com"
 
# 检查失败登录尝试
check_failed_logins() {
    failed_attempts=$(mysql -u$MYSQL_USER -p$MYSQL_PASS -h$MYSQL_HOST -e "
        SELECT COUNT(*) FROM mysql.general_log 
        WHERE command_type = 'Connect' 
        AND argument LIKE 'Access denied%' 
        AND event_time > NOW() - INTERVAL 1 HOUR;
    " 2>/dev/null | tail -1)
    
    if [ $failed_attempts -gt 10 ]; then
        echo "$(date): 警告 - 检测到 $failed_attempts 次失败登录尝试" >> $LOG_FILE
        echo "MySQL安全警告：检测到异常登录活动" | mail -s "MySQL安全告警" $ALERT_EMAIL
    fi
}
 
# 检查权限变更
check_privilege_changes() {
    mysql -u$MYSQL_USER -p$MYSQL_PASS -h$MYSQL_HOST -e "
        SELECT user, host, db, table_name, grantor, privilege_type, is_grantable
        FROM information_schema.schema_privileges 
        WHERE timestamp > NOW() - INTERVAL 1 HOUR;
    " 2>/dev/null | grep -v "^user"
    
    if [ $? -eq 0 ]; then
        echo "$(date): 警告 - 检测到权限变更操作" >> $LOG_FILE
    fi
}
 
# 主监控循环
main() {
    check_failed_logins
    check_privilege_changes
}
 
# 执行监控
main

-- MySQL 8.0+ 密码策略配置
-- 查看当前密码策略
SHOW VARIABLES LIKE 'validate_password%';
 
-- 设置密码策略（需要SUPER权限）
SET GLOBAL validate_password.policy = 'STRONG';
SET GLOBAL validate_password.length = 14;
SET GLOBAL validate_password.mixed_case_count = 2;
SET GLOBAL validate_password.number_count = 2;
SET GLOBAL validate_password.special_char_count = 2;
 
-- 密码过期策略
SET GLOBAL default_password_lifetime = 90; -- 90天过期
 
-- 账户锁定策略
SET GLOBAL connection_control_failed_connections_threshold = 5;
SET GLOBAL connection_control_max_connection_delay = 3600000; -- 1小时

-- 创建带有过期时间的用户
CREATE USER 'temp_user'@'localhost' 
IDENTIFIED BY 'TempPass123!@#'
PASSWORD EXPIRE INTERVAL 30 DAY
ACCOUNT LOCK;
 
-- 解锁账户并设置密码过期
ALTER USER 'temp_user'@'localhost' 
ACCOUNT UNLOCK
PASSWORD EXPIRE;
 
-- 强制用户下次登录时修改密码
ALTER USER 'temp_user'@'localhost' PASSWORD EXPIRE;
 
-- 禁用不再使用的账户（而非删除）
ALTER USER 'old_user'@'localhost' ACCOUNT LOCK;
 
-- 定期审计用户活动
SELECT user, host, account_locked, password_last_changed, 
       password_lifetime, account_locked
FROM mysql.user 
WHERE account_locked = 'N' 
AND password_last_changed < NOW() - INTERVAL 90 DAY;

# 使用Google Authenticator插件（需要安装）
# 安装插件
mysql> INSTALL PLUGIN authentication_pam SONAME 'authentication_pam.so';
 
# 创建使用2FA的用户
mysql> CREATE USER 'admin_user'@'localhost' IDENTIFIED WITH authentication_pam 
       AS 'mysql, mysqld:google_authenticator';
 
# 为用户生成2FA密钥
mysql> SET @secret = BASE32_ENCODE(RANDOM_BYTES(20));
mysql> SELECT @secret;

#!/bin/bash
# 加密备份脚本
 
# 配置变量
BACKUP_DIR="/secure/backup/mysql"
ENCRYPTION_KEY="/secure/path/backup.key"
RETENTION_DAYS=30
MYSQL_USER="backup_user"
MYSQL_PASS="backup_pass"
 
# 创建备份目录
mkdir -p $BACKUP_DIR/$(date +%Y%m%d)
 
# 使用mysqldump进行逻辑备份并加密
mysqldump -u$MYSQL_USER -p$MYSQL_PASS \
  --all-databases \
  --single-transaction \
  --routines \
  --triggers \
  --events \
  --hex-blob | \
openssl enc -aes-256-cbc -salt -out $BACKUP_DIR/$(date +%Y%m%d)/full_backup_$(date +%H%M%S).sql.enc -pass file:$ENCRYPTION_KEY
 
# 备份验证
echo "$(date): 备份完成，开始验证..." >> /var/log/mysql/backup.log
 
# 解密并验证备份文件
openssl enc -aes-256-cbc -d -in $BACKUP_DIR/$(date +%Y%m%d)/full_backup_*.sql.enc -pass file:$ENCRYPTION_KEY | head -10 > /tmp/backup_test.sql
 
if [ -s /tmp/backup_test.sql ]; then
    echo "$(date): 备份验证成功" >> /var/log/mysql/backup.log
else
    echo "$(date): 备份验证失败！" >> /var/log/mysql/backup.log
    # 发送告警邮件
    echo "MySQL备份验证失败" | mail -s "备份告警" dba@company.com
fi
 
# 清理旧备份
find $BACKUP_DIR -type f -name "*.enc" -mtime +$RETENTION_DAYS -delete
echo "$(date): 清理完成，保留最近 $RETENTION_DAYS 天备份" >> /var/log/mysql/backup.log

#!/bin/bash
# 使用XtraBackup进行加密物理备份
 
# 配置变量
BACKUP_DIR="/secure/backup/xtrabackup"
ENCRYPTION_KEY_FILE="/secure/path/xtrabackup_key"
MYSQL_USER="xtrabackup_user"
MYSQL_PASS="xtrabackup_pass"
 
# 创建加密密钥（如果不存在）
if [ ! -f $ENCRYPTION_KEY_FILE ]; then
    openssl rand -base64 32 > $ENCRYPTION_KEY_FILE
    chmod 600 $ENCRYPTION_KEY_FILE
fi
 
# 执行加密备份
xtrabackup --backup \
  --user=$MYSQL_USER \
  --password=$MYSQL_PASS \
  --target-dir=$BACKUP_DIR/$(date +%Y%m%d_%H%M%S) \
  --encrypt=AES256 \
  --encrypt-key-file=$ENCRYPTION_KEY_FILE \
  --compress \
  --parallel=4 \
  --slave-info \
  --safe-slave-backup
 
# 验证备份
xtrabackup --prepare \
  --target-dir=$BACKUP_DIR/$(date +%Y%m%d_%H%M%S) \
  --encrypt-key-file=$ENCRYPTION_KEY_FILE
 
echo "$(date): XtraBackup加密备份完成" >> /var/log/mysql/xtrabackup.log

#!/bin/bash
# 定期恢复测试脚本
 
# 配置变量
TEST_RESTORE_DIR="/tmp/mysql_restore_test"
BACKUP_KEY="/secure/path/backup.key"
PRODUCTION_DB="production_db"
TEST_DB="restore_test_$(date +%Y%m%d_%H%M%S)"
 
# 创建测试环境
mkdir -p $TEST_RESTORE_DIR
 
# 选择最新的备份文件进行恢复测试
LATEST_BACKUP=$(find /secure/backup/mysql -name "*.enc" -type f -exec ls -t {} + | head -1)
 
echo "$(date): 开始恢复测试，使用备份文件: $LATEST_BACKUP" >> /var/log/mysql/restore_test.log
 
# 解密并恢复备份
openssl enc -aes-256-cbc -d -in $LATEST_BACKUP -pass file:$BACKUP_KEY | \
mysql -e "CREATE DATABASE IF NOT EXISTS $TEST_DB; USE $TEST_DB;"
 
# 验证数据完整性
mysql $TEST_DB -e "
    SELECT COUNT(*) as total_records,
           MAX(updated_at) as latest_update,
           MIN(created_at) as earliest_record
    FROM information_schema.tables 
    WHERE table_schema = '$TEST_DB' 
    AND table_rows > 0;
" >> /var/log/mysql/restore_test.log
 
# 对比关键数据表记录数
mysql -e "
    SELECT 'Production' as source, 
           COUNT(*) as user_count 
    FROM $PRODUCTION_DB.users
    UNION ALL
    SELECT 'Restore_Test' as source, 
           COUNT(*) as user_count 
    FROM $TEST_DB.users;
" >> /var/log/mysql/restore_test.log
 
# 清理测试环境
mysql -e "DROP DATABASE IF EXISTS $TEST_DB;"
rm -rf $TEST_RESTORE_DIR
 
echo "$(date): 恢复测试完成" >> /var/log/mysql/restore_test.log

# 安全的Python代码示例（使用参数化查询）
import mysql.connector
from mysql.connector import Error
 
class SecureDatabase:
    def __init__(self, host, database, user, password):
        self.connection = mysql.connector.connect(
            host=host,
            database=database,
            user=user,
            password=password,
            charset='utf8mb4'
        )
    
    def get_user_by_id(self, user_id):
        """安全的查询方法 - 使用参数化查询防止SQL注入"""
        cursor = self.connection.cursor(dictionary=True)
        try:
            # 使用参数化查询，避免字符串拼接
            query = "SELECT id, username, email FROM users WHERE id = %s"
            cursor.execute(query, (user_id,))
            return cursor.fetchone()
        finally:
            cursor.close()
    
    def search_users(self, search_term):
        """安全的搜索方法 - 使用参数化查询"""
        cursor = self.connection.cursor(dictionary=True)
        try:
            # 使用参数化查询，LIKE语句也安全
            query = "SELECT id, username, email FROM users WHERE username LIKE %s"
            cursor.execute(query, (f"%{search_term}%",))
            return cursor.fetchall()
        finally:
            cursor.close()
 
# 不安全的做法（切勿使用）
def unsafe_get_user(user_id):
    # 危险！直接字符串拼接
    query = f"SELECT * FROM users WHERE id = {user_id}"
    # 攻击者可以传入: 1 OR 1=1 来获取所有用户
    return query

-- 防止权限提升的安全配置
-- 1. 禁用本地文件读取
SET GLOBAL local_infile = 0;
 
-- 2. 限制用户权限
REVOKE FILE ON *.* FROM 'web_user'@'localhost';
REVOKE SUPER ON *.* FROM 'web_user'@'localhost';
REVOKE GRANT OPTION ON *.* FROM 'web_user'@'localhost';
 
-- 3. 安全的数据库配置
SET GLOBAL secure_file_priv = '/secure/mysql-files/';
SET GLOBAL plugin_dir = '/usr/lib/mysql/plugin/';
 
-- 4. 监控权限变更
CREATE TABLE security_audit_log (
    id INT AUTO_INCREMENT PRIMARY KEY,
    user_host VARCHAR(255),
    query_text TEXT,
    execution_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_execution_time (execution_time)
);
 
-- 创建触发器监控GRANT操作（需要SUPER权限）
DELIMITER //
CREATE TRIGGER monitor_grant_operations
AFTER INSERT ON mysql.user
FOR EACH ROW
BEGIN
    INSERT INTO security_audit_log (user_host, query_text)
    VALUES (CONCAT(NEW.User, '@', NEW.Host), 'GRANT operation detected');
END//
DELIMITER ;

# MySQL配置防止DoS攻击
[mysqld]
# 连接数限制
max_connections = 200
max_user_connections = 50
 
# 查询限制
max_execution_time = 30000  # 30秒
max_statement_time = 30000   # 30秒
 
# 内存使用限制
max_heap_table_size = 64M
tmp_table_size = 64M
max_allowed_packet = 16M
 
# 表缓存限制
table_open_cache = 2000
table_definition_cache = 1400
 
# 线程缓存
thread_cache_size = 100

# 数据脱敏示例
import re
import hashlib
 
class DataMasking:
    @staticmethod
    def mask_email(email):
        """邮箱脱敏：john.doe@example.com -> j***e@example.com"""
        if '@' not in email:
            return email
        
        username, domain = email.split('@')
        if len(username) <= 3:
            return f"{username[0]}***@{domain}"
        
        masked_username = f"{username[0]}{'*' * (len(username) - 2)}{username[-1]}"
        return f"{masked_username}@{domain}"
    
    @staticmethod
    def mask_phone(phone):
        """手机号脱敏：13812345678 -> 138****5678"""
        phone = re.sub(r'\D', '', phone)  # 移除非数字字符
        if len(phone) == 11:
            return f"{phone[:3]}{'*' * 4}{phone[7:]}"
        return phone
    
    @staticmethod
    def mask_id_card(id_card):
        """身份证号脱敏：11010519900307283X -> 1101**********83X"""
        if len(id_card) == 18:
            return f"{id_card[:4]}{'*' * 10}{id_card[14:]}"
        return id_card
    
    @staticmethod
    def hash_sensitive_data(data, salt=""):
        """对敏感数据进行哈希处理"""
        return hashlib.sha256(f"{data}{salt}".encode()).hexdigest()
 
# 使用示例
masking = DataMasking()
print(masking.mask_email("john.doe@company.com"))  # j***e@company.com
print(masking.mask_phone("13812345678"))          # 138****5678
print(masking.mask_id_card("11010519900307283X")) # 1101**********83X

#!/bin/bash
# MySQL安全配置检查脚本
 
echo "=== MySQL安全配置检查清单 ==="
echo "检查时间: $(date)"
echo "================================"
 
# 1. 检查匿名用户
echo "1. 匿名用户检查:"
mysql -e "SELECT User, Host FROM mysql.user WHERE User = '';"
 
# 2. 检查测试数据库
echo -e "\n2. 测试数据库检查:"
mysql -e "SHOW DATABASES LIKE 'test%';"
 
# 3. 检查密码策略
echo -e "\n3. 密码策略检查:"
mysql -e "SHOW VARIABLES LIKE 'validate_password%';"
 
# 4. 检查SSL配置
echo -e "\n4. SSL配置检查:"
mysql -e "SHOW VARIABLES LIKE '%ssl%';"
 
# 5. 检查网络配置
echo -e "\n5. 网络配置检查:"
mysql -e "SHOW VARIABLES LIKE 'bind_address';"
mysql -e "SHOW VARIABLES LIKE 'skip_networking';"
 
# 6. 检查权限配置
echo -e "\n6. 权限配置检查:"
mysql -e "SELECT User, Host, Super_priv, Grant_priv FROM mysql.user WHERE Super_priv = 'Y';"
 
# 7. 检查日志配置
echo -e "\n7. 日志配置检查:"
mysql -e "SHOW VARIABLES LIKE 'general_log%';"
mysql -e "SHOW VARIABLES LIKE 'slow_query_log%';"
 
# 8. 检查文件权限
echo -e "\n8. 文件权限检查:"
ls -la /etc/mysql/my.cnf
ls -la /var/log/mysql/
 
echo -e "\n================================"
echo "检查完成！请根据结果进行相应配置优化。"

// .trae/security-config.json
{
  "mysql_security": {
    "enforce_ssl": true,
    "min_password_length": 14,
    "password_complexity": "strong",
    "max_connections_per_user": 50,
    "audit_log_enabled": true,
    "sql_injection_detection": true
  },
  "alerts": {
    "failed_login_threshold": 10,
    "privilege_change_notification": true,
    "unusual_query_detection": true
  }
}

# TRAE IDE安全测试集成
import unittest
from trae_security import MySQLSecurityTester
 
class TestMySQLSecurity(unittest.TestCase):
    def setUp(self):
        self.security_tester = MySQLSecurityTester(
            host='localhost',
            user='test_user',
            password='test_pass'
        )
    
    def test_ssl_enforcement(self):
        """测试SSL强制启用"""
        result = self.security_tester.check_ssl_enforcement()
        self.assertTrue(result['ssl_required'], "SSL应该被强制启用")
    
    def test_password_policy(self):
        """测试密码策略"""
        result = self.security_tester.check_password_policy()
        self.assertGreaterEqual(result['min_length'], 14, "密码最小长度应该至少14位")
        self.assertEqual(result['policy'], 'STRONG', "密码策略应该为STRONG")
    
    def test_privilege_separation(self):
        """测试权限分离"""
        result = self.security_tester.check_privilege_separation()
        self.assertFalse(result['has_all_privileges'], "不应该存在拥有所有权限的用户")
    
    def test_audit_logging(self):
        """测试审计日志"""
        result = self.security_tester.check_audit_logging()
        self.assertTrue(result['audit_enabled'], "审计日志应该被启用")
        self.assertIn('ALL', result['audit_events'], "应该记录所有类型的事件")
 
if __name__ == '__main__':
    unittest.main()