实验概述

• 动态脱敏：在数据使用（查询、导出）时，根据用户角色或访问场景，实时对敏感数据进行脱敏处理，原始数据不改变，仅向用户展示脱敏后的数据，实现“按需可见”；
• 视图脱敏：通过创建数据库视图，在视图中定义脱敏规则，用户查询视图而非原始表，实现动态脱敏；
• 操作审计：开启数据库审计日志，记录所有用户的操作行为（登录、查询、插入、更新、删除），包括操作时间、用户名、IP地址、执行的SQL语句等，用于追溯违规操作。

实验步骤，结果与问题解决过程

实验场景：模拟某银行客户信息查询系统，不同角色用户（普通柜员、客户经理、系统管理员）查询客户敏感数据（身份证号、银行卡号、余额）时，看到不同脱敏程度的结果，同时记录所有用户的操作行为，便于审计追溯。

模块一：准备实验环境与测试数据

步骤1：确认MySQL服务正常运行

• 登录Xshell连接CentOS 7服务器，执行systemctl status mysqld，确保MySQL服务已启动。

步骤2：创建客户信息表并插入数据

登录MySQL（root用户），执行以下SQL语句：

• • 创建数据库

CREATE DATABASE IF NOT EXISTS bank_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE bank_db;

• • 创建客户信息表（含敏感字段）

