数据库安全加固:从权限控制到备份恢复

一、数据库安全概述与重要性

1. 数据库安全面临的挑战

在数字化时代，数据库作为企业核心资产的存储中心，面临着来自内外部的多重安全威胁：

内部威胁

员工误操作、恶意行为或权限滥用导致的数据泄露和破坏。

外部攻击

SQL注入、暴力破解、DDoS攻击等外部恶意行为。

合规风险

违反GDPR、网络安全法等法律法规带来的法律风险。

🚨 数据泄露的代价：

根据IBM《2025年数据泄露成本报告》，平均数据泄露成本达到435万美元，比上一年增长2.6%。其中医疗行业数据泄露成本最高，平均达到1010万美元。

2. 数据库安全防护层级

全面的数据库安全防护需要覆盖多个层级：

1

网络层防护

防火墙配置、网络隔离、VPN访问控制、端口安全

2

访问控制层

身份认证、权限管理、最小权限原则、角色分离

3

数据层防护

数据加密、脱敏处理、数据分类、访问审计

4

应用层防护

输入验证、参数化查询、存储过程、ORM框架安全

5

运维层防护

定期备份、漏洞扫描、安全补丁、监控告警

二、访问控制与权限管理

🔐 最小权限原则：

每个用户和进程只能拥有完成其任务所必需的最小权限。这是数据库安全的核心原则之一。

1. 用户与角色管理

合理的用户角色划分是权限管理的基础：

-- MySQL 用户与角色管理示例
-- 1. 创建不同角色的用户
CREATE USER 'app_readonly'@'%' IDENTIFIED BY 'StrongPassword123!';
CREATE USER 'app_write'@'192.168.1.%' IDENTIFIED BY 'StrongPassword456!';
CREATE USER 'admin'@'localhost' IDENTIFIED BY 'AdminStrongPass789!';

-- 2. 创建角色
CREATE ROLE 'readonly_role', 'write_role', 'admin_role';

-- 3. 为角色分配权限
-- 只读角色：只能查询特定数据库
GRANT SELECT ON company_db.* TO 'readonly_role';

-- 读写角色：可以查询、插入、更新特定表
GRANT SELECT, INSERT, UPDATE ON company_db.customers TO 'write_role';
GRANT SELECT, INSERT, UPDATE ON company_db.orders TO 'write_role';

-- 管理员角色：完整权限（谨慎使用）
GRANT ALL PRIVILEGES ON company_db.* TO 'admin_role' WITH GRANT OPTION;

-- 4. 将角色分配给用户
GRANT 'readonly_role' TO 'app_readonly'@'%';
GRANT 'write_role' TO 'app_write'@'192.168.1.%';
GRANT 'admin_role' TO 'admin'@'localhost';

-- 5. 设置默认角色
SET DEFAULT ROLE 'readonly_role' FOR 'app_readonly'@'%';
SET DEFAULT ROLE 'write_role' FOR 'app_write'@'192.168.1.%';
SET DEFAULT ROLE 'admin_role' FOR 'admin'@'localhost';

-- 6. 查看用户权限
SHOW GRANTS FOR 'app_readonly'@'%';
SHOW GRANTS FOR 'app_write'@'192.168.1.%';

-- 7. 定期审计用户权限
SELECT * FROM mysql.user WHERE User NOT IN ('mysql.sys', 'mysql.session', 'mysql.infoschema');
SELECT User, Host, authentication_string FROM mysql.user WHERE authentication_string = ''; -- 查找空密码用户

2. 权限矩阵设计

企业级数据库权限矩阵示例：

操作/角色	只读用户	数据录入员	数据分析师	开发人员	DBA
SELECT（查询）	✓	✓	✓	✓	✓
INSERT（插入）	✗	✓	✗	✓	✓
UPDATE（更新）	✗	部分	✗	✓	✓
DELETE（删除）	✗	✗	✗	测试环境	✓
CREATE（创建表）	✗	✗	✗	✓	✓
DROP（删除表）	✗	✗	✗	✗	✓
GRANT（授权）	✗	✗	✗	✗	✓

3. 最佳实践：定期权限审计

#!/bin/bash
# database_permission_audit.sh
# 数据库权限审计脚本

DB_HOST="localhost"
DB_PORT="3306"
DB_USER="audit_user"
DB_PASS="AuditPass123!"
AUDIT_REPORT="/var/log/db_audit/report_$(date +%Y%m%d).txt"

# 创建审计目录
mkdir -p /var/log/db_audit

echo "=== 数据库权限审计报告 ===" > $AUDIT_REPORT
echo "审计时间: $(date)" >> $AUDIT_REPORT
echo "数据库主机: $DB_HOST" >> $AUDIT_REPORT
echo "==========================" >> $AUDIT_REPORT

# 1. 检查所有用户及其权限
echo -e "\n1. 所有数据库用户及权限:" >> $AUDIT_REPORT
mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS -e "
SELECT 
    User, 
    Host,
    Grant_priv,
    Alter_priv,
    Create_priv,
    Delete_priv,
    Drop_priv,
    Insert_priv,
    Select_priv,
    Update_priv
FROM mysql.user 
WHERE User NOT IN ('mysql.sys', 'mysql.session', 'mysql.infoschema')
ORDER BY User, Host;
" >> $AUDIT_REPORT

# 2. 检查空密码用户
echo -e "\n2. 空密码用户检查:" >> $AUDIT_REPORT
mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS -e "
SELECT User, Host 
FROM mysql.user 
WHERE authentication_string = '' OR authentication_string IS NULL;
" >> $AUDIT_REPORT

# 3. 检查弱密码用户（示例检查）
echo -e "\n3. 可能使用弱密码的用户:" >> $AUDIT_REPORT
mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS -e "
SELECT User, Host 
FROM mysql.user 
WHERE User IN ('root', 'admin', 'test', 'user', 'guest');
" >> $AUDIT_REPORT

# 4. 检查数据库级别的权限
echo -e "\n4. 数据库级别权限:" >> $AUDIT_REPORT
mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS -e "
SELECT 
    User, 
    Host, 
    Db, 
    Select_priv,
    Insert_priv,
    Update_priv,
    Delete_priv,
    Create_priv,
    Drop_priv
FROM mysql.db 
ORDER BY User, Db;
" >> $AUDIT_REPORT

# 5. 检查表级别权限
echo -e "\n5. 表级别权限:" >> $AUDIT_REPORT
mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS -e "
SELECT 
    User, 
    Host, 
    Db, 
    Table_name,
    Table_priv
FROM mysql.tables_priv 
ORDER BY User, Db, Table_name;
" >> $AUDIT_REPORT

# 6. 检查最近修改的权限
echo -e "\n6. 最近7天权限变更记录:" >> $AUDIT_REPORT
mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS -e "
SELECT * FROM mysql.general_log 
WHERE argument_text LIKE '%GRANT%' 
   OR argument_text LIKE '%REVOKE%'
   AND event_time > DATE_SUB(NOW(), INTERVAL 7 DAY)
ORDER BY event_time DESC
LIMIT 20;
" >> $AUDIT_REPORT

echo -e "\n=== 审计完成 ===" >> $AUDIT_REPORT
echo "报告已保存至: $AUDIT_REPORT"

# 发送审计报告（可选）
# mail -s "数据库权限审计报告" admin@example.com < $AUDIT_REPORT

三、SQL注入防护策略

💉 SQL注入威胁：

