172 lines
7.1 KiB
Python
172 lines
7.1 KiB
Python
"""
|
|
Tests for BATON-FIX-005: BOT_TOKEN leak prevention in logs.
|
|
|
|
Acceptance criteria covered by unit tests:
|
|
AC#4 — no places in source code where token is logged in plain text:
|
|
- _mask_token() returns masked representation (***XXXX format)
|
|
- validate_bot_token() exception handler does not log raw BOT_TOKEN
|
|
- validate_bot_token() exception handler logs type(exc).__name__ + masked token
|
|
- httpcore logger level >= WARNING (prevents URL leak via transport layer)
|
|
|
|
AC#1, AC#2, AC#3 (journalctl, webhook, service health) require live production
|
|
verification and are outside unit test scope.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
|
|
os.environ.setdefault("BOT_TOKEN", "test-bot-token")
|
|
os.environ.setdefault("CHAT_ID", "-1001234567890")
|
|
os.environ.setdefault("WEBHOOK_SECRET", "test-webhook-secret")
|
|
os.environ.setdefault("WEBHOOK_URL", "https://example.com/api/webhook/telegram")
|
|
os.environ.setdefault("FRONTEND_ORIGIN", "http://localhost:3000")
|
|
os.environ.setdefault("ADMIN_TOKEN", "test-admin-token")
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from backend import config
|
|
from backend.telegram import _mask_token, validate_bot_token
|
|
|
|
GET_ME_URL = f"https://api.telegram.org/bot{config.BOT_TOKEN}/getMe"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _mask_token helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_mask_token_shows_last_4_chars():
|
|
"""_mask_token returns '***XXXX' where XXXX is the last 4 chars of the token."""
|
|
token = "123456789:ABCDEFsomeLongTokenXYZW"
|
|
result = _mask_token(token)
|
|
assert result == f"***{token[-4:]}", f"Expected ***{token[-4:]}, got {result!r}"
|
|
|
|
|
|
def test_mask_token_hides_most_of_token():
|
|
"""_mask_token must NOT expose the full token — only last 4 chars."""
|
|
token = "123456789:ABCDEFsomeLongTokenXYZW"
|
|
result = _mask_token(token)
|
|
assert token[:-4] not in result, f"Masked token exposes too much: {result!r}"
|
|
|
|
|
|
def test_mask_token_short_token_returns_redacted():
|
|
"""_mask_token returns '***REDACTED***' for tokens shorter than 4 chars."""
|
|
assert _mask_token("abc") == "***REDACTED***"
|
|
|
|
|
|
def test_mask_token_empty_string_returns_redacted():
|
|
"""_mask_token on empty string returns '***REDACTED***'."""
|
|
assert _mask_token("") == "***REDACTED***"
|
|
|
|
|
|
def test_mask_token_exactly_4_chars_is_not_redacted():
|
|
"""_mask_token with exactly 4 chars returns '***XXXX' (not redacted)."""
|
|
result = _mask_token("1234")
|
|
assert result == "***1234", f"Expected ***1234, got {result!r}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# httpcore logger suppression (new in FIX-005; httpx covered in test_fix_011)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_httpcore_logger_level_is_warning_or_higher():
|
|
"""logging.getLogger('httpcore').level must be WARNING or higher after app import."""
|
|
import backend.main # noqa: F401 — ensures telegram.py module-level setLevel is called
|
|
|
|
httpcore_logger = logging.getLogger("httpcore")
|
|
assert httpcore_logger.level >= logging.WARNING, (
|
|
f"httpcore logger level must be >= WARNING (30), got {httpcore_logger.level}. "
|
|
"httpcore logs transport-level requests including URLs with BOT_TOKEN."
|
|
)
|
|
|
|
|
|
def test_httpcore_logger_info_not_enabled():
|
|
"""httpcore logger must not propagate INFO-level messages (would leak BOT_TOKEN URL)."""
|
|
import backend.main # noqa: F401
|
|
|
|
httpcore_logger = logging.getLogger("httpcore")
|
|
assert not httpcore_logger.isEnabledFor(logging.INFO), (
|
|
"httpcore logger must not process INFO messages — could leak BOT_TOKEN via URL"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# validate_bot_token() exception handler — AC#4: no raw token in error logs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_bot_token_network_error_does_not_log_raw_token(caplog):
|
|
"""validate_bot_token() on ConnectError must NOT log the raw BOT_TOKEN.
|
|
|
|
AC#4: The exception handler logs type(exc).__name__ + _mask_token() instead
|
|
of raw exc, which embeds the Telegram API URL containing the token.
|
|
"""
|
|
with respx.mock(assert_all_called=False) as mock:
|
|
mock.get(GET_ME_URL).mock(side_effect=httpx.ConnectError("connection refused"))
|
|
with caplog.at_level(logging.ERROR, logger="backend.telegram"):
|
|
result = await validate_bot_token()
|
|
|
|
assert result is False
|
|
|
|
raw_token = config.BOT_TOKEN
|
|
for record in caplog.records:
|
|
assert raw_token not in record.message, (
|
|
f"AC#4: Raw BOT_TOKEN leaked in log message: {record.message!r}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_bot_token_network_error_logs_exception_type_name(caplog):
|
|
"""validate_bot_token() on ConnectError logs the exception type name, not repr(exc).
|
|
|
|
The fixed handler: logger.error('...%s...', type(exc).__name__, ...) — not str(exc).
|
|
"""
|
|
with respx.mock(assert_all_called=False) as mock:
|
|
mock.get(GET_ME_URL).mock(side_effect=httpx.ConnectError("connection refused"))
|
|
with caplog.at_level(logging.ERROR, logger="backend.telegram"):
|
|
await validate_bot_token()
|
|
|
|
error_messages = [r.message for r in caplog.records if r.levelno >= logging.ERROR]
|
|
assert error_messages, "Expected at least one ERROR log on network failure"
|
|
assert any("ConnectError" in msg for msg in error_messages), (
|
|
f"Expected 'ConnectError' (type name) in error log, got: {error_messages}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_bot_token_network_error_logs_masked_token(caplog):
|
|
"""validate_bot_token() on network error logs masked token (***XXXX), not raw token."""
|
|
with respx.mock(assert_all_called=False) as mock:
|
|
mock.get(GET_ME_URL).mock(side_effect=httpx.ConnectError("connection refused"))
|
|
with caplog.at_level(logging.ERROR, logger="backend.telegram"):
|
|
await validate_bot_token()
|
|
|
|
token = config.BOT_TOKEN # "test-bot-token"
|
|
masked = f"***{token[-4:]}" # "***oken"
|
|
error_messages = [r.message for r in caplog.records if r.levelno >= logging.ERROR]
|
|
assert any(masked in msg for msg in error_messages), (
|
|
f"Expected masked token '{masked}' in error log. Got: {error_messages}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_bot_token_network_error_no_api_url_in_logs(caplog):
|
|
"""validate_bot_token() on network error must not log the Telegram API URL.
|
|
|
|
httpx embeds the request URL (including the token) into exception repr/str.
|
|
The fixed handler avoids logging exc directly to prevent this leak.
|
|
"""
|
|
with respx.mock(assert_all_called=False) as mock:
|
|
mock.get(GET_ME_URL).mock(side_effect=httpx.ConnectError("connection refused"))
|
|
with caplog.at_level(logging.ERROR, logger="backend.telegram"):
|
|
await validate_bot_token()
|
|
|
|
for record in caplog.records:
|
|
assert "api.telegram.org" not in record.message, (
|
|
f"AC#4: Telegram API URL (containing token) leaked in log: {record.message!r}"
|
|
)
|