CREATE TABLE IF NOT EXISTS customer (
    id INT PRIMARY KEY AUTO_INCREMENT COMMENT '客户ID',
    name VARCHAR(20) NOT NULL COMMENT '客户姓名',
    id_card VARCHAR(18) NOT NULL COMMENT '身份证号（敏感）',
    bank_card VARCHAR(19) NOT NULL COMMENT '银行卡号（敏感）',
    balance DECIMAL(10,2) NOT NULL COMMENT '账户余额（敏感）',
    phone VARCHAR(11) NOT NULL COMMENT '手机号（敏感）',
    register_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '注册时间'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='客户信息主表';

• • 插入测试数据

INSERT INTO customer (name, id_card, bank_card, balance, phone)
VALUES
('张三', '110101199001011234', '6222020108123456789', 50000.00, '13800138000'),
('李四', '310101199505056789', '6222030108987654321', 80000.00, '13900139000'),
('王五', '440101199808085678', '6222040108112233445', 30000.00, '13700137000'),
('赵六', '420101200002028901', '6222050108556677889', 100000.00, '13600136000');
验证
`SELECT * FROM customer;`

步骤3：创建不同角色用户

执行以下SQL语句创建3个不同角色的用户：

• • 1. 普通柜员（只能查看基础信息，敏感字段高度脱敏） CREATE USER IF NOT EXISTS 'teller'@'%' IDENTIFIED BY 'Teller@123';
• • 1. 客户经理（可查看部分敏感字段，轻度脱敏） CREATE USER IF NOT EXISTS 'manager'@'%' IDENTIFIED BY 'Manager@123';

• • 1. 系统管理员（可查看完整信息，无脱敏） CREATE USER IF NOT EXISTS 'admin'@'%' IDENTIFIED BY 'Admin@123';

• • 分配查询权限（仅允许查询customer表）

GRANT SELECT ON bank_db.customer TO 'teller'@'%';

GRANT SELECT ON bank_db.customer TO 'manager'@'%';

GRANT SELECT ON bank_db.customer TO 'admin'@'%';

刷新权限 FLUSH PRIVILEGES;

验证

模块二：设计动态脱敏规则

步骤1：基于角色创建脱敏视图

登录MySQL（root用户），为不同角色创建对应的脱敏视图：

USE bank_db;

• • 1. 普通柜员视图（高度脱敏）

仅显示信息片段

DROP VIEW IF EXISTS customer_view_teller;
CREATE VIEW customer_view_teller AS
SELECT
    id,
    name,
    CONCAT(SUBSTRING(id_card, 1, 6), '**********', SUBSTRING(id_card, 17, 2)) AS id_card,
    CONCAT('**** **** **** ', SUBSTRING(bank_card, 13, 7)) AS bank_card,
    CONCAT('≥', FLOOR(balance/10000)*10000, '元') AS balance,
    CONCAT(SUBSTRING(phone, 1, 3), '****', SUBSTRING(phone, 8, 4)) AS phone,
    register_time
FROM customer;

上个报错是因为MySQL的CREATE VIEW不支持IF NOT EXISTS语法。

• • 1. 客户经理视图（轻度脱敏）
• • 1. 管理员视图（无脱敏，显示完整信息）

验证试图 SHOW FULL TABLES IN bank_db WHERE TABLE_TYPE LIKE 'VIEW';

---

步骤2：分配视图访问权限

执行以下SQL语句，限制不同用户只能访问对应的脱敏视图：

• • 收回直接查询原始表的权限

REVOKE SELECT ON bank_db.customer FROM 'teller'@'%';

REVOKE SELECT ON bank_db.customer FROM 'manager'@'%';

• • 管理员保留原始表访问权限
• • 分配视图访问权限

GRANT SELECT ON bank_db.customer_view_teller TO 'teller'@'%';

GRANT SELECT ON bank_db.customer_view_manager TO 'manager'@'%';

GRANT SELECT ON bank_db.customer_view_admin TO 'admin'@'%';

FLUSH PRIVILEGES;

步骤3：验证动态脱敏效果

用普通柜员teller用户登录Navicat：

• 执行查询：SELECT * FROM bank_db.customer_view_teller;
• 验证结果：身份证号、银行卡号、手机号高度脱敏，余额显示区间（如“≥50000元”）。

用客户经理manager用户登录Navicat：

• 执行查询：SELECT * FROM bank_db.customer_view_manager;
• 验证结果：身份证号、银行卡号轻度脱敏，余额和手机号完整显示。

用管理员admin用户登录Navicat：

• 执行查询：SELECT * FROM bank_db.customer_view_admin;
• 验证结果：所有字段完整显示，无脱敏。

尝试用teller用户查询原始表：SELECT * FROM bank_db.customer; → 提示“权限不足”（验证权限控制）。

模块三：配置MySQL操作审计

步骤1：开启MySQL通用审计日志

登录Xshell连接CentOS 7服务器，编辑MySQL配置文件：

vi /etc/my.cnf

添加以下配置（开启通用日志，记录所有操作）：

[mysqld]

general_log = ON  # 开启通用日志

general_log_file = /var/log/mysql/bank_audit.log  # 审计日志文件路径

log_output = FILE  # 日志输出到文件（支持FILE/TABLE）

创建日志目录并授权：

mkdir -p /var/log/mysql

chown mysql:mysql /var/log/mysql  # 赋予MySQL用户读写权限

重启MySQL服务：

systemctl restart mysqld

步骤2：模拟用户操作行为

用3个不同角色用户依次执行以下操作（模拟正常与异常操作）：

• teller用户：查询脱敏视图2次，尝试查询原始表1次；

SELECT * FROM bank_db.customer_view_teller WHERE id = 1;

正常查询2

SELECT name, phone FROM bank_db.customer_view_teller;

异常操作

SELECT * FROM bank_db.customer;

• manager用户：查询脱敏视图3次，导出1条数据；
• • 查询1 SELECT * FROM bank_db.customer_view_manager WHERE balance > 50000;

• • 查询2 SELECT id_card, balance FROM bank_db.customer_view_manager;

• • 导出操作 在Navicat结果网格右键点击"导出向导"，选择Excel格式，导出当前查询结果

已导出

• admin用户：查询原始表2次，更新1条客户余额（UPDATE bank_db.customer SET balance = 55000.00 WHERE name = '张三';）；

-- 查询1
SELECT * FROM bank_db.customer_view_admin;

-- 查询2 SELECT COUNT(*), AVG(balance) FROM bank_db.customer;

-- 数据修改 UPDATE bank_db.customer SET balance = 55000.00 WHERE name = '张三';

-- 验证修改 SELECT name, balance FROM bank_db.customer WHERE name = '张三';

确保所有操作执行完成，生成审计日志。

tail -f /var/log/mysql/bank_audit.log

步骤3：分析审计日志

在CentOS 7服务器查看审计日志：

cat /var/log/mysql/bank_audit.log  # 查看完整日志

grep -E "SELECT|UPDATE|DELETE" /var/log/mysql/bank_audit.log  # 过滤DML操作

日志内容解析：每条记录包含“时间戳、用户名、主机地址、执行的SQL语句”；

用Python脚本分析审计日志：

• 新建Python脚本py，代码如下：

import re
import pandas as pd
from datetime import datetime
import matplotlib.pyplot as plt

def parse_mysql_audit_log(log_path):
    """解析MySQL审计日志"""
    print(f"开始解析日志文件: {log_path}")
    
    logs = []
    
    try:
        with open(log_path, 'r', encoding='utf-8', errors='ignore') as f:
            for line_num, line in enumerate(f, 1):
                line = line.strip()
                if not line:
                    continue
                
                # 调试：打印前几行日志
                if line_num <= 5:
                    print(f"第{line_num}行: {line[:100]}...")
                
                # 匹配MySQL通用日志格式: 时间戳 线程ID 操作类型 SQL语句
                # 示例: 2025-11-24T09:22:34.786985Z     9 Query SELECT * FROM table
                pattern1 = r'(\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d+Z)\\s+(\\d+)\\s+(\\w+)\\s+(.+)'
                match1 = re.match(pattern1, line)
                
                if match1:
                    timestamp = match1.group(1)
                    thread_id = match1.group(2)
                    operation = match1.group(3)
                    sql = match1.group(4)
                    
                    # 确定用户角色
                    user = "unknown"
                    if "teller" in sql.lower() or "customer_view_teller" in sql:
                        user = "teller"
                    elif "manager" in sql.lower() or "customer_view_manager" in sql:
                        user = "manager"  
                    elif "admin" in sql.lower() or "customer_view_admin" in sql:
                        user = "admin"
                    elif "root" in sql.lower():
                        user = "root"
                    
                    # 确定操作类型
                    sql_upper = sql.upper()
                    if sql_upper.startswith('SELECT'):
                        op_type = "查询"
                    elif sql_upper.startswith('UPDATE'):
                        op_type = "更新" 
                    elif sql_upper.startswith('INSERT'):
                        op_type = "插入"
                    elif sql_upper.startswith('DELETE'):
                        op_type = "删除"
                    elif sql_upper.startswith('CREATE'):
                        op_type = "创建"
                    elif sql_upper.startswith('DROP'):
                        op_type = "删除"
                    elif sql_upper.startswith('GRANT') or sql_upper.startswith('REVOKE'):
                        op_type = "权限"
                    else:
                        op_type = "其他"
                    
                    # 标记可疑操作
                    is_suspicious = False
                    risk_description = ""
                    
                    if user in ["teller", "manager"] and "customer" in sql and "view" not in sql:
                        is_suspicious = True
                        risk_description = "尝试访问原始表"
                    elif "DROP" in sql_upper or "DELETE" in sql_upper:
                        is_suspicious = True
                        risk_description = "高风险操作"
                    
                    logs.append([
                        timestamp, thread_id, user, op_type, 
                        sql[:200],  # 只保留前200字符
                        is_suspicious, risk_description
                    ])
                
                else:
                    # 尝试其他日志格式
                    print(f"无法解析的行 {line_num}: {line[:50]}...")
                    
        print(f"成功解析 {len(logs)} 条日志记录")
        return logs
        
    except Exception as e:
        print(f"解析日志时出错: {e}")
        return []

def analyze_audit_data(logs):
    """分析审计数据"""
    if not logs:
        print("没有可分析的数据")
        return
    
    # 转换为DataFrame
    df = pd.DataFrame(logs, columns=[
        "时间戳", "线程ID", "用户", "操作类型", "SQL语句", "是否异常", "风险描述"
    ])
    
    print("\\n" + "="*60)
    print("MySQL审计日志分析报告")
    print("="*60)
    
    # 基本统计
    print(f"\\n📊 基本统计:")
    print(f"   总操作记录: {len(df)} 条")
    print(f"   时间范围: {df['时间戳'].min()} 到 {df['时间戳'].max()}")
    
    # 用户统计
    print(f"\\n👥 用户操作统计:")
    user_stats = df['用户'].value_counts()
    for user, count in user_stats.items():
        percentage = (count / len(df)) * 100
        print(f"   {user}: {count} 次 ({percentage:.1f}%)")
    
    # 操作类型统计
    print(f"\\n🔧 操作类型分布:")
    op_stats = df['操作类型'].value_counts()
    for op_type, count in op_stats.items():
        percentage = (count / len(df)) * 100
        print(f"   {op_type}: {count} 次 ({percentage:.1f}%)")
    
    # 异常检测
    print(f"\\n🚨 安全检测结果:")
    suspicious_ops = df[df['是否异常'] == True]
    if len(suspicious_ops) > 0:
        print(f"   发现 {len(suspicious_ops)} 个异常操作:")
        for idx, row in suspicious_ops.iterrows():
            print(f"   ⚠️  时间: {row['时间戳']}")
            print(f"      用户: {row['用户']}")
            print(f"      操作: {row['SQL语句']}")
            print(f"      风险: {row['风险描述']}")
            print()
    else:
        print("   ✅ 未发现异常操作")
    
    # 安全评分
    security_score = (len(df) - len(suspicious_ops)) / len(df) * 100 if len(df) > 0 else 100
    print(f"\\n🛡️  安全评估:")
    print(f"   安全评分: {security_score:.1f}%")
    print(f"   正常操作: {len(df) - len(suspicious_ops)} 次")
    print(f"   异常操作: {len(suspicious_ops)} 次")
    
    return df, user_stats, op_stats, suspicious_ops

def create_visualization(user_stats, op_stats, output_path):
    """创建可视化图表"""
    try:
        plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
        plt.rcParams['axes.unicode_minus'] = False
        
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
        
        # 用户操作分布
        ax1.pie(user_stats.values, labels=user_stats.index, autopct='%1.1f%%', startangle=90)
        ax1.set_title('用户操作分布', fontsize=14, fontweight='bold')
        
        # 操作类型统计
        bars = ax2.bar(op_stats.index, op_stats.values, color=['#3498db', '#e74c3c', '#2ecc71', '#f39c12'])
        ax2.set_title('操作类型统计', fontsize=14, fontweight='bold')
        ax2.set_ylabel('操作次数')
        ax2.tick_params(axis='x', rotation=45)
        
        # 在柱子上显示数值
        for bar in bars:
            height = bar.get_height()
            ax2.text(bar.get_x() + bar.get_width()/2., height,
                    f'{int(height)}', ha='center', va='bottom')
        
        plt.tight_layout()
        plt.savefig(output_path, dpi=300, bbox_inches='tight')
        print(f"📈 可视化图表已保存: {output_path}")
        plt.show()
        
    except Exception as e:
        print(f"创建图表时出错: {e}")

def export_to_excel(df, output_path):
    """导出到Excel文件"""
    try:
        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
            # 完整数据
            df.to_excel(writer, sheet_name='完整日志', index=False)
            
            # 汇总统计
            summary_data = {
                '统计项': ['总操作数', '异常操作数', '安全评分', '开始时间', '结束时间'],
                '数值': [
                    len(df),
                    len(df[df['是否异常'] == True]),
                    f"{(len(df) - len(df[df['是否异常'] == True])) / len(df) * 100:.1f}%" if len(df) > 0 else "100%",
                    df['时间戳'].min(),
                    df['时间戳'].max()
                ]
            }
            pd.DataFrame(summary_data).to_excel(writer, sheet_name='汇总统计', index=False)
            
            # 用户统计
            user_stats = df['用户'].value_counts().reset_index()
            user_stats.columns = ['用户', '操作次数']
            user_stats.to_excel(writer, sheet_name='用户统计', index=False)
            
            # 异常操作
            suspicious_ops = df[df['是否异常'] == True]
            if len(suspicious_ops) > 0:
                suspicious_ops.to_excel(writer, sheet_name='异常操作', index=False)
        
        print(f"💾 Excel报告已导出: {output_path}")
        
    except Exception as e:
        print(f"导出Excel时出错: {e}")

def main():
    # 配置路径 - 请根据实际情况修改
    log_path = "D:/data_security/exp4/bank_audit.log"  # 本地日志文件路径
    output_dir = "D:/data_security/exp4"
    
    print("=" * 60)
    print("MySQL审计日志分析工具")
    print("=" * 60)
    
    # 解析日志
    logs = parse_mysql_audit_log(log_path)
    
    if not logs:
        print("解析失败，请检查日志文件路径和格式！")
        print("\\n可能的解决方案:")
        print("1. 检查日志文件路径是否正确")
        print("2. 确认日志文件不为空")
        print("3. 检查日志格式是否匹配")
        print("4. 尝试手动查看日志文件前几行")
        return
    
    # 分析数据
    df, user_stats, op_stats, suspicious_ops = analyze_audit_data(logs)
    
    # 创建可视化
    chart_path = f"{output_dir}/audit_analysis.png"
    create_visualization(user_stats, op_stats, chart_path)
    
    # 导出Excel
    excel_path = f"{output_dir}/audit_analysis_report.xlsx"
    export_to_excel(df, excel_path)
    
    print("\\n" + "="*60)
    print("分析完成！")
    print("="*60)
    print(f"📊 分析报告: {excel_path}")
    print(f"📈 可视化图表: {chart_path}")
    print(f"🔍 发现的异常操作: {len(suspicious_ops)} 个")

if __name__ == "__main__":
    main()

---

• 将服务器上的log下载到本地指定目录；
• 运行脚本，查看分析报告和可视化图表，识别异常操作（如teller用户尝试查询原始表）。

实验思考

一、动态脱敏与静态脱敏的本质区别

动态脱敏与静态脱敏代表了两种完全不同的数据安全防护理念。动态脱敏的核心是"按需可见"，在数据查询使用环节通过视图或函数实时转换敏感信息，原始数据始终保存在库中不发生任何改变，脱敏规则随用户角色动态生效。这种机制的最大优势在于灵活性与安全性兼得——当柜员需要查看客户身份证号时只显示前6后2位，而经理登录同一系统则可看到前8后2位，管理员则拥有完整访问权限，所有转换在毫秒级完成，无需为不同角色维护多套数据副本。相比之下，静态脱敏是在数据离开生产环境前，通过ETL工具提前生成一份"脱敏副本"，原始数据的敏感部分被永久替换或删除，适用于需要将数据提供给测试、开发或第三方机构的场景。本实验中我们采用的视图脱敏方案充分展现了动态脱敏的价值：柜员、经理和管理员看到的是同一张逻辑表的不同"安全投影"，权限变更只需调整视图访问策略即可实时生效，而静态脱敏一旦数据副本交出就失去了控制权，任何策略调整都需要重新执行脱敏作业，存在数据同步滞后和副本管理风险。因此，动态脱敏更适合生产环境的多角色访问控制，静态脱敏则专注于数据流转过程中的匿名化处理，两者共同构成完整的数据安全体系。

二、审计日志在安全事件追溯中的核心作用

审计日志是数据安全的"数字证据链"，其价值不仅在于记录操作，更在于构建不可抵赖的安全态势感知能力。在本次实验中，通用审计日志完整捕获了柜员teller尝试越权的全过程：时间戳精确到秒、客户端IP定位到192.168.3.1、SQL语句显示其直接访问customer表的意图。这种细粒度记录使得事后追溯不再是模糊的责任推断，而是可量化、可举证的事实还原。更重要的是，通过Python脚本的自动化分析，我们能够从海量日志中快速识别异常模式——当teller的查询行为偏离基线（如突然访问原始表），系统立即标记为"高风险"，实现了从"事后追查"到"事中感知"的升级。审计日志的另一个关键作用是满足合规审计要求，无论是等保2.0对三级以上系统的强制日志留存，还是GDPR对个人数据操作的可解释性需求，完整的审计记录都是不可或缺的举证材料。此外，日志分析还能构建用户行为画像，通过统计操作频率、时间分布和SQL模式，发现潜在的内部威胁或账户盗用风险。本次实验正是通过日志分析，精准识别出两条风险操作：柜员的越权尝试暴露了权限配置漏洞，管理员的余额修改需要二次审批确认，体现了审计在闭环安全管理中的枢纽地位。