OWASP Top 10 2023中，注入攻击（包括SQL注入）仍然位列前三。攻击者通过注入恶意SQL代码，可以窃取、篡改或破坏数据库数据。

1. SQL注入攻击示例

-- 易受SQL注入攻击的代码（Python示例）
import mysql.connector

def unsafe_login(username, password):
    """不安全的登录验证方式"""
    conn = mysql.connector.connect(host="localhost", database="app_db", user="app_user")
    cursor = conn.cursor()
    
    # 危险：直接拼接用户输入到SQL语句中
    query = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'"
    
    cursor.execute(query)
    result = cursor.fetchone()
    
    cursor.close()
    conn.close()
    return result

# 攻击者输入示例：
# username: admin' --
# password: anything
# 生成的SQL: SELECT * FROM users WHERE username = 'admin' --' AND password = 'anything'
# 注释符(--)使密码检查失效，攻击者可以以admin身份登录

# 更危险的攻击：
# username: admin' OR '1'='1
# password: ' OR '1'='1
# 生成的SQL: SELECT * FROM users WHERE username = 'admin' OR '1'='1' AND password = '' OR '1'='1'
# 这将返回所有用户记录

2. 防护措施：参数化查询

# 安全的参数化查询示例（Python）
import mysql.connector

def safe_login(username, password):
    """使用参数化查询的安全登录验证"""
    conn = mysql.connector.connect(host="localhost", database="app_db", user="app_user")
    cursor = conn.cursor(prepared=True)  # 启用预处理语句
    
    # 安全：使用参数化查询
    query = "SELECT * FROM users WHERE username = %s AND password = %s"
    
    # 参数会被自动转义，防止SQL注入
    cursor.execute(query, (username, password))
    result = cursor.fetchone()
    
    cursor.close()
    conn.close()
    return result

# 其他编程语言的参数化查询示例：

# PHP (PDO)
# $stmt = $pdo->prepare("SELECT * FROM users WHERE username = :username AND password = :password");
# $stmt->execute(['username' => $username, 'password' => $password]);

# Java (JDBC)
# String query = "SELECT * FROM users WHERE username = ? AND password = ?";
# PreparedStatement stmt = conn.prepareStatement(query);
# stmt.setString(1, username);
# stmt.setString(2, password);

# Node.js (mysql2)
# const [rows] = await connection.execute(
#     'SELECT * FROM users WHERE username = ? AND password = ?',
#     [username, password]
# );

3. 输入验证与过滤

import re

def validate_input(input_string, input_type):
    """输入验证函数"""
    
    validation_rules = {
        'username': {
            'pattern': r'^[a-zA-Z0-9_]{3,20}$',
            'message': '用户名必须是3-20位的字母、数字或下划线'
        },
        'email': {
            'pattern': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
            'message': '请输入有效的邮箱地址'
        },
        'phone': {
            'pattern': r'^1[3-9]\d{9}$',
            'message': '请输入有效的手机号码'
        },
        'integer': {
            'pattern': r'^\d+$',
            'message': '请输入有效的整数'
        },
        'float': {
            'pattern': r'^\d+(\.\d+)?$',
            'message': '请输入有效的浮点数'
        }
    }
    
    if input_type not in validation_rules:
        raise ValueError(f"未知的输入类型: {input_type}")
    
    rule = validation_rules[input_type]
    
    if not re.match(rule['pattern'], input_string):
        raise ValueError(rule['message'])
    
    return True

def sanitize_sql_input(input_string):
    """SQL输入清理（额外的安全层）"""
    # 移除或转义危险字符
    dangerous_patterns = [
        (r'--', ''),           # SQL注释
        (r';', '\\;'),         # 语句分隔符
        (r"'", "\\'"),         # 单引号
        (r'"', '\\"'),         # 双引号
        (r'/', '\\/'),         # 斜杠
        (r'\\', '\\\\'),       # 反斜杠
        (r'%', '\\%'),         # 通配符
        (r'_', '\\_'),         # 通配符
        (r'\(', '\\('),        # 括号
        (r'\)', '\\)'),        # 括号
        (r'union\s+select', 'union select'),  # 阻止union select
        (r'select.*from', 'select from'),     # 阻止select from
        (r'insert\s+into', 'insert into'),    # 阻止insert into
        (r'drop\s+table', 'drop table'),      # 阻止drop table
        (r'delete\s+from', 'delete from'),    # 阻止delete from
        (r'update\s+set', 'update set'),      # 阻止update set
    ]
    
    sanitized = input_string
    for pattern, replacement in dangerous_patterns:
        sanitized = re.sub(pattern, replacement, sanitized, flags=re.IGNORECASE)
    
    return sanitized

# 使用示例
try:
    username = "admin'; DROP TABLE users; --"
    
    # 1. 验证输入
    validate_input(username, 'username')  # 这会失败，因为包含非法字符
    
    # 2. 清理输入（即使验证通过也执行）
    safe_username = sanitize_sql_input(username)
    print(f"清理后的用户名: {safe_username}")
    
except ValueError as e:
    print(f"输入验证失败: {e}")
    
# 注意：输入清理不能替代参数化查询，应作为额外的安全层使用

4. Web应用防火墙（WAF）规则

# Nginx + ModSecurity WAF配置示例
# 防止SQL注入攻击的规则

# 1. 启用ModSecurity
load_module modules/ngx_http_modsecurity_module.so;

http {
    modsecurity on;
    modsecurity_rules_file /etc/nginx/modsec/main.conf;
    
    server {
        listen 80;
        server_name example.com;
        
        location / {
            # 启用ModSecurity
            modsecurity on;
            
            # SQL注入检测规则
            modsecurity_rules '
                # 检测SQL注入关键词
                SecRule ARGS|ARGS_NAMES|REQUEST_BODY|REQUEST_HEADERS 
                    "@rx (union\s+select|select\s+.*from|insert\s+into|drop\s+table|delete\s+from|update\s+set|create\s+table|alter\s+table|truncate\s+table)" 
                    "id:1001,phase:2,deny,status:403,msg:\'SQL Injection Attempt\',logdata:\'%{MATCHED_VAR}\'"
                
                # 检测SQL注释符
                SecRule ARGS|ARGS_NAMES|REQUEST_BODY|REQUEST_HEADERS 
                    "@rx (--|#|/\*|\*/)" 
                    "id:1002,phase:2,deny,status:403,msg:\'SQL Comment Detection\',logdata:\'%{MATCHED_VAR}\'"
                
                # 检测SQL语句分隔符
                SecRule ARGS|ARGS_NAMES|REQUEST_BODY|REQUEST_HEADERS 
                    "@rx (;|\`)" 
                    "id:1003,phase:2,deny,status:403,msg:\'SQL Statement Delimiter\',logdata:\'%{MATCHED_VAR}\'"
                
                # 检测SQL函数调用
                SecRule ARGS|ARGS_NAMES|REQUEST_BODY|REQUEST_HEADERS 
                    "@rx (database\(\)|version\(\)|user\(\)|sleep\(|benchmark\(|load_file\(|into\s+outfile|into\s+dumpfile)" 
                    "id:1004,phase:2,deny,status:403,msg:\'SQL Function Detection\',logdata:\'%{MATCHED_VAR}\'"
                
                # 检测SQL错误信息泄露（防止信息泄露）
                SecRule RESPONSE_BODY 
                    "@rx (SQL syntax|MySQL server version|You have an error in your SQL syntax|Unknown column|Table \'.*\' doesn\'t exist)" 
                    "id:1005,phase:4,deny,status:403,msg:\'SQL Error Information Leak\',logdata:\'%{MATCHED_VAR}\'"
                
                # 限制请求参数长度（防止过长的注入payload）
                SecRule &ARGS "@gt 20" 
                    "id:1006,phase:1,deny,status:403,msg:\'Too many parameters\'"
                
                SecRule ARGS "@gt 1000" 
                    "id:1007,phase:1,deny,status:403,msg:\'Parameter value too long\'"
            ';
        }
    }
}

