数据脱敏与加密

简介

数据脱敏与加密是保护敏感数据的核心手段。随着《个人信息保护法》（PIPL）、《数据安全法》以及 GDPR 等法规的实施，企业对个人信息、财务数据、医疗记录等敏感数据的保护已成为法律合规要求，而不仅仅是技术最佳实践。

数据脱敏（Data Masking）是指对敏感数据进行变形处理，使得脱敏后的数据在不改变业务特征的前提下无法识别原始敏感信息。常见策略包括掩码、替换、混淆、加密等。脱敏分为静态脱敏（Static Data Masking，对存储的数据进行不可逆脱敏）和动态脱敏（Dynamic Data Masking，查询时实时脱敏，原始数据不变）。

数据加密（Encryption）是通过密码学算法将明文数据转换为密文，只有持有密钥的授权方才能解密还原。加密分为传输加密（TLS/SSL）、存储加密（TDE、列加密）、应用层加密（应用代码中加密）三个层次，每层各有适用场景。

特点

• 1.合规驱动 — 法律法规要求保护个人信息和敏感数据
• 2.多层防护 — 传输加密、存储加密、应用层加密组合使用
• 3.最小权限 — 只有授权用户和系统才能访问明文数据
• 4.可审计 — 所有敏感数据访问都有日志记录
• 5.业务兼容 — 脱敏后的数据仍可用于开发、测试和分析

数据分类分级

-- 数据分类分级是脱敏和加密的前提

-- 创建数据分类表
CREATE TABLE data_classification (
    id INT PRIMARY KEY AUTO_INCREMENT,
    table_name VARCHAR(100) NOT NULL,
    column_name VARCHAR(100) NOT NULL,
    data_category ENUM('个人信息', '财务数据', '医疗数据', '商业秘密', '一般数据') NOT NULL,
    sensitivity_level ENUM('公开', '内部', '敏感', '机密') NOT NULL,
    masking_strategy VARCHAR(50),
    encryption_required BOOLEAN DEFAULT FALSE,
    description TEXT,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_table_column (table_name, column_name)
) ENGINE=InnoDB;

-- 示例：分类 users 表的敏感字段
INSERT INTO data_classification (table_name, column_name, data_category, sensitivity_level, masking_strategy, encryption_required) VALUES
('users', 'id_number', '个人信息', '机密', 'partial_mask', TRUE),
('users', 'phone', '个人信息', '敏感', 'phone_mask', FALSE),
('users', 'email', '个人信息', '敏感', 'email_mask', FALSE),
('users', 'real_name', '个人信息', '敏感', 'name_mask', FALSE),
('users', 'password_hash', '个人信息', '机密', 'full_mask', TRUE),
('users', 'bank_account', '财务数据', '机密', 'full_mask', TRUE),
('users', 'address', '个人信息', '敏感', 'partial_mask', FALSE),
('users', 'birth_date', '个人信息', '敏感', 'generalize', FALSE),
('users', 'username', '一般数据', '内部', NULL, FALSE),
('users', 'avatar_url', '一般数据', '公开', NULL, FALSE);

-- 查询所有需要脱敏的字段
SELECT
    table_name,
    column_name,
    data_category,
    sensitivity_level,
    masking_strategy,
    encryption_required
FROM data_classification
WHERE sensitivity_level IN ('敏感', '机密')
ORDER BY sensitivity_level DESC, table_name;

静态脱敏

-- 静态脱敏：永久修改数据，用于开发/测试环境

-- 1. 手机号脱敏 (13812345678 -> 138****5678)
UPDATE users SET phone = CONCAT(
    SUBSTRING(phone, 1, 3),
    '****',
    SUBSTRING(phone, 8, 4)
)
WHERE phone IS NOT NULL AND phone REGEXP '^1[3-9][0-9]{9}$';

-- 2. 身份证号脱敏 (110101199001011234 -> 110101****1234)
UPDATE users SET id_number = CONCAT(
    SUBSTRING(id_number, 1, 6),
    '********',
    SUBSTRING(id_number, 15, 4)
)
WHERE id_number IS NOT NULL AND LENGTH(id_number) = 18;

