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

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

规则：
- 已知 Secret / 密码 / Token 值出现即替换为 ***REDACTED***；
- 敏感 Header 值只保留 AccessKey 后四位，其余整体遮蔽；
- 连接串中 pwd=/password= 段遮蔽；
- 裸 JWT（eyJ…）与 URL 内嵌凭据（scheme://user:pwd@host）遮蔽；
- 已知 AccessKey 值出现在自由文本时只保留后四位（任务书 §24）。
"""
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)
# 中文自由文本口令：密码/口令 + 值（仅 ASCII 值，避免误伤普通描述）
_CN_PWD_RE = re.compile(r"(密码|口令)\s*[:：=]?\s*[A-Za-z0-9_\-@#$%!*]{4,}")
_BEARER_RE = re.compile(r"(Bearer|Basic)\s+[A-Za-z0-9\-._~+/=]+", re.IGNORECASE)
# 裸 JWT：header.payload.signature 三段 Base64URL（即使没有 Bearer 前缀）
_JWT_RE = re.compile(r"eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}")
# URL 内嵌凭据：mysql://user:pwd@host / https://user:pass@host
_URL_CRED_RE = re.compile(
    r"(?P<scheme>[a-zA-Z][a-zA-Z0-9+.-]*://)(?P<user>[^:/\s@]+):(?P<pwd>[^@/\s]+)@")


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,
                   access_keys: 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, access_keys)
    return out


def redact_text(text: str, secrets: list[str] | None = None,
                access_keys: 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 = _CN_PWD_RE.sub(lambda m: f"{m.group(1)} {REDACTED}", result)
    result = _BEARER_RE.sub(lambda m: f"{m.group(1)} {REDACTED}", result)
    result = _JWT_RE.sub(REDACTED, result)
    result = _URL_CRED_RE.sub(
        lambda m: f"{m.group('scheme')}{m.group('user')}:{REDACTED}@", result)
    # 已知 AccessKey 值：只保留后四位（优先于全量遮蔽）
    for access_key in access_keys or []:
        if access_key and access_key in result:
            result = result.replace(access_key, mask_access_key(access_key))
    return result


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

    - 敏感 Header 名：Authorization/签名类整体遮蔽，AccessKey 只留后四位；
    - 键名命中 SENSITIVE_KEY_RE 的字段整体遮蔽；
    - 其余字符串按已知 Secret 值、连接串/Bearer/JWT/URL 凭据模式遮蔽，
      已知 AccessKey 值只保留后四位。
    """
    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, access_keys)
        return out
    if isinstance(obj, list):
        return [redact_obj(v, secrets, access_keys) for v in obj]
    if isinstance(obj, str):
        return redact_text(obj, secrets, access_keys)
    return obj