Leak Containment & Anti-Extraction Defense — 泄露遏制、凭证轮换与数据掩码体系

当攻击者从 HTML 堆叠、JSON 响应或混合日志中清洗抽取数据时，他们依赖三件事：固定前缀与计数器规律、未受保护的原始 PII/卡密明文，以及从泄露到响应之间的时间窗口。蓝队必须用“抗聚类不透明编码 + PII 令牌化/假名化 + 金丝雀溯源 + 快速凭证轮换与遏制”打破整条数据利用链。

1. 对抗数据清洗与关联攻击的防御映射

对应 07-data-cleaning.md 中的格式提取、前缀聚类、计数器归一化、锚点映射与业务字段标记，建立以下四维防线：

攻击利用环节	攻击者依赖特征	蓝队反制与遏制措施
前缀聚类与计数器归一化	`VIP...`、`CDK...`、`TK0`–`TKF` 固定 2–4 位前缀与等长格式	改用 CSPRNG 不透明令牌，统一长度与随机分布；数据库仅存加椒哈希（`HMAC-SHA256`）
锚点映射与品牌关联	通过少量已知订单样本把 `prefix_A` 映射至对应商品/品牌	消除码面与 SKU 的字符规律耦合；兑换端强制校验 `user_id`/`order_id` 归属绑定
HTML/JSON 混合抽取	多段 `<textarea>`/`<table>` 或冗长 JSON 返回全量 PII、卡密与回调原文	接口与导出面最小化，PII 实施令牌化/假名化与字段遮罩，下载链接改为短时签名直链
泄露滞后与二次变现	泄露发生后未及时吊销未核销卡密、会话和支付密钥	植入金丝雀（Canary）记录第一时间感知泄露，触发自动遏制、轮换与批量作废

2. 核心防御与事件响应体系

2.1 抗前缀聚类：CSPRNG 不透明令牌与单向哈希存储

杜绝使用 VIP-0001、TK0...TKF 等携带业务/批次语义的可预测编码，避免攻击者仅凭格式与长度分布即可批量归类未核销资产：

生成方式：使用密码学安全随机数生成 128-bit 以上不透明兑换码（无业务前缀、无递增计数器），仅在发卡瞬间向授权订单拥有者展示一次。
落库方式：数据库不存明文激活码，仅存储 HMAC-SHA256(pepper, normalized_code) 摘要与后 4 位用于人工核对；即便发生全库转储，攻击者也无法直接提取可用卡密。

2.2 PII 令牌化、假名化与输出遮罩

在日志、分析库、管理后台列表与 JSON/HTML 响应中，严格隔离直接个人身份信息（邮箱、手机号、Session、回调 raw_body）：

令牌化（Tokenization）：将真实邮箱、手机号与支付卡号集中保存在独立 Vault 库，业务主表与日志仅保留无意义内部替代令牌（如 usr_tok_8f3a...）。
假名化与字段掩码（Pseudonymization & Masking）：展示与导出层仅返回部分遮罩结果（如 a***@example.com、138****5678），禁用整行冗余字段透传，消除“单次导出即拼出 identity + order_id + card_secret + callback”的条件。
短时效绑定下载：数字商品 download_url 绑定用户身份与 5 分钟过期 HMAC 签名，禁止静态永久链接出现在数据库或邮件回执日志中。

2.3 金丝雀（Canary）数据埋点与泄露检测

在生产库、分析副本及不同渠道的导出任务中植入具备唯一追踪标识的金丝雀记录（Honeytoken）：

记录类型：唯一虚构邮箱（带独立追踪子域或标签）、唯一金丝雀兑换码（无真实库存价值）、唯一监控 Webhook URL。
检测链路：
- 若金丝雀兑换码在兑换接口被尝试核销 → 立即告警并封禁源会话/IP，同时定位该码所属的副本或导出批次；
- 若金丝雀邮箱收到未知钓鱼/营销投递，或监控 Webhook 触发外部 DNS/HTTP 请求 → 确认该环境数据集已发生外泄，立即进入泄露遏制流程。

2.4 泄露事件后遏制与凭证轮换协议

一旦确认或高度疑似数据库/日志/导出发生泄露，按“先遏制扩散、再全量轮换、后吊销资产”顺序执行：