-- 3. 邮箱脱敏 (zhangsan@example.com -> z*****n@example.com)
UPDATE users SET email = CONCAT(
    SUBSTRING(email, 1, 1),
    '*****',
    SUBSTRING(email, LOCATE('@', email) - 1, 1),
    SUBSTRING(email, LOCATE('@', email))
)
WHERE email IS NOT NULL AND email LIKE '%@%';

-- 4. 姓名脱敏 (张三丰 -> 张*丰)
UPDATE users SET real_name = CONCAT(
    SUBSTRING(real_name, 1, 1),
    REPEAT('*', CHAR_LENGTH(real_name) - 1)
)
WHERE real_name IS NOT NULL;

-- 5. 银行卡号脱敏 (6222021234567890123 -> 6222********0123)
UPDATE users SET bank_account = CONCAT(
    SUBSTRING(bank_account, 1, 4),
    REPEAT('*', LENGTH(bank_account) - 8),
    SUBSTRING(bank_account, -4)
)
WHERE bank_account IS NOT NULL;

-- 6. 地址脱敏 (北京市朝阳区xx路xx号 -> 北京市朝阳区***)
UPDATE users SET address = CONCAT(
    SUBSTRING_INDEX(address, '区', 1),
    '区',
    '***'
)
WHERE address IS NOT NULL AND address LIKE '%区%';

-- 7. 日期泛化 (1990-01-15 -> 1990-01-01)
UPDATE users SET birth_date = DATE_FORMAT(birth_date, '%Y-%m-01')
WHERE birth_date IS NOT NULL;

动态脱敏

-- MySQL 8.0 动态数据脱敏（DML 视图方式）

-- 创建脱敏视图（查询时自动脱敏，不修改原始数据）
CREATE VIEW v_users_masked AS
SELECT
    id,
    username,
    -- 姓名脱敏
    CONCAT(SUBSTRING(real_name, 1, 1), REPEAT('*', CHAR_LENGTH(real_name) - 1)) AS real_name,
    -- 手机号脱敏
    CONCAT(SUBSTRING(phone, 1, 3), '****', SUBSTRING(phone, 8, 4)) AS phone,
    -- 邮箱脱敏
    CONCAT(SUBSTRING(email, 1, 1), '*****', SUBSTRING(email, LOCATE('@', email))) AS email,
    -- 身份证脱敏
    CONCAT(SUBSTRING(id_number, 1, 6), '********', SUBSTRING(id_number, 15, 4)) AS id_number,
    -- 其他非敏感字段直接透出
    avatar_url,
    created_at
FROM users;

-- 基于角色的动态脱敏
-- 普通用户只能看到脱敏视图
GRANT SELECT ON your_db.v_users_masked TO 'app_readonly'@'%';
-- 管理员可以看到原始数据
GRANT SELECT ON your_db.users TO 'app_admin'@'%';
-- 撤销普通用户对原始表的直接访问
REVOKE SELECT ON your_db.users FROM 'app_readonly'@'%';

-- 创建脱敏函数（可复用）
DELIMITER //
CREATE FUNCTION mask_phone(phone VARCHAR(20)) RETURNS VARCHAR(20)
DETERMINISTIC
BEGIN
    IF phone IS NULL OR phone = '' THEN
        RETURN phone;
    END IF;
    IF phone REGEXP '^1[3-9][0-9]{9}$' THEN
        RETURN CONCAT(SUBSTRING(phone, 1, 3), '****', SUBSTRING(phone, 8, 4));
    END IF;
    RETURN phone;
END //

CREATE FUNCTION mask_id_card(id_card VARCHAR(20)) RETURNS VARCHAR(20)
DETERMINISTIC
BEGIN
    IF id_card IS NULL OR id_card = '' THEN
        RETURN id_card;
    END IF;
    IF LENGTH(id_card) = 18 THEN
        RETURN CONCAT(SUBSTRING(id_card, 1, 6), '********', SUBSTRING(id_card, 15, 4));
    END IF;
    RETURN id_card;
END //

CREATE FUNCTION mask_email(email VARCHAR(100)) RETURNS VARCHAR(100)
DETERMINISTIC
BEGIN
    DECLARE at_pos INT;
    IF email IS NULL OR email = '' OR email NOT LIKE '%@%' THEN
        RETURN email;
    END IF;
    SET at_pos = LOCATE('@', email);
    RETURN CONCAT(SUBSTRING(email, 1, 1), '*****', SUBSTRING(email, at_pos));