四、数据加密技术应用

🔒 加密策略：

根据数据敏感程度采用不同的加密策略：传输加密、存储加密、字段级加密、全盘加密等。

1. 数据传输加密（SSL/TLS）

# MySQL SSL/TLS 配置示例

# 1. 生成SSL证书和密钥
# 生成CA证书
openssl genrsa 2048 > ca-key.pem
openssl req -new -x509 -nodes -days 3650 -key ca-key.pem -out ca-cert.pem

# 生成服务器证书
openssl req -newkey rsa:2048 -days 3650 -nodes -keyout server-key.pem -out server-req.pem
openssl rsa -in server-key.pem -out server-key.pem
openssl x509 -req -in server-req.pem -days 3650 -CA ca-cert.pem -CAkey ca-key.pem -set_serial 01 -out server-cert.pem

# 生成客户端证书
openssl req -newkey rsa:2048 -days 3650 -nodes -keyout client-key.pem -out client-req.pem
openssl rsa -in client-key.pem -out client-key.pem
openssl x509 -req -in client-req.pem -days 3650 -CA ca-cert.pem -CAkey ca-key.pem -set_serial 01 -out client-cert.pem

# 2. MySQL服务器SSL配置 (my.cnf)
[mysqld]
ssl-ca=/etc/mysql/ssl/ca-cert.pem
ssl-cert=/etc/mysql/ssl/server-cert.pem
ssl-key=/etc/mysql/ssl/server-key.pem

# 要求所有远程连接使用SSL
require_secure_transport=ON

# 3. 创建要求SSL的用户
CREATE USER 'secure_user'@'%' IDENTIFIED BY 'StrongPassword123!' REQUIRE SSL;
GRANT ALL PRIVILEGES ON secure_db.* TO 'secure_user'@'%';

# 4. 客户端连接示例
mysql --ssl-ca=/path/to/ca-cert.pem \
      --ssl-cert=/path/to/client-cert.pem \
      --ssl-key=/path/to/client-key.pem \
      -h db.example.com -u secure_user -p

# 5. 验证SSL连接状态
SHOW VARIABLES LIKE '%ssl%';
SHOW STATUS LIKE 'Ssl_cipher';
SHOW STATUS LIKE 'Ssl_version';

# 6. 检查当前连接的SSL状态
SELECT * FROM performance_schema.session_status 
WHERE VARIABLE_NAME IN ('Ssl_version', 'Ssl_cipher', 'Ssl_cipher_list');

2. 数据存储加密

# Python 数据加密示例
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2
import base64
import os

class DatabaseEncryptor:
    """数据库字段级加密工具类"""
    
    def __init__(self, master_key=None):
        """
        初始化加密器
        :param master_key: 主密钥，如果为None则生成新密钥
        """
        if master_key:
            self.master_key = master_key
        else:
            # 生成安全的主密钥
            self.master_key = Fernet.generate_key()
        
        self.cipher = Fernet(self.master_key)
    
    def encrypt_field(self, plaintext):
        """加密单个字段"""
        if plaintext is None:
            return None
        
        # 将字符串转换为字节
        if isinstance(plaintext, str):
            plaintext = plaintext.encode('utf-8')
        
        # 加密数据
        encrypted = self.cipher.encrypt(plaintext)
        
        # 返回Base64编码的字符串，便于数据库存储
        return base64.b64encode(encrypted).decode('utf-8')
    
    def decrypt_field(self, encrypted_text):
        """解密单个字段"""
        if encrypted_text is None:
            return None
        
        try:
            # 解码Base64字符串
            encrypted_bytes = base64.b64decode(encrypted_text.encode('utf-8'))
            
            # 解密数据
            decrypted = self.cipher.decrypt(encrypted_bytes)
            
            # 将字节转换回字符串
            return decrypted.decode('utf-8')
        except Exception as e:
            # 记录解密错误，但不暴露具体信息
            raise ValueError("解密失败，可能是密钥不正确或数据已损坏")
    
    def encrypt_sensitive_data(self, data_dict, sensitive_fields):
        """
        加密字典中的敏感字段
        :param data_dict: 包含数据的字典
        :param sensitive_fields: 需要加密的字段列表
        :return: 加密后的字典
        """
        encrypted_dict = data_dict.copy()
        
        for field in sensitive_fields:
            if field in encrypted_dict:
                encrypted_dict[field] = self.encrypt_field(encrypted_dict[field])
        
        return encrypted_dict
    
    def decrypt_sensitive_data(self, data_dict, sensitive_fields):
        """
        解字典中的敏感字段
        :param data_dict: 包含加密数据的字典
        :param sensitive_fields: 需要解密的字段列表
        :return: 解密后的字典
        """
        decrypted_dict = data_dict.copy()
        
        for field in sensitive_fields:
            if field in decrypted_dict:
                decrypted_dict[field] = self.decrypt_field(decrypted_dict[field])
        
        return decrypted_dict
    
    @staticmethod
    def generate_key_from_password(password, salt=None):
        """从密码派生加密密钥"""
        if salt is None:
            salt = os.urandom(16)  # 生成随机盐
        
        # 使用PBKDF2从密码派生密钥
        kdf = PBKDF2(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100000,
        )
        
        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
        return key, salt

# 使用示例
if __name__ == "__main__":
    # 1. 创建加密器
    encryptor = DatabaseEncryptor()
    
    # 2. 敏感数据
    sensitive_data = {
        'id': 1,
        'name': '张三',
        'email': 'zhangsan@example.com',
        'phone': '13800138000',
        'id_card': '110101199001011234',
        'bank_card': '6228480012345678901'
    }
    
    # 3. 定义需要加密的字段
    sensitive_fields = ['phone', 'id_card', 'bank_card']
    
    # 4. 加密敏感字段
    encrypted_data = encryptor.encrypt_sensitive_data(sensitive_data, sensitive_fields)
    print("加密后的数据:", encrypted_data)
    
    # 5. 解密数据
    decrypted_data = encryptor.decrypt_sensitive_data(encrypted_data, sensitive_fields)
    print("解密后的数据:", decrypted_data)
    
    # 6. 保存主密钥（安全存储！）
    print(f"主密钥（妥善保存）: {encryptor.master_key.decode()}")

3. MySQL透明数据加密（TDE）

# MySQL 8.0 透明数据加密配置

# 1. 安装TDE组件（企业版功能）
INSTALL COMPONENT "file://component_encryption";

# 2. 查看TDE状态
SELECT * FROM information_schema.`PLUGINS` 
WHERE PLUGIN_NAME LIKE '%keyring%';

# 3. 配置keyring插件（选择一种）
# 选项1: keyring_file（文件方式）
[mysqld]
early-plugin-load=keyring_file.so
keyring_file_data=/var/lib/mysql-keyring/keyring

# 选项2: keyring_encrypted_file（加密文件方式）
[mysqld]
early-plugin-load=keyring_encrypted_file.so
keyring_encrypted_file_data=/var/lib/mysql-keyring/keyring
keyring_encrypted_file_password=StrongKeyringPass123!