泄露遏制（Leak Containment）：下线或限流受影响的公开接口与导出端点；阻断异常批量抓取 IP 与爬虫会话；封存并隔离涉事服务器内存/磁盘快照以保留取证链。
凭证轮换（Credential Rotation）：
- 轮换数据库账号密码、Redis/Mongo 认证口令与对象存储 OSS_*/S3_* 密钥；
- 轮换 JWT 签名密钥（升级 kid）、Session 加密盐与第三方支付网关/Webhook 签名密钥（PAY_KEY、STRIPE_SECRET 等）；
- 替换兑换码存储 Pepper（对新签发资产启用新密钥版本）。
权益与会话吊销（Revocation）：
- 清空全部活跃 Session（使泄露的 PHPSESSID/Cookie 立即失效）并强制高风险账号重置口令；
- 对暴露时间窗口内尚未核销的旧版卡密/激活码/下载令牌执行批量失效并向合法订单用户重新签发。

3. 可运行防御自检脚本

以下 Python 脚本可本地直接运行，覆盖四项核心机制的自检：（1）生成抗前缀聚类的不透明随机令牌并计算 HMAC-SHA256 落库摘要；（2）对邮箱、手机号与卡密执行字段遮罩与假名化令牌替换；（3）植入并检测金丝雀记录命中；（4）模拟泄露后凭证版本轮换使旧密钥签发凭证立即失效：

python

#!/usr/bin/env python3
"""Self-check for anti-clustering tokens, PII masking, canary detection, and key rotation."""
import hashlib
import hmac
import re
import secrets


def generate_opaque_entitlement(pepper: bytes) -> tuple[str, str]:
    """Generate CSPRNG opaque code (no SKU prefix or hex counter) + HMAC digest for DB storage."""
    raw_token = secrets.token_urlsafe(18)  # 144-bit entropy, uniform distribution
    digest = hmac.new(pepper, raw_token.encode("utf-8"), hashlib.sha256).hexdigest()
    return raw_token, digest


def mask_email(email: str) -> str:
    local, _, domain = email.partition("@")
    if not domain:
        return "***"
    prefix = local[0] if local else "*"
    return f"{prefix}***@{domain}"


def mask_phone(phone: str) -> str:
    return re.sub(r"^(\d{3})\d{4}(\d{4})$", r"\1****\2", phone)


def pseudonymize_record(record: dict, salt: bytes) -> dict:
    uid_pseudo = hmac.new(salt, str(record["user_id"]).encode("utf-8"), hashlib.sha256).hexdigest()[:16]
    return {
        "pseudo_uid": f"ps_{uid_pseudo}",
        "email_masked": mask_email(record["email"]),
        "phone_masked": mask_phone(record["phone"]),
        "order_ref": record["order_id"],
    }


class CanaryRegistry:
    def __init__(self) -> None:
        self._canaries: dict[str, str] = {}

    def plant(self, source_tag: str) -> str:
        canary_code = "CN-" + secrets.token_hex(8)
        self._canaries[canary_code] = source_tag
        return canary_code

    def inspect_attempt(self, submitted_code: str) -> str | None:
        return self._canaries.get(submitted_code)


def verify_entitlement_with_rotation(
    submitted: str,
    stored_digest: str,
    key_version: int,
    active_keys: dict[int, bytes],
) -> bool:
    key = active_keys.get(key_version)
    if key is None:
        return False  # 已轮换吊销的旧密钥版本直接拒绝
    candidate = hmac.new(key, submitted.encode("utf-8"), hashlib.sha256).hexdigest()
    return hmac.compare_digest(candidate, stored_digest)