END //
DELIMITER ;

-- 使用脱敏函数
SELECT id, mask_phone(phone) AS phone, mask_email(email) AS email FROM users;

保持格式的脱敏 (FPE)

"""Format-Preserving Encryption (FPE) — 保持格式的加密/脱敏

FPE 的核心特点：加密后的数据与原始数据格式相同
- 手机号加密后还是手机号格式
- 身份证号加密后还是 18 位
- 邮箱加密后还是邮箱格式

这使得脱敏后的数据可以直接用于需要特定格式的业务系统。
"""

import hashlib
import struct

class FormatPreservingEncryption:
    """保持格式的加密（简化实现）

    生产环境建议使用:
    - NIST SP 800-38G (FF1/FF3-1 算法)
    - pyffx 库
    """

    @staticmethod
    def mask_phone(phone: str) -> str:
        """手机号脱敏: 保持格式，只遮盖中间4位"""
        if not phone or len(phone) != 11:
            return phone
        return phone[:3] + "****" + phone[7:]

    @staticmethod
    def mask_id_card(id_card: str) -> str:
        """身份证号脱敏: 保持格式，遮盖中间8位"""
        if not id_card or len(id_card) != 18:
            return id_card
        return id_card[:6] + "********" + id_card[14:]

    @staticmethod
    def mask_bank_card(card_no: str) -> str:
        """银行卡号脱敏: 只保留前4后4"""
        if not card_no or len(card_no) < 8:
            return card_no
        return card_no[:4] + "*" * (len(card_no) - 8) + card_no[-4:]

    @staticmethod
    def mask_email(email: str) -> str:
        """邮箱脱敏: 用户名只保留首尾字符"""
        if not email or "@" not in email:
            return email
        local, domain = email.split("@", 1)
        if len(local) <= 2:
            masked_local = local[0] + "*"
        else:
            masked_local = local[0] + "*" * (len(local) - 2) + local[-1]
        return f"{masked_local}@{domain}"

    @staticmethod
    def mask_name(name: str) -> str:
        """姓名脱敏: 只保留姓"""
        if not name:
            return name
        return name[0] + "*" * (len(name) - 1)

    @staticmethod
    def mask_address(address: str) -> str:
        """地址脱敏: 保留到区/县级别"""
        if not address:
            return address
        for separator in ["区", "县", "市"]:
            idx = address.find(separator)
            if idx > 0:
                return address[:idx + 1] + "***"
        return address[:3] + "***"

    @staticmethod
    def reversible_encrypt(value: str, key: str) -> str:
        """可逆加密 — 脱敏后可以还原（用于需要还原的场景）

        注意: 生产环境应使用 AES-256 等标准算法
        """
        result = []
        for i, char in enumerate(value):
            if char.isdigit():
                # 数字: 按密钥偏移
                shift = int(hashlib.md5(
                    (key + str(i)).encode()
                ).hexdigest()[:2], 16) % 10
                encrypted = str((int(char) + shift) % 10)
                result.append(encrypted)
            elif char.isalpha():
                # 字母: 按密钥偏移
                shift = int(hashlib.md5(
                    (key + str(i)).encode()
                ).hexdigest()[:2], 16) % 26
                if char.isupper():
                    encrypted = chr((ord(char) - ord('A') + shift) % 26 + ord('A'))
                else:
                    encrypted = chr((ord(char) - ord('a') + shift) % 26 + ord('a'))
                result.append(encrypted)
            else:
                result.append(char)
        return ''.join(result)


# 使用示例
fpe = FormatPreservingEncryption()

phone = "13812345678"
print(f"手机号: {phone} -> {fpe.mask_phone(phone)}")

id_card = "110101199001011234"
print(f"身份证: {id_card} -> {fpe.mask_id_card(id_card)}")

card = "6222021234567890123"
print(f"银行卡: {card} -> {fpe.mask_bank_card(card)}")

email = "zhangsan@example.com"
print(f"邮箱: {email} -> {fpe.mask_email(email)}")

列级加密

-- 列级加密：对特定列的数据进行加密存储

