kairee
e2cdf997af
Implement P0 of the reaction dashboard: config loading with hot reload, SQLite migrations with backups, single-writer queue, read-only Bilibili fav client, folder import/diff sync, invalid-resource handling, blacklist seed, scheduler, and JSON redacted logging. Includes offline unit and transport-level contract tests; all existing and new tests pass.
102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""脱敏 JSON 行日志。
|
|
|
|
日志只写 stdout，敏感字段（cookie / csrf / token / secret 等）在序列化前统一
|
|
替换为 ``***``；即使调用方误把异常文本或 Cookie 塞进普通字段，也会在字符串
|
|
层面再脱敏一次。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
from typing import Any, IO, Mapping
|
|
|
|
REDACTED = "***"
|
|
|
|
_SENSITIVE_KEY_PARTS = (
|
|
"cookie",
|
|
"csrf",
|
|
"token",
|
|
"secret",
|
|
"password",
|
|
"authorization",
|
|
"sessdata",
|
|
"bili_jct",
|
|
)
|
|
|
|
_STRING_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
|
|
(
|
|
re.compile(
|
|
r"(?i)\b(SESSDATA|bili_jct|buvid3|buvid4|DedeUserID__ckMd5|DedeUserID)"
|
|
r"\s*=\s*([^;\s]+)"
|
|
),
|
|
r"\1=" + REDACTED,
|
|
),
|
|
(re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._~+/=-]+"), "Bearer " + REDACTED),
|
|
(re.compile(r"\b\d{6,}:[A-Za-z0-9_-]{25,}\b"), REDACTED),
|
|
)
|
|
|
|
|
|
def _is_sensitive_key(key: str) -> bool:
|
|
normalized = key.lower().replace("-", "_")
|
|
return any(part in normalized for part in _SENSITIVE_KEY_PARTS)
|
|
|
|
|
|
def _redact_string(value: str) -> str:
|
|
for pattern, replacement in _STRING_PATTERNS:
|
|
value = pattern.sub(replacement, value)
|
|
return value
|
|
|
|
|
|
def redact(value: Any, *, key: str | None = None) -> Any:
|
|
"""递归脱敏。敏感键的整个值替换为 ``***``。"""
|
|
if key is not None and _is_sensitive_key(key):
|
|
return REDACTED
|
|
if isinstance(value, str):
|
|
return _redact_string(value)
|
|
if isinstance(value, Mapping):
|
|
return {k: redact(v, key=str(k)) for k, v in value.items()}
|
|
if isinstance(value, (list, tuple)):
|
|
return [redact(item) for item in value]
|
|
return value
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
class JsonLogger:
|
|
"""线程安全的 JSON 行 logger。"""
|
|
|
|
def __init__(self, stream: IO[str] | None = None) -> None:
|
|
self.stream = stream if stream is not None else sys.stdout
|
|
self._lock = threading.Lock()
|
|
|
|
def emit(self, level: str, event: str, **fields: Any) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"ts": _now_iso(),
|
|
"level": level,
|
|
"event": event,
|
|
}
|
|
payload.update(redact(fields))
|
|
line = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
|
with self._lock:
|
|
print(line, file=self.stream, flush=True)
|
|
return payload
|
|
|
|
def debug(self, event: str, **fields: Any) -> dict[str, Any]:
|
|
return self.emit("debug", event, **fields)
|
|
|
|
def info(self, event: str, **fields: Any) -> dict[str, Any]:
|
|
return self.emit("info", event, **fields)
|
|
|
|
def warning(self, event: str, **fields: Any) -> dict[str, Any]:
|
|
return self.emit("warning", event, **fields)
|
|
|
|
def error(self, event: str, **fields: Any) -> dict[str, Any]:
|
|
return self.emit("error", event, **fields)
|