7.2.2 数据脱敏 / 加密存储 / 访问审计三件套

# static_masking.py - 静态脱敏完整示例
import hashlib
import re
import secrets
from typing import Callable, Dict

class StaticMasker:
    """静态脱敏器:生产数据导出到测试环境前使用"""

    def __init__(self):
        self.strategies: Dict[str, Callable] = {
            "name": self._mask_name,
            "mobile": self._mask_mobile,
            "id_card": self._mask_id_card,
            "email": self._mask_email,
            "card_no": self._mask_card_no,
            "address": self._mask_address,
        }

    def _mask_name(self, value: str) -> str:
        """姓名脱敏:保留姓,名字 -> *"""
        if not value or len(value) == 0:
            return value
        if len(value) == 1:
            return value
        return value[0] + "*" * (len(value) - 1)

    def _mask_mobile(self, value: str) -> str:
        """手机号脱敏:138****1234"""
        if not value or len(value) != 11:
            return value
        return value[:3] + "****" + value[7:]

    def _mask_id_card(self, value: str) -> str:
        """身份证脱敏:保留前 6 后 4"""
        if not value or len(value) < 10:
            return value
        return value[:6] + "*" * (len(value) - 10) + value[-4:]

    def _mask_email(self, value: str) -> str:
        """邮箱脱敏:a****@example.com"""
        if not value or "@" not in value:
            return value
        local, domain = value.split("@", 1)
        if len(local) <= 1:
            masked_local = local + "***"
        else:
            masked_local = local[0] + "***" + local[-1]
        return f"{masked_local}@{domain}"

    def _mask_card_no(self, value: str) -> str:
        """银行卡脱敏:保留前 4 后 4"""
        if not value or len(value) < 8:
            return value
        return value[:4] + "*" * (len(value) - 8) + value[-4:]

    def _mask_address(self, value: str) -> str:
        """地址脱敏:保留前 6 字"""
        if not value or len(value) <= 6:
            return "***"
        return value[:6] + "***"

    def mask_dict(self, data: dict, rules: dict) -> dict:
        """按规则脱敏字典"""
        masked = {}
        for key, value in data.items():
            if key in rules and value is not None:
                strategy = rules[key]
                if strategy in self.strategies:
                    masked[key] = self.strategies[strategy](str(value))
                else:
                    masked[key] = value
            else:
                masked[key] = value
        return masked


# 使用示例
if __name__ == "__main__":
    masker = StaticMasker()
    user = {
        "id": 10001,
        "name": "张三丰",
        "mobile": "13800138000",
        "id_card": "110101199001011234",
        "email": "zhangsan@example.com",
        "card_no": "6222021234567890123",
        "address": "北京市朝阳区某某街道 100 号",
    }
    rules = {
        "name": "name",
        "mobile": "mobile",
        "id_card": "id_card",
        "email": "email",
        "card_no": "card_no",
        "address": "address",
    }
    masked = masker.mask_dict(user, rules)
    print(masked)
    # {'id': 10001, 'name': '张**',
    #  'mobile': '138****8000', 'id_card': '110101********1234',
    #  'email': 'z***n@example.com', 'card_no': '6222***********0123',
    #  'address': '北京市朝阳区***'}

# pseudonymize.py - 假名化(支持可逆还原)
import hashlib
import hmac
import os

class Pseudonymizer:
    """假名化:HMAC-SHA256 可在密钥控制下还原"""

    def __init__(self, secret_key: bytes = None):
        self.key = secret_key or os.urandom(32)

    def pseudonymize(self, identifier: str, salt: str = "") -> str:
        """生成假名(同输入同输出,确定性)"""
        h = hmac.new(self.key, (salt + identifier).encode(), hashlib.sha256)
        return h.hexdigest()[:16]

    def pseudonymize_with_mapping(self, identifier: str, mapping: dict) -> str:
        """基于映射表的假名化(可逆)"""
        if identifier in mapping:
            return mapping[identifier]
        new_id = f"user_{len(mapping) + 1:06d}"
        mapping[identifier] = new_id
        return new_id


# Faker 合成测试数据
from faker import Faker
fake = Faker("zh_CN")

def generate_test_users(n: int) -> list:
    """生成 n 条合成测试用户"""
    users = []
    for i in range(n):
        users.append({
            "name": fake.name(),
            "mobile": fake.phone_number(),
            "id_card": fake.ssn(),
            "email": fake.email(),
            "address": fake.address(),
        })
    return users

# 测试
if __name__ == "__main__":
    p = Pseudonymizer()
    # 同输入同输出
    print(p.pseudonymize("user_123"))  # 7a3f2c9e1b4d8f5a
    print(p.pseudonymize("user_123"))  # 7a3f2c9e1b4d8f5a
    print(p.pseudonymize("user_456"))  # 9c2e1d8a5f7b3c6e

# presidio_analyzer.py - Presidio 自动识别 + 脱敏
from presidio_analyzer import AnalyzerEngine
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

# 配置中文 NLP
nlp_config = {
    "nlp_engine_name": "spacy",
    "models": [{"lang_code": "zh", "model_name": "zh_core_web_sm"}],
}
nlp_engine = NlpEngineProvider(nlp_configuration=nlp_config).create_engine()
analyzer = AnalyzerEngine(nlp_engine=nlp_engine, supported_languages=["zh", "en"])
anonymizer = AnonymizerEngine()

text = "张三丰,身份证 110101199001011234,手机 13800138000,邮箱 zhang@example.com"

# 自动识别 PII
results = analyzer.analyze(text=text, language="zh")
print("识别结果:", [(r.entity_type, r.score) for r in results])

