AiDOPWarehouse/tools/integration-simulator/clients/redaction.py

"""日志、请求头与运行记录脱敏（任务书 P1-B 安全要求）。

规则：
- 已知 Secret / 密码 / Token 值出现即替换为 ***REDACTED***；
- 敏感 Header 值只保留 AccessKey 后四位，其余整体遮蔽；
- 连接串中 pwd=/password= 段遮蔽。
"""
from __future__ import annotations

import re

REDACTED = "***REDACTED***"

SENSITIVE_HEADERS = {
    "authorization", "x-signature", "x-access-key", "cookie", "set-cookie",
    "proxy-authorization", "x-csrf-token",
}

# 键名命中即整体遮蔽（用于运行记录/报告中的嵌套字段）
SENSITIVE_KEY_RE = re.compile(
    r"(password|passwd|pwd|secret|token|signature|access[-_]?key|credential|jwt)", re.IGNORECASE)

_CONN_PWD_RE = re.compile(r"(pwd|password)\s*=\s*[^;]+", re.IGNORECASE)
_BEARER_RE = re.compile(r"(Bearer|Basic)\s+[A-Za-z0-9\-._~+/=]+", re.IGNORECASE)


def mask_access_key(access_key: str | None) -> str:
    """报告中 AccessKey 只保留后四位。"""
    if not access_key:
        return ""
    return f"****{access_key[-4:]}" if len(access_key) > 4 else "****"


def redact_headers(headers: dict, secrets: list[str] | None = None) -> dict:
    out = {}
    for k, v in headers.items():
        if str(k).lower() in SENSITIVE_HEADERS:
            out[k] = mask_access_key(str(v)) if str(k).lower() == "x-access-key" else REDACTED
        else:
            out[k] = redact_text(str(v), secrets)
    return out


def redact_text(text: str, secrets: list[str] | None = None) -> str:
    if not text:
        return text
    result = text
    for secret in secrets or []:
        if secret and secret in result:
            result = result.replace(secret, REDACTED)
    result = _CONN_PWD_RE.sub(lambda m: f"{m.group(1)}={REDACTED}", result)
    result = _BEARER_RE.sub(lambda m: f"{m.group(1)} {REDACTED}", result)
    return result


def redact_obj(obj, secrets: list[str] | None = None):
    """递归脱敏 dict/list/str，用于运行记录与报告。

    - 敏感 Header 名：Authorization/签名类整体遮蔽，AccessKey 只留后四位；
    - 键名命中 SENSITIVE_KEY_RE 的字段整体遮蔽；
    - 其余字符串按已知 Secret 值与连接串/Bearer 模式遮蔽。
    """
    if isinstance(obj, dict):
        out = {}
        for k, v in obj.items():
            key = str(k).lower()
            if key in SENSITIVE_HEADERS or SENSITIVE_KEY_RE.search(key):
                out[k] = mask_access_key(str(v)) if "access" in key and "key" in key else REDACTED
            else:
                out[k] = redact_obj(v, secrets)
        return out
    if isinstance(obj, list):
        return [redact_obj(v, secrets) for v in obj]
    if isinstance(obj, str):
        return redact_text(obj, secrets)
    return obj