# 4. 创建加密表空间
CREATE TABLESPACE `encrypted_ts` 
ADD DATAFILE 'encrypted_ts.ibd' 
ENGINE=InnoDB 
ENCRYPTION='Y';

# 5. 创建使用加密表空间的表
CREATE TABLE sensitive_data (
    id INT PRIMARY KEY AUTO_INCREMENT,
    card_number VARCHAR(19) NOT NULL,
    card_holder VARCHAR(100) NOT NULL,
    expiry_date DATE NOT NULL,
    cvv VARCHAR(4) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) TABLESPACE encrypted_ts;

# 6. 加密现有表
ALTER TABLE existing_sensitive_table ENCRYPTION='Y';

# 7. 查看加密状态
SELECT 
    TABLE_SCHEMA,
    TABLE_NAME,
    CREATE_OPTIONS
FROM information_schema.TABLES 
WHERE CREATE_OPTIONS LIKE '%ENCRYPTION%';

# 8. 表空间加密信息
SELECT 
    SPACE, 
    NAME, 
    SPACE_TYPE, 
    ENCRYPTION 
FROM information_schema.INNODB_TABLESPACES 
WHERE ENCRYPTION = 'Y';

# 9. 备份加密表空间（需要特别处理）
# 备份前需要锁定表
FLUSH TABLES sensitive_data FOR EXPORT;
# 复制.ibd和.cfg文件
# 完成后解锁
UNLOCK TABLES;

# 10. 监控加密性能
SHOW STATUS LIKE 'keyring%';
SHOW GLOBAL STATUS LIKE 'innodb_encryption%';

# 注意事项：
# 1. TDE只加密数据文件，不加密日志文件、临时文件或内存中的数据
# 2. 备份和恢复需要特殊处理
# 3. 加密会影响性能（通常5-10%）
# 4. 需要企业版MySQL
# 5. 密钥管理是关键，确保密钥安全

五、定期备份与恢复策略

💾 备份原则（3-2-1规则）：

至少保存3份备份，使用2种不同介质，其中1份存放在异地。这是数据保护的最佳实践。

1. 数据库备份流程

1

📋 确定备份策略（全量/增量/差异）

2

⏰ 设置备份计划（自动化调度）

3

🔐 执行加密备份（保护备份数据）

4

📊 验证备份完整性（定期测试）

5

🌍 异地存储备份（灾难恢复）

6

📈 监控备份状态（及时告警）

2. 自动化备份脚本

#!/bin/bash
# mysql_automated_backup.sh
# MySQL自动化备份脚本

# 配置参数
DB_HOST="localhost"
DB_PORT="3306"
DB_USER="backup_user"
DB_PASS="BackupPass123!"
BACKUP_DIR="/data/backups/mysql"
RETENTION_DAYS=30
ENCRYPT_KEY="/etc/mysql/backup_key.pem"
LOG_FILE="/var/log/mysql_backup.log"
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_NAME="backup_${DATE}"

# 创建备份目录
mkdir -p ${BACKUP_DIR}/daily
mkdir -p ${BACKUP_DIR}/weekly
mkdir -p ${BACKUP_DIR}/monthly

# 日志函数
log_message() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a $LOG_FILE
}

# 错误处理函数
handle_error() {
    log_message "ERROR: $1"
    # 发送告警邮件
    echo "MySQL备份失败: $1" | mail -s "MySQL备份失败告警" admin@example.com
    exit 1
}

# 检查MySQL连接
check_mysql_connection() {
    if ! mysqladmin -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS ping > /dev/null 2>&1; then
        handle_error "无法连接到MySQL服务器"
    fi
    log_message "MySQL连接检查成功"
}

# 执行全量备份
perform_full_backup() {
    log_message "开始执行全量备份..."
    
    # 获取所有数据库（排除系统数据库）
    DATABASES=$(mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS -e "SHOW DATABASES;" | grep -Ev "(Database|information_schema|performance_schema|mysql|sys)")
    
    for DB in $DATABASES; do
        log_message "备份数据库: $DB"
        
        # 备份单个数据库
        BACKUP_FILE="${BACKUP_DIR}/daily/${BACKUP_NAME}_${DB}.sql"
        
        # 使用mysqldump备份
        if ! mysqldump -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS \
            --single-transaction \
            --routines \
            --triggers \
            --events \
            --add-drop-database \
            --databases "$DB" > "$BACKUP_FILE"; then
            handle_error "备份数据库 $DB 失败"
        fi
        
        # 压缩备份文件
        log_message "压缩备份文件: $BACKUP_FILE"
        gzip "$BACKUP_FILE"
        
        # 加密备份文件（可选）
        if [ -f "$ENCRYPT_KEY" ]; then
            log_message "加密备份文件"
            ENCRYPTED_FILE="${BACKUP_FILE}.gz.enc"
            openssl enc -aes-256-cbc -salt -in "${BACKUP_FILE}.gz" -out "$ENCRYPTED_FILE" -pass file:"$ENCRYPT_KEY"
            
            # 删除未加密的压缩文件
            rm "${BACKUP_FILE}.gz"
            log_message "加密完成: $ENCRYPTED_FILE"
        fi
        
        # 验证备份文件
        if [ -f "${BACKUP_FILE}.gz" ] || [ -f "$ENCRYPTED_FILE" ]; then
            log_message "数据库 $DB 备份成功"
        else
            handle_error "数据库 $DB 备份文件创建失败"
        fi
    done
    
    log_message "全量备份完成"
}

# 执行增量备份（需要启用二进制日志）
perform_incremental_backup() {
    log_message "开始执行增量备份..."
    
    # 刷新二进制日志
    mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS -e "FLUSH BINARY LOGS;"
    
    # 获取当前二进制日志文件
    CURRENT_BINLOG=$(mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASS -e "SHOW MASTER STATUS\G" | grep "File:" | awk '{print $2}')
    
    # 备份二进制日志
    BINLOG_BACKUP_DIR="${BACKUP_DIR}/binlogs"
    mkdir -p $BINLOG_BACKUP_DIR
    
    # 复制二进制日志文件（除了当前正在写的）
    for binlog in $(ls /var/lib/mysql/mysql-bin.* 2>/dev/null); do
        if [ "$(basename $binlog)" != "$CURRENT_BINLOG" ]; then
            log_message "备份二进制日志: $(basename $binlog)"
            cp "$binlog" "$BINLOG_BACKUP_DIR/"
        fi
    done
    
    log_message "增量备份完成"
}

# 清理旧备份
cleanup_old_backups() {
    log_message "清理超过 ${RETENTION_DAYS} 天的旧备份..."
    
    # 清理每日备份
    find ${BACKUP_DIR}/daily -name "backup_*.sql.gz" -mtime +${RETENTION_DAYS} -delete
    find ${BACKUP_DIR}/daily -name "backup_*.sql.gz.enc" -mtime +${RETENTION_DAYS} -delete
    
    # 清理二进制日志备份
    find ${BACKUP_DIR}/binlogs -name "mysql-bin.*" -mtime +${RETENTION_DAYS} -delete
    
    log_message "旧备份清理完成"
}