# 脱敏(不同字段不同策略)
operators = {
    "PERSON": OperatorConfig("replace", {"new_value": "[姓名]"}),
    "ID_CARD": OperatorConfig("mask", {"masking_char": "*", "chars_to_mask": 8, "from_end": False}),
    "PHONE_NUMBER": OperatorConfig("mask", {"masking_char": "*", "chars_to_mask": 4, "from_end": True}),
    "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "[邮箱]"}),
}
anonymized = anonymizer.anonymize(text=text, analyzer_results=results, operators=operators)
print(anonymized.text)
# "[姓名],身份证 110101********1234,手机 138****8000,邮箱 [邮箱]"

# aes_field_encrypt.py - 字段级 AES 加密
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import padding, hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import os
import base64

class FieldEncryptor:
    """AES-256-CBC 字段加密器"""

    def __init__(self, master_key: bytes = None):
        self.key = master_key or os.urandom(32)

    def _derive_key(self, password: str, salt: bytes) -> bytes:
        """从密码派生密钥"""
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100000,
        )
        return kdf.derive(password.encode())

    def encrypt(self, plaintext: str, salt: bytes = None) -> str:
        """加密字段,返回 base64"""
        salt = salt or os.urandom(16)
        iv = os.urandom(16)

        # PKCS7 填充
        padder = padding.PKCS7(128).padder()
        padded = padder.update(plaintext.encode()) + padder.finalize()

        cipher = Cipher(algorithms.AES(self.key), modes.CBC(iv))
        encryptor = cipher.encryptor()
        ciphertext = encryptor.update(padded) + encryptor.finalize()

        # iv + salt + ciphertext
        result = iv + salt + ciphertext
        return base64.b64encode(result).decode()

    def decrypt(self, token: str) -> str:
        """解密字段"""
        raw = base64.b64decode(token)
        iv = raw[:16]
        salt = raw[16:32]
        ciphertext = raw[32:]

        cipher = Cipher(algorithms.AES(self.key), modes.CBC(iv))
        decryptor = cipher.decryptor()
        padded = decryptor.update(ciphertext) + decryptor.finalize()

        unpadder = padding.PKCS7(128).unpadder()
        plaintext = unpadder.update(padded) + unpadder.finalize()
        return plaintext.decode()


# 使用示例
if __name__ == "__main__":
    enc = FieldEncryptor()
    cipher = enc.encrypt("110101199001011234")
    plain = enc.decrypt(cipher)
    print(f"密文: {cipher}")
    print(f"明文: {plain}")

# aliyun_kms.py - 集成阿里云 KMS 进行密钥管理
from alibabacloud_kms20160120.client import Client as KmsClient
from alibabacloud_kms20160120 import models as kms_models
from alibabacloud_tea_openapi import models as open_api_models

class AliyunKMS:
    """阿里云 KMS 集成,避免密钥硬编码在应用"""

    def __init__(self, access_key: str, secret_key: str, region: str, key_id: str):
        config = open_api_models.Config(
            access_key_id=access_key,
            access_key_secret=secret_key,
            region_id=region,
            endpoint=f"kms.{region}.aliyuncs.com",
        )
        self.client = KmsClient(config)
        self.key_id = key_id

    def encrypt(self, plaintext: str) -> str:
        """KMS 加密(信封加密,KMS 保护 DEK,DEK 加密数据)"""
        request = kms_models.EncryptRequest(key_id=self.key_id, plaintext=plaintext)
        response = self.client.encrypt(request)
        return response.body.ciphertext_blob

    def decrypt(self, ciphertext_blob: str) -> str:
        request = kms_models.DecryptRequest(ciphertext_blob=ciphertext_blob)
        response = self.client.decrypt(request)
        return response.body.plaintext

    def generate_data_key(self) -> tuple:
        """生成数据加密密钥(DEK),返回 (明文 DEK, 密文 DEK)"""
        request = kms_models.GenerateDataKeyRequest(
            key_id=self.key_id, number_of_bytes=32
        )
        response = self.client.generate_data_key(request)
        return response.body.plaintext, response.body.ciphertext_blob


# 使用:信封加密模式
if __name__ == "__main__":
    kms = AliyunKMS(
        access_key="LTAI***", secret_key="***",
        region="cn-hangzhou", key_id="alias/user-db-master"
    )
    # 1. 生成 DEK(每个用户/字段一个)
    plain_dek, cipher_dek = kms.generate_data_key()
    # 2. 用 DEK 在本地加密大量数据(AES)
    # 3. DEK 明文用完即丢,密文 DEK 存入数据库表
    # 4. 解密时:从 DB 取密文 DEK -> KMS 解出明文 DEK -> 本地解密数据

# gdpr_request.py - GDPR 数据主体权利响应
class DSARHandler:
    """Data Subject Access Request 处理器"""

    def handle_access_request(self, user_id: str) -> dict:
        """访问权:返回用户所有数据"""
        return {
            "profile": self.db.get_user(user_id),
            "orders": self.order_db.list_by_user(user_id),
            "logs": self.audit_db.list_by_user(user_id),
            "third_party_shares": self.consent_db.list_shares(user_id),
        }

    def handle_erasure_request(self, user_id: str) -> bool:
        """被遗忘权:删除用户所有数据"""
        # 1. 软删除(标记,30 天宽限期)
        self.db.soft_delete_user(user_id)
        # 2. 异步通知所有数据持有方
        for service in self.data_holders:
            service.schedule_deletion(user_id, delay=30*86400)
        # 3. 备份系统标记(下次轮转时删除)
        self.backup_db.mark_for_deletion(user_id)
        # 4. 第三方通知
        self.consent_db.revoke_all_consents(user_id)
        return True

    def handle_portability_request(self, user_id: str) -> bytes:
        """可携权:导出 JSON/CSV"""
        data = self.handle_access_request(user_id)
        return json.dumps(data, ensure_ascii=False, indent=2).encode()