-- 方案 1: MySQL 内置 AES 加密函数
-- 插入加密数据
INSERT INTO users_sensitive (
    user_id,
    id_number_encrypted,
    bank_account_encrypted
) VALUES (
    1,
    AES_ENCRYPT('110101199001011234', 'your-secret-key-32-bytes-long!!'),
    AES_ENCRYPT('6222021234567890123', 'your-secret-key-32-bytes-long!!')
);

-- 查询时解密
SELECT
    user_id,
    CAST(AES_DECRYPT(id_number_encrypted, 'your-secret-key-32-bytes-long!!') AS CHAR) AS id_number,
    CAST(AES_DECRYPT(bank_account_encrypted, 'your-secret-key-32-bytes-long!!') AS CHAR) AS bank_account
FROM users_sensitive
WHERE user_id = 1;

-- 方案 2: 使用 HEX 编码存储二进制密文
INSERT INTO users_sensitive (user_id, id_number_encrypted)
VALUES (
    1,
    HEX(AES_ENCRYPT('110101199001011234', 'your-secret-key-32-bytes-long!!'))
);

SELECT
    user_id,
    CAST(AES_DECRYPT(
        UNHEX(id_number_encrypted),
        'your-secret-key-32-bytes-long!!'
    ) AS CHAR) AS id_number
FROM users_sensitive;

-- 方案 3: 创建加密视图
CREATE VIEW v_users_decrypted AS
SELECT
    user_id,
    CAST(AES_DECRYPT(UNHEX(id_number_encrypted), 'your-secret-key-32-bytes-long!!') AS CHAR) AS id_number,
    CAST(AES_DECRYPT(UNHEX(bank_account_encrypted), 'your-secret-key-32-bytes-long!!') AS CHAR) AS bank_account
FROM users_sensitive;

-- 权限控制
GRANT SELECT ON your_db.v_users_decrypted TO 'app_authorized'@'%';
REVOKE SELECT ON your_db.users_sensitive FROM 'app_authorized'@'%';

TDE 透明数据加密

-- MySQL 透明数据加密 (Transparent Data Encryption)

-- MySQL 8.0+ 支持 InnoDB 静态数据加密
-- 加密在存储引擎层完成，对应用完全透明

-- 1. 查看当前加密状态
SELECT
    TABLE_SCHEMA,
    TABLE_NAME,
    ENCRYPTION
FROM information_schema.TABLES
WHERE ENCRYPTION = 'Y';

-- 2. 启用 InnoDB 表空间加密
-- 前提: 配置 keyring 插件
-- my.cnf 配置:
-- early-plugin-load=keyring_file.so
-- keyring_file_data=/var/lib/mysql-keyring/keyring

-- 3. 加密已有表
ALTER TABLE users ENCRYPTION = 'Y';

-- 4. 创建加密表
CREATE TABLE sensitive_data (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    data_type VARCHAR(50),
    encrypted_content TEXT,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
) ENCRYPTION = 'Y';

-- 5. 加密整个数据库的所有表
-- 生成批量加密语句
SELECT CONCAT('ALTER TABLE ', TABLE_SCHEMA, '.', TABLE_NAME, ' ENCRYPTION = "Y";') AS encrypt_sql
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'your_db'
  AND TABLE_TYPE = 'BASE TABLE'
  AND (ENCRYPTION IS NULL OR ENCRYPTION = 'N');

-- 6. 查看加密配置
SHOW VARIABLES LIKE '%encryption%';
SHOW VARIABLES LIKE '%keyring%';

密钥管理

"""密钥管理方案

密钥管理是数据加密安全的核心。密钥泄露等于加密形同虚设。
"""

import os
import hashlib
import base64
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives import hashes

