fix(logging): redact telegram bot urls
All checks were successful
Code Review / ai-code-review (push) Successful in 17s
CD Pipeline / tests (push) Successful in 1m14s
CD Pipeline / build-and-deploy (push) Successful in 3m19s
CD Pipeline / post-deploy-checks (push) Successful in 1m15s

This commit is contained in:
Your Name
2026-05-06 16:54:00 +08:00
parent 38a4748e17
commit 33f85ec8ca
3 changed files with 60 additions and 4 deletions

View File

@@ -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 URLTelegram 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:

View 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

View File

@@ -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
```