fix(logging): redact telegram bot urls
This commit is contained in:
@@ -11,6 +11,7 @@ Features:
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
@@ -19,6 +20,28 @@ from structlog.types import Processor
|
||||
|
||||
from src.core.config import settings
|
||||
|
||||
_TELEGRAM_BOT_URL_RE = re.compile(r"(api\.telegram\.org/bot)[^/\s]+")
|
||||
|
||||
|
||||
def _redact_sensitive_log_text(text: str) -> str:
|
||||
"""遮蔽可能出現在第三方 logger 訊息中的敏感 URL。"""
|
||||
return _TELEGRAM_BOT_URL_RE.sub(r"\1<redacted>", text)
|
||||
|
||||
|
||||
class SensitiveURLRedactionFilter(logging.Filter):
|
||||
"""標準 logging filter:避免 httpx 等第三方 logger 把 token URL 打進 log。"""
|
||||
|
||||
def filter(self, record: logging.LogRecord) -> bool:
|
||||
record.msg = _redact_sensitive_log_text(str(record.msg))
|
||||
if isinstance(record.args, tuple):
|
||||
record.args = tuple(_redact_sensitive_log_text(str(arg)) for arg in record.args)
|
||||
elif isinstance(record.args, dict):
|
||||
record.args = {
|
||||
key: _redact_sensitive_log_text(str(value))
|
||||
for key, value in record.args.items()
|
||||
}
|
||||
return True
|
||||
|
||||
|
||||
def setup_logging() -> None:
|
||||
"""Configure structlog for the application"""
|
||||
@@ -68,6 +91,15 @@ def setup_logging() -> None:
|
||||
stream=sys.stdout,
|
||||
level=logging.getLevelName(settings.LOG_LEVEL),
|
||||
)
|
||||
redaction_filter = SensitiveURLRedactionFilter()
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.addFilter(redaction_filter)
|
||||
for handler in root_logger.handlers:
|
||||
handler.addFilter(redaction_filter)
|
||||
|
||||
# httpx INFO 會輸出完整 request URL;Telegram Bot API URL 內含 token。
|
||||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||||
logging.getLogger("httpcore").setLevel(logging.WARNING)
|
||||
|
||||
|
||||
def get_logger(name: str | None = None, **initial_context: Any) -> structlog.BoundLogger:
|
||||
|
||||
23
apps/api/tests/test_logging_redaction.py
Normal file
23
apps/api/tests/test_logging_redaction.py
Normal file
@@ -0,0 +1,23 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from src.core.logging import SensitiveURLRedactionFilter
|
||||
|
||||
|
||||
def test_sensitive_url_redaction_filter_redacts_log_record_args() -> None:
|
||||
record = logging.LogRecord(
|
||||
name="httpx",
|
||||
level=logging.INFO,
|
||||
pathname=__file__,
|
||||
lineno=1,
|
||||
msg="HTTP Request: %s",
|
||||
args=("https://api.telegram.org/bot123456:SECRET/getWebhookInfo",),
|
||||
exc_info=None,
|
||||
)
|
||||
|
||||
assert SensitiveURLRedactionFilter().filter(record) is True
|
||||
|
||||
rendered = record.getMessage()
|
||||
assert "SECRET" not in rendered
|
||||
assert "bot<redacted>" in rendered
|
||||
@@ -3839,18 +3839,19 @@ ruff check apps/api/tests/test_approval_execution_mcp_audit.py
|
||||
| `failover_alerter.py` | 失敗時不再使用 `logger.exception()` 輸出 chained traceback,改記錄已遮蔽的錯誤文字與錯誤類型 |
|
||||
| MarkdownV2 | `_lines_from_list()` 將編號句點改為 `1\\.`,並補上 compact 省略文字的 MarkdownV2 escape,避免治理告警清單觸發 Telegram parse 400 |
|
||||
| `telegram_gateway.py` | HTTPStatusError 不再 `raise ... from e`,OTel span 也只記 sanitized gateway error,避免 httpx exception 字串帶出 Bot URL |
|
||||
| `core/logging.py` | 新增敏感 URL redaction filter,並將 `httpx/httpcore` logger 壓到 WARNING,避免成功 request log 輸出 Telegram Bot API token URL |
|
||||
| 測試 | 新增 Telegram error sanitizer 與 MarkdownV2 編號 escape 回歸測試 |
|
||||
|
||||
### 驗證
|
||||
|
||||
```text
|
||||
pytest apps/api/tests/test_failover_alerter.py apps/api/tests/test_telegram_gateway_error_sanitizer.py apps/api/tests/test_heartbeat_dedup_p0_4.py -q
|
||||
# 17 passed
|
||||
pytest apps/api/tests/test_failover_alerter.py apps/api/tests/test_telegram_gateway_error_sanitizer.py apps/api/tests/test_heartbeat_dedup_p0_4.py apps/api/tests/test_logging_redaction.py -q
|
||||
# 18 passed
|
||||
|
||||
py_compile apps/api/src/services/failover_alerter.py apps/api/src/services/telegram_gateway.py apps/api/tests/test_failover_alerter.py apps/api/tests/test_telegram_gateway_error_sanitizer.py
|
||||
py_compile apps/api/src/core/logging.py apps/api/src/services/failover_alerter.py apps/api/src/services/telegram_gateway.py apps/api/tests/test_failover_alerter.py apps/api/tests/test_telegram_gateway_error_sanitizer.py apps/api/tests/test_logging_redaction.py
|
||||
# 通過
|
||||
|
||||
ruff check apps/api/src/services/failover_alerter.py apps/api/tests/test_failover_alerter.py apps/api/tests/test_telegram_gateway_error_sanitizer.py
|
||||
ruff check apps/api/src/core/logging.py apps/api/src/services/failover_alerter.py apps/api/tests/test_failover_alerter.py apps/api/tests/test_telegram_gateway_error_sanitizer.py apps/api/tests/test_logging_redaction.py
|
||||
# All checks passed
|
||||
```
|
||||
|
||||
|
||||
Reference in New Issue
Block a user