def run_self_check() -> None:
    # 1. 抗前缀聚类令牌生成（无固定 SKU 前缀，各码相互独立）
    keys = {1: b"old-pepper-v1"}
    code1, digest1 = generate_opaque_entitlement(keys[1])
    code2, _ = generate_opaque_entitlement(keys[1])
    assert code1 != code2 and len(code1) >= 24
    assert verify_entitlement_with_rotation(code1, digest1, 1, keys) is True

    # 2. PII 掩码化与假名化
    masked = pseudonymize_record(
        {"user_id": 42, "email": "alice@example.com", "phone": "13800138000", "order_id": "ORD-9001"},
        b"analytics-salt",
    )
    assert masked["email_masked"] == "a***@example.com"
    assert masked["phone_masked"] == "138****8000"
    assert "alice" not in str(masked) and "13800138000" not in str(masked)

    # 3. 金丝雀检测命中定位泄露源
    registry = CanaryRegistry()
    canary = registry.plant("replica_export_20261112")
    assert registry.inspect_attempt(canary) == "replica_export_20261112"
    assert registry.inspect_attempt(code1) is None

    # 4. 泄露后轮换密钥版本，吊销旧版本密钥
    active_after_rotation = {2: b"new-pepper-v2"}
    assert verify_entitlement_with_rotation(code1, digest1, 1, active_after_rotation) is False
    print("[PASS] Leak containment & anti-extraction defense self-check succeeded.")


if __name__ == "__main__":
    run_self_check()

防御工作流

text

1. 改造发卡与激活码格式：淘汰固定前缀+十六进制计数器，全部替换为 CSPRNG 不透明令牌并以 HMAC 摘要存储
2. 收敛 HTML/JSON 接口与导出字段，对邮箱、手机号、回调原文及会话标识实施令牌化、假名化与字段遮罩
3. 在核心数据库表、分析副本及定时导出任务中植入唯一金丝雀（Canary）记录并接入实时告警规则
4. 当金丝雀告警或泄露情报触发时，立即封闭异常导出端点与高风险会话以完成泄露遏制
5. 按优先级执行全量凭证轮换（DB/Redis/支付网关/JWT 签名密钥）并批量作废窗口期内未核销的旧凭证
6. 运行自检脚本确认新密钥签发与旧版本吊销逻辑正确生效

验证与判定标准

防御能力	校验方法	通过标准	失败信号
抗前缀聚类	对连续签发的 100 个卡密/令牌做前缀与长度统计	无固定 2–4 位业务前缀，无十六进制连续计数器特征	同一商品生成码呈固定前缀或顺序递增
PII 遮罩与令牌化	检查导出 CSV、分析库与 HTML/JSON 响应体	仅含假名 ID 与遮罩字段，不含完整明文邮箱/手机/卡密	接口或导出直接透传明文 `password`/`card_secret`/`raw_body`
金丝雀溯源有效性	模拟核销一条金丝雀记录	在告警端精确返回对应 `source_tag` 并触发阻断标记	核销成功或系统无告警记录
泄露后轮换与吊销	轮换密钥版本并清理会话表后重放旧令牌/Session	旧 `kid`/旧密钥版本凭据 100% 验证失败，旧会话失效	轮换后旧 Session 或未核销旧凭证仍可正常使用

MCP 工具映射

防御步骤	MCP 工具	说明
检索泄露遏制与轮换指南	`kb_router`	按 `leak containment`、`tokenization`、`credential rotation`、`canary` 搜索
对照攻击者清洗思路	`kb_read_file`	读取 07-data-cleaning.md 识别需消除的可聚类特征与锚点
浏览数据库防御集	`kb_catalog`	获取 `24-database` 全量红蓝配对清单
暴露面复核	`http_probe`	在 `red`/`both` 模式下核对导出与查询接口返回已遮罩且无冗余 PII

关联技术

07-data-cleaning.md — 配对攻击面：泄露数据清洗方法论
05-backup-log-leak-defense.md — 备份加密与日志遮罩前置防护
04-config-exposure-defense.md — 数据库与支付凭证配置保护

Leak Containment & Anti-Extraction Defense — 泄露遏制、凭证轮换与数据掩码体系 ​

1. 对抗数据清洗与关联攻击的防御映射 ​

2. 核心防御与事件响应体系 ​

2.1 抗前缀聚类：CSPRNG 不透明令牌与单向哈希存储 ​

2.2 PII 令牌化、假名化与输出遮罩 ​

2.3 金丝雀（Canary）数据埋点与泄露检测 ​

2.4 泄露事件后遏制与凭证轮换协议 ​

3. 可运行防御自检脚本 ​

防御工作流 ​

验证与判定标准 ​

MCP 工具映射 ​

关联技术 ​