class KeyManager:
    """密钥管理器

    生产环境建议使用:
    - AWS KMS
    - Azure Key Vault
    - HashiCorp Vault
    - 硬件安全模块 (HSM)
    """

    def __init__(self, master_key_path: str = ".keys/master.key"):
        self.master_key_path = master_key_path

    def generate_data_encryption_key(self) -> bytes:
        """生成数据加密密钥 (DEK)"""
        return Fernet.generate_key()

    def encrypt_key(self, dek: bytes, master_key: bytes) -> bytes:
        """用主密钥加密数据密钥 (KEK 包装 DEK)"""
        f = Fernet(master_key)
        return f.encrypt(dek)

    def decrypt_key(self, encrypted_dek: bytes, master_key: bytes) -> bytes:
        """用主密钥解密数据密钥"""
        f = Fernet(master_key)
        return f.decrypt(encrypted_dek)

    def derive_key_from_password(self, password: str, salt: bytes) -> bytes:
        """从密码派生密钥 (PBKDF2)"""
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=600000,  # OWASP 推荐
        )
        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
        return key

    def key_rotation(self, old_key: bytes, data: bytes) -> tuple:
        """密钥轮换: 用新密钥重新加密数据"""
        new_key = self.generate_data_encryption_key()

        # 解密旧数据
        old_f = Fernet(old_key)
        plaintext = old_f.decrypt(data)

        # 用新密钥加密
        new_f = Fernet(new_key)
        new_encrypted = new_f.encrypt(plaintext)

        return new_key, new_encrypted


class ColumnEncryptor:
    """列级加密工具"""

    def __init__(self, key: bytes):
        self.fernet = Fernet(key)

    def encrypt(self, plaintext: str) -> str:
        """加密字符串，返回 hex 编码的密文"""
        encrypted = self.fernet.encrypt(plaintext.encode('utf-8'))
        return base64.b64encode(encrypted).decode('ascii')

    def decrypt(self, ciphertext: str) -> str:
        """解密 hex 编码的密文"""
        encrypted = base64.b64decode(ciphertext.encode('ascii'))
        decrypted = self.fernet.decrypt(encrypted)
        return decrypted.decode('utf-8')

    def encrypt_searchable(self, plaintext: str) -> str:
        """可搜索加密: 对精确匹配的查询场景
        使用 HMAC 生成确定性哈希值，支持等值查询
        """
        h = hashlib.sha256(plaintext.encode('utf-8'))
        return h.hexdigest()


# 使用示例
key_manager = KeyManager()

# 生成并保存主密钥（一次性操作）
master_key = key_manager.generate_data_encryption_key()

# 生成数据加密密钥
dek = key_manager.generate_data_encryption_key()
encrypted_dek = key_manager.encrypt_key(dek, master_key)

# 列级加密
encryptor = ColumnEncryptor(dek)
encrypted_id = encryptor.encrypt("110101199001011234")
print(f"加密后: {encrypted_id}")
print(f"解密后: {encryptor.decrypt(encrypted_id)}")

# 可搜索加密哈希
search_hash = encryptor.encrypt_searchable("110101199001011234")
print(f"搜索哈希: {search_hash}")

.NET 数据保护

// .NET 数据保护 API 使用示例

using Microsoft.AspNetCore.DataProtection;
using Microsoft.Extensions.DependencyInjection;

// 1. 注册数据保护服务
var services = new ServiceCollection();
services.AddDataProtection()
    .PersistKeysToFileSystem(new DirectoryInfo("./keys"))
    .SetApplicationName("MyApp")
    .SetDefaultKeyLifetime(TimeSpan.FromDays(90));

var provider = services.BuildServiceProvider();
var protector = provider.GetService<IDataProtectionProvider>()
    .CreateProtector("SensitiveDataProtection");

// 2. 加密敏感数据
string idNumber = "110101199001011234";
string encrypted = protector.Protect(idNumber);
Console.WriteLine($"加密后: {encrypted}");

// 3. 解密
string decrypted = protector.Unprotect(encrypted);
Console.WriteLine($"解密后: {decrypted}");

// 4. 带过期时间的保护
var timedProtector = provider.GetService<IDataProtectionProvider>()
    .CreateProtector("TimedProtection")
    .ToTimeLimitedDataProtector();

string protectedData = timedProtector.Protect("sensitive-value",
    lifetime: TimeSpan.FromHours(1));
// 1小时后解密会抛异常 CryptographicException

// 5. Entity Framework 列加密
// 使用 Shadow Property 实现自动加解密
public class User
{
    public int Id { get; set; }
    public string Username { get; set; }

    // 数据库存储加密值
    public string IdNumberEncrypted { get; set; }

    // 不映射到数据库，运行时使用
    [NotMapped]
    public string IdNumber { get; set; }
}

public class AppDbContext : DbContext
{
    private readonly IDataProtector _protector;