# 验证备份完整性
verify_backup_integrity() {
    log_message "验证备份完整性..."
    
    # 随机选择一个备份文件进行验证
    BACKUP_FILE=$(find ${BACKUP_DIR}/daily -name "*.sql.gz" -type f | head -1)
    
    if [ -n "$BACKUP_FILE" ]; then
        # 如果是加密文件，先解密
        if [[ "$BACKUP_FILE" == *.enc ]]; then
            TEMP_FILE="/tmp/backup_verify.sql.gz"
            openssl enc -aes-256-cbc -d -in "$BACKUP_FILE" -out "$TEMP_FILE" -pass file:"$ENCRYPT_KEY" 2>/dev/null
            
            if [ $? -eq 0 ]; then
                # 验证gzip文件完整性
                if gzip -t "$TEMP_FILE" 2>/dev/null; then
                    log_message "备份文件完整性验证成功: $BACKUP_FILE"
                    rm "$TEMP_FILE"
                else
                    log_message "WARNING: 备份文件可能损坏: $BACKUP_FILE"
                fi
            else
                log_message "WARNING: 无法解密备份文件: $BACKUP_FILE"
            fi
        else
            # 直接验证gzip文件
            if gzip -t "$BACKUP_FILE" 2>/dev/null; then
                log_message "备份文件完整性验证成功: $BACKUP_FILE"
            else
                log_message "WARNING: 备份文件可能损坏: $BACKUP_FILE"
            fi
        fi
    else
        log_message "没有找到备份文件进行验证"
    fi
}

# 生成备份报告
generate_backup_report() {
    log_message "生成备份报告..."
    
    REPORT_FILE="/tmp/backup_report_${DATE}.txt"
    
    echo "=== MySQL备份报告 ===" > $REPORT_FILE
    echo "生成时间: $(date)" >> $REPORT_FILE
    echo "备份目录: $BACKUP_DIR" >> $REPORT_FILE
    echo "=====================" >> $REPORT_FILE
    
    echo -e "\n备份文件统计:" >> $REPORT_FILE
    echo "每日备份文件数: $(find ${BACKUP_DIR}/daily -type f | wc -l)" >> $REPORT_FILE
    echo "二进制日志文件数: $(find ${BACKUP_DIR}/binlogs -type f | wc -l)" >> $REPORT_FILE
    echo "备份总大小: $(du -sh $BACKUP_DIR | awk '{print $1}')" >> $REPORT_FILE
    
    echo -e "\n最近备份文件:" >> $REPORT_FILE
    find ${BACKUP_DIR}/daily -type f -name "*.gz" -o -name "*.enc" | sort -r | head -5 | while read file; do
        echo "  $(basename $file) - $(stat -c %y $file | cut -d'.' -f1) - $(du -h $file | awk '{print $1}')" >> $REPORT_FILE
    done
    
    # 发送报告邮件
    mail -s "MySQL备份报告 ${DATE}" admin@example.com < $REPORT_FILE
    log_message "备份报告已发送"
    
    rm $REPORT_FILE
}

# 主函数
main() {
    log_message "=== MySQL自动化备份开始 ==="
    
    # 1. 检查MySQL连接
    check_mysql_connection
    
    # 2. 判断备份类型
    DAY_OF_WEEK=$(date +%u)  # 1=周一, 7=周日
    DAY_OF_MONTH=$(date +%d)
    
    if [ "$DAY_OF_WEEK" = "1" ]; then
        # 每周一执行全量备份
        log_message "周一执行全量备份"
        perform_full_backup
        
        # 复制到周备份目录
        cp ${BACKUP_DIR}/daily/backup_*.gz ${BACKUP_DIR}/weekly/ 2>/dev/null || true
        cp ${BACKUP_DIR}/daily/backup_*.gz.enc ${BACKUP_DIR}/weekly/ 2>/dev/null || true
        
    elif [ "$DAY_OF_MONTH" = "01" ]; then
        # 每月1号执行全量备份
        log_message "每月1号执行全量备份"
        perform_full_backup
        
        # 复制到月备份目录
        cp ${BACKUP_DIR}/daily/backup_*.gz ${BACKUP_DIR}/monthly/ 2>/dev/null || true
        cp ${BACKUP_DIR}/daily/backup_*.gz.enc ${BACKUP_DIR}/monthly/ 2>/dev/null || true
        
    else
        # 其他时间执行增量备份
        log_message "执行增量备份"
        perform_incremental_backup
    fi
    
    # 3. 清理旧备份
    cleanup_old_backups
    
    # 4. 验证备份完整性
    verify_backup_integrity
    
    # 5. 生成备份报告
    generate_backup_report
    
    log_message "=== MySQL自动化备份完成 ==="
}

# 执行主函数
main

3. 数据库恢复策略

故障类型	恢复策略	恢复时间目标 (RTO)	恢复点目标 (RPO)	恢复步骤
单表损坏	表级恢复	30分钟	15分钟	1. 从备份恢复表 2. 应用binlog增量
单库损坏	库级恢复	1小时	15分钟	1. 恢复完整备份 2. 应用增量日志
服务器故障	全实例恢复	2-4小时	1小时	1. 新服务器准备 2. 恢复所有数据
数据中心故障	异地灾备切换	4-8小时	2小时	1. 激活灾备中心 2. DNS切换
逻辑错误（误删）	时间点恢复	2小时	0（精确恢复）	1. 恢复到误操作前 2. 跳过错误事务

#!/bin/bash
# mysql_recovery_script.sh
# MySQL数据库恢复脚本

# 配置参数
BACKUP_DIR="/data/backups/mysql"
RESTORE_DIR="/tmp/mysql_restore"
MYSQL_DATA_DIR="/var/lib/mysql"
LOG_FILE="/var/log/mysql_recovery.log"
DATE_TO_RESTORE="2025-12-10 14:30:00"  # 恢复到的时间点

# 日志函数
log_message() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a $LOG_FILE
}

# 错误处理
handle_error() {
    log_message "ERROR: $1"
    exit 1
}

