43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
"""请求日志上下文和敏感摘要清理。"""
|
|
|
|
import contextvars
|
|
import logging
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
request_id_context: contextvars.ContextVar[str] = contextvars.ContextVar(
|
|
"request_id", default="-"
|
|
)
|
|
|
|
SENSITIVE_KEYS = {"api_key", "authorization", "cookie", "password", "secret", "token"}
|
|
|
|
|
|
def _is_sensitive_key(key: object) -> bool:
|
|
"""识别常见秘密字段，同时避免把 `token_count` 等统计字段误判为秘密。"""
|
|
|
|
normalized = str(key).strip().lower().replace("-", "_")
|
|
if normalized in SENSITIVE_KEYS:
|
|
return True
|
|
return normalized.endswith(("_api_key", "_password", "_secret", "_cookie", "_token"))
|
|
|
|
|
|
class RequestContextFilter(logging.Filter):
|
|
"""向每条日志补充请求关联标识，后台任务没有请求时使用短横线。"""
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
record.request_id = request_id_context.get()
|
|
return True
|
|
|
|
|
|
def sanitize_summary(value: Any) -> Any:
|
|
"""递归遮蔽常见敏感键；调用方仍需限制摘要的业务范围和长度。"""
|
|
|
|
if isinstance(value, Mapping):
|
|
return {
|
|
str(key): "***" if _is_sensitive_key(key) else sanitize_summary(item)
|
|
for key, item in value.items()
|
|
}
|
|
if isinstance(value, list):
|
|
return [sanitize_summary(item) for item in value]
|
|
return value
|