81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""采购工具日志的最小脱敏策略。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
import re
|
|
|
|
from .runtime import RuntimePaths
|
|
|
|
|
|
LOGGER_NAME = "cmbuyer_client"
|
|
REDACTED = "[已隐藏]"
|
|
|
|
_SENSITIVE_KEY_PATTERN = (
|
|
r"token|authorization|password|secret|api[_-]?key|"
|
|
r"address|phone|mobile|payment|pay|card|bank[_-]?account|"
|
|
r"令牌|授权|密码|密钥|地址|手机号|电话|支付|银行卡"
|
|
)
|
|
_KEY_VALUE_PATTERN = re.compile(
|
|
rf"(?P<key>{_SENSITIVE_KEY_PATTERN})\s*(?P<separator>[:=])\s*"
|
|
r"(?P<value>\"[^\"]*\"|'[^']*'|[^\s,;]+)",
|
|
flags=re.IGNORECASE,
|
|
)
|
|
_PHONE_PATTERN = re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)")
|
|
_BEARER_PATTERN = re.compile(r"(?i)\bBearer\s+[0-9a-f]{64}\b")
|
|
_BARE_TOKEN_PATTERN = re.compile(r"(?<![0-9a-fA-F])[0-9a-fA-F]{64}(?![0-9a-fA-F])")
|
|
|
|
|
|
def redact_text(message: str) -> str:
|
|
"""移除日志文本中的凭据、地址、手机号和支付字段值。"""
|
|
|
|
def replace_key_value(match: re.Match[str]) -> str:
|
|
return f"{match.group('key')}{match.group('separator')}{REDACTED}"
|
|
|
|
redacted = _BEARER_PATTERN.sub("Bearer " + REDACTED, message)
|
|
redacted = _KEY_VALUE_PATTERN.sub(replace_key_value, redacted)
|
|
redacted = _BARE_TOKEN_PATTERN.sub(REDACTED, redacted)
|
|
return _PHONE_PATTERN.sub(REDACTED, redacted)
|
|
|
|
|
|
class SensitiveDataFilter(logging.Filter):
|
|
"""在任何 handler 格式化记录前，清除敏感字段。
|
|
|
|
该过滤器在日志写入前替换 ``msg`` 与 ``args``，确保文件 handler 不会得到原始值。
|
|
"""
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
record.msg = redact_text(record.getMessage())
|
|
record.args = ()
|
|
return True
|
|
|
|
|
|
class RedactingFormatter(logging.Formatter):
|
|
"""再次处理完整格式化文本，覆盖异常 traceback 中的敏感值。"""
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
return redact_text(super().format(record))
|
|
|
|
def formatException(self, exc_info: tuple[type[BaseException], BaseException, object]) -> str:
|
|
return redact_text(super().formatException(exc_info))
|
|
|
|
|
|
def configure_application_logger(paths: RuntimePaths) -> logging.Logger:
|
|
"""配置唯一的 UTF-8 文件日志，并确保其先经过脱敏过滤。"""
|
|
|
|
paths.ensure_exists()
|
|
logger = logging.getLogger(LOGGER_NAME)
|
|
logger.setLevel(logging.INFO)
|
|
logger.propagate = False
|
|
|
|
for handler in tuple(logger.handlers):
|
|
logger.removeHandler(handler)
|
|
handler.close()
|
|
|
|
handler = logging.FileHandler(Path(paths.logs) / "client.log", encoding="utf-8")
|
|
handler.addFilter(SensitiveDataFilter())
|
|
handler.setFormatter(RedactingFormatter("%(asctime)s %(levelname)s %(message)s"))
|
|
logger.addHandler(handler)
|
|
return logger
|