# 准备工作
prepare_restore() {
    log_message "准备恢复环境..."
    
    # 停止MySQL服务
    systemctl stop mysql
    if [ $? -ne 0 ]; then
        handle_error "停止MySQL服务失败"
    fi
    
    # 备份当前数据目录
    if [ -d "$MYSQL_DATA_DIR" ]; then
        log_message "备份当前数据目录..."
        BACKUP_OLD_DATA="/tmp/mysql_data_backup_$(date +%Y%m%d_%H%M%S)"
        cp -r "$MYSQL_DATA_DIR" "$BACKUP_OLD_DATA"
        log_message "当前数据已备份到: $BACKUP_OLD_DATA"
    fi
    
    # 清理数据目录
    log_message "清理数据目录..."
    rm -rf ${MYSQL_DATA_DIR}/*
    
    # 创建恢复目录
    mkdir -p "$RESTORE_DIR"
}

# 恢复全量备份
restore_full_backup() {
    log_message "恢复全量备份..."
    
    # 查找最新的全量备份
    LATEST_BACKUP=$(find ${BACKUP_DIR}/daily -name "backup_*.sql.gz" -type f | sort -r | head -1)
    
    if [ -z "$LATEST_BACKUP" ]; then
        handle_error "未找到全量备份文件"
    fi
    
    log_message "使用备份文件: $LATEST_BACKUP"
    
    # 解压备份文件
    gunzip -c "$LATEST_BACKUP" > "${RESTORE_DIR}/full_backup.sql"
    
    if [ $? -ne 0 ]; then
        handle_error "解压备份文件失败"
    fi
    
    # 恢复备份
    log_message "执行全量备份恢复..."
    mysql < "${RESTORE_DIR}/full_backup.sql"
    
    if [ $? -ne 0 ]; then
        handle_error "全量备份恢复失败"
    fi
    
    log_message "全量备份恢复完成"
}

# 应用增量备份（二进制日志）
apply_incremental_backup() {
    log_message "应用增量备份..."
    
    # 查找全量备份的时间
    BACKUP_TIMESTAMP=$(echo "$LATEST_BACKUP" | grep -o '[0-9]\{8\}_[0-9]\{6\}')
    BACKUP_DATETIME="${BACKUP_TIMESTAMP:0:4}-${BACKUP_TIMESTAMP:4:2}-${BACKUP_TIMESTAMP:6:2} ${BACKUP_TIMESTAMP:9:2}:${BACKUP_TIMESTAMP:11:2}:${BACKUP_TIMESTAMP:13:2}"
    
    log_message "全量备份时间: $BACKUP_DATETIME"
    log_message "恢复到时间点: $DATE_TO_RESTORE"
    
    # 生成二进制日志列表
    BINLOG_FILES=$(find ${BACKUP_DIR}/binlogs -name "mysql-bin.*" -type f | sort)
    
    if [ -n "$BINLOG_FILES" ]; then
        log_message "应用二进制日志..."
        
        # 合并所有二进制日志
        for binlog in $BINLOG_FILES; do
            log_message "处理二进制日志: $(basename $binlog)"
            mysqlbinlog "$binlog" >> "${RESTORE_DIR}/all_binlogs.sql"
        done
        
        # 应用二进制日志（到指定时间点）
        mysqlbinlog --stop-datetime="$DATE_TO_RESTORE" \
                    --start-datetime="$BACKUP_DATETIME" \
                    ${BACKUP_DIR}/binlogs/mysql-bin.* > "${RESTORE_DIR}/incremental.sql"
        
        if [ -s "${RESTORE_DIR}/incremental.sql" ]; then
            mysql < "${RESTORE_DIR}/incremental.sql"
            if [ $? -ne 0 ]; then
                handle_error "应用增量备份失败"
            fi
            log_message "增量备份应用完成"
        else
            log_message "没有需要应用的增量备份"
        fi
    else
        log_message "没有找到二进制日志备份"
    fi
}

# 恢复后处理
post_restore() {
    log_message "执行恢复后处理..."
    
    # 修复权限
    chown -R mysql:mysql "$MYSQL_DATA_DIR"
    
    # 启动MySQL服务
    systemctl start mysql
    if [ $? -ne 0 ]; then
        handle_error "启动MySQL服务失败"
    fi
    
    # 检查MySQL状态
    sleep 5
    if mysqladmin ping > /dev/null 2>&1; then
        log_message "MySQL服务启动成功"
    else
        handle_error "MySQL服务启动后检查失败"
    fi
    
    # 验证数据完整性
    log_message "验证数据完整性..."
    mysql -e "SHOW DATABASES;"
    
    # 清理临时文件
    rm -rf "$RESTORE_DIR"
    
    log_message "数据库恢复完成"
}

# 主函数
main() {
    log_message "=== 开始数据库恢复 ==="
    
    # 1. 准备工作
    prepare_restore
    
    # 2. 恢复全量备份
    restore_full_backup
    
    # 3. 应用增量备份
    apply_incremental_backup
    
    # 4. 恢复后处理
    post_restore
    
    log_message "=== 数据库恢复完成 ==="
}

# 执行恢复（需要手动确认）
read -p "确认要恢复数据库吗？这将覆盖当前数据。(y/n): " -n 1 -r
echo
if [[ $REPLY =~ ^[Yy]$ ]]; then
    main
else
    echo "恢复操作已取消"
fi

六、监控与审计机制

1. 数据库活动审计

# MySQL 审计配置

# 1. 启用通用查询日志（生产环境慎用，性能影响大）
[mysqld]
general_log = 1
general_log_file = /var/log/mysql/general.log
log_output = FILE

# 2. 启用慢查询日志
slow_query_log = 1
slow_query_log_file = /var/log/mysql/slow.log
long_query_time = 2  # 超过2秒的查询
log_queries_not_using_indexes = 1  # 记录未使用索引的查询

# 3. 启用二进制日志（用于复制和恢复）
log_bin = /var/log/mysql/mysql-bin.log
expire_logs_days = 30
max_binlog_size = 100M
binlog_format = ROW  # ROW格式更安全，记录行级变更

# 4. 启用错误日志
log_error = /var/log/mysql/error.log

# 5. MySQL企业版审计插件（企业版功能）
INSTALL PLUGIN audit_log SONAME 'audit_log.so';

[mysqld]
audit_log_format = JSON
audit_log_policy = ALL  # 记录所有事件
audit_log_file = /var/log/mysql/audit.log
audit_log_rotate_on_size = 100M
audit_log_rotations = 10

# 6. 查看当前审计配置
SHOW VARIABLES LIKE '%audit%';

# 7. 自定义审计规则示例
# 审计敏感操作
SET GLOBAL audit_log_filter = '{
    "filter": {
        "class": {
            "name": "general",
            "event": {
                "name": "status",
                "log": true
            }
        }
    }
}';

# 8. 第三方审计工具：MySQL Enterprise Audit
# 提供更强大的审计功能：
# - 细粒度审计策略
# - 实时监控
# - 合规报告
# - 数据脱敏

# 9. 使用Percona审计插件（开源替代）
INSTALL PLUGIN audit_log SONAME 'audit_log.so';

[mysqld]
audit_log_format = JSON
audit_log_handler = FILE
audit_log_file = /var/log/mysql/audit.log
audit_log_policy = ALL
audit_log_rotate_on_size = 100M
audit_log_rotations = 10

# 10. 审计日志分析脚本示例
cat > analyze_audit_log.sh << 'EOF'
#!/bin/bash
# MySQL审计日志分析脚本

AUDIT_LOG="/var/log/mysql/audit.log"
REPORT_FILE="/tmp/audit_report_$(date +%Y%m%d).txt"

echo "=== MySQL审计报告 ===" > $REPORT_FILE
echo "生成时间: $(date)" >> $REPORT_FILE
echo "=====================" >> $REPORT_FILE

# 分析登录失败
echo -e "\n1. 登录失败统计:" >> $REPORT_FILE
grep '"command":"Connect"' $AUDIT_LOG | grep '"error":"1045"' | wc -l >> $REPORT_FILE

# 分析敏感操作
echo -e "\n2. 敏感操作统计:" >> $REPORT_FILE
echo "DROP操作: $(grep -i 'DROP' $AUDIT_LOG | wc -l)" >> $REPORT_FILE
echo "TRUNCATE操作: $(grep -i 'TRUNCATE' $AUDIT_LOG | wc -l)" >> $REPORT_FILE
echo "GRANT操作: $(grep -i 'GRANT' $AUDIT_LOG | wc -l)" >> $REPORT_FILE
echo "REVOKE操作: $(grep -i 'REVOKE' $AUDIT_LOG | wc -l)" >> $REPORT_FILE

# 分析用户活动
echo -e "\n3. 用户活动排名:" >> $REPORT_FILE
grep '"command":"Query"' $AUDIT_LOG | grep -o '"user":"[^"]*"' | sort | uniq -c | sort -rn | head -10 >> $REPORT_FILE

# 发送报告
mail -s "MySQL审计报告" admin@example.com < $REPORT_FILE
EOF

chmod +x analyze_audit_log.sh

2. 实时监控与告警

# Prometheus + Grafana MySQL监控配置

# 1. 安装MySQL exporter
# 下载地址：https://github.com/prometheus/mysqld_exporter

# 2. 创建监控用户
CREATE USER 'exporter'@'localhost' IDENTIFIED BY 'ExporterPass123!' WITH MAX_USER_CONNECTIONS 3;
GRANT PROCESS, REPLICATION CLIENT, SELECT ON *.* TO 'exporter'@'localhost';

# 3. 配置MySQL exporter
cat > /etc/mysql_exporter.yml << 'EOF'
# MySQL exporter配置
global:
  scrape_interval: 15s

mysql:
  # 监控目标
  - data_source_name: "exporter:ExporterPass123!@(localhost:3306)/"
    # 收集的指标
    collect:
      # 全局状态
      global_status: true
      # 全局变量
      global_variables: true
      # 进程列表
      processlist: true
      # 用户统计
      user_stats: true
      # 表统计
      table_stats: true
      # 索引统计
      index_stats: true
      # 复制状态
      slave_status: true
      # InnoDB状态
      innodb: true
      # 二进制日志
      binlog: true
EOF

# 4. 启动exporter
./mysqld_exporter --config.my-cnf=/etc/mysql_exporter.yml --web.listen-address=":9104"

# 5. Prometheus配置
cat > /etc/prometheus/prometheus.yml << 'EOF'
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'mysql'
    static_configs:
      - targets: ['localhost:9104']
    metrics_path: /metrics
    params:
      format: ['prometheus']
EOF

# 6. 关键监控指标
"""
mysql_global_status_threads_connected   # 当前连接数
mysql_global_status_max_used_connections # 历史最大连接数
mysql_global_status_questions           # 查询总数
mysql_global_status_slow_queries        # 慢查询数
mysql_global_status_innodb_row_lock_time_avg # 平均行锁等待时间
mysql_global_variables_max_connections  # 最大连接数限制
mysql_global_status_innodb_buffer_pool_pages_free # InnoDB缓冲池空闲页
mysql_slave_status_slave_io_running     # 从库IO线程状态
mysql_slave_status_slave_sql_running    # 从库SQL线程状态
"""

# 7. 告警规则（Prometheus rules）
cat > /etc/prometheus/rules/mysql_rules.yml << 'EOF'
groups:
  - name: mysql_alerts
    rules:
      # 连接数过高告警
      - alert: MySQL连接数过高
        expr: mysql_global_status_threads_connected / mysql_global_variables_max_connections > 0.8
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "MySQL连接数超过80%"
          description: "当前连接数: {{ $value }}%，接近最大连接数限制"
      
      # 慢查询过多告警
      - alert: MySQL慢查询过多
        expr: rate(mysql_global_status_slow_queries[5m]) > 10
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "MySQL慢查询数量异常"
          description: "过去5分钟平均每秒慢查询数: {{ $value }}"
      
      # 复制延迟告警
      - alert: MySQL复制延迟
        expr: mysql_slave_status_seconds_behind_master > 30
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "MySQL从库复制延迟超过30秒"
          description: "当前复制延迟: {{ $value }}秒"
      
      # InnoDB缓冲池命中率过低
      - alert: InnoDB缓冲池命中率低
        expr: (1 - (mysql_global_status_innodb_buffer_pool_reads / mysql_global_status_innodb_buffer_pool_read_requests)) * 100 < 90
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "InnoDB缓冲池命中率低于90%"
          description: "当前命中率: {{ $value }}%"
      
      # 死锁检测
      - alert: MySQL死锁频繁
        expr: rate(mysql_global_status_innodb_row_lock_time_avg[5m]) > 1000
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "MySQL死锁检测"
          description: "平均行锁等待时间异常: {{ $value }}ms"
      
      # 磁盘空间不足
      - alert: MySQL磁盘空间不足
        expr: (1 - (node_filesystem_avail_bytes{mountpoint="/var/lib/mysql"} / node_filesystem_size_bytes{mountpoint="/var/lib/mysql"})) * 100 > 85
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "MySQL数据目录磁盘空间不足"
          description: "磁盘使用率: {{ $value }}%"
EOF

# 8. Grafana仪表板
# 导入MySQL监控仪表板：https://grafana.com/grafana/dashboards/7362-mysql-overview/

# 9. 自定义监控脚本
cat > /usr/local/bin/mysql_health_check.sh << 'EOF'
#!/bin/bash
# MySQL健康检查脚本

MYSQL_HOST="localhost"
MYSQL_PORT="3306"
MYSQL_USER="health_check"
MYSQL_PASS="HealthCheck123!"

# 检查MySQL是否运行
check_mysql_running() {
    if ! systemctl is-active --quiet mysql; then
        echo "ERROR: MySQL服务未运行"
        return 1
    fi
    echo "✓ MySQL服务运行正常"
    return 0
}

# 检查MySQL连接
check_mysql_connection() {
    if ! mysqladmin -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASS ping > /dev/null 2>&1; then
        echo "ERROR: 无法连接到MySQL"
        return 1
    fi
    echo "✓ MySQL连接正常"
    return 0
}

# 检查复制状态
check_replication() {
    REPLICATION_STATUS=$(mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASS -e "SHOW SLAVE STATUS\G" 2>/dev/null)
    
    if [ -n "$REPLICATION_STATUS" ]; then
        IO_RUNNING=$(echo "$REPLICATION_STATUS" | grep "Slave_IO_Running:" | awk '{print $2}')
        SQL_RUNNING=$(echo "$REPLICATION_STATUS" | grep "Slave_SQL_Running:" | awk '{print $2}')
        
        if [ "$IO_RUNNING" = "Yes" ] && [ "$SQL_RUNNING" = "Yes" ]; then
            echo "✓ MySQL复制运行正常"
            return 0
        else
            echo "ERROR: MySQL复制异常 (IO: $IO_RUNNING, SQL: $SQL_RUNNING)"
            return 1
        fi
    else
        echo "⚠️  未配置复制"
        return 0
    fi
}

# 检查错误日志
check_error_log() {
    ERROR_LOG="/var/log/mysql/error.log"
    if [ -f "$ERROR_LOG" ]; then
        RECENT_ERRORS=$(tail -100 "$ERROR_LOG" | grep -E "ERROR|Error|error" | tail -5)
        
        if [ -n "$RECENT_ERRORS" ]; then
            echo "⚠️  发现最近错误:"
            echo "$RECENT_ERRORS"
            return 1
        else
            echo "✓ 错误日志正常"
            return 0
        fi
    else
        echo "⚠️  错误日志文件不存在"
        return 0
    fi
}

# 主函数
main() {
    echo "=== MySQL健康检查开始 ==="
    echo "检查时间: $(date)"
    echo ""
    
    # 执行检查
    check_mysql_running
    check_mysql_connection
    check_replication
    check_error_log
    
    echo ""
    echo "=== MySQL健康检查完成 ==="
}

main
EOF

chmod +x /usr/local/bin/mysql_health_check.sh

七、企业级数据库安全架构

🏢 企业级安全架构特点：

多层防护、统一管理、自动化运维、合规审计、容灾备份、安全开发生命周期集成。

1. 分层安全架构设计

1

物理层安全

数据中心访问控制、硬件加密、物理隔离、环境监控

2

网络层安全

防火墙、VPN、网络分段、IDS/IPS、DDoS防护、WAF

3

主机层安全

操作系统加固、最小化安装、定期打补、入侵检测、文件完整性监控

4

数据库层安全

访问控制、数据加密、审计日志、漏洞管理、备份恢复

5

应用层安全

输入验证、参数化查询、会话管理、错误处理、安全编码

6

数据层安全

数据分类、数据脱敏、数据防泄露、数据血缘追踪

2. 数据库安全开发生命周期（DBSDLC）

# 数据库安全开发生命周期（DBSDLC）流程

"""
阶段1: 需求分析
    - 识别敏感数据
    - 定义安全需求
    - 合规性要求分析
    - 数据分类分级

阶段2: 设计阶段
    - 安全架构设计
    - 权限模型设计
    - 加密策略设计
    - 审计方案设计
    - 备份恢复策略

阶段3: 开发阶段
    - 安全编码规范
    - 参数化查询实现
    - 输入验证实现
    - 错误处理设计
    - 安全函数开发

阶段4: 测试阶段
    - 安全功能测试
    - 渗透测试
    - SQL注入测试
    - 权限提升测试
    - 数据泄露测试

阶段5: 部署阶段
    - 安全配置加固
    - 最小权限配置
    - 加密证书部署
    - 监控告警配置
    - 备份恢复测试

阶段6: 运维阶段
    - 定期安全扫描
    - 漏洞管理
    - 日志审计分析
    - 权限定期评审
    - 安全补丁更新

阶段7: 退役阶段
    - 数据安全迁移
    - 数据安全销毁
    - 权限清理
    - 审计记录归档
"""

# DBSDLC检查清单
DBSDLC_CHECKLIST = {
    "需求分析": [
        "敏感数据识别完成",
        "安全需求文档化",
        "合规要求明确",
        "数据分类分级完成"
    ],
    "设计阶段": [
        "安全架构设计评审通过",
        "权限模型符合最小权限原则",
        "加密策略设计完成",
        "审计方案满足合规要求",
        "备份恢复策略制定"
    ],
    "开发阶段": [
        "代码安全扫描通过",
        "所有SQL查询使用参数化",
        "输入验证覆盖所有用户输入",
        "错误处理不泄露敏感信息",
        "安全函数单元测试通过"
    ],
    "测试阶段": [
        "安全功能测试通过",
        "渗透测试无高危漏洞",
        "SQL注入测试通过",
        "权限提升测试通过",
        "数据泄露测试通过"
    ],
    "部署阶段": [
        "安全配置加固完成",
        "最小权限配置验证",
        "加密证书部署完成",
        "监控告警配置测试",
        "备份恢复流程验证"
    ],
    "运维阶段": [
        "定期安全扫描计划",
        "漏洞管理流程建立",
        "日志审计分析机制",
        "权限定期评审计划",
        "安全补丁更新流程"
    ]
}

3. 云数据库安全最佳实践

网络隔离

使用VPC、安全组、网络ACL实现数据库网络隔离，限制访问来源IP。

密钥管理

使用KMS（密钥管理服务）管理加密密钥，实现自动密钥轮换。

版本管理

及时更新数据库版本，应用安全补丁，使用托管服务自动更新。

监控审计

启用云服务商提供的监控审计功能，如AWS CloudTrail、Azure Monitor。

自动备份

配置自动备份和快照，实现跨区域复制，满足合规要求。

身份管理

使用IAM角色和策略管理数据库访问权限，实现最小权限原则。

八、总结与最佳实践

✅ 数据库安全加固核心要点：

1. 最小权限原则：每个用户只能访问必需的数据
2. 纵深防御：多层安全防护，不依赖单一措施
3. 加密保护：传输加密和存储加密相结合
4. 持续监控：实时监控、定期审计、及时响应
5. 备份恢复：定期备份、验证恢复、灾难演练

1. 数据库安全实施路线图

阶段	时间	主要任务	关键成果
评估阶段	1-2周	现状评估、风险分析、合规要求	安全评估报告、风险清单
基础加固	2-4周	访问控制、基础监控、备份配置	权限矩阵、监控基线、备份策略
高级防护	4-8周	数据加密、WAF部署、审计增强	加密方案、WAF规则、审计报告
运维优化	持续进行	自动化运维、持续监控、定期评估	运维手册、监控仪表板、评估报告

2. 常见问题与解决方案

❓ 问题：权限管理复杂，难以维护

解决方案：
1. 实施基于角色的访问控制（RBAC）
2. 使用自动化工具管理权限
3. 定期审计和清理无用权限
4. 建立权限变更审批流程

❓ 问题：加密性能影响业务

解决方案：
1. 使用硬件加密加速
2. 实施分级加密策略
3. 优化加密算法和密钥长度
4. 使用透明数据加密（TDE）

❓ 问题：合规要求难以满足

解决方案：
1. 建立合规框架和控制矩阵
2. 使用自动化合规检查工具
3. 定期进行合规审计
4. 保持与法律团队的沟通

3. 数据库安全检查清单

# 数据库安全检查清单
# 定期（每月/每季度）执行以下检查

"""
1. 访问控制检查
    □ 所有数据库用户都有强密码
    □ 无默认用户或空密码用户
    □ 权限符合最小权限原则
    □ 定期清理离职员工权限
    □ 敏感操作需要多因素认证

2. 网络安全检查
    □ 数据库端口不对外网开放
    □ 防火墙规则限制访问来源
    □ 使用SSL/TLS加密连接
    □ 网络入侵检测系统运行正常
    □ 定期扫描网络漏洞

3. 数据安全检查
    □ 敏感数据已加密存储
    □ 加密密钥安全管理
    □ 数据备份完整且可恢复
    □ 备份数据加密存储
    □ 数据脱敏策略实施

4. 审计与监控检查
    □ 审计日志开启且完整
    □ 定期分析审计日志
    □ 实时监控数据库性能
    □ 设置关键指标告警
    □ 安全事件响应流程

5. 漏洞管理检查
    □ 数据库版本保持最新
    □ 定期进行漏洞扫描
    □ 安全补丁及时应用
    □ 配置符合安全基线
    □ 定期渗透测试

6. 备份恢复检查
    □ 备份策略符合3-2-1原则
    □ 备份文件定期验证
    □ 恢复演练定期进行
    □ 灾难恢复计划更新
    □ 备份加密和访问控制

7. 合规性检查
    □ 符合相关法律法规
    □ 数据保护政策执行
    □ 隐私保护措施到位
    □ 第三方审计通过
    □ 合规文档完整

8. 运维安全检查
    □ 运维操作有审计跟踪
    □ 变更管理流程规范
    □ 应急响应计划完善
    □ 安全培训定期开展
    □ 安全意识持续提升
"""

# 检查结果评分标准
SCORING_CRITERIA = {
    "优秀": "所有检查项通过，无安全问题",
    "良好": "少量低风险问题，已制定修复计划",
    "一般": "存在中风险问题，需要优先处理",
    "差": "存在高风险问题，需要立即处理"
}

4. 未来发展趋势

AI驱动的安全

机器学习异常检测、智能威胁分析、自动化响应。

同态加密

在加密数据上直接计算，保护数据在处理过程中的安全。

区块链审计

使用区块链技术实现不可篡改的审计日志。

零信任架构

不信任任何内部或外部实体，持续验证访问权限。

🎯 最后建议：

1. 从基础开始：先解决高风险问题，再逐步完善
2. 持续改进：安全是持续过程，不是一次性项目
3. 全员参与：安全不仅是技术问题，需要全员安全意识
4. 保持学习：安全威胁不断演变，需要持续学习新知识
5. 定期演练：定期进行安全演练和恢复测试，确保预案有效