    public AppDbContext(DbContextOptions options, IDataProtectionProvider dpProvider)
        : base(options)
    {
        _protector = dpProvider.CreateProtector("ColumnEncryption");
    }

    public DbSet<User> Users { get; set; }

    public override int SaveChanges()
    {
        foreach (var entry in ChangeTracker.Entries<User>())
        {
            if (entry.State == EntityState.Added || entry.State == EntityState.Modified)
            {
                if (!string.IsNullOrEmpty(entry.Entity.IdNumber))
                {
                    entry.Entity.IdNumberEncrypted =
                        _protector.Protect(entry.Entity.IdNumber);
                }
            }
        }
        return base.SaveChanges();
    }
}

审计日志

-- 敏感数据访问审计

-- 创建审计日志表
CREATE TABLE sensitive_data_audit (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_name VARCHAR(100) NOT NULL COMMENT '操作人',
    client_ip VARCHAR(45) NOT NULL COMMENT '客户端IP',
    action_type ENUM('SELECT', 'INSERT', 'UPDATE', 'DELETE') NOT NULL,
    table_name VARCHAR(100) NOT NULL,
    column_names VARCHAR(500) COMMENT '涉及的敏感列',
    row_identifier VARCHAR(200) COMMENT '数据行标识',
    query_text TEXT COMMENT '执行的SQL',
    masking_applied BOOLEAN DEFAULT FALSE COMMENT '是否应用了脱敏',
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_user_time (user_name, created_at),
    INDEX idx_table_time (table_name, created_at),
    INDEX idx_action_time (action_type, created_at)
) ENGINE=InnoDB;

-- MySQL 审计插件配置（企业版）
-- INSTALL PLUGIN audit_log SONAME 'audit_log.so';
-- SET GLOBAL audit_log_format = 'JSON';
-- SET GLOBAL audit_log_policy = 'ALL';

-- MariaDB 审计插件（社区版可用）
-- INSTALL PLUGIN server_audit SONAME 'server_audit.so';
-- SET GLOBAL server_audit_events = 'CONNECT,QUERY,TABLE';
-- SET GLOBAL server_audit_logging = 'ON';

-- 使用触发器记录敏感数据修改
DELIMITER //
CREATE TRIGGER trg_users_sensitive_update
AFTER UPDATE ON users
FOR EACH ROW
BEGIN
    -- 监控敏感字段变更
    IF OLD.phone != NEW.phone OR OLD.email != NEW.email
       OR OLD.id_number != NEW.id_number THEN
        INSERT INTO sensitive_data_audit (
            user_name, client_ip, action_type,
            table_name, column_names, row_identifier
        ) VALUES (
            CURRENT_USER(), SUBSTRING_INDEX(HOST, ':', 1),
            'UPDATE', 'users',
            CONCAT_WS(',',
                IF(OLD.phone != NEW.phone, 'phone', NULL),
                IF(OLD.email != NEW.email, 'email', NULL),
                IF(OLD.id_number != NEW.id_number, 'id_number', NULL)
            ),
            CONCAT('id=', NEW.id)
        );
    END IF;
END //
DELIMITER ;

-- 审计查询: 查看某人某天的敏感数据访问
SELECT
    user_name,
    action_type,
    table_name,
    column_names,
    row_identifier,
    created_at
FROM sensitive_data_audit
WHERE user_name = 'app_user@10.0.%'
  AND created_at >= '2024-01-15 00:00:00'
  AND created_at < '2024-01-16 00:00:00'
ORDER BY created_at DESC;

优点

• 1.合规保障 — 满足 PIPL、GDPR 等法规要求
• 2.数据安全 — 即使数据库被拖库，敏感数据也无法直接读取
• 3.灵活控制 — 基于角色和场景的差异化脱敏策略
• 4.业务兼容 — 脱敏数据保持格式，可用于测试开发
• 5.可追溯 — 审计日志记录所有敏感数据访问

缺点

• 1.性能开销 — 加解密消耗 CPU，查询变慢
• 2.密钥管理复杂 — 密钥轮换、分发、撤销流程复杂
• 3.开发成本 — 需要在应用层集成加解密逻辑
• 4.查询受限 — 加密列无法直接使用索引和 WHERE 条件
• 5.维护负担 — 脱敏规则和加密策略需要持续维护

性能注意事项

1. AES 加密开销：MySQL AES_ENCRYPT/AES_DECRYPT 比明文操作慢约 30-50%
2. 列加密索引：加密列无法建立有效索引，需要另存哈希列支持等值查询
3. TDE 影响：InnoDB TDE 的性能影响通常小于 5%，适合全库加密
4. 密钥缓存：应用层缓存解密后的密钥，避免频繁调用 KMS
5. 批量脱敏：静态脱敏大批量数据时，建议分批执行避免长事务
6. 审计日志大小：高频访问场景下审计日志增长很快，需要定期归档

总结

数据脱敏与加密是数据安全的最后防线。核心原则是分类分级、最小必要、加密存储、脱敏使用、全程审计。静态脱敏用于开发测试环境，动态脱敏用于生产查询，TDE 用于全库加密，列加密用于特定敏感字段。密钥管理是安全的关键，生产环境务必使用专业的 KMS 方案。

关键知识点

1. 数据分类分级 — 公开/内部/敏感/机密四级分类
2. 静态脱敏 vs 动态脱敏 — 前者不可逆改数据，后者查询时实时脱敏
3. FPE 保持格式加密 — 脱敏后数据保持原始格式特征
4. MySQL AES 加密 — AES_ENCRYPT/AES_DECRYPT 内置函数
5. TDE 透明加密 — 存储引擎层加密，对应用透明
6. 密钥层次结构 — 主密钥 (KEK) 加密数据密钥 (DEK)
7. .NET Data Protection API — 微软推荐的数据保护框架
8. 审计日志 — 记录所有敏感数据的访问和修改

常见误区

1. 只加密不管密钥：密钥硬编码在代码中，等于没加密
2. 脱敏即安全：掩码 * 不是加密，有经验的攻击者可以推断
3. 忽视内部威胁：只防外部攻击，忽视内部员工的越权访问
4. 过度加密：所有数据都加密，性能和开发成本不可接受
5. 密钥不轮换：同一密钥用几年，一旦泄露所有数据都暴露
6. 忘记录审计：没有审计日志，出了安全事件无法追溯

进阶路线

1. 入门：理解数据分类分级，实现基本的静态脱敏
2. 进阶：动态脱敏视图、列级加密、密钥管理
3. 高级：TDE 配置、FPE 实现、KMS 集成
4. 专家：全链路数据保护、零信任数据架构、隐私计算

适用场景

• 涉及个人信息（手机号、身份证、银行卡）的业务系统
• 金融行业的客户数据保护
• 医疗行业的患者数据保护
• 开发/测试环境的数据脱敏
• 跨境数据传输的合规要求

落地建议

1. 第一步：梳理所有敏感字段，完成数据分类分级
2. 第二步：实现静态脱敏，开发测试环境不再使用真实数据
3. 第三步：配置动态脱敏视图，生产环境按角色返回脱敏数据
4. 第四步：部署 TDE，加密存储层
5. 第五步：集成 KMS，统一管理密钥
6. 持续：定期审计日志，密钥定期轮换

排错清单

• 脱敏规则是否覆盖所有敏感字段？检查 data_classification 表
• 动态脱敏视图权限是否正确？普通用户不能访问原始表
• 加密列查询是否使用哈希索引？检查查询计划
• 密钥是否安全存储？不能硬编码在代码或配置文件中
• TDE 是否生效？检查 information_schema.TABLES 的 ENCRYPTION 字段
• 审计日志是否正常记录？插入测试数据验证
• 密钥轮换流程是否测试过？验证轮换后数据可正常解密
• 开发环境是否使用脱敏数据？禁止从生产库直接导出数据

复盘问题

1. 所有敏感字段是否都已分类分级并实施保护措施？
2. 上次密钥轮换是什么时候？是否按计划执行？
3. 审计日志中是否有异常的敏感数据访问？
4. 最近一次数据安全审计的结论是什么？
5. 开发/测试环境的数据来源是什么？是否经过脱敏？
6. 如果密钥泄露，多长时间能完成轮换？

延伸阅读

• 数据库备份与恢复
• 数据库连接池
• API 安全
• 数据库设计
• 数据库监控与告警